- commit
- dc79b65b7c894d8232d7fab7df7e4cc5e718b10c
- parent
- 1a9d5ab4ecd934cdc24e6913aecbc9fa521497c5
- author
- Dami <dami.etoile@gmail.com>
| R | crates/ | 461 | +----------------------------- |
| A | crates/ | 460 | ++++++++++++++++++++++++++++++ |
| M | crates/ | 4 | +- |
@@ -1,176 +1,10 @@
1-use std::future::Future;2-use std::pin::Pin;3-use std::sync::Arc;4-use std::sync::atomic::Ordering;5-use std::time::{Duration, Instant};1+use super::*;62 7-use anyhow::{Context, Result, bail, ensure};8-use iroh::EndpointId;9-use iroh::endpoint::{Connection, SendStream};10-use iroh::protocol::ProtocolHandler;11-use moq_net::{12- Broadcast, BroadcastProducer, Error as MoqError, GroupProducer, Origin, Track, TrackProducer,13-};14-use sigil_protocol::{15- AUDIO_HEADER_LEN, AdaptiveBitrateDecisionV1, AdaptiveBitrateReasonFlagsV1,16- AdaptiveBitrateStateV1, AudioFlags, AudioPacket, AudioPacketHeader, Capability, ClientHello,17- FrameFlags, HostHello, InputAck, InvitationGrants, KeyframeRequestReasonV3,18- MAX_AUDIO_PAYLOAD_LEN, MAX_MEDIA_GROUP_BYTES_V3, MAX_MEDIA_OBJECT_DELIVERY_TIMEOUT_MS,19- MAX_MEDIA_OBJECT_ID_V3, MIN_MEDIA_OBJECT_DELIVERY_TIMEOUT_MS, MOQ_VIDEO_H264_TRACK,20- MOQ_VIDEO_TRACK_PRIORITY, MediaControlRequestV3, MediaFeedbackFlags, MediaFeedbackReportV1,21- MediaFrame, MediaFrameHeader, MediaObjectHeaderV3, MediaObjectV3, encode_media_frame_object,22- media_moq_broadcast_name, read_client_hello, read_input_event, read_media_control_request_v3,23- read_media_feedback_report_v1, write_adaptive_bitrate_decision_v1, write_host_hello,24- write_input_ack, write_media_frame, write_media_object_v3,25-};26-use tracing::{debug, error, info, warn};27-28-use crate::audio::spawn_pipewire_audio;29-use crate::authorization::{AuthorizationPolicy, unix_timestamp_now};30-use crate::clock::SessionClock;31-use crate::config::{GamescopeEncoderBackend, HostConfig, VaapiRateControl, VideoSource};32-use crate::cursor::{PointerPositionTracker, PointerState};33-use crate::input::{InputBackend, InputDisposition};34-use crate::moq_catalog::publish_goq_catalog;35-use crate::source::{36- EncodedFrame, EncodedGop, EncodedSource, EncoderControl,37- spawn_gamescope_pipewire_after_static_preflight, spawn_test_pattern,38-};39-40-const MEDIA_CAPABILITIES: &[Capability] = &[Capability::VideoH264];3+pub(super) const MEDIA_CAPABILITIES: &[Capability] = &[Capability::VideoH264];414 const AUDIO_CAPABILITIES: &[Capability] = &[Capability::AudioOpus];42-const HANDSHAKE_TIMEOUT: Duration = Duration::from_secs(10);5+pub(super) const HANDSHAKE_TIMEOUT: Duration = Duration::from_secs(10);436 const INPUT_ACK_TIMEOUT: Duration = Duration::from_secs(1);447 const REJECTION_DRAIN_TIMEOUT: Duration = Duration::from_secs(1);45-const MAX_PENDING_HANDSHAKES: usize = 4;46-// Allow one frame of ordinary scheduler/write jitter beyond the frame being47-// sent, but never replay a suffix already more than two configured periods old.48-const MAX_MEDIA_REPLAY_AGE_FRAME_PERIODS: u64 = 2;49-const SOURCE_REAP_GRACE_TIMEOUT: Duration = Duration::from_secs(1);50-const ENCODER_CONTROL_COMMIT_TIMEOUT: Duration = Duration::from_secs(2);51-52-mod adaptive;53-mod media_v2;54-mod media_v3;55-mod moq;56-mod session;57-mod startup;58-59-60-pub(crate) use adaptive::MotionResolutionPolicy;61-pub(crate) use adaptive::VideoDimensions;62-use adaptive::serve_media_feedback;63-use media_v2::{media_frame_for_encoded, serve_media, serve_media_v2};64-use media_v3::{65- MediaV3GroupCursor, forward_media_v3_control_requests, new_current_gop_frames, serve_media_v3,66-};67-use moq::{MOQ_REJECT_CODE, serve_authorized_moq, serve_control_moq};68-69-pub use session::SessionRegistry;70-use session::{71- ClaimedMoqAttachment, ForcedIdrCoordinator, ForcedIdrDisposition, MediaV3Telemetry,72- MoqAttachmentWait, SourceTaskGuard,73-};74-use startup::select_gamescope_startup_source;75-76-77-enum MediaReplayDecision {78- Send { discontinuity: bool },79- SkipUntilKeyframe,80- DiscardStaleSuffix { through_sequence: u64 },81-}82-83-84-struct MediaV3ObjectPosition {85- group_id: u64,86- object_id: u32,87- discontinuity: bool,88-}89-90-91-enum MediaV3GroupDecision {92- Send(MediaV3ObjectPosition),93- SkipUntilKeyframe,94- EnterResync,95-}96-97-98-struct MediaReplayCursor {99- last_sequence: Option<u64>,100- waiting_for_keyframe: bool,101- discontinuity_pending: bool,102-}103-104-impl Default for MediaReplayCursor {105- fn default() -> Self {106- Self {107- last_sequence: None,108- waiting_for_keyframe: true,109- discontinuity_pending: false,110- }111- }112-}113-114-impl MediaReplayCursor {115- fn classify(116- &mut self,117- frame: &EncodedFrame,118- replay_through_sequence: u64,119- initial_replay_started_at: Option<Instant>,120- observed_now: Instant,121- maximum_replay_age: Duration,122- ) -> MediaReplayDecision {123- let replay_age = observed_now.saturating_duration_since(frame.observed_at);124- let initial_replay_within_budget = initial_replay_started_at.is_some_and(|started_at| {125- observed_now.saturating_duration_since(started_at) <= maximum_replay_age126- });127- if !initial_replay_within_budget && replay_age > maximum_replay_age {128- self.last_sequence = Some(replay_through_sequence);129- self.waiting_for_keyframe = true;130- self.discontinuity_pending = true;131- return MediaReplayDecision::DiscardStaleSuffix {132- through_sequence: replay_through_sequence,133- };134- }135-136- let sequence_discontinuity = self137- .last_sequence138- .is_some_and(|previous| previous.checked_add(1) != Some(frame.sequence));139- if sequence_discontinuity {140- self.waiting_for_keyframe = true;141- self.discontinuity_pending = true;142- }143- if self.waiting_for_keyframe && !(frame.keyframe && frame.codec_config) {144- return MediaReplayDecision::SkipUntilKeyframe;145- }146- MediaReplayDecision::Send {147- discontinuity: self.discontinuity_pending,148- }149- }150-151- fn commit_sent(&mut self, frame: &EncodedFrame) {152- self.last_sequence = Some(frame.sequence);153- self.waiting_for_keyframe = false;154- self.discontinuity_pending = false;155- }156-157- fn enter_resync_through(&mut self, through_sequence: Option<u64>) {158- if let Some(through_sequence) = through_sequence {159- self.last_sequence = Some(160- self.last_sequence161- .map_or(through_sequence, |last| last.max(through_sequence)),162- );163- }164- self.waiting_for_keyframe = true;165- self.discontinuity_pending = true;166- }167-}168-169-fn maximum_media_replay_age(framerate: u32) -> Duration {170- debug_assert!(framerate > 0);171- let frame_period_nanos = 1_000_000_000_u64.div_ceil(u64::from(framerate.max(1)));172- Duration::from_nanos(frame_period_nanos.saturating_mul(MAX_MEDIA_REPLAY_AGE_FRAME_PERIODS))173-}1748 1759 17610 pub struct MediaHandler {@@ -641,7 +475,7 @@ async fn serve_audio(
641475 session_result642476 }643477 644-async fn send_rejection(478+pub(super) async fn send_rejection(645479 send: &mut iroh::endpoint::SendStream,646480 message: impl Into<String>,647481 ) -> Result<()> {@@ -656,7 +490,10 @@ async fn send_rejection(
656490 Ok(())657491 }658492 659-fn negotiated_capabilities(hello: &ClientHello, supported: &[Capability]) -> Vec<Capability> {493+pub(super) fn negotiated_capabilities(494+ hello: &ClientHello,495+ supported: &[Capability],496+) -> Vec<Capability> {660497 supported661498 .iter()662499 .copied()@@ -737,7 +574,7 @@ fn input_event_type(event: &sigil_protocol::InputEvent) -> &'static str {
737574 }738575 }739576 740-async fn receive_hello<R>(reader: &mut R, required: Capability) -> Result<ClientHello>577+pub(super) async fn receive_hello<R>(reader: &mut R, required: Capability) -> Result<ClientHello>741578 where742579 R: tokio::io::AsyncRead + Unpin,743580 {@@ -762,68 +599,6 @@ where
762599 .context("client closed before hello")763600 }764601 765-pub fn install_panic_hook() {766- std::panic::set_hook(Box::new(|info| {767- error!(%info, "host panic");768- }));769-}770-771-772-fn endpoint(byte: u8) -> EndpointId {773- iroh::SecretKey::from_bytes(&[byte; 32]).public()774-}775-776-777-fn moq_test_config() -> HostConfig {778- HostConfig {779- identity_path: "identity".into(),780- state_path: "state".into(),781- source: VideoSource::TestPattern,782- width: Some(1280),783- height: Some(800),784- framerate: 60,785- codec: "h264".to_owned(),786- input_mode: crate::config::InputMode::Disabled,787- uinput: None,788- ffmpeg_path: "ffmpeg".into(),789- gamescope_pipewire: None,790- audio: None,791- }792-}793-794-795-fn media_v3_encoded_frame(796- sequence: u64,797- keyframe: bool,798- codec_config: bool,799- payload_len: usize,800-) -> EncodedFrame {801- EncodedFrame {802- sequence,803- width: 1_280,804- height: 800,805- capture_timestamp_micros: sequence,806- presentation_timestamp_micros: sequence as i64,807- observed_at: Instant::now(),808- keyframe,809- codec_config,810- discontinuity: false,811- data: Arc::from(vec![sequence as u8; payload_len]),812- }813-}814-815-816-struct DropNotify(Option<tokio::sync::oneshot::Sender<()>>);817-818-819-impl Drop for DropNotify {820- fn drop(&mut self) {821- if let Some(sender) = self.0.take() {822- let _ = sender.send(());823- }824- }825-}826-827602 828603 mod tests {829604 use super::*;@@ -1035,222 +810,4 @@ mod tests {
1035810 (Some(position), Some(false))1036811 );1037812 }1038-1039- 1040- fn initial_current_gop_replay_preserves_complete_startup_snapshot() {1041- let observed_now = Instant::now();1042- let maximum_replay_age = maximum_media_replay_age(60);1043- let old_observation = observed_now - Duration::from_secs(1);1044- let frame = |sequence, keyframe| EncodedFrame {1045- sequence,1046- width: 1_280,1047- height: 800,1048- capture_timestamp_micros: sequence,1049- presentation_timestamp_micros: sequence as i64,1050- observed_at: old_observation,1051- keyframe,1052- codec_config: keyframe,1053- discontinuity: false,1054- data: Arc::from([sequence as u8]),1055- };1056- let mut cursor = MediaReplayCursor::default();1057-1058- let keyframe = frame(10, true);1059- assert_eq!(1060- cursor.classify(1061- &keyframe,1062- 12,1063- Some(observed_now),1064- observed_now,1065- maximum_replay_age,1066- ),1067- MediaReplayDecision::Send {1068- discontinuity: false,1069- }1070- );1071- cursor.commit_sent(&keyframe);1072-1073- let delta = frame(11, false);1074- assert_eq!(1075- cursor.classify(1076- &delta,1077- 12,1078- Some(observed_now),1079- observed_now,1080- maximum_replay_age,1081- ),1082- MediaReplayDecision::Send {1083- discontinuity: false,1084- }1085- );1086- }1087-1088- 1089- fn stalled_initial_current_gop_discards_its_remaining_suffix() {1090- let replay_started_at = Instant::now();1091- let maximum_replay_age = maximum_media_replay_age(60);1092- let old_observation = replay_started_at - Duration::from_secs(1);1093- let frame = |sequence, keyframe| EncodedFrame {1094- sequence,1095- width: 1_280,1096- height: 800,1097- capture_timestamp_micros: sequence,1098- presentation_timestamp_micros: sequence as i64,1099- observed_at: old_observation,1100- keyframe,1101- codec_config: keyframe,1102- discontinuity: false,1103- data: Arc::from([sequence as u8]),1104- };1105- let mut cursor = MediaReplayCursor::default();1106-1107- let keyframe = frame(10, true);1108- assert_eq!(1109- cursor.classify(1110- &keyframe,1111- 12,1112- Some(replay_started_at),1113- replay_started_at,1114- maximum_replay_age,1115- ),1116- MediaReplayDecision::Send {1117- discontinuity: false,1118- }1119- );1120- cursor.commit_sent(&keyframe);1121-1122- let stalled_now = replay_started_at + maximum_replay_age + Duration::from_nanos(1);1123- let delta = frame(11, false);1124- assert_eq!(1125- cursor.classify(1126- &delta,1127- 12,1128- Some(replay_started_at),1129- stalled_now,1130- maximum_replay_age,1131- ),1132- MediaReplayDecision::DiscardStaleSuffix {1133- through_sequence: 12,1134- }1135- );1136- assert_eq!(cursor.last_sequence, Some(12));1137- assert!(cursor.waiting_for_keyframe);1138- assert!(cursor.discontinuity_pending);1139- }1140-1141- 1142- fn fresh_current_gop_suffix_replays_normally() {1143- let observed_now = Instant::now();1144- let maximum_replay_age = maximum_media_replay_age(60);1145- assert_eq!(maximum_replay_age, Duration::from_nanos(33_333_334));1146- let frame = EncodedFrame {1147- sequence: 11,1148- width: 1_280,1149- height: 800,1150- capture_timestamp_micros: 11,1151- presentation_timestamp_micros: 11,1152- observed_at: observed_now - maximum_replay_age,1153- keyframe: false,1154- codec_config: false,1155- discontinuity: false,1156- data: Arc::from([11]),1157- };1158- let mut cursor = MediaReplayCursor {1159- last_sequence: Some(10),1160- waiting_for_keyframe: false,1161- discontinuity_pending: false,1162- };1163-1164- assert_eq!(1165- cursor.classify(&frame, 12, None, observed_now, maximum_replay_age),1166- MediaReplayDecision::Send {1167- discontinuity: false,1168- }1169- );1170- cursor.commit_sent(&frame);1171- assert_eq!(cursor.last_sequence, Some(11));1172- }1173-1174- 1175- fn stale_current_gop_suffix_is_discarded_as_one_bounded_unit() {1176- let observed_now = Instant::now();1177- let maximum_replay_age = maximum_media_replay_age(60);1178- let frame = EncodedFrame {1179- sequence: 11,1180- width: 1_280,1181- height: 800,1182- capture_timestamp_micros: 11,1183- presentation_timestamp_micros: 11,1184- observed_at: observed_now - maximum_replay_age - Duration::from_nanos(1),1185- keyframe: false,1186- codec_config: false,1187- discontinuity: false,1188- data: Arc::from([11]),1189- };1190- let mut cursor = MediaReplayCursor {1191- last_sequence: Some(10),1192- waiting_for_keyframe: false,1193- discontinuity_pending: false,1194- };1195-1196- assert_eq!(1197- cursor.classify(&frame, 13, None, observed_now, maximum_replay_age),1198- MediaReplayDecision::DiscardStaleSuffix {1199- through_sequence: 13,1200- }1201- );1202- assert_eq!(cursor.last_sequence, Some(13));1203- assert!(cursor.waiting_for_keyframe);1204- assert!(cursor.discontinuity_pending);1205- }1206-1207- 1208- fn stale_suffix_recovers_only_on_idr_marked_discontinuity() {1209- let observed_now = Instant::now();1210- let maximum_replay_age = maximum_media_replay_age(60);1211- let frame = |sequence, keyframe| EncodedFrame {1212- sequence,1213- width: 1_280,1214- height: 800,1215- capture_timestamp_micros: sequence,1216- presentation_timestamp_micros: sequence as i64,1217- observed_at: observed_now,1218- keyframe,1219- codec_config: keyframe,1220- discontinuity: false,1221- data: Arc::from([sequence as u8]),1222- };1223- let mut cursor = MediaReplayCursor {1224- last_sequence: Some(13),1225- waiting_for_keyframe: true,1226- discontinuity_pending: true,1227- };1228-1229- let delta = frame(14, false);1230- assert_eq!(1231- cursor.classify(&delta, 14, None, observed_now, maximum_replay_age),1232- MediaReplayDecision::SkipUntilKeyframe1233- );1234- assert_eq!(cursor.last_sequence, Some(13));1235-1236- let idr = frame(15, true);1237- assert_eq!(1238- cursor.classify(&idr, 15, None, observed_now, maximum_replay_age),1239- MediaReplayDecision::Send {1240- discontinuity: true,1241- }1242- );1243- cursor.commit_sent(&idr);1244- assert_eq!(cursor.last_sequence, Some(15));1245- assert!(!cursor.waiting_for_keyframe);1246- assert!(!cursor.discontinuity_pending);1247-1248- let next_delta = frame(16, false);1249- assert_eq!(1250- cursor.classify(&next_delta, 16, None, observed_now, maximum_replay_age,),1251- MediaReplayDecision::Send {1252- discontinuity: false,1253- }1254- );1255- }1256813 }@@ -0,0 +1,460 @@
1+use std::future::Future;2+use std::pin::Pin;3+use std::sync::Arc;4+use std::sync::atomic::Ordering;5+use std::time::{Duration, Instant};6+7+use anyhow::{Context, Result, bail, ensure};8+use iroh::EndpointId;9+use iroh::endpoint::{Connection, SendStream};10+use iroh::protocol::ProtocolHandler;11+use moq_net::{12+ Broadcast, BroadcastProducer, Error as MoqError, GroupProducer, Origin, Track, TrackProducer,13+};14+use sigil_protocol::{15+ AUDIO_HEADER_LEN, AdaptiveBitrateDecisionV1, AdaptiveBitrateReasonFlagsV1,16+ AdaptiveBitrateStateV1, AudioFlags, AudioPacket, AudioPacketHeader, Capability, ClientHello,17+ FrameFlags, HostHello, InputAck, InvitationGrants, KeyframeRequestReasonV3,18+ MAX_AUDIO_PAYLOAD_LEN, MAX_MEDIA_GROUP_BYTES_V3, MAX_MEDIA_OBJECT_DELIVERY_TIMEOUT_MS,19+ MAX_MEDIA_OBJECT_ID_V3, MIN_MEDIA_OBJECT_DELIVERY_TIMEOUT_MS, MOQ_VIDEO_H264_TRACK,20+ MOQ_VIDEO_TRACK_PRIORITY, MediaControlRequestV3, MediaFeedbackFlags, MediaFeedbackReportV1,21+ MediaFrame, MediaFrameHeader, MediaObjectHeaderV3, MediaObjectV3, encode_media_frame_object,22+ media_moq_broadcast_name, read_client_hello, read_input_event, read_media_control_request_v3,23+ read_media_feedback_report_v1, write_adaptive_bitrate_decision_v1, write_host_hello,24+ write_input_ack, write_media_frame, write_media_object_v3,25+};26+use tracing::{debug, error, info, warn};27+28+use crate::audio::spawn_pipewire_audio;29+use crate::authorization::{AuthorizationPolicy, unix_timestamp_now};30+use crate::clock::SessionClock;31+use crate::config::{GamescopeEncoderBackend, HostConfig, VaapiRateControl, VideoSource};32+use crate::cursor::{PointerPositionTracker, PointerState};33+use crate::input::{InputBackend, InputDisposition};34+use crate::moq_catalog::publish_goq_catalog;35+use crate::source::{36+ EncodedFrame, EncodedGop, EncodedSource, EncoderControl,37+ spawn_gamescope_pipewire_after_static_preflight, spawn_test_pattern,38+};39+40+// Allow one frame of ordinary scheduler/write jitter beyond the frame being41+// sent, but never replay a suffix already more than two configured periods old.42+const MAX_MEDIA_REPLAY_AGE_FRAME_PERIODS: u64 = 2;43+const SOURCE_REAP_GRACE_TIMEOUT: Duration = Duration::from_secs(1);44+const ENCODER_CONTROL_COMMIT_TIMEOUT: Duration = Duration::from_secs(2);45+46+mod adaptive;47+mod handlers;48+mod media_v2;49+mod media_v3;50+mod moq;51+mod session;52+mod startup;53+54+55+pub(crate) use adaptive::MotionResolutionPolicy;56+pub(crate) use adaptive::VideoDimensions;57+use adaptive::serve_media_feedback;58+pub use handlers::{59+ AudioHandler, AuthorizedMoqHandler, ControlHandler, InputHandler, MediaFeedbackHandler,60+ MediaHandler, MediaV2Handler, MediaV3Handler,61+};62+use handlers::{63+ HANDSHAKE_TIMEOUT, MEDIA_CAPABILITIES, negotiated_capabilities, receive_hello, send_rejection,64+};65+use media_v2::{media_frame_for_encoded, serve_media, serve_media_v2};66+use media_v3::{67+ MediaV3GroupCursor, forward_media_v3_control_requests, new_current_gop_frames, serve_media_v3,68+};69+use moq::{MOQ_REJECT_CODE, serve_authorized_moq, serve_control_moq};70+71+pub use session::SessionRegistry;72+use session::{73+ ClaimedMoqAttachment, ForcedIdrCoordinator, ForcedIdrDisposition, MediaV3Telemetry,74+ MoqAttachmentWait, SourceTaskGuard,75+};76+use startup::select_gamescope_startup_source;77+78+79+enum MediaReplayDecision {80+ Send { discontinuity: bool },81+ SkipUntilKeyframe,82+ DiscardStaleSuffix { through_sequence: u64 },83+}84+85+86+struct MediaV3ObjectPosition {87+ group_id: u64,88+ object_id: u32,89+ discontinuity: bool,90+}91+92+93+enum MediaV3GroupDecision {94+ Send(MediaV3ObjectPosition),95+ SkipUntilKeyframe,96+ EnterResync,97+}98+99+100+struct MediaReplayCursor {101+ last_sequence: Option<u64>,102+ waiting_for_keyframe: bool,103+ discontinuity_pending: bool,104+}105+106+impl Default for MediaReplayCursor {107+ fn default() -> Self {108+ Self {109+ last_sequence: None,110+ waiting_for_keyframe: true,111+ discontinuity_pending: false,112+ }113+ }114+}115+116+impl MediaReplayCursor {117+ fn classify(118+ &mut self,119+ frame: &EncodedFrame,120+ replay_through_sequence: u64,121+ initial_replay_started_at: Option<Instant>,122+ observed_now: Instant,123+ maximum_replay_age: Duration,124+ ) -> MediaReplayDecision {125+ let replay_age = observed_now.saturating_duration_since(frame.observed_at);126+ let initial_replay_within_budget = initial_replay_started_at.is_some_and(|started_at| {127+ observed_now.saturating_duration_since(started_at) <= maximum_replay_age128+ });129+ if !initial_replay_within_budget && replay_age > maximum_replay_age {130+ self.last_sequence = Some(replay_through_sequence);131+ self.waiting_for_keyframe = true;132+ self.discontinuity_pending = true;133+ return MediaReplayDecision::DiscardStaleSuffix {134+ through_sequence: replay_through_sequence,135+ };136+ }137+138+ let sequence_discontinuity = self139+ .last_sequence140+ .is_some_and(|previous| previous.checked_add(1) != Some(frame.sequence));141+ if sequence_discontinuity {142+ self.waiting_for_keyframe = true;143+ self.discontinuity_pending = true;144+ }145+ if self.waiting_for_keyframe && !(frame.keyframe && frame.codec_config) {146+ return MediaReplayDecision::SkipUntilKeyframe;147+ }148+ MediaReplayDecision::Send {149+ discontinuity: self.discontinuity_pending,150+ }151+ }152+153+ fn commit_sent(&mut self, frame: &EncodedFrame) {154+ self.last_sequence = Some(frame.sequence);155+ self.waiting_for_keyframe = false;156+ self.discontinuity_pending = false;157+ }158+159+ fn enter_resync_through(&mut self, through_sequence: Option<u64>) {160+ if let Some(through_sequence) = through_sequence {161+ self.last_sequence = Some(162+ self.last_sequence163+ .map_or(through_sequence, |last| last.max(through_sequence)),164+ );165+ }166+ self.waiting_for_keyframe = true;167+ self.discontinuity_pending = true;168+ }169+}170+171+fn maximum_media_replay_age(framerate: u32) -> Duration {172+ debug_assert!(framerate > 0);173+ let frame_period_nanos = 1_000_000_000_u64.div_ceil(u64::from(framerate.max(1)));174+ Duration::from_nanos(frame_period_nanos.saturating_mul(MAX_MEDIA_REPLAY_AGE_FRAME_PERIODS))175+}176+177+pub fn install_panic_hook() {178+ std::panic::set_hook(Box::new(|info| {179+ error!(%info, "host panic");180+ }));181+}182+183+184+fn endpoint(byte: u8) -> EndpointId {185+ iroh::SecretKey::from_bytes(&[byte; 32]).public()186+}187+188+189+fn moq_test_config() -> HostConfig {190+ HostConfig {191+ identity_path: "identity".into(),192+ state_path: "state".into(),193+ source: VideoSource::TestPattern,194+ width: Some(1280),195+ height: Some(800),196+ framerate: 60,197+ codec: "h264".to_owned(),198+ input_mode: crate::config::InputMode::Disabled,199+ uinput: None,200+ ffmpeg_path: "ffmpeg".into(),201+ gamescope_pipewire: None,202+ audio: None,203+ }204+}205+206+207+fn media_v3_encoded_frame(208+ sequence: u64,209+ keyframe: bool,210+ codec_config: bool,211+ payload_len: usize,212+) -> EncodedFrame {213+ EncodedFrame {214+ sequence,215+ width: 1_280,216+ height: 800,217+ capture_timestamp_micros: sequence,218+ presentation_timestamp_micros: sequence as i64,219+ observed_at: Instant::now(),220+ keyframe,221+ codec_config,222+ discontinuity: false,223+ data: Arc::from(vec![sequence as u8; payload_len]),224+ }225+}226+227+228+struct DropNotify(Option<tokio::sync::oneshot::Sender<()>>);229+230+231+impl Drop for DropNotify {232+ fn drop(&mut self) {233+ if let Some(sender) = self.0.take() {234+ let _ = sender.send(());235+ }236+ }237+}238+239+240+mod tests {241+ use super::*;242+243+ 244+ fn initial_current_gop_replay_preserves_complete_startup_snapshot() {245+ let observed_now = Instant::now();246+ let maximum_replay_age = maximum_media_replay_age(60);247+ let old_observation = observed_now - Duration::from_secs(1);248+ let frame = |sequence, keyframe| EncodedFrame {249+ sequence,250+ width: 1_280,251+ height: 800,252+ capture_timestamp_micros: sequence,253+ presentation_timestamp_micros: sequence as i64,254+ observed_at: old_observation,255+ keyframe,256+ codec_config: keyframe,257+ discontinuity: false,258+ data: Arc::from([sequence as u8]),259+ };260+ let mut cursor = MediaReplayCursor::default();261+262+ let keyframe = frame(10, true);263+ assert_eq!(264+ cursor.classify(265+ &keyframe,266+ 12,267+ Some(observed_now),268+ observed_now,269+ maximum_replay_age,270+ ),271+ MediaReplayDecision::Send {272+ discontinuity: false,273+ }274+ );275+ cursor.commit_sent(&keyframe);276+277+ let delta = frame(11, false);278+ assert_eq!(279+ cursor.classify(280+ &delta,281+ 12,282+ Some(observed_now),283+ observed_now,284+ maximum_replay_age,285+ ),286+ MediaReplayDecision::Send {287+ discontinuity: false,288+ }289+ );290+ }291+292+ 293+ fn stalled_initial_current_gop_discards_its_remaining_suffix() {294+ let replay_started_at = Instant::now();295+ let maximum_replay_age = maximum_media_replay_age(60);296+ let old_observation = replay_started_at - Duration::from_secs(1);297+ let frame = |sequence, keyframe| EncodedFrame {298+ sequence,299+ width: 1_280,300+ height: 800,301+ capture_timestamp_micros: sequence,302+ presentation_timestamp_micros: sequence as i64,303+ observed_at: old_observation,304+ keyframe,305+ codec_config: keyframe,306+ discontinuity: false,307+ data: Arc::from([sequence as u8]),308+ };309+ let mut cursor = MediaReplayCursor::default();310+311+ let keyframe = frame(10, true);312+ assert_eq!(313+ cursor.classify(314+ &keyframe,315+ 12,316+ Some(replay_started_at),317+ replay_started_at,318+ maximum_replay_age,319+ ),320+ MediaReplayDecision::Send {321+ discontinuity: false,322+ }323+ );324+ cursor.commit_sent(&keyframe);325+326+ let stalled_now = replay_started_at + maximum_replay_age + Duration::from_nanos(1);327+ let delta = frame(11, false);328+ assert_eq!(329+ cursor.classify(330+ &delta,331+ 12,332+ Some(replay_started_at),333+ stalled_now,334+ maximum_replay_age,335+ ),336+ MediaReplayDecision::DiscardStaleSuffix {337+ through_sequence: 12,338+ }339+ );340+ assert_eq!(cursor.last_sequence, Some(12));341+ assert!(cursor.waiting_for_keyframe);342+ assert!(cursor.discontinuity_pending);343+ }344+345+ 346+ fn fresh_current_gop_suffix_replays_normally() {347+ let observed_now = Instant::now();348+ let maximum_replay_age = maximum_media_replay_age(60);349+ assert_eq!(maximum_replay_age, Duration::from_nanos(33_333_334));350+ let frame = EncodedFrame {351+ sequence: 11,352+ width: 1_280,353+ height: 800,354+ capture_timestamp_micros: 11,355+ presentation_timestamp_micros: 11,356+ observed_at: observed_now - maximum_replay_age,357+ keyframe: false,358+ codec_config: false,359+ discontinuity: false,360+ data: Arc::from([11]),361+ };362+ let mut cursor = MediaReplayCursor {363+ last_sequence: Some(10),364+ waiting_for_keyframe: false,365+ discontinuity_pending: false,366+ };367+368+ assert_eq!(369+ cursor.classify(&frame, 12, None, observed_now, maximum_replay_age),370+ MediaReplayDecision::Send {371+ discontinuity: false,372+ }373+ );374+ cursor.commit_sent(&frame);375+ assert_eq!(cursor.last_sequence, Some(11));376+ }377+378+ 379+ fn stale_current_gop_suffix_is_discarded_as_one_bounded_unit() {380+ let observed_now = Instant::now();381+ let maximum_replay_age = maximum_media_replay_age(60);382+ let frame = EncodedFrame {383+ sequence: 11,384+ width: 1_280,385+ height: 800,386+ capture_timestamp_micros: 11,387+ presentation_timestamp_micros: 11,388+ observed_at: observed_now - maximum_replay_age - Duration::from_nanos(1),389+ keyframe: false,390+ codec_config: false,391+ discontinuity: false,392+ data: Arc::from([11]),393+ };394+ let mut cursor = MediaReplayCursor {395+ last_sequence: Some(10),396+ waiting_for_keyframe: false,397+ discontinuity_pending: false,398+ };399+400+ assert_eq!(401+ cursor.classify(&frame, 13, None, observed_now, maximum_replay_age),402+ MediaReplayDecision::DiscardStaleSuffix {403+ through_sequence: 13,404+ }405+ );406+ assert_eq!(cursor.last_sequence, Some(13));407+ assert!(cursor.waiting_for_keyframe);408+ assert!(cursor.discontinuity_pending);409+ }410+411+ 412+ fn stale_suffix_recovers_only_on_idr_marked_discontinuity() {413+ let observed_now = Instant::now();414+ let maximum_replay_age = maximum_media_replay_age(60);415+ let frame = |sequence, keyframe| EncodedFrame {416+ sequence,417+ width: 1_280,418+ height: 800,419+ capture_timestamp_micros: sequence,420+ presentation_timestamp_micros: sequence as i64,421+ observed_at: observed_now,422+ keyframe,423+ codec_config: keyframe,424+ discontinuity: false,425+ data: Arc::from([sequence as u8]),426+ };427+ let mut cursor = MediaReplayCursor {428+ last_sequence: Some(13),429+ waiting_for_keyframe: true,430+ discontinuity_pending: true,431+ };432+433+ let delta = frame(14, false);434+ assert_eq!(435+ cursor.classify(&delta, 14, None, observed_now, maximum_replay_age),436+ MediaReplayDecision::SkipUntilKeyframe437+ );438+ assert_eq!(cursor.last_sequence, Some(13));439+440+ let idr = frame(15, true);441+ assert_eq!(442+ cursor.classify(&idr, 15, None, observed_now, maximum_replay_age),443+ MediaReplayDecision::Send {444+ discontinuity: true,445+ }446+ );447+ cursor.commit_sent(&idr);448+ assert_eq!(cursor.last_sequence, Some(15));449+ assert!(!cursor.waiting_for_keyframe);450+ assert!(!cursor.discontinuity_pending);451+452+ let next_delta = frame(16, false);453+ assert_eq!(454+ cursor.classify(&next_delta, 16, None, observed_now, maximum_replay_age,),455+ MediaReplayDecision::Send {456+ discontinuity: false,457+ }458+ );459+ }460+}@@ -9,10 +9,12 @@ use moq_net::BroadcastConsumer;
99 use sigil_protocol::{InvitationGrants, KeyframeRequestReasonV3};1010 use tracing::{debug, warn};1111 12-use super::{ENCODER_CONTROL_COMMIT_TIMEOUT, MAX_PENDING_HANDSHAKES, VideoDimensions};12+use super::{ENCODER_CONTROL_COMMIT_TIMEOUT, VideoDimensions};1313 use crate::clock::SessionClock;1414 use crate::source::EncoderControl;1515 16+const MAX_PENDING_HANDSHAKES: usize = 4;17+1618 1719 pub struct SessionRegistry {1820 active: Mutex<Option<ActiveSession>>,