broker_offline_status_subprocess_test.go RawHistoryFolder
1package app
2
3import (
4 "bytes"
5 "context"
6 "errors"
7 "fmt"
8 "os"
9 "os/exec"
10 "path/filepath"
11 "strconv"
12 "strings"
13 "sync"
14 "syscall"
15 "testing"
16 "time"
17
18 "github.com/stretchr/testify/require"
19
20 "github.com/bnema/vev/internal/adapters/brokerconfig"
21 "github.com/bnema/vev/internal/adapters/brokeripc"
22 "github.com/bnema/vev/internal/adapters/lifecycle"
23)
24
25// Real-subprocess coverage for the hidden launcher and status entry points.
26//
27// These tests re-execute the test binary through the production dispatch path:
28// TestMain intercepts the hidden broker commands when brokerHelperEnv is set and
29// runs app.Run, so `_broker-status` really elects a spawner, `_broker-launcher`
30// really double-forks, and `_broker-serve` really composes the offline broker.
31// Every helper records its PID under brokerHelperRecordDirEnv, so a test can
32// observe how many processes were spawned and terminate every one of them.
33
34const (
35 brokerHelperEnv = "VEV_BROKER_HELPER"
36 brokerHelperRecordDirEnv = "VEV_BROKER_HELPER_RECORD_DIR"
37)
38
39// Controlled-launcher test env: when set, the hidden `_broker-launcher` helper
40// records its PID and waits for the release marker instead of running the real
41// double-fork, so a test can hold a launcher open and prove the spawn wait is
42// bounded by its context.
43const (
44 brokerLauncherBlockRecordEnv = "VEV_BROKER_LAUNCHER_BLOCK_RECORD"
45 brokerLauncherBlockReleaseEnv = "VEV_BROKER_LAUNCHER_BLOCK_RELEASE"
46)
47
48// recordBrokerHelperProcess appends one helper process's PID, so a test can
49// discover and later terminate a detached broker or launcher.
50func recordBrokerHelperProcess(role string) {
51 dir := os.Getenv(brokerHelperRecordDirEnv)
52 if dir == "" {
53 return
54 }
55 file, err := os.OpenFile(filepath.Join(dir, role+".pids"), os.O_APPEND|os.O_CREATE|os.O_WRONLY, 0o600)
56 if err != nil {
57 return
58 }
59 defer func() { _ = file.Close() }()
60 _, _ = fmt.Fprintf(file, "%d\n", os.Getpid())
61}
62
63// shortTempDir creates one temporary directory with a short absolute path.
64//
65// The AF_UNIX pathname limit is about 103 bytes, and t.TempDir embeds the full
66// test name; the broker socket lives several components beneath the sandbox
67// root, so a long test name would overflow the socket path. A short base keeps
68// the real subprocess tests honest. It uses /tmp because macOS TMPDIR is
69// already a long /var/folders path.
70func shortTempDir(t *testing.T, prefix string) string {
71 t.Helper()
72 dir, err := os.MkdirTemp("/tmp", prefix)
73 require.NoError(t, err)
74 t.Cleanup(func() { _ = os.RemoveAll(dir) })
75 return dir
76}
77
78// brokerSubprocessSandbox isolates one real offline root and records every
79// spawned helper process.
80type brokerSubprocessSandbox struct {
81 t *testing.T
82 layout brokerconfig.Layout
83 recordDir string
84}
85
86// newBrokerSubprocessSandbox isolates XDG roots, writes one marker-valid empty
87// sandbox, and arranges cleanup of every recorded broker and launcher process.
88func newBrokerSubprocessSandbox(t *testing.T, idleGrace string) *brokerSubprocessSandbox {
89 t.Helper()
90 layout := emptyProductionBrokerLayout(t, idleGrace)
91 recordDir := filepath.Join(shortTempDir(t, "vevr"), "r")
92 require.NoError(t, os.MkdirAll(recordDir, 0o700))
93 t.Setenv(brokerHelperEnv, "1")
94 t.Setenv(brokerHelperRecordDirEnv, recordDir)
95 sandbox := &brokerSubprocessSandbox{t: t, layout: layout, recordDir: recordDir}
96 t.Cleanup(sandbox.cleanup)
97 return sandbox
98}
99
100// ensure uses the production connect-or-spawn entry point and observes the
101// registered epoch after the broker publishes its initial snapshot.
102func (s *brokerSubprocessSandbox) ensure(timeout time.Duration) (brokerStatusReport, error) {
103 ctx, cancel := context.WithTimeout(context.Background(), timeout)
104 defer cancel()
105 service, err := connectProductionBroker(ctx)
106 if err != nil {
107 return brokerStatusReport{}, err
108 }
109 defer service.Close()
110 return probeBrokerStatus(ctx, s.socketPath())
111}
112
113// runBrokerCommand executes one helper through the test binary.
114func runBrokerCommand(t *testing.T, args ...string) (stdout, stderr string, code int) {
115 t.Helper()
116 cmd := exec.Command(os.Args[0], args...)
117 cmd.Env = os.Environ()
118 var out, errb bytes.Buffer
119 cmd.Stdout = &out
120 cmd.Stderr = &errb
121 err := cmd.Run()
122 if err != nil {
123 var exitErr *exec.ExitError
124 if !errors.As(err, &exitErr) {
125 t.Fatalf("run broker command %v: %v", args, err)
126 }
127 code = exitErr.ExitCode()
128 }
129 return out.String(), errb.String(), code
130}
131
132// socketPath is the broker endpoint inside the sandbox runtime directory.
133func (s *brokerSubprocessSandbox) socketPath() string {
134 return brokeripc.SocketPath(s.layout.Runtime)
135}
136
137// recorded returns every PID a role recorded, in spawn order.
138func (s *brokerSubprocessSandbox) recorded(role string) []int {
139 s.t.Helper()
140 raw, err := os.ReadFile(filepath.Join(s.recordDir, role+".pids"))
141 if errors.Is(err, os.ErrNotExist) {
142 return nil
143 }
144 require.NoError(s.t, err)
145 var pids []int
146 for _, field := range strings.Fields(string(raw)) {
147 pid, convErr := strconv.Atoi(field)
148 require.NoError(s.t, convErr)
149 pids = append(pids, pid)
150 }
151 return pids
152}
153
154// live returns the recorded PIDs that still exist.
155func (s *brokerSubprocessSandbox) live(role string) []int {
156 s.t.Helper()
157 var live []int
158 for _, pid := range s.recorded(role) {
159 if err := syscall.Kill(pid, 0); err == nil || errors.Is(err, syscall.EPERM) {
160 live = append(live, pid)
161 }
162 }
163 return live
164}
165
166// terminate signals every recorded PID of one role and waits for each exit.
167func (s *brokerSubprocessSandbox) terminate(role string, signal syscall.Signal) {
168 s.t.Helper()
169 for _, pid := range s.recorded(role) {
170 if err := syscall.Kill(pid, signal); err != nil && !errors.Is(err, syscall.ESRCH) {
171 s.t.Errorf("signal %v pid %d: %v", signal, pid, err)
172 continue
173 }
174 require.NoError(s.t, waitForProcessExit(pid, 10*time.Second))
175 }
176}
177
178// waitForSocketGone waits until the broker endpoint no longer exists.
179func (s *brokerSubprocessSandbox) waitForSocketGone(timeout time.Duration) bool {
180 deadline := time.Now().Add(timeout)
181 for time.Now().Before(deadline) {
182 if _, err := os.Lstat(s.socketPath()); errors.Is(err, os.ErrNotExist) {
183 return true
184 }
185 time.Sleep(20 * time.Millisecond)
186 }
187 return false
188}
189
190// cleanup terminates every recorded broker and launcher so no test leaks an
191// orphan process.
192func (s *brokerSubprocessSandbox) cleanup() {
193 for _, role := range []string{productionBrokerServeCommand, productionBrokerLauncherCommand} {
194 for _, pid := range s.recorded(role) {
195 if err := syscall.Kill(pid, syscall.SIGKILL); err != nil && !errors.Is(err, syscall.ESRCH) {
196 continue
197 }
198 _ = waitForProcessExit(pid, 10*time.Second)
199 }
200 }
201}
202
203// TestBrokerStatusSubprocessElectsExactlyOneBroker races several real status
204// processes against one socket: exactly one broker process, one lifetime owner,
205// and one epoch must result.
206func TestBrokerStatusSubprocessElectsExactlyOneBroker(t *testing.T) {
207 sandbox := newBrokerSubprocessSandbox(t, "5s")
208 const racers = 4
209 type outcome struct {
210 report brokerStatusReport
211 err error
212 }
213 results := make([]outcome, racers)
214 var wg sync.WaitGroup
215 for i := 0; i < racers; i++ {
216 wg.Add(1)
217 go func(index int) {
218 defer wg.Done()
219 results[index].report, results[index].err = sandbox.ensure(20 * time.Second)
220 }(i)
221 }
222 wg.Wait()
223 epochs := make(map[uint64]struct{}, racers)
224 for _, result := range results {
225 require.NoError(t, result.err)
226 require.Equal(t, "ready", result.report.Status)
227 require.Equal(t, sandbox.socketPath(), result.report.Endpoint)
228 epochs[uint64(result.report.Epoch)] = struct{}{}
229 }
230 require.Len(t, epochs, 1)
231 require.Len(t, sandbox.recorded(productionBrokerServeCommand), 1)
232 require.Len(t, sandbox.live(productionBrokerServeCommand), 1)
233}
234
235// TestBrokerStatusSubprocessIdleExitRestartsWithNewEpoch lets the broker idle
236// out, then restarts it on demand and proves the new process samples a fresh
237// epoch.
238func TestBrokerStatusSubprocessIdleExitRestartsWithNewEpoch(t *testing.T) {
239 sandbox := newBrokerSubprocessSandbox(t, "2s")
240
241 first, err := sandbox.ensure(20 * time.Second)
242 require.NoError(t, err)
243 require.Equal(t, "ready", first.Status)
244 require.Positive(t, first.Epoch)
245
246 require.True(t, sandbox.waitForSocketGone(15*time.Second), "the idle broker must shut itself down")
247 for _, pid := range sandbox.recorded(productionBrokerServeCommand) {
248 require.NoError(t, waitForProcessExit(pid, 10*time.Second), "the idle broker must exit")
249 }
250 require.Empty(t, sandbox.live(productionBrokerServeCommand), "the idle broker must exit")
251
252 second, err := sandbox.ensure(20 * time.Second)
253 require.NoError(t, err)
254 require.Equal(t, "ready", second.Status)
255 require.NotEqual(t, first.Epoch, second.Epoch, "an on-demand restart must sample a new epoch")
256 require.Len(t, sandbox.recorded(productionBrokerServeCommand), 2, "one broker per ensure is expected")
257 require.Len(t, sandbox.live(productionBrokerServeCommand), 1, "only the restarted broker may remain")
258}
259
260// TestBrokerStatusSubprocessRecoversStaleSocket proves a stale socket left by a
261// dead owner is recoverable absence, not an incompatible endpoint.
262func TestBrokerStatusSubprocessRecoversStaleSocket(t *testing.T) {
263 sandbox := newBrokerSubprocessSandbox(t, "3s")
264 require.NoError(t, os.MkdirAll(sandbox.layout.Runtime, 0o700))
265 bindStaleUnixSocket(t, sandbox.socketPath())
266
267 report, err := sandbox.ensure(20 * time.Second)
268 require.NoError(t, err)
269 require.Equal(t, "ready", report.Status)
270 require.Len(t, sandbox.live(productionBrokerServeCommand), 1)
271}
272
273// TestBrokerStatusSubprocessWaitsForLiveSpawnerThenRecovers holds the
274// descriptor-backed election lock like a spawner that died mid-startup: the
275// waiter must not spawn, and must take over once the lock is released.
276func TestBrokerStatusSubprocessWaitsForLiveSpawnerThenRecovers(t *testing.T) {
277 sandbox := newBrokerSubprocessSandbox(t, "3s")
278
279 spawner, err := lifecycle.TryAcquire(sandbox.layout.Spawn)
280 require.NoError(t, err)
281
282 _, err = sandbox.ensure(500 * time.Millisecond)
283 require.Error(t, err)
284 require.Empty(t, sandbox.recorded(productionBrokerServeCommand), "a waiter must not spawn while another spawner is live")
285
286 // The kernel releases the spawner's descriptor when it dies; releasing here
287 // is exactly that event.
288 require.NoError(t, spawner.Release())
289 report, err := sandbox.ensure(20 * time.Second)
290 require.NoError(t, err)
291 require.Equal(t, "ready", report.Status)
292 require.Len(t, sandbox.live(productionBrokerServeCommand), 1)
293}
294
295// TestBrokerStatusSubprocessSilentEndpointFailsWithoutSpawning proves a live
296// endpoint that never answers the broker handshake is diagnosed as
297// incompatible under the probe budget, and never respawned over.
298func TestBrokerStatusSubprocessSilentEndpointFailsWithoutSpawning(t *testing.T) {
299 sandbox := newBrokerSubprocessSandbox(t, "3s")
300 require.NoError(t, os.MkdirAll(sandbox.layout.Runtime, 0o700))
301 // A listener that accepts and never answers: the broker preamble stalls.
302 _ = bindUnixSocket(t, sandbox.socketPath())
303
304 // The timeout must exceed the probe budget so the stall is classified as a
305 // live-but-incompatible endpoint instead of exhausting the overall deadline.
306 start := time.Now()
307 _, err := sandbox.ensure(5 * time.Second)
308 require.ErrorContains(t, err, "live but incompatible")
309 require.Less(t, time.Since(start), 5*time.Second, "the probe budget must classify a silent endpoint before the overall deadline")
310 require.Empty(t, sandbox.recorded(productionBrokerServeCommand), "a live-but-silent endpoint must not be respawned over")
311}
312
313// TestBrokerLauncherSpawnWaitIsBoundedByContext proves a controlled blocking
314// launcher is terminated when the spawn context is cancelled, so the overall
315// deadline/signal bounds the launcher wait. The helper never starts a detached
316// serve, so this is purely the bounded-wait half; other subprocess tests prove a
317// successful launcher's detached serve survives.
318func TestBrokerLauncherSpawnWaitIsBoundedByContext(t *testing.T) {
319 record := filepath.Join(shortTempDir(t, "vevl"), "launcher.pid")
320 release := filepath.Join(t.TempDir(), "release")
321 t.Setenv(brokerHelperEnv, "1")
322 t.Setenv(brokerLauncherBlockRecordEnv, record)
323 t.Setenv(brokerLauncherBlockReleaseEnv, release)
324 t.Cleanup(func() { _ = os.WriteFile(release, nil, 0o600) })
325
326 ctx, cancel := context.WithCancel(context.Background())
327 defer cancel()
328 result := make(chan error, 1)
329 go func() { result <- spawnProductionBrokerLauncher(ctx) }()
330
331 pid := waitForProcessRecord(t, record)
332 select {
333 case err := <-result:
334 t.Fatalf("spawnBrokerLauncher returned while the controlled launcher %d was still running: %v", pid, err)
335 default:
336 }
337
338 cancel()
339 select {
340 case err := <-result:
341 require.Error(t, err, "cancelling the context must abort the launcher wait")
342 case <-time.After(5 * time.Second):
343 t.Fatal("cancelling the context must bound the launcher spawn wait")
344 }
345 require.NoError(t, waitForProcessExit(pid, 5*time.Second), "CommandContext must terminate a blocking launcher")
346}
347
348// TestBrokerStatusSubprocessLauncherDiagnosesConflicts proves an explicit
349// launcher grace that conflicts with the provisioned one is refused before any
350// process is spawned, and that a missing configuration fails the launcher.
351// A production launcher accepts no idle override; malformed broker.json is
352// diagnosed by the detached serve before it can claim the lifetime lock.
353func TestBrokerStatusSubprocessLauncherDiagnosesConflicts(t *testing.T) {
354 t.Run("unsupported idle override", func(t *testing.T) {
355 sandbox := newBrokerSubprocessSandbox(t, "90s")
356 _, stderr, code := runBrokerCommand(t, productionBrokerLauncherCommand, "--idle-grace", "1m")
357 require.NotZero(t, code)
358 require.Contains(t, stderr, "does not accept arguments")
359 require.Empty(t, sandbox.recorded(productionBrokerServeCommand))
360 })
361 t.Run("invalid production configuration", func(t *testing.T) {
362 sandbox := newBrokerSubprocessSandbox(t, "")
363 require.NoError(t, os.WriteFile(productionBrokerConfigPath(), []byte("not json"), 0o600))
364 _, stderr, code := runBrokerCommand(t, productionBrokerServeCommand)
365 require.NotZero(t, code)
366 require.Contains(t, stderr, brokerconfig.ConfigFileName)
367 require.NoFileExists(t, filepath.Join(sandbox.layout.Runtime, "lifecycle.lock"))
368 })
369}
370
371// TestBrokerStatusSubprocessNoOrphanProcess proves the launcher exits on its
372// own, one broker remains, and terminating it leaves no recorded process live.
373func TestBrokerStatusSubprocessNoOrphanProcess(t *testing.T) {
374 sandbox := newBrokerSubprocessSandbox(t, "5s")
375
376 report, err := sandbox.ensure(20 * time.Second)
377 require.NoError(t, err)
378 require.Equal(t, "ready", report.Status)
379
380 require.NotEmpty(t, sandbox.recorded(productionBrokerLauncherCommand))
381 for _, pid := range sandbox.recorded(productionBrokerLauncherCommand) {
382 require.NoError(t, waitForProcessExit(pid, 5*time.Second), "the short-lived launcher must exit")
383 }
384 require.Len(t, sandbox.live(productionBrokerServeCommand), 1)
385
386 _, err = lifecycle.TryAcquire(sandbox.layout.Runtime)
387 require.ErrorIs(t, err, lifecycle.ErrBusy, "the live broker must own the lifetime lock")
388
389 sandbox.terminate(productionBrokerServeCommand, syscall.SIGTERM)
390 require.Empty(t, sandbox.live(productionBrokerLauncherCommand), "no launcher may be left behind")
391 require.Empty(t, sandbox.live(productionBrokerServeCommand), "no broker may be left behind")
392 require.True(t, sandbox.waitForSocketGone(5*time.Second), "an orderly broker shutdown must unlink its socket")
393 owner, err := lifecycle.TryAcquire(sandbox.layout.Runtime)
394 require.NoError(t, err, "the terminated broker must release the lifetime lock")
395 require.NoError(t, owner.Release())
396}
397
398// recordedPIDsIn reads one role's PID record from an arbitrary directory.
399func recordedPIDsIn(t *testing.T, dir, role string) []int {
400 t.Helper()
401 raw, err := os.ReadFile(filepath.Join(dir, role+".pids"))
402 if errors.Is(err, os.ErrNotExist) {
403 return nil
404 }
405 require.NoError(t, err)
406 var pids []int
407 for _, field := range strings.Fields(string(raw)) {
408 pid, convErr := strconv.Atoi(field)
409 require.NoError(t, convErr)
410 pids = append(pids, pid)
411 }
412 return pids
413}