handle weird ancestry during commit

effects/commit_ancestry_test.go

@@ -0,0 +1,168 @@
+package effects
+
+import (
+	"errors"
+	"sync"
+	"testing"
+
+	pb "github.com/swytchdb/engine/cluster/proto"
+)
+
+func commitAncestryFixture(t *testing.T) (*Engine, Tip, Tip, *pb.Effect) {
+	t.Helper()
+
+	e := newTxnTestEngine(nil)
+	historicalTestWrite(t, e, map[string]string{"k": "P"}, false)
+	p := e.index.Contains("k").Tips()[0]
+
+	var (
+		bindTip Tip
+		bind    *pb.Effect
+	)
+
+	e.SetOnLocalEffect(func(tip Tip, eff *pb.Effect) {
+		if eff.GetTxnBind() != nil {
+			bindTip, bind = tip, eff
+		}
+	})
+	historicalTestWrite(t, e, map[string]string{"a": "B", "k": "B"}, true)
+	e.SetOnLocalEffect(nil)
+
+	if bind == nil {
+		t.Fatal("missing fixture bind")
+	}
+
+	// Record the same observation as remote bind ingestion, and retain P
+	// alongside its descendant, as additional-key indexing can do.
+	e.handleRemoteBind(bind.GetTxnBind(), bindTip, bind.GetTxnId())
+	e.updateIndex("k", nil, p)
+
+	return e, p, bindTip, bind
+}
+
+func TestFlushTxRedundantConsumedAncestor(t *testing.T) {
+	for _, write := range []bool{false, true} {
+		name := "read-only"
+		if write {
+			name = "read-write"
+		}
+
+		t.Run(name, func(t *testing.T) {
+			e, p, _, _ := commitAncestryFixture(t)
+			for range 3 {
+				tx := e.NewContext()
+				tx.BeginTx()
+
+				result, _, err := tx.GetSnapshot("k")
+				if err != nil || string(result.GetScalar().GetRaw()) != "B" {
+					t.Fatalf("read: result=%v err=%v", result, err)
+				}
+
+				if !tx.keys["k"].initialTips.Contains(p) {
+					t.Fatal("fixture lost the redundant ancestor before commit")
+				}
+
+				if write {
+					if err := tx.Emit(scalarSet("other", "committed")); err != nil {
+						t.Fatal(err)
+					}
+				}
+
+				if err := tx.Flush(); err != nil {
+					t.Fatalf("redundant ancestor prevented progress: %v", err)
+				}
+
+				e.updateIndex("k", nil, p)
+			}
+		})
+	}
+}
+
+func TestCheckCompetingBindsAncestry(t *testing.T) {
+	for _, dependency := range []string{"shared-parent", "bind", "descendant"} {
+		t.Run(dependency, func(t *testing.T) {
+			e, p, b, prior := commitAncestryFixture(t)
+			consumed := []Tip{p}
+
+			switch dependency {
+			case "bind":
+				consumed = append(consumed, b)
+			case "descendant":
+				historicalTestWrite(t, e, map[string]string{"k": "C"}, false)
+				consumed = append(consumed, e.index.Contains("k").Tips()...)
+			}
+
+			candidate := &pb.TransactionalBindEffect{Keys: []*pb.TransactionalBindEffect_KeyBind{
+				{Key: []byte("k"), ConsumedTips: toPbRefs(consumed)},
+			}}
+
+			want := ""
+			if dependency == "shared-parent" {
+				want = prior.GetTxnId()
+			}
+
+			if got := e.checkCompetingBinds(candidate, "candidate"); got != want {
+				t.Fatalf("competing bind: got %q, want %q", got, want)
+			}
+		})
+	}
+}
+
+func TestFlushTxRequiresEverySpokenBind(t *testing.T) {
+	e, p, _, _ := commitAncestryFixture(t)
+	// Another bind also consumed P, but is not an ancestor of our snapshot.
+	// It may have been recorded before its index transition completed.
+	other := Tip{99, 1}
+	e.handleRemoteBind(&pb.TransactionalBindEffect{Keys: []*pb.TransactionalBindEffect_KeyBind{
+		{Key: []byte("k"), ConsumedTips: []*pb.EffectRef{toPbRef(p)}, NewTip: toPbRef(Tip{99, 2})},
+	}}, other, "other")
+
+	tx := e.NewContext()
+	tx.BeginTx()
+
+	if _, _, err := tx.GetSnapshot("k"); err != nil {
+		t.Fatal(err)
+	}
+
+	if err := tx.Emit(scalarSet("unrelated", "must not commit")); err != nil {
+		t.Fatal(err)
+	}
+
+	if err := tx.Flush(); !errors.Is(err, ErrTxnAborted) {
+		t.Fatalf("unproven consuming bind should abort, got %v", err)
+	}
+
+	if result, _, _, err := e.GetSnapshot("unrelated"); err != nil || result != nil {
+		t.Fatalf("aborted transaction leaked writes: result=%v err=%v", result, err)
+	}
+}
+
+func TestSpokenBindsRetainsConcurrentConsumers(t *testing.T) {
+	e := newTxnTestEngine(nil)
+	p := Tip{99, 1}
+	bind := &pb.TransactionalBindEffect{Keys: []*pb.TransactionalBindEffect_KeyBind{
+		{Key: []byte("k"), ConsumedTips: []*pb.EffectRef{toPbRef(p)}},
+	}}
+	e.handleRemoteBind(bind, Tip{99, 2}, "first")
+	history, _ := e.spokenBinds.Get(p, 0)
+	before := history.snapshot()
+
+	var writers sync.WaitGroup
+	for i := range 16 {
+		writers.Go(func() {
+			for range 2 { // Re-delivery must not grow the list.
+				e.handleRemoteBind(bind, Tip{99, uint64(i + 3)}, "later")
+			}
+		})
+	}
+
+	writers.Wait()
+
+	if got, _ := e.spokenBinds.Get(p, 0); len(got.snapshot()) != 17 {
+		t.Fatalf("lost or duplicated a consuming bind: %v", got)
+	}
+
+	if len(before) != 1 || before[0] != (Tip{99, 2}) {
+		t.Fatalf("mutated an already-published list: %v", before)
+	}
+}

effects/context.go

@@ -1199,6 +1199,8 @@ func (c *Context) flushTx() error {
 	// In (b), our emit won't dep-reference the remote bind so the two
 	// commit DAG-concurrent. Abort here rather than let fork-choice
 	// surface the inconsistency as G1a / incompatible-order later.
+	var spokenBinds []Tip
+
 	for key, ck := range c.keys {
 		if ck.initialTips == nil {
 			continue // first write to a brand-new key
@@ -1236,20 +1238,11 @@ func (c *Context) flushTx() error {
 
 				return ErrTxnAborted
 			}
-			// Abort if we already ACKed/NACKed a remote bind that consumed
-			// this tip — emitting now would commit DAG-concurrent with it.
+			// A retained consumed tip may be redundant ancestry. Remember the
+			// consuming binds so we can check actual reachability below.
 			if c.engine.spokenBinds != nil {
-				if _, spoken := c.engine.spokenBinds.Get(t, 0); spoken {
-					slog.Debug("flushTx: pre-flight aborting, spoke on consumed tip",
-						"key", key, "tip", t)
-
-					for _, ck3 := range c.keys {
-						c.engine.pendingTxTips.Delete(ck3.lastOffset)
-					}
-
-					c.reset()
-
-					return ErrTxnAborted
+				if binds, spoken := c.engine.spokenBinds.Get(t, 0); spoken {
+					spokenBinds = append(spokenBinds, binds.snapshot()...)
 				}
 			}
 		}
@@ -1282,6 +1275,29 @@ func (c *Context) flushTx() error {
 		}
 	}
 
+	if len(spokenBinds) > 0 {
+		// Reuse the NACK conflict check's DAG walk over our actual emitted
+		// dependencies. Including P alone is not enough: every known consumer
+		// of P must be in our past, even if ingestion has not indexed it yet.
+		candidate := &pendingTxn{keys: make([]pendingTxnKey, 0, len(c.keys))}
+		for _, ck := range c.keys {
+			candidate.keys = append(candidate.keys, pendingTxnKey{newTip: ck.lastOffset})
+		}
+
+		if !c.engine.reachesFromTx(candidate, spokenBinds...) {
+			slog.Debug("flushTx: pre-flight aborting, spoken bind is not an ancestor",
+				"binds", spokenBinds)
+
+			for _, ck := range c.keys {
+				c.engine.pendingTxTips.Delete(ck.lastOffset)
+			}
+
+			c.reset()
+
+			return ErrTxnAborted
+		}
+	}
+
 	// Step 1: Update index + broadcast individual effects per key
 	for key, ck := range c.keys {
 		c.engine.updateIndex(key, ck.initialTips, ck.lastOffset)

effects/effect.go

@@ -238,13 +238,12 @@ type Engine struct {
 	// Horizon wait for bind visibility
 	horizon *HorizonSet
 
-	// ConsumedTips of remote binds we have ACKed/NACKed. flushTx's
-	// pre-flight aborts when a new local bind's pre-tx tip is in here:
-	// emitting from that fork point would commit DAG-concurrent with the
-	// remote bind we already took a position on. Cache eviction is safe
-	// because the stale-tip / new-foreign-tip checks in flushTx catch
-	// the same divergence structurally.
-	spokenBinds *clox.CloxCache[Tip, struct{}]
+	// Remote binds indexed by the tips they consumed. A shared consumed
+	// tip is only a conflict when the new transaction does not also depend
+	// on every bind we already took a position on. The bounded cache
+	// supplements the structural preflight and checkCompetingBinds walk.
+	spokenBinds   *clox.CloxCache[Tip, *spokenBindHistory]
+	spokenBindsMu sync.Mutex // serializes append-only history updates
 
 	// Anti-entropy
 	antiEntropyStop chan struct{}
@@ -405,7 +404,7 @@ func newEngine(cfg EngineConfig) *Engine {
 		pendingBootstraps:  xsync.NewMap[string, *bootstrapCollector](),
 		peerKeyFilters:     make(map[pb.NodeID]*peerKeyFilter),
 		effectCache:        newVertexPool(),
-		spokenBinds:        clox.NewCloxCache[Tip, struct{}](clox.ConfigFromCapacity(8192)),
+		spokenBinds:        clox.NewCloxCache[Tip, *spokenBindHistory](clox.ConfigFromCapacity(8192)),
 		compressValues:     cfg.CompressValues,
 	}
 
@@ -1888,13 +1887,23 @@ func (e *Engine) handleRemoteBind(bind *pb.TransactionalBindEffect, bindOffset T
 		})
 	}
 
-	// Cache every consumed tip this bind forked from so flushTx's
-	// pre-flight will abort a future local emit that forks from the
-	// same point.
+	// Retain each consuming bind, not just a boolean for its shared base.
+	// flushTx must prove those binds are ancestors before reusing that base.
 	if e.spokenBinds != nil {
+		e.spokenBindsMu.Lock()
+		defer e.spokenBindsMu.Unlock()
+
 		for _, kb := range bind.GetKeys() {
 			for _, ct := range kb.GetConsumedTips() {
-				e.spokenBinds.Put(r(ct), struct{}{})
+				tip := r(ct)
+
+				history, ok := e.spokenBinds.Get(tip, 0)
+				if !ok {
+					history = &spokenBindHistory{}
+				}
+
+				history.add(bindOffset)
+				e.spokenBinds.Put(tip, history)
 			}
 		}
 	}

effects/emit_test.go

@@ -181,7 +181,7 @@ func newTestEngine(bc Broadcaster) *Engine {
 		pendingTxTips:     xsync.NewMap[Tip, []Tip](),
 		txAbortCounts:     xsync.NewMap[string, *atomic.Int32](),
 		pendingBootstraps: xsync.NewMap[string, *bootstrapCollector](),
-		spokenBinds:       clox.NewCloxCache[Tip, struct{}](clox.ConfigFromCapacity(256)),
+		spokenBinds:       clox.NewCloxCache[Tip, *spokenBindHistory](clox.ConfigFromCapacity(256)),
 		effectCache:       newVertexPool(),
 	}
 	e.safety.Store(&safetyMap{defaultMode: UnsafeMode})
@@ -818,7 +818,7 @@ func newTxnTestEngine(bc Broadcaster) *Engine {
 		pendingTxTips:     xsync.NewMap[Tip, []Tip](),
 		txAbortCounts:     xsync.NewMap[string, *atomic.Int32](),
 		pendingBootstraps: xsync.NewMap[string, *bootstrapCollector](),
-		spokenBinds:       clox.NewCloxCache[Tip, struct{}](clox.ConfigFromCapacity(256)),
+		spokenBinds:       clox.NewCloxCache[Tip, *spokenBindHistory](clox.ConfigFromCapacity(256)),
 		effectCache:       newVertexPool(),
 	}
 	e.safety.Store(&safetyMap{defaultMode: UnsafeMode})
@@ -2002,8 +2002,8 @@ func TestFlushTx_ConsumesTips_LinearChain(t *testing.T) {
 }
 
 // TestSpokenBinds_PopulatedOnHandleRemoteBind verifies that handleRemoteBind
-// inserts every (key, consumedTip) pair from the bind's ConsumedTips into
-// spokenBinds. handleRemoteBind is the canonical entry point HandleRemote
+// records the bind offset under every consumed tip in spokenBinds.
+// handleRemoteBind is the canonical entry point HandleRemote
 // uses for every remote bind it processes — ACK and NACK paths both pass
 // through here, so populating once at this site covers both response shapes.
 func TestSpokenBinds_PopulatedOnHandleRemoteBind(t *testing.T) {
@@ -2034,19 +2034,18 @@ func TestSpokenBinds_PopulatedOnHandleRemoteBind(t *testing.T) {
 
 	e.handleRemoteBind(bind, Tip{uint64(remoteNodeID), 12}, "txn-remote")
 
-	if _, ok := e.spokenBinds.Get(ct1, 0); !ok {
-		t.Fatalf("expected spokenBinds to contain %v after handleRemoteBind, got miss", ct1)
+	if binds, ok := e.spokenBinds.Get(ct1, 0); !ok || len(binds.snapshot()) != 1 || binds.snapshot()[0] != (Tip{uint64(remoteNodeID), 12}) {
+		t.Fatalf("expected consuming bind for %v, got %v", ct1, binds)
 	}
 
-	if _, ok := e.spokenBinds.Get(ct2, 0); !ok {
-		t.Fatalf("expected spokenBinds to contain %v after handleRemoteBind, got miss", ct2)
+	if binds, ok := e.spokenBinds.Get(ct2, 0); !ok || len(binds.snapshot()) != 1 || binds.snapshot()[0] != (Tip{uint64(remoteNodeID), 12}) {
+		t.Fatalf("expected consuming bind for %v, got %v", ct2, binds)
 	}
 }
 
 // TestSpokenBinds_PreflightAbortsOnHit verifies that emitting a local bind
-// whose initialTips include a spoken (key, tip) pair is rejected by the
-// pre-flight check with ErrTxnAborted. No predicate refinement: the speech
-// act is structural and unconditional.
+// whose initialTips share a spoken tip is rejected when the consuming bind
+// cannot be proved to be an ancestor, even if it is absent from the index.
 func TestSpokenBinds_PreflightAbortsOnHit(t *testing.T) {
 	e := NewTestEngine()
 	defer func() { _ = e.Close() }()
@@ -2069,16 +2068,16 @@ func TestSpokenBinds_PreflightAbortsOnHit(t *testing.T) {
 		t.Fatalf("seed flush: %v", err)
 	}
 
-	// Mark the current tip on "k" as spoken — simulating an earlier
-	// HandleRemote where we ACKed/NACKed a remote bind that consumed
-	// this tip.
+	// Record a consuming bind which has not yet appeared in the index.
 	tips := e.index.Contains("k")
 	if tips == nil || len(tips.Tips()) == 0 {
 		t.Fatalf("expected at least one tip on 'k' after seed")
 	}
 
 	spokenTip := tips.Tips()[0]
-	e.spokenBinds.Put(spokenTip, struct{}{})
+	history := &spokenBindHistory{}
+	history.add(Tip{99, 1})
+	e.spokenBinds.Put(spokenTip, history)
 
 	// Start a transactional emit on "k" — pre-flight should abort.
 	ctx2 := e.NewContext()
@@ -2131,11 +2130,7 @@ func TestSpokenBinds_PreflightAllowsOnMiss(t *testing.T) {
 		t.Fatalf("tx emit: %v", err)
 	}
 
-	if err := ctx2.Flush(); err != nil && !errors.Is(err, ErrTxnAborted) {
-		// Standalone tests have no broadcaster, so flushTx commits
-		// without subscriber-response waits. Any non-abort outcome is
-		// fine for this test; we just need to confirm the pre-flight
-		// did not fire.
-		t.Logf("flush returned %v (may be ok for standalone tx)", err)
+	if err := ctx2.Flush(); err != nil {
+		t.Fatalf("uncontended transaction should commit: %v", err)
 	}
 }

effects/handle_remote_ephemeral_test.go

@@ -52,7 +52,7 @@ func newTestEngineForEphemeral(t *testing.T) *Engine {
 		pendingTxTips:     xsync.NewMap[Tip, []Tip](),
 		txAbortCounts:     xsync.NewMap[string, *atomic.Int32](),
 		pendingBootstraps: xsync.NewMap[string, *bootstrapCollector](),
-		spokenBinds:       clox.NewCloxCache[Tip, struct{}](clox.ConfigFromCapacity(256)),
+		spokenBinds:       clox.NewCloxCache[Tip, *spokenBindHistory](clox.ConfigFromCapacity(256)),
 	}
 	e.safety.Store(&safetyMap{defaultMode: UnsafeMode})
 

effects/horizon_ancestor_invariant_test.go

@@ -278,7 +278,7 @@ func newSnapshotEngineWithHorizon(log *snapshotLog) *Engine {
 		pendingTxTips:     xsync.NewMap[Tip, []Tip](),
 		txAbortCounts:     xsync.NewMap[string, *atomic.Int32](),
 		pendingBootstraps: xsync.NewMap[string, *bootstrapCollector](),
-		spokenBinds:       clox.NewCloxCache[Tip, struct{}](clox.ConfigFromCapacity(256)),
+		spokenBinds:       clox.NewCloxCache[Tip, *spokenBindHistory](clox.ConfigFromCapacity(256)),
 	}
 	e.safety.Store(&safetyMap{defaultMode: UnsafeMode})
 	e.horizon = newHorizonSet(e)