test(distributed): add kill and restart primitives to the cluster harness

The point of running LocalAI as real child processes is to be able to take
one away. Add KillFrontend (SIGKILL, the lost replica), StopFrontendGracefully
(SIGTERM, the rolling update), KillWorker, RestartFrontend and FrontendAlive.

RestartFrontend pins the dead replica's original port. 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 exactly the workers that registered
with it and the failover under test never happens.

It also wipes the replica's data directory, so the process comes back with
empty local state and has to rehydrate node, session and job state from the
shared Postgres and NATS. Reusing the directory would model a pod with a
persistent volume and hide the class of bug these tests exist to find. That
is only safe because the harness pins LOCALAI_AUTH_HMAC_SECRET; otherwise the
wipe would take {DataPath}/.hmac_secret with it and every session minted
before the restart would 401 afterwards.

FrontendAlive consults the reaper's exited channel before signal 0: a child
that has died but has not yet been waited on is a zombie, and signal 0 to a
zombie succeeds, which would report a dead replica as alive.

The new specs cover argument validation only. Killing, stopping and
restarting a live process needs a built binary plus Postgres and NATS, so
those paths stay unexecuted until the failover suites land.

Assisted-by: Claude Opus 5 [claude-code]
Signed-off-by: Ettore Di Giacinto <mudler@localai.io>

tests/e2e/distributed/cluster/cluster.go

@@ -152,8 +152,8 @@ func (c *Cluster) startFrontend(i int, port int) (*Process, error) {
 		}
 		port = allocated
 	}
-	name := fmt.Sprintf("frontend-%d", i)
-	dir := filepath.Join(c.baseDir, name)
+	name := frontendName(i)
+	dir := c.frontendDir(i)
 	if err := os.MkdirAll(filepath.Join(dir, "models"), 0o755); err != nil {
 		return nil, fmt.Errorf("creating %s dirs: %w", name, err)
 	}
@@ -164,7 +164,7 @@ func (c *Cluster) startFrontend(i int, port int) (*Process, error) {
 	// ${cwd}/data (core/cli/run.go:48), which under `go test` is inside the
 	// source tree and shared by every replica: one collectiondb, one task and
 	// job store for processes that are meant to be independent.
-	dataPath := filepath.Join(dir, "data")
+	dataPath := c.frontendDataDir(i)
 	if err := os.MkdirAll(dataPath, 0o750); err != nil {
 		return nil, fmt.Errorf("creating %s dirs: %w", name, err)
 	}

tests/e2e/distributed/cluster/cluster_test.go

@@ -57,3 +57,31 @@ var _ = Describe("Admin session", Label("Distributed"), func() {
 		Expect(err.Error()).To(ContainSubstring("frontend 1"))
 	})
 })
+
+// Like the admin specs above, these cover argument validation only. Killing,
+// stopping and restarting a real replica needs a built local-ai plus Postgres
+// and NATS, so those paths stay unexecuted until the failover suites land.
+var _ = Describe("Failure primitives", Label("Distributed"), func() {
+	It("rejects an out-of-range frontend index rather than panicking", func() {
+		c := cluster.ForTestingEmpty()
+		Expect(c.KillFrontend(0)).To(MatchError(ContainSubstring("frontend 0 out of range")))
+		Expect(c.StopFrontendGracefully(2)).To(MatchError(ContainSubstring("frontend 2 out of range")))
+		Expect(c.KillWorker(1)).To(MatchError(ContainSubstring("worker 1 out of range")))
+	})
+
+	It("rejects a restart of a frontend index that does not exist", func() {
+		Expect(cluster.ForTestingEmpty().RestartFrontend(0)).
+			To(MatchError(ContainSubstring("frontend 0 out of range")))
+	})
+
+	It("rejects a negative index without treating it as an offset from the end", func() {
+		c := cluster.ForTestingEmpty()
+		Expect(c.KillFrontend(-1)).To(MatchError(ContainSubstring("frontend -1 out of range")))
+		Expect(c.KillWorker(-1)).To(MatchError(ContainSubstring("worker -1 out of range")))
+		Expect(c.FrontendAlive(-1)).To(BeFalse())
+	})
+
+	It("reports a frontend that was never started as not alive", func() {
+		Expect(cluster.ForTestingEmpty().FrontendAlive(0)).To(BeFalse())
+	})
+})

tests/e2e/distributed/cluster/failure.go

@@ -0,0 +1,144 @@
+package cluster
+
+import (
+	"fmt"
+	"os"
+	"path/filepath"
+	"syscall"
+)
+
+// KillFrontend SIGKILLs frontend i. This is the "replica died" case: no drain,
+// no graceful deregistration, sockets drop without a FIN from the application.
+//
+// The signal is delivered but not waited on, because a spec that asserts on the
+// cluster's reaction wants to observe the window between the death and the
+// survivors noticing it. Poll FrontendAlive with Eventually to join the exit.
+func (c *Cluster) KillFrontend(i int) error {
+	if err := c.checkFrontendIndex(i); err != nil {
+		return err
+	}
+	return signalProcess(c.frontends[i], syscall.SIGKILL)
+}
+
+// StopFrontendGracefully SIGTERMs frontend i. This is the rolling-update case:
+// the process gets a chance to drain and deregister. Like KillFrontend it does
+// not wait; the point of the distinction between the two is what the process
+// does with the time between the signal and its exit.
+func (c *Cluster) StopFrontendGracefully(i int) error {
+	if err := c.checkFrontendIndex(i); err != nil {
+		return err
+	}
+	return signalProcess(c.frontends[i], syscall.SIGTERM)
+}
+
+// KillWorker SIGKILLs worker i.
+func (c *Cluster) KillWorker(i int) error {
+	if i < 0 || i >= len(c.workers) {
+		return fmt.Errorf("worker %d out of range (cluster has %d)", i, len(c.workers))
+	}
+	return signalProcess(c.workers[i], syscall.SIGKILL)
+}
+
+// RestartFrontend brings frontend i back on its original port with an empty
+// data directory, modelling a replaced pod rather than a resumed one.
+//
+// The port is pinned rather than reallocated: workers read LOCALAI_REGISTER_TO
+// once at boot and never re-resolve it, so a replica that returns on a new port
+// is unreachable by exactly the workers that registered with it, and the
+// failover the spec means to observe never happens. Rebinding is safe because
+// the previous listener is fully closed before the new process starts (see the
+// terminate below) and Go's listeners set SO_REUSEADDR, so a lingering
+// TIME_WAIT on an accepted connection does not block the bind.
+//
+// The data directory is wiped so the replica must rehydrate node, session and
+// job state from the shared Postgres and NATS. Keeping it would model a pod
+// with a persistent volume and would hide the very class of bug these tests
+// exist to find. This is only safe because startFrontend pins
+// LOCALAI_AUTH_HMAC_SECRET: the secret otherwise lives at
+// {DataPath}/.hmac_secret, and wiping it would make every session minted before
+// the restart hash to a row the restarted replica cannot find, turning a
+// failover assertion into an unexplained 401.
+func (c *Cluster) RestartFrontend(i int) error {
+	if err := c.checkFrontendIndex(i); err != nil {
+		return err
+	}
+	old := c.frontends[i]
+	if old == nil {
+		return fmt.Errorf("frontend %d was never started, nothing to restart", i)
+	}
+	// The old process may still be running (a restart with no preceding kill) or
+	// already dead but unreaped. terminate is idempotent, bounds its wait, and
+	// releases the log handle the replacement is about to reopen; without it the
+	// replacement races the old listener for the port and leaks a file
+	// descriptor per restart.
+	old.terminate()
+
+	if err := os.RemoveAll(c.frontendDataDir(i)); err != nil {
+		return fmt.Errorf("wiping data dir of frontend %d: %w", i, err)
+	}
+
+	p, err := c.startFrontend(i, old.Port)
+	if err != nil {
+		return fmt.Errorf("restarting frontend %d: %w", i, err)
+	}
+	c.frontends[i] = p
+	return nil
+}
+
+// FrontendAlive reports whether frontend i's process is still running.
+func (c *Cluster) FrontendAlive(i int) bool {
+	if i < 0 || i >= len(c.frontends) {
+		return false
+	}
+	return c.frontends[i].alive()
+}
+
+// alive reports whether the process is still running. The reaper's exited
+// channel is authoritative and is consulted first: between a child's death and
+// the reaper's Wait returning, the child is a zombie, and signal 0 to a zombie
+// succeeds, which would report a dead replica as alive.
+func (p *Process) alive() bool {
+	if p == nil || p.Cmd == nil || p.Cmd.Process == nil {
+		return false
+	}
+	select {
+	case <-p.exited:
+		return false
+	default:
+	}
+	// Signal 0 tests for existence without delivering anything.
+	return p.Cmd.Process.Signal(syscall.Signal(0)) == nil
+}
+
+func signalProcess(p *Process, sig syscall.Signal) error {
+	if p == nil || p.Cmd == nil || p.Cmd.Process == nil {
+		return fmt.Errorf("process is not running")
+	}
+	if err := p.Cmd.Process.Signal(sig); err != nil {
+		return fmt.Errorf("signalling %s with %v: %w", p.Name, sig, err)
+	}
+	return nil
+}
+
+// checkFrontendIndex keeps the out-of-range wording identical across every
+// primitive, so a failing spec reads the same whichever one tripped.
+func (c *Cluster) checkFrontendIndex(i int) error {
+	if i < 0 || i >= len(c.frontends) {
+		return fmt.Errorf("frontend %d out of range (cluster has %d)", i, len(c.frontends))
+	}
+	return nil
+}
+
+func frontendName(i int) string {
+	return fmt.Sprintf("frontend-%d", i)
+}
+
+func (c *Cluster) frontendDir(i int) string {
+	return filepath.Join(c.baseDir, frontendName(i))
+}
+
+// frontendDataDir is LOCALAI_DATA_PATH for frontend i. RestartFrontend wipes it,
+// so it must be the exact path startFrontend hands the child.
+func (c *Cluster) frontendDataDir(i int) string {
+	return filepath.Join(c.frontendDir(i), "data")
+}