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

heddle_api/
v2.rs
