From 8a7b03ffa3962a34b1bb7d8f204aa51cedbc4027 Mon Sep 17 00:00:00 2001 From: Olaoluwa Osuntokun Date: Thu, 7 May 2026 13:55:39 -0500 Subject: [PATCH] channelnotifier: add NotifyEarlyClosedChannelEvent Today NotifyClosedChannelEvent rebuilds its event by round-tripping through FetchClosedChannel, which forces the caller to have already persisted the close summary to the closed-channel bucket. The chain watcher needs to surface a CLOSED_CHANNEL event to RPC subscribers as soon as a coop close spend is first detected on chain, well before the close has reached the required confirmation depth at which the state machine would normally call MarkChannelClosed. In this commit, we add NotifyEarlyClosedChannelEvent, which dispatches a ClosedChannelEvent built from a caller-supplied summary directly through the subscribe server. The summary is expected to carry IsPending=true so subscribers can recognize that the close has not yet been finalized in the database. Two unit tests assert that the new path delivers the supplied summary verbatim and produces exactly one event per call. (cherry picked from commit fc1391adb577a3561df86825e9de61ee77d30e2a) --- channelnotifier/channelnotifier.go | 17 +++++ channelnotifier/channelnotifier_test.go | 91 +++++++++++++++++++++++++ 2 files changed, 108 insertions(+) diff --git a/channelnotifier/channelnotifier.go b/channelnotifier/channelnotifier.go index 74fa92acc..99c291071 100644 --- a/channelnotifier/channelnotifier.go +++ b/channelnotifier/channelnotifier.go @@ -191,6 +191,23 @@ func (c *ChannelNotifier) NotifyClosedChannelEvent(chanPoint wire.OutPoint) { } } +// NotifyEarlyClosedChannelEvent dispatches a ClosedChannelEvent built from the +// supplied close summary, without consulting the channel database. This is +// used by the chain watcher to insta-dispatch CLOSED_CHANNEL events to RPC +// subscribers as soon as a coop close is first detected on chain, before the +// async N-conf path has persisted the close in the database. The summary's +// IsPending field will typically be true at this point; callers should set it +// accordingly. +func (c *ChannelNotifier) NotifyEarlyClosedChannelEvent( + summary *channeldb.ChannelCloseSummary) { + + event := ClosedChannelEvent{CloseSummary: summary} + if err := c.ntfnServer.SendUpdate(event); err != nil { + log.Warnf("Unable to send early closed channel update: %v", + err) + } +} + // NotifyFullyResolvedChannelEvent notifies the channelEventNotifier goroutine // that a channel was fully resolved on chain. func (c *ChannelNotifier) NotifyFullyResolvedChannelEvent( diff --git a/channelnotifier/channelnotifier_test.go b/channelnotifier/channelnotifier_test.go index 5dbdb4a45..0f815f730 100644 --- a/channelnotifier/channelnotifier_test.go +++ b/channelnotifier/channelnotifier_test.go @@ -4,6 +4,8 @@ import ( "testing" "time" + "github.com/btcsuite/btcd/chaincfg/chainhash" + "github.com/btcsuite/btcd/wire" "github.com/lightningnetwork/lnd/channeldb" "github.com/stretchr/testify/require" ) @@ -41,3 +43,92 @@ func TestChannelUpdateEvent(t *testing.T) { t.Fatalf("expected to receive channel update event") } } + +// TestNotifyEarlyClosedChannelEvent verifies that the early-dispatch path +// delivers exactly the supplied close summary to subscribers without +// consulting the channel database. This is the path used by the chain watcher +// at first conf to insta-dispatch CLOSED_CHANNEL events for cooperative +// closes, before the close summary is persisted. +func TestNotifyEarlyClosedChannelEvent(t *testing.T) { + t.Parallel() + + // Pass nil for chanDB; the early-dispatch path must not touch it. + ntfnServer := New(nil) + require.NoError(t, ntfnServer.Start()) + t.Cleanup(func() { + require.NoError(t, ntfnServer.Stop()) + }) + + sub, err := ntfnServer.SubscribeChannelEvents() + require.NoError(t, err) + t.Cleanup(sub.Cancel) + + // Build a close summary with IsPending=true to mirror what the chain + // watcher will hand in at first-conf detection. + chanPoint := wire.OutPoint{ + Hash: chainhash.Hash{0x01, 0x02, 0x03}, + Index: 4, + } + summary := &channeldb.ChannelCloseSummary{ + ChanPoint: chanPoint, + CloseType: channeldb.CooperativeClose, + IsPending: true, + } + + ntfnServer.NotifyEarlyClosedChannelEvent(summary) + + select { + case event := <-sub.Updates(): + closedEvent, ok := event.(ClosedChannelEvent) + require.True( + t, ok, "expected ClosedChannelEvent, got %T", event, + ) + require.NotNil(t, closedEvent.CloseSummary) + require.True(t, closedEvent.CloseSummary.IsPending, + "early dispatched summary must carry IsPending=true") + require.Equal(t, summary, closedEvent.CloseSummary, + "early dispatched summary must reach subscriber "+ + "verbatim") + + case <-time.After(time.Second): + t.Fatal("expected to receive early closed channel event") + } +} + +// TestNotifyEarlyClosedChannelEventSingleEvent guards against accidental +// re-dispatch: a single early-notify call must produce exactly one event, +// not two (e.g. a fan-out bug between the early and the legacy paths). +func TestNotifyEarlyClosedChannelEventSingleEvent(t *testing.T) { + t.Parallel() + + ntfnServer := New(nil) + require.NoError(t, ntfnServer.Start()) + t.Cleanup(func() { + require.NoError(t, ntfnServer.Stop()) + }) + + sub, err := ntfnServer.SubscribeChannelEvents() + require.NoError(t, err) + t.Cleanup(sub.Cancel) + + summary := &channeldb.ChannelCloseSummary{ + ChanPoint: wire.OutPoint{Index: 7}, + CloseType: channeldb.CooperativeClose, + IsPending: true, + } + ntfnServer.NotifyEarlyClosedChannelEvent(summary) + + // Drain the single expected event. + select { + case <-sub.Updates(): + case <-time.After(time.Second): + t.Fatal("expected to receive early closed channel event") + } + + // Any further read should not produce another event. + select { + case extra := <-sub.Updates(): + t.Fatalf("unexpected second event: %T", extra) + case <-time.After(50 * time.Millisecond): + } +}