detached 8c4b9b1

improve(dgram): expose privacy-safe UDP health diagnostics

commit
8c4b9b1df575554338f0c81b6a6e2ba44f2b7d2e
author
brice · · 2026-07-09 23:52 UTC
parent
f2100d6
signature
GPG VerifiedCD88F9E564E0102646FB6077A1A07C7C7C3E349D · arch-gpg Verified for the committer email
6 files · +277 -70
Unified
UnifiedSplit
A internal/adapters/dgram/diagnostics.go +47 -0
@@ -0,0 +1,47 @@
1+package dgram
2+
3+import (
4+ "log/slog"
5+ "time"
6+
7+ "github.com/bnema/vev/internal/ports"
8+)
9+
10+const diagnosticBufferSize = 16
11+
12+// Diagnostic is an immutable, privacy-safe snapshot of UDP link health.
13+// It intentionally contains no packet payloads, addresses, keys, or identities.
14+type Diagnostic struct {
15+ At time.Time
16+ State ports.LinkState
17+ SinceAuthenticatedPacket time.Duration
18+ SinceCompleteRecord time.Duration
19+ SinceACKProgress time.Duration
20+ PendingRecords int
21+ PendingBytes int
22+ Retransmits uint64
23+ ReassemblyInflight int
24+}
25+
26+// DiagnosticObserver receives bounded asynchronous UDP health snapshots.
27+type DiagnosticObserver func(Diagnostic)
28+
29+// DiagnosticLogObserver returns an observer suitable for composition roots. It
30+// logs only the privacy-safe diagnostic counters and ages.
31+func DiagnosticLogObserver(log *slog.Logger) DiagnosticObserver {
32+ if log == nil {
33+ return nil
34+ }
35+ return func(d Diagnostic) {
36+ log.Debug("udp health",
37+ "state", d.State,
38+ "since_authenticated_packet", d.SinceAuthenticatedPacket,
39+ "since_complete_record", d.SinceCompleteRecord,
40+ "since_ack_progress", d.SinceACKProgress,
41+ "pending_records", d.PendingRecords,
42+ "pending_bytes", d.PendingBytes,
43+ "retransmits", d.Retransmits,
44+ "reassembly_inflight", d.ReassemblyInflight,
45+ )
46+ }
47+}
@@ -0,0 +1,47 @@
1+package dgram
2+
3+import (
4+ "log/slog"
5+ "time"
6+
7+ "github.com/bnema/vev/internal/ports"
8+)
9+
10+const diagnosticBufferSize = 16
11+
12+// Diagnostic is an immutable, privacy-safe snapshot of UDP link health.
13+// It intentionally contains no packet payloads, addresses, keys, or identities.
14+type Diagnostic struct {
15+ At time.Time
16+ State ports.LinkState
17+ SinceAuthenticatedPacket time.Duration
18+ SinceCompleteRecord time.Duration
19+ SinceACKProgress time.Duration
20+ PendingRecords int
21+ PendingBytes int
22+ Retransmits uint64
23+ ReassemblyInflight int
24+}
25+
26+// DiagnosticObserver receives bounded asynchronous UDP health snapshots.
27+type DiagnosticObserver func(Diagnostic)
28+
29+// DiagnosticLogObserver returns an observer suitable for composition roots. It
30+// logs only the privacy-safe diagnostic counters and ages.
31+func DiagnosticLogObserver(log *slog.Logger) DiagnosticObserver {
32+ if log == nil {
33+ return nil
34+ }
35+ return func(d Diagnostic) {
36+ log.Debug("udp health",
37+ "state", d.State,
38+ "since_authenticated_packet", d.SinceAuthenticatedPacket,
39+ "since_complete_record", d.SinceCompleteRecord,
40+ "since_ack_progress", d.SinceACKProgress,
41+ "pending_records", d.PendingRecords,
42+ "pending_bytes", d.PendingBytes,
43+ "retransmits", d.Retransmits,
44+ "reassembly_inflight", d.ReassemblyInflight,
45+ )
46+ }
47+}
A internal/adapters/dgram/diagnostics_test.go +67 -0
@@ -0,0 +1,67 @@
1+package dgram
2+
3+import (
4+ "testing"
5+ "time"
6+
7+ "github.com/bnema/vev/internal/ports"
8+)
9+
10+func TestDiagnosticSnapshotSeparatesProgressAges(t *testing.T) {
11+ now := time.Date(2030, 1, 2, 3, 4, 5, 0, time.UTC)
12+ got := make(chan Diagnostic, 1)
13+ tr := &Transport{
14+ clock: fixedClock{now: now},
15+ linkState: ports.LinkStateDegraded,
16+ lastAuthenticatedPacket: now.Add(-time.Second),
17+ lastCompleteRecord: now.Add(-2 * time.Second),
18+ lastACKProgress: now.Add(-3 * time.Second),
19+ retransmits: 64,
20+ reassemblyInflight: 2,
21+ pending: map[uint64]*pending{
22+ 1: {frame: ports.Frame{Payload: make([]byte, 7)}},
23+ 2: {frame: ports.Frame{Payload: make([]byte, 11)}},
24+ },
25+ diagnosticCh: make(chan Diagnostic, diagnosticBufferSize),
26+ done: make(chan struct{}),
27+ }
28+ tr.observe = func(d Diagnostic) {
29+ // This would deadlock if the observer were invoked under Transport.mu.
30+ _ = tr.LinkState()
31+ got <- d
32+ }
33+ go tr.diagnosticLoop()
34+ defer close(tr.done)
35+
36+ tr.emitDiagnostic()
37+ select {
38+ case d := <-got:
39+ if d.At != now || d.State != ports.LinkStateDegraded {
40+ t.Fatalf("identity = %+v", d)
41+ }
42+ if d.SinceAuthenticatedPacket != time.Second || d.SinceCompleteRecord != 2*time.Second || d.SinceACKProgress != 3*time.Second {
43+ t.Fatalf("ages = packet %v, record %v, ack %v", d.SinceAuthenticatedPacket, d.SinceCompleteRecord, d.SinceACKProgress)
44+ }
45+ if d.PendingRecords != 2 || d.PendingBytes != 18 || d.Retransmits != 64 || d.ReassemblyInflight != 2 {
46+ t.Fatalf("counters = %+v", d)
47+ }
48+ case <-time.After(time.Second):
49+ t.Fatal("diagnostic observer was not called")
50+ }
51+}
52+
53+func TestDiagnosticObserverIsBounded(t *testing.T) {
54+ tr := &Transport{
55+ clock: fixedClock{now: time.Now()},
56+ pending: make(map[uint64]*pending),
57+ diagnosticCh: make(chan Diagnostic, 1),
58+ done: make(chan struct{}),
59+ }
60+ for range 100 {
61+ tr.emitDiagnostic()
62+ }
63+ if got := len(tr.diagnosticCh); got != 1 {
64+ t.Fatalf("queued diagnostics = %d, want bounded buffer", got)
65+ }
66+ close(tr.done)
67+}
@@ -0,0 +1,67 @@
1+package dgram
2+
3+import (
4+ "testing"
5+ "time"
6+
7+ "github.com/bnema/vev/internal/ports"
8+)
9+
10+func TestDiagnosticSnapshotSeparatesProgressAges(t *testing.T) {
11+ now := time.Date(2030, 1, 2, 3, 4, 5, 0, time.UTC)
12+ got := make(chan Diagnostic, 1)
13+ tr := &Transport{
14+ clock: fixedClock{now: now},
15+ linkState: ports.LinkStateDegraded,
16+ lastAuthenticatedPacket: now.Add(-time.Second),
17+ lastCompleteRecord: now.Add(-2 * time.Second),
18+ lastACKProgress: now.Add(-3 * time.Second),
19+ retransmits: 64,
20+ reassemblyInflight: 2,
21+ pending: map[uint64]*pending{
22+ 1: {frame: ports.Frame{Payload: make([]byte, 7)}},
23+ 2: {frame: ports.Frame{Payload: make([]byte, 11)}},
24+ },
25+ diagnosticCh: make(chan Diagnostic, diagnosticBufferSize),
26+ done: make(chan struct{}),
27+ }
28+ tr.observe = func(d Diagnostic) {
29+ // This would deadlock if the observer were invoked under Transport.mu.
30+ _ = tr.LinkState()
31+ got <- d
32+ }
33+ go tr.diagnosticLoop()
34+ defer close(tr.done)
35+
36+ tr.emitDiagnostic()
37+ select {
38+ case d := <-got:
39+ if d.At != now || d.State != ports.LinkStateDegraded {
40+ t.Fatalf("identity = %+v", d)
41+ }
42+ if d.SinceAuthenticatedPacket != time.Second || d.SinceCompleteRecord != 2*time.Second || d.SinceACKProgress != 3*time.Second {
43+ t.Fatalf("ages = packet %v, record %v, ack %v", d.SinceAuthenticatedPacket, d.SinceCompleteRecord, d.SinceACKProgress)
44+ }
45+ if d.PendingRecords != 2 || d.PendingBytes != 18 || d.Retransmits != 64 || d.ReassemblyInflight != 2 {
46+ t.Fatalf("counters = %+v", d)
47+ }
48+ case <-time.After(time.Second):
49+ t.Fatal("diagnostic observer was not called")
50+ }
51+}
52+
53+func TestDiagnosticObserverIsBounded(t *testing.T) {
54+ tr := &Transport{
55+ clock: fixedClock{now: time.Now()},
56+ pending: make(map[uint64]*pending),
57+ diagnosticCh: make(chan Diagnostic, 1),
58+ done: make(chan struct{}),
59+ }
60+ for range 100 {
61+ tr.emitDiagnostic()
62+ }
63+ if got := len(tr.diagnosticCh); got != 1 {
64+ t.Fatalf("queued diagnostics = %d, want bounded buffer", got)
65+ }
66+ close(tr.done)
67+}
M internal/adapters/dgram/ssh_dialer.go +1 -0
@@ -176,6 +176,7 @@ func (d RemoteDialer) Dial(ctx context.Context) (ports.Transport, error) {
176176 // fresh local UDP socket once the link is silent long enough to probe. The
177177 // bind reuses listenUDPPacket so tests can stub it.
178178 t, err := NewTransportWithOptions(pc, peer, ready.key, 1, 2, Options{
179+ Observe: DiagnosticLogObserver(d.Log),
179180 RebindPacketConn: func(net.PacketConn) (net.PacketConn, error) {
180181 return listenUDPPacket(ctx)
181182 },
@@ -176,6 +176,7 @@ func (d RemoteDialer) Dial(ctx context.Context) (ports.Transport, error) {
176 // fresh local UDP socket once the link is silent long enough to probe. The176 // fresh local UDP socket once the link is silent long enough to probe. The
177 // bind reuses listenUDPPacket so tests can stub it.177 // bind reuses listenUDPPacket so tests can stub it.
178 t, err := NewTransportWithOptions(pc, peer, ready.key, 1, 2, Options{178 t, err := NewTransportWithOptions(pc, peer, ready.key, 1, 2, Options{
179+ Observe: DiagnosticLogObserver(d.Log),
179 RebindPacketConn: func(net.PacketConn) (net.PacketConn, error) {180 RebindPacketConn: func(net.PacketConn) (net.PacketConn, error) {
180 return listenUDPPacket(ctx)181 return listenUDPPacket(ctx)
181 },182 },
M internal/adapters/dgram/transport.go +155 -69
@@ -64,6 +64,8 @@ type Options struct {
6464 MaxPendingWait time.Duration
6565 MaxRecvBuffer int
6666 Clock ports.Clock
67+ // Observe receives bounded, asynchronous, privacy-safe link health snapshots.
68+ Observe DiagnosticObserver
6769 // RebindPacketConn is an optional client-side hook used to hop the local UDP
6870 // socket when the peer is offline. Server/proxy transports should leave it nil.
6971 RebindPacketConn func(net.PacketConn) (net.PacketConn, error)
@@ -137,41 +139,48 @@ type Transport struct {
137139 sendDir, recvDir uint32
138140 mtu int
139141
140- mu sync.Mutex
141- peer net.Addr
142- ctr uint64
143- seq uint64
144- probeSeq uint64
145- pending map[uint64]*pending
146- closed bool
147- closeErr error
148- lastHeard time.Time
149- heartbeat time.Duration
150- resendAfter time.Duration
151- maxResendAfter time.Duration
152- maxResendPerTick int
153- srtt time.Duration
154- rttvar time.Duration
155- rto time.Duration
156- writeTimeout time.Duration
157- degradedAfter time.Duration
158- probeAfter time.Duration
159- offlineAfter time.Duration
160- deadAfter time.Duration
161- maxPending int
162- maxPendingWait time.Duration
163- maxRecvBuffer int
164- clock ports.Clock
165- linkState ports.LinkState
166- linkEvents chan ports.LinkEvent
167- probeWait map[uint64]chan struct{}
168- sendWake chan struct{}
169- probeReply chan uint64
170- rebind func(net.PacketConn) (net.PacketConn, error)
171- hoppedOffline bool
172- outputQueue []queuedSend
173- outputWake chan struct{}
174- outputNext time.Time
142+ mu sync.Mutex
143+ peer net.Addr
144+ ctr uint64
145+ seq uint64
146+ probeSeq uint64
147+ pending map[uint64]*pending
148+ closed bool
149+ closeErr error
150+ lastHeard time.Time
151+ lastAuthenticatedPacket time.Time
152+ lastCompleteRecord time.Time
153+ lastACKProgress time.Time
154+ retransmits uint64
155+ reassemblyInflight int
156+ observe DiagnosticObserver
157+ diagnosticCh chan Diagnostic
158+ heartbeat time.Duration
159+ resendAfter time.Duration
160+ maxResendAfter time.Duration
161+ maxResendPerTick int
162+ srtt time.Duration
163+ rttvar time.Duration
164+ rto time.Duration
165+ writeTimeout time.Duration
166+ degradedAfter time.Duration
167+ probeAfter time.Duration
168+ offlineAfter time.Duration
169+ deadAfter time.Duration
170+ maxPending int
171+ maxPendingWait time.Duration
172+ maxRecvBuffer int
173+ clock ports.Clock
174+ linkState ports.LinkState
175+ linkEvents chan ports.LinkEvent
176+ probeWait map[uint64]chan struct{}
177+ sendWake chan struct{}
178+ probeReply chan uint64
179+ rebind func(net.PacketConn) (net.PacketConn, error)
180+ hoppedOffline bool
181+ outputQueue []queuedSend
182+ outputWake chan struct{}
183+ outputNext time.Time
175184
176185 recvMu sync.Mutex
177186 replay *pdgram.ReplayWindow
@@ -214,41 +223,49 @@ func NewTransportWithOptions(pc net.PacketConn, peer net.Addr, key []byte, sendD
214223 }
215224 opts = normalizeOptions(opts)
216225 t := &Transport{
217- pc: pc,
218- codec: c,
219- sendDir: sendDir,
220- recvDir: recvDir,
221- mtu: opts.MTU,
222- peer: peer,
223- pending: make(map[uint64]*pending),
224- replay: pdgram.NewReplayWindow(),
225- reasm: pdgram.NewReassembler(),
226- in: make(chan ports.Frame, 32),
227- done: make(chan struct{}),
228- lastHeard: opts.Clock.Now(),
229- heartbeat: opts.Heartbeat,
230- resendAfter: opts.ResendAfter,
231- maxResendAfter: opts.MaxResendAfter,
232- maxResendPerTick: opts.MaxResendPerTick,
233- rto: opts.ResendAfter,
234- writeTimeout: opts.WriteTimeout,
235- degradedAfter: opts.DegradedAfter,
236- probeAfter: opts.ProbeAfter,
237- offlineAfter: opts.OfflineAfter,
238- deadAfter: opts.DeadAfter,
239- maxPending: opts.MaxPending,
240- maxPendingWait: opts.MaxPendingWait,
241- maxRecvBuffer: opts.MaxRecvBuffer,
242- clock: opts.Clock,
243- linkState: ports.LinkStateConnected,
244- linkEvents: make(chan ports.LinkEvent, linkEventBufferSize),
245- probeWait: make(map[uint64]chan struct{}),
246- rebind: opts.RebindPacketConn,
247- nextRecvSeq: 1,
248- recvBuf: make(map[uint64]ports.Frame),
226+ pc: pc,
227+ codec: c,
228+ sendDir: sendDir,
229+ recvDir: recvDir,
230+ mtu: opts.MTU,
231+ peer: peer,
232+ pending: make(map[uint64]*pending),
233+ replay: pdgram.NewReplayWindow(),
234+ reasm: pdgram.NewReassembler(),
235+ in: make(chan ports.Frame, 32),
236+ done: make(chan struct{}),
237+ lastHeard: opts.Clock.Now(),
238+ lastAuthenticatedPacket: opts.Clock.Now(),
239+ lastCompleteRecord: opts.Clock.Now(),
240+ lastACKProgress: opts.Clock.Now(),
241+ observe: opts.Observe,
242+ heartbeat: opts.Heartbeat,
243+ resendAfter: opts.ResendAfter,
244+ maxResendAfter: opts.MaxResendAfter,
245+ maxResendPerTick: opts.MaxResendPerTick,
246+ rto: opts.ResendAfter,
247+ writeTimeout: opts.WriteTimeout,
248+ degradedAfter: opts.DegradedAfter,
249+ probeAfter: opts.ProbeAfter,
250+ offlineAfter: opts.OfflineAfter,
251+ deadAfter: opts.DeadAfter,
252+ maxPending: opts.MaxPending,
253+ maxPendingWait: opts.MaxPendingWait,
254+ maxRecvBuffer: opts.MaxRecvBuffer,
255+ clock: opts.Clock,
256+ linkState: ports.LinkStateConnected,
257+ linkEvents: make(chan ports.LinkEvent, linkEventBufferSize),
258+ probeWait: make(map[uint64]chan struct{}),
259+ rebind: opts.RebindPacketConn,
260+ nextRecvSeq: 1,
261+ recvBuf: make(map[uint64]ports.Frame),
249262 }
250263 t.sendWake = make(chan struct{})
251264 t.outputWake = make(chan struct{})
265+ if t.observe != nil {
266+ t.diagnosticCh = make(chan Diagnostic, diagnosticBufferSize)
267+ go t.diagnosticLoop()
268+ }
252269 t.probeReply = make(chan uint64, probeReplyBufferSize)
253270 t.deliverCond = sync.NewCond(&t.deliverMu)
254271 go t.readLoop(pc)
@@ -256,6 +273,7 @@ func NewTransportWithOptions(pc net.PacketConn, peer net.Addr, key []byte, sendD
256273 go t.deliveryLoop()
257274 go t.probeReplyLoop()
258275 go t.outputPaceLoop()
276+ t.emitDiagnostic()
259277 return t, nil
260278 }
261279
@@ -666,16 +684,32 @@ func (t *Transport) readLoop(pc net.PacketConn) {
666684 t.recvMu.Unlock()
667685 continue
668686 }
669- payload, ok, err := t.reasm.Add(frag)
687+ payload, complete, err := t.reasm.Add(frag)
688+ inflight := t.reasm.Inflight()
670689 t.recvMu.Unlock()
671- if err != nil || !ok {
690+ t.recordAuthenticatedFragment(inflight)
691+ if err != nil || !complete {
672692 continue
673693 }
694+ t.recordCompleteRecord()
674695 t.updatePeerFromAuthenticated(addr)
675696 t.handleRecord(payload)
676697 }
677698 }
678699
700+func (t *Transport) recordAuthenticatedFragment(inflight int) {
701+ t.mu.Lock()
702+ t.lastAuthenticatedPacket = t.clock.Now()
703+ t.reassemblyInflight = inflight
704+ t.mu.Unlock()
705+}
706+
707+func (t *Transport) recordCompleteRecord() {
708+ t.mu.Lock()
709+ t.lastCompleteRecord = t.clock.Now()
710+ t.mu.Unlock()
711+}
712+
679713 func (t *Transport) updatePeerFromAuthenticated(addr net.Addr) {
680714 t.mu.Lock()
681715 t.peer = addr
@@ -708,9 +742,13 @@ func (t *Transport) handleRecord(p []byte) {
708742 acked = true
709743 }
710744 if acked {
745+ t.lastACKProgress = t.clock.Now()
711746 t.notifySendWaitersLocked()
712747 }
713748 t.mu.Unlock()
749+ if acked {
750+ t.emitDiagnostic()
751+ }
714752 case recProbe:
715753 if len(p) != 9 {
716754 return
@@ -871,6 +909,7 @@ func (t *Transport) resendPending() {
871909 }
872910 slices.Sort(seqs)
873911 limit := t.maxResendPerTick
912+ emitDiagnostic := false
874913 for _, seq := range seqs {
875914 if limit <= 0 {
876915 break
@@ -882,6 +921,8 @@ func (t *Transport) resendPending() {
882921 p.initialInFlight = true
883922 p.attempts++
884923 p.retransmitted = true
924+ t.retransmits++
925+ emitDiagnostic = emitDiagnostic || t.retransmits%64 == 0
885926 resend = append(resend, struct {
886927 seq uint64
887928 f ports.Frame
@@ -889,6 +930,9 @@ func (t *Transport) resendPending() {
889930 limit--
890931 }
891932 t.mu.Unlock()
933+ if emitDiagnostic {
934+ t.emitDiagnostic()
935+ }
892936 for _, r := range resend {
893937 if err := t.sendData(r.seq, true, r.f); err == nil {
894938 t.markPendingReady(r.seq, true)
@@ -1057,6 +1101,7 @@ func (t *Transport) setLinkState(state ports.LinkState, err error) {
10571101 }
10581102 t.notifySendWaitersLocked()
10591103 t.mu.Unlock()
1104+ t.emitDiagnostic()
10601105 event := ports.LinkEvent{State: state, At: now, Err: err}
10611106 select {
10621107 case t.linkEvents <- event:
@@ -1064,6 +1109,47 @@ func (t *Transport) setLinkState(state ports.LinkState, err error) {
10641109 }
10651110 }
10661111
1112+func (t *Transport) diagnosticLoop() {
1113+ for {
1114+ select {
1115+ case d := <-t.diagnosticCh:
1116+ t.observe(d)
1117+ case <-t.done:
1118+ return
1119+ }
1120+ }
1121+}
1122+
1123+func (t *Transport) emitDiagnostic() {
1124+ if t.diagnosticCh == nil {
1125+ return
1126+ }
1127+ now := t.clock.Now()
1128+ t.mu.Lock()
1129+ pendingBytes := 0
1130+ for _, p := range t.pending {
1131+ if p != nil {
1132+ pendingBytes += len(p.frame.Payload)
1133+ }
1134+ }
1135+ d := Diagnostic{
1136+ At: now,
1137+ State: t.linkState,
1138+ SinceAuthenticatedPacket: now.Sub(t.lastAuthenticatedPacket),
1139+ SinceCompleteRecord: now.Sub(t.lastCompleteRecord),
1140+ SinceACKProgress: now.Sub(t.lastACKProgress),
1141+ PendingRecords: len(t.pending),
1142+ PendingBytes: pendingBytes,
1143+ Retransmits: t.retransmits,
1144+ ReassemblyInflight: t.reassemblyInflight,
1145+ }
1146+ t.mu.Unlock()
1147+ select {
1148+ case t.diagnosticCh <- d:
1149+ default:
1150+ }
1151+}
1152+
10671153 func (t *Transport) closeWithError(err error) {
10681154 t.mu.Lock()
10691155 if t.closed {
@@ -64,6 +64,8 @@ type Options struct {
64 MaxPendingWait time.Duration64 MaxPendingWait time.Duration
65 MaxRecvBuffer int65 MaxRecvBuffer int
66 Clock ports.Clock66 Clock ports.Clock
67+ // Observe receives bounded, asynchronous, privacy-safe link health snapshots.
68+ Observe DiagnosticObserver
67 // RebindPacketConn is an optional client-side hook used to hop the local UDP69 // RebindPacketConn is an optional client-side hook used to hop the local UDP
68 // socket when the peer is offline. Server/proxy transports should leave it nil.70 // socket when the peer is offline. Server/proxy transports should leave it nil.
69 RebindPacketConn func(net.PacketConn) (net.PacketConn, error)71 RebindPacketConn func(net.PacketConn) (net.PacketConn, error)
@@ -137,41 +139,48 @@ type Transport struct {
137 sendDir, recvDir uint32139 sendDir, recvDir uint32
138 mtu int140 mtu int
139 141
140- mu sync.Mutex142+ mu sync.Mutex
141- peer net.Addr143+ peer net.Addr
142- ctr uint64144+ ctr uint64
143- seq uint64145+ seq uint64
144- probeSeq uint64146+ probeSeq uint64
145- pending map[uint64]*pending147+ pending map[uint64]*pending
146- closed bool148+ closed bool
147- closeErr error149+ closeErr error
148- lastHeard time.Time150+ lastHeard time.Time
149- heartbeat time.Duration151+ lastAuthenticatedPacket time.Time
150- resendAfter time.Duration152+ lastCompleteRecord time.Time
151- maxResendAfter time.Duration153+ lastACKProgress time.Time
152- maxResendPerTick int154+ retransmits uint64
153- srtt time.Duration155+ reassemblyInflight int
154- rttvar time.Duration156+ observe DiagnosticObserver
155- rto time.Duration157+ diagnosticCh chan Diagnostic
156- writeTimeout time.Duration158+ heartbeat time.Duration
157- degradedAfter time.Duration159+ resendAfter time.Duration
158- probeAfter time.Duration160+ maxResendAfter time.Duration
159- offlineAfter time.Duration161+ maxResendPerTick int
160- deadAfter time.Duration162+ srtt time.Duration
161- maxPending int163+ rttvar time.Duration
162- maxPendingWait time.Duration164+ rto time.Duration
163- maxRecvBuffer int165+ writeTimeout time.Duration
164- clock ports.Clock166+ degradedAfter time.Duration
165- linkState ports.LinkState167+ probeAfter time.Duration
166- linkEvents chan ports.LinkEvent168+ offlineAfter time.Duration
167- probeWait map[uint64]chan struct{}169+ deadAfter time.Duration
168- sendWake chan struct{}170+ maxPending int
169- probeReply chan uint64171+ maxPendingWait time.Duration
170- rebind func(net.PacketConn) (net.PacketConn, error)172+ maxRecvBuffer int
171- hoppedOffline bool173+ clock ports.Clock
172- outputQueue []queuedSend174+ linkState ports.LinkState
173- outputWake chan struct{}175+ linkEvents chan ports.LinkEvent
174- outputNext time.Time176+ probeWait map[uint64]chan struct{}
177+ sendWake chan struct{}
178+ probeReply chan uint64
179+ rebind func(net.PacketConn) (net.PacketConn, error)
180+ hoppedOffline bool
181+ outputQueue []queuedSend
182+ outputWake chan struct{}
183+ outputNext time.Time
175 184
176 recvMu sync.Mutex185 recvMu sync.Mutex
177 replay *pdgram.ReplayWindow186 replay *pdgram.ReplayWindow
@@ -214,41 +223,49 @@ func NewTransportWithOptions(pc net.PacketConn, peer net.Addr, key []byte, sendD
214 }223 }
215 opts = normalizeOptions(opts)224 opts = normalizeOptions(opts)
216 t := &Transport{225 t := &Transport{
217- pc: pc,226+ pc: pc,
218- codec: c,227+ codec: c,
219- sendDir: sendDir,228+ sendDir: sendDir,
220- recvDir: recvDir,229+ recvDir: recvDir,
221- mtu: opts.MTU,230+ mtu: opts.MTU,
222- peer: peer,231+ peer: peer,
223- pending: make(map[uint64]*pending),232+ pending: make(map[uint64]*pending),
224- replay: pdgram.NewReplayWindow(),233+ replay: pdgram.NewReplayWindow(),
225- reasm: pdgram.NewReassembler(),234+ reasm: pdgram.NewReassembler(),
226- in: make(chan ports.Frame, 32),235+ in: make(chan ports.Frame, 32),
227- done: make(chan struct{}),236+ done: make(chan struct{}),
228- lastHeard: opts.Clock.Now(),237+ lastHeard: opts.Clock.Now(),
229- heartbeat: opts.Heartbeat,238+ lastAuthenticatedPacket: opts.Clock.Now(),
230- resendAfter: opts.ResendAfter,239+ lastCompleteRecord: opts.Clock.Now(),
231- maxResendAfter: opts.MaxResendAfter,240+ lastACKProgress: opts.Clock.Now(),
232- maxResendPerTick: opts.MaxResendPerTick,241+ observe: opts.Observe,
233- rto: opts.ResendAfter,242+ heartbeat: opts.Heartbeat,
234- writeTimeout: opts.WriteTimeout,243+ resendAfter: opts.ResendAfter,
235- degradedAfter: opts.DegradedAfter,244+ maxResendAfter: opts.MaxResendAfter,
236- probeAfter: opts.ProbeAfter,245+ maxResendPerTick: opts.MaxResendPerTick,
237- offlineAfter: opts.OfflineAfter,246+ rto: opts.ResendAfter,
238- deadAfter: opts.DeadAfter,247+ writeTimeout: opts.WriteTimeout,
239- maxPending: opts.MaxPending,248+ degradedAfter: opts.DegradedAfter,
240- maxPendingWait: opts.MaxPendingWait,249+ probeAfter: opts.ProbeAfter,
241- maxRecvBuffer: opts.MaxRecvBuffer,250+ offlineAfter: opts.OfflineAfter,
242- clock: opts.Clock,251+ deadAfter: opts.DeadAfter,
243- linkState: ports.LinkStateConnected,252+ maxPending: opts.MaxPending,
244- linkEvents: make(chan ports.LinkEvent, linkEventBufferSize),253+ maxPendingWait: opts.MaxPendingWait,
245- probeWait: make(map[uint64]chan struct{}),254+ maxRecvBuffer: opts.MaxRecvBuffer,
246- rebind: opts.RebindPacketConn,255+ clock: opts.Clock,
247- nextRecvSeq: 1,256+ linkState: ports.LinkStateConnected,
248- recvBuf: make(map[uint64]ports.Frame),257+ linkEvents: make(chan ports.LinkEvent, linkEventBufferSize),
258+ probeWait: make(map[uint64]chan struct{}),
259+ rebind: opts.RebindPacketConn,
260+ nextRecvSeq: 1,
261+ recvBuf: make(map[uint64]ports.Frame),
249 }262 }
250 t.sendWake = make(chan struct{})263 t.sendWake = make(chan struct{})
251 t.outputWake = make(chan struct{})264 t.outputWake = make(chan struct{})
265+ if t.observe != nil {
266+ t.diagnosticCh = make(chan Diagnostic, diagnosticBufferSize)
267+ go t.diagnosticLoop()
268+ }
252 t.probeReply = make(chan uint64, probeReplyBufferSize)269 t.probeReply = make(chan uint64, probeReplyBufferSize)
253 t.deliverCond = sync.NewCond(&t.deliverMu)270 t.deliverCond = sync.NewCond(&t.deliverMu)
254 go t.readLoop(pc)271 go t.readLoop(pc)
@@ -256,6 +273,7 @@ func NewTransportWithOptions(pc net.PacketConn, peer net.Addr, key []byte, sendD
256 go t.deliveryLoop()273 go t.deliveryLoop()
257 go t.probeReplyLoop()274 go t.probeReplyLoop()
258 go t.outputPaceLoop()275 go t.outputPaceLoop()
276+ t.emitDiagnostic()
259 return t, nil277 return t, nil
260 }278 }
261 279
@@ -666,16 +684,32 @@ func (t *Transport) readLoop(pc net.PacketConn) {
666 t.recvMu.Unlock()684 t.recvMu.Unlock()
667 continue685 continue
668 }686 }
669- payload, ok, err := t.reasm.Add(frag)687+ payload, complete, err := t.reasm.Add(frag)
688+ inflight := t.reasm.Inflight()
670 t.recvMu.Unlock()689 t.recvMu.Unlock()
671- if err != nil || !ok {690+ t.recordAuthenticatedFragment(inflight)
691+ if err != nil || !complete {
672 continue692 continue
673 }693 }
694+ t.recordCompleteRecord()
674 t.updatePeerFromAuthenticated(addr)695 t.updatePeerFromAuthenticated(addr)
675 t.handleRecord(payload)696 t.handleRecord(payload)
676 }697 }
677 }698 }
678 699
700+func (t *Transport) recordAuthenticatedFragment(inflight int) {
701+ t.mu.Lock()
702+ t.lastAuthenticatedPacket = t.clock.Now()
703+ t.reassemblyInflight = inflight
704+ t.mu.Unlock()
705+}
706+
707+func (t *Transport) recordCompleteRecord() {
708+ t.mu.Lock()
709+ t.lastCompleteRecord = t.clock.Now()
710+ t.mu.Unlock()
711+}
712+
679 func (t *Transport) updatePeerFromAuthenticated(addr net.Addr) {713 func (t *Transport) updatePeerFromAuthenticated(addr net.Addr) {
680 t.mu.Lock()714 t.mu.Lock()
681 t.peer = addr715 t.peer = addr
@@ -708,9 +742,13 @@ func (t *Transport) handleRecord(p []byte) {
708 acked = true742 acked = true
709 }743 }
710 if acked {744 if acked {
745+ t.lastACKProgress = t.clock.Now()
711 t.notifySendWaitersLocked()746 t.notifySendWaitersLocked()
712 }747 }
713 t.mu.Unlock()748 t.mu.Unlock()
749+ if acked {
750+ t.emitDiagnostic()
751+ }
714 case recProbe:752 case recProbe:
715 if len(p) != 9 {753 if len(p) != 9 {
716 return754 return
@@ -871,6 +909,7 @@ func (t *Transport) resendPending() {
871 }909 }
872 slices.Sort(seqs)910 slices.Sort(seqs)
873 limit := t.maxResendPerTick911 limit := t.maxResendPerTick
912+ emitDiagnostic := false
874 for _, seq := range seqs {913 for _, seq := range seqs {
875 if limit <= 0 {914 if limit <= 0 {
876 break915 break
@@ -882,6 +921,8 @@ func (t *Transport) resendPending() {
882 p.initialInFlight = true921 p.initialInFlight = true
883 p.attempts++922 p.attempts++
884 p.retransmitted = true923 p.retransmitted = true
924+ t.retransmits++
925+ emitDiagnostic = emitDiagnostic || t.retransmits%64 == 0
885 resend = append(resend, struct {926 resend = append(resend, struct {
886 seq uint64927 seq uint64
887 f ports.Frame928 f ports.Frame
@@ -889,6 +930,9 @@ func (t *Transport) resendPending() {
889 limit--930 limit--
890 }931 }
891 t.mu.Unlock()932 t.mu.Unlock()
933+ if emitDiagnostic {
934+ t.emitDiagnostic()
935+ }
892 for _, r := range resend {936 for _, r := range resend {
893 if err := t.sendData(r.seq, true, r.f); err == nil {937 if err := t.sendData(r.seq, true, r.f); err == nil {
894 t.markPendingReady(r.seq, true)938 t.markPendingReady(r.seq, true)
@@ -1057,6 +1101,7 @@ func (t *Transport) setLinkState(state ports.LinkState, err error) {
1057 }1101 }
1058 t.notifySendWaitersLocked()1102 t.notifySendWaitersLocked()
1059 t.mu.Unlock()1103 t.mu.Unlock()
1104+ t.emitDiagnostic()
1060 event := ports.LinkEvent{State: state, At: now, Err: err}1105 event := ports.LinkEvent{State: state, At: now, Err: err}
1061 select {1106 select {
1062 case t.linkEvents <- event:1107 case t.linkEvents <- event:
@@ -1064,6 +1109,47 @@ func (t *Transport) setLinkState(state ports.LinkState, err error) {
1064 }1109 }
1065 }1110 }
1066 1111
1112+func (t *Transport) diagnosticLoop() {
1113+ for {
1114+ select {
1115+ case d := <-t.diagnosticCh:
1116+ t.observe(d)
1117+ case <-t.done:
1118+ return
1119+ }
1120+ }
1121+}
1122+
1123+func (t *Transport) emitDiagnostic() {
1124+ if t.diagnosticCh == nil {
1125+ return
1126+ }
1127+ now := t.clock.Now()
1128+ t.mu.Lock()
1129+ pendingBytes := 0
1130+ for _, p := range t.pending {
1131+ if p != nil {
1132+ pendingBytes += len(p.frame.Payload)
1133+ }
1134+ }
1135+ d := Diagnostic{
1136+ At: now,
1137+ State: t.linkState,
1138+ SinceAuthenticatedPacket: now.Sub(t.lastAuthenticatedPacket),
1139+ SinceCompleteRecord: now.Sub(t.lastCompleteRecord),
1140+ SinceACKProgress: now.Sub(t.lastACKProgress),
1141+ PendingRecords: len(t.pending),
1142+ PendingBytes: pendingBytes,
1143+ Retransmits: t.retransmits,
1144+ ReassemblyInflight: t.reassemblyInflight,
1145+ }
1146+ t.mu.Unlock()
1147+ select {
1148+ case t.diagnosticCh <- d:
1149+ default:
1150+ }
1151+}
1152+
1067 func (t *Transport) closeWithError(err error) {1153 func (t *Transport) closeWithError(err error) {
1068 t.mu.Lock()1154 t.mu.Lock()
1069 if t.closed {1155 if t.closed {
M internal/app/run.go +3 -1
@@ -745,7 +745,9 @@ func runUDPProxy(ctx context.Context, session string, ready io.Writer) error {
745745 return err
746746 }
747747 defer func() { _ = daemonTr.Close() }()
748- dg, err := dgram.NewTransportWithOptions(conn, nil, key, 2, 1, udpProxyClientTransportOptions)
748+ udpOptions := udpProxyClientTransportOptions
749+ udpOptions.Observe = dgram.DiagnosticLogObserver(log)
750+ dg, err := dgram.NewTransportWithOptions(conn, nil, key, 2, 1, udpOptions)
749751 if err != nil {
750752 return err
751753 }
@@ -745,7 +745,9 @@ func runUDPProxy(ctx context.Context, session string, ready io.Writer) error {
745 return err745 return err
746 }746 }
747 defer func() { _ = daemonTr.Close() }()747 defer func() { _ = daemonTr.Close() }()
748- dg, err := dgram.NewTransportWithOptions(conn, nil, key, 2, 1, udpProxyClientTransportOptions)748+ udpOptions := udpProxyClientTransportOptions
749+ udpOptions.Observe = dgram.DiagnosticLogObserver(log)
750+ dg, err := dgram.NewTransportWithOptions(conn, nil, key, 2, 1, udpOptions)
749 if err != nil {751 if err != nil {
750 return err752 return err
751 }753 }
M pkg/dgram/dgram.go +4 -0
@@ -191,6 +191,10 @@ type assembly struct {
191191 }
192192
193193 func NewReassembler() *Reassembler { return &Reassembler{inflight: make(map[uint64]*assembly)} }
194+
195+// Inflight returns the number of incomplete records currently being reassembled.
196+func (r *Reassembler) Inflight() int { return len(r.inflight) }
197+
194198 func (r *Reassembler) Add(f Fragment) ([]byte, bool, error) {
195199 if f.Count == 0 || int(f.Count) > maxFragmentCount || f.Index >= f.Count {
196200 return nil, false, ErrFragment
@@ -191,6 +191,10 @@ type assembly struct {
191 }191 }
192 192
193 func NewReassembler() *Reassembler { return &Reassembler{inflight: make(map[uint64]*assembly)} }193 func NewReassembler() *Reassembler { return &Reassembler{inflight: make(map[uint64]*assembly)} }
194+
195+// Inflight returns the number of incomplete records currently being reassembled.
196+func (r *Reassembler) Inflight() int { return len(r.inflight) }
197+
194 func (r *Reassembler) Add(f Fragment) ([]byte, bool, error) {198 func (r *Reassembler) Add(f Fragment) ([]byte, bool, error) {
195 if f.Count == 0 || int(f.Count) > maxFragmentCount || f.Index >= f.Count {199 if f.Count == 0 || int(f.Count) > maxFragmentCount || f.Index >= f.Count {
196 return nil, false, ErrFragment200 return nil, false, ErrFragment

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