benmichael/gods-eye-view
1
0 Fork 0
mirror of https://github.com/bilawalsidhu/gods-eye-view.git synced 2026-11-04 10:24:02 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix(sdr): bound decoder work and stabilize FM playback

Browse Source 
Keep USB reads in a bounded ordered pipeline, acknowledge and cap decoder
work, terminate stopped workers, and buffer short playback jitter. Throttle
display telemetry and compute spectra only while seeking.

Preserve PR #1174's wrapped seeking and fresh-spectrum tagging. Add opt-in
bounded timing diagnostics and decoder/playback lifecycle regressions.

Co-authored-by: Aditya <[email protected]>
This commit is contained in:
Mazeyar Moeini FeizabadiandAditya committed 2026-10-16 18:37:15 +02:00
1 parent 537e99dfa4
commit 240f7ae3da
10 files changed
+2305 -75
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

src/sdr/controller.js
+566 -44
View File
@@ -22,11 +22,45 @@ export { createRememberingWebUsb } from './usbDevices.js';
const FM_MIN_HZ = 87_500_000;
const FM_MAX_HZ = 108_000_000;
// Seek frames step around a lap one step longer than the band, so the frames
// either side of the wrap, clamped into the band, each reach their edge.
const FM_SEEK_STEP_HZ = 1_200_000;
const FM_SEEK_LAP_HZ = FM_MAX_HZ - FM_MIN_HZ + FM_SEEK_STEP_HZ;
const FM_SEEK_LAP_START_HZ = FM_MIN_HZ - FM_SEEK_STEP_HZ / 2;
// 19 frames; the last is one channel short of a full lap.
const FM_SEEK_FRAMES = Math.ceil(FM_SEEK_LAP_HZ / FM_SEEK_STEP_HZ);
const READ_SAMPLE_COUNT = 32_768;
// Bulk reads kept queued at the host. With one read at a time, nothing is
// requested while this thread is busy between reads, and delivered IQ falls
// short of the configured rate; queued reads keep a transfer pending meanwhile.
const READ_PIPELINE_DEPTH = 4;
const RATE_WINDOW_MS = 1_000;
// A bulk read that has not completed in this long is treated as a stalled
// USB transfer: the session is torn down instead of waiting forever.
export const SDR_READ_STALL_MS = 5_000;
// Decoder admission. Every block sent to the worker is numbered and must be
// acknowledged; past these bounds the receiver stops with an overload error
// rather than queueing raw IQ the decoder cannot keep up with. Eight 16 ms
// blocks is 128 ms of input, about forty times the measured decode cost.
export const SDR_WORKER_ADMISSION = Object.freeze({
maxBlocks: 8,
maxBytes: 8 * 65_536,
maxAgeMs: 1_000,
watchdogMs: 250,
// Pending-work diagnostics are published at most this often, never per ack.
reportMs: 1_000,
// A cold module worker can take far longer to load than 8 blocks of input:
// no block is sent before it reports ready, and a worker that never does
// stops the receiver.
readyTimeoutMs: 10_000,
});
const WORKER_READY = Symbol('workerReady');
// SNR and signal levels are display values: published at most this often.
const FM_LEVEL_STATE_MS = 250;
// Diagnostic mode: per-worker ring of recent block timings, and per-report
// samples kept for peaks. Both are bounded.
const DECODER_TIMING_RING = 256;
const DECODER_WINDOW_SAMPLES = 256;
// Upper bound on a graceful close queued behind other USB work; past it the
// raw USB device is closed directly, which aborts any pending transfer. The
// raw close is bounded by the same deadline, so a teardown settles within
@@ -177,6 +211,53 @@ function fenceDevice(device) {
}
}
/** Wall-aligned milliseconds comparable between this page and its workers. */
function absoluteNow() {
const clock = globalThis.performance;
return clock?.timeOrigin !== undefined
? clock.timeOrigin + clock.now()
: Date.now();
}
function newTimingWindow() {
return {
queueMs: [],
computeMs: [],
ackDelayMs: [],
rttMs: [],
audioDelayMs: [],
max: {},
pendingPeak: 0,
oldestPeakMs: 0,
};
}
function percentile(values, fraction) {
const sorted = [...values].sort((a, b) => a - b);
return sorted[
Math.min(sorted.length - 1, Math.floor(fraction * sorted.length))
];
}
/** Monotonic milliseconds for ages and deadlines (wall-clock can jump). */
function monotonicNow() {
return globalThis.performance?.now?.() ?? Date.now();
}
function newUsbGate() {
return { waiting: 0, idle: [], reads: [], retired: false };
}
function wakeUsbGate(gate) {
for (const resolve of gate.idle.splice(0)) resolve();
}
/** Resolves once the gate has no pending control work, or is retired. */
function usbGateIdle(gate) {
if (gate.waiting === 0 || gate.retired) return Promise.resolve();
return new Promise((resolve) => gate.idle.push(resolve));
}
/** Resolve with the operation's outcome, or reject once `ms` elapses. */
function withDeadline(promise, ms, message) {
let timer = null;
@@ -242,6 +323,8 @@ export class SdrController {
providerFactory = null,
readStallMs = SDR_READ_STALL_MS,
closeTimeoutMs = SDR_CLOSE_TIMEOUT_MS,
workerAdmission = SDR_WORKER_ADMISSION,
decoderDiagnostics = false,
storage = defaultStorage(),
webUsb = typeof navigator !== 'undefined' ? navigator.usb : null,
} = {}) {
@@ -294,6 +377,29 @@ export class SdrController {
// Bumped whenever a stalled queue is abandoned; work queued in an older
// epoch refuses to touch whatever device is current when it finally runs.
this._usbEpoch = 0;
// Control admission for one USB epoch: pending control work, read-loop
// waiters and the reads that control work drains. Abandoning the queue
// retires the gate, so nothing from an old session can hold a new one.
this._usbGate = newUsbGate();
this._admission = { ...SDR_WORKER_ADMISSION, ...workerAdmission };
// Blocks sent to the current worker and not yet acknowledged, in order.
this._pendingBlocks = new Map();
this._pendingBytes = 0;
this._nextBlockId = 0;
this._watchdog = null;
this._lastAdmissionReport = 0;
this._ackAnomalies = 0;
this._lastAckId = null;
this._workerSerial = 0;
// Optional timing instrumentation (test and soak use); off in production.
this._decoderDiagnostics = Boolean(decoderDiagnostics);
this._timingRing = [];
this._timingWindow = newTimingWindow();
this._fmLevelStateAt = -Infinity;
this._fmLevelPatch = null;
// Audio player epoch: only samples of the active FM session are accepted.
this._audioSessionId = 0;
this._audioSessionActive = false;
this._readGeneration = 0;
// Bumped by every connect() and stop(); a connect, mode change or
// configuration whose token is stale after an await has been superseded
@@ -313,6 +419,7 @@ export class SdrController {
this._fmFrequencyHz = FM_DEFAULT_HZ;
this._spectrum = null;
this._spectrumSequence = 0;
this._fmTune = 0;
this._spectrumWaiters = new Set();
this._seekToken = 0;
// Every provider opens its device through the selecting WebUSB wrapper.
@@ -370,12 +477,27 @@ export class SdrController {
_queueUsb(operation) {
const epoch = this._usbEpoch;
const run = () => {
const gate = this._usbGate;
// Reads stop being issued from here, and the ones already queued are
// delivered before control work runs, so every pre-control block reaches
// the worker ahead of any reset that follows.
gate.waiting += 1;
const run = async () => {
if (epoch !== this._usbEpoch)
throw new Error('USB work abandoned after a stalled transfer');
await Promise.allSettled(gate.reads);
// A stalled read may have ended this session while it drained.
if (epoch !== this._usbEpoch)
throw new Error('USB work abandoned after a stalled transfer');
return operation();
};
const queued = this._usbTail.then(run, run);
// Only the gate this work entered is updated, even if it settles late.
const done = () => {
gate.waiting -= 1;
if (gate.waiting === 0) wakeUsbGate(gate);
};
queued.then(done, done);
this._usbTail = queued.catch(() => {});
return queued;
}
@@ -384,6 +506,10 @@ export class SdrController {
_resetUsbQueue() {
this._usbEpoch += 1;
this._usbTail = Promise.resolve();
const retired = this._usbGate;
this._usbGate = newUsbGate();
retired.retired = true;
wakeUsbGate(retired);
}
_ensureWorker() {
@@ -392,9 +518,33 @@ export class SdrController {
type: 'module',
});
this._worker = worker;
// A new worker tags its spectra from 0 until its first reset.
this._fmTune = 0;
// Block numbers and pending work belong to one worker.
this._clearPendingBlocks();
this._nextBlockId = 0;
let markReady;
this._workerSerial += 1;
this._timingRing = [];
this._timingWindow = newTimingWindow();
this._lastAckId = null;
worker[WORKER_READY] = {
ready: false,
promise: new Promise((resolve) => (markReady = resolve)),
};
worker.onmessage = (event) => {
if (this._worker !== worker) return;
this._handleWorkerMessage(event.data || {});
const data = event.data || {};
if (data.type === 'ready') {
worker[WORKER_READY].ready = true;
markReady();
return;
}
if (data.type === 'ack') {
this._acknowledgeBlock(data.id, data);
return;
}
this._handleWorkerMessage(data);
};
worker.onerror = (event) => {
if (this._worker !== worker) return;
@@ -411,6 +561,7 @@ export class SdrController {
_discardWorker() {
const worker = this._worker;
this._worker = null;
this._clearPendingBlocks();
if (!worker) return;
worker.onmessage = null;
worker.onerror = null;
@@ -421,9 +572,255 @@ export class SdrController {
}
}
/** Start an audio epoch: the player accepts only this session's samples. */
_beginAudioSession() {
if (!this._audioNode) return;
this._audioSessionId += 1;
this._audioSessionActive = true;
const port = this._audioNode.port;
port.postMessage({
type: 'session',
sessionId: this._audioSessionId,
enabled: true,
});
port.postMessage({ type: 'diagnostics', enabled: true });
}
_endAudioSession() {
if (!this._audioSessionActive) return;
this._audioSessionActive = false;
const port = this._audioNode?.port;
port?.postMessage({ type: 'diagnostics', enabled: false });
port?.postMessage({
type: 'session',
sessionId: this._audioSessionId,
enabled: false,
});
port?.postMessage({ type: 'clear' });
}
/** Send one block to the worker if admission allows; otherwise stop. */
_postSamples(buffer) {
const worker = this._ensureWorker();
const { maxBlocks, maxBytes } = this._admission;
if (
this._pendingBlocks.size >= maxBlocks ||
this._pendingBytes + buffer.byteLength > maxBytes
) {
this._overload('backlog');
return false;
}
const id = this._nextBlockId++;
const postedAt = this._decoderDiagnostics ? absoluteNow() : undefined;
this._pendingBlocks.set(id, {
bytes: buffer.byteLength,
at: monotonicNow(),
postedAt,
});
this._pendingBytes += buffer.byteLength;
if (this._decoderDiagnostics) {
const window = this._timingWindow;
window.pendingPeak = Math.max(
window.pendingPeak,
this._pendingBlocks.size,
);
}
// Spectra are used only by seek, so the decoder computes them only then.
const spectrum = this.state.seeking === true;
worker.postMessage(
postedAt === undefined
? { type: 'samples', buffer, id, spectrum }
: { type: 'samples', buffer, id, spectrum, postedAt },
[buffer],
);
return true;
}
_acknowledgeBlock(id, ack = {}) {
const entry = this._pendingBlocks.get(id);
// Unknown, duplicate or already-cleared ids change nothing.
if (!entry) {
this._ackAnomalies += 1;
return;
}
this._pendingBlocks.delete(id);
this._pendingBytes -= entry.bytes;
this._lastAckId = id;
if (this._decoderDiagnostics && entry.postedAt !== undefined)
this._recordBlockTiming(id, entry.postedAt, ack);
}
/** Diagnostic mode: split one block's round trip into its hops. */
_recordBlockTiming(id, postedAt, ack) {
const received = absoluteNow();
const finite = (value) => (Number.isFinite(value) ? value : null);
const timing = {
id,
postedAt,
// Waiting in the worker's queue before it started this block.
queueMs: finite(ack.receivedAt - postedAt),
computeMs: finite(ack.computeMs),
// From the worker posting the ack to this thread handling it.
ackDelayMs: finite(received - ack.ackPostedAt),
rttMs: received - postedAt,
};
this._timingRing.push(timing);
if (this._timingRing.length > DECODER_TIMING_RING) this._timingRing.shift();
const window = this._timingWindow;
for (const key of ['queueMs', 'computeMs', 'ackDelayMs', 'rttMs']) {
if (timing[key] === null) continue;
const values = window[key];
if (values.length < DECODER_WINDOW_SAMPLES) values.push(timing[key]);
window.max[key] = Math.max(window.max[key] ?? -Infinity, timing[key]);
}
}
/** Diagnostic mode: worker post to this thread, for FM audio messages. */
_recordAudioDelay(workerPostedAt) {
if (!this._decoderDiagnostics || !Number.isFinite(workerPostedAt)) return;
const delay = absoluteNow() - workerPostedAt;
const window = this._timingWindow;
if (window.audioDelayMs.length < DECODER_WINDOW_SAMPLES)
window.audioDelayMs.push(delay);
window.max.audioDelayMs = Math.max(
window.max.audioDelayMs ?? -Infinity,
delay,
);
}
_publishTimingWindow(now) {
const window = this._timingWindow;
this._timingWindow = newTimingWindow();
const summary = {};
for (const key of [
'queueMs',
'computeMs',
'ackDelayMs',
'rttMs',
'audioDelayMs',
]) {
const values = window[key];
summary[key] = values.length
? {
count: values.length,
p95: percentile(values, 0.95),
max: window.max[key],
}
: null;
}
summary.pendingPeak = Math.max(
window.pendingPeak,
this._pendingBlocks.size,
);
summary.oldestPendingPeakMs = Math.max(
window.oldestPeakMs,
this._oldestPendingAgeMs(now),
);
return summary;
}
_clearPendingBlocks() {
this._pendingBlocks.clear();
this._pendingBytes = 0;
}
_oldestPendingAgeMs(now = monotonicNow()) {
const first = this._pendingBlocks.values().next().value;
return first ? now - first.at : 0;
}
/** Bounds unacked age even when the worker hangs and no block arrives. */
_startWatchdog() {
this._stopWatchdog();
this._watchdog = setInterval(() => {
if (!this._device) return;
const now = monotonicNow();
const oldest = this._oldestPendingAgeMs(now);
if (oldest > this._admission.maxAgeMs) {
this._overload('stalled');
return;
}
if (this._decoderDiagnostics) {
const window = this._timingWindow;
window.oldestPeakMs = Math.max(window.oldestPeakMs, oldest);
}
if (now - this._lastAdmissionReport >= this._admission.reportMs) {
this._lastAdmissionReport = now;
this._setState({
decoderPendingBlocks: this._pendingBlocks.size,
decoderPendingBytes: this._pendingBytes,
decoderOldestPendingMs: oldest,
...(this._decoderDiagnostics
? { decoderTimings: this._publishTimingWindow(now) }
: {}),
});
}
}, this._admission.watchdogMs);
this._watchdog.unref?.();
}
_stopWatchdog() {
if (this._watchdog) clearInterval(this._watchdog);
this._watchdog = null;
}
/** Containment: stop with an explicit error instead of queueing input. */
_overload(reason) {
if (!this._device) return;
console.warn(`[SDR] Decoder overload (${reason}); stopping receiver`);
// Captured before stop() clears pending work and discards the worker.
const snapshot = this._overloadSnapshot(reason);
// Published before stopping; stop() preserves it, and nothing is set
// afterwards, so a reconnect started during teardown keeps its own state.
this._setState({
status: 'error',
connected: false,
decoderOverload: reason,
decoderOverloadSnapshot: snapshot,
message:
reason === 'startup'
? 'SDR decoder did not start; receiver stopped'
: reason === 'stalled'
? 'SDR decoder stopped responding; receiver stopped'
: 'SDR decoder fell behind; receiver stopped',
});
void this.stop({ preserveMessage: true });
}
_overloadSnapshot(reason) {
const now = monotonicNow();
const snapshot = {
reason,
worker: this._workerSerial,
pendingBlocks: this._pendingBlocks.size,
pendingBytes: this._pendingBytes,
oldestPendingMs: this._oldestPendingAgeMs(now),
inflightReads: this._usbGate.reads.length,
controlWaiting: this._usbGate.waiting,
lastAckId: this._lastAckId,
nextBlockId: this._nextBlockId,
ackAnomalies: this._ackAnomalies,
};
if (this._decoderDiagnostics) {
snapshot.pending = [...this._pendingBlocks].map(([id, entry]) => ({
id,
ageMs: now - entry.at,
bytes: entry.bytes,
}));
snapshot.timings = this._timingRing.slice();
snapshot.window = this._publishTimingWindow(now);
}
return snapshot;
}
_configureWorker() {
this._ensureWorker().postMessage({
const worker = this._ensureWorker();
// Every configuration starts a new decoder epoch, so output of blocks
// queued before it (another mode, or a failed retune) is never current.
const tune = ++this._fmTune;
worker.postMessage({
type: 'configure',
tune,
mode: this.state.mode,
sampleRate: this.state.sampleRate,
location: this.state.receiverLocation,
@@ -452,6 +849,25 @@ export class SdrController {
outputChannelCount: [1],
},
);
// Local diagnostic candidate: expose playback accounting for the live test.
const audioNode = this._audioNode;
audioNode.port.onmessage = (event) => {
if (
this._audioNode !== audioNode ||
event.data?.type !== 'audioStats' ||
!this._device ||
this.state.mode !== 'fm'
)
return;
const { underruns, waiting, empty, dropped, queued } = event.data;
this._setState({
audioUnderruns: underruns,
audioWaitingFrames: waiting,
audioEmptyFrames: empty,
audioDroppedFrames: dropped,
audioQueuedFrames: queued,
});
};
this._audioNode.connect(this._audioContext.destination);
this._audioNode.port.postMessage({
type: 'volume',
@@ -466,37 +882,78 @@ export class SdrController {
}
_handleWorkerMessage(message) {
// Output of an earlier decoder epoch (before a retune or reconfiguration)
// is stale in every field, not only its audio.
if (message.tune !== this._fmTune) return;
const diagnostics = message.diagnostics;
if (diagnostics && Number.isFinite(diagnostics.workerBlocks)) {
this._setState({
workerBlocks: diagnostics.workerBlocks,
iqLevelDbfs: Number.isFinite(diagnostics.iqLevelDbfs)
? diagnostics.iqLevelDbfs
: this.state.iqLevelDbfs,
audioLevelDbfs: Number.isFinite(diagnostics.audioLevelDbfs)
? diagnostics.audioLevelDbfs
: this.state.audioLevelDbfs,
});
const levels =
diagnostics && Number.isFinite(diagnostics.workerBlocks)
? {
workerBlocks: diagnostics.workerBlocks,
iqLevelDbfs: Number.isFinite(diagnostics.iqLevelDbfs)
? diagnostics.iqLevelDbfs
: this.state.iqLevelDbfs,
audioLevelDbfs: Number.isFinite(diagnostics.audioLevelDbfs)
? diagnostics.audioLevelDbfs
: this.state.audioLevelDbfs,
}
: null;
if (message.type === 'fm') {
this._recordAudioDelay(message.workerPostedAt);
// Display values only: merged and published at most every 250 ms, so
// every subscriber is not run for each 16 ms block.
const patch = {
...(levels || {}),
...(Number.isFinite(message.snr) ? { snr: message.snr } : {}),
};
if (Object.keys(patch).length)
this._fmLevelPatch = { ...(this._fmLevelPatch || {}), ...patch };
const now = monotonicNow();
if (
this._fmLevelPatch &&
now - this._fmLevelStateAt >= FM_LEVEL_STATE_MS
) {
this._fmLevelStateAt = now;
const pending = this._fmLevelPatch;
this._fmLevelPatch = null;
this._setState(pending);
}
} else if (levels) {
this._setState(levels);
}
if (message.type === 'fm') {
// Audio from before the latest retune, from another mode, or from a
// session that has ended never reaches the player.
if (
message.samples instanceof Float32Array &&
message.samples.length > 0 &&
this._audioNode &&
this._device &&
this._audioSessionActive &&
this.state.mode === 'fm' &&
message.tune === this._fmTune &&
!this.state.seeking
) {
this._audioNode.port.postMessage(
{ type: 'samples', samples: message.samples },
{
type: 'samples',
sessionId: this._audioSessionId,
samples: message.samples,
},
[message.samples.buffer],
);
}
if (message.bins instanceof Float32Array) {
// A spectrum of samples from before the latest retune is dropped.
if (
message.bins instanceof Float32Array &&
message.tune === this._fmTune
) {
this._spectrum = message.bins;
this._spectrumSequence += 1;
for (const waiter of this._spectrumWaiters)
waiter(this._spectrumSequence);
this._spectrumWaiters.clear();
}
if (Number.isFinite(message.snr)) this._setState({ snr: message.snr });
return;
}
if (message.type === 'adsb') {
@@ -592,6 +1049,8 @@ export class SdrController {
if (!owned || !this._owns(token, device)) return false;
this._configureWorker();
this._audioNode?.port.postMessage({ type: 'clear' });
if (mode === 'fm') this._beginAudioSession();
else this._endAudioSession();
return true;
}
@@ -665,6 +1124,7 @@ export class SdrController {
audioLevelDbfs: null,
...this._resetAdsbCounters(),
});
this._startWatchdog();
void this._readLoop(this._readGeneration);
return true;
} catch (error) {
@@ -967,35 +1427,75 @@ export class SdrController {
return connected;
}
/** Wait (bounded) until the decoder has loaded; false ends this loop. */
async _decoderReady(generation) {
const startup = this._ensureWorker()[WORKER_READY];
if (!startup || startup.ready) return true;
let timer = null;
await Promise.race([
startup.promise,
new Promise((resolve) => {
timer = setTimeout(resolve, this._admission.readyTimeoutMs);
}),
]);
clearTimeout(timer);
if (generation !== this._readGeneration) return false;
if (startup.ready) return true;
this._overload('startup');
return false;
}
async _readLoop(generation) {
if (!(await this._decoderReady(generation))) return;
// Each read's delivery is chained to it in order, so a drained read has
// also been handed to the worker before any control work runs.
const inflight = [];
let chain = Promise.resolve();
const deliver = (block) => {
if (!this._device || generation !== this._readGeneration) return;
const buffer = block.data;
this._sampleWindowSamples += buffer.byteLength / 2;
const now = Date.now();
const elapsed = now - this._sampleWindowStartedAt;
if (elapsed >= RATE_WINDOW_MS) {
this._setState({
samplesPerSecond: Math.round(
(this._sampleWindowSamples * 1_000) / elapsed,
),
});
this._sampleWindowStartedAt = now;
this._sampleWindowSamples = 0;
}
this._postSamples(buffer);
};
while (this._device && generation === this._readGeneration) {
try {
const device = this._device;
// A transfer that never settles would hold the shared USB queue
// forever; past the deadline the session is torn down (stop() closes
// the raw USB device, which aborts the transfer).
const block = await this._queueUsb(() =>
withDeadline(
const gate = this._usbGate;
// Control work in this epoch drains this loop's reads; a retired
// gate's waiters are released and re-check the session above.
gate.reads = inflight;
// A transfer that never settles would hold control work forever; past
// the deadline the session is torn down (stop() closes the raw USB
// device, which aborts the transfer).
while (gate.waiting === 0 && inflight.length < READ_PIPELINE_DEPTH) {
const read = withDeadline(
device.readSamples(READ_SAMPLE_COUNT),
this._readStallMs,
'RTL-SDR sample read stalled',
),
);
if (!this._device || generation !== this._readGeneration) break;
const buffer = block.data;
this._sampleWindowSamples += buffer.byteLength / 2;
const now = Date.now();
const elapsed = now - this._sampleWindowStartedAt;
if (elapsed >= RATE_WINDOW_MS) {
this._setState({
samplesPerSecond: Math.round(
(this._sampleWindowSamples * 1_000) / elapsed,
),
});
this._sampleWindowStartedAt = now;
this._sampleWindowSamples = 0;
);
read.catch(() => {});
chain = chain.then(() => read).then(deliver);
const handled = chain;
handled.catch(() => {});
inflight.push(handled);
}
this._ensureWorker().postMessage({ type: 'samples', buffer }, [buffer]);
if (!inflight.length) {
await usbGateIdle(gate);
continue;
}
await inflight[0];
inflight.shift();
} catch (error) {
if (generation !== this._readGeneration) break;
console.warn('[SDR] Sample read failed:', error);
@@ -1146,6 +1646,9 @@ export class SdrController {
return true;
}
if (this.state.mode !== 'fm') await this.setMode('fm');
// Spectra are dropped from here until the worker takes this tag, so a
// retune that fails part-way never has them read at the wrong frequency.
const tune = ++this._fmTune;
try {
let actualFrequency = frequency;
await this._queueUsb(async () => {
@@ -1153,7 +1656,7 @@ export class SdrController {
if (Number.isFinite(tuned)) actualFrequency = tuned;
await this._device.resetBuffer();
});
this._worker?.postMessage({ type: 'reset' });
this._worker?.postMessage({ type: 'reset', tune });
this._audioNode?.port.postMessage({ type: 'clear' });
this._setState({
frequencyHz: actualFrequency,
@@ -1200,12 +1703,23 @@ export class SdrController {
});
this._audioNode?.port.postMessage({ type: 'volume', value: 0 });
try {
for (let scan = 0; scan < 18 && token === this._seekToken; scan += 1) {
const span = FM_MAX_HZ - FM_MIN_HZ;
const rawCenter = origin + seekDirection * scan * 1_200_000;
const center = FM_MIN_HZ + positiveModulo(rawCenter - FM_MIN_HZ, span);
for (
let scan = 0;
scan < FM_SEEK_FRAMES && token === this._seekToken;
scan += 1
) {
const rawCenter = origin + seekDirection * scan * FM_SEEK_STEP_HZ;
const position =
FM_SEEK_LAP_START_HZ +
positiveModulo(rawCenter - FM_SEEK_LAP_START_HZ, FM_SEEK_LAP_HZ);
const center = clampFmFrequency(position);
// tuneFm reports the failure; no spectrum is valid for this frame.
if (!(await this.tuneFm(center, { quiet: true }))) {
if (token === this._seekToken)
this._setState({ seeking: false, seekMessage: 'Seek stopped' });
return null;
}
const sequence = this._spectrumSequence;
await this.tuneFm(center, { quiet: true });
this._setState({
seekMessage: `Scanning ${(center / 1_000_000).toFixed(1)} MHz…`,
});
@@ -1219,7 +1733,10 @@ export class SdrController {
this.state.frequencyHz,
this.state.sampleRate,
seekDirection,
origin,
// Past the wrap the origin lies a lap behind this frame.
origin + position - rawCenter,
// A full lap ends back on the origin's own station.
{ skipFrequency: origin },
);
if (!peak) continue;
await this.tuneFm(peak.frequency, { quiet: true });
@@ -1311,6 +1828,11 @@ export class SdrController {
async stop({ preserveMessage = false, message = null } = {}) {
this.cancelSeek();
this._readGeneration += 1;
// Before any await: the worker and its queued input and output are gone,
// and the player refuses this session, so nothing late is handled.
this._stopWatchdog();
this._discardWorker();
this._endAudioSession();
const token = ++this._sessionToken;
const device = this._device;
const provider = this._provider;
src/sdr/controller.test.mjs
+268 -1
View File
@@ -33,7 +33,18 @@ function memoryStorage(initial = {}) {
}
class FakeWorker {
postMessage() {}
constructor() {
// Like the real worker, report ready once loaded.
queueMicrotask(() => this.onmessage?.({ data: { type: 'ready' } }));
}
// Like the real worker, every numbered samples block is acknowledged.
postMessage(message) {
if (message?.type === 'samples' && Number.isSafeInteger(message.id))
queueMicrotask(() =>
this.onmessage?.({ data: { type: 'ack', id: message.id } }),
);
}
terminate() {}
}
@@ -1020,3 +1031,259 @@ test("a device selection that resolves after its acquisition was superseded cann
assert.equal(state.status, 'streaming');
assert.equal(state.connected, true);
});
/** An FM spectrum frame centred on `center` with carriers at `stations`. */
function fmSpectrum(stations, center, sampleRate) {
const bins = new Float32Array(512).fill(-80);
for (const station of stations) {
const index = Math.round(
(station - center) / (sampleRate / bins.length) + bins.length / 2,
);
for (let bin = index - 5; bin <= index + 5; bin += 1) {
if (bin >= 0 && bin < bins.length) bins[bin] = -10;
}
}
return bins;
}
/**
* An open FM receiver hearing `stations`. Like the real worker, spectra of
* samples read before a retune keep arriving after it: one during the retune
* and one after the worker's reset, each drawn at the previous centre and
* tagged with the previous tune, before the first spectrum at the new centre.
*/
function fmSeekController(stations) {
const events = [];
const controller = new SdrController({
storage: memoryStorage(),
webUsb: {},
});
const { sampleRate } = controller.getState();
let center = null;
let workerTune = 0;
const deliver = (tune, at) =>
controller._handleWorkerMessage({
type: 'fm',
bins: fmSpectrum(stations, at, sampleRate),
tune,
});
let previous = null;
controller._worker = {
postMessage(message) {
if (message.type !== 'reset') return;
const stale = [workerTune, previous];
const fresh = [message.tune, center];
workerTune = message.tune;
if (stale[1] !== null) setTimeout(() => deliver(...stale), 0);
setTimeout(() => deliver(...fresh), 1);
},
terminate() {},
};
const device = fakeDevice(events);
const setCenterFrequency = device.setCenterFrequency;
device.setCenterFrequency = async (frequency) => {
if (center !== null) deliver(workerTune, center);
previous = center;
center = frequency;
return setCenterFrequency(frequency);
};
controller._device = device;
return { controller, events };
}
async function seekFrom(origin, direction, stations, prepare = () => {}) {
const { controller, events } = fmSeekController(stations);
await controller.tuneFm(origin);
await new Promise((resolve) => setTimeout(resolve, 0));
prepare(controller);
events.length = 0;
const scanned = new Set();
controller.subscribe(({ seekMessage }) => {
const match = /^Scanning ([\d.]+) MHz/.exec(seekMessage);
if (match) scanned.add(Number(match[1]));
});
const peak = await controller.seekFm(direction);
const tuned = events.filter((event) => event.startsWith('device:tune:'));
return { peak, state: controller.getState(), tuned, scanned };
}
test('FM seek wraps past either band edge and finds the next station', async () => {
const cases = [
// [origin, direction, stations, expected]
[107_500_000, 1, [88_500_000], 88_500_000],
[107_500_000, 1, [106_000_000, 88_500_000, 87_900_000], 87_900_000],
[107_900_000, 1, [87_500_000], 87_500_000],
[88_000_000, -1, [107_000_000], 107_000_000],
[88_000_000, -1, [89_000_000, 107_000_000, 107_600_000], 107_600_000],
[87_600_000, -1, [108_000_000], 108_000_000],
];
for (const [origin, direction, stations, expected] of cases) {
const { peak, state } = await seekFrom(origin, direction, stations);
const label = `${direction > 0 ? 'up' : 'down'} from ${origin}`;
assert.equal(peak?.frequency, expected, label);
assert.equal(state.frequencyHz, expected, label);
assert.equal(state.seeking, false, label);
}
});
test('FM seek reaches both band-edge channels from anywhere in the band', async () => {
for (let origin = 87_600_000; origin < 108_000_000; origin += 1_300_000) {
for (const direction of [1, -1]) {
for (const station of [87_500_000, 108_000_000]) {
const { peak } = await seekFrom(origin, direction, [station]);
assert.equal(
peak?.frequency,
station,
`${direction > 0 ? 'up' : 'down'} from ${origin} to ${station}`,
);
}
}
}
});
test('FM seek laps the band to a station just behind the origin', async () => {
assert.equal(
(await seekFrom(98_500_000, 1, [98_200_000])).peak?.frequency,
98_200_000,
);
assert.equal(
(await seekFrom(98_500_000, -1, [98_800_000])).peak?.frequency,
98_800_000,
);
});
test('FM seek with no other station stops after one lap and returns to the origin', async () => {
for (const stations of [[], [98_500_000]]) {
for (const direction of [1, -1]) {
const { peak, state, tuned, scanned } = await seekFrom(
98_500_000,
direction,
stations,
);
const label = `${stations.length} station(s), direction ${direction}`;
assert.equal(peak, null, label);
assert.equal(state.seeking, false, label);
assert.equal(state.seekMessage, 'No FM signal found', label);
assert.equal(state.frequencyHz, 98_500_000, label);
// 19 frames, then the retune to the origin.
assert.equal(tuned.length, 20, label);
for (const mhz of scanned) {
assert.ok(mhz >= 87.5 && mhz <= 108, `${label}: scanning ${mhz} MHz`);
}
}
}
});
test('FM seek passes over its own station to the next one in the same frame', async () => {
// The first frame's spectrum never arrives, so 98.7 is first seen in the
// last frame of the lap, behind the origin's own carrier.
const { peak } = await seekFrom(
98_500_000,
1,
[98_500_000, 98_700_000],
(controller) => {
const wait = controller._waitForSpectrum.bind(controller);
let first = true;
controller._waitForSpectrum = async (...args) => {
if (!first) return wait(...args);
first = false;
return false;
};
},
);
assert.equal(peak?.frequency, 98_700_000);
});
test('FM seek stops when a retune fails instead of reading the old frame', async () => {
const { peak, state, tuned } = await seekFrom(
98_500_000,
1,
[102_100_000],
(controller) => {
const device = controller._device;
const setCenterFrequency = device.setCenterFrequency;
let calls = 0;
device.setCenterFrequency = async (frequency) => {
calls += 1;
if (calls === 2) throw new Error('transfer error');
return setCenterFrequency(frequency);
};
},
);
assert.equal(peak, null);
assert.equal(state.seeking, false);
assert.equal(state.seekMessage, 'Seek stopped');
assert.equal(state.status, 'error');
assert.equal(tuned.length, 1, 'no frame is tuned after the failure');
});
test('a retune that fails part-way drops spectra until a retune succeeds', async () => {
const resets = [];
const controller = new SdrController({
storage: memoryStorage(),
webUsb: {},
});
const device = fakeDevice();
controller._device = device;
controller._worker = {
postMessage: (message) => {
if (message.type === 'reset') resets.push(message.tune);
},
terminate() {},
};
const spectrum = (tune) =>
controller._handleWorkerMessage({
type: 'fm',
bins: new Float32Array(512),
tune,
});
assert.equal(await controller.tuneFm(98_500_000), true);
const workerTune = resets.at(-1);
spectrum(workerTune);
const accepted = controller._spectrumSequence;
assert.equal(accepted, 1);
// The tuner moves, then the buffer reset fails: the worker is never reset,
// so samples at the new frequency still carry its old tag.
device.resetBuffer = async () => {
throw new Error('transfer error');
};
assert.equal(await controller.tuneFm(99_700_000), false);
assert.deepEqual(resets, [workerTune]);
spectrum(workerTune);
assert.equal(controller._spectrumSequence, accepted);
device.resetBuffer = async () => {};
assert.equal(await controller.tuneFm(99_700_000), true);
spectrum(resets.at(-1));
assert.equal(controller._spectrumSequence, accepted + 1);
});
test("a rebuilt decoder worker's spectra are accepted before its first reset", async (t) => {
const workers = [];
class TaggedWorker {
constructor() {
workers.push(this);
}
postMessage() {}
terminate() {}
}
t.after(replaceGlobal('Worker', TaggedWorker));
const controller = new SdrController({
storage: memoryStorage(),
webUsb: {},
});
controller._device = fakeDevice();
controller._ensureWorker();
await controller.tuneFm(98_500_000);
await controller.tuneFm(99_700_000);
controller._discardWorker();
controller._ensureWorker();
assert.equal(workers.length, 2);
const before = controller._spectrumSequence;
// Like sdr.worker.js, a fresh worker tags its spectra 0 until a reset.
workers[1].onmessage({
data: { type: 'fm', bins: new Float32Array(512), tune: 0 },
});
assert.equal(controller._spectrumSequence, before + 1);
});
src/sdr/decoderDiagnostics.test.mjs
+371
View File
@@ -0,0 +1,371 @@
import assert from 'node:assert/strict';
import test from 'node:test';
import { Worker as ThreadWorker } from 'node:worker_threads';
import { SdrController } from './controller.js';
const ADMISSION = {
maxBlocks: 4,
maxBytes: 4 * 64,
maxAgeMs: 200,
watchdogMs: 10,
reportMs: 20,
readyTimeoutMs: 500,
};
const pause = (ms = 0) => new Promise((resolve) => setTimeout(resolve, ms));
async function until(check, ms = 1_000) {
const deadline = Date.now() + ms;
while (!check()) {
if (Date.now() > deadline) throw new Error('condition not reached');
await pause(2);
}
}
/** Decoder stand-in; with `timings` it answers like the real worker in diagnostic mode. */
function installWorker(t, { ack = 'auto', timings = true } = {}) {
const workers = [];
class TimedWorker {
constructor() {
this.messages = [];
this.terminated = false;
workers.push(this);
setTimeout(() => this.emit({ type: 'ready' }), 0);
}
postMessage(message) {
this.messages.push(message);
if (message?.type !== 'samples') return;
if (ack === 'auto') queueMicrotask(() => this.ack(message));
}
ack(message) {
const now = performance.timeOrigin + performance.now();
this.emit(
timings && Number.isFinite(message.postedAt)
? {
type: 'ack',
id: message.id,
receivedAt: now,
computeMs: 0.5,
ackPostedAt: now,
}
: { type: 'ack', id: message.id },
);
}
emit(data) {
this.onmessage?.({ data });
}
terminate() {
this.terminated = true;
}
}
const prior = globalThis.Worker;
globalThis.Worker = TimedWorker;
t.after(() => {
globalThis.Worker = prior;
});
return workers;
}
function streaming(t, { reads = Infinity, diagnostics = false } = {}) {
const controller = new SdrController({
storage: { getItem: () => null, setItem() {} },
webUsb: {},
workerAdmission: ADMISSION,
decoderDiagnostics: diagnostics,
});
const audio = [];
controller._audioNode = {
port: { postMessage: (message) => audio.push(message) },
};
let served = 0;
const device = {
readSamples: () =>
served++ < reads
? new Promise((resolve) =>
setTimeout(() => resolve({ data: new ArrayBuffer(64) }), 1),
)
: new Promise(() => {}),
close: async () => {},
};
controller._device = device;
controller._setState({ mode: 'fm', status: 'streaming', connected: true });
controller._beginAudioSession();
controller._readGeneration += 1;
controller._startWatchdog();
void controller._readLoop(controller._readGeneration);
t.after(() => {
controller._stopWatchdog();
controller._readGeneration += 1;
controller._device = null;
});
return { controller, audio };
}
test('FM SNR and levels publish at most every 250 ms, while audio and spectrum stay immediate', async (t) => {
// Acknowledging decoder, so the 200 ms age bound does not stop the session.
const workers = installWorker(t);
const { controller, audio } = streaming(t, { reads: 1 });
await until(() => workers[0]?.messages.some((m) => m.type === 'samples'));
const worker = workers[0];
const tune = controller._fmTune;
// subscribe() reports the full state now and on every change, so count
// actual changes of the published SNR value.
let snrUpdates = 0;
let lastSnr;
let first = true;
controller.subscribe((state) => {
if (first) {
first = false;
lastSnr = state.snr;
return;
}
if (state.snr !== lastSnr) {
snrUpdates += 1;
lastSnr = state.snr;
}
});
const spectrumBefore = controller._spectrumSequence;
for (let index = 0; index < 20; index += 1) {
worker.emit({
type: 'fm',
tune,
snr: index,
samples: new Float32Array(4),
bins: new Float32Array(16),
diagnostics:
index === 7
? { workerBlocks: 700, iqLevelDbfs: -9, audioLevelDbfs: -12 }
: null,
});
}
const forwarded = audio.filter((m) => m.type === 'samples').length;
assert.equal(forwarded, 20, 'every block of audio forwarded immediately');
assert.equal(
controller._spectrumSequence - spectrumBefore,
20,
'spectrum immediate',
);
assert.equal(snrUpdates, 1, 'one display update for 20 blocks');
assert.equal(controller.getState().snr, 0);
await pause(260);
worker.emit({ type: 'fm', tune, snr: 99, samples: new Float32Array(4) });
assert.equal(snrUpdates, 2);
const state = controller.getState();
assert.equal(state.snr, 99, 'latest value published');
assert.equal(state.workerBlocks, 700, 'a level update held back is not lost');
assert.equal(state.iqLevelDbfs, -9);
});
test('diagnostics are off by default: no timestamps, no timing state', async (t) => {
const workers = installWorker(t);
const { controller } = streaming(t, { reads: 30 });
await until(
() => workers[0]?.messages.filter((m) => m.type === 'samples').length >= 30,
);
await pause(40);
const samples = workers[0].messages.filter((m) => m.type === 'samples');
assert.ok(samples.every((m) => !('postedAt' in m)));
assert.equal(controller.getState().decoderTimings, undefined);
assert.equal(controller._timingRing.length, 0);
});
test('diagnostic timings: fixed-size per-second summaries; the 256-record ring stays internal', async (t) => {
const workers = installWorker(t);
const { controller } = streaming(t, { reads: 300, diagnostics: true });
const published = [];
controller.subscribe((state) => {
if (state.decoderTimings) published.push(state.decoderTimings);
});
await until(
() =>
workers[0]?.messages.filter((m) => m.type === 'samples').length >= 300,
5_000,
);
await pause(40);
assert.equal(controller._timingRing.length, 256, 'ring capped at 256');
assert.ok(published.length > 0);
for (const summary of published) {
assert.deepEqual(Object.keys(summary).sort(), [
'ackDelayMs',
'audioDelayMs',
'computeMs',
'oldestPendingPeakMs',
'pendingPeak',
'queueMs',
'rttMs',
]);
assert.ok(!('timings' in summary) && !Array.isArray(summary.queueMs));
if (summary.computeMs) {
assert.ok(summary.computeMs.count <= 256);
assert.equal(summary.computeMs.max, 0.5);
}
assert.ok(summary.pendingPeak <= ADMISSION.maxBlocks);
}
});
test('an overload freezes one snapshot before pending work is cleared', async (t) => {
const workers = installWorker(t, { ack: 'manual' });
const { controller } = streaming(t, { diagnostics: true });
await until(() => controller.getState().status === 'error');
const snapshot = controller.getState().decoderOverloadSnapshot;
assert.equal(snapshot.reason, 'backlog');
assert.equal(snapshot.pendingBlocks, ADMISSION.maxBlocks);
assert.equal(snapshot.pendingBytes, ADMISSION.maxBytes);
assert.equal(snapshot.pending.length, ADMISSION.maxBlocks);
assert.ok(snapshot.pending.every((entry) => entry.bytes === 64));
assert.ok(Number.isFinite(snapshot.inflightReads));
assert.ok(Array.isArray(snapshot.timings) && snapshot.timings.length <= 256);
assert.equal(controller._pendingBlocks.size, 0, 'cleared after the snapshot');
assert.equal(workers[0].terminated, true);
});
test('without diagnostics the overload snapshot keeps counts only', async (t) => {
installWorker(t, { ack: 'manual' });
const { controller } = streaming(t);
await until(() => controller.getState().status === 'error');
const snapshot = controller.getState().decoderOverloadSnapshot;
assert.equal(snapshot.pendingBlocks, ADMISSION.maxBlocks);
assert.equal(snapshot.pending, undefined);
assert.equal(snapshot.timings, undefined);
});
test('the real decoder returns timings only for stamped blocks', async (t) => {
const module = new URL('./sdr.worker.js', import.meta.url).href;
const thread = new ThreadWorker(
`const { parentPort } = require('node:worker_threads');
globalThis.self = globalThis;
self.postMessage = (m, transfer) => parentPort.postMessage(m, transfer);
parentPort.on('message', (data) => self.onmessage && self.onmessage({ data }));
import(${JSON.stringify(module)});`,
{ eval: true },
);
t.after(() => thread.terminate());
const inbox = [];
thread.on('message', (message) => inbox.push(message));
await until(() => inbox.some((m) => m.type === 'ready'), 5_000);
thread.postMessage({
type: 'configure',
mode: 'fm',
sampleRate: 2_048_000,
tune: 1,
});
const block = () => {
const bytes = new Uint8Array(32_768 * 2).fill(128);
return bytes.buffer;
};
const stamped = block();
const postedAt = performance.timeOrigin + performance.now();
thread.postMessage({ type: 'samples', buffer: stamped, id: 0, postedAt }, [
stamped,
]);
const plain = block();
thread.postMessage({ type: 'samples', buffer: plain, id: 1 }, [plain]);
await until(() => inbox.filter((m) => m.type === 'ack').length === 2, 5_000);
const [timed, untimed] = inbox.filter((m) => m.type === 'ack');
assert.ok(
timed.receivedAt >= postedAt - 5,
'receipt after posting (clock skew tolerance)',
);
assert.ok(timed.computeMs >= 0 && timed.computeMs < 5_000);
assert.ok(timed.ackPostedAt >= timed.receivedAt);
assert.deepEqual(Object.keys(untimed).sort(), ['id', 'type']);
const fm = inbox.filter((m) => m.type === 'fm');
assert.ok(Number.isFinite(fm[0].workerPostedAt));
assert.equal(fm[1].workerPostedAt, undefined);
});
test('blocks request a spectrum only while seeking', (t) => {
// Deterministic: the _postSamples contract directly, no read loop or polling.
const workers = installWorker(t, { ack: 'manual' });
const controller = new SdrController({
storage: { getItem: () => null, setItem() {} },
webUsb: {},
workerAdmission: ADMISSION,
});
const post = () => {
assert.equal(controller._postSamples(new ArrayBuffer(64)), true);
const sent = workers[0].messages.filter((m) => m.type === 'samples');
return sent[sent.length - 1];
};
controller._setState({ seeking: false });
assert.equal(post().spectrum, false, 'listening: no spectrum');
controller._setState({ seeking: true });
assert.equal(post().spectrum, true, 'seeking: spectrum requested');
controller._setState({ seeking: false });
assert.equal(post().spectrum, false, 'after seek: no spectrum');
// The diagnostic branch carries the same flag.
controller._decoderDiagnostics = true;
controller._setState({ seeking: true });
const stamped = post();
assert.equal(stamped.spectrum, true);
assert.ok(Number.isFinite(stamped.postedAt));
});
test('the real decoder computes spectra only when requested, with audio unchanged', async (t) => {
const module = new URL('./sdr.worker.js', import.meta.url).href;
const thread = new ThreadWorker(
`const { parentPort } = require('node:worker_threads');
globalThis.self = globalThis;
self.postMessage = (m, transfer) => parentPort.postMessage(m, transfer);
parentPort.on('message', (data) => self.onmessage && self.onmessage({ data }));
import(${JSON.stringify(module)});`,
{ eval: true },
);
t.after(() => thread.terminate());
const inbox = [];
thread.on('message', (message) => inbox.push(message));
await until(() => inbox.some((m) => m.type === 'ready'), 5_000);
const tone = () => {
const bytes = new Uint8Array(32_768 * 2);
for (let n = 0; n < 32_768; n += 1) {
const phase = (2 * Math.PI * 100_000 * n) / 2_048_000;
bytes[2 * n] = Math.round(127.5 + 60 * Math.cos(phase));
bytes[2 * n + 1] = Math.round(127.5 + 60 * Math.sin(phase));
}
return bytes.buffer;
};
const run = async (spectrum, firstId) => {
// A reset starts each run at block 0 with identical decoder state.
thread.postMessage({ type: 'reset', tune: 1 });
thread.postMessage({
type: 'configure',
mode: 'fm',
sampleRate: 2_048_000,
tune: 1,
});
for (let index = 0; index < 6; index += 1) {
const buffer = tone();
thread.postMessage(
{ type: 'samples', buffer, id: firstId + index, spectrum },
[buffer],
);
}
await until(
() =>
inbox.filter((m) => m.type === 'ack' && m.id >= firstId).length === 6,
5_000,
);
return inbox.filter((m) => m.type === 'fm').splice(-6);
};
const listening = await run(false, 0);
const seeking = await run(true, 100);
assert.ok(
listening.every((m) => m.bins === null),
'no FFT while listening',
);
assert.equal(
seeking.filter((m) => m.bins instanceof Float32Array).length,
2,
'every third block',
);
assert.deepEqual(
listening.map((m) => m.samples.length),
seeking.map((m) => m.samples.length),
'audio output unchanged',
);
assert.deepEqual(
Array.from(listening[5].samples),
Array.from(seeking[5].samples),
);
});
src/sdr/readPipeline.diag.test.mjs
+226
View File
@@ -0,0 +1,226 @@
// Diagnostic prototype tests (isolated copy only): pipelined reads and control drain ordering.
import assert from 'node:assert/strict';
import test from 'node:test';
import { SdrController } from './controller.js';
function rig() {
const log = [];
const reads = [];
const c = new SdrController({
storage: { getItem: () => null, setItem() {} },
webUsb: {},
});
c._worker = {
postMessage(m) {
if (m.type === 'samples') log.push(`block:${m.buffer.byteLength}`);
else log.push(`worker:${m.type}`);
},
terminate() {},
};
c._device = {
readSamples: () =>
new Promise((resolve, reject) => {
const id = reads.length;
reads.push({ resolve, reject, id });
log.push(`issue:${id}`);
}),
};
c._readGeneration += 1;
void c._readLoop(c._readGeneration);
const settle = () => new Promise((r) => setTimeout(r, 0));
const finish = (id) => reads[id].resolve({ data: new ArrayBuffer(8 + id) });
return {
c,
log,
reads,
settle,
finish,
rejectFirst: () => reads[0].reject(new Error('transfer error')),
stop: () => {
c._readGeneration += 1;
c._device = null;
},
};
}
test('reads stay queued at the host while a completed block is handled', async () => {
const r = rig();
await r.settle();
assert.equal(
r.reads.length,
4,
'four reads outstanding before any completes',
);
r.finish(0);
await r.settle();
assert.equal(
r.reads.length,
5,
'a replacement read is issued as soon as one completes',
);
assert.deepEqual(
r.log.filter((e) => e.startsWith('block')),
['block:8'],
);
r.stop();
});
test('control work waits for in-flight reads, and no read is issued until it finishes', async () => {
const r = rig();
await r.settle();
let ran = false;
const control = r.c._queueUsb(async () => {
r.log.push('control');
ran = true;
});
await r.settle();
assert.equal(ran, false, 'control waits for queued reads');
for (const id of [0, 1, 2, 3]) r.finish(id);
await control;
await r.settle();
const at = r.log.indexOf('control');
assert.deepEqual(
r.log.slice(at - 4, at),
['block:8', 'block:9', 'block:10', 'block:11'],
'every pre-control block is delivered first',
);
assert.ok(
r.log.slice(at + 1).some((e) => e.startsWith('issue:')),
'reads resume after control',
);
assert.ok(
!r.log
.slice(r.log.indexOf('issue:3') + 1, at)
.some((e) => e.startsWith('issue:')),
'no read issued while control pending',
);
r.stop();
});
test('a retune posts its tagged reset only after every pre-retune block reached the worker', async () => {
const r = rig();
Object.assign(r.c._device, {
setCenterFrequency: async (f) => {
r.log.push(`tune:${f}`);
return f;
},
resetBuffer: async () => {
r.log.push('resetBuffer');
},
});
r.c._setState({ mode: 'fm', status: 'streaming' });
await r.settle();
const tuned = r.c.tuneFm(99_000_000, { quiet: true });
await r.settle();
assert.ok(
!r.log.includes('tune:99000000'),
'tuner untouched while reads are in flight',
);
for (const id of [0, 1, 2, 3]) r.finish(id);
assert.equal(await tuned, true);
await r.settle();
const reset = r.log.indexOf('worker:reset');
const tune = r.log.indexOf('tune:99000000');
const preBlocks = r.log.slice(0, tune).filter((e) => e.startsWith('block'));
assert.deepEqual(preBlocks, ['block:8', 'block:9', 'block:10', 'block:11']);
assert.ok(tune < reset, 'tuner and buffer reset before the worker reset');
assert.ok(
!r.log.slice(tune, reset).some((e) => e.startsWith('block')),
'no block between retune and reset',
);
r.stop();
});
function newSession(r, name) {
const reads = [];
r.c._device = {
readSamples: () =>
new Promise((resolve, reject) => {
reads.push({ resolve, reject });
r.log.push(`${name}:issue:${reads.length - 1}`);
}),
};
r.c._readGeneration += 1;
void r.c._readLoop(r.c._readGeneration);
return reads;
}
test('a hung control operation cannot block a new session after the queue is abandoned', async () => {
const r = rig();
await r.settle();
let release;
const hung = r.c._queueUsb(
() => new Promise((resolve) => (release = resolve)),
);
for (const id of [0, 1, 2, 3]) r.finish(id);
await r.settle();
assert.equal(typeof release, 'function', 'old control work is running');
// Forced teardown: abandon the queue, end the old generation, open a new session.
r.c._resetUsbQueue();
r.c._readGeneration += 1;
const reads = newSession(r, 'new');
await r.settle();
assert.equal(
reads.length,
4,
'the new session streams despite the hung old work',
);
// The old work settles late: it must not touch the new gate.
release();
await hung;
await r.settle();
assert.equal(r.c._usbGate.waiting, 0);
let ran = false;
const control = r.c._queueUsb(async () => (ran = true));
await r.settle();
assert.equal(ran, false, 'new control work still waits for the new reads');
reads.forEach((read) => read.resolve({ data: new ArrayBuffer(4) }));
await control;
assert.equal(ran, true);
r.c._readGeneration += 1;
r.c._device = null;
});
test('blocks read by an ended session are never delivered after reconnect', async () => {
const r = rig();
await r.settle();
assert.equal(r.reads.length, 4);
// Session ends with four reads outstanding; a new session starts.
r.c._readGeneration += 1;
const reads = newSession(r, 'new');
await r.settle();
for (const id of [0, 1, 2, 3]) r.finish(id); // old reads complete late
await r.settle();
assert.ok(
!r.log.some((e) => /^block:(8|9|10|11)$/.test(e)),
'no old block reached the worker',
);
reads[0].resolve({ data: new ArrayBuffer(100) });
await r.settle();
assert.ok(r.log.includes('block:100'), 'the new session delivers');
r.c._readGeneration += 1;
r.c._device = null;
});
test('a read that fails while control work is pending still lets the drain settle', async () => {
const r = rig();
await r.settle();
r.c.stop = async () => {
r.log.push('stop');
return true;
};
let ran = false;
const control = r.c._queueUsb(async () => (ran = true));
// First read fails (as a stalled or aborted transfer would); the rest complete.
r.finish(1);
r.finish(2);
r.finish(3);
await r.settle();
assert.equal(ran, false, 'drain waits for the outstanding failing read');
// Reject the first transfer.
r.rejectFirst();
await control;
await r.settle();
assert.equal(ran, true, 'control work runs once every read has settled');
assert.ok(r.log.includes('stop'), 'the failed read ended the session');
});
src/sdr/sdr.worker.js
+55 -2
View File
@@ -14,6 +14,9 @@ let sampleRate = 2_048_000;
let receiverLocation = null;
let demodulator = null;
let blockCount = 0;
let tune = 0;
// True while handling a block the controller stamped for diagnostics.
let timedBlock = false;
const aircraft = new Map();
// Positions refused by the decoder's speed check since the last clear.
const decoderStats = { positionsRejected: 0 };
@@ -85,9 +88,49 @@ function publicAircraft() {
return records;
}
/**
* Every samples message the controller numbered is acknowledged once, after
* any decoder output, whatever this worker did with it (FM, ADS-B, nothing,
* or an error, which still reaches onerror). The controller bounds unacked
* work, so an acknowledgement must never depend on producing output.
*/
self.onmessage = (event) => {
const message = event.data || {};
if (message.type !== 'samples' || !Number.isSafeInteger(message.id)) {
handleMessage(message);
return;
}
// Diagnostic mode only: the controller stamped the block, so time it here.
const timed = Number.isFinite(message.postedAt);
const receivedAt = timed ? absoluteNow() : 0;
const started = timed ? performance.now() : 0;
timedBlock = timed;
try {
handleMessage(message);
} finally {
timedBlock = false;
self.postMessage(
timed
? {
type: 'ack',
id: message.id,
receivedAt,
computeMs: performance.now() - started,
ackPostedAt: absoluteNow(),
}
: { type: 'ack', id: message.id },
);
}
};
/** Wall-aligned milliseconds comparable across the page and its workers. */
function absoluteNow() {
return performance.timeOrigin + performance.now();
}
function handleMessage(message) {
if (message.type === 'configure') {
if (Number.isSafeInteger(message.tune)) tune = message.tune;
configure(message.mode, message.sampleRate, message.location);
return;
}
@@ -97,6 +140,7 @@ self.onmessage = (event) => {
}
if (message.type === 'reset') {
blockCount = 0;
tune = message.tune;
adsbStream.reset();
if (message.clearAircraft) {
aircraft.clear();
@@ -110,7 +154,11 @@ self.onmessage = (event) => {
blockCount += 1;
if (mode === 'fm') {
const bins = blockCount % 3 === 0 ? analyzeSpectrum(message.buffer) : null;
// Only seek reads spectra: skip the FFT and its transfer otherwise.
const bins =
message.spectrum === true && blockCount % 3 === 0
? analyzeSpectrum(message.buffer)
: null;
const { i, q, iqLevelDbfs } = normalizedIq(message.buffer);
const decoded = demodulator.demodulate(i, q, 0);
// The demodulator owns pooled output arrays. Transfer a copy so detaching
@@ -119,8 +167,10 @@ self.onmessage = (event) => {
self.postMessage(
{
type: 'fm',
...(timedBlock ? { workerPostedAt: absoluteNow() } : {}),
samples,
bins,
tune,
snr:
Number.isFinite(decoded.snr) && decoded.snr > 0
? 10 * Math.log10(decoded.snr)
@@ -151,6 +201,7 @@ self.onmessage = (event) => {
if (decodedCount || removed || blockCount % 30 === 0) {
self.postMessage({
type: 'adsb',
tune,
aircraft: publicAircraft(),
decodedCount,
positionsRejected: decoderStats.positionsRejected,
@@ -163,6 +214,8 @@ self.onmessage = (event) => {
: null,
});
}
};
}
configure(mode, sampleRate, receiverLocation);
// Module loaded and configured: the controller may start sending blocks.
self.postMessage({ type: 'ready' });
src/sdr/sdrAudioWorklet.js
+143 -28
View File
@@ -1,44 +1,159 @@
class GevSdrAudioPlayer extends AudioWorkletProcessor {
constructor() {
constructor(options = {}) {
super();
const config = options.processorOptions || {};
const maxFrames = Number.isFinite(config.maxFrames)
? Math.floor(config.maxFrames)
: Math.round(sampleRate * 0.256);
this.maxFrames = Math.min(Math.round(sampleRate), Math.max(1, maxFrames));
const prebufferFrames = Number.isFinite(config.prebufferFrames)
? Math.floor(config.prebufferFrames)
: Math.round(sampleRate * 0.08);
this.prebufferFrames = Math.min(
Math.max(0, prebufferFrames),
this.maxFrames,
);
this.queue = [];
this.offset = 0;
this.queuedFrames = 0;
this.volume = 0.8;
this.port.onmessage = (event) => {
if (
event.data?.type === 'samples' &&
event.data.samples instanceof Float32Array
) {
this.queue.push(event.data.samples);
if (this.queue.length > 48)
this.queue.splice(0, this.queue.length - 48);
} else if (event.data?.type === 'volume') {
this.volume = Math.max(0, Math.min(1, Number(event.data.value) || 0));
} else if (event.data?.type === 'clear') {
this.queue.length = 0;
this.offset = 0;
}
this.sessionId = null;
this.enabled = false;
this.active = false;
this.playing = false;
this.diagnosticsEnabled = false;
this.diagnosticFrames = 0;
this.stats = {
enqueued: 0,
consumed: 0,
empty: 0,
waiting: 0,
dropped: 0,
cleared: 0,
underruns: 0,
output: 0,
};
this.port.onmessage = (event) => this.receive(event.data);
}
reportStats() {
this.port.postMessage({
type: 'audioStats',
...this.stats,
queued: this.queuedFrames,
active: this.active,
sessionId: this.sessionId,
enabled: this.enabled,
playing: this.playing,
});
}
receive(message) {
if (message?.type === 'session') {
if (!Number.isSafeInteger(message.sessionId) || message.sessionId < 0)
return;
if (message.enabled === true) {
// A retired epoch cannot be re-enabled by a late control message.
if (
this.sessionId !== null &&
(message.sessionId < this.sessionId ||
(message.sessionId === this.sessionId && !this.enabled))
)
return;
if (this.sessionId !== message.sessionId) {
this.receive({ type: 'clear' });
this.sessionId = message.sessionId;
}
this.enabled = true;
} else if (
message.enabled === false &&
this.sessionId === message.sessionId
) {
this.enabled = false;
this.receive({ type: 'clear' });
}
} else if (
message?.type === 'samples' &&
this.enabled &&
message.sessionId === this.sessionId &&
message.samples instanceof Float32Array &&
message.samples.length > 0
) {
this.active = true;
this.queue.push(message.samples);
this.queuedFrames += message.samples.length;
this.stats.enqueued += message.samples.length;
let drop = Math.max(0, this.queuedFrames - this.maxFrames);
this.stats.dropped += drop;
while (drop > 0 && this.queue.length > 0) {
const count = Math.min(drop, this.queue[0].length - this.offset);
this.offset += count;
this.queuedFrames -= count;
drop -= count;
if (this.offset === this.queue[0].length) {
this.queue.shift();
this.offset = 0;
}
}
} else if (message?.type === 'clear') {
this.stats.cleared += this.queuedFrames;
this.queue.length = 0;
this.offset = 0;
this.queuedFrames = 0;
this.active = false;
this.playing = false;
this.diagnosticFrames = 0;
if (this.diagnosticsEnabled) this.reportStats();
} else if (message?.type === 'volume') {
this.volume = Math.max(0, Math.min(1, Number(message.value) || 0));
} else if (message?.type === 'diagnostics') {
this.diagnosticsEnabled = message.enabled === true;
this.diagnosticFrames = 0;
} else if (message?.type === 'stats') {
this.reportStats();
}
}
process(_inputs, outputs) {
const output = outputs[0]?.[0];
if (!output) return true;
output.fill(0);
let outputOffset = 0;
while (outputOffset < output.length && this.queue.length) {
const samples = this.queue[0];
const available = samples.length - this.offset;
const count = Math.min(output.length - outputOffset, available);
for (let index = 0; index < count; index += 1) {
output[outputOffset + index] =
samples[this.offset + index] * this.volume;
if (!this.active) return true;
this.stats.output += output.length;
if (!this.playing && this.queuedFrames >= this.prebufferFrames)
this.playing = true;
if (!this.playing) {
this.stats.waiting += output.length;
} else {
let written = 0;
while (written < output.length && this.queue.length > 0) {
const samples = this.queue[0];
const count = Math.min(
output.length - written,
samples.length - this.offset,
);
for (let index = 0; index < count; index += 1)
output[written + index] = samples[this.offset + index] * this.volume;
written += count;
this.offset += count;
this.queuedFrames -= count;
this.stats.consumed += count;
if (this.offset === samples.length) {
this.queue.shift();
this.offset = 0;
}
}
outputOffset += count;
this.offset += count;
if (this.offset >= samples.length) {
this.queue.shift();
this.offset = 0;
if (written < output.length) {
this.stats.empty += output.length - written;
this.stats.underruns += 1;
this.playing = false;
}
}
if (this.diagnosticsEnabled) {
this.diagnosticFrames += output.length;
if (this.diagnosticFrames >= sampleRate) {
this.diagnosticFrames %= sampleRate;
this.reportStats();
}
}
return true;
src/sdr/sdrAudioWorklet.test.mjs
+234
View File
@@ -0,0 +1,234 @@
import assert from 'node:assert/strict';
import { readFileSync } from 'node:fs';
import vm from 'node:vm';
import test from 'node:test';
const source = readFileSync(
new URL('./sdrAudioWorklet.js', import.meta.url),
'utf8',
);
function rig(options = {}, rate = 48_000) {
let Processor;
const reports = [];
vm.runInNewContext(source, {
Float32Array,
sampleRate: rate,
AudioWorkletProcessor: class {
constructor() {
this.port = { postMessage: (m) => reports.push(m) };
}
},
registerProcessor: (_name, implementation) => {
Processor = implementation;
},
});
const player = new Processor({ processorOptions: options });
const send = (message) => player.port.onmessage({ data: message });
const enable = (sessionId = 1) =>
send({ type: 'session', sessionId, enabled: true });
const samples = (array, sessionId = 1) =>
send({ type: 'samples', samples: array, sessionId });
const tick = (length = 128) => {
const out = new Float32Array(length);
assert.equal(player.process([], [[out]]), true);
return out;
};
const snapshot = () => {
send({ type: 'stats' });
return reports.at(-1);
};
const conserve = () => {
const s = snapshot();
assert.equal(s.enqueued, s.consumed + s.dropped + s.cleared + s.queued);
assert.equal(s.output, s.consumed + s.empty + s.waiting);
};
return { player, reports, send, enable, samples, tick, snapshot, conserve };
}
test('idle worklet does not emit periodic diagnostics or count picker silence', () => {
const r = rig();
r.send({ type: 'diagnostics', enabled: true });
for (let i = 0; i < 1000; i++) assert.ok(r.tick().every((v) => v === 0));
assert.equal(r.reports.length, 0);
assert.equal(r.snapshot().output, 0);
r.conserve();
});
test('disabled, stale-session and zero-length messages cannot enqueue or reactivate', () => {
const r = rig({ prebufferFrames: 0 });
r.samples(new Float32Array(128).fill(1));
r.enable(2);
for (let i = 0; i < 10_000; i++) r.samples(new Float32Array(0), 2);
r.samples(new Float32Array(128).fill(1), 1);
assert.equal(r.player.queue.length, 0);
assert.equal(r.snapshot().active, false);
assert.ok(r.tick().every((v) => v === 0));
r.samples(new Float32Array(128).fill(1), 2);
r.send({ type: 'session', sessionId: 2, enabled: false });
r.samples(new Float32Array(128).fill(1), 2);
assert.equal(r.player.queue.length, 0);
assert.equal(r.snapshot().active, false);
r.conserve();
});
test('new session clears old audio and stale disable cannot disable current session', () => {
const r = rig({ prebufferFrames: 0 });
r.enable(1);
r.samples(new Float32Array(256).fill(1), 1);
r.enable(2);
r.samples(new Float32Array(256).fill(2), 2);
r.send({ type: 'session', sessionId: 1, enabled: false });
r.send({ type: 'volume', value: 1 });
assert.ok(r.tick().every((v) => v === 2));
assert.equal(r.snapshot().cleared, 256);
r.conserve();
});
test('prebuffer waits for threshold then consumes real samples in order', () => {
const r = rig({ prebufferFrames: 384, maxFrames: 1024 });
r.enable();
r.send({ type: 'volume', value: 1 });
r.samples(Float32Array.from({ length: 256 }, (_, i) => i + 1));
assert.ok(r.tick().every((v) => v === 0));
assert.equal(r.snapshot().waiting, 128);
r.samples(new Float32Array(256).fill(999));
assert.deepEqual(
[...r.tick()],
Array.from({ length: 128 }, (_, i) => i + 1),
);
assert.equal(r.snapshot().empty, 0);
r.conserve();
});
test('underrun zero-fills only missing frames and rebuffer waits before restarting', () => {
const r = rig({ prebufferFrames: 256 });
r.enable();
r.samples(new Float32Array(256).fill(1));
r.tick();
r.tick();
r.tick();
assert.equal(r.snapshot().empty, 128);
assert.equal(r.snapshot().underruns, 1);
r.samples(new Float32Array(128).fill(1));
assert.ok(r.tick().every((v) => v === 0));
r.samples(new Float32Array(128).fill(1));
assert.ok(r.tick().every((v) => v !== 0));
assert.equal(r.snapshot().waiting, 128);
r.conserve();
});
test('overflow trims exact frames across a partially consumed head', () => {
const r = rig({ prebufferFrames: 0, maxFrames: 1000 });
r.enable();
r.send({ type: 'volume', value: 1 });
r.samples(Float32Array.from({ length: 768 }, (_, i) => i + 1));
r.tick();
r.samples(new Float32Array(768).fill(999));
assert.equal(r.snapshot().dropped, 408);
assert.equal(r.tick(1)[0], 537);
assert.equal(r.snapshot().queued, 999);
r.conserve();
});
test('clear counts intentional discarded audio separately and stops idle reporting', () => {
const r = rig({ prebufferFrames: 0 });
r.enable();
r.send({ type: 'diagnostics', enabled: true });
r.samples(new Float32Array(768).fill(1));
r.tick();
r.send({ type: 'clear' });
const before = r.snapshot();
assert.equal(before.cleared, 640);
assert.equal(before.dropped, 0);
const reports = r.reports.length;
for (let i = 0; i < 750; i++) r.tick();
assert.equal(r.reports.length, reports);
assert.equal(r.snapshot().output, before.output);
assert.equal(r.snapshot().waiting, before.waiting);
r.conserve();
});
test('diagnostics are opt-in and paced by rendered frames rather than fixed quantum count', () => {
const r = rig({ prebufferFrames: 0 });
r.enable();
for (let i = 0; i < 200; i++) {
r.samples(new Float32Array(256).fill(1));
r.tick(256);
}
assert.equal(r.reports.length, 0);
r.send({ type: 'diagnostics', enabled: true });
for (let i = 0; i < 188; i++) {
r.samples(new Float32Array(256).fill(1));
r.tick(256);
}
assert.equal(r.reports.length, 1);
r.send({ type: 'diagnostics', enabled: false });
for (let i = 0; i < 200; i++) {
r.samples(new Float32Array(256).fill(1));
r.tick(256);
}
assert.equal(r.reports.length, 1);
r.conserve();
});
test('buffer options stay finite and bounded with prebuffer no greater than capacity', () => {
for (const options of [
{ maxFrames: Infinity, prebufferFrames: NaN },
{ maxFrames: -1, prebufferFrames: 1e9 },
{ maxFrames: 1e12, prebufferFrames: -Infinity },
{ maxFrames: '100', prebufferFrames: '200' },
]) {
const r = rig(options);
assert.ok(Number.isInteger(r.player.maxFrames));
assert.ok(r.player.maxFrames >= 1 && r.player.maxFrames <= 48_000);
assert.ok(
r.player.prebufferFrames >= 0 &&
r.player.prebufferFrames <= r.player.maxFrames,
);
}
});
test('full-rate jitter is absorbed but sustained undersupply remains visible', () => {
for (const rate of [48_000, 33_984]) {
const r = rig();
r.enable();
const arrivals = [];
for (let i = 0; i < Math.ceil((6 * rate) / 768); i++)
arrivals.push(
((i * 768) / rate) * 1000 + (rate === 48_000 && i % 8 >= 4 ? 40 : 0),
);
arrivals.sort((a, b) => a - b);
let index = 0;
for (let frame = 0; frame < 6 * 48_000; frame += 128) {
while (index < arrivals.length && arrivals[index] <= frame / 48) {
r.samples(new Float32Array(768).fill(1));
index++;
}
r.tick();
}
const s = r.snapshot();
if (rate === 48_000) {
assert.equal(s.empty, 0);
assert.equal(s.underruns, 0);
} else {
assert.ok(s.underruns > 0);
assert.ok(s.waiting + s.empty > 50_000);
}
r.conserve();
}
});
test('late enable cannot roll back a newer session or revive a retired epoch', () => {
const r = rig({ prebufferFrames: 0 });
r.enable(2);
r.enable(1);
assert.equal(r.snapshot().sessionId, 2);
r.send({ type: 'session', sessionId: 2, enabled: false });
r.enable(2);
r.samples(new Float32Array(128).fill(1), 2);
assert.equal(r.snapshot().enqueued, 0);
r.enable(3);
r.samples(new Float32Array(128).fill(1), 3);
assert.ok(r.tick().every((v) => v !== 0));
r.conserve();
});
src/sdr/spectrum.js
+2
View File
@@ -91,6 +91,7 @@ export function findDirectionalFmPeak(
minFrequency = 87_500_000,
maxFrequency = 108_000_000,
thresholdDb = 6,
skipFrequency = NaN,
} = {},
) {
if (
@@ -124,6 +125,7 @@ export function findDirectionalFmPeak(
) {
if (direction > 0 && frequency <= afterFrequency + 75_000) continue;
if (direction < 0 && frequency >= afterFrequency - 75_000) continue;
if (Math.abs(frequency - skipFrequency) <= 75_000) continue;
const index = Math.round(
(frequency - centerFrequency) / binWidth + bins.length / 2,
);
src/sdr/spectrum.test.mjs
+20
View File
@@ -41,3 +41,23 @@ test('directional FM seek chooses the nearest rounded station in-band', () => {
null,
);
});
test('directional FM seek passes over the skipped channel to the next station', () => {
const sampleRate = 2_048_000;
const bins = new Float32Array(2_048).fill(-80);
// Carriers at 98.8 and 99.1 MHz in a frame centred on 98.5 MHz. A carrier
// 100 kHz away would sit inside the peak test's 150 kHz comparison window.
for (const offset of [300_000, 600_000]) {
const index = offset / (sampleRate / bins.length) + bins.length / 2;
for (let bin = index - 5; bin <= index + 5; bin += 1) bins[bin] = -10;
}
const seek = (skipFrequency) =>
findDirectionalFmPeak(bins, 98_500_000, sampleRate, 1, 98_500_000, {
skipFrequency,
})?.frequency;
assert.equal(seek(undefined), 98_800_000);
assert.equal(seek(98_800_000), 99_100_000);
// The skip covers channels up to and including 75 kHz away.
assert.equal(seek(98_725_000), 99_100_000);
assert.equal(seek(98_724_999), 98_800_000);
});
src/sdr/workerAdmission.test.mjs
+420
View File
@@ -0,0 +1,420 @@
import assert from 'node:assert/strict';
import test from 'node:test';
import { Worker as ThreadWorker } from 'node:worker_threads';
import { SdrController } from './controller.js';
const ADMISSION = {
maxBlocks: 4,
maxBytes: 4 * 64,
maxAgeMs: 60,
watchdogMs: 10,
reportMs: 20,
readyTimeoutMs: 150,
};
const pause = (ms = 0) => new Promise((resolve) => setTimeout(resolve, ms));
/** Wait (bounded) until `check` holds. */
async function until(check, ms = 1_000) {
const deadline = Date.now() + ms;
while (!check()) {
if (Date.now() > deadline) throw new Error('condition not reached');
await pause(2);
}
}
/** A decoder stand-in that records blocks and acknowledges as told. */
function installWorker(t, { ack = 'auto', readyAfterMs = 0 } = {}) {
const workers = [];
class AckWorker {
constructor() {
this.blocks = [];
this.messages = [];
this.terminated = false;
workers.push(this);
// Like the real worker, report ready once the module has loaded.
if (readyAfterMs !== null)
setTimeout(() => this.emit({ type: 'ready' }), readyAfterMs);
}
postMessage(message) {
this.messages.push(message);
if (message?.type !== 'samples') return;
this.blocks.push(message.id);
if (ack === 'auto') queueMicrotask(() => this.ack(message.id));
}
ack(id) {
this.onmessage?.({ data: { type: 'ack', id } });
}
emit(data) {
this.onmessage?.({ data });
}
terminate() {
this.terminated = true;
}
}
const prior = globalThis.Worker;
globalThis.Worker = AckWorker;
t.after(() => {
globalThis.Worker = prior;
});
return workers;
}
/** A streaming controller with a device that yields one block per read. */
function streaming(t, { reads = Infinity, admission = ADMISSION } = {}) {
const controller = new SdrController({
storage: { getItem: () => null, setItem() {} },
webUsb: {},
workerAdmission: admission,
});
const audio = [];
controller._audioNode = {
port: { postMessage: (message) => audio.push(message) },
};
let served = 0;
const device = {
readSamples: (count) =>
served++ < reads
? new Promise((resolve) =>
setTimeout(() => resolve({ data: new ArrayBuffer(64) }), 1),
)
: new Promise(() => {}),
close: async () => {},
};
controller._device = device;
controller._setState({ mode: 'fm', status: 'streaming', connected: true });
controller._beginAudioSession();
controller._readGeneration += 1;
controller._startWatchdog();
void controller._readLoop(controller._readGeneration);
t.after(() => {
controller._stopWatchdog();
controller._readGeneration += 1;
controller._device = null;
});
return { controller, audio, device };
}
test('a slow decoder startup is not an overload: no block is sent before ready', async (t) => {
// Ready after 200 ms: longer than the 4 blocks the cap admits (and than the
// 128 ms the production cap of 8 blocks covers).
const workers = installWorker(t, { readyAfterMs: 200 });
const admission = { ...ADMISSION, readyTimeoutMs: 1_000 };
const { controller } = streaming(t, { reads: 10, admission });
await pause(150);
assert.equal(workers[0].blocks.length, 0, 'no block before ready');
assert.equal(controller.getState().status, 'streaming');
await until(() => workers[0].blocks.length >= 10);
assert.equal(controller.getState().status, 'streaming');
assert.equal(controller.getState().decoderOverload, undefined);
});
test('a decoder that never becomes ready stops the receiver at the startup bound', async (t) => {
const workers = installWorker(t, { readyAfterMs: null });
const { controller } = streaming(t, { reads: 10 });
await until(() => controller.getState().status === 'error', 1_000);
assert.equal(controller.getState().decoderOverload, 'startup');
assert.equal(workers[0].blocks.length, 0);
assert.equal(workers[0].terminated, true);
});
test('an acknowledging decoder streams with pending work bounded', async (t) => {
const workers = installWorker(t);
const { controller } = streaming(t, { reads: 40 });
await until(() => workers[0]?.blocks.length >= 40);
await pause(5);
assert.equal(controller.getState().status, 'streaming');
assert.equal(controller._pendingBlocks.size, 0);
assert.equal(workers[0].terminated, false);
});
test('a decoder that falls behind stops the receiver at the block bound', async (t) => {
const workers = installWorker(t, { ack: 'manual' });
const { controller } = streaming(t);
await until(() => controller.getState().status === 'error');
const worker = workers[0];
assert.equal(worker.blocks.length, ADMISSION.maxBlocks, 'never past the cap');
assert.equal(worker.terminated, true, 'decoder terminated');
assert.equal(controller.getState().decoderOverload, 'backlog');
assert.equal(controller._pendingBlocks.size, 0);
const posted = worker.blocks.length;
await pause(20);
assert.equal(worker.blocks.length, posted, 'nothing posted after overload');
});
test('a hung decoder is stopped by age even when no further USB block arrives', async (t) => {
const workers = installWorker(t, { ack: 'manual' });
const { controller } = streaming(t, { reads: 2 });
await until(() => workers[0]?.blocks.length === 2);
await until(() => controller.getState().status === 'error', 1_000);
assert.equal(controller.getState().decoderOverload, 'stalled');
assert.equal(workers[0].terminated, true);
});
test('pending diagnostics are published at low frequency, not per ack', async (t) => {
const workers = installWorker(t);
const { controller } = streaming(t, { reads: 30 });
let reports = 0;
controller.subscribe((state) => {
if ('decoderPendingBlocks' in state) reports += 1;
});
await until(() => workers[0]?.blocks.length >= 30);
const state = controller.getState();
assert.ok(Number.isFinite(state.decoderPendingBlocks));
assert.ok(Number.isFinite(state.decoderPendingBytes));
assert.ok(Number.isFinite(state.decoderOldestPendingMs));
assert.ok(reports < workers[0].blocks.length, 'fewer reports than acks');
});
test('duplicate, unknown and stale acknowledgements change nothing', async (t) => {
const workers = installWorker(t, { ack: 'manual' });
const { controller } = streaming(t, { reads: 2 });
await until(() => workers[0]?.blocks.length === 2);
const worker = workers[0];
worker.ack(0);
worker.ack(0); // duplicate
worker.ack(99); // never sent
assert.equal(controller._pendingBlocks.size, 1);
assert.equal(controller._pendingBytes, 64);
assert.equal(controller._ackAnomalies, 2);
// A replaced worker's late acknowledgement cannot touch the new worker.
const oldHandler = worker.onmessage;
controller._discardWorker();
controller._ensureWorker();
const fresh = workers[1];
controller._postSamples(new ArrayBuffer(64));
assert.equal(fresh.blocks[0], 0, 'block numbers restart per worker');
oldHandler?.({ data: { type: 'ack', id: 0 } });
assert.equal(controller._pendingBlocks.size, 1, 'stale ack ignored');
fresh.ack(0);
assert.equal(controller._pendingBlocks.size, 0);
});
test('stop terminates the decoder before cleanup and fences its late output', async (t) => {
const workers = installWorker(t);
const { controller, audio, device } = streaming(t, { reads: 5 });
await until(() => workers[0]?.blocks.length >= 3);
const worker = workers[0];
const handler = worker.onmessage;
let closed = false;
device.close = () =>
new Promise((resolve) =>
setTimeout(() => {
closed = true;
resolve();
}, 30),
);
const stopping = controller.stop();
assert.equal(worker.terminated, true, 'terminated synchronously');
assert.equal(closed, false, 'before the device close settled');
const disabled = audio.filter(
(message) => message.type === 'session' && message.enabled === false,
);
assert.equal(disabled.length, 1, 'player session disabled');
const before = audio.filter((message) => message.type === 'samples').length;
handler?.({
data: {
type: 'fm',
samples: new Float32Array(8),
tune: controller._fmTune,
},
});
await stopping;
assert.equal(
audio.filter((message) => message.type === 'samples').length,
before,
'no late audio after stop',
);
});
test('audio reaches the player only for the live FM session and current tune', async (t) => {
const workers = installWorker(t, { ack: 'manual' });
const { controller, audio } = streaming(t, { reads: 1 });
await until(() => workers[0]?.blocks.length === 1);
const worker = workers[0];
const samples = () => audio.filter((message) => message.type === 'samples');
worker.emit({ type: 'fm', samples: new Float32Array(4), tune: 0 });
assert.equal(samples().length, 1);
assert.equal(samples()[0].sessionId, controller._audioSessionId);
controller._fmTune = 3; // a retune is in flight
worker.emit({ type: 'fm', samples: new Float32Array(4), tune: 0 });
assert.equal(samples().length, 1, 'pre-retune audio dropped');
worker.emit({ type: 'fm', samples: new Float32Array(4), tune: 3 });
assert.equal(samples().length, 2);
controller._setState({ mode: 'adsb' });
worker.emit({ type: 'fm', samples: new Float32Array(4), tune: 3 });
assert.equal(samples().length, 2, 'no FM audio outside FM mode');
controller._setState({ mode: 'fm' });
worker.emit({ type: 'fm', samples: new Float32Array(0), tune: 3 });
assert.equal(samples().length, 2, 'empty audio is not forwarded');
});
test('an FM to ADS-B to FM round trip cannot relabel old output as current', async (t) => {
const workers = installWorker(t, { ack: 'manual' });
const { controller, audio } = streaming(t, { reads: 1 });
await until(() => workers[0]?.blocks.length === 1);
const worker = workers[0];
const samples = () => audio.filter((message) => message.type === 'samples');
controller._configureWorker();
const fmEpoch = controller._fmTune;
// Leave for ADS-B and come back, on the same worker, as _configureDevice does.
controller._setState({ mode: 'adsb' });
controller._configureWorker();
controller._setState({ mode: 'fm' });
controller._configureWorker();
controller._beginAudioSession();
const current = controller._fmTune;
assert.ok(current > fmEpoch);
worker.emit({ type: 'fm', samples: new Float32Array(4), tune: fmEpoch });
assert.equal(
samples().length,
0,
'old FM output dropped after the round trip',
);
worker.emit({ type: 'fm', samples: new Float32Array(4), tune: current });
assert.equal(samples().length, 1);
});
test('old-epoch diagnostics, SNR and ADS-B output are ignored', async (t) => {
const workers = installWorker(t, { ack: 'manual' });
const { controller } = streaming(t, { reads: 1 });
await until(() => workers[0]?.blocks.length === 1);
const worker = workers[0];
controller._configureWorker();
const old = controller._fmTune - 1;
const before = controller.getState();
worker.emit({
type: 'fm',
tune: old,
snr: 42,
samples: new Float32Array(4),
diagnostics: { workerBlocks: 999, iqLevelDbfs: -1, audioLevelDbfs: -1 },
});
worker.emit({ type: 'adsb', tune: old, aircraft: [{ icao: 'abc123' }] });
const after = controller.getState();
assert.equal(after.snr, before.snr);
assert.equal(after.workerBlocks, before.workerBlocks);
assert.deepEqual(after.aircraft, before.aircraft, 'old ADS-B output ignored');
// Control: the same output in the current epoch is applied, so the checks
// above are not vacuous.
worker.emit({
type: 'adsb',
tune: controller._fmTune,
aircraft: [{ icao: 'abc123' }],
});
assert.deepEqual(
controller.getState().aircraft.map((entry) => entry.icao),
['abc123'],
);
});
test('an overload publishes its error first and never overwrites a reconnect', async (t) => {
const workers = installWorker(t, { ack: 'manual' });
const { controller, device } = streaming(t);
device.close = () => new Promise((resolve) => setTimeout(resolve, 40));
await until(() => controller.getState().status === 'error');
assert.equal(controller.getState().connected, false);
assert.equal(workers[0].terminated, true);
// A reconnect starts while the overloaded session is still tearing down.
controller._sessionToken += 1;
controller._setState({ status: 'streaming', connected: true });
await pause(80);
assert.equal(controller.getState().status, 'streaming');
assert.equal(controller.getState().connected, true);
});
test('a reconnect gets a new decoder with fresh numbering and no carried backlog', async (t) => {
const workers = installWorker(t, { ack: 'manual' });
const first = streaming(t, { reads: 3 });
await until(() => workers[0]?.blocks.length === 3);
await first.controller.stop();
assert.equal(first.controller._pendingBlocks.size, 0);
first.controller._device = {
readSamples: () =>
new Promise((resolve) =>
setTimeout(() => resolve({ data: new ArrayBuffer(64) }), 1),
),
close: async () => {},
};
first.controller._readGeneration += 1;
first.controller._startWatchdog();
void first.controller._readLoop(first.controller._readGeneration);
await until(() => workers[1]?.blocks.length >= 1);
assert.equal(workers[1].blocks[0], 0);
assert.equal(
workers[0].blocks.length,
3,
'old decoder receives nothing more',
);
});
/** The real sdr.worker.js in a Node worker thread, with a browser-like `self`. */
function realWorker(t) {
const module = new URL('./sdr.worker.js', import.meta.url).href;
const thread = new ThreadWorker(
`const { parentPort } = require('node:worker_threads');
globalThis.self = globalThis;
self.postMessage = (m, transfer) => parentPort.postMessage(m, transfer);
parentPort.on('message', (data) => self.onmessage && self.onmessage({ data }));
import(${JSON.stringify(module)}).then(() => parentPort.postMessage({ type: 'ready' }));`,
{ eval: true },
);
const inbox = [];
thread.on('message', (message) => inbox.push(message));
t.after(() => thread.terminate());
return { thread, inbox };
}
function iqBlock(samples = 32_768) {
const bytes = new Uint8Array(samples * 2);
for (let n = 0; n < samples; n += 1) {
const phase = (2 * Math.PI * 100_000 * n) / 2_048_000;
bytes[2 * n] = Math.round(127.5 + 60 * Math.cos(phase));
bytes[2 * n + 1] = Math.round(127.5 + 60 * Math.sin(phase));
}
return bytes.buffer;
}
test('the real decoder acknowledges every numbered block after its output, in every path', async (t) => {
const { thread, inbox } = realWorker(t);
await until(() => inbox.some((m) => m.type === 'ready'), 5_000);
thread.postMessage({
type: 'configure',
mode: 'fm',
sampleRate: 2_048_000,
tune: 5,
});
for (const id of [0, 1, 2]) {
const buffer = iqBlock();
thread.postMessage({ type: 'samples', buffer, id }, [buffer]);
}
await until(() => inbox.filter((m) => m.type === 'ack').length === 3, 5_000);
const fm = inbox.filter((m) => m.type === 'fm');
assert.equal(fm.length, 3);
assert.ok(
fm.every((m) => m.tune === 5),
'configure re-synced the tune tag',
);
const order = inbox.filter((m) => m.type === 'fm' || m.type === 'ack');
assert.deepEqual(
order.map((m) => m.type),
['fm', 'ack', 'fm', 'ack', 'fm', 'ack'],
'each ack follows that block output',
);
// An invalid block produces no output and is still acknowledged.
thread.postMessage({ type: 'samples', buffer: null, id: 3 });
// ADS-B: output is rare, acknowledgement is not.
thread.postMessage({
type: 'configure',
mode: 'adsb',
sampleRate: 2_000_000,
});
const adsb = iqBlock(16_384);
thread.postMessage({ type: 'samples', buffer: adsb, id: 4 }, [adsb]);
await until(() => inbox.filter((m) => m.type === 'ack').length === 5, 5_000);
assert.deepEqual(
inbox.filter((m) => m.type === 'ack').map((m) => m.id),
[0, 1, 2, 3, 4],
);
});