From f5af7bcb495c470cedc97f165466e0aa6b12850a Mon Sep 17 00:00:00 2001 From: tac0turtle Date: Wed, 15 Jul 2026 15:24:20 +0200 Subject: [PATCH] fix(cache,da): drop finalized snapshot entries on restore and drain DA subscription channel MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Two memory fixes for long-running nodes: Cache restore: the DA-inclusion snapshot is only written on graceful shutdown, so after a crash (e.g. OOM kill) the restored snapshot can contain heights already below the persisted DA-included watermark. The inclusion loop never evicts below its watermark, so those placeholder entries leaked for the process lifetime and were re-persisted on every subsequent save, growing the snapshot monotonically across crash/restart cycles. RestoreFromStore now skips entries at or below the persisted DAIncludedHeight; skipped entries still seed maxDAHeight so DaHeight() is unchanged. DA subscription: the Subscribe wrapper goroutine exited on ctx cancellation without draining the underlying jsonrpc channel, leaving the go-jsonrpc delivery goroutine blocked on send โ€” it never observed the cancellation and never closed its channel, leaking one goroutine per watchdog reconnect. The wrapper now drains the raw channel on exit. Co-Authored-By: Claude Fable 5 --- block/internal/cache/generic_cache.go | 14 +++++- block/internal/cache/generic_cache_test.go | 54 ++++++++++++++++++---- block/internal/cache/manager.go | 13 +++++- block/internal/cache/manager_test.go | 8 +++- block/internal/da/client.go | 11 ++++- 5 files changed, 84 insertions(+), 16 deletions(-) diff --git a/block/internal/cache/generic_cache.go b/block/internal/cache/generic_cache.go index 0ef068c51a..543dfa4c48 100644 --- a/block/internal/cache/generic_cache.go +++ b/block/internal/cache/generic_cache.go @@ -191,7 +191,14 @@ func decodeSnapshot(buf []byte) []snapshotEntry { // RestoreFromStore loads the in-flight snapshot with a single store read. // Each entry is installed as a height placeholder; real hashes replace them // once the DA retriever re-fires SetHeaderDAIncluded after startup. -func (c *Cache) RestoreFromStore(ctx context.Context) error { +// +// Entries at or below finalizedHeight are already DA-included and finalized: +// the snapshot on disk can predate the current DA-included height (it is only +// written on graceful shutdown), and the inclusion loop never revisits heights +// below its watermark, so installing them would leak placeholders for the +// lifetime of the process and re-persist them on the next save. They are +// skipped, contributing only to maxDAHeight. +func (c *Cache) RestoreFromStore(ctx context.Context, finalizedHeight uint64) error { if c.store == nil || c.storeKeyPrefix == "" { return nil } @@ -208,10 +215,13 @@ func (c *Cache) RestoreFromStore(ctx context.Context) error { defer c.mu.Unlock() for _, e := range decodeSnapshot(buf) { + c.setMaxDAHeight(e.daHeight) + if e.blockHeight <= finalizedHeight { + continue + } placeholder := HeightPlaceholderKey(c.storeKeyPrefix, e.blockHeight) c.daIncluded[placeholder] = e.daHeight c.hashByHeight[e.blockHeight] = placeholder - c.setMaxDAHeight(e.daHeight) } return nil diff --git a/block/internal/cache/generic_cache_test.go b/block/internal/cache/generic_cache_test.go index 4af79f7b33..da7048c0ff 100644 --- a/block/internal/cache/generic_cache_test.go +++ b/block/internal/cache/generic_cache_test.go @@ -57,7 +57,7 @@ func TestCache_RestoreFromStore_EmptyChain(t *testing.T) { st := testMemStore(t) c := NewCache(st, "hdr/") - require.NoError(t, c.RestoreFromStore(context.Background())) + require.NoError(t, c.RestoreFromStore(context.Background(), 0)) assert.Equal(t, 0, c.daIncludedLen(), "no entries expected on empty chain") assert.Equal(t, uint64(0), c.daHeight()) @@ -75,7 +75,7 @@ func TestCache_RestoreFromStore_FullyFinalized(t *testing.T) { writeSnapshot(t, st, "hdr/", nil) c := NewCache(st, "hdr/") - require.NoError(t, c.RestoreFromStore(ctx)) + require.NoError(t, c.RestoreFromStore(ctx, 0)) assert.Equal(t, 0, c.daIncludedLen(), "no in-flight entries expected") assert.Equal(t, uint64(0), c.daHeight(), "no in-flight entries means daHeight is 0") @@ -93,7 +93,7 @@ func TestCache_RestoreFromStore_InFlightWindow(t *testing.T) { }) c := NewCache(st, "hdr/") - require.NoError(t, c.RestoreFromStore(ctx)) + require.NoError(t, c.RestoreFromStore(ctx, 0)) assert.Equal(t, 2, c.daIncludedLen(), "exactly the in-flight snapshot entries should be loaded") assert.Equal(t, uint64(14), c.daHeight(), "maxDAHeight should reflect the highest in-flight DA height") @@ -118,7 +118,7 @@ func TestCache_RestoreFromStore_SingleEntry(t *testing.T) { }) c := NewCache(st, "hdr/") - require.NoError(t, c.RestoreFromStore(ctx)) + require.NoError(t, c.RestoreFromStore(ctx, 0)) assert.Equal(t, 1, c.daIncludedLen(), "one entry should be in-flight") assert.Equal(t, uint64(20), c.daHeight()) @@ -131,7 +131,7 @@ func TestCache_RestoreFromStore_SingleEntry(t *testing.T) { func TestCache_RestoreFromStore_NilStore(t *testing.T) { c := NewCache(nil, "") - require.NoError(t, c.RestoreFromStore(context.Background())) + require.NoError(t, c.RestoreFromStore(context.Background(), 0)) assert.Equal(t, 0, c.daIncludedLen()) } @@ -147,7 +147,7 @@ func TestCache_RestoreFromStore_PlaceholderOverwrittenByRealHash(t *testing.T) { }) c := NewCache(st, "hdr/") - require.NoError(t, c.RestoreFromStore(ctx)) + require.NoError(t, c.RestoreFromStore(ctx, 0)) assert.Equal(t, 1, c.daIncludedLen(), "one placeholder for height 3") @@ -176,7 +176,7 @@ func TestCache_RestoreFromStore_RoundTrip(t *testing.T) { require.NoError(t, c1.SaveToStore(ctx)) c2 := NewCache(st, "rt/") - require.NoError(t, c2.RestoreFromStore(ctx)) + require.NoError(t, c2.RestoreFromStore(ctx, 0)) assert.Equal(t, 2, c2.daIncludedLen(), "only non-deleted entries should be restored") assert.Equal(t, uint64(30), c2.daHeight()) @@ -189,6 +189,42 @@ func TestCache_RestoreFromStore_RoundTrip(t *testing.T) { assert.True(t, ok, "height 3 placeholder should exist") } +// TestCache_RestoreFromStore_SkipsFinalizedEntries verifies that snapshot +// entries at or below the finalized height are not installed as placeholders +// (a stale snapshot from a previous graceful shutdown can contain heights the +// inclusion loop has already advanced past), while maxDAHeight still accounts +// for them. +func TestCache_RestoreFromStore_SkipsFinalizedEntries(t *testing.T) { + st := testMemStore(t) + ctx := context.Background() + + writeSnapshot(t, st, "hdr/", []snapshotEntry{ + {blockHeight: 3, daHeight: 10}, + {blockHeight: 4, daHeight: 11}, + {blockHeight: 5, daHeight: 12}, + }) + + c := NewCache(st, "hdr/") + require.NoError(t, c.RestoreFromStore(ctx, 4)) + + assert.Equal(t, 1, c.daIncludedLen(), "only entries above the finalized height should be installed") + assert.Equal(t, uint64(12), c.daHeight(), "maxDAHeight should include skipped entries") + + _, ok := c.getDAIncludedByHeight(3) + assert.False(t, ok, "height 3 is finalized, must not be restored") + _, ok = c.getDAIncludedByHeight(4) + assert.False(t, ok, "height 4 is finalized, must not be restored") + daH, ok := c.getDAIncludedByHeight(5) + require.True(t, ok, "height 5 is in-flight, must be restored") + assert.Equal(t, uint64(12), daH) + + // A fully-stale snapshot restores nothing but still seeds maxDAHeight. + c2 := NewCache(st, "hdr/") + require.NoError(t, c2.RestoreFromStore(ctx, 10)) + assert.Equal(t, 0, c2.daIncludedLen(), "all entries finalized, nothing restored") + assert.Equal(t, uint64(12), c2.daHeight()) +} + // --------------------------------------------------------------------------- // Basic operations // --------------------------------------------------------------------------- @@ -306,7 +342,7 @@ func TestCache_NoPlaceholderLeakAfterRefire(t *testing.T) { require.NoError(t, c1.SaveToStore(ctx)) c2 := NewCache(st, "pfx/") - require.NoError(t, c2.RestoreFromStore(ctx)) + require.NoError(t, c2.RestoreFromStore(ctx, 0)) placeholder := HeightPlaceholderKey("pfx/", 3) _, placeholderPresent := c2.getDAIncluded(placeholder) @@ -344,7 +380,7 @@ func TestCache_RestartIdempotent(t *testing.T) { for restart := 1; restart <= 3; restart++ { cR := NewCache(st, "pfx/") - require.NoError(t, cR.RestoreFromStore(ctx), "restart %d: RestoreFromStore", restart) + require.NoError(t, cR.RestoreFromStore(ctx, 0), "restart %d: RestoreFromStore", restart) assert.Equal(t, 1, cR.daIncludedLen(), "restart %d: one placeholder entry", restart) assert.Equal(t, daH, cR.daHeight(), "restart %d: daHeight correct", restart) diff --git a/block/internal/cache/manager.go b/block/internal/cache/manager.go index cbf1b8934d..1581c66f9d 100644 --- a/block/internal/cache/manager.go +++ b/block/internal/cache/manager.go @@ -294,13 +294,22 @@ func (m *implementation) SaveToStore() error { // RestoreFromStore loads the in-flight snapshot (O(1)) then seeds maxDAHeight // from the finalized-tip HeightToDAHeight metadata so DaHeight() is correct // even when the snapshot is empty (all blocks finalized). +// Snapshot entries at or below the persisted DA-included height are dropped: +// the snapshot may predate the watermark (it is only written on graceful +// shutdown) and the inclusion loop never evicts below it. func (m *implementation) RestoreFromStore() error { ctx := context.Background() - if err := m.headerCache.RestoreFromStore(ctx); err != nil { + daIncludedHeight, _, err := getMetadataUint64(ctx, m.store, store.DAIncludedHeightKey) + if err != nil { + m.logger.Warn().Err(err).Msg("failed to read DA included height, restoring cache unfiltered") + daIncludedHeight = 0 + } + + if err := m.headerCache.RestoreFromStore(ctx, daIncludedHeight); err != nil { return fmt.Errorf("failed to restore header cache from store: %w", err) } - if err := m.dataCache.RestoreFromStore(ctx); err != nil { + if err := m.dataCache.RestoreFromStore(ctx, daIncludedHeight); err != nil { return fmt.Errorf("failed to restore data cache from store: %w", err) } diff --git a/block/internal/cache/manager_test.go b/block/internal/cache/manager_test.go index b94ab81c4e..e220f1e503 100644 --- a/block/internal/cache/manager_test.go +++ b/block/internal/cache/manager_test.go @@ -136,7 +136,13 @@ func TestManager_SaveAndRestoreFromStore(t *testing.T) { // Height 1 is finalized (DAIncludedHeight = 1): IsHeightDAIncluded returns true // via the height comparison, so GetHeaderDAIncludedByHash is never consulted for it. - // The cache entry is not restored โ€” this is correct and intentional. + // The cache entry is not restored โ€” this is correct and intentional. Restoring it + // would leak a placeholder for the process lifetime (the inclusion loop never + // evicts below its watermark) and re-persist it on every subsequent save. + _, ok := m2.GetHeaderDAIncludedByHeight(1) + assert.False(t, ok, "finalized height 1 must not be restored from the snapshot") + _, ok = m2.GetDataDAIncludedByHeight(1) + assert.False(t, ok, "finalized height 1 must not be restored from the snapshot") // Height 2 is in-flight: the window restore loads a placeholder entry keyed by // height. The real content-hash entry is populated when the submitter re-processes diff --git a/block/internal/da/client.go b/block/internal/da/client.go index 0ffd7e41ed..5fad44feb8 100644 --- a/block/internal/da/client.go +++ b/block/internal/da/client.go @@ -497,8 +497,8 @@ func (c *client) SupportsSubscribe() bool { // Subscribe subscribes to blobs in the given namespace via the celestia-node // Subscribe API. It returns a channel that emits a SubscriptionEvent for every // DA block containing a matching blob. The channel is closed when ctx is -// cancelled. The caller must drain the channel after cancellation to avoid -// goroutine leaks. +// cancelled; the underlying jsonrpc subscription channel is drained on exit so +// its delivery goroutine is never left blocked on send. // Timestamps are included from the header if available (celestia-node v0.29.1+ยง), otherwise // fetched via a separate call when includeTimestamp is true. Be aware that fetching timestamps // separately is an additional call to the celestia node for each event. @@ -516,6 +516,13 @@ func (c *client) Subscribe(ctx context.Context, namespace []byte, includeTimesta out := make(chan datypes.SubscriptionEvent, 16) go func() { defer close(out) + // The jsonrpc layer closes rawCh only after its pending send is + // consumed and the ctx cancellation is processed. Drain it on exit + // so the delivery goroutine is not leaked on reconnect. + defer func() { + for range rawCh { //nolint:revive // draining until producer closes the channel + } + }() for { select { case <-ctx.Done():