diff --git a/crates/convex-engine/RUNTIME.md b/crates/convex-engine/RUNTIME.md index b11e9e8a..e5bbf344 100644 --- a/crates/convex-engine/RUNTIME.md +++ b/crates/convex-engine/RUNTIME.md @@ -113,8 +113,9 @@ A Live runtime resolves its pricing configuration and pinned references, with their issuer, calendar, convention and rate-index records, once per accepted activation, at the first snapshot that needs them, and reuses them at every tick. All are exact pinned versions, so none changes while that selection holds, and each -snapshot still checks them at its own cutoff. Quotes, fixings and stored curve -revisions are read at every snapshot's cutoff. A Live quote does not read the active +snapshot still checks them at its own cutoff. Every tick's snapshot shares one copy of +the references, written as JSON once and checked once apart from the cutoff (D68). +Quotes, fixings and stored curve revisions are read at every snapshot's cutoff. A Live quote does not read the active configuration: the runtime keeps the selection it accepted until the engine's activation refresh or a restart applies another. An activation committed without the engine therefore waits for one of those. Captured and Replay read the active diff --git a/crates/convex-engine/SNAPSHOTS.md b/crates/convex-engine/SNAPSHOTS.md index de3d2f9e..2dd1bff1 100644 --- a/crates/convex-engine/SNAPSHOTS.md +++ b/crates/convex-engine/SNAPSHOTS.md @@ -55,6 +55,11 @@ one nanosecond beyond it is stale. Reading a cache never renews its age. Snapshot identity is `sha256:<64 lowercase hexadecimal digits>` over canonical schema-1 content with domain `convex.snapshot.v1\0`, or complete recipe schema-2 content with domain `convex.snapshot.v2\0`. Original schema-1 bytes remain exact. +The references are one shared, immutable section (D68). Every snapshot of a Live +selection shares the section it pinned: its canonical JSON is written once, its checks +that read no cutoff run once, and each snapshot still checks it at its own cutoff. The +identity streams the same canonical bytes into the hash, so ids and stored snapshots +are unchanged. See [complete recipe replay](../../docs/CURVE-SNAPSHOT-REPLAY.md). Unordered manifests are normalized; duplicates, conflicting resource revisions, unknown fields and out-of-scope inputs fail validation. Deserialization recomputes content identity diff --git a/crates/convex-engine/src/snapshot.rs b/crates/convex-engine/src/snapshot.rs index a2517a3e..596b7bcb 100644 --- a/crates/convex-engine/src/snapshot.rs +++ b/crates/convex-engine/src/snapshot.rs @@ -209,7 +209,7 @@ impl SnapshotAssembler { configuration: convex_ports::storage::BondPricingConfig, ) -> Result { validate_request(request, events)?; - let references = self.references(request, &configuration).await?; + let references = SnapshotReferences::new(self.references(request, &configuration).await?); let capture = capture_ms(request); let mut quotes = IngestedQuotes::new()?; for event in events { @@ -290,7 +290,7 @@ impl SnapshotAssembler { pub(crate) async fn pin(&self, request: &SnapshotAssembly) -> Result { validate_request(request, &[])?; let configuration = self.configuration(request).await?; - let references = self.references(request, &configuration).await?; + let references = SnapshotReferences::new(self.references(request, &configuration).await?); Ok(PinnedInputs { activation: activation_revision(request), configuration_pin: request.configuration.clone(), @@ -326,7 +326,7 @@ impl SnapshotAssembler { &self, request: &SnapshotAssembly, configuration: convex_ports::storage::BondPricingConfig, - references: Vec, + references: SnapshotReferences, quotes: Vec, ) -> Result { let keys = required_fixings(request, &references)?; @@ -406,13 +406,14 @@ impl IngestedQuotes { /// The pricing configuration and pinned references of one selection, resolved once at /// the cutoff of its first snapshot (D56). Every record is an exact pinned version, so /// none changes while the selection holds; each snapshot still checks them at its own -/// cutoff (`MarketSnapshot::new`). +/// cutoff (`MarketSnapshot::new`). Every snapshot shares the one references section, so +/// its checks that read no cutoff and its canonical JSON are done once (D68). pub(crate) struct PinnedInputs { activation: Option, configuration_pin: ConfigurationPin, instruments: Vec, configuration: convex_ports::storage::BondPricingConfig, - references: Vec, + references: SnapshotReferences, } impl PinnedInputs { /// Whether these inputs were resolved for `request`'s activation, configuration and diff --git a/crates/convex-engine/tests/snapshots.rs b/crates/convex-engine/tests/snapshots.rs index fb664fb8..3042e09d 100644 --- a/crates/convex-engine/tests/snapshots.rs +++ b/crates/convex-engine/tests/snapshots.rs @@ -892,7 +892,7 @@ fn a_zero_prices_its_yield_measures_and_keeps_the_router_reason_for_curve_measur }) .collect(); let mut content = fixture::content(); - let reference = &mut content.references[0]; + let reference = &mut content.references.make_mut()[0]; reference.record.bond = Some(bond); reference.issuer = resources[0].clone(); reference.calendar = resources[1].clone(); diff --git a/crates/convex-engine/tests/support/calibration_replay.rs b/crates/convex-engine/tests/support/calibration_replay.rs index 998ed870..5b88cbae 100644 --- a/crates/convex-engine/tests/support/calibration_replay.rs +++ b/crates/convex-engine/tests/support/calibration_replay.rs @@ -64,7 +64,7 @@ pub fn runtime_capture() -> SnapshotCalibration { pub fn content(mut content: SnapshotContent, capture: SnapshotCalibration) -> SnapshotContent { content.schema_version = 3; // This standalone graph consumes no R0 bond reference or market mark. - content.references.clear(); + content.references.make_mut().clear(); content.quotes.clear(); // Explicit fictional capture configuration, valid at each original fixture instant. content.configuration.created_at = 0; diff --git a/crates/convex-ports/Cargo.toml b/crates/convex-ports/Cargo.toml index c2ef224b..5b736d06 100644 --- a/crates/convex-ports/Cargo.toml +++ b/crates/convex-ports/Cargo.toml @@ -17,7 +17,7 @@ convex-core = { workspace = true } # Trait definitions only - NO runtime deps (no reqwest, sqlx, redis, ...) async-trait = "0.1" serde = { workspace = true } -serde_json = { workspace = true, features = ["float_roundtrip"] } +serde_json = { workspace = true, features = ["float_roundtrip", "raw_value"] } sha2 = "0.10.9" thiserror = { workspace = true } rust_decimal = { workspace = true } diff --git a/crates/convex-ports/src/reference_master.rs b/crates/convex-ports/src/reference_master.rs index 113e6452..7a2e144a 100644 --- a/crates/convex-ports/src/reference_master.rs +++ b/crates/convex-ports/src/reference_master.rs @@ -1080,6 +1080,51 @@ pub fn validate_resolved_bond( } } +/// The checks of [`validate_resolved_bond`] that read `lookup`: the record was recorded +/// by its cutoff, its terms are effective at its instant, and so is every resource, +/// each recorded by the cutoff. Every other check reads the reference alone, so a +/// reference that passed [`validate_resolved_bond`] at one lookup passes it at another +/// exactly when it passes this (D68). +pub fn validate_resolved_bond_at( + reference: &ResolvedBondReference, + lookup: ReferenceLookup, +) -> Result<(), ReferenceError> { + let terms = reference + .record + .bond + .as_ref() + .ok_or(ReferenceError::NotFound)? + .contractual_terms + .as_ref() + .ok_or(ReferenceError::InvalidInput("missing terms"))?; + if reference.record.recorded_at > lookup.known_at || !terms.effective.contains(lookup.as_of) { + return Err(ReferenceError::NotFound); + } + let floating = reference + .floating + .iter() + .flat_map(|floating| [&floating.index, &floating.calendar]); + [ + &reference.issuer, + &reference.calendar, + &reference.convention, + ] + .into_iter() + .chain(floating) + .try_for_each(|record| resource_in_scope(record, lookup)) +} + +/// A resource record known by the lookup's cutoff and effective at its instant. +fn resource_in_scope(record: &MasterRecord, lookup: ReferenceLookup) -> Result<(), ReferenceError> { + if record.recorded_at > lookup.known_at || !record.definition.effective().contains(lookup.as_of) + { + return Err(ReferenceError::InvalidInput( + "reference resource outside effective/known-at scope", + )); + } + Ok(()) +} + /// A floating rate note the service values: floating terms without a cap or floor, and /// explicit mechanics pinning its rate index. pub fn is_valued_floating(bond: &BondReferenceData) -> bool { @@ -1114,13 +1159,7 @@ pub fn validate_floating_resources( .ok_or(ReferenceError::InvalidInput("missing floating terms"))?; for record in [&floating.index, &floating.calendar] { record.definition.validate()?; - if record.recorded_at > lookup.known_at - || !record.definition.effective().contains(lookup.as_of) - { - return Err(ReferenceError::InvalidInput( - "reference resource outside effective/known-at scope", - )); - } + resource_in_scope(record, lookup)?; } let (MasterDefinition::RateIndex(index), MasterDefinition::Calendar(calendar)) = (&floating.index.definition, &floating.calendar.definition) @@ -1205,13 +1244,7 @@ pub fn validate_service_resources( } for record in [issuer, calendar, convention] { record.definition.validate()?; - if record.recorded_at > lookup.known_at - || !record.definition.effective().contains(lookup.as_of) - { - return Err(ReferenceError::InvalidInput( - "reference resource outside effective/known-at scope", - )); - } + resource_in_scope(record, lookup)?; } if let MasterDefinition::Calendar(definition) = &calendar.definition { if definition.rule_version.as_str() != SUPPORTED_BOND_CALENDAR_RULES { diff --git a/crates/convex-ports/src/snapshot.rs b/crates/convex-ports/src/snapshot.rs index 11ffd16d..f7e167b3 100644 --- a/crates/convex-ports/src/snapshot.rs +++ b/crates/convex-ports/src/snapshot.rs @@ -7,8 +7,11 @@ use convex_core::contracts::{ SourceId, UtcTimestamp, Version, }; use serde::{Deserialize, Serialize}; +use serde_json::value::RawValue; use sha2::{Digest, Sha256}; use std::collections::BTreeSet; +use std::io::Write as _; +use std::sync::{Arc, OnceLock}; use crate::calibration::SnapshotCalibration; use crate::fixing::{FixingKey, PublishedFixing}; @@ -16,7 +19,8 @@ use crate::quote_event::{ DataState, QuoteEvent, QuoteEventKind, QuoteField, QuoteKey, QuoteQuality, QuoteSnapshot, }; use crate::reference_master::{ - validate_resolved_bond, MasterDefinition, ReferenceLookup, ResolvedBondReference, + validate_resolved_bond, validate_resolved_bond_at, MasterDefinition, ReferenceLookup, + ResolvedBondReference, }; use crate::storage::BondPricingConfig; use crate::TraitError; @@ -290,8 +294,8 @@ pub struct SnapshotContent { /// Omitted on older explicit-pin manifests to preserve their content identities. #[serde(default, skip_serializing_if = "Option::is_none")] pub configuration_capture: Option, - /// Resolved instrument, issuer, calendar and convention revisions. - pub references: Vec, + /// Resolved instrument, issuer, calendar and convention revisions, one shared section. + pub references: SnapshotReferences, /// Selected quote requirements, values and feed continuity. pub quotes: Vec, /// Required known-at fixings, including explicit missing entries. @@ -310,6 +314,174 @@ pub struct SnapshotContent { pub max_source_skew_ms: i64, } +/// The resolved references of a snapshot: one shared, immutable section (D68). A Live +/// selection pins its references once per activation, and every tick's snapshot shares +/// the section instead of copying it. The checks that read the records alone run once +/// per section, and its canonical JSON is written once; the checks that read a cutoff +/// run for every snapshot. [`SnapshotReferences::make_mut`] is the only way to change +/// the records, and it drops both. It serializes as the plain array of records, so +/// stored snapshots are unchanged. +#[derive(Clone)] +pub struct SnapshotReferences(Arc); + +struct ReferenceSection { + records: Vec, + /// Canonical JSON of `records`, written by the first identity that needs it. + json: OnceLock>, + /// Set once `records` pass the checks that read no cutoff. + checked: OnceLock<()>, +} + +impl ReferenceSection { + fn new(records: Vec) -> Self { + Self { + records, + json: OnceLock::new(), + checked: OnceLock::new(), + } + } +} + +impl Clone for ReferenceSection { + /// A copy is made only to change it, so it starts unchecked and unwritten. + fn clone(&self) -> Self { + Self::new(self.records.clone()) + } +} + +impl SnapshotReferences { + /// One section of `records`, sorted by instrument id as every snapshot holds them. + pub fn new(mut records: Vec) -> Self { + records.sort_by(by_instrument); + Self(Arc::new(ReferenceSection::new(records))) + } + /// The records to change, copied first when another holder shares them. The checks + /// and canonical JSON are dropped; the next snapshot sorts, checks and writes again. + pub fn make_mut(&mut self) -> &mut Vec { + let section = Arc::make_mut(&mut self.0); + section.json.take(); + section.checked.take(); + &mut section.records + } + /// Sort by instrument id, copying only when the records are out of order. A shared + /// section is already sorted, so a Live tick never copies it. + fn sort(&mut self) { + if !self + .0 + .records + .is_sorted_by(|a, b| by_instrument(a, b).is_le()) + { + self.make_mut().sort_by(by_instrument); + } + } + /// One instrument's reference; the records are sorted and distinct once checked. + fn find(&self, instrument_id: &str) -> Option<&ResolvedBondReference> { + let records = &self.0.records; + records + .binary_search_by(|r| r.record.instrument_id.as_str().cmp(instrument_id)) + .ok() + .map(|i| &records[i]) + } + /// Check sorted records at `lookup`. The checks that read the records alone run + /// once per section; the cutoff checks run at every call. + fn validate(&self, lookup: ReferenceLookup) -> Result<(), TraitError> { + use crate::reference_repository::validate_reference_id; + let invalid = |_| TraitError::InvalidInput("invalid resolved snapshot reference"); + if self.0.checked.get().is_some() { + return self + .0 + .records + .iter() + .try_for_each(|reference| validate_resolved_bond_at(reference, lookup)) + .map_err(invalid); + } + let mut resources = std::collections::BTreeMap::new(); + for (i, reference) in self.0.records.iter().enumerate() { + validate_reference_id(&reference.record.instrument_id) + .map_err(|_| TraitError::InvalidInput("invalid snapshot instrument"))?; + if i > 0 && self.0.records[i - 1].record.instrument_id == reference.record.instrument_id + { + return Err(TraitError::InvalidInput("duplicate snapshot reference")); + } + validate_resolved_bond(reference, lookup).map_err(invalid)?; + for record in [ + &reference.issuer, + &reference.calendar, + &reference.convention, + ] { + let key = (record.definition.key(), record.version); + if resources + .insert(key, record) + .is_some_and(|prior| prior != record) + { + return Err(TraitError::Conflict( + "different content for one reference revision", + )); + } + } + } + let _ = self.0.checked.set(()); + Ok(()) + } + /// The records' canonical JSON, written once per section. + fn json(&self) -> Result<&RawValue, TraitError> { + if let Some(json) = self.0.json.get() { + return Ok(json); + } + let json = serde_json::value::to_raw_value(&self.0.records) + .map_err(|_| TraitError::InvalidInput("snapshot serialization failed"))?; + Ok(self.0.json.get_or_init(|| json)) + } +} + +fn by_instrument(a: &ResolvedBondReference, b: &ResolvedBondReference) -> std::cmp::Ordering { + a.record + .instrument_id + .as_str() + .cmp(b.record.instrument_id.as_str()) +} + +impl std::ops::Deref for SnapshotReferences { + type Target = [ResolvedBondReference]; + fn deref(&self) -> &Self::Target { + &self.0.records + } +} + +impl<'a> IntoIterator for &'a SnapshotReferences { + type Item = &'a ResolvedBondReference; + type IntoIter = std::slice::Iter<'a, ResolvedBondReference>; + fn into_iter(self) -> Self::IntoIter { + self.0.records.iter() + } +} + +impl PartialEq for SnapshotReferences { + /// Equal records; the cached checks and JSON are not compared. + fn eq(&self, other: &Self) -> bool { + Arc::ptr_eq(&self.0, &other.0) || self.0.records == other.0.records + } +} + +impl std::fmt::Debug for SnapshotReferences { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + self.0.records.fmt(f) + } +} + +impl Serialize for SnapshotReferences { + fn serialize(&self, serializer: S) -> Result { + self.0.records.serialize(serializer) + } +} + +impl<'de> Deserialize<'de> for SnapshotReferences { + /// Records keep their wire order, so verification still refuses unsorted ones. + fn deserialize>(deserializer: D) -> Result { + Vec::deserialize(deserializer).map(|records| Self(Arc::new(ReferenceSection::new(records)))) + } +} + /// Machine-readable quality reason; identity is a selected dependency, never raw licensed data. #[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] #[serde(deny_unknown_fields)] @@ -422,23 +594,7 @@ impl MarketSnapshot { content.as_of.max(content.known_at), &InputSet::all(&content), ); - let bytes = serde_json::to_vec(&content) - .map_err(|_| TraitError::InvalidInput("snapshot serialization failed"))?; - if bytes.len() > MAX_SNAPSHOT_BYTES { - return Err(TraitError::CapacityExceeded); - } - let id = SnapshotId::new(format!( - "sha256:{}", - content_digest( - match content.schema_version { - 1 => b"convex.snapshot.v1\0", - 2 => b"convex.snapshot.v2\0", - _ => b"convex.snapshot.v3\0", - }, - &bytes - ) - )) - .map_err(|_| TraitError::InvalidInput("invalid snapshot identity"))?; + let id = identity(&content)?; Ok(Self(SnapshotWire { snapshot_id: id, content, @@ -460,11 +616,7 @@ impl MarketSnapshot { /// The resolved reference of one instrument. Every snapshot holds its references /// sorted by instrument id without duplicates, so this is a binary search. pub fn reference(&self, instrument_id: &str) -> Option<&ResolvedBondReference> { - let references = &self.0.content.references; - references - .binary_search_by(|r| r.record.instrument_id.as_str().cmp(instrument_id)) - .ok() - .map(|i| &references[i]) + self.0.content.references.find(instrument_id) } /// The selected quotes of one instrument, one per source, in source order. Every /// snapshot holds its quotes sorted by instrument id, then source. @@ -638,6 +790,99 @@ impl MarketSnapshot { Ok(()) } } +/// SHA-256 under the schema's domain over the canonical JSON of normalized content, the +/// bytes `serde_json::to_vec` writes for it. They stream into the hash, bounded as they +/// go, and the references are written from their section's cached JSON (D68). +fn identity(content: &SnapshotContent) -> Result { + let canonical = CanonicalContent::new(content)?; + let mut hash = Sha256::new(); + hash.update(match content.schema_version { + 1 => b"convex.snapshot.v1\0", + 2 => b"convex.snapshot.v2\0", + _ => b"convex.snapshot.v3\0", + }); + let mut writer = std::io::BufWriter::new(BoundedDigest { hash, len: 0 }); + let written = serde_json::to_writer(&mut writer, &canonical) + .map_err(drop) + .and_then(|()| writer.flush().map_err(drop)); + if written.is_err() { + return Err(if writer.get_ref().len > MAX_SNAPSHOT_BYTES { + TraitError::CapacityExceeded + } else { + TraitError::InvalidInput("snapshot serialization failed") + }); + } + let (digest, _) = writer.into_parts(); + SnapshotId::new(format!("sha256:{:x}", digest.hash.finalize())) + .map_err(|_| TraitError::InvalidInput("invalid snapshot identity")) +} + +/// A hash that refuses bytes beyond [`MAX_SNAPSHOT_BYTES`]. +struct BoundedDigest { + hash: Sha256, + len: usize, +} + +impl std::io::Write for BoundedDigest { + fn write(&mut self, bytes: &[u8]) -> std::io::Result { + self.len = self.len.saturating_add(bytes.len()); + if self.len > MAX_SNAPSHOT_BYTES { + return Err(std::io::Error::other("snapshot exceeds its bound")); + } + self.hash.update(bytes); + Ok(bytes.len()) + } + fn flush(&mut self) -> std::io::Result<()> { + Ok(()) + } +} + +/// [`SnapshotContent`] as identity writes it, with the references from their cached +/// JSON. The fields, their order and their skip rules mirror `SnapshotContent`; a test +/// compares the identity with SHA-256 over `serde_json::to_vec` of the content. +#[derive(Serialize)] +struct CanonicalContent<'a> { + schema_version: u32, + as_of: &'a UtcTimestamp, + known_at: &'a UtcTimestamp, + generation: &'a Version, + #[serde(serialize_with = "snapshot_configuration::serialize")] + configuration: &'a BondPricingConfig, + #[serde(skip_serializing_if = "Option::is_none")] + configuration_capture: Option<&'a crate::configuration::ConfigurationCapture>, + references: &'a RawValue, + quotes: &'a [SnapshotQuote], + fixings: &'a [SnapshotFixing], + required_curves: &'a [CurveId], + curves: &'a [SnapshotCurve], + #[serde(skip_serializing_if = "<[_]>::is_empty")] + curve_recipes: &'a [SnapshotCurveRecipe], + #[serde(skip_serializing_if = "Option::is_none")] + calibration: Option<&'a SnapshotCalibration>, + max_source_skew_ms: i64, +} + +impl<'a> CanonicalContent<'a> { + fn new(content: &'a SnapshotContent) -> Result { + Ok(Self { + schema_version: content.schema_version, + as_of: &content.as_of, + known_at: &content.known_at, + generation: &content.generation, + configuration: &content.configuration, + configuration_capture: content.configuration_capture.as_ref(), + references: content.references.json()?, + quotes: &content.quotes, + fixings: &content.fixings, + required_curves: &content.required_curves, + curves: &content.curves, + curve_recipes: &content.curve_recipes, + calibration: content.calibration.as_ref(), + max_source_skew_ms: content.max_source_skew_ms, + }) + } +} + impl TryFrom for MarketSnapshot { type Error = TraitError; fn try_from(value: SnapshotWire) -> Result { @@ -699,12 +944,7 @@ fn normalize(content: &mut SnapshotContent) -> Result<(), TraitError> { { return Err(TraitError::CapacityExceeded); } - content.references.sort_by(|a, b| { - a.record - .instrument_id - .as_str() - .cmp(b.record.instrument_id.as_str()) - }); + content.references.sort(); content.quotes.sort_by(|a, b| { ( a.requirement.key.instrument_id.as_str(), @@ -796,45 +1036,20 @@ fn validate_content(content: &SnapshotContent) -> Result<(), TraitError> { .applies_to .validate() .map_err(TraitError::InvalidInput)?; - let mut instruments = BTreeSet::new(); - let mut resources = std::collections::BTreeMap::new(); - for reference in &content.references { - validate_reference_id(&reference.record.instrument_id) - .map_err(|_| TraitError::InvalidInput("invalid snapshot instrument"))?; - if !instruments.insert(reference.record.instrument_id.as_str()) { - return Err(TraitError::InvalidInput("duplicate snapshot reference")); - } - validate_resolved_bond( - reference, - ReferenceLookup { - as_of: content.as_of, - known_at: content.known_at, - }, - ) - .map_err(|_| TraitError::InvalidInput("invalid resolved snapshot reference"))?; - for record in [ - &reference.issuer, - &reference.calendar, - &reference.convention, - ] { - let key = (record.definition.key(), record.version); - if resources - .insert(key, record) - .is_some_and(|prior| prior != record) - { - return Err(TraitError::Conflict( - "different content for one reference revision", - )); - } - } - } + content.references.validate(ReferenceLookup { + as_of: content.as_of, + known_at: content.known_at, + })?; let mut quotes = BTreeSet::new(); for quote in &content.quotes { let requirement = "e.requirement; validate_reference_id(&requirement.key.instrument_id) .map_err(|_| TraitError::InvalidInput("invalid quote instrument"))?; crate::quote_event::validate_text(&requirement.key.source, 128)?; - if !instruments.contains(requirement.key.instrument_id.as_str()) + if content + .references + .find(requirement.key.instrument_id.as_str()) + .is_none() || !quotes.insert(( requirement.key.instrument_id.as_str(), &requirement.key.source, diff --git a/crates/convex-ports/tests/calibration_snapshots.rs b/crates/convex-ports/tests/calibration_snapshots.rs index 5661cbb9..1ac3fbff 100644 --- a/crates/convex-ports/tests/calibration_snapshots.rs +++ b/crates/convex-ports/tests/calibration_snapshots.rs @@ -31,6 +31,12 @@ fn content(capture: SnapshotCalibration) -> SnapshotContent { content.calibration = Some(capture); content } +/// D68: a schema-3 identity is SHA-256 over the content's plain serialization. +#[test] +fn calibration_identity_is_the_hash_of_the_plain_content_bytes() { + fixture::assert_plain_identity(&MarketSnapshot::new(content(record())).unwrap()); +} + #[test] fn original_bounds_and_identity_pins_apply_to_the_whole_calibration() { let original = record(); diff --git a/crates/convex-ports/tests/curve_recipe_snapshots.rs b/crates/convex-ports/tests/curve_recipe_snapshots.rs index 896c9de7..375d3d89 100644 --- a/crates/convex-ports/tests/curve_recipe_snapshots.rs +++ b/crates/convex-ports/tests/curve_recipe_snapshots.rs @@ -27,6 +27,12 @@ fn content(records: Vec) -> SnapshotContent { c } +/// D68: a schema-2 identity is SHA-256 over the content's plain serialization. +#[test] +fn recipe_identity_is_the_hash_of_the_plain_content_bytes() { + fixture::assert_plain_identity(&MarketSnapshot::new(content(vec![record()])).unwrap()); +} + #[test] fn record_and_aggregate_bounds_are_shared_with_original_zero_records() { let mut r = record(); diff --git a/crates/convex-ports/tests/snapshot_contracts.rs b/crates/convex-ports/tests/snapshot_contracts.rs index 37741ee9..dc09b7e1 100644 --- a/crates/convex-ports/tests/snapshot_contracts.rs +++ b/crates/convex-ports/tests/snapshot_contracts.rs @@ -131,7 +131,10 @@ fn unknown_later_known_resources_and_invalid_curves_are_rejected_before_publicat for case in 0..6 { let mut content = fixture::content(); match case { - 0 => content.references[0].calendar.recorded_at = fixture::time("2026-10-03T00:00:00Z"), + 0 => { + content.references.make_mut()[0].calendar.recorded_at = + fixture::time("2026-10-03T00:00:00Z") + } 1 => content.configuration.version = 0, 2 => content.curves[0].points.swap(0, 1), 3 => content.curves[0].points[1].tenor_years = content.curves[0].points[0].tenor_years, @@ -145,6 +148,64 @@ fn unknown_later_known_resources_and_invalid_curves_are_rejected_before_publicat } } +/// D68: the identity equals SHA-256 over the content's plain serialization, one quote +/// field changes it, and a change to a section whose JSON was written drops that JSON. +#[test] +fn identity_is_the_hash_of_the_plain_content_bytes() { + let mut content = fixture::content(); + fixture::add_bond(&mut content, "BOND-002", 0); + let first = MarketSnapshot::new(content.clone()).unwrap(); + fixture::assert_plain_identity(&first); + let id = first.id().clone(); + + let mut quote = content.clone(); + for field in quote.quotes[0].image.as_mut().unwrap().fields.values_mut() { + field.source_time_ms -= 1; + } + let quote = MarketSnapshot::new(quote).unwrap(); + fixture::assert_plain_identity("e); + assert_ne!(quote.id(), &id); + assert_eq!(quote.content().references, first.content().references); + + // The section is no longer shared and its JSON is written: a change made in place. + drop((first, quote)); + fixture::add_bond(&mut content, "BOND-003", 0); + let grown = MarketSnapshot::new(content).unwrap(); + fixture::assert_plain_identity(&grown); + assert_ne!(grown.id(), &id); +} + +/// D68: a shared section's checks that read no cutoff run once, each snapshot still +/// checks it at its own cutoff, and a change drops the checks. +#[test] +fn a_checked_reference_section_is_still_checked_at_each_cutoff_and_after_a_change() { + let mut content = fixture::content(); + content.references.make_mut()[0].calendar.recorded_at = fixture::time("2026-10-02T12:00:00Z"); + let checked = MarketSnapshot::new(content.clone()).unwrap(); + + // Before the calendar revision was recorded. + let mut earlier = content.clone(); + earlier.as_of = fixture::time("2026-10-02T11:00:00Z"); + earlier.known_at = earlier.as_of; + assert!(matches!( + MarketSnapshot::new(earlier), + Err(convex_ports::TraitError::InvalidInput( + "invalid resolved snapshot reference" + )) + )); + + // A duplicate added in place, once the section is no longer shared. + drop(checked); + let duplicate = content.references[0].clone(); + content.references.make_mut().push(duplicate); + assert!(matches!( + MarketSnapshot::new(content), + Err(convex_ports::TraitError::InvalidInput( + "duplicate snapshot reference" + )) + )); +} + #[test] fn snapshot_configuration_keeps_exact_decimals_under_workspace_feature_unification() { let mut content = fixture::content(); @@ -170,7 +231,7 @@ fn instrument_lookups_find_each_reference_and_every_source_in_canonical_order() let mut content = fixture::content(); let reference = content.references[0].clone(); let quote = content.quotes[0].clone(); - content.references.clear(); + content.references.make_mut().clear(); content.quotes.clear(); // Supplied out of order; one instrument has two sources. for (id, sources) in [ @@ -183,7 +244,7 @@ fn instrument_lookups_find_each_reference_and_every_source_in_canonical_order() if let Some(bond) = &mut reference.record.bond { bond.instrument_id = reference.record.instrument_id.clone(); } - content.references.push(reference); + content.references.make_mut().push(reference); for source in sources { let mut quote = quote.clone(); quote.requirement.key.instrument_id = convex_core::ids::InstrumentId::new(id); @@ -353,7 +414,7 @@ fn an_instrument_reads_the_same_inputs_until_its_own_change() { assert_eq!(same(curve.clone(), &[]), (true, true)); assert_eq!(same(curve, &["USD-FIXTURE"]), (true, false)); let mut reference = base(); - reference.references[0].record.version = fixture::version(2); + reference.references.make_mut()[0].record.version = fixture::version(2); assert_eq!(same(reference, &[]), (true, false)); // The frame: the next day, the skew policy and the configuration. diff --git a/crates/convex-ports/tests/support/snapshot_fixtures.rs b/crates/convex-ports/tests/support/snapshot_fixtures.rs index be2662c3..da0c95e1 100644 --- a/crates/convex-ports/tests/support/snapshot_fixtures.rs +++ b/crates/convex-ports/tests/support/snapshot_fixtures.rs @@ -139,7 +139,7 @@ pub fn content() -> SnapshotContent { generation: version(1), configuration: config(), configuration_capture: None, - references: vec![ResolvedBondReference { + references: SnapshotReferences::new(vec![ResolvedBondReference { record: ReferenceRecord { schema_version: SchemaVersion::V2, instrument_id: bond.instrument_id.clone(), @@ -151,7 +151,7 @@ pub fn content() -> SnapshotContent { calendar: resources[1].clone(), convention: resources[2].clone(), floating: None, - }], + }]), quotes: vec![SnapshotQuote { requirement: quote_requirement(), delivery_watermark: Some(event.clone()), @@ -172,6 +172,21 @@ pub fn content() -> SnapshotContent { pub fn snapshot() -> MarketSnapshot { MarketSnapshot::new(content()).unwrap() } +/// The identity is SHA-256 over the content's plain serialization (D68): the hash streams +/// the references from their section's cached JSON, and this writes them record by record. +pub fn assert_plain_identity(snapshot: &MarketSnapshot) { + let content = snapshot.content(); + let domain: &[u8] = match content.schema_version { + 1 => b"convex.snapshot.v1\0", + 2 => b"convex.snapshot.v2\0", + _ => b"convex.snapshot.v3\0", + }; + let bytes = serde_json::to_vec(content).unwrap(); + assert_eq!( + snapshot.id().as_str(), + format!("sha256:{}", content_digest(domain, &bytes)) + ); +} /// Add a copy of the fixture bond under `id`, its mid observed `age_ms` before `as_of`. pub fn add_bond(content: &mut SnapshotContent, id: &str, age_ms: i64) { let mut reference = content.references[0].clone(); @@ -200,6 +215,6 @@ pub fn add_bond(content: &mut SnapshotContent, id: &str, age_ms: i64) { field.source_time_ms = observed; field.receive_time_ms = observed; } - content.references.push(reference); + content.references.make_mut().push(reference); content.quotes.push(quote); }