fix(broker): re-probe the local daemon after local streams complete

commit: 0125c1b41d13048914b32808a30ef725dd9d35a3
author: brice · 30d ago · 2026-09-22 20:09 UTC
parent: aa991bf
signature: GPG VerifiedCD88F9E564E0102646FB6077A1A07C7C7C3E349D · arch-gpg Verified for the committer email

4 files · +123 -5

M internal/usecase/broker/registry.go +16 -2


@@ -133,6 +133,12 @@ type Registry struct {
133	133		local *LocalObservation
134	134		localHost ports.BrokerDaemonObservation
135	135		localAttempt *probeAttempt
	136	+	// localPending marks explicit demand for a local re-probe ahead of its
	137	+	// scheduled freshness window: RequestProbe("") and a completed local control
	138	+	// stream or attach both set it. dispatchLocalLocked honors it the same way
	139	+	// pending honors remote demand, admitting the attempt before NextDue and
	140	+	// clearing the flag the moment the attempt starts.
	141	+	localPending bool
136	142
137	143		freshFor time.Duration
138	144		retryBase time.Duration

@@ -835,13 +841,21 @@ func (r *Registry) cancelAllInflightLocked() {
835	841
836	842		// RequestProbe is non-blocking and coalesces concurrent demand per endpoint.
837	843		// A disabled registry never observes, so the demand is dropped without
838		-	// scheduling anything.
	844	+	// scheduling anything. An empty endpoint requests a local re-probe instead of a
	845	+	// configured remote: the broker's own machine daemon is re-observed ahead of
	846	+	// its scheduled freshness window, honored by dispatchLocalLocked exactly like
	847	+	// pending honors remote demand. A registry with no local observation producer
	848	+	// drops it, just like an unknown remote endpoint is dropped.
839	849		func (r *Registry) RequestProbe(endpoint string) {
840	850		if r.observationDisabled {
841	851		return
842	852		}
843	853		r.mu.Lock()
844		-	if _, ok := r.hosts[endpoint]; ok {
	854	+	if endpoint == "" {
	855	+	if r.local != nil {
	856	+	r.localPending = true
	857	+	}
	858	+	} else if _, ok := r.hosts[endpoint]; ok {
845	859		r.pending[endpoint] = true
846	860		}
847	861		r.mu.Unlock()

 	local        *LocalObservation
 	localHost    ports.BrokerDaemonObservation
 	localAttempt *probeAttempt
+	// localPending marks explicit demand for a local re-probe ahead of its
+	// scheduled freshness window: RequestProbe("") and a completed local control
+	// stream or attach both set it. dispatchLocalLocked honors it the same way
+	// pending honors remote demand, admitting the attempt before NextDue and
+	// clearing the flag the moment the attempt starts.
+	localPending bool
 	freshFor  time.Duration
 	retryBase time.Duration
 // RequestProbe is non-blocking and coalesces concurrent demand per endpoint.
 // A disabled registry never observes, so the demand is dropped without
-// scheduling anything.
+// scheduling anything. An empty endpoint requests a local re-probe instead of a
+// configured remote: the broker's own machine daemon is re-observed ahead of
+// its scheduled freshness window, honored by dispatchLocalLocked exactly like
+// pending honors remote demand. A registry with no local observation producer
+// drops it, just like an unknown remote endpoint is dropped.
 func (r *Registry) RequestProbe(endpoint string) {
 	if r.observationDisabled {
 		return
 	}
 	r.mu.Lock()
-	if _, ok := r.hosts[endpoint]; ok {
+	if endpoint == "" {
+		if r.local != nil {
+			r.localPending = true
+		}
+	} else if _, ok := r.hosts[endpoint]; ok {
 		r.pending[endpoint] = true
 	}
 	r.mu.Unlock()

M internal/usecase/broker/registry_local.go +6 -3


@@ -79,15 +79,18 @@ func (l *LocalObservation) validate() error {
79	79		}
80	80
81	81		// dispatchLocalLocked admits at most one local probe attempt when its schedule
82		-	// says it is due. Callers must hold r.mu. It reports whether an attempt
83		-	// started, so dispatch publishes the transient Checking state exactly once.
	82	+	// says it is due, or immediately when localPending marks explicit demand for a
	83	+	// re-probe ahead of schedule. Callers must hold r.mu. It reports whether an
	84	+	// attempt started, so dispatch publishes the transient Checking state exactly
	85	+	// once.
84	86		func (r *Registry) dispatchLocalLocked(ctx context.Context, now time.Time, results chan<- probeResult) bool {
85	87		if r.local == nil \|\| r.localAttempt != nil {
86	88		return false
87	89		}
88		-	if !r.localHost.NextDue.IsZero() && now.Before(r.localHost.NextDue) {
	90	+	if !r.localPending && !r.localHost.NextDue.IsZero() && now.Before(r.localHost.NextDue) {
89	91		return false
90	92		}
	93	+	r.localPending = false
91	94		r.attempts++
92	95		attempt := r.attempts
93	96		probeCtx, cancel := context.WithCancel(ctx)

 }
 // dispatchLocalLocked admits at most one local probe attempt when its schedule
-// says it is due. Callers must hold r.mu. It reports whether an attempt
+// says it is due, or immediately when localPending marks explicit demand for a
-// started, so dispatch publishes the transient Checking state exactly once.
+// re-probe ahead of schedule. Callers must hold r.mu. It reports whether an
+// attempt started, so dispatch publishes the transient Checking state exactly
+// once.
 func (r *Registry) dispatchLocalLocked(ctx context.Context, now time.Time, results chan<- probeResult) bool {
 	if r.local == nil || r.localAttempt != nil {
 		return false
 	}
-	if !r.localHost.NextDue.IsZero() && now.Before(r.localHost.NextDue) {
+	if !r.localPending && !r.localHost.NextDue.IsZero() && now.Before(r.localHost.NextDue) {
 		return false
 	}
+	r.localPending = false
 	r.attempts++
 	attempt := r.attempts
 	probeCtx, cancel := context.WithCancel(ctx)

M internal/usecase/broker/registry_local_test.go +90 -0


@@ -69,6 +69,20 @@ func receiveLocalCall(t testing.T, probe scriptedLocalProbe) localProbeCall {
69	69		}
70	70		}
71	71
	72	+	// requireNoLocalProbeWithin asserts a negative scheduling outcome: no local
	73	+	// probe attempt starts inside window, so the test observes "nothing happened"
	74	+	// without sleeping on the behavior under test.
	75	+	func requireNoLocalProbeWithin(t testing.T, probe scriptedLocalProbe, window time.Duration) {
	76	+	t.Helper()
	77	+	timer := time.NewTimer(window)
	78	+	defer timer.Stop()
	79	+	select {
	80	+	case <-probe.calls:
	81	+	t.Fatal("unexpected local probe attempt")
	82	+	case <-timer.C:
	83	+	}
	84	+	}
	85	+
72	86		// newLocalTestRegistry builds a registry that observes the broker's own machine
73	87		// daemon through localProbe, beside an optional remote probe.
74	88		func newLocalTestRegistry(t testing.T, store ports.BrokerHostStore, remoteProbe ports.BrokerHostProbe, localProbe LocalProbe, clock manualClock) *Registry {

@@ -198,6 +212,82 @@ func TestRegistryLocalObservationNeverInventsState(t *testing.T) {
198	212		})
199	213		}
200	214
	215	+	// TestRegistryLocalPendingReprobeHonorsExplicitReconcileOnly proves the local
	216	+	// pending re-probe flag (set by RequestProbe("")) is the only thing that
	217	+	// admits a local attempt ahead of its scheduled NextDue: neither an explicit
	218	+	// unknown result nor a plain success schedules an early re-probe on its own,
	219	+	// but a reconcile requested right after either one is dispatched immediately,
	220	+	// on the fake clock, with no time advance at all. This is the fix for the
	221	+	// stale local catalogue after create/attach (`ls --all`/the picker showing a
	222	+	// new or newly-attached session for up to the retry/fresh interval): the
	223	+	// service now calls RequestProbe("") once a local control or attachment
	224	+	// stream completes, and this test pins the registry-side half of that fix.
	225	+	func TestRegistryLocalPendingReprobeHonorsExplicitReconcileOnly(t *testing.T) {
	226	+	reachable := localProbeAnswer{snapshot: ports.BrokerDaemonObservation{
	227	+	Identity: "local-daemon",
	228	+	Incarnation: ports.BrokerDaemonIncarnation{9},
	229	+	ProtocolVersion: protocol.Version,
	230	+	Availability: domain.RemoteAvailabilityReachable,
	231	+	}}
	232	+	unknown := localProbeAnswer{snapshot: ports.BrokerDaemonObservation{Availability: domain.RemoteAvailabilityUnknown}}
	233	+
	234	+	tests := []struct {
	235	+	name string
	236	+	first localProbeAnswer
	237	+	requestReconcile bool
	238	+	wantReprobe bool
	239	+	}{
	240	+	{
	241	+	name: "after an unknown result, no reconcile never re-probes before due",
	242	+	first: unknown,
	243	+	wantReprobe: false,
	244	+	},
	245	+	{
	246	+	name: "after a success, no reconcile never re-probes before due",
	247	+	first: reachable,
	248	+	wantReprobe: false,
	249	+	},
	250	+	{
	251	+	name: "after an unknown result, an explicit local reconcile re-probes before due",
	252	+	first: unknown,
	253	+	requestReconcile: true,
	254	+	wantReprobe: true,
	255	+	},
	256	+	{
	257	+	name: "after a success, an explicit local reconcile re-probes before due",
	258	+	first: reachable,
	259	+	requestReconcile: true,
	260	+	wantReprobe: true,
	261	+	},
	262	+	}
	263	+
	264	+	for _, tt := range tests {
	265	+	t.Run(tt.name, func(t *testing.T) {
	266	+	clock := newManualClock(time.Unix(100, 0))
	267	+	local := newScriptedLocalProbe(2)
	268	+	registry := newLocalTestRegistry(t, newTestStore(), nil, local, clock)
	269	+	startRegistry(t, registry)
	270	+
	271	+	call := receiveLocalCall(t, local)
	272	+	call.result <- tt.first
	273	+	waitLocal(t, registry, func(o ports.BrokerDaemonObservation) bool { return !o.Checking })
	274	+
	275	+	if tt.requestReconcile {
	276	+	registry.RequestProbe("")
	277	+	}
	278	+
	279	+	// No clock advance at all: NextDue (freshFor after success, the retry
	280	+	// delay after unknown) is still in the future either way, so only the
	281	+	// pending flag - never the schedule - can explain a dispatch here.
	282	+	if tt.wantReprobe {
	283	+	receiveLocalCall(t, local)
	284	+	} else {
	285	+	requireNoLocalProbeWithin(t, local, probeGrace)
	286	+	}
	287	+	})
	288	+	}
	289	+	}
	290	+
201	291		// TestRegistryLocalObservationValidatesProbeResults proves an invalid observed
202	292		// projection is classified as an invalid response, never published as an
203	293		// observation the wire refuses: a partial identity (identity without a version)


@@ -69,6 +69,20 @@ func receiveLocalCall(t testing.T, probe scriptedLocalProbe) localProbeCall {
69	}	69		}
70	}	70		}
71		71
		72	+	// requireNoLocalProbeWithin asserts a negative scheduling outcome: no local
		73	+	// probe attempt starts inside window, so the test observes "nothing happened"
		74	+	// without sleeping on the behavior under test.
		75	+	func requireNoLocalProbeWithin(t testing.T, probe scriptedLocalProbe, window time.Duration) {
		76	+	t.Helper()
		77	+	timer := time.NewTimer(window)
		78	+	defer timer.Stop()
		79	+	select {
		80	+	case <-probe.calls:
		81	+	t.Fatal("unexpected local probe attempt")
		82	+	case <-timer.C:
		83	+	}
		84	+	}
		85	+
72	// newLocalTestRegistry builds a registry that observes the broker's own machine	86		// newLocalTestRegistry builds a registry that observes the broker's own machine
73	// daemon through localProbe, beside an optional remote probe.	87		// daemon through localProbe, beside an optional remote probe.
74	func newLocalTestRegistry(t testing.T, store ports.BrokerHostStore, remoteProbe ports.BrokerHostProbe, localProbe LocalProbe, clock manualClock) *Registry {	88		func newLocalTestRegistry(t testing.T, store ports.BrokerHostStore, remoteProbe ports.BrokerHostProbe, localProbe LocalProbe, clock manualClock) *Registry {

@@ -198,6 +212,82 @@ func TestRegistryLocalObservationNeverInventsState(t *testing.T) {
198	})	212		})
199	}	213		}
200		214
		215	+	// TestRegistryLocalPendingReprobeHonorsExplicitReconcileOnly proves the local
		216	+	// pending re-probe flag (set by RequestProbe("")) is the only thing that
		217	+	// admits a local attempt ahead of its scheduled NextDue: neither an explicit
		218	+	// unknown result nor a plain success schedules an early re-probe on its own,
		219	+	// but a reconcile requested right after either one is dispatched immediately,
		220	+	// on the fake clock, with no time advance at all. This is the fix for the
		221	+	// stale local catalogue after create/attach (`ls --all`/the picker showing a
		222	+	// new or newly-attached session for up to the retry/fresh interval): the
		223	+	// service now calls RequestProbe("") once a local control or attachment
		224	+	// stream completes, and this test pins the registry-side half of that fix.
		225	+	func TestRegistryLocalPendingReprobeHonorsExplicitReconcileOnly(t *testing.T) {
		226	+	reachable := localProbeAnswer{snapshot: ports.BrokerDaemonObservation{
		227	+	Identity: "local-daemon",
		228	+	Incarnation: ports.BrokerDaemonIncarnation{9},
		229	+	ProtocolVersion: protocol.Version,
		230	+	Availability: domain.RemoteAvailabilityReachable,
		231	+	}}
		232	+	unknown := localProbeAnswer{snapshot: ports.BrokerDaemonObservation{Availability: domain.RemoteAvailabilityUnknown}}
		233	+
		234	+	tests := []struct {
		235	+	name string
		236	+	first localProbeAnswer
		237	+	requestReconcile bool
		238	+	wantReprobe bool
		239	+	}{
		240	+	{
		241	+	name: "after an unknown result, no reconcile never re-probes before due",
		242	+	first: unknown,
		243	+	wantReprobe: false,
		244	+	},
		245	+	{
		246	+	name: "after a success, no reconcile never re-probes before due",
		247	+	first: reachable,
		248	+	wantReprobe: false,
		249	+	},
		250	+	{
		251	+	name: "after an unknown result, an explicit local reconcile re-probes before due",
		252	+	first: unknown,
		253	+	requestReconcile: true,
		254	+	wantReprobe: true,
		255	+	},
		256	+	{
		257	+	name: "after a success, an explicit local reconcile re-probes before due",
		258	+	first: reachable,
		259	+	requestReconcile: true,
		260	+	wantReprobe: true,
		261	+	},
		262	+	}
		263	+
		264	+	for _, tt := range tests {
		265	+	t.Run(tt.name, func(t *testing.T) {
		266	+	clock := newManualClock(time.Unix(100, 0))
		267	+	local := newScriptedLocalProbe(2)
		268	+	registry := newLocalTestRegistry(t, newTestStore(), nil, local, clock)
		269	+	startRegistry(t, registry)
		270	+
		271	+	call := receiveLocalCall(t, local)
		272	+	call.result <- tt.first
		273	+	waitLocal(t, registry, func(o ports.BrokerDaemonObservation) bool { return !o.Checking })
		274	+
		275	+	if tt.requestReconcile {
		276	+	registry.RequestProbe("")
		277	+	}
		278	+
		279	+	// No clock advance at all: NextDue (freshFor after success, the retry
		280	+	// delay after unknown) is still in the future either way, so only the
		281	+	// pending flag - never the schedule - can explain a dispatch here.
		282	+	if tt.wantReprobe {
		283	+	receiveLocalCall(t, local)
		284	+	} else {
		285	+	requireNoLocalProbeWithin(t, local, probeGrace)
		286	+	}
		287	+	})
		288	+	}
		289	+	}
		290	+
201	// TestRegistryLocalObservationValidatesProbeResults proves an invalid observed	291		// TestRegistryLocalObservationValidatesProbeResults proves an invalid observed
202	// projection is classified as an invalid response, never published as an	292		// projection is classified as an invalid response, never published as an
203	// observation the wire refuses: a partial identity (identity without a version)	293		// observation the wire refuses: a partial identity (identity without a version)

M internal/usecase/broker/service.go +11 -0


@@ -266,6 +266,17 @@ func (s *Service) OpenStream(ctx context.Context, request ports.BrokerOpenStream
266	266		go func() {
267	267		<-stream.Done()
268	268		release()
	269	+	// A completed local control or attachment stream is the natural signal
	270	+	// that the local daemon's own catalogue may have just changed (a session
	271	+	// was created, killed, or its attachment state flipped): mark a local
	272	+	// re-probe pending so the next dispatch observes it ahead of the
	273	+	// freshness window instead of leaving `ls --all`/the picker stale for up
	274	+	// to the retry/fresh interval. An observation stream never triggers this:
	275	+	// it is the probe traffic itself, and re-arming from it would starve the
	276	+	// schedule instead of catching up to it.
	277	+	if request.Local && (request.Purpose == ports.BrokerStreamControl \|\| request.Purpose == ports.BrokerStreamAttachment) {
	278	+	s.registry.RequestProbe("")
	279	+	}
269	280		}()
270	281		return stream, nil
271	282		}


@@ -266,6 +266,17 @@ func (s *Service) OpenStream(ctx context.Context, request ports.BrokerOpenStream
266	go func() {	266		go func() {
267	<-stream.Done()	267		<-stream.Done()
268	release()	268		release()
		269	+	// A completed local control or attachment stream is the natural signal
		270	+	// that the local daemon's own catalogue may have just changed (a session
		271	+	// was created, killed, or its attachment state flipped): mark a local
		272	+	// re-probe pending so the next dispatch observes it ahead of the
		273	+	// freshness window instead of leaving `ls --all`/the picker stale for up
		274	+	// to the retry/fresh interval. An observation stream never triggers this:
		275	+	// it is the probe traffic itself, and re-arming from it would starve the
		276	+	// schedule instead of catching up to it.
		277	+	if request.Local && (request.Purpose == ports.BrokerStreamControl \|\| request.Purpose == ports.BrokerStreamAttachment) {
		278	+	s.registry.RequestProbe("")
		279	+	}
269	}()	280		}()
270	return stream, nil	281		return stream, nil
271	}	282		}