diff --git a/go.mod b/go.mod index 68b7690..9addab6 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-20260608034142-44aea235aa2d github.com/gorilla/mux v1.8.1 github.com/stretchr/testify v1.11.1 ) diff --git a/go.sum b/go.sum index 3625b30..1c6deb5 100644 --- a/go.sum +++ b/go.sum @@ -12,8 +12,8 @@ github.com/benitogf/jsonpatch v0.0.0-20260413094158-a4a6cc1a3382 h1:invi4tSUoH4H github.com/benitogf/jsonpatch v0.0.0-20260413094158-a4a6cc1a3382/go.mod h1:ZOQm93UFJwYDZLHQVGt2/JmNOltb6LRIJUuys0uMWKY= github.com/benitogf/ko v0.0.0-20260211072652-d48fcf4f8988 h1:sagWGc0GUBEMxa56HpjlYh6MmUt2O4vZrqlTspHotYc= 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-20260608034142-44aea235aa2d h1:OQH3vNscklTW6rFMcc5uQzORfRJ/a3wUCUPEyPhvvqw= +github.com/benitogf/ooo v0.0.0-20260608034142-44aea235aa2d/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.go b/handlers.go index 1bd5968..da11520 100644 --- a/handlers.go +++ b/handlers.go @@ -91,10 +91,18 @@ func GetSingle(db storage.Database, path string) func(w http.ResponseWriter, r * // peer woken by the trigger sees the bumped VV, not a stale one. type PostWriteFunc func(itemKey, op, originatorPeer string) +func handlerNeedsFallbackVVBump(handlerTracker *HandlerWriteTracker, itemKey string) bool { + if handlerTracker == nil { + return true + } + return handlerTracker.ConsumeBumpFallback(itemKey) +} + // Set set data on the pivot instance -// handlerTracker records "a handler will own the post-write work for this -// key" so the async storage callback skips its bump+fanout for this event; -// the handler does both, in order, after the storage write succeeds. +// handlerTracker records "a handler will own the post-write fanout for this +// key" so the async storage callback skips its fanout for this event. +// Attached storages bump through AfterWriteOp; unattached storages fall back +// to the handler after the storage write succeeds. // vvManager is the version vector manager for pivot servers (nil for nodes). // postWrite, if non-nil, runs synchronously after a successful VV bump to // fan out the change to peers. May be nil in tests that don't need fanout. @@ -163,7 +171,9 @@ func Set(db storage.Database, path string, handlerTracker *HandlerWriteTracker, // complete BEFORE we trigger peers — otherwise a peer woken by // the trigger could read /activity and see the pre-bump VV. if vvManager != nil { - vvManager.increment(path) + if handlerNeedsFallbackVVBump(handlerTracker, itemKey) { + vvManager.increment(path) + } // Merge-on-receive: integrate the originator's peer counters // into local VV. Without this, each node's VV is just // {"my-id": counter} and cross-node Compare always returns @@ -254,7 +264,9 @@ func Delete(db storage.Database, path string, handlerTracker *HandlerWriteTracke // reads a fresh VV. Merge-on-receive integrates the originator's // peer counters into local VV. if vvManager != nil { - vvManager.increment(path) + if handlerNeedsFallbackVVBump(handlerTracker, itemKey) { + vvManager.increment(path) + } if peerVV, ok := decodeVVHeader(r.Header.Get(VVHeader)); ok { vvManager.set(path, peerVV) } diff --git a/handlers_internal_test.go b/handlers_internal_test.go index ff88776..8d40020 100644 --- a/handlers_internal_test.go +++ b/handlers_internal_test.go @@ -477,7 +477,7 @@ 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 +// bumpPendingLen mirrors pendingLen for the fallback-bump counter so tests // can assert it drains independently. func (t *HandlerWriteTracker) bumpPendingLen() int { t.mu.Lock() @@ -486,11 +486,10 @@ func (t *HandlerWriteTracker) bumpPendingLen() int { } // TestHandlerWriteTracker_DualCounterConsumeSemantics pins that Mark -// sets BOTH the fanout-skip (pending) and bump-skip (bumpPending) +// sets BOTH the fanout-skip (pending) and fallback-bump (bumpPending) // counters, and each is consumed by exactly one consumer without -// depriving the other. The earlier Has() peek was non-consuming, so a -// stale mark could silently swallow a later direct write's VV bump when -// the watch goroutine never ran to drain it. +// depriving the other. The fallback marker is consumed so the handler +// can tell whether AfterWriteOp already performed the VV bump. func TestHandlerWriteTracker_DualCounterConsumeSemantics(t *testing.T) { tr := NewHandlerWriteTracker() const k = "things/abc" @@ -498,18 +497,17 @@ func TestHandlerWriteTracker_DualCounterConsumeSemantics(t *testing.T) { // Mark once: both counters carry the key. 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") + require.Equal(t, 1, tr.bumpPendingLen(), "Mark must set fallback-bump counter") - // AfterWrite consumes its own counter. The watch goroutine's + // AfterWriteOp consumes its own counter. The watch goroutine's // counter is untouched — the two consumers don't deprive each other. - require.True(t, tr.ConsumeBumpSkip(k), "first ConsumeBumpSkip sees the mark") - require.Equal(t, 0, tr.bumpPendingLen(), "ConsumeBumpSkip drains bump-skip counter") + require.True(t, tr.ConsumeBumpFallback(k), "first ConsumeBumpFallback sees the mark") + require.Equal(t, 0, tr.bumpPendingLen(), "ConsumeBumpFallback drains fallback-bump 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)") + // Consuming, not peeking: a second consume returns false. This tells + // the handler no fallback bump is needed. + require.False(t, tr.ConsumeBumpFallback(k), "second ConsumeBumpFallback must return false") // Watch goroutine consumes its counter independently. require.True(t, tr.Consume(k), "Consume sees the fanout-skip mark") @@ -519,8 +517,7 @@ func TestHandlerWriteTracker_DualCounterConsumeSemantics(t *testing.T) { // TestHandlerWriteTracker_UnmarkClearsBothCounters pins that the error // path (handler bails after Mark but before the storage write fires an -// event) drains BOTH counters — otherwise a leaked bump-skip mark would -// swallow a later direct write's VV bump for the same key. +// event) drains BOTH counters. func TestHandlerWriteTracker_UnmarkClearsBothCounters(t *testing.T) { tr := NewHandlerWriteTracker() const k = "things/abc" @@ -528,10 +525,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") + require.Equal(t, 0, tr.bumpPendingLen(), "Unmark must clear fallback-bump counter") // After Unmark, neither consumer sees a mark — a subsequent direct // write to the same key will correctly run its full path. - require.False(t, tr.ConsumeBumpSkip(k), "no bump-skip mark after Unmark") + require.False(t, tr.ConsumeBumpFallback(k), "no fallback-bump mark after Unmark") require.False(t, tr.Consume(k), "no fanout-skip mark after Unmark") } diff --git a/instance.go b/instance.go index fac8528..1e12d18 100644 --- a/instance.go +++ b/instance.go @@ -37,7 +37,7 @@ type Instance struct { ExtraNodeURLs []string // Additional node URLs (can be modified after Setup) VVManager *VVManager // Version vector manager (for both pivot and node servers) configKeys []Key // Configured keys from Setup; needed by the synchronous AfterWrite VV bump to find which base path scope to increment - handlerTracker *HandlerWriteTracker // Same tracker SyncCallback consumes from; AfterWrite peeks it (non-consuming) to skip its bump when a handler will bump explicitly + handlerTracker *HandlerWriteTracker // Same tracker SyncCallback consumes from; AfterWriteOp consumes handler fallback markers before bumping attachedDBs sync.Map // set of storage.Database -> struct{}; AfterWrite-driven sync bump is wired for these, so makeStorageSync's async bump must skip them to avoid double-counting syncerPool *syncerPool // Internal syncer pool for node servers (for testing hooks) nodesCache *nodesCache // Cache for NodesKey address list, invalidated by storage events @@ -245,7 +245,7 @@ func (i *Instance) IsAttached(db storage.Database) bool { // pushed write as VVEqual — a permanent node↔pivot divergence. Consuming only // the mark matching this write's own operation closes that cross-op steal. // -// Four cases skip the bump: +// Three cases skip the bump: // // - eventKey is a pivot-internal key (delete tombstones, VV storage, // health, etc.) — these are pivot's own bookkeeping, not application @@ -260,13 +260,12 @@ func (i *Instance) IsAttached(db storage.Database) bool { // is merged into local via vvManager.set later; bumping our own // counter on top would double-count. // -// - a handler has marked the key (HandlerWriteTracker.Has) — handlers -// bump explicitly after SetWithMeta returns, so AfterWrite would -// duplicate. The watch goroutine still Consumes the mark to gate -// the rest of SyncCallback; we only peek here. +// Handler writes also bump here, synchronously. The hook consumes the +// handler's fallback marker so the handler knows not to bump again after +// SetWithMeta/Del returns. // -// Direct user writes (db.Set / db.SetWithMeta on an attached storage) -// hit none of these cases and bump here, synchronously. +// Direct user writes (db.Set / db.SetWithMeta on an attached storage) hit none +// of the skip cases and bump here, synchronously. func (i *Instance) bumpVVForLocalWrite(eventKey string, op string) { if i.VVManager == nil { return @@ -274,8 +273,8 @@ func (i *Instance) bumpVVForLocalWrite(eventKey string, op string) { if strings.HasPrefix(eventKey, StoragePrefix) { return } - if i.handlerTracker != nil && i.handlerTracker.ConsumeBumpSkip(eventKey) { - return + if i.handlerTracker != nil { + i.handlerTracker.ConsumeBumpFallback(eventKey) } var matched Key found := false diff --git a/pivot.go b/pivot.go index 5f6145c..41fe4b4 100644 --- a/pivot.go +++ b/pivot.go @@ -619,10 +619,10 @@ func SetupWithError(server *ooo.Server, config Config) (*ooo.Server, error) { // Handler-write tracker is created on every server. The Set/Delete // handlers Mark before each storage write so the async storage event - // callback knows to skip its own bump+fanout for handler-driven events. + // callback knows to skip its own fanout for handler-driven events. handlerTracker := NewHandlerWriteTracker() - // AfterWrite (sync-bump path) needs to peek the same tracker so it - // skips its bump when a handler will bump explicitly post-SetWithMeta. + // AfterWriteOp consumes the same tracker's fallback-bump marker so the + // handler can tell whether the successful-write hook already bumped. instance.handlerTracker = handlerTracker // AfterWrite also matches eventKey against the configured Keys to find // the base path scope to increment. @@ -738,12 +738,12 @@ func SetupWithError(server *ooo.Server, config Config) (*ooo.Server, error) { // synchronous VV bump on the writer's goroutine, closing the // VV-lag race where /activity could return a pre-bump VV between // a write committing and the watch goroutine processing the event. - // The op-aware hook lets bumpVVForLocalWrite consume only the - // bump-skip mark matching the write's operation (set vs del), so a - // pulled delete's mark can't suppress a concurrent local set's bump. + // The op-aware hook lets bumpVVForLocalWrite consume only the pull mark + // matching the write's operation (set vs del), so a pulled delete's mark + // can't suppress a concurrent local set's bump. // instance.bumpVVForLocalWrite handles internal-prefix skip, // configured-key matching, pull-driven skip via the pullTracker - // consume, and handler-write skip via the HandlerWriteTracker peek. + // consume, and handler fallback-marker consumption via HandlerWriteTracker. server.AfterWriteOp = instance.bumpVVForLocalWrite // Record server.Storage as sync-bump-installed so makeStorageSync's // async-path bump skips events from it (the AfterWrite above already diff --git a/sync.go b/sync.go index ba36027..4b9f0e8 100644 --- a/sync.go +++ b/sync.go @@ -1202,17 +1202,15 @@ type StorageSyncCallback func(event storage.Event) // false and the callback runs its full path — that's how the callback // remains the source of truth for direct writes. // Two independent consumers recognize a handler-driven write, mirroring -// pullTracker: the watch goroutine (Consume, to skip its bump+fanout) -// and AfterWrite (ConsumeBumpSkip, to skip its synchronous bump because -// the handler bumps explicitly). Each drains its own counter; a single -// counter peeked by one and consumed by the other left a stale mark -// whenever the watch goroutine never ran (a pivot-synced key in -// NoBroadcastKeys, or a storage error before dispatch), which then -// swallowed a later direct write's legitimate bump. +// pullTracker: the watch goroutine (Consume, to skip fanout) and the VV +// bump path (ConsumeBumpFallback). If AfterWriteOp is installed, it consumes +// the fallback marker and bumps before storage callbacks can observe the write. +// If no hook is installed, the handler consumes the still-present marker +// after the write and performs the fallback bump. type HandlerWriteTracker struct { mu sync.Mutex pending map[string]int // consumed by the watch goroutine (fanout-skip) - bumpPending map[string]int // consumed by AfterWrite (bump-skip) + bumpPending map[string]int // consumed by AfterWriteOp or handler fallback } // NewHandlerWriteTracker creates an empty tracker. @@ -1269,12 +1267,11 @@ func (t *HandlerWriteTracker) Consume(key string) bool { return true } -// ConsumeBumpSkip returns true and decrements the bump-skip count if a -// mark is present (i.e. a handler will bump the VV for this write, so -// AfterWrite must skip). Consumed by bumpVVForLocalWrite. Replaces the -// earlier non-consuming Has peek — see the type doc for why consuming -// matters. -func (t *HandlerWriteTracker) ConsumeBumpSkip(key string) bool { +// ConsumeBumpFallback returns true and decrements the fallback-bump count if a +// handler mark is present. AfterWriteOp consumes this marker when it owns +// the successful-write bump; the handler consumes it only when no hook ran. +// Earlier non-consuming peeks leaked stale markers, so this must consume. +func (t *HandlerWriteTracker) ConsumeBumpFallback(key string) bool { t.mu.Lock() defer t.mu.Unlock() if t.bumpPending[key] == 0 { diff --git a/vv_path_scope_test.go b/vv_path_scope_test.go index d4c0330..ea0bfac 100644 --- a/vv_path_scope_test.go +++ b/vv_path_scope_test.go @@ -117,6 +117,53 @@ func TestCallbackIncrementsAtPathScope(t *testing.T) { "storage callback must increment path-scope VV on direct writes") } +// TestAttachedHandlerAfterWriteObservesSynchronousVVBump pins the handler +// hook-ordering contract pivot relies on for external storages. A caller +// waiting on its AfterWrite callback must observe the VV bump already applied; +// otherwise tests and peer pushes can see a committed write with an empty VV. +func TestAttachedHandlerAfterWriteObservesSynchronousVVBump(t *testing.T) { + monotonic.Init() + dataDB := storage.New(storage.LayeredConfig{Memory: storage.NewMemoryLayer()}) + vvDB := storage.New(storage.LayeredConfig{Memory: storage.NewMemoryLayer()}) + require.NoError(t, vvDB.Start(storage.Options{})) + defer vvDB.Close() + + vvm := NewVVManager(vvDB, LeaderID) + tracker := NewHandlerWriteTracker() + observedVV := make(chan VersionVector, 1) + instance := &Instance{ + VVManager: vvm, + configKeys: []Key{{Path: "policies", Database: dataDB}}, + handlerTracker: tracker, + SyncCallback: func(storage.Event) {}, + } + + require.NoError(t, instance.Attach(dataDB, storage.Options{ + AfterWrite: func(eventKey string) { + if eventKey == "policies" { + observedVV <- vvm.Get("policies") + } + }, + })) + defer dataDB.Close() + + handler := Set(dataDB, "policies", tracker, vvm, nil) + body := strings.NewReader(`{"created":1,"updated":1,"index":"policies","path":"policies","data":"eyJ2IjoicGhhc2UtMSJ9"}`) + req := httptest.NewRequest("POST", "/_pivot/pivot/policies", body) + req.Header.Set("Content-Type", "application/json") + w := httptest.NewRecorder() + handler(w, req) + require.Equal(t, 200, w.Code) + + select { + case vv := <-observedVV: + require.Equal(t, int64(1), vv[LeaderID], + "AfterWrite must not fire before the synchronous VV bump; got %v", vv) + case <-time.After(time.Second): + t.Fatal("timed out waiting for AfterWrite callback") + } +} + // TestHandlerIncrementMatchesActivityScope pins the symmetry: every // handler-driven write produces a VV that Activity exposes 1:1. If a // future change re-introduces an item-scope increment, OR if the