diff --git a/Cargo.lock b/Cargo.lock index 93cd29fb..7d7e48e6 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1796,6 +1796,7 @@ dependencies = [ "serde", "serde_json", "sha2", + "st2-resource-protocol", "st2-wire", "tempfile", "toml", @@ -1805,6 +1806,15 @@ dependencies = [ "tungstenite", ] +[[package]] +name = "st2-resource-protocol" +version = "0.1.0" +dependencies = [ + "serde", + "serde_json", + "sha2", +] + [[package]] name = "st2-wire" version = "0.1.0" diff --git a/Cargo.toml b/Cargo.toml index a99eb77e..167ea193 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -2,8 +2,8 @@ # `package.version` out of this file as the single source of truth for the # build, and a virtual root has no `[package]` to read. [workspace] -members = ["crates/agent-spec", "crates/st2-wire", "crates/demo-resolver-wasm"] -default-members = [".", "crates/agent-spec", "crates/st2-wire"] +members = ["crates/agent-spec", "crates/st2-resource-protocol", "crates/st2-wire", "crates/demo-resolver-wasm"] +default-members = [".", "crates/agent-spec", "crates/st2-resource-protocol", "crates/st2-wire"] [package] name = "st2" @@ -33,6 +33,7 @@ notify = "8" serde = { version = "1", features = ["derive"] } serde_json = "1" sha2 = "0.10" +st2-resource-protocol = { path = "crates/st2-resource-protocol" } st2-wire = { path = "crates/st2-wire" } tempfile = "3" toml = "0.9" diff --git a/crates/agent-spec/src/profile.rs b/crates/agent-spec/src/profile.rs index a21affc1..be30bc4e 100644 --- a/crates/agent-spec/src/profile.rs +++ b/crates/agent-spec/src/profile.rs @@ -36,7 +36,7 @@ pub const AGENT_GOAL_SCHEME: &str = "dev.schickling.agent-goal"; /// Maximum resolver module bytes admitted by both catalog transactions and the wasm runtime. pub const DEFAULT_MODULE_LIMIT_BYTES: usize = 16 * 1024 * 1024; /// Descriptor ABI implemented by this host. -pub const PROFILE_DESCRIPTOR_ABI_VERSION: u32 = 2; +pub const PROFILE_DESCRIPTOR_ABI_VERSION: u32 = 3; /// Maximum canonical compact JSON bytes accepted for one binding selector. pub const DEFAULT_SELECTOR_LIMIT_BYTES: usize = 16 * 1024; @@ -1091,7 +1091,7 @@ mod tests { use super::*; const VALID_DESCRIPTOR_JSON: &str = r#"{ - "abiVersion": 2, + "abiVersion": 3, "capabilities": ["resolve", "read", "observe"], "selectorSchema": { "type": "object", @@ -1157,7 +1157,7 @@ mod tests { } #[test] - fn valid_v2_descriptor_and_nested_selector_validate() { + fn valid_v3_descriptor_and_nested_selector_validate() { let descriptor = valid_descriptor(); assert_eq!(descriptor.abi_version, PROFILE_DESCRIPTOR_ABI_VERSION); assert_eq!(descriptor.runtime.topology, RuntimeTopology::Shared); @@ -1171,7 +1171,7 @@ mod tests { #[test] fn descriptor_rejects_unknown_abi_capability_and_fields() { - let unknown_abi = VALID_DESCRIPTOR_JSON.replace("\"abiVersion\": 2", "\"abiVersion\": 9"); + let unknown_abi = VALID_DESCRIPTOR_JSON.replace("\"abiVersion\": 3", "\"abiVersion\": 9"); assert!( ProfileDescriptor::from_json(unknown_abi.as_bytes()) .unwrap_err() @@ -1187,7 +1187,7 @@ mod tests { ); let unknown_field = - VALID_DESCRIPTOR_JSON.replace("\"abiVersion\": 2,", "\"abiVersion\": 2, \"extra\": true,"); + VALID_DESCRIPTOR_JSON.replace("\"abiVersion\": 3,", "\"abiVersion\": 3, \"extra\": true,"); assert!( ProfileDescriptor::from_json(unknown_field.as_bytes()) .unwrap_err() diff --git a/crates/agent-spec/tests/profile_wasm.rs b/crates/agent-spec/tests/profile_wasm.rs index 61aef4e1..5335c1bf 100644 --- a/crates/agent-spec/tests/profile_wasm.rs +++ b/crates/agent-spec/tests/profile_wasm.rs @@ -74,7 +74,7 @@ fn descriptor_module(payload: &[u8], reported_len: usize) -> WasmResolver { } const VALID_DESCRIPTOR: &str = r#"{ - "abiVersion": 2, + "abiVersion": 3, "capabilities": ["resolve", "read", "observe"], "selectorSchema": { "type": "object", @@ -112,12 +112,12 @@ fn current_rss_bytes() -> u64 { } #[test] -fn valid_v2_descriptor_executes_and_resolve_only_module_stays_passive() { +fn valid_v3_descriptor_executes_and_resolve_only_module_stays_passive() { let descriptor = descriptor_module(VALID_DESCRIPTOR.as_bytes(), VALID_DESCRIPTOR.len()) .describe_once() .expect("describe call succeeds") .expect("descriptor is present"); - assert_eq!(descriptor.abi_version, 2); + assert_eq!(descriptor.abi_version, 3); assert!( WasmResolver::load(Path::new(DEMO_WASM_PATH)) diff --git a/crates/st2-resource-protocol/Cargo.toml b/crates/st2-resource-protocol/Cargo.toml new file mode 100644 index 00000000..371772cf --- /dev/null +++ b/crates/st2-resource-protocol/Cargo.toml @@ -0,0 +1,11 @@ +[package] +name = "st2-resource-protocol" +version = "0.1.0" +edition = "2024" +description = "Resource Profile runtime wire types, framing, and codecs shared by st2 hosts and runtimes." +license = "MIT" + +[dependencies] +serde = { version = "1", features = ["derive"] } +serde_json = "1" +sha2 = "0.10" diff --git a/crates/st2-resource-protocol/src/lib.rs b/crates/st2-resource-protocol/src/lib.rs new file mode 100644 index 00000000..0c5e60b4 --- /dev/null +++ b/crates/st2-resource-protocol/src/lib.rs @@ -0,0 +1,1053 @@ +//! Resource Profile runtime wire types, framing, and codecs. + +use std::collections::BTreeSet; +use std::fmt; +use std::path::PathBuf; + +use serde::de::{self, Visitor}; +use serde::{Deserialize, Deserializer, Serialize, Serializer}; +use serde_json::Value; +use sha2::{Digest as _, Sha256}; + +pub const MAX_PROTOCOL_LINE_BYTES: usize = 2 * 1024 * 1024; +pub const MAX_SNAPSHOT_BYTES: usize = 1024 * 1024; +pub const MAX_SELECTOR_BYTES: usize = 16 * 1024; +pub const MAX_HEALTH_DETAIL_BYTES: usize = 16 * 1024; +const MAX_OPAQUE_ID_BYTES: usize = 16 * 1024; + +/// Fact bounds are deliberately small relative to the 2 MiB frame: even 32 facts whose strings +/// all require JSON escaping leave ample room beside a maximal base64-encoded 1 MiB snapshot. +pub const MAX_FACTS: usize = 32; +pub const MAX_FACT_KEY_BYTES: usize = 128; +pub const MAX_FACT_VALUE_BYTES: usize = 1024; + +#[derive(Debug, Clone, PartialEq, Eq, Default)] +pub enum FactValue { + #[default] + Omitted, + Null, + Value(String), +} + +impl FactValue { + pub fn value(value: impl Into) -> Self { + Self::Value(value.into()) + } + + pub fn as_option(&self) -> Option> { + match self { + Self::Omitted => None, + Self::Null => Some(None), + Self::Value(value) => Some(Some(value)), + } + } + + fn is_omitted(&self) -> bool { + matches!(self, Self::Omitted) + } +} + +impl Serialize for FactValue { + fn serialize(&self, serializer: S) -> Result + where + S: Serializer, + { + match self { + Self::Omitted => serializer.serialize_unit(), + Self::Null => serializer.serialize_none(), + Self::Value(value) => serializer.serialize_str(value), + } + } +} + +impl<'de> Deserialize<'de> for FactValue { + fn deserialize(deserializer: D) -> Result + where + D: Deserializer<'de>, + { + Option::::deserialize(deserializer) + .map(|value| value.map_or(Self::Null, Self::Value)) + } +} + +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "camelCase", deny_unknown_fields)] +pub struct ResourceFact { + key: String, + #[serde(default, skip_serializing_if = "FactValue::is_omitted")] + before: FactValue, + #[serde(default, skip_serializing_if = "FactValue::is_omitted")] + after: FactValue, +} + +impl ResourceFact { + pub fn new( + key: impl Into, + before: FactValue, + after: FactValue, + ) -> Result { + let fact = Self { + key: key.into(), + before, + after, + }; + fact.validate()?; + Ok(fact) + } + + pub fn current( + key: impl Into, + value: impl Into, + ) -> Result { + Self::new(key, FactValue::Omitted, FactValue::value(value)) + } + + pub fn transition( + key: impl Into, + before: Option>, + after: Option>, + ) -> Result { + Self::new( + key, + before.map_or(FactValue::Null, |value| FactValue::value(value)), + after.map_or(FactValue::Null, |value| FactValue::value(value)), + ) + } + + pub fn key(&self) -> &str { + &self.key + } + + pub fn before(&self) -> Option> { + self.before.as_option() + } + + pub fn after(&self) -> Option> { + self.after.as_option() + } + + pub fn validate(&self) -> Result<(), FactError> { + validate_fact_string("key", &self.key, MAX_FACT_KEY_BYTES, true)?; + if self.before.is_omitted() && self.after.is_omitted() { + return Err(FactError::MissingValue); + } + for (field, value) in [("before", &self.before), ("after", &self.after)] { + if let FactValue::Value(value) = value { + validate_fact_string(field, value, MAX_FACT_VALUE_BYTES, false)?; + } + } + Ok(()) + } +} + +fn validate_fact_string( + field: &'static str, + value: &str, + maximum: usize, + nonempty: bool, +) -> Result<(), FactError> { + if nonempty && value.is_empty() { + return Err(FactError::Empty { field }); + } + if value.len() > maximum { + return Err(FactError::TooLarge { + field, + actual: value.len(), + maximum, + }); + } + if value + .chars() + .any(|character| character.is_control() || matches!(character, '\u{2028}' | '\u{2029}')) + { + return Err(FactError::NotPrintable { field }); + } + Ok(()) +} + +#[derive(Debug, Clone, PartialEq, Eq)] +pub enum FactError { + TooMany { actual: usize }, + Empty { + field: &'static str, + }, + TooLarge { + field: &'static str, + actual: usize, + maximum: usize, + }, + NotPrintable { + field: &'static str, + }, + MissingValue, +} + +impl fmt::Display for FactError { + fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result { + match self { + Self::TooMany { actual } => { + write!(formatter, "fact list has {actual} entries; maximum is {MAX_FACTS}") + } + Self::Empty { field } => write!(formatter, "fact {field} must not be empty"), + Self::TooLarge { + field, + actual, + maximum, + } => write!( + formatter, + "fact {field} is {actual} bytes; maximum is {maximum}" + ), + Self::NotPrintable { field } => { + write!(formatter, "fact {field} must be one printable line") + } + Self::MissingValue => { + formatter.write_str("fact must include before, after, or both") + } + } + } +} + +impl std::error::Error for FactError {} + +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct OpaqueIdError { + kind: &'static str, + reason: &'static str, +} + +impl fmt::Display for OpaqueIdError { + fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result { + write!(formatter, "{} {}", self.kind, self.reason) + } +} + +impl std::error::Error for OpaqueIdError {} + +macro_rules! opaque_id { + ($name:ident, $kind:literal) => { + #[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord, Hash, Serialize)] + #[serde(transparent)] + pub struct $name(String); + + impl $name { + pub fn new(value: impl Into) -> Result { + let value = value.into(); + if value.is_empty() { + return Err(OpaqueIdError { + kind: $kind, + reason: "must not be empty", + }); + } + if value.len() > MAX_OPAQUE_ID_BYTES { + return Err(OpaqueIdError { + kind: $kind, + reason: "is too large", + }); + } + Ok(Self(value)) + } + + pub fn as_str(&self) -> &str { + &self.0 + } + } + + impl<'de> Deserialize<'de> for $name { + fn deserialize(deserializer: D) -> Result + where + D: Deserializer<'de>, + { + let value = String::deserialize(deserializer)?; + Self::new(value).map_err(de::Error::custom) + } + } + }; +} + +opaque_id!(RuntimeIncarnation, "runtime incarnation"); +opaque_id!(OwnerClaim, "owner claim"); +opaque_id!(BindingId, "binding id"); +opaque_id!(RegistrationToken, "registration token"); + +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "camelCase", deny_unknown_fields)] +pub struct RuntimeOwner { + incarnation: RuntimeIncarnation, + claim: OwnerClaim, +} + +impl RuntimeOwner { + pub fn new(incarnation: RuntimeIncarnation, claim: OwnerClaim) -> Self { + Self { incarnation, claim } + } + + pub fn incarnation(&self) -> &RuntimeIncarnation { + &self.incarnation + } + + pub fn claim(&self) -> &OwnerClaim { + &self.claim + } +} + +#[derive(Clone, PartialEq, Eq)] +pub struct SnapshotBytes(Vec); + +impl fmt::Debug for SnapshotBytes { + fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result { + formatter + .debug_struct("SnapshotBytes") + .field("len", &self.0.len()) + .finish() + } +} + +impl SnapshotBytes { + pub fn new(bytes: Vec) -> Result { + if bytes.len() > MAX_SNAPSHOT_BYTES { + return Err(SnapshotSizeError { + actual: bytes.len(), + }); + } + Ok(Self(bytes)) + } + + pub fn as_slice(&self) -> &[u8] { + &self.0 + } + + pub fn into_vec(self) -> Vec { + self.0 + } +} + +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub struct SnapshotSizeError { + pub actual: usize, +} + +impl fmt::Display for SnapshotSizeError { + fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result { + write!( + formatter, + "decoded snapshot is {} bytes; maximum is {MAX_SNAPSHOT_BYTES}", + self.actual + ) + } +} + +impl std::error::Error for SnapshotSizeError {} + +impl Serialize for SnapshotBytes { + fn serialize(&self, serializer: S) -> Result + where + S: Serializer, + { + serializer.serialize_str(&encode_base64(&self.0)) + } +} + +impl<'de> Deserialize<'de> for SnapshotBytes { + fn deserialize(deserializer: D) -> Result + where + D: Deserializer<'de>, + { + struct SnapshotBytesVisitor; + + impl Visitor<'_> for SnapshotBytesVisitor { + type Value = SnapshotBytes; + + fn expecting(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result { + formatter.write_str("an RFC 4648 padded base64 snapshot") + } + + fn visit_str(self, value: &str) -> Result + where + E: de::Error, + { + let bytes = decode_base64(value).map_err(E::custom)?; + SnapshotBytes::new(bytes).map_err(E::custom) + } + } + + deserializer.deserialize_str(SnapshotBytesVisitor) + } +} + +#[derive(Debug, Clone, PartialEq, Eq)] +struct Base64Error(&'static str); + +impl fmt::Display for Base64Error { + fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result { + formatter.write_str(self.0) + } +} + +fn encode_base64(bytes: &[u8]) -> String { + const ALPHABET: &[u8; 64] = b"ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz0123456789+/"; + let mut encoded = String::with_capacity(bytes.len().div_ceil(3) * 4); + for chunk in bytes.chunks(3) { + let first = chunk[0]; + let second = chunk.get(1).copied().unwrap_or(0); + let third = chunk.get(2).copied().unwrap_or(0); + encoded.push(ALPHABET[(first >> 2) as usize] as char); + encoded.push(ALPHABET[(((first & 0x03) << 4) | (second >> 4)) as usize] as char); + if chunk.len() > 1 { + encoded.push(ALPHABET[(((second & 0x0f) << 2) | (third >> 6)) as usize] as char); + } else { + encoded.push('='); + } + if chunk.len() > 2 { + encoded.push(ALPHABET[(third & 0x3f) as usize] as char); + } else { + encoded.push('='); + } + } + encoded +} + +fn decode_base64(encoded: &str) -> Result, Base64Error> { + if encoded.len() % 4 != 0 { + return Err(Base64Error("base64 length is not a multiple of four")); + } + let maximum_encoded = MAX_SNAPSHOT_BYTES.div_ceil(3) * 4; + if encoded.len() > maximum_encoded { + return Err(Base64Error("decoded snapshot exceeds the size limit")); + } + if encoded.is_empty() { + return Ok(Vec::new()); + } + + fn value(byte: u8) -> Result { + match byte { + b'A'..=b'Z' => Ok(byte - b'A'), + b'a'..=b'z' => Ok(byte - b'a' + 26), + b'0'..=b'9' => Ok(byte - b'0' + 52), + b'+' => Ok(62), + b'/' => Ok(63), + _ => Err(Base64Error("base64 contains an invalid character")), + } + } + + let input = encoded.as_bytes(); + let padding = + usize::from(input[input.len() - 1] == b'=') + usize::from(input[input.len() - 2] == b'='); + let decoded_len = input.len() / 4 * 3 - padding; + if decoded_len > MAX_SNAPSHOT_BYTES { + return Err(Base64Error("decoded snapshot exceeds the size limit")); + } + let mut decoded = Vec::with_capacity(decoded_len); + let chunks = input.chunks_exact(4); + let chunk_count = chunks.len(); + for (index, chunk) in chunks.enumerate() { + let last = index + 1 == chunk_count; + let a = value(chunk[0])?; + let b = value(chunk[1])?; + decoded.push((a << 2) | (b >> 4)); + match (chunk[2], chunk[3]) { + (b'=', b'=') if last => { + if b & 0x0f != 0 { + return Err(Base64Error("base64 has non-canonical trailing bits")); + } + } + (third, b'=') if last => { + let c = value(third)?; + if c & 0x03 != 0 { + return Err(Base64Error("base64 has non-canonical trailing bits")); + } + decoded.push((b << 4) | (c >> 2)); + } + (b'=', _) => return Err(Base64Error("base64 padding is misplaced")), + (third, fourth) => { + let c = value(third)?; + let d = value(fourth)?; + decoded.push((b << 4) | (c >> 2)); + decoded.push((c << 6) | d); + } + } + } + Ok(decoded) +} + +#[derive(Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Hash)] +pub struct SnapshotDigest([u8; 32]); + +impl SnapshotDigest { + pub fn of(bytes: &[u8]) -> Self { + Self(Sha256::digest(bytes).into()) + } + + pub fn as_bytes(&self) -> &[u8; 32] { + &self.0 + } +} + +impl fmt::Debug for SnapshotDigest { + fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result { + fmt::Display::fmt(self, formatter) + } +} + +impl fmt::Display for SnapshotDigest { + fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result { + for byte in self.0 { + write!(formatter, "{byte:02x}")?; + } + Ok(()) + } +} + +impl Serialize for SnapshotDigest { + fn serialize(&self, serializer: S) -> Result + where + S: Serializer, + { + serializer.collect_str(self) + } +} + +impl<'de> Deserialize<'de> for SnapshotDigest { + fn deserialize(deserializer: D) -> Result + where + D: Deserializer<'de>, + { + struct DigestVisitor; + + impl Visitor<'_> for DigestVisitor { + type Value = SnapshotDigest; + + fn expecting(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result { + formatter.write_str("a lowercase 64-character SHA-256 digest") + } + + fn visit_str(self, value: &str) -> Result + where + E: de::Error, + { + if value.len() != 64 || value.bytes().any(|byte| !byte.is_ascii_hexdigit()) { + return Err(E::custom("invalid SHA-256 digest")); + } + if value.bytes().any(|byte| byte.is_ascii_uppercase()) { + return Err(E::custom("SHA-256 digest must use lowercase hex")); + } + let mut digest = [0_u8; 32]; + for (index, pair) in value.as_bytes().chunks_exact(2).enumerate() { + let pair = std::str::from_utf8(pair).map_err(E::custom)?; + digest[index] = u8::from_str_radix(pair, 16).map_err(E::custom)?; + } + Ok(SnapshotDigest(digest)) + } + } + + deserializer.deserialize_str(DigestVisitor) + } +} + +#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)] +#[serde( + tag = "type", + rename_all = "camelCase", + rename_all_fields = "camelCase", + deny_unknown_fields +)] +pub enum HostMessage { + Register { + owner: RuntimeOwner, + binding_id: BindingId, + registration: RegistrationToken, + uri: String, + selector: Value, + carrier_path: PathBuf, + #[serde(skip_serializing_if = "Option::is_none")] + previous_digest: Option, + }, + Unregister { + owner: RuntimeOwner, + binding_id: BindingId, + registration: RegistrationToken, + }, +} + +#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "camelCase")] +pub enum RuntimeHealthState { + Starting, + Ready, + Degraded, + Failed, +} + +#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)] +#[serde( + tag = "type", + rename_all = "camelCase", + rename_all_fields = "camelCase", + deny_unknown_fields +)] +pub enum RuntimeMessage { + Publish { + owner: RuntimeOwner, + binding_id: BindingId, + registration: RegistrationToken, + schema_id: String, + media_type: String, + bytes: SnapshotBytes, + topics: Vec, + #[serde(skip_serializing_if = "Option::is_none")] + facts: Option>, + #[serde(skip_serializing_if = "Option::is_none")] + observed_at: Option, + }, + Health { + owner: RuntimeOwner, + #[serde(skip_serializing_if = "Option::is_none")] + binding_id: Option, + #[serde(skip_serializing_if = "Option::is_none")] + registration: Option, + state: RuntimeHealthState, + #[serde(skip_serializing_if = "Option::is_none")] + detail: Option, + }, +} + +#[derive(Debug)] +pub enum ProtocolError { + MissingNewline, + MultipleLines, + EmptyLine, + LineTooLarge { actual: usize }, + SelectorTooLarge { actual: usize }, + HealthDetailTooLarge { actual: usize }, + InvalidTopics(&'static str), + InvalidFacts(FactError), + InvalidHealthScope, + Json(serde_json::Error), +} + +impl fmt::Display for ProtocolError { + fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result { + match self { + Self::MissingNewline => formatter.write_str("protocol frame is missing its newline"), + Self::MultipleLines => formatter.write_str("protocol frame contains multiple lines"), + Self::EmptyLine => formatter.write_str("protocol frame is empty"), + Self::LineTooLarge { actual } => write!( + formatter, + "protocol line is {actual} bytes; maximum is {MAX_PROTOCOL_LINE_BYTES}" + ), + Self::SelectorTooLarge { actual } => write!( + formatter, + "selector is {actual} bytes; maximum is {MAX_SELECTOR_BYTES}" + ), + Self::HealthDetailTooLarge { actual } => write!( + formatter, + "health detail is {actual} bytes; maximum is {MAX_HEALTH_DETAIL_BYTES}" + ), + Self::InvalidTopics(reason) => write!(formatter, "invalid topics: {reason}"), + Self::InvalidFacts(error) => write!(formatter, "invalid facts: {error}"), + Self::InvalidHealthScope => formatter + .write_str("binding-scoped health must carry both bindingId and registration"), + Self::Json(error) => write!(formatter, "invalid protocol JSON: {error}"), + } + } +} + +impl std::error::Error for ProtocolError { + fn source(&self) -> Option<&(dyn std::error::Error + 'static)> { + match self { + Self::Json(error) => Some(error), + Self::InvalidFacts(error) => Some(error), + _ => None, + } + } +} + +pub fn decode_host_line(line: &[u8]) -> Result { + let payload = protocol_payload(line)?; + let message = serde_json::from_slice(payload).map_err(ProtocolError::Json)?; + validate_host_message(&message)?; + Ok(message) +} + +pub fn decode_runtime_line(line: &[u8]) -> Result { + let payload = protocol_payload(line)?; + let message = serde_json::from_slice(payload).map_err(ProtocolError::Json)?; + validate_runtime_message(&message)?; + Ok(message) +} + +pub fn encode_host_line(message: &HostMessage) -> Result, ProtocolError> { + validate_host_message(message)?; + encode_protocol_line(message) +} + +pub fn encode_runtime_line(message: &RuntimeMessage) -> Result, ProtocolError> { + validate_runtime_message(message)?; + encode_protocol_line(message) +} + +fn protocol_payload(line: &[u8]) -> Result<&[u8], ProtocolError> { + if line.len() > MAX_PROTOCOL_LINE_BYTES { + return Err(ProtocolError::LineTooLarge { actual: line.len() }); + } + let Some(payload) = line.strip_suffix(b"\n") else { + return Err(ProtocolError::MissingNewline); + }; + if payload.is_empty() { + return Err(ProtocolError::EmptyLine); + } + if payload.contains(&b'\n') || payload.contains(&b'\r') { + return Err(ProtocolError::MultipleLines); + } + Ok(payload) +} + +fn encode_protocol_line(message: &impl Serialize) -> Result, ProtocolError> { + let mut line = serde_json::to_vec(message).map_err(ProtocolError::Json)?; + line.push(b'\n'); + if line.len() > MAX_PROTOCOL_LINE_BYTES { + return Err(ProtocolError::LineTooLarge { actual: line.len() }); + } + Ok(line) +} + +fn validate_host_message(message: &HostMessage) -> Result<(), ProtocolError> { + if let HostMessage::Register { selector, .. } = message { + let actual = serde_json::to_vec(selector) + .map_err(ProtocolError::Json)? + .len(); + if actual > MAX_SELECTOR_BYTES { + return Err(ProtocolError::SelectorTooLarge { actual }); + } + } + Ok(()) +} + +fn validate_runtime_message(message: &RuntimeMessage) -> Result<(), ProtocolError> { + match message { + RuntimeMessage::Publish { topics, facts, .. } => { + validate_topics(topics)?; + validate_facts(facts.as_deref().unwrap_or_default()) + } + RuntimeMessage::Health { + binding_id, + registration, + detail, + .. + } => { + if binding_id.is_some() != registration.is_some() { + return Err(ProtocolError::InvalidHealthScope); + } + if let Some(detail) = detail + && detail.len() > MAX_HEALTH_DETAIL_BYTES + { + return Err(ProtocolError::HealthDetailTooLarge { + actual: detail.len(), + }); + } + Ok(()) + } + } +} +fn validate_facts(facts: &[ResourceFact]) -> Result<(), ProtocolError> { + if facts.len() > MAX_FACTS { + return Err(ProtocolError::InvalidFacts(FactError::TooMany { + actual: facts.len(), + })); + } + facts + .iter() + .try_for_each(ResourceFact::validate) + .map_err(ProtocolError::InvalidFacts) +} + + +fn validate_topics(topics: &[String]) -> Result<(), ProtocolError> { + let mut unique = BTreeSet::new(); + for topic in topics { + if topic.is_empty() { + return Err(ProtocolError::InvalidTopics( + "topic names must not be empty", + )); + } + if !unique.insert(topic.as_str()) { + return Err(ProtocolError::InvalidTopics("topic names must be unique")); + } + } + Ok(()) +} + +#[cfg(test)] +mod tests { + use super::*; + use serde_json::json; + + fn owner() -> RuntimeOwner { + RuntimeOwner::new( + RuntimeIncarnation::new("incarnation").unwrap(), + OwnerClaim::new("claim").unwrap(), + ) + } + + fn publish(bytes: &[u8]) -> RuntimeMessage { + RuntimeMessage::Publish { + owner: owner(), + binding_id: BindingId::new("binding").unwrap(), + registration: RegistrationToken::new("registration").unwrap(), + schema_id: "schema.v1".to_owned(), + media_type: "application/json".to_owned(), + bytes: SnapshotBytes::new(bytes.to_vec()).unwrap(), + topics: vec!["selected".to_owned()], + facts: None, + observed_at: Some("2026-08-30T00:00:00Z".to_owned()), + } + } + + #[test] + fn host_frames_have_exact_json_shape_and_newline() { + let register = HostMessage::Register { + owner: owner(), + binding_id: BindingId::new("binding").unwrap(), + registration: RegistrationToken::new("registration").unwrap(), + uri: "resource://one".to_owned(), + selector: json!({"kind": "one"}), + carrier_path: PathBuf::from("resources/one.json"), + previous_digest: Some(SnapshotDigest::of(b"previous")), + }; + assert_eq!( + encode_host_line(®ister).unwrap(), + b"{\"type\":\"register\",\"owner\":{\"incarnation\":\"incarnation\",\"claim\":\"claim\"},\"bindingId\":\"binding\",\"registration\":\"registration\",\"uri\":\"resource://one\",\"selector\":{\"kind\":\"one\"},\"carrierPath\":\"resources/one.json\",\"previousDigest\":\"6da0633528deaa0144e7b058315f0b753ec0b945163a72bf96a0d18180f9de0d\"}\n" + ); + let unregister = HostMessage::Unregister { + owner: owner(), + binding_id: BindingId::new("binding").unwrap(), + registration: RegistrationToken::new("registration").unwrap(), + }; + assert_eq!( + encode_host_line(&unregister).unwrap(), + b"{\"type\":\"unregister\",\"owner\":{\"incarnation\":\"incarnation\",\"claim\":\"claim\"},\"bindingId\":\"binding\",\"registration\":\"registration\"}\n" + ); + } + + #[test] + fn runtime_frames_have_exact_json_shape_and_padded_base64() { + assert_eq!( + encode_runtime_line(&publish(b"one byte")).unwrap(), + b"{\"type\":\"publish\",\"owner\":{\"incarnation\":\"incarnation\",\"claim\":\"claim\"},\"bindingId\":\"binding\",\"registration\":\"registration\",\"schemaId\":\"schema.v1\",\"mediaType\":\"application/json\",\"bytes\":\"b25lIGJ5dGU=\",\"topics\":[\"selected\"],\"observedAt\":\"2026-08-30T00:00:00Z\"}\n" + ); + let health = RuntimeMessage::Health { + owner: owner(), + binding_id: None, + registration: None, + state: RuntimeHealthState::Ready, + detail: None, + }; + assert_eq!( + encode_runtime_line(&health).unwrap(), + b"{\"type\":\"health\",\"owner\":{\"incarnation\":\"incarnation\",\"claim\":\"claim\"},\"state\":\"ready\"}\n" + ); + assert_eq!( + decode_runtime_line(&encode_runtime_line(&publish(b"one byte")).unwrap()).unwrap(), + publish(b"one byte") + ); + } + + #[test] + fn fact_wire_shape_distinguishes_omission_from_explicit_null() { + let mut message = publish(b"fact"); + let RuntimeMessage::Publish { facts, .. } = &mut message else { + unreachable!(); + }; + *facts = Some(vec![ + ResourceFact::current("state", "ready").unwrap(), + ResourceFact::transition("label", None::, Some("added")).unwrap(), + ResourceFact::transition("removed", Some("old"), None::).unwrap(), + ]); + let encoded = encode_runtime_line(&message).unwrap(); + let json: Value = serde_json::from_slice(encoded.strip_suffix(b"\n").unwrap()).unwrap(); + assert_eq!( + json["facts"], + json!([ + {"key": "state", "after": "ready"}, + {"key": "label", "before": null, "after": "added"}, + {"key": "removed", "before": "old", "after": null} + ]) + ); + assert_eq!(decode_runtime_line(&encoded).unwrap(), message); + } + + #[test] + fn invalid_fact_shapes_and_bounds_are_rejected() { + let missing_values = b"{\"type\":\"publish\",\"owner\":{\"incarnation\":\"i\",\"claim\":\"c\"},\"bindingId\":\"b\",\"registration\":\"r\",\"schemaId\":\"s\",\"mediaType\":\"m\",\"bytes\":\"\",\"topics\":[],\"facts\":[{\"key\":\"state\"}]}\n"; + assert!(matches!( + decode_runtime_line(missing_values), + Err(ProtocolError::InvalidFacts(FactError::MissingValue)) + )); + assert!(ResourceFact::current("", "value").is_err()); + assert!(ResourceFact::current("state", "two\nlines").is_err()); + assert!(ResourceFact::current("x".repeat(MAX_FACT_KEY_BYTES + 1), "value").is_err()); + assert!(ResourceFact::current("state", "x".repeat(MAX_FACT_VALUE_BYTES + 1)).is_err()); + + let mut message = publish(b"facts"); + let RuntimeMessage::Publish { facts, .. } = &mut message else { + unreachable!(); + }; + *facts = Some( + (0..=MAX_FACTS) + .map(|index| ResourceFact::current(format!("key-{index}"), "value").unwrap()) + .collect(), + ); + assert!(matches!( + encode_runtime_line(&message), + Err(ProtocolError::InvalidFacts(FactError::TooMany { .. })) + )); + } + + #[test] + fn decoding_is_strict_about_fields_ids_and_digest_encoding() { + let unknown_message_field = b"{\"type\":\"health\",\"owner\":{\"incarnation\":\"i\",\"claim\":\"c\"},\"state\":\"ready\",\"extra\":true}\n"; + assert!(matches!( + decode_runtime_line(unknown_message_field), + Err(ProtocolError::Json(_)) + )); + let unknown_owner_field = b"{\"type\":\"health\",\"owner\":{\"incarnation\":\"i\",\"claim\":\"c\",\"extra\":true},\"state\":\"ready\"}\n"; + assert!(matches!( + decode_runtime_line(unknown_owner_field), + Err(ProtocolError::Json(_)) + )); + let empty_id = b"{\"type\":\"unregister\",\"owner\":{\"incarnation\":\"\",\"claim\":\"c\"},\"bindingId\":\"b\",\"registration\":\"r\"}\n"; + assert!(matches!( + decode_host_line(empty_id), + Err(ProtocolError::Json(_)) + )); + let uppercase_digest = b"{\"type\":\"register\",\"owner\":{\"incarnation\":\"i\",\"claim\":\"c\"},\"bindingId\":\"b\",\"registration\":\"r\",\"uri\":\"u\",\"selector\":{},\"carrierPath\":\"p\",\"previousDigest\":\"AFA459DEEC028BA69A538AB3DF3ED61A63F9C8383C2FDAFA95ED544547FB675D\"}\n"; + assert!(matches!( + decode_host_line(uppercase_digest), + Err(ProtocolError::Json(_)) + )); + } + + #[test] + fn decoding_rejects_malformed_frames() { + assert!(matches!( + decode_runtime_line(b"{}"), + Err(ProtocolError::MissingNewline) + )); + assert!(matches!( + decode_runtime_line(b"\n"), + Err(ProtocolError::EmptyLine) + )); + assert!(matches!( + decode_runtime_line(b"{}\n{}\n"), + Err(ProtocolError::MultipleLines) + )); + assert!(matches!( + decode_runtime_line(b"{}\r\n"), + Err(ProtocolError::MultipleLines) + )); + assert!(matches!( + decode_runtime_line(b"not json\n"), + Err(ProtocolError::Json(_)) + )); + } + + #[test] + fn snapshot_base64_rejects_malformed_and_noncanonical_values() { + for encoded in ["A", "!!!!", "A===", "AB==", "AAB=", "AA=A", "AA==AAAA"] { + let line = format!( + "{{\"type\":\"publish\",\"owner\":{{\"incarnation\":\"i\",\"claim\":\"c\"}},\"bindingId\":\"b\",\"registration\":\"r\",\"schemaId\":\"s\",\"mediaType\":\"m\",\"bytes\":\"{encoded}\",\"topics\":[]}}\n" + ); + assert!( + matches!( + decode_runtime_line(line.as_bytes()), + Err(ProtocolError::Json(_)) + ), + "accepted {encoded}" + ); + } + } + + #[test] + fn protocol_enforces_all_byte_limits() { + assert!(matches!( + decode_runtime_line(&vec![b'x'; MAX_PROTOCOL_LINE_BYTES + 1]), + Err(ProtocolError::LineTooLarge { actual }) if actual == MAX_PROTOCOL_LINE_BYTES + 1 + )); + assert_eq!( + SnapshotBytes::new(vec![0; MAX_SNAPSHOT_BYTES + 1]).unwrap_err(), + SnapshotSizeError { + actual: MAX_SNAPSHOT_BYTES + 1 + } + ); + let oversized_snapshot = encode_base64(&vec![0; MAX_SNAPSHOT_BYTES + 1]); + let snapshot_line = format!( + "{{\"type\":\"publish\",\"owner\":{{\"incarnation\":\"i\",\"claim\":\"c\"}},\"bindingId\":\"b\",\"registration\":\"r\",\"schemaId\":\"s\",\"mediaType\":\"m\",\"bytes\":\"{oversized_snapshot}\",\"topics\":[]}}\n" + ); + assert!(matches!( + decode_runtime_line(snapshot_line.as_bytes()), + Err(ProtocolError::Json(_)) + )); + + let register = HostMessage::Register { + owner: owner(), + binding_id: BindingId::new("binding").unwrap(), + registration: RegistrationToken::new("registration").unwrap(), + uri: "u".to_owned(), + selector: Value::String("x".repeat(MAX_SELECTOR_BYTES)), + carrier_path: PathBuf::from("p"), + previous_digest: None, + }; + assert!(matches!( + encode_host_line(®ister), + Err(ProtocolError::SelectorTooLarge { .. }) + )); + let mut register_line = serde_json::to_vec(®ister).unwrap(); + register_line.push(b'\n'); + assert!(matches!( + decode_host_line(®ister_line), + Err(ProtocolError::SelectorTooLarge { .. }) + )); + + let health = RuntimeMessage::Health { + owner: owner(), + binding_id: None, + registration: None, + state: RuntimeHealthState::Degraded, + detail: Some("x".repeat(MAX_HEALTH_DETAIL_BYTES + 1)), + }; + assert!(matches!( + encode_runtime_line(&health), + Err(ProtocolError::HealthDetailTooLarge { .. }) + )); + let mut health_line = serde_json::to_vec(&health).unwrap(); + health_line.push(b'\n'); + assert!(matches!( + decode_runtime_line(&health_line), + Err(ProtocolError::HealthDetailTooLarge { .. }) + )); + assert!(RuntimeIncarnation::new("x".repeat(MAX_OPAQUE_ID_BYTES + 1)).is_err()); + } + + #[test] + fn runtime_semantic_validation_applies_on_encode_and_decode() { + let invalid_health = RuntimeMessage::Health { + owner: owner(), + binding_id: Some(BindingId::new("binding").unwrap()), + registration: None, + state: RuntimeHealthState::Failed, + detail: None, + }; + assert!(matches!( + encode_runtime_line(&invalid_health), + Err(ProtocolError::InvalidHealthScope) + )); + let mut duplicate_topics = publish(b"bytes"); + let RuntimeMessage::Publish { topics, .. } = &mut duplicate_topics else { + unreachable!(); + }; + *topics = vec!["same".to_owned(), "same".to_owned()]; + assert!(matches!( + encode_runtime_line(&duplicate_topics), + Err(ProtocolError::InvalidTopics(_)) + )); + } +} diff --git a/docs/vrs/07-resource-profile/requirements.md b/docs/vrs/07-resource-profile/requirements.md index 7d1bbe37..e5dfbbe9 100644 --- a/docs/vrs/07-resource-profile/requirements.md +++ b/docs/vrs/07-resource-profile/requirements.md @@ -207,18 +207,26 @@ one latest-state catch-up. The direction is recorded in - **PROFILE-R16A Finite protocol and publication bounds:** A selector's canonical compact JSON is at most 16 KiB. One encoded runtime-protocol line is at most 2 MiB including its newline. Decoded snapshot bytes are at most 1 MiB. - Health detail is at most 16 KiB of UTF-8. st2 rejects an oversized value - without truncation and contains the failure to the affected runtime or + Health detail is at most 16 KiB of UTF-8. One publication carries at most 32 + ordered facts; each fact key is at most 128 bytes and each before/after value + is at most 1 KiB of printable single-line UTF-8. st2 rejects an oversized + value without truncation and contains the failure to the affected runtime or binding. ### Must bound attention and catch up to current state -- **PROFILE-R17 Semantic invalidation:** When snapshot bytes change, including - on the first successful publication, the profile classifies the change with - zero or more descriptor-published semantic topics. st2 applies the binding - selector before delivery. A selected change emits one thin invalidation - carrying binding identity, current snapshot digest, and selected topics. It - does not copy snapshot bytes or a profile-rendered summary into the event. +- **PROFILE-R17 Semantic invalidation:** Every Resource invalidation carries the + same bounded ordered fact envelope in its durable body and renders at most + three whole facts into a subject of at most 96 Unicode scalars. Observable + profiles may publish facts and semantic topics beside changed snapshot bytes; + st2 validates the facts, applies the binding selector to topics, and retains + both through catch-up. Passive carrier changes publish one `content` topic + and a short digest transition fact. Agent Spec declaration changes publish + ordered binding-label facts for added, removed, and semantically changed + Resource declarations without exposing URIs or reasons; unavailable + declaration parsing falls back to a digest transition fact rather than + dropping the invalidation. Snapshot bytes and provider payloads remain out of + the event. - **PROFILE-R18 Built-in superseding delivery:** Smart Resource invalidations reuse one built-in per-agent delivery stream and the existing inbox, DING, deduplication, and producer-side supersession machinery. The binding name is diff --git a/docs/vrs/07-resource-profile/spec.md b/docs/vrs/07-resource-profile/spec.md index 8e0fa7fc..fcb367a0 100644 --- a/docs/vrs/07-resource-profile/spec.md +++ b/docs/vrs/07-resource-profile/spec.md @@ -50,12 +50,12 @@ closed wasm describe() -> capabilities + selector schema/default + topology catalog-trusted host runtime argv ----+ | v provider-native observation -publish(binding-id, bytes, topics) +publish(binding-id, bytes, topics, facts) | v host validation + contained atomic replacement canonical snapshot + current digest | - v selector + pending-relevance reducer + v selector + pending-relevance reducer retaining topics + facts built-in resync event (key=binding, supersede=true) | v existing inbox + DING @@ -309,10 +309,19 @@ For each active Resource binding, resync applies this precedence: 3. A schemeless path uses the existing agent-directory-relative rule. 4. Every other unregistered scheme remains opaque and unwatchable. -For a passive resolved carrier, Resource Profiles add no event semantics. -Parent-directory observation, rename replacement, digest seeding, equal-byte -deduplication, deterministic transition identity, bounded windows, and built-in -`resync` delivery remain the [`06-resync`](../06-resync/spec.md) pipeline. +Passive resolved carriers use the same Resource fact envelope as observable +publications. A content transition emits topic `content` and one +`digest=→` fact. Declaration subscriptions retain a +bounded summary keyed by binding label whose values digest URI, reason, +inactive state, and selector. On a declaration flush, one bounded catalog parse +derives the current summary: added labels transition absent→`declared`, removed +labels transition `declared`→absent, and changed summary digests publish +`label=changed`, ordered by label. URIs and reasons never enter the event. A +parse failure, unchanged Resource summary, or summary outside fact bounds falls +back to the declaration carrier's digest transition. Parent-directory +observation, rename replacement, digest seeding, equal-byte deduplication, +deterministic transition identity, and bounded windows otherwise remain the +[`06-resync`](../06-resync/spec.md) pipeline. `notify-chain #true` extends only subscription selection. For each active binding through that profile, resync validates the bound agent's supervisor @@ -341,7 +350,7 @@ fresh-instance policy, fuel budget, and no-import rule as `resolve`: ```json { - "abiVersion": 2, + "abiVersion": 3, "capabilities": ["resolve", "read", "observe"], "selectorSchema": { "type": "object", @@ -439,7 +448,7 @@ host -> unregister { runtime -> publish { owner: { incarnation, claim }, bindingId, registration, - schemaId, mediaType, bytes, topics, observedAt? + schemaId, mediaType, bytes, topics, facts?, observedAt? } runtime -> health { owner: { incarnation, claim }, @@ -465,9 +474,12 @@ Each encoded protocol line is at most 2 MiB, including the newline. `publish` encodes `bytes` as a padded RFC 4648 base64 string; the decoded snapshot is opaque. Selectors are at most 16 KiB when encoded as canonical compact JSON. Decoded snapshot bytes are at most 1 MiB. Health `detail` is at most 16 KiB of -UTF-8. These bounds are checked before allocation or decoding where the -transport permits and fail only the affected binding or runtime. st2 never -truncates canonical snapshot bytes or health text to satisfy a bound. +UTF-8. A publication has at most 32 ordered facts; keys are at most 128 bytes +and values at most 1 KiB of printable single-line UTF-8. A fact carries +`key` plus `before`, `after`, or both; explicit JSON null denotes absence. +These bounds are checked before allocation or decoding where the transport +permits and fail only the affected binding or runtime. st2 never truncates +canonical snapshot bytes, facts, or health text to satisfy a bound. The host rejects unknown bindings, stale owners or registrations, mismatched schema or media type, unpublished topics, invalid messages, output after @@ -484,7 +496,7 @@ The resolver's contained carrier path is the observable snapshot path. A successful `publish` follows one host-owned transaction: ```text -validate binding + schema + topics + size +validate binding + schema + topics + facts + size | v write new bytes to contained sibling temporary file @@ -495,8 +507,7 @@ write new bytes to contained sibling temporary file compute/record current sha256 digest and freshness | `-> equal digest: no invalidation - changed digest: apply binding selector -``` + changed digest: apply binding selector and retain selected topics + facts The runtime never writes the carrier directly. Descriptor-relative no-follow containment from the existing resolver contract applies to the temporary file, @@ -516,14 +527,17 @@ manifest, profile event log, or host retention policy. ## Semantic invalidation and catch-up (PROFILE-R17..R20) For every changed digest, including the first successful publication, the host -intersects `publish.topics` with the normalized binding selector. An empty -intersection updates the canonical snapshot and freshness without scheduling -delivery. A non-empty intersection updates this bounded per-binding state: +validates and preserves the runtime's ordered facts and intersects +`publish.topics` with the normalized binding selector. An empty intersection +updates the canonical snapshot and freshness without scheduling delivery. A +non-empty intersection updates this bounded per-binding state: ```text current_snapshot_digest: Digest? last_delivered_digest: Digest? pending_relevant_change: bool +pending_selected_topics: Topic[] +pending_facts: ResourceFact[] deliverable: bool ``` @@ -534,23 +548,26 @@ If delivery is available, st2 emits one event on the existing built-in stream = resync key = binding name supersede = true -subject = resource changed -body = { binding, snapshotDigest, topics } +subject = · [] +body = { binding, snapshotDigest, topics, facts } ``` -The body is a thin invalidation. It contains no snapshot bytes, provider -payload, rendered summary, credential, or provider cursor. Existing event +Subjects are at most 96 Unicode scalars. Facts retain publication order and are +included only whole; topic space is reserved before facts are admitted. If no +fact fits, a compatible bounded fallback remains. The durable body always +retains the complete bounded fact list. It contains no snapshot bytes, provider +payload, credential, URI, reason, or provider cursor. Existing event deduplication, inbox storage, DING rendering, and supersession apply unchanged. -Multiple topics for one atomic publication produce one invalidation, not one -stream or record per topic. +Multiple topics for one atomic publication produce one invalidation. -If delivery is unavailable, a relevant publication sets +If delivery is unavailable, a relevant publication replaces the pending +selected topics and facts with the latest relevant publication and sets `pending_relevant_change = true`. Later irrelevant publications may advance -`current_snapshot_digest` but do not clear the bit. When delivery becomes -available, st2 emits at most one invalidation for the then-current digest and -clears the bit only after event ingress accepts the record. No pending digest -or transition backlog exists. This is level-triggered current-state catch-up, -not event replay. +`current_snapshot_digest` but do not clear that state. When delivery becomes +available, st2 emits at most one invalidation for the then-current digest with +the retained latest relevant fact envelope, and clears it only after event +ingress accepts the record. No transition backlog exists. This is +level-triggered current-state catch-up, not event replay. Health has separate descriptor, selector, runtime, observation, publication, and delivery stages. Every stage reports affected scheme and binding without diff --git a/src/resource_profile.rs b/src/resource_profile.rs index f7ad9b78..55f588d3 100644 --- a/src/resource_profile.rs +++ b/src/resource_profile.rs @@ -14,566 +14,23 @@ use std::os::unix::ffi::OsStrExt as _; use std::path::{Component, Path, PathBuf}; use std::sync::atomic::{AtomicU64, Ordering}; -use serde::de::{self, Visitor}; -use serde::{Deserialize, Deserializer, Serialize, Serializer}; -use serde_json::Value; -use sha2::{Digest as _, Sha256}; - -pub const MAX_PROTOCOL_LINE_BYTES: usize = 2 * 1024 * 1024; -pub const MAX_SNAPSHOT_BYTES: usize = 1024 * 1024; -pub const MAX_SELECTOR_BYTES: usize = 16 * 1024; -pub const MAX_HEALTH_DETAIL_BYTES: usize = 16 * 1024; -const MAX_OPAQUE_ID_BYTES: usize = 16 * 1024; -const MAX_CATCH_UP_FILE_BYTES: usize = 16 * 1024; +use serde::{Deserialize, Serialize}; + +pub use st2_resource_protocol::{ + BindingId, FactError, FactValue, HostMessage, OpaqueIdError, OwnerClaim, ProtocolError, + RegistrationToken, ResourceFact, RuntimeHealthState, RuntimeIncarnation, RuntimeMessage, + RuntimeOwner, SnapshotBytes, SnapshotDigest, SnapshotSizeError, MAX_FACTS, MAX_FACT_KEY_BYTES, + MAX_FACT_VALUE_BYTES, MAX_HEALTH_DETAIL_BYTES, MAX_PROTOCOL_LINE_BYTES, MAX_SELECTOR_BYTES, + MAX_SNAPSHOT_BYTES, decode_host_line, decode_runtime_line, encode_host_line, + encode_runtime_line, +}; + +// Covers the prior state envelope plus 32 maximally sized facts after worst-case JSON escaping. +const MAX_CATCH_UP_FILE_BYTES: usize = 256 * 1024; const CATCH_UP_FILE: &str = "resource-profile-catch-up.json"; const PUBLICATION_INTENT_FILE: &str = "resource-profile-publication-intent.json"; static TEMPORARY_SEQUENCE: AtomicU64 = AtomicU64::new(0); -#[derive(Debug, Clone, PartialEq, Eq)] -pub struct OpaqueIdError { - kind: &'static str, - reason: &'static str, -} - -impl fmt::Display for OpaqueIdError { - fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result { - write!(formatter, "{} {}", self.kind, self.reason) - } -} - -impl std::error::Error for OpaqueIdError {} - -macro_rules! opaque_id { - ($name:ident, $kind:literal) => { - #[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord, Hash, Serialize)] - #[serde(transparent)] - pub struct $name(String); - - impl $name { - pub fn new(value: impl Into) -> Result { - let value = value.into(); - if value.is_empty() { - return Err(OpaqueIdError { - kind: $kind, - reason: "must not be empty", - }); - } - if value.len() > MAX_OPAQUE_ID_BYTES { - return Err(OpaqueIdError { - kind: $kind, - reason: "is too large", - }); - } - Ok(Self(value)) - } - - pub fn as_str(&self) -> &str { - &self.0 - } - } - - impl<'de> Deserialize<'de> for $name { - fn deserialize(deserializer: D) -> Result - where - D: Deserializer<'de>, - { - let value = String::deserialize(deserializer)?; - Self::new(value).map_err(de::Error::custom) - } - } - }; -} - -opaque_id!(RuntimeIncarnation, "runtime incarnation"); -opaque_id!(OwnerClaim, "owner claim"); -opaque_id!(BindingId, "binding id"); -opaque_id!(RegistrationToken, "registration token"); - -#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] -#[serde(rename_all = "camelCase", deny_unknown_fields)] -pub struct RuntimeOwner { - incarnation: RuntimeIncarnation, - claim: OwnerClaim, -} - -impl RuntimeOwner { - pub fn new(incarnation: RuntimeIncarnation, claim: OwnerClaim) -> Self { - Self { incarnation, claim } - } - - pub fn incarnation(&self) -> &RuntimeIncarnation { - &self.incarnation - } - - pub fn claim(&self) -> &OwnerClaim { - &self.claim - } -} - -#[derive(Clone, PartialEq, Eq)] -pub struct SnapshotBytes(Vec); - -impl fmt::Debug for SnapshotBytes { - fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result { - formatter - .debug_struct("SnapshotBytes") - .field("len", &self.0.len()) - .finish() - } -} - -impl SnapshotBytes { - pub fn new(bytes: Vec) -> Result { - if bytes.len() > MAX_SNAPSHOT_BYTES { - return Err(SnapshotSizeError { actual: bytes.len() }); - } - Ok(Self(bytes)) - } - - pub fn as_slice(&self) -> &[u8] { - &self.0 - } - - pub fn into_vec(self) -> Vec { - self.0 - } -} - -#[derive(Debug, Clone, Copy, PartialEq, Eq)] -pub struct SnapshotSizeError { - pub actual: usize, -} - -impl fmt::Display for SnapshotSizeError { - fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result { - write!( - formatter, - "decoded snapshot is {} bytes; maximum is {MAX_SNAPSHOT_BYTES}", - self.actual - ) - } -} - -impl std::error::Error for SnapshotSizeError {} - -impl Serialize for SnapshotBytes { - fn serialize(&self, serializer: S) -> Result - where - S: Serializer, - { - serializer.serialize_str(&encode_base64(&self.0)) - } -} - -impl<'de> Deserialize<'de> for SnapshotBytes { - fn deserialize(deserializer: D) -> Result - where - D: Deserializer<'de>, - { - struct SnapshotBytesVisitor; - - impl Visitor<'_> for SnapshotBytesVisitor { - type Value = SnapshotBytes; - - fn expecting(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result { - formatter.write_str("an RFC 4648 padded base64 snapshot") - } - - fn visit_str(self, value: &str) -> Result - where - E: de::Error, - { - let bytes = decode_base64(value).map_err(E::custom)?; - SnapshotBytes::new(bytes).map_err(E::custom) - } - } - - deserializer.deserialize_str(SnapshotBytesVisitor) - } -} - -#[derive(Debug, Clone, PartialEq, Eq)] -struct Base64Error(&'static str); - -impl fmt::Display for Base64Error { - fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result { - formatter.write_str(self.0) - } -} - -fn encode_base64(bytes: &[u8]) -> String { - const ALPHABET: &[u8; 64] = - b"ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz0123456789+/"; - let mut encoded = String::with_capacity(bytes.len().div_ceil(3) * 4); - for chunk in bytes.chunks(3) { - let first = chunk[0]; - let second = chunk.get(1).copied().unwrap_or(0); - let third = chunk.get(2).copied().unwrap_or(0); - encoded.push(ALPHABET[(first >> 2) as usize] as char); - encoded.push(ALPHABET[(((first & 0x03) << 4) | (second >> 4)) as usize] as char); - if chunk.len() > 1 { - encoded.push(ALPHABET[(((second & 0x0f) << 2) | (third >> 6)) as usize] as char); - } else { - encoded.push('='); - } - if chunk.len() > 2 { - encoded.push(ALPHABET[(third & 0x3f) as usize] as char); - } else { - encoded.push('='); - } - } - encoded -} - -fn decode_base64(encoded: &str) -> Result, Base64Error> { - if encoded.len() % 4 != 0 { - return Err(Base64Error("base64 length is not a multiple of four")); - } - let maximum_encoded = MAX_SNAPSHOT_BYTES.div_ceil(3) * 4; - if encoded.len() > maximum_encoded { - return Err(Base64Error("decoded snapshot exceeds the size limit")); - } - if encoded.is_empty() { - return Ok(Vec::new()); - } - - fn value(byte: u8) -> Result { - match byte { - b'A'..=b'Z' => Ok(byte - b'A'), - b'a'..=b'z' => Ok(byte - b'a' + 26), - b'0'..=b'9' => Ok(byte - b'0' + 52), - b'+' => Ok(62), - b'/' => Ok(63), - _ => Err(Base64Error("base64 contains an invalid character")), - } - } - - let input = encoded.as_bytes(); - let padding = usize::from(input[input.len() - 1] == b'=') - + usize::from(input[input.len() - 2] == b'='); - let decoded_len = input.len() / 4 * 3 - padding; - if decoded_len > MAX_SNAPSHOT_BYTES { - return Err(Base64Error("decoded snapshot exceeds the size limit")); - } - let mut decoded = Vec::with_capacity(decoded_len); - let chunks = input.chunks_exact(4); - let chunk_count = chunks.len(); - for (index, chunk) in chunks.enumerate() { - let last = index + 1 == chunk_count; - let a = value(chunk[0])?; - let b = value(chunk[1])?; - decoded.push((a << 2) | (b >> 4)); - match (chunk[2], chunk[3]) { - (b'=', b'=') if last => { - if b & 0x0f != 0 { - return Err(Base64Error("base64 has non-canonical trailing bits")); - } - } - (third, b'=') if last => { - let c = value(third)?; - if c & 0x03 != 0 { - return Err(Base64Error("base64 has non-canonical trailing bits")); - } - decoded.push((b << 4) | (c >> 2)); - } - (b'=', _) => return Err(Base64Error("base64 padding is misplaced")), - (third, fourth) => { - let c = value(third)?; - let d = value(fourth)?; - decoded.push((b << 4) | (c >> 2)); - decoded.push((c << 6) | d); - } - } - } - Ok(decoded) -} - -#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)] -#[serde( - tag = "type", - rename_all = "camelCase", - rename_all_fields = "camelCase", - deny_unknown_fields -)] -pub enum HostMessage { - Register { - owner: RuntimeOwner, - binding_id: BindingId, - registration: RegistrationToken, - uri: String, - selector: Value, - carrier_path: PathBuf, - #[serde(skip_serializing_if = "Option::is_none")] - previous_digest: Option, - }, - Unregister { - owner: RuntimeOwner, - binding_id: BindingId, - registration: RegistrationToken, - }, -} - -#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)] -#[serde(rename_all = "camelCase")] -pub enum RuntimeHealthState { - Starting, - Ready, - Degraded, - Failed, -} - -#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)] -#[serde( - tag = "type", - rename_all = "camelCase", - rename_all_fields = "camelCase", - deny_unknown_fields -)] -pub enum RuntimeMessage { - Publish { - owner: RuntimeOwner, - binding_id: BindingId, - registration: RegistrationToken, - schema_id: String, - media_type: String, - bytes: SnapshotBytes, - topics: Vec, - #[serde(skip_serializing_if = "Option::is_none")] - observed_at: Option, - }, - Health { - owner: RuntimeOwner, - #[serde(skip_serializing_if = "Option::is_none")] - binding_id: Option, - #[serde(skip_serializing_if = "Option::is_none")] - registration: Option, - state: RuntimeHealthState, - #[serde(skip_serializing_if = "Option::is_none")] - detail: Option, - }, -} - -#[derive(Debug)] -pub enum ProtocolError { - MissingNewline, - MultipleLines, - EmptyLine, - LineTooLarge { actual: usize }, - SelectorTooLarge { actual: usize }, - HealthDetailTooLarge { actual: usize }, - InvalidTopics(&'static str), - InvalidHealthScope, - Json(serde_json::Error), -} - -impl fmt::Display for ProtocolError { - fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result { - match self { - Self::MissingNewline => formatter.write_str("protocol frame is missing its newline"), - Self::MultipleLines => formatter.write_str("protocol frame contains multiple lines"), - Self::EmptyLine => formatter.write_str("protocol frame is empty"), - Self::LineTooLarge { actual } => write!( - formatter, - "protocol line is {actual} bytes; maximum is {MAX_PROTOCOL_LINE_BYTES}" - ), - Self::SelectorTooLarge { actual } => write!( - formatter, - "selector is {actual} bytes; maximum is {MAX_SELECTOR_BYTES}" - ), - Self::HealthDetailTooLarge { actual } => write!( - formatter, - "health detail is {actual} bytes; maximum is {MAX_HEALTH_DETAIL_BYTES}" - ), - Self::InvalidTopics(reason) => write!(formatter, "invalid topics: {reason}"), - Self::InvalidHealthScope => formatter.write_str( - "binding-scoped health must carry both bindingId and registration", - ), - Self::Json(error) => write!(formatter, "invalid protocol JSON: {error}"), - } - } -} - -impl std::error::Error for ProtocolError { - fn source(&self) -> Option<&(dyn std::error::Error + 'static)> { - match self { - Self::Json(error) => Some(error), - _ => None, - } - } -} - -pub fn decode_host_line(line: &[u8]) -> Result { - let payload = protocol_payload(line)?; - let message = serde_json::from_slice(payload).map_err(ProtocolError::Json)?; - validate_host_message(&message)?; - Ok(message) -} - -pub fn decode_runtime_line(line: &[u8]) -> Result { - let payload = protocol_payload(line)?; - let message = serde_json::from_slice(payload).map_err(ProtocolError::Json)?; - validate_runtime_message(&message)?; - Ok(message) -} - -pub fn encode_host_line(message: &HostMessage) -> Result, ProtocolError> { - validate_host_message(message)?; - encode_protocol_line(message) -} - -pub fn encode_runtime_line(message: &RuntimeMessage) -> Result, ProtocolError> { - validate_runtime_message(message)?; - encode_protocol_line(message) -} - -fn protocol_payload(line: &[u8]) -> Result<&[u8], ProtocolError> { - if line.len() > MAX_PROTOCOL_LINE_BYTES { - return Err(ProtocolError::LineTooLarge { actual: line.len() }); - } - let Some(payload) = line.strip_suffix(b"\n") else { - return Err(ProtocolError::MissingNewline); - }; - if payload.is_empty() { - return Err(ProtocolError::EmptyLine); - } - if payload.contains(&b'\n') || payload.contains(&b'\r') { - return Err(ProtocolError::MultipleLines); - } - Ok(payload) -} - -fn encode_protocol_line(message: &impl Serialize) -> Result, ProtocolError> { - let mut line = serde_json::to_vec(message).map_err(ProtocolError::Json)?; - line.push(b'\n'); - if line.len() > MAX_PROTOCOL_LINE_BYTES { - return Err(ProtocolError::LineTooLarge { actual: line.len() }); - } - Ok(line) -} - -fn validate_host_message(message: &HostMessage) -> Result<(), ProtocolError> { - if let HostMessage::Register { selector, .. } = message { - let actual = serde_json::to_vec(selector) - .map_err(ProtocolError::Json)? - .len(); - if actual > MAX_SELECTOR_BYTES { - return Err(ProtocolError::SelectorTooLarge { actual }); - } - } - Ok(()) -} - -fn validate_runtime_message(message: &RuntimeMessage) -> Result<(), ProtocolError> { - match message { - RuntimeMessage::Publish { topics, .. } => validate_topics(topics), - RuntimeMessage::Health { - binding_id, - registration, - detail, - .. - } => { - if binding_id.is_some() != registration.is_some() { - return Err(ProtocolError::InvalidHealthScope); - } - if let Some(detail) = detail - && detail.len() > MAX_HEALTH_DETAIL_BYTES - { - return Err(ProtocolError::HealthDetailTooLarge { - actual: detail.len(), - }); - } - Ok(()) - } - } -} - -fn validate_topics(topics: &[String]) -> Result<(), ProtocolError> { - let mut unique = BTreeSet::new(); - for topic in topics { - if topic.is_empty() { - return Err(ProtocolError::InvalidTopics("topic names must not be empty")); - } - if !unique.insert(topic.as_str()) { - return Err(ProtocolError::InvalidTopics("topic names must be unique")); - } - } - Ok(()) -} - -#[derive(Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Hash)] -pub struct SnapshotDigest([u8; 32]); - -impl SnapshotDigest { - pub fn of(bytes: &[u8]) -> Self { - Self(Sha256::digest(bytes).into()) - } - - pub fn as_bytes(&self) -> &[u8; 32] { - &self.0 - } -} - -impl fmt::Debug for SnapshotDigest { - fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result { - fmt::Display::fmt(self, formatter) - } -} - -impl fmt::Display for SnapshotDigest { - fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result { - for byte in self.0 { - write!(formatter, "{byte:02x}")?; - } - Ok(()) - } -} - -impl Serialize for SnapshotDigest { - fn serialize(&self, serializer: S) -> Result - where - S: Serializer, - { - serializer.collect_str(self) - } -} - -impl<'de> Deserialize<'de> for SnapshotDigest { - fn deserialize(deserializer: D) -> Result - where - D: Deserializer<'de>, - { - struct DigestVisitor; - - impl Visitor<'_> for DigestVisitor { - type Value = SnapshotDigest; - - fn expecting(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result { - formatter.write_str("a lowercase 64-character SHA-256 digest") - } - - fn visit_str(self, value: &str) -> Result - where - E: de::Error, - { - if value.len() != 64 || value.bytes().any(|byte| !byte.is_ascii_hexdigit()) { - return Err(E::custom("invalid SHA-256 digest")); - } - if value.bytes().any(|byte| byte.is_ascii_uppercase()) { - return Err(E::custom("SHA-256 digest must use lowercase hex")); - } - let mut digest = [0_u8; 32]; - for (index, pair) in value.as_bytes().chunks_exact(2).enumerate() { - let pair = std::str::from_utf8(pair).map_err(E::custom)?; - digest[index] = u8::from_str_radix(pair, 16).map_err(E::custom)?; - } - Ok(SnapshotDigest(digest)) - } - } - - deserializer.deserialize_str(DigestVisitor) - } -} - #[derive(Debug, Clone, PartialEq, Eq)] pub struct TopicSelection { topics: BTreeSet, @@ -913,6 +370,7 @@ impl RuntimeLifecycle { media_type, bytes, topics, + facts, observed_at, } => { let binding = self.require_registration(owner, binding_id, registration)?; @@ -935,6 +393,7 @@ impl RuntimeLifecycle { target: &binding.target, bytes, selected_topics: binding.contract.selection.select(topics), + facts: facts.as_deref().unwrap_or_default(), observed_at: observed_at.as_deref(), })) } @@ -1042,6 +501,7 @@ pub struct AcceptedPublication<'a> { target: &'a SnapshotTarget, bytes: &'a SnapshotBytes, selected_topics: Vec, + facts: &'a [ResourceFact], observed_at: Option<&'a str>, } @@ -1054,12 +514,21 @@ impl<'a> AcceptedPublication<'a> { &self.selected_topics } + pub fn facts(&self) -> &[ResourceFact] { + self.facts + } + pub fn observed_at(&self) -> Option<&str> { self.observed_at } fn prepare(self) -> Result, PublicationError> { - prepare_snapshot(self.target, self.bytes.as_slice(), self.selected_topics) + prepare_snapshot( + self.target, + self.bytes.as_slice(), + self.selected_topics, + self.facts.to_vec(), + ) } } @@ -1096,6 +565,7 @@ pub struct PublicationOutcome { digest: SnapshotDigest, change: SnapshotChange, selected_topics: Vec, + facts: Vec, } impl PublicationOutcome { @@ -1111,6 +581,10 @@ impl PublicationOutcome { &self.selected_topics } + pub fn facts(&self) -> &[ResourceFact] { + &self.facts + } + pub fn invalidating(&self) -> bool { self.change != SnapshotChange::Equal && !self.selected_topics.is_empty() } @@ -1182,6 +656,7 @@ fn prepare_snapshot<'a>( target: &'a SnapshotTarget, bytes: &'a [u8], selected_topics: Vec, + facts: Vec, ) -> Result, PublicationError> { if bytes.len() > MAX_SNAPSHOT_BYTES { return Err(PublicationError::SnapshotTooLarge { actual: bytes.len() }); @@ -1202,6 +677,7 @@ fn prepare_snapshot<'a>( digest, change, selected_topics, + facts, }, }) } @@ -1210,8 +686,9 @@ fn publish_snapshot( target: &SnapshotTarget, bytes: &[u8], selected_topics: Vec, + facts: Vec, ) -> Result { - let prepared = prepare_snapshot(target, bytes, selected_topics)?; + let prepared = prepare_snapshot(target, bytes, selected_topics, facts)?; prepared.commit()?; Ok(prepared.outcome) } @@ -1224,6 +701,8 @@ pub struct CatchUpState { pending_relevant_change: bool, #[serde(default)] pending_selected_topics: Vec, + #[serde(default)] + pending_facts: Vec, deliverable: bool, } @@ -1244,6 +723,10 @@ impl CatchUpState { &self.pending_selected_topics } + pub fn pending_facts(&self) -> &[ResourceFact] { + &self.pending_facts + } + pub fn deliverable(&self) -> bool { self.deliverable } @@ -1260,6 +743,12 @@ impl CatchUpState { )); } validate_persisted_topics(&self.pending_selected_topics)?; + validate_persisted_facts(&self.pending_facts)?; + if !self.pending_relevant_change && !self.pending_facts.is_empty() { + return Err(CatchUpError::InvalidState( + "pending facts require a pending relevant change", + )); + } Ok(()) } } @@ -1268,6 +757,7 @@ impl CatchUpState { pub struct DeliveryRequest { digest: SnapshotDigest, selected_topics: Vec, + facts: Vec, } impl DeliveryRequest { @@ -1278,6 +768,10 @@ impl DeliveryRequest { pub fn selected_topics(&self) -> &[String] { &self.selected_topics } + + pub fn facts(&self) -> &[ResourceFact] { + &self.facts + } } #[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] @@ -1285,6 +779,8 @@ impl DeliveryRequest { struct PublicationIntent { digest: SnapshotDigest, selected_topics: Vec, + #[serde(default)] + facts: Vec, } impl PublicationIntent { @@ -1292,11 +788,13 @@ impl PublicationIntent { Self { digest: outcome.digest, selected_topics: outcome.selected_topics.clone(), + facts: outcome.facts.clone(), } } fn validate(&self) -> Result<(), CatchUpError> { - validate_persisted_topics(&self.selected_topics) + validate_persisted_topics(&self.selected_topics)?; + validate_persisted_facts(&self.facts) } } @@ -1311,6 +809,13 @@ fn validate_persisted_topics(topics: &[String]) -> Result<(), CatchUpError> { } Ok(()) } +fn validate_persisted_facts(facts: &[ResourceFact]) -> Result<(), CatchUpError> { + if facts.len() > MAX_FACTS || facts.iter().any(|fact| fact.validate().is_err()) { + return Err(CatchUpError::InvalidState("persisted facts are invalid")); + } + Ok(()) +} + #[derive(Debug)] pub struct CatchUp { @@ -1395,6 +900,7 @@ impl CatchUp { if !intent.selected_topics.is_empty() { next.pending_relevant_change = true; next.pending_selected_topics = intent.selected_topics.clone(); + next.pending_facts = intent.facts.clone(); } } Some(_) | None => { @@ -1435,6 +941,7 @@ impl CatchUp { Some(DeliveryRequest { digest: self.state.current_snapshot_digest?, selected_topics: self.state.pending_selected_topics.clone(), + facts: self.state.pending_facts.clone(), }) } @@ -1451,6 +958,7 @@ impl CatchUp { next.last_delivered_digest = Some(digest); next.pending_relevant_change = false; next.pending_selected_topics.clear(); + next.pending_facts.clear(); self.commit(next)?; Ok(true) } @@ -1464,6 +972,7 @@ impl CatchUp { if outcome.invalidating() { next.pending_relevant_change = true; next.pending_selected_topics = outcome.selected_topics.clone(); + next.pending_facts = outcome.facts.clone(); } self.commit(next)?; Ok(self.pending_delivery()) @@ -1861,6 +1370,7 @@ mod tests { media_type: "application/json".to_owned(), bytes: SnapshotBytes::new(bytes.to_vec()).unwrap(), topics: topics.iter().map(|topic| (*topic).to_owned()).collect(), + facts: None, observed_at: None, } } @@ -1875,37 +1385,6 @@ mod tests { } } - #[test] - fn protocol_round_trips_padded_base64_and_rejects_malformed_and_oversized_lines() { - let message = publication(owner("one"), "registration", b"one byte?", &["selected"]); - let encoded = encode_runtime_line(&message).unwrap(); - assert!(encoded.ends_with(b"\n")); - assert_eq!(decode_runtime_line(&encoded).unwrap(), message); - - assert!(matches!( - decode_runtime_line(b"{\"type\":\"publish\"}\n"), - Err(ProtocolError::Json(_)) - )); - let oversized = vec![b'x'; MAX_PROTOCOL_LINE_BYTES + 1]; - assert!(matches!( - decode_runtime_line(&oversized), - Err(ProtocolError::LineTooLarge { .. }) - )); - } - - #[test] - fn protocol_rejects_oversized_decoded_snapshot_before_publication() { - let encoded = encode_base64(&vec![0_u8; MAX_SNAPSHOT_BYTES + 1]); - let line = format!( - "{{\"type\":\"publish\",\"owner\":{{\"incarnation\":\"i\",\"claim\":\"c\"}},\"bindingId\":\"b\",\"registration\":\"r\",\"schemaId\":\"s\",\"mediaType\":\"m\",\"bytes\":\"{encoded}\",\"topics\":[]}}\n" - ); - assert!(line.len() < MAX_PROTOCOL_LINE_BYTES); - assert!(matches!( - decode_runtime_line(line.as_bytes()), - Err(ProtocolError::Json(_)) - )); - assert!(SnapshotBytes::new(vec![0_u8; MAX_SNAPSHOT_BYTES + 1]).is_err()); - } #[test] fn stale_owner_and_registration_are_fenced() { @@ -2017,7 +1496,8 @@ mod tests { let target = SnapshotTarget::new(&root, "resources/github-pr/owner/repo/389.json").unwrap(); assert_eq!(target.current_digest().unwrap(), None); - let outcome = publish_snapshot(&target, b"bytes", vec!["selected".to_owned()]).unwrap(); + let outcome = + publish_snapshot(&target, b"bytes", vec!["selected".to_owned()], Vec::new()).unwrap(); assert_eq!(outcome.change(), SnapshotChange::First); assert_eq!( @@ -2067,6 +1547,7 @@ mod tests { digest: SnapshotDigest::of(b"relevant"), change: SnapshotChange::First, selected_topics: vec!["selected".to_owned()], + facts: vec![ResourceFact::current("state", "ready").unwrap()], }; let irrelevant = PublicationOutcome { digest: SnapshotDigest::of(b"later but irrelevant"), @@ -2074,6 +1555,7 @@ mod tests { previous: relevant.digest(), }, selected_topics: Vec::new(), + facts: Vec::new(), }; let mut catch_up = CatchUp::open(&state_directory).unwrap(); assert_eq!(catch_up.record_publication(&relevant).unwrap(), None); @@ -2083,6 +1565,7 @@ mod tests { let request = catch_up.set_deliverable(true).unwrap().unwrap(); assert_eq!(request.digest(), irrelevant.digest()); assert_eq!(request.selected_topics(), ["selected"]); + assert_eq!(request.facts(), relevant.facts()); assert!(!catch_up.acknowledge_delivery(relevant.digest()).unwrap()); assert!(catch_up.state().pending_relevant_change()); assert!(catch_up.acknowledge_delivery(irrelevant.digest()).unwrap()); @@ -2101,6 +1584,7 @@ mod tests { digest: SnapshotDigest::of(b"snapshot"), change: SnapshotChange::First, selected_topics: vec!["selected".to_owned()], + facts: vec![ResourceFact::current("state", "ready").unwrap()], }; { let mut catch_up = CatchUp::open(&state_directory).unwrap(); @@ -2122,6 +1606,36 @@ mod tests { catch_up.state().pending_selected_topics(), ["selected"] ); + assert_eq!(catch_up.state().pending_facts(), outcome.facts()); + } + + #[test] + fn catch_up_persists_a_maximal_worst_case_escaped_fact_envelope() { + let directory = tempfile::tempdir().unwrap(); + let state_directory = fs::canonicalize(directory.path()).unwrap(); + let facts = (0..MAX_FACTS) + .map(|_| { + ResourceFact::new( + "\"".repeat(MAX_FACT_KEY_BYTES), + FactValue::value("\"".repeat(MAX_FACT_VALUE_BYTES)), + FactValue::value("\\".repeat(MAX_FACT_VALUE_BYTES)), + ) + .unwrap() + }) + .collect::>(); + let outcome = PublicationOutcome { + digest: SnapshotDigest::of(b"snapshot"), + change: SnapshotChange::First, + selected_topics: vec!["selected".to_owned()], + facts, + }; + + { + let mut catch_up = CatchUp::open(&state_directory).unwrap(); + catch_up.record_publication(&outcome).unwrap(); + } + let catch_up = CatchUp::open(&state_directory).unwrap(); + assert_eq!(catch_up.state().pending_facts(), outcome.facts()); } #[test] @@ -2182,7 +1696,7 @@ mod tests { symlink(outside.path(), root.join("linked-parent")).unwrap(); let target = SnapshotTarget::new(&root, "linked-parent/snapshot.json").unwrap(); assert!(matches!( - publish_snapshot(&target, b"bytes", vec!["selected".to_owned()]), + publish_snapshot(&target, b"bytes", vec!["selected".to_owned()], Vec::new()), Err(PublicationError::Io(_)) )); diff --git a/src/resource_profile_supervisor.rs b/src/resource_profile_supervisor.rs index 043d4b7b..84beb250 100644 --- a/src/resource_profile_supervisor.rs +++ b/src/resource_profile_supervisor.rs @@ -24,7 +24,7 @@ use sha2::{Digest as _, Sha256}; use crate::catalog::CatalogConfig; use crate::resource_profile::{ AcceptedOutput, BindingId, BindingRegistration, CatchUp, HostMessage, OwnerClaim, - PublicationContract, RegistrationToken, RuntimeHealthState, RuntimeIncarnation, + PublicationContract, RegistrationToken, ResourceFact, RuntimeHealthState, RuntimeIncarnation, RuntimeLifecycle, RuntimeMessage, RuntimeOwner, SnapshotDigest, SnapshotTarget, TopicSelection, MAX_PROTOCOL_LINE_BYTES, decode_runtime_line, encode_host_line, }; @@ -982,20 +982,28 @@ fn emit_pending_at( binding: &'a str, snapshot_digest: String, topics: &'a [String], + facts: &'a [ResourceFact], } let digest = delivery.digest(); let topics = delivery.selected_topics().to_vec(); + let facts = delivery.facts(); let body = serde_json::to_string(&Body { binding: &active.desired.binding_name, snapshot_digest: digest.to_string(), topics: &topics, + facts, })?; let event_id = publication_event_id( &active.desired.recipient, &active.desired.binding_name, digest, ); - let subject = format!("resource {} changed", active.desired.binding_name); + let subject = resource_change_subject( + &active.desired.binding_name, + facts, + &topics, + "snapshot updated", + ); crate::event::emit_builtin_resync( catalog_root, this_host, @@ -1014,6 +1022,71 @@ fn publication_event_id(recipient: &str, binding: &str, digest: SnapshotDigest) hash_text(&format!("resource-profile\0{recipient}\0{binding}\0{digest}")) } +pub(crate) fn resource_change_subject( + binding: &str, + facts: &[ResourceFact], + topics: &[String], + fallback: &str, +) -> String { + const SUBJECT_MAX_SCALARS: usize = 96; + const MAX_RENDERED_FACTS: usize = 3; + + let topic_suffix = if topics.is_empty() { + String::new() + } else { + format!(" [{}]", topics.join(", ")) + }; + let base = format!("{binding} · "); + let mut rendered = Vec::new(); + for fact in facts.iter().take(MAX_RENDERED_FACTS) { + let fact = render_fact(fact); + let candidate = format!("{base}{}{topic_suffix}", { + let mut candidate_facts = rendered.clone(); + candidate_facts.push(fact.clone()); + candidate_facts.join("; ") + }); + if candidate.chars().count() > SUBJECT_MAX_SCALARS { + break; + } + rendered.push(fact); + } + + let detail = if rendered.is_empty() { + fallback.to_owned() + } else { + rendered.join("; ") + }; + let subject = format!("{base}{detail}{topic_suffix}"); + if subject.chars().count() <= SUBJECT_MAX_SCALARS { + return subject; + } + + // Facts and topic names are never clipped. A pathological oversized binding or topic suffix + // falls back to the binding and bounded generic detail; the durable body remains complete. + let fallback = format!("{binding} · {fallback}"); + if fallback.chars().count() <= SUBJECT_MAX_SCALARS { + fallback + } else { + fallback.chars().take(SUBJECT_MAX_SCALARS).collect() + } +} + +fn render_fact(fact: &ResourceFact) -> String { + match (fact.before(), fact.after()) { + (None, Some(Some(after))) => format!("{}={after}", fact.key()), + (None, Some(None)) => format!("{}=removed", fact.key()), + (Some(None), Some(Some(after))) => format!("{}=+{after}", fact.key()), + (Some(Some(before)), Some(None)) => format!("{}=-{before}", fact.key()), + (Some(Some(before)), Some(Some(after))) => { + format!("{}={before}→{after}", fact.key()) + } + (Some(None), Some(None)) => format!("{}=absent", fact.key()), + (Some(Some(before)), None) => format!("{} was {before}", fact.key()), + (Some(None), None) => format!("{} was absent", fact.key()), + (None, None) => unreachable!("validated facts always carry a value"), + } +} + fn binding_state_directory(desired: &DesiredBinding) -> anyhow::Result { let state = lexical_absolute(&crate::run::state_root())?; Ok(state @@ -1130,6 +1203,49 @@ mod tests { assert!(!owner_matches(None, &old)); } + #[test] + fn publication_subject_renders_ordered_facts_and_reserves_topics() { + let facts = vec![ + ResourceFact::current("state", "ready").unwrap(), + ResourceFact::transition("label", None::, Some("bug")).unwrap(), + ResourceFact::transition("owner", Some("alice"), Some("bob")).unwrap(), + ResourceFact::current("fourth", "omitted").unwrap(), + ]; + assert_eq!( + resource_change_subject( + "review", + &facts, + &["ci.failure".to_owned()], + "snapshot updated" + ), + "review · state=ready; label=+bug; owner=alice→bob [ci.failure]" + ); + } + + #[test] + fn publication_subject_drops_low_priority_facts_atomically_within_96_scalars() { + let facts = vec![ + ResourceFact::current("priority", "x".repeat(80)).unwrap(), + ResourceFact::current("lower", "must-not-leapfrog").unwrap(), + ]; + let subject = resource_change_subject( + "review", + &facts, + &["selected.topic".to_owned()], + "snapshot updated", + ); + assert_eq!(subject, "review · snapshot updated [selected.topic]"); + assert!(subject.chars().count() <= 96); + } + + #[test] + fn publication_subject_without_facts_has_a_useful_compatible_fallback() { + assert_eq!( + resource_change_subject("review", &[], &[], "snapshot updated"), + "review · snapshot updated" + ); + } + #[test] fn stale_protocol_failure_does_not_remove_the_replacement_process() { let old = RuntimeOwner::new( diff --git a/src/resync.rs b/src/resync.rs index 4e01854e..3f5707e0 100644 --- a/src/resync.rs +++ b/src/resync.rs @@ -16,11 +16,14 @@ use std::thread::JoinHandle; use std::time::{Duration, Instant}; use notify::Watcher as _; +use serde::Serialize; use sha2::{Digest as _, Sha256}; use agent_spec::profile::{ ProfileClass, ResourceProfileRefresh, ResourceProfileRegistry, }; -use agent_spec::spec::{AgentSpec, decode_percent_path}; +use agent_spec::spec::{AgentSpec, Resource, decode_percent_path}; + +use crate::resource_profile::{MAX_FACTS, MAX_FACT_KEY_BYTES, ResourceFact}; /// The reserved stream used only by the supervisor's crate-internal resync publisher. pub const RESYNC_STREAM: &str = "resync"; @@ -85,6 +88,57 @@ pub struct WatchableCarrier { pub class: CarrierClass, pub containment_root: Option, } +#[derive(Debug, Clone, PartialEq, Eq)] +struct DeclarationSummary { + bindings: BTreeMap, + complete: bool, +} + +fn declaration_summary(spec: &AgentSpec) -> DeclarationSummary { + let mut bindings = spec + .resources + .iter() + .map(|resource| { + ( + resource.name().to_owned(), + declaration_resource_digest(resource), + ) + }) + .collect::>(); + bindings.sort_by(|left, right| left.0.cmp(&right.0)); + let complete = bindings.len() <= MAX_FACTS + && bindings + .iter() + .all(|(label, _)| label.len() <= MAX_FACT_KEY_BYTES); + DeclarationSummary { + bindings: bindings.into_iter().take(MAX_FACTS).collect(), + complete, + } +} + +fn declaration_resource_digest(resource: &Resource) -> String { + let mut digest = Sha256::new(); + digest.update(b"st2.resync.resource-declaration.v1\0"); + update_digest_field(&mut digest, resource.uri().as_bytes()); + update_digest_field(&mut digest, resource.reason().as_bytes()); + match resource.inactive_reason() { + Some(reason) => { + digest.update([1]); + update_digest_field(&mut digest, reason.as_bytes()); + } + None => digest.update([0]), + } + let selector = serde_json::to_vec(&resource.selector()) + .expect("a parsed JSON selector always serializes"); + update_digest_field(&mut digest, &selector); + format!("{:x}", digest.finalize()) +} + +fn update_digest_field(digest: &mut Sha256, value: &[u8]) { + let length = u64::try_from(value.len()).expect("declaration fields fit in u64"); + digest.update(length.to_be_bytes()); + digest.update(value); +} /// The watchable carriers of one agent, keyed by its declaration path with current routing IDs. #[derive(Debug, Clone, PartialEq, Eq)] @@ -93,6 +147,7 @@ pub struct AgentWatchSet { pub bus_id: String, pub seat_id: Option, pub carriers: Vec, + declaration_summary: Option, } /// Resolve one spec's watch set: the declaration file plus every active resource binding whose @@ -207,6 +262,7 @@ fn resolve_watch_set( .unwrap_or_else(|| format!("{}.{}", spec.bus_id(this_host), task.name)) }), carriers, + declaration_summary: Some(declaration_summary(spec)), }, diagnostics, ) @@ -660,13 +716,72 @@ enum CarrierState { } impl CarrierState { - fn render(&self) -> &str { + fn fact_value(&self) -> String { match self { - Self::Present(digest) => digest, - Self::Missing => "missing", + Self::Present(digest) => digest.chars().take(12).collect(), + Self::Missing => "missing".to_owned(), } } } +fn digest_transition_fact(old: &CarrierState, new: &CarrierState) -> Vec { + vec![ + ResourceFact::transition("digest", Some(old.fact_value()), Some(new.fact_value())) + .expect("short carrier digests are valid facts"), + ] +} + +fn declaration_transition_facts( + old: Option<&DeclarationSummary>, + new: Option<&DeclarationSummary>, + old_state: &CarrierState, + new_state: &CarrierState, +) -> Vec { + let (Some(old), Some(new)) = (old, new) else { + return digest_transition_fact(old_state, new_state); + }; + if !old.complete || !new.complete { + return digest_transition_fact(old_state, new_state); + } + + let labels = old + .bindings + .keys() + .chain(new.bindings.keys()) + .cloned() + .collect::>(); + let facts = labels + .into_iter() + .filter_map(|label| match (old.bindings.get(&label), new.bindings.get(&label)) { + (None, Some(_)) => Some(ResourceFact::transition( + label, + None::, + Some("declared".to_owned()), + )), + (Some(_), None) => Some(ResourceFact::transition( + label, + Some("declared".to_owned()), + None::, + )), + (Some(before), Some(after)) if before != after => { + Some(ResourceFact::current(label, "changed")) + } + _ => None, + }) + .collect::, _>>(); + match facts { + Ok(facts) if !facts.is_empty() => facts, + Ok(_) | Err(_) => digest_transition_fact(old_state, new_state), + } +} + +fn current_declaration_summary(root: &Path, path: &Path) -> Option { + crate::discover_strict(root) + .specs + .into_iter() + .find(|spec| lexical_clean(&spec.path) == path) + .map(|spec| declaration_summary(&spec)) +} + #[derive(Debug, Clone, PartialEq, Eq)] struct PendingTransition { @@ -674,8 +789,11 @@ struct PendingTransition { path: PathBuf, old_state: CarrierState, new_state: CarrierState, + facts: Vec, + topics: Vec, body: String, event_id: String, + new_declaration_summary: Option, } impl PendingTransition { @@ -686,17 +804,73 @@ impl PendingTransition { new_state: &CarrierState, incarnation: crate::event::StreamOwnerIncarnation, sequence: u64, + ) -> Self { + Self::capture( + binding, + path, + old_state, + new_state, + digest_transition_fact(old_state, new_state), + vec!["content".to_owned()], + None, + incarnation, + sequence, + ) + } + + fn declaration( + path: &Path, + old_state: &CarrierState, + new_state: &CarrierState, + old_summary: Option<&DeclarationSummary>, + new_summary: Option, + incarnation: crate::event::StreamOwnerIncarnation, + sequence: u64, + ) -> Self { + let facts = declaration_transition_facts( + old_summary, + new_summary.as_ref(), + old_state, + new_state, + ); + Self::capture( + "declaration", + path, + old_state, + new_state, + facts, + vec!["declaration".to_owned()], + new_summary, + incarnation, + sequence, + ) + } + + #[allow(clippy::too_many_arguments)] + fn capture( + binding: &str, + path: &Path, + old_state: &CarrierState, + new_state: &CarrierState, + facts: Vec, + topics: Vec, + new_declaration_summary: Option, + incarnation: crate::event::StreamOwnerIncarnation, + sequence: u64, ) -> Self { let occurrence = incarnation.occurrence_token(sequence); - let body = render_body(binding, path, old_state, new_state, &occurrence); + let body = render_body(binding, &topics, &facts, &occurrence); let event_id = transition_identity(&body); Self { binding: binding.to_owned(), path: path.to_path_buf(), old_state: old_state.clone(), new_state: new_state.clone(), + facts, + topics, body, event_id, + new_declaration_summary, } } } @@ -708,6 +882,7 @@ struct Entry { class: CarrierClass, containment_root: Option, state: Option, + declaration_summary: Option, /// Last occurrence sequence reserved by this retained subscription. Sequence zero is the /// silent seeded state; only capturing a new immutable transition advances it. occurrence_sequence: u64, @@ -843,6 +1018,7 @@ fn rebuild_carriers( ) -> BTreeMap> { let mut next: BTreeMap> = BTreeMap::new(); for set in refresh.sets { + let seeded_declaration_summary = set.declaration_summary.clone(); for carrier in set.carriers { // The canonical recipient and binding label identify one subscription across // declaration and carrier relocation. Rebuild every retained entry from the current @@ -852,32 +1028,37 @@ fn rebuild_carriers( // recipient-scoped namespace. let identity = (set.bus_id.clone(), carrier.label.clone()); let retained = take_retained_entry(&mut previous, &set.bus_id, &carrier.label); - let (state, occurrence_sequence, pending_transition, dirty) = retained.map_or_else( - || { - let (state, dirty) = - match read_state(&carrier.path, carrier.containment_root.as_deref()) { - Ok(state) => (Some(state), false), - Err(error) => { - diagnose_read_error(&carrier.path, &error); - (None, true) - } - }; - ( - state, - subscription_sequences.get(&identity).copied().unwrap_or(0), - None, - dirty, - ) - }, - |entry| { - ( - entry.state, - entry.occurrence_sequence, - entry.pending_transition, - entry.dirty, - ) - }, - ); + let (state, declaration_summary, occurrence_sequence, pending_transition, dirty) = + retained.map_or_else( + || { + let (state, dirty) = + match read_state(&carrier.path, carrier.containment_root.as_deref()) { + Ok(state) => (Some(state), false), + Err(error) => { + diagnose_read_error(&carrier.path, &error); + (None, true) + } + }; + ( + state, + (carrier.label == "declaration") + .then(|| seeded_declaration_summary.clone()) + .flatten(), + subscription_sequences.get(&identity).copied().unwrap_or(0), + None, + dirty, + ) + }, + |entry| { + ( + entry.state, + entry.declaration_summary, + entry.occurrence_sequence, + entry.pending_transition, + entry.dirty, + ) + }, + ); let entry = Entry { bus_id: set.bus_id.clone(), seat_id: set.seat_id.clone(), @@ -885,6 +1066,7 @@ fn rebuild_carriers( class: carrier.class, containment_root: carrier.containment_root.clone(), state, + declaration_summary, occurrence_sequence, pending_transition, dirty, @@ -1235,6 +1417,11 @@ impl Worker { fn flush_path(&mut self, path: &Path, due_class: Option) { let occurrence_incarnation = crate::event::current_stream_owner_incarnation(&self.root, &self.this_host).ok(); + let observed_declaration_summary = self + .carriers + .get(path) + .is_some_and(|entries| entries.iter().any(|entry| entry.label == "declaration")) + .then(|| current_declaration_summary(&self.root, path)); let Some(entries) = self.carriers.get_mut(path) else { return; }; @@ -1253,6 +1440,9 @@ impl Worker { .take() .expect("pending transition was just observed"); entry.state = Some(completed.new_state); + if completed.binding == "declaration" { + entry.declaration_summary = completed.new_declaration_summary; + } // The current carrier may have advanced or rebound while the immutable // transition was pending. Complete it first, then schedule current state. match observed { @@ -1305,17 +1495,32 @@ impl Worker { retries.push(entry.class); continue; }; - let transition = PendingTransition::new( - &entry.label, - path, - old_state, - &target_state, - incarnation, - sequence, - ); + let transition = if entry.label == "declaration" { + PendingTransition::declaration( + path, + old_state, + &target_state, + entry.declaration_summary.as_ref(), + observed_declaration_summary.clone().flatten(), + incarnation, + sequence, + ) + } else { + PendingTransition::new( + &entry.label, + path, + old_state, + &target_state, + incarnation, + sequence, + ) + }; entry.occurrence_sequence = sequence; if emit_resync(&self.root, &self.this_host, &entry.bus_id, &transition) { entry.state = Some(target_state); + if transition.binding == "declaration" { + entry.declaration_summary = transition.new_declaration_summary; + } } else { entry.pending_transition = Some(transition); entry.dirty = true; @@ -1346,7 +1551,12 @@ fn emit_resync( bus_id: &str, transition: &PendingTransition, ) -> bool { - let subject = format!("resource {} changed", transition.binding); + let subject = crate::resource_profile_supervisor::resource_change_subject( + &transition.binding, + &transition.facts, + &transition.topics, + "content changed", + ); match crate::event::emit_builtin_resync( root, this_host, @@ -1368,6 +1578,7 @@ fn emit_resync( } } + fn read_state(path: &Path, containment_root: Option<&Path>) -> std::io::Result { match containment_root { Some(root) => read_confined(path, root), @@ -1539,19 +1750,28 @@ fn read_confined(_path: &Path, _root: &Path) -> std::io::Result { )) } +#[derive(Serialize)] +#[serde(rename_all = "camelCase")] +struct ResyncBody<'a> { + binding: &'a str, + topics: &'a [String], + facts: &'a [ResourceFact], + occurrence: &'a str, +} + fn render_body( - label: &str, - path: &Path, - old: &CarrierState, - new: &CarrierState, + binding: &str, + topics: &[String], + facts: &[ResourceFact], occurrence: &str, ) -> String { - format!( - "resource `{label}` changed\n\nbinding: {label}\npath: {}\nold: {}\nnew: {}\noccurrence: {occurrence}\n", - path.display(), - old.render(), - new.render(), - ) + serde_json::to_string(&ResyncBody { + binding, + topics, + facts, + occurrence, + }) + .expect("validated resync facts always serialize") } #[cfg(test)] mod tests { @@ -1565,6 +1785,28 @@ mod tests { } } + #[test] + fn resync_subject_uses_the_shared_three_fact_and_96_scalar_renderer() { + let facts = vec![ + ResourceFact::transition("alpha", None::, Some("declared")).unwrap(), + ResourceFact::current("beta", "changed").unwrap(), + ResourceFact::transition("charlie", Some("declared"), None::).unwrap(), + ResourceFact::current("delta", "omitted").unwrap(), + ]; + let subject = crate::resource_profile_supervisor::resource_change_subject( + "declaration", + &facts, + &["declaration".to_owned()], + "content changed", + ); + assert_eq!( + subject, + "declaration · alpha=+declared; beta=changed; charlie=-declared [declaration]" + ); + assert!(subject.chars().count() <= 96); + assert!(!subject.contains("delta")); + } + fn owner_incarnation(seed: u64) -> crate::event::StreamOwnerIncarnation { crate::event::StreamOwnerIncarnation::for_test(seed, seed + 1, 42, seed + 2) } @@ -1592,13 +1834,135 @@ mod tests { .collect() } + fn event_body(event: &str) -> serde_json::Value { + let body = event + .lines() + .rev() + .find(|line| line.starts_with('{')) + .expect("JSON resync body"); + serde_json::from_str(body).expect("valid JSON resync body") + } + fn event_field(event: &str, field: &str) -> String { - event + if let Some(value) = event .lines() .find_map(|line| line.strip_prefix(&format!("{field}: "))) + { + return value.to_owned(); + } + let body = event_body(event); + if matches!(field, "old" | "new") { + let digest = body["facts"] + .as_array() + .and_then(|facts| facts.iter().find(|fact| fact["key"] == "digest")) + .expect("digest transition fact"); + let value = if field == "old" { + &digest["before"] + } else { + &digest["after"] + }; + return value.as_str().expect("digest fact value").to_owned(); + } + body[field] + .as_str() .unwrap_or_else(|| panic!("missing {field} in event")) .to_owned() } + #[test] + fn declaration_facts_are_ordered_added_removed_and_semantically_changed_labels() { + let root = tempfile::tempdir().unwrap(); + let declaration = root.path().join("agent.kdl"); + std::fs::write( + &declaration, + r#"agent "worker" { + host "host" + command "true" + resource "inactive" uri="file:///inactive" reason="kept" + resource "reason" uri="file:///reason" reason="before" + resource "removed" uri="file:///removed" reason="gone" + resource "uri" uri="file:///before" reason="same" +}"#, + ) + .unwrap(); + let before = declaration_summary(&discover(root.path())); + std::fs::write( + &declaration, + r#"agent "worker" { + host "host" + command "true" + resource "added" uri="file:///added" reason="new" + resource "inactive" uri="file:///inactive" reason="kept" inactive-reason="paused" + resource "reason" uri="file:///reason" reason="after" + resource "uri" uri="file:///after" reason="same" +}"#, + ) + .unwrap(); + let after = declaration_summary(&discover(root.path())); + let facts = declaration_transition_facts( + Some(&before), + Some(&after), + &CarrierState::Present("before-digest".to_owned()), + &CarrierState::Present("after-digest".to_owned()), + ); + assert_eq!( + facts.iter().map(ResourceFact::key).collect::>(), + vec!["added", "inactive", "reason", "removed", "uri"] + ); + assert_eq!(facts[0].before(), Some(None)); + assert_eq!(facts[0].after(), Some(Some("declared"))); + for index in [1, 2, 4] { + assert_eq!(facts[index].before(), None); + assert_eq!(facts[index].after(), Some(Some("changed"))); + } + assert_eq!(facts[3].before(), Some(Some("declared"))); + assert_eq!(facts[3].after(), Some(None)); + } + + #[test] + fn declaration_parse_failure_retains_a_digest_fact_for_later_delivery() { + let root = tempfile::tempdir().unwrap(); + let agent_dir = root.path().join("agents/host/worker"); + std::fs::create_dir_all(&agent_dir).unwrap(); + let declaration = agent_dir.join("agent.kdl"); + let valid = r#"agent "worker" { + host "host" + command "true" + resource "goal" uri="resources/goal.md" reason="Mission." +}"#; + std::fs::write(&declaration, valid).unwrap(); + crate::event::publish_owner_binding_for_test(root.path(), "host").unwrap(); + let set = watch_set_for(&discover(root.path()), "host", &Default::default()); + let mut worker = Worker { + root: root.path().to_path_buf(), + this_host: "host".to_owned(), + carriers: BTreeMap::new(), + subscription_sequences: BTreeMap::new(), + deadlines: BTreeMap::new(), + watched: BTreeMap::new(), + watcher: None, + }; + worker.apply_watch_sets(refresh_for(vec![set])); + std::fs::write(&declaration, "not an Agent Spec").unwrap(); + worker.flush_path(&declaration, None); + let pending = worker.carriers[&declaration][0] + .pending_transition + .as_ref() + .expect("malformed declaration keeps its digest fallback"); + assert_eq!(pending.facts.len(), 1); + assert_eq!(pending.facts[0].key(), "digest"); + assert_eq!(pending.topics, ["declaration"]); + assert_eq!(event_body(&pending.body)["facts"].as_array().unwrap().len(), 1); + std::fs::write(&declaration, valid).unwrap(); + worker.flush_path(&declaration, None); + let delivered = resync_inbox_event(&agent_dir); + let body = event_body(&delivered); + assert_eq!(body["binding"], "declaration"); + assert_eq!(body["topics"], serde_json::json!(["declaration"])); + assert_eq!(body["facts"][0]["key"], "digest"); + assert!(!delivered.contains("file:///")); + assert!(!delivered.contains("Mission.")); + } + #[test] fn watch_set_covers_declaration_and_local_bindings_only() { @@ -1735,6 +2099,7 @@ mod tests { class: CarrierClass::Immediate, containment_root: None, state: Some(CarrierState::Present("alpha-before".to_owned())), + declaration_summary: None, occurrence_sequence: 4, pending_transition: None, dirty: true, @@ -1746,6 +2111,7 @@ mod tests { class: CarrierClass::Coalesced, containment_root: None, state: Some(CarrierState::Present("beta-before".to_owned())), + declaration_summary: None, occurrence_sequence: 9, dirty: true, pending_transition: None, @@ -1763,6 +2129,7 @@ mod tests { class: CarrierClass::Immediate, containment_root: None, }], + declaration_summary: None, }, AgentWatchSet { declaration_path: PathBuf::from("/catalog/beta/agent.kdl"), @@ -1774,6 +2141,7 @@ mod tests { class: CarrierClass::Coalesced, containment_root: None, }], + declaration_summary: None, }, ]; @@ -1830,6 +2198,7 @@ mod tests { class: CarrierClass::Coalesced, containment_root: None, state: Some(CarrierState::Present("old-digest".to_owned())), + declaration_summary: None, occurrence_sequence: 3, pending_transition: None, dirty: true, @@ -1899,6 +2268,7 @@ mod tests { class: CarrierClass::Immediate, containment_root: None, state: Some(CarrierState::Present("old-digest".to_owned())), + declaration_summary: None, occurrence_sequence: 0, pending_transition: None, dirty: true, @@ -1937,11 +2307,11 @@ mod tests { .expect("the retry routes through the refreshed recipient"); assert!(event.contains(&format!("event-id: {}", pending.event_id)), "{event}"); assert!(event.contains(&pending.body), "{event}"); - assert!(pending.body.contains(&format!("path: {}", old_path.display()))); + let pending_body: serde_json::Value = serde_json::from_str(&pending.body).unwrap(); + assert_eq!(pending_body["binding"], "goal"); + assert_eq!(pending_body["facts"][0]["before"], "old-digest"); assert!( - !pending - .body - .contains(&format!("path: {}", current_path.display())), + !pending.body.contains(¤t_path.display().to_string()), "rebinding must not rewrite bytes reserved under the pending event identity" ); let entry = &worker.carriers[¤t_path][0]; @@ -1975,6 +2345,7 @@ mod tests { class: CarrierClass::Immediate, containment_root: None, state: read_state(&old_path, None).ok(), + declaration_summary: None, occurrence_sequence: 0, pending_transition: None, dirty: false, @@ -1996,6 +2367,7 @@ mod tests { class: CarrierClass::Immediate, containment_root: None, }], + declaration_summary: None, }])); assert!(!worker.carriers.contains_key(&old_path)); @@ -2025,6 +2397,7 @@ mod tests { class: CarrierClass::Immediate, containment_root: None, state: read_state(&carrier, None).ok(), + declaration_summary: None, occurrence_sequence: 0, pending_transition: None, dirty: true, @@ -2046,6 +2419,7 @@ mod tests { class, containment_root: None, }], + declaration_summary: None, }]) }; @@ -2079,6 +2453,7 @@ mod tests { class: CarrierClass::Immediate, containment_root: None, state: Some(CarrierState::Present("before".to_owned())), + declaration_summary: None, occurrence_sequence: 1, pending_transition: Some(PendingTransition::new( "declaration", @@ -2163,6 +2538,7 @@ mod tests { class: CarrierClass::Immediate, containment_root: None, state: Some(CarrierState::Present("old-digest".to_owned())), + declaration_summary: None, occurrence_sequence: 1, pending_transition: Some(PendingTransition::new( "goal", @@ -2193,8 +2569,8 @@ mod tests { .map(|entry| std::fs::read_to_string(entry.unwrap().path()).unwrap()) .find(|body| body.contains("stream: resync")) .expect("pending transition is replayed"); - assert!(event.contains("old: old-digest"), "{event}"); - assert!(event.contains("new: pending-target"), "{event}"); + assert_eq!(event_field(&event, "old"), "old-digest"); + assert_eq!(event_field(&event, "new"), "pending-targ"); let entry = &worker.carriers[&carrier][0]; assert_eq!( entry.state, @@ -2226,6 +2602,7 @@ mod tests { class: CarrierClass::Coalesced, containment_root: None, state: baseline.clone(), + declaration_summary: None, occurrence_sequence: 0, pending_transition: None, dirty: false, @@ -2248,6 +2625,7 @@ mod tests { class: CarrierClass::Coalesced, containment_root: None, }], + declaration_summary: None, }])); worker.flush_due(now + IMMEDIATE_WINDOW + Duration::from_secs(1)); @@ -2312,7 +2690,7 @@ mod tests { .collect::>(); assert_eq!(events.len(), 1); assert!(events[0].contains("stream: resync")); - assert!(events[0].contains("binding: goal")); + assert!(events[0].contains(r#""binding":"goal""#)); } #[test] @@ -2433,6 +2811,7 @@ mod tests { class, containment_root: None, state: state.clone(), + declaration_summary: None, occurrence_sequence: 0, pending_transition: None, dirty: true, @@ -2548,7 +2927,10 @@ mod tests { worker.flush_path(&carrier, None); let deletion = resync_inbox_events(&agent_dir); assert_eq!(deletion.len(), 1); - assert_eq!(event_field(&deletion[0], "old"), original_digest); + assert_eq!( + event_field(&deletion[0], "old"), + original_digest.chars().take(12).collect::() + ); assert_eq!(event_field(&deletion[0], "new"), "missing"); assert!(event_field(&deletion[0], "occurrence").ends_with(":1")); assert_eq!( @@ -2864,6 +3246,7 @@ mod tests { class: CarrierClass::Immediate, containment_root: None, state: Some(CarrierState::Present("old-digest".to_owned())), + declaration_summary: None, occurrence_sequence: 0, pending_transition: None, dirty: true, @@ -2932,6 +3315,7 @@ mod tests { class: CarrierClass::Immediate, containment_root: None, state: Some(CarrierState::Present("old-digest".to_owned())), + declaration_summary: None, occurrence_sequence: 0, pending_transition: None, dirty: true, @@ -2950,8 +3334,8 @@ mod tests { .clone() .expect("failed transition snapshot is retained"); assert_eq!(pending_transition.new_state, CarrierState::Missing); - assert!(pending_transition.body.contains("old: old-digest")); - assert!(pending_transition.body.contains("new: missing")); + assert_eq!(event_field(&pending_transition.body, "old"), "old-digest"); + assert_eq!(event_field(&pending_transition.body, "new"), "missing"); assert_eq!(worker.carriers[&carrier][0].occurrence_sequence, 1); std::fs::write(&carrier, "old bytes").unwrap(); worker.flush_path(&carrier, None); @@ -2972,91 +3356,34 @@ mod tests { #[test] fn transition_identity_covers_every_rendered_transition_dimension() { - let old = CarrierState::Present("old-digest".to_owned()); - let new = CarrierState::Present("new-digest".to_owned()); - let baseline = render_body( - "goal", - Path::new("/agent/goal.md"), - &old, - &new, - "v1:1:2:42:3:1", - ); + let topics = vec!["content".to_owned()]; + let facts = + vec![ResourceFact::transition("digest", Some("old"), Some("new")).unwrap()]; + let baseline = render_body("goal", &topics, &facts, "v1:1:2:42:3:1"); assert_eq!( transition_identity(&baseline), transition_identity(&baseline), "replaying one canonical body must reproduce its identity" ); + let changed_facts = + vec![ResourceFact::transition("digest", Some("old"), Some("other")).unwrap()]; for (dimension, changed) in [ ( "binding", - render_body( - "spec", - Path::new("/agent/goal.md"), - &old, - &new, - "v1:1:2:42:3:1", - ), - ), - ( - "path", - render_body( - "goal", - Path::new("/other/goal.md"), - &old, - &new, - "v1:1:2:42:3:1", - ), + render_body("spec", &topics, &facts, "v1:1:2:42:3:1"), ), ( - "old state", - render_body( - "goal", - Path::new("/agent/goal.md"), - &CarrierState::Present("other-old".to_owned()), - &new, - "v1:1:2:42:3:1", - ), + "topic", + render_body("goal", &["other".to_owned()], &facts, "v1:1:2:42:3:1"), ), ( - "missing old state", - render_body( - "goal", - Path::new("/agent/goal.md"), - &CarrierState::Missing, - &new, - "v1:1:2:42:3:1", - ), - ), - ( - "new state", - render_body( - "goal", - Path::new("/agent/goal.md"), - &old, - &CarrierState::Present("other-new".to_owned()), - "v1:1:2:42:3:1", - ), - ), - ( - "missing new state", - render_body( - "goal", - Path::new("/agent/goal.md"), - &old, - &CarrierState::Missing, - "v1:1:2:42:3:1", - ), + "fact", + render_body("goal", &topics, &changed_facts, "v1:1:2:42:3:1"), ), ( "occurrence", - render_body( - "goal", - Path::new("/agent/goal.md"), - &old, - &new, - "v1:1:2:42:3:2", - ), + render_body("goal", &topics, &facts, "v1:1:2:42:3:2"), ), ] { assert_ne!( diff --git a/src/run.rs b/src/run.rs index 7f1fcd6e..241a4a88 100644 --- a/src/run.rs +++ b/src/run.rs @@ -4477,7 +4477,7 @@ mod tests { std::fs::write(&live_goal, "changed while compile failed\n").unwrap(); let first_event = wait_for_resync_event(&live_dir) .expect("the already-live valid seat must stay watched across the compile error"); - assert!(first_event.contains("binding: goal"), "{first_event}"); + assert!(first_event.contains(r#""binding":"goal""#), "{first_event}"); std::fs::write(&broken_goal, "invalid seat changed\n").unwrap(); std::thread::sleep(Duration::from_millis(750)); @@ -4520,7 +4520,7 @@ mod tests { ); let corrected_event = wait_for_resync_event_change(&live_dir, &first_event) .expect("correcting another declaration must not reseed and hide the live transition"); - assert!(corrected_event.contains("binding: goal"), "{corrected_event}"); + assert!(corrected_event.contains(r#""binding":"goal""#), "{corrected_event}"); } #[test] @@ -4589,7 +4589,7 @@ mod tests { std::fs::write(&dormant_goal, "unwatched while materialization failed\n").unwrap(); let first_event = wait_for_resync_event(&live_dir) .expect("the observed live seat must remain watched through materialization failure"); - assert!(first_event.contains("binding: goal"), "{first_event}"); + assert!(first_event.contains(r#""binding":"goal""#), "{first_event}"); std::thread::sleep(Duration::from_millis(750)); assert!( current_resync_event(&dormant_dir).is_none(), @@ -4618,7 +4618,7 @@ mod tests { let recovered_event = wait_for_resync_event_change(&live_dir, &first_event) .expect("recovery must preserve the pending transition instead of silently reseeding"); assert!( - recovered_event.contains("binding: goal"), + recovered_event.contains(r#""binding":"goal""#), "{recovered_event}" ); } @@ -4771,7 +4771,7 @@ mod tests { let event = wait_for_resync_event(&first_dir) .expect("the first seat must observe a carrier transition during the later launch"); - assert!(event.contains("binding: goal"), "{event}"); + assert!(event.contains(r#""binding":"goal""#), "{event}"); } #[test] @@ -4890,7 +4890,7 @@ mod tests { std::fs::write(&goal, "changed after replacement launch\n").unwrap(); let event = wait_for_resync_event(&agent_dir) .expect("the successful replacement must receive a fresh silent baseline"); - assert!(event.contains("binding: goal"), "{event}"); + assert!(event.contains(r#""binding":"goal""#), "{event}"); } #[test] @@ -4963,7 +4963,7 @@ mod tests { ); let event = wait_for_resync_event(&agent_dir) .expect("the companion launch and final refresh must preserve the canonical baseline"); - assert!(event.contains("binding: goal"), "{event}"); + assert!(event.contains(r#""binding":"goal""#), "{event}"); } #[test] diff --git a/tests/resource_profile_supervisor_e2e.rs b/tests/resource_profile_supervisor_e2e.rs index 1f976854..8a980597 100755 --- a/tests/resource_profile_supervisor_e2e.rs +++ b/tests/resource_profile_supervisor_e2e.rs @@ -9,8 +9,8 @@ use std::path::{Path, PathBuf}; use std::time::{Duration, Instant}; use st2::resource_profile::{ - BindingId, HostMessage, RegistrationToken, RuntimeHealthState, RuntimeMessage, RuntimeOwner, - SnapshotBytes, decode_host_line, encode_runtime_line, + BindingId, HostMessage, RegistrationToken, ResourceFact, RuntimeHealthState, RuntimeMessage, + RuntimeOwner, SnapshotBytes, decode_host_line, encode_runtime_line, }; use st2::resource_profile_supervisor::ResourceProfileSupervisor; @@ -67,6 +67,9 @@ impl RuntimeControl { media_type: MEDIA_TYPE.to_owned(), bytes: SnapshotBytes::new(bytes.to_vec()).unwrap(), topics: topics.iter().map(|topic| (*topic).to_owned()).collect(), + facts: Some(vec![ + ResourceFact::current("revision", health_marker).unwrap(), + ]), observed_at: None, }; let health = RuntimeMessage::Health { @@ -220,6 +223,16 @@ fn observable_publication_reaches_builtin_resync_with_filter_catch_up_and_scope_ 1, "the first selected publication must create one built-in resync record" ); + assert!( + first_inbox[0].contains("subject: observed · revision=primary-first [selected]"), + "{}", + first_inbox[0] + ); + assert!( + first_inbox[0].contains(r#""facts":[{"key":"revision","after":"primary-first"}]"#), + "{}", + first_inbox[0] + ); primary .runtime @@ -274,6 +287,11 @@ fn observable_publication_reaches_builtin_resync_with_filter_catch_up_and_scope_ caught_up_inbox, first_inbox, "restoring delivery must replace the old head with the pending digest" ); + assert!( + caught_up_inbox[0].contains("revision=delivery-unavailable"), + "{}", + caught_up_inbox[0] + ); let caught_up_projection = file_tree(&primary.agent_dir.join("resources")); primary.refresh(&primary_supervisor); assert_eq!( @@ -391,7 +409,7 @@ fn file_tree(root: &Path) -> Vec<(PathBuf, Vec)> { } fn observable_resolver_wasm() -> Vec { - const DESCRIPTOR: &[u8] = br#"{"abiVersion":2,"capabilities":["resolve","read","observe"],"selectorSchema":{"type":"object","properties":{"topics":{"type":"array","items":{"type":"string"},"uniqueItems":true}},"required":["topics"],"additionalProperties":false},"defaultSelector":{"topics":["selected"]},"topics":[{"name":"selected"},{"name":"ignored"}],"runtime":{"topology":"shared"},"snapshot":{"mediaType":"application/json","schemaId":"dev.example.observable.snapshot.v1"}}"#; + const DESCRIPTOR: &[u8] = br#"{"abiVersion":3,"capabilities":["resolve","read","observe"],"selectorSchema":{"type":"object","properties":{"topics":{"type":"array","items":{"type":"string"},"uniqueItems":true}},"required":["topics"],"additionalProperties":false},"defaultSelector":{"topics":["selected"]},"topics":[{"name":"selected"},{"name":"ignored"}],"runtime":{"topology":"shared"},"snapshot":{"mediaType":"application/json","schemaId":"dev.example.observable.snapshot.v1"}}"#; const RESOLUTION: &[u8] = br#"{"path":"resources/snapshot.json","class":"observable"}"#; const DESCRIPTOR_PTR: i64 = 1024; const RESOLUTION_PTR: i64 = 4096; diff --git a/tests/resync.rs b/tests/resync.rs index 6cde6bb7..81a27e3d 100644 --- a/tests/resync.rs +++ b/tests/resync.rs @@ -95,8 +95,8 @@ fn carrier_change_emits_one_superseded_resync_event_and_silent_stores_stay_quiet "goal change must produce exactly one resync event within the immediate window" ); let body = &resync_events(&agent_dir)[0]; - assert!(body.contains("resource goal changed"), "{body}"); - assert!(body.contains("binding: goal"), "{body}"); + assert!(body.contains("subject: goal · digest="), "{body}"); + assert!(body.contains(r#""binding":"goal""#), "{body}"); let legitimate_event_id = body .lines() .find(|line| line.starts_with("event-id:")) @@ -153,7 +153,7 @@ fn carrier_change_emits_one_superseded_resync_event_and_silent_stores_stay_quiet || { resync_events(&agent_dir) .iter() - .filter(|b| !b.contains(&first_event_id) && b.contains("binding: goal")) + .filter(|b| !b.contains(&first_event_id) && b.contains(r#""binding":"goal""#)) .count() }, 1 @@ -189,7 +189,7 @@ fn whole_file_declaration_replacement_by_rename_notifies_immediately() { || { resync_events(&agent_dir) .iter() - .filter(|b| b.contains("binding: declaration")) + .filter(|b| b.contains(r#""binding":"declaration""#)) .count() }, 1 @@ -283,7 +283,7 @@ fn declared_wasm_profile_resolves_a_scheme_uri_goal_binding_and_fires_on_change( resync_events(&agent_dir) ); let body = &resync_events(&agent_dir)[0]; - assert!(body.contains("resource goal changed"), "{body}"); + assert!(body.contains("subject: goal · digest="), "{body}"); } #[test]