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,