From f31ddbe50fefb9611126da8a02776482c9b23deb Mon Sep 17 00:00:00 2001 From: FrameAutomata Date: Fri, 4 Sep 2026 10:16:53 -0500 Subject: [PATCH] fix: keep systemd from killing a slow boot, and a healthy process notifySystemd() sent READY=1 only after migrations, the backfills, the project cache and the port bind, with nothing before it. A Type=notify unit whose boot outlasts the default TimeoutStartSec=90s is therefore SIGTERM'd and restarted into the same migration -- the systemd version of the loop #351 fixed for Kubernetes by budgeting 300s for the startup probe. beginBoot() extends the deadline every 10s from the top of Run() and returns the function that ends the boot phase, so stopping the heartbeat and reporting ready cannot be separated or reordered: once the service is running, EXTEND_TIMEOUT_USEC extends the watchdog deadline instead of the startup one, where a stray message would paper over a hung process. The extension is bounded at 30 minutes, because these messages go out on a timer rather than on boot progress -- unbounded, a boot wedged against an unreachable database would delete TimeoutStartSec rather than defer it. The watchdog ticker was hardcoded at 15s whatever systemd asked for in WATCHDOG_USEC, so WatchdogSec=10s had systemd kill a perfectly healthy process one ping in. It now pings at half the interval systemd asks for, and starts no goroutine at all when the watchdog is off or another process is the watched PID. Closes #361 Co-Authored-By: Claude Opus 5 (1M context) --- backend/cmd/run.go | 26 +---- backend/cmd/systemd.go | 145 ++++++++++++++++++++++++++ backend/cmd/systemd_test.go | 199 ++++++++++++++++++++++++++++++++++++ 3 files changed, 349 insertions(+), 21 deletions(-) create mode 100644 backend/cmd/systemd.go create mode 100644 backend/cmd/systemd_test.go diff --git a/backend/cmd/run.go b/backend/cmd/run.go index be09d6de..3f9f1c3b 100644 --- a/backend/cmd/run.go +++ b/backend/cmd/run.go @@ -35,7 +35,6 @@ import ( "github.com/tracewayapp/traceway/backend/app/synthetics" "github.com/tracewayapp/traceway/backend/static" - "github.com/coreos/go-systemd/v22/daemon" "github.com/gin-gonic/gin" "github.com/joho/godotenv" traceway "go.tracewayapp.com" @@ -83,6 +82,10 @@ func Run(opts ...Option) { } config.Init(cfg) + // Everything below runs before the service reports itself started, and + // migrations alone can outlast systemd's default 90s start timeout. + ready := beginBoot(bootExtendInterval, bootExtendWindow, bootExtendBudget) + if err := services.InitJWT(); err != nil { // A missing signing key is a configuration mistake, not a crash: print // something actionable instead of burying the message under a panic @@ -325,7 +328,7 @@ func Run(opts ...Option) { }(listener) } - notifySystemd() + ready() config.Logln("Starting server on " + listeners[0].Addr().String()) serveHTTP(router, listeners[0]) } @@ -505,25 +508,6 @@ func parsePositiveInt(s string, def int) int { return v } -func notifySystemd() { - sent, err := daemon.SdNotify(false, daemon.SdNotifyReady) - if err != nil { - config.Logf("Failed to notify systemd: %v", err) - } else if sent { - config.Logln("Notified systemd that service is ready") - } - - go func() { - defer traceway.Recover() - - ticker := time.NewTicker(15 * time.Second) - defer ticker.Stop() - for range ticker.C { - daemon.SdNotify(false, daemon.SdNotifyWatchdog) - } - }() -} - func mustSubFS(fsys fs.FS, dir string) fs.FS { sub, err := fs.Sub(fsys, dir) if err != nil { diff --git a/backend/cmd/systemd.go b/backend/cmd/systemd.go new file mode 100644 index 00000000..d2949c12 --- /dev/null +++ b/backend/cmd/systemd.go @@ -0,0 +1,145 @@ +package cmd + +import ( + "fmt" + "os" + "sync" + "time" + + "github.com/coreos/go-systemd/v22/daemon" + "github.com/tracewayapp/traceway/backend/app/config" + traceway "go.tracewayapp.com" +) + +// Boot can outlast systemd's default TimeoutStartSec=90s -- an index-building +// migration is the usual reason, which is why the Helm chart budgets 300s for +// its startup probe. Rather than ask operators to guess a TimeoutStartSec for +// their largest migration, hold the deadline open while boot runs: every +// EXTEND_TIMEOUT_USEC message pushes it out from the moment systemd receives +// it. +// +// The window is several times the interval so that one late heartbeat -- a +// stalled disk, a stop-the-world pause -- cannot let the deadline lapse. +// +// The total is bounded because these messages go out on a timer rather than on +// boot progress: without a budget, a boot wedged against an unreachable +// database would hold the deadline open forever, which deletes +// TimeoutStartSec instead of deferring it. Once the budget is spent, systemd's +// own timeout applies again. +const ( + bootExtendInterval = 10 * time.Second + bootExtendWindow = 60 * time.Second + bootExtendBudget = 30 * time.Minute +) + +// beginBoot holds the service manager's startup timeout open, and returns the +// function that ends the boot phase by reporting the service ready. +// +// Reporting ready is the same call so that no caller can leave a heartbeat +// running past it: once the service is running, EXTEND_TIMEOUT_USEC extends +// the *watchdog* deadline rather than the startup one, where a stray message +// would paper over a hung process. +// +// All of it is a no-op when the process was not started by a service manager +// (NOTIFY_SOCKET unset), which is every other way Traceway runs: Docker, +// Kubernetes, go run. +func beginBoot(interval, window, budget time.Duration) (ready func()) { + stop := startBootHeartbeat(interval, window, budget) + + return sync.OnceFunc(func() { + stop() + notifyReady() + }) +} + +func startBootHeartbeat(interval, window, budget time.Duration) (stop func()) { + // Ask the environment rather than infer it from the first send: SdNotify + // also reports "not sent" for a transient socket error, and treating that + // as "no service manager" would abandon the heartbeat for the whole boot + // and hand the process back to the 90s default this exists to survive. + if os.Getenv("NOTIFY_SOCKET") == "" { + return func() {} + } + + extendStartTimeout(window) + + stopped := make(chan struct{}) + done := make(chan struct{}) + + go func() { + defer traceway.Recover() + defer close(done) + + ticker := time.NewTicker(interval) + defer ticker.Stop() + spent := time.After(budget) + + for { + select { + case <-ticker.C: + extendStartTimeout(window) + case <-spent: + config.Logf("No longer extending the systemd startup timeout: boot has taken %s", budget) + return + case <-stopped: + return + } + } + }() + + return sync.OnceFunc(func() { + close(stopped) + <-done + }) +} + +func extendStartTimeout(window time.Duration) { + if _, err := daemon.SdNotify(false, fmt.Sprintf("EXTEND_TIMEOUT_USEC=%d", window.Microseconds())); err != nil { + config.Logf("Failed to extend the systemd startup timeout: %v", err) + } +} + +func notifyReady() { + sent, err := daemon.SdNotify(false, daemon.SdNotifyReady) + if err != nil { + config.Logf("Failed to notify systemd: %v", err) + } else if sent { + config.Logln("Notified systemd that service is ready") + } + + startWatchdog() +} + +// startWatchdog pings the watchdog at the interval systemd asked for rather +// than a fixed one: WatchdogSec=10s expects a ping every 5s, and the hardcoded +// 15s this replaces would have had systemd kill a perfectly healthy process. +func startWatchdog() { + ping, err := watchdogPingInterval() + if err != nil { + config.Logf("Failed to read the systemd watchdog interval: %v", err) + return + } + if ping <= 0 { + return + } + + config.Logf("Pinging the systemd watchdog every %s", ping) + + go func() { + defer traceway.Recover() + + ticker := time.NewTicker(ping) + defer ticker.Stop() + for range ticker.C { + daemon.SdNotify(false, daemon.SdNotifyWatchdog) + } + }() +} + +// watchdogPingInterval is half of WATCHDOG_USEC, the rate sd_notify(3) asks +// for. It is 0 when the watchdog is disabled or another process is the watched +// PID, in which case nothing should be sent at all. +func watchdogPingInterval() (time.Duration, error) { + interval, err := daemon.SdWatchdogEnabled(false) + return interval / 2, err +} diff --git a/backend/cmd/systemd_test.go b/backend/cmd/systemd_test.go new file mode 100644 index 00000000..532867f5 --- /dev/null +++ b/backend/cmd/systemd_test.go @@ -0,0 +1,199 @@ +package cmd + +import ( + "fmt" + "net" + "os" + "path/filepath" + "testing" + "time" +) + +const readTimeout = 2 * time.Second + +// notifySocket binds a unixgram socket and points NOTIFY_SOCKET at it, the way +// systemd does for a Type=notify unit. +func notifySocket(t *testing.T) *net.UnixConn { + t.Helper() + + // Keep test names short: the socket path has to fit sun_path's ~107 bytes. + path := filepath.Join(t.TempDir(), "n") + conn, err := net.ListenUnixgram("unixgram", &net.UnixAddr{Name: path, Net: "unixgram"}) + if err != nil { + t.Fatalf("bind %s: %v", path, err) + } + t.Cleanup(func() { conn.Close() }) + + t.Setenv("NOTIFY_SOCKET", path) + return conn +} + +func readMessage(t *testing.T, conn *net.UnixConn) string { + t.Helper() + + message, ok := readMessageWithin(t, conn, readTimeout) + if !ok { + t.Fatalf("no message arrived within %s", readTimeout) + } + return message +} + +func expectNoMessage(t *testing.T, conn *net.UnixConn, within time.Duration) { + t.Helper() + + if message, ok := readMessageWithin(t, conn, within); ok { + t.Fatalf("unexpected message %q", message) + } +} + +func readMessageWithin(t *testing.T, conn *net.UnixConn, within time.Duration) (string, bool) { + t.Helper() + + if err := conn.SetReadDeadline(time.Now().Add(within)); err != nil { + t.Fatalf("set deadline: %v", err) + } + buf := make([]byte, 256) + n, err := conn.Read(buf) + if err != nil { + return "", false + } + return string(buf[:n]), true +} + +func TestBootHeartbeatKeepsExtendingTheStartTimeout(t *testing.T) { + conn := notifySocket(t) + + stop := startBootHeartbeat(20*time.Millisecond, 5*time.Second, time.Minute) + defer stop() + + // The first extension goes out before Run() touches the database, so a boot + // that stalls immediately is still covered. + want := "EXTEND_TIMEOUT_USEC=5000000" + for i := range 3 { + if got := readMessage(t, conn); got != want { + t.Fatalf("message %d = %q, want %q", i, got, want) + } + } +} + +func TestNothingExtendsTheStartTimeoutAfterReady(t *testing.T) { + conn := notifySocket(t) + + ready := beginBoot(20*time.Millisecond, 5*time.Second, time.Minute) + readMessage(t, conn) + ready() + + // Anything after READY=1 extends the *watchdog* deadline instead of the + // startup one, so a heartbeat outliving ready() would hide a hung process. + for { + message := readMessage(t, conn) + if message == "READY=1" { + break + } + if message != "EXTEND_TIMEOUT_USEC=5000000" { + t.Fatalf("unexpected message before READY=1: %q", message) + } + } + expectNoMessage(t, conn, 100*time.Millisecond) + + // A second call must neither re-report ready nor start a second watchdog. + ready() + expectNoMessage(t, conn, 100*time.Millisecond) +} + +func TestBootHeartbeatStopsExtendingOnceTheBudgetIsSpent(t *testing.T) { + conn := notifySocket(t) + + // A boot wedged against an unreachable database must not hold systemd's + // start deadline open forever. + stop := startBootHeartbeat(10*time.Millisecond, 5*time.Second, 50*time.Millisecond) + defer stop() + + readMessage(t, conn) + deadline := time.Now().Add(readTimeout) + for { + if _, ok := readMessageWithin(t, conn, 150*time.Millisecond); !ok { + break + } + if time.Now().After(deadline) { + t.Fatal("heartbeat still extending well past its budget") + } + } +} + +func TestBootHeartbeatSurvivesAFailedFirstSend(t *testing.T) { + // NOTIFY_SOCKET names a socket nothing has bound yet, so the first send + // fails. Reading "no service manager" into that failure would abandon the + // heartbeat and drop boot back to the default TimeoutStartSec. + path := filepath.Join(t.TempDir(), "n") + t.Setenv("NOTIFY_SOCKET", path) + + stop := startBootHeartbeat(20*time.Millisecond, 5*time.Second, time.Minute) + defer stop() + + conn, err := net.ListenUnixgram("unixgram", &net.UnixAddr{Name: path, Net: "unixgram"}) + if err != nil { + t.Fatalf("bind %s: %v", path, err) + } + defer conn.Close() + + if got := readMessage(t, conn); got != "EXTEND_TIMEOUT_USEC=5000000" { + t.Fatalf("got %q, want the heartbeat to have retried", got) + } +} + +func TestBootHeartbeatIsANoOpWithoutAServiceManager(t *testing.T) { + t.Setenv("NOTIFY_SOCKET", "") + + done := make(chan struct{}) + go func() { + defer close(done) + beginBoot(bootExtendInterval, bootExtendWindow, bootExtendBudget)() + }() + + select { + case <-done: + case <-time.After(readTimeout): + t.Fatal("ready() blocked when NOTIFY_SOCKET is unset") + } +} + +func TestWatchdogPingIntervalIsHalfOfWhatSystemdAsksFor(t *testing.T) { + tests := []struct { + name string + usec string + pid string + want time.Duration + wantErr bool + }{ + {name: "disabled", usec: "", want: 0}, + // The interval this replaces was hardcoded at 15s, so systemd would + // have killed a healthy process one ping into a 10s watchdog. + {name: "shorter than the old hardcoded interval", usec: "10000000", want: 5 * time.Second}, + {name: "longer than the old hardcoded interval", usec: "60000000", want: 30 * time.Second}, + {name: "watched pid is this process", usec: "30000000", pid: fmt.Sprint(os.Getpid()), want: 15 * time.Second}, + {name: "watched pid is another process", usec: "30000000", pid: fmt.Sprint(os.Getpid() + 1), want: 0}, + {name: "malformed", usec: "soon", wantErr: true}, + } + + for _, test := range tests { + t.Run(test.name, func(t *testing.T) { + t.Setenv("WATCHDOG_USEC", test.usec) + t.Setenv("WATCHDOG_PID", test.pid) + + got, err := watchdogPingInterval() + if test.wantErr { + if err == nil { + t.Fatalf("got %s, want an error", got) + } + return + } + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + if got != test.want { + t.Fatalf("got %s, want %s", got, test.want) + } + }) + } +}