feat(broker): reconcile configured observations

commit: a6ff6975dbc97e1361234a01ab8db767c0eccbf8
author: brice · 32d ago · 2026-09-20 10:37 UTC
parent: b337f41
signature: GPG VerifiedCD88F9E564E0102646FB6077A1A07C7C7C3E349D · arch-gpg Verified for the committer email

2 files · +51 -5

M internal/usecase/broker/service.go +12 -3


@@ -312,9 +312,18 @@ func membershipError(err error) error {
312	312		return ports.BrokerError{Code: code, Text: text, Cause: err}
313	313		}
314	314
315		-	// RequestReconcile remains a no-op until the observation service exposes a
316		-	// connection-safe reconcile seam.
317		-	func (s *Service) RequestReconcile(string) {}
	315	+	// RequestReconcile schedules one coalesced observation for a configured
	316	+	// endpoint. The registry validates membership and drops unknown endpoints, so
	317	+	// this hint cannot expand authority or start an attachment.
	318	+	func (s *Service) RequestReconcile(endpoint string) {
	319	+	s.mu.Lock()
	320	+	closed := s.closed
	321	+	s.mu.Unlock()
	322	+	if closed {
	323	+	return
	324	+	}
	325	+	s.registry.RequestProbe(endpoint)
	326	+	}
318	327
319	328		// Close cancels and drains every owned resource exactly once: in-flight setup
320	329		// opens and live streams stop, subscriptions close, the pool client is retired,

 	return ports.BrokerError{Code: code, Text: text, Cause: err}
 }
-// RequestReconcile remains a no-op until the observation service exposes a
+// RequestReconcile schedules one coalesced observation for a configured
-// connection-safe reconcile seam.
+// endpoint. The registry validates membership and drops unknown endpoints, so
-func (s *Service) RequestReconcile(string) {}
+// this hint cannot expand authority or start an attachment.
+func (s *Service) RequestReconcile(endpoint string) {
+	s.mu.Lock()
+	closed := s.closed
+	s.mu.Unlock()
+	if closed {
+		return
+	}
+	s.registry.RequestProbe(endpoint)
+}
 // Close cancels and drains every owned resource exactly once: in-flight setup
 // opens and live streams stop, subscriptions close, the pool client is retired,

M internal/usecase/broker/service_test.go +39 -2


@@ -531,7 +531,7 @@ func TestServiceShutdownTerminatesOwnedWork(t *testing.T) {
531	531		// TestServiceImmutableMembershipRefusesEveryMutation proves the immutable
532	532		// connection refuses all three membership mutations with the typed immutable
533	533		// code 12 and the ports sentinel before any durable authority moves: no CAS is
534		-	// attempted, no publication changes, and reconcile stays a no-op.
	534	+	// attempted and no publication changes.
535	535		func TestServiceImmutableMembershipRefusesEveryMutation(t *testing.T) {
536	536		store := newMembershipStore()
537	537		authority, registry, _, _ := newServiceMembershipFixture(t, store, MembershipImmutable)

@@ -555,7 +555,44 @@ func TestServiceImmutableMembershipRefusesEveryMutation(t *testing.T) {
555	555		require.Equal(t, before, registry.Snapshot(), "a refused mutation must not touch authority")
556	556
557	557		service.RequestReconcile("new@host:22")
558		-	require.Equal(t, before, registry.Snapshot(), "reconcile must be a no-op")
	558	+	require.Equal(t, before, registry.Snapshot(), "an unknown reconcile hint cannot change authority")
	559	+	}
	560	+
	561	+	func TestServiceRequestReconcileDelegatesToRegistry(t *testing.T) {
	562	+	clock := newManualClock(time.Unix(0, 0))
	563	+	store := newTestStore()
	564	+	registration := registration(t, "user@host:22", 1)
	565	+	store.hosts = ports.BrokerHosts{Revision: 1, Hosts: []ports.BrokerHostRecord{{Registration: registration, Pinned: true, Policy: poolPolicy()}}}
	566	+	probe := newTestProbe(2)
	567	+	registry, err := NewRegistry(1, store, probe, clock, nil)
	568	+	require.NoError(t, err)
	569	+	registry.jitter = identityJitter
	570	+	authority, _, _, _ := composeTestAuthority(t, 1, registry, immediateConnector, clock)
	571	+	service, err := authority.AdmitClient(context.Background())
	572	+	require.NoError(t, err)
	573	+	t.Cleanup(func() { require.NoError(t, service.Close()) })
	574	+
	575	+	var initial probeCall
	576	+	require.Eventually(t, func() bool {
	577	+	select {
	578	+	case initial = <-probe.calls:
	579	+	return true
	580	+	default:
	581	+	return false
	582	+	}
	583	+	}, time.Second, time.Millisecond)
	584	+	initial.result <- probeAnswer{snapshot: ports.BrokerDaemonObservation{Identity: "daemon", Incarnation: ports.BrokerDaemonIncarnation{1}, ProtocolVersion: 1, Availability: domain.RemoteAvailabilityReachable}}
	585	+	service.RequestReconcile(registration.Endpoint)
	586	+	var reconciled probeCall
	587	+	require.Eventually(t, func() bool {
	588	+	select {
	589	+	case reconciled = <-probe.calls:
	590	+	return true
	591	+	default:
	592	+	return false
	593	+	}
	594	+	}, time.Second, time.Millisecond)
	595	+	require.Equal(t, registration, reconciled.registration)
559	596		}
560	597
561	598		// TestServiceMutableMembershipDelegatesOnce proves an admitted mutable


@@ -531,7 +531,7 @@ func TestServiceShutdownTerminatesOwnedWork(t *testing.T) {
531		// TestServiceImmutableMembershipRefusesEveryMutation proves the immutable	531		// TestServiceImmutableMembershipRefusesEveryMutation proves the immutable
532		// connection refuses all three membership mutations with the typed immutable	532		// connection refuses all three membership mutations with the typed immutable
533		// code 12 and the ports sentinel before any durable authority moves: no CAS is	533		// code 12 and the ports sentinel before any durable authority moves: no CAS is
534	-	// attempted, no publication changes, and reconcile stays a no-op.	534	+	// attempted and no publication changes.
535		func TestServiceImmutableMembershipRefusesEveryMutation(t *testing.T) {	535		func TestServiceImmutableMembershipRefusesEveryMutation(t *testing.T) {
536		store := newMembershipStore()	536		store := newMembershipStore()
537		authority, registry, _, _ := newServiceMembershipFixture(t, store, MembershipImmutable)	537		authority, registry, _, _ := newServiceMembershipFixture(t, store, MembershipImmutable)

@@ -555,7 +555,44 @@ func TestServiceImmutableMembershipRefusesEveryMutation(t *testing.T) {
555		require.Equal(t, before, registry.Snapshot(), "a refused mutation must not touch authority")	555		require.Equal(t, before, registry.Snapshot(), "a refused mutation must not touch authority")
556			556
557		service.RequestReconcile("new@host:22")	557		service.RequestReconcile("new@host:22")
558	-	require.Equal(t, before, registry.Snapshot(), "reconcile must be a no-op")	558	+	require.Equal(t, before, registry.Snapshot(), "an unknown reconcile hint cannot change authority")
			559	+	}
			560	+
			561	+	func TestServiceRequestReconcileDelegatesToRegistry(t *testing.T) {
			562	+	clock := newManualClock(time.Unix(0, 0))
			563	+	store := newTestStore()
			564	+	registration := registration(t, "user@host:22", 1)
			565	+	store.hosts = ports.BrokerHosts{Revision: 1, Hosts: []ports.BrokerHostRecord{{Registration: registration, Pinned: true, Policy: poolPolicy()}}}
			566	+	probe := newTestProbe(2)
			567	+	registry, err := NewRegistry(1, store, probe, clock, nil)
			568	+	require.NoError(t, err)
			569	+	registry.jitter = identityJitter
			570	+	authority, _, _, _ := composeTestAuthority(t, 1, registry, immediateConnector, clock)
			571	+	service, err := authority.AdmitClient(context.Background())
			572	+	require.NoError(t, err)
			573	+	t.Cleanup(func() { require.NoError(t, service.Close()) })
			574	+
			575	+	var initial probeCall
			576	+	require.Eventually(t, func() bool {
			577	+	select {
			578	+	case initial = <-probe.calls:
			579	+	return true
			580	+	default:
			581	+	return false
			582	+	}
			583	+	}, time.Second, time.Millisecond)
			584	+	initial.result <- probeAnswer{snapshot: ports.BrokerDaemonObservation{Identity: "daemon", Incarnation: ports.BrokerDaemonIncarnation{1}, ProtocolVersion: 1, Availability: domain.RemoteAvailabilityReachable}}
			585	+	service.RequestReconcile(registration.Endpoint)
			586	+	var reconciled probeCall
			587	+	require.Eventually(t, func() bool {
			588	+	select {
			589	+	case reconciled = <-probe.calls:
			590	+	return true
			591	+	default:
			592	+	return false
			593	+	}
			594	+	}, time.Second, time.Millisecond)
			595	+	require.Equal(t, registration, reconciled.registration)
559		}	596		}
560			597
561		// TestServiceMutableMembershipDelegatesOnce proves an admitted mutable	598		// TestServiceMutableMembershipDelegatesOnce proves an admitted mutable