Module: RAK3172-E (STM32WLE5CC), RAK19007 base
Firmware: RUI3 4.2.3 (custom Arduino sketch; UART1 in RAK_AT_MODE). The same receive code is in the 4.1.0 BSP source and is unchanged in the latest 4.3.0 (HAL_UART_RxCpltCallback).
Mode: LoRa P2P (AT+NWM=0), 915 MHz, SF10 / BW125 / CR 4/5, preamble 8, continuous RX (AT+PRECV=65533)
Host link: UART1 (PB6/PB7) at 115200 baud, Raspberry Pi 4
Previously reported on the RAK forum (“RAK3172-E AT command processor wedges under sustained P2P TX (UART1 only)”, now closed for inactivity). A RAK engineer could not reproduce it with 8-byte AT+PSENDs at 9600 baud every 5 s and asked for our exact commands. We have now found the trigger: command length over 128 bytes, not TX rate.
Summary
When a host sends an AT command longer than 128 bytes in one continuous burst while the radio is receiving, the module sometimes loses bytes from the middle of the command. The damaged line then sits in the AT parser: the command gets no reply and the following commands fail or time out, while +EVT:RXP2P URCs keep arriving normally. Commands of 128 bytes or less never fail, and the same long command written in short pieces with idle gaps between them never fails.
Likely cause (from the RUI3 BSP source)
cores/STM32WLE/component/core/mcu/stm32wle5xx/uhal/uhal_uart.c receives UART1 (and LPUART1) by DMA into a 128-byte buffer (#define BUFFER_SIZE 128):
- A line that ends while the UART goes idle is handled by the IDLE interrupt path, with nothing else in flight. These lines are always fine.
- A line longer than 128 bytes fills the buffer mid-line.
HAL_UART_RxCpltCallbackthen callsHAL_UART_DMAStop, feeds the 128 bytes to the parser, pushes them to the queue, and only then restarts reception withHAL_UART_Receive_DMA. - Bytes that arrive on the wire between
DMAStopand the restart are dropped. At 115200 baud one byte takes about 87 µs, so any delay in that window (a radio RX interrupt, or URC output from the same core) loses bytes.
Suggested fix: use circular DMA with half/full-transfer and IDLE events (HAL_UARTEx_ReceiveToIdle_DMA), or double-buffer, so reception never stops while a full buffer is processed.
Reproduction (no over-the-air TX needed, ~10 minutes)
Script: rak_uart_collision_test.py (full source at the end of this issue; Python 3 + pyserial). Run: python3 rak_uart_collision_test.py --port /dev/ttyAMA0 --duration 600, then --analyze /tmp/rak-uart-collision.jsonl.
- Configure P2P as above and put the radio in continuous RX. A second node transmitting nearby every few seconds makes it fail much faster, since the failures cluster around incoming packets.
- In random order, with 50–300 ms between commands, send one of:
long_bulk: a 170-character unknown command (AT+ZZ0123…) in one write. Expected reply:AT_ERROR.long_chunked: the same 170 characters in 16-byte writes, 2 ms apart. Expected:AT_ERROR.mid_bulk: a 120-character command in one write. Expected:AT_ERROR.short:AT. Expected:OK.
- A command fails if its expected reply doesn’t arrive within 1 s. After a failure the script sends
\r\none byte at a time to clear the parser before continuing.
No AT+PSEND is used here, so the radio never transmits. That rules out TX timing.
Results (RUI 4.2.3, UART1, 115200 baud, 2,873 commands, 115 URCs)
| Command | URC arriving during the write | Sent | Failed |
|---|---|---|---|
| 170 chars, one write | yes | 9 | 6 (67%) |
| 170 chars, one write | no | 744 | 10 (1.3%) |
| 170 chars, 16-byte chunks | either | 709 | 0 |
| 120 chars, one write | either | 728 | 0 |
AT |
either | 683 | 0 |
Every failed command got no reply at all. In the gateway, without the flush, the next command then fails too, because the parser still holds the partial line.
Why your test didn’t reproduce it
AT+PSEND=1234567812345678 is 25 characters, far under 128 bytes, so it never takes the DMA-full path. A full-size P2P packet does: an 80-byte payload is AT+PSEND= + 160 hex characters = 171 characters.
In-service impact and our workaround
On our mesh only full-size AT+PSENDs were affected. Over one night, 5 of 29 two-fragment heartbeats (whose first fragment is ~171 characters) failed, against 0 of about 4,300 shorter sends.
Our host now writes any command over 32 bytes in 32-byte pieces with a short idle gap, so every piece ends on an idle line. Result: a 23-hour soak with 52 full-size AT+PSENDs on UART1 had 0 failures (before: about 1 in 6). Hosts that can’t pace their writes, such as a microcontroller sending one buffered string, will still hit this.
What would help
- Confirmation of the DMA stop/restart gap in
uhal_uart.c, and a fix in a future RUI3 release (circular / ReceiveToIdle DMA). - Whether the LPUART1 (USB) path, which uses the same code, is affected in your testing.
- Until then, documenting a maximum safe single-burst command length (128 bytes) for UART AT hosts.
We can supply full logs (every line timestamped) and test a patched firmware.
rak_uart_collision_test.py
#!/usr/bin/env python3
"""RAK3172 UART collision test: does the AT parser lose input bytes while the
module is emitting a +EVT:RXP2P URC? No over-the-air TX at all.
The radio sits in continuous RX (peers keep transmitting) while the host sends
commands that never transmit, picked at random per iteration:
long_bulk ~170-char unknown command in one write (like a heartbeat PSEND)
mid_bulk ~120-char command in one write (under RUI's 128-byte UART1 DMA
buffer: if the DMA-full stop/restart gap drops bytes, this
should be safe and long_bulk should not)
long_chunked the same line written in CHUNK-byte pieces, CHUNK_GAP_S apart
short bare "AT"
Expected replies: AT_ERROR for the long lines (RUI 4.2.3), OK for AT. A reader
thread timestamps every received line (first and last byte) so each failure can
be matched against URCs that arrived while the command was on the wire.
Log: one JSON object per line ("cmd", "rx", "recover", "config" records).
Analyze with --analyze LOG.
"""
from __future__ import annotations
import argparse
import json
import queue
import random
import sys
import threading
import time
LONG_LEN = 170
MID_LEN = 120
CHUNK = 16
CHUNK_GAP_S = 0.002
REPLY_TIMEOUT_S = 1.0
OVERLAP_SLACK_S = 0.005 # URC bytes seen within this margin of the write count as overlapping
REPLIES = ("OK", "AT_COMMAND_NOT_FOUND", "AT_PARAM_ERROR", "AT_ERROR", "AT_BUSY_ERROR")
class Link:
def __init__(self, port: str, baud: int, log):
import serial
self.ser = serial.Serial(port, baud, timeout=0.05)
self.log = log
self.lines: "queue.Queue[dict]" = queue.Queue()
self._stop = False
self._t = threading.Thread(target=self._reader, daemon=True)
self._t.start()
def _reader(self) -> None:
buf = b""
t_first = None
while not self._stop:
chunk = self.ser.read(self.ser.in_waiting or 1)
now = time.monotonic()
if not chunk:
continue
for b in chunk:
if t_first is None:
t_first = now
buf += bytes([b])
if b == 0x0A:
text = buf.decode(errors="replace").strip()
if text:
rec = {"type": "rx", "t0": t_first, "t1": now, "line": text}
self.log(rec)
self.lines.put(rec)
buf = b""
t_first = None
def write(self, data: bytes, chunked: bool) -> tuple[float, float]:
t0 = time.monotonic()
if chunked:
for i in range(0, len(data), CHUNK):
self.ser.write(data[i:i + CHUNK])
self.ser.flush()
time.sleep(CHUNK_GAP_S)
else:
self.ser.write(data)
self.ser.flush()
return t0, time.monotonic()
def drain_lines(self) -> None:
while not self.lines.empty():
self.lines.get_nowait()
def wait_reply(self, timeout: float) -> tuple[str | None, list[str]]:
"""Return (reply, other_lines). Reply is the first line that is (or
ends with) a known AT reply; URCs and junk go to other_lines."""
deadline = time.monotonic() + timeout
other = []
while True:
left = deadline - time.monotonic()
if left <= 0:
return None, other
try:
rec = self.lines.get(timeout=left)
except queue.Empty:
return None, other
line = rec["line"]
if line.startswith("+EVT:"):
other.append(line[:40])
continue
for r in REPLIES:
if line == r:
return r, other
other.append(line[:60])
for r in REPLIES:
if line.endswith(r):
return "merged:" + r, other
def paced_flush(self) -> None:
for _ in range(3):
for b in b"\r\n":
self.ser.write(bytes([b]))
self.ser.flush()
time.sleep(0.03)
def close(self) -> None:
self._stop = True
self._t.join(timeout=1)
self.ser.close()
def command(link: Link, cmd: str, timeout: float = 2.0) -> str | None:
link.drain_lines()
link.write(cmd.encode() + b"\r\n", chunked=False)
reply, _ = link.wait_reply(timeout)
return reply
def recover(link: Link, log) -> bool:
for attempt in range(6):
link.paced_flush()
time.sleep(0.3)
link.drain_lines()
for _ in range(3):
if command(link, "AT", 1.0) == "OK":
log({"type": "recover", "t": time.monotonic(), "attempts": attempt + 1, "ok": True})
return True
log({"type": "recover", "t": time.monotonic(), "ok": False})
return False
def configure(link: Link, preamble: int, log) -> bool:
link.paced_flush()
time.sleep(0.3)
cmds = ["AT", "AT+PRECV=0", "AT+NWM=0", "AT+PFREQ=915000000", "AT+PSF=10",
"AT+PBW=0", "AT+PCR=0", f"AT+PPL={preamble}", "AT+PTP=10",
"AT+CAD=1", "AT+PRECV=65533"]
for c in cmds:
r = command(link, c, 10.0 if c == "AT+NWM=0" else 3.0)
log({"type": "config", "cmd": c, "reply": r})
if r not in ("OK", "AT_PARAM_ERROR") and not (r or "").endswith("OK"):
print(f"configure failed at {c}: {r}", file=sys.stderr)
return False
return True
def run(args) -> None:
out = open(args.log, "a", buffering=1)
lock = threading.Lock()
def log(rec):
with lock:
out.write(json.dumps(rec) + "\n")
link = Link(args.port, args.baud, log)
if not configure(link, args.preamble, log):
if not (recover(link, log) and configure(link, args.preamble, log)):
sys.exit(1)
print(f"configured; running {args.duration}s of no-TX commands")
end = time.monotonic() + args.duration
filler = "AT+ZZ" + "0123456789ABCDEF" * 20
long_line = filler[:LONG_LEN].encode() + b"\r\n"
mid_line = filler[:MID_LEN].encode() + b"\r\n"
n = 0
while time.monotonic() < end:
kind = random.choice(("long_bulk", "long_chunked", "mid_bulk", "short"))
data = {"long_bulk": long_line, "long_chunked": long_line,
"mid_bulk": mid_line, "short": b"AT\r\n"}[kind]
expect = "OK" if kind == "short" else "AT_ERROR" # RUI 4.2.3 reply to an unknown long line
link.drain_lines()
w0, w1 = link.write(data, chunked=(kind == "long_chunked"))
reply, other = link.wait_reply(REPLY_TIMEOUT_S)
rec = {"type": "cmd", "kind": kind, "w0": w0, "w1": w1, "reply": reply,
"ok": reply == expect, "other": other}
log(rec)
n += 1
if not rec["ok"]:
# A wrong or missing reply may leave a partial line in the parser:
# flush before the next command so failures stay independent.
if not recover(link, log):
print("radio not recovering; stopping", file=sys.stderr)
break
time.sleep(random.uniform(args.min_gap, args.max_gap))
if n % 200 == 0:
print(f"{n} commands")
link.close()
print("done")
def analyze(path: str) -> None:
rx, cmds = [], []
for l in open(path):
d = json.loads(l)
if d["type"] == "rx" and d["line"].startswith("+EVT:RXP2P"):
rx.append((d["t0"], d["t1"]))
elif d["type"] == "cmd":
cmds.append(d)
print(f"URCs: {len(rx)} commands: {len(cmds)}")
def overlaps(c):
a, b = c["w0"] - OVERLAP_SLACK_S, c["w1"] + OVERLAP_SLACK_S
return any(t0 <= b and t1 >= a for t0, t1 in rx)
print(f"{'kind':14s} {'overlap':8s} {'n':>5s} {'fail':>5s} {'fail%':>6s}")
for kind in ("long_bulk", "long_chunked", "mid_bulk", "short"):
for ov in (True, False):
sel = [c for c in cmds if c["kind"] == kind and overlaps(c) == ov]
f = sum(not c["ok"] for c in sel)
pct = f"{100 * f / len(sel):.1f}" if sel else "-"
print(f"{kind:14s} {str(ov):8s} {len(sel):5d} {f:5d} {pct:>6s}")
fails = [c for c in cmds if not c["ok"]]
print("sample failures:")
for c in fails[:12]:
print(f" {c['kind']:13s} reply={c['reply']!s:28s} other={c['other'][:2]}")
def main() -> None:
p = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter)
p.add_argument("--port", default="/dev/ttyAMA0")
p.add_argument("--baud", type=int, default=115200)
p.add_argument("--preamble", type=int, default=8)
p.add_argument("--duration", type=int, default=2400)
p.add_argument("--min-gap", type=float, default=0.05)
p.add_argument("--max-gap", type=float, default=0.3)
p.add_argument("--log", default="/tmp/rak-uart-collision.jsonl")
p.add_argument("--analyze", metavar="LOG")
args = p.parse_args()
if args.analyze:
analyze(args.analyze)
else:
run(args)
if __name__ == "__main__":
main()