detached 1d7fc8e

fix(broker): give preview streams their own connection identity

commit
1d7fc8e4782336aa4713efdddbb4decc2f15e998
author
brice · · 2026-09-23 08:19 UTC
parent
d6a1b89
signature
GPG VerifiedCD88F9E564E0102646FB6077A1A07C7C7C3E349D · arch-gpg Verified for the committer email
4 files · +63 -11
Unified
UnifiedSplit
M internal/usecase/broker/service.go +14 -4
@@ -89,12 +89,19 @@ func (a *Authority) AdmitClient(ctx context.Context) (ports.BrokerService, error
8989 lease.Release()
9090 return nil, err
9191 }
92+ previewID, err := a.pool.RegisterClient()
93+ if err != nil {
94+ a.pool.CloseClient(id)
95+ lease.Release()
96+ return nil, err
97+ }
9298 if err := ctx.Err(); err != nil {
99+ a.pool.CloseClient(previewID)
93100 a.pool.CloseClient(id)
94101 lease.Release()
95102 return nil, err
96103 }
97- return newService(a.epoch, id, a.registry, a.pool, a.supervisor, lease, a.clock), nil
104+ return newService(a.epoch, id, previewID, a.registry, a.pool, a.supervisor, lease, a.clock), nil
98105 }
99106
100107 // Service is one admitted client connection: the core's service for exactly one
@@ -102,6 +109,7 @@ func (a *Authority) AdmitClient(ctx context.Context) (ports.BrokerService, error
102109 type Service struct {
103110 epoch ports.BrokerEpoch
104111 id ports.BrokerConnectionID
112+ previewID ports.BrokerConnectionID
105113 registry *Registry
106114 pool *Pool
107115 supervisor *Supervisor
@@ -124,7 +132,8 @@ type Service struct {
124132 // supervisor and the broker operations each read one identity here and carry
125133 // it on OpenStream rather than relying on an implicit allocation there. An
126134 // allocated identity is consumed even when the open it names is refused.
127- nextStream ports.BrokerStreamID
135+ nextStream ports.BrokerStreamID
136+ previewStream ports.BrokerStreamID
128137 // wg counts in-flight stream opens and membership mutations so Close drains
129138 // them, and their operation leases, before releasing connection resources.
130139 wg sync.WaitGroup
@@ -144,10 +153,10 @@ type Service struct {
144153
145154 var _ ports.BrokerService = (*Service)(nil)
146155
147-func newService(epoch ports.BrokerEpoch, id ports.BrokerConnectionID, registry *Registry, pool *Pool, supervisor *Supervisor, lease *Lease, clock ports.Clock) *Service {
156+func newService(epoch ports.BrokerEpoch, id, previewID ports.BrokerConnectionID, registry *Registry, pool *Pool, supervisor *Supervisor, lease *Lease, clock ports.Clock) *Service {
148157 ctx, cancel := context.WithCancel(supervisor.RootContext())
149158 s := &Service{
150- epoch: epoch, id: id, registry: registry, pool: pool, supervisor: supervisor, lease: lease, clock: clock,
159+ epoch: epoch, id: id, previewID: previewID, registry: registry, pool: pool, supervisor: supervisor, lease: lease, clock: clock,
151160 ctx: ctx, cancel: cancel, subs: make(map[*serviceSubscription]struct{}),
152161 done: make(chan struct{}),
153162 }
@@ -410,6 +419,7 @@ func (s *Service) Close() error {
410419 for _, sub := range subs {
411420 sub.Close()
412421 }
422+ s.pool.CloseClient(s.previewID)
413423 s.pool.CloseClient(s.id)
414424 s.lease.Release()
415425 // The connection is locally and orderly closed: detach the broker-root
@@ -89,12 +89,19 @@ func (a *Authority) AdmitClient(ctx context.Context) (ports.BrokerService, error
89 lease.Release()89 lease.Release()
90 return nil, err90 return nil, err
91 }91 }
92+ previewID, err := a.pool.RegisterClient()
93+ if err != nil {
94+ a.pool.CloseClient(id)
95+ lease.Release()
96+ return nil, err
97+ }
92 if err := ctx.Err(); err != nil {98 if err := ctx.Err(); err != nil {
99+ a.pool.CloseClient(previewID)
93 a.pool.CloseClient(id)100 a.pool.CloseClient(id)
94 lease.Release()101 lease.Release()
95 return nil, err102 return nil, err
96 }103 }
97- return newService(a.epoch, id, a.registry, a.pool, a.supervisor, lease, a.clock), nil104+ return newService(a.epoch, id, previewID, a.registry, a.pool, a.supervisor, lease, a.clock), nil
98 }105 }
99 106
100 // Service is one admitted client connection: the core's service for exactly one107 // Service is one admitted client connection: the core's service for exactly one
@@ -102,6 +109,7 @@ func (a *Authority) AdmitClient(ctx context.Context) (ports.BrokerService, error
102 type Service struct {109 type Service struct {
103 epoch ports.BrokerEpoch110 epoch ports.BrokerEpoch
104 id ports.BrokerConnectionID111 id ports.BrokerConnectionID
112+ previewID ports.BrokerConnectionID
105 registry *Registry113 registry *Registry
106 pool *Pool114 pool *Pool
107 supervisor *Supervisor115 supervisor *Supervisor
@@ -124,7 +132,8 @@ type Service struct {
124 // supervisor and the broker operations each read one identity here and carry132 // supervisor and the broker operations each read one identity here and carry
125 // it on OpenStream rather than relying on an implicit allocation there. An133 // it on OpenStream rather than relying on an implicit allocation there. An
126 // allocated identity is consumed even when the open it names is refused.134 // allocated identity is consumed even when the open it names is refused.
127- nextStream ports.BrokerStreamID135+ nextStream ports.BrokerStreamID
136+ previewStream ports.BrokerStreamID
128 // wg counts in-flight stream opens and membership mutations so Close drains137 // wg counts in-flight stream opens and membership mutations so Close drains
129 // them, and their operation leases, before releasing connection resources.138 // them, and their operation leases, before releasing connection resources.
130 wg sync.WaitGroup139 wg sync.WaitGroup
@@ -144,10 +153,10 @@ type Service struct {
144 153
145 var _ ports.BrokerService = (*Service)(nil)154 var _ ports.BrokerService = (*Service)(nil)
146 155
147-func newService(epoch ports.BrokerEpoch, id ports.BrokerConnectionID, registry *Registry, pool *Pool, supervisor *Supervisor, lease *Lease, clock ports.Clock) *Service {156+func newService(epoch ports.BrokerEpoch, id, previewID ports.BrokerConnectionID, registry *Registry, pool *Pool, supervisor *Supervisor, lease *Lease, clock ports.Clock) *Service {
148 ctx, cancel := context.WithCancel(supervisor.RootContext())157 ctx, cancel := context.WithCancel(supervisor.RootContext())
149 s := &Service{158 s := &Service{
150- epoch: epoch, id: id, registry: registry, pool: pool, supervisor: supervisor, lease: lease, clock: clock,159+ epoch: epoch, id: id, previewID: previewID, registry: registry, pool: pool, supervisor: supervisor, lease: lease, clock: clock,
151 ctx: ctx, cancel: cancel, subs: make(map[*serviceSubscription]struct{}),160 ctx: ctx, cancel: cancel, subs: make(map[*serviceSubscription]struct{}),
152 done: make(chan struct{}),161 done: make(chan struct{}),
153 }162 }
@@ -410,6 +419,7 @@ func (s *Service) Close() error {
410 for _, sub := range subs {419 for _, sub := range subs {
411 sub.Close()420 sub.Close()
412 }421 }
422+ s.pool.CloseClient(s.previewID)
413 s.pool.CloseClient(s.id)423 s.pool.CloseClient(s.id)
414 s.lease.Release()424 s.lease.Release()
415 // The connection is locally and orderly closed: detach the broker-root425 // The connection is locally and orderly closed: detach the broker-root
M internal/usecase/broker/service_preview.go +10 -5
@@ -93,12 +93,17 @@ func previewPublication(request ports.BrokerPreviewRequest, preview protocol.Rem
9393 // previewStreamRequest builds the observation stream for one watch. The broker
9494 // allocates the stream identity, so a client never names one.
9595 func (s *Service) previewStreamRequest(route ports.BrokerPreviewRoute) (ports.BrokerOpenStreamRequest, error) {
96- stream, err := s.NextStreamID()
97- if err != nil {
98- return ports.BrokerOpenStreamRequest{}, err
96+ s.mu.Lock()
97+ defer s.mu.Unlock()
98+ if s.closed {
99+ return ports.BrokerOpenStreamRequest{}, ports.BrokerAdmissionClosed
100+ }
101+ if s.previewStream == ^ports.BrokerStreamID(0) {
102+ return ports.BrokerOpenStreamRequest{}, ports.BrokerAdmissionLimit
99103 }
104+ s.previewStream++
100105 return ports.BrokerOpenStreamRequest{
101- Epoch: s.epoch, Connection: s.id, Stream: stream,
106+ Epoch: s.epoch, Connection: s.previewID, Stream: s.previewStream,
102107 Purpose: ports.BrokerStreamObservation, StartMode: ports.BrokerDaemonExistingOnly,
103108 Local: route.Local, Endpoint: route.Endpoint, Registration: route.Registration, Policy: route.Policy,
104109 }, nil
@@ -196,7 +201,7 @@ func (p *servicePreviewSubscription) watch() (reopen, delivered bool) {
196201 route, err := p.service.previewStreamRequest(p.request.Route)
197202 var stream ports.BrokerLogicalConnection
198203 if err == nil {
199- stream, err = p.service.OpenStream(p.ctx, route)
204+ stream, err = p.service.pool.OpenStream(p.ctx, route)
200205 }
201206 if err != nil {
202207 return p.fail(err), false
@@ -93,12 +93,17 @@ func previewPublication(request ports.BrokerPreviewRequest, preview protocol.Rem
93 // previewStreamRequest builds the observation stream for one watch. The broker93 // previewStreamRequest builds the observation stream for one watch. The broker
94 // allocates the stream identity, so a client never names one.94 // allocates the stream identity, so a client never names one.
95 func (s *Service) previewStreamRequest(route ports.BrokerPreviewRoute) (ports.BrokerOpenStreamRequest, error) {95 func (s *Service) previewStreamRequest(route ports.BrokerPreviewRoute) (ports.BrokerOpenStreamRequest, error) {
96- stream, err := s.NextStreamID()96+ s.mu.Lock()
97- if err != nil {97+ defer s.mu.Unlock()
98- return ports.BrokerOpenStreamRequest{}, err98+ if s.closed {
99+ return ports.BrokerOpenStreamRequest{}, ports.BrokerAdmissionClosed
100+ }
101+ if s.previewStream == ^ports.BrokerStreamID(0) {
102+ return ports.BrokerOpenStreamRequest{}, ports.BrokerAdmissionLimit
99 }103 }
104+ s.previewStream++
100 return ports.BrokerOpenStreamRequest{105 return ports.BrokerOpenStreamRequest{
101- Epoch: s.epoch, Connection: s.id, Stream: stream,106+ Epoch: s.epoch, Connection: s.previewID, Stream: s.previewStream,
102 Purpose: ports.BrokerStreamObservation, StartMode: ports.BrokerDaemonExistingOnly,107 Purpose: ports.BrokerStreamObservation, StartMode: ports.BrokerDaemonExistingOnly,
103 Local: route.Local, Endpoint: route.Endpoint, Registration: route.Registration, Policy: route.Policy,108 Local: route.Local, Endpoint: route.Endpoint, Registration: route.Registration, Policy: route.Policy,
104 }, nil109 }, nil
@@ -196,7 +201,7 @@ func (p *servicePreviewSubscription) watch() (reopen, delivered bool) {
196 route, err := p.service.previewStreamRequest(p.request.Route)201 route, err := p.service.previewStreamRequest(p.request.Route)
197 var stream ports.BrokerLogicalConnection202 var stream ports.BrokerLogicalConnection
198 if err == nil {203 if err == nil {
199- stream, err = p.service.OpenStream(p.ctx, route)204+ stream, err = p.service.pool.OpenStream(p.ctx, route)
200 }205 }
201 if err != nil {206 if err != nil {
202 return p.fail(err), false207 return p.fail(err), false
M internal/usecase/broker/service_preview_test.go +37 -0
@@ -183,6 +183,43 @@ func awaitClosed(t *testing.T, conn *previewWatchConn) {
183183 }
184184 }
185185
186+func TestServicePreviewStreamIDsDoNotConsumeClientIDs(t *testing.T) {
187+ for _, previewFirst := range []bool{false, true} {
188+ name := "client first"
189+ if previewFirst {
190+ name = "preview first"
191+ }
192+ t.Run(name, func(t *testing.T) {
193+ f := newPreviewFixture(t, previewTarget(0x11, "tab-alpha"))
194+ service := f.service.(*Service)
195+ openClient := func(want ports.BrokerStreamID) {
196+ t.Helper()
197+ id, err := service.NextStreamID()
198+ require.NoError(t, err)
199+ require.Equal(t, want, id)
200+ stream, err := service.OpenStream(context.Background(), poolRequest(service.ConnectionID(), uint64(id)))
201+ require.NoError(t, err)
202+ t.Cleanup(func() { require.NoError(t, stream.Close()) })
203+ }
204+ if !previewFirst {
205+ openClient(1)
206+ }
207+ for range 3 {
208+ req, err := service.previewStreamRequest(ports.BrokerPreviewRoute{Local: true, Policy: poolPolicy()})
209+ require.NoError(t, err)
210+ stream, err := service.pool.OpenStream(context.Background(), req)
211+ require.NoError(t, err)
212+ require.NoError(t, stream.Close())
213+ }
214+ if previewFirst {
215+ openClient(1)
216+ } else {
217+ openClient(2)
218+ }
219+ })
220+ }
221+}
222+
186223 func TestServicePreviewKeepsOneStreamAcrossFrames(t *testing.T) {
187224 target := previewTarget(0x11, "tab-alpha")
188225 f := newPreviewFixture(t, target)
@@ -183,6 +183,43 @@ func awaitClosed(t *testing.T, conn *previewWatchConn) {
183 }183 }
184 }184 }
185 185
186+func TestServicePreviewStreamIDsDoNotConsumeClientIDs(t *testing.T) {
187+ for _, previewFirst := range []bool{false, true} {
188+ name := "client first"
189+ if previewFirst {
190+ name = "preview first"
191+ }
192+ t.Run(name, func(t *testing.T) {
193+ f := newPreviewFixture(t, previewTarget(0x11, "tab-alpha"))
194+ service := f.service.(*Service)
195+ openClient := func(want ports.BrokerStreamID) {
196+ t.Helper()
197+ id, err := service.NextStreamID()
198+ require.NoError(t, err)
199+ require.Equal(t, want, id)
200+ stream, err := service.OpenStream(context.Background(), poolRequest(service.ConnectionID(), uint64(id)))
201+ require.NoError(t, err)
202+ t.Cleanup(func() { require.NoError(t, stream.Close()) })
203+ }
204+ if !previewFirst {
205+ openClient(1)
206+ }
207+ for range 3 {
208+ req, err := service.previewStreamRequest(ports.BrokerPreviewRoute{Local: true, Policy: poolPolicy()})
209+ require.NoError(t, err)
210+ stream, err := service.pool.OpenStream(context.Background(), req)
211+ require.NoError(t, err)
212+ require.NoError(t, stream.Close())
213+ }
214+ if previewFirst {
215+ openClient(1)
216+ } else {
217+ openClient(2)
218+ }
219+ })
220+ }
221+}
222+
186 func TestServicePreviewKeepsOneStreamAcrossFrames(t *testing.T) {223 func TestServicePreviewKeepsOneStreamAcrossFrames(t *testing.T) {
187 target := previewTarget(0x11, "tab-alpha")224 target := previewTarget(0x11, "tab-alpha")
188 f := newPreviewFixture(t, target)225 f := newPreviewFixture(t, target)
M internal/usecase/broker/service_test.go +2 -2
@@ -154,12 +154,12 @@ func TestAuthorityAdmissionRollback(t *testing.T) {
154154
155155 // Exhaust the pool's connection bound: the second admission must release the
156156 // supervisor client lease it already took.
157- pool.limits.Clients = 1
157+ pool.limits.Clients = 2
158158 _, err = authority.AdmitClient(context.Background())
159159 require.ErrorIs(t, err, ports.BrokerAdmissionLimit)
160160 clients, _, _, _ := supervisorState(supervisor)
161161 require.Equal(t, 1, clients, "a refused admission must release its client lease")
162- require.Equal(t, 1, poolClientCount(pool), "a refused admission must not leave a pool client")
162+ require.Equal(t, 2, poolClientCount(pool), "a refused admission must not leave a pool client")
163163
164164 // A cancelled admission context takes no lease and registers no client.
165165 ctx, cancel := context.WithCancel(context.Background())
@@ -154,12 +154,12 @@ func TestAuthorityAdmissionRollback(t *testing.T) {
154 154
155 // Exhaust the pool's connection bound: the second admission must release the155 // Exhaust the pool's connection bound: the second admission must release the
156 // supervisor client lease it already took.156 // supervisor client lease it already took.
157- pool.limits.Clients = 1157+ pool.limits.Clients = 2
158 _, err = authority.AdmitClient(context.Background())158 _, err = authority.AdmitClient(context.Background())
159 require.ErrorIs(t, err, ports.BrokerAdmissionLimit)159 require.ErrorIs(t, err, ports.BrokerAdmissionLimit)
160 clients, _, _, _ := supervisorState(supervisor)160 clients, _, _, _ := supervisorState(supervisor)
161 require.Equal(t, 1, clients, "a refused admission must release its client lease")161 require.Equal(t, 1, clients, "a refused admission must release its client lease")
162- require.Equal(t, 1, poolClientCount(pool), "a refused admission must not leave a pool client")162+ require.Equal(t, 2, poolClientCount(pool), "a refused admission must not leave a pool client")
163 163
164 // A cancelled admission context takes no lease and registers no client.164 // A cancelled admission context takes no lease and registers no client.
165 ctx, cancel := context.WithCancel(context.Background())165 ctx, cancel := context.WithCancel(context.Background())

Keyboard shortcuts

Single-key shortcuts are on. ctrl k and esc work whatever this says.

ctrl k :
go to a section, branch, tag, file or repository
?
this list
j k
move the row cursor down / up
enter
open the row under the cursor; on Files, a directory opens or closes in place
↑ ↓ ← → home end
in the file tree (focus on a row): previous / next row, close / open a directory or go to its parent / first entry, first / last row. They never open a file
l h
on Files: open a directory or a file / close a directory or go up to its parent, then to the parent page. Elsewhere: into the side panel (on a merge request, its files) / back to the content; there j k move between its controls
g g G
first / last row
/
filter the list; on Files, find a file anywhere in the repository (esc clears, enter opens the best match)
b
switch branch or tag
g s g f g c g b g m g i g r g a
tabs: summary, files, commits, branches & tags, merge requests, issues, releases, actions (when available)
g h
home
w
wide mode: use the whole width of the screen, or not (remembered)
y
copy a permalink pinned to the commit
d u x
diff: split / unified / collapse or expand all
e E
diff: the whole file in view / every file, with the lines the hunks hide
n
merge request: next note in the diff
esc
close a dialog, leave a field

Branches 12

Tags 11

↑↓ moveenter openesc close