9b767e7 fix(client,app): fall back to creation when the remote stays unobserved · vev

M internal/app/broker_client.go +9 -2


@@ -455,7 +455,8 @@ func remoteNamedCreateResolver(endpoint, name string) client.InitialNavigationRe
455	455		// attach when the endpoint's committed observation carries it, and to a named
456	456		// creation fenced to the same registration once an inventory observed at or
457	457		// after requestedAt proves it absent. Before such an observation it asks the
458		-	// supervisor to request one and resolve again. An endpoint the broker does not
	458	+	// supervisor to request one and resolve again, with the creation as the
	459	+	// fallback should the host stay unobserved. An endpoint the broker does not
459	460		// carry is refused rather than dialed.
460	461		func remoteAttachOrCreateResolver(endpoint, name string, requestedAt time.Time) client.InitialNavigationResolver {
461	462		attach := remoteExactAttachResolver(endpoint, name)

@@ -469,7 +470,13 @@ func remoteAttachOrCreateResolver(endpoint, name string, requestedAt time.Time)
469	470		return attach(snapshot)
470	471		}
471	472		if !observation.InventoryKnown \|\| observation.LastSuccess.Before(requestedAt) {
472		-	return client.InitialNavigation{}, client.InitialNavigationNotObserved{Endpoint: endpoint}
	473	+	// Should the host stay unobserved, creating is still the only
	474	+	// resolvable intent; the remote daemon refuses a live name.
	475	+	fallback, err := create(snapshot)
	476	+	if err != nil {
	477	+	fallback = client.InitialNavigation{}
	478	+	}
	479	+	return client.InitialNavigation{}, client.InitialNavigationNotObserved{Endpoint: endpoint, Fallback: fallback}
473	480		}
474	481		return create(snapshot)
475	482		}

 // attach when the endpoint's committed observation carries it, and to a named
 // creation fenced to the same registration once an inventory observed at or
 // after requestedAt proves it absent. Before such an observation it asks the
-// supervisor to request one and resolve again. An endpoint the broker does not
+// supervisor to request one and resolve again, with the creation as the
+// fallback should the host stay unobserved. An endpoint the broker does not
 // carry is refused rather than dialed.
 func remoteAttachOrCreateResolver(endpoint, name string, requestedAt time.Time) client.InitialNavigationResolver {
 	attach := remoteExactAttachResolver(endpoint, name)
 			return attach(snapshot)
 		}
 		if !observation.InventoryKnown || observation.LastSuccess.Before(requestedAt) {
-			return client.InitialNavigation{}, client.InitialNavigationNotObserved{Endpoint: endpoint}
+			// Should the host stay unobserved, creating is still the only
+			// resolvable intent; the remote daemon refuses a live name.
+			fallback, err := create(snapshot)
+			if err != nil {
+				fallback = client.InitialNavigation{}
+			}
+			return client.InitialNavigation{}, client.InitialNavigationNotObserved{Endpoint: endpoint, Fallback: fallback}
 		}
 		return create(snapshot)
 	}

M internal/app/ui_driver_acceptance_test.go +3 -0


@@ -268,6 +268,9 @@ func TestUIDriverNavigationReusesTheSharedTerminalTranslation(t *testing.T) {
268	268		var pending client.InitialNavigationNotObserved
269	269		require.ErrorAs(t, resolveErr, &pending)
270	270		require.Equal(t, "user@example.com", pending.Endpoint, "the wait names the endpoint to reconcile")
	271	+	require.Equal(t, client.InitialNavigationCreateNamed, pending.Fallback.Kind, "an unobservable host still falls back to creation")
	272	+	require.Equal(t, tt.options.session, pending.Fallback.Name)
	273	+	require.NoError(t, pending.Fallback.Validate())
271	274		return
272	275		}
273	276		require.NoError(t, resolveErr)


@@ -268,6 +268,9 @@ func TestUIDriverNavigationReusesTheSharedTerminalTranslation(t *testing.T) {
268	var pending client.InitialNavigationNotObserved	268		var pending client.InitialNavigationNotObserved
269	require.ErrorAs(t, resolveErr, &pending)	269		require.ErrorAs(t, resolveErr, &pending)
270	require.Equal(t, "user@example.com", pending.Endpoint, "the wait names the endpoint to reconcile")	270		require.Equal(t, "user@example.com", pending.Endpoint, "the wait names the endpoint to reconcile")
		271	+	require.Equal(t, client.InitialNavigationCreateNamed, pending.Fallback.Kind, "an unobservable host still falls back to creation")
		272	+	require.Equal(t, tt.options.session, pending.Fallback.Name)
		273	+	require.NoError(t, pending.Fallback.Validate())
271	return	274		return
272	}	275		}
273	require.NoError(t, resolveErr)	276		require.NoError(t, resolveErr)

M internal/usecase/client/initial_navigation_test.go +15 -3


@@ -512,10 +512,13 @@ func TestInitialNavigationWaitsForTargetObservation(t *testing.T) {
512	512		// observedAt is the first revision the resolver can decide on; zero
513	513		// means the target is never observed.
514	514		observedAt ports.BrokerRevision
515		-	wantOpen bool
	515	+	// fallback is offered with the pending resolution.
	516	+	fallback bool
	517	+	wantOpen bool
516	518		}{
517	519		{name: "a later publication resolves the armed intent", observedAt: 2, wantOpen: true},
518	520		{name: "the observation budget refuses once", observedAt: 0},
	521	+	{name: "the observation budget takes the resolver's fallback", observedAt: 0, fallback: true, wantOpen: true},
519	522		}
520	523		for _, tt := range tests {
521	524		t.Run(tt.name, func(t *testing.T) {

@@ -541,7 +544,11 @@ func TestInitialNavigationWaitsForTargetObservation(t *testing.T) {
541	544		ResolveInitialNavigation: func(snapshot ports.BrokerSnapshot) (InitialNavigation, error) {
542	545		calls.Add(1)
543	546		if tt.observedAt == 0 \|\| snapshot.Revision < tt.observedAt {
544		-	return InitialNavigation{}, InitialNavigationNotObserved{Endpoint: "user@host"}
	547	+	pending := InitialNavigationNotObserved{Endpoint: "user@host"}
	548	+	if tt.fallback {
	549	+	pending.Fallback = InitialNavigation{Kind: InitialNavigationCreateEphemeral, Destination: ports.BrokerEndpointFence{Local: true}}
	550	+	}
	551	+	return InitialNavigation{}, pending
545	552		}
546	553		// The target exists only in the later publication: the stream
547	554		// request must resolve against the snapshot that decided.

@@ -570,7 +577,12 @@ func TestInitialNavigationWaitsForTargetObservation(t *testing.T) {
570	577		service.mu.Unlock()
571	578		require.Equal(t, []string{"user@host"}, reconciles, "the wait asks the broker for one fresh observation")
572	579
573		-	if tt.wantOpen {
	580	+	if tt.fallback {
	581	+	budget.fire()
	582	+	require.Eventually(t, func() bool { return opened.Load() == 1 }, 5*time.Second, time.Millisecond)
	583	+	require.Equal(t, protocol.ExactSessionTarget{}, service.openedRequests()[0].Target, "the fallback creation opens, not an exact attach")
	584	+	require.Equal(t, int64(1), calls.Load())
	585	+	} else if tt.wantOpen {
574	586		service.publishSnapshot(ports.BrokerSnapshot{Epoch: 3, Revision: 2, Daemons: []ports.BrokerDaemonObservation{
575	587		pickerTestLocalObservation(time.Unix(1000, 0), pickerTestSession("late", 5, catalogue_Up)),
576	588		}})

 		// observedAt is the first revision the resolver can decide on; zero
 		// means the target is never observed.
 		observedAt ports.BrokerRevision
-		wantOpen   bool
+		// fallback is offered with the pending resolution.
+		fallback bool
+		wantOpen bool
 	}{
 		{name: "a later publication resolves the armed intent", observedAt: 2, wantOpen: true},
 		{name: "the observation budget refuses once", observedAt: 0},
+		{name: "the observation budget takes the resolver's fallback", observedAt: 0, fallback: true, wantOpen: true},
 	}
 	for _, tt := range tests {
 		t.Run(tt.name, func(t *testing.T) {
 				ResolveInitialNavigation: func(snapshot ports.BrokerSnapshot) (InitialNavigation, error) {
 					calls.Add(1)
 					if tt.observedAt == 0 || snapshot.Revision < tt.observedAt {
-						return InitialNavigation{}, InitialNavigationNotObserved{Endpoint: "user@host"}
+						pending := InitialNavigationNotObserved{Endpoint: "user@host"}
+						if tt.fallback {
+							pending.Fallback = InitialNavigation{Kind: InitialNavigationCreateEphemeral, Destination: ports.BrokerEndpointFence{Local: true}}
+						}
+						return InitialNavigation{}, pending
 					}
 					// The target exists only in the later publication: the stream
 					// request must resolve against the snapshot that decided.
 			service.mu.Unlock()
 			require.Equal(t, []string{"user@host"}, reconciles, "the wait asks the broker for one fresh observation")
-			if tt.wantOpen {
+			if tt.fallback {
+				budget.fire()
+				require.Eventually(t, func() bool { return opened.Load() == 1 }, 5*time.Second, time.Millisecond)
+				require.Equal(t, protocol.ExactSessionTarget{}, service.openedRequests()[0].Target, "the fallback creation opens, not an exact attach")
+				require.Equal(t, int64(1), calls.Load())
+			} else if tt.wantOpen {
 				service.publishSnapshot(ports.BrokerSnapshot{Epoch: 3, Revision: 2, Daemons: []ports.BrokerDaemonObservation{
 					pickerTestLocalObservation(time.Unix(1000, 0), pickerTestSession("late", 5, catalogue_Up)),
 				}})

M internal/usecase/client/supervisor_attach.go +14 -5


@@ -396,8 +396,13 @@ var ErrInitialNavigationNotObserved = errors.New("vev: initial navigation target
396	396		// InitialNavigationNotObserved is the resolver's ErrInitialNavigationNotObserved
397	397		// naming the configured endpoint whose fresh observation it needs; the
398	398		// supervisor asks the broker to reconcile that endpoint once per wait.
	399	+	// Fallback, when it is not the picker, is the navigation to take if the budget
	400	+	// expires before the endpoint is observed (a remote host whose observation
	401	+	// fails can still serve a stream); the destination daemon stays the final
	402	+	// authority over it.
399	403		type InitialNavigationNotObserved struct {
400	404		Endpoint string
	405	+	Fallback InitialNavigation
401	406		}
402	407
403	408		func (e InitialNavigationNotObserved) Error() string {

@@ -426,10 +431,10 @@ type initialNavigationTake struct {
426	431
427	432		// awaitInitialNavigationObservation re-resolves the armed initial navigation
428	433		// on each committed publication until the resolver can decide, the budget
429		-	// expires (a bounded refusal), or the run ends. The returned snapshot is the
430		-	// publication the decision was made on, so the stream request resolves against
431		-	// the same authority.
432		-	func (s Supervisor) awaitInitialNavigationObservation(ctx context.Context, input terminalInputLifetime, service ports.BrokerService) initialNavigationTake {
	434	+	// expires (the resolver's fallback, or a bounded refusal without one), or the
	435	+	// run ends. The returned snapshot is the publication the decision was made on,
	436	+	// so the stream request resolves against the same authority.
	437	+	func (s Supervisor) awaitInitialNavigationObservation(ctx context.Context, input terminalInputLifetime, service ports.BrokerService, pending InitialNavigationNotObserved) initialNavigationTake {
433	438		timer := s.cfg.Clock.NewTimer(initialNavigationObservationBudget)
434	439		defer timer.Stop()
435	440		for {

@@ -438,6 +443,7 @@ func (s *Supervisor) awaitInitialNavigationObservation(ctx context.Context, inpu
438	443		snapshot := service.Snapshot()
439	444		navigation, consumed, err := s.takeInitialNavigation(snapshot)
440	445		if errors.Is(err, ErrInitialNavigationNotObserved) {
	446	+	_ = errors.As(err, &pending)
441	447		continue
442	448		}
443	449		return initialNavigationTake{snapshot: snapshot, navigation: navigation, consumed: consumed, err: err}

@@ -445,6 +451,9 @@ func (s *Supervisor) awaitInitialNavigationObservation(ctx context.Context, inpu
445	451		s.mu.Lock()
446	452		s.navigationConsumed = true
447	453		s.mu.Unlock()
	454	+	if pending.Fallback.Kind != InitialNavigationPicker {
	455	+	return initialNavigationTake{snapshot: service.Snapshot(), navigation: pending.Fallback, consumed: true}
	456	+	}
448	457		return initialNavigationTake{consumed: true, err: ErrInitialNavigationNotObserved}
449	458		case <-service.Done():
450	459		// The ready loop observes the loss; the intent stays armed for the

@@ -475,7 +484,7 @@ func (s Supervisor) runInitialNavigation(ctx context.Context, input terminalIn
475	484		if errors.As(err, &pending) && pending.Endpoint != "" {
476	485		service.RequestReconcile(pending.Endpoint)
477	486		}
478		-	take := s.awaitInitialNavigationObservation(ctx, input, service)
	487	+	take := s.awaitInitialNavigationObservation(ctx, input, service, pending)
479	488		if take.terminated {
480	489		return true, take.termErr
481	490		}

 // InitialNavigationNotObserved is the resolver's ErrInitialNavigationNotObserved
 // naming the configured endpoint whose fresh observation it needs; the
 // supervisor asks the broker to reconcile that endpoint once per wait.
+// Fallback, when it is not the picker, is the navigation to take if the budget
+// expires before the endpoint is observed (a remote host whose observation
+// fails can still serve a stream); the destination daemon stays the final
+// authority over it.
 type InitialNavigationNotObserved struct {
 	Endpoint string
+	Fallback InitialNavigation
 }
 func (e InitialNavigationNotObserved) Error() string {
 // awaitInitialNavigationObservation re-resolves the armed initial navigation
 // on each committed publication until the resolver can decide, the budget
-// expires (a bounded refusal), or the run ends. The returned snapshot is the
+// expires (the resolver's fallback, or a bounded refusal without one), or the
-// publication the decision was made on, so the stream request resolves against
+// run ends. The returned snapshot is the publication the decision was made on,
-// the same authority.
+// so the stream request resolves against the same authority.
-func (s *Supervisor) awaitInitialNavigationObservation(ctx context.Context, input *terminalInputLifetime, service ports.BrokerService) initialNavigationTake {
+func (s *Supervisor) awaitInitialNavigationObservation(ctx context.Context, input *terminalInputLifetime, service ports.BrokerService, pending InitialNavigationNotObserved) initialNavigationTake {
 	timer := s.cfg.Clock.NewTimer(initialNavigationObservationBudget)
 	defer timer.Stop()
 	for {
 			snapshot := service.Snapshot()
 			navigation, consumed, err := s.takeInitialNavigation(snapshot)
 			if errors.Is(err, ErrInitialNavigationNotObserved) {
+				_ = errors.As(err, &pending)
 				continue
 			}
 			return initialNavigationTake{snapshot: snapshot, navigation: navigation, consumed: consumed, err: err}
 			s.mu.Lock()
 			s.navigationConsumed = true
 			s.mu.Unlock()
+			if pending.Fallback.Kind != InitialNavigationPicker {
+				return initialNavigationTake{snapshot: service.Snapshot(), navigation: pending.Fallback, consumed: true}
+			}
 			return initialNavigationTake{consumed: true, err: ErrInitialNavigationNotObserved}
 		case <-service.Done():
 			// The ready loop observes the loss; the intent stays armed for the
 		if errors.As(err, &pending) && pending.Endpoint != "" {
 			service.RequestReconcile(pending.Endpoint)
 		}
-		take := s.awaitInitialNavigationObservation(ctx, input, service)
+		take := s.awaitInitialNavigationObservation(ctx, input, service, pending)
 		if take.terminated {
 			return true, take.termErr
 		}