rpcv2: version the on-disk format before the first release freezes it

cmd/stellar-rpc/internal/rpcv2/backfill/perf_test.go

@@ -105,10 +105,10 @@ func TestStreamingRebuild_ByteIdenticalToColdPath(t *testing.T) {
 // ---------------------------------------------------------------------------
 
 // TestStreamingBin_MatchesSpecFormat asserts the .bin a frozen chunk leaves on
-// disk matches gettransaction §6.1: a uint64-LE entry-count header, then the
-// 16-byte index secret the keys were blinded with, then 20-byte [16-byte key |
-// 4-byte LE seq] entries. freezeChunkBin uses the real txhash.WriteColdBin, so
-// this is the producer's actual on-disk contract.
+// disk matches gettransaction §6.1: the "SBIN" magic-and-version prelude, a
+// uint64-LE entry-count, the 16-byte index secret the keys were blinded with,
+// then 20-byte [16-byte key | 4-byte LE seq] entries. freezeChunkBin uses the
+// real txhash.WriteColdBin, so this is the producer's actual on-disk contract.
 func TestStreamingBin_MatchesSpecFormat(t *testing.T) {
 	cat, _ := smallTxHashIndexCatalog(t, 4)
 
@@ -119,12 +119,13 @@ func TestStreamingBin_MatchesSpecFormat(t *testing.T) {
 	raw, err := os.ReadFile(cat.Layout().TxHashBinPath(0))
 	require.NoError(t, err)
 
-	// §6.1: 24-byte header (8-byte uint64-LE count + 16-byte index secret) +
-	// N * 20-byte entries.
+	// §6.1: 32-byte header (8-byte magic/version prelude + 8-byte uint64-LE
+	// count + 16-byte index secret) + N * 20-byte entries.
 	const (
+		preludeW  = 8                // magic "SBIN" + version + reserved
 		countW    = 8                // uint64-LE entry count
 		secretW   = stores.SecretLen // index secret the keys were blinded with
-		hdrSize   = countW + secretW
+		hdrSize   = preludeW + countW + secretW
 		keyW      = 16 // streamhash.MinKeySize
 		seqW      = 4
 		entryW    = keyW + seqW // 20 bytes exactly
@@ -134,7 +135,9 @@ func TestStreamingBin_MatchesSpecFormat(t *testing.T) {
 	require.Equal(t, streamhash.MinKeySize, keyW, "16-byte key == streamhash routing-key width")
 	require.Len(t, raw, hdrSize+wantCount*entryW, "header + 20-byte entries")
 
-	count := binary.LittleEndian.Uint64(raw[:countW])
+	require.Equal(t, []byte("SBIN"), raw[:4], "magic in on-disk byte order")
+	require.Equal(t, byte(1), raw[4], ".bin format version")
+	count := binary.LittleEndian.Uint64(raw[preludeW : preludeW+countW])
 	require.Equal(t, uint64(wantCount), count, "uint64-LE entry-count header")
 
 	// Each entry: the 16-byte secret-keyed routing key (keyed at ingest — never

cmd/stellar-rpc/internal/rpcv2/catalog/catalog.go

@@ -46,6 +46,15 @@ func Open(
 		return nil, err
 	}
 	c := &Catalog{store: store, logger: logger, layout: layout, txhashIndex: txhashIndex}
+	// Census before the secret mint below: a catalog holding entries outside
+	// this binary's vocabulary (a newer binary's formats, or corruption) is
+	// refused here, so Open writes no catalog entry of its own into a tree it
+	// refuses. RocksDB itself may still create housekeeping files and flush a
+	// previous binary's recovered WAL on Close.
+	if err := c.census(); err != nil {
+		_ = c.Close()
+		return nil, err
+	}
 	// Mint-or-load the cold-index secret up front (get-or-create is not atomic;
 	// here it runs single-threaded) and cache it, so post-Open Secret() reads are
 	// lock-free and cannot fail.

cmd/stellar-rpc/internal/rpcv2/catalog/census.go

@@ -0,0 +1,129 @@
+package catalog
+
+import (
+	"errors"
+	"fmt"
+	"strconv"
+	"strings"
+
+	"github.com/stellar/stellar-rpc/cmd/stellar-rpc/internal/rpcv2/geometry"
+)
+
+// ErrForeignCatalog marks a catalog holding entries outside this binary's
+// vocabulary. The daemon refuses to start on it: the entries were either
+// written by a newer stellar-rpc (whose formats this binary cannot read, and
+// whose artifacts the resolver would otherwise overwrite) or corrupted.
+var ErrForeignCatalog = errors.New("catalog: entries unknown to this version of stellar-rpc")
+
+// censusMaxDetailed caps how many offending entries the refusal error spells
+// out; the total count is always reported.
+const censusMaxDetailed = 10
+
+// census validates every key and value in the store against the exact
+// vocabulary this binary writes. It runs once inside Open, before the secret
+// mint, so a foreign catalog is refused before anything writes into it. The
+// scan is read-only and touches only the catalog (never a hot DB or a file).
+//
+// Everything this binary durably writes must parse here; the whitelist spans
+// the geometry key families AND the meta/catalog-secret key, and the value
+// checks are the census's own exact-token comparisons (State/HotState reads
+// elsewhere are raw casts, not validators).
+func (c *Catalog) census() error {
+	var (
+		offenders []string
+		total     int
+	)
+	flag := func(key, detail string) {
+		total++
+		if len(offenders) < censusMaxDetailed {
+			offenders = append(offenders, fmt.Sprintf("%q: %s", key, detail))
+		}
+	}
+
+	for e, err := range c.prefixScan("") {
+		if err != nil {
+			return fmt.Errorf("catalog: census scan: %w", err)
+		}
+		switch e.Key {
+		case geometry.ConfigEarliestLedger:
+			if !isCanonicalUint32(e.Value) {
+				flag(e.Key, fmt.Sprintf("value %q is not a canonical decimal uint32", e.Value))
+			}
+		case catalogSecretStoreKey:
+			// Never print the secret: it is the live index-blinding key. The
+			// width comes from the field's type, valid before the mint.
+			if len(e.Value) != len(c.secret) {
+				flag(e.Key, fmt.Sprintf("value is %d bytes, want %d (value redacted)", len(e.Value), len(c.secret)))
+			}
+		default:
+			if detail, ok := c.censusArtifactEntry(e.Key, e.Value); !ok {
+				flag(e.Key, detail)
+			}
+		}
+	}
+
+	if total == 0 {
+		return nil
+	}
+	return fmt.Errorf("%w — either written by a newer stellar-rpc (deploy that version or newer) "+
+		"or corrupted; %d offending entr%s: %s",
+		ErrForeignCatalog, total, plural(total), strings.Join(offenders, "; "))
+}
+
+// censusArtifactEntry validates one non-config entry against the three state
+// key families. ok=false returns the reason.
+func (c *Catalog) censusArtifactEntry(key, value string) (string, bool) {
+	switch {
+	case strings.HasPrefix(key, geometry.HotChunkPrefix):
+		if _, ok := geometry.ParseHotChunkKey(key); !ok {
+			return "malformed hot-chunk key", false
+		}
+		if !geometry.IsKnownHotState(geometry.HotState(value)) {
+			return fmt.Sprintf("unknown hot state %q", value), false
+		}
+	case strings.HasPrefix(key, geometry.ChunkPrefix):
+		if _, _, ok := geometry.ParseChunkKey(key); !ok {
+			return "malformed per-chunk artifact key", false
+		}
+		if !geometry.IsKnownState(geometry.State(value)) {
+			return fmt.Sprintf("unknown artifact state %q", value), false
+		}
+	case strings.HasPrefix(key, geometry.TxHashIndexPrefix):
+		cov, ok := geometry.ParseTxHashIndexKey(key)
+		if !ok {
+			return "malformed index coverage key", false
+		}
+		// The builder only ever covers chunks of the key's own index, so a
+		// cross-window coverage is something no binary wrote. Left accepted,
+		// a corrupt frozen one would suppress legitimate index rebuilds.
+		if c.txhashIndex.TxHashIndexID(cov.Lo) != cov.Index ||
+			c.txhashIndex.TxHashIndexID(cov.Hi) != cov.Index {
+			return "index coverage endpoints outside the key's own index", false
+		}
+		if !geometry.IsKnownState(geometry.State(value)) {
+			return fmt.Sprintf("unknown artifact state %q", value), false
+		}
+	default:
+		// Never print an unknown key's value: a newer binary may store key
+		// material under a name this binary does not recognize.
+		return fmt.Sprintf("unknown key (value redacted, %d bytes)", len(value)), false
+	}
+	return "", true
+}
+
+// isCanonicalUint32 reports whether v round-trips through ParseUint and
+// FormatUint byte-identically — the exact form PinEarliestLedger writes.
+func isCanonicalUint32(v string) bool {
+	n, err := strconv.ParseUint(v, 10, 32)
+	if err != nil {
+		return false
+	}
+	return strconv.FormatUint(n, 10) == v
+}
+
+func plural(n int) string {
+	if n == 1 {
+		return "y"
+	}
+	return "ies"
+}

cmd/stellar-rpc/internal/rpcv2/catalog/census_test.go

@@ -0,0 +1,133 @@
+package catalog
+
+import (
+	"fmt"
+	"testing"
+
+	"github.com/stretchr/testify/require"
+
+	"github.com/stellar/stellar-rpc/cmd/stellar-rpc/internal/rpcv2/chunk"
+	"github.com/stellar/stellar-rpc/cmd/stellar-rpc/internal/rpcv2/geometry"
+	"github.com/stellar/stellar-rpc/cmd/stellar-rpc/internal/rpcv2/rocksdb"
+)
+
+// reopenAfter opens a catalog at path, applies mutate, closes it, and returns
+// the error from reopening — the census's verdict on the mutated store.
+func reopenAfter(t *testing.T, mutate func(c *Catalog)) error {
+	t.Helper()
+	path := t.TempDir()
+	cat, err := openKVAt(t, path)
+	require.NoError(t, err)
+	mutate(cat)
+	require.NoError(t, cat.Close())
+	reopened, err := openKVAt(t, path)
+	if reopened != nil {
+		t.Cleanup(func() { _ = reopened.Close() })
+	}
+	return err
+}
+
+// TestCensus_AcceptsOwnVocabulary pins that every key and value this binary
+// writes — all three state families in every state, the pin, and the minted
+// secret — reopens cleanly. This is the whole release-1 write surface.
+func TestCensus_AcceptsOwnVocabulary(t *testing.T) {
+	err := reopenAfter(t, func(c *Catalog) {
+		for i, s := range geometry.AllStates() {
+			id := chunk.ID(i)
+			for _, kind := range geometry.AllKinds() {
+				require.NoError(t, c.put(geometry.ChunkKey(id, kind), string(s)))
+			}
+			// Coverage endpoints must lie in the key's own index window.
+			first := chunk.ID(i) * chunk.ID(geometry.ChunksPerTxhashIndex)
+			idxKey := geometry.TxHashIndexKey(geometry.TxHashIndexID(i), first, first+5)
+			require.NoError(t, c.put(idxKey, string(s)))
+		}
+		for i, s := range geometry.AllHotStates() {
+			require.NoError(t, c.put(geometry.HotChunkKey(chunk.ID(7+i)), string(s)))
+		}
+		require.NoError(t, c.PinEarliestLedger(2))
+	})
+	require.NoError(t, err)
+}
+
+// TestCensus_AcceptsFirstStartResidues pins the crash-shaped first-start
+// catalogs: secret-only (crash after Open, before the pin) and secret+pin.
+func TestCensus_AcceptsFirstStartResidues(t *testing.T) {
+	require.NoError(t, reopenAfter(t, func(*Catalog) {}), "secret-only catalog")
+	require.NoError(t, reopenAfter(t, func(c *Catalog) {
+		require.NoError(t, c.PinEarliestLedger(10002))
+	}), "secret+pin catalog")
+}
+
+// TestCensus_RefusesForeignEntries walks the refusal matrix: every entry shape
+// a newer binary (or corruption) could leave must fail reopen with
+// ErrForeignCatalog and an actionable message.
+func TestCensus_RefusesForeignEntries(t *testing.T) {
+	cases := []struct {
+		name, key, value string
+	}{
+		{"novel prefix", "format:events", "2"},
+		{"novel meta key", "meta/other", "WOULD-BE-SECRET-BYTES"},
+		{"unknown kind under chunk prefix", "chunk:00000001:bogus", string(geometry.StateFrozen)},
+		{"version suffix on chunk state", geometry.ChunkKey(1, geometry.KindEvents), "frozen@2"},
+		{"version suffix on hot state", geometry.HotChunkKey(3), "ready@2"},
+		{"version suffix on index state", geometry.TxHashIndexKey(0, 0, 9), "frozen@2"},
+		{"unknown hot state", geometry.HotChunkKey(4), "warm"},
+		{"unpadded chunk id", "chunk:123:ledgers", string(geometry.StateFrozen)},
+		{"index lo above hi", "index:00000000:00000005:00000002", string(geometry.StateFrozen)},
+		{"cross-window index coverage", "index:00000001:00000000:00003000", string(geometry.StateFrozen)},
+		{"non-canonical pin", geometry.ConfigEarliestLedger, "007"},
+	}
+	for _, tc := range cases {
+		t.Run(tc.name, func(t *testing.T) {
+			err := reopenAfter(t, func(c *Catalog) {
+				require.NoError(t, c.put(tc.key, tc.value))
+			})
+			require.ErrorIs(t, err, ErrForeignCatalog)
+			require.ErrorContains(t, err, "deploy that version or newer")
+			require.ErrorContains(t, err, tc.key)
+			require.NotContains(t, err.Error(), "WOULD-BE-SECRET-BYTES",
+				"an unknown key's value must never be printed; it may be a newer binary's key material")
+		})
+	}
+}
+
+// TestCensus_CapsDetailButCountsAll pins the refusal shape on a store with
+// more offenders than the detail cap: every offender is counted, only the
+// first censusMaxDetailed are spelled out.
+func TestCensus_CapsDetailButCountsAll(t *testing.T) {
+	const n = censusMaxDetailed + 2
+	err := reopenAfter(t, func(c *Catalog) {
+		for i := range n {
+			require.NoError(t, c.put(fmt.Sprintf("future:%02d", i), "x"))
+		}
+	})
+	require.ErrorIs(t, err, ErrForeignCatalog)
+	require.ErrorContains(t, err, fmt.Sprintf("%d offending entries", n))
+	require.ErrorContains(t, err, fmt.Sprintf("future:%02d", censusMaxDetailed-1))
+	require.NotContains(t, err.Error(), fmt.Sprintf("future:%02d", censusMaxDetailed))
+}
+
+// TestCensus_RefusalIsWriteFree pins that a refused Open leaves the store
+// untouched: in particular it must NOT mint a secret into a foreign tree (the
+// census runs before ensureSecret). Simulates a tree whose secret a newer
+// binary relocated: a foreign key present, the secret key absent.
+func TestCensus_RefusalIsWriteFree(t *testing.T) {
+	path := t.TempDir()
+	cat, err := openKVAt(t, path)
+	require.NoError(t, err)
+	require.NoError(t, cat.put("future:key", "x"))
+	require.NoError(t, cat.del(catalogSecretStoreKey))
+	require.NoError(t, cat.Close())
+
+	_, err = openKVAt(t, path)
+	require.ErrorIs(t, err, ErrForeignCatalog)
+
+	// Inspect the raw store: the secret key must still be absent.
+	store, err := rocksdb.New(rocksdb.Config{Path: path, Logger: silentLogger()})
+	require.NoError(t, err)
+	t.Cleanup(func() { _ = store.Close() })
+	_, found, err := store.Get("", []byte(catalogSecretStoreKey))
+	require.NoError(t, err)
+	require.False(t, found, "the refused Open must not have minted a secret")
+}

cmd/stellar-rpc/internal/rpcv2/catalog/kv_test.go

@@ -293,18 +293,19 @@ func TestKV_GracefulCloseAndReopen(t *testing.T) {
 
 	first, err := openKVAt(t, path)
 	require.NoError(t, err)
-	require.NoError(t, first.put("streaming:last_committed_ledger", "999"))
-	require.NoError(t, first.put("config:ledgers_per_tx_index", "100000"))
-	require.NoError(t, first.put("chunk:00000001:lfs", "1"))
+	// Keys must stay inside the census vocabulary — reopen refuses anything else.
+	require.NoError(t, first.put(geometry.ConfigEarliestLedger, "999"))
+	require.NoError(t, first.put(geometry.HotChunkKey(3), string(geometry.HotReady)))
+	require.NoError(t, first.put(geometry.ChunkKey(1, geometry.KindLedgers), string(geometry.StateFrozen)))
 	require.NoError(t, first.Close())
 
 	second, err := openKVAt(t, path)
 	require.NoError(t, err)
 	t.Cleanup(func() { _ = second.Close() })
 
-	assert.Equal(t, "999", kvGetHit(t, second, "streaming:last_committed_ledger"))
-	assert.Equal(t, "100000", kvGetHit(t, second, "config:ledgers_per_tx_index"))
-	assert.Equal(t, "1", kvGetHit(t, second, "chunk:00000001:lfs"))
+	assert.Equal(t, "999", kvGetHit(t, second, geometry.ConfigEarliestLedger))
+	assert.Equal(t, string(geometry.HotReady), kvGetHit(t, second, geometry.HotChunkKey(3)))
+	assert.Equal(t, string(geometry.StateFrozen), kvGetHit(t, second, geometry.ChunkKey(1, geometry.KindLedgers)))
 }
 
 func TestKV_ConcurrentOpsAndCloseRaceFree(t *testing.T) {

cmd/stellar-rpc/internal/rpcv2/catalog/secret.go

@@ -2,7 +2,6 @@ package catalog
 
 import (
 	"crypto/rand"
-	"fmt"
 )
 
 // catalogSecretStoreKey holds the deployment's cold-index secret.
@@ -17,7 +16,8 @@ const catalogSecretStoreKey = "meta/catalog-secret"
 func (c *Catalog) Secret() [32]byte { return c.secret }
 
 // ensureSecret loads the persisted cold-index secret, minting and persisting a
-// fresh random one on first call. Open runs it single-threaded and caches the
+// fresh random one on first call. Open runs it single-threaded, after the
+// census has already validated any persisted value's width, and caches the
 // result; nothing else should call it (get-or-create is not atomic).
 func (c *Catalog) ensureSecret() ([32]byte, error) {
 	var s [32]byte
@@ -26,9 +26,6 @@ func (c *Catalog) ensureSecret() ([32]byte, error) {
 		return s, err
 	}
 	if found {
-		if len(v) != len(s) {
-			return s, fmt.Errorf("persisted cold-index secret is %d bytes, want %d", len(v), len(s))
-		}
 		copy(s[:], v)
 		return s, nil
 	}

cmd/stellar-rpc/internal/rpcv2/catalog/secret_test.go

@@ -29,7 +29,8 @@ func TestSecret_MintedOnceAndStable(t *testing.T) {
 // TestSecret_RejectsCorruptedPersisted pins that a wrong-length persisted
 // secret fails Open loudly instead of silently reminting: a truncated write or
 // downgraded encoding must not swap in a fresh secret, which would orphan every
-// txhash .bin already keyed under the old one.
+// txhash .bin already keyed under the old one. The census catches it (before
+// the mint), and its message must not print the secret's value.
 func TestSecret_RejectsCorruptedPersisted(t *testing.T) {
 	path := t.TempDir()
 
@@ -40,5 +41,7 @@ func TestSecret_RejectsCorruptedPersisted(t *testing.T) {
 
 	_, err = openKVAt(t, path)
 	require.Error(t, err, "a corrupted persisted secret must fail Open, not remint")
-	require.ErrorContains(t, err, "persisted cold-index secret is 5 bytes")
+	require.ErrorIs(t, err, ErrForeignCatalog)
+	require.ErrorContains(t, err, "value is 5 bytes, want 32")
+	require.NotContains(t, err.Error(), "short", "the secret's value must never appear in the error")
 }