zencodec_media/
staged.rs

1//! Finite video encoding without final-destination waits while the encoder is alive.
2//!
3//! The operation owns the encoder and actual prepared container/destination.
4//! Every frame is resolved through policy before conversion and native input.
5//! Encoded payloads use explicit memory or disk staging. Disk writes may block
6//! while the encoder owns raw state. Finish drains and drops
7//! the encoder before publishing any bytes. This intentionally stages the full
8//! compressed sequence; it is not a live streaming/backpressure mechanism.
9//! Caller-held pictures, codec/CMS workspace and bookkeeping are outside the
10//! payload budget. Unsupported timing/layout fails before the first frame.
11//! Optional borrowed Pulse arguments add per-call control. `None` leaves retained
12//! native controls, storage admission and required policy checks in force.
13//! ```
14//! use zencodec::{frame::Frame, operation::PolicySet, storage::StorageBudget};
15//! use zencodec_media::{container::Muxer, output::Forward, session::VideoEncoder,
16//!     staged::StagedEncode};
17//! fn encode<E: VideoEncoder, M: Muxer, W: std::io::Write>(
18//!     encoder: E, mux: M, writer: W, frames: impl IntoIterator<Item = Frame>,
19//! ) -> Result<(), Box<dyn core::error::Error>> {
20//!     let mut job = StagedEncode::new(encoder, mux, Forward(writer),
21//!         PolicySet::new(), StorageBudget::new(64 * 1024 * 1024, 4096), None)?;
22//!     for frame in frames { job.push(frame, None)?; }
23//!     let completed = job.finish(None)?;
24//!     let _ = (completed.stats, completed.report);
25//!     Ok(())
26//! }
27//! ```
28use crate::staging::Staging;
29use crate::{
30    container::Muxer,
31    output::{Destination, PreparedOutput, Stats},
32    packet::Packet,
33    session::{CodecConfig, Receive, SendStatus, VideoEncoder},
34    track::TrackInfo,
35};
36use alloc::boxed::Box;
37use core::{error::Error as StdError, fmt};
38
39use zencodec::{
40    At,
41    frame::{Frame, FrameInterpretation},
42    operation::{
43        FrameConversionOffer, FrameFacts, FrameRoute, FrameRouteOffer, FrameRouteState,
44        OperationPolicy, OperationReport, PolicySet, resolve_frame_route,
45    },
46    storage::StorageBudget,
47};
48
49/// Concrete source, native codec, conversion, policy or destination failure.
50pub type Cause = Box<dyn StdError + Send + Sync>;
51
52/// Boundary at which the owned operation stopped.
53#[derive(Clone, Copy, Debug, PartialEq, Eq)]
54#[non_exhaustive]
55pub enum Stage {
56    /// Prepare the container from the actual encoder configuration.
57    Prepare,
58    /// Resolve the held frame and conversion through application policy.
59    Policy,
60    /// Execute and validate the selected conversion.
61    Convert,
62    /// Send, receive or stage native encoded packets.
63    Encode,
64    /// Drain and release the encoder.
65    Finish,
66    /// Publish the container; partial effects remain in the cause.
67    Publish,
68    /// An operation already failed or unwound.
69    Terminal,
70}
71/// Failure with bounded policy history and a retained concrete cause.
72/// Default formatting omits arbitrary backend and application payloads.
73pub struct Failure {
74    /// Failed boundary.
75    pub stage: Stage,
76    /// Confirmed destination appends, excluding spool writes and patch overwrites.
77    /// Failed final patches can also leave partial overwrites; inspect the cause.
78    pub emitted: u64,
79    /// Decisions up to this failure.
80    pub report: Box<OperationReport>,
81    cause: Cause,
82}
83impl fmt::Debug for Failure {
84    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
85        f.debug_struct("StagedEncodeFailure")
86            .field("stage", &self.stage)
87            .field("emitted", &self.emitted)
88            .field("report", &self.report)
89            .finish_non_exhaustive()
90    }
91}
92impl fmt::Display for Failure {
93    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
94        write!(f, "staged encode failed during {:?}", self.stage)
95    }
96}
97impl StdError for Failure {
98    fn source(&self) -> Option<&(dyn StdError + 'static)> {
99        Some(self.cause.as_ref())
100    }
101}
102/// Typed refusal of the fixed-configuration staging contract.
103#[derive(Clone, Copy, Debug, PartialEq, Eq)]
104#[non_exhaustive]
105pub enum Refusal {
106    /// Encoder cannot declare its container configuration before input.
107    ConfigurationUnavailable,
108    /// Configuration does not describe video.
109    VideoConfigurationRequired,
110    /// Conversion's bounded facts disagree with its full target.
111    OfferMismatch,
112    /// Conversion returned different interpretation or timing.
113    ConversionMismatch,
114    /// Encoder promised pending output but did not return a packet.
115    ReceivePendingWithoutPacket,
116    /// Encoder ended before the caller finished input.
117    EarlyEnd,
118    /// Encoder changed the configuration or fidelity bound at preparation.
119    ConfigurationChanged,
120    /// Encoder requested input after end_input.
121    InputAfterEnd,
122    /// Encoder returned an empty or taken packet.
123    EmptyPacket,
124    /// Operation previously failed or unwound.
125    Terminal,
126    /// No input frame completed policy admission. Empty output is unsupported.
127    NoAdmittedFrames,
128    /// The legacy preset setter was called after input had been accepted.
129    PolicyChangedAfterInput,
130}
131impl fmt::Display for Refusal {
132    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
133        write!(f, "staged encode refusal: {self:?}")
134    }
135}
136impl StdError for Refusal {}
137fn invalid(reason: Refusal) -> Cause {
138    Box::new(reason)
139}
140fn check(pulse: &dyn zencodec::Pulse) -> Result<(), Cause> {
141    pulse
142        .check()
143        .map_err(|reason| At::wrap(zencodec::CodecError::new(None, reason.into())).into())
144}
145fn same_config(a: &CodecConfig, b: &CodecConfig) -> bool {
146    a.codec == b.codec
147        && a.private == b.private
148        && a.epoch == b.epoch
149        && a.time_base == b.time_base
150        && a.video == b.video
151}
152struct State<E, M, D> {
153    encoder: E,
154    output: PreparedOutput<M, D>,
155    config: CodecConfig,
156    fidelity: Option<zencodec::encode::Fidelity>,
157    packets: Staging,
158    previous: Option<FrameFacts>,
159}
160/// Completed output and its bounded application decision history.
161#[derive(Debug)]
162pub struct Completed {
163    /// Confirmed destination effects.
164    pub stats: Stats,
165    /// Decisions made before accepting frames.
166    pub report: OperationReport,
167}
168
169/// An owned, fixed-configuration video-to-container operation.
170///
171/// Failure or unwind makes it terminal and drops encoder, staged packets and
172/// private destination preparation. Push may write an explicit disk spool, but
173/// neither push nor drop writes the final destination. Finish
174/// consumes the operation and releases native state before the first destination write.
175/// No implicit segmentation, timing rewrite or fallback is performed.
176pub struct StagedEncode<E, M, D, R> {
177    state: Option<State<E, M, D>>,
178    resolver: R,
179    preset: PolicySet,
180    report: OperationReport,
181    policy_changed_after_input: bool,
182}
183impl<E: VideoEncoder, M: Muxer, D: Destination, R: OperationPolicy> StagedEncode<E, M, D, R> {
184    /// Retain the encoder and prepare exact-timing output from its actual config.
185    /// This single-video-track operation requires configuration before input.
186    /// Unknown explicit workspace bounds refuse before reading encoder config or
187    /// preparing the muxer. Supplied encoder/mux/store construction has already
188    /// happened and is outside this admission boundary.
189    pub fn new(
190        encoder: E,
191        mux: M,
192        destination: D,
193        resolver: R,
194        storage: StorageBudget,
195        pulse: Option<&dyn zencodec::Pulse>,
196    ) -> Result<Self, At<Failure>> {
197        Self::with_staging(
198            encoder,
199            mux,
200            destination,
201            resolver,
202            Staging::memory(storage),
203            pulse,
204        )
205    }
206    /// Retain an explicit compressed store. Disk staging may block while the
207    /// encoder is alive; policy sees the selected mechanism before accepting each frame.
208    pub fn with_staging(
209        encoder: E,
210        mux: M,
211        destination: D,
212        resolver: R,
213        staging: Staging,
214        pulse: Option<&dyn zencodec::Pulse>,
215    ) -> Result<Self, At<Failure>> {
216        let local_pulse = pulse.unwrap_or(&zencodec::NoPulse);
217        let mut stage = Stage::Prepare;
218        let prepare = || -> Result<_, Cause> {
219            check(local_pulse)?;
220            stage = Stage::Policy;
221            if let Some(required) = resolver.requirements() {
222                // No native/mux preparation bound is available. A later frame
223                // decision cannot authorize the allocations about to happen.
224                required.check_workspace(None)?;
225            }
226            stage = Stage::Prepare;
227            let config = encoder
228                .output_config()
229                .ok_or_else(|| invalid(Refusal::ConfigurationUnavailable))?
230                .clone();
231            let video = config
232                .video
233                .ok_or_else(|| invalid(Refusal::VideoConfigurationRequired))?;
234            let mut track = TrackInfo::video(0, config.codec.clone(), config.time_base, video)
235                .with_config_epoch(config.epoch);
236            if let Some(private) = &config.private {
237                track = track.with_codec_private(private.clone());
238            }
239            let output = PreparedOutput::new(mux, destination, &[track.build()?], pulse)?;
240            let fidelity = encoder.output_fidelity();
241            Ok(State {
242                encoder,
243                fidelity,
244                output,
245                config,
246                packets: staging,
247                previous: None,
248            })
249        };
250        let state = prepare().map_err(|cause| {
251            At::wrap(Failure {
252                stage,
253                emitted: 0,
254                report: Box::default(),
255                cause,
256            })
257        })?;
258        Ok(Self {
259            state: Some(state),
260            resolver,
261            preset: PolicySet::new(),
262            report: OperationReport::default(),
263            policy_changed_after_input: false,
264        })
265    }
266    /// Preset consulted when the resolver returns `UsePreset`.
267    /// During migration, these requirements and the policy argument
268    /// both apply. A bound policy supplies preferences in place of these.
269    /// Call before the first accepted frame. Calling it afterward makes the
270    /// next push or finish refuse without publishing the staged output.
271    #[deprecated(note = "Pass PolicySet or BoundPolicy as the operation policy argument")]
272    pub fn with_policy_set(mut self, preset: PolicySet) -> Self {
273        self.policy_changed_after_input |= self
274            .state
275            .as_ref()
276            .is_some_and(|state| state.previous.is_some());
277        self.preset = preset;
278        self
279    }
280    /// Bounded decisions, including after a terminal failure.
281    pub fn report(&self) -> &OperationReport {
282        &self.report
283    }
284    /// Whether failure or unwind has made this operation unusable.
285    pub fn is_failed(&self) -> bool {
286        self.state.is_none()
287    }
288    fn failure(&self, stage: Stage, cause: Cause) -> At<Failure> {
289        self.failure_after(stage, cause, 0)
290    }
291    fn failure_after(&self, stage: Stage, cause: Cause, emitted: u64) -> At<Failure> {
292        let emitted = zencodec::find_cause::<crate::output::Failure>(cause.as_ref())
293            .map_or(emitted, |failure| failure.emitted);
294        At::wrap(Failure {
295            stage,
296            emitted,
297            report: Box::new(self.report.clone()),
298            cause,
299        })
300    }
301    /// Send the exact held picture without conversion or timing changes.
302    /// Prepared encoder fidelity is disclosed; native workspace remains unqualified.
303    pub fn push(
304        &mut self,
305        input: Frame,
306        pulse: Option<&dyn zencodec::Pulse>,
307    ) -> Result<(), At<Failure>> {
308        let target = input.interpretation().clone();
309        let mut offer = FrameConversionOffer::new(FrameFacts::new(&target, input.timing()));
310        offer.preserves_samples = true;
311        offer.preserves_extended_range = true;
312        self.push_converted(input, &target, offer, Ok, pulse)
313    }
314    /// Retain this input and selected converter through policy, then execute once.
315    /// `target` is the complete required interpretation; timing comes from the
316    /// offer. Output must match both exactly before reaching the encoder. The
317    /// closure owns its prepared conversion/reservations; it must not publish.
318    pub fn push_converted(
319        &mut self,
320        input: Frame,
321        target: &FrameInterpretation,
322        conversion: FrameConversionOffer,
323        execute: impl FnOnce(Frame) -> Result<Frame, Cause>,
324        pulse: Option<&dyn zencodec::Pulse>,
325    ) -> Result<(), At<Failure>> {
326        let local_pulse = pulse.unwrap_or(&zencodec::NoPulse);
327        let mut state = self
328            .state
329            .take()
330            .ok_or_else(|| self.failure(Stage::Terminal, invalid(Refusal::Terminal)))?;
331        let mut stage = Stage::Policy;
332        let run = || -> Result<(), Cause> {
333            check(local_pulse)?;
334            if self.policy_changed_after_input {
335                return Err(invalid(Refusal::PolicyChangedAfterInput));
336            }
337            if state.encoder.output_fidelity() != state.fidelity {
338                return Err(invalid(Refusal::ConfigurationChanged));
339            }
340            if conversion.encoder_input != FrameFacts::new(target, conversion.encoder_input.timing)
341            {
342                return Err(invalid(Refusal::OfferMismatch));
343            }
344            let mut offer = FrameRouteOffer::new(if state.previous.is_none() {
345                FrameRoute::BeginSegment
346            } else {
347                FrameRoute::Continue
348            });
349            offer.conversion = Some(conversion);
350            offer.fidelity = state.fidelity;
351            offer.destination = Some(state.output.facts());
352            offer.admitted_encoded_bytes = state.packets.admitted_memory();
353            offer.staging = Some(state.packets.facts());
354            offer.output_schedule = Some(zencodec::operation::OutputScheduleOffer::new(
355                zencodec::operation::OutputSchedule::FinishFirst,
356            ));
357            let mut progress = FrameRouteState::default();
358            progress.previous = state.previous;
359            match resolve_frame_route(
360                input.interpretation(),
361                input.timing(),
362                progress,
363                &[offer],
364                &mut self.resolver,
365                self.preset,
366                pulse,
367            ) {
368                Ok((_, record)) => self.report.record(record),
369                Err(error) => {
370                    self.report.record(error.error().record());
371                    return Err(error.into());
372                }
373            }
374            stage = Stage::Convert;
375            let previous = FrameFacts::new(input.interpretation(), input.timing());
376            let converted = execute(input)?;
377            check(local_pulse)?;
378            if converted.interpretation().as_ref() != target
379                || converted.timing() != conversion.encoder_input.timing
380            {
381                return Err(invalid(Refusal::ConversionMismatch));
382            }
383            stage = Stage::Encode;
384            loop {
385                check(local_pulse)?;
386                match state.encoder.send(&converted, pulse)? {
387                    SendStatus::Accepted => break,
388                    SendStatus::ReceivePending => match state.encoder.receive(pulse)? {
389                        Receive::Ready(packet) => {
390                            stage_packet(&mut state.packets, packet, &state.output, local_pulse)?
391                        }
392                        _ => return Err(invalid(Refusal::ReceivePendingWithoutPacket)),
393                    },
394                }
395            }
396            loop {
397                check(local_pulse)?;
398                match state.encoder.receive(pulse)? {
399                    Receive::Ready(packet) => {
400                        stage_packet(&mut state.packets, packet, &state.output, local_pulse)?
401                    }
402                    Receive::NeedInput => break,
403                    Receive::EndOfStream => return Err(invalid(Refusal::EarlyEnd)),
404                }
405            }
406            if !state
407                .encoder
408                .output_config()
409                .is_some_and(|config| same_config(config, &state.config))
410                || state.encoder.output_fidelity() != state.fidelity
411            {
412                return Err(invalid(Refusal::ConfigurationChanged));
413            }
414            state.previous = Some(previous);
415            check(local_pulse)
416        };
417        run().map_err(|cause| self.failure(stage, cause))?;
418        self.state = Some(state);
419        Ok(())
420    }
421    /// Drain, release all owned encoder state, then publish the retained output.
422    /// Writer failure may leave partial bytes; its concrete output failure and
423    /// confirmed append count remain in the error chain. No retry is performed.
424    /// An operation with no admitted input frames refuses before draining;
425    /// empty-container publication has no supported admission route.
426    pub fn finish(mut self, pulse: Option<&dyn zencodec::Pulse>) -> Result<Completed, At<Failure>> {
427        let local_pulse = pulse.unwrap_or(&zencodec::NoPulse);
428        let mut state = self
429            .state
430            .take()
431            .ok_or_else(|| self.failure(Stage::Terminal, invalid(Refusal::Terminal)))?;
432        let mut drain = || -> Result<(), Cause> {
433            check(local_pulse)?;
434            if self.policy_changed_after_input {
435                return Err(invalid(Refusal::PolicyChangedAfterInput));
436            }
437            if state.previous.is_none() {
438                return Err(invalid(Refusal::NoAdmittedFrames));
439            }
440            state.encoder.end_input()?;
441            loop {
442                check(local_pulse)?;
443                match state.encoder.receive(pulse)? {
444                    Receive::Ready(packet) => {
445                        stage_packet(&mut state.packets, packet, &state.output, local_pulse)?
446                    }
447                    Receive::EndOfStream => break,
448                    Receive::NeedInput => {
449                        return Err(invalid(Refusal::InputAfterEnd));
450                    }
451                }
452            }
453            if !state
454                .encoder
455                .output_config()
456                .is_some_and(|config| same_config(config, &state.config))
457                || state.encoder.output_fidelity() != state.fidelity
458            {
459                return Err(invalid(Refusal::ConfigurationChanged));
460            }
461            check(local_pulse)
462        };
463        drain().map_err(|cause| self.failure(Stage::Finish, cause))?;
464        let State {
465            encoder,
466            output,
467            mut packets,
468            ..
469        } = state;
470        drop(encoder);
471        let mut emitted = 0;
472        let publish = || -> Result<Stats, Cause> {
473            packets.begin_replay(local_pulse)?;
474            // Admit/read the first packet before any final destination effects.
475            let mut packet = packets.next(local_pulse)?;
476            let mut output = output.publish(pulse)?;
477            emitted = output.stats().emitted;
478            while let Some(held) = packet {
479                output.send(held, pulse)?;
480                emitted = output.stats().emitted;
481                packet = packets.next(local_pulse)?;
482            }
483            Ok(output.finish(pulse)?)
484        };
485        let stats =
486            publish().map_err(|cause| self.failure_after(Stage::Publish, cause, emitted))?;
487        Ok(Completed {
488            stats,
489            report: self.report,
490        })
491    }
492}
493fn stage_packet<M: Muxer, D: Destination>(
494    packets: &mut Staging,
495    packet: Packet,
496    output: &PreparedOutput<M, D>,
497    pulse: &dyn zencodec::Pulse,
498) -> Result<(), Cause> {
499    output.validate_packet(&packet)?;
500    if packet.is_empty() {
501        return Err(invalid(Refusal::EmptyPacket));
502    }
503    packets.push(packet, pulse)
504}
zencodec_media/staged.rs

zencodec_media/
staged.rs
