use std::error::Error;
use std::fmt::{self, Display, Formatter};
use lodestone_types::{CollectionIndexRef, DocumentId, FenceToken, RecoveryEra, RoutingRevision, SeqNo, ShardId};
use serde::{Deserialize, Serialize};
use serde_json::Value;
use serde_json::value::RawValue;
use crate::RequestId;
pub const SUBSCRIPTION_FORMAT_VERSION: u16 = 1;
pub const MAX_SUBSCRIPTION_INDEX_PATTERNS: usize = 128;
pub const MAX_SUBSCRIPTION_PATTERN_BYTES: usize = 1_024;
pub const MAX_SUBSCRIPTION_FILTER_NODES: usize = 64;
pub const MAX_SUBSCRIPTION_FILTER_TERMS: usize = 1_024;
pub const MAX_SUBSCRIPTION_FILTER_BYTES: usize = 64 * 1_024;
pub const MAX_SUBSCRIPTION_SOURCE_FIELDS: usize = 128;
pub const MAX_SUBSCRIPTION_FIELD_BYTES: usize = 1_024;
pub const MAX_SUBSCRIPTION_ROUTE_SCOPES: usize = 128;
pub const MAX_SUBSCRIPTION_ROUTE_SHARDS: usize = 4_096;
pub const MAX_SUBSCRIPTION_REQUEST_BYTES: usize = 2 * 1024 * 1024;
pub const MAX_SUBSCRIPTION_FRAME_BYTES: usize = 2 * 1024 * 1024;
#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct FilterSubscriptionV1 {
pub format_version: u16,
pub request_id: RequestId,
pub index_patterns: Vec<String>,
#[serde(default)]
pub filter: Option<SubscriptionFilterV1>,
#[serde(default)]
pub include_deletes: bool,
#[serde(default)]
pub source_projection: Vec<String>,
pub reconnect_attempt: u16,
pub start: SubscriptionStartV1,
}
impl FilterSubscriptionV1 {
pub fn validate(&self) -> Result<(), SubscriptionWireError> {
if self.format_version != SUBSCRIPTION_FORMAT_VERSION {
return Err(SubscriptionWireError::UnsupportedVersion(self.format_version));
}
if self.index_patterns.is_empty() || self.index_patterns.len() > MAX_SUBSCRIPTION_INDEX_PATTERNS {
return Err(SubscriptionWireError::IndexPatternCount(self.index_patterns.len()));
}
for pattern in &self.index_patterns {
validate_nonempty_bounded(pattern, MAX_SUBSCRIPTION_PATTERN_BYTES, "index pattern")?;
if pattern.bytes().any(|byte| byte.is_ascii_control()) {
return Err(SubscriptionWireError::InvalidPattern);
}
}
if self.source_projection.len() > MAX_SUBSCRIPTION_SOURCE_FIELDS {
return Err(SubscriptionWireError::SourceProjectionCount(
self.source_projection.len(),
));
}
for field in &self.source_projection {
validate_field_path(field)?;
}
if let Some(filter) = &self.filter {
let mut budget = FilterBudget::default();
filter.validate(&mut budget)?;
let encoded = serde_json::to_vec(filter).map_err(|error| SubscriptionWireError::Json(error.to_string()))?;
if encoded.len() > MAX_SUBSCRIPTION_FILTER_BYTES {
return Err(SubscriptionWireError::FilterBytes(encoded.len()));
}
if self.include_deletes && filter.references_source() {
return Err(SubscriptionWireError::FilteredDeletesRequirePriorImage);
}
}
self.start.validate()
}
}
pub fn parse_filter_subscription(bytes: &[u8]) -> Result<FilterSubscriptionV1, SubscriptionWireError> {
if bytes.len() > MAX_SUBSCRIPTION_REQUEST_BYTES {
return Err(SubscriptionWireError::RequestBytes(bytes.len()));
}
let request: FilterSubscriptionV1 =
serde_json::from_slice(bytes).map_err(|error| SubscriptionWireError::Json(error.to_string()))?;
request.validate()?;
Ok(request)
}
#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
#[serde(tag = "kind", rename_all = "snake_case", deny_unknown_fields)]
pub enum SubscriptionStartV1 {
FromNow,
Resume {
routes: Vec<SubscriptionRouteIdentityV1>,
cut: Vec<SubscriptionShardCursorV1>,
},
}
impl SubscriptionStartV1 {
fn validate(&self) -> Result<(), SubscriptionWireError> {
let Self::Resume { routes, cut } = self else {
return Ok(());
};
if routes.is_empty() || routes.len() > MAX_SUBSCRIPTION_ROUTE_SCOPES {
return Err(SubscriptionWireError::RouteCount(routes.len()));
}
if cut.is_empty() || cut.len() > MAX_SUBSCRIPTION_ROUTE_SHARDS {
return Err(SubscriptionWireError::CursorCount(cut.len()));
}
let mut prior_scope = None;
let mut expected_cursors = 0usize;
for route in routes {
if prior_scope.is_some_and(|prior| prior >= route.collection_index_ref) {
return Err(SubscriptionWireError::NonCanonicalRouteVector);
}
prior_scope = Some(route.collection_index_ref);
route.validate()?;
expected_cursors = expected_cursors
.checked_add(route.shard_ids.len())
.ok_or(SubscriptionWireError::CursorCount(usize::MAX))?;
}
if expected_cursors != cut.len() {
return Err(SubscriptionWireError::IncompleteCursorVector);
}
let mut prior = None;
for cursor in cut {
let key = (cursor.collection_index_ref, cursor.shard_id);
if prior.is_some_and(|candidate| candidate >= key) {
return Err(SubscriptionWireError::NonCanonicalCursorVector);
}
prior = Some(key);
let matching_route = routes
.binary_search_by_key(&cursor.collection_index_ref, |route| route.collection_index_ref)
.ok()
.and_then(|index| routes.get(index));
if matching_route.is_none_or(|route| route.shard_ids.binary_search(&cursor.shard_id).is_err()) {
return Err(SubscriptionWireError::IncompleteCursorVector);
}
}
Ok(())
}
}
#[derive(Clone, Debug, Eq, PartialEq, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct SubscriptionRouteIdentityV1 {
pub collection_index_ref: CollectionIndexRef,
pub routing_revision: RoutingRevision,
pub shard_ids: Vec<ShardId>,
}
impl SubscriptionRouteIdentityV1 {
fn validate(&self) -> Result<(), SubscriptionWireError> {
if self.shard_ids.is_empty() || self.shard_ids.len() > MAX_SUBSCRIPTION_ROUTE_SHARDS {
return Err(SubscriptionWireError::RouteShardCount(self.shard_ids.len()));
}
if self.shard_ids.windows(2).any(|pair| pair[0] >= pair[1]) {
return Err(SubscriptionWireError::NonCanonicalRouteVector);
}
Ok(())
}
}
#[derive(Clone, Copy, Debug, Eq, PartialEq, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct SubscriptionShardCursorV1 {
pub collection_index_ref: CollectionIndexRef,
pub shard_id: ShardId,
pub recovery_era: RecoveryEra,
pub seq_no: SeqNo,
}
#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
#[serde(tag = "kind", rename_all = "snake_case", deny_unknown_fields)]
pub enum SubscriptionFilterV1 {
All,
Bool {
#[serde(default)]
must: Vec<SubscriptionFilterV1>,
#[serde(default)]
should: Vec<SubscriptionFilterV1>,
#[serde(default)]
must_not: Vec<SubscriptionFilterV1>,
#[serde(default)]
minimum_should_match: usize,
},
Term {
field: SubscriptionFieldV1,
value: Value,
},
Terms {
field: SubscriptionFieldV1,
values: Vec<Value>,
},
Range {
field: SubscriptionFieldV1,
#[serde(default)]
gt: Option<Value>,
#[serde(default)]
gte: Option<Value>,
#[serde(default)]
lt: Option<Value>,
#[serde(default)]
lte: Option<Value>,
},
Exists {
field: SubscriptionFieldV1,
},
Prefix {
field: SubscriptionFieldV1,
value: String,
},
}
impl SubscriptionFilterV1 {
fn validate(&self, budget: &mut FilterBudget) -> Result<(), SubscriptionWireError> {
budget.nodes = budget.nodes.saturating_add(1);
if budget.nodes > MAX_SUBSCRIPTION_FILTER_NODES {
return Err(SubscriptionWireError::FilterNodes(budget.nodes));
}
match self {
Self::All => Ok(()),
Self::Bool {
must,
should,
must_not,
minimum_should_match,
} => {
let children = must.len().saturating_add(should.len()).saturating_add(must_not.len());
if children == 0 || *minimum_should_match > should.len() {
return Err(SubscriptionWireError::InvalidBool);
}
for child in must.iter().chain(should).chain(must_not) {
child.validate(budget)?;
}
Ok(())
}
Self::Term { field, value } => {
field.validate()?;
validate_scalar(value)?;
budget.add_terms(1)
}
Self::Terms { field, values } => {
field.validate()?;
if values.is_empty() {
return Err(SubscriptionWireError::EmptyTerms);
}
for value in values {
validate_scalar(value)?;
}
budget.add_terms(values.len())
}
Self::Range {
field,
gt,
gte,
lt,
lte,
} => {
field.validate()?;
if (gt.is_some() && gte.is_some()) || (lt.is_some() && lte.is_some()) {
return Err(SubscriptionWireError::InvalidRange);
}
let bounds = [gt, gte, lt, lte].into_iter().flatten().collect::<Vec<_>>();
if bounds.is_empty() {
return Err(SubscriptionWireError::InvalidRange);
}
for value in bounds {
validate_scalar(value)?;
}
Ok(())
}
Self::Exists { field } => field.validate(),
Self::Prefix { field, value } => {
field.validate()?;
validate_nonempty_bounded(value, MAX_SUBSCRIPTION_PATTERN_BYTES, "prefix")
}
}
}
#[must_use]
pub fn references_source(&self) -> bool {
match self {
Self::All => false,
Self::Bool {
must, should, must_not, ..
} => must.iter().chain(should).chain(must_not).any(Self::references_source),
Self::Term { field, .. }
| Self::Terms { field, .. }
| Self::Range { field, .. }
| Self::Exists { field }
| Self::Prefix { field, .. } => matches!(field, SubscriptionFieldV1::Source { .. }),
}
}
}
#[derive(Clone, Debug, Eq, PartialEq, Serialize, Deserialize)]
#[serde(tag = "kind", rename_all = "snake_case", deny_unknown_fields)]
pub enum SubscriptionFieldV1 {
Source { path: String },
Metadata { field: SubscriptionMetadataFieldV1 },
}
impl SubscriptionFieldV1 {
fn validate(&self) -> Result<(), SubscriptionWireError> {
match self {
Self::Source { path } => validate_field_path(path),
Self::Metadata { .. } => Ok(()),
}
}
}
#[derive(Clone, Copy, Debug, Eq, PartialEq, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum SubscriptionMetadataFieldV1 {
Id,
RoutingHash,
Operation,
Shard,
Sequence,
}
#[derive(Clone, Copy, Debug, Eq, PartialEq, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum SubscriptionOperationV1 {
Put,
Update,
Delete,
}
#[derive(Clone, Debug, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct SubscriptionMutationEventV1 {
pub format_version: u16,
pub collection_index: CollectionIndexRef,
pub shard_id: ShardId,
pub recovery_era: RecoveryEra,
pub writer_fence: FenceToken,
pub routing_revision: RoutingRevision,
pub seq_no: SeqNo,
pub operation: SubscriptionOperationV1,
pub document_id: DocumentId,
pub routing_hash: u32,
pub event_identity: [u8; 32],
#[serde(default, skip_serializing_if = "Option::is_none")]
pub source: Option<Box<RawValue>>,
}
impl PartialEq for SubscriptionMutationEventV1 {
fn eq(&self, other: &Self) -> bool {
self.format_version == other.format_version
&& self.collection_index == other.collection_index
&& self.shard_id == other.shard_id
&& self.recovery_era == other.recovery_era
&& self.writer_fence == other.writer_fence
&& self.routing_revision == other.routing_revision
&& self.seq_no == other.seq_no
&& self.operation == other.operation
&& self.document_id == other.document_id
&& self.routing_hash == other.routing_hash
&& self.event_identity == other.event_identity
&& self.source.as_deref().map(RawValue::get) == other.source.as_deref().map(RawValue::get)
}
}
#[derive(Clone, Debug, Eq, PartialEq, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum SubscriptionGapReasonV1 {
RetentionFloor,
RecoveryEraChanged,
RouteWatchGap,
RouteHandoffUnavailable,
SlowConsumer,
ReplayBudget,
EventTooLarge,
RegisteredWalUnavailable,
}
#[derive(Clone, Debug, Eq, PartialEq, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct SubscriptionGapV1 {
pub format_version: u16,
pub affected: Vec<SubscriptionShardCursorV1>,
pub last_confirmed: Vec<SubscriptionShardCursorV1>,
#[serde(default, skip_serializing_if = "Vec::is_empty")]
pub earliest_available: Vec<SubscriptionShardCursorV1>,
pub reason: SubscriptionGapReasonV1,
}
#[derive(Clone, Debug, Eq, PartialEq, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct SubscriptionAcceptedV1 {
pub format_version: u16,
pub routes: Vec<SubscriptionRouteIdentityV1>,
pub initial_cut: Vec<SubscriptionShardCursorV1>,
}
#[derive(Clone, Debug, Eq, PartialEq, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct SubscriptionHeartbeatV1 {
pub format_version: u16,
pub confirmed_cut: Vec<SubscriptionShardCursorV1>,
pub route_revision: u64,
pub authorization_revision: u64,
pub authorization_valid_until_millis: u64,
}
#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
#[serde(tag = "kind", content = "body", rename_all = "snake_case", deny_unknown_fields)]
pub enum SubscriptionFrameV1 {
Accepted(SubscriptionAcceptedV1),
Event(SubscriptionMutationEventV1),
Gap(SubscriptionGapV1),
Heartbeat(SubscriptionHeartbeatV1),
Closed { format_version: u16, reason: String },
}
impl SubscriptionFrameV1 {
pub fn validate(&self) -> Result<(), SubscriptionWireError> {
match self {
Self::Accepted(accepted) => {
validate_frame_version(accepted.format_version)?;
SubscriptionStartV1::Resume {
routes: accepted.routes.clone(),
cut: accepted.initial_cut.clone(),
}
.validate()
}
Self::Event(event) => {
validate_frame_version(event.format_version)?;
if event.seq_no == SeqNo::new(0)
|| event.event_identity == [0; 32]
|| (event.operation == SubscriptionOperationV1::Delete && event.source.is_some())
|| (event.operation != SubscriptionOperationV1::Delete && event.source.is_none())
|| event.source.as_ref().is_some_and(|source| {
serde_json::from_str::<serde_json::Map<String, Value>>(source.get()).is_err()
})
{
return Err(SubscriptionWireError::InvalidFrame("mutation event"));
}
Ok(())
}
Self::Gap(gap) => {
validate_frame_version(gap.format_version)?;
if gap.affected.is_empty()
|| gap.last_confirmed.is_empty()
|| !valid_frame_cursor_vector(&gap.affected)
|| !valid_frame_cursor_vector(&gap.last_confirmed)
|| (!gap.earliest_available.is_empty() && !valid_frame_cursor_vector(&gap.earliest_available))
{
return Err(SubscriptionWireError::InvalidFrame("gap"));
}
Ok(())
}
Self::Heartbeat(heartbeat) => {
validate_frame_version(heartbeat.format_version)?;
if heartbeat.confirmed_cut.is_empty()
|| !valid_frame_cursor_vector(&heartbeat.confirmed_cut)
|| heartbeat.route_revision == 0
|| heartbeat.authorization_revision == 0
|| heartbeat.authorization_valid_until_millis == 0
{
return Err(SubscriptionWireError::InvalidFrame("heartbeat"));
}
Ok(())
}
Self::Closed { format_version, reason } => {
validate_frame_version(*format_version)?;
validate_nonempty_bounded(reason, MAX_SUBSCRIPTION_FIELD_BYTES, "close reason")
}
}
}
}
pub fn parse_subscription_frame_v1(bytes: &[u8]) -> Result<SubscriptionFrameV1, SubscriptionWireError> {
if bytes.len() > MAX_SUBSCRIPTION_FRAME_BYTES {
return Err(SubscriptionWireError::FrameBytes(bytes.len()));
}
let frame: SubscriptionFrameV1 =
serde_json::from_slice(bytes).map_err(|error| SubscriptionWireError::Json(error.to_string()))?;
frame.validate()?;
Ok(frame)
}
fn validate_frame_version(version: u16) -> Result<(), SubscriptionWireError> {
if version != SUBSCRIPTION_FORMAT_VERSION {
return Err(SubscriptionWireError::UnsupportedVersion(version));
}
Ok(())
}
fn valid_frame_cursor_vector(cursors: &[SubscriptionShardCursorV1]) -> bool {
if cursors.len() > MAX_SUBSCRIPTION_ROUTE_SHARDS {
return false;
}
cursors
.windows(2)
.all(|pair| (pair[0].collection_index_ref, pair[0].shard_id) < (pair[1].collection_index_ref, pair[1].shard_id))
}
#[derive(Default)]
struct FilterBudget {
nodes: usize,
terms: usize,
}
impl FilterBudget {
fn add_terms(&mut self, count: usize) -> Result<(), SubscriptionWireError> {
self.terms = self.terms.saturating_add(count);
if self.terms > MAX_SUBSCRIPTION_FILTER_TERMS {
return Err(SubscriptionWireError::FilterTerms(self.terms));
}
Ok(())
}
}
fn validate_nonempty_bounded(value: &str, maximum: usize, field: &'static str) -> Result<(), SubscriptionWireError> {
if value.is_empty() || value.len() > maximum {
return Err(SubscriptionWireError::StringBound {
field,
actual: value.len(),
maximum,
});
}
Ok(())
}
fn validate_field_path(path: &str) -> Result<(), SubscriptionWireError> {
validate_nonempty_bounded(path, MAX_SUBSCRIPTION_FIELD_BYTES, "field path")?;
if path.starts_with('.')
|| path.ends_with('.')
|| path.split('.').any(|segment| {
segment.is_empty()
|| !segment
.bytes()
.all(|byte| byte.is_ascii_alphanumeric() || matches!(byte, b'_' | b'-'))
})
{
return Err(SubscriptionWireError::InvalidFieldPath);
}
Ok(())
}
fn validate_scalar(value: &Value) -> Result<(), SubscriptionWireError> {
match value {
Value::Null | Value::Bool(_) | Value::Number(_) | Value::String(_) => Ok(()),
Value::Array(_) | Value::Object(_) => Err(SubscriptionWireError::NonScalarTerm),
}
}
#[derive(Clone, Debug, Eq, PartialEq)]
pub enum SubscriptionWireError {
UnsupportedVersion(u16),
RequestBytes(usize),
FrameBytes(usize),
IndexPatternCount(usize),
SourceProjectionCount(usize),
StringBound {
field: &'static str,
actual: usize,
maximum: usize,
},
InvalidPattern,
InvalidFieldPath,
FilterNodes(usize),
FilterTerms(usize),
FilterBytes(usize),
InvalidBool,
EmptyTerms,
InvalidRange,
NonScalarTerm,
FilteredDeletesRequirePriorImage,
RouteCount(usize),
RouteShardCount(usize),
CursorCount(usize),
NonCanonicalRouteVector,
NonCanonicalCursorVector,
IncompleteCursorVector,
InvalidFrame(&'static str),
Json(String),
}
impl Display for SubscriptionWireError {
fn fmt(&self, formatter: &mut Formatter<'_>) -> fmt::Result {
match self {
Self::UnsupportedVersion(version) => write!(formatter, "unsupported subscription format version {version}"),
Self::RequestBytes(actual) => write!(formatter, "subscription request is too large: {actual} bytes"),
Self::FrameBytes(actual) => write!(formatter, "subscription frame is too large: {actual} bytes"),
Self::IndexPatternCount(actual) => write!(formatter, "invalid subscription index-pattern count {actual}"),
Self::SourceProjectionCount(actual) => write!(formatter, "invalid source projection count {actual}"),
Self::StringBound { field, actual, maximum } => {
write!(formatter, "{field} length {actual} is outside 1..={maximum}")
}
Self::InvalidPattern => formatter.write_str("subscription index pattern contains a control byte"),
Self::InvalidFieldPath => formatter.write_str("subscription field path is not canonical"),
Self::FilterNodes(actual) => write!(formatter, "subscription filter has too many nodes: {actual}"),
Self::FilterTerms(actual) => write!(formatter, "subscription filter has too many terms: {actual}"),
Self::FilterBytes(actual) => write!(formatter, "subscription filter is too large: {actual} bytes"),
Self::InvalidBool => formatter.write_str("subscription bool filter is empty or has an invalid minimum"),
Self::EmptyTerms => formatter.write_str("subscription terms filter is empty"),
Self::InvalidRange => formatter.write_str("subscription range filter has invalid bounds"),
Self::NonScalarTerm => formatter.write_str("subscription term or range value must be scalar"),
Self::FilteredDeletesRequirePriorImage => {
formatter.write_str("source-filtered delete delivery would require an unavailable prior image")
}
Self::RouteCount(actual) => write!(formatter, "invalid subscription route count {actual}"),
Self::RouteShardCount(actual) => write!(formatter, "invalid subscription route shard count {actual}"),
Self::CursorCount(actual) => write!(formatter, "invalid subscription cursor count {actual}"),
Self::NonCanonicalRouteVector => formatter.write_str("subscription route vector is not canonical"),
Self::NonCanonicalCursorVector => formatter.write_str("subscription cursor vector is not canonical"),
Self::IncompleteCursorVector => formatter.write_str("subscription cursor vector does not match routes"),
Self::InvalidFrame(frame) => write!(formatter, "subscription {frame} frame is invalid"),
Self::Json(error) => write!(formatter, "invalid subscription JSON: {error}"),
}
}
}
impl Error for SubscriptionWireError {}
#[cfg(test)]
mod tests {
use lodestone_types::{CollectionId, CollectionIndexRef, IndexId, RecoveryEra, RoutingRevision, SeqNo, ShardId};
use super::{
FilterSubscriptionV1, SubscriptionAcceptedV1, SubscriptionFieldV1, SubscriptionFilterV1, SubscriptionFrameV1,
SubscriptionMetadataFieldV1, SubscriptionRouteIdentityV1, SubscriptionShardCursorV1, SubscriptionStartV1,
SubscriptionWireError, parse_filter_subscription, parse_subscription_frame_v1,
};
use crate::RequestId;
fn scope() -> Result<CollectionIndexRef, Box<dyn std::error::Error>> {
Ok(CollectionIndexRef::new(CollectionId::new(7)?, IndexId::new(9)?))
}
fn request() -> Result<FilterSubscriptionV1, Box<dyn std::error::Error>> {
Ok(FilterSubscriptionV1 {
format_version: 1,
request_id: RequestId::new("sub-1")?,
index_patterns: vec!["logs-*".to_owned()],
filter: Some(SubscriptionFilterV1::Term {
field: SubscriptionFieldV1::Metadata {
field: SubscriptionMetadataFieldV1::Operation,
},
value: serde_json::json!("put"),
}),
include_deletes: true,
source_projection: vec!["service.name".to_owned()],
reconnect_attempt: 1,
start: SubscriptionStartV1::FromNow,
})
}
#[test]
fn request_round_trips_and_refuses_unknown_or_newer_shapes() -> Result<(), Box<dyn std::error::Error>> {
let request = request()?;
let bytes = serde_json::to_vec(&request)?;
assert_eq!(parse_filter_subscription(&bytes)?, request);
let mut newer: serde_json::Value = serde_json::from_slice(&bytes)?;
newer["format_version"] = serde_json::json!(2);
assert!(matches!(
parse_filter_subscription(&serde_json::to_vec(&newer)?),
Err(SubscriptionWireError::UnsupportedVersion(2))
));
let mut smuggled: serde_json::Value = serde_json::from_slice(&bytes)?;
smuggled["collection_id"] = serde_json::json!(44);
assert!(matches!(
parse_filter_subscription(&serde_json::to_vec(&smuggled)?),
Err(SubscriptionWireError::Json(_))
));
Ok(())
}
#[test]
fn source_filtered_deletes_are_refused() -> Result<(), Box<dyn std::error::Error>> {
let mut request = request()?;
request.filter = Some(SubscriptionFilterV1::Exists {
field: SubscriptionFieldV1::Source {
path: "service.name".to_owned(),
},
});
assert_eq!(
request.validate(),
Err(SubscriptionWireError::FilteredDeletesRequirePriorImage)
);
Ok(())
}
#[test]
fn resume_requires_one_canonical_cursor_per_route_shard() -> Result<(), Box<dyn std::error::Error>> {
let mut request = request()?;
request.start = SubscriptionStartV1::Resume {
routes: vec![SubscriptionRouteIdentityV1 {
collection_index_ref: scope()?,
routing_revision: RoutingRevision::new(3),
shard_ids: vec![ShardId::new(1), ShardId::new(2)],
}],
cut: vec![SubscriptionShardCursorV1 {
collection_index_ref: scope()?,
shard_id: ShardId::new(1),
recovery_era: RecoveryEra::new(4)?,
seq_no: SeqNo::new(8),
}],
};
assert_eq!(request.validate(), Err(SubscriptionWireError::IncompleteCursorVector));
Ok(())
}
#[test]
fn response_frames_refuse_newer_versions_and_invalid_vectors() -> Result<(), Box<dyn std::error::Error>> {
let route = SubscriptionRouteIdentityV1 {
collection_index_ref: scope()?,
routing_revision: RoutingRevision::new(3),
shard_ids: vec![ShardId::new(1)],
};
let cursor = SubscriptionShardCursorV1 {
collection_index_ref: scope()?,
shard_id: ShardId::new(1),
recovery_era: RecoveryEra::new(4)?,
seq_no: SeqNo::new(8),
};
let accepted = SubscriptionFrameV1::Accepted(SubscriptionAcceptedV1 {
format_version: 1,
routes: vec![route],
initial_cut: vec![cursor],
});
assert_eq!(parse_subscription_frame_v1(&serde_json::to_vec(&accepted)?)?, accepted);
let newer = SubscriptionFrameV1::Closed {
format_version: 2,
reason: "closed".to_owned(),
};
assert!(matches!(
parse_subscription_frame_v1(&serde_json::to_vec(&newer)?),
Err(SubscriptionWireError::UnsupportedVersion(2))
));
Ok(())
}
}