1//go:build linux
2
3package app
4
5import (
6 "context"
7 "encoding/json"
8 "errors"
9 "io"
10 "log/slog"
11 "net"
12 "os"
13 "path/filepath"
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/brokerstore"
23 "github.com/bnema/vev/internal/adapters/clock"
24 "github.com/bnema/vev/internal/adapters/daemonidentity"
25 "github.com/bnema/vev/internal/adapters/daemonmux"
26 "github.com/bnema/vev/internal/adapters/ipc"
27 "github.com/bnema/vev/internal/domain"
28 "github.com/bnema/vev/internal/ports"
29 "github.com/bnema/vev/internal/protocol"
30 "github.com/bnema/vev/internal/protocol/catalogue"
31 "github.com/bnema/vev/internal/usecase/broker"
32)
33
34// testLocalSandboxConfig composes the production local authority with a
35// daemon-owned identity under private XDG directories.
36func testLocalSandboxConfig(t *testing.T) (*brokerconfig.Config, brokerconfig.LocalBinding) {
37 t.Helper()
38 layout := emptyProductionBrokerLayout(t, "")
39 identity, err := daemonidentity.LoadOrCreate(filepath.Dir(layout.Root))
40 require.NoError(t, err)
41 config, err := brokerconfig.LoadProduction(layout, productionBrokerConfigPath(), identity, localDaemonPolicy(), daemonmux.SocketPath(ipc.SocketDir()))
42 require.NoError(t, err)
43 binding, ok := config.LocalBinding()
44 require.True(t, ok)
45 return config, binding
46}
47
48// rawFramedTransport adapts one in-memory pipe end to the raw framed carriage
49// the daemonmux bridge consumes. No real socket is opened.
50func rawFramedTransport(raw net.Conn) daemonmux.RawFramedTransport {
51 return ipc.NewTransport(raw).(daemonmux.RawFramedTransport)
52}
53
54// testServerBinding builds the daemon-side binding matching one local binding.
55func testServerBinding(t *testing.T, binding brokerconfig.LocalBinding) daemonmux.ServerBinding {
56 t.Helper()
57 serverBinding, err := daemonmux.NewServerBinding(binding.Identity, ports.BrokerDaemonIncarnation{7, 1}, binding.Policy)
58 require.NoError(t, err)
59 return serverBinding
60}
61
62// testLocalConnector builds the broker-owned endpoint connector a probe
63// composes, over an explicit dial seam. It is the same connector type the pool
64// uses, so a test exercises the real composition path.
65func testLocalConnector(t *testing.T, dial localCarrierDialer) ports.BrokerEndpointConnector {
66 t.Helper()
67 connector, err := daemonmux.NewEndpointConnector(daemonmux.RawCarrierDialer(dial), daemonmux.DefaultMuxCeilings())
68 require.NoError(t, err)
69 return connector
70}
71
72// unreachableCarriageConnector builds a connector whose dial always fails, so a
73// composition test can build a probe without a scripted daemon.
74func unreachableCarriageConnector(t *testing.T) ports.BrokerEndpointConnector {
75 t.Helper()
76 return testLocalConnector(t, func(context.Context, ports.BrokerDialTarget) (daemonmux.RawFramedTransport, error) {
77 return nil, errors.New("no carriage")
78 })
79}
80
81// testLocalUpSession builds one exact, attachable catalogue session with an
82// ordered tab list.
83func testLocalUpSession(lifecycle domain.SessionLifecycleID, name string) catalogue.RemoteCatalogSession {
84 return catalogue.RemoteCatalogSession{
85 LifecycleID: lifecycle,
86 Name: name,
87 State: catalogue.RemoteCatalogSessionUp,
88 Tabs: []catalogue.RemoteCatalogTab{},
89 }
90}
91
92// scriptedLocalDaemon is a scripted daemonmux daemon for the broker-owned local
93// probe: it completes the authenticated physical preamble for the supplied
94// binding, accepts the logical streams the probe opens, records every typed
95// client message, and answers from a scripted responder. It never creates a
96// session or a process and never interprets the probe's traffic beyond the
97// responder.
98type scriptedLocalDaemon struct {
99 supervisor *daemonmux.ServerSupervisor
100 respond func(protocol.ClientMessage) protocol.ServerMessage
101
102 dials atomic.Int32
103
104 mu sync.Mutex
105 streams int
106 received []protocol.ClientMessage
107}
108
109// newScriptedLocalDaemon serves one scripted daemonmux daemon over in-memory
110// carriages. The responder answers one typed client message; a nil reply ends
111// the stream without an answer.
112func newScriptedLocalDaemon(t *testing.T, binding daemonmux.ServerBinding, respond func(protocol.ClientMessage) protocol.ServerMessage) *scriptedLocalDaemon {
113 t.Helper()
114 daemon := &scriptedLocalDaemon{respond: respond}
115 aggregate := daemonmux.NewAggregateListener()
116 supervisor, err := daemonmux.NewServerSupervisor(aggregate, binding, daemonmux.DefaultMuxCeilings(), 0)
117 require.NoError(t, err)
118 t.Cleanup(func() {
119 _ = supervisor.Close()
120 _ = aggregate.Close()
121 })
122 go func() {
123 for {
124 connection, acceptErr := aggregate.Accept()
125 if acceptErr != nil {
126 return
127 }
128 daemon.mu.Lock()
129 daemon.streams++
130 daemon.mu.Unlock()
131 go daemon.serveStream(connection)
132 }
133 }()
134 daemon.supervisor = supervisor
135 return daemon
136}
137
138// dial returns the probe's carrier dialer: one fresh in-memory carriage whose
139// daemon end is adopted by the scripted daemon.
140func (d *scriptedLocalDaemon) dial() localCarrierDialer {
141 return func(ctx context.Context, _ ports.BrokerDialTarget) (daemonmux.RawFramedTransport, error) {
142 d.dials.Add(1)
143 client, server := net.Pipe()
144 go func() { _ = d.supervisor.Adopt(context.Background(), rawFramedTransport(server)) }()
145 return rawFramedTransport(client), nil
146 }
147}
148
149// serveStream records every typed client message of one admitted stream and
150// answers it from the responder. A responder that returns nil ends the stream
151// without an answer, so a test can exercise the peer-closed-without-replying
152// path; a responder that blocks holds the stream open, so a test can exercise
153// the caller-bounded wait.
154func (d *scriptedLocalDaemon) serveStream(connection ports.ServerConnection) {
155 defer func() { _ = connection.Close() }()
156 for {
157 message, err := connection.ReceiveClient()
158 if err != nil {
159 return
160 }
161 d.mu.Lock()
162 d.received = append(d.received, message)
163 d.mu.Unlock()
164 if d.respond == nil {
165 return
166 }
167 reply := d.respond(message)
168 if reply == nil {
169 return
170 }
171 if err := connection.SendServer(reply); err != nil {
172 return
173 }
174 }
175}
176
177// messages returns a defensive copy of every recorded client message.
178func (d *scriptedLocalDaemon) messages() []protocol.ClientMessage {
179 d.mu.Lock()
180 defer d.mu.Unlock()
181 return append([]protocol.ClientMessage(nil), d.received...)
182}
183
184// awaitMessages waits until the daemon recorded at least count messages.
185func (d *scriptedLocalDaemon) awaitMessages(t *testing.T, count int) []protocol.ClientMessage {
186 t.Helper()
187 require.Eventually(t, func() bool {
188 d.mu.Lock()
189 defer d.mu.Unlock()
190 return len(d.received) >= count
191 }, brokerTestWait, time.Millisecond, "scripted local daemon recorded %d messages", len(d.messages()))
192 return d.messages()
193}
194
195// awaitStreams waits until the daemon admitted at least count logical streams.
196func (d *scriptedLocalDaemon) awaitStreams(t *testing.T, count int) {
197 t.Helper()
198 require.Eventually(t, func() bool {
199 d.mu.Lock()
200 defer d.mu.Unlock()
201 return d.streams >= count
202 }, brokerTestWait, time.Millisecond, "scripted local daemon admitted too few streams")
203}
204
205// TestBrokerLocalProbeObservesAuthenticatedCatalogue proves the broker-owned
206// local probe opens exactly one logical observation stream over the
207// authenticated daemonmux carriage, asks the daemon for its own catalogue
208// through the existing typed control protocol, and publishes the observed
209// identity, incarnation, protocol version, and exact session inventory. It
210// never sends a Hello, requests no admission, claims no geometry, and creates
211// nothing.
212func TestBrokerLocalProbeObservesAuthenticatedCatalogue(t *testing.T) {
213 _, binding := testLocalSandboxConfig(t)
214 serverBinding := testServerBinding(t, binding)
215 lifecycle := domain.SessionLifecycleID{1}
216 session := testLocalUpSession(lifecycle, "work")
217
218 daemon := newScriptedLocalDaemon(t, serverBinding, func(message protocol.ClientMessage) protocol.ServerMessage {
219 request, ok := message.(protocol.CommandRequest)
220 if !ok {
221 return nil
222 }
223 return protocol.CommandResult{RequestID: request.RequestID, Outcome: protocol.CommandSucceeded, Output: testLocalCatalogue(t, session)}
224 })
225
226 probe, err := newLocalRouteProbe(binding, brokerLocalTestEpoch, daemon.dial(), daemonmux.DefaultMuxCeilings())
227 require.NoError(t, err)
228
229 observation, err := probe.ProbeLocal(context.Background())
230 require.NoError(t, err)
231 require.Equal(t, binding.Identity, observation.Identity)
232 require.Equal(t, serverBinding.Incarnation(), observation.Incarnation)
233 require.Equal(t, protocol.Version, observation.ProtocolVersion)
234 require.Equal(t, domain.RemoteAvailabilityReachable, observation.Availability)
235 require.True(t, observation.InventoryKnown, "an answered catalogue is known inventory")
236 require.Equal(t, []catalogue.RemoteCatalogSession{session}, observation.Sessions)
237 require.Zero(t, observation.Capabilities, "the catalogue command carries no capabilities; they must stay unknown")
238 require.False(t, observation.Local, "the probe supplies observed state only, never configured authority")
239 require.Empty(t, observation.Endpoint)
240 require.Equal(t, domain.RemoteRegistration{}, observation.Registration)
241
242 // Exactly one observation stream opened, and its only traffic is one typed,
243 // correlated remote-catalog request: no Hello means no attachment, no
244 // session creation, and no geometry claim.
245 daemon.awaitStreams(t, 1)
246 messages := daemon.awaitMessages(t, 1)
247 require.Len(t, messages, 1)
248 request, ok := messages[0].(protocol.CommandRequest)
249 require.True(t, ok, "the observation stream must send only a typed command, got %#v", messages[0])
250 require.Equal(t, "remote-catalog", request.Slug)
251 require.True(t, request.JSON)
252 require.False(t, request.Attached, "an observation stream never claims an attachment")
253 require.NotZero(t, request.RequestID)
254 require.Equal(t, protocol.Version, request.Version)
255 for _, message := range messages {
256 _, hello := message.(protocol.Hello)
257 require.False(t, hello, "an observation must never send a Hello")
258 }
259 require.EqualValues(t, 1, daemon.dials.Load())
260}
261
262// TestBrokerLocalProbeFailsClosed proves an unavailable carriage yields zero
263// observed identity, no inventory, and an explicit availability, never an
264// invented one, and that the probe only dials the provisioned local address: it
265// has no spawn path and never starts a stopped daemon.
266func TestBrokerLocalProbeFailsClosed(t *testing.T) {
267 _, binding := testLocalSandboxConfig(t)
268
269 var addresses []string
270 dial := func(_ context.Context, target ports.BrokerDialTarget) (daemonmux.RawFramedTransport, error) {
271 require.Equal(t, ports.BrokerDaemonExistingOnly, target.StartMode, "an observation dial never authorizes starting the daemon")
272 addresses = append(addresses, target.Address)
273 return nil, errors.New("no carriage")
274 }
275 probe, err := newLocalRouteProbe(binding, brokerLocalTestEpoch, dial, daemonmux.DefaultMuxCeilings())
276 require.NoError(t, err)
277
278 observation, err := probe.ProbeLocal(context.Background())
279 require.Error(t, err)
280 require.Equal(t, domain.RemoteAvailabilityUnreachable, observation.Availability)
281 require.Zero(t, observation.Identity)
282 require.True(t, observation.Incarnation.IsZero())
283 require.Zero(t, observation.ProtocolVersion)
284 require.False(t, observation.InventoryKnown)
285 require.Nil(t, observation.Sessions)
286 require.Equal(t, []string{binding.Route.Address()}, addresses)
287
288 t.Run("a nil carriage is refused", func(t *testing.T) {
289 probe, err := newLocalRouteProbe(binding, brokerLocalTestEpoch, func(context.Context, ports.BrokerDialTarget) (daemonmux.RawFramedTransport, error) {
290 return nil, nil
291 }, daemonmux.DefaultMuxCeilings())
292 require.NoError(t, err)
293 observation, err := probe.ProbeLocal(context.Background())
294 require.Error(t, err)
295 require.Zero(t, observation.Identity)
296 require.Equal(t, domain.RemoteAvailabilityUnreachable, observation.Availability)
297 })
298
299 t.Run("construction refuses a missing dialer", func(t *testing.T) {
300 _, err := newLocalRouteProbe(binding, brokerLocalTestEpoch, nil, daemonmux.DefaultMuxCeilings())
301 require.Error(t, err)
302 })
303
304 t.Run("construction refuses a zero epoch", func(t *testing.T) {
305 _, err := newLocalRouteProbe(binding, 0, func(context.Context, ports.BrokerDialTarget) (daemonmux.RawFramedTransport, error) {
306 return nil, nil
307 }, daemonmux.DefaultMuxCeilings())
308 require.Error(t, err)
309 })
310}
311
312// TestBrokerLocalProbeRejectsInvalidObservation proves every malformed or
313// mistyped answer fails closed: an uncorrelated result, an explicit failure, a
314// result that is not a result at all, malformed JSON, and a catalogue outside
315// the exact current schema all yield an error and an explicit unreachable
316// observation with zero identity, zero version, and no inventory. A probe never
317// publishes inventory it could not validate.
318func TestBrokerLocalProbeRejectsInvalidObservation(t *testing.T) {
319 _, binding := testLocalSandboxConfig(t)
320 serverBinding := testServerBinding(t, binding)
321 session := testLocalUpSession(domain.SessionLifecycleID{1}, "work")
322
323 // stalled is released when the subtest ends, so a responder that holds an
324 // answer open never leaks its goroutine past the test that needed it.
325 stalled := make(chan struct{})
326 t.Cleanup(func() { close(stalled) })
327
328 valid := testLocalCatalogue(t, session)
329 badSchema, err := json.Marshal(catalogue.RemoteCatalog{
330 ProtocolVersion: protocol.Version,
331 SchemaVersion: catalogue.RemoteCatalogSchemaVersion + 1,
332 Sessions: []catalogue.RemoteCatalogSession{session},
333 })
334 require.NoError(t, err)
335
336 tests := []struct {
337 name string
338 respond func(protocol.ClientMessage) protocol.ServerMessage
339 timeout time.Duration
340 }{
341 {
342 name: "uncorrelated result",
343 respond: func(message protocol.ClientMessage) protocol.ServerMessage {
344 request, _ := message.(protocol.CommandRequest)
345 return protocol.CommandResult{RequestID: request.RequestID + 1, Outcome: protocol.CommandSucceeded, Output: valid}
346 },
347 },
348 {
349 name: "explicit failure",
350 respond: func(protocol.ClientMessage) protocol.ServerMessage {
351 return protocol.CommandResult{RequestID: 0, Outcome: protocol.CommandFailed, Code: protocol.ErrInternal, Text: "catalogue unavailable"}
352 },
353 },
354 {
355 name: "wrong message type",
356 respond: func(protocol.ClientMessage) protocol.ServerMessage { return protocol.Pong{} },
357 },
358 {
359 name: "malformed json",
360 respond: func(message protocol.ClientMessage) protocol.ServerMessage {
361 request, _ := message.(protocol.CommandRequest)
362 return protocol.CommandResult{RequestID: request.RequestID, Outcome: protocol.CommandSucceeded, Output: "{not json"}
363 },
364 },
365 {
366 name: "schema mismatch",
367 respond: func(message protocol.ClientMessage) protocol.ServerMessage {
368 request, _ := message.(protocol.CommandRequest)
369 return protocol.CommandResult{RequestID: request.RequestID, Outcome: protocol.CommandSucceeded, Output: string(badSchema)}
370 },
371 },
372 {
373 name: "peer closes without answering",
374 respond: func(protocol.ClientMessage) protocol.ServerMessage { return nil },
375 },
376 {
377 name: "daemon stalls and the attempt deadline expires",
378 respond: func(protocol.ClientMessage) protocol.ServerMessage {
379 <-stalled
380 return nil
381 },
382 timeout: 200 * time.Millisecond,
383 },
384 }
385 for _, tt := range tests {
386 t.Run(tt.name, func(t *testing.T) {
387 daemon := newScriptedLocalDaemon(t, serverBinding, tt.respond)
388 probe, err := newLocalRouteProbe(binding, brokerLocalTestEpoch, daemon.dial(), daemonmux.DefaultMuxCeilings())
389 require.NoError(t, err)
390
391 wait := brokerTestWait
392 if tt.timeout > 0 {
393 wait = tt.timeout
394 }
395 ctx, cancel := context.WithTimeout(context.Background(), wait)
396 defer cancel()
397 observation, err := probe.ProbeLocal(ctx)
398 require.Error(t, err)
399 require.Equal(t, domain.RemoteAvailabilityUnreachable, observation.Availability)
400 require.Zero(t, observation.Identity)
401 require.True(t, observation.Incarnation.IsZero())
402 require.Zero(t, observation.ProtocolVersion)
403 require.False(t, observation.InventoryKnown)
404 require.Empty(t, observation.Sessions)
405 })
406 }
407}
408
409// TestBrokerLocalObservationComposition proves the composition derives the
410// local producer's authority from the validated configuration and refuses a
411// configuration that provisions no local binding.
412func TestBrokerLocalObservationComposition(t *testing.T) {
413 config, binding := testLocalSandboxConfig(t)
414 connector := unreachableCarriageConnector(t)
415 observation, err := brokerLocalObservation(config, brokerLocalTestEpoch, connector)
416 require.NoError(t, err)
417 require.Equal(t, binding.DisplayOrigin, observation.DisplayOrigin)
418 require.Equal(t, binding.Policy, observation.Policy)
419 require.NotNil(t, observation.Probe)
420 _, ok := observation.Probe.(*broker.LocalDaemonProbe)
421 require.True(t, ok)
422}
423
424// Production always composes local observation and mutable membership, even
425// when broker.json contains no remote hosts.
426func TestProductionRegistryConfigAlwaysSelectsLocalObservation(t *testing.T) {
427 config, binding := testLocalSandboxConfig(t)
428 selected, err := offlineRegistryConfig(config, brokerLocalTestEpoch, unreachableCarriageConnector(t))
429 require.NoError(t, err)
430 require.False(t, selected.ObservationDisabled)
431 require.Equal(t, broker.MembershipMutable, selected.MembershipMode)
432 require.NotNil(t, selected.Local)
433 require.Equal(t, binding.DisplayOrigin, selected.Local.DisplayOrigin)
434 require.Equal(t, binding.Policy, selected.Local.Policy)
435}
436
437// productionLocalConfig builds one production-shaped broker configuration whose
438// local binding carries an empty identity, exactly as the broker composes it
439// before the daemon's first start. The identity is supplied only by the loader.
440func productionLocalConfig(t *testing.T) (*brokerconfig.Config, brokerconfig.LocalBinding) {
441 t.Helper()
442 t.Setenv("XDG_RUNTIME_DIR", t.TempDir())
443 t.Setenv("XDG_STATE_HOME", t.TempDir())
444 t.Setenv("XDG_CONFIG_HOME", t.TempDir())
445 layout := productionBrokerLayout()
446 path := productionBrokerConfigPath()
447 require.NoError(t, os.MkdirAll(filepath.Dir(path), 0o700))
448 raw, err := json.Marshal(sandboxEmptyDocument("3s"))
449 require.NoError(t, err)
450 require.NoError(t, os.WriteFile(path, raw, 0o600))
451 config, err := brokerconfig.LoadProduction(layout, path, "", localDaemonPolicy(), daemonmux.SocketPath(ipc.SocketDir()))
452 require.NoError(t, err)
453 binding, ok := config.LocalBinding()
454 require.True(t, ok)
455 require.Empty(t, binding.Identity, "production local authority starts with no identity")
456 return config, binding
457}
458
459// TestBrokerLocalObservationWithoutIdentityPublishesLocalUnknown proves the
460// composition observes the local daemon even before an identity exists: the
461// published snapshot carries exactly one local entry with the explicit unknown
462// availability and zero observed identity, at a nonzero epoch and revision. A
463// broker may therefore precede its daemon and still publish a valid catalogue.
464func TestBrokerLocalObservationWithoutIdentityPublishesLocalUnknown(t *testing.T) {
465 config, _ := productionLocalConfig(t)
466 // The loader re-reads an absent identity on every attempt, so the probe is
467 // dynamic: it never dials and never spawns while no identity exists.
468 loader := broker.LocalIdentityLoader(func() (ports.BrokerDaemonIdentity, error) { return "", nil })
469 var dials atomic.Int32
470 connector := testLocalConnector(t, func(context.Context, ports.BrokerDialTarget) (daemonmux.RawFramedTransport, error) {
471 dials.Add(1)
472 return nil, errors.New("an absent identity must never dial")
473 })
474 local, err := brokerLocalObservation(config, brokerLocalTestEpoch, connector, loader)
475 require.NoError(t, err)
476
477 stateDir := filepath.Join(shortTempDir(t, "vevl"), "state")
478 store, err := brokerstore.Open(brokerstore.Options{Dir: stateDir})
479 require.NoError(t, err)
480 t.Cleanup(func() { require.NoError(t, store.Close()) })
481 registry, err := broker.NewRegistryWithConfig(brokerLocalTestEpoch, store, nil, clock.New(), discardLog(), broker.RegistryConfig{Local: local, MembershipMode: broker.MembershipMutable})
482 require.NoError(t, err)
483 startLocalRegistry(t, registry)
484
485 observation := waitLocalEntry(t, registry, func(entry ports.BrokerDaemonObservation) bool { return !entry.Checking })
486 require.True(t, observation.Local)
487 require.Equal(t, domain.RemoteAvailabilityUnknown, observation.Availability, "an unobserved local daemon is explicitly unknown, never invented")
488 require.Empty(t, observation.Identity)
489 require.True(t, observation.Incarnation.IsZero())
490 require.Zero(t, observation.ProtocolVersion)
491 require.False(t, observation.InventoryKnown)
492 require.Empty(t, observation.Sessions)
493
494 snapshot := registry.Snapshot()
495 require.NotZero(t, snapshot.Epoch, "a published snapshot always carries an epoch")
496 require.NotZero(t, snapshot.Revision, "a published snapshot always carries a nonzero revision")
497 require.Len(t, snapshot.Daemons, 1, "exactly one local entry is published")
498 require.True(t, snapshot.Daemons[0].Local)
499 require.Equal(t, int32(0), dials.Load(), "an absent identity never dials and never spawns")
500
501 // Never durable: the store is asked for its durable snapshot and the local
502 // entry is absent from it.
503 durable, err := store.Load()
504 require.NoError(t, err)
505 require.Empty(t, durable.Daemons, "a local observation is never durable")
506}
507
508// TestBrokerLocalObservationPublishesReachableCatalogueWhenIdentityAppears
509// proves the same broker starts observing without a restart once the daemon
510// publishes its identity: the loader is re-read on every attempt, so the next
511// observation authenticates the local carriage and publishes the reachable
512// catalogue.
513func TestBrokerLocalObservationPublishesReachableCatalogueWhenIdentityAppears(t *testing.T) {
514 config, binding := productionLocalConfig(t)
515 // The daemon publishes its own identity; the composition's binding starts
516 // empty and is filled by the loader the moment the daemon first starts.
517 daemonIdentity := ports.BrokerDaemonIdentity(brokerLocalTestIdentity)
518 serverBinding, err := daemonmux.NewServerBinding(daemonIdentity, ports.BrokerDaemonIncarnation{7, 1}, binding.Policy)
519 require.NoError(t, err)
520 session := testLocalUpSession(domain.SessionLifecycleID{5}, "appeared")
521 daemon := newScriptedLocalDaemon(t, serverBinding, func(message protocol.ClientMessage) protocol.ServerMessage {
522 request, ok := message.(protocol.CommandRequest)
523 if !ok {
524 return nil
525 }
526 return protocol.CommandResult{RequestID: request.RequestID, Outcome: protocol.CommandSucceeded, Output: testLocalCatalogue(t, session)}
527 })
528
529 // The identity is absent at first and appears before the next attempt; the
530 // scripted daemon answers under that identity.
531 var identity atomic.Value
532 identity.Store(ports.BrokerDaemonIdentity(""))
533 connector := testLocalConnector(t, daemon.dial())
534 local, err := brokerLocalObservation(config, brokerLocalTestEpoch, connector, func() (ports.BrokerDaemonIdentity, error) {
535 return identity.Load().(ports.BrokerDaemonIdentity), nil
536 })
537 require.NoError(t, err)
538
539 stateDir := filepath.Join(shortTempDir(t, "vevl"), "state")
540 store, err := brokerstore.Open(brokerstore.Options{Dir: stateDir})
541 require.NoError(t, err)
542 t.Cleanup(func() { require.NoError(t, store.Close()) })
543 // A deterministic clock: the identity retry is a jittered multi-second
544 // delay, so wall time would race the test's wait bound.
545 clk := newSandboxClock()
546 registry, err := broker.NewRegistryWithConfig(brokerLocalTestEpoch, store, nil, clk, discardLog(), broker.RegistryConfig{Local: local, MembershipMode: broker.MembershipMutable})
547 require.NoError(t, err)
548 startLocalRegistry(t, registry)
549
550 waitLocalEntry(t, registry, func(entry ports.BrokerDaemonObservation) bool { return !entry.Checking })
551 require.Equal(t, int32(0), daemon.dials.Load(), "no dial happens while no identity exists")
552
553 identity.Store(daemonIdentity)
554 // Step past the longest jittered retry (retry base plus 10%) so the next
555 // attempt reads the published identity.
556 clk.Advance(6 * time.Second)
557 observed := waitLocalEntry(t, registry, func(entry ports.BrokerDaemonObservation) bool {
558 return entry.Availability == domain.RemoteAvailabilityReachable
559 })
560 require.Equal(t, daemonIdentity, observed.Identity)
561 require.Equal(t, serverBinding.Incarnation(), observed.Incarnation)
562 require.Equal(t, protocol.Version, observed.ProtocolVersion)
563 require.True(t, observed.InventoryKnown)
564 require.Equal(t, []catalogue.RemoteCatalogSession{session}, observed.Sessions)
565 require.Positive(t, daemon.dials.Load())
566
567 // The published catalogue resolves the exact attach target the client
568 // composition uses, so the appearing daemon is immediately usable.
569 _, ok := brokerObservationExactTarget(observed, "appeared")
570 require.True(t, ok)
571}
572
573// TestBrokerLocalObservationNeverSpawnsDaemon proves the local observation is
574// spawn-free: every dial carries the existing-only authorization, so observing
575// a stopped daemon can never start one, and no dial happens without an identity.
576func TestBrokerLocalObservationNeverSpawnsDaemon(t *testing.T) {
577 _, binding := testLocalSandboxConfig(t)
578 var modes []ports.BrokerDaemonStartMode
579 var mu sync.Mutex
580 probe, err := newDynamicLocalRouteProbe(binding.Route, binding.Policy, brokerLocalTestEpoch, testEndpointConnectorFunc(func(_ context.Context, target ports.BrokerDialTarget) (ports.BrokerPhysicalConnection, error) {
581 mu.Lock()
582 modes = append(modes, target.StartMode)
583 mu.Unlock()
584 return nil, errors.New("socket absent")
585 }), func() (ports.BrokerDaemonIdentity, error) { return binding.Identity, nil })
586 require.NoError(t, err)
587
588 _, err = probe.ProbeLocal(context.Background())
589 require.Error(t, err)
590 mu.Lock()
591 require.Equal(t, []ports.BrokerDaemonStartMode{ports.BrokerDaemonExistingOnly}, modes, "an observation never authorizes starting the daemon")
592 mu.Unlock()
593}
594
595// TestOfflineRegistryConfigExplicitDisabledFixtureStaysDisabled proves an
596// TestProductionRegistryOwnsMutableMembershipWithZeroHosts proves the production
597// composition owns host membership for its whole run even when it starts with
598// no configured host at all: the broker may precede every registration, so a
599// first host can be added and removed through the same authority the zero-host
600// run already published.
601func TestProductionRegistryOwnsMutableMembershipWithZeroHosts(t *testing.T) {
602 config, _ := productionLocalConfig(t)
603 local, err := brokerLocalObservation(config, brokerLocalTestEpoch, unreachableCarriageConnector(t), func() (ports.BrokerDaemonIdentity, error) { return "", nil })
604 require.NoError(t, err)
605 selected, err := offlineRegistryConfig(config, brokerLocalTestEpoch, unreachableCarriageConnector(t))
606 require.NoError(t, err)
607 require.Equal(t, broker.MembershipMutable, selected.MembershipMode)
608
609 stateDir := filepath.Join(shortTempDir(t, "vevl"), "state")
610 store, err := brokerstore.Open(brokerstore.Options{Dir: stateDir})
611 require.NoError(t, err)
612 t.Cleanup(func() { require.NoError(t, store.Close()) })
613 // The production remote probe is required exactly when membership is added,
614 // so this fixture supplies one that never dials: the test asserts membership
615 // authority, not observation.
616 registry, err := broker.NewRegistryWithConfig(brokerLocalTestEpoch, store, unreachableHostProbe(t), clock.New(), discardLog(), broker.RegistryConfig{Local: local, MembershipMode: broker.MembershipMutable})
617 require.NoError(t, err)
618 startLocalRegistry(t, registry)
619
620 require.Empty(t, registry.Snapshot().Daemons[0].Registration, "the local entry carries no configured registration")
621 registration, err := registry.AddHost(context.Background(), "user@example.com", remoteBrokerPolicy(hostTransportQUIC))
622 require.NoError(t, err, "a zero-host production run must admit a first host")
623 require.Equal(t, "user@example.com", registration.Endpoint)
624
625 removed, err := registry.RemoveHost(context.Background(), registration)
626 require.NoError(t, err)
627 require.True(t, removed)
628}
629
630// TestBrokerLocalObservationRegistryPublishesLocalFirst proves the composition
631// end to end: the registry observes the local daemon through the broker-owned
632// probe and publishes it at index zero with the observed inventory, and the
633// durable store persists no local entry (the local observation is never
634// durable).
635func TestBrokerLocalObservationRegistryPublishesLocalFirst(t *testing.T) {
636 config, binding := testLocalSandboxConfig(t)
637 serverBinding := testServerBinding(t, binding)
638 session := testLocalUpSession(domain.SessionLifecycleID{9}, "registry")
639
640 daemon := newScriptedLocalDaemon(t, serverBinding, func(message protocol.ClientMessage) protocol.ServerMessage {
641 request, ok := message.(protocol.CommandRequest)
642 if !ok {
643 return nil
644 }
645 return protocol.CommandResult{RequestID: request.RequestID, Outcome: protocol.CommandSucceeded, Output: testLocalCatalogue(t, session)}
646 })
647 local, err := brokerLocalObservation(config, brokerLocalTestEpoch, testLocalConnector(t, daemon.dial()))
648 require.NoError(t, err)
649
650 stateDir := filepath.Join(shortTempDir(t, "vevs"), "state")
651 require.NoError(t, os.MkdirAll(stateDir, 0o700))
652 store, err := brokerstore.Open(brokerstore.Options{Dir: stateDir})
653 require.NoError(t, err)
654 t.Cleanup(func() { require.NoError(t, store.Close()) })
655
656 registry, err := broker.NewRegistryWithConfig(brokerLocalTestEpoch, store, nil, clock.New(), slog.New(slog.NewTextHandler(io.Discard, nil)), broker.RegistryConfig{Local: local})
657 require.NoError(t, err)
658
659 ctx, cancel := context.WithCancel(context.Background())
660 done := make(chan struct{})
661 go func() { defer close(done); registry.Run(ctx) }()
662 t.Cleanup(func() { cancel(); <-done })
663 registry.RequestProbe("")
664
665 require.Eventually(t, func() bool {
666 snapshot := registry.Snapshot()
667 return len(snapshot.Daemons) == 1 && snapshot.Daemons[0].Local &&
668 !snapshot.Daemons[0].Checking && snapshot.Daemons[0].Availability == domain.RemoteAvailabilityReachable
669 }, 3*time.Second, 2*time.Millisecond)
670
671 snapshot := registry.Snapshot()
672 require.Len(t, snapshot.Daemons, 1)
673 require.True(t, snapshot.Daemons[0].Local)
674 require.Equal(t, binding.Identity, snapshot.Daemons[0].Identity)
675 require.Equal(t, serverBinding.Incarnation(), snapshot.Daemons[0].Incarnation)
676 require.True(t, snapshot.Daemons[0].InventoryKnown)
677 require.Equal(t, []catalogue.RemoteCatalogSession{session}, snapshot.Daemons[0].Sessions)
678 require.Positive(t, daemon.dials.Load())
679
680 // The durable store persists remote observations only: the local entry never
681 // reaches disk.
682 require.Eventually(t, func() bool {
683 stored, err := store.Load()
684 return err == nil && stored.Revision > 0
685 }, 3*time.Second, 2*time.Millisecond)
686 stored, err := store.Load()
687 require.NoError(t, err)
688 for _, entry := range stored.Daemons {
689 require.False(t, entry.Local, "a local observation must never be durable")
690 }
691}
692
693// TestBrokerLocalProbeDialTargetIsExistingOnly proves the broker-owned local
694// observation dial target carries the existing-only authorization: an
695// observation must never start a stopped daemon, and the resolved target is
696// where that decision is carried to the transport.
697type testEndpointConnectorFunc func(context.Context, ports.BrokerDialTarget) (ports.BrokerPhysicalConnection, error)
698
699func (f testEndpointConnectorFunc) Connect(ctx context.Context, target ports.BrokerDialTarget) (ports.BrokerPhysicalConnection, error) {
700 return f(ctx, target)
701}
702
703func TestBrokerLocalProbeLoadsIdentityOnEveryAttempt(t *testing.T) {
704 _, binding := testLocalSandboxConfig(t)
705 var identity ports.BrokerDaemonIdentity
706 var dials []ports.BrokerDialTarget
707 probe, err := newDynamicLocalRouteProbe(binding.Route, binding.Policy, brokerLocalTestEpoch, testEndpointConnectorFunc(func(_ context.Context, target ports.BrokerDialTarget) (ports.BrokerPhysicalConnection, error) {
708 dials = append(dials, target)
709 return nil, errors.New("socket absent")
710 }), func() (ports.BrokerDaemonIdentity, error) {
711 return identity, nil
712 })
713 require.NoError(t, err)
714
715 observation, err := probe.ProbeLocal(context.Background())
716 require.NoError(t, err)
717 require.Equal(t, domain.RemoteAvailabilityUnknown, observation.Availability)
718 require.Empty(t, dials, "an absent identity cannot dial or spawn")
719
720 identity = binding.Identity
721 observation, err = probe.ProbeLocal(context.Background())
722 require.Error(t, err)
723 require.Equal(t, domain.RemoteAvailabilityUnreachable, observation.Availability)
724 require.Len(t, dials, 1)
725 require.Equal(t, ports.BrokerExpectedIdentity{Identity: binding.Identity, Bound: true}, dials[0].ExpectedIdentity)
726 require.Equal(t, ports.BrokerDaemonExistingOnly, dials[0].StartMode)
727}
728
729func TestBrokerLocalProbeIdentityCorruptionIsDiagnosticWithoutDial(t *testing.T) {
730 _, binding := testLocalSandboxConfig(t)
731 dialed := false
732 probe, err := newDynamicLocalRouteProbe(binding.Route, binding.Policy, brokerLocalTestEpoch, testEndpointConnectorFunc(func(context.Context, ports.BrokerDialTarget) (ports.BrokerPhysicalConnection, error) {
733 dialed = true
734 return nil, nil
735 }), func() (ports.BrokerDaemonIdentity, error) {
736 return "", daemonidentity.ErrInvalid
737 })
738 require.NoError(t, err)
739 observation, err := probe.ProbeLocal(context.Background())
740 require.ErrorIs(t, err, daemonidentity.ErrInvalid)
741 require.Equal(t, domain.RemoteAvailabilityUnreachable, observation.Availability)
742 require.False(t, dialed)
743}
744
745func TestBrokerLocalProbeDialTargetIsExistingOnly(t *testing.T) {
746 _, binding := testLocalSandboxConfig(t)
747 var target ports.BrokerDialTarget
748 probe, err := newLocalRouteProbe(binding, brokerLocalTestEpoch, func(_ context.Context, observed ports.BrokerDialTarget) (daemonmux.RawFramedTransport, error) {
749 target = observed
750 return nil, errors.New("unreachable")
751 }, daemonmux.DefaultMuxCeilings())
752 require.NoError(t, err)
753 _, err = probe.ProbeLocal(context.Background())
754 require.Error(t, err)
755 require.Equal(t, ports.BrokerDaemonExistingOnly, target.StartMode,
756 "an observation dial target never authorizes starting the daemon")
757 require.True(t, target.Fence.Local)
758 require.Equal(t, binding.Identity, target.ExpectedIdentity.Identity)
759}
760
761// TestBrokerLocalProbeRefusesNonLocalRoute proves the probe refuses a
762// hand-built binding whose route is an ssh helper route: probing must never
763// start a child process, so only a Unix route this process dials directly is
764// acceptable. brokerconfig exposes no exported route constructor and its route
765// fields are unexported, so the ssh route is obtained the only way this package
766// can: through a provisioned ssh registration (loadTestRoute, shared with the
767// mux tests) and used directly as the local binding's route.
768func TestBrokerLocalProbeRefusesNonLocalRoute(t *testing.T) {
769 isolateSandboxEnv(t)
770 sshRoute := loadTestRoute(t, map[string]any{
771 "kind": "ssh-stdio",
772 "target": "user@host:2222",
773 "argv": []any{"vev", brokerMuxStdioCommand, "--offline-root", "/srv/remote"},
774 })
775 require.Equal(t, brokerconfig.RouteSSHStdio, sshRoute.Kind())
776 require.False(t, sshRoute.IsLocal(), "an ssh helper route is never dialed directly")
777
778 binding := brokerconfig.LocalBinding{
779 Identity: brokerLocalTestIdentity,
780 DisplayOrigin: "local",
781 Policy: brokerTestPolicy(),
782 Route: sshRoute,
783 }
784 _, err := newLocalRouteProbe(binding, brokerLocalTestEpoch, func(context.Context, ports.BrokerDialTarget) (daemonmux.RawFramedTransport, error) {
785 t.Fatal("an ssh route must never be dialed by the local probe")
786 return nil, nil
787 }, daemonmux.DefaultMuxCeilings())
788 require.Error(t, err)
789 require.Contains(t, err.Error(), "local route")
790}
791
792// startLocalRegistry runs one registry for the duration of a test.
793func startLocalRegistry(t *testing.T, registry *broker.Registry) {
794 t.Helper()
795 registry.SetDemand(true)
796 ctx, cancel := context.WithCancel(context.Background())
797 done := make(chan struct{})
798 go func() { defer close(done); registry.Run(ctx) }()
799 t.Cleanup(func() { cancel(); <-done; registry.SetDemand(false) })
800}
801
802// waitLocalEntry waits for the published local entry (always index zero) to
803// satisfy predicate and returns it.
804func waitLocalEntry(t *testing.T, registry *broker.Registry, predicate func(ports.BrokerDaemonObservation) bool) ports.BrokerDaemonObservation {
805 t.Helper()
806 sub := registry.Subscribe()
807 defer sub.Close()
808 deadline := time.After(brokerTestWait)
809 for {
810 snapshot := registry.Snapshot()
811 if len(snapshot.Daemons) > 0 && snapshot.Daemons[0].Local && predicate(snapshot.Daemons[0]) {
812 return snapshot.Daemons[0]
813 }
814 select {
815 case <-sub.Changed():
816 case <-deadline:
817 t.Fatal("timed out waiting for the published local entry")
818 return ports.BrokerDaemonObservation{}
819 }
820 }
821}
822
823// noDialHostProbe is a ports.BrokerHostProbe that never dials: a membership test
824// supplies it only because the registry requires a remote probe to observe
825// hosts, never to own their authority.
826type noDialHostProbe struct{}
827
828func (noDialHostProbe) Probe(context.Context, domain.RemoteRegistration) (ports.BrokerDaemonObservation, error) {
829 return ports.BrokerDaemonObservation{}, errors.New("membership authority must not dial")
830}
831
832func unreachableHostProbe(t *testing.T) ports.BrokerHostProbe {
833 t.Helper()
834 return noDialHostProbe{}
835}