diff --git a/packages/cultmesh-rs/src/rudp_document_server.rs b/packages/cultmesh-rs/src/rudp_document_server.rs index c6825c8e..40dfb767 100644 --- a/packages/cultmesh-rs/src/rudp_document_server.rs +++ b/packages/cultmesh-rs/src/rudp_document_server.rs @@ -3,7 +3,7 @@ use cultnet_rs::{ CultNetErrorCode, CultNetErrorDetails, CultNetMessage, CultNetRawDocumentRecord, CultNetRudpPacket, CultNetRudpPacketType, CultNetRudpSendOptions, CultNetRudpSession, CultNetRudpSessionOptions, CultNetWireContract, decode_cultnet_message_from_slice, decode_rudp_packet, encode_cultnet_message_to_vec, - encode_rudp_packet, is_permanent_send_error, readable_message_id, send_error_code, write_content_hash, + encode_rudp_packet, is_permanent_send_error, readable_message_id, message_id_digest, send_error_code, write_content_hash, }; use std::collections::{BTreeMap, HashMap}; #[cfg(test)] @@ -189,6 +189,15 @@ pub struct CultMeshRudpDocumentServerOptions { pub resend_delay: Duration, pub max_pending_reliable_packets_per_session: usize, pub max_fragment_bytes: usize, + /// Server-wide memory budget for the remembered writes of all receipt sessions + /// (write-receipt.md, "Remembered-write limits"): at most this many bytes, counted as + /// `REMEMBERED_ENTRY_BYTES` per entry. A new write over it is refused `limit_reached`, the session kept. + pub max_remembered_write_bytes: usize, +} + +impl CultMeshRudpDocumentServerOptions { + /// What one remembered write is charged: the digests of its id and content, and its answer. + pub const REMEMBERED_ENTRY_BYTES: usize = 32 + 32 + 1; } impl Default for CultMeshRudpDocumentServerOptions { @@ -204,6 +213,7 @@ impl Default for CultMeshRudpDocumentServerOptions { resend_delay: Duration::from_millis(50), max_pending_reliable_packets_per_session: 1024, max_fragment_bytes: 1200, + max_remembered_write_bytes: 32 * 1024 * 1024, } } } @@ -275,6 +285,10 @@ pub enum CultMeshRudpRejectionReason { /// A receipt-capable session already remembers `MAX_REMEMBERED_WRITES_PER_SESSION` writes and takes no new /// id. Nothing reached the sink. WriteLimitReached, + /// A receipt-capable session sent a message id over `MAX_MESSAGE_ID_BYTES`. Nothing was remembered. + MessageIdTooLong, + /// A receipt-capable session sent a write on an unreliable channel, where no answer could be owed once. + ChannelNotReliable, } impl std::fmt::Display for CultMeshRudpRejectionReason { @@ -285,6 +299,8 @@ impl std::fmt::Display for CultMeshRudpRejectionReason { Self::MessageUnreadable(error) => write!(f, "message could not be read: {error}"), Self::MessageIdReused => f.write_str("the message id was already used for different content"), Self::WriteLimitReached => f.write_str("the session remembers as many writes as it will"), + Self::MessageIdTooLong => f.write_str("the message id is longer than 256 bytes"), + Self::ChannelNotReliable => f.write_str("a receipt session's write arrived on an unreliable channel"), Self::DocumentUnservable { response_bytes, max_snapshot_response_bytes, @@ -325,7 +341,7 @@ impl std::fmt::Display for CultMeshRudpRejectionReason { /// The texts a refused peer is sent, one per rejection variant. They are fixed: /// a sink's or source's error text can quote what a peer sent, so it stays with /// the caller that receives the rejection and never reaches the wire. -pub(crate) const REFUSAL_TEXTS: [&str; 13] = [ +pub(crate) const REFUSAL_TEXTS: [&str; 15] = [ "the catalog refused the document", "the document can never be served", "the snapshot source failed", @@ -339,12 +355,40 @@ pub(crate) const REFUSAL_TEXTS: [&str; 13] = [ "the message could not be read", "the message id was already used for different content", "the session's remembered-write limit is full", + "messageId exceeds 256 bytes", + "receipt sessions require the reliable channel", ]; impl CultMeshRudpRejectionReason { /// The fixed text the refused peer is sent for this variant. fn refusal_text(&self) -> &'static str { - REFUSAL_TEXTS[match self { + REFUSAL_TEXTS[usize::from(self.refusal_index())] + } + + /// The reason a remembered refusal replays as. Its payload (a sink's text, a size) is not remembered: the + /// peer is only ever sent the variant's fixed text, and the caller's rejection is reported once, at first sight. + fn from_refusal_index(index: u8) -> Self { + match index { + 0 => Self::SinkRefused(String::new()), + 1 => Self::DocumentUnservable { response_bytes: 0, max_snapshot_response_bytes: 0, fragment_count: 0, max_fragment_count: 0 }, + 2 => Self::SnapshotSourceFailed(String::new()), + 3 => Self::SnapshotTooManyDocuments { documents: 0, max_snapshot_documents: 0 }, + 4 => Self::SnapshotResponseTooLarge { response_bytes: 0, max_snapshot_response_bytes: 0 }, + 5 => Self::ResponseEncodingFailed(String::new()), + 6 => Self::PayloadBudgetFull, + 7 => Self::ResponseQueueFailed(String::new()), + 8 => Self::ResponseSendFailed(""), + 9 => Self::DeleteNotSupported, + 10 => Self::MessageUnreadable(String::new()), + 11 => Self::MessageIdReused, + 12 => Self::WriteLimitReached, + 13 => Self::MessageIdTooLong, + _ => Self::ChannelNotReliable, + } + } + + fn refusal_index(&self) -> u8 { + match self { Self::SinkRefused(_) => 0, Self::DocumentUnservable { .. } => 1, Self::SnapshotSourceFailed(_) => 2, @@ -358,7 +402,9 @@ impl CultMeshRudpRejectionReason { Self::MessageUnreadable(_) => 10, Self::MessageIdReused => 11, Self::WriteLimitReached => 12, - }] + Self::MessageIdTooLong => 13, + Self::ChannelNotReliable => 14, + } } } @@ -437,6 +483,8 @@ impl CultMeshRudpApplicationRejection { (CultMeshRudpRejectionReason::MessageIdReused, _) => { message_invalid_message("messageId", text) } + (CultMeshRudpRejectionReason::MessageIdTooLong, _) => message_invalid_message("messageId", text), + (CultMeshRudpRejectionReason::ChannelNotReliable, _) => message_invalid_message("channel", text), (CultMeshRudpRejectionReason::WriteLimitReached, _) => CultNetMessage::Error { error: format!("limit_reached: {text}"), code: Some(CultNetErrorCode::LimitReached), @@ -475,6 +523,8 @@ enum SessionEnd { Rejected(CultNetMessage), /// The session owes its peer a packet that can never be sent as built. Unsendable(std::io::Error), + /// The server could not answer a write it owed an answer to. Not a refusal of anything the peer sent. + LocalFailure, } fn write_rejection( @@ -512,9 +562,31 @@ struct SessionEntry { /// Answered writes of a receipt-capable session: message id -> (content hash, the answer given), refusals /// included, so a resend of the same content is answered again without being decided again and the same /// id with other content is refused. - answered_writes: HashMap, + answered_writes: HashMap<[u8; 32], RememberedWrite>, + /// Writes refused `limit_reached`, kept so their resend replays the refusal. A bounded set outside the + /// entry budget (`MAX_REMEMBERED_WRITES_PER_SESSION`); past it a new write is refused unremembered. + limit_refusals: HashMap<[u8; 32], [u8; 32]>, } +/// What the server remembers of a write it answered: digests only, so an entry costs +/// `REMEMBERED_ENTRY_BYTES` whatever the id or payload size. The answer's body is rebuilt from the resend, +/// whose content hash has been checked equal to the original's. +#[derive(Clone, Copy)] +struct RememberedWrite { + content_hash: [u8; 32], + answer: StoredAnswer, +} + +#[derive(Clone, Copy, PartialEq, Eq)] +enum StoredAnswer { + Admitted, + /// A refusal, by its `REFUSAL_TEXTS` index (`CultMeshRudpRejectionReason::refusal_index`). + Refused(u8), +} + +/// The longest `messageId` a receipt session may send, in UTF-8 bytes. +const MAX_MESSAGE_ID_BYTES: usize = 256; + /// The most answered writes a receipt-capable session remembers; a session at the limit refuses new writes /// (`WriteLimitReached`) rather than forget an id it promised to answer idempotently. const WRITE_ADMITTED_VERSION: &str = "cultnet.write_admitted.v0"; @@ -663,7 +735,9 @@ where && packet.channel_id == "schema" && !packet.reliable { - return Ok(CultMeshRudpPollOutcome::Handled); + // A receipt session's write that could be lost without trace cannot be answered once: it is + // refused on the reliable channel, never applied. Anything else unreliable is dropped as before. + return self.refuse_unreliable_write(key, &packet.payload); } let payload_charge = if packet.packet_type == CultNetRudpPacketType::Data { packet.payload.len() @@ -789,7 +863,7 @@ where Answered::Sent => {} Answered::Unsendable(error) => unsendable = Some(error), Answered::Unqueueable => { - self.end_session(key, SessionEnd::Refused)?; + self.end_session(key, SessionEnd::LocalFailure)?; return Ok(CultMeshRudpPollOutcome::ApplicationRejected(rejection)); } } @@ -814,7 +888,7 @@ where // An unanswerable write is answered by ending the session: its peer settles it // as `session_ended`, never as silence. Answered::Unqueueable => { - self.end_session(key, SessionEnd::Refused)?; + self.end_session(key, SessionEnd::LocalFailure)?; return Ok(CultMeshRudpPollOutcome::ApplicationRejected(rejection)); } } @@ -897,6 +971,39 @@ where }) } + /// Refuses a put or delete that arrived on an unreliable channel of a receipt session with + /// `message_invalid` (`details.field` `channel`) naming its id. Nothing is remembered or applied: the id + /// stays free for the same write sent reliably. + fn refuse_unreliable_write( + &mut self, + key: CultMeshRudpSessionKey, + payload: &[u8], + ) -> Result { + if !self.receipts(key) { + return Ok(CultMeshRudpPollOutcome::Handled); + } + let Ok(message) = decode_cultnet_message_from_slice(payload, CultNetWireContract::CultNetSchemaV0) else { + return Ok(CultMeshRudpPollOutcome::Handled); + }; + let (operation, message_id, record_key) = match message { + CultNetMessage::DocumentPutRaw { message_id, document } => { + (CultMeshRudpApplicationOperation::DocumentPutRaw, message_id, document.record_key) + } + CultNetMessage::DocumentDelete { message_id, record_key, .. } => { + (CultMeshRudpApplicationOperation::DocumentDelete, message_id, record_key) + } + _ => return Ok(CultMeshRudpPollOutcome::Handled), + }; + let rejection = write_rejection(key, operation, &message_id, &record_key, CultMeshRudpRejectionReason::ChannelNotReliable); + let refusal = rejection.refusal_message(true); + match self.send_on_live_session(key, &refusal, true)? { + Answered::Sent => {} + Answered::Unsendable(error) => self.end_session(key, SessionEnd::Unsendable(error))?, + Answered::Unqueueable => self.end_session(key, SessionEnd::LocalFailure)?, + } + Ok(CultMeshRudpPollOutcome::ApplicationRejected(rejection)) + } + /// Expire idle sessions and resend reliable packets whose deadline passed. pub fn maintain(&mut self) -> Result { let now = self.clock.now_monotonic_millis(); @@ -1011,6 +1118,7 @@ where receipts: false, capability_settled: false, answered_writes: HashMap::new(), + limit_refusals: HashMap::new(), }, ); accept @@ -1067,6 +1175,9 @@ where schema_id: receipt.document.schema_id.clone(), record_key: receipt.document.record_key.clone(), }; + if message_id.len() > MAX_MESSAGE_ID_BYTES { + return Ok(Some(rejection(CultMeshRudpRejectionReason::MessageIdTooLong))); + } let content_hash = match write_content_hash( "put", &receipt.document.schema_id, @@ -1082,20 +1193,21 @@ where }; // The id is looked up first: a resend is answered with what the session already said, and // the sink is asked only about an id the session has not seen. + let id_digest = message_id_digest(&message_id); if let ControlFlow::Break(outcome) = - self.seen_write(key, now, &message_id, content_hash, &rejection)? + self.seen_write(key, now, id_digest, content_hash, &admitted, &rejection)? { return Ok(outcome); } match self.admit_put(receipt, true) { Ok(()) => { // Admitted: the sink holds it. Only now is a receipt owed, never before the commit point. - self.remember(key, &message_id, content_hash, admitted.clone()); + self.remember(key, id_digest, content_hash, StoredAnswer::Admitted); return self.answer_sent(key, now, &admitted, &rejection); } Err(reason) => { let refused = rejection(reason); - self.remember(key, &message_id, content_hash, refused.refusal_message(true)); + self.remember(key, id_digest, content_hash, StoredAnswer::Refused(refused.reason.refusal_index())); return Ok(Some(refused)); } } @@ -1278,6 +1390,9 @@ where if !self.receipts(key) { return Ok(Some(rejection(CultMeshRudpRejectionReason::DeleteNotSupported))); } + if message_id.len() > MAX_MESSAGE_ID_BYTES { + return Ok(Some(rejection(CultMeshRudpRejectionReason::MessageIdTooLong))); + } let content_hash = match write_content_hash("delete", &schema_id, &record_key, None) { Ok(hash) => hash, Err(error) => { @@ -1286,13 +1401,21 @@ where )))); } }; + let id_digest = message_id_digest(&message_id); + // A delete is never admitted; the body is the receipt it would be answered with. + let would_admit = CultNetMessage::WriteAdmitted { + message_id: message_id.clone(), + operation: "delete".into(), + schema_id: schema_id.clone(), + record_key: record_key.clone(), + }; if let ControlFlow::Break(outcome) = - self.seen_write(key, now, &message_id, content_hash, &rejection)? + self.seen_write(key, now, id_digest, content_hash, &would_admit, &rejection)? { return Ok(outcome); } let refused = rejection(CultMeshRudpRejectionReason::DeleteNotSupported); - self.remember(key, &message_id, content_hash, refused.refusal_message(true)); + self.remember(key, id_digest, content_hash, StoredAnswer::Refused(refused.reason.refusal_index())); return Ok(Some(refused)); } _ => {} @@ -1300,36 +1423,63 @@ where Ok(None) } - /// What a receipt-capable session already knows of `message_id`. `Continue` is a first sighting that - /// the session has room to remember. A repeat with the same content is answered with the stored answer, - /// a repeat with other content (a put or a delete of any record) is `MessageIdReused`, and a session - /// that remembers its limit takes no new id. + /// What a receipt-capable session already knows of a write, by the digest of its id. `Continue` is a first + /// sighting that fits both bounds. A repeat with the same content is answered with its remembered answer + /// (rebuilt from this resend, whose content hash is the original's), a repeat with other content is + /// `MessageIdReused`. A first sighting past the session table or the server-wide budget is refused + /// `limit_reached` for that write alone, and that refusal is remembered (`limit_refusals`) so its resend + /// replays it; no slot is freed, so the id is never admitted later in the session. fn seen_write( &mut self, key: CultMeshRudpSessionKey, now: u64, - message_id: &str, + id_digest: [u8; 32], content_hash: [u8; 32], + admitted: &CultNetMessage, rejection: &impl Fn(CultMeshRudpRejectionReason) -> CultMeshRudpApplicationRejection, ) -> Result>> { - let (prior, remembered) = self - .sessions - .get(&key) - .map(|entry| (entry.answered_writes.get(message_id).cloned(), entry.answered_writes.len())) - .unwrap_or_default(); + let Some(entry) = self.sessions.get(&key) else { + return Ok(ControlFlow::Continue(())); + }; + let prior = entry.answered_writes.get(&id_digest).copied().or_else(|| { + entry.limit_refusals.get(&id_digest).map(|&content_hash| RememberedWrite { + content_hash, + answer: StoredAnswer::Refused(CultMeshRudpRejectionReason::WriteLimitReached.refusal_index()), + }) + }); + let session_full = entry.answered_writes.len() >= MAX_REMEMBERED_WRITES_PER_SESSION; + let refusal_room = entry.limit_refusals.len() < MAX_REMEMBERED_WRITES_PER_SESSION; + let remembered_total: usize = self.sessions.values().map(|entry| entry.answered_writes.len()).sum(); + let budget_full = remembered_total + .checked_add(1) + .and_then(|entries| entries.checked_mul(CultMeshRudpDocumentServerOptions::REMEMBERED_ENTRY_BYTES)) + .is_none_or(|bytes| bytes > self.options.max_remembered_write_bytes); Ok(ControlFlow::Break(match prior { - Some((hash, answer)) if hash == content_hash => self.answer_sent(key, now, &answer, rejection)?, + Some(remembered) if remembered.content_hash == content_hash => { + let answer = match remembered.answer { + StoredAnswer::Admitted => admitted.clone(), + StoredAnswer::Refused(index) => { + rejection(CultMeshRudpRejectionReason::from_refusal_index(index)).refusal_message(true) + } + }; + self.answer_sent(key, now, &answer, rejection)? + } Some(_) => Some(rejection(CultMeshRudpRejectionReason::MessageIdReused)), - None if remembered >= MAX_REMEMBERED_WRITES_PER_SESSION => { + None if session_full || budget_full => { + if refusal_room { + if let Some(entry) = self.sessions.get_mut(&key) { + entry.limit_refusals.insert(id_digest, content_hash); + } + } Some(rejection(CultMeshRudpRejectionReason::WriteLimitReached)) } None => return Ok(ControlFlow::Continue(())), })) } - fn remember(&mut self, key: CultMeshRudpSessionKey, message_id: &str, content_hash: [u8; 32], answer: CultNetMessage) { + fn remember(&mut self, key: CultMeshRudpSessionKey, id_digest: [u8; 32], content_hash: [u8; 32], answer: StoredAnswer) { if let Some(entry) = self.sessions.get_mut(&key) { - entry.answered_writes.insert(message_id.to_owned(), (content_hash, answer)); + entry.answered_writes.insert(id_digest, RememberedWrite { content_hash, answer }); } } @@ -1346,7 +1496,7 @@ where if self.send_reliable_schema(key, now, answer) { return Ok(None); } - self.end_session(key, SessionEnd::Refused)?; + self.end_session(key, SessionEnd::LocalFailure)?; Ok(Some(rejection(CultMeshRudpRejectionReason::ResponseSendFailed( "the write answer could not be sent".into(), )))) @@ -1463,10 +1613,12 @@ where let Some(mut entry) = self.sessions.remove(&key) else { return Ok(()); }; + let local_failure = matches!(end, SessionEnd::LocalFailure); let (refusal, mut unsendable) = match end { SessionEnd::Refused => (Vec::new(), None), SessionEnd::Rejected(refusal) => (self.refusal_packets(&mut entry.session, &refusal, false), None), SessionEnd::Unsendable(error) => (Vec::new(), Some(error)), + SessionEnd::LocalFailure => (Vec::new(), None), }; entry.session.reset_peer_state(); let nothing = entry.session.create_ack(); @@ -1478,9 +1630,10 @@ where break; } } - let goodbye = match &unsendable { - Some(error) => entry.session.end_unsendable_session(error), - None => entry.session.end_refused_session(), + let goodbye = match (&unsendable, local_failure) { + (Some(error), _) => entry.session.end_unsendable_session(error), + (None, true) => entry.session.end_local_failure_session(), + (None, false) => entry.session.end_refused_session(), }; // The goodbye is the session's last datagram. If it too can never be // sent, nobody is left to tell. @@ -1603,6 +1756,9 @@ fn validate_options(options: &CultMeshRudpDocumentServerOptions) -> Result<()> { if options.session_max_lifetime.is_zero() { return Err(anyhow!("session_max_lifetime must be greater than zero")); } + if options.max_remembered_write_bytes == 0 { + return Err(anyhow!("max_remembered_write_bytes must be greater than zero")); + } if options.max_admitted_payload_bytes == 0 { return Err(anyhow!( "max_admitted_payload_bytes must be greater than zero" diff --git a/packages/cultmesh-rs/tests/rudp_document_server.rs b/packages/cultmesh-rs/tests/rudp_document_server.rs index 79e9e14c..0957d008 100644 --- a/packages/cultmesh-rs/tests/rudp_document_server.rs +++ b/packages/cultmesh-rs/tests/rudp_document_server.rs @@ -2332,8 +2332,9 @@ fn a_full_dedupe_table_refuses_new_writes_and_keeps_the_session_and_its_receipts send_reliable(&mut peer, &put("w0", "note:0", vec![1]))?; let answers = frames_after(&mut server, &mut peer, 2)?; assert_eq!(refusal_of(&answers[0]), (Some(CultNetErrorCode::LimitReached), Some("over".into()))); - let CultNetMessage::ErrorV1 { error, .. } = &answers[0] else { unreachable!() }; + let CultNetMessage::ErrorV1 { error, details, .. } = &answers[0] else { unreachable!() }; assert_eq!(error, "limit_reached: the session's remembered-write limit is full", "the cap has its own fixed text"); + assert_eq!(details, &Some(CultNetErrorDetails::default()), "limit_reached names no field: details all null"); assert_eq!(answers[1], admitted("w0", "note:0"), "a remembered id is still answered idempotently"); assert_eq!(server.session_count(), 1); Ok(()) @@ -2423,6 +2424,17 @@ fn a_receipt_that_cannot_be_queued_ends_the_session_instead_of_leaving_the_write } } assert_eq!(server.session_count(), 0, "a receipt the queue could not take ended the session"); + // The goodbye names a local failure: it is not the "session refused a packet" goodbye of a protocol refusal. + let mut reason = None; + for _ in 0..200 { + while peer.receive_once()?.is_some() {} + if let Some(found) = peer.disconnect_reason() { + reason = Some(found.to_vec()); + break; + } + thread::sleep(Duration::from_millis(1)); + } + assert_eq!(reason.expect("the peer was told the session ended"), b"server could not answer a write".to_vec()); Ok(()) } @@ -2623,3 +2635,115 @@ fn a_put_that_reuses_a_deletes_id_is_message_invalid() -> Result<()> { assert_eq!(sink.0.lock().unwrap().receipts.len(), 0, "the reusing put was not applied"); Ok(()) } + +fn error_text_and_field(message: &CultNetMessage) -> (String, Option) { + let CultNetMessage::ErrorV1 { error, details, .. } = message else { panic!("expected error.v1, got {message:?}") }; + (error.clone(), details.as_ref().and_then(|d| d.field.clone())) +} + +/// The server-wide budget counts fixed-size entries across sessions; a write over it is refused for that write +/// alone, the refusal is remembered (a resend replays it, no slot is freed), and every session is kept. +#[test] +fn a_server_wide_memory_budget_refuses_the_write_over_it_and_keeps_the_sessions() -> Result<()> { + let options = CultMeshRudpDocumentServerOptions { + max_remembered_write_bytes: 2 * CultMeshRudpDocumentServerOptions::REMEMBERED_ENTRY_BYTES, + ..Default::default() + }; + let sink = Sink::default(); + let mut server = server(options, Clock::new(70_000), sink.clone(), Source::default())?; + let mut a = client(server.local_addr()?, 230)?; + let mut b = client(server.local_addr()?, 231)?; + connect(&mut server, &mut [&mut a, &mut b])?; + send_reliable(&mut a, &receipt_hello())?; + send_reliable(&mut b, &receipt_hello())?; + send_reliable(&mut a, &put("a1", "note:a1", vec![1]))?; + send_reliable(&mut b, &put("b1", "note:b1", vec![1]))?; + for _ in 0..200 { + server.poll_once()?; + a.poll_resends()?; + b.poll_resends()?; + while a.receive_once()?.is_some() {} + while b.receive_once()?.is_some() {} + } + assert_eq!(sink.0.lock().unwrap().receipts.len(), 2); + send_reliable(&mut a, &put("a2", "note:a2", vec![1]))?; + send_reliable(&mut a, &put("a2", "note:a2", vec![1]))?; + let answers = frames_after(&mut server, &mut a, 2)?; + for answer in &answers { + assert_eq!(refusal_of(answer), (Some(CultNetErrorCode::LimitReached), Some("a2".into())), "the resend replays the refusal"); + let CultNetMessage::ErrorV1 { details, .. } = answer else { unreachable!() }; + assert_eq!(details, &Some(CultNetErrorDetails::default())); + } + assert_eq!(sink.0.lock().unwrap().receipts.len(), 2, "the refused write was not applied"); + assert_eq!(server.session_count(), 2, "both sessions are kept"); + send_reliable(&mut a, &put("a1", "note:a1", vec![1]))?; + assert_eq!(frames_after(&mut server, &mut a, 1)?, vec![admitted("a1", "note:a1")], "a remembered id is still answered"); + Ok(()) +} + +/// A message id over 256 bytes is refused before anything is remembered: no entry, no budget charge. +#[test] +fn a_message_id_over_256_bytes_is_refused_before_anything_is_remembered() -> Result<()> { + let options = CultMeshRudpDocumentServerOptions { + max_remembered_write_bytes: CultMeshRudpDocumentServerOptions::REMEMBERED_ENTRY_BYTES, + ..Default::default() + }; + let sink = Sink::default(); + let mut server = server(options, Clock::new(70_000), sink.clone(), Source::default())?; + let mut peer = client(server.local_addr()?, 232)?; + connect(&mut server, &mut [&mut peer])?; + send_reliable(&mut peer, &receipt_hello())?; + let long = "x".repeat(257); + send_reliable(&mut peer, &put(&long, "note:1", vec![1]))?; + send_reliable(&mut peer, &put(&long, "note:1", vec![1]))?; + send_reliable(&mut peer, &put(&"y".repeat(256), "note:2", vec![1]))?; + let answers = answers_after(&mut server, &mut peer, 3)?; + for answer in &answers[..2] { + assert_eq!(refusal_of(answer), (Some(CultNetErrorCode::MessageInvalid), Some(long.clone()))); + assert_eq!(error_text_and_field(answer), ("message_invalid: messageId exceeds 256 bytes".into(), Some("messageId".into()))); + } + assert_eq!(answers[2], admitted(&"y".repeat(256), "note:2"), "256 bytes is allowed, and the long id charged no budget"); + assert_eq!(sink.0.lock().unwrap().receipts.len(), 1, "the long-id writes were never applied"); + Ok(()) +} + +/// A write on an unreliable channel of a receipt session is refused, not dropped and not accepted. +#[test] +fn an_unreliable_write_on_a_receipt_session_is_refused_message_invalid_naming_the_channel() -> Result<()> { + let sink = Sink::default(); + let mut server = server(Default::default(), Clock::new(70_000), sink.clone(), Source::default())?; + let mut peer = RawPeer::connect(&mut server, 233)?; + peer.send(&receipt_hello())?; + server.poll_once()?; + let payload = encode_cultnet_message_to_vec(&put("u1", "note:1", vec![1]), CultNetWireContract::CultNetSchemaV0)?; + let packets = peer.session.send_many( + "schema", + payload, + cultnet_rs::CultNetRudpSendOptions { reliable: false, ordered: false, sequenced: false, now_ms: 3, reliable_expire_after_ms: None }, + Some(1200), + )?; + for packet in &packets { + peer.send_packet(packet)?; + } + server.poll_once()?; + let mut answers = Vec::new(); + for packet in peer.drain() { + for frame in peer.session.receive(&packet, 4)?.delivered { + answers.push(decode_cultnet_message_from_slice(&frame.payload, CultNetWireContract::CultNetSchemaV0)?); + } + } + let refusals: Vec<_> = answers.iter().filter(|answer| matches!(answer, CultNetMessage::ErrorV1 { .. })).collect(); + assert_eq!(refusals.len(), 1, "the unreliable write is answered, not dropped: {answers:?}"); + assert_eq!(refusal_of(refusals[0]), (Some(CultNetErrorCode::MessageInvalid), Some("u1".into()))); + assert_eq!( + error_text_and_field(refusals[0]), + ("message_invalid: receipt sessions require the reliable channel".into(), Some("channel".into())) + ); + assert_eq!(sink.0.lock().unwrap().receipts.len(), 0, "nothing was applied"); + assert_eq!(server.session_count(), 1); + // The id was not consumed: the same write on the reliable channel is admitted. + peer.send(&put("u1", "note:1", vec![1]))?; + server.poll_once()?; + assert_eq!(sink.0.lock().unwrap().receipts.len(), 1); + Ok(()) +} diff --git a/packages/cultnet-rs/src/contracts.rs b/packages/cultnet-rs/src/contracts.rs index 1aa39d11..531fe7ec 100644 --- a/packages/cultnet-rs/src/contracts.rs +++ b/packages/cultnet-rs/src/contracts.rs @@ -569,7 +569,8 @@ pub fn encode_cultnet_message_for_wire( | CultNetMessage::ContentChunkRequest { .. } | CultNetMessage::ContentChunkResponse { .. } | CultNetMessage::Error { .. } - | CultNetMessage::ErrorV1 { .. } => encode_raw_cultnet_schema_message(message), + | CultNetMessage::ErrorV1 { .. } + | CultNetMessage::WriteAdmitted { .. } => encode_raw_cultnet_schema_message(message), _ => Ok(cultcache_rs::decode_payload(&rmp_serde::to_vec( &serde_json::to_value(message)?, )?)?), @@ -1146,6 +1147,14 @@ fn encode_raw_cultnet_schema_message(message: &CultNetMessage) -> Result vec![ + (rmpv::Value::from("schemaVersion"), rmpv::Value::from("cultnet.write_admitted.v0")), + (rmpv::Value::from("messageId"), rmpv::Value::from(message_id.as_str())), + (rmpv::Value::from("operation"), rmpv::Value::from(operation.as_str())), + (rmpv::Value::from("schemaId"), rmpv::Value::from(schema_id.as_str())), + (rmpv::Value::from("recordKey"), rmpv::Value::from(record_key.as_str())), + ], CultNetMessage::DocumentPutRaw { message_id, document, @@ -1991,6 +2000,13 @@ pub fn write_content_hash( Ok(Sha256::digest(&bytes).into()) } +/// SHA-256 of a write's `messageId`: what a receipt server remembers in place of the id itself, so an +/// entry's memory does not depend on the id's length (write-receipt.md, "What is remembered"). +pub fn message_id_digest(message_id: &str) -> [u8; 32] { + use sha2::{Digest, Sha256}; + Sha256::digest(message_id.as_bytes()).into() +} + /// The `messageId` of a schema frame that does not decode as a message (write-receipt.md, rule 3: a refusal /// carries the write's id whenever the frame is readable enough to find it). `None` when the bytes are not /// a MessagePack map or hold no non-empty string `messageId`. diff --git a/packages/cultnet-rs/src/rudp.rs b/packages/cultnet-rs/src/rudp.rs index 6a2aa20b..eae7a2ee 100644 --- a/packages/cultnet-rs/src/rudp.rs +++ b/packages/cultnet-rs/src/rudp.rs @@ -468,6 +468,14 @@ impl CultNetRudpSession { self.create_disconnect(RUDP_REFUSED_PACKET_REASON.to_vec()) } + /// Ends a session because the server itself could not answer a write it owes an answer to (its answer + /// could not be encoded, queued or sent). The peer did nothing wrong: this goodbye is not the + /// "session refused a packet" of a protocol refusal. + pub fn end_local_failure_session(&mut self) -> CultNetRudpPacket { + self.reset_peer_state(); + self.create_disconnect(RUDP_LOCAL_FAILURE_REASON.to_vec()) + } + /// Ends a session that owes its peer a packet that can never be sent as /// built (see `is_permanent_send_error`). Resending it would fail forever, /// and dropping it would leave a reliable sequence the peer waits on for @@ -2609,6 +2617,8 @@ where /// Sent to a peer whose session refused one of its packets, and reported to /// the local caller as the disconnect reason. const RUDP_REFUSED_PACKET_REASON: &[u8] = b"session refused a packet"; +/// Goodbye reason of a session whose server could not answer a write (`end_local_failure_session`). +const RUDP_LOCAL_FAILURE_REASON: &[u8] = b"server could not answer a write"; const SESSION_TIMED_OUT_REASON: &[u8] = b"session timed out"; /// Prefix of the goodbye reason for a session that owed a packet which can /// never be sent; the failure's fixed name (`send_error_code`) follows it. diff --git a/packages/cultnet-rs/tests/write_receipt.rs b/packages/cultnet-rs/tests/write_receipt.rs index 4ad7a826..6d06a68c 100644 --- a/packages/cultnet-rs/tests/write_receipt.rs +++ b/packages/cultnet-rs/tests/write_receipt.rs @@ -101,7 +101,7 @@ fn error_v1_limit_reached_is_the_spec_vector() { message_id: Some("put-1025".into()), }; let bytes = encode_cultnet_message_to_vec(&message, CultNetWireContract::CultNetSchemaV0).unwrap(); - assert_eq!(hex(&bytes), vector); + assert_eq!(bytes, unhex(vector), "encoded bytes must equal the spec vector"); let CultNetMessage::ErrorV1 { error, code, details, message_id } = decode_cultnet_message_from_slice(&unhex(vector), CultNetWireContract::CultNetSchemaV0).unwrap() else { @@ -113,6 +113,25 @@ fn error_v1_limit_reached_is_the_spec_vector() { assert_eq!(details, Some(CultNetErrorDetails::default())); } +/// Spec vectors `write_admitted_put` and `write_admitted_delete`: the encoder emits the spec's field order. +#[test] +fn write_admitted_encodes_to_the_spec_vector_bytes() { + let cases = [ + ("put-1", "put", "85AD736368656D6156657273696F6EB963756C746E65742E77726974655F61646D69747465642E7630A96D6573736167654964A57075742D31A96F7065726174696F6EA3707574A8736368656D614964A76E6F74652E7630A97265636F72644B6579A66E6F74653A31"), + ("del-1", "delete", "85AD736368656D6156657273696F6EB963756C746E65742E77726974655F61646D69747465642E7630A96D6573736167654964A564656C2D31A96F7065726174696F6EA664656C657465A8736368656D614964A76E6F74652E7630A97265636F72644B6579A66E6F74653A31"), + ]; + for (id, operation, vector) in cases { + let message = CultNetMessage::WriteAdmitted { + message_id: id.into(), + operation: operation.into(), + schema_id: "note.v0".into(), + record_key: "note:1".into(), + }; + let bytes = encode_cultnet_message_to_vec(&message, CultNetWireContract::CultNetSchemaV0).unwrap(); + assert_eq!(bytes, unhex(vector)); + } +} + /// `limit_reached` is in the error.v1 enum only: a v0 error neither encodes nor decodes it. #[test] fn error_v0_refuses_the_v1_only_limit_reached_code() {