- commit
- ccaf747f3ab708dbb429a052c32452ba89e06d52
- parent
- 8b95efdae1c2500238fe857368b145fe1984d34c
- author
- Dami <dami.etoile@gmail.com>
| M | src-tauri/ | 649 | +---------------------------- |
| M | src-tauri/ | 10 | + |
| M | src-tauri/ | 1 | + |
| A | src-tauri/ | 658 | ++++++++++++++++++++++++++++++ |
@@ -9,12 +9,10 @@ use crate::media::adaptive_feedback::{
99 use crate::media::audio_delivery::{1010 self, AudioStartRequest, lock_audio_deliveries, next_audio_generation, try_start_audio,1111 };12+1213 pub use crate::media::frame_channel::FramePayload;1314 use crate::media::frame_channel::{14- FrameEnvelopeMetadata, byte_to_codec, close_generation_connection, emit_frame_error,15- encode_frame_envelope, next_media_generation, release_frame_channel_slot,16- release_frame_channel_slot_for_generation, try_reserve_frame_channel_slot,17- validate_legacy_media_header,15+ close_generation_connection, next_media_generation, release_frame_channel_slot_for_generation,1816 };1917 use crate::media::input_delivery::{2018 CLIENT_INPUT_QUEUE_CAPACITY, InputSession, open_input_session, run_input_feedback,@@ -22,35 +20,18 @@ use crate::media::input_delivery::{
2220 };2321 2422 pub use crate::media::input_delivery::{PointerFeedbackPayload, PointerFeedbackTerminalReason};25-use crate::media::media_control::{26- request_keyframe, run_media_control_writer_v3, try_queue_media_keyframe_request,27-};28-use crate::media::metrics::{ClientMediaMetrics, lock_client_media_metrics};29-use crate::media::moq_receiver::{30- MoqMediaReadOutcome, open_upstream_moq_media, retire_upstream_moq_generation,31-};32-use crate::media::network_diagnostics::{33- NetworkLeg, NetworkSessionDiagnostics, lock_network_diagnostics,34-};35-use crate::media::object_receiver::{36- MediaObjectReadOutcome, MediaObjectReadOutcomeV3, MediaObjectReceiver, MediaObjectReceiverV3,37- MediaObjectSequence, MediaObjectSequenceDecision, MediaObjectSequenceDecisionV3,38- MediaObjectSequenceV3,39-};23+use crate::media::media_control::{request_keyframe, run_media_control_writer_v3};24+use crate::media::moq_receiver::open_upstream_moq_media;25+use crate::media::network_diagnostics::NetworkSessionDiagnostics;4026 use crate::media::transport::{4127 CLIENT_ENDPOINT_CLOSE_TIMEOUT, MediaTransport, open_negotiated_media_stream,4228 };43-use base64::Engine;29+use crate::media::video_delivery::{VideoDeliveryRequest, run_video_delivery};4430 use iroh::{Endpoint, SecretKey, endpoint::presets};45-use openh264::{formats::YUVSource, nal_units};4631 use serde::Serialize;4732 use sigil_protocol::{48- FrameFlags, InputEvent, InvitationGrants, KeyframeRequestReasonV3, MediaCodec,49- PointerSurfaceDimensions, read_media_frame,33+ InputEvent, InvitationGrants, KeyframeRequestReasonV3, PointerSurfaceDimensions,5034 };51-use std::io::Cursor;52-53-use std::sync::atomic::AtomicU64;5435 use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering};5536 use std::sync::{Arc, Mutex as StdMutex};5637 use std::time::{Duration, Instant};@@ -61,9 +42,6 @@ use tauri::{
6142 6243 // ─── Client commands ──────────────────────────────────────────────────────────6344 64-const LEGACY_MEDIA_IDLE_TIMEOUT: Duration = Duration::from_secs(10);65-const CLIENT_FRAME_STATS_INTERVAL: Duration = Duration::from_millis(250);66-6745 struct ClientConnectGuard {6846 active: Arc<AtomicBool>,6947 committed: bool,@@ -197,7 +175,7 @@ pub async fn iroh_client_connect(
197175 // protocols. The inherited v0 leg is retained below solely as migration198176 // code and is no longer selected by an ordinary Portal connection.199177 let use_v1 = true;200- let (frame_conn, mut frame_recv, media_control_stream, media_negotiation, media_transport) =178+ let (frame_conn, frame_recv, media_control_stream, media_negotiation, media_transport) =201179 if use_v1 {202180 let first_attempt = open_negotiated_media_stream(203181 &endpoint,@@ -437,569 +415,23 @@ pub async fn iroh_client_connect(
437415 438416 // Frame reader — dual path: WebCodecs (raw bytes) or software JPEG decode439417 let use_webcodecs = state.webcodecs.load(Ordering::SeqCst);440- tokio::spawn(async move {441- let metrics_started = Instant::now();442- let mut initial_metrics = ClientMediaMetrics::default();443- // Joining a running encoder commonly starts in the middle of a GOP.444- // The initial keyframe wait is a real resync episode, just like a wait445- // entered after frontend backpressure, so account for it from t=0.446- initial_metrics.begin_frontend_resync(Duration::ZERO);447- let metrics = Arc::new(StdMutex::new(initial_metrics));448- let mut previous_sequence: Option<u64> = None;449- let mut frontend_waiting_for_keyframe = true;450- let mut media_objects = (media_transport == MediaTransport::IndependentObjectsV2)451- .then(|| MediaObjectReceiver::new(frame_connection_for_stats.clone()));452- let mut media_object_sequence = MediaObjectSequence::new();453- let mut media_objects_v3 = (media_transport == MediaTransport::GroupedObjectsV3)454- .then(|| MediaObjectReceiverV3::new(frame_connection_for_stats.clone()));455- let mut media_object_sequence_v3 = MediaObjectSequenceV3::new();456- let mut upstream_moq_media = upstream_moq_media;457-458- let mut decoder = if use_webcodecs {459- None460- } else {461- match openh264::decoder::Decoder::new() {462- Ok(d) => Some(d),463- Err(e) => {464- emit_frame_error(&app, media_generation, format!("Decoder init failed: {e}"));465- if media_transport == MediaTransport::UpstreamMoq {466- retire_upstream_moq_generation(467- &app,468- media_generation,469- connected_audio_generation,470- )471- .await;472- }473- return;474- }475- }476- };477-478- let (stats_stop, mut stats_stop_rx) = tokio::sync::watch::channel(false);479- let stats_app = app.clone();480- let stats_metrics = Arc::clone(&metrics);481- let stats_in_flight = Arc::clone(&frame_events_in_flight);482- let stats_connection = frame_connection_for_stats.clone();483- let stats_input_connection = input_connection.clone();484- let stats_audio_connection = audio_connection_for_stats.clone();485- let stats_network_diagnostics = Arc::clone(&network_diagnostics);486- let stats_task = tokio::spawn(async move {487- let mut interval = tokio::time::interval(CLIENT_FRAME_STATS_INTERVAL);488- interval.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip);489- // Tokio intervals tick immediately once. Consume that tick so the490- // first payload represents a full diagnostics interval.491- interval.tick().await;492- let mut last_path_sample = Instant::now() - Duration::from_secs(1);493-494- loop {495- tokio::select! {496- _ = interval.tick() => {497- if last_path_sample.elapsed() >= Duration::from_secs(1) {498- let mut diagnostics = lock_network_diagnostics(&stats_network_diagnostics);499- diagnostics.observe_connection(NetworkLeg::Media, &stats_connection);500- diagnostics.observe_connection(NetworkLeg::Input, &stats_input_connection);501- if let Some(audio_connection) = stats_audio_connection.as_ref() {502- diagnostics.observe_connection(NetworkLeg::Audio, audio_connection);503- }504- last_path_sample = Instant::now();505- }506- let queue_depth = stats_in_flight.load(Ordering::SeqCst);507- let network_snapshot = lock_network_diagnostics(&stats_network_diagnostics)508- .snapshot(Instant::now());509- let payload = lock_client_media_metrics(&stats_metrics).snapshot(510- metrics_started.elapsed(),511- queue_depth,512- network_snapshot,513- media_generation,514- );515- let _ = stats_app.emit("frame-stats", payload);516- }517- changed = stats_stop_rx.changed() => {518- if changed.is_err() || *stats_stop_rx.borrow() {519- break;520- }521- }522- }523- }524- });525-526- 'frames: loop {527- let (528- w,529- h,530- frame_buf,531- is_keyframe,532- codec,533- sequence,534- capture_timestamp_micros,535- pts_micros,536- discontinuity,537- codec_config,538- ) = match media_transport {539- MediaTransport::UpstreamMoq => {540- let receiver = upstream_moq_media541- .as_mut()542- .expect("upstream MoQ receiver must exist for MoQ transport");543- loop {544- let outcome = match receiver.next().await {545- Ok(Some(outcome)) => outcome,546- Ok(None) => {547- emit_frame_error(548- &app,549- media_generation,550- "Upstream MoQ video track closed",551- );552- break 'frames;553- }554- Err(error) => {555- emit_frame_error(&app, media_generation, error);556- break 'frames;557- }558- };559- match outcome {560- MoqMediaReadOutcome::Dropped { reason } => {561- frontend_waiting_for_keyframe = true;562- let mut metrics = lock_client_media_metrics(&metrics);563- metrics.observe_transport_object_drop(false);564- metrics.begin_frontend_resync(metrics_started.elapsed());565- try_queue_media_keyframe_request(566- media_control_requests.as_ref(),567- reason,568- receiver.last_frame_sequence(),569- );570- }571- MoqMediaReadOutcome::Malformed(error) => {572- emit_frame_error(&app, media_generation, error);573- break 'frames;574- }575- MoqMediaReadOutcome::Frame {576- frame,577- discontinuity,578- } => {579- let codec = match frame.header.codec {580- MediaCodec::H264 => "h264".to_string(),581- };582- break (583- u32::from(frame.header.width),584- u32::from(frame.header.height),585- frame.payload,586- frame.header.flags.contains(FrameFlags::KEYFRAME),587- codec,588- Some(frame.header.sequence),589- Some(frame.header.capture_timestamp_us),590- Some(frame.header.pts_us),591- discontinuity,592- frame.header.flags.contains(FrameFlags::CODEC_CONFIG),593- );594- }595- }596- }597- }598- MediaTransport::GroupedObjectsV3 => {599- let receiver = media_objects_v3600- .as_mut()601- .expect("media v3 receiver must exist for media v3 transport");602- loop {603- let outcome = match receiver.next().await {604- Ok(Some(outcome)) => outcome,605- Ok(None) => {606- emit_frame_error(&app, media_generation, "Connection closed");607- break 'frames;608- }609- Err(error) => {610- emit_frame_error(&app, media_generation, error);611- break 'frames;612- }613- };614- match outcome {615- MediaObjectReadOutcomeV3::Dropped { reason, .. } => {616- let begins_resync = media_object_sequence_v3.note_dropped_object();617- let mut metrics = lock_client_media_metrics(&metrics);618- metrics.observe_transport_object_drop(!begins_resync);619- if begins_resync {620- frontend_waiting_for_keyframe = true;621- metrics.begin_frontend_resync(metrics_started.elapsed());622- try_queue_media_keyframe_request(623- media_control_requests.as_ref(),624- reason,625- media_object_sequence_v3.last_sequence(),626- );627- }628- }629- MediaObjectReadOutcomeV3::Malformed(error) => {630- emit_frame_error(&app, media_generation, error);631- break 'frames;632- }633- MediaObjectReadOutcomeV3::Object { object, .. } => {634- let was_waiting = media_object_sequence_v3.waiting_for_keyframe();635- let discontinuity = match media_object_sequence_v3.classify(&object)636- {637- MediaObjectSequenceDecisionV3::Deliver { discontinuity } => {638- discontinuity639- }640- MediaObjectSequenceDecisionV3::DropLate => {641- lock_client_media_metrics(&metrics)642- .observe_transport_object_drop(true);643- continue;644- }645- MediaObjectSequenceDecisionV3::DropUntilKeyframe => {646- frontend_waiting_for_keyframe = true;647- let mut metrics = lock_client_media_metrics(&metrics);648- metrics.observe_transport_object_drop(false);649- metrics.begin_frontend_resync(metrics_started.elapsed());650- if !was_waiting {651- try_queue_media_keyframe_request(652- media_control_requests.as_ref(),653- KeyframeRequestReasonV3::TransportGap,654- media_object_sequence_v3.last_sequence(),655- );656- }657- continue;658- }659- };660- let codec = match object.header.codec {661- MediaCodec::H264 => "h264".to_string(),662- };663- break (664- u32::from(object.header.width),665- u32::from(object.header.height),666- object.payload,667- object.header.flags.contains(FrameFlags::KEYFRAME),668- codec,669- Some(object.header.sequence),670- Some(object.header.capture_timestamp_us),671- Some(object.header.pts_us),672- discontinuity,673- object.header.flags.contains(FrameFlags::CODEC_CONFIG),674- );675- }676- }677- }678- }679- MediaTransport::IndependentObjectsV2 => {680- let receiver = media_objects681- .as_mut()682- .expect("media v2 receiver must exist for media v2 transport");683- loop {684- let outcome = match receiver.next().await {685- Ok(Some(outcome)) => outcome,686- Ok(None) => {687- emit_frame_error(&app, media_generation, "Connection closed");688- break 'frames;689- }690- Err(error) => {691- emit_frame_error(&app, media_generation, error);692- break 'frames;693- }694- };695- match outcome {696- MediaObjectReadOutcome::Dropped { object_index } => {697- let begins_resync =698- media_object_sequence.note_dropped_object(object_index);699- let mut metrics = lock_client_media_metrics(&metrics);700- metrics.observe_transport_object_drop(!begins_resync);701- if begins_resync {702- frontend_waiting_for_keyframe = true;703- metrics.begin_frontend_resync(metrics_started.elapsed());704- }705- }706- MediaObjectReadOutcome::Malformed(error) => {707- emit_frame_error(&app, media_generation, error);708- break 'frames;709- }710- MediaObjectReadOutcome::Frame {711- object_index,712- frame,713- } => {714- let discontinuity = match media_object_sequence715- .classify(object_index, &frame)716- {717- MediaObjectSequenceDecision::Deliver { discontinuity } => {718- discontinuity719- }720- MediaObjectSequenceDecision::DropLate => {721- lock_client_media_metrics(&metrics)722- .observe_transport_object_drop(true);723- continue;724- }725- MediaObjectSequenceDecision::DropUntilKeyframe => {726- frontend_waiting_for_keyframe = true;727- let mut metrics = lock_client_media_metrics(&metrics);728- metrics.observe_transport_object_drop(false);729- metrics.begin_frontend_resync(metrics_started.elapsed());730- continue;731- }732- };733- let codec = match frame.header.codec {734- MediaCodec::H264 => "h264".to_string(),735- };736- break (737- u32::from(frame.header.width),738- u32::from(frame.header.height),739- frame.payload,740- frame.header.flags.contains(FrameFlags::KEYFRAME),741- codec,742- Some(frame.header.sequence),743- Some(frame.header.capture_timestamp_us),744- Some(frame.header.pts_us),745- discontinuity,746- frame.header.flags.contains(FrameFlags::CODEC_CONFIG),747- );748- }749- }750- }751- }752- MediaTransport::ReliableStreamV1 => {753- // Gamescope's PipeWire stream is damage-driven: a static754- // screen can legitimately produce no encoded frame for an755- // arbitrary period. Connection closure and parser/source756- // errors remain terminal; frame silence does not.757- let frame = match read_media_frame(&mut frame_recv).await {758- Ok(Some(frame)) => frame,759- Ok(None) => {760- emit_frame_error(&app, media_generation, "Connection closed");761- break;762- }763- Err(error) => {764- emit_frame_error(765- &app,766- media_generation,767- format!("Invalid media stream: {error}"),768- );769- break;770- }771- };772- let codec = match frame.header.codec {773- MediaCodec::H264 => "h264".to_string(),774- };775- (776- u32::from(frame.header.width),777- u32::from(frame.header.height),778- frame.payload,779- frame.header.flags.contains(FrameFlags::KEYFRAME),780- codec,781- Some(frame.header.sequence),782- Some(frame.header.capture_timestamp_us),783- Some(frame.header.pts_us),784- frame.header.flags.contains(FrameFlags::DISCONTINUITY),785- frame.header.flags.contains(FrameFlags::CODEC_CONFIG),786- )787- }788- MediaTransport::LegacyV0 => {789- let mut header = [0u8; 14];790- match tokio::time::timeout(791- LEGACY_MEDIA_IDLE_TIMEOUT,792- frame_recv.read_exact(&mut header),793- )794- .await795- {796- Err(_) => {797- emit_frame_error(&app, media_generation, "Media stream idle timeout");798- break;799- }800- Ok(Err(_)) => {801- emit_frame_error(&app, media_generation, "Connection lost");802- break;803- }804- Ok(Ok(_)) => {}805- }806- let w = u32::from_be_bytes([header[0], header[1], header[2], header[3]]);807- let h = u32::from_be_bytes([header[4], header[5], header[6], header[7]]);808- let frame_len =809- u32::from_be_bytes([header[8], header[9], header[10], header[11]]) as usize;810- if let Err(error) = validate_legacy_media_header(w, h, frame_len) {811- emit_frame_error(&app, media_generation, error);812- break;813- }814-815- let is_keyframe = header[12] == 1;816- let codec = byte_to_codec(header[13]).to_string();817-818- let mut frame_buf = vec![0u8; frame_len];819- match tokio::time::timeout(820- LEGACY_MEDIA_IDLE_TIMEOUT,821- frame_recv.read_exact(&mut frame_buf),822- )823- .await824- {825- Err(_) => {826- emit_frame_error(&app, media_generation, "Media payload idle timeout");827- break;828- }829- Ok(Err(_)) => {830- emit_frame_error(&app, media_generation, "Connection lost");831- break;832- }833- Ok(Ok(_)) => {}834- }835- (836- w,837- h,838- frame_buf,839- is_keyframe,840- codec,841- None,842- None,843- None,844- false,845- false,846- )847- }848- };849- lock_client_media_metrics(&metrics).observe_transport_receive(850- metrics_started.elapsed(),851- sequence,852- is_keyframe,853- );854-855- let sequence_gap = match sequence.zip(previous_sequence) {856- Some((current, previous)) => match sequence_gap(previous, current) {857- Ok(gap) => gap,858- Err(error) => {859- emit_frame_error(&app, media_generation, error);860- break;861- }862- },863- None => 0,864- };865- if sequence.is_some() {866- previous_sequence = sequence;867- }868- lock_client_media_metrics(&metrics).observe_sequence_drop(sequence_gap);869-870- if frontend_waiting_for_keyframe && !is_keyframe {871- lock_client_media_metrics(&metrics).observe_frontend_resync_drop();872- continue;873- }874- if !try_reserve_frame_channel_slot(&frame_events_in_flight) {875- let begins_resync = !frontend_waiting_for_keyframe;876- frontend_waiting_for_keyframe = true;877- let mut metrics = lock_client_media_metrics(&metrics);878- metrics.observe_frontend_queue_drop();879- metrics.begin_frontend_resync(metrics_started.elapsed());880- if begins_resync {881- try_queue_media_keyframe_request(882- media_control_requests.as_ref(),883- KeyframeRequestReasonV3::FrontendBackpressure,884- sequence,885- );886- }887- continue;888- }889- lock_client_media_metrics(&metrics)890- .observe_frontend_queue_depth(frame_events_in_flight.load(Ordering::SeqCst));891-892- let delivered_to_frontend;893- if use_webcodecs {894- let envelope = match encode_frame_envelope(895- FrameEnvelopeMetadata {896- width: w,897- height: h,898- codec: &codec,899- keyframe: is_keyframe,900- discontinuity,901- codec_config,902- sequence,903- capture_timestamp_micros,904- pts_micros,905- },906- &frame_buf,907- ) {908- Ok(envelope) => envelope,909- Err(error) => {910- emit_frame_error(&app, media_generation, error);911- release_frame_channel_slot(&frame_events_in_flight);912- break;913- }914- };915- let ipc_send_started = Instant::now();916- let send_result = frame_channel.send(Response::new(envelope));917- lock_client_media_metrics(&metrics).observe_frontend_ipc_send_duration(918- metrics_started.elapsed(),919- ipc_send_started.elapsed(),920- );921- if send_result.is_err() {922- release_frame_channel_slot(&frame_events_in_flight);923- break;924- }925- delivered_to_frontend = true;926- } else if let Some(ref mut dec) = decoder {927- let mut emitted = false;928- for nal in nal_units(&frame_buf) {929- if let Ok(Some(yuv)) = dec.decode(nal) {930- let (yw, yh) = yuv.dimensions();931- let rgb_len = yuv.rgb8_len();932- let mut rgb_raw = vec![0u8; rgb_len];933- yuv.write_rgb8(&mut rgb_raw);934-935- let img = match image::RgbImage::from_raw(yw as u32, yh as u32, rgb_raw) {936- Some(img) => img,937- None => continue,938- };939- let mut jpeg_buf = Vec::with_capacity(30_000);940- if image::DynamicImage::ImageRgb8(img)941- .write_to(&mut Cursor::new(&mut jpeg_buf), image::ImageFormat::Jpeg)942- .is_err()943- {944- continue;945- }946-947- let b64 = base64::engine::general_purpose::STANDARD.encode(&jpeg_buf);948- let ipc_send_started = Instant::now();949- let send_result = app.emit(950- "frame",951- FramePayload {952- generation: media_generation,953- width: yw as u32,954- height: yh as u32,955- data: b64,956- keyframe: is_keyframe,957- codec: codec.clone(),958- capture_timestamp_micros,959- pts_micros,960- discontinuity,961- },962- );963- lock_client_media_metrics(&metrics).observe_frontend_ipc_send_duration(964- metrics_started.elapsed(),965- ipc_send_started.elapsed(),966- );967- if send_result.is_err() {968- release_frame_channel_slot(&frame_events_in_flight);969- break 'frames;970- }971- emitted = true;972- break;973- }974- }975- if !emitted {976- release_frame_channel_slot(&frame_events_in_flight);977- }978- delivered_to_frontend = emitted;979- } else {980- release_frame_channel_slot(&frame_events_in_flight);981- delivered_to_frontend = false;982- }983-984- if delivered_to_frontend {985- frontend_waiting_for_keyframe = false;986- let mut metrics = lock_client_media_metrics(&metrics);987- let elapsed = metrics_started.elapsed();988- metrics.observe_frontend_send(elapsed);989- metrics.finish_frontend_resync(elapsed);990- }991-992- tokio::task::yield_now().await;993- }994-995- let _ = stats_stop.send(true);996- let _ = stats_task.await;997- if media_transport == MediaTransport::UpstreamMoq {998- retire_upstream_moq_generation(&app, media_generation, connected_audio_generation)999- .await;1000- }1001- drop(endpoint);1002- });418+ tokio::spawn(run_video_delivery(VideoDeliveryRequest {419+ app,420+ endpoint,421+ frame_recv,422+ frame_connection: frame_connection_for_stats,423+ input_connection,424+ audio_connection: audio_connection_for_stats,425+ network_diagnostics,426+ media_control_requests,427+ upstream_moq_media,428+ frame_events_in_flight,429+ frame_channel,430+ media_transport,431+ media_generation,432+ connected_audio_generation,433+ use_webcodecs,434+ }));1003435 1004436 let result = ConnectResult {1005437 connected: true,@@ -1026,32 +458,10 @@ pub async fn iroh_client_connect(
1026458 Ok(result)1027459 }1028460 1029-fn sequence_gap(previous: u64, current: u64) -> Result<u64, String> {1030- let expected = previous1031- .checked_add(1)1032- .ok_or_else(|| format!("Media sequence overflowed after {previous}"))?;1033- if current < expected {1034- return Err(format!(1035- "Non-monotonic media sequence: previous={previous}, current={current}"1036- ));1037- }1038- Ok(current - expected)1039-}1040-1041461 1042462 mod tests {1043463 use super::*;1044464 1045- 1046- fn media_generations_are_nonzero_monotonic_and_checked_for_overflow() {1047- let counter = AtomicU64::new(0);1048- assert_eq!(next_media_generation(&counter).unwrap(), 1);1049- assert_eq!(next_media_generation(&counter).unwrap(), 2);1050-1051- let exhausted = AtomicU64::new(u64::MAX);1052- assert!(next_media_generation(&exhausted).is_err());1053- }1054-1055465 1056466 fn client_connect_guard_rejects_overlap_and_resets_only_failed_attempts() {1057467 let active = Arc::new(AtomicBool::new(false));@@ -1067,15 +477,6 @@ mod tests {
1067477 assert!(ClientConnectGuard::acquire(Arc::clone(&active)).is_err());1068478 active.store(false, Ordering::SeqCst);1069479 }1070-1071- 1072- fn sequence_checks_reject_duplicates_regressions_and_overflow() {1073- assert_eq!(sequence_gap(41, 42).unwrap(), 0);1074- assert_eq!(sequence_gap(41, 45).unwrap(), 3);1075- assert!(sequence_gap(41, 41).is_err());1076- assert!(sequence_gap(41, 40).is_err());1077- assert!(sequence_gap(u64::MAX, 0).is_err());1078- }1079480 }1080481 1081482 @@ -250,6 +250,16 @@ pub(crate) fn validate_legacy_media_header(
250250 mod tests {251251 use super::*;252252 253+ 254+ fn media_generations_are_nonzero_monotonic_and_checked_for_overflow() {255+ let counter = AtomicU64::new(0);256+ assert_eq!(next_media_generation(&counter).unwrap(), 1);257+ assert_eq!(next_media_generation(&counter).unwrap(), 2);258+259+ let exhausted = AtomicU64::new(u64::MAX);260+ assert!(next_media_generation(&exhausted).is_err());261+ }262+253263 254264 fn binary_frame_envelope_has_exact_stable_layout() {255265 let payload = [0, 0, 0, 1, 0x65];@@ -9,3 +9,4 @@ pub(crate) mod moq_receiver;
99 pub(crate) mod network_diagnostics;1010 pub(crate) mod object_receiver;1111 pub(crate) mod transport;12+pub(crate) mod video_delivery;@@ -0,0 +1,658 @@
1+use std::io::Cursor;2+use std::sync::atomic::{AtomicUsize, Ordering};3+use std::sync::{Arc, Mutex as StdMutex};4+use std::time::{Duration, Instant};5+6+use base64::Engine;7+use openh264::{formats::YUVSource, nal_units};8+use sigil_protocol::{FrameFlags, KeyframeRequestReasonV3, MediaCodec, read_media_frame};9+use tauri::{10+ AppHandle, Emitter,11+ ipc::{Channel, Response},12+};13+14+use crate::commands::state::MediaControlRequestSender;15+use crate::media::frame_channel::{16+ FrameEnvelopeMetadata, FramePayload, byte_to_codec, emit_frame_error, encode_frame_envelope,17+ release_frame_channel_slot, try_reserve_frame_channel_slot, validate_legacy_media_header,18+};19+use crate::media::media_control::try_queue_media_keyframe_request;20+use crate::media::metrics::{ClientMediaMetrics, lock_client_media_metrics};21+use crate::media::moq_receiver::{22+ MoqMediaReadOutcome, MoqMediaReceiver, retire_upstream_moq_generation,23+};24+use crate::media::network_diagnostics::{25+ NetworkLeg, NetworkSessionDiagnostics, lock_network_diagnostics,26+};27+use crate::media::object_receiver::{28+ MediaObjectReadOutcome, MediaObjectReadOutcomeV3, MediaObjectReceiver, MediaObjectReceiverV3,29+ MediaObjectSequence, MediaObjectSequenceDecision, MediaObjectSequenceDecisionV3,30+ MediaObjectSequenceV3,31+};32+use crate::media::transport::MediaTransport;33+34+const LEGACY_MEDIA_IDLE_TIMEOUT: Duration = Duration::from_secs(10);35+const CLIENT_FRAME_STATS_INTERVAL: Duration = Duration::from_millis(250);36+37+pub(crate) struct VideoDeliveryRequest {38+ pub(crate) app: AppHandle,39+ pub(crate) endpoint: iroh::Endpoint,40+ pub(crate) frame_recv: iroh::endpoint::RecvStream,41+ pub(crate) frame_connection: iroh::endpoint::Connection,42+ pub(crate) input_connection: iroh::endpoint::Connection,43+ pub(crate) audio_connection: Option<iroh::endpoint::Connection>,44+ pub(crate) network_diagnostics: Arc<StdMutex<NetworkSessionDiagnostics>>,45+ pub(crate) media_control_requests: Option<MediaControlRequestSender>,46+ pub(crate) upstream_moq_media: Option<MoqMediaReceiver>,47+ pub(crate) frame_events_in_flight: Arc<AtomicUsize>,48+ pub(crate) frame_channel: Channel<Response>,49+ pub(crate) media_transport: MediaTransport,50+ pub(crate) media_generation: u64,51+ pub(crate) connected_audio_generation: Option<u64>,52+ pub(crate) use_webcodecs: bool,53+}54+55+pub(crate) async fn run_video_delivery(request: VideoDeliveryRequest) {56+ let VideoDeliveryRequest {57+ app,58+ endpoint,59+ mut frame_recv,60+ frame_connection: frame_connection_for_stats,61+ input_connection,62+ audio_connection: audio_connection_for_stats,63+ network_diagnostics,64+ media_control_requests,65+ upstream_moq_media,66+ frame_events_in_flight,67+ frame_channel,68+ media_transport,69+ media_generation,70+ connected_audio_generation,71+ use_webcodecs,72+ } = request;73+74+ let metrics_started = Instant::now();75+ let mut initial_metrics = ClientMediaMetrics::default();76+ // Joining a running encoder commonly starts in the middle of a GOP.77+ // The initial keyframe wait is a real resync episode, just like a wait78+ // entered after frontend backpressure, so account for it from t=0.79+ initial_metrics.begin_frontend_resync(Duration::ZERO);80+ let metrics = Arc::new(StdMutex::new(initial_metrics));81+ let mut previous_sequence: Option<u64> = None;82+ let mut frontend_waiting_for_keyframe = true;83+ let mut media_objects = (media_transport == MediaTransport::IndependentObjectsV2)84+ .then(|| MediaObjectReceiver::new(frame_connection_for_stats.clone()));85+ let mut media_object_sequence = MediaObjectSequence::new();86+ let mut media_objects_v3 = (media_transport == MediaTransport::GroupedObjectsV3)87+ .then(|| MediaObjectReceiverV3::new(frame_connection_for_stats.clone()));88+ let mut media_object_sequence_v3 = MediaObjectSequenceV3::new();89+ let mut upstream_moq_media = upstream_moq_media;90+91+ let mut decoder = if use_webcodecs {92+ None93+ } else {94+ match openh264::decoder::Decoder::new() {95+ Ok(d) => Some(d),96+ Err(e) => {97+ emit_frame_error(&app, media_generation, format!("Decoder init failed: {e}"));98+ if media_transport == MediaTransport::UpstreamMoq {99+ retire_upstream_moq_generation(100+ &app,101+ media_generation,102+ connected_audio_generation,103+ )104+ .await;105+ }106+ return;107+ }108+ }109+ };110+111+ let (stats_stop, mut stats_stop_rx) = tokio::sync::watch::channel(false);112+ let stats_app = app.clone();113+ let stats_metrics = Arc::clone(&metrics);114+ let stats_in_flight = Arc::clone(&frame_events_in_flight);115+ let stats_connection = frame_connection_for_stats.clone();116+ let stats_input_connection = input_connection.clone();117+ let stats_audio_connection = audio_connection_for_stats.clone();118+ let stats_network_diagnostics = Arc::clone(&network_diagnostics);119+ let stats_task = tokio::spawn(async move {120+ let mut interval = tokio::time::interval(CLIENT_FRAME_STATS_INTERVAL);121+ interval.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip);122+ // Tokio intervals tick immediately once. Consume that tick so the123+ // first payload represents a full diagnostics interval.124+ interval.tick().await;125+ let mut last_path_sample = Instant::now() - Duration::from_secs(1);126+127+ loop {128+ tokio::select! {129+ _ = interval.tick() => {130+ if last_path_sample.elapsed() >= Duration::from_secs(1) {131+ let mut diagnostics = lock_network_diagnostics(&stats_network_diagnostics);132+ diagnostics.observe_connection(NetworkLeg::Media, &stats_connection);133+ diagnostics.observe_connection(NetworkLeg::Input, &stats_input_connection);134+ if let Some(audio_connection) = stats_audio_connection.as_ref() {135+ diagnostics.observe_connection(NetworkLeg::Audio, audio_connection);136+ }137+ last_path_sample = Instant::now();138+ }139+ let queue_depth = stats_in_flight.load(Ordering::SeqCst);140+ let network_snapshot = lock_network_diagnostics(&stats_network_diagnostics)141+ .snapshot(Instant::now());142+ let payload = lock_client_media_metrics(&stats_metrics).snapshot(143+ metrics_started.elapsed(),144+ queue_depth,145+ network_snapshot,146+ media_generation,147+ );148+ let _ = stats_app.emit("frame-stats", payload);149+ }150+ changed = stats_stop_rx.changed() => {151+ if changed.is_err() || *stats_stop_rx.borrow() {152+ break;153+ }154+ }155+ }156+ }157+ });158+159+ 'frames: loop {160+ let (161+ w,162+ h,163+ frame_buf,164+ is_keyframe,165+ codec,166+ sequence,167+ capture_timestamp_micros,168+ pts_micros,169+ discontinuity,170+ codec_config,171+ ) = match media_transport {172+ MediaTransport::UpstreamMoq => {173+ let receiver = upstream_moq_media174+ .as_mut()175+ .expect("upstream MoQ receiver must exist for MoQ transport");176+ loop {177+ let outcome = match receiver.next().await {178+ Ok(Some(outcome)) => outcome,179+ Ok(None) => {180+ emit_frame_error(181+ &app,182+ media_generation,183+ "Upstream MoQ video track closed",184+ );185+ break 'frames;186+ }187+ Err(error) => {188+ emit_frame_error(&app, media_generation, error);189+ break 'frames;190+ }191+ };192+ match outcome {193+ MoqMediaReadOutcome::Dropped { reason } => {194+ frontend_waiting_for_keyframe = true;195+ let mut metrics = lock_client_media_metrics(&metrics);196+ metrics.observe_transport_object_drop(false);197+ metrics.begin_frontend_resync(metrics_started.elapsed());198+ try_queue_media_keyframe_request(199+ media_control_requests.as_ref(),200+ reason,201+ receiver.last_frame_sequence(),202+ );203+ }204+ MoqMediaReadOutcome::Malformed(error) => {205+ emit_frame_error(&app, media_generation, error);206+ break 'frames;207+ }208+ MoqMediaReadOutcome::Frame {209+ frame,210+ discontinuity,211+ } => {212+ let codec = match frame.header.codec {213+ MediaCodec::H264 => "h264".to_string(),214+ };215+ break (216+ u32::from(frame.header.width),217+ u32::from(frame.header.height),218+ frame.payload,219+ frame.header.flags.contains(FrameFlags::KEYFRAME),220+ codec,221+ Some(frame.header.sequence),222+ Some(frame.header.capture_timestamp_us),223+ Some(frame.header.pts_us),224+ discontinuity,225+ frame.header.flags.contains(FrameFlags::CODEC_CONFIG),226+ );227+ }228+ }229+ }230+ }231+ MediaTransport::GroupedObjectsV3 => {232+ let receiver = media_objects_v3233+ .as_mut()234+ .expect("media v3 receiver must exist for media v3 transport");235+ loop {236+ let outcome = match receiver.next().await {237+ Ok(Some(outcome)) => outcome,238+ Ok(None) => {239+ emit_frame_error(&app, media_generation, "Connection closed");240+ break 'frames;241+ }242+ Err(error) => {243+ emit_frame_error(&app, media_generation, error);244+ break 'frames;245+ }246+ };247+ match outcome {248+ MediaObjectReadOutcomeV3::Dropped { reason, .. } => {249+ let begins_resync = media_object_sequence_v3.note_dropped_object();250+ let mut metrics = lock_client_media_metrics(&metrics);251+ metrics.observe_transport_object_drop(!begins_resync);252+ if begins_resync {253+ frontend_waiting_for_keyframe = true;254+ metrics.begin_frontend_resync(metrics_started.elapsed());255+ try_queue_media_keyframe_request(256+ media_control_requests.as_ref(),257+ reason,258+ media_object_sequence_v3.last_sequence(),259+ );260+ }261+ }262+ MediaObjectReadOutcomeV3::Malformed(error) => {263+ emit_frame_error(&app, media_generation, error);264+ break 'frames;265+ }266+ MediaObjectReadOutcomeV3::Object { object, .. } => {267+ let was_waiting = media_object_sequence_v3.waiting_for_keyframe();268+ let discontinuity = match media_object_sequence_v3.classify(&object) {269+ MediaObjectSequenceDecisionV3::Deliver { discontinuity } => {270+ discontinuity271+ }272+ MediaObjectSequenceDecisionV3::DropLate => {273+ lock_client_media_metrics(&metrics)274+ .observe_transport_object_drop(true);275+ continue;276+ }277+ MediaObjectSequenceDecisionV3::DropUntilKeyframe => {278+ frontend_waiting_for_keyframe = true;279+ let mut metrics = lock_client_media_metrics(&metrics);280+ metrics.observe_transport_object_drop(false);281+ metrics.begin_frontend_resync(metrics_started.elapsed());282+ if !was_waiting {283+ try_queue_media_keyframe_request(284+ media_control_requests.as_ref(),285+ KeyframeRequestReasonV3::TransportGap,286+ media_object_sequence_v3.last_sequence(),287+ );288+ }289+ continue;290+ }291+ };292+ let codec = match object.header.codec {293+ MediaCodec::H264 => "h264".to_string(),294+ };295+ break (296+ u32::from(object.header.width),297+ u32::from(object.header.height),298+ object.payload,299+ object.header.flags.contains(FrameFlags::KEYFRAME),300+ codec,301+ Some(object.header.sequence),302+ Some(object.header.capture_timestamp_us),303+ Some(object.header.pts_us),304+ discontinuity,305+ object.header.flags.contains(FrameFlags::CODEC_CONFIG),306+ );307+ }308+ }309+ }310+ }311+ MediaTransport::IndependentObjectsV2 => {312+ let receiver = media_objects313+ .as_mut()314+ .expect("media v2 receiver must exist for media v2 transport");315+ loop {316+ let outcome = match receiver.next().await {317+ Ok(Some(outcome)) => outcome,318+ Ok(None) => {319+ emit_frame_error(&app, media_generation, "Connection closed");320+ break 'frames;321+ }322+ Err(error) => {323+ emit_frame_error(&app, media_generation, error);324+ break 'frames;325+ }326+ };327+ match outcome {328+ MediaObjectReadOutcome::Dropped { object_index } => {329+ let begins_resync =330+ media_object_sequence.note_dropped_object(object_index);331+ let mut metrics = lock_client_media_metrics(&metrics);332+ metrics.observe_transport_object_drop(!begins_resync);333+ if begins_resync {334+ frontend_waiting_for_keyframe = true;335+ metrics.begin_frontend_resync(metrics_started.elapsed());336+ }337+ }338+ MediaObjectReadOutcome::Malformed(error) => {339+ emit_frame_error(&app, media_generation, error);340+ break 'frames;341+ }342+ MediaObjectReadOutcome::Frame {343+ object_index,344+ frame,345+ } => {346+ let discontinuity =347+ match media_object_sequence.classify(object_index, &frame) {348+ MediaObjectSequenceDecision::Deliver { discontinuity } => {349+ discontinuity350+ }351+ MediaObjectSequenceDecision::DropLate => {352+ lock_client_media_metrics(&metrics)353+ .observe_transport_object_drop(true);354+ continue;355+ }356+ MediaObjectSequenceDecision::DropUntilKeyframe => {357+ frontend_waiting_for_keyframe = true;358+ let mut metrics = lock_client_media_metrics(&metrics);359+ metrics.observe_transport_object_drop(false);360+ metrics.begin_frontend_resync(metrics_started.elapsed());361+ continue;362+ }363+ };364+ let codec = match frame.header.codec {365+ MediaCodec::H264 => "h264".to_string(),366+ };367+ break (368+ u32::from(frame.header.width),369+ u32::from(frame.header.height),370+ frame.payload,371+ frame.header.flags.contains(FrameFlags::KEYFRAME),372+ codec,373+ Some(frame.header.sequence),374+ Some(frame.header.capture_timestamp_us),375+ Some(frame.header.pts_us),376+ discontinuity,377+ frame.header.flags.contains(FrameFlags::CODEC_CONFIG),378+ );379+ }380+ }381+ }382+ }383+ MediaTransport::ReliableStreamV1 => {384+ // Gamescope's PipeWire stream is damage-driven: a static385+ // screen can legitimately produce no encoded frame for an386+ // arbitrary period. Connection closure and parser/source387+ // errors remain terminal; frame silence does not.388+ let frame = match read_media_frame(&mut frame_recv).await {389+ Ok(Some(frame)) => frame,390+ Ok(None) => {391+ emit_frame_error(&app, media_generation, "Connection closed");392+ break;393+ }394+ Err(error) => {395+ emit_frame_error(396+ &app,397+ media_generation,398+ format!("Invalid media stream: {error}"),399+ );400+ break;401+ }402+ };403+ let codec = match frame.header.codec {404+ MediaCodec::H264 => "h264".to_string(),405+ };406+ (407+ u32::from(frame.header.width),408+ u32::from(frame.header.height),409+ frame.payload,410+ frame.header.flags.contains(FrameFlags::KEYFRAME),411+ codec,412+ Some(frame.header.sequence),413+ Some(frame.header.capture_timestamp_us),414+ Some(frame.header.pts_us),415+ frame.header.flags.contains(FrameFlags::DISCONTINUITY),416+ frame.header.flags.contains(FrameFlags::CODEC_CONFIG),417+ )418+ }419+ MediaTransport::LegacyV0 => {420+ let mut header = [0u8; 14];421+ match tokio::time::timeout(422+ LEGACY_MEDIA_IDLE_TIMEOUT,423+ frame_recv.read_exact(&mut header),424+ )425+ .await426+ {427+ Err(_) => {428+ emit_frame_error(&app, media_generation, "Media stream idle timeout");429+ break;430+ }431+ Ok(Err(_)) => {432+ emit_frame_error(&app, media_generation, "Connection lost");433+ break;434+ }435+ Ok(Ok(_)) => {}436+ }437+ let w = u32::from_be_bytes([header[0], header[1], header[2], header[3]]);438+ let h = u32::from_be_bytes([header[4], header[5], header[6], header[7]]);439+ let frame_len =440+ u32::from_be_bytes([header[8], header[9], header[10], header[11]]) as usize;441+ if let Err(error) = validate_legacy_media_header(w, h, frame_len) {442+ emit_frame_error(&app, media_generation, error);443+ break;444+ }445+446+ let is_keyframe = header[12] == 1;447+ let codec = byte_to_codec(header[13]).to_string();448+449+ let mut frame_buf = vec![0u8; frame_len];450+ match tokio::time::timeout(451+ LEGACY_MEDIA_IDLE_TIMEOUT,452+ frame_recv.read_exact(&mut frame_buf),453+ )454+ .await455+ {456+ Err(_) => {457+ emit_frame_error(&app, media_generation, "Media payload idle timeout");458+ break;459+ }460+ Ok(Err(_)) => {461+ emit_frame_error(&app, media_generation, "Connection lost");462+ break;463+ }464+ Ok(Ok(_)) => {}465+ }466+ (467+ w,468+ h,469+ frame_buf,470+ is_keyframe,471+ codec,472+ None,473+ None,474+ None,475+ false,476+ false,477+ )478+ }479+ };480+ lock_client_media_metrics(&metrics).observe_transport_receive(481+ metrics_started.elapsed(),482+ sequence,483+ is_keyframe,484+ );485+486+ let sequence_gap = match sequence.zip(previous_sequence) {487+ Some((current, previous)) => match sequence_gap(previous, current) {488+ Ok(gap) => gap,489+ Err(error) => {490+ emit_frame_error(&app, media_generation, error);491+ break;492+ }493+ },494+ None => 0,495+ };496+ if sequence.is_some() {497+ previous_sequence = sequence;498+ }499+ lock_client_media_metrics(&metrics).observe_sequence_drop(sequence_gap);500+501+ if frontend_waiting_for_keyframe && !is_keyframe {502+ lock_client_media_metrics(&metrics).observe_frontend_resync_drop();503+ continue;504+ }505+ if !try_reserve_frame_channel_slot(&frame_events_in_flight) {506+ let begins_resync = !frontend_waiting_for_keyframe;507+ frontend_waiting_for_keyframe = true;508+ let mut metrics = lock_client_media_metrics(&metrics);509+ metrics.observe_frontend_queue_drop();510+ metrics.begin_frontend_resync(metrics_started.elapsed());511+ if begins_resync {512+ try_queue_media_keyframe_request(513+ media_control_requests.as_ref(),514+ KeyframeRequestReasonV3::FrontendBackpressure,515+ sequence,516+ );517+ }518+ continue;519+ }520+ lock_client_media_metrics(&metrics)521+ .observe_frontend_queue_depth(frame_events_in_flight.load(Ordering::SeqCst));522+523+ let delivered_to_frontend;524+ if use_webcodecs {525+ let envelope = match encode_frame_envelope(526+ FrameEnvelopeMetadata {527+ width: w,528+ height: h,529+ codec: &codec,530+ keyframe: is_keyframe,531+ discontinuity,532+ codec_config,533+ sequence,534+ capture_timestamp_micros,535+ pts_micros,536+ },537+ &frame_buf,538+ ) {539+ Ok(envelope) => envelope,540+ Err(error) => {541+ emit_frame_error(&app, media_generation, error);542+ release_frame_channel_slot(&frame_events_in_flight);543+ break;544+ }545+ };546+ let ipc_send_started = Instant::now();547+ let send_result = frame_channel.send(Response::new(envelope));548+ lock_client_media_metrics(&metrics).observe_frontend_ipc_send_duration(549+ metrics_started.elapsed(),550+ ipc_send_started.elapsed(),551+ );552+ if send_result.is_err() {553+ release_frame_channel_slot(&frame_events_in_flight);554+ break;555+ }556+ delivered_to_frontend = true;557+ } else if let Some(ref mut dec) = decoder {558+ let mut emitted = false;559+ for nal in nal_units(&frame_buf) {560+ if let Ok(Some(yuv)) = dec.decode(nal) {561+ let (yw, yh) = yuv.dimensions();562+ let rgb_len = yuv.rgb8_len();563+ let mut rgb_raw = vec![0u8; rgb_len];564+ yuv.write_rgb8(&mut rgb_raw);565+566+ let img = match image::RgbImage::from_raw(yw as u32, yh as u32, rgb_raw) {567+ Some(img) => img,568+ None => continue,569+ };570+ let mut jpeg_buf = Vec::with_capacity(30_000);571+ if image::DynamicImage::ImageRgb8(img)572+ .write_to(&mut Cursor::new(&mut jpeg_buf), image::ImageFormat::Jpeg)573+ .is_err()574+ {575+ continue;576+ }577+578+ let b64 = base64::engine::general_purpose::STANDARD.encode(&jpeg_buf);579+ let ipc_send_started = Instant::now();580+ let send_result = app.emit(581+ "frame",582+ FramePayload {583+ generation: media_generation,584+ width: yw as u32,585+ height: yh as u32,586+ data: b64,587+ keyframe: is_keyframe,588+ codec: codec.clone(),589+ capture_timestamp_micros,590+ pts_micros,591+ discontinuity,592+ },593+ );594+ lock_client_media_metrics(&metrics).observe_frontend_ipc_send_duration(595+ metrics_started.elapsed(),596+ ipc_send_started.elapsed(),597+ );598+ if send_result.is_err() {599+ release_frame_channel_slot(&frame_events_in_flight);600+ break 'frames;601+ }602+ emitted = true;603+ break;604+ }605+ }606+ if !emitted {607+ release_frame_channel_slot(&frame_events_in_flight);608+ }609+ delivered_to_frontend = emitted;610+ } else {611+ release_frame_channel_slot(&frame_events_in_flight);612+ delivered_to_frontend = false;613+ }614+615+ if delivered_to_frontend {616+ frontend_waiting_for_keyframe = false;617+ let mut metrics = lock_client_media_metrics(&metrics);618+ let elapsed = metrics_started.elapsed();619+ metrics.observe_frontend_send(elapsed);620+ metrics.finish_frontend_resync(elapsed);621+ }622+623+ tokio::task::yield_now().await;624+ }625+626+ let _ = stats_stop.send(true);627+ let _ = stats_task.await;628+ if media_transport == MediaTransport::UpstreamMoq {629+ retire_upstream_moq_generation(&app, media_generation, connected_audio_generation).await;630+ }631+ drop(endpoint);632+}633+634+fn sequence_gap(previous: u64, current: u64) -> Result<u64, String> {635+ let expected = previous636+ .checked_add(1)637+ .ok_or_else(|| format!("Media sequence overflowed after {previous}"))?;638+ if current < expected {639+ return Err(format!(640+ "Non-monotonic media sequence: previous={previous}, current={current}"641+ ));642+ }643+ Ok(current - expected)644+}645+646+647+mod tests {648+ use super::*;649+650+ 651+ fn sequence_checks_reject_duplicates_regressions_and_overflow() {652+ assert_eq!(sequence_gap(41, 42).unwrap(), 0);653+ assert_eq!(sequence_gap(41, 45).unwrap(), 3);654+ assert!(sequence_gap(41, 41).is_err());655+ assert!(sequence_gap(41, 40).is_err());656+ assert!(sequence_gap(u64::MAX, 0).is_err());657+ }658+}