Merge branch 'main' into solana-logtrigger-cap-bump

Author: silaslenihan

go.mod

@@ -7,9 +7,7 @@ require (
 	github.com/XSAM/otelsql v0.37.0
 	github.com/andybalholm/brotli v1.1.1
 	github.com/atombender/go-jsonschema v0.16.1-0.20240916205339-a74cd4e2851c
-	github.com/buraksezer/consistent v0.10.0
 	github.com/bytecodealliance/wasmtime-go/v28 v28.0.0
-	github.com/cespare/xxhash/v2 v2.3.0
 	github.com/davecgh/go-spew v1.1.2-0.20180830191138-d8f796af33cc
 	github.com/dominikbraun/graph v0.23.0
 	github.com/fxamacker/cbor/v2 v2.7.0
@@ -90,6 +88,7 @@ require (
 	github.com/buger/goterm v1.0.4 // indirect
 	github.com/buger/jsonparser v1.1.1 // indirect
 	github.com/cenkalti/backoff/v5 v5.0.2 // indirect
+	github.com/cespare/xxhash/v2 v2.3.0 // indirect
 	github.com/cloudevents/sdk-go/binding/format/protobuf/v2 v2.16.1 // indirect
 	github.com/cloudevents/sdk-go/v2 v2.16.1 // indirect
 	github.com/fatih/color v1.18.0 // indirect

go.sum

@@ -28,8 +28,6 @@ github.com/buger/goterm v1.0.4 h1:Z9YvGmOih81P0FbVtEYTFF6YsSgxSUKEhf/f9bTMXbY=
 github.com/buger/goterm v1.0.4/go.mod h1:HiFWV3xnkolgrBV3mY8m0X0Pumt4zg4QhbdOzQtB8tE=
 github.com/buger/jsonparser v1.1.1 h1:2PnMjfWD7wBILjqQbt530v576A/cAbQvEW9gGIpYMUs=
 github.com/buger/jsonparser v1.1.1/go.mod h1:6RYKKt7H4d4+iWqouImQ9R2FZql3VbhNgx27UK13J/0=
-github.com/buraksezer/consistent v0.10.0 h1:hqBgz1PvNLC5rkWcEBVAL9dFMBWz6I0VgUCW25rrZlU=
-github.com/buraksezer/consistent v0.10.0/go.mod h1:6BrVajWq7wbKZlTOUPs/XVfR8c0maujuPowduSpZqmw=
 github.com/bytecodealliance/wasmtime-go/v28 v28.0.0 h1:aBU8cexP2rPZ0Qz488kvn2NXvWZHL2aG1/+n7Iv+xGc=
 github.com/bytecodealliance/wasmtime-go/v28 v28.0.0/go.mod h1:4OCU0xAW9ycwtX4nMF4zxwgJBJ5/0eMfJiHB0wAmkV4=
 github.com/cenkalti/backoff/v5 v5.0.2 h1:rIfFVxEf1QsI7E1ZHfp/B4DF/6QBAUhmgkxc0H7Zss8=

pkg/capabilities/registry/base.go

@@ -356,20 +356,22 @@ func (a *atomicTriggerCapability) Update(c capabilities.BaseCapability) error {
 
 func (a *atomicTriggerCapability) Info(ctx context.Context) (capabilities.CapabilityInfo, error) {
 	a.mu.RLock()
-	defer a.mu.RUnlock()
-	if a.cap == nil {
+	cap := a.cap
+	a.mu.RUnlock()
+	if cap == nil {
 		return capabilities.CapabilityInfo{}, errors.New("capability unavailable")
 	}
-	return a.cap.Info(ctx)
+	return cap.Info(ctx)
 }
 
 func (a *atomicTriggerCapability) GetState() connectivity.State {
 	a.mu.RLock()
-	defer a.mu.RUnlock()
-	if a.cap == nil {
+	cap := a.cap
+	a.mu.RUnlock()
+	if cap == nil {
 		return connectivity.Shutdown
 	}
-	if sg, ok := a.cap.(StateGetter); ok {
+	if sg, ok := cap.(StateGetter); ok {
 		return sg.GetState()
 	}
 	return connectivity.State(-1) // unknown
@@ -427,20 +429,22 @@ func (a *atomicExecuteCapability) Update(c capabilities.BaseCapability) error {
 
 func (a *atomicExecuteCapability) Info(ctx context.Context) (capabilities.CapabilityInfo, error) {
 	a.mu.RLock()
-	defer a.mu.RUnlock()
-	if a.cap == nil {
+	cap := a.cap
+	a.mu.RUnlock()
+	if cap == nil {
 		return capabilities.CapabilityInfo{}, errors.New("capability unavailable")
 	}
-	return a.cap.Info(ctx)
+	return cap.Info(ctx)
 }
 
 func (a *atomicExecuteCapability) GetState() connectivity.State {
 	a.mu.RLock()
-	defer a.mu.RUnlock()
-	if a.cap == nil {
+	cap := a.cap
+	a.mu.RUnlock()
+	if cap == nil {
 		return connectivity.Shutdown
 	}
-	if sg, ok := a.cap.(StateGetter); ok {
+	if sg, ok := cap.(StateGetter); ok {
 		return sg.GetState()
 	}
 	return connectivity.State(-1) // unknown
@@ -458,29 +462,32 @@ func (a *atomicExecuteCapability) Load() *capabilities.ExecutableCapability {
 
 func (a *atomicExecuteCapability) RegisterToWorkflow(ctx context.Context, request capabilities.RegisterToWorkflowRequest) error {
 	a.mu.RLock()
-	defer a.mu.RUnlock()
-	if a.cap == nil {
+	cap := a.cap
+	a.mu.RUnlock()
+	if cap == nil {
 		return errors.New("capability unavailable")
 	}
-	return a.cap.RegisterToWorkflow(ctx, request)
+	return cap.RegisterToWorkflow(ctx, request)
 }
 
 func (a *atomicExecuteCapability) UnregisterFromWorkflow(ctx context.Context, request capabilities.UnregisterFromWorkflowRequest) error {
 	a.mu.RLock()
-	defer a.mu.RUnlock()
-	if a.cap == nil {
+	cap := a.cap
+	a.mu.RUnlock()
+	if cap == nil {
 		return errors.New("capability unavailable")
 	}
-	return a.cap.UnregisterFromWorkflow(ctx, request)
+	return cap.UnregisterFromWorkflow(ctx, request)
 }
 
 func (a *atomicExecuteCapability) Execute(ctx context.Context, request capabilities.CapabilityRequest) (capabilities.CapabilityResponse, error) {
 	a.mu.RLock()
-	defer a.mu.RUnlock()
-	if a.cap == nil {
+	cap := a.cap
+	a.mu.RUnlock()
+	if cap == nil {
 		return capabilities.CapabilityResponse{}, errors.New("capability unavailable")
 	}
-	return a.cap.Execute(ctx, request)
+	return cap.Execute(ctx, request)
 }
 
 var _ capabilities.ExecutableAndTriggerCapability = &atomicExecuteAndTriggerCapability{}
@@ -516,20 +523,22 @@ func (a *atomicExecuteAndTriggerCapability) Update(c capabilities.BaseCapability
 
 func (a *atomicExecuteAndTriggerCapability) Info(ctx context.Context) (capabilities.CapabilityInfo, error) {
 	a.mu.RLock()
-	defer a.mu.RUnlock()
-	if a.cap == nil {
+	cap := a.cap
+	a.mu.RUnlock()
+	if cap == nil {
 		return capabilities.CapabilityInfo{}, errors.New("capability unavailable")
 	}
-	return a.cap.Info(ctx)
+	return cap.Info(ctx)
 }
 
 func (a *atomicExecuteAndTriggerCapability) GetState() connectivity.State {
 	a.mu.RLock()
-	defer a.mu.RUnlock()
+	cap := a.cap
+	a.mu.RUnlock()
 	if a.cap == nil {
 		return connectivity.Shutdown
 	}
-	if sg, ok := a.cap.(StateGetter); ok {
+	if sg, ok := cap.(StateGetter); ok {
 		return sg.GetState()
 	}
 	return connectivity.State(-1) // unknown
@@ -565,27 +574,30 @@ func (a *atomicExecuteAndTriggerCapability) UnregisterTrigger(ctx context.Contex
 
 func (a *atomicExecuteAndTriggerCapability) RegisterToWorkflow(ctx context.Context, request capabilities.RegisterToWorkflowRequest) error {
 	a.mu.RLock()
-	defer a.mu.RUnlock()
-	if a.cap == nil {
+	cap := a.cap
+	a.mu.RUnlock()
+	if cap == nil {
 		return errors.New("capability unavailable")
 	}
-	return a.cap.RegisterToWorkflow(ctx, request)
+	return cap.RegisterToWorkflow(ctx, request)
 }
 
 func (a *atomicExecuteAndTriggerCapability) UnregisterFromWorkflow(ctx context.Context, request capabilities.UnregisterFromWorkflowRequest) error {
 	a.mu.RLock()
-	defer a.mu.RUnlock()
-	if a.cap == nil {
+	cap := a.cap
+	a.mu.RUnlock()
+	if cap == nil {
 		return errors.New("capability unavailable")
 	}
-	return a.cap.UnregisterFromWorkflow(ctx, request)
+	return cap.UnregisterFromWorkflow(ctx, request)
 }
 
 func (a *atomicExecuteAndTriggerCapability) Execute(ctx context.Context, request capabilities.CapabilityRequest) (capabilities.CapabilityResponse, error) {
 	a.mu.RLock()
-	defer a.mu.RUnlock()
-	if a.cap == nil {
+	cap := a.cap
+	a.mu.RUnlock()
+	if cap == nil {
 		return capabilities.CapabilityResponse{}, errors.New("capability unavailable")
 	}
-	return a.cap.Execute(ctx, request)
+	return cap.Execute(ctx, request)
 }

pkg/loop/config.go

@@ -229,8 +229,12 @@ func (e *EnvConfig) AsCmdEnv() (env []string) {
 	add(envChipIngressEndpoint, e.ChipIngressEndpoint)
 	add(envChipIngressInsecureConnection, strconv.FormatBool(e.ChipIngressInsecureConnection))
 
-	add(envCRESettings, e.CRESettings)
-	add(envCRESettingsDefault, e.CRESettingsDefault)
+	if e.CRESettings != "" {
+		add(envCRESettings, e.CRESettings)
+	}
+	if e.CRESettingsDefault != "" {
+		add(envCRESettingsDefault, e.CRESettingsDefault)
+	}
 
 	return
 }

pkg/settings/cresettings/README.md

@@ -0,0 +1,178 @@
+# CRE Settings
+
+```mermaid
+---
+title: Legend
+---
+flowchart
+    bound{{Bound}}
+    gate[/Gate\]
+    queue[[Queue]]
+    rate[\Rate/]
+    resource([Resource])
+    time>Time]
+    
+    bound:::bound
+    gate:::gate
+    queue:::queue
+    rate:::rate
+    resource:::resource
+    time:::time
+
+    classDef bound stroke:#f00
+    classDef gate stroke:#0f0
+    classDef queue stroke:#00f
+    classDef rate stroke:#ff0
+    classDef resource stroke:#f0f
+    classDef time stroke:#0ff
+```
+
+
+```mermaid
+---
+title: Limits
+---
+flowchart
+    subgraph handleRequest[httpServer/websocketServer.handleRequest]
+        GatewayIncomingPayloadSizeLimit{{GatewayIncomingPayloadSizeLimit}}:::bound
+    end
+%%    WorkflowLimit - Deprecated
+%%    TODO unused
+%%    PerOrg.ZeroBalancePruningTimeout
+
+    subgraph Store.FetchWorkflowArtifacts
+        PerWorkflow.WASMConfigSizeLimit{{PerWorkflow.WASMConfigSizeLimit}}:::bound
+        PerWorkflow.WASMBinarySizeLimit{{PerWorkflow.WASMBinarySizeLimit}}:::bound
+        PerWorkflow.WASMSecretsSizeLimit{{PerWorkflow.WASMSecretsSizeLimit}}:::bound
+    end
+
+    subgraph host.NewModule
+        PerWorkflow.WASMCompressedBinarySizeLimit{{PerWorkflow.WASMCompressedBinarySizeLimit}}:::bound
+    end
+    
+    subgraph Engine.init
+        WorkflowExecutionConcurrencyLimit([WorkflowExecutionConcurrencyLimit]):::resource
+        PerOwner.WorkflowExecutionConcurrencyLimit([PerOwner.WorkflowExecutionConcurrencyLimit]):::resource
+
+        WorkflowExecutionConcurrencyLimit-->PerOwner.WorkflowExecutionConcurrencyLimit
+    end
+    
+    subgraph Engine.runTriggerSubscriptionPhase
+
+        PerWorkflow.TriggerSubscriptionTimeout>PerWorkflow.TriggerSubscriptionTimeout]:::time
+        PerWorkflow.WASMMemoryLimit{{PerWorkflow.WASMMemoryLimit}}:::bound
+        PerWorkflow.TriggerRegistrationsTimeout>PerWorkflow.TriggerRegistrationsTimeout]:::time
+        PerWorkflow.TriggerSubscriptionLimit{{PerWorkflow.TriggerSubscriptionLimit}}:::bound
+
+        PerWorkflow.TriggerSubscriptionTimeout-->PerWorkflow.WASMMemoryLimit-->PerWorkflow.TriggerSubscriptionLimit-->PerWorkflow.TriggerRegistrationsTimeout
+    end
+
+    subgraph triggers
+        direction TB
+
+        subgraph PerWorkflow.CRONTrigger
+            FastestScheduleInterval>FastestScheduleInterval]:::time
+        end
+        subgraph PerWorkflow.HTTPTrigger
+            RateLimit[\RateLimit/]:::rate
+        end
+        subgraph PerWorkflow.LogTrigger
+            direction LR
+
+            EventRateLimit[\EventRateLimit/]:::rate
+            EventSizeLimit{{EventSizeLimit}}:::bound
+            FilterAddressLimit{{FilterAddressLimit}}:::bound
+            FilterTopicsPerSlotLimit{{FilterTopicsPerSlotLimit}}:::bound
+        end
+    end
+
+    subgraph Engine.handleAllTriggerEvents
+        PerWorkflow.TriggerEventQueueLimit[[PerWorkflow.TriggerEventQueueLimit]]:::queue
+        PerWorkflow.TriggerEventQueueTimeout>PerWorkflow.TriggerEventQueueTimeout]:::time
+        PerWorkflow.ExecutionConcurrencyLimit([PerWorkflow.ExecutionConcurrencyLimit]):::resource
+
+        PerWorkflow.TriggerEventQueueLimit-->PerWorkflow.TriggerEventQueueTimeout-->PerWorkflow.ExecutionConcurrencyLimit
+    end
+
+    subgraph Engine.startExecution
+        direction TB
+    
+        subgraph logs
+            PerWorkflow.LogLineLimit{{PerWorkflow.LogLineLimit}}:::bound
+            PerWorkflow.LogEventLimit{{PerWorkflow.LogEventLimit}}:::bound
+        end
+        
+        PerWorkflow.ExecutionTimeout>PerWorkflow.ExecutionTimeout]:::time
+        PerWorkflow.ExecutionResponseLimit{{PerWorkflow.ExecutionResponseLimit}}:::bound
+
+        PerWorkflow.ExecutionTimeout-->PerWorkflow.ExecutionResponseLimit
+    end
+        
+    subgraph ExecutionHelper.GetSecrets
+        PerWorkflow.SecretsConcurrencyLimit([PerWorkflow.SecretsConcurrencyLimit]):::resource
+    end
+    subgraph ExecutionHelper.CallCapability
+        PerWorkflow.ChainAllowed[/PerWorkflow.ChainAllowed\]:::gate
+        PerWorkflow.CapabilityConcurrencyLimit([PerWorkflow.CapabilityConcurrencyLimit]):::resource
+        PerWorkflow.CapabilityCallTimeout>PerWorkflow.CapabilityCallTimeout]:::time
+        
+        PerWorkflow.ChainAllowed-->PerWorkflow.CapabilityConcurrencyLimit-->PerWorkflow.CapabilityCallTimeout
+    end
+    
+    subgraph actions
+        direction TB
+        
+        subgraph PerWorkflow.ChainWrite
+            direction LR
+            
+            TargetsLimit{{TargetsLimit}}:::bound
+            ReportSizeLimit{{ReportSizeLimit}}:::bound
+
+            subgraph EVM
+                GasLimit{{GasLimit}}:::bound
+            end
+        end
+        subgraph PerWorkflow.ChainRead
+            direction LR
+            chainread.CallLimit{{CallLimit}}:::bound
+            LogQueryBlockLimit{{LogQueryBlockLimit}}:::bound
+            PayloadSizeLimit{{PayloadSizeLimit}}:::bound
+        end
+        subgraph PerWorkflow.Consensus
+            ObservationSizeLimit{{ObservationSizeLimit}}:::bound
+            consensus.CallLimit{{CallLimit}}:::bound
+        end
+        subgraph PerWorkflow.HTTPAction
+            direction LR
+            
+            httpaction.CallLimit{{CallLimit}}:::bound
+            CacheAgeLimit{{CacheAgeLimit}}:::bound
+            ConnectionTimeout{{ConnectionTimeout}}:::bound
+            RequestSizeLimit{{RequestSizeLimit}}:::bound
+            ResponseSizeLimit{{ResponseSizeLimit}}:::bound
+        end
+        subgraph PerWorkflow.Secrets
+            secrets.CallLimit{{CallLimit}}:::bound
+        end
+    end
+    subgraph vault
+        VaultCiphertextSizeLimit{{VaultCiphertextSizeLimit}}:::bound
+        VaultIdentifierKeySizeLimit{{VaultIdentifierKeySizeLimit}}:::bound
+        VaultIdentifierOwnerSizeLimit{{VaultIdentifierOwnerSizeLimit}}:::bound
+        VaultIdentifierNamespaceSizeLimit{{VaultIdentifierNamespaceSizeLimit}}:::bound
+        VaultPluginBatchSizeLimit{{VaultPluginBatchSizeLimit}}:::bound
+        VaultRequestBatchSizeLimit{{VaultRequestBatchSizeLimit}}:::bound
+        PerOwner.VaultSecretsLimit{{PerOwner.VaultSecretsLimit}}:::bound
+    end
+
+    handleRequest-->Store.FetchWorkflowArtifacts-->host.NewModule-->Engine.init-->Engine.runTriggerSubscriptionPhase-->triggers-->Engine.handleAllTriggerEvents-->Engine.startExecution
+    Engine.startExecution-->ExecutionHelper.CallCapability-->actions
+    Engine.startExecution-->PerWorkflow.SecretsConcurrencyLimit-->vault
+
+    classDef bound stroke:#f00
+    classDef gate stroke:#0f0
+    classDef queue stroke:#00f
+    classDef rate stroke:#ff0
+    classDef resource stroke:#f0f
+    classDef time stroke:#0ff
+```