1use std::future::Future;2use std::sync::Arc;3use std::sync::atomic::Ordering;4use std::time::{Duration, Instant};56use anyhow::{Context, Result, bail, ensure};7use iroh::EndpointId;8use iroh::endpoint::{Connection, SendStream};9use iroh::protocol::ProtocolHandler;10use moq_net::{11 Broadcast, BroadcastProducer, Error as MoqError, GroupProducer, Origin, Track, TrackProducer,12};13use sigil_protocol::{14 AUDIO_HEADER_LEN, AdaptiveBitrateDecisionV1, AdaptiveBitrateReasonFlagsV1,15 AdaptiveBitrateStateV1, AudioFlags, AudioPacket, AudioPacketHeader, Capability,16 ClientControlEnvelopeV2, ClientHello, FocusCommandResultV2, FocusTransitionReasonV2,17 FrameFlags, HostHello, HostHelloV2, InputAck, InputAckV2, InputClientHelloV2, InputEventV2,18 InputHostHelloV2, InvitationGrants, KeyframeRequestReasonV3, MAX_AUDIO_PAYLOAD_LEN,19 MAX_MEDIA_GROUP_BYTES_V3, MAX_MEDIA_OBJECT_DELIVERY_TIMEOUT_MS, MAX_MEDIA_OBJECT_ID_V3,20 MIN_MEDIA_OBJECT_DELIVERY_TIMEOUT_MS, MOQ_VIDEO_H264_TRACK, MOQ_VIDEO_TRACK_PRIORITY,21 MediaControlRequestV3, MediaFeedbackFlags, MediaFeedbackReportV1, MediaFrame, MediaFrameHeader,22 MediaGenerationSigningKey, MediaObjectCoordinates, MediaObjectHeaderV3, MediaObjectV3,23 MediaTrack, ServerControlEnvelopeV2, SignedSubscriptionCapability, SubscriptionClaims,24 SubscriptionTracks, encode_media_frame_object, media_moq_broadcast_name,25 read_client_control_v2, read_client_hello, read_client_hello_v2, read_input_client_hello_v2,26 read_input_event, read_input_event_v2, read_media_control_request_v3,27 read_media_feedback_report_v1, write_adaptive_bitrate_decision_v1, write_host_hello,28 write_host_hello_v2, write_input_ack, write_input_ack_v2, write_input_host_hello_v2,29 write_media_object_v3, write_server_control_v2,30};31use tracing::{debug, error, info, warn};3233use crate::audio::spawn_pipewire_audio;34use crate::authorization::{AuthorizationPolicy, unix_timestamp_now};35use crate::clock::SessionClock;36use crate::config::{GamescopeEncoderBackend, HostConfig, VaapiRateControl, VideoSource};37use crate::cursor::{PointerPositionTracker, PointerState};38use crate::input::{InputBackend, InputDisposition};39use crate::moq_catalog::publish_goq_catalog;40use crate::source::{41 EncodedFrame, EncodedGop, EncodedSource, EncoderControl,42 spawn_gamescope_pipewire_after_static_preflight, spawn_test_pattern,43};4445// Allow one frame of ordinary scheduler/write jitter beyond the frame being46// sent, but never replay a suffix already more than two configured periods old.47const MAX_MEDIA_REPLAY_AGE_FRAME_PERIODS: u64 = 2;48const SOURCE_REAP_GRACE_TIMEOUT: Duration = Duration::from_secs(1);49const ENCODER_CONTROL_COMMIT_TIMEOUT: Duration = Duration::from_secs(2);5051mod adaptive;52mod focus;53mod generation;54mod handlers;55mod media_v3;56mod moq;57mod session;58mod startup;5960pub(crate) use adaptive::VideoDimensions;61use adaptive::serve_media_feedback;62pub(crate) use generation::MediaGenerationManager;63pub use handlers::{64 AudioHandler, AuthorizedMoqHandler, ControlHandler, ControlV2Handler, InputHandler,65 InputOperations, InputV2Handler, MediaFeedbackHandler, MediaV3Handler,66};67use handlers::{68 HANDSHAKE_TIMEOUT, MEDIA_CAPABILITIES, negotiated_capabilities, receive_hello, send_rejection,69 send_rejection_v2,70};71use media_v3::{72 MediaV3GroupCursor, forward_media_v3_control_requests, new_current_gop_frames, serve_media_v3,73};74use moq::{MOQ_REJECT_CODE, serve_authorized_moq, serve_control_moq, serve_control_moq_v2};7576pub use session::SessionRegistry;77pub(crate) use session::SessionRuntimeStatus;78use session::{79 ClaimedMoqAttachment, ForcedIdrCoordinator, ForcedIdrDisposition, MediaV3Telemetry,80 MoqAttachmentWait, SourceTaskGuard,81};82use startup::select_gamescope_startup_source;838485enum MediaReplayDecision {86 Send { discontinuity: bool },87 SkipUntilKeyframe,88 DiscardStaleSuffix { through_sequence: u64 },89}909192struct MediaV3ObjectPosition {93 group_id: u64,94 object_id: u32,95 discontinuity: bool,96}979899enum MediaV3GroupDecision {100 Send(MediaV3ObjectPosition),101 SkipUntilKeyframe,102 EnterResync,103}104105106struct MediaReplayCursor {107 last_sequence: Option<u64>,108 waiting_for_keyframe: bool,109 discontinuity_pending: bool,110}111112impl Default for MediaReplayCursor {113 fn default() -> Self {114 Self {115 last_sequence: None,116 waiting_for_keyframe: true,117 discontinuity_pending: false,118 }119 }120}121122impl MediaReplayCursor {123 fn classify(124 &mut self,125 frame: &EncodedFrame,126 replay_through_sequence: u64,127 initial_replay_started_at: Option<Instant>,128 observed_now: Instant,129 maximum_replay_age: Duration,130 ) -> MediaReplayDecision {131 let replay_age = observed_now.saturating_duration_since(frame.observed_at);132 let initial_replay_within_budget = initial_replay_started_at.is_some_and(|started_at| {133 observed_now.saturating_duration_since(started_at) <= maximum_replay_age134 });135 if !initial_replay_within_budget && replay_age > maximum_replay_age {136 self.last_sequence = Some(replay_through_sequence);137 self.waiting_for_keyframe = true;138 self.discontinuity_pending = true;139 return MediaReplayDecision::DiscardStaleSuffix {140 through_sequence: replay_through_sequence,141 };142 }143144 let sequence_discontinuity = self145 .last_sequence146 .is_some_and(|previous| previous.checked_add(1) != Some(frame.sequence));147 if sequence_discontinuity {148 self.waiting_for_keyframe = true;149 self.discontinuity_pending = true;150 }151 if self.waiting_for_keyframe && !(frame.keyframe && frame.codec_config) {152 return MediaReplayDecision::SkipUntilKeyframe;153 }154 MediaReplayDecision::Send {155 discontinuity: self.discontinuity_pending,156 }157 }158159 fn commit_sent(&mut self, frame: &EncodedFrame) {160 self.last_sequence = Some(frame.sequence);161 self.waiting_for_keyframe = false;162 self.discontinuity_pending = false;163 }164165 fn enter_resync_through(&mut self, through_sequence: Option<u64>) {166 if let Some(through_sequence) = through_sequence {167 self.last_sequence = Some(168 self.last_sequence169 .map_or(through_sequence, |last| last.max(through_sequence)),170 );171 }172 self.waiting_for_keyframe = true;173 self.discontinuity_pending = true;174 }175}176177fn maximum_media_replay_age(framerate: u32) -> Duration {178 debug_assert!(framerate > 0);179 let frame_period_nanos = 1_000_000_000_u64.div_ceil(u64::from(framerate.max(1)));180 Duration::from_nanos(frame_period_nanos.saturating_mul(MAX_MEDIA_REPLAY_AGE_FRAME_PERIODS))181}182183pub fn install_panic_hook() {184 std::panic::set_hook(Box::new(|info| {185 error!(%info, "host panic");186 }));187}188189190fn endpoint(byte: u8) -> EndpointId {191 iroh::SecretKey::from_bytes(&[byte; 32]).public()192}193194195pub(crate) fn moq_test_config() -> HostConfig {196 HostConfig {197 identity_path: "identity".into(),198 state_path: "state".into(),199 max_viewers: crate::config::DEFAULT_MAX_VIEWERS,200 focus_owner: None,201 source: VideoSource::TestPattern,202 width: Some(1280),203 height: Some(800),204 framerate: 60,205 codec: "h264".to_owned(),206 input_mode: crate::config::InputMode::Disabled,207 uinput: None,208 ffmpeg_path: "ffmpeg".into(),209 gamescope_pipewire: None,210 audio: None,211 }212}213214215fn media_v3_encoded_frame(216 sequence: u64,217 keyframe: bool,218 codec_config: bool,219 payload_len: usize,220) -> EncodedFrame {221 EncodedFrame {222 sequence,223 width: 1_280,224 height: 800,225 capture_timestamp_micros: sequence,226 presentation_timestamp_micros: sequence as i64,227 observed_at: Instant::now(),228 keyframe,229 codec_config,230 discontinuity: false,231 data: Arc::from(vec![sequence as u8; payload_len]),232 }233}234235236struct DropNotify(Option<tokio::sync::oneshot::Sender<()>>);237238239impl Drop for DropNotify {240 fn drop(&mut self) {241 if let Some(sender) = self.0.take() {242 let _ = sender.send(());243 }244 }245}246247248mod tests {249 use super::*;250251 252 fn initial_current_gop_replay_preserves_complete_startup_snapshot() {253 let observed_now = Instant::now();254 let maximum_replay_age = maximum_media_replay_age(60);255 let old_observation = observed_now - Duration::from_secs(1);256 let frame = |sequence, keyframe| EncodedFrame {257 sequence,258 width: 1_280,259 height: 800,260 capture_timestamp_micros: sequence,261 presentation_timestamp_micros: sequence as i64,262 observed_at: old_observation,263 keyframe,264 codec_config: keyframe,265 discontinuity: false,266 data: Arc::from([sequence as u8]),267 };268 let mut cursor = MediaReplayCursor::default();269270 let keyframe = frame(10, true);271 assert_eq!(272 cursor.classify(273 &keyframe,274 12,275 Some(observed_now),276 observed_now,277 maximum_replay_age,278 ),279 MediaReplayDecision::Send {280 discontinuity: false,281 }282 );283 cursor.commit_sent(&keyframe);284285 let delta = frame(11, false);286 assert_eq!(287 cursor.classify(288 &delta,289 12,290 Some(observed_now),291 observed_now,292 maximum_replay_age,293 ),294 MediaReplayDecision::Send {295 discontinuity: false,296 }297 );298 }299300 301 fn stalled_initial_current_gop_discards_its_remaining_suffix() {302 let replay_started_at = Instant::now();303 let maximum_replay_age = maximum_media_replay_age(60);304 let old_observation = replay_started_at - Duration::from_secs(1);305 let frame = |sequence, keyframe| EncodedFrame {306 sequence,307 width: 1_280,308 height: 800,309 capture_timestamp_micros: sequence,310 presentation_timestamp_micros: sequence as i64,311 observed_at: old_observation,312 keyframe,313 codec_config: keyframe,314 discontinuity: false,315 data: Arc::from([sequence as u8]),316 };317 let mut cursor = MediaReplayCursor::default();318319 let keyframe = frame(10, true);320 assert_eq!(321 cursor.classify(322 &keyframe,323 12,324 Some(replay_started_at),325 replay_started_at,326 maximum_replay_age,327 ),328 MediaReplayDecision::Send {329 discontinuity: false,330 }331 );332 cursor.commit_sent(&keyframe);333334 let stalled_now = replay_started_at + maximum_replay_age + Duration::from_nanos(1);335 let delta = frame(11, false);336 assert_eq!(337 cursor.classify(338 &delta,339 12,340 Some(replay_started_at),341 stalled_now,342 maximum_replay_age,343 ),344 MediaReplayDecision::DiscardStaleSuffix {345 through_sequence: 12,346 }347 );348 assert_eq!(cursor.last_sequence, Some(12));349 assert!(cursor.waiting_for_keyframe);350 assert!(cursor.discontinuity_pending);351 }352353 354 fn fresh_current_gop_suffix_replays_normally() {355 let observed_now = Instant::now();356 let maximum_replay_age = maximum_media_replay_age(60);357 assert_eq!(maximum_replay_age, Duration::from_nanos(33_333_334));358 let frame = EncodedFrame {359 sequence: 11,360 width: 1_280,361 height: 800,362 capture_timestamp_micros: 11,363 presentation_timestamp_micros: 11,364 observed_at: observed_now - maximum_replay_age,365 keyframe: false,366 codec_config: false,367 discontinuity: false,368 data: Arc::from([11]),369 };370 let mut cursor = MediaReplayCursor {371 last_sequence: Some(10),372 waiting_for_keyframe: false,373 discontinuity_pending: false,374 };375376 assert_eq!(377 cursor.classify(&frame, 12, None, observed_now, maximum_replay_age),378 MediaReplayDecision::Send {379 discontinuity: false,380 }381 );382 cursor.commit_sent(&frame);383 assert_eq!(cursor.last_sequence, Some(11));384 }385386 387 fn stale_current_gop_suffix_is_discarded_as_one_bounded_unit() {388 let observed_now = Instant::now();389 let maximum_replay_age = maximum_media_replay_age(60);390 let frame = EncodedFrame {391 sequence: 11,392 width: 1_280,393 height: 800,394 capture_timestamp_micros: 11,395 presentation_timestamp_micros: 11,396 observed_at: observed_now - maximum_replay_age - Duration::from_nanos(1),397 keyframe: false,398 codec_config: false,399 discontinuity: false,400 data: Arc::from([11]),401 };402 let mut cursor = MediaReplayCursor {403 last_sequence: Some(10),404 waiting_for_keyframe: false,405 discontinuity_pending: false,406 };407408 assert_eq!(409 cursor.classify(&frame, 13, None, observed_now, maximum_replay_age),410 MediaReplayDecision::DiscardStaleSuffix {411 through_sequence: 13,412 }413 );414 assert_eq!(cursor.last_sequence, Some(13));415 assert!(cursor.waiting_for_keyframe);416 assert!(cursor.discontinuity_pending);417 }418419 420 fn stale_suffix_recovers_only_on_idr_marked_discontinuity() {421 let observed_now = Instant::now();422 let maximum_replay_age = maximum_media_replay_age(60);423 let frame = |sequence, keyframe| EncodedFrame {424 sequence,425 width: 1_280,426 height: 800,427 capture_timestamp_micros: sequence,428 presentation_timestamp_micros: sequence as i64,429 observed_at: observed_now,430 keyframe,431 codec_config: keyframe,432 discontinuity: false,433 data: Arc::from([sequence as u8]),434 };435 let mut cursor = MediaReplayCursor {436 last_sequence: Some(13),437 waiting_for_keyframe: true,438 discontinuity_pending: true,439 };440441 let delta = frame(14, false);442 assert_eq!(443 cursor.classify(&delta, 14, None, observed_now, maximum_replay_age),444 MediaReplayDecision::SkipUntilKeyframe445 );446 assert_eq!(cursor.last_sequence, Some(13));447448 let idr = frame(15, true);449 assert_eq!(450 cursor.classify(&idr, 15, None, observed_now, maximum_replay_age),451 MediaReplayDecision::Send {452 discontinuity: true,453 }454 );455 cursor.commit_sent(&idr);456 assert_eq!(cursor.last_sequence, Some(15));457 assert!(!cursor.waiting_for_keyframe);458 assert!(!cursor.discontinuity_pending);459460 let next_delta = frame(16, false);461 assert_eq!(462 cursor.classify(&next_delta, 16, None, observed_now, maximum_replay_age,),463 MediaReplayDecision::Send {464 discontinuity: false,465 }466 );467 }468}