From 1d864233b0abbb0dd3c367834ed7513e75800645 Mon Sep 17 00:00:00 2001 From: root Date: Tue, 16 Jun 2026 20:27:35 +0800 Subject: [PATCH] =?UTF-8?q?test(cluster):=20fix=20~1/300=20TestClusterSync?= =?UTF-8?q?=20flake=20via=20ooo=20lossless=20watch;=20port=20Audit?= =?UTF-8?q?=E2=86=92Router.Use?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Bump ooo to the lossless-watch release. TestClusterSyncLocal/Remote hung intermittently (~1/300 -race iterations, only under -count) because ooo dropped a storage→broadcast event when a consumer stalled under load, so a subscriber missed a delivery and the test's WaitGroup waited forever. The WaitGroup counts were correct all along — the loss was below them. FakeServer now enables ooo's LosslessWatch (block instead of drop) with an OnDroppedEvent guard that fails the test if any event is ever dropped, so every committed write deterministically reaches every live subscriber. The cluster sync test's WaitGroup logic is unchanged. The ooo bump also removes Server.Audit (now gated via Router.Use middleware): drop the allow-all Audit test hooks (no middleware = allow all by default) and port the one deny gate (nodehealth ping probe test) to Router.Use. Pass the instance context to WatchWithCallback, now context-cancellable, so attached storages' watch workers stop on shutdown instead of leaking. Verified with /flaky-check: 300 cumulative full-suite -race iterations clean against the released ooo (540 against the same fix pre-release), zero drops. Co-Authored-By: Claude Opus 4.8 (1M context) --- .gitignore | 3 ++- benchmark_test.go | 5 ---- clock_drift_test.go | 2 -- cluster_test.go | 14 ++++++++--- crab_test.go | 12 ---------- edge_test.go | 2 -- go.mod | 2 +- go.sum | 2 ++ handlers_internal_test.go | 19 ++++++++------- idempotency_guard_test.go | 17 ++++++------- instance.go | 4 ++-- multi_cluster_test.go | 5 +--- nodehealth.go | 12 +++++----- nodehealth_pingnode_test.go | 37 ++++++++++++++++++----------- offline_sync_test.go | 2 -- sync_glob_clock_skew_test.go | 5 ++-- sync_singlekey_tombstone_vv_test.go | 3 ++- sync_vv_divergence_test.go | 15 ++++++------ synctoleader_tombstone_test.go | 7 +++--- version_e2e_test.go | 1 - version_vector_internal_test.go | 3 ++- vv_bump_bench_test.go | 1 - vv_glob_pull_delete_test.go | 2 -- vv_merge_test.go | 6 ++--- vv_path_scope_test.go | 7 +++--- 25 files changed, 93 insertions(+), 95 deletions(-) diff --git a/.gitignore b/.gitignore index fadf458..0fcac69 100644 --- a/.gitignore +++ b/.gitignore @@ -2,4 +2,5 @@ pivot.test test_* test/ -*.code-workspace \ No newline at end of file +*.code-workspace +.flaky/ diff --git a/benchmark_test.go b/benchmark_test.go index 1b2919f..5bd2ecf 100644 --- a/benchmark_test.go +++ b/benchmark_test.go @@ -30,7 +30,6 @@ func createBenchServer() *ooo.Server { DisableKeepAlives: true, }, } - server.Audit = func(r *http.Request) bool { return true } server.OpenFilter("things/*") server.OpenFilter("settings") server.Start("localhost:0") @@ -56,7 +55,6 @@ func createBenchPivotServer() *ooo.Server { server.Static = true server.Storage = storage.New(storage.LayeredConfig{Memory: storage.NewMemoryLayer()}) server.Router = mux.NewRouter() - server.Audit = func(r *http.Request) bool { return true } config := pivot.Config{ Keys: []pivot.Key{{Path: "settings"}}, @@ -79,7 +77,6 @@ func createBenchNodeServer(pivotAddress string) *ooo.Server { server.Static = true server.Storage = storage.New(storage.LayeredConfig{Memory: storage.NewMemoryLayer()}) server.Router = mux.NewRouter() - server.Audit = func(r *http.Request) bool { return true } config := pivot.Config{ Keys: []pivot.Key{{Path: "settings"}}, @@ -268,7 +265,6 @@ func createBenchPivotServerWithWaiter(waiter *eventWaiter) *ooo.Server { server.Static = true server.Storage = storage.New(storage.LayeredConfig{Memory: storage.NewMemoryLayer()}) server.Router = mux.NewRouter() - server.Audit = func(r *http.Request) bool { return true } config := pivot.Config{ Keys: []pivot.Key{{Path: "settings"}}, @@ -302,7 +298,6 @@ func createBenchNodeServerWithWaiter(pivotAddress string, waiter *eventWaiter) * server.Static = true server.Storage = storage.New(storage.LayeredConfig{Memory: storage.NewMemoryLayer()}) server.Router = mux.NewRouter() - server.Audit = func(r *http.Request) bool { return true } config := pivot.Config{ Keys: []pivot.Key{{Path: "settings"}}, diff --git a/clock_drift_test.go b/clock_drift_test.go index 6d9112a..f73878b 100644 --- a/clock_drift_test.go +++ b/clock_drift_test.go @@ -279,7 +279,6 @@ func setupClobberServers(t *testing.T) *clobberServers { pivotServer.Storage = storage.New(storage.LayeredConfig{Memory: storage.NewMemoryLayer()}) pivotServer.Router = mux.NewRouter() pivotServer.Client = mkClient() - pivotServer.Audit = func(r *http.Request) bool { return true } pivot.Setup(pivotServer, pivot.Config{ Keys: []pivot.Key{{Path: "policies", Database: pivotPoliciesStorage}}, @@ -298,7 +297,6 @@ func setupClobberServers(t *testing.T) *clobberServers { nodeServer.Storage = storage.New(storage.LayeredConfig{Memory: storage.NewMemoryLayer()}) nodeServer.Router = mux.NewRouter() nodeServer.Client = mkClient() - nodeServer.Audit = func(r *http.Request) bool { return true } pivot.Setup(nodeServer, pivot.Config{ Keys: []pivot.Key{{Path: "policies", Database: nodePoliciesStorage}}, diff --git a/cluster_test.go b/cluster_test.go index e022e56..19e753a 100644 --- a/cluster_test.go +++ b/cluster_test.go @@ -96,6 +96,17 @@ func FakeServer(t *testing.T, clusterURL string, onPolicyWrite func()) *ooo.Serv server := &ooo.Server{} server.Silence = true server.Static = true + // Lossless watch: never drop a storage→broadcast event under consumer + // stall. The default drop-after-timeout is a production write-hang + // resilience feature, but in this deterministic test a dropped broadcast + // would silently desync a subscriber and surface as a flaky hung Wait. With + // it off, every committed write deterministically reaches every live sub — + // exactly what the per-operation WaitGroups below assume. OnDroppedEvent is + // wired as a guard: it must never fire. + server.LosslessWatch = true + server.OnDroppedEvent = func(ev storage.Event) { + t.Errorf("watch event dropped (key=%q op=%q) — lossless watch should prevent this", ev.Key, ev.Operation) + } server.Storage = storage.New(storage.LayeredConfig{Memory: storage.NewMemoryLayer()}) server.Router = mux.NewRouter() server.Client = &http.Client{ @@ -108,9 +119,6 @@ func FakeServer(t *testing.T, clusterURL string, onPolicyWrite func()) *ooo.Serv DisableKeepAlives: true, }, } - server.Audit = func(r *http.Request) bool { - return true - } // Create auth store authStorage := storage.New(storage.LayeredConfig{Memory: storage.NewMemoryLayer()}) diff --git a/crab_test.go b/crab_test.go index fff419a..b72b4f1 100644 --- a/crab_test.go +++ b/crab_test.go @@ -207,9 +207,6 @@ func startNodeServerPerKey(globalPivotURL, devicesPivotURL string, nodeStorage s DisableKeepAlives: true, }, } - server.Audit = func(r *http.Request) bool { - return true - } config := pivot.Config{ Keys: []pivot.Key{ @@ -246,9 +243,6 @@ func startPivotServerWithDevices(pivotIP string) *ooo.Server { DisableKeepAlives: true, }, } - server.Audit = func(r *http.Request) bool { - return true - } config := pivot.Config{ Keys: []pivot.Key{ @@ -291,9 +285,6 @@ func startPivotServer(pivotIP string) (*ooo.Server, *sync.WaitGroup) { DisableKeepAlives: true, }, } - server.Audit = func(r *http.Request) bool { - return true - } config := pivot.Config{ Keys: []pivot.Key{ @@ -346,9 +337,6 @@ func startNodeServer(pivotIP string, nodeStorage storage.Database) (*ooo.Server, DisableKeepAlives: true, }, } - server.Audit = func(r *http.Request) bool { - return true - } config := pivot.Config{ Keys: []pivot.Key{ diff --git a/edge_test.go b/edge_test.go index 3eaff5a..83dbad2 100644 --- a/edge_test.go +++ b/edge_test.go @@ -167,7 +167,6 @@ func createEdgeTestServer(pivotIP string, nodeStorage storage.Database) (*ooo.Se DisableKeepAlives: true, }, } - server.Audit = func(r *http.Request) bool { return true } config := pivot.Config{ Keys: []pivot.Key{{Path: "settings"}}, @@ -210,7 +209,6 @@ func createEdgeTestServerNoSync(pivotIP string) *ooo.Server { DisableKeepAlives: true, }, } - server.Audit = func(r *http.Request) bool { return true } config := pivot.Config{ Keys: []pivot.Key{{Path: "settings"}}, diff --git a/go.mod b/go.mod index 68b7690..e337f5e 100644 --- a/go.mod +++ b/go.mod @@ -7,7 +7,7 @@ require ( github.com/benitogf/coat v0.0.0-20200402073050-ff807656cbec github.com/benitogf/go-json v0.0.0-20260410172501-727f5690408b github.com/benitogf/ko v0.0.0-20260211072652-d48fcf4f8988 - github.com/benitogf/ooo v0.0.0-20260606052832-8a306d163ab8 + github.com/benitogf/ooo v0.0.0-20260616085443-cb886859c622 github.com/gorilla/mux v1.8.1 github.com/stretchr/testify v1.11.1 ) diff --git a/go.sum b/go.sum index 3625b30..aa8baa3 100644 --- a/go.sum +++ b/go.sum @@ -14,6 +14,8 @@ github.com/benitogf/ko v0.0.0-20260211072652-d48fcf4f8988 h1:sagWGc0GUBEMxa56Hpj github.com/benitogf/ko v0.0.0-20260211072652-d48fcf4f8988/go.mod h1:fRbtp9nrkNeDXowzM7KRJP6TL6OMsAyroUylQCw+77s= github.com/benitogf/ooo v0.0.0-20260606052832-8a306d163ab8 h1:W3gw3dIrDiDZa5JK32lROYm/KZK7ieOFvl9zti748eA= github.com/benitogf/ooo v0.0.0-20260606052832-8a306d163ab8/go.mod h1:WvPwWgfK2mo3UKfR+BM/9hwHe+ZjVwZZ3rxOPcBcXnw= +github.com/benitogf/ooo v0.0.0-20260616085443-cb886859c622 h1:g7Ht7IhjO+jVVh6scWacwyVATP9sdure7SNAXCqRutw= +github.com/benitogf/ooo v0.0.0-20260616085443-cb886859c622/go.mod h1:WvPwWgfK2mo3UKfR+BM/9hwHe+ZjVwZZ3rxOPcBcXnw= github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c= github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38= github.com/felixge/httpsnoop v1.0.4 h1:NFTV2Zj1bL4mc9sqWACXbQFVBBg2W3GPvqp8/ESS2Wg= diff --git a/handlers_internal_test.go b/handlers_internal_test.go index ff88776..bcb3536 100644 --- a/handlers_internal_test.go +++ b/handlers_internal_test.go @@ -1,6 +1,7 @@ package pivot import ( + "context" "encoding/json" "errors" "net/http/httptest" @@ -61,7 +62,7 @@ func TestDeleteTombstoneAtomicity(t *testing.T) { real := storage.New(storage.LayeredConfig{Memory: storage.NewMemoryLayer()}) require.NoError(t, real.Start(storage.Options{})) defer real.Close() - storage.WatchWithCallback(real, func(storage.Event) {}) + storage.WatchWithCallback(context.Background(), real, func(storage.Event) {}) itemKey := "things/abc" tombstoneKey := StoragePrefix + "things" @@ -129,7 +130,7 @@ func TestSetVVIncrementsExactlyOnce(t *testing.T) { vvm := NewVVManager(db, "leader") keys := []Key{{Path: "things/*", Database: db}} instance := &Instance{VVManager: vvm} - storage.WatchWithCallback(db, makeStorageSync(StorageSyncConfig{ + storage.WatchWithCallback(context.Background(), db, makeStorageSync(StorageSyncConfig{ Keys: keys, GetNodes: func() []string { return nil }, HandlerTracker: tracker, @@ -176,7 +177,7 @@ func TestSetVVIncrementsExactlyOnceNodeRole(t *testing.T) { keys := []Key{{Path: "things/*", Database: db}} instance := &Instance{VVManager: vvm} - storage.WatchWithCallback(db, makeStorageSync(StorageSyncConfig{ + storage.WatchWithCallback(context.Background(), db, makeStorageSync(StorageSyncConfig{ Keys: keys, ConfigClusterURL: "127.0.0.1:8000", // non-empty -> node mode GetNodes: func() []string { return nil }, @@ -220,7 +221,7 @@ func TestSetVVIncrementsExactlyOnceUnderBurst(t *testing.T) { vvm := NewVVManager(db, "leader") keys := []Key{{Path: "things/*", Database: db}} instance := &Instance{VVManager: vvm} - storage.WatchWithCallback(db, makeStorageSync(StorageSyncConfig{ + storage.WatchWithCallback(context.Background(), db, makeStorageSync(StorageSyncConfig{ Keys: keys, GetNodes: func() []string { return nil }, HandlerTracker: tracker, @@ -267,7 +268,7 @@ func TestDeleteDoesNotLeakHandlerMarks(t *testing.T) { vvm := NewVVManager(db, "leader") keys := []Key{{Path: "things/*", Database: db}} instance := &Instance{VVManager: vvm} - storage.WatchWithCallback(db, makeStorageSync(StorageSyncConfig{ + storage.WatchWithCallback(context.Background(), db, makeStorageSync(StorageSyncConfig{ Keys: keys, GetNodes: func() []string { return nil }, HandlerTracker: tracker, @@ -323,7 +324,7 @@ func TestSetPostWriteSeesBumpedVV(t *testing.T) { db := storage.New(storage.LayeredConfig{Memory: storage.NewMemoryLayer()}) require.NoError(t, db.Start(storage.Options{})) defer db.Close() - storage.WatchWithCallback(db, func(storage.Event) {}) + storage.WatchWithCallback(context.Background(), db, func(storage.Event) {}) tracker := NewHandlerWriteTracker() vvm := NewVVManager(db, "leader") @@ -358,7 +359,7 @@ func TestSetVVDoesNotBumpOnStorageFailure(t *testing.T) { real := storage.New(storage.LayeredConfig{Memory: storage.NewMemoryLayer()}) require.NoError(t, real.Start(storage.Options{})) defer real.Close() - storage.WatchWithCallback(real, func(storage.Event) {}) + storage.WatchWithCallback(context.Background(), real, func(storage.Event) {}) failing := &failingItemStorage{ Database: real, @@ -407,7 +408,7 @@ func TestDeleteVVDoesNotBumpOnStorageFailure(t *testing.T) { real := storage.New(storage.LayeredConfig{Memory: storage.NewMemoryLayer()}) require.NoError(t, real.Start(storage.Options{})) defer real.Close() - storage.WatchWithCallback(real, func(storage.Event) {}) + storage.WatchWithCallback(context.Background(), real, func(storage.Event) {}) itemKey := "things/abc" tombstoneKey := StoragePrefix + "things" @@ -448,7 +449,7 @@ func TestDeleteHappyPathStillCommitsBoth(t *testing.T) { db := storage.New(storage.LayeredConfig{Memory: storage.NewMemoryLayer()}) require.NoError(t, db.Start(storage.Options{})) defer db.Close() - storage.WatchWithCallback(db, func(storage.Event) {}) + storage.WatchWithCallback(context.Background(), db, func(storage.Event) {}) itemKey := "things/abc" tombstoneKey := StoragePrefix + "things" diff --git a/idempotency_guard_test.go b/idempotency_guard_test.go index 52d5601..dfa85f8 100644 --- a/idempotency_guard_test.go +++ b/idempotency_guard_test.go @@ -21,6 +21,7 @@ package pivot import ( "bytes" + "context" "encoding/json" "net/http/httptest" "strconv" @@ -43,7 +44,7 @@ func TestSetGuardSkipsStaleRetry(t *testing.T) { db := storage.New(storage.LayeredConfig{Memory: storage.NewMemoryLayer()}) require.NoError(t, db.Start(storage.Options{})) defer db.Close() - storage.WatchWithCallback(db, func(storage.Event) {}) + storage.WatchWithCallback(context.Background(), db, func(storage.Event) {}) // Seed local with the newer "operator" value. Local VV reflects a // prior synced write from node A (A:2) and the operator's local @@ -82,7 +83,7 @@ func TestSetGuardSkipsExactRetry(t *testing.T) { db := storage.New(storage.LayeredConfig{Memory: storage.NewMemoryLayer()}) require.NoError(t, db.Start(storage.Options{})) defer db.Close() - storage.WatchWithCallback(db, func(storage.Event) {}) + storage.WatchWithCallback(context.Background(), db, func(storage.Event) {}) _, err := db.SetWithMeta("things/x", []byte(`{"v":"v1"}`), 100, 100) require.NoError(t, err) @@ -114,7 +115,7 @@ func TestSetGuardAcceptsHigherVV(t *testing.T) { db := storage.New(storage.LayeredConfig{Memory: storage.NewMemoryLayer()}) require.NoError(t, db.Start(storage.Options{})) defer db.Close() - storage.WatchWithCallback(db, func(storage.Event) {}) + storage.WatchWithCallback(context.Background(), db, func(storage.Event) {}) _, err := db.SetWithMeta("things/x", []byte(`{"v":"older"}`), 100, 100) require.NoError(t, err) @@ -148,7 +149,7 @@ func TestSetGuardProceedsWithoutHeader(t *testing.T) { db := storage.New(storage.LayeredConfig{Memory: storage.NewMemoryLayer()}) require.NoError(t, db.Start(storage.Options{})) defer db.Close() - storage.WatchWithCallback(db, func(storage.Event) {}) + storage.WatchWithCallback(context.Background(), db, func(storage.Event) {}) vvm := NewVVManager(db, "leader") // Seed local VV that *would* dominate if compared. @@ -183,7 +184,7 @@ func TestSetGuardPreservesClockDriftScenario(t *testing.T) { db := storage.New(storage.LayeredConfig{Memory: storage.NewMemoryLayer()}) require.NoError(t, db.Start(storage.Options{})) defer db.Close() - storage.WatchWithCallback(db, func(storage.Event) {}) + storage.WatchWithCallback(context.Background(), db, func(storage.Event) {}) // Local has the future-timestamped write at A:1. Pivot's local VV // reflects {A:1}. @@ -229,7 +230,7 @@ func TestSetGuardProceedsOnVVConcurrent(t *testing.T) { db := storage.New(storage.LayeredConfig{Memory: storage.NewMemoryLayer()}) require.NoError(t, db.Start(storage.Options{})) defer db.Close() - storage.WatchWithCallback(db, func(storage.Event) {}) + storage.WatchWithCallback(context.Background(), db, func(storage.Event) {}) _, err := db.SetWithMeta("things/x", []byte(`{"v":"local"}`), 100, 100) require.NoError(t, err) @@ -273,7 +274,7 @@ func TestSetGuardProceedsOnMalformedHeader(t *testing.T) { db := storage.New(storage.LayeredConfig{Memory: storage.NewMemoryLayer()}) require.NoError(t, db.Start(storage.Options{})) defer db.Close() - storage.WatchWithCallback(db, func(storage.Event) {}) + storage.WatchWithCallback(context.Background(), db, func(storage.Event) {}) vvm := NewVVManager(db, "leader") vvm.set("things/*", VersionVector{"leader": 5}) // would dominate if compared @@ -303,7 +304,7 @@ func TestDeleteGuardSkipsStaleTombstone(t *testing.T) { db := storage.New(storage.LayeredConfig{Memory: storage.NewMemoryLayer()}) require.NoError(t, db.Start(storage.Options{})) defer db.Close() - storage.WatchWithCallback(db, func(storage.Event) {}) + storage.WatchWithCallback(context.Background(), db, func(storage.Event) {}) _, err := db.SetWithMeta("things/x", []byte(`{"v":"newer"}`), 100, 100) require.NoError(t, err) diff --git a/instance.go b/instance.go index fac8528..63c26bb 100644 --- a/instance.go +++ b/instance.go @@ -321,7 +321,7 @@ func (i *Instance) bumpVVForLocalWrite(eventKey string, op string) { // This is a convenience method that replaces the manual setup: // // db.Start(storage.Options{BeforeRead: instance.BeforeRead}) -// storage.WatchWithCallback(db, instance.SyncCallback) +// storage.WatchWithCallback(instance.ctx, db, instance.SyncCallback) // // Optional storageOpts can be provided to pass additional storage options (e.g., AfterWrite for testing). // @@ -396,7 +396,7 @@ func (i *Instance) Attach(db storage.Database, storageOpts ...storage.Options) e return err } } - storage.WatchWithCallback(db, i.SyncCallback) + storage.WatchWithCallback(i.ctx, db, i.SyncCallback) return nil } diff --git a/multi_cluster_test.go b/multi_cluster_test.go index f49ab82..37de199 100644 --- a/multi_cluster_test.go +++ b/multi_cluster_test.go @@ -11,11 +11,11 @@ import ( "testing" "time" + "github.com/benitogf/go-json" "github.com/benitogf/ooo" ooio "github.com/benitogf/ooo/io" "github.com/benitogf/ooo/storage" "github.com/benitogf/pivot" - "github.com/benitogf/go-json" "github.com/gorilla/mux" "github.com/stretchr/testify/require" ) @@ -380,7 +380,6 @@ func MultiClusterAuthServer(t *testing.T, deviceClusterURL string, afterAuthWrit server.Storage = storage.New(storage.LayeredConfig{Memory: storage.NewMemoryLayer()}) server.Router = mux.NewRouter() server.Client = clusterTestClient() - server.Audit = func(r *http.Request) bool { return true } // Create separate storage for users and policies (external storage pattern) authStorage := storage.New(storage.LayeredConfig{Memory: storage.NewMemoryLayer()}) @@ -437,7 +436,6 @@ func MultiClusterDevicePivot(t *testing.T, afterDeviceWrite func(key string)) *o server.Storage = storage.New(storage.LayeredConfig{Memory: storage.NewMemoryLayer()}) server.Router = mux.NewRouter() server.Client = clusterTestClient() - server.Audit = func(r *http.Request) bool { return true } config := pivot.Config{ Keys: []pivot.Key{ @@ -475,7 +473,6 @@ func MultiClusterNodeDevice(t *testing.T, authServerURL string, deviceClusterURL server.Storage = storage.New(storage.LayeredConfig{Memory: storage.NewMemoryLayer()}) server.Router = mux.NewRouter() server.Client = clusterTestClient() - server.Audit = func(r *http.Request) bool { return true } // Create separate storage for users and policies authStorage := storage.New(storage.LayeredConfig{Memory: storage.NewMemoryLayer()}) diff --git a/nodehealth.go b/nodehealth.go index 5613d9d..61b8a5b 100644 --- a/nodehealth.go +++ b/nodehealth.go @@ -317,12 +317,12 @@ func (nh *NodeHealth) IsCompatible(node string) bool { // pingNode checks if a node is reachable. It probes the pivot-internal // /_pivot/version route rather than the ooo root ("/"): the root is served by -// the UI handler behind the Audit gate, so a node hardened with a non-trivial -// Audit returns 401 on "/" and would be permanently marked unhealthy. The -// /_pivot/version route is registered without the Audit wrapper, so it stays -// reachable. GET (not HEAD) keeps the probe compatible with older nodes that -// registered /_pivot/version for GET only, so a rolling upgrade never marks a -// not-yet-upgraded node unhealthy. +// the UI handler, so a node hardened with an auth gate (request middleware) +// returns 401 on "/" and would be permanently marked unhealthy. A hardened +// deployment leaves /_pivot/version reachable as the node-to-node probe +// endpoint, so it stays usable for health checks. GET (not HEAD) keeps the +// probe compatible with older nodes that registered /_pivot/version for GET +// only, so a rolling upgrade never marks a not-yet-upgraded node unhealthy. func (nh *NodeHealth) pingNode(node string) bool { select { case <-nh.ctx.Done(): diff --git a/nodehealth_pingnode_test.go b/nodehealth_pingnode_test.go index 60ec89b..1769683 100644 --- a/nodehealth_pingnode_test.go +++ b/nodehealth_pingnode_test.go @@ -3,6 +3,7 @@ package pivot import ( "net/http" "os" + "strings" "testing" "time" @@ -13,15 +14,15 @@ import ( ) // TestPingNodeProbesUnauthedEndpoint pins that pingNode can reach a hardened -// node — one whose ooo Audit denies requests — by probing the pivot-internal -// /_pivot/version route, which is registered without the Audit wrapper. +// node — one whose request gate denies the ooo root — by probing the +// pivot-internal /_pivot/version route, which the gate leaves reachable. // // Before the fix, pingNode issued GET "/" against the ooo root. The root is -// served by the UI handler, which returns 401 whenever a non-trivial Audit +// served by the UI handler, which returns 401 whenever the operator's gate // denies the request. Every health check against a hardened node therefore // failed, and the node was permanently marked unhealthy. The probe now targets -// /_pivot/version, which the Audit gate does not cover. (GET, not HEAD, so the -// probe stays compatible with older nodes whose route registered GET only.) +// /_pivot/version, which the gate exempts. (GET, not HEAD, so the probe stays +// compatible with older nodes whose route registered GET only.) func TestPingNodeProbesUnauthedEndpoint(t *testing.T) { server := &ooo.Server{} server.Silence = true @@ -29,27 +30,35 @@ func TestPingNodeProbesUnauthedEndpoint(t *testing.T) { server.Storage = storage.New(storage.LayeredConfig{Memory: storage.NewMemoryLayer()}) server.Router = mux.NewRouter() server.Client = &http.Client{Timeout: 500 * time.Millisecond} - // Hardened node: Audit denies everything, so the ooo root ("/") returns - // 401. /_pivot/version stays reachable because pivot registers it without - // the Audit wrapper. - server.Audit = func(r *http.Request) bool { return false } + // Hardened node: a gate middleware denies everything except the + // pivot-internal routes, so the ooo root ("/") returns 401 while + // /_pivot/version stays reachable (the probe endpoint pingNode must use). + server.Router.Use(func(next http.Handler) http.Handler { + return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if strings.HasPrefix(r.URL.Path, RoutePrefix+"/") { + next.ServeHTTP(w, r) + return + } + w.WriteHeader(http.StatusUnauthorized) + }) + }) Setup(server, Config{Keys: []Key{{Path: "items/*"}}, NodesKey: "nodes/*"}) server.Start("localhost:0") defer server.Close(os.Interrupt) - // Premise check: the ooo root IS behind the Audit gate, so a direct GET - // "/" on this hardened server is rejected. If this ever stops being 401, - // the bug this test guards against no longer exists. + // Premise check: the ooo root IS behind the gate, so a direct GET "/" on + // this hardened server is rejected. If this ever stops being 401, the bug + // this test guards against no longer exists. rootResp, err := http.Get("http://" + server.Address + "/") require.NoError(t, err) rootResp.Body.Close() require.Equal(t, http.StatusUnauthorized, rootResp.StatusCode, - "premise: the ooo root must be Audit-gated on a hardened node") + "premise: the ooo root must be gated on a hardened node") nh := NewNodeHealth(nil) defer nh.Stop() require.True(t, nh.pingNode(server.Address), - "pingNode must reach a hardened node via the Audit-exempt /_pivot/version route") + "pingNode must reach a hardened node via the gate-exempt /_pivot/version route") } diff --git a/offline_sync_test.go b/offline_sync_test.go index f80c88b..b3a1c43 100644 --- a/offline_sync_test.go +++ b/offline_sync_test.go @@ -121,7 +121,6 @@ func setupOfflineServers(t *testing.T) *OfflineTestServers { DisableKeepAlives: true, }, } - pivotServer.Audit = func(r *http.Request) bool { return true } // Configure pivot with policies using SEPARATE storage pivotConfig := pivot.Config{ @@ -182,7 +181,6 @@ func setupOfflineServers(t *testing.T) *OfflineTestServers { DisableKeepAlives: true, }, } - nodeServer.Audit = func(r *http.Request) bool { return true } // Configure node with policies using SEPARATE storage nodeConfig := pivot.Config{ diff --git a/sync_glob_clock_skew_test.go b/sync_glob_clock_skew_test.go index 97ba3d3..256699d 100644 --- a/sync_glob_clock_skew_test.go +++ b/sync_glob_clock_skew_test.go @@ -1,6 +1,7 @@ package pivot import ( + "context" "encoding/json" "net/http" "net/http/httptest" @@ -17,7 +18,7 @@ func TestGlobPullUsesVVAuthorityWhenTimestampsDisagree(t *testing.T) { localDB := storage.New(storage.LayeredConfig{Memory: storage.NewMemoryLayer()}) require.NoError(t, localDB.Start(storage.Options{})) defer localDB.Close() - storage.WatchWithCallback(localDB, func(storage.Event) {}) + storage.WatchWithCallback(context.Background(), localDB, func(storage.Event) {}) // Local has a future timestamp with stale payload. _, err := localDB.SetWithMeta("things/x", []byte(`{"v":"stale-local"}`), 9_999, 9_999) @@ -71,7 +72,7 @@ func TestGlobPushUsesVVAuthorityWhenTimestampsDisagree(t *testing.T) { localDB := storage.New(storage.LayeredConfig{Memory: storage.NewMemoryLayer()}) require.NoError(t, localDB.Start(storage.Options{})) defer localDB.Close() - storage.WatchWithCallback(localDB, func(storage.Event) {}) + storage.WatchWithCallback(context.Background(), localDB, func(storage.Event) {}) // Local has causally newer data but smaller wall-clock timestamp. _, err := localDB.SetWithMeta("things/x", []byte(`{"v":"local-causally-newer"}`), 100, 100) diff --git a/sync_singlekey_tombstone_vv_test.go b/sync_singlekey_tombstone_vv_test.go index f436ebe..2fd8a84 100644 --- a/sync_singlekey_tombstone_vv_test.go +++ b/sync_singlekey_tombstone_vv_test.go @@ -1,6 +1,7 @@ package pivot import ( + "context" "encoding/json" "net/http" "net/http/httptest" @@ -16,7 +17,7 @@ func TestSingleKeyPullIgnoresSkewedDeleteTimestampWhenVVLess(t *testing.T) { localDB := storage.New(storage.LayeredConfig{Memory: storage.NewMemoryLayer()}) require.NoError(t, localDB.Start(storage.Options{})) defer localDB.Close() - storage.WatchWithCallback(localDB, func(storage.Event) {}) + storage.WatchWithCallback(context.Background(), localDB, func(storage.Event) {}) // Simulate a skewed future tombstone timestamp for this key. _, err := localDB.Set(StoragePrefix+"things/x", []byte("999999999999999999")) diff --git a/sync_vv_divergence_test.go b/sync_vv_divergence_test.go index 43411c1..9b1da0d 100644 --- a/sync_vv_divergence_test.go +++ b/sync_vv_divergence_test.go @@ -13,6 +13,7 @@ package pivot // (backward compatibility with old peers). import ( + "context" "encoding/json" "net/http" "net/http/httptest" @@ -60,7 +61,7 @@ func TestSyncDetectsVVDivergenceOnEqualLastEntryPullCase(t *testing.T) { localDB := storage.New(storage.LayeredConfig{Memory: storage.NewMemoryLayer()}) require.NoError(t, localDB.Start(storage.Options{})) defer localDB.Close() - storage.WatchWithCallback(localDB, func(storage.Event) {}) + storage.WatchWithCallback(context.Background(), localDB, func(storage.Event) {}) collidedTS := int64(1_700_000_000_000_000_000) // Seed a single local item so checkActivity returns collidedTS. @@ -117,7 +118,7 @@ func TestSyncDetectsVVDivergenceOnEqualLastEntryPushCase(t *testing.T) { localDB := storage.New(storage.LayeredConfig{Memory: storage.NewMemoryLayer()}) require.NoError(t, localDB.Start(storage.Options{})) defer localDB.Close() - storage.WatchWithCallback(localDB, func(storage.Event) {}) + storage.WatchWithCallback(context.Background(), localDB, func(storage.Event) {}) collidedTS := int64(1_700_000_000_000_000_000) obj := meta.Object{Created: collidedTS, Updated: collidedTS, Index: "x", Path: "things/x", Data: []byte(`{"v":"local"}`)} @@ -172,7 +173,7 @@ func TestSyncPushDirectionPushesWhenItemTimestampsAllow(t *testing.T) { localDB := storage.New(storage.LayeredConfig{Memory: storage.NewMemoryLayer()}) require.NoError(t, localDB.Start(storage.Options{})) defer localDB.Close() - storage.WatchWithCallback(localDB, func(storage.Event) {}) + storage.WatchWithCallback(context.Background(), localDB, func(storage.Event) {}) leaderTS := int64(1_700_000_000_000_000_000) localTS := leaderTS + 1_000_000 // strictly newer item @@ -210,7 +211,7 @@ func TestSyncSkipsWhenVVsAreEqual(t *testing.T) { localDB := storage.New(storage.LayeredConfig{Memory: storage.NewMemoryLayer()}) require.NoError(t, localDB.Start(storage.Options{})) defer localDB.Close() - storage.WatchWithCallback(localDB, func(storage.Event) {}) + storage.WatchWithCallback(context.Background(), localDB, func(storage.Event) {}) collidedTS := int64(1_700_000_000_000_000_000) vvm := NewVVManager(localDB, "127.0.0.1:9000") @@ -253,7 +254,7 @@ func TestSyncConcurrentVVsLogsAndPullsLeader(t *testing.T) { localDB := storage.New(storage.LayeredConfig{Memory: storage.NewMemoryLayer()}) require.NoError(t, localDB.Start(storage.Options{})) defer localDB.Close() - storage.WatchWithCallback(localDB, func(storage.Event) {}) + storage.WatchWithCallback(context.Background(), localDB, func(storage.Event) {}) collidedTS := int64(1_700_000_000_000_000_000) obj := meta.Object{Created: collidedTS, Updated: collidedTS, Index: "x", Path: "things/x", Data: []byte(`{"v":"local"}`)} @@ -309,7 +310,7 @@ func TestSyncFallsBackToLastEntryWhenLeaderHasNoVV(t *testing.T) { localDB := storage.New(storage.LayeredConfig{Memory: storage.NewMemoryLayer()}) require.NoError(t, localDB.Start(storage.Options{})) defer localDB.Close() - storage.WatchWithCallback(localDB, func(storage.Event) {}) + storage.WatchWithCallback(context.Background(), localDB, func(storage.Event) {}) leaderTS := int64(1_700_000_000_000_000_500) leaderItem := meta.Object{Created: leaderTS, Updated: leaderTS, Index: "x", Path: "things/x", Data: []byte(`{"v":"from-leader"}`)} @@ -343,7 +344,7 @@ func TestSyncFallsBackToLastEntryWhenLocalHasNoVV(t *testing.T) { localDB := storage.New(storage.LayeredConfig{Memory: storage.NewMemoryLayer()}) require.NoError(t, localDB.Start(storage.Options{})) defer localDB.Close() - storage.WatchWithCallback(localDB, func(storage.Event) {}) + storage.WatchWithCallback(context.Background(), localDB, func(storage.Event) {}) leaderTS := int64(1_700_000_000_000_000_500) leaderItem := meta.Object{Created: leaderTS, Updated: leaderTS, Index: "x", Path: "things/x", Data: []byte(`{"v":"from-leader"}`)} diff --git a/synctoleader_tombstone_test.go b/synctoleader_tombstone_test.go index 0653d6f..e82e0d7 100644 --- a/synctoleader_tombstone_test.go +++ b/synctoleader_tombstone_test.go @@ -12,6 +12,7 @@ package pivot // Both must reach the same conclusion. import ( + "context" "encoding/json" "net/http" "net/http/httptest" @@ -27,8 +28,8 @@ import ( func TestSyncToLeaderHonoursTombstone(t *testing.T) { cases := []struct { - name string - emitHeader bool + name string + emitHeader bool serveActivity bool }{ {name: "new_leader_with_header", emitHeader: true, serveActivity: false}, @@ -45,7 +46,7 @@ func TestSyncToLeaderHonoursTombstone(t *testing.T) { if err := localDB.Start(storage.Options{}); err != nil { t.Fatal(err) } - storage.WatchWithCallback(localDB, func(storage.Event) {}) + storage.WatchWithCallback(context.Background(), localDB, func(storage.Event) {}) defer localDB.Close() oldTime := time.Now().UTC().Add(-2 * time.Hour).UnixNano() diff --git a/version_e2e_test.go b/version_e2e_test.go index 137536c..e53d4d9 100644 --- a/version_e2e_test.go +++ b/version_e2e_test.go @@ -34,7 +34,6 @@ func VersionTestServer(t *testing.T, clusterURL string) *ooo.Server { server.Storage = storage.New(storage.LayeredConfig{Memory: storage.NewMemoryLayer()}) server.Router = mux.NewRouter() server.Client = &http.Client{Timeout: 500 * time.Millisecond} - server.Audit = func(r *http.Request) bool { return true } config := pivot.Config{ Keys: []pivot.Key{{Path: "items/*"}}, diff --git a/version_vector_internal_test.go b/version_vector_internal_test.go index 17e429b..b6de9ab 100644 --- a/version_vector_internal_test.go +++ b/version_vector_internal_test.go @@ -7,6 +7,7 @@ package pivot import ( "bytes" + "context" "encoding/json" "errors" "log" @@ -43,7 +44,7 @@ func BenchmarkVVManagerIncrement(b *testing.B) { if err := db.Start(storage.Options{}); err != nil { b.Fatal(err) } - storage.WatchWithCallback(db, func(storage.Event) {}) + storage.WatchWithCallback(context.Background(), db, func(storage.Event) {}) b.Cleanup(func() { db.Close() }) m := NewVVManager(db, "leader") diff --git a/vv_bump_bench_test.go b/vv_bump_bench_test.go index 1bc206c..8655398 100644 --- a/vv_bump_bench_test.go +++ b/vv_bump_bench_test.go @@ -60,7 +60,6 @@ func benchPivotServer(b *testing.B) (*ooo.Server, storage.Database, func()) { DisableKeepAlives: true, }, } - server.Audit = func(r *http.Request) bool { return true } pivot.Setup(server, pivot.Config{ Keys: []pivot.Key{{Path: "policies", Database: policiesStorage}}, diff --git a/vv_glob_pull_delete_test.go b/vv_glob_pull_delete_test.go index b06d293..be0cceb 100644 --- a/vv_glob_pull_delete_test.go +++ b/vv_glob_pull_delete_test.go @@ -91,7 +91,6 @@ func setupGlobServers(t *testing.T) *globServers { pivotServer.Storage = storage.New(storage.LayeredConfig{Memory: storage.NewMemoryLayer()}) pivotServer.Router = mux.NewRouter() pivotServer.Client = mkClient() - pivotServer.Audit = func(r *http.Request) bool { return true } pivot.Setup(pivotServer, pivot.Config{ Keys: []pivot.Key{{Path: "things/*", Database: pivotThings}}, ClusterURL: "", @@ -107,7 +106,6 @@ func setupGlobServers(t *testing.T) *globServers { nodeServer.Storage = storage.New(storage.LayeredConfig{Memory: storage.NewMemoryLayer()}) nodeServer.Router = mux.NewRouter() nodeServer.Client = mkClient() - nodeServer.Audit = func(r *http.Request) bool { return true } pivot.Setup(nodeServer, pivot.Config{ Keys: []pivot.Key{{Path: "things/*", Database: nodeThings}}, ClusterURL: pivotServer.Address, diff --git a/vv_merge_test.go b/vv_merge_test.go index 85af7c9..cd73556 100644 --- a/vv_merge_test.go +++ b/vv_merge_test.go @@ -36,7 +36,7 @@ func TestSetHandlerMergesInboundVV(t *testing.T) { db := storage.New(storage.LayeredConfig{Memory: storage.NewMemoryLayer()}) require.NoError(t, db.Start(storage.Options{})) defer db.Close() - storage.WatchWithCallback(db, func(storage.Event) {}) + storage.WatchWithCallback(context.Background(), db, func(storage.Event) {}) vvm := NewVVManager(db, "leader") handler := Set(db, "things", NewHandlerWriteTracker(), vvm, nil) @@ -70,7 +70,7 @@ func TestSetHandlerNoMergeWithoutHeader(t *testing.T) { db := storage.New(storage.LayeredConfig{Memory: storage.NewMemoryLayer()}) require.NoError(t, db.Start(storage.Options{})) defer db.Close() - storage.WatchWithCallback(db, func(storage.Event) {}) + storage.WatchWithCallback(context.Background(), db, func(storage.Event) {}) vvm := NewVVManager(db, "leader") handler := Set(db, "things", NewHandlerWriteTracker(), vvm, nil) @@ -98,7 +98,7 @@ func TestDeleteHandlerMergesInboundVV(t *testing.T) { db := storage.New(storage.LayeredConfig{Memory: storage.NewMemoryLayer()}) require.NoError(t, db.Start(storage.Options{})) defer db.Close() - storage.WatchWithCallback(db, func(storage.Event) {}) + storage.WatchWithCallback(context.Background(), db, func(storage.Event) {}) // Seed an item so Delete has something to remove. _, err := db.SetWithMeta("things/x", []byte(`{"v":"v1"}`), 1, 1) diff --git a/vv_path_scope_test.go b/vv_path_scope_test.go index d4c0330..22c4b91 100644 --- a/vv_path_scope_test.go +++ b/vv_path_scope_test.go @@ -15,6 +15,7 @@ package pivot // share state. import ( + "context" "net/http/httptest" "strings" "testing" @@ -45,7 +46,7 @@ func TestActivityExposesVVAfterGlobWrite(t *testing.T) { keys := []Key{{Path: "things/*", Database: db}} instance := &Instance{VVManager: vvm} - storage.WatchWithCallback(db, makeStorageSync(StorageSyncConfig{ + storage.WatchWithCallback(context.Background(), db, makeStorageSync(StorageSyncConfig{ Keys: keys, GetNodes: func() []string { return nil }, HandlerTracker: tracker, @@ -97,7 +98,7 @@ func TestCallbackIncrementsAtPathScope(t *testing.T) { vvm := NewVVManager(db, "leader") keys := []Key{{Path: "things/*", Database: db}} instance := &Instance{VVManager: vvm} - storage.WatchWithCallback(db, makeStorageSync(StorageSyncConfig{ + storage.WatchWithCallback(context.Background(), db, makeStorageSync(StorageSyncConfig{ Keys: keys, GetNodes: func() []string { return nil }, Instance: instance, @@ -133,7 +134,7 @@ func TestHandlerIncrementMatchesActivityScope(t *testing.T) { keys := []Key{{Path: "things/*", Database: db}} instance := &Instance{VVManager: vvm} - storage.WatchWithCallback(db, makeStorageSync(StorageSyncConfig{ + storage.WatchWithCallback(context.Background(), db, makeStorageSync(StorageSyncConfig{ Keys: keys, GetNodes: func() []string { return nil }, HandlerTracker: tracker,