From 1b934733774cfd49a666aa94b2c7602853897a3e Mon Sep 17 00:00:00 2001 From: Abhishek Choudhary Date: Wed, 12 Aug 2026 21:27:31 +0545 Subject: [PATCH 1/2] feat: add a WebSocket watch session with in-stream progress requests etcd's JSON gateway is half-duplex over plain HTTP/1.1: the server emits no response bytes while the request body is still open, so the existing /v3/watch stream is write-once and a WatchProgressRequest can never be sent on a live watch. etcd wraps /v3/ in grpc-websocket-proxy, so a WebSocket upgrade on /v3/watch gives a genuinely full-duplex stream. create_ws_watch_session() opens such a stream and returns a session with recv(), request_progress() and close(). A progress reply proves the stream already delivered everything up to its revision, which gives watchers a delivery barrier for keeping their revision fresh on an idle prefix (see apache/apisix#13777). Session creation validates the transport by consuming the first WatchResponse, because resty.websocket.client accepts any HTTP/1.1 status line as a handshake, so an intermediary that strips the Upgrade header would otherwise look connected (openresty/lua-resty-websocket#104 fixes that upstream). --- api_v3.md | 65 ++++++++++ lib/resty/etcd/v3.lua | 195 ++++++++++++++++++++++++++++++ t/v3/ws_watch.t | 268 ++++++++++++++++++++++++++++++++++++++++++ 3 files changed, 528 insertions(+) create mode 100644 t/v3/ws_watch.t diff --git a/api_v3.md b/api_v3.md index 3dd9e37d..f8d92949 100644 --- a/api_v3.md +++ b/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])` diff --git a/lib/resty/etcd/v3.lua b/lib/resty/etcd/v3.lua index be5898d5..6ff435f9 100644 --- a/lib/resty/etcd/v3.lua +++ b/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,200 @@ 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 + buf = nil + 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 = { diff --git a/t/v3/ws_watch.t b/t/v3/ws_watch.t new file mode 100644 index 00000000..3c78cebd --- /dev/null +++ b/t/v3/ws_watch.t @@ -0,0 +1,268 @@ +use Test::Nginx::Socket::Lua; + +log_level('info'); +no_long_string(); +repeat_each(1); + +my $etcd_version = `etcd --version`; +if ($etcd_version =~ /^etcd Version: 2/ || $etcd_version =~ /^etcd Version: 3.1./ + || $etcd_version =~ /^etcd Version: 3.2./ || $etcd_version =~ /^etcd Version: 3.3./) { + plan(skip_all => "etcd is too old, progress requests need etcd >= 3.4"); +} else { + plan 'no_plan'; +} + +our $HttpConfig = <<'_EOC_'; + lua_socket_log_errors off; + lua_package_path 'lib/?.lua;/usr/local/share/lua/5.3/?.lua;/usr/share/lua/5.1/?.lua;;'; + init_by_lua_block { + function new_cli() + local etcd, err = require("resty.etcd").new({ + protocol = "v3", + http_host = "http://127.0.0.1:2379", + timeout = 5, + }) + if not etcd then + ngx.say("failed to new etcd: ", err) + ngx.exit(200) + end + return etcd + end + + function current_rev(etcd, key) + local res, err = etcd:get(key) + if not res then + ngx.say("failed to get: ", err) + ngx.exit(200) + end + return tonumber(res.body.header.revision) + end + } +_EOC_ + +# a plain HTTP endpoint that accepts the websocket upgrade request but never +# actually upgrades: the handshake reply is a normal 200 response +our $HttpConfigNoUpgrade = <<'_EOC_'; + lua_socket_log_errors off; + lua_package_path 'lib/?.lua;/usr/local/share/lua/5.3/?.lua;/usr/share/lua/5.1/?.lua;;'; + server { + listen 1985; + location /v3/watch { + content_by_lua_block { + ngx.print("not a websocket endpoint") + } + } + } +_EOC_ + +run_tests(); + +__DATA__ + +=== TEST 1: created ack, then a live event while the request side stays open +--- http_config eval: $::HttpConfig +--- config + location /t { + content_by_lua_block { + local etcd = new_cli() + assert(etcd:set("/ws_watch/seed", "seed")) + local rev = current_rev(etcd, "/ws_watch/seed") + + local sess, err = etcd:create_ws_watch_session("/ws_watch", + {start_revision = rev + 1}) + if not sess then + ngx.say("failed to create session: ", err) + return + end + + local res, err = sess:recv(2) + if not res then + ngx.say("failed to recv created ack: ", err) + return + end + ngx.say("created: ", res.result.created == true) + + assert(etcd:set("/ws_watch/live", "abc")) + res, err = sess:recv(2) + if not res then + ngx.say("failed to recv event: ", err) + return + end + local event = res.result.events[1] + ngx.say("event: ", event.type or "PUT", " ", event.kv.key, "=", event.kv.value) + sess:close() + } + } +--- request +GET /t +--- response_body +created: true +event: PUT /ws_watch/live=abc +--- no_error_log +[error] + + + +=== TEST 2: a progress request on an idle prefix reports the global revision +--- http_config eval: $::HttpConfig +--- config + location /t { + content_by_lua_block { + local etcd = new_cli() + assert(etcd:set("/ws_watch/seed", "seed")) + local rev = current_rev(etcd, "/ws_watch/seed") + + local sess = assert(etcd:create_ws_watch_session("/ws_watch", + {start_revision = rev + 1})) + assert(sess:recv(2)) -- created ack + + -- etcd only answers once the watcher is synced; a request landing + -- in the pre-sync window (<= 100ms) is dropped, so retry once + ngx.sleep(0.3) + assert(sess:request_progress()) + local res, err = sess:recv(2) + if not res then + assert(sess:request_progress()) + res, err = sess:recv(2) + end + if not res then + ngx.say("failed to recv progress: ", err) + return + end + -- >= rev rather than == rev: anything else sharing the etcd may + -- have bumped the global revision since current_rev() sampled it + ngx.say("idle progress at current revision: ", + res.result.events == nil + and tonumber(res.result.header.revision) >= rev) + + -- writes OUTSIDE the watched prefix produce no events, yet the + -- barrier must follow the global revision + for i = 1, 3 do + assert(etcd:set("/ws_elsewhere/k", "v" .. i)) + end + assert(sess:request_progress()) + res, err = sess:recv(2) + if not res then + ngx.say("failed to recv progress: ", err) + return + end + ngx.say("barrier followed foreign writes: ", + res.result.events == nil + and tonumber(res.result.header.revision) >= rev + 3) + sess:close() + } + } +--- request +GET /t +--- response_body +idle progress at current revision: true +barrier followed foreign writes: true +--- no_error_log +[error] + + + +=== TEST 3: a stream resumed from the barrier revision survives compaction +--- http_config eval: $::HttpConfig +--- config + location /t { + content_by_lua_block { + local cjson = require("cjson.safe") + local http = require("resty.http") + local etcd = new_cli() + assert(etcd:set("/ws_watch/seed", "seed")) + local stale_rev = current_rev(etcd, "/ws_watch/seed") + + -- the global revision moves on while the watched prefix is idle + for i = 1, 3 do + assert(etcd:set("/ws_elsewhere/k", "v" .. i)) + end + + -- learn the barrier from a live session, exactly as a watcher would + local sess = assert(etcd:create_ws_watch_session("/ws_watch", + {start_revision = stale_rev + 1})) + assert(sess:recv(2)) -- created ack + ngx.sleep(0.3) + assert(sess:request_progress()) + local res = sess:recv(2) + if not res then + assert(sess:request_progress()) + res = assert(sess:recv(2)) + end + local barrier = tonumber(res.result.header.revision) + sess:close() + + -- compact away every revision below the barrier + local httpc = http.new() + local cres = assert(httpc:request_uri("http://127.0.0.1:2379/v3/kv/compaction", { + method = "POST", + body = cjson.encode({revision = tostring(barrier), physical = true}), + })) + ngx.say("compacted: ", cres.status == 200) + + -- resuming from the barrier is legal: created ack, no compact cancel + local sess2 = assert(etcd:create_ws_watch_session("/ws_watch", + {start_revision = barrier + 1})) + local res2 = assert(sess2:recv(2)) + ngx.say("resume ok: ", res2.result.created == true + and res2.result.canceled == nil) + + assert(etcd:set("/ws_watch/after-compact", "z")) + res2 = assert(sess2:recv(2)) + ngx.say("delivery after compaction: ", + res2.result.events[1].kv.key == "/ws_watch/after-compact") + sess2:close() + + -- control: resuming from the pre-barrier revision must be refused, + -- which is the expensive resync path the barrier avoids + local sess3 = assert(etcd:create_ws_watch_session("/ws_watch", + {start_revision = stale_rev + 1})) + local canceled = false + for _ = 1, 3 do + local res3 = sess3:recv(2) + if res3 and res3.result.canceled and res3.result.compact_revision then + canceled = true + break + end + end + ngx.say("stale revision compact-canceled: ", canceled) + sess3:close() + } + } +--- request +GET /t +--- response_body +compacted: true +resume ok: true +delivery after compaction: true +stale revision compact-canceled: true +--- no_error_log +[error] + + + +=== TEST 4: an endpoint that cannot upgrade is rejected at session creation +--- http_config eval: $::HttpConfigNoUpgrade +--- config + location /t { + content_by_lua_block { + local etcd, err = require("resty.etcd").new({ + protocol = "v3", + http_host = "http://127.0.0.1:1985", + timeout = 2, + }) + if not etcd then + ngx.say("failed to new etcd: ", err) + return + end + + local sess, err = etcd:create_ws_watch_session("/ws_watch", {}) + ngx.say("session refused: ", sess == nil, ", err: ", err ~= nil) + } + } +--- request +GET /t +--- response_body +session refused: true, err: true +--- no_error_log +[error] From f9967c7cee3dfb8e7270e70890d9f853886931c4 Mon Sep 17 00:00:00 2001 From: Abhishek Choudhary Date: Mon, 17 Aug 2026 21:05:35 +0545 Subject: [PATCH 2/2] style: drop a dead assignment flagged by luacheck --- lib/resty/etcd/v3.lua | 1 - 1 file changed, 1 deletion(-) diff --git a/lib/resty/etcd/v3.lua b/lib/resty/etcd/v3.lua index 6ff435f9..9c748ffb 100644 --- a/lib/resty/etcd/v3.lua +++ b/lib/resty/etcd/v3.lua @@ -1101,7 +1101,6 @@ function ws_session_mt.recv(self, timeout) else if buf then data = buf .. data - buf = nil end return ws_decode_frame(self, data) end