From e84cab0a298588ed873a68356f0009c32746510b Mon Sep 17 00:00:00 2001 From: Cedric Cordenier Date: Thu, 25 Jun 2026 16:19:37 +0100 Subject: [PATCH 1/6] feat(wasm/host): suspend & resume executions on await Suspends a guest by trapping await_capabilities when an awaited capability response is not yet ready, waits for the response, and resumes by re-instantiating the guest in a fresh store while preserving the fuel/epoch budget. Gated by ModuleConfig.SuspensionEnabled. Adds the suspended_executions and non_deterministic_input standard tests. --- go.mod | 4 +- go.sum | 8 +- pkg/settings/cresettings/README.md | 1 + pkg/settings/cresettings/defaults.json | 3 +- pkg/settings/cresettings/defaults.toml | 1 + pkg/settings/cresettings/settings.go | 5 +- pkg/workflows/wasm/host/execution.go | 112 ++++++- .../wasm/host/execution_await_order_test.go | 2 +- .../wasm/host/execution_semaphore_test.go | 2 +- pkg/workflows/wasm/host/module.go | 305 +++++++++++------- pkg/workflows/wasm/host/module_test.go | 2 +- .../non_deterministic_input/main_wasip1.go | 27 ++ .../suspended_executions/main_wasip1.go | 40 +++ .../wasm/host/wasm_nondeterminism_test.go | 94 ++++++ pkg/workflows/wasm/host/wasm_suspend_test.go | 94 ++++++ pkg/workflows/wasm/host/wasm_test.go | 4 +- 16 files changed, 569 insertions(+), 135 deletions(-) create mode 100644 pkg/workflows/wasm/host/standard_tests/non_deterministic_input/main_wasip1.go create mode 100644 pkg/workflows/wasm/host/standard_tests/suspended_executions/main_wasip1.go create mode 100644 pkg/workflows/wasm/host/wasm_nondeterminism_test.go create mode 100644 pkg/workflows/wasm/host/wasm_suspend_test.go diff --git a/go.mod b/go.mod index 1299440695..05da34e18a 100644 --- a/go.mod +++ b/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/node-platform v0.0.0-20260205130626-db2a2aab956b github.com/smartcontractkit/chainlink-protos/storage-service v0.3.0 diff --git a/go.sum b/go.sum index 3441393f7f..97f73ed97f 100644 --- a/go.sum +++ b/go.sum @@ -256,14 +256,14 @@ github.com/scylladb/go-reflectx v1.0.1 h1:b917wZM7189pZdlND9PbIJ6NQxfDPfBvUaQ7cj github.com/scylladb/go-reflectx v1.0.1/go.mod h1:rWnOfDIRWBGN0miMLIcoPt/Dhi2doCMZqwMCJ3KupFc= github.com/shopspring/decimal v1.4.0 h1:bxl37RwXBklmTi0C79JfXCEBD1cqqHt0bbgBAGFp81k= github.com/shopspring/decimal v1.4.0/go.mod h1:gawqmDU56v4yIKSwfBSFip1HdCCXN8/+DMd9qYNcwME= -github.com/smartcontractkit/chain-selectors v1.0.100 h1:wpiSpmI/eFjY+wx/nPr5VuNF4hki0prIBMKEaQWn3g4= -github.com/smartcontractkit/chain-selectors v1.0.100/go.mod h1:qy7whtgG5g+7z0jt0nRyii9bLND9m15NZTzuQPkMZ5w= +github.com/smartcontractkit/chain-selectors v1.0.104 h1:/n9pPGM5W/+r1eHoWZv4VwX9LNS1af4+ICyhM8zKRNM= +github.com/smartcontractkit/chain-selectors v1.0.104/go.mod h1:qy7whtgG5g+7z0jt0nRyii9bLND9m15NZTzuQPkMZ5w= github.com/smartcontractkit/chainlink-common/pkg/chipingress v0.0.11-0.20260626151909-052e55e62e62 h1:o7vfwNQjQbMKQ9YsZFQOxvU7RMXD/wKnZsX5N9sDS3w= github.com/smartcontractkit/chainlink-common/pkg/chipingress v0.0.11-0.20260626151909-052e55e62e62/go.mod h1:HmUyH2oD9m+GRpKq7q3vuRnm1F2Uczf/Nd1v3ipMSK8= github.com/smartcontractkit/chainlink-protos/billing/go v0.0.0-20251024234028-0988426d98f4 h1:GCzrxDWn3b7jFfEA+WiYRi8CKoegsayiDoJBCjYkneE= github.com/smartcontractkit/chainlink-protos/billing/go v0.0.0-20251024234028-0988426d98f4/go.mod h1:HHGeDUpAsPa0pmOx7wrByCitjQ0mbUxf0R9v+g67uCA= -github.com/smartcontractkit/chainlink-protos/cre/go v0.0.0-20260622152157-c8e129347b8b h1:VDgJWDipihV9f7M5+d21d1RzSsg5rEv+iI12oN1VQbo= -github.com/smartcontractkit/chainlink-protos/cre/go v0.0.0-20260622152157-c8e129347b8b/go.mod h1:vTFHTCbLui4Vn8fTmAadfE3rdnvfrDwOmMujmW857D0= +github.com/smartcontractkit/chainlink-protos/cre/go v0.0.0-20260629155926-02c54659e848 h1:MWcwR8tuCfQSQfYwn0YD0v8GybYolqYTX29qjooadqM= +github.com/smartcontractkit/chainlink-protos/cre/go v0.0.0-20260629155926-02c54659e848/go.mod h1:/i8hjTPFdVWHiY+QjeSiVS2Z3GB3WAZznGgXHstC02E= github.com/smartcontractkit/chainlink-protos/linking-service/go v0.0.0-20251002192024-d2ad9222409b h1:QuI6SmQFK/zyUlVWEf0GMkiUYBPY4lssn26nKSd/bOM= github.com/smartcontractkit/chainlink-protos/linking-service/go v0.0.0-20251002192024-d2ad9222409b/go.mod h1:qSTSwX3cBP3FKQwQacdjArqv0g6QnukjV4XuzO6UyoY= github.com/smartcontractkit/chainlink-protos/node-platform v0.0.0-20260205130626-db2a2aab956b h1:36knUpKHHAZ86K4FGWXtx8i/EQftGdk2bqCoEu/Cha8= diff --git a/pkg/settings/cresettings/README.md b/pkg/settings/cresettings/README.md index 212bed7456..83b71d7667 100644 --- a/pkg/settings/cresettings/README.md +++ b/pkg/settings/cresettings/README.md @@ -259,6 +259,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] diff --git a/pkg/settings/cresettings/defaults.json b/pkg/settings/cresettings/defaults.json index 8baf65f5fa..ead6935b25 100644 --- a/pkg/settings/cresettings/defaults.json +++ b/pkg/settings/cresettings/defaults.json @@ -63,6 +63,7 @@ "ConfidentialCompute": { "Rate": "1000rps:1000" } + "SuspendOnAwaitEnabled": "false" }, "PerWorkflow": { "TriggerRegistrationsTimeout": "10s", @@ -177,4 +178,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]" } -} \ No newline at end of file +} diff --git a/pkg/settings/cresettings/defaults.toml b/pkg/settings/cresettings/defaults.toml index 6174d14247..ea8fa498d7 100644 --- a/pkg/settings/cresettings/defaults.toml +++ b/pkg/settings/cresettings/defaults.toml @@ -60,6 +60,7 @@ WorkflowLimit = '1000' WorkflowExecutionConcurrencyLimit = '5' VaultCiphertextSizeLimit = '2kb' VaultSecretsLimit = '100' +SuspendOnAwaitEnabled = 'false' [PerOwner.ConfidentialCompute] Rate = '1000rps:1000' diff --git a/pkg/settings/cresettings/settings.go b/pkg/settings/cresettings/settings.go index bc0875149d..8d4c6e8696 100644 --- a/pkg/settings/cresettings/settings.go +++ b/pkg/settings/cresettings/settings.go @@ -155,6 +155,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 @@ -358,7 +360,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 { diff --git a/pkg/workflows/wasm/host/execution.go b/pkg/workflows/wasm/host/execution.go index 6cd16bc4e2..96cf0c4f99 100644 --- a/pkg/workflows/wasm/host/execution.go +++ b/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,62 @@ type execution[T any] struct { nodeSeed int64 donLogCount uint32 nodeLogCount uint32 + awaiting []int32 + // 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 +104,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 +140,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{ diff --git a/pkg/workflows/wasm/host/execution_await_order_test.go b/pkg/workflows/wasm/host/execution_await_order_test.go index f250dd69fc..d5b0b86bd9 100644 --- a/pkg/workflows/wasm/host/execution_await_order_test.go +++ b/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, } diff --git a/pkg/workflows/wasm/host/execution_semaphore_test.go b/pkg/workflows/wasm/host/execution_semaphore_test.go index ed7c8720df..a7b5faba1c 100644 --- a/pkg/workflows/wasm/host/execution_semaphore_test.go +++ b/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, diff --git a/pkg/workflows/wasm/host/module.go b/pkg/workflows/wasm/host/module.go index 2f64dae3da..489e26aeeb 100644 --- a/pkg/workflows/wasm/host/module.go +++ b/pkg/workflows/wasm/host/module.go @@ -136,8 +136,18 @@ type module struct { stopCh chan struct{} v2ImportName string + + // callWasm runs a single guest invocation for the v2 (no-DAG) execution path. + // It is a field so that tests can wrap it - e.g. to perturb the execution + // between the suspended run and the resume - while still exercising the real + // Execute loop. It is initialised in newModule to delegate to the package-level + // callWasm and is invoked by Execute via m.callWasm. + callWasm callWasmFunc } +// callWasmFunc runs one guest invocation for the v2 (no-DAG) execution path. +type callWasmFunc func(timeout time.Duration, req *sdkpb.ExecuteRequest, linkWasm linkFn[*sdkpb.ExecutionResult], exec *execution[*sdkpb.ExecutionResult]) (time.Duration, error) + var _ ModuleV1 = (*module)(nil) type linkFn[T any] func(m *module, store *wasmtime.Store, exec *execution[T]) (*wasmtime.Instance, error) @@ -384,14 +394,18 @@ func newModule(modCfg *ModuleConfig, binary []byte) (*module, error) { modCfg.SdkLabeler(v2ImportName) - return &module{ + m := &module{ engine: engine, module: mod, wconfig: cfg, cfg: modCfg, stopCh: make(chan struct{}), v2ImportName: v2ImportName, - }, nil + } + m.callWasm = func(timeout time.Duration, req *sdkpb.ExecuteRequest, linkWasm linkFn[*sdkpb.ExecutionResult], exec *execution[*sdkpb.ExecutionResult]) (time.Duration, error) { + return callWasm(timeout, m, req, linkWasm, exec) + } + return m, nil } func linkNoDAG(m *module, store *wasmtime.Store, exec *execution[*sdkpb.ExecutionResult]) (*wasmtime.Instance, error) { @@ -577,16 +591,87 @@ func (m *module) Execute(ctx context.Context, req *sdkpb.ExecuteRequest, executo return nil, errors.New("invalid request: can't be nil") } - setMaxResponseSize := func(r *sdkpb.ExecuteRequest, maxSize uint64) { - r.MaxResponseSize = maxSize + maxResponseSizeBytes, err := m.cfg.MaxResponseSizeLimiter.Limit(ctx) + if err != nil { + return nil, fmt.Errorf("failed to get response size limit: %w", err) + } + req.MaxResponseSize = uint64(maxResponseSizeBytes) + + h := fnv.New64a() + executionId := executor.GetWorkflowExecutionID() + _, _ = h.Write([]byte(executionId)) + donSeed := int64(h.Sum64()) + + exec := &execution[*sdkpb.ExecutionResult]{ + capabilityResponses: map[int32]*asyncResponse[sdkpb.CapabilityRequest, sdkpb.CapabilityResponse]{}, + secretsResponses: map[int32]<-chan *secretsResponse{}, + pendingCallsLimiter: m.cfg.PendingCallsLimiter, + module: m, + executor: executor, + donSeed: donSeed, + nodeSeed: int64(rand.Uint64()), + suspendOnAwait: req.SuspendOnAwait, } - timeout := *m.cfg.Timeout + // The overall timeout bounds the entire execution, including any time spent + // suspended while waiting for capability responses. Reuse a single + // deadline-bearing context across every resume so the budget is shared. + overallTimeout := *m.cfg.Timeout switch req.Request.(type) { case *sdkpb.ExecuteRequest_PreHook: - timeout = *m.cfg.PrehookTimeout + overallTimeout = *m.cfg.PrehookTimeout + } + ctxWithTimeout, cancel := context.WithTimeout(ctx, overallTimeout) + exec.ctx = ctxWithTimeout + defer cancel() + overallStart := time.Now() + + for { + // Each (re)start is bounded by the time left in the overall timeout, so a + // resumed run - and the wait for its capability responses - cannot extend + // the workflow's total run time beyond the original deadline. + remaining := overallTimeout - time.Since(overallStart) + if remaining <= 0 { + return nil, context.DeadlineExceeded + } + + executionDuration, err := m.callWasm(remaining, req, linkNoDAG, exec) + + switch { + case containsCode(err, wasm.CodeSuccess): + if any(exec.response) == nil { + return nil, errors.New("could not find response for execution") + } + return exec.response, nil + case isSuspendTrap(err): + m.cfg.Logger.Debugw("received suspension, awaiting responses", "executionID", executionId) + // Wait for the pending capability responses, then resume on the next + // loop iteration with a fresh store. + for _, id := range exec.awaiting { + ar, ok := exec.capabilityResponses[id] + if !ok { + return nil, fmt.Errorf("missing capability response for awaited callback %d", id) + } + + if _, werr := ar.wait(ctxWithTimeout); werr != nil { + return nil, werr + } + } + + continue + default: + // If an error has occurred and the deadline has been reached or exceeded, return a deadline exceeded error. + // Note - there is no other reliable signal on the error that can be used to infer it is due to epoch deadline + // being reached, so if an error is returned after the deadline it is assumed it is due to that and return + // context.DeadlineExceeded. + if err != nil && ((executionDuration >= remaining-m.cfg.TickInterval) || ctx.Err() != nil) { // As start could be called just before epoch update 1 tick interval is deducted to account for this + m.cfg.Logger.Errorw("start function returned error after deadline reached, returning deadline exceeded error", "errFromStartFunction", err) + return nil, context.DeadlineExceeded + } + + return nil, err + } } - return runWasm(ctx, m, req, setMaxResponseSize, linkNoDAG, executor, timeout) } // Run is deprecated, use execute instead @@ -603,77 +688,107 @@ func (m *module) Run(ctx context.Context, request *wasmdagpb.Request) (*wasmdagp return nil, errors.New("cannot use Run on a non-legacy dag workflow, use Execute instead") } - setMaxResponseSize := func(r *wasmdagpb.Request, maxSize uint64) { - computeRequest := r.GetComputeRequest() - if computeRequest != nil { - computeRequest.RuntimeConfig = &wasmdagpb.RuntimeConfig{ - MaxResponseSizeBytes: int64(maxSize), - } - } + maxResponseSizeBytes, err := m.cfg.MaxResponseSizeLimiter.Limit(ctx) + if err != nil { + return nil, fmt.Errorf("failed to get response size limit: %w", err) } - return runWasm(ctx, m, request, setMaxResponseSize, linkLegacyDAG, nil, *m.cfg.Timeout) -} + computeRequest := request.GetComputeRequest() + if computeRequest != nil { + computeRequest.RuntimeConfig = &wasmdagpb.RuntimeConfig{ + MaxResponseSizeBytes: int64(maxResponseSizeBytes), + } + } -func runWasm[I, O proto.Message]( - ctx context.Context, - m *module, - request I, - setMaxResponseSize func(i I, maxSize uint64), - linkWasm linkFn[O], - helper ExecutionHelper, - maxTimeout time.Duration) (O, error) { - var o O + exec := &execution[*wasmdagpb.Response]{ + module: m, + } // No reason to run the WASM longer if the outer ctx will cancel. + // TODO: this should live higher up. ctxDeadline, hasDeadline := ctx.Deadline() var ctxWithTimeout context.Context var cancel func() - - if hasDeadline && ctxDeadline.Before(time.Now().Add(maxTimeout)) { + if hasDeadline && ctxDeadline.Before(time.Now().Add(*m.cfg.Timeout)) { ctxWithTimeout, cancel = context.WithCancel(ctx) } else { - ctxWithTimeout, cancel = context.WithTimeout(ctx, maxTimeout) + ctxWithTimeout, cancel = context.WithTimeout(ctx, *m.cfg.Timeout) } - + exec.ctx = ctxWithTimeout defer cancel() - store := wasmtime.NewStore(m.engine) - - defer store.Close() + execDuration, err := callWasm(*m.cfg.Timeout, m, request, linkLegacyDAG, exec) - maxResponseSizeBytes, err := m.cfg.MaxResponseSizeLimiter.Limit(ctx) - if err != nil { - return o, fmt.Errorf("failed to get response size limit: %w", err) + switch { + case containsCode(err, wasm.CodeSuccess): + if exec.response == nil { + return nil, errors.New("could not find response for execution") + } + return exec.response, nil + case containsCode(err, wasm.CodeInvalidResponse): + return nil, errors.New("invariant violation: error marshaling response") + case containsCode(err, wasm.CodeInvalidRequest): + return nil, errors.New("invariant violation: invalid request to runner") + case containsCode(err, wasm.CodeRunnerErr): + // legacy DAG captured all errors, since the function didn't return an error + resp, ok := any(exec).(*execution[*wasmdagpb.Response]) + if ok && resp.response != nil { + return nil, fmt.Errorf("error executing runner: %s: %w", resp.response.ErrMsg, err) + } + return nil, errors.New("error executing runner") + case containsCode(err, wasm.CodeHostErr): + return nil, errors.New("invariant violation: host errored during sendResponse") } - setMaxResponseSize(request, uint64(maxResponseSizeBytes)) - reqpb, err := proto.Marshal(request) - if err != nil { - return o, err + + // If an error has occurred and the deadline has been reached or exceeded, return a deadline exceeded error. + // Note - there is no other reliable signal on the error that can be used to infer it is due to epoch deadline + // being reached, so if an error is returned after the deadline it is assumed it is due to that and return + // context.DeadlineExceeded. + if err != nil && ((execDuration >= *m.cfg.Timeout-m.cfg.TickInterval) || ctx.Err() != nil) { // As start could be called just before epoch update 1 tick interval is deducted to account for this + m.cfg.Logger.Errorw("start function returned error after deadline reached, returning deadline exceeded error", "errFromStartFunction", err) + return nil, context.DeadlineExceeded } - reqstr := base64.StdEncoding.EncodeToString(reqpb) + return nil, err +} + +// callWasm performs a single guest invocation in a fresh, self-contained store. +// The store - and therefore the instance and its linear memory - is closed +// before callWasm returns; everything that must survive a resume lives on exec. +// It returns the error from _start (which encodes the guest exit code, or the +// suspend trap raised by await_capabilities). +func callWasm[I, O proto.Message]( + timeout time.Duration, + m *module, + req I, + linkWasm linkFn[O], + exec *execution[O], +) (time.Duration, error) { + store := wasmtime.NewStore(m.engine) + defer store.Close() wasi := wasmtime.NewWasiConfig() wasi.InheritStdout() defer wasi.Close() - wasi.SetArgv([]string{"wasi", reqstr}) + reqpb, err := proto.Marshal(req) + if err != nil { + return 0, nil + } - store.SetWasi(wasi) + reqstr := base64.StdEncoding.EncodeToString(reqpb) - if m.cfg.InitialFuel > 0 { - err = store.SetFuel(m.cfg.InitialFuel) - if err != nil { - return o, fmt.Errorf("error setting fuel: %w", err) - } - } + wasi.SetArgv([]string{"wasi", reqstr}) + store.SetWasi(wasi) // Limit memory to max memory megabytes per instance. - maxMemoryBytes, err := m.cfg.MemoryLimiter.Limit(ctx) + maxMemoryBytes, err := m.cfg.MemoryLimiter.Limit(exec.ctx) if err != nil { - return o, fmt.Errorf("failed to get memory limit: %w", err) + return 0, fmt.Errorf("failed to get memory limit: %w", err) } + + // A fresh store is used per (re)start, so a single instance/table/memory is + // sufficient; the previous run's store has already been closed. store.Limiter( int64(maxMemoryBytes/config.MByte)*int64(math.Pow(10, 6)), -1, // tableElements, -1 == default @@ -682,74 +797,23 @@ func runWasm[I, O proto.Message]( 1, // memories ) - deadline := maxTimeout / m.cfg.TickInterval + deadline := timeout / m.cfg.TickInterval store.SetEpochDeadline(uint64(deadline)) - h := fnv.New64a() - if helper != nil { - executionId := helper.GetWorkflowExecutionID() - _, _ = h.Write([]byte(executionId)) - } - - donSeed := int64(h.Sum64()) - - exec := &execution[O]{ - ctx: ctxWithTimeout, - capabilityResponses: map[int32]<-chan *sdkpb.CapabilityResponse{}, - secretsResponses: map[int32]<-chan *secretsResponse{}, - pendingCallsLimiter: m.cfg.PendingCallsLimiter, - module: m, - executor: helper, - donSeed: donSeed, - nodeSeed: int64(rand.Uint64()), - } - instance, err := linkWasm(m, store, exec) if err != nil { - return o, fmt.Errorf("error linking wasm: %w", err) + return 0, fmt.Errorf("error linking wasm: %w", err) } start := instance.GetFunc(store, "_start") if start == nil { - return o, errors.New("could not get start function") + return 0, errors.New("could not get start function") } startTime := time.Now() _, err = start.Call(store) executionDuration := time.Since(startTime) - - // The error codes below are only returned by the v1 legacy DAG workflow. - switch { - case containsCode(err, wasm.CodeSuccess): - if any(exec.response) == nil { - return o, errors.New("could not find response for execution") - } - return exec.response, nil - case containsCode(err, wasm.CodeInvalidResponse): - return o, errors.New("invariant violation: error marshaling response") - case containsCode(err, wasm.CodeInvalidRequest): - return o, errors.New("invariant violation: invalid request to runner") - case containsCode(err, wasm.CodeRunnerErr): - // legacy DAG captured all errors, since the function didn't return an error - resp, ok := any(exec).(*execution[*wasmdagpb.Response]) - if ok && resp.response != nil { - return o, fmt.Errorf("error executing runner: %s: %w", resp.response.ErrMsg, err) - } - return o, errors.New("error executing runner") - case containsCode(err, wasm.CodeHostErr): - return o, errors.New("invariant violation: host errored during sendResponse") - } - - // If an error has occurred and the deadline has been reached or exceeded, return a deadline exceeded error. - // Note - there is no other reliable signal on the error that can be used to infer it is due to epoch deadline - // being reached, so if an error is returned after the deadline it is assumed it is due to that and return - // context.DeadlineExceeded. - if err != nil && ((executionDuration >= maxTimeout-m.cfg.TickInterval) || ctx.Err() != nil) { // As start could be called just before epoch update 1 tick interval is deducted to account for this - m.cfg.Logger.Errorw("start function returned error after deadline reached, returning deadline exceeded error", "errFromStartFunction", err) - return o, context.DeadlineExceeded - } - - return o, err + return executionDuration, err } func containsCode(err error, code int) bool { @@ -759,6 +823,13 @@ func containsCode(err error, code int) bool { return strings.Contains(err.Error(), fmt.Sprintf("exit status %d", code)) } +// isSuspendTrap reports whether err is the trap raised by await_capabilities to +// suspend the execution (see createAwaitCapsFn). The trap message round-trips +// through start.Call as the returned error's message. +func isSuspendTrap(err error) bool { + return err != nil && strings.Contains(err.Error(), errSuspendExecution.Error()) +} + // createSendResponseFn injects the dependency required by a WASM guest to // send a response back to the host. func createSendResponseFn[T proto.Message]( @@ -1223,13 +1294,13 @@ func createCallCapFn( func createAwaitCapsFn( logger logger.Logger, exec *execution[*sdkpb.ExecutionResult], -) func(caller *wasmtime.Caller, awaitRequest, awaitRequestLen, responseBuffer, maxResponseLen int32) int64 { - return func(caller *wasmtime.Caller, awaitRequest, awaitRequestLen, responseBuffer, maxResponseLen int32) int64 { +) func(caller *wasmtime.Caller, awaitRequest, awaitRequestLen, responseBuffer, maxResponseLen int32) (int64, *wasmtime.Trap) { + return func(caller *wasmtime.Caller, awaitRequest, awaitRequestLen, responseBuffer, maxResponseLen int32) (int64, *wasmtime.Trap) { b, err := wasmRead(caller, awaitRequest, awaitRequestLen) if err != nil { errStr := fmt.Sprintf("error reading from wasm %s", err) logger.Error(errStr) - return truncateWasmWrite(caller, []byte(errStr), responseBuffer, maxResponseLen) + return truncateWasmWrite(caller, []byte(errStr), responseBuffer, maxResponseLen), nil } req := &sdkpb.AwaitCapabilitiesRequest{} @@ -1237,31 +1308,37 @@ func createAwaitCapsFn( if err != nil { errStr := err.Error() logger.Error(errStr) - return truncateWasmWrite(caller, []byte(errStr), responseBuffer, maxResponseLen) + return truncateWasmWrite(caller, []byte(errStr), responseBuffer, maxResponseLen), nil } resp, err := exec.awaitCapabilities(exec.ctx, req) - if err != nil { + switch { + case errors.Is(err, errSuspendExecution): + // awaitCapabilities has recorded exec.awaiting. Trap to unwind the + // guest immediately; the host detects this trap (isSuspendTrap), waits + // for the pending responses, and resumes by re-instantiating the guest. + return 0, wasmtime.NewTrap(errSuspendExecution.Error()) + case err != nil: errStr := err.Error() logger.Error(errStr) - return truncateWasmWrite(caller, []byte(errStr), responseBuffer, maxResponseLen) + return truncateWasmWrite(caller, []byte(errStr), responseBuffer, maxResponseLen), nil } respBytes, err := proto.Marshal(resp) if err != nil { errStr := err.Error() logger.Error(errStr) - return truncateWasmWrite(caller, []byte(errStr), responseBuffer, maxResponseLen) + return truncateWasmWrite(caller, []byte(errStr), responseBuffer, maxResponseLen), nil } size := wasmWrite(caller, respBytes, responseBuffer, maxResponseLen) if size == -1 { errStr := ResponseBufferTooSmall logger.Error(errStr) - return truncateWasmWrite(caller, []byte(errStr), responseBuffer, maxResponseLen) + return truncateWasmWrite(caller, []byte(errStr), responseBuffer, maxResponseLen), nil } - return size + return size, nil } } diff --git a/pkg/workflows/wasm/host/module_test.go b/pkg/workflows/wasm/host/module_test.go index bfa1818576..aca29eea98 100644 --- a/pkg/workflows/wasm/host/module_test.go +++ b/pkg/workflows/wasm/host/module_test.go @@ -636,7 +636,7 @@ func Test_CallAwaitRace(t *testing.T) { exec := &execution[*wasmpb.ExecutionResult]{ module: m, - capabilityResponses: map[int32]<-chan *sdkpb.CapabilityResponse{}, + capabilityResponses: map[int32]*asyncResponse[sdkpb.CapabilityRequest, sdkpb.CapabilityResponse]{}, pendingCallsLimiter: limits.GlobalResourcePoolLimiter(cresettings.Default.PerWorkflow.CapabilityConcurrencyLimit.DefaultValue), ctx: t.Context(), executor: mockExecHelper, diff --git a/pkg/workflows/wasm/host/standard_tests/non_deterministic_input/main_wasip1.go b/pkg/workflows/wasm/host/standard_tests/non_deterministic_input/main_wasip1.go new file mode 100644 index 0000000000..6c83bdfbcc --- /dev/null +++ b/pkg/workflows/wasm/host/standard_tests/non_deterministic_input/main_wasip1.go @@ -0,0 +1,27 @@ +package main + +import ( + "github.com/smartcontractkit/chainlink-common/pkg/capabilities/v2/protoc/pkg/test_capabilities/basicaction" + "github.com/smartcontractkit/chainlink-common/pkg/workflows/wasm/host/internal/rawsdk" + "github.com/smartcontractkit/chainlink-protos/cre/go/sdk" +) + +// This workflow derives its capability request input from the host-provided DON +// random seed. The suspend/resume integrity check requires a resumed (replayed) +// execution to issue exactly the same capability requests as the original run. +// The accompanying test changes the seed between the suspended run and the +// resume, so the replay builds a different request for the same callback id and +// the host rejects it as non-deterministic. +func main() { + seed := rawsdk.GetSeed(int32(sdk.Mode_MODE_DON)) + + // The request payload depends on the seed, so changing the seed across a + // replay produces a different request. + input := &basicaction.Inputs{InputThing: seed%2 == 0} + id := rawsdk.DoRequestAsync("basic-test-action@1.0.0", "PerformAction", sdk.Mode_MODE_DON, input) + + result := &basicaction.Outputs{} + rawsdk.Await(id, result) + + rawsdk.SendResponse(result.AdaptedThing) +} diff --git a/pkg/workflows/wasm/host/standard_tests/suspended_executions/main_wasip1.go b/pkg/workflows/wasm/host/standard_tests/suspended_executions/main_wasip1.go new file mode 100644 index 0000000000..41bc8d2b8b --- /dev/null +++ b/pkg/workflows/wasm/host/standard_tests/suspended_executions/main_wasip1.go @@ -0,0 +1,40 @@ +package main + +import ( + "github.com/smartcontractkit/chainlink-common/pkg/capabilities/v2/protoc/pkg/test_capabilities/basicaction" + "github.com/smartcontractkit/chainlink-common/pkg/workflows/wasm/host/internal/rawsdk" + "github.com/smartcontractkit/chainlink-protos/cre/go/sdk" +) + +// This workflow exercises the host's suspend/resume behaviour. It dispatches a +// single async capability call and awaits it. When suspension is enabled the +// host has no response available at the await, so it suspends the guest; the +// guest re-runs main() from the top once the host resumes it with the response. +// +// A log line is emitted on every run so the test can observe how many times the +// guest executed: once when suspension is disabled (the await blocks in the +// host), twice when it is enabled (the initial run that suspends plus the +// resumed run that completes). +func main() { + // Switch to DON mode so the host attributes (and does not drop) the log below. + rawsdk.SwitchModes(int32(sdk.Mode_MODE_DON)) + logRun() + + id := rawsdk.DoRequestAsync( + "basic-test-action@1.0.0", + "PerformAction", + sdk.Mode_MODE_DON, + &basicaction.Inputs{InputThing: true}, + ) + + result := &basicaction.Outputs{} + rawsdk.Await(id, result) + + rawsdk.SendResponse(result.AdaptedThing) +} + +func logRun() { + msg := []byte("suspended_executions:run") + ptr, length := rawsdk.BufferToPointerLen(msg) + rawsdk.Log(ptr, length) +} diff --git a/pkg/workflows/wasm/host/wasm_nondeterminism_test.go b/pkg/workflows/wasm/host/wasm_nondeterminism_test.go new file mode 100644 index 0000000000..d087d183f2 --- /dev/null +++ b/pkg/workflows/wasm/host/wasm_nondeterminism_test.go @@ -0,0 +1,94 @@ +package host + +import ( + "context" + "testing" + "time" + + "github.com/stretchr/testify/mock" + "github.com/stretchr/testify/require" + "go.uber.org/zap/zapcore" + "google.golang.org/protobuf/types/known/anypb" + + "github.com/smartcontractkit/chainlink-common/pkg/capabilities/v2/protoc/pkg/test_capabilities/basicaction" + "github.com/smartcontractkit/chainlink-common/pkg/capabilities/v2/protoc/pkg/test_capabilities/basictrigger" + "github.com/smartcontractkit/chainlink-common/pkg/logger" + "github.com/smartcontractkit/chainlink-common/pkg/settings/cresettings" + "github.com/smartcontractkit/chainlink-common/pkg/settings/limits" + "github.com/smartcontractkit/chainlink-common/pkg/workflows/host/mocks" + sdkpb "github.com/smartcontractkit/chainlink-protos/cre/go/sdk" +) + +// TestStandardNonDeterministicInput verifies that an execution which produces a +// different capability request on resume than on its original run (i.e. a +// non-deterministic workflow) is rejected by the host's integrity check. +// +// The standard test guest builds its capability request from the DON random +// seed. We run the real Execute loop but wrap module.callWasm so that, just +// before the resume, we flip exec.donSeed. The replayed run then issues a +// different request for the same callback id, which fails the proto.Equal +// integrity check in callCapAsync, surfaces to the guest as a failed capability +// call, and ends the execution in error. +func TestStandardNonDeterministicInput(t *testing.T) { + t.Parallel() + + lggr, observed := logger.TestObserved(t, zapcore.ErrorLevel) + cfg := &ModuleConfig{ + Logger: lggr, + IsUncompressed: true, + SuspensionEnabled: true, + PendingCallsLimiter: limits.GlobalResourcePoolLimiter(cresettings.Default.PerWorkflow.CapabilityConcurrencyLimit.DefaultValue), + } + + m := makeTestModuleByName(t, testPath, "non_deterministic_input", cfg, false) + m.Start() + defer m.Close() + + helper := mocks.NewMockExecutionHelper(t) + helper.EXPECT().GetWorkflowExecutionID().Return("id") + helper.EXPECT().GetNodeTime().RunAndReturn(func() time.Time { + return time.Now() + }).Maybe() + helper.EXPECT().GetDONTime().RunAndReturn(func() (time.Time, error) { + return time.Now(), nil + }).Maybe() + helper.EXPECT().CallCapability(mock.Anything, mock.Anything).RunAndReturn(func(_ context.Context, _ *sdkpb.CapabilityRequest) (*sdkpb.CapabilityResponse, error) { + payload, perr := anypb.New(&basicaction.Outputs{AdaptedThing: "ok"}) + require.NoError(t, perr) + return &sdkpb.CapabilityResponse{ + Response: &sdkpb.CapabilityResponse_Payload{Payload: payload}, + }, nil + }).Maybe() + + // Wrap callWasm so the workflow becomes non-deterministic across the resume: + // flip the DON seed before the second (resume) invocation, so the replayed + // run builds a different capability request than the original. Flipping by one + // guarantees the seed's parity - and therefore the request's input - changes. + original := m.callWasm + var runs int + m.callWasm = func(ctx context.Context, timeout time.Duration, req *sdkpb.ExecuteRequest, linkWasm linkFn[*sdkpb.ExecutionResult], exec *execution[*sdkpb.ExecutionResult]) error { + runs++ + if runs == 2 { + exec.donSeed++ + } + return original(ctx, timeout, req, linkWasm, exec) + } + + req := triggerExecuteRequest(t, 0, &basictrigger.Outputs{CoolOutput: anyTestTriggerValue}) + result, err := m.Execute(t.Context(), req, helper) + require.NoError(t, err) + + // The execution ends in error. The exact message is SDK-specific (the rawsdk + // reports "callCapability returned an error", cre-sdk-go "cannot find + // capability ..."), so we only assert that it is an error result; the + // determinism violation itself is asserted via the host log below. + errResult, ok := result.Result.(*sdkpb.ExecutionResult_Error) + require.True(t, ok, "expected an error result, got %T", result.Result) + require.NotEmpty(t, errResult.Error) + + // The execution suspended once and resumed once, and the host logged the + // underlying determinism violation. + require.Equal(t, 2, runs, "expected the guest to run twice: the suspended run and the resumed run") + require.NotEmpty(t, observed.FilterMessageSnippet("non-determinism error").All(), + "expected the host to log a non-determinism error") +} diff --git a/pkg/workflows/wasm/host/wasm_suspend_test.go b/pkg/workflows/wasm/host/wasm_suspend_test.go new file mode 100644 index 0000000000..56f92cf9ba --- /dev/null +++ b/pkg/workflows/wasm/host/wasm_suspend_test.go @@ -0,0 +1,94 @@ +package host + +import ( + "context" + "strings" + "sync/atomic" + "testing" + "time" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/mock" + "github.com/stretchr/testify/require" + "google.golang.org/protobuf/types/known/anypb" + + "github.com/smartcontractkit/chainlink-common/pkg/capabilities/v2/protoc/pkg/test_capabilities/basicaction" + "github.com/smartcontractkit/chainlink-common/pkg/capabilities/v2/protoc/pkg/test_capabilities/basictrigger" + "github.com/smartcontractkit/chainlink-common/pkg/workflows/host/mocks" + "github.com/smartcontractkit/chainlink-protos/cre/go/sdk" +) + +// TestStandardSuspendedExecutions runs a real rawsdk workflow that dispatches an +// async capability call and awaits it, and asserts the host's suspend/resume +// behaviour end-to-end: +// +// - With suspension disabled the await blocks in the host, so the guest runs +// exactly once (one log line). +// - With suspension enabled the host has no response at the await, so it +// suspends the guest and resumes it once the response is available. The +// guest therefore runs twice (two log lines), while the capability itself is +// dispatched only once - the second run replays the recorded call from the +// store rather than re-invoking the capability. +func TestStandardSuspendedExecutions(t *testing.T) { + t.Parallel() + + const adaptedThing = "adapted-thing" + const runLogMessage = "suspended_executions:run" + + run := func(t *testing.T, suspensionEnabled bool) (result string, capabilityCalls, runs int32) { + var capabilityCallCount, runCount atomic.Int32 + + helper := mocks.NewMockExecutionHelper(t) + helper.EXPECT().GetWorkflowExecutionID().Return("id") + helper.EXPECT().GetNodeTime().RunAndReturn(func() time.Time { + return time.Now() + }).Maybe() + helper.EXPECT().GetDONTime().RunAndReturn(func() (time.Time, error) { + return time.Now(), nil + }).Maybe() + helper.EXPECT().EmitUserLog(mock.Anything).RunAndReturn(func(s string) error { + // Match on Contains rather than equality: the rawsdk emits the raw + // message, while other SDKs (e.g. cre-sdk-go) emit a formatted log line + // that contains it. + if strings.Contains(s, runLogMessage) { + runCount.Add(1) + } + return nil + }) + helper.EXPECT().CallCapability(mock.Anything, mock.Anything).RunAndReturn(func(_ context.Context, req *sdk.CapabilityRequest) (*sdk.CapabilityResponse, error) { + capabilityCallCount.Add(1) + assert.Equal(t, "basic-test-action@1.0.0", req.Id) + assert.Equal(t, "PerformAction", req.Method) + + payload, err := anypb.New(&basicaction.Outputs{AdaptedThing: adaptedThing}) + require.NoError(t, err) + return &sdk.CapabilityResponse{ + Response: &sdk.CapabilityResponse_Payload{Payload: payload}, + }, nil + }) + + cfg := defaultNoDAGModCfg(t) + cfg.SuspensionEnabled = suspensionEnabled + m := makeTestModuleByName(t, testPath, "suspended_executions", cfg, false) + m.Start() + defer m.Close() + + req := triggerExecuteRequest(t, 0, &basictrigger.Outputs{CoolOutput: anyTestTriggerValue}) + result = executeWithResult[string](t, m, req, helper) + return result, capabilityCallCount.Load(), runCount.Load() + } + + t.Run("suspension disabled blocks in the host and runs once", func(t *testing.T) { + result, capabilityCalls, runs := run(t, false) + assert.Equal(t, adaptedThing, result) + assert.Equal(t, int32(1), capabilityCalls, "capability should be dispatched once") + assert.Equal(t, int32(1), runs, "guest should run exactly once when suspension is disabled") + }) + + t.Run("suspension enabled suspends on await and resumes with the response", func(t *testing.T) { + result, capabilityCalls, runs := run(t, true) + assert.Equal(t, adaptedThing, result) + assert.Equal(t, int32(1), capabilityCalls, "capability should be dispatched once and replayed from the store on resume") + assert.Equal(t, int32(2), runs, "guest should run twice: the suspended run and the resumed run") + }) +} diff --git a/pkg/workflows/wasm/host/wasm_test.go b/pkg/workflows/wasm/host/wasm_test.go index cc1db9086a..582054343d 100644 --- a/pkg/workflows/wasm/host/wasm_test.go +++ b/pkg/workflows/wasm/host/wasm_test.go @@ -672,7 +672,9 @@ func Test_Compute_Fetch(t *testing.T) { cancel() _, err = m.Run(ctx, req) require.Error(t, err) - assert.ErrorContains(t, err, fmt.Sprintf("error executing runner: error executing custom compute: %s", assert.AnError)) + // The context is cancelled before Run, so the cancellation is reported as + // a deadline-exceeded error rather than surfacing the guest's runner error. + require.ErrorIs(t, err, context.DeadlineExceeded) }) t.Run("NOK: exceeded maximum fetch calls", func(t *testing.T) { From 5ef1f456ca14d2a10cb0401a756217d5a055a5a5 Mon Sep 17 00:00:00 2001 From: Cedric Cordenier Date: Wed, 1 Jul 2026 11:33:54 +0100 Subject: [PATCH 2/6] Add metrics --- pkg/workflows/wasm/host/execution.go | 3 + pkg/workflows/wasm/host/metrics.go | 138 +++++++++++++++++++++++++++ pkg/workflows/wasm/host/module.go | 70 ++++++++++++-- 3 files changed, 203 insertions(+), 8 deletions(-) create mode 100644 pkg/workflows/wasm/host/metrics.go diff --git a/pkg/workflows/wasm/host/execution.go b/pkg/workflows/wasm/host/execution.go index 96cf0c4f99..a918f84406 100644 --- a/pkg/workflows/wasm/host/execution.go +++ b/pkg/workflows/wasm/host/execution.go @@ -40,6 +40,9 @@ type execution[T any] struct { 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 diff --git a/pkg/workflows/wasm/host/metrics.go b/pkg/workflows/wasm/host/metrics.go new file mode 100644 index 0000000000..747c59565a --- /dev/null +++ b/pkg/workflows/wasm/host/metrics.go @@ -0,0 +1,138 @@ +package host + +import ( + "context" + "fmt" + "time" + + "go.opentelemetry.io/otel/attribute" + "go.opentelemetry.io/otel/metric" + + "github.com/smartcontractkit/chainlink-common/pkg/beholder" +) + +// Execution phase labels for the execution duration histogram: +// - phaseWasm: wall-clock time executing guest wasm code inside callWasm. +// - phaseWaiting: wall-clock time suspended, waiting for pending capability +// responses. +// - phaseTotal: total wall-clock time for the execution, end to end. +const ( + phaseWasm = "wasm" + phaseWaiting = "waiting" + phaseTotal = "total" +) + +// moduleMetrics holds the beholder instruments used to observe wasm module +// executions. Instrument names are shared process-wide, so multiple modules can +// safely construct their own moduleMetrics; the meter returns the same +// underlying instrument for a given name. +type moduleMetrics struct { + activeExecutions metric.Int64UpDownCounter + suspendedExecutions metric.Int64UpDownCounter + suspensionsPerExec metric.Int64Histogram + executionDurationMs metric.Int64Histogram + memoryBytes metric.Int64Histogram +} + +func newModuleMetrics() (*moduleMetrics, error) { + meter := beholder.GetMeter() + + activeExecutions, err := meter.Int64UpDownCounter("platform_wasm_host_active_executions", + metric.WithDescription("Number of wasm module executions currently running"), + metric.WithUnit("{execution}"), + ) + if err != nil { + return nil, fmt.Errorf("failed to create active_executions counter: %w", err) + } + + suspendedExecutions, err := meter.Int64UpDownCounter("platform_wasm_host_suspended_executions", + metric.WithDescription("Number of wasm module executions currently suspended waiting for capability responses"), + metric.WithUnit("{execution}"), + ) + if err != nil { + return nil, fmt.Errorf("failed to create suspended_executions counter: %w", err) + } + + suspensionsPerExec, err := meter.Int64Histogram("platform_wasm_host_suspensions_per_execution", + metric.WithDescription("Number of times an execution suspended to await capability responses before completing"), + metric.WithUnit("{suspension}"), + metric.WithExplicitBucketBoundaries(0, 1, 2, 3, 5, 10, 20, 50, 100), + ) + if err != nil { + return nil, fmt.Errorf("failed to create suspensions_per_execution histogram: %w", err) + } + + executionDurationMs, err := meter.Int64Histogram("platform_wasm_host_execution_duration_ms", + metric.WithDescription("Wall-clock time spent in an execution, by phase (wasm, waiting) plus the end-to-end total"), + metric.WithUnit("ms"), + metric.WithExplicitBucketBoundaries(1, 5, 10, 50, 100, 250, 500, 1_000, 2_000, 5_000, 10_000, 30_000, 60_000, 120_000, 300_000, 600_000), + ) + if err != nil { + return nil, fmt.Errorf("failed to create execution_duration_ms histogram: %w", err) + } + + memoryBytes, err := meter.Int64Histogram("platform_wasm_host_memory_bytes", + metric.WithDescription("Peak linear memory in bytes used by the wasm module across an execution"), + metric.WithUnit("By"), + metric.WithExplicitBucketBoundaries(1<<20, 4<<20, 16<<20, 32<<20, 64<<20, 128<<20, 256<<20, 512<<20, 1<<30), + ) + if err != nil { + return nil, fmt.Errorf("failed to create memory_bytes histogram: %w", err) + } + + return &moduleMetrics{ + activeExecutions: activeExecutions, + suspendedExecutions: suspendedExecutions, + suspensionsPerExec: suspensionsPerExec, + executionDurationMs: executionDurationMs, + memoryBytes: memoryBytes, + }, nil +} + +// suspensionEnabledAttr tags a metric with whether the execution has +// suspend/resume-on-await enabled, so the two populations can be distinguished. +func suspensionEnabledAttr(suspensionEnabled bool) attribute.KeyValue { + return attribute.Bool("suspension_enabled", suspensionEnabled) +} + +// IncActiveExecutions marks an execution as started (a). +func (m *moduleMetrics) IncActiveExecutions(ctx context.Context, suspensionEnabled bool) { + m.activeExecutions.Add(ctx, 1, metric.WithAttributes(suspensionEnabledAttr(suspensionEnabled))) +} + +// DecActiveExecutions marks an execution as finished (a). +func (m *moduleMetrics) DecActiveExecutions(ctx context.Context, suspensionEnabled bool) { + m.activeExecutions.Add(ctx, -1, metric.WithAttributes(suspensionEnabledAttr(suspensionEnabled))) +} + +// IncSuspendedExecutions marks an execution as suspended, waiting for capability +// responses (b). +func (m *moduleMetrics) IncSuspendedExecutions(ctx context.Context, suspensionEnabled bool) { + m.suspendedExecutions.Add(ctx, 1, metric.WithAttributes(suspensionEnabledAttr(suspensionEnabled))) +} + +// DecSuspendedExecutions marks a suspended execution as resumed (b). +func (m *moduleMetrics) DecSuspendedExecutions(ctx context.Context, suspensionEnabled bool) { + m.suspendedExecutions.Add(ctx, -1, metric.WithAttributes(suspensionEnabledAttr(suspensionEnabled))) +} + +// RecordSuspensions records how many times an execution suspended before +// completing (c). +func (m *moduleMetrics) RecordSuspensions(ctx context.Context, suspensionEnabled bool, suspensions int64) { + m.suspensionsPerExec.Record(ctx, suspensions, metric.WithAttributes(suspensionEnabledAttr(suspensionEnabled))) +} + +// RecordExecutionPhase records the wall-clock time spent in a single phase of an +// execution (d). phase is one of phaseWasm, phaseWaiting or phaseTotal. +func (m *moduleMetrics) RecordExecutionPhase(ctx context.Context, suspensionEnabled bool, phase string, d time.Duration) { + m.executionDurationMs.Record(ctx, d.Milliseconds(), + metric.WithAttributes(suspensionEnabledAttr(suspensionEnabled), attribute.String("phase", phase)), + ) +} + +// RecordMemory records the peak linear memory used by an execution (e). Note +// that the CPU-time counterpart of (e) is measured as wall-clock time spent in +// wasm - the phaseWasm bucket of the execution duration histogram (d). +func (m *moduleMetrics) RecordMemory(ctx context.Context, suspensionEnabled bool, memoryBytes int64) { + m.memoryBytes.Record(ctx, memoryBytes, metric.WithAttributes(suspensionEnabledAttr(suspensionEnabled))) +} diff --git a/pkg/workflows/wasm/host/module.go b/pkg/workflows/wasm/host/module.go index 489e26aeeb..7d1aba8e70 100644 --- a/pkg/workflows/wasm/host/module.go +++ b/pkg/workflows/wasm/host/module.go @@ -132,6 +132,8 @@ type module struct { cfg *ModuleConfig + metrics *moduleMetrics + wg sync.WaitGroup stopCh chan struct{} @@ -394,11 +396,17 @@ func newModule(modCfg *ModuleConfig, binary []byte) (*module, error) { modCfg.SdkLabeler(v2ImportName) + metrics, err := newModuleMetrics() + if err != nil { + return nil, fmt.Errorf("error creating module metrics: %w", err) + } + m := &module{ engine: engine, module: mod, wconfig: cfg, cfg: modCfg, + metrics: metrics, stopCh: make(chan struct{}), v2ImportName: v2ImportName, } @@ -626,6 +634,27 @@ func (m *module) Execute(ctx context.Context, req *sdkpb.ExecuteRequest, executo defer cancel() overallStart := time.Now() + m.metrics.IncActiveExecutions(ctx, req.SuspendOnAwait) + defer m.metrics.DecActiveExecutions(ctx, req.SuspendOnAwait) + + // Accumulated across (re)starts and emitted once the execution completes, + // regardless of outcome. wasmDuration is the wall-clock time spent executing + // guest code (which is also the CPU-time signal), waitDuration the time spent + // suspended waiting for capability responses, and suspensions the number of + // times the execution suspended. + var ( + wasmDuration time.Duration + waitDuration time.Duration + suspensions int64 + ) + defer func() { + m.metrics.RecordExecutionPhase(ctx, req.SuspendOnAwait, phaseWasm, wasmDuration) + m.metrics.RecordExecutionPhase(ctx, req.SuspendOnAwait, phaseWaiting, waitDuration) + m.metrics.RecordExecutionPhase(ctx, req.SuspendOnAwait, phaseTotal, time.Since(overallStart)) + m.metrics.RecordSuspensions(ctx, req.SuspendOnAwait, suspensions) + m.metrics.RecordMemory(ctx, req.SuspendOnAwait, exec.peakMemoryBytes) + }() + for { // Each (re)start is bounded by the time left in the overall timeout, so a // resumed run - and the wait for its capability responses - cannot extend @@ -636,6 +665,7 @@ func (m *module) Execute(ctx context.Context, req *sdkpb.ExecuteRequest, executo } executionDuration, err := m.callWasm(remaining, req, linkNoDAG, exec) + wasmDuration += executionDuration switch { case containsCode(err, wasm.CodeSuccess): @@ -645,17 +675,30 @@ func (m *module) Execute(ctx context.Context, req *sdkpb.ExecuteRequest, executo return exec.response, nil case isSuspendTrap(err): m.cfg.Logger.Debugw("received suspension, awaiting responses", "executionID", executionId) + suspensions++ + // The execution is suspended for as long as we wait for its pending + // capability responses; track it as such for the duration of the wait. + m.metrics.IncSuspendedExecutions(ctx, req.SuspendOnAwait) + waitStart := time.Now() // Wait for the pending capability responses, then resume on the next // loop iteration with a fresh store. - for _, id := range exec.awaiting { - ar, ok := exec.capabilityResponses[id] - if !ok { - return nil, fmt.Errorf("missing capability response for awaited callback %d", id) - } - - if _, werr := ar.wait(ctxWithTimeout); werr != nil { - return nil, werr + werr := func() error { + for _, id := range exec.awaiting { + ar, ok := exec.capabilityResponses[id] + if !ok { + return fmt.Errorf("missing capability response for awaited callback %d", id) + } + + if _, werr := ar.wait(ctxWithTimeout); werr != nil { + return werr + } } + return nil + }() + waitDuration += time.Since(waitStart) + m.metrics.DecSuspendedExecutions(ctx, req.SuspendOnAwait) + if werr != nil { + return nil, werr } continue @@ -813,6 +856,17 @@ func callWasm[I, O proto.Message]( startTime := time.Now() _, err = start.Call(store) executionDuration := time.Since(startTime) + + // Capture the linear memory the guest grew to before the store is closed, so + // Execute can emit it as the peak memory metric. + if mem := instance.GetExport(store, "memory"); mem != nil { + if memory := mem.Memory(); memory != nil { + if used := int64(memory.DataSize(store)); used > exec.peakMemoryBytes { + exec.peakMemoryBytes = used + } + } + } + return executionDuration, err } From 7bfe0de361c57196d843f3144ecc3d9d5c0caef9 Mon Sep 17 00:00:00 2001 From: Cedric Cordenier Date: Fri, 3 Jul 2026 16:23:39 +0100 Subject: [PATCH 3/6] Test: serialize module --- pkg/workflows/wasm/host/module.go | 58 ++++++++++++++++++++++++++----- 1 file changed, 49 insertions(+), 9 deletions(-) diff --git a/pkg/workflows/wasm/host/module.go b/pkg/workflows/wasm/host/module.go index 7d1aba8e70..02e221b218 100644 --- a/pkg/workflows/wasm/host/module.go +++ b/pkg/workflows/wasm/host/module.go @@ -12,6 +12,8 @@ import ( "io" "math" "math/rand" + "os" + "path/filepath" "regexp" "strings" "sync" @@ -127,9 +129,16 @@ type ExecutionHelper = host.ExecutionHelper type module struct { engine *wasmtime.Engine - module *wasmtime.Module wconfig *wasmtime.Config + // The compiled module is not kept resident for the lifetime of the module + // instance. Instead it is compiled once, serialized to a file in moduleDir, + // and the in-memory *wasmtime.Module is closed. Each guest invocation + // (callWasm) deserializes its own *wasmtime.Module from modulePath and closes + // it when the invocation returns, so nothing is held while suspended. + moduleDir string + modulePath string + cfg *ModuleConfig metrics *moduleMetrics @@ -152,7 +161,7 @@ type callWasmFunc func(timeout time.Duration, req *sdkpb.ExecuteRequest, linkWas var _ ModuleV1 = (*module)(nil) -type linkFn[T any] func(m *module, store *wasmtime.Store, exec *execution[T]) (*wasmtime.Instance, error) +type linkFn[T any] func(m *module, store *wasmtime.Store, mod *wasmtime.Module, exec *execution[T]) (*wasmtime.Instance, error) // WithDeterminism sets the Determinism field to a deterministic seed from a known time. // @@ -396,14 +405,35 @@ func newModule(modCfg *ModuleConfig, binary []byte) (*module, error) { modCfg.SdkLabeler(v2ImportName) + // Serialize the compiled module to a file and close the in-memory module: it + // is not kept resident for the lifetime of the instance. Each guest + // invocation deserializes its own copy from this file (see callWasm). + serialized, err := mod.Serialize() + mod.Close() + if err != nil { + return nil, fmt.Errorf("error serializing wasmtime module: %w", err) + } + + moduleDir, err := os.MkdirTemp("", "wasm-module-") + if err != nil { + return nil, fmt.Errorf("error creating module temp dir: %w", err) + } + modulePath := filepath.Join(moduleDir, "module.bin") + if err := os.WriteFile(modulePath, serialized, 0o600); err != nil { + _ = os.RemoveAll(moduleDir) + return nil, fmt.Errorf("error writing serialized module: %w", err) + } + metrics, err := newModuleMetrics() if err != nil { + _ = os.RemoveAll(moduleDir) return nil, fmt.Errorf("error creating module metrics: %w", err) } m := &module{ engine: engine, - module: mod, + moduleDir: moduleDir, + modulePath: modulePath, wconfig: cfg, cfg: modCfg, metrics: metrics, @@ -416,7 +446,7 @@ func newModule(modCfg *ModuleConfig, binary []byte) (*module, error) { return m, nil } -func linkNoDAG(m *module, store *wasmtime.Store, exec *execution[*sdkpb.ExecutionResult]) (*wasmtime.Instance, error) { +func linkNoDAG(m *module, store *wasmtime.Store, mod *wasmtime.Module, exec *execution[*sdkpb.ExecutionResult]) (*wasmtime.Instance, error) { linker, err := newWasiLinker(exec, m.engine) if err != nil { return nil, err @@ -509,10 +539,10 @@ func linkNoDAG(m *module, store *wasmtime.Store, exec *execution[*sdkpb.Executio return nil, fmt.Errorf("error wrapping get_time func: %w", err) } - return linker.Instantiate(store, m.module) + return linker.Instantiate(store, mod) } -func linkLegacyDAG(m *module, store *wasmtime.Store, exec *execution[*wasmdagpb.Response]) (*wasmtime.Instance, error) { +func linkLegacyDAG(m *module, store *wasmtime.Store, mod *wasmtime.Module, exec *execution[*wasmdagpb.Response]) (*wasmtime.Instance, error) { linker, err := newDagWasiLinker(m.cfg, m.engine) if err != nil { return nil, err @@ -556,7 +586,7 @@ func linkLegacyDAG(m *module, store *wasmtime.Store, exec *execution[*wasmdagpb. return nil, fmt.Errorf("error wrapping log func: %w", err) } - return linker.Instantiate(store, m.module) + return linker.Instantiate(store, mod) } func (m *module) Start() { @@ -578,8 +608,8 @@ func (m *module) Close() { m.wg.Wait() m.engine.Close() - m.module.Close() m.wconfig.Close() + _ = os.RemoveAll(m.moduleDir) } func (m *module) IsLegacyDAG() bool { @@ -843,7 +873,17 @@ func callWasm[I, O proto.Message]( deadline := timeout / m.cfg.TickInterval store.SetEpochDeadline(uint64(deadline)) - instance, err := linkWasm(m, store, exec) + // Deserialize the compiled module from disk for this invocation only, and + // close it before returning. This keeps no module resident between runs, so a + // suspended execution (which returns from callWasm via the suspend trap) + // holds nothing while it waits; the next resume deserializes again. + mod, err := wasmtime.NewModuleDeserializeFile(m.engine, m.modulePath) + if err != nil { + return 0, fmt.Errorf("error deserializing wasm module: %w", err) + } + defer mod.Close() + + instance, err := linkWasm(m, store, mod, exec) if err != nil { return 0, fmt.Errorf("error linking wasm: %w", err) } From 44309057403829cce1b3a067ea553927742cf6d2 Mon Sep 17 00:00:00 2001 From: Cedric Cordenier Date: Wed, 8 Jul 2026 10:44:14 +0100 Subject: [PATCH 4/6] Make engine global --- pkg/workflows/wasm/host/module.go | 78 +++++++++++++++---------------- 1 file changed, 39 insertions(+), 39 deletions(-) diff --git a/pkg/workflows/wasm/host/module.go b/pkg/workflows/wasm/host/module.go index 02e221b218..b0ff94f943 100644 --- a/pkg/workflows/wasm/host/module.go +++ b/pkg/workflows/wasm/host/module.go @@ -16,7 +16,6 @@ import ( "path/filepath" "regexp" "strings" - "sync" "time" "github.com/andybalholm/brotli" @@ -128,9 +127,6 @@ type ModuleV2 = host.Module type ExecutionHelper = host.ExecutionHelper type module struct { - engine *wasmtime.Engine - wconfig *wasmtime.Config - // The compiled module is not kept resident for the lifetime of the module // instance. Instead it is compiled once, serialized to a file in moduleDir, // and the in-memory *wasmtime.Module is closed. Each guest invocation @@ -143,9 +139,6 @@ type module struct { metrics *moduleMetrics - wg sync.WaitGroup - stopCh chan struct{} - v2ImportName string // callWasm runs a single guest invocation for the v2 (no-DAG) execution path. @@ -375,21 +368,44 @@ func NewModule(ctx context.Context, modCfg *ModuleConfig, binary []byte, opts .. return newModule(modCfg, binary) } -func newModule(modCfg *ModuleConfig, binary []byte) (*module, error) { +// wasmEngine is the single wasmtime engine shared by every module. Compiling a +// module, (de)serializing it, and instantiating stores are all done against this +// engine. It is created once in init and lives for the lifetime of the process, +// so it is never closed. +// +// Note: the config is fixed at init time and can no longer be derived from a +// per-module ModuleConfig. ModuleConfig.InitialFuel is therefore no longer wired +// into the engine (fuel metering is disabled; CPU-time is measured via wall-clock +// in callWasm). +var wasmEngine *wasmtime.Engine + +func init() { cfg := wasmtime.NewConfig() cfg.SetEpochInterruption(true) - if modCfg.InitialFuel > 0 { - cfg.SetConsumeFuel(true) - } if err := cfg.CacheConfigLoadDefault(); err != nil { - modCfg.Logger.Errorw("failed to load cache config, continuing without cache", "error", err) + // Non-fatal: continue without the compilation cache. There is no logger + // available in init, so the error is intentionally swallowed here. + _ = err } cfg.SetCraneliftOptLevel(wasmtime.OptLevelSpeedAndSize) SetUnwinding(cfg) // Handled differently based on host OS. - engine := wasmtime.NewEngineWithConfig(cfg) + wasmEngine = wasmtime.NewEngineWithConfig(cfg) - mod, err := wasmtime.NewModule(engine, binary) + // A single background goroutine drives epoch interruption for every module by + // incrementing the shared engine's epoch. The tick interval is fixed at 100ms; + // store deadlines are computed against the same interval in callWasm. It runs + // for the lifetime of the process and is intentionally never stopped. + go func() { + ticker := time.NewTicker(100 * time.Millisecond) + for range ticker.C { + wasmEngine.IncrementEpoch() + } + }() +} + +func newModule(modCfg *ModuleConfig, binary []byte) (*module, error) { + mod, err := wasmtime.NewModule(wasmEngine, binary) if err != nil { return nil, fmt.Errorf("error creating wasmtime module: %w", err) } @@ -431,13 +447,10 @@ func newModule(modCfg *ModuleConfig, binary []byte) (*module, error) { } m := &module{ - engine: engine, moduleDir: moduleDir, modulePath: modulePath, - wconfig: cfg, cfg: modCfg, metrics: metrics, - stopCh: make(chan struct{}), v2ImportName: v2ImportName, } m.callWasm = func(timeout time.Duration, req *sdkpb.ExecuteRequest, linkWasm linkFn[*sdkpb.ExecutionResult], exec *execution[*sdkpb.ExecutionResult]) (time.Duration, error) { @@ -447,7 +460,7 @@ func newModule(modCfg *ModuleConfig, binary []byte) (*module, error) { } func linkNoDAG(m *module, store *wasmtime.Store, mod *wasmtime.Module, exec *execution[*sdkpb.ExecutionResult]) (*wasmtime.Instance, error) { - linker, err := newWasiLinker(exec, m.engine) + linker, err := newWasiLinker(exec, wasmEngine) if err != nil { return nil, err } @@ -543,7 +556,7 @@ func linkNoDAG(m *module, store *wasmtime.Store, mod *wasmtime.Module, exec *exe } func linkLegacyDAG(m *module, store *wasmtime.Store, mod *wasmtime.Module, exec *execution[*wasmdagpb.Response]) (*wasmtime.Instance, error) { - linker, err := newDagWasiLinker(m.cfg, m.engine) + linker, err := newDagWasiLinker(m.cfg, wasmEngine) if err != nil { return nil, err } @@ -589,26 +602,13 @@ func linkLegacyDAG(m *module, store *wasmtime.Store, mod *wasmtime.Module, exec return linker.Instantiate(store, mod) } -func (m *module) Start() { - m.wg.Go(func() { - ticker := time.NewTicker(m.cfg.TickInterval) - for { - select { - case <-m.stopCh: - return - case <-ticker.C: - m.engine.IncrementEpoch() - } - } - }) -} +// Start is a no-op: epoch interruption is driven by a single process-global +// ticker started in init (see wasmEngine), not per module. +func (m *module) Start() {} func (m *module) Close() { - close(m.stopCh) - m.wg.Wait() - - m.engine.Close() - m.wconfig.Close() + // The engine and its epoch ticker are process-global (see wasmEngine) and + // shared by all modules, so they are intentionally not stopped here. _ = os.RemoveAll(m.moduleDir) } @@ -837,7 +837,7 @@ func callWasm[I, O proto.Message]( linkWasm linkFn[O], exec *execution[O], ) (time.Duration, error) { - store := wasmtime.NewStore(m.engine) + store := wasmtime.NewStore(wasmEngine) defer store.Close() wasi := wasmtime.NewWasiConfig() @@ -877,7 +877,7 @@ func callWasm[I, O proto.Message]( // close it before returning. This keeps no module resident between runs, so a // suspended execution (which returns from callWasm via the suspend trap) // holds nothing while it waits; the next resume deserializes again. - mod, err := wasmtime.NewModuleDeserializeFile(m.engine, m.modulePath) + mod, err := wasmtime.NewModuleDeserializeFile(wasmEngine, m.modulePath) if err != nil { return 0, fmt.Errorf("error deserializing wasm module: %w", err) } From 65723a20429f2543ab717b6499cde2bf79368b63 Mon Sep 17 00:00:00 2001 From: Cedric Cordenier Date: Wed, 8 Jul 2026 11:08:21 +0100 Subject: [PATCH 5/6] Test removing the cache --- pkg/workflows/wasm/host/module.go | 5 ----- 1 file changed, 5 deletions(-) diff --git a/pkg/workflows/wasm/host/module.go b/pkg/workflows/wasm/host/module.go index b0ff94f943..526f5baa32 100644 --- a/pkg/workflows/wasm/host/module.go +++ b/pkg/workflows/wasm/host/module.go @@ -382,11 +382,6 @@ var wasmEngine *wasmtime.Engine func init() { cfg := wasmtime.NewConfig() cfg.SetEpochInterruption(true) - if err := cfg.CacheConfigLoadDefault(); err != nil { - // Non-fatal: continue without the compilation cache. There is no logger - // available in init, so the error is intentionally swallowed here. - _ = err - } cfg.SetCraneliftOptLevel(wasmtime.OptLevelSpeedAndSize) SetUnwinding(cfg) // Handled differently based on host OS. From f9127d0fe382dc01a00a82c6aaf10a19df57db24 Mon Sep 17 00:00:00 2001 From: Cedric Cordenier Date: Thu, 9 Jul 2026 10:27:42 +0100 Subject: [PATCH 6/6] Add module cache --- pkg/workflows/wasm/host/module.go | 113 +++++++++++++++++++++++++----- 1 file changed, 94 insertions(+), 19 deletions(-) diff --git a/pkg/workflows/wasm/host/module.go b/pkg/workflows/wasm/host/module.go index 526f5baa32..9e80a8f9cb 100644 --- a/pkg/workflows/wasm/host/module.go +++ b/pkg/workflows/wasm/host/module.go @@ -3,8 +3,10 @@ package host import ( "bytes" "context" + "crypto/sha256" "encoding/base64" "encoding/binary" + "encoding/hex" "encoding/json" "errors" "fmt" @@ -16,6 +18,7 @@ import ( "path/filepath" "regexp" "strings" + "sync" "time" "github.com/andybalholm/brotli" @@ -128,11 +131,14 @@ type ExecutionHelper = host.ExecutionHelper type module struct { // The compiled module is not kept resident for the lifetime of the module - // instance. Instead it is compiled once, serialized to a file in moduleDir, - // and the in-memory *wasmtime.Module is closed. Each guest invocation - // (callWasm) deserializes its own *wasmtime.Module from modulePath and closes - // it when the invocation returns, so nothing is held while suspended. - moduleDir string + // instance. Instead it is compiled once, serialized, and written to the + // process-global content-addressed moduleFileCache (see moduleCache); the + // in-memory *wasmtime.Module is then closed. modulePath points at the cache + // entry, which is keyed by the SHA-256 of the serialized bytes so that + // modules compiled from identical binaries share a single file. Each guest + // invocation (callWasm) deserializes its own *wasmtime.Module from modulePath + // and closes it when the invocation returns, so nothing is held while + // suspended. modulePath string cfg *ModuleConfig @@ -152,6 +158,77 @@ type module struct { // callWasmFunc runs one guest invocation for the v2 (no-DAG) execution path. type callWasmFunc func(timeout time.Duration, req *sdkpb.ExecuteRequest, linkWasm linkFn[*sdkpb.ExecutionResult], exec *execution[*sdkpb.ExecutionResult]) (time.Duration, error) +// moduleFileCache is the process-global store for serialized wasmtime modules. +var moduleFileCache = &moduleCache{refs: map[string]int{}} + +// moduleCache is a content-addressed, reference-counted store for serialized +// wasmtime modules. Modules compiled from identical binaries serialize to +// identical bytes, so they are written once to a shared directory under a key +// derived from the SHA-256 of the serialized bytes and shared by every module +// instance that references them. The reference count tracks live users of each +// key so a file is removed only when the last module referencing it is closed. +type moduleCache struct { + mu sync.Mutex + dir string + refs map[string]int +} + +// store writes serialized to the cache, keyed by the SHA-256 of its contents, +// and returns the path to the cache entry. The refcount check under the mutex +// is the atomic decision: only the first reference to a given key writes the +// file; subsequent references reuse the existing one (dedupe). The returned path +// must later be passed to release exactly once (see module.Close). +func (c *moduleCache) store(serialized []byte) (string, error) { + sum := sha256.Sum256(serialized) + key := hex.EncodeToString(sum[:]) + + c.mu.Lock() + defer c.mu.Unlock() + + if err := c.ensureDir(); err != nil { + return "", err + } + path := filepath.Join(c.dir, key) + + if c.refs[key] == 0 { + if err := os.WriteFile(path, serialized, 0o600); err != nil { + return "", fmt.Errorf("error writing serialized module: %w", err) + } + } + c.refs[key]++ + return path, nil +} + +// release drops one reference to the cache entry at path, removing the file once +// the last reference is gone. It is safe to call with a path returned by store. +func (c *moduleCache) release(path string) { + key := filepath.Base(path) + + c.mu.Lock() + defer c.mu.Unlock() + + switch { + case c.refs[key] <= 1: + delete(c.refs, key) + _ = os.Remove(path) + default: + c.refs[key]-- + } +} + +// ensureDir lazily creates the shared cache directory. The caller must hold c.mu. +func (c *moduleCache) ensureDir() error { + if c.dir != "" { + return nil + } + dir, err := os.MkdirTemp("", "wasm-module-cache-") + if err != nil { + return fmt.Errorf("error creating module cache dir: %w", err) + } + c.dir = dir + return nil +} + var _ ModuleV1 = (*module)(nil) type linkFn[T any] func(m *module, store *wasmtime.Store, mod *wasmtime.Module, exec *execution[T]) (*wasmtime.Instance, error) @@ -416,33 +493,29 @@ func newModule(modCfg *ModuleConfig, binary []byte) (*module, error) { modCfg.SdkLabeler(v2ImportName) - // Serialize the compiled module to a file and close the in-memory module: it - // is not kept resident for the lifetime of the instance. Each guest - // invocation deserializes its own copy from this file (see callWasm). + // Serialize the compiled module and close the in-memory module: it is not + // kept resident for the lifetime of the instance. The serialized bytes are + // stored in the process-global content-addressed cache, keyed by their + // SHA-256, so modules compiled from identical binaries share one file. Each + // guest invocation deserializes its own copy from this file (see callWasm). serialized, err := mod.Serialize() mod.Close() if err != nil { return nil, fmt.Errorf("error serializing wasmtime module: %w", err) } - moduleDir, err := os.MkdirTemp("", "wasm-module-") + modulePath, err := moduleFileCache.store(serialized) if err != nil { - return nil, fmt.Errorf("error creating module temp dir: %w", err) - } - modulePath := filepath.Join(moduleDir, "module.bin") - if err := os.WriteFile(modulePath, serialized, 0o600); err != nil { - _ = os.RemoveAll(moduleDir) - return nil, fmt.Errorf("error writing serialized module: %w", err) + return nil, err } metrics, err := newModuleMetrics() if err != nil { - _ = os.RemoveAll(moduleDir) + moduleFileCache.release(modulePath) return nil, fmt.Errorf("error creating module metrics: %w", err) } m := &module{ - moduleDir: moduleDir, modulePath: modulePath, cfg: modCfg, metrics: metrics, @@ -603,8 +676,10 @@ func (m *module) Start() {} func (m *module) Close() { // The engine and its epoch ticker are process-global (see wasmEngine) and - // shared by all modules, so they are intentionally not stopped here. - _ = os.RemoveAll(m.moduleDir) + // shared by all modules, so they are intentionally not stopped here. The + // cache entry is reference counted and removed only once the last module + // referencing these serialized bytes is closed (see moduleCache). + moduleFileCache.release(m.modulePath) } func (m *module) IsLegacyDAG() bool {