diff --git a/fraud/actor_test.go b/fraud/actor_test.go index 082ebf76e..7f52997c7 100644 --- a/fraud/actor_test.go +++ b/fraud/actor_test.go @@ -19,6 +19,19 @@ import ( const testTimeout = time.Second +// The fraud watcher is intentionally a one-way alarm: it triggers +// unroll on every observed ancestor spend and does not subscribe to +// the chainsource reorg-aware lifecycle. The reorg-safety roadmap's +// third invariant — "unroll reconciliation resumes safely" — is owned +// by the unroll package and exercised there +// (see unroll/reorg_safety_test.go::TestOfflineReorgRestartReconciles +// BeforeSideEffects and the related restart-reconcile tests). The +// fraud-side tests in this file pin only the parts of the invariant +// the fraud watcher itself owns: spend watch is not reorg-aware, +// triggering the unroll does not unregister the watch, repeated +// observations are forwarded, and benign edge cases (unknown +// outpoint, shutdown) do not falsely raise a fraud alarm. + // spendRef aliases the chainsource spend notification target. type spendRef = actor.TellOnlyRef[chainsource.SpendEvent] @@ -378,3 +391,285 @@ func TestWatcherSpendFanoutBestEffort(t *testing.T) { require.Error(t, err) require.Equal(t, 2, unrollRef.requestCount()) } + +// TestFraudWatcherSpendWatchIsNotReorgAware pins the fraud-watcher +// semantics flagged by the reorg-safety roadmap: the chainsource +// spend watch is intentionally a one-way alarm, NOT reorg-aware. A +// fraud trigger that fires on a transient (later-reorged) ancestor +// spend is acceptable because the unroll the trigger spawns is itself +// idempotent and reorg-safe, and the unroll's restart reconciliation +// path will land the recovery on the canonical chain regardless of +// the trigger's chain history. Wiring NotifyReorged / NotifyDone here +// would only let a transient ancestor-spend reorg unwind the fraud +// alarm, which we explicitly do not want. +// +// This test exists so that "fraud watcher must not be reversible" +// stays a documented, enforceable invariant rather than an +// undocumented assumption. +func TestFraudWatcherSpendWatchIsNotReorgAware(t *testing.T) { + treePath, _ := testLeafTree(t, 60) + target := testInput(61) + desc := testDescriptor(target, treePath) + + chainRef := &fakeChainSourceRef{} + unrollRef := &fakeUnrollRef{} + watcher := NewWatcherActor(WatcherConfig{ + ChainSource: chainRef, + UnrollRef: unrollRef, + Log: fn.None[btclog.Logger](), + }) + t.Cleanup(watcher.Stop) + + _, err := watcher.Ref().Ask(t.Context(), &TrackVTXOsRequest{ + VTXOs: []*vtxo.Descriptor{desc}, + }).Await(t.Context()).Unpack() + require.NoError(t, err) + + chainRef.mu.Lock() + defer chainRef.mu.Unlock() + require.NotEmpty(t, chainRef.spendReqs) + + for i, req := range chainRef.spendReqs { + require.True( + t, req.NotifyActor.IsSome(), + "req[%d] must always wire NotifyActor", i, + ) + require.True( + t, req.NotifyReorged.IsNone(), + "req[%d] must NOT wire NotifyReorged — fraud watch "+ + "is intentionally one-way; unroll's own "+ + "reorg-aware lifecycle handles "+ + "canonical-chain reconciliation downstream", i, + ) + require.True( + t, req.NotifyDone.IsNone(), + "req[%d] must NOT wire NotifyDone for the same reason", + i, + ) + } +} + +// TestFraudWatcherSpendTriggerDoesNotUnregisterWatch verifies that +// observing a watched ancestor spend (and firing the fraud unroll +// trigger) does NOT tear down the chainsource spend watch for that +// outpoint. The watch stays armed so a subsequent re-confirmation / +// re-spend of the same ancestor would re-trigger; only an explicit +// UntrackRequest on the dependent target tears the watch down (see +// TestWatcherTriggersUnrollOnAncestorSpend's untrack assertion). +// +// This is the structural complement to +// TestFraudWatcherSpendWatchIsNotReorgAware: the watch is one-way +// at the wiring level AND retained at the lifecycle level. Together +// they prove that a transient ancestor-spend reorg cannot terminally +// fail the fraud defenses — the watch is still listening and will +// re-fire the trigger on the next observation. +func TestFraudWatcherSpendTriggerDoesNotUnregisterWatch(t *testing.T) { + treePath, source := testLeafTree(t, 70) + target := testInput(71) + desc := testDescriptor(target, treePath) + + chainRef := &fakeChainSourceRef{} + unrollRef := &fakeUnrollRef{} + watcher := NewWatcherActor(WatcherConfig{ + ChainSource: chainRef, + UnrollRef: unrollRef, + Log: fn.None[btclog.Logger](), + }) + t.Cleanup(watcher.Stop) + + _, err := watcher.Ref().Ask(t.Context(), &TrackVTXOsRequest{ + VTXOs: []*vtxo.Descriptor{desc}, + }).Await(t.Context()).Unpack() + require.NoError(t, err) + + chainRef.emitSpend(t, source) + + require.Eventually(t, func() bool { + return unrollRef.requestCount() == 1 + }, testTimeout, 10*time.Millisecond) + + // Watch must still be registered after the trigger fires. + chainRef.mu.Lock() + require.Empty( + t, chainRef.unregisters, "firing the unroll trigger must "+ + "not unregister the watch; the watch has to stay "+ + "armed so a subsequent observation of the same "+ + "ancestor still re-triggers", + ) + chainRef.mu.Unlock() +} + +// TestFraudWatcherSpendTriggerIsIdempotentOnRepeatedObservations +// verifies that observing the same ancestor spend twice fires the +// unroll trigger twice. The fraud watcher does not need to dedupe +// itself; downstream idempotency in the unroll registry (tested in +// `unroll/`) absorbs the duplicate without spawning a second child +// actor. This test only pins the FRAUD-side behavior — the fake +// unroll ref here records both calls without enforcing the registry's +// dedup contract, which is the right scope: that contract is owned +// by the unroll package. +// +// The scenario this anticipates is a backend that re-delivers a +// `SpendEvent` for the same outpoint after a reorg-and-reconfirm of +// the spending block. The fraud watcher's `RegisterSpendRequest` +// deliberately does not subscribe to the chainsource reorg-aware +// lifecycle (see TestFraudWatcherSpendWatchIsNotReorgAware), so this +// test simulates what would happen IF a duplicate `SpendEvent` +// reached the watcher mailbox by another route (e.g. a backend that +// natively redelivers positive events on canonical-chain change). +func TestFraudWatcherSpendTriggerIsIdempotentOnRepeatedObservations( + t *testing.T) { + + treePath, source := testLeafTree(t, 80) + target := testInput(81) + desc := testDescriptor(target, treePath) + + chainRef := &fakeChainSourceRef{} + unrollRef := &fakeUnrollRef{} + watcher := NewWatcherActor(WatcherConfig{ + ChainSource: chainRef, + UnrollRef: unrollRef, + Log: fn.None[btclog.Logger](), + }) + t.Cleanup(watcher.Stop) + + _, err := watcher.Ref().Ask(t.Context(), &TrackVTXOsRequest{ + VTXOs: []*vtxo.Descriptor{desc}, + }).Await(t.Context()).Unpack() + require.NoError(t, err) + + // First observation. + chainRef.emitSpend(t, source) + require.Eventually(t, func() bool { + return unrollRef.requestCount() == 1 + }, testTimeout, 10*time.Millisecond) + + // Second observation (simulates the reorg-aware path where the + // SpendActor re-fires after a reorg-then-reconfirm of the + // spending block — even though fraud watch itself is not wired + // for reorgs, a real reorg-aware backend could re-deliver the + // underlying SpendEvent on the new canonical chain). + chainRef.emitSpend(t, source) + require.Eventually(t, func() bool { + return unrollRef.requestCount() == 2 + }, testTimeout, 10*time.Millisecond) + + // Both requests must target the same VTXO with the FraudSpend + // trigger; the downstream unroll registry's idempotency makes + // the second a no-op. + chainRef.mu.Lock() + require.Empty( + t, chainRef.unregisters, + "repeated observations must keep the watch armed", + ) + chainRef.mu.Unlock() + + unrollRef.mu.Lock() + for i, req := range unrollRef.requests { + require.Equal( + t, target, req.Outpoint, "request[%d] must target "+ + "the same outpoint", i, + ) + require.Equal( + t, unroll.TriggerFraudSpend, req.Trigger, "request[%"+ + "d] must use the FraudSpend trigger", i, + ) + } + unrollRef.mu.Unlock() +} + +// TestFraudWatcherOnStopUnregistersWithoutTriggering verifies the +// shutdown path: stopping the watcher tears down every active +// chainsource spend watch but does NOT fire any unroll trigger. +// A future refactor that accidentally re-uses the spend-event path +// for shutdown cleanup would falsely trigger fraud unrolls on every +// daemon shutdown. +func TestFraudWatcherOnStopUnregistersWithoutTriggering(t *testing.T) { + treePath, _ := testLeafTree(t, 90) + target := testInput(91) + desc := testDescriptor(target, treePath) + + chainRef := &fakeChainSourceRef{} + unrollRef := &fakeUnrollRef{} + watcher := NewWatcherActor(WatcherConfig{ + ChainSource: chainRef, + UnrollRef: unrollRef, + Log: fn.None[btclog.Logger](), + }) + + _, err := watcher.Ref().Ask(t.Context(), &TrackVTXOsRequest{ + VTXOs: []*vtxo.Descriptor{desc}, + }).Await(t.Context()).Unpack() + require.NoError(t, err) + + require.Greater( + t, chainRef.spendWatchCount(), 0, + "setup: watcher must have registered at least one watch", + ) + + // Shutdown. + watcher.Stop() + + // On shutdown the watcher should release its watches via the + // chainsource Unregister path, but must not have triggered any + // unroll calls. + require.Eventually(t, func() bool { + chainRef.mu.Lock() + defer chainRef.mu.Unlock() + + return len(chainRef.unregisters) > 0 + }, testTimeout, 10*time.Millisecond, + "OnStop must unregister the active spend watches") + + require.Zero( + t, unrollRef.requestCount(), + "OnStop must NOT fire any unroll trigger — shutdown is not "+ + "a fraud event", + ) +} + +// TestFraudWatcherSpendObservedForUnknownOutpointIsBenign verifies +// that a SpendObservedMsg for an outpoint with no tracked targets +// (e.g. a late chainsource delivery for an outpoint that has since +// been untracked, or a backend that natively re-delivers a positive +// event after a reorg-and-reconfirm on a target that the watcher no +// longer cares about) returns AckResp{} without firing any unroll +// trigger. This is the load-bearing safety case for the "transient +// ancestor-spend reorg does not terminally fail anything" invariant: +// a re-delivered spend on a now-untracked outpoint must be silently +// dropped, not erroneously promoted into a fraud alarm. +func TestFraudWatcherSpendObservedForUnknownOutpointIsBenign(t *testing.T) { + chainRef := &fakeChainSourceRef{} + unrollRef := &fakeUnrollRef{} + watcher := NewWatcherActor(WatcherConfig{ + ChainSource: chainRef, + UnrollRef: unrollRef, + Log: fn.None[btclog.Logger](), + }) + t.Cleanup(watcher.Stop) + + unknownOutpoint := testInput(99) + + resp, err := watcher.Ref().Ask( + t.Context(), &SpendObservedMsg{ + Outpoint: unknownOutpoint, + SpendingTxid: testInput(100).Hash, + Height: 42, + }, + ).Await(t.Context()).Unpack() + require.NoError( + t, err, "SpendObservedMsg for an unknown outpoint must ack "+ + "without error — the late-delivery scenario is benign", + ) + _, ok := resp.(*AckResp) + require.True( + t, ok, + "SpendObservedMsg for an unknown outpoint must return AckResp", + ) + + require.Zero( + t, unrollRef.requestCount(), + "SpendObservedMsg for an unknown outpoint must NOT trigger "+ + "any unroll call", + ) +} diff --git a/round/actor.go b/round/actor.go index b7a390ec5..872421580 100644 --- a/round/actor.go +++ b/round/actor.go @@ -229,6 +229,29 @@ type RoundClientActor struct { // keys for routing confirmation events. commitmentTxIndex map[chainhash.Hash]RoundKeyStr + // pendingCommitmentConfs caches the most recent ConfirmationEvent + // for each tracked commitment tx, so handleCommitmentFinalized can + // build the BoardingConfirmed FSM event using the canonical-chain + // confirmation height observed before finality. The entry is + // installed on every ConfirmationEvent (first conf or any + // re-confirmation after a reorg) and consumed on the matching + // CommitmentFinalizedEvent. A reorg without a follow-up + // re-confirmation leaves the stale entry in place, but the + // chainsource finality synthesizer requires a non-zero + // confirmHeight to fire a Done event, so finality cannot land on + // the stale entry; the next re-confirmation overwrites it before + // any Done could be synthesized. + // + // The cache exists because the round FSM intentionally delays its + // terminal transition (InputSigSent -> ConfirmedState) until the + // commitment-tx confirmation is past the chainsource backend's + // reorg-safety depth. The first ConfirmationEvent on its own is + // not sufficient to commit user-visible state (VTXOs marked live, + // ledger emissions, indexer notifications) because a reorg of the + // confirmation block would otherwise leave that state inconsistent + // with the canonical chain. + pendingCommitmentConfs map[chainhash.Hash]*ConfirmationEvent + // pendingQuotes buffers JoinRoundQuoteReceived envelopes that // arrive before the matching RoundJoined re-keys the FSM. The // mailbox contract (docs/RPC_MAILBOX_CONTRACT.md:90-98) allows @@ -423,8 +446,11 @@ func NewRoundClientActor(cfg *RoundClientConfig) fn.Result[*RoundClientActor] { log: actorLog, rounds: make(map[RoundKeyStr]*RoundFSM), commitmentTxIndex: make(map[chainhash.Hash]RoundKeyStr), - pendingQuotes: make(map[RoundID]*JoinRoundQuoteReceived), - env: env, + pendingCommitmentConfs: make( + map[chainhash.Hash]*ConfirmationEvent, + ), + pendingQuotes: make(map[RoundID]*JoinRoundQuoteReceived), + env: env, } // The base env is used as a template for per-round FSM environments. @@ -1007,6 +1033,25 @@ func (a *RoundClientActor) registerCommitmentConfirmation(ctx context.Context, }, ) + // Reorg-aware lifecycle refs. The actor currently logs these + // rather than reversing state: see the doc on CommitmentReorgedEvent + // for why FSM-level rollback is a follow-up. Wiring the refs now + // means the chainsource conf sub-actor stays alive past first + // confirmation, height-based finality synthesis fires, and a + // future FSM-rollback patch only has to consume the events. + reorgedRef := chainsource.MapConfReorgedEvent( + a.cfg.SelfRef, + func(ev chainsource.ConfReorgedEvent) actormsg.RoundReceivable { + return &CommitmentReorgedEvent{Txid: ev.Txid} + }, + ) + finalizedRef := chainsource.MapConfDoneEvent( + a.cfg.SelfRef, + func(ev chainsource.ConfDoneEvent) actormsg.RoundReceivable { + return &CommitmentFinalizedEvent{Txid: ev.Txid} + }, + ) + // Extract the pkScript LND needs for confirmation tracking. Watch the // validated batch output (the output that receives this client's // funds) rather than assuming output 0; confirmationWatchScript falls @@ -1042,12 +1087,14 @@ func (a *RoundClientActor) registerCommitmentConfirmation(ctx context.Context, } confReq := &chainsource.RegisterConfRequest{ - CallerID: callerID, - Txid: &txid, - PkScript: pkScript, - TargetConfs: a.cfg.OperatorTerms.MinConfirmations, - HeightHint: heightHint, - NotifyActor: fn.Some(mappedRef), + CallerID: callerID, + Txid: &txid, + PkScript: pkScript, + TargetConfs: a.cfg.OperatorTerms.MinConfirmations, + HeightHint: heightHint, + NotifyActor: fn.Some(mappedRef), + NotifyReorged: fn.Some(reorgedRef), + NotifyDone: fn.Some(finalizedRef), } if err := a.cfg.ChainSource.Tell( @@ -1300,6 +1347,12 @@ func (a *RoundClientActor) Receive(ctx context.Context, case *ConfirmationEvent: return a.handleConfirmation(ctx, m) + case *CommitmentReorgedEvent: + return a.handleCommitmentReorged(ctx, m) + + case *CommitmentFinalizedEvent: + return a.handleCommitmentFinalized(ctx, m) + case *TimeoutMsg: return a.handleTimeout(ctx, m) @@ -2140,12 +2193,27 @@ func (a *RoundClientActor) reapFailedRounds(ctx context.Context) { // from ChainSource. Boarding address confirmations are now handled via // WalletBoardingConfirmed events from the wallet actor. // +// The commitment-tx confirmation is treated as PROVISIONAL: the FSM +// is NOT transitioned to terminal ConfirmedState on first conf. The +// event is cached on pendingCommitmentConfs so that the matching +// CommitmentFinalizedEvent (synthesized by the chainsource backend at +// the reorg-safety horizon, default six blocks past the latest +// confirmation) can replay it as BoardingConfirmed. A re-confirmation +// after a reorg overwrites the cache entry with the new canonical- +// chain height, and the finality synthesizer's depth counter resets +// on the reorg, so finality is only ever reported for the latest +// re-confirmation. This preserves the property the round FSM relies +// on for safety: user-visible side effects (VTXOs marked live in the +// local store, ledger entries, indexer notifications) only fire once +// the commitment is past the reorg-safety horizon. +// // Concurrency: The actor framework serializes all messages through Receive(), // so no synchronization is needed for rounds map access. func (a *RoundClientActor) handleConfirmation(ctx context.Context, event *ConfirmationEvent) fn.Result[actormsg.RoundActorResp] { - a.log.InfoS(ctx, "Received commitment transaction confirmation", + a.log.InfoS(ctx, "Received provisional commitment-tx confirmation; "+ + "deferring FSM terminal transition to finality", slog.String("txid", event.Txid.String()), slog.Int("block_height", int(event.BlockHeight)), slog.Int("confirmations", int(event.Confirmations)), @@ -2165,7 +2233,90 @@ func (a *RoundClientActor) handleConfirmation(ctx context.Context, return fn.Ok[actormsg.RoundActorResp](nil) } - // Route to the specific round's FSM. + // Sanity-check the routing target exists, but do NOT advance the + // FSM yet. handleCommitmentFinalized is the trigger for the + // terminal transition. + if _, exists := a.rounds[keyStr]; !exists { + return fn.Err[actormsg.RoundActorResp]( + fmt.Errorf("round FSM not found for key %s", keyStr), + ) + } + + // Cache the conf so the matching finality event can replay it. + // Every ConfirmationEvent overwrites — the cache always reflects + // the LATEST positive confirmation, which is what the finality + // synthesizer counts depth from. + cached := *event + a.pendingCommitmentConfs[event.Txid] = &cached + + return fn.Ok[actormsg.RoundActorResp](nil) +} + +// handleCommitmentReorged processes a chainsource ConfReorgedEvent on +// a commitment transaction. The provisional/finalized split means the +// FSM is still in its pre-confirmation state when a reorg lands — +// user-visible side effects have not yet committed — so there is +// nothing to roll back. The cached ConfirmationEvent stays in place; +// either a follow-up re-confirmation overwrites it before the +// chainsource finality synthesizer can fire (the synthesizer's depth +// counter resets on the reorg), or the chain genuinely abandons the +// confirmation and no Done event is ever synthesized. +func (a *RoundClientActor) handleCommitmentReorged(ctx context.Context, + event *CommitmentReorgedEvent) fn.Result[actormsg.RoundActorResp] { + + keyStr, tracked := a.commitmentTxIndex[event.Txid] + if !tracked { + // Round is no longer tracked: either it finalized cleanly + // and the cleanup path already ran, or it was never one of + // ours. Either way there's nothing to undo here. + a.log.DebugS(ctx, "Commitment-tx reorged on untracked txid", + slog.String("txid", event.Txid.String()), + ) + + return fn.Ok[actormsg.RoundActorResp](nil) + } + + a.log.InfoS(ctx, "Commitment-tx provisional confirmation reorged "+ + "out; FSM remains pre-confirmation, awaiting "+ + "re-confirmation or finality on the canonical chain", + slog.String("txid", event.Txid.String()), + slog.String("round_key", string(keyStr)), + ) + + return fn.Ok[actormsg.RoundActorResp](nil) +} + +// handleCommitmentFinalized processes a chainsource ConfDoneEvent on a +// commitment transaction. This is the trigger for the FSM's +// InputSigSent -> ConfirmedState transition: the cached +// ConfirmationEvent's height/hash/numConfs are replayed as the +// BoardingConfirmed FSM event so the terminal-state side effects +// (ledger emission, indexer publish, onRoundComplete cleanup) fire +// only after the chainsource backend has reported the confirmation is +// past the reorg-safety horizon. +// +// If the cache is empty for this txid (no prior ConfirmationEvent +// observed, e.g. a late or duplicate Done event after the round was +// already cleaned up), the handler logs and acks without touching +// the FSM. +func (a *RoundClientActor) handleCommitmentFinalized(ctx context.Context, + event *CommitmentFinalizedEvent) fn.Result[actormsg.RoundActorResp] { + + a.log.InfoS(ctx, "Commitment-tx confirmation finalized; promoting "+ + "round FSM to ConfirmedState", + slog.String("txid", event.Txid.String()), + ) + + keyStr, tracked := a.commitmentTxIndex[event.Txid] + if !tracked { + a.log.DebugS(ctx, "Commitment-tx finalized on untracked txid; "+ + "round already cleaned up", + slog.String("txid", event.Txid.String()), + ) + + return fn.Ok[actormsg.RoundActorResp](nil) + } + roundFSM, exists := a.rounds[keyStr] if !exists { return fn.Err[actormsg.RoundActorResp]( @@ -2173,23 +2324,37 @@ func (a *RoundClientActor) handleConfirmation(ctx context.Context, ) } - a.log.InfoS(ctx, "Routing confirmation to round FSM", - slog.String("key", string(keyStr)), - slog.String("round_id", roundFSM.RoundID.String()), - ) + cached, ok := a.pendingCommitmentConfs[event.Txid] + if !ok { + // Defensive: chainsource should not synthesize a Done event + // without a prior positive ConfirmationEvent (the + // finality-depth synthesizer is gated on a non-zero + // confirmHeight). If we see one anyway, ack without + // transitioning — the FSM cannot reach ConfirmedState + // without the confirmation's height/hash. + a.log.WarnS(ctx, "Commitment-tx finalized without a cached "+ + "prior ConfirmationEvent; skipping FSM transition", + nil, + slog.String("txid", event.Txid.String()), + slog.String("round_key", string(keyStr)), + ) + + return fn.Ok[actormsg.RoundActorResp](nil) + } + delete(a.pendingCommitmentConfs, event.Txid) confirmEvt := &BoardingConfirmed{ - TxID: event.Txid, - BlockHeight: event.BlockHeight, - BlockHash: event.BlockHash, - Confirmations: int32(event.Confirmations), + TxID: cached.Txid, + BlockHeight: cached.BlockHeight, + BlockHash: cached.BlockHash, + Confirmations: int32(cached.Confirmations), } err := a.askEventAndProcessOutbox(ctx, roundFSM, confirmEvt) if err != nil { return fn.Err[actormsg.RoundActorResp]( - fmt.Errorf("FSM error processing commitment "+ - "confirmation: %w", err), + fmt.Errorf("FSM error promoting round on finality: %w", + err), ) } @@ -2687,6 +2852,22 @@ func (a *RoundClientActor) processConfirmationRequest( }, ) + // Reorg-aware lifecycle refs — see registerCommitmentConfirmation + // for the rationale. Detection-only today; FSM rollback is a + // follow-up. + reorgedRef := chainsource.MapConfReorgedEvent( + a.cfg.SelfRef, + func(ev chainsource.ConfReorgedEvent) actormsg.RoundReceivable { + return &CommitmentReorgedEvent{Txid: ev.Txid} + }, + ) + finalizedRef := chainsource.MapConfDoneEvent( + a.cfg.SelfRef, + func(ev chainsource.ConfDoneEvent) actormsg.RoundReceivable { + return &CommitmentFinalizedEvent{Txid: ev.Txid} + }, + ) + // Query ChainSource for current block height to use as // HeightHint. LND requires HeightHint > 0 for confirmation // scanning. @@ -2712,12 +2893,14 @@ func (a *RoundClientActor) processConfirmationRequest( // Build the complete RegisterConfRequest with the mapper as // the NotifyActor target. confReq := &chainsource.RegisterConfRequest{ - CallerID: callerID, - Txid: m.Txid, - PkScript: m.PkScript, - TargetConfs: m.TargetConfs, - HeightHint: heightHint, - NotifyActor: fn.Some(mappedRef), + CallerID: callerID, + Txid: m.Txid, + PkScript: m.PkScript, + TargetConfs: m.TargetConfs, + HeightHint: heightHint, + NotifyActor: fn.Some(mappedRef), + NotifyReorged: fn.Some(reorgedRef), + NotifyDone: fn.Some(finalizedRef), } a.log.InfoS(ctx, "Sending RegisterConfRequest to ChainSource", diff --git a/round/actor_messages.go b/round/actor_messages.go index c9d434f3f..e1884725a 100644 --- a/round/actor_messages.go +++ b/round/actor_messages.go @@ -268,6 +268,55 @@ func (m *ConfirmationEvent) MessageType() string { // RoundReceivable implements actormsg.RoundReceivable marker interface. func (m *ConfirmationEvent) RoundReceivable() {} +// CommitmentReorgedEvent wraps a chainsource ConfReorgedEvent that +// reports a previously delivered ConfirmationEvent for a commitment +// transaction was rolled back by a reorg of the canonical chain. +// +// Reorg semantics for the round FSM are not yet implemented: the +// commitment-tx confirmation drives the FSM's `InputSigSent -> +// Confirmed` transition (and the actor's `onRoundComplete` cleanup), +// both of which are terminal. Until the FSM gains a provisional/ +// finalized split, the actor-level handler for this event can only +// log the divergence so an operator notices and the future systests +// have something to assert against. Routing to the (now stopped) FSM +// would be a no-op even if the round were still tracked, because +// ConfirmedState has no transition for a "commitment reorged" event. +type CommitmentReorgedEvent struct { + actor.BaseMessage + + // Txid identifies the commitment transaction whose previously + // observed confirmation has been rolled back. + Txid chainhash.Hash +} + +func (m *CommitmentReorgedEvent) MessageType() string { + return "CommitmentReorgedEvent" +} + +// RoundReceivable implements actormsg.RoundReceivable marker interface. +func (m *CommitmentReorgedEvent) RoundReceivable() {} + +// CommitmentFinalizedEvent wraps a chainsource ConfDoneEvent that +// reports a commitment-tx confirmation is past the backend's reorg- +// safety depth and is no longer reversible. The current FSM treats +// the first confirmation as terminal; once the provisional/finalized +// FSM split lands, this is the signal that promotes the round from +// provisional to truly final. +type CommitmentFinalizedEvent struct { + actor.BaseMessage + + // Txid identifies the commitment transaction whose confirmation + // is now past the reorg-safety horizon. + Txid chainhash.Hash +} + +func (m *CommitmentFinalizedEvent) MessageType() string { + return "CommitmentFinalizedEvent" +} + +// RoundReceivable implements actormsg.RoundReceivable marker interface. +func (m *CommitmentFinalizedEvent) RoundReceivable() {} + // TimeoutMsg is sent to the round actor when a timeout expires. type TimeoutMsg struct { actor.BaseMessage diff --git a/round/actor_test.go b/round/actor_test.go index 8bdc8d2da..f872c80b3 100644 --- a/round/actor_test.go +++ b/round/actor_test.go @@ -290,6 +290,24 @@ func TestActorRecovery(t *testing.T) { reg := h.chainSource.registrations[0] require.NotNil(t, reg.Txid) require.True(t, reg.Txid.IsEqual(&txid)) + + // Reorg-aware lifecycle refs must be wired so the chainsource + // conf sub-actor keeps the registration alive past first + // confirmation, synthesizes a Done at the reorg-safety + // horizon, and surfaces TxReorged on rollback. Without these + // refs, the actor would never see reorg or finality signals + // for the commitment tx. + require.True( + t, reg.NotifyReorged.IsSome(), + "RegisterConfRequest must wire NotifyReorged so "+ + "the commitment-tx reorg lifecycle reaches "+ + "the actor", + ) + require.True( + t, reg.NotifyDone.IsSome(), + "RegisterConfRequest must wire NotifyDone so the "+ + "finality horizon reaches the actor", + ) }) t.Run("multiple_active_rounds", func(t *testing.T) { @@ -987,6 +1005,235 @@ func TestActorGetStateWithActiveRounds(t *testing.T) { } } +// TestActorRoundCommitmentLifecycleGatedOnFinality pins the round's +// reorg-safety contract end-to-end at the actor level: +// +// - A ConfirmationEvent is PROVISIONAL. It caches the event on +// pendingCommitmentConfs[txid] and does NOT advance the FSM — +// user-visible side effects (VTXOs marked live in the local +// store, ledger emissions, indexer notifications, onRoundComplete +// cleanup) must not fire until the commitment is past the +// reorg-safety horizon. +// +// - A CommitmentReorgedEvent leaves the FSM in its pre-confirmation +// state (no rollback needed; nothing was committed). The cached +// entry is retained so a follow-up re-confirmation overwrites it +// before the chainsource finality synthesizer can fire. +// +// - A second ConfirmationEvent (re-confirmation after a reorg) +// overwrites the cached entry with the new canonical-chain +// height; the finality synthesizer's depth counter resets on +// reorg, so the eventual Done event always reflects the latest +// re-confirmation. +// +// - A CommitmentFinalizedEvent consumes the cached entry, drives +// BoardingConfirmed into the FSM, and the FSM's terminal-state +// transition fires onRoundComplete which clears the round from +// the actor's tracking maps. +// +// - A CommitmentFinalizedEvent without a cached prior conf +// (defensive: chainsource should not synthesize Done without a +// prior positive event) is a no-op rather than a crash. +// +// - Events for untracked / already-cleaned-up txids ack without +// affecting any other rounds. +func TestActorRoundCommitmentLifecycleGatedOnFinality(t *testing.T) { + t.Parallel() + + t.Run("untracked_txid_is_benign", func(t *testing.T) { + t.Parallel() + + h := newActorTestHarness(t) + h.setupMockRoundStoreForStart() + require.NoError(t, h.start()) + + untrackedTxid := chainhash.Hash{0xfe} + + require.True( + t, h.receive( + &ConfirmationEvent{Txid: untrackedTxid}, + ).IsOk(), + "ConfirmationEvent on untracked txid must ack", + ) + require.True( + t, h.receive( + &CommitmentReorgedEvent{Txid: untrackedTxid}, + ).IsOk(), + "Reorged on untracked txid must ack", + ) + require.True( + t, h.receive( + &CommitmentFinalizedEvent{Txid: untrackedTxid}, + ).IsOk(), + "Finalized on untracked txid must ack", + ) + + // No cache entries should have been installed for the + // untracked txid (the index gate prevents it). + require.NotContains( + t, h.actor.pendingCommitmentConfs, untrackedTxid, + ) + }) + + t.Run("confirmation_caches_without_fsm_transition", + func(t *testing.T) { + t.Parallel() + + h := newActorTestHarness(t) + h.setupMockRoundStoreForStart() + require.NoError(t, h.start()) + + roundID := testRoundID("test-round-conf") + packet := h.setupRoundInInputSigSentState(roundID) + txid := packet.UnsignedTx.TxHash() + h.actor.commitmentTxIndex[txid] = RoundKeyStr( + roundID.KeyString(), + ) + + require.True( + t, h.receive(&ConfirmationEvent{ + Txid: txid, + BlockHeight: 105, + Confirmations: 1, + }).IsOk(), + ) + + // Cache populated, FSM untouched: the round is still + // in the actor's tracking maps because onRoundComplete + // has not run. + cached, ok := h.actor.pendingCommitmentConfs[txid] + require.True( + t, ok, + "ConfirmationEvent must populate the cache", + ) + require.Equal(t, int32(105), cached.BlockHeight) + require.Contains( + t, h.actor.rounds, + RoundKeyStr( + roundID.KeyString(), + ), + "FSM must remain tracked before finality", + ) + require.Contains(t, h.actor.commitmentTxIndex, txid) + }) + + t.Run("reorg_before_finality_keeps_state", + func(t *testing.T) { + t.Parallel() + + h := newActorTestHarness(t) + h.setupMockRoundStoreForStart() + require.NoError(t, h.start()) + + roundID := testRoundID("test-round-reorg") + packet := h.setupRoundInInputSigSentState(roundID) + txid := packet.UnsignedTx.TxHash() + keyStr := RoundKeyStr(roundID.KeyString()) + h.actor.commitmentTxIndex[txid] = keyStr + + require.True( + t, h.receive(&ConfirmationEvent{ + Txid: txid, + BlockHeight: 200, + Confirmations: 1, + }).IsOk(), + ) + require.True( + t, h.receive( + &CommitmentReorgedEvent{Txid: txid}, + ).IsOk(), + ) + + // Reorg before finality: no rollback work to do + // because nothing user-visible was committed. The + // round stays tracked; the cache stays populated + // (a follow-up re-confirmation will overwrite it). + require.Contains(t, h.actor.rounds, keyStr) + require.Contains(t, h.actor.commitmentTxIndex, txid) + require.Contains( + t, h.actor.pendingCommitmentConfs, txid, + ) + }) + + t.Run("reconfirmation_overwrites_cache_with_canonical_height", + func(t *testing.T) { + t.Parallel() + + h := newActorTestHarness(t) + h.setupMockRoundStoreForStart() + require.NoError(t, h.start()) + + roundID := testRoundID("test-round-reconf") + packet := h.setupRoundInInputSigSentState(roundID) + txid := packet.UnsignedTx.TxHash() + h.actor.commitmentTxIndex[txid] = RoundKeyStr( + roundID.KeyString(), + ) + + require.True( + t, h.receive(&ConfirmationEvent{ + Txid: txid, + BlockHeight: 300, + Confirmations: 1, + }).IsOk(), + ) + require.True( + t, h.receive( + &CommitmentReorgedEvent{Txid: txid}, + ).IsOk(), + ) + require.True( + t, h.receive(&ConfirmationEvent{ + Txid: txid, + BlockHeight: 301, + Confirmations: 1, + }).IsOk(), + ) + + cached := h.actor.pendingCommitmentConfs[txid] + require.NotNil(t, cached) + require.Equal( + t, int32(301), cached.BlockHeight, "second "+ + "confirmation must overwrite the "+ + "cache with the canonical-chain "+ + "height; finality will replay the "+ + "latest entry, not a stale "+ + "pre-reorg observation", + ) + }) + + t.Run("finality_without_prior_conf_is_a_no_op", + func(t *testing.T) { + t.Parallel() + + h := newActorTestHarness(t) + h.setupMockRoundStoreForStart() + require.NoError(t, h.start()) + + roundID := testRoundID("test-round-bare-final") + packet := h.setupRoundInInputSigSentState(roundID) + txid := packet.UnsignedTx.TxHash() + keyStr := RoundKeyStr(roundID.KeyString()) + h.actor.commitmentTxIndex[txid] = keyStr + + // Defensive path: chainsource should not synthesize + // Done without a prior positive event (the depth + // synthesizer is gated on confirmHeight != 0), but + // the handler must not crash if it ever does. + require.True( + t, h.receive( + &CommitmentFinalizedEvent{Txid: txid}, + ).IsOk(), + ) + + require.Contains( + t, h.actor.rounds, keyStr, "Finalized "+ + "without cached conf must not "+ + "trigger an FSM transition", + ) + }) +} + // TestActorReceiveUnknownMessageType ensures that the actor rejects // unrecognized message types with an appropriate error rather than silently // ignoring them. diff --git a/systest/boarding_sweep_reorg_test.go b/systest/boarding_sweep_reorg_test.go new file mode 100644 index 000000000..608dff989 --- /dev/null +++ b/systest/boarding_sweep_reorg_test.go @@ -0,0 +1,299 @@ +//go:build systest + +package systest + +import ( + "context" + "crypto/sha256" + "testing" + "time" + + btcaddr "github.com/btcsuite/btcd/address/v2" + "github.com/btcsuite/btcd/btcutil/v2" + "github.com/btcsuite/btcd/chaincfg/v2" + "github.com/btcsuite/btcd/txscript/v2" + "github.com/lightninglabs/darepo-client/baselib/actor" + "github.com/lightninglabs/darepo-client/txconfirm" + "github.com/lightninglabs/darepo-client/wallet" + "github.com/stretchr/testify/require" +) + +// recordingBoardingSweepRef collects the BoardingSweepTxNotification +// values that txconfirm's lifecycle, fanned through the boarding +// sweep MapNotification, delivers to the actor's mailbox. The test +// uses this to assert that each chainsource event maps to the right +// BoardingSweepTxStatus rather than collapsing reorgs into a failure. +type recordingBoardingSweepRef struct { + id string + msgs chan wallet.BoardingSweepTxNotification +} + +func newRecordingBoardingSweepRef(id string) *recordingBoardingSweepRef { + return &recordingBoardingSweepRef{ + id: id, + msgs: make(chan wallet.BoardingSweepTxNotification, 16), + } +} + +// ID returns the subscriber identifier. +func (r *recordingBoardingSweepRef) ID() string { + return r.id +} + +// Tell records the inbound boarding-sweep notification on the channel +// for the test to consume. +func (r *recordingBoardingSweepRef) Tell(_ context.Context, + msg wallet.WalletMsg) error { + + notif, ok := msg.(wallet.BoardingSweepTxNotification) + if !ok { + + // Not a notification we care about (and there should be + // no others on this subscriber path). + return nil + } + + r.msgs <- notif + + return nil +} + +// await pulls the next boarding-sweep notification or fails the test +// on timeout. +func (r *recordingBoardingSweepRef) await( + t *testing.T) wallet.BoardingSweepTxNotification { + + t.Helper() + + select { + case msg := <-r.msgs: + return msg + + case <-time.After(txConfirmSystestEventTimeout): + t.Fatalf("timeout waiting for boarding-sweep notification (%s)", + txConfirmSystestEventTimeout) + + return wallet.BoardingSweepTxNotification{} + } +} + +// TestBoardingSweepReorgRoundTrip exercises the boarding sweep +// MapNotification under a real bitcoind reorg, end-to-end through the +// full txconfirm pipeline. It is the boarding-sweep complement of +// TestTxConfirmReorgRoundTrip: the same chain-event substrate, but +// asserting on the wallet-level lifecycle statuses that drive +// handleSweepTxNotification rather than the raw txconfirm +// notifications. +// +// Lifecycle asserted: +// +// BoardingSweepTxStatusConfirmed (first conf landed) +// BoardingSweepTxStatusReorged (conf block disconnected) +// BoardingSweepTxStatusConfirmed (re-confirmation on canonical chain) +// +// TxFinalized is exercised by TestTxConfirmReorgRoundTrip — that test +// proves the chainsource finality-depth synthesizer fires after the +// reorg-safety horizon, which is the same wire that would deliver +// BoardingSweepTxStatusFinalized to the wallet handler. Including a +// dedicated Finalized assertion here would require mining +// DefaultFinalityDepth more blocks (~6 additional rounds of harness +// generation) without exercising any boarding-sweep-specific code +// path that the unit tests in +// wallet/boarding_sweep_actor_test.go::TestSweepTxNotificationFinalizedIsBenign +// do not already cover. +// +// What this test specifically pins: +// +// - The boarding-sweep MapNotification correctly classifies a real +// chainsource Confirmed event as +// BoardingSweepTxStatusConfirmed (not Failed). +// - The boarding-sweep MapNotification correctly classifies a real +// chainsource Reorged event as +// BoardingSweepTxStatusReorged (not Failed). This is the load- +// bearing regression the wallet refactor was designed to prevent. +// - The handler is multi-shot: txconfirm keeps the watch armed +// past the first TxConfirmed, so a re-confirmation on the +// canonical chain re-fires +// BoardingSweepTxStatusConfirmed. +// - BlockHeight is preserved across the wire on each Confirmed +// event. +func TestBoardingSweepReorgRoundTrip(t *testing.T) { + ParallelN(t) + + h := NewSysTestHarness(t) + ctx := h.Context() + + chainSource := h.NewChainSourceActor() + + // Spawn a real txconfirm actor over the real chainsource. Wallet + // is nil because the systest tx is non-anchor; CPFP fee-input + // selection is never triggered. + txconfBehavior := txconfirm.NewTxBroadcasterActor(txconfirm.Config{ + ChainSource: chainSource, + }) + txconfInstance := actor.NewActor(actor.ActorConfig[ + txconfirm.Msg, txconfirm.Resp, + ]{ + ID: "txconfirm-boarding-sweep-reorg", + Behavior: txconfBehavior, + MailboxSize: 64, + }) + txconfBehavior.SetSelfRef(txconfInstance.TellRef()) + txconfInstance.Start() + t.Cleanup(txconfInstance.Stop) + + // Synthetic watched pkScript — same trick as + // TestTxConfirmReorgRoundTrip. + pubKeyHash := sha256.Sum256([]byte(t.Name())) + addr, err := btcaddr.NewAddressWitnessPubKeyHash( + pubKeyHash[:20], &chaincfg.RegressionNetParams, + ) + require.NoError(t, err, "build synthetic P2WPKH address") + pkScript, err := txscript.PayToAddrScript(addr) + require.NoError(t, err, "derive pkScript for synthetic address") + + heightHint := h.Harness.BlockCount() + + signedTx := h.Harness.SignedV3Tx( + pkScript, btcutil.Amount(btcutil.SatoshiPerBitcoin/100), + ) + txidVal := signedTx.TxHash() + txid := &txidVal + t.Logf("constructed v3 tx: txid=%s", txid) + + // Build the subscriber chain that the production code uses: + // + // walletNotif (TellOnlyRef[BoardingSweepTxNotification], wrap + // via MapMessage so it can sit on a WalletMsg pipe) + // -> wallet.MapBoardingSweepNotification (the production + // classifier from boarding_sweep_actor.go) hosted on a + // txconfirm subscriber. + // + // The recording ref stands in for the wallet actor's mailbox; the + // MapNotification function is the same one the production + // submitSweepConfirmer wires up. + walletRecorder := newRecordingBoardingSweepRef( + "boarding-sweep-systest-sub", + ) + subscriber := wallet.NewBoardingSweepTxconfirmSubscriber( + walletRecorder, + ) + + // Register the EnsureConfirmedReq BEFORE mining so we exercise + // the live-detection path. + _, err = txconfInstance.Ref().Ask( + ctx, &txconfirm.EnsureConfirmedReq{ + Tx: signedTx, + ConfirmationPkScript: pkScript, + Label: "systest-boarding-sweep-reorg", + HeightHint: heightHint, + TargetConfs: 1, + Subscriber: subscriber, + }, + ).Await(ctx).Unpack() + require.NoError(t, err, "EnsureConfirmedReq failed") + + // 1. Mine the block that confirms the systest tx. + originalBlocks := h.Harness.Generate(1) + require.Len(t, originalBlocks, 1) + originalBlock := originalBlocks[0] + + // 2. Expect Status=Confirmed at the original block height with + // at least one confirmation. NumConfs being non-zero pins the + // classifier's assignment of ev.NumConfs through the + // txconfirm.TxConfirmed -> BoardingSweepTxNotification mapping; + // a regression that zeroed it (e.g. by accidentally pulling + // from the wrong field on the event) would otherwise be silently + // invisible to consumers that read NumConfs to gate on + // confirmation depth. + first := walletRecorder.await(t) + require.Equal( + t, wallet.BoardingSweepTxStatusConfirmed, first.Status, "fir"+ + "st notification must be Confirmed, got status=%d", + first.Status, + ) + require.Equal(t, *txid, first.Txid) + require.Equal( + t, int32(originalBlock.Height), first.BlockHeight, + "first Confirmed BlockHeight should match the mined block", + ) + require.GreaterOrEqual( + t, first.NumConfs, uint32(1), + "first Confirmed NumConfs must be at least the target (1)", + ) + t.Logf( + "first Confirmed: txid=%s height=%d num_confs=%d", first.Txid, + first.BlockHeight, first.NumConfs, + ) + + // 3. Reorg the conf block out, mine a strictly longer replacement + // branch, wait for lnd to chain-sync. + reorg := h.Harness.Reorg(1, 2) + require.Equal( + t, originalBlock.Hash, reorg.Disconnected[0].Hash, + "the reorg should have disconnected the conf block", + ) + require.Len(t, reorg.Connected, 2) + t.Logf( + "reorg: disconnected=%d connected=%d fork=%d", + len(reorg.Disconnected), len(reorg.Connected), + reorg.ForkPoint.Height, + ) + + // 4. Expect Status=Reorged. The load-bearing assertion: a real + // chainsource Reorged event must NOT classify as Failed (which + // would trigger MarkBoardingSweepFailed in the production + // handler). + second := walletRecorder.await(t) + require.Equal( + t, wallet.BoardingSweepTxStatusReorged, second.Status, "seco"+ + "nd notification must be Reorged (NOT Failed); got "+ + "status=%d. A Reorged-as-Failed regression is "+ + "exactly what this systest exists to catch", + second.Status, + ) + require.Equal(t, *txid, second.Txid) + require.Empty( + t, second.Reason, "Reorged must not carry a failure reason", + ) + t.Logf("Reorged: txid=%s", second.Txid) + + // 5. Expect a fresh Status=Confirmed on the canonical chain. The + // re-confirmation must land in one of the new connected blocks + // (NumConfs is also asserted to pin the field-through invariant + // under the re-confirmation path — the classifier's TxConfirmed + // arm is exercised twice in this lifecycle). Note: the + // re-confirmation can legitimately land at the SAME block height + // as the disconnected block — Reorg(1, 2) replaces 1 block with + // 2, so the first new block sits at the same height as the old + // one, just with a different hash. The notification doesn't + // carry BlockHash so we can't directly assert hash inequality; + // the canonical-height membership check below is the strongest + // available signal that this is a fresh confirmation on the new + // tip rather than a recycled stale event. + third := walletRecorder.await(t) + require.Equal( + t, wallet.BoardingSweepTxStatusConfirmed, third.Status, "thi"+ + "rd notification must be Confirmed "+ + "(re-confirmation), got status=%d", third.Status, + ) + require.Equal(t, *txid, third.Txid) + canonicalHeights := make(map[int32]struct{}, len(reorg.Connected)) + for _, blk := range reorg.Connected { + canonicalHeights[int32(blk.Height)] = struct{}{} + } + _, onCanonical := canonicalHeights[third.BlockHeight] + require.True( + t, onCanonical, "re-confirmation BlockHeight=%d must be one "+ + "of the new connected block heights %v", + third.BlockHeight, reorg.Connected, + ) + require.GreaterOrEqual( + t, third.NumConfs, uint32(1), + "re-confirmation NumConfs must be at least the target (1)", + ) + t.Logf( + "re-Confirmed: txid=%s height=%d num_confs=%d", third.Txid, + third.BlockHeight, third.NumConfs, + ) +} diff --git a/systest/round_commitment_reorg_test.go b/systest/round_commitment_reorg_test.go new file mode 100644 index 000000000..16f623305 --- /dev/null +++ b/systest/round_commitment_reorg_test.go @@ -0,0 +1,210 @@ +//go:build systest + +package systest + +import ( + "crypto/sha256" + "testing" + + btcaddr "github.com/btcsuite/btcd/address/v2" + "github.com/btcsuite/btcd/btcutil/v2" + "github.com/btcsuite/btcd/chaincfg/v2" + "github.com/btcsuite/btcd/chainhash/v2" + "github.com/btcsuite/btcd/txscript/v2" + "github.com/lightninglabs/darepo-client/baselib/actor" + "github.com/lightninglabs/darepo-client/chainsource" + "github.com/lightningnetwork/lnd/fn/v2" + "github.com/stretchr/testify/require" +) + +// TestRoundCommitmentReorgRoundTrip drives the full reorg-aware +// confirmation lifecycle through real bitcoind for a watch shaped +// like the one round/actor.go registers for a server-built commitment +// transaction. The round actor's Option B finality-gating contract +// depends on three properties of the chainsource pipeline holding +// over the real gRPC transport: +// +// 1. The watch is multi-shot once NotifyReorged or NotifyDone is +// wired — a single registration must surface +// ConfirmationEvent -> ConfReorgedEvent -> ConfirmationEvent -> +// ConfDoneEvent, in that order, without the sub-actor tearing +// itself down after the first positive event. +// +// 2. The chainsource FinalityDepth synthesizer (default six blocks) +// actually fires on a real chain. lndclient over gRPC does not +// write the upstream Done channel, so without the synthesizer +// no Done event would ever arrive at the round actor and the +// FSM would stay parked in InputSigSent forever. This is the +// property TestChainSourceConfReorgRoundTrip deliberately does +// NOT cover — it leaves Done unwired — so this test is the +// systest-level oracle for the synthesizer's wire integration +// under a real reorg cycle. +// +// 3. A reorg between the first confirmation and finality resets the +// synthesizer's depth counter. Done must only fire once the +// re-confirmation is past the reorg-safety depth, not the +// pre-reorg observation. +// +// What this test pins specifically: the chainsource events reaching +// the actor layer match what the round actor's Option B handlers +// (handleConfirmation, handleCommitmentReorged, +// handleCommitmentFinalized) expect. The round actor's behavior +// under that event sequence is unit-tested in +// round/actor_test.go::TestActorRoundCommitmentLifecycleGatedOnFinality; +// this test proves the wire-level events the unit test mocks are +// genuinely delivered by the production pipeline. +func TestRoundCommitmentReorgRoundTrip(t *testing.T) { + ParallelN(t) + + h := NewSysTestHarness(t) + ctx := h.Context() + + chainSource := h.NewChainSourceActor() + + // Synthetic watched address — deterministic per test name. + pubKeyHash := sha256.Sum256([]byte(t.Name())) + addr, err := btcaddr.NewAddressWitnessPubKeyHash( + pubKeyHash[:20], &chaincfg.RegressionNetParams, + ) + require.NoError(t, err, "build synthetic P2WPKH address") + pkScript, err := txscript.PayToAddrScript(addr) + require.NoError(t, err, "derive pkScript for synthetic address") + + // Three recording refs mirroring the round actor's wiring in + // registerCommitmentConfirmation: MapConfirmationEvent, + // MapConfReorgedEvent, MapConfDoneEvent. Using + // ChannelTellOnlyRef instead of a real actor keeps the test + // focused on the wire-level event sequence without standing up + // the round FSM (whose behavior is unit-tested separately). + confRef := actor.NewChannelTellOnlyRef[chainsource.ConfirmationEvent]( + "round-commit-conf", 8, + ) + reorgRef := actor.NewChannelTellOnlyRef[chainsource.ConfReorgedEvent]( + "round-commit-reorged", 4, + ) + doneRef := actor.NewChannelTellOnlyRef[chainsource.ConfDoneEvent]( + "round-commit-done", 4, + ) + + // Faucet first so we have a known txid, then register, then mine. + // The watch is keyed on (txid, pkScript) so the ordering between + // mempool entry and registration is fine. Mining BEFORE + // registration would dispatch historical confirmation state and + // race against the live path. + heightHint := h.Harness.BlockCount() + amount := btcutil.Amount(btcutil.SatoshiPerBitcoin / 100) + txidStr := h.Harness.Faucet(addr.String(), amount) + txid, err := chainhash.NewHashFromStr(txidStr) + require.NoError(t, err, "parse faucet txid") + t.Logf("faucet tx: txid=%s", txid) + + confNotify := actor.TellOnlyRef[chainsource.ConfirmationEvent]( + confRef, + ) + reorgNotify := actor.TellOnlyRef[chainsource.ConfReorgedEvent]( + reorgRef, + ) + doneNotify := actor.TellOnlyRef[chainsource.ConfDoneEvent]( + doneRef, + ) + + _, err = chainSource.Ask(ctx, &chainsource.RegisterConfRequest{ + CallerID: "round-commit-reorg-systest-" + txidStr, + Txid: txid, + PkScript: pkScript, + TargetConfs: 1, + HeightHint: heightHint, + NotifyActor: fn.Some(confNotify), + NotifyReorged: fn.Some(reorgNotify), + NotifyDone: fn.Some(doneNotify), + }).Await(ctx).Unpack() + require.NoError(t, err, "RegisterConfRequest failed") + + // 1. Mine the conf block. + originalBlocks := h.Harness.Generate(1) + require.Len(t, originalBlocks, 1) + originalBlock := originalBlocks[0] + originalHash, err := chainhash.NewHashFromStr(originalBlock.Hash) + require.NoError(t, err, "parse original block hash") + + // 2. Expect ConfirmationEvent at the original block height. + first := awaitConfEvent(t, confRef) + require.Equal(t, *txid, first.Txid) + require.Equal(t, int32(originalBlock.Height), first.BlockHeight) + require.Equal(t, *originalHash, first.BlockHash) + t.Logf( + "first Confirmation: txid=%s height=%d hash=%s", first.Txid, + first.BlockHeight, first.BlockHash, + ) + + // 3. Reorg the conf block out. + reorg := h.Harness.Reorg(1, 2) + require.Equal( + t, originalBlock.Hash, reorg.Disconnected[0].Hash, + "reorg must disconnect the original conf block", + ) + require.Len(t, reorg.Connected, 2) + t.Logf( + "reorg: disconnected=%d connected=%d fork=%d", + len(reorg.Disconnected), len(reorg.Connected), + reorg.ForkPoint.Height, + ) + + // 4. Expect ConfReorgedEvent — the property the round actor's + // handleCommitmentReorged consumes. The FSM stays in + // InputSigSent because nothing user-visible was committed. + reorgEvt := awaitReorgEvent(t, reorgRef) + require.Equal(t, *txid, reorgEvt.Txid) + t.Logf("Reorged: txid=%s", reorgEvt.Txid) + + // 5. Re-ConfirmationEvent on the canonical chain. Bitcoind + // preserves the tx in its mempool across the invalidate, so the + // faucet tx re-confirms in one of the new connected blocks. + second := awaitConfEvent(t, confRef) + require.Equal(t, *txid, second.Txid) + require.NotEqual( + t, *originalHash, second.BlockHash, "re-confirmation "+ + "BlockHash must differ from the disconnected "+ + "block; if it matches, the watch is replaying the "+ + "stale pre-reorg event", + ) + t.Logf( + "re-Confirmation: txid=%s height=%d hash=%s", second.Txid, + second.BlockHeight, second.BlockHash, + ) + + // 6. Mine `DefaultFinalityDepth - 1` additional blocks past the + // re-conf height. The synthesizer fires Done when the block + // epoch subscription observes a height of + // `confirmHeight + FinalityDepth - 1`. The re-confirmation block + // itself counts as depth=1, so we need (FinalityDepth-1) more + // blocks. With DefaultFinalityDepth=6 that's 5 more blocks. + additionalBlocks := int(chainsource.DefaultFinalityDepth) - 1 + t.Logf( + "mining %d additional blocks to trigger finality "+ + "synthesizer (FinalityDepth=%d)", additionalBlocks, + chainsource.DefaultFinalityDepth, + ) + h.Harness.Generate(additionalBlocks) + + // 7. Expect ConfDoneEvent — the property the round actor's + // handleCommitmentFinalized consumes. Without this event, the + // round FSM under Option B would never reach its terminal state + // (no user-visible side effects, no onRoundComplete cleanup). + doneEvt, ok := doneRef.AwaitMessage(reorgSystestEventTimeout) + require.True( + t, ok, "timeout waiting for ConfDoneEvent from the "+ + "FinalityDepth synthesizer (%s); without this "+ + "signal the round FSM would never reach ConfirmedState", + reorgSystestEventTimeout, + ) + require.Equal( + t, *txid, doneEvt.Txid, + "Done event txid must match the watched commitment tx", + ) + t.Logf( + "Done: txid=%s (synthesizer fired at depth=%d past "+ + "re-confirmation)", doneEvt.Txid, + int(chainsource.DefaultFinalityDepth), + ) +} diff --git a/wallet/boarding_sweep_actor.go b/wallet/boarding_sweep_actor.go index 2620ee413..ac15ca42c 100644 --- a/wallet/boarding_sweep_actor.go +++ b/wallet/boarding_sweep_actor.go @@ -210,19 +210,63 @@ func (m *ReplayPendingIntentsResponse) MessageType() string { func (m *ReplayPendingIntentsResponse) walletRespSealed() {} -// BoardingSweepSpendNotification is a Tell carrying a chainsource spend -// event for a boarding-sweep input. Emitted by the chainsource subscription -// the wallet actor sets up via MapSpendEvent. +// BoardingSweepSpendStatus identifies which point in the chainsource +// spend-watch reorg-aware lifecycle drove this notification. Splitting +// the status out makes it impossible for the handler to confuse a +// reorg-out with a fresh spend: a reorged-out spending block should +// not advance the input's persistent state machine further than the +// canonical chain has. +type BoardingSweepSpendStatus int + +const ( + // BoardingSweepSpendStatusUnknown is the zero value; receiving it + // indicates a programmer error (MapSpendEvent fell through). + BoardingSweepSpendStatusUnknown BoardingSweepSpendStatus = iota + + // BoardingSweepSpendStatusSpent reports that the watched outpoint + // was observed spent on the canonical chain by SpendingTxid at + // SpendingHeight. The observation is provisional until + // BoardingSweepSpendStatusDone arrives — a reorg can roll it back + // via BoardingSweepSpendStatusReorged. + BoardingSweepSpendStatusSpent + + // BoardingSweepSpendStatusReorged reports that a previously + // delivered Spent observation was reorged out. The handler logs + // the divergence and leaves the watch armed; the chainsource + // sub-actor stays alive in reorg-aware mode so a subsequent + // re-spend on the new canonical chain re-fires + // BoardingSweepSpendStatusSpent. + BoardingSweepSpendStatusReorged + + // BoardingSweepSpendStatusDone reports that the spend observation + // is past the chainsource backend's reorg-safety depth and is no + // longer reversible. Synthesized by the spend-actor's + // FinalityDepth synthesizer when the backend (notably lndclient + // over gRPC) does not surface a native Done signal. + BoardingSweepSpendStatusDone +) + +// BoardingSweepSpendNotification is a Tell carrying one event from the +// chainsource spend-watch reorg-aware lifecycle for a boarding-sweep +// input. Emitted by the subscription the wallet actor sets up via +// MapSpendEvent / MapSpendReorgedEvent / MapSpendDoneEvent. type BoardingSweepSpendNotification struct { actor.BaseMessage - // Outpoint is the boarding UTXO that was spent. + // Status identifies which lifecycle event this notification + // carries. See BoardingSweepSpendStatus. + Status BoardingSweepSpendStatus + + // Outpoint is the boarding UTXO whose spend lifecycle changed. Outpoint wire.OutPoint - // SpendingTxid is the transaction that confirmed the spend. + // SpendingTxid is the transaction that spent the outpoint when + // Status=Spent; zero on Reorged / Done (the wire-level event + // carries no spender identity past the first Spent observation). SpendingTxid chainhash.Hash - // SpendingHeight is the block height of the spending transaction. + // SpendingHeight is the block height of the spending transaction + // when Status=Spent; zero on Reorged / Done. SpendingHeight int32 } @@ -924,6 +968,7 @@ func (a *Ark) registerSweepSpendWatch(ctx context.Context, notify := chainsource.MapSpendEvent(a.selfRef, func(ev chainsource.SpendEvent) WalletMsg { return BoardingSweepSpendNotification{ + Status: BoardingSweepSpendStatusSpent, Outpoint: ev.Outpoint, SpendingTxid: ev.SpendingTxid, SpendingHeight: ev.SpendingHeight, @@ -931,16 +976,44 @@ func (a *Ark) registerSweepSpendWatch(ctx context.Context, }, ) + // Reorg-aware lifecycle refs. Without these, a reorged-out + // external spender would leave the input row marked + // external_spent in the persistent store with no rollback path — + // the chainsource SpendActor would either tear the registration + // down (legacy mode) or hold reorg events for nobody (reorg-aware + // mode missing the refs). Wiring them keeps the registration + // alive past the first Spent event, surfaces Reorged on rollback, + // and lets the FinalityDepth synthesizer eventually fire Done so + // the sub-actor exits cleanly. + reorgedNotify := chainsource.MapSpendReorgedEvent(a.selfRef, + func(ev chainsource.SpendReorgedEvent) WalletMsg { + return BoardingSweepSpendNotification{ + Status: BoardingSweepSpendStatusReorged, + Outpoint: ev.Outpoint, + } + }, + ) + doneNotify := chainsource.MapSpendDoneEvent(a.selfRef, + func(ev chainsource.SpendDoneEvent) WalletMsg { + return BoardingSweepSpendNotification{ + Status: BoardingSweepSpendStatusDone, + Outpoint: ev.Outpoint, + } + }, + ) + callerID := boardingSweepCallerID(op) notifyOpt := fn.Some[actor.TellOnlyRef[chainsource.SpendEvent]]( notify, ) req := &chainsource.RegisterSpendRequest{ - CallerID: callerID, - Outpoint: &op, - PkScript: out.PkScript, - HeightHint: heightHint, - NotifyActor: notifyOpt, + CallerID: callerID, + Outpoint: &op, + PkScript: out.PkScript, + HeightHint: heightHint, + NotifyActor: notifyOpt, + NotifyReorged: fn.Some(reorgedNotify), + NotifyDone: fn.Some(doneNotify), } future := a.chainSource.Ask(ctx, req) @@ -1070,8 +1143,31 @@ func (a *Ark) submitSweepToConfirm(ctx context.Context, tx *wire.MsgTx, return nil } -// handleSweepSpendNotification updates persistent state when an input of -// an in-flight aggregate sweep is observed spent on chain. +// handleSweepSpendNotification updates persistent state when the +// reorg-aware spend lifecycle on an in-flight aggregate-sweep input +// changes: +// +// - Status=Spent: record the spend in the persistent store via +// MarkBoardingSweepInputSpent. This is the existing happy path. +// +// - Status=Reorged: a previously delivered Spent observation was +// rolled back on the canonical chain. The handler logs and does +// NOT undo the store row — the txid-keyed idempotency of +// MarkBoardingSweepInputSpent means a re-spend on the new +// canonical chain (which chainsource will surface as another +// Spent event because the watch stays alive in reorg-aware mode) +// either matches the already-stored spender (no-op) or rejects +// the row update with sql.ErrNoRows (existing handler treats as +// debug). Rolling the row back to pending on reorg would require +// a new store method; for now we accept the brief inconsistency +// window, which is bounded by the FinalityDepth synthesizer +// firing Done at the reorg-safety horizon. Documented as a +// known limitation in the boarding-sweep reorg handler. +// +// - Status=Done: the spend is past the reorg-safety horizon and +// the chainsource sub-actor will exit. Informational at the +// wallet layer; the row state was already committed on the +// original Spent event and is no longer reversible. func (a *Ark) handleSweepSpendNotification(ctx context.Context, notif BoardingSweepSpendNotification) fn.Result[WalletResp] { @@ -1079,6 +1175,45 @@ func (a *Ark) handleSweepSpendNotification(ctx context.Context, return fn.Ok[WalletResp](&BoardingSweepNotificationAck{}) } + switch notif.Status { + case BoardingSweepSpendStatusSpent: + // Fall through to the existing Spent handling below. + + case BoardingSweepSpendStatusReorged: + // Reorg-out of a previously delivered Spent observation. + // Leave the store row in place; the next Spent event on + // the canonical chain (or, in the worst case, manual + // reconciliation at restart) brings local state back in + // sync. We log so operators have a signal and so future + // regression tests can assert the reorg path actually + // reaches the wallet layer. + a.logger(ctx).WarnS(ctx, "Boarding-sweep input spend "+ + "reorged out; leaving store row as-is and "+ + "waiting for re-spend on the canonical chain", + nil, + slog.String("outpoint", notif.Outpoint.String()), + ) + + return fn.Ok[WalletResp](&BoardingSweepNotificationAck{}) + + case BoardingSweepSpendStatusDone: + a.logger(ctx).DebugS(ctx, "Boarding-sweep input spend "+ + "finalized past reorg-safety depth", + slog.String("outpoint", notif.Outpoint.String()), + ) + + return fn.Ok[WalletResp](&BoardingSweepNotificationAck{}) + + default: + a.logger(ctx).WarnS(ctx, "Boarding-sweep spend notification "+ + "with unknown status", + fmt.Errorf("status=%d", notif.Status), + slog.String("outpoint", notif.Outpoint.String()), + ) + + return fn.Ok[WalletResp](&BoardingSweepNotificationAck{}) + } + resolved, err := a.sweepStore.MarkBoardingSweepInputSpent( ctx, notif.Outpoint, notif.SpendingTxid, notif.SpendingHeight, ) diff --git a/wallet/boarding_sweep_actor_test.go b/wallet/boarding_sweep_actor_test.go index a8d103eae..ad6e8f2a8 100644 --- a/wallet/boarding_sweep_actor_test.go +++ b/wallet/boarding_sweep_actor_test.go @@ -432,6 +432,7 @@ func TestSweepSpendNotificationMarksInputSpent(t *testing.T) { result := a.handleSweepSpendNotification( t.Context(), BoardingSweepSpendNotification{ + Status: BoardingSweepSpendStatusSpent, Outpoint: op, SpendingTxid: spendingTxid, SpendingHeight: 220, @@ -1229,3 +1230,132 @@ func TestSweepTxNotificationUnknownStatusIsBenign(t *testing.T) { store.AssertExpectations(t) } + +// TestSweepSpendNotificationReorgedDoesNotMarkSpent verifies that a +// chainsource SpendReorgedEvent on a watched boarding-sweep input is +// classified as BoardingSweepSpendStatusReorged and processed by the +// Reorged arm of handleSweepSpendNotification — specifically, the +// store's MarkBoardingSweepInputSpent must NOT be called (would +// double-spend the row's state machine) and pendingSweeps tracking +// must be left intact (the chainsource sub-actor stays alive in +// reorg-aware mode; a follow-up Spent on the canonical chain will +// drive the row through the existing happy path). +func TestSweepSpendNotificationReorgedDoesNotMarkSpent(t *testing.T) { + t.Parallel() + + op := wire.OutPoint{Hash: chainhash.Hash{0xb1}, Index: 0} + + store := &MockBoardingSweepStore{} + // CRITICAL: MarkBoardingSweepInputSpent must NOT be called on + // reorg. The testify/mock will fail the test if any unexpected + // method is invoked, so we simply do not register + // MarkBoardingSweepInputSpent here. + + a := newSweepTestArk(t, store, nil, 0, 0) + + pendingTxid := chainhash.Hash{0xb2} + pending := &pendingSweepState{ + txid: pendingTxid, + inputs: map[wire.OutPoint]string{ + op: boardingSweepCallerID(op), + }, + } + a.pendingSweeps[pendingTxid] = pending + a.pendingSweepInputs[op] = pendingTxid + + result := a.handleSweepSpendNotification( + t.Context(), BoardingSweepSpendNotification{ + Status: BoardingSweepSpendStatusReorged, + Outpoint: op, + }, + ) + require.True(t, result.IsOk()) + + // Reorg must leave both in-memory tracking maps alone — the + // sub-actor stays alive in reorg-aware mode and a follow-up + // Spent event on the canonical chain will re-drive the row. + require.Same( + t, pending, a.pendingSweeps[pendingTxid], + "reorg must not evict pending sweep tracking", + ) + require.Equal( + t, pendingTxid, a.pendingSweepInputs[op], + "reorg must not evict per-input back-pointer", + ) + + store.AssertExpectations(t) +} + +// TestSweepSpendNotificationDoneIsBenign verifies that a chainsource +// SpendDoneEvent (the reorg-safety horizon is reached for the spend +// observation) does not call any store method and does not perturb +// in-memory tracking. The row state was already committed on the +// original Spent event; Done is informational at the wallet layer. +func TestSweepSpendNotificationDoneIsBenign(t *testing.T) { + t.Parallel() + + op := wire.OutPoint{Hash: chainhash.Hash{0xc1}, Index: 0} + + store := &MockBoardingSweepStore{} + // No store expectations — Done must touch nothing. + + a := newSweepTestArk(t, store, nil, 0, 0) + + pendingTxid := chainhash.Hash{0xc2} + pending := &pendingSweepState{ + txid: pendingTxid, + inputs: map[wire.OutPoint]string{ + op: boardingSweepCallerID(op), + }, + } + a.pendingSweeps[pendingTxid] = pending + a.pendingSweepInputs[op] = pendingTxid + + result := a.handleSweepSpendNotification( + t.Context(), BoardingSweepSpendNotification{ + Status: BoardingSweepSpendStatusDone, + Outpoint: op, + }, + ) + require.True(t, result.IsOk()) + + require.Same(t, pending, a.pendingSweeps[pendingTxid]) + require.Equal(t, pendingTxid, a.pendingSweepInputs[op]) + + store.AssertExpectations(t) +} + +// TestSweepSpendNotificationUnknownStatusIsBenign verifies the default +// arm of handleSweepSpendNotification handles an unrecognised status +// without touching the store. Guards against a future chainsource +// lifecycle addition being delivered without a matching MapSpendEvent +// classification arm. +func TestSweepSpendNotificationUnknownStatusIsBenign(t *testing.T) { + t.Parallel() + + op := wire.OutPoint{Hash: chainhash.Hash{0xd1}, Index: 0} + store := &MockBoardingSweepStore{} + // No expectations — unknown must touch nothing. + + a := newSweepTestArk(t, store, nil, 0, 0) + pendingTxid := chainhash.Hash{0xd2} + pending := &pendingSweepState{ + txid: pendingTxid, + inputs: map[wire.OutPoint]string{ + op: boardingSweepCallerID(op), + }, + } + a.pendingSweeps[pendingTxid] = pending + a.pendingSweepInputs[op] = pendingTxid + + result := a.handleSweepSpendNotification( + t.Context(), BoardingSweepSpendNotification{ + Status: BoardingSweepSpendStatusUnknown, + Outpoint: op, + }, + ) + require.True(t, result.IsOk()) + require.Same(t, pending, a.pendingSweeps[pendingTxid]) + + store.AssertExpectations(t) +}