diff --git a/contractcourt/chain_arbitrator.go b/contractcourt/chain_arbitrator.go index eac63cb32..1b8b00e2a 100644 --- a/contractcourt/chain_arbitrator.go +++ b/contractcourt/chain_arbitrator.go @@ -165,6 +165,15 @@ type ChainArbitratorConfig struct { // will use to notify the ChannelNotifier about a newly closed channel. NotifyClosedChannel func(wire.OutPoint) + // NotifyEarlyClosedChannel is invoked by the chain watcher when a + // cooperative close spend is first detected on chain, before the close + // summary has been persisted to the closed-channel bucket. It allows + // the channel notifier to dispatch a CLOSED_CHANNEL event over RPC at + // the same depth it did before the multi-confirmation reorg-aware + // dispatch was introduced. The follow-up persist + state advance still + // waits for the full required confirmation count. + NotifyEarlyClosedChannel func(*channeldb.ChannelCloseSummary) + // NotifyFullyResolvedChannel is a function closure that the // ChainArbitrator will use to notify the ChannelNotifier about a newly // resolved channel. The main difference to NotifyClosedChannel is that @@ -424,6 +433,18 @@ func (a *arbChannel) ForceCloseChan() (*wire.MsgTx, error) { return closeSummary.CloseTx, nil } +// shouldSuppressClosedChannelNotify reports whether MarkChannelClosed should +// skip firing NotifyClosedChannel because the chain watcher already emitted a +// preliminary CLOSED_CHANNEL via the early-dispatch path. Only the +// cooperative-close path is gated; force, breach, and abandon closes never +// take the early-dispatch path, so their NotifyClosedChannel must always +// fire from MarkChannelClosed. +func shouldSuppressClosedChannelNotify(closeType channeldb.ClosureType, + earlyDispatched bool) bool { + + return closeType == channeldb.CooperativeClose && earlyDispatched +} + // newActiveChannelArbitrator creates a new instance of an active channel // arbitrator given the state of the target channel. func newActiveChannelArbitrator(channel *channeldb.OpenChannel, @@ -451,7 +472,35 @@ func newActiveChannelArbitrator(channel *channeldb.OpenChannel, if err != nil { return err } + + // In the async multi-conf path the chain watcher + // already fires a preliminary CLOSED_CHANNEL event + // over the channel notifier as soon as the coop + // close spend lands on chain. Suppressing the + // duplicate notify here keeps the + // SubscribeChannelEvents stream emitting a single + // CLOSED_CHANNEL per close, matching the v0.20.1 + // surface. In the fast path (numConfs == 1) no early + // dispatch fires, so we still need to fire the + // CLOSED_CHANNEL event from here. Force/breach + // closes never take the early-dispatch path and so + // always notify here. + c.Lock() + w := c.activeWatchers[summary.ChanPoint] + c.Unlock() + + earlyDispatched := w != nil && + w.EarlyCoopCloseDispatched() + + if shouldSuppressClosedChannelNotify( + summary.CloseType, earlyDispatched, + ) { + + return nil + } + c.cfg.NotifyClosedChannel(summary.ChanPoint) + return nil }, IsPendingClose: false, @@ -1145,11 +1194,12 @@ func (c *ChainArbitrator) WatchNewChannel(newChan *channeldb.OpenChannel) error chanPoint, retInfo, ) }, - extractStateNumHint: lnwallet.GetStateNumHint, - auxLeafStore: c.cfg.AuxLeafStore, - auxResolver: c.cfg.AuxResolver, - auxCloser: c.cfg.AuxCloser, - chanCloseConfs: c.cfg.ChannelCloseConfs, + extractStateNumHint: lnwallet.GetStateNumHint, + auxLeafStore: c.cfg.AuxLeafStore, + auxResolver: c.cfg.AuxResolver, + auxCloser: c.cfg.AuxCloser, + chanCloseConfs: c.cfg.ChannelCloseConfs, + notifyEarlyCoopClose: c.cfg.NotifyEarlyClosedChannel, }, ) if err != nil { @@ -1317,18 +1367,20 @@ func (c *ChainArbitrator) loadOpenChannels() error { return c.cfg.ContractBreach(chanPoint, ret) } + notifyEarlyClose := c.cfg.NotifyEarlyClosedChannel chainWatcher, err := newChainWatcher( chainWatcherConfig{ - chanState: channel, - notifier: c.cfg.Notifier, - signer: c.cfg.Signer, - isOurAddr: c.cfg.IsOurAddress, - contractBreach: breachClosure, - extractStateNumHint: lnwallet.GetStateNumHint, - auxLeafStore: c.cfg.AuxLeafStore, - auxResolver: c.cfg.AuxResolver, - auxCloser: c.cfg.AuxCloser, - chanCloseConfs: c.cfg.ChannelCloseConfs, + chanState: channel, + notifier: c.cfg.Notifier, + signer: c.cfg.Signer, + isOurAddr: c.cfg.IsOurAddress, + contractBreach: breachClosure, + extractStateNumHint: lnwallet.GetStateNumHint, + auxLeafStore: c.cfg.AuxLeafStore, + auxResolver: c.cfg.AuxResolver, + auxCloser: c.cfg.AuxCloser, + chanCloseConfs: c.cfg.ChannelCloseConfs, + notifyEarlyCoopClose: notifyEarlyClose, }, ) if err != nil { diff --git a/contractcourt/chain_arbitrator_test.go b/contractcourt/chain_arbitrator_test.go index 622686f76..d859d2d79 100644 --- a/contractcourt/chain_arbitrator_test.go +++ b/contractcourt/chain_arbitrator_test.go @@ -219,3 +219,69 @@ func TestResolveContract(t *testing.T) { err = chainArb.ResolveContract(channel.FundingOutpoint) require.NoError(t, err, "second resolve call shouldn't fail") } + +// TestShouldSuppressClosedChannelNotify pins down the gate that prevents +// MarkChannelClosed from firing a duplicate NotifyClosedChannel after the +// chain watcher has already emitted a preliminary CLOSED_CHANNEL via the +// early-dispatch path. Only the cooperative-close path can be suppressed; +// every other CloseType (force, breach, abandon) must always notify here +// regardless of the early-dispatched flag. The fast path (numConfs==1) +// never sets the early-dispatched flag, so cooperative closes on that path +// also fall through to NotifyClosedChannel. +func TestShouldSuppressClosedChannelNotify(t *testing.T) { + t.Parallel() + + cases := []struct { + name string + closeType channeldb.ClosureType + earlyDispatched bool + wantSuppress bool + }{ + { + name: "coop close with early dispatch", + closeType: channeldb.CooperativeClose, + earlyDispatched: true, + wantSuppress: true, + }, + { + name: "coop close without early dispatch " + + "(fast path or no watcher)", + closeType: channeldb.CooperativeClose, + earlyDispatched: false, + wantSuppress: false, + }, + { + name: "local force close", + closeType: channeldb.LocalForceClose, + earlyDispatched: true, + wantSuppress: false, + }, + { + name: "remote force close", + closeType: channeldb.RemoteForceClose, + earlyDispatched: true, + wantSuppress: false, + }, + { + name: "breach close", + closeType: channeldb.BreachClose, + earlyDispatched: true, + wantSuppress: false, + }, + { + name: "abandoned close", + closeType: channeldb.Abandoned, + earlyDispatched: true, + wantSuppress: false, + }, + } + + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + got := shouldSuppressClosedChannelNotify( + tc.closeType, tc.earlyDispatched, + ) + require.Equal(t, tc.wantSuppress, got) + }) + } +} diff --git a/contractcourt/chain_watcher.go b/contractcourt/chain_watcher.go index e45bb3dc9..577dcc3aa 100644 --- a/contractcourt/chain_watcher.go +++ b/contractcourt/chain_watcher.go @@ -279,6 +279,16 @@ type chainWatcherConfig struct { // the normal capacity-based scaling. This is only available in // dev/integration builds for testing purposes. chanCloseConfs fn.Option[uint32] + + // notifyEarlyCoopClose, if set, is invoked with a synthesized + // ChannelCloseSummary the first time a cooperative close spend is + // detected on chain. It dispatches a CLOSED_CHANNEL event over the + // channel notifier so RPC subscribers see the close at the same + // block depth they did before the multi-confirmation reorg-aware + // dispatch was introduced. The follow-up state transition (DB persist + // + state machine advance + FULLY_RESOLVED_CHANNEL) still waits for + // the full required confirmation depth via the existing async path. + notifyEarlyCoopClose func(*channeldb.ChannelCloseSummary) } // chainWatcher is a system that's assigned to every active channel. The duty @@ -329,6 +339,26 @@ type chainWatcher struct { // ensure that the outpoint+pkscript pair is confirmed before calling // `RegisterSpendNtfn`. fundingConfirmedNtfn *chainntnfs.ConfirmationEvent + + // coopCloseEarlyDispatched is set when we have already insta-dispatched + // a preliminary CLOSED_CHANNEL event for a coop close upon first spend + // detection. It is cleared on a deep reorg of the close so a re-mined + // close still re-fires the early event. The closeObserver goroutine is + // the only writer, but the channel arbitrator's MarkChannelClosed + // callback reads it from its own goroutine to decide whether to + // suppress the duplicate notify at full conf depth, so the field is an + // atomic.Bool to make that cross-goroutine read race-free. + coopCloseEarlyDispatched atomic.Bool +} + +// EarlyCoopCloseDispatched reports whether the chain watcher already fired the +// preliminary CLOSED_CHANNEL event for the in-flight cooperative close. The +// channel arbitrator uses this to gate the duplicate CLOSED_CHANNEL that +// MarkChannelClosed would otherwise fire at full conf depth. The flag is only +// ever set on the async multi-conf path; the fast-path (numConfs == 1) leaves +// it false and so the regular MarkChannelClosed-driven notify still fires. +func (c *chainWatcher) EarlyCoopCloseDispatched() bool { + return c.coopCloseEarlyDispatched.Load() } // newChainWatcher returns a new instance of a chainWatcher for a channel given @@ -711,13 +741,21 @@ type spendProcessResult struct { // // For single-confirmation mode (numConfs == 1), it immediately dispatches the // close event and returns empty result. For multi-confirmation mode, it -// registers for confirmations and returns the new pending state. +// registers for confirmations and returns the new pending state. In the +// async path, a coop close also triggers an early CLOSED_CHANNEL event over +// the channel notifier so RPC subscribers see the close at the same depth +// they did before the multi-confirmation reorg-aware dispatch was introduced. func (c *chainWatcher) processDetectedSpend( spend *chainntnfs.SpendDetail, source string, currentPendingSpend *chainntnfs.SpendDetail, currentConfNtfn *chainntnfs.ConfirmationEvent) spendProcessResult { - // FAST PATH: Single confirmation mode dispatches immediately. + // FAST PATH: Single confirmation mode dispatches immediately. In this + // mode the existing flow already drives MarkChannelClosed at the + // single conf, which fires CLOSED_CHANNEL with a fully populated + // summary (including close initiator from the historical bucket), so + // the early dispatch is not needed and would actually deliver a + // summary with an unknown close initiator to subscribers. if c.handleSpendDispatch(spend, source) { if currentConfNtfn != nil { currentConfNtfn.Cancel() @@ -727,12 +765,22 @@ func (c *chainWatcher) processDetectedSpend( } // ASYNC PATH: Multiple confirmations (production). + // // STATE TRANSITION: None -> Pending. log.Infof("ChannelPoint(%v): detected spend from %s, "+ "transitioning to %v", c.cfg.chanState.FundingOutpoint, source, spendStatePending) - // Check for duplicate spend detection. + // Reconcile against any spend we're already tracking *before* firing + // the preliminary CLOSED_CHANNEL. If a replacement coop close arrives + // while the previous spend's NegativeConf has not yet been drained + // (e.g. a deep reorg removed the old spend and a different coop close + // then confirmed), the early-dispatch flag may still be set from the + // stale spend. Clearing it on the replacement path lets the next + // maybeDispatchEarlyCoopClose call fire a fresh event for the new tx, + // so subscribers observe the replacement instead of being left with + // the stale event (which the arbitrator's CloseType-gated suppression + // would otherwise let stand). if currentPendingSpend != nil { if *currentPendingSpend.SpenderTxHash == *spend.SpenderTxHash { log.Debugf("ChannelPoint(%v): ignoring duplicate "+ @@ -746,7 +794,10 @@ func (c *chainWatcher) processDetectedSpend( } } - // Different spend detected. Cancel existing confNtfn. + // Different spend detected (e.g. an RBF replacement). Cancel + // the existing confNtfn so we can re-register for the new tx, + // and clear the early-dispatch flag so the replacement's own + // preliminary CLOSED_CHANNEL event below can fire. log.Warnf("ChannelPoint(%v): detected different spend tx %v, "+ "replacing pending tx %v", c.cfg.chanState.FundingOutpoint, @@ -755,8 +806,18 @@ func (c *chainWatcher) processDetectedSpend( if currentConfNtfn != nil { currentConfNtfn.Cancel() } + + c.coopCloseEarlyDispatched.Store(false) } + // Fire a preliminary CLOSED_CHANNEL event over the channel notifier + // as soon as the spend is first detected so SubscribeChannelEvents + // subscribers see the close at the same depth they did before the + // multi-confirmation reorg-aware dispatch was introduced. The + // suppression of the duplicate notify at MarkChannelClosed time is + // handled in chain_arbitrator.go via a CloseType check. + c.maybeDispatchEarlyCoopClose(spend) + numConfs := c.requiredConfsForSpend() txid := spend.SpenderTxHash @@ -943,6 +1004,11 @@ func (c *chainWatcher) closeObserver() { confNtfn = nil pendingSpend = nil + // Clear the early-dispatch flag so a re-mined coop + // close re-fires the preliminary CLOSED_CHANNEL event + // with its own close summary. + c.coopCloseEarlyDispatched.Store(false) + // Reset the close confirmation height since the spend // was reorged out. err := c.cfg.chanState.ResetCloseConfirmationHeight() @@ -1347,26 +1413,78 @@ func (c *chainWatcher) requiredConfsForSpend() uint32 { }) } -// dispatchCooperativeClose processed a detect cooperative channel closure. -// We'll use the spending transaction to locate our output within the -// transaction, then clean up the database state. We'll also dispatch a -// notification to all subscribers that the channel has been closed in this -// manner. -func (c *chainWatcher) dispatchCooperativeClose(commitSpend *chainntnfs.SpendDetail) error { +// isCoopCloseSpend reports whether the supplied spending tx looks like a +// cooperative close. A coop close has a finalized input sequence number +// (either MaxTxInSequenceNum or MaxRBFSequence); regular commitment txns +// carry an obfuscated state hint in the sequence + locktime fields and +// won't match either constant. +func isCoopCloseSpend(spendingTx *wire.MsgTx) bool { + if len(spendingTx.TxIn) == 0 { + return false + } + + switch spendingTx.TxIn[0].Sequence { + case wire.MaxTxInSequenceNum: + return true + case mempool.MaxRBFSequence: + return true + } + + return false +} + +// maybeDispatchEarlyCoopClose fires a preliminary CLOSED_CHANNEL event over +// the channel notifier the first time a coop close spend is detected on +// chain. It is a no-op if no early-dispatch callback was wired in, the spend +// is not a coop close, or an early dispatch has already happened for this +// close. The flag is cleared on a deep reorg of the close (in the closeObserver +// negativeConfChan handler) so a re-mined close re-fires. +func (c *chainWatcher) maybeDispatchEarlyCoopClose( + spend *chainntnfs.SpendDetail) { + + if c.coopCloseEarlyDispatched.Load() { + return + } + if c.cfg.notifyEarlyCoopClose == nil { + return + } + + // We only insta-dispatch for coop closes. Force-close, breach, and + // abandon spends intentionally remain on the existing N-confirmation + // dispatch contract: their CLOSED_CHANNEL event is driven from the + // channel arbitrator's MarkChannelClosed callback at the required + // confirmation depth, so an early dispatch here would either deliver + // a duplicate event or, worse, surface a "cooperative close" summary + // for a unilateral spend. + if !isCoopCloseSpend(spend.SpendingTx) { + return + } + + summary := c.buildCoopCloseSummary(spend) + + log.Infof("ChannelPoint(%v): dispatching early CLOSED_CHANNEL "+ + "event for coop close tx %v at height %d", + c.cfg.chanState.FundingOutpoint, spend.SpenderTxHash, + spend.SpendingHeight) + + c.cfg.notifyEarlyCoopClose(summary) + c.coopCloseEarlyDispatched.Store(true) +} + +// buildCoopCloseSummary constructs a ChannelCloseSummary for a cooperative +// close from the supplied spend detail. The summary is returned with +// IsPending=true; the channel arbitrator's MarkChannelClosed callback flips +// this to false after the close reaches the required confirmation depth. This +// helper is shared between the early insta-dispatch path (first conf, no DB +// persist) and the post-N-conf dispatch path so both surfaces produce +// equivalent summaries. +func (c *chainWatcher) buildCoopCloseSummary( + commitSpend *chainntnfs.SpendDetail) *channeldb.ChannelCloseSummary { + broadcastTx := commitSpend.SpendingTx - - log.Infof("Cooperative closure for ChannelPoint(%v): %v", - c.cfg.chanState.FundingOutpoint, - lnutils.SpewLogClosure(broadcastTx)) - - // If the input *is* final, then we'll check to see which output is - // ours. localAmt := c.toSelfAmount(broadcastTx) - // Once this is known, we'll mark the state as fully closed in the - // database. For cooperative closes, we wait for a confirmation depth - // determined by channel capacity before dispatching this event. - closeSummary := &channeldb.ChannelCloseSummary{ + summary := &channeldb.ChannelCloseSummary{ ChanPoint: c.cfg.chanState.FundingOutpoint, ChainHash: c.cfg.chanState.ChainHash, ClosingTXID: *commitSpend.SpenderTxHash, @@ -1388,9 +1506,28 @@ func (c *chainWatcher) dispatchCooperativeClose(commitSpend *chainntnfs.SpendDet log.Errorf("ChannelPoint(%v): unable to create channel sync "+ "message: %v", c.cfg.chanState.FundingOutpoint, err) } else { - closeSummary.LastChanSyncMsg = chanSync + summary.LastChanSyncMsg = chanSync } + return summary +} + +// dispatchCooperativeClose processed a detect cooperative channel closure. +// We'll use the spending transaction to locate our output within the +// transaction, then clean up the database state. We'll also dispatch a +// notification to all subscribers that the channel has been closed in this +// manner. +func (c *chainWatcher) dispatchCooperativeClose( + commitSpend *chainntnfs.SpendDetail) error { + + broadcastTx := commitSpend.SpendingTx + + log.Infof("Cooperative closure for ChannelPoint(%v): %v", + c.cfg.chanState.FundingOutpoint, + lnutils.SpewLogClosure(broadcastTx)) + + closeSummary := c.buildCoopCloseSummary(commitSpend) + // Create a summary of all the information needed to handle the // cooperative closure. closeInfo := &CooperativeCloseInfo{ @@ -1399,7 +1536,7 @@ func (c *chainWatcher) dispatchCooperativeClose(commitSpend *chainntnfs.SpendDet // If we have an aux closer, finalize the cooperative close now that // it's confirmed. - err = fn.MapOptionZ( + err := fn.MapOptionZ( c.cfg.auxCloser, func(aux AuxChanCloser) error { return c.finalizeCoopClose(aux, broadcastTx) }, diff --git a/rpcserver.go b/rpcserver.go index 53248e891..947bc401c 100644 --- a/rpcserver.go +++ b/rpcserver.go @@ -5432,10 +5432,12 @@ func rpcChannelResolution(report *channeldb.ResolverReport) (*lnrpc.Resolution, } // getInitiators returns an initiator enum that provides information about the -// party that initiated channel's open and close. This information is obtained -// from the historical channel bucket, so unknown values are returned when the -// channel is not present (which indicates that it was closed before we started -// writing channels to the historical close bucket). +// party that initiated channel's open and close. The information is normally +// read from the historical channel bucket; for early-dispatched coop closes +// the channel is still live in the open bucket at notify time (the historical +// bucket is only populated at MarkChannelClosed time), so we fall back to the +// open channel state in that case. Unknown values are returned when neither +// bucket can provide the channel. func (r *rpcServer) getInitiators(chanPoint *wire.OutPoint) ( lnrpc.Initiator, lnrpc.Initiator, error) { @@ -5455,10 +5457,20 @@ func (r *rpcServer) getInitiators(chanPoint *wire.OutPoint) ( case err == channeldb.ErrNoHistoricalBucket: return openInitiator, closeInitiator, nil - // The channel was closed before we started storing historical - // channels. Do not return an error, initiator values are unknown. + // The channel was either closed before we started storing + // historical channels OR the historical bucket has not been + // populated yet because this is an early-dispatched + // CLOSED_CHANNEL event for a coop close that hasn't reached its + // full confirmation depth. Try the open channel bucket so the + // early dispatch still carries close-initiator info. case err == channeldb.ErrChannelNotFound: - return openInitiator, closeInitiator, nil + openChan, openErr := r.server.chanStateDB.FetchChannel( + *chanPoint, + ) + if openErr != nil { + return openInitiator, closeInitiator, nil + } + histChan = openChan case err != nil: return 0, 0, err diff --git a/server.go b/server.go index 09cb4858b..ddbc81e28 100644 --- a/server.go +++ b/server.go @@ -1404,6 +1404,7 @@ func newServer(ctx context.Context, cfg *Config, listenAddrs []net.Addr, Sweeper: s.sweeper, Registry: s.invoices, NotifyClosedChannel: s.channelNotifier.NotifyClosedChannelEvent, + NotifyEarlyClosedChannel: s.channelNotifier.NotifyEarlyClosedChannelEvent, NotifyFullyResolvedChannel: s.channelNotifier.NotifyFullyResolvedChannelEvent, OnionProcessor: s.sphinxPayment, PaymentsExpirationGracePeriod: cfg.PaymentsExpirationGracePeriod,