fix: keep systemd from killing a slow boot, and a healthy process

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 {

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

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)
+			}
+		})
+	}
+}