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, ClientHello,16 FrameFlags, HostHello, InputAck, InvitationGrants, KeyframeRequestReasonV3,17 MAX_AUDIO_PAYLOAD_LEN, MAX_MEDIA_GROUP_BYTES_V3, MAX_MEDIA_OBJECT_DELIVERY_TIMEOUT_MS,18 MAX_MEDIA_OBJECT_ID_V3, MIN_MEDIA_OBJECT_DELIVERY_TIMEOUT_MS, MOQ_VIDEO_H264_TRACK,19 MOQ_VIDEO_TRACK_PRIORITY, MediaControlRequestV3, MediaFeedbackFlags, MediaFeedbackReportV1,20 MediaFrame, MediaFrameHeader, MediaObjectHeaderV3, MediaObjectV3, encode_media_frame_object,21 media_moq_broadcast_name, read_client_hello, read_input_event, read_media_control_request_v3,22 read_media_feedback_report_v1, write_adaptive_bitrate_decision_v1, write_host_hello,23 write_input_ack, write_media_object_v3,24};25use tracing::{debug, error, info, warn};2627use crate::audio::spawn_pipewire_audio;28use crate::authorization::{AuthorizationPolicy, unix_timestamp_now};29use crate::clock::SessionClock;30use crate::config::{GamescopeEncoderBackend, HostConfig, VaapiRateControl, VideoSource};31use crate::cursor::{PointerPositionTracker, PointerState};32use crate::input::{InputBackend, InputDisposition};33use crate::moq_catalog::publish_goq_catalog;34use crate::source::{35 EncodedFrame, EncodedGop, EncodedSource, EncoderControl,36 spawn_gamescope_pipewire_after_static_preflight, spawn_test_pattern,37};3839// Allow one frame of ordinary scheduler/write jitter beyond the frame being40// sent, but never replay a suffix already more than two configured periods old.41const MAX_MEDIA_REPLAY_AGE_FRAME_PERIODS: u64 = 2;42const SOURCE_REAP_GRACE_TIMEOUT: Duration = Duration::from_secs(1);43const ENCODER_CONTROL_COMMIT_TIMEOUT: Duration = Duration::from_secs(2);4445mod adaptive;46mod handlers;47mod media_v3;48mod moq;49mod session;50mod startup;5152pub(crate) use adaptive::VideoDimensions;53use adaptive::serve_media_feedback;54pub use handlers::{55 AudioHandler, AuthorizedMoqHandler, ControlHandler, InputHandler, MediaFeedbackHandler,56 MediaV3Handler,57};58use handlers::{59 HANDSHAKE_TIMEOUT, MEDIA_CAPABILITIES, negotiated_capabilities, receive_hello, send_rejection,60};61use media_v3::{62 MediaV3GroupCursor, forward_media_v3_control_requests, new_current_gop_frames, serve_media_v3,63};64use moq::{MOQ_REJECT_CODE, serve_authorized_moq, serve_control_moq};6566pub use session::SessionRegistry;67use session::{68 ClaimedMoqAttachment, ForcedIdrCoordinator, ForcedIdrDisposition, MediaV3Telemetry,69 MoqAttachmentWait, SourceTaskGuard,70};71use startup::select_gamescope_startup_source;727374enum MediaReplayDecision {75 Send { discontinuity: bool },76 SkipUntilKeyframe,77 DiscardStaleSuffix { through_sequence: u64 },78}798081struct MediaV3ObjectPosition {82 group_id: u64,83 object_id: u32,84 discontinuity: bool,85}868788enum MediaV3GroupDecision {89 Send(MediaV3ObjectPosition),90 SkipUntilKeyframe,91 EnterResync,92}939495struct MediaReplayCursor {96 last_sequence: Option<u64>,97 waiting_for_keyframe: bool,98 discontinuity_pending: bool,99}100101impl Default for MediaReplayCursor {102 fn default() -> Self {103 Self {104 last_sequence: None,105 waiting_for_keyframe: true,106 discontinuity_pending: false,107 }108 }109}110111impl MediaReplayCursor {112 fn classify(113 &mut self,114 frame: &EncodedFrame,115 replay_through_sequence: u64,116 initial_replay_started_at: Option<Instant>,117 observed_now: Instant,118 maximum_replay_age: Duration,119 ) -> MediaReplayDecision {120 let replay_age = observed_now.saturating_duration_since(frame.observed_at);121 let initial_replay_within_budget = initial_replay_started_at.is_some_and(|started_at| {122 observed_now.saturating_duration_since(started_at) <= maximum_replay_age123 });124 if !initial_replay_within_budget && replay_age > maximum_replay_age {125 self.last_sequence = Some(replay_through_sequence);126 self.waiting_for_keyframe = true;127 self.discontinuity_pending = true;128 return MediaReplayDecision::DiscardStaleSuffix {129 through_sequence: replay_through_sequence,130 };131 }132133 let sequence_discontinuity = self134 .last_sequence135 .is_some_and(|previous| previous.checked_add(1) != Some(frame.sequence));136 if sequence_discontinuity {137 self.waiting_for_keyframe = true;138 self.discontinuity_pending = true;139 }140 if self.waiting_for_keyframe && !(frame.keyframe && frame.codec_config) {141 return MediaReplayDecision::SkipUntilKeyframe;142 }143 MediaReplayDecision::Send {144 discontinuity: self.discontinuity_pending,145 }146 }147148 fn commit_sent(&mut self, frame: &EncodedFrame) {149 self.last_sequence = Some(frame.sequence);150 self.waiting_for_keyframe = false;151 self.discontinuity_pending = false;152 }153154 fn enter_resync_through(&mut self, through_sequence: Option<u64>) {155 if let Some(through_sequence) = through_sequence {156 self.last_sequence = Some(157 self.last_sequence158 .map_or(through_sequence, |last| last.max(through_sequence)),159 );160 }161 self.waiting_for_keyframe = true;162 self.discontinuity_pending = true;163 }164}165166fn maximum_media_replay_age(framerate: u32) -> Duration {167 debug_assert!(framerate > 0);168 let frame_period_nanos = 1_000_000_000_u64.div_ceil(u64::from(framerate.max(1)));169 Duration::from_nanos(frame_period_nanos.saturating_mul(MAX_MEDIA_REPLAY_AGE_FRAME_PERIODS))170}171172pub fn install_panic_hook() {173 std::panic::set_hook(Box::new(|info| {174 error!(%info, "host panic");175 }));176}177178179fn endpoint(byte: u8) -> EndpointId {180 iroh::SecretKey::from_bytes(&[byte; 32]).public()181}182183184fn moq_test_config() -> HostConfig {185 HostConfig {186 identity_path: "identity".into(),187 state_path: "state".into(),188 source: VideoSource::TestPattern,189 width: Some(1280),190 height: Some(800),191 framerate: 60,192 codec: "h264".to_owned(),193 input_mode: crate::config::InputMode::Disabled,194 uinput: None,195 ffmpeg_path: "ffmpeg".into(),196 gamescope_pipewire: None,197 audio: None,198 }199}200201202fn media_v3_encoded_frame(203 sequence: u64,204 keyframe: bool,205 codec_config: bool,206 payload_len: usize,207) -> EncodedFrame {208 EncodedFrame {209 sequence,210 width: 1_280,211 height: 800,212 capture_timestamp_micros: sequence,213 presentation_timestamp_micros: sequence as i64,214 observed_at: Instant::now(),215 keyframe,216 codec_config,217 discontinuity: false,218 data: Arc::from(vec![sequence as u8; payload_len]),219 }220}221222223struct DropNotify(Option<tokio::sync::oneshot::Sender<()>>);224225226impl Drop for DropNotify {227 fn drop(&mut self) {228 if let Some(sender) = self.0.take() {229 let _ = sender.send(());230 }231 }232}233234235mod tests {236 use super::*;237238 239 fn initial_current_gop_replay_preserves_complete_startup_snapshot() {240 let observed_now = Instant::now();241 let maximum_replay_age = maximum_media_replay_age(60);242 let old_observation = observed_now - Duration::from_secs(1);243 let frame = |sequence, keyframe| EncodedFrame {244 sequence,245 width: 1_280,246 height: 800,247 capture_timestamp_micros: sequence,248 presentation_timestamp_micros: sequence as i64,249 observed_at: old_observation,250 keyframe,251 codec_config: keyframe,252 discontinuity: false,253 data: Arc::from([sequence as u8]),254 };255 let mut cursor = MediaReplayCursor::default();256257 let keyframe = frame(10, true);258 assert_eq!(259 cursor.classify(260 &keyframe,261 12,262 Some(observed_now),263 observed_now,264 maximum_replay_age,265 ),266 MediaReplayDecision::Send {267 discontinuity: false,268 }269 );270 cursor.commit_sent(&keyframe);271272 let delta = frame(11, false);273 assert_eq!(274 cursor.classify(275 &delta,276 12,277 Some(observed_now),278 observed_now,279 maximum_replay_age,280 ),281 MediaReplayDecision::Send {282 discontinuity: false,283 }284 );285 }286287 288 fn stalled_initial_current_gop_discards_its_remaining_suffix() {289 let replay_started_at = Instant::now();290 let maximum_replay_age = maximum_media_replay_age(60);291 let old_observation = replay_started_at - Duration::from_secs(1);292 let frame = |sequence, keyframe| EncodedFrame {293 sequence,294 width: 1_280,295 height: 800,296 capture_timestamp_micros: sequence,297 presentation_timestamp_micros: sequence as i64,298 observed_at: old_observation,299 keyframe,300 codec_config: keyframe,301 discontinuity: false,302 data: Arc::from([sequence as u8]),303 };304 let mut cursor = MediaReplayCursor::default();305306 let keyframe = frame(10, true);307 assert_eq!(308 cursor.classify(309 &keyframe,310 12,311 Some(replay_started_at),312 replay_started_at,313 maximum_replay_age,314 ),315 MediaReplayDecision::Send {316 discontinuity: false,317 }318 );319 cursor.commit_sent(&keyframe);320321 let stalled_now = replay_started_at + maximum_replay_age + Duration::from_nanos(1);322 let delta = frame(11, false);323 assert_eq!(324 cursor.classify(325 &delta,326 12,327 Some(replay_started_at),328 stalled_now,329 maximum_replay_age,330 ),331 MediaReplayDecision::DiscardStaleSuffix {332 through_sequence: 12,333 }334 );335 assert_eq!(cursor.last_sequence, Some(12));336 assert!(cursor.waiting_for_keyframe);337 assert!(cursor.discontinuity_pending);338 }339340 341 fn fresh_current_gop_suffix_replays_normally() {342 let observed_now = Instant::now();343 let maximum_replay_age = maximum_media_replay_age(60);344 assert_eq!(maximum_replay_age, Duration::from_nanos(33_333_334));345 let frame = EncodedFrame {346 sequence: 11,347 width: 1_280,348 height: 800,349 capture_timestamp_micros: 11,350 presentation_timestamp_micros: 11,351 observed_at: observed_now - maximum_replay_age,352 keyframe: false,353 codec_config: false,354 discontinuity: false,355 data: Arc::from([11]),356 };357 let mut cursor = MediaReplayCursor {358 last_sequence: Some(10),359 waiting_for_keyframe: false,360 discontinuity_pending: false,361 };362363 assert_eq!(364 cursor.classify(&frame, 12, None, observed_now, maximum_replay_age),365 MediaReplayDecision::Send {366 discontinuity: false,367 }368 );369 cursor.commit_sent(&frame);370 assert_eq!(cursor.last_sequence, Some(11));371 }372373 374 fn stale_current_gop_suffix_is_discarded_as_one_bounded_unit() {375 let observed_now = Instant::now();376 let maximum_replay_age = maximum_media_replay_age(60);377 let frame = EncodedFrame {378 sequence: 11,379 width: 1_280,380 height: 800,381 capture_timestamp_micros: 11,382 presentation_timestamp_micros: 11,383 observed_at: observed_now - maximum_replay_age - Duration::from_nanos(1),384 keyframe: false,385 codec_config: false,386 discontinuity: false,387 data: Arc::from([11]),388 };389 let mut cursor = MediaReplayCursor {390 last_sequence: Some(10),391 waiting_for_keyframe: false,392 discontinuity_pending: false,393 };394395 assert_eq!(396 cursor.classify(&frame, 13, None, observed_now, maximum_replay_age),397 MediaReplayDecision::DiscardStaleSuffix {398 through_sequence: 13,399 }400 );401 assert_eq!(cursor.last_sequence, Some(13));402 assert!(cursor.waiting_for_keyframe);403 assert!(cursor.discontinuity_pending);404 }405406 407 fn stale_suffix_recovers_only_on_idr_marked_discontinuity() {408 let observed_now = Instant::now();409 let maximum_replay_age = maximum_media_replay_age(60);410 let frame = |sequence, keyframe| EncodedFrame {411 sequence,412 width: 1_280,413 height: 800,414 capture_timestamp_micros: sequence,415 presentation_timestamp_micros: sequence as i64,416 observed_at: observed_now,417 keyframe,418 codec_config: keyframe,419 discontinuity: false,420 data: Arc::from([sequence as u8]),421 };422 let mut cursor = MediaReplayCursor {423 last_sequence: Some(13),424 waiting_for_keyframe: true,425 discontinuity_pending: true,426 };427428 let delta = frame(14, false);429 assert_eq!(430 cursor.classify(&delta, 14, None, observed_now, maximum_replay_age),431 MediaReplayDecision::SkipUntilKeyframe432 );433 assert_eq!(cursor.last_sequence, Some(13));434435 let idr = frame(15, true);436 assert_eq!(437 cursor.classify(&idr, 15, None, observed_now, maximum_replay_age),438 MediaReplayDecision::Send {439 discontinuity: true,440 }441 );442 cursor.commit_sent(&idr);443 assert_eq!(cursor.last_sequence, Some(15));444 assert!(!cursor.waiting_for_keyframe);445 assert!(!cursor.discontinuity_pending);446447 let next_delta = frame(16, false);448 assert_eq!(449 cursor.classify(&next_delta, 16, None, observed_now, maximum_replay_age,),450 MediaReplayDecision::Send {451 discontinuity: false,452 }453 );454 }455}