detached 1744d68

fix(app): deliver the broker mux QUIC proxy close before exiting

When the remote daemon was not running, the existing-only helper dial
failed and _broker-mux-quic-proxy exited before QUIC's deferred close
ran, so neither FIN nor CONNECTION_CLOSE reached the broker. Its probe
then sat in the daemonmux preamble read until the 60s QUIC idle timeout.
The proxy now waits for the bounded graceful teardown, as _quic-proxy
already does. Adds coverage that a silent daemon on a borrowed pooled
transport never triggers a redial.
commit
1744d6827ecb592daea2717a482a98410ac733fd
author
brice · · 2026-09-23 03:23 UTC
parent
3fbd5a1
signature
GPG VerifiedCD88F9E564E0102646FB6077A1A07C7C7C3E349D · arch-gpg Verified for the committer email
2 files · +103 -1
Unified
UnifiedSplit
M internal/app/broker_offline_mux.go +9 -1
@@ -622,7 +622,15 @@ func runBrokerMuxQUICProxyCommand(ctx context.Context, options brokerMuxOptions)
622622 if err != nil {
623623 return err
624624 }
625- defer func() { _ = transport.Close() }()
625+ // The proxy owns its process: QUIC's orderly Close only FINs the stream and
626+ // defers the connection close, so returning (and exiting) right away would
627+ // drop both and leave the broker's preamble read parked until the QUIC idle
628+ // timeout. Waiting for the bounded teardown delivers the FIN, so a refused
629+ // existing-only dial (no daemon running) fails the broker's attempt at once.
630+ defer func() {
631+ _ = transport.Close()
632+ _ = waitGracefulTeardown(transport)
633+ }()
626634 raw, err := dialBrokerMuxHelper(ctx, options)
627635 if err != nil {
628636 return fmt.Errorf("vev: broker mux quic proxy: dial daemonmux carriage: %w", err)
@@ -622,7 +622,15 @@ func runBrokerMuxQUICProxyCommand(ctx context.Context, options brokerMuxOptions)
622 if err != nil {622 if err != nil {
623 return err623 return err
624 }624 }
625- defer func() { _ = transport.Close() }()625+ // The proxy owns its process: QUIC's orderly Close only FINs the stream and
626+ // defers the connection close, so returning (and exiting) right away would
627+ // drop both and leave the broker's preamble read parked until the QUIC idle
628+ // timeout. Waiting for the bounded teardown delivers the FIN, so a refused
629+ // existing-only dial (no daemon running) fails the broker's attempt at once.
630+ defer func() {
631+ _ = transport.Close()
632+ _ = waitGracefulTeardown(transport)
633+ }()
626 raw, err := dialBrokerMuxHelper(ctx, options)634 raw, err := dialBrokerMuxHelper(ctx, options)
627 if err != nil {635 if err != nil {
628 return fmt.Errorf("vev: broker mux quic proxy: dial daemonmux carriage: %w", err)636 return fmt.Errorf("vev: broker mux quic proxy: dial daemonmux carriage: %w", err)
M internal/app/broker_remote_probe_shared_test.go +94 -0
@@ -3,7 +3,9 @@ package app
33 import (
44 "context"
55 "errors"
6+ "sync"
67 "testing"
8+ "time"
79
810 "github.com/bnema/vev/internal/adapters/clock"
911 "github.com/bnema/vev/internal/domain"
@@ -113,3 +115,95 @@ func TestBrokerRemoteProbeBorrowsWarmTransport(t *testing.T) {
113115 })
114116 }
115117 }
118+
119+// TestBrokerRemoteProbeSilentBorrowedTransportNeverRedials proves a probe on a
120+// healthy borrowed transport whose daemon never answers is bounded by the
121+// attempt context (the registry's ports.Clock deadline cancels it) and returns
122+// without bootstrapping a second transport: a live pooled physical is never
123+// abandoned for a redial just because one observation went unanswered.
124+func TestBrokerRemoteProbeSilentBorrowedTransportNeverRedials(t *testing.T) {
125+ registration := domain.RemoteRegistration{Endpoint: "user@example.test", Incarnation: [16]byte{1}, Generation: 1}
126+ policy := remoteBrokerPolicy("quic")
127+ const identity ports.BrokerDaemonIdentity = "remote-daemon"
128+ tests := []struct {
129+ name string
130+ // silentOpen: the daemon never answers the observation open; otherwise
131+ // the stream opens and the catalogue command is never answered.
132+ silentOpen bool
133+ }{
134+ {name: "unanswered observation open", silentOpen: true},
135+ {name: "unanswered catalogue command"},
136+ }
137+ for _, tt := range tests {
138+ t.Run(tt.name, func(t *testing.T) {
139+ record := ports.BrokerHostRecord{Registration: registration, Policy: policy, Identity: identity}
140+ routes := portsmocks.NewMockBrokerRouteAuthority(t)
141+ routes.EXPECT().ResolveDialTarget(mock.Anything, mock.Anything).Return(ports.BrokerDialTarget{
142+ Fence: ports.BrokerEndpointFence{Registration: registration}, Policy: policy, Address: "opaque", StartMode: ports.BrokerDaemonExistingOnly,
143+ ExpectedIdentity: ports.BrokerExpectedIdentity{Identity: identity, Bound: true},
144+ }, nil)
145+ // No Connect expectation: any redial fails the mock.
146+ connector := portsmocks.NewMockBrokerEndpointConnector(t)
147+ physical := portsmocks.NewMockBrokerPhysicalConnection(t)
148+ healthy := make(chan struct{})
149+ physical.EXPECT().Policy().Return(policy).Maybe()
150+ physical.EXPECT().Identity().Return(identity).Maybe()
151+ physical.EXPECT().Incarnation().Return(ports.BrokerDaemonIncarnation{1}).Maybe()
152+ physical.EXPECT().Done().Return(healthy).Maybe()
153+ physical.EXPECT().Close().Return(nil).Maybe()
154+
155+ ctx, cancel := context.WithCancel(context.Background())
156+ defer cancel()
157+ // waiting signals that the probe is parked on the silent daemon; the
158+ // worker only signals, the test goroutine cancels and asserts.
159+ waiting := make(chan struct{}, 1)
160+ if tt.silentOpen {
161+ physical.EXPECT().OpenStream(mock.Anything, mock.Anything).RunAndReturn(func(ctx context.Context, _ ports.BrokerOpenStreamRequest) (ports.BrokerLogicalConnection, error) {
162+ waiting <- struct{}{}
163+ <-ctx.Done()
164+ return nil, ctx.Err()
165+ })
166+ } else {
167+ logical := portsmocks.NewMockBrokerLogicalConnection(t)
168+ closed := make(chan struct{})
169+ var once sync.Once
170+ logical.EXPECT().SendClient(mock.Anything).Return(nil)
171+ logical.EXPECT().ReceiveServer().RunAndReturn(func() (protocol.ServerMessage, error) {
172+ waiting <- struct{}{}
173+ <-closed
174+ return nil, errors.New("stream closed")
175+ })
176+ logical.EXPECT().Close().RunAndReturn(func() error { once.Do(func() { close(closed) }); return nil })
177+ physical.EXPECT().OpenStream(mock.Anything, mock.Anything).Return(logical, nil)
178+ }
179+ probe := &brokerRemoteProbe{
180+ epoch: 7, routes: routes, binder: portsmocks.NewMockBrokerIdentityBinder(t), connector: connector,
181+ hosts: &routeTestHosts{record: record, found: true},
182+ }
183+ pool, err := broker.NewPool(7, routes, probe.binder, connector, clock.New(), broker.PoolLimits{Physical: 1, Clients: 1, Streams: 2, StreamsPerClient: 2, Warm: 1})
184+ require.NoError(t, err)
185+ t.Cleanup(func() { require.NoError(t, pool.Close()) })
186+ require.True(t, pool.AdoptPhysical(physical, policy))
187+ probe.shared.share(pool)
188+
189+ type outcome struct{ err error }
190+ done := make(chan outcome, 1)
191+ go func() {
192+ _, err := probe.Probe(ctx, registration)
193+ done <- outcome{err: err}
194+ }()
195+ select {
196+ case <-waiting:
197+ case <-time.After(2 * time.Second):
198+ t.Fatal("probe never reached the silent daemon")
199+ }
200+ cancel()
201+ select {
202+ case result := <-done:
203+ require.ErrorIs(t, result.err, context.Canceled)
204+ case <-time.After(2 * time.Second):
205+ t.Fatal("probe outlived its attempt context")
206+ }
207+ })
208+ }
209+}
@@ -3,7 +3,9 @@ package app
3 import (3 import (
4 "context"4 "context"
5 "errors"5 "errors"
6+ "sync"
6 "testing"7 "testing"
8+ "time"
7 9
8 "github.com/bnema/vev/internal/adapters/clock"10 "github.com/bnema/vev/internal/adapters/clock"
9 "github.com/bnema/vev/internal/domain"11 "github.com/bnema/vev/internal/domain"
@@ -113,3 +115,95 @@ func TestBrokerRemoteProbeBorrowsWarmTransport(t *testing.T) {
113 })115 })
114 }116 }
115 }117 }
118+
119+// TestBrokerRemoteProbeSilentBorrowedTransportNeverRedials proves a probe on a
120+// healthy borrowed transport whose daemon never answers is bounded by the
121+// attempt context (the registry's ports.Clock deadline cancels it) and returns
122+// without bootstrapping a second transport: a live pooled physical is never
123+// abandoned for a redial just because one observation went unanswered.
124+func TestBrokerRemoteProbeSilentBorrowedTransportNeverRedials(t *testing.T) {
125+ registration := domain.RemoteRegistration{Endpoint: "user@example.test", Incarnation: [16]byte{1}, Generation: 1}
126+ policy := remoteBrokerPolicy("quic")
127+ const identity ports.BrokerDaemonIdentity = "remote-daemon"
128+ tests := []struct {
129+ name string
130+ // silentOpen: the daemon never answers the observation open; otherwise
131+ // the stream opens and the catalogue command is never answered.
132+ silentOpen bool
133+ }{
134+ {name: "unanswered observation open", silentOpen: true},
135+ {name: "unanswered catalogue command"},
136+ }
137+ for _, tt := range tests {
138+ t.Run(tt.name, func(t *testing.T) {
139+ record := ports.BrokerHostRecord{Registration: registration, Policy: policy, Identity: identity}
140+ routes := portsmocks.NewMockBrokerRouteAuthority(t)
141+ routes.EXPECT().ResolveDialTarget(mock.Anything, mock.Anything).Return(ports.BrokerDialTarget{
142+ Fence: ports.BrokerEndpointFence{Registration: registration}, Policy: policy, Address: "opaque", StartMode: ports.BrokerDaemonExistingOnly,
143+ ExpectedIdentity: ports.BrokerExpectedIdentity{Identity: identity, Bound: true},
144+ }, nil)
145+ // No Connect expectation: any redial fails the mock.
146+ connector := portsmocks.NewMockBrokerEndpointConnector(t)
147+ physical := portsmocks.NewMockBrokerPhysicalConnection(t)
148+ healthy := make(chan struct{})
149+ physical.EXPECT().Policy().Return(policy).Maybe()
150+ physical.EXPECT().Identity().Return(identity).Maybe()
151+ physical.EXPECT().Incarnation().Return(ports.BrokerDaemonIncarnation{1}).Maybe()
152+ physical.EXPECT().Done().Return(healthy).Maybe()
153+ physical.EXPECT().Close().Return(nil).Maybe()
154+
155+ ctx, cancel := context.WithCancel(context.Background())
156+ defer cancel()
157+ // waiting signals that the probe is parked on the silent daemon; the
158+ // worker only signals, the test goroutine cancels and asserts.
159+ waiting := make(chan struct{}, 1)
160+ if tt.silentOpen {
161+ physical.EXPECT().OpenStream(mock.Anything, mock.Anything).RunAndReturn(func(ctx context.Context, _ ports.BrokerOpenStreamRequest) (ports.BrokerLogicalConnection, error) {
162+ waiting <- struct{}{}
163+ <-ctx.Done()
164+ return nil, ctx.Err()
165+ })
166+ } else {
167+ logical := portsmocks.NewMockBrokerLogicalConnection(t)
168+ closed := make(chan struct{})
169+ var once sync.Once
170+ logical.EXPECT().SendClient(mock.Anything).Return(nil)
171+ logical.EXPECT().ReceiveServer().RunAndReturn(func() (protocol.ServerMessage, error) {
172+ waiting <- struct{}{}
173+ <-closed
174+ return nil, errors.New("stream closed")
175+ })
176+ logical.EXPECT().Close().RunAndReturn(func() error { once.Do(func() { close(closed) }); return nil })
177+ physical.EXPECT().OpenStream(mock.Anything, mock.Anything).Return(logical, nil)
178+ }
179+ probe := &brokerRemoteProbe{
180+ epoch: 7, routes: routes, binder: portsmocks.NewMockBrokerIdentityBinder(t), connector: connector,
181+ hosts: &routeTestHosts{record: record, found: true},
182+ }
183+ pool, err := broker.NewPool(7, routes, probe.binder, connector, clock.New(), broker.PoolLimits{Physical: 1, Clients: 1, Streams: 2, StreamsPerClient: 2, Warm: 1})
184+ require.NoError(t, err)
185+ t.Cleanup(func() { require.NoError(t, pool.Close()) })
186+ require.True(t, pool.AdoptPhysical(physical, policy))
187+ probe.shared.share(pool)
188+
189+ type outcome struct{ err error }
190+ done := make(chan outcome, 1)
191+ go func() {
192+ _, err := probe.Probe(ctx, registration)
193+ done <- outcome{err: err}
194+ }()
195+ select {
196+ case <-waiting:
197+ case <-time.After(2 * time.Second):
198+ t.Fatal("probe never reached the silent daemon")
199+ }
200+ cancel()
201+ select {
202+ case result := <-done:
203+ require.ErrorIs(t, result.err, context.Canceled)
204+ case <-time.After(2 * time.Second):
205+ t.Fatal("probe outlived its attempt context")
206+ }
207+ })
208+ }
209+}

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