1package app
2
3import (
4 "bytes"
5 "context"
6 "encoding/json"
7 "errors"
8 "io"
9 "log/slog"
10 "os"
11 "path/filepath"
12 "sort"
13 "strings"
14 "sync"
15 "sync/atomic"
16 "testing"
17 "time"
18
19 "github.com/stretchr/testify/require"
20
21 "github.com/bnema/vev/internal/adapters/brokerconfig"
22 "github.com/bnema/vev/internal/adapters/brokeripc"
23 "github.com/bnema/vev/internal/adapters/daemonmux"
24 "github.com/bnema/vev/internal/adapters/ipc"
25 "github.com/bnema/vev/internal/adapters/lifecycle"
26 "github.com/bnema/vev/internal/domain"
27 "github.com/bnema/vev/internal/ports"
28 "github.com/bnema/vev/internal/protocol"
29)
30
31const (
32 brokerTestEndpoint = "offline@daemon:2222"
33 brokerTestIdentity = "offline-daemon"
34 brokerTestWait = 5 * time.Second
35 brokerTestPollTick = 5 * time.Millisecond
36 brokerTestIdleAdvance = time.Minute
37)
38
39// brokerTestPolicy is the exact policy provisioned and offered by the fixture.
40func brokerTestPolicy() ports.BrokerPolicy {
41 return ports.BrokerPolicy{
42 ProtocolVersion: protocol.Version,
43 CatalogSchemaVersion: 3,
44 EnvironmentPolicy: protocol.EnvironmentPolicyClientOwned,
45 Transport: "unix-mux",
46 Trust: "offline-trust",
47 Launch: "offline-launch",
48 Isolation: "offline-isolation",
49 }
50}
51
52// brokerTestRegistration is the exact registration provisioned by the sandbox
53// config and carried by the request.
54func brokerTestRegistration() domain.RemoteRegistration {
55 var registration domain.RemoteRegistration
56 registration.Endpoint = brokerTestEndpoint
57 for i := range registration.Incarnation {
58 registration.Incarnation[i] = byte(i + 1)
59 }
60 registration.Generation = 1
61 return registration
62}
63
64// sandboxRegistrationDocument renders one registration document with the given
65// route (a Unix path string or an explicit route object).
66func sandboxRegistrationDocument(route any) map[string]any {
67 policy := brokerTestPolicy()
68 return map[string]any{
69 "marker": brokerconfig.Marker,
70 "registrations": []any{map[string]any{
71 "endpoint": brokerTestEndpoint,
72 "incarnation": "0102030405060708090a0b0c0d0e0f10",
73 "generation": 1,
74 "identity": brokerTestIdentity,
75 "route": route,
76 "policy": map[string]any{
77 "protocolVersion": policy.ProtocolVersion,
78 "catalogSchemaVersion": policy.CatalogSchemaVersion,
79 "environmentPolicy": "client-owned",
80 "transport": policy.Transport,
81 "trust": policy.Trust,
82 "launch": policy.Launch,
83 "isolation": policy.Isolation,
84 },
85 }},
86 }
87}
88
89// isolateSandboxEnv redirects production XDG roots into private temporary
90// directories and returns one fresh offline root beside them, so the sandbox
91// can never touch the real runtime or state paths.
92func isolateSandboxEnv(t *testing.T) (root, prodRuntime, prodState string) {
93 t.Helper()
94 prodRuntime = t.TempDir()
95 prodState = t.TempDir()
96 t.Setenv("VEV_ENV", "")
97 t.Setenv("VEV_ENV_ROOT", "")
98 t.Setenv("XDG_RUNTIME_DIR", prodRuntime)
99 t.Setenv("XDG_STATE_HOME", prodState)
100 root = filepath.Join(shortTempDir(t, "vevs"), "sandbox")
101 return root, prodRuntime, prodState
102}
103
104// testBrokerServeDeps builds sandbox deps with a deterministic clock and a
105// discard logger, and reports readiness on the returned channel.
106func testBrokerServeDeps(clk ports.Clock) (brokerServeDeps, <-chan string) {
107 deps := defaultBrokerServeDeps()
108 deps.clock = func() ports.Clock { return clk }
109 deps.newLogger = func(string) (*slog.Logger, io.Closer, error) {
110 return slog.New(slog.NewTextHandler(io.Discard, nil)), io.NopCloser(strings.NewReader("")), nil
111 }
112 ready := make(chan string, 1)
113 deps.onReady = func(socketPath string) {
114 select {
115 case ready <- socketPath:
116 default:
117 }
118 }
119 return deps, ready
120}
121
122// runSandbox starts one sandbox serve goroutine and returns its result channel.
123func runSandbox(ctx context.Context, options brokerServeOptions, deps brokerServeDeps) <-chan error {
124 done := make(chan error, 1)
125 go func() { done <- runBrokerServe(ctx, options, deps) }()
126 return done
127}
128
129// awaitSandboxCancel cancels the sandbox and requires an orderly shutdown.
130func awaitSandboxCancel(t *testing.T, cancel context.CancelFunc, done <-chan error) {
131 t.Helper()
132 cancel()
133 select {
134 case err := <-done:
135 require.NoError(t, err)
136 case <-time.After(brokerTestWait):
137 t.Fatal("sandbox did not shut down after cancellation")
138 }
139}
140
141// fatalListener is a ports.BrokerListener whose Accept always reports one
142// fixed terminal error, so a test can drive the accept drain without a carriage.
143type fatalListener struct{ err error }
144
145func (l fatalListener) Accept() (ports.BrokerCoreService, error) { return nil, l.err }
146func (l fatalListener) Close() error { return nil }
147func (l fatalListener) Addr() string { return "fatal" }
148
149// TestDrainBrokerAcceptClassifiesTerminalFailure proves the drain treats a
150// listener close as an orderly end with no log and no failure signal, while any
151// other terminal accept failure is logged and reported on the failure channel so
152// the caller can commit shutdown.
153func TestDrainBrokerAcceptClassifiesTerminalFailure(t *testing.T) {
154 tests := []struct {
155 name string
156 err error
157 wantLogged bool
158 wantFailed bool
159 }{
160 {name: "listener close is orderly and quiet", err: brokeripc.ErrListenerClosed},
161 {name: "unexpected terminal failure is logged and signaled", err: errors.New("carriage accept failed"), wantLogged: true, wantFailed: true},
162 }
163 for _, tt := range tests {
164 t.Run(tt.name, func(t *testing.T) {
165 var buffer bytes.Buffer
166 log := slog.New(slog.NewTextHandler(&buffer, nil))
167 started := make(chan struct{})
168 drained := make(chan struct{})
169 failed := make(chan struct{})
170 go drainBrokerAccept(fatalListener{err: tt.err}, log, started, drained, failed)
171
172 select {
173 case <-started:
174 case <-time.After(brokerTestWait):
175 t.Fatal("the drain never started")
176 }
177 select {
178 case <-drained:
179 case <-time.After(brokerTestWait):
180 t.Fatal("the drain did not stop on a terminal accept failure")
181 }
182
183 select {
184 case <-failed:
185 require.True(t, tt.wantFailed, "an orderly listener close must not signal failure")
186 default:
187 require.False(t, tt.wantFailed, "an unexpected terminal failure must signal shutdown")
188 }
189
190 if tt.wantLogged {
191 require.Contains(t, buffer.String(), "broker_offline_accept_failed")
192 return
193 }
194 require.Empty(t, strings.TrimSpace(buffer.String()), "an orderly listener close must not be logged")
195 })
196 }
197}
198
199func TestProductionBrokerServeRejectsUnsafeRoot(t *testing.T) {
200 layout := emptyProductionBrokerLayout(t, "")
201 require.NoError(t, os.MkdirAll(layout.Root, 0o755))
202 deps, _ := testBrokerServeDeps(newSandboxClock())
203 err := runBrokerServe(context.Background(), brokerServeOptions{}, deps)
204 require.ErrorContains(t, err, "0755")
205 require.NoFileExists(t, filepath.Join(layout.Runtime, "lifecycle.lock"))
206}
207
208func TestBrokerServeRoundTrip(t *testing.T) {
209 emptyProductionBrokerLayout(t, "")
210 policy := brokerTestPolicy()
211 fixture := newBrokerMuxFixture(t, policy)
212 writeProductionBrokerDocument(t, sandboxRegistrationDocument(fixture.route))
213
214 clk := newSandboxClock()
215 deps, ready := testBrokerServeDeps(clk)
216 ctx, cancel := context.WithCancel(context.Background())
217 defer cancel()
218 done := runSandbox(ctx, brokerServeOptions{}, deps)
219
220 socketPath := awaitSandboxReady(t, ready, done)
221
222 clientCtx, clientCancel := context.WithTimeout(context.Background(), brokerTestWait)
223 defer clientCancel()
224 service, err := brokeripc.Dial(clientCtx, socketPath, brokeripc.Config{})
225 require.NoError(t, err)
226 defer func() { _ = service.Close() }()
227
228 streamID, err := service.NextStreamID()
229 require.NoError(t, err)
230 stream, err := service.OpenStream(clientCtx, ports.BrokerOpenStreamRequest{
231 Purpose: ports.BrokerStreamControl,
232 Stream: streamID,
233 Endpoint: brokerTestEndpoint,
234 Registration: brokerTestRegistration(),
235 Policy: policy,
236 StartMode: ports.BrokerDaemonStartIfNeeded,
237 })
238 require.NoError(t, err)
239 defer func() { _ = stream.Close() }()
240
241 type pingResult struct {
242 message protocol.ServerMessage
243 err error
244 }
245 result := make(chan pingResult, 1)
246 go func() {
247 if sendErr := stream.SendClient(protocol.Ping{}); sendErr != nil {
248 result <- pingResult{err: sendErr}
249 return
250 }
251 message, recvErr := stream.ReceiveServer()
252 result <- pingResult{message: message, err: recvErr}
253 }()
254 select {
255 case outcome := <-result:
256 require.NoError(t, outcome.err)
257 require.Equal(t, protocol.Pong{}, outcome.message)
258 case <-time.After(brokerTestWait):
259 t.Fatal("typed IPC -> pool -> mux round trip timed out")
260 }
261 require.Positive(t, fixture.streams.Load())
262
263 awaitSandboxCancel(t, cancel, done)
264 requireSocketRemoved(t, socketPath)
265}
266
267// writeProductionBrokerDocument replaces only the private per-test bootstrap.
268func writeProductionBrokerDocument(t *testing.T, document any) {
269 t.Helper()
270 raw, err := json.Marshal(document)
271 require.NoError(t, err)
272 require.NoError(t, os.WriteFile(productionBrokerConfigPath(), raw, 0o600))
273}
274
275// awaitSandboxReady waits for the sandbox readiness hook and returns the
276// reported socket path.
277func awaitSandboxReady(t *testing.T, ready <-chan string, done <-chan error) string {
278 t.Helper()
279 select {
280 case socketPath := <-ready:
281 require.NotEmpty(t, socketPath)
282 return socketPath
283 case err := <-done:
284 t.Fatalf("sandbox exited before readiness: %v", err)
285 return ""
286 case <-time.After(brokerTestWait):
287 t.Fatal("sandbox never reported readiness and did not exit")
288 return ""
289 }
290}
291
292// requireSocketRemoved asserts the sandbox socket inode was unlinked.
293func requireSocketRemoved(t *testing.T, socketPath string) {
294 t.Helper()
295 _, err := os.Lstat(socketPath)
296 require.ErrorIs(t, err, os.ErrNotExist)
297}
298
299// waitForSandboxShutdown advances the manual clock until the sandbox has shut
300// down, so a real idle grace is never awaited.
301func waitForSandboxShutdown(t *testing.T, clk *sandboxClock, done <-chan error) {
302 t.Helper()
303 deadline := time.After(brokerTestWait)
304 for {
305 select {
306 case err := <-done:
307 require.NoError(t, err)
308 return
309 case <-time.After(brokerTestPollTick):
310 clk.Advance(brokerTestIdleAdvance)
311 case <-deadline:
312 t.Fatal("idle sandbox did not shut down")
313 }
314 }
315}
316
317func TestBrokerServeIdleShutdownOnEmptyConfig(t *testing.T) {
318 layout := emptyProductionBrokerLayout(t, "")
319 clk := newSandboxClock()
320
321 deps, ready := testBrokerServeDeps(clk)
322 ctx, cancel := context.WithCancel(context.Background())
323 defer cancel()
324 done := runSandbox(ctx, brokerServeOptions{}, deps)
325
326 socketPath := awaitSandboxReady(t, ready, done)
327 waitForSandboxShutdown(t, clk, done)
328
329 requireSocketRemoved(t, socketPath)
330 requireOwnerLockReleased(t, layout.Runtime)
331
332}
333
334// requireOwnerLockReleased proves the lifetime lock is no longer held.
335func requireOwnerLockReleased(t *testing.T, runtimeDir string) {
336 t.Helper()
337 owner, err := lifecycle.TryAcquire(runtimeDir)
338 require.NoError(t, err)
339 require.NoError(t, owner.Release())
340}
341
342func TestBrokerServeCancelCleanup(t *testing.T) {
343 layout := emptyProductionBrokerLayout(t, "")
344 clk := newSandboxClock()
345
346 deps, ready := testBrokerServeDeps(clk)
347 ctx, cancel := context.WithCancel(context.Background())
348 defer cancel()
349 done := runSandbox(ctx, brokerServeOptions{}, deps)
350
351 socketPath := awaitSandboxReady(t, ready, done)
352 awaitSandboxCancel(t, cancel, done)
353
354 requireSocketRemoved(t, socketPath)
355 requireOwnerLockReleased(t, layout.Runtime)
356 _, err := os.Lstat(filepath.Join(layout.State, "state.json"))
357 require.NoError(t, err, "durable broker state must remain after shutdown")
358
359}
360
361// wedgedListener is a ports.BrokerListener whose Accept always reports one
362// unexpected terminal failure and whose Close records that the sandbox tore it
363// down, so a test can prove an accept failure does not leave a bound broker.
364type wedgedListener struct {
365 err error
366 addr string
367 closed atomic.Bool
368}
369
370func (l *wedgedListener) Accept() (ports.BrokerCoreService, error) { return nil, l.err }
371func (l *wedgedListener) Close() error {
372 l.closed.Store(true)
373 return nil
374}
375func (l *wedgedListener) Addr() string { return l.addr }
376
377// TestBrokerServeShutsDownOnUnexpectedAcceptFailure proves an unexpected
378// terminal accept failure commits shutdown and drains the listener, instead of
379// leaving a bound broker whose socket can no longer admit work.
380func TestBrokerServeShutsDownOnUnexpectedAcceptFailure(t *testing.T) {
381 layout := emptyProductionBrokerLayout(t, "")
382 clk := newSandboxClock()
383
384 listener := &wedgedListener{err: errors.New("carriage accept failed"), addr: "wedged"}
385 deps, ready := testBrokerServeDeps(clk)
386 deps.listen = func(string, ports.BrokerEpoch, ports.BrokerAuthority, brokeripc.Config) (ports.BrokerListener, error) {
387 return listener, nil
388 }
389 ctx, cancel := context.WithCancel(context.Background())
390 defer cancel()
391 done := runSandbox(ctx, brokerServeOptions{}, deps)
392
393 awaitSandboxReady(t, ready, done)
394 select {
395 case err := <-done:
396 require.NoError(t, err)
397 case <-time.After(brokerTestWait):
398 t.Fatal("the broker wedged on an unexpected accept failure")
399 }
400 require.True(t, listener.closed.Load(), "shutdown must close the wedged listener")
401 requireOwnerLockReleased(t, layout.Runtime)
402
403}
404
405func TestBrokerServeRefusesDuplicateOwnership(t *testing.T) {
406 emptyProductionBrokerLayout(t, "")
407 clk := newSandboxClock()
408
409 deps, ready := testBrokerServeDeps(clk)
410 ctx, cancel := context.WithCancel(context.Background())
411 defer cancel()
412 done := runSandbox(ctx, brokerServeOptions{}, deps)
413 socketPath := awaitSandboxReady(t, ready, done)
414
415 // A second foreground owner fails fast on the lifetime lock.
416 secondDeps, _ := testBrokerServeDeps(newSandboxClock())
417 err := runBrokerServe(context.Background(), brokerServeOptions{}, secondDeps)
418 require.ErrorIs(t, err, lifecycle.ErrBusy)
419
420 // A duplicate listener on the live socket is refused as well.
421 duplicate, err := ipc.ListenMux(socketPath)
422 if duplicate != nil {
423 _ = duplicate.Close()
424 }
425 require.ErrorIs(t, err, ipc.ErrDaemonRunning)
426
427 awaitSandboxCancel(t, cancel, done)
428}
429
430// serveTestPongs answers every typed Ping of one accepted daemonmux stream with
431// exactly one Pong until the stream ends.
432func serveTestPongs(connection ports.ServerConnection) {
433 for {
434 message, err := connection.ReceiveClient()
435 if err != nil {
436 return
437 }
438 if _, ok := message.(protocol.Ping); !ok {
439 continue
440 }
441 if err := connection.SendServer(protocol.Pong{}); err != nil {
442 return
443 }
444 }
445}
446
447// brokerMuxFixture is a real Unix daemonmux server: one private mux carriage
448// adopted by a supervisor into an aggregate listener that answers typed Pings.
449// streams counts admitted typed logical streams so a test can prove the
450// request actually reached the mux fixture.
451type brokerMuxFixture struct {
452 route string
453 streams atomic.Int64
454}
455
456// newBrokerMuxFixture binds one real private Unix mux carriage and serves typed
457// Pings on every admitted logical stream.
458func newBrokerMuxFixture(t *testing.T, policy ports.BrokerPolicy) *brokerMuxFixture {
459 t.Helper()
460 routeDir := shortTempDir(t, "mux")
461 require.NoError(t, os.Chmod(routeDir, 0o700))
462 return newBrokerMuxFixtureAt(t, policy, filepath.Join(routeDir, "mux.sock"))
463}
464
465// newBrokerMuxFixtureAt binds at a caller-chosen production daemonmux path.
466func newBrokerMuxFixtureAt(t *testing.T, policy ports.BrokerPolicy, route string) *brokerMuxFixture {
467 t.Helper()
468 require.NoError(t, os.MkdirAll(filepath.Dir(route), 0o700))
469 listener, err := ipc.ListenMux(route)
470 require.NoError(t, err)
471
472 aggregate := daemonmux.NewAggregateListener()
473 var incarnation ports.BrokerDaemonIncarnation
474 for i := range incarnation {
475 incarnation[i] = byte(0x10 + i)
476 }
477 binding, err := daemonmux.NewServerBindings(ports.BrokerDaemonIdentity(brokerTestIdentity), incarnation, []daemonmux.ServerPolicyAdmission{{Policy: policy, Origin: ports.SessionOriginRemote}})
478 require.NoError(t, err)
479 supervisor, err := daemonmux.NewServerSupervisor(aggregate, binding, daemonmux.DefaultMuxCeilings(), 8)
480 require.NoError(t, err)
481
482 fixture := &brokerMuxFixture{route: route}
483 ctx, cancel := context.WithCancel(context.Background())
484 t.Cleanup(func() {
485 cancel()
486 _ = supervisor.Close()
487 _ = aggregate.Close()
488 _ = listener.Close()
489 })
490
491 go func() {
492 for {
493 raw, acceptErr := listener.Accept()
494 if acceptErr != nil {
495 return
496 }
497 go func() { _ = supervisor.Adopt(ctx, raw) }()
498 }
499 }()
500 go func() {
501 for {
502 connection, acceptErr := aggregate.Accept()
503 if acceptErr != nil {
504 return
505 }
506 fixture.streams.Add(1)
507 go serveTestPongs(connection)
508 }
509 }()
510 return fixture
511}
512
513// sandboxClock is a deterministic ports.Clock/ports.Timer pair: time only moves
514// when Advance is called, so idle shutdown is exercised without wall time.
515type sandboxClock struct {
516 mu sync.Mutex
517 now time.Time
518 order uint64
519 timers map[*sandboxTimer]struct{}
520}
521
522func newSandboxClock() *sandboxClock {
523 return &sandboxClock{now: time.Unix(0, 0), timers: make(map[*sandboxTimer]struct{})}
524}
525
526func (c *sandboxClock) Now() time.Time {
527 c.mu.Lock()
528 defer c.mu.Unlock()
529 return c.now
530}
531
532func (c *sandboxClock) NewTimer(d time.Duration) ports.Timer {
533 timer := &sandboxTimer{clock: c, ch: make(chan time.Time, 1)}
534 c.mu.Lock()
535 defer c.mu.Unlock()
536 timer.arm(d)
537 return timer
538}
539
540// Advance moves time forward and fires every timer that came due.
541func (c *sandboxClock) Advance(d time.Duration) {
542 if d < 0 {
543 panic("sandboxClock.Advance requires a non-negative duration")
544 }
545 c.mu.Lock()
546 defer c.mu.Unlock()
547 c.now = c.now.Add(d)
548 due := make([]*sandboxTimer, 0, len(c.timers))
549 for timer := range c.timers {
550 if !timer.when.After(c.now) {
551 due = append(due, timer)
552 }
553 }
554 sort.Slice(due, func(i, j int) bool {
555 if due[i].when.Equal(due[j].when) {
556 return due[i].order < due[j].order
557 }
558 return due[i].when.Before(due[j].when)
559 })
560 for _, timer := range due {
561 timer.active = false
562 delete(c.timers, timer)
563 select {
564 case timer.ch <- c.now:
565 default:
566 }
567 }
568}
569
570type sandboxTimer struct {
571 clock *sandboxClock
572 ch chan time.Time
573 when time.Time
574 order uint64
575 active bool
576}
577
578func (t *sandboxTimer) C() <-chan time.Time { return t.ch }
579
580func (t *sandboxTimer) Stop() bool {
581 t.clock.mu.Lock()
582 defer t.clock.mu.Unlock()
583 if !t.active {
584 return false
585 }
586 t.active = false
587 delete(t.clock.timers, t)
588 return true
589}
590
591func (t *sandboxTimer) Reset(d time.Duration) bool {
592 t.clock.mu.Lock()
593 defer t.clock.mu.Unlock()
594 wasActive := t.active
595 if t.active {
596 t.active = false
597 delete(t.clock.timers, t)
598 }
599 select {
600 case <-t.ch:
601 default:
602 }
603 t.arm(d)
604 return wasActive
605}
606
607func (t *sandboxTimer) arm(d time.Duration) {
608 t.order = t.clock.order
609 t.clock.order++
610 t.when = t.clock.now.Add(d)
611 t.active = true
612 t.clock.timers[t] = struct{}{}
613}
614
615func TestSandboxClockFiresOnAdvance(t *testing.T) {
616 clk := newSandboxClock()
617 timer := clk.NewTimer(time.Minute)
618 require.True(t, timer.Stop())
619 timer = clk.NewTimer(time.Minute)
620 clk.Advance(time.Minute)
621 select {
622 case <-timer.C():
623 case <-time.After(brokerTestWait):
624 t.Fatal("manual timer did not fire")
625 }
626}