aequora_protocol/
lib.rs

1//! Stable wire data-transfer objects for synchronization exchanges.
2
3use aequora_types::{
4    ActorId, AuthorityEpoch, AuthorityId, Cursor, DeviceId, EntityRef, EntityVersion, EventId,
5    HybridTimestamp, LineageContext, OperationId, ProtocolVersion, RegionId, RequestId,
6    SchemaVersion, Sequence, SessionId, SnapshotId, SyncScopeId, TenantId,
7};
8use serde::{Deserialize, Serialize};
9use smallvec::SmallVec;
10
11/// Absolute allocation ceilings enforced while deserializing untrusted transport DTOs.
12/// Deployments normally configure lower runtime limits in `aequora-validator` and transports.
13pub mod wire_limits {
14    use aequora_types::OperationId;
15    use serde::{
16        Deserialize,
17        de::{self, Deserializer, SeqAccess, Visitor},
18    };
19    use smallvec::SmallVec;
20    use std::{fmt, marker::PhantomData};
21
22    /// Absolute operations accepted by the wire DTO decoder.
23    pub const OPERATIONS: usize = 4_096;
24    /// Absolute dependencies accepted on one wire operation.
25    pub const DEPENDENCIES: usize = 1_024;
26    /// Absolute partial-scope selectors accepted by the decoder.
27    pub const PARTITIONS: usize = 512;
28    /// Absolute feature capabilities accepted by the decoder.
29    pub const CAPABILITIES: usize = 64;
30    /// Absolute bytes accepted in any individual domain payload or partition value.
31    pub const PAYLOAD_BYTES: usize = 16 * 1_024 * 1_024;
32    /// Absolute entries accepted in an individual server result collection.
33    pub const RESULTS: usize = 8_192;
34    /// Absolute entities accepted in one decoded bootstrap page.
35    pub const SNAPSHOT_ENTITIES: usize = 8_192;
36
37    struct BoundedSequence<T, const MAX: usize>(PhantomData<T>);
38
39    impl<'de, T, const MAX: usize> Visitor<'de> for BoundedSequence<T, MAX>
40    where
41        T: Deserialize<'de>,
42    {
43        type Value = Vec<T>;
44
45        fn expecting(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
46            write!(formatter, "a sequence containing at most {MAX} elements")
47        }
48
49        fn visit_seq<A>(self, mut sequence: A) -> Result<Self::Value, A::Error>
50        where
51            A: SeqAccess<'de>,
52        {
53            if sequence.size_hint().is_some_and(|length| length > MAX) {
54                return Err(de::Error::invalid_length(MAX.saturating_add(1), &self));
55            }
56            let mut items = Vec::with_capacity(sequence.size_hint().unwrap_or(0).min(MAX));
57            while let Some(item) = sequence.next_element()? {
58                if items.len() == MAX {
59                    return Err(de::Error::invalid_length(MAX.saturating_add(1), &self));
60                }
61                items.push(item);
62            }
63            Ok(items)
64        }
65    }
66
67    fn bounded_vec<'de, D, T, const MAX: usize>(deserializer: D) -> Result<Vec<T>, D::Error>
68    where
69        D: Deserializer<'de>,
70        T: Deserialize<'de>,
71    {
72        deserializer.deserialize_seq(BoundedSequence::<T, MAX>(PhantomData))
73    }
74
75    pub(crate) fn operations<'de, D, T>(deserializer: D) -> Result<Vec<T>, D::Error>
76    where
77        D: Deserializer<'de>,
78        T: Deserialize<'de>,
79    {
80        bounded_vec::<D, T, OPERATIONS>(deserializer)
81    }
82
83    pub(crate) fn partitions<'de, D, T>(deserializer: D) -> Result<Vec<T>, D::Error>
84    where
85        D: Deserializer<'de>,
86        T: Deserialize<'de>,
87    {
88        bounded_vec::<D, T, PARTITIONS>(deserializer)
89    }
90
91    pub(crate) fn capabilities<'de, D, T>(deserializer: D) -> Result<Vec<T>, D::Error>
92    where
93        D: Deserializer<'de>,
94        T: Deserialize<'de>,
95    {
96        bounded_vec::<D, T, CAPABILITIES>(deserializer)
97    }
98
99    pub(crate) fn payload<'de, D>(deserializer: D) -> Result<Vec<u8>, D::Error>
100    where
101        D: Deserializer<'de>,
102    {
103        bounded_vec::<D, u8, PAYLOAD_BYTES>(deserializer)
104    }
105
106    pub(crate) fn results<'de, D, T>(deserializer: D) -> Result<Vec<T>, D::Error>
107    where
108        D: Deserializer<'de>,
109        T: Deserialize<'de>,
110    {
111        bounded_vec::<D, T, RESULTS>(deserializer)
112    }
113
114    pub(crate) fn snapshot_entities<'de, D, T>(deserializer: D) -> Result<Vec<T>, D::Error>
115    where
116        D: Deserializer<'de>,
117        T: Deserialize<'de>,
118    {
119        bounded_vec::<D, T, SNAPSHOT_ENTITIES>(deserializer)
120    }
121
122    struct BoundedDependencies;
123
124    impl<'de> Visitor<'de> for BoundedDependencies {
125        type Value = SmallVec<[OperationId; 4]>;
126
127        fn expecting(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
128            write!(
129                formatter,
130                "a dependency sequence containing at most {DEPENDENCIES} elements"
131            )
132        }
133
134        fn visit_seq<A>(self, mut sequence: A) -> Result<Self::Value, A::Error>
135        where
136            A: SeqAccess<'de>,
137        {
138            if sequence
139                .size_hint()
140                .is_some_and(|length| length > DEPENDENCIES)
141            {
142                return Err(de::Error::invalid_length(
143                    DEPENDENCIES.saturating_add(1),
144                    &self,
145                ));
146            }
147            let mut items = SmallVec::new();
148            while let Some(item) = sequence.next_element()? {
149                if items.len() == DEPENDENCIES {
150                    return Err(de::Error::invalid_length(
151                        DEPENDENCIES.saturating_add(1),
152                        &self,
153                    ));
154                }
155                items.push(item);
156            }
157            Ok(items)
158        }
159    }
160
161    pub(crate) fn dependencies<'de, D>(
162        deserializer: D,
163    ) -> Result<SmallVec<[OperationId; 4]>, D::Error>
164    where
165        D: Deserializer<'de>,
166    {
167        deserializer.deserialize_seq(BoundedDependencies)
168    }
169
170    #[cfg(test)]
171    mod tests {
172        use super::*;
173        use serde::{Deserialize, Serialize};
174
175        #[derive(Debug, Deserialize, Eq, PartialEq, Serialize)]
176        struct TinySequence(#[serde(deserialize_with = "tiny")] Vec<u8>);
177
178        fn tiny<'de, D>(deserializer: D) -> Result<Vec<u8>, D::Error>
179        where
180            D: Deserializer<'de>,
181        {
182            bounded_vec::<D, u8, 2>(deserializer)
183        }
184
185        #[test]
186        fn declared_collection_length_is_rejected_by_the_deserializer() {
187            let encoded = postcard::to_stdvec(&TinySequence(vec![1, 2, 3]))
188                .unwrap_or_else(|error| panic!("{error}"));
189            let decoded = postcard::from_bytes::<TinySequence>(&encoded);
190            assert!(decoded.is_err());
191        }
192    }
193}
194
195/// Numeric identifier registered by an application for an operation payload.
196#[derive(Clone, Copy, Debug, Deserialize, Eq, Hash, PartialEq, Serialize)]
197#[serde(transparent)]
198pub struct OperationKind(pub u16);
199
200/// Metadata that is useful to the application but independent of its payload.
201#[derive(Clone, Debug, Default, Deserialize, Eq, PartialEq, Serialize)]
202pub struct OperationMetadata {
203    /// Optional opaque trace identifier. It must not contain business payload data.
204    pub trace_id: Option<String>,
205    /// Operation IDs that must execute before this operation.
206    #[serde(deserialize_with = "wire_limits::dependencies")]
207    pub dependencies: SmallVec<[OperationId; 4]>,
208    /// Retry-stable root correlation and direct semantic cause.
209    #[serde(default = "legacy_lineage")]
210    pub lineage: LineageContext,
211}
212
213fn legacy_lineage() -> LineageContext {
214    LineageContext::legacy_missing()
215}
216
217/// A domain operation and the synchronization metadata needed to process it safely.
218#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
219pub struct OperationEnvelope {
220    /// Transport protocol spoken by the producer.
221    pub protocol_version: ProtocolVersion,
222    /// Permanent key used to make retries idempotent.
223    pub operation_id: OperationId,
224    /// Claimed tenant; the server must compare it with authenticated context.
225    pub tenant_id: TenantId,
226    /// Actor that originated the operation.
227    pub actor_id: ActorId,
228    /// Device that originated the operation.
229    pub device_id: DeviceId,
230    /// Aggregate root or entity targeted by the operation.
231    pub entity: EntityRef,
232    /// Authoritative version on which the local edit was based.
233    pub base_version: Option<EntityVersion>,
234    /// Causal timestamp metadata, never used as a cursor or entity version.
235    pub created_at: HybridTimestamp,
236    /// Application payload schema version.
237    pub schema_version: SchemaVersion,
238    /// Application-registered operation decoder/handler identifier.
239    pub operation_kind: OperationKind,
240    /// Postcard-encoded application command. Raw SQL is never valid here.
241    #[serde(deserialize_with = "wire_limits::payload")]
242    pub payload: Vec<u8>,
243    /// Dependencies and non-sensitive diagnostic metadata.
244    pub metadata: OperationMetadata,
245}
246
247/// Authenticated session metadata sent on every exchange.
248#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
249pub struct SessionMetadata {
250    /// Client session identity.
251    pub session_id: SessionId,
252    /// Client device identity.
253    pub device_id: DeviceId,
254    /// Authenticated actor identity as understood by the client.
255    pub actor_id: ActorId,
256    /// Tenant requested by the client.
257    pub tenant_id: TenantId,
258    /// Scope of the requested journal cursor.
259    pub scope_id: SyncScopeId,
260    /// Opaque application-defined filters that make up this partial synchronization scope.
261    #[serde(deserialize_with = "wire_limits::partitions")]
262    pub partitions: Vec<Partition>,
263}
264
265/// One opaque partial-synchronization partition selector.
266#[derive(Clone, Debug, Deserialize, Eq, Hash, PartialEq, Serialize)]
267pub struct Partition {
268    /// Compact application-defined partition kind.
269    pub kind: u16,
270    /// Opaque bounded value interpreted only by the application/server adapter.
271    #[serde(deserialize_with = "wire_limits::payload")]
272    pub value: Vec<u8>,
273}
274
275/// A feature that can be negotiated additively.
276#[derive(Clone, Copy, Debug, Deserialize, Eq, Hash, PartialEq, Serialize)]
277#[non_exhaustive]
278pub enum Capability {
279    /// Postcard payloads using protocol version one.
280    PostcardV1,
281    /// Zstandard compression is supported.
282    Zstd,
283    /// Snapshot bootstrap version one is supported.
284    SnapshotV1,
285    /// The peer understands tombstones.
286    Tombstones,
287    /// The transport can deliver multiple snapshot pages on one bounded stream.
288    StreamingSnapshots,
289    /// The transport can deliver payload-free journal-advance hints.
290    PushHints,
291    /// The peer supports the Aequora QUIC framing profile.
292    Quic,
293    /// The peer understands region-routing metadata.
294    MultiRegion,
295    /// The peer preserves correlation and direct-causation metadata version one.
296    LineageV1,
297    /// The peer supports canonical integrity generation one and bounded repair negotiation.
298    IntegrityV1,
299    /// The peer can negotiate versioned scope transitions outside the legacy v1 exchange body.
300    ScopeV1,
301    /// The peer can negotiate the additive transport-neutral live control protocol.
302    LiveV1,
303    /// The peer can verify signed snapshot manifests version one.
304    SignedSnapshotV1,
305    /// The peer can decrypt encrypted snapshot chunks version one.
306    EncryptedSnapshotV1,
307    /// The peer can produce or verify device operation signatures version one.
308    DeviceSignatureV1,
309    /// The peer binds every synchronization cursor to an authority ID and epoch.
310    AuthorityEpochV1,
311    /// The peer supplies coarse resource-aware transport limits without authorization meaning.
312    ResourceConstrainedV1,
313    /// The peer supports the explicit Part 21 client/server negotiation handshake.
314    CompatibilityNegotiationV1,
315}
316
317impl Capability {
318    /// Stable Part 21 registry ID. Values are append-only and never reused.
319    #[must_use]
320    pub const fn stable_id(self) -> u32 {
321        match self {
322            Self::PostcardV1 => 1,
323            Self::Zstd => 2,
324            Self::SnapshotV1 => 3,
325            Self::Tombstones => 4,
326            Self::StreamingSnapshots => 5,
327            Self::PushHints => 6,
328            Self::Quic => 7,
329            Self::MultiRegion => 8,
330            Self::LineageV1 => 9,
331            Self::IntegrityV1 => 10,
332            Self::ScopeV1 => 11,
333            Self::LiveV1 => 12,
334            Self::SignedSnapshotV1 => 13,
335            Self::EncryptedSnapshotV1 => 14,
336            Self::DeviceSignatureV1 => 15,
337            Self::AuthorityEpochV1 => 16,
338            Self::ResourceConstrainedV1 => 17,
339            Self::CompatibilityNegotiationV1 => 18,
340        }
341    }
342}
343
344/// Client-enforced response limits advertised to the server.
345#[derive(Clone, Copy, Debug, Deserialize, Eq, PartialEq, Serialize)]
346pub struct ClientLimits {
347    /// Maximum number of remote changes accepted in this response.
348    pub max_changes: u32,
349    /// Maximum uncompressed response size the client is prepared to accept.
350    pub max_response_bytes: u32,
351}
352
353impl Default for ClientLimits {
354    fn default() -> Self {
355        Self {
356            max_changes: 1_024,
357            max_response_bytes: 4 * 1_024 * 1_024,
358        }
359    }
360}
361
362/// One bidirectional push/pull synchronization request.
363#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
364pub struct SyncRequest {
365    /// Wire protocol version.
366    pub protocol: ProtocolVersion,
367    /// Unique request identity used only for correlation and diagnostics.
368    pub request_id: RequestId,
369    /// Client session and identity claims.
370    pub session: SessionMetadata,
371    /// Last authoritative sequence durably reconciled by the client.
372    pub cursor: Option<Cursor>,
373    /// Pending domain operations.
374    #[serde(deserialize_with = "wire_limits::operations")]
375    pub operations: Vec<OperationEnvelope>,
376    /// Limits the server must honor.
377    pub limits: ClientLimits,
378    /// Features supported by the client.
379    #[serde(deserialize_with = "wire_limits::capabilities")]
380    pub capabilities: Vec<Capability>,
381}
382
383/// Result retained by the server operation ledger and returned for retries.
384#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
385pub struct OperationAck {
386    /// Operation that produced this acknowledgement.
387    pub operation_id: OperationId,
388    /// Stable authoritative event returned for both first execution and duplicate replay.
389    pub event_id: EventId,
390    /// Original retry-stable operation lineage retained by the idempotency ledger.
391    pub lineage: LineageContext,
392    /// Resulting authoritative entity version.
393    pub entity_version: EntityVersion,
394    /// Journal sequence produced by the operation.
395    pub sequence: Sequence,
396    /// True when this response was replayed from the idempotency ledger.
397    pub duplicate: bool,
398}
399
400/// Stable machine-readable rejection categories.
401#[derive(Clone, Copy, Debug, Deserialize, Eq, PartialEq, Serialize)]
402#[non_exhaustive]
403pub enum RejectionCode {
404    /// Request identity did not match the authenticated context.
405    IdentityMismatch,
406    /// The actor is not allowed to perform the operation.
407    Unauthorized,
408    /// Wire shape or bounded-field validation failed.
409    InvalidOperation,
410    /// The application rejected the operation's business semantics.
411    BusinessRule,
412    /// The operation depends on an unavailable or rejected operation.
413    Dependency,
414    /// Operation schema is outside the application's compatibility window.
415    SchemaIncompatible,
416}
417
418/// A permanent operation rejection that should not be retried unchanged.
419#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
420pub struct OperationRejection {
421    /// Rejected operation.
422    pub operation_id: OperationId,
423    /// Machine-readable category.
424    pub code: RejectionCode,
425    /// Bounded, non-sensitive explanation suitable for a conflict inbox.
426    pub message: String,
427}
428
429/// Conflict behavior selected by application policy.
430#[derive(Clone, Copy, Debug, Deserialize, Eq, PartialEq, Serialize)]
431#[non_exhaustive]
432pub enum ConflictPolicy {
433    /// Reject stale writes.
434    Reject,
435    /// Keep authoritative state.
436    ServerWins,
437    /// Application explicitly permits client replacement.
438    ClientWins,
439    /// Application-specific merger is required.
440    CustomMerge,
441    /// A human must resolve the conflict.
442    ManualResolution,
443    /// Merge independently timestamped application fields.
444    FieldMerge,
445    /// Apply an application-defined commutative mutation.
446    CommutativeOperation,
447    /// Merge an application-defined convergent replicated data type.
448    Crdt,
449    /// Deterministically keep the newest application-timestamped whole value.
450    LastWriterWins,
451}
452
453/// A stale-base conflict surfaced to the client.
454#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
455pub struct Conflict {
456    /// Operation that encountered the conflict.
457    pub operation_id: OperationId,
458    /// Entity whose version diverged.
459    pub entity: EntityRef,
460    /// Version from which the client edited.
461    pub client_base: Option<EntityVersion>,
462    /// Current authoritative version, if the entity exists.
463    pub server_version: Option<EntityVersion>,
464    /// Policy applied by the server.
465    pub policy: ConflictPolicy,
466    /// Non-sensitive application explanation.
467    pub message: String,
468}
469
470/// Kind of authoritative state transition.
471#[derive(Clone, Copy, Debug, Deserialize, Eq, PartialEq, Serialize)]
472pub enum ChangeKind {
473    /// Entity was created or replaced with active state.
474    Upsert,
475    /// Entity was deleted but remains represented for synchronization.
476    Tombstone,
477}
478
479/// An authoritative journal entry pulled by a client.
480#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
481pub struct RemoteChange {
482    /// Tenant owning the change.
483    pub tenant_id: TenantId,
484    /// Scope in which `sequence` is monotonic.
485    pub scope_id: SyncScopeId,
486    /// Authoritative journal position.
487    pub sequence: Sequence,
488    /// Operation that produced the change.
489    pub operation_id: OperationId,
490    /// Stable identity of this authoritative event, independent from its journal sequence.
491    pub event_id: EventId,
492    /// Root correlation and direct cause retained across adapters and consumers.
493    pub lineage: LineageContext,
494    /// Changed entity.
495    pub entity: EntityRef,
496    /// Resulting entity version.
497    pub version: EntityVersion,
498    /// Upsert or tombstone.
499    pub change_kind: ChangeKind,
500    /// Application-defined authoritative snapshot/event payload.
501    #[serde(deserialize_with = "wire_limits::payload")]
502    pub payload: Vec<u8>,
503    /// Authoritative event timestamp.
504    pub timestamp: HybridTimestamp,
505}
506
507/// Reason incremental synchronization must restart from a consistent snapshot.
508#[derive(Clone, Copy, Debug, Deserialize, Eq, PartialEq, Serialize)]
509#[non_exhaustive]
510pub enum ResyncReason {
511    /// The client's cursor predates retained journal history.
512    CursorExpired,
513    /// The requested partial synchronization scope changed incompatibly.
514    ScopeChanged,
515    /// Domain schema cannot be migrated incrementally.
516    SchemaIncompatible,
517    /// The device exceeded the deployment's inactivity window.
518    DeviceInactive,
519    /// Local or authoritative consistency checks detected corruption.
520    CorruptionDetected,
521}
522
523/// Typed server instruction accompanying every synchronization response.
524#[derive(Clone, Copy, Debug, Default, Deserialize, Eq, PartialEq, Serialize)]
525pub enum SyncDirective {
526    /// Reconcile this normal incremental response.
527    #[default]
528    Continue,
529    /// This client protocol falls outside the server compatibility window.
530    UpgradeRequired {
531        /// Oldest protocol accepted by the server.
532        minimum: ProtocolVersion,
533        /// Current protocol emitted by the server.
534        current: ProtocolVersion,
535    },
536    /// Discard incremental progress only through the normal atomic bootstrap flow.
537    ResyncRequired {
538        /// Stable reason suitable for application policy and diagnostics.
539        reason: ResyncReason,
540    },
541    /// The authority timeline advanced and incremental replay must freeze before rebootstrap.
542    AuthorityChanged {
543        /// Logical authority that owns the replacement timeline.
544        authority_id: AuthorityId,
545        /// Epoch supplied by the rejected client cursor.
546        previous_epoch: AuthorityEpoch,
547        /// Current epoch that must be bootstrapped.
548        current_epoch: AuthorityEpoch,
549    },
550}
551
552/// One response containing push results and incremental pull changes.
553#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
554pub struct SyncResponse {
555    /// Server protocol version.
556    pub protocol: ProtocolVersion,
557    /// Compatibility or recovery instruction evaluated before reconciliation.
558    pub directive: SyncDirective,
559    /// Accepted operations, including deterministic duplicate replies.
560    #[serde(deserialize_with = "wire_limits::results")]
561    pub acknowledged: Vec<OperationAck>,
562    /// Permanently rejected operations.
563    #[serde(deserialize_with = "wire_limits::results")]
564    pub rejected: Vec<OperationRejection>,
565    /// Operations requiring conflict handling.
566    #[serde(deserialize_with = "wire_limits::results")]
567    pub conflicts: Vec<Conflict>,
568    /// Authoritative journal page after the client's cursor.
569    #[serde(deserialize_with = "wire_limits::results")]
570    pub changes: Vec<RemoteChange>,
571    /// Cursor through which returned changes are complete.
572    pub next_cursor: Cursor,
573    /// True when another exchange is needed to finish the current pull.
574    pub has_more: bool,
575    /// Hybrid timestamp emitted by the server.
576    pub server_time: HybridTimestamp,
577}
578
579/// Client limits for one bootstrap snapshot page.
580#[derive(Clone, Copy, Debug, Deserialize, Eq, PartialEq, Serialize)]
581pub struct SnapshotLimits {
582    /// Maximum entities accepted in one page.
583    pub max_entities: u32,
584    /// Maximum aggregate application payload bytes accepted in one page.
585    pub max_payload_bytes: u32,
586}
587
588impl Default for SnapshotLimits {
589    fn default() -> Self {
590        Self {
591            max_entities: 512,
592            max_payload_bytes: 4 * 1_024 * 1_024,
593        }
594    }
595}
596
597/// Request to begin or resume a consistent snapshot bootstrap.
598#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
599pub struct BootstrapRequest {
600    /// Wire protocol version.
601    pub protocol: ProtocolVersion,
602    /// Unique request identity used only for correlation and diagnostics.
603    pub request_id: RequestId,
604    /// Authenticated session and partial scope.
605    pub session: SessionMetadata,
606    /// Existing consistent snapshot when resuming, or `None` to begin.
607    pub snapshot_id: Option<SnapshotId>,
608    /// Zero-based entity offset requested from the snapshot.
609    pub offset: u64,
610    /// Page bounds the server must honor.
611    pub limits: SnapshotLimits,
612    /// Client features used for negotiation.
613    #[serde(deserialize_with = "wire_limits::capabilities")]
614    pub capabilities: Vec<Capability>,
615}
616
617/// One authoritative entity in a consistent bootstrap snapshot.
618#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
619pub struct SnapshotEntity {
620    /// Stable entity identity.
621    pub entity: EntityRef,
622    /// Authoritative version at the snapshot boundary.
623    pub version: EntityVersion,
624    /// Application-owned authoritative state.
625    #[serde(deserialize_with = "wire_limits::payload")]
626    pub payload: Vec<u8>,
627    /// True when the snapshot preserves a deletion tombstone.
628    pub tombstone: bool,
629}
630
631/// One resumable page from a consistent bootstrap snapshot.
632#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
633pub struct BootstrapResponse {
634    /// Server wire protocol version.
635    pub protocol: ProtocolVersion,
636    /// Stable snapshot identity shared by every page.
637    pub snapshot_id: SnapshotId,
638    /// Cursor at the logical snapshot boundary.
639    pub cursor: Cursor,
640    /// Offset represented by the first entity in this page.
641    pub offset: u64,
642    /// Bounded page of authoritative entity state.
643    #[serde(deserialize_with = "wire_limits::snapshot_entities")]
644    pub entities: Vec<SnapshotEntity>,
645    /// Offset to request next.
646    pub next_offset: u64,
647    /// True when another page remains.
648    pub has_more: bool,
649    /// Server timestamp for causal observation.
650    pub server_time: HybridTimestamp,
651}
652
653/// Reason a server suggests that the client perform a normal synchronization exchange.
654#[derive(Clone, Copy, Debug, Deserialize, Eq, PartialEq, Serialize)]
655#[non_exhaustive]
656pub enum PushHintReason {
657    /// The authoritative journal advanced beyond the hinted cursor.
658    JournalAdvanced,
659    /// A previously captured bootstrap snapshot should no longer be resumed.
660    SnapshotInvalidated,
661    /// The preferred serving region changed.
662    RegionChanged,
663}
664
665/// Payload-free notification that prompts a client to use the normal pull protocol.
666/// Hints are advisory and never carry authoritative business state.
667#[derive(Clone, Copy, Debug, Deserialize, Eq, PartialEq, Serialize)]
668pub struct PushHint {
669    /// Wire protocol spoken by the sender.
670    pub protocol: ProtocolVersion,
671    /// Tenant boundary to validate before acting on the hint.
672    pub tenant_id: TenantId,
673    /// Scope whose journal may have advanced.
674    pub scope_id: SyncScopeId,
675    /// Greatest journal position known when the hint was emitted.
676    pub sequence: Sequence,
677    /// Why the hint was emitted.
678    pub reason: PushHintReason,
679    /// Serving region that emitted the hint, when region routing is enabled.
680    pub region_id: Option<RegionId>,
681}
682
683#[cfg(test)]
684mod compatibility_tests {
685    use super::*;
686
687    #[test]
688    fn conflict_policy_wire_discriminants_remain_append_only() {
689        let policies = [
690            ConflictPolicy::Reject,
691            ConflictPolicy::ServerWins,
692            ConflictPolicy::ClientWins,
693            ConflictPolicy::CustomMerge,
694            ConflictPolicy::ManualResolution,
695            ConflictPolicy::FieldMerge,
696            ConflictPolicy::CommutativeOperation,
697            ConflictPolicy::Crdt,
698            ConflictPolicy::LastWriterWins,
699        ];
700        for (discriminant, policy) in policies.into_iter().enumerate() {
701            assert_eq!(
702                postcard::to_stdvec(&policy).unwrap_or_else(|error| panic!("{error}")),
703                vec![u8::try_from(discriminant).unwrap_or(u8::MAX)]
704            );
705        }
706    }
707
708    #[test]
709    fn capability_wire_discriminants_and_registry_ids_remain_append_only() {
710        let capabilities = [
711            Capability::PostcardV1,
712            Capability::Zstd,
713            Capability::SnapshotV1,
714            Capability::Tombstones,
715            Capability::StreamingSnapshots,
716            Capability::PushHints,
717            Capability::Quic,
718            Capability::MultiRegion,
719            Capability::LineageV1,
720            Capability::IntegrityV1,
721            Capability::ScopeV1,
722            Capability::LiveV1,
723            Capability::SignedSnapshotV1,
724            Capability::EncryptedSnapshotV1,
725            Capability::DeviceSignatureV1,
726            Capability::AuthorityEpochV1,
727            Capability::ResourceConstrainedV1,
728            Capability::CompatibilityNegotiationV1,
729        ];
730        for (discriminant, capability) in capabilities.into_iter().enumerate() {
731            assert_eq!(
732                postcard::to_stdvec(&capability).unwrap_or_else(|error| panic!("{error}")),
733                vec![u8::try_from(discriminant).unwrap_or(u8::MAX)]
734            );
735            assert_eq!(
736                capability.stable_id(),
737                u32::try_from(discriminant).unwrap_or(u32::MAX) + 1
738            );
739        }
740    }
741}
aequora_protocol/lib.rs

aequora_protocol/
lib.rs
