wip

Author: wkozyra95

smelter-core/CLAUDE.md

@@ -30,11 +30,30 @@ Pipeline orchestration: `Pipeline` in `./src/pipeline/instance.rs`. Shared state
   - RTP, RTMP client, MP4, HLS, WHIP, WHEP (WebRTC egress), RawData/EncodedData (channel-based)
   - Example: `RtpOutput` `./src/pipeline/rtp/rtp_output.rs`
 
-- **Protocol support** — `./src/pipeline/rtp/` (payloader/depayloader/jitter buffer), `./src/pipeline/webrtc/server.rs` (WHIP/WHEP HTTP endpoints), `./src/pipeline/utils/` (H.264 format conversions)
-
-
 ## Multimedia data flow
 
 Input (demuxer) → Decoder → Queue → Rendering/AudioMixer → Encoder → Output (muxer) 
 
 In most cases, each element spawns at least one thread and communicates with other elements via channels.
+
+## Timestamps
+
+- Inputs always start PTS from `queue_sync_point.elapsed()` when first packet arrives. If ahead of time processing is enabled it might lead to unexpected behaviors.
+- Input is responsible for maintaining continuity of the stream until it sends EOS event. After EOS, it is expected to synchronize with `queue_sync_point.elspased()` again.
+- Queue handles input offsets. If input does not have offset then timestamps from inputs are preserved. If input has an offset then queue calculates the value based on first packet (from audio or video track).
+- Queue handles pausing inputs. Pause offset is calculated based on difference between `queue_sync_point.elapsed()` on pause and resume calls.
+- When `ahead_of_time_processing` is enabled queue might be processing PTS value a lot larger than `queue_sync_point.elapsed()`
+
+## Alternative approach
+
+- Inputs always start from some epoch.
+- Queue synchronizes on the first packet
+  - How audio and video sync should work?
+    - Eos needs and generation number
+    - On each track audio and video synchronizes for current epoch
+- Sending Reset event will reset timestamps
+- Pause 
+- Seek
+  - Send Reset
+  - Start stream
+

smelter-core/src/queue/audio_queue.rs

@@ -83,18 +83,17 @@ impl AudioQueue {
     }
 
     pub fn pause_input(&mut self, input_id: &InputId, pts: Duration) {
-        if let Some(input) = self.inputs.get_mut(input_id) {
-            if input.pause_state.pause(pts) {
-                self.event_emitter
-                    .emit(Event::AudioInputStreamPaused(input_id.clone()));
-            }
+        if let Some(input) = self.inputs.get_mut(input_id)
+            && input.pause_state.pause(pts, &input.state)
+        {
+            self.event_emitter
+                .emit(Event::AudioInputStreamPaused(input_id.clone()));
         }
     }
 
     pub fn resume_input(&mut self, input_id: &InputId, pts: Duration) {
         if let Some(input) = self.inputs.get_mut(input_id) {
-            let first_pts_received = input.shared_state.first_pts().is_some();
-            if input.pause_state.resume(pts, first_pts_received) && first_pts_received {
+            if input.pause_state.resume(pts, &input.state) {
                 self.event_emitter
                     .emit(Event::AudioInputStreamPlaying(input_id.clone()));
             }
@@ -281,7 +280,7 @@ impl AudioQueueInput {
     /// queue start.
     fn drop_old_samples_before_start(&mut self) {
         loop {
-            if let QueueState::Draining = self.state  {
+            if let QueueState::Draining = self.state {
                 self.reset_after_eos();
             }
             // if offset is defined try_enqueue_frame will always return err

smelter-core/src/queue/utils.rs

@@ -34,6 +34,7 @@ impl EmitOnceGuard {
     }
 }
 
+#[derive(PartialEq, Eq, Clone, Copy)]
 pub enum QueueState {
     WaitingForFirstPacket,
     Running,
@@ -43,7 +44,6 @@ pub enum QueueState {
 }
 
 pub struct PauseState {
-    paused: bool,
     /// Internal PTS (relative to sync_point) when input was paused.
     paused_at_pts: Option<Duration>,
     /// Accumulated pause duration to add to frame/sample PTS after resume.
@@ -53,36 +53,48 @@ pub struct PauseState {
 impl PauseState {
     pub fn new() -> Self {
         Self {
-            paused: false,
             paused_at_pts: None,
             pts_offset: Duration::ZERO,
         }
     }
 
-    /// Returns `true` if the state actually changed (was not already paused).
+    /// Sets paused state. Returns `true` if the pause state was changed
     pub fn pause(&mut self, pts: Duration) -> bool {
-        // Make pause idempotent: if we're already paused, do not update the
-        // original pause start timestamp to avoid distorting pts_offset.
-        if self.paused {
+        if self.paused_at_pts.is_some() {
             return false;
         }
-        self.paused = true;
         self.paused_at_pts = Some(pts);
         true
     }
 
-    /// Returns `true` if the state actually changed (was paused before).
-    pub fn resume(&mut self, pts: Duration, first_pts_received: bool) -> bool {
-        if !self.paused {
+    /// Clears paused state. Returns `true` if pause state was changed
+    pub fn resume(&mut self, pts: Duration, state: QueueState) -> bool {
+        if self.paused_at_pts.is_none() {
             return false;
         }
-        self.paused = false;
-        if let Some(pause_start) = self.paused_at_pts.take()
-            && first_pts_received
-        {
-            self.pts_offset += pts.saturating_sub(pause_start);
+        if let Some(pause_start) = self.paused_at_pts.take() {
+            match state {
+                QueueState::WaitingForFirstPacket => {
+                    // Input without offset
+                    // If input already resolved elapsed time before
+                    //  - we need to add offset
+                    // If input did not resolve elapse time yet
+                    //  - 
+                    // If input resolved elapsed time during pause
+                    //  - we need to know where
+                    //
+                },
+                QueueState::Running => {
+                    self.pts_offset += pts.saturating_sub(pause_start);
+                }
+                QueueState::Draining => {
+                    // We will clear the queue, send EOS on the next loop
+                    // and state will be reset, so offset does not matter
+                    // after this
+                },
+            }
         }
-        true
+        return true;
     }
 
     pub fn is_paused(&self) -> bool {
@@ -92,4 +104,8 @@ impl PauseState {
     pub fn pts_offset(&self) -> Duration {
         self.pts_offset
     }
+
+    pub fn paused_at_pts(&self) -> Option<Duration> {
+        self.paused_at_pts
+    }
 }

smelter-core/src/queue/video_queue.rs

@@ -57,8 +57,7 @@ impl VideoQueue {
 
                 offset_from_start: opts.offset,
 
-                pause_state: PauseState::new(),
-                paused_last_frame: None,
+                pause_state: VideoPauseState::new(),
 
                 state: QueueState::WaitingForFirstPacket,
 
@@ -83,23 +82,22 @@ impl VideoQueue {
     }
 
     pub fn pause_input(&mut self, input_id: &InputId, pts: Duration) {
-        if let Some(input) = self.inputs.get_mut(input_id) {
-            input.paused_last_frame = input.queue.front().cloned();
-            if input.pause_state.pause(pts) {
-                self.event_emitter
-                    .emit(Event::VideoInputStreamPaused(input_id.clone()));
-            }
+        if let Some(input) = self.inputs.get_mut(input_id)
+            && input
+                .pause_state
+                .pause(pts, &input.state, input.queue.front().cloned())
+        {
+            self.event_emitter
+                .emit(Event::VideoInputStreamPaused(input_id.clone()));
         }
     }
 
     pub fn resume_input(&mut self, input_id: &InputId, pts: Duration) {
         if let Some(input) = self.inputs.get_mut(input_id) {
-            let first_pts_received = input.shared_state.first_pts().is_some();
-            if input.pause_state.resume(pts, first_pts_received) && first_pts_received {
+            if input.pause_state.resume(pts, &input.state) {
                 self.event_emitter
                     .emit(Event::VideoInputStreamPlaying(input_id.clone()));
             }
-            input.paused_last_frame = None;
             input.queue.clear();
         }
     }
@@ -176,6 +174,52 @@ struct FrameEvent {
     event: PipelineEvent<Frame>,
 }
 
+struct VideoPauseState {
+    inner: PauseState,
+    paused_frame: Option<Frame>,
+}
+
+impl VideoPauseState {
+    fn new() -> Self {
+        Self {
+            inner: PauseState::new(),
+            paused_frame: None,
+        }
+    }
+
+    fn pause(&mut self, pts: Duration, state: &QueueState, frame: Option<Frame>) -> bool {
+        let was_paused = self.inner.is_paused();
+        let should_emit = self.inner.pause(pts, state);
+        if !was_paused {
+            self.paused_frame = frame;
+        }
+        should_emit
+    }
+
+    fn resume(&mut self, pts: Duration, state: &QueueState) -> bool {
+        self.paused_frame = None;
+        self.inner.resume(pts, state)
+    }
+
+    /// Returns the paused frame as a PipelineEvent with PTS shifted by time elapsed since pause.
+    fn paused_frame(&self, buffer_pts: Duration) -> Option<PipelineEvent<Frame>> {
+        self.paused_frame.clone().map(|mut frame| {
+            if let Some(paused_at) = self.inner.paused_at_pts() {
+                frame.pts += buffer_pts.saturating_sub(paused_at);
+            }
+            PipelineEvent::Data(frame)
+        })
+    }
+
+    fn is_paused(&self) -> bool {
+        self.inner.is_paused()
+    }
+
+    fn pts_offset(&self) -> Duration {
+        self.inner.pts_offset()
+    }
+}
+
 pub struct VideoQueueInput {
     /// Frames are PTS ordered where PTS=0 represents beginning of the stream.
     queue: VecDeque<Frame>,
@@ -192,10 +236,7 @@ pub struct VideoQueueInput {
     sync_point: Instant,
     shared_state: SharedState,
 
-    pause_state: PauseState,
-    /// Last frame captured at the moment of pause. Returned with updated PTS
-    /// on every `get_frame` call while paused.
-    paused_last_frame: Option<Frame>,
+    pause_state: VideoPauseState,
 
     state: QueueState,
 
@@ -209,12 +250,9 @@ impl VideoQueueInput {
     /// whether stream is required or not.
     fn get_frame(&mut self, buffer_pts: Duration, queue_start_pts: Duration) -> Option<FrameEvent> {
         if self.pause_state.is_paused() {
-            return self.paused_last_frame.clone().map(|mut frame| {
-                frame.pts = buffer_pts;
-                FrameEvent {
-                    required: self.required,
-                    event: PipelineEvent::Data(frame),
-                }
+            return self.pause_state.paused_frame(buffer_pts).map(|event| FrameEvent {
+                required: self.required,
+                event,
             });
         }
 
@@ -345,7 +383,7 @@ impl VideoQueueInput {
     /// queue start.
     fn drop_old_frames_before_start(&mut self) {
         loop {
-            if let QueueState::Draining = self.state  {
+            if let QueueState::Draining = self.state {
                 self.reset_after_eos();
             }
             // if offset is defined try_enqueue_frame will always return err