heddle_api/
v2.rs

1//! Shared v2 client behavior. Transport adapters retain key and connection ownership.
2pub mod account_metadata;
3pub mod approval_groups;
4pub mod client;
5pub mod collaboration;
6pub mod content;
7pub mod custodial_recovery;
8pub mod identity_management;
9pub mod invitation;
10pub mod invitation_code;
11pub mod membership_floor;
12pub mod notifications;
13pub mod passkey_label;
14pub mod people;
15pub mod read_budget;
16pub mod spool_settings;
17use crate::StreamingShape;
18use crate::heddle::api::common::{
19    AuthorizationAccess, AuthorizationExistence, AuthorizationRole, AuthorizationScopeSource,
20    CallContext, DeploymentTarget, RetryBehavior, RpcEffect, ServiceMaturity, SigningTier,
21};
22use crate::heddle::api::v1alpha2::{StreamDataKind, StreamFrame, stream_frame};
23pub use read_budget::{
24    GUARANTEED_READ_BUDGET, ReadBudgetError, negotiate_read_budget, validate_accepted_read_budget,
25};
26
27include!(concat!(env!("OUT_DIR"), "/heddle_api_v2_methods.rs"));
28
29/// Generated resource guard selected by a protobuf request field path.
30#[derive(Clone, Copy, Debug, Eq, PartialEq)]
31pub struct AuthorizationTarget {
32    pub path: &'static str,
33    pub role: AuthorizationRole,
34}
35
36/// Complete method authorization declaration. Hosts must resolve every
37/// populated target and enforce the role against current authority; metadata
38/// describes that obligation and is never itself a capability.
39#[derive(Clone, Copy, Debug, Eq, PartialEq)]
40pub struct AuthorizationPolicy {
41    pub role: AuthorizationRole,
42    pub scope_source: AuthorizationScopeSource,
43    pub existence: AuthorizationExistence,
44    pub targets: &'static [AuthorizationTarget],
45}
46
47/// Decoded transport context bound to a v2 route, before body interpretation.
48#[derive(Debug)]
49pub struct RoutedCall<'a> {
50    pub method: &'static MethodDescriptor,
51    pub context: &'a CallContext,
52}
53
54impl<'a> RoutedCall<'a> {
55    pub fn new(path: &str, context: &'a CallContext) -> Option<Self> {
56        method_descriptor(path).map(|method| Self { method, context })
57    }
58}
59
60impl MethodDescriptor {
61    /// Only safe reads can be sent on replayable 0-RTT connections.
62    pub const fn allows_zero_rtt(&self) -> bool {
63        matches!(self.effect, RpcEffect::ReadOnly)
64            && matches!(self.retry_behavior, RetryBehavior::Safe)
65    }
66    /// Extract the operation ID for the transport context from the exact request
67    /// bytes. Adapters must not maintain a second route/field-number inventory.
68    pub fn client_operation_id<'a>(
69        &self,
70        request: &'a [u8],
71    ) -> Result<Option<&'a str>, crate::RequestMetadataError> {
72        self.client_operation_id_field_number
73            .map(|field| crate::transport::protobuf_string_field(request, field))
74            .transpose()
75            .map(Option::flatten)
76    }
77}
78
79/// Upper bound for an opaque observation cursor; independent of payload limits.
80pub const MAX_CURSOR_BYTES: usize = 4096;
81
82/// Invalid stream framing must never advance a client's durable observation.
83#[derive(Clone, Copy, Debug, PartialEq, Eq, thiserror::Error)]
84pub enum StreamProtocolError {
85    #[error("non-contiguous stream sequence")]
86    Sequence,
87    #[error("stream query binding mismatch")]
88    Binding,
89    #[error("stream resumed from an unexpected cursor")]
90    Resume,
91    #[error("invalid stream phase")]
92    Phase,
93    #[error("payload does not match the frame kind")]
94    Payload,
95    #[error("invalid or oversized checkpoint cursor")]
96    Cursor,
97    #[error("missing or invalid accepted ReadBudget")]
98    Budget,
99}
100
101#[derive(Clone, Copy, Debug, PartialEq, Eq)]
102enum Phase {
103    Opening,
104    Snapshot,
105    Live,
106    Reset,
107    Complete,
108}
109
110/// Instructions for the typed view reducer. Apply staged data atomically at
111/// Commit, and persist the cursor only after that reducer succeeds.
112#[derive(Clone, Copy, Debug, PartialEq, Eq)]
113pub enum ObservationAction {
114    BeginSnapshot,
115    Resumed,
116    Stage(StreamDataKind),
117    Commit,
118    Heartbeat,
119    Reset,
120    Complete,
121}
122
123#[derive(Debug, thiserror::Error)]
124pub enum ObservationApplyError<E> {
125    #[error("stream protocol failed: {0}")]
126    Protocol(#[from] StreamProtocolError),
127    #[error("view reducer failed: {0}")]
128    Reducer(E),
129}
130
131/// Cursor and lifecycle tracker; typed consumers own staged view data.
132#[derive(Clone)]
133pub struct ObservationState {
134    binding_digest: [u8; 32],
135    cursor: Vec<u8>,
136    sequence: u64,
137    phase: Phase,
138    pending: bool,
139}
140
141impl ObservationState {
142    /// Advances this tracker only after the view reducer succeeds. At Commit,
143    /// the reducer must atomically apply staged data and persist the supplied
144    /// cursor. A failed reducer leaves the tracker at its previous checkpoint.
145    pub async fn apply<E, F, Fut>(
146        &mut self,
147        frame: &StreamFrame,
148        has_payload: bool,
149        reducer: F,
150    ) -> Result<ObservationAction, ObservationApplyError<E>>
151    where
152        F: FnOnce(ObservationAction, Vec<u8>) -> Fut,
153        Fut: std::future::Future<Output = Result<(), E>>,
154    {
155        let mut next = self.clone();
156        let action = next.accept(frame, has_payload)?;
157        reducer(action, next.cursor.clone())
158            .await
159            .map_err(ObservationApplyError::Reducer)?;
160        *self = next;
161        Ok(action)
162    }
163
164    pub fn new(binding_digest: [u8; 32], cursor: Vec<u8>) -> Self {
165        Self {
166            binding_digest,
167            cursor,
168            sequence: 0,
169            phase: Phase::Opening,
170            pending: false,
171        }
172    }
173
174    pub fn cursor(&self) -> &[u8] {
175        &self.cursor
176    }
177
178    pub fn is_complete(&self) -> bool {
179        self.phase == Phase::Complete
180    }
181
182    /// Validate a frame before changing lifecycle state. A rejected frame leaves
183    /// the state unchanged. Transport adapters enforce negotiated byte budgets
184    /// before decoding; callers validate the method-specific typed payload.
185    ///
186    /// For reducers that can fail, run this on a clone, apply the returned action,
187    /// then replace the original tracker only if the reducer succeeds.
188    pub fn accept(
189        &mut self,
190        frame: &StreamFrame,
191        has_payload: bool,
192    ) -> Result<ObservationAction, StreamProtocolError> {
193        use stream_frame::Body;
194        if matches!(self.phase, Phase::Reset | Phase::Complete) {
195            return Err(StreamProtocolError::Phase);
196        }
197        if self.sequence.checked_add(1) != Some(frame.sequence) {
198            return Err(StreamProtocolError::Sequence);
199        }
200        let body = frame.body.as_ref().ok_or(StreamProtocolError::Phase)?;
201        if has_payload != matches!(body, Body::Data(_)) {
202            return Err(StreamProtocolError::Payload);
203        }
204        let action = match body {
205            Body::Open(open) => {
206                if self.phase != Phase::Opening {
207                    return Err(StreamProtocolError::Phase);
208                }
209                if open.binding_digest.as_slice() != self.binding_digest {
210                    return Err(StreamProtocolError::Binding);
211                }
212                if open.resumed_from != self.cursor {
213                    return Err(StreamProtocolError::Resume);
214                }
215                if self.cursor.len() > MAX_CURSOR_BYTES {
216                    return Err(StreamProtocolError::Cursor);
217                }
218                validate_accepted_read_budget(&Default::default(), open.accepted_budget.as_ref())
219                    .map_err(|_| StreamProtocolError::Budget)?;
220                if self.cursor.is_empty() {
221                    self.phase = Phase::Snapshot;
222                    ObservationAction::BeginSnapshot
223                } else {
224                    self.phase = Phase::Live;
225                    ObservationAction::Resumed
226                }
227            }
228            Body::Data(data) => {
229                let kind =
230                    StreamDataKind::try_from(data.kind).map_err(|_| StreamProtocolError::Phase)?;
231                let valid = match self.phase {
232                    Phase::Snapshot => kind == StreamDataKind::Snapshot,
233                    Phase::Live => matches!(kind, StreamDataKind::Upsert | StreamDataKind::Remove),
234                    _ => false,
235                };
236                if !valid {
237                    return Err(StreamProtocolError::Phase);
238                }
239                self.pending = true;
240                ObservationAction::Stage(kind)
241            }
242            Body::Checkpoint(checkpoint) => {
243                if !matches!(self.phase, Phase::Snapshot | Phase::Live)
244                    || checkpoint.snapshot_complete != (self.phase == Phase::Snapshot)
245                {
246                    return Err(StreamProtocolError::Phase);
247                }
248                if checkpoint.previous_cursor != self.cursor
249                    || checkpoint.cursor.is_empty()
250                    || checkpoint.cursor.len() > MAX_CURSOR_BYTES
251                    || checkpoint.cursor == self.cursor
252                {
253                    return Err(StreamProtocolError::Cursor);
254                }
255                self.cursor.clone_from(&checkpoint.cursor);
256                self.phase = Phase::Live;
257                self.pending = false;
258                ObservationAction::Commit
259            }
260            Body::Reset(_) => {
261                self.phase = Phase::Reset;
262                self.cursor.clear();
263                self.pending = false;
264                ObservationAction::Reset
265            }
266            Body::Complete(complete) => {
267                if self.phase != Phase::Live || self.pending {
268                    return Err(StreamProtocolError::Phase);
269                }
270                if complete.cursor != self.cursor {
271                    return Err(StreamProtocolError::Cursor);
272                }
273                self.phase = Phase::Complete;
274                ObservationAction::Complete
275            }
276            Body::Heartbeat(_) => {
277                if self.phase == Phase::Opening {
278                    return Err(StreamProtocolError::Phase);
279                }
280                ObservationAction::Heartbeat
281            }
282        };
283        self.sequence = frame.sequence;
284        Ok(action)
285    }
286}
heddle_api/v2.rs

heddle_api/
v2.rs
