test(cluster): fix flaky sync hang via lossless watch; adopt store update

.gitignore

@@ -2,4 +2,5 @@
 pivot.test
 test_*
 test/
-*.code-workspace
+*.code-workspace
+.flaky/

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"}},

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}},

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()})

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{

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"}},

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
 )

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=

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"

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)

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
 }
 

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()})