fix(distributed): survive a slow control-plane database

.agents/api-endpoints-and-auth.md

@@ -236,6 +236,58 @@ Use these HTTP status codes:
 
 If your endpoint should be tracked for usage (token counts, request counts), add the `usageMiddleware` to its middleware chain. See `core/http/middleware/usage.go` and how it's applied in `routes/openai.go`.
 
+## Control-plane database health metrics
+
+In distributed mode the frontend registers three OpenTelemetry gauges over the
+PostgreSQL control-plane database (`core/services/monitoring/control_plane_db.go`,
+wired in `core/application/distributed.go`). They reach `/metrics` through the
+same Prometheus exporter as the rest of the API metrics.
+
+| Metric | Meaning | Page when |
+|--------|---------|-----------|
+| `localai_control_plane_oldest_xmin_age` | Transactions elapsed since the oldest snapshot any backend still holds | above a few million, and rising |
+| `localai_control_plane_longest_transaction_seconds` | Age of the longest open transaction | above 3600 |
+| `localai_control_plane_dead_tuple_ratio` | Dead tuples per live tuple, labelled by `table`, on `backend_nodes`, `node_models` and `gallery_operations` | sustained above ~10 on a small table |
+
+A sustained high `localai_control_plane_oldest_xmin_age` is the one to page on.
+While it grows, autovacuum can reclaim nothing anywhere in the database no
+matter how often it runs, so the dead tuple ratio keeps climbing and a six-row
+registry table can reach hundreds of megabytes. Tuning autovacuum does not help.
+The fix is to find the transaction holding the horizon open and clear it:
+
+```sql
+SELECT pid, state, age(backend_xmin) AS xmin_age, now() - xact_start AS xact_age, query
+FROM pg_stat_activity
+WHERE backend_xmin IS NOT NULL
+ORDER BY age(backend_xmin) DESC;
+```
+
+Then `pg_terminate_backend(pid)` on the offenders, and `VACUUM (VERBOSE)` the
+bloated tables once the horizon has moved.
+
+**A healthy-looking xmin age does not on its own prove the horizon is free.**
+The gauge reads `pg_stat_activity`, which only sees live backends. Two other
+things pin the very same horizon and are invisible there, so either one can hold
+vacuum back while the gauge reads 0:
+
+```sql
+SELECT gid, prepared, database, transaction FROM pg_prepared_xacts;
+SELECT slot_name, active, xmin, catalog_xmin FROM pg_replication_slots;
+```
+
+An orphaned prepared transaction is cleared with `ROLLBACK PREPARED '<gid>'`,
+and a stale slot with `pg_drop_replication_slot('<slot_name>')`. Check both
+before concluding that a bloated table has some other cause.
+
+Sampling is scrape-driven behind a 30 second cache, so scrape frequency does not
+translate into database load. Failed and timed-out samples cost the same interval
+as successful ones, so a database that is already struggling is not retried on
+every scrape. A failed sample reports the last good values rather than failing the
+scrape, because these gauges matter most when the database is struggling. Before
+the first successful sample the gauges are absent rather than zero, since a zero
+xmin age would read as a healthy horizon: alert on `absent()` too if you need to
+distinguish "healthy" from "never sampled".
+
 ## Advertising surfaces — where to register a new capability
 
 Beyond routing and auth, LocalAI publishes its capability surface in **four independent places**. When you add an endpoint — especially one introducing a net-new capability like a new media type or a new auth-gated feature — you must update every relevant surface. These aren't optional: missing them means the endpoint works but is invisible to clients, admins, and the UI.

core/application/distributed.go

@@ -15,6 +15,7 @@ import (
 	"github.com/mudler/LocalAI/core/services/distributed"
 	"github.com/mudler/LocalAI/core/services/jobs"
 	"github.com/mudler/LocalAI/core/services/messaging"
+	"github.com/mudler/LocalAI/core/services/monitoring"
 	"github.com/mudler/LocalAI/core/services/nodes"
 	"github.com/mudler/LocalAI/core/services/nodes/prefixcache"
 	"github.com/mudler/LocalAI/core/services/storage"
@@ -162,6 +163,17 @@ func initDistributed(cfg *config.ApplicationConfig, authDB *gorm.DB, configLoade
 	}
 	xlog.Info("Node registry initialized")
 
+	// Bound durable heartbeat writes: a beat that only carries a fresher
+	// timestamp is what turned backend_nodes into a 460 MB six-row table.
+	registry.SetHeartbeatCheckpoint(cfg.Distributed.NodeHeartbeatCheckpointOrDefault())
+
+	// Measure the vacuum horizon. The 42 days it stayed open went unnoticed
+	// because no gauge reported it until models started failing to load.
+	if err := monitoring.RegisterControlPlaneDBMetrics(authDB, 30*time.Second); err != nil {
+		// Metrics are diagnostic; a failure here must not stop the frontend.
+		xlog.Warn("Control-plane database metrics unavailable", "error", err)
+	}
+
 	// Let scheduling rules be keyed by a model alias. The registry resolves a
 	// rule's name through the config loader to find the model it governs, so an
 	// operator can pin placement to a stable name like "production" and have it

core/cli/run.go

@@ -182,6 +182,8 @@ type RunCMD struct {
 	BackendUpgradeTimeout        string `env:"LOCALAI_NATS_BACKEND_UPGRADE_TIMEOUT" help:"NATS round-trip timeout for backend.upgrade requests (default 15m)." group:"distributed"`
 	ModelLoadTimeout             string `env:"LOCALAI_NATS_MODEL_LOAD_TIMEOUT" help:"Fixed gRPC deadline for the remote LoadModel call sent to a worker node once its backend is installed and model files are staged. Unset (the default), the deadline is derived from the checkpoint size instead: 5m plus 20s per GiB, capped at 6h, so multi-tens-of-GB diffusion/video checkpoints get the minutes they need without a fixed cliff. Set this only to pin a specific budget; the value is used verbatim, including when it is shorter than the derived one." group:"distributed"`
 	ModelLoadWait                string `env:"LOCALAI_MODEL_LOAD_WAIT" help:"How long an inference request waits for a model that is still cold-loading onto a worker before it is answered with 503, a Retry-After header and live staging progress (default 60s). The request is served the moment the model becomes ready, so a model already most of the way staged needs no client retry. Set to 0 to wait as long as the load takes — only safe when no ingress or load balancer with an idle timeout sits in front." group:"distributed"`
+	StaleNodeThreshold           string `env:"LOCALAI_STALE_NODE_THRESHOLD" help:"How long a worker node may go without a durable heartbeat before the health monitor marks it offline (default 5m). Because a beat that only carries a fresher timestamp is held back by --node-heartbeat-checkpoint, this must stay comfortably wider than that interval; raise both together. Dead-node detection through the per-model gRPC health check and through request-time failure is unaffected by this knob." group:"distributed"`
+	NodeHeartbeatCheckpoint      string `env:"LOCALAI_NODE_HEARTBEAT_CHECKPOINT" help:"Minimum gap between durable heartbeat writes for a worker node (default 60s). A beat that only carries a fresher timestamp is dropped until this interval elapses; every field is compared against the value last written, so a node's first beat, a changed total VRAM/total disk/GPU vendor, and a free VRAM/RAM/disk reading that has moved more than 256 MiB from the written value all still write immediately, and a node that is not active is never suppressed. Set below the worker heartbeat interval to write on every beat." group:"distributed"`
 	NatsAccountSeed              string `env:"LOCALAI_NATS_ACCOUNT_SEED" help:"NATS account signing seed (SU...) used to mint per-node worker JWTs at registration" group:"distributed"`
 	NatsServiceJWT               string `env:"LOCALAI_NATS_SERVICE_JWT" help:"NATS user JWT for the frontend (and agent workers) to publish control-plane messages" group:"distributed"`
 	NatsServiceSeed              string `env:"LOCALAI_NATS_SERVICE_SEED" help:"NATS user signing seed (SU...) paired with LOCALAI_NATS_SERVICE_JWT" group:"distributed"`
@@ -397,6 +399,20 @@ func (r *RunCMD) Run(ctx *cliContext.Context) error {
 		}
 		opts = append(opts, config.WithModelLoadWait(d))
 	}
+	if r.StaleNodeThreshold != "" {
+		d, err := parseDistributedDuration("LOCALAI_STALE_NODE_THRESHOLD", r.StaleNodeThreshold)
+		if err != nil {
+			return err
+		}
+		opts = append(opts, config.WithStaleNodeThreshold(d))
+	}
+	if r.NodeHeartbeatCheckpoint != "" {
+		d, err := parseDistributedDuration("LOCALAI_NODE_HEARTBEAT_CHECKPOINT", r.NodeHeartbeatCheckpoint)
+		if err != nil {
+			return err
+		}
+		opts = append(opts, config.WithNodeHeartbeatCheckpoint(d))
+	}
 	if r.RegistrationToken != "" {
 		opts = append(opts, config.WithRegistrationToken(r.RegistrationToken))
 	}

core/config/distributed_config.go

@@ -57,12 +57,13 @@ type DistributedConfig struct {
 	StorageSecretKey string // --storage-secret-key / LOCALAI_STORAGE_SECRET_KEY
 
 	// Timeout configuration (all have sensible defaults — zero means use default)
-	MCPToolTimeout      time.Duration // MCP tool execution timeout (default 360s)
-	MCPDiscoveryTimeout time.Duration // MCP discovery timeout (default 60s)
-	WorkerWaitTimeout   time.Duration // Max wait for healthy worker at startup (default 5m)
-	DrainTimeout        time.Duration // Time to wait for in-flight requests during drain (default 30s)
-	HealthCheckInterval time.Duration // Health monitor check interval (default 15s)
-	StaleNodeThreshold  time.Duration // Time before a node is considered stale (default 60s)
+	MCPToolTimeout          time.Duration // MCP tool execution timeout (default 360s)
+	MCPDiscoveryTimeout     time.Duration // MCP discovery timeout (default 60s)
+	WorkerWaitTimeout       time.Duration // Max wait for healthy worker at startup (default 5m)
+	DrainTimeout            time.Duration // Time to wait for in-flight requests during drain (default 30s)
+	HealthCheckInterval     time.Duration // Health monitor check interval (default 15s)
+	StaleNodeThreshold      time.Duration // Time before a node is considered stale (default 5m)
+	NodeHeartbeatCheckpoint time.Duration // Minimum gap between durable heartbeat writes (default 60s, 0 = every beat)
 	// DisablePerModelHealthCheck turns off the health monitor's per-model
 	// gRPC probe. When enabled (the default), the monitor pings each model's
 	// gRPC address and removes stale node_models rows whose backend has
@@ -165,16 +166,17 @@ func (c DistributedConfig) Validate() error {
 	c.NatsAuthConfig().WarnIfInsecure(true)
 	// Check for negative durations
 	for name, d := range map[string]time.Duration{
-		FlagMCPToolTimeout:        c.MCPToolTimeout,
-		FlagMCPDiscoveryTimeout:   c.MCPDiscoveryTimeout,
-		FlagWorkerWaitTimeout:     c.WorkerWaitTimeout,
-		FlagDrainTimeout:          c.DrainTimeout,
-		FlagHealthCheckInterval:   c.HealthCheckInterval,
-		FlagStaleNodeThreshold:    c.StaleNodeThreshold,
-		FlagMCPCIJobTimeout:       c.MCPCIJobTimeout,
-		FlagBackendInstallTimeout: c.BackendInstallTimeout,
-		FlagBackendUpgradeTimeout: c.BackendUpgradeTimeout,
-		FlagModelLoadTimeout:      c.ModelLoadTimeout,
+		FlagMCPToolTimeout:          c.MCPToolTimeout,
+		FlagMCPDiscoveryTimeout:     c.MCPDiscoveryTimeout,
+		FlagWorkerWaitTimeout:       c.WorkerWaitTimeout,
+		FlagDrainTimeout:            c.DrainTimeout,
+		FlagHealthCheckInterval:     c.HealthCheckInterval,
+		FlagStaleNodeThreshold:      c.StaleNodeThreshold,
+		FlagNodeHeartbeatCheckpoint: c.NodeHeartbeatCheckpoint,
+		FlagMCPCIJobTimeout:         c.MCPCIJobTimeout,
+		FlagBackendInstallTimeout:   c.BackendInstallTimeout,
+		FlagBackendUpgradeTimeout:   c.BackendUpgradeTimeout,
+		FlagModelLoadTimeout:        c.ModelLoadTimeout,
 	} {
 		if d < 0 {
 			return fmt.Errorf("%s must not be negative", name)
@@ -337,6 +339,27 @@ func WithModelLoadWait(d time.Duration) AppOption {
 	}
 }
 
+// WithStaleNodeThreshold sets how long a node may go without a durable
+// heartbeat before the health monitor marks it offline. It has to be raised
+// alongside WithNodeHeartbeatCheckpoint: a checkpoint interval wider than this
+// threshold makes every healthy node look dead the moment its beats start
+// being suppressed.
+func WithStaleNodeThreshold(d time.Duration) AppOption {
+	return func(o *ApplicationConfig) {
+		o.Distributed.StaleNodeThreshold = d
+	}
+}
+
+// WithNodeHeartbeatCheckpoint bounds durable heartbeat writes. A zero d is
+// deliberately not special-cased into "unbounded": NodeHeartbeatCheckpointOrDefault
+// reads zero as unset, and an operator who wants a write per beat sets a value
+// below the worker's heartbeat interval instead.
+func WithNodeHeartbeatCheckpoint(d time.Duration) AppOption {
+	return func(o *ApplicationConfig) {
+		o.Distributed.NodeHeartbeatCheckpoint = d
+	}
+}
+
 var EnableAutoApproveNodes = func(o *ApplicationConfig) {
 	o.Distributed.AutoApproveNodes = true
 }
@@ -391,17 +414,18 @@ func WithModelSchedulingConfigPath(path string) AppOption {
 // them as constants prevents the string from drifting from the actual
 // flag a future rename would produce.
 const (
-	FlagMCPToolTimeout        = "mcp-tool-timeout"
-	FlagMCPDiscoveryTimeout   = "mcp-discovery-timeout"
-	FlagWorkerWaitTimeout     = "worker-wait-timeout"
-	FlagDrainTimeout          = "drain-timeout"
-	FlagHealthCheckInterval   = "health-check-interval"
-	FlagStaleNodeThreshold    = "stale-node-threshold"
-	FlagMCPCIJobTimeout       = "mcp-ci-job-timeout"
-	FlagBackendInstallTimeout = "backend-install-timeout"
-	FlagBackendUpgradeTimeout = "backend-upgrade-timeout"
-	FlagModelLoadTimeout      = "model-load-timeout"
-	FlagModelLoadWait         = "model-load-wait"
+	FlagMCPToolTimeout          = "mcp-tool-timeout"
+	FlagMCPDiscoveryTimeout     = "mcp-discovery-timeout"
+	FlagWorkerWaitTimeout       = "worker-wait-timeout"
+	FlagDrainTimeout            = "drain-timeout"
+	FlagHealthCheckInterval     = "health-check-interval"
+	FlagStaleNodeThreshold      = "stale-node-threshold"
+	FlagNodeHeartbeatCheckpoint = "node-heartbeat-checkpoint"
+	FlagMCPCIJobTimeout         = "mcp-ci-job-timeout"
+	FlagBackendInstallTimeout   = "backend-install-timeout"
+	FlagBackendUpgradeTimeout   = "backend-upgrade-timeout"
+	FlagModelLoadTimeout        = "model-load-timeout"
+	FlagModelLoadWait           = "model-load-wait"
 	// FlagDiskHeadroomCheck names the disk-headroom toggle. It is quoted in
 	// the warning the check emits while disabled, so the operator reading a
 	// log line knows exactly which knob produced it.
@@ -410,16 +434,22 @@ const (
 
 // Defaults for distributed timeouts.
 const (
-	DefaultMCPToolTimeout        = 360 * time.Second
-	DefaultMCPDiscoveryTimeout   = 60 * time.Second
-	DefaultWorkerWaitTimeout     = 5 * time.Minute
-	DefaultDrainTimeout          = 30 * time.Second
-	DefaultHealthCheckInterval   = 15 * time.Second
-	DefaultStaleNodeThreshold    = 60 * time.Second
-	DefaultMCPCIJobTimeout       = 10 * time.Minute
-	DefaultBackendInstallTimeout = 15 * time.Minute
-	DefaultBackendUpgradeTimeout = 15 * time.Minute
-	DefaultModelLoadTimeout      = 5 * time.Minute
+	DefaultMCPToolTimeout      = 360 * time.Second
+	DefaultMCPDiscoveryTimeout = 60 * time.Second
+	DefaultWorkerWaitTimeout   = 5 * time.Minute
+	DefaultDrainTimeout        = 30 * time.Second
+	DefaultHealthCheckInterval = 15 * time.Second
+	// A beat that only refreshes the timestamp is now dropped until the
+	// checkpoint interval elapses, so the persisted column is up to one
+	// interval stale by design. The threshold covers that plus jitter.
+	// A genuinely dead node is still caught sooner by the per-model gRPC
+	// health check and by request-time failure, neither of which reads this.
+	DefaultStaleNodeThreshold      = 5 * time.Minute
+	DefaultNodeHeartbeatCheckpoint = 60 * time.Second
+	DefaultMCPCIJobTimeout         = 10 * time.Minute
+	DefaultBackendInstallTimeout   = 15 * time.Minute
+	DefaultBackendUpgradeTimeout   = 15 * time.Minute
+	DefaultModelLoadTimeout        = 5 * time.Minute
 	// DefaultModelLoadWait is how long a request waits for a cold-loading model
 	// before it is answered with 503 and live progress. Chosen to sit under the
 	// idle timeout of typical ingress/LB defaults, so the answer comes from
@@ -519,6 +549,14 @@ func (c DistributedConfig) StaleNodeThresholdOrDefault() time.Duration {
 	return cmp.Or(c.StaleNodeThreshold, DefaultStaleNodeThreshold)
 }
 
+// NodeHeartbeatCheckpointOrDefault returns the configured interval or the
+// default. A configured zero is indistinguishable from unset here, which is
+// intentional: cmp.Or falls back to the default, and an operator who wants a
+// write per beat sets a value below the heartbeat interval instead.
+func (c DistributedConfig) NodeHeartbeatCheckpointOrDefault() time.Duration {
+	return cmp.Or(c.NodeHeartbeatCheckpoint, DefaultNodeHeartbeatCheckpoint)
+}
+
 // MCPCIJobTimeoutOrDefault returns the configured MCP CI job timeout or the default.
 func (c DistributedConfig) MCPCIJobTimeoutOrDefault() time.Duration {
 	return cmp.Or(c.MCPCIJobTimeout, DefaultMCPCIJobTimeout)

core/config/distributed_config_test.go

@@ -47,6 +47,27 @@ var _ = Describe("DistributedConfig backend NATS timeouts", func() {
 	})
 })
 
+// Heartbeat checkpointing makes last_heartbeat up to one checkpoint interval
+// stale by design, which is why the threshold defaults to 5 minutes. An
+// operator who widens the checkpoint has to widen this to match, so it has to
+// be reachable from the CLI rather than being a compile-time constant.
+var _ = Describe("DistributedConfig stale node threshold", func() {
+	It("defaults to 5 minutes, wide enough to cover a suppressed beat", func() {
+		Expect(config.DistributedConfig{}.StaleNodeThresholdOrDefault()).
+			To(Equal(5 * time.Minute))
+		Expect(config.DefaultStaleNodeThreshold).
+			To(BeNumerically(">", config.DefaultNodeHeartbeatCheckpoint),
+				"a threshold at or below the checkpoint interval marks healthy, "+
+					"beating nodes offline every cycle")
+	})
+
+	It("is configurable, so a widened checkpoint can be matched", func() {
+		o := config.NewApplicationConfig(config.WithStaleNodeThreshold(20 * time.Minute))
+		Expect(o.Distributed.StaleNodeThreshold).To(Equal(20 * time.Minute))
+		Expect(o.Distributed.StaleNodeThresholdOrDefault()).To(Equal(20 * time.Minute))
+	})
+})
+
 var _ = Describe("DistributedConfig flag-name constants", func() {
 	// Pin the kebab-case strings so a rename of the Go field name (or a
 	// CLI flag naming convention change) forces the constant to update,
@@ -62,6 +83,7 @@ var _ = Describe("DistributedConfig flag-name constants", func() {
 		Entry("drain timeout", config.FlagDrainTimeout, "drain-timeout"),
 		Entry("health check interval", config.FlagHealthCheckInterval, "health-check-interval"),
 		Entry("stale node threshold", config.FlagStaleNodeThreshold, "stale-node-threshold"),
+		Entry("node heartbeat checkpoint", config.FlagNodeHeartbeatCheckpoint, "node-heartbeat-checkpoint"),
 		Entry("MCP CI job timeout", config.FlagMCPCIJobTimeout, "mcp-ci-job-timeout"),
 		Entry("backend install timeout", config.FlagBackendInstallTimeout, "backend-install-timeout"),
 		Entry("backend upgrade timeout", config.FlagBackendUpgradeTimeout, "backend-upgrade-timeout"),