From a7847b8a37198560f940c863eb274c9851ca30dd Mon Sep 17 00:00:00 2001 From: Ettore Di Giacinto Date: Mon, 31 Aug 2026 11:18:49 +0000 Subject: [PATCH] test(distributed): make the cluster harness survive a restart Restarting a frontend replica must not move it: workers read LOCALAI_REGISTER_TO once at boot and never re-resolve it, so a replica that returns on a fresh port is unreachable by the workers that registered with it. startFrontend now takes the port, with <= 0 meaning "allocate". Process logs are opened for append rather than truncated, so a restarted process cannot erase the log of the instance that died, which is the log a failover post-mortem needs. The post-SIGKILL wait is bounded, so one stuck child no longer becomes a suite-wide timeout that names nothing. Stop is nil-safe because Start returns a nil cluster after stopping itself. Start's doc comment no longer claims to wait for worker registration; that needs an authenticated admin session, so it now says callers must poll /api/nodes themselves. Assisted-by: Claude Opus 5 [claude-code] Signed-off-by: Ettore Di Giacinto --- tests/e2e/distributed/cluster/cluster.go | 47 +++++++++++++++++++----- 1 file changed, 37 insertions(+), 10 deletions(-) diff --git a/tests/e2e/distributed/cluster/cluster.go b/tests/e2e/distributed/cluster/cluster.go index 306980e00..0445fe386 100644 --- a/tests/e2e/distributed/cluster/cluster.go +++ b/tests/e2e/distributed/cluster/cluster.go @@ -44,7 +44,10 @@ type Options struct { // Process is one running local-ai. type Process struct { - Name string + Name string + // Cmd is exposed for signalling only. Never call Cmd.Wait on it: the reaper + // goroutine started in spawn owns it, a second Wait races the first, and + // waitErr is only safe to read after <-p.exited. Cmd *exec.Cmd Port int LogPath string @@ -70,6 +73,10 @@ const ( defaultAdminEmail = "admin@e2e.local" readinessTimeout = 90 * time.Second readinessPoll = 200 * time.Millisecond + // processExitTimeout bounds the post-SIGKILL wait in terminate. An unbounded + // wait turns one stuck child (D state, or a Wait that never returns) into a + // suite-wide Ginkgo timeout that names nothing. + processExitTimeout = 10 * time.Second ) func (o *Options) applyDefaults() { @@ -94,8 +101,10 @@ func (o Options) validate() error { return nil } -// Start brings up the cluster and blocks until every frontend answers /readyz -// and every worker has registered. +// Start brings up the cluster. It blocks until every frontend answers /readyz +// and every worker process has been spawned. It does NOT wait for workers to +// register: that needs an authenticated admin session, so a caller that depends +// on registration must poll /api/nodes itself. func Start(opts Options) (*Cluster, error) { opts.applyDefaults() if err := opts.validate(); err != nil { @@ -110,7 +119,7 @@ func Start(opts Options) (*Cluster, error) { c := &Cluster{opts: opts, baseDir: baseDir} for i := 0; i < opts.Frontends; i++ { - p, err := c.startFrontend(i) + p, err := c.startFrontend(i, 0) if err != nil { c.Stop() return nil, err @@ -128,10 +137,17 @@ func Start(opts Options) (*Cluster, error) { return c, nil } -func (c *Cluster) startFrontend(i int) (*Process, error) { - port, err := freeport.GetFreePort() - if err != nil { - return nil, fmt.Errorf("allocating frontend port: %w", err) +// startFrontend starts frontend i. A port <= 0 allocates a fresh one; a pinned +// port exists for restart: workers take LOCALAI_REGISTER_TO once at boot and +// never re-resolve it, so a replica that comes back on a new port is +// unreachable by exactly the workers that registered with it. +func (c *Cluster) startFrontend(i int, port int) (*Process, error) { + if port <= 0 { + allocated, err := freeport.GetFreePort() + if err != nil { + return nil, fmt.Errorf("allocating frontend port: %w", err) + } + port = allocated } name := fmt.Sprintf("frontend-%d", i) dir := filepath.Join(c.baseDir, name) @@ -218,7 +234,9 @@ func (c *Cluster) startWorker(i int) (*Process, error) { func (c *Cluster) spawn(name string, cmd *exec.Cmd, port int) (*Process, error) { logPath := filepath.Join(c.opts.LogDir, name+".log") - f, err := os.Create(logPath) + // Append rather than truncate: a restarted process reopens the same path, and + // the log of the instance that died is the one a failover post-mortem needs. + f, err := os.OpenFile(logPath, os.O_CREATE|os.O_WRONLY|os.O_APPEND, 0o644) if err != nil { return nil, fmt.Errorf("creating log file for %s: %w", name, err) } @@ -246,7 +264,11 @@ func (p *Process) terminate() { return } _ = p.Cmd.Process.Kill() - <-p.exited + select { + case <-p.exited: + case <-time.After(processExitTimeout): + fmt.Printf("warning: %s did not exit within %s after SIGKILL; continuing teardown\n", p.Name, processExitTimeout) + } if p.logFile != nil { _ = p.logFile.Close() } @@ -265,6 +287,11 @@ func (c *Cluster) WorkerName(i int) string { // Stop terminates every process and removes the work directory. Logs survive in // LogDir, which the caller owns. func (c *Cluster) Stop() { + // Start returns (nil, err) after stopping itself, so a spec that defers + // c.Stop before asserting the error would otherwise nil-deref. + if c == nil { + return + } for _, p := range append(append([]*Process{}, c.workers...), c.frontends...) { p.terminate() }