diff --git a/docs/00-architecture-overview.md b/docs/00-architecture-overview.md index bf4c553b..a14beee6 100644 --- a/docs/00-architecture-overview.md +++ b/docs/00-architecture-overview.md @@ -389,6 +389,35 @@ flowchart TD | `team` | 100 | `GOCLAW_LANE_TEAM` | Agent team/delegation executions | | `cron` | 30 | `GOCLAW_LANE_CRON` | Scheduled cron jobs | +### Scaling Beyond the Lane Defaults + +`main` bounds how many users can be answered at once, but raising it alone just +moves the queue downstream. These knobs need to move with it: + +| Setting | Default | Env Override | Why it binds | +|---------|:-------:|--------------|--------------| +| Postgres max open conns | 25 | `GOCLAW_PG_MAX_OPEN_CONNS` | Held per query, not per run — undersizing shows up as burst latency when many runs load context at once | +| Postgres max idle conns | 10 | `GOCLAW_PG_MAX_IDLE_CONNS` | Clamped to max open | +| Provider idle conns | 100 | `GOCLAW_HTTP_MAX_IDLE_CONNS` | Total across all provider hosts | +| Provider idle conns per host | 10 | `GOCLAW_HTTP_MAX_IDLE_CONNS_PER_HOST` | When one provider takes nearly all traffic, every request past this limit pays a fresh TCP+TLS handshake. Self-hosted/in-network providers should match `main`. | +| Outbound dispatch shards | 8 | `GOCLAW_OUTBOUND_SHARDS` | See below | + +Channel-side rate limits are metered by the *remote* platform and are not +covered by any of the above — DingTalk's AI Card API, for instance, meters per +app, so `card_update_interval_ms` decides how many conversations can stream +concurrently regardless of local capacity. + +### Outbound Dispatch + +Outbound delivery is sharded by `channel + ChatID`. A conversation always maps +to the same shard and is delivered serially there, so the ordering a run +depends on — block replies, retry notices, then the final answer — is +preserved. Unrelated conversations run in parallel, so one slow send (a media +upload of several seconds) no longer stalls every other reply in the process. + +Raising `GOCLAW_OUTBOUND_SHARDS` past the point where the remote API meters the +gateway only moves queueing from this process into theirs. + ### Session Queue Concurrency Per-session queues now support configurable `maxConcurrent`: diff --git a/internal/channels/dispatch.go b/internal/channels/dispatch.go index c338ba72..a4df9743 100644 --- a/internal/channels/dispatch.go +++ b/internal/channels/dispatch.go @@ -3,11 +3,14 @@ package channels import ( "context" "fmt" + "hash/fnv" "log/slog" "net/http" "os" "regexp" + "strconv" "strings" + "sync" "github.com/google/uuid" "github.com/nextlevelbuilder/goclaw/internal/bus" @@ -20,85 +23,205 @@ type WebhookRoute struct { Handler http.Handler } -// dispatchOutbound consumes outbound messages from the bus and routes them -// to the appropriate channel. Internal channels are silently skipped. +// Outbound dispatch is sharded by conversation. +// +// A single dispatch goroutine made every reply in the process — every channel, +// every user — wait behind the slowest Send. One media upload of a few seconds +// froze all outbound traffic for that long. The ordering the old loop actually +// protected is per-conversation (a run emits block replies, retry notices and +// the final answer to one ChatID, and those must arrive in that order); it +// never needed a global order. +// +// Sharding on channel+ChatID keeps that guarantee — a conversation always maps +// to the same shard and is processed serially there — while letting unrelated +// conversations proceed in parallel. +const ( + defaultOutboundShards = 8 + + // outboundShardQueue buffers each shard so a brief stall on one Send does + // not immediately push back on the reader (which would re-serialize the + // shards). The bus itself remains the real backpressure boundary. + outboundShardQueue = 256 +) + +// outboundShardCount resolves the dispatch worker count. +// +// GOCLAW_OUTBOUND_SHARDS=8 +// +// Raising it past the point where the *remote* API meters us (DingTalk's +// per-app card QPS, a provider's rate limit) only moves queueing from our +// process into theirs. +func outboundShardCount() int { + if v := os.Getenv("GOCLAW_OUTBOUND_SHARDS"); v != "" { + if n, err := strconv.Atoi(v); err == nil && n > 0 { + return n + } + } + return defaultOutboundShards +} + +// outboundShardIndex maps a conversation to a stable shard. +// +// The channel name is part of the key because ChatIDs are only unique within a +// channel — two channels can legitimately use the same conversation id. +func outboundShardIndex(channel, chatID string, shards int) int { + h := fnv.New32a() + _, _ = h.Write([]byte(channel)) + _, _ = h.Write([]byte{0}) + _, _ = h.Write([]byte(chatID)) + return int(h.Sum32() % uint32(shards)) +} + +// dispatchOutbound consumes outbound messages from the bus and fans them out to +// per-conversation shard workers. Internal channels are silently skipped. func (m *Manager) dispatchOutbound(ctx context.Context) { - slog.Info("outbound dispatcher started") + shards := outboundShardCount() + + queues := make([]chan bus.OutboundMessage, shards) + var wg sync.WaitGroup + for i := range queues { + queues[i] = make(chan bus.OutboundMessage, outboundShardQueue) + wg.Add(1) + go func(id int, q <-chan bus.OutboundMessage) { + defer wg.Done() + m.dispatchShard(ctx, id, q) + }(i, queues[i]) + } + + slog.Info("outbound dispatcher started", "shards", shards) + + defer func() { + // Closing the queues drains whatever each shard already accepted, so a + // reply that made it out of the bus is not dropped on shutdown. + for _, q := range queues { + close(q) + } + wg.Wait() + slog.Info("outbound dispatcher stopped") + }() for { select { case <-ctx.Done(): - slog.Info("outbound dispatcher stopped") return default: - msg, ok := m.bus.SubscribeOutbound(ctx) - if !ok { - continue - } - - // Skip internal channels - if IsInternalChannel(msg.Channel) { - continue - } - - m.mu.RLock() - channel, exists := m.channels[msg.Channel] - m.mu.RUnlock() - - if !exists { - slog.Warn("unknown channel for outbound message", "channel", msg.Channel) - continue - } - - // Filter out temp media files that no longer exist (already sent by another dispatch). - if len(msg.Media) > 0 { - tmpDir := os.TempDir() - filtered := msg.Media[:0] - for _, media := range msg.Media { - if media.URL != "" && strings.HasPrefix(media.URL, tmpDir) { - if _, err := os.Stat(media.URL); err != nil { - slog.Debug("skipping already-delivered temp media", "path", media.URL) - continue - } - } - filtered = append(filtered, media) - } - msg.Media = filtered - // If only media was in this message and all files are gone, skip entirely. - if len(msg.Media) == 0 && msg.Content == "" { - continue - } - } - - // Add tenant context for per-tenant TTS auto-apply - sendCtx := ctx - if msg.TenantID != uuid.Nil { - sendCtx = store.WithTenantID(ctx, msg.TenantID) - } - - // Add agent audio context for per-agent TTS voice override - if msg.AgentID != uuid.Nil && len(msg.AgentOtherConfig) > 0 { - sendCtx = store.WithAgentAudio(sendCtx, store.AgentAudioSnapshot{ - AgentID: msg.AgentID, - OtherConfig: msg.AgentOtherConfig, - }) - } - - if err := channel.Send(sendCtx, msg); err != nil { - m.handleSendFailure(sendCtx, channel, msg, err) - } - - // Clean up temp media files only. Workspace-generated files are preserved - // so they remain accessible via workspace/web UI after delivery. - tmpDir := os.TempDir() - for _, media := range msg.Media { - if media.URL != "" && strings.HasPrefix(media.URL, tmpDir) { - if err := os.Remove(media.URL); err != nil { - slog.Debug("failed to clean up media file", "path", media.URL, "error", err) - } - } - } } + + msg, ok := m.bus.SubscribeOutbound(ctx) + if !ok { + return + } + if IsInternalChannel(msg.Channel) { + continue + } + + idx := outboundShardIndex(msg.Channel, msg.ChatID, shards) + select { + case queues[idx] <- msg: + case <-ctx.Done(): + return + } + } +} + +// dispatchShard delivers one shard's messages, strictly in order. +func (m *Manager) dispatchShard(ctx context.Context, id int, q <-chan bus.OutboundMessage) { + for msg := range q { + m.deliverOutbound(ctx, msg) + } + slog.Debug("outbound shard stopped", "shard", id) +} + +// deliverOutbound sends one message and cleans up its temp media. +func (m *Manager) deliverOutbound(ctx context.Context, msg bus.OutboundMessage) { + m.mu.RLock() + channel, exists := m.channels[msg.Channel] + m.mu.RUnlock() + + if !exists { + slog.Warn("unknown channel for outbound message", "channel", msg.Channel) + return + } + + kept, claimed := m.claimTempMedia(msg.Media) + defer m.releaseTempMedia(claimed) + msg.Media = kept + + // If only media was in this message and every file is gone, skip entirely. + if len(msg.Media) == 0 && msg.Content == "" { + return + } + + // Add tenant context for per-tenant TTS auto-apply + sendCtx := ctx + if msg.TenantID != uuid.Nil { + sendCtx = store.WithTenantID(ctx, msg.TenantID) + } + + // Add agent audio context for per-agent TTS voice override + if msg.AgentID != uuid.Nil && len(msg.AgentOtherConfig) > 0 { + sendCtx = store.WithAgentAudio(sendCtx, store.AgentAudioSnapshot{ + AgentID: msg.AgentID, + OtherConfig: msg.AgentOtherConfig, + }) + } + + if err := channel.Send(sendCtx, msg); err != nil { + m.handleSendFailure(sendCtx, channel, msg, err) + } +} + +// claimTempMedia filters msg.Media down to the files this dispatch owns, and +// returns the temp paths it claimed so they can be released after the send. +// +// Serial dispatch deduped temp media implicitly: the first delivery removed the +// file, and a later message carrying the same path found os.Stat failing and +// skipped it. Across concurrent shards that check alone is a TOCTOU race — two +// shards can both stat a live file and upload it twice. +// +// The claim set closes the window. Entries live only for the duration of a +// send, so the set stays bounded, and a message that arrives after the owner +// finished still finds the file gone and skips it exactly as before. +// +// Non-temp media (workspace files) is passed through untouched: it is never +// removed after delivery, so it needs no claim. +func (m *Manager) claimTempMedia(media []bus.MediaAttachment) (kept []bus.MediaAttachment, claimed []string) { + if len(media) == 0 { + return nil, nil + } + + tmpDir := os.TempDir() + kept = make([]bus.MediaAttachment, 0, len(media)) + + for _, mf := range media { + if mf.URL == "" || !strings.HasPrefix(mf.URL, tmpDir) { + kept = append(kept, mf) + continue + } + if _, err := os.Stat(mf.URL); err != nil { + slog.Debug("skipping already-delivered temp media", "path", mf.URL) + continue + } + if _, loaded := m.mediaClaims.LoadOrStore(mf.URL, struct{}{}); loaded { + slog.Debug("skipping temp media already claimed by another shard", "path", mf.URL) + continue + } + kept = append(kept, mf) + claimed = append(claimed, mf.URL) + } + return kept, claimed +} + +// releaseTempMedia removes delivered temp files and drops their claims. +// +// Workspace-generated files are preserved so they remain accessible via the +// workspace/web UI after delivery; only os.TempDir() entries reach here. +func (m *Manager) releaseTempMedia(claimed []string) { + for _, path := range claimed { + if err := os.Remove(path); err != nil { + slog.Debug("failed to clean up media file", "path", path, "error", err) + } + m.mediaClaims.Delete(path) } } diff --git a/internal/channels/dispatch_shard_test.go b/internal/channels/dispatch_shard_test.go new file mode 100644 index 00000000..3a6f6160 --- /dev/null +++ b/internal/channels/dispatch_shard_test.go @@ -0,0 +1,316 @@ +package channels + +import ( + "context" + "fmt" + "os" + "path/filepath" + "sync" + "testing" + "time" + + "github.com/nextlevelbuilder/goclaw/internal/bus" +) + +// recorderChannel records every delivered message in arrival order, and can +// block inside Send to simulate a slow upload. +type recorderChannel struct { + BaseChannel + + mu sync.Mutex + received []bus.OutboundMessage + + // blockOn holds Send hostage for messages whose ChatID matches, until + // release is closed. Empty means never block. + blockOn string + release chan struct{} +} + +func newRecorderChannel(name string) *recorderChannel { + rc := &recorderChannel{release: make(chan struct{})} + rc.BaseChannel = BaseChannel{name: name} + return rc +} + +func (r *recorderChannel) Type() string { return TypeTelegram } +func (r *recorderChannel) Start(_ context.Context) error { return nil } +func (r *recorderChannel) Stop(_ context.Context) error { return nil } +func (r *recorderChannel) IsRunning() bool { return true } +func (r *recorderChannel) IsAllowed(_ string) bool { return true } + +func (r *recorderChannel) Send(_ context.Context, msg bus.OutboundMessage) error { + if r.blockOn != "" && msg.ChatID == r.blockOn { + <-r.release + } + r.mu.Lock() + r.received = append(r.received, msg) + r.mu.Unlock() + return nil +} + +func (r *recorderChannel) contentsFor(chatID string) []string { + r.mu.Lock() + defer r.mu.Unlock() + var out []string + for _, m := range r.received { + if m.ChatID == chatID { + out = append(out, m.Content) + } + } + return out +} + +func (r *recorderChannel) count() int { + r.mu.Lock() + defer r.mu.Unlock() + return len(r.received) +} + +// waitFor polls cond until it holds or the deadline passes. +func waitFor(t *testing.T, timeout time.Duration, cond func() bool) bool { + t.Helper() + deadline := time.Now().Add(timeout) + for time.Now().Before(deadline) { + if cond() { + return true + } + time.Sleep(2 * time.Millisecond) + } + return cond() +} + +// distinctShardChats returns two chat IDs that hash to different shards, so a +// test can prove two conversations really do run on separate workers. +func distinctShardChats(t *testing.T, channel string, shards int) (string, string) { + t.Helper() + base := "chat-0" + baseIdx := outboundShardIndex(channel, base, shards) + for i := 1; i < 500; i++ { + other := fmt.Sprintf("chat-%d", i) + if outboundShardIndex(channel, other, shards) != baseIdx { + return base, other + } + } + t.Fatalf("no two chat ids landed on different shards out of %d", shards) + return "", "" +} + +// A conversation must always map to the same shard — that mapping is the only +// thing preserving per-chat ordering once dispatch runs in parallel. +func TestOutboundShardIndex_StableAndBounded(t *testing.T) { + t.Parallel() + + const shards = 8 + for i := range 200 { + chatID := fmt.Sprintf("chat-%d", i) + got := outboundShardIndex("telegram-test", chatID, shards) + if got < 0 || got >= shards { + t.Fatalf("shard index %d out of range [0,%d)", got, shards) + } + if again := outboundShardIndex("telegram-test", chatID, shards); again != got { + t.Fatalf("shard index for %q not stable: %d then %d", chatID, got, again) + } + } +} + +// ChatIDs are only unique within a channel, so the channel name must be part +// of the shard key — otherwise two channels sharing a conversation id would +// have their messages interleaved on one worker. +func TestOutboundShardIndex_ChannelIsPartOfKey(t *testing.T) { + t.Parallel() + + const shards = 64 + same := 0 + for i := range 100 { + chatID := fmt.Sprintf("chat-%d", i) + if outboundShardIndex("alpha", chatID, shards) == outboundShardIndex("beta", chatID, shards) { + same++ + } + } + // With 64 shards, identical placement for every id would mean the channel + // name was ignored entirely. + if same == 100 { + t.Fatal("channel name does not affect shard placement") + } +} + +// The ordering guarantee dispatch actually owes callers: a run emits block +// replies, retry notices and the final answer to one ChatID, and they must +// arrive in that order. +func TestDispatchOutbound_PreservesPerChatOrder(t *testing.T) { + t.Parallel() + + mb := bus.New() + mgr := NewManager(mb) + ch := newRecorderChannel("telegram-test") + mgr.channels["telegram-test"] = ch + + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + go mgr.dispatchOutbound(ctx) + + const n = 50 + for i := range n { + mb.PublishOutbound(bus.OutboundMessage{ + Channel: "telegram-test", + ChatID: "chat-order", + Content: fmt.Sprintf("msg-%d", i), + }) + } + + if !waitFor(t, 5*time.Second, func() bool { return ch.count() >= n }) { + t.Fatalf("only %d/%d messages delivered", ch.count(), n) + } + + got := ch.contentsFor("chat-order") + for i := range n { + want := fmt.Sprintf("msg-%d", i) + if got[i] != want { + t.Fatalf("message %d out of order: got %q, want %q", i, got[i], want) + } + } +} + +// The regression this whole change exists for: one stalled conversation must +// not hold up every other conversation in the process. +func TestDispatchOutbound_SlowChatDoesNotBlockOthers(t *testing.T) { + t.Parallel() + + shards := outboundShardCount() + slowChat, fastChat := distinctShardChats(t, "telegram-test", shards) + + mb := bus.New() + mgr := NewManager(mb) + ch := newRecorderChannel("telegram-test") + ch.blockOn = slowChat + mgr.channels["telegram-test"] = ch + + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + go mgr.dispatchOutbound(ctx) + + // The stuck conversation goes first, exactly as it would when a media + // upload lands ahead of everyone else's replies. + mb.PublishOutbound(bus.OutboundMessage{ + Channel: "telegram-test", + ChatID: slowChat, + Content: "slow-upload", + }) + mb.PublishOutbound(bus.OutboundMessage{ + Channel: "telegram-test", + ChatID: fastChat, + Content: "should-not-wait", + }) + + if !waitFor(t, 5*time.Second, func() bool { return len(ch.contentsFor(fastChat)) == 1 }) { + close(ch.release) + t.Fatal("a reply on an unrelated conversation was blocked behind a slow send") + } + + // The slow one still completes once it is unblocked. + close(ch.release) + if !waitFor(t, 5*time.Second, func() bool { return len(ch.contentsFor(slowChat)) == 1 }) { + t.Fatal("the slow conversation never completed after release") + } +} + +// Temp media was deduped by the fact that serial dispatch removed the file +// before the next message was looked at. Concurrent shards make that a TOCTOU +// race, so the claim must be exclusive. +func TestClaimTempMedia_ExclusiveUnderConcurrency(t *testing.T) { + t.Parallel() + + mgr := NewManager(bus.New()) + + path := filepath.Join(t.TempDir(), "shared.jpg") + if err := os.WriteFile(path, []byte("x"), 0o600); err != nil { + t.Fatalf("write temp media: %v", err) + } + // claimTempMedia only guards files under os.TempDir(); t.TempDir() lives + // there, so this mirrors the real path. + if !isUnderTempDir(path) { + t.Skipf("t.TempDir() %q is not under os.TempDir() %q on this platform", path, os.TempDir()) + } + + media := []bus.MediaAttachment{{URL: path, ContentType: "image/jpeg"}} + + const racers = 16 + var wg sync.WaitGroup + winners := make(chan []string, racers) + for range racers { + wg.Add(1) + go func() { + defer wg.Done() + kept, claimed := mgr.claimTempMedia(media) + if len(kept) > 0 { + winners <- claimed + } + }() + } + wg.Wait() + close(winners) + + if n := len(winners); n != 1 { + t.Fatalf("%d dispatchers claimed the same temp file; want exactly 1", n) + } +} + +// Non-temp media (workspace files) is never removed after delivery, so it must +// pass through unclaimed — otherwise a second message referencing the same +// workspace file would silently lose it. +func TestClaimTempMedia_PassesThroughNonTempAndSkipsMissing(t *testing.T) { + t.Parallel() + + mgr := NewManager(bus.New()) + + missing := filepath.Join(os.TempDir(), "goclaw-does-not-exist-12345.bin") + workspace := "/workspace/report.pdf" + + kept, claimed := mgr.claimTempMedia([]bus.MediaAttachment{ + {URL: workspace}, + {URL: missing}, + }) + + if len(kept) != 1 || kept[0].URL != workspace { + t.Fatalf("kept = %+v, want only the workspace file", kept) + } + if len(claimed) != 0 { + t.Fatalf("claimed = %v, want no claims on non-temp media", claimed) + } +} + +// releaseTempMedia must drop the claim as well as the file, so the claim set +// stays bounded across a long-running process. +func TestReleaseTempMedia_ClearsClaim(t *testing.T) { + t.Parallel() + + mgr := NewManager(bus.New()) + + path := filepath.Join(t.TempDir(), "once.jpg") + if err := os.WriteFile(path, []byte("x"), 0o600); err != nil { + t.Fatalf("write temp media: %v", err) + } + if !isUnderTempDir(path) { + t.Skipf("t.TempDir() %q is not under os.TempDir() %q on this platform", path, os.TempDir()) + } + + media := []bus.MediaAttachment{{URL: path}} + + kept, claimed := mgr.claimTempMedia(media) + if len(kept) != 1 || len(claimed) != 1 { + t.Fatalf("first claim failed: kept=%d claimed=%d", len(kept), len(claimed)) + } + mgr.releaseTempMedia(claimed) + + if _, stillHeld := mgr.mediaClaims.Load(path); stillHeld { + t.Fatal("claim survived release; the claim set would grow without bound") + } + if _, err := os.Stat(path); err == nil { + t.Fatal("temp file survived release") + } +} + +func isUnderTempDir(path string) bool { + rel, err := filepath.Rel(os.TempDir(), path) + return err == nil && !filepath.IsAbs(rel) && rel != ".." && len(rel) > 0 && rel[0] != '.' +} diff --git a/internal/channels/manager.go b/internal/channels/manager.go index 19936a35..c571ab7c 100644 --- a/internal/channels/manager.go +++ b/internal/channels/manager.go @@ -76,6 +76,7 @@ type Manager struct { health map[string]ChannelHealth bus *bus.MessageBus runs sync.Map // runID string → *RunContext + mediaClaims sync.Map // temp media path → struct{}, in-flight dispatch claims dispatchTask *asyncTask mu sync.RWMutex contactCollector *store.ContactCollector diff --git a/internal/providers/defaults.go b/internal/providers/defaults.go index de098ca8..b6def05b 100644 --- a/internal/providers/defaults.go +++ b/internal/providers/defaults.go @@ -2,6 +2,8 @@ package providers import ( "net/http" + "os" + "strconv" "time" ) @@ -21,6 +23,32 @@ const ( StdioScanBufMax = 10 * 1024 * 1024 // 10MB max for large protocol messages ) +// Idle connection pooling, overridable for deployments that raise agent-run +// concurrency: +// +// GOCLAW_HTTP_MAX_IDLE_CONNS=100 +// GOCLAW_HTTP_MAX_IDLE_CONNS_PER_HOST=10 +// +// When nearly all traffic goes to one provider host, the per-host limit is the +// one that binds: every concurrent request past it gets a fresh TCP+TLS +// handshake, and its connection is closed rather than pooled on completion. +// A self-hosted or in-network provider has no reason to be throttled this way, +// so those deployments should raise it to match LaneMain. +const ( + defaultMaxIdleConns = 100 + defaultMaxIdleConnsPerHost = 10 +) + +// transportEnv reads a positive int from an env var, falling back to defaultVal. +func transportEnv(key string, defaultVal int) int { + if v := os.Getenv(key); v != "" { + if n, err := strconv.Atoi(v); err == nil && n > 0 { + return n + } + } + return defaultVal +} + // NewDefaultTransport returns an http.Transport with per-stage timeouts but no // overall deadline. The absence of Client.Timeout allows LLM streaming responses // (extended thinking, long completions) to run indefinitely once the response has @@ -57,8 +85,8 @@ func NewDefaultTransport() *http.Transport { IdleConnTimeout: 90 * time.Second, // close idle keep-alive connections TLSHandshakeTimeout: 10 * time.Second, ExpectContinueTimeout: 1 * time.Second, - MaxIdleConns: 100, - MaxIdleConnsPerHost: 10, + MaxIdleConns: transportEnv("GOCLAW_HTTP_MAX_IDLE_CONNS", defaultMaxIdleConns), + MaxIdleConnsPerHost: transportEnv("GOCLAW_HTTP_MAX_IDLE_CONNS_PER_HOST", defaultMaxIdleConnsPerHost), } } diff --git a/internal/scheduler/lanes.go b/internal/scheduler/lanes.go index 60d9ddcd..e04e174b 100644 --- a/internal/scheduler/lanes.go +++ b/internal/scheduler/lanes.go @@ -154,6 +154,11 @@ func NewLaneManager(configs []LaneConfig) *LaneManager { // GOCLAW_LANE_SUBAGENT=50 // GOCLAW_LANE_TEAM=100 // GOCLAW_LANE_CRON=30 +// +// LaneMain bounds concurrent agent runs across every channel, so on a bot +// shared by a whole org it is the ceiling on how many people can be answered +// at once. A run is overwhelmingly blocked on the LLM, not on CPU, so raising +// it trades provider load for wait time, not local work. func DefaultLanes() []LaneConfig { return []LaneConfig{ {Name: LaneMain, Concurrency: laneEnv("GOCLAW_LANE_MAIN", 30)}, diff --git a/internal/store/pg/pool.go b/internal/store/pg/pool.go index b2251f8a..7c7fb404 100644 --- a/internal/store/pg/pool.go +++ b/internal/store/pg/pool.go @@ -4,6 +4,8 @@ import ( "database/sql" "fmt" "log/slog" + "os" + "strconv" "github.com/jackc/pgx/v5" "github.com/jackc/pgx/v5/stdlib" @@ -14,6 +16,34 @@ import ( // the gateway's own connections apart from external clients (issue #1338). const PoolApplicationName = "goclaw" +// Connection pool sizing, overridable for deployments that raise LaneMain: +// +// GOCLAW_PG_MAX_OPEN_CONNS=25 +// GOCLAW_PG_MAX_IDLE_CONNS=10 +// +// A connection is held per query, not per agent run, so the pool does not cap +// concurrency directly — it caps how many queries can be in flight. The cost of +// undersizing shows up as latency at burst, when many runs start together and +// each loads context and history at the same moment, which is why the ceiling +// wants to move with LaneMain rather than stay fixed below it. +// +// Postgres' own max_connections is the real limit: keep the sum across all +// gateway processes under it. +const ( + defaultMaxOpenConns = 25 + defaultMaxIdleConns = 10 +) + +// poolEnv reads a positive int from an env var, falling back to defaultVal. +func poolEnv(key string, defaultVal int) int { + if v := os.Getenv(key); v != "" { + if n, err := strconv.Atoi(v); err == nil && n > 0 { + return n + } + } + return defaultVal +} + // OpenDB creates a database/sql connection to Postgres using pgx driver. func OpenDB(dsn string) (*sql.DB, error) { config, err := pgx.ParseConfig(dsn) @@ -29,14 +59,19 @@ func OpenDB(dsn string) (*sql.DB, error) { } db := stdlib.OpenDB(*config) - db.SetMaxOpenConns(25) - db.SetMaxIdleConns(10) + maxOpen := poolEnv("GOCLAW_PG_MAX_OPEN_CONNS", defaultMaxOpenConns) + maxIdle := poolEnv("GOCLAW_PG_MAX_IDLE_CONNS", defaultMaxIdleConns) + if maxIdle > maxOpen { + maxIdle = maxOpen + } + db.SetMaxOpenConns(maxOpen) + db.SetMaxIdleConns(maxIdle) if err := db.Ping(); err != nil { db.Close() return nil, fmt.Errorf("ping postgres: %w", err) } - slog.Info("postgres connected", "dsn_len", len(dsn)) + slog.Info("postgres connected", "dsn_len", len(dsn), "max_open", maxOpen, "max_idle", maxIdle) return db, nil }