feat: add a WebSocket watch session with in-stream progress requests

api_v3.md

@@ -12,6 +12,7 @@ API V3
     * [watchcancel](#watchcancel)
     * [readdir](#readdir)
     * [watchdir](#watchdir)
+    * [create_ws_watch_session](#create_ws_watch_session)
     * [rmdir](#rmdir)
     * [txn](#txn)
     * [version](#version)
@@ -236,6 +237,70 @@ local res, err = cli:watchdir('/path/to/dir')
 [Back to TOP](#api-v3)
 
 
+### create_ws_watch_session
+
+`syntax: session, err = cli:create_ws_watch_session(dir:string [, opts:table])`
+
+* `dir`: string value, the key prefix to watch (the range end is derived the same way as `watchdir`).
+* `opts`: optional options.
+    * `timeout`: (int) default timeout seconds for `recv` when it is called without one.
+    * `start_revision`: (int) start_revision is an optional revision to watch from (inclusive). No start_revision is "now".
+    * `progress_notify`: (bool) ask the etcd server to periodically send a WatchResponse with no events.
+    * `filters`: (slice of [enum FilterType {NOPUT = 0;NODELETE = 1;}]) filters filter the events at server side before it sends back to the watcher.
+    * `prev_kv`: (bool) If prev_kv is set, created watcher gets the previous KV before the event happens.
+    * `watch_id`: (int) If watch_id is provided and non-zero, it will be assigned to this watcher.
+    * `max_payload_len`: (int) maximal length of a single WebSocket frame accepted from etcd, defaults to 32MB.
+
+Watch a key prefix over a full-duplex WebSocket stream to etcd's JSON gateway.
+
+etcd's gateway is half-duplex over plain HTTP/1.1: the server sends no response
+bytes while the request body is still open, so a watch opened with `watchdir`
+can never write to the stream again. etcd also wraps `/v3/` in
+grpc-websocket-proxy, and a WebSocket upgrade on `/v3/watch` carries one
+WatchRequest per text frame in and one WatchResponse per text frame out. That
+makes `WatchProgressRequest` usable on a live watch: its reply proves this very
+stream already delivered every event up to the reported revision, which is safe
+to resume from after a reconnect, unlike a revision learned from a second
+connection.
+
+The returned session has three methods:
+
+* `res, err = session:recv(timeout)`: waits up to `timeout` seconds for the
+  next WatchResponse, decoded exactly like a `watchdir` response. `err` is
+  `"timeout"` when nothing arrived in time and `"closed"` when the stream is
+  gone.
+* `ok, err = session:request_progress()`: sends a `progress_request` on the
+  live stream. The reply arrives via `recv()` as a WatchResponse without
+  `events`; its `header.revision` is the delivery barrier. etcd only answers
+  once every watcher on the stream is synced, so a request sent immediately
+  after session creation may be dropped (etcd >= 3.6) and should be retried.
+* `session:close()`: closes the stream.
+
+Session creation performs the upgrade, sends the `create_request`, and waits
+for the first WatchResponse; an endpoint that cannot upgrade (for example a
+proxy that strips the `Upgrade` header) is therefore rejected here, so callers
+can fall back to `watchdir`. The consumed response is returned by the first
+`recv()` call.
+
+Note: requires the etcd server to answer progress requests (etcd >= 3.4), and
+etcd >= 3.5 for watch responses larger than 64KB (older gateways buffer
+response lines with a fixed 64KB limit).
+
+```lua
+local session, err = cli:create_ws_watch_session('/path/to/dir', {start_revision = rev})
+local res, err = session:recv(50)
+if not res and err == "timeout" then
+    session:request_progress()
+    local progress = session:recv(3)
+    if progress and not progress.result.events then
+        rev = tonumber(progress.result.header.revision) + 1
+    end
+end
+```
+
+[Back to TOP](#api-v3)
+
+
 ### rmdir
 
 `syntax: res, err = cli:rmdir(dir:string [, opts:table])`

lib/resty/etcd/v3.lua

@@ -31,6 +31,7 @@ local semaphore     = require("ngx.semaphore")
 local health_check  = require("resty.etcd.health_check")
 local pl_path       = require("pl.path")
 local grpc_proto    = require("resty.etcd.proto")
+local ws_client     = require("resty.websocket.client")
 math.randomseed(now() * 1000 + ngx.worker.pid())
 
 local INIT_COUNT_RESIZE = 2e8
@@ -1029,6 +1030,199 @@ local function create_watch_request(key, attr)
 end
 
 
+-- WebSocket watch session.
+--
+-- etcd's JSON gateway is half-duplex over plain HTTP/1.1: Go's net/http
+-- server emits no response bytes while the request body is still open.
+-- etcd also wraps /v3/ in grpc-websocket-proxy, so a WebSocket upgrade on
+-- /v3/watch gives a full-duplex stream: one WatchRequest per text frame in,
+-- one WatchResponse per text frame out. That makes in-stream
+-- progress_request usable as a delivery barrier (apache/apisix#13777).
+
+local ws_session_mt = {}
+ws_session_mt.__index = ws_session_mt
+
+
+local function ws_decode_frame(self, frame)
+    local body, err = decode_json(frame)
+    if not body then
+        return nil, "failed to decode json body: " .. (err or " unknown")
+    end
+
+    if body.error and body.error.http_code and body.error.http_code >= 500 then
+        health_check.report_failure(self.endpoint.http_host)
+        return nil, self.endpoint.http_host .. ": "
+                    .. (body.error.http_status or body.error.http_code)
+    end
+
+    if body.result and body.result.events then
+        for _, event in ipairs(body.result.events) do
+            if event.kv.value then    -- DELETE not have value
+                event.kv.value = decode_base64(event.kv.value or "")
+                event.kv.value = self.cli.serializer.deserialize(event.kv.value)
+            end
+            event.kv.key = decode_base64(event.kv.key)
+            if event.prev_kv then
+                event.prev_kv.value = decode_base64(event.prev_kv.value or "")
+                event.prev_kv.value = self.cli.serializer.deserialize(event.prev_kv.value)
+                event.prev_kv.key = decode_base64(event.prev_kv.key)
+            end
+        end
+    end
+
+    return body
+end
+
+
+-- returns one decoded WatchResponse, or nil + "timeout"/"closed"/error
+function ws_session_mt.recv(self, timeout)
+    if self.pending then
+        local res = self.pending
+        self.pending = nil
+        return res
+    end
+
+    local ws = self.ws
+    ws:set_timeout((timeout or self.timeout) * 1000)
+
+    local buf
+    while true do
+        local data, typ, err = ws:recv_frame()
+        if not data then
+            if err and str_find(err, "timeout", 1, true) then
+                return nil, "timeout"
+            end
+            return nil, err or "closed"
+        end
+
+        if typ == "text" or typ == "binary" or typ == "continuation" then
+            if err == "again" then    -- fragmented frame, more to come
+                buf = (buf or "") .. data
+            else
+                if buf then
+                    data = buf .. data
+                end
+                return ws_decode_frame(self, data)
+            end
+        elseif typ == "ping" then
+            ws:send_pong(data)
+        elseif typ == "close" then
+            return nil, "closed"
+        end
+        -- pong or unknown frame: keep reading
+    end
+end
+
+
+-- asks etcd how far this stream has delivered; the reply arrives via recv()
+-- as a WatchResponse with no events
+function ws_session_mt.request_progress(self)
+    local bytes, err = self.ws:send_text('{"progress_request":{}}')
+    if not bytes then
+        return nil, err
+    end
+    return true
+end
+
+
+function ws_session_mt.close(self)
+    return self.ws:close()
+end
+
+
+function _M.create_ws_watch_session(self, key, opts)
+    if self.unix_socket_proxy then
+        return nil, "websocket watch does not support unix socket proxy"
+    end
+
+    key = utils.get_real_key(self.key_prefix, key)
+
+    local attr = {
+        range_end       = get_range_end(key),
+        start_revision  = opts and opts.start_revision,
+        progress_notify = opts and opts.progress_notify,
+        filters         = opts and opts.filters,
+        prev_kv         = opts and opts.prev_kv,
+        watch_id        = opts and opts.watch_id,
+    }
+
+    local create_request = create_watch_request(key, attr)
+    create_request.key = encode_base64(key)
+    create_request.range_end = encode_base64(attr.range_end)
+
+    local endpoint, err = choose_endpoint(self)
+    if not endpoint then
+        return nil, err
+    end
+
+    local conn_opts = {}
+    if self.is_auth then
+        local _, auth_err = refresh_jwt_token(self, (opts and opts.timeout) or self.timeout)
+        if auth_err then
+            return nil, auth_err
+        end
+        -- grpc-websocket-proxy turns this subprotocol into an Authorization header
+        conn_opts.protocols = "Bearer," .. self.jwt_token
+    end
+
+    local scheme = "ws"
+    if endpoint.scheme == "https" then
+        scheme = "wss"
+        conn_opts.ssl_verify = self.ssl_verify
+        conn_opts.server_name = self.sni or endpoint.host
+        conn_opts.client_cert = self.ssl_cert
+        conn_opts.client_priv_key = self.ssl_key
+    end
+
+    local ws, new_err = ws_client:new({
+        max_payload_len = (opts and opts.max_payload_len) or 32 * 1024 * 1024,
+    })
+    if not ws then
+        return nil, new_err
+    end
+
+    local uri = scheme .. "://" .. endpoint.address .. ":" .. endpoint.port
+                .. endpoint.api_prefix .. "/watch"
+
+    local ok, conn_err = ws:connect(uri, conn_opts)
+    if not ok then
+        health_check.report_failure(endpoint.http_host)
+        return nil, endpoint.http_host .. ": " .. conn_err
+    end
+
+    local req_body, encode_err = encode_json({create_request = create_request})
+    if not req_body then
+        ws:close()
+        return nil, encode_err
+    end
+
+    local bytes, send_err = ws:send_text(req_body)
+    if not bytes then
+        ws:close()
+        return nil, send_err
+    end
+
+    local session = setmetatable({
+        cli      = self,
+        ws       = ws,
+        endpoint = endpoint,
+        timeout  = (opts and opts.timeout) or self.timeout,
+    }, ws_session_mt)
+
+    -- resty.websocket.client accepts any HTTP/1.1 status line as a handshake,
+    -- so an intermediary that stripped the Upgrade header still "connects".
+    -- Only a real WatchResponse proves the stream works; keep it for recv().
+    local first, ferr = session:recv()
+    if not first then
+        ws:close()
+        return nil, "websocket watch stream unusable: " .. (ferr or "unknown")
+    end
+    session.pending = first
+
+    return session
+end
+
+
 local get_grpc_metadata
 do
     local metadata = {