refactor(broker): build services from a validated config

commit: d98363fe3845b4920466d3c6b2e26b52e7ff4dba
author: brice · 29d ago · 2026-09-23 14:43 UTC
committer: brice · 29d ago · 2026-09-23 14:49 UTC
parent: 0cfd75a
signature: GPG VerifiedCD88F9E564E0102646FB6077A1A07C7C7C3E349D · arch-gpg Verified for the committer email

2 files · +55 -5

M internal/usecase/broker/service.go +29 -5


@@ -101,7 +101,10 @@ func (a *Authority) AdmitClient(ctx context.Context) (ports.BrokerService, error
101	101		lease.Release()
102	102		return nil, err
103	103		}
104		-	return newService(a.epoch, id, previewID, a.registry, a.pool, a.supervisor, lease, a.clock), nil
	104	+	return newService(serviceConfig{
	105	+	epoch: a.epoch, id: id, previewID: previewID, registry: a.registry,
	106	+	pool: a.pool, supervisor: a.supervisor, lease: lease, clock: a.clock,
	107	+	}), nil
105	108		}
106	109
107	110		// Service is one admitted client connection: the core's service for exactly one

@@ -153,10 +156,31 @@ type Service struct {
153	156
154	157		var _ ports.BrokerService = (*Service)(nil)
155	158
156		-	func newService(epoch ports.BrokerEpoch, id, previewID ports.BrokerConnectionID, registry Registry, pool Pool, supervisor Supervisor, lease Lease, clock ports.Clock) *Service {
157		-	ctx, cancel := context.WithCancel(supervisor.RootContext())
	159	+	type serviceConfig struct {
	160	+	epoch ports.BrokerEpoch
	161	+	id, previewID ports.BrokerConnectionID
	162	+	registry *Registry
	163	+	pool *Pool
	164	+	supervisor *Supervisor
	165	+	lease *Lease
	166	+	clock ports.Clock
	167	+	}
	168	+
	169	+	func (c serviceConfig) validate() error {
	170	+	if c.epoch == 0 \|\| c.id == (ports.BrokerConnectionID{}) \|\| c.previewID == (ports.BrokerConnectionID{}) \|\| c.id == c.previewID \|\|
	171	+	nilDependency(c.registry) \|\| nilDependency(c.pool) \|\| nilDependency(c.supervisor) \|\| nilDependency(c.lease) \|\| nilDependency(c.clock) {
	172	+	return errors.New("broker: invalid service dependencies")
	173	+	}
	174	+	return nil
	175	+	}
	176	+
	177	+	func newService(c serviceConfig) *Service {
	178	+	if err := c.validate(); err != nil {
	179	+	panic(err) // Only the admitted authority constructs a service after acquiring both IDs and lease.
	180	+	}
	181	+	ctx, cancel := context.WithCancel(c.supervisor.RootContext())
158	182		s := &Service{
159		-	epoch: epoch, id: id, previewID: previewID, registry: registry, pool: pool, supervisor: supervisor, lease: lease, clock: clock,
	183	+	epoch: c.epoch, id: c.id, previewID: c.previewID, registry: c.registry, pool: c.pool, supervisor: c.supervisor, lease: c.lease, clock: c.clock,
160	184		ctx: ctx, cancel: cancel, subs: make(map[*serviceSubscription]struct{}),
161	185		done: make(chan struct{}),
162	186		}

@@ -164,7 +188,7 @@ func newService(epoch ports.BrokerEpoch, id, previewID ports.BrokerConnectionID,
164	188		// loss as the terminal cause unless a local Close has already settled the
165	189		// connection, so a connection closed by its owner reports nil and one lost
166	190		// to a broker shutdown reports the typed loss.
167		-	s.stopRoot = context.AfterFunc(supervisor.RootContext(), func() {
	191	+	s.stopRoot = context.AfterFunc(c.supervisor.RootContext(), func() {
168	192		s.terminalize(ports.BrokerError{Code: ports.BrokerErrorUnavailable, Text: "broker shutdown", Cause: context.Canceled})
169	193		})
170	194		return s

 		lease.Release()
 		return nil, err
 	}
-	return newService(a.epoch, id, previewID, a.registry, a.pool, a.supervisor, lease, a.clock), nil
+	return newService(serviceConfig{
+		epoch: a.epoch, id: id, previewID: previewID, registry: a.registry,
+		pool: a.pool, supervisor: a.supervisor, lease: lease, clock: a.clock,
+	}), nil
 }
 // Service is one admitted client connection: the core's service for exactly one
 var _ ports.BrokerService = (*Service)(nil)
-func newService(epoch ports.BrokerEpoch, id, previewID ports.BrokerConnectionID, registry *Registry, pool *Pool, supervisor *Supervisor, lease *Lease, clock ports.Clock) *Service {
+type serviceConfig struct {
-	ctx, cancel := context.WithCancel(supervisor.RootContext())
+	epoch         ports.BrokerEpoch
+	id, previewID ports.BrokerConnectionID
+	registry      *Registry
+	pool          *Pool
+	supervisor    *Supervisor
+	lease         *Lease
+	clock         ports.Clock
+}
+func (c serviceConfig) validate() error {
+	if c.epoch == 0 || c.id == (ports.BrokerConnectionID{}) || c.previewID == (ports.BrokerConnectionID{}) || c.id == c.previewID ||
+		nilDependency(c.registry) || nilDependency(c.pool) || nilDependency(c.supervisor) || nilDependency(c.lease) || nilDependency(c.clock) {
+		return errors.New("broker: invalid service dependencies")
+	}
+	return nil
+}
+func newService(c serviceConfig) *Service {
+	if err := c.validate(); err != nil {
+		panic(err) // Only the admitted authority constructs a service after acquiring both IDs and lease.
+	}
+	ctx, cancel := context.WithCancel(c.supervisor.RootContext())
 	s := &Service{
-		epoch: epoch, id: id, previewID: previewID, registry: registry, pool: pool, supervisor: supervisor, lease: lease, clock: clock,
+		epoch: c.epoch, id: c.id, previewID: c.previewID, registry: c.registry, pool: c.pool, supervisor: c.supervisor, lease: c.lease, clock: c.clock,
 		ctx: ctx, cancel: cancel, subs: make(map[*serviceSubscription]struct{}),
 		done: make(chan struct{}),
 	}
 	// loss as the terminal cause unless a local Close has already settled the
 	// connection, so a connection closed by its owner reports nil and one lost
 	// to a broker shutdown reports the typed loss.
-	s.stopRoot = context.AfterFunc(supervisor.RootContext(), func() {
+	s.stopRoot = context.AfterFunc(c.supervisor.RootContext(), func() {
 		s.terminalize(ports.BrokerError{Code: ports.BrokerErrorUnavailable, Text: "broker shutdown", Cause: context.Canceled})
 	})
 	return s

M internal/usecase/broker/service_test.go +26 -0


@@ -1154,3 +1154,29 @@ func TestRegistryObservationDisabledDrainsWriter(t *testing.T) {
1154	1154		require.NotEmpty(t, store.writtenRevisions())
1155	1155		require.Zero(t, clock.activeTimers())
1156	1156		}
	1157	+
	1158	+	func TestServiceConfigValidation(t *testing.T) {
	1159	+	valid := serviceConfig{epoch: 1, id: ports.BrokerConnectionID{1}, previewID: ports.BrokerConnectionID{2}, registry: &Registry{}, pool: &Pool{}, supervisor: &Supervisor{}, lease: &Lease{}, clock: newManualClock(time.Unix(0, 0))}
	1160	+	tests := []struct {
	1161	+	name string
	1162	+	mutate func(*serviceConfig)
	1163	+	}{
	1164	+	{"zero epoch", func(c *serviceConfig) { c.epoch = 0 }},
	1165	+	{"zero id", func(c *serviceConfig) { c.id = ports.BrokerConnectionID{} }},
	1166	+	{"zero preview", func(c *serviceConfig) { c.previewID = ports.BrokerConnectionID{} }},
	1167	+	{"same ids", func(c *serviceConfig) { c.previewID = c.id }},
	1168	+	{"nil registry", func(c *serviceConfig) { c.registry = nil }},
	1169	+	{"nil pool", func(c *serviceConfig) { c.pool = nil }},
	1170	+	{"nil supervisor", func(c *serviceConfig) { c.supervisor = nil }},
	1171	+	{"nil lease", func(c *serviceConfig) { c.lease = nil }},
	1172	+	{"nil clock", func(c *serviceConfig) { c.clock = nil }},
	1173	+	}
	1174	+	require.NoError(t, valid.validate())
	1175	+	for _, tt := range tests {
	1176	+	t.Run(tt.name, func(t *testing.T) {
	1177	+	config := valid
	1178	+	tt.mutate(&config)
	1179	+	require.Error(t, config.validate())
	1180	+	})
	1181	+	}
	1182	+	}


@@ -1154,3 +1154,29 @@ func TestRegistryObservationDisabledDrainsWriter(t *testing.T) {
1154	require.NotEmpty(t, store.writtenRevisions())	1154		require.NotEmpty(t, store.writtenRevisions())
1155	require.Zero(t, clock.activeTimers())	1155		require.Zero(t, clock.activeTimers())
1156	}	1156		}
		1157	+
		1158	+	func TestServiceConfigValidation(t *testing.T) {
		1159	+	valid := serviceConfig{epoch: 1, id: ports.BrokerConnectionID{1}, previewID: ports.BrokerConnectionID{2}, registry: &Registry{}, pool: &Pool{}, supervisor: &Supervisor{}, lease: &Lease{}, clock: newManualClock(time.Unix(0, 0))}
		1160	+	tests := []struct {
		1161	+	name string
		1162	+	mutate func(*serviceConfig)
		1163	+	}{
		1164	+	{"zero epoch", func(c *serviceConfig) { c.epoch = 0 }},
		1165	+	{"zero id", func(c *serviceConfig) { c.id = ports.BrokerConnectionID{} }},
		1166	+	{"zero preview", func(c *serviceConfig) { c.previewID = ports.BrokerConnectionID{} }},
		1167	+	{"same ids", func(c *serviceConfig) { c.previewID = c.id }},
		1168	+	{"nil registry", func(c *serviceConfig) { c.registry = nil }},
		1169	+	{"nil pool", func(c *serviceConfig) { c.pool = nil }},
		1170	+	{"nil supervisor", func(c *serviceConfig) { c.supervisor = nil }},
		1171	+	{"nil lease", func(c *serviceConfig) { c.lease = nil }},
		1172	+	{"nil clock", func(c *serviceConfig) { c.clock = nil }},
		1173	+	}
		1174	+	require.NoError(t, valid.validate())
		1175	+	for _, tt := range tests {
		1176	+	t.Run(tt.name, func(t *testing.T) {
		1177	+	config := valid
		1178	+	tt.mutate(&config)
		1179	+	require.Error(t, config.validate())
		1180	+	})
		1181	+	}
		1182	+	}