diff --git a/batchcanon/AGENTS.md b/batchcanon/AGENTS.md new file mode 100644 index 000000000..a384e0365 --- /dev/null +++ b/batchcanon/AGENTS.md @@ -0,0 +1,90 @@ +# batchcanon + +## Purpose + +Client-side **batch canonicality data model** for the reorg-safety epic +(darepo#454, task C2). Holds the durable, reorg-aware record of how each batch +(commitment) transaction is faring against the best chain: its canonicality +state, current confirmation observation, recompute inputs for effective +expiry, the inputs it consumes, the VTXOs it anchors, and the reverse +dependencies needed to restore a provisionally consumed VTXO. + +This package is **data + query/update interface only**. It contains no +interpretation, no chain watching, and no admission behavior — those belong to +the (later) `BatchCanonicalityManager` and the VTXO manager. Keeping the model +in its own package, separate from `chainsource` (raw observation) and `vtxo` +(admission), preserves the epic's observation → interpretation → action split. + +## Key Types + +- `State` — canonicality state enum: `StateUnseen`, `StateProvisional`, + `StateFinalized`, `StateReorgedOut`, `StateConflictProvisional`, + `StateConflictFinalized`. Reorg-reversible; **no state is a terminal + verdict** at this layer. Persisted as an append-only typed INTEGER column — + values must never be renumbered. +- `PolicyState` — reserved policy classification slot (`PolicyStateDefault` + only); persisted and round-tripped, no business meaning yet. +- `Record` — per-batch record keyed by `BatchTxID`. Identity is by **txid**, + never `(txid, block hash)`; `ConfirmationBlock` is an observation attribute + only. `EffectiveExpiry()` derives the absolute expiry as + `ConfirmationHeight + CSVExpiryDelta`, returning `None` when unconfirmed — + the structural guarantee that expiry is recomputed on every + reconfirmation rather than frozen. +- `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`/`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 + +- **Depends on**: `btcd/chaincfg/chainhash`, `btcd/wire`, `lnd/fn/v2` only. +- **Depended on by**: `db` (concrete store), and — in later tasks — the + batch canonicality manager and `vtxo` admission. + +## Invariants + +- Identity is by txid / outpoint, never by `(txid, block hash)`. +- Expiry is never persisted as a standalone or terminal value; it is always + derived from `CSVExpiryDelta` + the current confirmation observation. +- State enum integer values are append-only (persisted column). + +## Expiry-as-terminal audit (darepo#454 C2) + +C2 requires auditing every site that treats `BatchExpiry`/`Expired` as a +one-way terminal fact. These are flagged for rework when the +BatchCanonicalityManager (task C3/C4) rewires expiry consumers onto +`Record.EffectiveExpiry()`; **no behavior is changed by C2**: + +- `vtxo/transitions.go` (`ExpiryStatusExpired → FailedState{Recoverable: + false}`, and the Critical/Expired escalations) — the primary offender: a + reorg that lowers the confirmation height could otherwise push a VTXO + permanently into non-recoverable `Failed`. +- `vtxo/expiry.go` (`CheckExpiry`, `BlocksUntilExpiry`) — compute from the + frozen absolute `vtxo.BatchExpiry`; must consume effective (recomputable) + expiry instead. +- `vtxo/actor.go` — schedules on the frozen absolute `BatchExpiry`. +- `darepod/vhtlc_recovery_target.go` — folds multiple roots into a + most-restrictive absolute `batchExpiry`. +- `unroll/proof_assembler.go` (`BatchExpiry == 0`) — treats zero as "unset", + not terminal; benign, documented for completeness. + +## Deep Docs + +- [ARCHITECTURE.md](../ARCHITECTURE.md) — System-wide package map. + diff --git a/batchcanon/CLAUDE.md b/batchcanon/CLAUDE.md new file mode 100644 index 000000000..a384e0365 --- /dev/null +++ b/batchcanon/CLAUDE.md @@ -0,0 +1,90 @@ +# batchcanon + +## Purpose + +Client-side **batch canonicality data model** for the reorg-safety epic +(darepo#454, task C2). Holds the durable, reorg-aware record of how each batch +(commitment) transaction is faring against the best chain: its canonicality +state, current confirmation observation, recompute inputs for effective +expiry, the inputs it consumes, the VTXOs it anchors, and the reverse +dependencies needed to restore a provisionally consumed VTXO. + +This package is **data + query/update interface only**. It contains no +interpretation, no chain watching, and no admission behavior — those belong to +the (later) `BatchCanonicalityManager` and the VTXO manager. Keeping the model +in its own package, separate from `chainsource` (raw observation) and `vtxo` +(admission), preserves the epic's observation → interpretation → action split. + +## Key Types + +- `State` — canonicality state enum: `StateUnseen`, `StateProvisional`, + `StateFinalized`, `StateReorgedOut`, `StateConflictProvisional`, + `StateConflictFinalized`. Reorg-reversible; **no state is a terminal + verdict** at this layer. Persisted as an append-only typed INTEGER column — + values must never be renumbered. +- `PolicyState` — reserved policy classification slot (`PolicyStateDefault` + only); persisted and round-tripped, no business meaning yet. +- `Record` — per-batch record keyed by `BatchTxID`. Identity is by **txid**, + never `(txid, block hash)`; `ConfirmationBlock` is an observation attribute + only. `EffectiveExpiry()` derives the absolute expiry as + `ConfirmationHeight + CSVExpiryDelta`, returning `None` when unconfirmed — + the structural guarantee that expiry is recomputed on every + reconfirmation rather than frozen. +- `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`/`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 + +- **Depends on**: `btcd/chaincfg/chainhash`, `btcd/wire`, `lnd/fn/v2` only. +- **Depended on by**: `db` (concrete store), and — in later tasks — the + batch canonicality manager and `vtxo` admission. + +## Invariants + +- Identity is by txid / outpoint, never by `(txid, block hash)`. +- Expiry is never persisted as a standalone or terminal value; it is always + derived from `CSVExpiryDelta` + the current confirmation observation. +- State enum integer values are append-only (persisted column). + +## Expiry-as-terminal audit (darepo#454 C2) + +C2 requires auditing every site that treats `BatchExpiry`/`Expired` as a +one-way terminal fact. These are flagged for rework when the +BatchCanonicalityManager (task C3/C4) rewires expiry consumers onto +`Record.EffectiveExpiry()`; **no behavior is changed by C2**: + +- `vtxo/transitions.go` (`ExpiryStatusExpired → FailedState{Recoverable: + false}`, and the Critical/Expired escalations) — the primary offender: a + reorg that lowers the confirmation height could otherwise push a VTXO + permanently into non-recoverable `Failed`. +- `vtxo/expiry.go` (`CheckExpiry`, `BlocksUntilExpiry`) — compute from the + frozen absolute `vtxo.BatchExpiry`; must consume effective (recomputable) + expiry instead. +- `vtxo/actor.go` — schedules on the frozen absolute `BatchExpiry`. +- `darepod/vhtlc_recovery_target.go` — folds multiple roots into a + most-restrictive absolute `batchExpiry`. +- `unroll/proof_assembler.go` (`BatchExpiry == 0`) — treats zero as "unset", + not terminal; benign, documented for completeness. + +## Deep Docs + +- [ARCHITECTURE.md](../ARCHITECTURE.md) — System-wide package map. + diff --git a/batchcanon/doc.go b/batchcanon/doc.go new file mode 100644 index 000000000..59a5064de --- /dev/null +++ b/batchcanon/doc.go @@ -0,0 +1,26 @@ +// Package batchcanon holds the client-side batch canonicality data model: +// the durable record of how each batch (commitment) transaction is faring +// against the best chain, the inputs it consumes, the VTXOs it anchors, and +// the reverse-dependency edges needed to restore a provisionally consumed +// VTXO if its consumer batch never becomes canonical. +// +// This package is the data substrate for the reorg-safety epic +// (darepo#454). It deliberately contains NO interpretation or admission +// behavior: it persists and retrieves observations only. The +// BatchCanonicalityManager (a later task) is the sole interpreter that +// drives state transitions from chainsource observations, and the VTXO +// manager remains the admission boundary. Keeping the model here, separate +// from both chainsource (raw observation) and vtxo (admission), preserves +// the observation -> interpretation -> action split the epic mandates. +// +// Two principles shape the model: +// +// - Identity is by txid / outpoint, never by (txid, block hash). A reorg +// that re-mines the same batch tx in a different block is the SAME +// batch; the block hash is only an observation attribute. +// +// - Expiry is never stored as a terminal fact. The model stores a +// CSV-relative delta plus the current confirmation height and derives +// the effective (absolute) expiry on demand, so a reorg-and-reconfirm +// at a new height recomputes expiry instead of freezing it. +package batchcanon 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 new file mode 100644 index 000000000..bd891eeb3 --- /dev/null +++ b/batchcanon/record.go @@ -0,0 +1,90 @@ +package batchcanon + +import ( + "github.com/btcsuite/btcd/chainhash/v2" + "github.com/btcsuite/btcd/wire/v2" + fn "github.com/lightningnetwork/lnd/fn/v2" +) + +// Record is the durable canonicality view of one batch (commitment) +// transaction, keyed by its txid. It bundles the interpreted State, the +// current confirmation observation, the recompute inputs for effective +// expiry, the consumed inputs and dependent VTXOs, and the reserved policy +// slot. +type Record struct { + // BatchTxID is the commitment transaction id and the record's + // identity. Identity is by txid, never by (txid, block hash): a reorg + // that re-mines the same tx in a different block is the same batch. + BatchTxID chainhash.Hash + + // State is the interpreted canonicality state. + State State + + // ConfirmationHeight is the best-chain height at which the batch tx + // is currently observed confirmed. None when the batch is not + // currently confirmed (unseen or reorged out). A reorg clears it; a + // reconfirmation sets it to the new height. + ConfirmationHeight fn.Option[int32] + + // ConfirmationBlock is the hash of the block currently confirming the + // batch tx. It is an observation attribute only and is NOT part of + // the batch identity. None when the batch is not currently confirmed. + ConfirmationBlock fn.Option[chainhash.Hash] + + // CSVExpiryDelta is the batch's CSV-relative expiry timeout, in + // blocks. The effective (absolute) expiry height is derived from this + // plus the current confirmation height, so it tracks reconfirmations + // 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 + + // ConsumedInputs are the outpoints this batch tx spends. They are + // tracked so the manager can watch each one for a conflicting spend. + ConsumedInputs []wire.OutPoint + + // DependentVTXOs are the VTXO outpoints anchored by this batch. Their + // derived availability follows this batch's canonicality. + DependentVTXOs []wire.OutPoint +} + +// EffectiveExpiry derives the absolute expiry height from the current +// confirmation observation: ConfirmationHeight + CSVExpiryDelta. It returns +// None when the batch is not currently confirmed. +// +// Deriving expiry on demand (rather than persisting an absolute height) is +// what keeps expiry reorg-safe: a confirmation that is reorged out clears +// ConfirmationHeight and so erases the effective expiry, and a +// reconfirmation at a different height yields a fresh effective expiry. +// Expiry is therefore never a one-way terminal fact at this layer. +func (r *Record) EffectiveExpiry() fn.Option[int32] { + return fn.MapOption( + func(height int32) int32 { + return height + r.CSVExpiryDelta + })(r.ConfirmationHeight) +} + +// ProvisionalConsumer records that a (locally relevant) VTXO has been +// provisionally consumed by a not-yet-canonical consumer batch. It is the +// reverse-dependency edge that lets a provisionally consumed VTXO be restored +// if the consumer batch never becomes canonical — for example a round-2 +// forfeit whose commitment tx is reorged out, which must restore the round-1 +// VTXO it consumed. +type ProvisionalConsumer struct { + // ConsumedVTXO is the outpoint of the VTXO consumed by ConsumerBatch. + ConsumedVTXO wire.OutPoint + + // ConsumerBatch is the batch tx that provisionally consumes + // ConsumedVTXO. + ConsumerBatch chainhash.Hash +} diff --git a/batchcanon/record_test.go b/batchcanon/record_test.go new file mode 100644 index 000000000..bf217f87c --- /dev/null +++ b/batchcanon/record_test.go @@ -0,0 +1,70 @@ +package batchcanon + +import ( + "testing" + + "github.com/btcsuite/btcd/chainhash/v2" + fn "github.com/lightningnetwork/lnd/fn/v2" + "github.com/stretchr/testify/require" +) + +// TestEffectiveExpiryNoneWhenUnconfirmed verifies that a batch with no +// current confirmation observation has no effective expiry — the structural +// guarantee that expiry is not a standalone terminal fact. +func TestEffectiveExpiryNoneWhenUnconfirmed(t *testing.T) { + t.Parallel() + + rec := &Record{ + BatchTxID: chainhash.Hash{ + 0x01, + }, + State: StateUnseen, + ConfirmationHeight: fn.None[int32](), + CSVExpiryDelta: 144, + } + + require.True(t, rec.EffectiveExpiry().IsNone()) +} + +// TestEffectiveExpiryDerivesFromConfirmation verifies the effective expiry is +// the confirmation height plus the CSV-relative delta. +func TestEffectiveExpiryDerivesFromConfirmation(t *testing.T) { + t.Parallel() + + rec := &Record{ + BatchTxID: chainhash.Hash{ + 0x02, + }, + State: StateProvisional, + ConfirmationHeight: fn.Some[int32](100), + CSVExpiryDelta: 144, + } + + got := rec.EffectiveExpiry() + require.True(t, got.IsSome()) + require.Equal(t, int32(244), got.UnwrapOr(0)) +} + +// TestEffectiveExpiryRecomputesAfterReconfirm verifies that re-confirming the +// same batch at a different height (as happens after a reorg) yields a fresh +// effective expiry rather than a value frozen at first confirmation. +func TestEffectiveExpiryRecomputesAfterReconfirm(t *testing.T) { + t.Parallel() + + rec := &Record{ + BatchTxID: chainhash.Hash{ + 0x03, + }, + ConfirmationHeight: fn.Some[int32](100), + CSVExpiryDelta: 144, + } + require.Equal(t, int32(244), rec.EffectiveExpiry().UnwrapOr(0)) + + // Reorg: the confirmation leaves the best chain. + rec.ConfirmationHeight = fn.None[int32]() + require.True(t, rec.EffectiveExpiry().IsNone()) + + // Reconfirmation at a higher height on the new best chain. + rec.ConfirmationHeight = fn.Some[int32](103) + require.Equal(t, int32(247), rec.EffectiveExpiry().UnwrapOr(0)) +} diff --git a/batchcanon/state.go b/batchcanon/state.go new file mode 100644 index 000000000..803432a65 --- /dev/null +++ b/batchcanon/state.go @@ -0,0 +1,101 @@ +package batchcanon + +import "fmt" + +// State is the canonicality state of a batch (commitment) transaction as +// interpreted from raw chain observation. It is reorg-reversible: a batch may +// move between states any number of times (e.g. provisional -> finalized -> +// reorged_out -> provisional) as the chain evolves. No state at this layer is +// a one-way terminal verdict — even ConflictFinalized can be undone if the +// conflicting transaction itself later reorgs out. +// +// State is persisted as a typed INTEGER column. Values are append-only and +// MUST NOT be renumbered, because persisted rows reference them directly. +type State int + +const ( + // StateUnseen indicates the batch tx has not been observed confirmed + // on the best chain. This is the zero value: a freshly recorded + // batch with no confirmation observation is unseen. + StateUnseen State = iota + + // StateProvisional indicates the batch tx is confirmed but has not + // yet matured past the configured finality depth, so its + // confirmation may still be reorged out. + StateProvisional + + // StateFinalized indicates the batch tx confirmation has matured past + // the configured finality depth. This is policy finality at the + // configured depth, not a claim of absolute Bitcoin finality. + StateFinalized + + // StateReorgedOut indicates a previously observed confirmation left + // the best chain and no consumed input has been seen double-spent. + // The batch may reconfirm, so dependent VTXOs enter limbo rather than + // being invalidated. + StateReorgedOut + + // StateConflictProvisional indicates a consumed batch input was + // double-spent by a conflicting transaction on the best chain, and + // that conflicting spend has not yet matured past the finality depth. + StateConflictProvisional + + // StateConflictFinalized indicates a consumed-input conflict has + // matured past the finality depth. It is the strongest negative + // signal but, like every state here, remains reversible if the + // conflicting transaction is itself reorged out. + StateConflictFinalized +) + +// String returns a stable lower-snake-case name for the state, matching the +// vocabulary used in darepo#454 and the persisted-column documentation. +func (s State) String() string { + switch s { + case StateUnseen: + return "unseen" + + case StateProvisional: + return "provisional" + + case StateFinalized: + return "finalized" + + case StateReorgedOut: + return "reorged_out" + + case StateConflictProvisional: + return "conflict_provisional" + + case StateConflictFinalized: + return "conflict_finalized" + + default: + return fmt.Sprintf("unknown(%d)", int(s)) + } +} + +// PolicyState is a durable, reorg-independent policy classification slot for a +// batch. darepo#454 reserves this field in the data model; this layer +// persists and round-trips it but assigns no business meaning yet. The +// BatchCanonicalityManager and the admission gates in later tasks own its +// interpretation. +// +// Like State, it is persisted as an append-only typed INTEGER column. +type PolicyState int + +const ( + // PolicyStateDefault is the zero value and the only policy state + // defined at the data-model layer. + PolicyStateDefault PolicyState = iota +) + +// String returns a stable lower-snake-case name for the policy state. +func (p PolicyState) String() string { + switch p { + case PolicyStateDefault: + return "default" + + default: + return fmt.Sprintf("unknown(%d)", int(p)) + } +} diff --git a/batchcanon/state_test.go b/batchcanon/state_test.go new file mode 100644 index 000000000..6fcd89ee2 --- /dev/null +++ b/batchcanon/state_test.go @@ -0,0 +1,75 @@ +package batchcanon + +import ( + "testing" + + "github.com/stretchr/testify/require" +) + +// TestStateValuesStable pins the integer value and string name of every +// canonicality state. These values are persisted as a typed INTEGER column, +// so a change here would silently re-interpret existing rows — the test +// exists to make any renumbering a deliberate, visible edit. +func TestStateValuesStable(t *testing.T) { + t.Parallel() + + cases := []struct { + state State + value int + name string + }{ + { + StateUnseen, + 0, + "unseen", + }, + { + StateProvisional, + 1, + "provisional", + }, + { + StateFinalized, + 2, + "finalized", + }, + { + StateReorgedOut, + 3, + "reorged_out", + }, + { + StateConflictProvisional, + 4, + "conflict_provisional", + }, + { + StateConflictFinalized, + 5, + "conflict_finalized", + }, + } + + for _, tc := range cases { + require.Equal(t, tc.value, int(tc.state), tc.name) + require.Equal(t, tc.name, tc.state.String()) + } +} + +// TestStateStringUnknown verifies an out-of-range state stringifies to a +// diagnosable unknown form rather than an empty string. +func TestStateStringUnknown(t *testing.T) { + t.Parallel() + + require.Equal(t, "unknown(99)", State(99).String()) +} + +// TestPolicyStateStable pins the policy-state value and name. PolicyState is +// also persisted as an append-only typed INTEGER column. +func TestPolicyStateStable(t *testing.T) { + t.Parallel() + + require.Equal(t, 0, int(PolicyStateDefault)) + require.Equal(t, "default", PolicyStateDefault.String()) + require.Equal(t, "unknown(7)", PolicyState(7).String()) +} diff --git a/batchcanon/store.go b/batchcanon/store.go new file mode 100644 index 000000000..d81bd7f0f --- /dev/null +++ b/batchcanon/store.go @@ -0,0 +1,77 @@ +package batchcanon + +import ( + "context" + "errors" + + "github.com/btcsuite/btcd/chainhash/v2" + "github.com/btcsuite/btcd/wire/v2" +) + +// ErrBatchNotFound is returned by Store.GetBatch when no canonicality record +// exists for the requested batch txid. +var ErrBatchNotFound = errors.New("batch canonicality record not found") + +// Store is the durable query/update surface for batch canonicality records. +// It is intentionally behavior-free: it persists and retrieves observations +// and reverse-dependency edges, leaving all interpretation — state +// transitions, chain watching, and admission — to the BatchCanonicalityManager +// and the later tasks of the reorg-safety epic. +type Store interface { + // UpsertBatch inserts or replaces the canonicality record for a + // batch, including its consumed inputs and dependent VTXOs. It is the + // single entry point for first-seeing a batch and for wholesale + // rewrites; targeted mutations use the methods below. + UpsertBatch(ctx context.Context, record *Record) error + + // GetBatch returns the canonicality record for a batch txid. It + // returns ErrBatchNotFound when no record exists. + GetBatch(ctx context.Context, txid chainhash.Hash) (*Record, error) + + // ListBatchesByState returns every batch currently in the given + // state. Used by the manager to find batches needing a particular + // follow-up (e.g. all provisional batches to re-check for finality). + ListBatchesByState(ctx context.Context, state State) ([]*Record, error) + + // UpdateBatchState transitions a batch to a new canonicality state + // without touching its other fields. + UpdateBatchState(ctx context.Context, txid chainhash.Hash, + state State) error + + // RecordConfirmation records that the batch tx is confirmed at the + // given best-chain height and block hash. A later RecordConfirmation + // at a different height (after a reorg) overwrites the observation so + // the effective expiry tracks the new confirmation. + RecordConfirmation(ctx context.Context, txid chainhash.Hash, + height int32, block chainhash.Hash) error + + // ClearConfirmation clears the confirmation observation for a batch, + // reflecting that its confirming block left the best chain. It does + // not set any terminal flag: the batch may reconfirm. + ClearConfirmation(ctx context.Context, txid chainhash.Hash) error + + // FindBatchesConsumingOutpoint returns the txids of every recorded + // batch that consumes the given outpoint. Used to detect input + // conflicts: two batches consuming the same outpoint are in conflict. + FindBatchesConsumingOutpoint(ctx context.Context, + outpoint wire.OutPoint) ([]chainhash.Hash, error) + + // AddProvisionalConsumer records a reverse-dependency edge: the given + // VTXO outpoint has been provisionally consumed by the given consumer + // batch. Idempotent on (consumedVTXO, consumerBatch). + AddProvisionalConsumer(ctx context.Context, consumedVTXO wire.OutPoint, + consumerBatch chainhash.Hash) error + + // ListProvisionalConsumersForBatch returns the VTXO outpoints that + // the given consumer batch provisionally consumes. Used to find the + // VTXOs to restore when the consumer batch is invalidated. + ListProvisionalConsumersForBatch(ctx context.Context, + consumerBatch chainhash.Hash) ([]wire.OutPoint, error) + + // DeleteProvisionalConsumersForBatch removes every reverse-dependency + // edge for the given consumer batch, used once the batch is canonical + // (the consumption is no longer provisional) or fully invalidated and + // reconciled. + DeleteProvisionalConsumersForBatch(ctx context.Context, + consumerBatch chainhash.Hash) error +} diff --git a/db/CLAUDE.md b/db/CLAUDE.md index 3cf684ded..e99300a16 100644 --- a/db/CLAUDE.md +++ b/db/CLAUDE.md @@ -71,7 +71,16 @@ For field-level detail, use `go doc github.com/lightninglabs/darepo-client/db.= int32(finalityDepth) { + state = batchcanon.StateFinalized + } + + err = q.UpsertBatchCanonicality( + ctx, sqlc.UpsertBatchCanonicalityParams{ + BatchTxid: txid[:], + State: int32(state), + ConfirmationHeight: sql.NullInt32{ + Int32: g.createdHeight, + Valid: true, + }, + // The confirming block hash is not + // recorded on VTXO rows; it is an + // observation attribute the manager + // fills on its next confirmation + // sighting. + ConfirmationBlockHash: nil, + CsvExpiryDelta: csvDelta, + PolicyState: int32( + batchcanon.PolicyStateDefault, + ), + CreatedAt: now, + UpdatedAt: now, + }, + ) + if err != nil { + return err + } + + err = insertDependentVTXOs(ctx, q, txid, g.dependents) + if err != nil { + return err + } + + created++ + } + + return nil + }) + + return created, err +} + +// hydrateRecord builds a batchcanon.Record from a canonicality row, loading +// its consumed inputs and dependent VTXOs through the same query handle (and +// therefore the same transaction). +func (s *BatchCanonicalityPersistenceStore) hydrateRecord(ctx context.Context, + q BatchCanonicalityStore, row sqlc.BatchCanonicality) ( + *batchcanon.Record, error) { + + txid, err := chainhash.NewHash(row.BatchTxid) + if err != nil { + return nil, err + } + + confBlock, err := bytesToOptionHash(row.ConfirmationBlockHash) + if err != nil { + return nil, err + } + + inputRows, err := q.ListBatchConsumedInputs(ctx, row.BatchTxid) + if err != nil { + return nil, err + } + inputs := make([]wire.OutPoint, 0, len(inputRows)) + for _, in := range inputRows { + hash, err := chainhash.NewHash(in.InputHash) + if err != nil { + return nil, err + } + inputs = append(inputs, wire.OutPoint{ + Hash: *hash, + Index: uint32(in.InputIndex), + }) + } + + depRows, err := q.ListBatchDependentVTXOs(ctx, row.BatchTxid) + if err != nil { + return nil, err + } + deps := make([]wire.OutPoint, 0, len(depRows)) + for _, dep := range depRows { + hash, err := chainhash.NewHash(dep.VtxoOutpointHash) + if err != nil { + return nil, err + } + deps = append(deps, wire.OutPoint{ + Hash: *hash, + Index: uint32(dep.VtxoOutpointIndex), + }) + } + + return &batchcanon.Record{ + 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 +} + +// optionToNullInt32 maps an optional int32 to a sql.NullInt32. +func optionToNullInt32(o fn.Option[int32]) sql.NullInt32 { + if o.IsNone() { + return sql.NullInt32{} + } + + return sql.NullInt32{Int32: o.UnwrapOr(0), Valid: true} +} + +// nullInt32ToOption maps a sql.NullInt32 back to an optional int32. +func nullInt32ToOption(n sql.NullInt32) fn.Option[int32] { + if !n.Valid { + return fn.None[int32]() + } + + return fn.Some(n.Int32) +} + +// optionHashToBytes maps an optional hash to its raw bytes, or nil when None. +func optionHashToBytes(o fn.Option[chainhash.Hash]) []byte { + if o.IsNone() { + return nil + } + + h := o.UnwrapOr(chainhash.Hash{}) + + return h[:] +} + +// bytesToOptionHash maps a (possibly nil) raw hash to an optional hash. A nil +// or empty slice yields None; any other length is validated to 32 bytes. +func bytesToOptionHash(raw []byte) (fn.Option[chainhash.Hash], error) { + if len(raw) == 0 { + return fn.None[chainhash.Hash](), nil + } + + hash, err := chainhash.NewHash(raw) + if err != nil { + return fn.None[chainhash.Hash](), err + } + + return fn.Some(*hash), nil +} + +// Compile-time check that the persistence store satisfies the domain Store +// interface. +var _ batchcanon.Store = (*BatchCanonicalityPersistenceStore)(nil) diff --git a/db/batch_canonicality_store_test.go b/db/batch_canonicality_store_test.go new file mode 100644 index 000000000..a128cd9e1 --- /dev/null +++ b/db/batch_canonicality_store_test.go @@ -0,0 +1,467 @@ +package db + +import ( + "database/sql" + "testing" + + "github.com/btcsuite/btcd/chainhash/v2" + "github.com/btcsuite/btcd/wire/v2" + "github.com/btcsuite/btclog/v2" + "github.com/lightninglabs/darepo-client/batchcanon" + "github.com/lightninglabs/darepo-client/lib/tree" + "github.com/lightninglabs/darepo-client/round" + "github.com/lightningnetwork/lnd/clock" + fn "github.com/lightningnetwork/lnd/fn/v2" + "github.com/stretchr/testify/require" +) + +// newBatchCanonicalityStoreForTest creates a batch canonicality store backed +// by a fresh test database. +func newBatchCanonicalityStoreForTest( + t *testing.T) *BatchCanonicalityPersistenceStore { + + t.Helper() + + db := NewTestDB(t) + + canonDB := NewTransactionExecutor( + db.BaseDB, + func(tx *sql.Tx) BatchCanonicalityStore { + return db.WithTx(tx) + }, + btclog.Disabled, + ) + + return NewBatchCanonicalityPersistenceStore( + canonDB, clock.NewDefaultClock(), + ) +} + +// outpoint is a small test helper building a deterministic outpoint. +func outpoint(b byte, index uint32) wire.OutPoint { + return wire.OutPoint{Hash: chainhash.Hash{b}, Index: index} +} + +// TestBatchCanonicalityUpsertRoundTrip verifies a record survives an upsert +// and read with all of its fields, consumed inputs, and dependent VTXOs. +func TestBatchCanonicalityUpsertRoundTrip(t *testing.T) { + t.Parallel() + + ctx := t.Context() + store := newBatchCanonicalityStoreForTest(t) + + txid := chainhash.Hash{0xaa} + rec := &batchcanon.Record{ + BatchTxID: txid, + State: batchcanon.StateProvisional, + ConfirmationHeight: fn.Some[int32](100), + ConfirmationBlock: fn.Some(chainhash.Hash{0xbb}), + CSVExpiryDelta: 144, + PolicyState: batchcanon.PolicyStateDefault, + ConsumedInputs: []wire.OutPoint{ + outpoint(0x01, 0), outpoint(0x02, 3), + }, + DependentVTXOs: []wire.OutPoint{ + outpoint(0x03, 1), + }, + } + require.NoError(t, store.UpsertBatch(ctx, rec)) + + got, err := store.GetBatch(ctx, txid) + require.NoError(t, err) + require.Equal(t, txid, got.BatchTxID) + require.Equal(t, batchcanon.StateProvisional, got.State) + require.Equal(t, int32(100), got.ConfirmationHeight.UnwrapOr(0)) + require.True(t, got.ConfirmationBlock.IsSome()) + require.Equal(t, int32(144), got.CSVExpiryDelta) + require.Equal(t, batchcanon.PolicyStateDefault, got.PolicyState) + require.ElementsMatch(t, rec.ConsumedInputs, got.ConsumedInputs) + require.ElementsMatch(t, rec.DependentVTXOs, got.DependentVTXOs) + + // Effective expiry derives from the stored confirmation. + require.Equal(t, int32(244), got.EffectiveExpiry().UnwrapOr(0)) +} + +// TestBatchCanonicalityGetNotFound verifies the not-found sentinel. +func TestBatchCanonicalityGetNotFound(t *testing.T) { + t.Parallel() + + ctx := t.Context() + store := newBatchCanonicalityStoreForTest(t) + + _, err := store.GetBatch(ctx, chainhash.Hash{0xff}) + require.ErrorIs(t, err, batchcanon.ErrBatchNotFound) +} + +// TestBatchCanonicalityUpsertReplacesEdges verifies a re-upsert replaces the +// consumed-input and dependent-VTXO sets rather than appending. +func TestBatchCanonicalityUpsertReplacesEdges(t *testing.T) { + t.Parallel() + + ctx := t.Context() + store := newBatchCanonicalityStoreForTest(t) + + txid := chainhash.Hash{0xa1} + require.NoError( + t, + store.UpsertBatch( + ctx, &batchcanon.Record{ + BatchTxID: txid, + State: batchcanon.StateUnseen, + CSVExpiryDelta: 10, + ConsumedInputs: []wire.OutPoint{ + outpoint(0x01, 0), + }, + DependentVTXOs: []wire.OutPoint{ + outpoint(0x02, 0), + }, + }, + ), + ) + + // Re-upsert with a different edge set. + require.NoError( + t, + store.UpsertBatch( + ctx, &batchcanon.Record{ + BatchTxID: txid, + State: batchcanon.StateProvisional, + CSVExpiryDelta: 10, + ConsumedInputs: []wire.OutPoint{ + outpoint(0x09, 2), + }, + DependentVTXOs: nil, + }, + ), + ) + + got, err := store.GetBatch(ctx, txid) + require.NoError(t, err) + require.Equal(t, []wire.OutPoint{outpoint(0x09, 2)}, got.ConsumedInputs) + require.Empty(t, got.DependentVTXOs) +} + +// TestBatchCanonicalityReorgRecomputesExpiry verifies the reorg-aware expiry +// contract end to end through the store: a confirmation yields an effective +// expiry, a reorg (ClearConfirmation) erases it, and a reconfirmation at a new +// height yields a fresh effective expiry. Expiry is never frozen. +func TestBatchCanonicalityReorgRecomputesExpiry(t *testing.T) { + t.Parallel() + + ctx := t.Context() + store := newBatchCanonicalityStoreForTest(t) + + txid := chainhash.Hash{0xc0} + require.NoError( + t, + store.UpsertBatch( + ctx, &batchcanon.Record{ + BatchTxID: txid, + State: batchcanon.StateUnseen, + CSVExpiryDelta: 144, + }, + ), + ) + + // Unconfirmed: no effective expiry. + got, err := store.GetBatch(ctx, txid) + require.NoError(t, err) + require.True(t, got.EffectiveExpiry().IsNone()) + + // Confirm at height 100. + require.NoError( + t, + store.RecordConfirmation( + ctx, txid, 100, chainhash.Hash{0xc1}, + ), + ) + got, err = store.GetBatch(ctx, txid) + require.NoError(t, err) + require.Equal(t, int32(244), got.EffectiveExpiry().UnwrapOr(0)) + + // Reorg out: confirmation cleared, effective expiry erased. + require.NoError(t, store.ClearConfirmation(ctx, txid)) + got, err = store.GetBatch(ctx, txid) + require.NoError(t, err) + require.True(t, got.ConfirmationHeight.IsNone()) + require.True(t, got.EffectiveExpiry().IsNone()) + + // Reconfirm at a higher height: fresh effective expiry. + require.NoError( + t, + store.RecordConfirmation( + ctx, txid, 105, chainhash.Hash{0xc2}, + ), + ) + got, err = store.GetBatch(ctx, txid) + require.NoError(t, err) + require.Equal(t, int32(249), got.EffectiveExpiry().UnwrapOr(0)) +} + +// TestBatchCanonicalityStateNotTerminal verifies state can move freely in any +// direction (finalized -> reorged_out -> provisional), proving no state is +// persisted as an irreversible terminal verdict. +func TestBatchCanonicalityStateNotTerminal(t *testing.T) { + t.Parallel() + + ctx := t.Context() + store := newBatchCanonicalityStoreForTest(t) + + txid := chainhash.Hash{0xd0} + require.NoError( + t, + store.UpsertBatch( + ctx, &batchcanon.Record{ + BatchTxID: txid, + State: batchcanon.StateFinalized, + CSVExpiryDelta: 10, + }, + ), + ) + + for _, want := range []batchcanon.State{ + batchcanon.StateReorgedOut, + batchcanon.StateConflictFinalized, + batchcanon.StateProvisional, + } { + require.NoError(t, store.UpdateBatchState(ctx, txid, want)) + got, err := store.GetBatch(ctx, txid) + require.NoError(t, err) + require.Equal(t, want, got.State) + } +} + +// TestBatchCanonicalityListByState verifies state-filtered listing. +func TestBatchCanonicalityListByState(t *testing.T) { + t.Parallel() + + ctx := t.Context() + store := newBatchCanonicalityStoreForTest(t) + + require.NoError( + t, + store.UpsertBatch( + ctx, &batchcanon.Record{ + BatchTxID: chainhash.Hash{0xe0}, + State: batchcanon.StateProvisional, + CSVExpiryDelta: 1, + }, + ), + ) + require.NoError( + t, + store.UpsertBatch( + ctx, &batchcanon.Record{ + BatchTxID: chainhash.Hash{0xe1}, + State: batchcanon.StateProvisional, + CSVExpiryDelta: 1, + }, + ), + ) + require.NoError( + t, + store.UpsertBatch( + ctx, &batchcanon.Record{ + BatchTxID: chainhash.Hash{0xe2}, + State: batchcanon.StateFinalized, + CSVExpiryDelta: 1, + }, + ), + ) + + prov, err := store.ListBatchesByState(ctx, batchcanon.StateProvisional) + require.NoError(t, err) + require.Len(t, prov, 2) + + final, err := store.ListBatchesByState(ctx, batchcanon.StateFinalized) + require.NoError(t, err) + require.Len(t, final, 1) +} + +// TestBatchCanonicalityFindByConsumedOutpoint verifies input-conflict +// detection: two batches consuming the same outpoint are both found. +func TestBatchCanonicalityFindByConsumedOutpoint(t *testing.T) { + t.Parallel() + + ctx := t.Context() + store := newBatchCanonicalityStoreForTest(t) + + shared := outpoint(0x55, 1) + batchA := chainhash.Hash{0xa0} + batchB := chainhash.Hash{0xb0} + + require.NoError( + t, + store.UpsertBatch( + ctx, &batchcanon.Record{ + BatchTxID: batchA, + State: batchcanon.StateProvisional, + CSVExpiryDelta: 1, + ConsumedInputs: []wire.OutPoint{shared}, + }, + ), + ) + require.NoError( + t, + store.UpsertBatch( + ctx, &batchcanon.Record{ + BatchTxID: batchB, + State: batchcanon.StateProvisional, + CSVExpiryDelta: 1, + ConsumedInputs: []wire.OutPoint{shared}, + }, + ), + ) + + found, err := store.FindBatchesConsumingOutpoint(ctx, shared) + require.NoError(t, err) + require.ElementsMatch(t, []chainhash.Hash{batchA, batchB}, found) + + none, err := store.FindBatchesConsumingOutpoint(ctx, outpoint(0x99, 0)) + require.NoError(t, err) + require.Empty(t, none) +} + +// TestBatchCanonicalityProvisionalConsumerRestore verifies the reverse- +// dependency lifecycle: a provisionally consumed VTXO is listed for its +// consumer batch (so it can be restored if the batch is invalidated), survives +// a batch state change, and is removed on delete. +func TestBatchCanonicalityProvisionalConsumerRestore(t *testing.T) { + t.Parallel() + + ctx := t.Context() + store := newBatchCanonicalityStoreForTest(t) + + consumerBatch := chainhash.Hash{0xf0} + require.NoError( + t, + store.UpsertBatch( + ctx, &batchcanon.Record{ + BatchTxID: consumerBatch, + State: batchcanon.StateProvisional, + CSVExpiryDelta: 1, + }, + ), + ) + + consumed := outpoint(0x44, 2) + require.NoError( + t, store.AddProvisionalConsumer( + ctx, consumed, consumerBatch, + ), + ) + + // Listed for the consumer batch. + got, err := store.ListProvisionalConsumersForBatch(ctx, consumerBatch) + require.NoError(t, err) + require.Equal(t, []wire.OutPoint{consumed}, got) + + // Idempotent re-add. + require.NoError( + t, store.AddProvisionalConsumer( + ctx, consumed, consumerBatch, + ), + ) + got, err = store.ListProvisionalConsumersForBatch(ctx, consumerBatch) + require.NoError(t, err) + require.Len(t, got, 1) + + // The edge survives the batch being marked reorged/invalidated — that + // is exactly when the restore caller needs to read it. + require.NoError( + t, store.UpdateBatchState( + ctx, consumerBatch, batchcanon.StateReorgedOut, + ), + ) + got, err = store.ListProvisionalConsumersForBatch(ctx, consumerBatch) + require.NoError(t, err) + require.Equal(t, []wire.OutPoint{consumed}, got) + + // Deleting clears the edges (e.g. once the consumption is canonical or + // fully reconciled). + require.NoError( + t, store.DeleteProvisionalConsumersForBatch( + ctx, consumerBatch, + ), + ) + got, err = store.ListProvisionalConsumersForBatch(ctx, consumerBatch) + require.NoError(t, err) + require.Empty(t, got) +} + +// TestBatchCanonicalityBackfillFromVTXOs verifies that backfill derives one +// canonicality record per distinct batch present in the VTXO store, with the +// CSV-relative expiry delta recovered from the stored absolute batch_expiry, +// the right provisional/finalized classification, and the dependent VTXO +// linked. It also verifies idempotency: a re-run creates nothing and does not +// clobber state the manager has since advanced. +func TestBatchCanonicalityBackfillFromVTXOs(t *testing.T) { + t.Parallel() + + ctx := t.Context() + + vtxoStore, roundStore, baseDB := newVTXOStoreForTest(t) + + canonDB := NewTransactionExecutor( + baseDB, + func(tx *sql.Tx) BatchCanonicalityStore { + return baseDB.WithTx(tx) + }, + btclog.Disabled, + ) + canon := NewBatchCanonicalityPersistenceStore( + canonDB, clock.NewDefaultClock(), + ) + + // A round must exist to satisfy the VTXO foreign key. + roundID := testRoundIDDB("backfill-round") + testRound := createTestRound(t, roundID) + sigState := &round.InputSigSentState{ + RoundID: testRound.RoundID, + ClientTrees: make(map[round.SignerKey]*tree.Tree), + } + require.NoError(t, roundStore.CommitState(ctx, testRound, sigState)) + + // Two VTXOs in two distinct batches: + // idx 0: batch_expiry 1000, created_height 500 + // idx 1: batch_expiry 1100, created_height 510 + desc0 := createTestVTXODescriptor(t, roundID, 0) + desc1 := createTestVTXODescriptor(t, roundID, 1) + require.NoError(t, vtxoStore.SaveVTXO(ctx, desc0)) + require.NoError(t, vtxoStore.SaveVTXO(ctx, desc1)) + + // best height 505, finality depth 6: + // batch 0: depth = 505-500+1 = 6 >= 6 -> finalized + // batch 1: depth = 505-510+1 < 6 -> provisional + n, err := canon.BackfillFromVTXOs(ctx, 505, 6) + require.NoError(t, err) + require.Equal(t, 2, n) + + rec0, err := canon.GetBatch(ctx, desc0.CommitmentTxID) + require.NoError(t, err) + require.Equal(t, batchcanon.StateFinalized, rec0.State) + require.Equal(t, int32(500), rec0.ConfirmationHeight.UnwrapOr(0)) + require.Equal(t, int32(500), rec0.CSVExpiryDelta) + require.Equal(t, int32(1000), rec0.EffectiveExpiry().UnwrapOr(0)) + require.Equal(t, []wire.OutPoint{desc0.Outpoint}, rec0.DependentVTXOs) + + rec1, err := canon.GetBatch(ctx, desc1.CommitmentTxID) + require.NoError(t, err) + require.Equal(t, batchcanon.StateProvisional, rec1.State) + require.Equal(t, int32(590), rec1.CSVExpiryDelta) + + // Idempotency: advance one batch's state, re-run backfill, and verify + // it creates nothing new and leaves the advanced state untouched. + require.NoError( + t, canon.UpdateBatchState( + ctx, desc0.CommitmentTxID, batchcanon.StateReorgedOut, + ), + ) + n, err = canon.BackfillFromVTXOs(ctx, 505, 6) + require.NoError(t, err) + require.Equal(t, 0, n) + + rec0, err = canon.GetBatch(ctx, desc0.CommitmentTxID) + require.NoError(t, err) + require.Equal(t, batchcanon.StateReorgedOut, rec0.State) +} diff --git a/db/migrations.go b/db/migrations.go index d5b37ce9f..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 = 10 + 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 new file mode 100644 index 000000000..7729a8049 --- /dev/null +++ b/db/sqlc/batch_canonicality.sql.go @@ -0,0 +1,490 @@ +// Code generated by sqlc. DO NOT EDIT. +// versions: +// sqlc v1.29.0 +// source: batch_canonicality.sql + +package sqlc + +import ( + "context" + "database/sql" +) + +const ClearBatchConfirmation = `-- name: ClearBatchConfirmation :exec +UPDATE batch_canonicality +SET confirmation_height = NULL, confirmation_block_hash = NULL, updated_at = $2 +WHERE batch_txid = $1 +` + +type ClearBatchConfirmationParams struct { + BatchTxid []byte + UpdatedAt int64 +} + +// ClearBatchConfirmation nulls the confirmation observation, reflecting that +// the confirming block left the best chain. It sets no terminal flag. +func (q *Queries) ClearBatchConfirmation(ctx context.Context, arg ClearBatchConfirmationParams) error { + _, err := q.db.ExecContext(ctx, ClearBatchConfirmation, arg.BatchTxid, arg.UpdatedAt) + return err +} + +const DeleteBatchConsumedInputs = `-- name: DeleteBatchConsumedInputs :exec +DELETE FROM batch_consumed_inputs WHERE batch_txid = $1 +` + +// DeleteBatchConsumedInputs removes every consumed-input row for a batch, +// used by the store's upsert to replace the set atomically. +func (q *Queries) DeleteBatchConsumedInputs(ctx context.Context, batchTxid []byte) error { + _, err := q.db.ExecContext(ctx, DeleteBatchConsumedInputs, batchTxid) + return err +} + +const DeleteBatchDependentVTXOs = `-- name: DeleteBatchDependentVTXOs :exec +DELETE FROM batch_dependent_vtxos WHERE batch_txid = $1 +` + +// DeleteBatchDependentVTXOs removes every dependent-VTXO row for a batch, +// used by the store's upsert to replace the set atomically. +func (q *Queries) DeleteBatchDependentVTXOs(ctx context.Context, batchTxid []byte) error { + _, err := q.db.ExecContext(ctx, DeleteBatchDependentVTXOs, batchTxid) + return err +} + +const DeleteProvisionalConsumersForBatch = `-- name: DeleteProvisionalConsumersForBatch :exec +DELETE FROM batch_provisional_consumers WHERE consumer_batch_txid = $1 +` + +// DeleteProvisionalConsumersForBatch removes every reverse-dependency edge +// for the given consumer batch. +func (q *Queries) DeleteProvisionalConsumersForBatch(ctx context.Context, consumerBatchTxid []byte) error { + _, err := q.db.ExecContext(ctx, DeleteProvisionalConsumersForBatch, consumerBatchTxid) + return err +} + +const FindBatchesByConsumedOutpoint = `-- name: FindBatchesByConsumedOutpoint :many +SELECT batch_txid +FROM batch_consumed_inputs +WHERE input_hash = $1 AND input_index = $2 +` + +type FindBatchesByConsumedOutpointParams struct { + InputHash []byte + InputIndex int32 +} + +// FindBatchesByConsumedOutpoint returns the txids of every batch that +// consumes the given outpoint. +func (q *Queries) FindBatchesByConsumedOutpoint(ctx context.Context, arg FindBatchesByConsumedOutpointParams) ([][]byte, error) { + rows, err := q.db.QueryContext(ctx, FindBatchesByConsumedOutpoint, arg.InputHash, arg.InputIndex) + if err != nil { + return nil, err + } + defer rows.Close() + var items [][]byte + for rows.Next() { + var batch_txid []byte + if err := rows.Scan(&batch_txid); err != nil { + return nil, err + } + items = append(items, batch_txid) + } + if err := rows.Close(); err != nil { + return nil, err + } + if err := rows.Err(); err != nil { + return nil, err + } + return items, nil +} + +const GetBatchCanonicality = `-- name: GetBatchCanonicality :one +SELECT batch_txid, state, confirmation_height, confirmation_block_hash, + 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. 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 + err := row.Scan( + &i.BatchTxid, + &i.State, + &i.ConfirmationHeight, + &i.ConfirmationBlockHash, + &i.CsvExpiryDelta, + &i.PolicyState, + &i.CreatedAt, + &i.UpdatedAt, + &i.ConfirmationPkScript, + ) + return i, err +} + +const InsertBatchConsumedInput = `-- name: InsertBatchConsumedInput :exec +INSERT INTO batch_consumed_inputs (batch_txid, input_hash, input_index) +VALUES ($1, $2, $3) +ON CONFLICT (batch_txid, input_hash, input_index) DO NOTHING +` + +type InsertBatchConsumedInputParams struct { + BatchTxid []byte + InputHash []byte + InputIndex int32 +} + +// InsertBatchConsumedInput records one outpoint consumed by a batch. +func (q *Queries) InsertBatchConsumedInput(ctx context.Context, arg InsertBatchConsumedInputParams) error { + _, err := q.db.ExecContext(ctx, InsertBatchConsumedInput, arg.BatchTxid, arg.InputHash, arg.InputIndex) + return err +} + +const InsertBatchDependentVTXO = `-- name: InsertBatchDependentVTXO :exec +INSERT INTO batch_dependent_vtxos ( + batch_txid, vtxo_outpoint_hash, vtxo_outpoint_index +) VALUES ($1, $2, $3) +ON CONFLICT (batch_txid, vtxo_outpoint_hash, vtxo_outpoint_index) DO NOTHING +` + +type InsertBatchDependentVTXOParams struct { + BatchTxid []byte + VtxoOutpointHash []byte + VtxoOutpointIndex int32 +} + +// InsertBatchDependentVTXO records one VTXO outpoint anchored by a batch. +func (q *Queries) InsertBatchDependentVTXO(ctx context.Context, arg InsertBatchDependentVTXOParams) error { + _, err := q.db.ExecContext(ctx, InsertBatchDependentVTXO, arg.BatchTxid, arg.VtxoOutpointHash, arg.VtxoOutpointIndex) + return err +} + +const InsertProvisionalConsumer = `-- name: InsertProvisionalConsumer :exec +INSERT INTO batch_provisional_consumers ( + consumed_vtxo_hash, consumed_vtxo_index, consumer_batch_txid, created_at +) VALUES ($1, $2, $3, $4) +ON CONFLICT ( + consumed_vtxo_hash, consumed_vtxo_index, consumer_batch_txid +) DO NOTHING +` + +type InsertProvisionalConsumerParams struct { + ConsumedVtxoHash []byte + ConsumedVtxoIndex int32 + ConsumerBatchTxid []byte + CreatedAt int64 +} + +// InsertProvisionalConsumer records a reverse-dependency edge: consumed_vtxo +// is provisionally consumed by consumer_batch. Idempotent. +func (q *Queries) InsertProvisionalConsumer(ctx context.Context, arg InsertProvisionalConsumerParams) error { + _, err := q.db.ExecContext(ctx, InsertProvisionalConsumer, + arg.ConsumedVtxoHash, + arg.ConsumedVtxoIndex, + arg.ConsumerBatchTxid, + arg.CreatedAt, + ) + return err +} + +const ListBatchCanonicalityByState = `-- name: ListBatchCanonicalityByState :many +SELECT batch_txid, state, confirmation_height, confirmation_block_hash, + csv_expiry_delta, policy_state, created_at, updated_at, + confirmation_pk_script +FROM batch_canonicality +WHERE state = $1 +` + +// ListBatchCanonicalityByState returns every batch currently in the given +// state. +func (q *Queries) ListBatchCanonicalityByState(ctx context.Context, state int32) ([]BatchCanonicality, error) { + rows, err := q.db.QueryContext(ctx, ListBatchCanonicalityByState, state) + if err != nil { + return nil, err + } + defer rows.Close() + var items []BatchCanonicality + for rows.Next() { + var i BatchCanonicality + if err := rows.Scan( + &i.BatchTxid, + &i.State, + &i.ConfirmationHeight, + &i.ConfirmationBlockHash, + &i.CsvExpiryDelta, + &i.PolicyState, + &i.CreatedAt, + &i.UpdatedAt, + &i.ConfirmationPkScript, + ); err != nil { + return nil, err + } + items = append(items, i) + } + if err := rows.Close(); err != nil { + return nil, err + } + if err := rows.Err(); err != nil { + return nil, err + } + return items, nil +} + +const ListBatchConsumedInputs = `-- name: ListBatchConsumedInputs :many +SELECT input_hash, input_index +FROM batch_consumed_inputs +WHERE batch_txid = $1 +` + +type ListBatchConsumedInputsRow struct { + InputHash []byte + InputIndex int32 +} + +// ListBatchConsumedInputs returns the outpoints a batch consumes. +func (q *Queries) ListBatchConsumedInputs(ctx context.Context, batchTxid []byte) ([]ListBatchConsumedInputsRow, error) { + rows, err := q.db.QueryContext(ctx, ListBatchConsumedInputs, batchTxid) + if err != nil { + return nil, err + } + defer rows.Close() + var items []ListBatchConsumedInputsRow + for rows.Next() { + var i ListBatchConsumedInputsRow + if err := rows.Scan(&i.InputHash, &i.InputIndex); err != nil { + return nil, err + } + items = append(items, i) + } + if err := rows.Close(); err != nil { + return nil, err + } + if err := rows.Err(); err != nil { + return nil, err + } + return items, nil +} + +const ListBatchDependentVTXOs = `-- name: ListBatchDependentVTXOs :many +SELECT vtxo_outpoint_hash, vtxo_outpoint_index +FROM batch_dependent_vtxos +WHERE batch_txid = $1 +` + +type ListBatchDependentVTXOsRow struct { + VtxoOutpointHash []byte + VtxoOutpointIndex int32 +} + +// ListBatchDependentVTXOs returns the VTXO outpoints a batch anchors. +func (q *Queries) ListBatchDependentVTXOs(ctx context.Context, batchTxid []byte) ([]ListBatchDependentVTXOsRow, error) { + rows, err := q.db.QueryContext(ctx, ListBatchDependentVTXOs, batchTxid) + if err != nil { + return nil, err + } + defer rows.Close() + var items []ListBatchDependentVTXOsRow + for rows.Next() { + var i ListBatchDependentVTXOsRow + if err := rows.Scan(&i.VtxoOutpointHash, &i.VtxoOutpointIndex); err != nil { + return nil, err + } + items = append(items, i) + } + if err := rows.Close(); err != nil { + return nil, err + } + if err := rows.Err(); err != nil { + return nil, err + } + return items, nil +} + +const ListProvisionalConsumersForBatch = `-- name: ListProvisionalConsumersForBatch :many +SELECT consumed_vtxo_hash, consumed_vtxo_index +FROM batch_provisional_consumers +WHERE consumer_batch_txid = $1 +` + +type ListProvisionalConsumersForBatchRow struct { + ConsumedVtxoHash []byte + ConsumedVtxoIndex int32 +} + +// ListProvisionalConsumersForBatch returns the VTXO outpoints that the given +// consumer batch provisionally consumes (the VTXOs to restore if the batch +// is invalidated). +func (q *Queries) ListProvisionalConsumersForBatch(ctx context.Context, consumerBatchTxid []byte) ([]ListProvisionalConsumersForBatchRow, error) { + rows, err := q.db.QueryContext(ctx, ListProvisionalConsumersForBatch, consumerBatchTxid) + if err != nil { + return nil, err + } + defer rows.Close() + var items []ListProvisionalConsumersForBatchRow + for rows.Next() { + var i ListProvisionalConsumersForBatchRow + if err := rows.Scan(&i.ConsumedVtxoHash, &i.ConsumedVtxoIndex); err != nil { + return nil, err + } + items = append(items, i) + } + if err := rows.Close(); err != nil { + return nil, err + } + if err := rows.Err(); err != nil { + return nil, err + } + return items, nil +} + +const ListVTXOsForCanonicalityBackfill = `-- name: ListVTXOsForCanonicalityBackfill :many +SELECT outpoint_hash, outpoint_index, commitment_txid, batch_expiry, + created_height +FROM vtxos +WHERE length(commitment_txid) = 32 +` + +type ListVTXOsForCanonicalityBackfillRow struct { + OutpointHash []byte + OutpointIndex int32 + CommitmentTxid []byte + BatchExpiry int32 + CreatedHeight int32 +} + +// ListVTXOsForCanonicalityBackfill returns the columns needed to derive +// initial batch canonicality records from already-persisted VTXOs: each +// VTXO's outpoint, its commitment (batch) txid, the absolute batch expiry +// height, and the height at which it was created (confirmed). The backfill +// groups these by commitment txid in Go and recomputes the CSV-relative +// expiry delta as batch_expiry - created_height. +func (q *Queries) ListVTXOsForCanonicalityBackfill(ctx context.Context) ([]ListVTXOsForCanonicalityBackfillRow, error) { + rows, err := q.db.QueryContext(ctx, ListVTXOsForCanonicalityBackfill) + if err != nil { + return nil, err + } + defer rows.Close() + var items []ListVTXOsForCanonicalityBackfillRow + for rows.Next() { + var i ListVTXOsForCanonicalityBackfillRow + if err := rows.Scan( + &i.OutpointHash, + &i.OutpointIndex, + &i.CommitmentTxid, + &i.BatchExpiry, + &i.CreatedHeight, + ); err != nil { + return nil, err + } + items = append(items, i) + } + if err := rows.Close(); err != nil { + return nil, err + } + if err := rows.Err(); err != nil { + return nil, err + } + return items, nil +} + +const RecordBatchConfirmation = `-- name: RecordBatchConfirmation :exec +UPDATE batch_canonicality +SET confirmation_height = $2, confirmation_block_hash = $3, updated_at = $4 +WHERE batch_txid = $1 +` + +type RecordBatchConfirmationParams struct { + BatchTxid []byte + ConfirmationHeight sql.NullInt32 + ConfirmationBlockHash []byte + UpdatedAt int64 +} + +// RecordBatchConfirmation records the best-chain height and block hash at +// which the batch tx is confirmed. A later call at a different height (after +// a reorg) overwrites the observation so effective expiry tracks the new +// confirmation. +func (q *Queries) RecordBatchConfirmation(ctx context.Context, arg RecordBatchConfirmationParams) error { + _, err := q.db.ExecContext(ctx, RecordBatchConfirmation, + arg.BatchTxid, + arg.ConfirmationHeight, + arg.ConfirmationBlockHash, + arg.UpdatedAt, + ) + return err +} + +const UpdateBatchCanonicalityState = `-- name: UpdateBatchCanonicalityState :exec +UPDATE batch_canonicality +SET state = $2, updated_at = $3 +WHERE batch_txid = $1 +` + +type UpdateBatchCanonicalityStateParams struct { + BatchTxid []byte + State int32 + UpdatedAt int64 +} + +// UpdateBatchCanonicalityState transitions a batch to a new state without +// touching its other fields. +func (q *Queries) UpdateBatchCanonicalityState(ctx context.Context, arg UpdateBatchCanonicalityStateParams) error { + _, err := q.db.ExecContext(ctx, UpdateBatchCanonicalityState, arg.BatchTxid, arg.State, arg.UpdatedAt) + return err +} + +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, + confirmation_pk_script +) VALUES ( + $1, $2, $3, $4, $5, $6, $7, $8, $9 +) +ON CONFLICT (batch_txid) DO UPDATE SET + state = EXCLUDED.state, + confirmation_height = EXCLUDED.confirmation_height, + 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 +` + +type UpsertBatchCanonicalityParams struct { + BatchTxid []byte + State int32 + ConfirmationHeight sql.NullInt32 + ConfirmationBlockHash []byte + CsvExpiryDelta int32 + PolicyState int32 + CreatedAt int64 + UpdatedAt int64 + ConfirmationPkScript []byte +} + +// Batch canonicality queries. +// These maintain the durable, reorg-aware record of how each batch +// (commitment) transaction is faring against the best chain, the inputs it +// consumes, the VTXOs it anchors, and the reverse-dependency edges needed to +// restore a provisionally consumed VTXO. The queries are behavior-free; all +// interpretation lives in the batch canonicality manager. +// UpsertBatchCanonicality inserts or replaces the canonicality row for a +// batch. created_at is preserved on conflict; everything else is overwritten. +func (q *Queries) UpsertBatchCanonicality(ctx context.Context, arg UpsertBatchCanonicalityParams) error { + _, err := q.db.ExecContext(ctx, UpsertBatchCanonicality, + arg.BatchTxid, + arg.State, + arg.ConfirmationHeight, + arg.ConfirmationBlockHash, + arg.CsvExpiryDelta, + arg.PolicyState, + arg.CreatedAt, + arg.UpdatedAt, + arg.ConfirmationPkScript, + ) + return err +} diff --git a/db/sqlc/migrations/000011_batch_canonicality.down.sql b/db/sqlc/migrations/000011_batch_canonicality.down.sql new file mode 100644 index 000000000..c21529715 --- /dev/null +++ b/db/sqlc/migrations/000011_batch_canonicality.down.sql @@ -0,0 +1,4 @@ +DROP TABLE IF EXISTS batch_provisional_consumers; +DROP TABLE IF EXISTS batch_dependent_vtxos; +DROP TABLE IF EXISTS batch_consumed_inputs; +DROP TABLE IF EXISTS batch_canonicality; diff --git a/db/sqlc/migrations/000011_batch_canonicality.up.sql b/db/sqlc/migrations/000011_batch_canonicality.up.sql new file mode 100644 index 000000000..498632459 --- /dev/null +++ b/db/sqlc/migrations/000011_batch_canonicality.up.sql @@ -0,0 +1,117 @@ +-- batch_canonicality is the durable, reorg-aware record of how each batch +-- (commitment) transaction is faring against the best chain. It is keyed by +-- the batch txid: identity is by txid, never by (txid, block hash), so a +-- reorg that re-mines the same batch in a different block is the same row. +-- +-- Effective (absolute) expiry is intentionally NOT stored. The row keeps the +-- CSV-relative delta plus the current confirmation height; the effective +-- expiry is derived as confirmation_height + csv_expiry_delta and is therefore +-- recomputed on every (re)confirmation rather than frozen at first +-- confirmation. Expiry is never persisted as a one-way terminal fact. +CREATE TABLE IF NOT EXISTS batch_canonicality ( + -- batch_txid is the 32-byte commitment transaction id and primary key. + batch_txid BLOB NOT NULL CHECK (length(batch_txid) = 32), + + -- state is the interpreted canonicality state (batchcanon.State): + -- 0 = unseen + -- 1 = provisional + -- 2 = finalized + -- 3 = reorged_out + -- 4 = conflict_provisional + -- 5 = conflict_finalized + -- Values are append-only and must never be renumbered. + state INTEGER NOT NULL DEFAULT 0, + + -- confirmation_height is the best-chain height at which the batch tx is + -- currently observed confirmed. NULL means the batch is not currently + -- confirmed (unseen or reorged out). A reorg clears it; a reconfirmation + -- sets it to the new height. + confirmation_height INTEGER, + + -- confirmation_block_hash is the hash of the block currently confirming + -- the batch tx. It is an observation attribute only and is NOT part of + -- the batch identity. NULL when not currently confirmed. + confirmation_block_hash BLOB + CHECK (confirmation_block_hash IS NULL + OR length(confirmation_block_hash) = 32), + + -- csv_expiry_delta is the batch's CSV-relative expiry timeout, in blocks. + -- Combined with confirmation_height it yields the effective expiry. + csv_expiry_delta INTEGER NOT NULL, + + -- policy_state is a reserved policy classification slot + -- (batchcanon.PolicyState); 0 = default. The data-model layer persists + -- and round-trips it but assigns no business meaning. + policy_state INTEGER NOT NULL DEFAULT 0, + + -- created_at / updated_at are unix timestamps. + created_at BIGINT NOT NULL, + updated_at BIGINT NOT NULL, + + PRIMARY KEY (batch_txid) +); + +-- Index supporting "find every batch in a given state" (e.g. all provisional +-- batches the manager must re-check for finality). +CREATE INDEX IF NOT EXISTS idx_batch_canonicality_state + ON batch_canonicality(state); + +-- batch_consumed_inputs records the outpoints each batch tx spends, so the +-- canonicality manager can watch every consumed input for a conflicting +-- spend. +CREATE TABLE IF NOT EXISTS batch_consumed_inputs ( + batch_txid BLOB NOT NULL CHECK (length(batch_txid) = 32), + input_hash BLOB NOT NULL CHECK (length(input_hash) = 32), + input_index INTEGER NOT NULL CHECK (input_index >= 0), + + PRIMARY KEY (batch_txid, input_hash, input_index), + FOREIGN KEY (batch_txid) + REFERENCES batch_canonicality(batch_txid) ON DELETE CASCADE +); + +-- Index supporting input-conflict detection: given an outpoint, find every +-- batch that consumes it (two batches consuming the same outpoint conflict). +CREATE INDEX IF NOT EXISTS idx_batch_consumed_inputs_outpoint + ON batch_consumed_inputs(input_hash, input_index); + +-- batch_dependent_vtxos records the VTXO outpoints anchored by each batch. +-- Their derived availability follows the batch's canonicality. There is +-- intentionally no FK to vtxos: a batch may anchor VTXOs the local wallet +-- does not own or persist. +CREATE TABLE IF NOT EXISTS batch_dependent_vtxos ( + batch_txid BLOB NOT NULL CHECK (length(batch_txid) = 32), + vtxo_outpoint_hash BLOB NOT NULL CHECK (length(vtxo_outpoint_hash) = 32), + vtxo_outpoint_index INTEGER NOT NULL CHECK (vtxo_outpoint_index >= 0), + + PRIMARY KEY (batch_txid, vtxo_outpoint_hash, vtxo_outpoint_index), + FOREIGN KEY (batch_txid) + REFERENCES batch_canonicality(batch_txid) ON DELETE CASCADE +); + +-- Index supporting "given a VTXO outpoint, which batch anchors it". +CREATE INDEX IF NOT EXISTS idx_batch_dependent_vtxos_vtxo + ON batch_dependent_vtxos(vtxo_outpoint_hash, vtxo_outpoint_index); + +-- batch_provisional_consumers is the reverse-dependency table that lets a +-- provisionally consumed VTXO be restored if its consumer batch never becomes +-- canonical (e.g. a round-2 forfeit whose commitment tx is reorged out must +-- restore the round-1 VTXO it consumed). Each row says "consumed_vtxo is +-- provisionally consumed by consumer_batch". +CREATE TABLE IF NOT EXISTS batch_provisional_consumers ( + consumed_vtxo_hash BLOB NOT NULL CHECK (length(consumed_vtxo_hash) = 32), + consumed_vtxo_index INTEGER NOT NULL CHECK (consumed_vtxo_index >= 0), + consumer_batch_txid BLOB NOT NULL + CHECK (length(consumer_batch_txid) = 32), + created_at BIGINT NOT NULL, + + PRIMARY KEY ( + consumed_vtxo_hash, consumed_vtxo_index, consumer_batch_txid + ), + FOREIGN KEY (consumer_batch_txid) + REFERENCES batch_canonicality(batch_txid) ON DELETE CASCADE +); + +-- Index supporting "given an invalidated consumer batch, list the VTXOs to +-- restore". +CREATE INDEX IF NOT EXISTS idx_batch_prov_consumers_batch + ON batch_provisional_consumers(consumer_batch_txid); 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 f68c79c07..251dd87cc 100644 --- a/db/sqlc/models.go +++ b/db/sqlc/models.go @@ -61,6 +61,37 @@ type ActivityStatus struct { Name string } +type BatchCanonicality struct { + BatchTxid []byte + State int32 + ConfirmationHeight sql.NullInt32 + ConfirmationBlockHash []byte + CsvExpiryDelta int32 + PolicyState int32 + CreatedAt int64 + UpdatedAt int64 + ConfirmationPkScript []byte +} + +type BatchConsumedInput struct { + BatchTxid []byte + InputHash []byte + InputIndex int32 +} + +type BatchDependentVtxo struct { + BatchTxid []byte + VtxoOutpointHash []byte + VtxoOutpointIndex int32 +} + +type BatchProvisionalConsumer struct { + ConsumedVtxoHash []byte + ConsumedVtxoIndex int32 + ConsumerBatchTxid []byte + CreatedAt int64 +} + type BoardingAddress struct { PkScript []byte AddressString string diff --git a/db/sqlc/querier.go b/db/sqlc/querier.go index ec186c3b8..8efbbacd4 100644 --- a/db/sqlc/querier.go +++ b/db/sqlc/querier.go @@ -15,6 +15,9 @@ type Querier interface { // contiguous). Callers use it as the resumable-subscribe cursor for the update. AppendActivityEvent(ctx context.Context, arg AppendActivityEventParams) (int64, error) CancelVHTLCRecoveryJob(ctx context.Context, arg CancelVHTLCRecoveryJobParams) (int64, error) + // ClearBatchConfirmation nulls the confirmation observation, reflecting that + // the confirming block left the best chain. It sets no terminal flag. + ClearBatchConfirmation(ctx context.Context, arg ClearBatchConfirmationParams) error ClearPendingIntentAnchorByOutpoint(ctx context.Context, arg ClearPendingIntentAnchorByOutpointParams) error CompleteVHTLCRecoveryJob(ctx context.Context, arg CompleteVHTLCRecoveryJobParams) (int64, error) // CountActivityEntriesByStatus returns the number of current-state rows in the @@ -28,6 +31,12 @@ type Querier interface { // CountVTXOsByStatus returns the count of VTXOs with the specified status. CountVTXOsByStatus(ctx context.Context, status int32) (int64, error) CountWalletUTXOLog(ctx context.Context) (int64, error) + // DeleteBatchConsumedInputs removes every consumed-input row for a batch, + // used by the store's upsert to replace the set atomically. + DeleteBatchConsumedInputs(ctx context.Context, batchTxid []byte) error + // DeleteBatchDependentVTXOs removes every dependent-VTXO row for a batch, + // used by the store's upsert to replace the set atomically. + DeleteBatchDependentVTXOs(ctx context.Context, batchTxid []byte) error DeleteClientTreeTxids(ctx context.Context, arg DeleteClientTreeTxidsParams) error DeleteOORPackageCheckpoints(ctx context.Context, sessionID []byte) error DeleteOrphanedPendingBoardIntents(ctx context.Context) error @@ -41,6 +50,9 @@ type Querier interface { DeletePendingIntentsByKind(ctx context.Context, kind string) error DeletePendingSendIntentByID(ctx context.Context, intentID []byte) error DeletePendingSendIntentsAll(ctx context.Context) error + // DeleteProvisionalConsumersForBatch removes every reverse-dependency edge + // for the given consumer batch. + DeleteProvisionalConsumersForBatch(ctx context.Context, consumerBatchTxid []byte) error // DeleteSpendingReservation removes the reservation for one outpoint. Called // when the VTXO leaves SpendingState (released or completed). DeleteSpendingReservation(ctx context.Context, arg DeleteSpendingReservationParams) error @@ -54,8 +66,14 @@ type Querier interface { EscalateVHTLCRecoveryJob(ctx context.Context, arg EscalateVHTLCRecoveryJobParams) (int64, error) FailVHTLCRecoveryJob(ctx context.Context, arg FailVHTLCRecoveryJobParams) (int64, error) FinalizeRound(ctx context.Context, arg FinalizeRoundParams) error + // FindBatchesByConsumedOutpoint returns the txids of every batch that + // consumes the given outpoint. + 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. 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) GetBoardingSweep(ctx context.Context, txid []byte) (BoardingSweep, error) @@ -97,6 +115,10 @@ type Querier interface { // GetVTXOReplacement retrieves the replacement VTXO outpoint for a forfeited // VTXO. Returns NULL if not forfeited or no replacement recorded. GetVTXOReplacement(ctx context.Context, arg GetVTXOReplacementParams) (GetVTXOReplacementRow, error) + // InsertBatchConsumedInput records one outpoint consumed by a batch. + InsertBatchConsumedInput(ctx context.Context, arg InsertBatchConsumedInputParams) error + // InsertBatchDependentVTXO records one VTXO outpoint anchored by a batch. + InsertBatchDependentVTXO(ctx context.Context, arg InsertBatchDependentVTXOParams) error // Boarding address queries. InsertBoardingAddress(ctx context.Context, arg InsertBoardingAddressParams) error // Boarding intent queries. @@ -126,6 +148,9 @@ type Querier interface { InsertClientTreeTxid(ctx context.Context, arg InsertClientTreeTxidParams) error InsertMacaroonRootKey(ctx context.Context, arg InsertMacaroonRootKeyParams) error InsertOORPackageCheckpoint(ctx context.Context, arg InsertOORPackageCheckpointParams) error + // InsertProvisionalConsumer records a reverse-dependency edge: consumed_vtxo + // is provisionally consumed by consumer_batch. Idempotent. + InsertProvisionalConsumer(ctx context.Context, arg InsertProvisionalConsumerParams) error // Round queries. InsertRound(ctx context.Context, arg InsertRoundParams) error // Round boarding intents queries. @@ -161,6 +186,13 @@ type Querier interface { ListAllCreditOperations(ctx context.Context) ([]CreditOperation, error) ListAllOORSessionRegistry(ctx context.Context) ([]OorSessionRegistry, error) ListAllVTXOs(ctx context.Context) ([]Vtxo, error) + // ListBatchCanonicalityByState returns every batch currently in the given + // state. + ListBatchCanonicalityByState(ctx context.Context, state int32) ([]BatchCanonicality, error) + // ListBatchConsumedInputs returns the outpoints a batch consumes. + ListBatchConsumedInputs(ctx context.Context, batchTxid []byte) ([]ListBatchConsumedInputsRow, error) + // ListBatchDependentVTXOs returns the VTXO outpoints a batch anchors. + ListBatchDependentVTXOs(ctx context.Context, batchTxid []byte) ([]ListBatchDependentVTXOsRow, error) ListBoardingIntentOutpoints(ctx context.Context) ([]ListBoardingIntentOutpointsRow, error) ListBoardingIntentsByConfHeight(ctx context.Context, confHeight int32) ([]BoardingIntent, error) ListBoardingIntentsByPkScript(ctx context.Context, pkScript []byte) ([]BoardingIntent, error) @@ -210,6 +242,10 @@ type Querier interface { ListPendingBoardingSweeps(ctx context.Context) ([]BoardingSweep, error) ListPendingIntentAnchorsByKind(ctx context.Context, kind string) ([]PendingIntentAnchor, error) ListPendingSendIntents(ctx context.Context) ([]ListPendingSendIntentsRow, error) + // ListProvisionalConsumersForBatch returns the VTXO outpoints that the given + // consumer batch provisionally consumes (the VTXOs to restore if the batch + // is invalidated). + ListProvisionalConsumersForBatch(ctx context.Context, consumerBatchTxid []byte) ([]ListProvisionalConsumersForBatchRow, error) ListRoundsByStatus(ctx context.Context, status string) ([]Round, error) // ListRoundsPaginated returns rounds ordered by round_id with cursor- // based pagination. When cursor is empty, returns from the beginning. @@ -249,6 +285,13 @@ type Querier interface { // management, including status transitions and forfeit transaction tracking. // ListVTXOsByStatus returns all VTXOs with the specified status. ListVTXOsByStatus(ctx context.Context, status int32) ([]Vtxo, error) + // ListVTXOsForCanonicalityBackfill returns the columns needed to derive + // initial batch canonicality records from already-persisted VTXOs: each + // VTXO's outpoint, its commitment (batch) txid, the absolute batch expiry + // height, and the height at which it was created (confirmed). The backfill + // groups these by commitment txid in Go and recomputes the CSV-relative + // expiry delta as batch_expiry - created_height. + ListVTXOsForCanonicalityBackfill(ctx context.Context) ([]ListVTXOsForCanonicalityBackfillRow, error) ListWalletUTXOLog(ctx context.Context, arg ListWalletUTXOLogParams) ([]WalletUtxoLog, error) ListWalletUTXOLogByBlock(ctx context.Context, blockHeight int32) ([]WalletUtxoLog, error) ListWalletUTXOLogByClassification(ctx context.Context, arg ListWalletUTXOLogByClassificationParams) ([]WalletUtxoLog, error) @@ -287,8 +330,16 @@ type Querier interface { // PullActivityEvents returns transition rows strictly after the cursor in // event_seq order, the resumable-subscribe replay primitive. PullActivityEvents(ctx context.Context, arg PullActivityEventsParams) ([]ActivityEvent, error) + // RecordBatchConfirmation records the best-chain height and block hash at + // which the batch tx is confirmed. A later call at a different height (after + // a reorg) overwrites the observation so effective expiry tracks the new + // confirmation. + RecordBatchConfirmation(ctx context.Context, arg RecordBatchConfirmationParams) error SumBoardingIntentAmountsByStatus(ctx context.Context, status string) (interface{}, error) SumUnspentVTXOAmounts(ctx context.Context) (interface{}, error) + // UpdateBatchCanonicalityState transitions a batch to a new state without + // touching its other fields. + UpdateBatchCanonicalityState(ctx context.Context, arg UpdateBatchCanonicalityStateParams) error UpdateBoardingIntentStatus(ctx context.Context, arg UpdateBoardingIntentStatusParams) error UpdateRoundBoardingIntentSignature(ctx context.Context, arg UpdateRoundBoardingIntentSignatureParams) error UpdateRoundStatus(ctx context.Context, arg UpdateRoundStatusParams) error @@ -305,6 +356,15 @@ type Querier interface { // correlation handles are COALESCEd so an early projection that does not yet // know a txid never clobbers one a later projection already recorded. UpsertActivityEntry(ctx context.Context, arg UpsertActivityEntryParams) error + // Batch canonicality queries. + // These maintain the durable, reorg-aware record of how each batch + // (commitment) transaction is faring against the best chain, the inputs it + // consumes, the VTXOs it anchors, and the reverse-dependency edges needed to + // restore a provisionally consumed VTXO. The queries are behavior-free; all + // interpretation lives in the batch canonicality manager. + // UpsertBatchCanonicality inserts or replaces the canonicality row for a + // batch. created_at is preserved on conflict; everything else is overwritten. + UpsertBatchCanonicality(ctx context.Context, arg UpsertBatchCanonicalityParams) error UpsertChainInfo(ctx context.Context, arg UpsertChainInfoParams) error // Credit operations control-plane queries. UpsertCreditOperation(ctx context.Context, arg UpsertCreditOperationParams) error diff --git a/db/sqlc/queries/batch_canonicality.sql b/db/sqlc/queries/batch_canonicality.sql new file mode 100644 index 000000000..8be642c5a --- /dev/null +++ b/db/sqlc/queries/batch_canonicality.sql @@ -0,0 +1,143 @@ +-- Batch canonicality queries. +-- These maintain the durable, reorg-aware record of how each batch +-- (commitment) transaction is faring against the best chain, the inputs it +-- consumes, the VTXOs it anchors, and the reverse-dependency edges needed to +-- restore a provisionally consumed VTXO. The queries are behavior-free; all +-- interpretation lives in the batch canonicality manager. + +-- name: UpsertBatchCanonicality :exec +-- UpsertBatchCanonicality inserts or replaces the canonicality row for a +-- 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, + confirmation_pk_script +) VALUES ( + $1, $2, $3, $4, $5, $6, $7, $8, $9 +) +ON CONFLICT (batch_txid) DO UPDATE SET + state = EXCLUDED.state, + confirmation_height = EXCLUDED.confirmation_height, + 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. 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, + confirmation_pk_script +FROM batch_canonicality +WHERE batch_txid = $1; + +-- name: ListBatchCanonicalityByState :many +-- 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, + confirmation_pk_script +FROM batch_canonicality +WHERE state = $1; + +-- name: UpdateBatchCanonicalityState :exec +-- UpdateBatchCanonicalityState transitions a batch to a new state without +-- touching its other fields. +UPDATE batch_canonicality +SET state = $2, updated_at = $3 +WHERE batch_txid = $1; + +-- name: RecordBatchConfirmation :exec +-- RecordBatchConfirmation records the best-chain height and block hash at +-- which the batch tx is confirmed. A later call at a different height (after +-- a reorg) overwrites the observation so effective expiry tracks the new +-- confirmation. +UPDATE batch_canonicality +SET confirmation_height = $2, confirmation_block_hash = $3, updated_at = $4 +WHERE batch_txid = $1; + +-- name: ClearBatchConfirmation :exec +-- ClearBatchConfirmation nulls the confirmation observation, reflecting that +-- the confirming block left the best chain. It sets no terminal flag. +UPDATE batch_canonicality +SET confirmation_height = NULL, confirmation_block_hash = NULL, updated_at = $2 +WHERE batch_txid = $1; + +-- name: InsertBatchConsumedInput :exec +-- InsertBatchConsumedInput records one outpoint consumed by a batch. +INSERT INTO batch_consumed_inputs (batch_txid, input_hash, input_index) +VALUES ($1, $2, $3) +ON CONFLICT (batch_txid, input_hash, input_index) DO NOTHING; + +-- name: DeleteBatchConsumedInputs :exec +-- DeleteBatchConsumedInputs removes every consumed-input row for a batch, +-- used by the store's upsert to replace the set atomically. +DELETE FROM batch_consumed_inputs WHERE batch_txid = $1; + +-- name: ListBatchConsumedInputs :many +-- ListBatchConsumedInputs returns the outpoints a batch consumes. +SELECT input_hash, input_index +FROM batch_consumed_inputs +WHERE batch_txid = $1; + +-- name: FindBatchesByConsumedOutpoint :many +-- FindBatchesByConsumedOutpoint returns the txids of every batch that +-- consumes the given outpoint. +SELECT batch_txid +FROM batch_consumed_inputs +WHERE input_hash = $1 AND input_index = $2; + +-- name: InsertBatchDependentVTXO :exec +-- InsertBatchDependentVTXO records one VTXO outpoint anchored by a batch. +INSERT INTO batch_dependent_vtxos ( + batch_txid, vtxo_outpoint_hash, vtxo_outpoint_index +) VALUES ($1, $2, $3) +ON CONFLICT (batch_txid, vtxo_outpoint_hash, vtxo_outpoint_index) DO NOTHING; + +-- name: DeleteBatchDependentVTXOs :exec +-- DeleteBatchDependentVTXOs removes every dependent-VTXO row for a batch, +-- used by the store's upsert to replace the set atomically. +DELETE FROM batch_dependent_vtxos WHERE batch_txid = $1; + +-- name: ListBatchDependentVTXOs :many +-- ListBatchDependentVTXOs returns the VTXO outpoints a batch anchors. +SELECT vtxo_outpoint_hash, vtxo_outpoint_index +FROM batch_dependent_vtxos +WHERE batch_txid = $1; + +-- name: InsertProvisionalConsumer :exec +-- InsertProvisionalConsumer records a reverse-dependency edge: consumed_vtxo +-- is provisionally consumed by consumer_batch. Idempotent. +INSERT INTO batch_provisional_consumers ( + consumed_vtxo_hash, consumed_vtxo_index, consumer_batch_txid, created_at +) VALUES ($1, $2, $3, $4) +ON CONFLICT ( + consumed_vtxo_hash, consumed_vtxo_index, consumer_batch_txid +) DO NOTHING; + +-- name: ListProvisionalConsumersForBatch :many +-- ListProvisionalConsumersForBatch returns the VTXO outpoints that the given +-- consumer batch provisionally consumes (the VTXOs to restore if the batch +-- is invalidated). +SELECT consumed_vtxo_hash, consumed_vtxo_index +FROM batch_provisional_consumers +WHERE consumer_batch_txid = $1; + +-- name: DeleteProvisionalConsumersForBatch :exec +-- DeleteProvisionalConsumersForBatch removes every reverse-dependency edge +-- for the given consumer batch. +DELETE FROM batch_provisional_consumers WHERE consumer_batch_txid = $1; + +-- name: ListVTXOsForCanonicalityBackfill :many +-- ListVTXOsForCanonicalityBackfill returns the columns needed to derive +-- initial batch canonicality records from already-persisted VTXOs: each +-- VTXO's outpoint, its commitment (batch) txid, the absolute batch expiry +-- height, and the height at which it was created (confirmed). The backfill +-- groups these by commitment txid in Go and recomputes the CSV-relative +-- expiry delta as batch_expiry - created_height. +SELECT outpoint_hash, outpoint_index, commitment_txid, batch_expiry, + created_height +FROM vtxos +WHERE length(commitment_txid) = 32; diff --git a/db/sqlc/schemas/generated_schema.sql b/db/sqlc/schemas/generated_schema.sql index 27f17a868..acfaffa4f 100644 --- a/db/sqlc/schemas/generated_schema.sql +++ b/db/sqlc/schemas/generated_schema.sql @@ -98,6 +98,83 @@ CREATE TABLE ask_results ( expires_at BIGINT NOT NULL ); +CREATE TABLE batch_canonicality ( + -- batch_txid is the 32-byte commitment transaction id and primary key. + batch_txid BLOB NOT NULL CHECK (length(batch_txid) = 32), + + -- state is the interpreted canonicality state (batchcanon.State): + -- 0 = unseen + -- 1 = provisional + -- 2 = finalized + -- 3 = reorged_out + -- 4 = conflict_provisional + -- 5 = conflict_finalized + -- Values are append-only and must never be renumbered. + state INTEGER NOT NULL DEFAULT 0, + + -- confirmation_height is the best-chain height at which the batch tx is + -- currently observed confirmed. NULL means the batch is not currently + -- confirmed (unseen or reorged out). A reorg clears it; a reconfirmation + -- sets it to the new height. + confirmation_height INTEGER, + + -- confirmation_block_hash is the hash of the block currently confirming + -- the batch tx. It is an observation attribute only and is NOT part of + -- the batch identity. NULL when not currently confirmed. + confirmation_block_hash BLOB + CHECK (confirmation_block_hash IS NULL + OR length(confirmation_block_hash) = 32), + + -- csv_expiry_delta is the batch's CSV-relative expiry timeout, in blocks. + -- Combined with confirmation_height it yields the effective expiry. + csv_expiry_delta INTEGER NOT NULL, + + -- policy_state is a reserved policy classification slot + -- (batchcanon.PolicyState); 0 = default. The data-model layer persists + -- and round-trips it but assigns no business meaning. + policy_state INTEGER NOT NULL DEFAULT 0, + + -- created_at / updated_at are unix timestamps. + created_at BIGINT NOT NULL, + updated_at BIGINT NOT NULL, confirmation_pk_script BLOB, + + PRIMARY KEY (batch_txid) +); + +CREATE TABLE batch_consumed_inputs ( + batch_txid BLOB NOT NULL CHECK (length(batch_txid) = 32), + input_hash BLOB NOT NULL CHECK (length(input_hash) = 32), + input_index INTEGER NOT NULL CHECK (input_index >= 0), + + PRIMARY KEY (batch_txid, input_hash, input_index), + FOREIGN KEY (batch_txid) + REFERENCES batch_canonicality(batch_txid) ON DELETE CASCADE +); + +CREATE TABLE batch_dependent_vtxos ( + batch_txid BLOB NOT NULL CHECK (length(batch_txid) = 32), + vtxo_outpoint_hash BLOB NOT NULL CHECK (length(vtxo_outpoint_hash) = 32), + vtxo_outpoint_index INTEGER NOT NULL CHECK (vtxo_outpoint_index >= 0), + + PRIMARY KEY (batch_txid, vtxo_outpoint_hash, vtxo_outpoint_index), + FOREIGN KEY (batch_txid) + REFERENCES batch_canonicality(batch_txid) ON DELETE CASCADE +); + +CREATE TABLE batch_provisional_consumers ( + consumed_vtxo_hash BLOB NOT NULL CHECK (length(consumed_vtxo_hash) = 32), + consumed_vtxo_index INTEGER NOT NULL CHECK (consumed_vtxo_index >= 0), + consumer_batch_txid BLOB NOT NULL + CHECK (length(consumer_batch_txid) = 32), + created_at BIGINT NOT NULL, + + PRIMARY KEY ( + consumed_vtxo_hash, consumed_vtxo_index, consumer_batch_txid + ), + FOREIGN KEY (consumer_batch_txid) + REFERENCES batch_canonicality(batch_txid) ON DELETE CASCADE +); + CREATE TABLE boarding_addresses ( -- pk_script is the raw output script (P2TR script) and serves as the -- primary key since it uniquely identifies an address. @@ -410,6 +487,18 @@ CREATE INDEX idx_activity_events_canonical CREATE INDEX idx_ask_results_expires ON ask_results(expires_at); +CREATE INDEX idx_batch_canonicality_state + ON batch_canonicality(state); + +CREATE INDEX idx_batch_consumed_inputs_outpoint + ON batch_consumed_inputs(input_hash, input_index); + +CREATE INDEX idx_batch_dependent_vtxos_vtxo + ON batch_dependent_vtxos(vtxo_outpoint_hash, vtxo_outpoint_index); + +CREATE INDEX idx_batch_prov_consumers_batch + ON batch_provisional_consumers(consumer_batch_txid); + CREATE INDEX idx_boarding_addresses_creation_time ON boarding_addresses(creation_time DESC); diff --git a/db/store.go b/db/store.go index b1ee2da11..819dc99ee 100644 --- a/db/store.go +++ b/db/store.go @@ -347,6 +347,29 @@ func (s *Store) NewActivityStore(clk clock.Clock) *ActivityPersistenceStore { return NewActivityPersistenceStore(activityDB, clk) } +// NewBatchCanonicalityStore builds the batch canonicality persistence store +// with transactional query execution. +// +// The store holds the durable, reorg-aware record of how each batch +// (commitment) tx is faring against the best chain, plus the reverse +// dependencies needed to restore a provisionally consumed VTXO. It is +// behavior-free; interpretation lives in the batch canonicality manager. +func (s *Store) NewBatchCanonicalityStore( + clk clock.Clock) *BatchCanonicalityPersistenceStore { + + baseDB := s.BaseDB() + + canonDB := NewTransactionExecutor( + baseDB, + func(tx *sql.Tx) BatchCanonicalityStore { + return s.queries.WithTx(tx) + }, + s.log, + ) + + return NewBatchCanonicalityPersistenceStore(canonDB, clk) +} + // NewUnilateralExitStore builds the unilateral-exit persistence store with // transactional query execution. func (s *Store) NewUnilateralExitStore(