use aequora_types::{
ActorId, AuthorityEpoch, AuthorityId, Cursor, DeviceId, EntityRef, EntityVersion, EventId,
HybridTimestamp, LineageContext, OperationId, ProtocolVersion, RegionId, RequestId,
SchemaVersion, Sequence, SessionId, SnapshotId, SyncScopeId, TenantId,
};
use serde::{Deserialize, Serialize};
use smallvec::SmallVec;
pub mod wire_limits {
use aequora_types::OperationId;
use serde::{
Deserialize,
de::{self, Deserializer, SeqAccess, Visitor},
};
use smallvec::SmallVec;
use std::{fmt, marker::PhantomData};
pub const OPERATIONS: usize = 4_096;
pub const DEPENDENCIES: usize = 1_024;
pub const PARTITIONS: usize = 512;
pub const CAPABILITIES: usize = 64;
pub const PAYLOAD_BYTES: usize = 16 * 1_024 * 1_024;
pub const RESULTS: usize = 8_192;
pub const SNAPSHOT_ENTITIES: usize = 8_192;
struct BoundedSequence<T, const MAX: usize>(PhantomData<T>);
impl<'de, T, const MAX: usize> Visitor<'de> for BoundedSequence<T, MAX>
where
T: Deserialize<'de>,
{
type Value = Vec<T>;
fn expecting(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
write!(formatter, "a sequence containing at most {MAX} elements")
}
fn visit_seq<A>(self, mut sequence: A) -> Result<Self::Value, A::Error>
where
A: SeqAccess<'de>,
{
if sequence.size_hint().is_some_and(|length| length > MAX) {
return Err(de::Error::invalid_length(MAX.saturating_add(1), &self));
}
let mut items = Vec::with_capacity(sequence.size_hint().unwrap_or(0).min(MAX));
while let Some(item) = sequence.next_element()? {
if items.len() == MAX {
return Err(de::Error::invalid_length(MAX.saturating_add(1), &self));
}
items.push(item);
}
Ok(items)
}
}
fn bounded_vec<'de, D, T, const MAX: usize>(deserializer: D) -> Result<Vec<T>, D::Error>
where
D: Deserializer<'de>,
T: Deserialize<'de>,
{
deserializer.deserialize_seq(BoundedSequence::<T, MAX>(PhantomData))
}
pub(crate) fn operations<'de, D, T>(deserializer: D) -> Result<Vec<T>, D::Error>
where
D: Deserializer<'de>,
T: Deserialize<'de>,
{
bounded_vec::<D, T, OPERATIONS>(deserializer)
}
pub(crate) fn partitions<'de, D, T>(deserializer: D) -> Result<Vec<T>, D::Error>
where
D: Deserializer<'de>,
T: Deserialize<'de>,
{
bounded_vec::<D, T, PARTITIONS>(deserializer)
}
pub(crate) fn capabilities<'de, D, T>(deserializer: D) -> Result<Vec<T>, D::Error>
where
D: Deserializer<'de>,
T: Deserialize<'de>,
{
bounded_vec::<D, T, CAPABILITIES>(deserializer)
}
pub(crate) fn payload<'de, D>(deserializer: D) -> Result<Vec<u8>, D::Error>
where
D: Deserializer<'de>,
{
bounded_vec::<D, u8, PAYLOAD_BYTES>(deserializer)
}
pub(crate) fn results<'de, D, T>(deserializer: D) -> Result<Vec<T>, D::Error>
where
D: Deserializer<'de>,
T: Deserialize<'de>,
{
bounded_vec::<D, T, RESULTS>(deserializer)
}
pub(crate) fn snapshot_entities<'de, D, T>(deserializer: D) -> Result<Vec<T>, D::Error>
where
D: Deserializer<'de>,
T: Deserialize<'de>,
{
bounded_vec::<D, T, SNAPSHOT_ENTITIES>(deserializer)
}
struct BoundedDependencies;
impl<'de> Visitor<'de> for BoundedDependencies {
type Value = SmallVec<[OperationId; 4]>;
fn expecting(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
write!(
formatter,
"a dependency sequence containing at most {DEPENDENCIES} elements"
)
}
fn visit_seq<A>(self, mut sequence: A) -> Result<Self::Value, A::Error>
where
A: SeqAccess<'de>,
{
if sequence
.size_hint()
.is_some_and(|length| length > DEPENDENCIES)
{
return Err(de::Error::invalid_length(
DEPENDENCIES.saturating_add(1),
&self,
));
}
let mut items = SmallVec::new();
while let Some(item) = sequence.next_element()? {
if items.len() == DEPENDENCIES {
return Err(de::Error::invalid_length(
DEPENDENCIES.saturating_add(1),
&self,
));
}
items.push(item);
}
Ok(items)
}
}
pub(crate) fn dependencies<'de, D>(
deserializer: D,
) -> Result<SmallVec<[OperationId; 4]>, D::Error>
where
D: Deserializer<'de>,
{
deserializer.deserialize_seq(BoundedDependencies)
}
#[cfg(test)]
mod tests {
use super::*;
use serde::{Deserialize, Serialize};
#[derive(Debug, Deserialize, Eq, PartialEq, Serialize)]
struct TinySequence(#[serde(deserialize_with = "tiny")] Vec<u8>);
fn tiny<'de, D>(deserializer: D) -> Result<Vec<u8>, D::Error>
where
D: Deserializer<'de>,
{
bounded_vec::<D, u8, 2>(deserializer)
}
#[test]
fn declared_collection_length_is_rejected_by_the_deserializer() {
let encoded = postcard::to_stdvec(&TinySequence(vec![1, 2, 3]))
.unwrap_or_else(|error| panic!("{error}"));
let decoded = postcard::from_bytes::<TinySequence>(&encoded);
assert!(decoded.is_err());
}
}
}
#[derive(Clone, Copy, Debug, Deserialize, Eq, Hash, PartialEq, Serialize)]
#[serde(transparent)]
pub struct OperationKind(pub u16);
#[derive(Clone, Debug, Default, Deserialize, Eq, PartialEq, Serialize)]
pub struct OperationMetadata {
pub trace_id: Option<String>,
#[serde(deserialize_with = "wire_limits::dependencies")]
pub dependencies: SmallVec<[OperationId; 4]>,
#[serde(default = "legacy_lineage")]
pub lineage: LineageContext,
}
fn legacy_lineage() -> LineageContext {
LineageContext::legacy_missing()
}
#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
pub struct OperationEnvelope {
pub protocol_version: ProtocolVersion,
pub operation_id: OperationId,
pub tenant_id: TenantId,
pub actor_id: ActorId,
pub device_id: DeviceId,
pub entity: EntityRef,
pub base_version: Option<EntityVersion>,
pub created_at: HybridTimestamp,
pub schema_version: SchemaVersion,
pub operation_kind: OperationKind,
#[serde(deserialize_with = "wire_limits::payload")]
pub payload: Vec<u8>,
pub metadata: OperationMetadata,
}
#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
pub struct SessionMetadata {
pub session_id: SessionId,
pub device_id: DeviceId,
pub actor_id: ActorId,
pub tenant_id: TenantId,
pub scope_id: SyncScopeId,
#[serde(deserialize_with = "wire_limits::partitions")]
pub partitions: Vec<Partition>,
}
#[derive(Clone, Debug, Deserialize, Eq, Hash, PartialEq, Serialize)]
pub struct Partition {
pub kind: u16,
#[serde(deserialize_with = "wire_limits::payload")]
pub value: Vec<u8>,
}
#[derive(Clone, Copy, Debug, Deserialize, Eq, Hash, PartialEq, Serialize)]
#[non_exhaustive]
pub enum Capability {
PostcardV1,
Zstd,
SnapshotV1,
Tombstones,
StreamingSnapshots,
PushHints,
Quic,
MultiRegion,
LineageV1,
IntegrityV1,
ScopeV1,
LiveV1,
SignedSnapshotV1,
EncryptedSnapshotV1,
DeviceSignatureV1,
AuthorityEpochV1,
ResourceConstrainedV1,
CompatibilityNegotiationV1,
}
impl Capability {
#[must_use]
pub const fn stable_id(self) -> u32 {
match self {
Self::PostcardV1 => 1,
Self::Zstd => 2,
Self::SnapshotV1 => 3,
Self::Tombstones => 4,
Self::StreamingSnapshots => 5,
Self::PushHints => 6,
Self::Quic => 7,
Self::MultiRegion => 8,
Self::LineageV1 => 9,
Self::IntegrityV1 => 10,
Self::ScopeV1 => 11,
Self::LiveV1 => 12,
Self::SignedSnapshotV1 => 13,
Self::EncryptedSnapshotV1 => 14,
Self::DeviceSignatureV1 => 15,
Self::AuthorityEpochV1 => 16,
Self::ResourceConstrainedV1 => 17,
Self::CompatibilityNegotiationV1 => 18,
}
}
}
#[derive(Clone, Copy, Debug, Deserialize, Eq, PartialEq, Serialize)]
pub struct ClientLimits {
pub max_changes: u32,
pub max_response_bytes: u32,
}
impl Default for ClientLimits {
fn default() -> Self {
Self {
max_changes: 1_024,
max_response_bytes: 4 * 1_024 * 1_024,
}
}
}
#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
pub struct SyncRequest {
pub protocol: ProtocolVersion,
pub request_id: RequestId,
pub session: SessionMetadata,
pub cursor: Option<Cursor>,
#[serde(deserialize_with = "wire_limits::operations")]
pub operations: Vec<OperationEnvelope>,
pub limits: ClientLimits,
#[serde(deserialize_with = "wire_limits::capabilities")]
pub capabilities: Vec<Capability>,
}
#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
pub struct OperationAck {
pub operation_id: OperationId,
pub event_id: EventId,
pub lineage: LineageContext,
pub entity_version: EntityVersion,
pub sequence: Sequence,
pub duplicate: bool,
}
#[derive(Clone, Copy, Debug, Deserialize, Eq, PartialEq, Serialize)]
#[non_exhaustive]
pub enum RejectionCode {
IdentityMismatch,
Unauthorized,
InvalidOperation,
BusinessRule,
Dependency,
SchemaIncompatible,
}
#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
pub struct OperationRejection {
pub operation_id: OperationId,
pub code: RejectionCode,
pub message: String,
}
#[derive(Clone, Copy, Debug, Deserialize, Eq, PartialEq, Serialize)]
#[non_exhaustive]
pub enum ConflictPolicy {
Reject,
ServerWins,
ClientWins,
CustomMerge,
ManualResolution,
FieldMerge,
CommutativeOperation,
Crdt,
LastWriterWins,
}
#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
pub struct Conflict {
pub operation_id: OperationId,
pub entity: EntityRef,
pub client_base: Option<EntityVersion>,
pub server_version: Option<EntityVersion>,
pub policy: ConflictPolicy,
pub message: String,
}
#[derive(Clone, Copy, Debug, Deserialize, Eq, PartialEq, Serialize)]
pub enum ChangeKind {
Upsert,
Tombstone,
}
#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
pub struct RemoteChange {
pub tenant_id: TenantId,
pub scope_id: SyncScopeId,
pub sequence: Sequence,
pub operation_id: OperationId,
pub event_id: EventId,
pub lineage: LineageContext,
pub entity: EntityRef,
pub version: EntityVersion,
pub change_kind: ChangeKind,
#[serde(deserialize_with = "wire_limits::payload")]
pub payload: Vec<u8>,
pub timestamp: HybridTimestamp,
}
#[derive(Clone, Copy, Debug, Deserialize, Eq, PartialEq, Serialize)]
#[non_exhaustive]
pub enum ResyncReason {
CursorExpired,
ScopeChanged,
SchemaIncompatible,
DeviceInactive,
CorruptionDetected,
}
#[derive(Clone, Copy, Debug, Default, Deserialize, Eq, PartialEq, Serialize)]
pub enum SyncDirective {
#[default]
Continue,
UpgradeRequired {
minimum: ProtocolVersion,
current: ProtocolVersion,
},
ResyncRequired {
reason: ResyncReason,
},
AuthorityChanged {
authority_id: AuthorityId,
previous_epoch: AuthorityEpoch,
current_epoch: AuthorityEpoch,
},
}
#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
pub struct SyncResponse {
pub protocol: ProtocolVersion,
pub directive: SyncDirective,
#[serde(deserialize_with = "wire_limits::results")]
pub acknowledged: Vec<OperationAck>,
#[serde(deserialize_with = "wire_limits::results")]
pub rejected: Vec<OperationRejection>,
#[serde(deserialize_with = "wire_limits::results")]
pub conflicts: Vec<Conflict>,
#[serde(deserialize_with = "wire_limits::results")]
pub changes: Vec<RemoteChange>,
pub next_cursor: Cursor,
pub has_more: bool,
pub server_time: HybridTimestamp,
}
#[derive(Clone, Copy, Debug, Deserialize, Eq, PartialEq, Serialize)]
pub struct SnapshotLimits {
pub max_entities: u32,
pub max_payload_bytes: u32,
}
impl Default for SnapshotLimits {
fn default() -> Self {
Self {
max_entities: 512,
max_payload_bytes: 4 * 1_024 * 1_024,
}
}
}
#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
pub struct BootstrapRequest {
pub protocol: ProtocolVersion,
pub request_id: RequestId,
pub session: SessionMetadata,
pub snapshot_id: Option<SnapshotId>,
pub offset: u64,
pub limits: SnapshotLimits,
#[serde(deserialize_with = "wire_limits::capabilities")]
pub capabilities: Vec<Capability>,
}
#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
pub struct SnapshotEntity {
pub entity: EntityRef,
pub version: EntityVersion,
#[serde(deserialize_with = "wire_limits::payload")]
pub payload: Vec<u8>,
pub tombstone: bool,
}
#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
pub struct BootstrapResponse {
pub protocol: ProtocolVersion,
pub snapshot_id: SnapshotId,
pub cursor: Cursor,
pub offset: u64,
#[serde(deserialize_with = "wire_limits::snapshot_entities")]
pub entities: Vec<SnapshotEntity>,
pub next_offset: u64,
pub has_more: bool,
pub server_time: HybridTimestamp,
}
#[derive(Clone, Copy, Debug, Deserialize, Eq, PartialEq, Serialize)]
#[non_exhaustive]
pub enum PushHintReason {
JournalAdvanced,
SnapshotInvalidated,
RegionChanged,
}
#[derive(Clone, Copy, Debug, Deserialize, Eq, PartialEq, Serialize)]
pub struct PushHint {
pub protocol: ProtocolVersion,
pub tenant_id: TenantId,
pub scope_id: SyncScopeId,
pub sequence: Sequence,
pub reason: PushHintReason,
pub region_id: Option<RegionId>,
}
#[cfg(test)]
mod compatibility_tests {
use super::*;
#[test]
fn conflict_policy_wire_discriminants_remain_append_only() {
let policies = [
ConflictPolicy::Reject,
ConflictPolicy::ServerWins,
ConflictPolicy::ClientWins,
ConflictPolicy::CustomMerge,
ConflictPolicy::ManualResolution,
ConflictPolicy::FieldMerge,
ConflictPolicy::CommutativeOperation,
ConflictPolicy::Crdt,
ConflictPolicy::LastWriterWins,
];
for (discriminant, policy) in policies.into_iter().enumerate() {
assert_eq!(
postcard::to_stdvec(&policy).unwrap_or_else(|error| panic!("{error}")),
vec![u8::try_from(discriminant).unwrap_or(u8::MAX)]
);
}
}
#[test]
fn capability_wire_discriminants_and_registry_ids_remain_append_only() {
let capabilities = [
Capability::PostcardV1,
Capability::Zstd,
Capability::SnapshotV1,
Capability::Tombstones,
Capability::StreamingSnapshots,
Capability::PushHints,
Capability::Quic,
Capability::MultiRegion,
Capability::LineageV1,
Capability::IntegrityV1,
Capability::ScopeV1,
Capability::LiveV1,
Capability::SignedSnapshotV1,
Capability::EncryptedSnapshotV1,
Capability::DeviceSignatureV1,
Capability::AuthorityEpochV1,
Capability::ResourceConstrainedV1,
Capability::CompatibilityNegotiationV1,
];
for (discriminant, capability) in capabilities.into_iter().enumerate() {
assert_eq!(
postcard::to_stdvec(&capability).unwrap_or_else(|error| panic!("{error}")),
vec![u8::try_from(discriminant).unwrap_or(u8::MAX)]
);
assert_eq!(
capability.stable_id(),
u32::try_from(discriminant).unwrap_or(u32::MAX) + 1
);
}
}
}