fix(client,app): wait for the remote inventory before attach-or-create
--remote HOST --session NAME resolved to a named creation whenever the first committed publication did not yet carry the host's inventory, so a freshly spawned broker collided with an existing remote session. The resolver now reports ErrInitialNavigationNotObserved until the host's inventory is observed; the supervisor keeps the intent armed, re-resolves it on each publication, and refuses it only after a bounded budget.
- commit
- 112960a995af0e07ebe056243664f9dc6b1dbfec
- author
- brice · · 2026-09-23 02:57 UTC
- parent
- 8360111
- signature
- GPG Verified
CD88F9E564E0102646FB6077A1A07C7C7C3E349D· arch-gpg Verified for the committer email
4 files · +165 -1
M internal/app/broker_client.go +7 -1
| @@ -446,7 +446,8 @@ func remoteNamedCreateResolver(endpoint, name string) client.InitialNavigationRe | |||
| 446 | 446 | ||
| 447 | 447 | // remoteAttachOrCreateResolver resolves one remote session name to an exact | |
| 448 | 448 | // attach when the endpoint's committed observation carries it, and to a named | |
| 449 | - | // creation fenced to the same registration otherwise. An endpoint the broker | |
| 449 | + | // creation fenced to the same registration once the observed inventory proves | |
| 450 | + | // it absent. Before the host's first inventory it asks to be resolved again. An endpoint the broker | |
| 450 | 451 | // does not carry is refused rather than dialed. | |
| 451 | 452 | func remoteAttachOrCreateResolver(endpoint, name string) client.InitialNavigationResolver { | |
| 452 | 453 | attach := remoteExactAttachResolver(endpoint, name) | |
| @@ -459,6 +460,11 @@ func remoteAttachOrCreateResolver(endpoint, name string) client.InitialNavigatio | |||
| 459 | 460 | if _, exists := brokerObservationExactTarget(observation, name); exists { | |
| 460 | 461 | return attach(snapshot) | |
| 461 | 462 | } | |
| 463 | + | if !observation.InventoryKnown { | |
| 464 | + | // Absence is only known once the host's inventory was observed; | |
| 465 | + | // creating before that would collide with a live remote name. | |
| 466 | + | return client.InitialNavigation{}, client.ErrInitialNavigationNotObserved | |
| 467 | + | } | |
| 462 | 468 | return create(snapshot) | |
| 463 | 469 | } | |
| 464 | 470 | } | |
| @@ -446,7 +446,8 @@ func remoteNamedCreateResolver(endpoint, name string) client.InitialNavigationRe | |||||
| 446 | 446 | ||||
| 447 | // remoteAttachOrCreateResolver resolves one remote session name to an exact | 447 | // remoteAttachOrCreateResolver resolves one remote session name to an exact | ||
| 448 | // attach when the endpoint's committed observation carries it, and to a named | 448 | // attach when the endpoint's committed observation carries it, and to a named | ||
| 449 | - | // creation fenced to the same registration otherwise. An endpoint the broker | 449 | + | // creation fenced to the same registration once the observed inventory proves |
| 450 | + | // it absent. Before the host's first inventory it asks to be resolved again. An endpoint the broker | |||
| 450 | // does not carry is refused rather than dialed. | 451 | // does not carry is refused rather than dialed. | ||
| 451 | func remoteAttachOrCreateResolver(endpoint, name string) client.InitialNavigationResolver { | 452 | func remoteAttachOrCreateResolver(endpoint, name string) client.InitialNavigationResolver { | ||
| 452 | attach := remoteExactAttachResolver(endpoint, name) | 453 | attach := remoteExactAttachResolver(endpoint, name) | ||
| @@ -459,6 +460,11 @@ func remoteAttachOrCreateResolver(endpoint, name string) client.InitialNavigatio | |||||
| 459 | if _, exists := brokerObservationExactTarget(observation, name); exists { | 460 | if _, exists := brokerObservationExactTarget(observation, name); exists { | ||
| 460 | return attach(snapshot) | 461 | return attach(snapshot) | ||
| 461 | } | 462 | } | ||
| 463 | + | if !observation.InventoryKnown { | |||
| 464 | + | // Absence is only known once the host's inventory was observed; | |||
| 465 | + | // creating before that would collide with a live remote name. | |||
| 466 | + | return client.InitialNavigation{}, client.ErrInitialNavigationNotObserved | |||
| 467 | + | } | |||
| 462 | return create(snapshot) | 468 | return create(snapshot) | ||
| 463 | } | 469 | } | ||
| 464 | } | 470 | } | ||
M internal/app/ui_driver_acceptance_test.go +15 -0
| @@ -212,6 +212,15 @@ func TestUIDriverNavigationReusesTheSharedTerminalTranslation(t *testing.T) { | |||
| 212 | 212 | require.Equal(t, "work", named.Name) | |
| 213 | 213 | require.True(t, named.Destination.Local) | |
| 214 | 214 | ||
| 215 | + | unobserved := ports.BrokerSnapshot{ | |
| 216 | + | Epoch: terminalCompositionEpoch, | |
| 217 | + | Revision: 1, | |
| 218 | + | Daemons: []ports.BrokerDaemonObservation{{ | |
| 219 | + | Endpoint: "user@example.com", | |
| 220 | + | DisplayOrigin: "user@example.com", | |
| 221 | + | Registration: terminalCompositionRegistration("user@example.com"), | |
| 222 | + | }}, | |
| 223 | + | } | |
| 215 | 224 | remoteSnapshot := func(sessions ...string) ports.BrokerSnapshot { | |
| 216 | 225 | return ports.BrokerSnapshot{ | |
| 217 | 226 | Epoch: terminalCompositionEpoch, | |
| @@ -231,7 +240,9 @@ func TestUIDriverNavigationReusesTheSharedTerminalTranslation(t *testing.T) { | |||
| 231 | 240 | snapshot ports.BrokerSnapshot | |
| 232 | 241 | wantKind client.InitialNavigationKind | |
| 233 | 242 | wantName string | |
| 243 | + | wantErr error | |
| 234 | 244 | }{ | |
| 245 | + | {name: "remote named before first inventory waits", options: uiDriverOptions{remote: "user@example.com", session: "work"}, snapshot: unobserved, wantErr: client.ErrInitialNavigationNotObserved}, | |
| 235 | 246 | {name: "remote ephemeral", options: uiDriverOptions{remote: "user@example.com"}, snapshot: remoteSnapshot("work"), wantKind: client.InitialNavigationCreateEphemeral}, | |
| 236 | 247 | {name: "remote named existing attaches exact lifecycle", options: uiDriverOptions{remote: "user@example.com", session: "work"}, snapshot: remoteSnapshot("other", "work"), wantKind: client.InitialNavigationAttachExact, wantName: "work"}, | |
| 237 | 248 | {name: "remote named missing creates", options: uiDriverOptions{remote: "user@example.com", session: "fresh"}, snapshot: remoteSnapshot("work"), wantKind: client.InitialNavigationCreateNamed, wantName: "fresh"}, | |
| @@ -245,6 +256,10 @@ func TestUIDriverNavigationReusesTheSharedTerminalTranslation(t *testing.T) { | |||
| 245 | 256 | require.NotNil(t, resolver, "a remote target resolves against the committed publication") | |
| 246 | 257 | ||
| 247 | 258 | resolved, resolveErr := resolver(tt.snapshot) | |
| 259 | + | if tt.wantErr != nil { | |
| 260 | + | require.ErrorIs(t, resolveErr, tt.wantErr) | |
| 261 | + | return | |
| 262 | + | } | |
| 248 | 263 | require.NoError(t, resolveErr) | |
| 249 | 264 | require.NoError(t, resolved.Validate()) | |
| 250 | 265 | require.False(t, resolved.Destination.Local) | |
| @@ -212,6 +212,15 @@ func TestUIDriverNavigationReusesTheSharedTerminalTranslation(t *testing.T) { | |||||
| 212 | require.Equal(t, "work", named.Name) | 212 | require.Equal(t, "work", named.Name) | ||
| 213 | require.True(t, named.Destination.Local) | 213 | require.True(t, named.Destination.Local) | ||
| 214 | 214 | ||||
| 215 | + | unobserved := ports.BrokerSnapshot{ | |||
| 216 | + | Epoch: terminalCompositionEpoch, | |||
| 217 | + | Revision: 1, | |||
| 218 | + | Daemons: []ports.BrokerDaemonObservation{{ | |||
| 219 | + | Endpoint: "user@example.com", | |||
| 220 | + | DisplayOrigin: "user@example.com", | |||
| 221 | + | Registration: terminalCompositionRegistration("user@example.com"), | |||
| 222 | + | }}, | |||
| 223 | + | } | |||
| 215 | remoteSnapshot := func(sessions ...string) ports.BrokerSnapshot { | 224 | remoteSnapshot := func(sessions ...string) ports.BrokerSnapshot { | ||
| 216 | return ports.BrokerSnapshot{ | 225 | return ports.BrokerSnapshot{ | ||
| 217 | Epoch: terminalCompositionEpoch, | 226 | Epoch: terminalCompositionEpoch, | ||
| @@ -231,7 +240,9 @@ func TestUIDriverNavigationReusesTheSharedTerminalTranslation(t *testing.T) { | |||||
| 231 | snapshot ports.BrokerSnapshot | 240 | snapshot ports.BrokerSnapshot | ||
| 232 | wantKind client.InitialNavigationKind | 241 | wantKind client.InitialNavigationKind | ||
| 233 | wantName string | 242 | wantName string | ||
| 243 | + | wantErr error | |||
| 234 | }{ | 244 | }{ | ||
| 245 | + | {name: "remote named before first inventory waits", options: uiDriverOptions{remote: "user@example.com", session: "work"}, snapshot: unobserved, wantErr: client.ErrInitialNavigationNotObserved}, | |||
| 235 | {name: "remote ephemeral", options: uiDriverOptions{remote: "user@example.com"}, snapshot: remoteSnapshot("work"), wantKind: client.InitialNavigationCreateEphemeral}, | 246 | {name: "remote ephemeral", options: uiDriverOptions{remote: "user@example.com"}, snapshot: remoteSnapshot("work"), wantKind: client.InitialNavigationCreateEphemeral}, | ||
| 236 | {name: "remote named existing attaches exact lifecycle", options: uiDriverOptions{remote: "user@example.com", session: "work"}, snapshot: remoteSnapshot("other", "work"), wantKind: client.InitialNavigationAttachExact, wantName: "work"}, | 247 | {name: "remote named existing attaches exact lifecycle", options: uiDriverOptions{remote: "user@example.com", session: "work"}, snapshot: remoteSnapshot("other", "work"), wantKind: client.InitialNavigationAttachExact, wantName: "work"}, | ||
| 237 | {name: "remote named missing creates", options: uiDriverOptions{remote: "user@example.com", session: "fresh"}, snapshot: remoteSnapshot("work"), wantKind: client.InitialNavigationCreateNamed, wantName: "fresh"}, | 248 | {name: "remote named missing creates", options: uiDriverOptions{remote: "user@example.com", session: "fresh"}, snapshot: remoteSnapshot("work"), wantKind: client.InitialNavigationCreateNamed, wantName: "fresh"}, | ||
| @@ -245,6 +256,10 @@ func TestUIDriverNavigationReusesTheSharedTerminalTranslation(t *testing.T) { | |||||
| 245 | require.NotNil(t, resolver, "a remote target resolves against the committed publication") | 256 | require.NotNil(t, resolver, "a remote target resolves against the committed publication") | ||
| 246 | 257 | ||||
| 247 | resolved, resolveErr := resolver(tt.snapshot) | 258 | resolved, resolveErr := resolver(tt.snapshot) | ||
| 259 | + | if tt.wantErr != nil { | |||
| 260 | + | require.ErrorIs(t, resolveErr, tt.wantErr) | |||
| 261 | + | return | |||
| 262 | + | } | |||
| 248 | require.NoError(t, resolveErr) | 263 | require.NoError(t, resolveErr) | ||
| 249 | require.NoError(t, resolved.Validate()) | 264 | require.NoError(t, resolved.Validate()) | ||
| 250 | require.False(t, resolved.Destination.Local) | 265 | require.False(t, resolved.Destination.Local) | ||
M internal/usecase/client/initial_navigation_test.go +84 -0
| @@ -500,3 +500,87 @@ func TestInitialNavigationResolverFailureIsSelectionUnavailable(t *testing.T) { | |||
| 500 | 500 | cancel() | |
| 501 | 501 | require.ErrorIs(t, <-done, context.Canceled) | |
| 502 | 502 | } | |
| 503 | + | ||
| 504 | + | // TestInitialNavigationWaitsForTargetObservation pins that a resolver whose | |
| 505 | + | // target daemon is not observed yet keeps the intent armed: it resolves again | |
| 506 | + | // on the next committed publication, and only the bounded observation budget | |
| 507 | + | // turns the wait into a selection-unavailable refusal. | |
| 508 | + | func TestInitialNavigationWaitsForTargetObservation(t *testing.T) { | |
| 509 | + | tests := []struct { | |
| 510 | + | name string | |
| 511 | + | // observedAt is the first revision the resolver can decide on; zero | |
| 512 | + | // means the target is never observed. | |
| 513 | + | observedAt ports.BrokerRevision | |
| 514 | + | wantOpen bool | |
| 515 | + | }{ | |
| 516 | + | {name: "a later publication resolves the armed intent", observedAt: 2, wantOpen: true}, | |
| 517 | + | {name: "the observation budget refuses once", observedAt: 0}, | |
| 518 | + | } | |
| 519 | + | for _, tt := range tests { | |
| 520 | + | t.Run(tt.name, func(t *testing.T) { | |
| 521 | + | ctx, cancel := context.WithCancel(context.Background()) | |
| 522 | + | defer cancel() | |
| 523 | + | clock := newSupervisorTestClock() | |
| 524 | + | reader := newAttachTestReader() | |
| 525 | + | terminal := &attachTestTerminal{in: reader} | |
| 526 | + | picker := newAttachTestPicker() | |
| 527 | + | service := newSupervisorTestService(ports.BrokerConnectionID{1}) | |
| 528 | + | service.publishSnapshot(localDaemonSnapshot(3, 1)) | |
| 529 | + | var opened atomic.Int64 | |
| 530 | + | service.setOpenStream(func(context.Context, ports.BrokerOpenStreamRequest) (ports.BrokerLogicalConnection, error) { | |
| 531 | + | opened.Add(1) | |
| 532 | + | return newSessionTestStream(), nil | |
| 533 | + | }) | |
| 534 | + | notices := make(chan LifecycleNotice, 4) | |
| 535 | + | var calls atomic.Int64 | |
| 536 | + | sup := mustSupervisor(t, SupervisorConfig{ | |
| 537 | + | Connector: newSupervisorTestConnector(func(context.Context, int) (ports.BrokerService, error) { return service, nil }), | |
| 538 | + | Terminal: terminal, Clock: clock, Jitter: func() float64 { return 0 }, Picker: picker, | |
| 539 | + | NotifyLifecycle: func(notice LifecycleNotice) { notices <- notice }, | |
| 540 | + | ResolveInitialNavigation: func(snapshot ports.BrokerSnapshot) (InitialNavigation, error) { | |
| 541 | + | calls.Add(1) | |
| 542 | + | if tt.observedAt == 0 || snapshot.Revision < tt.observedAt { | |
| 543 | + | return InitialNavigation{}, ErrInitialNavigationNotObserved | |
| 544 | + | } | |
| 545 | + | return InitialNavigation{Kind: InitialNavigationCreateEphemeral, Destination: ports.BrokerEndpointFence{Local: true}}, nil | |
| 546 | + | }, | |
| 547 | + | }) | |
| 548 | + | done := make(chan error, 1) | |
| 549 | + | go func() { done <- sup.Run(ctx) }() | |
| 550 | + | ||
| 551 | + | require.Eventually(t, func() bool { return calls.Load() == 1 }, 5*time.Second, time.Millisecond) | |
| 552 | + | var budget *supervisorTestTimer | |
| 553 | + | require.Eventually(t, func() bool { | |
| 554 | + | clock.mu.Lock() | |
| 555 | + | defer clock.mu.Unlock() | |
| 556 | + | for _, timer := range clock.timers { | |
| 557 | + | if timer.delay == initialNavigationObservationBudget { | |
| 558 | + | budget = timer | |
| 559 | + | return true | |
| 560 | + | } | |
| 561 | + | } | |
| 562 | + | return false | |
| 563 | + | }, 5*time.Second, time.Millisecond, "the wait is bounded by the observation budget") | |
| 564 | + | require.Zero(t, opened.Load(), "an unobserved target never dials") | |
| 565 | + | ||
| 566 | + | if tt.wantOpen { | |
| 567 | + | service.publishSnapshot(localDaemonSnapshot(3, 2)) | |
| 568 | + | require.Eventually(t, func() bool { return opened.Load() == 1 }, 5*time.Second, time.Millisecond) | |
| 569 | + | require.Equal(t, int64(2), calls.Load(), "the armed intent resolves once per publication") | |
| 570 | + | } else { | |
| 571 | + | budget.fire() | |
| 572 | + | select { | |
| 573 | + | case notice := <-notices: | |
| 574 | + | require.Equal(t, LifecycleNoticeSelectionUnavailable, notice.Kind) | |
| 575 | + | case <-time.After(5 * time.Second): | |
| 576 | + | t.Fatal("the budget never refused the unobserved target") | |
| 577 | + | } | |
| 578 | + | require.Zero(t, opened.Load()) | |
| 579 | + | service.publishSnapshot(localDaemonSnapshot(3, 2)) | |
| 580 | + | require.Never(t, func() bool { return calls.Load() > 1 }, 100*time.Millisecond, 5*time.Millisecond, "a refused intent is never re-armed") | |
| 581 | + | } | |
| 582 | + | cancel() | |
| 583 | + | require.ErrorIs(t, <-done, context.Canceled) | |
| 584 | + | }) | |
| 585 | + | } | |
| 586 | + | } | |
| @@ -500,3 +500,87 @@ func TestInitialNavigationResolverFailureIsSelectionUnavailable(t *testing.T) { | |||||
| 500 | cancel() | 500 | cancel() | ||
| 501 | require.ErrorIs(t, <-done, context.Canceled) | 501 | require.ErrorIs(t, <-done, context.Canceled) | ||
| 502 | } | 502 | } | ||
| 503 | + | ||||
| 504 | + | // TestInitialNavigationWaitsForTargetObservation pins that a resolver whose | |||
| 505 | + | // target daemon is not observed yet keeps the intent armed: it resolves again | |||
| 506 | + | // on the next committed publication, and only the bounded observation budget | |||
| 507 | + | // turns the wait into a selection-unavailable refusal. | |||
| 508 | + | func TestInitialNavigationWaitsForTargetObservation(t *testing.T) { | |||
| 509 | + | tests := []struct { | |||
| 510 | + | name string | |||
| 511 | + | // observedAt is the first revision the resolver can decide on; zero | |||
| 512 | + | // means the target is never observed. | |||
| 513 | + | observedAt ports.BrokerRevision | |||
| 514 | + | wantOpen bool | |||
| 515 | + | }{ | |||
| 516 | + | {name: "a later publication resolves the armed intent", observedAt: 2, wantOpen: true}, | |||
| 517 | + | {name: "the observation budget refuses once", observedAt: 0}, | |||
| 518 | + | } | |||
| 519 | + | for _, tt := range tests { | |||
| 520 | + | t.Run(tt.name, func(t *testing.T) { | |||
| 521 | + | ctx, cancel := context.WithCancel(context.Background()) | |||
| 522 | + | defer cancel() | |||
| 523 | + | clock := newSupervisorTestClock() | |||
| 524 | + | reader := newAttachTestReader() | |||
| 525 | + | terminal := &attachTestTerminal{in: reader} | |||
| 526 | + | picker := newAttachTestPicker() | |||
| 527 | + | service := newSupervisorTestService(ports.BrokerConnectionID{1}) | |||
| 528 | + | service.publishSnapshot(localDaemonSnapshot(3, 1)) | |||
| 529 | + | var opened atomic.Int64 | |||
| 530 | + | service.setOpenStream(func(context.Context, ports.BrokerOpenStreamRequest) (ports.BrokerLogicalConnection, error) { | |||
| 531 | + | opened.Add(1) | |||
| 532 | + | return newSessionTestStream(), nil | |||
| 533 | + | }) | |||
| 534 | + | notices := make(chan LifecycleNotice, 4) | |||
| 535 | + | var calls atomic.Int64 | |||
| 536 | + | sup := mustSupervisor(t, SupervisorConfig{ | |||
| 537 | + | Connector: newSupervisorTestConnector(func(context.Context, int) (ports.BrokerService, error) { return service, nil }), | |||
| 538 | + | Terminal: terminal, Clock: clock, Jitter: func() float64 { return 0 }, Picker: picker, | |||
| 539 | + | NotifyLifecycle: func(notice LifecycleNotice) { notices <- notice }, | |||
| 540 | + | ResolveInitialNavigation: func(snapshot ports.BrokerSnapshot) (InitialNavigation, error) { | |||
| 541 | + | calls.Add(1) | |||
| 542 | + | if tt.observedAt == 0 || snapshot.Revision < tt.observedAt { | |||
| 543 | + | return InitialNavigation{}, ErrInitialNavigationNotObserved | |||
| 544 | + | } | |||
| 545 | + | return InitialNavigation{Kind: InitialNavigationCreateEphemeral, Destination: ports.BrokerEndpointFence{Local: true}}, nil | |||
| 546 | + | }, | |||
| 547 | + | }) | |||
| 548 | + | done := make(chan error, 1) | |||
| 549 | + | go func() { done <- sup.Run(ctx) }() | |||
| 550 | + | ||||
| 551 | + | require.Eventually(t, func() bool { return calls.Load() == 1 }, 5*time.Second, time.Millisecond) | |||
| 552 | + | var budget *supervisorTestTimer | |||
| 553 | + | require.Eventually(t, func() bool { | |||
| 554 | + | clock.mu.Lock() | |||
| 555 | + | defer clock.mu.Unlock() | |||
| 556 | + | for _, timer := range clock.timers { | |||
| 557 | + | if timer.delay == initialNavigationObservationBudget { | |||
| 558 | + | budget = timer | |||
| 559 | + | return true | |||
| 560 | + | } | |||
| 561 | + | } | |||
| 562 | + | return false | |||
| 563 | + | }, 5*time.Second, time.Millisecond, "the wait is bounded by the observation budget") | |||
| 564 | + | require.Zero(t, opened.Load(), "an unobserved target never dials") | |||
| 565 | + | ||||
| 566 | + | if tt.wantOpen { | |||
| 567 | + | service.publishSnapshot(localDaemonSnapshot(3, 2)) | |||
| 568 | + | require.Eventually(t, func() bool { return opened.Load() == 1 }, 5*time.Second, time.Millisecond) | |||
| 569 | + | require.Equal(t, int64(2), calls.Load(), "the armed intent resolves once per publication") | |||
| 570 | + | } else { | |||
| 571 | + | budget.fire() | |||
| 572 | + | select { | |||
| 573 | + | case notice := <-notices: | |||
| 574 | + | require.Equal(t, LifecycleNoticeSelectionUnavailable, notice.Kind) | |||
| 575 | + | case <-time.After(5 * time.Second): | |||
| 576 | + | t.Fatal("the budget never refused the unobserved target") | |||
| 577 | + | } | |||
| 578 | + | require.Zero(t, opened.Load()) | |||
| 579 | + | service.publishSnapshot(localDaemonSnapshot(3, 2)) | |||
| 580 | + | require.Never(t, func() bool { return calls.Load() > 1 }, 100*time.Millisecond, 5*time.Millisecond, "a refused intent is never re-armed") | |||
| 581 | + | } | |||
| 582 | + | cancel() | |||
| 583 | + | require.ErrorIs(t, <-done, context.Canceled) | |||
| 584 | + | }) | |||
| 585 | + | } | |||
| 586 | + | } | |||
M internal/usecase/client/supervisor_attach.go +59 -0
| @@ -364,6 +364,14 @@ func (s *Supervisor) takeInitialNavigation(snapshot ports.BrokerSnapshot) (Initi | |||
| 364 | 364 | ||
| 365 | 365 | if resolver != nil { | |
| 366 | 366 | resolved, err := resolver(snapshot.Clone()) | |
| 367 | + | if errors.Is(err, ErrInitialNavigationNotObserved) { | |
| 368 | + | // Not a refusal: the target's daemon has not been observed yet. | |
| 369 | + | // The intent stays armed for a later publication. | |
| 370 | + | s.mu.Lock() | |
| 371 | + | s.navigationConsumed = false | |
| 372 | + | s.mu.Unlock() | |
| 373 | + | return InitialNavigation{}, false, err | |
| 374 | + | } | |
| 367 | 375 | if err != nil { | |
| 368 | 376 | return InitialNavigation{}, true, err | |
| 369 | 377 | } | |
| @@ -378,6 +386,49 @@ func (s *Supervisor) takeInitialNavigation(snapshot ports.BrokerSnapshot) (Initi | |||
| 378 | 386 | return navigation, true, nil | |
| 379 | 387 | } | |
| 380 | 388 | ||
| 389 | + | // ErrInitialNavigationNotObserved is returned by an InitialNavigationResolver | |
| 390 | + | // whose decision depends on a daemon inventory the committed publication does | |
| 391 | + | // not carry yet (for example attach-or-create before the host's first | |
| 392 | + | // observation). The supervisor keeps the intent armed and resolves it again on | |
| 393 | + | // later publications, bounded by initialNavigationObservationBudget. | |
| 394 | + | var ErrInitialNavigationNotObserved = errors.New("vev: initial navigation target is not observed yet") | |
| 395 | + | ||
| 396 | + | // initialNavigationObservationBudget bounds how long an initial navigation | |
| 397 | + | // waits for its target's first observation; it stays inside the 15-second | |
| 398 | + | // handshake budget. | |
| 399 | + | const initialNavigationObservationBudget = 10 * time.Second | |
| 400 | + | ||
| 401 | + | // awaitInitialNavigationObservation re-resolves the armed initial navigation | |
| 402 | + | // on each committed publication until the resolver can decide, the budget | |
| 403 | + | // expires (a bounded refusal), or the run ends. | |
| 404 | + | func (s *Supervisor) awaitInitialNavigationObservation(ctx context.Context, input *terminalInputLifetime, service ports.BrokerService) (InitialNavigation, bool, error, bool, error) { | |
| 405 | + | timer := s.cfg.Clock.NewTimer(initialNavigationObservationBudget) | |
| 406 | + | defer timer.Stop() | |
| 407 | + | for { | |
| 408 | + | select { | |
| 409 | + | case <-s.brokerChanged(): | |
| 410 | + | navigation, consumed, err := s.takeInitialNavigation(service.Snapshot()) | |
| 411 | + | if errors.Is(err, ErrInitialNavigationNotObserved) { | |
| 412 | + | continue | |
| 413 | + | } | |
| 414 | + | return navigation, consumed, err, false, nil | |
| 415 | + | case <-timer.C(): | |
| 416 | + | s.mu.Lock() | |
| 417 | + | s.navigationConsumed = true | |
| 418 | + | s.mu.Unlock() | |
| 419 | + | return InitialNavigation{}, true, ErrInitialNavigationNotObserved, false, nil | |
| 420 | + | case <-service.Done(): | |
| 421 | + | // The ready loop observes the loss; the intent stays armed for the | |
| 422 | + | // next adopted connection. | |
| 423 | + | return InitialNavigation{}, false, nil, false, nil | |
| 424 | + | case <-ctx.Done(): | |
| 425 | + | return InitialNavigation{}, false, nil, true, ctx.Err() | |
| 426 | + | case err := <-input.EOF(): | |
| 427 | + | return InitialNavigation{}, false, nil, true, terminalReadCause(err) | |
| 428 | + | } | |
| 429 | + | } | |
| 430 | + | } | |
| 431 | + | ||
| 381 | 432 | // runCommittedAttachment admits one committed selection end to end: resolve, | |
| 382 | 433 | // open the exact stream under the one absolute deadline, run the real | |
| 383 | 434 | // attachment worker through the supervisor's foreground grant, and return to | |
| @@ -390,6 +441,14 @@ func (s *Supervisor) runInitialNavigation(ctx context.Context, input *terminalIn | |||
| 390 | 441 | // read. | |
| 391 | 442 | snapshot := service.Snapshot() | |
| 392 | 443 | navigation, consumed, err := s.takeInitialNavigation(snapshot) | |
| 444 | + | if errors.Is(err, ErrInitialNavigationNotObserved) { | |
| 445 | + | var terminated bool | |
| 446 | + | var termErr error | |
| 447 | + | navigation, consumed, err, terminated, termErr = s.awaitInitialNavigationObservation(ctx, input, service) | |
| 448 | + | if terminated { | |
| 449 | + | return true, termErr | |
| 450 | + | } | |
| 451 | + | } | |
| 393 | 452 | if !consumed { | |
| 394 | 453 | return false, nil | |
| 395 | 454 | } | |
| @@ -364,6 +364,14 @@ func (s *Supervisor) takeInitialNavigation(snapshot ports.BrokerSnapshot) (Initi | |||||
| 364 | 364 | ||||
| 365 | if resolver != nil { | 365 | if resolver != nil { | ||
| 366 | resolved, err := resolver(snapshot.Clone()) | 366 | resolved, err := resolver(snapshot.Clone()) | ||
| 367 | + | if errors.Is(err, ErrInitialNavigationNotObserved) { | |||
| 368 | + | // Not a refusal: the target's daemon has not been observed yet. | |||
| 369 | + | // The intent stays armed for a later publication. | |||
| 370 | + | s.mu.Lock() | |||
| 371 | + | s.navigationConsumed = false | |||
| 372 | + | s.mu.Unlock() | |||
| 373 | + | return InitialNavigation{}, false, err | |||
| 374 | + | } | |||
| 367 | if err != nil { | 375 | if err != nil { | ||
| 368 | return InitialNavigation{}, true, err | 376 | return InitialNavigation{}, true, err | ||
| 369 | } | 377 | } | ||
| @@ -378,6 +386,49 @@ func (s *Supervisor) takeInitialNavigation(snapshot ports.BrokerSnapshot) (Initi | |||||
| 378 | return navigation, true, nil | 386 | return navigation, true, nil | ||
| 379 | } | 387 | } | ||
| 380 | 388 | ||||
| 389 | + | // ErrInitialNavigationNotObserved is returned by an InitialNavigationResolver | |||
| 390 | + | // whose decision depends on a daemon inventory the committed publication does | |||
| 391 | + | // not carry yet (for example attach-or-create before the host's first | |||
| 392 | + | // observation). The supervisor keeps the intent armed and resolves it again on | |||
| 393 | + | // later publications, bounded by initialNavigationObservationBudget. | |||
| 394 | + | var ErrInitialNavigationNotObserved = errors.New("vev: initial navigation target is not observed yet") | |||
| 395 | + | ||||
| 396 | + | // initialNavigationObservationBudget bounds how long an initial navigation | |||
| 397 | + | // waits for its target's first observation; it stays inside the 15-second | |||
| 398 | + | // handshake budget. | |||
| 399 | + | const initialNavigationObservationBudget = 10 * time.Second | |||
| 400 | + | ||||
| 401 | + | // awaitInitialNavigationObservation re-resolves the armed initial navigation | |||
| 402 | + | // on each committed publication until the resolver can decide, the budget | |||
| 403 | + | // expires (a bounded refusal), or the run ends. | |||
| 404 | + | func (s *Supervisor) awaitInitialNavigationObservation(ctx context.Context, input *terminalInputLifetime, service ports.BrokerService) (InitialNavigation, bool, error, bool, error) { | |||
| 405 | + | timer := s.cfg.Clock.NewTimer(initialNavigationObservationBudget) | |||
| 406 | + | defer timer.Stop() | |||
| 407 | + | for { | |||
| 408 | + | select { | |||
| 409 | + | case <-s.brokerChanged(): | |||
| 410 | + | navigation, consumed, err := s.takeInitialNavigation(service.Snapshot()) | |||
| 411 | + | if errors.Is(err, ErrInitialNavigationNotObserved) { | |||
| 412 | + | continue | |||
| 413 | + | } | |||
| 414 | + | return navigation, consumed, err, false, nil | |||
| 415 | + | case <-timer.C(): | |||
| 416 | + | s.mu.Lock() | |||
| 417 | + | s.navigationConsumed = true | |||
| 418 | + | s.mu.Unlock() | |||
| 419 | + | return InitialNavigation{}, true, ErrInitialNavigationNotObserved, false, nil | |||
| 420 | + | case <-service.Done(): | |||
| 421 | + | // The ready loop observes the loss; the intent stays armed for the | |||
| 422 | + | // next adopted connection. | |||
| 423 | + | return InitialNavigation{}, false, nil, false, nil | |||
| 424 | + | case <-ctx.Done(): | |||
| 425 | + | return InitialNavigation{}, false, nil, true, ctx.Err() | |||
| 426 | + | case err := <-input.EOF(): | |||
| 427 | + | return InitialNavigation{}, false, nil, true, terminalReadCause(err) | |||
| 428 | + | } | |||
| 429 | + | } | |||
| 430 | + | } | |||
| 431 | + | ||||
| 381 | // runCommittedAttachment admits one committed selection end to end: resolve, | 432 | // runCommittedAttachment admits one committed selection end to end: resolve, | ||
| 382 | // open the exact stream under the one absolute deadline, run the real | 433 | // open the exact stream under the one absolute deadline, run the real | ||
| 383 | // attachment worker through the supervisor's foreground grant, and return to | 434 | // attachment worker through the supervisor's foreground grant, and return to | ||
| @@ -390,6 +441,14 @@ func (s *Supervisor) runInitialNavigation(ctx context.Context, input *terminalIn | |||||
| 390 | // read. | 441 | // read. | ||
| 391 | snapshot := service.Snapshot() | 442 | snapshot := service.Snapshot() | ||
| 392 | navigation, consumed, err := s.takeInitialNavigation(snapshot) | 443 | navigation, consumed, err := s.takeInitialNavigation(snapshot) | ||
| 444 | + | if errors.Is(err, ErrInitialNavigationNotObserved) { | |||
| 445 | + | var terminated bool | |||
| 446 | + | var termErr error | |||
| 447 | + | navigation, consumed, err, terminated, termErr = s.awaitInitialNavigationObservation(ctx, input, service) | |||
| 448 | + | if terminated { | |||
| 449 | + | return true, termErr | |||
| 450 | + | } | |||
| 451 | + | } | |||
| 393 | if !consumed { | 452 | if !consumed { | ||
| 394 | return false, nil | 453 | return false, nil | ||
| 395 | } | 454 | } | ||