Retry scripts on transient gRPC and pruned-state errors (#57)

* 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.

* 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.

Author: janezpodhostnik

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{}{}
 		}
 	}

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{})

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
 	})

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<address>\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{}
 	}