mirror of
https://github.com/tiennm99/goclaw.git
synced 2026-10-12 18:16:09 +00:00
Conner Mo
9996a16f83
A single dispatch goroutine delivered every reply in the process — every channel, every user — each Send blocking the next. One media upload of a few seconds froze all outbound traffic for that long, and sustained throughput was capped at one Send round-trip regardless of lane capacity. The ordering that loop actually protected is per-conversation: a run emits block replies, retry notices and a final answer to one ChatID, and those must arrive in that order. A global order was never required. Dispatch now shards on channel+ChatID. A conversation always maps to the same shard and is delivered serially there, so the ordering guarantee is unchanged, while unrelated conversations proceed in parallel. GOCLAW_OUTBOUND_SHARDS tunes the worker count (default 8). Temp media needed a real fix to go with it. Serial dispatch deduped it 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, so claims are now taken atomically and released after the send. Also makes the pool limits that bound this path configurable, defaults unchanged: GOCLAW_PG_MAX_OPEN_CONNS / GOCLAW_PG_MAX_IDLE_CONNS, and GOCLAW_HTTP_MAX_IDLE_CONNS / GOCLAW_HTTP_MAX_IDLE_CONNS_PER_HOST. The per-host limit binds when one provider takes nearly all traffic: every request past it pays a fresh TCP+TLS handshake, which a deployment raising LaneMain needs to be able to lift.
78 lines
2.3 KiB
Go
78 lines
2.3 KiB
Go
package pg
|
|
|
|
import (
|
|
"database/sql"
|
|
"fmt"
|
|
"log/slog"
|
|
"os"
|
|
"strconv"
|
|
|
|
"github.com/jackc/pgx/v5"
|
|
"github.com/jackc/pgx/v5/stdlib"
|
|
)
|
|
|
|
// PoolApplicationName tags every gateway pool connection via the PostgreSQL
|
|
// application_name runtime parameter. Pre-restore safety checks use it to tell
|
|
// 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)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("parse postgres dsn: %w", err)
|
|
}
|
|
if config.RuntimeParams == nil {
|
|
config.RuntimeParams = map[string]string{}
|
|
}
|
|
// Respect a caller-supplied application_name; otherwise tag as the gateway.
|
|
if config.RuntimeParams["application_name"] == "" {
|
|
config.RuntimeParams["application_name"] = PoolApplicationName
|
|
}
|
|
db := stdlib.OpenDB(*config)
|
|
|
|
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), "max_open", maxOpen, "max_idle", maxIdle)
|
|
return db, nil
|
|
}
|