From 2f3f0ac8345579eb7002c63b36fa513caa550450 Mon Sep 17 00:00:00 2001 From: cfal Date: Wed, 2 Sep 2026 23:05:10 +0700 Subject: [PATCH] workflows/wasm/host: prevent limiter and Wasmtime resource leaks --- pkg/workflows/wasm/host/execution.go | 12 +- pkg/workflows/wasm/host/module.go | 180 ++++++++++-------- pkg/workflows/wasm/host/module_test.go | 155 +++++++++++++++ .../wasm/host/poll_oneoff_regression_test.go | 2 +- pkg/workflows/wasm/host/wasip1.go | 30 ++- pkg/workflows/wasm/host/wasm.go | 3 +- 6 files changed, 290 insertions(+), 92 deletions(-) diff --git a/pkg/workflows/wasm/host/execution.go b/pkg/workflows/wasm/host/execution.go index fbb87323ab..08af1ace6e 100644 --- a/pkg/workflows/wasm/host/execution.go +++ b/pkg/workflows/wasm/host/execution.go @@ -230,7 +230,7 @@ func (e *execution[T]) log(caller *wasmtime.Caller, ptr int32, ptrlen int32) { } func (e *execution[T]) emitMetric(caller *wasmtime.Caller, ptr int32, ptrlen int32) int32 { - if err := e.module.cfg.EnableUserMetricsLimiter.AllowErr(e.ctx); err != nil { + if err := limiterOrDefault(e.module.cfg.EnableUserMetricsLimiter, e.module.defaultLimiters.enableUserMetrics).AllowErr(e.ctx); err != nil { return -1 } @@ -238,7 +238,7 @@ func (e *execution[T]) emitMetric(caller *wasmtime.Caller, ptr int32, ptrlen int return -1 } - if err := e.module.cfg.MaxUserMetricPayloadLimiter.Check(e.ctx, config.Size(ptrlen)); err != nil { + if err := limiterOrDefault(e.module.cfg.MaxUserMetricPayloadLimiter, e.module.defaultLimiters.maxUserMetricPayload).Check(e.ctx, config.Size(ptrlen)); err != nil { e.module.cfg.Logger.Warnf("metric payload too large: %d bytes - dropping: %s", ptrlen, err) return -1 } @@ -260,18 +260,18 @@ func (e *execution[T]) emitMetric(caller *wasmtime.Caller, ptr int32, ptrlen int return -1 } - if err := e.module.cfg.MaxUserMetricNameLengthLimiter.Check(e.ctx, len(metric.Name)); err != nil { + if err := limiterOrDefault(e.module.cfg.MaxUserMetricNameLengthLimiter, e.module.defaultLimiters.maxUserMetricNameLength).Check(e.ctx, len(metric.Name)); err != nil { e.module.cfg.Logger.Warnf("metric name too long: %d chars - dropping: %s", len(metric.Name), err) return -1 } - if err := e.module.cfg.MaxUserMetricLabelsPerMetricLimiter.Check(e.ctx, len(metric.Labels)); err != nil { + if err := limiterOrDefault(e.module.cfg.MaxUserMetricLabelsPerMetricLimiter, e.module.defaultLimiters.maxUserMetricLabelsPerMetric).Check(e.ctx, len(metric.Labels)); err != nil { e.module.cfg.Logger.Warnf("too many labels on metric %q: %d - dropping: %s", metric.Name, len(metric.Labels), err) return -1 } for k, v := range metric.Labels { - if err := e.module.cfg.MaxUserMetricLabelValueLengthLimiter.Check(e.ctx, len(v)); err != nil { + if err := limiterOrDefault(e.module.cfg.MaxUserMetricLabelValueLengthLimiter, e.module.defaultLimiters.maxUserMetricLabelValueLength).Check(e.ctx, len(v)); err != nil { e.module.cfg.Logger.Warnf("label value too long for key %q on metric %q: %d chars - dropping: %s", k, metric.Name, len(v), err) return -1 } @@ -357,7 +357,7 @@ func (e *execution[T]) pollOneoff(caller *wasmtime.Caller, subscriptionptr int32 if nsubscriptions <= 0 || nsubscriptions > max(math.MaxInt32/subscriptionLen, math.MaxInt32/eventsLen) { return ErrnoInval } - if err := e.module.cfg.MaxSubscriptionsLimiter.Check(e.ctx, int(nsubscriptions)); err != nil { + if err := limiterOrDefault(e.module.cfg.MaxSubscriptionsLimiter, e.module.defaultLimiters.maxSubscriptions).Check(e.ctx, int(nsubscriptions)); err != nil { return ErrnoInval } diff --git a/pkg/workflows/wasm/host/module.go b/pkg/workflows/wasm/host/module.go index 5e8e129518..4d9c6d7f7c 100644 --- a/pkg/workflows/wasm/host/module.go +++ b/pkg/workflows/wasm/host/module.go @@ -27,7 +27,6 @@ import ( "github.com/smartcontractkit/chainlink-common/pkg/config" "github.com/smartcontractkit/chainlink-common/pkg/custmsg" "github.com/smartcontractkit/chainlink-common/pkg/logger" - "github.com/smartcontractkit/chainlink-common/pkg/settings" "github.com/smartcontractkit/chainlink-common/pkg/settings/cresettings" "github.com/smartcontractkit/chainlink-common/pkg/settings/limits" dagsdk "github.com/smartcontractkit/chainlink-common/pkg/workflows/sdk" @@ -96,7 +95,8 @@ type ModuleConfig struct { MaxResponseSizeLimiter limits.BoundLimiter[config.Size] // supersedes MaxResponseSizeBytes if set // MaxSubscriptionsLimiter bounds nsubscriptions in the WASI poll_oneoff host - // call. Defaults to cresettings.Default.WASMPollOneoffSubscriptionLimit. + // call. It uses the default value of cresettings.Default.WASMPollOneoffSubscriptionLimit; + // inject a limiter to provide dynamic settings. MaxSubscriptionsLimiter limits.BoundLimiter[int] MaxLogLenBytes uint32 @@ -150,7 +150,8 @@ type module struct { module *wasmtime.Module wconfig *wasmtime.Config - cfg *ModuleConfig + cfg *ModuleConfig + defaultLimiters moduleLimiters metrics moduleMetrics @@ -175,6 +176,48 @@ var _ ModuleV1 = (*module)(nil) type linkFn[T any] func(ctx context.Context, m *module, store *wasmtime.Store, exec *execution[T]) (*wasmtime.Instance, error) +type moduleLimiters struct { + pendingCalls limits.ResourcePoolLimiter[int] + enableUserMetrics limits.GateLimiter + maxUserMetricPayload limits.BoundLimiter[config.Size] + maxUserMetricNameLength limits.BoundLimiter[int] + maxUserMetricLabelsPerMetric limits.BoundLimiter[int] + maxUserMetricLabelValueLength limits.BoundLimiter[int] + memory limits.BoundLimiter[config.Size] + maxCompressedBinary limits.BoundLimiter[config.Size] + maxDecompressedBinary limits.BoundLimiter[config.Size] + maxResponseSize limits.BoundLimiter[config.Size] + maxSubscriptions limits.BoundLimiter[int] +} + +func (l *moduleLimiters) close() { + closers := [...]io.Closer{ + l.pendingCalls, + l.enableUserMetrics, + l.maxUserMetricPayload, + l.maxUserMetricNameLength, + l.maxUserMetricLabelsPerMetric, + l.maxUserMetricLabelValueLength, + l.memory, + l.maxCompressedBinary, + l.maxDecompressedBinary, + l.maxResponseSize, + l.maxSubscriptions, + } + for i := len(closers) - 1; i >= 0; i-- { + if closers[i] != nil { + _ = closers[i].Close() + } + } +} + +func limiterOrDefault[T io.Closer](configured, defaultLimiter T) T { + if any(configured) != nil { + return configured + } + return defaultLimiter +} + // WithDeterminism sets the Determinism field to a deterministic seed from a known time. // // "The Times 03/Jan/2009 Chancellor on brink of second bailout for banks" @@ -189,7 +232,20 @@ func WithDeterminism() func(*ModuleConfig) { } } +// NewModule creates a WASM module. Limiters omitted from modCfg are created +// internally, owned by the returned module, and not written back to modCfg. +// Caller-provided or subsequently configured limiters remain caller-owned. +// Limiter fields may be set or replaced after construction, but must not be +// reset to nil. func NewModule(ctx context.Context, modCfg *ModuleConfig, binary []byte, opts ...func(*ModuleConfig)) (*module, error) { + var defaultLimiters moduleLimiters + cleanupLimiters := true + defer func() { + if cleanupLimiters { + defaultLimiters.close() + } + }() + // Apply options to the module config. for _, opt := range opts { opt(modCfg) @@ -209,15 +265,6 @@ func NewModule(ctx context.Context, modCfg *ModuleConfig, binary []byte, opts .. modCfg.MaxFetchRequests = defaultMaxFetchRequests } - if modCfg.PendingCallsLimiter == nil { - lf := limits.Factory{Logger: modCfg.Logger} - var err error - modCfg.PendingCallsLimiter, err = limits.MakeResourcePoolLimiter(lf, cresettings.Default.PerWorkflow.CapabilityConcurrencyLimit) - if err != nil { - return nil, fmt.Errorf("failed to make pending calls limiter: %w", err) - } - } - if modCfg.Labeler == nil { modCfg.Labeler = &unimplementedMessageEmitter{} } @@ -286,41 +333,29 @@ func NewModule(ctx context.Context, modCfg *ModuleConfig, binary []byte, opts .. lf := limits.Factory{Logger: modCfg.Logger} + if modCfg.PendingCallsLimiter == nil { + limiter, err := limits.MakeResourcePoolLimiter(lf, cresettings.Default.PerWorkflow.CapabilityConcurrencyLimit) + if err != nil { + return nil, fmt.Errorf("failed to make pending calls limiter: %w", err) + } + defaultLimiters.pendingCalls = limiter + } + if modCfg.EnableUserMetricsLimiter == nil { - modCfg.EnableUserMetricsLimiter = limits.NewGateLimiter(false) + defaultLimiters.enableUserMetrics = limits.NewGateLimiter(false) } if modCfg.MaxUserMetricPayloadLimiter == nil { - limit := settings.Size(config.Size(modCfg.MaxUserMetricPayloadBytes)) - var err error - modCfg.MaxUserMetricPayloadLimiter, err = limits.MakeUpperBoundLimiter(lf, limit) - if err != nil { - return nil, fmt.Errorf("failed to make metric payload size limiter: %w", err) - } + defaultLimiters.maxUserMetricPayload = limits.NewUpperBoundLimiter(config.Size(modCfg.MaxUserMetricPayloadBytes)) } if modCfg.MaxUserMetricNameLengthLimiter == nil { - limit := settings.Int(int(modCfg.MaxUserMetricNameLength)) - var err error - modCfg.MaxUserMetricNameLengthLimiter, err = limits.MakeUpperBoundLimiter(lf, limit) - if err != nil { - return nil, fmt.Errorf("failed to make metric name length limiter: %w", err) - } + defaultLimiters.maxUserMetricNameLength = limits.NewUpperBoundLimiter(int(modCfg.MaxUserMetricNameLength)) } if modCfg.MaxUserMetricLabelsPerMetricLimiter == nil { - limit := settings.Int(int(modCfg.MaxUserMetricLabelsPerMetric)) - var err error - modCfg.MaxUserMetricLabelsPerMetricLimiter, err = limits.MakeUpperBoundLimiter(lf, limit) - if err != nil { - return nil, fmt.Errorf("failed to make labels per metric limiter: %w", err) - } + defaultLimiters.maxUserMetricLabelsPerMetric = limits.NewUpperBoundLimiter(int(modCfg.MaxUserMetricLabelsPerMetric)) } if modCfg.MaxUserMetricLabelValueLengthLimiter == nil { - limit := settings.Int(int(modCfg.MaxUserMetricLabelValueLength)) - var err error - modCfg.MaxUserMetricLabelValueLengthLimiter, err = limits.MakeUpperBoundLimiter(lf, limit) - if err != nil { - return nil, fmt.Errorf("failed to make label value length limiter: %w", err) - } + defaultLimiters.maxUserMetricLabelValueLength = limits.NewUpperBoundLimiter(int(modCfg.MaxUserMetricLabelValueLength)) } if modCfg.MemoryLimiter == nil { // Take the max of the min and the configured max memory mbs. @@ -328,55 +363,34 @@ func NewModule(ctx context.Context, modCfg *ModuleConfig, binary []byte, opts .. // and local testing has shown that with less than the min, some // binaries may error sporadically. modCfg.MaxMemoryMBs = uint64(math.Max(float64(modCfg.MinMemoryMBs), float64(modCfg.MaxMemoryMBs))) - limit := settings.Size(config.Size(modCfg.MaxMemoryMBs) * config.MByte) - var err error - modCfg.MemoryLimiter, err = limits.MakeUpperBoundLimiter(lf, limit) - if err != nil { - return nil, fmt.Errorf("failed to make memory limiter: %w", err) - } + defaultLimiters.memory = limits.NewUpperBoundLimiter(config.Size(modCfg.MaxMemoryMBs) * config.MByte) } if modCfg.MaxCompressedBinaryLimiter == nil { - limit := settings.Size(config.Size(modCfg.MaxCompressedBinarySize)) - var err error - modCfg.MaxCompressedBinaryLimiter, err = limits.MakeUpperBoundLimiter(lf, limit) - if err != nil { - return nil, fmt.Errorf("failed to make compressed binary size limiter: %w", err) - } + defaultLimiters.maxCompressedBinary = limits.NewUpperBoundLimiter(config.Size(modCfg.MaxCompressedBinarySize)) } if modCfg.MaxDecompressedBinaryLimiter == nil { - limit := settings.Size(config.Size(modCfg.MaxDecompressedBinarySize)) - var err error - modCfg.MaxDecompressedBinaryLimiter, err = limits.MakeUpperBoundLimiter(lf, limit) - if err != nil { - return nil, fmt.Errorf("failed to make decompressed binary size limiter: %w", err) - } + defaultLimiters.maxDecompressedBinary = limits.NewUpperBoundLimiter(config.Size(modCfg.MaxDecompressedBinarySize)) } if modCfg.MaxResponseSizeLimiter == nil { - limit := settings.Size(config.Size(modCfg.MaxResponseSizeBytes)) - var err error - modCfg.MaxResponseSizeLimiter, err = limits.MakeUpperBoundLimiter(lf, limit) - if err != nil { - return nil, fmt.Errorf("failed to make response size limiter: %w", err) - } + defaultLimiters.maxResponseSize = limits.NewUpperBoundLimiter(config.Size(modCfg.MaxResponseSizeBytes)) } if modCfg.MaxSubscriptionsLimiter == nil { - var err error - modCfg.MaxSubscriptionsLimiter, err = limits.MakeUpperBoundLimiter(lf, cresettings.Default.WASMPollOneoffSubscriptionLimit) - if err != nil { - return nil, fmt.Errorf("failed to make poll_oneoff subscription limiter: %w", err) - } + defaultLimiters.maxSubscriptions = limits.NewUpperBoundLimiter(cresettings.Default.WASMPollOneoffSubscriptionLimit.DefaultValue) } + maxCompressedBinaryLimiter := limiterOrDefault(modCfg.MaxCompressedBinaryLimiter, defaultLimiters.maxCompressedBinary) + maxDecompressedBinaryLimiter := limiterOrDefault(modCfg.MaxDecompressedBinaryLimiter, defaultLimiters.maxDecompressedBinary) + if !modCfg.IsUncompressed { // validate the binary size before decompressing // this is to prevent decompression bombs - if err := modCfg.MaxCompressedBinaryLimiter.Check(ctx, config.SizeOf(binary)); err != nil { + if err := maxCompressedBinaryLimiter.Check(ctx, config.SizeOf(binary)); err != nil { if errors.Is(err, limits.ErrorBoundLimited[config.Size]{}) { return nil, fmt.Errorf("compressed binary size exceeds the maximum allowed size: %w", err) } return nil, fmt.Errorf("failed to check compressed binary size limit: %w", err) } - maxDecompressedBinarySize, err := modCfg.MaxDecompressedBinaryLimiter.Limit(ctx) + maxDecompressedBinarySize, err := maxDecompressedBinaryLimiter.Limit(ctx) if err != nil { return nil, fmt.Errorf("failed to get decompressed binary size limit: %w", err) } @@ -392,7 +406,7 @@ func NewModule(ctx context.Context, modCfg *ModuleConfig, binary []byte, opts .. // Validate the decompressed binary size. // io.LimitReader prevents decompression bombs by reading up to a set limit, but it will not return an error if the limit is reached. // The Read() method will return io.EOF, and ReadAll will gracefully handle it and return nil. - if err := modCfg.MaxDecompressedBinaryLimiter.Check(ctx, config.SizeOf(binary)); err != nil { + if err := maxDecompressedBinaryLimiter.Check(ctx, config.SizeOf(binary)); err != nil { if errors.Is(err, limits.ErrorBoundLimited[config.Size]{}) { return nil, fmt.Errorf("decompressed binary size reached the maximum allowed size: %w", err) } @@ -404,7 +418,13 @@ func NewModule(ctx context.Context, modCfg *ModuleConfig, binary []byte, opts .. return nil, fmt.Errorf("failed to create module metrics: %w", err) } - return newModule(modCfg, binary, metrics) + m, err := newModule(modCfg, binary, metrics) + if err != nil { + return nil, err + } + m.defaultLimiters = defaultLimiters + cleanupLimiters = false + return m, nil } func newModule(modCfg *ModuleConfig, binary []byte, metrics moduleMetrics) (*module, error) { @@ -423,6 +443,7 @@ func newModule(modCfg *ModuleConfig, binary []byte, metrics moduleMetrics) (*mod mod, err := wasmtime.NewModule(engine, binary) if err != nil { + engine.Close() return nil, fmt.Errorf("error creating wasmtime module: %w", err) } @@ -431,6 +452,8 @@ func newModule(modCfg *ModuleConfig, binary []byte, metrics moduleMetrics) (*mod // at all. Reject it here rather than letting the first callback dereference // a missing or wrong-typed export. if err = requireMemoryExport(mod); err != nil { + mod.Close() + engine.Close() return nil, err } @@ -470,6 +493,7 @@ func linkNoDAG(_ context.Context, m *module, store *wasmtime.Store, exec *execut if err != nil { return nil, err } + defer linker.Close() if err = linker.FuncWrap( "env", @@ -562,10 +586,11 @@ func linkNoDAG(_ context.Context, m *module, store *wasmtime.Store, exec *execut } func linkLegacyDAG(ctx context.Context, m *module, store *wasmtime.Store, exec *execution[*wasmdagpb.Response]) (*wasmtime.Instance, error) { - linker, err := newDagWasiLinker(ctx, m.cfg, m.engine) + linker, err := newDagWasiLinker(ctx, m) if err != nil { return nil, err } + defer linker.Close() logger := m.cfg.Logger @@ -611,6 +636,7 @@ func linkLegacyDAG(ctx context.Context, m *module, store *wasmtime.Store, exec * func (m *module) Start() { m.wg.Go(func() { ticker := time.NewTicker(m.cfg.TickInterval) + defer ticker.Stop() for { select { case <-m.stopCh: @@ -622,10 +648,13 @@ func (m *module) Start() { }) } +// Close may wait for a blocked acquisition from the internally owned pending +// calls limiter. func (m *module) Close() { close(m.stopCh) m.wg.Wait() + m.defaultLimiters.close() m.engine.Close() m.module.Close() m.wconfig.Close() @@ -736,7 +765,7 @@ func runWasm[I, O proto.Message]( defer store.Close() - maxResponseSizeBytes, err := m.cfg.MaxResponseSizeLimiter.Limit(ctx) + maxResponseSizeBytes, err := limiterOrDefault(m.cfg.MaxResponseSizeLimiter, m.defaultLimiters.maxResponseSize).Limit(ctx) if err != nil { return o, fmt.Errorf("failed to get response size limit: %w", err) } @@ -749,14 +778,13 @@ func runWasm[I, O proto.Message]( reqstr := base64.StdEncoding.EncodeToString(reqpb) wasi := wasmtime.NewWasiConfig() + defer wasi.Close() if err := wasi.SetStdoutFile(m.cfg.guestStdoutFile); err != nil { return o, fmt.Errorf("error setting guest stdout file: %w", err) } if err := wasi.SetStderrFile(m.cfg.guestStderrFile); err != nil { return o, fmt.Errorf("error setting guest stderr file: %w", err) } - defer wasi.Close() - wasi.SetArgv([]string{"wasi", reqstr}) store.SetWasi(wasi) @@ -769,7 +797,7 @@ func runWasm[I, O proto.Message]( } // Limit memory to max memory megabytes per instance. - maxMemoryBytes, err := m.cfg.MemoryLimiter.Limit(ctx) + maxMemoryBytes, err := limiterOrDefault(m.cfg.MemoryLimiter, m.defaultLimiters.memory).Limit(ctx) if err != nil { return o, fmt.Errorf("failed to get memory limit: %w", err) } @@ -797,7 +825,7 @@ func runWasm[I, O proto.Message]( capabilityResponses: map[int32]<-chan *sdkpb.CapabilityResponse{}, secretsResponses: map[int32]<-chan *secretsResponse{}, usedCallbackIDs: map[string]bool{}, - pendingCallsLimiter: m.cfg.PendingCallsLimiter, + pendingCallsLimiter: limiterOrDefault(m.cfg.PendingCallsLimiter, m.defaultLimiters.pendingCalls), module: m, executor: helper, donSeed: donSeed, diff --git a/pkg/workflows/wasm/host/module_test.go b/pkg/workflows/wasm/host/module_test.go index 7dc01dabe6..3be89397c9 100644 --- a/pkg/workflows/wasm/host/module_test.go +++ b/pkg/workflows/wasm/host/module_test.go @@ -1,9 +1,11 @@ package host import ( + "bytes" "context" "encoding/binary" "math" + "runtime/pprof" "strings" "sync" "testing" @@ -17,6 +19,7 @@ import ( "google.golang.org/protobuf/proto" "github.com/smartcontractkit/chainlink-common/pkg/capabilities/v2/protoc/pkg/test_capabilities/basictrigger" + "github.com/smartcontractkit/chainlink-common/pkg/contexts" "github.com/smartcontractkit/chainlink-common/pkg/workflows/host/mocks" "github.com/smartcontractkit/chainlink-common/pkg/custmsg" @@ -29,6 +32,158 @@ import ( "github.com/smartcontractkit/chainlink-protos/cre/go/values/pb" ) +func countLimiterUpdaterGoroutines(t *testing.T) int { + t.Helper() + + var profile bytes.Buffer + require.NoError(t, pprof.Lookup("goroutine").WriteTo(&profile, 2)) + count := 0 + for _, goroutine := range strings.Split(profile.String(), "\n\n") { + if strings.Contains(goroutine, "pkg/settings/limits.(*updater") && strings.Contains(goroutine, ").updateLoop(") { + count++ + } + } + return count +} + +// This test inspects the process-wide goroutine profile and must remain serial. +func TestNewModuleClosesDefaultLimiters(t *testing.T) { + t.Run("constructor error", func(t *testing.T) { + before := countLimiterUpdaterGoroutines(t) + cfg := &ModuleConfig{ + Logger: logger.Test(t), + IsUncompressed: true, + } + + _, err := NewModule(t.Context(), cfg, []byte("invalid wasm")) + require.Error(t, err) + require.Equal(t, before, countLimiterUpdaterGoroutines(t)) + require.Nil(t, cfg.PendingCallsLimiter) + require.Nil(t, cfg.MemoryLimiter) + }) + + t.Run("module close", func(t *testing.T) { + binary, err := wasmtime.Wat2Wasm(`(module (memory (export "memory") 1))`) + require.NoError(t, err) + + before := countLimiterUpdaterGoroutines(t) + mod, err := NewModule(t.Context(), &ModuleConfig{ + Logger: logger.Test(t), + IsUncompressed: true, + }, binary) + require.NoError(t, err) + require.Equal(t, before, countLimiterUpdaterGoroutines(t)) + free, err := limiterOrDefault(mod.cfg.PendingCallsLimiter, mod.defaultLimiters.pendingCalls).Wait(contexts.WithCRE(t.Context(), contexts.CRE{Workflow: "workflow-id"}), 1) + require.NoError(t, err) + free() + require.Eventually(t, func() bool { + return before+1 == countLimiterUpdaterGoroutines(t) + }, time.Second, time.Millisecond) + mod.Start() + mod.Close() + require.Eventually(t, func() bool { + return before == countLimiterUpdaterGoroutines(t) + }, time.Second, time.Millisecond) + }) + + t.Run("reused config", func(t *testing.T) { + binary, err := wasmtime.Wat2Wasm(`(module (memory (export "memory") 1))`) + require.NoError(t, err) + cfg := &ModuleConfig{ + Logger: logger.Test(t), + IsUncompressed: true, + } + before := countLimiterUpdaterGoroutines(t) + + for range 2 { + mod, err := NewModule(t.Context(), cfg, binary) + require.NoError(t, err) + mod.Close() + require.Nil(t, cfg.PendingCallsLimiter) + require.Nil(t, cfg.MaxResponseSizeLimiter) + } + require.Equal(t, before, countLimiterUpdaterGoroutines(t)) + }) + + t.Run("simultaneous modules sharing config", func(t *testing.T) { + binary, err := wasmtime.Wat2Wasm(`(module (memory (export "memory") 1) (func (export "_start")))`) + require.NoError(t, err) + cfg := &ModuleConfig{ + Logger: logger.Test(t), + IsUncompressed: true, + } + + first, err := NewModule(t.Context(), cfg, binary) + require.NoError(t, err) + second, err := NewModule(t.Context(), cfg, binary) + require.NoError(t, err) + t.Cleanup(second.Close) + + first.Close() + request := &wasmpb.Request{Id: "request-id"} + var runErr error + var subscriptionErr error + require.NotPanics(t, func() { + _, runErr = second.Run(t.Context(), request) + subscriptionErr = limiterOrDefault(second.cfg.MaxSubscriptionsLimiter, second.defaultLimiters.maxSubscriptions).Check(t.Context(), 1) + }) + require.NoError(t, runErr) + require.NoError(t, subscriptionErr) + }) + + t.Run("caller-provided limiter", func(t *testing.T) { + binary, err := wasmtime.Wat2Wasm(`(module (memory (export "memory") 1))`) + require.NoError(t, err) + limiter := limits.NewUpperBoundLimiter(1) + t.Cleanup(func() { require.NoError(t, limiter.Close()) }) + + mod, err := NewModule(t.Context(), &ModuleConfig{ + Logger: logger.Test(t), + IsUncompressed: true, + MaxSubscriptionsLimiter: limiter, + }, binary) + require.NoError(t, err) + mod.Close() + require.NoError(t, limiter.Check(t.Context(), 1)) + }) + + t.Run("caller-replaced limiter", func(t *testing.T) { + binary, err := wasmtime.Wat2Wasm(`(module (memory (export "memory") 1))`) + require.NoError(t, err) + cfg := &ModuleConfig{ + Logger: logger.Test(t), + IsUncompressed: true, + } + mod, err := NewModule(t.Context(), cfg, binary) + require.NoError(t, err) + + limiter := &closeTrackingGateLimiter{} + cfg.EnableUserMetricsLimiter = limiter + require.NoError(t, limiterOrDefault(mod.cfg.EnableUserMetricsLimiter, mod.defaultLimiters.enableUserMetrics).AllowErr(t.Context())) + + mod.Close() + require.Same(t, limiter, cfg.EnableUserMetricsLimiter) + require.Zero(t, limiter.closeCalls) + }) +} + +type closeTrackingGateLimiter struct { + closeCalls int +} + +func (l *closeTrackingGateLimiter) Close() error { + l.closeCalls++ + return nil +} + +func (*closeTrackingGateLimiter) Limit(context.Context) (bool, error) { + return true, nil +} + +func (*closeTrackingGateLimiter) AllowErr(context.Context) error { + return nil +} + type mockMessageEmitter struct { e func(context.Context, string, map[string]string) error labels map[string]string diff --git a/pkg/workflows/wasm/host/poll_oneoff_regression_test.go b/pkg/workflows/wasm/host/poll_oneoff_regression_test.go index 2fb84a4607..711299831d 100644 --- a/pkg/workflows/wasm/host/poll_oneoff_regression_test.go +++ b/pkg/workflows/wasm/host/poll_oneoff_regression_test.go @@ -26,7 +26,7 @@ func TestRegressionPollOneoffRejectsExcessiveSubscriptions(t *testing.T) { cfg := &ModuleConfig{MaxSubscriptionsLimiter: limiter} exec := &execution[*sdkpb.ExecutionResult]{ctx: t.Context(), module: &module{cfg: cfg}} - legacyPollOneoff := createPollOneoff(t.Context(), cfg) + legacyPollOneoff := createPollOneoff(t.Context(), limiter) tests := []struct { name string diff --git a/pkg/workflows/wasm/host/wasip1.go b/pkg/workflows/wasm/host/wasip1.go index 5c5a2dabef..3d340e9b7f 100644 --- a/pkg/workflows/wasm/host/wasip1.go +++ b/pkg/workflows/wasm/host/wasip1.go @@ -11,6 +11,8 @@ import ( "github.com/bytecodealliance/wasmtime-go/v48" "github.com/jonboulle/clockwork" + + "github.com/smartcontractkit/chainlink-common/pkg/settings/limits" ) var ( @@ -21,6 +23,12 @@ var ( func newWasiLinker[T any](exec *execution[T], engine *wasmtime.Engine) (*wasmtime.Linker, error) { linker := wasmtime.NewLinker(engine) + cleanupLinker := true + defer func() { + if cleanupLinker { + linker.Close() + } + }() linker.AllowShadowing(true) err := linker.DefineWasi() @@ -49,11 +57,18 @@ func newWasiLinker[T any](exec *execution[T], engine *wasmtime.Engine) (*wasmtim return nil, err } + cleanupLinker = false return linker, nil } -func newDagWasiLinker(ctx context.Context, modCfg *ModuleConfig, engine *wasmtime.Engine) (*wasmtime.Linker, error) { - linker := wasmtime.NewLinker(engine) +func newDagWasiLinker(ctx context.Context, m *module) (*wasmtime.Linker, error) { + linker := wasmtime.NewLinker(m.engine) + cleanupLinker := true + defer func() { + if cleanupLinker { + linker.Close() + } + }() linker.AllowShadowing(true) err := linker.DefineWasi() @@ -64,7 +79,7 @@ func newDagWasiLinker(ctx context.Context, modCfg *ModuleConfig, engine *wasmtim err = linker.FuncWrap( "wasi_snapshot_preview1", "poll_oneoff", - createPollOneoff(ctx, modCfg), + createPollOneoff(ctx, limiterOrDefault(m.cfg.MaxSubscriptionsLimiter, m.defaultLimiters.maxSubscriptions)), ) if err != nil { return nil, err @@ -79,17 +94,18 @@ func newDagWasiLinker(ctx context.Context, modCfg *ModuleConfig, engine *wasmtim return nil, err } - if modCfg.Determinism != nil { + if m.cfg.Determinism != nil { err = linker.FuncWrap( "wasi_snapshot_preview1", "random_get", - createRandomGet(modCfg), + createRandomGet(m.cfg), ) if err != nil { return nil, err } } + cleanupLinker = false return linker, nil } @@ -138,12 +154,12 @@ const ( // https://github.com/WebAssembly/WASI/blob/snapshot-01/phases/snapshot/docs.md // This implementation only responds to clock events, not to file descriptor notifications. // It doesn't actually sleep though, and will instead advance our fake clock by the sleep duration. -func createPollOneoff(ctx context.Context, cfg *ModuleConfig) func(caller *wasmtime.Caller, subscriptionptr int32, eventsptr int32, nsubscriptions int32, resultNevents int32) int32 { +func createPollOneoff(ctx context.Context, limiter limits.BoundLimiter[int]) func(caller *wasmtime.Caller, subscriptionptr int32, eventsptr int32, nsubscriptions int32, resultNevents int32) int32 { return func(caller *wasmtime.Caller, subscriptionptr int32, eventsptr int32, nsubscriptions int32, resultNevents int32) int32 { if nsubscriptions <= 0 || nsubscriptions > max(math.MaxInt32/subscriptionLen, math.MaxInt32/eventsLen) { return ErrnoInval } - if err := cfg.MaxSubscriptionsLimiter.Check(ctx, int(nsubscriptions)); err != nil { + if err := limiter.Check(ctx, int(nsubscriptions)); err != nil { return ErrnoInval } diff --git a/pkg/workflows/wasm/host/wasm.go b/pkg/workflows/wasm/host/wasm.go index d8c4bae7f1..45de99a78d 100644 --- a/pkg/workflows/wasm/host/wasm.go +++ b/pkg/workflows/wasm/host/wasm.go @@ -19,6 +19,7 @@ func GetWorkflowSpec(ctx context.Context, modCfg *ModuleConfig, binary []byte, c } m.Start() + defer m.Close() rid := uuid.New().String() req := &legacywasmpb.Request{ @@ -38,7 +39,5 @@ func GetWorkflowSpec(ctx context.Context, modCfg *ModuleConfig, binary []byte, c return nil, errors.New("unexpected response from WASM binary: got nil spec response") } - m.Close() - return legacywasmpb.ProtoToWorkflowSpec(sr) }