feat(broker): observe remote daemons without attaching

commit: 0ab3226b2f17e6d5c3bc3077c29868b8ee76b3da
author: brice · 32d ago · 2026-09-20 12:27 UTC
parent: a6ff697
signature: GPG VerifiedCD88F9E564E0102646FB6077A1A07C7C7C3E349D · arch-gpg Verified for the committer email

4 files · +199 -6

M internal/adapters/brokerconfig/config.go +9 -0


@@ -441,6 +441,15 @@ func (c *Config) Endpoints() []string {
441	441		return append([]string(nil), c.order...)
442	442		}
443	443
	444	+	// Registration returns the immutable configured authority for one endpoint.
	445	+	func (c *Config) Registration(endpoint string) (Registration, bool) {
	446	+	if c == nil {
	447	+	return Registration{}, false
	448	+	}
	449	+	registration, ok := c.byEndpoint[endpoint]
	450	+	return registration, ok
	451	+	}
	452	+
444	453		// IdleGrace returns the provisioned effective idle grace and whether the
445	454		// configuration provisioned one. A nil configuration provisions nothing.
446	455		func (c *Config) IdleGrace() (time.Duration, bool) {


@@ -441,6 +441,15 @@ func (c *Config) Endpoints() []string {
441	return append([]string(nil), c.order...)	441		return append([]string(nil), c.order...)
442	}	442		}
443		443
		444	+	// Registration returns the immutable configured authority for one endpoint.
		445	+	func (c *Config) Registration(endpoint string) (Registration, bool) {
		446	+	if c == nil {
		447	+	return Registration{}, false
		448	+	}
		449	+	registration, ok := c.byEndpoint[endpoint]
		450	+	return registration, ok
		451	+	}
		452	+
444	// IdleGrace returns the provisioned effective idle grace and whether the	453		// IdleGrace returns the provisioned effective idle grace and whether the
445	// configuration provisioned one. A nil configuration provisions nothing.	454		// configuration provisioned one. A nil configuration provisions nothing.
446	func (c *Config) IdleGrace() (time.Duration, bool) {	455		func (c *Config) IdleGrace() (time.Duration, bool) {

M internal/app/broker_offline.go +17 -6


@@ -287,7 +287,22 @@ func runBrokerServe(ctx context.Context, options brokerServeOptions, deps broker
287	287		if err != nil {
288	288		return err
289	289		}
290		-	registry, err := broker.NewRegistryWithConfig(epoch, store, nil, clk, log, registryConfig)
	290	+	connector, err := brokerMuxConnector(config, log)
	291	+	if err != nil {
	292	+	return err
	293	+	}
	294	+	resolver := config.Resolver()
	295	+	var remoteProbe ports.BrokerHostProbe
	296	+	if len(config.Endpoints()) != 0 {
	297	+	remoteProbe = &brokerRemoteProbe{
	298	+	epoch: epoch, resolver: resolver, connector: connector,
	299	+	policy: func(endpoint string) (ports.BrokerPolicy, bool) {
	300	+	registration, ok := config.Registration(endpoint)
	301	+	return registration.Policy, ok
	302	+	},
	303	+	}
	304	+	}
	305	+	registry, err := broker.NewRegistryWithConfig(epoch, store, remoteProbe, clk, log, registryConfig)
291	306		if err != nil {
292	307		return err
293	308		}

@@ -297,11 +312,7 @@ func runBrokerServe(ctx context.Context, options brokerServeOptions, deps broker
297	312		if err := supervisor.RegisterRunner("registry", registry); err != nil {
298	313		return err
299	314		}
300		-	connector, err := brokerMuxConnector(config, log)
301		-	if err != nil {
302		-	return err
303		-	}
304		-	pool, err := broker.NewPool(epoch, config.Resolver(), connector, clk, brokerServePoolLimits)
	315	+	pool, err := broker.NewPool(epoch, resolver, connector, clk, brokerServePoolLimits)
305	316		if err != nil {
306	317		return err
307	318		}

 	if err != nil {
 		return err
 	}
-	registry, err := broker.NewRegistryWithConfig(epoch, store, nil, clk, log, registryConfig)
+	connector, err := brokerMuxConnector(config, log)
+	if err != nil {
+		return err
+	}
+	resolver := config.Resolver()
+	var remoteProbe ports.BrokerHostProbe
+	if len(config.Endpoints()) != 0 {
+		remoteProbe = &brokerRemoteProbe{
+			epoch: epoch, resolver: resolver, connector: connector,
+			policy: func(endpoint string) (ports.BrokerPolicy, bool) {
+				registration, ok := config.Registration(endpoint)
+				return registration.Policy, ok
+			},
+		}
+	}
+	registry, err := broker.NewRegistryWithConfig(epoch, store, remoteProbe, clk, log, registryConfig)
 	if err != nil {
 		return err
 	}
 	if err := supervisor.RegisterRunner("registry", registry); err != nil {
 		return err
 	}
-	connector, err := brokerMuxConnector(config, log)
+	pool, err := broker.NewPool(epoch, resolver, connector, clk, brokerServePoolLimits)
-	if err != nil {
-		return err
-	}
-	pool, err := broker.NewPool(epoch, config.Resolver(), connector, clk, brokerServePoolLimits)
 	if err != nil {
 		return err
 	}

A internal/app/broker_remote_probe.go +136 -0

@@ -0,0 +1,136 @@
	1	+	package app
	2	+
	3	+	import (
	4	+	"context"
	5	+	"crypto/rand"
	6	+	"encoding/binary"
	7	+	"encoding/json"
	8	+	"errors"
	9	+	"fmt"
	10	+
	11	+	"github.com/bnema/vev/internal/domain"
	12	+	"github.com/bnema/vev/internal/ports"
	13	+	"github.com/bnema/vev/internal/protocol"
	14	+	"github.com/bnema/vev/internal/protocol/catalogue"
	15	+	)
	16	+
	17	+	// brokerRemoteProbe observes one configured daemon over the broker-owned
	18	+	// multiplexed carriage. It opens an observation stream, never an attachment,
	19	+	// and asks the daemon for its local catalogue through the existing typed
	20	+	// command protocol. Authenticated daemon identity and process incarnation come
	21	+	// from the physical handshake rather than from the configured hostname.
	22	+	type brokerRemoteProbe struct {
	23	+	epoch ports.BrokerEpoch
	24	+	resolver ports.BrokerEndpointResolver
	25	+	connector ports.BrokerEndpointConnector
	26	+	policy func(string) (ports.BrokerPolicy, bool)
	27	+	}
	28	+
	29	+	var _ ports.BrokerHostProbe = (*brokerRemoteProbe)(nil)
	30	+
	31	+	func (p *brokerRemoteProbe) Probe(ctx context.Context, registration domain.RemoteRegistration) (ports.BrokerDaemonObservation, error) {
	32	+	if p == nil \|\| p.epoch == 0 \|\| p.resolver == nil \|\| p.connector == nil {
	33	+	return ports.BrokerDaemonObservation{}, errors.New("vev: incomplete broker remote probe")
	34	+	}
	35	+	request, err := p.request(registration)
	36	+	if err != nil {
	37	+	return ports.BrokerDaemonObservation{}, err
	38	+	}
	39	+	endpoint, err := p.resolver.Resolve(ctx, request)
	40	+	if err != nil {
	41	+	return ports.BrokerDaemonObservation{}, err
	42	+	}
	43	+	physical, err := p.connector.Connect(ctx, endpoint)
	44	+	if err != nil {
	45	+	return ports.BrokerDaemonObservation{}, err
	46	+	}
	47	+	defer physical.Close()
	48	+	if physical.Identity() != endpoint.Identity \|\| physical.Policy() != endpoint.Policy {
	49	+	return ports.BrokerDaemonObservation{}, errors.New("vev: remote probe authenticated binding mismatch")
	50	+	}
	51	+	logical, err := physical.OpenStream(ctx, request)
	52	+	if err != nil {
	53	+	return ports.BrokerDaemonObservation{}, err
	54	+	}
	55	+	defer logical.Close()
	56	+	commandID, err := randomNonzeroUint64()
	57	+	if err != nil {
	58	+	return ports.BrokerDaemonObservation{}, err
	59	+	}
	60	+	if err := logical.SendClient(protocol.CommandRequest{RequestID: commandID, Version: protocol.Version, Slug: "remote-catalog", JSON: true}); err != nil {
	61	+	return ports.BrokerDaemonObservation{}, err
	62	+	}
	63	+	reply, err := logical.ReceiveServer()
	64	+	if err != nil {
	65	+	return ports.BrokerDaemonObservation{}, err
	66	+	}
	67	+	result, ok := reply.(protocol.CommandResult)
	68	+	if !ok \|\| result.RequestID != commandID \|\| !result.Valid() {
	69	+	return ports.BrokerDaemonObservation{}, errors.New("vev: invalid remote observation result")
	70	+	}
	71	+	if result.Outcome != protocol.CommandSucceeded {
	72	+	return ports.BrokerDaemonObservation{}, fmt.Errorf("vev: remote observation failed: %s", result.Text)
	73	+	}
	74	+	var catalog catalogue.RemoteCatalog
	75	+	if err := json.Unmarshal([]byte(result.Output), &catalog); err != nil {
	76	+	return ports.BrokerDaemonObservation{}, fmt.Errorf("vev: decode remote observation: %w", err)
	77	+	}
	78	+	if err := catalogue.ValidateRemoteCatalog(catalog); err != nil {
	79	+	return ports.BrokerDaemonObservation{}, err
	80	+	}
	81	+	return ports.BrokerDaemonObservation{
	82	+	Identity: physical.Identity(),
	83	+	Incarnation: physical.Incarnation(),
	84	+	ProtocolVersion: catalog.ProtocolVersion,
	85	+	Capabilities: 0,
	86	+	Availability: domain.RemoteAvailabilityReachable,
	87	+	InventoryKnown: true,
	88	+	Sessions: catalog.Sessions,
	89	+	}, nil
	90	+	}
	91	+
	92	+	func (p *brokerRemoteProbe) request(registration domain.RemoteRegistration) (ports.BrokerOpenStreamRequest, error) {
	93	+	if p.policy == nil {
	94	+	return ports.BrokerOpenStreamRequest{}, errors.New("vev: remote probe has no policy authority")
	95	+	}
	96	+	policy, ok := p.policy(registration.Endpoint)
	97	+	if !ok {
	98	+	return ports.BrokerOpenStreamRequest{}, errors.New("vev: remote probe endpoint is not configured")
	99	+	}
	100	+	connection, err := randomConnectionID()
	101	+	if err != nil {
	102	+	return ports.BrokerOpenStreamRequest{}, err
	103	+	}
	104	+	stream, err := randomNonzeroUint64()
	105	+	if err != nil {
	106	+	return ports.BrokerOpenStreamRequest{}, err
	107	+	}
	108	+	return ports.BrokerOpenStreamRequest{
	109	+	Epoch: p.epoch, Purpose: ports.BrokerStreamObservation,
	110	+	Connection: connection, Stream: ports.BrokerStreamID(stream),
	111	+	Endpoint: registration.Endpoint, Registration: registration, Policy: policy,
	112	+	}, nil
	113	+	}
	114	+
	115	+	func randomConnectionID() (ports.BrokerConnectionID, error) {
	116	+	var id ports.BrokerConnectionID
	117	+	if _, err := rand.Read(id[:]); err != nil {
	118	+	return id, err
	119	+	}
	120	+	if err := id.Validate(); err != nil {
	121	+	return randomConnectionID()
	122	+	}
	123	+	return id, nil
	124	+	}
	125	+
	126	+	func randomNonzeroUint64() (uint64, error) {
	127	+	var raw [8]byte
	128	+	if _, err := rand.Read(raw[:]); err != nil {
	129	+	return 0, err
	130	+	}
	131	+	value := binary.BigEndian.Uint64(raw[:])
	132	+	if value == 0 {
	133	+	return randomNonzeroUint64()
	134	+	}
	135	+	return value, nil
	136	+	}

@@ -0,0 +1,136 @@
	1	+	package app
	2	+
	3	+	import (
	4	+	"context"
	5	+	"crypto/rand"
	6	+	"encoding/binary"
	7	+	"encoding/json"
	8	+	"errors"
	9	+	"fmt"
	10	+
	11	+	"github.com/bnema/vev/internal/domain"
	12	+	"github.com/bnema/vev/internal/ports"
	13	+	"github.com/bnema/vev/internal/protocol"
	14	+	"github.com/bnema/vev/internal/protocol/catalogue"
	15	+	)
	16	+
	17	+	// brokerRemoteProbe observes one configured daemon over the broker-owned
	18	+	// multiplexed carriage. It opens an observation stream, never an attachment,
	19	+	// and asks the daemon for its local catalogue through the existing typed
	20	+	// command protocol. Authenticated daemon identity and process incarnation come
	21	+	// from the physical handshake rather than from the configured hostname.
	22	+	type brokerRemoteProbe struct {
	23	+	epoch ports.BrokerEpoch
	24	+	resolver ports.BrokerEndpointResolver
	25	+	connector ports.BrokerEndpointConnector
	26	+	policy func(string) (ports.BrokerPolicy, bool)
	27	+	}
	28	+
	29	+	var _ ports.BrokerHostProbe = (*brokerRemoteProbe)(nil)
	30	+
	31	+	func (p *brokerRemoteProbe) Probe(ctx context.Context, registration domain.RemoteRegistration) (ports.BrokerDaemonObservation, error) {
	32	+	if p == nil \|\| p.epoch == 0 \|\| p.resolver == nil \|\| p.connector == nil {
	33	+	return ports.BrokerDaemonObservation{}, errors.New("vev: incomplete broker remote probe")
	34	+	}
	35	+	request, err := p.request(registration)
	36	+	if err != nil {
	37	+	return ports.BrokerDaemonObservation{}, err
	38	+	}
	39	+	endpoint, err := p.resolver.Resolve(ctx, request)
	40	+	if err != nil {
	41	+	return ports.BrokerDaemonObservation{}, err
	42	+	}
	43	+	physical, err := p.connector.Connect(ctx, endpoint)
	44	+	if err != nil {
	45	+	return ports.BrokerDaemonObservation{}, err
	46	+	}
	47	+	defer physical.Close()
	48	+	if physical.Identity() != endpoint.Identity \|\| physical.Policy() != endpoint.Policy {
	49	+	return ports.BrokerDaemonObservation{}, errors.New("vev: remote probe authenticated binding mismatch")
	50	+	}
	51	+	logical, err := physical.OpenStream(ctx, request)
	52	+	if err != nil {
	53	+	return ports.BrokerDaemonObservation{}, err
	54	+	}
	55	+	defer logical.Close()
	56	+	commandID, err := randomNonzeroUint64()
	57	+	if err != nil {
	58	+	return ports.BrokerDaemonObservation{}, err
	59	+	}
	60	+	if err := logical.SendClient(protocol.CommandRequest{RequestID: commandID, Version: protocol.Version, Slug: "remote-catalog", JSON: true}); err != nil {
	61	+	return ports.BrokerDaemonObservation{}, err
	62	+	}
	63	+	reply, err := logical.ReceiveServer()
	64	+	if err != nil {
	65	+	return ports.BrokerDaemonObservation{}, err
	66	+	}
	67	+	result, ok := reply.(protocol.CommandResult)
	68	+	if !ok \|\| result.RequestID != commandID \|\| !result.Valid() {
	69	+	return ports.BrokerDaemonObservation{}, errors.New("vev: invalid remote observation result")
	70	+	}
	71	+	if result.Outcome != protocol.CommandSucceeded {
	72	+	return ports.BrokerDaemonObservation{}, fmt.Errorf("vev: remote observation failed: %s", result.Text)
	73	+	}
	74	+	var catalog catalogue.RemoteCatalog
	75	+	if err := json.Unmarshal([]byte(result.Output), &catalog); err != nil {
	76	+	return ports.BrokerDaemonObservation{}, fmt.Errorf("vev: decode remote observation: %w", err)
	77	+	}
	78	+	if err := catalogue.ValidateRemoteCatalog(catalog); err != nil {
	79	+	return ports.BrokerDaemonObservation{}, err
	80	+	}
	81	+	return ports.BrokerDaemonObservation{
	82	+	Identity: physical.Identity(),
	83	+	Incarnation: physical.Incarnation(),
	84	+	ProtocolVersion: catalog.ProtocolVersion,
	85	+	Capabilities: 0,
	86	+	Availability: domain.RemoteAvailabilityReachable,
	87	+	InventoryKnown: true,
	88	+	Sessions: catalog.Sessions,
	89	+	}, nil
	90	+	}
	91	+
	92	+	func (p *brokerRemoteProbe) request(registration domain.RemoteRegistration) (ports.BrokerOpenStreamRequest, error) {
	93	+	if p.policy == nil {
	94	+	return ports.BrokerOpenStreamRequest{}, errors.New("vev: remote probe has no policy authority")
	95	+	}
	96	+	policy, ok := p.policy(registration.Endpoint)
	97	+	if !ok {
	98	+	return ports.BrokerOpenStreamRequest{}, errors.New("vev: remote probe endpoint is not configured")
	99	+	}
	100	+	connection, err := randomConnectionID()
	101	+	if err != nil {
	102	+	return ports.BrokerOpenStreamRequest{}, err
	103	+	}
	104	+	stream, err := randomNonzeroUint64()
	105	+	if err != nil {
	106	+	return ports.BrokerOpenStreamRequest{}, err
	107	+	}
	108	+	return ports.BrokerOpenStreamRequest{
	109	+	Epoch: p.epoch, Purpose: ports.BrokerStreamObservation,
	110	+	Connection: connection, Stream: ports.BrokerStreamID(stream),
	111	+	Endpoint: registration.Endpoint, Registration: registration, Policy: policy,
	112	+	}, nil
	113	+	}
	114	+
	115	+	func randomConnectionID() (ports.BrokerConnectionID, error) {
	116	+	var id ports.BrokerConnectionID
	117	+	if _, err := rand.Read(id[:]); err != nil {
	118	+	return id, err
	119	+	}
	120	+	if err := id.Validate(); err != nil {
	121	+	return randomConnectionID()
	122	+	}
	123	+	return id, nil
	124	+	}
	125	+
	126	+	func randomNonzeroUint64() (uint64, error) {
	127	+	var raw [8]byte
	128	+	if _, err := rand.Read(raw[:]); err != nil {
	129	+	return 0, err
	130	+	}
	131	+	value := binary.BigEndian.Uint64(raw[:])
	132	+	if value == 0 {
	133	+	return randomNonzeroUint64()
	134	+	}
	135	+	return value, nil
	136	+	}

A internal/app/broker_remote_probe_test.go +37 -0

@@ -0,0 +1,37 @@
	1	+	package app
	2	+
	3	+	import (
	4	+	"context"
	5	+	"testing"
	6	+
	7	+	"github.com/bnema/vev/internal/domain"
	8	+	"github.com/bnema/vev/internal/ports"
	9	+	"github.com/bnema/vev/internal/protocol"
	10	+	"github.com/stretchr/testify/require"
	11	+	)
	12	+
	13	+	func TestBrokerRemoteProbeRequestIsObservationOnly(t *testing.T) {
	14	+	registration := domain.RemoteRegistration{Endpoint: "user@example.test", Incarnation: [16]byte{1}, Generation: 1}
	15	+	policy := ports.BrokerPolicy{ProtocolVersion: protocol.Version, CatalogSchemaVersion: 3, Transport: "stdio", Trust: "ssh", Launch: "explicit", Isolation: "user", EnvironmentPolicy: protocol.EnvironmentPolicyDaemonOwned}
	16	+	probe := &brokerRemoteProbe{
	17	+	epoch: 7,
	18	+	policy: func(endpoint string) (ports.BrokerPolicy, bool) {
	19	+	return policy, endpoint == registration.Endpoint
	20	+	},
	21	+	}
	22	+
	23	+	request, err := probe.request(registration)
	24	+	require.NoError(t, err)
	25	+	require.NoError(t, request.Validate())
	26	+	require.Equal(t, ports.BrokerStreamObservation, request.Purpose)
	27	+	require.Zero(t, request.Admission)
	28	+	require.Empty(t, request.Env)
	29	+	require.Equal(t, registration, request.Registration)
	30	+	require.Equal(t, policy, request.Policy)
	31	+	}
	32	+
	33	+	func TestBrokerRemoteProbeRejectsUnconfiguredEndpointBeforeDial(t *testing.T) {
	34	+	probe := &brokerRemoteProbe{epoch: 7, policy: func(string) (ports.BrokerPolicy, bool) { return ports.BrokerPolicy{}, false }}
	35	+	_, err := probe.Probe(context.Background(), domain.RemoteRegistration{Endpoint: "user@example.test", Incarnation: [16]byte{1}, Generation: 1})
	36	+	require.ErrorContains(t, err, "incomplete broker remote probe")
	37	+	}

@@ -0,0 +1,37 @@
	1	+	package app
	2	+
	3	+	import (
	4	+	"context"
	5	+	"testing"
	6	+
	7	+	"github.com/bnema/vev/internal/domain"
	8	+	"github.com/bnema/vev/internal/ports"
	9	+	"github.com/bnema/vev/internal/protocol"
	10	+	"github.com/stretchr/testify/require"
	11	+	)
	12	+
	13	+	func TestBrokerRemoteProbeRequestIsObservationOnly(t *testing.T) {
	14	+	registration := domain.RemoteRegistration{Endpoint: "user@example.test", Incarnation: [16]byte{1}, Generation: 1}
	15	+	policy := ports.BrokerPolicy{ProtocolVersion: protocol.Version, CatalogSchemaVersion: 3, Transport: "stdio", Trust: "ssh", Launch: "explicit", Isolation: "user", EnvironmentPolicy: protocol.EnvironmentPolicyDaemonOwned}
	16	+	probe := &brokerRemoteProbe{
	17	+	epoch: 7,
	18	+	policy: func(endpoint string) (ports.BrokerPolicy, bool) {
	19	+	return policy, endpoint == registration.Endpoint
	20	+	},
	21	+	}
	22	+
	23	+	request, err := probe.request(registration)
	24	+	require.NoError(t, err)
	25	+	require.NoError(t, request.Validate())
	26	+	require.Equal(t, ports.BrokerStreamObservation, request.Purpose)
	27	+	require.Zero(t, request.Admission)
	28	+	require.Empty(t, request.Env)
	29	+	require.Equal(t, registration, request.Registration)
	30	+	require.Equal(t, policy, request.Policy)
	31	+	}
	32	+
	33	+	func TestBrokerRemoteProbeRejectsUnconfiguredEndpointBeforeDial(t *testing.T) {
	34	+	probe := &brokerRemoteProbe{epoch: 7, policy: func(string) (ports.BrokerPolicy, bool) { return ports.BrokerPolicy{}, false }}
	35	+	_, err := probe.Probe(context.Background(), domain.RemoteRegistration{Endpoint: "user@example.test", Incarnation: [16]byte{1}, Generation: 1})
	36	+	require.ErrorContains(t, err, "incomplete broker remote probe")
	37	+	}