diff --git a/docs/daemon-single-owner.md b/docs/daemon-single-owner.md new file mode 100644 index 0000000000..ad435e66a5 --- /dev/null +++ b/docs/daemon-single-owner.md @@ -0,0 +1,122 @@ +# Single-daemon ownership and version compatibility + +A machine may run at most **one** Multica agent daemon at a time. Two daemons on +one host — most commonly the CLI-spawned daemon and the Desktop-spawned daemon +running under different profiles — share the same local checkouts and the same +in-process local-directory mutex, so both can write the same repository at once +and corrupt it (VWO-364 / VWO-365). + +This is enforced by a **machine-global advisory lock** and a **client/server +version gate**. This document is the operator reference for both. + +## How ownership is enforced + +- On startup, before it binds its health port or registers any runtime, a + daemon takes an exclusive advisory lock on `~/.multica/daemon.lock`. +- That path lives in the **base** config directory, which every profile shares, + so the lock is contended across profiles — unlike the per-profile health-port + and PID-file guards, which never see a daemon under a different profile. +- The lock is held for the life of the process via an open file handle. The OS + releases it automatically when the process exits — cleanly **or on a crash** — + so there is no lease TTL to tune and no stale-lock window. A crashed owner's + lock is gone the instant the process dies. +- A second daemon that tries to start while an owner is live fails immediately + with an actionable error naming the incumbent (pid, version, profile, health + port) instead of silently double-registering. +- The launcher also refuses to start while a daemon that does NOT hold the lock + is alive on any known profile's health port — that can only be a daemon from a + release predating single-daemon ownership (or one started under the + break-glass env). This closes the rolling-upgrade window: when upgrading, stop + every old daemon (`multica [--profile ] daemon stop`) before starting + the new one; the sweep enforces it rather than silently running alongside. + +## How version compatibility is enforced + +- The server reports its build version in the daemon registration response + (`server_version`). The daemon compares it against the minimum server version + it requires (currently the first release carrying the task `prepare-lease` + route) and, if the server is **older**, fails to start with a clear error + rather than 404-looping on unsupported routes forever. +- A server that reports **no** version (older than this field, or built without + a version stamp) is treated as "unknown": the daemon warns and proceeds. If it + then hits a missing route (a bare `prepare-lease` 404), it logs one loud error + and stops that retry loop — it never floods the log with repeated 404s. + +## Operator verification + +Confirm exactly one daemon owns the machine: + +```bash +# 1. Exactly one daemon process. +pgrep -fl 'multica daemon start --foreground' + +# 2. The lock records the live owner (pid, health port, version, profile). +cat ~/.multica/daemon.lock + +# 3. The recorded pid is the running daemon. +ps -p "$(python3 -c 'import json,os;print(json.load(open(os.path.expanduser("~/.multica/daemon.lock")))["pid"])')" + +# 4. Health endpoint agrees. +multica daemon status +``` + +Confirm the second daemon is refused (safe to run — it will not start): + +```bash +# With a daemon already running, this must exit non-zero with an ownership error: +multica daemon start --profile some-other-profile +# => "another Multica daemon already owns this machine (pid …) … `multica daemon stop` … `--takeover`" +``` + +Confirm version compatibility on a self-hosted server: + +```bash +# The server build the daemon negotiated against. +multica daemon logs | grep -Ei 'server version|incompatible|prepare-lease' +``` + +## Handing ownership over (supported takeover) + +To replace the running daemon (e.g. switch the owner from the CLI daemon to the +Desktop daemon) without editing any file or database row: + +```bash +# Ask the current owner to stop, wait for it to release, then start: +multica daemon start --takeover +# (or `multica daemon restart` for a same-profile stop+start) +``` + +`--takeover` reads the incumbent's health port from the lock file, requests a +graceful shutdown over HTTP (cross-platform; no OS signals), waits up to 45s for +the lock to free (sized to the incumbent's full graceful shutdown: a 30s +in-flight task drain plus deregistration), then acquires it. If the incumbent +does not release in time it fails loudly with a manual remedy — it never +force-kills, since that would skip the task drain and could orphan agent +subprocesses mid-write. + +## Rollback / break-glass + +The ownership guard can be disabled without redeploying: + +```bash +# Restore the pre-change behavior (NO single-owner guard). Only for a deliberate +# multi-backend setup that accepts the shared-checkout risk, or emergency rollback. +export MULTICA_DAEMON_ALLOW_MULTIPLE=1 +multica daemon start +``` + +With the variable set (any value other than empty / `0` / `false`), the daemon +skips lock acquisition and logs a loud warning on every start. Unset it and +restart to restore the guard. The version gate is independent and is not +affected by this variable. + +## Recovery scenarios + +| Situation | What happens | Operator action | +|---|---|---| +| Owner exits cleanly | Lock released on shutdown | Next daemon starts normally | +| Owner crashes / is killed | OS drops the lock on process death | Next daemon starts immediately; no cleanup | +| Leftover `daemon.lock` from a dead owner | File content is advisory only; no live lock | Next daemon acquires and overwrites it; no manual deletion | +| Second daemon attempted | Refused with an actionable error | Use `daemon stop`, or `daemon start --takeover` | +| Old (pre-lock) daemon still running during an upgrade | New daemon refuses to start alongside it | Stop the old daemon (`multica [--profile ] daemon stop`), then start | +| Daemon newer than server | Startup fails with a version error (or one loud `prepare-lease` error) | Upgrade the server, or run a matching daemon | diff --git a/server/cmd/multica/cmd_daemon.go b/server/cmd/multica/cmd_daemon.go index abeee1fb69..9de6a064f8 100644 --- a/server/cmd/multica/cmd_daemon.go +++ b/server/cmd/multica/cmd_daemon.go @@ -11,6 +11,7 @@ import ( "os" "os/exec" "path/filepath" + "runtime" "sort" "strconv" "strings" @@ -81,6 +82,7 @@ var daemonDiskUsageCmd = &cobra.Command{ func init() { f := daemonStartCmd.Flags() f.Bool("foreground", false, "Run in the foreground instead of background") + f.Bool("takeover", false, "If another daemon already owns this machine, ask it to stop and take over instead of failing") f.String("daemon-id", "", "Unique daemon identifier (env: MULTICA_DAEMON_ID)") f.String("device-name", "", "Human-readable device name (env: MULTICA_DAEMON_DEVICE_NAME)") f.String("runtime-name", "", "Runtime display name (env: MULTICA_AGENT_RUNTIME_NAME)") @@ -101,6 +103,7 @@ func init() { // restart shares all the same flags as start rf := daemonRestartCmd.Flags() rf.Bool("foreground", false, "Run in the foreground instead of background") + rf.Bool("takeover", false, "If another daemon already owns this machine, ask it to stop and take over instead of failing") rf.String("daemon-id", "", "Unique daemon identifier (env: MULTICA_DAEMON_ID)") rf.String("device-name", "", "Human-readable device name (env: MULTICA_DAEMON_DEVICE_NAME)") rf.String("runtime-name", "", "Runtime display name (env: MULTICA_AGENT_RUNTIME_NAME)") @@ -234,6 +237,179 @@ func healthPortForProfile(profile string) int { return daemon.DefaultHealthPort + 1 + (h % 1000) } +// prepareDaemonOwnership makes sure it is safe to (re)start a daemon on this +// machine before one is spawned. At most one daemon may own the machine-global +// lock at cli.ProfileDir("")/daemon.lock, which spans every profile — so it +// catches a Desktop-spawned daemon running under a different profile that the +// per-profile health-port guard cannot see (VWO-365). +// +// It runs in the LAUNCHER purely to give a fast, actionable error (or perform a +// requested takeover); the spawned daemon's Run re-acquires the lock for real. +// +// - break-glass (MULTICA_DAEMON_ALLOW_MULTIPLE set) → skip. +// - lock free → nil. +// - lock held, --takeover → ask the incumbent to stop, wait for release. +// - lock held, no --takeover → the actionable conflict error. +// +// ownershipHandoffWait bounds how long the launcher waits for a shutting-down +// incumbent to release the machine lock. It covers the gap between a daemon's +// health port closing (at the start of shutdown) and its ownership lock +// releasing at the very end, so a `daemon restart`, a stop-then-start, or a +// --takeover does not spuriously report a conflict/timeout mid-shutdown. +// Sized to the full graceful-shutdown bound: pollLoop waits up to 30s for +// in-flight tasks, then deregisterRuntimes has a 5s timeout — 45s adds margin. +const ownershipHandoffWait = 45 * time.Second + +func prepareDaemonOwnership(cmd *cobra.Command) error { + if daemon.OwnershipBypassed() { + return nil + } + baseDir, err := cli.ProfileDir("") + if err != nil { + return fmt.Errorf("resolve base config dir for daemon ownership lock: %w", err) + } + free, incumbent, ok, err := daemon.ProbeOwnership(baseDir) + if err != nil { + return err + } + if !free { + if takeover, _ := cmd.Flags().GetBool("takeover"); takeover { + if err := takeoverDaemonOwner(baseDir, incumbent); err != nil { + return err + } + } else if ok && incumbent.HealthPort > 0 && daemonAliveOnPort(incumbent.HealthPort) { + // The lock is held and the incumbent is genuinely live: reject + // immediately with an actionable error. + return ownershipConflictErr(baseDir, incumbent, ok) + } else if !waitForOwnershipFree(baseDir, ownershipHandoffWait) { + // Not answering health means it is shutting down — the daemon a + // `restart`/`stop` just asked to exit, still draining tasks and + // deregistering before it releases the lock — so wait a bounded + // time for the clean handoff rather than reject a legitimate + // restart. A wedged daemon that never releases lands here. + return ownershipConflictErr(baseDir, incumbent, ok) + } + } + // The lock is (now) free. A daemon can still be running WITHOUT it: one + // from a release predating single-daemon ownership, or one started under + // the break-glass env. Starting alongside it would recreate the exact + // two-writer hazard the lock exists to prevent, so sweep every known + // profile's health port and refuse while such a daemon is alive. + return detectUnlockedDaemon(baseDir) +} + +// detectUnlockedDaemon scans every known profile's health port for a daemon +// running WITHOUT holding the machine lock. Only reachable while the lock is +// free — and a lock-aware daemon acquires the lock BEFORE it binds its health +// port — so an alive health endpoint here can only be a daemon from a pre-lock +// release or one bypassing the guard. This closes the rolling-upgrade blind +// spot: without it, upgrading the CLI while a legacy daemon runs under another +// profile (e.g. the Desktop's desktop- profile) would let the new daemon +// start alongside it. Ports are derived from the profile directories that +// actually exist plus the default profile, so the sweep is a handful of +// fail-fast local probes, not a port scan. +func detectUnlockedDaemon(baseDir string) error { + if daemon.OwnershipBypassed() { + return nil + } + ports := map[int]string{daemon.DefaultHealthPort: ""} // port -> profile ("" = default) + if entries, err := os.ReadDir(filepath.Join(baseDir, "profiles")); err == nil { + for _, e := range entries { + if e.IsDir() { + ports[healthPortForProfile(e.Name())] = e.Name() + } + } + } + for port, profile := range ports { + health := probeDaemonHealth(port) + if !daemonAlive(health) { + continue + } + // A daemon visible on localhost but running in an environment this + // launcher can't manage — e.g. a Linux daemon inside WSL2 reachable + // through Windows localhost forwarding (#3916) — is not sharing this + // home's checkouts the way a native daemon does, and its lifecycle is + // not ours to gate on. Skip it rather than refuse with a stop hint the + // user can't act on. A daemon that reports no OS (older release) is + // treated as native: fail safe toward refusing. + if hostOS, _ := health["os"].(string); hostOS != "" && hostOS != runtime.GOOS { + continue + } + label := "the default profile" + stopHint := "`multica daemon stop`" + if profile != "" { + label = fmt.Sprintf("profile %q", profile) + stopHint = fmt.Sprintf("`multica --profile %s daemon stop`", profile) + } + return fmt.Errorf( + "a daemon that does not hold the machine ownership lock is running for %s (health port %d) — likely an older Multica release from before single-daemon ownership; stop it with %s and retry, or set MULTICA_DAEMON_ALLOW_MULTIPLE=1 to bypass the guard at your own risk", + label, port, stopHint, + ) + } + return nil +} + +func ownershipConflictErr(baseDir string, incumbent daemon.OwnerInfo, hasInfo bool) error { + return &daemon.OwnershipConflict{Path: daemon.OwnershipLockPath(baseDir), Incumbent: incumbent, HasInfo: hasInfo} +} + +// daemonAliveOnPort reports whether a daemon health endpoint answers on port. +func daemonAliveOnPort(port int) bool { + ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second) + defer cancel() + return daemonAlive(checkDaemonHealthOnPort(ctx, port)) +} + +// probeDaemonHealth is an indirection over checkDaemonHealthOnPort so tests can +// stub port liveness instead of binding real well-known ports (which would +// collide with an actual daemon on a developer machine). Returns the raw +// health map; a dead port yields a map daemonAlive() rejects. +var probeDaemonHealth = func(port int) map[string]any { + ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second) + defer cancel() + return checkDaemonHealthOnPort(ctx, port) +} + +// waitForOwnershipFree polls until the machine lock is free or timeout elapses. +func waitForOwnershipFree(baseDir string, timeout time.Duration) bool { + deadline := time.Now().Add(timeout) + for time.Now().Before(deadline) { + if free, _, _, err := daemon.ProbeOwnership(baseDir); err == nil && free { + return true + } + time.Sleep(200 * time.Millisecond) + } + return false +} + +// takeoverDaemonOwner asks the incumbent daemon to shut down (via its recorded +// health port — cross-platform, no OS signals) and waits until it releases the +// machine lock. The OS drops the incumbent's advisory lock the instant its +// process exits, so a successful shutdown frees the lock with no stale window. +func takeoverDaemonOwner(baseDir string, incumbent daemon.OwnerInfo) error { + if incumbent.HealthPort <= 0 { + return fmt.Errorf("cannot take over: owner health port unknown (owner pid %d); stop it manually with `multica daemon stop`", incumbent.PID) + } + fmt.Fprintf(os.Stderr, "Taking over from daemon pid %d (profile %q) on health port %d...\n", + incumbent.PID, incumbent.Profile, incumbent.HealthPort) + if err := requestDaemonShutdown(incumbent.HealthPort); err != nil { + // A refused connection here is EXPECTED when the incumbent is already + // shutting down: it closes its health listener at the start of shutdown + // while holding the lock through the 30s task drain + deregister. Never + // force-kill on this path — SIGKILL would skip the drain and the + // deregister, and orphaned agent subprocesses could still be writing a + // checkout when the new owner starts dispatching. Fall through to the + // bounded wait; a genuinely wedged incumbent fails loudly below with a + // manual remedy instead of being forced. + fmt.Fprintf(os.Stderr, "Shutdown request not delivered (%v); waiting for the daemon to release the machine on its own...\n", err) + } + if waitForOwnershipFree(baseDir, ownershipHandoffWait) { + fmt.Fprintln(os.Stderr, "Previous daemon released the machine; continuing.") + return nil + } + return fmt.Errorf("takeover timed out: daemon pid %d did not release the machine lock within %s (it may still be draining an in-flight task); retry, or if it is truly stuck stop it manually (kill %d) and start again", incumbent.PID, ownershipHandoffWait, incumbent.PID) +} + // --- daemon start --- func runDaemonStart(cmd *cobra.Command, _ []string) error { @@ -261,6 +437,14 @@ func runDaemonBackground(cmd *cobra.Command) error { return fmt.Errorf("%s is already running (pid %v). Use 'daemon restart' to restart it", label, int(pid)) } + // Reject (or take over) a daemon owning this machine under ANY profile — the + // cross-profile case the same-profile health check above misses (VWO-365). + // Done before spawning so the user sees an actionable error immediately, not + // the generic "may not have started" timeout after the child fails to lock. + if err := prepareDaemonOwnership(cmd); err != nil { + return err + } + // Resolve current executable so the foreground child reuses this binary. exePath, err := selfexec.Resolve() if err != nil { @@ -424,6 +608,30 @@ func runDaemonForeground(cmd *cobra.Command) error { profile := resolveProfile(cmd) + // Honor --takeover on a direct `daemon start --foreground --takeover`: free + // the incumbent before Run tries to acquire the machine lock. Without + // --takeover we skip the lock pre-check — Run's acquire is the sole + // authority and returns the conflict error itself — so the common + // background→foreground child path (which inherits no --takeover) does not + // re-probe the lock. + if takeover, _ := cmd.Flags().GetBool("takeover"); takeover { + if err := prepareDaemonOwnership(cmd); err != nil { + return err + } + } else if baseDir, err := cli.ProfileDir(""); err == nil { + // The unlocked-daemon sweep, however, must run on EVERY launch path: + // Run's lock acquire cannot see a pre-lock daemon (it holds no lock), so + // without this the documented --foreground debugging path could start + // alongside a legacy daemon during the rolling-upgrade window. Cheap — + // a handful of fail-fast local probes — and a no-op re-check when this + // process is the background launcher's child (the launcher swept + // moments earlier). The takeover branch above already ends with the + // same sweep inside prepareDaemonOwnership. + if err := detectUnlockedDaemon(baseDir); err != nil { + return err + } + } + // Pick the log sink. A user who runs `daemon start --foreground` in a shell // keeps live, colored logging on their terminal (a documented debugging // path — see docs troubleshooting). A detached/background child, whose diff --git a/server/cmd/multica/cmd_daemon_ownership_test.go b/server/cmd/multica/cmd_daemon_ownership_test.go new file mode 100644 index 0000000000..fe7c195b27 --- /dev/null +++ b/server/cmd/multica/cmd_daemon_ownership_test.go @@ -0,0 +1,165 @@ +package main + +import ( + "net/http" + "net/http/httptest" + "net/url" + "os" + "path/filepath" + "runtime" + "strconv" + "strings" + "sync" + "testing" + "time" + + "github.com/multica-ai/multica/server/internal/daemon" +) + +// TestTakeoverDaemonOwner_ShutsDownAndWaits covers supported takeover: the +// launcher asks the incumbent to stop via its recorded health port, waits for +// it to release the machine lock, and then the lock is free for the successor. +// The incumbent is stood in for by a held same-process lock that a fake +// /shutdown endpoint releases (flock treats two opens of one file as +// contending, even in one process). +func TestTakeoverDaemonOwner_ShutsDownAndWaits(t *testing.T) { + baseDir := t.TempDir() + + held, err := daemon.AcquireOwnership(baseDir, daemon.OwnerInfo{PID: 4242, Version: "0.4.2", StartedAt: time.Now()}) + if err != nil { + t.Fatalf("seed incumbent lock: %v", err) + } + var releaseOnce sync.Once + defer held.Release() + + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if r.URL.Path == "/shutdown" { + releaseOnce.Do(func() { _ = held.Release() }) + w.WriteHeader(http.StatusOK) + return + } + w.WriteHeader(http.StatusNotFound) + })) + defer srv.Close() + + port := mustPort(t, srv.URL) + if err := takeoverDaemonOwner(baseDir, daemon.OwnerInfo{PID: 4242, HealthPort: port}); err != nil { + t.Fatalf("takeover should succeed once the incumbent releases: %v", err) + } + + // The lock must now be free for the successor. + free, _, _, err := daemon.ProbeOwnership(baseDir) + if err != nil { + t.Fatalf("probe after takeover: %v", err) + } + if !free { + t.Fatal("lock should be free after a successful takeover") + } +} + +// TestTakeoverDaemonOwner_NoHealthPort covers the case where the incumbent +// recorded no reachable health port: takeover cannot proceed and must return an +// actionable error rather than hang. +func TestTakeoverDaemonOwner_NoHealthPort(t *testing.T) { + baseDir := t.TempDir() + err := takeoverDaemonOwner(baseDir, daemon.OwnerInfo{PID: 4242, HealthPort: 0}) + if err == nil { + t.Fatal("takeover with no health port must error") + } +} + +// TestDetectUnlockedDaemon covers the rolling-upgrade blind spot: a daemon +// running WITHOUT the machine lock (a pre-lock release, or one started under +// the break-glass env) must block a new daemon from starting alongside it. +// Port liveness is stubbed so the test never touches real well-known ports — +// a developer machine may have an actual daemon on them. +func TestDetectUnlockedDaemon(t *testing.T) { + restore := probeDaemonHealth + t.Cleanup(func() { probeDaemonHealth = restore }) + + nativeAlive := map[string]any{"status": "running", "os": runtime.GOOS} + dead := map[string]any{} + foreignOS := "windows" + if runtime.GOOS == "windows" { + foreignOS = "linux" + } + + t.Run("refuses while an unlocked daemon is alive", func(t *testing.T) { + baseDir := t.TempDir() + if err := os.MkdirAll(filepath.Join(baseDir, "profiles", "desktop-host"), 0o755); err != nil { + t.Fatal(err) + } + legacyPort := healthPortForProfile("desktop-host") + probeDaemonHealth = func(port int) map[string]any { + if port == legacyPort { + return nativeAlive + } + return dead + } + + err := detectUnlockedDaemon(baseDir) + if err == nil { + t.Fatal("an alive unlocked daemon must refuse the start") + } + for _, want := range []string{"--profile desktop-host", "MULTICA_DAEMON_ALLOW_MULTIPLE"} { + if !strings.Contains(err.Error(), want) { + t.Fatalf("error missing %q: %v", want, err) + } + } + }) + + t.Run("clean when nothing is alive", func(t *testing.T) { + baseDir := t.TempDir() + probeDaemonHealth = func(int) map[string]any { return dead } + if err := detectUnlockedDaemon(baseDir); err != nil { + t.Fatalf("no live daemons must mean no error: %v", err) + } + }) + + t.Run("skips a foreign-environment daemon", func(t *testing.T) { + // #3916 topology: a daemon visible through localhost forwarding but + // running under a different OS (e.g. WSL2 behind Windows). Its + // lifecycle is not this launcher's to gate on — must not refuse. + baseDir := t.TempDir() + probeDaemonHealth = func(int) map[string]any { + return map[string]any{"status": "running", "os": foreignOS} + } + if err := detectUnlockedDaemon(baseDir); err != nil { + t.Fatalf("a foreign-OS daemon must not block startup: %v", err) + } + }) + + t.Run("fails safe when the daemon reports no OS", func(t *testing.T) { + // An older release that predates the os field is exactly the legacy + // daemon the sweep exists to catch — missing os must refuse. + baseDir := t.TempDir() + probeDaemonHealth = func(int) map[string]any { + return map[string]any{"status": "running"} + } + if err := detectUnlockedDaemon(baseDir); err == nil { + t.Fatal("a live daemon with no reported OS must refuse the start") + } + }) + + t.Run("break-glass bypasses the sweep", func(t *testing.T) { + baseDir := t.TempDir() + probeDaemonHealth = func(int) map[string]any { return nativeAlive } + t.Setenv("MULTICA_DAEMON_ALLOW_MULTIPLE", "1") + if err := detectUnlockedDaemon(baseDir); err != nil { + t.Fatalf("break-glass must bypass the sweep: %v", err) + } + }) +} + +func mustPort(t *testing.T, rawURL string) int { + t.Helper() + u, err := url.Parse(rawURL) + if err != nil { + t.Fatalf("parse %q: %v", rawURL, err) + } + p, err := strconv.Atoi(u.Port()) + if err != nil { + t.Fatalf("port from %q: %v", rawURL, err) + } + return p +} diff --git a/server/internal/daemon/client.go b/server/internal/daemon/client.go index 196866f65c..db16c3a321 100644 --- a/server/internal/daemon/client.go +++ b/server/internal/daemon/client.go @@ -56,6 +56,32 @@ func isTaskNotFoundError(err error) bool { return strings.Contains(strings.ToLower(reqErr.Body), "task not found") } +// isPrepareLeaseUnsupported returns true when err is a 404 from the +// /prepare-lease route that means the ROUTE itself is unregistered on an +// un-upgraded server (< MinServerVersion), not that the task/runtime row is +// gone. The daemon uses this to fail loudly on a client/server version skew +// the first time it bites, instead of Warn-looping every 15s forever +// (VWO-364/VWO-365). +// +// It matches the POSITIVE signature of an unregistered route: chi's default +// NotFound handler writes plain-text "404 page not found" (the router does not +// override it), while every handler-level 404 goes through writeError as JSON +// {"error": ...} — including "task not found"/"runtime not found" AND the +// access-check paths that map transient DB errors to a generic not-found. +// Exclusion-based matching here would let one DB hiccup on a single refresh +// tick masquerade as version skew and permanently stop the lease extender, +// lapsing the lease mid-task; a transient handler 404 must stay retryable. +func isPrepareLeaseUnsupported(err error) bool { + var reqErr *requestError + if !errors.As(err, &reqErr) { + return false + } + if reqErr.StatusCode != http.StatusNotFound { + return false + } + return strings.Contains(strings.ToLower(reqErr.Body), "404 page not found") +} + // isUnauthorizedError returns true if the error is a 401 from the server. // Used by the token-renewal loop to surface a clear "re-login required" // message instead of a generic transport-level retry. @@ -665,6 +691,10 @@ type RegisterResponse struct { Repos []RepoData `json:"repos"` ReposVersion string `json:"repos_version"` Settings json.RawMessage `json:"settings,omitempty"` + // ServerVersion is the running server build version, used to detect a + // client/server version skew loudly at startup. Empty against a server + // that predates this field (treated as "unknown", not incompatible). + ServerVersion string `json:"server_version,omitempty"` } func (c *Client) Register(ctx context.Context, req map[string]any) (*RegisterResponse, error) { diff --git a/server/internal/daemon/daemon.go b/server/internal/daemon/daemon.go index bea576437e..40276ce8b4 100644 --- a/server/internal/daemon/daemon.go +++ b/server/internal/daemon/daemon.go @@ -176,6 +176,9 @@ type Daemon struct { versionsMu sync.RWMutex // guards agentVersions agentVersions map[string]string // provider -> detected CLI version (set during registration) + serverVersionMu sync.Mutex // guards serverVersion + serverVersion string // last non-empty server_version reported at register (VWO-365 skew gate) + // resolvedPathsMu guards resolvedPaths, the self-healed executable paths. // The daemon pins each agent's absolute path at startup so a later PATH // change can't redirect a task launch. When that pinned path later vanishes @@ -922,7 +925,27 @@ func (d *Daemon) Run(ctx context.Context) error { d.cancelFunc = cancel d.rootCtx = ctx - // Bind health port early to detect another running daemon. + // Take the machine-global ownership lock before ANY side effect (health + // port, registration, task claim). This is the single authoritative guard + // that a second daemon process on this machine — most commonly the + // Desktop-spawned daemon racing the CLI one under a different profile — + // cannot advertise the same identity/runtime IDs or write the same + // checkout while an owner is live. The per-profile health-port bind below + // only catches a same-profile duplicate; this catches every profile. + // Deferred first so it releases LAST, after deregisterRuntimes. + ownership, err := d.acquireOwnership() + if err != nil { + return err + } + if ownership != nil { + defer func() { + if rerr := ownership.Release(); rerr != nil { + d.logger.Warn("release daemon ownership lock failed", "error", rerr) + } + }() + } + + // Bind health port early to detect a same-profile running daemon. healthLn, err := d.listenHealth() if err != nil { return err @@ -988,9 +1011,21 @@ func (d *Daemon) Run(ctx context.Context) error { return err } - // Deregister runtimes on shutdown (uses a fresh context since ctx will be cancelled). + // Deregister runtimes on shutdown (uses a fresh context since ctx will be + // cancelled). Installed BEFORE the version gate below: preflightAuth has + // already registered runtimes, so a version-mismatch exit must still take + // them offline rather than leave the server routing tasks to a dead + // process until the stale-runtime sweeper catches up. defer d.deregisterRuntimes() + // Fail loudly on a client/server version skew now, after the initial + // register learned the server_version but before any task is claimed — + // instead of leaving a newer daemon to 404-loop on routes an older server + // lacks (VWO-365). + if err := d.checkServerCompatibility(); err != nil { + return err + } + // Start workspace sync loop to discover newly created workspaces. go d.workspaceSyncLoop(ctx) @@ -1019,6 +1054,84 @@ func (d *Daemon) RestartBinary() string { return d.restartBinary } +// acquireOwnership takes the machine-global daemon ownership lock unless the +// operator has explicitly disabled the guard. It returns (nil, nil) when the +// guard is bypassed so Run proceeds without a lock (and without a release). +// +// The lock lives at cli.ProfileDir("")/daemon.lock — the base config dir shared +// by every profile — so it excludes a Desktop-spawned daemon under a different +// profile, which the per-profile health-port guard does not. +func (d *Daemon) acquireOwnership() (*OwnershipLock, error) { + if OwnershipBypassed() { + d.logger.Warn("daemon ownership guard disabled via MULTICA_DAEMON_ALLOW_MULTIPLE; " + + "multiple daemons may write the same checkout concurrently and corrupt it — unset this to restore the guard") + return nil, nil + } + baseDir, err := cli.ProfileDir("") + if err != nil { + return nil, fmt.Errorf("resolve base config dir for daemon ownership lock: %w", err) + } + lock, err := AcquireOwnership(baseDir, OwnerInfo{ + PID: os.Getpid(), + HealthPort: d.cfg.HealthPort, + DaemonID: d.cfg.DaemonID, + Version: d.cfg.CLIVersion, + Profile: d.cfg.Profile, + StartedAt: time.Now(), + }) + if err != nil { + return nil, err + } + d.logger.Info("acquired daemon ownership lock", "path", OwnershipLockPath(baseDir)) + return lock, nil +} + +// recordServerVersion stores the server_version reported in a register +// response. Only a non-empty value overwrites, so a later response from a +// server that omits the field (or a different, older backend) never clobbers a +// version we already learned. +func (d *Daemon) recordServerVersion(v string) { + if strings.TrimSpace(v) == "" { + return + } + d.serverVersionMu.Lock() + d.serverVersion = strings.TrimSpace(v) + d.serverVersionMu.Unlock() +} + +// serverVersionSeen returns the last non-empty server_version learned at +// register, or "" if the server never reported one. +func (d *Daemon) serverVersionSeen() string { + d.serverVersionMu.Lock() + defer d.serverVersionMu.Unlock() + return d.serverVersion +} + +// checkServerCompatibility fails loudly when the daemon is talking to a server +// too old to serve the routes it depends on (VWO-365). It runs once after the +// initial registration in preflight, so a skew is caught before any task is +// claimed — not discovered later as a stream of silent /prepare-lease 404s. An +// unreported/unparseable/dev server version is treated as "unknown": we warn +// and lean on the per-route capability backstop rather than refuse to start and +// false-reject a server that may well be compatible. +func (d *Daemon) checkServerCompatibility() error { + seen := d.serverVersionSeen() + switch err := agent.CheckMinServerVersion(seen); { + case errors.Is(err, agent.ErrServerVersionTooOld): + return fmt.Errorf( + "incompatible Multica server: this daemon (version %s) requires server >= %s but the server reports %s; upgrade the server, or downgrade/upgrade the daemon so client and server match", + d.cfg.CLIVersion, agent.MinServerVersion, seen, + ) + case errors.Is(err, agent.ErrServerVersionUnknown): + d.logger.Warn("server did not report a version at registration; cannot pre-verify compatibility (a too-old server will still be caught loudly the first time a task needs an unsupported route)", + "min_server_version", agent.MinServerVersion) + return nil + default: + d.logger.Debug("server version compatible", "server_version", seen, "min_server_version", agent.MinServerVersion) + return nil + } +} + // deregisterRuntimes notifies the server that all runtimes are going offline. func (d *Daemon) deregisterRuntimes() { runtimeIDs := d.allRuntimeIDs() @@ -1192,6 +1305,7 @@ func (d *Daemon) registerRuntimesForWorkspace(ctx context.Context, workspaceID s if err != nil { return nil, "", fmt.Errorf("register runtimes: %w", err) } + d.recordServerVersion(resp.ServerVersion) if len(resp.Runtimes) == 0 && len(failedProfiles) == 0 { return nil, "", fmt.Errorf("register runtimes: empty response") } @@ -3678,6 +3792,18 @@ func (d *Daemon) startTaskPrepareLeaseExtender(ctx context.Context, task Task, t err := d.client.ExtendTaskPrepareLease(reqCtx, task.RuntimeID, task.ID) reqCancel() if err != nil { + if isPrepareLeaseUnsupported(err) { + // The /prepare-lease route is missing: this daemon is + // newer than the server. Say so loudly ONCE and stop the + // loop rather than emit an identical Warn every 15s — + // "do not rely on repeated 404s" (VWO-365). The server + // version gate in Run already fails startup when the + // server reports a too-old version; this is the backstop + // for a server old enough to predate that handshake. + taskLog.Error("stopping task prepare-lease refresh: the server does not support the /prepare-lease route, so this daemon is incompatible with it — upgrade the server to at least "+agent.MinServerVersion+" (or run a daemon matching the server version)", + "error", err) + return + } taskLog.Warn("extend task prepare lease failed", "error", err) } } diff --git a/server/internal/daemon/ownership.go b/server/internal/daemon/ownership.go new file mode 100644 index 0000000000..9e33c0b0e1 --- /dev/null +++ b/server/internal/daemon/ownership.go @@ -0,0 +1,243 @@ +package daemon + +import ( + "encoding/json" + "fmt" + "os" + "path/filepath" + "strings" + "time" +) + +// ownershipLockFileName is the machine-global file whose advisory lock decides +// which daemon process owns this host. It lives directly under the base config +// directory (`~/.multica`), NOT under a profile subdirectory, so every daemon +// on the machine — the CLI-spawned one (default profile) and the Desktop-spawned +// one (a `--profile desktop-` daemon) — resolves the SAME path and +// contends for the SAME lock. That is the invariant the per-profile health-port +// and PID-file guards miss: they key on the profile, so two profiles never see +// each other and both start, each holding an independent in-process +// local-directory mutex that lets two agents write the same checkout at once +// (VWO-364 / VWO-365). +const ownershipLockFileName = "daemon.lock" + +// OwnerInfo is the diagnostic record the owning daemon writes into the lock +// file body. It is advisory only — the OS advisory lock on the file, not this +// content, is what enforces ownership. It exists so a would-be second daemon +// (and `daemon stop/restart --takeover`) can name the incumbent and reach its +// health port for a graceful takeover. +type OwnerInfo struct { + PID int `json:"pid"` + HealthPort int `json:"health_port"` + DaemonID string `json:"daemon_id"` + Version string `json:"version"` + Profile string `json:"profile,omitempty"` + StartedAt time.Time `json:"started_at"` +} + +// OwnershipLock is a held machine-level daemon ownership lock. The advisory +// lock is held for the lifetime of the open file handle: the kernel releases +// it automatically when the process exits (clean or crashed), so there is no +// TTL to tune and no stale-lock window — a crashed owner's lock is gone the +// instant it dies and the next daemon acquires immediately, with no manual +// database or file surgery. Call Release on clean shutdown. +type OwnershipLock struct { + path string + f *os.File +} + +// OwnershipConflict is returned by AcquireOwnership when another live daemon +// already holds the machine lock. Incumbent carries whatever could be read +// from the lock body (best-effort; HasInfo is false when it was empty or +// unreadable, e.g. an owner that has not finished writing it yet). +type OwnershipConflict struct { + Path string + Incumbent OwnerInfo + HasInfo bool +} + +func (e *OwnershipConflict) Error() string { + if e.HasInfo { + // The stop hint must target the OWNER's profile: a bare `multica daemon + // stop` only checks the default profile's health port, which is exactly + // the cross-profile blindness this lock exists to catch. + stopHint := "`multica daemon stop`" + if e.Incumbent.Profile != "" { + stopHint = fmt.Sprintf("`multica --profile %s daemon stop`", e.Incumbent.Profile) + } + return fmt.Sprintf( + "another Multica daemon already owns this machine (pid %d, version %s, profile %q, health port %d, since %s) via lock %s; stop it with %s or take over with `multica daemon start --takeover`", + e.Incumbent.PID, orUnknown(e.Incumbent.Version), e.Incumbent.Profile, e.Incumbent.HealthPort, + e.Incumbent.StartedAt.Format(time.RFC3339), e.Path, stopHint, + ) + } + return fmt.Sprintf( + "another Multica daemon already owns this machine via lock %s (owner details unavailable); stop it with `multica daemon stop` (add --profile if it runs under a named profile) or take over with `multica daemon start --takeover`", + e.Path, + ) +} + +func orUnknown(s string) string { + if s == "" { + return "unknown" + } + return s +} + +// OwnershipLockPath returns the machine-global daemon lock path for baseDir. +// baseDir is the base config directory (cli.ProfileDir("") == ~/.multica), +// shared by every profile on the machine. +func OwnershipLockPath(baseDir string) string { + return filepath.Join(baseDir, ownershipLockFileName) +} + +// AcquireOwnership takes the machine-global daemon ownership lock at +// baseDir/daemon.lock WITHOUT blocking, and records self in the lock body on +// success. It is the single authoritative ownership mechanism: acquire it +// before a daemon binds its health port, registers runtimes, or claims any +// task, so a second process never advertises the same identity or runtime IDs +// while an owner is live. +// +// Returns: +// - (*OwnershipLock, nil) when this process is now the owner. +// - (nil, *OwnershipConflict) when another live daemon holds it. The error +// names the incumbent so the caller can print an actionable message. +// - (nil, err) on an I/O failure opening or locking the file. +// +// The lock file is intentionally NOT removed on release: a persistent inode +// avoids the classic flock+unlink race (unlinking a locked file lets a racing +// acquirer create a fresh inode and both "win"). Stale content from a prior +// owner is harmless — the OS lock state, not the body, is authoritative, and +// the body is truncated and rewritten on every successful acquire. +func AcquireOwnership(baseDir string, self OwnerInfo) (*OwnershipLock, error) { + if baseDir == "" { + return nil, fmt.Errorf("acquire daemon ownership: empty base directory") + } + if err := os.MkdirAll(baseDir, 0o755); err != nil { + return nil, fmt.Errorf("create config dir for daemon lock: %w", err) + } + path := OwnershipLockPath(baseDir) + f, err := os.OpenFile(path, os.O_RDWR|os.O_CREATE, 0o644) + if err != nil { + return nil, fmt.Errorf("open daemon lock %s: %w", path, err) + } + locked, err := tryLockExclusive(f) + if err != nil { + f.Close() + return nil, fmt.Errorf("lock daemon lock %s: %w", path, err) + } + if !locked { + info, ok := readOwnerInfo(path) + f.Close() + return nil, &OwnershipConflict{Path: path, Incumbent: info, HasInfo: ok} + } + if err := writeOwnerInfo(f, self); err != nil { + _ = unlock(f) + f.Close() + return nil, fmt.Errorf("record daemon lock owner: %w", err) + } + return &OwnershipLock{path: path, f: f}, nil +} + +// Release drops the advisory lock and closes the handle. Safe to call on a nil +// receiver or an already-released lock. The lock file itself is left in place +// (see AcquireOwnership). +func (l *OwnershipLock) Release() error { + if l == nil || l.f == nil { + return nil + } + unlockErr := unlock(l.f) + closeErr := l.f.Close() + l.f = nil + if unlockErr != nil { + return unlockErr + } + return closeErr +} + +// writeOwnerInfo truncates the lock file and writes self as JSON. The caller +// holds the exclusive advisory lock, so this write is uncontended. +func writeOwnerInfo(f *os.File, self OwnerInfo) error { + data, err := json.Marshal(self) + if err != nil { + return err + } + if err := f.Truncate(0); err != nil { + return err + } + if _, err := f.Seek(0, 0); err != nil { + return err + } + if _, err := f.Write(data); err != nil { + return err + } + return f.Sync() +} + +// readOwnerInfo reads the lock body without taking the lock. It is used on the +// conflict path (to name the incumbent) and by the takeover path (to reach the +// incumbent's health port). Returns (info, false) when the file is missing, +// empty, or unparseable — e.g. a brand-new owner that locked but has not yet +// written its body. +func readOwnerInfo(path string) (OwnerInfo, bool) { + data, err := os.ReadFile(path) + if err != nil || len(data) == 0 { + return OwnerInfo{}, false + } + var info OwnerInfo + if err := json.Unmarshal(data, &info); err != nil { + return OwnerInfo{}, false + } + return info, true +} + +// ReadOwnerInfo exposes the lock body to callers outside the daemon package +// (the CLI takeover path). Returns ok=false when there is no readable owner. +func ReadOwnerInfo(baseDir string) (OwnerInfo, bool) { + return readOwnerInfo(OwnershipLockPath(baseDir)) +} + +// ProbeOwnership reports whether the machine ownership lock is currently free, +// WITHOUT recording a new owner. When held, it returns the incumbent (ok=true +// when the body was readable). It is used by the CLI launcher to give a fast, +// actionable conflict message (or drive a takeover) before spawning the real +// daemon, which acquires the lock authoritatively in Run. +// +// "free" is an instantaneous observation: a probe that finds the lock free does +// not reserve it, so the caller must still let the spawned daemon acquire it. +func ProbeOwnership(baseDir string) (free bool, incumbent OwnerInfo, ok bool, err error) { + if baseDir == "" { + return false, OwnerInfo{}, false, fmt.Errorf("probe daemon ownership: empty base directory") + } + if err := os.MkdirAll(baseDir, 0o755); err != nil { + return false, OwnerInfo{}, false, fmt.Errorf("create config dir for daemon lock: %w", err) + } + path := OwnershipLockPath(baseDir) + f, err := os.OpenFile(path, os.O_RDWR|os.O_CREATE, 0o644) + if err != nil { + return false, OwnerInfo{}, false, fmt.Errorf("open daemon lock %s: %w", path, err) + } + locked, err := tryLockExclusive(f) + if err != nil { + f.Close() + return false, OwnerInfo{}, false, fmt.Errorf("probe daemon lock %s: %w", path, err) + } + if !locked { + info, has := readOwnerInfo(path) + f.Close() + return false, info, has, nil + } + _ = unlock(f) + f.Close() + return true, OwnerInfo{}, false, nil +} + +// OwnershipBypassed reports whether the operator disabled the single-owner +// guard via MULTICA_DAEMON_ALLOW_MULTIPLE. This is the documented break-glass / +// rollback for the ownership lock, and the escape hatch for a deliberate +// multi-backend setup that knowingly accepts the shared-checkout risk. Any +// value other than empty / "0" / "false" enables the bypass. +func OwnershipBypassed() bool { + v := strings.TrimSpace(os.Getenv("MULTICA_DAEMON_ALLOW_MULTIPLE")) + return v != "" && v != "0" && !strings.EqualFold(v, "false") +} diff --git a/server/internal/daemon/ownership_test.go b/server/internal/daemon/ownership_test.go new file mode 100644 index 0000000000..6d505fced7 --- /dev/null +++ b/server/internal/daemon/ownership_test.go @@ -0,0 +1,285 @@ +package daemon + +import ( + "encoding/json" + "errors" + "os" + "os/exec" + "path/filepath" + "strconv" + "strings" + "testing" + "time" +) + +// TestAcquireOwnership_SecondFailsWhileHeld covers simultaneous startup / +// duplicate identity: a second acquirer of the same base dir is rejected while +// the first holds the lock. flock treats two independent opens of the same +// file as contending even within one process (flock(2)), so this is a faithful +// stand-in for two daemon processes racing the same machine lock. +func TestAcquireOwnership_SecondFailsWhileHeld(t *testing.T) { + baseDir := t.TempDir() + + first, err := AcquireOwnership(baseDir, OwnerInfo{PID: 4242, HealthPort: 19514, DaemonID: "d-1", Version: "0.4.2", Profile: "", StartedAt: time.Now()}) + if err != nil { + t.Fatalf("first acquire: %v", err) + } + defer first.Release() + + _, err = AcquireOwnership(baseDir, OwnerInfo{PID: 5555, HealthPort: 19999, DaemonID: "d-2", Version: "0.4.2", Profile: "desktop-host", StartedAt: time.Now()}) + var conflict *OwnershipConflict + if !errors.As(err, &conflict) { + t.Fatalf("second acquire: want *OwnershipConflict, got %v", err) + } + if !conflict.HasInfo || conflict.Incumbent.PID != 4242 { + t.Fatalf("conflict should name the incumbent (pid 4242): %+v", conflict.Incumbent) + } + // The message must be actionable — name the stop and takeover remedies. + msg := conflict.Error() + for _, want := range []string{"daemon stop", "--takeover", baseDir} { + if !strings.Contains(msg, want) { + t.Fatalf("conflict message missing %q: %s", want, msg) + } + } +} + +// TestOwnershipConflict_ProfileAwareStopHint pins that the stop remedy targets +// the OWNER's profile: a bare `multica daemon stop` only reaches the default +// profile's daemon, which is exactly the cross-profile blindness the lock +// exists to catch. +func TestOwnershipConflict_ProfileAwareStopHint(t *testing.T) { + withProfile := &OwnershipConflict{ + Path: "/tmp/x/daemon.lock", + Incumbent: OwnerInfo{PID: 1, Profile: "desktop-host", HealthPort: 20000}, + HasInfo: true, + } + if !strings.Contains(withProfile.Error(), "`multica --profile desktop-host daemon stop`") { + t.Fatalf("named-profile incumbent must get a profile-scoped stop hint: %s", withProfile.Error()) + } + + defaultProfile := &OwnershipConflict{ + Path: "/tmp/x/daemon.lock", + Incumbent: OwnerInfo{PID: 1, Profile: "", HealthPort: 19514}, + HasInfo: true, + } + msg := defaultProfile.Error() + if !strings.Contains(msg, "`multica daemon stop`") || strings.Contains(msg, "--profile ") { + t.Fatalf("default-profile incumbent must get the plain stop hint: %s", msg) + } +} + +// TestAcquireOwnership_ReleaseAllowsReacquire covers clean shutdown: after the +// owner releases, the next daemon acquires immediately. +func TestAcquireOwnership_ReleaseAllowsReacquire(t *testing.T) { + baseDir := t.TempDir() + + first, err := AcquireOwnership(baseDir, OwnerInfo{PID: 1, Version: "0.4.2", StartedAt: time.Now()}) + if err != nil { + t.Fatalf("first acquire: %v", err) + } + if err := first.Release(); err != nil { + t.Fatalf("release: %v", err) + } + // Release is idempotent / nil-safe. + if err := first.Release(); err != nil { + t.Fatalf("second release should be a no-op: %v", err) + } + + second, err := AcquireOwnership(baseDir, OwnerInfo{PID: 2, Version: "0.4.2", StartedAt: time.Now()}) + if err != nil { + t.Fatalf("reacquire after clean release: %v", err) + } + defer second.Release() + + // The body should now describe the new owner. + info, ok := ReadOwnerInfo(baseDir) + if !ok || info.PID != 2 { + t.Fatalf("lock body should name the new owner (pid 2): ok=%v info=%+v", ok, info) + } +} + +// TestAcquireOwnership_StaleLockFileReclaimed covers stale-lease recovery: a +// leftover lock file from a dead owner (content present, no live lock) must not +// block a new daemon, and requires no manual file surgery. +func TestAcquireOwnership_StaleLockFileReclaimed(t *testing.T) { + baseDir := t.TempDir() + + stale := OwnerInfo{PID: 999999, HealthPort: 19514, DaemonID: "dead", Version: "0.3.0", StartedAt: time.Now().Add(-time.Hour)} + data, _ := json.Marshal(stale) + if err := os.WriteFile(OwnershipLockPath(baseDir), data, 0o644); err != nil { + t.Fatalf("seed stale lock file: %v", err) + } + + lock, err := AcquireOwnership(baseDir, OwnerInfo{PID: os.Getpid(), Version: "0.4.2", StartedAt: time.Now()}) + if err != nil { + t.Fatalf("stale leftover lock file must not block acquire: %v", err) + } + defer lock.Release() + + info, ok := ReadOwnerInfo(baseDir) + if !ok || info.PID != os.Getpid() { + t.Fatalf("acquire should overwrite the stale body with our own: ok=%v info=%+v", ok, info) + } +} + +// TestProbeOwnership_FreeAndHeld covers the launcher probe used to give a fast +// conflict message (and drive takeover) without recording a new owner. +func TestProbeOwnership_FreeAndHeld(t *testing.T) { + baseDir := t.TempDir() + + free, _, _, err := ProbeOwnership(baseDir) + if err != nil { + t.Fatalf("probe (free): %v", err) + } + if !free { + t.Fatal("probe should report a fresh base dir as free") + } + // Probing must NOT have recorded an owner. + if _, ok := ReadOwnerInfo(baseDir); ok { + t.Fatal("probe must not write an owner record") + } + + lock, err := AcquireOwnership(baseDir, OwnerInfo{PID: 7, HealthPort: 20200, Version: "0.4.2", StartedAt: time.Now()}) + if err != nil { + t.Fatalf("acquire: %v", err) + } + defer lock.Release() + + free, incumbent, ok, err := ProbeOwnership(baseDir) + if err != nil { + t.Fatalf("probe (held): %v", err) + } + if free { + t.Fatal("probe should report a held lock as not free") + } + if !ok || incumbent.PID != 7 || incumbent.HealthPort != 20200 { + t.Fatalf("probe should surface the incumbent (pid 7, port 20200): ok=%v %+v", ok, incumbent) + } +} + +// TestOwnershipBypassed covers the break-glass / rollback env parsing. +func TestOwnershipBypassed(t *testing.T) { + cases := []struct { + val string + set bool + want bool + }{ + {set: false, want: false}, + {val: "", set: true, want: false}, + {val: "0", set: true, want: false}, + {val: "false", set: true, want: false}, + {val: "FALSE", set: true, want: false}, + {val: "1", set: true, want: true}, + {val: "true", set: true, want: true}, + {val: "yes", set: true, want: true}, + } + for _, tc := range cases { + if tc.set { + t.Setenv("MULTICA_DAEMON_ALLOW_MULTIPLE", tc.val) + } else { + os.Unsetenv("MULTICA_DAEMON_ALLOW_MULTIPLE") + } + if got := OwnershipBypassed(); got != tc.want { + t.Fatalf("OwnershipBypassed(set=%v val=%q)=%v, want %v", tc.set, tc.val, got, tc.want) + } + } +} + +// TestAcquireOwnership_CrashRecovery covers crash recovery across REAL OS +// processes: a helper process acquires the lock, is SIGKILLed (no clean +// Release runs), and the parent must then acquire immediately — proving the +// kernel drops the advisory lock on process death with no stale window and no +// manual cleanup. This subprocess test is also the end-to-end cross-process +// reproduction (a two-goroutine test cannot distinguish the fixed lock from the +// old process-local mutex). It is fully isolated: a temp base dir, no health +// port, no server. +func TestAcquireOwnership_CrashRecovery(t *testing.T) { + baseDir := t.TempDir() + flag := filepath.Join(baseDir, "acquired.flag") + + helper := exec.Command(os.Args[0], "-test.run=TestOwnershipHelperProcess") + helper.Env = append(os.Environ(), + "MULTICA_OWNERSHIP_HELPER=1", + "MULTICA_OWNERSHIP_BASEDIR="+baseDir, + ) + helper.Stdout = os.Stderr // fold helper test output into the parent log + helper.Stderr = os.Stderr + if err := helper.Start(); err != nil { + t.Fatalf("start helper: %v", err) + } + t.Cleanup(func() { + _ = helper.Process.Kill() + _, _ = helper.Process.Wait() + }) + + // Wait for the helper to signal it holds the lock. + if !waitForFile(flag, 10*time.Second) { + t.Fatal("helper did not acquire the lock within 10s") + } + + // While the helper holds it, the parent cannot acquire. + if _, err := AcquireOwnership(baseDir, OwnerInfo{PID: os.Getpid(), Version: "0.4.2", StartedAt: time.Now()}); err == nil { + t.Fatal("parent acquired while helper still held the lock") + } + + // Crash the helper (SIGKILL — no defer, no Release). + if err := helper.Process.Kill(); err != nil { + t.Fatalf("kill helper: %v", err) + } + _, _ = helper.Process.Wait() + + // The OS should have released the lock on death; the parent acquires, + // retrying briefly to absorb kernel reap latency. + var ( + lock *OwnershipLock + err error + ) + deadline := time.Now().Add(10 * time.Second) + for time.Now().Before(deadline) { + lock, err = AcquireOwnership(baseDir, OwnerInfo{PID: os.Getpid(), Version: "0.4.2", StartedAt: time.Now()}) + if err == nil { + break + } + time.Sleep(50 * time.Millisecond) + } + if err != nil { + t.Fatalf("crash of the owner should have freed the lock: %v", err) + } + _ = lock.Release() +} + +// TestOwnershipHelperProcess is the child-process entrypoint for +// TestAcquireOwnership_CrashRecovery. It runs only when invoked with the helper +// env var set; otherwise it is a no-op so the normal suite does not spawn +// anything. +func TestOwnershipHelperProcess(t *testing.T) { + if os.Getenv("MULTICA_OWNERSHIP_HELPER") != "1" { + t.Skip("child-process entrypoint; not run directly") + } + baseDir := os.Getenv("MULTICA_OWNERSHIP_BASEDIR") + hp, _ := strconv.Atoi(os.Getenv("MULTICA_OWNERSHIP_HEALTHPORT")) + lock, err := AcquireOwnership(baseDir, OwnerInfo{ + PID: os.Getpid(), HealthPort: hp, DaemonID: "helper", Version: "9.9.9", StartedAt: time.Now(), + }) + if err != nil { + // Signal failure by leaving the flag absent; parent times out. + os.Exit(3) + } + // Signal "acquired" via a sentinel file the parent polls for. + _ = os.WriteFile(filepath.Join(baseDir, "acquired.flag"), []byte("ok"), 0o644) + // Hold until killed, with a safety cap so a leaked helper self-exits. + time.Sleep(60 * time.Second) + _ = lock.Release() + os.Exit(0) +} + +func waitForFile(path string, timeout time.Duration) bool { + deadline := time.Now().Add(timeout) + for time.Now().Before(deadline) { + if _, err := os.Stat(path); err == nil { + return true + } + time.Sleep(20 * time.Millisecond) + } + return false +} diff --git a/server/internal/daemon/ownership_unix.go b/server/internal/daemon/ownership_unix.go new file mode 100644 index 0000000000..85e8b1fea5 --- /dev/null +++ b/server/internal/daemon/ownership_unix.go @@ -0,0 +1,35 @@ +//go:build !windows + +package daemon + +import ( + "errors" + "os" + + "golang.org/x/sys/unix" +) + +// tryLockExclusive takes a non-blocking exclusive advisory lock on the whole +// file via flock(2). flock is inherited across fork and, crucially, released +// by the kernel when the last descriptor closes — including on process death — +// so a crashed owner never leaves a lock behind. flock is purely advisory +// among flock callers and does not block read(2)/write(2), so readOwnerInfo can +// still read the body while another process holds the lock. +// +// Returns (true, nil) on success, (false, nil) when another process holds it +// (EWOULDBLOCK), and (false, err) on any other failure. +func tryLockExclusive(f *os.File) (bool, error) { + err := unix.Flock(int(f.Fd()), unix.LOCK_EX|unix.LOCK_NB) + if err == nil { + return true, nil + } + if errors.Is(err, unix.EWOULDBLOCK) { + return false, nil + } + return false, err +} + +// unlock releases the flock held on f. +func unlock(f *os.File) error { + return unix.Flock(int(f.Fd()), unix.LOCK_UN) +} diff --git a/server/internal/daemon/ownership_windows.go b/server/internal/daemon/ownership_windows.go new file mode 100644 index 0000000000..b565c161e6 --- /dev/null +++ b/server/internal/daemon/ownership_windows.go @@ -0,0 +1,46 @@ +//go:build windows + +package daemon + +import ( + "errors" + "os" + + "golang.org/x/sys/windows" +) + +// lockRegionOffsetHigh places the single locked byte at a very high file +// offset (0x40000000_00000000) that no lock-file content ever reaches, so the +// byte-range lock acts as a pure ownership token and never blocks another +// process from reading the JSON body in the low bytes (readOwnerInfo). This +// mirrors the whole-file, read-transparent semantics flock gives on Unix. +const lockRegionOffsetHigh = 0x40000000 + +// tryLockExclusive takes a non-blocking exclusive byte-range lock via +// LockFileEx. Windows releases LockFileEx locks when the handle closes or the +// process dies, so — like flock — a crashed owner leaves no lingering lock. +// +// Returns (true, nil) on success, (false, nil) when another process holds it +// (ERROR_LOCK_VIOLATION, the immediate-fail result of LOCKFILE_FAIL_IMMEDIATELY), +// and (false, err) on any other failure. +func tryLockExclusive(f *os.File) (bool, error) { + ol := &windows.Overlapped{OffsetHigh: lockRegionOffsetHigh} + err := windows.LockFileEx( + windows.Handle(f.Fd()), + windows.LOCKFILE_EXCLUSIVE_LOCK|windows.LOCKFILE_FAIL_IMMEDIATELY, + 0, 1, 0, ol, + ) + if err == nil { + return true, nil + } + if errors.Is(err, windows.ERROR_LOCK_VIOLATION) { + return false, nil + } + return false, err +} + +// unlock releases the byte-range lock held on f. +func unlock(f *os.File) error { + ol := &windows.Overlapped{OffsetHigh: lockRegionOffsetHigh} + return windows.UnlockFileEx(windows.Handle(f.Fd()), 0, 1, 0, ol) +} diff --git a/server/internal/daemon/version_gate_test.go b/server/internal/daemon/version_gate_test.go new file mode 100644 index 0000000000..7e9ccb2099 --- /dev/null +++ b/server/internal/daemon/version_gate_test.go @@ -0,0 +1,190 @@ +package daemon + +import ( + "context" + "io" + "log/slog" + "net/http" + "net/http/httptest" + "strings" + "sync/atomic" + "testing" + "time" +) + +// TestServerCompatibilityGate covers version mismatch at startup: a server that +// reports a version below MinServerVersion fails loudly; a compatible or +// unreported version does not. +func TestServerCompatibilityGate(t *testing.T) { + newDaemon := func() *Daemon { + return &Daemon{ + logger: slog.New(slog.NewTextHandler(io.Discard, nil)), + cfg: Config{CLIVersion: "0.4.2"}, + } + } + + t.Run("too old fails loudly", func(t *testing.T) { + d := newDaemon() + d.recordServerVersion("0.3.22") + err := d.checkServerCompatibility() + if err == nil { + t.Fatal("a 0.3.22 server must fail the compatibility gate") + } + for _, want := range []string{"incompatible", "0.3.22"} { + if !strings.Contains(err.Error(), want) { + t.Fatalf("error missing %q: %v", want, err) + } + } + }) + + t.Run("compatible passes", func(t *testing.T) { + d := newDaemon() + d.recordServerVersion("0.4.2") + if err := d.checkServerCompatibility(); err != nil { + t.Fatalf("a matching server must pass: %v", err) + } + }) + + t.Run("unreported version warns but proceeds", func(t *testing.T) { + d := newDaemon() + // No recordServerVersion call: server never reported a version. + if err := d.checkServerCompatibility(); err != nil { + t.Fatalf("an unreported server version must not fail startup: %v", err) + } + }) + + t.Run("empty does not clobber a known version", func(t *testing.T) { + d := newDaemon() + d.recordServerVersion("0.4.2") + d.recordServerVersion("") // e.g. a later response from an older backend + if got := d.serverVersionSeen(); got != "0.4.2" { + t.Fatalf("empty must not clobber a known version, got %q", got) + } + }) +} + +// TestPrepareLeaseExtender_UnsupportedRouteStopsLoudly covers the version-skew +// backstop: when /prepare-lease 404s because the route is missing on an +// un-upgraded server, the extender logs once and STOPS instead of Warn-looping +// forever. +func TestPrepareLeaseExtender_UnsupportedRouteStopsLoudly(t *testing.T) { + oldRefresh := taskPrepareLeaseRefresh + oldTimeout := taskPrepareLeaseTimeout + taskPrepareLeaseRefresh = 10 * time.Millisecond + taskPrepareLeaseTimeout = 500 * time.Millisecond + t.Cleanup(func() { + taskPrepareLeaseRefresh = oldRefresh + taskPrepareLeaseTimeout = oldTimeout + }) + + var calls atomic.Int32 + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if strings.HasSuffix(r.URL.Path, "/prepare-lease") { + calls.Add(1) + // Route missing on an old server: a bare 404, NOT "task not found". + http.Error(w, "404 page not found", http.StatusNotFound) + return + } + w.WriteHeader(http.StatusOK) + })) + t.Cleanup(srv.Close) + + d := &Daemon{ + client: NewClient(srv.URL), + logger: slog.New(slog.NewTextHandler(io.Discard, nil)), + } + task := Task{ID: "task-1", RuntimeID: "rt-1"} + taskLog := slog.New(slog.NewTextHandler(io.Discard, nil)) + + stop := d.startTaskPrepareLeaseExtender(context.Background(), task, taskLog) + t.Cleanup(stop) + + // Let many refresh intervals elapse. If the loop stopped after the first + // 404 (correct), the server sees exactly one call; if it kept looping (the + // bug), it would see ~30. + time.Sleep(300 * time.Millisecond) + if got := calls.Load(); got != 1 { + t.Fatalf("prepare-lease extender should stop after the first unsupported-route 404; got %d calls", got) + } +} + +// TestPrepareLeaseExtender_Handler404KeepsRetrying guards the classifier's +// positive matching: a HANDLER-level 404 (JSON {"error": ...} body — a deleted +// row, or a transient DB error the server's access checks map to "not found") +// must NOT be mistaken for a missing route. Stopping the extender on one such +// blip would lapse the lease mid-task and let the server re-dispatch a second +// execution — the duplicate-writer class this change exists to prevent. +func TestPrepareLeaseExtender_Handler404KeepsRetrying(t *testing.T) { + oldRefresh := taskPrepareLeaseRefresh + oldTimeout := taskPrepareLeaseTimeout + taskPrepareLeaseRefresh = 10 * time.Millisecond + taskPrepareLeaseTimeout = 500 * time.Millisecond + t.Cleanup(func() { + taskPrepareLeaseRefresh = oldRefresh + taskPrepareLeaseTimeout = oldTimeout + }) + + var calls atomic.Int32 + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if strings.HasSuffix(r.URL.Path, "/prepare-lease") { + calls.Add(1) + // Handler-shaped 404: what writeError produces for a gone row or a + // transient access-check failure. NOT chi's plain catch-all. + w.Header().Set("Content-Type", "application/json") + w.WriteHeader(http.StatusNotFound) + w.Write([]byte(`{"error":"not found"}`)) + return + } + w.WriteHeader(http.StatusOK) + })) + t.Cleanup(srv.Close) + + d := &Daemon{ + client: NewClient(srv.URL), + logger: slog.New(slog.NewTextHandler(io.Discard, nil)), + } + stop := d.startTaskPrepareLeaseExtender(context.Background(), Task{ID: "task-1", RuntimeID: "rt-1"}, slog.New(slog.NewTextHandler(io.Discard, nil))) + t.Cleanup(stop) + + time.Sleep(150 * time.Millisecond) + if got := calls.Load(); got < 2 { + t.Fatalf("a handler-level 404 must keep the loop retrying; got only %d calls", got) + } +} + +// TestPrepareLeaseExtender_TransientErrorKeepsRetrying guards that the backstop +// only fires on a missing route: an ordinary transient failure must NOT stop +// the loop. +func TestPrepareLeaseExtender_TransientErrorKeepsRetrying(t *testing.T) { + oldRefresh := taskPrepareLeaseRefresh + oldTimeout := taskPrepareLeaseTimeout + taskPrepareLeaseRefresh = 10 * time.Millisecond + taskPrepareLeaseTimeout = 500 * time.Millisecond + t.Cleanup(func() { + taskPrepareLeaseRefresh = oldRefresh + taskPrepareLeaseTimeout = oldTimeout + }) + + var calls atomic.Int32 + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if strings.HasSuffix(r.URL.Path, "/prepare-lease") { + calls.Add(1) + http.Error(w, "boom", http.StatusInternalServerError) // transient 500 + return + } + w.WriteHeader(http.StatusOK) + })) + t.Cleanup(srv.Close) + + d := &Daemon{ + client: NewClient(srv.URL), + logger: slog.New(slog.NewTextHandler(io.Discard, nil)), + } + stop := d.startTaskPrepareLeaseExtender(context.Background(), Task{ID: "task-1", RuntimeID: "rt-1"}, slog.New(slog.NewTextHandler(io.Discard, nil))) + t.Cleanup(stop) + + time.Sleep(150 * time.Millisecond) + if got := calls.Load(); got < 2 { + t.Fatalf("a transient error must keep the loop retrying; got only %d calls", got) + } +} diff --git a/server/internal/handler/daemon.go b/server/internal/handler/daemon.go index 40b91211be..edd6fe6ee4 100644 --- a/server/internal/handler/daemon.go +++ b/server/internal/handler/daemon.go @@ -658,6 +658,16 @@ func (h *Handler) DaemonRegister(w http.ResponseWriter, r *http.Request) { "repos": repoResp.Repos, "repos_version": repoResp.ReposVersion, "settings": repoResp.Settings, + // The running server build version, so the daemon can detect a + // client/server version skew loudly at registration instead of + // 404-looping later (e.g. a newer daemon calling the /prepare-lease + // route an older server lacks — VWO-364/VWO-365). Always populated for + // daemons (this is the internal daemon channel, distinct from the + // user-facing /api/config which omits it on official cloud). Empty only + // on a server built without a version stamp or one that predates this + // field — the daemon treats that as "unknown" and relies on the + // per-route capability backstop rather than failing. + "server_version": h.cfg.ServerVersion, }) } diff --git a/server/pkg/agent/version.go b/server/pkg/agent/version.go index aa3eeb221b..801c68076a 100644 --- a/server/pkg/agent/version.go +++ b/server/pkg/agent/version.go @@ -66,6 +66,57 @@ var ( ErrCLIVersionTooOld = errors.New("multica CLI version is below required minimum") ) +// MinServerVersion is the lowest Multica server version this daemon build is +// compatible with. v0.3.28 is the first server release carrying the +// /api/daemon/runtimes/{id}/tasks/{id}/prepare-lease route the daemon's task +// lease extender depends on; against an older server that call 404-loops +// silently (VWO-364/VWO-365). The daemon compares the server_version reported +// in the register response against this and fails loudly on a +// parseable-but-older server, rather than discovering the skew as a stream of +// 404s once tasks run. +const MinServerVersion = "0.3.28" + +// Errors returned by CheckMinServerVersion, mirroring the CLI-version taxonomy. +// ErrServerVersionUnknown is deliberately NON-fatal to the caller: a server +// that reports no version (built without a stamp, or one predating the +// register-response server_version field) may still be route-compatible, so the +// daemon warns and relies on the per-route capability backstop instead of +// refusing to start and false-rejecting a working server. +var ( + ErrServerVersionUnknown = errors.New("multica server version not reported") + ErrServerVersionTooOld = errors.New("multica server version is below required minimum") +) + +// CheckMinServerVersion returns nil when `serverVersion` parses as ≥ +// MinServerVersion. Returns ErrServerVersionUnknown for empty or unparsable +// input (caller should warn, not fail), and ErrServerVersionTooOld when +// parsable but below the minimum (caller should fail loudly). Dev-built servers +// (git-describe shape) always pass, matching CheckMinCLIVersion so a `make dev` +// server is never gated. +func CheckMinServerVersion(serverVersion string) error { + v := strings.TrimSpace(serverVersion) + if v == "" { + return ErrServerVersionUnknown + } + if devDescribeRe.MatchString(v) { + return nil + } + parsed, err := parseSemver(v) + if err != nil { + return ErrServerVersionUnknown + } + min, err := parseSemver(MinServerVersion) + if err != nil { + // Misconfiguration in the constant itself — do not hard-fail the daemon + // over our own typo; treat as unknown so the backstop still guards. + return ErrServerVersionUnknown + } + if parsed.lessThan(min) { + return ErrServerVersionTooOld + } + return nil +} + // devDescribeRe matches the `git describe --tags --always --dirty` output for // a build past the latest tag, e.g. `v0.2.15-235-gdaf0e935` (optionally with a // trailing `-dirty`). Daemons built from source (Makefile `make build` / `make diff --git a/server/pkg/agent/version_test.go b/server/pkg/agent/version_test.go index c78b908b48..13280271aa 100644 --- a/server/pkg/agent/version_test.go +++ b/server/pkg/agent/version_test.go @@ -79,6 +79,33 @@ func TestCheckMinCLIVersion(t *testing.T) { } } +func TestCheckMinServerVersion(t *testing.T) { + tests := []struct { + name string + input string + wantErr error + }{ + {"tagged release at minimum", "0.3.28", nil}, + {"tagged release above minimum", "0.4.2", nil}, + {"v-prefixed above minimum", "v0.4.2", nil}, + {"below minimum (the VWO-364 skew)", "0.3.22", ErrServerVersionTooOld}, + {"well below minimum", "v0.3.0", ErrServerVersionTooOld}, + {"empty (server predates the field / no stamp)", "", ErrServerVersionUnknown}, + {"unparsable", "not-a-version", ErrServerVersionUnknown}, + {"git-describe dev build passes", "v0.4.2-5-gabc1234", nil}, + {"git-describe dirty dev build passes", "v0.4.2-5-gabc1234-dirty", nil}, + } + for _, tt := range tests { + err := CheckMinServerVersion(tt.input) + if tt.wantErr == nil && err != nil { + t.Errorf("%s: CheckMinServerVersion(%q) = %v, want nil", tt.name, tt.input, err) + } + if tt.wantErr != nil && !errors.Is(err, tt.wantErr) { + t.Errorf("%s: CheckMinServerVersion(%q) = %v, want %v", tt.name, tt.input, err, tt.wantErr) + } + } +} + func TestExtractVersionLine(t *testing.T) { tests := []struct { name string