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

heddle_api/
v2.rs
