tapgarden: hardening phase two (batch invariants)

cmd/tapd/main.go

@@ -35,6 +35,20 @@ func main() {
 		os.Exit(0)
 	}
 
+	// If the operator has invoked tapd in one-shot repair mode, run
+	// the requested repair against the database and exit before
+	// constructing the full server. The repair tool opens the DB
+	// with migrations skipped, so it can recover a legacy DB whose
+	// state would otherwise block a migration from applying.
+	if cfg.Repair != nil && cfg.Repair.CancelDuplicateBatches {
+		err := tapcfg.RunRepairTool(cfg, cfgLogger, shutdownInterceptor)
+		if err != nil {
+			_, _ = fmt.Fprintln(os.Stderr, err)
+			os.Exit(1)
+		}
+		os.Exit(0)
+	}
+
 	// Enable http profiling server if requested.
 	if cfg.Profile != "" {
 		go func() {

docs/release-notes/release-notes-0.9.0.md

@@ -25,6 +25,10 @@
   fixes a bug that could cause minted assets to commit to the wrong
   address.
 
+* [PR#2203](https://github.com/lightninglabs/taproot-assets/pull/2203)
+  fixes a bug that could leave invalid minting batch state on disk after
+  failure.
+
 # New Features
 
 ## Functional Enhancements

rpcserver/rpcserver.go

@@ -5744,7 +5744,7 @@ func (r *RPCServer) SubscribeMintEvents(req *mintrpc.SubscribeMintEventsRequest,
 			return nil, fmt.Errorf("invalid event type: %T", event)
 		}
 
-		rpcState, err := marshalBatchState(e.BatchState)
+		rpcState, err := marshalBatchState(e.Batch.State())
 		if err != nil {
 			return nil, fmt.Errorf("error marshaling batch state: "+
 				"%w", err)

sample-tapd.conf

@@ -543,4 +543,13 @@
 
 ; The amount of time we should wait between certificate expiration health checks.
 ; This value must be >= 1m.
-; healthcheck.tls.interval=1m
+; healthcheck.tls.interval=1m
+
+[repair]
+
+; If set, tapd cancels all but the most recent minting batch in
+; BatchStatePending or BatchStateFrozen and then exits. Used to recover
+; from a database that violates the singleton pre-broadcast batch
+; invariant added in migration 000061 (e.g. a legacy DB with duplicate
+; pending batches that blocks the migration).
+; repair.cancel-duplicate-batches=false

tapcfg/config.go

@@ -368,6 +368,15 @@ type ExperimentalConfig struct {
 	Rfq rfq.CliConfig `group:"rfq" namespace:"rfq"`
 }
 
+// RepairConfig houses one-shot recovery flags that, when set, cause
+// tapd to perform a targeted repair action against the database and
+// exit before constructing the full server. These flags are intended
+// for operator use after a constraint or invariant failure has
+// prevented normal startup.
+type RepairConfig struct {
+	CancelDuplicateBatches bool `long:"cancel-duplicate-batches" description:"If set, tapd cancels all but the most recent minting batch in BatchStatePending or BatchStateFrozen and then exits. Used to recover from a database that violates the singleton pre-broadcast batch invariant added in migration 000061 (e.g. a legacy DB with duplicate pending batches that blocks the migration)."`
+}
+
 // CleanAndValidate performs final processing on the ExperimentalConfig,
 // returning an error if the configuration is invalid.
 func (c *ExperimentalConfig) CleanAndValidate() error {
@@ -428,6 +437,8 @@ type Config struct {
 
 	Experimental *ExperimentalConfig `group:"experimental" namespace:"experimental"`
 
+	Repair *RepairConfig `group:"repair" namespace:"repair"`
+
 	HealthChecks *HealthCheckConfig `group:"healthcheck" namespace:"healthcheck"`
 
 	// LogWriter is the root logger that all of the daemon's subloggers are
@@ -548,6 +559,7 @@ func DefaultConfig() Config {
 				AcceptPriceDeviationPpm: rfq.DefaultAcceptPriceDeviationPpm,
 			},
 		},
+		Repair: &RepairConfig{},
 		HealthChecks: &HealthCheckConfig{
 			TLSCheck: &CheckConfig{
 				Interval: defaultTLSInterval,

tapcfg/repair.go

@@ -0,0 +1,151 @@
+package tapcfg
+
+import (
+	"context"
+	"database/sql"
+	"fmt"
+	"sort"
+	"time"
+
+	"github.com/btcsuite/btclog/v2"
+	"github.com/lightninglabs/taproot-assets/tapdb"
+	"github.com/lightninglabs/taproot-assets/tapgarden"
+	"github.com/lightningnetwork/lnd/signal"
+)
+
+// RunRepairTool inspects the configured database for batches that
+// violate the singleton "≤ 1 in {Pending, Frozen}" invariant added
+// in migration 000060, and cancels all but the most recent. The
+// preserved batch is the one with the latest CreationTime; cancelled
+// batches transition to BatchStateSeedlingCancelled, leaving their
+// row and seedlings on disk for later inspection.
+//
+// The function opens the database with migrations skipped, so it can
+// run against a legacy database whose state would otherwise fail the
+// migration.
+//
+// NOTE: With migration 000061's self-heal in place, restarting tapd
+// normally will cancel the duplicates as part of applying the
+// migration. This tool is retained as a diagnostic that surfaces the
+// same repair outside the migration stream (e.g. after operator
+// intervention that re-introduces duplicates).
+func RunRepairTool(cfg *Config, cfgLogger btclog.Logger,
+	shutdownInterceptor signal.Interceptor) error {
+
+	// Derive a cancellable context that trips on shutdown. Without
+	// this, a Ctrl+C mid-repair would leave partial state behind
+	// (some batches cancelled, others not).
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+	go func() {
+		select {
+		case <-shutdownInterceptor.ShutdownChannel():
+			cancel()
+		case <-ctx.Done():
+		}
+	}()
+
+	// Open the database with migrations skipped. We want to inspect
+	// and repair a database whose state would otherwise prevent
+	// migration 000060 from applying; running migrations as part of
+	// opening the DB would defeat the purpose.
+	var (
+		db  tapdb.DatabaseBackend
+		err error
+	)
+	switch cfg.DatabaseBackend {
+	case DatabaseBackendSqlite:
+		sqliteCfg := *cfg.Sqlite
+		sqliteCfg.SkipMigrations = true
+		cfgLogger.Infof("repair: opening sqlite3 database at %v "+
+			"(migrations skipped)",
+			sqliteCfg.DatabaseFileName)
+		db, err = tapdb.NewSqliteStore(&sqliteCfg)
+
+	case DatabaseBackendPostgres:
+		pgCfg := *cfg.Postgres
+		pgCfg.SkipMigrations = true
+		cfgLogger.Infof("repair: opening postgres database " +
+			"(migrations skipped)")
+		db, err = tapdb.NewPostgresStore(&pgCfg)
+
+	default:
+		return fmt.Errorf("unknown database backend: %s",
+			cfg.DatabaseBackend)
+	}
+	if err != nil {
+		return fmt.Errorf("repair: unable to open database: %w", err)
+	}
+
+	mintingExec := tapdb.NewTransactionExecutor(
+		db, func(tx *sql.Tx) tapdb.PendingAssetStore {
+			return db.WithTx(tx)
+		},
+	)
+	store := tapdb.NewAssetMintingStore(mintingExec)
+
+	nonFinal, err := store.FetchNonFinalBatches(ctx)
+	if err != nil {
+		return fmt.Errorf("repair: unable to fetch non-final "+
+			"batches: %w", err)
+	}
+
+	var preBroadcast []*tapgarden.MintingBatch
+	for _, batch := range nonFinal {
+		switch batch.State() {
+		case tapgarden.BatchStatePending,
+			tapgarden.BatchStateFrozen:
+
+			preBroadcast = append(preBroadcast, batch)
+
+		default:
+			// Only pre-broadcast states are constrained by
+			// the singleton index; ignore everything else.
+		}
+	}
+
+	if len(preBroadcast) <= 1 {
+		cfgLogger.Infof("repair: nothing to do; found %d batches "+
+			"in pre-broadcast state", len(preBroadcast))
+		return nil
+	}
+
+	// Sort newest-first by CreationTime; preserve [0], cancel the
+	// rest. SliceStable gives a deterministic winner when two
+	// batches share a timestamp -- the input order (from
+	// FetchNonFinalBatches) then acts as the tiebreak.
+	sort.SliceStable(preBroadcast, func(i, j int) bool {
+		return preBroadcast[i].CreationTime.After(
+			preBroadcast[j].CreationTime,
+		)
+	})
+
+	preserved := preBroadcast[0]
+	cfgLogger.Infof("repair: preserving most recent pre-broadcast "+
+		"batch %x (state=%v, created=%s)",
+		preserved.BatchKey.PubKey.SerializeCompressed(),
+		preserved.State(),
+		preserved.CreationTime.Format(time.RFC3339))
+
+	for _, batch := range preBroadcast[1:] {
+		cfgLogger.Warnf("repair: cancelling pre-broadcast batch "+
+			"%x (state=%v, created=%s)",
+			batch.BatchKey.PubKey.SerializeCompressed(),
+			batch.State(),
+			batch.CreationTime.Format(time.RFC3339))
+
+		err := store.UpdateBatchState(
+			ctx, batch, tapgarden.BatchStateSeedlingCancelled,
+		)
+		if err != nil {
+			return fmt.Errorf("repair: unable to cancel batch "+
+				"%x: %w",
+				batch.BatchKey.PubKey.SerializeCompressed(),
+				err)
+		}
+	}
+
+	cfgLogger.Infof("repair: complete; cancelled %d duplicate "+
+		"batches, preserved 1.", len(preBroadcast)-1)
+	return nil
+}

tapdb/asset_minting.go

@@ -1319,7 +1319,7 @@ func marshalMintingBatch(ctx context.Context, q PendingAssetStore,
 		return nil, err
 	}
 
-	batch.UpdateState(batchState)
+	batch.SetStateOnDBSuccess(batchState)
 
 	if len(dbBatch.TapscriptSibling) != 0 {
 		batchSibling, err := chainhash.NewHash(dbBatch.TapscriptSibling)
@@ -1472,15 +1472,22 @@ func marshalMintingBatch(ctx context.Context, q PendingAssetStore,
 
 // UpdateBatchState updates the state of a batch based on the batch key.
 func (a *AssetMintingStore) UpdateBatchState(ctx context.Context,
-	batchKey *btcec.PublicKey, newState tapgarden.BatchState) error {
+	batch *tapgarden.MintingBatch, newState tapgarden.BatchState) error {
 
+	batchKey := batch.BatchKey.PubKey
 	var writeTxOpts AssetStoreTxOptions
-	return a.db.ExecTx(ctx, &writeTxOpts, func(q PendingAssetStore) error {
+	err := a.db.ExecTx(ctx, &writeTxOpts, func(q PendingAssetStore) error {
 		return q.UpdateMintingBatchState(ctx, BatchStateUpdate{
 			RawKey:     batchKey.SerializeCompressed(),
 			BatchState: int16(newState),
 		})
 	})
+	if err != nil {
+		return err
+	}
+
+	batch.SetStateOnDBSuccess(newState)
+	return nil
 }
 
 // encodeOutpoint encodes the outpoint point in Bitcoin wire format, returning
@@ -1793,7 +1800,7 @@ func fetchSeedlingGroups(ctx context.Context, q PendingAssetStore,
 // binds the genesis transaction (which will create the set of assets in the
 // batch) to the batch itself.
 func (a *AssetMintingStore) AddSproutsToBatch(ctx context.Context,
-	batchKey *btcec.PublicKey,
+	batch *tapgarden.MintingBatch,
 	genesisPacket *tapgarden.FundedMintAnchorPsbt,
 	assetRoot *commitment.TapCommitment) error {
 
@@ -1824,10 +1831,11 @@ func (a *AssetMintingStore) AddSproutsToBatch(ctx context.Context,
 		return err
 	}
 
+	batchKey := batch.BatchKey.PubKey
 	rawBatchKey := batchKey.SerializeCompressed()
 
 	var writeTxOpts AssetStoreTxOptions
-	return a.db.ExecTx(ctx, &writeTxOpts, func(q PendingAssetStore) error {
+	err = a.db.ExecTx(ctx, &writeTxOpts, func(q PendingAssetStore) error {
 		// Upsert the assets with genesis.
 		_, _, err := upsertAssetsWithGenesis(
 			ctx, q, genesisOutpoint, sortedAssets, nil,
@@ -1852,6 +1860,12 @@ func (a *AssetMintingStore) AddSproutsToBatch(ctx context.Context,
 			BatchState: int16(tapgarden.BatchStateCommitted),
 		})
 	})
+	if err != nil {
+		return err
+	}
+
+	batch.SetStateOnDBSuccess(tapgarden.BatchStateCommitted)
+	return nil
 }
 
 // CommitSignedGenesisTx binds a fully signed genesis transaction to a pending
@@ -1863,10 +1877,12 @@ func (a *AssetMintingStore) AddSproutsToBatch(ctx context.Context,
 // TODO(roasbeef): or could just re-read assets from disk and set the script
 // root manually?
 func (a *AssetMintingStore) CommitSignedGenesisTx(ctx context.Context,
-	batchKey *btcec.PublicKey, genesisPkt *tapsend.FundedPsbt,
+	batch *tapgarden.MintingBatch, genesisPkt *tapsend.FundedPsbt,
 	anchorOutputIndex uint32, merkleRoot, tapTreeRoot []byte,
 	tapSibling []byte) error {
 
+	batchKey := batch.BatchKey.PubKey
+
 	// The managed UTXO we'll insert only contains the raw tx of the
 	// genesis packet, so we'll extract that now.
 	//
@@ -1902,7 +1918,7 @@ func (a *AssetMintingStore) CommitSignedGenesisTx(ctx context.Context,
 	}
 
 	var writeTxOpts AssetStoreTxOptions
-	return a.db.ExecTx(ctx, &writeTxOpts, func(q PendingAssetStore) error {
+	err = a.db.ExecTx(ctx, &writeTxOpts, func(q PendingAssetStore) error {
 		// First, we'll update the genesis packet stored as part of the
 		// batch, as this packet is now fully signed.
 		pktBytes, err := fn.Serialize(genesisPkt.Pkt)
@@ -1977,19 +1993,26 @@ func (a *AssetMintingStore) CommitSignedGenesisTx(ctx context.Context,
 			BatchState: int16(tapgarden.BatchStateBroadcast),
 		})
 	})
+	if err != nil {
+		return err
+	}
+
+	batch.SetStateOnDBSuccess(tapgarden.BatchStateBroadcast)
+	return nil
 }
 
 // MarkBatchConfirmed stores final confirmation information for a batch on
 // disk.
 func (a *AssetMintingStore) MarkBatchConfirmed(ctx context.Context,
-	batchKey *btcec.PublicKey, blockHash *chainhash.Hash,
+	batch *tapgarden.MintingBatch, blockHash *chainhash.Hash,
 	blockHeight uint32, txIndex uint32,
 	mintingProofs proof.AssetBlobs) error {
 
+	batchKey := batch.BatchKey.PubKey
 	rawBatchKey := batchKey.SerializeCompressed()
 
 	var writeTxOpts AssetStoreTxOptions
-	return a.db.ExecTx(ctx, &writeTxOpts, func(q PendingAssetStore) error {
+	err := a.db.ExecTx(ctx, &writeTxOpts, func(q PendingAssetStore) error {
 		// First, we'll update the state of the target batch to reflect
 		// that the batch is fully finalized.
 		err := q.UpdateMintingBatchState(ctx, BatchStateUpdate{
@@ -2044,6 +2067,12 @@ func (a *AssetMintingStore) MarkBatchConfirmed(ctx context.Context,
 		}
 		return nil
 	})
+	if err != nil {
+		return err
+	}
+
+	batch.SetStateOnDBSuccess(tapgarden.BatchStateConfirmed)
+	return nil
 }
 
 // FetchGroupByGenesis fetches the asset group created by the genesis referenced