unroll: gate admission on source-lineage canonicality (C8)

chainsource/finality.go

@@ -21,16 +21,6 @@ var finalityBlockSubscriptionBackoffs = []time.Duration{
 	2 * time.Second,
 }
 
-// finalityBlockSubscriptionAttemptTimeout bounds each individual
-// RegisterBlocks attempt. Without it a single hung RegisterBlocks call
-// (e.g. a wedged lndclient gRPC stream) would block the conf/spend
-// monitoring goroutine indefinitely — stalling Confirmed/Reorged/Done
-// delivery on that watch — since the retry schedule only bounds the gaps
-// between attempts, not the attempts themselves. 10s mirrors the per-call
-// registration timeout used in conf_actor.go's handleRegisterConf so the
-// whole file behaves consistently under a slow backend.
-const finalityBlockSubscriptionAttemptTimeout = 10 * time.Second
-
 // registerBlocksForFinality registers a block-epoch subscription used
 // to synthesize a Done signal at FinalityDepth past an observed
 // confirmation or spend. The call is retried with a short bounded
@@ -39,11 +29,22 @@ const finalityBlockSubscriptionAttemptTimeout = 10 * time.Second
 // lndclient over gRPC); a one-shot RegisterBlocks attempt that
 // briefly hiccups would leak the per-watch sub-actor indefinitely.
 //
-// The retries run in the calling sub-actor's monitoring goroutine, so
-// brief blocking here is safe: more confirmation/spend events on this
-// specific watch are not expected during the retry window (we already
-// consumed the one that triggered the arm), and ctx cancellation
-// breaks out promptly.
+// The retries run in a dedicated arming goroutine (not the sub-actor's
+// select loop), so brief blocking here is safe: more confirmation/spend
+// events on this specific watch are not expected during the retry window
+// (we already consumed the one that triggered the arm), and ctx
+// cancellation breaks out promptly.
+//
+// The passed ctx MUST be the sub-actor's long-lived context, and it is
+// handed to RegisterBlocks unwrapped: for in-process backends the
+// block-epoch forwarder goroutine is tied to the ctx it receives, so
+// bounding each attempt with a cancellable child ctx (and cancelling it
+// once the call returns) would tear the subscription down the instant it
+// was armed — starving finality synthesis of the very epochs it needs.
+// A hung RegisterBlocks can therefore stall this arming goroutine, but
+// that is contained: it is off the select loop (fix moved arming there
+// precisely so a slow backend cannot wedge Confirmed/Reorged/Done
+// delivery), and a genuinely wedged backend is a lost watch regardless.
 //
 // Returns the registration on success, or a non-nil error after
 // retries are exhausted. Callers should log the error at warn level
@@ -54,14 +55,7 @@ func registerBlocksForFinality(ctx context.Context, backend ChainBackend,
 
 	var lastErr error
 	for attempt, backoff := range finalityBlockSubscriptionBackoffs {
-		// Bound each attempt so a hung RegisterBlocks cannot wedge the
-		// monitoring goroutine; the retry schedule only bounds the gaps
-		// between attempts, not a single stuck call.
-		attemptCtx, cancel := context.WithTimeout(
-			ctx, finalityBlockSubscriptionAttemptTimeout,
-		)
-		reg, err := backend.RegisterBlocks(attemptCtx)
-		cancel()
+		reg, err := backend.RegisterBlocks(ctx)
 		if err == nil {
 			return reg, nil
 		}

oor/lineage_batch_canon_test.go

@@ -0,0 +1,165 @@
+package oor
+
+import (
+	"testing"
+	"time"
+
+	"github.com/btcsuite/btcd/chainhash/v2"
+	"github.com/btcsuite/btcd/wire/v2"
+	"github.com/btcsuite/btclog/v2"
+	"github.com/lightninglabs/darepo-client/baselib/actor"
+	"github.com/lightninglabs/darepo-client/batchcanon"
+	"github.com/lightninglabs/darepo-client/lib/tree"
+	"github.com/lightninglabs/darepo-client/vtxo"
+	fn "github.com/lightningnetwork/lnd/fn/v2"
+	"github.com/stretchr/testify/require"
+)
+
+// oorBCRef aliases the canonicality manager tell-ref to keep test literals
+// within the line limit.
+type oorBCRef = actor.TellOnlyRef[batchcanon.ManagerMsg]
+
+// lineageFragment builds an ancestry fragment anchored at the given commitment
+// txid whose tree root carries the supplied batch-output pkScript.
+func lineageFragment(txid chainhash.Hash, pkScript []byte) vtxo.Ancestry {
+	return vtxo.Ancestry{
+		CommitmentTxID: txid,
+		TreePath: &tree.Tree{
+			BatchOutput: &wire.TxOut{
+				PkScript: pkScript,
+			},
+		},
+	}
+}
+
+func lineageOutpoint(seed byte) wire.OutPoint {
+	var h chainhash.Hash
+	h[0] = seed
+
+	return wire.OutPoint{Hash: h, Index: uint32(seed)}
+}
+
+// TestRegisterLineageBatchesRegistersDistinctAncestors verifies that receiving
+// OOR VTXOs registers one RegisterBatchRequest per distinct ancestor commitment
+// tx, carrying the tree-root batch pkScript and the dependent VTXO outpoints,
+// and that a batch shared across two received VTXOs accumulates both.
+func TestRegisterLineageBatchesRegistersDistinctAncestors(t *testing.T) {
+	t.Parallel()
+
+	ref := actor.NewChannelTellOnlyRef[batchcanon.ManagerMsg](
+		"oor-batchcanon-test", 8,
+	)
+	b := &sessionBehavior{
+		cfg: SessionActorConfig{
+			BatchCanonicality: fn.Some[oorBCRef](ref),
+		},
+		log: btclog.Disabled,
+	}
+
+	txA := chainhash.Hash{0xaa}
+	txB := chainhash.Hash{0xbb}
+	scriptA := []byte{0x51, 0x20, 0xaa}
+	scriptB := []byte{0x51, 0x20, 0xbb}
+
+	vtxo1 := lineageOutpoint(1)
+	vtxo2 := lineageOutpoint(2)
+
+	descs := []*vtxo.Descriptor{
+		{
+			Outpoint: vtxo1,
+			Ancestry: []vtxo.Ancestry{
+				lineageFragment(txA, scriptA),
+			},
+		},
+		{
+			// Multi-parent: shares txA and adds txB.
+			Outpoint: vtxo2,
+			Ancestry: []vtxo.Ancestry{
+				lineageFragment(txA, scriptA),
+				lineageFragment(txB, scriptB),
+			},
+		},
+	}
+
+	b.registerLineageBatches(t.Context(), descs)
+
+	got := make(map[chainhash.Hash]*batchcanon.RegisterBatchRequest)
+	for range 2 {
+		msg, ok := ref.AwaitMessage(time.Second)
+		require.True(t, ok, "expected a RegisterBatchRequest")
+		req, ok := msg.(*batchcanon.RegisterBatchRequest)
+		require.True(t, ok)
+		got[req.BatchTxID] = req
+	}
+
+	// No third registration.
+	_, extra := ref.AwaitMessage(100 * time.Millisecond)
+	require.False(t, extra, "expected exactly two distinct batches")
+
+	require.Contains(t, got, txA)
+	require.Equal(t, scriptA, got[txA].ConfirmationPkScript)
+	require.ElementsMatch(
+		t, []wire.OutPoint{vtxo1, vtxo2}, got[txA].DependentVTXOs,
+	)
+
+	require.Contains(t, got, txB)
+	require.Equal(t, scriptB, got[txB].ConfirmationPkScript)
+	require.Equal(t, []wire.OutPoint{vtxo2}, got[txB].DependentVTXOs)
+}
+
+// TestRegisterLineageBatchesDormantWhenUnwired verifies registration is a no-op
+// when no canonicality manager ref is configured.
+func TestRegisterLineageBatchesDormantWhenUnwired(t *testing.T) {
+	t.Parallel()
+
+	b := &sessionBehavior{
+		cfg: SessionActorConfig{
+			BatchCanonicality: fn.None[oorBCRef](),
+		},
+		log: btclog.Disabled,
+	}
+
+	// Must not panic with a populated lineage and no ref.
+	b.registerLineageBatches(t.Context(), []*vtxo.Descriptor{
+		{
+			Outpoint: lineageOutpoint(1),
+			Ancestry: []vtxo.Ancestry{
+				lineageFragment(
+					chainhash.Hash{0xaa}, []byte{0x51},
+				),
+			},
+		},
+	})
+}
+
+// TestRegisterLineageBatchesSkipsIncompleteFragments verifies fragments with no
+// tree path / batch output or a zero txid are skipped (they cannot be watched),
+// leaving the gate permissive rather than registering an unwatchable batch.
+func TestRegisterLineageBatchesSkipsIncompleteFragments(t *testing.T) {
+	t.Parallel()
+
+	ref := actor.NewChannelTellOnlyRef[batchcanon.ManagerMsg](
+		"oor-batchcanon-skip", 4,
+	)
+	b := &sessionBehavior{
+		cfg: SessionActorConfig{
+			BatchCanonicality: fn.Some[oorBCRef](ref),
+		},
+		log: btclog.Disabled,
+	}
+
+	b.registerLineageBatches(t.Context(), []*vtxo.Descriptor{
+		{
+			Outpoint: lineageOutpoint(1),
+			Ancestry: []vtxo.Ancestry{
+				// Nil tree path: skipped.
+				{CommitmentTxID: chainhash.Hash{0xaa}},
+				// Zero txid: skipped.
+				lineageFragment(chainhash.Hash{}, []byte{0x51}),
+			},
+		},
+	})
+
+	_, ok := ref.AwaitMessage(200 * time.Millisecond)
+	require.False(t, ok, "no registration expected for incomplete lineage")
+}

oor/session_actor.go

@@ -9,6 +9,7 @@ import (
 
 	"github.com/btcsuite/btclog/v2"
 	"github.com/lightninglabs/darepo-client/baselib/actor"
+	"github.com/lightninglabs/darepo-client/batchcanon"
 	"github.com/lightninglabs/darepo-client/build"
 	clientdb "github.com/lightninglabs/darepo-client/db"
 	"github.com/lightninglabs/darepo-client/ledger"
@@ -64,6 +65,14 @@ type SessionActorConfig struct {
 	// materialized so it can spawn monitoring actors.
 	VTXOManager actor.TellOnlyRef[vtxo.ManagerMsg]
 
+	// BatchCanonicality, when set, receives a RegisterBatchRequest for
+	// each commitment batch in a received OOR VTXO's lineage so the
+	// reorg-safety availability gate (darepo#454) can govern the received
+	// VTXO: a reorg-out or invalidation of any ancestor batch marks the
+	// VTXO limbo. None disables registration (the gate stays dormant),
+	// matching the C5/C6 dormancy contract.
+	BatchCanonicality fn.Option[actor.TellOnlyRef[batchcanon.ManagerMsg]]
+
 	// SpendCompleter routes outgoing input-spend completion through the
 	// VTXO manager. The manager's status write commits in the VTXO actor's
 	// own transaction, so it does NOT join this actor's turn: the spend is

oor/session_actor_handlers.go

@@ -9,6 +9,7 @@ import (
 
 	"github.com/btcsuite/btcd/chainhash/v2"
 	"github.com/btcsuite/btcd/wire/v2"
+	"github.com/lightninglabs/darepo-client/batchcanon"
 	clientdb "github.com/lightninglabs/darepo-client/db"
 	"github.com/lightninglabs/darepo-client/ledger"
 	libtypes "github.com/lightninglabs/darepo-client/lib/types"
@@ -757,10 +758,111 @@ func (b *sessionBehavior) notifyMaterialized(ctx context.Context, event Event) {
 						"observer failed", err)
 				}
 			}
+
+			b.registerLineageBatches(ctx, descs)
 		}()
 	})
 }
 
+// registerLineageBatches registers every commitment batch in the received
+// VTXOs' lineage with the canonicality manager so the multi-parent admission
+// gate can govern these OOR-received VTXOs: a reorg-out or invalidation of any
+// ancestor batch marks the dependent VTXO limbo, excluding it from selection
+// until its lineage is canonical again (darepo#454, F4/F5).
+//
+// It is a no-op when no manager ref is wired (the gate stays dormant). It runs
+// on the post-commit best-effort goroutine alongside the VTXO-manager
+// notification: a dropped registration only leaves the gate permissive for
+// these VTXOs (the safe default), and the manager's RegisterBatch is
+// idempotent so a re-materialization re-registers harmlessly.
+//
+// The batch-output pkScript comes from each ancestry fragment's tree root
+// (BatchOutput), which is what script-filtering light-client backends filter
+// on, so confirmation detection of the ancestor batch works on Esplora/Neutrino
+// receivers too. ConsumedInputs are left empty: the receiver does not hold the
+// ancestor commitment txs' inputs at this seam, so per-input double-spend
+// watches are a follow-up — a reorg-out of an ancestor batch is still detected
+// via its confirmation watch and marks the VTXO limbo.
+func (b *sessionBehavior) registerLineageBatches(ctx context.Context,
+	descs []*vtxo.Descriptor) {
+
+	if b.cfg.BatchCanonicality.IsNone() {
+		return
+	}
+	ref := b.cfg.BatchCanonicality.UnsafeFromSome()
+
+	// Collect, per distinct ancestor commitment tx, its batch-output
+	// pkScript and the deduped set of received VTXO outpoints that depend
+	// on it (a desc may carry the same commitment txid across more than one
+	// ancestry fragment).
+	type batchReg struct {
+		pkScript   []byte
+		depSeen    map[wire.OutPoint]struct{}
+		dependents []wire.OutPoint
+	}
+	batches := make(map[chainhash.Hash]*batchReg)
+	order := make([]chainhash.Hash, 0)
+
+	for _, desc := range descs {
+		for i := range desc.Ancestry {
+			frag := desc.Ancestry[i]
+			if frag.TreePath == nil ||
+				frag.TreePath.BatchOutput == nil {
+
+				continue
+			}
+			txid := frag.CommitmentTxID
+			if txid == (chainhash.Hash{}) {
+				continue
+			}
+
+			reg, ok := batches[txid]
+			if !ok {
+				reg = &batchReg{
+					pkScript: frag.TreePath.
+						BatchOutput.PkScript,
+					depSeen: make(
+						map[wire.OutPoint]struct{},
+					),
+				}
+				batches[txid] = reg
+				order = append(order, txid)
+			}
+			if _, dup := reg.depSeen[desc.Outpoint]; !dup {
+				reg.depSeen[desc.Outpoint] = struct{}{}
+				reg.dependents = append(
+					reg.dependents, desc.Outpoint,
+				)
+			}
+		}
+	}
+
+	for _, txid := range order {
+		reg := batches[txid]
+
+		// CSVExpiryDelta is intentionally left zero here: the
+		// admission gate consults only the batch State (reorged /
+		// conflict), never EffectiveExpiry, and the per-batch
+		// effective expiry is not consumed for OOR-registered ancestor
+		// batches (each received VTXO carries its own BatchExpiry).
+		// Threading the real per-fragment CSV delta is a follow-up
+		// alongside ConsumedInputs.
+		err := ref.Tell(ctx, &batchcanon.RegisterBatchRequest{
+			BatchTxID:            txid,
+			ConfirmationPkScript: reg.pkScript,
+			DependentVTXOs:       reg.dependents,
+		})
+		if err != nil {
+			b.log.WarnS(ctx, "Failed to register OOR lineage "+
+				"batch canonicality", err,
+				slog.String("commitment_txid", txid.String()),
+				slog.Int("dependent_vtxos",
+					len(reg.dependents)),
+			)
+		}
+	}
+}
+
 // queueVTXOsReceived stages one VTXOReceivedMsg per materialized incoming
 // VTXO for the durable outbox enqueue in commitAck.
 func (b *sessionBehavior) queueVTXOsReceived(ctx context.Context,