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 {