1package client
2
3import (
4 "errors"
5 "fmt"
6 "sync"
7 "testing"
8 "time"
9
10 "github.com/stretchr/testify/require"
11
12 "github.com/bnema/vev/internal/domain"
13 "github.com/bnema/vev/internal/ports"
14 "github.com/bnema/vev/internal/protocol"
15 "github.com/bnema/vev/internal/protocol/catalogue"
16)
17
18// Client-owned picker catalogue tests (Plan 001 P5.2b). Every case is
19// deterministic: the fake clock, not wall time, decides freshness, and no case
20// sleeps.
21
22const pickerTestFreshness = 30 * time.Second
23
24// pickerCatalogueErrorIs reports whether err is a typed catalogue refusal with
25// the supplied code, so tests classify a resolution failure without matching
26// message text.
27func pickerCatalogueErrorIs(err error, code pickerCatalogueErrorCode) bool {
28 var typed pickerCatalogueError
29 return errors.As(err, &typed) && typed.Code == code
30}
31
32func pickerTestPolicy() ports.BrokerPolicy {
33 return ports.BrokerPolicy{
34 ProtocolVersion: protocol.Version,
35 CatalogSchemaVersion: catalogue.RemoteCatalogSchemaVersion,
36 EnvironmentPolicy: protocol.EnvironmentPolicyDaemonOwned,
37 Transport: "quic",
38 Trust: "trust",
39 Launch: "launch",
40 Isolation: "user",
41 }
42}
43
44func pickerTestIncarnation(seed byte) [16]byte {
45 var id [16]byte
46 id[0] = seed
47 return id
48}
49
50func pickerTestRegistration(endpoint string, seed byte) domain.RemoteRegistration {
51 return domain.RemoteRegistration{Endpoint: endpoint, Incarnation: pickerTestIncarnation(seed), Generation: 1}
52}
53
54func pickerTestLifecycle(seed byte) domain.SessionLifecycleID { return pickerTestIncarnation(seed) }
55
56func pickerTestSession(name string, seed byte, state catalogue.RemoteCatalogSessionState) catalogue.RemoteCatalogSession {
57 return catalogue.RemoteCatalogSession{LifecycleID: pickerTestLifecycle(seed), Name: name, State: state}
58}
59
60func pickerTestSessionWithLifecycle(name string, lifecycle domain.SessionLifecycleID) catalogue.RemoteCatalogSession {
61 return catalogue.RemoteCatalogSession{LifecycleID: lifecycle, Name: name, State: catalogue_Up}
62}
63
64func pickerTestLocalObservation(now time.Time, sessions ...catalogue.RemoteCatalogSession) ports.BrokerDaemonObservation {
65 return ports.BrokerDaemonObservation{
66 Local: true,
67 DisplayOrigin: "local",
68 Policy: pickerTestPolicy(),
69 Identity: "local-daemon",
70 Incarnation: ports.BrokerDaemonIncarnation(pickerTestIncarnation(9)),
71 ProtocolVersion: protocol.Version,
72 Availability: domain.RemoteAvailabilityReachable,
73 LastSuccess: now,
74 InventoryKnown: true,
75 Sessions: sessions,
76 }
77}
78
79func pickerTestRemoteObservation(endpoint string, registrationSeed, incarnationSeed byte, now time.Time, sessions ...catalogue.RemoteCatalogSession) ports.BrokerDaemonObservation {
80 return ports.BrokerDaemonObservation{
81 Endpoint: endpoint,
82 DisplayOrigin: endpoint,
83 Registration: pickerTestRegistration(endpoint, registrationSeed),
84 Policy: pickerTestPolicy(),
85 Identity: ports.BrokerDaemonIdentity("identity-" + endpoint),
86 Incarnation: ports.BrokerDaemonIncarnation(pickerTestIncarnation(incarnationSeed)),
87 ProtocolVersion: protocol.Version,
88 Availability: domain.RemoteAvailabilityReachable,
89 LastSuccess: now,
90 InventoryKnown: true,
91 Sessions: sessions,
92 }
93}
94
95func pickerTestUnobservedObservation(now time.Time, sessions ...catalogue.RemoteCatalogSession) ports.BrokerDaemonObservation {
96 return ports.BrokerDaemonObservation{
97 Local: true,
98 DisplayOrigin: "local",
99 Policy: pickerTestPolicy(),
100 Availability: domain.RemoteAvailabilityUnknown,
101 InventoryKnown: len(sessions) > 0,
102 Sessions: sessions,
103 }
104}
105
106func advancePickerClock(clock *supervisorTestClock, delta time.Duration) {
107 clock.mu.Lock()
108 clock.now = clock.now.Add(delta)
109 clock.mu.Unlock()
110}
111
112func pickerTestCatalogue(t *testing.T) (*pickerCatalogue, *supervisorTestClock) {
113 t.Helper()
114 clock := newSupervisorTestClock()
115 return newPickerCatalogue(pickerCatalogueConfig{Clock: clock, Freshness: pickerTestFreshness}), clock
116}
117
118func pickerLineByLabel(lines []protocol.PickerLine, label string) (protocol.PickerLine, bool) {
119 for _, line := range lines {
120 if line.Kind != protocol.PickerLineSection && line.Label == label {
121 return line, true
122 }
123 }
124 return protocol.PickerLine{}, false
125}
126
127func pickerSectionByLabel(lines []protocol.PickerLine, label string) (protocol.PickerLine, bool) {
128 for _, line := range lines {
129 if line.Kind == protocol.PickerLineSection && line.Label == label {
130 return line, true
131 }
132 }
133 return protocol.PickerLine{}, false
134}
135
136func pickerHostLine(lines []protocol.PickerLine) (protocol.PickerLine, bool) {
137 for _, line := range lines {
138 if line.Kind == protocol.PickerLineHost {
139 return line, true
140 }
141 }
142 return protocol.PickerLine{}, false
143}
144
145func pickerSectionLabels(lines []protocol.PickerLine) []string {
146 sections := make([]string, 0)
147 for _, line := range lines {
148 if line.Kind == protocol.PickerLineSection {
149 sections = append(sections, line.Label)
150 }
151 }
152 return sections
153}
154
155func pickerSessionLabels(lines []protocol.PickerLine) []string {
156 labels := make([]string, 0)
157 for _, line := range lines {
158 if line.Kind == protocol.PickerLineSession {
159 labels = append(labels, line.Label)
160 }
161 }
162 return labels
163}
164
165func pickerTestBase() pickerResolveBase {
166 return pickerResolveBase{Connection: ports.BrokerConnectionID{1}, Stream: ports.BrokerStreamID(1)}
167}
168
169// localDaemonSnapshot publishes one fresh, reachable, compatible local daemon
170// observation with no sessions, which is what the supervisor's readiness gate
171// and its initial navigation both need.
172func localDaemonSnapshot(epoch ports.BrokerEpoch, revision ports.BrokerRevision) ports.BrokerSnapshot {
173 return ports.BrokerSnapshot{Epoch: epoch, Revision: revision, Daemons: []ports.BrokerDaemonObservation{
174 pickerTestLocalObservation(time.Unix(1000, 0)),
175 }}
176}
177
178func TestPickerCatalogueEmptySnapshot(t *testing.T) {
179 catalogue, _ := pickerTestCatalogue(t)
180 require.False(t, catalogue.Apply(ports.BrokerSnapshot{Epoch: 0, Revision: 1}), "an invalid snapshot is never applied")
181 require.True(t, catalogue.Apply(ports.BrokerSnapshot{Epoch: 5, Revision: 1}))
182 require.Empty(t, catalogue.Lines())
183 require.Equal(t, ports.BrokerEpoch(5), catalogue.Epoch())
184 require.Equal(t, ports.BrokerRevision(1), catalogue.Revision())
185}
186
187func TestPickerCatalogueProjectionCases(t *testing.T) {
188 clock := newSupervisorTestClock()
189 now := clock.Now()
190 navigateKill := protocol.PickerCanNavigate | protocol.PickerCanKill
191
192 tests := []struct {
193 name string
194 daemons []ports.BrokerDaemonObservation
195 wantSections []string
196 wantSessions []string
197 wantActions map[string]protocol.PickerLineActions
198 wantStatus map[string]protocol.PickerLineStatus
199 wantSection map[string]protocol.PickerLineStatus
200 wantDim map[string]bool
201 wantHostRow bool
202 wantNoDaemon bool
203 }{
204 {
205 name: "local only",
206 daemons: []ports.BrokerDaemonObservation{pickerTestLocalObservation(now, pickerTestSession("alpha", 1, catalogue_Up), pickerTestSession("beta", 2, catalogue_Up))},
207 wantSections: []string{"local"},
208 wantSessions: []string{"alpha", "beta"},
209 wantActions: map[string]protocol.PickerLineActions{"alpha": navigateKill, "beta": navigateKill},
210 },
211 {
212 name: "unobserved local daemon with no session shows its status",
213 daemons: []ports.BrokerDaemonObservation{pickerTestUnobservedObservation(now)},
214 wantSections: []string{"local"},
215 wantSessions: []string{},
216 wantStatus: map[string]protocol.PickerLineStatus{"local": protocol.PickerLineStatusStale},
217 wantHostRow: true,
218 },
219 {
220 name: "no local daemon entry",
221 daemons: []ports.BrokerDaemonObservation{pickerTestRemoteObservation("user@arch", 1, 1, now, pickerTestSession("remote-a", 3, catalogue_Up))},
222 wantSections: []string{"user@arch"},
223 wantSessions: []string{"remote-a"},
224 wantActions: map[string]protocol.PickerLineActions{"remote-a": protocol.PickerCanNavigate},
225 },
226 {
227 name: "reachable host with no daemon offers a create row", wantNoDaemon: true,
228 daemons: []ports.BrokerDaemonObservation{
229 func() ports.BrokerDaemonObservation {
230 observation := pickerTestRemoteObservation("user@arch", 1, 1, now)
231 observation.Identity = ""
232 observation.Incarnation = ports.BrokerDaemonIncarnation{}
233 observation.ProtocolVersion = 0
234 observation.Availability = domain.RemoteAvailabilityNoDaemon
235 observation.LastSuccess = time.Time{}
236 observation.InventoryKnown = false
237 return observation
238 }(),
239 },
240 wantSections: []string{"user@arch"},
241 wantSessions: []string{},
242 wantStatus: map[string]protocol.PickerLineStatus{"user@arch": protocol.PickerLineStatusNoDaemon},
243 wantHostRow: true,
244 },
245 {
246 name: "stale observation stays explicitly selectable",
247 daemons: []ports.BrokerDaemonObservation{
248 func() ports.BrokerDaemonObservation {
249 observation := pickerTestLocalObservation(now, pickerTestSession("alpha", 1, catalogue_Up))
250 observation.LastSuccess = now.Add(-2 * pickerTestFreshness)
251 return observation
252 }(),
253 },
254 wantSections: []string{"local"},
255 wantSessions: []string{"alpha"},
256 wantActions: map[string]protocol.PickerLineActions{"alpha": navigateKill},
257 wantStatus: map[string]protocol.PickerLineStatus{"alpha": protocol.PickerLineStatusNone},
258 wantSection: map[string]protocol.PickerLineStatus{"local": protocol.PickerLineStatusNone},
259 wantDim: map[string]bool{"alpha": true},
260 },
261 {
262 name: "unavailable host keeps its rows and hides the redundant status row",
263 daemons: []ports.BrokerDaemonObservation{
264 func() ports.BrokerDaemonObservation {
265 observation := pickerTestRemoteObservation("user@arch", 1, 1, now, pickerTestSession("remote-a", 3, catalogue_Up))
266 observation.Availability = domain.RemoteAvailabilityUnreachable
267 return observation
268 }(),
269 },
270 wantSections: []string{"user@arch"},
271 wantSessions: []string{"remote-a"},
272 wantActions: map[string]protocol.PickerLineActions{"remote-a": protocol.PickerCanNavigate},
273 wantStatus: map[string]protocol.PickerLineStatus{"remote-a": protocol.PickerLineStatusNone},
274 wantSection: map[string]protocol.PickerLineStatus{"user@arch": protocol.PickerLineStatusDown},
275 wantDim: map[string]bool{"remote-a": true},
276 },
277 {
278 name: "unavailable host without sessions shows its status row",
279 daemons: []ports.BrokerDaemonObservation{
280 func() ports.BrokerDaemonObservation {
281 observation := pickerTestRemoteObservation("user@arch", 1, 1, now)
282 observation.Availability = domain.RemoteAvailabilityUnreachable
283 return observation
284 }(),
285 },
286 wantSections: []string{"user@arch"},
287 wantSessions: []string{},
288 wantStatus: map[string]protocol.PickerLineStatus{"user@arch": protocol.PickerLineStatusDown},
289 wantSection: map[string]protocol.PickerLineStatus{"user@arch": protocol.PickerLineStatusNone},
290 wantHostRow: true,
291 },
292 {
293 name: "incompatible host",
294 daemons: []ports.BrokerDaemonObservation{
295 func() ports.BrokerDaemonObservation {
296 observation := pickerTestRemoteObservation("user@arch", 1, 1, now, pickerTestSession("remote-a", 3, catalogue_Up))
297 observation.ProtocolVersion = protocol.Version + 1
298 return observation
299 }(),
300 },
301 wantSections: []string{"user@arch"},
302 wantSessions: []string{"remote-a"},
303 wantActions: map[string]protocol.PickerLineActions{"remote-a": 0},
304 wantStatus: map[string]protocol.PickerLineStatus{"remote-a": protocol.PickerLineStatusNone},
305 wantSection: map[string]protocol.PickerLineStatus{"user@arch": protocol.PickerLineStatusVersion},
306 wantDim: map[string]bool{"remote-a": true},
307 },
308 {
309 name: "broken session is only killable",
310 daemons: []ports.BrokerDaemonObservation{pickerTestLocalObservation(now, pickerTestSession("alpha", 1, catalogue_Broken))},
311 wantSections: []string{"local"},
312 wantSessions: []string{"alpha"},
313 wantActions: map[string]protocol.PickerLineActions{"alpha": protocol.PickerCanKill},
314 wantStatus: map[string]protocol.PickerLineStatus{"alpha": protocol.PickerLineStatusError},
315 wantDim: map[string]bool{"alpha": true},
316 },
317 }
318
319 for _, tt := range tests {
320 t.Run(tt.name, func(t *testing.T) {
321 catalogue, _ := pickerTestCatalogue(t)
322 require.True(t, catalogue.Apply(ports.BrokerSnapshot{Epoch: 7, Revision: 1, Daemons: tt.daemons}))
323 lines := catalogue.Lines()
324 require.Equal(t, tt.wantSections, pickerSectionLabels(lines))
325 require.Equal(t, tt.wantSessions, pickerSessionLabels(lines))
326 hostLine, hasHost := pickerHostLine(lines)
327 require.Equal(t, tt.wantHostRow, hasHost, "a host status row is published only when it adds information")
328 if tt.wantNoDaemon {
329 require.Equal(t, protocol.PickerLineStatusNoDaemon, hostLine.Status)
330 require.Equal(t, protocol.PickerCanNavigate, hostLine.Actions, "Enter creates an ephemeral session on this host")
331 require.True(t, hostLine.Focusable)
332 require.Contains(t, hostLine.StatusDetail, "no_daemon")
333 }
334 for label, actions := range tt.wantActions {
335 line, ok := pickerLineByLabel(lines, label)
336 require.True(t, ok, "session %q must be projected", label)
337 require.True(t, line.Focusable, "a projected row is always a cursor destination")
338 require.Equal(t, actions, line.Actions, "row %q actions", label)
339 }
340 for label, status := range tt.wantStatus {
341 line, ok := pickerLineByLabel(lines, label)
342 require.True(t, ok, "row %q must be projected", label)
343 require.Equal(t, status, line.Status, "row %q badge", label)
344 }
345 for label, status := range tt.wantSection {
346 line, ok := pickerSectionByLabel(lines, label)
347 require.True(t, ok, "section %q must be projected", label)
348 require.Equal(t, status, line.Status, "section %q problem dot", label)
349 }
350 for label, dim := range tt.wantDim {
351 line, ok := pickerLineByLabel(lines, label)
352 require.True(t, ok)
353 require.Equal(t, dim, line.Dim, "row %q dim", label)
354 }
355 })
356 }
357}
358
359func TestPickerCatalogueMonotonicUpdates(t *testing.T) {
360 clock := newSupervisorTestClock()
361 now := clock.Now()
362 catalogue, _ := pickerTestCatalogue(t)
363
364 newer := ports.BrokerSnapshot{Epoch: 3, Revision: 4, Daemons: []ports.BrokerDaemonObservation{
365 pickerTestLocalObservation(now, pickerTestSession("alpha", 1, catalogue_Up)),
366 }}
367 require.True(t, catalogue.Apply(newer))
368 before := catalogue.Lines()
369
370 // A lower revision in the same epoch never supersedes and never regresses.
371 older := ports.BrokerSnapshot{Epoch: 3, Revision: 2, Daemons: []ports.BrokerDaemonObservation{
372 pickerTestLocalObservation(now, pickerTestSession("beta", 2, catalogue_Up)),
373 }}
374 require.False(t, catalogue.Apply(older))
375 require.Equal(t, before, catalogue.Lines())
376 require.Equal(t, ports.BrokerRevision(4), catalogue.Revision())
377
378 // A new epoch is a full resynchronization and replaces every row.
379 epoch := ports.BrokerSnapshot{Epoch: 4, Revision: 1, Daemons: []ports.BrokerDaemonObservation{
380 pickerTestLocalObservation(now, pickerTestSession("gamma", 3, catalogue_Up)),
381 }}
382 require.True(t, catalogue.Apply(epoch))
383 require.Equal(t, []string{"gamma"}, pickerSessionLabels(catalogue.Lines()))
384}
385
386// catalogue_Up/Broken keep the tables above short without importing the
387// catalogue state into every literal.
388const (
389 catalogue_Up = catalogue.RemoteCatalogSessionState("up")
390 catalogue_Down = catalogue.RemoteCatalogSessionState("down")
391 catalogue_Broken = catalogue.RemoteCatalogSessionState("broken")
392)
393
394func TestPickerCatalogueSlowObservationKeepsNewerState(t *testing.T) {
395 clock := newSupervisorTestClock()
396 now := clock.Now()
397 catalogue, _ := pickerTestCatalogue(t)
398
399 fresh := pickerTestRemoteObservation("user@arch", 1, 1, now, pickerTestSession("alpha", 1, catalogue_Up))
400 require.True(t, catalogue.Apply(ports.BrokerSnapshot{Epoch: 3, Revision: 2, Daemons: []ports.BrokerDaemonObservation{fresh}}))
401
402 regressed := pickerTestRemoteObservation("user@arch", 1, 1, now.Add(-time.Minute), pickerTestSession("beta", 2, catalogue_Up))
403 require.True(t, catalogue.Apply(ports.BrokerSnapshot{Epoch: 3, Revision: 3, Daemons: []ports.BrokerDaemonObservation{regressed}}), "the publication supersedes")
404
405 require.Equal(t, []string{"alpha"}, pickerSessionLabels(catalogue.Lines()), "an older observed inventory never regresses a row")
406}
407
408func TestPickerCatalogueIdentityReplacement(t *testing.T) {
409 clock := newSupervisorTestClock()
410 now := clock.Now()
411 catalogue, _ := pickerTestCatalogue(t)
412
413 original := ports.BrokerSnapshot{Epoch: 3, Revision: 1, Daemons: []ports.BrokerDaemonObservation{
414 pickerTestRemoteObservation("user@arch", 1, 1, now, pickerTestSession("alpha", 1, catalogue_Up)),
415 }}
416 require.True(t, catalogue.Apply(original))
417 oldLine, ok := pickerLineByLabel(catalogue.Lines(), "alpha")
418 require.True(t, ok)
419 oldKey := oldLine.Key
420
421 replaced := ports.BrokerSnapshot{Epoch: 3, Revision: 2, Daemons: []ports.BrokerDaemonObservation{
422 pickerTestRemoteObservation("user@arch", 2, 2, now, pickerTestSession("alpha", 9, catalogue_Up)),
423 }}
424 require.True(t, catalogue.Apply(replaced))
425 newLine, ok := pickerLineByLabel(catalogue.Lines(), "alpha")
426 require.True(t, ok)
427 require.NotEqual(t, oldKey, newLine.Key, "a replaced identity gets a new opaque key")
428
429 // The old key is still recognized and refused as a replacement, not silently
430 // re-bound to the new session.
431 _, err := catalogue.Resolve(oldKey, pickerTestBase())
432 require.Error(t, err)
433 require.True(t, pickerCatalogueErrorIs(err, pickerCatalogueReplaced), "same endpoint with a new incarnation is a replacement")
434}
435
436// TestPickerCatalogueRegistrationGenerationChangeRequiresFreshSelection pins the
437// complete-registration equality fence on an interactive selection: a
438// registration whose generation advanced under the same endpoint and
439// incarnation is a replacement, so the displayed row cannot silently re-bind
440// to changed authority.
441func TestPickerCatalogueRegistrationGenerationChangeRequiresFreshSelection(t *testing.T) {
442 clock := newSupervisorTestClock()
443 now := clock.Now()
444 catalogue, _ := pickerTestCatalogue(t)
445
446 require.True(t, catalogue.Apply(ports.BrokerSnapshot{Epoch: 3, Revision: 1, Daemons: []ports.BrokerDaemonObservation{
447 pickerTestRemoteObservation("user@arch", 1, 1, now, pickerTestSession("alpha", 1, catalogue_Up)),
448 }}))
449 line, ok := pickerLineByLabel(catalogue.Lines(), "alpha")
450 require.True(t, ok)
451
452 advanced := pickerTestRemoteObservation("user@arch", 1, 1, now, pickerTestSession("alpha", 1, catalogue_Up))
453 advanced.Registration.Generation = 2
454 require.True(t, catalogue.Apply(ports.BrokerSnapshot{Epoch: 3, Revision: 2, Daemons: []ports.BrokerDaemonObservation{advanced}}))
455
456 _, err := catalogue.Resolve(line.Key, pickerTestBase())
457 require.Error(t, err)
458 require.True(t, pickerCatalogueErrorIs(err, pickerCatalogueReplaced), "an advanced generation is a replacement: %v", err)
459}
460
461func TestPickerCatalogueStableRowOrder(t *testing.T) {
462 clock := newSupervisorTestClock()
463 now := clock.Now()
464 catalogue, _ := pickerTestCatalogue(t)
465
466 first := ports.BrokerSnapshot{Epoch: 3, Revision: 1, Daemons: []ports.BrokerDaemonObservation{
467 pickerTestLocalObservation(now, pickerTestSession("alpha", 1, catalogue_Up)),
468 pickerTestRemoteObservation("host-a", 1, 1, now, pickerTestSession("a-one", 2, catalogue_Up)),
469 pickerTestRemoteObservation("host-b", 2, 2, now, pickerTestSession("b-one", 3, catalogue_Up)),
470 }}
471 require.True(t, catalogue.Apply(first))
472 require.Equal(t, []string{"local", "host-a", "host-b"}, pickerSectionLabels(catalogue.Lines()))
473
474 // The publication reorders the remote hosts and changes one unrelated
475 // session; the first-seen order never reshuffles. The local daemon stays
476 // first, as every valid broker publication requires.
477 second := ports.BrokerSnapshot{Epoch: 3, Revision: 2, Daemons: []ports.BrokerDaemonObservation{
478 pickerTestLocalObservation(now, pickerTestSession("alpha", 1, catalogue_Up)),
479 pickerTestRemoteObservation("host-b", 2, 2, now, pickerTestSession("b-one", 3, catalogue_Up), pickerTestSession("b-two", 4, catalogue_Up)),
480 pickerTestRemoteObservation("host-a", 1, 1, now, pickerTestSession("a-one", 2, catalogue_Up)),
481 }}
482 require.True(t, catalogue.Apply(second))
483 require.Equal(t, []string{"local", "host-a", "host-b"}, pickerSectionLabels(catalogue.Lines()))
484 require.Equal(t, []string{"alpha", "a-one", "b-one", "b-two"}, pickerSessionLabels(catalogue.Lines()))
485}
486
487func TestPickerCatalogueDeterministicKeys(t *testing.T) {
488 clock := newSupervisorTestClock()
489 now := clock.Now()
490 daemons := []ports.BrokerDaemonObservation{
491 pickerTestLocalObservation(now, pickerTestSession("alpha", 1, catalogue_Up)),
492 pickerTestRemoteObservation("host-a", 1, 1, now, pickerTestSession("a-one", 2, catalogue_Up)),
493 }
494 first, _ := pickerTestCatalogue(t)
495 second, _ := pickerTestCatalogue(t)
496 require.True(t, first.Apply(ports.BrokerSnapshot{Epoch: 3, Revision: 1, Daemons: daemons}))
497 require.True(t, second.Apply(ports.BrokerSnapshot{Epoch: 3, Revision: 1, Daemons: daemons}))
498 require.Equal(t, first.Lines(), second.Lines())
499}
500
501func TestPickerCatalogueSelectionResolution(t *testing.T) {
502 clock := newSupervisorTestClock()
503 now := clock.Now()
504 base := pickerTestBase()
505
506 tests := []struct {
507 name string
508 build func(t *testing.T, catalogue *pickerCatalogue, clock *supervisorTestClock) string
509 wantErr pickerCatalogueErrorCode
510 wantLocal bool
511 wantAdmission ports.BrokerStreamAdmission
512 }{
513 {
514 name: "local exact attaches to the lifecycle",
515 build: func(t *testing.T, catalogue *pickerCatalogue, _ *supervisorTestClock) string {
516 require.True(t, catalogue.Apply(ports.BrokerSnapshot{Epoch: 3, Revision: 1, Daemons: []ports.BrokerDaemonObservation{
517 pickerTestLocalObservation(now, pickerTestSession("alpha", 1, catalogue_Up)),
518 }}))
519 line, ok := pickerLineByLabel(catalogue.Lines(), "alpha")
520 require.True(t, ok)
521 return line.Key
522 },
523 wantLocal: true,
524 wantAdmission: ports.BrokerAdmissionExact,
525 },
526 {
527 name: "remote exact carries the configured authority",
528 build: func(t *testing.T, catalogue *pickerCatalogue, _ *supervisorTestClock) string {
529 require.True(t, catalogue.Apply(ports.BrokerSnapshot{Epoch: 3, Revision: 1, Daemons: []ports.BrokerDaemonObservation{
530 pickerTestRemoteObservation("user@arch", 1, 1, now, pickerTestSession("remote-a", 2, catalogue_Up)),
531 }}))
532 line, ok := pickerLineByLabel(catalogue.Lines(), "remote-a")
533 require.True(t, ok)
534 return line.Key
535 },
536 wantAdmission: ports.BrokerAdmissionExact,
537 },
538 {
539 name: "gone host",
540 build: func(t *testing.T, catalogue *pickerCatalogue, _ *supervisorTestClock) string {
541 require.True(t, catalogue.Apply(ports.BrokerSnapshot{Epoch: 3, Revision: 1, Daemons: []ports.BrokerDaemonObservation{
542 pickerTestRemoteObservation("user@arch", 1, 1, now, pickerTestSession("remote-a", 2, catalogue_Up)),
543 }}))
544 line, ok := pickerLineByLabel(catalogue.Lines(), "remote-a")
545 require.True(t, ok)
546 key := line.Key
547 require.True(t, catalogue.Apply(ports.BrokerSnapshot{Epoch: 3, Revision: 2}))
548 return key
549 },
550 wantErr: pickerCatalogueGone,
551 },
552 {
553 name: "stale observation is still explicitly selectable",
554 build: func(t *testing.T, catalogue *pickerCatalogue, clock *supervisorTestClock) string {
555 require.True(t, catalogue.Apply(ports.BrokerSnapshot{Epoch: 3, Revision: 1, Daemons: []ports.BrokerDaemonObservation{
556 pickerTestLocalObservation(now, pickerTestSession("alpha", 1, catalogue_Up)),
557 }}))
558 line, ok := pickerLineByLabel(catalogue.Lines(), "alpha")
559 require.True(t, ok)
560 advancePickerClock(clock, 2*pickerTestFreshness)
561 return line.Key
562 },
563 wantLocal: true,
564 wantAdmission: ports.BrokerAdmissionExact,
565 },
566 {
567 name: "incompatible host",
568 build: func(t *testing.T, catalogue *pickerCatalogue, _ *supervisorTestClock) string {
569 incompatible := pickerTestRemoteObservation("user@arch", 1, 1, now, pickerTestSession("remote-a", 2, catalogue_Up))
570 incompatible.ProtocolVersion = protocol.Version + 1
571 require.True(t, catalogue.Apply(ports.BrokerSnapshot{Epoch: 3, Revision: 1, Daemons: []ports.BrokerDaemonObservation{incompatible}}))
572 line, ok := pickerLineByLabel(catalogue.Lines(), "remote-a")
573 require.True(t, ok)
574 return line.Key
575 },
576 wantErr: pickerCatalogueIncompatible,
577 },
578 {
579 // Plan 003 B4: a remote host the broker observed failing refuses
580 // instantly instead of a slow attempt that fails the same way.
581 name: "unreachable remote host refuses instantly",
582 build: func(t *testing.T, catalogue *pickerCatalogue, _ *supervisorTestClock) string {
583 unavailable := pickerTestRemoteObservation("user@arch", 1, 1, now, pickerTestSession("remote-a", 2, catalogue_Up))
584 unavailable.Availability = domain.RemoteAvailabilityUnreachable
585 require.True(t, catalogue.Apply(ports.BrokerSnapshot{Epoch: 3, Revision: 1, Daemons: []ports.BrokerDaemonObservation{unavailable}}))
586 line, ok := pickerLineByLabel(catalogue.Lines(), "remote-a")
587 require.True(t, ok)
588 return line.Key
589 },
590 wantErr: pickerCatalogueUnavailable,
591 },
592 {
593 name: "stale reachable remote host stays attemptable",
594 build: func(t *testing.T, catalogue *pickerCatalogue, _ *supervisorTestClock) string {
595 stale := pickerTestRemoteObservation("user@arch", 1, 1, now.Add(-24*time.Hour), pickerTestSession("remote-a", 2, catalogue_Up))
596 require.True(t, catalogue.Apply(ports.BrokerSnapshot{Epoch: 3, Revision: 1, Daemons: []ports.BrokerDaemonObservation{stale}}))
597 line, ok := pickerLineByLabel(catalogue.Lines(), "remote-a")
598 require.True(t, ok)
599 return line.Key
600 },
601 wantAdmission: ports.BrokerAdmissionExact,
602 },
603 {
604 name: "epoch change",
605 build: func(t *testing.T, catalogue *pickerCatalogue, _ *supervisorTestClock) string {
606 require.True(t, catalogue.Apply(ports.BrokerSnapshot{Epoch: 3, Revision: 1, Daemons: []ports.BrokerDaemonObservation{
607 pickerTestLocalObservation(now, pickerTestSession("alpha", 1, catalogue_Up)),
608 }}))
609 line, ok := pickerLineByLabel(catalogue.Lines(), "alpha")
610 require.True(t, ok)
611 key := line.Key
612 require.True(t, catalogue.Apply(ports.BrokerSnapshot{Epoch: 4, Revision: 1, Daemons: []ports.BrokerDaemonObservation{
613 pickerTestLocalObservation(now, pickerTestSession("beta", 2, catalogue_Up)),
614 }}))
615 return key
616 },
617 wantErr: pickerCatalogueUnknown,
618 },
619 {
620 name: "session lifecycle replaced",
621 build: func(t *testing.T, catalogue *pickerCatalogue, _ *supervisorTestClock) string {
622 require.True(t, catalogue.Apply(ports.BrokerSnapshot{Epoch: 3, Revision: 1, Daemons: []ports.BrokerDaemonObservation{
623 pickerTestLocalObservation(now, pickerTestSession("alpha", 1, catalogue_Up)),
624 }}))
625 line, ok := pickerLineByLabel(catalogue.Lines(), "alpha")
626 require.True(t, ok)
627 key := line.Key
628 require.True(t, catalogue.Apply(ports.BrokerSnapshot{Epoch: 3, Revision: 2, Daemons: []ports.BrokerDaemonObservation{
629 pickerTestLocalObservation(now, pickerTestSession("alpha", 5, catalogue_Up)),
630 }}))
631 return key
632 },
633 wantErr: pickerCatalogueReplaced,
634 },
635 {
636 name: "session gone",
637 build: func(t *testing.T, catalogue *pickerCatalogue, _ *supervisorTestClock) string {
638 require.True(t, catalogue.Apply(ports.BrokerSnapshot{Epoch: 3, Revision: 1, Daemons: []ports.BrokerDaemonObservation{
639 pickerTestLocalObservation(now, pickerTestSession("alpha", 1, catalogue_Up), pickerTestSession("beta", 2, catalogue_Up)),
640 }}))
641 line, ok := pickerLineByLabel(catalogue.Lines(), "alpha")
642 require.True(t, ok)
643 key := line.Key
644 require.True(t, catalogue.Apply(ports.BrokerSnapshot{Epoch: 3, Revision: 2, Daemons: []ports.BrokerDaemonObservation{
645 pickerTestLocalObservation(now, pickerTestSession("beta", 2, catalogue_Up)),
646 }}))
647 return key
648 },
649 wantErr: pickerCatalogueGone,
650 },
651 {
652 name: "unobserved host refuses as unavailable",
653 build: func(t *testing.T, catalogue *pickerCatalogue, _ *supervisorTestClock) string {
654 require.True(t, catalogue.Apply(ports.BrokerSnapshot{Epoch: 3, Revision: 1, Daemons: []ports.BrokerDaemonObservation{
655 pickerTestUnobservedObservation(now, pickerTestSession("alpha", 1, catalogue_Up)),
656 }}))
657 line, ok := pickerLineByLabel(catalogue.Lines(), "alpha")
658 require.True(t, ok)
659 return line.Key
660 },
661 wantLocal: true,
662 wantAdmission: ports.BrokerAdmissionExact,
663 },
664 }
665
666 for _, tt := range tests {
667 t.Run(tt.name, func(t *testing.T) {
668 catalogue, clock := pickerTestCatalogue(t)
669 key := tt.build(t, catalogue, clock)
670 request, err := catalogue.Resolve(key, base)
671 if tt.wantErr != 0 {
672 require.Error(t, err)
673 require.True(t, pickerCatalogueErrorIs(err, tt.wantErr), "want %v, got %v", tt.wantErr, err)
674 return
675 }
676 require.NoError(t, err)
677 require.NoError(t, request.Validate(), "a resolved request always validates")
678 require.Equal(t, ports.BrokerStreamAttachment, request.Purpose)
679 require.Equal(t, tt.wantAdmission, request.Admission)
680 require.Equal(t, tt.wantLocal, request.Local)
681 require.Equal(t, ports.BrokerEpoch(3), request.Epoch)
682 require.Equal(t, base.Connection, request.Connection)
683 require.Equal(t, base.Stream, request.Stream)
684 })
685 }
686}
687
688func TestPickerCatalogueRemoteSelectionCarriesExactIdentity(t *testing.T) {
689 clock := newSupervisorTestClock()
690 now := clock.Now()
691 catalogue, _ := pickerTestCatalogue(t)
692 registration := pickerTestRegistration("user@arch", 1)
693 lifecycle := pickerTestLifecycle(2)
694 require.True(t, catalogue.Apply(ports.BrokerSnapshot{Epoch: 3, Revision: 1, Daemons: []ports.BrokerDaemonObservation{
695 pickerTestRemoteObservation("user@arch", 1, 1, now, pickerTestSession("remote-a", 2, catalogue_Up)),
696 }}))
697 line, ok := pickerLineByLabel(catalogue.Lines(), "remote-a")
698 require.True(t, ok)
699
700 request, err := catalogue.Resolve(line.Key, pickerTestBase())
701 require.NoError(t, err)
702 require.Equal(t, "user@arch", request.Endpoint)
703 require.Equal(t, registration, request.Registration)
704 require.Equal(t, protocol.ExactSessionTarget{LifecycleID: lifecycle, SessionName: "remote-a"}, request.Target)
705 require.Equal(t, pickerTestPolicy(), request.Policy)
706}
707
708func TestPickerCatalogueUnknownKeyRefused(t *testing.T) {
709 catalogue, _ := pickerTestCatalogue(t)
710 _, err := catalogue.Resolve("missing", pickerTestBase())
711 require.Error(t, err)
712 require.True(t, pickerCatalogueErrorIs(err, pickerCatalogueUnknown))
713}
714
715// TestPickerCatalogueEpochFenceOnCapturedRef pins the selection revision: a
716// driver that captured a row's exact identity at one broker epoch may not
717// resolve it after the broker process was replaced, even when the endpoint and
718// session lifecycle happen to look the same.
719func TestPickerCatalogueEpochFenceOnCapturedRef(t *testing.T) {
720 clock := newSupervisorTestClock()
721 now := clock.Now()
722 catalogue, _ := pickerTestCatalogue(t)
723 require.True(t, catalogue.Apply(ports.BrokerSnapshot{Epoch: 3, Revision: 1, Daemons: []ports.BrokerDaemonObservation{
724 pickerTestLocalObservation(now, pickerTestSession("alpha", 1, catalogue_Up)),
725 }}))
726 line, ok := pickerLineByLabel(catalogue.Lines(), "alpha")
727 require.True(t, ok)
728 ref, ok := catalogue.Ref(line.Key)
729 require.True(t, ok)
730
731 // A new broker epoch resynchronizes the catalogue; the captured identity is
732 // fenced even though its key is recomputed identically.
733 require.True(t, catalogue.Apply(ports.BrokerSnapshot{Epoch: 4, Revision: 1, Daemons: []ports.BrokerDaemonObservation{
734 pickerTestLocalObservation(now, pickerTestSession("alpha", 1, catalogue_Up)),
735 }}))
736 _, err := catalogue.ResolveRef(ref, pickerTestBase())
737 require.Error(t, err)
738 require.True(t, pickerCatalogueErrorIs(err, pickerCatalogueEpochStale))
739}
740
741func TestPickerCatalogueRetiredRefFenceIsBounded(t *testing.T) {
742 catalogue, _ := pickerTestCatalogue(t)
743 base := pickerTestBase()
744 const inserted = pickerCatalogueMaxRetiredRefs + 8
745 keys := make([]string, 0, inserted)
746 for i := range inserted {
747 name := fmt.Sprintf("s-%d", i)
748 session := pickerTestSession(name, byte(i%200+1), catalogue_Up)
749 require.True(t, catalogue.Apply(ports.BrokerSnapshot{Epoch: 3, Revision: ports.BrokerRevision(i + 1), Daemons: []ports.BrokerDaemonObservation{
750 pickerTestLocalObservation(time.Unix(1000, 0), session),
751 }}))
752 line, ok := pickerLineByLabel(catalogue.Lines(), name)
753 require.True(t, ok)
754 keys = append(keys, line.Key)
755 }
756 // The fence holds the exact bound: only the newest retired keys survive.
757 require.Len(t, catalogue.retired, pickerCatalogueMaxRetiredRefs)
758 require.Len(t, catalogue.retiredOrder, pickerCatalogueMaxRetiredRefs)
759
760 // A retained retired key still classifies as gone: its lifecycle left the
761 // projection and no same-name replacement exists.
762 retainedKey := keys[len(keys)-2]
763 _, err := catalogue.Resolve(retainedKey, base)
764 require.Error(t, err)
765 require.True(t, pickerCatalogueErrorIs(err, pickerCatalogueGone), "a retained retired key must classify as gone: %v", err)
766
767 // The oldest key was evicted from the bounded fence and is now unknown.
768 evictedKey := keys[0]
769 _, err = catalogue.Resolve(evictedKey, base)
770 require.Error(t, err)
771 require.True(t, pickerCatalogueErrorIs(err, pickerCatalogueUnknown), "an evicted retired key must classify as unknown: %v", err)
772}
773
774// TestPickerCatalogueRetiredRefFenceConcurrentApply exercises the retired-ref
775// fence under -race: many goroutines fold distinct publications concurrently
776// and the fence may never exceed its bound.
777func TestPickerCatalogueRetiredRefFenceConcurrentApply(t *testing.T) {
778 catalogue, _ := pickerTestCatalogue(t)
779 const (
780 workers = 8
781 perWorker = 64
782 )
783 var wait sync.WaitGroup
784 for worker := range workers {
785 wait.Add(1)
786 go func(worker int) {
787 defer wait.Done()
788 for i := range perWorker {
789 var lifecycle domain.SessionLifecycleID
790 lifecycle[0] = byte(i + 1)
791 lifecycle[1] = byte(worker + 1)
792 session := pickerTestSessionWithLifecycle(fmt.Sprintf("w%d-%d", worker, i), lifecycle)
793 catalogue.Apply(ports.BrokerSnapshot{Epoch: 3, Revision: ports.BrokerRevision(worker*perWorker + i + 1), Daemons: []ports.BrokerDaemonObservation{
794 pickerTestLocalObservation(time.Unix(1000, 0), session),
795 }})
796 }
797 }(worker)
798 }
799 wait.Wait()
800 require.LessOrEqual(t, len(catalogue.retired), pickerCatalogueMaxRetiredRefs)
801 require.LessOrEqual(t, len(catalogue.retiredOrder), pickerCatalogueMaxRetiredRefs)
802}
803
804// TestPickerCatalogueAvailabilityClassifiedBeforeCompatibility pins the
805// authoritative ordering: availability first, then the reachable version
806// branch. An unobserved (ProtocolVersion 0) or unreachable daemon is never a
807// version mismatch.
808func TestPickerCatalogueAvailabilityClassifiedBeforeCompatibility(t *testing.T) {
809 now := time.Unix(1000, 0)
810 reachable := func(mutate func(*ports.BrokerDaemonObservation)) ports.BrokerDaemonObservation {
811 observation := pickerTestLocalObservation(now, pickerTestSession("alpha", 1, catalogue_Up))
812 if mutate != nil {
813 mutate(&observation)
814 }
815 return observation
816 }
817 unobserved := func(mutate func(*ports.BrokerDaemonObservation)) ports.BrokerDaemonObservation {
818 observation := pickerTestUnobservedObservation(now, pickerTestSession("alpha", 1, catalogue_Up))
819 if mutate != nil {
820 mutate(&observation)
821 }
822 return observation
823 }
824
825 tests := []struct {
826 name string
827 observation ports.BrokerDaemonObservation
828 fresh bool
829 wantStatus protocol.PickerLineStatus
830 wantReason string
831 }{
832 {name: "unobserved unknown is stale and refreshing", observation: unobserved(nil), fresh: false, wantStatus: protocol.PickerLineStatusStale, wantReason: domain.RemoteReasonRefreshing},
833 {name: "unreachable is down", observation: unobserved(func(o *ports.BrokerDaemonObservation) { o.Availability = domain.RemoteAvailabilityUnreachable }), wantStatus: protocol.PickerLineStatusDown, wantReason: domain.RemoteReasonHostUnreachable},
834 {name: "auth failed is down", observation: unobserved(func(o *ports.BrokerDaemonObservation) { o.Availability = domain.RemoteAvailabilityAuthFailed }), wantStatus: protocol.PickerLineStatusDown, wantReason: domain.RemoteReasonAuthFailure},
835 {name: "reachable without daemon is distinct and createable", observation: unobserved(func(o *ports.BrokerDaemonObservation) { o.Availability = domain.RemoteAvailabilityNoDaemon }), wantStatus: protocol.PickerLineStatusNoDaemon, wantReason: domain.RemoteReasonNoDaemon},
836 {name: "invalid response is error", observation: unobserved(func(o *ports.BrokerDaemonObservation) { o.Availability = domain.RemoteAvailabilityInvalidResponse }), wantStatus: protocol.PickerLineStatusError, wantReason: domain.RemoteReasonMalformed},
837 {name: "unreachable observed mismatch is down", observation: reachable(func(o *ports.BrokerDaemonObservation) {
838 o.Availability = domain.RemoteAvailabilityUnreachable
839 o.ProtocolVersion = protocol.Version + 1
840 }), wantStatus: protocol.PickerLineStatusDown, wantReason: domain.RemoteReasonHostUnreachable},
841 {name: "reachable mismatch is version", observation: reachable(func(o *ports.BrokerDaemonObservation) { o.ProtocolVersion = protocol.Version + 1 }), fresh: true, wantStatus: protocol.PickerLineStatusVersion, wantReason: domain.RemoteReasonVersionMismatch},
842 {name: "reachable compatible fresh is up", observation: reachable(nil), fresh: true, wantStatus: protocol.PickerLineStatusNone, wantReason: ""},
843 {name: "reachable compatible stale is stale", observation: reachable(nil), fresh: false, wantStatus: protocol.PickerLineStatusStale, wantReason: domain.RemoteReasonCatalogStale},
844 {name: "reachable unobserved is stale and refreshing", observation: unobserved(func(o *ports.BrokerDaemonObservation) { o.Availability = domain.RemoteAvailabilityReachable }), fresh: true, wantStatus: protocol.PickerLineStatusStale, wantReason: domain.RemoteReasonRefreshing},
845 {name: "legacy incompatible availability is version", observation: reachable(func(o *ports.BrokerDaemonObservation) { o.Availability = domain.RemoteAvailabilityIncompatible }), fresh: true, wantStatus: protocol.PickerLineStatusVersion, wantReason: domain.RemoteReasonVersionMismatch},
846 {name: "reachable checking is up and refreshing", observation: reachable(func(o *ports.BrokerDaemonObservation) { o.Checking = true }), fresh: true, wantStatus: protocol.PickerLineStatusNone, wantReason: domain.RemoteReasonRefreshing},
847 }
848 for _, tt := range tests {
849 t.Run(tt.name, func(t *testing.T) {
850 require.Equal(t, tt.wantStatus, pickerObservationStatus(tt.observation, tt.fresh))
851 require.Equal(t, tt.wantReason, pickerObservationReason(tt.observation, tt.fresh))
852 })
853 }
854}
855
856// TestPickerCatalogueResolutionAdmission pins the narrow refusal rules of the
857// shared resolver: only a known incompatibility (a version mismatch or an
858// availability classified as incompatible) is refused immediately. An
859// unobserved, unreachable, auth-failed, invalid-response, or stale daemon stays
860// explicitly attemptable, because only the destination revalidates the exact
861// identity and only it may start a stopped target under the resolved policy.
862func TestPickerCatalogueResolutionAdmission(t *testing.T) {
863 now := time.Unix(1000, 0)
864 tests := []struct {
865 name string
866 daemon ports.BrokerDaemonObservation
867 wantErr pickerCatalogueErrorCode
868 hostRow bool
869 create bool
870 exact bool
871 named bool
872 }{
873 {
874 name: "unobserved unknown stays attemptable",
875 daemon: pickerTestUnobservedObservation(now, pickerTestSession("alpha", 1, catalogue_Up)),
876 },
877 {
878 name: "unreachable observed mismatch refuses incompatible",
879 daemon: func() ports.BrokerDaemonObservation {
880 o := pickerTestLocalObservation(now, pickerTestSession("alpha", 1, catalogue_Up))
881 o.Availability = domain.RemoteAvailabilityUnreachable
882 o.ProtocolVersion = protocol.Version + 1
883 return o
884 }(),
885 wantErr: pickerCatalogueIncompatible,
886 },
887 {
888 name: "auth failed stays attemptable",
889 daemon: func() ports.BrokerDaemonObservation {
890 o := pickerTestUnobservedObservation(now, pickerTestSession("alpha", 1, catalogue_Up))
891 o.Availability = domain.RemoteAvailabilityAuthFailed
892 return o
893 }(),
894 },
895 {
896 name: "invalid response stays attemptable",
897 daemon: func() ports.BrokerDaemonObservation {
898 o := pickerTestUnobservedObservation(now, pickerTestSession("alpha", 1, catalogue_Up))
899 o.Availability = domain.RemoteAvailabilityInvalidResponse
900 return o
901 }(),
902 },
903 {
904 name: "reachable unobserved stays attemptable",
905 daemon: func() ports.BrokerDaemonObservation {
906 o := pickerTestUnobservedObservation(now, pickerTestSession("alpha", 1, catalogue_Up))
907 o.Availability = domain.RemoteAvailabilityReachable
908 return o
909 }(),
910 },
911 {
912 name: "no daemon allows create admission", hostRow: true, create: true,
913 daemon: func() ports.BrokerDaemonObservation {
914 o := pickerTestRemoteObservation("user@arch", 1, 1, now)
915 o.Identity = ""
916 o.Incarnation = ports.BrokerDaemonIncarnation{}
917 o.ProtocolVersion = 0
918 o.Availability = domain.RemoteAvailabilityNoDaemon
919 o.LastSuccess = time.Time{}
920 o.InventoryKnown = false
921 return o
922 }(),
923 },
924 {
925 name: "no daemon refuses exact absent session attach", hostRow: true, exact: true,
926 daemon: func() ports.BrokerDaemonObservation {
927 o := pickerTestRemoteObservation("user@arch", 1, 1, now)
928 o.Identity = ""
929 o.Incarnation = ports.BrokerDaemonIncarnation{}
930 o.ProtocolVersion = 0
931 o.Availability = domain.RemoteAvailabilityNoDaemon
932 o.LastSuccess = time.Time{}
933 o.InventoryKnown = false
934 return o
935 }(),
936 wantErr: pickerCatalogueUnavailable,
937 },
938 {
939 name: "no daemon lets the daemon decide a named attach", named: true,
940 daemon: func() ports.BrokerDaemonObservation {
941 o := pickerTestRemoteObservation("user@arch", 1, 1, now)
942 o.Identity = ""
943 o.Incarnation = ports.BrokerDaemonIncarnation{}
944 o.ProtocolVersion = 0
945 o.Availability = domain.RemoteAvailabilityNoDaemon
946 o.LastSuccess = time.Time{}
947 o.InventoryKnown = false
948 return o
949 }(),
950 },
951 {
952 name: "reachable mismatch refuses incompatible",
953 daemon: func() ports.BrokerDaemonObservation {
954 o := pickerTestLocalObservation(now, pickerTestSession("alpha", 1, catalogue_Up))
955 o.ProtocolVersion = protocol.Version + 1
956 return o
957 }(),
958 wantErr: pickerCatalogueIncompatible,
959 },
960 {
961 name: "legacy incompatible availability refuses incompatible",
962 daemon: func() ports.BrokerDaemonObservation {
963 o := pickerTestLocalObservation(now, pickerTestSession("alpha", 1, catalogue_Up))
964 o.Availability = domain.RemoteAvailabilityIncompatible
965 return o
966 }(),
967 wantErr: pickerCatalogueIncompatible,
968 },
969 }
970 for _, tt := range tests {
971 t.Run(tt.name, func(t *testing.T) {
972 catalogue, _ := pickerTestCatalogue(t)
973 require.True(t, catalogue.Apply(ports.BrokerSnapshot{Epoch: 3, Revision: 1, Daemons: []ports.BrokerDaemonObservation{tt.daemon}}))
974 lines := catalogue.Lines()
975 line, ok := pickerLineByLabel(lines, "alpha")
976 if tt.hostRow {
977 line, ok = pickerHostLine(lines)
978 }
979 if tt.create {
980 require.True(t, ok, "the no-daemon host row is visible")
981 ref, exists := catalogue.Ref(line.Key)
982 require.True(t, exists)
983 request, err := resolvePickerRequest(catalogue.snapshot.Epoch, pickerAuthorityInSnapshot(catalogue.snapshot, ref), ref, pickerTestBase())
984 require.NoError(t, err)
985 require.Equal(t, ports.BrokerAdmissionCreateEphemeral, request.Admission)
986 require.Equal(t, ports.BrokerDaemonStartIfNeeded, request.StartMode)
987 return
988 }
989 if tt.named {
990 ref := pickerSelectionRef{kind: pickerSelectionAttachNamed, epoch: catalogue.snapshot.Epoch, endpoint: tt.daemon.Endpoint,
991 registration: tt.daemon.Registration, name: "restored"}
992 request, err := resolvePickerRequest(catalogue.snapshot.Epoch, pickerAuthorityInSnapshot(catalogue.snapshot, ref), ref, pickerTestBase())
993 require.NoError(t, err)
994 require.Equal(t, ports.BrokerAdmissionAttachNamed, request.Admission)
995 require.Equal(t, ports.BrokerDaemonStartIfNeeded, request.StartMode)
996 return
997 }
998 if tt.exact {
999 ref := pickerSelectionRef{kind: pickerSelectionExact, epoch: catalogue.snapshot.Epoch, endpoint: tt.daemon.Endpoint,
1000 registration: tt.daemon.Registration, lifecycle: domain.SessionLifecycleID{1}, name: "absent"}
1001 _, err := resolvePickerRequest(catalogue.snapshot.Epoch, pickerAuthorityInSnapshot(catalogue.snapshot, ref), ref, pickerTestBase())
1002 require.Error(t, err)
1003 require.True(t, pickerCatalogueErrorIs(err, tt.wantErr), "want %v, got %v", tt.wantErr, err)
1004 return
1005 }
1006 require.True(t, ok)
1007 if tt.wantErr != 0 {
1008 _, err := catalogue.Resolve(line.Key, pickerTestBase())
1009 require.Error(t, err)
1010 require.True(t, pickerCatalogueErrorIs(err, tt.wantErr), "want %v, got %v", tt.wantErr, err)
1011 return
1012 }
1013 request, err := catalogue.Resolve(line.Key, pickerTestBase())
1014 require.NoError(t, err)
1015 require.NoError(t, request.Validate())
1016 require.Equal(t, ports.BrokerDaemonStartIfNeeded, request.StartMode, "an explicit attempt keeps the start authorization")
1017 })
1018 }
1019}
1020
1021// TestPickerCatalogueRejectsInvalidSnapshot pins the projection trust
1022// boundary: a snapshot that fails its own contract is refused and the previous
1023// projection is kept intact.
1024func TestPickerCatalogueRejectsInvalidSnapshot(t *testing.T) {
1025 catalogue, clock := pickerTestCatalogue(t)
1026 valid := pickerTestSnapshot(3, 1, clock.Now(), "alpha")
1027 require.True(t, catalogue.Apply(valid))
1028 validLines := catalogue.Lines()
1029 require.Equal(t, ports.BrokerRevision(1), catalogue.Revision())
1030
1031 invalid := pickerTestSnapshot(3, 2, clock.Now(), "beta")
1032 invalid.Daemons[0].Availability = 0 // out of range: availability is never zero
1033 require.False(t, catalogue.Apply(invalid), "an invalid snapshot is refused")
1034 require.Equal(t, ports.BrokerRevision(1), catalogue.Revision(), "the previous projection is kept")
1035 require.Equal(t, validLines, catalogue.Lines())
1036}
1037
1038// TestPickerCatalogueUnobservedDaemonProjects proves an unobserved daemon
1039// (Availability Unknown, ProtocolVersion 0, DisplayOrigin set) still projects a
1040// section and host row, classified as stale/refreshing rather than as a version
1041// mismatch.
1042func TestPickerCatalogueUnobservedDaemonProjects(t *testing.T) {
1043 catalogue, _ := pickerTestCatalogue(t)
1044 unobserved := pickerTestUnobservedObservation(time.Unix(1000, 0), pickerTestSession("alpha", 1, catalogue_Up))
1045 require.True(t, catalogue.Apply(ports.BrokerSnapshot{Epoch: 3, Revision: 1, Daemons: []ports.BrokerDaemonObservation{unobserved}}))
1046
1047 lines := catalogue.Lines()
1048 require.Equal(t, []string{"local"}, pickerSectionLabels(lines))
1049 require.Equal(t, []string{"alpha"}, pickerSessionLabels(lines))
1050 _, hasHost := pickerHostLine(lines)
1051 require.False(t, hasHost, "the session rows already carry the host state")
1052 session, ok := pickerLineByLabel(lines, "alpha")
1053 require.True(t, ok)
1054 require.True(t, session.Focusable)
1055 require.Equal(t, protocol.PickerCanNavigate|protocol.PickerCanKill, session.Actions, "an unobserved daemon may still be attempted explicitly")
1056 require.True(t, session.Dim)
1057 require.Equal(t, domain.RemoteReasonRefreshing, session.StatusDetail, "an unobserved daemon refreshes, it is not a version mismatch")
1058}
1059
1060// TestPickerCatalogueOriginLabelNeverShowsRawEndpoint proves display labels
1061// never fall back to a raw endpoint.
1062func TestPickerCatalogueOriginLabelNeverShowsRawEndpoint(t *testing.T) {
1063 raw := pickerTestRemoteObservation("user@arch", 1, 1, time.Unix(1000, 0))
1064 raw.DisplayOrigin = ""
1065 require.Equal(t, pickerCatalogueOriginFallback, pickerOriginLabel(raw))
1066 require.NotContains(t, pickerOriginLabel(raw), "user@arch")
1067
1068 require.Equal(t, "local", pickerOriginLabel(pickerTestUnobservedObservation(time.Unix(1000, 0))))
1069 require.Equal(t, "user@arch", pickerOriginLabel(pickerTestRemoteObservation("user@arch", 1, 1, time.Unix(1000, 0))))
1070}
1071
1072func TestPickerHostHealthNotice(t *testing.T) {
1073 now := time.Unix(1000, 0)
1074 host := func(availability domain.RemoteAvailability, kind domain.RemoteFailureKind, mismatch bool) pickerHostHealth {
1075 o := pickerTestRemoteObservation("user@arch", 1, 1, now)
1076 o.Availability = availability
1077 o.LastFailure = domain.RemoteFailure{Kind: kind}
1078 return pickerHostHealth{key: "host-key", observation: o, health: domain.RemoteHealth{Availability: availability, VersionMismatch: mismatch}}
1079 }
1080 const failed = "Remote check failed: user@arch — "
1081 tests := []struct {
1082 name string
1083 host pickerHostHealth
1084 event domain.RemoteHealthEvent
1085 severity domain.NoticeSeverity
1086 message string
1087 }{
1088 {name: "recovered", host: host(domain.RemoteAvailabilityReachable, 0, false), event: domain.RemoteHealthRecovered, severity: domain.NoticeInfo, message: "Remote host reconnected: user@arch"},
1089 {name: "timeout", host: host(domain.RemoteAvailabilityUnreachable, domain.RemoteFailureTimeout, false), event: domain.RemoteHealthFailed, severity: domain.NoticeError, message: failed + "SSH timed out"},
1090 {name: "authentication", host: host(domain.RemoteAvailabilityUnreachable, domain.RemoteFailureAuthentication, false), event: domain.RemoteHealthFailed, severity: domain.NoticeError, message: failed + "SSH authentication failed; verify non-interactive SSH access"},
1091 {name: "trust", host: host(domain.RemoteAvailabilityUnreachable, domain.RemoteFailureTrust, false), event: domain.RemoteHealthFailed, severity: domain.NoticeError, message: failed + "SSH host verification failed; verify the host key policy"},
1092 {name: "invalid response kind", host: host(domain.RemoteAvailabilityUnreachable, domain.RemoteFailureInvalidResponse, false), event: domain.RemoteHealthFailed, severity: domain.NoticeError, message: failed + "remote catalog response is invalid"},
1093 {name: "auth failed availability fallback", host: host(domain.RemoteAvailabilityAuthFailed, 0, false), event: domain.RemoteHealthFailed, severity: domain.NoticeError, message: failed + "SSH authentication failed; verify non-interactive SSH access"},
1094 {name: "invalid response availability fallback", host: host(domain.RemoteAvailabilityInvalidResponse, 0, false), event: domain.RemoteHealthFailed, severity: domain.NoticeError, message: failed + "remote catalog response is invalid"},
1095 {name: "unreachable default", host: host(domain.RemoteAvailabilityUnreachable, 0, false), event: domain.RemoteHealthFailed, severity: domain.NoticeError, message: failed + "SSH connection failed; verify SSH access"},
1096 {name: "incompatible", host: host(domain.RemoteAvailabilityIncompatible, 0, false), event: domain.RemoteHealthFailed, severity: domain.NoticeError, message: failed + "remote vev version is incompatible"},
1097 {name: "version mismatch", host: host(domain.RemoteAvailabilityReachable, 0, true), event: domain.RemoteHealthFailed, severity: domain.NoticeError, message: failed + "remote vev version is incompatible"},
1098 {name: "no daemon warns", host: host(domain.RemoteAvailabilityNoDaemon, 0, false), event: domain.RemoteHealthFailed, severity: domain.NoticeWarn, message: failed + "no vev daemon is running"},
1099 }
1100 for _, tt := range tests {
1101 t.Run(tt.name, func(t *testing.T) {
1102 n := pickerHostHealthNotice(tt.host, tt.event)
1103 require.Equal(t, domain.NoticeRemoteObservation, n.Code)
1104 require.Equal(t, "host-key", n.Scope, "failure and recovery share the host subject")
1105 require.Equal(t, tt.severity, n.Severity)
1106 require.Equal(t, tt.message, n.Message)
1107 })
1108 }
1109}