diff --git a/CHANGELOG.md b/CHANGELOG.md index 63e47833..b8c771f7 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -87,6 +87,16 @@ APIs change between minors without a deprecation period. ### Fixed +- **A lost store machine stops costing requests most of a minute.** The store's connections try each member for 3 seconds, + not until the kernel gives up, and a connection to a machine that is gone is closed within about 10 seconds (TCP + keepalive, and `TCP_USER_TIMEOUT` on Linux). Patroni replaces a lost primary within 20 seconds (`ttl` 20, `loop_wait` + 5, `retry_timeout` 5). Running `server init --patroni` again applies this to an existing cluster. Measured on three + VMs with the primary's power cut: about 45 seconds of hanging requests before. +- **A new leader waits for the judge instead of refusing at once.** The moment a control plane takes the lead it knows no + judge yet. When every judge fails at once, they are asked again every 250 ms while the policy's `latency_budget` lasts, + so `fail: closed` no longer answers every request in those seconds with 503 (about 17 seconds of them in the test). A + judge that is slow is not asked again, and an alarm is raised once per request, not once per try. + - Node discovery for Prometheus listed only the control plane asked and the store, so a machine that is both was a control plane to one and a store member to the other — scraped twice — and a standby asked alone lost every diff --git a/go.mod b/go.mod index e763318f..13d3ebc5 100644 --- a/go.mod +++ b/go.mod @@ -35,6 +35,7 @@ require ( golang.org/x/net v0.59.0 golang.org/x/oauth2 v0.37.0 golang.org/x/sync v0.23.0 + golang.org/x/sys v0.48.0 golang.org/x/term v0.46.0 golang.org/x/text v0.42.0 google.golang.org/protobuf v1.36.12 @@ -88,7 +89,6 @@ require ( go.opentelemetry.io/otel/exporters/otlp/otlptrace v1.46.0 // indirect go.opentelemetry.io/otel/metric v1.46.0 // indirect go.opentelemetry.io/proto/otlp v1.11.0 // indirect - golang.org/x/sys v0.48.0 // indirect google.golang.org/genproto/googleapis/api v0.0.0-20260819154853-08b0e4226688 // indirect google.golang.org/genproto/googleapis/rpc v0.0.0-20260819154853-08b0e4226688 // indirect google.golang.org/grpc v1.83.1 // indirect diff --git a/internal/guard/judge.go b/internal/guard/judge.go index 3bd9abc4..5303abed 100644 --- a/internal/guard/judge.go +++ b/internal/guard/judge.go @@ -198,25 +198,70 @@ func (j *judge) ask(ctx context.Context, jp *JudgePolicy, user, answer string) ( } j.mu.Unlock() - err := errNoJudge - for _, model := range models { - var res judgment - res, err = j.askOne(ctx, jp, model, user, answer) - if err == nil { - j.remember(key, res) - return res, nil + start := time.Now() + for { + err := errNoJudge + var failed []string + var errs []error + // fast is every judge failing at once, as one does that is not + // there yet: the moment a control plane takes the lead, before + // the agents have told it where the judges are. + fast := true + for _, model := range models { + var res judgment + t := time.Now() + res, err = j.askOne(ctx, jp, model, user, answer) + if err == nil { + // The judges before it that failed are said to have, as + // one on a GPU that is down while the CPUs' answers. + j.report(failed, errs) + j.remember(key, res) + return res, nil + } + failed, errs = append(failed, model), append(errs, err) + if time.Since(t) > judgeFastFailure || errors.Is(err, errJudgeFormat) { + fast = false + } + if ctx.Err() != nil { + // The request itself is gone: no judge after this one is wanted. + break + } } - if j.failed != nil { - j.failed(model, err) - } - if ctx.Err() != nil { - // The request itself is gone: no judge after this one is wanted. - break + // Asked again while the budget lasts, rather than refused at once: + // with fail: closed, every request in those seconds was a 503. + if fast && ctx.Err() == nil && time.Since(start)+judgeRetryEvery < jp.budget() { + select { + case <-ctx.Done(): + case <-time.After(judgeRetryEvery): + } + if ctx.Err() == nil { + continue + } } + // Told once, for the last round: not twenty times in five seconds. + j.report(failed, errs) + return judgment{}, err } - return judgment{}, err } +// report tells of each judge that gave no verdict. +func (j *judge) report(models []string, errs []error) { + if j.failed == nil { + return + } + for i, model := range models { + j.failed(model, errs[i]) + } +} + +// A judge that fails within judgeFastFailure is one that is not there, and +// is asked again every judgeRetryEvery while the policy's budget lasts. One +// that is slow, or answers what cannot be read, is not. +const ( + judgeFastFailure = time.Second + judgeRetryEvery = 250 * time.Millisecond +) + // askOne asks one judge, waiting the policy's budget for it. func (j *judge) askOne(ctx context.Context, jp *JudgePolicy, model, user, answer string) (res judgment, err error) { start := time.Now() diff --git a/internal/guard/judge_test.go b/internal/guard/judge_test.go index 308af0e4..2e9c3600 100644 --- a/internal/guard/judge_test.go +++ b/internal/guard/judge_test.go @@ -222,7 +222,7 @@ func TestJudgesAreAskedInTurn(t *testing.T) { var asked []string gpuDown := errors.New("no replica of gpu-judge is serving") g := gate(t, lib(map[string]Policy{"p": {Fail: "closed", - Judge: &JudgePolicy{Models: []string{"gpu-judge", "cpu-judge"}, Input: "block"}}}, "p")) + Judge: &JudgePolicy{Models: []string{"gpu-judge", "cpu-judge"}, Input: "block", LatencyBudget: 600 * time.Millisecond}}}, "p")) g.SetJudge(func(_ context.Context, model, _ string, body []byte) ([]byte, error) { asked = append(asked, model) if model == "gpu-judge" { @@ -317,3 +317,44 @@ func TestALongTextIsJudgedByItsEnds(t *testing.T) { t.Errorf("a short text was cut: %q", got) } } + +// The moment a control plane takes the lead it knows no judge yet; one that +// arrives within the budget judges the request, instead of fail: closed +// refusing it at once. One that is slow is not asked again. +func TestAJudgeNotThereYetIsWaitedForWithinTheBudget(t *testing.T) { + ctx := context.Background() + g := gate(t, lib(map[string]Policy{"p": {Fail: "closed", + Judge: &JudgePolicy{Model: "guard", Input: "block", LatencyBudget: 2 * time.Second}}}, "p")) + arrives := time.Now().Add(400 * time.Millisecond) + calls := 0 + g.SetJudge(func(context.Context, string, string, []byte) ([]byte, error) { + calls++ + if time.Now().Before(arrives) { + return nil, errors.New("no replica of guard is serving") + } + return json.Marshal(map[string]any{"choices": []any{map[string]any{"text": "Safety: Safe\nCategories: None"}}}) + }, func(string) (string, []string, bool) { return "qwen3guard", nil, true }) + var alarmed []string + g.SetJudgeAlarm(func(model string, _ error) { alarmed = append(alarmed, model) }) + if v, err := g.Input(ctx, Subject{}, chat("user", "hello")); err != nil || v.Action != Allow { + t.Fatalf("refused while the judge was on its way: %+v %v", v, err) + } + if calls < 2 || len(alarmed) != 0 { + t.Errorf("asked %d times, alarms %v", calls, alarmed) + } + + // Slow, and failing: given up on after one try, not the whole budget again. + calls = 0 + g.SetJudge(func(ctx context.Context, _, _ string, _ []byte) ([]byte, error) { + calls++ + time.Sleep(1100 * time.Millisecond) + return nil, errors.New("overloaded") + }, func(string) (string, []string, bool) { return "qwen3guard", nil, true }) + start := time.Now() + if _, err := g.Input(ctx, Subject{}, chat("user", "hello again")); err == nil { + t.Error("fail closed served what no judge read") + } + if calls != 1 || time.Since(start) > 1800*time.Millisecond { + t.Errorf("a slow judge was asked %d times over %s", calls, time.Since(start)) + } +} diff --git a/internal/pgha/config.go b/internal/pgha/config.go index eb3820fb..2e029e3d 100644 --- a/internal/pgha/config.go +++ b/internal/pgha/config.go @@ -169,9 +169,9 @@ etcd3: %[14]s bootstrap: dcs: - ttl: 30 - loop_wait: 10 - retry_timeout: 10 + ttl: %[15]d + loop_wait: %[16]d + retry_timeout: %[17]d maximum_lag_on_failover: 1048576 postgresql: use_pg_rewind: true @@ -211,7 +211,7 @@ postgresql: parameters: unix_socket_directories: /var/run/postgresql `, Scope, m.Name, m.IP, PatroniPort, strings.Join(hosts, ","), version, s.Superuser, s.Replicator, hbaLines, - s.RestAPI, RestUser, sslLines, restTLSLines(tls), etcdTLSLines(tls)) + s.RestAPI, RestUser, sslLines, restTLSLines(tls), etcdTLSLines(tls), leaderTTL, loopWait, retryTimeout) b.WriteString(` tags: nofailover: false diff --git a/internal/pgha/pgha_test.go b/internal/pgha/pgha_test.go index e63cfef5..e920089e 100644 --- a/internal/pgha/pgha_test.go +++ b/internal/pgha/pgha_test.go @@ -559,3 +559,42 @@ func TestAMemberLeftPendingARestartIsRestartedOnTheNextRun(t *testing.T) { t.Errorf("restarted %v; only the member pending", restarted) } } + +func TestALostPrimaryIsReplacedWithinTwentySeconds(t *testing.T) { + cfg := map[string]any{"ttl": 30, "loop_wait": 10, "retry_timeout": 10} + patches := 0 + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + switch r.Method { + case http.MethodGet: + _ = json.NewEncoder(w).Encode(cfg) + case http.MethodPatch: + patches++ + _ = json.NewDecoder(r.Body).Decode(&cfg) + } + })) + defer srv.Close() + u, _ := url.Parse(srv.URL) + old := restPort + restPort, _ = strconv.Atoi(u.Port()) + defer func() { restPort = old }() + ca, err := tlsconf.NewCA("test") + if err != nil { + t.Fatal(err) + } + s := Setup{DB: []Node{{Member{"m", "127.0.0.1"}, &machine{}}}, CA: ca} + for range 2 { + if err := s.timings(context.Background(), Secrets{RestAPI: "pw"}); err != nil { + t.Fatal(err) + } + } + if patches != 1 || cfg["ttl"].(float64) != 20 || cfg["loop_wait"].(float64) != 5 || cfg["retry_timeout"].(float64) != 5 { + t.Errorf("patched %d times to %v", patches, cfg) + } + // Patroni's own rule, or it refuses them. + if loopWait+2*retryTimeout > leaderTTL { + t.Error("loop_wait + 2*retry_timeout is more than ttl") + } + if c := Patroni(pg01, []Member{pg01, ctl02}, []Member{pg01, ctl02, ctl}, nil, "16", Secrets{}, false); !strings.Contains(c, "ttl: 20\n loop_wait: 5\n retry_timeout: 5\n") { + t.Errorf("a new cluster bootstraps with others:\n%s", c) + } +} diff --git a/internal/pgha/setup.go b/internal/pgha/setup.go index 1407db67..8774146f 100644 --- a/internal/pgha/setup.go +++ b/internal/pgha/setup.go @@ -131,6 +131,9 @@ func (s *Setup) Run(ctx context.Context) (string, error) { if err := s.memory(ctx, sec); err != nil { return "", fmt.Errorf("memory: %w", err) } + if err := s.timings(ctx, sec); err != nil { + return "", fmt.Errorf("failover: %w", err) + } if err := s.database(ctx); err != nil { return "", fmt.Errorf("database: %w", err) } diff --git a/internal/pgha/timings.go b/internal/pgha/timings.go new file mode 100644 index 00000000..71f6818f --- /dev/null +++ b/internal/pgha/timings.go @@ -0,0 +1,60 @@ +package pgha + +import ( + "context" + "fmt" +) + +// How soon a lost primary is replaced. Patroni's own defaults — the leader +// key lives 30 seconds, checked every 10 — had the control planes wait the +// better part of a minute for a store after a database machine lost its +// power. Patroni asks that loop_wait + 2*retry_timeout <= ttl, so a lost +// primary is replaced within ttl and a slow etcd round is not mistaken for +// one. +const ( + leaderTTL = 20 + loopWait = 5 + retryTimeout = 5 +) + +// timingSettings are the shared configuration's keys for them. +func timingSettings() map[string]any { + return map[string]any{"ttl": leaderTTL, "loop_wait": loopWait, "retry_timeout": retryTimeout} +} + +// timings puts them in the cluster's shared configuration, through +// Patroni's API: a cluster set up with the old ones gets these in place. +func (s *Setup) timings(ctx context.Context, sec Secrets) error { + // As with TLS, changed in place by the machine with the authority: init + // has it, and a new cluster has these from its bootstrap already. + if s.CA == nil { + return nil + } + var lastErr error + for _, d := range s.DB { + cfg, err := getConfig(ctx, d.IP) + if err != nil { + lastErr = err + continue + } + want := timingSettings() + same := true + for k, v := range want { + // JSON numbers come back as float64. + if n, ok := cfg[k].(float64); !ok || int(n) != v.(int) { + same = false + } + } + if same { + s.emit("failover", false, "a lost primary is replaced within %d seconds", leaderTTL) + return nil + } + if err := patchConfig(ctx, d.IP, sec.RestAPI, want); err != nil { + return fmt.Errorf("Patroni on %s: %w", d.IP, err) + } + s.emit("failover", true, "a lost primary is replaced within %d seconds (ttl %d, loop_wait %d, retry_timeout %d)", + leaderTTL, leaderTTL, loopWait, retryTimeout) + return nil + } + return fmt.Errorf("no member's Patroni answered: %w", lastErr) +} diff --git a/internal/store/postgres.go b/internal/store/postgres.go index a17779d6..ddd2c1b6 100644 --- a/internal/store/postgres.go +++ b/internal/store/postgres.go @@ -2,17 +2,46 @@ package store import ( "context" - "database/sql" "fmt" + "net" "time" // The pure-Go PostgreSQL driver. Pure-Go is not a preference here: the // whole binary is built with CGO_ENABLED=0 so that it runs on a node whose // libc is older than the builder's, and a cgo driver would take that away // from every deployment, including the ones still on SQLite. - _ "github.com/jackc/pgx/v5/stdlib" + "github.com/jackc/pgx/v5" + "github.com/jackc/pgx/v5/stdlib" ) +// How soon a store machine that is gone is given up on. A replicated store's +// address names every member, and a new connection tries them in order: with +// no timeout, one to a machine that had lost its power waited for the +// kernel's own — half a minute and more — before the next member was asked, +// and every request whose key was not cached waited with it. +const ( + // storeConnectTimeout is per member: the next is tried after it. + storeConnectTimeout = 3 * time.Second + // storeDeadAfter is how long a connection's sent data may go unanswered, + // or an idle one's probes, before it is closed: a connection to the + // primary that was lost fails, and the next opens to the new one. + storeDeadAfter = 10 * time.Second +) + +// failFast gives a connection configuration those limits, where its address +// gave none of its own. +func failFast(cfg *pgx.ConnConfig) { + if cfg.ConnectTimeout == 0 { + cfg.ConnectTimeout = storeConnectTimeout + } + d := &net.Dialer{ + KeepAliveConfig: net.KeepAliveConfig{Enable: true, Idle: 5 * time.Second, Interval: time.Second, + Count: int((storeDeadAfter - 5*time.Second) / time.Second)}, + Control: userTimeout(storeDeadAfter), + } + cfg.DialFunc = d.DialContext +} + // OpenPostgres opens the store against PostgreSQL. // // The same queries run here as on SQLite; only the placeholders differ, and @@ -29,12 +58,14 @@ func OpenPostgres(ctx context.Context, dsn string) (*DB, error) { if dsn == "" { return nil, fmt.Errorf("store: no PostgreSQL address configured") } - db, err := sql.Open("pgx", dsn) + cfg, err := pgx.ParseConfig(dsn) if err != nil { // Deliberately not wrapped with the DSN: it holds a password, and an // error string ends up in logs, tickets and screenshots. return nil, fmt.Errorf("opening the PostgreSQL store: %w", err) } + failFast(cfg) + db := stdlib.OpenDB(*cfg) // Sized for a control plane rather than for a fleet of application // servers. The request path barely touches the store — usage is batched, diff --git a/internal/store/postgres_failfast_test.go b/internal/store/postgres_failfast_test.go new file mode 100644 index 00000000..a060f95a --- /dev/null +++ b/internal/store/postgres_failfast_test.go @@ -0,0 +1,25 @@ +package store + +import ( + "testing" + "time" + + "github.com/jackc/pgx/v5" +) + +func TestAStoreMachineThatIsGoneIsGivenUpOnInSeconds(t *testing.T) { + cfg, err := pgx.ParseConfig("postgres://u:p@10.0.0.158:5432,10.0.0.159:5432/llmgw?sslmode=disable&target_session_attrs=read-write") + if err != nil { + t.Fatal(err) + } + failFast(cfg) + if cfg.ConnectTimeout != storeConnectTimeout || cfg.DialFunc == nil { + t.Errorf("timeout %s, dialer set %v", cfg.ConnectTimeout, cfg.DialFunc != nil) + } + // One the address asks for is its own. + cfg, _ = pgx.ParseConfig("postgres://u:p@h:5432/llmgw?sslmode=disable&connect_timeout=12") + failFast(cfg) + if cfg.ConnectTimeout != 12*time.Second { + t.Errorf("the address's own timeout was replaced: %s", cfg.ConnectTimeout) + } +} diff --git a/internal/store/usertimeout_linux.go b/internal/store/usertimeout_linux.go new file mode 100644 index 00000000..baf5a24c --- /dev/null +++ b/internal/store/usertimeout_linux.go @@ -0,0 +1,23 @@ +package store + +import ( + "syscall" + "time" + + "golang.org/x/sys/unix" +) + +// userTimeout sets TCP_USER_TIMEOUT: data sent and not acknowledged for that +// long closes the connection, where keepalive probes wait for it to go idle. +func userTimeout(d time.Duration) func(network, address string, c syscall.RawConn) error { + return func(_, _ string, c syscall.RawConn) error { + var serr error + err := c.Control(func(fd uintptr) { + serr = unix.SetsockoptInt(int(fd), unix.IPPROTO_TCP, unix.TCP_USER_TIMEOUT, int(d.Milliseconds())) + }) + if err != nil { + return err + } + return serr + } +} diff --git a/internal/store/usertimeout_other.go b/internal/store/usertimeout_other.go new file mode 100644 index 00000000..62c9f76d --- /dev/null +++ b/internal/store/usertimeout_other.go @@ -0,0 +1,13 @@ +//go:build !linux + +package store + +import ( + "syscall" + "time" +) + +// userTimeout is Linux's alone; elsewhere the keepalive probes do it, later. +func userTimeout(time.Duration) func(network, address string, c syscall.RawConn) error { + return nil +}