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/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/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/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(), } 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]) +}