fix(server): drain fully before the self-update/restore restart handoff (#1380)

* feat(server): supervisor detection and server.restart_mode config key

RunningUnderSupervisor detects systemd (INVOCATION_ID) and, best-effort,
NSSM (NSSM_SERVICE_NAME — 2.24 does not set it, so NSSM deployments set
the mode explicitly). server.restart_mode (auto|spawn|supervised, default
auto, env OWNCORD_SERVER_RESTART_MODE) selects how a self-restart hands
off after the server drains: exit for the supervisor to relaunch, or
spawn the replacement directly.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01Ngzj2Rx9UGC35uLHAfErMp

* fix(server): make the self-restart handoff drain fully before starting the successor

The update/restore/wizard restart previously spawned the replacement
while the old server was still serving, then SIGTERMed itself and
hard-exited after 10s. That design failed in every documented deployment
mode: under the shipped systemd unit the spawned child (same cgroup) was
killed when the old main process exited and Restart=on-failure never
relaunched a clean exit; on Windows the self-SIGTERM is unsupported and
silently dropped, so graceful shutdown never ran — hub.GracefulStop (the
only caller of LiveKitProcess.Stop) was skipped, orphaning livekit-server
on TCP 7880/UDP 50000-60000 and dropping queued event/audit batches; and
NSSM's relaunch raced the self-spawned replacement for the database lock.

Admin handlers now perform only the on-disk swap and request a restart
through an injected hook (admin.SetRestartHandoff). The main package's
restart coordinator cancels the parent of run()'s signal.NotifyContext —
the exact drain a SIGTERM triggers, on every platform — and after run()
has fully torn down (listeners closed, hub and LiveKit stopped, queues
flushed, DB closed and its lock released) main() performs the handoff:
spawn the replacement in spawn mode, or exit 0 for the supervisor in
supervised mode. A 90s backstop force-exits a wedged teardown; the
DB-lock and bind retries demote to safety nets.

A three-state guard (idle/busy/restart-pending) serializes update apply,
backup restore, and setup-wizard restarts against each other: concurrent
applies no longer race the same staged .new file or broadcast a spurious
update_aborted, and conflicting requests get 409 UPDATE_IN_PROGRESS /
RESTART_PENDING. The swap being free of process side effects also makes
the apply success path unit-testable for the first time.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01Ngzj2Rx9UGC35uLHAfErMp

* fix(server): errno-based bind-conflict detection, ACME bind retry, LiveKit Pdeathsig

isAddrInUse now unwraps to the platform errno (EADDRINUSE; WSAEADDRINUSE
10048 on Windows) with the English strings kept only as fallback — the
string-only match never fired on localized Windows, silently disabling
the bind retry. The retry loop is extracted into serveWithBindRetry and
now also covers the ACME :80 challenge server, which previously gave up
on first conflict and stayed dead (breaking HTTP-01 renewals) until the
next restart. The .old-binary boot cleanup retries briefly for the
window where a spawn-mode predecessor has not fully exited. The
companion livekit-server gets Pdeathsig SIGKILL on Linux so a parent
killed without teardown (kill -9, OOM, backstop exit) cannot orphan it
with the voice ports held.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01Ngzj2Rx9UGC35uLHAfErMp

* docs(deploy): Restart=always unit and per-supervisor restart-mode guidance

Restart=always is what lets the deliberate clean exit after a
self-update/restore relaunch under systemd (systemctl stop is never
auto-restarted; failure exits behave as before). Deployment docs gain
the required NSSM AppEnvironmentExtra line, the Task Scheduler and
Docker restart-policy notes, and the new drain-then-handoff update flow.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01Ngzj2Rx9UGC35uLHAfErMp

---------

Co-authored-by: Claude <noreply@anthropic.com>
This commit is contained in:
J3vb
2026-08-16 08:25:40 +02:00
committed by GitHub
co-authored by Claude Fable 5
parent a366160dc8
commit 6a26f2a839
29 changed files with 1540 additions and 171 deletions
+21
View File
@@ -0,0 +1,21 @@
package main
import "strings"
// isAddrInUse reports whether err is an "address already in use" bind
// failure. The errno check (platform files: addrinuse_unix.go /
// addrinuse_windows.go) is authoritative; the string checks remain as a
// fallback for errors that arrive with the errno wrapped away. String
// matching alone was the original implementation and silently disabled the
// bind retry on non-English Windows, where the system error text is
// localized.
func isAddrInUse(err error) bool {
if err == nil {
return false
}
if errnoIsAddrInUse(err) {
return true
}
return strings.Contains(err.Error(), "address already in use") ||
strings.Contains(err.Error(), "Only one usage of each socket address")
}
+114
View File
@@ -0,0 +1,114 @@
package main
import (
"errors"
"fmt"
"io"
"log/slog"
"net"
"net/http"
"testing"
"time"
)
// A real double-bind must be recognized through the errno chain
// (net.OpError → os.SyscallError → Errno) — this is what keeps the bind
// retry alive on non-English Windows, where the old string-only match never
// fired against localized error text.
func TestIsAddrInUse_RealBindConflict(t *testing.T) {
l, err := net.Listen("tcp", "127.0.0.1:0")
if err != nil {
t.Fatalf("listen: %v", err)
}
defer l.Close() //nolint:errcheck
_, err = net.Listen("tcp", l.Addr().String())
if err == nil {
t.Fatal("second listen on the same address unexpectedly succeeded")
}
if !isAddrInUse(err) {
t.Errorf("isAddrInUse(%v) = false, want true for a real bind conflict", err)
}
}
func TestIsAddrInUse_Table(t *testing.T) {
cases := []struct {
name string
err error
want bool
}{
{"nil", nil, false},
{"unrelated error", errors.New("connection refused"), false},
{"unix string fallback", errors.New("listen tcp :8443: bind: address already in use"), true},
{"windows string fallback", errors.New("listen tcp :8443: bind: Only one usage of each socket address (protocol/network address/port) is normally permitted."), true},
{"out of range port", fmt.Errorf("listen tcp: address 99999: invalid port"), false},
}
for _, tc := range cases {
t.Run(tc.name, func(t *testing.T) {
if got := isAddrInUse(tc.err); got != tc.want {
t.Errorf("isAddrInUse(%v) = %v, want %v", tc.err, got, tc.want)
}
})
}
}
func TestServeWithBindRetry(t *testing.T) {
log := slog.New(slog.NewTextHandler(io.Discard, nil))
addrInUse := errors.New("listen tcp :1: bind: address already in use")
prevEvery := bindRetryEvery
bindRetryEvery = time.Millisecond
t.Cleanup(func() { bindRetryEvery = prevEvery })
t.Run("retries through a transient conflict", func(t *testing.T) {
calls := 0
err := serveWithBindRetry(log, "test", func() error {
calls++
if calls < 3 {
return addrInUse
}
return http.ErrServerClosed
})
if !errors.Is(err, http.ErrServerClosed) {
t.Errorf("err = %v, want ErrServerClosed after the conflict clears", err)
}
if calls != 3 {
t.Errorf("serve calls = %d, want 3", calls)
}
})
t.Run("gives up on a non-conflict error immediately", func(t *testing.T) {
calls := 0
otherErr := errors.New("listen tcp: address 99999: invalid port")
err := serveWithBindRetry(log, "test", func() error {
calls++
return otherErr
})
if !errors.Is(err, otherErr) {
t.Errorf("err = %v, want the serve error passed through", err)
}
if calls != 1 {
t.Errorf("serve calls = %d, want 1 (no retry for non-conflict errors)", calls)
}
})
t.Run("bounded attempts on a persistent conflict", func(t *testing.T) {
calls := 0
err := serveWithBindRetry(log, "test", func() error {
calls++
return addrInUse
})
if !errors.Is(err, addrInUse) {
t.Errorf("err = %v, want the final conflict error", err)
}
if calls != 20 {
t.Errorf("serve calls = %d, want exactly 20", calls)
}
})
t.Run("clean shutdown passes through untouched", func(t *testing.T) {
if err := serveWithBindRetry(log, "test", func() error { return http.ErrServerClosed }); !errors.Is(err, http.ErrServerClosed) {
t.Errorf("err = %v, want ErrServerClosed", err)
}
})
}
+14
View File
@@ -0,0 +1,14 @@
//go:build !windows
package main
import (
"errors"
"syscall"
)
// errnoIsAddrInUse unwraps err (net.OpError → os.SyscallError → Errno) and
// compares against EADDRINUSE.
func errnoIsAddrInUse(err error) bool {
return errors.Is(err, syscall.EADDRINUSE)
}
+21
View File
@@ -0,0 +1,21 @@
//go:build windows
package main
import (
"errors"
"syscall"
)
// wsaeAddrInUse is WSAEADDRINUSE (10048), the errno a Windows bind conflict
// actually carries. The standard syscall package does not export it (it
// lives in golang.org/x/sys/windows, which this module does not depend on
// directly), and syscall.EADDRINUSE on Windows is Go's distinct
// APPLICATION_ERROR-block value that errors.Is does not map to WSA codes.
const wsaeAddrInUse = syscall.Errno(10048)
// errnoIsAddrInUse unwraps err (net.OpError → os.SyscallError → Errno) and
// compares against both Windows bind-conflict errnos.
func errnoIsAddrInUse(err error) bool {
return errors.Is(err, wsaeAddrInUse) || errors.Is(err, syscall.EADDRINUSE)
}
+82 -6
View File
@@ -2,6 +2,7 @@ package admin
import (
"errors"
"sync"
"sync/atomic"
"time"
@@ -64,16 +65,65 @@ func StubCloseError(msg string) (restore func()) {
}
}
// ApplyStagedUpdate exposes applyStagedUpdate (the on-disk swap + respawn
// logic behind POST /updates/apply's background goroutine) so tests can drive
// its abort paths directly with fake filesystem paths, instead of exercising
// the full HTTP handler — which resolves exePath via os.Executable() and
// would rename/replace the running test binary itself.
// ApplyStagedUpdate exposes applyStagedUpdate (the on-disk swap behind
// POST /updates/apply's background goroutine) so tests can drive it directly
// with fake filesystem paths, instead of exercising the full HTTP handler —
// which resolves exePath via os.Executable() and would rename/replace the
// running test binary itself. The swap has no process side effects anymore
// (no spawn, no signal, no exit), so the success path is testable too.
var ApplyStagedUpdate = applyStagedUpdate
// ApplyAndRestart exposes the whole background tail (countdown broadcast →
// swap → restart request / guard release) for tests, combined with
// StubRestart and SetApplyRestartDelay.
var ApplyAndRestart = applyAndRestart
// SetApplyRestartDelay shrinks the client-facing restart countdown so tests
// of applyAndRestart don't sleep the real 5 seconds.
func SetApplyRestartDelay(d time.Duration) (restore func()) {
prev := applyRestartDelay
applyRestartDelay = d
return func() { applyRestartDelay = prev }
}
// ResetRestartState returns the restart-serialization guard to idle. Tests
// that drive a restart-committing path (restore, apply success, setup
// restart) must call it afterwards — the state is process-global and would
// otherwise 409 every later test's request.
func ResetRestartState() { restartState.Store(restartStateIdle) }
// ForceRestartState pins the guard for conflict-path tests: busy=false sets
// restart-pending, busy=true sets an in-flight exclusive operation.
func ForceRestartState(busy bool) {
if busy {
restartState.Store(restartStateBusy)
} else {
restartState.Store(restartStatePending)
}
}
// CurrentRestartState reports the guard state by name for assertions.
func CurrentRestartState() string {
switch restartState.Load() {
case restartStateBusy:
return "busy"
case restartStatePending:
return "pending"
default:
return "idle"
}
}
// RequestRestartForTest exposes requestRestart so the unwired default hook's
// inertness (log, no exit, no spawn) is directly testable.
var RequestRestartForTest = requestRestart
// StubRestart replaces the process-restart hook for the duration of a test and
// returns a func reporting whether a restart was requested. Without this the
// restore handler would respawn and os.Exit the test binary.
// restore handler's request would hit the unwired-hook error log; with it the
// test can assert the request happened. The restore func also returns the
// restart-serialization guard to idle, since any test that triggered the hook
// has necessarily left it in restart-pending.
func StubRestart() (restarted func() bool, restore func()) {
restartMu.Lock()
prev := restartSelf
@@ -84,5 +134,31 @@ func StubRestart() (restarted func() bool, restore func()) {
restartMu.Lock()
restartSelf = prev
restartMu.Unlock()
ResetRestartState()
}
}
// StubRestartCapture is StubRestart recording the reasons passed to the hook,
// for tests that assert which path requested the restart.
func StubRestartCapture() (reasons func() []string, restore func()) {
restartMu.Lock()
prev := restartSelf
var mu sync.Mutex
var got []string
restartSelf = func(reason string) {
mu.Lock()
got = append(got, reason)
mu.Unlock()
}
restartMu.Unlock()
return func() []string {
mu.Lock()
defer mu.Unlock()
return append([]string(nil), got...)
}, func() {
restartMu.Lock()
restartSelf = prev
restartMu.Unlock()
ResetRestartState()
}
}
+19 -61
View File
@@ -13,19 +13,8 @@ import (
"sync"
"time"
"syscall"
"github.com/go-chi/chi/v5"
"github.com/owncord/server/db"
"github.com/owncord/server/updater"
)
const (
// restartGraceDelay lets the HTTP response and the server_restart
// broadcast reach clients before the process goes away.
restartGraceDelay = 2 * time.Second
// shutdownGraceDelay is how long SIGTERM gets before the os.Exit backstop.
shutdownGraceDelay = 10 * time.Second
)
// backupBaseDir is the directory for backup files, resolved to an absolute
@@ -211,6 +200,18 @@ func handleRestoreBackup(database *db.DB, hub HubBroadcaster) http.Handler {
return
}
// Serialize against a concurrent update apply (which stages a new
// binary and then restarts) and an already-requested restart — a
// restore must not close the database out from under either. The
// deferred release is a busy→idle CAS, so it covers every failure
// return below and harmlessly no-ops once a branch has promoted the
// state to restart-pending via commitRestartPending.
if !beginRestartSensitiveOp() {
writeRestartConflict(w)
return
}
defer abortRestartSensitiveOp()
if _, err := os.Stat(target); os.IsNotExist(err) { //nolint:gosec // G703: path sanitized by HasPrefix check above
writeErr(w, http.StatusNotFound, "NOT_FOUND", "backup not found")
return
@@ -276,13 +277,14 @@ func handleRestoreBackup(database *db.DB, hub HubBroadcaster) http.Handler {
// database.Close() closes the writer and reader pools regardless of
// the error it returns (Server/db/db.go), so this process cannot
// serve anything more either way — every other failure path below
// (copyFile failing, and the success path itself) respawns for
// (copyFile failing, and the success path itself) restarts for
// exactly that reason. The live database file is still intact here
// (copyFile hasn't run yet), so the respawned process comes back on
// (copyFile hasn't run yet), so the restarted process comes back on
// the pre-restore data rather than leaving clients pinned on
// "Reconnecting..." against a process that never actually restarts.
slog.Error("failed to close database before restore — restarting anyway, DB pools are closed either way", "err", err)
writeErr(w, http.StatusInternalServerError, "INTERNAL_ERROR", "failed to close database")
commitRestartPending()
go requestRestart("backup_restore_close_failed")
return
}
@@ -304,8 +306,9 @@ func handleRestoreBackup(database *db.DB, hub HubBroadcaster) http.Handler {
}
writeErr(w, http.StatusInternalServerError, "INTERNAL_ERROR", msg)
// The database was closed before the copy: this process cannot serve
// anything more either way, so it must respawn exactly as it does on
// anything more either way, so it must restart exactly as it does on
// the success path.
commitRestartPending()
go requestRestart("backup_restore_failed")
return
}
@@ -321,19 +324,12 @@ func handleRestoreBackup(database *db.DB, hub HubBroadcaster) http.Handler {
// this process can serve nothing more. It used to stop here, leaving a
// live server answering every request against a closed DB while the
// response and the restart broadcast both claimed a restart was
// happening. Respawn for real, the same way applying an update does.
// happening. Restart for real, the same way applying an update does.
commitRestartPending()
go requestRestart("backup_restore")
})
}
// restartSelf is the process-restart hook, swappable in tests (which must not
// respawn or exit the test binary). Guarded because the swap happens on the
// test goroutine while the restore handler reads it from its own.
var (
restartMu sync.Mutex
restartSelf = restartProcess
)
// dbCloser is swappable in tests to simulate database.Close() returning an
// error. modernc.org/sqlite's sqlite3_close_v2 essentially never fails on a
// normally-open connection, so there is no portable way to provoke a genuine
@@ -353,44 +349,6 @@ func closeDatabase(database *db.DB) error {
return fn(database)
}
// requestRestart invokes the current restart hook.
func requestRestart(reason string) {
restartMu.Lock()
fn := restartSelf
restartMu.Unlock()
fn(reason)
}
// restartProcess spawns a fresh copy of this server and shuts the current one
// down. Mirrors the update-apply path (update_handlers.go): SIGTERM first so
// main.go's graceful shutdown runs, os.Exit as the backstop.
func restartProcess(reason string) {
// Give the HTTP response and the restart broadcast a moment to flush.
time.Sleep(restartGraceDelay)
exePath, err := os.Executable()
if err != nil {
slog.Error("restart: cannot determine executable path — manual restart required",
"reason", reason, "error", err)
return
}
if resolved, symErr := filepath.EvalSymlinks(exePath); symErr == nil {
exePath = resolved
}
if err := updater.SpawnDetached(exePath, os.Args[1:]); err != nil {
slog.Error("restart: spawning the replacement process failed — manual restart required",
"reason", reason, "error", err)
return
}
slog.Info("restart: replacement process spawned, shutting down", "reason", reason)
if p, findErr := os.FindProcess(os.Getpid()); findErr == nil {
_ = p.Signal(syscall.SIGTERM)
time.Sleep(shutdownGraceDelay)
}
os.Exit(0) //nolint:gocritic // backstop if the SIGTERM handler didn't exit
}
// copyBackupFile is the restore path's file-copy hook. It exists as a var so
// tests can inject the hard-to-simulate mid-copy failure (truncate-then-fail)
// the rollback branch exists for; production never swaps it.
+112
View File
@@ -0,0 +1,112 @@
package admin
import (
"log/slog"
"net/http"
"sync"
"sync/atomic"
)
// The admin panel has three actions that end in a process restart: applying a
// server update, restoring a database backup, and finishing the setup wizard
// with listener-affecting changes. None of them stop or spawn processes
// directly — they request a restart through the hook below, and the main
// package performs the actual work: trigger the same graceful drain a SIGTERM
// would (which also works on Windows, where a process cannot signal itself),
// and only after run() has fully torn down either spawn the replacement
// binary or exit for the process supervisor to relaunch, per
// server.restart_mode (see Server/restart.go).
// restartSelf is the process-restart request hook. Swapped by
// SetRestartHandoff at startup and by tests (StubRestart); the default logs
// loudly instead of exiting, so a mis-wired binary degrades to "restart
// manually" rather than to a silent no-op or a test-killing os.Exit.
var (
restartMu sync.Mutex
restartSelf = restartUnwired
)
func restartUnwired(reason string) {
slog.Error("restart requested but no restart coordinator is wired — restart the server manually",
"reason", reason)
}
// SetRestartHandoff wires restart requests to the main package's restart
// coordinator. Call once at startup, before the router serves (main.go, next
// to SetDatabasePath).
func SetRestartHandoff(fn func(reason string)) {
restartMu.Lock()
restartSelf = fn
restartMu.Unlock()
}
// requestRestart invokes the current restart hook.
func requestRestart(reason string) {
restartMu.Lock()
fn := restartSelf
restartMu.Unlock()
fn(reason)
}
// restartState serializes the restart-ending admin actions against each
// other. Two problems it closes: concurrent update applies raced the same
// staged .new file (each download deletes the other's staged file, and the
// loser broadcast a spurious update_aborted to every client while a restart
// was actually happening), and a restore could tear the database down under
// an apply that had already responded 200.
//
// idle ──beginRestartSensitiveOp──▶ busy ──commitRestartPending──▶ pending
// ▲ │
// └────abortRestartSensitiveOp─────┘
//
// pending is terminal: it means a restart request has been (or is about to
// be) issued and only the process replacement clears it. Ownership of the
// busy state transfers to the background goroutine that finishes the work —
// the HTTP handler responds while the state is still busy, so it must NOT
// defer a reset.
const (
restartStateIdle int32 = 0
restartStateBusy int32 = 1 // an update apply or restore is in flight
restartStatePending int32 = 2 // a restart has been requested
)
var restartState atomic.Int32
// beginRestartSensitiveOp claims the exclusive restart-sensitive slot.
// Callers that fail any later step must release it with
// abortRestartSensitiveOp; callers that reach the point of no return promote
// it with commitRestartPending.
func beginRestartSensitiveOp() bool {
return restartState.CompareAndSwap(restartStateIdle, restartStateBusy)
}
// abortRestartSensitiveOp releases the slot after a failed operation. CAS,
// not a blind store: an abort must never demote an already-pending restart.
func abortRestartSensitiveOp() {
restartState.CompareAndSwap(restartStateBusy, restartStateIdle)
}
// commitRestartPending marks the process as committed to restarting.
func commitRestartPending() {
restartState.Store(restartStatePending)
}
// tryDirectRestartPending is commitRestartPending for paths with no failable
// work between claiming the slot and requesting the restart (setup wizard):
// idle → pending in one step. Reports whether the claim won.
func tryDirectRestartPending() bool {
return restartState.CompareAndSwap(restartStateIdle, restartStatePending)
}
// writeRestartConflict answers a request that lost to an in-flight
// restart-sensitive operation, distinguishing "busy, try again shortly" from
// "the process is about to be replaced".
func writeRestartConflict(w http.ResponseWriter) {
if restartState.Load() == restartStatePending {
writeErr(w, http.StatusConflict, "RESTART_PENDING",
"the server is restarting — retry after it comes back")
return
}
writeErr(w, http.StatusConflict, "UPDATE_IN_PROGRESS",
"another update or restore is already in progress")
}
+263
View File
@@ -0,0 +1,263 @@
package admin_test
import (
"context"
"crypto/sha256"
"encoding/hex"
"encoding/json"
"net/http"
"os"
"path/filepath"
"testing"
"time"
"github.com/owncord/server/admin"
"github.com/owncord/server/updater"
)
// ─── The restart handoff: swap success paths and the serialization guard ────
//
// applyStagedUpdate no longer spawns, signals, or exits — the restart goes
// through the coordinator hook (admin.SetRestartHandoff) and the main package
// performs the drain + handoff. That is what makes the success paths below
// testable at all, and the three-state guard (idle → busy → restart-pending)
// is what keeps concurrent applies/restores/restarts from racing each other.
// stageFakeUpdate lays out a fake current binary and a staged .new whose
// hash matches, returning the three paths plus the staged hash.
func stageFakeUpdate(t *testing.T) (exePath, oldPath, newPath, stagedHash string) {
t.Helper()
dir := t.TempDir()
exePath = filepath.Join(dir, "chatserver")
oldPath = exePath + ".old"
newPath = exePath + ".new"
if err := os.WriteFile(exePath, []byte("old binary"), 0o755); err != nil {
t.Fatalf("writing fake exe: %v", err)
}
staged := []byte("verified staged bytes")
if err := os.WriteFile(newPath, staged, 0o755); err != nil {
t.Fatalf("writing staged binary: %v", err)
}
sum := sha256.Sum256(staged)
return exePath, oldPath, newPath, hex.EncodeToString(sum[:])
}
// The success path: the verified staged binary ends up at exePath, the
// previous binary at .old, no corrective broadcast is sent, and the swap
// reports committed.
func TestApplyStagedUpdate_Success_SwapsWithoutAbortBroadcast(t *testing.T) {
exePath, oldPath, newPath, stagedHash := stageFakeUpdate(t)
hub := &mockHub{}
if !admin.ApplyStagedUpdate(hub, exePath, oldPath, newPath, stagedHash) {
t.Fatal("ApplyStagedUpdate = false, want committed swap")
}
if len(hub.restartCalls) != 0 {
t.Errorf("restartCalls = %+v, want none (no corrective broadcast on success)", hub.restartCalls)
}
if got, err := os.ReadFile(exePath); err != nil || string(got) != "verified staged bytes" {
t.Errorf("exePath contents = %q, err=%v; want the staged bytes", got, err)
}
if got, err := os.ReadFile(oldPath); err != nil || string(got) != "old binary" {
t.Errorf(".old contents = %q, err=%v; want the previous binary", got, err)
}
if _, err := os.Stat(newPath); !os.IsNotExist(err) {
t.Errorf(".new still exists after commit (stat err=%v)", err)
}
}
// The full background tail on success: countdown broadcast, swap, guard
// promoted to restart-pending, restart requested through the hook with
// reason "update".
func TestApplyAndRestart_Success_RequestsRestartAndMarksPending(t *testing.T) {
exePath, oldPath, newPath, stagedHash := stageFakeUpdate(t)
admin.ResetRestartState()
admin.ForceRestartState(true) // the handler claims busy before spawning the goroutine
reasons, restoreHook := admin.StubRestartCapture()
defer restoreHook()
defer admin.SetApplyRestartDelay(time.Millisecond)()
hub := &mockHub{}
admin.ApplyAndRestart(hub, exePath, oldPath, newPath, stagedHash)
if len(hub.restartCalls) != 1 || hub.restartCalls[0].reason != "update" {
t.Fatalf("restartCalls = %+v, want exactly the update countdown", hub.restartCalls)
}
if got := reasons(); len(got) != 1 || got[0] != "update" {
t.Errorf("restart hook reasons = %v, want [update]", got)
}
if got := admin.CurrentRestartState(); got != "pending" {
t.Errorf("restart state after committed swap = %q, want pending", got)
}
}
// The background tail on a failed swap: corrective broadcast (covered in
// depth by the OC-0226 tests), no restart request, and the busy slot released
// so a corrected release can be applied without a manual restart.
func TestApplyAndRestart_Abort_ReleasesGuard(t *testing.T) {
dir := t.TempDir()
exePath := filepath.Join(dir, "chatserver")
oldPath := exePath + ".old"
newPath := exePath + ".new" // never written → re-verification fails
admin.ResetRestartState()
admin.ForceRestartState(true)
reasons, restoreHook := admin.StubRestartCapture()
defer restoreHook()
defer admin.SetApplyRestartDelay(time.Millisecond)()
hub := &mockHub{}
admin.ApplyAndRestart(hub, exePath, oldPath, newPath,
"0000000000000000000000000000000000000000000000000000000000000000")
if got := reasons(); len(got) != 0 {
t.Errorf("restart hook reasons = %v, want none on abort", got)
}
if got := admin.CurrentRestartState(); got != "idle" {
t.Errorf("restart state after aborted swap = %q, want idle (slot released)", got)
}
if len(hub.restartCalls) != 2 || hub.restartCalls[1].reason != "update_aborted" {
t.Errorf("restartCalls = %+v, want countdown then update_aborted", hub.restartCalls)
}
}
// POST /updates/apply answers 409 without touching the updater when a restart
// is already pending or another restart-sensitive operation is in flight.
// The guard sits before CheckForUpdate, so no GitHub call is attempted — the
// updater here has no reachable base URL and would error loudly if consulted.
func TestApplyUpdate_Conflict409(t *testing.T) {
t.Setenv("OWNCORD_CONTAINER", "0")
database := openAdminTestDB(t)
u := updater.NewUpdater("1.0.0", "", "J3vb", "OwnCord")
handler := admin.NewAdminAPI(database, "1.0.0", nil, u, nil, nil, nil, newTestModService(database), newTestRoleService(database))
token := createAdminUser(t, database)
cases := []struct {
name string
busy bool
wantCode string
}{
{"restart pending", false, "RESTART_PENDING"},
{"apply or restore in flight", true, "UPDATE_IN_PROGRESS"},
}
for _, tc := range cases {
t.Run(tc.name, func(t *testing.T) {
admin.ForceRestartState(tc.busy)
t.Cleanup(admin.ResetRestartState)
w := doRequest(t, handler, http.MethodPost, "/updates/apply", token, nil)
if w.Code != http.StatusConflict {
t.Fatalf("status = %d, want 409; body: %s", w.Code, w.Body.String())
}
var resp map[string]string
if err := json.Unmarshal(w.Body.Bytes(), &resp); err != nil {
t.Fatalf("unmarshal: %v", err)
}
if resp["error"] != tc.wantCode {
t.Errorf("error code = %q, want %q", resp["error"], tc.wantCode)
}
})
}
}
// POST /backups/{name}/restore refuses the same way — before any disk I/O.
func TestRestoreBackup_Conflict409(t *testing.T) {
database := openAdminTestDB(t)
handler := admin.NewAdminAPI(database, "1.0.0", &mockHub{}, nil, nil, nil, nil, newTestModService(database), newTestRoleService(database))
token := createAdminUser(t, database)
admin.ForceRestartState(false) // restart pending
t.Cleanup(admin.ResetRestartState)
w := doRequest(t, handler, http.MethodPost, "/backups/chatserver_x.db/restore", token, nil)
if w.Code != http.StatusConflict {
t.Fatalf("status = %d, want 409; body: %s", w.Code, w.Body.String())
}
var resp map[string]string
if err := json.Unmarshal(w.Body.Bytes(), &resp); err != nil {
t.Fatalf("unmarshal: %v", err)
}
if resp["error"] != "RESTART_PENDING" {
t.Errorf("error code = %q, want RESTART_PENDING", resp["error"])
}
}
// A restore whose database.Close() fails is still committed to dying: it must
// request a restart AND leave the guard in restart-pending so nothing else
// starts an update against a process with closed DB pools.
func TestRestore_CloseFailure_StillMarksPendingAndRequestsRestart(t *testing.T) {
tmpDir := chdirTemp(t)
database := openAdminTestDB(t)
handler := admin.NewAdminAPI(database, "1.0.0", &mockHub{}, nil, nil, nil, nil, newTestModService(database), newTestRoleService(database))
token := createAdminUser(t, database)
backupDir := filepath.Join(tmpDir, "data", "backups")
if err := os.MkdirAll(backupDir, 0o750); err != nil {
t.Fatalf("MkdirAll backups: %v", err)
}
backupName := "chatserver_20240101_120000.db"
if err := database.BackupToSafe(context.Background(), filepath.Join(backupDir, backupName), backupDir); err != nil {
t.Fatalf("BackupToSafe fixture: %v", err)
}
reasons, restoreHook := admin.StubRestartCapture()
defer restoreHook()
restoreClose := admin.StubCloseError("injected close failure")
defer restoreClose()
w := doRequest(t, handler, http.MethodPost, "/backups/"+backupName+"/restore", token, nil)
if w.Code != http.StatusInternalServerError {
t.Fatalf("status = %d, want 500; body: %s", w.Code, w.Body.String())
}
deadline := time.Now().Add(2 * time.Second)
for len(reasons()) == 0 && time.Now().Before(deadline) {
time.Sleep(10 * time.Millisecond)
}
if got := reasons(); len(got) != 1 || got[0] != "backup_restore_close_failed" {
t.Errorf("restart hook reasons = %v, want [backup_restore_close_failed]", got)
}
if got := admin.CurrentRestartState(); got != "pending" {
t.Errorf("restart state = %q, want pending (process is committed to dying)", got)
}
}
// A setup-wizard restart is skipped (with the response already written) when
// an update or restore already owns the restart: two restart paths must never
// race each other's teardown.
func TestSetupRestart_SkippedWhenPending(t *testing.T) {
database := openAdminTestDB(t)
cfgPath := filepath.Join(t.TempDir(), "config.yaml")
restarted := make(chan string, 1)
handler := wizardHandler(t, database, cfgPath, restarted)
admin.ForceRestartState(false) // restart pending
rr := doRequest(t, handler, "POST", "/setup", "", map[string]any{
"username": "owner",
"password": "SecurePass123!",
"wizard": map[string]any{
"server_name": "S",
"port": 9000,
"tls_mode": "off",
},
})
if rr.Code != http.StatusCreated {
t.Fatalf("POST /setup = %d, want 201; body=%s", rr.Code, rr.Body.String())
}
select {
case reason := <-restarted:
t.Errorf("setup requested restart %q despite a pending restart", reason)
case <-time.After(200 * time.Millisecond):
}
}
// The unwired default hook must degrade to a loud log — never exit, spawn,
// or panic — so a binary that misses the SetRestartHandoff wiring (or a test
// that forgets to stub) fails soft.
func TestRequestRestart_UnwiredDefaultIsInert(t *testing.T) {
admin.RequestRestartForTest("unwired-test")
}
+9 -2
View File
@@ -254,9 +254,16 @@ func handleSetup(database *db.DB, limiter *auth.RateLimiter, allowedOrigins []st
// Restart after the response is written so the browser receives the
// token and the reconnect URL. Mirrors handleRestoreBackup /
// handleApplyUpdate: broadcast, then respawn in a goroutine
// (requestRestart sleeps a grace delay before acting).
// handleApplyUpdate: broadcast, then request the restart in a
// goroutine — main.go drains the server and performs the handoff.
// tryDirectRestartPending loses only to an already in-flight update
// or restore, which will itself restart the process; skipping is
// correct then (the response above is already written either way).
if restartRequired {
if !tryDirectRestartPending() {
slog.Warn("setup restart skipped: another restart-sensitive operation is already in progress")
return
}
if hub != nil {
hub.BroadcastServerRestart("setup", restartBroadcastDelaySeconds)
}
+5 -1
View File
@@ -33,9 +33,13 @@ func wizardRunningCfg() *config.Config {
}
// wizardHandler builds the admin API with wizard options and a restart stub
// that signals restarted (buffered) instead of respawning the process.
// that signals restarted (buffered) instead of restarting the process. A
// wizard run that triggers a restart leaves the process-global
// restart-serialization guard in restart-pending, so it is reset after every
// wizard test.
func wizardHandler(t *testing.T, database *db.DB, cfgPath string, restarted chan string) http.Handler {
t.Helper()
t.Cleanup(admin.ResetRestartState)
return admin.NewAdminAPI(database, "1.0.0", nil, nil, nil, nil, nil, newTestModService(database), newTestRoleService(database),
admin.SetupOptions{
ConfigPath: cfgPath,
+70 -55
View File
@@ -6,7 +6,6 @@ import (
"net/http"
"os"
"path/filepath"
"syscall"
"time"
"github.com/owncord/server/updater"
@@ -35,6 +34,11 @@ func handleCheckUpdate(u *updater.Updater) http.HandlerFunc {
}
}
// applyRestartDelay is how long the "restarting in 5s" countdown broadcast to
// clients actually gets before the swap + restart request. A var so tests can
// shrink it; the broadcast countdown below stays 5 to match this value.
var applyRestartDelay = 5 * time.Second
// handleApplyUpdate downloads and applies a server update.
func handleApplyUpdate(u *updater.Updater, hub HubBroadcaster, _ string) http.Handler {
return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
@@ -52,6 +56,24 @@ func handleApplyUpdate(u *updater.Updater, hub HubBroadcaster, _ string) http.Ha
return
}
// Serialize against concurrent applies, restores, and an
// already-requested restart — claimed before any updater work so a
// pending restart answers 409 without an outbound GitHub call, and
// so two concurrent applies can never both stage into the same .new
// path (each download removes the other's staged file). The deferred
// release covers every early return; ownership transfers to the
// applyAndRestart goroutine at the bottom.
if !beginRestartSensitiveOp() {
writeRestartConflict(w)
return
}
claimed := true
defer func() {
if claimed {
abortRestartSensitiveOp()
}
}()
// Check for available update.
info, err := u.CheckForUpdate(r.Context())
if err != nil {
@@ -94,7 +116,7 @@ func handleApplyUpdate(u *updater.Updater, hub HubBroadcaster, _ string) http.Ha
// DownloadAndVerify stages the binary and returns its trusted hash
// (bound to the signed release manifest). The apply goroutine below
// re-verifies the staged file against this hash through an open
// handle — never by path — before the rename+spawn.
// handle — never by path — before the rename.
stagedHash, err := u.DownloadAndVerify(ctx, info.Latest, info.DownloadURL, info.ChecksumURL, info.SignatureURL, info.ManifestURL, info.ManifestSignatureURL, newPath)
if err != nil {
slog.Error("update download/verify failed", "err", err)
@@ -108,41 +130,54 @@ func handleApplyUpdate(u *updater.Updater, hub HubBroadcaster, _ string) http.Ha
"version": info.Latest,
})
// Broadcast restart notification and apply in background.
go func() {
if hub != nil {
hub.BroadcastServerRestart("update", 5)
}
time.Sleep(5 * time.Second)
if applyStagedUpdate(hub, exePath, oldPath, newPath, stagedHash) {
// Every deferred cleanup inside applyStagedUpdate has run by
// now, which is why the exit lives out here.
os.Exit(0) // fallback if the SIGTERM handler didn't exit
}
}()
// Broadcast the restart countdown and finish in the background — the
// goroutine takes over the busy state claimed above, so the deferred
// release must stand down.
claimed = false
go applyAndRestart(hub, exePath, oldPath, newPath, stagedHash)
})
}
// applyStagedUpdate performs the on-disk swap (verified staged binary ->
// exePath) and spawns the replacement process. The caller has already
// broadcast "restarting in 5s" to every connected client before invoking
// this, so every return path that does NOT end in a successful respawn must
// correct that promise — otherwise the client's restart banner counts down
// to a permanent "Reconnecting..." over a connection that never actually
// dropped (OC-0226). The deferred broadcast below covers all such paths
// (verification failure, rename failure, commit failure, spawn failure) with
// one guard instead of one broadcast per failure branch; it is cancelled by
// setting restarting=true immediately before the process commits to
// respawning.
// It reports whether the process is committed to exiting for the replacement.
// The exit itself belongs to the caller: calling os.Exit here would skip both
// deferred cleanups below (the staged-file handle and the corrective
// broadcast), and on Windows releasing that handle is the very thing the
// restart is for.
// applyAndRestart is POST /updates/apply's background tail: give clients the
// promised countdown, swap the binary on disk, and hand the process over to
// the main package's restart coordinator. On a failed swap it releases the
// exclusive slot claimed by the handler so a corrected release can be applied
// without a manual restart (the corrective update_aborted broadcast is sent
// by applyStagedUpdate's deferred guard).
func applyAndRestart(hub HubBroadcaster, exePath, oldPath, newPath, stagedHash string) {
if hub != nil {
hub.BroadcastServerRestart("update", 5)
}
time.Sleep(applyRestartDelay)
if applyStagedUpdate(hub, exePath, oldPath, newPath, stagedHash) {
commitRestartPending()
requestRestart("update")
return
}
abortRestartSensitiveOp()
}
// applyStagedUpdate performs the on-disk swap: verified staged binary ->
// exePath, previous binary -> .old. It reports whether the swap committed —
// on true the caller must request a restart, because the file at exePath is
// no longer the binary this process is running.
//
// The caller has already broadcast "restarting in 5s" to every connected
// client before invoking this, so every failure path must correct that
// promise — otherwise the client's restart banner counts down to a permanent
// "Reconnecting..." over a connection that never actually dropped (OC-0226).
// The deferred broadcast below covers all such paths (verification failure,
// rename failure, commit failure) with one guard; it is cancelled by setting
// committed=true once the verified binary is in place.
//
// It does NOT spawn, signal, or exit: the restart itself is the main
// package's job, after run() has fully drained (Server/restart.go). Keeping
// the swap free of process side effects is also what makes the success path
// unit-testable.
func applyStagedUpdate(hub HubBroadcaster, exePath, oldPath, newPath, stagedHash string) bool {
restarting := false
committed := false
defer func() {
if !restarting && hub != nil {
if !committed && hub != nil {
hub.BroadcastServerRestart("update_aborted", 0)
}
}()
@@ -150,7 +185,7 @@ func applyStagedUpdate(hub HubBroadcaster, exePath, oldPath, newPath, stagedHash
// TOCTOU guard: open the staged binary once, verify its hash
// through that handle, and commit (rename) that exact file.
// Commit fails if the path was swapped after verification, so
// the bytes verified are the bytes spawned.
// the bytes verified are the bytes the restart will execute.
staged, err := updater.OpenVerifiedBinary(newPath, stagedHash)
if err != nil {
slog.Error("update: staged binary re-verification failed, aborting update", "error", err)
@@ -176,27 +211,7 @@ func applyStagedUpdate(hub HubBroadcaster, exePath, oldPath, newPath, stagedHash
return false
}
// Spawn new process.
if err := updater.SpawnDetached(exePath, os.Args[1:]); err != nil {
slog.Error("update: spawn new process failed", "error", err)
return false
}
// The replacement process is spawned: from here on this process is
// committed to shutting down for it, so the "restarting" promise made at
// the top of handleApplyUpdate's goroutine is about to come true. Cancel
// the deferred corrective broadcast.
restarting = true
// Signal the process to shut down gracefully before exiting.
// We use SIGTERM on Unix to trigger the graceful shutdown handler
// in main.go. On Windows, os.Exit is unavoidable because the
// process must release its file lock on the binary.
slog.Info("update: new process spawned, shutting down current process")
if p, err := os.FindProcess(os.Getpid()); err == nil {
_ = p.Signal(syscall.SIGTERM)
// Give graceful shutdown a few seconds before force-killing.
time.Sleep(10 * time.Second)
}
committed = true
slog.Info("update: staged binary committed — requesting restart", "path", exePath)
return true
}
+3 -2
View File
@@ -418,8 +418,9 @@ func TestAdminAPI_ApplyUpdate_ContainerOptOut(t *testing.T) {
// "Reconnecting..." state) with no corrective signal ever sent. These tests
// call the swap logic directly — admin.ApplyStagedUpdate — with inputs
// engineered to fail at different points, and assert a corrective
// "update_aborted" broadcast follows. They deliberately do not exercise the
// success path: that ends in os.Exit(0), which would kill the test binary.
// "update_aborted" broadcast follows. The success path is covered too, now
// that the swap has no process side effects (the restart happens through the
// coordinator hook, not an in-package spawn + os.Exit).
// TestApplyStagedUpdate_VerifyFails_BroadcastsAbort covers the earliest abort
// point: the staged binary re-verification (OpenVerifiedBinary) fails
+20 -1
View File
@@ -200,6 +200,19 @@ type ServerConfig struct {
// longer has to be added to the ADMIN allowlist. Empty (default) falls
// back to AdminAllowedCIDRs.
LiveKitWebhookAllowedCIDRs []string `koanf:"livekit_webhook_allowed_cidrs"`
// RestartMode selects how a self-restart (update apply, backup restore,
// setup wizard) hands the process over to its replacement once the server
// has fully drained:
// - "supervised": exit cleanly and rely on the process supervisor
// (systemd Restart=, NSSM AppExit, Docker restart policy) to relaunch.
// - "spawn": start the replacement binary directly before exiting
// (unmanaged deployments: console, Task Scheduler).
// - "auto" (default): "supervised" when a supervisor or container is
// detected (updater.RunningUnderSupervisor / RunningInContainer),
// otherwise "spawn".
// Env override: OWNCORD_SERVER_RESTART_MODE. NSSM deployments must set
// this to "supervised" — NSSM 2.24 is not auto-detectable.
RestartMode string `koanf:"restart_mode"`
}
// MetricsCIDRs returns the effective allowlist for the metrics surfaces.
@@ -290,7 +303,8 @@ func defaults() Config {
"192.168.0.0/16", // private class C
"fc00::/7", // IPv6 unique local
},
WAFCRSMode: "detect",
WAFCRSMode: "detect",
RestartMode: "auto",
},
Database: DatabaseConfig{
Type: "sqlite",
@@ -369,6 +383,11 @@ server:
# waf_paranoia_level: 2 # OWASP CRS paranoia level 1-4
# waf_crs_mode: "detect" # off | detect | block — CRS layer mode; "detect" logs
# # CRS matches without blocking (safe default for chat traffic)
# restart_mode: "auto" # auto | spawn | supervised — how self-restarts (update,
# # restore, setup wizard) hand off. "supervised" exits and
# # lets systemd/NSSM/Docker relaunch; "spawn" starts the
# # replacement directly; "auto" detects (NSSM users: set
# # "supervised" explicitly, NSSM is not auto-detectable)
database:
type: "sqlite" # "sqlite" is the only supported backend
+42
View File
@@ -505,3 +505,45 @@ func TestLoadEnvOverride_EventPersistence(t *testing.T) {
t.Errorf("EventPersistence.RetentionHours = %d, want 48", cfg.EventPersistence.RetentionHours)
}
}
func TestLoadRestartMode(t *testing.T) {
// server.restart_mode drives the self-restart handoff (see main.go's
// resolveRestartMode): default "auto", overridable via YAML and via
// OWNCORD_SERVER_RESTART_MODE — the env case pins envKeyToKoanf's
// server_restart_mode -> server.restart_mode mapping.
t.Run("default", func(t *testing.T) {
cfg, err := config.Load(filepath.Join(t.TempDir(), "config.yaml"))
if err != nil {
t.Fatalf("Load() returned error: %v", err)
}
if cfg.Server.RestartMode != "auto" {
t.Errorf("Server.RestartMode = %q, want %q", cfg.Server.RestartMode, "auto")
}
})
t.Run("yaml override", func(t *testing.T) {
cfgPath := filepath.Join(t.TempDir(), "config.yaml")
yaml := "server:\n restart_mode: \"supervised\"\n"
if err := os.WriteFile(cfgPath, []byte(yaml), 0o644); err != nil {
t.Fatalf("failed to write yaml: %v", err)
}
cfg, err := config.Load(cfgPath)
if err != nil {
t.Fatalf("Load() returned error: %v", err)
}
if cfg.Server.RestartMode != "supervised" {
t.Errorf("Server.RestartMode = %q, want %q", cfg.Server.RestartMode, "supervised")
}
})
t.Run("env override", func(t *testing.T) {
t.Setenv("OWNCORD_SERVER_RESTART_MODE", "spawn")
cfg, err := config.Load(filepath.Join(t.TempDir(), "config.yaml"))
if err != nil {
t.Fatalf("Load() returned error: %v", err)
}
if cfg.Server.RestartMode != "spawn" {
t.Errorf("Server.RestartMode = %q, want %q", cfg.Server.RestartMode, "spawn")
}
})
}
+8 -6
View File
@@ -18,12 +18,14 @@ func lockFilePath(dbPath string) string { return dbPath + ".lock" }
// acquireProcessLock takes the single-process lock for dbPath, retrying for
// a bounded window before giving up with errAlreadyLocked.
//
// The retry exists for the restart handoff: self-update and backup-restore
// spawn the replacement process while the old one is still draining (worst
// case ~12s — restartProcess SIGTERMs itself and hard-exits after a 10s
// grace), so the successor must wait for the lock rather than die on it.
// A genuinely concurrent long-lived second process still fails, just after
// the wait.
// The restart handoff no longer overlaps by design — the old process closes
// the database (releasing this lock) and exits before its replacement is
// started, in both spawn and supervised restart modes (Server/restart.go).
// The retry survives as a safety net for the cases that can still race: a
// supervisor relaunching the service while a wedged predecessor is being
// backstop-killed, and the final old-style update from a release that still
// spawned mid-drain. A genuinely concurrent long-lived second process still
// fails, just after the wait.
func acquireProcessLock(dbPath string) (release func(), err error) {
const (
retryFor = 30 * time.Second
+38
View File
@@ -0,0 +1,38 @@
package main
import (
"errors"
"log/slog"
"net/http"
"time"
)
// serveWithBindRetry runs serve, retrying bounded times while the failure is
// an address-in-use bind conflict. http.ErrServerClosed (and nil) pass
// straight through — those are clean shutdowns, not failures.
//
// This is a safety net, not part of the restart design: the restart handoff
// releases the port before the successor starts (Server/restart.go), so the
// retry only matters for external squatters, supervisor relaunch races
// against a not-yet-dead predecessor, and platform TIME_WAIT edge cases.
// bindRetryEvery spaces the bind retries; a var only so the give-up-bound
// test doesn't sleep the real ~10 seconds.
var bindRetryEvery = 500 * time.Millisecond
func serveWithBindRetry(log *slog.Logger, label string, serve func() error) error {
const attempts = 20
var err error
for attempt := range attempts {
err = serve()
if err == nil || errors.Is(err, http.ErrServerClosed) {
return err
}
if attempt < attempts-1 && isAddrInUse(err) {
log.Warn("port in use, retrying...", "listener", label, "attempt", attempt+1, "error", err)
time.Sleep(bindRetryEvery)
continue
}
break
}
return err
}
+73 -31
View File
@@ -19,7 +19,6 @@ import (
"path/filepath"
"runtime"
"strconv"
"strings"
"syscall"
"time"
@@ -71,15 +70,41 @@ func main() {
log := slog.New(logctx.New(multiHandler))
slog.SetDefault(log)
if err := run(log, logBuf, levelVar); err != nil {
// The restart coordinator carries a self-restart request (update apply,
// backup restore, setup wizard) across run()'s teardown — see restart.go.
// The backstop closure fires only if a requested restart's drain wedges
// past restartBackstopDelay: it performs the handoff and force-exits,
// mirroring what the code below does on the healthy path.
var rc *restartCoordinator
rc = newRestartCoordinator(restartBackstopDelay, func() {
slog.Error("restart backstop fired — teardown exceeded its budget, exiting for handoff")
reason, _ := rc.Requested()
performRestartHandoff(reason, rc.Mode(), slog.Default())
os.Exit(0)
})
err := run(log, logBuf, levelVar, rc)
rc.disarm()
// Perform the handoff even when run() returned an error: a restart is
// only ever requested after a committed binary swap or a restore that
// closed the database, so not restarting is strictly worse than
// restarting into whatever the error was.
if reason, ok := rc.Requested(); ok {
performRestartHandoff(reason, rc.Mode(), log)
}
if err != nil {
_, _ = fmt.Fprintf(os.Stderr, "\n [ERROR] %v\n\n", err)
log.Error("server exited with error", "error", err)
os.Exit(1)
}
}
// run is the real entrypoint — separated for testability.
func run(log *slog.Logger, logBuf *admin.RingBuffer, levelVar *slog.LevelVar) error {
// run is the real entrypoint — separated for testability. rc carries a
// self-restart request out to main(), which performs the actual handoff once
// everything here has drained (see restart.go).
func run(log *slog.Logger, logBuf *admin.RingBuffer, levelVar *slog.LevelVar, rc *restartCoordinator) error {
// bgCtx is a cancellable context shared by all background goroutines
// (event persister, event pruner, plugin loader, maintenance loop).
//
@@ -92,14 +117,27 @@ func run(log *slog.Logger, logBuf *admin.RingBuffer, levelVar *slog.LevelVar) er
bgCtx, bgCancel := context.WithCancel(context.Background())
defer bgCancel()
// Clean up old binary from a previous update.
// Clean up old binary from a previous update. Bounded retry: in spawn
// mode the predecessor spawns this process as its very last act, so for
// the first few hundred milliseconds it may not have fully exited — and
// on Windows its image file (the .old after the swap) stays locked until
// it does.
exePath, exeErr := os.Executable()
if exeErr != nil {
log.Warn("failed to determine executable path", "error", exeErr)
} else {
oldPath := exePath + ".old"
if _, statErr := os.Stat(oldPath); statErr == nil {
if rmErr := os.Remove(oldPath); rmErr != nil {
var rmErr error
for attempt := range 5 {
if attempt > 0 {
time.Sleep(250 * time.Millisecond)
}
if rmErr = os.Remove(oldPath); rmErr == nil {
break
}
}
if rmErr != nil {
log.Warn("failed to remove old binary", "path", oldPath, "error", rmErr)
} else {
log.Info("removed old binary from previous update", "path", oldPath)
@@ -122,6 +160,11 @@ func run(log *slog.Logger, logBuf *admin.RingBuffer, levelVar *slog.LevelVar) er
log.Warn("unknown logging.level, keeping info", "value", cfg.Logging.Level)
}
// Resolve how a self-restart hands off (spawn the replacement vs exit
// for a supervisor) now that config is loaded — main() reads it back
// after run() returns.
rc.SetMode(resolveRestartMode(cfg.Server.RestartMode, log))
// ── 2. Ensure data directory exists ────────────────────────────────────
if mkdirErr := os.MkdirAll(cfg.Server.DataDir, 0o750); mkdirErr != nil {
return fmt.Errorf("creating data dir %s: %w", cfg.Server.DataDir, mkdirErr)
@@ -167,6 +210,11 @@ func run(log *slog.Logger, logBuf *admin.RingBuffer, levelVar *slog.LevelVar) er
// Backup handlers and the scheduled-backup maintenance write to the
// configured backup directory (defaults to data/backups).
admin.SetBackupDir(cfg.Backup.Dir)
// Admin restart requests (update apply, backup restore, setup wizard)
// land in the coordinator, which drains this process and lets main()
// perform the handoff. Wired before the listener starts serving, so no
// admin request can ever hit the unwired default hook.
admin.SetRestartHandoff(rc.Request)
if err := db.Migrate(database); err != nil {
return fmt.Errorf("running migrations: %w", err)
@@ -316,8 +364,8 @@ func run(log *slog.Logger, logBuf *admin.RingBuffer, levelVar *slog.LevelVar) er
}
go func() {
log.Info("ACME HTTP challenge server starting on :80")
if err := acmeSrv.ListenAndServe(); err != nil && !errors.Is(err, http.ErrServerClosed) {
log.Error("ACME HTTP server error", "error", err)
if err := serveWithBindRetry(log, "acme-http", acmeSrv.ListenAndServe); err != nil && !errors.Is(err, http.ErrServerClosed) {
log.Error("ACME HTTP server error — HTTP-01 challenges and certificate renewal will fail until the next restart", "error", err)
}
}()
}
@@ -410,8 +458,12 @@ func run(log *slog.Logger, logBuf *admin.RingBuffer, levelVar *slog.LevelVar) er
}
}()
// Listen for OS signals for graceful shutdown.
ctx, stop := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM)
// Listen for OS signals for graceful shutdown. The coordinator's context
// is the parent, so a programmatic restart request (rc.Request) drains
// exactly like a SIGTERM — including on Windows, where a process cannot
// signal itself. Signals arriving mid-drain are swallowed until stop()
// runs, same as on the real-signal path.
ctx, stop := signal.NotifyContext(rc.Context(), os.Interrupt, syscall.SIGTERM)
defer stop()
// Start serving in a goroutine.
@@ -419,23 +471,14 @@ func run(log *slog.Logger, logBuf *admin.RingBuffer, levelVar *slog.LevelVar) er
go func() {
log.Info("server starting", "addr", addr, "tls", tlsCfg != nil, "version", version)
for attempt := range 20 {
var listenErr error
err := serveWithBindRetry(log, "server", func() error {
if tlsCfg != nil {
listenErr = srv.ListenAndServeTLS("", "")
} else {
listenErr = srv.ListenAndServe()
return srv.ListenAndServeTLS("", "")
}
if listenErr != nil && !errors.Is(listenErr, http.ErrServerClosed) {
// Check if it's an "address already in use" error (port not released yet from old process)
if attempt < 19 && isAddrInUse(listenErr) {
log.Warn("port in use, retrying...", "attempt", attempt+1, "error", listenErr)
time.Sleep(500 * time.Millisecond)
continue
}
serveErr <- listenErr
}
break
return srv.ListenAndServe()
})
if err != nil && !errors.Is(err, http.ErrServerClosed) {
serveErr <- err
}
close(serveErr)
}()
@@ -447,7 +490,11 @@ func run(log *slog.Logger, logBuf *admin.RingBuffer, levelVar *slog.LevelVar) er
return fmt.Errorf("server error: %w", err)
}
case <-ctx.Done():
log.Info("shutdown signal received, draining connections (30s timeout)")
if reason, ok := rc.Requested(); ok {
log.Info("restart requested, draining connections (30s timeout)", "reason", reason)
} else {
log.Info("shutdown signal received, draining connections (30s timeout)")
}
}
// Graceful shutdown.
@@ -645,11 +692,6 @@ func seedHubReplayState(ctx context.Context, hub *ws.Hub, database *db.DB, log *
hub.MarkVisibilityChanged()
}
// isAddrInUse checks if an error is an "address already in use" error.
func isAddrInUse(err error) bool {
return err != nil && (strings.Contains(err.Error(), "address already in use") || strings.Contains(err.Error(), "Only one usage of each socket address"))
}
// printBanner writes the startup banner to stderr (so it doesn't mix with
// the structured log output on stdout).
func printBanner(cfg *config.Config, ver string, tls bool) {
+5 -1
View File
@@ -45,9 +45,13 @@ func TestRun_ServeErrorReturn_StopsHubDispatchGoroutine(t *testing.T) {
leakOpt := goleak.IgnoreCurrent()
if err := run(log, logBuf, levelVar); err == nil {
rc := newRestartCoordinator(time.Hour, nil)
if err := run(log, logBuf, levelVar, rc); err == nil {
t.Fatal("expected run() to return an error for an out-of-range port")
}
if _, requested := rc.Requested(); requested {
t.Error("no restart was requested, but the coordinator reports one")
}
// hub.Run's dispatch goroutine only exits once hub.stop is closed, which
// only happens inside hub.GracefulStop(). If run() returned without
+189
View File
@@ -0,0 +1,189 @@
// Restart coordination for self-restarts (update apply, backup restore,
// setup wizard).
//
// The admin package never stops or spawns processes; it requests a restart
// through admin.SetRestartHandoff, which lands in restartCoordinator.Request
// here. Request cancels the parent context of run()'s signal.NotifyContext,
// so the process drains through the exact same graceful path a SIGTERM takes
// — including on Windows, where a process cannot deliver a signal to itself.
// Only after run() has fully torn down (HTTP listeners closed, hub and
// LiveKit stopped, queues flushed, database closed and its process lock
// released) does main() perform the handoff: spawn the replacement binary
// when self-managed, or just exit and let the process supervisor relaunch
// the service. The old process being completely gone before the successor
// starts is what makes the handoff deterministic — the DB-lock and bind
// retries in db/ and main.go survive only as safety nets.
package main
import (
"context"
"log/slog"
"os"
"path/filepath"
"time"
"github.com/owncord/server/syncutil"
"github.com/owncord/server/updater"
)
const (
restartModeSpawn = "spawn"
restartModeSupervised = "supervised"
// restartBackstopDelay bounds how long a requested restart may drain
// before the process force-exits (performing the handoff first). run()'s
// worst-case legitimate teardown is ≈55s — the 30s shutdown budget plus
// its sequential bounded defers — so 90s only ever fires on a genuinely
// wedged teardown. The successor's lock/bind retries absorb whatever a
// backstop exit leaves unreleased.
restartBackstopDelay = 90 * time.Second
)
// restartCoordinator owns the lifecycle of one restart request. It is
// created in main(), threaded into run() (as a parameter, so tests drive
// run() with their own instance), and consulted by main() again after run()
// returns.
type restartCoordinator struct {
ctx context.Context
cancel context.CancelFunc
backstopDelay time.Duration
onBackstop func()
mu syncutil.Mutex
requested bool
reason string
mode string
backstop *time.Timer
}
// newRestartCoordinator builds a coordinator whose Context() is the parent
// for run()'s signal.NotifyContext. onBackstop runs once if a requested
// restart's drain exceeds backstopDelay; production passes handoff+os.Exit,
// tests pass a recorder.
func newRestartCoordinator(backstopDelay time.Duration, onBackstop func()) *restartCoordinator {
ctx, cancel := context.WithCancel(context.Background())
return &restartCoordinator{
ctx: ctx,
cancel: cancel,
backstopDelay: backstopDelay,
onBackstop: onBackstop,
}
}
// Context is the parent context for run()'s signal handling: cancelling it
// (Request) is indistinguishable from a shutdown signal to everything
// downstream. A Request issued before run() reaches NotifyContext is safe —
// NotifyContext over an already-cancelled parent starts out done, and run()
// falls straight through to graceful teardown.
func (rc *restartCoordinator) Context() context.Context { return rc.ctx }
// SetMode records the resolved restart mode ("spawn"/"supervised") once
// config is loaded; Mode reads it back for the handoff.
func (rc *restartCoordinator) SetMode(mode string) {
rc.mu.Lock()
rc.mode = mode
rc.mu.Unlock()
}
func (rc *restartCoordinator) Mode() string {
rc.mu.Lock()
defer rc.mu.Unlock()
return rc.mode
}
// Request records a restart request and starts the drain. Idempotent: the
// first reason wins, duplicates are logged and dropped. Arms the backstop
// timer before cancelling so a wedged teardown can never outlive it.
func (rc *restartCoordinator) Request(reason string) {
rc.mu.Lock()
if rc.requested {
pending := rc.reason
rc.mu.Unlock()
slog.Info("restart already pending — duplicate request dropped",
"reason", reason, "pending_reason", pending)
return
}
rc.requested = true
rc.reason = reason
mode := rc.mode
if rc.onBackstop != nil {
rc.backstop = time.AfterFunc(rc.backstopDelay, rc.onBackstop)
}
rc.mu.Unlock()
slog.Info("restart requested — draining for handoff", "reason", reason, "mode", mode)
rc.cancel()
}
// Requested reports whether a restart was requested, and its reason.
func (rc *restartCoordinator) Requested() (reason string, ok bool) {
rc.mu.Lock()
defer rc.mu.Unlock()
return rc.reason, rc.requested
}
// disarm stops the backstop timer. main() calls it the moment run() returns:
// from there the handoff is in main()'s hands and a delayed force-exit would
// only race it.
func (rc *restartCoordinator) disarm() {
rc.mu.Lock()
if rc.backstop != nil {
rc.backstop.Stop()
rc.backstop = nil
}
rc.mu.Unlock()
}
// resolveRestartMode turns cfg.Server.RestartMode into the effective handoff
// mode. Explicit "spawn"/"supervised" win; "auto" (or empty, or an unknown
// value after a warning) detects: containers and supervised services exit
// for their supervisor/engine to relaunch, everything else spawns its own
// replacement.
func resolveRestartMode(cfgVal string, log *slog.Logger) string {
switch cfgVal {
case restartModeSpawn, restartModeSupervised:
return cfgVal
case "", "auto":
default:
log.Warn("unknown server.restart_mode, using auto detection",
"value", cfgVal, "valid", "auto|spawn|supervised")
}
if updater.RunningInContainer() || updater.RunningUnderSupervisor() {
return restartModeSupervised
}
return restartModeSpawn
}
// spawnReplacement is the replacement-process spawner, swappable in tests
// (which must not start real processes).
var spawnReplacement = updater.SpawnDetached
// performRestartHandoff completes a requested restart after run() has fully
// drained. In supervised mode the handoff IS the exit — the supervisor
// (systemd Restart=, NSSM AppExit, Docker restart policy) relaunches the
// service, now running the swapped binary. In spawn mode the replacement is
// started directly; every resource is already released, so the successor
// boots with no lock or port contention. A failed spawn leaves the server
// down — loudly logged; there is no hub left to notify clients through.
func performRestartHandoff(reason, mode string, log *slog.Logger) {
if mode == restartModeSupervised {
log.Info("restart: exiting for the supervisor to relaunch", "reason", reason, "mode", mode)
return
}
exePath, err := os.Executable()
if err != nil {
log.Error("restart: cannot determine executable path — manual restart required",
"reason", reason, "error", err)
return
}
if resolved, symErr := filepath.EvalSymlinks(exePath); symErr == nil {
exePath = resolved
}
if err := spawnReplacement(exePath, os.Args[1:]); err != nil {
log.Error("restart: spawning the replacement process FAILED — manual restart required",
"reason", reason, "error", err)
return
}
log.Info("restart: replacement process spawned", "reason", reason, "path", exePath)
}
+260
View File
@@ -0,0 +1,260 @@
package main
import (
"fmt"
"io"
"log/slog"
"net"
"net/http"
"os"
"os/signal"
"testing"
"time"
"go.uber.org/goleak"
"github.com/owncord/server/admin"
)
func TestRestartCoordinator_RequestIdempotent(t *testing.T) {
rc := newRestartCoordinator(time.Hour, nil)
defer rc.disarm()
if reason, ok := rc.Requested(); ok || reason != "" {
t.Fatalf("fresh coordinator Requested() = %q, %v; want none", reason, ok)
}
rc.Request("first")
rc.Request("second")
reason, ok := rc.Requested()
if !ok || reason != "first" {
t.Errorf("Requested() = %q, %v; want first reason to win", reason, ok)
}
select {
case <-rc.Context().Done():
default:
t.Error("coordinator context not cancelled after Request")
}
}
// Cancelling the coordinator's context must drive a signal.NotifyContext
// built on top of it — that is the whole mechanism by which a restart
// request drains run() exactly like a SIGTERM, on every platform.
func TestRestartCoordinator_CancelDrivesNotifyContext(t *testing.T) {
rc := newRestartCoordinator(time.Hour, nil)
defer rc.disarm()
ctx, stop := signal.NotifyContext(rc.Context(), os.Interrupt)
defer stop()
rc.Request("test")
select {
case <-ctx.Done():
case <-time.After(2 * time.Second):
t.Fatal("NotifyContext not done after coordinator Request — restart requests would never drain the server")
}
}
func TestRestartCoordinator_Backstop(t *testing.T) {
t.Run("fires after the delay", func(t *testing.T) {
fired := make(chan struct{})
rc := newRestartCoordinator(10*time.Millisecond, func() { close(fired) })
rc.Request("test")
select {
case <-fired:
case <-time.After(2 * time.Second):
t.Fatal("backstop did not fire")
}
})
t.Run("disarm stops it", func(t *testing.T) {
fired := make(chan struct{})
rc := newRestartCoordinator(50*time.Millisecond, func() { close(fired) })
rc.Request("test")
rc.disarm()
select {
case <-fired:
t.Fatal("backstop fired after disarm")
case <-time.After(200 * time.Millisecond):
}
})
t.Run("not armed without a request", func(t *testing.T) {
fired := make(chan struct{})
_ = newRestartCoordinator(10*time.Millisecond, func() { close(fired) })
select {
case <-fired:
t.Fatal("backstop fired without a restart request")
case <-time.After(100 * time.Millisecond):
}
})
}
func TestResolveRestartMode(t *testing.T) {
log := slog.New(slog.NewTextHandler(io.Discard, nil))
// Pin every detector input: CI runners themselves can execute under
// systemd and carry a real INVOCATION_ID into the test process.
// present-but-empty counts as unset for all three.
pinBare := func(t *testing.T) {
t.Setenv("INVOCATION_ID", "")
t.Setenv("NSSM_SERVICE_NAME", "")
t.Setenv("OWNCORD_CONTAINER", "0")
}
t.Run("explicit values win over detection", func(t *testing.T) {
t.Setenv("INVOCATION_ID", "detected-supervisor")
if got := resolveRestartMode("spawn", log); got != restartModeSpawn {
t.Errorf("resolveRestartMode(spawn) = %q under systemd, want spawn", got)
}
pinBare(t)
if got := resolveRestartMode("supervised", log); got != restartModeSupervised {
t.Errorf("resolveRestartMode(supervised) = %q on bare metal, want supervised", got)
}
})
t.Run("auto on bare metal spawns", func(t *testing.T) {
pinBare(t)
for _, v := range []string{"auto", "", "bogus-mode"} {
if got := resolveRestartMode(v, log); got != restartModeSpawn {
t.Errorf("resolveRestartMode(%q) = %q, want spawn", v, got)
}
}
})
t.Run("auto under systemd is supervised", func(t *testing.T) {
pinBare(t)
t.Setenv("INVOCATION_ID", "4a1f3b0e9c8d4e2f")
if got := resolveRestartMode("auto", log); got != restartModeSupervised {
t.Errorf("resolveRestartMode(auto) = %q, want supervised", got)
}
})
t.Run("auto in a container is supervised", func(t *testing.T) {
pinBare(t)
t.Setenv("OWNCORD_CONTAINER", "1")
if got := resolveRestartMode("auto", log); got != restartModeSupervised {
t.Errorf("resolveRestartMode(auto) = %q, want supervised (restore/wizard restarts in Docker rely on the restart policy)", got)
}
})
}
func TestPerformRestartHandoff(t *testing.T) {
log := slog.New(slog.NewTextHandler(io.Discard, nil))
type call struct {
exe string
args []string
}
var calls []call
prev := spawnReplacement
spawnReplacement = func(exePath string, args []string) error {
calls = append(calls, call{exePath, args})
return nil
}
defer func() { spawnReplacement = prev }()
performRestartHandoff("update", restartModeSupervised, log)
if len(calls) != 0 {
t.Fatalf("supervised handoff spawned a process: %+v — the supervisor owns the relaunch", calls)
}
performRestartHandoff("update", restartModeSpawn, log)
if len(calls) != 1 {
t.Fatalf("spawn handoff made %d spawn calls, want 1", len(calls))
}
if calls[0].exe == "" {
t.Error("spawn handoff passed an empty executable path")
}
// A failing spawn must be survivable (logged, no panic) — there is no
// hub left to notify at this point.
spawnReplacement = func(string, []string) error { return fmt.Errorf("injected spawn failure") }
performRestartHandoff("update", restartModeSpawn, log)
}
// TestRun_RestartRequest_DrainsCleanly drives run() end to end: boot on a
// real port, request a restart through the coordinator (exactly what an
// admin update/restore does via admin.SetRestartHandoff), and assert run()
// drains and returns nil with no leaked goroutines — the property main()'s
// post-run handoff depends on (DB closed and lock released, port free,
// LiveKit stopped) before it starts the successor.
func TestRun_RestartRequest_DrainsCleanly(t *testing.T) {
t.Chdir(t.TempDir())
// Grab a free port, release it, and hand it to run(). The tiny window
// where something else could take it is absorbed by run()'s bind retry.
l, err := net.Listen("tcp", "127.0.0.1:0")
if err != nil {
t.Fatalf("probe listen: %v", err)
}
port := l.Addr().(*net.TCPAddr).Port
_ = l.Close()
t.Setenv("OWNCORD_SERVER_PORT", fmt.Sprint(port))
t.Setenv("OWNCORD_TLS_MODE", "off")
t.Setenv("OWNCORD_VOICE_AUTO_DOWNLOAD_LIVEKIT", "false") // keep the test offline
logBuf := admin.NewRingBuffer(64)
levelVar := new(slog.LevelVar)
log := slog.New(slog.NewTextHandler(io.Discard, &slog.HandlerOptions{Level: levelVar}))
leakOpt := goleak.IgnoreCurrent()
rc := newRestartCoordinator(time.Hour, nil)
runErr := make(chan error, 1)
go func() { runErr <- run(log, logBuf, levelVar, rc) }()
// Wait for the server to actually serve before requesting the restart.
healthURL := fmt.Sprintf("http://127.0.0.1:%d/health", port)
up := false
for deadline := time.Now().Add(15 * time.Second); time.Now().Before(deadline); {
resp, healthErr := http.Get(healthURL) //nolint:gosec // G107: loopback URL built from the test's own port
if healthErr == nil {
_, _ = io.Copy(io.Discard, resp.Body)
_ = resp.Body.Close()
up = true
break
}
select {
case err := <-runErr:
t.Fatalf("run() exited before serving: %v", err)
case <-time.After(50 * time.Millisecond):
}
}
if !up {
t.Fatal("server never became reachable on /health")
}
rc.Request("test-restart")
select {
case err := <-runErr:
if err != nil {
t.Fatalf("run() after restart request = %v, want nil (clean drain)", err)
}
case <-time.After(60 * time.Second):
t.Fatal("run() did not return after the restart request")
}
rc.disarm()
if reason, ok := rc.Requested(); !ok || reason != "test-restart" {
t.Errorf("Requested() = %q, %v; want the recorded restart", reason, ok)
}
// Everything must have drained: this is the guarantee that lets main()
// start the successor with zero lock/port contention.
if err := goleak.Find(leakOpt); err != nil {
t.Errorf("goroutines leaked after a restart-request drain: %v", err)
}
// The port must actually be free again.
relisten, err := net.Listen("tcp", fmt.Sprintf("127.0.0.1:%d", port))
if err != nil {
t.Errorf("port still held after drain: %v", err)
} else {
_ = relisten.Close()
}
}
+27
View File
@@ -0,0 +1,27 @@
package updater
import "os"
// RunningUnderSupervisor reports whether the server appears to be running
// under a process supervisor that will relaunch it after a clean exit. It
// decides how a self-restart (update apply, backup restore, setup wizard)
// hands off: under a supervisor the server just exits and lets the supervisor
// start the new binary; unsupervised it spawns the replacement itself.
//
// Detected supervisors:
// - systemd: INVOCATION_ID is set for every process systemd starts as a
// unit (v232+). Interactive shells are started by terminal emulators,
// not by the service manager, so they do not carry it.
// - NSSM: NSSM_SERVICE_NAME. NSSM 2.24 (the release the deployment docs
// install) does NOT set this for the service process — only newer
// pre-releases do — so NSSM deployments must set
// OWNCORD_SERVER_RESTART_MODE=supervised explicitly (docs/deployment.md).
// The check stays because it makes future NSSM releases work unconfigured.
//
// This is only the auto-detection half of the decision: server.restart_mode
// ("spawn"/"supervised") overrides it in both directions, and containers
// (RunningInContainer) are treated as supervised by the caller because the
// engine's restart policy is what relaunches PID 1.
func RunningUnderSupervisor() bool {
return os.Getenv("INVOCATION_ID") != "" || os.Getenv("NSSM_SERVICE_NAME") != ""
}
+34
View File
@@ -0,0 +1,34 @@
package updater
import "testing"
// Non-empty INVOCATION_ID (systemd) or NSSM_SERVICE_NAME (NSSM) means
// supervised; empty counts as unset. Both are pinned empty in the negative
// case because CI runners themselves can execute under systemd and carry a
// real INVOCATION_ID into the test process.
func TestRunningUnderSupervisor(t *testing.T) {
cases := []struct {
name string
invocationID string
nssmService string
want bool
}{
{"bare (both empty)", "", "", false},
{"systemd", "4a1f3b0e9c8d4e2f8a7b6c5d4e3f2a1b", "", true},
{"nssm", "", "OwnCord", true},
{"both", "abc", "OwnCord", true},
}
for _, tc := range cases {
t.Run(tc.name, func(t *testing.T) {
// t.Setenv with "" makes the variable present-but-empty, which the
// non-empty check treats exactly like unset — the same trick
// container_test.go relies on for hermetic negatives.
t.Setenv("INVOCATION_ID", tc.invocationID)
t.Setenv("NSSM_SERVICE_NAME", tc.nssmService)
if got := RunningUnderSupervisor(); got != tc.want {
t.Errorf("RunningUnderSupervisor() = %v, want %v (INVOCATION_ID=%q NSSM_SERVICE_NAME=%q)",
got, tc.want, tc.invocationID, tc.nssmService)
}
})
}
}
+19
View File
@@ -0,0 +1,19 @@
//go:build linux
package ws
import "syscall"
// liveKitSysProcAttr asks the kernel to SIGKILL the companion livekit-server
// if this process dies without running its own teardown (kill -9, OOM kill,
// a wedged shutdown force-exited by the restart backstop). The normal stop
// path is still LiveKitProcess.Stop via hub.GracefulStop — this only closes
// the hole where an orphaned livekit-server keeps TCP 7880 and the UDP media
// range bound, crash-looping the successor's LiveKit until someone kills the
// orphan by hand.
//
// Pdeathsig is Linux-only (prctl PR_SET_PDEATHSIG); other platforms return
// nil and rely on the graceful path alone.
func liveKitSysProcAttr() *syscall.SysProcAttr {
return &syscall.SysProcAttr{Pdeathsig: syscall.SIGKILL}
}
+21
View File
@@ -0,0 +1,21 @@
//go:build linux
package ws
import (
"syscall"
"testing"
)
// The companion livekit-server must die with a parent that never ran its
// teardown (kill -9, OOM, backstop force-exit): without Pdeathsig the orphan
// keeps 7880/UDP bound and the successor's LiveKit crash-loops.
func TestLiveKitSysProcAttr_SetsPdeathsig(t *testing.T) {
attr := liveKitSysProcAttr()
if attr == nil {
t.Fatal("liveKitSysProcAttr() = nil on linux, want Pdeathsig attr")
}
if attr.Pdeathsig != syscall.SIGKILL {
t.Errorf("Pdeathsig = %v, want SIGKILL", attr.Pdeathsig)
}
}
+14
View File
@@ -0,0 +1,14 @@
//go:build !linux
package ws
import "syscall"
// liveKitSysProcAttr returns nil: parent-death signaling (Pdeathsig) is a
// Linux prctl feature. On Windows and macOS the companion livekit-server is
// stopped only by the graceful path (LiveKitProcess.Stop via
// hub.GracefulStop); a Windows job object would be the equivalent hardening
// and is deliberately out of scope here.
func liveKitSysProcAttr() *syscall.SysProcAttr {
return nil
}
+6
View File
@@ -261,6 +261,12 @@ func (p *LiveKitProcess) runLoop(ctx context.Context, cfgPath, binPath string) {
cmd.Stdout = os.Stdout
cmd.Stderr = os.Stderr
cmd.WaitDelay = 6 * time.Second // bound Wait to prevent goroutine leak on Windows
if attr := liveKitSysProcAttr(); attr != nil {
// Linux: die with a parent that never ran its teardown (kill -9,
// OOM, restart-backstop force-exit) instead of orphaning with
// 7880/UDP still bound.
cmd.SysProcAttr = attr
}
slog.Info("livekit: starting process",
"binary", binPath,