use std::error::Error;
use std::fmt::{self, Display, Formatter};
use std::num::NonZeroU128;
use serde::de::{DeserializeOwned, Visitor};
use serde::{Deserialize, Deserializer, Serialize, Serializer};
use crate::{
ClusterRevision, CollectionIndexRef, FieldPathDigestV1, ManifestGeneration, PhysicalObjectId, PolicyGeneration,
RootSetRef, SchemaGeneration, SeqNo, ShardId, SourceCutV1, SourceIncarnation,
};
pub const LIFECYCLE_COMPLETION_FORMAT_VERSION: u16 = 2;
pub const EXTERNAL_KEY_FORMAT_VERSION: u16 = 3;
pub const MAX_LIFECYCLE_CONTRACT_BYTES: usize = 1_048_576;
pub const MAX_OPAQUE_EXTERNAL_KEY_BINDING_BYTES: usize = 4 * 1_024;
pub const MAX_ROLLUP_DIMENSIONS: usize = 32;
pub const MAX_ROLLUP_MEASURES: usize = 64;
pub const MAX_ROLLUP_COVERAGES: usize = 4_096;
pub const MAX_ROLLUP_SHARDS: usize = 512;
pub const MAX_ROLLUP_OUTPUT_ROWS: usize = 262_144;
pub const MAX_ROLLUP_OUTPUT_BYTES: usize = 16 * 1_024 * 1_024;
pub const MAX_ROLLUP_SCALAR_BYTES: usize = 4_096;
pub const MAX_ROLLUP_BUCKET_MILLIS: u64 = 366 * 24 * 60 * 60 * 1_000;
pub const MAX_CONTENT_KEY_CACHE_MILLIS: u64 = 300_000;
pub const ROLLUP_QUERY_CONTRACT_DIGEST_V1: [u8; 32] = [
0x7c, 0x3e, 0x41, 0x11, 0x81, 0xaa, 0x97, 0x39, 0x2d, 0x63, 0x22, 0xcd, 0xd8, 0xf0, 0x02, 0x07, 0xea, 0x5f, 0x06,
0xa7, 0x90, 0x6b, 0x28, 0x81, 0x42, 0x89, 0x9c, 0xb8, 0x02, 0x17, 0x57, 0x80,
];
fn digest_is_nonzero(value: &[u8; 32]) -> bool {
value.iter().any(|byte| *byte != 0)
}
fn require_digest(value: &[u8; 32], name: &'static str) -> Result<(), LifecycleContractError> {
if digest_is_nonzero(value) {
Ok(())
} else {
Err(LifecycleContractError::Violation(name))
}
}
macro_rules! semantic_digest {
($name:ident, $description:literal) => {
#[doc = $description]
#[derive(Clone, Copy, Debug, Deserialize, Eq, Hash, Ord, PartialEq, PartialOrd, Serialize)]
#[serde(transparent)]
pub struct $name([u8; 32]);
impl $name {
pub fn new(value: [u8; 32]) -> Result<Self, LifecycleContractError> {
require_digest(&value, concat!(stringify!($name), " cannot be zero"))?;
Ok(Self(value))
}
#[must_use]
pub const fn get(self) -> [u8; 32] {
self.0
}
fn validate(self) -> Result<(), LifecycleContractError> {
require_digest(&self.0, concat!(stringify!($name), " cannot be zero"))
}
}
};
}
macro_rules! nonzero_decimal_identity {
($name:ident, $description:literal) => {
#[doc = $description]
#[derive(Clone, Copy, Debug, Eq, Hash, Ord, PartialEq, PartialOrd)]
pub struct $name(NonZeroU128);
impl $name {
pub fn new(value: u128) -> Result<Self, LifecycleContractError> {
NonZeroU128::new(value)
.map(Self)
.ok_or(LifecycleContractError::ZeroIdentity(stringify!($name)))
}
#[must_use]
pub const fn get(self) -> u128 {
self.0.get()
}
}
impl Display for $name {
fn fmt(&self, formatter: &mut Formatter<'_>) -> fmt::Result {
Display::fmt(&self.get(), formatter)
}
}
impl Serialize for $name {
fn serialize<S>(&self, serializer: S) -> Result<S::Ok, S::Error>
where
S: Serializer,
{
serializer.serialize_str(&self.get().to_string())
}
}
impl<'de> Deserialize<'de> for $name {
fn deserialize<D>(deserializer: D) -> Result<Self, D::Error>
where
D: Deserializer<'de>,
{
struct CanonicalDecimalVisitor;
impl Visitor<'_> for CanonicalDecimalVisitor {
type Value = $name;
fn expecting(&self, formatter: &mut Formatter<'_>) -> fmt::Result {
formatter.write_str("a canonical nonzero u128 decimal string")
}
fn visit_str<E>(self, value: &str) -> Result<Self::Value, E>
where
E: serde::de::Error,
{
let bytes = value.as_bytes();
if bytes.is_empty()
|| !matches!(bytes[0], b'1'..=b'9')
|| !bytes[1..].iter().all(u8::is_ascii_digit)
{
return Err(E::custom("expected a canonical nonzero u128 decimal string"));
}
let parsed = value
.parse::<u128>()
.map_err(|_| E::custom("identity exceeds u128"))?;
NonZeroU128::new(parsed)
.map($name)
.ok_or_else(|| E::custom("identity cannot be zero"))
}
}
deserializer.deserialize_str(CanonicalDecimalVisitor)
}
}
};
}
semantic_digest!(
ExternalKeyContextDigestV1,
"Digest of the exact non-secret external-key context."
);
semantic_digest!(
RollupQueryContractDigestV1,
"Digest of the admitted query shapes proven by the rollup planner."
);
semantic_digest!(
RollupCoverageDigestV1,
"Digest of the exact coverage/root state proven safe for routing."
);
nonzero_decimal_identity!(ContentKeyIdV1, "Never-reused identity of one content-encryption key.");
nonzero_decimal_identity!(
ExternalKeyBindingIdV1,
"Never-reused identity of one opaque index-authorized external-key binding."
);
nonzero_decimal_identity!(
RollupPolicyIdV1,
"Never-reused identity of one aggregate-then-delete policy."
);
#[derive(Clone, Copy, Debug, Deserialize, Eq, Hash, PartialEq, Serialize)]
#[serde(rename_all = "snake_case")]
pub enum DocumentEncryptionAlgorithmV1 {
Aes256Gcm,
}
#[derive(Clone, Copy, Debug, Deserialize, Eq, PartialEq, Serialize)]
#[serde(deny_unknown_fields)]
pub struct ExternalKeyContextV1 {
pub format_version: u16,
pub collection_index_ref: CollectionIndexRef,
pub content_key_id: ContentKeyIdV1,
pub content_key_generation: u64,
pub binding_id: ExternalKeyBindingIdV1,
pub binding_generation: u64,
pub access_generation: u64,
pub object_context_digest: [u8; 32],
}
impl ExternalKeyContextV1 {
pub fn validate(self) -> Result<(), LifecycleContractError> {
validate_external_key_format(self.format_version)?;
if self.content_key_generation == 0 || self.binding_generation == 0 || self.access_generation == 0 {
return Err(LifecycleContractError::Violation(
"external-key generations cannot be zero",
));
}
require_digest(
&self.object_context_digest,
"external-key object context cannot be zero",
)
}
pub fn digest(self) -> Result<ExternalKeyContextDigestV1, LifecycleContractError> {
self.validate()?;
let bytes = serde_json::to_vec(&self).map_err(LifecycleContractError::serde)?;
ExternalKeyContextDigestV1::new(*blake3::hash(&bytes).as_bytes())
}
}
#[derive(Clone, Deserialize, Eq, PartialEq, Serialize)]
#[serde(deny_unknown_fields)]
pub struct IndexExternalKeyBindingV1 {
pub format_version: u16,
pub binding_id: ExternalKeyBindingIdV1,
pub collection_index_ref: CollectionIndexRef,
pub content_key_id: ContentKeyIdV1,
pub content_key_generation: u64,
pub binding_generation: u64,
pub access_generation: u64,
pub opaque_binding: Vec<u8>,
pub context: ExternalKeyContextV1,
pub context_digest: ExternalKeyContextDigestV1,
pub created_at_revision: ClusterRevision,
}
impl fmt::Debug for IndexExternalKeyBindingV1 {
fn fmt(&self, formatter: &mut Formatter<'_>) -> fmt::Result {
formatter
.debug_struct("IndexExternalKeyBindingV1")
.field("format_version", &self.format_version)
.field("binding_id", &self.binding_id)
.field("collection_index_ref", &self.collection_index_ref)
.field("content_key_id", &self.content_key_id)
.field("content_key_generation", &self.content_key_generation)
.field("binding_generation", &self.binding_generation)
.field("access_generation", &self.access_generation)
.field("opaque_binding", &"[redacted]")
.field("context", &self.context)
.field("context_digest", &self.context_digest)
.field("created_at_revision", &self.created_at_revision)
.finish()
}
}
impl IndexExternalKeyBindingV1 {
pub fn validate(&self) -> Result<(), LifecycleContractError> {
validate_external_key_format(self.format_version)?;
if self.content_key_generation == 0
|| self.binding_generation == 0
|| self.access_generation == 0
|| self.created_at_revision.get() == 0
{
return Err(LifecycleContractError::Violation(
"external-key binding requires nonzero generations",
));
}
if self.opaque_binding.is_empty()
|| self.opaque_binding.len() > MAX_OPAQUE_EXTERNAL_KEY_BINDING_BYTES
|| self.opaque_binding.iter().all(|byte| *byte == 0)
{
return Err(LifecycleContractError::Budget {
field: "opaque external-key binding bytes",
observed: self.opaque_binding.len() as u64,
maximum: MAX_OPAQUE_EXTERNAL_KEY_BINDING_BYTES as u64,
});
}
self.context.validate()?;
self.context_digest.validate()?;
if self.context.collection_index_ref != self.collection_index_ref
|| self.context.content_key_id != self.content_key_id
|| self.context.content_key_generation != self.content_key_generation
|| self.context.binding_id != self.binding_id
|| self.context.binding_generation != self.binding_generation
|| self.context.access_generation != self.access_generation
|| self.context.digest()? != self.context_digest
{
return Err(LifecycleContractError::Violation(
"external-key binding and context identities differ",
));
}
Ok(())
}
}
#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
#[serde(deny_unknown_fields)]
pub struct EncryptedObjectKeyRefV1 {
pub format_version: u16,
pub physical_object: PhysicalObjectId,
pub algorithm: DocumentEncryptionAlgorithmV1,
pub content_key_id: ContentKeyIdV1,
pub content_key_generation: u64,
pub nonce: [u8; 12],
pub authenticated_context_digest: [u8; 32],
pub ciphertext_bytes: u64,
}
impl EncryptedObjectKeyRefV1 {
pub fn validate(&self) -> Result<(), LifecycleContractError> {
validate_external_key_format(self.format_version)?;
self.physical_object
.validate()
.map_err(|error| LifecycleContractError::Nested(error.to_string()))?;
if self.content_key_generation == 0 || self.ciphertext_bytes < 16 {
return Err(LifecycleContractError::Violation(
"encrypted objects require a nonzero key generation and an authentication tag",
));
}
if self.nonce != object_nonce_v1(self.content_key_id, self.content_key_generation, &self.physical_object)? {
return Err(LifecycleContractError::Violation(
"encrypted-object nonce is not derived from its never-reused physical identity",
));
}
let expected = object_authenticated_context_digest_v1(
self.algorithm,
self.content_key_id,
self.content_key_generation,
&self.physical_object,
self.ciphertext_bytes,
)?;
if self.authenticated_context_digest != expected {
return Err(LifecycleContractError::Violation(
"encrypted-object authenticated context does not exact-match metadata",
));
}
Ok(())
}
}
pub fn object_nonce_v1(
content_key_id: ContentKeyIdV1,
content_key_generation: u64,
physical_object: &PhysicalObjectId,
) -> Result<[u8; 12], LifecycleContractError> {
if content_key_generation == 0 {
return Err(LifecycleContractError::Violation(
"content-key generation cannot be zero",
));
}
physical_object
.validate()
.map_err(|error| LifecycleContractError::Nested(error.to_string()))?;
let mut hasher = blake3::Hasher::new();
hasher.update(b"lodestone/p13/object-nonce/v1\0");
hasher.update(&content_key_id.get().to_be_bytes());
hasher.update(&content_key_generation.to_be_bytes());
hasher.update(&serde_json::to_vec(physical_object).map_err(LifecycleContractError::serde)?);
let digest = hasher.finalize();
let mut nonce = [0_u8; 12];
nonce.copy_from_slice(&digest.as_bytes()[..12]);
Ok(nonce)
}
pub fn object_authenticated_context_digest_v1(
algorithm: DocumentEncryptionAlgorithmV1,
content_key_id: ContentKeyIdV1,
content_key_generation: u64,
physical_object: &PhysicalObjectId,
ciphertext_bytes: u64,
) -> Result<[u8; 32], LifecycleContractError> {
if content_key_generation == 0 || ciphertext_bytes < 16 {
return Err(LifecycleContractError::Violation(
"authenticated context requires a key generation and ciphertext tag",
));
}
physical_object
.validate()
.map_err(|error| LifecycleContractError::Nested(error.to_string()))?;
let binding = (
EXTERNAL_KEY_FORMAT_VERSION,
algorithm,
content_key_id,
content_key_generation,
physical_object,
ciphertext_bytes,
);
let mut hasher = blake3::Hasher::new();
hasher.update(b"lodestone/p13/object-aad/v1\0");
hasher.update(&serde_json::to_vec(&binding).map_err(LifecycleContractError::serde)?);
Ok(*hasher.finalize().as_bytes())
}
#[derive(Clone, Copy, Debug, Deserialize, Eq, Hash, Ord, PartialEq, PartialOrd, Serialize)]
#[serde(rename_all = "snake_case")]
pub enum RollupTimeZoneV1 {
Utc,
}
#[derive(Clone, Copy, Debug, Deserialize, Eq, Hash, Ord, PartialEq, PartialOrd, Serialize)]
#[serde(tag = "kind", rename_all = "snake_case", deny_unknown_fields)]
pub enum RollupMeasureV1 {
Count,
Sum { field_path_digest: FieldPathDigestV1 },
Min { field_path_digest: FieldPathDigestV1 },
Max { field_path_digest: FieldPathDigestV1 },
}
#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
#[serde(deny_unknown_fields)]
pub struct RollupPolicyV1 {
pub format_version: u16,
pub collection_index_ref: CollectionIndexRef,
pub policy_id: RollupPolicyIdV1,
pub policy_generation: PolicyGeneration,
pub detail_retirement_acknowledged: bool,
pub time_zone: RollupTimeZoneV1,
pub timestamp_field_digest: FieldPathDigestV1,
pub bucket_width_millis: u64,
pub bucket_alignment_millis: u64,
pub dimensions: Vec<FieldPathDigestV1>,
pub measures: Vec<RollupMeasureV1>,
}
impl RollupPolicyV1 {
pub fn validate(&self) -> Result<(), LifecycleContractError> {
validate_format(self.format_version)?;
if self.policy_generation.get() == 0
|| !self.detail_retirement_acknowledged
|| self.bucket_width_millis == 0
|| self.bucket_width_millis > MAX_ROLLUP_BUCKET_MILLIS
|| self.bucket_alignment_millis >= self.bucket_width_millis
{
return Err(LifecycleContractError::Violation(
"rollup generation or bucket geometry is outside the current contract",
));
}
require_digest(
&self.timestamp_field_digest.get(),
"rollup timestamp-field digest cannot be zero",
)?;
if self.dimensions.len() > MAX_ROLLUP_DIMENSIONS
|| self.dimensions.windows(2).any(|pair| pair[0] >= pair[1])
|| self.measures.is_empty()
|| self.measures.len() > MAX_ROLLUP_MEASURES
|| self.measures.windows(2).any(|pair| pair[0] >= pair[1])
{
return Err(LifecycleContractError::Violation(
"rollup dimensions and measures must be bounded, unique, and canonically ordered",
));
}
for dimension in &self.dimensions {
require_digest(&dimension.get(), "rollup dimension digest cannot be zero")?;
}
for measure in &self.measures {
match measure {
RollupMeasureV1::Count => {}
RollupMeasureV1::Sum { field_path_digest }
| RollupMeasureV1::Min { field_path_digest }
| RollupMeasureV1::Max { field_path_digest } => {
require_digest(&field_path_digest.get(), "rollup measure digest cannot be zero")?;
}
}
}
Ok(())
}
pub fn boundary_aligned(&self, value: u64) -> bool {
value >= self.bucket_alignment_millis
&& (value - self.bucket_alignment_millis).is_multiple_of(self.bucket_width_millis)
}
}
#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
#[serde(deny_unknown_fields)]
pub struct RollupWorkPlanV1 {
pub format_version: u16,
pub policy: RollupPolicyV1,
pub schema_generation: SchemaGeneration,
pub source_incarnation: Option<SourceIncarnation>,
pub source_cut: Option<SourceCutV1>,
pub source_generations: Vec<RollupSourceGenerationV1>,
pub range_start_millis: u64,
pub range_end_millis: u64,
pub late_data_watermark_millis: u64,
}
impl RollupWorkPlanV1 {
pub fn validate(&self) -> Result<(), LifecycleContractError> {
validate_format(self.format_version)?;
self.policy.validate()?;
if self.schema_generation.get() == 0
|| self.range_start_millis >= self.range_end_millis
|| !self.policy.boundary_aligned(self.range_start_millis)
|| !self.policy.boundary_aligned(self.range_end_millis)
|| self.late_data_watermark_millis < self.range_end_millis
{
return Err(LifecycleContractError::Violation(
"rollup work plan schema, aligned range, or late-data watermark is invalid",
));
}
if self.source_generations.is_empty()
|| self.source_generations.len() > MAX_ROLLUP_SHARDS
|| self
.source_generations
.windows(2)
.any(|pair| pair[0].shard_id >= pair[1].shard_id)
|| self
.source_generations
.iter()
.any(|source| source.manifest_generation.get() == 0)
{
return Err(LifecycleContractError::Violation(
"rollup source generations must be complete, bounded, and canonically ordered",
));
}
match (&self.source_incarnation, &self.source_cut) {
(None, None) => Ok(()),
(Some(incarnation), Some(cut)) if *incarnation == cut.source_incarnation => cut
.validate()
.map_err(|error| LifecycleContractError::Nested(error.to_string())),
_ => Err(LifecycleContractError::Violation(
"replayable-source rollup work requires an exact matching source cut",
)),
}
}
pub fn digest(&self) -> Result<[u8; 32], LifecycleContractError> {
self.validate()?;
let bytes = serde_json::to_vec(self).map_err(LifecycleContractError::serde)?;
let mut hasher = blake3::Hasher::new();
hasher.update(b"lodestone/p13/rollup-work-plan/v1\0");
hasher.update(&bytes);
Ok(*hasher.finalize().as_bytes())
}
}
#[derive(Clone, Debug, Deserialize, Eq, Ord, PartialEq, PartialOrd, Serialize)]
#[serde(tag = "kind", content = "value", rename_all = "snake_case", deny_unknown_fields)]
pub enum RollupScalarV1 {
Null,
Bool(bool),
I64(i64),
U64(u64),
String(String),
}
impl RollupScalarV1 {
fn validate(&self) -> Result<(), LifecycleContractError> {
if matches!(self, Self::String(value) if value.len() > MAX_ROLLUP_SCALAR_BYTES) {
return Err(LifecycleContractError::Budget {
field: "rollup dimension scalar bytes",
observed: match self {
Self::String(value) => value.len() as u64,
_ => 0,
},
maximum: MAX_ROLLUP_SCALAR_BYTES as u64,
});
}
Ok(())
}
}
#[derive(Clone, Copy, Debug, Deserialize, Eq, PartialEq, Serialize)]
#[serde(tag = "kind", content = "value", rename_all = "snake_case", deny_unknown_fields)]
pub enum RollupAggregateValueV1 {
Count(u64),
Sum(i64),
Min(i64),
Max(i64),
}
#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
#[serde(deny_unknown_fields)]
pub struct RollupOutputRowV1 {
pub bucket_start_millis: u64,
pub dimensions: Vec<RollupScalarV1>,
pub measures: Vec<RollupAggregateValueV1>,
}
#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
#[serde(deny_unknown_fields)]
pub struct RollupOutputV1 {
pub format_version: u16,
pub work_plan_digest: [u8; 32],
pub rows: Vec<RollupOutputRowV1>,
pub input_documents: u64,
pub input_bytes: u64,
}
impl RollupOutputV1 {
pub fn validate(&self, plan: &RollupWorkPlanV1) -> Result<(), LifecycleContractError> {
validate_format(self.format_version)?;
plan.validate()?;
if self.work_plan_digest != plan.digest()?
|| self.rows.is_empty()
|| self.rows.len() > MAX_ROLLUP_OUTPUT_ROWS
|| u64::try_from(self.rows.len()).is_err()
|| u64::try_from(self.rows.len()).is_ok_and(|rows| rows > self.input_documents)
|| self.input_documents == 0
|| self.input_bytes == 0
{
return Err(LifecycleContractError::Violation(
"rollup output does not bind the exact bounded work plan and input",
));
}
let mut previous_key: Option<(u64, &[RollupScalarV1])> = None;
let count_position = plan
.policy
.measures
.iter()
.position(|measure| *measure == RollupMeasureV1::Count);
let mut counted_documents = 0_u64;
for row in &self.rows {
if row.bucket_start_millis < plan.range_start_millis
|| row.bucket_start_millis >= plan.range_end_millis
|| !plan.policy.boundary_aligned(row.bucket_start_millis)
|| row.dimensions.len() != plan.policy.dimensions.len()
|| row.measures.len() != plan.policy.measures.len()
|| row.dimensions.iter().any(|value| value.validate().is_err())
|| row.measures.iter().zip(&plan.policy.measures).any(|(value, measure)| {
!matches!(
(value, measure),
(RollupAggregateValueV1::Count(_), RollupMeasureV1::Count)
| (RollupAggregateValueV1::Sum(_), RollupMeasureV1::Sum { .. })
| (RollupAggregateValueV1::Min(_), RollupMeasureV1::Min { .. })
| (RollupAggregateValueV1::Max(_), RollupMeasureV1::Max { .. })
)
})
|| previous_key.is_some_and(|prior| prior >= (row.bucket_start_millis, &row.dimensions))
{
return Err(LifecycleContractError::Violation(
"rollup rows must match policy shape and be canonically ordered",
));
}
if let Some(position) = count_position {
let RollupAggregateValueV1::Count(count) = row.measures[position] else {
return Err(LifecycleContractError::Violation(
"rollup count measure does not match the current policy",
));
};
if count == 0 {
return Err(LifecycleContractError::Violation(
"rollup groups cannot contain an empty count",
));
}
counted_documents = counted_documents
.checked_add(count)
.ok_or(LifecycleContractError::Violation("rollup count total overflowed"))?;
}
previous_key = Some((row.bucket_start_millis, &row.dimensions));
}
if count_position.is_some() && counted_documents != self.input_documents {
return Err(LifecycleContractError::Violation(
"rollup count total does not match the exact input document count",
));
}
Ok(())
}
}
#[derive(Clone, Copy, Debug, Deserialize, Eq, PartialEq, Serialize)]
#[serde(deny_unknown_fields)]
pub struct RollupSourceGenerationV1 {
pub shard_id: ShardId,
pub manifest_generation: ManifestGeneration,
pub correction_through: SeqNo,
}
#[derive(Clone, Copy, Debug, Deserialize, Eq, PartialEq, Serialize)]
#[serde(deny_unknown_fields)]
pub struct RollupRoutingProofV1 {
pub query_contract_digest: RollupQueryContractDigestV1,
pub coverage_digest: RollupCoverageDigestV1,
pub proven_at_revision: ClusterRevision,
}
impl RollupRoutingProofV1 {
pub fn validate(self) -> Result<(), LifecycleContractError> {
self.query_contract_digest.validate()?;
self.coverage_digest.validate()?;
if self.query_contract_digest != rollup_query_contract_digest_v1()? || self.proven_at_revision.get() == 0 {
return Err(LifecycleContractError::Violation(
"rollup routing proof requires the current exact query contract and a nonzero catalog revision",
));
}
Ok(())
}
}
pub fn rollup_query_contract_digest_v1() -> Result<RollupQueryContractDigestV1, LifecycleContractError> {
let mut hasher = blake3::Hasher::new();
hasher.update(b"lodestone/p13/rollup-query-contract/v1\0");
hasher.update(
b"utc-fixed-buckets;closed-count-sum-min-max;half-open-gap-free-non-overlapping;exact-match-or-raw-unavailable",
);
let observed = *hasher.finalize().as_bytes();
if observed != ROLLUP_QUERY_CONTRACT_DIGEST_V1 {
return Err(LifecycleContractError::Violation(
"rollup query semantics changed without a current-format digest revision",
));
}
RollupQueryContractDigestV1::new(observed)
}
#[derive(Clone, Copy, Debug, Deserialize, Eq, PartialEq, Serialize)]
#[serde(rename_all = "snake_case")]
pub enum RawRetirementStateV1 {
RawPresent,
RoutingProven,
RawRetired,
}
#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
#[serde(deny_unknown_fields)]
pub struct RollupCoverageV1 {
pub format_version: u16,
pub collection_index_ref: CollectionIndexRef,
pub policy: RollupPolicyV1,
pub schema_generation: SchemaGeneration,
pub source_incarnation: Option<SourceIncarnation>,
pub source_cut: Option<SourceCutV1>,
pub source_generations: Vec<RollupSourceGenerationV1>,
pub range_start_millis: u64,
pub range_end_millis: u64,
pub late_data_watermark_millis: u64,
pub output_root: RootSetRef,
pub raw_retirement_state: RawRetirementStateV1,
pub routing_proof: Option<RollupRoutingProofV1>,
}
impl RollupCoverageV1 {
pub fn coverage_digest(&self) -> Result<RollupCoverageDigestV1, LifecycleContractError> {
let binding = (
self.format_version,
self.collection_index_ref,
&self.policy,
self.schema_generation,
self.source_incarnation,
&self.source_cut,
&self.source_generations,
self.range_start_millis,
self.range_end_millis,
self.late_data_watermark_millis,
&self.output_root,
);
let mut hasher = blake3::Hasher::new();
hasher.update(b"lodestone/p13/rollup-coverage/v1\0");
hasher.update(&serde_json::to_vec(&binding).map_err(LifecycleContractError::serde)?);
RollupCoverageDigestV1::new(*hasher.finalize().as_bytes())
}
pub fn validate(&self) -> Result<(), LifecycleContractError> {
validate_format(self.format_version)?;
self.policy.validate()?;
if self.policy.collection_index_ref != self.collection_index_ref
|| self.schema_generation.get() == 0
|| self.range_start_millis >= self.range_end_millis
|| !self.policy.boundary_aligned(self.range_start_millis)
|| !self.policy.boundary_aligned(self.range_end_millis)
|| self.late_data_watermark_millis < self.range_end_millis
{
return Err(LifecycleContractError::Violation(
"rollup coverage scope, schema, aligned range, or late-data watermark is invalid",
));
}
if self.source_generations.is_empty()
|| self.source_generations.len() > MAX_ROLLUP_SHARDS
|| self
.source_generations
.windows(2)
.any(|pair| pair[0].shard_id >= pair[1].shard_id)
|| self
.source_generations
.iter()
.any(|source| source.manifest_generation.get() == 0)
{
return Err(LifecycleContractError::Violation(
"rollup source generations must be complete, bounded, and canonically ordered",
));
}
match (&self.source_incarnation, &self.source_cut) {
(None, None) => {}
(Some(incarnation), Some(cut)) if *incarnation == cut.source_incarnation => cut
.validate()
.map_err(|error| LifecycleContractError::Nested(error.to_string()))?,
_ => {
return Err(LifecycleContractError::Violation(
"replayable-source rollup coverage requires an exact matching source cut",
));
}
}
self.output_root
.validate()
.map_err(|error| LifecycleContractError::Nested(error.to_string()))?;
if self.output_root.item_count != 1
|| self.output_root.page_count != 1
|| self.output_root.schema_generations.as_slice() != [self.schema_generation]
{
return Err(LifecycleContractError::Violation(
"current rollup coverage requires one output object for the exact schema generation",
));
}
match (self.raw_retirement_state, self.routing_proof) {
(RawRetirementStateV1::RawPresent, None) => {}
(RawRetirementStateV1::RoutingProven | RawRetirementStateV1::RawRetired, Some(proof)) => {
proof.validate()?;
if proof.coverage_digest != self.coverage_digest()? {
return Err(LifecycleContractError::Violation(
"raw-retirement proof does not bind the exact current coverage and output root",
));
}
}
_ => {
return Err(LifecycleContractError::Violation(
"raw retirement requires an exact planner routing proof",
));
}
}
Ok(())
}
}
#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
#[serde(deny_unknown_fields)]
pub struct RollupCoverageLedgerV1 {
pub format_version: u16,
pub collection_index_ref: CollectionIndexRef,
pub revision: u64,
pub coverages: Vec<RollupCoverageV1>,
}
impl RollupCoverageLedgerV1 {
pub fn validate(&self) -> Result<(), LifecycleContractError> {
validate_format(self.format_version)?;
if self.revision == 0 || self.coverages.len() > MAX_ROLLUP_COVERAGES {
return Err(LifecycleContractError::Budget {
field: "rollup coverage rows or revision",
observed: self.coverages.len() as u64,
maximum: MAX_ROLLUP_COVERAGES as u64,
});
}
let mut prior_end = None;
for coverage in &self.coverages {
coverage.validate()?;
if coverage.collection_index_ref != self.collection_index_ref
|| prior_end.is_some_and(|end| end > coverage.range_start_millis)
|| coverage
.routing_proof
.is_some_and(|proof| proof.proven_at_revision.get() > self.revision)
{
return Err(LifecycleContractError::Violation(
"rollup coverages must be non-overlapping and canonically ordered for one index",
));
}
prior_end = Some(coverage.range_end_millis);
}
Ok(())
}
}
#[derive(Clone, Debug, Eq, PartialEq)]
pub struct RollupQueryShapeV1 {
pub range_start_millis: u64,
pub range_end_millis: u64,
pub time_zone: RollupTimeZoneV1,
pub bucket_width_millis: u64,
pub bucket_alignment_millis: u64,
pub dimensions: Vec<FieldPathDigestV1>,
pub measures: Vec<RollupMeasureV1>,
}
impl RollupQueryShapeV1 {
fn matches(&self, policy: &RollupPolicyV1) -> bool {
self.time_zone == policy.time_zone
&& self.bucket_width_millis == policy.bucket_width_millis
&& self.bucket_alignment_millis == policy.bucket_alignment_millis
&& self.dimensions == policy.dimensions
&& self.measures == policy.measures
&& policy.boundary_aligned(self.range_start_millis)
&& policy.boundary_aligned(self.range_end_millis)
}
}
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub enum RollupRouteKindV1 {
Raw,
Rollup,
}
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub struct RollupRouteSliceV1 {
pub range_start_millis: u64,
pub range_end_millis: u64,
pub kind: RollupRouteKindV1,
pub rollup_coverage_digest: Option<RollupCoverageDigestV1>,
}
pub fn route_rollup_query_v1(
ledger: &RollupCoverageLedgerV1,
query: &RollupQueryShapeV1,
) -> Result<Vec<RollupRouteSliceV1>, LifecycleContractError> {
ledger.validate()?;
if query.range_start_millis >= query.range_end_millis {
return Err(LifecycleContractError::Violation(
"rollup query range is empty or reversed",
));
}
let mut slices = Vec::new();
let mut cursor = query.range_start_millis;
for coverage in &ledger.coverages {
if coverage.range_end_millis <= cursor || coverage.range_start_millis >= query.range_end_millis {
continue;
}
if cursor < coverage.range_start_millis {
push_route_slice(
&mut slices,
cursor,
coverage.range_start_millis.min(query.range_end_millis),
RollupRouteKindV1::Raw,
None,
);
cursor = coverage.range_start_millis.min(query.range_end_millis);
}
if cursor >= query.range_end_millis {
break;
}
let end = coverage.range_end_millis.min(query.range_end_millis);
let exact_rollup = query.matches(&coverage.policy);
if !exact_rollup && coverage.raw_retirement_state == RawRetirementStateV1::RawRetired {
return Err(LifecycleContractError::DetailRetiredByPolicy);
}
let rollup_coverage_digest = exact_rollup.then(|| coverage.coverage_digest()).transpose()?;
push_route_slice(
&mut slices,
cursor.max(coverage.range_start_millis),
end,
if exact_rollup {
RollupRouteKindV1::Rollup
} else {
RollupRouteKindV1::Raw
},
rollup_coverage_digest,
);
cursor = end;
}
if cursor < query.range_end_millis {
push_route_slice(
&mut slices,
cursor,
query.range_end_millis,
RollupRouteKindV1::Raw,
None,
);
}
if slices.is_empty()
|| slices[0].range_start_millis != query.range_start_millis
|| slices
.last()
.is_none_or(|slice| slice.range_end_millis != query.range_end_millis)
|| slices
.windows(2)
.any(|pair| pair[0].range_end_millis != pair[1].range_start_millis)
|| slices
.iter()
.any(|slice| matches!(slice.kind, RollupRouteKindV1::Raw) != slice.rollup_coverage_digest.is_none())
{
return Err(LifecycleContractError::Violation(
"rollup routing did not cover the query exactly once",
));
}
Ok(slices)
}
fn push_route_slice(
slices: &mut Vec<RollupRouteSliceV1>,
start: u64,
end: u64,
kind: RollupRouteKindV1,
rollup_coverage_digest: Option<RollupCoverageDigestV1>,
) {
if start >= end {
return;
}
if let Some(previous) = slices.last_mut()
&& previous.kind == kind
&& previous.range_end_millis == start
&& previous.rollup_coverage_digest == rollup_coverage_digest
{
previous.range_end_millis = end;
return;
}
slices.push(RollupRouteSliceV1 {
range_start_millis: start,
range_end_millis: end,
kind,
rollup_coverage_digest,
});
}
pub fn encode_encrypted_object_key_ref_v1(value: &EncryptedObjectKeyRefV1) -> Result<Vec<u8>, LifecycleContractError> {
encode_current(value, EncryptedObjectKeyRefV1::validate)
}
pub fn decode_encrypted_object_key_ref_v1(bytes: &[u8]) -> Result<EncryptedObjectKeyRefV1, LifecycleContractError> {
decode_current(bytes, EncryptedObjectKeyRefV1::validate)
}
pub fn encode_index_external_key_binding_v1(
value: &IndexExternalKeyBindingV1,
) -> Result<Vec<u8>, LifecycleContractError> {
encode_current(value, IndexExternalKeyBindingV1::validate)
}
pub fn decode_index_external_key_binding_v1(bytes: &[u8]) -> Result<IndexExternalKeyBindingV1, LifecycleContractError> {
decode_current(bytes, IndexExternalKeyBindingV1::validate)
}
pub fn encode_rollup_policy_v1(value: &RollupPolicyV1) -> Result<Vec<u8>, LifecycleContractError> {
encode_current(value, RollupPolicyV1::validate)
}
pub fn decode_rollup_policy_v1(bytes: &[u8]) -> Result<RollupPolicyV1, LifecycleContractError> {
decode_current(bytes, RollupPolicyV1::validate)
}
pub fn encode_rollup_output_v1(
value: &RollupOutputV1,
plan: &RollupWorkPlanV1,
) -> Result<Vec<u8>, LifecycleContractError> {
value.validate(plan)?;
let encoded = serde_json::to_vec(value).map_err(LifecycleContractError::serde)?;
if encoded.len() > MAX_ROLLUP_OUTPUT_BYTES {
return Err(LifecycleContractError::Budget {
field: "rollup output bytes",
observed: encoded.len() as u64,
maximum: MAX_ROLLUP_OUTPUT_BYTES as u64,
});
}
Ok(encoded)
}
pub fn decode_rollup_output_v1(
bytes: &[u8],
plan: &RollupWorkPlanV1,
) -> Result<RollupOutputV1, LifecycleContractError> {
if bytes.is_empty() || bytes.len() > MAX_ROLLUP_OUTPUT_BYTES {
return Err(LifecycleContractError::Budget {
field: "rollup output bytes",
observed: bytes.len() as u64,
maximum: MAX_ROLLUP_OUTPUT_BYTES as u64,
});
}
let value: RollupOutputV1 = serde_json::from_slice(bytes).map_err(LifecycleContractError::serde)?;
value.validate(plan)?;
let canonical = serde_json::to_vec(&value).map_err(LifecycleContractError::serde)?;
if canonical != bytes {
return Err(LifecycleContractError::Violation(
"rollup output is not the canonical current-format encoding",
));
}
Ok(value)
}
pub fn encode_rollup_coverage_ledger_v1(value: &RollupCoverageLedgerV1) -> Result<Vec<u8>, LifecycleContractError> {
encode_current(value, RollupCoverageLedgerV1::validate)
}
pub fn decode_rollup_coverage_ledger_v1(bytes: &[u8]) -> Result<RollupCoverageLedgerV1, LifecycleContractError> {
decode_current(bytes, RollupCoverageLedgerV1::validate)
}
fn encode_current<T>(
value: &T,
validate: impl FnOnce(&T) -> Result<(), LifecycleContractError>,
) -> Result<Vec<u8>, LifecycleContractError>
where
T: Serialize,
{
validate(value)?;
let bytes = serde_json::to_vec(value).map_err(LifecycleContractError::serde)?;
if bytes.len() > MAX_LIFECYCLE_CONTRACT_BYTES {
return Err(LifecycleContractError::Budget {
field: "lifecycle contract bytes",
observed: bytes.len() as u64,
maximum: MAX_LIFECYCLE_CONTRACT_BYTES as u64,
});
}
Ok(bytes)
}
fn decode_current<T>(
bytes: &[u8],
validate: impl FnOnce(&T) -> Result<(), LifecycleContractError>,
) -> Result<T, LifecycleContractError>
where
T: DeserializeOwned + Serialize,
{
if bytes.len() > MAX_LIFECYCLE_CONTRACT_BYTES {
return Err(LifecycleContractError::Budget {
field: "lifecycle contract bytes",
observed: bytes.len() as u64,
maximum: MAX_LIFECYCLE_CONTRACT_BYTES as u64,
});
}
let probe: serde_json::Value = serde_json::from_slice(bytes).map_err(LifecycleContractError::serde)?;
let _version = probe
.get("format_version")
.ok_or(LifecycleContractError::MissingFormatVersion)?
.as_u64()
.and_then(|value| u16::try_from(value).ok())
.ok_or_else(|| LifecycleContractError::Serde("format_version must be a u16".to_owned()))?;
let value: T = serde_json::from_slice(bytes).map_err(LifecycleContractError::serde)?;
validate(&value)?;
if serde_json::to_vec(&value).map_err(LifecycleContractError::serde)? != bytes {
return Err(LifecycleContractError::NonCanonical);
}
Ok(value)
}
fn validate_format(version: u16) -> Result<(), LifecycleContractError> {
if version == LIFECYCLE_COMPLETION_FORMAT_VERSION {
Ok(())
} else {
Err(LifecycleContractError::UnsupportedFormat(version))
}
}
fn validate_external_key_format(version: u16) -> Result<(), LifecycleContractError> {
if version == EXTERNAL_KEY_FORMAT_VERSION {
Ok(())
} else {
Err(LifecycleContractError::UnsupportedFormat(version))
}
}
#[derive(Clone, Debug, Eq, PartialEq)]
pub enum LifecycleContractError {
UnsupportedFormat(u16),
MissingFormatVersion,
NonCanonical,
ZeroIdentity(&'static str),
Budget {
field: &'static str,
observed: u64,
maximum: u64,
},
Violation(&'static str),
DetailRetiredByPolicy,
Nested(String),
Serde(String),
}
impl LifecycleContractError {
fn serde(error: serde_json::Error) -> Self {
Self::Serde(error.to_string())
}
}
impl Display for LifecycleContractError {
fn fmt(&self, formatter: &mut Formatter<'_>) -> fmt::Result {
match self {
Self::UnsupportedFormat(version) => write!(formatter, "unsupported lifecycle format {version}"),
Self::MissingFormatVersion => formatter.write_str("lifecycle format version is missing"),
Self::NonCanonical => formatter.write_str("lifecycle bytes are not canonical current format"),
Self::ZeroIdentity(name) => write!(formatter, "{name} cannot be zero"),
Self::Budget {
field,
observed,
maximum,
} => write!(formatter, "{field} {observed} exceeds {maximum}"),
Self::Violation(message) => formatter.write_str(message),
Self::DetailRetiredByPolicy => {
formatter.write_str("detail_retired_by_policy: raw detail is retired and this query is outside the declared rollup contract")
}
Self::Nested(message) => write!(formatter, "nested lifecycle authority refused input: {message}"),
Self::Serde(message) => write!(formatter, "lifecycle codec refused input: {message}"),
}
}
}
impl Error for LifecycleContractError {}
#[cfg(test)]
mod tests {
use super::*;
use crate::{BackendAuthorityId, CollectionId, IndexId, ObjectIncarnation, ObjectKey};
fn scope() -> Result<CollectionIndexRef, Box<dyn Error>> {
Ok(CollectionIndexRef::new(CollectionId::new(1)?, IndexId::new(1)?))
}
fn physical(key: &str) -> Result<PhysicalObjectId, Box<dyn Error>> {
Ok(PhysicalObjectId {
backend_authority: BackendAuthorityId::new("backend-1")?,
object_incarnation: ObjectIncarnation::new(1)?,
object_key: ObjectKey::new(key)?,
})
}
fn binding() -> Result<IndexExternalKeyBindingV1, Box<dyn Error>> {
let context = ExternalKeyContextV1 {
format_version: EXTERNAL_KEY_FORMAT_VERSION,
collection_index_ref: scope()?,
content_key_id: ContentKeyIdV1::new(1)?,
content_key_generation: 1,
binding_id: ExternalKeyBindingIdV1::new(1)?,
binding_generation: 1,
access_generation: 1,
object_context_digest: [8; 32],
};
Ok(IndexExternalKeyBindingV1 {
format_version: EXTERNAL_KEY_FORMAT_VERSION,
binding_id: context.binding_id,
collection_index_ref: context.collection_index_ref,
content_key_id: context.content_key_id,
content_key_generation: context.content_key_generation,
binding_generation: context.binding_generation,
access_generation: context.access_generation,
opaque_binding: vec![7; 64],
context,
context_digest: context.digest()?,
created_at_revision: ClusterRevision::new(1),
})
}
#[test]
fn encrypted_object_and_binding_are_exact_current_format() -> Result<(), Box<dyn Error>> {
let physical_object = physical("encrypted/object-1")?;
let content_key_id = ContentKeyIdV1::new(1)?;
let object = EncryptedObjectKeyRefV1 {
format_version: EXTERNAL_KEY_FORMAT_VERSION,
nonce: object_nonce_v1(content_key_id, 1, &physical_object)?,
authenticated_context_digest: object_authenticated_context_digest_v1(
DocumentEncryptionAlgorithmV1::Aes256Gcm,
content_key_id,
1,
&physical_object,
128,
)?,
physical_object,
algorithm: DocumentEncryptionAlgorithmV1::Aes256Gcm,
content_key_id,
content_key_generation: 1,
ciphertext_bytes: 128,
};
let bytes = encode_encrypted_object_key_ref_v1(&object)?;
assert_eq!(decode_encrypted_object_key_ref_v1(&bytes)?, object);
let binding = binding()?;
let bytes = encode_index_external_key_binding_v1(&binding)?;
assert_eq!(decode_index_external_key_binding_v1(&bytes)?, binding);
Ok(())
}
#[test]
fn rollup_route_is_exact_and_retired_raw_refuses_mismatch() -> Result<(), Box<dyn Error>> {
let dimension = FieldPathDigestV1::new([1; 32])?;
let policy = RollupPolicyV1 {
format_version: LIFECYCLE_COMPLETION_FORMAT_VERSION,
collection_index_ref: scope()?,
policy_id: RollupPolicyIdV1::new(1)?,
policy_generation: PolicyGeneration::new(1),
detail_retirement_acknowledged: true,
time_zone: RollupTimeZoneV1::Utc,
timestamp_field_digest: FieldPathDigestV1::new([3; 32])?,
bucket_width_millis: 60_000,
bucket_alignment_millis: 0,
dimensions: vec![dimension],
measures: vec![RollupMeasureV1::Count],
};
let policy_bytes = encode_rollup_policy_v1(&policy)?;
assert_eq!(decode_rollup_policy_v1(&policy_bytes)?, policy);
let mut unacknowledged = policy.clone();
unacknowledged.detail_retirement_acknowledged = false;
assert!(matches!(
unacknowledged.validate(),
Err(LifecycleContractError::Violation(_))
));
let root = RootSetRef {
format_version: crate::ROOT_SET_REF_FORMAT_VERSION,
root: physical("rollup/root")?,
content_digest: [9; 32],
item_count: 1,
page_count: 1,
schema_generations: vec![SchemaGeneration::new(1)],
};
let mut coverage = RollupCoverageV1 {
format_version: LIFECYCLE_COMPLETION_FORMAT_VERSION,
collection_index_ref: scope()?,
policy: policy.clone(),
schema_generation: SchemaGeneration::new(1),
source_incarnation: None,
source_cut: None,
source_generations: vec![RollupSourceGenerationV1 {
shard_id: ShardId::new(0),
manifest_generation: ManifestGeneration::new(1),
correction_through: SeqNo::new(10),
}],
range_start_millis: 60_000,
range_end_millis: 180_000,
late_data_watermark_millis: 180_000,
output_root: root,
raw_retirement_state: RawRetirementStateV1::RawPresent,
routing_proof: None,
};
let mut invalid_output_root = coverage.clone();
invalid_output_root.output_root.item_count = 2;
assert!(matches!(
invalid_output_root.validate(),
Err(LifecycleContractError::Violation(_))
));
coverage.raw_retirement_state = RawRetirementStateV1::RawRetired;
coverage.routing_proof = Some(RollupRoutingProofV1 {
query_contract_digest: rollup_query_contract_digest_v1()?,
coverage_digest: coverage.coverage_digest()?,
proven_at_revision: ClusterRevision::new(2),
});
let ledger = RollupCoverageLedgerV1 {
format_version: LIFECYCLE_COMPLETION_FORMAT_VERSION,
collection_index_ref: scope()?,
revision: 2,
coverages: vec![coverage],
};
let bytes = encode_rollup_coverage_ledger_v1(&ledger)?;
assert_eq!(decode_rollup_coverage_ledger_v1(&bytes)?, ledger);
let coverage_digest = ledger.coverages[0].coverage_digest()?;
let query = RollupQueryShapeV1 {
range_start_millis: 0,
range_end_millis: 240_000,
time_zone: RollupTimeZoneV1::Utc,
bucket_width_millis: 60_000,
bucket_alignment_millis: 0,
dimensions: vec![dimension],
measures: vec![RollupMeasureV1::Count],
};
assert_eq!(
route_rollup_query_v1(&ledger, &query)?,
vec![
RollupRouteSliceV1 {
range_start_millis: 0,
range_end_millis: 60_000,
kind: RollupRouteKindV1::Raw,
rollup_coverage_digest: None,
},
RollupRouteSliceV1 {
range_start_millis: 60_000,
range_end_millis: 180_000,
kind: RollupRouteKindV1::Rollup,
rollup_coverage_digest: Some(coverage_digest),
},
RollupRouteSliceV1 {
range_start_millis: 180_000,
range_end_millis: 240_000,
kind: RollupRouteKindV1::Raw,
rollup_coverage_digest: None,
},
]
);
let mismatch = RollupQueryShapeV1 {
dimensions: Vec::new(),
..query
};
assert_eq!(
route_rollup_query_v1(&ledger, &mismatch),
Err(LifecycleContractError::DetailRetiredByPolicy)
);
let mut stale_proof = ledger.clone();
stale_proof.coverages[0].routing_proof = Some(RollupRoutingProofV1 {
query_contract_digest: RollupQueryContractDigestV1::new([5; 32])?,
coverage_digest: stale_proof.coverages[0].coverage_digest()?,
proven_at_revision: ClusterRevision::new(2),
});
assert!(matches!(
stale_proof.validate(),
Err(LifecycleContractError::Violation(_))
));
let mut future_proof = ledger.clone();
future_proof.coverages[0]
.routing_proof
.as_mut()
.ok_or("routing proof missing")?
.proven_at_revision = ClusterRevision::new(3);
assert!(matches!(
future_proof.validate(),
Err(LifecycleContractError::Violation(_))
));
Ok(())
}
#[test]
fn rollup_output_refuses_empty_or_miscounted_groups() -> Result<(), Box<dyn Error>> {
let plan = RollupWorkPlanV1 {
format_version: LIFECYCLE_COMPLETION_FORMAT_VERSION,
policy: RollupPolicyV1 {
format_version: LIFECYCLE_COMPLETION_FORMAT_VERSION,
collection_index_ref: scope()?,
policy_id: RollupPolicyIdV1::new(1)?,
policy_generation: PolicyGeneration::new(1),
detail_retirement_acknowledged: true,
time_zone: RollupTimeZoneV1::Utc,
timestamp_field_digest: FieldPathDigestV1::new([3; 32])?,
bucket_width_millis: 1_000,
bucket_alignment_millis: 0,
dimensions: Vec::new(),
measures: vec![RollupMeasureV1::Count],
},
schema_generation: SchemaGeneration::new(1),
source_incarnation: None,
source_cut: None,
source_generations: vec![RollupSourceGenerationV1 {
shard_id: ShardId::new(0),
manifest_generation: ManifestGeneration::new(1),
correction_through: SeqNo::new(2),
}],
range_start_millis: 0,
range_end_millis: 1_000,
late_data_watermark_millis: 1_000,
};
let valid = RollupOutputV1 {
format_version: LIFECYCLE_COMPLETION_FORMAT_VERSION,
work_plan_digest: plan.digest()?,
rows: vec![RollupOutputRowV1 {
bucket_start_millis: 0,
dimensions: Vec::new(),
measures: vec![RollupAggregateValueV1::Count(2)],
}],
input_documents: 2,
input_bytes: 64,
};
encode_rollup_output_v1(&valid, &plan)?;
let mut empty = valid.clone();
empty.rows.clear();
assert!(encode_rollup_output_v1(&empty, &plan).is_err());
let mut miscounted = valid;
miscounted.rows[0].measures[0] = RollupAggregateValueV1::Count(1);
assert!(encode_rollup_output_v1(&miscounted, &plan).is_err());
Ok(())
}
#[test]
fn adjacent_rollup_roots_remain_distinct_route_slices() -> Result<(), Box<dyn Error>> {
let dimension = FieldPathDigestV1::new([1; 32])?;
let policy = RollupPolicyV1 {
format_version: LIFECYCLE_COMPLETION_FORMAT_VERSION,
collection_index_ref: scope()?,
policy_id: RollupPolicyIdV1::new(1)?,
policy_generation: PolicyGeneration::new(1),
detail_retirement_acknowledged: true,
time_zone: RollupTimeZoneV1::Utc,
timestamp_field_digest: FieldPathDigestV1::new([3; 32])?,
bucket_width_millis: 60_000,
bucket_alignment_millis: 0,
dimensions: vec![dimension],
measures: vec![RollupMeasureV1::Count],
};
let root = |name: &str| -> Result<RootSetRef, Box<dyn Error>> {
Ok(RootSetRef {
format_version: crate::ROOT_SET_REF_FORMAT_VERSION,
root: physical(name)?,
content_digest: if name.ends_with('1') { [8; 32] } else { [9; 32] },
item_count: 1,
page_count: 1,
schema_generations: vec![SchemaGeneration::new(1)],
})
};
let coverage = |start, end, output_root| -> Result<RollupCoverageV1, Box<dyn Error>> {
Ok(RollupCoverageV1 {
format_version: LIFECYCLE_COMPLETION_FORMAT_VERSION,
collection_index_ref: scope()?,
policy: policy.clone(),
schema_generation: SchemaGeneration::new(1),
source_incarnation: None,
source_cut: None,
source_generations: vec![RollupSourceGenerationV1 {
shard_id: ShardId::new(0),
manifest_generation: ManifestGeneration::new(1),
correction_through: SeqNo::new(10),
}],
range_start_millis: start,
range_end_millis: end,
late_data_watermark_millis: end,
output_root,
raw_retirement_state: RawRetirementStateV1::RawPresent,
routing_proof: None,
})
};
let first = coverage(60_000, 120_000, root("rollup/root-1")?)?;
let second = coverage(120_000, 180_000, root("rollup/root-2")?)?;
let first_digest = first.coverage_digest()?;
let second_digest = second.coverage_digest()?;
assert_ne!(first_digest, second_digest);
let ledger = RollupCoverageLedgerV1 {
format_version: LIFECYCLE_COMPLETION_FORMAT_VERSION,
collection_index_ref: scope()?,
revision: 1,
coverages: vec![first, second],
};
let query = RollupQueryShapeV1 {
range_start_millis: 60_000,
range_end_millis: 180_000,
time_zone: RollupTimeZoneV1::Utc,
bucket_width_millis: 60_000,
bucket_alignment_millis: 0,
dimensions: vec![dimension],
measures: vec![RollupMeasureV1::Count],
};
assert_eq!(
route_rollup_query_v1(&ledger, &query)?,
vec![
RollupRouteSliceV1 {
range_start_millis: 60_000,
range_end_millis: 120_000,
kind: RollupRouteKindV1::Rollup,
rollup_coverage_digest: Some(first_digest),
},
RollupRouteSliceV1 {
range_start_millis: 120_000,
range_end_millis: 180_000,
kind: RollupRouteKindV1::Rollup,
rollup_coverage_digest: Some(second_digest),
},
]
);
Ok(())
}
#[test]
fn codec_refuses_old_future_missing_unknown_duplicate_noncanonical_and_oversized() -> Result<(), Box<dyn Error>> {
let bytes = encode_index_external_key_binding_v1(&binding()?)?;
let mut old: serde_json::Value = serde_json::from_slice(&bytes)?;
let old_version = EXTERNAL_KEY_FORMAT_VERSION - 1;
old["format_version"] = serde_json::Value::from(old_version);
assert_eq!(
decode_index_external_key_binding_v1(&serde_json::to_vec(&old)?),
Err(LifecycleContractError::UnsupportedFormat(old_version))
);
let mut future: serde_json::Value = serde_json::from_slice(&bytes)?;
let future_version = EXTERNAL_KEY_FORMAT_VERSION + 1;
future["format_version"] = serde_json::Value::from(future_version);
assert_eq!(
decode_index_external_key_binding_v1(&serde_json::to_vec(&future)?),
Err(LifecycleContractError::UnsupportedFormat(future_version))
);
assert_eq!(
decode_index_external_key_binding_v1(br#"{"binding_id":"1"}"#),
Err(LifecycleContractError::MissingFormatVersion)
);
let mut unknown: serde_json::Value = serde_json::from_slice(&bytes)?;
unknown["unknown_binding"] = serde_json::json!(true);
assert!(matches!(
decode_index_external_key_binding_v1(&serde_json::to_vec(&unknown)?),
Err(LifecycleContractError::Serde(_))
));
let version_prefix = format!("{{\"format_version\":{EXTERNAL_KEY_FORMAT_VERSION},");
let duplicate_prefix = format!(
"{{\"format_version\":{EXTERNAL_KEY_FORMAT_VERSION},\"format_version\":{EXTERNAL_KEY_FORMAT_VERSION},"
);
let duplicate = String::from_utf8(bytes.clone())?.replacen(&version_prefix, &duplicate_prefix, 1);
assert!(matches!(
decode_index_external_key_binding_v1(duplicate.as_bytes()),
Err(LifecycleContractError::Serde(_))
));
let mut whitespace = bytes;
whitespace.push(b'\n');
assert_eq!(
decode_index_external_key_binding_v1(&whitespace),
Err(LifecycleContractError::NonCanonical)
);
assert!(matches!(
decode_index_external_key_binding_v1(&vec![b' '; MAX_LIFECYCLE_CONTRACT_BYTES + 1]),
Err(LifecycleContractError::Budget { .. })
));
let mut mismatched = binding()?;
mismatched.binding_id = ExternalKeyBindingIdV1::new(2)?;
assert!(matches!(
mismatched.validate(),
Err(LifecycleContractError::Violation(_))
));
Ok(())
}
#[test]
fn serialized_binding_contains_only_opaque_non_plaintext_material() -> Result<(), Box<dyn Error>> {
let value = serde_json::to_value(binding()?)?;
assert!(value.get("opaque_binding").is_some());
assert!(value.get("plaintext_key").is_none());
assert!(value.get("content_key_bytes").is_none());
assert_eq!(value.as_object().map(serde_json::Map::len), Some(11));
Ok(())
}
}