From a5f9ec0c0eec57bf5493640029c7378ef954e908 Mon Sep 17 00:00:00 2001 From: Andrew Davis <1709934+Savid@users.noreply.github.com> Date: Tue, 28 Jul 2026 14:11:06 +1000 Subject: [PATCH 1/3] fix(agent): bound concurrent beacon state and bad block fetches Beacon states and execution bad blocks are both read fully into memory before being compressed and stored. In `single` mode one process runs an agent per node and every agent reacts to the same block event, so nothing limited how many of those reads ran at once - peak memory scaled with the node count rather than with any configured bound. On a glamsterdam devnet with ~100MB states, 47 agents sawtoothed between 2.3GB and 5.67GB against a 6GB limit and were OOM killed every 35-64s. Trimming the agent list only moved the deadline (20 agents lasted ~101s), since the peak is a function of concurrent fetches. Add a process-wide semaphore for each path. The slot is held across the fetch, the compression and the upload, because both the raw and compressed copies are live for that whole window; for bad blocks it is held across the indexing pass, since the decoded slice is retained until it completes. Also drop the raw state reference once compressed so only one copy survives the upload. Measured with 47 agents against a mock beacon node serving 100MB states, 6GB / 2 CPU, same config otherwise: before peak 5.67GB (94% of limit), 2.3-5.67GB sawtooth limit 4 peak 1.09GB (18%), 679 states indexed limit 10 peak 1.44GB (24%), 752 states indexed Bounding concurrency also made state archival work: previously the fetches contended badly enough that a production deployment persisted 20 states in 10 days. The limits are process-wide and fixed by whichever agent starts first, as agents are constructed independently. Threading a limiter through the single config would avoid that at the cost of changing agent.New. Co-Authored-By: Claude Opus 5 (1M context) --- pkg/agent/consensus.go | 14 +++++++ pkg/agent/ethereum/config.go | 33 +++++++++++++++ pkg/agent/execution.go | 10 +++++ pkg/agent/limiter.go | 81 ++++++++++++++++++++++++++++++++++++ 4 files changed, 138 insertions(+) create mode 100644 pkg/agent/limiter.go diff --git a/pkg/agent/consensus.go b/pkg/agent/consensus.go index d706246..396f667 100644 --- a/pkg/agent/consensus.go +++ b/pkg/agent/consensus.go @@ -77,6 +77,16 @@ func (s *agent) fetchAndIndexBeaconState(ctx context.Context, slot phase0.Slot) stateID = fmt.Sprintf("%d", slot) } + // Bound how many states are in flight process-wide. The raw state and its + // compressed copy are both held until the upload finishes, so the slot is + // kept for that whole window rather than just the fetch. + releaseFetchSlot, err := s.acquireBeaconStateFetch(ctx) + if err != nil { + return errors.Wrap(err, "failed to acquire beacon state fetch slot") + } + + defer releaseFetchSlot() + // Fetch the state state, err := s.node.Beacon().Node().FetchRawBeaconState(ctx, stateID, string(mime.ContentTypeOctet)) if err != nil { @@ -91,6 +101,10 @@ func (s *agent) fetchAndIndexBeaconState(ctx context.Context, slot phase0.Slot) return errors.Wrap(err, "failed to compress beacon state") } + // Drop the raw state before the upload so only the compressed copy is held + // for the duration of the store write. + state = nil + // Upload the state to the store location, err = s.store.SaveBeaconState(ctx, &store.SaveParams{ Data: &compressedState, diff --git a/pkg/agent/ethereum/config.go b/pkg/agent/ethereum/config.go index 2aa99a4..8409235 100644 --- a/pkg/agent/ethereum/config.go +++ b/pkg/agent/ethereum/config.go @@ -27,6 +27,39 @@ type Config struct { // which can cause the agent to get stuck as old states might not be available in // the beacon node cache. BeaconStateAgeThresholdEpochs uint64 `yaml:"beaconStateAgeThresholdEpochs" default:"1"` + + // MaxConcurrentBeaconStateFetches bounds how many beacon states may be fetched + // at once across every agent in the process. Each fetch holds the raw state and + // its compressed copy in memory, so this caps peak memory independently of how + // many agents are configured. Applied process-wide by whichever agent starts + // first, so it is only meaningful when set consistently across agents. + MaxConcurrentBeaconStateFetches int `yaml:"maxConcurrentBeaconStateFetches" default:"10"` + + // MaxConcurrentExecutionBadBlockFetches bounds how many bad block fetches may + // be in flight at once across every agent in the process. The response carries + // every bad block the node still holds and is decoded into memory for the whole + // indexing pass. Applied process-wide, as above. + MaxConcurrentExecutionBadBlockFetches int `yaml:"maxConcurrentExecutionBadBlockFetches" default:"4"` +} + +// GetMaxConcurrentBeaconStateFetches returns the configured beacon state fetch +// concurrency, or 0 to use the agent default. +func (c *Config) GetMaxConcurrentBeaconStateFetches() int { + if c.MaxConcurrentBeaconStateFetches <= 0 { + return 0 + } + + return c.MaxConcurrentBeaconStateFetches +} + +// GetMaxConcurrentExecutionBadBlockFetches returns the configured bad block fetch +// concurrency, or 0 to use the agent default. +func (c *Config) GetMaxConcurrentExecutionBadBlockFetches() int { + if c.MaxConcurrentExecutionBadBlockFetches <= 0 { + return 0 + } + + return c.MaxConcurrentExecutionBadBlockFetches } func (c *Config) Validate() error { diff --git a/pkg/agent/execution.go b/pkg/agent/execution.go index 0ac7db7..5b83dd5 100644 --- a/pkg/agent/execution.go +++ b/pkg/agent/execution.go @@ -97,6 +97,16 @@ func (s *agent) fetchAndIndexExecutionBlockTrace(ctx context.Context, blockNumbe } func (s *agent) fetchAndIndexExecutionBadBlocks(ctx context.Context) error { + // Bound how many bad block responses are decoded at once process-wide. The + // decoded slice is retained for the whole indexing pass below, so the slot is + // held until indexing finishes rather than released after the fetch. + releaseFetchSlot, err := s.acquireExecutionBadBlockFetch(ctx) + if err != nil { + return errors.Wrap(err, "failed to acquire execution bad block fetch slot") + } + + defer releaseFetchSlot() + // Fetch the bad blocks from the execution node. blocks, err := s.node.Execution().GetBadBlocks(ctx) if err != nil { diff --git a/pkg/agent/limiter.go b/pkg/agent/limiter.go new file mode 100644 index 0000000..e04f7c3 --- /dev/null +++ b/pkg/agent/limiter.go @@ -0,0 +1,81 @@ +package agent + +import ( + "context" + "sync" + + "golang.org/x/sync/semaphore" +) + +// Fetch concurrency defaults. Both paths read a whole response into memory, so +// peak usage scales with how many are in flight rather than with the response +// size alone. +// +// In `single` mode one process runs an agent per node and every agent reacts to +// the same block event, so without a bound the peak scales with the node count: +// on a devnet with ~100MB states, 47 agents peaked near 6GB and were OOM killed. +// Bounding in-flight fetches makes peak memory a function of these limits rather +// than of how many nodes are configured. +const ( + defaultMaxConcurrentBeaconStateFetches = 10 + // Bad block responses carry every bad block the node still holds, so they can + // be large and are decoded rather than streamed. Kept lower than the state + // limit because the decoded slice is retained for the whole indexing pass. + defaultMaxConcurrentExecutionBadBlockFetches = 4 +) + +// fetchLimiter is a lazily sized process-wide concurrency budget. +// +// The budget is package-level so that agents constructed independently still +// share it; in `single` mode they are separate agent instances in the same +// process. The first caller fixes the size, so mixed per-agent limits are not +// supported - the value from whichever agent starts first wins. +type fetchLimiter struct { + once sync.Once + sem *semaphore.Weighted +} + +// acquire blocks until a slot is free or ctx is cancelled, returning the release +// function. def is used when limit is unset. +func (l *fetchLimiter) acquire(ctx context.Context, limit, def int) (func(), error) { + l.once.Do(func() { + if limit <= 0 { + limit = def + } + + l.sem = semaphore.NewWeighted(int64(limit)) + }) + + if err := l.sem.Acquire(ctx, 1); err != nil { + return nil, err + } + + return func() { l.sem.Release(1) }, nil +} + +var ( + beaconStateFetchLimiter fetchLimiter + executionBadBlockFetchLimiter fetchLimiter +) + +// acquireBeaconStateFetch bounds concurrent beacon state fetches. The returned +// release function must be called once the state and its compressed copy are no +// longer referenced. +func (s *agent) acquireBeaconStateFetch(ctx context.Context) (func(), error) { + return beaconStateFetchLimiter.acquire( + ctx, + s.Config.Ethereum.GetMaxConcurrentBeaconStateFetches(), + defaultMaxConcurrentBeaconStateFetches, + ) +} + +// acquireExecutionBadBlockFetch bounds concurrent bad block fetches. The +// returned release function must be called once the decoded blocks are no longer +// referenced, which is after indexing rather than after the fetch. +func (s *agent) acquireExecutionBadBlockFetch(ctx context.Context) (func(), error) { + return executionBadBlockFetchLimiter.acquire( + ctx, + s.Config.Ethereum.GetMaxConcurrentExecutionBadBlockFetches(), + defaultMaxConcurrentExecutionBadBlockFetches, + ) +} From 7ae88eb2c48ff4fe64a91b8d3c754b592c77642f Mon Sep 17 00:00:00 2001 From: Andrew Davis <1709934+Savid@users.noreply.github.com> Date: Tue, 28 Jul 2026 14:15:19 +1000 Subject: [PATCH 2/3] fix(agent): raise bad block fetch limit to 10, trim comments Bad block responses are roughly an order of magnitude smaller than beacon states, so the same limit costs far less memory on that path. The slot is also held across the whole indexing pass, which is slow enough that a lower limit throttles throughput without meaningfully lowering the peak. Co-Authored-By: Claude Opus 5 (1M context) --- pkg/agent/consensus.go | 5 ++--- pkg/agent/ethereum/config.go | 20 ++++++------------- pkg/agent/execution.go | 5 ++--- pkg/agent/limiter.go | 37 +++++++++--------------------------- 4 files changed, 19 insertions(+), 48 deletions(-) diff --git a/pkg/agent/consensus.go b/pkg/agent/consensus.go index 396f667..712b903 100644 --- a/pkg/agent/consensus.go +++ b/pkg/agent/consensus.go @@ -77,9 +77,8 @@ func (s *agent) fetchAndIndexBeaconState(ctx context.Context, slot phase0.Slot) stateID = fmt.Sprintf("%d", slot) } - // Bound how many states are in flight process-wide. The raw state and its - // compressed copy are both held until the upload finishes, so the slot is - // kept for that whole window rather than just the fetch. + // Held until the upload finishes, not just the fetch, as the raw state and its + // compressed copy are both live until then. releaseFetchSlot, err := s.acquireBeaconStateFetch(ctx) if err != nil { return errors.Wrap(err, "failed to acquire beacon state fetch slot") diff --git a/pkg/agent/ethereum/config.go b/pkg/agent/ethereum/config.go index 8409235..495be7d 100644 --- a/pkg/agent/ethereum/config.go +++ b/pkg/agent/ethereum/config.go @@ -28,22 +28,16 @@ type Config struct { // the beacon node cache. BeaconStateAgeThresholdEpochs uint64 `yaml:"beaconStateAgeThresholdEpochs" default:"1"` - // MaxConcurrentBeaconStateFetches bounds how many beacon states may be fetched - // at once across every agent in the process. Each fetch holds the raw state and - // its compressed copy in memory, so this caps peak memory independently of how - // many agents are configured. Applied process-wide by whichever agent starts - // first, so it is only meaningful when set consistently across agents. + // MaxConcurrentBeaconStateFetches bounds in-flight beacon state fetches across + // every agent in the process, capping peak memory independently of how many + // agents are configured. Applied process-wide by whichever agent starts first. MaxConcurrentBeaconStateFetches int `yaml:"maxConcurrentBeaconStateFetches" default:"10"` - // MaxConcurrentExecutionBadBlockFetches bounds how many bad block fetches may - // be in flight at once across every agent in the process. The response carries - // every bad block the node still holds and is decoded into memory for the whole - // indexing pass. Applied process-wide, as above. - MaxConcurrentExecutionBadBlockFetches int `yaml:"maxConcurrentExecutionBadBlockFetches" default:"4"` + // MaxConcurrentExecutionBadBlockFetches bounds in-flight bad block fetches, + // as above. + MaxConcurrentExecutionBadBlockFetches int `yaml:"maxConcurrentExecutionBadBlockFetches" default:"10"` } -// GetMaxConcurrentBeaconStateFetches returns the configured beacon state fetch -// concurrency, or 0 to use the agent default. func (c *Config) GetMaxConcurrentBeaconStateFetches() int { if c.MaxConcurrentBeaconStateFetches <= 0 { return 0 @@ -52,8 +46,6 @@ func (c *Config) GetMaxConcurrentBeaconStateFetches() int { return c.MaxConcurrentBeaconStateFetches } -// GetMaxConcurrentExecutionBadBlockFetches returns the configured bad block fetch -// concurrency, or 0 to use the agent default. func (c *Config) GetMaxConcurrentExecutionBadBlockFetches() int { if c.MaxConcurrentExecutionBadBlockFetches <= 0 { return 0 diff --git a/pkg/agent/execution.go b/pkg/agent/execution.go index 5b83dd5..d73bcb4 100644 --- a/pkg/agent/execution.go +++ b/pkg/agent/execution.go @@ -97,9 +97,8 @@ func (s *agent) fetchAndIndexExecutionBlockTrace(ctx context.Context, blockNumbe } func (s *agent) fetchAndIndexExecutionBadBlocks(ctx context.Context) error { - // Bound how many bad block responses are decoded at once process-wide. The - // decoded slice is retained for the whole indexing pass below, so the slot is - // held until indexing finishes rather than released after the fetch. + // Held across the indexing pass below, as the decoded slice is retained until + // it completes. releaseFetchSlot, err := s.acquireExecutionBadBlockFetch(ctx) if err != nil { return errors.Wrap(err, "failed to acquire execution bad block fetch slot") diff --git a/pkg/agent/limiter.go b/pkg/agent/limiter.go index e04f7c3..3eb7786 100644 --- a/pkg/agent/limiter.go +++ b/pkg/agent/limiter.go @@ -7,36 +7,23 @@ import ( "golang.org/x/sync/semaphore" ) -// Fetch concurrency defaults. Both paths read a whole response into memory, so -// peak usage scales with how many are in flight rather than with the response -// size alone. -// -// In `single` mode one process runs an agent per node and every agent reacts to -// the same block event, so without a bound the peak scales with the node count: -// on a devnet with ~100MB states, 47 agents peaked near 6GB and were OOM killed. -// Bounding in-flight fetches makes peak memory a function of these limits rather -// than of how many nodes are configured. +// Both paths read a whole response into memory, so peak usage is the limit +// multiplied by the response size. States are roughly an order of magnitude +// larger than bad block responses. const ( - defaultMaxConcurrentBeaconStateFetches = 10 - // Bad block responses carry every bad block the node still holds, so they can - // be large and are decoded rather than streamed. Kept lower than the state - // limit because the decoded slice is retained for the whole indexing pass. - defaultMaxConcurrentExecutionBadBlockFetches = 4 + defaultMaxConcurrentBeaconStateFetches = 10 + defaultMaxConcurrentExecutionBadBlockFetches = 10 ) -// fetchLimiter is a lazily sized process-wide concurrency budget. -// -// The budget is package-level so that agents constructed independently still -// share it; in `single` mode they are separate agent instances in the same -// process. The first caller fixes the size, so mixed per-agent limits are not -// supported - the value from whichever agent starts first wins. +// fetchLimiter is a lazily sized process-wide concurrency budget. It is +// package-level because `single` mode runs an agent per node in one process and +// constructs each independently, so a per-agent budget would bound nothing. The +// first caller fixes the size. type fetchLimiter struct { once sync.Once sem *semaphore.Weighted } -// acquire blocks until a slot is free or ctx is cancelled, returning the release -// function. def is used when limit is unset. func (l *fetchLimiter) acquire(ctx context.Context, limit, def int) (func(), error) { l.once.Do(func() { if limit <= 0 { @@ -58,9 +45,6 @@ var ( executionBadBlockFetchLimiter fetchLimiter ) -// acquireBeaconStateFetch bounds concurrent beacon state fetches. The returned -// release function must be called once the state and its compressed copy are no -// longer referenced. func (s *agent) acquireBeaconStateFetch(ctx context.Context) (func(), error) { return beaconStateFetchLimiter.acquire( ctx, @@ -69,9 +53,6 @@ func (s *agent) acquireBeaconStateFetch(ctx context.Context) (func(), error) { ) } -// acquireExecutionBadBlockFetch bounds concurrent bad block fetches. The -// returned release function must be called once the decoded blocks are no longer -// referenced, which is after indexing rather than after the fetch. func (s *agent) acquireExecutionBadBlockFetch(ctx context.Context) (func(), error) { return executionBadBlockFetchLimiter.acquire( ctx, From ec559c1016f80d674d050f31184f137da3435eb0 Mon Sep 17 00:00:00 2001 From: Andrew Davis <1709934+Savid@users.noreply.github.com> Date: Tue, 28 Jul 2026 14:23:59 +1000 Subject: [PATCH 3/3] chore: extract repeated string literals into constants Clears the goconst failures blocking CI. No behaviour change - every replacement is a literal swapped for a constant holding the same value. Reuses the existing indexer Key* constants where the literals duplicated them, and adds constants for the remaining repeated logrus field keys, prometheus labels, gorm order clauses and test fixtures. Test-only values live in _test.go files so they are not compiled into the binary. The four lock tests in pkg/server/persistence fail on master as well and are untouched by this change. Co-Authored-By: Claude Opus 5 (1M context) --- pkg/agent/agent.go | 21 +++-- pkg/agent/consensus.go | 4 +- pkg/agent/metrics.go | 8 +- pkg/compression/compressor_test.go | 23 +++-- .../persistence/beacon_bad_blob_test.go | 20 ++--- .../persistence/beacon_bad_block_test.go | 20 ++--- pkg/server/persistence/beacon_block_test.go | 20 ++--- pkg/server/persistence/beacon_state_test.go | 18 ++-- .../persistence/execution_bad_block_test.go | 12 +-- .../persistence/execution_block_trace_test.go | 12 +-- pkg/server/persistence/filter_names_test.go | 17 ++++ pkg/server/persistence/lock.go | 25 +++--- pkg/server/service/api/api.go | 14 +-- .../service/indexer/beacon_bad_block_test.go | 2 +- .../service/indexer/beacon_block_test.go | 2 +- .../service/indexer/beacon_state_test.go | 2 +- .../indexer/execution_bad_block_test.go | 2 +- .../indexer/execution_block_trace_test.go | 2 +- pkg/server/service/indexer/indexer.go | 20 +++-- pkg/server/service/indexer/permanent_store.go | 86 +++++++++---------- .../service/indexer/permanent_store_test.go | 20 ++--- pkg/server/service/indexer/retention.go | 58 ++++++------- pkg/server/service/indexer/testdata_test.go | 7 ++ pkg/store/metrics.go | 16 ++-- pkg/store/mock.go | 12 ++- pkg/store/type.go | 4 + 26 files changed, 252 insertions(+), 195 deletions(-) create mode 100644 pkg/server/persistence/filter_names_test.go create mode 100644 pkg/server/service/indexer/testdata_test.go diff --git a/pkg/agent/agent.go b/pkg/agent/agent.go index 6bcbe5e..6c3a9af 100644 --- a/pkg/agent/agent.go +++ b/pkg/agent/agent.go @@ -55,7 +55,14 @@ type agent struct { compressor *compression.Compressor } -const namespace = "tracoor_agent" +const ( + namespace = "tracoor_agent" + + logKeyPurpose = "purpose" + logKeySlot = "slot" + labelAgent = "agent" + labelQueue = "queue" +) func New(ctx context.Context, log logrus.FieldLogger, config *Config) (*agent, error) { if config == nil { @@ -127,9 +134,9 @@ func (s *agent) Start(ctx context.Context) error { } logCtx := s.log.WithFields(logrus.Fields{ - "event_slot": event.Slot, - "event_root": fmt.Sprintf("%#x", event.Block), - "purpose": "execution_block_trace", + "event_slot": event.Slot, + "event_root": fmt.Sprintf("%#x", event.Block), + logKeyPurpose: "execution_block_trace", }) // Check if the block is too old to bother fetching. @@ -195,9 +202,9 @@ func (s *agent) Start(ctx context.Context) error { s.node.Beacon().Node().OnBlock(ctx, func(ctx context.Context, event *eth2v1.BlockEvent) error { logCtx := s.log.WithFields(logrus.Fields{ "event_topic": "block", - "slot": event.Slot, + logKeySlot: event.Slot, "root": fmt.Sprintf("%#x", event.Block), - "purpose": "beacon_state_and_block", + logKeyPurpose: "beacon_state_and_block", }) if ignore, err := s.node.ShouldIgnoreEventFromSlot(event.Slot); err != nil { @@ -224,7 +231,7 @@ func (s *agent) Start(ctx context.Context) error { "event_old_head_block": rootAsString(chainReorg.OldHeadBlock), "event_new_head_block": rootAsString(chainReorg.NewHeadBlock), "event_depth": chainReorg.Depth, - "purpose": "chain_reorg", + logKeyPurpose: "chain_reorg", "event_slot": chainReorg.Slot, }, ) diff --git a/pkg/agent/consensus.go b/pkg/agent/consensus.go index 712b903..583f23b 100644 --- a/pkg/agent/consensus.go +++ b/pkg/agent/consensus.go @@ -398,7 +398,7 @@ func (s *agent) fetchAndIndexBeaconBadBlocks(ctx context.Context, path string) e }) if err != nil { s.log.WithFields(logrus.Fields{ - "slot": slot, + logKeySlot: slot, "blockRoot": blockRoot, "filePath": filePath, }).WithError(err).Error("Failed to save beacon bad block to store") @@ -616,7 +616,7 @@ func (s *agent) fetchAndIndexBeaconBadBlobs(ctx context.Context, path string) er }) if err != nil { s.log.WithFields(logrus.Fields{ - "slot": slot, + logKeySlot: slot, "blockRoot": blockRoot, "index": index, "filePath": filePath, diff --git a/pkg/agent/metrics.go b/pkg/agent/metrics.go index 541e77d..a82c9ed 100644 --- a/pkg/agent/metrics.go +++ b/pkg/agent/metrics.go @@ -39,23 +39,23 @@ func GetMetricsInstance(namespace string) *Metrics { Namespace: namespace, Name: "queue_size", Help: "The size of the queue", - }, []string{"queue", "agent"}), + }, []string{labelQueue, labelAgent}), queueItemProcessingTime: prometheus.NewHistogramVec(prometheus.HistogramOpts{ Namespace: namespace, Name: "queue_item_processing_time_seconds", Help: "The time it takes to process an item from the queue", Buckets: prometheus.LinearBuckets(0, 3, 10), - }, []string{"queue", "agent"}), + }, []string{labelQueue, labelAgent}), itemExported: prometheus.NewCounterVec(prometheus.CounterOpts{ Namespace: namespace, Name: "item_exported", Help: "The number of items exported", - }, []string{"queue", "agent"}), + }, []string{labelQueue, labelAgent}), queueItemSkipped: prometheus.NewCounterVec(prometheus.CounterOpts{ Namespace: namespace, Name: "queue_item_skipped", Help: "The number of items skipped", - }, []string{"queue", "agent"}), + }, []string{labelQueue, labelAgent}), } prometheus.MustRegister(metricsInstance.queueSize) diff --git a/pkg/compression/compressor_test.go b/pkg/compression/compressor_test.go index e443183..3605256 100644 --- a/pkg/compression/compressor_test.go +++ b/pkg/compression/compressor_test.go @@ -10,6 +10,11 @@ import ( "github.com/ethpandaops/tracoor/pkg/compression" ) +const ( + testFilename = "test" + testFilenameGz = "test.gz" +) + func TestNewCompressor(t *testing.T) { c := compression.NewCompressor() assert.NotNil(t, c) @@ -70,13 +75,13 @@ func TestCompressor_Decompress(t *testing.T) { { name: "Decompress Gzip", data: compressed, - filename: "test.gz", + filename: testFilenameGz, wantErr: false, }, { name: "Decompress with nil data", data: nil, - filename: "test.gz", + filename: testFilenameGz, wantErr: true, }, { @@ -111,13 +116,13 @@ func TestAddExtension(t *testing.T) { }{ { name: "Add Gzip extension", - filename: "test", + filename: testFilename, algorithm: compression.Gzip, want: "test.gz", }, { name: "Extension already present", - filename: "test.gz", + filename: testFilenameGz, algorithm: compression.Gzip, want: "test.gz", }, @@ -142,13 +147,13 @@ func TestRemoveExtension(t *testing.T) { }{ { name: "Remove Gzip extension", - filename: "test.gz", + filename: testFilenameGz, algorithm: compression.Gzip, want: "test", }, { name: "No extension to remove", - filename: "test", + filename: testFilename, algorithm: compression.Gzip, want: "test", }, @@ -179,13 +184,13 @@ func TestHasCompressionExtension(t *testing.T) { }{ { name: "Has Gzip extension", - filename: "test.gz", + filename: testFilenameGz, algorithm: compression.Gzip, want: true, }, { name: "No Gzip extension", - filename: "test", + filename: testFilename, algorithm: compression.Gzip, want: false, }, @@ -210,7 +215,7 @@ func TestGetCompressionAlgorithm(t *testing.T) { }{ { name: "Get Gzip algorithm", - filename: "test.gz", + filename: testFilenameGz, want: compression.Gzip, wantErr: false, }, diff --git a/pkg/server/persistence/beacon_bad_blob_test.go b/pkg/server/persistence/beacon_bad_blob_test.go index 04a7966..f87c3a5 100644 --- a/pkg/server/persistence/beacon_bad_blob_test.go +++ b/pkg/server/persistence/beacon_bad_blob_test.go @@ -98,7 +98,7 @@ func TestListBeaconBadBlob(t *testing.T) { } page := &PaginationCursor{} - mock.ExpectQuery("SELECT \\* FROM").WithArgs(filter.ID).WillReturnRows(sqlmock.NewRows([]string{"id", "node"}).AddRow("test-id", "test-node")) + mock.ExpectQuery("SELECT \\* FROM").WithArgs(filter.ID).WillReturnRows(sqlmock.NewRows([]string{columnID, columnNode}).AddRow("test-id", "test-node")) blobs, err := indexer.ListBeaconBadBlob(ctx, filter, page) assert.NoError(t, err) @@ -301,15 +301,15 @@ func TestBeaconBadBlobIndividualFilters(t *testing.T) { name string filter BeaconBadBlobFilter }{ - {"ID", BeaconBadBlobFilter{ID: &beaconBlob.ID}}, - {"Node", BeaconBadBlobFilter{Node: &beaconBlob.Node}}, - {"Slot", BeaconBadBlobFilter{Slot: &slot}}, - {"Epoch", BeaconBadBlobFilter{Epoch: &epoch}}, - {"BlockRoot", BeaconBadBlobFilter{BlockRoot: &beaconBlob.BlockRoot}}, - {"NodeVersion", BeaconBadBlobFilter{NodeVersion: &beaconBlob.NodeVersion}}, - {"Location", BeaconBadBlobFilter{Location: &beaconBlob.Location}}, - {"Network", BeaconBadBlobFilter{Network: &beaconBlob.Network}}, - {"BeaconImplementation", BeaconBadBlobFilter{BeaconImplementation: &beaconBlob.BeaconImplementation}}, + {fieldID, BeaconBadBlobFilter{ID: &beaconBlob.ID}}, + {fieldNode, BeaconBadBlobFilter{Node: &beaconBlob.Node}}, + {fieldSlot, BeaconBadBlobFilter{Slot: &slot}}, + {fieldEpoch, BeaconBadBlobFilter{Epoch: &epoch}}, + {fieldBlockRoot, BeaconBadBlobFilter{BlockRoot: &beaconBlob.BlockRoot}}, + {fieldNodeVersion, BeaconBadBlobFilter{NodeVersion: &beaconBlob.NodeVersion}}, + {fieldLocation, BeaconBadBlobFilter{Location: &beaconBlob.Location}}, + {fieldNetwork, BeaconBadBlobFilter{Network: &beaconBlob.Network}}, + {fieldBeaconImplementation, BeaconBadBlobFilter{BeaconImplementation: &beaconBlob.BeaconImplementation}}, {"Index", BeaconBadBlobFilter{Index: &index}}, } diff --git a/pkg/server/persistence/beacon_bad_block_test.go b/pkg/server/persistence/beacon_bad_block_test.go index 7bd296b..594fe48 100644 --- a/pkg/server/persistence/beacon_bad_block_test.go +++ b/pkg/server/persistence/beacon_bad_block_test.go @@ -96,7 +96,7 @@ func TestListBeaconBadBlock(t *testing.T) { } page := &PaginationCursor{} - mock.ExpectQuery("SELECT \\* FROM").WithArgs(filter.ID).WillReturnRows(sqlmock.NewRows([]string{"id", "node"}).AddRow("test-id", "test-node")) + mock.ExpectQuery("SELECT \\* FROM").WithArgs(filter.ID).WillReturnRows(sqlmock.NewRows([]string{columnID, columnNode}).AddRow("test-id", "test-node")) blocks, err := indexer.ListBeaconBadBlock(ctx, filter, page) assert.NoError(t, err) @@ -288,15 +288,15 @@ func TestBeaconBadBlockIndividualFilters(t *testing.T) { name string filter BeaconBadBlockFilter }{ - {"ID", BeaconBadBlockFilter{ID: &beaconBlock.ID}}, - {"Node", BeaconBadBlockFilter{Node: &beaconBlock.Node}}, - {"Slot", BeaconBadBlockFilter{Slot: &slot}}, - {"Epoch", BeaconBadBlockFilter{Epoch: &epoch}}, - {"BlockRoot", BeaconBadBlockFilter{BlockRoot: &beaconBlock.BlockRoot}}, - {"NodeVersion", BeaconBadBlockFilter{NodeVersion: &beaconBlock.NodeVersion}}, - {"Location", BeaconBadBlockFilter{Location: &beaconBlock.Location}}, - {"Network", BeaconBadBlockFilter{Network: &beaconBlock.Network}}, - {"BeaconImplementation", BeaconBadBlockFilter{BeaconImplementation: &beaconBlock.BeaconImplementation}}, + {fieldID, BeaconBadBlockFilter{ID: &beaconBlock.ID}}, + {fieldNode, BeaconBadBlockFilter{Node: &beaconBlock.Node}}, + {fieldSlot, BeaconBadBlockFilter{Slot: &slot}}, + {fieldEpoch, BeaconBadBlockFilter{Epoch: &epoch}}, + {fieldBlockRoot, BeaconBadBlockFilter{BlockRoot: &beaconBlock.BlockRoot}}, + {fieldNodeVersion, BeaconBadBlockFilter{NodeVersion: &beaconBlock.NodeVersion}}, + {fieldLocation, BeaconBadBlockFilter{Location: &beaconBlock.Location}}, + {fieldNetwork, BeaconBadBlockFilter{Network: &beaconBlock.Network}}, + {fieldBeaconImplementation, BeaconBadBlockFilter{BeaconImplementation: &beaconBlock.BeaconImplementation}}, } for _, tc := range testCases { diff --git a/pkg/server/persistence/beacon_block_test.go b/pkg/server/persistence/beacon_block_test.go index 237e3c9..4fa1527 100644 --- a/pkg/server/persistence/beacon_block_test.go +++ b/pkg/server/persistence/beacon_block_test.go @@ -97,7 +97,7 @@ func TestListBeaconBlock(t *testing.T) { } page := &PaginationCursor{} - mock.ExpectQuery("SELECT \\* FROM").WithArgs(filter.ID).WillReturnRows(sqlmock.NewRows([]string{"id", "node"}).AddRow(testID, "test-node")) + mock.ExpectQuery("SELECT \\* FROM").WithArgs(filter.ID).WillReturnRows(sqlmock.NewRows([]string{columnID, columnNode}).AddRow(testID, "test-node")) blocks, err := indexer.ListBeaconBlock(ctx, filter, page) assert.NoError(t, err) @@ -289,15 +289,15 @@ func TestBeaconBlockIndividualFilters(t *testing.T) { name string filter BeaconBlockFilter }{ - {"ID", BeaconBlockFilter{ID: &beaconBlock.ID}}, - {"Node", BeaconBlockFilter{Node: &beaconBlock.Node}}, - {"Slot", BeaconBlockFilter{Slot: &slot}}, - {"Epoch", BeaconBlockFilter{Epoch: &epoch}}, - {"BlockRoot", BeaconBlockFilter{BlockRoot: &beaconBlock.BlockRoot}}, - {"NodeVersion", BeaconBlockFilter{NodeVersion: &beaconBlock.NodeVersion}}, - {"Location", BeaconBlockFilter{Location: &beaconBlock.Location}}, - {"Network", BeaconBlockFilter{Network: &beaconBlock.Network}}, - {"BeaconImplementation", BeaconBlockFilter{BeaconImplementation: &beaconBlock.BeaconImplementation}}, + {fieldID, BeaconBlockFilter{ID: &beaconBlock.ID}}, + {fieldNode, BeaconBlockFilter{Node: &beaconBlock.Node}}, + {fieldSlot, BeaconBlockFilter{Slot: &slot}}, + {fieldEpoch, BeaconBlockFilter{Epoch: &epoch}}, + {fieldBlockRoot, BeaconBlockFilter{BlockRoot: &beaconBlock.BlockRoot}}, + {fieldNodeVersion, BeaconBlockFilter{NodeVersion: &beaconBlock.NodeVersion}}, + {fieldLocation, BeaconBlockFilter{Location: &beaconBlock.Location}}, + {fieldNetwork, BeaconBlockFilter{Network: &beaconBlock.Network}}, + {fieldBeaconImplementation, BeaconBlockFilter{BeaconImplementation: &beaconBlock.BeaconImplementation}}, } for _, tc := range testCases { diff --git a/pkg/server/persistence/beacon_state_test.go b/pkg/server/persistence/beacon_state_test.go index 3152f33..95adc0b 100644 --- a/pkg/server/persistence/beacon_state_test.go +++ b/pkg/server/persistence/beacon_state_test.go @@ -96,7 +96,7 @@ func TestListBeaconState(t *testing.T) { } page := &PaginationCursor{} - mock.ExpectQuery("SELECT \\* FROM").WithArgs(filter.ID).WillReturnRows(sqlmock.NewRows([]string{"id", "node"}).AddRow("test-id", "test-node")) + mock.ExpectQuery("SELECT \\* FROM").WithArgs(filter.ID).WillReturnRows(sqlmock.NewRows([]string{columnID, columnNode}).AddRow("test-id", "test-node")) states, err := indexer.ListBeaconState(ctx, filter, page) assert.NoError(t, err) @@ -288,15 +288,15 @@ func TestBeaconStateIndividualFilters(t *testing.T) { name string filter BeaconStateFilter }{ - {"ID", BeaconStateFilter{ID: &beaconState.ID}}, - {"Node", BeaconStateFilter{Node: &beaconState.Node}}, - {"Slot", BeaconStateFilter{Slot: &slot}}, - {"Epoch", BeaconStateFilter{Epoch: &epoch}}, + {fieldID, BeaconStateFilter{ID: &beaconState.ID}}, + {fieldNode, BeaconStateFilter{Node: &beaconState.Node}}, + {fieldSlot, BeaconStateFilter{Slot: &slot}}, + {fieldEpoch, BeaconStateFilter{Epoch: &epoch}}, {"StateRoot", BeaconStateFilter{StateRoot: &beaconState.StateRoot}}, - {"NodeVersion", BeaconStateFilter{NodeVersion: &beaconState.NodeVersion}}, - {"Location", BeaconStateFilter{Location: &beaconState.Location}}, - {"Network", BeaconStateFilter{Network: &beaconState.Network}}, - {"BeaconImplementation", BeaconStateFilter{BeaconImplementation: &beaconState.BeaconImplementation}}, + {fieldNodeVersion, BeaconStateFilter{NodeVersion: &beaconState.NodeVersion}}, + {fieldLocation, BeaconStateFilter{Location: &beaconState.Location}}, + {fieldNetwork, BeaconStateFilter{Network: &beaconState.Network}}, + {fieldBeaconImplementation, BeaconStateFilter{BeaconImplementation: &beaconState.BeaconImplementation}}, } for _, tc := range testCases { diff --git a/pkg/server/persistence/execution_bad_block_test.go b/pkg/server/persistence/execution_bad_block_test.go index a54dcda..62b8c40 100644 --- a/pkg/server/persistence/execution_bad_block_test.go +++ b/pkg/server/persistence/execution_bad_block_test.go @@ -96,7 +96,7 @@ func TestListExecutionBadBlock(t *testing.T) { } page := &PaginationCursor{} - mock.ExpectQuery("SELECT \\* FROM").WithArgs(filter.ID).WillReturnRows(sqlmock.NewRows([]string{"id", "node"}).AddRow("test-id", "test-node")) + mock.ExpectQuery("SELECT \\* FROM").WithArgs(filter.ID).WillReturnRows(sqlmock.NewRows([]string{columnID, columnNode}).AddRow("test-id", "test-node")) blocks, err := indexer.ListExecutionBadBlock(ctx, filter, page) assert.NoError(t, err) @@ -264,14 +264,14 @@ func TestExecutionBadBlockIndividualFilters(t *testing.T) { name string filter ExecutionBadBlockFilter }{ - {"ID", ExecutionBadBlockFilter{ID: &block.ID}}, - {"Node", ExecutionBadBlockFilter{Node: &block.Node}}, + {fieldID, ExecutionBadBlockFilter{ID: &block.ID}}, + {fieldNode, ExecutionBadBlockFilter{Node: &block.Node}}, {"BlockHash", ExecutionBadBlockFilter{BlockHash: &block.BlockHash}}, {"BlockNumber", ExecutionBadBlockFilter{BlockNumber: &block.BlockNumber.Int64}}, {"BlockExtraData", ExecutionBadBlockFilter{BlockExtraData: &block.BlockExtraData.String}}, - {"NodeVersion", ExecutionBadBlockFilter{NodeVersion: &block.NodeVersion}}, - {"Location", ExecutionBadBlockFilter{Location: &block.Location}}, - {"Network", ExecutionBadBlockFilter{Network: &block.Network}}, + {fieldNodeVersion, ExecutionBadBlockFilter{NodeVersion: &block.NodeVersion}}, + {fieldLocation, ExecutionBadBlockFilter{Location: &block.Location}}, + {fieldNetwork, ExecutionBadBlockFilter{Network: &block.Network}}, {"ExecutionImplementation", ExecutionBadBlockFilter{ExecutionImplementation: &block.ExecutionImplementation}}, } diff --git a/pkg/server/persistence/execution_block_trace_test.go b/pkg/server/persistence/execution_block_trace_test.go index bd700b3..7c90149 100644 --- a/pkg/server/persistence/execution_block_trace_test.go +++ b/pkg/server/persistence/execution_block_trace_test.go @@ -94,7 +94,7 @@ func TestListExecutionBlockTrace(t *testing.T) { } page := &PaginationCursor{} - mock.ExpectQuery("SELECT \\* FROM").WithArgs(filter.ID).WillReturnRows(sqlmock.NewRows([]string{"id", "node"}).AddRow("test-id", "test-node")) + mock.ExpectQuery("SELECT \\* FROM").WithArgs(filter.ID).WillReturnRows(sqlmock.NewRows([]string{columnID, columnNode}).AddRow("test-id", "test-node")) traces, err := indexer.ListExecutionBlockTrace(ctx, filter, page) assert.NoError(t, err) @@ -123,13 +123,13 @@ func TestExecutionBlockTraceFilters(t *testing.T) { name string filter ExecutionBlockTraceFilter }{ - {"ID", ExecutionBlockTraceFilter{ID: &trace.ID}}, - {"Node", ExecutionBlockTraceFilter{Node: &trace.Node}}, + {fieldID, ExecutionBlockTraceFilter{ID: &trace.ID}}, + {fieldNode, ExecutionBlockTraceFilter{Node: &trace.Node}}, {"BlockHash", ExecutionBlockTraceFilter{BlockHash: &trace.BlockHash}}, {"BlockNumber", ExecutionBlockTraceFilter{BlockNumber: &trace.BlockNumber}}, - {"NodeVersion", ExecutionBlockTraceFilter{NodeVersion: &trace.NodeVersion}}, - {"Location", ExecutionBlockTraceFilter{Location: &trace.Location}}, - {"Network", ExecutionBlockTraceFilter{Network: &trace.Network}}, + {fieldNodeVersion, ExecutionBlockTraceFilter{NodeVersion: &trace.NodeVersion}}, + {fieldLocation, ExecutionBlockTraceFilter{Location: &trace.Location}}, + {fieldNetwork, ExecutionBlockTraceFilter{Network: &trace.Network}}, {"ExecutionImplementation", ExecutionBlockTraceFilter{ExecutionImplementation: &trace.ExecutionImplementation}}, } diff --git a/pkg/server/persistence/filter_names_test.go b/pkg/server/persistence/filter_names_test.go new file mode 100644 index 0000000..110fce4 --- /dev/null +++ b/pkg/server/persistence/filter_names_test.go @@ -0,0 +1,17 @@ +package persistence + +// Filter field names shared by the table-driven filter tests. +const ( + fieldID = "ID" + fieldNode = "Node" + fieldSlot = "Slot" + fieldEpoch = "Epoch" + fieldBlockRoot = "BlockRoot" + fieldNodeVersion = "NodeVersion" + fieldLocation = "Location" + fieldNetwork = "Network" + fieldBeaconImplementation = "BeaconImplementation" + + columnID = "id" + columnNode = "node" +) diff --git a/pkg/server/persistence/lock.go b/pkg/server/persistence/lock.go index 171684b..820992f 100644 --- a/pkg/server/persistence/lock.go +++ b/pkg/server/persistence/lock.go @@ -10,6 +10,11 @@ import ( "gorm.io/gorm" ) +const ( + logKeyLock = "key" + logKeyOwner = "owner" +) + // DistributedLock represents a lock in the database. type DistributedLock struct { gorm.Model @@ -95,18 +100,18 @@ func (i *Indexer) AcquireLock(ctx context.Context, key, owner string, ttl time.D }) if err != nil { i.log.WithFields(logrus.Fields{ - "key": key, - "owner": owner, - "error": err.Error(), + logKeyLock: key, + logKeyOwner: owner, + "error": err.Error(), }).Debug("Failed to acquire lock") return false, errors.Wrap(err, "failed to acquire lock") } i.log.WithFields(logrus.Fields{ - "key": key, - "owner": owner, - "ttl": ttl, + logKeyLock: key, + logKeyOwner: owner, + "ttl": ttl, }).Debug("Acquired lock") return true, nil @@ -121,16 +126,16 @@ func (i *Indexer) ReleaseLock(ctx context.Context, key, owner string) error { if result.RowsAffected == 0 { i.log.WithFields(logrus.Fields{ - "key": key, - "owner": owner, + logKeyLock: key, + logKeyOwner: owner, }).Debug("Lock not found or not owned by the given owner") return nil } i.log.WithFields(logrus.Fields{ - "key": key, - "owner": owner, + logKeyLock: key, + logKeyOwner: owner, }).Debug("Released lock") return nil diff --git a/pkg/server/service/api/api.go b/pkg/server/service/api/api.go index ee91bb5..4b69a98 100644 --- a/pkg/server/service/api/api.go +++ b/pkg/server/service/api/api.go @@ -16,6 +16,8 @@ import ( const ( ServiceType = "tracoor.api" + + OrderFetchedAtDesc = "fetched_at DESC" ) type API struct { @@ -104,7 +106,7 @@ func (i *API) ListBeaconState(ctx context.Context, req *api.ListBeaconStateReque pagination := &indexer.PaginationCursor{ Limit: 100, Offset: 0, - OrderBy: "fetched_at DESC", + OrderBy: OrderFetchedAtDesc, } if req.Pagination != nil { @@ -237,7 +239,7 @@ func (i *API) ListBeaconBlock(ctx context.Context, req *api.ListBeaconBlockReque pagination := &indexer.PaginationCursor{ Limit: 100, Offset: 0, - OrderBy: "fetched_at DESC", + OrderBy: OrderFetchedAtDesc, } if req.Pagination != nil { @@ -370,7 +372,7 @@ func (i *API) ListBeaconBadBlock(ctx context.Context, req *api.ListBeaconBadBloc pagination := &indexer.PaginationCursor{ Limit: 100, Offset: 0, - OrderBy: "fetched_at DESC", + OrderBy: OrderFetchedAtDesc, } if req.Pagination != nil { @@ -503,7 +505,7 @@ func (i *API) ListBeaconBadBlob(ctx context.Context, req *api.ListBeaconBadBlobR pagination := &indexer.PaginationCursor{ Limit: 100, Offset: 0, - OrderBy: "fetched_at DESC", + OrderBy: OrderFetchedAtDesc, } if req.Pagination != nil { @@ -642,7 +644,7 @@ func (i *API) ListExecutionBlockTrace(ctx context.Context, req *api.ListExecutio pagination := &indexer.PaginationCursor{ Limit: 100, Offset: 0, - OrderBy: "fetched_at DESC", + OrderBy: OrderFetchedAtDesc, } if req.Pagination != nil { @@ -767,7 +769,7 @@ func (i *API) ListExecutionBadBlock(ctx context.Context, req *api.ListExecutionB pagination := &indexer.PaginationCursor{ Limit: 100, Offset: 0, - OrderBy: "fetched_at DESC", + OrderBy: OrderFetchedAtDesc, } if req.Pagination != nil { diff --git a/pkg/server/service/indexer/beacon_bad_block_test.go b/pkg/server/service/indexer/beacon_bad_block_test.go index 46d8814..4bacf3b 100644 --- a/pkg/server/service/indexer/beacon_bad_block_test.go +++ b/pkg/server/service/indexer/beacon_bad_block_test.go @@ -88,7 +88,7 @@ func TestIndexerBeaconBadBlockDownloading(t *testing.T) { location, err := index.Store().SaveBeaconBadBlock(ctx, &store.SaveParams{ Data: &compressedData, - Location: "data.json", + Location: testDataLocation, ContentEncoding: compression.Gzip.ContentEncoding, }) if err != nil { diff --git a/pkg/server/service/indexer/beacon_block_test.go b/pkg/server/service/indexer/beacon_block_test.go index a618d37..0fa7df7 100644 --- a/pkg/server/service/indexer/beacon_block_test.go +++ b/pkg/server/service/indexer/beacon_block_test.go @@ -90,7 +90,7 @@ func TestIndexerBeaconBlockDownloading(t *testing.T) { location, err := index.Store().SaveBeaconBlock(ctx, &store.SaveParams{ Data: &compressedData, - Location: "data.json", + Location: testDataLocation, ContentEncoding: compression.Gzip.ContentEncoding, }) if err != nil { diff --git a/pkg/server/service/indexer/beacon_state_test.go b/pkg/server/service/indexer/beacon_state_test.go index b968a1e..ae4f434 100644 --- a/pkg/server/service/indexer/beacon_state_test.go +++ b/pkg/server/service/indexer/beacon_state_test.go @@ -88,7 +88,7 @@ func TestIndexerBeaconStateDownloading(t *testing.T) { location, err := index.Store().SaveBeaconState(ctx, &store.SaveParams{ Data: &compressedData, - Location: "data.json", + Location: testDataLocation, ContentEncoding: compression.Gzip.ContentEncoding, }) if err != nil { diff --git a/pkg/server/service/indexer/execution_bad_block_test.go b/pkg/server/service/indexer/execution_bad_block_test.go index c462ba2..7d3e502 100644 --- a/pkg/server/service/indexer/execution_bad_block_test.go +++ b/pkg/server/service/indexer/execution_bad_block_test.go @@ -270,7 +270,7 @@ func TestIndexerExecutionBadBlockDownloading(t *testing.T) { location, err := index.Store().SaveExecutionBadBlock(ctx, &store.SaveParams{ Data: &compressedData, - Location: "data.json", + Location: testDataLocation, ContentEncoding: compression.Gzip.ContentEncoding, }) if err != nil { diff --git a/pkg/server/service/indexer/execution_block_trace_test.go b/pkg/server/service/indexer/execution_block_trace_test.go index 4786eb4..d210cf8 100644 --- a/pkg/server/service/indexer/execution_block_trace_test.go +++ b/pkg/server/service/indexer/execution_block_trace_test.go @@ -269,7 +269,7 @@ func TestIndexerExecutionBlockTraceDownloading(t *testing.T) { location, err := index.Store().SaveExecutionBlockTrace(ctx, &store.SaveParams{ Data: &compressedData, - Location: "data.json", + Location: testDataLocation, ContentEncoding: compression.Gzip.ContentEncoding, }) if err != nil { diff --git a/pkg/server/service/indexer/indexer.go b/pkg/server/service/indexer/indexer.go index 108e876..0ae565d 100644 --- a/pkg/server/service/indexer/indexer.go +++ b/pkg/server/service/indexer/indexer.go @@ -33,6 +33,12 @@ const ( KeyLocation = "location" KeyFetchedAt = "fetched_at" KeyBeaconImplementation = "beacon_implementation" + KeyID = "id" + KeyIndex = "index" + KeyLockKey = "lock_key" + + OrderFetchedAtDesc = "fetched_at DESC" + OrderFetchedAtAsc = "fetched_at ASC" ) type Indexer struct { @@ -291,7 +297,7 @@ func (i *Indexer) ListBeaconState(ctx context.Context, req *indexer.ListBeaconSt pagination := &persistence.PaginationCursor{ Limit: 1000, Offset: 0, - OrderBy: "fetched_at DESC", + OrderBy: OrderFetchedAtDesc, } if req.Pagination != nil { @@ -557,7 +563,7 @@ func (i *Indexer) ListBeaconBlock(ctx context.Context, req *indexer.ListBeaconBl pagination := &persistence.PaginationCursor{ Limit: 1000, Offset: 0, - OrderBy: "fetched_at DESC", + OrderBy: OrderFetchedAtDesc, } if req.Pagination != nil { @@ -815,7 +821,7 @@ func (i *Indexer) ListBeaconBadBlock(ctx context.Context, req *indexer.ListBeaco pagination := &persistence.PaginationCursor{ Limit: 1000, Offset: 0, - OrderBy: "fetched_at DESC", + OrderBy: OrderFetchedAtDesc, } if req.Pagination != nil { @@ -1010,7 +1016,7 @@ func (i *Indexer) CreateBeaconBadBlob(ctx context.Context, req *indexer.CreateBe KeyLocation: req.GetLocation().GetValue(), KeyFetchedAt: req.GetFetchedAt().AsTime(), KeyBeaconImplementation: req.GetBeaconImplementation().GetValue(), - "index": req.GetIndex().GetValue(), + KeyIndex: req.GetIndex().GetValue(), } if err := i.db.InsertBeaconBadBlob(ctx, ProtoBeaconBadBlobToDBBeaconBadBlob(badBlob)); err != nil { @@ -1080,7 +1086,7 @@ func (i *Indexer) ListBeaconBadBlob(ctx context.Context, req *indexer.ListBeacon pagination := &persistence.PaginationCursor{ Limit: 1000, Offset: 0, - OrderBy: "fetched_at DESC", + OrderBy: OrderFetchedAtDesc, } if req.Pagination != nil { @@ -1301,7 +1307,7 @@ func (i *Indexer) ListExecutionBlockTrace(ctx context.Context, req *indexer.List pagination := &persistence.PaginationCursor{ Limit: 1000, Offset: 0, - OrderBy: "fetched_at DESC", + OrderBy: OrderFetchedAtDesc, } if req.Pagination != nil { @@ -1511,7 +1517,7 @@ func (i *Indexer) ListExecutionBadBlock(ctx context.Context, req *indexer.ListEx pagination := &persistence.PaginationCursor{ Limit: 1000, Offset: 0, - OrderBy: "fetched_at DESC", + OrderBy: OrderFetchedAtDesc, } if req.Pagination != nil { diff --git a/pkg/server/service/indexer/permanent_store.go b/pkg/server/service/indexer/permanent_store.go index e6d9814..af73bd7 100644 --- a/pkg/server/service/indexer/permanent_store.go +++ b/pkg/server/service/indexer/permanent_store.go @@ -122,15 +122,15 @@ func (p *PermanentStore) QueueBlock(block PermanentStoreBlock) { select { case p.queue <- block: p.log.WithFields(logrus.Fields{ - "block_root": block.BlockRoot, - "network": block.Network, - "location": block.Location, + KeyBlockRoot: block.BlockRoot, + KeyNetwork: block.Network, + KeyLocation: block.Location, }).Debug("Queued block for permanent storage") default: p.log.WithFields(logrus.Fields{ - "block_root": block.BlockRoot, - "network": block.Network, - "location": block.Location, + KeyBlockRoot: block.BlockRoot, + KeyNetwork: block.Network, + KeyLocation: block.Location, }).Warn("Failed to queue block for permanent storage, queue is full") } } @@ -154,9 +154,9 @@ func (p *PermanentStore) processQueue(ctx context.Context) { if err := p.processBlock(ctx, block); err != nil { p.log.WithError(err).WithFields(logrus.Fields{ - "block_root": block.BlockRoot, - "network": block.Network, - "location": block.Location, + KeyBlockRoot: block.BlockRoot, + KeyNetwork: block.Network, + KeyLocation: block.Location, }).Error("Failed to process block for permanent storage") } } @@ -178,8 +178,8 @@ func (p *PermanentStore) processBlock(ctx context.Context, block PermanentStoreB // Check if we've already processed this block if _, ok := p.cache.Get(cacheKey); ok { p.log.WithFields(logrus.Fields{ - "block_root": block.BlockRoot, - "network": block.Network, + KeyBlockRoot: block.BlockRoot, + KeyNetwork: block.Network, }).Debug("Block already processed (cache hit)") return nil @@ -211,9 +211,9 @@ func (p *PermanentStore) processBlock(ctx context.Context, block PermanentStoreB // If the error indicates someone else has the lock, retry if err.Error() != "" && time.Since(startTime) < maxRetryDuration { p.log.WithFields(logrus.Fields{ - "block_root": block.BlockRoot, - "network": block.Network, - "lock_key": lockKey, + KeyBlockRoot: block.BlockRoot, + KeyNetwork: block.Network, + KeyLockKey: lockKey, "error": err.Error(), "elapsed": time.Since(startTime).String(), }).Debug("Failed to acquire lock, retrying...") @@ -233,9 +233,9 @@ func (p *PermanentStore) processBlock(ctx context.Context, block PermanentStoreB // If we couldn't acquire the lock but there's no error, retry if time.Since(startTime) < maxRetryDuration { p.log.WithFields(logrus.Fields{ - "block_root": block.BlockRoot, - "network": block.Network, - "lock_key": lockKey, + KeyBlockRoot: block.BlockRoot, + KeyNetwork: block.Network, + KeyLockKey: lockKey, "elapsed": time.Since(startTime).String(), }).Debug("Failed to acquire lock, retrying...") @@ -245,9 +245,9 @@ func (p *PermanentStore) processBlock(ctx context.Context, block PermanentStoreB } p.log.WithFields(logrus.Fields{ - "block_root": block.BlockRoot, - "network": block.Network, - "lock_key": lockKey, + KeyBlockRoot: block.BlockRoot, + KeyNetwork: block.Network, + KeyLockKey: lockKey, }).Debug("Failed to acquire lock after retries, another instance is processing this block") return nil @@ -256,9 +256,9 @@ func (p *PermanentStore) processBlock(ctx context.Context, block PermanentStoreB if !acquired { p.log.WithFields(logrus.Fields{ - "block_root": block.BlockRoot, - "network": block.Network, - "lock_key": lockKey, + KeyBlockRoot: block.BlockRoot, + KeyNetwork: block.Network, + KeyLockKey: lockKey, }).Debug("Failed to acquire lock after maximum retry duration") return nil @@ -267,9 +267,9 @@ func (p *PermanentStore) processBlock(ctx context.Context, block PermanentStoreB defer func() { if lerr := p.db.ReleaseLock(ctx, lockKey, p.nodeID); lerr != nil { p.log.WithError(lerr).WithFields(logrus.Fields{ - "block_root": block.BlockRoot, - "network": block.Network, - "lock_key": lockKey, + KeyBlockRoot: block.BlockRoot, + KeyNetwork: block.Network, + KeyLockKey: lockKey, }).Error("Failed to release lock") } }() @@ -277,8 +277,8 @@ func (p *PermanentStore) processBlock(ctx context.Context, block PermanentStoreB // Check again after acquiring the lock if _, ok := p.cache.Get(cacheKey); ok { p.log.WithFields(logrus.Fields{ - "block_root": block.BlockRoot, - "network": block.Network, + KeyBlockRoot: block.BlockRoot, + KeyNetwork: block.Network, }).Debug("Block already processed (cache hit after lock)") return nil @@ -288,13 +288,13 @@ func (p *PermanentStore) processBlock(ctx context.Context, block PermanentStoreB permanentBlock, err := p.db.GetPermanentBlockByBlockRoot(ctx, block.BlockRoot, block.Network) if err != nil { p.log.WithError(err).WithFields(logrus.Fields{ - "block_root": block.BlockRoot, - "network": block.Network, + KeyBlockRoot: block.BlockRoot, + KeyNetwork: block.Network, }).Error("Failed to check if block is already recorded in database") } else if permanentBlock != nil { p.log.WithFields(logrus.Fields{ - "block_root": block.BlockRoot, - "network": block.Network, + KeyBlockRoot: block.BlockRoot, + KeyNetwork: block.Network, }).Debug("Block already recorded in database") // Add to cache to avoid future checks @@ -314,9 +314,9 @@ func (p *PermanentStore) processBlock(ctx context.Context, block PermanentStoreB if exists { p.log.WithFields(logrus.Fields{ - "block_root": block.BlockRoot, - "network": block.Network, - "location": permanentLocation, + KeyBlockRoot: block.BlockRoot, + KeyNetwork: block.Network, + KeyLocation: permanentLocation, }).Debug("Block already exists in permanent location") // Add to cache to avoid future checks @@ -325,9 +325,9 @@ func (p *PermanentStore) processBlock(ctx context.Context, block PermanentStoreB // 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{ - "block_root": block.BlockRoot, - "network": block.Network, - "slot": block.Slot, + KeyBlockRoot: block.BlockRoot, + KeyNetwork: block.Network, + KeySlot: block.Slot, }).Error("Failed to record permanent block in database") } @@ -344,8 +344,8 @@ func (p *PermanentStore) processBlock(ctx context.Context, block PermanentStoreB } p.log.WithFields(logrus.Fields{ - "block_root": block.BlockRoot, - "network": block.Network, + KeyBlockRoot: block.BlockRoot, + KeyNetwork: block.Network, "from": block.Location, "to": permanentLocation, }).Info("Copied block to permanent location") @@ -353,9 +353,9 @@ func (p *PermanentStore) processBlock(ctx context.Context, block PermanentStoreB // Record the block in the database if perr := p.recordPermanentBlock(ctx, block); perr != nil { p.log.WithError(perr).WithFields(logrus.Fields{ - "block_root": block.BlockRoot, - "network": block.Network, - "slot": block.Slot, + KeyBlockRoot: block.BlockRoot, + KeyNetwork: block.Network, + KeySlot: block.Slot, }).Error("Failed to record permanent block in database") } diff --git a/pkg/server/service/indexer/permanent_store_test.go b/pkg/server/service/indexer/permanent_store_test.go index de654fa..4b36e0d 100644 --- a/pkg/server/service/indexer/permanent_store_test.go +++ b/pkg/server/service/indexer/permanent_store_test.go @@ -96,7 +96,7 @@ func TestPermanentStoreQueueAndProcess(t *testing.T) { blockInfo := PermanentStoreBlock{ Location: blockLocation, BlockRoot: "0x1234", - Network: "mainnet", + Network: testNetwork, Slot: 123, ProcessedChan: processChan, } @@ -157,7 +157,7 @@ func TestPermanentStoreProcessSameBlockTwice(t *testing.T) { blockInfo := PermanentStoreBlock{ Location: blockLocation, BlockRoot: "0x1234", - Network: "mainnet", + Network: testNetwork, Slot: 123, ProcessedChan: processChan1, } @@ -195,7 +195,7 @@ func TestPermanentStoreProcessSameBlockTwice(t *testing.T) { blockInfo2 := PermanentStoreBlock{ Location: blockLocation, BlockRoot: "0x1234", - Network: "mainnet", + Network: testNetwork, Slot: 123, ProcessedChan: processChan2, } @@ -251,7 +251,7 @@ func TestPermanentStoreDifferentNetworks(t *testing.T) { blockInfo1 := PermanentStoreBlock{ Location: blockLocation, BlockRoot: "0x1234", - Network: "mainnet", + Network: testNetwork, Slot: 123, ProcessedChan: make(chan struct{}), } @@ -378,7 +378,7 @@ func TestPermanentStoreDistributedLock(t *testing.T) { blockInfo1 := PermanentStoreBlock{ Location: blockLocation, BlockRoot: "0xabcd", - Network: "mainnet", + Network: testNetwork, ProcessedChan: make(chan struct{}), } @@ -419,7 +419,7 @@ func TestPermanentStoreDistributedLock(t *testing.T) { blockInfo2 := PermanentStoreBlock{ Location: blockLocation, BlockRoot: "0xabcd", - Network: "mainnet", + Network: testNetwork, ProcessedChan: make(chan struct{}), } @@ -462,7 +462,7 @@ func TestPermanentStoreStop(t *testing.T) { blockInfo := PermanentStoreBlock{ Location: blockLocation, BlockRoot: "0xstop", - Network: "mainnet", + Network: testNetwork, ProcessedChan: processChan, Slot: 1, } @@ -490,7 +490,7 @@ func TestPermanentStoreStop(t *testing.T) { queuedBlock := PermanentStoreBlock{ Location: blockLocation, BlockRoot: "0xqueued", - Network: "mainnet", + Network: testNetwork, ProcessedChan: processChan2, Slot: 2, } @@ -533,7 +533,7 @@ func TestPermanentStoreStop(t *testing.T) { unprocessedBlock := PermanentStoreBlock{ Location: blockLocation, BlockRoot: "0xunprocessed", - Network: "mainnet", + Network: testNetwork, ProcessedChan: make(chan struct{}), Slot: 3, } @@ -562,7 +562,7 @@ func TestPermanentStoreLocation(t *testing.T) { blockInfo := PermanentStoreBlock{ Location: "test/location/block.ssz", BlockRoot: "0xabcd1234", - Network: "mainnet", + Network: testNetwork, Slot: 123456, } diff --git a/pkg/server/service/indexer/retention.go b/pkg/server/service/indexer/retention.go index a2ac489..67ae175 100644 --- a/pkg/server/service/indexer/retention.go +++ b/pkg/server/service/indexer/retention.go @@ -61,7 +61,7 @@ func (i *Indexer) purgeOldBeaconStates(ctx context.Context) error { Before: &before, } - states, err := i.db.ListBeaconState(ctx, filter, &persistence.PaginationCursor{Limit: 10000, Offset: 0, OrderBy: "fetched_at ASC"}) + states, err := i.db.ListBeaconState(ctx, filter, &persistence.PaginationCursor{Limit: 10000, Offset: 0, OrderBy: OrderFetchedAtAsc}) if err != nil { return err } @@ -89,10 +89,10 @@ func (i *Indexer) purgeOldBeaconStates(ctx context.Context) error { i.log.WithFields( logrus.Fields{ - "node": state.Node, - "network": state.Network, - "slot": state.Slot, - "id": state.ID, + KeyNode: state.Node, + KeyNetwork: state.Network, + KeySlot: state.Slot, + KeyID: state.ID, }, ).Debug("Deleted beacon state") } @@ -107,7 +107,7 @@ func (i *Indexer) purgeOldBeaconBlocks(ctx context.Context) error { Before: &before, } - blocks, err := i.db.ListBeaconBlock(ctx, filter, &persistence.PaginationCursor{Limit: 10000, Offset: 0, OrderBy: "fetched_at ASC"}) + blocks, err := i.db.ListBeaconBlock(ctx, filter, &persistence.PaginationCursor{Limit: 10000, Offset: 0, OrderBy: OrderFetchedAtAsc}) if err != nil { return err } @@ -150,10 +150,10 @@ func (i *Indexer) purgeOldBeaconBlocks(ctx context.Context) error { i.log.WithFields( logrus.Fields{ - "node": block.Node, - "network": block.Network, - "slot": block.Slot, - "id": block.ID, + KeyNode: block.Node, + KeyNetwork: block.Network, + KeySlot: block.Slot, + KeyID: block.ID, }, ).Debug("Deleted beacon block") } @@ -168,7 +168,7 @@ func (i *Indexer) purgeOldBeaconBadBlocks(ctx context.Context) error { Before: &before, } - blocks, err := i.db.ListBeaconBadBlock(ctx, filter, &persistence.PaginationCursor{Limit: 10000, Offset: 0, OrderBy: "fetched_at ASC"}) + blocks, err := i.db.ListBeaconBadBlock(ctx, filter, &persistence.PaginationCursor{Limit: 10000, Offset: 0, OrderBy: OrderFetchedAtAsc}) if err != nil { return err } @@ -196,10 +196,10 @@ func (i *Indexer) purgeOldBeaconBadBlocks(ctx context.Context) error { i.log.WithFields( logrus.Fields{ - "node": block.Node, - "network": block.Network, - "slot": block.Slot, - "id": block.ID, + KeyNode: block.Node, + KeyNetwork: block.Network, + KeySlot: block.Slot, + KeyID: block.ID, }, ).Debug("Deleted beacon bad block") } @@ -214,7 +214,7 @@ func (i *Indexer) purgeOldBeaconBadBlobs(ctx context.Context) error { Before: &before, } - blobs, err := i.db.ListBeaconBadBlob(ctx, filter, &persistence.PaginationCursor{Limit: 10000, Offset: 0, OrderBy: "fetched_at ASC"}) + blobs, err := i.db.ListBeaconBadBlob(ctx, filter, &persistence.PaginationCursor{Limit: 10000, Offset: 0, OrderBy: OrderFetchedAtAsc}) if err != nil { return err } @@ -242,11 +242,11 @@ func (i *Indexer) purgeOldBeaconBadBlobs(ctx context.Context) error { i.log.WithFields( logrus.Fields{ - "node": blob.Node, - "network": blob.Network, - "slot": blob.Slot, - "index": blob.Index, - "id": blob.ID, + KeyNode: blob.Node, + KeyNetwork: blob.Network, + KeySlot: blob.Slot, + KeyIndex: blob.Index, + KeyID: blob.ID, }, ).Debug("Deleted beacon bad blob") } @@ -261,7 +261,7 @@ func (i *Indexer) purgeOldExecutionTraces(ctx context.Context) error { Before: &before, } - traces, err := i.db.ListExecutionBlockTrace(ctx, filter, &persistence.PaginationCursor{Limit: 10000, Offset: 0, OrderBy: "fetched_at ASC"}) + traces, err := i.db.ListExecutionBlockTrace(ctx, filter, &persistence.PaginationCursor{Limit: 10000, Offset: 0, OrderBy: OrderFetchedAtAsc}) if err != nil { return err } @@ -289,10 +289,10 @@ func (i *Indexer) purgeOldExecutionTraces(ctx context.Context) error { i.log.WithFields( logrus.Fields{ - "node": trace.Node, - "network": trace.Network, + KeyNode: trace.Node, + KeyNetwork: trace.Network, "block_number": trace.BlockNumber, - "id": trace.ID, + KeyID: trace.ID, }, ).Debug("Deleted execution block trace") } @@ -307,7 +307,7 @@ func (i *Indexer) purgeOldExecutionBadBlocks(ctx context.Context) error { Before: &before, } - blocks, err := i.db.ListExecutionBadBlock(ctx, filter, &persistence.PaginationCursor{Limit: 10000, Offset: 0, OrderBy: "fetched_at ASC"}) + blocks, err := i.db.ListExecutionBadBlock(ctx, filter, &persistence.PaginationCursor{Limit: 10000, Offset: 0, OrderBy: OrderFetchedAtAsc}) if err != nil { return err } @@ -335,10 +335,10 @@ func (i *Indexer) purgeOldExecutionBadBlocks(ctx context.Context) error { i.log.WithFields( logrus.Fields{ - "node": block.Node, - "network": block.Network, + KeyNode: block.Node, + KeyNetwork: block.Network, "block_hash": block.BlockHash, - "id": block.ID, + KeyID: block.ID, }, ).Debug("Deleted execution block trace") } diff --git a/pkg/server/service/indexer/testdata_test.go b/pkg/server/service/indexer/testdata_test.go new file mode 100644 index 0000000..6e74e01 --- /dev/null +++ b/pkg/server/service/indexer/testdata_test.go @@ -0,0 +1,7 @@ +package indexer + +// Values shared by the indexer tests. +const ( + testDataLocation = "data.json" + testNetwork = "mainnet" +) diff --git a/pkg/store/metrics.go b/pkg/store/metrics.go index c8fd2d0..86f87f7 100644 --- a/pkg/store/metrics.go +++ b/pkg/store/metrics.go @@ -41,43 +41,43 @@ func GetBasicMetricsInstance(namespace, storeType string, enabled bool) *BasicMe Namespace: namespace, Name: "items_added_count", Help: "Number of items added to the store", - }, []string{"type"}), + }, []string{labelType}), itemsRemoved: prometheus.NewCounterVec(prometheus.CounterOpts{ Namespace: namespace, Name: "items_removed_count", Help: "Number of items removed from the store", - }, []string{"type"}), + }, []string{labelType}), itemsRetreived: prometheus.NewCounterVec(prometheus.CounterOpts{ Namespace: namespace, Name: "items_retrieved_count", Help: "Number of items retreived from the store", - }, []string{"type"}), + }, []string{labelType}), itemsStored: prometheus.NewGaugeVec(prometheus.GaugeOpts{ Namespace: namespace, Name: "items_stored_total", Help: "Number of items stored in the store", - }, []string{"type"}), + }, []string{labelType}), itemsUrlsRetreived: prometheus.NewCounterVec(prometheus.CounterOpts{ Namespace: namespace, Name: "items_urls_retrieved_count", Help: "Number of items URLs retreived", - }, []string{"type"}), + }, []string{labelType}), cacheHit: prometheus.NewCounterVec(prometheus.CounterOpts{ Namespace: namespace, Name: "cache_hit_count", Help: "Number of cache hits", - }, []string{"type"}), + }, []string{labelType}), cacheMiss: prometheus.NewCounterVec(prometheus.CounterOpts{ Namespace: namespace, Name: "cache_miss_count", Help: "Number of cache misses", - }, []string{"type"}), + }, []string{labelType}), itemsAddedBytes: prometheus.NewHistogramVec(prometheus.HistogramOpts{ Namespace: namespace, Name: "items_added_bytes", Help: "Size of items added to the store", Buckets: prometheus.ExponentialBuckets(1024000, 2, 13), - }, []string{"type"}), + }, []string{labelType}), } if enabled { diff --git a/pkg/store/mock.go b/pkg/store/mock.go index c51e6e9..93b8649 100644 --- a/pkg/store/mock.go +++ b/pkg/store/mock.go @@ -11,13 +11,17 @@ import ( "github.com/testcontainers/testcontainers-go/wait" ) +// minioTestCredential is the access key and secret the mock minio container is +// started with. Test-only; not a real credential. +const minioTestCredential = "minioadmin" + func setupMinioContainer(ctx context.Context, bucketName string) (testcontainers.Container, string, error) { req := testcontainers.ContainerRequest{ Image: "minio/minio", ExposedPorts: []string{"9000/tcp"}, Env: map[string]string{ - "MINIO_ACCESS_KEY": "minioadmin", - "MINIO_SECRET_KEY": "minioadmin", + "MINIO_ACCESS_KEY": minioTestCredential, + "MINIO_SECRET_KEY": minioTestCredential, }, Cmd: []string{"server", "/data"}, WaitingFor: wait.ForListeningPort("9000/tcp").WithStartupTimeout(2 * time.Minute), @@ -60,8 +64,8 @@ func NewMockS3Store(ctx context.Context, bucket string) (Store, func() error, er store, err := NewS3Store("throwaway", logrus.New(), &S3StoreConfig{ Endpoint: "http://" + endpoint, Region: "us-east-1", - AccessKey: "minioadmin", - AccessSecret: "minioadmin", + AccessKey: minioTestCredential, + AccessSecret: minioTestCredential, BucketName: bucket, }, DefaultOptions().SetMetricsEnabled(false)) if err != nil { diff --git a/pkg/store/type.go b/pkg/store/type.go index eacd373..712fcd7 100644 --- a/pkg/store/type.go +++ b/pkg/store/type.go @@ -2,6 +2,10 @@ package store type Type string +const ( + labelType = "type" +) + const ( UnknownStore Type = "unknown" S3StoreType Type = "s3"