Snapshot Engine + flatKV integration: phase 1

app/abci_test.go

@@ -85,41 +85,60 @@ func TestBeginBlockAppliesMigrationBatchSize(t *testing.T) {
 	require.Equal(t, 321, after, "BeginBlock should push the gov param into the SC store")
 }
 
-// TestMigrationBatchSizeTakesEffectNextBlock is the full end-to-end timing
-// check: a governance proposal committed in block N (written into the block's
-// deliver state, then Commit) only changes the SC store's migration rate when
-// block N+1's BeginBlock runs and reads it from committed state.
+// TestMigrationBatchSizeTakesEffectNextBlock pins when a param change reaches the SC store: BeginBlock
+// applies whatever the param says at the moment it runs, so a proposal that lands later in a block cannot
+// change the rate that block is already migrating at — only the next block's BeginBlock picks it up.
+//
+// The BeginBlock step is driven directly rather than through FinalizeBlock/Commit because the sequence
+// this pins has no legal block-level spelling: getting the param written after BeginBlock but committed at
+// the same height means writing between FinalizeBlock and Commit, which changes committed state for a
+// height whose app hash was already announced to Tendermint. See TestMigrationBatchSizeAppliedAtBlockStart
+// for the same path through a real block.
 func TestMigrationBatchSizeTakesEffectNextBlock(t *testing.T) {
 	a := Setup(t, false, false, false)
-	bg := context.Background()
+	ctx := a.GetContextForDeliverTx([]byte{})
 
-	// Block 1: BeginBlock runs first (param still unset), then the gov
-	// proposal lands by writing into this block's deliver state, then Commit
-	// persists it to the committed multistore.
-	_, err := a.FinalizeBlock(bg, &abci.RequestFinalizeBlock{
-		Header: &tmproto.Header{ChainID: "sei-test", Height: 1, Time: time.Now()},
-	})
-	require.NoError(t, err)
+	// A block begins with the param unset, so the lazily-persisted default leaves migration paused.
+	a.applyMigrationBatchSize(ctx)
+	got, ok := a.rootStore.GetMigrationBatchSize()
+	require.True(t, ok)
+	require.Equal(t, 0, got, "the default param must leave migration paused")
 
+	// A gov proposal raises the rate part-way through that same block.
 	subspace, ok := a.ParamsKeeper.GetSubspace(migration.SubspaceName)
 	require.True(t, ok)
-	subspace.Set(a.GetContextForDeliverTx([]byte{}), migration.KeyNumKeysToMigratePerBlock, uint64(640))
+	subspace.Set(ctx, migration.KeyNumKeysToMigratePerBlock, uint64(640))
 
-	_, err = a.Commit(bg)
-	require.NoError(t, err)
+	got, ok = a.rootStore.GetMigrationBatchSize()
+	require.True(t, ok)
+	require.Equal(t, 0, got, "a param write must not change the rate the current block is migrating at")
 
-	// The param was committed in block 1, but BeginBlock(1) ran before it
-	// existed, so the rate is still paused at this point.
-	got, ok := a.rootStore.GetMigrationBatchSize()
+	// The next block's BeginBlock reads the param and applies it.
+	a.applyMigrationBatchSize(ctx)
+	got, ok = a.rootStore.GetMigrationBatchSize()
 	require.True(t, ok)
-	require.Equal(t, 0, got, "param committed in block 1 must not take effect within block 1")
+	require.Equal(t, 640, got, "the next BeginBlock must apply the new rate")
+}
 
-	// Block 2: BeginBlock reads the now-committed param and applies it.
-	_, err = a.FinalizeBlock(bg, &abci.RequestFinalizeBlock{
-		Header: &tmproto.Header{ChainID: "sei-test", Height: 2, Time: time.Now().Add(time.Second)},
+// TestMigrationBatchSizeAppliedAtBlockStart covers the same path through a real block: a param already in
+// state when the block starts is applied by that block's BeginBlock and survives the commit.
+func TestMigrationBatchSizeAppliedAtBlockStart(t *testing.T) {
+	a := Setup(t, false, false, false)
+	bg := context.Background()
+
+	// Written into the genesis deliver state, before any block's app hash has been taken.
+	subspace, ok := a.ParamsKeeper.GetSubspace(migration.SubspaceName)
+	require.True(t, ok)
+	subspace.Set(a.GetContextForDeliverTx([]byte{}), migration.KeyNumKeysToMigratePerBlock, uint64(640))
+
+	_, err := a.FinalizeBlock(bg, &abci.RequestFinalizeBlock{
+		Header: &tmproto.Header{ChainID: "sei-test", Height: 1, Time: time.Now()},
 	})
 	require.NoError(t, err)
+	_, err = a.Commit(bg)
+	require.NoError(t, err)
 
-	got, _ = a.rootStore.GetMigrationBatchSize()
-	require.Equal(t, 640, got, "migration rate must take effect on the block after the param is committed")
+	got, ok := a.rootStore.GetMigrationBatchSize()
+	require.True(t, ok)
+	require.Equal(t, 640, got, "BeginBlock must apply a param already in state when the block starts")
 }

sei-cosmos/storev2/rootmulti/store.go

@@ -74,6 +74,11 @@ type Store struct {
 	// captured only once (with the real, non-empty changeset) per block.
 	blockChangeSets          []*proto.NamedChangeSet
 	changesetCapturedVersion int64
+	// flushedVersion is the height whose changesets have already been handed to the commit store. A
+	// height is handed over at most once: baseapp asks for the working hash twice per block and then
+	// commits, so flush runs three times per height, and only the first run carries the block's writes.
+	// Handing the later, empty runs down would tell the commit store it has moved on to another block.
+	flushedVersion int64
 	// nextBlockHash is the Tendermint block hash supplied by baseapp for the block being committed.
 	nextBlockHash []byte
 	// nextResultHash is the result hash (merkle root over the block's deterministic tx results)
@@ -134,6 +139,8 @@ func NewStore(
 		hashLoggerConfig:   scConfig.HashLogger,
 		hashLoggerDisabled: !scConfig.HashLogger.Enable,
 		scDir:              scDir,
+		// No height has been flushed yet, and the first block is 1, so -1 cannot collide with it.
+		flushedVersion: -1,
 	}
 	if ssConfig.Enable {
 		ssStore, err := ss.NewStateStore(homeDir, ssConfig)
@@ -222,11 +229,25 @@ func (rs *Store) flush() error {
 			return changeSets[i].Name < changeSets[j].Name
 		})
 	}
-	// Capture the (sorted) aggregate changeset for hash logging once per block. rootmulti flushes twice
-	// per block (GetWorkingHash then Commit) but only the first flush carries the real changeset — the
-	// second sees an empty set because PopChangeSet already drained it — so capture only the first time.
-	// nil is normalized to an empty (non-nil) set so an empty block records the stable empty-changeset
-	// hash rather than a nil one.
+	// A height is handed down once. baseapp requests the working hash in FinalizeBlock and again in
+	// Commit before committing, so flush runs three times per height, and PopChangeSet has already
+	// drained the block's writes by the second run. Handing an empty changeset down is not harmless:
+	// the commit store stamps it with a height it derives from its own last committed block, which the
+	// first working-hash request already advanced, so it would conclude the chain had moved to the next
+	// block and commit one that never existed. Draining above is what makes this emptiness check
+	// meaningful, and it also keeps a stray pair from being attributed to the following height.
+	if rs.flushedVersion == currentVersion {
+		if len(changeSets) > 0 {
+			return fmt.Errorf("rootmulti: %d changeset(s) arrived for height %d after its working hash "+
+				"was taken; the app hash already announced no longer describes the state being committed",
+				len(changeSets), currentVersion)
+		}
+		return nil
+	}
+	rs.flushedVersion = currentVersion
+
+	// Capture the (sorted) aggregate changeset for hash logging once per block. nil is normalized to an
+	// empty (non-nil) set so an empty block records the stable empty-changeset hash rather than a nil one.
 	if !rs.hashLoggerDisabled && rs.changesetCapturedVersion != currentVersion {
 		if changeSets == nil {
 			rs.blockChangeSets = []*proto.NamedChangeSet{}