1package app
2
3import (
4 "context"
5 "encoding/json"
6 "errors"
7 "net"
8 "os"
9 "path/filepath"
10 "sync/atomic"
11 "testing"
12 "time"
13
14 "github.com/stretchr/testify/require"
15
16 "github.com/bnema/vev/internal/adapters/brokerconfig"
17 "github.com/bnema/vev/internal/adapters/brokeripc"
18 "github.com/bnema/vev/internal/adapters/daemonmux"
19 "github.com/bnema/vev/internal/adapters/ipc"
20 "github.com/bnema/vev/internal/adapters/lifecycle"
21)
22
23// sandboxEmptyDocument is a marker-valid sandbox configuration with no
24// registrations and an optional provisioned idle grace.
25func sandboxEmptyDocument(idleGrace string) map[string]any {
26 document := map[string]any{"marker": brokerconfig.Marker, "registrations": []any{}}
27 if idleGrace != "" {
28 document["idleGrace"] = idleGrace
29 }
30 return document
31}
32
33// emptyProductionBrokerLayout provisions the real broker config in private XDG
34// directories; each test receives an independent runtime, state, and config.
35func emptyProductionBrokerLayout(t *testing.T, idleGrace string) brokerconfig.Layout {
36 t.Helper()
37 t.Setenv("VEV_ENV", "")
38 t.Setenv("VEV_ENV_ROOT", "")
39 t.Setenv("XDG_RUNTIME_DIR", shortTempDir(t, "vb"))
40 t.Setenv("XDG_STATE_HOME", shortTempDir(t, "vb"))
41 t.Setenv("XDG_CONFIG_HOME", shortTempDir(t, "vb"))
42 require.NoError(t, ensureProductionBrokerConfig())
43 if idleGrace != "" {
44 raw, err := json.Marshal(sandboxEmptyDocument(idleGrace))
45 require.NoError(t, err)
46 require.NoError(t, os.WriteFile(productionBrokerConfigPath(), raw, 0o600))
47 }
48 return productionBrokerLayout()
49}
50
51// productionStatusRequest provides the inputs used by production connect-or-spawn.
52func productionStatusRequest(layout brokerconfig.Layout, timeout time.Duration) brokerStatusRequest {
53 return brokerStatusRequest{
54 layout: layout, socketPath: brokeripc.SocketPath(layout.Runtime),
55 deadline: time.Now().Add(timeout), timeout: timeout,
56 }
57}
58
59// fakeStatusLock records releases for one held election.
60type fakeStatusLock struct{ releases atomic.Int32 }
61
62func (l *fakeStatusLock) Release() error {
63 l.releases.Add(1)
64 return nil
65}
66
67// readyReport is the report returned when a scripted probe succeeds.
68func readyReport(socketPath string) brokerStatusReport {
69 return brokerStatusReport{Status: "ready", Endpoint: socketPath, Epoch: 7, Revision: 3, HostCount: 2}
70}
71
72// testStatusDeps builds deterministic status seams. Virtual time starts at the
73// real instant and only advances when the loop waits, so a short timeout
74// exhausts without real sleeping while the production context deadline stays
75// comfortably in the future. The default probe always reports absence.
76func testStatusDeps(t *testing.T) (brokerStatusDeps, *statusRecorder) {
77 t.Helper()
78 rec := &statusRecorder{
79 now: time.Now(),
80 }
81 deps := brokerStatusDeps{
82 now: func() time.Time { return rec.now },
83 wait: func(ctx context.Context, d time.Duration) error {
84 rec.now = rec.now.Add(d)
85 return ctx.Err()
86 },
87 acquireElection: func(string) (brokerStatusLock, error) {
88 rec.elections.Add(1)
89 lock := &fakeStatusLock{}
90 rec.locks = append(rec.locks, lock)
91 return lock, nil
92 },
93 lifetimeOwner: func(string) (bool, error) { return false, nil },
94 spawn: func(context.Context, string, time.Duration, bool) error {
95 rec.spawns.Add(1)
96 return nil
97 },
98 }
99 rec.setProbe(&deps, func(_ int, socketPath string) (brokerStatusReport, error) {
100 return brokerStatusReport{}, absentError(socketPath)
101 })
102 return deps, rec
103}
104
105// statusRecorder captures one status run's observable effects.
106type statusRecorder struct {
107 now time.Time
108 probes atomic.Int32
109 spawns atomic.Int32
110 elections atomic.Int32
111 locks []*fakeStatusLock
112}
113
114// setProbe installs one scripted probe that sees its 1-based attempt number.
115func (r *statusRecorder) setProbe(deps *brokerStatusDeps, probe func(attempt int, socketPath string) (brokerStatusReport, error)) {
116 deps.probe = func(_ context.Context, socketPath string) (brokerStatusReport, error) {
117 return probe(int(r.probes.Add(1)), socketPath)
118 }
119}
120
121func TestEffectiveIdleGrace(t *testing.T) {
122 layout := emptyProductionBrokerLayout(t, "90s")
123 config, err := brokerconfig.LoadProduction(layout, productionBrokerConfigPath(), "", localDaemonPolicy(), daemonmux.SocketPath(ipc.SocketDir()))
124 require.NoError(t, err)
125 unprovisioned := emptyProductionBrokerLayout(t, "")
126 plain, err := brokerconfig.LoadProduction(unprovisioned, productionBrokerConfigPath(), "", localDaemonPolicy(), daemonmux.SocketPath(ipc.SocketDir()))
127 require.NoError(t, err)
128
129 tests := []struct {
130 name string
131 config *brokerconfig.Config
132 explicit time.Duration
133 want time.Duration
134 wantErr string
135 }{
136 {name: "unset uses provisioned", config: config, want: 90 * time.Second},
137 {name: "matching explicit is accepted", config: config, explicit: 90 * time.Second, want: 90 * time.Second},
138 {name: "conflicting explicit is diagnosed", config: config, explicit: time.Minute, wantErr: "conflicts with the provisioned idle grace"},
139 {name: "unprovisioned explicit wins", config: plain, explicit: 42 * time.Second, want: 42 * time.Second},
140 {name: "unprovisioned unset defers to the default", config: plain, want: 0},
141 }
142 for _, tt := range tests {
143 t.Run(tt.name, func(t *testing.T) {
144 got, err := effectiveIdleGrace(tt.config, tt.explicit)
145 if tt.wantErr != "" {
146 require.Error(t, err)
147 require.Contains(t, err.Error(), tt.wantErr)
148 return
149 }
150 require.NoError(t, err)
151 require.Equal(t, tt.want, got)
152 })
153 }
154}
155
156func TestBrokerServeRefusesConflictingIdleGrace(t *testing.T) {
157 layout := emptyProductionBrokerLayout(t, "90s")
158 ctx, cancel := context.WithCancel(context.Background())
159 defer cancel()
160 deps, _ := testBrokerServeDeps(newSandboxClock())
161 err := runBrokerServe(ctx, brokerServeOptions{idleGrace: time.Minute}, deps)
162 require.Error(t, err)
163 require.Contains(t, err.Error(), "conflicts with the provisioned idle grace")
164 require.NoFileExists(t, filepath.Join(layout.Runtime, "lifecycle.lock"), "a refused serve must not take the lifetime lock")
165}
166
167func TestBrokerLauncherRefusesConflictingIdleGraceBeforeSpawning(t *testing.T) {
168 layout := emptyProductionBrokerLayout(t, "90s")
169 config, err := brokerconfig.LoadProduction(layout, productionBrokerConfigPath(), "", localDaemonPolicy(), daemonmux.SocketPath(ipc.SocketDir()))
170 require.NoError(t, err)
171 _, err = effectiveIdleGrace(config, time.Minute)
172 require.Error(t, err)
173 require.Contains(t, err.Error(), "conflicts with the provisioned idle grace")
174}
175
176func TestBrokerStatusEnsureSpawnsOnceThenReportsReady(t *testing.T) {
177 layout := emptyProductionBrokerLayout(t, "")
178 deps, rec := testStatusDeps(t)
179 rec.setProbe(&deps, func(_ int, socketPath string) (brokerStatusReport, error) {
180 if rec.spawns.Load() > 0 {
181 return readyReport(socketPath), nil
182 }
183 return brokerStatusReport{}, absentError(socketPath)
184 })
185 report, err := ensureBrokerReady(context.Background(), productionStatusRequest(layout, time.Second), deps)
186 require.NoError(t, err)
187
188 require.Equal(t, int32(1), rec.spawns.Load(), "exactly one spawn is expected")
189 require.Equal(t, int32(1), rec.elections.Load(), "exactly one election is expected")
190 require.Len(t, rec.locks, 1)
191 require.Equal(t, int32(1), rec.locks[0].releases.Load(), "the election must be released after readiness")
192 require.Equal(t, "ready", report.Status)
193}
194
195func TestBrokerStatusEnsureWaitsForAnotherSpawner(t *testing.T) {
196 layout := emptyProductionBrokerLayout(t, "")
197 var readyAfter int32 = 4
198 deps, rec := testStatusDeps(t)
199 rec.setProbe(&deps, func(attempt int, socketPath string) (brokerStatusReport, error) {
200 if int32(attempt) >= readyAfter {
201 return readyReport(socketPath), nil
202 }
203 return brokerStatusReport{}, absentError(socketPath)
204 })
205 deps.acquireElection = func(string) (brokerStatusLock, error) {
206 rec.elections.Add(1)
207 return nil, lifecycle.ErrBusy
208 }
209 report, err := ensureBrokerReady(context.Background(), productionStatusRequest(layout, time.Second), deps)
210 require.NoError(t, err)
211
212 require.Zero(t, rec.spawns.Load(), "a waiter must never spawn while another caller is elected")
213 require.Positive(t, rec.elections.Load())
214 require.Equal(t, "ready", report.Status)
215}
216
217func TestBrokerStatusEnsureDefersToLifetimeOwnerBeforeSpawning(t *testing.T) {
218 layout := emptyProductionBrokerLayout(t, "")
219 var ownerCalls int32
220 deps, rec := testStatusDeps(t)
221 rec.setProbe(&deps, func(_ int, socketPath string) (brokerStatusReport, error) {
222 if rec.spawns.Load() > 0 {
223 return readyReport(socketPath), nil
224 }
225 return brokerStatusReport{}, absentError(socketPath)
226 })
227 deps.lifetimeOwner = func(string) (bool, error) {
228 if atomic.AddInt32(&ownerCalls, 1) <= 3 {
229 return true, nil
230 }
231 return false, nil
232 }
233 _, err := ensureBrokerReady(context.Background(), productionStatusRequest(layout, time.Second), deps)
234 require.NoError(t, err)
235 require.Equal(t, int32(1), rec.spawns.Load(), "a live lifetime owner must suppress the spawn until it is gone")
236 require.Equal(t, int32(1), rec.elections.Load(), "the election is taken only after the live owner releases the broker")
237}
238
239func TestBrokerStatusEnsureRecoversAfterSpawnerDeath(t *testing.T) {
240 layout := emptyProductionBrokerLayout(t, "")
241 var electionAttempts int32
242 deps, rec := testStatusDeps(t)
243 rec.setProbe(&deps, func(_ int, socketPath string) (brokerStatusReport, error) {
244 if rec.spawns.Load() > 0 {
245 return readyReport(socketPath), nil
246 }
247 return brokerStatusReport{}, absentError(socketPath)
248 })
249 deps.acquireElection = func(string) (brokerStatusLock, error) {
250 rec.elections.Add(1)
251 if atomic.AddInt32(&electionAttempts, 1) <= 3 {
252 // The previous elected spawner is still alive (or mid-startup).
253 return nil, lifecycle.ErrBusy
254 }
255 lock := &fakeStatusLock{}
256 rec.locks = append(rec.locks, lock)
257 return lock, nil
258 }
259 report, err := ensureBrokerReady(context.Background(), productionStatusRequest(layout, time.Second), deps)
260 require.NoError(t, err)
261 require.Equal(t, int32(1), rec.spawns.Load(), "a waiter must take over once the elected spawner dies")
262 require.Equal(t, "ready", report.Status)
263}
264
265func TestBrokerStatusEnsureIncompatibleEndpointFailsWithoutSpawning(t *testing.T) {
266 layout := emptyProductionBrokerLayout(t, "")
267 deps, rec := testStatusDeps(t)
268 rec.setProbe(&deps, func(_ int, socketPath string) (brokerStatusReport, error) {
269 return brokerStatusReport{}, incompatibleError(socketPath, errors.New("foreign endpoint"))
270 })
271 _, err := ensureBrokerReady(context.Background(), productionStatusRequest(layout, time.Second), deps)
272 require.ErrorIs(t, err, errBrokerIncompatible)
273 require.Zero(t, rec.spawns.Load())
274 require.Zero(t, rec.elections.Load())
275}
276
277func TestBrokerStatusEnsureBoundedTimeoutReportsOffline(t *testing.T) {
278 layout := emptyProductionBrokerLayout(t, "")
279 deps, rec := testStatusDeps(t)
280 _, err := ensureBrokerReady(context.Background(), productionStatusRequest(layout, 50*time.Millisecond), deps)
281 require.ErrorIs(t, err, errBrokerNotReady)
282 require.Equal(t, int32(1), rec.spawns.Load())
283}
284
285func TestBrokerStatusEnsureSpawnFailureIsReported(t *testing.T) {
286 layout := emptyProductionBrokerLayout(t, "")
287 spawnErr := errors.New("launcher exec failed")
288 deps, _ := testStatusDeps(t)
289 deps.spawn = func(context.Context, string, time.Duration, bool) error {
290 return spawnErr
291 }
292 _, err := ensureBrokerReady(context.Background(), productionStatusRequest(layout, time.Second), deps)
293 require.ErrorIs(t, err, spawnErr)
294}
295
296// A corrupt production broker.json is rejected before broker ownership.
297func TestBrokerEnsureProductionConfigFailure(t *testing.T) {
298 layout := emptyProductionBrokerLayout(t, "")
299 require.NoError(t, os.WriteFile(productionBrokerConfigPath(), []byte("not json"), 0o600))
300 deps, _ := testBrokerServeDeps(newSandboxClock())
301 err := runBrokerServe(context.Background(), brokerServeOptions{}, deps)
302 require.Error(t, err)
303 require.NoFileExists(t, filepath.Join(layout.Runtime, "lifecycle.lock"))
304}
305
306// TestBrokerStatusEnsureSecuresRootBeforeSpawning rejects an insecure
307// production spawn directory before electing a launcher.
308func TestBrokerStatusEnsureSecuresRootBeforeSpawning(t *testing.T) {
309 layout := emptyProductionBrokerLayout(t, "")
310 require.NoError(t, os.MkdirAll(layout.Runtime, 0o700))
311 require.NoError(t, os.MkdirAll(layout.Spawn, 0o700))
312 require.NoError(t, os.Chmod(layout.Spawn, 0o755))
313 deps, rec := testStatusDeps(t)
314 _, err := ensureBrokerReady(context.Background(), productionStatusRequest(layout, time.Second), deps)
315 require.Error(t, err)
316 require.Contains(t, err.Error(), "0755")
317 require.Zero(t, rec.spawns.Load())
318 require.Zero(t, rec.elections.Load())
319 require.NoFileExists(t, filepath.Join(layout.Spawn, "lifecycle.lock"), "an insecure spawn directory must be refused before election")
320}
321
322// TestBrokerStatusEnsureDeadlineBoundsSpawnWait proves the overall ensure
323// deadline bounds a launcher spawn that never returns, so a blocking launcher
324// cannot wedge connect-or-spawn.
325func TestBrokerStatusEnsureDeadlineBoundsSpawnWait(t *testing.T) {
326 layout := emptyProductionBrokerLayout(t, "")
327 deps, _ := testStatusDeps(t)
328 deps.spawn = func(ctx context.Context, _ string, _ time.Duration, _ bool) error {
329 <-ctx.Done()
330 return ctx.Err()
331 }
332 start := time.Now()
333 _, err := ensureBrokerReady(context.Background(), productionStatusRequest(layout, 200*time.Millisecond), deps)
334 require.ErrorIs(t, err, context.DeadlineExceeded)
335 require.Less(t, time.Since(start), 5*time.Second, "the overall deadline must bound the launcher spawn wait")
336}
337
338func TestBrokerStatusEnsureHonorsCancellation(t *testing.T) {
339 layout := emptyProductionBrokerLayout(t, "")
340 deps, _ := testStatusDeps(t)
341 ctx, cancel := context.WithCancel(context.Background())
342 cancel()
343 _, err := ensureBrokerReady(ctx, productionStatusRequest(layout, time.Second), deps)
344 require.ErrorIs(t, err, context.Canceled)
345}
346
347func TestProbeBrokerStatusClassifiesAbsenceAndIncompatibility(t *testing.T) {
348 layout := emptyProductionBrokerLayout(t, "")
349 socketPath := brokeripc.SocketPath(layout.Runtime)
350
351 t.Run("missing endpoint is absence", func(t *testing.T) {
352 _, err := probeBrokerStatus(context.Background(), socketPath)
353 require.ErrorIs(t, err, errBrokerAbsent)
354 })
355
356 t.Run("foreign path is incompatible", func(t *testing.T) {
357 require.NoError(t, os.MkdirAll(layout.Runtime, 0o700))
358 require.NoError(t, os.WriteFile(socketPath, []byte("not a socket"), 0o600))
359 _, err := probeBrokerStatus(context.Background(), socketPath)
360 require.ErrorIs(t, err, errBrokerIncompatible)
361 })
362
363 t.Run("stale socket is absence", func(t *testing.T) {
364 require.NoError(t, os.Remove(socketPath))
365 bindStaleUnixSocket(t, socketPath)
366 _, err := probeBrokerStatus(context.Background(), socketPath)
367 require.ErrorIs(t, err, errBrokerAbsent)
368 })
369
370 t.Run("live foreign endpoint is incompatible", func(t *testing.T) {
371 require.NoError(t, os.Remove(socketPath))
372 listener := bindUnixSocket(t, socketPath)
373 go func() {
374 for {
375 conn, err := listener.Accept()
376 if err != nil {
377 return
378 }
379 _ = conn.Close()
380 }
381 }()
382 _, err := probeBrokerStatus(context.Background(), socketPath)
383 require.ErrorIs(t, err, errBrokerIncompatible)
384 })
385}
386
387func TestProbeBrokerStatusSilentRegistrationIsBounded(t *testing.T) {
388 layout := emptyProductionBrokerLayout(t, "")
389 socketPath := brokeripc.SocketPath(layout.Runtime)
390 require.NoError(t, os.MkdirAll(layout.Runtime, 0o700))
391 // A listener that never reads or answers: the dial succeeds and the broker
392 // handshake stalls until the probe budget expires.
393 _ = bindUnixSocket(t, socketPath)
394
395 ctx, cancel := context.WithTimeout(context.Background(), 150*time.Millisecond)
396 defer cancel()
397 start := time.Now()
398 _, err := probeBrokerStatus(ctx, socketPath)
399 require.Error(t, err)
400 require.False(t, errors.Is(err, errBrokerAbsent), "a stalled handshake must not be classified as absence")
401 require.Less(t, time.Since(start), 5*time.Second)
402}
403
404// bindUnixSocket binds one real Unix listener at path.
405func bindUnixSocket(t *testing.T, path string) net.Listener {
406 t.Helper()
407 listener, err := net.Listen("unix", path)
408 require.NoError(t, err)
409 t.Cleanup(func() { _ = listener.Close() })
410 return listener
411}
412
413// bindStaleUnixSocket leaves a socket file behind with no live listener, exactly
414// like an owner that died without unlinking its endpoint.
415func bindStaleUnixSocket(t *testing.T, path string) {
416 t.Helper()
417 listener, err := net.Listen("unix", path)
418 require.NoError(t, err)
419 if unix, ok := listener.(*net.UnixListener); ok {
420 unix.SetUnlinkOnClose(false)
421 }
422 require.NoError(t, listener.Close())
423 require.FileExists(t, path)
424}