diff --git a/crates/eureka-state/tests/client.rs b/crates/eureka-state/tests/client.rs index 38550b9..e474276 100644 --- a/crates/eureka-state/tests/client.rs +++ b/crates/eureka-state/tests/client.rs @@ -218,7 +218,7 @@ fn client_persona_round_trip_against_a_live_daemon() { assert_eq!(view.value, persona(INSTANCE, T1)); assert_eq!((view.updated_at.as_str(), view.receipt_id.as_str()), (T1, receipt_id.as_str())); let HuginnMindResponse::Whoami(status) = client.call(HuginnMindRequest::Whoami).unwrap() else { panic!("a status") }; - assert_eq!(status.persona.as_deref(), Some(T1)); + assert_eq!(status.persona, huginn_mind::wire::PersonaStatus::Stored { updated_at: T1.into() }); // A stale put is an answer. assert_eq!( diff --git a/crates/huginn-daemon/src/daemon.rs b/crates/huginn-daemon/src/daemon.rs index 34a54c7..afe0185 100644 --- a/crates/huginn-daemon/src/daemon.rs +++ b/crates/huginn-daemon/src/daemon.rs @@ -590,7 +590,7 @@ pub(crate) mod tests { documents: 1, receipts: 1, index: IndexStatus::Current, - persona: None, + persona: huginn_mind::wire::PersonaStatus::Absent, } ); assert_eq!(daemon.runtime_id(), "huginn-yggdrasil"); @@ -704,7 +704,7 @@ pub(crate) mod tests { expected_updated_at: None, }; assert_eq!(daemon.handle(put, now()).answered(), HuginnMindResponse::Refused(foreign)); - assert_eq!(status(&mut daemon).persona, None); + assert_eq!(status(&mut daemon).persona, huginn_mind::wire::PersonaStatus::Absent); } /// A declared instance outside `Slug`'s grammar is refused by its own diff --git a/crates/huginn-mind/Cargo.toml b/crates/huginn-mind/Cargo.toml index bcbfde6..48e61ad 100644 --- a/crates/huginn-mind/Cargo.toml +++ b/crates/huginn-mind/Cargo.toml @@ -26,6 +26,7 @@ rmp-serde = "1" schemars = "1" serde = { version = "1", features = ["derive"] } serde_json = "1" +serde_path_to_error = "0.1" sha2 = "0.10" [dev-dependencies] diff --git a/crates/huginn-mind/src/envelope.rs b/crates/huginn-mind/src/envelope.rs index a314e37..4ac9544 100644 --- a/crates/huginn-mind/src/envelope.rs +++ b/crates/huginn-mind/src/envelope.rs @@ -70,28 +70,50 @@ pub fn decode_request(message: &CultNetMessage) -> Result<(String, HuginnMindReq )); }; if service_id != MIND_SERVICE_ID { - return Err(OperationFailure::new("wrong-service", format!("{service_id} is not {MIND_SERVICE_ID}"))); + return Err(OperationFailure::new("wrong-service", format!("the service is not {MIND_SERVICE_ID}"))); } if payload_schema != MIND_REQUEST_SCHEMA { return Err(OperationFailure::new( "wrong-payload-schema", - format!("{payload_schema} is not {MIND_REQUEST_SCHEMA}"), + format!("the payload schema is not {MIND_REQUEST_SCHEMA}"), )); } let bytes = STANDARD .decode(payload) - .map_err(|error| OperationFailure::new("payload-not-base64", error.to_string()))?; - let request: HuginnMindRequest = rmp_serde::from_slice(&bytes) - .map_err(|error| OperationFailure::new("payload-not-a-request", error.to_string()))?; + .map_err(|_| OperationFailure::new("payload-not-base64", "the payload is not standard base64"))?; + let request = read_request(&bytes)?; if operation != request.operation() { return Err(OperationFailure::new( "operation-mismatch", - format!("the envelope says {operation} and the payload is {}", request.operation()), + format!("the envelope's operation is not the payload's {}", request.operation()), )); } Ok((message_id.clone(), request)) } +/// The payload as a request, or a failure naming the field path and the class +/// of error and nothing the caller sent: a decoder's own text quotes the +/// offending value (an unknown variant, an integer), so no decoder text +/// leaves this function. +fn read_request(bytes: &[u8]) -> Result { + use rmp_serde::decode::Error as Decode; + let mut decoder = rmp_serde::Deserializer::new(bytes); + serde_path_to_error::deserialize(&mut decoder).map_err(|error| { + let class = match error.inner() { + Decode::InvalidMarkerRead(_) | Decode::InvalidDataRead(_) => "truncated", + Decode::TypeMismatch(_) => "wrong-type", + Decode::OutOfRange => "out-of-range", + Decode::LengthMismatch(_) => "wrong-length", + Decode::Uncategorized(_) | Decode::Syntax(_) => "invalid-value", + Decode::Utf8Error(_) => "invalid-text", + Decode::DepthLimitExceeded => "too-deep", + }; + let path = error.path().to_string(); + let at = if path == "." { "the payload".to_string() } else { format!("field {path}") }; + OperationFailure::new("payload-not-a-request", format!("{at}: {class}")) + }) +} + /// One answer, whose status the response derives. pub fn encode_response( message_id: &str, @@ -184,7 +206,7 @@ mod tests { documents: 0, receipts: 0, index: crate::wire::IndexStatus::Current, - persona: None, + persona: crate::wire::PersonaStatus::Absent, }), "accepted", ), @@ -210,6 +232,82 @@ mod tests { assert_eq!(decode_response(&message).unwrap(), ("m-2".into(), Err(failure))); } + fn request_message(service_id: &str, operation: &str, schema: &str, payload: String) -> CultNetMessage { + CultNetMessage::OperationRequest { + message_id: "m-9".into(), + service_id: service_id.into(), + operation: operation.into(), + payload_schema: schema.into(), + payload_encoding: "messagepack-base64".into(), + payload, + source_runtime_id: None, + target_runtime_id: None, + } + } + + /// The payload of `request` with the value at `path` replaced, encoded the + /// way a client does. + fn payload_with(request: &HuginnMindRequest, path: &[&str], bad: serde_json::Value) -> String { + let mut value = serde_json::to_value(request).unwrap(); + let mut at = &mut value; + for key in path { + at = at.get_mut(*key).unwrap_or_else(|| panic!("no {key} in the request")); + } + *at = bad; + STANDARD.encode(rmp_serde::to_vec_named(&value).unwrap()) + } + + /// A refusal of the decoder names the field and the class of error and + /// nothing the caller sent: not an unknown variant, not an integer, not + /// an envelope field, not a byte of bad base64. Over the persona wire and + /// over the pipeline's, because the decoder is one. + #[test] + fn the_decoder_never_echoes_what_the_caller_sent() { + const CANARY: &str = "CANARY-5c0de1-do-not-echo"; + const NUMBER: i64 = 90817263; + let put = HuginnMindRequest::PersonaPut { + instance: slug(INSTANCE), + provenance: provenance(Faculty::Hands), + state: serde_json::json!({}), + expected_updated_at: None, + }; + let admit = HuginnMindRequest::Admit(PipelineAdmissionBatch { + instance: slug(INSTANCE), + provenance: provenance(Faculty::Hands), + documents: vec![instance(INSTANCE)], + }); + let cases = [ + (&put, vec!["PersonaPut", "provenance", "faculty"], serde_json::json!(CANARY), "provenance.faculty"), + (&put, vec!["PersonaPut", "provenance", "agent"], serde_json::json!(NUMBER), "provenance.agent"), + (&put, vec!["PersonaPut", "expected_updated_at"], serde_json::json!(NUMBER), "expected_updated_at"), + (&put, vec!["PersonaPut", "instance"], serde_json::json!(NUMBER), "instance"), + (&admit, vec!["Admit", "provenance", "faculty"], serde_json::json!(CANARY), "provenance.faculty"), + (&admit, vec!["Admit", "instance"], serde_json::json!(NUMBER), "instance"), + ]; + for (request, path, bad, field) in cases { + let message = request_message(MIND_SERVICE_ID, request.operation(), MIND_REQUEST_SCHEMA, payload_with(request, &path, bad)); + let failure = decode_request(&message).unwrap_err(); + assert_eq!(failure.code, "payload-not-a-request"); + assert!(failure.message.contains(field), "{field} is named: {}", failure.message); + assert!(failure.message.ends_with(": invalid-value"), "the class is named: {}", failure.message); + for echoed in [CANARY.to_string(), NUMBER.to_string()] { + assert!(!failure.message.contains(&echoed), "{echoed} came back: {}", failure.message); + } + } + + let good = payload_with(&put, &["PersonaPut", "expected_updated_at"], serde_json::Value::Null); + let envelope_cases = [ + request_message(CANARY, "persona_put", MIND_REQUEST_SCHEMA, good.clone()), + request_message(MIND_SERVICE_ID, "persona_put", CANARY, good.clone()), + request_message(MIND_SERVICE_ID, CANARY, MIND_REQUEST_SCHEMA, good), + request_message(MIND_SERVICE_ID, "persona_put", MIND_REQUEST_SCHEMA, format!("{CANARY}!!")), + ]; + for message in envelope_cases { + let failure = decode_request(&message).unwrap_err(); + assert!(!failure.message.contains(CANARY), "{} came back: {}", failure.code, failure.message); + } + } + /// The round trip the client and the daemon share, and the one code that /// `serve::answer` cannot reach because it answers a foreign message with /// `Error` instead. diff --git a/crates/huginn-mind/src/lib.rs b/crates/huginn-mind/src/lib.rs index 0e9dc19..67b76d6 100644 --- a/crates/huginn-mind/src/lib.rs +++ b/crates/huginn-mind/src/lib.rs @@ -55,7 +55,7 @@ pub use refusal::MindRefusal; pub use store::{MindStore, OwnedRedbMessagePackBackingStore}; pub use wire::{ DeferredAnswer, HuginnMindRequest, HuginnMindResponse, IndexStatus, MAX_DEFERRED_BODY_BYTES, MIND_REQUEST_SCHEMA, MIND_REQUEST_SCHEMA_JSON, - MIND_RESPONSE_SCHEMA, MIND_RESPONSE_SCHEMA_JSON, MIND_SERVICE_ID, MindStatus, + MIND_RESPONSE_SCHEMA, MIND_RESPONSE_SCHEMA_JSON, MIND_SERVICE_ID, MindStatus, PersonaStatus, }; /// The leaf, whole, through the one crate that pins its rev: the daemon and diff --git a/crates/huginn-mind/src/persona.rs b/crates/huginn-mind/src/persona.rs index 8e1a7a9..9c47518 100644 --- a/crates/huginn-mind/src/persona.rs +++ b/crates/huginn-mind/src/persona.rs @@ -243,7 +243,7 @@ mod tests { mind } - fn persona_in_status(mind: &Mind) -> Option { + fn persona_in_status(mind: &Mind) -> crate::wire::PersonaStatus { mind.status(crate::wire::IndexStatus::Current).persona } @@ -436,7 +436,7 @@ mod tests { fn a_put_is_a_compare_and_swap_on_updated_at() { let store = MemoryStore::new(); let mut mind = seeded_over(&store); - assert_eq!(persona_in_status(&mind), None); + assert_eq!(persona_in_status(&mind), crate::wire::PersonaStatus::Absent); let receipts = |mind: &Mind| mind.receipts().unwrap().len(); let base = receipts(&mind); @@ -447,7 +447,7 @@ mod tests { let view = mind.persona().unwrap().unwrap(); assert_eq!((view.updated_at.as_str(), view.receipt_id.as_str()), (T1, first.as_str())); assert_eq!(view.value, persona(INSTANCE, T1)); - assert_eq!(persona_in_status(&mind).as_deref(), Some(T1)); + assert_eq!(persona_in_status(&mind), crate::wire::PersonaStatus::Stored { updated_at: T1.into() }); // Two puts that read the same stored document: the second loses. let second = committed(put(&mut mind, persona(INSTANCE, T2), Some(T1))); @@ -521,7 +521,7 @@ mod tests { with_one.push(entry(INSTANCE, persona(INSTANCE, T1))); let mind = Mind::open_with(planted(with_one.clone()), &yggdrasil).unwrap(); assert_eq!(mind.envelopes().len(), 3); - assert_eq!(persona_in_status(&mind).as_deref(), Some(T1)); + assert_eq!(persona_in_status(&mind), crate::wire::PersonaStatus::Stored { updated_at: T1.into() }); // Two are no mind's. with_one.push(entry("second", persona(INSTANCE, T2))); @@ -537,6 +537,33 @@ mod tests { assert!(matches!(Mind::open_with(planted(stranger), &yggdrasil).err(), Some(MindRefusal::ForeignStore { .. }))); } + /// Status answers in the door's own terms: a stored persona the door + /// cannot read is `Unreadable`, not the `Absent` of a mind with none, while + /// `PersonaGet` answers `Unavailable` for the same bytes. + #[test] + fn status_says_unreadable_where_the_door_says_unavailable() { + let yggdrasil = slug(INSTANCE); + let cache = crate::mind::schema_cache().unwrap(); + let readable = cache.prepare_entry_named(INSTANCE, &HuginnPersonaEntry { value: persona(INSTANCE, T1) }).unwrap().0; + let no_stamp = cache.prepare_entry_named(INSTANCE, &HuginnPersonaEntry { value: json!({}) }).unwrap().0; + let mut undecodable = readable.clone(); + undecodable.payload = vec![0xc1]; + for (row, status) in [ + (readable, crate::wire::PersonaStatus::Stored { updated_at: T1.into() }), + (no_stamp, crate::wire::PersonaStatus::Unreadable), + (undecodable, crate::wire::PersonaStatus::Unreadable), + ] { + let store = MemoryStore::new(); + store.plant(epoch()); + store.plant(prepare(&instance(INSTANCE))); + store.plant(row); + let mind = Mind::open_with(store, &yggdrasil).unwrap(); + assert_eq!(persona_in_status(&mind), status); + assert_eq!(mind.persona().is_err(), status == crate::wire::PersonaStatus::Unreadable); + } + assert_eq!(persona_in_status(&seeded_over(&MemoryStore::new())), crate::wire::PersonaStatus::Absent); + } + #[test] fn the_stored_bytes_are_the_one_slot_array_the_catalogue_describes() { let store = MemoryStore::new(); diff --git a/crates/huginn-mind/src/wire.rs b/crates/huginn-mind/src/wire.rs index e68805a..00cbce0 100644 --- a/crates/huginn-mind/src/wire.rs +++ b/crates/huginn-mind/src/wire.rs @@ -192,8 +192,18 @@ pub struct MindStatus { pub documents: u32, pub receipts: u32, pub index: IndexStatus, - /// The stored persona document's `updatedAt`; none until the first put. - pub persona: Option, + /// The persona door as `PersonaGet` would find it. + pub persona: PersonaStatus, +} + +/// What the persona door holds, in the one vocabulary the door itself answers +/// in: nothing yet, a document with this `updatedAt`, or a stored document the +/// door cannot read (`PersonaGet` and every put answer `Unavailable`). +#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize, JsonSchema)] +pub enum PersonaStatus { + Absent, + Stored { updated_at: String }, + Unreadable, } impl Mind { @@ -215,7 +225,11 @@ impl Mind { documents, receipts, index, - persona: self.stored_updated_at().ok().flatten(), + persona: match self.stored_updated_at() { + Ok(None) => PersonaStatus::Absent, + Ok(Some(updated_at)) => PersonaStatus::Stored { updated_at }, + Err(_) => PersonaStatus::Unreadable, + }, } } }