detached 489a275

Merge pull request #56 from bnema/fix/udp-proxy-idlettl

fix(remote): honor UDP proxy idle TTL
commit
489a2758e556a3272ce6bf2979ae59939e64af8c
author
bnema · · 2026-07-09 16:24 UTC
committer
GitHub · · 2026-07-09 16:24 UTC
merge of
348a1c1 + 18e2515 · diff against the first parent
signature
GPG UnverifiedB5690EEEBB952194 The key is unknown
4 files · +192 -10
Split
UnifiedSplit
M internal/adapters/dgram/proxy.go +12 -5
@@ -93,6 +93,9 @@ func (p ProxyRuntime) Run(ctx context.Context) error {
9393 }
9494 }()
9595 startIdle := func() {
96+ if idle != nil {
97+ return
98+ }
9699 if timer == nil {
97100 timer = clk.NewTimer(idleTTL)
98101 } else {
@@ -116,12 +119,10 @@ func (p ProxyRuntime) Run(ctx context.Context) error {
116119 continue
117120 }
118121 switch ev.State {
119- case ports.LinkStateDegraded, ports.LinkStateProbing, ports.LinkStateOffline:
122+ case ports.LinkStateDegraded, ports.LinkStateProbing, ports.LinkStateOffline, ports.LinkStateDead:
120123 startIdle()
121124 case ports.LinkStateConnected:
122125 stopIdle()
123- case ports.LinkStateDead:
124- return ErrLinkDead
125126 }
126127 case <-idle:
127128 return ErrLinkDead
@@ -132,11 +133,17 @@ func (p ProxyRuntime) Run(ctx context.Context) error {
132133 return nil
133134 }
134135 return ce.err
136+ case proxyClientRecv, proxyClientSend:
137+ if errors.Is(ce.err, ErrLinkDead) {
138+ startIdle()
139+ continue
140+ }
141+ return ce.err
135142 default:
136143 // Recoverable client-side errors are retried inside the copier
137144 // without dropping the frame; anything that surfaces here (a
138- // daemon send failure, or a non-recoverable client error such as
139- // a dead link) is fatal.
145+ // daemon send failure, or a non-recoverable client send failure)
146+ // is fatal.
140147 return ce.err
141148 }
142149 }
@@ -93,6 +93,9 @@ func (p ProxyRuntime) Run(ctx context.Context) error {
93 }93 }
94 }()94 }()
95 startIdle := func() {95 startIdle := func() {
96+ if idle != nil {
97+ return
98+ }
96 if timer == nil {99 if timer == nil {
97 timer = clk.NewTimer(idleTTL)100 timer = clk.NewTimer(idleTTL)
98 } else {101 } else {
@@ -116,12 +119,10 @@ func (p ProxyRuntime) Run(ctx context.Context) error {
116 continue119 continue
117 }120 }
118 switch ev.State {121 switch ev.State {
119- case ports.LinkStateDegraded, ports.LinkStateProbing, ports.LinkStateOffline:122+ case ports.LinkStateDegraded, ports.LinkStateProbing, ports.LinkStateOffline, ports.LinkStateDead:
120 startIdle()123 startIdle()
121 case ports.LinkStateConnected:124 case ports.LinkStateConnected:
122 stopIdle()125 stopIdle()
123- case ports.LinkStateDead:
124- return ErrLinkDead
125 }126 }
126 case <-idle:127 case <-idle:
127 return ErrLinkDead128 return ErrLinkDead
@@ -132,11 +133,17 @@ func (p ProxyRuntime) Run(ctx context.Context) error {
132 return nil133 return nil
133 }134 }
134 return ce.err135 return ce.err
136+ case proxyClientRecv, proxyClientSend:
137+ if errors.Is(ce.err, ErrLinkDead) {
138+ startIdle()
139+ continue
140+ }
141+ return ce.err
135 default:142 default:
136 // Recoverable client-side errors are retried inside the copier143 // Recoverable client-side errors are retried inside the copier
137 // without dropping the frame; anything that surfaces here (a144 // without dropping the frame; anything that surfaces here (a
138- // daemon send failure, or a non-recoverable client error such as145+ // daemon send failure, or a non-recoverable client send failure)
139- // a dead link) is fatal.146+ // is fatal.
140 return ce.err147 return ce.err
141 }148 }
142 }149 }
M internal/adapters/dgram/proxy_test.go +168 -3
@@ -182,21 +182,186 @@ func TestProxyRuntimeOfflineLinkEventExpiresViaIdleTTL(t *testing.T) {
182182 }
183183 }
184184
185-func TestProxyRuntimeDeadLinkEventTearsDown(t *testing.T) {
185+type proxyManualClock struct {
186+ mu sync.Mutex
187+ now time.Time
188+ timers chan *proxyManualTimer
189+}
190+
191+func newProxyManualClock(now time.Time) *proxyManualClock {
192+ return &proxyManualClock{now: now, timers: make(chan *proxyManualTimer, 4)}
193+}
194+
195+func (c *proxyManualClock) Now() time.Time {
196+ c.mu.Lock()
197+ defer c.mu.Unlock()
198+ return c.now
199+}
200+
201+func (c *proxyManualClock) NewTimer(d time.Duration) ports.Timer {
202+ tm := &proxyManualTimer{c: make(chan time.Time, 1), d: d}
203+ c.timers <- tm
204+ return tm
205+}
206+
207+func (c *proxyManualClock) advance(d time.Duration) {
208+ c.mu.Lock()
209+ c.now = c.now.Add(d)
210+ now := c.now
211+ c.mu.Unlock()
212+ for {
213+ select {
214+ case tm := <-c.timers:
215+ if tm.duration() <= d {
216+ tm.fire(now)
217+ } else {
218+ c.timers <- tm
219+ return
220+ }
221+ default:
222+ return
223+ }
224+ }
225+}
226+
227+type proxyManualTimer struct {
228+ mu sync.Mutex
229+ c chan time.Time
230+ d time.Duration
231+}
232+
233+func (t *proxyManualTimer) C() <-chan time.Time { return t.c }
234+func (t *proxyManualTimer) Reset(d time.Duration) bool {
235+ t.mu.Lock()
236+ defer t.mu.Unlock()
237+ t.d = d
238+ return true
239+}
240+func (t *proxyManualTimer) Stop() bool { return true }
241+func (t *proxyManualTimer) duration() time.Duration {
242+ t.mu.Lock()
243+ defer t.mu.Unlock()
244+ return t.d
245+}
246+func (t *proxyManualTimer) fire(now time.Time) {
247+ select {
248+ case t.c <- now:
249+ default:
250+ }
251+}
252+
253+func waitForProxyManualTimers(t *testing.T, clk *proxyManualClock, n int) {
254+ t.Helper()
255+ deadline := time.Now().Add(time.Second)
256+ for time.Now().Before(deadline) {
257+ if len(clk.timers) >= n {
258+ return
259+ }
260+ time.Sleep(time.Millisecond)
261+ }
262+ t.Fatalf("manual timers=%d, want at least %d", len(clk.timers), n)
263+}
264+
265+func TestProxyRuntimeClientDeadWaitsForIdleTTL(t *testing.T) {
186266 client := newFakeTransport()
187267 daemon := newFakeTransport()
268+ clk := newProxyManualClock(time.Unix(0, 0))
188269 errCh := make(chan error, 1)
189270 go func() {
190- errCh <- ProxyRuntime{Client: client, Daemon: daemon, IdleTTL: time.Hour}.Run(t.Context())
271+ errCh <- ProxyRuntime{Client: client, Daemon: daemon, IdleTTL: time.Minute, Clock: clk}.Run(t.Context())
191272 }()
273+
192274 client.linkEvents <- ports.LinkEvent{State: ports.LinkStateDead}
275+ waitForProxyManualTimers(t, clk, 1)
276+ select {
277+ case err := <-errCh:
278+ t.Fatalf("Run returned before IdleTTL after client LinkStateDead: %v", err)
279+ default:
280+ }
281+
282+ clk.advance(time.Minute)
283+ select {
284+ case err := <-errCh:
285+ if !errors.Is(err, ErrLinkDead) {
286+ t.Fatalf("Run err=%v, want ErrLinkDead", err)
287+ }
288+ case <-time.After(time.Second):
289+ t.Fatal("timeout waiting for idle expiry after client LinkStateDead")
290+ }
291+}
292+
293+func TestProxyRuntimeClientRecvDeadWaitsForIdleTTL(t *testing.T) {
294+ client := newFakeTransport()
295+ daemon := newFakeTransport()
296+ clk := newProxyManualClock(time.Unix(0, 0))
297+ errCh := make(chan error, 1)
298+ go func() {
299+ errCh <- ProxyRuntime{Client: client, Daemon: daemon, IdleTTL: time.Minute, Clock: clk}.Run(t.Context())
300+ }()
301+
302+ client.recv <- recvResult{err: ErrLinkDead}
303+ waitForProxyManualTimers(t, clk, 1)
304+ select {
305+ case err := <-errCh:
306+ t.Fatalf("Run returned before IdleTTL after client Recv ErrLinkDead: %v", err)
307+ default:
308+ }
309+
310+ clk.advance(time.Minute)
311+ select {
312+ case err := <-errCh:
313+ if !errors.Is(err, ErrLinkDead) {
314+ t.Fatalf("Run err=%v, want ErrLinkDead", err)
315+ }
316+ case <-time.After(time.Second):
317+ t.Fatal("timeout waiting for idle expiry after client Recv ErrLinkDead")
318+ }
319+}
320+
321+func TestProxyRuntimeClientSendDeadWaitsForIdleTTL(t *testing.T) {
322+ client := newFakeTransport()
323+ daemon := newFakeTransport()
324+ clk := newProxyManualClock(time.Unix(0, 0))
325+ errCh := make(chan error, 1)
326+ go func() {
327+ errCh <- ProxyRuntime{Client: client, Daemon: daemon, IdleTTL: time.Minute, Clock: clk}.Run(t.Context())
328+ }()
329+
330+ client.sendErr <- ErrLinkDead
331+ daemon.recv <- recvResult{frame: ports.Frame{Type: ports.MsgOutput, Payload: []byte("client is gone")}}
332+ waitForProxyManualTimers(t, clk, 1)
333+ select {
334+ case err := <-errCh:
335+ t.Fatalf("Run returned before IdleTTL after client Send ErrLinkDead: %v", err)
336+ default:
337+ }
338+
339+ clk.advance(time.Minute)
340+ select {
341+ case err := <-errCh:
342+ if !errors.Is(err, ErrLinkDead) {
343+ t.Fatalf("Run err=%v, want ErrLinkDead", err)
344+ }
345+ case <-time.After(time.Second):
346+ t.Fatal("timeout waiting for idle expiry after client Send ErrLinkDead")
347+ }
348+}
349+
350+func TestProxyRuntimeDaemonRecvDeadTearsDown(t *testing.T) {
351+ client := newFakeTransport()
352+ daemon := newFakeTransport()
353+ errCh := make(chan error, 1)
354+ go func() {
355+ errCh <- ProxyRuntime{Client: client, Daemon: daemon, IdleTTL: time.Hour}.Run(t.Context())
356+ }()
357+ daemon.recv <- recvResult{err: ErrLinkDead}
193358 select {
194359 case err := <-errCh:
195360 if !errors.Is(err, ErrLinkDead) {
196361 t.Fatalf("Run err=%v, want ErrLinkDead", err)
197362 }
198363 case <-time.After(time.Second):
199- t.Fatal("dead link event did not tear down proxy")
364+ t.Fatal("daemon recv dead did not tear down proxy")
200365 }
201366 }
202367
@@ -182,21 +182,186 @@ func TestProxyRuntimeOfflineLinkEventExpiresViaIdleTTL(t *testing.T) {
182 }182 }
183 }183 }
184 184
185-func TestProxyRuntimeDeadLinkEventTearsDown(t *testing.T) {185+type proxyManualClock struct {
186+ mu sync.Mutex
187+ now time.Time
188+ timers chan *proxyManualTimer
189+}
190+
191+func newProxyManualClock(now time.Time) *proxyManualClock {
192+ return &proxyManualClock{now: now, timers: make(chan *proxyManualTimer, 4)}
193+}
194+
195+func (c *proxyManualClock) Now() time.Time {
196+ c.mu.Lock()
197+ defer c.mu.Unlock()
198+ return c.now
199+}
200+
201+func (c *proxyManualClock) NewTimer(d time.Duration) ports.Timer {
202+ tm := &proxyManualTimer{c: make(chan time.Time, 1), d: d}
203+ c.timers <- tm
204+ return tm
205+}
206+
207+func (c *proxyManualClock) advance(d time.Duration) {
208+ c.mu.Lock()
209+ c.now = c.now.Add(d)
210+ now := c.now
211+ c.mu.Unlock()
212+ for {
213+ select {
214+ case tm := <-c.timers:
215+ if tm.duration() <= d {
216+ tm.fire(now)
217+ } else {
218+ c.timers <- tm
219+ return
220+ }
221+ default:
222+ return
223+ }
224+ }
225+}
226+
227+type proxyManualTimer struct {
228+ mu sync.Mutex
229+ c chan time.Time
230+ d time.Duration
231+}
232+
233+func (t *proxyManualTimer) C() <-chan time.Time { return t.c }
234+func (t *proxyManualTimer) Reset(d time.Duration) bool {
235+ t.mu.Lock()
236+ defer t.mu.Unlock()
237+ t.d = d
238+ return true
239+}
240+func (t *proxyManualTimer) Stop() bool { return true }
241+func (t *proxyManualTimer) duration() time.Duration {
242+ t.mu.Lock()
243+ defer t.mu.Unlock()
244+ return t.d
245+}
246+func (t *proxyManualTimer) fire(now time.Time) {
247+ select {
248+ case t.c <- now:
249+ default:
250+ }
251+}
252+
253+func waitForProxyManualTimers(t *testing.T, clk *proxyManualClock, n int) {
254+ t.Helper()
255+ deadline := time.Now().Add(time.Second)
256+ for time.Now().Before(deadline) {
257+ if len(clk.timers) >= n {
258+ return
259+ }
260+ time.Sleep(time.Millisecond)
261+ }
262+ t.Fatalf("manual timers=%d, want at least %d", len(clk.timers), n)
263+}
264+
265+func TestProxyRuntimeClientDeadWaitsForIdleTTL(t *testing.T) {
186 client := newFakeTransport()266 client := newFakeTransport()
187 daemon := newFakeTransport()267 daemon := newFakeTransport()
268+ clk := newProxyManualClock(time.Unix(0, 0))
188 errCh := make(chan error, 1)269 errCh := make(chan error, 1)
189 go func() {270 go func() {
190- errCh <- ProxyRuntime{Client: client, Daemon: daemon, IdleTTL: time.Hour}.Run(t.Context())271+ errCh <- ProxyRuntime{Client: client, Daemon: daemon, IdleTTL: time.Minute, Clock: clk}.Run(t.Context())
191 }()272 }()
273+
192 client.linkEvents <- ports.LinkEvent{State: ports.LinkStateDead}274 client.linkEvents <- ports.LinkEvent{State: ports.LinkStateDead}
275+ waitForProxyManualTimers(t, clk, 1)
276+ select {
277+ case err := <-errCh:
278+ t.Fatalf("Run returned before IdleTTL after client LinkStateDead: %v", err)
279+ default:
280+ }
281+
282+ clk.advance(time.Minute)
283+ select {
284+ case err := <-errCh:
285+ if !errors.Is(err, ErrLinkDead) {
286+ t.Fatalf("Run err=%v, want ErrLinkDead", err)
287+ }
288+ case <-time.After(time.Second):
289+ t.Fatal("timeout waiting for idle expiry after client LinkStateDead")
290+ }
291+}
292+
293+func TestProxyRuntimeClientRecvDeadWaitsForIdleTTL(t *testing.T) {
294+ client := newFakeTransport()
295+ daemon := newFakeTransport()
296+ clk := newProxyManualClock(time.Unix(0, 0))
297+ errCh := make(chan error, 1)
298+ go func() {
299+ errCh <- ProxyRuntime{Client: client, Daemon: daemon, IdleTTL: time.Minute, Clock: clk}.Run(t.Context())
300+ }()
301+
302+ client.recv <- recvResult{err: ErrLinkDead}
303+ waitForProxyManualTimers(t, clk, 1)
304+ select {
305+ case err := <-errCh:
306+ t.Fatalf("Run returned before IdleTTL after client Recv ErrLinkDead: %v", err)
307+ default:
308+ }
309+
310+ clk.advance(time.Minute)
311+ select {
312+ case err := <-errCh:
313+ if !errors.Is(err, ErrLinkDead) {
314+ t.Fatalf("Run err=%v, want ErrLinkDead", err)
315+ }
316+ case <-time.After(time.Second):
317+ t.Fatal("timeout waiting for idle expiry after client Recv ErrLinkDead")
318+ }
319+}
320+
321+func TestProxyRuntimeClientSendDeadWaitsForIdleTTL(t *testing.T) {
322+ client := newFakeTransport()
323+ daemon := newFakeTransport()
324+ clk := newProxyManualClock(time.Unix(0, 0))
325+ errCh := make(chan error, 1)
326+ go func() {
327+ errCh <- ProxyRuntime{Client: client, Daemon: daemon, IdleTTL: time.Minute, Clock: clk}.Run(t.Context())
328+ }()
329+
330+ client.sendErr <- ErrLinkDead
331+ daemon.recv <- recvResult{frame: ports.Frame{Type: ports.MsgOutput, Payload: []byte("client is gone")}}
332+ waitForProxyManualTimers(t, clk, 1)
333+ select {
334+ case err := <-errCh:
335+ t.Fatalf("Run returned before IdleTTL after client Send ErrLinkDead: %v", err)
336+ default:
337+ }
338+
339+ clk.advance(time.Minute)
340+ select {
341+ case err := <-errCh:
342+ if !errors.Is(err, ErrLinkDead) {
343+ t.Fatalf("Run err=%v, want ErrLinkDead", err)
344+ }
345+ case <-time.After(time.Second):
346+ t.Fatal("timeout waiting for idle expiry after client Send ErrLinkDead")
347+ }
348+}
349+
350+func TestProxyRuntimeDaemonRecvDeadTearsDown(t *testing.T) {
351+ client := newFakeTransport()
352+ daemon := newFakeTransport()
353+ errCh := make(chan error, 1)
354+ go func() {
355+ errCh <- ProxyRuntime{Client: client, Daemon: daemon, IdleTTL: time.Hour}.Run(t.Context())
356+ }()
357+ daemon.recv <- recvResult{err: ErrLinkDead}
193 select {358 select {
194 case err := <-errCh:359 case err := <-errCh:
195 if !errors.Is(err, ErrLinkDead) {360 if !errors.Is(err, ErrLinkDead) {
196 t.Fatalf("Run err=%v, want ErrLinkDead", err)361 t.Fatalf("Run err=%v, want ErrLinkDead", err)
197 }362 }
198 case <-time.After(time.Second):363 case <-time.After(time.Second):
199- t.Fatal("dead link event did not tear down proxy")364+ t.Fatal("daemon recv dead did not tear down proxy")
200 }365 }
201 }366 }
202 367
M internal/app/run.go +7 -2
@@ -770,10 +770,15 @@ func runUDPProxy(ctx context.Context, session string, ready io.Writer) error {
770770 if session != "" {
771771 dgramTr = preHelloNameTransport{Transport: dg, session: session}
772772 }
773- return dgram.ProxyRuntime{Client: dgramTr, Daemon: daemonTr, Log: log}.Run(ctx)
773+ return dgram.ProxyRuntime{Client: dgramTr, Daemon: daemonTr, Log: log, IdleTTL: udpProxyIdleTTL}.Run(ctx)
774774 }
775775
776-var udpProxyClientTransportOptions = dgram.Options{MaxPending: 32}
776+const udpProxyIdleTTL = 15 * time.Minute
777+
778+var udpProxyClientTransportOptions = dgram.Options{
779+ MaxPending: 32,
780+ DeadAfter: 2 * udpProxyIdleTTL,
781+}
777782
778783 const (
779784 // envUDPPortRange configures the remote UDP proxy's listen port range so a
@@ -770,10 +770,15 @@ func runUDPProxy(ctx context.Context, session string, ready io.Writer) error {
770 if session != "" {770 if session != "" {
771 dgramTr = preHelloNameTransport{Transport: dg, session: session}771 dgramTr = preHelloNameTransport{Transport: dg, session: session}
772 }772 }
773- return dgram.ProxyRuntime{Client: dgramTr, Daemon: daemonTr, Log: log}.Run(ctx)773+ return dgram.ProxyRuntime{Client: dgramTr, Daemon: daemonTr, Log: log, IdleTTL: udpProxyIdleTTL}.Run(ctx)
774 }774 }
775 775
776-var udpProxyClientTransportOptions = dgram.Options{MaxPending: 32}776+const udpProxyIdleTTL = 15 * time.Minute
777+
778+var udpProxyClientTransportOptions = dgram.Options{
779+ MaxPending: 32,
780+ DeadAfter: 2 * udpProxyIdleTTL,
781+}
777 782
778 const (783 const (
779 // envUDPPortRange configures the remote UDP proxy's listen port range so a784 // envUDPPortRange configures the remote UDP proxy's listen port range so a
M internal/app/run_test.go +5 -0
@@ -29,6 +29,11 @@ func TestRunUDPProxyUsesBoundedClientMaxPending(t *testing.T) {
2929 require.Equal(t, 32, udpProxyClientTransportOptions.MaxPending)
3030 }
3131
32+func TestRunUDPProxyClientDeadAfterExceedsIdleTTL(t *testing.T) {
33+ require.Positive(t, udpProxyIdleTTL)
34+ require.Greater(t, udpProxyClientTransportOptions.DeadAfter, udpProxyIdleTTL)
35+}
36+
3237 func TestParseArgs(t *testing.T) {
3338 tests := []struct {
3439 name string
@@ -29,6 +29,11 @@ func TestRunUDPProxyUsesBoundedClientMaxPending(t *testing.T) {
29 require.Equal(t, 32, udpProxyClientTransportOptions.MaxPending)29 require.Equal(t, 32, udpProxyClientTransportOptions.MaxPending)
30 }30 }
31 31
32+func TestRunUDPProxyClientDeadAfterExceedsIdleTTL(t *testing.T) {
33+ require.Positive(t, udpProxyIdleTTL)
34+ require.Greater(t, udpProxyClientTransportOptions.DeadAfter, udpProxyIdleTTL)
35+}
36+
32 func TestParseArgs(t *testing.T) {37 func TestParseArgs(t *testing.T) {
33 tests := []struct {38 tests := []struct {
34 name string39 name string

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 2

Tags 13

↑↓ moveenter openesc close