Fix handler VV bump ordering for PR #66

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
 )

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=

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

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,30 +486,28 @@ 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"
 
 	// 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,19 +517,18 @@ 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"
 
 	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")
 }

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,22 +260,21 @@ 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
 	}
 	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

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

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 {

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