fix(watch): do not drop events when a read carries multiple responses

.github/workflows/ci.yml

@@ -22,7 +22,7 @@ jobs:
           - version: 3.4.0
             conf: Procfile-single-enable-mtls
 
-    runs-on: "ubuntu-20.04"
+    runs-on: "ubuntu-22.04"
     env:
       OPENRESTY_PREFIX: "/usr/local/openresty"
       AUTH_ENDPOINT_V3: "127.0.0.1:12379"
@@ -35,10 +35,12 @@ jobs:
       - uses: actions/checkout@v2
 
       - name: setup go
-        uses: actions/setup-go@v2.1.3
+        uses: actions/setup-go@v5
         with:
-          # goreman requires 1.17
-          go-version: "1.17"
+          # Keep this in sync with the goreman pin in the install step: it must
+          # be >= goreman's module `go` directive, otherwise the toolchain gets
+          # downloaded implicitly (see GOTOOLCHAIN=local note below).
+          go-version: "1.22"
 
       - name: get dependencies
         run: sudo apt install -y cpanminus build-essential libncurses5-dev libreadline-dev libssl-dev perl
@@ -68,7 +70,12 @@ jobs:
           tar xf etcd-v$ETCD_VER-linux-amd64.tar.gz
           # run etcd local cluster, startup at localhost:2379, localhost:22379, and localhost:32379
           # see more https://github.com/etcd-io/etcd/blob/master/Documentation/dev-guide/local_cluster.md
-          go get github.com/mattn/goreman
+          # Pin goreman to a fixed release for reproducible CI. v0.3.15 declares
+          # `go 1.19` and pulls golang.org/x/sys v0.6.0, both of which build with
+          # the pinned toolchain above. GOTOOLCHAIN=local forbids Go from
+          # silently downloading a newer toolchain (goreman >= v0.3.19 declares
+          # `go 1.25`, which would otherwise defeat the setup-go pin).
+          GOTOOLCHAIN=local go install github.com/mattn/goreman@v0.3.15
 
       - name: script
         if: matrix.conf != 'Procfile-single-enable-mtls'

lib/resty/etcd/v3.lua

@@ -939,7 +939,12 @@ local function request_chunk(self, method, path, opts, timeout)
             end
         end
 
-        if #all_events > 1 then
+        -- one read may carry several watch responses separated by "\n", and the
+        -- last one is not necessarily the one holding the events, e.g.
+        -- [response with events, progress notification]. `body` points at the
+        -- last decoded response, so the collected events must always be
+        -- attached to it, not only when more than one event was collected.
+        if #all_events > 0 and body.result then
             body.result.events = all_events
         end
         return body

t/v3/tls.t

@@ -103,7 +103,7 @@ GET /t
 --- no_error_log
 [error]
 --- response_body eval
-qr/18: self signed certificate/
+qr/18: self[ -]signed certificate/
 
 
 

t/v3/watch.t

@@ -0,0 +1,223 @@
+use Test::Nginx::Socket::Lua;
+
+log_level('info');
+no_long_string();
+repeat_each(1);
+plan 'no_plan';
+
+# etcd coalesces watch responses into a single body read whenever they are
+# written close enough together (very common once progress_notify is on), and a
+# real etcd cannot be driven into that state reliably -- so each test below uses
+# a fake endpoint that emits a fixed, hand-crafted coalesced read. The point of
+# the fix under test is that the response holding the events is NOT necessarily
+# the last one in the read, so we exercise both orderings and an interleaving.
+
+# [event, progress]: event (rev 7) followed by a progress notification (rev 9).
+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;;';
+    server {
+        listen 1985;
+        location /v3/watch {
+            content_by_lua_block {
+                ngx.print('{"result":{"header":{"revision":"7"},"events":' ..
+                          '[{"type":"PUT","kv":{"key":"L3Rlc3Q=","value":"ImFiYyI=",' ..
+                          '"mod_revision":"7"}}]}}\n' ..
+                          '{"result":{"header":{"revision":"9"}}}\n')
+            }
+        }
+    }
+_EOC_
+
+# [progress, event]: progress notification (rev 7) followed by an event (rev 9).
+our $HttpConfigProgressFirst = <<'_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('{"result":{"header":{"revision":"7"}}}\n' ..
+                          '{"result":{"header":{"revision":"9"},"events":' ..
+                          '[{"type":"PUT","kv":{"key":"L3Rlc3Q=","value":"ImFiYyI=",' ..
+                          '"mod_revision":"9"}}]}}\n')
+            }
+        }
+    }
+_EOC_
+
+# [event, progress, event]: an event (rev 7), a progress notification (rev 8),
+# and a second event (rev 9) all in one read.
+our $HttpConfigInterleaved = <<'_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('{"result":{"header":{"revision":"7"},"events":' ..
+                          '[{"type":"PUT","kv":{"key":"L3Rlc3Q=","value":"ImFiYyI=",' ..
+                          '"mod_revision":"7"}}]}}\n' ..
+                          '{"result":{"header":{"revision":"8"}}}\n' ..
+                          '{"result":{"header":{"revision":"9"},"events":' ..
+                          '[{"type":"PUT","kv":{"key":"L3Rlc3Qy","value":"ImJjZCI=",' ..
+                          '"mod_revision":"9"}}]}}\n')
+            }
+        }
+    }
+_EOC_
+
+run_tests();
+
+__DATA__
+
+=== TEST 1: a coalesced read keeps the events of every response it contains
+--- http_config eval: $::HttpConfig
+--- config
+    location /t {
+        content_by_lua_block {
+            local etcd, err = require "resty.etcd" .new({
+                protocol = "v3",
+                http_host = "http://127.0.0.1:1985",
+            })
+            if not etcd then
+                ngx.say("failed to new: ", err)
+                return
+            end
+
+            local res_func, err = etcd:watchdir("/test", {timeout = 5})
+            if not res_func then
+                ngx.say("failed to watchdir: ", err)
+                return
+            end
+
+            local res, err = res_func()
+            if not res then
+                ngx.say("failed to read watch: ", err)
+                return
+            end
+
+            -- The read is [event(rev 7), progress(rev 9)]. We report the
+            -- progress notification's revision (9, the last response in the
+            -- read) while still delivering the rev-7 event: a progress
+            -- notification guarantees nothing happened between the event and
+            -- it, so the caller can safely advance start_revision past 9.
+            ngx.say("revision: ", res.result.header.revision)
+            local events = res.result.events
+            ngx.say("events: ", events and #events or 0)
+            if events and events[1] then
+                ngx.say("key: ", events[1].kv.key)
+                ngx.say("value: ", events[1].kv.value)
+            end
+        }
+    }
+--- request
+GET /t
+--- response_body
+revision: 9
+events: 1
+key: /test
+value: abc
+--- no_error_log
+[error]
+
+
+
+=== TEST 2: a coalesced read keeps the event when it is the last response
+--- http_config eval: $::HttpConfigProgressFirst
+--- config
+    location /t {
+        content_by_lua_block {
+            local etcd, err = require "resty.etcd" .new({
+                protocol = "v3",
+                http_host = "http://127.0.0.1:1985",
+            })
+            if not etcd then
+                ngx.say("failed to new: ", err)
+                return
+            end
+
+            local res_func, err = etcd:watchdir("/test", {timeout = 5})
+            if not res_func then
+                ngx.say("failed to watchdir: ", err)
+                return
+            end
+
+            local res, err = res_func()
+            if not res then
+                ngx.say("failed to read watch: ", err)
+                return
+            end
+
+            -- The read is [progress(rev 7), event(rev 9)]. The event is the
+            -- last response, so its revision (9) is reported and its event
+            -- delivered.
+            ngx.say("revision: ", res.result.header.revision)
+            local events = res.result.events
+            ngx.say("events: ", events and #events or 0)
+            if events and events[1] then
+                ngx.say("key: ", events[1].kv.key)
+                ngx.say("value: ", events[1].kv.value)
+            end
+        }
+    }
+--- request
+GET /t
+--- response_body
+revision: 9
+events: 1
+key: /test
+value: abc
+--- no_error_log
+[error]
+
+
+
+=== TEST 3: a coalesced read keeps events from every response when interleaved
+--- http_config eval: $::HttpConfigInterleaved
+--- config
+    location /t {
+        content_by_lua_block {
+            local etcd, err = require "resty.etcd" .new({
+                protocol = "v3",
+                http_host = "http://127.0.0.1:1985",
+            })
+            if not etcd then
+                ngx.say("failed to new: ", err)
+                return
+            end
+
+            local res_func, err = etcd:watchdir("/test", {timeout = 5})
+            if not res_func then
+                ngx.say("failed to watchdir: ", err)
+                return
+            end
+
+            local res, err = res_func()
+            if not res then
+                ngx.say("failed to read watch: ", err)
+                return
+            end
+
+            -- The read is [event(rev 7), progress(rev 8), event(rev 9)]. Both
+            -- events must survive, in order, under the highest revision (9).
+            ngx.say("revision: ", res.result.header.revision)
+            local events = res.result.events
+            ngx.say("events: ", events and #events or 0)
+            if events then
+                for i = 1, #events do
+                    ngx.say("key", i, ": ", events[i].kv.key,
+                            " value", i, ": ", events[i].kv.value)
+                end
+            end
+        }
+    }
+--- request
+GET /t
+--- response_body
+revision: 9
+events: 2
+key1: /test value1: abc
+key2: /test2 value2: bcd
+--- no_error_log
+[error]