evmonly: persist receipts and encode state changes behind the block; Commit advances the head once the receipts have landed

giga/evmonly/README.md

@@ -81,22 +81,26 @@ For each block the executor opens a current `giga.StateView`, executes against
 its EVM-native read methods, converts the resulting `StateChangeSet`, and calls
 `CommitStateChanges`. Execution and commit on an executor are serialized so
 blocks cannot share a stale snapshot or overlap commits; callers must still
-submit block heights in order. The snapshot stays open through the commit and
-is always closed afterward. An empty block still commits an encoded empty
-changeset so the store can advance its height. Stateless preparation can
-continue concurrently with store-backed execution.
+submit block heights in order. The snapshot is closed once execution returns;
+the commit reads from the changeset, not the snapshot. An empty block still
+commits an encoded empty changeset so the store can advance its height.
+Stateless preparation can continue concurrently with store-backed execution.
 
 The encoder is explicit because `giga.StateDB` defines the protobuf commit
 transport but does not define an on-disk key layout. In particular, an encoder
 must preserve `StorageClears` as prefix clears rather than silently dropping
-persisted slots that were not read during execution. Encoding, state commit, or
-receipt-store failures release the block result and return an error without
-invoking `ResultSink`. Ethereum receipts are converted into
+persisted slots that were not read during execution. Encoding and both store
+writes run behind the block, one block at a time in block order; `ResultSink`
+runs on the block loop once that work has been handed off, so it may see a
+result whose writes have not landed yet. Ethereum receipts are converted into
 `receipt.ReceiptRecord` values and persisted through the shared
 `receipt.ReceiptStore` interface before the height-advancing state commit,
-including for empty blocks. A receipt failure leaves state unchanged so the
-block can be retried. A state failure can leave receipts behind, but retrying
-the block overwrites them. `ResultSink` runs only after both stores succeed.
+including for empty blocks; `AwaitReceipts` blocks until the newest block's
+receipts are readable. A failure in either write latches: `AwaitReceipts`
+reports a receipt failure, the next block's execution reports a state failure,
+and the executor accepts no further blocks. A receipt failure leaves state
+unchanged; a state failure can leave receipts behind, which re-executing the
+block after a restart overwrites.
 
 `ExecuteBlock` advances the state store's version itself, independently of any
 ABCI `Commit`, so what the store holds after a restart is decided by the

giga/evmonly/executor.go

@@ -46,13 +46,20 @@ type Executor struct {
 	// Breaks a store-backed block into its stages. That path is serialized by storeMu, so one timer
 	// serves the executor.
 	blockPhases *seidbmetrics.PhaseTimer
+	// Breaks the background persistence of a block into its stages. One block is persisted at a
+	// time, so one timer serves it.
+	pipelinePhases *seidbmetrics.PhaseTimer
 
 	// The commit running behind the current block, and what it will write. A block reads the latter
 	// through an overlay so it need not wait for the former.
-	pipelineMu      sync.Mutex
-	pipelineDone    chan struct{}
-	pipelineErr     error
-	pipelineChanges *pendingChanges
+	pipelineMu sync.Mutex
+	// Closed once the block is fully persisted.
+	pipelineDone chan struct{}
+	pipelineErr  error
+	// The most recent block's receipt write; kept after the block retires so a waiter that arrives
+	// late still finds its answer.
+	pipelineReceipts *receiptWrite
+	pipelineChanges  *pendingChanges
 	// Counts commits started, so a reader can tell that a block landed between two of its steps.
 	pipelineGeneration uint64
 	// The first commit that failed, kept so no caller can miss it.
@@ -98,9 +105,10 @@ func WithBlockChangeSetEncoder(encoder BlockChangeSetEncoder) Option {
 // execution on this executor.
 func NewExecutor(cfg Config, opts ...Option) *Executor {
 	e := &Executor{
-		cfg:         cfg.WithDefaults(),
-		resultPool:  newBlockResultPool(cfg.BlockResultPoolSize),
-		blockPhases: seidbmetrics.NewPhaseTimer(otel.Meter(executorMeterName), "evmonly_block"),
+		cfg:            cfg.WithDefaults(),
+		resultPool:     newBlockResultPool(cfg.BlockResultPoolSize),
+		blockPhases:    seidbmetrics.NewPhaseTimer(otel.Meter(executorMeterName), "evmonly_block"),
+		pipelinePhases: seidbmetrics.NewPhaseTimer(otel.Meter(executorMeterName), "evmonly_pipeline"),
 	}
 	if e.cfg.OCCWorkers > 1 {
 		e.occPool = newOCCWorkerPool(e.cfg.OCCWorkers)

giga/evmonly/executor_test.go

@@ -174,19 +174,21 @@ func TestExecutorReturnsReceiptStoreError(t *testing.T) {
 
 	require.ErrorIs(t, err, storeErr)
 	require.Nil(t, result)
-	require.Empty(t, sink.results)
+	// Receipts are written behind the block, so the sink has already seen the result.
+	require.Len(t, sink.results, 1)
+	sink.releases[0]()
 	require.Equal(t, BlockResultPoolStats{Capacity: 1, Available: 1}, executor.ResultPoolStats())
 	view := stateStore.OpenView()
 	require.Zero(t, view.GetBlockHeight())
 	view.Close()
 
+	// A failed write is latched: the executor refuses further blocks rather than run ahead of a
+	// store that is missing a block.
 	receiptStore.err = nil
 	result, err = executor.ExecuteBlock(t.Context(), request)
-	require.NoError(t, err)
-	require.NotNil(t, result)
+	require.ErrorIs(t, err, storeErr)
+	require.Nil(t, result)
 	require.Len(t, sink.results, 1)
-	result.Release()
-	sink.releases[0]()
 }
 
 func TestExecutorPooledResultRelease(t *testing.T) {

giga/evmonly/giga_store.go

@@ -10,6 +10,7 @@ import (
 
 	gigametrics "github.com/sei-protocol/sei-chain/giga/metrics"
 	"github.com/sei-protocol/sei-chain/sei-db/common/keys"
+	seidbtypes "github.com/sei-protocol/sei-chain/sei-db/db_engine/types"
 	"github.com/sei-protocol/sei-chain/sei-db/proto"
 	gigatypes "github.com/sei-protocol/sei-chain/sei-db/state_db/giga/types"
 )
@@ -26,17 +27,12 @@ var (
 var _ StateReader = gigaSnapshotStateReader{}
 
 // NamedChangeSetEncoder converts an executor-native state result into the
-// on-disk changesets understood by a giga store. It is called synchronously
-// while the block's read snapshot is still open. It must treat the input as
-// immutable and must not retain references to it after returning.
+// on-disk changesets understood by a giga store. It is called in the background,
+// after the previous block's commit has landed and before this block's starts, on a
+// copy of the block's changes that it must treat as immutable.
 //
-// What it returns must not alias the input either. The commit runs in the background, outliving
-// the block result and its return to the pool, so an aliasing pair would be rewritten underneath
-// the write by the next block.
-//
-// It must read only the changeset it is given. Encoding overlaps the previous block's commit, so
-// an encoder that reads the store would see a store mid-write. Expanding a storage clear is the
-// one exception, and the executor waits for that commit before encoding a block that has one.
+// It may read the store, which then holds every earlier block and none of this one; expanding a
+// storage clear does so.
 type NamedChangeSetEncoder func(StateChangeSet) ([]*proto.NamedChangeSet, error)
 
 // BlockChangeSetEncoder contributes named changesets that are committed in the
@@ -54,8 +50,7 @@ func (e *Executor) executePreparedBlockWithStore(ctx context.Context, req Prepar
 	if stateStore == nil {
 		return nil, errMissingStateStore
 	}
-	receiptStore := e.receiptStore
-	if receiptStore == nil {
+	if e.receiptStore == nil {
 		return nil, errMissingReceiptStore
 	}
 	if e.changeSetEncoder == nil {
@@ -115,23 +110,24 @@ func (e *Executor) executePreparedBlockWithStore(ctx context.Context, req Prepar
 		return nil, err
 	}
 	gigametrics.SetPhase(gigametrics.PhaseStorage)
-	// Encoding that reads the store has to see a store holding every earlier block and none of
-	// this one, so the previous commit lands first. Encoding that reads only this block's own
-	// changes runs while that commit is still going, and waits below instead.
-	settleBeforeEncoding := e.encodingReadsTheStore(&result.ChangeSet)
-	if settleBeforeEncoding {
+	// One commit is in flight at a time, so the previous one lands before this block starts its
+	// own. It has had this block's whole execution to run, so it rarely still holds. A block encoder
+	// that reads the store needs it landed before it runs; one that reads only the result overlaps
+	// it. The state encoder runs in the background after this wait either way, so a storage clear
+	// it expands against the store sees every earlier block and none of this one.
+	settleBeforeBlockEncoder := e.blockChangeSetEncoder != nil && e.blockEncoderReadsStore
+	if settleBeforeBlockEncoder {
 		e.blockPhases.SetPhase("await_commit")
 		if err := e.awaitPipelineCommit(); err != nil {
 			return nil, err
 		}
 	}
-	e.blockPhases.SetPhase("encode_changesets")
-	changesets, err := e.changeSetEncoder(result.ChangeSet)
-	if err != nil {
-		return nil, fmt.Errorf("encode state changes for block %d: %w", req.Context.Number, err)
-	}
+	// The block encoder stays on the loop: what it stages is what the caller reports for the block,
+	// so it has to have run when this returns.
+	var extra []*proto.NamedChangeSet
 	if e.blockChangeSetEncoder != nil {
-		extra, err := e.blockChangeSetEncoder(req.Context, result)
+		e.blockPhases.SetPhase("encode_block_changesets")
+		extra, err = e.blockChangeSetEncoder(req.Context, result)
 		if err != nil {
 			return nil, fmt.Errorf("encode block changes for block %d: %w", req.Context.Number, err)
 		}
@@ -143,45 +139,48 @@ func (e *Executor) executePreparedBlockWithStore(ctx context.Context, req Prepar
 					cs.Name, req.Context.Number, errBlockEncoderUsedEVMStoreKey)
 			}
 		}
-		changesets = append(changesets, extra...)
 	}
 	if err := ctx.Err(); err != nil {
 		return nil, err
 	}
-	e.blockPhases.SetPhase("encode_receipts")
-	records, err := e.receiptRecordsParallel(ctx, req.Context.Number, result)
-	if err != nil {
-		return nil, fmt.Errorf("encode receipts for block %d: %w", req.Context.Number, err)
-	}
-	e.blockPhases.SetPhase("write_receipts")
-	if err := receiptStore.SetReceipts(newReceiptContext(ctx, blockNumber), records); err != nil {
-		return nil, fmt.Errorf("store receipts for block %d: %w", req.Context.Number, err)
-	}
-	// One commit is in flight at a time, so the previous one lands before this block starts its
-	// own. It has had this block's whole execution to run, so it rarely still holds.
-	if !settleBeforeEncoding {
+	if !settleBeforeBlockEncoder {
 		e.blockPhases.SetPhase("await_commit")
 		if err := e.awaitPipelineCommit(); err != nil {
 			return nil, err
 		}
 	}
-	e.blockPhases.SetPhase("commit_state")
-	if err := e.startPipelineCommit(blockNumber, changesets, &result.ChangeSet); err != nil {
+	e.blockPhases.SetPhase("start_commit")
+	if err := e.startPipelineCommit(ctx, blockNumber, result, extra); err != nil {
 		return nil, fmt.Errorf("commit state changes for block %d: %w", req.Context.Number, err)
 	}
 	ok = true
 	return result, nil
 }
 
-// encodingReadsTheStore reports whether encoding this block's changesets reads the store as well as
-// the block's own changes, which decides whether encoding may overlap the previous block's commit.
+// receiptWrite is one block's background receipt write: done is closed once err is set.
+type receiptWrite struct {
+	done chan struct{}
+	err  error
+}
+
+// AwaitReceipts blocks until the receipts of the last block this executor ran are in the receipt
+// store, and reports the write's failure if it had one.
 //
-// Expanding a storage clear iterates the live store to find the slots to delete, so a block that
-// clears one must not be encoded against a store mid-commit. A block encoder is caller-supplied and
-// free to read whatever it likes, so one is assumed to read the store unless it was registered as
-// store-independent: assuming otherwise would surrender the receipt-stage slack on every block.
-func (e *Executor) encodingReadsTheStore(changes *StateChangeSet) bool {
-	return len(changes.StorageClears) > 0 || (e.blockChangeSetEncoder != nil && e.blockEncoderReadsStore)
+// Receipts are written in the background behind ExecutePreparedBlock. A caller that publishes a
+// block to readers who expect its receipts has to wait here first. Unlike AwaitCommits it does not
+// wait for the block's state commit, which keeps landing behind the next block.
+func (e *Executor) AwaitReceipts() error {
+	if e == nil {
+		return nil
+	}
+	e.pipelineMu.Lock()
+	write := e.pipelineReceipts
+	e.pipelineMu.Unlock()
+	if write == nil {
+		return nil
+	}
+	<-write.done
+	return write.err
 }
 
 // AwaitCommits blocks until every block this executor has run is committed, and reports the first
@@ -271,7 +270,7 @@ func (e *Executor) pipelineFailureLocked() error {
 		return e.pipelineFailure
 	}
 	if e.pipelineErr != nil {
-		return fmt.Errorf("commit state changes: %w", e.pipelineErr)
+		return fmt.Errorf("persist block: %w", e.pipelineErr)
 	}
 	return nil
 }
@@ -298,7 +297,7 @@ func (e *Executor) awaitPipelineCommit() error {
 		// Only the waiters on this commit retire it; a later one owns its own state.
 		if e.pipelineDone == done {
 			if e.pipelineErr != nil && e.pipelineFailure == nil {
-				e.pipelineFailure = fmt.Errorf("commit state changes: %w", e.pipelineErr)
+				e.pipelineFailure = fmt.Errorf("persist block: %w", e.pipelineErr)
 			}
 			e.pipelineDone = nil
 			e.pipelineChanges = nil
@@ -311,13 +310,17 @@ func (e *Executor) awaitPipelineCommit() error {
 	return e.pipelineFailure
 }
 
-// startPipelineCommit writes the block in the background and records what it changed, so the next
-// block reads those changes through an overlay rather than waiting for the write.
+// startPipelineCommit persists the block in the background, receipts first and then the encoded
+// state changes, and records what it changed, so the next block reads those changes through an
+// overlay rather than waiting for the write. The block result is held until its receipts are
+// encoded; the state changes are encoded from a copy that outlives it.
 //
 // Commits stay ordered because only one is ever in flight: awaitPipelineCommit lands the previous
 // one before this is called.
-func (e *Executor) startPipelineCommit(blockNumber int64, changesets []*proto.NamedChangeSet, changes *StateChangeSet) error {
-	pending := newPendingChanges(changes.clone())
+func (e *Executor) startPipelineCommit(ctx context.Context, blockNumber int64, result *BlockResult, extra []*proto.NamedChangeSet) error {
+	changes := result.ChangeSet.clone()
+	pending := newPendingChanges(changes)
+	receipts := &receiptWrite{done: make(chan struct{})}
 	done := make(chan struct{})
 	e.pipelineMu.Lock()
 	if failure := e.pipelineFailure; failure != nil {
@@ -327,19 +330,75 @@ func (e *Executor) startPipelineCommit(blockNumber int64, changesets []*proto.Na
 	e.pipelineChanges = pending
 	e.pipelineGeneration++
 	e.pipelineDone = done
+	e.pipelineReceipts = receipts
 	e.pipelineErr = nil
 	e.pipelineMu.Unlock()
 
+	// The write finishes even if the request that ran the block is cancelled: Close waits for it,
+	// and a failure is reported through the pipeline rather than by dropping the block.
+	bgCtx := context.WithoutCancel(ctx)
+	releaseResult := result.retain()
 	go func() {
-		err := e.stateStore.CommitStateChanges(blockNumber, changesets)
+		defer close(done)
+		defer e.pipelinePhases.Reset()
+		receipts.err = e.persistReceipts(bgCtx, blockNumber, result)
+		releaseResult()
+		close(receipts.done)
+		if receipts.err != nil {
+			e.pipelineMu.Lock()
+			e.pipelineErr = receipts.err
+			e.pipelineMu.Unlock()
+			return
+		}
+		err := e.commitStateChanges(blockNumber, changes, extra)
 		e.pipelineMu.Lock()
 		e.pipelineErr = err
 		e.pipelineMu.Unlock()
-		close(done)
 	}()
 	return nil
 }
 
+// persistReceipts encodes the block's receipts and writes them to the receipt store, returning once
+// they are readable there.
+func (e *Executor) persistReceipts(ctx context.Context, blockNumber int64, result *BlockResult) error {
+	e.pipelinePhases.SetPhase("encode_receipts")
+	records, err := e.receiptRecordsParallel(ctx, uint64(blockNumber), result) //nolint:gosec // G115: non-negative, checked by the caller.
+	if err != nil {
+		return fmt.Errorf("encode receipts for block %d: %w", blockNumber, err)
+	}
+	e.pipelinePhases.SetPhase("write_receipts")
+	if err := e.receiptStore.SetReceipts(newReceiptContext(ctx, blockNumber), records); err != nil {
+		return fmt.Errorf("store receipts for block %d: %w", blockNumber, err)
+	}
+	// A store that applies writes from its own queue reports the landing through its version;
+	// a write it dropped after accepting shows up as a version short of this block.
+	if waiter, ok := e.receiptStore.(seidbtypes.PendingWriteWaiter); ok {
+		e.pipelinePhases.SetPhase("await_receipts")
+		waiter.WaitForPendingWrites()
+		if latest := e.receiptStore.LatestVersion(); latest < blockNumber {
+			return fmt.Errorf("receipts for block %d did not land: receipt store is at block %d", blockNumber, latest)
+		}
+	}
+	return nil
+}
+
+// commitStateChanges encodes the block's state changes, appends the block encoder's, and commits
+// them to the state store.
+func (e *Executor) commitStateChanges(blockNumber int64, changes *StateChangeSet, extra []*proto.NamedChangeSet) error {
+	e.pipelinePhases.SetPhase("encode_changesets")
+	var stateChanges StateChangeSet
+	if changes != nil {
+		stateChanges = *changes
+	}
+	changesets, err := e.changeSetEncoder(stateChanges)
+	if err != nil {
+		return fmt.Errorf("encode state changes for block %d: %w", blockNumber, err)
+	}
+	changesets = append(changesets, extra...)
+	e.pipelinePhases.SetPhase("commit_state")
+	return e.stateStore.CommitStateChanges(blockNumber, changesets)
+}
+
 type gigaSnapshotStateReader struct {
 	snapshot     gigatypes.EVMStateView
 	missingState StateReader

giga/evmonly/giga_store_test.go

@@ -338,7 +338,7 @@ func TestExecutorGigaStoreFailuresDoNotCommitPartialState(t *testing.T) {
 		require.Equal(t, 1, snapshot.closeCount)
 	})
 
-	t.Run("context canceled during encoding", func(t *testing.T) {
+	t.Run("context canceled during encoding still persists the executed block", func(t *testing.T) {
 		snapshot := newMemoryGigaSnapshot(0)
 		store := &recordingGigaStore{snapshot: snapshot}
 		ctx, cancel := context.WithCancel(t.Context())
@@ -349,9 +349,10 @@ func TestExecutorGigaStoreFailuresDoNotCommitPartialState(t *testing.T) {
 
 		result, err := executor.ExecuteBlock(ctx, BlockRequest{Context: blockContext(big.NewInt(testChainID))})
 
-		require.ErrorIs(t, err, context.Canceled)
-		require.Nil(t, result)
-		require.Empty(t, store.commits)
+		require.NoError(t, err)
+		require.NotNil(t, result)
+		result.Release()
+		require.Len(t, store.commits, 1)
 		require.Equal(t, 1, snapshot.closeCount)
 		require.Equal(t, BlockResultPoolStats{Capacity: 1, Available: 1}, executor.ResultPoolStats())
 	})