add ambiguous result and remove optimization

effects/context.go

@@ -1685,11 +1685,11 @@ func (c *Context) flushTx() error {
 
 			if ptxn.state.Load() != txnStateAborted {
 				if responseCount < len(subscribers) {
-					slog.Warn("flushTx: not all subscribers responded, aborting",
+					slog.Warn("flushTx: not all subscribers responded, outcome unknown",
 						"bind_offset", bindOffset,
 						"responses", responseCount,
 						"subscribers", len(subscribers))
-					abortPendingTxn(ptxn)
+					markPendingTxnAmbiguous(ptxn)
 				} else if commitPendingTxn(ptxn) {
 					slog.Debug("flushTx: all subscribers responded, committed",
 						"bind_offset", bindOffset,
@@ -1701,7 +1701,7 @@ func (c *Context) flushTx() error {
 			// chance to set txnStateAborted via isRealConflict. The bind
 			// must not become visible before this decision is made.
 			if c.engine.horizon != nil {
-				if ptxn.state.Load() == txnStateAborted {
+				if ptxn.state.Load() != txnStateCommitted {
 					c.engine.horizon.Abort(c.txnID)
 				} else {
 					c.engine.horizon.MakeVisible(c.txnID)
@@ -1777,16 +1777,24 @@ func (c *Context) flushTx() error {
 
 	c.engine.pendingTxns.Delete(bindOffset)
 
-	if state == txnStateAborted {
-		slog.Debug("Flush: transaction aborted", "bind_offset", bindOffset)
-		// Clean up pending tx tips — but NOT when horizon is active
-		// (effects stay invisible, timer cleans up)
+	if state != txnStateCommitted {
+		// Release local pending state. Horizon.Abort already did this when
+		// enabled; neither path retracts the published bind from peers.
 		if c.engine.horizon == nil {
 			for _, pk := range ptxn.keys {
 				c.engine.pendingTxTips.Delete(pk.newTip)
 			}
 		}
 
+		if state == txnStateAmbiguous {
+			slog.Debug("Flush: transaction outcome unknown", "bind_offset", bindOffset)
+			c.reset()
+
+			return ErrTxnAmbiguous
+		}
+
+		slog.Debug("Flush: transaction aborted", "bind_offset", bindOffset)
+
 		// Check serialization escalation (only if no leader already active)
 		for _, pk := range ptxn.keys {
 			if c.engine.CheckSerializationLeader(pk.key) == nil && c.engine.incrementAbortCount(pk.key) {
@@ -1847,8 +1855,8 @@ func (c *Context) flushTx() error {
 // The membership filter is what lets a departed peer's stale id fall out
 // lazily instead of needing an eager scrub on peer removal: PeerIDs() shrinks
 // only on genuine topology removal, so a member that is merely unreachable
-// stays in the required set (flushTx then aborts when it can't respond,
-// preserving AtMostOneCommit), while a truly departed peer is dropped so the
+// stays in the required set (flushTx then reports an ambiguous outcome when it
+// can't respond), while a truly departed peer is dropped so the
 // key can still make progress.
 func (c *Context) collectSubscribers() []pb.NodeID {
 	var members map[pb.NodeID]struct{}

effects/fork_choice_test.go

@@ -95,3 +95,70 @@ func TestForkChoiceLess_EliminatesHLCBias(t *testing.T) {
 		t.Fatalf("expected mixed winners, got %d/%d for node1", wins, trials)
 	}
 }
+
+func TestLosersOnKeyAt_SharedRootFork(t *testing.T) {
+	for _, shape := range []string{"fork", "merge", "snapshot", "snapshot-descendant"} {
+		for _, reverse := range []bool{false, true} {
+			name := shape
+			if reverse {
+				name += "/reversed"
+			}
+
+			t.Run(name, func(t *testing.T) {
+				log := newSnapshotLog()
+				e := newSnapshotEngine(log)
+				addBind := func(txn string, hash byte, consumed ...Tip) Tip {
+					data := log.putEffect(&pb.Effect{
+						Key: []byte("k"), NodeId: 1, Hlc: sTs(100), TxnId: txn,
+						Deps: toPbRefs(consumed), Kind: &pb.Effect_Noop{Noop: &pb.NoopEffect{}},
+					})
+
+					return log.putEffect(&pb.Effect{
+						Key: []byte("k"), NodeId: 1, Hlc: sTs(100), TxnId: txn, ForkChoiceHash: []byte{hash},
+						Deps: toPbRefs([]Tip{data}),
+						Kind: &pb.Effect_TxnBind{TxnBind: &pb.TransactionalBindEffect{
+							Keys: []*pb.TransactionalBindEffect_KeyBind{{
+								Key: []byte("k"), NewTip: toPbRef(data), ConsumedTips: toPbRefs(consumed),
+							}},
+						}},
+					})
+				}
+
+				// A common root does not make the siblings a chain. Adding a
+				// merge or snapshot must not change which sibling loses.
+				root := addBind("root", 0x40)
+				winner := addBind("winner", 0x10, root)
+				loser := addBind("loser", 0x20, root)
+
+				tips := []Tip{winner, loser}
+				if reverse {
+					tips[0], tips[1] = tips[1], tips[0]
+				}
+
+				switch shape {
+				case "merge":
+					tips = []Tip{addBind("descendant", 0x30, tips...)}
+				case "snapshot", "snapshot-descendant":
+					snapshot := log.putEffect(&pb.Effect{
+						Key: []byte("k"), NodeId: 1, Hlc: sTs(100), Deps: toPbRefs(tips),
+						Kind: &pb.Effect_Snapshot{Snapshot: &pb.SnapshotEffect{}},
+					})
+
+					tips = []Tip{snapshot}
+					if shape == "snapshot-descendant" {
+						tips = []Tip{addBind("descendant", 0x30, snapshot)}
+					}
+				}
+
+				losers, _ := e.losersOnKeyAt("k", tips, nil, nil, nil, false)
+				if _, found := losers["loser"]; !found {
+					t.Fatalf("concurrent sibling escaped fork choice: %v", losers)
+				}
+
+				if len(losers) != 1 {
+					t.Fatalf("unexpected other losers: %v", losers)
+				}
+			})
+		}
+	}
+}

effects/snapshot.go

@@ -1903,7 +1903,6 @@ func (e *Engine) losersOnKeyAt(key string, tips, extraTips []Tip, snapshotVerdic
 		bindOffset Tip // bind anchor offset (the TransactionalBindEffect's own offset)
 		newTip     Tip
 		hash       []byte
-		consumed   []Tip // ConsumedTips on this key
 	}
 
 	var binds []seenBind
@@ -1948,15 +1947,11 @@ func (e *Engine) losersOnKeyAt(key string, tips, extraTips []Tip, snapshotVerdic
 
 		if bind := eff.GetTxnBind(); bind != nil {
 			if isInvisible == nil || !isInvisible(eff.GetTxnId()) {
-				var (
-					newTip   Tip
-					consumed []Tip
-				)
+				var newTip Tip
 
 				for _, kb := range bind.GetKeys() {
 					if string(kb.GetKey()) == key {
 						newTip = r(kb.GetNewTip())
-						consumed = fromPbRefs(kb.GetConsumedTips())
 
 						break
 					}
@@ -1969,7 +1964,6 @@ func (e *Engine) losersOnKeyAt(key string, tips, extraTips []Tip, snapshotVerdic
 						bindOffset: t,
 						newTip:     newTip,
 						hash:       eff.GetForkChoiceHash(),
-						consumed:   consumed,
 					})
 				}
 			}
@@ -2017,54 +2011,10 @@ func (e *Engine) losersOnKeyAt(key string, tips, extraTips []Tip, snapshotVerdic
 		candidates = append(candidates, b)
 	}
 
-	// Chain detection: if every candidate bind's ConsumedTips on this
-	// key contains another candidate bind's anchor offset, the binds
-	// form a total order via consumed-bind edges (each bind explicitly
-	// built on its predecessor's committed bind). No two can be
-	// DAG-concurrent on this key, so no losers — skip the O(N²) pairwise
-	// and ancestor work entirely. Common in sequential single-writer
-	// workloads (XADD, LPUSH, INCR), where this is the difference
-	// between O(N²) and O(N).
-	if len(candidates) > 1 {
-		bindOffsetToIdx := make(map[Tip]int, len(candidates))
-		for i := range candidates {
-			bindOffsetToIdx[candidates[i].bindOffset] = i
-		}
-
-		rootBinds := 0
-		chained := true
-
-		for i := range candidates {
-			consumesSibling := false
-
-			for _, ct := range candidates[i].consumed {
-				if _, ok := bindOffsetToIdx[ct]; ok {
-					consumesSibling = true
-
-					break
-				}
-			}
-
-			if !consumesSibling {
-				rootBinds++
-				if rootBinds > 1 {
-					chained = false
-
-					break
-				}
-			}
-		}
-
-		if chained {
-			return losers, walked
-		}
-	}
-
 	// Ancestor closures computed lazily on first reach query. Cached so
 	// each bind's parent DAG is walked at most once across all pairs.
 	// Fetches via e.getEffect (CloxCache-backed) instead of holding a
-	// separate nodes map, so the discovery walk doesn't pay for an
-	// extra map slot per visited effect when chain-detect short-circuits.
+	// separate nodes map during discovery.
 	ancestorsOf := make(map[Tip]map[Tip]struct{})
 	buildAncestors := func(start Tip) map[Tip]struct{} {
 		if cached, ok := ancestorsOf[start]; ok {

effects/txn.go

@@ -36,6 +36,11 @@ import (
 // encounters a real conflict that cannot be resolved.
 var ErrTxnAborted = errors.New("transaction aborted")
 
+// ErrTxnAmbiguous is returned when a published bind could not be confirmed by
+// every subscriber. Its effects may become visible; callers must not treat
+// this as a definite abort or automatically retry the transaction.
+var ErrTxnAmbiguous = errors.New("transaction outcome unknown; commit may have occurred")
+
 // ErrRegionPartitioned is returned by Flush in SafeMode when not all
 // same-region peers are reachable.
 var ErrRegionPartitioned = errors.New("region partitioned: not all same-region peers are reachable")
@@ -63,6 +68,7 @@ const (
 	txnStatePending   uint32 = 0
 	txnStateCommitted uint32 = 1
 	txnStateAborted   uint32 = 2
+	txnStateAmbiguous uint32 = 3
 )
 
 // pendingTxn tracks a transaction awaiting NACK resolution.
@@ -509,3 +515,15 @@ func abortPendingTxn(ptxn *pendingTxn) bool {
 
 	return false
 }
+
+// markPendingTxnAmbiguous ends the local wait without claiming a cluster-wide
+// commit or abort. A known conflict that already aborted the txn takes priority.
+func markPendingTxnAmbiguous(ptxn *pendingTxn) bool {
+	if ptxn.state.CompareAndSwap(txnStatePending, txnStateAmbiguous) {
+		close(ptxn.done)
+
+		return true
+	}
+
+	return false
+}