pocketstation 1.0.0

Source-aware desktop audio Session SDK
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108

//! Deterministic continuity validation for source-aware signals.

use crate::graph::signal::{SignalEnvelope, SignalEnvelopeError, SignalLineage, SignalTiming};

#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct SignalContinuityObservation {
    pub discontinuity_observed: bool,
    pub source_recovered: bool,
    pub policy_changed: bool,
}

#[derive(Debug, Default)]
pub struct SignalContinuityTracker {
    previous: Option<(SignalLineage, SignalTiming)>,
}

impl SignalContinuityTracker {
    pub fn observe(
        &mut self,
        envelope: &SignalEnvelope,
    ) -> Result<SignalContinuityObservation, SignalContinuityError> {
        envelope
            .validate()
            .map_err(SignalContinuityError::InvalidEnvelope)?;
        let current = envelope
            .lineage()
            .ok_or(SignalContinuityError::MissingLineage)?;
        let current_timing = envelope.timing();
        if let Some((previous, previous_timing)) = self.previous {
            if current.session_id != previous.session_id
                || current.stream_id != previous.stream_id
                || current.source_id != previous.source_id
                || current.clock_id != previous.clock_id
            {
                return Err(SignalContinuityError::IdentityChanged);
            }
            if current.discontinuity_epoch < previous.discontinuity_epoch {
                return Err(SignalContinuityError::DiscontinuityRegressed);
            }
            if current.source_generation < previous.source_generation {
                return Err(SignalContinuityError::GenerationRegressed);
            }
            if current.policy_epoch < previous.policy_epoch {
                return Err(SignalContinuityError::PolicyRegressed);
            }
            let discontinuity_observed = current.discontinuity_epoch > previous.discontinuity_epoch;
            if !discontinuity_observed
                && current.sequence_number != previous.sequence_number.saturating_add(1)
            {
                return Err(SignalContinuityError::SequenceGapWithoutDiscontinuity);
            }
            let source_recovered = current.source_generation > previous.source_generation;
            if source_recovered && !discontinuity_observed {
                return Err(SignalContinuityError::RecoveryWithoutDiscontinuity);
            }
            if timestamp_regressed(previous_timing, current_timing) {
                return Err(SignalContinuityError::TimestampRegression);
            }
            let observation = SignalContinuityObservation {
                discontinuity_observed,
                source_recovered,
                policy_changed: current.policy_epoch > previous.policy_epoch,
            };
            self.previous = Some((current, current_timing));
            return Ok(observation);
        }
        self.previous = Some((current, current_timing));
        Ok(SignalContinuityObservation {
            discontinuity_observed: false,
            source_recovered: false,
            policy_changed: false,
        })
    }
}

fn timestamp_regressed(previous: SignalTiming, current: SignalTiming) -> bool {
    current.observed_timestamp_ns < previous.observed_timestamp_ns
        || previous
            .source_timestamp_ns
            .zip(current.source_timestamp_ns)
            .is_some_and(|(previous, current)| current < previous)
        || previous
            .session_timestamp_ns
            .zip(current.session_timestamp_ns)
            .is_some_and(|(previous, current)| current < previous)
}

#[derive(Debug, Clone, Copy, PartialEq, Eq, thiserror::Error)]
pub enum SignalContinuityError {
    #[error("signal envelope is invalid: {0}")]
    InvalidEnvelope(SignalEnvelopeError),
    #[error("signal continuity requires source-independent lineage")]
    MissingLineage,
    #[error("signal identity changed within one continuity tracker")]
    IdentityChanged,
    #[error("signal sequence gap occurred without a discontinuity epoch change")]
    SequenceGapWithoutDiscontinuity,
    #[error("signal timestamp regressed")]
    TimestampRegression,
    #[error("signal discontinuity epoch regressed")]
    DiscontinuityRegressed,
    #[error("signal source generation regressed")]
    GenerationRegressed,
    #[error("signal source recovery occurred without a discontinuity")]
    RecoveryWithoutDiscontinuity,
    #[error("signal policy epoch regressed")]
    PolicyRegressed,
}