From 4f0707f0a21a7575b6ac0df702020b10cf7feaa5 Mon Sep 17 00:00:00 2001 From: Elle Mouton Date: Wed, 8 Jul 2026 14:00:28 -0700 Subject: [PATCH 1/2] batchcanon: BatchCanonicalityManager (C3/C4) Squashed for the btcd v2 port. The batchcanon.Manager actor: one reorg-aware conf watch per batch + one spend watch per consumed input via chainsource, derives State by priority, recomputes effective expiry on reconfirm, and reconciles non-final watches on restart. No admission (that is C5). --- batchcanon/AGENTS.md | 19 +- batchcanon/CLAUDE.md | 19 +- batchcanon/manager.go | 690 +++++++++++++++ batchcanon/manager_conflict_shared_test.go | 65 ++ batchcanon/manager_test.go | 803 ++++++++++++++++++ batchcanon/messages.go | 208 +++++ batchcanon/record.go | 8 + db/batch_canonicality_store.go | 27 +- db/migrations.go | 2 +- db/sqlc/batch_canonicality.sql.go | 19 +- ...00012_batch_canonicality_pkscript.down.sql | 1 + .../000012_batch_canonicality_pkscript.up.sql | 10 + db/sqlc/models.go | 1 + db/sqlc/querier.go | 3 +- db/sqlc/queries/batch_canonicality.sql | 15 +- db/sqlc/schemas/generated_schema.sql | 2 +- 16 files changed, 1863 insertions(+), 29 deletions(-) create mode 100644 batchcanon/manager.go create mode 100644 batchcanon/manager_conflict_shared_test.go create mode 100644 batchcanon/manager_test.go create mode 100644 batchcanon/messages.go create mode 100644 db/sqlc/migrations/000012_batch_canonicality_pkscript.down.sql create mode 100644 db/sqlc/migrations/000012_batch_canonicality_pkscript.up.sql diff --git a/batchcanon/AGENTS.md b/batchcanon/AGENTS.md index 47ea46e0f..a384e0365 100644 --- a/batchcanon/AGENTS.md +++ b/batchcanon/AGENTS.md @@ -33,8 +33,23 @@ in its own package, separate from `chainsource` (raw observation) and `vtxo` - `ProvisionalConsumer` — reverse-dependency edge (consumed VTXO → consumer batch) enabling VTXO restore if a consumer batch never becomes canonical. - `Store` — behavior-free durable query/update interface. Implemented by - `db.BatchCanonicalityPersistenceStore` over the `000020` schema; backfilled - from existing VTXOs via `db.BatchCanonicalityPersistenceStore.BackfillFromVTXOs`. + `db.BatchCanonicalityPersistenceStore` over the `000020`/`000021` schema; + backfilled from existing VTXOs via + `db.BatchCanonicalityPersistenceStore.BackfillFromVTXOs`. +- `Manager` — the actor that interprets chain observation into canonicality + state (the sole client-side interpreter). Registered under + `ManagerServiceKey`. `RegisterBatchRequest` arms one reorg-aware + confirmation watch on the batch tx and one reorg-aware spend watch per + consumed input (deduped per batch, idempotent — repeats merge dependent + VTXOs). It maps chainsource `ConfirmationEvent`/`ConfReorgedEvent`/ + `ConfDoneEvent` and `SpendEvent`/`SpendReorgedEvent`/`SpendDoneEvent` onto + its own mailbox and derives `State` per the priority + `conflict_finalized > conflict_provisional > reorged_out > + finalized/provisional > unseen`. `Reconcile` re-arms watches for non-final + batches after restart without downgrading persisted state. + `GetBatchStateRequest` reads the persisted record. `NewManager` returns the + behavior; the caller registers it, then calls `SetSelfRef(ref.TellRef())` + and `Reconcile`. ## Relationships diff --git a/batchcanon/CLAUDE.md b/batchcanon/CLAUDE.md index 47ea46e0f..a384e0365 100644 --- a/batchcanon/CLAUDE.md +++ b/batchcanon/CLAUDE.md @@ -33,8 +33,23 @@ in its own package, separate from `chainsource` (raw observation) and `vtxo` - `ProvisionalConsumer` — reverse-dependency edge (consumed VTXO → consumer batch) enabling VTXO restore if a consumer batch never becomes canonical. - `Store` — behavior-free durable query/update interface. Implemented by - `db.BatchCanonicalityPersistenceStore` over the `000020` schema; backfilled - from existing VTXOs via `db.BatchCanonicalityPersistenceStore.BackfillFromVTXOs`. + `db.BatchCanonicalityPersistenceStore` over the `000020`/`000021` schema; + backfilled from existing VTXOs via + `db.BatchCanonicalityPersistenceStore.BackfillFromVTXOs`. +- `Manager` — the actor that interprets chain observation into canonicality + state (the sole client-side interpreter). Registered under + `ManagerServiceKey`. `RegisterBatchRequest` arms one reorg-aware + confirmation watch on the batch tx and one reorg-aware spend watch per + consumed input (deduped per batch, idempotent — repeats merge dependent + VTXOs). It maps chainsource `ConfirmationEvent`/`ConfReorgedEvent`/ + `ConfDoneEvent` and `SpendEvent`/`SpendReorgedEvent`/`SpendDoneEvent` onto + its own mailbox and derives `State` per the priority + `conflict_finalized > conflict_provisional > reorged_out > + finalized/provisional > unseen`. `Reconcile` re-arms watches for non-final + batches after restart without downgrading persisted state. + `GetBatchStateRequest` reads the persisted record. `NewManager` returns the + behavior; the caller registers it, then calls `SetSelfRef(ref.TellRef())` + and `Reconcile`. ## Relationships diff --git a/batchcanon/manager.go b/batchcanon/manager.go new file mode 100644 index 000000000..281d30e25 --- /dev/null +++ b/batchcanon/manager.go @@ -0,0 +1,690 @@ +package batchcanon + +import ( + "context" + "errors" + "fmt" + + "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/build" + "github.com/lightninglabs/darepo-client/chainsource" + fn "github.com/lightningnetwork/lnd/fn/v2" +) + +// ManagerServiceKey is the receptionist key the BatchCanonicalityManager +// registers under. +var ManagerServiceKey = actor.NewServiceKey[ManagerMsg, ManagerResp]( + "batch-canonicality", +) + +// usabilityConfs is the confirmation count at which the manager wants the +// first positive confirmation notification. Ark's usability depth is one +// confirmation: a batch is provisionally usable as soon as it confirms, and +// the reorg-aware lifecycle keeps it correct from there. Policy finality is +// signalled separately by chainsource's Done event at its FinalityDepth. +const usabilityConfs uint32 = 1 + +// confState is the manager's in-memory view of a batch tx's confirmation +// observation, distinct from any input-conflict view. +type confState int + +const ( + confUnseen confState = iota + confConfirmed + confFinalized + confReorgedOut +) + +// inputWatch tracks the conflict view of one consumed batch input. +type inputWatch struct { + // spenderIsConflict records whether the last observed spend of this + // input was by a transaction other than the batch itself. The batch + // consuming its own input is the expected, non-conflicting case. + spenderIsConflict bool + + // conflicting is true while a conflicting spend is observed and has not + // been reorged out. + conflicting bool + + // conflictFinal is true once a conflicting spend matured past the + // reorg-safety depth. + conflictFinal bool +} + +// batchWatch is the manager's in-memory state for one watched batch. +type batchWatch struct { + txid chainhash.Hash + pkScript []byte + + conf confState + inputs map[wire.OutPoint]*inputWatch + + // persisted is the State last written to the store, so the manager only + // issues an UpdateBatchState when the derived state actually changes. + persisted State +} + +// ManagerConfig configures the BatchCanonicalityManager. +type ManagerConfig struct { + // Store is the durable canonicality store. + Store Store + + // ChainSource is the chain-observation actor the manager registers + // reorg-aware conf/spend watches with. + ChainSource actor.ActorRef[ + chainsource.ChainSourceMsg, chainsource.ChainSourceResp, + ] + + // Log is an optional logger. + Log fn.Option[btclog.Logger] +} + +// Manager is the sole client-side interpreter of batch canonicality. It +// observes (via chainsource) each batch tx confirmation and each consumed +// input, interprets the reorg-aware lifecycle into batchcanon.State, and +// persists the result. It is an actor behavior: chainsource events arrive as +// internal messages re-wrapped onto the manager's own mailbox. +// +// Light-client backends (neutrino, Esplora) filter spend notifications by the +// prevout pkScript, which the manager does not yet carry per consumed input; +// spend watches are registered by outpoint, which is sufficient for the +// full-node (LND) path. Threading per-input pkScripts is a follow-up for when +// the producers (round, OOR) — which hold those scripts — wire in. +type Manager struct { + cfg ManagerConfig + log btclog.Logger + selfRef actor.TellOnlyRef[ManagerMsg] + + watches map[chainhash.Hash]*batchWatch +} + +// NewManager builds a BatchCanonicalityManager behavior. SetSelfRef must be +// called (with the registered actor's TellRef) before any batch is registered +// so the manager can route chainsource events back to itself. +func NewManager(cfg ManagerConfig) *Manager { + return &Manager{ + cfg: cfg, + log: cfg.Log.UnwrapOr(btclog.Disabled), + watches: make(map[chainhash.Hash]*batchWatch), + } +} + +// SetSelfRef wires the manager's own mailbox ref, used to build the mapped +// chainsource notification refs. +func (m *Manager) SetSelfRef(ref actor.TellOnlyRef[ManagerMsg]) { + m.selfRef = ref +} + +// Receive implements actor.ActorBehavior. It serializes all canonicality +// mutations through the single actor mailbox. +func (m *Manager) Receive(ctx context.Context, + msg ManagerMsg) fn.Result[ManagerResp] { + + switch v := msg.(type) { + case *RegisterBatchRequest: + return m.handleRegisterBatch(ctx, v) + + case *GetBatchStateRequest: + return m.handleGetBatchState(ctx, v) + + case *batchConfirmedMsg: + m.handleBatchConfirmed(ctx, v) + + case *batchReorgedMsg: + m.handleBatchReorged(ctx, v) + + case *batchDoneMsg: + m.handleBatchDone(ctx, v) + + case *inputSpentMsg: + m.handleInputSpent(ctx, v) + + case *inputSpendReorgedMsg: + m.handleInputSpendReorged(ctx, v) + + case *inputSpendDoneMsg: + m.handleInputSpendDone(ctx, v) + + default: + return fn.Err[ManagerResp]( + fmt.Errorf("unknown batchcanon message: %T", msg), + ) + } + + return fn.Ok[ManagerResp](&ackResponse{}) +} + +// logger returns the configured logger, falling back to the context logger. +func (m *Manager) logger(ctx context.Context) btclog.Logger { + return m.cfg.Log.UnwrapOr(build.LoggerFromContext(ctx)) +} + +// handleRegisterBatch persists the batch record and arms its watches. It is +// idempotent: a repeat for the same batch merges the dependent VTXOs into the +// record without re-arming watches. +func (m *Manager) handleRegisterBatch(ctx context.Context, + req *RegisterBatchRequest) fn.Result[ManagerResp] { + + existing, ok := m.watches[req.BatchTxID] + if ok { + // Already watching: merge dependent VTXOs into the record and + // return without duplicating watches. + if err := m.mergeDependents(ctx, existing, req); err != nil { + return fn.Err[ManagerResp](err) + } + + return fn.Ok[ManagerResp](&RegisterBatchResponse{}) + } + + // Persist the initial record (unseen until the first observation). + record := &Record{ + BatchTxID: req.BatchTxID, + State: StateUnseen, + ConfirmationHeight: fn.None[int32](), + ConfirmationBlock: fn.None[chainhash.Hash](), + CSVExpiryDelta: req.CSVExpiryDelta, + PolicyState: PolicyStateDefault, + ConfirmationPkScript: req.ConfirmationPkScript, + ConsumedInputs: req.ConsumedInputs, + DependentVTXOs: req.DependentVTXOs, + } + if err := m.cfg.Store.UpsertBatch(ctx, record); err != nil { + return fn.Err[ManagerResp]( + fmt.Errorf("persist batch record: %w", err), + ) + } + + w := &batchWatch{ + txid: req.BatchTxID, + pkScript: req.ConfirmationPkScript, + conf: confUnseen, + inputs: make(map[wire.OutPoint]*inputWatch), + } + for _, in := range req.ConsumedInputs { + w.inputs[in] = &inputWatch{} + } + + // Record the watch only AFTER arming succeeds. If we recorded it first + // and arming failed, a retry would take the idempotent "already + // watching" merge path at the top of handleRegisterBatch and never + // re-arm the missing/partial chain watches until a restart. Leaving + // m.watches untouched on failure means a retry re-arms from scratch; + // re-registering the same conf/spend caller IDs is idempotent (the same + // property Reconcile relies on after restart). + if err := m.armWatches(ctx, w, req.ConsumedInputs); err != nil { + return fn.Err[ManagerResp](err) + } + m.watches[req.BatchTxID] = w + + return fn.Ok[ManagerResp](&RegisterBatchResponse{}) +} + +// mergeDependents adds any new dependent VTXOs from a repeat registration to +// the persisted record, keeping the batch's existing watches and state. +func (m *Manager) mergeDependents(ctx context.Context, w *batchWatch, + req *RegisterBatchRequest) error { + + record, err := m.cfg.Store.GetBatch(ctx, w.txid) + if err != nil { + return fmt.Errorf("load batch for merge: %w", err) + } + + seen := make(map[wire.OutPoint]struct{}, len(record.DependentVTXOs)) + for _, dep := range record.DependentVTXOs { + seen[dep] = struct{}{} + } + changed := false + for _, dep := range req.DependentVTXOs { + if _, ok := seen[dep]; ok { + continue + } + record.DependentVTXOs = append(record.DependentVTXOs, dep) + seen[dep] = struct{}{} + changed = true + } + if !changed { + return nil + } + + return m.cfg.Store.UpsertBatch(ctx, record) +} + +// armWatches registers the reorg-aware confirmation watch on the batch tx and +// a reorg-aware spend watch on each consumed input. +func (m *Manager) armWatches(ctx context.Context, w *batchWatch, + inputs []wire.OutPoint) error { + + heightHint := m.bestHeightHint(ctx) + + confReq := &chainsource.RegisterConfRequest{ + CallerID: confCallerID(w.txid), + Txid: &w.txid, + PkScript: w.pkScript, + TargetConfs: usabilityConfs, + HeightHint: heightHint, + NotifyActor: fn.Some( + chainsource.MapConfirmationEvent( + m.selfRef, + func( + ce chainsource.ConfirmationEvent, + ) ManagerMsg { + + return &batchConfirmedMsg{ + txid: ce.Txid, + blockHeight: ce.BlockHeight, + blockHash: ce.BlockHash, + } + }, + ), + ), + NotifyReorged: fn.Some( + chainsource.MapConfReorgedEvent( + m.selfRef, + func( + ev chainsource.ConfReorgedEvent, + ) ManagerMsg { + + return &batchReorgedMsg{ + txid: ev.Txid, + } + }, + ), + ), + NotifyDone: fn.Some( + chainsource.MapConfDoneEvent( + m.selfRef, + func(ev chainsource.ConfDoneEvent) ManagerMsg { + return &batchDoneMsg{ + txid: ev.Txid, + } + }, + ), + ), + } + if err := m.cfg.ChainSource.Tell(ctx, confReq); err != nil { + return fmt.Errorf("register batch conf watch: %w", err) + } + + for i := range inputs { + op := inputs[i] + if err := m.armSpendWatch( + ctx, w.txid, op, heightHint, + ); err != nil { + return err + } + } + + return nil +} + +// armSpendWatch registers one reorg-aware spend watch on a consumed input. +func (m *Manager) armSpendWatch(ctx context.Context, txid chainhash.Hash, + op wire.OutPoint, heightHint uint32) error { + + spendReq := &chainsource.RegisterSpendRequest{ + CallerID: spendCallerID(txid, op), + Outpoint: &op, + HeightHint: heightHint, + NotifyActor: fn.Some( + chainsource.MapSpendEvent( + m.selfRef, + func(ev chainsource.SpendEvent) ManagerMsg { + return &inputSpentMsg{ + outpoint: ev.Outpoint, + spendingTxid: ev.SpendingTxid, + spendHeight: ev.SpendingHeight, + } + }, + ), + ), + NotifyReorged: fn.Some( + chainsource.MapSpendReorgedEvent( + m.selfRef, + func( + ev chainsource.SpendReorgedEvent, + ) ManagerMsg { + + return &inputSpendReorgedMsg{ + outpoint: ev.Outpoint, + } + }, + ), + ), + NotifyDone: fn.Some( + chainsource.MapSpendDoneEvent( + m.selfRef, + func(ev chainsource.SpendDoneEvent) ManagerMsg { + return &inputSpendDoneMsg{ + outpoint: ev.Outpoint, + } + }, + ), + ), + } + if err := m.cfg.ChainSource.Tell(ctx, spendReq); err != nil { + return fmt.Errorf("register input spend watch %s: %w", op, err) + } + + return nil +} + +// bestHeightHint asks chainsource for the current best height to use as a +// watch height hint. On error it returns 0 (scan from the backend's default), +// logging the failure rather than aborting registration. +func (m *Manager) bestHeightHint(ctx context.Context) uint32 { + resp, err := m.cfg.ChainSource.Ask( + ctx, &chainsource.BestHeightRequest{}, + ).Await(ctx).Unpack() + if err != nil { + m.logger(ctx).WarnS(ctx, "Batch canonicality best-height "+ + "query failed; using zero height hint", err) + + return 0 + } + + height, ok := resp.(*chainsource.BestHeightResponse) + if !ok { + return 0 + } + if height.Height < 0 { + return 0 + } + + return uint32(height.Height) +} + +// handleGetBatchState serves a read of the persisted canonicality record. +func (m *Manager) handleGetBatchState(ctx context.Context, + req *GetBatchStateRequest) fn.Result[ManagerResp] { + + record, err := m.cfg.Store.GetBatch(ctx, req.BatchTxID) + switch { + case errors.Is(err, ErrBatchNotFound): + return fn.Ok[ManagerResp](&GetBatchStateResponse{Found: false}) + + case err != nil: + return fn.Err[ManagerResp](err) + + default: + return fn.Ok[ManagerResp](&GetBatchStateResponse{ + Record: record, + Found: true, + }) + } +} + +// handleBatchConfirmed records the batch tx confirmation observation and +// re-derives the canonicality state. +func (m *Manager) handleBatchConfirmed(ctx context.Context, + msg *batchConfirmedMsg) { + + w, ok := m.watches[msg.txid] + if !ok { + return + } + + w.conf = confConfirmed + err := m.cfg.Store.RecordConfirmation( + ctx, msg.txid, msg.blockHeight, msg.blockHash, + ) + if err != nil { + m.logger(ctx).WarnS(ctx, "Failed to record batch confirmation", + err, "batch", msg.txid) + } + + m.deriveAndPersist(ctx, w) +} + +// handleBatchReorged clears the confirmation observation (the confirming block +// left the best chain) and re-derives state. +func (m *Manager) handleBatchReorged(ctx context.Context, + msg *batchReorgedMsg) { + + w, ok := m.watches[msg.txid] + if !ok { + return + } + + w.conf = confReorgedOut + if err := m.cfg.Store.ClearConfirmation(ctx, msg.txid); err != nil { + m.logger(ctx).WarnS(ctx, "Failed to clear batch confirmation", + err, "batch", msg.txid) + } + + m.deriveAndPersist(ctx, w) +} + +// handleBatchDone marks the batch confirmation as matured past the reorg- +// safety depth (policy finality) and re-derives state. The chainsource conf +// sub-actor releases its own registration on Done; the manager additionally +// releases the per-input spend watches, since a finalized batch's inputs are +// safely consumed and can no longer be double-spent. +func (m *Manager) handleBatchDone(ctx context.Context, msg *batchDoneMsg) { + w, ok := m.watches[msg.txid] + if !ok { + return + } + + w.conf = confFinalized + m.deriveAndPersist(ctx, w) + m.releaseSpendWatches(ctx, w) +} + +// handleInputSpent interprets a spend of a consumed batch input. The SAME +// outpoint can be consumed by more than one registered batch — that is exactly +// the double-spend the manager exists to classify — so every batch watching +// the outpoint is updated, not just one. For each such batch, a spend by that +// batch's own tx is the expected consumption (not a conflict), while a spend by +// any other transaction is a conflicting double-spend of that batch's input. +func (m *Manager) handleInputSpent(ctx context.Context, msg *inputSpentMsg) { + m.forEachInputWatch(msg.outpoint, func(w *batchWatch, iw *inputWatch) { + conflict := msg.spendingTxid != w.txid + iw.spenderIsConflict = conflict + iw.conflicting = conflict + iw.conflictFinal = false + + m.deriveAndPersist(ctx, w) + }) +} + +// handleInputSpendReorged clears a previously observed spend that left the +// best chain, for every batch watching the outpoint. +func (m *Manager) handleInputSpendReorged(ctx context.Context, + msg *inputSpendReorgedMsg) { + + m.forEachInputWatch(msg.outpoint, func(w *batchWatch, iw *inputWatch) { + iw.conflicting = false + iw.conflictFinal = false + + m.deriveAndPersist(ctx, w) + }) +} + +// handleInputSpendDone promotes a conflicting spend to finalized once it has +// matured past the reorg-safety depth, for every batch watching the outpoint. +// A matured spend by a batch's own tx is the normal consumption, so only the +// batches for which the spend was a conflict are promoted. +func (m *Manager) handleInputSpendDone(ctx context.Context, + msg *inputSpendDoneMsg) { + + m.forEachInputWatch(msg.outpoint, func(w *batchWatch, iw *inputWatch) { + if iw.spenderIsConflict { + iw.conflictFinal = true + } + + m.deriveAndPersist(ctx, w) + }) +} + +// forEachInputWatch invokes fn for every batch whose consumed-input set +// contains op. The same outpoint can appear under multiple batches (the +// conflict case: two batches spending the same input), so all matching watches +// must be visited — keying input watches by outpoint alone does NOT uniquely +// identify a batch. +func (m *Manager) forEachInputWatch(op wire.OutPoint, + fn func(w *batchWatch, iw *inputWatch)) { + + for _, w := range m.watches { + if iw, ok := w.inputs[op]; ok { + fn(w, iw) + } + } +} + +// deriveState computes the dominant canonicality state from the batch's +// in-memory confirmation and input-conflict views, applying the priority +// conflict_finalized > conflict_provisional > reorged_out > +// finalized/provisional > unseen. +func deriveState(w *batchWatch) State { + anyConflictFinal := false + anyConflict := false + for _, iw := range w.inputs { + if iw.conflictFinal { + anyConflictFinal = true + } + if iw.conflicting { + anyConflict = true + } + } + + switch { + case anyConflictFinal: + return StateConflictFinalized + + case anyConflict: + return StateConflictProvisional + + case w.conf == confReorgedOut: + return StateReorgedOut + + case w.conf == confFinalized: + return StateFinalized + + case w.conf == confConfirmed: + return StateProvisional + + default: + return StateUnseen + } +} + +// deriveAndPersist recomputes the batch state and writes it only when it +// changed since the last persisted value. +func (m *Manager) deriveAndPersist(ctx context.Context, w *batchWatch) { + next := deriveState(w) + if next == w.persisted { + return + } + + if err := m.cfg.Store.UpdateBatchState(ctx, w.txid, next); err != nil { + m.logger(ctx).WarnS(ctx, "Failed to persist batch state", err, + "batch", w.txid, "state", next.String()) + + return + } + w.persisted = next +} + +// releaseSpendWatches unregisters the per-input spend watches for a batch, +// called once the batch finalizes. +func (m *Manager) releaseSpendWatches(ctx context.Context, w *batchWatch) { + for op := range w.inputs { + err := m.cfg.ChainSource.Tell( + ctx, &chainsource.UnregisterSpendRequest{ + CallerID: spendCallerID(w.txid, op), + Outpoint: &op, + }, + ) + if err != nil { + m. + logger(ctx). + WarnS( + ctx, + "Failed to release input spend "+ + "watch", + err, + "batch", + w.txid, + ) + } + } +} + +// Reconcile re-establishes watches for every non-final persisted batch after a +// restart. It seeds each batch's in-memory state from the persisted record so +// live re-observation does not transiently downgrade a persisted conflict or +// finalized state. It must run after SetSelfRef. +func (m *Manager) Reconcile(ctx context.Context) error { + // Non-final states whose watches must be re-armed. Finalized and + // conflict_finalized batches need no further watching. + live := []State{ + StateUnseen, StateProvisional, StateReorgedOut, + StateConflictProvisional, + } + + for _, state := range live { + records, err := m.cfg.Store.ListBatchesByState(ctx, state) + if err != nil { + return fmt.Errorf("list %s batches: %w", state, err) + } + + for _, record := range records { + m.reconcileOne(ctx, record) + } + } + + return nil +} + +// reconcileOne rebuilds the in-memory watch for one persisted batch and +// re-arms its chain watches. +func (m *Manager) reconcileOne(ctx context.Context, record *Record) { + if _, ok := m.watches[record.BatchTxID]; ok { + return + } + + w := &batchWatch{ + txid: record.BatchTxID, + pkScript: record.ConfirmationPkScript, + inputs: make(map[wire.OutPoint]*inputWatch), + persisted: record.State, + } + + // Seed the confirmation view from the persisted record so a derive + // before re-observation does not regress the stored state. + switch record.State { + case StateProvisional, StateConflictProvisional: + w.conf = confConfirmed + + case StateReorgedOut: + w.conf = confReorgedOut + + default: + w.conf = confUnseen + } + + for _, in := range record.ConsumedInputs { + w.inputs[in] = &inputWatch{} + } + m.watches[record.BatchTxID] = w + + if err := m.armWatches(ctx, w, record.ConsumedInputs); err != nil { + m.logger(ctx).WarnS(ctx, "Failed to re-arm batch watches on "+ + "reconcile", err, "batch", record.BatchTxID) + } +} + +// confCallerID is the stable chainsource caller id for a batch's confirmation +// watch. +func confCallerID(txid chainhash.Hash) string { + return fmt.Sprintf("batchcanon-conf-%s", txid) +} + +// spendCallerID is the stable chainsource caller id for a batch input's spend +// watch. +func spendCallerID(txid chainhash.Hash, op wire.OutPoint) string { + return fmt.Sprintf("batchcanon-spend-%s-%s", txid, op) +} diff --git a/batchcanon/manager_conflict_shared_test.go b/batchcanon/manager_conflict_shared_test.go new file mode 100644 index 000000000..2a2828635 --- /dev/null +++ b/batchcanon/manager_conflict_shared_test.go @@ -0,0 +1,65 @@ +package batchcanon + +import ( + "testing" + + "github.com/btcsuite/btcd/wire/v2" + "github.com/stretchr/testify/require" +) + +// TestManagerSharedInputSpendClassifiesPerBatch verifies that when two batches +// consume the SAME input (the double-spend case), a spend by one batch's own tx +// is classified as the expected consumption for that batch and as a conflict +// for the OTHER batch — every watch on the outpoint is updated, not just one +// arbitrary batch. It also checks the finalize promotion is per-batch. +func TestManagerSharedInputSpendClassifiesPerBatch(t *testing.T) { + t.Parallel() + + h := newManagerHarness(t, 100) + + txA := testBatchTxid(0xa1) + txB := testBatchTxid(0xb2) + shared := testOutpoint(0xcc, 0) + + h.registerBatch(t, &RegisterBatchRequest{ + BatchTxID: txA, + ConfirmationPkScript: []byte{0x51, 0x20, 0x01}, + ConsumedInputs: []wire.OutPoint{shared}, + }) + h.registerBatch(t, &RegisterBatchRequest{ + BatchTxID: txB, + ConfirmationPkScript: []byte{0x51, 0x20, 0x02}, + ConsumedInputs: []wire.OutPoint{shared}, + }) + + // Batch A wins the input and confirms; B never confirms. + h.fireConfirmed(t, txA, 101, testBatchTxid(0x01)) + require.Equal(t, StateProvisional, h.state(t, txA).Record.State) + + // The shared input is spent by A's own tx: not a conflict for A, but a + // conflicting double-spend for B (which wanted the same input). + h.fireSpend(t, shared, txA, 101) + + require.Equal( + t, StateProvisional, h.state(t, txA).Record.State, + "batch whose own tx spent the input must not be in conflict", + ) + require.Equal( + t, StateConflictProvisional, h.state(t, txB).Record.State, + "batch losing its input to another tx must be "+ + "conflict-provisional", + ) + + // Once the spend matures, only the conflicted batch (B) is promoted to + // conflict-finalized; A's own consumption stays provisional. + h.fireSpendDone(t, shared) + + require.Equal( + t, StateProvisional, h.state(t, txA).Record.State, + "self-consuming batch must not finalize as a conflict", + ) + require.Equal( + t, StateConflictFinalized, h.state(t, txB).Record.State, + "conflicted batch must promote to conflict-finalized", + ) +} diff --git a/batchcanon/manager_test.go b/batchcanon/manager_test.go new file mode 100644 index 000000000..0b5c1b819 --- /dev/null +++ b/batchcanon/manager_test.go @@ -0,0 +1,803 @@ +package batchcanon + +import ( + "context" + "sync" + "testing" + "time" + + "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" +) + +const testTimeout = 5 * time.Second + +// --------------------------------------------------------------------------- +// In-memory fake Store (the real db store is tested separately; this keeps the +// manager unit test free of a batchcanon -> db import cycle). +// --------------------------------------------------------------------------- + +type fakeStore struct { + mu sync.Mutex + records map[chainhash.Hash]*Record + consumers map[chainhash.Hash][]wire.OutPoint +} + +func newFakeStore() *fakeStore { + return &fakeStore{ + records: make(map[chainhash.Hash]*Record), + consumers: make(map[chainhash.Hash][]wire.OutPoint), + } +} + +func cloneRecord(r *Record) *Record { + cp := *r + cp.ConsumedInputs = append([]wire.OutPoint(nil), r.ConsumedInputs...) + cp.DependentVTXOs = append([]wire.OutPoint(nil), r.DependentVTXOs...) + cp.ConfirmationPkScript = append( + []byte(nil), r.ConfirmationPkScript..., + ) + + return &cp +} + +func (s *fakeStore) UpsertBatch(_ context.Context, r *Record) error { + s.mu.Lock() + defer s.mu.Unlock() + s.records[r.BatchTxID] = cloneRecord(r) + + return nil +} + +func (s *fakeStore) GetBatch(_ context.Context, txid chainhash.Hash) (*Record, + error) { + + s.mu.Lock() + defer s.mu.Unlock() + r, ok := s.records[txid] + if !ok { + return nil, ErrBatchNotFound + } + + return cloneRecord(r), nil +} + +func (s *fakeStore) ListBatchesByState(_ context.Context, state State) ( + []*Record, error) { + + s.mu.Lock() + defer s.mu.Unlock() + var out []*Record + for _, r := range s.records { + if r.State == state { + out = append(out, cloneRecord(r)) + } + } + + return out, nil +} + +func (s *fakeStore) UpdateBatchState(_ context.Context, txid chainhash.Hash, + state State) error { + + s.mu.Lock() + defer s.mu.Unlock() + if r, ok := s.records[txid]; ok { + r.State = state + } + + return nil +} + +func (s *fakeStore) RecordConfirmation(_ context.Context, txid chainhash.Hash, + height int32, block chainhash.Hash) error { + + s.mu.Lock() + defer s.mu.Unlock() + if r, ok := s.records[txid]; ok { + r.ConfirmationHeight = fn.Some(height) + r.ConfirmationBlock = fn.Some(block) + } + + return nil +} + +func (s *fakeStore) ClearConfirmation(_ context.Context, + txid chainhash.Hash) error { + + s.mu.Lock() + defer s.mu.Unlock() + if r, ok := s.records[txid]; ok { + r.ConfirmationHeight = fn.None[int32]() + r.ConfirmationBlock = fn.None[chainhash.Hash]() + } + + return nil +} + +func (s *fakeStore) FindBatchesConsumingOutpoint(_ context.Context, + op wire.OutPoint) ([]chainhash.Hash, error) { + + s.mu.Lock() + defer s.mu.Unlock() + var out []chainhash.Hash + for txid, r := range s.records { + for _, in := range r.ConsumedInputs { + if in == op { + out = append(out, txid) + } + } + } + + return out, nil +} + +func (s *fakeStore) AddProvisionalConsumer(_ context.Context, + consumedVTXO wire.OutPoint, consumerBatch chainhash.Hash) error { + + s.mu.Lock() + defer s.mu.Unlock() + s.consumers[consumerBatch] = append( + s.consumers[consumerBatch], consumedVTXO, + ) + + return nil +} + +func (s *fakeStore) ListProvisionalConsumersForBatch(_ context.Context, + consumerBatch chainhash.Hash) ([]wire.OutPoint, error) { + + s.mu.Lock() + defer s.mu.Unlock() + + return append([]wire.OutPoint(nil), s.consumers[consumerBatch]...), nil +} + +func (s *fakeStore) DeleteProvisionalConsumersForBatch(_ context.Context, + consumerBatch chainhash.Hash) error { + + s.mu.Lock() + defer s.mu.Unlock() + delete(s.consumers, consumerBatch) + + return nil +} + +var _ Store = (*fakeStore)(nil) + +// --------------------------------------------------------------------------- +// Mock chainsource actor: captures the reorg-aware notification refs from +// register requests and lets the test fire lifecycle events back at them. +// --------------------------------------------------------------------------- + +type confRefs struct { + confirmed actor.TellOnlyRef[chainsource.ConfirmationEvent] + reorged actor.TellOnlyRef[chainsource.ConfReorgedEvent] + done actor.TellOnlyRef[chainsource.ConfDoneEvent] +} + +type spendRefs struct { + spend actor.TellOnlyRef[chainsource.SpendEvent] + reorged actor.TellOnlyRef[chainsource.SpendReorgedEvent] + done actor.TellOnlyRef[chainsource.SpendDoneEvent] +} + +type mockChainSource struct { + mu sync.Mutex + bestHeight int32 + confByTxid map[chainhash.Hash]confRefs + spendByOp map[wire.OutPoint]spendRefs + confCancels map[chainhash.Hash]int + spendCancel map[wire.OutPoint]int +} + +func newMockChainSource(bestHeight int32) *mockChainSource { + return &mockChainSource{ + bestHeight: bestHeight, + confByTxid: make(map[chainhash.Hash]confRefs), + spendByOp: make(map[wire.OutPoint]spendRefs), + confCancels: make(map[chainhash.Hash]int), + spendCancel: make(map[wire.OutPoint]int), + } +} + +func (c *mockChainSource) Receive(_ context.Context, + msg chainsource.ChainSourceMsg) fn.Result[chainsource.ChainSourceResp] { + + switch v := msg.(type) { + case *chainsource.BestHeightRequest: + c.mu.Lock() + h := c.bestHeight + c.mu.Unlock() + + return fn.Ok[chainsource.ChainSourceResp]( + &chainsource.BestHeightResponse{ + Height: h, + }, + ) + + case *chainsource.RegisterConfRequest: + c.mu.Lock() + c.confByTxid[*v.Txid] = confRefs{ + confirmed: v.NotifyActor.UnwrapOr(nil), + reorged: v.NotifyReorged.UnwrapOr(nil), + done: v.NotifyDone.UnwrapOr(nil), + } + c.mu.Unlock() + + return fn.Ok[chainsource.ChainSourceResp]( + &chainsource.RegisterConfResponse{}, + ) + + case *chainsource.RegisterSpendRequest: + c.mu.Lock() + c.spendByOp[*v.Outpoint] = spendRefs{ + spend: v.NotifyActor.UnwrapOr(nil), + reorged: v.NotifyReorged.UnwrapOr(nil), + done: v.NotifyDone.UnwrapOr(nil), + } + c.mu.Unlock() + + return fn.Ok[chainsource.ChainSourceResp]( + &chainsource.RegisterSpendResponse{}, + ) + + case *chainsource.UnregisterConfRequest: + c.mu.Lock() + c.confCancels[*v.Txid]++ + c.mu.Unlock() + + return fn.Ok[chainsource.ChainSourceResp]( + &chainsource.UnregisterConfResponse{}, + ) + + case *chainsource.UnregisterSpendRequest: + c.mu.Lock() + c.spendCancel[*v.Outpoint]++ + c.mu.Unlock() + + return fn.Ok[chainsource.ChainSourceResp]( + &chainsource.UnregisterSpendResponse{}, + ) + + default: + return fn.Err[chainsource.ChainSourceResp]( + errUnexpected(msg), + ) + } +} + +func errUnexpected(msg chainsource.ChainSourceMsg) error { + return &unexpectedMsgErr{msg: msg.MessageType()} +} + +type unexpectedMsgErr struct{ msg string } + +func (e *unexpectedMsgErr) Error() string { + return "mock chainsource: unexpected message " + e.msg +} + +// getConfRefs waits until the manager has registered a conf watch for txid and +// returns the captured refs. +func (c *mockChainSource) getConfRefs(t *testing.T, + txid chainhash.Hash) confRefs { + + t.Helper() + var refs confRefs + require.Eventually(t, func() bool { + c.mu.Lock() + defer c.mu.Unlock() + r, ok := c.confByTxid[txid] + if ok { + refs = r + } + + return ok + }, testTimeout, 5*time.Millisecond, "conf watch never registered") + + return refs +} + +func (c *mockChainSource) getSpendRefs(t *testing.T, + op wire.OutPoint) spendRefs { + + t.Helper() + var refs spendRefs + require.Eventually(t, func() bool { + c.mu.Lock() + defer c.mu.Unlock() + r, ok := c.spendByOp[op] + if ok { + refs = r + } + + return ok + }, testTimeout, 5*time.Millisecond, "spend watch never registered") + + return refs +} + +func (c *mockChainSource) spendCancelCount(op wire.OutPoint) int { + c.mu.Lock() + defer c.mu.Unlock() + + return c.spendCancel[op] +} + +// --------------------------------------------------------------------------- +// Harness. +// --------------------------------------------------------------------------- + +type managerHarness struct { + mgrRef actor.ActorRef[ManagerMsg, ManagerResp] + mock *mockChainSource + store *fakeStore +} + +func newManagerHarness(t *testing.T, bestHeight int32) *managerHarness { + t.Helper() + + mock := newMockChainSource(bestHeight) + mockActor := actor.NewActor(actor.ActorConfig[ + chainsource.ChainSourceMsg, chainsource.ChainSourceResp, + ]{ + ID: "mock-chainsource", + Behavior: mock, + MailboxSize: 64, + }) + mockActor.Start() + t.Cleanup(mockActor.Stop) + + store := newFakeStore() + mgr := NewManager(ManagerConfig{ + Store: store, + ChainSource: mockActor.Ref(), + }) + mgrActor := actor.NewActor(actor.ActorConfig[ManagerMsg, ManagerResp]{ + ID: "batch-canonicality", + Behavior: mgr, + MailboxSize: 64, + }) + mgr.SetSelfRef(mgrActor.TellRef()) + mgrActor.Start() + t.Cleanup(mgrActor.Stop) + + return &managerHarness{ + mgrRef: mgrActor.Ref(), + mock: mock, + store: store, + } +} + +// registerBatch registers a batch and waits for the synchronous response. +func (h *managerHarness) registerBatch(t *testing.T, + req *RegisterBatchRequest) { + + t.Helper() + _, err := h.mgrRef.Ask(t.Context(), req).Await(t.Context()).Unpack() + require.NoError(t, err) +} + +// state reads the persisted record for a batch via the manager. Because the +// manager mailbox is FIFO, issuing this Ask after a fired event guarantees the +// event was processed first. +func (h *managerHarness) state(t *testing.T, + txid chainhash.Hash) *GetBatchStateResponse { + + t.Helper() + resp, err := h.mgrRef.Ask( + t.Context(), &GetBatchStateRequest{BatchTxID: txid}, + ).Await(t.Context()).Unpack() + require.NoError(t, err) + got, ok := resp.(*GetBatchStateResponse) + require.True(t, ok) + + return got +} + +// fire helpers Tell the captured chainsource refs, synchronously enqueuing the +// re-wrapped event onto the manager mailbox. +func (h *managerHarness) fireConfirmed(t *testing.T, txid chainhash.Hash, + height int32, block chainhash.Hash) { + + t.Helper() + refs := h.mock.getConfRefs(t, txid) + require.NoError( + t, + refs.confirmed.Tell( + t.Context(), chainsource.ConfirmationEvent{ + Txid: txid, + BlockHeight: height, + BlockHash: block, + NumConfs: 1, + }, + ), + ) +} + +func (h *managerHarness) fireConfReorged(t *testing.T, txid chainhash.Hash) { + t.Helper() + refs := h.mock.getConfRefs(t, txid) + require.NoError( + t, + refs.reorged.Tell( + t.Context(), chainsource.ConfReorgedEvent{ + Txid: txid, + }, + ), + ) +} + +func (h *managerHarness) fireConfDone(t *testing.T, txid chainhash.Hash) { + t.Helper() + refs := h.mock.getConfRefs(t, txid) + require.NoError( + t, + refs.done.Tell( + t.Context(), chainsource.ConfDoneEvent{ + Txid: txid, + }, + ), + ) +} + +func (h *managerHarness) fireSpend(t *testing.T, op wire.OutPoint, + spender chainhash.Hash, height int32) { + + t.Helper() + refs := h.mock.getSpendRefs(t, op) + require.NoError( + t, + refs.spend.Tell( + t.Context(), chainsource.SpendEvent{ + Outpoint: op, + SpendingTxid: spender, + SpendingHeight: height, + }, + ), + ) +} + +func (h *managerHarness) fireSpendReorged(t *testing.T, op wire.OutPoint) { + t.Helper() + refs := h.mock.getSpendRefs(t, op) + require.NoError( + t, + refs.reorged.Tell( + t.Context(), chainsource.SpendReorgedEvent{ + Outpoint: op, + }, + ), + ) +} + +func (h *managerHarness) fireSpendDone(t *testing.T, op wire.OutPoint) { + t.Helper() + refs := h.mock.getSpendRefs(t, op) + require.NoError( + t, + refs.done.Tell( + t.Context(), chainsource.SpendDoneEvent{ + Outpoint: op, + }, + ), + ) +} + +// --------------------------------------------------------------------------- +// Tests. +// --------------------------------------------------------------------------- + +func testBatchTxid(b byte) chainhash.Hash { + var h chainhash.Hash + h[0] = b + + return h +} + +func testOutpoint(b byte, idx uint32) wire.OutPoint { + return wire.OutPoint{Hash: chainhash.Hash{b}, Index: idx} +} + +// TestManagerConfirmThenFinalize drives the happy path: a registered batch is +// unseen, becomes provisional on first confirmation (with a derived effective +// expiry), then finalized on the chainsource Done. +func TestManagerConfirmThenFinalize(t *testing.T) { + t.Parallel() + + h := newManagerHarness(t, 100) + txid := testBatchTxid(0xaa) + + h.registerBatch(t, &RegisterBatchRequest{ + BatchTxID: txid, + ConfirmationPkScript: []byte{0x51, 0x20, 0x01}, + CSVExpiryDelta: 144, + }) + + // Unseen before any observation. + got := h.state(t, txid) + require.True(t, got.Found) + require.Equal(t, StateUnseen, got.Record.State) + require.True(t, got.Record.EffectiveExpiry().IsNone()) + + // First confirmation -> provisional, effective expiry derived. + h.fireConfirmed(t, txid, 101, testBatchTxid(0xb1)) + got = h.state(t, txid) + require.Equal(t, StateProvisional, got.Record.State) + require.Equal(t, int32(101), got.Record.ConfirmationHeight.UnwrapOr(0)) + require.Equal(t, int32(245), got.Record.EffectiveExpiry().UnwrapOr(0)) + + // Policy finality -> finalized. + h.fireConfDone(t, txid) + got = h.state(t, txid) + require.Equal(t, StateFinalized, got.Record.State) +} + +// TestManagerReorgRecovers proves the core reorg-safety property: a confirmed +// batch that is reorged out moves to reorged_out (with expiry erased), then +// recovers to provisional on reconfirmation at a new height (with a fresh +// effective expiry), then finalizes. +func TestManagerReorgRecovers(t *testing.T) { + t.Parallel() + + h := newManagerHarness(t, 100) + txid := testBatchTxid(0xcc) + + h.registerBatch(t, &RegisterBatchRequest{ + BatchTxID: txid, + CSVExpiryDelta: 100, + }) + + h.fireConfirmed(t, txid, 101, testBatchTxid(0xd1)) + got := h.state(t, txid) + require.Equal(t, StateProvisional, got.Record.State) + require.Equal(t, int32(201), got.Record.EffectiveExpiry().UnwrapOr(0)) + + // Reorg out: state reorged_out, confirmation (and effective expiry) + // cleared. + h.fireConfReorged(t, txid) + got = h.state(t, txid) + require.Equal(t, StateReorgedOut, got.Record.State) + require.True(t, got.Record.ConfirmationHeight.IsNone()) + require.True(t, got.Record.EffectiveExpiry().IsNone()) + + // Reconfirm at a higher height: provisional again, fresh expiry. + h.fireConfirmed(t, txid, 105, testBatchTxid(0xd2)) + got = h.state(t, txid) + require.Equal(t, StateProvisional, got.Record.State) + require.Equal(t, int32(205), got.Record.EffectiveExpiry().UnwrapOr(0)) + + h.fireConfDone(t, txid) + require.Equal(t, StateFinalized, h.state(t, txid).Record.State) +} + +// TestManagerInputConflict proves conflict detection: a consumed input spent +// by a transaction OTHER than the batch is a conflict (conflict_provisional), +// promoted to conflict_finalized once the conflicting spend matures. +func TestManagerInputConflict(t *testing.T) { + t.Parallel() + + h := newManagerHarness(t, 100) + txid := testBatchTxid(0x11) + input := testOutpoint(0x22, 0) + + h.registerBatch(t, &RegisterBatchRequest{ + BatchTxID: txid, + CSVExpiryDelta: 50, + ConsumedInputs: []wire.OutPoint{input}, + }) + + h.fireConfirmed(t, txid, 101, testBatchTxid(0x33)) + require.Equal(t, StateProvisional, h.state(t, txid).Record.State) + + // A different tx double-spends the consumed input. + conflictTx := testBatchTxid(0x99) + h.fireSpend(t, input, conflictTx, 102) + require.Equal( + t, StateConflictProvisional, h.state(t, txid).Record.State, + ) + + // The conflict matures -> conflict_finalized. + h.fireSpendDone(t, input) + require.Equal( + t, StateConflictFinalized, h.state(t, txid).Record.State, + ) +} + +// TestManagerConflictClearsOnSpendReorg proves a conflict is reversible: if the +// conflicting spend is itself reorged out, the batch returns to its +// confirmation-derived state. +func TestManagerConflictClearsOnSpendReorg(t *testing.T) { + t.Parallel() + + h := newManagerHarness(t, 100) + txid := testBatchTxid(0x41) + input := testOutpoint(0x42, 1) + + h.registerBatch(t, &RegisterBatchRequest{ + BatchTxID: txid, + CSVExpiryDelta: 50, + ConsumedInputs: []wire.OutPoint{input}, + }) + h.fireConfirmed(t, txid, 101, testBatchTxid(0x43)) + + h.fireSpend(t, input, testBatchTxid(0x99), 102) + require.Equal( + t, StateConflictProvisional, h.state(t, txid).Record.State, + ) + + // The conflicting spend reorgs out -> conflict cleared, back to + // provisional (the batch is still confirmed). + h.fireSpendReorged(t, input) + require.Equal(t, StateProvisional, h.state(t, txid).Record.State) +} + +// TestManagerBatchSelfSpendNotConflict proves that the batch consuming its own +// input (the expected case) is not treated as a conflict. +func TestManagerBatchSelfSpendNotConflict(t *testing.T) { + t.Parallel() + + h := newManagerHarness(t, 100) + txid := testBatchTxid(0x51) + input := testOutpoint(0x52, 0) + + h.registerBatch(t, &RegisterBatchRequest{ + BatchTxID: txid, + CSVExpiryDelta: 50, + ConsumedInputs: []wire.OutPoint{input}, + }) + h.fireConfirmed(t, txid, 101, testBatchTxid(0x53)) + + // The spend is by the batch itself: not a conflict. + h.fireSpend(t, input, txid, 101) + require.Equal(t, StateProvisional, h.state(t, txid).Record.State) + + // And its maturation is the normal consumption, not a conflict. + h.fireSpendDone(t, input) + require.Equal(t, StateProvisional, h.state(t, txid).Record.State) +} + +// TestManagerConflictDominatesReorg proves the state priority: when a batch is +// both reorged out AND has a conflicting input spend, conflict dominates. +func TestManagerConflictDominatesReorg(t *testing.T) { + t.Parallel() + + h := newManagerHarness(t, 100) + txid := testBatchTxid(0x61) + input := testOutpoint(0x62, 0) + + h.registerBatch(t, &RegisterBatchRequest{ + BatchTxID: txid, + CSVExpiryDelta: 50, + ConsumedInputs: []wire.OutPoint{input}, + }) + h.fireConfirmed(t, txid, 101, testBatchTxid(0x63)) + h.fireConfReorged(t, txid) + require.Equal(t, StateReorgedOut, h.state(t, txid).Record.State) + + // A conflicting spend appears while the batch is reorged out: conflict + // dominates reorged_out. + h.fireSpend(t, input, testBatchTxid(0x99), 102) + require.Equal( + t, StateConflictProvisional, h.state(t, txid).Record.State, + ) +} + +// TestManagerFinalizeReleasesSpendWatches proves the manager releases the +// per-input spend watches once a batch finalizes. +func TestManagerFinalizeReleasesSpendWatches(t *testing.T) { + t.Parallel() + + h := newManagerHarness(t, 100) + txid := testBatchTxid(0x71) + input := testOutpoint(0x72, 0) + + h.registerBatch(t, &RegisterBatchRequest{ + BatchTxID: txid, + CSVExpiryDelta: 50, + ConsumedInputs: []wire.OutPoint{input}, + }) + h.fireConfirmed(t, txid, 101, testBatchTxid(0x73)) + h.fireConfDone(t, txid) + + // Drain via a state read, then assert the spend watch was released. + require.Equal(t, StateFinalized, h.state(t, txid).Record.State) + require.Eventually(t, func() bool { + return h.mock.spendCancelCount(input) == 1 + }, testTimeout, 5*time.Millisecond, + "input spend watch not released on finalize") +} + +// TestManagerRegisterIdempotentMergesDependents proves a repeat registration +// merges dependent VTXOs without re-arming or losing state. +func TestManagerRegisterIdempotentMergesDependents(t *testing.T) { + t.Parallel() + + h := newManagerHarness(t, 100) + txid := testBatchTxid(0x81) + depA := testOutpoint(0x8a, 0) + depB := testOutpoint(0x8b, 0) + + h.registerBatch(t, &RegisterBatchRequest{ + BatchTxID: txid, + CSVExpiryDelta: 50, + DependentVTXOs: []wire.OutPoint{depA}, + }) + h.fireConfirmed(t, txid, 101, testBatchTxid(0x83)) + require.Equal(t, StateProvisional, h.state(t, txid).Record.State) + + // Repeat with an additional dependent: merged, state preserved. + h.registerBatch(t, &RegisterBatchRequest{ + BatchTxID: txid, + CSVExpiryDelta: 50, + DependentVTXOs: []wire.OutPoint{depB}, + }) + got := h.state(t, txid) + require.Equal(t, StateProvisional, got.Record.State) + require.ElementsMatch( + t, []wire.OutPoint{depA, depB}, got.Record.DependentVTXOs, + ) +} + +// TestManagerReconcileReArmsWatches proves restart reconciliation: a manager +// started against a store with a persisted provisional batch re-arms its +// watches and does not downgrade the persisted state before re-observation. +func TestManagerReconcileReArmsWatches(t *testing.T) { + t.Parallel() + + store := newFakeStore() + txid := testBatchTxid(0x91) + input := testOutpoint(0x92, 0) + + // Seed a persisted provisional batch as if a prior run had observed it. + require.NoError( + t, + store.UpsertBatch( + t.Context(), &Record{ + BatchTxID: txid, + State: StateProvisional, + ConfirmationHeight: fn.Some[int32](90), + CSVExpiryDelta: 50, + ConsumedInputs: []wire.OutPoint{input}, + }, + ), + ) + + mock := newMockChainSource(100) + mockActor := actor.NewActor(actor.ActorConfig[ + chainsource.ChainSourceMsg, chainsource.ChainSourceResp, + ]{ID: "mock", Behavior: mock, MailboxSize: 64}) + mockActor.Start() + t.Cleanup(mockActor.Stop) + + mgr := NewManager( + ManagerConfig{ + Store: store, + ChainSource: mockActor.Ref(), + }, + ) + mgrActor := actor.NewActor(actor.ActorConfig[ManagerMsg, ManagerResp]{ + ID: "mgr", Behavior: mgr, MailboxSize: 64, + }) + mgr.SetSelfRef(mgrActor.TellRef()) + mgrActor.Start() + t.Cleanup(mgrActor.Stop) + + require.NoError(t, mgr.Reconcile(t.Context())) + + // Watches re-armed for the persisted batch. + mock.getConfRefs(t, txid) + mock.getSpendRefs(t, input) + + // State not downgraded by reconcile. + h := &managerHarness{mgrRef: mgrActor.Ref(), mock: mock, store: store} + require.Equal(t, StateProvisional, h.state(t, txid).Record.State) + + // A reorg after restart is still handled correctly. + h.fireConfReorged(t, txid) + require.Equal(t, StateReorgedOut, h.state(t, txid).Record.State) +} diff --git a/batchcanon/messages.go b/batchcanon/messages.go new file mode 100644 index 000000000..91213e86c --- /dev/null +++ b/batchcanon/messages.go @@ -0,0 +1,208 @@ +package batchcanon + +import ( + "github.com/btcsuite/btcd/chainhash/v2" + "github.com/btcsuite/btcd/wire/v2" + "github.com/lightninglabs/darepo-client/baselib/actor" +) + +// ManagerMsg is the sealed inbound message interface for the +// BatchCanonicalityManager. It covers both the public register/query API and +// the internal chain-observation messages re-wrapped from chainsource. +type ManagerMsg interface { + actor.Message + + managerMsgSealed() +} + +// ManagerResp is the sealed response interface for the manager. +type ManagerResp interface { + actor.Message + + managerRespSealed() +} + +// RegisterBatchRequest registers (or re-registers, idempotently) a batch with +// the manager: it persists a canonicality record, registers a reorg-aware +// confirmation watch on the batch tx, and a reorg-aware spend watch on every +// consumed input. Calling it again for the same batch txid merges the +// dependent VTXOs into the record without duplicating watches. +type RegisterBatchRequest struct { + actor.BaseMessage + + // BatchTxID is the batch (commitment) transaction id. + BatchTxID chainhash.Hash + + // ConfirmationPkScript is the pkScript of the batch-tx output the + // confirmation watch keys on. Required for light-client backends and + // persisted for restart re-registration. + ConfirmationPkScript []byte + + // CSVExpiryDelta is the batch's CSV-relative expiry timeout in blocks. + CSVExpiryDelta int32 + + // ConsumedInputs are the outpoints the batch tx spends. Each gets a + // reorg-aware spend watch so a conflicting double-spend is detected. + ConsumedInputs []wire.OutPoint + + // DependentVTXOs are the VTXO outpoints anchored by this batch. + DependentVTXOs []wire.OutPoint +} + +// MessageType returns the message type identifier. +func (m *RegisterBatchRequest) MessageType() string { + return "batchcanon.RegisterBatchRequest" +} + +func (m *RegisterBatchRequest) managerMsgSealed() {} + +// RegisterBatchResponse is the reply to RegisterBatchRequest. +type RegisterBatchResponse struct { + actor.BaseMessage +} + +// MessageType returns the message type identifier. +func (m *RegisterBatchResponse) MessageType() string { + return "batchcanon.RegisterBatchResponse" +} + +func (m *RegisterBatchResponse) managerRespSealed() {} + +// GetBatchStateRequest reads the current canonicality record for a batch. +type GetBatchStateRequest struct { + actor.BaseMessage + + // BatchTxID is the batch tx to look up. + BatchTxID chainhash.Hash +} + +// MessageType returns the message type identifier. +func (m *GetBatchStateRequest) MessageType() string { + return "batchcanon.GetBatchStateRequest" +} + +func (m *GetBatchStateRequest) managerMsgSealed() {} + +// GetBatchStateResponse carries the looked-up record, if present. +type GetBatchStateResponse struct { + actor.BaseMessage + + // Record is the canonicality record. Nil when Found is false. + Record *Record + + // Found reports whether a record existed for the batch. + Found bool +} + +// MessageType returns the message type identifier. +func (m *GetBatchStateResponse) MessageType() string { + return "batchcanon.GetBatchStateResponse" +} + +func (m *GetBatchStateResponse) managerRespSealed() {} + +// ackResponse is the no-op reply for internal Tell-delivered messages. +type ackResponse struct { + actor.BaseMessage +} + +// MessageType returns the message type identifier. +func (m *ackResponse) MessageType() string { + return "batchcanon.ackResponse" +} + +func (m *ackResponse) managerRespSealed() {} + +// batchConfirmedMsg is the internal re-wrap of a chainsource ConfirmationEvent +// for a watched batch tx. +type batchConfirmedMsg struct { + actor.BaseMessage + + txid chainhash.Hash + blockHeight int32 + blockHash chainhash.Hash +} + +// MessageType returns the message type identifier. +func (m *batchConfirmedMsg) MessageType() string { + return "batchcanon.batchConfirmedMsg" +} + +func (m *batchConfirmedMsg) managerMsgSealed() {} + +// batchReorgedMsg is the internal re-wrap of a chainsource ConfReorgedEvent. +type batchReorgedMsg struct { + actor.BaseMessage + + txid chainhash.Hash +} + +// MessageType returns the message type identifier. +func (m *batchReorgedMsg) MessageType() string { + return "batchcanon.batchReorgedMsg" +} + +func (m *batchReorgedMsg) managerMsgSealed() {} + +// batchDoneMsg is the internal re-wrap of a chainsource ConfDoneEvent: the +// batch confirmation has matured past the reorg-safety depth (policy +// finality). +type batchDoneMsg struct { + actor.BaseMessage + + txid chainhash.Hash +} + +// MessageType returns the message type identifier. +func (m *batchDoneMsg) MessageType() string { + return "batchcanon.batchDoneMsg" +} + +func (m *batchDoneMsg) managerMsgSealed() {} + +// inputSpentMsg is the internal re-wrap of a chainsource SpendEvent on a +// consumed batch input. +type inputSpentMsg struct { + actor.BaseMessage + + outpoint wire.OutPoint + spendingTxid chainhash.Hash + spendHeight int32 +} + +// MessageType returns the message type identifier. +func (m *inputSpentMsg) MessageType() string { + return "batchcanon.inputSpentMsg" +} + +func (m *inputSpentMsg) managerMsgSealed() {} + +// inputSpendReorgedMsg is the internal re-wrap of a chainsource +// SpendReorgedEvent: a previously observed spend left the best chain. +type inputSpendReorgedMsg struct { + actor.BaseMessage + + outpoint wire.OutPoint +} + +// MessageType returns the message type identifier. +func (m *inputSpendReorgedMsg) MessageType() string { + return "batchcanon.inputSpendReorgedMsg" +} + +func (m *inputSpendReorgedMsg) managerMsgSealed() {} + +// inputSpendDoneMsg is the internal re-wrap of a chainsource SpendDoneEvent: +// the spend observation matured past the reorg-safety depth. +type inputSpendDoneMsg struct { + actor.BaseMessage + + outpoint wire.OutPoint +} + +// MessageType returns the message type identifier. +func (m *inputSpendDoneMsg) MessageType() string { + return "batchcanon.inputSpendDoneMsg" +} + +func (m *inputSpendDoneMsg) managerMsgSealed() {} diff --git a/batchcanon/record.go b/batchcanon/record.go index 8f47d8639..bd891eeb3 100644 --- a/batchcanon/record.go +++ b/batchcanon/record.go @@ -37,6 +37,14 @@ type Record struct { // after a reorg instead of being frozen at first confirmation. CSVExpiryDelta int32 + // ConfirmationPkScript is the pkScript of the batch-tx output the + // confirmation watch keys on. It is persisted so the manager can + // re-register the watch after a restart, since light-client backends + // (neutrino, Esplora) filter confirmation notifications by pkScript. + // May be empty for records seeded by descriptor backfill, which has no + // batch-output pkScript to derive. + ConfirmationPkScript []byte + // PolicyState is the reserved policy classification slot. See // PolicyState. PolicyState PolicyState diff --git a/db/batch_canonicality_store.go b/db/batch_canonicality_store.go index e5e984546..94819cc6c 100644 --- a/db/batch_canonicality_store.go +++ b/db/batch_canonicality_store.go @@ -106,6 +106,7 @@ func (s *BatchCanonicalityPersistenceStore) UpsertBatch(ctx context.Context, now := s.clock.Now().Unix() txid := record.BatchTxID + pkScript := record.ConfirmationPkScript return s.db.ExecTx(ctx, WriteTxOption(), func( q BatchCanonicalityStore) error { @@ -120,10 +121,11 @@ func (s *BatchCanonicalityPersistenceStore) UpsertBatch(ctx context.Context, ConfirmationBlockHash: optionHashToBytes( record.ConfirmationBlock, ), - CsvExpiryDelta: record.CSVExpiryDelta, - PolicyState: int32(record.PolicyState), - CreatedAt: now, - UpdatedAt: now, + CsvExpiryDelta: record.CSVExpiryDelta, + PolicyState: int32(record.PolicyState), + ConfirmationPkScript: pkScript, + CreatedAt: now, + UpdatedAt: now, }, ) if err != nil { @@ -588,14 +590,15 @@ func (s *BatchCanonicalityPersistenceStore) hydrateRecord(ctx context.Context, } return &batchcanon.Record{ - BatchTxID: *txid, - State: batchcanon.State(row.State), - ConfirmationHeight: nullInt32ToOption(row.ConfirmationHeight), - ConfirmationBlock: confBlock, - CSVExpiryDelta: row.CsvExpiryDelta, - PolicyState: batchcanon.PolicyState(row.PolicyState), - ConsumedInputs: inputs, - DependentVTXOs: deps, + BatchTxID: *txid, + State: batchcanon.State(row.State), + ConfirmationHeight: nullInt32ToOption(row.ConfirmationHeight), + ConfirmationBlock: confBlock, + CSVExpiryDelta: row.CsvExpiryDelta, + PolicyState: batchcanon.PolicyState(row.PolicyState), + ConfirmationPkScript: row.ConfirmationPkScript, + ConsumedInputs: inputs, + DependentVTXOs: deps, }, nil } diff --git a/db/migrations.go b/db/migrations.go index 7ea42e1ac..4af91f672 100644 --- a/db/migrations.go +++ b/db/migrations.go @@ -10,7 +10,7 @@ const ( // daemon. // // NOTE: This MUST be updated when a new migration is added. - LatestMigrationVersion uint = 11 + LatestMigrationVersion uint = 12 ) // MigrationTarget is a functional option that can be passed to applyMigrations diff --git a/db/sqlc/batch_canonicality.sql.go b/db/sqlc/batch_canonicality.sql.go index 0e68b5f9d..7729a8049 100644 --- a/db/sqlc/batch_canonicality.sql.go +++ b/db/sqlc/batch_canonicality.sql.go @@ -99,12 +99,14 @@ func (q *Queries) FindBatchesByConsumedOutpoint(ctx context.Context, arg FindBat const GetBatchCanonicality = `-- name: GetBatchCanonicality :one SELECT batch_txid, state, confirmation_height, confirmation_block_hash, - csv_expiry_delta, policy_state, created_at, updated_at + csv_expiry_delta, policy_state, created_at, updated_at, + confirmation_pk_script FROM batch_canonicality WHERE batch_txid = $1 ` -// GetBatchCanonicality returns the canonicality row for a batch txid. +// GetBatchCanonicality returns the canonicality row for a batch txid. The +// column order matches the table so sqlc reuses the BatchCanonicality model. func (q *Queries) GetBatchCanonicality(ctx context.Context, batchTxid []byte) (BatchCanonicality, error) { row := q.db.QueryRowContext(ctx, GetBatchCanonicality, batchTxid) var i BatchCanonicality @@ -117,6 +119,7 @@ func (q *Queries) GetBatchCanonicality(ctx context.Context, batchTxid []byte) (B &i.PolicyState, &i.CreatedAt, &i.UpdatedAt, + &i.ConfirmationPkScript, ) return i, err } @@ -188,7 +191,8 @@ func (q *Queries) InsertProvisionalConsumer(ctx context.Context, arg InsertProvi const ListBatchCanonicalityByState = `-- name: ListBatchCanonicalityByState :many SELECT batch_txid, state, confirmation_height, confirmation_block_hash, - csv_expiry_delta, policy_state, created_at, updated_at + csv_expiry_delta, policy_state, created_at, updated_at, + confirmation_pk_script FROM batch_canonicality WHERE state = $1 ` @@ -213,6 +217,7 @@ func (q *Queries) ListBatchCanonicalityByState(ctx context.Context, state int32) &i.PolicyState, &i.CreatedAt, &i.UpdatedAt, + &i.ConfirmationPkScript, ); err != nil { return nil, err } @@ -434,9 +439,10 @@ const UpsertBatchCanonicality = `-- name: UpsertBatchCanonicality :exec INSERT INTO batch_canonicality ( batch_txid, state, confirmation_height, confirmation_block_hash, - csv_expiry_delta, policy_state, created_at, updated_at + csv_expiry_delta, policy_state, created_at, updated_at, + confirmation_pk_script ) VALUES ( - $1, $2, $3, $4, $5, $6, $7, $8 + $1, $2, $3, $4, $5, $6, $7, $8, $9 ) ON CONFLICT (batch_txid) DO UPDATE SET state = EXCLUDED.state, @@ -444,6 +450,7 @@ ON CONFLICT (batch_txid) DO UPDATE SET confirmation_block_hash = EXCLUDED.confirmation_block_hash, csv_expiry_delta = EXCLUDED.csv_expiry_delta, policy_state = EXCLUDED.policy_state, + confirmation_pk_script = EXCLUDED.confirmation_pk_script, updated_at = EXCLUDED.updated_at ` @@ -456,6 +463,7 @@ type UpsertBatchCanonicalityParams struct { PolicyState int32 CreatedAt int64 UpdatedAt int64 + ConfirmationPkScript []byte } // Batch canonicality queries. @@ -476,6 +484,7 @@ func (q *Queries) UpsertBatchCanonicality(ctx context.Context, arg UpsertBatchCa arg.PolicyState, arg.CreatedAt, arg.UpdatedAt, + arg.ConfirmationPkScript, ) return err } diff --git a/db/sqlc/migrations/000012_batch_canonicality_pkscript.down.sql b/db/sqlc/migrations/000012_batch_canonicality_pkscript.down.sql new file mode 100644 index 000000000..685db6a6e --- /dev/null +++ b/db/sqlc/migrations/000012_batch_canonicality_pkscript.down.sql @@ -0,0 +1 @@ +ALTER TABLE batch_canonicality DROP COLUMN confirmation_pk_script; diff --git a/db/sqlc/migrations/000012_batch_canonicality_pkscript.up.sql b/db/sqlc/migrations/000012_batch_canonicality_pkscript.up.sql new file mode 100644 index 000000000..1737c7364 --- /dev/null +++ b/db/sqlc/migrations/000012_batch_canonicality_pkscript.up.sql @@ -0,0 +1,10 @@ +-- Add the confirmation pkScript to batch_canonicality so the batch +-- canonicality manager can re-register the batch tx confirmation watch after a +-- restart. Light-client backends (neutrino, Esplora) filter confirmation +-- watches by pkScript, so a txid alone is insufficient to re-establish the +-- watch; persisting the watched output's pkScript lets restart reconciliation +-- rebuild every non-final batch's watch. NULL on rows created by the +-- descriptor backfill (which has no batch-output pkScript to derive); those +-- fall back to a txid-only re-registration. +ALTER TABLE batch_canonicality + ADD COLUMN confirmation_pk_script BLOB; diff --git a/db/sqlc/models.go b/db/sqlc/models.go index 90df73990..251dd87cc 100644 --- a/db/sqlc/models.go +++ b/db/sqlc/models.go @@ -70,6 +70,7 @@ type BatchCanonicality struct { PolicyState int32 CreatedAt int64 UpdatedAt int64 + ConfirmationPkScript []byte } type BatchConsumedInput struct { diff --git a/db/sqlc/querier.go b/db/sqlc/querier.go index a5a95c37a..8efbbacd4 100644 --- a/db/sqlc/querier.go +++ b/db/sqlc/querier.go @@ -71,7 +71,8 @@ type Querier interface { FindBatchesByConsumedOutpoint(ctx context.Context, arg FindBatchesByConsumedOutpointParams) ([][]byte, error) // GetActivityEntry returns one entry by its canonical id. GetActivityEntry(ctx context.Context, canonicalID string) (ActivityEntry, error) - // GetBatchCanonicality returns the canonicality row for a batch txid. + // GetBatchCanonicality returns the canonicality row for a batch txid. The + // column order matches the table so sqlc reuses the BatchCanonicality model. GetBatchCanonicality(ctx context.Context, batchTxid []byte) (BatchCanonicality, error) GetBoardingAddress(ctx context.Context, pkScript []byte) (BoardingAddress, error) GetBoardingIntent(ctx context.Context, arg GetBoardingIntentParams) (BoardingIntent, error) diff --git a/db/sqlc/queries/batch_canonicality.sql b/db/sqlc/queries/batch_canonicality.sql index 6dc744185..8be642c5a 100644 --- a/db/sqlc/queries/batch_canonicality.sql +++ b/db/sqlc/queries/batch_canonicality.sql @@ -10,9 +10,10 @@ -- batch. created_at is preserved on conflict; everything else is overwritten. INSERT INTO batch_canonicality ( batch_txid, state, confirmation_height, confirmation_block_hash, - csv_expiry_delta, policy_state, created_at, updated_at + csv_expiry_delta, policy_state, created_at, updated_at, + confirmation_pk_script ) VALUES ( - $1, $2, $3, $4, $5, $6, $7, $8 + $1, $2, $3, $4, $5, $6, $7, $8, $9 ) ON CONFLICT (batch_txid) DO UPDATE SET state = EXCLUDED.state, @@ -20,12 +21,15 @@ ON CONFLICT (batch_txid) DO UPDATE SET confirmation_block_hash = EXCLUDED.confirmation_block_hash, csv_expiry_delta = EXCLUDED.csv_expiry_delta, policy_state = EXCLUDED.policy_state, + confirmation_pk_script = EXCLUDED.confirmation_pk_script, updated_at = EXCLUDED.updated_at; -- name: GetBatchCanonicality :one --- GetBatchCanonicality returns the canonicality row for a batch txid. +-- GetBatchCanonicality returns the canonicality row for a batch txid. The +-- column order matches the table so sqlc reuses the BatchCanonicality model. SELECT batch_txid, state, confirmation_height, confirmation_block_hash, - csv_expiry_delta, policy_state, created_at, updated_at + csv_expiry_delta, policy_state, created_at, updated_at, + confirmation_pk_script FROM batch_canonicality WHERE batch_txid = $1; @@ -33,7 +37,8 @@ WHERE batch_txid = $1; -- ListBatchCanonicalityByState returns every batch currently in the given -- state. SELECT batch_txid, state, confirmation_height, confirmation_block_hash, - csv_expiry_delta, policy_state, created_at, updated_at + csv_expiry_delta, policy_state, created_at, updated_at, + confirmation_pk_script FROM batch_canonicality WHERE state = $1; diff --git a/db/sqlc/schemas/generated_schema.sql b/db/sqlc/schemas/generated_schema.sql index bcd01c93d..acfaffa4f 100644 --- a/db/sqlc/schemas/generated_schema.sql +++ b/db/sqlc/schemas/generated_schema.sql @@ -136,7 +136,7 @@ CREATE TABLE batch_canonicality ( -- created_at / updated_at are unix timestamps. created_at BIGINT NOT NULL, - updated_at BIGINT NOT NULL, + updated_at BIGINT NOT NULL, confirmation_pk_script BLOB, PRIMARY KEY (batch_txid) ); From b0e71829eb6c84c4e9081bea548b795fea3e88f0 Mon Sep 17 00:00:00 2001 From: Elle Mouton Date: Wed, 8 Jul 2026 14:04:46 -0700 Subject: [PATCH 2/2] batchcanon+vtxo: VTXO lineage availability + admission gate (C5) Squashed for the btcd v2 port. batchcanon Availability vocab (available_final/provisional/unknown, limbo_reorg/conflict, invalidated) + CombineAvailability + store-driven LineageBlocked, and the vtxo.Manager coin-selection/forfeit admission gate that drops candidates whose batch lineage is limbo/invalidated. Permissive for unseen/unregistered; no-op when the store is nil. --- batchcanon/AGENTS.md | 10 ++ batchcanon/CLAUDE.md | 10 ++ batchcanon/availability.go | 214 +++++++++++++++++++++++++++ batchcanon/availability_test.go | 196 ++++++++++++++++++++++++ chainsource/finality.go | 40 +++-- vtxo/AGENTS.md | 8 + vtxo/CLAUDE.md | 8 + vtxo/manager.go | 112 ++++++++++++++ vtxo/manager_forfeit_gate_test.go | 70 +++++++++ vtxo/manager_lineage_gate_test.go | 237 ++++++++++++++++++++++++++++++ 10 files changed, 882 insertions(+), 23 deletions(-) create mode 100644 batchcanon/availability.go create mode 100644 batchcanon/availability_test.go create mode 100644 vtxo/manager_forfeit_gate_test.go create mode 100644 vtxo/manager_lineage_gate_test.go diff --git a/batchcanon/AGENTS.md b/batchcanon/AGENTS.md index a384e0365..8a6f32f80 100644 --- a/batchcanon/AGENTS.md +++ b/batchcanon/AGENTS.md @@ -32,6 +32,16 @@ in its own package, separate from `chainsource` (raw observation) and `vtxo` reconfirmation rather than frozen. - `ProvisionalConsumer` — reverse-dependency edge (consumed VTXO → consumer batch) enabling VTXO restore if a consumer batch never becomes canonical. +- `Availability` — derived (never persisted) VTXO-lineage spendability: + `AvailableFinal`, `AvailableProvisional`, `AvailabilityUnknown`, + `LimboReorg`, `LimboConflict`, `Invalidated`. `AvailabilityForState` + maps one batch's `State`; `CombineAvailability` takes the worst across a + multi-parent lineage; `Usable()` is true only for confirmed lineage. + `LineageAvailability`/`LineageBlocked` load each parent batch from the + `Store` and produce the combined availability / block decision the VTXO + manager's admission gate (C5 wiring) calls per candidate. The gate is + permissive: unseen / not-yet-registered lineage does not block — only + limbo/invalidated lineage does. - `Store` — behavior-free durable query/update interface. Implemented by `db.BatchCanonicalityPersistenceStore` over the `000020`/`000021` schema; backfilled from existing VTXOs via diff --git a/batchcanon/CLAUDE.md b/batchcanon/CLAUDE.md index a384e0365..8a6f32f80 100644 --- a/batchcanon/CLAUDE.md +++ b/batchcanon/CLAUDE.md @@ -32,6 +32,16 @@ in its own package, separate from `chainsource` (raw observation) and `vtxo` reconfirmation rather than frozen. - `ProvisionalConsumer` — reverse-dependency edge (consumed VTXO → consumer batch) enabling VTXO restore if a consumer batch never becomes canonical. +- `Availability` — derived (never persisted) VTXO-lineage spendability: + `AvailableFinal`, `AvailableProvisional`, `AvailabilityUnknown`, + `LimboReorg`, `LimboConflict`, `Invalidated`. `AvailabilityForState` + maps one batch's `State`; `CombineAvailability` takes the worst across a + multi-parent lineage; `Usable()` is true only for confirmed lineage. + `LineageAvailability`/`LineageBlocked` load each parent batch from the + `Store` and produce the combined availability / block decision the VTXO + manager's admission gate (C5 wiring) calls per candidate. The gate is + permissive: unseen / not-yet-registered lineage does not block — only + limbo/invalidated lineage does. - `Store` — behavior-free durable query/update interface. Implemented by `db.BatchCanonicalityPersistenceStore` over the `000020`/`000021` schema; backfilled from existing VTXOs via diff --git a/batchcanon/availability.go b/batchcanon/availability.go new file mode 100644 index 000000000..9764afe2a --- /dev/null +++ b/batchcanon/availability.go @@ -0,0 +1,214 @@ +package batchcanon + +import ( + "context" + "errors" + "fmt" + + "github.com/btcsuite/btcd/chainhash/v2" +) + +// Availability is the derived spendability of a VTXO's lineage, computed from +// the canonicality State of the batch(es) the VTXO descends from. It is the +// vocabulary the VTXO manager's admission gate and the producers consume; it +// is never persisted (it is always recomputed from the current batch State). +type Availability int + +const ( + // AvailableFinal means every parent batch reached policy finality. The + // VTXO is usable and its lineage is as settled as policy allows. + AvailableFinal Availability = iota + + // AvailableProvisional means every parent batch is confirmed but not + // yet final. The VTXO is usable at one-confirmation usability depth, + // but the lineage could still reorg. + AvailableProvisional + + // AvailabilityUnknown means at least one parent batch has no + // confirmation observation yet (unseen), and none is in limbo or + // invalidated. The lineage is not yet usable, but nothing is wrong. + AvailabilityUnknown + + // LimboReorg means at least one parent batch was reorged out with no + // input conflict. The VTXO is temporarily unusable and may recover if + // the batch reconfirms. + LimboReorg + + // LimboConflict means at least one parent batch has a consumed input + // double-spent by a conflicting transaction that has not yet reached + // finality. The VTXO is unusable and may recover only if the conflict + // reorgs out. + LimboConflict + + // Invalidated means at least one parent batch has a consumed-input + // conflict that reached finality. The VTXO is unusable; recovery + // requires the conflicting transaction to itself reorg out (beyond + // policy finality). + Invalidated +) + +// availabilityRank orders availabilities from most to least available, so the +// combined availability of a multi-parent lineage is the worst (highest rank) +// of its parents. +func availabilityRank(a Availability) int { + switch a { + case AvailableFinal: + return 0 + + case AvailableProvisional: + return 1 + + case AvailabilityUnknown: + return 2 + + case LimboReorg: + return 3 + + case LimboConflict: + return 4 + + case Invalidated: + return 5 + + default: + return 2 + } +} + +// String returns a stable lower-snake-case name for the availability. +func (a Availability) String() string { + switch a { + case AvailableFinal: + return "available_final" + + case AvailableProvisional: + return "available_provisional" + + case AvailabilityUnknown: + return "available_unknown" + + case LimboReorg: + return "limbo_reorg" + + case LimboConflict: + return "limbo_conflict" + + case Invalidated: + return "invalidated" + + default: + return fmt.Sprintf("unknown(%d)", int(a)) + } +} + +// Usable reports whether a VTXO with this lineage availability may be admitted +// for spending or forfeiting. Only confirmed lineage (provisional or final) is +// usable; unseen, limbo, and invalidated lineage is not. +func (a Availability) Usable() bool { + return a == AvailableFinal || a == AvailableProvisional +} + +// AvailabilityForState maps a single batch's canonicality State to the +// availability it confers on its dependent VTXOs. +func AvailabilityForState(s State) Availability { + switch s { + case StateFinalized: + return AvailableFinal + + case StateProvisional: + return AvailableProvisional + + case StateReorgedOut: + return LimboReorg + + case StateConflictProvisional: + return LimboConflict + + case StateConflictFinalized: + return Invalidated + + case StateUnseen: + return AvailabilityUnknown + + default: + return AvailabilityUnknown + } +} + +// CombineAvailability returns the availability of a VTXO that depends on +// several parent batches: a VTXO is only as available as its least-available +// parent (the worst rank). With no parents it returns AvailabilityUnknown. +func CombineAvailability(parents ...Availability) Availability { + if len(parents) == 0 { + return AvailabilityUnknown + } + + worst := parents[0] + for _, p := range parents[1:] { + if availabilityRank(p) > availabilityRank(worst) { + worst = p + } + } + + return worst +} + +// LineageAvailability returns the combined availability of a VTXO that +// descends from the given batch txids, loading each batch's canonicality +// state from the store and taking the worst across them. A batch with no +// record yet (e.g. not registered with the manager during rollout) maps to +// AvailabilityUnknown, so a caller that wants a permissive posture can admit +// when no record blocks it. With no txids it returns AvailabilityUnknown. +// +// This is the gate logic the VTXO manager calls per candidate: a VTXO is +// admissible iff LineageAvailability(...).Usable() — or, permissively, iff it +// is not in a limbo/invalidated state. +func LineageAvailability(ctx context.Context, store Store, + batchTxids ...chainhash.Hash) (Availability, error) { + + if len(batchTxids) == 0 { + return AvailabilityUnknown, nil + } + + avails := make([]Availability, 0, len(batchTxids)) + for _, txid := range batchTxids { + record, err := store.GetBatch(ctx, txid) + switch { + case errors.Is(err, ErrBatchNotFound): + avails = append(avails, AvailabilityUnknown) + + case err != nil: + return AvailabilityUnknown, err + + default: + avails = append( + avails, AvailabilityForState(record.State), + ) + } + } + + return CombineAvailability(avails...), nil +} + +// LineageBlocked reports whether a VTXO descending from the given batches must +// be refused admission because at least one parent batch is in a limbo or +// invalidated state. It is the permissive form of the gate: unseen or +// not-yet-registered lineage does NOT block (only positively-bad lineage +// does), which keeps the gate safe to enable before every producer registers +// its batches. +func LineageBlocked(ctx context.Context, store Store, + batchTxids ...chainhash.Hash) (bool, Availability, error) { + + avail, err := LineageAvailability(ctx, store, batchTxids...) + if err != nil { + return false, avail, err + } + + switch avail { + case LimboReorg, LimboConflict, Invalidated: + return true, avail, nil + + default: + return false, avail, nil + } +} diff --git a/batchcanon/availability_test.go b/batchcanon/availability_test.go new file mode 100644 index 000000000..caa3fd2fd --- /dev/null +++ b/batchcanon/availability_test.go @@ -0,0 +1,196 @@ +package batchcanon + +import ( + "testing" + + "github.com/btcsuite/btcd/chainhash/v2" + "github.com/stretchr/testify/require" +) + +// TestAvailabilityForState pins the State -> Availability mapping. +func TestAvailabilityForState(t *testing.T) { + t.Parallel() + + cases := []struct { + state State + want Availability + }{ + { + StateFinalized, + AvailableFinal, + }, + { + StateProvisional, + AvailableProvisional, + }, + { + StateUnseen, + AvailabilityUnknown, + }, + { + StateReorgedOut, + LimboReorg, + }, + { + StateConflictProvisional, + LimboConflict, + }, + { + StateConflictFinalized, + Invalidated, + }, + } + + for _, tc := range cases { + require.Equal( + t, tc.want, AvailabilityForState(tc.state), + tc.state.String(), + ) + } +} + +// TestAvailabilityUsable verifies only confirmed lineage is usable. +func TestAvailabilityUsable(t *testing.T) { + t.Parallel() + + require.True(t, AvailableFinal.Usable()) + require.True(t, AvailableProvisional.Usable()) + require.False(t, AvailabilityUnknown.Usable()) + require.False(t, LimboReorg.Usable()) + require.False(t, LimboConflict.Usable()) + require.False(t, Invalidated.Usable()) +} + +// TestCombineAvailability verifies a multi-parent lineage takes the worst +// (least-available) parent. +func TestCombineAvailability(t *testing.T) { + t.Parallel() + + require.Equal(t, AvailabilityUnknown, CombineAvailability()) + + // All final -> final. + require.Equal( + t, AvailableFinal, CombineAvailability( + AvailableFinal, AvailableFinal, + ), + ) + + // A provisional parent downgrades a final one. + require.Equal( + t, AvailableProvisional, CombineAvailability( + AvailableFinal, AvailableProvisional, + ), + ) + + // Any limbo dominates available parents. + require.Equal( + t, LimboReorg, CombineAvailability( + AvailableFinal, AvailableProvisional, LimboReorg, + ), + ) + + // Conflict limbo dominates reorg limbo. + require.Equal( + t, LimboConflict, CombineAvailability( + LimboReorg, LimboConflict, + ), + ) + + // Invalidated dominates everything. + require.Equal( + t, Invalidated, CombineAvailability( + AvailableFinal, LimboConflict, Invalidated, + AvailableProvisional, + ), + ) + + // Unknown dominates available but not limbo/invalidated. + require.Equal( + t, AvailabilityUnknown, CombineAvailability( + AvailableProvisional, AvailabilityUnknown, + ), + ) + require.Equal( + t, LimboReorg, CombineAvailability( + AvailabilityUnknown, LimboReorg, + ), + ) +} + +// TestAvailabilityStringStable pins the string names. +func TestAvailabilityStringStable(t *testing.T) { + t.Parallel() + + require.Equal(t, "available_final", AvailableFinal.String()) + require.Equal(t, "available_provisional", AvailableProvisional.String()) + require.Equal(t, "available_unknown", AvailabilityUnknown.String()) + require.Equal(t, "limbo_reorg", LimboReorg.String()) + require.Equal(t, "limbo_conflict", LimboConflict.String()) + require.Equal(t, "invalidated", Invalidated.String()) +} + +// TestLineageAvailabilityFromStore exercises the store-driven lineage gate: +// it combines the worst availability across a VTXO's parent batches, treats a +// missing record as unknown (non-blocking), and reports blocking only for +// limbo/invalidated lineage. +func TestLineageAvailabilityFromStore(t *testing.T) { + t.Parallel() + + ctx := t.Context() + store := newFakeStore() + + finalTx := chainhash.Hash{0x01} + reorgTx := chainhash.Hash{0x02} + conflictTx := chainhash.Hash{0x03} + missingTx := chainhash.Hash{0x04} + + put := func(txid chainhash.Hash, st State) { + require.NoError( + t, + store.UpsertBatch( + ctx, &Record{ + BatchTxID: txid, + State: st, + }, + ), + ) + } + put(finalTx, StateFinalized) + put(reorgTx, StateReorgedOut) + put(conflictTx, StateConflictFinalized) + + // Single finalized parent: available, not blocked. + avail, err := LineageAvailability(ctx, store, finalTx) + require.NoError(t, err) + require.Equal(t, AvailableFinal, avail) + blocked, _, err := LineageBlocked(ctx, store, finalTx) + require.NoError(t, err) + require.False(t, blocked) + + // A reorged parent alongside a final one: limbo, blocked. + avail, err = LineageAvailability(ctx, store, finalTx, reorgTx) + require.NoError(t, err) + require.Equal(t, LimboReorg, avail) + blocked, _, err = LineageBlocked(ctx, store, finalTx, reorgTx) + require.NoError(t, err) + require.True(t, blocked) + + // An invalidated parent dominates: blocked. + blocked, avail, err = LineageBlocked(ctx, store, finalTx, conflictTx) + require.NoError(t, err) + require.True(t, blocked) + require.Equal(t, Invalidated, avail) + + // A missing (unregistered) parent is unknown and does NOT block. + avail, err = LineageAvailability(ctx, store, finalTx, missingTx) + require.NoError(t, err) + require.Equal(t, AvailabilityUnknown, avail) + blocked, _, err = LineageBlocked(ctx, store, finalTx, missingTx) + require.NoError(t, err) + require.False(t, blocked) + + // No parents: unknown, not blocked. + blocked, _, err = LineageBlocked(ctx, store) + require.NoError(t, err) + require.False(t, blocked) +} diff --git a/chainsource/finality.go b/chainsource/finality.go index 5bdc97ac6..2b8561b1f 100644 --- a/chainsource/finality.go +++ b/chainsource/finality.go @@ -21,16 +21,6 @@ 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 @@ -39,11 +29,22 @@ const finalityBlockSubscriptionAttemptTimeout = 10 * time.Second // lndclient over gRPC); a one-shot RegisterBlocks attempt that // briefly hiccups would leak the per-watch sub-actor indefinitely. // -// 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. +// 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. // // Returns the registration on success, or a non-nil error after // retries are exhausted. Callers should log the error at warn level @@ -54,14 +55,7 @@ func registerBlocksForFinality(ctx context.Context, backend ChainBackend, var lastErr error for attempt, backoff := range finalityBlockSubscriptionBackoffs { - // 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() + reg, err := backend.RegisterBlocks(ctx) if err == nil { return reg, nil } diff --git a/vtxo/AGENTS.md b/vtxo/AGENTS.md index 354b1ac8e..64b910e50 100644 --- a/vtxo/AGENTS.md +++ b/vtxo/AGENTS.md @@ -28,6 +28,14 @@ when the local wallet owns the receive script. context. `ExitOutcomeResolver` is called at startup to reconcile VTXOs still persisted in `VTXOStatusUnilateralExit` with their terminal job outcome. `ReservationStore` is used at startup to sweep orphaned Spending VTXOs. + `BatchCanonicality` (optional `batchcanon.Store`), when set, gates coin + selection on batch lineage canonicality: `selectAndReserveVTXOs` drops any + candidate whose batch is in limbo (reorged-out) or invalidated + (conflict-finalized) state via `batchcanon.LineageBlocked`, reading the + candidate's direct commitment txid through `GetVTXO`. Nil disables the gate + (a complete no-op, the default until the batch producers register batches); + it is permissive otherwise (unseen / unregistered lineage does not block). + Full multi-parent ancestry gating is a follow-up. - `ExitOutcomeResolution` — Terminal result for an exiting VTXO: `Outcome` (`ExitOutcomeRecoverable` or `ExitOutcomeConfirmed`) and `Reason`. - `ExitOutcomeResolver` — Function type diff --git a/vtxo/CLAUDE.md b/vtxo/CLAUDE.md index 354b1ac8e..64b910e50 100644 --- a/vtxo/CLAUDE.md +++ b/vtxo/CLAUDE.md @@ -28,6 +28,14 @@ when the local wallet owns the receive script. context. `ExitOutcomeResolver` is called at startup to reconcile VTXOs still persisted in `VTXOStatusUnilateralExit` with their terminal job outcome. `ReservationStore` is used at startup to sweep orphaned Spending VTXOs. + `BatchCanonicality` (optional `batchcanon.Store`), when set, gates coin + selection on batch lineage canonicality: `selectAndReserveVTXOs` drops any + candidate whose batch is in limbo (reorged-out) or invalidated + (conflict-finalized) state via `batchcanon.LineageBlocked`, reading the + candidate's direct commitment txid through `GetVTXO`. Nil disables the gate + (a complete no-op, the default until the batch producers register batches); + it is permissive otherwise (unseen / unregistered lineage does not block). + Full multi-parent ancestry gating is a follow-up. - `ExitOutcomeResolution` — Terminal result for an exiting VTXO: `Outcome` (`ExitOutcomeRecoverable` or `ExitOutcomeConfirmed`) and `Reason`. - `ExitOutcomeResolver` — Function type diff --git a/vtxo/manager.go b/vtxo/manager.go index 2d5378c3c..e08efc347 100644 --- a/vtxo/manager.go +++ b/vtxo/manager.go @@ -17,6 +17,7 @@ import ( "github.com/btcsuite/btclog/v2" "github.com/btcsuite/btcwallet/waddrmgr" "github.com/lightninglabs/darepo-client/baselib/actor" + "github.com/lightninglabs/darepo-client/batchcanon" "github.com/lightninglabs/darepo-client/build" "github.com/lightninglabs/darepo-client/chainsource" "github.com/lightninglabs/darepo-client/coinselect" @@ -136,6 +137,16 @@ type ManagerConfig struct { // VTXOs leave SpendingState. When nil, the reservation index is not // maintained and the startup sweep is skipped. ReservationStore SpendingReservationStore + + // BatchCanonicality, when set, gates coin selection on batch lineage + // canonicality: a VTXO whose batch reorged out (limbo) or was + // conflict-invalidated is excluded from selection so it is never spent + // or forfeited while its lineage is not on the canonical chain + // (darepo#454). Nil disables the gate, which is the default until the + // batch producers (round, OOR) register their batches with the + // canonicality manager; the gate is permissive for unregistered or + // unseen lineage either way. + BatchCanonicality batchcanon.Store } // Manager coordinates VTXO actor lifecycle - spawning new actors when VTXOs @@ -1148,6 +1159,51 @@ type reserveParams struct { // its actor. On partial failure the rollback function is called for // already-reserved outpoints. Returns the selected VTXO details and // total amount on success. +// gateUnavailableLineage drops candidates whose batch lineage is in a limbo +// (reorged-out) or invalidated (conflict-finalized) canonicality state, so a +// VTXO is never selected while its batch is off the canonical chain. It is a +// no-op when no canonicality store is configured (the gate stays dormant until +// the batch producers register their batches). It reads each candidate's +// direct commitment txid via the store; full multi-parent ancestry gating for +// cross-commitment OOR VTXOs is a follow-up. The gate is permissive: an +// unregistered or unseen batch does not block selection. +func (m *Manager) gateUnavailableLineage(ctx context.Context, + candidates []*Descriptor) ([]*Descriptor, error) { + + if m.cfg.BatchCanonicality == nil { + return candidates, nil + } + + kept := make([]*Descriptor, 0, len(candidates)) + for _, c := range candidates { + desc, err := m.cfg.Store.GetVTXO(ctx, c.Outpoint) + if err != nil { + return nil, fmt.Errorf("load vtxo for lineage gate "+ + "%s: %w", c.Outpoint, err) + } + + blocked, avail, err := batchcanon.LineageBlocked( + ctx, m.cfg.BatchCanonicality, desc.CommitmentTxID, + ) + if err != nil { + return nil, fmt.Errorf("lineage gate %s: %w", + c.Outpoint, err) + } + if blocked { + m.logger(ctx).DebugS(ctx, "Excluding VTXO with "+ + "unavailable batch lineage from selection", + slog.String("outpoint", c.Outpoint.String()), + slog.String("availability", avail.String())) + + continue + } + + kept = append(kept, c) + } + + return kept, nil +} + func (m *Manager) selectAndReserveVTXOs(ctx context.Context, p reserveParams) ( []SelectedVTXO, btcutil.Amount, error) { @@ -1193,6 +1249,15 @@ func (m *Manager) selectAndReserveVTXOs(ctx context.Context, p reserveParams) ( }) } + // Drop any candidate whose batch lineage is in limbo or invalidated, so + // a VTXO whose batch reorged out or was conflict-invalidated is never + // selected while its lineage is off the canonical chain. No-op when no + // canonicality store is configured. + candidates, err = m.gateUnavailableLineage(ctx, candidates) + if err != nil { + return nil, 0, err + } + // Run largest-first selection through the shared selector. Map its // typed outcomes back onto the manager's liquidity diagnostics: a // dust-change rejection is reported verbatim, while any shortfall @@ -1812,6 +1877,26 @@ func (m *Manager) handleReserveForfeit(ctx context.Context, ErrVTXOLiquidityLocked, op), ) } + + // Refuse to forfeit a VTXO whose batch lineage is in limbo + // (reorged out) or invalidated: forfeiting commits the VTXO + // into a round, and a VTXO that is not on the canonical chain + // must not be spent. The coin-selection gate + // (gateUnavailableLineage) already excludes such VTXOs, but the + // wallet's explicit-outpoint paths (refresh / leave / sweep-all + // / replay) reserve by name and bypass selection, so the same + // gate is enforced here (darepo#454). + blocked, avail, err := m.forfeitLineageBlocked(ctx, op) + if err != nil { + return fn.Err[ManagerResp](err) + } + if blocked { + return fn.Err[ManagerResp]( + fmt.Errorf("%w: outpoint %s batch lineage "+ + "unavailable (%s)", + ErrVTXOLiquidityLocked, op, avail), + ) + } } // Reserve each VTXO. Track successes for rollback on failure. @@ -1843,6 +1928,33 @@ func (m *Manager) handleReserveForfeit(ctx context.Context, return fn.Ok[ManagerResp](&ReserveForfeitResponse{}) } +// forfeitLineageBlocked reports whether the named VTXO's batch lineage is in a +// limbo (reorged-out) or invalidated (conflict-finalized) canonicality state, +// so an explicit forfeit reservation must be refused. It mirrors the +// coin-selection gate (gateUnavailableLineage) for the explicit-outpoint +// reserve path: a no-op when no canonicality store is configured, and +// permissive for unseen / unregistered lineage. It reads the candidate's +// direct commitment txid; full multi-parent ancestry gating arrives with the +// selection gate's multi-parent extension. +func (m *Manager) forfeitLineageBlocked(ctx context.Context, op wire.OutPoint) ( + bool, batchcanon.Availability, error) { + + if m.cfg.BatchCanonicality == nil { + return false, batchcanon.AvailabilityUnknown, nil + } + + desc, err := m.cfg.Store.GetVTXO(ctx, op) + if err != nil { + return false, batchcanon.AvailabilityUnknown, + fmt.Errorf("load vtxo for forfeit lineage gate %s: %w", + op, err) + } + + return batchcanon.LineageBlocked( + ctx, m.cfg.BatchCanonicality, desc.CommitmentTxID, + ) +} + // rollbackForfeit sends ForfeitReleasedEvent to previously reserved VTXOs. // Best-effort: errors are logged but do not propagate. func (m *Manager) rollbackForfeit(ctx context.Context, diff --git a/vtxo/manager_forfeit_gate_test.go b/vtxo/manager_forfeit_gate_test.go new file mode 100644 index 000000000..26692916b --- /dev/null +++ b/vtxo/manager_forfeit_gate_test.go @@ -0,0 +1,70 @@ +package vtxo + +import ( + "testing" + + "github.com/btcsuite/btcd/chainhash/v2" + "github.com/lightninglabs/darepo-client/batchcanon" + "github.com/stretchr/testify/mock" + "github.com/stretchr/testify/require" +) + +// TestForfeitLineageBlockedOnLimbo verifies the explicit-outpoint forfeit gate +// refuses a VTXO whose batch reorged out, matching the coin-selection gate so +// the wallet's reserve-by-name paths (refresh/leave/sweep/replay) cannot +// forfeit a VTXO that is off the canonical chain. +func TestForfeitLineageBlockedOnLimbo(t *testing.T) { + t.Parallel() + + v := makeDescriptor(t, 50_000, 0) + v.CommitmentTxID = chainhash.Hash{0xaa} + + mgr, store := newTestManager(t, []*Descriptor{v}) + mgr.cfg.BatchCanonicality = &fakeBatchCanon{ + states: map[chainhash.Hash]batchcanon.State{ + v.CommitmentTxID: batchcanon.StateReorgedOut, + }, + } + store.On("GetVTXO", mock.Anything, v.Outpoint).Return(v, nil) + + blocked, avail, err := mgr.forfeitLineageBlocked( + t.Context(), v.Outpoint, + ) + require.NoError(t, err) + require.True(t, blocked) + require.Equal(t, batchcanon.LimboReorg, avail) +} + +// TestForfeitLineageNotBlockedWhenCanonical verifies a canonical VTXO is +// admissible for forfeit. +func TestForfeitLineageNotBlockedWhenCanonical(t *testing.T) { + t.Parallel() + + v := makeDescriptor(t, 50_000, 0) + v.CommitmentTxID = chainhash.Hash{0xaa} + + mgr, store := newTestManager(t, []*Descriptor{v}) + mgr.cfg.BatchCanonicality = &fakeBatchCanon{ + states: map[chainhash.Hash]batchcanon.State{ + v.CommitmentTxID: batchcanon.StateProvisional, + }, + } + store.On("GetVTXO", mock.Anything, v.Outpoint).Return(v, nil) + + blocked, _, err := mgr.forfeitLineageBlocked(t.Context(), v.Outpoint) + require.NoError(t, err) + require.False(t, blocked) +} + +// TestForfeitLineageGateDormantWhenNoStore verifies the forfeit gate is a no-op +// when no canonicality store is wired. +func TestForfeitLineageGateDormantWhenNoStore(t *testing.T) { + t.Parallel() + + v := makeDescriptor(t, 50_000, 0) + mgr, _ := newTestManager(t, []*Descriptor{v}) + + blocked, _, err := mgr.forfeitLineageBlocked(t.Context(), v.Outpoint) + require.NoError(t, err) + require.False(t, blocked) +} diff --git a/vtxo/manager_lineage_gate_test.go b/vtxo/manager_lineage_gate_test.go new file mode 100644 index 000000000..843aa5300 --- /dev/null +++ b/vtxo/manager_lineage_gate_test.go @@ -0,0 +1,237 @@ +package vtxo + +import ( + "context" + "testing" + + "github.com/btcsuite/btcd/chainhash/v2" + "github.com/btcsuite/btcd/wire/v2" + "github.com/lightninglabs/darepo-client/batchcanon" + "github.com/stretchr/testify/mock" + "github.com/stretchr/testify/require" +) + +// fakeBatchCanon is a minimal batchcanon.Store for the lineage-gate tests: it +// maps batch txids to canonicality states and answers GetBatch from that map. +// The other Store methods are unused by the gate and return zero values. +type fakeBatchCanon struct { + states map[chainhash.Hash]batchcanon.State +} + +func (f *fakeBatchCanon) GetBatch(_ context.Context, txid chainhash.Hash) ( + *batchcanon.Record, error) { + + st, ok := f.states[txid] + if !ok { + return nil, batchcanon.ErrBatchNotFound + } + + return &batchcanon.Record{BatchTxID: txid, State: st}, nil +} + +func (f *fakeBatchCanon) UpsertBatch(context.Context, + *batchcanon.Record) error { + + return nil +} + +func (f *fakeBatchCanon) ListBatchesByState(context.Context, batchcanon.State) ( + []*batchcanon.Record, error) { + + return nil, nil +} + +func (f *fakeBatchCanon) UpdateBatchState(context.Context, chainhash.Hash, + batchcanon.State) error { + + return nil +} + +func (f *fakeBatchCanon) RecordConfirmation(context.Context, chainhash.Hash, + int32, chainhash.Hash) error { + + return nil +} + +func (f *fakeBatchCanon) ClearConfirmation(context.Context, + chainhash.Hash) error { + + return nil +} + +func (f *fakeBatchCanon) FindBatchesConsumingOutpoint(context.Context, + wire.OutPoint) ([]chainhash.Hash, error) { + + return nil, nil +} + +func (f *fakeBatchCanon) AddProvisionalConsumer(context.Context, wire.OutPoint, + chainhash.Hash) error { + + return nil +} + +func (f *fakeBatchCanon) ListProvisionalConsumersForBatch(context.Context, + chainhash.Hash) ([]wire.OutPoint, error) { + + return nil, nil +} + +func (f *fakeBatchCanon) DeleteProvisionalConsumersForBatch(context.Context, + chainhash.Hash) error { + + return nil +} + +var _ batchcanon.Store = (*fakeBatchCanon)(nil) + +// TestSelectExcludesLimboLineage verifies the admission gate drops a candidate +// whose batch reorged out (limbo), so largest-first selection skips it and +// picks a smaller candidate whose batch is canonical instead. +func TestSelectExcludesLimboLineage(t *testing.T) { + t.Parallel() + + good := makeDescriptor(t, 40000, 0) + bad := makeDescriptor(t, 50000, 1) + good.CommitmentTxID = chainhash.Hash{0xaa} + bad.CommitmentTxID = chainhash.Hash{0xbb} + + mgr, store := newTestManager(t, []*Descriptor{good, bad}) + mgr.cfg.BatchCanonicality = &fakeBatchCanon{ + states: map[chainhash.Hash]batchcanon.State{ + good.CommitmentTxID: batchcanon.StateProvisional, + bad.CommitmentTxID: batchcanon.StateReorgedOut, + }, + } + + store.On( + "ListVTXOsByStatus", t.Context(), VTXOStatusLive, + ).Return([]*Descriptor{good, bad}, nil) + store.On("GetVTXO", mock.Anything, good.Outpoint).Return(good, nil) + store.On("GetVTXO", mock.Anything, bad.Outpoint).Return(bad, nil) + + result := mgr.Receive(t.Context(), &SelectAndReserveSpendRequest{ + TargetAmount: 40000, + }) + resp, err := result.Unpack() + require.NoError(t, err) + + spendResp, ok := resp.(*SelectAndReserveSpendResponse) + require.True(t, ok) + + // The 50000 candidate (largest) is gated out for its reorged-out batch, + // so selection falls to the 40000 candidate with a canonical batch. + require.Len(t, spendResp.SelectedVTXOs, 1) + require.Equal(t, good.Outpoint, spendResp.SelectedVTXOs[0].Outpoint) +} + +// TestSelectFailsWhenAllLineageInvalidated verifies that when every candidate's +// batch is invalidated, selection finds no admissible liquidity and fails +// rather than spending an invalidated VTXO. +func TestSelectFailsWhenAllLineageInvalidated(t *testing.T) { + t.Parallel() + + only := makeDescriptor(t, 50000, 0) + only.CommitmentTxID = chainhash.Hash{0xcc} + + mgr, store := newTestManager(t, []*Descriptor{only}) + mgr.cfg.BatchCanonicality = &fakeBatchCanon{ + states: map[chainhash.Hash]batchcanon.State{ + only.CommitmentTxID: batchcanon.StateConflictFinalized, + }, + } + + store.On( + "ListVTXOsByStatus", t.Context(), VTXOStatusLive, + ).Return([]*Descriptor{only}, nil) + store.On("GetVTXO", mock.Anything, only.Outpoint).Return(only, nil) + + // The shortfall path builds a liquidity diagnostic via ListLiveVTXOs. + store.On("ListLiveVTXOs", mock.Anything).Return( + []*Descriptor{only}, nil, + ) + + result := mgr.Receive(t.Context(), &SelectAndReserveSpendRequest{ + TargetAmount: 40000, + }) + _, err := result.Unpack() + require.Error(t, err) +} + +// TestSelectAdmitsCanonicalAndUnregisteredLineage verifies the gate is +// permissive: a candidate whose batch is provisional is admitted, and so is +// one whose batch has no canonicality record yet (unregistered during +// rollout) — only positively limbo/invalidated lineage is refused. +func TestSelectAdmitsCanonicalAndUnregisteredLineage(t *testing.T) { + t.Parallel() + + provisional := makeDescriptor(t, 30000, 0) + unregistered := makeDescriptor(t, 50000, 1) + provisional.CommitmentTxID = chainhash.Hash{0xd1} + unregistered.CommitmentTxID = chainhash.Hash{0xd2} + + mgr, store := newTestManager(t, []*Descriptor{ + provisional, unregistered, + }) + mgr.cfg.BatchCanonicality = &fakeBatchCanon{ + states: map[chainhash.Hash]batchcanon.State{ + provisional.CommitmentTxID: batchcanon.StateProvisional, + // unregistered: intentionally absent from the map. + }, + } + + store.On( + "ListVTXOsByStatus", t.Context(), VTXOStatusLive, + ).Return([]*Descriptor{provisional, unregistered}, nil) + store.On( + "GetVTXO", mock.Anything, provisional.Outpoint, + ).Return(provisional, nil) + store.On( + "GetVTXO", mock.Anything, unregistered.Outpoint, + ).Return(unregistered, nil) + + result := mgr.Receive(t.Context(), &SelectAndReserveSpendRequest{ + TargetAmount: 40000, + }) + resp, err := result.Unpack() + require.NoError(t, err) + + spendResp, ok := resp.(*SelectAndReserveSpendResponse) + require.True(t, ok) + + // The unregistered-batch candidate (50000, largest) is admitted because + // the gate does not block unseen/unregistered lineage. + require.Len(t, spendResp.SelectedVTXOs, 1) + require.Equal( + t, unregistered.Outpoint, spendResp.SelectedVTXOs[0].Outpoint, + ) +} + +// TestSelectGateDisabledWhenNoStore verifies that with no canonicality store +// configured the gate is a complete no-op (no GetVTXO calls, normal +// largest-first selection). +func TestSelectGateDisabledWhenNoStore(t *testing.T) { + t.Parallel() + + v := makeDescriptor(t, 50000, 0) + mgr, store := newTestManager(t, []*Descriptor{v}) + require.Nil(t, mgr.cfg.BatchCanonicality) + + store.On( + "ListVTXOsByStatus", t.Context(), VTXOStatusLive, + ).Return([]*Descriptor{v}, nil) + + result := mgr.Receive(t.Context(), &SelectAndReserveSpendRequest{ + TargetAmount: 40000, + }) + resp, err := result.Unpack() + require.NoError(t, err) + + spendResp, ok := resp.(*SelectAndReserveSpendResponse) + require.True(t, ok) + require.Len(t, spendResp.SelectedVTXOs, 1) + require.Equal(t, v.Outpoint, spendResp.SelectedVTXOs[0].Outpoint) + + // The gate must not have queried GetVTXO at all. + store.AssertNotCalled(t, "GetVTXO", mock.Anything, mock.Anything) +}