From 74eca2c32f73666134893699bb1d75aedce85afd Mon Sep 17 00:00:00 2001 From: Benedict Lau Date: Tue, 27 Oct 2026 18:03:03 -0400 Subject: [PATCH] Add prototype for nested-computations --- CHANGELOG.md | 17 + Cargo.lock | 8 + integrity-jsonld/src/lib.rs | 14 + integrity-jsonld/src/loader.rs | 4 +- ...cmx5n4ydefbbc62ngyannzi566zv5fsozgbwwfw27u | 28 ++ integrity-lineage-models/Cargo.toml | 1 + .../statements/computation_statement.rs | 337 +++++++++++++++++- .../src/models/statements/mod.rs | 4 +- 8 files changed, 407 insertions(+), 6 deletions(-) create mode 100644 integrity-jsonld/static_contexts/cid/bafkr4iduavib2lgjcmx5n4ydefbbc62ngyannzi566zv5fsozgbwwfw27u diff --git a/CHANGELOG.md b/CHANGELOG.md index 2e3b11f..6f69265 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -16,6 +16,23 @@ not always match the Git tags; section versions follow the tags. ## [Unreleased] +### Added + +- `integrity-lineage-models`: a `ComputationRegistration` can record when it ran + and which OpenTelemetry span it was, through `ComputationStatement::create_with_span` + and `ComputationSpan`. `startTime` and `endTime` are when the work began and ended, + separate from the registration `timestamp`. `span` and `parentSpan` identify the + computation's span and the span it ran inside, as `urn:uuid` version 5 UUIDs of + `"{traceId}/{spanId}"` in `OTEL_SPAN_NAMESPACE` (`otel_span_urn`, + `ComputationSpan::from_otel`). A computation names its parent rather than its + children, so each one can be registered as soon as its own span ends, on any node; + the nesting is read back by matching each `parentSpan` to a `span`. +- `integrity-jsonld`: the computation context + (`urn:cid:bafkr4iduavib2lgjcmx5n4ydefbbc62ngyannzi566zv5fsozgbwwfw27u`, + `ig_computation_context_link`), which extends the common context with those + four terms. Only statements that use them reference it; every other + `ComputationRegistration` keeps the common context and its CID. + ## [0.0.19] - 2026-10-16 ### Added diff --git a/Cargo.lock b/Cargo.lock index 7fb84b2..94552fa 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -3996,6 +3996,7 @@ dependencies = [ "ssi 0.16.0", "tokio", "utoipa", + "uuid", ] [[package]] @@ -6857,6 +6858,12 @@ dependencies = [ "digest 0.10.7", ] +[[package]] +name = "sha1_smol" +version = "1.0.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "bbfa15b3dddfee50a0fff136974b3e1bde555604ba463834a7eb7deb6417705d" + [[package]] name = "sha2" version = "0.9.9" @@ -8766,6 +8773,7 @@ dependencies = [ "getrandom 0.3.4", "js-sys", "serde_core", + "sha1_smol", "wasm-bindgen", ] diff --git a/integrity-jsonld/src/lib.rs b/integrity-jsonld/src/lib.rs index df57205..66b6993 100644 --- a/integrity-jsonld/src/lib.rs +++ b/integrity-jsonld/src/lib.rs @@ -25,6 +25,20 @@ pub fn ig_common_context_link() -> String { format!("urn:cid:{cid}").trim_end().to_owned() } +/// Returns the computation context link as a CID URN. +/// +/// This context extends the IG common context with the `startTime`, `endTime`, +/// `span` and `parentSpan` terms of a `ComputationRegistration`. Statements that +/// use none of those terms keep the common context, so their CIDs are unchanged. +/// +/// # Returns +/// URN string in format `urn:cid:{cid}` for the computation context. +pub fn ig_computation_context_link() -> String { + let cid: &'static str = "bafkr4iduavib2lgjcmx5n4ydefbbc62ngyannzi566zv5fsozgbwwfw27u"; + + format!("urn:cid:{cid}") +} + /// Computes RDFC CID for JSON-LD data. /// /// Converts JSON-LD to N-Quads, canonicalizes using URDNA2015, and computes a CID. diff --git a/integrity-jsonld/src/loader.rs b/integrity-jsonld/src/loader.rs index 02bdda3..d94e287 100644 --- a/integrity-jsonld/src/loader.rs +++ b/integrity-jsonld/src/loader.rs @@ -136,8 +136,8 @@ mod tests { .keys() .filter(|k| k.starts_with("urn:cid:")) .count(), - 25, - "expected all 25 urn:cid: contexts to be embedded" + 26, + "expected all 26 urn:cid: contexts to be embedded" ); // A representative CID context (also returned by `ig_common_context_link`). diff --git a/integrity-jsonld/static_contexts/cid/bafkr4iduavib2lgjcmx5n4ydefbbc62ngyannzi566zv5fsozgbwwfw27u b/integrity-jsonld/static_contexts/cid/bafkr4iduavib2lgjcmx5n4ydefbbc62ngyannzi566zv5fsozgbwwfw27u new file mode 100644 index 0000000..bd521c8 --- /dev/null +++ b/integrity-jsonld/static_contexts/cid/bafkr4iduavib2lgjcmx5n4ydefbbc62ngyannzi566zv5fsozgbwwfw27u @@ -0,0 +1,28 @@ +{ + "@context": [ + "urn:cid:bafkr4icploa577ziqnb57jlpoj7l2hi5kgt3knxpdtunlttjd3q33zeqpy", + { + "@protected": true, + + "startTime": { + "@id": "https://eqtylab.io/terms/startTime", + "@type": "http://www.w3.org/2001/XMLSchema#dateTime" + }, + + "endTime": { + "@id": "https://eqtylab.io/terms/endTime", + "@type": "http://www.w3.org/2001/XMLSchema#dateTime" + }, + + "span": { + "@id": "https://eqtylab.io/terms/span", + "@type": "@id" + }, + + "parentSpan": { + "@id": "https://eqtylab.io/terms/parentSpan", + "@type": "@id" + } + } + ] +} diff --git a/integrity-lineage-models/Cargo.toml b/integrity-lineage-models/Cargo.toml index d3a422d..d4a4211 100644 --- a/integrity-lineage-models/Cargo.toml +++ b/integrity-lineage-models/Cargo.toml @@ -23,6 +23,7 @@ serde_jcs = "0.2.0" serde_json = "1.0" ssi = { version = "0.16", default-features = false, features = ["w3c", "ed25519", "secp256k1", "secp256r1"] } utoipa = { version = "3", features = ["axum_extras", "openapi_extensions"] } +uuid = { version = "1.4", features = ["v5"] } [target.'cfg(target_arch = "wasm32")'.dependencies] blake3 = "1.5" diff --git a/integrity-lineage-models/src/models/statements/computation_statement.rs b/integrity-lineage-models/src/models/statements/computation_statement.rs index 182554a..347a672 100644 --- a/integrity-lineage-models/src/models/statements/computation_statement.rs +++ b/integrity-lineage-models/src/models/statements/computation_statement.rs @@ -1,17 +1,112 @@ -use anyhow::Result; -use integrity_jsonld::ig_common_context_link; +use anyhow::{anyhow, bail, Result}; +use chrono::DateTime; +use integrity_jsonld::{ig_common_context_link, ig_computation_context_link}; use serde::{Deserialize, Serialize}; +use uuid::Uuid; use super::{ compute_cid, format_cids, format_timestamp, get_jsonld_filename, StatementTrait, ValueOrArray, }; use crate::cid::prepend_urn_cid; +/// Namespace of the name-based UUIDs that identify OpenTelemetry spans (see [`otel_span_urn`]). +pub const OTEL_SPAN_NAMESPACE: Uuid = Uuid::from_u128(0x3eeccd21_6917_4419_a92d_893e3fd3a821); + +/// The `urn:uuid` that identifies an OpenTelemetry span: a version 5 (name-based) +/// UUID in [`OTEL_SPAN_NAMESPACE`] of `"{trace_id}/{span_id}"`. +/// +/// The same trace and span always give the same URN, so a child can name its +/// parent from the `traceparent` it is called with, before the parent has ended. +/// IDs are the W3C trace-context hex forms (32 and 16 digits, not all zero) and +/// are read case-insensitively. +pub fn otel_span_urn(trace_id: &str, span_id: &str) -> Result { + let trace_id = otel_hex_id(trace_id, 32, "trace")?; + let span_id = otel_hex_id(span_id, 16, "span")?; + let uuid = Uuid::new_v5( + &OTEL_SPAN_NAMESPACE, + format!("{trace_id}/{span_id}").as_bytes(), + ); + Ok(format!("urn:uuid:{uuid}")) +} + +fn otel_hex_id(id: &str, digits: usize, kind: &str) -> Result { + let id = id.to_ascii_lowercase(); + if id.len() != digits || !id.bytes().all(|b| b.is_ascii_hexdigit()) { + bail!("OpenTelemetry {kind} ID '{id}' is not {digits} hex digits."); + } + if id.bytes().all(|b| b == b'0') { + bail!("OpenTelemetry {kind} ID '{id}' is all zeros, which is invalid."); + } + Ok(id) +} + +/// When a computation ran and which span it was, as an OpenTelemetry span records. +/// +/// `start_time` and `end_time` are when the work itself began and ended, which is +/// different from the statement's `timestamp` (when it was registered). `span` and +/// `parent_span` are `urn:uuid` identifiers of the computation's own span and of the +/// span it ran inside (see [`otel_span_urn`]). A computation names its parent, not +/// its children, so every statement can be registered as soon as its own span ends; +/// the nesting is read back by matching each `parentSpan` to a `span`. +#[derive(Clone, Debug, Default, PartialEq)] +pub struct ComputationSpan { + /// RFC 3339 time at which the computation started + pub start_time: Option, + /// RFC 3339 time at which the computation ended + pub end_time: Option, + /// `urn:uuid` of this computation's span + pub span: Option, + /// `urn:uuid` of the span this computation ran inside + pub parent_span: Option, +} + +impl ComputationSpan { + /// The span of an OpenTelemetry span: its times and its, and its parent's, `urn:uuid`. + pub fn from_otel( + trace_id: &str, + span_id: &str, + parent_span_id: Option<&str>, + start_time: Option, + end_time: Option, + ) -> Result { + Ok(Self { + start_time, + end_time, + span: Some(otel_span_urn(trace_id, span_id)?), + parent_span: parent_span_id + .map(|parent| otel_span_urn(trace_id, parent)) + .transpose()?, + }) + } + + fn is_empty(&self) -> bool { + self.start_time.is_none() + && self.end_time.is_none() + && self.span.is_none() + && self.parent_span.is_none() + } +} + +/// A span identifier in its canonical form, `urn:uuid:` and a lowercase hyphenated UUID. +fn canonical_span_urn(urn: &str, term: &str) -> Result { + let uuid = urn + .strip_prefix("urn:uuid:") + .ok_or_else(|| anyhow!("Computation {term} '{urn}' is not a urn:uuid.")) + .and_then(|uuid| { + Uuid::parse_str(uuid).map_err(|e| anyhow!("Computation {term} '{urn}': {e}")) + })?; + Ok(format!("urn:uuid:{}", uuid.hyphenated())) +} + /// Records a computational process with its inputs and outputs /// /// This statement type captures the lineage of data transformations, /// including the computation performed, input data, output data, /// and the entity that executed the computation. +/// +/// A computation may also record when it ran and which span it was, and the span +/// it ran inside (see [`ComputationSpan`]). Statements that do so use the +/// computation context, which extends the common context with those terms. #[derive(Clone, Debug, Serialize, Deserialize, utoipa::ToSchema, PartialEq)] #[serde(rename_all = "camelCase")] pub struct ComputationStatement { @@ -38,6 +133,18 @@ pub struct ComputationStatement { /// Optional identifier of the system where computation was executed #[serde(skip_serializing_if = "Option::is_none")] pub executed_on: Option, + /// Optional RFC 3339 time at which the computation started + #[serde(default, skip_serializing_if = "Option::is_none")] + pub start_time: Option, + /// Optional RFC 3339 time at which the computation ended + #[serde(default, skip_serializing_if = "Option::is_none")] + pub end_time: Option, + /// Optional `urn:uuid` of this computation's span + #[serde(default, skip_serializing_if = "Option::is_none")] + pub span: Option, + /// Optional `urn:uuid` of the span this computation ran inside + #[serde(default, skip_serializing_if = "Option::is_none")] + pub parent_span: Option, /// DID of the entity that registered this statement pub registered_by: String, /// ISO 8601 timestamp of when the statement was created @@ -88,6 +195,35 @@ impl ComputationStatement { executed_on: Option, registered_by: String, timestamp: Option, + ) -> Result { + Self::create_with_span( + computation, + input, + output, + operated_by, + executed_on, + registered_by, + timestamp, + ComputationSpan::default(), + ) + .await + } + + /// Creates a new ComputationRegistrationStatement object that records when it + /// ran, which span it was, and the span it ran inside. + /// + /// An empty `span` produces the same statement, and the same CID, as [`Self::create`]. + /// A parent span needs the computation's own span, and the two must differ. + #[allow(clippy::too_many_arguments)] + pub async fn create_with_span( + computation: Option, + input: Vec, + output: Vec, + operated_by: String, + executed_on: Option, + registered_by: String, + timestamp: Option, + span: ComputationSpan, ) -> Result { let computation = computation .map(|s| prepend_urn_cid(s.as_str())) @@ -95,8 +231,39 @@ impl ComputationStatement { let input = format_cids(input)?; let output = format_cids(output)?; + let start = span.start_time.as_deref().map(parse_time).transpose()?; + let end = span.end_time.as_deref().map(parse_time).transpose()?; + if let (Some(start), Some(end)) = (start, end) { + if end < start { + bail!("Computation endTime must not be before its startTime."); + } + } + + let context = if span.is_empty() { + ig_common_context_link() + } else { + ig_computation_context_link() + }; + let own_span = span + .span + .as_deref() + .map(|urn| canonical_span_urn(urn, "span")) + .transpose()?; + let parent_span = span + .parent_span + .as_deref() + .map(|urn| canonical_span_urn(urn, "parentSpan")) + .transpose()?; + match (&own_span, &parent_span) { + (None, Some(_)) => bail!("Computation parentSpan needs the computation's own span."), + (Some(own), Some(parent)) if own == parent => { + bail!("Computation span and parentSpan must differ.") + } + _ => {} + } + let statement = Self { - context: ig_common_context_link(), + context, id: String::from("in-progress"), type_: "ComputationRegistration".to_owned(), computation, @@ -104,6 +271,10 @@ impl ComputationStatement { output, operated_by, executed_on, + start_time: span.start_time, + end_time: span.end_time, + span: own_span, + parent_span, registered_by, timestamp: format_timestamp(timestamp), }; @@ -116,6 +287,11 @@ impl ComputationStatement { } } +fn parse_time(time: &str) -> Result> { + DateTime::parse_from_rfc3339(time) + .map_err(|e| anyhow::anyhow!("Computation time '{time}' is not RFC 3339: {e}")) +} + #[cfg(test)] mod tests { use chrono::{DateTime, Utc}; @@ -379,4 +555,159 @@ mod tests { "Type match failed" ); } + + const DID: &str = "did:key:z6Mkvt1grez4Avdvhqc196hTs6Lxb4qmu1NUdGk2An7QKqnT"; + const INPUT: &str = "urn:cid:bafkr4ia3wmrvedxwkjm6jfmtqy2bdcpi47hv5bni7twshohepck3gsgodi"; + const OUTPUT: &str = "urn:cid:bafkr4ifqnz4knhvvgorsjl6hwhtmzfkfurebfslr3mwe2bqygcgu4bq3wi"; + const TRACE: &str = "01a1401d684b7f40a85aba2fa15ba51d"; + const ROOT: &str = "a85aba2fa15ba51d"; + const CHILD: &str = "93460e50e31602c0"; + + async fn with_span(span: ComputationSpan) -> Result { + ComputationStatement::create_with_span( + None, + vec![INPUT.to_owned()], + vec![OUTPUT.to_owned()], + DID.to_owned(), + None, + DID.to_owned(), + Some("2024-06-27T14:36:35Z".to_owned()), + span, + ) + .await + } + + fn full_span() -> ComputationSpan { + ComputationSpan::from_otel( + TRACE, + CHILD, + Some(ROOT), + Some("2024-06-27T14:36:20.404544Z".to_owned()), + Some("2024-06-27T14:36:34.768539Z".to_owned()), + ) + .unwrap() + } + + #[test] + fn otel_span_urn_is_a_v5_uuid_of_trace_and_span() { + let urn = otel_span_urn(TRACE, CHILD).unwrap(); + let uuid = Uuid::parse_str(urn.strip_prefix("urn:uuid:").unwrap()).unwrap(); + assert_eq!(uuid.get_version_num(), 5); + assert_eq!( + uuid, + Uuid::new_v5(&OTEL_SPAN_NAMESPACE, format!("{TRACE}/{CHILD}").as_bytes()) + ); + // Stable across releases: other implementations derive the same value + assert_eq!(urn, "urn:uuid:44f895c1-bcfc-5980-ac00-84ace1063312"); + // Hex case does not matter; another span or trace gives another URN + assert_eq!( + otel_span_urn(&TRACE.to_uppercase(), &CHILD.to_uppercase()).unwrap(), + urn + ); + assert_ne!(otel_span_urn(TRACE, ROOT).unwrap(), urn); + } + + #[test] + fn otel_span_urn_rejects_invalid_ids() { + assert!(otel_span_urn("01a1401d", CHILD).is_err()); + assert!(otel_span_urn(TRACE, "93460e50e31602cz").is_err()); + assert!(otel_span_urn(&"0".repeat(32), CHILD).is_err()); + assert!(otel_span_urn(TRACE, &"0".repeat(16)).is_err()); + } + + #[tokio::test] + async fn empty_span_keeps_common_context_and_cid() { + let plain = with_span(ComputationSpan::default()).await.unwrap(); + let created = ComputationStatement::create( + None, + vec![INPUT.to_owned()], + vec![OUTPUT.to_owned()], + DID.to_owned(), + None, + DID.to_owned(), + Some("2024-06-27T14:36:35Z".to_owned()), + ) + .await + .unwrap(); + + assert_eq!(plain, created); + assert_eq!(plain.context, ig_common_context_link()); + let json = serde_json::to_value(&plain).unwrap(); + for key in ["startTime", "endTime", "span", "parentSpan"] { + assert!( + json.get(key).is_none(), + "{key} serialized on a plain statement" + ); + } + } + + #[tokio::test] + async fn span_uses_computation_context() { + let statement = with_span(full_span()).await.unwrap(); + + assert_eq!(statement.context, ig_computation_context_link()); + let json = serde_json::to_value(&statement).unwrap(); + assert_eq!(json["startTime"], "2024-06-27T14:36:20.404544Z"); + assert_eq!(json["endTime"], "2024-06-27T14:36:34.768539Z"); + assert_eq!(json["span"], otel_span_urn(TRACE, CHILD).unwrap()); + assert_eq!(json["parentSpan"], otel_span_urn(TRACE, ROOT).unwrap()); + assert!(json.get("subcomputations").is_none()); + + let parsed: ComputationStatement = serde_json::from_value(json).unwrap(); + assert_eq!(parsed, statement); + } + + #[tokio::test] + async fn span_terms_are_covered_by_the_cid() { + let base = with_span(full_span()).await.unwrap(); + + let mut later_start = full_span(); + later_start.start_time = Some("2024-06-27T14:36:21Z".to_owned()); + let mut later_end = full_span(); + later_end.end_time = Some("2024-06-27T14:36:35Z".to_owned()); + let mut other_span = full_span(); + other_span.span = Some(otel_span_urn(TRACE, "b0ce072c04d26a6c").unwrap()); + let mut root = full_span(); + root.parent_span = None; + + for span in [later_start, later_end, other_span, root] { + let changed = with_span(span).await.unwrap(); + assert_ne!(changed.get_id(), base.get_id(), "span term not in the CID"); + } + } + + #[tokio::test] + async fn span_urns_are_canonical() { + let mut span = full_span(); + span.span = span + .span + .map(|urn| urn.to_uppercase().replace("URN:UUID:", "urn:uuid:")); + + let statement = with_span(span).await.unwrap(); + assert_eq!(statement.span, Some(otel_span_urn(TRACE, CHILD).unwrap())); + } + + #[tokio::test] + async fn span_rejects_bad_times_and_spans() { + let mut reversed = full_span(); + reversed.end_time = Some("2024-06-27T14:36:00Z".to_owned()); + assert!(with_span(reversed).await.is_err()); + + let mut unparsable = full_span(); + unparsable.start_time = Some("yesterday".to_owned()); + assert!(with_span(unparsable).await.is_err()); + + let mut orphan_parent = full_span(); + orphan_parent.span = None; + assert!(with_span(orphan_parent).await.is_err()); + + let mut own_parent = full_span(); + own_parent.parent_span = own_parent.span.clone(); + assert!(with_span(own_parent).await.is_err()); + + let mut not_a_uuid = full_span(); + not_a_uuid.span = + Some("urn:cid:bafkr4ia3wmrvedxwkjm6jfmtqy2bdcpi47hv5bni7twshohepck3gsgodi".to_owned()); + assert!(with_span(not_a_uuid).await.is_err()); + } } diff --git a/integrity-lineage-models/src/models/statements/mod.rs b/integrity-lineage-models/src/models/statements/mod.rs index 56eb6e6..3c12515 100644 --- a/integrity-lineage-models/src/models/statements/mod.rs +++ b/integrity-lineage-models/src/models/statements/mod.rs @@ -8,7 +8,9 @@ pub mod association_statement; pub use association_statement::{AssociationStatement, AssociationType}; /// Computation statement for recording computational processes pub mod computation_statement; -pub use computation_statement::ComputationStatement; +pub use computation_statement::{ + otel_span_urn, ComputationSpan, ComputationStatement, OTEL_SPAN_NAMESPACE, +}; pub mod common; /// Data statement for recording data artifacts pub mod data_statement;