perf(channels): shard outbound dispatch per conversation

docs/00-architecture-overview.md

@@ -380,6 +380,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`:

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
-			}
+		msg, ok := m.bus.SubscribeOutbound(ctx)
+		if !ok {
+			return
+		}
+		if IsInternalChannel(msg.Channel) {
+			continue
+		}
 
-			m.mu.RLock()
-			channel, exists := m.channels[msg.Channel]
-			m.mu.RUnlock()
+		idx := outboundShardIndex(msg.Channel, msg.ChatID, shards)
+		select {
+		case queues[idx] <- msg:
+		case <-ctx.Done():
+			return
+		}
+	}
+}
 
-			if !exists {
-				slog.Warn("unknown channel for outbound message", "channel", msg.Channel)
-				continue
-			}
+// 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)
+}
 
-			// 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
-				}
-			}
+// 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()
 
-			// Add tenant context for per-tenant TTS auto-apply
-			sendCtx := ctx
-			if msg.TenantID != uuid.Nil {
-				sendCtx = store.WithTenantID(ctx, msg.TenantID)
-			}
+	if !exists {
+		slog.Warn("unknown channel for outbound message", "channel", msg.Channel)
+		return
+	}
 
-			// 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,
-				})
-			}
+	kept, claimed := m.claimTempMedia(msg.Media)
+	defer m.releaseTempMedia(claimed)
+	msg.Media = kept
 
-			if err := channel.Send(sendCtx, msg); err != nil {
-				m.handleSendFailure(sendCtx, channel, msg, err)
-			}
+	// If only media was in this message and every file is gone, skip entirely.
+	if len(msg.Media) == 0 && msg.Content == "" {
+		return
+	}
 
-			// 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)
-					}
-				}
-			}
+	// 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)
 	}
 }