From 32030ef100a5c2f94899ccbef59bfbe75c643d49 Mon Sep 17 00:00:00 2001 From: Janez Podhostnik Date: Mon, 10 Aug 2026 14:36:38 +0200 Subject: [PATCH 1/2] Retry on Unavailable gRPC errors, reduce result channel buffer Retry when the connection drops (codes.Unavailable); gRPC reconnects on the next attempt so these errors are transient. Reduce the script result channel buffer from 10000 to 500 to apply backpressure to producers earlier and bound memory usage. --- client/interceptors/uci_retry.go | 6 ++++++ scanner/output_handler.go | 2 +- 2 files changed, 7 insertions(+), 1 deletion(-) diff --git a/client/interceptors/uci_retry.go b/client/interceptors/uci_retry.go index ace9747..ceefa9e 100644 --- a/client/interceptors/uci_retry.go +++ b/client/interceptors/uci_retry.go @@ -53,6 +53,12 @@ func RetryUnaryClientInterceptor( return true } + // Retry when the connection drops (e.g. "connection timed out"). + // This is transient: gRPC will reconnect on the next attempt. + if code == codes.Unavailable { + return true + } + if strings.Contains(err.Error(), "please retry for collection in finalized block") { return true } diff --git a/scanner/output_handler.go b/scanner/output_handler.go index 1627a9e..2c85ebc 100644 --- a/scanner/output_handler.go +++ b/scanner/output_handler.go @@ -35,7 +35,7 @@ func NewScriptResultProcessor( handler ScriptResultHandler, ) *ScriptResultProcessor { r := &ScriptResultProcessor{ - resultsChan: make(chan ProcessedAddressBatch, 10000), + resultsChan: make(chan ProcessedAddressBatch, 500), handler: handler, log: logger.With().Str("component", "script_result_processor").Logger(), } From b4e20462724ae30efece892f09c394564a8d91fd Mon Sep 17 00:00:00 2001 From: Janez Podhostnik Date: Thu, 13 Aug 2026 16:28:00 +0200 Subject: [PATCH 2/2] Retry batches at latest scanned height when reference block state is pruned When execution nodes prune state for a batch's reference block, retrying or splitting at the same height can never succeed. Add ScriptErrorActionRetryAtLatestHeight: the batch is resubmitted whole at the latest scanned height once it advances past the batch's height. If the batch is within PrunedStateFatalHeightGap of the latest scanned height (or nothing has been scanned yet), the error is fatal instead: state that close to the tip should never be pruned. All other resubmissions (retry, split, exclude) are also moved to the latest scanned height when it is higher, so retries never re-execute against potentially pruned state. AddressBatch.WithBlockHeight returns a height-adjusted copy that shares done-tracking with the original. Also skip the empty address in event candidate scanning: it is not a real account (e.g. TokensWithdrawn during the initial FLOW mint has no source) and can never be scanned. --- candidates/event_scanner.go | 5 + scanner/address_batch.go | 14 + scanner/engine_builder.go | 9 +- scanner/script_runner.go | 122 ++++++- scanner/script_runner_retry_test.go | 493 ++++++++++++++++++++++++++++ 5 files changed, 638 insertions(+), 5 deletions(-) create mode 100644 scanner/script_runner_retry_test.go diff --git a/candidates/event_scanner.go b/candidates/event_scanner.go index 14c27da..b9e818f 100644 --- a/candidates/event_scanner.go +++ b/candidates/event_scanner.go @@ -83,6 +83,11 @@ func (s *EventCandidatesScanner) Scan( Msg("could not get candidate address from event") return NewCandidatesResultError(err) } + // The empty address is not a real account (e.g. TokensWithdrawn during + // the initial FLOW mint has no source); it can never be scanned. + if address == flow.EmptyAddress { + continue + } addresses[address] = struct{}{} } } diff --git a/scanner/address_batch.go b/scanner/address_batch.go index 0c59567..0df1cfb 100644 --- a/scanner/address_batch.go +++ b/scanner/address_batch.go @@ -83,6 +83,20 @@ func (b *AddressBatch) ExcludeAddress(address flow.Address) { } } +// WithBlockHeight returns a copy of the batch with a new block height. +// The copy shares the done-tracking state with the original, so +// DoneHandling still fires exactly once for the batch. +func (b *AddressBatch) WithBlockHeight(blockHeight uint64) AddressBatch { + return AddressBatch{ + Addresses: b.Addresses, + BlockHeight: blockHeight, + doneHandling: b.doneHandling, + isValid: b.isValid, + + doneOnce: b.doneOnce, + } +} + // Split splits the batch into two batches of equal size. func (b *AddressBatch) Split() (AddressBatch, AddressBatch) { leftDone := make(chan struct{}) diff --git a/scanner/engine_builder.go b/scanner/engine_builder.go index 454118a..465f83f 100644 --- a/scanner/engine_builder.go +++ b/scanner/engine_builder.go @@ -60,11 +60,18 @@ func NewScannerEngineBuilder(cfg *engineConfig) *engine.BuilderBase[*engineConfi // Component: ScriptRunner (depends on ResultProcessor) var scriptRunner *ScriptRunner builder.Component("script_runner", func(cfg *engineConfig) (module.ReadyDoneAware, error) { + // Share the latest-scanned height with the script runner, so batches whose + // reference block became unservable (e.g. "execution state is pruned") can + // be retried at a fresh height. + scriptRunnerConfig := cfg.Config.ScriptRunnerConfig + if scriptRunnerConfig.LatestScannedHeight == nil { + scriptRunnerConfig.LatestScannedHeight = cfg.LatestScanned.GetIfScanned + } scriptRunner = NewScriptRunner( cfg.Config.Logger, cfg.Client, resultProcessor, - cfg.Config.ScriptRunnerConfig, + scriptRunnerConfig, ) return scriptRunner, nil }) diff --git a/scanner/script_runner.go b/scanner/script_runner.go index a6450b1..df6ffcf 100644 --- a/scanner/script_runner.go +++ b/scanner/script_runner.go @@ -19,8 +19,10 @@ package scanner import ( "context" "errors" + "fmt" "regexp" "strings" + "time" "github.com/onflow/cadence" "github.com/onflow/flow-go-sdk" @@ -36,19 +38,45 @@ import ( // As long as they don't wait too long, this is not a problem. const DefaultScriptRunnerMaxConcurrentScripts = 20 +// DefaultPrunedStateFatalHeightGap is the default minimum number of blocks the +// latest scanned height must be ahead of an unservable batch's height for the +// batch to be rescheduled instead of treated as fatal. +const DefaultPrunedStateFatalHeightGap = 10 + type ScriptRunnerConfig struct { Script []byte MaxConcurrentScripts int HandleScriptError func(AddressBatch, error) ScriptErrorAction + + // LatestScannedHeight returns the latest block height known to have servable + // state (typically the latest incrementally scanned height). + // Before a batch is resubmitted (retry, split, or exclude), it is moved to + // this height if it is higher than the batch's current height, so retries + // never re-execute against state that may have been pruned. + // For ScriptErrorActionRetryAtLatestHeight, the resubmission is deferred + // until this advances past the batch's height (a new scanned block). + // May be nil: resubmissions then keep the batch's original height. + LatestScannedHeight func() (uint64, bool) + + // PrunedStateFatalHeightGap is the minimum number of blocks the latest + // scanned height must be ahead of a batch's block height for + // ScriptErrorActionRetryAtLatestHeight to reschedule it. If the gap is + // smaller (or no height has been scanned yet), the error is fatal + // (ctx.Throw) instead: state that close to the tip should never be + // pruned, so waiting for a new block cannot be expected to help. + // Zero disables the fatal check. + // Only evaluated when LatestScannedHeight is non-nil. + PrunedStateFatalHeightGap uint64 } func DefaultScriptRunnerConfig() ScriptRunnerConfig { return ScriptRunnerConfig{ Script: []byte(defaultScript), - MaxConcurrentScripts: DefaultScriptRunnerMaxConcurrentScripts, - HandleScriptError: DefaultHandleScriptError, + MaxConcurrentScripts: DefaultScriptRunnerMaxConcurrentScripts, + HandleScriptError: DefaultHandleScriptError, + PrunedStateFatalHeightGap: DefaultPrunedStateFatalHeightGap, } } @@ -152,7 +180,52 @@ func (r *ScriptRunner) handleBatch(ctx irrecoverable.SignalerContext, input Addr Info(). Msg("retrying") go func() { - r.handleBatch(ctx, input) + r.handleBatch(ctx, r.atLatestHeight(input)) + }() + return + case ScriptErrorActionRetryAtLatestHeight: + // The batch's reference block is no longer servable (e.g. its state was + // pruned on the execution nodes). Splitting is pointless — every half + // would fail the same way — and retrying at the same height can never + // succeed, so resubmit the whole batch once the latest scanned height + // advances to a new block. + // + // Exception: if the batch's height is within PrunedStateFatalHeightGap + // of the latest scanned height, state that close to the tip should + // never be unservable, so the error is fatal instead of reschedulable. + if r.LatestScannedHeight != nil && r.PrunedStateFatalHeightGap > 0 { + if latest, ok := r.LatestScannedHeight(); !ok || + latest < input.BlockHeight || + latest-input.BlockHeight < r.PrunedStateFatalHeightGap { + ctx.Throw(fmt.Errorf( + "batch at height %d is not servable while latest scanned height is %d (fatal gap < %d): %w", + input.BlockHeight, latest, r.PrunedStateFatalHeightGap, err)) + return + } + } + r.log. + Info(). + Uint64("block_height", input.BlockHeight). + Msg("reference block not servable, waiting for a new scanned block to retry") + go func() { + ticker := time.NewTicker(newScannedHeightPollInterval) + defer ticker.Stop() + for { + select { + case <-ctx.Done(): + return + case <-ticker.C: + } + if r.LatestScannedHeight == nil { + // no way to detect a new block: retry at the same height + r.handleBatch(ctx, input) + return + } + if h, ok := r.LatestScannedHeight(); ok && h > input.BlockHeight { + r.handleBatch(ctx, input.WithBlockHeight(h)) + return + } + } }() return case ScriptErrorActionSplit: @@ -165,6 +238,7 @@ func (r *ScriptRunner) handleBatch(ctx irrecoverable.SignalerContext, input Addr Info(). Int("addresses", len(input.Addresses)). Msg("retrying by splitting") + input = r.atLatestHeight(input) left, right := input.Split() go func() { r.handleBatch(ctx, left) @@ -191,7 +265,7 @@ func (r *ScriptRunner) handleBatch(ctx irrecoverable.SignalerContext, input Addr input.ExcludeAddress(address) } go func() { - r.handleBatch(ctx, input) + r.handleBatch(ctx, r.atLatestHeight(input)) }() return case ScriptErrorActionNone: @@ -211,8 +285,30 @@ func (r *ScriptRunner) handleBatch(ctx irrecoverable.SignalerContext, input Addr }() } +// atLatestHeight moves the batch to the latest scanned height if that is +// higher than the batch's current height, so a resubmitted batch never +// re-executes against state that may have been pruned. Returns the batch +// unchanged when LatestScannedHeight is nil or has no higher height. +func (r *ScriptRunner) atLatestHeight(input AddressBatch) AddressBatch { + if r.LatestScannedHeight == nil { + return input + } + if h, ok := r.LatestScannedHeight(); ok && h > input.BlockHeight { + return input.WithBlockHeight(h) + } + return input +} + var accountFrozenRegex = regexp.MustCompile(`\[Error Code: 1204] account (?P
\w{16}) is frozen`) +// newScannedHeightPollInterval is how often the script runner checks whether +// the latest scanned height has advanced while waiting to resubmit a batch via +// ScriptErrorActionRetryAtLatestHeight. There is no retry cap: a long-running +// scan should ride out a multi-hour execution-node state-availability outage +// rather than crash. +// It is a variable so tests can stub it. +var newScannedHeightPollInterval = time.Second + // executeScript retries running the cadence script until we get a successful response back, // returning an array of Balance pairs, along with a boolean representing whether we can continue // or are finished processing. @@ -254,6 +350,18 @@ var _ ScriptErrorAction = ScriptErrorActionRetry{} func (s ScriptErrorActionRetry) isScriptErrorAction() {} +// ScriptErrorActionRetryAtLatestHeight resubmits the whole batch at the latest +// scanned block height (see ScriptRunnerConfig.LatestScannedHeight), after a +// backoff. Use it for errors where the batch's reference block can never +// succeed again, e.g. "execution state is pruned". +// If the batch's height is within ScriptRunnerConfig.PrunedStateFatalHeightGap +// of the latest scanned height, the error is treated as fatal instead. +type ScriptErrorActionRetryAtLatestHeight struct{} + +var _ ScriptErrorAction = ScriptErrorActionRetryAtLatestHeight{} + +func (s ScriptErrorActionRetryAtLatestHeight) isScriptErrorAction() {} + type ScriptErrorActionNone struct{} var _ ScriptErrorAction = ScriptErrorActionNone{} @@ -285,6 +393,12 @@ func DefaultHandleScriptError(_ AddressBatch, err error) ScriptErrorAction { return ScriptErrorActionNone{} } + // The execution nodes no longer serve state for the batch's reference + // block; splitting or retrying at the same height cannot succeed. + if strings.Contains(err.Error(), "execution state is pruned") { + return ScriptErrorActionRetryAtLatestHeight{} + } + if strings.Contains(err.Error(), "state commitment not found") { return ScriptErrorActionNone{} } diff --git a/scanner/script_runner_retry_test.go b/scanner/script_runner_retry_test.go new file mode 100644 index 0000000..596cb81 --- /dev/null +++ b/scanner/script_runner_retry_test.go @@ -0,0 +1,493 @@ +// Flow Batch Scan +// +// Copyright Flow Foundation +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package scanner + +import ( + "context" + "errors" + "sync" + "sync/atomic" + "testing" + "time" + + "github.com/onflow/cadence" + "github.com/onflow/flow-go-sdk" + flowgrpc "github.com/onflow/flow-go-sdk/access/grpc" + "github.com/onflow/flow-go/module/irrecoverable" + "github.com/rs/zerolog" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/mock" + "github.com/stretchr/testify/require" +) + +// fakeClient implements client.Client for script runner tests. +type fakeClient struct { + mu sync.Mutex + + // execute is called by ExecuteScriptAtBlockHeight; heights records every call. + execute func(height uint64) (cadence.Value, error) + heights []uint64 +} + +func (c *fakeClient) ExecuteScriptAtBlockHeight( + _ context.Context, + height uint64, + _ []byte, + _ []cadence.Value, +) (cadence.Value, error) { + c.mu.Lock() + defer c.mu.Unlock() + c.heights = append(c.heights, height) + return c.execute(height) +} + +func (c *fakeClient) GetLatestBlockHeader(context.Context, bool) (*flow.BlockHeader, error) { + return nil, errors.New("not implemented") +} +func (c *fakeClient) GetBlockByHeight(context.Context, uint64) (*flow.Block, error) { + return nil, errors.New("not implemented") +} +func (c *fakeClient) GetTransaction(context.Context, flow.Identifier) (*flow.Transaction, error) { + return nil, errors.New("not implemented") +} +func (c *fakeClient) GetEventsForHeightRange(context.Context, flowgrpc.EventRangeQuery) ([]flow.BlockEvents, error) { + return nil, errors.New("not implemented") +} +func (c *fakeClient) GetCollection(context.Context, flow.Identifier) (*flow.Collection, error) { + return nil, errors.New("not implemented") +} +func (c *fakeClient) SubscribeBlockDigestsFromLatest(context.Context, flow.BlockStatus) (<-chan *flow.BlockDigest, <-chan error, error) { + return nil, nil, errors.New("not implemented") +} +func (c *fakeClient) SubscribeBlockDigestsFromStartHeight(context.Context, uint64, flow.BlockStatus) (<-chan *flow.BlockDigest, <-chan error, error) { + return nil, nil, errors.New("not implemented") +} + +type recordingResultHandler struct { + mu sync.Mutex + heights []uint64 +} + +func (h *recordingResultHandler) Handle(batch ProcessedAddressBatch) error { + h.mu.Lock() + defer h.mu.Unlock() + h.heights = append(h.heights, batch.BlockHeight) + return nil +} + +// TestScriptRunner_RetriesAtLatestHeightOnPruned verifies that a batch failing +// with "execution state is pruned" is retried unsplit once the latest scanned +// height advances to a new block, and that done-tracking fires exactly once. +func TestScriptRunner_RetriesAtLatestHeightOnPruned(t *testing.T) { + // stub the poll interval to keep the test fast + origPoll := newScannedHeightPollInterval + newScannedHeightPollInterval = time.Millisecond + defer func() { newScannedHeightPollInterval = origPoll }() + + prunedErr := errors.New(`rpc error: code = InvalidArgument desc = failed to create storage snapshot: state not found in ledger: execution state is pruned`) + + const originalHeight uint64 = 100 + + // the latest scanned height advances by one block on every pruned failure, + // so each retry waits for and then runs at a new block + var latest atomic.Uint64 + latest.Store(200) + + failuresLeft := 2 + client := &fakeClient{ + execute: func(height uint64) (cadence.Value, error) { + if failuresLeft > 0 { + failuresLeft-- + latest.Add(1) + return nil, prunedErr + } + return cadence.NewArray(nil), nil + }, + } + + resultHandler := &recordingResultHandler{} + resultProcessor := NewScriptResultProcessor(zerolog.Nop(), resultHandler) + + runner := NewScriptRunner( + zerolog.Nop(), + client, + resultProcessor, + ScriptRunnerConfig{ + Script: []byte("access(all) fun main(addresses: [Address]): [Int] { return [] }"), + MaxConcurrentScripts: 2, + HandleScriptError: DefaultHandleScriptError, + LatestScannedHeight: func() (uint64, bool) { return latest.Load(), true }, + }, + ) + ctx, cancel := context.WithCancel(context.Background()) + signalerCtx := irrecoverable.NewMockSignalerContext(t, ctx) + resultProcessor.Start(signalerCtx) + runner.Start(signalerCtx) + defer func() { + cancel() + <-resultProcessor.Done() + <-runner.Done() + }() + + <-resultProcessor.Ready() + <-runner.Ready() + + done := make(chan struct{}) + batch := NewAddressBatch( + []flow.Address{flow.HexToAddress("0x01"), flow.HexToAddress("0x02")}, + originalHeight, + func() { close(done) }, + nil, + ) + runner.Submit(batch) + + select { + case <-done: + case <-time.After(10 * time.Second): + t.Fatal("batch was not processed in time") + } + + client.mu.Lock() + heights := append([]uint64(nil), client.heights...) + client.mu.Unlock() + + // first attempt at the original height, each retry at the newly scanned + // height, and never split (every call has the full address set) + require.Len(t, heights, 3) + assert.Equal(t, originalHeight, heights[0]) + assert.Equal(t, uint64(201), heights[1]) + assert.Equal(t, uint64(202), heights[2]) + + resultHandler.mu.Lock() + defer resultHandler.mu.Unlock() + require.Len(t, resultHandler.heights, 1) + assert.Equal(t, uint64(202), resultHandler.heights[0]) +} + +// TestScriptRunner_RetryUsesLatestHeight verifies that any resubmission (not +// just ScriptErrorActionRetryAtLatestHeight) moves the batch to the latest +// scanned height when it is higher than the batch's height. +func TestScriptRunner_RetryUsesLatestHeight(t *testing.T) { + const originalHeight uint64 = 100 + const latestHeight uint64 = 200 + + failuresLeft := 1 + client := &fakeClient{ + execute: func(height uint64) (cadence.Value, error) { + if failuresLeft > 0 { + failuresLeft-- + return nil, errors.New("boom") + } + return cadence.NewArray(nil), nil + }, + } + + resultHandler := &recordingResultHandler{} + resultProcessor := NewScriptResultProcessor(zerolog.Nop(), resultHandler) + + runner := NewScriptRunner( + zerolog.Nop(), + client, + resultProcessor, + ScriptRunnerConfig{ + Script: []byte("access(all) fun main(addresses: [Address]): [Int] { return [] }"), + MaxConcurrentScripts: 2, + HandleScriptError: func(AddressBatch, error) ScriptErrorAction { + return ScriptErrorActionRetry{} + }, + LatestScannedHeight: func() (uint64, bool) { return latestHeight, true }, + }, + ) + ctx, cancel := context.WithCancel(context.Background()) + signalerCtx := irrecoverable.NewMockSignalerContext(t, ctx) + resultProcessor.Start(signalerCtx) + runner.Start(signalerCtx) + defer func() { + cancel() + <-resultProcessor.Done() + <-runner.Done() + }() + + <-resultProcessor.Ready() + <-runner.Ready() + + done := make(chan struct{}) + batch := NewAddressBatch( + []flow.Address{flow.HexToAddress("0x01")}, + originalHeight, + func() { close(done) }, + nil, + ) + runner.Submit(batch) + + select { + case <-done: + case <-time.After(10 * time.Second): + t.Fatal("batch was not processed in time") + } + + client.mu.Lock() + heights := append([]uint64(nil), client.heights...) + client.mu.Unlock() + + require.Len(t, heights, 2) + assert.Equal(t, originalHeight, heights[0]) + assert.Equal(t, latestHeight, heights[1]) +} + +// TestDefaultHandleScriptError_Pruned verifies the default handler maps the +// pruned-state error to ScriptErrorActionRetryAtLatestHeight. +func TestDefaultHandleScriptError_Pruned(t *testing.T) { + err := errors.New("rpc error: code = InvalidArgument desc = failed to create storage snapshot: state not found in ledger for commit abc (block def): execution state is pruned") + action := DefaultHandleScriptError(AddressBatch{}, err) + assert.IsType(t, ScriptErrorActionRetryAtLatestHeight{}, action) +} + +// TestAddressBatch_WithBlockHeightSharesDoneTracking verifies that a +// height-adjusted copy still completes the original batch's done-tracking. +func TestAddressBatch_WithBlockHeightSharesDoneTracking(t *testing.T) { + done := make(chan struct{}) + var once sync.Once + original := NewAddressBatch( + []flow.Address{flow.HexToAddress("0x01")}, + 100, + func() { once.Do(func() { close(done) }) }, + nil, + ) + + moved := original.WithBlockHeight(200) + assert.Equal(t, uint64(200), moved.BlockHeight) + assert.Equal(t, uint64(100), original.BlockHeight) + + moved.DoneHandling() + moved.DoneHandling() + original.DoneHandling() + + select { + case <-done: + default: + t.Fatal("DoneHandling on the copy did not complete the original batch") + } +} + +// TestScriptRunner_PrunedFatalWhenGapTooSmall verifies that a pruned-state +// error for a batch whose height is within PrunedStateFatalHeightGap of the +// latest scanned height is fatal (ctx.Throw) instead of being rescheduled: +// state that close to the tip should never be pruned, so retrying cannot help. +func TestScriptRunner_PrunedFatalWhenGapTooSmall(t *testing.T) { + prunedErr := errors.New(`rpc error: code = InvalidArgument desc = failed to create storage snapshot: state not found in ledger: execution state is pruned`) + + const batchHeight uint64 = 100 + const latestScanned uint64 = 105 // gap of 5 < threshold of 10 + + client := &fakeClient{ + execute: func(uint64) (cadence.Value, error) { + return nil, prunedErr + }, + } + + resultProcessor := NewScriptResultProcessor(zerolog.Nop(), &recordingResultHandler{}) + + runner := NewScriptRunner( + zerolog.Nop(), + client, + resultProcessor, + ScriptRunnerConfig{ + Script: []byte("access(all) fun main(addresses: [Address]): [Int] { return [] }"), + MaxConcurrentScripts: 2, + HandleScriptError: DefaultHandleScriptError, + LatestScannedHeight: func() (uint64, bool) { return latestScanned, true }, + PrunedStateFatalHeightGap: 10, + }, + ) + ctx, cancel := context.WithCancel(context.Background()) + signalerCtx := irrecoverable.NewMockSignalerContext(t, ctx) + thrown := make(chan error, 1) + signalerCtx.On("Throw", mock.Anything). + Once(). + Run(func(args mock.Arguments) { thrown <- args.Error(0) }). + Return() + resultProcessor.Start(signalerCtx) + runner.Start(signalerCtx) + defer func() { + cancel() + <-resultProcessor.Done() + <-runner.Done() + }() + + <-resultProcessor.Ready() + <-runner.Ready() + + runner.Submit(NewAddressBatch( + []flow.Address{flow.HexToAddress("0x01")}, + batchHeight, + func() {}, + nil, + )) + + select { + case err := <-thrown: + assert.ErrorIs(t, err, prunedErr) + assert.Contains(t, err.Error(), "100") + assert.Contains(t, err.Error(), "105") + case <-time.After(10 * time.Second): + t.Fatal("expected fatal Throw for unservable near-tip batch") + } + + // the script ran once and was never rescheduled + client.mu.Lock() + defer client.mu.Unlock() + assert.Equal(t, []uint64{batchHeight}, client.heights) +} + +// TestScriptRunner_PrunedFatalWhenNothingScanned verifies that a pruned-state +// error is fatal when no height has been scanned yet: there is no fresh block +// to wait for, so rescheduling cannot help. +func TestScriptRunner_PrunedFatalWhenNothingScanned(t *testing.T) { + prunedErr := errors.New(`rpc error: code = InvalidArgument desc = failed to create storage snapshot: state not found in ledger: execution state is pruned`) + + const batchHeight uint64 = 100 + + client := &fakeClient{ + execute: func(uint64) (cadence.Value, error) { + return nil, prunedErr + }, + } + + resultProcessor := NewScriptResultProcessor(zerolog.Nop(), &recordingResultHandler{}) + + runner := NewScriptRunner( + zerolog.Nop(), + client, + resultProcessor, + ScriptRunnerConfig{ + Script: []byte("access(all) fun main(addresses: [Address]): [Int] { return [] }"), + MaxConcurrentScripts: 2, + HandleScriptError: DefaultHandleScriptError, + LatestScannedHeight: func() (uint64, bool) { return 0, false }, + PrunedStateFatalHeightGap: 10, + }, + ) + ctx, cancel := context.WithCancel(context.Background()) + signalerCtx := irrecoverable.NewMockSignalerContext(t, ctx) + thrown := make(chan error, 1) + signalerCtx.On("Throw", mock.Anything). + Once(). + Run(func(args mock.Arguments) { thrown <- args.Error(0) }). + Return() + resultProcessor.Start(signalerCtx) + runner.Start(signalerCtx) + defer func() { + cancel() + <-resultProcessor.Done() + <-runner.Done() + }() + + <-resultProcessor.Ready() + <-runner.Ready() + + runner.Submit(NewAddressBatch( + []flow.Address{flow.HexToAddress("0x01")}, + batchHeight, + func() {}, + nil, + )) + + select { + case err := <-thrown: + assert.ErrorIs(t, err, prunedErr) + case <-time.After(10 * time.Second): + t.Fatal("expected fatal Throw when no height has been scanned yet") + } +} + +// TestScriptRunner_PrunedRetriedWhenGapEqualsThreshold verifies the boundary: +// a gap exactly equal to PrunedStateFatalHeightGap is not fatal, and the batch +// is rescheduled at the latest scanned height as usual. +func TestScriptRunner_PrunedRetriedWhenGapEqualsThreshold(t *testing.T) { + // stub the poll interval to keep the test fast + origPoll := newScannedHeightPollInterval + newScannedHeightPollInterval = time.Millisecond + defer func() { newScannedHeightPollInterval = origPoll }() + + prunedErr := errors.New(`rpc error: code = InvalidArgument desc = failed to create storage snapshot: state not found in ledger: execution state is pruned`) + + const batchHeight uint64 = 100 + const latestScanned uint64 = 110 // gap of 10 == threshold of 10 + + failuresLeft := 1 + client := &fakeClient{ + execute: func(uint64) (cadence.Value, error) { + if failuresLeft > 0 { + failuresLeft-- + return nil, prunedErr + } + return cadence.NewArray(nil), nil + }, + } + + resultHandler := &recordingResultHandler{} + resultProcessor := NewScriptResultProcessor(zerolog.Nop(), resultHandler) + + runner := NewScriptRunner( + zerolog.Nop(), + client, + resultProcessor, + ScriptRunnerConfig{ + Script: []byte("access(all) fun main(addresses: [Address]): [Int] { return [] }"), + MaxConcurrentScripts: 2, + HandleScriptError: DefaultHandleScriptError, + LatestScannedHeight: func() (uint64, bool) { return latestScanned, true }, + PrunedStateFatalHeightGap: 10, + }, + ) + ctx, cancel := context.WithCancel(context.Background()) + // no Throw expectation: any Throw call fails the test + signalerCtx := irrecoverable.NewMockSignalerContext(t, ctx) + resultProcessor.Start(signalerCtx) + runner.Start(signalerCtx) + defer func() { + cancel() + <-resultProcessor.Done() + <-runner.Done() + }() + + <-resultProcessor.Ready() + <-runner.Ready() + + done := make(chan struct{}) + runner.Submit(NewAddressBatch( + []flow.Address{flow.HexToAddress("0x01")}, + batchHeight, + func() { close(done) }, + nil, + )) + + select { + case <-done: + case <-time.After(10 * time.Second): + t.Fatal("batch was not processed in time") + } + + client.mu.Lock() + heights := append([]uint64(nil), client.heights...) + client.mu.Unlock() + + require.Len(t, heights, 2) + assert.Equal(t, batchHeight, heights[0]) + assert.Equal(t, latestScanned, heights[1]) +}