peer: create new rbfCloseActor to decouple RPC RBF close bumps

diff --git a/peer/rbf_close_wrapper_actor.go b/peer/rbf_close_wrapper_actor.go
new file mode 100644
index 0000000..5200f64
--- /dev/null
+++ b/peer/rbf_close_wrapper_actor.go
@@ -0,0 +1,189 @@
+package peer
+
+import (
+	"context"
+	"fmt"
+
+	"github.com/btcsuite/btcd/wire"
+	"github.com/lightningnetwork/lnd/actor"
+	"github.com/lightningnetwork/lnd/contractcourt"
+	"github.com/lightningnetwork/lnd/fn/v2"
+	"github.com/lightningnetwork/lnd/htlcswitch"
+	"github.com/lightningnetwork/lnd/lnwallet/chainfee"
+	"github.com/lightningnetwork/lnd/lnwire"
+)
+
+// rbfCloseMessage is a message type that is used to trigger a cooperative fee
+// bump, or initiate a close for the first time.
+type rbfCloseMessage struct {
+	actor.BaseMessage
+
+	// Ctx is the context of the caller that initiated the RBF close. This
+	// is propagated to the underlying close request so that cancellation
+	// of the caller (e.g. RPC stream disconnect) tears down the associated
+	// observer goroutine. The caller's context is distinct from the
+	// actor's own lifecycle context that is passed to Receive.
+	Ctx context.Context //nolint:containedctx
+
+	// ChanPoint is the channel point of the channel to be closed.
+	ChanPoint wire.OutPoint
+
+	// FeeRate is the fee rate to use for the transaction.
+	FeeRate chainfee.SatPerKWeight
+
+	// DeliveryScript is the script to use for the transaction.
+	DeliveryScript lnwire.DeliveryAddress
+}
+
+// MessageType returns the type of the message.
+//
+// NOTE: This is part of the actor.Message interface.
+func (r rbfCloseMessage) MessageType() string {
+	return fmt.Sprintf("RbfCloseMessage(%v)", r.ChanPoint)
+}
+
+// NewRbfBumpCloseMsg returns a message that can be sent to the RBF actor to
+// initiate a new fee bump.
+func NewRbfBumpCloseMsg(ctx context.Context, op wire.OutPoint,
+	feeRate chainfee.SatPerKWeight,
+	deliveryScript lnwire.DeliveryAddress) rbfCloseMessage {
+
+	return rbfCloseMessage{
+		Ctx:            ctx,
+		ChanPoint:      op,
+		FeeRate:        feeRate,
+		DeliveryScript: deliveryScript,
+	}
+}
+
+// RbfCloseActorServiceKey is a service key that can be used to reach an RBF
+// chan closer.
+type RbfCloseActorServiceKey = actor.ServiceKey[
+	rbfCloseMessage, *CoopCloseUpdates,
+]
+
+// NewRbfCloserPeerServiceKey returns a new service key that can be used to
+// reach an RBF chan closer, via an active peer.
+func NewRbfCloserPeerServiceKey(op wire.OutPoint) RbfCloseActorServiceKey {
+	opStr := op.String()
+
+	// Just using the channel point here is enough, as we have a unique
+	// type here rbfCloseMessage which will handle the final actor
+	// selection.
+	actorKey := fmt.Sprintf("Peer(RbfChanCloser(%v))", opStr)
+
+	return actor.NewServiceKey[rbfCloseMessage, *CoopCloseUpdates](actorKey)
+}
+
+// rbfCloseActor is a wrapper around the Brontide peer to expose the internal
+// RBF close state machine as an actor. This is intended for callers that need
+// to obtain streaming close updates related to the RBF close process.
+type rbfCloseActor struct {
+	chanPeer    *Brontide
+	actorSystem *actor.ActorSystem
+	chanPoint   wire.OutPoint
+}
+
+// newRbfCloseActor creates a new instance of the RBF close wrapper actor.
+func newRbfCloseActor(chanPoint wire.OutPoint,
+	chanPeer *Brontide, actorSystem *actor.ActorSystem) *rbfCloseActor {
+
+	return &rbfCloseActor{
+		chanPeer:    chanPeer,
+		actorSystem: actorSystem,
+		chanPoint:   chanPoint,
+	}
+}
+
+// registerActor registers a new RBF close actor with the actor system. If an
+// instance with the same service key and types are registered, we'll
+// unregister before proceeding.
+func (r *rbfCloseActor) registerActor() error {
+	// First, we'll make the service key of this RBF actor. This'll allow
+	// us to spawn the actor in the actor system.
+	actorKey := NewRbfCloserPeerServiceKey(r.chanPoint)
+
+	// We only want to have a single actor instance for this rbf closer,
+	// so we'll now attempt to unregister any other instances.
+	actorKey.UnregisterAll(r.actorSystem)
+
+	// Now that we know that no instances of the actor are present, let's
+	// register a new instance. We don't actually need the ref though, as
+	// any interested parties can look up the actor via the service key.
+	actorID := fmt.Sprintf(
+		"PeerWrapper(RbfChanCloser(%s))", r.chanPoint,
+	)
+	if _, err := actorKey.Spawn(r.actorSystem, actorID, r); err != nil {
+		return fmt.Errorf("unable to spawn RBF close actor for "+
+			"channel %v: %w", r.chanPoint, err)
+	}
+
+	return nil
+}
+
+// Receive implements the actor.ActorBehavior interface for the rbf closer
+// wrapper. This allows us to expose our specific processes around the coop
+// close flow as an actor.
+//
+// NOTE: This implements the actor.ActorBehavior interface.
+func (r *rbfCloseActor) Receive(_ context.Context,
+	msg rbfCloseMessage) fn.Result[*CoopCloseUpdates] {
+
+	type retType = *CoopCloseUpdates
+
+	// If RBF coop close isn't permitted, then we'll return an error.
+	if !r.chanPeer.rbfCoopCloseAllowed() {
+		return fn.Errf[retType]("rbf coop close not enabled for " +
+			"channel")
+	}
+
+	closeUpdates := &CoopCloseUpdates{
+		UpdateChan: make(chan interface{}, 1),
+		ErrChan:    make(chan error, 1),
+	}
+
+	// We'll re-use the existing switch struct here, even though we're
+	// bypassing the switch entirely. We use the caller's context from the
+	// message so that canceling the caller (e.g., RPC stream close) also
+	// tears down the observer goroutine.
+	closeReq := htlcswitch.ChanClose{
+		CloseType:      contractcourt.CloseRegular,
+		ChanPoint:      &msg.ChanPoint,
+		TargetFeePerKw: msg.FeeRate,
+		DeliveryScript: msg.DeliveryScript,
+		Updates:        closeUpdates.UpdateChan,
+		Err:            closeUpdates.ErrChan,
+		Ctx:            msg.Ctx,
+	}
+
+	err := r.chanPeer.startRbfChanCloser(
+		newRPCShutdownInit(&closeReq), msg.ChanPoint,
+	)
+	if err != nil {
+		peerLog.Errorf("unable to start RBF chan closer for "+
+			"channel %v: %v", msg.ChanPoint, err)
+
+		return fn.Errf[retType]("unable to start RBF chan "+
+			"closer: %w", err)
+	}
+
+	return fn.Ok(closeUpdates)
+}
+
+// RbfChanCloseActor is a router that will route messages to the relevant RBF
+// chan closer.
+type RbfChanCloseActor = actor.Router[rbfCloseMessage, *CoopCloseUpdates]
+
+// RbfChanCloserRouter creates a new router that will route messages to the
+// relevant RBF chan closer.
+func RbfChanCloserRouter(actorSystem *actor.ActorSystem,
+	serviceKey RbfCloseActorServiceKey) *RbfChanCloseActor {
+
+	strategy := actor.NewRoundRobinStrategy[
+		rbfCloseMessage, *CoopCloseUpdates,
+	]()
+
+	return actor.NewRouter(
+		actorSystem.Receptionist(), serviceKey, strategy, nil,
+	)
+}
diff --git a/peer/rbf_close_wrapper_actor_test.go b/peer/rbf_close_wrapper_actor_test.go
new file mode 100644
index 0000000..56fe3af
--- /dev/null
+++ b/peer/rbf_close_wrapper_actor_test.go
@@ -0,0 +1,46 @@
+package peer
+
+import (
+	"testing"
+
+	"github.com/btcsuite/btcd/wire"
+	"github.com/lightningnetwork/lnd/actor"
+	"github.com/stretchr/testify/require"
+)
+
+// TestRbfCloseActorSingleton verifies that registering an RBF close actor for
+// the same channel point twice results in only a single registered actor. The
+// second call to registerActor should unregister the first actor before
+// spawning a replacement.
+func TestRbfCloseActorSingleton(t *testing.T) {
+	t.Parallel()
+
+	actorSystem := actor.NewActorSystem()
+	t.Cleanup(func() {
+		require.NoError(t, actorSystem.Shutdown())
+	})
+
+	chanPoint := wire.OutPoint{Index: 1}
+	serviceKey := NewRbfCloserPeerServiceKey(chanPoint)
+
+	// Register the actor for the first time.
+	actor1 := newRbfCloseActor(chanPoint, nil, actorSystem)
+	require.NoError(t, actor1.registerActor())
+
+	// Verify exactly one actor is registered.
+	refs := actor.FindInReceptionist(
+		actorSystem.Receptionist(), serviceKey,
+	)
+	require.Len(t, refs, 1)
+
+	// Register the actor again for the same channel point.
+	actor2 := newRbfCloseActor(chanPoint, nil, actorSystem)
+	require.NoError(t, actor2.registerActor())
+
+	// Verify there is still exactly one actor registered (the second one
+	// replaced the first).
+	refs = actor.FindInReceptionist(
+		actorSystem.Receptionist(), serviceKey,
+	)
+	require.Len(t, refs, 1)
+}