detached d626c6d

refactor(dgram): split transport health and diagnostics

commit
d626c6dc4efedf461946741531bd63acb3901968
author
brice · · 2026-07-14 20:49 UTC
parent
3b6ed69
signature
GPG VerifiedCD88F9E564E0102646FB6077A1A07C7C7C3E349D · arch-gpg Verified for the committer email
3 files · +318 -304
Unified
UnifiedSplit
M internal/adapters/dgram/transport.go +0 -304
@@ -386,309 +386,5 @@ func (t *Transport) pendingExists(seq uint64) bool {
386386 return ok
387387 }
388388
389-// healthLoop owns link-health timing so bulk retransmission cannot delay state
390-// transitions or socket hopping.
391-func (t *Transport) healthLoop() {
392- health := t.clock.NewTimer(t.resendAfter)
393- defer health.Stop()
394- for {
395- select {
396- case <-health.C():
397- t.checkSilence()
398- health.Reset(t.resendAfter)
399- t.notifyTimerHook(func() func() { return t.afterHealthTimer })
400- case <-t.done:
401- return
402- }
403- }
404-}
405-
406-func (t *Transport) heartbeatLoop() {
407- heartbeat := t.clock.NewTimer(t.heartbeat)
408- defer heartbeat.Stop()
409- for {
410- select {
411- case <-heartbeat.C():
412- t.queueControl(recProbe, 0)
413- heartbeat.Reset(t.heartbeat)
414- t.notifyTimerHook(func() func() { return t.afterHeartbeatTimer })
415- case <-t.done:
416- return
417- }
418- }
419-}
420-
421-func (t *Transport) notifyTimerHook(get func() func()) {
422- t.mu.Lock()
423- hook := get()
424- t.mu.Unlock()
425- if hook != nil {
426- hook()
427- }
428-}
429-
430-func (t *Transport) hopPacketConnOnce(generation uint64) {
431- t.mu.Lock()
432- if t.rebind == nil || t.hoppedOffline || t.closed || t.linkState != ports.LinkStateProbing || t.health.generation != generation {
433- t.mu.Unlock()
434- return
435- }
436- old := t.pc
437- rebind := t.rebind
438- t.hoppedOffline = true
439- t.hopGeneration = generation
440- t.mu.Unlock()
441- pc, err := rebind(old)
442-
443- // Serialize the swap with writes. writePacket takes writeMu before mu, so
444- // preserve that order here while retiring old.
445- t.writeMu.Lock()
446- defer t.writeMu.Unlock()
447- t.mu.Lock()
448- if err != nil {
449- t.rollbackHopLocked(generation)
450- t.mu.Unlock()
451- return
452- }
453- if t.closed || t.pc != old || t.health.generation != generation || t.linkState != ports.LinkStateProbing {
454- t.rollbackHopLocked(generation)
455- t.mu.Unlock()
456- _ = pc.Close()
457- return
458- }
459- t.pc = pc
460- t.writeDeadlines = newWriteDeadlineState(pc)
461- t.mu.Unlock()
462- go t.readLoop(pc)
463- _ = old.Close()
464-}
465-
466-func (t *Transport) rollbackHopLocked(generation uint64) {
467- if t.hoppedOffline && t.hopGeneration == generation {
468- t.hoppedOffline = false
469- t.hopGeneration = 0
470- }
471-}
472-
473-func (t *Transport) checkSilence() {
474- runHook := true
475- for {
476- t.mu.Lock()
477- if t.closed {
478- t.mu.Unlock()
479- return
480- }
481- now := t.clock.Now()
482- generation := t.health.generation
483- state, hop, dead := t.health.decide(now, len(t.pending) > 0, t.degradedAfter, t.probeAfter, t.offlineAfter, t.deadAfter)
484- afterDecision := t.afterHealthDecision
485- t.mu.Unlock()
486-
487- if runHook {
488- runHook = false
489- if afterDecision != nil {
490- afterDecision()
491- }
492- }
493-
494- t.mu.Lock()
495- if t.closed {
496- t.mu.Unlock()
497- return
498- }
499- if t.health.generation != generation {
500- t.mu.Unlock()
501- continue
502- }
503- changed, stateGeneration := t.setLinkStateLocked(state)
504- afterStateCommit := t.afterLinkStateCommit
505- closed := false
506- if dead {
507- closed = t.closeWithErrorLocked(ErrLinkDead)
508- }
509- t.mu.Unlock()
510-
511- if changed {
512- if afterStateCommit != nil {
513- afterStateCommit(state)
514- }
515- var linkErr error
516- if dead {
517- linkErr = ErrLinkDead
518- }
519- t.publishLinkState(state, now, linkErr, stateGeneration)
520- }
521- if closed {
522- t.broadcastClosed()
523- return
524- }
525- if hop {
526- t.hopPacketConnOnce(generation)
527- }
528- return
529- }
530-}
531-
532-func (t *Transport) setLinkState(state ports.LinkState, err error) {
533- now := t.clock.Now()
534- t.mu.Lock()
535- changed, stateGeneration := t.setLinkStateLocked(state)
536- afterStateCommit := t.afterLinkStateCommit
537- t.mu.Unlock()
538- if changed {
539- if afterStateCommit != nil {
540- afterStateCommit(state)
541- }
542- t.publishLinkState(state, now, err, stateGeneration)
543- }
544-}
545-
546-func (t *Transport) setLinkStateLocked(state ports.LinkState) (bool, uint64) {
547- if t.linkState == state {
548- return false, t.linkStateGeneration
549- }
550- t.linkState = state
551- t.linkStateGeneration++
552- if state == ports.LinkStateConnected {
553- t.hoppedOffline = false
554- t.hopGeneration = 0
555- for _, p := range t.pending {
556- if p != nil {
557- p.attempts = 0
558- }
559- }
560- }
561- t.notifySendWaitersLocked()
562- return true, t.linkStateGeneration
563-}
564-
565-func (t *Transport) publishLinkState(state ports.LinkState, now time.Time, err error, generation uint64) {
566- t.linkEventMu.Lock()
567- defer t.linkEventMu.Unlock()
568- t.mu.Lock()
569- if t.linkState != state || t.linkStateGeneration != generation {
570- t.mu.Unlock()
571- return
572- }
573- event := ports.LinkEvent{State: state, At: now, Err: err}
574- select {
575- case t.linkEvents <- event:
576- default:
577- }
578- t.mu.Unlock()
579- t.emitDiagnostic()
580-}
581-
582-func (t *Transport) diagnosticLoop() {
583- for {
584- select {
585- case d := <-t.diagnosticCh:
586- t.observe(d)
587- case <-t.done:
588- return
589- }
590- }
591-}
592-
593-func (t *Transport) emitDiagnostic() {
594- if t.diagnosticCh == nil {
595- return
596- }
597- d := t.diagnosticSnapshot()
598- select {
599- case t.diagnosticCh <- d:
600- default:
601- }
602-}
603-
604-func (t *Transport) diagnosticSnapshot() Diagnostic {
605- t.mu.Lock()
606- now := t.clock.Now()
607- lastAuthenticatedPacket := t.lastAuthenticatedPacket
608- lastCompleteRecord := t.lastCompleteRecord
609- lastACKProgress := t.lastACKProgress
610- pendingBytes := 0
611- for _, p := range t.pending {
612- if p != nil {
613- pendingBytes += len(p.frame.Payload)
614- }
615- }
616- state := t.linkState
617- pendingRecords := len(t.pending)
618- retransmits := t.retransmits
619- reassemblyInflight := t.reassemblyInflight
620- t.mu.Unlock()
621-
622- return Diagnostic{
623- At: now,
624- State: state,
625- SinceAuthenticatedPacket: diagnosticAge(now, lastAuthenticatedPacket),
626- SinceCompleteRecord: diagnosticAge(now, lastCompleteRecord),
627- SinceACKProgress: diagnosticAge(now, lastACKProgress),
628- PendingRecords: pendingRecords,
629- PendingBytes: pendingBytes,
630- Retransmits: retransmits,
631- ReassemblyInflight: reassemblyInflight,
632- }
633-}
634-
635-func diagnosticAge(now, then time.Time) time.Duration {
636- if then.After(now) {
637- return 0
638- }
639- return now.Sub(then)
640-}
641-
642389 // DatagramTransport marks Transport as a UDP-style datagram transport.
643-func (t *Transport) beginRuntimeOperation(start ports.RuntimeMarkKind, bytes uint64) func(bool) {
644- if t.runtimeObserver == nil {
645- return func(bool) {}
646- }
647- if !t.beginObservedOperation() {
648- return func(bool) {}
649- }
650- correlation := ports.NewRuntimeCorrelation()
651- t.runtimeObserver.ObserveRuntime(ports.NewRuntimeMarkWithCorrelation("dgram", correlation, start, bytes, true))
652- end := ports.RuntimeAdapterSendEnd
653- if start == ports.RuntimeAdapterReceiveStart {
654- end = ports.RuntimeAdapterReceiveEnd
655- }
656- return func(valid bool) {
657- defer t.finishObservedOperation()
658- t.runtimeObserver.ObserveRuntime(ports.NewRuntimeMarkWithCorrelation("dgram", correlation, end, bytes, valid))
659- }
660-}
661-
662-func (t *Transport) beginObservedOperation() bool {
663- t.operationMu.Lock()
664- defer t.operationMu.Unlock()
665- if t.closing {
666- return false
667- }
668- if t.operationCount == 0 {
669- t.operationsDone = make(chan struct{})
670- }
671- t.operationCount++
672- return true
673-}
674-
675-func (t *Transport) finishObservedOperation() {
676- t.operationMu.Lock()
677- defer t.operationMu.Unlock()
678- t.operationCount--
679- if t.operationCount == 0 {
680- close(t.operationsDone)
681- }
682-}
683-
684-func (t *Transport) beginShutdown() <-chan struct{} {
685- t.operationMu.Lock()
686- defer t.operationMu.Unlock()
687- t.closing = true
688- if t.operationCount == 0 {
689- return nil
690- }
691- return t.operationsDone
692-}
693-
694390 func (*Transport) DatagramTransport() {}
@@ -386,309 +386,5 @@ func (t *Transport) pendingExists(seq uint64) bool {
386 return ok386 return ok
387 }387 }
388 388
389-// healthLoop owns link-health timing so bulk retransmission cannot delay state
390-// transitions or socket hopping.
391-func (t *Transport) healthLoop() {
392- health := t.clock.NewTimer(t.resendAfter)
393- defer health.Stop()
394- for {
395- select {
396- case <-health.C():
397- t.checkSilence()
398- health.Reset(t.resendAfter)
399- t.notifyTimerHook(func() func() { return t.afterHealthTimer })
400- case <-t.done:
401- return
402- }
403- }
404-}
405-
406-func (t *Transport) heartbeatLoop() {
407- heartbeat := t.clock.NewTimer(t.heartbeat)
408- defer heartbeat.Stop()
409- for {
410- select {
411- case <-heartbeat.C():
412- t.queueControl(recProbe, 0)
413- heartbeat.Reset(t.heartbeat)
414- t.notifyTimerHook(func() func() { return t.afterHeartbeatTimer })
415- case <-t.done:
416- return
417- }
418- }
419-}
420-
421-func (t *Transport) notifyTimerHook(get func() func()) {
422- t.mu.Lock()
423- hook := get()
424- t.mu.Unlock()
425- if hook != nil {
426- hook()
427- }
428-}
429-
430-func (t *Transport) hopPacketConnOnce(generation uint64) {
431- t.mu.Lock()
432- if t.rebind == nil || t.hoppedOffline || t.closed || t.linkState != ports.LinkStateProbing || t.health.generation != generation {
433- t.mu.Unlock()
434- return
435- }
436- old := t.pc
437- rebind := t.rebind
438- t.hoppedOffline = true
439- t.hopGeneration = generation
440- t.mu.Unlock()
441- pc, err := rebind(old)
442-
443- // Serialize the swap with writes. writePacket takes writeMu before mu, so
444- // preserve that order here while retiring old.
445- t.writeMu.Lock()
446- defer t.writeMu.Unlock()
447- t.mu.Lock()
448- if err != nil {
449- t.rollbackHopLocked(generation)
450- t.mu.Unlock()
451- return
452- }
453- if t.closed || t.pc != old || t.health.generation != generation || t.linkState != ports.LinkStateProbing {
454- t.rollbackHopLocked(generation)
455- t.mu.Unlock()
456- _ = pc.Close()
457- return
458- }
459- t.pc = pc
460- t.writeDeadlines = newWriteDeadlineState(pc)
461- t.mu.Unlock()
462- go t.readLoop(pc)
463- _ = old.Close()
464-}
465-
466-func (t *Transport) rollbackHopLocked(generation uint64) {
467- if t.hoppedOffline && t.hopGeneration == generation {
468- t.hoppedOffline = false
469- t.hopGeneration = 0
470- }
471-}
472-
473-func (t *Transport) checkSilence() {
474- runHook := true
475- for {
476- t.mu.Lock()
477- if t.closed {
478- t.mu.Unlock()
479- return
480- }
481- now := t.clock.Now()
482- generation := t.health.generation
483- state, hop, dead := t.health.decide(now, len(t.pending) > 0, t.degradedAfter, t.probeAfter, t.offlineAfter, t.deadAfter)
484- afterDecision := t.afterHealthDecision
485- t.mu.Unlock()
486-
487- if runHook {
488- runHook = false
489- if afterDecision != nil {
490- afterDecision()
491- }
492- }
493-
494- t.mu.Lock()
495- if t.closed {
496- t.mu.Unlock()
497- return
498- }
499- if t.health.generation != generation {
500- t.mu.Unlock()
501- continue
502- }
503- changed, stateGeneration := t.setLinkStateLocked(state)
504- afterStateCommit := t.afterLinkStateCommit
505- closed := false
506- if dead {
507- closed = t.closeWithErrorLocked(ErrLinkDead)
508- }
509- t.mu.Unlock()
510-
511- if changed {
512- if afterStateCommit != nil {
513- afterStateCommit(state)
514- }
515- var linkErr error
516- if dead {
517- linkErr = ErrLinkDead
518- }
519- t.publishLinkState(state, now, linkErr, stateGeneration)
520- }
521- if closed {
522- t.broadcastClosed()
523- return
524- }
525- if hop {
526- t.hopPacketConnOnce(generation)
527- }
528- return
529- }
530-}
531-
532-func (t *Transport) setLinkState(state ports.LinkState, err error) {
533- now := t.clock.Now()
534- t.mu.Lock()
535- changed, stateGeneration := t.setLinkStateLocked(state)
536- afterStateCommit := t.afterLinkStateCommit
537- t.mu.Unlock()
538- if changed {
539- if afterStateCommit != nil {
540- afterStateCommit(state)
541- }
542- t.publishLinkState(state, now, err, stateGeneration)
543- }
544-}
545-
546-func (t *Transport) setLinkStateLocked(state ports.LinkState) (bool, uint64) {
547- if t.linkState == state {
548- return false, t.linkStateGeneration
549- }
550- t.linkState = state
551- t.linkStateGeneration++
552- if state == ports.LinkStateConnected {
553- t.hoppedOffline = false
554- t.hopGeneration = 0
555- for _, p := range t.pending {
556- if p != nil {
557- p.attempts = 0
558- }
559- }
560- }
561- t.notifySendWaitersLocked()
562- return true, t.linkStateGeneration
563-}
564-
565-func (t *Transport) publishLinkState(state ports.LinkState, now time.Time, err error, generation uint64) {
566- t.linkEventMu.Lock()
567- defer t.linkEventMu.Unlock()
568- t.mu.Lock()
569- if t.linkState != state || t.linkStateGeneration != generation {
570- t.mu.Unlock()
571- return
572- }
573- event := ports.LinkEvent{State: state, At: now, Err: err}
574- select {
575- case t.linkEvents <- event:
576- default:
577- }
578- t.mu.Unlock()
579- t.emitDiagnostic()
580-}
581-
582-func (t *Transport) diagnosticLoop() {
583- for {
584- select {
585- case d := <-t.diagnosticCh:
586- t.observe(d)
587- case <-t.done:
588- return
589- }
590- }
591-}
592-
593-func (t *Transport) emitDiagnostic() {
594- if t.diagnosticCh == nil {
595- return
596- }
597- d := t.diagnosticSnapshot()
598- select {
599- case t.diagnosticCh <- d:
600- default:
601- }
602-}
603-
604-func (t *Transport) diagnosticSnapshot() Diagnostic {
605- t.mu.Lock()
606- now := t.clock.Now()
607- lastAuthenticatedPacket := t.lastAuthenticatedPacket
608- lastCompleteRecord := t.lastCompleteRecord
609- lastACKProgress := t.lastACKProgress
610- pendingBytes := 0
611- for _, p := range t.pending {
612- if p != nil {
613- pendingBytes += len(p.frame.Payload)
614- }
615- }
616- state := t.linkState
617- pendingRecords := len(t.pending)
618- retransmits := t.retransmits
619- reassemblyInflight := t.reassemblyInflight
620- t.mu.Unlock()
621-
622- return Diagnostic{
623- At: now,
624- State: state,
625- SinceAuthenticatedPacket: diagnosticAge(now, lastAuthenticatedPacket),
626- SinceCompleteRecord: diagnosticAge(now, lastCompleteRecord),
627- SinceACKProgress: diagnosticAge(now, lastACKProgress),
628- PendingRecords: pendingRecords,
629- PendingBytes: pendingBytes,
630- Retransmits: retransmits,
631- ReassemblyInflight: reassemblyInflight,
632- }
633-}
634-
635-func diagnosticAge(now, then time.Time) time.Duration {
636- if then.After(now) {
637- return 0
638- }
639- return now.Sub(then)
640-}
641-
642 // DatagramTransport marks Transport as a UDP-style datagram transport.389 // DatagramTransport marks Transport as a UDP-style datagram transport.
643-func (t *Transport) beginRuntimeOperation(start ports.RuntimeMarkKind, bytes uint64) func(bool) {
644- if t.runtimeObserver == nil {
645- return func(bool) {}
646- }
647- if !t.beginObservedOperation() {
648- return func(bool) {}
649- }
650- correlation := ports.NewRuntimeCorrelation()
651- t.runtimeObserver.ObserveRuntime(ports.NewRuntimeMarkWithCorrelation("dgram", correlation, start, bytes, true))
652- end := ports.RuntimeAdapterSendEnd
653- if start == ports.RuntimeAdapterReceiveStart {
654- end = ports.RuntimeAdapterReceiveEnd
655- }
656- return func(valid bool) {
657- defer t.finishObservedOperation()
658- t.runtimeObserver.ObserveRuntime(ports.NewRuntimeMarkWithCorrelation("dgram", correlation, end, bytes, valid))
659- }
660-}
661-
662-func (t *Transport) beginObservedOperation() bool {
663- t.operationMu.Lock()
664- defer t.operationMu.Unlock()
665- if t.closing {
666- return false
667- }
668- if t.operationCount == 0 {
669- t.operationsDone = make(chan struct{})
670- }
671- t.operationCount++
672- return true
673-}
674-
675-func (t *Transport) finishObservedOperation() {
676- t.operationMu.Lock()
677- defer t.operationMu.Unlock()
678- t.operationCount--
679- if t.operationCount == 0 {
680- close(t.operationsDone)
681- }
682-}
683-
684-func (t *Transport) beginShutdown() <-chan struct{} {
685- t.operationMu.Lock()
686- defer t.operationMu.Unlock()
687- t.closing = true
688- if t.operationCount == 0 {
689- return nil
690- }
691- return t.operationsDone
692-}
693-
694 func (*Transport) DatagramTransport() {}390 func (*Transport) DatagramTransport() {}
A internal/adapters/dgram/transport_diagnostics.go +118 -0
@@ -0,0 +1,118 @@
1+package dgram
2+
3+import (
4+ "time"
5+
6+ "github.com/bnema/vev/internal/ports"
7+)
8+
9+func (t *Transport) diagnosticLoop() {
10+ for {
11+ select {
12+ case d := <-t.diagnosticCh:
13+ t.observe(d)
14+ case <-t.done:
15+ return
16+ }
17+ }
18+}
19+
20+func (t *Transport) emitDiagnostic() {
21+ if t.diagnosticCh == nil {
22+ return
23+ }
24+ d := t.diagnosticSnapshot()
25+ select {
26+ case t.diagnosticCh <- d:
27+ default:
28+ }
29+}
30+
31+func (t *Transport) diagnosticSnapshot() Diagnostic {
32+ t.mu.Lock()
33+ now := t.clock.Now()
34+ lastAuthenticatedPacket := t.lastAuthenticatedPacket
35+ lastCompleteRecord := t.lastCompleteRecord
36+ lastACKProgress := t.lastACKProgress
37+ pendingBytes := 0
38+ for _, p := range t.pending {
39+ if p != nil {
40+ pendingBytes += len(p.frame.Payload)
41+ }
42+ }
43+ state := t.linkState
44+ pendingRecords := len(t.pending)
45+ retransmits := t.retransmits
46+ reassemblyInflight := t.reassemblyInflight
47+ t.mu.Unlock()
48+
49+ return Diagnostic{
50+ At: now,
51+ State: state,
52+ SinceAuthenticatedPacket: diagnosticAge(now, lastAuthenticatedPacket),
53+ SinceCompleteRecord: diagnosticAge(now, lastCompleteRecord),
54+ SinceACKProgress: diagnosticAge(now, lastACKProgress),
55+ PendingRecords: pendingRecords,
56+ PendingBytes: pendingBytes,
57+ Retransmits: retransmits,
58+ ReassemblyInflight: reassemblyInflight,
59+ }
60+}
61+
62+func diagnosticAge(now, then time.Time) time.Duration {
63+ if then.After(now) {
64+ return 0
65+ }
66+ return now.Sub(then)
67+}
68+
69+func (t *Transport) beginRuntimeOperation(start ports.RuntimeMarkKind, bytes uint64) func(bool) {
70+ if t.runtimeObserver == nil {
71+ return func(bool) {}
72+ }
73+ if !t.beginObservedOperation() {
74+ return func(bool) {}
75+ }
76+ correlation := ports.NewRuntimeCorrelation()
77+ t.runtimeObserver.ObserveRuntime(ports.NewRuntimeMarkWithCorrelation("dgram", correlation, start, bytes, true))
78+ end := ports.RuntimeAdapterSendEnd
79+ if start == ports.RuntimeAdapterReceiveStart {
80+ end = ports.RuntimeAdapterReceiveEnd
81+ }
82+ return func(valid bool) {
83+ defer t.finishObservedOperation()
84+ t.runtimeObserver.ObserveRuntime(ports.NewRuntimeMarkWithCorrelation("dgram", correlation, end, bytes, valid))
85+ }
86+}
87+
88+func (t *Transport) beginObservedOperation() bool {
89+ t.operationMu.Lock()
90+ defer t.operationMu.Unlock()
91+ if t.closing {
92+ return false
93+ }
94+ if t.operationCount == 0 {
95+ t.operationsDone = make(chan struct{})
96+ }
97+ t.operationCount++
98+ return true
99+}
100+
101+func (t *Transport) finishObservedOperation() {
102+ t.operationMu.Lock()
103+ defer t.operationMu.Unlock()
104+ t.operationCount--
105+ if t.operationCount == 0 {
106+ close(t.operationsDone)
107+ }
108+}
109+
110+func (t *Transport) beginShutdown() <-chan struct{} {
111+ t.operationMu.Lock()
112+ defer t.operationMu.Unlock()
113+ t.closing = true
114+ if t.operationCount == 0 {
115+ return nil
116+ }
117+ return t.operationsDone
118+}
@@ -0,0 +1,118 @@
1+package dgram
2+
3+import (
4+ "time"
5+
6+ "github.com/bnema/vev/internal/ports"
7+)
8+
9+func (t *Transport) diagnosticLoop() {
10+ for {
11+ select {
12+ case d := <-t.diagnosticCh:
13+ t.observe(d)
14+ case <-t.done:
15+ return
16+ }
17+ }
18+}
19+
20+func (t *Transport) emitDiagnostic() {
21+ if t.diagnosticCh == nil {
22+ return
23+ }
24+ d := t.diagnosticSnapshot()
25+ select {
26+ case t.diagnosticCh <- d:
27+ default:
28+ }
29+}
30+
31+func (t *Transport) diagnosticSnapshot() Diagnostic {
32+ t.mu.Lock()
33+ now := t.clock.Now()
34+ lastAuthenticatedPacket := t.lastAuthenticatedPacket
35+ lastCompleteRecord := t.lastCompleteRecord
36+ lastACKProgress := t.lastACKProgress
37+ pendingBytes := 0
38+ for _, p := range t.pending {
39+ if p != nil {
40+ pendingBytes += len(p.frame.Payload)
41+ }
42+ }
43+ state := t.linkState
44+ pendingRecords := len(t.pending)
45+ retransmits := t.retransmits
46+ reassemblyInflight := t.reassemblyInflight
47+ t.mu.Unlock()
48+
49+ return Diagnostic{
50+ At: now,
51+ State: state,
52+ SinceAuthenticatedPacket: diagnosticAge(now, lastAuthenticatedPacket),
53+ SinceCompleteRecord: diagnosticAge(now, lastCompleteRecord),
54+ SinceACKProgress: diagnosticAge(now, lastACKProgress),
55+ PendingRecords: pendingRecords,
56+ PendingBytes: pendingBytes,
57+ Retransmits: retransmits,
58+ ReassemblyInflight: reassemblyInflight,
59+ }
60+}
61+
62+func diagnosticAge(now, then time.Time) time.Duration {
63+ if then.After(now) {
64+ return 0
65+ }
66+ return now.Sub(then)
67+}
68+
69+func (t *Transport) beginRuntimeOperation(start ports.RuntimeMarkKind, bytes uint64) func(bool) {
70+ if t.runtimeObserver == nil {
71+ return func(bool) {}
72+ }
73+ if !t.beginObservedOperation() {
74+ return func(bool) {}
75+ }
76+ correlation := ports.NewRuntimeCorrelation()
77+ t.runtimeObserver.ObserveRuntime(ports.NewRuntimeMarkWithCorrelation("dgram", correlation, start, bytes, true))
78+ end := ports.RuntimeAdapterSendEnd
79+ if start == ports.RuntimeAdapterReceiveStart {
80+ end = ports.RuntimeAdapterReceiveEnd
81+ }
82+ return func(valid bool) {
83+ defer t.finishObservedOperation()
84+ t.runtimeObserver.ObserveRuntime(ports.NewRuntimeMarkWithCorrelation("dgram", correlation, end, bytes, valid))
85+ }
86+}
87+
88+func (t *Transport) beginObservedOperation() bool {
89+ t.operationMu.Lock()
90+ defer t.operationMu.Unlock()
91+ if t.closing {
92+ return false
93+ }
94+ if t.operationCount == 0 {
95+ t.operationsDone = make(chan struct{})
96+ }
97+ t.operationCount++
98+ return true
99+}
100+
101+func (t *Transport) finishObservedOperation() {
102+ t.operationMu.Lock()
103+ defer t.operationMu.Unlock()
104+ t.operationCount--
105+ if t.operationCount == 0 {
106+ close(t.operationsDone)
107+ }
108+}
109+
110+func (t *Transport) beginShutdown() <-chan struct{} {
111+ t.operationMu.Lock()
112+ defer t.operationMu.Unlock()
113+ t.closing = true
114+ if t.operationCount == 0 {
115+ return nil
116+ }
117+ return t.operationsDone
118+}
A internal/adapters/dgram/transport_health.go +200 -0
@@ -0,0 +1,200 @@
1+package dgram
2+
3+import (
4+ "time"
5+
6+ "github.com/bnema/vev/internal/ports"
7+)
8+
9+// healthLoop owns link-health timing so bulk retransmission cannot delay state
10+// transitions or socket hopping.
11+func (t *Transport) healthLoop() {
12+ health := t.clock.NewTimer(t.resendAfter)
13+ defer health.Stop()
14+ for {
15+ select {
16+ case <-health.C():
17+ t.checkSilence()
18+ health.Reset(t.resendAfter)
19+ t.notifyTimerHook(func() func() { return t.afterHealthTimer })
20+ case <-t.done:
21+ return
22+ }
23+ }
24+}
25+
26+func (t *Transport) heartbeatLoop() {
27+ heartbeat := t.clock.NewTimer(t.heartbeat)
28+ defer heartbeat.Stop()
29+ for {
30+ select {
31+ case <-heartbeat.C():
32+ t.queueControl(recProbe, 0)
33+ heartbeat.Reset(t.heartbeat)
34+ t.notifyTimerHook(func() func() { return t.afterHeartbeatTimer })
35+ case <-t.done:
36+ return
37+ }
38+ }
39+}
40+
41+func (t *Transport) notifyTimerHook(get func() func()) {
42+ t.mu.Lock()
43+ hook := get()
44+ t.mu.Unlock()
45+ if hook != nil {
46+ hook()
47+ }
48+}
49+
50+func (t *Transport) hopPacketConnOnce(generation uint64) {
51+ t.mu.Lock()
52+ if t.rebind == nil || t.hoppedOffline || t.closed || t.linkState != ports.LinkStateProbing || t.health.generation != generation {
53+ t.mu.Unlock()
54+ return
55+ }
56+ old := t.pc
57+ rebind := t.rebind
58+ t.hoppedOffline = true
59+ t.hopGeneration = generation
60+ t.mu.Unlock()
61+ pc, err := rebind(old)
62+
63+ // Serialize the swap with writes. writePacket takes writeMu before mu, so
64+ // preserve that order here while retiring old.
65+ t.writeMu.Lock()
66+ defer t.writeMu.Unlock()
67+ t.mu.Lock()
68+ if err != nil {
69+ t.rollbackHopLocked(generation)
70+ t.mu.Unlock()
71+ return
72+ }
73+ if t.closed || t.pc != old || t.health.generation != generation || t.linkState != ports.LinkStateProbing {
74+ t.rollbackHopLocked(generation)
75+ t.mu.Unlock()
76+ _ = pc.Close()
77+ return
78+ }
79+ t.pc = pc
80+ t.writeDeadlines = newWriteDeadlineState(pc)
81+ t.mu.Unlock()
82+ go t.readLoop(pc)
83+ _ = old.Close()
84+}
85+
86+func (t *Transport) rollbackHopLocked(generation uint64) {
87+ if t.hoppedOffline && t.hopGeneration == generation {
88+ t.hoppedOffline = false
89+ t.hopGeneration = 0
90+ }
91+}
92+
93+func (t *Transport) checkSilence() {
94+ runHook := true
95+ for {
96+ t.mu.Lock()
97+ if t.closed {
98+ t.mu.Unlock()
99+ return
100+ }
101+ now := t.clock.Now()
102+ generation := t.health.generation
103+ state, hop, dead := t.health.decide(now, len(t.pending) > 0, t.degradedAfter, t.probeAfter, t.offlineAfter, t.deadAfter)
104+ afterDecision := t.afterHealthDecision
105+ t.mu.Unlock()
106+
107+ if runHook {
108+ runHook = false
109+ if afterDecision != nil {
110+ afterDecision()
111+ }
112+ }
113+
114+ t.mu.Lock()
115+ if t.closed {
116+ t.mu.Unlock()
117+ return
118+ }
119+ if t.health.generation != generation {
120+ t.mu.Unlock()
121+ continue
122+ }
123+ changed, stateGeneration := t.setLinkStateLocked(state)
124+ afterStateCommit := t.afterLinkStateCommit
125+ closed := false
126+ if dead {
127+ closed = t.closeWithErrorLocked(ErrLinkDead)
128+ }
129+ t.mu.Unlock()
130+
131+ if changed {
132+ if afterStateCommit != nil {
133+ afterStateCommit(state)
134+ }
135+ var linkErr error
136+ if dead {
137+ linkErr = ErrLinkDead
138+ }
139+ t.publishLinkState(state, now, linkErr, stateGeneration)
140+ }
141+ if closed {
142+ t.broadcastClosed()
143+ return
144+ }
145+ if hop {
146+ t.hopPacketConnOnce(generation)
147+ }
148+ return
149+ }
150+}
151+
152+func (t *Transport) setLinkState(state ports.LinkState, err error) {
153+ now := t.clock.Now()
154+ t.mu.Lock()
155+ changed, stateGeneration := t.setLinkStateLocked(state)
156+ afterStateCommit := t.afterLinkStateCommit
157+ t.mu.Unlock()
158+ if changed {
159+ if afterStateCommit != nil {
160+ afterStateCommit(state)
161+ }
162+ t.publishLinkState(state, now, err, stateGeneration)
163+ }
164+}
165+
166+func (t *Transport) setLinkStateLocked(state ports.LinkState) (bool, uint64) {
167+ if t.linkState == state {
168+ return false, t.linkStateGeneration
169+ }
170+ t.linkState = state
171+ t.linkStateGeneration++
172+ if state == ports.LinkStateConnected {
173+ t.hoppedOffline = false
174+ t.hopGeneration = 0
175+ for _, p := range t.pending {
176+ if p != nil {
177+ p.attempts = 0
178+ }
179+ }
180+ }
181+ t.notifySendWaitersLocked()
182+ return true, t.linkStateGeneration
183+}
184+
185+func (t *Transport) publishLinkState(state ports.LinkState, now time.Time, err error, generation uint64) {
186+ t.linkEventMu.Lock()
187+ defer t.linkEventMu.Unlock()
188+ t.mu.Lock()
189+ if t.linkState != state || t.linkStateGeneration != generation {
190+ t.mu.Unlock()
191+ return
192+ }
193+ event := ports.LinkEvent{State: state, At: now, Err: err}
194+ select {
195+ case t.linkEvents <- event:
196+ default:
197+ }
198+ t.mu.Unlock()
199+ t.emitDiagnostic()
200+}
@@ -0,0 +1,200 @@
1+package dgram
2+
3+import (
4+ "time"
5+
6+ "github.com/bnema/vev/internal/ports"
7+)
8+
9+// healthLoop owns link-health timing so bulk retransmission cannot delay state
10+// transitions or socket hopping.
11+func (t *Transport) healthLoop() {
12+ health := t.clock.NewTimer(t.resendAfter)
13+ defer health.Stop()
14+ for {
15+ select {
16+ case <-health.C():
17+ t.checkSilence()
18+ health.Reset(t.resendAfter)
19+ t.notifyTimerHook(func() func() { return t.afterHealthTimer })
20+ case <-t.done:
21+ return
22+ }
23+ }
24+}
25+
26+func (t *Transport) heartbeatLoop() {
27+ heartbeat := t.clock.NewTimer(t.heartbeat)
28+ defer heartbeat.Stop()
29+ for {
30+ select {
31+ case <-heartbeat.C():
32+ t.queueControl(recProbe, 0)
33+ heartbeat.Reset(t.heartbeat)
34+ t.notifyTimerHook(func() func() { return t.afterHeartbeatTimer })
35+ case <-t.done:
36+ return
37+ }
38+ }
39+}
40+
41+func (t *Transport) notifyTimerHook(get func() func()) {
42+ t.mu.Lock()
43+ hook := get()
44+ t.mu.Unlock()
45+ if hook != nil {
46+ hook()
47+ }
48+}
49+
50+func (t *Transport) hopPacketConnOnce(generation uint64) {
51+ t.mu.Lock()
52+ if t.rebind == nil || t.hoppedOffline || t.closed || t.linkState != ports.LinkStateProbing || t.health.generation != generation {
53+ t.mu.Unlock()
54+ return
55+ }
56+ old := t.pc
57+ rebind := t.rebind
58+ t.hoppedOffline = true
59+ t.hopGeneration = generation
60+ t.mu.Unlock()
61+ pc, err := rebind(old)
62+
63+ // Serialize the swap with writes. writePacket takes writeMu before mu, so
64+ // preserve that order here while retiring old.
65+ t.writeMu.Lock()
66+ defer t.writeMu.Unlock()
67+ t.mu.Lock()
68+ if err != nil {
69+ t.rollbackHopLocked(generation)
70+ t.mu.Unlock()
71+ return
72+ }
73+ if t.closed || t.pc != old || t.health.generation != generation || t.linkState != ports.LinkStateProbing {
74+ t.rollbackHopLocked(generation)
75+ t.mu.Unlock()
76+ _ = pc.Close()
77+ return
78+ }
79+ t.pc = pc
80+ t.writeDeadlines = newWriteDeadlineState(pc)
81+ t.mu.Unlock()
82+ go t.readLoop(pc)
83+ _ = old.Close()
84+}
85+
86+func (t *Transport) rollbackHopLocked(generation uint64) {
87+ if t.hoppedOffline && t.hopGeneration == generation {
88+ t.hoppedOffline = false
89+ t.hopGeneration = 0
90+ }
91+}
92+
93+func (t *Transport) checkSilence() {
94+ runHook := true
95+ for {
96+ t.mu.Lock()
97+ if t.closed {
98+ t.mu.Unlock()
99+ return
100+ }
101+ now := t.clock.Now()
102+ generation := t.health.generation
103+ state, hop, dead := t.health.decide(now, len(t.pending) > 0, t.degradedAfter, t.probeAfter, t.offlineAfter, t.deadAfter)
104+ afterDecision := t.afterHealthDecision
105+ t.mu.Unlock()
106+
107+ if runHook {
108+ runHook = false
109+ if afterDecision != nil {
110+ afterDecision()
111+ }
112+ }
113+
114+ t.mu.Lock()
115+ if t.closed {
116+ t.mu.Unlock()
117+ return
118+ }
119+ if t.health.generation != generation {
120+ t.mu.Unlock()
121+ continue
122+ }
123+ changed, stateGeneration := t.setLinkStateLocked(state)
124+ afterStateCommit := t.afterLinkStateCommit
125+ closed := false
126+ if dead {
127+ closed = t.closeWithErrorLocked(ErrLinkDead)
128+ }
129+ t.mu.Unlock()
130+
131+ if changed {
132+ if afterStateCommit != nil {
133+ afterStateCommit(state)
134+ }
135+ var linkErr error
136+ if dead {
137+ linkErr = ErrLinkDead
138+ }
139+ t.publishLinkState(state, now, linkErr, stateGeneration)
140+ }
141+ if closed {
142+ t.broadcastClosed()
143+ return
144+ }
145+ if hop {
146+ t.hopPacketConnOnce(generation)
147+ }
148+ return
149+ }
150+}
151+
152+func (t *Transport) setLinkState(state ports.LinkState, err error) {
153+ now := t.clock.Now()
154+ t.mu.Lock()
155+ changed, stateGeneration := t.setLinkStateLocked(state)
156+ afterStateCommit := t.afterLinkStateCommit
157+ t.mu.Unlock()
158+ if changed {
159+ if afterStateCommit != nil {
160+ afterStateCommit(state)
161+ }
162+ t.publishLinkState(state, now, err, stateGeneration)
163+ }
164+}
165+
166+func (t *Transport) setLinkStateLocked(state ports.LinkState) (bool, uint64) {
167+ if t.linkState == state {
168+ return false, t.linkStateGeneration
169+ }
170+ t.linkState = state
171+ t.linkStateGeneration++
172+ if state == ports.LinkStateConnected {
173+ t.hoppedOffline = false
174+ t.hopGeneration = 0
175+ for _, p := range t.pending {
176+ if p != nil {
177+ p.attempts = 0
178+ }
179+ }
180+ }
181+ t.notifySendWaitersLocked()
182+ return true, t.linkStateGeneration
183+}
184+
185+func (t *Transport) publishLinkState(state ports.LinkState, now time.Time, err error, generation uint64) {
186+ t.linkEventMu.Lock()
187+ defer t.linkEventMu.Unlock()
188+ t.mu.Lock()
189+ if t.linkState != state || t.linkStateGeneration != generation {
190+ t.mu.Unlock()
191+ return
192+ }
193+ event := ports.LinkEvent{State: state, At: now, Err: err}
194+ select {
195+ case t.linkEvents <- event:
196+ default:
197+ }
198+ t.mu.Unlock()
199+ t.emitDiagnostic()
200+}

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