fix(dgram): track pending UDP round trips

commit: 7f56b4a7041c6056211542a91ee74cfa7b5b071c
author: brice · 3mo ago · 2026-07-10 06:46 UTC
parent: 68a2ea2
signature: GPG VerifiedCD88F9E564E0102646FB6077A1A07C7C7C3E349D · arch-gpg Verified for the committer email

3 files · +110 -5

M internal/adapters/dgram/health.go +19 -2


@@ -12,6 +12,7 @@ type healthTracker struct {
12	12		lastPacket time.Time
13	13		lastRecord time.Time
14	14		lastProgress time.Time
	15	+	pendingSince time.Time
15	16		generation uint64
16	17		}
17	18

@@ -34,6 +35,16 @@ func (h *healthTracker) ackProgress(now time.Time) {
34	35		h.generation++
35	36		}
36	37
	38	+	func (h *healthTracker) pendingStarted(now time.Time) {
	39	+	h.pendingSince = now
	40	+	h.generation++
	41	+	}
	42	+
	43	+	func (h *healthTracker) pendingCleared() {
	44	+	h.pendingSince = time.Time{}
	45	+	h.generation++
	46	+	}
	47	+
37	48		func (h healthTracker) decide(now time.Time, hasPending bool, degradedAfter, probeAfter, offlineAfter, deadAfter time.Duration) (ports.LinkState, bool, bool) {
38	49		packetAge := now.Sub(h.lastPacket)
39	50		switch {

@@ -50,8 +61,14 @@ func (h healthTracker) decide(now time.Time, hasPending bool, degradedAfter, pro
50	61		qualityAt = h.lastProgress
51	62		}
52	63		qualityAge := now.Sub(qualityAt)
53		-	if hasPending && probeAfter > 0 && qualityAge >= probeAfter {
54		-	return ports.LinkStateProbing, true, false
	64	+	if hasPending && probeAfter > 0 {
	65	+	pendingBaseline := h.pendingSince
	66	+	if pendingBaseline.IsZero() \|\| h.lastProgress.After(pendingBaseline) {
	67	+	pendingBaseline = h.lastProgress
	68	+	}
	69	+	if now.Sub(pendingBaseline) >= probeAfter {
	70	+	return ports.LinkStateProbing, true, false
	71	+	}
55	72		}
56	73		if degradedAfter > 0 && qualityAge >= degradedAfter {
57	74		return ports.LinkStateDegraded, false, false

 	lastPacket   time.Time
 	lastRecord   time.Time
 	lastProgress time.Time
+	pendingSince time.Time
 	generation   uint64
 }
 	h.generation++
 }
+func (h *healthTracker) pendingStarted(now time.Time) {
+	h.pendingSince = now
+	h.generation++
+}
+func (h *healthTracker) pendingCleared() {
+	h.pendingSince = time.Time{}
+	h.generation++
+}
 func (h healthTracker) decide(now time.Time, hasPending bool, degradedAfter, probeAfter, offlineAfter, deadAfter time.Duration) (ports.LinkState, bool, bool) {
 	packetAge := now.Sub(h.lastPacket)
 	switch {
 		qualityAt = h.lastProgress
 	}
 	qualityAge := now.Sub(qualityAt)
-	if hasPending && probeAfter > 0 && qualityAge >= probeAfter {
+	if hasPending && probeAfter > 0 {
-		return ports.LinkStateProbing, true, false
+		pendingBaseline := h.pendingSince
+		if pendingBaseline.IsZero() || h.lastProgress.After(pendingBaseline) {
+			pendingBaseline = h.lastProgress
+		}
+		if now.Sub(pendingBaseline) >= probeAfter {
+			return ports.LinkStateProbing, true, false
+		}
 	}
 	if degradedAfter > 0 && qualityAge >= degradedAfter {
 		return ports.LinkStateDegraded, false, false

M internal/adapters/dgram/transport.go +12 -2


@@ -395,7 +395,11 @@ func (t *Transport) send(f ports.Frame, async bool) error {
395	395		t.seq++
396	396		seq := t.seq
397	397		if reliable {
398		-	t.pending[seq] = &pending{frame: f, enqueued: t.clock.Now()}
	398	+	now := t.clock.Now()
	399	+	if len(t.pending) == 0 {
	400	+	t.health.pendingStarted(now)
	401	+	}
	402	+	t.pending[seq] = &pending{frame: f, enqueued: now}
399	403		}
400	404		if shouldPaceOutput(f) {
401	405		now := t.clock.Now()

@@ -554,6 +558,9 @@ func (t *Transport) removePendingLocked(seq uint64) bool {
554	558		return false
555	559		}
556	560		delete(t.pending, seq)
	561	+	if len(t.pending) == 0 {
	562	+	t.health.pendingCleared()
	563	+	}
557	564		return true
558	565		}
559	566

@@ -838,6 +845,7 @@ func (t *Transport) readLoop(pc net.PacketConn) {
838	845		}
839	846		t.recordCompleteRecord()
840	847		t.handleRecord(payload)
	848	+	t.checkSilence()
841	849		}
842	850		}
843	851

@@ -856,7 +864,6 @@ func (t *Transport) recordCompleteRecord() {
856	864		t.health.completeRecord(now)
857	865		t.lastCompleteRecord = now
858	866		t.mu.Unlock()
859		-	t.setLinkState(ports.LinkStateConnected, nil)
860	867		}
861	868
862	869		func (t *Transport) handleRecord(p []byte) {

@@ -884,6 +891,9 @@ func (t *Transport) handleRecord(p []byte) {
884	891		if acked {
885	892		now := t.clock.Now()
886	893		t.health.ackProgress(now)
	894	+	if len(t.pending) == 0 {
	895	+	t.health.pendingCleared()
	896	+	}
887	897		t.lastACKProgress = now
888	898		t.notifySendWaitersLocked()
889	899		}

 	t.seq++
 	seq := t.seq
 	if reliable {
-		t.pending[seq] = &pending{frame: f, enqueued: t.clock.Now()}
+		now := t.clock.Now()
+		if len(t.pending) == 0 {
+			t.health.pendingStarted(now)
+		}
+		t.pending[seq] = &pending{frame: f, enqueued: now}
 	}
 	if shouldPaceOutput(f) {
 		now := t.clock.Now()
 		return false
 	}
 	delete(t.pending, seq)
+	if len(t.pending) == 0 {
+		t.health.pendingCleared()
+	}
 	return true
 }
 		}
 		t.recordCompleteRecord()
 		t.handleRecord(payload)
+		t.checkSilence()
 	}
 }
 	t.health.completeRecord(now)
 	t.lastCompleteRecord = now
 	t.mu.Unlock()
-	t.setLinkState(ports.LinkStateConnected, nil)
 }
 func (t *Transport) handleRecord(p []byte) {
 		if acked {
 			now := t.clock.Now()
 			t.health.ackProgress(now)
+			if len(t.pending) == 0 {
+				t.health.pendingCleared()
+			}
 			t.lastACKProgress = now
 			t.notifySendWaitersLocked()
 		}

M internal/adapters/dgram/transport_test.go +79 -1


@@ -1843,7 +1843,8 @@ func TestLinkHealthSeparatesPacketContactFromUsefulProgress(t *testing.T) {
1843	1843		{name: "packet silence probes and hops", packetAge: threshold(20), recordAge: threshold(20), ackAge: threshold(20), wantState: ports.LinkStateProbing, wantHop: true},
1844	1844		{name: "packet silence becomes offline", packetAge: threshold(30), recordAge: threshold(30), ackAge: threshold(30), wantState: ports.LinkStateOffline},
1845	1845		{name: "packet silence becomes dead", packetAge: threshold(60), recordAge: threshold(60), ackAge: threshold(60), wantState: ports.LinkStateDead, wantDead: true},
1846		-	{name: "complete record restores connected", packetAge: 0, recordAge: 0, ackAge: threshold(30), hasPending: true, wantState: ports.LinkStateConnected},
	1846	+	{name: "complete record cannot mask pending stall", packetAge: 0, recordAge: 0, ackAge: threshold(30), hasPending: true, wantState: ports.LinkStateProbing, wantHop: true},
	1847	+	{name: "complete record restores connected without pending data", packetAge: 0, recordAge: 0, ackAge: threshold(30), wantState: ports.LinkStateConnected},
1847	1848		{name: "ack progress restores connected", packetAge: 0, recordAge: threshold(30), ackAge: 0, hasPending: true, wantState: ports.LinkStateConnected},
1848	1849		}
1849	1850

@@ -1896,6 +1897,83 @@ func TestLinkHealthPendingStallWithFreshContactProbesOnceWithoutDying(t *testing
1896	1897		}
1897	1898		}
1898	1899
	1900	+	func TestLinkHealthHeartbeatControlCannotMaskPendingACKStall(t *testing.T) {
	1901	+	aPC, bPC := newPair()
	1902	+	clk := newManualClock(time.Date(2030, 1, 2, 3, 4, 5, 0, time.UTC))
	1903	+	var hops atomic.Int32
	1904	+	tr, err := NewTransportWithOptions(aPC, testAddr("b"), key(), 1, 2, Options{
	1905	+	Clock: clk, ResendAfter: time.Hour, Heartbeat: time.Hour,
	1906	+	RebindPacketConn: func(net.PacketConn) (net.PacketConn, error) {
	1907	+	attempt := hops.Add(1)
	1908	+	return &fakePC{addr: testAddr(fmt.Sprintf("a%d", attempt+1)), in: make(chan packet, 100), peers: map[string]*fakePC{"b": bPC}}, nil
	1909	+	},
	1910	+	})
	1911	+	if err != nil {
	1912	+	t.Fatal(err)
	1913	+	}
	1914	+	defer func() { _ = tr.Close() }()
	1915	+	defer func() { _ = bPC.Close() }()
	1916	+	if err := tr.Send(ports.Frame{Type: ports.MsgInput, Payload: []byte("unacked")}); err != nil {
	1917	+	t.Fatal(err)
	1918	+	}
	1919	+
	1920	+	codec, err := pdgram.NewCodec(key())
	1921	+	if err != nil {
	1922	+	t.Fatal(err)
	1923	+	}
	1924	+	authenticated := make(chan struct{}, 32)
	1925	+	tr.mu.Lock()
	1926	+	tr.afterAuthenticatedPacket = func() { authenticated <- struct{}{} }
	1927	+	tr.mu.Unlock()
	1928	+	var counter uint64
	1929	+	sendHeartbeatPong := func() {
	1930	+	t.Helper()
	1931	+	counter++
	1932	+	frag := pdgram.Fragment{Seq: counter, Index: 0, Count: 1, Data: probeRecord(recPong, counter)}
	1933	+	raw, marshalErr := pdgram.MarshalFragment(frag)
	1934	+	if marshalErr != nil {
	1935	+	t.Fatal(marshalErr)
	1936	+	}
	1937	+	tr.mu.Lock()
	1938	+	pc := tr.pc.(*fakePC)
	1939	+	tr.mu.Unlock()
	1940	+	pc.in <- packet{b: codec.Seal(2, counter, raw, nil), addr: testAddr("b")}
	1941	+	select {
	1942	+	case <-authenticated:
	1943	+	case <-time.After(time.Second):
	1944	+	t.Fatal("heartbeat pong was not authenticated")
	1945	+	}
	1946	+	}
	1947	+
	1948	+	for elapsed := 3 * time.Second; elapsed <= 66time.Second; elapsed += 3 time.Second {
	1949	+	clk.advance(3 * time.Second)
	1950	+	sendHeartbeatPong()
	1951	+	eventually(t, time.Second, func() bool {
	1952	+	tr.mu.Lock()
	1953	+	defer tr.mu.Unlock()
	1954	+	return tr.health.lastRecord.Equal(clk.Now())
	1955	+	})
	1956	+	if elapsed >= 20*time.Second {
	1957	+	eventually(t, time.Second, func() bool {
	1958	+	tr.mu.Lock()
	1959	+	defer tr.mu.Unlock()
	1960	+	return tr.pc != aPC
	1961	+	})
	1962	+	}
	1963	+	}
	1964	+	if got := tr.LinkState(); got != ports.LinkStateProbing {
	1965	+	t.Fatalf("state=%v, want probing despite fresh heartbeat control traffic", got)
	1966	+	}
	1967	+	if got := hops.Load(); got != 1 {
	1968	+	t.Fatalf("socket hops=%d, want one pending round-trip recovery hop", got)
	1969	+	}
	1970	+	select {
	1971	+	case <-tr.done:
	1972	+	t.Fatal("fresh authenticated control contact must prevent offline/dead closure")
	1973	+	default:
	1974	+	}
	1975	+	}
	1976	+
1899	1977		func TestLinkHealthOnlyAckProgressThatRemovesPendingRestoresConnected(t *testing.T) {
1900	1978		tr, _ := newResendTestTransport(t, 10)
1901	1979		now := tr.clock.Now()


@@ -1843,7 +1843,8 @@ func TestLinkHealthSeparatesPacketContactFromUsefulProgress(t *testing.T) {
1843		{name: "packet silence probes and hops", packetAge: threshold(20), recordAge: threshold(20), ackAge: threshold(20), wantState: ports.LinkStateProbing, wantHop: true},	1843		{name: "packet silence probes and hops", packetAge: threshold(20), recordAge: threshold(20), ackAge: threshold(20), wantState: ports.LinkStateProbing, wantHop: true},
1844		{name: "packet silence becomes offline", packetAge: threshold(30), recordAge: threshold(30), ackAge: threshold(30), wantState: ports.LinkStateOffline},	1844		{name: "packet silence becomes offline", packetAge: threshold(30), recordAge: threshold(30), ackAge: threshold(30), wantState: ports.LinkStateOffline},
1845		{name: "packet silence becomes dead", packetAge: threshold(60), recordAge: threshold(60), ackAge: threshold(60), wantState: ports.LinkStateDead, wantDead: true},	1845		{name: "packet silence becomes dead", packetAge: threshold(60), recordAge: threshold(60), ackAge: threshold(60), wantState: ports.LinkStateDead, wantDead: true},
1846	-	{name: "complete record restores connected", packetAge: 0, recordAge: 0, ackAge: threshold(30), hasPending: true, wantState: ports.LinkStateConnected},	1846	+	{name: "complete record cannot mask pending stall", packetAge: 0, recordAge: 0, ackAge: threshold(30), hasPending: true, wantState: ports.LinkStateProbing, wantHop: true},
			1847	+	{name: "complete record restores connected without pending data", packetAge: 0, recordAge: 0, ackAge: threshold(30), wantState: ports.LinkStateConnected},
1847		{name: "ack progress restores connected", packetAge: 0, recordAge: threshold(30), ackAge: 0, hasPending: true, wantState: ports.LinkStateConnected},	1848		{name: "ack progress restores connected", packetAge: 0, recordAge: threshold(30), ackAge: 0, hasPending: true, wantState: ports.LinkStateConnected},
1848		}	1849		}
1849			1850

@@ -1896,6 +1897,83 @@ func TestLinkHealthPendingStallWithFreshContactProbesOnceWithoutDying(t *testing
1896		}	1897		}
1897		}	1898		}
1898			1899
			1900	+	func TestLinkHealthHeartbeatControlCannotMaskPendingACKStall(t *testing.T) {
			1901	+	aPC, bPC := newPair()
			1902	+	clk := newManualClock(time.Date(2030, 1, 2, 3, 4, 5, 0, time.UTC))
			1903	+	var hops atomic.Int32
			1904	+	tr, err := NewTransportWithOptions(aPC, testAddr("b"), key(), 1, 2, Options{
			1905	+	Clock: clk, ResendAfter: time.Hour, Heartbeat: time.Hour,
			1906	+	RebindPacketConn: func(net.PacketConn) (net.PacketConn, error) {
			1907	+	attempt := hops.Add(1)
			1908	+	return &fakePC{addr: testAddr(fmt.Sprintf("a%d", attempt+1)), in: make(chan packet, 100), peers: map[string]*fakePC{"b": bPC}}, nil
			1909	+	},
			1910	+	})
			1911	+	if err != nil {
			1912	+	t.Fatal(err)
			1913	+	}
			1914	+	defer func() { _ = tr.Close() }()
			1915	+	defer func() { _ = bPC.Close() }()
			1916	+	if err := tr.Send(ports.Frame{Type: ports.MsgInput, Payload: []byte("unacked")}); err != nil {
			1917	+	t.Fatal(err)
			1918	+	}
			1919	+
			1920	+	codec, err := pdgram.NewCodec(key())
			1921	+	if err != nil {
			1922	+	t.Fatal(err)
			1923	+	}
			1924	+	authenticated := make(chan struct{}, 32)
			1925	+	tr.mu.Lock()
			1926	+	tr.afterAuthenticatedPacket = func() { authenticated <- struct{}{} }
			1927	+	tr.mu.Unlock()
			1928	+	var counter uint64
			1929	+	sendHeartbeatPong := func() {
			1930	+	t.Helper()
			1931	+	counter++
			1932	+	frag := pdgram.Fragment{Seq: counter, Index: 0, Count: 1, Data: probeRecord(recPong, counter)}
			1933	+	raw, marshalErr := pdgram.MarshalFragment(frag)
			1934	+	if marshalErr != nil {
			1935	+	t.Fatal(marshalErr)
			1936	+	}
			1937	+	tr.mu.Lock()
			1938	+	pc := tr.pc.(*fakePC)
			1939	+	tr.mu.Unlock()
			1940	+	pc.in <- packet{b: codec.Seal(2, counter, raw, nil), addr: testAddr("b")}
			1941	+	select {
			1942	+	case <-authenticated:
			1943	+	case <-time.After(time.Second):
			1944	+	t.Fatal("heartbeat pong was not authenticated")
			1945	+	}
			1946	+	}
			1947	+
			1948	+	for elapsed := 3 * time.Second; elapsed <= 66time.Second; elapsed += 3 time.Second {
			1949	+	clk.advance(3 * time.Second)
			1950	+	sendHeartbeatPong()
			1951	+	eventually(t, time.Second, func() bool {
			1952	+	tr.mu.Lock()
			1953	+	defer tr.mu.Unlock()
			1954	+	return tr.health.lastRecord.Equal(clk.Now())
			1955	+	})
			1956	+	if elapsed >= 20*time.Second {
			1957	+	eventually(t, time.Second, func() bool {
			1958	+	tr.mu.Lock()
			1959	+	defer tr.mu.Unlock()
			1960	+	return tr.pc != aPC
			1961	+	})
			1962	+	}
			1963	+	}
			1964	+	if got := tr.LinkState(); got != ports.LinkStateProbing {
			1965	+	t.Fatalf("state=%v, want probing despite fresh heartbeat control traffic", got)
			1966	+	}
			1967	+	if got := hops.Load(); got != 1 {
			1968	+	t.Fatalf("socket hops=%d, want one pending round-trip recovery hop", got)
			1969	+	}
			1970	+	select {
			1971	+	case <-tr.done:
			1972	+	t.Fatal("fresh authenticated control contact must prevent offline/dead closure")
			1973	+	default:
			1974	+	}
			1975	+	}
			1976	+
1899		func TestLinkHealthOnlyAckProgressThatRemovesPendingRestoresConnected(t *testing.T) {	1977		func TestLinkHealthOnlyAckProgressThatRemovesPendingRestoresConnected(t *testing.T) {
1900		tr, _ := newResendTestTransport(t, 10)	1978		tr, _ := newResendTestTransport(t, 10)
1901		now := tr.clock.Now()	1979		now := tr.clock.Now()