From bb23ecc3a99ad77b05e1d3f43b57f2e65703e4c8 Mon Sep 17 00:00:00 2001 From: Tom M0LTE Date: Sat, 6 Jun 2026 12:08:59 +0000 Subject: [PATCH] router: model the channel as a continuous carrier (never drop mid-transmission) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The per-link audio queue was a fixed ~3 s buffered channel that dropped the oldest block on overflow. A TNC bursts a whole keyup's audio with no real-time pacing, so any transmission longer than ~3 s overran the buffer and net-sim dropped blocks *mid-transmission* — gapping the receiver's audio, collapsing its DCD / carrier sense, and making the far end key up on top of the in-progress transmission. That overlap is a collision (the mixer's MixCollision → silence), the frame is lost, and the loss triggers a go-back-N retransmit storm that overruns the buffer further. A real continuous-carrier channel has no such limit. Measured on a real LinBPQ ↔ packet-node link over the sim (a chat help-text dump = back-to-back long I-frames): the receiver decoded only ~26% of the sender's I-frames; ~36-63% of the receiver's own transmissions collided with the sender's (composite-WAV L/R overlap analysis); and the collisions began a median ~3.3 s into each transmission — i.e. exactly when the 3 s buffer saturated and started dropping. Fix: linkQueue becomes a non-dropping FIFO that grows as needed. The real-time rxFeeder still meters it out at the channel sample rate, so channel timing is unchanged, but the receiver now hears a gap-free carrier for the full transmission and its carrier sense holds. Memory is bounded in practice (a keyup is finite; the backing array is released once drained); only a pathological runaway past a 60 s safety cap ever drops, logged as the anomaly it is. After the fix, on the same link: overflow 0, 0 collisions, receiver I-frame delivery 26% → ~90%, the full help text delivered, and the go-back-N storm gone. Adds TestLinkQueueDeliversLongBurstGapFree: a ~10 s burst (far past the old 3 s cap) must be delivered FIFO and gap-free. Co-Authored-By: Claude Opus 4.8 (1M context) --- internal/router/linkqueue_test.go | 45 ++++++++++++++++ internal/router/router.go | 89 ++++++++++++++++++------------- 2 files changed, 96 insertions(+), 38 deletions(-) create mode 100644 internal/router/linkqueue_test.go diff --git a/internal/router/linkqueue_test.go b/internal/router/linkqueue_test.go new file mode 100644 index 0000000..a9aa9f6 --- /dev/null +++ b/internal/router/linkqueue_test.go @@ -0,0 +1,45 @@ +package router + +import ( + "log/slog" + "testing" + + "github.com/packethacking/net-sim/internal/audio" + "github.com/packethacking/net-sim/internal/config" +) + +// A transmission far longer than the old fixed 3 s buffer must be delivered +// gap-free: net-sim models a continuous carrier and must never drop audio +// within a transmission. Dropping mid-transmission gaps the receiver's audio, +// collapses its carrier sense, and makes the far end key up on top of the +// in-progress transmission (a collision) — the bug this non-dropping FIFO +// fixes. Regression for the 3 s drop-on-overflow `linkQueue`. +func TestLinkQueueDeliversLongBurstGapFree(t *testing.T) { + q := newLinkQueue( + config.PortRef{NodeID: "a", PortID: "vhf"}, + config.PortRef{NodeID: "b", PortID: "vhf"}, + 0, 0, + ) + + // ~10 s of audio — well past the old 3 s cap that used to drop. + n := 10 * audio.SampleRate / audio.BlockSamples + for i := 0; i < n; i++ { + blk := make(audio.Block, audio.BlockBytes) + blk[0] = byte(i) + blk[1] = byte(i >> 8) + q.push(blk, slog.Default()) + } + + for i := 0; i < n; i++ { + blk, ok := q.pop() + if !ok { + t.Fatalf("block %d of %d was dropped — the queue must never drop within a transmission", i, n) + } + if got := int(blk[0]) | int(blk[1])<<8; got != i { + t.Fatalf("FIFO order broken at block %d: got marker %d", i, got) + } + } + if _, ok := q.pop(); ok { + t.Fatal("queue should be empty after draining the whole transmission") + } +} diff --git a/internal/router/router.go b/internal/router/router.go index b782b4f..a96cac2 100644 --- a/internal/router/router.go +++ b/internal/router/router.go @@ -146,57 +146,70 @@ type txTracker struct { lastBusyNanos atomic.Int64 } -// linkQueue is one source→destination link's audio buffer. The capacity -// is sized for ~3 s of audio at SampleRate / BlockSamples blocks per -// second — large enough to absorb a full samoyed TX burst (preamble + -// frame + postamble for typical AX.25) without dropping. +// maxQueueBlocks is a safety cap on a single link's audio backlog (~60 s). +// The channel is modelled as a CONTINUOUS CARRIER: blocks are never dropped +// within a transmission (see linkQueue.push). 60 s only triggers on a genuine +// runaway (a wedged rxFeeder), where dropping the head is the lesser evil — a +// real keyup is at most a few seconds. +const maxQueueBlocks = 60 * audio.SampleRate / audio.BlockSamples + +// linkQueue is one source→destination link's audio buffer: a non-dropping FIFO +// of PCM blocks the source has transmitted, drained by the destination's +// rxFeeder at the channel sample rate (so the receiver hears the carrier in +// real time). +// +// A TNC bursts a whole keyup's worth of audio with no real-time pacing on its +// end (see txReader), so the buffer must hold an entire transmission and the +// rxFeeder plays it out at real time. It is non-dropping and grows as needed: +// dropping mid-transmission would gap the receiver's audio, collapse its DCD / +// carrier sense, and make the far end key up on top of an in-progress +// transmission — a collision that cannot happen on a real continuous-carrier +// channel. Memory is bounded in practice (a keyup is finite; the backing array +// is released once drained); only a pathological runaway past maxQueueBlocks +// ever drops. type linkQueue struct { src, dst config.PortRef loss float64 noise float64 - ch chan audio.Block + + mu sync.Mutex + buf []audio.Block // FIFO; index 0 = oldest } func newLinkQueue(src, dst config.PortRef, loss, noise float64) *linkQueue { - const capBlocks = 3 * audio.SampleRate / audio.BlockSamples - return &linkQueue{ - src: src, - dst: dst, - loss: loss, - noise: noise, - ch: make(chan audio.Block, capBlocks), - } + return &linkQueue{src: src, dst: dst, loss: loss, noise: noise} } -// pushNonBlocking enqueues a block; drops oldest if full. Logs the drop. -func (q *linkQueue) pushNonBlocking(blk audio.Block, logger *slog.Logger) { - select { - case q.ch <- blk: - return - default: - } - // Full: drop the oldest block to make room. This indicates the - // downstream rxFeeder is falling behind — usually a sign something - // has stalled rather than a normal-operations event, so log it. - select { - case <-q.ch: - default: - } - select { - case q.ch <- blk: - default: - } - logger.Warn("audio queue overflow", "from", q.src, "to", q.dst) +// push appends a block to the link's buffer. It never blocks and — modelling a +// continuous carrier — never drops within a transmission, no matter how far the +// source TNC has run ahead of real time. The only drop is the maxQueueBlocks +// safety valve, which signals a stalled rxFeeder rather than normal operation, +// so it is logged. +func (q *linkQueue) push(blk audio.Block, logger *slog.Logger) { + q.mu.Lock() + defer q.mu.Unlock() + if len(q.buf) >= maxQueueBlocks { + q.buf[0] = nil + q.buf = q.buf[1:] + logger.Warn("audio queue safety cap reached (rxFeeder stalled?)", "from", q.src, "to", q.dst) + } + q.buf = append(q.buf, blk) } -// pop returns one block if immediately available. +// pop returns one block if immediately available (FIFO, oldest first). func (q *linkQueue) pop() (audio.Block, bool) { - select { - case b := <-q.ch: - return b, true - default: + q.mu.Lock() + defer q.mu.Unlock() + if len(q.buf) == 0 { return nil, false } + blk := q.buf[0] + q.buf[0] = nil // release the reference + q.buf = q.buf[1:] + if len(q.buf) == 0 { + q.buf = nil // release the backing array once drained + } + return blk, true } // Start spawns all samoyed children and begins routing audio. @@ -481,7 +494,7 @@ func (r *Router) txReader(ctx context.Context, ref config.PortRef, c *tnc.Child) } } for _, q := range outgoing { - q.pushNonBlocking(blk, r.logger) + q.push(blk, r.logger) } } }