diff --git a/native/src/linux_capture.cpp b/native/src/linux_capture.cpp index d770478..ea0f7b1 100644 --- a/native/src/linux_capture.cpp +++ b/native/src/linux_capture.cpp @@ -17,6 +17,7 @@ namespace { constexpr size_t kMaxQueuedChunks = 256; constexpr size_t kDefaultDrainChunkLimit = 64; +constexpr pa_usec_t kTargetRecordFragmentMicroseconds = 10000; struct OutputDeviceInfo { std::string id; @@ -574,6 +575,21 @@ private: return std::string(prefix) + " " + pulseError; } + static pa_buffer_attr buildRecordBufferAttr(const pa_sample_spec& sampleSpec) { + pa_buffer_attr attr{}; + attr.maxlength = UINT32_MAX; + attr.tlength = UINT32_MAX; + attr.prebuf = UINT32_MAX; + attr.minreq = UINT32_MAX; + + const size_t requestedFragSize = + pa_usec_to_bytes(kTargetRecordFragmentMicroseconds, &sampleSpec); + attr.fragsize = requestedFragSize == 0 + ? 1 + : static_cast(std::min(requestedFragSize, UINT32_MAX)); + return attr; + } + bool startInternal(const std::string& requestedDeviceId, std::string* outErrorMessage) { stopInternal(); @@ -645,13 +661,14 @@ private: pa_stream_set_state_callback(stream_, &HandleStreamState, connection_.mainloop()); pa_stream_set_read_callback(stream_, &HandleStreamRead, this); + const pa_buffer_attr requestedBufferAttr = buildRecordBufferAttr(selected->sampleSpec); const pa_stream_flags_t flags = static_cast( PA_STREAM_ADJUST_LATENCY | PA_STREAM_AUTO_TIMING_UPDATE | PA_STREAM_INTERPOLATE_TIMING | PA_STREAM_DONT_MOVE); const int connectResult = pa_stream_connect_record( - stream_, selected->monitorSourceName.c_str(), nullptr, flags); + stream_, selected->monitorSourceName.c_str(), &requestedBufferAttr, flags); if (connectResult < 0) { const std::string errorMessage = buildContextErrorMessage( connection_.context(), "Could not start Linux monitor capture."); @@ -686,6 +703,12 @@ private: } else { sampleSpec_ = selected->sampleSpec; } + const pa_buffer_attr* activeBufferAttr = pa_stream_get_buffer_attr(stream_); + if (activeBufferAttr != nullptr) { + bufferAttr_ = *activeBufferAttr; + } else { + bufferAttr_ = requestedBufferAttr; + } { std::lock_guard lock(stateMutex_); @@ -751,6 +774,7 @@ private: channelCount_ = 2; sequence_ = 0; sampleSpec_ = pa_sample_spec{}; + bufferAttr_ = pa_buffer_attr{}; } { @@ -856,6 +880,7 @@ private: uint32_t channelCount_ = 2; uint64_t sequence_ = 0; pa_sample_spec sampleSpec_{}; + pa_buffer_attr bufferAttr_{}; std::deque chunkQueue_; size_t overwriteCount_ = 0; }; diff --git a/src/renderer/audio/AudioCapture.ts b/src/renderer/audio/AudioCapture.ts index 6148564..cb0f4d7 100644 --- a/src/renderer/audio/AudioCapture.ts +++ b/src/renderer/audio/AudioCapture.ts @@ -53,15 +53,58 @@ interface CaptureBackend { getStatus(): CaptureBackendStatus } +const NATIVE_BACKLOG_CATCH_UP_CHUNK_THRESHOLD = 4 +const NATIVE_BACKLOG_LIVE_WINDOW_MS = 50 + function isDocumentHidden(): boolean { return typeof document !== 'undefined' && document.hidden === true } -function resolveNativeCapturePollDelay(chunkCount: number): number { +function resolveNativeCapturePollDelay(chunkCount: number, queueDepth = 0): number { if (isDocumentHidden()) { return 16 } - return chunkCount > 0 ? 0 : 2 + return chunkCount > 0 || queueDepth > 0 ? 0 : 2 +} + +function trimNativeChunksToLiveWindow(chunks: NativeCaptureDrainResult['chunks']): NativeCaptureDrainResult['chunks'] { + if (chunks.length <= 1) { + return chunks + } + + const latestChunk = chunks[chunks.length - 1] + const latestCapturedAt = latestChunk?.capturedAtMilliseconds ?? NaN + if (!Number.isFinite(latestCapturedAt)) { + return latestChunk ? [latestChunk] : [] + } + + const firstLiveChunkIndex = chunks.findIndex((chunk) => ( + Number.isFinite(chunk.capturedAtMilliseconds) + ? latestCapturedAt - chunk.capturedAtMilliseconds <= NATIVE_BACKLOG_LIVE_WINDOW_MS + : false + )) + + if (firstLiveChunkIndex === -1) { + return latestChunk ? [latestChunk] : [] + } + + return chunks.slice(firstLiveChunkIndex) +} + +function selectNativeChunksForDelivery( + result: NativeCaptureDrainResult, + trimBacklog: boolean, +): NativeCaptureDrainResult['chunks'] { + if (!trimBacklog) { + return result.chunks + } + + const shouldCatchUp = result.queueDepth > 0 || result.chunks.length > NATIVE_BACKLOG_CATCH_UP_CHUNK_THRESHOLD + if (!shouldCatchUp) { + return result.chunks + } + + return trimNativeChunksToLiveWindow(result.chunks) } function resolvePlatform(): string { @@ -400,6 +443,10 @@ export abstract class NativePolledCaptureBackend implements CaptureBackend { } } + protected shouldTrimBacklogForLiveCapture(): boolean { + return false + } + private startPolling(): void { this.stopPolling() @@ -407,11 +454,17 @@ export abstract class NativePolledCaptureBackend implements CaptureBackend { if (!this.active) return let chunkCount = 0 + let queueDepth = 0 try { const result = this.getNativeCaptureModule()?.drain(32) as NativeCaptureDrainResult | undefined chunkCount = result?.chunks.length ?? 0 + queueDepth = result?.queueDepth ?? 0 if (result) { - for (const chunk of result.chunks) { + const deliveredChunks = selectNativeChunksForDelivery( + result, + this.shouldTrimBacklogForLiveCapture(), + ) + for (const chunk of deliveredChunks) { const routedChunk: CaptureChunk = { left: chunk.left, right: chunk.right, @@ -432,7 +485,7 @@ export abstract class NativePolledCaptureBackend implements CaptureBackend { return } - this.pollTimer = window.setTimeout(poll, resolveNativeCapturePollDelay(chunkCount)) + this.pollTimer = window.setTimeout(poll, resolveNativeCapturePollDelay(chunkCount, queueDepth)) } this.pollTimer = window.setTimeout(poll, 0) @@ -483,6 +536,10 @@ class NativeLinuxCaptureBackend extends NativePolledCaptureBackend { protected getBackendLabel(): string { return 'Native Linux' } + + protected shouldTrimBacklogForLiveCapture(): boolean { + return true + } } class NativeUnavailableCaptureBackend implements CaptureBackend { diff --git a/test/renderer-helpers.test.ts b/test/renderer-helpers.test.ts index 199297c..b569cbb 100644 --- a/test/renderer-helpers.test.ts +++ b/test/renderer-helpers.test.ts @@ -3158,7 +3158,7 @@ test('LUFSMeter keeps integrated history bounded over long runs', () => { assert.equal(Number.isFinite((meter as unknown as { integratedLUFS: number }).integratedLUFS), true) }) -test('NativePolledCaptureBackend schedules immediate, backoff, and hidden-document polls and cancels on stop', async () => { +test('NativePolledCaptureBackend forwards all drained chunks, respects hidden-document backoff, and cancels on stop', async () => { const timers = installFakeTimeouts() try { @@ -3166,13 +3166,22 @@ test('NativePolledCaptureBackend schedules immediate, backoff, and hidden-docume const drainResults = [ { - chunks: [{ - left: new Float32Array([0.1, 0.2]), - right: new Float32Array([0.3, 0.4]), - channelCount: 2, - capturedAtMilliseconds: 5, - sequence: 1, - }], + chunks: [ + { + left: new Float32Array([0.1, 0.2]), + right: new Float32Array([0.3, 0.4]), + channelCount: 2, + capturedAtMilliseconds: 5, + sequence: 1, + }, + { + left: new Float32Array([0.5, 0.6]), + right: new Float32Array([0.7, 0.8]), + channelCount: 2, + capturedAtMilliseconds: 15, + sequence: 2, + }, + ], overwriteCount: 0, queueDepth: 0, }, @@ -3224,15 +3233,18 @@ test('NativePolledCaptureBackend schedules immediate, backoff, and hidden-docume reason: null, }) const receivedSequences: number[] = [] + const receivedChunkTimes: number[] = [] backend.subscribe((chunk) => { receivedSequences.push(chunk.sequence) + receivedChunkTimes.push(chunk.capturedAt) }) await backend.start() assert.equal(timers.nextDelay(), 0) timers.runNext() - assert.deepEqual(receivedSequences, [1]) + assert.deepEqual(receivedSequences, [1, 2]) + assert.equal(receivedChunkTimes.length, 2) assert.equal(timers.nextDelay(), 0) timers.runNext() @@ -3248,3 +3260,134 @@ test('NativePolledCaptureBackend schedules immediate, backoff, and hidden-docume timers.restore() } }) + +test('NativePolledCaptureBackend trims stale backlog to the newest live slice when catch-up mode is enabled', async () => { + const timers = installFakeTimeouts() + + try { + const { NativePolledCaptureBackend } = await import('../src/renderer/audio/AudioCapture') + + const drainResults = [ + { + chunks: [ + { + left: new Float32Array([0.1]), + right: new Float32Array([0.1]), + channelCount: 2, + capturedAtMilliseconds: 0, + sequence: 1, + }, + { + left: new Float32Array([0.2]), + right: new Float32Array([0.2]), + channelCount: 2, + capturedAtMilliseconds: 10, + sequence: 2, + }, + { + left: new Float32Array([0.3]), + right: new Float32Array([0.3]), + channelCount: 2, + capturedAtMilliseconds: 30, + sequence: 3, + }, + { + left: new Float32Array([0.4]), + right: new Float32Array([0.4]), + channelCount: 2, + capturedAtMilliseconds: 70, + sequence: 4, + }, + { + left: new Float32Array([0.5]), + right: new Float32Array([0.5]), + channelCount: 2, + capturedAtMilliseconds: 90, + sequence: 5, + }, + ], + overwriteCount: 0, + queueDepth: 3, + }, + { + chunks: [{ + left: new Float32Array([0.6]), + right: new Float32Array([0.6]), + channelCount: 2, + capturedAtMilliseconds: 120, + sequence: 6, + }], + overwriteCount: 0, + queueDepth: 0, + }, + { + chunks: [], + overwriteCount: 0, + queueDepth: 0, + }, + ] + + const nativeModule = { + getSupport: () => ({ available: true, reason: null }), + listOutputDevices: () => [], + start: () => ({ + sampleRate: 48000, + channelCount: 2, + deviceId: 'device', + deviceLabel: 'Device', + }), + stop: () => {}, + drain: () => drainResults.shift() ?? { + chunks: [], + overwriteCount: 0, + queueDepth: 0, + }, + nowMilliseconds: () => 0, + } + + class TestNativeBackend extends NativePolledCaptureBackend { + readonly kind = 'native-linux' as const + + protected getNativeCaptureModule() { + return nativeModule + } + + protected getBackendLabel(): string { + return 'Test Native' + } + + protected shouldTrimBacklogForLiveCapture(): boolean { + return true + } + } + + const backend = new TestNativeBackend({ + kind: 'native-linux', + available: true, + reason: null, + }) + const receivedSequences: number[] = [] + backend.subscribe((chunk) => { + receivedSequences.push(chunk.sequence) + }) + + await backend.start() + assert.equal(timers.nextDelay(), 0) + + timers.runNext() + assert.deepEqual(receivedSequences, [4, 5]) + assert.equal(timers.nextDelay(), 0) + + timers.runNext() + assert.deepEqual(receivedSequences, [4, 5, 6]) + assert.equal(timers.nextDelay(), 0) + + timers.runNext() + assert.equal(timers.nextDelay(), 2) + + await backend.stop() + assert.equal(timers.pendingCount(), 0) + } finally { + timers.restore() + } +})