defer free()

Author: justinkaseman

pkg/workflows/wasm/host/execution.go

@@ -23,7 +23,6 @@ type execution[T any] struct {
 	capabilityResponses  map[int32]<-chan *sdkpb.CapabilityResponse
 	secretsResponses     map[int32]<-chan *secretsResponse
 	pendingCallsLimiter  limits.ResourcePoolLimiter[int]
-	pendingCallsFree     map[int32]func()
 	lock                 sync.RWMutex
 	module               *module
 	executor             ExecutionHelper
@@ -51,9 +50,10 @@ func (e *execution[T]) callCapAsync(ctx context.Context, req *sdkpb.CapabilityRe
 	e.lock.Lock()
 	defer e.lock.Unlock()
 	e.capabilityResponses[req.CallbackId] = ch
-	e.pendingCallsFree[req.CallbackId] = free
 
 	go func() {
+		defer free()
+
 		resp, err := e.executor.CallCapability(ctx, req)
 
 		if err != nil {
@@ -92,10 +92,6 @@ func (e *execution[T]) awaitCapabilities(ctx context.Context, acr *sdkpb.AwaitCa
 		}
 
 		delete(e.capabilityResponses, callId)
-		if free, ok := e.pendingCallsFree[callId]; ok {
-			free()
-			delete(e.pendingCallsFree, callId)
-		}
 	}
 
 	return &sdkpb.AwaitCapabilitiesResponse{
@@ -119,9 +115,10 @@ func (e *execution[T]) getSecretsAsync(ctx context.Context, req *sdkpb.GetSecret
 	e.lock.Lock()
 	defer e.lock.Unlock()
 	e.secretsResponses[req.CallbackId] = ch
-	e.pendingCallsFree[req.CallbackId] = free
 
 	go func() {
+		defer free()
+
 		resp, err := e.executor.GetSecrets(ctx, req)
 		sr := &secretsResponse{responses: resp, err: err}
 
@@ -157,10 +154,6 @@ func (e *execution[T]) awaitSecrets(ctx context.Context, acr *sdkpb.AwaitSecrets
 		}
 
 		delete(e.secretsResponses, callId)
-		if free, ok := e.pendingCallsFree[callId]; ok {
-			free()
-			delete(e.pendingCallsFree, callId)
-		}
 	}
 
 	return &sdkpb.AwaitSecretsResponse{

pkg/workflows/wasm/host/execution_await_order_test.go

@@ -71,7 +71,6 @@ func TestAwaitCapabilities_headOfLineBlocksOnEarlierID(t *testing.T) {
 		ctx:                 t.Context(),
 		capabilityResponses: make(map[int32]<-chan *sdkpb.CapabilityResponse),
 		pendingCallsLimiter: limits.GlobalResourcePoolLimiter(defaultMaxPendingCalls),
-		pendingCallsFree:    map[int32]func(){},
 		executor:            stub,
 	}
 

pkg/workflows/wasm/host/module.go

@@ -72,9 +72,9 @@ type ModuleConfig struct {
 	IsUncompressed   bool
 	Fetch            func(ctx context.Context, req *FetchRequest) (*FetchResponse, error)
 	MaxFetchRequests int
-	// MaxPendingCalls bounds the number of concurrent in-flight capability call
-	// goroutines per execution. Additional calls block until a slot is freed.
-	MaxPendingCalls              int
+	// MaxPendingCalls bounds concurrent in-flight capability calls per workflow.
+	MaxPendingCalls int
+	// When PendingCallsLimiter is set, it enforces a separate pending calls pool per workflow ID.
 	PendingCallsLimiter          limits.ResourcePoolLimiter[int] // supersedes MaxPendingCalls if set
 	MaxCompressedBinarySize      uint64
 	MaxCompressedBinaryLimiter   limits.BoundLimiter[config.Size] // supersedes MaxCompressedBinarySize if set
@@ -711,7 +711,6 @@ func runWasm[I, O proto.Message](
 		capabilityResponses: map[int32]<-chan *sdkpb.CapabilityResponse{},
 		secretsResponses:    map[int32]<-chan *secretsResponse{},
 		pendingCallsLimiter: m.cfg.PendingCallsLimiter,
-		pendingCallsFree:    map[int32]func(){},
 		module:              m,
 		executor:            helper,
 		donSeed:             donSeed,

pkg/workflows/wasm/host/module_test.go

@@ -635,7 +635,6 @@ func Test_CallAwaitRace(t *testing.T) {
 		module:              m,
 		capabilityResponses: map[int32]<-chan *sdkpb.CapabilityResponse{},
 		pendingCallsLimiter: limits.GlobalResourcePoolLimiter[int](defaultMaxPendingCalls),
-		pendingCallsFree:    map[int32]func(){},
 		ctx:                 t.Context(),
 		executor:            mockExecHelper,
 	}