From 6dfc760e2df5c075ebec38c6614bcf37c0b1fc60 Mon Sep 17 00:00:00 2001 From: root Date: Tue, 9 Jun 2026 15:28:45 +0800 Subject: [PATCH 1/3] test: drop test-only production hooks; enforce deterministic async testing MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Replace bespoke test instrumentation and timing-based test synchronization with the established observable-event WaitGroup discipline, and separate internal-mechanism unit tests from external-API e2e tests. Production: - Remove VVManager.SetBumpObserver/onBump (the bespoke test-only hook) and revert increment() to its plain `defer mu.Unlock()`. The VV bump is already observable through the standard storage AfterWrite (it persists to pivot/vv/) and the external /activity endpoint, so no special hook is warranted. Tests: - clock_drift: assert the data-sync effect (a present-time write wins over a future-timestamped one — only possible if VV ordering, not the wall clock, governs) instead of internal VV counters; existing NodeWg/PivotWg already synchronize it, so no wait-on-bump is needed. - offline_sync: TestOfflineNodeWriteAndSync asserts the data effect; TestVersionVectorActivityEndpoint waits on the VV-persistence storage write (standard AfterWrite, via setupOfflineServers' opt-in observer) and asserts the VV through the external /activity endpoint — no VVManager polling. - handlers_internal + vv_path_scope: replace time.Sleep / Eventually(pendingLen) with a WaitGroup signalled when the watch callback finishes processing each event (the path that could double-bump). Reframe the HandlerWriteTracker unit tests to assert the public Mark/Consume/ConsumeBumpSkip contract, and remove the test-only pendingLen/bumpPendingLen accessors. - version_e2e: wait on the node's pivot/status feed (the external cluster-status API the health check broadcasts to) instead of polling GetPivotInfo on a sleep loop. - vv_merge: wait on the mock leader's request handler instead of Eventually(len(received)). Justified unchanged (race/stress or correct failure-deadline patterns, not result synchronization): nodehealth_race_test (race tests — sleeps are concurrency windows for -race and negative assertions, real sync is via channels/WaitGroups), remote_context_test (channel-based success sync; time.After is the failure deadline), trigger/syncer_init (sync.Once first-hit on the unbounded coalescer stream). Verified: full -race suite green; 40x serial + 60x parallel -race of the converted tests with zero hangs or flakes. Co-Authored-By: Claude Opus 4.8 (1M context) --- clock_drift_test.go | 49 ++------------ handlers_internal_test.go | 136 ++++++++++++++++++++------------------ offline_sync_test.go | 113 +++++++++++++++---------------- version_e2e_test.go | 64 +++++++++++------- version_vector.go | 32 +-------- vv_merge_test.go | 16 ++--- vv_path_scope_test.go | 52 +++++++++------ 7 files changed, 215 insertions(+), 247 deletions(-) diff --git a/clock_drift_test.go b/clock_drift_test.go index 6d9112a..2c1568b 100644 --- a/clock_drift_test.go +++ b/clock_drift_test.go @@ -64,25 +64,13 @@ func TestClockDriftScenario(t *testing.T) { servers := setupOfflineServers(t) defer servers.Close() - pivotInstance := pivot.GetInstance(servers.Pivot) - nodeInstance := pivot.GetInstance(servers.Node) - require.NotNil(t, pivotInstance.VVManager) - require.NotNil(t, nodeInstance.VVManager) - - // The pivot Set handler increments its VV counter AFTER db.SetWithMeta - // returns — i.e. after the storage AfterWrite that drives PivotWg.Done(). - // Waiting on PivotWg alone proves the pushed data landed, not that the - // counter bump did, so reading the VV right after PivotWg.Wait() races the - // bump (empty/stale leader counter). pivotBump fires once per leader-counter - // increment for "policies"; pairing Add(1)/Wait() with each push makes the - // counter reads below deterministic. Disarmed before phase 4, whose - // pivot-side write bumps via a path this test does not assert on. - var pivotBump sync.WaitGroup - pivotInstance.VVManager.SetBumpObserver(func(baseKey string) { - if baseKey == "policies" { - pivotBump.Done() - } - }) + // This test asserts the OBSERVABLE EFFECT of version-vector ordering — that a + // present-time write wins over an earlier future-timestamped one — through the + // synced data, not through pivot's internal VV counters. The decisive proof is + // in phase 3: phase-2 carries a numerically SMALLER Updated than phase-1, so it + // can only win if logical (VV) ordering, not the wall clock, governs. Reads are + // synchronised by the storage-write WaitGroups (NodeWg/PivotWg) the harness + // exposes — no reaching into VVManager. sixHours := int64(6 * 60 * 60 * 1000000000) // 6 hours in nanoseconds now := time.Now().UnixNano() @@ -101,12 +89,10 @@ func TestClockDriftScenario(t *testing.T) { // in the wire-format body. servers.NodeWg.Add(1) servers.PivotWg.Add(1) - pivotBump.Add(1) _, err := servers.NodePolicies.SetWithMeta("policies", futureBytes, futureTimestamp, futureTimestamp) require.NoError(t, err) servers.NodeWg.Wait() servers.PivotWg.Wait() - pivotBump.Wait() // Phase 1 must reach pivot with the future Updated intact. phase1Obj, err := servers.PivotPolicies.Get("policies") @@ -117,12 +103,6 @@ func TestClockDriftScenario(t *testing.T) { require.Equal(t, futureTimestamp, phase1Obj.Updated, "Pivot's stored Updated should reflect the node-pushed future timestamp") - // Record VV state after phase 1. - pivotVV1 := pivotInstance.VVManager.Get("policies") - t.Logf("Pivot VV after phase 1: %v", pivotVV1) - require.Greater(t, pivotVV1["leader"], int64(0), - "phase-1 push should have bumped pivot's leader counter") - // === Phase 2: Clock goes back to ACTUAL time === t.Log("Phase 2: Clock returns to actual time, new update happens") @@ -139,12 +119,10 @@ func TestClockDriftScenario(t *testing.T) { // Updated. servers.NodeWg.Add(1) servers.PivotWg.Add(1) - pivotBump.Add(1) _, err = servers.NodePolicies.SetWithMeta("policies", currentBytes, currentTimestamp, currentTimestamp) require.NoError(t, err) servers.NodeWg.Wait() servers.PivotWg.Wait() - pivotBump.Wait() // === Phase 3: Verify VV prevents the "future" data from winning === t.Log("Phase 3: Verifying Version Vector prevents future-timestamp overwrite") @@ -160,22 +138,9 @@ func TestClockDriftScenario(t *testing.T) { require.Equal(t, currentTimestamp, pivotObj.Updated, "pivot should have phase-2's present-time Updated, proving the wall clock did not gate this") - pivotVV2 := pivotInstance.VVManager.Get("policies") - t.Logf("Pivot VV after phase 2: %v", pivotVV2) - require.Greater(t, pivotVV2["leader"], pivotVV1["leader"], - "VV counter should have incremented, proving logical ordering over timestamps") - // === Phase 4: Simulate reverse sync (pivot -> node) to ensure no regression === t.Log("Phase 4: Verify pivot -> node sync also respects VV") - // Phase 4 writes to pivot directly (HTTP POST on its own storage), which - // bumps the pivot VV via the AfterWriteOp path rather than the Set handler. - // This test makes no VV-counter assertions past here, so disarm the observer - // — leaving it armed would Done() pivotBump with no matching Add(). The - // disarm happens-after phase 2's pivotBump.Wait(), so no in-flight bump is - // lost, and SetBumpObserver serialises with increment under the VV mutex. - pivotInstance.VVManager.SetBumpObserver(nil) - // Write on pivot via HTTP, then manually trigger node sync // (Pivot doesn't auto-sync to node since node isn't registered). servers.PivotWg.Add(1) diff --git a/handlers_internal_test.go b/handlers_internal_test.go index ff88776..8e70863 100644 --- a/handlers_internal_test.go +++ b/handlers_internal_test.go @@ -6,9 +6,11 @@ import ( "net/http/httptest" "strconv" "strings" + "sync" "testing" "time" + "github.com/benitogf/ooo/key" "github.com/benitogf/ooo/meta" "github.com/benitogf/ooo/monotonic" "github.com/benitogf/ooo/storage" @@ -16,14 +18,21 @@ import ( "github.com/stretchr/testify/require" ) -// pendingLen exposes the tracker's pending count for tests that need to -// wait for the watch goroutine to drain. Lives in the test file so it -// doesn't leak into the production API surface — production code has no -// legitimate need to introspect the tracker. -func (t *HandlerWriteTracker) pendingLen() int { - t.mu.Lock() - defer t.mu.Unlock() - return len(t.pending) +// watchProcessed wraps a storage sync callback so wg.Done() fires after the +// watch goroutine finishes processing each event whose key matches glob. It +// lets these handler+callback tests wait deterministically for the async +// callback to drain the writes they care about — the watch goroutine is what +// could double-bump the VV, so observing its completion is the right signal — +// without sleeps or polling internal tracker counters. Bumps that persist the +// VV write to StoragePrefix keys don't match the data glob, so they don't +// inflate the count. +func watchProcessed(db storage.Database, cb StorageSyncCallback, glob string, wg *sync.WaitGroup) { + storage.WatchWithCallback(db, func(e storage.Event) { + cb(e) + if key.Match(glob, e.Key) { + wg.Done() + } + }) } // failingTombstoneStorage forces Set on the pivot tombstone prefix to fail. @@ -129,12 +138,13 @@ func TestSetVVIncrementsExactlyOnce(t *testing.T) { vvm := NewVVManager(db, "leader") keys := []Key{{Path: "things/*", Database: db}} instance := &Instance{VVManager: vvm} - storage.WatchWithCallback(db, makeStorageSync(StorageSyncConfig{ + var processed sync.WaitGroup + watchProcessed(db, makeStorageSync(StorageSyncConfig{ Keys: keys, GetNodes: func() []string { return nil }, HandlerTracker: tracker, Instance: instance, - })) + }), "things/*", &processed) handler := Set(db, "things", tracker, vvm, nil) @@ -148,14 +158,17 @@ func TestSetVVIncrementsExactlyOnce(t *testing.T) { require.Equal(t, 200, w.Code) } + // Wait for the watch goroutine to process each write before reading the + // counter: it is the path that could double-bump, so its completion (not a + // sleep) is the deterministic signal that the count has settled. + processed.Add(1) doSet("abc") - // Settle: callback runs in the watch goroutine, so we need to let any - // stray increment race past the handler's own Get before reading. - time.Sleep(100 * time.Millisecond) + processed.Wait() require.Equal(t, int64(1), vvm.Get("things")["leader"], "first Set must bump leader counter to exactly 1") + processed.Add(1) doSet("abc") - time.Sleep(100 * time.Millisecond) + processed.Wait() require.Equal(t, int64(2), vvm.Get("things")["leader"], "second Set must bump leader counter to exactly 2") } @@ -176,13 +189,14 @@ func TestSetVVIncrementsExactlyOnceNodeRole(t *testing.T) { keys := []Key{{Path: "things/*", Database: db}} instance := &Instance{VVManager: vvm} - storage.WatchWithCallback(db, makeStorageSync(StorageSyncConfig{ + var processed sync.WaitGroup + watchProcessed(db, makeStorageSync(StorageSyncConfig{ Keys: keys, ConfigClusterURL: "127.0.0.1:8000", // non-empty -> node mode GetNodes: func() []string { return nil }, HandlerTracker: tracker, Instance: instance, - })) + }), "things/*", &processed) handler := Set(db, "things", tracker, vvm, nil) body := strings.NewReader(`{"created":0,"updated":0,"index":"abc","path":"things/abc","data":"e30="}`) @@ -190,9 +204,10 @@ func TestSetVVIncrementsExactlyOnceNodeRole(t *testing.T) { req = mux.SetURLVars(req, map[string]string{"index": "abc"}) req.Header.Set("Content-Type", "application/json") w := httptest.NewRecorder() + processed.Add(1) handler(w, req) require.Equal(t, 200, w.Code) - time.Sleep(100 * time.Millisecond) + processed.Wait() // watch goroutine processed the write (no second bump) got := vvm.Get("things") require.Equal(t, int64(1), got["127.0.0.1:9999"], @@ -220,16 +235,18 @@ func TestSetVVIncrementsExactlyOnceUnderBurst(t *testing.T) { vvm := NewVVManager(db, "leader") keys := []Key{{Path: "things/*", Database: db}} instance := &Instance{VVManager: vvm} - storage.WatchWithCallback(db, makeStorageSync(StorageSyncConfig{ + var processed sync.WaitGroup + watchProcessed(db, makeStorageSync(StorageSyncConfig{ Keys: keys, GetNodes: func() []string { return nil }, HandlerTracker: tracker, Instance: instance, - })) + }), "things/*", &processed) handler := Set(db, "things", tracker, vvm, nil) const burst = 10 + processed.Add(burst) for range burst { body := strings.NewReader(`{"created":0,"updated":0,"index":"abc","path":"things/abc","data":"e30="}`) req := httptest.NewRequest("POST", "/_pivot/pivot/things/abc", body) @@ -240,10 +257,10 @@ func TestSetVVIncrementsExactlyOnceUnderBurst(t *testing.T) { require.Equal(t, 200, w.Code) } - // Wait deterministically for the watch goroutine to drain every event; - // when tracker.pendingLen() hits zero, every Mark has been Consumed. - require.Eventually(t, func() bool { return tracker.pendingLen() == 0 }, 2*time.Second, 5*time.Millisecond, - "watch goroutine never drained all %d events", burst) + // Wait for the watch goroutine to process all burst events (each Consumes + // one Mark). Its completion is the deterministic signal that every event + // drained — no polling of internal tracker state. + processed.Wait() got := vvm.Get("things")["leader"] require.Equal(t, int64(burst), got, @@ -267,17 +284,20 @@ func TestDeleteDoesNotLeakHandlerMarks(t *testing.T) { vvm := NewVVManager(db, "leader") keys := []Key{{Path: "things/*", Database: db}} instance := &Instance{VVManager: vvm} - storage.WatchWithCallback(db, makeStorageSync(StorageSyncConfig{ + var processed sync.WaitGroup + watchProcessed(db, makeStorageSync(StorageSyncConfig{ Keys: keys, GetNodes: func() []string { return nil }, HandlerTracker: tracker, Instance: instance, - })) + }), "things/*", &processed) // Seed a few items so each Delete actually has something to remove. // Seeding goes through db.SetWithMeta directly (no Mark), so each event - // flows through the callback's empty-tracker branch and bumps VV. + // flows through the callback's empty-tracker branch and bumps VV once at + // path scope. Wait for all seed events to be processed (deterministic). indices := []string{"a", "b", "c"} + processed.Add(len(indices)) for _, idx := range indices { nowUnix := time.Now().UTC().UnixNano() obj := meta.Object{Created: nowUnix, Updated: nowUnix, Index: idx, Path: "things/" + idx, Data: []byte(`{"v":1}`)} @@ -286,16 +306,12 @@ func TestDeleteDoesNotLeakHandlerMarks(t *testing.T) { _, err = db.SetWithMeta("things/"+idx, body, nowUnix, nowUnix) require.NoError(t, err) } - // Wait until the seed VV bumps actually landed via the callback. Each - // seed bumps path-scope "things" once (the callback increments at the - // matched key's base, not the storage event's full key), so after three - // seeds the path-scope leader counter must be 3. - require.Eventually(t, func() bool { - return vvm.Get("things")["leader"] >= 3 - }, 2*time.Second, 5*time.Millisecond, "seed VV bumps never landed") + processed.Wait() + require.Equal(t, int64(len(indices)), vvm.Get("things")["leader"], "each seed must bump path-scope VV exactly once") handler := Delete(db, "things", tracker, vvm, nil) - for _, idx := range []string{"a", "b", "c"} { + processed.Add(len(indices)) + for _, idx := range indices { ts := strconv.FormatInt(time.Now().UTC().UnixNano(), 10) req := httptest.NewRequest("DELETE", "/_pivot/pivot/things/"+idx+"/"+ts, nil) req = mux.SetURLVars(req, map[string]string{"index": idx, "time": ts}) @@ -303,9 +319,17 @@ func TestDeleteDoesNotLeakHandlerMarks(t *testing.T) { handler(w, req) require.Equal(t, 200, w.Code) } - - require.Eventually(t, func() bool { return tracker.pendingLen() == 0 }, 2*time.Second, 5*time.Millisecond, - "tracker leaked entries after Deletes drained: Len=%d", tracker.pendingLen()) + processed.Wait() + + // No leaked handler marks. Post-fix the Delete handler Marks only the item + // key (Consumed by the item's del event the watch goroutine just processed), + // never the tombstone key. Asserted via the public Consume contract, not an + // internal length: a leaked tombstone Mark would make Consume return true, + // and any unconsumed item Mark likewise. + require.False(t, tracker.Consume(StoragePrefix+"things"), "Delete must not leak a tombstone-key handler mark") + for _, idx := range indices { + require.False(t, tracker.Consume("things/"+idx), "Delete item Mark must have been consumed by the watch goroutine") + } } // TestSetPostWriteSeesBumpedVV pins the in-handler ordering: the post-write @@ -477,14 +501,6 @@ func TestDeleteHappyPathStillCommitsBoth(t *testing.T) { require.Equal(t, deleteTS, strings.TrimSpace(string(tomb.Data)), "tombstone payload must be the delete timestamp") } -// bumpPendingLen mirrors pendingLen for the bump-skip counter so tests -// can assert it drains independently. -func (t *HandlerWriteTracker) bumpPendingLen() int { - t.mu.Lock() - defer t.mu.Unlock() - return len(t.bumpPending) -} - // TestHandlerWriteTracker_DualCounterConsumeSemantics pins that Mark // sets BOTH the fanout-skip (pending) and bump-skip (bumpPending) // counters, and each is consumed by exactly one consumer without @@ -495,25 +511,20 @@ func TestHandlerWriteTracker_DualCounterConsumeSemantics(t *testing.T) { tr := NewHandlerWriteTracker() const k = "things/abc" - // Mark once: both counters carry the key. + // Mark sets both the bump-skip and fanout-skip counters. Assert the contract + // through the public consume methods (not an internal length): each counter + // is consumable exactly once, and consuming one does not deprive the other. tr.Mark(k) - require.Equal(t, 1, tr.pendingLen(), "Mark must set fanout-skip counter") - require.Equal(t, 1, tr.bumpPendingLen(), "Mark must set bump-skip counter") - // AfterWrite consumes its own counter. The watch goroutine's - // counter is untouched — the two consumers don't deprive each other. + // The bump-skip consumer drains its own counter; consuming, not peeking, so a + // second consume returns false — the property that stops a stale mark from + // swallowing a later direct write's bump. require.True(t, tr.ConsumeBumpSkip(k), "first ConsumeBumpSkip sees the mark") - require.Equal(t, 0, tr.bumpPendingLen(), "ConsumeBumpSkip drains bump-skip counter") - require.Equal(t, 1, tr.pendingLen(), "fanout-skip counter unaffected by bump consume") - - // Consuming, not peeking: a second consume returns false. This is - // the property that prevents a stale mark from swallowing a later - // direct write's bump. require.False(t, tr.ConsumeBumpSkip(k), "second ConsumeBumpSkip must return false (consumed, not peeked)") - // Watch goroutine consumes its counter independently. - require.True(t, tr.Consume(k), "Consume sees the fanout-skip mark") - require.Equal(t, 0, tr.pendingLen(), "Consume drains fanout-skip counter") + // The fanout-skip mark is still present — the two counters are independent, + // so consuming bump-skip above did not drain it. + require.True(t, tr.Consume(k), "Consume still sees the fanout-skip mark after the bump-skip consume") require.False(t, tr.Consume(k), "second Consume must return false") } @@ -527,11 +538,10 @@ func TestHandlerWriteTracker_UnmarkClearsBothCounters(t *testing.T) { tr.Mark(k) tr.Unmark(k) - require.Equal(t, 0, tr.pendingLen(), "Unmark must clear fanout-skip counter") - require.Equal(t, 0, tr.bumpPendingLen(), "Unmark must clear bump-skip counter") - // After Unmark, neither consumer sees a mark — a subsequent direct - // write to the same key will correctly run its full path. + // After Unmark, neither consumer sees a mark — both counters were cleared, + // so a subsequent direct write to the same key runs its full path. Asserted + // through the public consume contract rather than an internal length. require.False(t, tr.ConsumeBumpSkip(k), "no bump-skip mark after Unmark") require.False(t, tr.Consume(k), "no fanout-skip mark after Unmark") } diff --git a/offline_sync_test.go b/offline_sync_test.go index f80c88b..7489598 100644 --- a/offline_sync_test.go +++ b/offline_sync_test.go @@ -79,7 +79,29 @@ type OfflineTestServers struct { NodeWg *sync.WaitGroup } -func setupOfflineServers(t *testing.T) *OfflineTestServers { +// offlineOpt configures setupOfflineServers before the servers start. +type offlineOpt func(*offlineConfig) + +type offlineConfig struct { + // pivotVVWrites, if set, has Done() called for every committed write to the + // pivot's "pivot/vv/policies" key — the leader's version-vector persistence. + // It rides the standard server.AfterWrite storage callback (wired before + // Start), so a test can wait on the leader's post-write VV bump by observing + // the storage event the bump produces, rather than polling VVManager. + pivotVVWrites *sync.WaitGroup +} + +// withPivotVVWrites wires wg to the leader's VV-persistence storage writes. +func withPivotVVWrites(wg *sync.WaitGroup) offlineOpt { + return func(c *offlineConfig) { c.pivotVVWrites = wg } +} + +func setupOfflineServers(t *testing.T, opts ...offlineOpt) *OfflineTestServers { + var cfg offlineConfig + for _, o := range opts { + o(&cfg) + } + pivotWg := &sync.WaitGroup{} nodeWg := &sync.WaitGroup{} @@ -164,6 +186,18 @@ func setupOfflineServers(t *testing.T) *OfflineTestServers { } }).Methods(http.MethodGet, http.MethodPost) + // Observe the leader's VV persistence (pivot/vv/policies) through the standard + // server.AfterWrite storage callback. pivot uses server.AfterWriteOp for its + // own bump, leaving server.AfterWrite free for this. Set before Start so the + // storage layer captures it; nil unless a test opted in via withPivotVVWrites. + if cfg.pivotVVWrites != nil { + pivotServer.AfterWrite = func(key string) { + if key == pivot.VVKeyPrefix+"policies" { + cfg.pivotVVWrites.Done() + } + } + } + pivotServer.Start("localhost:0") // Create node server (follower) @@ -251,55 +285,35 @@ func TestOfflineNodeWriteAndSync(t *testing.T) { t.Logf("Pivot policies storage active: %v", servers.PivotPolicies.Active()) t.Logf("Node policies storage active: %v", servers.NodePolicies.Active()) - // Get instances to access VVManager - pivotInstance := pivot.GetInstance(servers.Pivot) - nodeInstance := pivot.GetInstance(servers.Node) - require.NotNil(t, pivotInstance, "Pivot instance should exist") - require.NotNil(t, nodeInstance, "Node instance should exist") - require.NotNil(t, pivotInstance.VVManager, "Pivot should have VVManager") - require.NotNil(t, nodeInstance.VVManager, "Node should have VVManager") + // This test asserts the observable data-sync effect (a node write reaches the + // pivot and a pivot write reaches the node), synchronised by the storage-write + // WaitGroups — not pivot's internal VV counters. The VV machinery is exercised + // through its effect on the available API (GET /_pivot/activity) in + // TestVersionVectorActivityEndpoint. - // Phase 1: Node writes data via HTTP, syncs to pivot - // Expect: 1 node write (local) + 1 pivot write (from sync) + // Phase 1: a node write syncs to the pivot. NodeWg fires on the node-local + // write, PivotWg on the pivot-applied write; once both drain the synced data + // is observable on the pivot. servers.NodeWg.Add(1) servers.PivotWg.Add(1) - // Use HTTP POST to trigger proper AfterWrite callback payload := []byte(`{"value": "from-node"}`) resp, err := servers.Node.Client.Post("http://"+servers.Node.Address+"/policies", "application/json", bytes.NewBuffer(payload)) require.NoError(t, err) resp.Body.Close() require.Equal(t, http.StatusOK, resp.StatusCode) - // Wait for both writes to complete. The wg decrements fire from - // AfterWrite (synchronous with the storage Set), but the VV bump - // runs in the storage event callback on the watch goroutine — those - // two paths are decoupled. Poll the VV reads instead of asserting - // once and racing the callback. servers.NodeWg.Wait() servers.PivotWg.Wait() - require.Eventually(t, func() bool { - return len(nodeInstance.VVManager.Get("policies")) > 0 - }, 2*time.Second, 5*time.Millisecond, "node VV never bumped") - nodeVV := nodeInstance.VVManager.Get("policies") - t.Logf("Node VV after write: %v", nodeVV) - - // Verify pivot received the data + // Effect: the pivot received the node's data. pivotObj, err := servers.PivotPolicies.Get("policies") require.NoError(t, err) var pivotData map[string]string json.Unmarshal(pivotObj.Data, &pivotData) require.Equal(t, "from-node", pivotData["value"], "Pivot should have received data from node") - // Verify pivot incremented its VV (via Set handler) - require.Eventually(t, func() bool { - return pivotInstance.VVManager.Get("policies")["leader"] > 0 - }, 2*time.Second, 5*time.Millisecond, "pivot VV never bumped after node-driven write") - pivotVV := pivotInstance.VVManager.Get("policies") - t.Logf("Pivot VV after receiving: %v", pivotVV) - - t.Log("Phase 1 passed: Node write syncs to pivot with VV tracking") + t.Log("Phase 1 passed: node write syncs to pivot") // Phase 2: Pivot writes via HTTP, then manually trigger node sync // (Pivot doesn't auto-sync to node since node isn't registered in NodesKey) @@ -329,26 +343,23 @@ func TestOfflineNodeWriteAndSync(t *testing.T) { json.Unmarshal(nodeObj2.Data, &nodeData) require.Equal(t, "from-pivot", nodeData["value"], "Node should have received update from pivot") - // Check pivot VV incremented again — same poll-vs-callback race story. - require.Eventually(t, func() bool { - return pivotInstance.VVManager.Get("policies")["leader"] > pivotVV["leader"] - }, 2*time.Second, 5*time.Millisecond, "pivot VV never re-bumped after the second write") - pivotVV2 := pivotInstance.VVManager.Get("policies") - t.Logf("Pivot VV after second write: %v", pivotVV2) - - t.Log("Phase 2 passed: Pivot write syncs to node") - - t.Log("Offline sync test completed successfully") + t.Log("Phase 2 passed: pivot write syncs to node") } func TestVersionVectorActivityEndpoint(t *testing.T) { - servers := setupOfflineServers(t) + // pivotVVWrites observes the leader's VV persistence through the standard + // server.AfterWrite storage callback (see setupOfflineServers). A node push + // drives the pivot Set handler, which increments then merges its VV — two + // writes to pivot/vv/policies. Waiting on those makes the bump durable before + // we read it back through the external /activity endpoint, with no polling of + // internal VV state. + var pivotVVWrites sync.WaitGroup + servers := setupOfflineServers(t, withPivotVVWrites(&pivotVVWrites)) defer servers.Close() - // Write data via HTTP to trigger VV increment - // Expect: 1 node write (local) + 1 pivot write (from sync) servers.NodeWg.Add(1) servers.PivotWg.Add(1) + pivotVVWrites.Add(2) // pivot Set handler: increment-save + merge-save payload := []byte(`{"value": "test"}`) resp, err := servers.Node.Client.Post("http://"+servers.Node.Address+"/policies", "application/json", bytes.NewBuffer(payload)) @@ -356,21 +367,11 @@ func TestVersionVectorActivityEndpoint(t *testing.T) { resp.Body.Close() require.Equal(t, http.StatusOK, resp.StatusCode) - // Wait for both writes to complete servers.NodeWg.Wait() servers.PivotWg.Wait() + pivotVVWrites.Wait() - // Verify pivot has VV. The wg decrements fire from AfterWrite - // (synchronous with the storage Set), but the VV bump runs in the - // storage event callback on the watch goroutine — those two paths - // are decoupled. Poll until the bump has landed instead of asserting - // once and racing the callback. - pivotInstance := pivot.GetInstance(servers.Pivot) - require.Eventually(t, func() bool { - return pivotInstance.VVManager.Get("policies")["leader"] > 0 - }, 2*time.Second, 5*time.Millisecond, "pivot VV never bumped after the policies write drained") - - // Check activity endpoint on pivot includes VV + // Assert through the external API: /activity exposes the leader's VV. resp, err = servers.Pivot.Client.Get("http://" + servers.Pivot.Address + "/_pivot/activity/policies") require.NoError(t, err) defer resp.Body.Close() diff --git a/version_e2e_test.go b/version_e2e_test.go index 137536c..a8ab801 100644 --- a/version_e2e_test.go +++ b/version_e2e_test.go @@ -51,6 +51,40 @@ func VersionTestServer(t *testing.T, clusterURL string) *ooo.Server { return server } +// awaitPivotStatus subscribes to the node's pivot/status feed — the external +// cluster-status API the UI consumes — and returns the first status whose pivot +// protocol has been detected. The background health check broadcasts pivot/status +// on every status change (pivot.go), so this is the event-driven, deterministic +// replacement for polling GetPivotInfo while that check runs. sync.Once is the +// sanctioned pattern here: pivot/status is an unbounded broadcast stream and we +// want the first delivery that satisfies the condition. +func awaitPivotStatus(t *testing.T, nodeServer *ooo.Server) ui.PivotInfo { + t.Helper() + var done sync.WaitGroup + done.Add(1) + var once sync.Once + var mu sync.Mutex + var got ui.PivotInfo + go client.Subscribe(client.SubscribeConfig{ + Ctx: t.Context(), + Server: client.Server{Protocol: "ws", Host: nodeServer.Address}, + Silence: true, + }, "pivot/status", client.SubscribeEvents[ui.PivotInfo]{ + OnMessage: func(m client.Meta[ui.PivotInfo]) { + mu.Lock() + got = m.Data + mu.Unlock() + if m.Data.PivotProtocol != "unknown" { + once.Do(done.Done) + } + }, + }) + done.Wait() + mu.Lock() + defer mu.Unlock() + return got +} + func TestE2E_VersionSync_CompatibleServers(t *testing.T) { t.Parallel() // This test verifies that compatible servers can detect each other's version @@ -247,18 +281,9 @@ func TestE2E_VersionSync_NodeDetectsPivotProtocol(t *testing.T) { nodeServer := VersionTestServer(t, "http://"+pivotServer.Address) defer nodeServer.Close(os.Interrupt) - // Poll until health check detects pivot protocol (initial check runs async) - var nodeInfo *ui.PivotInfo - deadline := time.Now().Add(5 * time.Second) - for time.Now().Before(deadline) { - nodeInfo = pivot.GetPivotInfo(nodeServer)() - if nodeInfo != nil && nodeInfo.PivotProtocol != "unknown" { - break - } - time.Sleep(50 * time.Millisecond) - } - - require.NotNil(t, nodeInfo) + // Wait (event-driven) for the health check to detect the pivot's protocol, + // observed through the node's pivot/status feed. + nodeInfo := awaitPivotStatus(t, nodeServer) require.Equal(t, "node", nodeInfo.Role) require.Equal(t, "http://"+pivotServer.Address, nodeInfo.PivotIP) require.Equal(t, pivot.ProtocolVersion, nodeInfo.PivotProtocol, "node should detect pivot's protocol version") @@ -285,18 +310,9 @@ func TestE2E_VersionSync_NodeDetectsIncompatiblePivotProtocol(t *testing.T) { nodeServer := VersionTestServer(t, "http://"+mockPivot.Listener.Addr().String()) defer nodeServer.Close(os.Interrupt) - // Poll until health check detects pivot protocol - var nodeInfo *ui.PivotInfo - deadline := time.Now().Add(5 * time.Second) - for time.Now().Before(deadline) { - nodeInfo = pivot.GetPivotInfo(nodeServer)() - if nodeInfo != nil && nodeInfo.PivotProtocol != "unknown" { - break - } - time.Sleep(50 * time.Millisecond) - } - - require.NotNil(t, nodeInfo) + // Wait (event-driven) for the health check to detect the (incompatible) + // pivot protocol, observed through the node's pivot/status feed. + nodeInfo := awaitPivotStatus(t, nodeServer) require.Equal(t, "node", nodeInfo.Role) require.Equal(t, "1.0", nodeInfo.PivotProtocol, "node should detect pivot's protocol version 1.0") require.False(t, nodeInfo.PivotCompatible, "node should report pivot as incompatible") diff --git a/version_vector.go b/version_vector.go index 0a7f995..2e1c8d1 100644 --- a/version_vector.go +++ b/version_vector.go @@ -182,13 +182,6 @@ type VVManager struct { storage storage.Database nodeID string // ID of this node ("leader" for pivot, node path for nodes) shutdown bool // guarded by mu; prevents writes during shutdown - // onBump, when non-nil, fires (without m.mu held) after this node's own - // counter has been incremented and persisted — once per increment, never - // on a merge. Test-only synchronisation hook: the Set/Delete handlers bump - // the counter AFTER db.SetWithMeta returns (and thus after the storage - // AfterWrite that tests wait on), so a test that reads the VV right after - // the storage write races the bump. nil in production. See SetBumpObserver. - onBump func(baseKey string) } // NewVVManager creates a new version vector manager. @@ -231,6 +224,7 @@ func (m *VVManager) increment(keyPath string) { baseKey := normalizeKeyPath(keyPath) m.mu.Lock() + defer m.mu.Unlock() // Node servers create their VVManager with nodeID="" during Setup and only // call SetNodeID once server.Address is known (inside OnStart). The TCP @@ -239,7 +233,6 @@ func (m *VVManager) increment(keyPath string) { // ever increments and live in storage forever. Skip and log loudly so the // regression surfaces if a caller starts incrementing pre-SetNodeID. if m.nodeID == "" { - m.mu.Unlock() log.Printf("[pivot] VVManager.increment skipped for %q: nodeID not set yet", keyPath) return } @@ -253,14 +246,6 @@ func (m *VVManager) increment(keyPath string) { m.vectors[baseKey][m.nodeID]++ m.saveToStorage(baseKey) - // Capture under the lock, fire after releasing it: the observer must not - // be able to re-enter VVManager under m.mu, and the bump is already - // durable by here. - cb := m.onBump - m.mu.Unlock() - if cb != nil { - cb(baseKey) - } } // set merges a remote version vector into the local one to ensure @@ -337,21 +322,6 @@ func (m *VVManager) Shutdown() { m.mu.Unlock() } -// SetBumpObserver installs a callback fired after each successful counter -// increment+persist, for any key (pass nil to clear). It exists so async tests -// can deterministically synchronise on the pivot's post-write VV bump: the -// Set/Delete handlers increment the counter after the storage write — and thus -// after the storage AfterWrite a test waits on — so reading the VV immediately -// after the write races the bump. The callback fires once per increment and -// never on a merge, so the count is one per write that advances this node's own -// counter regardless of code path. Production never sets it. Safe to call from -// another goroutine; serialised with increment via m.mu. -func (m *VVManager) SetBumpObserver(cb func(baseKey string)) { - m.mu.Lock() - m.onBump = cb - m.mu.Unlock() -} - // SetNodeID sets the node ID for this manager. // Used by node servers to set their address once known. func (m *VVManager) SetNodeID(nodeID string) { diff --git a/vv_merge_test.go b/vv_merge_test.go index 85af7c9..7bccce9 100644 --- a/vv_merge_test.go +++ b/vv_merge_test.go @@ -18,7 +18,6 @@ import ( "strconv" "sync" "testing" - "time" "github.com/benitogf/ooo/meta" "github.com/benitogf/ooo/monotonic" @@ -169,11 +168,13 @@ func TestQueuedOpCarriesQueueTimeVV(t *testing.T) { mu struct{ sync.Mutex } received []recv ) + var delivered sync.WaitGroup // Done() per inbound POST — the deterministic drain signal srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { mu.Lock() received = append(received, recv{key: r.URL.Path, vv: r.Header.Get(VVHeader)}) mu.Unlock() w.WriteHeader(http.StatusOK) + delivered.Done() })) defer srv.Close() @@ -193,15 +194,12 @@ func TestQueuedOpCarriesQueueTimeVV(t *testing.T) { pool.syncers[srv.URL[len("http://"):]].QueueOrSendSet("things/b", meta.Object{Created: 2, Updated: 2, Index: "b", Path: "things/b", Data: []byte(`{}`)}) - // Drain by setting the node addr — the queue runs through sendToLeader - // with op.vv as the header value. + // Drain by setting the node addr — the queue runs through sendToLeader with + // op.vv as the header value. Each delivered op fires the mock leader's + // handler Done(); waiting on the two is deterministic, no polling. + delivered.Add(2) pool.SetNodeAddr("10.0.0.1:9000") - - require.Eventually(t, func() bool { - mu.Lock() - defer mu.Unlock() - return len(received) == 2 - }, time.Second, 5*time.Millisecond, "queue never drained both ops") + delivered.Wait() mu.Lock() defer mu.Unlock() diff --git a/vv_path_scope_test.go b/vv_path_scope_test.go index d4c0330..f936b2c 100644 --- a/vv_path_scope_test.go +++ b/vv_path_scope_test.go @@ -17,8 +17,8 @@ package pivot import ( "net/http/httptest" "strings" + "sync" "testing" - "time" "github.com/benitogf/ooo/monotonic" "github.com/benitogf/ooo/storage" @@ -45,26 +45,32 @@ func TestActivityExposesVVAfterGlobWrite(t *testing.T) { keys := []Key{{Path: "things/*", Database: db}} instance := &Instance{VVManager: vvm} - storage.WatchWithCallback(db, makeStorageSync(StorageSyncConfig{ + var processed sync.WaitGroup + watchProcessed(db, makeStorageSync(StorageSyncConfig{ Keys: keys, GetNodes: func() []string { return nil }, HandlerTracker: tracker, Instance: instance, - })) + }), "things/*", &processed) // Wire a Set handler the way a glob-path key registers it: path= // "things" (the registered base, with the /* stripped before // mounting) and items go under "things/". setHandler := Set(db, "things", tracker, vvm, nil) - // Drive a write under the glob. + // Drive a write under the glob, then wait for the watch goroutine to process + // it. Waiting matters for the double-bump claim below: a regression that let + // the callback also bump would only show as leader==2 AFTER the callback ran, + // so reading before it processed would race past the bug. body := strings.NewReader(`{"created":0,"updated":0,"index":"x","path":"things/x","data":"e30="}`) req := httptest.NewRequest("POST", "/_pivot/pivot/things/x", body) req = mux.SetURLVars(req, map[string]string{"index": "x"}) req.Header.Set("Content-Type", "application/json") w := httptest.NewRecorder() + processed.Add(1) setHandler(w, req) require.Equal(t, 200, w.Code) + processed.Wait() // Stand up the Activity handler the way pivot.go registers it: // the Key.Path is the glob, "things/*". Activity normalizes that @@ -97,24 +103,23 @@ func TestCallbackIncrementsAtPathScope(t *testing.T) { vvm := NewVVManager(db, "leader") keys := []Key{{Path: "things/*", Database: db}} instance := &Instance{VVManager: vvm} - storage.WatchWithCallback(db, makeStorageSync(StorageSyncConfig{ + var processed sync.WaitGroup + watchProcessed(db, makeStorageSync(StorageSyncConfig{ Keys: keys, GetNodes: func() []string { return nil }, Instance: instance, - })) + }), "things/*", &processed) - // Direct storage write — bypasses the Set handler, so the storage - // callback is the one bumping VV. + // Direct storage write — bypasses the Set handler, so the storage callback is + // the one bumping VV. Wait for the watch goroutine to process that event, then + // assert exactly 1 (deterministic — no polling). + processed.Add(1) _, err := db.SetWithMeta("things/x", []byte(`{"v":"v1"}`), 1, 1) require.NoError(t, err) + processed.Wait() - // Settle for the watch goroutine. The callback is the only thing - // touching VV here, and increments synchronously inside the - // goroutine — a small wait is enough. - require.Eventually(t, func() bool { - return vvm.Get("things/*")["leader"] >= 1 - }, time.Second, 5*time.Millisecond, - "storage callback must increment path-scope VV on direct writes") + require.Equal(t, int64(1), vvm.Get("things/*")["leader"], + "storage callback must increment path-scope VV exactly once on a direct write") } // TestHandlerIncrementMatchesActivityScope pins the symmetry: every @@ -133,16 +138,19 @@ func TestHandlerIncrementMatchesActivityScope(t *testing.T) { keys := []Key{{Path: "things/*", Database: db}} instance := &Instance{VVManager: vvm} - storage.WatchWithCallback(db, makeStorageSync(StorageSyncConfig{ + var processed sync.WaitGroup + watchProcessed(db, makeStorageSync(StorageSyncConfig{ Keys: keys, GetNodes: func() []string { return nil }, HandlerTracker: tracker, Instance: instance, - })) + }), "things/*", &processed) setHandler := Set(db, "things", tracker, vvm, nil) - for i, idx := range []string{"a", "b", "c"} { + indices := []string{"a", "b", "c"} + processed.Add(len(indices)) + for i, idx := range indices { body := strings.NewReader(`{"created":0,"updated":0,"index":"` + idx + `","path":"things/` + idx + `","data":"e30="}`) req := httptest.NewRequest("POST", "/_pivot/pivot/things/"+idx, body) req = mux.SetURLVars(req, map[string]string{"index": idx}) @@ -152,10 +160,10 @@ func TestHandlerIncrementMatchesActivityScope(t *testing.T) { require.Equal(t, 200, w.Code, "write %d failed", i) } - // Wait for the watch goroutine to drain. Once tracker is empty, every - // handler Mark has been Consumed by the callback (callback skipped - // each). - require.Eventually(t, func() bool { return tracker.pendingLen() == 0 }, time.Second, 5*time.Millisecond) + // Wait for the watch goroutine to process all three writes (it Consumes each + // handler Mark and skips its own bump). Its completion is the deterministic + // signal — no polling of internal tracker state. + processed.Wait() // Three writes under the same registered glob path → one VV with // leader counter at 3. Pre-fix each write bumped a separate item- From 963af6b5d01d7568a8a9f0ebfc943c21f47941ce Mon Sep 17 00:00:00 2001 From: root Date: Tue, 9 Jun 2026 22:14:06 +0800 Subject: [PATCH 2/3] test(version-e2e): poll detected protocol instead of subscribing (fix CI hang) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The prior commit's awaitPivotStatus subscribed to the node's pivot/status feed and waited for a detected protocol. That hangs: pivot/status broadcasts only on a status CHANGE, so the single unknown→detected transition can be missed in the subscription's connect window (initial snapshot still "unknown", the one change broadcast lost) — after which no further broadcast arrives and the wait never returns. CI hit this (TestE2E_VersionSync_NodeDetectsIncompatiblePivotProtocol ran 48s and blew the 60s package timeout); reproduced locally 4/6 under parallel -race load. Protocol detection is a level (current state of an async background health check), not a reliably observable edge, so read the level: awaitDetectedProtocol polls GetPivotInfo until it leaves "unknown" with a bounded deadline. Documented why this is the right tool here rather than an event wait, and the PR's triage table moves version_e2e to the justified bucket. Verified: the parallel -race load that hung 4/6 now passes all 6. Co-Authored-By: Claude Opus 4.8 (1M context) --- version_e2e_test.go | 57 +++++++++++++++++++-------------------------- 1 file changed, 24 insertions(+), 33 deletions(-) diff --git a/version_e2e_test.go b/version_e2e_test.go index a8ab801..1e050d9 100644 --- a/version_e2e_test.go +++ b/version_e2e_test.go @@ -51,38 +51,29 @@ func VersionTestServer(t *testing.T, clusterURL string) *ooo.Server { return server } -// awaitPivotStatus subscribes to the node's pivot/status feed — the external -// cluster-status API the UI consumes — and returns the first status whose pivot -// protocol has been detected. The background health check broadcasts pivot/status -// on every status change (pivot.go), so this is the event-driven, deterministic -// replacement for polling GetPivotInfo while that check runs. sync.Once is the -// sanctioned pattern here: pivot/status is an unbounded broadcast stream and we -// want the first delivery that satisfies the condition. -func awaitPivotStatus(t *testing.T, nodeServer *ooo.Server) ui.PivotInfo { +// awaitDetectedProtocol reads the node's detected pivot protocol until it leaves +// "unknown" or the deadline elapses. +// +// This deliberately polls rather than subscribing to an event. Protocol detection +// is a LEVEL produced by an async background health check, not a reliably +// observable edge: the node broadcasts pivot/status only on a status *change*, so +// the single unknown→detected transition can be missed in a subscriber's connect +// window (initial snapshot still "unknown", the one change broadcast lost), after +// which no further broadcast ever arrives and a subscribe-and-wait hangs. Reading +// the current level is the robust tool — /testing-go-backend-async's "wait on a +// callback" guidance assumes a discrete completion event, which a background +// level-detector with a change-only broadcast does not provide. +func awaitDetectedProtocol(t *testing.T, nodeServer *ooo.Server) *ui.PivotInfo { t.Helper() - var done sync.WaitGroup - done.Add(1) - var once sync.Once - var mu sync.Mutex - var got ui.PivotInfo - go client.Subscribe(client.SubscribeConfig{ - Ctx: t.Context(), - Server: client.Server{Protocol: "ws", Host: nodeServer.Address}, - Silence: true, - }, "pivot/status", client.SubscribeEvents[ui.PivotInfo]{ - OnMessage: func(m client.Meta[ui.PivotInfo]) { - mu.Lock() - got = m.Data - mu.Unlock() - if m.Data.PivotProtocol != "unknown" { - once.Do(done.Done) - } - }, - }) - done.Wait() - mu.Lock() - defer mu.Unlock() - return got + deadline := time.Now().Add(5 * time.Second) + for time.Now().Before(deadline) { + if info := pivot.GetPivotInfo(nodeServer)(); info != nil && info.PivotProtocol != "unknown" { + return info + } + time.Sleep(50 * time.Millisecond) + } + t.Fatal("health check did not detect the pivot protocol within 5s") + return nil } func TestE2E_VersionSync_CompatibleServers(t *testing.T) { @@ -283,7 +274,7 @@ func TestE2E_VersionSync_NodeDetectsPivotProtocol(t *testing.T) { // Wait (event-driven) for the health check to detect the pivot's protocol, // observed through the node's pivot/status feed. - nodeInfo := awaitPivotStatus(t, nodeServer) + nodeInfo := awaitDetectedProtocol(t, nodeServer) require.Equal(t, "node", nodeInfo.Role) require.Equal(t, "http://"+pivotServer.Address, nodeInfo.PivotIP) require.Equal(t, pivot.ProtocolVersion, nodeInfo.PivotProtocol, "node should detect pivot's protocol version") @@ -312,7 +303,7 @@ func TestE2E_VersionSync_NodeDetectsIncompatiblePivotProtocol(t *testing.T) { // Wait (event-driven) for the health check to detect the (incompatible) // pivot protocol, observed through the node's pivot/status feed. - nodeInfo := awaitPivotStatus(t, nodeServer) + nodeInfo := awaitDetectedProtocol(t, nodeServer) require.Equal(t, "node", nodeInfo.Role) require.Equal(t, "1.0", nodeInfo.PivotProtocol, "node should detect pivot's protocol version 1.0") require.False(t, nodeInfo.PivotCompatible, "node should report pivot as incompatible") From f55626fe42fa74c45cf1eec3a86f92013e562dd7 Mon Sep 17 00:00:00 2001 From: root Date: Mon, 29 Jun 2026 16:42:00 +0800 Subject: [PATCH 3/3] docs(version-e2e): reword call-site comments to describe the bounded poll MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The 963af6b rewrite swapped subscribe->poll but left the two awaitDetectedProtocol call-site comments asserting "event-driven ... observed through the node's pivot/status feed" — the opposite of what the helper now does. Reword both to describe the bounded poll and defer to awaitDetectedProtocol's own doc for the rationale. Co-Authored-By: Claude Opus 4.8 (1M context) --- version_e2e_test.go | 10 ++++++---- 1 file changed, 6 insertions(+), 4 deletions(-) diff --git a/version_e2e_test.go b/version_e2e_test.go index 1e050d9..7915f9b 100644 --- a/version_e2e_test.go +++ b/version_e2e_test.go @@ -272,8 +272,9 @@ func TestE2E_VersionSync_NodeDetectsPivotProtocol(t *testing.T) { nodeServer := VersionTestServer(t, "http://"+pivotServer.Address) defer nodeServer.Close(os.Interrupt) - // Wait (event-driven) for the health check to detect the pivot's protocol, - // observed through the node's pivot/status feed. + // Bounded poll of the node's detected protocol level until it leaves + // "unknown" (see awaitDetectedProtocol for why this polls rather than + // subscribes). nodeInfo := awaitDetectedProtocol(t, nodeServer) require.Equal(t, "node", nodeInfo.Role) require.Equal(t, "http://"+pivotServer.Address, nodeInfo.PivotIP) @@ -301,8 +302,9 @@ func TestE2E_VersionSync_NodeDetectsIncompatiblePivotProtocol(t *testing.T) { nodeServer := VersionTestServer(t, "http://"+mockPivot.Listener.Addr().String()) defer nodeServer.Close(os.Interrupt) - // Wait (event-driven) for the health check to detect the (incompatible) - // pivot protocol, observed through the node's pivot/status feed. + // Bounded poll of the node's detected protocol level until it leaves + // "unknown" (see awaitDetectedProtocol for why this polls rather than + // subscribes). nodeInfo := awaitDetectedProtocol(t, nodeServer) require.Equal(t, "node", nodeInfo.Role) require.Equal(t, "1.0", nodeInfo.PivotProtocol, "node should detect pivot's protocol version 1.0")