diff --git a/app/seidb.go b/app/seidb.go index 9307af2bb0..b00d389416 100644 --- a/app/seidb.go +++ b/app/seidb.go @@ -53,6 +53,16 @@ const ( FlagEVMSSDirectory = "state-store.evm-ss-db-directory" FlagEVMSSSplit = "state-store.evm-ss-split" FlagEVMSSSeparateDBs = "state-store.evm-ss-separate-dbs" + + // Historical SS offload fallback. + FlagHistoricalOffloadBackend = "state-store.historical-offload-backend" + FlagHistoricalOffloadBigtableProjectID = "state-store.historical-offload-bigtable-project-id" + FlagHistoricalOffloadBigtableInstance = "state-store.historical-offload-bigtable-instance" + FlagHistoricalOffloadBigtableTable = "state-store.historical-offload-bigtable-table" + FlagHistoricalOffloadBigtableFamily = "state-store.historical-offload-bigtable-family" + FlagHistoricalOffloadBigtableAppProfile = "state-store.historical-offload-bigtable-app-profile" + FlagHistoricalOffloadBigtableShards = "state-store.historical-offload-bigtable-shards" + FlagHistoricalOffloadEarliestVersion = "state-store.historical-offload-earliest-version" ) var GigaKeys = []string{"evm", "bank"} @@ -208,5 +218,13 @@ func parseSSConfigs(appOpts servertypes.AppOptions) config.StateStoreConfig { ssConfig.EVMDBDirectory = cast.ToString(appOpts.Get(FlagEVMSSDirectory)) ssConfig.SeparateEVMSubDBs = cast.ToBool(appOpts.Get(FlagEVMSSSeparateDBs)) ssConfig.EVMSplit = cast.ToBool(appOpts.Get(FlagEVMSSSplit)) + ssConfig.HistoricalOffloadBackend = cast.ToString(appOpts.Get(FlagHistoricalOffloadBackend)) + ssConfig.HistoricalOffloadBigtableProjectID = cast.ToString(appOpts.Get(FlagHistoricalOffloadBigtableProjectID)) + ssConfig.HistoricalOffloadBigtableInstance = cast.ToString(appOpts.Get(FlagHistoricalOffloadBigtableInstance)) + ssConfig.HistoricalOffloadBigtableTable = cast.ToString(appOpts.Get(FlagHistoricalOffloadBigtableTable)) + ssConfig.HistoricalOffloadBigtableFamily = cast.ToString(appOpts.Get(FlagHistoricalOffloadBigtableFamily)) + ssConfig.HistoricalOffloadBigtableAppProfile = cast.ToString(appOpts.Get(FlagHistoricalOffloadBigtableAppProfile)) + ssConfig.HistoricalOffloadBigtableShards = cast.ToInt(appOpts.Get(FlagHistoricalOffloadBigtableShards)) + ssConfig.HistoricalOffloadEarliestVersion = cast.ToInt64(appOpts.Get(FlagHistoricalOffloadEarliestVersion)) return ssConfig } diff --git a/app/seidb_test.go b/app/seidb_test.go index f089826f2b..b9a3f5c010 100644 --- a/app/seidb_test.go +++ b/app/seidb_test.go @@ -65,6 +65,22 @@ func (t TestSeiDBAppOpts) Get(s string) interface{} { return defaultSSConfig.EVMSplit case FlagEVMSSSeparateDBs: return defaultSSConfig.SeparateEVMSubDBs + case FlagHistoricalOffloadBackend: + return defaultSSConfig.HistoricalOffloadBackend + case FlagHistoricalOffloadBigtableProjectID: + return defaultSSConfig.HistoricalOffloadBigtableProjectID + case FlagHistoricalOffloadBigtableInstance: + return defaultSSConfig.HistoricalOffloadBigtableInstance + case FlagHistoricalOffloadBigtableTable: + return defaultSSConfig.HistoricalOffloadBigtableTable + case FlagHistoricalOffloadBigtableFamily: + return defaultSSConfig.HistoricalOffloadBigtableFamily + case FlagHistoricalOffloadBigtableAppProfile: + return defaultSSConfig.HistoricalOffloadBigtableAppProfile + case FlagHistoricalOffloadBigtableShards: + return defaultSSConfig.HistoricalOffloadBigtableShards + case FlagHistoricalOffloadEarliestVersion: + return defaultSSConfig.HistoricalOffloadEarliestVersion } return nil } @@ -204,6 +220,32 @@ func TestParseSSConfigs_EVMFlags(t *testing.T) { assert.True(t, ssConfig.SeparateEVMSubDBs) } +func TestParseSSConfigs_HistoricalBigtableFlags(t *testing.T) { + appOpts := mapAppOpts{ + FlagSSEnable: true, + FlagHistoricalOffloadBackend: "bigtable", + FlagHistoricalOffloadBigtableProjectID: "sei-project", + FlagHistoricalOffloadBigtableInstance: "sei-history", + FlagHistoricalOffloadBigtableTable: "state_mutations", + FlagHistoricalOffloadBigtableFamily: "state", + FlagHistoricalOffloadBigtableAppProfile: "historical", + FlagHistoricalOffloadBigtableShards: 512, + FlagHistoricalOffloadEarliestVersion: int64(1234), + FlagSSAsyncWriterBuffer: 0, + } + + ssConfig := parseSSConfigs(appOpts) + assert.True(t, ssConfig.Enable) + assert.Equal(t, "bigtable", ssConfig.HistoricalOffloadBackend) + assert.Equal(t, "sei-project", ssConfig.HistoricalOffloadBigtableProjectID) + assert.Equal(t, "sei-history", ssConfig.HistoricalOffloadBigtableInstance) + assert.Equal(t, "state_mutations", ssConfig.HistoricalOffloadBigtableTable) + assert.Equal(t, "state", ssConfig.HistoricalOffloadBigtableFamily) + assert.Equal(t, "historical", ssConfig.HistoricalOffloadBigtableAppProfile) + assert.Equal(t, 512, ssConfig.HistoricalOffloadBigtableShards) + assert.Equal(t, int64(1234), ssConfig.HistoricalOffloadEarliestVersion) +} + func TestParseSSConfigs_ReadWriteMetrics(t *testing.T) { ssConfig := parseSSConfigs(mapAppOpts{ FlagSSEnable: true, diff --git a/sei-db/config/ss_config.go b/sei-db/config/ss_config.go index a3a89b898e..820406e2f1 100644 --- a/sei-db/config/ss_config.go +++ b/sei-db/config/ss_config.go @@ -80,6 +80,26 @@ type StateStoreConfig struct { // When true, data is routed to separate DBs by EVM key family while // preserving the same logical store key and full key encoding inside each DB. SeparateEVMSubDBs bool `mapstructure:"evm-separate-dbs"` + + // HistoricalOffloadBackend selects the historical offload read fallback. + // Empty (the default) disables the fallback entirely; "bigtable" enables + // the Bigtable reader configured by the connection fields below. + HistoricalOffloadBackend string `mapstructure:"historical-offload-backend"` + + // HistoricalOffloadBigtable* are the Bigtable connection parameters used + // when HistoricalOffloadBackend is "bigtable". + HistoricalOffloadBigtableProjectID string `mapstructure:"historical-offload-bigtable-project-id"` + HistoricalOffloadBigtableInstance string `mapstructure:"historical-offload-bigtable-instance"` + HistoricalOffloadBigtableTable string `mapstructure:"historical-offload-bigtable-table"` + HistoricalOffloadBigtableFamily string `mapstructure:"historical-offload-bigtable-family"` + HistoricalOffloadBigtableAppProfile string `mapstructure:"historical-offload-bigtable-app-profile"` + HistoricalOffloadBigtableShards int `mapstructure:"historical-offload-bigtable-shards"` + + // HistoricalOffloadEarliestVersion is the operator-declared earliest version + // (inclusive) fully ingested into the historical backend. When > 0 it is + // advertised as the store's earliest version so height gates admit pruned + // reads; reads below it stay on the primary. + HistoricalOffloadEarliestVersion int64 `mapstructure:"historical-offload-earliest-version"` } // DefaultStateStoreConfig returns the default StateStoreConfig diff --git a/sei-db/config/toml_test.go b/sei-db/config/toml_test.go index deee57a2ea..81522f947a 100644 --- a/sei-db/config/toml_test.go +++ b/sei-db/config/toml_test.go @@ -99,6 +99,10 @@ func TestStateStoreConfigTemplate(t *testing.T) { require.Contains(t, output, `evm-ss-db-directory = ""`, "Missing evm-ss-db-directory") require.Contains(t, output, `evm-ss-split = false`, "Missing or incorrect evm-ss-split") require.Contains(t, output, "evm-ss-separate-dbs = false", "Missing or incorrect evm-ss-separate-dbs") + // The historical-offload keys are deliberately absent from the generated + // default config (internal/Giga-only for now); they still parse when an + // operator adds them to app.toml manually. + require.NotContains(t, output, "historical-offload", "historical-offload keys must not be in the default template") } // TestReceiptStoreConfigTemplate verifies that all field paths in the receipt-store TOML template diff --git a/sei-db/state_db/ss/composite/store.go b/sei-db/state_db/ss/composite/store.go index a4077cca09..062157e4cc 100644 --- a/sei-db/state_db/ss/composite/store.go +++ b/sei-db/state_db/ss/composite/store.go @@ -219,6 +219,17 @@ func (s *CompositeStateStore) GetEarliestVersion() int64 { return s.cosmosStore.GetEarliestVersion() } +// GetEarliestVersionForKey reports the earliest version still held locally by +// the store that serves storeKey. Cosmos and EVM prune independently (an EVM +// prune failure is only logged), so pruned-read fallback must consult the +// horizon of the store that will actually serve the read. +func (s *CompositeStateStore) GetEarliestVersionForKey(storeKey string) int64 { + if s.evmRouted(storeKey) { + return s.evmStore.GetEarliestVersion() + } + return s.cosmosStore.GetEarliestVersion() +} + func (s *CompositeStateStore) Close() error { s.closeOnce.Do(func() { if s.pruningManager != nil { diff --git a/sei-db/state_db/ss/offload/consumer/README.md b/sei-db/state_db/ss/offload/consumer/README.md index 7e2ec5cef1..5a3435ed60 100644 --- a/sei-db/state_db/ss/offload/consumer/README.md +++ b/sei-db/state_db/ss/offload/consumer/README.md @@ -47,10 +47,55 @@ SASL/PLAIN using service-account credentials: } ``` +## Node Read Fallback + +Enable fallback reads in the node config: + +```toml +[state-store] +historical-offload-backend = "bigtable" +historical-offload-bigtable-project-id = "my-gcp-project" +historical-offload-bigtable-instance = "sei-history" +historical-offload-bigtable-table = "state_mutations" +historical-offload-bigtable-family = "state" +historical-offload-bigtable-app-profile = "" +historical-offload-bigtable-shards = 256 +``` + +These keys are deliberately not part of the generated default config +(internal/Giga-only for now) and must be added to `app.toml` manually. + +Fallback activates only for point reads where the requested version is below the +local SS earliest version. Missing rows and tombstones return empty state, same +as local SS. Reads ahead of the backend's last ingested version (consumer lag) +return an error rather than empty state. + +To open RPC height gates for pruned heights, declare how far back the backend +has full coverage (the version at which ingestion/backfill started): + +```toml +[state-store] +historical-offload-earliest-version = 123456789 +``` + +When set (> 0), the node advertises this as its earliest version so height +checks admit heights the fallback can serve; point reads below it stay on the +local store. Leave it 0 until the backend actually covers the target range — +heights the backend never ingested would otherwise read as empty state. + +## Operational preconditions + +Before enabling the node read fallback in production: + +- The Bigtable table exists with the same family/shards the consumer wrote with. +- The consumer has been ingesting continuously; check + `consumer_kafka_lag` / `bigtable_rows_mutated_total`. +- `historical-offload-earliest-version` is set to the true coverage floor. + ## Current Limits -- The node-side read fallback lands in part 2; this part is the client library - and the ingestion pipeline. +- No offload iterator path; range queries between the coverage floor and the + local prune horizon see only local data. - No cross-row transaction on ingest; mutation rows are written first and the version marker is written last, so replay is idempotent after partial failure. - No automatic table creation from the binary. diff --git a/sei-db/state_db/ss/offload/historical/metrics.go b/sei-db/state_db/ss/offload/historical/metrics.go new file mode 100644 index 0000000000..273cb68c85 --- /dev/null +++ b/sei-db/state_db/ss/offload/historical/metrics.go @@ -0,0 +1,49 @@ +package historical + +import ( + "context" + + "go.opentelemetry.io/otel" + "go.opentelemetry.io/otel/attribute" + "go.opentelemetry.io/otel/metric" +) + +// fallbackMeterName is the OpenTelemetry meter for the node-side pruned-read +// fallback path, exported through the same process-wide MeterProvider. +const fallbackMeterName = "seidb_historical_fallback" + +const ( + fallbackOutcomeCacheHit = "cache_hit" + fallbackOutcomeBackendHit = "backend_hit" + fallbackOutcomeBackendMiss = "backend_miss" + fallbackOutcomeBackendBehind = "backend_behind" + fallbackOutcomeError = "error" +) + +// fallbackMetrics counts pruned point reads by operation (get/has) and outcome +// so operators can see how much load the read cache absorbs and how many reads +// actually reach the historical backend. Attributes are a closed set, so +// cardinality stays flat. +type fallbackMetrics struct { + reads metric.Int64Counter +} + +func newFallbackMetrics() *fallbackMetrics { + meter := otel.Meter(fallbackMeterName) + reads, _ := meter.Int64Counter( + "historical_fallback_reads_total", + metric.WithDescription("Pruned point reads routed to the historical fallback, by operation and outcome"), + metric.WithUnit("{read}"), + ) + return &fallbackMetrics{reads: reads} +} + +func (m *fallbackMetrics) recordRead(op, outcome string) { + if m == nil { + return + } + m.reads.Add(context.Background(), 1, metric.WithAttributes( + attribute.String("op", op), + attribute.String("outcome", outcome), + )) +} diff --git a/sei-db/state_db/ss/offload/historical/metrics_test.go b/sei-db/state_db/ss/offload/historical/metrics_test.go new file mode 100644 index 0000000000..d7786e82ea --- /dev/null +++ b/sei-db/state_db/ss/offload/historical/metrics_test.go @@ -0,0 +1,21 @@ +package historical + +import "testing" + +func TestFallbackMetricsNilSafe(t *testing.T) { + var m *fallbackMetrics + m.recordRead("get", fallbackOutcomeCacheHit) +} + +func TestFallbackMetricsRecordNoPanic(t *testing.T) { + m := newFallbackMetrics() + for _, outcome := range []string{ + fallbackOutcomeCacheHit, + fallbackOutcomeBackendHit, + fallbackOutcomeBackendMiss, + fallbackOutcomeError, + } { + m.recordRead("get", outcome) + m.recordRead("has", outcome) + } +} diff --git a/sei-db/state_db/ss/offload/historical/store.go b/sei-db/state_db/ss/offload/historical/store.go new file mode 100644 index 0000000000..b74452bd37 --- /dev/null +++ b/sei-db/state_db/ss/offload/historical/store.go @@ -0,0 +1,332 @@ +package historical + +import ( + "bytes" + "context" + "errors" + "fmt" + "sync" + "sync/atomic" + "time" + + lru "github.com/hashicorp/golang-lru/v2" + "github.com/sei-protocol/sei-chain/sei-db/db_engine/types" + "github.com/sei-protocol/sei-chain/sei-db/proto" + dbm "github.com/tendermint/tm-db" +) + +const ( + // Cache entries and the per-value cap bound worst-case memory at + // entries*cap (256 MiB) even when an external caller crafts distinct + // large-value historical queries. + defaultReadCacheEntries = 32 * 1024 + maxReadCacheValueBytes = 8 * 1024 + + // readTimeout bounds one backend point read. types.StateStore has + // no context parameter, so the deadline must be injected here; without it a + // silently dropped connection can park an RPC goroutine in a backend read + // until the OS TCP timeout. + readTimeout = 10 * time.Second + + // missCacheTTL bounds how long a backend miss is trusted. Hits are + // immutable (a version's value never changes once written) but a miss can + // flip to a hit once the offload consumer catches up. + missCacheTTL = time.Minute + + // backendVersionRecheckInterval rate-limits LastVersion refreshes when a + // requested version is ahead of the backend's cached ingestion watermark. + backendVersionRecheckInterval = 30 * time.Second +) + +type readCacheKey struct { + storeKey string + version int64 + key string +} + +type readCacheValue struct { + value []byte + found bool + valueKnown bool + // missExpiresAt is set only on miss entries; found entries never expire. + missExpiresAt time.Time +} + +// PerKeyEarliestVersioner is implemented by primaries whose stores prune +// independently per store key (e.g. the composite store's cosmos and EVM +// backends), so the fallback horizon can be checked against the store that +// will actually serve the read. +type PerKeyEarliestVersioner interface { + GetEarliestVersionForKey(storeKey string) int64 +} + +// FallbackStateStore routes pruned point reads to the historical reader. +// Iteration and writes stay on the primary state store. +type FallbackStateStore struct { + primary types.StateStore + perKeyEarliest PerKeyEarliestVersioner + reader Reader + cache *lru.Cache[readCacheKey, readCacheValue] + metrics *fallbackMetrics + coverageFloor int64 + + // backendLast caches the backend's last ingested version; it is refreshed + // at most every backendVersionRecheckInterval when a read runs ahead of it. + backendLast atomic.Int64 + backendCheckedAt atomic.Int64 + backendMu sync.Mutex +} + +var _ types.StateStore = (*FallbackStateStore)(nil) + +// NewFallbackStateStore takes ownership of primary and reader for Close. +// coverageFloor is the operator-declared earliest version (inclusive) fully +// ingested into the backend: when > 0 it becomes the store's advertised +// earliest version, so height gates such as the EVM RPC watermark admit +// pruned heights the fallback can serve, while reads below it stay on the +// primary. When zero, the advertised earliest remains the local prune horizon. +func NewFallbackStateStore(primary types.StateStore, reader Reader, coverageFloor int64) *FallbackStateStore { + cache, err := lru.New[readCacheKey, readCacheValue](defaultReadCacheEntries) + if err != nil { + panic(err) + } + perKeyEarliest, _ := primary.(PerKeyEarliestVersioner) + return &FallbackStateStore{ + primary: primary, + perKeyEarliest: perKeyEarliest, + reader: reader, + cache: cache, + metrics: newFallbackMetrics(), + coverageFloor: coverageFloor, + } +} + +func (s *FallbackStateStore) shouldFallback(storeKey string, version int64) bool { + if version < 0 || (s.coverageFloor > 0 && version < s.coverageFloor) { + return false + } + earliest := s.earliestVersionFor(storeKey) + return earliest > 0 && version < earliest +} + +// errBackendBehind marks reads refused because the backend has not ingested +// the requested version yet, distinguishing lag from backend failures. +var errBackendBehind = errors.New("historical backend behind requested version") + +// ensureBackendCoverage refuses fallback reads above the backend's last +// ingested version so consumer lag surfaces as an error instead of silently +// empty state (which would also poison the miss cache). +func (s *FallbackStateStore) ensureBackendCoverage(ctx context.Context, version int64) error { + if version <= s.backendLast.Load() { + return nil + } + s.backendMu.Lock() + defer s.backendMu.Unlock() + last := s.backendLast.Load() + if version <= last { + return nil + } + if time.Now().UnixNano()-s.backendCheckedAt.Load() >= backendVersionRecheckInterval.Nanoseconds() { + refreshed, err := s.reader.LastVersion(ctx) + if err != nil { + return fmt.Errorf("check historical backend coverage: %w", err) + } + s.backendCheckedAt.Store(time.Now().UnixNano()) + if refreshed > last { + last = refreshed + s.backendLast.Store(refreshed) + } + } + if version > last { + return fmt.Errorf("%w: ingested up to version %d, requested %d", errBackendBehind, last, version) + } + return nil +} + +func coverageOutcome(err error) string { + if errors.Is(err, errBackendBehind) { + return fallbackOutcomeBackendBehind + } + return fallbackOutcomeError +} + +func (s *FallbackStateStore) earliestVersionFor(storeKey string) int64 { + if s.perKeyEarliest != nil { + return s.perKeyEarliest.GetEarliestVersionForKey(storeKey) + } + return s.primary.GetEarliestVersion() +} + +func (s *FallbackStateStore) readContext() (context.Context, context.CancelFunc) { + return context.WithTimeout(context.Background(), readTimeout) +} + +func (s *FallbackStateStore) Get(storeKey string, version int64, key []byte) ([]byte, error) { + if !s.shouldFallback(storeKey, version) { + return s.primary.Get(storeKey, version, key) + } + cacheKey := readCacheKey{storeKey: storeKey, version: version, key: string(key)} + if value, found, ok := s.getCachedValue(cacheKey); ok { + s.metrics.recordRead("get", fallbackOutcomeCacheHit) + if !found { + return nil, nil + } + return value, nil + } + ctx, cancel := s.readContext() + defer cancel() + if err := s.ensureBackendCoverage(ctx, version); err != nil { + s.metrics.recordRead("get", coverageOutcome(err)) + return nil, err + } + v, err := s.reader.Get(ctx, storeKey, key, version) + if err != nil { + if errors.Is(err, ErrNotFound) { + s.metrics.recordRead("get", fallbackOutcomeBackendMiss) + s.cacheMiss(cacheKey) + return nil, nil + } + s.metrics.recordRead("get", fallbackOutcomeError) + return nil, err + } + s.metrics.recordRead("get", fallbackOutcomeBackendHit) + s.cacheValue(cacheKey, v.Bytes) + return v.Bytes, nil +} + +func (s *FallbackStateStore) getCachedValue(key readCacheKey) ([]byte, bool, bool) { + value, ok := s.cache.Get(key) + if !ok { + return nil, false, false + } + if !value.found { + if s.missExpired(key, value) { + return nil, false, false + } + return nil, false, true + } + if !value.valueKnown { + return nil, false, false + } + return bytes.Clone(value.value), true, true +} + +func (s *FallbackStateStore) getCachedHas(key readCacheKey) (bool, bool) { + value, ok := s.cache.Get(key) + if !ok { + return false, false + } + if !value.found && s.missExpired(key, value) { + return false, false + } + return value.found, true +} + +// missExpired evicts and reports miss entries whose TTL has lapsed so a +// backend that has since ingested the version gets re-queried. +func (s *FallbackStateStore) missExpired(key readCacheKey, value readCacheValue) bool { + if value.missExpiresAt.IsZero() || time.Now().Before(value.missExpiresAt) { + return false + } + s.cache.Remove(key) + return true +} + +func (s *FallbackStateStore) cacheValue(key readCacheKey, value []byte) { + if value == nil || len(value) > maxReadCacheValueBytes { + return + } + s.cache.Add(key, readCacheValue{value: bytes.Clone(value), found: true, valueKnown: true}) +} + +func (s *FallbackStateStore) cacheMiss(key readCacheKey) { + s.cache.Add(key, readCacheValue{valueKnown: true, missExpiresAt: time.Now().Add(missCacheTTL)}) +} + +func (s *FallbackStateStore) cacheHas(key readCacheKey) { + s.cache.Add(key, readCacheValue{found: true}) +} + +func (s *FallbackStateStore) Has(storeKey string, version int64, key []byte) (bool, error) { + if !s.shouldFallback(storeKey, version) { + return s.primary.Has(storeKey, version, key) + } + cacheKey := readCacheKey{storeKey: storeKey, version: version, key: string(key)} + if found, ok := s.getCachedHas(cacheKey); ok { + s.metrics.recordRead("has", fallbackOutcomeCacheHit) + return found, nil + } + ctx, cancel := s.readContext() + defer cancel() + if err := s.ensureBackendCoverage(ctx, version); err != nil { + s.metrics.recordRead("has", coverageOutcome(err)) + return false, err + } + found, err := s.reader.Has(ctx, storeKey, key, version) + if err != nil { + s.metrics.recordRead("has", fallbackOutcomeError) + return false, err + } + if found { + s.metrics.recordRead("has", fallbackOutcomeBackendHit) + s.cacheHas(cacheKey) + } else { + s.metrics.recordRead("has", fallbackOutcomeBackendMiss) + s.cacheMiss(cacheKey) + } + return found, nil +} + +func (s *FallbackStateStore) Iterator(storeKey string, version int64, start, end []byte) (dbm.Iterator, error) { + return s.primary.Iterator(storeKey, version, start, end) +} + +func (s *FallbackStateStore) ReverseIterator(storeKey string, version int64, start, end []byte) (dbm.Iterator, error) { + return s.primary.ReverseIterator(storeKey, version, start, end) +} + +func (s *FallbackStateStore) RawIterate(storeKey string, fn func([]byte, []byte, int64) bool) (bool, error) { + return s.primary.RawIterate(storeKey, fn) +} + +func (s *FallbackStateStore) GetLatestVersion() int64 { return s.primary.GetLatestVersion() } + +func (s *FallbackStateStore) SetLatestVersion(version int64) error { + return s.primary.SetLatestVersion(version) +} + +// GetEarliestVersion advertises the earliest queryable version. With a +// configured coverage floor the fallback serves point reads below the local +// prune horizon, so height gates (e.g. the EVM RPC watermark) must see the +// floor rather than the local horizon or every historical query is rejected +// before it reaches the store. Iterators do not use the fallback; range +// queries between the floor and the local horizon see only local data. +func (s *FallbackStateStore) GetEarliestVersion() int64 { + local := s.primary.GetEarliestVersion() + if s.coverageFloor > 0 && local > 0 && s.coverageFloor < local { + return s.coverageFloor + } + return local +} + +func (s *FallbackStateStore) SetEarliestVersion(version int64, ignoreVersion bool) error { + return s.primary.SetEarliestVersion(version, ignoreVersion) +} + +func (s *FallbackStateStore) ApplyChangesetSync(version int64, changesets []*proto.NamedChangeSet) error { + return s.primary.ApplyChangesetSync(version, changesets) +} + +func (s *FallbackStateStore) ApplyChangesetAsync(version int64, changesets []*proto.NamedChangeSet) error { + return s.primary.ApplyChangesetAsync(version, changesets) +} + +func (s *FallbackStateStore) Prune(version int64) error { return s.primary.Prune(version) } + +func (s *FallbackStateStore) Import(version int64, ch <-chan types.SnapshotNode) error { + return s.primary.Import(version, ch) +} + +func (s *FallbackStateStore) Close() error { + return errors.Join(s.primary.Close(), s.reader.Close()) +} diff --git a/sei-db/state_db/ss/offload/historical/store_test.go b/sei-db/state_db/ss/offload/historical/store_test.go new file mode 100644 index 0000000000..d57f7f0b28 --- /dev/null +++ b/sei-db/state_db/ss/offload/historical/store_test.go @@ -0,0 +1,326 @@ +package historical + +import ( + "context" + "errors" + "testing" + "time" + + "github.com/sei-protocol/sei-chain/sei-db/db_engine/types" + "github.com/sei-protocol/sei-chain/sei-db/proto" + "github.com/stretchr/testify/require" + dbm "github.com/tendermint/tm-db" +) + +type fakeStateStore struct { + earliest int64 + gets int + has int +} + +func (f *fakeStateStore) Get(_ string, _ int64, _ []byte) ([]byte, error) { + f.gets++ + return []byte("primary"), nil +} + +func (f *fakeStateStore) Has(_ string, _ int64, _ []byte) (bool, error) { + f.has++ + return true, nil +} + +func (f *fakeStateStore) Iterator(string, int64, []byte, []byte) (dbm.Iterator, error) { + return nil, nil +} + +func (f *fakeStateStore) ReverseIterator(string, int64, []byte, []byte) (dbm.Iterator, error) { + return nil, nil +} + +func (f *fakeStateStore) RawIterate(string, func([]byte, []byte, int64) bool) (bool, error) { + return false, nil +} + +func (f *fakeStateStore) GetLatestVersion() int64 { return 0 } +func (f *fakeStateStore) SetLatestVersion(int64) error { return nil } +func (f *fakeStateStore) GetEarliestVersion() int64 { return f.earliest } +func (f *fakeStateStore) SetEarliestVersion(version int64, _ bool) error { + f.earliest = version + return nil +} +func (f *fakeStateStore) ApplyChangesetSync(int64, []*proto.NamedChangeSet) error { return nil } +func (f *fakeStateStore) ApplyChangesetAsync(int64, []*proto.NamedChangeSet) error { return nil } +func (f *fakeStateStore) Prune(int64) error { return nil } +func (f *fakeStateStore) Import(int64, <-chan types.SnapshotNode) error { return nil } +func (f *fakeStateStore) Close() error { return nil } + +type fakeReader struct { + gets int + has int + getErr error + hasResult bool + hasSet bool + // lastVersion overrides the reported ingestion watermark; zero means + // "fully caught up" so coverage gating stays out of unrelated tests. + lastVersion int64 + lastVersions int +} + +func (f *fakeReader) Get(context.Context, string, []byte, int64) (Value, error) { + f.gets++ + if f.getErr != nil { + return Value{}, f.getErr + } + return Value{Bytes: []byte("historical"), Version: 7}, nil +} + +func (f *fakeReader) Has(context.Context, string, []byte, int64) (bool, error) { + f.has++ + if f.hasSet { + return f.hasResult, nil + } + return true, nil +} + +func (f *fakeReader) LastVersion(context.Context) (int64, error) { + f.lastVersions++ + if f.lastVersion != 0 { + return f.lastVersion, nil + } + return 1 << 62, nil +} +func (f *fakeReader) Close() error { return nil } + +func TestFallbackStateStoreRoutesPrunedPointReads(t *testing.T) { + primary := &fakeStateStore{earliest: 10} + reader := &fakeReader{} + store := NewFallbackStateStore(primary, reader, 0) + + value, err := store.Get("bank", 7, []byte("k")) + require.NoError(t, err) + require.Equal(t, []byte("historical"), value) + require.Equal(t, 0, primary.gets) + require.Equal(t, 1, reader.gets) + + ok, err := store.Has("bank", 7, []byte("k")) + require.NoError(t, err) + require.True(t, ok) + require.Equal(t, 0, primary.has) + require.Equal(t, 0, reader.has) +} + +func TestFallbackStateStoreKeepsRecentPointReadsOnPrimary(t *testing.T) { + primary := &fakeStateStore{earliest: 10} + reader := &fakeReader{} + store := NewFallbackStateStore(primary, reader, 0) + + value, err := store.Get("bank", 10, []byte("k")) + require.NoError(t, err) + require.Equal(t, []byte("primary"), value) + require.Equal(t, 1, primary.gets) + require.Equal(t, 0, reader.gets) +} + +func TestFallbackStateStoreCachesHistoricalPointReads(t *testing.T) { + primary := &fakeStateStore{earliest: 10} + reader := &fakeReader{} + store := NewFallbackStateStore(primary, reader, 0) + + value, err := store.Get("bank", 7, []byte("k")) + require.NoError(t, err) + require.Equal(t, []byte("historical"), value) + value[0] = 'H' + + value, err = store.Get("bank", 7, []byte("k")) + require.NoError(t, err) + require.Equal(t, []byte("historical"), value) + value[0] = 'H' + + value, err = store.Get("bank", 7, []byte("k")) + require.NoError(t, err) + require.Equal(t, []byte("historical"), value) + require.Equal(t, 1, reader.gets) +} + +func TestFallbackStateStoreCachesHistoricalMisses(t *testing.T) { + primary := &fakeStateStore{earliest: 10} + reader := &fakeReader{getErr: ErrNotFound, hasSet: true} + store := NewFallbackStateStore(primary, reader, 0) + + value, err := store.Get("bank", 7, []byte("missing")) + require.NoError(t, err) + require.Nil(t, value) + + value, err = store.Get("bank", 7, []byte("missing")) + require.NoError(t, err) + require.Nil(t, value) + + ok, err := store.Has("bank", 7, []byte("missing")) + require.NoError(t, err) + require.False(t, ok) + require.Equal(t, 1, reader.gets) + require.Equal(t, 0, reader.has) +} + +func TestFallbackStateStoreCachesHistoricalHasResults(t *testing.T) { + primary := &fakeStateStore{earliest: 10} + reader := &fakeReader{hasResult: true, hasSet: true} + store := NewFallbackStateStore(primary, reader, 0) + + ok, err := store.Has("bank", 7, []byte("k")) + require.NoError(t, err) + require.True(t, ok) + + ok, err = store.Has("bank", 7, []byte("k")) + require.NoError(t, err) + require.True(t, ok) + require.Equal(t, 1, reader.has) +} + +func TestFallbackStateStoreDoesNotUseHasOnlyCacheForGet(t *testing.T) { + primary := &fakeStateStore{earliest: 10} + reader := &fakeReader{hasResult: true, hasSet: true} + store := NewFallbackStateStore(primary, reader, 0) + + ok, err := store.Has("bank", 7, []byte("k")) + require.NoError(t, err) + require.True(t, ok) + + value, err := store.Get("bank", 7, []byte("k")) + require.NoError(t, err) + require.Equal(t, []byte("historical"), value) + require.Equal(t, 1, reader.gets) +} + +type fakePerKeyStateStore struct { + fakeStateStore + perKey map[string]int64 +} + +func (f *fakePerKeyStateStore) GetEarliestVersionForKey(storeKey string) int64 { + if v, ok := f.perKey[storeKey]; ok { + return v + } + return f.earliest +} + +func TestFallbackStateStoreUsesPerKeyEarliestVersion(t *testing.T) { + primary := &fakePerKeyStateStore{ + fakeStateStore: fakeStateStore{earliest: 10}, + perKey: map[string]int64{"evm": 5}, + } + reader := &fakeReader{} + store := NewFallbackStateStore(primary, reader, 0) + + // The EVM store still holds version 7 locally even though the cosmos store + // pruned to 10; the read must stay on the primary. + value, err := store.Get("evm", 7, []byte("k")) + require.NoError(t, err) + require.Equal(t, []byte("primary"), value) + require.Equal(t, 1, primary.gets) + require.Equal(t, 0, reader.gets) + + // Other store keys fall back below the cosmos horizon as before. + value, err = store.Get("bank", 7, []byte("k")) + require.NoError(t, err) + require.Equal(t, []byte("historical"), value) + require.Equal(t, 1, reader.gets) + + ok, err := store.Has("evm", 7, []byte("k")) + require.NoError(t, err) + require.True(t, ok) + require.Equal(t, 1, primary.has) + require.Equal(t, 0, reader.has) +} + +func TestFallbackStateStoreCoverageFloor(t *testing.T) { + primary := &fakeStateStore{earliest: 100} + reader := &fakeReader{} + store := NewFallbackStateStore(primary, reader, 50) + + // The floor becomes the advertised earliest so height gates admit pruned + // heights the fallback can serve. + require.Equal(t, int64(50), store.GetEarliestVersion()) + + // In [floor, local earliest): fallback serves the read. + value, err := store.Get("bank", 60, []byte("k")) + require.NoError(t, err) + require.Equal(t, []byte("historical"), value) + require.Equal(t, 1, reader.gets) + + // Below the floor: the backend has no coverage; stay on the primary. + value, err = store.Get("bank", 49, []byte("k")) + require.NoError(t, err) + require.Equal(t, []byte("primary"), value) + require.Equal(t, 1, reader.gets) + + // Negative versions never fall back. + _, err = store.Get("bank", -1, []byte("k")) + require.NoError(t, err) + require.Equal(t, 1, reader.gets) +} + +func TestFallbackStateStoreRejectsReadsBeyondBackendCoverage(t *testing.T) { + primary := &fakeStateStore{earliest: 100} + reader := &fakeReader{lastVersion: 40} + store := NewFallbackStateStore(primary, reader, 0) + + // The backend has only ingested up to 40; a pruned read at 90 must error + // instead of returning silently-empty state. + _, err := store.Get("bank", 90, []byte("k")) + require.ErrorContains(t, err, "behind requested version") + require.Equal(t, 0, reader.gets) + + _, err = store.Has("bank", 90, []byte("k")) + require.ErrorContains(t, err, "behind requested version") + require.Equal(t, 0, reader.has) + + // The watermark refresh is rate-limited: the second miss must not trigger + // another LastVersion call. + require.Equal(t, 1, reader.lastVersions) + + // Reads at or below the ingested watermark are served. + value, err := store.Get("bank", 40, []byte("k")) + require.NoError(t, err) + require.Equal(t, []byte("historical"), value) + require.Equal(t, 1, reader.gets) +} + +func TestFallbackStateStoreExpiresCachedMisses(t *testing.T) { + primary := &fakeStateStore{earliest: 10} + reader := &fakeReader{getErr: ErrNotFound} + store := NewFallbackStateStore(primary, reader, 0) + + value, err := store.Get("bank", 7, []byte("missing")) + require.NoError(t, err) + require.Nil(t, value) + require.Equal(t, 1, reader.gets) + + // Backdate the cached miss to simulate the TTL lapsing after the offload + // consumer catches up. + cacheKey := readCacheKey{storeKey: "bank", version: 7, key: "missing"} + entry, ok := store.cache.Get(cacheKey) + require.True(t, ok) + entry.missExpiresAt = time.Now().Add(-time.Second) + store.cache.Add(cacheKey, entry) + + reader.getErr = nil + value, err = store.Get("bank", 7, []byte("missing")) + require.NoError(t, err) + require.Equal(t, []byte("historical"), value) + require.Equal(t, 2, reader.gets) +} + +func TestFallbackStateStoreDoesNotCacheHistoricalErrors(t *testing.T) { + primary := &fakeStateStore{earliest: 10} + reader := &fakeReader{getErr: errors.New("boom")} + store := NewFallbackStateStore(primary, reader, 0) + + _, err := store.Get("bank", 7, []byte("k")) + require.Error(t, err) + + reader.getErr = nil + value, err := store.Get("bank", 7, []byte("k")) + require.NoError(t, err) + require.Equal(t, []byte("historical"), value) + require.Equal(t, 2, reader.gets) +} diff --git a/sei-db/state_db/ss/store.go b/sei-db/state_db/ss/store.go index 0fb4b5184e..e59ad4a1d8 100644 --- a/sei-db/state_db/ss/store.go +++ b/sei-db/state_db/ss/store.go @@ -1,15 +1,45 @@ package ss import ( + "fmt" + "github.com/sei-protocol/sei-chain/sei-db/config" + "github.com/sei-protocol/sei-chain/sei-db/db_engine/bigtable" "github.com/sei-protocol/sei-chain/sei-db/db_engine/types" "github.com/sei-protocol/sei-chain/sei-db/state_db/ss/composite" + "github.com/sei-protocol/sei-chain/sei-db/state_db/ss/offload/historical" ) // NewStateStore creates a CompositeStateStore which handles both Cosmos and EVM data. // The backend (pebbledb or rocksdb) is resolved at compile time via build-tag-gated // files in the backend package. When WriteMode/ReadMode are both cosmos_only (the default), // the EVM stores are not opened and the composite store behaves identically to a plain cosmos state store. +// When historical-offload-backend selects a backend (currently "bigtable"), +// the store is wrapped so pruned point reads fall back to it. func NewStateStore(homeDir string, ssConfig config.StateStoreConfig) (types.StateStore, error) { - return composite.NewCompositeStateStore(ssConfig, homeDir) + primary, err := composite.NewCompositeStateStore(ssConfig, homeDir) + if err != nil { + return nil, err + } + switch ssConfig.HistoricalOffloadBackend { + case "": + return primary, nil + case "bigtable": + reader, err := historical.NewBigtableReader(bigtable.Config{ + ProjectID: ssConfig.HistoricalOffloadBigtableProjectID, + InstanceID: ssConfig.HistoricalOffloadBigtableInstance, + Table: ssConfig.HistoricalOffloadBigtableTable, + Family: ssConfig.HistoricalOffloadBigtableFamily, + AppProfile: ssConfig.HistoricalOffloadBigtableAppProfile, + Shards: ssConfig.HistoricalOffloadBigtableShards, + }) + if err != nil { + _ = primary.Close() + return nil, fmt.Errorf("open bigtable historical offload reader: %w", err) + } + return historical.NewFallbackStateStore(primary, reader, ssConfig.HistoricalOffloadEarliestVersion), nil + default: + _ = primary.Close() + return nil, fmt.Errorf("unsupported historical offload backend %q", ssConfig.HistoricalOffloadBackend) + } } diff --git a/sei-db/state_db/ss/store_test.go b/sei-db/state_db/ss/store_test.go index f6e7686c48..5ab0526358 100644 --- a/sei-db/state_db/ss/store_test.go +++ b/sei-db/state_db/ss/store_test.go @@ -50,3 +50,15 @@ func TestNewStateStore(t *testing.T) { require.Equal(t, fmt.Sprintf("value%d", i), string(value)) } } + +func TestNewStateStoreRejectsUnsupportedHistoricalOffloadBackend(t *testing.T) { + ssConfig := config.StateStoreConfig{ + Backend: config.PebbleDBBackend, + AsyncWriteBuffer: 100, + KeepRecent: 500, + HistoricalOffloadBackend: "scylla", + } + + _, err := NewStateStore(filepath.Join(t.TempDir(), "pebbledb"), ssConfig) + require.ErrorContains(t, err, `unsupported historical offload backend "scylla"`) +}