Part 2: Bigtable node read fallback

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
 }

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,

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

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

sei-db/state_db/ss/composite/store.go

@@ -219,6 +219,17 @@
 	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 {

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.

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),
+	))
+}

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)
+	}
+}