From 6d5a4052f9d9980cf8b7ec0946b373632e13aafc Mon Sep 17 00:00:00 2001 From: Brandon Chatham Date: Sun, 20 Sep 2026 14:38:26 +0000 Subject: [PATCH 1/4] evmonly: start the receipt write as soon as execution returns; the state commit waits for it Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> --- giga/evmonly/README.md | 16 ++++--- giga/evmonly/executor.go | 12 +++-- giga/evmonly/giga_store.go | 70 +++++++++++++++++++++-------- giga/evmonly/pipeline_store_test.go | 68 ++++++++++++++++++++++++++++ 4 files changed, 138 insertions(+), 28 deletions(-) diff --git a/giga/evmonly/README.md b/giga/evmonly/README.md index 64ce24e987..4e008ef1a7 100644 --- a/giga/evmonly/README.md +++ b/giga/evmonly/README.md @@ -94,13 +94,15 @@ 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; `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. +`receipt.ReceiptStore` interface, including for empty blocks. That write starts +as soon as execution returns, before the block encoder runs and the previous +commit is waited on, and the block's height-advancing state commit waits for it +to land; `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, +or a block that fails after its receipts were handed off, 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 diff --git a/giga/evmonly/executor.go b/giga/evmonly/executor.go index 7aad6c539e..d3783e9095 100644 --- a/giga/evmonly/executor.go +++ b/giga/evmonly/executor.go @@ -48,9 +48,12 @@ 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. + // Breaks the background persistence of a block's state into its stages. One block is + // committed at a time, so one timer serves it. pipelinePhases *seidbmetrics.PhaseTimer + // Breaks the background receipt write into its stages. Receipt writes run one at a time, in + // block order, but overlap the state commit, so they have a timer of their own. + receiptPhases *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. @@ -111,6 +114,7 @@ func NewExecutor(cfg Config, opts ...Option) *Executor { resultPool: newBlockResultPool(cfg.BlockResultPoolSize), blockPhases: seidbmetrics.NewPhaseTimer(otel.Meter(executorMeterName), "evmonly_block"), pipelinePhases: seidbmetrics.NewPhaseTimer(otel.Meter(executorMeterName), "evmonly_pipeline"), + receiptPhases: seidbmetrics.NewPhaseTimer(otel.Meter(executorMeterName), "evmonly_pipeline"), } e.parseSizer = newParseSizer(e.cfg.ParseWorkers) if e.cfg.OCCWorkers > 1 { @@ -128,8 +132,10 @@ func (e *Executor) Close() { } e.closed.Store(true) // Land the commit running behind the last block before the pool it may need goes away. The - // failure is kept rather than reported, for the next AwaitCommits to return. + // failure is kept rather than reported, for the next AwaitCommits to return. The receipt write + // is waited on separately: a block that failed after starting it has no commit to land. _ = e.awaitPipelineCommit() + _ = e.AwaitReceipts() if e.occPool != nil { e.occPool.Close() } diff --git a/giga/evmonly/giga_store.go b/giga/evmonly/giga_store.go index 1c6709910c..53ee41f7c0 100644 --- a/giga/evmonly/giga_store.go +++ b/giga/evmonly/giga_store.go @@ -110,6 +110,10 @@ func (e *Executor) executePreparedBlockWithStore(ctx context.Context, req Prepar return nil, err } gigametrics.SetPhase(gigametrics.PhaseStorage) + // The receipts need nothing but the result, so their write starts here and runs under the rest + // of the block's tail and the caller's, instead of after it. + e.blockPhases.SetPhase("start_receipts") + receipts := e.startReceiptWrite(ctx, blockNumber, result) // 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 @@ -150,7 +154,7 @@ func (e *Executor) executePreparedBlockWithStore(ctx context.Context, req Prepar } } e.blockPhases.SetPhase("start_commit") - if err := e.startPipelineCommit(ctx, blockNumber, result, extra); err != nil { + if err := e.startPipelineCommit(blockNumber, result, receipts, extra); err != nil { return nil, fmt.Errorf("commit state changes for block %d: %w", req.Context.Number, err) } ok = true @@ -310,17 +314,51 @@ func (e *Executor) awaitPipelineCommit() error { return e.pipelineFailure } -// 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. +// startReceiptWrite persists the block's receipts in the background, after the previous block's +// have landed, and returns the write to wait on. The block result is held until its receipts are +// encoded. +// +// The write is recorded as the executor's newest, so AwaitReceipts finds it whether or not the +// block's commit is started afterwards. +func (e *Executor) startReceiptWrite(ctx context.Context, blockNumber int64, result *BlockResult) *receiptWrite { + receipts := &receiptWrite{done: make(chan struct{})} + e.pipelineMu.Lock() + previous := e.pipelineReceipts + e.pipelineReceipts = receipts + 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() { + defer close(receipts.done) + defer releaseResult() + // Receipts land in block order, so the store's version means every block up to it. The + // previous write owns the phase timer until it is done. + if previous != nil { + <-previous.done + if previous.err != nil { + receipts.err = fmt.Errorf("receipts for block %d not written after an earlier failure: %w", blockNumber, previous.err) + return + } + } + defer e.receiptPhases.Reset() + receipts.err = e.persistReceipts(bgCtx, blockNumber, result) + }() + return receipts +} + +// startPipelineCommit commits the block's encoded state changes in the background once its +// receipts have landed, and records what it changed, so the next block reads those changes through +// an overlay rather than waiting for the write. The state changes are encoded from a copy that +// outlives the block result. // // Commits stay ordered because only one is ever in flight: awaitPipelineCommit lands the previous // one before this is called. -func (e *Executor) startPipelineCommit(ctx context.Context, blockNumber int64, result *BlockResult, extra []*proto.NamedChangeSet) error { +func (e *Executor) startPipelineCommit(blockNumber int64, result *BlockResult, receipts *receiptWrite, 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 { @@ -330,20 +368,16 @@ func (e *Executor) startPipelineCommit(ctx context.Context, blockNumber int64, r 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() { defer close(done) defer e.pipelinePhases.Reset() - receipts.err = e.persistReceipts(bgCtx, blockNumber, result) - releaseResult() - close(receipts.done) + // A block whose receipts were lost is a failed block: its state is not committed, so the + // store never holds a block whose receipts cannot be read. + e.pipelinePhases.SetPhase("await_receipts") + <-receipts.done if receipts.err != nil { e.pipelineMu.Lock() e.pipelineErr = receipts.err @@ -361,19 +395,19 @@ func (e *Executor) startPipelineCommit(ctx context.Context, blockNumber int64, r // 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") + e.receiptPhases.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") + e.receiptPhases.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") + e.receiptPhases.SetPhase("await_store") 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) diff --git a/giga/evmonly/pipeline_store_test.go b/giga/evmonly/pipeline_store_test.go index 3abe38438f..250249d06c 100644 --- a/giga/evmonly/pipeline_store_test.go +++ b/giga/evmonly/pipeline_store_test.go @@ -389,6 +389,74 @@ func TestAwaitReceiptsReturnsBeforeTheStateCommitLands(t *testing.T) { require.Equal(t, []int64{41}, store.commitBlock) } +// signallingReceiptStore reports each block whose receipts it was handed on written. +type signallingReceiptStore struct { + *MemoryReceiptStore + written chan int64 +} + +func (s *signallingReceiptStore) SetReceipts(ctx sdk.Context, records []receipt.ReceiptRecord) error { + if err := s.MemoryReceiptStore.SetReceipts(ctx, records); err != nil { + return err + } + s.written <- ctx.BlockHeight() + return nil +} + +// A block's receipts are written while the loop is still waiting on the previous block's commit, +// rather than after it: the write needs only the result. +func TestReceiptWriteStartsBeforeThePreviousCommitLands(t *testing.T) { + chainID := big.NewInt(testChainID) + key, err := crypto.GenerateKey() + require.NoError(t, err) + sender := crypto.PubkeyToAddress(key.PublicKey) + recipient := testAddress(0xa9) + + snapshot := newMemoryGigaSnapshot(40) + snapshot.setBalance(sender, big.NewInt(testFundedBalanceWei)) + store := &gatedCommitStore{recordingGigaStore: &recordingGigaStore{snapshot: snapshot}, release: make(chan struct{})} + receipts := &signallingReceiptStore{MemoryReceiptStore: NewMemoryReceiptStore(), written: make(chan int64, 2)} + // A store-reading block encoder makes the loop wait for the previous commit before it runs. + blockEncoder := func(BlockContext, *BlockResult) ([]*proto.NamedChangeSet, error) { return nil, nil } + executor := NewExecutor(Config{}, + withTestStores(store, receipts, noopChangeSetEncoder), + WithBlockChangeSetEncoder(blockEncoder)) + defer executor.Close() + + executePipelinedBlock(t, executor, chainID, 41, + signLegacyTx(t, key, chainID, 0, &recipient, big.NewInt(7), nil)).Release() + require.NoError(t, executor.AwaitReceipts()) + require.Equal(t, int64(41), <-receipts.written) + + // Block 42 blocks on the loop until 41's commit is released. + blockCtx := blockContext(chainID) + blockCtx.Number = 42 + prepared, err := executor.PrepareBlock(t.Context(), BlockRequest{Context: blockCtx, + Txs: [][]byte{signLegacyTx(t, key, chainID, 1, &recipient, big.NewInt(9), nil)}}) + require.NoError(t, err) + executed := make(chan error, 1) + go func() { + result, err := executor.ExecutePreparedBlock(t.Context(), prepared) + if err == nil { + result.Release() + } + executed <- err + }() + + require.Equal(t, int64(42), <-receipts.written, "block 42's receipts are written while its loop waits") + require.Empty(t, store.commits, "the previous commit is still held") + select { + case err := <-executed: + t.Fatalf("block 42 returned before the previous commit landed: %v", err) + default: + } + + close(store.release) + require.NoError(t, <-executed) + require.NoError(t, executor.AwaitCommits()) + require.Equal(t, []int64{41, 42}, store.commitBlock) +} + // A receipt write that fails is a failed block: the state commit is not attempted and both waiters // report it. func TestFailedReceiptWriteFailsTheBlockBeforeItsStateCommit(t *testing.T) { From 5cd99b7855dcb4c01b69f3d63a6d07feb6ad6400 Mon Sep 17 00:00:00 2001 From: Brandon Chatham Date: Sun, 20 Sep 2026 14:49:29 +0000 Subject: [PATCH 2/4] evmonly: label the receipt and state pipeline timers by stage; test Close draining a failed block's receipt write Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> --- giga/evmonly/executor.go | 9 +++-- giga/evmonly/pipeline_store_test.go | 62 +++++++++++++++++++++++++++++ 2 files changed, 68 insertions(+), 3 deletions(-) diff --git a/giga/evmonly/executor.go b/giga/evmonly/executor.go index d3783e9095..54a2a0edd1 100644 --- a/giga/evmonly/executor.go +++ b/giga/evmonly/executor.go @@ -20,6 +20,7 @@ import ( "github.com/sei-protocol/sei-chain/sei-db/ledger_db/receipt" gigatypes "github.com/sei-protocol/sei-chain/sei-db/state_db/giga/types" "go.opentelemetry.io/otel" + "go.opentelemetry.io/otel/attribute" ) // executorMeterName is the OTel meter this package's instruments are created on. @@ -52,7 +53,8 @@ type Executor struct { // committed at a time, so one timer serves it. pipelinePhases *seidbmetrics.PhaseTimer // Breaks the background receipt write into its stages. Receipt writes run one at a time, in - // block order, but overlap the state commit, so they have a timer of their own. + // block order, but overlap the state commit, so they have a timer of their own on the same + // metric, told apart by a stage label. receiptPhases *seidbmetrics.PhaseTimer // The commit running behind the current block, and what it will write. A block reads the latter @@ -109,12 +111,13 @@ func WithBlockChangeSetEncoder(encoder BlockChangeSetEncoder) Option { // NewExecutor constructs an EVM-only executor. Call Close to disable future OCC // execution on this executor. func NewExecutor(cfg Config, opts ...Option) *Executor { + pipelineTimers := seidbmetrics.NewPhaseTimerFactory(otel.Meter(executorMeterName), "evmonly_pipeline") e := &Executor{ cfg: cfg.WithDefaults(), resultPool: newBlockResultPool(cfg.BlockResultPoolSize), blockPhases: seidbmetrics.NewPhaseTimer(otel.Meter(executorMeterName), "evmonly_block"), - pipelinePhases: seidbmetrics.NewPhaseTimer(otel.Meter(executorMeterName), "evmonly_pipeline"), - receiptPhases: seidbmetrics.NewPhaseTimer(otel.Meter(executorMeterName), "evmonly_pipeline"), + pipelinePhases: pipelineTimers.Build(attribute.String("stage", "state")), + receiptPhases: pipelineTimers.Build(attribute.String("stage", "receipts")), } e.parseSizer = newParseSizer(e.cfg.ParseWorkers) if e.cfg.OCCWorkers > 1 { diff --git a/giga/evmonly/pipeline_store_test.go b/giga/evmonly/pipeline_store_test.go index 250249d06c..ec237eedee 100644 --- a/giga/evmonly/pipeline_store_test.go +++ b/giga/evmonly/pipeline_store_test.go @@ -4,6 +4,7 @@ import ( "errors" "math/big" "testing" + "time" "github.com/ethereum/go-ethereum/crypto" "github.com/stretchr/testify/require" @@ -550,3 +551,64 @@ func TestBackgroundEncoderSeesTheBlocksOwnChanges(t *testing.T) { require.Equal(t, want, encoded[0]) require.Equal(t, []int64{41, 42}, store.commitBlock) } + +// errTestEncoderFailed is the failure a test block encoder reports. +var errTestEncoderFailed = errors.New("block encoder failed") + +// gatedReceiptStore holds every receipt write until release is closed. +type gatedReceiptStore struct { + *MemoryReceiptStore + release chan struct{} +} + +func (s *gatedReceiptStore) SetReceipts(ctx sdk.Context, records []receipt.ReceiptRecord) error { + <-s.release + return s.MemoryReceiptStore.SetReceipts(ctx, records) +} + +// A block that fails on the loop after its receipts were handed off has no state commit for Close to +// drain, so Close waits for the receipt write itself and the receipts are readable once it returns. +func TestCloseWaitsForTheReceiptsOfABlockThatFailedAfterHandingThemOff(t *testing.T) { + chainID := big.NewInt(testChainID) + key, err := crypto.GenerateKey() + require.NoError(t, err) + sender := crypto.PubkeyToAddress(key.PublicKey) + recipient := testAddress(0xa9) + + snapshot := newMemoryGigaSnapshot(40) + snapshot.setBalance(sender, big.NewInt(testFundedBalanceWei)) + store := &recordingGigaStore{snapshot: snapshot} + receipts := &gatedReceiptStore{MemoryReceiptStore: NewMemoryReceiptStore(), release: make(chan struct{})} + blockEncoder := func(BlockContext, *BlockResult) ([]*proto.NamedChangeSet, error) { + return nil, errTestEncoderFailed + } + executor := NewExecutor(Config{}, + withTestStores(store, receipts, noopChangeSetEncoder), + WithBlockChangeSetEncoder(blockEncoder)) + + rawTx := signLegacyTx(t, key, chainID, 0, &recipient, big.NewInt(7), nil) + blockCtx := blockContext(chainID) + blockCtx.Number = 41 + prepared, err := executor.PrepareBlock(t.Context(), BlockRequest{Context: blockCtx, Txs: [][]byte{rawTx}}) + require.NoError(t, err) + _, err = executor.ExecutePreparedBlock(t.Context(), prepared) + require.ErrorIs(t, err, errTestEncoderFailed) + + closed := make(chan struct{}) + go func() { + executor.Close() + close(closed) + }() + select { + case <-closed: + t.Fatal("Close returned while the receipt write was still held") + case <-time.After(50 * time.Millisecond): + } + + close(receipts.release) + <-closed + stored, err := receipts.GetReceipt(newReceiptContext(t.Context(), 41), decodeTx(t, rawTx).Hash()) + require.NoError(t, err) + require.Equal(t, uint64(41), stored.BlockNumber) + require.Empty(t, store.commits, "a block that failed on the loop commits no state") +} From 5f7e57fa4a99c58625d4be24ab54c68710b0b125 Mon Sep 17 00:00:00 2001 From: Brandon Chatham Date: Sun, 20 Sep 2026 14:59:27 +0000 Subject: [PATCH 3/4] evmonly: rename the state commit's receipt wait to await_receipt_write; test the write after a failed one; state the result's retention window Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> --- giga/evmonly/giga_store.go | 6 ++--- giga/evmonly/pipeline_store_test.go | 36 +++++++++++++++++++++++++++++ 2 files changed, 39 insertions(+), 3 deletions(-) diff --git a/giga/evmonly/giga_store.go b/giga/evmonly/giga_store.go index 53ee41f7c0..18e61763c7 100644 --- a/giga/evmonly/giga_store.go +++ b/giga/evmonly/giga_store.go @@ -315,8 +315,8 @@ func (e *Executor) awaitPipelineCommit() error { } // startReceiptWrite persists the block's receipts in the background, after the previous block's -// have landed, and returns the write to wait on. The block result is held until its receipts are -// encoded. +// have landed, and returns the write to wait on. The block result is held until the write has +// landed. // // The write is recorded as the executor's newest, so AwaitReceipts finds it whether or not the // block's commit is started afterwards. @@ -376,7 +376,7 @@ func (e *Executor) startPipelineCommit(blockNumber int64, result *BlockResult, r defer e.pipelinePhases.Reset() // A block whose receipts were lost is a failed block: its state is not committed, so the // store never holds a block whose receipts cannot be read. - e.pipelinePhases.SetPhase("await_receipts") + e.pipelinePhases.SetPhase("await_receipt_write") <-receipts.done if receipts.err != nil { e.pipelineMu.Lock() diff --git a/giga/evmonly/pipeline_store_test.go b/giga/evmonly/pipeline_store_test.go index ec237eedee..43231fe3fe 100644 --- a/giga/evmonly/pipeline_store_test.go +++ b/giga/evmonly/pipeline_store_test.go @@ -481,6 +481,42 @@ func TestFailedReceiptWriteFailsTheBlockBeforeItsStateCommit(t *testing.T) { require.Empty(t, store.commits, "state must not be committed for a block whose receipts were not") } +// The block after a failed receipt write fails too, before its own receipts are attempted: a store +// whose version has moved past a block whose receipts it never got would claim them as written. +func TestReceiptWriteAfterAFailedOneIsNotAttempted(t *testing.T) { + chainID := big.NewInt(testChainID) + key, err := crypto.GenerateKey() + require.NoError(t, err) + sender := crypto.PubkeyToAddress(key.PublicKey) + recipient := testAddress(0xa9) + + snapshot := newMemoryGigaSnapshot(40) + snapshot.setBalance(sender, big.NewInt(testFundedBalanceWei)) + store := &recordingGigaStore{snapshot: snapshot} + receipts := &failingReceiptStore{MemoryReceiptStore: NewMemoryReceiptStore(), err: errTestReceiptWriteFailed} + executor := NewExecutor(Config{}, withTestStores(store, receipts, noopChangeSetEncoder)) + defer executor.Close() + + executePipelinedBlock(t, executor, chainID, 41, + signLegacyTx(t, key, chainID, 0, &recipient, big.NewInt(7), nil)) + require.ErrorIs(t, executor.AwaitReceipts(), errTestReceiptWriteFailed) + + // The store would accept block 42's receipts now; the executor must not offer them. + receipts.err = nil + blockCtx := blockContext(chainID) + blockCtx.Number = 42 + rawTx := signLegacyTx(t, key, chainID, 1, &recipient, big.NewInt(9), nil) + prepared, err := executor.PrepareBlock(t.Context(), BlockRequest{Context: blockCtx, Txs: [][]byte{rawTx}}) + require.NoError(t, err) + _, err = executor.ExecutePreparedBlock(t.Context(), prepared) + require.ErrorIs(t, err, errTestReceiptWriteFailed) + + require.ErrorIs(t, executor.AwaitReceipts(), errTestReceiptWriteFailed) + _, err = receipts.GetReceipt(newReceiptContext(t.Context(), 42), decodeTx(t, rawTx).Hash()) + require.ErrorIs(t, err, receipt.ErrNotFound, "block 42's receipts must not land over the hole at 41") + require.Empty(t, store.commits) +} + // droppingReceiptStore accepts every write and applies none of them, the way a queued store behaves // once an earlier write has failed: it takes the block, waits out its queue, and its version never // reaches it. From 1644b9b85199af142999f81dff8191c6ae170591 Mon Sep 17 00:00:00 2001 From: Brandon Chatham Date: Sun, 20 Sep 2026 15:11:16 +0000 Subject: [PATCH 4/4] ci: retrigger race shards (privval/inspect/wal timeouts on shared runner) Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com>