feat(installer): activate source builds atomically

internal/installer/sourceinstall_ssh_test.go

@@ -4,6 +4,7 @@ import (
 	"crypto/ed25519"
 	"crypto/rand"
 	"fmt"
+	"io"
 	"net"
 	"strings"
 	"sync"
@@ -106,7 +107,18 @@ func (s *fakeSSHServer) handleConn(conn net.Conn, config *ssh.ServerConfig) {
 		}
 		go func() {
 			defer channel.Close()
+			// Drain the client's stdin so uploads (which stream base64
+			// over the session stdin) never block the channel window.
+			// The channel is not closed until the client's write side is
+			// exhausted, mirroring the real server behavior.
+			var drained sync.WaitGroup
+			drained.Add(1)
+			go func() {
+				defer drained.Done()
+				_, _ = io.Copy(io.Discard, channel)
+			}()
 			s.handleSession(channel, requests)
+			drained.Wait()
 		}()
 	}
 }
@@ -151,9 +163,10 @@ func (s *fakeSSHServer) execCommand(channel ssh.Channel, command string) {
 
 // defaultFakeExec simulates a minimal Linux host: it answers os-release,
 // uname, the init-marker file probe, origin URL, branch resolution, and
-// rev-parse, and accepts every install step. A pinned-branch existence
-// check (show-ref) fails by default so the missing-branch path is
-// exercised without extra setup.
+// rev-parse, accepts every install step, and simulates the work-package-4
+// activation (running-init probe, env/unit uploads, and the activate
+// script succeeding). A pinned-branch existence check (show-ref) fails by
+// default so the missing-branch path is exercised without extra setup.
 func defaultFakeExec(command string) (string, string, int) {
 	switch {
 	case strings.Contains(command, "cat /etc/os-release"):
@@ -170,6 +183,18 @@ func defaultFakeExec(command string) (string, string, int) {
 		return fakeCommitHex + "\n", "", 0
 	case strings.Contains(command, "show-ref"):
 		return "", "branch not found", 1
+	case strings.Contains(command, "softlevel"):
+		// running-init probe: no systemd/openrc is actually booted.
+		return "none\n", "", 0
+	case strings.Contains(command, "mktemp -d /tmp/rsmon-worker-act"):
+		// server-side secure 0700 upload dir.
+		return "/tmp/rsmon-worker-act-fake\n", "", 0
+	case strings.Contains(command, "base64 -d"):
+		// env/unit uploads succeed.
+		return "", "", 0
+	case strings.Contains(command, "rsmon-worker activated"):
+		// the activate script completed successfully.
+		return "", "", 0
 	default:
 		return "", "", 0
 	}
@@ -184,9 +209,25 @@ func testSSHOptions(port int) SourceInstallOptions {
 			Password:        fakeSSHPassword,
 			InsecureHostKey: true,
 		},
+		// Staging-only by default so the work-package-3 orchestration
+		// tests keep their historical shape; activation tests opt in.
+		Activation: ActivationOptions{Activate: false},
 	}
 }
 
+// testActivationOptions wraps testSSHOptions with the credentials and
+// activation flag needed to run the work-package-4 flow against the fake
+// server.
+func testActivationOptions(port int) SourceInstallOptions {
+	opts := testSSHOptions(port)
+	opts.Activation = ActivationOptions{
+		Activate: true,
+		URL:      "https://rsmon.ru",
+		Token:    fakeSSHPassword + "-token",
+	}
+	return opts
+}
+
 func TestSourceInstallSSHFlow(t *testing.T) {
 	srv := startFakeSSHServer(t, nil)
 	res, err := SourceInstall(testSSHOptions(srv.Port()))
@@ -349,3 +390,147 @@ func TestSourceInstallSSHCommandTimeout(t *testing.T) {
 		t.Fatalf("err = %v, want command timeout", err)
 	}
 }
+
+// TestSourceInstallSSHActivationFlow verifies the work-package-4 flow
+// over a real SSH session: after the staging build and commit record, the
+// installer probes the running init, uploads the env and service
+// definition, runs the atomic activation script, and reports the installed
+// layout. The secrets-absent contract holds: the worker token never
+// reaches a remote command.
+func TestSourceInstallSSHActivationFlow(t *testing.T) {
+	srv := startFakeSSHServer(t, nil)
+	res, err := SourceInstall(testActivationOptions(srv.Port()))
+	if err != nil {
+		t.Fatal(err)
+	}
+	if res.Activation == nil {
+		t.Fatal("activation result missing")
+	}
+	if res.Activation.Binary != "/usr/local/bin/rsmon-worker" ||
+		res.Activation.EnvFile != "/etc/rsmon-worker/worker.env" ||
+		res.Activation.DataDir != "/var/lib/rsmon-worker" ||
+		res.Activation.UnitFile != "/etc/systemd/system/rsmon-worker.service" ||
+		res.Activation.Supervisor != "none" ||
+		!res.Activation.Started {
+		t.Fatalf("activation result = %+v", res.Activation)
+	}
+
+	commands := srv.Commands()
+	var joined strings.Builder
+	for _, c := range commands {
+		joined.WriteString(c)
+		joined.WriteString("\n")
+	}
+	for _, want := range []string{
+		"/run/systemd/system",
+		"softlevel",
+		"mktemp -d /tmp/rsmon-worker-act",
+		"umask 077; base64 -d >",
+		"rsmon-worker activated",
+		"install -m 0755",
+		"install -m 0600",
+		"chmod 0750",
+		"worker.pid",
+		"liveness",
+		"rm -rf --",
+	} {
+		if !strings.Contains(joined.String(), want) {
+			t.Fatalf("activation commands missing %q:\n%s", want, joined.String())
+		}
+	}
+
+	// Ordering: build and commit record, then a server-side secure upload
+	// dir is created, the env/unit are uploaded into it, and the activation
+	// script runs last.
+	idx := func(sub string) int {
+		for i, c := range commands {
+			if strings.Contains(c, sub) {
+				return i
+			}
+		}
+		t.Fatalf("command %q not found in %v", sub, commands)
+		return -1
+	}
+	if idx("branch=%s") >= idx("mktemp -d /tmp/rsmon-worker-act") ||
+		idx("mktemp -d /tmp/rsmon-worker-act") >= idx("base64 -d >") ||
+		idx("base64 -d >") >= idx("rsmon-worker activated") {
+		t.Fatalf("activation order wrong: %v", commands)
+	}
+	if strings.Contains(joined.String(), fakeSSHPassword+"-token") {
+		t.Fatal("worker token leaked into a remote command")
+	}
+}
+
+// TestSourceInstallSSHNoActivationSkipsService proves the staging
+// boundary: without Activation the installer never probes the init system,
+// uploads an env/unit, or runs the activation script.
+func TestSourceInstallSSHNoActivationSkipsService(t *testing.T) {
+	srv := startFakeSSHServer(t, nil)
+	res, err := SourceInstall(testSSHOptions(srv.Port()))
+	if err != nil {
+		t.Fatal(err)
+	}
+	if res.Activation != nil {
+		t.Fatalf("activation must not run without the flag: %+v", res.Activation)
+	}
+	joined := strings.Join(srv.Commands(), "\n")
+	for _, forbidden := range []string{"softlevel", "base64 -d >", "rsmon-worker activated", "worker.pid"} {
+		if strings.Contains(joined, forbidden) {
+			t.Fatalf("staging-only flow ran activation step %q:\n%s", forbidden, joined)
+		}
+	}
+}
+
+// TestSourceInstallSSHActivationRequiresToken verifies the credentials
+// gate: activation without a token fails before any remote connection.
+func TestSourceInstallSSHActivationRequiresToken(t *testing.T) {
+	srv := startFakeSSHServer(t, nil)
+	opts := testSSHOptions(srv.Port())
+	opts.Activation = ActivationOptions{Activate: true, URL: "https://rsmon.ru"}
+	_, err := SourceInstall(opts)
+	if err == nil || !strings.Contains(err.Error(), "RSMON_TOKEN") {
+		t.Fatalf("err = %v, want token requirement", err)
+	}
+	if got := len(srv.Commands()); got != 0 {
+		t.Fatalf("commands ran before validation failed: %d", got)
+	}
+}
+
+// TestSourceInstallSSHActivationFailure verifies the bounded failure
+// path: a failing activate script surfaces as a labelled error with the
+// remote stderr, and the uploaded temp env/unit are cleaned up by the
+// controller defer regardless of the outcome.
+func TestSourceInstallSSHActivationFailure(t *testing.T) {
+	srv := startFakeSSHServer(t, func(command string) (string, string, int) {
+		if strings.Contains(command, "rsmon-worker activated") {
+			return "", "activation exploded: disk full", 1
+		}
+		return defaultFakeExec(command)
+	})
+	_, err := SourceInstall(testActivationOptions(srv.Port()))
+	if err == nil || !strings.Contains(err.Error(), "activate worker service") || !strings.Contains(err.Error(), "disk full") {
+		t.Fatalf("err = %v, want bounded activation failure", err)
+	}
+	joined := strings.Join(srv.Commands(), "\n")
+	if !strings.Contains(joined, "rm -rf --") {
+		t.Fatalf("secure upload dir cleanup not issued after activation failure:\n%s", joined)
+	}
+}
+
+// TestSourceInstallSSHActivationHealthFailure verifies the health gate
+// surfaces as an activation failure and the temp files are still cleaned.
+// The fake cannot run the shell script, so the failure is injected at the
+// activate-command level; the real rollback semantics are covered by the
+// Docker/OpenSSH E2E tests.
+func TestSourceInstallSSHActivationHealthFailure(t *testing.T) {
+	srv := startFakeSSHServer(t, func(command string) (string, string, int) {
+		if strings.Contains(command, "rsmon-worker activated") {
+			return "", "health failure: worker did not answer /healthz", 1
+		}
+		return defaultFakeExec(command)
+	})
+	_, err := SourceInstall(testActivationOptions(srv.Port()))
+	if err == nil || !strings.Contains(err.Error(), "did not answer /healthz") {
+		t.Fatalf("err = %v, want /healthz verification failure", err)
+	}
+}