peer: register the rbfCloseActor, have RPC route fee bumps to it

diff --git a/peer/brontide.go b/peer/brontide.go
index 5ab4e8f..01d948b 100644
--- a/peer/brontide.go
+++ b/peer/brontide.go
@@ -1421,7 +1421,7 @@ func (p *Brontide) loadActiveChannels(chans []*channeldb.OpenChannel) (
 			// Create the Shutdown message.
 			shutdown, err := negotiateChanCloser.ShutdownChan()
 			if err != nil {
-				p.activeChanCloses.Delete(chanID)
+				p.deleteActiveChanCloser(chanID, chanPoint)
 				shutdownInfoErr = err
 
 				return
@@ -1465,7 +1465,7 @@ func (p *Brontide) loadActiveChannels(chans []*channeldb.OpenChannel) (
 		// Creating this here ensures that any shutdown messages sent
 		// will be automatically routed by the msg router.
 		if _, err := p.initRbfChanCloser(lnChan); err != nil {
-			p.activeChanCloses.Delete(chanID)
+			p.deleteActiveChanCloser(chanID, chanPoint)
 
 			return nil, fmt.Errorf("unable to init RBF chan "+
 				"closer during peer connect: %w", err)
@@ -1762,6 +1762,10 @@ func (p *Brontide) Disconnect(reason error) {
 	// Stop the onion peer actor if one was spawned.
 	p.StopOnionActorIfExists()
 
+	// Unregister any RBF close actors registered for channels of this
+	// peer so we don't leave stale entries in the actor system.
+	p.unregisterRbfCloseActors()
+
 	// Ensure that the TCP connection is properly closed before continuing.
 	p.cfg.Conn.Close()
 
@@ -1793,6 +1797,54 @@ func (p *Brontide) StopOnionActorIfExists() {
 	)
 }
 
+// unregisterRbfCloseActor removes any RBF close actor registered for the
+// given channel point from the actor system. This is idempotent and safe to
+// call whether or not an actor was registered for the channel point.
+func (p *Brontide) unregisterRbfCloseActor(chanPoint wire.OutPoint) {
+	if p.cfg.ActorSystem == nil {
+		return
+	}
+
+	actorKey := NewRbfCloserPeerServiceKey(chanPoint)
+	actorKey.UnregisterAll(p.cfg.ActorSystem)
+}
+
+// deleteActiveChanCloser removes the chan closer for the given channel ID and
+// also unregisters any RBF close actor associated with the channel point from
+// the actor system. Callers should prefer this over calling
+// activeChanCloses.Delete directly so the actor registry stays in sync with
+// the active closers map.
+func (p *Brontide) deleteActiveChanCloser(chanID lnwire.ChannelID,
+	chanPoint wire.OutPoint) {
+
+	p.activeChanCloses.Delete(chanID)
+	p.unregisterRbfCloseActor(chanPoint)
+}
+
+// unregisterRbfCloseActors removes any RBF close actors registered for this
+// peer's active channels from the actor system. This should be called on
+// disconnect so we don't leave stale RBF close actors for a peer that is no
+// longer connected. This is idempotent and safe to call multiple times.
+func (p *Brontide) unregisterRbfCloseActors() {
+	if p.cfg.ActorSystem == nil {
+		return
+	}
+
+	p.activeChannels.Range(func(_ lnwire.ChannelID,
+		channel *lnwallet.LightningChannel) bool {
+
+		// Pending channels are tracked with a nil value in the map,
+		// so skip those as they have no channel point to look up.
+		if channel == nil {
+			return true
+		}
+
+		p.unregisterRbfCloseActor(channel.ChannelPoint())
+
+		return true
+	})
+}
+
 // readNextMessage reads, and returns the next message on the wire along with
 // any additional raw payload.
 func (p *Brontide) readNextMessage() (lnwire.Message, error) {
@@ -3679,7 +3731,7 @@ func (p *Brontide) restartCoopClose(lnChan *lnwallet.LightningChannel) (
 	shutdownMsg, err := chanCloser.ShutdownChan()
 	if err != nil {
 		p.log.Errorf("unable to create shutdown message: %v", err)
-		p.activeChanCloses.Delete(chanID)
+		p.deleteActiveChanCloser(chanID, c.FundingOutpoint)
 		return nil, err
 	}
 
@@ -3784,7 +3836,7 @@ func (p *Brontide) initNegotiateChanCloser(req *htlcswitch.ChanClose,
 		// back to its normal state.
 		defer channel.ResetState()
 
-		p.activeChanCloses.Delete(chanID)
+		p.deleteActiveChanCloser(chanID, channel.ChannelPoint())
 
 		return fmt.Errorf("unable to shutdown channel: %w", err)
 	}
@@ -3955,7 +4007,9 @@ func (p *Brontide) observeRbfCloseUpdates(chanCloser *chancloser.RbfChanCloser,
 				chanID := lnwire.NewChanIDFromOutPoint(
 					*closeReq.ChanPoint,
 				)
-				p.activeChanCloses.Delete(chanID)
+				p.deleteActiveChanCloser(
+					chanID, *closeReq.ChanPoint,
+				)
 
 				return
 			}
@@ -4029,7 +4083,9 @@ func (c *chanErrorReporter) ReportError(chanErr error) {
 	}
 
 	if _, err := c.peer.initRbfChanCloser(lnChan); err != nil {
-		c.peer.activeChanCloses.Delete(c.chanID)
+		c.peer.deleteActiveChanCloser(
+			c.chanID, lnChan.ChannelPoint(),
+		)
 
 		c.peer.log.Errorf("unable to init RBF chan closer after "+
 			"error case: %v", err)
@@ -4233,8 +4289,30 @@ func (p *Brontide) initRbfChanCloser(
 			"close: %w", err)
 	}
 
+	// We store the closer first so that any lookups that race with actor
+	// registration will find the chan closer already in place.
 	p.activeChanCloses.Store(chanID, makeRbfCloser(&chanCloser))
 
+	// In addition to the message router, we'll register the state machine
+	// with the actor system.
+	if p.cfg.ActorSystem != nil {
+		p.log.Infof("Registering RBF actor for channel %v",
+			channel.ChannelPoint())
+
+		actorWrapper := newRbfCloseActor(
+			channel.ChannelPoint(), p, p.cfg.ActorSystem,
+		)
+		if err := actorWrapper.registerActor(); err != nil {
+			chanCloser.Stop()
+			p.deleteActiveChanCloser(
+				chanID, channel.ChannelPoint(),
+			)
+
+			return nil, fmt.Errorf("unable to register RBF close "+
+				"actor: %w", err)
+		}
+	}
+
 	// Now that we've created the rbf closer state machine, we'll launch a
 	// new goroutine to eventually send in the ChannelFlushed event once
 	// needed.
@@ -4666,9 +4744,12 @@ func (p *Brontide) finalizeChanClosure(chanCloser *chancloser.ChanCloser) {
 	chanPoint := chanCloser.Channel().ChannelPoint()
 	p.WipeChannel(&chanPoint)
 
-	// Also clear the activeChanCloses map of this channel.
+	// Also clear the activeChanCloses map of this channel, and unregister
+	// any RBF close actor that was registered for this channel point.
+	//
+	// TODO(roasbeef): existing race.
 	cid := lnwire.NewChanIDFromOutPoint(chanPoint)
-	p.activeChanCloses.Delete(cid) // TODO(roasbeef): existing race
+	p.deleteActiveChanCloser(cid, chanPoint)
 
 	// Next, we'll launch a goroutine which will request to be notified by
 	// the ChainNotifier once the closure transaction obtains a single
@@ -5211,7 +5292,9 @@ func (p *Brontide) handleCloseMsg(msg *closeMsg) {
 			chanCloser.CloseRequest().Err <- err
 		}
 
-		p.activeChanCloses.Delete(msg.cid)
+		p.deleteActiveChanCloser(
+			msg.cid, chanCloser.Channel().ChannelPoint(),
+		)
 
 		p.Disconnect(err)
 	}
@@ -5551,7 +5634,7 @@ func (p *Brontide) addActiveChannel(c *lnpeer.NewChannel) error {
 	// Creating this here ensures that any shutdown messages sent will be
 	// automatically routed by the msg router.
 	if _, err := p.initRbfChanCloser(lnChan); err != nil {
-		p.activeChanCloses.Delete(chanID)
+		p.deleteActiveChanCloser(chanID, lnChan.ChannelPoint())
 
 		return fmt.Errorf("unable to init RBF chan closer for new "+
 			"chan: %w", err)
@@ -5817,42 +5900,3 @@ func (p *Brontide) ChanHasRbfCoopCloser(chanPoint wire.OutPoint) bool {
 
 	return chanCloser.IsRight()
 }
-
-// TriggerCoopCloseRbfBump given a chan ID, and the params needed to trigger a
-// new RBF co-op close update, a bump is attempted. A channel used for updates,
-// along with one used to o=communicate any errors is returned. If no chan
-// closer is found, then false is returned for the second argument.
-func (p *Brontide) TriggerCoopCloseRbfBump(ctx context.Context,
-	chanPoint wire.OutPoint, feeRate chainfee.SatPerKWeight,
-	deliveryScript lnwire.DeliveryAddress) (*CoopCloseUpdates, error) {
-
-	// If RBF coop close isn't permitted, then we'll an error.
-	if !p.rbfCoopCloseAllowed() {
-		return nil, fmt.Errorf("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.
-	closeReq := htlcswitch.ChanClose{
-		CloseType:      contractcourt.CloseRegular,
-		ChanPoint:      &chanPoint,
-		TargetFeePerKw: feeRate,
-		DeliveryScript: deliveryScript,
-		Updates:        closeUpdates.UpdateChan,
-		Err:            closeUpdates.ErrChan,
-		Ctx:            ctx,
-	}
-
-	err := p.startRbfChanCloser(newRPCShutdownInit(&closeReq), chanPoint)
-	if err != nil {
-		return nil, err
-	}
-
-	return closeUpdates, nil
-}
diff --git a/rpcserver.go b/rpcserver.go
index e62de95..d4a9ce5 100644
--- a/rpcserver.go
+++ b/rpcserver.go
@@ -3018,13 +3018,27 @@ func (r *rpcServer) CloseChannel(in *lnrpc.CloseChannelRequest,
 			rpcsLog.Infof("Bypassing Switch to do fee bump "+
 				"for ChannelPoint(%v)", chanPoint)
 
-			closeUpdates, err := r.server.AttemptRBFCloseUpdate(
-				updateStream.Context(), *chanPoint, feeRate,
-				deliveryScript,
+			// To perform this RBF bump, we'll send a bump message
+			// to the RBF close actor. We propagate the stream
+			// context so that cancellation of the RPC client also
+			// tears down the observer goroutine.
+			ctx := updateStream.Context()
+			rbfBumpMsg := peer.NewRbfBumpCloseMsg(
+				ctx, *chanPoint, feeRate, deliveryScript,
+			)
+			rbfActorKey := peer.NewRbfCloserPeerServiceKey(
+				*chanPoint,
 			)
+			rbfRouter := peer.RbfChanCloserRouter(
+				r.server.actorSystem, rbfActorKey,
+			)
+
+			closeUpdates, err := rbfRouter.Ask(
+				ctx, rbfBumpMsg,
+			).Await(ctx).Unpack()
 			if err != nil {
-				return fmt.Errorf("unable to do RBF close "+
-					"update: %w", err)
+				return fmt.Errorf("unable to ask for RBF "+
+					"close: %w", err)
 			}
 
 			updateChan = closeUpdates.UpdateChan
diff --git a/server.go b/server.go
index 09cb485..83131c6 100644
--- a/server.go
+++ b/server.go
@@ -2859,6 +2859,12 @@ func (s *server) Stop() error {
 		s.sigPool.Stop()
 		s.writePool.Stop()
 		s.readPool.Stop()
+
+		// Shut down the actor system last so any in-flight actor work
+		// triggered by the subsystems above has a chance to complete.
+		if err := s.actorSystem.Shutdown(); err != nil {
+			srvrLog.Warnf("failed to stop actor system: %v", err)
+		}
 	})
 
 	return nil
@@ -5629,74 +5635,6 @@ func (s *server) ChanHasRbfCoopCloser(peerPub *btcec.PublicKey,
 	return targetPeer.ChanHasRbfCoopCloser(chanPoint)
 }
 
-// attemptCoopRbfFeeBump attempts to look up the active chan closer for a
-// channel given the outpoint. If found, we'll attempt to do a fee bump,
-// returning channels used for updates. If the channel isn't currently active
-// (p2p connection established), then his function will return an error.
-func (s *server) attemptCoopRbfFeeBump(ctx context.Context,
-	chanPoint wire.OutPoint, feeRate chainfee.SatPerKWeight,
-	deliveryScript lnwire.DeliveryAddress) (*peer.CoopCloseUpdates, error) {
-
-	// First, we'll attempt to look up the channel based on it's
-	// ChannelPoint.
-	channel, err := s.chanStateDB.FetchChannel(chanPoint)
-	if err != nil {
-		return nil, fmt.Errorf("unable to fetch channel: %w", err)
-	}
-
-	// From the channel, we can now get the pubkey of the peer, then use
-	// that to eventually get the chan closer.
-	peerPub := channel.IdentityPub.SerializeCompressed()
-
-	// Now that we have the peer pub, we can look up the peer itself.
-	s.mu.RLock()
-	targetPeer, ok := s.peersByPub[string(peerPub)]
-	s.mu.RUnlock()
-	if !ok {
-		return nil, fmt.Errorf("peer for ChannelPoint(%v) is "+
-			"not online", chanPoint)
-	}
-
-	closeUpdates, err := targetPeer.TriggerCoopCloseRbfBump(
-		ctx, chanPoint, feeRate, deliveryScript,
-	)
-	if err != nil {
-		return nil, fmt.Errorf("unable to trigger coop rbf fee bump: "+
-			"%w", err)
-	}
-
-	return closeUpdates, nil
-}
-
-// AttemptRBFCloseUpdate attempts to trigger a new RBF iteration for a co-op
-// close update. This route it to be used only if the target channel in question
-// is no longer active in the link. This can happen when we restart while we
-// already have done a single RBF co-op close iteration.
-func (s *server) AttemptRBFCloseUpdate(ctx context.Context,
-	chanPoint wire.OutPoint, feeRate chainfee.SatPerKWeight,
-	deliveryScript lnwire.DeliveryAddress) (*peer.CoopCloseUpdates, error) {
-
-	// If the channel is present in the switch, then the request should flow
-	// through the switch instead.
-	chanID := lnwire.NewChanIDFromOutPoint(chanPoint)
-	if _, err := s.htlcSwitch.GetLink(chanID); err == nil {
-		return nil, fmt.Errorf("ChannelPoint(%v) is active in link, "+
-			"invalid request", chanPoint)
-	}
-
-	// At this point, we know that the channel isn't present in the link, so
-	// we'll check to see if we have an entry in the active chan closer map.
-	updates, err := s.attemptCoopRbfFeeBump(
-		ctx, chanPoint, feeRate, deliveryScript,
-	)
-	if err != nil {
-		return nil, fmt.Errorf("unable to attempt coop rbf fee bump "+
-			"ChannelPoint(%v)", chanPoint)
-	}
-
-	return updates, nil
-}
-
 // calculateNodeAnnouncementTimestamp returns the timestamp to use for a node
 // announcement, ensuring it's at least one second after the previously
 // persisted timestamp. This ensures BOLT-07 compliance, which requires node