1package app
2
3// Hidden `_broker-ready` dial-only probe contract (Plan 001 P3.4).
4//
5// The probe selects one broker IPC endpoint, dials it, registers, subscribes,
6// and reads complete snapshots. These tests pin the whole caller contract
7// without a production broker process: argument parsing, the bounded JSON
8// document, the closed exit-code table, the readiness evaluation of both
9// `--require` modes, the terminal classification that must never retry, the
10// reconnect on a new broker epoch, and the zero-effect guarantee (no ensure, no
11// spawn, no logical stream, no reconcile, and no membership mutation).
12
13import (
14 "bytes"
15 "context"
16 "encoding/json"
17 "errors"
18 "net"
19 "os"
20 "path/filepath"
21 "strconv"
22 "strings"
23 "sync"
24 "syscall"
25 "testing"
26 "time"
27
28 "github.com/stretchr/testify/require"
29
30 "github.com/bnema/vev/internal/adapters/brokeripc"
31 "github.com/bnema/vev/internal/adapters/ipc"
32 "github.com/bnema/vev/internal/domain"
33 "github.com/bnema/vev/internal/ports"
34 "github.com/bnema/vev/internal/protocol"
35)
36
37const (
38 readyTestEpoch = ports.BrokerEpoch(0x2a)
39 readyTestRevision = ports.BrokerRevision(7)
40 readyTestIdentity = ports.BrokerDaemonIdentity("ready-local-daemon")
41 readyTestObserveMax = 60 * time.Millisecond
42)
43
44// readyTestIncarnate is one valid, non-zero daemon incarnation.
45var readyTestIncarnate = ports.BrokerDaemonIncarnation{1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13, 14, 15, 16}
46
47// readyTestPolicy is one valid exactly-provisioned connection policy.
48func readyTestPolicy() ports.BrokerPolicy {
49 return ports.BrokerPolicy{
50 ProtocolVersion: protocol.Version,
51 CatalogSchemaVersion: 3,
52 EnvironmentPolicy: protocol.EnvironmentPolicyDaemonOwned,
53 Transport: "unix-mux",
54 Trust: "same-user",
55 Launch: "explicit",
56 Isolation: "per-user",
57 }
58}
59
60// readyTestLocal builds one local daemon observation with the supplied observed
61// state, so a test states exactly which readiness input it varies.
62func readyTestLocal(mutate func(*ports.BrokerDaemonObservation)) ports.BrokerDaemonObservation {
63 local := ports.BrokerDaemonObservation{
64 Local: true,
65 DisplayOrigin: "local",
66 Policy: readyTestPolicy(),
67 Identity: readyTestIdentity,
68 Incarnation: readyTestIncarnate,
69 ProtocolVersion: protocol.Version,
70 Availability: domain.RemoteAvailabilityReachable,
71 InventoryKnown: true,
72 LastSuccess: time.Now().UTC(),
73 }
74 if mutate != nil {
75 mutate(&local)
76 }
77 return local
78}
79
80// readyTestSnapshot builds one valid publication carrying exactly one local
81// daemon observation.
82func readyTestSnapshot(mutate func(*ports.BrokerDaemonObservation)) ports.BrokerSnapshot {
83 return ports.BrokerSnapshot{
84 Epoch: readyTestEpoch,
85 Revision: readyTestRevision,
86 Daemons: []ports.BrokerDaemonObservation{readyTestLocal(mutate)},
87 }
88}
89
90// readyScriptedSubscription is one coalescing capacity-one subscription.
91type readyScriptedSubscription struct {
92 mu sync.Mutex
93 changed chan struct{}
94 closed int
95}
96
97func newReadyScriptedSubscription() *readyScriptedSubscription {
98 return &readyScriptedSubscription{changed: make(chan struct{}, 1)}
99}
100
101func (s *readyScriptedSubscription) Changed() <-chan struct{} { return s.changed }
102
103func (s *readyScriptedSubscription) Close() {
104 s.mu.Lock()
105 s.closed++
106 s.mu.Unlock()
107}
108
109func (s *readyScriptedSubscription) closeCount() int {
110 s.mu.Lock()
111 defer s.mu.Unlock()
112 return s.closed
113}
114
115// readyScriptedService is a scripted ports.BrokerService. It serves one fixed
116// snapshot, records every mutating or stream method it is asked for, and lets a
117// test end its connection to drive the reconnect path.
118type readyScriptedService struct {
119 snapshot ports.BrokerSnapshot
120 subErr error
121
122 mu sync.Mutex
123 done chan struct{}
124 doneOnce sync.Once
125 subs []*readyScriptedSubscription
126 closes int
127 mutations []string
128}
129
130func newReadyScriptedService(snapshot ports.BrokerSnapshot) *readyScriptedService {
131 return &readyScriptedService{snapshot: snapshot, done: make(chan struct{})}
132}
133
134func (s *readyScriptedService) record(name string) {
135 s.mu.Lock()
136 s.mutations = append(s.mutations, name)
137 s.mu.Unlock()
138}
139
140func (s *readyScriptedService) ConnectionID() ports.BrokerConnectionID {
141 return ports.BrokerConnectionID{0x2a}
142}
143
144func (s *readyScriptedService) NextStreamID() (ports.BrokerStreamID, error) { return 1, nil }
145
146func (s *readyScriptedService) Done() <-chan struct{} { return s.done }
147
148func (s *readyScriptedService) Err() error { return nil }
149
150func (s *readyScriptedService) Snapshot() ports.BrokerSnapshot { return s.snapshot.Clone() }
151
152func (s *readyScriptedService) Subscribe() (ports.BrokerSubscription, error) {
153 if s.subErr != nil {
154 return nil, s.subErr
155 }
156 sub := newReadyScriptedSubscription()
157 s.mu.Lock()
158 s.subs = append(s.subs, sub)
159 s.mu.Unlock()
160 return sub, nil
161}
162
163func (s *readyScriptedService) SubscribePreview(ports.BrokerPreviewRequest) (ports.BrokerPreviewSubscription, error) {
164 s.record("SubscribePreview")
165 return nil, errors.New("ready scripted service does not support preview")
166}
167
168func (s *readyScriptedService) OpenStream(context.Context, ports.BrokerOpenStreamRequest) (ports.BrokerLogicalConnection, error) {
169 s.record("OpenStream")
170 return nil, ports.BrokerAdmissionClosed
171}
172
173func (s *readyScriptedService) CloseStream(ports.BrokerConnectionID, ports.BrokerStreamID) error {
174 s.record("CloseStream")
175 return nil
176}
177
178func (s *readyScriptedService) AddHost(context.Context, string, ports.BrokerPolicy) (domain.RemoteRegistration, error) {
179 s.record("AddHost")
180 return domain.RemoteRegistration{}, nil
181}
182
183func (s *readyScriptedService) RemoveHost(context.Context, domain.RemoteRegistration) (bool, error) {
184 s.record("RemoveHost")
185 return false, nil
186}
187
188func (s *readyScriptedService) UpdateHostPolicy(context.Context, domain.RemoteRegistration, ports.BrokerPolicy) (domain.RemoteRegistration, error) {
189 s.record("UpdateHostPolicy")
190 return domain.RemoteRegistration{}, nil
191}
192
193func (s *readyScriptedService) RequestReconcile(string) { s.record("RequestReconcile") }
194
195func (s *readyScriptedService) Close() error {
196 s.mu.Lock()
197 s.closes++
198 s.mu.Unlock()
199 return nil
200}
201
202// finish closes Done, so the probe observes the connection going terminal.
203func (s *readyScriptedService) finish() { s.doneOnce.Do(func() { close(s.done) }) }
204
205func (s *readyScriptedService) closeCount() int {
206 s.mu.Lock()
207 defer s.mu.Unlock()
208 return s.closes
209}
210
211// mutationNames returns the effectful methods the probe asked for, if any.
212func (s *readyScriptedService) mutationNames() []string {
213 s.mu.Lock()
214 defer s.mu.Unlock()
215 return append([]string(nil), s.mutations...)
216}
217
218// subscriptionCloses returns the total Close count across this service's
219// subscriptions.
220func (s *readyScriptedService) subscriptionCloses() int {
221 s.mu.Lock()
222 subs := append([]*readyScriptedSubscription(nil), s.subs...)
223 s.mu.Unlock()
224 total := 0
225 for _, sub := range subs {
226 total += sub.closeCount()
227 }
228 return total
229}
230
231var _ ports.BrokerService = (*readyScriptedService)(nil)
232
233// readyDialScript records every dial the probe performs and serves one fresh
234// scripted service per call, so a probe is proven to reconnect rather than
235// reuse a connection.
236type readyDialScript struct {
237 mu sync.Mutex
238 dials int
239 services []*readyScriptedService
240 produce func() (ports.BrokerService, error)
241}
242
243// scriptReadyDial replaces the probe's dial seam for one test.
244func scriptReadyDial(t *testing.T, produce func() (ports.BrokerService, error)) *readyDialScript {
245 t.Helper()
246 script := &readyDialScript{produce: produce}
247 previous := brokerReadyDial
248 brokerReadyDial = func(context.Context, string) (ports.BrokerService, error) {
249 script.mu.Lock()
250 script.dials++
251 script.mu.Unlock()
252 service, err := script.produce()
253 if recorded, ok := service.(*readyScriptedService); ok {
254 script.mu.Lock()
255 script.services = append(script.services, recorded)
256 script.mu.Unlock()
257 }
258 return service, err
259 }
260 t.Cleanup(func() { brokerReadyDial = previous })
261 return script
262}
263
264func (s *readyDialScript) dialCount() int {
265 s.mu.Lock()
266 defer s.mu.Unlock()
267 return s.dials
268}
269
270func (s *readyDialScript) createdServices() []*readyScriptedService {
271 s.mu.Lock()
272 defer s.mu.Unlock()
273 return append([]*readyScriptedService(nil), s.services...)
274}
275
276// readyTestOptions builds one parsed probe invocation over a fresh isolated
277// offline root, so no production path is ever inspected.
278func readyTestOptions(t *testing.T, requirement string, timeout time.Duration) brokerReadyOptions {
279 t.Helper()
280 emptyProductionBrokerLayout(t, "")
281 options, err := parseBrokerReadyArgs([]string{"--require", requirement, "--timeout", timeout.String()})
282 require.NoError(t, err)
283 return options
284}
285
286// runReadyProbe drives the real probe entry point over the scripted dial seam.
287func runReadyProbe(t *testing.T, options brokerReadyOptions) (string, error) {
288 t.Helper()
289 var out bytes.Buffer
290 err := runBrokerReady(context.Background(), options, &out)
291 return out.String(), err
292}
293
294// decodeReadyDocument asserts stdout holds exactly one bounded JSON line and
295// returns the decoded object.
296func decodeReadyDocument(t *testing.T, raw string) map[string]any {
297 t.Helper()
298 trimmed := strings.TrimSpace(raw)
299 require.Equal(t, 1, strings.Count(trimmed, "\n")+1, "the probe writes exactly one JSON line")
300 require.LessOrEqual(t, len(trimmed)+1, 4096, "the document stays inside its bound")
301 var decoded map[string]any
302 require.NoError(t, json.Unmarshal([]byte(trimmed), &decoded))
303 require.Equal(t, "vev.broker-ready/v1", decoded["schema"])
304 return decoded
305}
306
307// TestParseBrokerReadyArgs pins the strict argument contract: unknown,
308// duplicated, incomplete, and non-positive options are refused, a require mode
309// outside the closed pair is refused, the observation-age bound is admitted only
310// for the catalogue mode, and every default is explicit.
311func TestParseBrokerReadyArgs(t *testing.T) {
312 t.Parallel()
313
314 tests := []struct {
315 name string
316 args []string
317 want brokerReadyOptions
318 wantErr string
319 }{
320 {
321 name: "authority defaults",
322 args: []string{"--require", "local-authority"},
323 want: brokerReadyOptions{require: "local-authority", timeout: 10 * time.Second, maxAge: 5 * time.Second},
324 },
325 {
326 name: "catalogue with an explicit observation age",
327 args: []string{"--require", "local-catalogue", "--max-observation-age", "2s", "--timeout", "3s"},
328 want: brokerReadyOptions{require: "local-catalogue", timeout: 3 * time.Second, maxAge: 2 * time.Second},
329 },
330 {name: "missing require", args: nil, wantErr: "`--require` must be local-authority or local-catalogue"},
331 {name: "unknown require", args: []string{"--require", "remote-catalogue"}, wantErr: "`--require` must be local-authority or local-catalogue"},
332 {name: "missing value", args: []string{"--require"}, wantErr: "requires a value"},
333 {name: "duplicate option", args: []string{"--require", "local-authority", "--require", "local-authority"}, wantErr: "duplicate"},
334 {name: "unknown option", args: []string{"--require", "local-authority", "--ensure", "1"}, wantErr: "unknown `--ensure` option"},
335 {name: "timeout is not a duration", args: []string{"--require", "local-authority", "--timeout", "soon"}, wantErr: "`--timeout` must be positive"},
336 {name: "timeout is zero", args: []string{"--require", "local-authority", "--timeout", "0s"}, wantErr: "`--timeout` must be positive"},
337 {name: "timeout is negative", args: []string{"--require", "local-authority", "--timeout", "-1s"}, wantErr: "`--timeout` must be positive"},
338 {name: "observation age is not a duration", args: []string{"--require", "local-catalogue", "--max-observation-age", "later"}, wantErr: "`--max-observation-age` must be positive"},
339 {name: "observation age is zero", args: []string{"--require", "local-catalogue", "--max-observation-age", "0s"}, wantErr: "`--max-observation-age` must be positive"},
340 {
341 name: "observation age is authority-only",
342 args: []string{"--require", "local-authority", "--max-observation-age", "2s"},
343 wantErr: "`--max-observation-age` is only valid for local-catalogue",
344 },
345 }
346 for _, tt := range tests {
347 t.Run(tt.name, func(t *testing.T) {
348 t.Parallel()
349 got, err := parseBrokerReadyArgs(tt.args)
350 if tt.wantErr != "" {
351 require.Error(t, err)
352 require.ErrorContains(t, err, tt.wantErr)
353 require.Equal(t, 2, ExitCode(err), "an argument refusal is a usage exit")
354 return
355 }
356 require.NoError(t, err)
357 require.Equal(t, tt.want, got)
358 })
359 }
360}
361
362// TestBrokerReadyDocumentIsOneBoundedJSONLine pins the machine contract: exactly
363// one JSON line, decimal-string epoch/revision IDs, the pending list, and the
364// observed local projection, including its hex incarnation and RFC3339
365// observation time.
366func TestBrokerReadyDocumentIsOneBoundedJSONLine(t *testing.T) {
367 t.Parallel()
368
369 local := readyTestLocal(func(local *ports.BrokerDaemonObservation) {
370 local.LastSuccess = time.Date(2026, 9, 21, 12, 0, 0, 123456789, time.UTC)
371 })
372 options := brokerReadyOptions{require: "local-catalogue", timeout: time.Second, maxAge: time.Second}
373
374 tests := []struct {
375 name string
376 result brokerReadyResult
377 }{
378 {name: "absent local", result: readyResult(options, "pending", "local-not-ready", nil)},
379 {name: "observed local", result: readyResult(options, "pending", "local-not-ready", &local)},
380 {name: "ready snapshot", result: evaluateReady(options, readyTestSnapshot(func(observed *ports.BrokerDaemonObservation) { *observed = local }), local)},
381 }
382 for _, tt := range tests {
383 t.Run(tt.name, func(t *testing.T) {
384 t.Parallel()
385 var out bytes.Buffer
386 // emitReady reports the exit code for a non-ready status; the document
387 // is written either way, which is the property under test here.
388 _ = emitReady(&out, tt.result)
389
390 raw := out.String()
391 require.True(t, strings.HasSuffix(raw, "\n"), "the document ends with exactly one newline")
392 decoded := decodeReadyDocument(t, raw)
393 require.Equal(t, tt.result.Status, decoded["status"])
394 require.Equal(t, tt.result.Require, decoded["require"])
395 require.Equal(t, tt.result.Reason, decoded["reason"])
396 require.Equal(t, toAnySlice(tt.result.Pending), decoded["pending"])
397 require.Equal(t, tt.result.Epoch, decoded["epoch"], "epoch travels as a decimal string")
398 require.Equal(t, tt.result.Revision, decoded["revision"], "revision travels as a decimal string")
399
400 if tt.result.Local == nil {
401 require.Nil(t, decoded["local"])
402 return
403 }
404 observed, ok := decoded["local"].(map[string]any)
405 require.True(t, ok, "an observed local is a JSON object")
406 require.Equal(t, "ready-local-daemon", observed["identity"])
407 require.Equal(t, "0102030405060708090a0b0c0d0e0f10", observed["incarnation"])
408 require.Equal(t, strconv.FormatUint(uint64(protocol.Version), 10), observed["version"])
409 require.Equal(t, "reachable", observed["availability"])
410 require.Equal(t, true, observed["inventory_known"])
411 require.Equal(t, "2026-09-21T12:00:00.123456789Z", observed["last_success"])
412 })
413 }
414}
415
416// TestBrokerReadyExitCodeTable pins the closed status-to-exit-code mapping: 0
417// ready, 3 timeout (including the default pending case), 4 terminal, and 5
418// cancellation. The document is written before the exit code is returned.
419func TestBrokerReadyExitCodeTable(t *testing.T) {
420 t.Parallel()
421
422 tests := []struct {
423 name string
424 status string
425 wantCode int
426 wantErrMsg string
427 }{
428 {name: "ready", status: "ready", wantCode: 0},
429 {name: "timeout", status: "timeout", wantCode: 3, wantErrMsg: "broker readiness timeout"},
430 {name: "terminal", status: "terminal", wantCode: 4, wantErrMsg: "broker readiness terminal failure"},
431 {name: "canceled", status: "canceled", wantCode: 5, wantErrMsg: "context canceled"},
432 {name: "pending", status: "pending", wantCode: 3, wantErrMsg: "broker readiness timeout"},
433 }
434 for _, tt := range tests {
435 t.Run(tt.name, func(t *testing.T) {
436 t.Parallel()
437 var out bytes.Buffer
438 result := brokerReadyResult{Schema: "vev.broker-ready/v1", Status: tt.status, Require: "local-authority", Pending: []string{}}
439 err := emitReady(&out, result)
440 require.Equal(t, tt.wantCode, ExitCode(err))
441 if tt.wantErrMsg == "" {
442 require.NoError(t, err)
443 } else {
444 require.ErrorContains(t, err, tt.wantErrMsg)
445 }
446 require.Equal(t, 1, strings.Count(out.String(), "\n"), "the report is written even when the exit code is non-zero")
447 })
448 }
449}
450
451// TestBrokerReadyObservationTable pins the readiness evaluation itself:
452// `local-authority` is satisfied by any snapshot that carries a valid local
453// entry, `local-catalogue` requires a reachable, known, fresh, version-exact
454// observation, an empty known inventory is a success, and a structurally
455// invalid or doubled local publication is terminal rather than pending.
456func TestBrokerReadyObservationTable(t *testing.T) {
457 tests := []struct {
458 name string
459 require string
460 snapshot ports.BrokerSnapshot
461 subErr error
462 wantNil bool
463 wantStatus string
464 wantReason string
465 wantEpoch string
466 wantRevision string
467 wantPending []string
468 }{
469 {
470 name: "unknown availability satisfies the authority requirement",
471 require: "local-authority",
472 snapshot: readyTestSnapshot(func(local *ports.BrokerDaemonObservation) { local.Availability = domain.RemoteAvailabilityUnknown }),
473 wantStatus: "ready",
474 wantReason: "authority",
475 wantEpoch: "42",
476 wantRevision: "7",
477 wantPending: []string{},
478 },
479 {
480 name: "reachable known empty inventory satisfies the catalogue requirement",
481 require: "local-catalogue",
482 snapshot: readyTestSnapshot(nil),
483 wantStatus: "ready",
484 wantReason: "catalogue",
485 wantEpoch: "42",
486 wantRevision: "7",
487 wantPending: []string{},
488 },
489 {
490 name: "unreachable local stays pending",
491 require: "local-catalogue",
492 snapshot: readyTestSnapshot(func(local *ports.BrokerDaemonObservation) { local.Availability = domain.RemoteAvailabilityUnreachable }),
493 wantNil: true,
494 wantStatus: "pending",
495 wantPending: []string{"local-catalogue"},
496 },
497 {
498 name: "unknown inventory stays pending",
499 require: "local-catalogue",
500 snapshot: readyTestSnapshot(func(local *ports.BrokerDaemonObservation) { local.InventoryKnown = false }),
501 wantNil: true,
502 wantStatus: "pending",
503 wantPending: []string{"local-catalogue"},
504 },
505 {
506 name: "stale observation stays pending",
507 require: "local-catalogue",
508 snapshot: readyTestSnapshot(func(local *ports.BrokerDaemonObservation) { local.LastSuccess = time.Now().Add(-time.Hour) }),
509 wantNil: true,
510 wantStatus: "pending",
511 wantPending: []string{"local-catalogue"},
512 },
513 {
514 name: "missing observation time stays pending",
515 require: "local-catalogue",
516 snapshot: readyTestSnapshot(func(local *ports.BrokerDaemonObservation) { local.LastSuccess = time.Time{} }),
517 wantNil: true,
518 wantStatus: "pending",
519 wantPending: []string{"local-catalogue"},
520 },
521 {
522 name: "incompatible protocol version stays pending",
523 require: "local-catalogue",
524 snapshot: readyTestSnapshot(func(local *ports.BrokerDaemonObservation) { local.ProtocolVersion = protocol.Version + 1 }),
525 wantNil: true,
526 wantStatus: "pending",
527 wantPending: []string{"local-catalogue"},
528 },
529 {
530 name: "unobserved local stays pending",
531 require: "local-catalogue",
532 snapshot: readyTestSnapshot(func(local *ports.BrokerDaemonObservation) {
533 *local = ports.BrokerDaemonObservation{Local: true, DisplayOrigin: "local", Policy: readyTestPolicy(), Availability: domain.RemoteAvailabilityUnknown}
534 }),
535 wantNil: true,
536 wantStatus: "pending",
537 wantPending: []string{"local-catalogue"},
538 },
539 {
540 name: "snapshot without an epoch is no observation at all",
541 require: "local-authority",
542 snapshot: ports.BrokerSnapshot{},
543 wantNil: true,
544 wantStatus: "pending",
545 wantPending: []string{"local-authority"},
546 },
547 {
548 name: "snapshot without a revision is terminal",
549 require: "local-authority",
550 snapshot: ports.BrokerSnapshot{Epoch: readyTestEpoch, Revision: 0, Daemons: []ports.BrokerDaemonObservation{readyTestLocal(nil)}},
551 wantStatus: "terminal",
552 wantReason: "protocol",
553 wantEpoch: "0",
554 wantRevision: "0",
555 wantPending: []string{"local-authority"},
556 },
557 {
558 name: "two local daemons are terminal",
559 require: "local-authority",
560 snapshot: ports.BrokerSnapshot{
561 Epoch: readyTestEpoch, Revision: readyTestRevision,
562 Daemons: []ports.BrokerDaemonObservation{readyTestLocal(nil), readyTestLocal(nil)},
563 },
564 wantStatus: "terminal",
565 wantReason: "protocol",
566 wantEpoch: "0",
567 wantRevision: "0",
568 wantPending: []string{"local-authority"},
569 },
570 {
571 name: "subscription refusal is terminal",
572 require: "local-authority",
573 snapshot: readyTestSnapshot(nil),
574 subErr: errors.Join(brokeripc.ErrProtocol, errors.New("broker refused the subscription")),
575 wantStatus: "terminal",
576 wantReason: "protocol",
577 wantEpoch: "0",
578 wantRevision: "0",
579 wantPending: []string{"local-authority"},
580 },
581 }
582 for _, tt := range tests {
583 t.Run(tt.name, func(t *testing.T) {
584 service := newReadyScriptedService(tt.snapshot)
585 service.subErr = tt.subErr
586 ctx, cancel := context.WithTimeout(context.Background(), readyTestObserveMax)
587 defer cancel()
588
589 result := observeBrokerReady(ctx, brokerReadyOptions{require: tt.require, timeout: time.Second, maxAge: time.Second}, service)
590 require.Empty(t, service.mutationNames(), "readiness must never open a stream, reconcile, or mutate membership")
591 if tt.subErr != nil {
592 require.Zero(t, service.subscriptionCloses(), "a refused subscription is never closed as if it existed")
593 } else {
594 require.Equal(t, 1, service.subscriptionCloses(), "the probe closes the subscription it opened")
595 }
596 if tt.wantNil {
597 require.Nil(t, result, "an unsatisfied observation publishes nothing")
598 return
599 }
600 require.NotNil(t, result)
601 require.Equal(t, tt.wantStatus, result.Status)
602 require.Equal(t, tt.wantReason, result.Reason)
603 require.Equal(t, tt.wantEpoch, result.Epoch)
604 require.Equal(t, tt.wantRevision, result.Revision)
605 require.Equal(t, tt.wantPending, result.Pending)
606 require.Equal(t, "vev.broker-ready/v1", result.Schema)
607 })
608 }
609}
610
611// TestBrokerReadyRunOutcomeTable drives the real probe entry point for every
612// exit class: an on-demand broker that is merely absent or refused is retried
613// into the deadline (3), a security, configuration, or protocol refusal is
614// terminal (4), a satisfied requirement is 0, and every connection the probe
615// opened is closed exactly once with no effectful call.
616func TestBrokerReadyRunOutcomeTable(t *testing.T) {
617 tests := []struct {
618 name string
619 require string
620 produce func() (ports.BrokerService, error)
621 wantCode int
622 wantStatus string
623 wantReason string
624 wantPending []string
625 }{
626 {
627 name: "authority satisfied",
628 require: "local-authority",
629 produce: func() (ports.BrokerService, error) { return newReadyScriptedService(readyTestSnapshot(nil)), nil },
630 wantCode: 0,
631 wantStatus: "ready",
632 wantReason: "authority",
633 wantPending: []string{},
634 },
635 {
636 name: "catalogue satisfied by a known empty inventory",
637 require: "local-catalogue",
638 produce: func() (ports.BrokerService, error) { return newReadyScriptedService(readyTestSnapshot(nil)), nil },
639 wantCode: 0,
640 wantStatus: "ready",
641 wantReason: "catalogue",
642 wantPending: []string{},
643 },
644 {
645 name: "catalogue with an incompatible version times out",
646 require: "local-catalogue",
647 produce: func() (ports.BrokerService, error) {
648 return newReadyScriptedService(readyTestSnapshot(func(local *ports.BrokerDaemonObservation) { local.ProtocolVersion = protocol.Version + 1 })), nil
649 },
650 wantCode: 3,
651 wantStatus: "timeout",
652 wantReason: "deadline",
653 wantPending: []string{"local-catalogue"},
654 },
655 {
656 name: "stale catalogue observation times out",
657 require: "local-catalogue",
658 produce: func() (ports.BrokerService, error) {
659 return newReadyScriptedService(readyTestSnapshot(func(local *ports.BrokerDaemonObservation) { local.LastSuccess = time.Now().Add(-time.Hour) })), nil
660 },
661 wantCode: 3,
662 wantStatus: "timeout",
663 wantReason: "deadline",
664 wantPending: []string{"local-catalogue"},
665 },
666 {
667 name: "no snapshot times out",
668 require: "local-authority",
669 produce: func() (ports.BrokerService, error) { return newReadyScriptedService(ports.BrokerSnapshot{}), nil },
670 wantCode: 3,
671 wantStatus: "timeout",
672 wantReason: "deadline",
673 wantPending: []string{"local-authority"},
674 },
675 {
676 name: "absent endpoint is retried then times out",
677 require: "local-authority",
678 produce: func() (ports.BrokerService, error) {
679 return nil, &net.OpError{Op: "dial", Net: "unix", Err: syscall.ENOENT}
680 },
681 wantCode: 3,
682 wantStatus: "timeout",
683 wantReason: "deadline",
684 wantPending: []string{"local-authority"},
685 },
686 {
687 name: "refused endpoint is retried then times out",
688 require: "local-authority",
689 produce: func() (ports.BrokerService, error) {
690 return nil, &net.OpError{Op: "dial", Net: "unix", Err: syscall.ECONNREFUSED}
691 },
692 wantCode: 3,
693 wantStatus: "timeout",
694 wantReason: "deadline",
695 wantPending: []string{"local-authority"},
696 },
697 {
698 name: "invalid snapshot is terminal",
699 require: "local-authority",
700 produce: func() (ports.BrokerService, error) {
701 return newReadyScriptedService(ports.BrokerSnapshot{Epoch: readyTestEpoch}), nil
702 },
703 wantCode: 4,
704 wantStatus: "terminal",
705 wantReason: "protocol",
706 wantPending: []string{"local-authority"},
707 },
708 {
709 name: "rejected peer is terminal",
710 require: "local-authority",
711 produce: func() (ports.BrokerService, error) { return nil, ipc.ErrMuxPeerRejected },
712 wantCode: 4,
713 wantStatus: "terminal",
714 wantReason: "security",
715 wantPending: []string{"local-authority"},
716 },
717 {
718 name: "permission refusal is terminal",
719 require: "local-authority",
720 produce: func() (ports.BrokerService, error) { return nil, os.ErrPermission },
721 wantCode: 4,
722 wantStatus: "terminal",
723 wantReason: "security",
724 wantPending: []string{"local-authority"},
725 },
726 {
727 name: "invalid configuration is terminal",
728 require: "local-authority",
729 produce: func() (ports.BrokerService, error) { return nil, brokeripc.ErrConfig },
730 wantCode: 4,
731 wantStatus: "terminal",
732 wantReason: "config",
733 wantPending: []string{"local-authority"},
734 },
735 {
736 name: "broker protocol violation is terminal",
737 require: "local-authority",
738 produce: func() (ports.BrokerService, error) {
739 return nil, errors.Join(brokeripc.ErrMalformedFrame, errors.New("truncated"))
740 },
741 wantCode: 4,
742 wantStatus: "terminal",
743 wantReason: "protocol",
744 wantPending: []string{"local-authority"},
745 },
746 }
747 for _, tt := range tests {
748 t.Run(tt.name, func(t *testing.T) {
749 options := readyTestOptions(t, tt.require, 80*time.Millisecond)
750 script := scriptReadyDial(t, tt.produce)
751
752 out, err := runReadyProbe(t, options)
753 require.Equal(t, tt.wantCode, ExitCode(err), "probe outcome %q", out)
754
755 decoded := decodeReadyDocument(t, out)
756 require.Equal(t, tt.wantStatus, decoded["status"])
757 require.Equal(t, tt.wantReason, decoded["reason"])
758 require.Equal(t, tt.require, decoded["require"])
759 require.Equal(t, toAnySlice(tt.wantPending), decoded["pending"])
760
761 services := script.createdServices()
762 for _, service := range services {
763 require.Equal(t, 1, service.closeCount(), "every dialed connection is closed exactly once")
764 require.Empty(t, service.mutationNames(), "readiness opens no stream, reconciles nothing, and mutates no membership")
765 require.Equal(t, 1, service.subscriptionCloses(), "the probe closes the subscription it opened")
766 }
767 if tt.wantCode == 0 {
768 require.Equal(t, 1, script.dialCount(), "a satisfied probe dials exactly once")
769 }
770 })
771 }
772}
773
774// TestBrokerReadyNeverEnsuresSpawnsOrMutates pins the zero-effect guarantee at
775// the probe level: the dial seam is the only I/O seam it uses, every connection
776// it opens is closed, every subscription is closed, and no logical stream,
777// reconcile, or membership mutation is ever attempted.
778func TestBrokerReadyNeverEnsuresSpawnsOrMutates(t *testing.T) {
779 options := readyTestOptions(t, "local-authority", 80*time.Millisecond)
780 script := scriptReadyDial(t, func() (ports.BrokerService, error) {
781 return newReadyScriptedService(readyTestSnapshot(nil)), nil
782 })
783
784 out, err := runReadyProbe(t, options)
785 require.NoError(t, err, "outcome %q", out)
786 require.Equal(t, 1, script.dialCount(), "a satisfied probe dials exactly once")
787 require.Len(t, script.createdServices(), 1)
788
789 service := script.createdServices()[0]
790 require.Empty(t, service.mutationNames(), "readiness opens no stream, reconciles nothing, and mutates no membership")
791 require.Equal(t, 1, service.closeCount())
792 require.Equal(t, 1, service.subscriptionCloses())
793
794 decoded := decodeReadyDocument(t, out)
795 require.Equal(t, "ready", decoded["status"])
796 observed, ok := decoded["local"].(map[string]any)
797 require.True(t, ok)
798 require.Equal(t, "reachable", observed["availability"])
799 require.Equal(t, true, observed["inventory_known"])
800}
801
802// TestBrokerReadyReconnectUsesTheNewEpoch pins that a broker connection that
803// ends before readiness is replaced by a fresh dial, and that the report carries
804// the new broker epoch: a retired epoch is never reused as readiness authority.
805func TestBrokerReadyReconnectUsesTheNewEpoch(t *testing.T) {
806 freshSnapshot := ports.BrokerSnapshot{Epoch: ports.BrokerEpoch(2), Revision: 3, Daemons: []ports.BrokerDaemonObservation{readyTestLocal(nil)}}
807
808 options := readyTestOptions(t, "local-authority", 2*time.Second)
809 var (
810 mu sync.Mutex
811 first *readyScriptedService
812 attempts int
813 )
814 scriptReadyDial(t, func() (ports.BrokerService, error) {
815 mu.Lock()
816 defer mu.Unlock()
817 attempts++
818 if attempts == 1 {
819 // The first connection commits no snapshot before it retires: the
820 // probe must reconnect rather than report the retired connection.
821 first = newReadyScriptedService(ports.BrokerSnapshot{})
822 first.finish()
823 return first, nil
824 }
825 return newReadyScriptedService(freshSnapshot), nil
826 })
827
828 out, err := runReadyProbe(t, options)
829 require.NoError(t, err, "outcome %q", out)
830
831 decoded := decodeReadyDocument(t, out)
832 require.Equal(t, "ready", decoded["status"])
833 require.Equal(t, "2", decoded["epoch"], "the report carries the epoch of the connection that answered")
834 require.NotEqual(t, "1", decoded["epoch"], "a retired epoch is never reused")
835
836 mu.Lock()
837 defer mu.Unlock()
838 require.Equal(t, 2, attempts, "a retired connection is replaced by exactly one fresh dial")
839 require.NotNil(t, first)
840 require.Equal(t, 1, first.closeCount(), "the retired connection is closed exactly once")
841}
842
843// TestBrokerReadyCancellationIsExitFiveAndClosesEverything pins the
844// cancellation contract: the caller sees exit 5, the document says canceled,
845// and the connection and subscription the probe opened are closed rather than
846// leaked.
847func TestBrokerReadyCancellationIsExitFiveAndClosesEverything(t *testing.T) {
848 options := readyTestOptions(t, "local-authority", 5*time.Second)
849 script := scriptReadyDial(t, func() (ports.BrokerService, error) {
850 return newReadyScriptedService(ports.BrokerSnapshot{}), nil
851 })
852
853 ctx, cancel := context.WithCancel(context.Background())
854 var out bytes.Buffer
855 results := make(chan error, 1)
856 go func() { results <- runBrokerReady(ctx, options, &out) }()
857
858 require.Eventually(t, func() bool { return script.dialCount() == 1 }, time.Second, 5*time.Millisecond, "the probe dials before it is cancelled")
859 cancel()
860
861 select {
862 case err := <-results:
863 require.Equal(t, 5, ExitCode(err))
864 case <-time.After(2 * time.Second):
865 t.Fatal("a cancelled probe must return promptly")
866 }
867
868 decoded := decodeReadyDocument(t, out.String())
869 require.Equal(t, "canceled", decoded["status"])
870 require.Equal(t, "canceled", decoded["reason"])
871
872 services := script.createdServices()
873 require.Len(t, services, 1)
874 require.Equal(t, 1, services[0].closeCount(), "cancellation closes the connection")
875 require.Equal(t, 1, services[0].subscriptionCloses(), "cancellation closes the subscription")
876 require.Empty(t, services[0].mutationNames())
877}
878
879// TestBrokerReadyEndpointSecurityClassificationTable pins the endpoint check the
880// probe performs before any dial: absence is retryable, and a foreign path or an
881// uninspectable path is a terminal security failure.
882func TestBrokerReadyEndpointSecurityClassificationTable(t *testing.T) {
883 t.Parallel()
884
885 dir := t.TempDir()
886 regular := filepath.Join(dir, "not-a-socket")
887 require.NoError(t, os.WriteFile(regular, []byte("x"), 0o600))
888
889 socketDir := shortTempDir(t, "ready")
890 socketPath := filepath.Join(socketDir, "broker.sock")
891 listener, err := net.Listen("unix", socketPath)
892 require.NoError(t, err)
893 t.Cleanup(func() { _ = listener.Close() })
894
895 unreadableDir := filepath.Join(t.TempDir(), "unreadable")
896 require.NoError(t, os.Mkdir(unreadableDir, 0o700))
897 unreadable := filepath.Join(unreadableDir, "broker.sock")
898 require.NoError(t, os.Chmod(unreadableDir, 0o000))
899 t.Cleanup(func() { _ = os.Chmod(unreadableDir, 0o700) })
900
901 tests := []struct {
902 name string
903 path string
904 wantTerminal bool
905 wantReason string
906 }{
907 {name: "absent endpoint is retryable", path: filepath.Join(dir, "absent.sock")},
908 {name: "bound socket is dialable", path: socketPath},
909 {name: "regular file is terminal", path: regular, wantTerminal: true, wantReason: "security"},
910 {name: "directory is terminal", path: socketDir, wantTerminal: true, wantReason: "security"},
911 {name: "uninspectable path is terminal", path: unreadable, wantTerminal: true, wantReason: "security"},
912 }
913 for _, tt := range tests {
914 t.Run(tt.name, func(t *testing.T) {
915 t.Parallel()
916 reason, terminal := brokerReadyEndpointSecurity(tt.path)
917 require.Equal(t, tt.wantTerminal, terminal)
918 require.Equal(t, tt.wantReason, reason)
919 })
920 }
921}
922
923// TestBrokerReadyDialClassificationTable pins the dial-failure taxonomy: only a
924// refusal that proves the peer, the path, the credentials, the configuration, or
925// the broker conversation itself is not the expected broker is terminal.
926func TestBrokerReadyDialClassificationTable(t *testing.T) {
927 t.Parallel()
928
929 tests := []struct {
930 name string
931 err error
932 wantTerminal bool
933 wantReason string
934 }{
935 {name: "nil is not a failure", err: nil},
936 {name: "absent endpoint is retryable", err: &net.OpError{Op: "dial", Net: "unix", Err: syscall.ENOENT}},
937 {name: "refused endpoint is retryable", err: &net.OpError{Op: "dial", Net: "unix", Err: syscall.ECONNREFUSED}},
938 {name: "cancellation is retryable", err: context.Canceled},
939 {name: "deadline is retryable", err: context.DeadlineExceeded},
940 {name: "unclassified failure is retryable", err: errors.New("io error")},
941 {name: "rejected peer is terminal", err: ipc.ErrMuxPeerRejected, wantTerminal: true, wantReason: "security"},
942 {name: "foreign path is terminal", err: ipc.ErrMuxForeignPath, wantTerminal: true, wantReason: "security"},
943 {name: "invalid mux path is terminal", err: ipc.ErrMuxPath, wantTerminal: true, wantReason: "security"},
944 {name: "unsupported peer check is terminal", err: ipc.ErrMuxUnsupported, wantTerminal: true, wantReason: "security"},
945 {name: "permission failure is terminal", err: &net.OpError{Op: "dial", Net: "unix", Err: syscall.EACCES}, wantTerminal: true, wantReason: "security"},
946 {name: "invalid configuration is terminal", err: brokeripc.ErrConfig, wantTerminal: true, wantReason: "config"},
947 {name: "broker protocol violation is terminal", err: brokeripc.ErrProtocol, wantTerminal: true, wantReason: "protocol"},
948 {name: "malformed frame is terminal", err: brokeripc.ErrMalformedFrame, wantTerminal: true, wantReason: "protocol"},
949 {name: "oversize non-broker frame is terminal", err: ipc.ErrFrameTooLarge, wantTerminal: true, wantReason: "protocol"},
950 {name: "zero-length non-broker frame is terminal", err: ipc.ErrZeroLengthFrame, wantTerminal: true, wantReason: "protocol"},
951 }
952 for _, tt := range tests {
953 t.Run(tt.name, func(t *testing.T) {
954 t.Parallel()
955 reason, terminal := terminalReadyReason(tt.err)
956 require.Equal(t, tt.wantTerminal, terminal, "cause %v", tt.err)
957 require.Equal(t, tt.wantReason, reason)
958 })
959 }
960}
961
962// toAnySlice renders a decoded JSON string array comparison input.
963func toAnySlice(values []string) []any {
964 out := make([]any, 0, len(values))
965 for _, value := range values {
966 out = append(out, value)
967 }
968 return out
969}