- commit
- a0e7153ef0506b893e6eed5f8f6638444e19679a
- parent
- 6aa71e0356f4f1f5f9b1acb011e0bb9bdb5fdb73
- author
- Dami <dami.etoile@gmail.com>
| M | crates/ | 334 | +----------------------------- |
| A | crates/ | 321 | +++++++++++++++++++++++++++++ |
@@ -63,10 +63,6 @@ const MAX_PENDING_HANDSHAKES: usize = 4;
6363 // Allow one frame of ordinary scheduler/write jitter beyond the frame being6464 // sent, but never replay a suffix already more than two configured periods old.6565 const MAX_MEDIA_REPLAY_AGE_FRAME_PERIODS: u64 = 2;66-const GAMESCOPE_STARTUP_SAMPLE_WINDOW: Duration = Duration::from_millis(500);67-const GAMESCOPE_STARTUP_MIN_OBSERVATION_SPAN: Duration = Duration::from_millis(100);68-const GAMESCOPE_STARTUP_TARGET_MIN_FRAMES: u64 = 8;69-const GAMESCOPE_STARTUP_TARGET_MIN_FPS: f64 = 45.0;7066 const SOURCE_REAP_GRACE_TIMEOUT: Duration = Duration::from_secs(1);7167 const MOQ_ATTACHMENT_TIMEOUT: Duration = Duration::from_secs(10);7268 const MOQ_REJECT_CODE: u32 = 0x534d;@@ -74,6 +70,7 @@ const ENCODER_CONTROL_COMMIT_TIMEOUT: Duration = Duration::from_secs(2);
7470 7571 mod adaptive;7672 mod session;73+mod startup;7774 7875 7976 pub(crate) use adaptive::MotionResolutionPolicy;@@ -85,186 +82,7 @@ use session::{
8582 ClaimedMoqAttachment, ForcedIdrCoordinator, ForcedIdrDisposition, MediaV3Telemetry,8683 MoqAttachmentWait, SourceTaskGuard,8784 };88-89-struct StartupCadenceSample {90- first_sequence: Option<u64>,91- last_sequence: Option<u64>,92- first_observed_at: Option<Instant>,93- last_observed_at: Option<Instant>,94- receiver_open: bool,95- decodable_gop_ready: bool,96-}97-98-impl StartupCadenceSample {99- fn observe(&mut self, frame: &EncodedFrame) {100- if self.first_sequence.is_none() {101- self.first_sequence = Some(frame.sequence);102- self.first_observed_at = Some(frame.observed_at);103- }104- if self105- .last_sequence106- .is_none_or(|sequence| frame.sequence >= sequence)107- {108- self.last_sequence = Some(frame.sequence);109- self.last_observed_at = Some(frame.observed_at);110- }111- }112-113- fn frame_progress(self) -> u64 {114- self.first_sequence115- .zip(self.last_sequence)116- .map_or(0, |(first, last)| last.saturating_sub(first) + 1)117- }118-119- fn fps(self) -> f64 {120- let frames = self.frame_progress();121- let elapsed = self.observation_span();122- if frames < 2 || elapsed.is_zero() {123- return 0.0;124- }125- (frames - 1) as f64 / elapsed.as_secs_f64()126- }127-128- fn observation_span(self) -> Duration {129- self.first_observed_at130- .zip(self.last_observed_at)131- .map_or(Duration::ZERO, |(first, last)| {132- last.saturating_duration_since(first)133- })134- }135-136- fn has_representative_span(self) -> bool {137- self.observation_span() >= GAMESCOPE_STARTUP_MIN_OBSERVATION_SPAN138- }139-140- fn is_usable(self) -> bool {141- self.receiver_open && self.frame_progress() > 0 && self.decodable_gop_ready142- }143-144- fn meets_target_cadence(self) -> bool {145- self.receiver_open146- && self.has_representative_span()147- && self.frame_progress() >= GAMESCOPE_STARTUP_TARGET_MIN_FRAMES148- && self.fps() >= GAMESCOPE_STARTUP_TARGET_MIN_FPS149- }150-}151-152-fn startup_source_needs_restart(sample: StartupCadenceSample) -> bool {153- !sample.is_usable()154-}155-156-async fn sample_startup_cadence(157- receiver: &mut tokio::sync::watch::Receiver<Option<EncodedFrame>>,158- current_gop: &tokio::sync::watch::Receiver<Option<EncodedGop>>,159- window: Duration,160-) -> StartupCadenceSample {161- let mut sample = StartupCadenceSample {162- receiver_open: true,163- ..StartupCadenceSample::default()164- };165- if let Some(frame) = receiver.borrow_and_update().as_ref() {166- sample.observe(frame);167- }168- sample.decodable_gop_ready = current_gop.borrow().is_some();169- let deadline = tokio::time::Instant::now() + window;170- while !sample.is_usable() || !sample.has_representative_span() {171- match tokio::time::timeout_at(deadline, receiver.changed()).await {172- Ok(Ok(())) => {173- if let Some(frame) = receiver.borrow_and_update().as_ref() {174- sample.observe(frame);175- }176- sample.decodable_gop_ready = current_gop.borrow().is_some();177- }178- Ok(Err(_)) => {179- sample.receiver_open = false;180- break;181- }182- Err(_) => break,183- }184- }185- sample186-}187-188-async fn reap_encoded_source(source: EncodedSource) {189- reap_encoded_source_with_timeout(source, SOURCE_REAP_GRACE_TIMEOUT).await;190-}191-192-async fn reap_encoded_source_with_timeout(source: EncodedSource, grace_timeout: Duration) {193- let EncodedSource {194- frames,195- current_gop,196- task,197- pointer_surface_dimensions: _,198- encoder_control: _,199- } = source;200- // Closing both bounded outputs gives the capture task a chance to kill and201- // wait for its GStreamer child itself. Abort is only the bounded fallback.202- drop(frames);203- drop(current_gop);204- SourceTaskGuard::new(task)205- .wait_or_abort(grace_timeout)206- .await;207-}208-209-async fn select_gamescope_startup_source(210- config: HostConfig,211- session_clock: SessionClock,212- mut primary: EncodedSource,213-) -> Result<EncodedSource> {214- let primary_sample = sample_startup_cadence(215- &mut primary.frames,216- &primary.current_gop,217- GAMESCOPE_STARTUP_SAMPLE_WINDOW,218- )219- .await;220- info!(221- frames = primary_sample.frame_progress(),222- fps = primary_sample.fps(),223- receiver_open = primary_sample.receiver_open,224- decodable_gop_ready = primary_sample.decodable_gop_ready,225- target_cadence = primary_sample.meets_target_cadence(),226- "sampled primary Gamescope capture startup"227- );228- if !startup_source_needs_restart(primary_sample) {229- return Ok(primary);230- }231-232- // Gamescope's PipeWire export does not reliably fan out full cadence to233- // two simultaneous consumers. Reap an unhealthy startup pipeline before234- // opening its replacement; overlapping them can divide vblank deliveries235- // and permanently strand the selected source at half rate.236- reap_encoded_source(primary).await;237- let mut replacement = spawn_gamescope_pipewire_after_static_preflight(config, session_clock)238- .await239- .context("restarting unhealthy Gamescope capture pipeline")?;240- let replacement_sample = sample_startup_cadence(241- &mut replacement.frames,242- &replacement.current_gop,243- GAMESCOPE_STARTUP_SAMPLE_WINDOW,244- )245- .await;246- info!(247- frames = replacement_sample.frame_progress(),248- fps = replacement_sample.fps(),249- receiver_open = replacement_sample.receiver_open,250- decodable_gop_ready = replacement_sample.decodable_gop_ready,251- target_cadence = replacement_sample.meets_target_cadence(),252- "sampled sequential Gamescope capture startup replacement"253- );254- if !replacement_sample.is_usable() {255- let frames = replacement_sample.frame_progress();256- let fps = replacement_sample.fps();257- let receiver_open = replacement_sample.receiver_open;258- let decodable_gop_ready = replacement_sample.decodable_gop_ready;259- reap_encoded_source(replacement).await;260- bail!(261- "replacement Gamescope capture pipeline remained unhealthy: \262- frames={frames}, fps={fps:.2}, receiver_open={receiver_open}, \263- decodable_gop_ready={decodable_gop_ready}"264- );265- }266- Ok(replacement)267-}85+use startup::select_gamescope_startup_source;26886 26987 27088 enum MediaActivity {@@ -3029,18 +2847,20 @@ fn moq_test_config() -> HostConfig {
30292847 }30302848 30312849 3032-mod tests {3033- use super::*;3034-3035- struct DropNotify(Option<tokio::sync::oneshot::Sender<()>>);2850+struct DropNotify(Option<tokio::sync::oneshot::Sender<()>>);30362851 3037- impl Drop for DropNotify {3038- fn drop(&mut self) {3039- if let Some(sender) = self.0.take() {3040- let _ = sender.send(());3041- }2852+2853+impl Drop for DropNotify {2854+ fn drop(&mut self) {2855+ if let Some(sender) = self.0.take() {2856+ let _ = sender.send(());30422857 }30432858 }2859+}2860+2861+2862+mod tests {2863+ use super::*;30442864 30452865 30462866 fn media_v2_scheduler_is_bounded_and_drops_a_saturated_delta_suffix() {@@ -3764,134 +3584,6 @@ mod tests {
37643584 assert!(media_v3_transport_priority(1).is_err());37653585 }37663586 3767- fn startup_sample(fps: f64, frames: u64, receiver_open: bool) -> StartupCadenceSample {3768- if frames == 0 {3769- return StartupCadenceSample {3770- receiver_open,3771- ..StartupCadenceSample::default()3772- };3773- }3774- let first = Instant::now();3775- let span = if frames > 1 && fps > 0.0 {3776- Duration::from_secs_f64((frames - 1) as f64 / fps)3777- } else {3778- Duration::ZERO3779- };3780- StartupCadenceSample {3781- first_sequence: Some(0),3782- last_sequence: Some(frames - 1),3783- first_observed_at: Some(first),3784- last_observed_at: Some(first + span),3785- receiver_open,3786- decodable_gop_ready: true,3787- }3788- }3789-3790- 3791- fn startup_restart_requires_a_live_decodable_source_not_target_cadence() {3792- let slow = startup_sample(12.0, 6, true);3793- let target = startup_sample(60.0, 8, true);3794- assert!(!startup_source_needs_restart(slow));3795- assert!(!slow.meets_target_cadence());3796- assert!(!startup_source_needs_restart(target));3797- assert!(target.meets_target_cadence());3798- assert!(startup_source_needs_restart(startup_sample(60.0, 8, false)));3799- assert!(startup_source_needs_restart(startup_sample(0.0, 0, true)));3800- let mut undecodable = startup_sample(60.0, 8, true);3801- undecodable.decodable_gop_ready = false;3802- assert!(startup_source_needs_restart(undecodable));3803- }3804-3805- 3806- fn startup_bursts_are_not_sustained_health() {3807- let burst = startup_sample(10_000.0, 8, true);3808- assert!(burst.fps() > GAMESCOPE_STARTUP_TARGET_MIN_FPS);3809- assert!(!burst.has_representative_span());3810- assert!(!burst.meets_target_cadence());3811- assert!(!startup_source_needs_restart(burst));3812- }3813-3814- 3815- async fn startup_sampling_preserves_current_decodable_gop() {3816- let now = Instant::now();3817- let frame = EncodedFrame {3818- sequence: 0,3819- width: 1_280,3820- height: 800,3821- capture_timestamp_micros: 0,3822- presentation_timestamp_micros: 0,3823- observed_at: now,3824- keyframe: true,3825- codec_config: true,3826- discontinuity: false,3827- data: Arc::from([1_u8, 2, 3]),3828- };3829- let (_frame_sender, mut frame_receiver) = tokio::sync::watch::channel(Some(frame.clone()));3830- let (_gop_sender, gop_receiver) = tokio::sync::watch::channel(Some(EncodedGop {3831- frames: vec![frame],3832- payload_bytes: 3,3833- }));3834-3835- let sample =3836- sample_startup_cadence(&mut frame_receiver, &gop_receiver, Duration::ZERO).await;3837- assert_eq!(sample.frame_progress(), 1);3838- let current_gop = gop_receiver.borrow().clone().unwrap();3839- assert_eq!(current_gop.frames.len(), 1);3840- assert!(current_gop.frames[0].keyframe && current_gop.frames[0].codec_config);3841- }3842-3843- 3844- async fn reaping_source_closes_outputs_before_waiting_for_task_cleanup() {3845- let (frame_sender, frame_receiver) = tokio::sync::watch::channel(None);3846- let (gop_sender, gop_receiver) = tokio::sync::watch::channel(None);3847- let (reaped_tx, reaped_rx) = tokio::sync::oneshot::channel();3848- let task = tokio::spawn(async move {3849- frame_sender.closed().await;3850- gop_sender.closed().await;3851- let _ = reaped_tx.send(());3852- Ok(())3853- });3854- let source = EncodedSource {3855- frames: frame_receiver,3856- current_gop: gop_receiver,3857- task,3858- pointer_surface_dimensions: None,3859- encoder_control: None,3860- };3861-3862- reap_encoded_source_with_timeout(source, Duration::from_millis(100)).await;3863- reaped_rx3864- .await3865- .expect("source task did not observe both closed outputs");3866- }3867-3868- 3869- async fn reaping_stalled_source_aborts_after_bounded_grace() {3870- let (_frame_sender, frame_receiver) = tokio::sync::watch::channel(None);3871- let (_gop_sender, gop_receiver) = tokio::sync::watch::channel(None);3872- let (started_tx, started_rx) = tokio::sync::oneshot::channel();3873- let (reaped_tx, reaped_rx) = tokio::sync::oneshot::channel();3874- let task = tokio::spawn(async move {3875- let _notify = DropNotify(Some(reaped_tx));3876- let _ = started_tx.send(());3877- std::future::pending::<Result<()>>().await3878- });3879- let source = EncodedSource {3880- frames: frame_receiver,3881- current_gop: gop_receiver,3882- task,3883- pointer_surface_dimensions: None,3884- encoder_control: None,3885- };3886-3887- started_rx.await.unwrap();3888- reap_encoded_source_with_timeout(source, Duration::from_millis(10)).await;3889- tokio::time::timeout(Duration::from_millis(100), reaped_rx)3890- .await3891- .expect("stalled source task was not aborted and reaped")3892- .unwrap();3893- }3894-38953587 38963588 fn capability_negotiation_is_an_exact_intersection() {38973589 let hello = ClientHello::new(@@ -0,0 +1,321 @@
1+use super::*;2+3+const GAMESCOPE_STARTUP_SAMPLE_WINDOW: Duration = Duration::from_millis(500);4+const GAMESCOPE_STARTUP_MIN_OBSERVATION_SPAN: Duration = Duration::from_millis(100);5+const GAMESCOPE_STARTUP_TARGET_MIN_FRAMES: u64 = 8;6+const GAMESCOPE_STARTUP_TARGET_MIN_FPS: f64 = 45.0;7+8+9+struct StartupCadenceSample {10+ first_sequence: Option<u64>,11+ last_sequence: Option<u64>,12+ first_observed_at: Option<Instant>,13+ last_observed_at: Option<Instant>,14+ receiver_open: bool,15+ decodable_gop_ready: bool,16+}17+18+impl StartupCadenceSample {19+ fn observe(&mut self, frame: &EncodedFrame) {20+ if self.first_sequence.is_none() {21+ self.first_sequence = Some(frame.sequence);22+ self.first_observed_at = Some(frame.observed_at);23+ }24+ if self25+ .last_sequence26+ .is_none_or(|sequence| frame.sequence >= sequence)27+ {28+ self.last_sequence = Some(frame.sequence);29+ self.last_observed_at = Some(frame.observed_at);30+ }31+ }32+33+ fn frame_progress(self) -> u64 {34+ self.first_sequence35+ .zip(self.last_sequence)36+ .map_or(0, |(first, last)| last.saturating_sub(first) + 1)37+ }38+39+ fn fps(self) -> f64 {40+ let frames = self.frame_progress();41+ let elapsed = self.observation_span();42+ if frames < 2 || elapsed.is_zero() {43+ return 0.0;44+ }45+ (frames - 1) as f64 / elapsed.as_secs_f64()46+ }47+48+ fn observation_span(self) -> Duration {49+ self.first_observed_at50+ .zip(self.last_observed_at)51+ .map_or(Duration::ZERO, |(first, last)| {52+ last.saturating_duration_since(first)53+ })54+ }55+56+ fn has_representative_span(self) -> bool {57+ self.observation_span() >= GAMESCOPE_STARTUP_MIN_OBSERVATION_SPAN58+ }59+60+ fn is_usable(self) -> bool {61+ self.receiver_open && self.frame_progress() > 0 && self.decodable_gop_ready62+ }63+64+ fn meets_target_cadence(self) -> bool {65+ self.receiver_open66+ && self.has_representative_span()67+ && self.frame_progress() >= GAMESCOPE_STARTUP_TARGET_MIN_FRAMES68+ && self.fps() >= GAMESCOPE_STARTUP_TARGET_MIN_FPS69+ }70+}71+72+fn startup_source_needs_restart(sample: StartupCadenceSample) -> bool {73+ !sample.is_usable()74+}75+76+async fn sample_startup_cadence(77+ receiver: &mut tokio::sync::watch::Receiver<Option<EncodedFrame>>,78+ current_gop: &tokio::sync::watch::Receiver<Option<EncodedGop>>,79+ window: Duration,80+) -> StartupCadenceSample {81+ let mut sample = StartupCadenceSample {82+ receiver_open: true,83+ ..StartupCadenceSample::default()84+ };85+ if let Some(frame) = receiver.borrow_and_update().as_ref() {86+ sample.observe(frame);87+ }88+ sample.decodable_gop_ready = current_gop.borrow().is_some();89+ let deadline = tokio::time::Instant::now() + window;90+ while !sample.is_usable() || !sample.has_representative_span() {91+ match tokio::time::timeout_at(deadline, receiver.changed()).await {92+ Ok(Ok(())) => {93+ if let Some(frame) = receiver.borrow_and_update().as_ref() {94+ sample.observe(frame);95+ }96+ sample.decodable_gop_ready = current_gop.borrow().is_some();97+ }98+ Ok(Err(_)) => {99+ sample.receiver_open = false;100+ break;101+ }102+ Err(_) => break,103+ }104+ }105+ sample106+}107+108+async fn reap_encoded_source(source: EncodedSource) {109+ reap_encoded_source_with_timeout(source, SOURCE_REAP_GRACE_TIMEOUT).await;110+}111+112+async fn reap_encoded_source_with_timeout(source: EncodedSource, grace_timeout: Duration) {113+ let EncodedSource {114+ frames,115+ current_gop,116+ task,117+ pointer_surface_dimensions: _,118+ encoder_control: _,119+ } = source;120+ // Closing both bounded outputs gives the capture task a chance to kill and121+ // wait for its GStreamer child itself. Abort is only the bounded fallback.122+ drop(frames);123+ drop(current_gop);124+ SourceTaskGuard::new(task)125+ .wait_or_abort(grace_timeout)126+ .await;127+}128+129+pub(super) async fn select_gamescope_startup_source(130+ config: HostConfig,131+ session_clock: SessionClock,132+ mut primary: EncodedSource,133+) -> Result<EncodedSource> {134+ let primary_sample = sample_startup_cadence(135+ &mut primary.frames,136+ &primary.current_gop,137+ GAMESCOPE_STARTUP_SAMPLE_WINDOW,138+ )139+ .await;140+ info!(141+ frames = primary_sample.frame_progress(),142+ fps = primary_sample.fps(),143+ receiver_open = primary_sample.receiver_open,144+ decodable_gop_ready = primary_sample.decodable_gop_ready,145+ target_cadence = primary_sample.meets_target_cadence(),146+ "sampled primary Gamescope capture startup"147+ );148+ if !startup_source_needs_restart(primary_sample) {149+ return Ok(primary);150+ }151+152+ // Gamescope's PipeWire export does not reliably fan out full cadence to153+ // two simultaneous consumers. Reap an unhealthy startup pipeline before154+ // opening its replacement; overlapping them can divide vblank deliveries155+ // and permanently strand the selected source at half rate.156+ reap_encoded_source(primary).await;157+ let mut replacement = spawn_gamescope_pipewire_after_static_preflight(config, session_clock)158+ .await159+ .context("restarting unhealthy Gamescope capture pipeline")?;160+ let replacement_sample = sample_startup_cadence(161+ &mut replacement.frames,162+ &replacement.current_gop,163+ GAMESCOPE_STARTUP_SAMPLE_WINDOW,164+ )165+ .await;166+ info!(167+ frames = replacement_sample.frame_progress(),168+ fps = replacement_sample.fps(),169+ receiver_open = replacement_sample.receiver_open,170+ decodable_gop_ready = replacement_sample.decodable_gop_ready,171+ target_cadence = replacement_sample.meets_target_cadence(),172+ "sampled sequential Gamescope capture startup replacement"173+ );174+ if !replacement_sample.is_usable() {175+ let frames = replacement_sample.frame_progress();176+ let fps = replacement_sample.fps();177+ let receiver_open = replacement_sample.receiver_open;178+ let decodable_gop_ready = replacement_sample.decodable_gop_ready;179+ reap_encoded_source(replacement).await;180+ bail!(181+ "replacement Gamescope capture pipeline remained unhealthy: \182+ frames={frames}, fps={fps:.2}, receiver_open={receiver_open}, \183+ decodable_gop_ready={decodable_gop_ready}"184+ );185+ }186+ Ok(replacement)187+}188+189+190+mod tests {191+ use super::super::DropNotify;192+ use super::*;193+194+ fn startup_sample(fps: f64, frames: u64, receiver_open: bool) -> StartupCadenceSample {195+ if frames == 0 {196+ return StartupCadenceSample {197+ receiver_open,198+ ..StartupCadenceSample::default()199+ };200+ }201+ let first = Instant::now();202+ let span = if frames > 1 && fps > 0.0 {203+ Duration::from_secs_f64((frames - 1) as f64 / fps)204+ } else {205+ Duration::ZERO206+ };207+ StartupCadenceSample {208+ first_sequence: Some(0),209+ last_sequence: Some(frames - 1),210+ first_observed_at: Some(first),211+ last_observed_at: Some(first + span),212+ receiver_open,213+ decodable_gop_ready: true,214+ }215+ }216+217+ 218+ fn startup_restart_requires_a_live_decodable_source_not_target_cadence() {219+ let slow = startup_sample(12.0, 6, true);220+ let target = startup_sample(60.0, 8, true);221+ assert!(!startup_source_needs_restart(slow));222+ assert!(!slow.meets_target_cadence());223+ assert!(!startup_source_needs_restart(target));224+ assert!(target.meets_target_cadence());225+ assert!(startup_source_needs_restart(startup_sample(60.0, 8, false)));226+ assert!(startup_source_needs_restart(startup_sample(0.0, 0, true)));227+ let mut undecodable = startup_sample(60.0, 8, true);228+ undecodable.decodable_gop_ready = false;229+ assert!(startup_source_needs_restart(undecodable));230+ }231+232+ 233+ fn startup_bursts_are_not_sustained_health() {234+ let burst = startup_sample(10_000.0, 8, true);235+ assert!(burst.fps() > GAMESCOPE_STARTUP_TARGET_MIN_FPS);236+ assert!(!burst.has_representative_span());237+ assert!(!burst.meets_target_cadence());238+ assert!(!startup_source_needs_restart(burst));239+ }240+241+ 242+ async fn startup_sampling_preserves_current_decodable_gop() {243+ let now = Instant::now();244+ let frame = EncodedFrame {245+ sequence: 0,246+ width: 1_280,247+ height: 800,248+ capture_timestamp_micros: 0,249+ presentation_timestamp_micros: 0,250+ observed_at: now,251+ keyframe: true,252+ codec_config: true,253+ discontinuity: false,254+ data: Arc::from([1_u8, 2, 3]),255+ };256+ let (_frame_sender, mut frame_receiver) = tokio::sync::watch::channel(Some(frame.clone()));257+ let (_gop_sender, gop_receiver) = tokio::sync::watch::channel(Some(EncodedGop {258+ frames: vec![frame],259+ payload_bytes: 3,260+ }));261+262+ let sample =263+ sample_startup_cadence(&mut frame_receiver, &gop_receiver, Duration::ZERO).await;264+ assert_eq!(sample.frame_progress(), 1);265+ let current_gop = gop_receiver.borrow().clone().unwrap();266+ assert_eq!(current_gop.frames.len(), 1);267+ assert!(current_gop.frames[0].keyframe && current_gop.frames[0].codec_config);268+ }269+270+ 271+ async fn reaping_source_closes_outputs_before_waiting_for_task_cleanup() {272+ let (frame_sender, frame_receiver) = tokio::sync::watch::channel(None);273+ let (gop_sender, gop_receiver) = tokio::sync::watch::channel(None);274+ let (reaped_tx, reaped_rx) = tokio::sync::oneshot::channel();275+ let task = tokio::spawn(async move {276+ frame_sender.closed().await;277+ gop_sender.closed().await;278+ let _ = reaped_tx.send(());279+ Ok(())280+ });281+ let source = EncodedSource {282+ frames: frame_receiver,283+ current_gop: gop_receiver,284+ task,285+ pointer_surface_dimensions: None,286+ encoder_control: None,287+ };288+289+ reap_encoded_source_with_timeout(source, Duration::from_millis(100)).await;290+ reaped_rx291+ .await292+ .expect("source task did not observe both closed outputs");293+ }294+295+ 296+ async fn reaping_stalled_source_aborts_after_bounded_grace() {297+ let (_frame_sender, frame_receiver) = tokio::sync::watch::channel(None);298+ let (_gop_sender, gop_receiver) = tokio::sync::watch::channel(None);299+ let (started_tx, started_rx) = tokio::sync::oneshot::channel();300+ let (reaped_tx, reaped_rx) = tokio::sync::oneshot::channel();301+ let task = tokio::spawn(async move {302+ let _notify = DropNotify(Some(reaped_tx));303+ let _ = started_tx.send(());304+ std::future::pending::<Result<()>>().await305+ });306+ let source = EncodedSource {307+ frames: frame_receiver,308+ current_gop: gop_receiver,309+ task,310+ pointer_surface_dimensions: None,311+ encoder_control: None,312+ };313+314+ started_rx.await.unwrap();315+ reap_encoded_source_with_timeout(source, Duration::from_millis(10)).await;316+ tokio::time::timeout(Duration::from_millis(100), reaped_rx)317+ .await318+ .expect("stalled source task was not aborted and reaped")319+ .unwrap();320+ }321+}