feat: add a WebSocket watch session with in-stream progress requests - #226
feat: add a WebSocket watch session with in-stream progress requests#226shreemaan-abhishek wants to merge 1 commit into
Conversation
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).
📝 WalkthroughWalkthroughThe client now supports full-duplex etcd WebSocket watch sessions. It adds session creation, event and progress handling, connection cleanup, API documentation, and tests for events, compaction, resumption, and failed upgrades. ChangesWebSocket watch sessions
Estimated code review effort: 4 (Complex) | ~45 minutes 🚥 Pre-merge checks | ✅ 5 | ❌ 1❌ Failed checks (1 warning)
✅ Passed checks (5 passed)
✨ Finishing Touches📝 Generate docstrings
🧪 Generate unit tests (beta)
Comment |
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 3
🧹 Nitpick comments (3)
lib/resty/etcd/v3.lua (1)
1089-1114: 🩺 Stability & Availability | 🔵 Trivial | ⚡ Quick winBound the total wait with a deadline.
set_timeoutapplies perrecv_framecall. Each ping, pong, or fragment restarts the budget, sorecv(2)can block far longer than 2 seconds when etcd keeps the connection busy. api_v3.md states thatrecvwaits up totimeoutseconds.Track a deadline and shorten the socket timeout on each iteration, or return
"timeout"once the deadline passes.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@lib/resty/etcd/v3.lua` around lines 1089 - 1114, Update the websocket receive loop around ws:recv_frame() to enforce one total timeout deadline for the recv operation, rather than restarting the full per-call timeout after pings, pongs, or fragmented frames. Track the deadline and reduce each iteration’s socket timeout accordingly, returning "timeout" once it expires while preserving existing frame assembly and close/error handling.t/v3/ws_watch.t (2)
15-56: 📐 Maintainability & Code Quality | 🔵 Trivial | 🏗️ Heavy liftConsider covering authentication and
wss.The suite exercises plaintext
wsonly.create_ws_watch_sessionalso passes the JWT throughSec-WebSocket-Protocoland mapshttpsendpoints towsswith mTLS options. Those two paths carry the highest risk of silent option loss across lua-resty-websocket versions.Do you want me to draft the auth and
wsstest cases?🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@t/v3/ws_watch.t` around lines 15 - 56, Extend the websocket watch test suite beyond plaintext ws by adding coverage for JWT authentication via the Sec-WebSocket-Protocol header and HTTPS endpoints mapped to wss. Exercise create_ws_watch_session with mTLS options and verify both authentication and TLS settings are preserved through the connection.
195-201: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick winPhysical compaction affects the whole shared etcd instance.
This request compacts every revision below the barrier on the etcd instance that all test files share. Any other test that resumes a watch or reads from an earlier revision can then fail, and the failure depends on test execution order.
Consider running this test against a dedicated etcd instance, or moving it into a separately gated test file so the destructive step is explicit.
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@t/v3/ws_watch.t` around lines 195 - 201, Isolate the destructive physical compaction in the watch test around the compaction request so it cannot affect the shared etcd instance used by other tests. Run this scenario against a dedicated etcd instance, or move it into a separately gated test file with explicit setup and teardown, while preserving the existing compaction assertion.
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Inline comments:
In `@lib/resty/etcd/v3.lua`:
- Around line 1150-1152: Update the key assignment in the watch request
construction after create_watch_request to base64-encode create_request.key
rather than the outer key variable, preserving the helper’s empty-key
substitution before sending the request.
- Line 34: Update the module-level websocket dependency around ws_client and
create_ws_watch_session: either declare lua-resty-websocket in the rockspec, or
move the require("resty.websocket.client") into create_ws_watch_session so
resty.etcd.v3 can load without the optional dependency. Prefer lazy-loading if
websocket support is only needed for watch sessions.
- Around line 1169-1192: Declare and pin the lua-resty-websocket dependency to
version 0.10 or later in the project’s rockspec configuration. Preserve the
existing ws_client:new options in the WebSocket connection flow so server_name,
client_cert, and client_priv_key are honored.
---
Nitpick comments:
In `@lib/resty/etcd/v3.lua`:
- Around line 1089-1114: Update the websocket receive loop around
ws:recv_frame() to enforce one total timeout deadline for the recv operation,
rather than restarting the full per-call timeout after pings, pongs, or
fragmented frames. Track the deadline and reduce each iteration’s socket timeout
accordingly, returning "timeout" once it expires while preserving existing frame
assembly and close/error handling.
In `@t/v3/ws_watch.t`:
- Around line 15-56: Extend the websocket watch test suite beyond plaintext ws
by adding coverage for JWT authentication via the Sec-WebSocket-Protocol header
and HTTPS endpoints mapped to wss. Exercise create_ws_watch_session with mTLS
options and verify both authentication and TLS settings are preserved through
the connection.
- Around line 195-201: Isolate the destructive physical compaction in the watch
test around the compaction request so it cannot affect the shared etcd instance
used by other tests. Run this scenario against a dedicated etcd instance, or
move it into a separately gated test file with explicit setup and teardown,
while preserving the existing compaction assertion.
🪄 Autofix
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: Organization UI
Review profile: CHILL
Plan: Pro
Run ID: 7f7473fb-3a80-4a15-8d7a-c959953dd1e9
📒 Files selected for processing (3)
api_v3.mdlib/resty/etcd/v3.luat/v3/ws_watch.t
| 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") |
There was a problem hiding this comment.
🩺 Stability & Availability | 🟠 Major | ⚡ Quick win
🧩 Analysis chain
🏁 Script executed:
#!/bin/bash
# List rockspec files and their dependency blocks.
fd -e rockspec --exec sh -c 'echo "== {}"; cat -n "{}"'
# Check whether the websocket lib is referenced anywhere in packaging or CI.
rg -n 'lua-resty-websocket|resty\.websocket' --glob '!lib/**'Repository: api7/lua-resty-etcd
Length of output: 41762
🏁 Script executed:
#!/bin/bash
set -eu
echo '== websocket require and module header =='
cat -n lib/resty/etcd/v3.lua | sed -n '1,70p'
echo '== module imports and v3 call sites =='
rg -n 'require\(|resty\.etcd\.v3|create_ws_watch_session|ws_client' lib spec t 2>/dev/null || true
echo '== current rockspec candidates =='
fd -e rockspec -x sh -c 'grep -Hn -A8 -B2 "^dependencies" "$1"' sh {}Repository: api7/lua-resty-etcd
Length of output: 35786
Declare lua-resty-websocket or lazy-load the client.
resty.etcd loads resty.etcd.v3, which requires resty.websocket.client at module load time. Without the dependency, existing API calls fail before they run. Add lua-resty-websocket to the rockspec, or move the require into create_ws_watch_session.
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@lib/resty/etcd/v3.lua` at line 34, Update the module-level websocket
dependency around ws_client and create_ws_watch_session: either declare
lua-resty-websocket in the rockspec, or move the
require("resty.websocket.client") into create_ws_watch_session so resty.etcd.v3
can load without the optional dependency. Prefer lazy-loading if websocket
support is only needed for watch sessions.
| local create_request = create_watch_request(key, attr) | ||
| create_request.key = encode_base64(key) | ||
| create_request.range_end = encode_base64(attr.range_end) |
There was a problem hiding this comment.
🎯 Functional Correctness | 🟡 Minor | ⚡ Quick win
Encode the key produced by create_watch_request.
create_watch_request replaces an empty key with str_char(0). Line 1151 base64-encodes the outer key variable instead, so that substitution is discarded and an empty key is sent as "". Encode the value already stored in create_request.
🐛 Proposed fix
local create_request = create_watch_request(key, attr)
- create_request.key = encode_base64(key)
- create_request.range_end = encode_base64(attr.range_end)
+ create_request.key = encode_base64(create_request.key)
+ create_request.range_end = encode_base64(create_request.range_end or "")📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
| local create_request = create_watch_request(key, attr) | |
| create_request.key = encode_base64(key) | |
| create_request.range_end = encode_base64(attr.range_end) | |
| local create_request = create_watch_request(key, attr) | |
| create_request.key = encode_base64(create_request.key) | |
| create_request.range_end = encode_base64(create_request.range_end or "") |
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@lib/resty/etcd/v3.lua` around lines 1150 - 1152, Update the key assignment in
the watch request construction after create_watch_request to base64-encode
create_request.key rather than the outer key variable, preserving the helper’s
empty-key substitution before sending the request.
| 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 |
There was a problem hiding this comment.
🔒 Security & Privacy | 🟠 Major | ⚡ Quick win
🧩 Analysis chain
🏁 Script executed:
#!/bin/bash
# 2. endpoint field names produced by the endpoint constructor / choose_endpoint.
rg -n -C6 'api_prefix|http_host\s*=|address\s*=' lib/resty/etcd/ | sed -n '1,120p'
# 3. health_check usage symmetry.
rg -n -C3 'health_check\.(report_failure|report_success)' lib/resty/etcd/Repository: api7/lua-resty-etcd
Length of output: 10264
🌐 Web query:
lua-resty-websocket client connect options client_cert client_priv_key server_name ssl_verify supported version
💡 Result:
In lua-resty-websocket, the client:connect method accepts an optional table of parameters to configure the connection, including support for SSL/TLS settings [1][2]. The following options are supported: ssl_verify: A boolean that specifies whether to perform SSL certificate verification during the SSL handshake when using the wss:// scheme [3]. client_cert: Specifies a client certificate chain cdata object to be used during the TLS handshake [3]. This requires the use of the ngx.ssl.parse_pem_cert function from lua-resty-core to create the object [3]. If client_cert is provided, client_priv_key must also be provided [4][3]. client_priv_key: Specifies the private key corresponding to the client_cert option [3]. This object can be created using the ngx.ssl.parse_pem_priv_key function from lua-resty-core [3]. server_name: Specifies the server name (SNI) to be used during the SSL/TLS handshake with the remote server [5]. Additional supported connection options include: protocols: Used to specify the Sec-WebSocket-Protocol header [4]. origin: Used to specify the Origin header [4]. pool and pool_size: Control connection pooling behavior [1][2]. backlog: Sets a limit on the number of waiting connect operations when the pool is full [1][3]. Regarding supported versions, the library generally requires OpenResty with an Nginx version that supports the necessary cosocket and SSL features (specifically ngx_lua 0.9.11+ is required for SSL/wss support) [4]. Version-specific features such as pool and backlog management were introduced in v0.10.14 [1][5].
Citations:
- 1: https://github.com/openresty/lua-resty-websocket/blob/master/README.markdown
- 2: https://opm.openresty.org/package/openresty/lua-resty-websocket/
- 3: https://nginx-extras.getpagespeed.com/lua/websocket/
- 4: https://github.com/openresty/lua-resty-websocket/blob/master/lib/resty/websocket/client.lua
- 5: https://github.com/openresty/lua-resty-websocket
🏁 Script executed:
#!/bin/bash
set -eu
printf '%s\n' '--- dependency declarations ---'
rg -n -C4 'lua-resty-websocket|resty.websocket|websocket' \
.github Makefile README.md *.rockspec *.rock 2>/dev/null || true
printf '%s\n' '--- health-check implementation and success API ---'
fd -t f . lib | sort | grep -E 'health|etcd|resty'
rg -n -C8 'function .*report_(success|failure)|report_success|report_failure' lib
printf '%s\n' '--- endpoint selection and websocket session context ---'
rg -n -C12 'choose_endpoint|create_ws_watch_session|ws_client:new|ws:connect' lib/resty/etcd/v3.luaRepository: api7/lua-resty-etcd
Length of output: 16960
🏁 Script executed:
#!/bin/bash
set -eu
printf '%s\n' '--- tracked dependency and packaging files ---'
git ls-files | grep -Ei '(^|/)(rockspec|rock|manifest|makefile|cmake|docker|deps|requirements|.*lock)|websocket' | sed -n '1,160p'
printf '%s\n' '--- all websocket dependency references ---'
rg -n -i -C3 'lua-resty-websocket|resty\.websocket|websocket' . --glob '!*.md' --glob '!t/data/**' | sed -n '1,220p'
printf '%s\n' '--- health-check module ---'
cat -n lib/resty/etcd/health_check.lua | sed -n '1,190p'
printf '%s\n' '--- websocket watch call path after connect ---'
cat -n lib/resty/etcd/v3.lua | sed -n '1128,1235p'Repository: api7/lua-resty-etcd
Length of output: 15723
🏁 Script executed:
#!/bin/bash
set -eu
printf '%s\n' '--- websocket dependency declarations in rockspecs ---'
for f in $(git ls-files 'rockspec/*.rockspec'); do
if grep -qi 'websocket' "$f"; then
printf '\n[%s]\n' "$f"
grep -in -C3 'websocket' "$f"
fi
done
printf '%s\n' '--- current build and test dependency metadata ---'
cat -n Makefile | sed -n '1,180p'
cat -n rockspec/lua-resty-etcd-master-0.1-0.rockspec | sed -n '1,180p'
printf '%s\n' '--- repository revision and changed-file context ---'
git status --short
git branch --show-current
git rev-parse --short HEADRepository: api7/lua-resty-etcd
Length of output: 2781
🌐 Web query:
site:github.com/openresty/lua-resty-websocket "max_payload_len" "client_priv_key"
💡 Result:
In the context of the openresty/lua-resty-websocket library, max_payload_len and client_priv_key are configuration options used when initializing WebSocket connections [1][2]. max_payload_len This parameter defines the maximum length of the payload allowed for both sending and receiving WebSocket frames [2][3]. It defaults to 65535 [2][4]. When a frame exceeds this limit, the library will reject it [5]. Related configuration options include max_recv_len (defaults to max_payload_len) and max_send_len (defaults to max_payload_len) [1][2][4]. client_priv_key This parameter specifies the private key used during the TLS handshake for secure (WSS) WebSocket connections [2][3]. It must be provided alongside the client_cert option [1][2]. The private key object is typically created using the ngx.ssl.parse_pem_priv_key function from the lua-resty-core library [2][3]. If client_cert is provided, the library asserts that client_priv_key is also present [1]. It is subsequently used in the socket's TLS configuration (e.g., via sock:setclientcert) during the handshake process [1][6].
Citations:
- 1: https://github.com/openresty/lua-resty-websocket/blob/master/lib/resty/websocket/client.lua
- 2: https://github.com/openresty/lua-resty-websocket/blob/master/README.markdown
- 3: https://github.com/openresty/lua-resty-websocket
- 4: https://github.com/openresty/lua-resty-websocket/blob/master/lib/resty/websocket/server.lua
- 5: https://github.com/openresty/lua-resty-websocket/blob/master/lib/resty/websocket/protocol.lua
- 6: ssl session repeated sslhandshake openresty/lua-resty-websocket#79
🌐 Web query:
site:github.com/openresty/lua-resty-websocket/releases lua-resty-websocket version client_priv_key server_name
💡 Result:
The lua-resty-websocket library does not natively support specific configuration parameters named client_priv_key or server_name in its connection methods [1]. Based on the available documentation, there is no evidence that these parameters exist within the library's API. Generally, SSL/TLS configurations for OpenResty libraries are handled by the underlying lua-resty-core or Nginx stream/http SSL directives rather than the websocket library itself.
Citations:
🏁 Script executed:
#!/bin/bash
set -eu
printf '%s\n' '--- upstream websocket client option implementation ---'
curl -fsSL https://raw.githubusercontent.com/openresty/lua-resty-websocket/master/lib/resty/websocket/client.lua \
| nl -ba | grep -n -C5 -E 'max_payload_len|ssl_verify|server_name|client_cert|client_priv_key|function _M:new|function _M:connect' \
| sed -n '1,240p'
printf '%s\n' '--- upstream websocket release tags ---'
curl -fsSL 'https://api.github.com/repos/openresty/lua-resty-websocket/tags?per_page=100' \
| jq -r '.[] | [.name, .commit.sha] | `@tsv`' | sed -n '1,120p'Repository: api7/lua-resty-etcd
Length of output: 1271
🏁 Script executed:
#!/bin/bash
set -eu
base='https://raw.githubusercontent.com/openresty/lua-resty-websocket'
for tag in v0.01 v0.02 v0.03 v0.04 v0.05 v0.06 v0.07 v0.08 v0.09 v0.10 v0.11 v0.12 v0.13 v0.14 master; do
url="$base/$tag/lib/resty/websocket/client.lua"
body=$(curl -fsSL "$url" 2>/dev/null || true)
printf '\n[%s]\n' "$tag"
printf '%s\n' "$body" | grep -n -E 'max_payload_len|ssl_verify|server_name|client_cert|client_priv_key|function .*connect|function .*new' || true
done
printf '\n--- README option references ---\n'
curl -fsSL "$base/master/README.markdown" \
| grep -n -C4 -E 'max_payload_len|ssl_verify|server_name|client_cert|client_priv_key' \
| sed -n '1,220p'Repository: api7/lua-resty-etcd
Length of output: 16777
Declare and pin lua-resty-websocket to v0.10 or later. The rockspecs do not declare this dependency. max_payload_len is widely supported, but server_name, client_cert, and client_priv_key are unsupported before v0.10 and are ignored by older clients, which can bypass configured SNI and mTLS. The endpoint fields are populated correctly, and health entries recover when fail_timeout expires; no report_success call is required.
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@lib/resty/etcd/v3.lua` around lines 1169 - 1192, Declare and pin the
lua-resty-websocket dependency to version 0.10 or later in the project’s
rockspec configuration. Preserve the existing ws_client:new options in the
WebSocket connection flow so server_name, client_cert, and client_priv_key are
honored.
1 participant
Why
etcd's JSON gateway is half-duplex over plain HTTP/1.1: Go's
net/httpserver drains the unread request body before writing any response bytes, so the existing/v3/watchstream is write-once. AWatchProgressRequestcan never be sent on a live watch, which blocks apache/apisix#13777 (keeping an idle watch's revision fresh so etcd compaction stops forcing full config reloads).etcd also wraps
/v3/in grpc-websocket-proxy (all supported versions), and over a WebSocket upgrade the same endpoint is genuinely full-duplex: one WatchRequest per text frame in, one WatchResponse per frame out. Verified against etcd 3.5.12 and 3.6.4: progress requests on an idle watched prefix are answered once the watcher is synced, live events keep flowing while the client write side stays open, and the reply revision is a true in-stream delivery barrier (progressIfSynconly answers when every watcher on the stream is synced).What
cli:create_ws_watch_session(dir, opts)built onresty.websocket.client(bundled with OpenResty). Returns a session with:recv(timeout): next WatchResponse, decoded identically towatchdirresponses (base64 + serializer handling shared semantics).request_progress(): sendsprogress_requeston the live stream; the reply arrives viarecv()as an events-free WatchResponse whoseheader.revisionis the barrier.close().Sec-WebSocket-Protocol: Bearer,<token>which the etcd proxy converts into theAuthorizationheader;wss/mTLS options are passed through.resty.websocket.clientaccepts any HTTP/1.1 status line as a handshake (bugfix: client: reject a handshake response whose status is not 101 openresty/lua-resty-websocket#104 fixes that upstream), so an intermediary that stripsUpgradewould otherwise look connected. Rejecting it at creation lets callers fall back towatchdir. The consumed response is replayed on the firstrecv().api_v3.md, tests int/v3/ws_watch.t(created ack + live event on an open stream, idle-prefix progress barrier following foreign writes, resume-from-barrier surviving compaction with a stale-revision control, and rejection of a non-upgrading endpoint).Notes
WithMaxRespBodyBufferSize, so watch responses over 64KB need etcd >= 3.5.Summary by CodeRabbit