commit 41acfb8af066a1f08e870eb43a09ffed8c428b22
parent b82931ce699202529c35e3a9d21a2be80a533695
Author: triesap <tyson@radroots.org>
Date: Tue, 28 Jul 2026 11:23:49 +0000
event_store: make transition invariants structural
- bind page arithmetic to governed event wire bounds
- remove duplicate checks after typed and scoped validation
- preserve typed cause keys and validate stored timestamps
- apply production AST authority to immutable predecessors
Diffstat:
7 files changed, 73 insertions(+), 123 deletions(-)
diff --git a/contracts/event_store_production_sources.toml b/contracts/event_store_production_sources.toml
@@ -19,7 +19,7 @@ sha256 = "14a98fe4361baa90e74c499ca96cd47822531041dfd874281b1758796b8fd22e"
[[sources]]
path = "crates/event_store/src/model/addressable_transition_feed_v1.rs"
-sha256 = "da82c889de4c131b33f40c88aced0a381b7f213b297d125327334714ab4c2511"
+sha256 = "bf0ed43e03af0d45f9406f2ca2463b7ab740b0582191c9ba44a36cdc77de3f4e"
[[sources]]
path = "crates/event_store/src/model/current_visibility_v1.rs"
@@ -71,11 +71,11 @@ sha256 = "ef277328d99ef75a978f2907dc654cbf391436b9db693c3be69655dccbe74f89"
[[sources]]
path = "crates/event_store/src/store/addressable_transition_feed_v1.rs"
-sha256 = "4d1d9b09bacbfc1bcbedcfc641a50679b6b4b51e39786ab9e321a6d78c04f744"
+sha256 = "510f9cdeb3756b0f1ae4fcd5c85cb68ff5cb91344bc15cab0aa9c473938d2743"
[[sources]]
path = "crates/event_store/src/store/current_visibility_v1.rs"
-sha256 = "56a21d2ec6a408a2dd5131db0e3c7343c67e60efc67a9c8b2f97fbb90d2d5788"
+sha256 = "ba4f77e758d98b25f034cbcc554dda833c839f3ccb8671c0b43997aaac764806"
[[sources]]
path = "crates/event_store/src/store/food_availability_projection_v1.rs"
diff --git a/crates/event_store/src/model/addressable_transition_feed_v1.rs b/crates/event_store/src/model/addressable_transition_feed_v1.rs
@@ -17,6 +17,14 @@ pub const RADROOTS_ADDRESSABLE_TRANSITION_D_TAG_MAX_BYTES_V1: usize =
radroots_event::wire::v1::DEFAULT_TAG_ELEMENT_MAX_BYTES;
pub const RADROOTS_ADDRESSABLE_TRANSITION_CURSOR_JSON_MAX_BYTES_V1: usize = 512;
const SCOPE_FINGERPRINT_DOMAIN_V1: &[u8] = b"radroots.addressable-transition-scope.v1\0";
+const _: () = assert!(
+ radroots_event::wire::v1::DEFAULT_RAW_JSON_MAX_BYTES
+ <= RADROOTS_ADDRESSABLE_TRANSITION_PAGE_RAW_JSON_MAX_BYTES_V1
+);
+const _: () = assert!(
+ RADROOTS_ADDRESSABLE_TRANSITION_PAGE_LIMIT_MAX_V1 as usize
+ <= usize::MAX / radroots_event::wire::v1::DEFAULT_RAW_JSON_MAX_BYTES
+);
#[derive(Clone, Copy, Debug, PartialEq, Eq, Hash)]
pub struct RadrootsAddressableTransitionScopeFingerprintV1([u8; 32]);
@@ -61,14 +69,6 @@ impl RadrootsAddressableTransitionScopeV1 {
if kinds.is_empty() {
return Err(RadrootsEventStoreError::AddressableTransitionScopeEmpty);
}
- if kinds.len() > RADROOTS_ADDRESSABLE_TRANSITION_SCOPE_KIND_MAX_V1 {
- return Err(
- RadrootsEventStoreError::AddressableTransitionScopeTooLarge {
- max: RADROOTS_ADDRESSABLE_TRANSITION_SCOPE_KIND_MAX_V1,
- actual: kinds.len(),
- },
- );
- }
if let Some(kind) = kinds
.iter()
.copied()
@@ -209,11 +209,10 @@ fn decode_cursor_hex(
{
return Err(RadrootsEventStoreError::AddressableTransitionCursorEncoding { field });
}
- let decoded = hex::decode(value)
- .map_err(|_| RadrootsEventStoreError::AddressableTransitionCursorEncoding { field })?;
- decoded
- .try_into()
- .map_err(|_| RadrootsEventStoreError::AddressableTransitionCursorEncoding { field })
+ let mut decoded = [0_u8; 32];
+ hex::decode_to_slice(value, &mut decoded)
+ .expect("the exact lowercase hexadecimal cursor encoding was validated");
+ Ok(decoded)
}
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
diff --git a/crates/event_store/src/store.rs b/crates/event_store/src/store.rs
@@ -8803,6 +8803,12 @@ CREATE TABLE aux.event_transport_observation (event_id TEXT);",
"stored address deletion cutoff is invalid",
),
(
+ "negative stored created-at",
+ true,
+ "UPDATE radroots_event_store_addressable_head_state SET raw_head_created_at = -1",
+ "stored raw-head created-at is invalid",
+ ),
+ (
"contract disagreement",
false,
"UPDATE radroots_event_store_addressable_head_state SET contract_id = 'radroots.event.invalid.v1'",
@@ -9289,6 +9295,14 @@ CREATE TABLE aux.event_transport_observation (event_id TEXT);",
"UPDATE radroots_event_store_addressable_head_transition SET d_tag = replace(hex(zeroblob(513)), '00', 'x')",
),
(
+ "negative kind",
+ "UPDATE radroots_event_store_addressable_head_transition SET kind = -1",
+ ),
+ (
+ "kind above u32",
+ "UPDATE radroots_event_store_addressable_head_transition SET kind = 4294967296",
+ ),
+ (
"raw-head sequence",
"UPDATE radroots_event_store_addressable_head_transition SET raw_head_event_seq = 0",
),
diff --git a/crates/event_store/src/store/addressable_transition_feed_v1.rs b/crates/event_store/src/store/addressable_transition_feed_v1.rs
@@ -70,11 +70,8 @@ pub(super) async fn addressable_transition_page_in_transaction_v1(
let mut transitions = Vec::with_capacity(rows.len());
let mut canonical_payload_bytes = 0usize;
let mut last_scanned_sequence = start;
- let mut stopped_before_row = false;
for row in rows {
- let transition_seq: i64 = row.try_get("transition_seq").map_err(|error| {
- corruption(format!("transition sequence cannot be decoded: {error}"))
- })?;
+ let transition_seq: i64 = row.try_get("transition_seq")?;
let expected_sequence = last_scanned_sequence
.checked_add(1)
.ok_or_else(|| corruption("transition sequence overflow"))?;
@@ -85,57 +82,27 @@ pub(super) async fn addressable_transition_page_in_transaction_v1(
),
});
}
- let kind = u32_from_i64(
- "transition.kind",
- row.try_get("kind").map_err(|error| {
- corruption(format!("transition kind cannot be decoded: {error}"))
- })?,
- )
- .map_err(|error| corruption(error.to_string()))?;
+ let kind = u32_from_i64("transition.kind", row.try_get("kind")?)
+ .map_err(|error| corruption(error.to_string()))?;
if !scope.kinds().contains(&kind) {
last_scanned_sequence = transition_seq;
continue;
}
if transitions.len() == usize::try_from(limit).expect("u32 fits usize") {
- stopped_before_row = true;
break;
}
let transition = transition_from_row(connection, row, source.generation).await?;
let transition_payload_bytes = transition
.visible_event()
.map_or(0, |event| event.raw_json().len());
- let next_payload_bytes = canonical_payload_bytes
- .checked_add(transition_payload_bytes)
- .ok_or(
- RadrootsEventStoreError::AddressableTransitionPagePayloadTooLarge {
- max: RADROOTS_ADDRESSABLE_TRANSITION_PAGE_RAW_JSON_MAX_BYTES_V1,
- actual: usize::MAX,
- },
- )?;
+ let next_payload_bytes = canonical_payload_bytes + transition_payload_bytes;
if next_payload_bytes > RADROOTS_ADDRESSABLE_TRANSITION_PAGE_RAW_JSON_MAX_BYTES_V1 {
- if transitions.is_empty() {
- return Err(
- RadrootsEventStoreError::AddressableTransitionPagePayloadTooLarge {
- max: RADROOTS_ADDRESSABLE_TRANSITION_PAGE_RAW_JSON_MAX_BYTES_V1,
- actual: next_payload_bytes,
- },
- );
- }
- stopped_before_row = true;
break;
}
canonical_payload_bytes = next_payload_bytes;
last_scanned_sequence = transition_seq;
transitions.push(transition);
}
- if !scan_limited && !stopped_before_row && last_scanned_sequence < source.high_water {
- return Err(RadrootsEventStoreError::AddressableTransitionSequenceGap {
- reason: format!(
- "sealed interval ends at {}, but the last stored transition is {last_scanned_sequence}",
- source.high_water
- ),
- });
- }
let has_more = last_scanned_sequence < source.high_water;
Ok(RadrootsAddressableTransitionPageV1 {
source_high_water: source.high_water,
@@ -144,7 +111,8 @@ pub(super) async fn addressable_transition_page_in_transaction_v1(
source.generation,
scope.fingerprint(),
last_scanned_sequence,
- )?,
+ )
+ .expect("the validated source interval and cursor preserve a nonnegative sequence"),
has_more,
})
}
@@ -255,14 +223,6 @@ async fn validate_or_create_cursor(
let Some(cursor) = cursor else {
return Ok(source.floor);
};
- if cursor.feed_version() != RADROOTS_ADDRESSABLE_TRANSITION_FEED_VERSION_V1 {
- return Err(
- RadrootsEventStoreError::AddressableTransitionFeedVersionMismatch {
- expected: RADROOTS_ADDRESSABLE_TRANSITION_FEED_VERSION_V1,
- actual: cursor.feed_version(),
- },
- );
- }
if cursor.scope_fingerprint() != scope.fingerprint() {
return Err(RadrootsEventStoreError::AddressableTransitionScopeMismatch);
}
@@ -308,21 +268,9 @@ async fn validate_or_create_cursor(
async fn transition_from_row(
connection: &mut SqliteConnection,
row: sqlx::sqlite::SqliteRow,
- expected_generation: RadrootsEventStoreSourceGeneration,
+ source_generation: RadrootsEventStoreSourceGeneration,
) -> Result<RadrootsAddressableTransitionV1, RadrootsEventStoreError> {
let transition_seq: i64 = row.try_get("transition_seq")?;
- if transition_seq <= 0 {
- return Err(corruption(format!(
- "transition sequence {transition_seq} is not positive"
- )));
- }
- let source_generation = generation_from_blob(row.try_get("source_generation")?)
- .map_err(|error| corruption(format!("transition generation is invalid: {error}")))?;
- if source_generation != expected_generation {
- return Err(corruption(
- "scoped query returned a transition from another generation",
- ));
- }
let origin =
RadrootsAddressableTransitionOriginV1::parse(row.try_get::<String, _>("origin")?.as_str())
.map_err(|error| corruption(error.to_string()))?;
@@ -396,7 +344,8 @@ async fn transition_from_row(
raw_head_created_at,
)?;
- let (raw_event, admission) = load_and_validate_stored_event(connection, &raw_head).await?;
+ let (raw_event, admission, raw_event_pubkey) =
+ load_and_validate_stored_event(connection, &raw_head).await?;
validate_addressable_reference(
connection,
source_generation,
@@ -415,12 +364,7 @@ async fn transition_from_row(
)));
}
- let visible_event = if let Some(reference) = visible_reference.as_ref() {
- if reference != &raw_head {
- return Err(corruption(format!(
- "transition {transition_seq} visible event is not the raw head"
- )));
- }
+ let visible_event = if visible_reference.is_some() {
Some(RadrootsStoreProducedCanonicalEventV1 {
event_id: raw_head.event_id().clone(),
pubkey: coordinate.pubkey().clone(),
@@ -433,7 +377,7 @@ async fn transition_from_row(
};
if let Some(reference) = retracted_event.as_ref() {
- let (event, admission) = load_and_validate_stored_event(connection, reference).await?;
+ let (event, admission, _) = load_and_validate_stored_event(connection, reference).await?;
validate_addressable_reference(
connection,
source_generation,
@@ -450,7 +394,11 @@ async fn transition_from_row(
}
let cause = if let Some(reference) = cause_reference.as_ref() {
if reference == &raw_head {
- Some((raw_event.clone(), admission.clone()))
+ Some((
+ raw_event.clone(),
+ admission.clone(),
+ raw_event_pubkey.clone(),
+ ))
} else {
Some(load_and_validate_stored_event(connection, reference).await?)
}
@@ -488,27 +436,20 @@ async fn transition_from_row(
retracted_event.as_ref(),
)
.await?;
- let cause_event = cause
- .map(|(event, admission)| {
- let event_reference = cause_reference
- .clone()
- .ok_or_else(|| corruption("loaded transition cause has no reference"))?;
- let pubkey = RadrootsPublicKey::parse(event.pubkey.as_str()).map_err(|error| {
- corruption(format!("transition cause pubkey is invalid: {error}"))
- })?;
- Ok::<RadrootsAddressableTransitionCauseV1, RadrootsEventStoreError>(
+ let cause_event =
+ cause
+ .zip(cause_reference)
+ .map(|((event, admission, pubkey), event_reference)| {
RadrootsAddressableTransitionCauseV1 {
- event: event_reference,
+ event: event_reference.clone(),
pubkey,
created_at: event.created_at,
kind: event.kind,
admission_status: admission.status,
admission_code: admission.code,
contract_id: admission.contract.map(|contract| contract.id.to_owned()),
- },
- )
- })
- .transpose()?;
+ }
+ });
Ok(RadrootsAddressableTransitionV1 {
transition_seq,
@@ -548,7 +489,7 @@ async fn validate_incremental_cause(
coordinate: &RadrootsAddressableTransitionCoordinateV1,
raw_head: &RadrootsAddressableTransitionEventReferenceV1,
cause_reference: Option<&RadrootsAddressableTransitionEventReferenceV1>,
- cause: Option<&(RadrootsStoredRawEvent, EventAdmission)>,
+ cause: Option<&(RadrootsStoredRawEvent, EventAdmission, RadrootsPublicKey)>,
suppression: Option<&RadrootsNip09SuppressionEvidenceV1>,
decision: RadrootsAddressableTransitionRawHeadDecisionV1,
) -> Result<(), RadrootsEventStoreError> {
@@ -557,7 +498,7 @@ async fn validate_incremental_cause(
}
let cause_reference = cause_reference
.ok_or_else(|| corruption("incremental transition has no cause reference"))?;
- let (cause_event, cause_admission) =
+ let (cause_event, cause_admission, _) =
cause.ok_or_else(|| corruption("incremental transition cause could not be loaded"))?;
match decision {
RadrootsAddressableTransitionRawHeadDecisionV1::Applied => {
@@ -676,9 +617,6 @@ async fn validate_retraction_lineage(
|| prior_state.raw_head != current.raw_head))
.then_some(prior_state.raw_head)
} else {
- if origin == RadrootsAddressableTransitionOriginV1::Baseline && retracted.is_some() {
- return Err(corruption("baseline transition retracts prior state"));
- }
None
};
if expected.as_ref() != retracted {
@@ -857,7 +795,7 @@ fn parse_event_id(
async fn load_and_validate_stored_event(
connection: &mut SqliteConnection,
reference: &RadrootsAddressableTransitionEventReferenceV1,
-) -> Result<(RadrootsStoredRawEvent, EventAdmission), RadrootsEventStoreError> {
+) -> Result<(RadrootsStoredRawEvent, EventAdmission, RadrootsPublicKey), RadrootsEventStoreError> {
let row = sqlx::query(
"SELECT seq, event_id, pubkey, created_at, kind, tags_json, content, sig, raw_json, verification_status, contract_status, contract_id, event_class, projection_eligible, inserted_at_ms, updated_at_ms FROM event_envelopes WHERE seq = ? AND event_id = ?",
)
@@ -877,11 +815,9 @@ async fn load_and_validate_stored_event(
let reconstructed = RadrootsEventIngest::from_raw_json(stored.raw_json.clone(), 0)
.map_err(|error| corruption(format!("stored raw event cannot be reverified: {error}")))?;
let event = reconstructed.event();
- let tags_json = serde_json::to_string(&event.tags_as_vec()).map_err(|error| {
- corruption(format!(
- "stored raw event tags cannot be canonicalized: {error}"
- ))
- })?;
+ let event_pubkey = event.author().clone();
+ let tags_json = serde_json::to_string(&event.tags_as_vec())
+ .expect("an in-memory vector of string tags always serializes as JSON");
if stored.event_id != event.id_str()
|| stored.pubkey != event.author_str()
|| stored.created_at != event.created_at_u64()
@@ -909,7 +845,7 @@ async fn load_and_validate_stored_event(
reference.event_id()
)));
}
- Ok((stored, admission))
+ Ok((stored, admission, event_pubkey))
}
async fn validate_addressable_reference(
diff --git a/crates/event_store/src/store/current_visibility_v1.rs b/crates/event_store/src/store/current_visibility_v1.rs
@@ -1,5 +1,5 @@
use super::protocol_storage_v1::stored_raw_event_from_row;
-use super::{RadrootsEventStore, bool_from_i64};
+use super::{RadrootsEventStore, bool_from_i64, u64_from_i64};
use crate::RadrootsEventStoreError;
use crate::model::{
RadrootsCurrentEventVisibilityV1, RadrootsCurrentVisibilityDecisionV1,
@@ -274,6 +274,11 @@ async fn validate_addressable_head_projection(
));
};
let evidence = visibility.suppression.as_ref();
+ let stored_created_at = u64_from_i64(
+ "addressable_head_state.raw_head_created_at",
+ row.try_get("raw_head_created_at")?,
+ )
+ .map_err(|error| visibility_authority_error("stored raw-head created-at", error))?;
let stored_cutoff = row
.try_get::<Option<i64>, _>("address_reference_cutoff")?
.map(|value| {
@@ -286,15 +291,7 @@ async fn validate_addressable_head_projection(
.map_err(|error| visibility_authority_error("stored address deletion cutoff", error))?;
if row.try_get::<String, _>("raw_head_event_id")? != visibility.event.event_id
|| row.try_get::<i64, _>("raw_head_event_seq")? != visibility.event.seq
- || row.try_get::<i64, _>("raw_head_created_at")?
- != i64::try_from(visibility.event.created_at).map_err(|_| {
- RadrootsEventStoreError::CurrentVisibilityDrift {
- reason: format!(
- "addressable event `{}` timestamp is outside SQLite range",
- visibility.event.event_id
- ),
- }
- })?
+ || stored_created_at != visibility.event.created_at
|| row.try_get::<String, _>("admission_status")?
!= visibility.event.admission_status.as_str()
|| row.try_get::<Option<String>, _>("admission_code")?
diff --git a/tools/xtask/src/contract/phase1_publication_artifact.rs b/tools/xtask/src/contract/phase1_publication_artifact.rs
@@ -1,6 +1,9 @@
use super::artifact_bundle::{
GeneratedArtifact, read_regular_file, with_artifact_bundle_transaction,
};
+use super::raw_source_rebuild::{
+ is_semantic_event_store_production_source, validate_event_store_production_source_authority,
+};
use radroots_event_codec::wire::publication::{
RADROOTS_PHASE1_PUBLICATION_ARTIFACT_MAX_BYTES,
RADROOTS_PHASE1_PUBLICATION_ARTIFACT_SCHEMA_VERSION,
@@ -1378,6 +1381,7 @@ fn validate_result_vector(workspace_root: &Path) -> Result<ValidatedResultVector
}
fn validate_immutable_raw_predecessor_under_lock(workspace_root: &Path) -> Result<(), String> {
+ validate_event_store_production_source_authority(workspace_root)?;
for spec in RAW_IMMUTABLE_ARTIFACTS {
let bytes = read_regular_file(workspace_root, spec.relative)?;
if bytes.len() != spec.byte_length || sha256_hex(&bytes) != spec.sha256 {
@@ -1434,7 +1438,7 @@ fn validate_immutable_raw_predecessor_under_lock(workspace_root: &Path) -> Resul
if !seen.insert(path) {
return Err(format!("{RAW_MANIFEST_RELATIVE} duplicates source {path}"));
}
- if !superseded.contains(path) {
+ if !superseded.contains(path) && !is_semantic_event_store_production_source(path) {
validate_value_descriptor(workspace_root, source, "raw predecessor source")?;
}
}
diff --git a/tools/xtask/src/contract/raw_source_rebuild.rs b/tools/xtask/src/contract/raw_source_rebuild.rs
@@ -1217,7 +1217,7 @@ fn validate_event_store_production_source_inventory(
let source = std::str::from_utf8(&source_bytes).map_err(|error| {
format!("{EVENT_STORE_PRODUCTION_SOURCES_RELATIVE} must be UTF-8 TOML: {error}")
})?;
- let inventory: EventStoreProductionSourceInventory = toml::from_str(&source)
+ let inventory: EventStoreProductionSourceInventory = toml::from_str(source)
.map_err(|error| format!("parse {EVENT_STORE_PRODUCTION_SOURCES_RELATIVE}: {error}"))?;
if inventory.schema_version != 1 || inventory.hash_algorithm != PRODUCTION_AST_HASH_ALGORITHM {
return Err(format!(
@@ -1266,7 +1266,7 @@ fn validate_event_store_production_source_inventory(
Ok(actual_paths)
}
-fn is_semantic_event_store_production_source(path: &str) -> bool {
+pub(super) fn is_semantic_event_store_production_source(path: &str) -> bool {
path.starts_with("crates/event_store/src/")
&& path.ends_with(".rs")
&& path != "crates/event_store/src/generated.rs"
