fix(cache,da): drop finalized snapshot entries on restore and drain DA subscription channel

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

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)

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

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

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():