From dc6ef58eb427f2d72b77663e01dbfb269f71ae3a Mon Sep 17 00:00:00 2001 From: anhthii Date: Thu, 8 Oct 2026 19:36:19 +0700 Subject: [PATCH] feat(config): default from_latest to true, honor start_block on cold start from_latest is now a *bool that defaults to true when unset in both defaults and the chain config. On a cold start, a chain with from_latest: false begins at its configured start_block instead of the chain head. Config validation rejects from_latest: false without a start_block. --- configs/config.example.yaml | 2 +- internal/worker/regular.go | 10 ++++++-- internal/worker/regular_test.go | 37 ++++++++++++++++++++++++++++++ pkg/common/config/chains.go | 17 +++++++++++--- pkg/common/config/chains_test.go | 24 +++++++++++++++++++ pkg/common/config/types.go | 4 ++-- pkg/common/config/validate.go | 3 +++ pkg/common/config/validate_test.go | 11 +++++++++ 8 files changed, 100 insertions(+), 8 deletions(-) diff --git a/configs/config.example.yaml b/configs/config.example.yaml index dc219d7..d21bc4e 100644 --- a/configs/config.example.yaml +++ b/configs/config.example.yaml @@ -9,7 +9,7 @@ logging: # Global defaults - applies to all chains unless overridden defaults: enabled: false # chains inherit this unless they override enabled below - from_latest: true # start indexing from the latest block + from_latest: true # on cold start begin at chain head; set false per chain to start from start_block poll_interval: "5s" # how often to poll for new blocks reorg_rollback_window: 20 # number of blocks to roll back on reorg two_way_indexing: false # enable two-way indexing for all chains diff --git a/internal/worker/regular.go b/internal/worker/regular.go index 72243db..6c759fd 100644 --- a/internal/worker/regular.go +++ b/internal/worker/regular.go @@ -200,8 +200,8 @@ func (rw *RegularWorker) commitProgress(lastSuccess uint64, lastSuccessHash stri // determineStartingBlock picks a start block from the last indexed block and the // chain head. GetLatestBlock returns (0, nil) for a cold start, so a non-nil -// kvErr means the store is genuinely down. When there is no prior block to -// resume from, the chain head is the only safe anchor and we wait for it. +// kvErr means the store is genuinely down. On a cold start we begin at the +// configured start_block when from_latest is false, otherwise at the chain head. func (rw *RegularWorker) determineStartingBlock() uint64 { kvLatest, kvErr := rw.blockStore.GetLatestBlock(rw.chain.GetNetworkInternalCode()) @@ -231,6 +231,12 @@ func (rw *RegularWorker) determineStartingBlock() uint64 { if kvErr != nil { rw.logger.Error("Block store unavailable, starting from chain head", "chain", rw.chain.GetName(), "error", kvErr) + return rw.waitForChainHead() + } + if !rw.config.IsFromLatest() && rw.config.StartBlock > 0 { + rw.logger.Info("Cold start from configured start_block", + "chain", rw.chain.GetName(), "start_block", rw.config.StartBlock) + return uint64(rw.config.StartBlock) } return rw.waitForChainHead() } diff --git a/internal/worker/regular_test.go b/internal/worker/regular_test.go index 2b9f89c..70e409a 100644 --- a/internal/worker/regular_test.go +++ b/internal/worker/regular_test.go @@ -247,6 +247,43 @@ func TestRegularWorkerDetermineStartingBlockColdStartChainUp(t *testing.T) { require.Equal(t, uint64(500), rw.determineStartingBlock()) } +func TestRegularWorkerDetermineStartingBlockColdStartFromConfiguredStartBlock(t *testing.T) { + t.Parallel() + + chain := &stubIndexer{name: "ethereum", internalCode: "ETH", networkType: enum.NetworkTypeEVM, latest: 500} + store := &stubBlockStore{latestBlock: 0} + rw := newTestRegularWorker(chain, store, 0, 2) + fromLatest := false + rw.config.FromLatest = &fromLatest + rw.config.StartBlock = 120 + + require.Equal(t, uint64(120), rw.determineStartingBlock()) +} + +func TestRegularWorkerDetermineStartingBlockColdStartFromLatestIgnoresStartBlock(t *testing.T) { + t.Parallel() + + chain := &stubIndexer{name: "ethereum", internalCode: "ETH", networkType: enum.NetworkTypeEVM, latest: 500} + store := &stubBlockStore{latestBlock: 0} + rw := newTestRegularWorker(chain, store, 0, 2) + rw.config.StartBlock = 120 + + require.Equal(t, uint64(500), rw.determineStartingBlock()) +} + +func TestRegularWorkerDetermineStartingBlockResumeIgnoresStartBlock(t *testing.T) { + t.Parallel() + + chain := &stubIndexer{name: "ethereum", internalCode: "ETH", networkType: enum.NetworkTypeEVM, latest: 300} + store := &stubBlockStore{latestBlock: 300} + rw := newTestRegularWorker(chain, store, 0, 2) + fromLatest := false + rw.config.FromLatest = &fromLatest + rw.config.StartBlock = 120 + + require.Equal(t, uint64(300), rw.determineStartingBlock()) +} + func TestRegularWorkerDetermineStartingBlockUsesConfirmedHead(t *testing.T) { t.Parallel() diff --git a/pkg/common/config/chains.go b/pkg/common/config/chains.go index ce81537..c8ef455 100644 --- a/pkg/common/config/chains.go +++ b/pkg/common/config/chains.go @@ -56,7 +56,8 @@ func (c Chains) Validate(names []string) error { func (c Chains) OverrideFromLatest(names []string) { for _, name := range names { if chain, ok := c[name]; ok { - chain.FromLatest = true + fromLatest := true + chain.FromLatest = &fromLatest c[name] = chain } } @@ -75,8 +76,9 @@ func (c Chains) ApplyDefaults(def Defaults) error { enabled := *def.Enabled chain.Enabled = &enabled } - if !chain.FromLatest { - chain.FromLatest = def.FromLatest + if chain.FromLatest == nil && def.FromLatest != nil { + fromLatest := *def.FromLatest + chain.FromLatest = &fromLatest } if !chain.TwoWayIndexing { chain.TwoWayIndexing = def.TwoWayIndexing @@ -101,3 +103,12 @@ func (c Chains) ApplyDefaults(def Defaults) error { } return nil } + +// IsFromLatest reports whether a cold start begins at the chain head. Defaults to true when unset. +func (c ChainConfig) IsFromLatest() bool { + return boolOrTrue(c.FromLatest) +} + +func boolOrTrue(b *bool) bool { + return b == nil || *b +} diff --git a/pkg/common/config/chains_test.go b/pkg/common/config/chains_test.go index 0e65cee..0024e78 100644 --- a/pkg/common/config/chains_test.go +++ b/pkg/common/config/chains_test.go @@ -124,3 +124,27 @@ func TestEnabledNames_IncludesExplicitlyEnabledChains(t *testing.T) { require.ElementsMatch(t, []string{"ethereum_mainnet"}, chains.EnabledNames()) } + +func TestApplyDefaults_FromLatest(t *testing.T) { + t.Parallel() + + f, tr := false, true + tests := []struct { + name string + def *bool + chain *bool + expected bool + }{ + {"unset everywhere defaults to true", nil, nil, true}, + {"inherits default false", &f, nil, false}, + {"chain overrides default", &tr, &f, false}, + {"chain true overrides default false", &f, &tr, true}, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + chains := Chains{"eth": {FromLatest: tt.chain}} + require.NoError(t, chains.ApplyDefaults(Defaults{FromLatest: tt.def})) + require.Equal(t, tt.expected, chains["eth"].IsFromLatest()) + }) + } +} diff --git a/pkg/common/config/types.go b/pkg/common/config/types.go index 97761d9..c15032f 100644 --- a/pkg/common/config/types.go +++ b/pkg/common/config/types.go @@ -52,7 +52,7 @@ func (c *LoggingConfig) Normalize() { type Defaults struct { Enabled *bool `yaml:"enabled"` - FromLatest bool `yaml:"from_latest"` + FromLatest *bool `yaml:"from_latest"` TwoWayIndexing bool `yaml:"two_way_indexing"` PollInterval time.Duration `yaml:"poll_interval" validate:"required"` ReorgRollbackWindow int `yaml:"reorg_rollback_window" validate:"required,min=1"` @@ -71,7 +71,7 @@ type ChainConfig struct { InternalCode string `yaml:"internal_code"` NativeDenom string `yaml:"native_denom"` Type enum.NetworkType `yaml:"type" validate:"required"` - FromLatest bool `yaml:"from_latest"` + FromLatest *bool `yaml:"from_latest"` StartBlock int `yaml:"start_block" validate:"min=0"` PollInterval time.Duration `yaml:"poll_interval"` ReorgRollbackWindow int `yaml:"reorg_rollback_window"` diff --git a/pkg/common/config/validate.go b/pkg/common/config/validate.go index e00b080..83d7113 100644 --- a/pkg/common/config/validate.go +++ b/pkg/common/config/validate.go @@ -10,6 +10,9 @@ func validateChainConfig(chain ChainConfig) error { if chain.Type == enum.NetworkTypeCosmos && chain.NativeDenom == "" { return fmt.Errorf("native_denom is required for cosmos chains") } + if !chain.IsFromLatest() && chain.StartBlock == 0 { + return fmt.Errorf("start_block is required when from_latest is false") + } statusCfg := chain.Status.Normalize() if statusCfg.HealthyMaxPendingBlocks >= statusCfg.SlowMaxPendingBlocks { return fmt.Errorf( diff --git a/pkg/common/config/validate_test.go b/pkg/common/config/validate_test.go index f6e4eb0..3afdd40 100644 --- a/pkg/common/config/validate_test.go +++ b/pkg/common/config/validate_test.go @@ -72,3 +72,14 @@ func TestLoggingConfigValidate_RejectsUnknownModeAndFormat(t *testing.T) { invalid := LoggingConfig{Mode: "somewhere", Format: "plain"} require.Error(t, validate.Struct(&invalid)) } + +func TestValidateChainConfig_RequiresStartBlockWhenNotFromLatest(t *testing.T) { + t.Parallel() + + fromLatest := false + chain := ChainConfig{Type: enum.NetworkTypeEVM, FromLatest: &fromLatest} + require.Error(t, validateChainConfig(chain)) + + chain.StartBlock = 100 + require.NoError(t, validateChainConfig(chain)) +}