Stop the permanent store from caching a block as done when it isn't

pkg/server/service/indexer/permanent_store.go

@@ -20,7 +20,7 @@ type PermanentStoreBlock struct {
 	BlockRoot     string
 	Network       string
 	Slot          phase0.Slot
-	ProcessedChan chan struct{}
+	ProcessedChan chan error
 }
 
 // PermanentStore ensures that at least one copy of each block per network is retained
@@ -110,12 +110,21 @@ func (p *PermanentStore) IsEnabled() bool {
 
 // QueueBlock adds a block to the queue for processing.
 func (p *PermanentStore) QueueBlock(block PermanentStoreBlock) {
-	// Check if the permanent store is enabled
+	// Check if the permanent store is enabled. There was never a permanence
+	// guarantee to keep in a deployment where this feature isn't turned on,
+	// so this is reported as success rather than failure.
 	if !p.IsEnabled() {
+		sendProcessedResult(block, nil)
+
 		return
 	}
 
+	// The store is shutting down. Unlike being disabled, this is a transient
+	// state: the block wasn't archived, so callers waiting on the result
+	// should treat it as not yet safe to act on rather than as success.
 	if p.stopped {
+		sendProcessedResult(block, fmt.Errorf("permanent store is stopped"))
+
 		return
 	}
 
@@ -132,6 +141,22 @@ func (p *PermanentStore) QueueBlock(block PermanentStoreBlock) {
 			"network":    block.Network,
 			"location":   block.Location,
 		}).Warn("Failed to queue block for permanent storage, queue is full")
+
+		sendProcessedResult(block, fmt.Errorf("permanent store queue is full"))
+	}
+}
+
+// sendProcessedResult reports the outcome of attempting to process a block,
+// if the caller asked to be told (ProcessedChan is non-nil). The channel is
+// expected to be buffered by at least one slot, so this never blocks.
+func sendProcessedResult(block PermanentStoreBlock, err error) {
+	if block.ProcessedChan == nil {
+		return
+	}
+
+	select {
+	case block.ProcessedChan <- err:
+	default:
 	}
 }
 
@@ -163,16 +188,16 @@ func (p *PermanentStore) processQueue(ctx context.Context) {
 	}
 }
 
-// processBlock processes a single block.
-func (p *PermanentStore) processBlock(ctx context.Context, block PermanentStoreBlock) error {
+// processBlock processes a single block. The named return value is reported
+// back to whoever is waiting on block.ProcessedChan, so every exit path
+// (including the early returns below) must leave err set correctly rather
+// than swallowing a failure.
+func (p *PermanentStore) processBlock(ctx context.Context, block PermanentStoreBlock) (err error) {
 	// Create a cache key for this block
 	cacheKey := fmt.Sprintf("%s:%s", block.Network, block.BlockRoot)
 
-	// Close the processed channel so that the caller can wait for the block to be processed
 	defer func() {
-		if block.ProcessedChan != nil {
-			close(block.ProcessedChan)
-		}
+		sendProcessedResult(block, err)
 	}()
 
 	// Check if we've already processed this block
@@ -191,8 +216,6 @@ func (p *PermanentStore) processBlock(ctx context.Context, block PermanentStoreB
 	// Try to acquire a distributed lock with retries
 	var acquired bool
 
-	var err error
-
 	retryInterval := 200 * time.Millisecond
 	maxRetryDuration := 35 * time.Second
 	startTime := time.Now()
@@ -285,9 +308,9 @@ func (p *PermanentStore) processBlock(ctx context.Context, block PermanentStoreB
 	}
 
 	// Check if block is already recorded in database before checking the store
-	permanentBlock, err := p.db.GetPermanentBlockByBlockRoot(ctx, block.BlockRoot, block.Network)
-	if err != nil {
-		p.log.WithError(err).WithFields(logrus.Fields{
+	permanentBlock, lookupErr := p.db.GetPermanentBlockByBlockRoot(ctx, block.BlockRoot, block.Network)
+	if lookupErr != nil {
+		p.log.WithError(lookupErr).WithFields(logrus.Fields{
 			"block_root": block.BlockRoot,
 			"network":    block.Network,
 		}).Error("Failed to check if block is already recorded in database")
@@ -319,27 +342,31 @@ func (p *PermanentStore) processBlock(ctx context.Context, block PermanentStoreB
 			"location":   permanentLocation,
 		}).Debug("Block already exists in permanent location")
 
-		// Add to cache to avoid future checks
-		p.cache.Add(cacheKey, true)
-
-		// Ensure the block is recorded in the database even if it already exists in storage
-		if perr := p.recordPermanentBlock(ctx, block); perr != nil {
-			p.log.WithError(perr).WithFields(logrus.Fields{
+		// Ensure the block is recorded in the database even if it already
+		// exists in storage. Only cache success once the record actually
+		// lands: caching before this would let a later check for the same
+		// block believe it's fully durable when the database still has no
+		// row for it.
+		if err = p.recordPermanentBlock(ctx, block); err != nil {
+			p.log.WithError(err).WithFields(logrus.Fields{
 				"block_root": block.BlockRoot,
 				"network":    block.Network,
 				"slot":       block.Slot,
 			}).Error("Failed to record permanent block in database")
+
+			return err
 		}
 
+		p.cache.Add(cacheKey, true)
+
 		return nil
 	}
 
 	// Copy the block to the permanent location
-	err = p.store.Copy(ctx, &store.CopyParams{
+	if err = p.store.Copy(ctx, &store.CopyParams{
 		Source:      block.Location,
 		Destination: permanentLocation,
-	})
-	if err != nil {
+	}); err != nil {
 		return fmt.Errorf("failed to copy block to permanent location: %w", err)
 	}
 
@@ -350,13 +377,16 @@ func (p *PermanentStore) processBlock(ctx context.Context, block PermanentStoreB
 		"to":         permanentLocation,
 	}).Info("Copied block to permanent location")
 
-	// Record the block in the database
-	if perr := p.recordPermanentBlock(ctx, block); perr != nil {
-		p.log.WithError(perr).WithFields(logrus.Fields{
+	// Record the block in the database. As above, only cache success once
+	// the record actually lands.
+	if err = p.recordPermanentBlock(ctx, block); err != nil {
+		p.log.WithError(err).WithFields(logrus.Fields{
 			"block_root": block.BlockRoot,
 			"network":    block.Network,
 			"slot":       block.Slot,
 		}).Error("Failed to record permanent block in database")
+
+		return err
 	}
 
 	// Add to cache to avoid future checks

pkg/server/service/indexer/permanent_store_cache_test.go

@@ -0,0 +1,220 @@
+package indexer
+
+import (
+	"context"
+	"database/sql"
+	"fmt"
+	"os"
+	"testing"
+
+	"github.com/attestantio/go-eth2-client/spec/phase0"
+	"github.com/ethpandaops/tracoor/pkg/server/persistence"
+	"github.com/ethpandaops/tracoor/pkg/store"
+	"github.com/sirupsen/logrus"
+)
+
+// newFileBackedPermanentStore builds a real PermanentStore backed by a
+// file-backed SQLite database and an FS store, avoiding the Docker/Minio
+// dependency the rest of this package's permanent store tests require. It
+// returns the store, the underlying store.Store, and the raw database path
+// so callers can inject a real, targeted database failure.
+func newFileBackedPermanentStore(t *testing.T) (*PermanentStore, store.Store, string) {
+	t.Helper()
+
+	ctx := context.Background()
+
+	dbFile, err := os.CreateTemp("", "permanent_store_cache_*.db")
+	if err != nil {
+		t.Fatalf("failed to create temp db file: %v", err)
+	}
+	dbPath := dbFile.Name()
+	dbFile.Close()
+	os.Remove(dbPath)
+
+	t.Cleanup(func() {
+		os.Remove(dbPath)
+		os.Remove(dbPath + "-wal")
+		os.Remove(dbPath + "-shm")
+	})
+
+	db, err := persistence.NewIndexer("permanent-store-cache-test", logrus.New(), persistence.Config{
+		DSN:        fmt.Sprintf("file:%s?parseTime=True", dbPath),
+		DriverName: "sqlite",
+	}, persistence.DefaultOptions().SetMetricsEnabled(false))
+	if err != nil {
+		t.Fatalf("failed to create persistence indexer: %v", err)
+	}
+	if err := db.Start(ctx); err != nil {
+		t.Fatalf("failed to migrate: %v", err)
+	}
+
+	basePath, err := os.MkdirTemp("", "permanent_store_cache_fs")
+	if err != nil {
+		t.Fatalf("failed to create temp fs dir: %v", err)
+	}
+	t.Cleanup(func() { os.RemoveAll(basePath) })
+
+	st, err := store.NewFSStore("permanent-store-cache-test", logrus.New(), &store.FSStoreConfig{BasePath: basePath}, &store.Options{})
+	if err != nil {
+		t.Fatalf("failed to create FS store: %v", err)
+	}
+
+	permanentStore, err := NewPermanentStore(logrus.New(), st, db, "cache-test-node", &PermanentStoreConfig{
+		Blocks: BlockConfig{Enabled: true},
+	})
+	if err != nil {
+		t.Fatalf("failed to create permanent store: %v", err)
+	}
+
+	return permanentStore, st, dbPath
+}
+
+// dropPermanentBlocksTable forces every future recordPermanentBlock call to
+// fail with a real, clean SQL error, without disturbing the distributed_locks
+// table AcquireLock/ReleaseLock depend on. The connection used for the DDL
+// is opened, used, and closed immediately so it never contends with the
+// permanent store's own connection pool.
+func dropPermanentBlocksTable(t *testing.T, dbPath string) {
+	t.Helper()
+
+	conn, err := sql.Open("sqlite", dbPath)
+	if err != nil {
+		t.Fatalf("failed to open raw connection: %v", err)
+	}
+	defer conn.Close()
+
+	if _, err := conn.ExecContext(context.Background(), "DROP TABLE permanent_blocks"); err != nil {
+		t.Fatalf("failed to drop permanent_blocks table: %v", err)
+	}
+}
+
+// TestProcessBlock_DoesNotCacheWhenRecordingFails is a regression test: a
+// real (not mocked) database failure while recording a block must not be
+// treated as success. Before this fix, processBlock logged the failure but
+// still cached the block as done and returned nil, so a later retry (for
+// example retention's confirm-before-delete check) would short-circuit on
+// the cache without ever noticing the block was never actually durable.
+func TestProcessBlock_DoesNotCacheWhenRecordingFails(t *testing.T) {
+	permanentStore, st, dbPath := newFileBackedPermanentStore(t)
+	ctx := context.Background()
+
+	blockData := []byte("block bytes")
+	blockLocation := "beacon_block/cache-poison-test.ssz"
+
+	if _, err := st.SaveBeaconBlock(ctx, &store.SaveParams{Data: &blockData, Location: blockLocation}); err != nil {
+		t.Fatalf("failed to pre-upload block: %v", err)
+	}
+
+	dropPermanentBlocksTable(t, dbPath)
+
+	block := PermanentStoreBlock{
+		Location:      blockLocation,
+		BlockRoot:     "0xcachepoison",
+		Network:       "mainnet",
+		Slot:          phase0.Slot(1),
+		ProcessedChan: make(chan error, 1),
+	}
+
+	// First attempt: the copy succeeds, but recording it fails because the
+	// table is gone.
+	firstErr := permanentStore.processBlock(ctx, block)
+	if firstErr == nil {
+		t.Fatal("expected processBlock to return an error when recording fails")
+	}
+
+	select {
+	case reported := <-block.ProcessedChan:
+		if reported == nil {
+			t.Fatal("expected a non-nil error to be reported on ProcessedChan")
+		}
+	default:
+		t.Fatal("expected a result to be available on ProcessedChan immediately")
+	}
+
+	permanentLocation := permanentStore.GetPermanentLocation(block)
+
+	copied, err := st.Exists(ctx, permanentLocation)
+	if err != nil || !copied {
+		t.Fatalf("expected the blob to have been copied despite the record failure, exists=%v err=%v", copied, err)
+	}
+
+	// Second attempt, same block: if the cache had been poisoned, this
+	// would hit the cache-hit shortcut and return nil without even trying
+	// to record again. It must instead try again and fail again, since the
+	// table is still gone.
+	block2 := PermanentStoreBlock{
+		Location:      blockLocation,
+		BlockRoot:     "0xcachepoison",
+		Network:       "mainnet",
+		Slot:          phase0.Slot(1),
+		ProcessedChan: make(chan error, 1),
+	}
+
+	secondErr := permanentStore.processBlock(ctx, block2)
+	if secondErr == nil {
+		t.Fatal("expected the second processBlock call to also fail, cache was incorrectly poisoned by the first failure")
+	}
+}
+
+// TestProcessBlock_RecordsSuccessfullyAfterATransientFailureClears shows the
+// positive side of the same fix: once the underlying problem is gone, a
+// retried attempt for the same block succeeds and is cached normally.
+func TestProcessBlock_RecordsSuccessfullyAfterATransientFailureClears(t *testing.T) {
+	permanentStore, st, dbPath := newFileBackedPermanentStore(t)
+	ctx := context.Background()
+
+	blockData := []byte("block bytes")
+	blockLocation := "beacon_block/recovers.ssz"
+
+	if _, err := st.SaveBeaconBlock(ctx, &store.SaveParams{Data: &blockData, Location: blockLocation}); err != nil {
+		t.Fatalf("failed to pre-upload block: %v", err)
+	}
+
+	dropPermanentBlocksTable(t, dbPath)
+
+	block := PermanentStoreBlock{
+		Location:      blockLocation,
+		BlockRoot:     "0xrecovers",
+		Network:       "mainnet",
+		Slot:          phase0.Slot(1),
+		ProcessedChan: make(chan error, 1),
+	}
+
+	if err := permanentStore.processBlock(ctx, block); err == nil {
+		t.Fatal("expected the first attempt to fail while the table is missing")
+	}
+
+	// Recreate the table, standing in for whatever transient problem caused
+	// the original failure being resolved (a DB coming back, disk space
+	// freed up, and so on).
+	db, err := persistence.NewIndexer("permanent-store-cache-test-recover", logrus.New(), persistence.Config{
+		DSN:        fmt.Sprintf("file:%s?parseTime=True", dbPath),
+		DriverName: "sqlite",
+	}, persistence.DefaultOptions().SetMetricsEnabled(false))
+	if err != nil {
+		t.Fatalf("failed to reconnect: %v", err)
+	}
+	if err := db.Start(ctx); err != nil {
+		t.Fatalf("failed to re-migrate: %v", err)
+	}
+
+	block2 := PermanentStoreBlock{
+		Location:      blockLocation,
+		BlockRoot:     "0xrecovers",
+		Network:       "mainnet",
+		Slot:          phase0.Slot(1),
+		ProcessedChan: make(chan error, 1),
+	}
+
+	if err := permanentStore.processBlock(ctx, block2); err != nil {
+		t.Fatalf("expected the retried attempt to succeed once the table exists again, got: %v", err)
+	}
+
+	permanentBlock, err := db.GetPermanentBlockByBlockRoot(ctx, block2.BlockRoot, block2.Network)
+	if err != nil {
+		t.Fatalf("expected the block to now be recorded in the database: %v", err)
+	}
+	if permanentBlock == nil {
+		t.Fatal("expected a non-nil permanent block record")
+	}
+}