Skip to content

Commit 4bcd3c7

Browse files
committed
Fix retransmission starvation by requesting oldest gaps first
TCP retransmission documentation Change missing sequence requests from newest-first ([-100:]) to oldest-first ([:100]) to prioritize recovery of gaps closest to aging out of the car's 60s ring buffer. This prevents starvation under sustained packet loss where old gaps would never be recovered before expiring. Adds comprehensive unit tests to verify oldest-first recovery behavior under various loss scenarios. Updates README to reflect the new strategy and removes the now-obsolete starvation limitation.
1 parent c52733c commit 4bcd3c7

3 files changed

Lines changed: 250 additions & 3 deletions

File tree

universal-telemetry-software/README.md

Lines changed: 96 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -37,6 +37,100 @@ The deployed car service sets `ROLE=car` explicitly in `deploy/car-telemetry.ser
3737

3838
---
3939

40+
## Reliability & Retransmission
41+
42+
The link between car and base is a **best-effort UDP stream with a pull-based recovery channel over TCP**. The car never waits for acknowledgements — it fires CAN batches over UDP as fast as it reads them and keeps a short backlog in memory. The base station detects the gaps and asks for the missing pieces back. This keeps the live path low-latency while still recovering most dropped packets a few seconds later.
43+
44+
All of this lives in `src/data.py`: the car side in `run_car()` (`udp_sender`, `handle_resend`), the base side in `run_base()` (`udp_receiver`, `missing_reporter`).
45+
46+
### Data flow
47+
48+
```mermaid
49+
sequenceDiagram
50+
participant CAN as CAN bus
51+
participant Car as Car (udp_sender)
52+
participant RB as Ring buffer (60s)
53+
participant Base as Base (udp_receiver)
54+
participant MR as Base (missing_reporter)
55+
56+
CAN->>Car: raw frames
57+
Note over Car: batch 20 msgs OR 50ms<br/>assign seq_num (uint64)
58+
Car->>RB: store (seq, batch, t)
59+
Car--)Base: UDP :5005 [seq | count | msgs]
60+
Note over Base: track expected_seq<br/>seq > expected ⇒ gap<br/>add skipped seqs to missing set
61+
loop every 10s, if missing set non-empty
62+
MR->>Car: TCP :5006 {"missing": [seq, ...]}
63+
Car->>RB: look up seqs still in buffer
64+
Car-->>MR: JSON [{seq, msgs:[{t,id,d}]}]
65+
Note over MR: re-publish to Redis<br/>mark seq recovered
66+
end
67+
```
68+
69+
### Sender: batching, sequencing, ring buffer (car)
70+
71+
- **Batching** — CAN frames are accumulated and flushed when the batch hits `BATCH_SIZE` (20 messages) **or** `BATCH_TIMEOUT` (50 ms) elapses, whichever comes first. Each flush is one UDP datagram.
72+
- **Sequencing** — every batch gets a monotonic `seq_num` (uint64). This is the unit of loss detection and retransmission; individual CAN messages are never tracked, only whole batches.
73+
- **Wire format** — the UDP payload is a 10-byte header followed by fixed-size messages:
74+
75+
| Field | Type | Bytes | Notes |
76+
|-------|------|-------|-------|
77+
| `seq` | `uint64` big-endian (`!Q`) | 8 | batch sequence number |
78+
| `count` | `uint16` big-endian (`!H`) | 2 | number of messages in this batch |
79+
| `messages` | `count × CANMessage` | 20 each | packed via `CANMessage.pack()` |
80+
81+
- **Ring buffer** — after sending, the car also appends `(seq, batch, timestamp)` to an in-memory `deque` and evicts anything older than `BUFFER_DURATION` (**60 seconds**). This is the *only* copy available for retransmission. A gap that isn't requested within ~60s is gone for good on the live link (it may still be recovered later from the car's own CSV log, but not through this channel).
82+
83+
### Receiver: gap detection state machine (base)
84+
85+
The base tracks a single `expected_seq` and classifies every incoming datagram:
86+
87+
| Condition | Meaning | Action |
88+
|-----------|---------|--------|
89+
| `seq == expected_seq` | in order | process; `expected_seq = seq + 1` |
90+
| `seq > expected_seq` | **gap** | add every seq in `[expected_seq, seq)` to the `missing` set; process; advance `expected_seq` |
91+
| `seq < expected_seq`, in `missing` set | late/out-of-order arrival | remove from `missing`, count as recovered |
92+
| `seq < expected_seq`, not in `missing` | duplicate | ignore |
93+
| `seq < expected_seq − 1000` | **sequence reset** (car restarted) | reset `expected_seq`, clear `missing` |
94+
95+
The `missing` set is capped at **1000 entries**; when it overflows the oldest sequence is evicted (and will never be requested). A per-second status map (`0` = missing, `1` = UDP, `2` = TCP-recovered) is maintained purely for the PECAN link-health visualization.
96+
97+
### Recovery: request / response schema (TCP :5006)
98+
99+
A `missing_reporter` task wakes every `MISSING_CHECK_INTERVAL` (**10 seconds**). If the `missing` set is non-empty it opens a TCP connection to the car's resend server and sends:
100+
101+
```json
102+
{ "missing": [12043, 12044, 12051] }
103+
```
104+
105+
Only the **100 oldest** missing sequences are sent per cycle — `sorted(missing)[:100]` — so that gaps closest to aging out of the car's 60s ring buffer get priority. Newer gaps can wait; they have more buffer time left. The car's `handle_resend` looks each one up in the ring buffer and returns whatever it still holds:
106+
107+
```json
108+
[
109+
{ "seq": 12043, "msgs": [ { "t": 1752345600.12, "id": 256, "d": "a1b2c3..." } ] }
110+
]
111+
```
112+
113+
Recovered messages are re-published to the same `can_messages` Redis channel as the live path, so downstream consumers (WebSocket bridge, TimescaleDB) see one merged stream.
114+
115+
> **Format note:** the resend response uses a JSON shape (`t`/`id`/`d`-hex) that differs from the binary `CANMessage` on the primary UDP path. Both are normalized to the same `{time, canId, data}` shape before hitting Redis, but the two encodings are worth keeping in mind when touching either path.
116+
117+
### Tuning knobs & limits
118+
119+
| Constant | Default | Location | Effect |
120+
|----------|---------|----------|--------|
121+
| `BATCH_SIZE` | `20` | `src/data.py` | messages per UDP datagram |
122+
| `BATCH_TIMEOUT` | `0.05` s | `src/data.py` | max latency before a partial batch is flushed |
123+
| `BUFFER_DURATION` | `60` s | `src/data.py` | how far back the car can retransmit |
124+
| `MISSING_CHECK_INTERVAL` | `10` s | `src/data.py` | how often the base requests resends |
125+
| `UDP_PORT` / `TCP_PORT` | `5005` / `5006` | `src/config.py` | live stream / recovery channel |
126+
127+
**Known limitations** (by design — this is a soft-real-time telemetry link, not a guaranteed-delivery bus):
128+
129+
- Recovery is best-effort. Nothing is retried indefinitely; a gap older than ~60s or beyond the 1000-entry `missing` cap is dropped from the live stream.
130+
- There is no end-to-end ACK, so the car has no knowledge of what the base did or didn't receive — all loss detection is receiver-driven.
131+
132+
---
133+
40134
## Hardware Setup (Ubuntu)
41135

42136
This section covers setting up a CAN HAT (e.g. MCP2515-based) on Ubuntu before running the software.
@@ -238,8 +332,8 @@ Images are built for both `linux/amd64` and `linux/arm64` (Raspberry Pi).
238332

239333
| Port | Protocol | Purpose |
240334
|------|----------|---------|
241-
| 5005 | UDP | CAN data streaming |
242-
| 5006 | TCP | Packet retransmission |
335+
| 5005 | UDP | CAN data streaming (see [Reliability & Retransmission](#reliability--retransmission)) |
336+
| 5006 | TCP | Packet retransmission (see [Reliability & Retransmission](#reliability--retransmission)) |
243337
| 6379 | TCP | Redis (internal) |
244338
| 8080 | HTTP | Status monitoring page |
245339
| 9080 | WebSocket | PECAN dashboard feed |

universal-telemetry-software/src/data.py

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -752,7 +752,7 @@ async def missing_reporter():
752752
logger.info(f"Requesting resend for {len(missing_seqs)} batches")
753753
try:
754754
reader, writer = await asyncio.open_connection(REMOTE_IP, TCP_PORT)
755-
request = {"missing": sorted(list(missing_seqs))[-100:]} # Limit request size
755+
request = {"missing": sorted(list(missing_seqs))[:100]}
756756
writer.write(json.dumps(request).encode())
757757
await writer.drain()
758758

Lines changed: 153 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,153 @@
1+
"""
2+
Unit test for retransmission request-slice behavior.
3+
4+
Reproduces the gap-detection + missing_reporter request-slice logic
5+
from data.py without needing sockets, Redis, or Docker.
6+
7+
The request slice uses sorted()[:100] (oldest first) so that gaps
8+
closest to aging out of the car's 60s ring buffer get priority.
9+
"""
10+
import pytest
11+
12+
13+
def simulate_gap_detection(seq_stream: list[int]):
14+
"""Replay the base receiver's gap-detection state machine.
15+
16+
Returns the final (missing_seqs, expected_seq) after processing all seqs.
17+
This mirrors data.py lines 618-655 exactly.
18+
"""
19+
expected_seq = None
20+
missing_seqs: set[int] = set()
21+
22+
for seq in seq_stream:
23+
if expected_seq is None:
24+
expected_seq = seq
25+
if expected_seq is not None and seq < expected_seq - 1000:
26+
expected_seq = seq
27+
missing_seqs.clear()
28+
if seq > expected_seq:
29+
for s in range(expected_seq, seq):
30+
missing_seqs.add(s)
31+
if len(missing_seqs) > 1000:
32+
oldest = min(missing_seqs)
33+
missing_seqs.remove(oldest)
34+
elif seq < expected_seq:
35+
if seq in missing_seqs:
36+
missing_seqs.remove(seq)
37+
else:
38+
continue
39+
expected_seq = max(expected_seq, seq + 1)
40+
41+
return missing_seqs, expected_seq
42+
43+
44+
def simulate_request_slice(missing_seqs: set[int]) -> list[int]:
45+
"""The exact line from missing_reporter: sorted(list(missing_seqs))[:100]"""
46+
return sorted(list(missing_seqs))[:100]
47+
48+
49+
class TestOldestFirstRecovery:
50+
"""Verify oldest-first request slice prevents starvation."""
51+
52+
def test_small_gap_all_requested(self):
53+
"""<=100 missing seqs: everything gets requested."""
54+
missing = set(range(10, 60))
55+
requested = simulate_request_slice(missing)
56+
assert set(requested) == missing
57+
58+
def test_large_gap_oldest_requested_first(self):
59+
""">100 missing: oldest 100 are requested, not newest."""
60+
missing = set(range(10, 210)) # 200 missing
61+
requested = simulate_request_slice(missing)
62+
63+
assert len(requested) == 100
64+
assert set(requested) == set(range(10, 110))
65+
66+
def test_single_burst_self_heals(self):
67+
"""A burst of 200 missing seqs recovers fully in 2 cycles."""
68+
missing = set(range(10, 210))
69+
70+
# Cycle 1: oldest 100 recovered
71+
requested = simulate_request_slice(missing)
72+
assert set(requested) == set(range(10, 110))
73+
missing -= set(requested)
74+
75+
# Cycle 2: remaining 100 recovered
76+
requested = simulate_request_slice(missing)
77+
assert set(requested) == set(range(110, 210))
78+
missing -= set(requested)
79+
assert len(missing) == 0
80+
81+
def test_sustained_loss_old_gaps_still_recovered(self):
82+
"""
83+
Under sustained loss (>100 new gaps per cycle), oldest-first
84+
ensures old gaps are recovered before they age out of the
85+
car's ring buffer.
86+
"""
87+
missing = set(range(100, 150)) # 50 old gaps
88+
next_seq = 200
89+
90+
for cycle in range(3):
91+
for s in range(next_seq, next_seq + 120):
92+
missing.add(s)
93+
next_seq += 140
94+
95+
requested = simulate_request_slice(missing)
96+
missing -= set(requested)
97+
98+
old_remaining = missing & set(range(100, 150))
99+
assert len(old_remaining) == 0, (
100+
"Oldest-first should recover old gaps even under sustained loss"
101+
)
102+
103+
def test_sustained_loss_newest_delayed_not_lost(self):
104+
"""
105+
Under sustained loss, newest gaps are delayed (not requested
106+
immediately) but eventually get their turn as older ones clear.
107+
"""
108+
missing = set(range(100, 150)) # 50 old
109+
next_seq = 200
110+
all_ever_missing: set[int] = set(missing)
111+
112+
for cycle in range(10):
113+
new_gaps = set(range(next_seq, next_seq + 120))
114+
missing |= new_gaps
115+
all_ever_missing |= new_gaps
116+
next_seq += 140
117+
118+
requested = simulate_request_slice(missing)
119+
missing -= set(requested)
120+
121+
# The set grows but oldest are always drained first.
122+
# With 120 new per cycle and 100 recovered, net growth is 20/cycle.
123+
# After 10 cycles: 200 remaining (all from recent cycles).
124+
# None of the old (100-149) should remain.
125+
old_remaining = missing & set(range(100, 150))
126+
assert len(old_remaining) == 0
127+
128+
def test_1000_cap_evicts_oldest(self):
129+
"""
130+
The missing set cap at 1000 still evicts the oldest seqs.
131+
This is a separate mechanism from the request slice.
132+
"""
133+
stream = [0, 1501]
134+
missing, _ = simulate_gap_detection(stream)
135+
136+
assert len(missing) <= 1000
137+
assert min(missing) == 501
138+
139+
140+
class TestRecoveryRemovesFromMissing:
141+
"""Verify that successful recovery correctly shrinks the missing set."""
142+
143+
def test_recovery_removes_seq(self):
144+
missing = set(range(10, 20))
145+
missing.remove(15)
146+
assert 15 not in missing
147+
assert len(missing) == 9
148+
149+
def test_full_recovery_empties_set(self):
150+
missing = set(range(10, 20))
151+
for seq in range(10, 20):
152+
missing.discard(seq)
153+
assert len(missing) == 0

0 commit comments

Comments
 (0)