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): + } +}