diff --git a/effects/context.go b/effects/context.go index b81aaa6..d78bf4d 100644 --- a/effects/context.go +++ b/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{} diff --git a/effects/fork_choice_test.go b/effects/fork_choice_test.go index 35fc92a..2c7470c 100644 --- a/effects/fork_choice_test.go +++ b/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) + } + }) + } + } +} diff --git a/effects/snapshot.go b/effects/snapshot.go index c05f146..1b23a67 100644 --- a/effects/snapshot.go +++ b/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 { diff --git a/effects/txn.go b/effects/txn.go index 267bc1b..e77aa58 100644 --- a/effects/txn.go +++ b/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 +}