fix: Linux PipeWire engine two-level Frame enum + bounded frame channel (all platforms)

[Tristan-Stoltz-ERC]

Summary

The Frame enum was refactored into a two-level shape (Frame::Audio(AudioFrame) / Frame::Video(VideoFrame::*)) and the per-format structs' display_time field was changed from u64 to SystemTime, but the Linux PipeWire engine in src/capturer/engine/linux/mod.rs was not updated to match.

Result: 8 compile errors on Linux; 0.1.0-beta.1 on crates.io does not build there, and neither does the current main (verified on Ubuntu 24 with pipewire 1.0). fix-windows addressed the Windows path but left Linux untouched.

While fixing that, I found and fixed a real memory-growth bug in the shared frame channel — and then, in fixing that, broke the Windows/macOS build the same way the original Linux bug happened, so I fixed that too. Details below, split by commit since they're three separate concerns.

Commit 1 — fix(linux): use two-level Frame::Video(VideoFrame::...) enum + SystemTime display_time

• Wraps the four emit sites (RGBx / RGB / xBGR / BGRx) in Frame::Video(VideoFrame::...(...)) to match frame::Frame's current two-level shape.
• Replaces display_time: timestamp as u64 with display_time: SystemTime::now(), matching what the macOS (src/capturer/engine/mac/mod.rs) and Windows (src/capturer/engine/win/mod.rs) engines already do. The raw PipeWire pts from spa_meta_header is monotonic ns since an arbitrary reference, not wall-clock — it cannot be converted losslessly to SystemTime. Relative frame ordering survives via channel-send order.
• Adds SystemTime to the std::time imports and VideoFrame to the crate::frame imports.

Commit 2 — fix(linux): bound the internal frame channel to stop unbounded memory growth

Capturer::build() wired an unbounded std::sync::mpsc::channel() between each platform's capture callback and the consumer. On Linux, the PipeWire callback thread copies every frame via to_vec() unconditionally, independent of whether get_next_frame() is being drained — so any time the consumer lags even briefly, frames pile up with zero backpressure.

Confirmed via heaptrack on a real KDE-Wayland capture session: ~1.44G of 1.47G leaked in a 25.76s run, traced entirely to this call site. Bounded the shared channel to mpsc::sync_channel(2) instead. Re-verified with the same repro run 5x longer (136.74s): total leaked dropped to ~77.6MB (unrelated debug-backtrace-formatting overhead), and the process_callback leak site is completely gone from the profile.

This changes Capturer::build() and the shared (non-cfg-gated) Engine::new() from mpsc::Sender<ChannelItem> to mpsc::SyncSender<ChannelItem>.

Commit 3 — fix: propagate SyncSender to win/mac + avoid blocking send on all engines

Commit 2's type change is not cfg-gated — Engine::new() unconditionally forwards the new SyncSender into win::create_capturer() and mac::create_capturer(), both of which still declared the old mpsc::Sender type. mpsc::Sender and mpsc::SyncSender are distinct, non-coercible std types, so commit 2 alone breaks the Windows and macOS build — exactly the same class of bug this PR opened by fixing on Linux, just introduced in the opposite direction. This repo's PR CI doesn't build those targets, so it wouldn't have surfaced before merge; I only caught it by reading win/mod.rs/mac/mod.rs directly against the new signature.

This commit:

• Propagates mpsc::Sender → mpsc::SyncSender through win/mod.rs (Capturer, FlagStruct, create_capturer, spawn_audio_stream) and mac/mod.rs (CapturerInner, create_capturer), matching Engine::new().
• Replaces blocking .send() with .try_send() on every frame/sample emit site across all three engines, dropping the frame when the channel is full instead of blocking the capture callback thread. This matters most on Linux: process_callback runs on the same thread that pipewire_capturer's loop polls CAPTURER_STATE on, and LinuxCapturer::stop_capture() .join()s that thread. A blocking send() on a full bounded channel (e.g. the consumer paused draining get_next_frame()) would leave that thread permanently parked inside send(), so stop_capture() would hang forever. try_send() + drop-on-full keeps commit 2's bounded-memory fix without introducing that deadlock. Applied the same pattern to Windows' capture/audio callbacks and macOS' ScreenCaptureKit delivery callback, since an OS-driven callback thread blocking indefinitely isn't something worth newly introducing there either.

Verification

• cargo check on Linux (Ubuntu 24, pipewire 1.0, via nix-shell -p pipewire dbus pkg-config clang) passes clean after all three commits: same 16 pre-existing lifetime-elision warnings as before, zero new warnings or errors.
• win/mod.rs and mac/mod.rs could not be compiled or type-checked in my environment (no Windows/macOS toolchain, and their #[cfg(target_os = ...)]-gated mod declarations mean rustc never even parses them on a Linux host). I verified commit 3's changes to those files by: (1) rustfmt --check, which requires syntactically valid Rust to run at all — both pass; (2) manually cross-referencing every mpsc type at each call site against its declaration. I'd appreciate a maintainer (or CI) actually building this branch on Windows and macOS before merging — that's the one thing I can't self-verify here.

Follow-ups not in this PR

• The pts value is currently discarded (let _ = timestamp;) since we can't faithfully convert monotonic-ns to wall-clock SystemTime. A follow-up could add a monotonic_time: Option<Duration> sibling field if callers want sub-ms buffer timing on Linux. Left as a separate concern.
• The lifetime-elision warnings in portal.rs and a few other files are out of scope here but would be an easy clippy pass.
• A channel capacity of 2 is a somewhat arbitrary starting bound (chosen to eliminate the observed leak with margin) — if real-world usage shows it dropping frames too eagerly under normal jitter, it's an easy follow-up to raise it or make it configurable via Options.

Context

Found while integrating scap into xenia-capture, the screen-capture abstraction of the Xenia remote-session stack — we picked scap as our primary cross-platform backend specifically because of your PipeWire-native Wayland path. Thanks for the work on this crate.

[tembo]

Now that the producer/consumer path is sync_channel(2), send() here can block the PipeWire process callback when the consumer is behind. In a realtime-ish callback, that usually hurts more than dropping a frame.

Consider switching to try_send() and treating Full as a silent drop (and only logging on Disconnected).

Suggested change

	VideoFormat::RGBx => user_data.tx.send(Frame::Video(VideoFrame::RGBx(RGBxFrame {
	let send_result = match user_data.format.format() {
	VideoFormat::RGBx => user_data
	.tx
	.try_send(Frame::Video(VideoFrame::RGBx(RGBxFrame {
	display_time,
	width: frame_size.width as i32,
	height: frame_size.height as i32,
	data: frame_data,
	}))),
	VideoFormat::RGB => user_data
	.tx
	.try_send(Frame::Video(VideoFrame::RGB(RGBFrame {
	display_time,
	width: frame_size.width as i32,
	height: frame_size.height as i32,
	data: frame_data,
	}))),
	VideoFormat::xBGR => user_data
	.tx
	.try_send(Frame::Video(VideoFrame::XBGR(XBGRFrame {
	display_time,
	width: frame_size.width as i32,
	height: frame_size.height as i32,
	data: frame_data,
	}))),
	VideoFormat::BGRx => user_data
	.tx
	.try_send(Frame::Video(VideoFrame::BGRx(BGRxFrame {
	display_time,
	width: frame_size.width as i32,
	height: frame_size.height as i32,
	data: frame_data,
	}))),
	_ => panic!("Unsupported frame format received"),
	};
	match send_result {
	Ok(()) => {}
	Err(mpsc::TrySendError::Full(_)) => {}
	Err(mpsc::TrySendError::Disconnected(_)) => {
	eprintln!("frame channel disconnected");
	}
	}

[Tristan-Stoltz-ERC]

This was already fixed in 9ed0211 before I saw this comment — switched to try_send() with Full as a silent drop and only logging on Disconnected (and per your later comment on this same line, that Disconnected log is now also deduped via STREAM_STATE_CHANGED_TO_ERROR.swap() in 2ae046c). Matches your suggestion almost exactly — good instinct.

[tembo]

Minor wording nit: this is unbounded growth / retention rather than a true leak, so the comment reads a bit scary as-written.

Suggested change

	// limit -- confirmed via heaptrack: ~1.44G leaked over a 25s run,
	// consumer (encode/network backpressure) grows this queue without
	// limit -- confirmed via heaptrack: ~1.44GiB retained over a 25s run,

[Tristan-Stoltz-ERC]

Fixed in 9b30e91 — changed to "~1.44GiB retained", matching your wording exactly.

[tembo]

If the receiver is disconnected, this will eprintln! once per frame until the capture thread stops. Since you already have STREAM_STATE_CHANGED_TO_ERROR, you can set it here (to exit the loop) and use swap to log only once.

Suggested change

	if let Err(mpsc::TrySendError::Disconnected(_)) = send_result {
	if let Err(mpsc::TrySendError::Disconnected(_)) = send_result {
	if !STREAM_STATE_CHANGED_TO_ERROR.swap(true, std::sync::atomic::Ordering::Relaxed) {
	eprintln!("Frame receiver disconnected");
	}
	}

[Tristan-Stoltz-ERC]

Good catch — applied exactly as suggested in 2ae046c. Reused STREAM_STATE_CHANGED_TO_ERROR so a disconnected receiver now ends the capture loop instead of spinning forever, and swapped to log the disconnect once.

[tembo]

Minor nit: 2 feels a bit magic here. Maybe make it a named const so it's easier to tune later without hunting the callsite.

Suggested change

	let (tx, rx) = mpsc::sync_channel(2);
	const FRAME_CHANNEL_CAPACITY: usize = 2;
	let (tx, rx) = mpsc::sync_channel(FRAME_CHANNEL_CAPACITY);

[Tristan-Stoltz-ERC]

Done in 95e1e9e — extracted to FRAME_CHANNEL_CAPACITY.

[tembo]

Small readability thing: STREAM_STATE_CHANGED_TO_ERROR is now also used for "receiver disconnected" (not just PipeWire state error). Might be worth renaming to something like STREAM_SHOULD_EXIT (or splitting into two flags) so the intent stays obvious as this grows.

[Tristan-Stoltz-ERC]

Renamed to STREAM_SHOULD_EXIT in 95e1e9e — that's more accurate now anyway since it's a shared 'stop the loop' signal for two distinct causes (stream error, disconnected receiver), not just the error case. Also fixed the now-stale inline comment on the loop condition that only mentioned the error case.

[tembo]

try_send makes sense here. Minor follow-up: if this starts returning TrySendError::Disconnected, you may want to treat that as a signal to stop the sc::Stream (or flip some shared flag) so ScreenCaptureKit isn’t continuously delivering buffers with no consumer.

[Tristan-Stoltz-ERC]

Fixed in 72c78cd -- went with your 'flip some shared flag' alternative since impl_stream_did_output_sample_buf has no return channel to signal a stop the way Windows' on_frame_arrived does (it's an extern C callback returning unit). Added a disconnected: Arc on CapturerInner, set once on the first Disconnected (swap, logged once) and checked at the top of the callback so later frames skip the retain+send work entirely. Didn't attempt to call into sc::Stream's stop path from inside the delivery callback itself -- no way for me to verify that's reentrant-safe without a macOS toolchain, so I kept this to the safer flag-only mitigation you offered as the alternative.

[tembo]

One edge case with STREAM_SHOULD_EXIT being a static: if the receiver is dropped without stop_capture(), the flag gets set to true and never reset, so a later capture session in the same process can exit immediately.

Suggested change

	// User has called Capturer::start() and we start the main loop
	while CAPTURER_STATE.load(std::sync::atomic::Ordering::Relaxed) == 1
	&& /* If the stream state got changed to `Error`, we exit. TODO: tell user that we exited */
	!STREAM_STATE_CHANGED_TO_ERROR.load(std::sync::atomic::Ordering::Relaxed)
	&& /* Exit early on a PipeWire stream error or a disconnected frame receiver. TODO: tell user that we exited */
	!STREAM_SHOULD_EXIT.load(std::sync::atomic::Ordering::Relaxed)
	{
	// User has called Capturer::start() and we start the main loop
	STREAM_SHOULD_EXIT.store(false, std::sync::atomic::Ordering::Relaxed);
	while CAPTURER_STATE.load(std::sync::atomic::Ordering::Relaxed) == 1
	&& /* Exit early on a PipeWire stream error or a disconnected frame receiver. TODO: tell user that we exited */
	!STREAM_SHOULD_EXIT.load(std::sync::atomic::Ordering::Relaxed)
	{