From 1b423f4936d40498844e7ab7ca2eeb2962393075 Mon Sep 17 00:00:00 2001 From: Damilola Edwards Date: Sun, 26 Jul 2026 15:35:10 +0100 Subject: [PATCH] Stop the permanent store from caching a block as done when it isn't processBlock logged a failure to record a block in the permanent_blocks table but still marked the block as done in its in-memory cache and returned success. Any later attempt to process that same block, such as retention re-checking before deleting the original, would hit the cache and skip straight to success without ever noticing the database never actually got the record. Retention would then delete the original block and its blob, leaving an orphaned copy at the permanent location that nothing in the system tracks anymore. The cache is now only updated once recording actually succeeds, and the failure is reported back to the caller instead of swallowed. ProcessedChan changes from a plain close signal to a channel carrying the actual result, so retention can tell the difference between "this block is safe to delete because it was archived" and "archiving failed, leave it alone and retry later." Queuing outcomes that were previously silent (store disabled, stopped, queue full) now report through the same channel so callers get one consistent way to find out what happened. --- pkg/server/service/indexer/permanent_store.go | 80 +++++-- .../indexer/permanent_store_cache_test.go | 220 ++++++++++++++++++ .../service/indexer/permanent_store_test.go | 22 +- pkg/server/service/indexer/retention.go | 19 +- 4 files changed, 302 insertions(+), 39 deletions(-) create mode 100644 pkg/server/service/indexer/permanent_store_cache_test.go diff --git a/pkg/server/service/indexer/permanent_store.go b/pkg/server/service/indexer/permanent_store.go index e6d9814..7a7fcbf 100644 --- a/pkg/server/service/indexer/permanent_store.go +++ b/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 diff --git a/pkg/server/service/indexer/permanent_store_cache_test.go b/pkg/server/service/indexer/permanent_store_cache_test.go new file mode 100644 index 0000000..d455059 --- /dev/null +++ b/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") + } +} diff --git a/pkg/server/service/indexer/permanent_store_test.go b/pkg/server/service/indexer/permanent_store_test.go index de654fa..b3f346a 100644 --- a/pkg/server/service/indexer/permanent_store_test.go +++ b/pkg/server/service/indexer/permanent_store_test.go @@ -92,7 +92,7 @@ func TestPermanentStoreQueueAndProcess(t *testing.T) { t.Run("queue and process block", func(t *testing.T) { // Queue a block for processing with a channel - processChan := make(chan struct{}) + processChan := make(chan error, 1) blockInfo := PermanentStoreBlock{ Location: blockLocation, BlockRoot: "0x1234", @@ -153,7 +153,7 @@ func TestPermanentStoreProcessSameBlockTwice(t *testing.T) { require.NoError(t, err) // Queue a block for processing with a channel - processChan1 := make(chan struct{}) + processChan1 := make(chan error, 1) blockInfo := PermanentStoreBlock{ Location: blockLocation, BlockRoot: "0x1234", @@ -191,7 +191,7 @@ func TestPermanentStoreProcessSameBlockTwice(t *testing.T) { assert.Len(t, permanentBlocks, 1, "Permanent block should be recorded in the database") // Process the same block again with a new channel - processChan2 := make(chan struct{}) + processChan2 := make(chan error, 1) blockInfo2 := PermanentStoreBlock{ Location: blockLocation, BlockRoot: "0x1234", @@ -253,7 +253,7 @@ func TestPermanentStoreDifferentNetworks(t *testing.T) { BlockRoot: "0x1234", Network: "mainnet", Slot: 123, - ProcessedChan: make(chan struct{}), + ProcessedChan: make(chan error, 1), } // Second block @@ -262,7 +262,7 @@ func TestPermanentStoreDifferentNetworks(t *testing.T) { BlockRoot: "0x1234", // Same root Network: "goerli", // Different network Slot: 123, - ProcessedChan: make(chan struct{}), + ProcessedChan: make(chan error, 1), } permanentStore.QueueBlock(blockInfo1) @@ -379,7 +379,7 @@ func TestPermanentStoreDistributedLock(t *testing.T) { Location: blockLocation, BlockRoot: "0xabcd", Network: "mainnet", - ProcessedChan: make(chan struct{}), + ProcessedChan: make(chan error, 1), } permanentStore1.QueueBlock(blockInfo1) @@ -420,7 +420,7 @@ func TestPermanentStoreDistributedLock(t *testing.T) { Location: blockLocation, BlockRoot: "0xabcd", Network: "mainnet", - ProcessedChan: make(chan struct{}), + ProcessedChan: make(chan error, 1), } permanentStore2.QueueBlock(blockInfo2) @@ -456,7 +456,7 @@ func TestPermanentStoreStop(t *testing.T) { require.NoError(t, err) // Create a channel to track when processing is complete - processChan := make(chan struct{}) + processChan := make(chan error, 1) // Queue a block for processing with the channel blockInfo := PermanentStoreBlock{ @@ -486,7 +486,7 @@ func TestPermanentStoreStop(t *testing.T) { // Now test the stop procedure with a block in the queue // Queue another block before stopping - processChan2 := make(chan struct{}) + processChan2 := make(chan error, 1) queuedBlock := PermanentStoreBlock{ Location: blockLocation, BlockRoot: "0xqueued", @@ -534,7 +534,7 @@ func TestPermanentStoreStop(t *testing.T) { Location: blockLocation, BlockRoot: "0xunprocessed", Network: "mainnet", - ProcessedChan: make(chan struct{}), + ProcessedChan: make(chan error, 1), Slot: 3, } permanentStore.QueueBlock(unprocessedBlock) @@ -583,7 +583,7 @@ func TestPermanentStoreLocation(t *testing.T) { require.NoError(t, err) // Verify we can find blocks by querying the permanent block table - processChan := make(chan struct{}) + processChan := make(chan error, 1) blockInfo.ProcessedChan = processChan // Queue the block for processing diff --git a/pkg/server/service/indexer/retention.go b/pkg/server/service/indexer/retention.go index a2ac489..b0f89f0 100644 --- a/pkg/server/service/indexer/retention.go +++ b/pkg/server/service/indexer/retention.go @@ -120,15 +120,28 @@ func (i *Indexer) purgeOldBeaconBlocks(ctx context.Context) error { Location: block.Location, BlockRoot: block.BlockRoot, Network: block.Network, - ProcessedChan: make(chan struct{}), + ProcessedChan: make(chan error, 1), //nolint:gosec // This is a valid conversion Slot: phase0.Slot(block.Slot), } i.permanentStore.QueueBlock(b) - // Wait for the block to be processed - <-b.ProcessedChan + // Wait for the permanent store to finish with this block, without + // blocking forever if the context is cancelled (for example during + // shutdown). If it could not guarantee a durable copy, leave the + // original alone rather than delete it out from under a copy that + // was never actually made -- it will be retried on the next cycle. + select { + case permErr := <-b.ProcessedChan: + if permErr != nil { + i.log.WithError(permErr).WithField("block_id", block.ID).Error("Permanent store could not process block, will retry next time") + + continue + } + case <-ctx.Done(): + return ctx.Err() + } // Delete from the store first if err := i.store.DeleteBeaconBlock(ctx, block.Location); err != nil {