diff --git a/chainsource/finality.go b/chainsource/finality.go index 2b8561b1f..5bdc97ac6 100644 --- a/chainsource/finality.go +++ b/chainsource/finality.go @@ -21,6 +21,16 @@ var finalityBlockSubscriptionBackoffs = []time.Duration{ 2 * time.Second, } +// finalityBlockSubscriptionAttemptTimeout bounds each individual +// RegisterBlocks attempt. Without it a single hung RegisterBlocks call +// (e.g. a wedged lndclient gRPC stream) would block the conf/spend +// monitoring goroutine indefinitely — stalling Confirmed/Reorged/Done +// delivery on that watch — since the retry schedule only bounds the gaps +// between attempts, not the attempts themselves. 10s mirrors the per-call +// registration timeout used in conf_actor.go's handleRegisterConf so the +// whole file behaves consistently under a slow backend. +const finalityBlockSubscriptionAttemptTimeout = 10 * time.Second + // registerBlocksForFinality registers a block-epoch subscription used // to synthesize a Done signal at FinalityDepth past an observed // confirmation or spend. The call is retried with a short bounded @@ -29,22 +39,11 @@ var finalityBlockSubscriptionBackoffs = []time.Duration{ // lndclient over gRPC); a one-shot RegisterBlocks attempt that // briefly hiccups would leak the per-watch sub-actor indefinitely. // -// The retries run in a dedicated arming goroutine (not the sub-actor's -// select loop), so brief blocking here is safe: more confirmation/spend -// events on this specific watch are not expected during the retry window -// (we already consumed the one that triggered the arm), and ctx -// cancellation breaks out promptly. -// -// The passed ctx MUST be the sub-actor's long-lived context, and it is -// handed to RegisterBlocks unwrapped: for in-process backends the -// block-epoch forwarder goroutine is tied to the ctx it receives, so -// bounding each attempt with a cancellable child ctx (and cancelling it -// once the call returns) would tear the subscription down the instant it -// was armed — starving finality synthesis of the very epochs it needs. -// A hung RegisterBlocks can therefore stall this arming goroutine, but -// that is contained: it is off the select loop (fix moved arming there -// precisely so a slow backend cannot wedge Confirmed/Reorged/Done -// delivery), and a genuinely wedged backend is a lost watch regardless. +// The retries run in the calling sub-actor's monitoring goroutine, so +// brief blocking here is safe: more confirmation/spend events on this +// specific watch are not expected during the retry window (we already +// consumed the one that triggered the arm), and ctx cancellation +// breaks out promptly. // // Returns the registration on success, or a non-nil error after // retries are exhausted. Callers should log the error at warn level @@ -55,7 +54,14 @@ func registerBlocksForFinality(ctx context.Context, backend ChainBackend, var lastErr error for attempt, backoff := range finalityBlockSubscriptionBackoffs { - reg, err := backend.RegisterBlocks(ctx) + // Bound each attempt so a hung RegisterBlocks cannot wedge the + // monitoring goroutine; the retry schedule only bounds the gaps + // between attempts, not a single stuck call. + attemptCtx, cancel := context.WithTimeout( + ctx, finalityBlockSubscriptionAttemptTimeout, + ) + reg, err := backend.RegisterBlocks(attemptCtx) + cancel() if err == nil { return reg, nil } diff --git a/darepod/config.go b/darepod/config.go index 1304759eb..e3f3f9b15 100644 --- a/darepod/config.go +++ b/darepod/config.go @@ -389,6 +389,15 @@ type UnrollConfig struct { // MaxFeeRateSatPerVByte caps fee estimates to prevent runaway // fees. Zero uses the default of 100 sat/vB. MaxFeeRateSatPerVByte int64 `mapstructure:"maxfeeratesatpervbyte"` + + // ReconcileProbeTimeoutSec bounds each per-anchor restart- + // reconciliation probe issued by the chainsource-backed + // ChainReconciler (in seconds). A probe that times out is + // treated as "not on chain" and triggers a conservative + // rollback of the affected anchor; operators running against + // a slow backend can raise this to avoid spurious rollbacks. + // Zero uses the reconciler's internal default (10s). + ReconcileProbeTimeoutSec int64 `mapstructure:"reconcileprobetimeoutsec"` } // FeeEstimationConfig groups optional external chain fee providers used by the diff --git a/darepod/server.go b/darepod/server.go index c7f5a7e2b..f4623b8c9 100644 --- a/darepod/server.go +++ b/darepod/server.go @@ -5299,6 +5299,36 @@ func (s *Server) initUnrollSubsystem(ctx context.Context, exitObserver = fn.Some[actor.TellOnlyRef[vtxo.ManagerMsg]](ref) }) + // Build a chainsource-backed reconciler factory so every per- + // target actor verifies its checkpoint anchors against the + // canonical chain on restart. Without this, a daemon that was + // offline during a reorg would silently resume against stale + // PlannerState — see unroll/reconcile.go for the safety + // rationale. + // + // The factory bakes the target outpoint into the chainsource + // caller-ID prefix so two actors that reconcile the same + // shared proof-graph ancestor concurrently land on distinct + // service keys (chainsource keys on (CallerID, Txid, PkScript, + // TargetConfs)); a static prefix would collide. + reconcileLog := s.subLogger("UREC") + probeTimeout := s.unrollReconcileProbeTimeout() + reconcilerFactory := func(target wire.OutPoint, + proof *recovery.Proof) unroll.ChainReconciler { + + return unroll.NewChainSourceReconciler( + unroll.ChainSourceReconcilerConfig{ + ChainSource: chainSourceRef, + Proof: proof, + CallerID: fmt.Sprintf( + "unroll-reconcile-%s", target, + ), + ProbeTimeout: probeTimeout, + Log: fn.Some(reconcileLog), + }, + ) + } + registry := unroll.NewUnrollRegistryActor(unroll.RegistryConfig{ Store: &unroll.DBRegistryStore{ UEStore: ueStore, @@ -5319,6 +5349,9 @@ func (s *Server) initUnrollSubsystem(ctx context.Context, Preimage: preimages, }, VTXOExitObserver: exitObserver, + ChainReconcilerFactory: fn.Some( + unroll.ChainReconcilerFactory(reconcilerFactory), + ), }) s.unrollRegistry = registry s.unrollRegistryRef = fn.Some(registry.Ref()) @@ -5659,6 +5692,20 @@ func (s *Server) unrollMaxFeeRate() int64 { return 0 } +// unrollReconcileProbeTimeout returns the configured per-anchor probe +// timeout for the chainsource-backed restart reconciler, or zero so +// the reconciler falls back to defaultReconcileProbeTimeout. +func (s *Server) unrollReconcileProbeTimeout() time.Duration { + if s.cfg.Unroll != nil && + s.cfg.Unroll.ReconcileProbeTimeoutSec > 0 { + return time.Duration( + s.cfg.Unroll.ReconcileProbeTimeoutSec, + ) * time.Second + } + + return 0 +} + // unrollBumpAfterBlocks returns the configured fee-bump cadence (in // blocks) for the shared txconfirm actor used by the unroll subsystem, // or zero to let txconfirm fall back to DefaultFeeBumpIntervalBlocks. diff --git a/sample-darepod.conf b/sample-darepod.conf index 25705421a..3b325d3fd 100644 --- a/sample-darepod.conf +++ b/sample-darepod.conf @@ -199,6 +199,14 @@ # Maximum unroll fee rate in sat/vB. The zero value uses the unroll default. # unroll.maxfeeratesatpervbyte=0 +# Per-anchor restart-reconciliation probe timeout in seconds. The chainsource- +# backed ChainReconciler issues one probe per persisted anchor on Resume; a +# probe that exceeds this budget is treated as "not on chain" and triggers a +# conservative rollback of the affected anchor. Operators running against a +# slow chain backend can raise this to avoid spurious rollbacks. The zero +# value uses the reconciler's internal default (10s). +# unroll.reconcileprobetimeoutsec=0 + # Swap server gRPC address for swapruntime builds. # swap.serveraddress=localhost:10030 diff --git a/unroll/actor.go b/unroll/actor.go index 322cff497..c73906798 100644 --- a/unroll/actor.go +++ b/unroll/actor.go @@ -74,6 +74,21 @@ type Config struct { // LedgerSink receives the confirmed on-chain exit fee once the // final sweep has confirmed. LedgerSink fn.Option[ledger.Sink] + + // ChainReconcilerFactory, when set, is invoked after the actor + // loads its proof to construct a ChainReconciler that reconciles + // the persisted checkpoint against the canonical chain at + // startup. Anchors that are no longer live on chain (typically + // reorged out while the daemon was offline) are pruned before + // the FSM session is built so the actor does not broadcast a + // sweep on top of stale planner state. + // + // The factory shape lets the production implementation + // (NewChainSourceReconciler) bind to the actor's proof for + // pkScript lookups while still letting unit tests inject a stub + // that ignores the proof. When None the actor skips + // reconciliation entirely. + ChainReconcilerFactory fn.Option[ChainReconcilerFactory] } // VTXOUnrollActor wraps one durable per-target unroll actor. @@ -157,6 +172,22 @@ type behavior struct { proofSpendWatches map[wire.OutPoint]struct{} terminalNotified bool exitCostNotified bool + + // reconciled records that the restored checkpoint has been + // reconciled against the canonical chain via cfg.ChainReconciler. + // Reconciliation runs exactly once per actor lifetime, on the + // first ensureLoaded call, before the FSM session is bound. + reconciled bool + + // sweepFinalized latches true after a TxFinalizedMsg arrives for + // the txid currently recorded as the sweep in PlannerState. While + // false, PhaseCompleted is treated as PROVISIONAL: the registry is + // NOT told the actor has terminated, so a reorg of the sweep + // confirmation still has a live actor to deliver the rollback to. + // Once sweepFinalized is true the next notifyRegistryOfTerminal + // call fires UnrollTerminatedMsg and the registry evicts the + // child. + sweepFinalized bool } // unrollTx is the transaction-scoped store handed to the unroll behavior inside @@ -287,9 +318,37 @@ func (b *behavior) dispatch(ctx context.Context, ax actor.Exec[unrollTx], Reason: b.failureReasonForTx(m.Txid, m.Reason), }) + case *TxReorgedMsg: + return b.handleEvent(ctx, ax, &TxReorgedEvent{ + Txid: m.Txid, + }) + + case *TxFinalizedMsg: + // Latch sweepFinalized BEFORE handleEvent so the + // notifyRegistry call inside driveEvent sees the + // post-finalization view and fires UnrollTerminatedMsg + // instead of holding the actor in provisional Completed. + if err := b.ensureLoaded(ctx); err != nil { + return fn.Err[Resp](err) + } + b.maybeLatchSweepFinalized(m.Txid) + + return b.handleEvent(ctx, ax, &TxFinalizedEvent{ + Txid: m.Txid, + }) + case *SpendObservedMsg: return b.handleSpendObserved(ctx, ax, m) + case *SpendReorgedMsg: + return b.handleEvent(ctx, ax, &SpendReorgedEvent{}) + + case *SpendFinalizedMsg: + return b.handleEvent(ctx, ax, &SpendFinalizedEvent{}) + + case *GetStateRequest: + return fn.Ok[Resp](b.stateResponse()) + default: return fn.Err[Resp]( fmt.Errorf("unknown unroll message: %T", msg), @@ -779,14 +838,35 @@ func (b *behavior) watchDeferredCheckpoint(ctx context.Context, }, ) + // Deferred-checkpoint watches register directly against + // chainsource rather than txconfirm, but the rollback contract + // is identical: an operator-confirmed checkpoint that reorgs out + // while the actor is live must drop from ConfirmedTxids so the + // planner does not advance off a stale anchor, and a finality + // signal lets the chainsource sub-actor release its registration + // at the reorg-safety horizon. Wire both lifecycle refs through + // the same selfRef the positive event uses. + reorgRef := chainsource.MapConfReorgedEvent( + b.selfRef, func(event chainsource.ConfReorgedEvent) Msg { + return &TxReorgedMsg{Txid: event.Txid} + }, + ) + doneRef := chainsource.MapConfDoneEvent( + b.selfRef, func(event chainsource.ConfDoneEvent) Msg { + return &TxFinalizedMsg{Txid: event.Txid} + }, + ) + txidCopy := txid _, err = b.cfg.ChainSource.Ask(ctx, &chainsource.RegisterConfRequest{ - CallerID: b.deferredCheckpointCallerID(), - Txid: &txidCopy, - PkScript: append([]byte(nil), pkScript...), - TargetConfs: 1, - HeightHint: proofNodeHeightHint, - NotifyActor: fn.Some(notifyRef), + CallerID: b.deferredCheckpointCallerID(), + Txid: &txidCopy, + PkScript: append([]byte(nil), pkScript...), + TargetConfs: 1, + HeightHint: proofNodeHeightHint, + NotifyActor: fn.Some(notifyRef), + NotifyReorged: fn.Some(reorgRef), + NotifyDone: fn.Some(doneRef), }).Await(ctx).Unpack() if err != nil { return err @@ -884,6 +964,23 @@ func (b *behavior) ensureLoaded(ctx context.Context) error { b.planner = planner } + // Reconcile the restored checkpoint against the canonical chain + // before binding the FSM session. Any confirmed-tx anchor the + // reconciler reports as absent gets dropped (with its + // descendants), and a target / sweep confirmation that vanished + // while the daemon was offline downgrades the sweep state. This + // must run before stateFromCheckpoint so the FSM is built from + // the post-reconciliation snapshot, and before any side effects + // (block subscription, spend watch, ResumeEvent reissue) so a + // stale sweep is never re-broadcast on a now-unconfirmed target. + if !b.reconciled { + if err := b.reconcileOnRestart(ctx); err != nil { + return err + } + + b.reconciled = true + } + if b.session == nil { initialState := State(&Idle{}) if b.pending != nil && b.pending.Started { @@ -946,6 +1043,16 @@ func (b *behavior) notificationRef() actor.TellOnlyRef[txconfirm.Notification] { Reason: m.Reason, } + case *txconfirm.TxReorged: + return &TxReorgedMsg{ + Txid: m.Txid, + } + + case *txconfirm.TxFinalized: + return &TxFinalizedMsg{ + Txid: m.Txid, + } + default: return &TxFailedMsg{ Reason: fmt.Sprintf( @@ -988,10 +1095,40 @@ func (b *behavior) restoreCheckpoint(ctx context.Context) error { b.pending = decoded b.sweepTx = copyTx(decoded.SweepTx) + b.sweepFinalized = decoded.SweepFinalized return nil } +// reconcileOnRestart asks the configured ChainReconciler whether each +// anchor recorded in the restored checkpoint is still live on the +// canonical chain, and prunes the in-memory checkpoint accordingly. +// Called from ensureLoaded exactly once per actor lifetime, BEFORE the +// FSM session is bound. +// +// A missing reconciler is a no-op: backends that cannot answer +// historical confirmation queries fall back to the conservative +// behavior of treating the restored checkpoint as authoritative. +// Transport errors from the reconciler are surfaced upward; the actor +// must not proceed to broadcast off a checkpoint we could not verify. +func (b *behavior) reconcileOnRestart(ctx context.Context) error { + if b.cfg.ChainReconcilerFactory.IsNone() { + return nil + } + + if b.pending == nil || !b.pending.Started { + return nil + } + + factory := b.cfg.ChainReconcilerFactory.UnsafeFromSome() + reconciler := factory(b.cfg.TargetOutpoint, b.proof) + if reconciler == nil { + return nil + } + + return reconcileCheckpoint(ctx, reconciler, b.proof, b.pending) +} + // ensureBlockSubscription starts the actor's shared block epoch subscription on // first use so CSV waits advance in the live daemon. func (b *behavior) ensureBlockSubscription(ctx context.Context) error { @@ -1077,14 +1214,28 @@ func (b *behavior) ensureSpendWatch(ctx context.Context) error { } }, ) + reorgRef := chainsource.MapSpendReorgedEvent( + b.selfRef, + func(_ chainsource.SpendReorgedEvent) Msg { + return &SpendReorgedMsg{} + }, + ) + doneRef := chainsource.MapSpendDoneEvent( + b.selfRef, + func(_ chainsource.SpendDoneEvent) Msg { + return &SpendFinalizedMsg{} + }, + ) _, err = b.cfg.ChainSource.Ask( ctx, &chainsource.RegisterSpendRequest{ - CallerID: b.spendCallerID(), - Outpoint: &targetOutpoint, - PkScript: pkScript, - HeightHint: uint32(b.desc.CreatedHeight), - NotifyActor: fn.Some(notifyRef), + CallerID: b.spendCallerID(), + Outpoint: &targetOutpoint, + PkScript: pkScript, + HeightHint: uint32(b.desc.CreatedHeight), + NotifyActor: fn.Some(notifyRef), + NotifyReorged: fn.Some(reorgRef), + NotifyDone: fn.Some(doneRef), }, ).Await(ctx).Unpack() if err != nil { @@ -1340,18 +1491,36 @@ func (b *behavior) handleSpendObserved(ctx context.Context, } } - // Case 4: neither of the above. Someone else spent the watched output. - // This happens if the operator cooperatively claimed the VTXO, - // if a reorg replaced history, or in fraud scenarios. There is - // no way for this unroll to proceed, so terminate with a - // reason string that identifies the spender for operator - // triage. - spentOutpoint := b.cfg.TargetOutpoint - if msg.Outpoint != (wire.OutPoint{}) { - spentOutpoint = msg.Outpoint + // Case 4: neither of the above. Someone else spent the watched + // output. This can happen for two structurally different + // outputs, with different reversibility properties: + // + // a. The target outpoint was spent by an unknown party. This + // is the cooperative-claim / fraud scenario the external- + // spend reorg-safety work is built for: a reorg of the + // spending block can resurrect the recovery job, so the + // actor parks in AwaitingExternalSpendFinality. A + // subsequent SpendFinalizedMsg promotes the spend to a + // permanent FailReason; a SpendReorgedMsg clears the + // observation and resumes planning. + // + // b. A proof-node output was spent by a transaction that is + // not itself part of the proof graph (ackProofOutputSpend + // acked the parent confirmation already, but returned + // false for the spender lookup). The proof graph cannot + // complete through this fork, so the unroll job is + // terminally dead. Fail with a reason that identifies the + // spender for operator triage. + if msg.Outpoint == (wire.OutPoint{}) || + msg.Outpoint == b.cfg.TargetOutpoint { + return b.handleEvent(ctx, ax, &ExternalSpendObservedEvent{ + SpendingTxid: msg.SpendingTxid, + SpendingHeight: msg.SpendingHeight, + }) } + reason := fmt.Sprintf("watched outpoint %s spent externally by tx %s "+ - "at height %d", spentOutpoint, msg.SpendingTxid, + "at height %d", msg.Outpoint, msg.SpendingTxid, msg.SpendingHeight) return b.handleEvent(ctx, ax, &FailEvent{Reason: reason}) @@ -1424,6 +1593,14 @@ func (b *behavior) checkpointWrite() (*actorCheckpoint, } checkpoint := checkpointFromState(state, b.sweepTx) + + // Persist the sweep-finalized latch so it survives restart: commitAck + // re-persists this checkpoint atomically with the message ack, so a + // finalized sweep whose terminal handoff was deferred/failed is not + // lost — the actor rehydrates as terminal-eligible rather than stuck + // "provisional completed". + checkpoint.SweepFinalized = b.sweepFinalized + raw, err := encodeCheckpoint(checkpoint) if err != nil { return nil, nil, err @@ -1728,19 +1905,60 @@ func (b *behavior) failureReasonForTx(txid chainhash.Hash, return fmt.Sprintf("proof tx %s failed: %s", txid, reason) } +// maybeLatchSweepFinalized sets b.sweepFinalized when an incoming +// TxFinalizedMsg refers to the txid currently recorded as the sweep in +// PlannerState. The flag gates notifyRegistryIfTerminal so that a +// PhaseCompleted entry without a matching finalization stays +// provisional — the actor is kept alive by the registry until the +// sweep is past the backend's reorg-safety depth, so a reorg of the +// sweep confirmation has a live actor to receive the rollback. +// +// Late finalizations (e.g. for a proof tx, or for a sweep txid that has +// since been replaced by a re-broadcast) are ignored. +func (b *behavior) maybeLatchSweepFinalized(txid chainhash.Hash) { + state, err := b.currentState() + if err != nil { + return + } + + job := stateJob(state) + if job.PlannerState.Sweep.Txid.IsNone() { + return + } + if job.PlannerState.Sweep.Txid.UnsafeFromSome() != txid { + return + } + + b.sweepFinalized = true +} + // notifyRegistryIfTerminal forwards one UnrollTerminatedMsg to the -// registry when the FSM reaches Completed or Failed, at most once per -// actor lifetime. +// registry when the FSM reaches a TRULY terminal state, at most once +// per actor lifetime. +// +// PhaseFailed always means FailReason has been set (proof-tx terminal +// failure, sweep retry budget exhausted, or a finalized external +// spend); none of those are reorg-recoverable so the actor is terminal +// the moment we get there. +// +// PhaseCompleted is treated as PROVISIONAL until b.sweepFinalized is +// latched by a matching TxFinalizedMsg for the recorded sweep txid. +// While provisional, the registry keeps the child in its active map so +// a reorg of the sweep confirmation has a live actor to deliver the +// rollback to. The chainsource transport (lndclient over gRPC) does +// not surface a finality signal today, so production deployments may +// retain the child indefinitely after a sweep confirms; height-based +// finality gating is a Phase 7 follow-up. // -// The registry uses this to move the outpoint out of its active map and -// mark the durable store terminal. If the FSM receives additional events -// after reaching terminal (e.g. a late TxConfirmed for a proof node that -// materialized before we failed for other reasons) terminalNotified -// keeps us from spamming the registry with repeats. +// PhaseExternalSpendObserved is reversible by design and never reaches +// this function as terminal; SpendFinalizedMsg promotes the +// provisional anchor to FailReason and the actor moves through +// PhaseFailed instead. // -// Failure to Tell is warned but not fatal — the registry will rediscover -// the terminal phase the next time it queries child state, and it holds -// its own persistence retry loop for the control-plane record. +// Failure to Tell is warned but not fatal — the registry will +// rediscover the terminal phase the next time it queries child state, +// and it holds its own persistence retry loop for the control-plane +// record. func (b *behavior) notifyRegistryIfTerminal(ctx context.Context) { state, err := b.currentState() if err != nil { @@ -1750,7 +1968,9 @@ func (b *behavior) notifyRegistryIfTerminal(ctx context.Context) { } phase := phaseFromState(state) - if phase != PhaseCompleted && phase != PhaseFailed { + trulyTerminal := phase == PhaseFailed || + (phase == PhaseCompleted && b.sweepFinalized) + if !trulyTerminal { return } diff --git a/unroll/actor_test.go b/unroll/actor_test.go index cb71d0ad3..2dbc59367 100644 --- a/unroll/actor_test.go +++ b/unroll/actor_test.go @@ -392,6 +392,42 @@ func (f *fakeTxConfirmRef) emitFailed(t *testing.T, index int, require.NoError(t, err) } +// emitReorged delivers a txconfirm reorg notification to the subscriber +// behind the request at index. +func (f *fakeTxConfirmRef) emitReorged(t *testing.T, index int, + txid chainhash.Hash) { + + t.Helper() + + f.mu.Lock() + require.Less(t, index, len(f.requests)) + subscriber := f.requests[index].Subscriber + f.mu.Unlock() + + err := subscriber.Tell(t.Context(), &txconfirm.TxReorged{ + Txid: txid, + }) + require.NoError(t, err) +} + +// emitFinalized delivers a txconfirm finalized notification to the +// subscriber behind the request at index. +func (f *fakeTxConfirmRef) emitFinalized(t *testing.T, index int, + txid chainhash.Hash) { + + t.Helper() + + f.mu.Lock() + require.Less(t, index, len(f.requests)) + subscriber := f.requests[index].Subscriber + f.mu.Unlock() + + err := subscriber.Tell(t.Context(), &txconfirm.TxFinalized{ + Txid: txid, + }) + require.NoError(t, err) +} + // confRef aliases the chainsource confirmation notification target. type confRef = actor.TellOnlyRef[chainsource.ConfirmationEvent] @@ -401,15 +437,17 @@ type confReq = chainsource.RegisterConfRequest // fakeChainSourceRef is a minimal chainsource actor ref for sweep fee // estimation tests. type fakeChainSourceRef struct { - mu sync.Mutex - bestHeight int32 - feeRate int64 - feeErr error - blockRef actor.TellOnlyRef[chainsource.BlockEpoch] - spendRefs map[wire.OutPoint]spendEventRef - spendRegs []wire.OutPoint - confRefs map[chainhash.Hash]confRef - confReqs map[chainhash.Hash]*confReq + mu sync.Mutex + bestHeight int32 + feeRate int64 + feeErr error + blockRef actor.TellOnlyRef[chainsource.BlockEpoch] + spendRefs map[wire.OutPoint]spendEventRef + spendRegs []wire.OutPoint + spendReorgedRef actor.TellOnlyRef[chainsource.SpendReorgedEvent] + spendFinalizedRef actor.TellOnlyRef[chainsource.SpendDoneEvent] + confRefs map[chainhash.Hash]confRef + confReqs map[chainhash.Hash]*confReq } // spendEventRef is the fake chain-source spend notification actor reference. @@ -507,6 +545,18 @@ func (f *fakeChainSourceRef) Ask(_ context.Context, } f.spendRefs[outpoint] = msg.NotifyActor.UnwrapOr(nil) f.spendRegs = append(f.spendRegs, outpoint) + + // Reorg/finalized refs are only wired by ensureSpendWatch + // (target outpoint). Proof-node spend watches from + // ensureProofSpendWatches leave these unset; capture them + // only when the caller actually provided them so a later + // proof-node registration cannot wipe the target's refs. + if msg.NotifyReorged.IsSome() { + f.spendReorgedRef = msg.NotifyReorged.UnwrapOr(nil) + } + if msg.NotifyDone.IsSome() { + f.spendFinalizedRef = msg.NotifyDone.UnwrapOr(nil) + } f.mu.Unlock() promise.Complete( fn.Ok[chainsource.ChainSourceResp]( @@ -637,6 +687,40 @@ func (f *fakeChainSourceRef) spendRegistrations() []wire.OutPoint { return append([]wire.OutPoint(nil), f.spendRegs...) } +// emitSpendReorged delivers a SpendReorgedEvent to the subscribed actor. +func (f *fakeChainSourceRef) emitSpendReorged(t *testing.T) { + t.Helper() + + f.mu.Lock() + ref := f.spendReorgedRef + f.mu.Unlock() + + require.NotNil(t, ref) + require.NoError( + t, + ref.Tell( + t.Context(), chainsource.SpendReorgedEvent{}, + ), + ) +} + +// emitSpendFinalized delivers a SpendDoneEvent to the subscribed actor. +func (f *fakeChainSourceRef) emitSpendFinalized(t *testing.T) { + t.Helper() + + f.mu.Lock() + ref := f.spendFinalizedRef + f.mu.Unlock() + + require.NotNil(t, ref) + require.NoError( + t, + ref.Tell( + t.Context(), chainsource.SpendDoneEvent{}, + ), + ) +} + // fakeSweepWallet is a minimal signer plus wallet-destination test double. type fakeSweepWallet struct{} @@ -2706,6 +2790,12 @@ func TestSweepConfirmationCompletesActor(t *testing.T) { ) require.True(t, checkpoint.State.Sweep.ConfirmHeight.IsSome()) + // Reorg-safe completion treats PhaseCompleted as provisional on + // confirmation and defers the terminal handoff (and the ExitCostMsg + // emission) until the sweep finalizes past reorg-safety depth. Drive + // the finalize so the exit cost lands. + txconfirmRef.emitFinalized(t, 2, sweepTxid) + ledgerMsg, ok := ledgerSink.AwaitMessage(testTimeout) require.True(t, ok) exitCostMsg, ok := ledgerMsg.(*ledger.ExitCostMsg) @@ -2807,6 +2897,12 @@ func TestExitCostTellFailureDefersTerminalHandoff(t *testing.T) { return stateResp.Phase == PhaseCompleted }, testTimeout, 10*time.Millisecond) + // Finalize the sweep so PhaseCompleted is no longer provisional and + // the actor becomes terminal-eligible. The terminal handoff still + // requires a successful exit-cost emission, which fails here because + // the sink has no ledger actor behind it. + txconfirmRef.emitFinalized(t, 2, sweepTxid) + // The failing ledger sink must have deferred the terminal handoff: // the registry sees no UnrollTerminatedMsg. _, ok := registryRef.AwaitMessage(50 * time.Millisecond) @@ -3170,8 +3266,11 @@ func TestSweepFailureRetriesThenFails(t *testing.T) { require.Equal(t, maxSweepAttempts, checkpoint.SweepAttempts) } -// TestExternalSpendTerminatesActor verifies that an external spend of the -// target VTXO (not our proof nodes or sweep) terminates the actor. +// TestExternalSpendTerminatesActor verifies that an external spend of +// the target VTXO is treated as a provisional block, and that +// finalization promotes it to a terminal failure. Without the +// finalization signal the actor stays in AwaitingExternalSpendFinality +// so a reorg of the spending block has a live actor to resume. func TestExternalSpendTerminatesActor(t *testing.T) { proof := buildLinearProof(t) desc := testDescriptor(t, proof.TargetOutpoint(), proof.CSVDelay()) @@ -3185,15 +3284,26 @@ func TestExternalSpendTerminatesActor(t *testing.T) { Trigger: TriggerManual, }) - // Ensure spend watch is registered. + // Ensure spend watch is registered for the target outpoint with + // reorg / done callback refs wired so the actor can be driven + // through the full reversible lifecycle. require.Eventually(t, func() bool { + var targetRegistered bool for _, outpoint := range chainSource.spendRegistrations() { if outpoint == proof.TargetOutpoint() { - return true + targetRegistered = true + break } } + if !targetRegistered { + return false + } - return false + chainSource.mu.Lock() + defer chainSource.mu.Unlock() + + return chainSource.spendReorgedRef != nil && + chainSource.spendFinalizedRef != nil }, testTimeout, 10*time.Millisecond) // Simulate an external party spending the target VTXO. @@ -3202,6 +3312,21 @@ func TestExternalSpendTerminatesActor(t *testing.T) { t, proof.TargetOutpoint(), externalTxid, 101, ) + // The actor must enter the reversible + // AwaitingExternalSpendFinality phase rather than terminating. + require.Eventually(t, func() bool { + stateResp, ok := mustAsk( + t, unrollActor.Ref(), &GetStateRequest{}, + ).(*GetStateResp) + require.True(t, ok) + + return stateResp.Phase == PhaseExternalSpendObserved + }, testTimeout, 10*time.Millisecond) + + // Finalize the spend. The actor promotes the provisional anchor + // to a permanent FailReason and transitions to PhaseFailed. + chainSource.emitSpendFinalized(t) + require.Eventually(t, func() bool { stateResp, ok := mustAsk( t, unrollActor.Ref(), &GetStateRequest{}, diff --git a/unroll/fsm_logic.go b/unroll/fsm_logic.go index 862e5235c..a73a5d94a 100644 --- a/unroll/fsm_logic.go +++ b/unroll/fsm_logic.go @@ -91,6 +91,44 @@ func processEventWithJob(ctx context.Context, job *JobState, event Event, case *TxFailedEvent: applyFailedEvent(nextJob, e) + case *TxReorgedEvent: + applyReorgedEvent(nextJob, e, env) + + case *TxFinalizedEvent: + // TxFinalized is informational at the unroll layer: the + // underlying chain anchor is no longer reversible, but the + // planner has already accounted for the confirmation. Nothing + // to mutate; we still run the planner below in case the + // finality changes any derived decision. + + case *ExternalSpendObservedEvent: + nextJob.ProvisionalExternalSpend = fn.Some(ExternalSpendAnchor{ + SpendingTxid: e.SpendingTxid, + SpendingHeight: e.SpendingHeight, + }) + if e.SpendingHeight > nextJob.Height { + nextJob.Height = e.SpendingHeight + } + + case *SpendReorgedEvent: + nextJob.ProvisionalExternalSpend = + fn.None[ExternalSpendAnchor]() + + case *SpendFinalizedEvent: + // Finalizing a provisional external spend promotes it to a + // terminal FailReason. If no provisional anchor was held, + // SpendFinalized is informational and ignored. + nextJob.ProvisionalExternalSpend.WhenSome( + func(anchor ExternalSpendAnchor) { + nextJob.FailReason = fmt.Sprintf("target "+ + "spent externally by tx %s at height "+ + "%d (finalized)", anchor.SpendingTxid, + anchor.SpendingHeight) + nextJob.ProvisionalExternalSpend = + fn.None[ExternalSpendAnchor]() + }, + ) + case *SweepBroadcastedEvent: nextJob.PlannerState.Sweep.Status = unrollplan.SweepStatusBroadcasted @@ -195,9 +233,9 @@ func deriveStateTransition(ctx context.Context, job *JobState, env *Environment, } // Terminal short-circuit. applyFailedEvent populates FailReason - // for proof-tx terminal failures, and handleSpendObserved emits - // explicit FailEvents for external-spend detection; both land - // here before any planner work. + // for proof-tx terminal failures, and SpendFinalizedEvent promotes + // a provisional external spend to a permanent FailReason; both + // land here before any planner work. if job.FailReason != "" { return &StateTransition{ NextState: &Failed{ @@ -206,6 +244,22 @@ func deriveStateTransition(ctx context.Context, job *JobState, env *Environment, }, nil } + // Provisional external spend short-circuit. While the spend is + // observed but not finalized, the planner must not advance toward + // a sweep: on chain the target output no longer exists, and + // broadcasting a sweep on top would fail. A SpendReorgedEvent + // clears ProvisionalExternalSpend and the next derivation falls + // through to the normal planner phase decision; a + // SpendFinalizedEvent promotes it to FailReason and the previous + // branch handles the terminal transition. + if job.ProvisionalExternalSpend.IsSome() { + return &StateTransition{ + NextState: &AwaitingExternalSpendFinality{ + Job: job.Copy(), + }, + }, nil + } + // Consult the pure planner. Plan() is stateless; it reads // PlannerState + the proof graph and returns a snapshot with Done // / NeedSweep / CSV / Ready fields. All phase decisions below @@ -441,6 +495,151 @@ func applyFailedEvent(job *JobState, event *TxFailedEvent) { job.FailReason = event.Reason } +// applyReorgedEvent rolls back the chain anchor for a previously +// confirmed proof or sweep transaction. The semantics mirror +// applyConfirmedEvent in reverse: +// +// - If the reorged txid matches the recorded sweep txid, downgrade +// SweepStatus from Confirmed back to Broadcasted and clear the +// stored ConfirmHeight. The signed sweep bytes are still durable +// in the actor checkpoint and the txconfirm subscription is still +// live, so the actor naturally re-runs AwaitingSweepConfirmation +// until the sweep reconfirms. +// +// - Otherwise the reorged txid is a proof node: drop it from +// ConfirmedTxids so the planner stops treating it as ready. If +// the reorged tx is the target itself, clear TargetConfirmHeight +// so CSV maturity is recomputed when the target reconfirms, and +// downgrade any sweep that depended on the now-invalidated target +// confirmation. The proof node remains broadcastable; the next +// deriveStateTransition will route an EnsureReadyTransactions for +// it if the planner's frontier turns up the txid again. +// +// Height is never rolled back: chain height is a global property +// reported by the block subscription, not by the per-tx watch, so a +// reorged confirmation does not move best-height down even though it +// invalidates the per-tx anchor. +func applyReorgedEvent(job *JobState, event *TxReorgedEvent, env *Environment) { + if job == nil || event == nil { + return + } + + // Sweep reorg: downgrade SweepStatus so deriveStateTransition lands + // in AwaitingSweepConfirmation (the sweep tx is still on chain, + // pending re-confirmation) rather than Completed. + if job.PlannerState.Sweep.Txid.IsSome() && + job.PlannerState.Sweep.Txid.UnsafeFromSome() == event.Txid { + + if job.PlannerState.Sweep.Status == + unrollplan.SweepStatusConfirmed { + + job.PlannerState.Sweep.Status = + unrollplan.SweepStatusBroadcasted + } + job.PlannerState.Sweep.ConfirmHeight = fn.None[int32]() + + return + } + + // Proof-node reorg: clear the confirmed anchor for this txid AND + // every descendant that we had recorded as confirmed or in-flight. + // State.Validate enforces a topological invariant (every + // confirmed/in-flight node has confirmed parents) that would fail + // if we dropped only the immediate ancestor and left descendants + // in place. Pruning the entire reorged subtree keeps the planner + // state internally consistent; txconfirm's txid-keyed dedup + // absorbs the re-submits when the planner re-emits the same nodes + // on its ready frontier after the parent reconfirms. + reorgedSubtree := collectReorgedSubtree(env, event.Txid) + job.PlannerState.ConfirmedTxids = removeHashes( + job.PlannerState.ConfirmedTxids, reorgedSubtree, + ) + job.PlannerState.InFlightTxids = removeHashes( + job.PlannerState.InFlightTxids, reorgedSubtree, + ) + for txid := range reorgedSubtree { + job.DeferredCheckpoints = removeDeferredCheckpoint( + job.DeferredCheckpoints, txid, + ) + } + + // If this proof tx was the target, the CSV anchor is also gone. + // unrollplan's State.Validate enforces "broadcasted / confirmed + // sweep requires confirmed target" so any non-pending sweep must + // be reset to Pending when the target loses its anchor. The + // signed sweep bytes still live in the actor checkpoint + // (b.sweepTx); a re-confirmed target drives NeedSweep again and + // startSweep reuses those bytes rather than deriving a new wallet + // pkScript or producing a different sweep txid. + if env != nil && env.Proof != nil && + event.Txid == env.Proof.TargetOutpoint().Hash { + + job.PlannerState.TargetConfirmHeight = fn.None[int32]() + + if job.PlannerState.Sweep.Status != + unrollplan.SweepStatusPending { + + job.PlannerState.Sweep.Status = + unrollplan.SweepStatusPending + job.PlannerState.Sweep.Txid = fn.None[chainhash.Hash]() + job.PlannerState.Sweep.ConfirmHeight = fn.None[int32]() + } + } +} + +// collectReorgedSubtree returns the set of every txid transitively +// descended from root within the proof graph, inclusive of root itself. +// A nil environment or missing root yields a singleton set so the +// reducer can still drop the reorged txid from local bookkeeping even +// when the proof is not loaded. +func collectReorgedSubtree(env *Environment, + root chainhash.Hash) map[chainhash.Hash]struct{} { + + subtree := map[chainhash.Hash]struct{}{root: {}} + if env == nil || env.Proof == nil { + return subtree + } + + queue := []chainhash.Hash{root} + for len(queue) > 0 { + next := queue[0] + queue = queue[1:] + + children, err := env.Proof.ChildTxids(next) + if err != nil { + continue + } + for _, child := range children { + if _, ok := subtree[child]; ok { + continue + } + subtree[child] = struct{}{} + queue = append(queue, child) + } + } + + return subtree +} + +// removeHashes returns hashes with every entry in drop removed. +func removeHashes(hashes []chainhash.Hash, + drop map[chainhash.Hash]struct{}) []chainhash.Hash { + + if len(hashes) == 0 || len(drop) == 0 { + return hashes + } + + filtered := hashes[:0:0] + for _, h := range hashes { + if _, ok := drop[h]; ok { + continue + } + filtered = append(filtered, h) + } + + return filtered +} + // applySweepBuildFailed records a sweep build or broadcast failure and // decides whether to terminate or retry. // diff --git a/unroll/fsm_types.go b/unroll/fsm_types.go index 13f40db04..e21db6c51 100644 --- a/unroll/fsm_types.go +++ b/unroll/fsm_types.go @@ -103,6 +103,31 @@ type JobState struct { // SweepAttempts counts sweep build or broadcast failures so the actor // can retry up to maxSweepAttempts before giving up. SweepAttempts int + + // ProvisionalExternalSpend records an external spend of the target + // outpoint that has been observed but has not yet been finalized + // past the backend's reorg-safety depth. While set, the actor is + // parked: deriveStateTransition refuses to advance toward a sweep, + // since the on-chain state says the target output no longer exists. + // A SpendReorgedEvent clears the anchor and lets planning resume; + // a SpendFinalizedEvent promotes it to a permanent FailReason. + // + // This field is intentionally NOT persisted in the actor checkpoint. + // On restart the chainsource spend watch is re-registered and the + // current spend state (if any) is re-delivered before the actor + // drives any side effects. + ProvisionalExternalSpend fn.Option[ExternalSpendAnchor] +} + +// ExternalSpendAnchor captures the identity of an external spend of the +// target outpoint that the actor has observed but has not yet treated +// as final. +type ExternalSpendAnchor struct { + // SpendingTxid is the txid that consumed the target outpoint. + SpendingTxid chainhash.Hash + + // SpendingHeight is the block height the spending tx confirmed at. + SpendingHeight int32 } // Copy returns a deep copy of the job state. @@ -113,14 +138,15 @@ func (j *JobState) Copy() *JobState { deferred := copyDeferredCheckpoints(j.DeferredCheckpoints) copyState := &JobState{ - Height: j.Height, - Trigger: j.Trigger, - ExitPolicyKind: exitPolicyKind(j.ExitPolicyKind), - ExitPolicyRef: j.ExitPolicyRef, - PlannerState: copyPlannerState(j.PlannerState), - DeferredCheckpoints: deferred, - FailReason: j.FailReason, - SweepAttempts: j.SweepAttempts, + Height: j.Height, + Trigger: j.Trigger, + ExitPolicyKind: exitPolicyKind(j.ExitPolicyKind), + ExitPolicyRef: j.ExitPolicyRef, + PlannerState: copyPlannerState(j.PlannerState), + DeferredCheckpoints: deferred, + FailReason: j.FailReason, + SweepAttempts: j.SweepAttempts, + ProvisionalExternalSpend: j.ProvisionalExternalSpend, } return copyState @@ -203,6 +229,57 @@ type TxFailedEvent struct { // eventSealed marks TxFailedEvent as an FSM event. func (e *TxFailedEvent) eventSealed() {} +// TxReorgedEvent records that a previously confirmed proof or sweep +// transaction was reorged out of the canonical chain. +type TxReorgedEvent struct { + // Txid is the reorged transaction hash. + Txid chainhash.Hash +} + +// eventSealed marks TxReorgedEvent as an FSM event. +func (e *TxReorgedEvent) eventSealed() {} + +// TxFinalizedEvent records that a confirmation is past the backend's +// reorg-safety depth. +type TxFinalizedEvent struct { + // Txid is the finalized transaction hash. + Txid chainhash.Hash +} + +// eventSealed marks TxFinalizedEvent as an FSM event. +func (e *TxFinalizedEvent) eventSealed() {} + +// ExternalSpendObservedEvent records an external spend of the target +// outpoint that has not yet been finalized. The reducer parks the +// actor in AwaitingExternalSpendFinality. +type ExternalSpendObservedEvent struct { + // SpendingTxid is the txid that consumed the target outpoint. + SpendingTxid chainhash.Hash + + // SpendingHeight is the block height the spending tx confirmed at. + SpendingHeight int32 +} + +// eventSealed marks ExternalSpendObservedEvent as an FSM event. +func (e *ExternalSpendObservedEvent) eventSealed() {} + +// SpendReorgedEvent records that a previously observed spend of the +// target outpoint was reorged out of the canonical chain. The reducer +// clears any provisional external-spend block on JobState. +type SpendReorgedEvent struct{} + +// eventSealed marks SpendReorgedEvent as an FSM event. +func (e *SpendReorgedEvent) eventSealed() {} + +// SpendFinalizedEvent records that a previously observed external spend +// of the target outpoint is past the backend's reorg-safety depth. The +// reducer promotes any provisional external-spend block to a permanent +// FailReason. +type SpendFinalizedEvent struct{} + +// eventSealed marks SpendFinalizedEvent as an FSM event. +func (e *SpendFinalizedEvent) eventSealed() {} + // SweepBroadcastedEvent records that the actor built the final sweep and // submitted it to txconfirm. type SweepBroadcastedEvent struct { @@ -445,7 +522,46 @@ func (s *AwaitingSweepConfirmation) ProcessEvent(ctx context.Context, return processEventWithJob(ctx, s.Job, event, env) } +// AwaitingExternalSpendFinality indicates the actor has observed an +// external spend of the target outpoint but is waiting for either a +// reorg (which clears the observation and resumes planning) or a +// finality signal (which promotes the observation to a permanent +// failure). +type AwaitingExternalSpendFinality struct { + // Job is the durable FSM state. + Job *JobState +} + +// String returns a human-readable state label. +func (s *AwaitingExternalSpendFinality) String() string { + return "AwaitingExternalSpendFinality" +} + +// IsTerminal returns false because the observation is reversible. +func (s *AwaitingExternalSpendFinality) IsTerminal() bool { + return false +} + +// stateSealed marks AwaitingExternalSpendFinality as implementing State. +func (s *AwaitingExternalSpendFinality) stateSealed() {} + +// ProcessEvent delegates to the shared reducer so every event kind is +// applied uniformly; the reducer's planner-gate keeps the actor parked +// in this state until SpendReorgedEvent or SpendFinalizedEvent clears +// or promotes the provisional anchor. +func (s *AwaitingExternalSpendFinality) ProcessEvent(ctx context.Context, + event Event, env *Environment) (*StateTransition, error) { + + return processEventWithJob(ctx, s.Job, event, env) +} + // Completed indicates the final sweep has confirmed. +// +// Completed is reversible until finality: a TxReorgedEvent for the +// recorded sweep txid (or for the target tx that the sweep depended on) +// must be routed through the same reducer as the non-terminal states so +// the actor can roll back to AwaitingSweepConfirmation when the chain +// disagrees. Any other event is rejected as before. type Completed struct { // Job is the durable FSM state. Job *JobState @@ -456,19 +572,33 @@ func (s *Completed) String() string { return "Completed" } -// IsTerminal returns true because this state is terminal. +// IsTerminal reports whether the planner-derived state is still +// terminal. Completed remains the public phase label, but a reorg event +// can re-enter the FSM so it cannot be marked terminal at the protofsm +// level — otherwise the engine refuses to deliver the rollback event. func (s *Completed) IsTerminal() bool { - return true + return false } // stateSealed marks Completed as implementing State. func (s *Completed) stateSealed() {} -// ProcessEvent rejects further events in the terminal completed state. -func (s *Completed) ProcessEvent(context.Context, Event, *Environment) ( - *StateTransition, error) { +// ProcessEvent applies reorg / finality events in the completed state +// and absorbs every other event kind as an idempotent no-op. Late +// chain notifications (a TxConfirmedEvent or HeightUpdatedEvent that +// raced the terminal transition while the registry is draining the +// actor for cleanup) are already reflected in the terminal checkpoint +// and should not error. +func (s *Completed) ProcessEvent(ctx context.Context, event Event, + env *Environment) (*StateTransition, error) { + + switch event.(type) { + case *TxReorgedEvent, *TxFinalizedEvent: + return processEventWithJob(ctx, s.Job, event, env) - return nil, fmt.Errorf("completed state is terminal") + default: + return &StateTransition{NextState: s}, nil + } } // Failed indicates the actor reached terminal failure. diff --git a/unroll/messages.go b/unroll/messages.go index deb518646..31f191855 100644 --- a/unroll/messages.go +++ b/unroll/messages.go @@ -41,6 +41,10 @@ const ( txFailedMsgTLVType tlv.Type = 0x7904 getStateRequestTLVType tlv.Type = 0x7905 spendObservedMsgTLVType tlv.Type = 0x7906 + txReorgedMsgTLVType tlv.Type = 0x7907 + txFinalizedMsgTLVType tlv.Type = 0x7908 + spendReorgedMsgTLVType tlv.Type = 0x7909 + spendFinalizedMsgTLVType tlv.Type = 0x790a ) // Durable mailbox priorities. Admission stays at the default priority so a @@ -76,6 +80,9 @@ const ( spendObservedHeightRecType tlv.Type = 3 spendObservedOutHashRecType tlv.Type = 5 spendObservedOutIndexRecType tlv.Type = 7 + + txReorgedTxidRecType tlv.Type = 1 + txFinalizedTxidRecType tlv.Type = 1 ) // StartTrigger identifies what caused the unroll actor to start. @@ -129,6 +136,14 @@ const ( // PhaseFailed indicates the actor reached terminal failure. PhaseFailed Phase = "failed" + + // PhaseExternalSpendObserved indicates the actor has observed an + // external spend of the target outpoint that has not yet been + // finalized past the backend's reorg-safety depth. The actor is + // parked: it does not advance toward a sweep, but it has not + // terminated either, since a reorg of the spending block can + // resurrect the recovery job. + PhaseExternalSpendObserved Phase = "external_spend_observed" ) // Msg is the durable mailbox surface accepted by the VTXO unroll actor. @@ -736,6 +751,258 @@ func newCodec() *actor.MessageCodec { spendObservedMsgTLVType, func() actor.TLVMessage { return &SpendObservedMsg{} }, ) + codec.MustRegister( + txReorgedMsgTLVType, + func() actor.TLVMessage { return &TxReorgedMsg{} }, + ) + codec.MustRegister( + txFinalizedMsgTLVType, + func() actor.TLVMessage { return &TxFinalizedMsg{} }, + ) + codec.MustRegister( + spendReorgedMsgTLVType, + func() actor.TLVMessage { return &SpendReorgedMsg{} }, + ) + codec.MustRegister( + spendFinalizedMsgTLVType, + func() actor.TLVMessage { return &SpendFinalizedMsg{} }, + ) return codec } + +// SpendReorgedMsg reports that a previously delivered SpendObservedMsg +// for the target outpoint has been reorged out of the canonical chain. +// The actor must roll back any provisional external-spend state it +// recorded for the prior observation; if a new spend on the new tip +// follows, it arrives as a fresh SpendObservedMsg on the same +// subscription. +// +// The payload is intentionally empty: each unroll actor watches +// exactly one target outpoint, and the chainsource sub-actor is +// already keyed on that outpoint, so no additional correlation +// metadata is needed. +type SpendReorgedMsg struct { + actor.BaseMessage +} + +// MessageType returns the stable message type identifier. +func (m *SpendReorgedMsg) MessageType() string { + return "SpendReorgedMsg" +} + +// TLVType returns the durable mailbox type ID. +func (m *SpendReorgedMsg) TLVType() tlv.Type { + return spendReorgedMsgTLVType +} + +// Priority returns the durable mailbox priority for spend reorg events. +func (m *SpendReorgedMsg) Priority() int { + return unrollProgressPriority +} + +// Encode serializes the message as a (currently empty) TLV stream. +func (m *SpendReorgedMsg) Encode(w io.Writer) error { + stream, err := tlv.NewStream() + if err != nil { + return fmt.Errorf("create stream: %w", err) + } + + return stream.Encode(w) +} + +// Decode deserializes the message from a TLV stream. +func (m *SpendReorgedMsg) Decode(r io.Reader) error { + stream, err := tlv.NewStream() + if err != nil { + return fmt.Errorf("create stream: %w", err) + } + + if err := stream.Decode(r); err != nil { + return fmt.Errorf("decode: %w", err) + } + + return nil +} + +// unrollMsgSealed seals SpendReorgedMsg into the message surface. +func (m *SpendReorgedMsg) unrollMsgSealed() {} + +// SpendFinalizedMsg reports that a previously observed external spend +// of the target outpoint is past the backend's reorg-safety depth. +// Receiving it converts any provisional external-spend block on the +// actor into a terminal failure. +type SpendFinalizedMsg struct { + actor.BaseMessage +} + +// MessageType returns the stable message type identifier. +func (m *SpendFinalizedMsg) MessageType() string { + return "SpendFinalizedMsg" +} + +// TLVType returns the durable mailbox type ID. +func (m *SpendFinalizedMsg) TLVType() tlv.Type { + return spendFinalizedMsgTLVType +} + +// Priority returns the durable mailbox priority for spend finality +// events. +func (m *SpendFinalizedMsg) Priority() int { + return unrollProgressPriority +} + +// Encode serializes the message as a (currently empty) TLV stream. +func (m *SpendFinalizedMsg) Encode(w io.Writer) error { + stream, err := tlv.NewStream() + if err != nil { + return fmt.Errorf("create stream: %w", err) + } + + return stream.Encode(w) +} + +// Decode deserializes the message from a TLV stream. +func (m *SpendFinalizedMsg) Decode(r io.Reader) error { + stream, err := tlv.NewStream() + if err != nil { + return fmt.Errorf("create stream: %w", err) + } + + if err := stream.Decode(r); err != nil { + return fmt.Errorf("decode: %w", err) + } + + return nil +} + +// unrollMsgSealed seals SpendFinalizedMsg into the message surface. +func (m *SpendFinalizedMsg) unrollMsgSealed() {} + +// TxReorgedMsg reports that a previously delivered TxConfirmedMsg has +// been reorged out of the canonical chain. Subscribers should treat the +// prior confirmation as no longer valid; if the transaction re-confirms, +// a fresh TxConfirmedMsg will follow on the same subscription. +type TxReorgedMsg struct { + actor.BaseMessage + + // Txid identifies the reorged transaction. + Txid chainhash.Hash +} + +// MessageType returns the stable message type identifier. +func (m *TxReorgedMsg) MessageType() string { + return "TxReorgedMsg" +} + +// TLVType returns the durable mailbox type ID. +func (m *TxReorgedMsg) TLVType() tlv.Type { + return txReorgedMsgTLVType +} + +// Priority returns the durable mailbox priority for reorg events. Reorg +// rollback must run before low-value reads and height ticks because +// downstream side effects (e.g. sweep gating) gate on its outcome. +func (m *TxReorgedMsg) Priority() int { + return unrollProgressPriority +} + +// Encode serializes the message as a TLV stream. +func (m *TxReorgedMsg) Encode(w io.Writer) error { + txid := [32]byte(m.Txid) + + stream, err := tlv.NewStream( + tlv.MakePrimitiveRecord(txReorgedTxidRecType, &txid), + ) + if err != nil { + return fmt.Errorf("create stream: %w", err) + } + + return stream.Encode(w) +} + +// Decode deserializes the message from a TLV stream. +func (m *TxReorgedMsg) Decode(r io.Reader) error { + var txid [32]byte + + stream, err := tlv.NewStream( + tlv.MakePrimitiveRecord(txReorgedTxidRecType, &txid), + ) + if err != nil { + return fmt.Errorf("create stream: %w", err) + } + + if err := stream.Decode(r); err != nil { + return fmt.Errorf("decode: %w", err) + } + + m.Txid = chainhash.Hash(txid) + + return nil +} + +// unrollMsgSealed seals TxReorgedMsg into the message surface. +func (m *TxReorgedMsg) unrollMsgSealed() {} + +// TxFinalizedMsg reports that a tracked transaction is past the +// backend's reorg-safety depth and will receive no further events. +// Consumers may use this signal to drop any reorg-recovery bookkeeping +// they were holding for the watch. +type TxFinalizedMsg struct { + actor.BaseMessage + + // Txid identifies the finalized transaction. + Txid chainhash.Hash +} + +// MessageType returns the stable message type identifier. +func (m *TxFinalizedMsg) MessageType() string { + return "TxFinalizedMsg" +} + +// TLVType returns the durable mailbox type ID. +func (m *TxFinalizedMsg) TLVType() tlv.Type { + return txFinalizedMsgTLVType +} + +// Priority returns the durable mailbox priority for finality events. +func (m *TxFinalizedMsg) Priority() int { + return unrollProgressPriority +} + +// Encode serializes the message as a TLV stream. +func (m *TxFinalizedMsg) Encode(w io.Writer) error { + txid := [32]byte(m.Txid) + + stream, err := tlv.NewStream( + tlv.MakePrimitiveRecord(txFinalizedTxidRecType, &txid), + ) + if err != nil { + return fmt.Errorf("create stream: %w", err) + } + + return stream.Encode(w) +} + +// Decode deserializes the message from a TLV stream. +func (m *TxFinalizedMsg) Decode(r io.Reader) error { + var txid [32]byte + + stream, err := tlv.NewStream( + tlv.MakePrimitiveRecord(txFinalizedTxidRecType, &txid), + ) + if err != nil { + return fmt.Errorf("create stream: %w", err) + } + + if err := stream.Decode(r); err != nil { + return fmt.Errorf("decode: %w", err) + } + + m.Txid = chainhash.Hash(txid) + + return nil +} + +// unrollMsgSealed seals TxFinalizedMsg into the message surface. +func (m *TxFinalizedMsg) unrollMsgSealed() {} diff --git a/unroll/messages_test.go b/unroll/messages_test.go index 097e39204..e4321bcb4 100644 --- a/unroll/messages_test.go +++ b/unroll/messages_test.go @@ -127,6 +127,56 @@ func TestDurableMessageTLVRoundTrip(t *testing.T) { require.Equal(t, orig.SpendingHeight, got.SpendingHeight) }) + t.Run("TxReorgedMsg", func(t *testing.T) { + t.Parallel() + + orig := &TxReorgedMsg{Txid: txid} + + var buf bytes.Buffer + require.NoError(t, orig.Encode(&buf)) + + got := &TxReorgedMsg{} + require.NoError(t, got.Decode(bytes.NewReader(buf.Bytes()))) + require.Equal(t, orig.Txid, got.Txid) + }) + + t.Run("SpendReorgedMsg", func(t *testing.T) { + t.Parallel() + + orig := &SpendReorgedMsg{} + + var buf bytes.Buffer + require.NoError(t, orig.Encode(&buf)) + + got := &SpendReorgedMsg{} + require.NoError(t, got.Decode(bytes.NewReader(buf.Bytes()))) + }) + + t.Run("SpendFinalizedMsg", func(t *testing.T) { + t.Parallel() + + orig := &SpendFinalizedMsg{} + + var buf bytes.Buffer + require.NoError(t, orig.Encode(&buf)) + + got := &SpendFinalizedMsg{} + require.NoError(t, got.Decode(bytes.NewReader(buf.Bytes()))) + }) + + t.Run("TxFinalizedMsg", func(t *testing.T) { + t.Parallel() + + orig := &TxFinalizedMsg{Txid: txid} + + var buf bytes.Buffer + require.NoError(t, orig.Encode(&buf)) + + got := &TxFinalizedMsg{} + require.NoError(t, got.Decode(bytes.NewReader(buf.Bytes()))) + require.Equal(t, orig.Txid, got.Txid) + }) + t.Run("GetStateRequest", func(t *testing.T) { t.Parallel() diff --git a/unroll/reconcile.go b/unroll/reconcile.go new file mode 100644 index 000000000..adbc3d213 --- /dev/null +++ b/unroll/reconcile.go @@ -0,0 +1,302 @@ +package unroll + +import ( + "context" + "fmt" + + "github.com/btcsuite/btcd/chainhash/v2" + "github.com/btcsuite/btcd/wire/v2" + "github.com/lightninglabs/darepo-client/lib/recovery" + "github.com/lightninglabs/darepo-client/unrollplan" + fn "github.com/lightningnetwork/lnd/fn/v2" +) + +// ConfirmedAnchor describes a tx confirmation that the backend reports +// is currently live on the canonical chain. +type ConfirmedAnchor struct { + // Txid is the confirmed transaction hash. + Txid chainhash.Hash + + // Height is the block height the transaction confirmed at. + Height int32 +} + +// SpendAnchor describes an outpoint spend that the backend reports is +// currently live on the canonical chain. +type SpendAnchor struct { + // Outpoint is the spent outpoint. + Outpoint wire.OutPoint + + // SpendingTxid is the transaction that consumed the outpoint. + SpendingTxid chainhash.Hash + + // SpendingHeight is the block height the spending tx confirmed at. + SpendingHeight int32 +} + +// ChainReconciler is the narrow interface the unroll actor uses on +// restart to verify that the chain anchors recorded in its checkpoint +// are still live on the canonical chain. +// +// The interface is intentionally small: it answers two questions, both +// keyed on data the actor already has in its checkpoint. Implementations +// query the backend (lnd's chainntnfs, neutrino, or a bitcoind RPC) and +// return fn.None for "not on chain anymore" so the unroll layer can +// stay backend-agnostic. +// +// Both methods return errors for transport-level failures the caller +// should retry; an "anchor is not on chain" answer is conveyed via +// fn.None, not via error. +type ChainReconciler interface { + // ConfirmedTx reports the current confirmation status of a tx on + // the canonical chain. fn.None means the tx is no longer + // confirmed (typically because the block was reorged out while + // the daemon was offline). + ConfirmedTx(ctx context.Context, + txid chainhash.Hash) (fn.Option[ConfirmedAnchor], error) + + // SpentOutpoint reports the current spend status of an outpoint + // on the canonical chain. fn.None means the outpoint is currently + // unspent. + SpentOutpoint(ctx context.Context, + outpoint wire.OutPoint) (fn.Option[SpendAnchor], error) +} + +// ChainReconcilerFactory builds a ChainReconciler bound to a specific +// per-target unroll actor. Per-target actors invoke the factory after +// loading their proof; the target outpoint identifies the calling +// actor and lets the implementation namespace its chainsource probe +// caller IDs so two actors that share a proof-graph ancestor (the +// common case for sibling VTXOs) cannot collide on chainsource +// service keys when they reconcile concurrently. The proof argument +// supplies output scripts because lnd's tx index is not always +// available, so the (txid, pkScript) pair is the canonical way to +// identify a tx during a historical block scan. Test stubs that do +// not need either argument can ignore them. +type ChainReconcilerFactory func(target wire.OutPoint, + proof *recovery.Proof) ChainReconciler + +// reconcileCheckpoint walks the persisted checkpoint and prunes any +// chain anchor that the reconciler reports is no longer live on the +// canonical chain. The checkpoint is mutated in place. +// +// The reconciliation order mirrors applyReorgedEvent: +// +// - For every confirmed proof-graph txid, ask the reconciler whether +// it is still confirmed. Anchors that are not are dropped, and so +// are their descendants in the proof graph (State.Validate's +// topological invariant requires every confirmed/in-flight node to +// have confirmed parents). +// - If the target tx was confirmed but is now absent, clear +// TargetConfirmHeight and downgrade the sweep to Pending — same +// gate the live reorg reducer enforces. +// - If a Broadcasted / Confirmed sweep tx is no longer on chain, +// reset the sweep state so the actor re-broadcasts on its next +// planning step. +// - Reconcile the target outpoint spend status against any persisted +// ProvisionalExternalSpend anchor: set, refresh, or clear it so +// the restored FSM enters the same parked state a live +// SpendObservedEvent / SpendReorgedEvent sequence would have +// produced. +// +// On any reconciler error we surface it untouched: the actor must not +// proceed to broadcast off a checkpoint we could not verify. +func reconcileCheckpoint(ctx context.Context, reconciler ChainReconciler, + proof *recovery.Proof, checkpoint *actorCheckpoint) error { + + if reconciler == nil || checkpoint == nil || proof == nil { + return nil + } + + // Snapshot the input list so the loop iterates over a stable + // slice while we mutate the live ConfirmedTxids field. + confirmedIn := append( + []chainhash.Hash(nil), checkpoint.State.ConfirmedTxids..., + ) + + for _, txid := range confirmedIn { + anchor, err := reconciler.ConfirmedTx(ctx, txid) + if err != nil { + return fmt.Errorf("reconcile confirmed tx %s: %w", txid, + err) + } + if anchor.IsSome() { + // The tx is still confirmed. Refresh the + // target-derived height in case the tx was re-mined + // at a different block height while the daemon was + // offline; leaving the stale height in place would + // make CSV maturity look closer than it actually is + // and could drive a premature sweep on the next + // planning step. Also raise the checkpoint's + // best-height watermark to at least the live anchor + // height so derived state is internally consistent. + liveHeight := anchor.UnsafeFromSome().Height + if txid == proof.TargetOutpoint().Hash { + checkpoint.State.TargetConfirmHeight = fn.Some( + liveHeight, + ) + } + if liveHeight > checkpoint.Height { + checkpoint.Height = liveHeight + } + + continue + } + + pruneReorgedSubtree(proof, &checkpoint.State, txid) + + if txid == proof.TargetOutpoint().Hash { + downgradeSweepOnTargetLoss(&checkpoint.State) + } + } + + // If a sweep was already broadcast or confirmed, verify that it + // is still on chain. A sweep that vanished while the daemon was + // offline must be re-broadcastable; resetting Status to Pending + // triggers a fresh RequestSweepBuild on the next derivation, and + // the cached b.sweepTx (restored from the checkpoint at boot) is + // reused so the txid stays stable. + if checkpoint.State.Sweep.Txid.IsSome() && + checkpoint.State.Sweep.Status != + unrollplan.SweepStatusPending { + + sweepTxid := checkpoint.State.Sweep.Txid.UnsafeFromSome() + anchor, err := reconciler.ConfirmedTx(ctx, sweepTxid) + if err != nil { + return fmt.Errorf("reconcile sweep tx %s: %w", + sweepTxid, err) + } + switch { + case anchor.IsNone(): + // Sweep is no longer on chain. Reset so the next + // derivation re-broadcasts using the cached bytes. + checkpoint.State.Sweep.Status = + unrollplan.SweepStatusPending + checkpoint.State.Sweep.Txid = + fn.None[chainhash.Hash]() + checkpoint.State.Sweep.ConfirmHeight = + fn.None[int32]() + + case checkpoint.State.Sweep.Status == + unrollplan.SweepStatusConfirmed: + + // Sweep is still on chain but at potentially a new + // height. Recompute ConfirmHeight from the live + // answer so a sweep that confirmed in a different + // block is not stuck reporting a stale height. + checkpoint.State.Sweep.ConfirmHeight = fn.Some( + anchor.UnsafeFromSome().Height, + ) + } + } + + if err := reconcileExternalSpend( + ctx, reconciler, proof, checkpoint, + ); err != nil { + return err + } + + return nil +} + +// reconcileExternalSpend cross-checks the target outpoint's spend +// status on the canonical chain against any ProvisionalExternalSpend +// anchor the checkpoint carries. Four cases land here: +// +// 1. Chain says target spent + checkpoint anchor matches the same +// spender txid: no-op (refresh the height in case it shifted to a +// re-organized block). +// 2. Chain says target spent + checkpoint anchor for a DIFFERENT +// spender: update the anchor to the live spender. +// 3. Chain says target spent + no checkpoint anchor: install one. +// This is the "external spend confirmed while the daemon was +// offline" path. +// 4. Chain says target unspent + checkpoint anchor present: clear +// the anchor. The spending block was reorged out during downtime. +// +// In cases 2/3, "spent by a proof-graph node" or "spent by our own +// sweep" are treated as benign: those are the same classifications +// the live spend-watch handler runs in handleSpendObserved. +func reconcileExternalSpend(ctx context.Context, reconciler ChainReconciler, + proof *recovery.Proof, checkpoint *actorCheckpoint) error { + + targetOutpoint := proof.TargetOutpoint() + anchor, err := reconciler.SpentOutpoint(ctx, targetOutpoint) + if err != nil { + return fmt.Errorf("reconcile target spend %s: %w", + targetOutpoint, err) + } + + if anchor.IsNone() { + // Case 4: chain reports target unspent. Any persisted + // anchor was reorged out while we were offline; clear it + // so the FSM does NOT park in + // AwaitingExternalSpendFinality on restore. + checkpoint.ProvisionalExternalSpend = + fn.None[ExternalSpendAnchor]() + + return nil + } + + live := anchor.UnsafeFromSome() + + // Skip benign spenders that the live spend-watch path would also + // classify as "expected materialization traffic": a proof-graph + // node, or our own sweep. + if proof != nil { + if _, ok := proof.Node(live.SpendingTxid); ok { + checkpoint.ProvisionalExternalSpend = + fn.None[ExternalSpendAnchor]() + + return nil + } + } + if checkpoint.State.Sweep.Txid.IsSome() && + checkpoint.State.Sweep.Txid.UnsafeFromSome() == + live.SpendingTxid { + + checkpoint.ProvisionalExternalSpend = + fn.None[ExternalSpendAnchor]() + + return nil + } + + // Cases 1, 2, 3: install or refresh the anchor with the live + // spender so the restored FSM enters AwaitingExternalSpendFinality + // rather than blindly resuming planner-driven materialization. + checkpoint.ProvisionalExternalSpend = fn.Some(ExternalSpendAnchor{ + SpendingTxid: live.SpendingTxid, + SpendingHeight: live.SpendingHeight, + }) + + return nil +} + +// pruneReorgedSubtree drops a txid and every transitive descendant from +// both ConfirmedTxids and InFlightTxids, and clears any deferred +// checkpoints referencing the subtree. +func pruneReorgedSubtree(proof *recovery.Proof, state *unrollplan.State, + root chainhash.Hash) { + + subtree := collectReorgedSubtree( + &Environment{Proof: proof}, root, + ) + state.ConfirmedTxids = removeHashes(state.ConfirmedTxids, subtree) + state.InFlightTxids = removeHashes(state.InFlightTxids, subtree) +} + +// downgradeSweepOnTargetLoss resets the target-derived planner state +// when the target tx is no longer on the canonical chain. Mirrors the +// "target reorg" branch of applyReorgedEvent so restart reconciliation +// and live rollback converge on identical post-conditions. +func downgradeSweepOnTargetLoss(state *unrollplan.State) { + state.TargetConfirmHeight = fn.None[int32]() + + if state.Sweep.Status == unrollplan.SweepStatusPending { + return + } + + state.Sweep.Status = unrollplan.SweepStatusPending + state.Sweep.Txid = fn.None[chainhash.Hash]() + state.Sweep.ConfirmHeight = fn.None[int32]() +} diff --git a/unroll/reconcile_chainsource.go b/unroll/reconcile_chainsource.go new file mode 100644 index 000000000..4be3b403a --- /dev/null +++ b/unroll/reconcile_chainsource.go @@ -0,0 +1,318 @@ +package unroll + +import ( + "context" + "errors" + "fmt" + "log/slog" + "time" + + "github.com/btcsuite/btcd/chainhash/v2" + "github.com/btcsuite/btcd/wire/v2" + "github.com/btcsuite/btclog/v2" + "github.com/lightninglabs/darepo-client/baselib/actor" + "github.com/lightninglabs/darepo-client/chainsource" + "github.com/lightninglabs/darepo-client/lib/recovery" + fn "github.com/lightningnetwork/lnd/fn/v2" +) + +// defaultReconcileProbeTimeout bounds how long the chainsource-backed +// reconciler waits for a confirmation or spend event before treating +// the probed anchor as no longer live. lnd's chainntnfs dispatches +// historical confirmation events immediately when the tx is in chain +// and the height-hint is fresh; legitimate "still confirmed" answers +// return well under a second on a healthy node, so a 10-second +// budget leaves comfortable headroom for slow restarts while still +// failing fast on truly absent anchors. +const defaultReconcileProbeTimeout = 10 * time.Second + +// ChainSourceReconcilerConfig configures NewChainSourceReconciler. +type ChainSourceReconcilerConfig struct { + // ChainSource is the actor ref used to issue RegisterConf / + // RegisterSpend probes. + ChainSource actor.ActorRef[ + chainsource.ChainSourceMsg, chainsource.ChainSourceResp, + ] + + // Proof is the immutable recovery graph the reconciler queries + // pkScripts from. Required. + Proof *recovery.Proof + + // CallerID prefix used when registering probes; the reconciler + // appends per-probe suffixes so chainsource sees a unique key + // per (caller, txid / outpoint, confs) tuple. + // + // Callers that share the underlying ChainSource between multiple + // reconcilers MUST give each instance a unique prefix (typically + // by including the per-actor target identity) so concurrent + // probes of the same shared proof-graph txid do not collide on + // the chainsource service-key namespace. + CallerID string + + // ProbeTimeout bounds each individual confirmation / spend + // probe. Zero falls back to defaultReconcileProbeTimeout. + // Slow chainsource backends can legitimately need a higher + // budget here: a probe that times out is treated as "not on + // chain", which on restart triggers a conservative + // re-broadcast / planner rollback. Operators running against + // a slow backend should raise this to avoid spurious + // rollbacks; see the timeout warn log emitted on miss for the + // signal to watch for. + ProbeTimeout time.Duration + + // CleanupTimeout bounds the best-effort unregister send that + // runs when a probe is interrupted before its future fires. + // Zero falls back to 5 seconds. + CleanupTimeout time.Duration + + // Log is an optional logger. Timeout-as-absent answers are + // surfaced at warn level here so a slow backend masquerading + // as a reorged-out anchor is visible in production logs + // instead of silently driving conservative rollbacks. Zero + // falls back to btclog.Disabled. + Log fn.Option[btclog.Logger] +} + +// chainSourceReconciler is a ChainReconciler that answers queries by +// probing the chainsource actor with short-timeout RegisterConf and +// RegisterSpend requests in future mode. +// +// The implementation is intentionally simple: it does NOT consume a +// dedicated "is this tx in chain right now" API (chainsource does not +// expose one today). Instead it leans on lnd's chainntnfs behavior of +// dispatching a historical confirmation immediately when the watched +// tx is currently on chain. A timed-out probe is treated as "not on +// chain" — conservative for restart reconciliation, which is the only +// caller, because the worst-case false-negative is a re-broadcast of +// an already-confirmed proof node (idempotent thanks to txconfirm's +// txid-keyed dedup). +// +// Probes that complete on their own self-clean (the chainsource +// sub-actor exits after delivering the single positive event in +// future mode); probes that time out enqueue a best-effort +// UnregisterConfRequest on a fresh background context so the +// long-lived chainsource sub-actor does not leak per restart. +type chainSourceReconciler struct { + chainSource actor.ActorRef[ + chainsource.ChainSourceMsg, chainsource.ChainSourceResp, + ] + proof *recovery.Proof + callerID string + probeTimeout time.Duration + cleanupTimeout time.Duration + log btclog.Logger +} + +// NewChainSourceReconciler constructs a chainsource-backed +// ChainReconciler. The proof is required because chainsource probes +// rely on (txid, pkScript) pairs for historical scans. +func NewChainSourceReconciler(cfg ChainSourceReconcilerConfig) ChainReconciler { + if cfg.Proof == nil { + return nil + } + probeTimeout := cfg.ProbeTimeout + if probeTimeout <= 0 { + probeTimeout = defaultReconcileProbeTimeout + } + cleanupTimeout := cfg.CleanupTimeout + if cleanupTimeout <= 0 { + cleanupTimeout = 5 * time.Second + } + + return &chainSourceReconciler{ + chainSource: cfg.ChainSource, + proof: cfg.Proof, + callerID: cfg.CallerID, + probeTimeout: probeTimeout, + cleanupTimeout: cleanupTimeout, + log: cfg.Log.UnwrapOr(btclog.Disabled), + } +} + +// ConfirmedTx probes chainsource for the current confirmation status +// of txid. The probe registers a one-shot conf watch in future mode +// using the txid + first-output pkScript from the proof, then awaits +// the future with a bounded timeout. +func (r *chainSourceReconciler) ConfirmedTx(ctx context.Context, + txid chainhash.Hash) (fn.Option[ConfirmedAnchor], error) { + + node, ok := r.proof.Node(txid) + if !ok || node == nil || node.Tx == nil || + len(node.Tx.TxOut) == 0 { + // Without an output script we cannot drive a historical + // scan; treat as "unknown / not on chain" so reconciliation + // stays conservative. + return fn.None[ConfirmedAnchor](), nil + } + pkScript := append([]byte(nil), node.Tx.TxOut[0].PkScript...) + + callerID := fmt.Sprintf("%s-conf-%s", r.callerID, txid) + probeCtx, cancel := context.WithTimeout(ctx, r.probeTimeout) + defer cancel() + + resp, err := r.chainSource.Ask( + probeCtx, &chainsource.RegisterConfRequest{ + CallerID: callerID, + Txid: &txid, + PkScript: pkScript, + TargetConfs: 1, + }, + ).Await(probeCtx).Unpack() + if err != nil { + return fn.None[ConfirmedAnchor](), fmt.Errorf("register conf "+ + "probe %s: %w", txid, err) + } + + confResp, ok := resp.(*chainsource.RegisterConfResponse) + if !ok || confResp.Future == nil { + return fn.None[ConfirmedAnchor](), fmt.Errorf("register conf "+ + "probe %s: unexpected response %T", txid, resp) + } + + // Schedule a best-effort cleanup unregister in case the probe + // times out: the chainsource sub-actor would otherwise stay + // alive watching for a confirmation that, by assumption, is + // never coming. The cleanup runs on a fresh background context + // so it executes even when the probe context was cancelled. + probeTxid := txid + probePkScript := pkScript + //nolint:contextcheck // cleanup intentionally uses its own context + defer r.cleanupConfWatch(callerID, &probeTxid, probePkScript) + + event, err := confResp.Future.Await(probeCtx).Unpack() + if err != nil { + if errors.Is(err, context.DeadlineExceeded) || + errors.Is(err, context.Canceled) { + // Timeout-as-absent is a conservative answer: the + // reconciler tells the caller "not on chain" and + // any persisted anchor will be rolled back. That is + // safe but expensive when the real cause is a slow + // backend rather than a vanished tx, so surface it + // at warn level to make slow-backend regressions + // visible without changing planner behavior. + r.log.WarnS(ctx, "Reconciler conf probe timed out; "+ + "treating tx as not on chain", err, + slog.String("txid", txid.String()), + slog.Duration("timeout", r.probeTimeout), + ) + + return fn.None[ConfirmedAnchor](), nil + } + + return fn.None[ConfirmedAnchor](), fmt.Errorf("await conf "+ + "probe %s: %w", txid, err) + } + + return fn.Some(ConfirmedAnchor{ + Txid: event.Txid, + Height: event.BlockHeight, + }), nil +} + +// SpentOutpoint probes chainsource for the current spend status of +// outpoint. Symmetric to ConfirmedTx: registers a future-mode spend +// watch with the outpoint's pkScript and awaits the future with a +// bounded timeout, scheduling an UnregisterSpendRequest cleanup on +// the way out. +func (r *chainSourceReconciler) SpentOutpoint(ctx context.Context, + outpoint wire.OutPoint) (fn.Option[SpendAnchor], error) { + + node, ok := r.proof.Node(outpoint.Hash) + if !ok || node == nil || node.Tx == nil { + return fn.None[SpendAnchor](), nil + } + if int(outpoint.Index) >= len(node.Tx.TxOut) { + return fn.None[SpendAnchor](), nil + } + pkScript := append( + []byte(nil), node.Tx.TxOut[outpoint.Index].PkScript..., + ) + + callerID := fmt.Sprintf("%s-spend-%s", r.callerID, outpoint) + probeCtx, cancel := context.WithTimeout(ctx, r.probeTimeout) + defer cancel() + + probeOutpoint := outpoint + resp, err := r.chainSource.Ask( + probeCtx, &chainsource.RegisterSpendRequest{ + CallerID: callerID, + Outpoint: &probeOutpoint, + PkScript: pkScript, + }, + ).Await(probeCtx).Unpack() + if err != nil { + return fn.None[SpendAnchor](), fmt.Errorf("register spend "+ + "probe %s: %w", outpoint, err) + } + + spendResp, ok := resp.(*chainsource.RegisterSpendResponse) + if !ok || spendResp.Future == nil { + return fn.None[SpendAnchor](), fmt.Errorf("register spend "+ + "probe %s: unexpected response %T", outpoint, resp) + } + + //nolint:contextcheck // cleanup intentionally uses its own context + defer r.cleanupSpendWatch(callerID, &probeOutpoint) + + event, err := spendResp.Future.Await(probeCtx).Unpack() + if err != nil { + if errors.Is(err, context.DeadlineExceeded) || + errors.Is(err, context.Canceled) { + r.log.WarnS(ctx, "Reconciler spend probe timed "+ + "out; treating outpoint as unspent", err, + slog.String("outpoint", outpoint.String()), + slog.Duration("timeout", r.probeTimeout), + ) + + return fn.None[SpendAnchor](), nil + } + + return fn.None[SpendAnchor](), fmt.Errorf("await spend probe "+ + "%s: %w", outpoint, err) + } + + return fn.Some(SpendAnchor{ + Outpoint: event.Outpoint, + SpendingTxid: event.SpendingTxid, + SpendingHeight: event.SpendingHeight, + }), nil +} + +// cleanupConfWatch sends a best-effort UnregisterConfRequest on a +// fresh background context so a timed-out probe does not leak the +// underlying chainsource sub-actor. Errors are intentionally ignored +// because the cleanup is purely a hygiene operation. +func (r *chainSourceReconciler) cleanupConfWatch(callerID string, + txid *chainhash.Hash, pkScript []byte) { + + cleanupCtx, cancel := context.WithTimeout( + context.Background(), r.cleanupTimeout, + ) + defer cancel() + + _, _ = r.chainSource.Ask( + cleanupCtx, &chainsource.UnregisterConfRequest{ + CallerID: callerID, + Txid: txid, + PkScript: pkScript, + TargetConfs: 1, + }, + ).Await(cleanupCtx).Unpack() +} + +// cleanupSpendWatch is the spend-side analogue of cleanupConfWatch. +func (r *chainSourceReconciler) cleanupSpendWatch(callerID string, + outpoint *wire.OutPoint) { + + cleanupCtx, cancel := context.WithTimeout( + context.Background(), r.cleanupTimeout, + ) + defer cancel() + + _, _ = r.chainSource.Ask( + cleanupCtx, &chainsource.UnregisterSpendRequest{ + CallerID: callerID, + Outpoint: outpoint, + }, + ).Await(cleanupCtx).Unpack() +} diff --git a/unroll/reconcile_chainsource_test.go b/unroll/reconcile_chainsource_test.go new file mode 100644 index 000000000..dee557b96 --- /dev/null +++ b/unroll/reconcile_chainsource_test.go @@ -0,0 +1,272 @@ +package unroll + +import ( + "context" + "fmt" + "sync" + "sync/atomic" + "testing" + "time" + + "github.com/btcsuite/btcd/btcutil/v2" + "github.com/btcsuite/btcd/chainhash/v2" + "github.com/btcsuite/btcd/wire/v2" + "github.com/lightninglabs/darepo-client/baselib/actor" + "github.com/lightninglabs/darepo-client/chainsource" + fn "github.com/lightningnetwork/lnd/fn/v2" + "github.com/stretchr/testify/require" +) + +// reconcileMockBackend is a minimal ChainBackend test double tailored +// for the reconciler concurrency test. It records every RegisterConf +// call so the test can assert per-probe caller-ID uniqueness, and +// pre-arms a positive confirmation reply that fires immediately when +// the txid matches the configured target. +type reconcileMockBackend struct { + confirmTxid chainhash.Hash + confirmHeight int32 + + // registers records (callerID-like txid, pkScript) pairs as seen + // from successive RegisterConf calls. The caller-ID is not visible + // at this layer (chainsource consumes it before spawning the sub- + // actor), so the test instead asserts that the sub-actor IDs the + // system spawned are distinct via System.Has on the service-key. + mu sync.Mutex + registers []struct { + Txid chainhash.Hash + PkScript []byte + } + + // bestHeight is reported by BestBlock. + bestHeight int32 + + // epochCh / epochCancel back RegisterBlocks. + epochCh chan *chainsource.BlockEpoch + epochCancel atomic.Int32 +} + +func newReconcileMockBackend() *reconcileMockBackend { + return &reconcileMockBackend{ + bestHeight: 200, + epochCh: make(chan *chainsource.BlockEpoch, 1), + } +} + +func (m *reconcileMockBackend) EstimateFee(context.Context, uint32) ( + btcutil.Amount, error) { + + return 1000, nil +} + +func (m *reconcileMockBackend) BestBlock(context.Context) (int32, + chainhash.Hash, error) { + + return m.bestHeight, chainhash.Hash{}, nil +} + +func (m *reconcileMockBackend) TestMempoolAccept(context.Context, + ...*wire.MsgTx) ([]chainsource.MempoolAcceptResult, error) { + + return nil, nil +} + +func (m *reconcileMockBackend) BroadcastTx(context.Context, *wire.MsgTx, + string) error { + + return nil +} + +func (m *reconcileMockBackend) SubmitPackage(context.Context, []*wire.MsgTx, + *wire.MsgTx) error { + + return nil +} + +// RegisterConf records the registration and arms a one-shot positive +// confirmation when the requested txid matches the configured target. +// Every call returns a fresh ConfRegistration with its own channels so +// concurrent registrants do not contend on a shared buffered channel. +func (m *reconcileMockBackend) RegisterConf(_ context.Context, + txid *chainhash.Hash, pkScript []byte, _, _ uint32, _ bool) ( + *chainsource.ConfRegistration, error) { + + m.mu.Lock() + m.registers = append(m.registers, struct { + Txid chainhash.Hash + PkScript []byte + }{ + Txid: *txid, + PkScript: append([]byte(nil), pkScript...), + }) + m.mu.Unlock() + + confCh := make(chan *chainsource.TxConfirmation, 1) + reorged := make(chan uint64, 1) + done := make(chan struct{}, 1) + + if txid != nil && *txid == m.confirmTxid { + blockTx := wire.NewMsgTx(2) + blockHash := chainhash.Hash{0x42} + confCh <- &chainsource.TxConfirmation{ + BlockHash: &blockHash, + BlockHeight: uint32(m.confirmHeight), + Tx: blockTx, + } + } + + return &chainsource.ConfRegistration{ + Confirmed: confCh, + Reorged: reorged, + Done: done, + Cancel: func() {}, + }, nil +} + +func (m *reconcileMockBackend) RegisterSpend(context.Context, *wire.OutPoint, + []byte, uint32) (*chainsource.SpendRegistration, error) { + + return &chainsource.SpendRegistration{ + Spend: make(chan *chainsource.SpendDetail, 1), + Reorged: make(chan uint64, 1), + Done: make(chan struct{}, 1), + Cancel: func() {}, + }, nil +} + +func (m *reconcileMockBackend) RegisterBlocks(context.Context) ( + *chainsource.BlockRegistration, error) { + + return &chainsource.BlockRegistration{ + Epochs: m.epochCh, + Cancel: func() { + m.epochCancel.Add(1) + }, + }, nil +} + +func (m *reconcileMockBackend) Start() error { return nil } +func (m *reconcileMockBackend) Stop() error { return nil } + +// TestChainSourceReconcilerConcurrentProbesDoNotCollide exercises the +// invariant that two reconcilers probing the same shared proof-graph +// txid against a single chainsource actor do NOT collide on chainsource +// service keys when each reconciler is built with a target-specific +// caller-ID prefix. +// +// Chainsource keys its per-probe sub-actors on +// (CallerID, txid/pkScript, TargetConfs) (see handleRegisterConf in +// chainsource/chainsource.go). Two reconcilers built with the same +// static prefix probing the same txid would map to identical service +// keys; the second Spawn would collide with the first sub-actor's +// registration and drop or merge the second probe. Production wiring +// in darepod/server.go bakes the per-actor target outpoint into the +// caller-ID prefix exactly to avoid this; this test pins that +// invariant for the chainsource-backed reconciler. +func TestChainSourceReconcilerConcurrentProbesDoNotCollide(t *testing.T) { + t.Parallel() + + proof := buildLinearProof(t) + roots := proof.RootTxids() + require.Len(t, roots, 1, "linear proof should have one root") + rootTxid := roots[0] + + backend := newReconcileMockBackend() + backend.confirmTxid = rootTxid + backend.confirmHeight = 150 + + system := actor.NewActorSystem() + defer func() { + _ = system.Shutdown(t.Context()) + }() + + chainSource := chainsource.NewChainSourceActor( + chainsource.ChainSourceConfig{ + Backend: backend, + System: system, + }, + ) + chainRef := chainsource.ChainSourceKey.Spawn( + system, "chainsource-reconcile", chainSource, + ) + + target1 := proof.TargetOutpoint() + target2 := wire.OutPoint{ + Hash: chainhash.Hash{ + 0xaa, + }, + Index: 7, + } + + mkReconciler := func(target wire.OutPoint) ChainReconciler { + return NewChainSourceReconciler(ChainSourceReconcilerConfig{ + ChainSource: chainRef, + Proof: proof, + CallerID: fmt.Sprintf( + "unroll-reconcile-%s", target, + ), + ProbeTimeout: 5 * time.Second, + }) + } + + rec1 := mkReconciler(target1) + rec2 := mkReconciler(target2) + + type probeResult struct { + anchor fn.Option[ConfirmedAnchor] + err error + } + + results := make(chan probeResult, 2) + probe := func(r ChainReconciler) { + ctx, cancel := context.WithTimeout( + t.Context(), 10*time.Second, + ) + defer cancel() + + anchor, err := r.ConfirmedTx(ctx, rootTxid) + results <- probeResult{anchor: anchor, err: err} + } + + go probe(rec1) + go probe(rec2) + + for range 2 { + select { + case res := <-results: + require.NoError( + t, res.err, + "concurrent reconciler probe failed", + ) + require.True( + t, res.anchor.IsSome(), + "reconciler did not see confirmation; a "+ + "static caller-ID collision would "+ + "silently swallow the second probe", + ) + require.Equal( + t, backend.confirmHeight, + res.anchor.UnsafeFromSome().Height, + ) + + case <-time.After(15 * time.Second): + t.Fatal( + "timed out waiting for concurrent " + + "reconciler probes; static " + + "caller-ID would cause one probe " + + "to wait forever", + ) + } + } + + // Both reconcilers must have hit the backend with a RegisterConf + // for the shared txid — confirming no internal short-circuit. + backend.mu.Lock() + defer backend.mu.Unlock() + require.Len( + t, backend.registers, 2, + "expected both reconciler probes to reach the backend", + ) + for _, r := range backend.registers { + require.Equal(t, rootTxid, r.Txid) + } +} diff --git a/unroll/reconcile_test.go b/unroll/reconcile_test.go new file mode 100644 index 000000000..27bd46c05 --- /dev/null +++ b/unroll/reconcile_test.go @@ -0,0 +1,552 @@ +package unroll + +import ( + "context" + "errors" + "testing" + "time" + + "github.com/btcsuite/btcd/chainhash/v2" + "github.com/btcsuite/btcd/wire/v2" + "github.com/btcsuite/btclog/v2" + "github.com/lightninglabs/darepo-client/baselib/actor" + "github.com/lightninglabs/darepo-client/lib/recovery" + "github.com/lightninglabs/darepo-client/unrollplan" + fn "github.com/lightningnetwork/lnd/fn/v2" + "github.com/stretchr/testify/require" +) + +// stubChainReconciler is a fully controllable ChainReconciler test +// double. The two maps drive the ConfirmedTx / SpentOutpoint answers +// and a non-nil err field lets a test inject a transport failure on +// either query. +type stubChainReconciler struct { + confirmed map[chainhash.Hash]ConfirmedAnchor + spent map[wire.OutPoint]SpendAnchor + err error +} + +// ConfirmedTx returns the configured anchor for txid, or fn.None. +func (s *stubChainReconciler) ConfirmedTx(_ context.Context, + txid chainhash.Hash) (fn.Option[ConfirmedAnchor], error) { + + if s.err != nil { + return fn.None[ConfirmedAnchor](), s.err + } + + anchor, ok := s.confirmed[txid] + if !ok { + return fn.None[ConfirmedAnchor](), nil + } + + return fn.Some(anchor), nil +} + +// SpentOutpoint returns the configured anchor for outpoint, or fn.None. +func (s *stubChainReconciler) SpentOutpoint(_ context.Context, + outpoint wire.OutPoint) (fn.Option[SpendAnchor], error) { + + if s.err != nil { + return fn.None[SpendAnchor](), s.err + } + + anchor, ok := s.spent[outpoint] + if !ok { + return fn.None[SpendAnchor](), nil + } + + return fn.Some(anchor), nil +} + +// restoreHarness boots a fresh unroll behavior from a fabricated +// checkpoint so the reconcile tests can inspect the post-reconciliation +// FSM state without running the rest of the actor's IO surface. +func restoreHarness(t *testing.T, proof *recovery.Proof, + checkpoint *actorCheckpoint, reconciler ChainReconciler) (*behavior, + *fakeTxConfirmRef, *fakeChainSourceRef) { + + t.Helper() + + desc := testDescriptor(t, proof.TargetOutpoint(), proof.CSVDelay()) + store := newMemCheckpointStore() + txconfirmRef := &fakeTxConfirmRef{} + chainRef := &fakeChainSourceRef{} + + raw, err := encodeCheckpoint(checkpoint) + require.NoError(t, err) + + err = store.SaveCheckpoint(t.Context(), actor.CheckpointParams{ + ActorID: "reconcile-test", + StateType: checkpointStateType, + StateData: raw, + Version: checkpointVersion, + }) + require.NoError(t, err) + + cfg := Config{ + TargetOutpoint: proof.TargetOutpoint(), + ActorID: "reconcile-test", + DeliveryStore: store, + ProofAssembler: &mockProofAssembler{ + proof: proof, + }, + VTXOStore: &mockVTXOStore{ + desc: desc, + }, + TxConfirmRef: txconfirmRef, + ChainSource: chainRef, + Wallet: &fakeSweepWallet{}, + Log: fn.Some(btclog.Disabled), + } + if reconciler != nil { + // Wrap the stub in a factory that ignores the proof, so + // the production code path (which always calls the + // factory) sees the test stub unchanged. + stub := reconciler + cfg.ChainReconcilerFactory = fn.Some( + ChainReconcilerFactory( + func(wire.OutPoint, + *recovery.Proof) ChainReconciler { + + return stub + }, + ), + ) + } + + beh := &behavior{ + cfg: cfg, + log: btclog.Disabled, + } + require.NoError(t, beh.restoreCheckpoint(t.Context())) + + return beh, txconfirmRef, chainRef +} + +// TestReconcileTargetReorgedOfflineResumesMaterialization restores a +// checkpoint whose target tx is recorded as confirmed but is no longer +// on the canonical chain. After reconciliation the actor must restart +// without the stale TargetConfirmHeight, in PhaseMaterializing, so it +// does not broadcast a sweep on a target that has been reorged out. +func TestReconcileTargetReorgedOfflineResumesMaterialization(t *testing.T) { + proof := buildLinearProof(t) + rootTxid := proof.RootTxids()[0] + targetTxid := proof.TargetOutpoint().Hash + + checkpoint := &actorCheckpoint{ + Version: checkpointVersion, + Height: 103, + Started: true, + Trigger: TriggerRestart, + State: unrollplan.State{ + ConfirmedTxids: []chainhash.Hash{ + rootTxid, targetTxid, + }, + TargetConfirmHeight: fn.Some[int32](102), + }, + } + + // Reconciler reports the root is still on chain at H=101 but the + // target tx is absent — its confirming block was reorged out + // while the daemon was offline. + reconciler := &stubChainReconciler{ + confirmed: map[chainhash.Hash]ConfirmedAnchor{ + rootTxid: { + Txid: rootTxid, + Height: 101, + }, + }, + } + + beh, _, _ := restoreHarness(t, proof, checkpoint, reconciler) + + actorInstance := actor.NewActor(actor.ActorConfig[Msg, Resp]{ + ID: "reconcile-test", + Behavior: &txExecAdapter{b: beh, ax: newMemExecFor(beh)}, + MailboxSize: 16, + }) + beh.selfRef = actorInstance.TellRef() + actorInstance.Start() + t.Cleanup(actorInstance.Stop) + + mustAsk(t, actorInstance.Ref(), &ResumeUnrollRequest{Height: 103}) + + require.Eventually(t, func() bool { + resp, ok := mustAsk( + t, actorInstance.Ref(), &GetStateRequest{}, + ).(*GetStateResp) + require.True(t, ok) + + // Target anchor must be gone, planner must be back in + // materialization, and no FailReason should have leaked + // through. + for _, h := range resp.PlannerState.ConfirmedTxids { + if h == targetTxid { + return false + } + } + + return resp.Phase == PhaseMaterializing && + resp.PlannerState.TargetConfirmHeight.IsNone() && + resp.FailReason == "" + }, testTimeout, 10*time.Millisecond, + "reconciliation never cleared the stale target anchor") +} + +// TestReconcileSweepReorgedOfflineDowngradesSweep restores a +// checkpoint with a Confirmed sweep that is no longer on chain. After +// reconciliation the sweep state must reset so the next planner pass +// re-emits NeedSweep and the cached sweep tx is re-broadcast. +func TestReconcileSweepReorgedOfflineDowngradesSweep(t *testing.T) { + proof := buildLinearProof(t) + rootTxid := proof.RootTxids()[0] + targetTxid := proof.TargetOutpoint().Hash + sweepTxid := chainhash.Hash{0xaa} + + checkpoint := &actorCheckpoint{ + Version: checkpointVersion, + Height: 110, + Started: true, + Trigger: TriggerRestart, + State: unrollplan.State{ + ConfirmedTxids: []chainhash.Hash{ + rootTxid, targetTxid, + }, + TargetConfirmHeight: fn.Some[int32](102), + Sweep: unrollplan.SweepState{ + Status: unrollplan.SweepStatusConfirmed, + Txid: fn.Some(sweepTxid), + ConfirmHeight: fn.Some[int32](108), + }, + }, + } + + // Reconciler: every proof anchor still confirmed, but the sweep + // is no longer on chain. + reconciler := &stubChainReconciler{ + confirmed: map[chainhash.Hash]ConfirmedAnchor{ + rootTxid: { + Txid: rootTxid, + Height: 101, + }, + targetTxid: { + Txid: targetTxid, + Height: 102, + }, + }, + } + + beh, _, _ := restoreHarness(t, proof, checkpoint, reconciler) + + actorInstance := actor.NewActor(actor.ActorConfig[Msg, Resp]{ + ID: "reconcile-test", + Behavior: &txExecAdapter{b: beh, ax: newMemExecFor(beh)}, + MailboxSize: 16, + }) + beh.selfRef = actorInstance.TellRef() + actorInstance.Start() + t.Cleanup(actorInstance.Stop) + + mustAsk(t, actorInstance.Ref(), &ResumeUnrollRequest{Height: 110}) + + require.Eventually(t, func() bool { + resp, ok := mustAsk( + t, actorInstance.Ref(), &GetStateRequest{}, + ).(*GetStateResp) + require.True(t, ok) + + // The target anchor must survive a sweep-only reorg — + // CSV maturity is still valid even though the sweep + // itself needs to be re-broadcast. The planner should + // move the actor back into AwaitingSweepBroadcast (or + // have already requested a fresh sweep build). + sweep := resp.PlannerState.Sweep + + return resp.PlannerState.TargetConfirmHeight.IsSome() && + (sweep.Status == unrollplan.SweepStatusPending || + sweep.Status == + unrollplan.SweepStatusBroadcasted) && + sweep.ConfirmHeight.IsNone() + }, testTimeout, 10*time.Millisecond, + "reconciliation never downgraded the stale sweep anchor") +} + +// TestReconcileRefreshesTargetHeightAfterOfflineReMine restores a +// checkpoint that recorded the target tx confirmed at one height, but +// the reconciler reports the target as still confirmed at a DIFFERENT +// (higher) height — i.e. the target tx was re-mined into a different +// block while the daemon was offline. The reconciler must refresh +// TargetConfirmHeight to the live value so CSV maturity is computed +// off the correct on-chain anchor, not the stale checkpoint height. +func TestReconcileRefreshesTargetHeightAfterOfflineReMine(t *testing.T) { + proof := buildLinearProof(t) + rootTxid := proof.RootTxids()[0] + targetTxid := proof.TargetOutpoint().Hash + + const ( + staleHeight = int32(102) + liveHeight = int32(105) + ) + + checkpoint := &actorCheckpoint{ + Version: checkpointVersion, + Height: staleHeight, + Started: true, + Trigger: TriggerRestart, + State: unrollplan.State{ + ConfirmedTxids: []chainhash.Hash{ + rootTxid, targetTxid, + }, + TargetConfirmHeight: fn.Some(staleHeight), + }, + } + + reconciler := &stubChainReconciler{ + confirmed: map[chainhash.Hash]ConfirmedAnchor{ + rootTxid: { + Txid: rootTxid, + Height: 101, + }, + targetTxid: { + Txid: targetTxid, + Height: liveHeight, + }, + }, + } + + err := reconcileCheckpoint(t.Context(), reconciler, proof, checkpoint) + require.NoError(t, err) + + require.True(t, checkpoint.State.TargetConfirmHeight.IsSome()) + require.Equal( + t, liveHeight, + checkpoint.State.TargetConfirmHeight.UnsafeFromSome(), + "TargetConfirmHeight must refresh to the live anchor height", + ) + require.GreaterOrEqual( + t, checkpoint.Height, liveHeight, + "checkpoint best-height should not lag the live anchor", + ) +} + +// TestReconcileTransportErrorSurfacesUpward asserts that a transport +// failure from the reconciler propagates out of ensureLoaded rather +// than silently letting the actor proceed off an unverified +// checkpoint. +func TestReconcileTransportErrorSurfacesUpward(t *testing.T) { + proof := buildLinearProof(t) + rootTxid := proof.RootTxids()[0] + + checkpoint := &actorCheckpoint{ + Version: checkpointVersion, + Height: 101, + Started: true, + Trigger: TriggerRestart, + State: unrollplan.State{ + ConfirmedTxids: []chainhash.Hash{ + rootTxid, + }, + }, + } + + reconciler := &stubChainReconciler{ + err: errors.New("backend unreachable"), + } + + beh, _, _ := restoreHarness(t, proof, checkpoint, reconciler) + + err := beh.ensureLoaded(t.Context()) + require.Error(t, err) + require.Contains(t, err.Error(), "backend unreachable") +} + +// TestReconcileExternalSpendStillLiveRestoresParkedActor restores a +// checkpoint that already carries a ProvisionalExternalSpend anchor +// and verifies that the reconciler keeps it set when the chain still +// reports the target as spent — the actor must enter +// AwaitingExternalSpendFinality on the next Resume rather than +// resuming planner-driven materialization. +func TestReconcileExternalSpendStillLiveRestoresParkedActor(t *testing.T) { + proof := buildLinearProof(t) + rootTxid := proof.RootTxids()[0] + externalTxid := chainhash.Hash{0xee} + + checkpoint := &actorCheckpoint{ + Version: checkpointVersion, + Height: 155, + Started: true, + Trigger: TriggerRestart, + State: unrollplan.State{ + ConfirmedTxids: []chainhash.Hash{ + rootTxid, + }, + }, + ProvisionalExternalSpend: fn.Some(ExternalSpendAnchor{ + SpendingTxid: externalTxid, + SpendingHeight: 155, + }), + } + + reconciler := &stubChainReconciler{ + confirmed: map[chainhash.Hash]ConfirmedAnchor{ + rootTxid: { + Txid: rootTxid, + Height: 101, + }, + }, + spent: map[wire.OutPoint]SpendAnchor{ + proof.TargetOutpoint(): { + Outpoint: proof.TargetOutpoint(), + SpendingTxid: externalTxid, + SpendingHeight: 155, + }, + }, + } + + beh, _, _ := restoreHarness(t, proof, checkpoint, reconciler) + + actorInstance := actor.NewActor(actor.ActorConfig[Msg, Resp]{ + ID: "reconcile-test", + Behavior: &txExecAdapter{b: beh, ax: newMemExecFor(beh)}, + MailboxSize: 16, + }) + beh.selfRef = actorInstance.TellRef() + actorInstance.Start() + t.Cleanup(actorInstance.Stop) + + mustAsk(t, actorInstance.Ref(), &ResumeUnrollRequest{Height: 155}) + + require.Eventually(t, func() bool { + resp, ok := mustAsk( + t, actorInstance.Ref(), &GetStateRequest{}, + ).(*GetStateResp) + require.True(t, ok) + + return resp.Phase == PhaseExternalSpendObserved && + resp.FailReason == "" + }, testTimeout, 10*time.Millisecond, + "actor did not restore into PhaseExternalSpendObserved") +} + +// TestReconcileExternalSpendReorgedOfflineResumesActor restores a +// checkpoint carrying a ProvisionalExternalSpend anchor but the +// reconciler reports the target outpoint as currently unspent: the +// spending block was reorged out while the daemon was offline. +// Reconciliation must clear the anchor so the actor resumes +// materialization instead of staying parked. +func TestReconcileExternalSpendReorgedOfflineResumesActor(t *testing.T) { + proof := buildLinearProof(t) + rootTxid := proof.RootTxids()[0] + externalTxid := chainhash.Hash{0xee} + + checkpoint := &actorCheckpoint{ + Version: checkpointVersion, + Height: 155, + Started: true, + Trigger: TriggerRestart, + State: unrollplan.State{ + InFlightTxids: []chainhash.Hash{ + rootTxid, + }, + }, + ProvisionalExternalSpend: fn.Some(ExternalSpendAnchor{ + SpendingTxid: externalTxid, + SpendingHeight: 155, + }), + } + + // Reconciler reports the target outpoint as unspent. The + // proof root is still in-flight; no confirmed anchors to verify + // here. + reconciler := &stubChainReconciler{} + + beh, _, _ := restoreHarness(t, proof, checkpoint, reconciler) + + actorInstance := actor.NewActor(actor.ActorConfig[Msg, Resp]{ + ID: "reconcile-test", + Behavior: &txExecAdapter{b: beh, ax: newMemExecFor(beh)}, + MailboxSize: 16, + }) + beh.selfRef = actorInstance.TellRef() + actorInstance.Start() + t.Cleanup(actorInstance.Stop) + + mustAsk(t, actorInstance.Ref(), &ResumeUnrollRequest{Height: 155}) + + require.Eventually(t, func() bool { + resp, ok := mustAsk( + t, actorInstance.Ref(), &GetStateRequest{}, + ).(*GetStateResp) + require.True(t, ok) + + return resp.Phase == PhaseMaterializing && + resp.FailReason == "" + }, testTimeout, 10*time.Millisecond, + "actor did not resume materialization after offline "+ + "external-spend reorg") +} + +// TestReconcileExternalSpendObservedOfflineParksActor restores a +// checkpoint that does NOT carry a ProvisionalExternalSpend anchor, +// but the reconciler reports the target outpoint as currently spent +// by a txid that is neither in the proof graph nor our sweep. The +// reconciler must install the anchor so the actor enters +// AwaitingExternalSpendFinality on the next Resume. +func TestReconcileExternalSpendObservedOfflineParksActor(t *testing.T) { + proof := buildLinearProof(t) + rootTxid := proof.RootTxids()[0] + externalTxid := chainhash.Hash{0xee} + + checkpoint := &actorCheckpoint{ + Version: checkpointVersion, + Height: 155, + Started: true, + Trigger: TriggerRestart, + State: unrollplan.State{ + ConfirmedTxids: []chainhash.Hash{ + rootTxid, + }, + }, + } + + reconciler := &stubChainReconciler{ + confirmed: map[chainhash.Hash]ConfirmedAnchor{ + rootTxid: { + Txid: rootTxid, + Height: 101, + }, + }, + spent: map[wire.OutPoint]SpendAnchor{ + proof.TargetOutpoint(): { + Outpoint: proof.TargetOutpoint(), + SpendingTxid: externalTxid, + SpendingHeight: 155, + }, + }, + } + + beh, _, _ := restoreHarness(t, proof, checkpoint, reconciler) + + actorInstance := actor.NewActor(actor.ActorConfig[Msg, Resp]{ + ID: "reconcile-test", + Behavior: &txExecAdapter{b: beh, ax: newMemExecFor(beh)}, + MailboxSize: 16, + }) + beh.selfRef = actorInstance.TellRef() + actorInstance.Start() + t.Cleanup(actorInstance.Stop) + + mustAsk(t, actorInstance.Ref(), &ResumeUnrollRequest{Height: 155}) + + require.Eventually(t, func() bool { + resp, ok := mustAsk( + t, actorInstance.Ref(), &GetStateRequest{}, + ).(*GetStateResp) + require.True(t, ok) + + return resp.Phase == PhaseExternalSpendObserved && + resp.FailReason == "" + }, testTimeout, 10*time.Millisecond, + "actor did not park in PhaseExternalSpendObserved after "+ + "offline external spend") +} diff --git a/unroll/registry.go b/unroll/registry.go index 881ab30c8..bbdd652d8 100644 --- a/unroll/registry.go +++ b/unroll/registry.go @@ -160,6 +160,16 @@ type RegistryConfig struct { // to exit (darepo-client#602). When None, terminal outcomes are not // forwarded (used by tests that don't exercise the manager). VTXOExitObserver fn.Option[actor.TellOnlyRef[vtxo.ManagerMsg]] + + // ChainReconcilerFactory, when set, is forwarded to each spawned + // per-target actor so it can verify its checkpoint anchors + // against the canonical chain on restart. The factory is invoked + // once per actor lifetime after the proof loads. When None + // children skip reconciliation; production wiring should pass a + // factory backed by NewChainSourceReconciler so an offline + // reorg window does not silently leave actors driving side + // effects off stale planner state. + ChainReconcilerFactory fn.Option[ChainReconcilerFactory] } // UnrollRegistryActor wraps the thin unroll registry actor. @@ -1446,6 +1456,7 @@ func (r *registryBehavior) childConfig(target wire.OutPoint) Config { ExitSpendPolicyResolver: r.cfg.ExitSpendPolicyResolver, FraudCheckpointSafetyMargin: r.cfg.FraudCheckpointSafetyMargin, RegistryRef: r.selfRef, + ChainReconcilerFactory: r.cfg.ChainReconcilerFactory, } } diff --git a/unroll/reorg_safety_test.go b/unroll/reorg_safety_test.go new file mode 100644 index 000000000..f510413bd --- /dev/null +++ b/unroll/reorg_safety_test.go @@ -0,0 +1,663 @@ +package unroll + +import ( + "context" + "sync" + "testing" + "time" + + "github.com/btcsuite/btcd/chainhash/v2" + "github.com/btcsuite/btclog/v2" + "github.com/lightninglabs/darepo-client/baselib/actor" + "github.com/lightninglabs/darepo-client/unrollplan" + fn "github.com/lightningnetwork/lnd/fn/v2" + "github.com/stretchr/testify/require" +) + +// TestProofRootReorgBlocksDownstreamMaterialization drives a proof root +// through Confirmed -> Reorged -> Confirmed and asserts that the unroll +// actor does NOT submit the target transaction while the root anchor is +// torn down. Only once the root reconfirms does the actor advance the +// proof graph. +// +// This is the load-bearing reorg-safety case: before the rollback work +// the actor treated the first TxConfirmedMsg as a monotonic fact and +// would have submitted the target while the chain was still missing the +// root. After rollback, the target submission is gated on the root +// anchor being live. +func TestProofRootReorgBlocksDownstreamMaterialization(t *testing.T) { + proof := buildLinearProof(t) + desc := testDescriptor(t, proof.TargetOutpoint(), proof.CSVDelay()) + unrollActor, _, txconfirmRef, _ := newActorHarness(t, proof, desc) + + mustAsk(t, unrollActor.Ref(), &StartUnrollRequest{ + Height: 100, + Trigger: TriggerManual, + }) + + rootTxid := proof.RootTxids()[0] + targetTxid := proof.TargetOutpoint().Hash + + // Wait for the actor to register the root with txconfirm. + require.Eventually(t, func() bool { + return txconfirmRef.requestCountForTxid(rootTxid) == 1 + }, testTimeout, 10*time.Millisecond, "root never submitted") + + // 1. Root confirms; the actor should immediately submit the target. + txconfirmRef.emitConfirmed(t, 0, rootTxid, 101) + require.Eventually(t, func() bool { + return txconfirmRef.requestCountForTxid(targetTxid) == 1 + }, testTimeout, 10*time.Millisecond, "target never submitted") + + // 2. Root reorgs out. The actor must drop the root anchor AND the + // dependent target anchor from in-flight, since State.Validate + // requires every in-flight node to have confirmed parents in the + // proof graph. + targetSubmissionsBeforeReorg := txconfirmRef.requestCountForTxid( + targetTxid, + ) + txconfirmRef.emitReorged(t, 0, rootTxid) + + require.Eventually(t, func() bool { + resp, ok := mustAsk( + t, unrollActor.Ref(), &GetStateRequest{}, + ).(*GetStateResp) + require.True(t, ok) + for _, h := range resp.PlannerState.ConfirmedTxids { + if h == rootTxid { + return false + } + } + for _, h := range resp.PlannerState.InFlightTxids { + if h == targetTxid { + return false + } + } + + return true + }, testTimeout, 10*time.Millisecond, + "root + target anchors never cleared after reorg") + + // The actor naturally re-submits the root on its ready frontier + // after the reorg drops it from ConfirmedTxids — txconfirm dedup + // absorbs that re-submit. What matters for reorg safety is that the + // TARGET was NOT re-submitted while its parent anchor was missing: + // driving target broadcast off a non-anchored parent is exactly the + // unsafe behavior this work fixes. + time.Sleep(50 * time.Millisecond) + require.Equal( + t, targetSubmissionsBeforeReorg, + txconfirmRef.requestCountForTxid(targetTxid), + "target was re-submitted while its parent was reorged out", + ) + + // 3. Root reconfirms in a different block. The actor should keep + // progressing without resubmitting target (txconfirm dedup already + // holds the in-flight subscription, but the planner state must be + // coherent). + txconfirmRef.emitConfirmed(t, 0, rootTxid, 102) + require.Eventually(t, func() bool { + resp, ok := mustAsk( + t, unrollActor.Ref(), &GetStateRequest{}, + ).(*GetStateResp) + require.True(t, ok) + for _, h := range resp.PlannerState.ConfirmedTxids { + if h == rootTxid { + return true + } + } + + return false + }, testTimeout, 10*time.Millisecond, + "root anchor never re-recorded after reconfirm") +} + +// TestTargetReorgClearsCSVMaturity confirms the target tx, waits for the +// sweep to broadcast, then reorgs the target out. The actor must clear +// TargetConfirmHeight and downgrade the sweep so the planner stops +// reporting Done. +func TestTargetReorgClearsCSVMaturity(t *testing.T) { + proof := buildLinearProof(t) + desc := testDescriptor(t, proof.TargetOutpoint(), proof.CSVDelay()) + unrollActor, _, txconfirmRef, _ := newActorHarness(t, proof, desc) + + mustAsk(t, unrollActor.Ref(), &StartUnrollRequest{ + Height: 100, + Trigger: TriggerManual, + }) + + rootTxid := proof.RootTxids()[0] + targetTxid := proof.TargetOutpoint().Hash + + require.Eventually(t, func() bool { + return txconfirmRef.requestCountForTxid(rootTxid) == 1 + }, testTimeout, 10*time.Millisecond) + + txconfirmRef.emitConfirmed(t, 0, rootTxid, 101) + require.Eventually(t, func() bool { + return txconfirmRef.requestCountForTxid(targetTxid) == 1 + }, testTimeout, 10*time.Millisecond) + + txconfirmRef.emitConfirmed(t, 1, targetTxid, 102) + + // Advance height past CSV maturity so the sweep is built and + // broadcast. + mustAsk(t, unrollActor.Ref(), &HeightObservedMsg{ + Height: 102 + int32(proof.CSVDelay()) + 1, + }) + + require.Eventually(t, func() bool { + return txconfirmRef.requestCount() >= 3 + }, testTimeout, 10*time.Millisecond, + "sweep was never broadcast") + + sweepReq := txconfirmRef.lastRequest(t) + sweepTxid := sweepReq.Tx.TxHash() + + // Sweep confirms. + txconfirmRef.emitConfirmed(t, 2, sweepTxid, 110) + require.Eventually(t, func() bool { + resp, ok := mustAsk( + t, unrollActor.Ref(), &GetStateRequest{}, + ).(*GetStateResp) + require.True(t, ok) + + return resp.Phase == PhaseCompleted + }, testTimeout, 10*time.Millisecond, + "sweep never reached Completed") + + // 1. Target reorg invalidates the CSV anchor AND the sweep + // confirmation (which depended on it). + txconfirmRef.emitReorged(t, 1, targetTxid) + + require.Eventually(t, func() bool { + resp, ok := mustAsk( + t, unrollActor.Ref(), &GetStateRequest{}, + ).(*GetStateResp) + require.True(t, ok) + + csvCleared := resp.PlannerState.TargetConfirmHeight.IsNone() + targetCleared := true + for _, h := range resp.PlannerState.ConfirmedTxids { + if h == targetTxid { + targetCleared = false + } + } + + return csvCleared && targetCleared + }, testTimeout, 10*time.Millisecond, + "target anchor never cleared after reorg") +} + +// recordingRegistryRef is a stub RegistryRef that captures every +// message delivered by the per-target actor's notifyRegistryIfTerminal +// path so a test can assert "the registry has / has not been told the +// actor is terminal". +type recordingRegistryRef struct { + id string + mu sync.Mutex + msgs []RegistryMsg +} + +// ID returns the stub registry ref identifier. +func (r *recordingRegistryRef) ID() string { + return r.id +} + +// Tell records the inbound RegistryMsg and acks. +func (r *recordingRegistryRef) Tell(_ context.Context, msg RegistryMsg) error { + r.mu.Lock() + defer r.mu.Unlock() + r.msgs = append(r.msgs, msg) + + return nil +} + +// terminatedCount returns how many UnrollTerminatedMsg messages have +// been delivered so far. +func (r *recordingRegistryRef) terminatedCount() int { + r.mu.Lock() + defer r.mu.Unlock() + + count := 0 + for _, m := range r.msgs { + if _, ok := m.(*UnrollTerminatedMsg); ok { + count++ + } + } + + return count +} + +// TestSweepCompletionStaysProvisionalUntilFinalized proves the Phase 7 +// invariant: a per-target actor that has reached PhaseCompleted does +// NOT notify the registry as terminal until a TxFinalizedMsg for the +// sweep txid arrives. Until then the actor stays alive so a reorg of +// the sweep confirmation has a live actor to deliver the rollback to. +func TestSweepCompletionStaysProvisionalUntilFinalized(t *testing.T) { + proof := buildLinearProof(t) + desc := testDescriptor(t, proof.TargetOutpoint(), proof.CSVDelay()) + + store := newMemCheckpointStore() + txconfirmRef := &fakeTxConfirmRef{} + registryRef := &recordingRegistryRef{id: "reg-stub"} + + cfg := Config{ + TargetOutpoint: proof.TargetOutpoint(), + ActorID: "unroll-finality-test", + DeliveryStore: store, + ProofAssembler: &mockProofAssembler{ + proof: proof, + }, + VTXOStore: &mockVTXOStore{ + desc: desc, + }, + TxConfirmRef: txconfirmRef, + ChainSource: &fakeChainSourceRef{}, + Wallet: &fakeSweepWallet{}, + Log: fn.Some(btclog.Disabled), + RegistryRef: registryRef, + } + beh := &behavior{cfg: cfg, log: btclog.Disabled} + require.NoError(t, beh.restoreCheckpoint(t.Context())) + + actorInstance := actor.NewActor(actor.ActorConfig[Msg, Resp]{ + ID: "unroll-finality-test", + Behavior: &txExecAdapter{b: beh, ax: newMemExecFor(beh)}, + MailboxSize: 32, + }) + beh.selfRef = actorInstance.TellRef() + actorInstance.Start() + t.Cleanup(actorInstance.Stop) + + mustAsk(t, actorInstance.Ref(), &StartUnrollRequest{ + Height: 100, + Trigger: TriggerManual, + }) + + rootTxid := proof.RootTxids()[0] + targetTxid := proof.TargetOutpoint().Hash + + require.Eventually(t, func() bool { + return txconfirmRef.requestCountForTxid(rootTxid) == 1 + }, testTimeout, 10*time.Millisecond) + + txconfirmRef.emitConfirmed(t, 0, rootTxid, 101) + require.Eventually(t, func() bool { + return txconfirmRef.requestCountForTxid(targetTxid) == 1 + }, testTimeout, 10*time.Millisecond) + + txconfirmRef.emitConfirmed(t, 1, targetTxid, 102) + mustAsk(t, actorInstance.Ref(), &HeightObservedMsg{ + Height: 102 + int32(proof.CSVDelay()) + 1, + }) + + require.Eventually(t, func() bool { + return txconfirmRef.requestCount() >= 3 + }, testTimeout, 10*time.Millisecond) + + sweepTxid := txconfirmRef.lastRequest(t).Tx.TxHash() + txconfirmRef.emitConfirmed(t, 2, sweepTxid, 110) + + // Wait for the actor to settle on PhaseCompleted. + require.Eventually(t, func() bool { + resp, ok := mustAsk( + t, actorInstance.Ref(), &GetStateRequest{}, + ).(*GetStateResp) + require.True(t, ok) + + return resp.Phase == PhaseCompleted + }, testTimeout, 10*time.Millisecond) + + // 1. PROVISIONAL: registry must NOT have been told the actor is + // terminal. The sweep can still reorg out. + time.Sleep(50 * time.Millisecond) + require.Equal( + t, 0, registryRef.terminatedCount(), + "registry was told terminal before TxFinalized arrived", + ) + + // 2. Sweep reorgs out; the live actor rolls back to + // AwaitingSweepConfirmation. Registry still has not been + // terminated. + txconfirmRef.emitReorged(t, 2, sweepTxid) + require.Eventually(t, func() bool { + resp, ok := mustAsk( + t, actorInstance.Ref(), &GetStateRequest{}, + ).(*GetStateResp) + require.True(t, ok) + + return resp.Phase == PhaseSweepConfirmation + }, testTimeout, 10*time.Millisecond, + "actor never rolled back to AwaitingSweepConfirmation") + require.Equal( + t, 0, registryRef.terminatedCount(), + "registry was told terminal during sweep reorg recovery", + ) + + // 3. Sweep re-confirms; actor returns to PhaseCompleted. + txconfirmRef.emitConfirmed(t, 2, sweepTxid, 111) + require.Eventually(t, func() bool { + resp, ok := mustAsk( + t, actorInstance.Ref(), &GetStateRequest{}, + ).(*GetStateResp) + require.True(t, ok) + + return resp.Phase == PhaseCompleted + }, testTimeout, 10*time.Millisecond) + require.Equal( + t, 0, registryRef.terminatedCount(), + "registry was told terminal after sweep re-confirm but "+ + "before TxFinalized", + ) + + // 4. Finality. TxFinalizedMsg matches the recorded sweep txid, + // the actor latches sweepFinalized, and the next driveEvent + // fires UnrollTerminatedMsg. + txconfirmRef.emitFinalized(t, 2, sweepTxid) + require.Eventually(t, func() bool { + return registryRef.terminatedCount() == 1 + }, testTimeout, 10*time.Millisecond, + "registry was never told terminal after TxFinalized") +} + +// TestExternalSpendReorgResumesActor observes an external spend of the +// target outpoint while materialization is in flight, then drives a +// reorg of the spending block. The actor must park in +// AwaitingExternalSpendFinality while the spend is provisional, then +// resume normal planning once the spend reorgs out. Without this +// behavior, a transient reorg-out replacement of the target would +// permanently terminate the recovery job. +func TestExternalSpendReorgResumesActor(t *testing.T) { + proof := buildLinearProof(t) + desc := testDescriptor(t, proof.TargetOutpoint(), proof.CSVDelay()) + unrollActor, beh, txconfirmRef, _ := newActorHarness(t, proof, desc) + + chainSource, ok := beh.cfg.ChainSource.(*fakeChainSourceRef) + require.True(t, ok) + + mustAsk(t, unrollActor.Ref(), &StartUnrollRequest{ + Height: 100, + Trigger: TriggerManual, + }) + + rootTxid := proof.RootTxids()[0] + require.Eventually(t, func() bool { + return txconfirmRef.requestCountForTxid(rootTxid) == 1 + }, testTimeout, 10*time.Millisecond, + "root never submitted") + + // Ensure spend watch on the target is registered with the + // reorg / done refs wired before driving any events. + require.Eventually(t, func() bool { + var targetRegistered bool + for _, op := range chainSource.spendRegistrations() { + if op == proof.TargetOutpoint() { + targetRegistered = true + break + } + } + if !targetRegistered { + return false + } + + chainSource.mu.Lock() + defer chainSource.mu.Unlock() + + return chainSource.spendReorgedRef != nil && + chainSource.spendFinalizedRef != nil + }, testTimeout, 10*time.Millisecond) + + // 1. An external party broadcasts a spend of the target outpoint. + externalTxid := chainhash.Hash{0xee} + chainSource.emitSpendForOutpoint( + t, proof.TargetOutpoint(), externalTxid, 110, + ) + + // The actor must park in AwaitingExternalSpendFinality rather + // than transitioning to Failed; the spend is provisional. + require.Eventually(t, func() bool { + resp, ok := mustAsk( + t, unrollActor.Ref(), &GetStateRequest{}, + ).(*GetStateResp) + require.True(t, ok) + + return resp.Phase == PhaseExternalSpendObserved && + resp.FailReason == "" + }, testTimeout, 10*time.Millisecond, + "actor never parked in PhaseExternalSpendObserved") + + // 2. The spending block is reorged out. The actor must clear the + // provisional anchor and resume normal planning. Because the + // proof root is still in-flight (no confirmation has arrived + // yet), the actor goes back to PhaseMaterializing. + chainSource.emitSpendReorged(t) + + require.Eventually(t, func() bool { + resp, ok := mustAsk( + t, unrollActor.Ref(), &GetStateRequest{}, + ).(*GetStateResp) + require.True(t, ok) + + return resp.Phase == PhaseMaterializing && + resp.FailReason == "" + }, testTimeout, 10*time.Millisecond, + "actor did not resume materialization after spend reorg") +} + +// TestSweepConfirmationReorgReversesCompleted drives the full happy path +// to Completed and then reorgs the sweep confirmation out. The actor +// must drop SweepStatusConfirmed back to Broadcasted and clear the +// stored ConfirmHeight so the planner stops reporting Done. +func TestSweepConfirmationReorgReversesCompleted(t *testing.T) { + proof := buildLinearProof(t) + desc := testDescriptor(t, proof.TargetOutpoint(), proof.CSVDelay()) + unrollActor, _, txconfirmRef, _ := newActorHarness(t, proof, desc) + + mustAsk(t, unrollActor.Ref(), &StartUnrollRequest{ + Height: 100, + Trigger: TriggerManual, + }) + + rootTxid := proof.RootTxids()[0] + targetTxid := proof.TargetOutpoint().Hash + + require.Eventually(t, func() bool { + return txconfirmRef.requestCountForTxid(rootTxid) == 1 + }, testTimeout, 10*time.Millisecond) + + txconfirmRef.emitConfirmed(t, 0, rootTxid, 101) + require.Eventually(t, func() bool { + return txconfirmRef.requestCountForTxid(targetTxid) == 1 + }, testTimeout, 10*time.Millisecond) + + txconfirmRef.emitConfirmed(t, 1, targetTxid, 102) + mustAsk(t, unrollActor.Ref(), &HeightObservedMsg{ + Height: 102 + int32(proof.CSVDelay()) + 1, + }) + + require.Eventually(t, func() bool { + return txconfirmRef.requestCount() >= 3 + }, testTimeout, 10*time.Millisecond) + + sweepReq := txconfirmRef.lastRequest(t) + sweepTxid := sweepReq.Tx.TxHash() + + txconfirmRef.emitConfirmed(t, 2, sweepTxid, 110) + require.Eventually(t, func() bool { + resp, ok := mustAsk( + t, unrollActor.Ref(), &GetStateRequest{}, + ).(*GetStateResp) + require.True(t, ok) + + return resp.Phase == PhaseCompleted + }, testTimeout, 10*time.Millisecond) + + // 1. Sweep confirmation reorgs out. The actor must roll back to + // AwaitingSweepConfirmation: the signed sweep is still durable, the + // txconfirm subscription is still live, but the planner stops + // reporting Done. + txconfirmRef.emitReorged(t, 2, sweepTxid) + + require.Eventually(t, func() bool { + resp, ok := mustAsk( + t, unrollActor.Ref(), &GetStateRequest{}, + ).(*GetStateResp) + require.True(t, ok) + + return resp.Phase == PhaseSweepConfirmation && + resp.PlannerState.Sweep.Status == + unrollplan.SweepStatusBroadcasted && + resp.PlannerState.Sweep.ConfirmHeight.IsNone() + }, testTimeout, 10*time.Millisecond, + "sweep reorg did not roll the actor back to "+ + "AwaitingSweepConfirmation") + + // 2. Sweep reconfirms in a different block. The actor must move + // back to Completed. + txconfirmRef.emitConfirmed(t, 2, sweepTxid, 111) + require.Eventually(t, func() bool { + resp, ok := mustAsk( + t, unrollActor.Ref(), &GetStateRequest{}, + ).(*GetStateResp) + require.True(t, ok) + + sweep := resp.PlannerState.Sweep + + return resp.Phase == PhaseCompleted && + sweep.ConfirmHeight.IsSome() && + sweep.ConfirmHeight.UnsafeFromSome() == 111 + }, testTimeout, 10*time.Millisecond, + "sweep did not re-Complete after reconfirm") +} + +// TestOfflineReorgRestartReconcilesBeforeSideEffects bundles the three +// offline-reorg scenarios (target reorged out, sweep reorged out, and +// external-spend reorged out) into one restart and asserts that the +// reconciler runs BEFORE the FSM session is built — i.e. before the +// actor can broadcast a stale sweep or park in +// AwaitingExternalSpendFinality on a vanished spender. +// +// The setup mimics a daemon that committed a checkpoint reflecting: +// +// - root + target proof nodes confirmed +// - a sweep tx broadcast AND confirmed at H=108 +// - an external spender provisionally observed at H=110 +// +// While the daemon was offline, the chain reorged: the target's +// confirming block was orphaned (so the sweep that spent it can no +// longer be confirmed either), and the external spender's block was +// also dropped. The reconciler the actor consults on restart returns +// "root still confirmed" and "everything else absent". +// +// After Resume the actor must end up in the post-reconcile state +// (sweep downgraded to Pending, TargetConfirmHeight cleared, target +// pruned from ConfirmedTxids, ProvisionalExternalSpend cleared) and +// must NOT have re-submitted the stale sweep tx to txconfirm before +// the planner re-derives broadcast intent from the post-reconcile +// PlannerState. The latter is the load-bearing safety invariant: a +// re-broadcast off a stale checkpoint would race a fresh wallet +// pkScript against any future re-mining of the target. +func TestOfflineReorgRestartReconcilesBeforeSideEffects(t *testing.T) { + proof := buildLinearProof(t) + rootTxid := proof.RootTxids()[0] + targetTxid := proof.TargetOutpoint().Hash + staleSweepTxid := chainhash.Hash{0xaa} + staleExternalTxid := chainhash.Hash{0xbb} + + checkpoint := &actorCheckpoint{ + Version: checkpointVersion, + Height: 112, + Started: true, + Trigger: TriggerRestart, + State: unrollplan.State{ + ConfirmedTxids: []chainhash.Hash{ + rootTxid, targetTxid, + }, + TargetConfirmHeight: fn.Some[int32](102), + Sweep: unrollplan.SweepState{ + Status: unrollplan.SweepStatusConfirmed, + Txid: fn.Some(staleSweepTxid), + ConfirmHeight: fn.Some[int32](108), + }, + }, + ProvisionalExternalSpend: fn.Some(ExternalSpendAnchor{ + SpendingTxid: staleExternalTxid, + SpendingHeight: 110, + }), + } + + // Reconciler reports: + // - root still confirmed (so the planner keeps the partial + // proof-graph progress) + // - target absent (offline reorg) + // - stale sweep absent (offline reorg) + // - target outpoint unspent (external spender reorged out) + reconciler := &stubChainReconciler{ + confirmed: map[chainhash.Hash]ConfirmedAnchor{ + rootTxid: { + Txid: rootTxid, + Height: 101, + }, + }, + } + + beh, txconfirmRef, _ := restoreHarness( + t, proof, checkpoint, reconciler, + ) + + actorInstance := actor.NewActor(actor.ActorConfig[Msg, Resp]{ + ID: "reconcile-test", + Behavior: &txExecAdapter{b: beh, ax: newMemExecFor(beh)}, + MailboxSize: 16, + }) + beh.selfRef = actorInstance.TellRef() + actorInstance.Start() + t.Cleanup(actorInstance.Stop) + + mustAsk(t, actorInstance.Ref(), &ResumeUnrollRequest{Height: 112}) + + // The actor must end up in the post-reconcile state: sweep + // downgraded, target-derived height cleared, target pruned from + // ConfirmedTxids, external spend cleared, root preserved. + require.Eventually(t, func() bool { + resp, ok := mustAsk( + t, actorInstance.Ref(), &GetStateRequest{}, + ).(*GetStateResp) + require.True(t, ok) + + for _, h := range resp.PlannerState.ConfirmedTxids { + if h == targetTxid { + return false + } + } + sweep := resp.PlannerState.Sweep + rootPresent := false + for _, h := range resp.PlannerState.ConfirmedTxids { + if h == rootTxid { + rootPresent = true + break + } + } + + return rootPresent && + sweep.Status == unrollplan.SweepStatusPending && + sweep.Txid.IsNone() && + sweep.ConfirmHeight.IsNone() && + resp.PlannerState.TargetConfirmHeight.IsNone() && + resp.Phase == PhaseMaterializing && + resp.FailReason == "" + }, testTimeout, 10*time.Millisecond, + "actor did not reach post-reconcile state on restart") + + // Load-bearing safety invariant: every txconfirm request issued + // after restart must target a tx the post-reconcile PlannerState + // authorizes. In particular, the orphaned sweep txid must never + // be re-submitted — that would replay a sweep against a target + // the chain says no longer exists. + require.Zero( + t, txconfirmRef.requestCountForTxid(staleSweepTxid), + "stale sweep tx was re-submitted to txconfirm before "+ + "reconciliation downgraded it", + ) +} diff --git a/unroll/snapshot.go b/unroll/snapshot.go index 8f6668dd0..c9518c344 100644 --- a/unroll/snapshot.go +++ b/unroll/snapshot.go @@ -2,10 +2,13 @@ package unroll import ( "bytes" + "encoding/binary" "fmt" + "github.com/btcsuite/btcd/chainhash/v2" "github.com/btcsuite/btcd/wire/v2" "github.com/lightninglabs/darepo-client/unrollplan" + fn "github.com/lightningnetwork/lnd/fn/v2" "github.com/lightningnetwork/lnd/tlv" ) @@ -93,21 +96,76 @@ const ( // checkpointExitPolicyRefRecordType carries the policy-specific // durable-state reference. checkpointExitPolicyRefRecordType tlv.Type = 21 + + // checkpointExternalSpendRecordType is optional; present only when + // the actor has observed an external spend of the target outpoint + // that has not yet been finalized. Payload is a fixed-layout + // 36-byte blob (32-byte SpendingTxid + 4-byte big-endian + // SpendingHeight) so a daemon restarting mid-spend-finality-window + // can rehydrate into AwaitingExternalSpendFinality instead of + // dropping the provisional anchor and broadcasting a sweep on a + // target the chain says no longer exists. + checkpointExternalSpendRecordType tlv.Type = 23 + + // checkpointSweepFinalizedRecordType is optional; present (value 1) + // only when the sweep has finalized past the backend's reorg-safety + // depth. It makes the in-memory sweep-finalized latch durable so a + // PhaseCompleted entry whose terminal handoff was deferred or failed + // does not restart as permanently "provisional completed" (the latch + // gates notifyRegistryIfTerminal). Omitted when false. + checkpointSweepFinalizedRecordType tlv.Type = 25 ) +// externalSpendBlobSize is the canonical wire size of the persisted +// ProvisionalExternalSpend payload: 32-byte txid + 4-byte big-endian +// height. +const externalSpendBlobSize = chainhash.HashSize + 4 + // actorCheckpoint is the durable checkpoint shape for one VTXO unroll actor. type actorCheckpoint struct { - Version uint8 - Height int32 - Started bool - Trigger StartTrigger - State unrollplan.State - ExitPolicyKind ExitPolicyKind - ExitPolicyRef string - SweepTx *wire.MsgTx - Fail string - SweepAttempts int - DeferredCheckpoints []DeferredCheckpoint + Version uint8 + Height int32 + Started bool + Trigger StartTrigger + State unrollplan.State + ExitPolicyKind ExitPolicyKind + ExitPolicyRef string + SweepTx *wire.MsgTx + Fail string + SweepAttempts int + DeferredCheckpoints []DeferredCheckpoint + ProvisionalExternalSpend fn.Option[ExternalSpendAnchor] + SweepFinalized bool +} + +// encodeExternalSpendBlob serializes an ExternalSpendAnchor into the +// fixed-layout payload carried by checkpointExternalSpendRecordType. +func encodeExternalSpendBlob(anchor ExternalSpendAnchor) []byte { + blob := make([]byte, externalSpendBlobSize) + copy(blob[:chainhash.HashSize], anchor.SpendingTxid[:]) + binary.BigEndian.PutUint32( + blob[chainhash.HashSize:], uint32(anchor.SpendingHeight), + ) + + return blob +} + +// decodeExternalSpendBlob parses a fixed-layout external-spend payload +// back into an ExternalSpendAnchor. +func decodeExternalSpendBlob(blob []byte) (ExternalSpendAnchor, error) { + if len(blob) != externalSpendBlobSize { + return ExternalSpendAnchor{}, fmt.Errorf("external spend blob "+ + "has %d bytes, want %d", len(blob), + externalSpendBlobSize) + } + + var anchor ExternalSpendAnchor + copy(anchor.SpendingTxid[:], blob[:chainhash.HashSize]) + anchor.SpendingHeight = int32( + binary.BigEndian.Uint32(blob[chainhash.HashSize:]), + ) + + return anchor, nil } // encodeCheckpoint serializes one actor checkpoint into canonical TLV @@ -226,6 +284,27 @@ func encodeCheckpoint(value *actorCheckpoint) ([]byte, error) { ) } + value.ProvisionalExternalSpend.WhenSome( + func(anchor ExternalSpendAnchor) { + blob := encodeExternalSpendBlob(anchor) + records = append( + records, tlv.MakePrimitiveRecord( + checkpointExternalSpendRecordType, + &blob, + ), + ) + }, + ) + + if value.SweepFinalized { + finalized := uint8(1) + records = append( + records, tlv.MakePrimitiveRecord( + checkpointSweepFinalizedRecordType, &finalized, + ), + ) + } + stream, err := tlv.NewStream(records...) if err != nil { return nil, fmt.Errorf("create checkpoint stream: %w", err) @@ -255,17 +334,19 @@ func encodeCheckpoint(value *actorCheckpoint) ([]byte, error) { // truncated state. func decodeCheckpoint(raw []byte) (*actorCheckpoint, error) { var ( - version uint8 - height uint32 - started uint8 - trigger uint32 - stateBytes []byte - sweepBytes []byte - failBytes []byte - attempts uint32 - deferredBytes []byte - policyKind []byte - policyRef []byte + version uint8 + height uint32 + started uint8 + trigger uint32 + stateBytes []byte + sweepBytes []byte + failBytes []byte + attempts uint32 + deferredBytes []byte + policyKind []byte + policyRef []byte + extSpendBlob []byte + sweepFinalized uint8 ) stream, err := tlv.NewStream( @@ -302,6 +383,12 @@ func decodeCheckpoint(raw []byte) (*actorCheckpoint, error) { tlv.MakePrimitiveRecord( checkpointExitPolicyRefRecordType, &policyRef, ), + tlv.MakePrimitiveRecord( + checkpointExternalSpendRecordType, &extSpendBlob, + ), + tlv.MakePrimitiveRecord( + checkpointSweepFinalizedRecordType, &sweepFinalized, + ), ) if err != nil { return nil, fmt.Errorf("create checkpoint stream: %w", err) @@ -367,6 +454,18 @@ func decodeCheckpoint(raw []byte) (*actorCheckpoint, error) { checkpoint.DeferredCheckpoints = checkpoints } + if _, ok := parsed[checkpointExternalSpendRecordType]; ok { + anchor, err := decodeExternalSpendBlob(extSpendBlob) + if err != nil { + return nil, fmt.Errorf("decode external spend: %w", err) + } + checkpoint.ProvisionalExternalSpend = fn.Some(anchor) + } + + if _, ok := parsed[checkpointSweepFinalizedRecordType]; ok { + checkpoint.SweepFinalized = sweepFinalized != 0 + } + return checkpoint, nil } diff --git a/unroll/snapshot_test.go b/unroll/snapshot_test.go index 866725c25..6064684fd 100644 --- a/unroll/snapshot_test.go +++ b/unroll/snapshot_test.go @@ -81,6 +81,28 @@ func TestCheckpointCodecRoundTripHandcrafted(t *testing.T) { SweepAttempts: 1, }, }, + { + name: "sweep_finalized", + checkpoint: &actorCheckpoint{ + Version: checkpointVersion, + Height: 210, + Started: true, + Trigger: TriggerManual, + State: unrollplan.State{ + ConfirmedTxids: []chainhash.Hash{ + targetTxid, + }, + Sweep: unrollplan.SweepState{ + Status: unrollplan. + SweepStatusBroadcasted, + Txid: fn.Some(sweepTxid), + }, + }, + SweepTx: sweepTx, + SweepAttempts: 1, + SweepFinalized: true, + }, + }, { name: "failed", checkpoint: &actorCheckpoint{ @@ -111,6 +133,31 @@ func TestCheckpointCodecRoundTripHandcrafted(t *testing.T) { }}, }, }, + { + name: "provisional_external_spend", + checkpoint: &actorCheckpoint{ + Version: checkpointVersion, + Height: 155, + Started: true, + Trigger: TriggerManual, + State: unrollplan.State{ + ConfirmedTxids: []chainhash.Hash{ + targetTxid, + }, + TargetConfirmHeight: fn.Some[int32]( + 154, + ), + }, + ProvisionalExternalSpend: fn.Some( + ExternalSpendAnchor{ + SpendingTxid: hashFromByteCk( + 0xee, + ), + SpendingHeight: 155, + }, + ), + }, + }, } for _, tc := range cases { @@ -597,6 +644,16 @@ func requireCheckpointEqual(t *testing.T, want, got *actorCheckpoint) { t, txsEqualCk(want.SweepTx, got.SweepTx), "sweep transactions differ", ) + require.Equal( + t, want.ProvisionalExternalSpend.IsSome(), + got.ProvisionalExternalSpend.IsSome(), + ) + if want.ProvisionalExternalSpend.IsSome() { + require.Equal( + t, want.ProvisionalExternalSpend.UnsafeFromSome(), + got.ProvisionalExternalSpend.UnsafeFromSome(), + ) + } } // hashFromByteCk builds a chainhash.Hash with the first byte set to b, useful diff --git a/unroll/state_snapshot.go b/unroll/state_snapshot.go index a3292759e..23d682d2c 100644 --- a/unroll/state_snapshot.go +++ b/unroll/state_snapshot.go @@ -40,6 +40,7 @@ func checkpointFromState(state State, sweepTx *wire.MsgTx) *actorCheckpoint { } checkpoint.Fail = job.FailReason checkpoint.SweepAttempts = job.SweepAttempts + checkpoint.ProvisionalExternalSpend = job.ProvisionalExternalSpend return checkpoint } @@ -102,14 +103,17 @@ func stateFromCheckpoint(checkpoint *actorCheckpoint) State { deferred := copyDeferredCheckpoints(checkpoint.DeferredCheckpoints) job := &JobState{ - Height: checkpoint.Height, - Trigger: checkpoint.Trigger, - ExitPolicyKind: exitPolicyKind(checkpoint.ExitPolicyKind), - ExitPolicyRef: checkpoint.ExitPolicyRef, - PlannerState: copyPlannerState(checkpoint.State), - DeferredCheckpoints: deferred, - FailReason: checkpoint.Fail, - SweepAttempts: checkpoint.SweepAttempts, + Height: checkpoint.Height, + Trigger: checkpoint.Trigger, + ExitPolicyKind: exitPolicyKind( + checkpoint.ExitPolicyKind, + ), + ExitPolicyRef: checkpoint.ExitPolicyRef, + PlannerState: copyPlannerState(checkpoint.State), + DeferredCheckpoints: deferred, + FailReason: checkpoint.Fail, + SweepAttempts: checkpoint.SweepAttempts, + ProvisionalExternalSpend: checkpoint.ProvisionalExternalSpend, } switch phaseFromPlannerState(job) { @@ -119,6 +123,9 @@ func stateFromCheckpoint(checkpoint *actorCheckpoint) State { case PhaseFailed: return &Failed{Job: job} + case PhaseExternalSpendObserved: + return &AwaitingExternalSpendFinality{Job: job} + case PhaseSweepConfirmation: return &AwaitingSweepConfirmation{Job: job} @@ -152,6 +159,9 @@ func phaseFromState(state State) Phase { case *AwaitingSweepConfirmation: return PhaseSweepConfirmation + case *AwaitingExternalSpendFinality: + return PhaseExternalSpendObserved + case *Completed: return PhaseCompleted @@ -174,6 +184,15 @@ func phaseFromPlannerState(job *JobState) Phase { return PhaseFailed } + // A persisted provisional external spend takes precedence over the + // sweep-based phase derivation: the actor was parked waiting for + // either a reorg (which clears the anchor) or finality (which + // promotes it to FailReason). Surfacing this phase keeps restart + // reconciliation and the live reducer aligned on the same state. + if job.ProvisionalExternalSpend.IsSome() { + return PhaseExternalSpendObserved + } + switch { case job.PlannerState.Sweep.Status == unrollplan.SweepStatusConfirmed: return PhaseCompleted @@ -207,6 +226,9 @@ func stateJob(state State) *JobState { case *AwaitingSweepConfirmation: return s.Job.Copy() + case *AwaitingExternalSpendFinality: + return s.Job.Copy() + case *Completed: return s.Job.Copy()