multi: add unilateral-exit subsystem

baselib/actor/durable_actor.go

@@ -6,6 +6,8 @@ import (
 	"sync"
 	"time"
 
+	"github.com/btcsuite/btclog/v2"
+	"github.com/lightninglabs/darepo-client/build"
 	"github.com/lightningnetwork/lnd/clock"
 	"github.com/lightningnetwork/lnd/fn/v2"
 )
@@ -35,6 +37,10 @@ type DurableActorConfig[M TLVMessage, R any] struct {
 	// ID is the unique identifier for the actor.
 	ID string
 
+	// Log is the logger attached to the durable actor runtime context.
+	// When unset, the runtime falls back to btclog.Disabled.
+	Log fn.Option[btclog.Logger]
+
 	// Behavior defines how the actor responds to messages.
 	// The runtime handles ack/nack automatically based on the result.
 	Behavior ActorBehavior[M, R]
@@ -101,6 +107,7 @@ func DefaultDurableActorConfig[M TLVMessage, R any](
 
 	return DurableActorConfig[M, R]{
 		ID:                id,
+		Log:               fn.None[btclog.Logger](),
 		Behavior:          behavior,
 		Store:             store,
 		Codec:             codec,
@@ -188,7 +195,10 @@ func NewDurableActor[M TLVMessage, R any](
 	cfg DurableActorConfig[M, R],
 ) *DurableActor[M, R] {
 
-	ctx, cancel := context.WithCancel(context.Background())
+	baseCtx := build.ContextWithLogger(
+		context.Background(), cfg.Log.UnwrapOr(btclog.Disabled),
+	)
+	ctx, cancel := context.WithCancel(baseCtx)
 
 	mailboxCfg := DurableMailboxConfig{
 		MailboxID:     cfg.ID,
@@ -398,9 +408,13 @@ func (a *DurableActor[M, R]) processInTransaction(
 	})
 
 	if err != nil {
-		logger(ctx).WarnS(ctx, "Transaction failed, nacking message", err,
+		logger(ctx).WarnS(ctx,
+			"Transaction failed, nacking message",
+			err,
 			"actor_id", a.id,
-			"delivery_id", delivery.ID)
+			"delivery_id", delivery.ID,
+			"msg_type", delivery.Message.MessageType(),
+		)
 
 		// Transaction failed - Nack for retry.
 		if nackErr := delivery.Nack(ctx, err, 10*time.Second); nackErr != nil {
@@ -609,6 +623,15 @@ func (a *DurableActor[M, R]) handleResultInTx(
 
 	// For Tell messages, handle based on success/error.
 	if err := result.Err(); err != nil {
+		logger(ctx).WarnS(ctx,
+			"Durable actor Tell message failed",
+			err,
+			"actor_id", a.id,
+			"delivery_id", delivery.ID,
+			"msg_type", delivery.Message.MessageType(),
+			"attempts", delivery.Attempts,
+		)
+
 		// Apply Tell retry policy.
 		retry, delay := a.tellRetryPolicy(err, delivery.Attempts)
 		if retry {
@@ -711,6 +734,15 @@ func (a *DurableActor[M, R]) handleResult(
 
 	// For Tell messages, handle based on success/error.
 	if err := result.Err(); err != nil {
+		logger(ctx).WarnS(ctx,
+			"Durable actor Tell message failed",
+			err,
+			"actor_id", a.id,
+			"delivery_id", delivery.ID,
+			"msg_type", delivery.Message.MessageType(),
+			"attempts", delivery.Attempts,
+		)
+
 		// Apply Tell retry policy.
 		retry, delay := a.tellRetryPolicy(err, delivery.Attempts)
 		if retry {

btcwbackend/chain_backend.go

@@ -281,6 +281,15 @@ func (b *ChainBackend) BroadcastTx(ctx context.Context,
 	return nil
 }
 
+// SubmitPackage is not currently supported by the neutrino backend.
+// Neutrino can broadcast individual transactions, but it does not expose
+// a package-submission path comparable to the other backends.
+func (b *ChainBackend) SubmitPackage(_ context.Context,
+	_ []*wire.MsgTx, _ *wire.MsgTx) error {
+
+	return fmt.Errorf("submit package not supported by neutrino backend")
+}
+
 // RegisterConf registers for confirmation notifications using
 // neutrino's chain notifier. The registration returns a
 // ConfRegistration with channels for receiving confirmation events.

chainbackends/lnd.go

@@ -6,6 +6,7 @@ import (
 	"fmt"
 	"log/slog"
 
+	"github.com/btcsuite/btcd/btcjson"
 	"github.com/btcsuite/btcd/btcutil"
 	"github.com/btcsuite/btcd/chaincfg/chainhash"
 	"github.com/btcsuite/btcd/wire"
@@ -27,6 +28,16 @@ type TxBroadcaster interface {
 		label string) error
 }
 
+// PackageSubmitter atomically submits parent+child transaction packages.
+// Bitcoind-backed implementations can satisfy this interface to expose v3
+// package relay through the LND chain backend.
+type PackageSubmitter interface {
+	// SubmitPackage submits the package. The maxFeeRate parameter is
+	// optional and nil leaves the node default unchanged.
+	SubmitPackage(parents []*wire.MsgTx, child *wire.MsgTx,
+		maxFeeRate *float64) (*btcjson.SubmitPackageResult, error)
+}
+
 // LNDBackend implements the chainsource.ChainBackend interface by wrapping
 // lnd's chain notification and fee estimation interfaces. This backend provides
 // full-node functionality and is suitable for production deployments where lnd
@@ -44,6 +55,9 @@ type LNDBackend struct {
 	// broadcaster provides transaction broadcasting capabilities.
 	broadcaster TxBroadcaster
 
+	// packageSubmitter optionally provides atomic package relay support.
+	packageSubmitter PackageSubmitter
+
 	// Log is an optional logger for this backend. If None, the backend
 	// falls back to extracting a logger from context.
 	Log fn.Option[btclog.Logger]
@@ -62,6 +76,13 @@ func NewLNDBackend(notifier chainntnfs.ChainNotifier,
 	}
 }
 
+// SetPackageSubmitter attaches optional package relay support to the backend.
+func (b *LNDBackend) SetPackageSubmitter(
+	packageSubmitter PackageSubmitter) {
+
+	b.packageSubmitter = packageSubmitter
+}
+
 // logger returns the configured logger, falling back to the context logger.
 func (b *LNDBackend) logger(ctx context.Context) btclog.Logger {
 	return b.Log.UnwrapOr(build.LoggerFromContext(ctx))
@@ -176,6 +197,58 @@ func (b *LNDBackend) BroadcastTx(ctx context.Context, tx *wire.MsgTx,
 	return nil
 }
 
+// SubmitPackage submits a parent+child package through the configured
+// PackageSubmitter. This is required for v3 package relay when a fee-paying
+// child must accompany otherwise non-relayable parents.
+func (b *LNDBackend) SubmitPackage(ctx context.Context,
+	parents []*wire.MsgTx, child *wire.MsgTx) error {
+
+	if b.packageSubmitter == nil {
+		return fmt.Errorf("package submission not supported by " +
+			"LND backend")
+	}
+
+	result, err := b.packageSubmitter.SubmitPackage(
+		parents, child, nil,
+	)
+	if err != nil {
+		return fmt.Errorf("submit package RPC: %w", err)
+	}
+	if result == nil {
+		return fmt.Errorf("submit package RPC returned nil result")
+	}
+	// Log per-tx results and collect errors.
+	var txErrors []error
+	for wtxid, txResult := range result.TxResults {
+		b.logger(ctx).DebugS(ctx, "Package tx result",
+			slog.String("wtxid", wtxid),
+			slog.String("txid", txResult.TxID.String()))
+
+		if txResult.Error != nil {
+			txErrors = append(txErrors, fmt.Errorf(
+				"wtxid=%s txid=%s: %s",
+				wtxid, txResult.TxID,
+				*txResult.Error,
+			))
+		}
+	}
+
+	if result.PackageMsg != "success" {
+		return fmt.Errorf("package not accepted: %s: %w",
+			result.PackageMsg, errors.Join(txErrors...))
+	}
+
+	if len(txErrors) > 0 {
+		return fmt.Errorf("package tx rejected: %w",
+			errors.Join(txErrors...))
+	}
+
+	b.logger(ctx).InfoS(ctx, "Submitted transaction package",
+		slog.Int("parent_count", len(parents)))
+
+	return nil
+}
+
 // RegisterConf registers for confirmation notifications using lnd's chain
 // notifier. The registration returns a ConfRegistration with channels for
 // receiving confirmation events.

chainbackends/lnd_test.go

@@ -5,6 +5,7 @@ import (
 	"testing"
 	"time"
 
+	"github.com/btcsuite/btcd/btcjson"
 	"github.com/btcsuite/btcd/chaincfg/chainhash"
 	"github.com/btcsuite/btcd/wire"
 	"github.com/lightninglabs/darepo-client/chainsource"
@@ -75,6 +76,18 @@ func (s *stubBroadcaster) PublishTransaction(
 	return nil
 }
 
+type stubPackageSubmitter struct {
+	result *btcjson.SubmitPackageResult
+	err    error
+}
+
+func (s *stubPackageSubmitter) SubmitPackage(parents []*wire.MsgTx,
+	child *wire.MsgTx,
+	maxFeeRate *float64) (*btcjson.SubmitPackageResult, error) {
+
+	return s.result, s.err
+}
+
 func TestRegisterConfSurvivesCallerContextCancellation(t *testing.T) {
 	t.Parallel()
 
@@ -182,6 +195,72 @@ func TestRegisterSpendSurvivesCallerContextCancellation(t *testing.T) {
 	reg.Cancel()
 }
 
+func TestSubmitPackageUnsupported(t *testing.T) {
+	t.Parallel()
+
+	backend := NewLNDBackend(
+		&stubNotifier{}, &stubFeeEstimator{}, &stubBroadcaster{},
+	)
+
+	err := backend.SubmitPackage(
+		t.Context(), []*wire.MsgTx{wire.NewMsgTx(3)},
+		wire.NewMsgTx(3),
+	)
+	require.Error(t, err)
+	require.Contains(t, err.Error(), "not supported")
+}
+
+func TestSubmitPackageSuccess(t *testing.T) {
+	t.Parallel()
+
+	backend := NewLNDBackend(
+		&stubNotifier{}, &stubFeeEstimator{}, &stubBroadcaster{},
+	)
+	backend.SetPackageSubmitter(&stubPackageSubmitter{
+		result: &btcjson.SubmitPackageResult{
+			PackageMsg: "success",
+			TxResults: map[string]btcjson.SubmitPackageTxResult{
+				"wtxid-1": {
+					TxID: chainhash.Hash{1},
+				},
+			},
+		},
+	})
+
+	err := backend.SubmitPackage(
+		t.Context(), []*wire.MsgTx{wire.NewMsgTx(3)},
+		wire.NewMsgTx(3),
+	)
+	require.NoError(t, err)
+}
+
+func TestSubmitPackageRejectsRejectedTransactions(t *testing.T) {
+	t.Parallel()
+
+	rejectReason := "insufficient fee"
+	backend := NewLNDBackend(
+		&stubNotifier{}, &stubFeeEstimator{}, &stubBroadcaster{},
+	)
+	backend.SetPackageSubmitter(&stubPackageSubmitter{
+		result: &btcjson.SubmitPackageResult{
+			PackageMsg: "success",
+			TxResults: map[string]btcjson.SubmitPackageTxResult{
+				"wtxid-1": {
+					TxID:  chainhash.Hash{1},
+					Error: &rejectReason,
+				},
+			},
+		},
+	})
+
+	err := backend.SubmitPackage(
+		t.Context(), []*wire.MsgTx{wire.NewMsgTx(3)},
+		wire.NewMsgTx(3),
+	)
+	require.Error(t, err)
+	require.Contains(t, err.Error(), "insufficient fee")
+}
+
 const (
 	pollInterval = 50 * time.Millisecond
 	testTimeout  = 5 * pollInterval

chainbackends/lndclient_adapters.go

@@ -183,14 +183,43 @@ func (n *LndClientChainNotifier) RegisterConfirmationsNtfn(
 		slog.Int("num_confs", int(numConfs)),
 		slog.Int("height_hint", int(heightHint)))
 
-	confChan, errChan, err := chainNotifier.RegisterConfirmationsNtfn(
-		ctx, txid, pkScript, int32(numConfs), int32(heightHint),
-		lndOpts...,
-	)
-	if err != nil {
+	// Run the registration in a goroutine with a timeout to
+	// prevent hanging when LND is slow under block load.
+	type regResult struct {
+		confChan chan *chainntnfs.TxConfirmation
+		errChan  chan error
+		err      error
+	}
+
+	resultCh := make(chan regResult, 1)
+	go func() {
+		cc, ec, err := chainNotifier.RegisterConfirmationsNtfn(
+			ctx, txid, pkScript, int32(numConfs),
+			int32(heightHint), lndOpts...,
+		)
+		resultCh <- regResult{cc, ec, err}
+	}()
+
+	var confChan chan *chainntnfs.TxConfirmation
+	var errChan chan error
+
+	select {
+	case r := <-resultCh:
+		if r.err != nil {
+			cancel()
+
+			return nil, fmt.Errorf(
+				"register confirmations: %w", r.err)
+		}
+
+		confChan = r.confChan
+		errChan = r.errChan
+
+	case <-time.After(15 * time.Second):
 		cancel()
 
-		return nil, fmt.Errorf("register confirmations: %w", err)
+		return nil, fmt.Errorf(
+			"register confirmations timed out after 15s")
 	}
 
 	go func() {
@@ -346,6 +375,12 @@ type LNDBackendFromLndClientConfig struct {
 	// backend falls back to extracting a logger from context or uses
 	// btclog.Disabled.
 	Log fn.Option[btclog.Logger]
+
+	// PackageSubmitter is an optional package submitter for atomic
+	// parent+child package submission. When nil, SubmitPackage
+	// returns an "unsupported" error. Typically backed by a direct
+	// bitcoind RPC client.
+	PackageSubmitter PackageSubmitter
 }
 
 // WithLogger returns a new config with the given logger set.
@@ -368,7 +403,6 @@ func NewLNDBackendFromLndClient(cfg LNDBackendFromLndClientConfig) *LNDBackend {
 
 	// Use explicit struct initialization instead of type cast for safety -
 	// this ensures we don't silently miss fields if the types diverge.
-	//nolint:gosimple
 	notifier := NewLndClientChainNotifier(LndClientChainNotifierConfig{
 		LND: cfg.LND,
 		Log: cfg.Log,
@@ -378,6 +412,7 @@ func NewLNDBackendFromLndClient(cfg LNDBackendFromLndClientConfig) *LNDBackend {
 
 	backend := NewLNDBackend(notifier, feeEstimator, broadcaster)
 	backend.Log = cfg.Log
+	backend.packageSubmitter = cfg.PackageSubmitter
 
 	return backend
 }