diff --git a/CHANGELOG.md b/CHANGELOG.md index a3e622b52..d70ba76e8 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,5 +1,14 @@ # Changelog +- Local FM seek finds stations after wrapping around 87.5–108 MHz, covers both + band edges, skips the starting station and rejects spectra from an earlier + tune (Aditya, @Rohilalala, #1174). +- Local RTL-SDR keeps USB reads queued in order and bounds pending decoder + work. A stalled or overloaded decoder stops with an explicit error; stopping + terminates its worker and rejects late audio. FM playback buffers short + scheduling gaps, caps its audio queue at 256 ms, and reduces display updates + and spectrum calculations while listening. + - Add Bogotá (Tembici, 250 stations) and Santiago de Chile (Bike Itaú, 200 stations) to the GBFS bikeshare registry, the layer's first stations in Colombia and Chile. Both are PBSC feeds like the Brazilian Bike Itaú cities, diff --git a/docs/CURRENT-STATE.md b/docs/CURRENT-STATE.md index 4c06dd3cb..62c279847 100644 --- a/docs/CURRENT-STATE.md +++ b/docs/CURRENT-STATE.md @@ -5396,6 +5396,22 @@ Local FM and internet radio are one listening surface: starting local FM stops internet-radio playback, and starting internet radio stops local FM. ADS-B reception does not interrupt internet radio. +- **FM seek.** Seeking wraps across 87.5–108 MHz, covers both band edges and + skips the starting station. Spectra carry a decoder epoch, so a retune or mode + change rejects earlier output. A failed scan retune stops seeking. This + integrates Aditya's (@Rohilalala) contribution in #1174; it does not change + the peak-selection algorithm or guarantee that a final station retune succeeds. +- **Streaming and playback.** Four ordered USB reads keep input available while + the page handles other work; control changes drain those reads before touching + the receiver. The worker must finish loading before reads start. Decoder work + is limited to eight unacknowledged blocks / 512 KiB, with a one-second oldest + block watchdog. Overload stops the receiver with an explicit error. Stop + terminates the worker immediately and disables the audio session; stale decoder + output and audio from retired sessions are refused. FM audio starts after an + 80 ms prebuffer and retains at most 256 ms, dropping the oldest frames on + overflow. FM signal display updates are coalesced to at most four per second; + spectra are computed only while seeking. Optional `decoderDiagnostics` adds + bounded timing samples and overload snapshots without recording audio or IQ. - **Modes.** FM (87.5–108 MHz, tune, seek, volume) or ADS-B at 1090 MHz. With a receiver open, selecting ADS-B enables the Local ADS-B layer; selecting FM disables it. diff --git a/scripts/format-scope.json b/scripts/format-scope.json index c62c6886f..c25797f38 100644 --- a/scripts/format-scope.json +++ b/scripts/format-scope.json @@ -872,9 +872,13 @@ "src/layers/weather/imageryHost.test.mjs", "src/sdr/adsbDecoder.test.mjs", "src/sdr/controller.test.mjs", + "src/sdr/decoderDiagnostics.test.mjs", "src/sdr/gain.test.mjs", + "src/sdr/readPipeline.diag.test.mjs", + "src/sdr/sdrAudioWorklet.test.mjs", "src/sdr/spectrum.test.mjs", "src/sdr/usbDevices.test.mjs", + "src/sdr/workerAdmission.test.mjs", "src/sources/adsbRecords.test.mjs", "src/layers/localAdsb/index.test.mjs", "src/ui/localSdrControls.test.mjs", diff --git a/scripts/qa-radio.mjs b/scripts/qa-radio.mjs index adbbfdf81..648075bd2 100644 --- a/scripts/qa-radio.mjs +++ b/scripts/qa-radio.mjs @@ -167,6 +167,13 @@ async function main() { page.on('request', (request) => { const url = new URL(request.url()); + // Radio QA uses keyless maps; an optional Google token is not offered. + if (request.method() === 'GET' && url.origin === APP_ORIGIN + && url.pathname === '/api/google/tiles-token') { + request.respond({ status: 200, contentType: 'application/json', body: '{}' }); + return; + } + // These scenarios exercise Context lifecycle and keyboard ownership, not // live orbit accuracy. Reuse the tracking suite's fixed element sets so // CelesTrak outages cannot invalidate an otherwise clean UI run. diff --git a/src/sdr/controller.js b/src/sdr/controller.js index 1c86658ec..bcd1d1750 100644 --- a/src/sdr/controller.js +++ b/src/sdr/controller.js @@ -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; diff --git a/src/sdr/controller.test.mjs b/src/sdr/controller.test.mjs index f8078dbe5..6abc1b6f4 100644 --- a/src/sdr/controller.test.mjs +++ b/src/sdr/controller.test.mjs @@ -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); +}); diff --git a/src/sdr/decoderDiagnostics.test.mjs b/src/sdr/decoderDiagnostics.test.mjs new file mode 100644 index 000000000..39b7e4d0b --- /dev/null +++ b/src/sdr/decoderDiagnostics.test.mjs @@ -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), + ); +}); diff --git a/src/sdr/readPipeline.diag.test.mjs b/src/sdr/readPipeline.diag.test.mjs new file mode 100644 index 000000000..0a2253d17 --- /dev/null +++ b/src/sdr/readPipeline.diag.test.mjs @@ -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'); +}); diff --git a/src/sdr/sdr.worker.js b/src/sdr/sdr.worker.js index 07f3c5bc6..ef1cb22b7 100644 --- a/src/sdr/sdr.worker.js +++ b/src/sdr/sdr.worker.js @@ -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' }); diff --git a/src/sdr/sdrAudioWorklet.js b/src/sdr/sdrAudioWorklet.js index 4983e43bf..6ea87d98b 100644 --- a/src/sdr/sdrAudioWorklet.js +++ b/src/sdr/sdrAudioWorklet.js @@ -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; diff --git a/src/sdr/sdrAudioWorklet.test.mjs b/src/sdr/sdrAudioWorklet.test.mjs new file mode 100644 index 000000000..ddb328b76 --- /dev/null +++ b/src/sdr/sdrAudioWorklet.test.mjs @@ -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(); +}); diff --git a/src/sdr/spectrum.js b/src/sdr/spectrum.js index acec025c4..4b4f22de8 100644 --- a/src/sdr/spectrum.js +++ b/src/sdr/spectrum.js @@ -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, ); diff --git a/src/sdr/spectrum.test.mjs b/src/sdr/spectrum.test.mjs index 551b3f824..65104e6b2 100644 --- a/src/sdr/spectrum.test.mjs +++ b/src/sdr/spectrum.test.mjs @@ -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); +}); diff --git a/src/sdr/workerAdmission.test.mjs b/src/sdr/workerAdmission.test.mjs new file mode 100644 index 000000000..f7392b66a --- /dev/null +++ b/src/sdr/workerAdmission.test.mjs @@ -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], + ); +});