LocalAI/core/config/distributed_config.go

package config

import (
	"cmp"
	"fmt"
	"time"

	"github.com/mudler/xlog"
)

// DistributedConfig holds configuration for horizontal scaling mode.
// When Enabled is true, PostgreSQL is required. Nothing else is: fan-out rides
// PostgreSQL and every worker is reached over the tunnel it dials out.
type DistributedConfig struct {
	Enabled    bool   // --distributed / LOCALAI_DISTRIBUTED
	InstanceID string // --instance-id / LOCALAI_INSTANCE_ID (auto-generated UUID if empty)
	// AdvertiseAddr is the host:port OTHER REPLICAS dial to reach this one,
	// which is not the address this process binds: a replica behind a service
	// or a NAT binds one and is reached at another. Empty means "work it out",
	// by asking the kernel which local address routes to PostgreSQL; that
	// answer is only usable when the database is remote, so a deployment with
	// a local or sidecar database has to set this.
	AdvertiseAddr     string // LOCALAI_DISTRIBUTED_ADVERTISE_ADDR
	StorageURL        string // --storage-url / LOCALAI_STORAGE_URL (S3 endpoint)
	RegistrationToken string // --registration-token / LOCALAI_REGISTRATION_TOKEN (required token for node registration)
	// RegistrationRequireAuth fails startup when distributed mode is enabled but
	// RegistrationToken is empty. The default (false) keeps the historical
	// fail-open behavior with a loud warning; production should set it so the
	// node-register endpoints and the worker file-transfer server cannot run
	// unauthenticated.
	RegistrationRequireAuth bool // LOCALAI_REGISTRATION_REQUIRE_AUTH
	// RequireAuth is the umbrella switch (LOCALAI_DISTRIBUTED_REQUIRE_AUTH) for
	// distributed-mode auth. It implies RegistrationRequireAuth, and that is
	// now ALL it implies: it used to imply a NatsRequireAuth as well, and there
	// is no message bus left for that half to lock down. The granular flag
	// remains available.
	RequireAuth      bool // LOCALAI_DISTRIBUTED_REQUIRE_AUTH
	AutoApproveNodes bool // --auto-approve-nodes / LOCALAI_AUTO_APPROVE_NODES (skip admin approval for new workers)
	// SharedModels asserts that every node (frontend and workers) mounts the
	// SAME models directory at the SAME path (e.g. a shared volume, as in
	// docker-compose.distributed.yaml). When true, the router skips staging
	// model files to workers entirely: the frontend's absolute model paths are
	// already valid on the worker, so re-uploading them into a per-model
	// subdirectory only re-downloads what is already present (#10556). Default
	// false preserves the historical per-node staging behavior.
	SharedModels bool // --distributed-shared-models / LOCALAI_DISTRIBUTED_SHARED_MODELS

	// S3 configuration (used when StorageURL is set)
	StorageBucket    string // --storage-bucket / LOCALAI_STORAGE_BUCKET
	StorageRegion    string // --storage-region / LOCALAI_STORAGE_REGION
	StorageAccessKey string // --storage-access-key / LOCALAI_STORAGE_ACCESS_KEY
	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 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
	// crashed even though the worker's node-level heartbeat is still arriving.
	// Without per-model probing, /embeddings and /completions can be dispatched
	// to a backend that silently returns garbage (see also the cascading
	// model-row cleanup on MarkUnhealthy / MarkDraining).
	DisablePerModelHealthCheck bool

	MCPCIJobTimeout time.Duration // MCP CI job execution timeout (default 10m)

	// WorkerReconnectGrace is how long a worker whose tunnel was lost is
	// treated as reconnecting rather than gone. It is the ONLY thing that
	// separates a worker re-homing between frontend replicas from one that has
	// left, and absence is what makes the scheduler stop placing work and reap
	// the worker's rows, so a grace shorter than the worker's own reconnect
	// backoff condemns workers that are behaving exactly as designed.
	//
	// Zero means unset (DefaultWorkerReconnectGrace applies). Measured on the
	// database clock, so every replica agrees on when the window ends.
	WorkerReconnectGrace time.Duration // LOCALAI_WORKER_RECONNECT_GRACE

	BackendInstallTimeout time.Duration // control round-trip timeout for backend.install (default 15m)
	BackendUpgradeTimeout time.Duration // control round-trip timeout for backend.upgrade (default 15m)
	// ModelLoadTimeout is the gRPC deadline for the remote LoadModel call the
	// router issues once a worker has the backend installed and the model files
	// staged. It therefore covers only the backend's own checkpoint load and
	// pipeline init, which for a multi-tens-of-GB diffusion/video checkpoint on
	// unified memory can far exceed the 5m default.
	ModelLoadTimeout time.Duration // gRPC deadline for remote LoadModel (default 5m)
	// ModelLoadWait bounds how long an inference request waits for a model that
	// is being cold-loaded before it is answered with 503 plus live progress. A
	// held HTTP request cannot survive real infrastructure — an ingress or LB
	// idle timeout kills a 20-minute request regardless of what LocalAI does —
	// so the wait is bounded by default.
	//
	// Zero means unset (DefaultModelLoadWait applies); ModelLoadWaitUnbounded
	// records the operator asking for unbounded waiting with
	// LOCALAI_MODEL_LOAD_WAIT=0.
	ModelLoadWait time.Duration

	MaxUploadSize int64 // Maximum upload body size in bytes (default 50 GB)

	AgentWorkerConcurrency int `yaml:"agent_worker_concurrency" json:"agent_worker_concurrency" env:"LOCALAI_AGENT_WORKER_CONCURRENCY"`
	JobWorkerConcurrency   int `yaml:"job_worker_concurrency" json:"job_worker_concurrency" env:"LOCALAI_JOB_WORKER_CONCURRENCY"`

	// DiskHeadroomDisabled turns off the scheduler's free-disk admission check,
	// restoring the pre-#11054 behaviour where node selection ignores whether a
	// node can actually store the model. The check is ON by default because it
	// prevents a measured failure (a node with 0 bytes free accepted a 70GB
	// model and failed 16 minutes into staging); this is the escape hatch for
	// setups where our size estimate is wrong (deduplicating filesystems, a
	// worker that fetches its own weights), not the norm.
	//
	// Disabling does NOT silence the check: it still runs and warns when it
	// would have rejected every node, so the operator keeps the diagnosis
	// without being blocked. See SmartRouter.scheduleNewModel.
	//
	// Stored as the negation of the CLI/runtime flag so the zero value is
	// "enabled" (mirrors PrefixCacheDisabled).
	DiskHeadroomDisabled bool

	// PrefixCacheDisabled turns off prefix-cache-aware routing, falling back to
	// round-robin (the floor). Prefix-cache routing is ON by default in
	// distributed mode; this flag exists so operators can opt out. The CLI
	// surfaces a default-true --distributed-prefix-cache enable flag and sets
	// this when the operator passes --distributed-prefix-cache=false.
	PrefixCacheDisabled bool
	// PrefixCacheTTL is the idle-timeout for prefix-cache index entries and
	// drives the background eviction cadence (eviction runs every TTL/2). Zero
	// means use the prefixcache package default (5m).
	PrefixCacheTTL time.Duration
	// ModelSchedulingJSON is an inline JSON list of per-model scheduling configs
	// applied authoritatively at startup (LOCALAI_MODEL_SCHEDULING).
	ModelSchedulingJSON string
	// ModelSchedulingConfigPath is a path to a YAML file with the same list
	// (LOCALAI_MODEL_SCHEDULING_CONFIG).
	ModelSchedulingConfigPath string
}

// Validate checks that the distributed configuration is internally consistent.
// It returns nil if distributed mode is disabled.
func (c DistributedConfig) Validate() error {
	if !c.Enabled {
		return nil
	}
	// No message-bus URL is required, and none is dialled. The last family
	// that needed one was agent.<agent>.cancel, which now rides the agent
	// worker's own tunnel as a control verb; a distributed deployment needs
	// PostgreSQL and the frontends' own HTTP listener. The flag is still
	// accepted so an existing command line starts unchanged.
	// S3 credentials must be paired
	if (c.StorageAccessKey != "" && c.StorageSecretKey == "") ||
		(c.StorageAccessKey == "" && c.StorageSecretKey != "") {
		return fmt.Errorf("storage-access-key and storage-secret-key must both be set or both empty")
	}
	// The registration token guards both the node HTTP register/heartbeat
	// endpoints and the worker file-transfer server (which fails open on an
	// empty token). Enforce it when registration auth is required (the granular
	// flag or the umbrella); otherwise warn.
	if c.RegistrationToken == "" {
		if c.RegistrationAuthRequired() {
			return fmt.Errorf("registration auth is required (LOCALAI_REGISTRATION_REQUIRE_AUTH or LOCALAI_DISTRIBUTED_REQUIRE_AUTH) but LOCALAI_REGISTRATION_TOKEN is empty")
		}
		xlog.Warn("distributed mode running without registration token — node endpoints and the worker file-transfer server are unprotected; set LOCALAI_REGISTRATION_TOKEN, or LOCALAI_DISTRIBUTED_REQUIRE_AUTH=true to fail closed")
	}
	// 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,
		FlagNodeHeartbeatCheckpoint: c.NodeHeartbeatCheckpoint,
		FlagMCPCIJobTimeout:         c.MCPCIJobTimeout,
		FlagBackendInstallTimeout:   c.BackendInstallTimeout,
		FlagBackendUpgradeTimeout:   c.BackendUpgradeTimeout,
		FlagModelLoadTimeout:        c.ModelLoadTimeout,
		FlagWorkerReconnectGrace:    c.WorkerReconnectGrace,
	} {
		if d < 0 {
			return fmt.Errorf("%s must not be negative", name)
		}
	}
	return nil
}

// Distributed config options

var EnableDistributed = func(o *ApplicationConfig) {
	o.Distributed.Enabled = true
}

func WithDistributedInstanceID(id string) AppOption {
	return func(o *ApplicationConfig) {
		o.Distributed.InstanceID = id
	}
}

// WithDistributedAdvertiseAddr pins the host:port peers dial to reach this
// replica, overriding the route-based discovery.
func WithDistributedAdvertiseAddr(addr string) AppOption {
	return func(o *ApplicationConfig) {
		o.Distributed.AdvertiseAddr = addr
	}
}

func WithRegistrationToken(token string) AppOption {
	return func(o *ApplicationConfig) {
		o.Distributed.RegistrationToken = token
	}
}

// EnableRegistrationRequireAuth makes an empty registration token a hard error
// in distributed mode (see DistributedConfig.RegistrationRequireAuth).
var EnableRegistrationRequireAuth = func(o *ApplicationConfig) {
	o.Distributed.RegistrationRequireAuth = true
}

// EnableDistributedRequireAuth is the umbrella switch implying
// RegistrationRequireAuth (see DistributedConfig.RequireAuth).
var EnableDistributedRequireAuth = func(o *ApplicationConfig) {
	o.Distributed.RequireAuth = true
}

// RegistrationAuthRequired reports whether an empty registration token must be
// treated as a fatal misconfiguration — the granular flag or the umbrella.
func (c DistributedConfig) RegistrationAuthRequired() bool {
	return c.RegistrationRequireAuth || c.RequireAuth
}

func WithStorageURL(url string) AppOption {
	return func(o *ApplicationConfig) {
		o.Distributed.StorageURL = url
	}
}

func WithStorageBucket(bucket string) AppOption {
	return func(o *ApplicationConfig) {
		o.Distributed.StorageBucket = bucket
	}
}

func WithStorageRegion(region string) AppOption {
	return func(o *ApplicationConfig) {
		o.Distributed.StorageRegion = region
	}
}

func WithStorageAccessKey(key string) AppOption {
	return func(o *ApplicationConfig) {
		o.Distributed.StorageAccessKey = key
	}
}

func WithStorageSecretKey(key string) AppOption {
	return func(o *ApplicationConfig) {
		o.Distributed.StorageSecretKey = key
	}
}

// WithWorkerReconnectGrace sets how long a lost worker tunnel is read as
// reconnecting rather than gone (see DistributedConfig.WorkerReconnectGrace).
func WithWorkerReconnectGrace(d time.Duration) AppOption {
	return func(o *ApplicationConfig) {
		o.Distributed.WorkerReconnectGrace = d
	}
}

func WithBackendInstallTimeout(d time.Duration) AppOption {
	return func(o *ApplicationConfig) {
		o.Distributed.BackendInstallTimeout = d
	}
}

func WithBackendUpgradeTimeout(d time.Duration) AppOption {
	return func(o *ApplicationConfig) {
		o.Distributed.BackendUpgradeTimeout = d
	}
}

func WithModelLoadTimeout(d time.Duration) AppOption {
	return func(o *ApplicationConfig) {
		o.Distributed.ModelLoadTimeout = d
	}
}

// WithModelLoadWait sets how long a request waits for a cold-loading model. A
// zero d records the operator asking for unbounded waiting: "set the knob to
// zero" cannot sensibly mean "use the default".
func WithModelLoadWait(d time.Duration) AppOption {
	return func(o *ApplicationConfig) {
		if d == 0 {
			d = ModelLoadWaitUnbounded
		}
		o.Distributed.ModelLoadWait = d
	}
}

// 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
}

// EnableDistributedSharedModels marks the cluster as sharing one models
// directory across all nodes, so the router skips staging model files to
// workers (see DistributedConfig.SharedModels).
var EnableDistributedSharedModels = func(o *ApplicationConfig) {
	o.Distributed.SharedModels = true
}

// DisableDiskHeadroomCheck turns off the scheduler's free-disk admission
// check (see DistributedConfig.DiskHeadroomDisabled). The check is enabled by
// default in distributed mode.
var DisableDiskHeadroomCheck = func(o *ApplicationConfig) {
	o.Distributed.DiskHeadroomDisabled = true
}

// DisablePrefixCache turns off prefix-cache-aware routing (falls back to
// round-robin). Prefix-cache routing is enabled by default in distributed mode.
var DisablePrefixCache = func(o *ApplicationConfig) {
	o.Distributed.PrefixCacheDisabled = true
}

// WithPrefixCacheTTL sets the prefix-cache index idle-timeout (and the
// background eviction cadence, which runs every TTL/2).
func WithPrefixCacheTTL(d time.Duration) AppOption {
	return func(o *ApplicationConfig) {
		o.Distributed.PrefixCacheTTL = d
	}
}

// WithModelSchedulingJSON sets the inline-JSON declarative scheduling config.
func WithModelSchedulingJSON(s string) AppOption {
	return func(o *ApplicationConfig) {
		o.Distributed.ModelSchedulingJSON = s
	}
}

// WithModelSchedulingConfigPath sets the path to a YAML declarative scheduling
// config file.
func WithModelSchedulingConfigPath(path string) AppOption {
	return func(o *ApplicationConfig) {
		o.Distributed.ModelSchedulingConfigPath = path
	}
}

// Flag names for distributed timeout / interval configuration. These are
// the kebab-case identifiers kong derives from the matching RunCMD struct
// fields; they appear in Validate error messages and any other operator-
// facing surface that needs to reference a specific knob by name. Keeping
// 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"
	FlagNodeHeartbeatCheckpoint = "node-heartbeat-checkpoint"
	FlagMCPCIJobTimeout         = "mcp-ci-job-timeout"
	FlagBackendInstallTimeout   = "backend-install-timeout"
	FlagBackendUpgradeTimeout   = "backend-upgrade-timeout"
	FlagModelLoadTimeout        = "model-load-timeout"
	// FlagWorkerReconnectGrace names the reconnect-grace knob. Validate quotes
	// it when the operator hands it a negative duration.
	FlagWorkerReconnectGrace = "worker-reconnect-grace"
	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.
	FlagDiskHeadroomCheck = "distributed-disk-headroom-check"
)

// 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
	// 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
	// LocalAI (with progress the client can act on) rather than from a proxy
	// dropping the connection.
	DefaultModelLoadWait = 60 * time.Second
	// DefaultWorkerReconnectGrace covers a worker that misses one reconnect at
	// the ceiling and lands on the next, with margin. The worker's own numbers
	// (core/services/worker/tunnel.go) are a 30s backoff ceiling
	// (tunnelBackoffMax) and a 10s dial budget (tunnelHandshakeTimeout), so two
	// ceiling waits with a hung dial between them puts the worker back at 70s,
	// not 60s: two waits alone is the boundary, not a bound.
	//
	// The ceiling is reachable precisely when it matters. The backoff resets
	// only after a session that lasted tunnelHealthyAfter (30s), which a
	// replica accepting a dial and then dying denies, so a worker crossing a
	// rolling frontend restart climbs to the ceiling rather than sitting near
	// the 500ms floor.
	//
	// 90s therefore has margin where 60s sat on the edge. The asymmetry is
	// deliberate: too short and a worker that is reconnecting exactly as
	// designed is reported GONE, which licenses a reap and costs a model
	// reload; too long and a worker that really has died is reaped later. The
	// second is cheaper, so the default errs long.
	//
	// Raising it further makes a rolling frontend restart safer still; lowering
	// it reaps a dead worker sooner. There IS a value at which a live worker is
	// reported as gone: any grace shorter than that worker's actual reconnect.
	// That is why this is a duration and not a boolean.
	DefaultWorkerReconnectGrace = 90 * time.Second
)

// ModelLoadWaitUnbounded records LOCALAI_MODEL_LOAD_WAIT=0 — "wait as long as
// it takes" — which a plain zero cannot express, since zero also means "unset,
// use the default". Only deployments with no proxy in front should use it.
const ModelLoadWaitUnbounded = -1 * time.Second

// DefaultMaxUploadSize is the default maximum upload body size (50 GB).
const DefaultMaxUploadSize int64 = 50 << 30

// ReconnectGraceOrDefault returns the configured worker reconnect grace or the
// default.
//
// A non-positive value falls back rather than being taken verbatim, which is
// the opposite of what the timeout knobs above do, and deliberately. A
// negative grace makes every departure older than the window the instant it is
// stamped, so a worker two seconds into a normal reconnect reports as GONE, and
// gone is the one answer a caller may reap and evict on. Validate rejects a
// negative duration at startup; this is the second line, for a config built in
// code that never went through it.
func (c DistributedConfig) ReconnectGraceOrDefault() time.Duration {
	if c.WorkerReconnectGrace <= 0 {
		return DefaultWorkerReconnectGrace
	}
	return c.WorkerReconnectGrace
}

// BackendInstallTimeoutOrDefault returns the configured timeout or the default.
func (c DistributedConfig) BackendInstallTimeoutOrDefault() time.Duration {
	return cmp.Or(c.BackendInstallTimeout, DefaultBackendInstallTimeout)
}

// BackendUpgradeTimeoutOrDefault returns the configured timeout or the default.
func (c DistributedConfig) BackendUpgradeTimeoutOrDefault() time.Duration {
	return cmp.Or(c.BackendUpgradeTimeout, DefaultBackendUpgradeTimeout)
}

// ModelLoadTimeoutOrDefault returns the configured timeout or the default.
func (c DistributedConfig) ModelLoadTimeoutOrDefault() time.Duration {
	return cmp.Or(c.ModelLoadTimeout, DefaultModelLoadTimeout)
}

// MCPToolTimeoutOrDefault returns the configured timeout or the default.
func (c DistributedConfig) MCPToolTimeoutOrDefault() time.Duration {
	return cmp.Or(c.MCPToolTimeout, DefaultMCPToolTimeout)
}

// MCPDiscoveryTimeoutOrDefault returns the configured timeout or the default.
func (c DistributedConfig) MCPDiscoveryTimeoutOrDefault() time.Duration {
	return cmp.Or(c.MCPDiscoveryTimeout, DefaultMCPDiscoveryTimeout)
}

// WorkerWaitTimeoutOrDefault returns the configured timeout or the default.
func (c DistributedConfig) WorkerWaitTimeoutOrDefault() time.Duration {
	return cmp.Or(c.WorkerWaitTimeout, DefaultWorkerWaitTimeout)
}

// DrainTimeoutOrDefault returns the configured timeout or the default.
func (c DistributedConfig) DrainTimeoutOrDefault() time.Duration {
	return cmp.Or(c.DrainTimeout, DefaultDrainTimeout)
}

// HealthCheckIntervalOrDefault returns the configured interval or the default.
func (c DistributedConfig) HealthCheckIntervalOrDefault() time.Duration {
	return cmp.Or(c.HealthCheckInterval, DefaultHealthCheckInterval)
}

// StaleNodeThresholdOrDefault returns the configured threshold or the default.
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)
}

// MaxUploadSizeOrDefault returns the configured max upload size or the default.
func (c DistributedConfig) MaxUploadSizeOrDefault() int64 {
	return cmp.Or(c.MaxUploadSize, DefaultMaxUploadSize)
}