fix(blockassembly): stop re-marking the whole mempool on every reorg

CHANGELOG.md

@@ -106,6 +106,16 @@ Upgrading from v0.16.0 changes nothing for them; upgrading from an earlier relea
 
 ### Breaking — behaviour changes on upgrade
 
+- **A reorg no longer re-marks block assembly's whole mempool.** Block assembly marks
+  only the transactions it moved back and that are not mined on the new chain. Block
+  validation now repairs a block it marked mined that a competing chain later overtakes,
+  on its setMined worker, in slices of `blockvalidation_tipRepairSliceSize` transactions
+  (new setting, default 250000). Each slice can delay the next block's mined status by
+  one slice. The repair assumes a single block validation instance writes mined status
+  for the chain, and its state is in memory: fork blocks more than 10 below the tip at a
+  restart are not repaired. A deployment that runs more than one block validation
+  replica loses no data, but transactions can stay flagged unmined until block
+  assembly's next unmined load. No action needed for a single-instance deployment. (#2065)
 - **Catch-up quick validation waits for each block's `.subtree` writes before it commits
   the block, and a shutdown waits for a block's in-flight commit.** A block was stored with
   `subtrees_set` while its `.subtree` files could still be in flight, so a failed write

docs/references/settings/services/blockvalidation_settings.md

@@ -69,6 +69,7 @@ loader (0, false and false), so for these keys the two agree. The
 | MaxBlocksBehindBlockAssembly | int | 20 | blockvalidation_maxBlocksBehindBlockAssembly | **CRITICAL** - Max blocks behind block assembly |
 | PeriodicProcessingInterval | time.Duration | 1m | blockvalidation_periodic_processing_interval | Periodic processing interval |
 | RecentBlockIDsLimit | uint64 | 50000 | blockvalidation_recentBlockIDsLimit | **CRITICAL** - Fast-path double-spend checking window |
+| TipRepairSliceSize | int | 250000 | blockvalidation_tipRepairSliceSize | Transactions per overtaken-tip repair step on the setMined worker; bounds how long a step delays the next block's mined status |
 | MaxParallelForks | int | 4 | blockvalidation_max_parallel_forks | Maximum parallel fork processing |
 | MaxTrackedForks | int | 1000 | blockvalidation_max_tracked_forks | Maximum total forks tracked |
 | NearForkThreshold | int | 0 | blockvalidation_near_fork_threshold | Near fork detection (0=coinbase maturity/2) |

docs/references/stores/utxo_reference.md

@@ -256,7 +256,9 @@ type Store interface {
     // MarkTransactionsOnLongestChain marks transactions as being on the longest chain or not.
     // When onLongestChain is true, the unminedSince field is unset (transaction is mined).
     // When onLongestChain is false, the unminedSince field is set to the current block height.
-    MarkTransactionsOnLongestChain(ctx context.Context, txHashes []chainhash.Hash, onLongestChain bool) error
+    // WithSkipMissing() makes a record missing from the store a logged skip instead of an
+    // error for onLongestChain=false.
+    MarkTransactionsOnLongestChain(ctx context.Context, txHashes []chainhash.Hash, onLongestChain bool, opts ...MarkLongestChainOption) error
 
     // SetBlockHeight updates the current block height in the store.
     //

docs/topics/stores/utxo.md

@@ -632,8 +632,8 @@ The current writers, re-derived directly from the code (not from the settings/de
 | Service | File(s) | Writes | Why |
 |---|---|---|---|
 | **Validator** | `services/validator/Validator.go` | `SpendAndCreate` (creates the tx's outputs, spends its inputs), `SetLocked` | Core transaction validation: on accepting a new transaction it atomically creates its UTXOs and spends the ones it consumes. `SetLocked` completes a two-phase commit on the tx's own hash: when the tx is queued for block assembly its outputs are created locked (`WithLocked(true)`), and unlocked only once it has been added successfully (`twoPhaseCommitTransaction`) — a crash in between leaves them unspendable rather than spendable. |
-| **Block Validation** | `services/blockvalidation/BlockValidation.go`, `services/blockvalidation/quick_validate.go` | `SpendAndCreate` (`WithCreateOnly`/`WithSpendOnly`, quick-validate replay), `SetMinedMulti` (steady state, via `model.UpdateTxMinedStatus` in `setTxMined`), `utxo.SetMinedMultiChunked` (→ `SetMinedMulti`, quick-validate replay), `SetLocked` | Validates incoming blocks: creates/spends UTXOs for transactions not already known, marks the block's transactions as mined (records `BlockID`/`SubtreeIdx`), then — unless `quickValidateSkipsUtxoLock` applies to this block — unlocks the block's own subtree transactions at the end of quick-validate (`unlockSubtreeTransactions`). |
-| **Block Assembly** | `services/blockassembly/BlockAssembler.go`, `services/blockassembly/subtreeprocessor/SubtreeProcessor.go` | `SpendAndCreate` (`WithCreateOnly` — creates the coinbase UTXOs, `processCoinbaseUtxos`), `Delete` (coinbase and its child spends on reorg, `removeCoinbaseUtxos` — Block Assembly is the only service that calls `Delete` directly), `MarkTransactionsOnLongestChain`, `SetLocked` (unlocks unmined transactions reloaded at startup; unlocks conflict-locked parents when healing a stale conflict intent), `utxo.ProcessConflicting` / `utxo.ReverseProcessConflicting` (→ `SetConflicting`, `Unspend`, `SpendAndCreate`, `BeginConflictIntent`/`CompleteConflictIntent`) | Owns chain-tip bookkeeping during block building and reorgs: creates the coinbase UTXO and deletes it (with its child spends) on reorg, flags which transactions sit on the currently-longest chain, unlocks unmined transactions reloaded at startup and conflict-locked parents when healing a stale intent, and drives the conflicting-transaction state machine (marking/unmarking double-spends and reversing that on reorg). |
+| **Block Validation** | `services/blockvalidation/BlockValidation.go`, `services/blockvalidation/quick_validate.go`, `services/blockvalidation/overtaken_tip_repair.go` | `SpendAndCreate` (`WithCreateOnly`/`WithSpendOnly`, quick-validate replay), `SetMinedMulti` (steady state, via `model.UpdateTxMinedStatus` in `setTxMined`), `utxo.SetMinedMultiChunked` (→ `SetMinedMulti`, quick-validate replay), `SetLocked`, `MarkTransactionsOnLongestChain` (overtaken-tip repair, `overtaken_tip_repair.go`) | Validates incoming blocks: creates/spends UTXOs for transactions not already known, marks the block's transactions as mined (records `BlockID`/`SubtreeIdx`), re-marks the transactions of a block it marked mined that has since left the longest chain, then — unless `quickValidateSkipsUtxoLock` applies to this block — unlocks the block's own subtree transactions at the end of quick-validate (`unlockSubtreeTransactions`). |
+| **Block Assembly** | `services/blockassembly/BlockAssembler.go`, `services/blockassembly/subtreeprocessor/SubtreeProcessor.go` | `SpendAndCreate` (`WithCreateOnly` — creates the coinbase UTXOs, `processCoinbaseUtxos`), `Delete` (coinbase and its child spends on reorg, `removeCoinbaseUtxos` — Block Assembly is the only service that calls `Delete` directly), `MarkTransactionsOnLongestChain`, `SetLocked` (unlocks unmined transactions reloaded at startup; unlocks conflict-locked parents when healing a stale conflict intent), `utxo.ProcessConflicting` / `utxo.ReverseProcessConflicting` (→ `SetConflicting`, `Unspend`, `SpendAndCreate`, `BeginConflictIntent`/`CompleteConflictIntent`) | Owns chain-tip bookkeeping during block building and reorgs: creates the coinbase UTXO and deletes it (with its child spends) on reorg, flags which transactions sit on the currently-longest chain (on a reorg, only the transactions it moved, not its whole mempool), unlocks unmined transactions reloaded at startup and conflict-locked parents when healing a stale intent, and drives the conflicting-transaction state machine (marking/unmarking double-spends and reversing that on reorg). |
 | **Legacy (p2p sync)** | `services/legacy/netsync/handle_block.go` | `SpendAndCreate`, `utxo.SetMinedMultiChunked` (→ `SetMinedMulti`) | The legacy Bitcoin-protocol block-sync path validates and applies blocks received from legacy peers the same way Block Validation does for the native protocol — same create/spend/set-mined writes, different ingestion path. |
 | **Alert System** | `services/alert/node.go` | `FreezeUTXOs`, `UnFreezeUTXOs`, `ReAssignUTXO` | Executes alert-network directives (court orders / network alerts) to freeze, unfreeze, or reassign specific UTXOs. |
 | **RPC** | `services/rpc/handlers.go` | `FreezeUTXOs`, `UnFreezeUTXOs`, `ReAssignUTXO` | Exposes the same freeze/unfreeze/reassign operations as operator-facing RPC calls (e.g. for manual intervention), independent of the Alert System's automated path. |

model/filter-not-on-longest-chain.go

@@ -0,0 +1,200 @@
+package model
+
+import (
+	"context"
+	"fmt"
+	"slices"
+
+	"github.com/bsv-blockchain/go-bt/v2/chainhash"
+	"github.com/bsv-blockchain/teranode/errors"
+	"github.com/bsv-blockchain/teranode/settings"
+	"github.com/bsv-blockchain/teranode/stores/utxo"
+	"github.com/bsv-blockchain/teranode/stores/utxo/fields"
+	"github.com/bsv-blockchain/teranode/ulogger"
+	"github.com/bsv-blockchain/teranode/util"
+	"golang.org/x/sync/errgroup"
+)
+
+// filterNotOnLongestChainBatchSize is how many records the filters here read per
+// BatchDecorate call.
+const filterNotOnLongestChainBatchSize = 1024
+
+// txMetaBatchReader is the slice of the UTXO store FilterTxsNotOnLongestChain reads from.
+type txMetaBatchReader interface {
+	BatchDecorate(ctx context.Context, unresolvedMetaDataSlice []*utxo.UnresolvedMetaData, fields ...fields.FieldName) error
+}
+
+// longestChainClient is the slice of the blockchain client FilterTxsNotOnLongestChain needs.
+type longestChainClient interface {
+	CheckBlockIsInCurrentChain(ctx context.Context, blockIDs []uint32) (bool, error)
+}
+
+// recentChainHeadersClient is the slice of the blockchain client RecentChainBlockIDs needs.
+type recentChainHeadersClient interface {
+	GetBlockHeaders(ctx context.Context, blockHash *chainhash.Hash, numberOfHeaders uint64) ([]*BlockHeader, []*BlockHeaderMeta, error)
+}
+
+// RecentChainBlockIDs returns the ids of the n blocks ending at blockHash, for
+// FilterTxsNotOnLongestChain's fast path when blockHash is on the longest chain.
+func RecentChainBlockIDs(ctx context.Context, blockchainClient recentChainHeadersClient, blockHash *chainhash.Hash, n uint64) (map[uint32]bool, error) {
+	_, metas, err := blockchainClient.GetBlockHeaders(ctx, blockHash, n)
+	if err != nil {
+		return nil, errors.NewProcessingError("[RecentChainBlockIDs] error getting last block headers", err)
+	}
+
+	ids := make(map[uint32]bool, len(metas))
+	for _, meta := range metas {
+		ids[meta.ID] = true
+	}
+
+	return ids, nil
+}
+
+// FilterTxsNotOnLongestChain returns the subset of hashes that still look mined
+// (unmined_since absent) but are not mined in any block on the current longest chain:
+// exactly the txs whose MarkTransactionsOnLongestChain(false) would change something.
+//
+// It exists for blocks that left the longest chain after their txs were written as mined.
+// A tx the block shares with the longest chain is mined there and must keep its cleared
+// unmined_since; a tx that already has unmined_since needs no write at all.
+//
+//   - offChainBlockID is a block the caller has established is off the longest chain. A tx
+//     whose only block it is gets returned without a chain lookup. The caller owns that
+//     claim: if the chain can move back onto the block, re-check before marking.
+//   - recentMainChainIDs is a set of block ids on the longest chain, typically the most
+//     recent ones (RecentChainBlockIDs). A tx in any of them is on the longest chain without
+//     a lookup. Other block id sets are resolved with CheckBlockIsInCurrentChain, once per
+//     distinct set: the txs of one block share a handful of them.
+//
+// A hash missing from the UTXO store is an error, unless skipMissing is set: a caller
+// re-marking an old block's txs passes it, since a tx pruned since then has nothing to mark.
+func FilterTxsNotOnLongestChain(ctx context.Context, logger ulogger.Logger, tSettings *settings.Settings, utxoStore txMetaBatchReader,
+	blockchainClient longestChainClient, recentMainChainIDs map[uint32]bool, offChainBlockID uint32, hashes []chainhash.Hash, skipMissing bool) ([]chainhash.Hash, error) {
+	notOnLongestChain := make([]chainhash.Hash, 0, len(hashes))
+	inCurrentChain := make(map[string]bool) // CheckBlockIsInCurrentChain answers, by block id set
+
+	items, err := batchReadTxMeta(ctx, logger, tSettings, utxoStore, hashes, fields.BlockIDs, fields.UnminedSince)
+	if err != nil {
+		return nil, errors.NewProcessingError("[FilterTxsNotOnLongestChain] error getting transactions from utxo store", err)
+	}
+
+	for _, item := range items {
+		if item.Err != nil {
+			if skipMissing && errors.Is(item.Err, errors.ErrTxNotFound) {
+				continue
+			}
+
+			return nil, errors.NewProcessingError("[FilterTxsNotOnLongestChain] error getting transaction %s from utxo store", item.Hash.String(), item.Err)
+		}
+
+		txMeta := item.Data
+		if txMeta == nil {
+			if skipMissing {
+				continue
+			}
+
+			// transaction not found, not OK
+			return nil, errors.NewProcessingError("[FilterTxsNotOnLongestChain] transaction %s not found on longest chain", item.Hash.String())
+		}
+
+		if txMeta.UnminedSince != 0 {
+			// already unmined: marking it would change nothing (issue 1768 keeps an existing value)
+			continue
+		}
+
+		if len(txMeta.BlockIDs) == 1 && txMeta.BlockIDs[0] == offChainBlockID {
+			// the transaction is only in the off-chain block, so it is definitely not on the longest chain
+			notOnLongestChain = append(notOnLongestChain, item.Hash)
+			continue
+		}
+
+		foundInRecentBlock := false
+		for _, blockID := range txMeta.BlockIDs {
+			if recentMainChainIDs[blockID] {
+				// the transaction is still in one of the last 1000 blocks, so it is still on the longest chain
+				foundInRecentBlock = true
+				break
+			}
+		}
+
+		if foundInRecentBlock {
+			continue
+		}
+
+		// check BlockIDs to see if the transaction is still on the longest chain
+		key := fmt.Sprint(txMeta.BlockIDs)
+
+		onLongestChain, known := inCurrentChain[key]
+		if !known {
+			var err error
+			if onLongestChain, err = blockchainClient.CheckBlockIsInCurrentChain(ctx, txMeta.BlockIDs); err != nil {
+				return nil, errors.NewProcessingError("[FilterTxsNotOnLongestChain] error checking if transaction is on longest chain", err)
+			}
+
+			inCurrentChain[key] = onLongestChain
+		}
+
+		if !onLongestChain {
+			notOnLongestChain = append(notOnLongestChain, item.Hash)
+		}
+	}
+
+	return notOnLongestChain, nil
+}
+
+// FilterTxsUnmined returns the subset of hashes whose unmined_since is set and that are
+// mined in blockID: the txs of that block a MarkTransactionsOnLongestChain(true) would
+// change. Hashes missing from the UTXO store are skipped.
+func FilterTxsUnmined(ctx context.Context, logger ulogger.Logger, tSettings *settings.Settings, utxoStore txMetaBatchReader, blockID uint32, hashes []chainhash.Hash) ([]chainhash.Hash, error) {
+	items, err := batchReadTxMeta(ctx, logger, tSettings, utxoStore, hashes, fields.UnminedSince, fields.BlockIDs)
+	if err != nil {
+		return nil, errors.NewProcessingError("[FilterTxsUnmined] error getting transactions from utxo store", err)
+	}
+
+	unmined := make([]chainhash.Hash, 0, len(hashes))
+
+	for _, item := range items {
+		if item.Err != nil {
+			if errors.Is(item.Err, errors.ErrTxNotFound) {
+				continue
+			}
+
+			return nil, errors.NewProcessingError("[FilterTxsUnmined] error getting transaction %s from utxo store", item.Hash.String(), item.Err)
+		}
+
+		if item.Data != nil && item.Data.UnminedSince != 0 && slices.Contains(item.Data.BlockIDs, blockID) {
+			unmined = append(unmined, item.Hash)
+		}
+	}
+
+	return unmined, nil
+}
+
+// batchReadTxMeta reads the given fields of hashes in concurrent BatchDecorate batches. The
+// per-item result, including a per-item error, is on each returned item.
+func batchReadTxMeta(ctx context.Context, logger ulogger.Logger, tSettings *settings.Settings, utxoStore txMetaBatchReader,
+	hashes []chainhash.Hash, readFields ...fields.FieldName) ([]*utxo.UnresolvedMetaData, error) {
+	items := make([]*utxo.UnresolvedMetaData, len(hashes))
+	for idx, hash := range hashes {
+		items[idx] = &utxo.UnresolvedMetaData{Hash: hash, Idx: idx}
+	}
+
+	g, gCtx := errgroup.WithContext(ctx)
+	// SafeSetLimit panics loudly on a zero limit (MaxMinedRoutines and GetBatcherSize both 0)
+	// rather than letting g.Go deadlock on a zero-capacity semaphore.
+	util.SafeSetLimit(logger, g, max(tSettings.UtxoStore.MaxMinedRoutines, tSettings.UtxoStore.GetBatcherSize*2))
+
+	for start := 0; start < len(items); start += filterNotOnLongestChainBatchSize {
+		batch := items[start:min(start+filterNotOnLongestChainBatchSize, len(items))]
+
+		g.Go(func() error {
+			return utxoStore.BatchDecorate(gCtx, batch, readFields...)
+		})
+	}
+
+	if err := g.Wait(); err != nil {
+		return nil, err
+	}
+
+	return items, nil
+}

services/blockassembly/liveness_reset_test.go

@@ -87,12 +87,12 @@ type probeUtxoStore struct {
 	scanIterator  utxo.ConsistencyScanIterator
 }
 
-func (s *probeUtxoStore) MarkTransactionsOnLongestChain(ctx context.Context, txHashes []chainhash.Hash, onLongestChain bool) error {
+func (s *probeUtxoStore) MarkTransactionsOnLongestChain(ctx context.Context, txHashes []chainhash.Hash, onLongestChain bool, opts ...utxo.MarkLongestChainOption) error {
 	if !onLongestChain && s.onMarkUnmined != nil {
 		s.onMarkUnmined()
 	}
 
-	return s.Store.MarkTransactionsOnLongestChain(ctx, txHashes, onLongestChain)
+	return s.Store.MarkTransactionsOnLongestChain(ctx, txHashes, onLongestChain, opts...)
 }
 
 // ScanInconsistentUnminedTxs stands in the synthetic scan when one is set. The

services/blockassembly/reset_pruned_moveback_test.go

@@ -30,14 +30,14 @@ type markRecorderStore struct {
 	itemErrs map[chainhash.Hash]error
 }
 
-func (s *markRecorderStore) MarkTransactionsOnLongestChain(ctx context.Context, txHashes []chainhash.Hash, onLongestChain bool) error {
+func (s *markRecorderStore) MarkTransactionsOnLongestChain(ctx context.Context, txHashes []chainhash.Hash, onLongestChain bool, opts ...utxo.MarkLongestChainOption) error {
 	if !onLongestChain {
 		s.mu.Lock()
 		s.markedFalse = append(s.markedFalse, txHashes...)
 		s.mu.Unlock()
 	}
 
-	return s.Store.MarkTransactionsOnLongestChain(ctx, txHashes, onLongestChain)
+	return s.Store.MarkTransactionsOnLongestChain(ctx, txHashes, onLongestChain, opts...)
 }
 
 func (s *markRecorderStore) BatchDecorate(ctx context.Context, items []*utxo.UnresolvedMetaData, f ...fields.FieldName) error {