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], + ); +});