mirror of
https://github.com/tiennm99/goclaw.git
synced 2026-10-13 02:14:39 +00:00
feat(webhooks): configurable agent run timeout (default 600s) (#1267)
Replace the hardcoded 30s webhook agent-run deadline with a configurable
timeout (default 600s, cap 3600s) for both the async worker and the
sync/test HTTP handler. Legitimate multi-step runs were dying at 30s
mid-tool-call; the same request over WebSocket completed fine.
- internal/webhooks/timeout.go: ResolveTimeoutSec helper (<=0 → 600s, cap 3600s)
- WorkerConfig.AsyncAgentTimeout (async worker) + WebhookLLMHandler.syncTimeout
(sync + admin test), wired from config
- config keys gateway.webhook_{async,sync}_timeout_sec, also settable via
GOCLAW_WEBHOOK_{ASYNC,SYNC}_TIMEOUT_SEC env (env overrides config)
- tests: timeout bounds + config file/env override
This commit is contained in:
1 parent
eaa011dae3
commit
a10464290d
11 files changed
+127
-8
No files matched your search
@@ -16,6 +16,7 @@ import (
|
||||
"github.com/nextlevelbuilder/goclaw/internal/store"
|
||||
"github.com/nextlevelbuilder/goclaw/internal/store/pg"
|
||||
"github.com/nextlevelbuilder/goclaw/internal/tools"
|
||||
"github.com/nextlevelbuilder/goclaw/internal/webhooks"
|
||||
)
|
||||
|
||||
// httpHandlers bundles the results of wireHTTP() for passing to wireHTTPHandlersOnServer.
|
||||
@@ -232,6 +233,7 @@ func (d *gatewayDeps) wireHTTPHandlersOnServer(
|
||||
d.pgStores.Webhooks,
|
||||
sharedWebhookLimiter, // K10: shared limiter
|
||||
nil, // lane: nil → internal default (4-slot); configurable in future via cfg
|
||||
webhooks.ResolveTimeoutSec(d.cfg.Gateway.WebhookSyncTimeoutSec),
|
||||
)
|
||||
llmH.SetEncKey(webhookEncKey) // K6: decrypt secret at HMAC verify time
|
||||
d.server.SetWebhookLLMHandler(llmH)
|
||||
|
||||
@@ -172,6 +172,7 @@ func (d *gatewayDeps) runLifecycle(
|
||||
webhooks.WorkerConfig{
|
||||
WorkerConcurrency: workerConcurrency,
|
||||
PerTenantConcurrency: 4,
|
||||
AsyncAgentTimeout: webhooks.ResolveTimeoutSec(d.cfg.Gateway.WebhookAsyncTimeoutSec),
|
||||
},
|
||||
)
|
||||
// K6: decrypt raw secret for outbound HMAC signing using the same key as inbound verify.
|
||||
|
||||
+12
-1
@@ -254,7 +254,18 @@ Triggers an agent with an input prompt. Available in all editions.
|
||||
}
|
||||
```
|
||||
|
||||
Sync mode times out at **30 seconds**. On timeout: `504 Gateway Timeout` with `webhook.llm_timeout`.
|
||||
Sync mode times out after the configured deadline (default **600s**). On timeout: `504 Gateway Timeout` with `webhook.llm_timeout`.
|
||||
|
||||
#### Agent-run timeouts (configurable)
|
||||
|
||||
Both webhook agent-run deadlines default to **600s** and are capped at **3600s**. Configure via `config.json` or environment variables (env overrides config):
|
||||
|
||||
| Setting (config) | Env var | Applies to | Default |
|
||||
|------------------|---------|-----------|---------|
|
||||
| `gateway.webhook_sync_timeout_sec` | `GOCLAW_WEBHOOK_SYNC_TIMEOUT_SEC` | sync + admin test calls | 600 |
|
||||
| `gateway.webhook_async_timeout_sec` | `GOCLAW_WEBHOOK_ASYNC_TIMEOUT_SEC` | async worker runs | 600 |
|
||||
|
||||
> Sync mode holds the HTTP connection open for the whole run — a value above an upstream proxy/load-balancer read timeout may be cut before the agent finishes. Async mode returns `202` immediately and runs in the background, so a longer deadline is safe.
|
||||
|
||||
### Async Response — 202 Accepted
|
||||
|
||||
|
||||
@@ -426,6 +426,8 @@ type GatewayConfig struct {
|
||||
ChatBehavior *ChatBehaviorConfig `json:"chat_behavior,omitempty"` // human-like channel delivery behavior (default disabled)
|
||||
ToolStatus *bool `json:"tool_status,omitempty"` // show tool name in streaming preview during tool execution (default true)
|
||||
TaskRecoveryIntervalSec int `json:"task_recovery_interval_sec,omitempty"` // team task recovery ticker interval in seconds (default 300 = 5min)
|
||||
WebhookAsyncTimeoutSec int `json:"webhook_async_timeout_sec,omitempty"` // async webhook worker agent-run deadline in seconds (default 600, cap 3600)
|
||||
WebhookSyncTimeoutSec int `json:"webhook_sync_timeout_sec,omitempty"` // sync + test webhook handler agent-run deadline in seconds (default 600, cap 3600). NOTE: sync holds the HTTP connection open for this duration — a value above an upstream proxy/LB read timeout may be cut.
|
||||
BackgroundProvider string `json:"background_provider,omitempty"` // LLM provider for background workers (vault enrichment, consolidation)
|
||||
BackgroundModel string `json:"background_model,omitempty"` // LLM model for background workers
|
||||
PublicURL string `json:"public_url,omitempty"` // public base URL for OAuth callbacks (e.g. "https://goclaw.example.com")
|
||||
|
||||
@@ -251,6 +251,18 @@ func (c *Config) applyEnvOverrides() {
|
||||
c.Skills.MaxUploadSizeMB = ClampSkillMaxUploadSizeMB(mb)
|
||||
}
|
||||
}
|
||||
// Webhook agent-run timeouts (seconds). Bounds (default 600, cap 3600) are
|
||||
// applied at consumption via webhooks.ResolveTimeoutSec.
|
||||
if v := os.Getenv("GOCLAW_WEBHOOK_ASYNC_TIMEOUT_SEC"); v != "" {
|
||||
if sec, err := strconv.Atoi(v); err == nil && sec > 0 {
|
||||
c.Gateway.WebhookAsyncTimeoutSec = sec
|
||||
}
|
||||
}
|
||||
if v := os.Getenv("GOCLAW_WEBHOOK_SYNC_TIMEOUT_SEC"); v != "" {
|
||||
if sec, err := strconv.Atoi(v); err == nil && sec > 0 {
|
||||
c.Gateway.WebhookSyncTimeoutSec = sec
|
||||
}
|
||||
}
|
||||
envBoolPtr := func(key string, dst **bool) {
|
||||
if v := os.Getenv(key); v != "" {
|
||||
b := parseEnvBool(v)
|
||||
|
||||
@@ -119,6 +119,37 @@ func TestLoad_EnvVarOverrides(t *testing.T) {
|
||||
}
|
||||
}
|
||||
|
||||
func TestLoad_WebhookTimeoutsFromFileAndEnv(t *testing.T) {
|
||||
dir := t.TempDir()
|
||||
cfgPath := filepath.Join(dir, "config.json5")
|
||||
os.WriteFile(cfgPath, []byte(`{"gateway":{"webhook_async_timeout_sec":120,"webhook_sync_timeout_sec":90}}`), 0644)
|
||||
|
||||
cfg, err := Load(cfgPath)
|
||||
if err != nil {
|
||||
t.Fatalf("load error: %v", err)
|
||||
}
|
||||
if cfg.Gateway.WebhookAsyncTimeoutSec != 120 {
|
||||
t.Fatalf("file async timeout: got %d, want 120", cfg.Gateway.WebhookAsyncTimeoutSec)
|
||||
}
|
||||
if cfg.Gateway.WebhookSyncTimeoutSec != 90 {
|
||||
t.Fatalf("file sync timeout: got %d, want 90", cfg.Gateway.WebhookSyncTimeoutSec)
|
||||
}
|
||||
|
||||
// Env overrides the file values.
|
||||
t.Setenv("GOCLAW_WEBHOOK_ASYNC_TIMEOUT_SEC", "300")
|
||||
t.Setenv("GOCLAW_WEBHOOK_SYNC_TIMEOUT_SEC", "240")
|
||||
cfg, err = Load(cfgPath)
|
||||
if err != nil {
|
||||
t.Fatalf("load with env error: %v", err)
|
||||
}
|
||||
if cfg.Gateway.WebhookAsyncTimeoutSec != 300 {
|
||||
t.Fatalf("env async timeout: got %d, want 300", cfg.Gateway.WebhookAsyncTimeoutSec)
|
||||
}
|
||||
if cfg.Gateway.WebhookSyncTimeoutSec != 240 {
|
||||
t.Fatalf("env sync timeout: got %d, want 240", cfg.Gateway.WebhookSyncTimeoutSec)
|
||||
}
|
||||
}
|
||||
|
||||
func TestLoad_SkillsMaxUploadSizeFromFileAndEnv(t *testing.T) {
|
||||
dir := t.TempDir()
|
||||
cfgPath := filepath.Join(dir, "config.json5")
|
||||
|
||||
@@ -52,6 +52,8 @@ func (c *Config) ApplySystemConfigs(configs map[string]string) {
|
||||
boolean("gateway.block_reply", &c.Gateway.BlockReply)
|
||||
boolean("gateway.tool_status", &c.Gateway.ToolStatus)
|
||||
integer("gateway.task_recovery_interval_sec", &c.Gateway.TaskRecoveryIntervalSec)
|
||||
integer("gateway.webhook_async_timeout_sec", &c.Gateway.WebhookAsyncTimeoutSec)
|
||||
integer("gateway.webhook_sync_timeout_sec", &c.Gateway.WebhookSyncTimeoutSec)
|
||||
|
||||
// Background workers (vault enrichment, consolidation)
|
||||
str("background.provider", &c.Gateway.BackgroundProvider)
|
||||
|
||||
@@ -21,8 +21,9 @@ import (
|
||||
)
|
||||
|
||||
const (
|
||||
// webhookLLMTimeout is the hard deadline for synchronous LLM invocations.
|
||||
webhookLLMTimeout = 30 * time.Second
|
||||
// webhookLLMTimeout is the default hard deadline for webhook LLM invocations
|
||||
// (sync + admin test). Overridable per-handler via syncTimeout from config.
|
||||
webhookLLMTimeout = 600 * time.Second
|
||||
|
||||
// webhookLLMResponseTruncate is the maximum bytes stored in the audit row response column.
|
||||
webhookLLMResponseTruncate = 32 * 1024
|
||||
@@ -100,7 +101,8 @@ type WebhookLLMHandler struct {
|
||||
limiter *webhookLimiter
|
||||
lane *scheduler.Lane
|
||||
encKey string // AES-256-GCM key for decrypting encrypted_secret at HMAC verify time
|
||||
// syncTimeout overrides webhookLLMTimeout (30s) — set in tests only.
|
||||
// syncTimeout overrides the webhookLLMTimeout default (600s); set from
|
||||
// gateway.webhook_sync_timeout_sec config (and in tests). 0 → use default.
|
||||
syncTimeout time.Duration
|
||||
}
|
||||
|
||||
@@ -112,6 +114,7 @@ func NewWebhookLLMHandler(
|
||||
webhooks store.WebhookStore,
|
||||
limiter *webhookLimiter,
|
||||
lane *scheduler.Lane,
|
||||
syncTimeout time.Duration,
|
||||
) *WebhookLLMHandler {
|
||||
if lane == nil {
|
||||
lane = scheduler.NewLane(webhookLaneName, webhookLaneDefaultConcurrency)
|
||||
@@ -122,6 +125,7 @@ func NewWebhookLLMHandler(
|
||||
webhooks: webhooks,
|
||||
limiter: limiter,
|
||||
lane: lane,
|
||||
syncTimeout: syncTimeout,
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
@@ -0,0 +1,24 @@
|
||||
package webhooks
|
||||
|
||||
import "time"
|
||||
|
||||
const (
|
||||
// DefaultAgentTimeout is the fallback webhook agent-run deadline when config is unset.
|
||||
DefaultAgentTimeout = 600 * time.Second
|
||||
// MaxAgentTimeout caps the configurable webhook agent-run deadline (1h) so a config
|
||||
// typo cannot hold worker/lane slots indefinitely.
|
||||
MaxAgentTimeout = 3600 * time.Second
|
||||
)
|
||||
|
||||
// ResolveTimeoutSec converts a config value (seconds) into a bounded duration:
|
||||
// <= 0 → DefaultAgentTimeout; > 3600 → MaxAgentTimeout; otherwise the value.
|
||||
func ResolveTimeoutSec(sec int) time.Duration {
|
||||
if sec <= 0 {
|
||||
return DefaultAgentTimeout
|
||||
}
|
||||
d := time.Duration(sec) * time.Second
|
||||
if d > MaxAgentTimeout {
|
||||
return MaxAgentTimeout
|
||||
}
|
||||
return d
|
||||
}
|
||||
@@ -0,0 +1,27 @@
|
||||
package webhooks
|
||||
|
||||
import (
|
||||
"testing"
|
||||
"time"
|
||||
)
|
||||
|
||||
func TestResolveTimeoutSec(t *testing.T) {
|
||||
cases := []struct {
|
||||
name string
|
||||
in int
|
||||
want time.Duration
|
||||
}{
|
||||
{"zero uses default", 0, 600 * time.Second},
|
||||
{"negative uses default", -5, 600 * time.Second},
|
||||
{"passthrough", 120, 120 * time.Second},
|
||||
{"caps at 3600", 99999, 3600 * time.Second},
|
||||
{"exactly cap", 3600, 3600 * time.Second},
|
||||
}
|
||||
for _, c := range cases {
|
||||
t.Run(c.name, func(t *testing.T) {
|
||||
if got := ResolveTimeoutSec(c.in); got != c.want {
|
||||
t.Fatalf("ResolveTimeoutSec(%d) = %v, want %v", c.in, got, c.want)
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
@@ -54,9 +54,6 @@ const (
|
||||
// callbackResponseStorageLimit is the max bytes stored in webhook_calls.response.
|
||||
callbackResponseStorageLimit = 32 * 1024 // 32 KB
|
||||
|
||||
// asyncAgentTimeout is the max time to invoke the LLM agent for async_llm mode.
|
||||
asyncAgentTimeout = 30 * time.Second
|
||||
|
||||
// retryAfterCap caps the Retry-After header value to 6 hours.
|
||||
retryAfterCap = 6 * time.Hour
|
||||
)
|
||||
@@ -117,6 +114,9 @@ type WorkerConfig struct {
|
||||
// PerTenantConcurrency is the per-tenant cap passed to CallbackLimiter.
|
||||
// 0 = default (4).
|
||||
PerTenantConcurrency int
|
||||
|
||||
// AsyncAgentTimeout is the deadline for each async agent run. 0 → DefaultAgentTimeout (600s).
|
||||
AsyncAgentTimeout time.Duration
|
||||
}
|
||||
|
||||
// WebhookWorker is the background callback delivery service. It is started once per
|
||||
@@ -152,6 +152,9 @@ func NewWebhookWorker(
|
||||
if cfg.WorkerConcurrency <= 0 {
|
||||
cfg.WorkerConcurrency = 4
|
||||
}
|
||||
if cfg.AsyncAgentTimeout <= 0 {
|
||||
cfg.AsyncAgentTimeout = DefaultAgentTimeout
|
||||
}
|
||||
if limiter == nil {
|
||||
limiter = NewCallbackLimiter(cfg.PerTenantConcurrency)
|
||||
}
|
||||
@@ -762,7 +765,7 @@ func (w *WebhookWorker) invokeAgent(
|
||||
TraceTags: []string{"webhook", "async"},
|
||||
}
|
||||
|
||||
runCtx, cancel := context.WithTimeout(context.WithoutCancel(ctx), asyncAgentTimeout)
|
||||
runCtx, cancel := context.WithTimeout(context.WithoutCancel(ctx), w.cfg.AsyncAgentTimeout)
|
||||
defer cancel()
|
||||
|
||||
result, runErr := ag.Run(runCtx, rr)
|
||||
|
||||
Reference in new issue
Block a user