diff --git a/peer/brontide.go b/peer/brontide.go index 5ab4e8fa6..01d948beb 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 ca33b0ed5..45579da6c 100644 --- a/rpcserver.go +++ b/rpcserver.go @@ -3000,13 +3000,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 0bea9471f..0c864d988 100644 --- a/server.go +++ b/server.go @@ -2915,6 +2915,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 @@ -5687,74 +5693,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