fix handling block execution timeout and returning error

streamingfast · Aug 20, 2024 · 850a4f8 · 850a4f8
1 parent 9ba80f6
commit 850a4f8
Show file tree

Hide file tree

Showing 7 changed files with 11 additions and 8 deletions.
diff --git a/pipeline/exec/baseexec.go b/pipeline/exec/baseexec.go
@@ -108,8 +108,8 @@ func (e *BaseExecutor) wasmCall(outputGetter execout.ExecutionOutputGetter) (cal
 			return nil, fmt.Errorf("block %d: module %q: general wasm execution panicked: %w: %s", clock.Number, e.moduleName, ErrWasmDeterministicExec, errExecutor.Error())
 		}
 		if err != nil {
-			if err := e.ctx.Err(); err != nil {
-				return nil, fmt.Errorf("block %d: module %q: general wasm execution failed: %w", clock.Number, e.moduleName, err)
+			if ctxErr := e.ctx.Err(); ctxErr != nil {
+				return nil, fmt.Errorf("block %d: module %q: general wasm execution failed: %w, %w", clock.Number, e.moduleName, err, ctxErr)
 			}
 			return nil, fmt.Errorf("block %d: module %q: general wasm execution failed: %w: %s", clock.Number, e.moduleName, ErrWasmDeterministicExec, err)
 		}

diff --git a/pipeline/exec/indexexec.go b/pipeline/exec/indexexec.go
@@ -32,6 +32,7 @@ func (i *IndexModuleExecutor) applyCachedOutput([]byte) error {
 func (i *IndexModuleExecutor) run(ctx context.Context, reader execout.ExecutionOutputGetter) (out []byte, outForFiles []byte, moduleOutputData *pbssinternal.ModuleOutput, err error) {
 	_, span := reqctx.WithModuleExecutionSpan(ctx, "exec_index")
 	defer span.EndWithErr(&err)
+	i.ctx = ctx
 
 	var call *wasm.Call
 	if call, err = i.wasmCall(reader); err != nil {

diff --git a/pipeline/exec/mapexec.go b/pipeline/exec/mapexec.go
@@ -35,6 +35,7 @@ func (e *MapperModuleExecutor) run(ctx context.Context, reader execout.Execution
 	_, span := reqctx.WithModuleExecutionSpan(ctx, "exec_map")
 	defer span.EndWithErr(&err)
 
+	e.ctx = ctx
 	var call *wasm.Call
 	if call, err = e.wasmCall(reader); err != nil {
 		return nil, nil, nil, fmt.Errorf("maps wasm call: %w", err)

diff --git a/pipeline/exec/storeexec.go b/pipeline/exec/storeexec.go
@@ -35,6 +35,7 @@ func (e *StoreModuleExecutor) applyCachedOutput(value []byte) error {
 func (e *StoreModuleExecutor) run(ctx context.Context, reader execout.ExecutionOutputGetter) (out []byte, outForFiles []byte, moduleOutputData *pbssinternal.ModuleOutput, err error) {
 	_, span := reqctx.WithModuleExecutionSpan(ctx, "exec_store")
 	defer span.EndWithErr(&err)
+	e.ctx = ctx
 
 	if _, err := e.wasmCall(reader); err != nil {
 		return nil, nil, nil, fmt.Errorf("store wasm call: %w", err)

diff --git a/pipeline/process_block.go b/pipeline/process_block.go
@@ -128,7 +128,7 @@ func (p *Pipeline) processBlock(
 		dmetering.GetBytesMeter(ctx).AddBytesRead(execOutput.Len())
 		err = p.handleStepNew(ctx, clock, cursor, execOutput)
 		if err != nil && err != io.EOF {
-			return fmt.Errorf("step new: handler step new: %w", err)
+			return err
 		}
 		if err == io.EOF {
 			eof = true
@@ -137,20 +137,20 @@ func (p *Pipeline) processBlock(
 		p.blockStepMap[bstream.StepNewIrreversible]++
 		err = p.handleStepNew(ctx, clock, cursor, execOutput)
 		if err != nil && err != io.EOF {
-			return fmt.Errorf("step new irr: handler step new: %w", err)
+			return err
 		}
 		if err == io.EOF {
 			eof = true
 		}
 		err = p.handleStepFinal(clock)
 		if err != nil {
-			return fmt.Errorf("handling step irreversible: %w", err)
+			return err
 		}
 	case bstream.StepIrreversible:
 		p.blockStepMap[bstream.StepIrreversible]++
 		err = p.handleStepFinal(clock)
 		if err != nil {
-			return fmt.Errorf("handling step irreversible: %w", err)
+			return err
 		}
 	}
 

diff --git a/pipeline/terminate.go b/pipeline/terminate.go
@@ -49,7 +49,7 @@ func (p *Pipeline) OnStreamTerminated(ctx context.Context, err error) error {
 	}
 
 	if err := p.stores.flushStores(ctx, p.executionStages, reqDetails.StopBlockNum); err != nil {
-		return fmt.Errorf("step new irr: stores end of stream: %w", err)
+		return fmt.Errorf("flushing stores on termination: %w", err)
 	}
 
 	if reqctx.Details(ctx).IsTier2Request {

diff --git a/service/failure_backoff.go b/service/failure_backoff.go
@@ -57,7 +57,7 @@ func (s *Tier1Service) errorFromRecordedFailure(id string, isProductionMode bool
 	return nil
 }
 
-// Error: rpc error: code = InvalidArgument desc = step new irr: handler step new: execute modules: applying executor results ... store wasm call: block 300: module "store_eth_stats": wasm execution failed ...
+// Error: rpc error: code = InvalidArgument desc = execute modules: applying executor results ... store wasm call: block 300: module "store_eth_stats": wasm execution failed ...
 var blockFailureRE = regexp.MustCompile(`store wasm call: block ([0-9]*): module "([^"]*)"`)
 
 func (s *Tier1Service) recordFailure(requestID string, err error) {