render: fix 4K playback memory growth and decode-to-target-size

Root causes found for the 4K stalls and the second-footage memory
blowup (audit + code review):

- ticket bookkeeping leaked unbounded: the procpool ticket table and
  the arena slot map only ever grew (50-100 tickets/sec during
  playback, each pinning montage params and shm region views).
  Completed/cancelled/superseded/crashed entries are now removed, and
  the arena reaps fire-and-forget tickets once finished; the sync poll
  path reaps via a terminal result() read. InFlight duplicate submits
  now answer State immediately instead of sitting in the map forever.
- decode ran a full-resolution swscale to F32 RGBA (~132 MB at 4K)
  plus a second full-res copy before downscaling to the 480px proxy:
  RetrieveVideoParams.target_size lets swscale convert AND resize in
  one pass (bilinear, matching the old Rust resampler), so a 4K
  preview frame costs ~1 MB instead of ~260 MB of churn. This applies
  to proxy AND full-res requests alike.
- per-process decoder cache was unbounded (each session pins an FFmpeg
  context + 2 native decoded frames): LRU-capped at 16, eviction drops
  the map entry (in-flight renders keep their Arc; Drop releases
  FFmpeg).
- playback window completions were not generation-gated: a stale
  render from before an edit landed in the rebuilt window (wrong frame
  displayed, fresh request blocked). Stale completions now return
  their shm slot credit instead.
- async audio prefetch used the polling ticket submit without ever
  polling: switched to the fire-and-forget submit so entries reap.

crates/oak-render/src/eval.rs

@@ -485,14 +485,31 @@ pub fn render_produced_frame(
 
 /// Process-wide open decoder sessions, keyed by (filename, stream).
 /// Sessions are mutex-serialized inside the oakcodec box, so sharing
-/// one handle across worker threads is safe.
+/// one handle across worker threads is safe. The value carries an LRU
+/// tick: the map is capped ([`MAX_CACHED_DECODERS`]) because every
+/// session pins an FFmpeg context plus up to two native decoded frames
+/// (~50 MB at 4K) — before the cap, scrubbing a footage bin grew the
+/// map without bound.
 static DECODERS: std::sync::OnceLock<
-	std::sync::Mutex<std::collections::HashMap<(String, i32), Arc<dyn oak_codec::decoder::Decoder>>>,
+	std::sync::Mutex<
+		std::collections::HashMap<(String, i32), (Arc<dyn oak_codec::decoder::Decoder>, u64)>,
+	>,
 > = std::sync::OnceLock::new();
 
+/// Cap on cached decoder sessions per process (LRU beyond this). 16
+/// covers heavy multi-clip montages without reopen thrash; eviction only
+/// drops the map entry — an in-flight render keeps its Arc alive and the
+/// session dies with the last reference (Drop releases FFmpeg).
+const MAX_CACHED_DECODERS: usize = 16;
+
+/// LRU tick source for [`DECODERS`].
+static DECODER_TICK: std::sync::atomic::AtomicU64 = std::sync::atomic::AtomicU64::new(1);
+
 fn decoders(
-) -> std::sync::MutexGuard<'static, std::collections::HashMap<(String, i32), Arc<dyn oak_codec::decoder::Decoder>>>
-{
+) -> std::sync::MutexGuard<
+	'static,
+	std::collections::HashMap<(String, i32), (Arc<dyn oak_codec::decoder::Decoder>, u64)>,
+> {
 	DECODERS
 		.get_or_init(|| std::sync::Mutex::new(std::collections::HashMap::new()))
 		.lock()
@@ -501,9 +518,12 @@ fn decoders(
 
 /// Open (or reuse) the decoder session for `(filename, stream_index)`.
 fn open_decoder(filename: &str, stream_index: i32) -> Result<Arc<dyn oak_codec::decoder::Decoder>> {
+	let key = (filename.to_string(), stream_index);
+	let tick = DECODER_TICK.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
 	{
-		let cache = decoders();
-		if let Some(d) = cache.get(&(filename.to_string(), stream_index)) {
+		let mut cache = decoders();
+		if let Some((d, t)) = cache.get_mut(&key) {
+			*t = tick;
 			return Ok(d.clone());
 		}
 	}
@@ -513,7 +533,19 @@ fn open_decoder(filename: &str, stream_index: i32) -> Result<Arc<dyn oak_codec::
 		.open(&stream)
 		.map_err(|e| Error::Failed(format!("footage decode open: {e:?}")))?;
 	let mut cache = decoders();
-	cache.insert((filename.to_string(), stream_index), decoder.clone());
+	// LRU eviction: the session is dropped here, but an in-flight render
+	// holds its own Arc — the FFmpeg context dies with the last reference.
+	while cache.len() >= MAX_CACHED_DECODERS {
+		let victim = cache
+			.iter()
+			.min_by_key(|(_, (_, t))| *t)
+			.map(|(k, _)| k.clone());
+		let Some(victim) = victim else {
+			break;
+		};
+		cache.remove(&victim);
+	}
+	cache.insert(key, (decoder.clone(), tick));
 	Ok(decoder)
 }
 
@@ -527,6 +559,7 @@ pub fn render_footage_frame(
 	format: PixelFormat,
 ) -> Result<Texture> {
 	let decoder = open_decoder(filename, stream_index)?;
+	let (w, h) = size;
 	let params = RetrieveVideoParams {
 		stream: CodecStream::with_block(filename.to_string(), stream_index, None),
 		time,
@@ -537,6 +570,13 @@ pub fn render_footage_frame(
 		image_sequence_number: 0,
 		mode: RenderMode::Offline,
 		alpha_is_premultiplied: false,
+		// 直接按目标尺寸出帧：swscale 一次完成格式转换 + 缩放，不再
+		// 产生全分辨率 F32 中间帧（4K 预览每帧省 ~260MB 瞬时拷贝）。
+		target_size: if w > 0 && h > 0 {
+			Some((w as u32, h as u32))
+		} else {
+			None
+		},
 	};
 	let decoded = decoder
 		.retrieve_video_frame(&params)
@@ -545,7 +585,6 @@ pub fn render_footage_frame(
 	let src_w = decoded.width();
 	let src_h = decoded.height();
 	let src_linesize = decoded.linesize_bytes();
-	let (w, h) = size;
 	if src_w <= 0 || src_h <= 0 || src_linesize <= 0 || !decoded.is_allocated() {
 		return Err(Error::Failed("footage decode: bad decoded frame".into()));
 	}

crates/oak-render/src/procpool.rs

@@ -1005,7 +1005,9 @@ impl ProcessDispatcher {
 				.find(|(_, pt)| &pt.key == key)
 				.map(|(id, _)| *id);
 			if let Some(id) = ticket {
-				if let Some(pt) = inner.tickets.get_mut(&id) {
+				// 完成即移除（ticket 表只增不减是内存泄漏——播放时
+				// 每秒 50-100 个 ticket 全部永久驻留）。
+				if let Some(mut pt) = inner.tickets.remove(&id) {
 					if let Some(done) = pt.done.take() {
 						fired.push((done, Err(Error::State)));
 					}
@@ -1224,9 +1226,9 @@ impl ProcessDispatcher {
 		let shm = handle.shm.clone();
 		handle.held.insert(slot as u32);
 
-		let pt = inner.tickets.get_mut(&ticket);
+		let pt = inner.tickets.remove(&ticket);
 		match pt {
-			Some(pt) => {
+			Some(mut pt) => {
 				let key = pt.key;
 				inner.scheduler.frame_done(&key);
 				if let Some(done) = pt.done.take() {
@@ -1291,7 +1293,7 @@ impl ProcessDispatcher {
 		// The worker acquired the slot but never published it: the
 		// dispatcher (drainer) hands it back to the free pool.
 		self.recycle_slot(inner, worker, slot);
-		if let Some(pt) = inner.tickets.get_mut(&ticket) {
+		if let Some(mut pt) = inner.tickets.remove(&ticket) {
 			inner.scheduler.frame_failed(&pt.key);
 			if let Some(done) = pt.done.take() {
 				fired.push((done, Err(Error::Failed(format!("render failed: {error}")))));
@@ -1573,7 +1575,7 @@ impl ProcessDispatcher {
 			inner.workers[worker].state = WorkerState::PermanentlyDead;
 			for req in reclaimed {
 				inner.scheduler.cancel_key(&req.key);
-				if let Some(pt) = inner.tickets.get_mut(&req.payload) {
+				if let Some(mut pt) = inner.tickets.remove(&req.payload) {
 					if let Some(done) = pt.done.take() {
 						fired.push((
 							done,
@@ -1733,17 +1735,25 @@ impl JobDispatch for ProcessDispatcher {
 				SubmitOutcome::Accepted => {}
 				SubmitOutcome::Replaced(old) => {
 					// A newer request for the same key superseded the old
-					// pending one: cancel the old ticket's completion.
-					if let Some(pt) = inner.tickets.get_mut(&old.payload) {
+					// pending one: cancel the old ticket's completion (and
+					// reap the entry — the table must not grow unbounded).
+					if let Some(mut pt) = inner.tickets.remove(&old.payload) {
 						if let Some(done) = pt.done.take() {
 							fired.push((done, Err(Error::State)));
 						}
 					}
 				}
 				SubmitOutcome::InFlight => {
-					// Already claimed by a worker; the rendered frame is
-					// still valid for the same params (playback window
-					// slides re-request frames that are in flight).
+					// Already claimed by a worker under the same key: the
+					// worker will deliver the OLD ticket only. Fire the new
+					// ticket's completion as cancelled so its entry does
+					// not leak and the caller (playback window) may
+					// re-request once the in-flight render lands.
+					if let Some(mut pt) = inner.tickets.remove(&id) {
+						if let Some(done) = pt.done.take() {
+							fired.push((done, Err(Error::State)));
+						}
+					}
 				}
 			}
 		}
@@ -1766,7 +1776,7 @@ impl JobDispatch for ProcessDispatcher {
 			let mut inner = lock(&self.inner);
 			let dropped = inner.scheduler.cancel_sequence(sequence);
 			for request in dropped {
-				if let Some(pt) = inner.tickets.get_mut(&request.payload) {
+				if let Some(mut pt) = inner.tickets.remove(&request.payload) {
 					if let Some(done) = pt.done.take() {
 						fired.push((done, Err(Error::State)));
 					}
@@ -1866,12 +1876,15 @@ impl JobDispatch for ProcessDispatcher {
 				}
 				w.stdin = None;
 			}
-			// Every ticket still open completes with cancellation.
+			// Every ticket still open completes with cancellation; the map
+			// is dropped with the dispatcher (clear for hygiene — leaked
+			// entries pin shm region views).
 			for (_, pt) in inner.tickets.iter_mut() {
 				if let Some(done) = pt.done.take() {
 					fired.push((done, Err(Error::State)));
 				}
 			}
+			inner.tickets.clear();
 		}
 		for (done, result) in fired {
 			done(result);

crates/oak-render/src/ticket.rs

@@ -231,6 +231,13 @@ struct TicketSlot {
 	/// whose render completed must recycle its shm slot — the consumer
 	/// never sees the `ShmFrame` payload).
 	dispatch: Arc<dyn JobDispatch>,
+	/// The caller polls `wait()`/`result()` after completion (the sync
+	/// render path): the arena entry must survive `finish()` until
+	/// `result()` reaps it. Fire-and-forget tickets (completion-only:
+	/// playback window, autocacher, manager) are reaped at the next
+	/// `allocate()` once finished — the arena map must not grow
+	/// unbounded (playback posts 50-100 tickets/sec).
+	poll_after: bool,
 }
 
 impl TicketSlot {
@@ -348,7 +355,12 @@ impl TicketArena {
 
 	fn allocate(&self, slot: Arc<TicketSlot>) -> TicketId {
 		let id = slot.id;
-		lock(&self.slots).insert(id, slot);
+		let mut slots = lock(&self.slots);
+		slots.insert(id, slot);
+		// Reap finished fire-and-forget tickets (their completions were
+		// delivered at finish(); nobody will poll them). Poll-path slots
+		// survive until `result()` reaps them.
+		slots.retain(|_, s| !(s.is_finished() && !s.poll_after));
 		id
 	}
 
@@ -365,24 +377,29 @@ impl TicketArena {
 	/// on cancellation (with `Error::State`). The job posts as a Seek
 	/// single-frame request (M15 S2; see
 	/// [`TicketArena::submit_playback`] for the pre-render window).
+	/// The caller is expected to poll `wait()`/`result()` (the sync
+	/// render path); the arena entry survives until `result()` reaps it.
 	pub fn submit_video_with_id(
 		&self,
 		id: TicketId,
 		params: VideoTicketParams,
 		done: Completion,
 	) -> TicketId {
-		self.submit_video_impl(id, params, done, JobSchedule::seek())
+		self.submit_video_impl(id, params, done, JobSchedule::seek(), true)
 	}
 
 	/// The shared video-ticket submit path: register the slot, post the
 	/// job with `schedule` (M15 S2), and deliver `Error::State`
-	/// immediately when the backend is gone.
+	/// immediately when the backend is gone. `poll_after` marks entries
+	/// the caller reaps via `result()` (sync path) rather than the
+	/// completion (fire-and-forget — reaped once finished).
 	fn submit_video_impl(
 		&self,
 		id: TicketId,
 		params: VideoTicketParams,
 		done: Completion,
 		schedule: JobSchedule,
+		poll_after: bool,
 	) -> TicketId {
 		let meta = TicketMeta {
 			kind: Some(ticket_kind::VIDEO),
@@ -404,6 +421,7 @@ impl TicketArena {
 			completion: Mutex::new(Some(done)),
 			result: Mutex::new(None),
 			dispatch: self.dispatch.clone(),
+			poll_after,
 		});
 		self.allocate(slot.clone());
 
@@ -446,6 +464,7 @@ impl TicketArena {
 			params,
 			done,
 			JobSchedule::playback(frame, distance, version),
+			false,
 		)
 	}
 
@@ -468,24 +487,38 @@ impl TicketArena {
 		params: VideoTicketParams,
 		done: Completion,
 	) -> TicketId {
-		self.submit_video_impl(id, params, done, JobSchedule::background())
+		self.submit_video_impl(id, params, done, JobSchedule::background(), false)
 	}
 
 	/// Submit a video ticket; completion fires exactly once, including
 	/// on cancellation (with `Error::State`).
 	pub fn submit_video(&self, params: VideoTicketParams, done: Completion) -> TicketId {
 		let id = self.next_id();
-		self.submit_video_with_id(id, params, done)
+		self.submit_video_impl(id, params, done, JobSchedule::seek(), false)
 	}
 
 	/// Submit an audio ticket (range pull; C++ render_audio) with a
 	/// caller-reserved id (allocated by [`TicketArena::next_id`]). The
-	/// completion fires exactly once.
+	/// completion fires exactly once. The caller is expected to poll
+	/// `wait()`/`result()` (the sync path); the arena entry survives
+	/// until `result()` reaps it.
 	pub fn submit_audio_with_id(
 		&self,
 		id: TicketId,
 		params: AudioTicketParams,
 		done: Completion,
+	) -> TicketId {
+		self.submit_audio_impl(id, params, done, true)
+	}
+
+	/// The shared audio-ticket submit path (`poll_after` 语义同
+	/// [`TicketArena::submit_video_impl`]).
+	fn submit_audio_impl(
+		&self,
+		id: TicketId,
+		params: AudioTicketParams,
+		done: Completion,
+		poll_after: bool,
 	) -> TicketId {
 		let range = params.range;
 		let meta = TicketMeta {
@@ -506,6 +539,7 @@ impl TicketArena {
 			completion: Mutex::new(Some(done)),
 			result: Mutex::new(None),
 			dispatch: self.audio_dispatch.clone(),
+			poll_after,
 		});
 		self.allocate(slot.clone());
 
@@ -562,7 +596,7 @@ impl TicketArena {
 	/// completion fires exactly once.
 	pub fn submit_audio(&self, params: AudioTicketParams, done: Completion) -> TicketId {
 		let id = self.next_id();
-		self.submit_audio_with_id(id, params, done)
+		self.submit_audio_impl(id, params, done, false)
 	}
 
 	/// True when the ticket has finished.
@@ -602,7 +636,9 @@ impl TicketArena {
 	/// The ticket result, when finished (clone; unknown/unfinished ids give
 	/// `None`).
 	pub fn result(&self, id: TicketId) -> Option<TicketResult> {
-		lock(&self.slots).get(&id).and_then(|s| s.result())
+		// Terminal read: the poll path's only observation point — reap
+		// the entry with it (the arena map must not grow unbounded).
+		lock(&self.slots).remove(&id).and_then(|s| s.result())
 	}
 
 	/// Ticket metadata query (C++ property()).
@@ -716,6 +752,9 @@ mod tests {
 			"exactly once"
 		);
 
+		// is_finished 必须在 result() 之前查：result() 是终止性读取
+		// （连同条目一起回收，arena 表不能无限增长）。
+		assert!(arena.is_finished(id));
 		let res = arena.result(id).unwrap().unwrap();
 		assert_eq!(
 			match res {
@@ -724,7 +763,6 @@ mod tests {
 			},
 			(4, 4)
 		);
-		assert!(arena.is_finished(id));
 		video.shutdown();
 	}