Suspended executions wasm

go.mod

@@ -42,10 +42,10 @@ require (
 	github.com/santhosh-tekuri/jsonschema/v5 v5.3.1
 	github.com/scylladb/go-reflectx v1.0.1
 	github.com/shopspring/decimal v1.4.0
-	github.com/smartcontractkit/chain-selectors v1.0.100
+	github.com/smartcontractkit/chain-selectors v1.0.104
 	github.com/smartcontractkit/chainlink-common/pkg/chipingress v0.0.11-0.20260626151909-052e55e62e62
 	github.com/smartcontractkit/chainlink-protos/billing/go v0.0.0-20251024234028-0988426d98f4
-	github.com/smartcontractkit/chainlink-protos/cre/go v0.0.0-20260622152157-c8e129347b8b
+	github.com/smartcontractkit/chainlink-protos/cre/go v0.0.0-20260629155926-02c54659e848
 	github.com/smartcontractkit/chainlink-protos/linking-service/go v0.0.0-20251002192024-d2ad9222409b
 	github.com/smartcontractkit/chainlink-protos/metering/go v0.0.0-20260706185759-873029fd9019
 	github.com/smartcontractkit/chainlink-protos/node-platform v0.0.0-20260205130626-db2a2aab956b

pkg/settings/cresettings/README.md

@@ -260,6 +260,7 @@ flowchart
         VaultMaxPerOracleUnexpiredBlobCount{{VaultMaxPerOracleUnexpiredBlobCount}}:::bound
         PerOwner.VaultCiphertextSizeLimit{{PerOwner.VaultCiphertextSizeLimit}}:::bound
         PerOwner.VaultSecretsLimit{{PerOwner.VaultSecretsLimit}}:::bound
+        PerOwner.SuspendOnAwaitEnabled[/PerOwner.SuspendOnAwaitEnabled\]:::gate
     end
 
     subgraph ConfidentialCompute[Confidential Compute executor]

pkg/settings/cresettings/defaults.json

@@ -83,6 +83,7 @@
 		"ConfidentialCompute": {
 			"Rate": "1000rps:1000"
 		}
+		"SuspendOnAwaitEnabled": "false"
 	},
 	"PerWorkflow": {
 		"TriggerRegistrationsTimeout": "10s",
@@ -197,4 +198,4 @@
 		"FeatureEVMWriteReportL1FeeActivePeriod": "[2100-01-01 00:00:00 +0000 UTC,2101-01-01 00:00:00 +0000 UTC]",
 		"FeatureAptosWriteReportBlockTimestampActivePeriod": "[2100-01-01 00:00:00 +0000 UTC,2101-01-01 00:00:00 +0000 UTC]"
 	}
-}
+}

pkg/settings/cresettings/defaults.toml

@@ -80,6 +80,7 @@ WorkflowLimit = '1000'
 WorkflowExecutionConcurrencyLimit = '5'
 VaultCiphertextSizeLimit = '2kb'
 VaultSecretsLimit = '100'
+SuspendOnAwaitEnabled = 'false'
 
 [PerOwner.ConfidentialCompute]
 Rate = '1000rps:1000'

pkg/settings/cresettings/settings.go

@@ -175,6 +175,8 @@ var Default = Schema{
 		WorkflowLimit:                     Int(1000),
 		WorkflowExecutionConcurrencyLimit: Int(5),
 
+		SuspendOnAwaitEnabled: Bool(false),
+
 		// DANGER(cedric): Be extremely careful changing these vault limits below as they act as a default value
 		// used by the Vault OCR plugin -- changing these values could cause issues with the plugin during an image
 		// upgrade as nodes apply the old and new values inconsistently. A safe upgrade path
@@ -379,7 +381,8 @@ type Owners struct {
 	VaultSecretsLimit                 Setting[int] `unit:"{secret}"`
 
 	// ConfidentialCompute holds the per-workflow-owner Confidential Compute settings.
-	ConfidentialCompute ownerConfidentialCompute
+	ConfidentialCompute   ownerConfidentialCompute
+	SuspendOnAwaitEnabled Setting[bool]
 }
 
 type Workflows struct {

pkg/workflows/wasm/host/execution.go

@@ -19,11 +19,13 @@ import (
 	wfpb "github.com/smartcontractkit/chainlink-protos/workflows/go/v2"
 )
 
+// Here we need to distinguish between variables that should survive a replay and
+// variables that need to be re-initialized every replay.
 type execution[T any] struct {
 	fetchRequestsCounter int
 	response             T
 	ctx                  context.Context
-	capabilityResponses  map[int32]<-chan *sdkpb.CapabilityResponse
+	capabilityResponses  map[int32]*asyncResponse[sdkpb.CapabilityRequest, sdkpb.CapabilityResponse]
 	secretsResponses     map[int32]<-chan *secretsResponse
 	pendingCallsLimiter  limits.ResourcePoolLimiter[int]
 	lock                 sync.RWMutex
@@ -37,12 +39,65 @@ type execution[T any] struct {
 	nodeSeed             int64
 	donLogCount          uint32
 	nodeLogCount         uint32
+	awaiting             []int32
+	// peakMemoryBytes is the largest linear memory observed across (re)starts.
+	// It is populated by callWasm and read by Execute to emit the memory metric.
+	peakMemoryBytes int64
+	// suspendOnAwait gates the suspend/resume behaviour. When false, the
+	// execution behaves as it did before suspension was introduced:
+	// awaitCapabilities blocks until each response is available and callCapAsync
+	// always dispatches a fresh call. When true, awaitCapabilities returns
+	// errSuspendExecution while responses are pending and callCapAsync replays
+	// recorded calls instead of re-dispatching them.
+	suspendOnAwait bool
+}
+
+type asyncResponse[I, O any] struct {
+	mu   sync.Mutex
+	ch   <-chan *O
+	resp *O
+	req  *I
+}
+
+func (a *asyncResponse[I, O]) wait(ctx context.Context) (*O, error) {
+	if resp := a.getResp(ctx); resp != nil {
+		return resp, nil
+	}
+
+	select {
+	case <-ctx.Done():
+		return nil, ctx.Err()
+	case o := <-a.ch:
+		a.mu.Lock()
+		defer a.mu.Unlock()
+		a.resp = o
+		return o, nil
+	}
+}
+
+func (a *asyncResponse[I, O]) getResp(ctx context.Context) *O {
+	a.mu.Lock()
+	defer a.mu.Unlock()
+	return a.resp
 }
 
 // callCapAsync async calls a capability by placing execution results onto a
 // channel and storing each channel with a unique identifier for future
 // retrieval on await.
 func (e *execution[T]) callCapAsync(ctx context.Context, req *sdkpb.CapabilityRequest) error {
+	if e.suspendOnAwait {
+		// check if there is already an item in the capabilityResponses for a given callback id.
+		// if there is -> integrity check (matching requests); return without firing goroutine.
+		// else -> legacy path.
+		if asyncResponse, ok := e.capabilityResponses[req.CallbackId]; ok {
+			// there is already an item for this callback id; we must therefore be replaying
+			// perform an integrity check to enforce determinism and return early.
+			if !proto.Equal(asyncResponse.req, req) {
+				return errors.New("non-determinism error")
+			}
+			return nil
+		}
+	}
 	// Acquire a slot from the pool limiter to bound concurrency.
 	free, err := e.pendingCallsLimiter.Wait(ctx, 1)
 	if err != nil {
@@ -52,7 +107,13 @@ func (e *execution[T]) callCapAsync(ctx context.Context, req *sdkpb.CapabilityRe
 	ch := make(chan *sdkpb.CapabilityResponse, 1)
 	e.lock.Lock()
 	defer e.lock.Unlock()
-	e.capabilityResponses[req.CallbackId] = ch
+	e.capabilityResponses[req.CallbackId] = &asyncResponse[
+		sdkpb.CapabilityRequest,
+		sdkpb.CapabilityResponse,
+	]{
+		ch:  ch,
+		req: req,
+	}
 
 	go func() {
 		defer free()
@@ -82,25 +143,61 @@ func (e *execution[T]) callCapAsync(ctx context.Context, req *sdkpb.CapabilityRe
 	return nil
 }
 
+var errSuspendExecution = errors.New("__SUSPEND_EXECUTION__")
+
 func (e *execution[T]) awaitCapabilities(ctx context.Context, acr *sdkpb.AwaitCapabilitiesRequest) (*sdkpb.AwaitCapabilitiesResponse, error) {
 	responses := make(map[int32]*sdkpb.CapabilityResponse, len(acr.Ids))
 
 	e.lock.Lock()
 	defer e.lock.Unlock()
+
+	if !e.suspendOnAwait {
+		// Legacy behaviour: block until each requested response is available,
+		// then consume and remove it from the store.
+		for _, callId := range acr.Ids {
+			ar, ok := e.capabilityResponses[callId]
+			if !ok {
+				return nil, fmt.Errorf("failed to get call from store : %d", callId)
+			}
+
+			select {
+			case <-ctx.Done():
+				return nil, fmt.Errorf("failed to wait for capability response %d : %w", callId, ctx.Err())
+			case resp := <-ar.ch:
+				responses[callId] = resp
+			}
+
+			delete(e.capabilityResponses, callId)
+		}
+
+		return &sdkpb.AwaitCapabilitiesResponse{
+			Responses: responses,
+		}, nil
+	}
+
+	// for all ids, check whether we have a response
+	// if yes: return the response
+	// if no: return suspend execution error, record the ids for which we are still waiting
+	// 	to be picked up by the runWasm routine
+	responsesForAll := true
 	for _, callId := range acr.Ids {
-		ch, ok := e.capabilityResponses[callId]
+		ar, ok := e.capabilityResponses[callId]
 		if !ok {
 			return nil, fmt.Errorf("failed to get call from store : %d", callId)
 		}
 
-		select {
-		case <-ctx.Done():
-			return nil, fmt.Errorf("failed to wait for capability response %d : %w", callId, ctx.Err())
-		case resp := <-ch:
-			responses[callId] = resp
+		resp := ar.getResp(ctx)
+		if resp == nil {
+			responsesForAll = false
+			break
 		}
 
-		delete(e.capabilityResponses, callId)
+		responses[callId] = resp
+	}
+
+	if !responsesForAll {
+		e.awaiting = acr.Ids
+		return nil, errSuspendExecution
 	}
 
 	return &sdkpb.AwaitCapabilitiesResponse{

pkg/workflows/wasm/host/execution_await_order_test.go

@@ -70,7 +70,7 @@ func TestAwaitCapabilities_headOfLineBlocksOnEarlierID(t *testing.T) {
 
 	exec := &execution[*sdkpb.ExecutionResult]{
 		ctx:                 t.Context(),
-		capabilityResponses: make(map[int32]<-chan *sdkpb.CapabilityResponse),
+		capabilityResponses: make(map[int32]*asyncResponse[sdkpb.CapabilityRequest, sdkpb.CapabilityResponse]),
 		pendingCallsLimiter: limits.GlobalResourcePoolLimiter(cresettings.Default.PerWorkflow.CapabilityConcurrencyLimit.DefaultValue),
 		executor:            stub,
 	}

pkg/workflows/wasm/host/execution_semaphore_test.go

@@ -59,7 +59,7 @@ var _ ExecutionHelper = (*slowCapStub)(nil)
 func newTestExec(maxPending int, stub ExecutionHelper) *execution[*sdkpb.ExecutionResult] {
 	return &execution[*sdkpb.ExecutionResult]{
 		ctx:                 context.Background(),
-		capabilityResponses: make(map[int32]<-chan *sdkpb.CapabilityResponse),
+		capabilityResponses: make(map[int32]*asyncResponse[sdkpb.CapabilityRequest, sdkpb.CapabilityResponse]),
 		secretsResponses:    make(map[int32]<-chan *secretsResponse),
 		pendingCallsLimiter: limits.GlobalResourcePoolLimiter[int](maxPending),
 		executor:            stub,