From 2d383d781885a20d7b5860118da03ddb8447a3ed Mon Sep 17 00:00:00 2001 From: Meta Vens Date: Wed, 28 Oct 2026 06:05:09 +0100 Subject: [PATCH] Fix pass 4: long request ids are refused on the wire, frame rule is public, a FrameBound completion frees its slot, replays are re-measured against the frame - begin(): an id past MAX_IDENTITY_BYTES gets Refusal::Malformed naming the request by its SHA-256, so the reply fits the smallest frame; no dropped connection - frame_bytes_for_payload, payload_bytes_for_frame and MAX_IDENTITY_BYTES are public; ServeError::FrameBound states the frame needed - complete(): FrameBound releases the caller's slot - begin() takes the frame: a recorded result that no longer fits is replaced by the stand-in failure Co-Authored-By: Claude Opus 5.5 --- README.md | 7 + crates/model-connector/src/envelope.rs | 54 +++-- crates/model-connector/src/serve.rs | 107 +++++++-- crates/model-connector/src/service.rs | 288 ++++++++++++++++++++----- 4 files changed, 372 insertions(+), 84 deletions(-) diff --git a/README.md b/README.md index 1155500..d01fb9e 100644 --- a/README.md +++ b/README.md @@ -24,6 +24,13 @@ Crate `crates/model-connector`: `ProviderBackend` trait. A connector binary implements `ProviderBackend` for its provider and calls `serve`; the kit names no provider. +Frame arithmetic is public in `src/envelope.rs`: `frame_bytes_for_payload(n)` is the frame a +caller payload bound of `n` needs, `payload_bytes_for_frame(frame)` is the largest payload bound +a frame allows, and `MAX_IDENTITY_BYTES` (256) bounds caller ids, request ids and model names. +`serve` refuses a frame/payload pair that violates them (`ServeError::FrameBound` states the frame +needed); the smallest legal frame is `frame_bytes_for_payload(MIN_PAYLOAD_BYTES)`. A request id past +the bound is refused on the wire as `Refusal::Malformed`, the reply naming it by its SHA-256. + The default feature set compiles contract, envelope and client only. Schema ids are still `gamecult.codex.*.v2`; renaming them changes the wire and belongs to a later contract version. diff --git a/crates/model-connector/src/envelope.rs b/crates/model-connector/src/envelope.rs index 48e1548..a0df968 100644 --- a/crates/model-connector/src/envelope.rs +++ b/crates/model-connector/src/envelope.rs @@ -28,17 +28,18 @@ pub struct TransportEnvelope { } /// The longest caller id, request id or model name a service admits. The -/// frame arithmetic below depends on it, so a longer one is refused up front. -#[cfg(feature = "daemon")] -pub(crate) const MAX_IDENTITY_BYTES: usize = 256; +/// frame arithmetic below depends on it. A request whose id is longer is +/// refused on the wire as malformed. +pub const MAX_IDENTITY_BYTES: usize = 256; /// The most bytes a serialized envelope holding `ciphertext_len` ciphertext -/// bytes can occupy. The v2 wire serializes the ciphertext as a MessagePack -/// array of integers: a byte of 128 or more takes two bytes, so the worst case -/// is twice the ciphertext, plus the array header and the envelope's other +/// bytes can occupy, with ids at their longest: the frame a caller payload +/// bound (`CallerAdmission`'s `max_payload_bytes`, a bound on ciphertext +/// length) needs. The v2 wire serializes the ciphertext as a MessagePack array +/// of integers: a byte of 128 or more takes two bytes, so the worst case is +/// twice the ciphertext, plus the array header and the envelope's other /// fields at their longest. Measured from the real encoding, not assumed. -#[cfg(feature = "daemon")] -pub(crate) fn worst_frame_bytes(ciphertext_len: usize) -> usize { +pub fn frame_bytes_for_payload(ciphertext_len: usize) -> usize { let shell = rmp_serde::to_vec(&TransportEnvelope { schema_id: ENVELOPE_SCHEMA_ID.to_string(), caller_runtime_id: "x".repeat(MAX_IDENTITY_BYTES), @@ -60,6 +61,26 @@ pub(crate) fn worst_frame_bytes(ciphertext_len: usize) -> usize { .saturating_add(ciphertext_len.saturating_mul(2)) } +/// The largest payload bound whose worst-case frame fits `max_frame_bytes`: +/// the inverse of [`frame_bytes_for_payload`]. `None` when the frame cannot +/// carry even an empty envelope. +pub fn payload_bytes_for_frame(max_frame_bytes: usize) -> Option { + if frame_bytes_for_payload(0) > max_frame_bytes { + return None; + } + // frame_bytes_for_payload is increasing; the answer is at most half the frame. + let (mut low, mut high) = (0, max_frame_bytes / 2); + while low < high { + let mid = low + (high - low).div_ceil(2); + if frame_bytes_for_payload(mid) <= max_frame_bytes { + low = mid; + } else { + high = mid - 1; + } + } + Some(low) +} + pub fn encrypt_invocation( invocation: &TransportInvocation, security: &TransportKey, @@ -431,7 +452,6 @@ mod tests { assert!(matches!(write(&[], 1_000), Err(TransportFrameError::Size))); } - #[cfg(feature = "daemon")] #[test] fn the_worst_frame_is_the_real_encoding_of_an_all_high_ciphertext() { let long = "x".repeat(MAX_IDENTITY_BYTES); @@ -446,13 +466,12 @@ mod tests { }; assert_eq!( rmp_serde::to_vec(&envelope).unwrap().len(), - worst_frame_bytes(length), + frame_bytes_for_payload(length), "{length}" ); } } - #[cfg(feature = "daemon")] #[test] fn a_sealed_envelope_never_exceeds_the_worst_frame_for_its_ciphertext() { let key = security("epiphany-distinct-test-key"); @@ -462,7 +481,7 @@ mod tests { encrypt_message(&id, &id, EnvelopeKind::Result, &vec![0x41; plaintext], &key) .unwrap(); let real = rmp_serde::to_vec(&sealed).unwrap().len(); - let worst = worst_frame_bytes(sealed.ciphertext.len()); + let worst = frame_bytes_for_payload(sealed.ciphertext.len()); assert!(real <= worst, "{real} > {worst}"); // Real ciphertext is random, so about half its bytes take two // bytes: the worst case is not wildly above what is written. @@ -475,4 +494,15 @@ mod tests { } } } + + #[test] + fn the_payload_bound_a_frame_allows_inverts_the_frame_a_payload_needs() { + assert_eq!(payload_bytes_for_frame(frame_bytes_for_payload(0) - 1), None); + assert_eq!(payload_bytes_for_frame(frame_bytes_for_payload(0)), Some(0)); + for bytes in [1, 15, 16, 4096, 65_535, 65_536, 100_000, 2_000_000] { + let frame = frame_bytes_for_payload(bytes); + assert_eq!(payload_bytes_for_frame(frame), Some(bytes), "{bytes}"); + assert!(payload_bytes_for_frame(frame - 1).unwrap() < bytes, "{bytes}"); + } + } } diff --git a/crates/model-connector/src/serve.rs b/crates/model-connector/src/serve.rs index 1c763e9..e27a501 100644 --- a/crates/model-connector/src/serve.rs +++ b/crates/model-connector/src/serve.rs @@ -12,7 +12,7 @@ use zeroize::Zeroizing; use crate::contract::{TransportInvocation, TransportResult}; use crate::envelope::{ - ServiceError, TransportEnvelope, TransportFrameError, read_transport_frame, worst_frame_bytes, + ServiceError, TransportEnvelope, TransportFrameError, read_transport_frame, frame_bytes_for_payload, write_transport_frame, }; use crate::service::{CallerAdmission, TransportAdmission, TransportService}; @@ -39,8 +39,9 @@ pub struct ServeLimits { pub max_expiry_skew_ms: u64, } -/// The smallest caller payload bound the base daemon admitted. -const MIN_PAYLOAD_BYTES: usize = 4096; +/// The smallest caller payload bound serve admits. The smallest legal frame is +/// `frame_bytes_for_payload(MIN_PAYLOAD_BYTES)`. +pub const MIN_PAYLOAD_BYTES: usize = 4096; impl ServeLimits { /// The bounds serve refuses before it opens anything: each limit nonzero, @@ -51,7 +52,15 @@ impl ServeLimits { /// An admitted payload therefore always fits, and `complete` is given the /// same frame bound to fit every result in. fn validate(&self, admissions: &[CallerAdmission]) -> Result<(), ServeError> { - if self.max_frame_bytes < worst_frame_bytes(MIN_PAYLOAD_BYTES) + let smallest = frame_bytes_for_payload(MIN_PAYLOAD_BYTES); + if self.max_frame_bytes > 0 && self.max_frame_bytes < smallest { + return Err(ServeError::FrameBound { + max_frame_bytes: self.max_frame_bytes, + payload_bytes: MIN_PAYLOAD_BYTES, + needed_frame_bytes: smallest, + }); + } + if self.max_frame_bytes == 0 || self.max_frame_bytes > u32::MAX as usize || self.max_connections == 0 || self.socket_timeout_ms == 0 @@ -71,11 +80,18 @@ impl ServeLimits { if self.max_connections < slots.max(8) { return Err(ServeError::InvalidConfig("connection capacity")); } - if admissions.iter().any(|caller| { - caller.max_payload_bytes() < MIN_PAYLOAD_BYTES - || worst_frame_bytes(caller.max_payload_bytes()) > self.max_frame_bytes - }) { - return Err(ServeError::InvalidConfig("caller admission")); + for caller in admissions { + if caller.max_payload_bytes() < MIN_PAYLOAD_BYTES { + return Err(ServeError::InvalidConfig("caller admission")); + } + let needed = frame_bytes_for_payload(caller.max_payload_bytes()); + if needed > self.max_frame_bytes { + return Err(ServeError::FrameBound { + max_frame_bytes: self.max_frame_bytes, + payload_bytes: caller.max_payload_bytes(), + needed_frame_bytes: needed, + }); + } } Ok(()) } @@ -211,7 +227,7 @@ fn execute_inference_frame( let admission = service .lock() .map_err(|_| ServeError::ServicePoisoned)? - .begin(&envelope, unix_ms()?)?; + .begin(&envelope, unix_ms()?, limits.max_frame_bytes)?; let response = match admission { TransportAdmission::Reply(response) => response, TransportAdmission::Execute(claim) => { @@ -274,6 +290,14 @@ impl Drop for ConnectionPermit { pub enum ServeError { #[error("invalid serve configuration field {0}")] InvalidConfig(&'static str), + #[error( + "max_frame_bytes {max_frame_bytes} cannot carry a {payload_bytes}-byte payload bound: at least {needed_frame_bytes} frame bytes are needed" + )] + FrameBound { + max_frame_bytes: usize, + payload_bytes: usize, + needed_frame_bytes: usize, + }, #[error("connector connection key is unavailable")] ConnectionKey(#[source] std::io::Error), #[error(transparent)] @@ -496,6 +520,45 @@ mod tests { assert_eq!(backend.executions.load(Ordering::SeqCst), 1); } + #[test] + fn serve_refuses_a_long_request_id_on_the_wire_without_dropping_the_connection() { + let store = TestStore::new(); + let backend = CountingBackend::new(true); + let (_client, endpoint) = start(backend.clone(), &store); + let security = crate::fixtures::security(KEY); + let send = |id: &str| { + let request = rmp_serde::to_vec( + &crate::envelope::encrypt_invocation(&fresh(id), &security).unwrap(), + ) + .unwrap(); + let mut stream = TcpStream::connect(endpoint).unwrap(); + write_transport_frame(&mut stream, &request, FRAME).unwrap(); + let reply = read_transport_frame(&mut stream, FRAME).unwrap(); + let envelope: TransportEnvelope = rmp_serde::from_slice(&reply).unwrap(); + let result: TransportResult = rmp_serde::from_slice( + &crate::envelope::decrypt_message(&envelope, &security).unwrap(), + ) + .unwrap(); + (reply.len(), result) + }; + let (_, served) = send(&"i".repeat(crate::envelope::MAX_IDENTITY_BYTES)); + assert!(matches!( + served.disposition, + TransportDisposition::Transported { .. } + )); + assert_eq!(backend.executions.load(Ordering::SeqCst), 1); + for length in [257, 258, 5_000, 60_000] { + let (size, result) = send(&"i".repeat(length)); + assert_eq!( + result.disposition, + TransportDisposition::Refused(Refusal::Malformed), + "{length}" + ); + assert!(size <= smallest_frame(), "{length}: {size}"); + } + assert_eq!(backend.executions.load(Ordering::SeqCst), 1); + } + #[test] fn serve_refuses_to_start_on_an_unready_backend() { let store = TestStore::new(); @@ -547,7 +610,7 @@ mod tests { /// The smallest frame serve admits: the smallest payload at its worst-case /// sealed encoding. fn smallest_frame() -> usize { - worst_frame_bytes(MIN_PAYLOAD_BYTES) + frame_bytes_for_payload(MIN_PAYLOAD_BYTES) } #[test] @@ -558,16 +621,15 @@ mod tests { )) .unwrap() .len(); - let worst = worst_frame_bytes(plaintext + GCM_TAG); + let worst = frame_bytes_for_payload(plaintext + GCM_TAG); assert!(worst <= smallest_frame(), "{worst} > {}", smallest_frame()); } #[test] fn serve_refuses_unbounded_limits_and_admits_the_widest_frame() { let store = TestStore::new(); - let mutations: [fn(&mut ServeLimits); 6] = [ + let mutations: [fn(&mut ServeLimits); 5] = [ |limits| limits.max_frame_bytes = 0, - |limits| limits.max_frame_bytes = smallest_frame() - 1, |limits| limits.max_frame_bytes = u32::MAX as usize + 1, |limits| limits.max_connections = 0, |limits| limits.socket_timeout_ms = 0, @@ -658,6 +720,7 @@ mod tests { ) { None => None, Some(Err(ServeError::InvalidConfig(field))) => Some(field), + Some(Err(ServeError::FrameBound { .. })) => Some("frame bound"), Some(other) => panic!("unexpected serve outcome {other:?}"), } } @@ -699,13 +762,13 @@ mod tests { let tightest = smallest_frame(); assert_eq!( refusal(frame(tightest - 1), payload(4096)), - Some("serve limits") + Some("frame bound") ); // The tightest legal pair: the smallest payload in the smallest frame. assert_eq!(refusal(frame(tightest), payload(4096)), None); assert_eq!( refusal(frame(tightest), payload(4097)), - Some("caller admission") + Some("frame bound") ); assert_eq!( refusal(frame(tightest), payload(4095)), @@ -715,14 +778,14 @@ mod tests { // a payload that is the whole frame cannot be carried. assert_eq!( refusal(frame(FRAME), payload(FRAME)), - Some("caller admission") + Some("frame bound") ); for bytes in [8192, 65_535, 65_536, 100_000] { - let widest = worst_frame_bytes(bytes); + let widest = frame_bytes_for_payload(bytes); assert_eq!(refusal(frame(widest), payload(bytes)), None, "{bytes}"); assert_eq!( refusal(frame(widest - 1), payload(bytes)), - Some("caller admission"), + Some("frame bound"), "{bytes}" ); } @@ -730,7 +793,7 @@ mod tests { admission("a", "key-a-distinct", 4, 8192), admission("b", "key-b-distinct", 4, FRAME), ]; - assert_eq!(refusal(frame(FRAME), mixed), Some("caller admission")); + assert_eq!(refusal(frame(FRAME), mixed), Some("frame bound")); } #[test] @@ -744,7 +807,7 @@ mod tests { let key = crate::fixtures::security(KEY); let sealed = crate::envelope::encrypt_invocation(&invocation, &key).unwrap(); let bound = sealed.ciphertext.len(); - let frame = worst_frame_bytes(bound); + let frame = frame_bytes_for_payload(bound); let listener = TcpListener::bind("127.0.0.1:0").unwrap(); let endpoint = listener.local_addr().unwrap(); let path = store.path.clone(); @@ -920,7 +983,7 @@ mod tests { /// The largest result plaintext whose sealed frame fits at the worst encoding. fn widest_guaranteed_plaintext(base: usize) -> usize { (base..=FRAME) - .take_while(|plaintext| worst_frame_bytes(plaintext + GCM_TAG) <= FRAME) + .take_while(|plaintext| frame_bytes_for_payload(plaintext + GCM_TAG) <= FRAME) .last() .unwrap_or(base) } diff --git a/crates/model-connector/src/service.rs b/crates/model-connector/src/service.rs index 6836df1..4ec8cb6 100644 --- a/crates/model-connector/src/service.rs +++ b/crates/model-connector/src/service.rs @@ -195,15 +195,19 @@ impl TransportService { }) } + /// Admits one request. `max_frame_bytes` is the frame every reply must + /// fit, the same bound `complete` is given: a recorded result that no + /// longer fits it (the service reopened at a smaller frame) is answered + /// with the typed `result_size` failure, which is recorded in its place. + /// A request id longer than `MAX_IDENTITY_BYTES` is refused as malformed; + /// the refusal names the id by its digest so it fits any legal frame. pub fn begin( &mut self, envelope: &TransportEnvelope, now_unix_ms: u64, + max_frame_bytes: usize, ) -> Result { validate_envelope(envelope, EnvelopeKind::Invocation)?; - if envelope.request_id.len() > MAX_IDENTITY_BYTES { - return Err(ContractError::Invalid("request_id").into()); - } let caller = self .callers .get(&envelope.caller_runtime_id) @@ -211,6 +215,12 @@ impl TransportService { let security = caller.security.clone(); let max_concurrent_requests = caller.max_concurrent_requests; let invocation = decrypt_invocation(envelope, &security)?; + if envelope.request_id.len() > MAX_IDENTITY_BYTES { + return Ok(TransportAdmission::Reply(encrypt_result( + &unnamed_refusal(&invocation, Refusal::Malformed), + &security, + )?)); + } if envelope.ciphertext.len() > caller.max_payload_bytes { return self.reply_refusal(&invocation, Refusal::Policy, &security); } @@ -240,8 +250,25 @@ impl TransportService { if record.invocation_sha256 != invocation_sha256 { return self.reply_refusal(&invocation, Refusal::ReplayConflict, &security); } - return match record.state { - ReplayState::Completed { response } => Ok(TransportAdmission::Reply(response)), + return match &record.state { + ReplayState::Completed { response } => { + if fits_frame(response, max_frame_bytes)? { + return Ok(TransportAdmission::Reply(response.clone())); + } + let stand_in = self.seal_stand_in(&invocation, &security, max_frame_bytes)?; + self.replay_store + .put( + replay_key(&identity), + &ReplayRecord { + state: ReplayState::Completed { + response: stand_in.clone(), + }, + ..record.clone() + }, + ) + .map_err(replay_store_error)?; + Ok(TransportAdmission::Reply(stand_in)) + } ReplayState::Active => { let refusal = if self.live.contains(&identity) { Refusal::InFlight @@ -318,22 +345,23 @@ impl TransportService { .callers .get(&identity.caller_runtime_id) .ok_or(ServiceError::CallerNotAdmitted)?; - let mut response = encrypt_result(&result, &caller.security)?; + let security = caller.security.clone(); + let mut response = encrypt_result(&result, &security)?; if !fits_frame(&response, max_frame_bytes)? { - response = encrypt_result( - &result_size_failure( - &claim.invocation.caller_runtime_id, - claim.invocation.request_id(), - &claim.invocation.request.model, - claim.invocation.native_request_sha256, - claim.invocation.provider_request_sha256, - ), - &caller.security, - )?; - if !fits_frame(&response, max_frame_bytes)? { - return Err(ServiceError::FrameBound); - } + response = match self.seal_stand_in(&claim.invocation, &security, max_frame_bytes) { + Ok(stand_in) => stand_in, + Err(error) => { + // Nothing is recorded, but the caller's slot is not held + // by a request that can no longer finish. + self.live.remove(&identity); + return Err(error); + } + }; } + let caller = self + .callers + .get(&identity.caller_runtime_id) + .ok_or(ServiceError::CallerNotAdmitted)?; let replay_record = ReplayRecord { caller_runtime_id: identity.caller_runtime_id.clone(), request_id: identity.request_id.clone(), @@ -351,6 +379,30 @@ impl TransportService { Ok(response) } + /// The `result_size` failure sealed for `invocation`, if it fits the frame. + fn seal_stand_in( + &self, + invocation: &TransportInvocation, + security: &TransportKey, + max_frame_bytes: usize, + ) -> Result { + let stand_in = encrypt_result( + &result_size_failure( + &invocation.caller_runtime_id, + invocation.request_id(), + &invocation.request.model, + invocation.native_request_sha256, + invocation.provider_request_sha256, + ), + security, + )?; + if fits_frame(&stand_in, max_frame_bytes)? { + Ok(stand_in) + } else { + Err(ServiceError::FrameBound) + } + } + fn reply_refusal( &self, invocation: &TransportInvocation, @@ -412,14 +464,27 @@ fn replay_key(identity: &RequestIdentity) -> String { hasher.update(identity.caller_runtime_id.as_bytes()); hasher.update((identity.request_id.len() as u64).to_be_bytes()); hasher.update(identity.request_id.as_bytes()); + hex(hasher.finalize().into()) +} + +fn hex(digest: [u8; 32]) -> String { const HEX: &[u8; 16] = b"0123456789abcdef"; - let digest = hasher.finalize(); - let mut key = String::with_capacity(64); + let mut text = String::with_capacity(64); for byte in digest { - key.push(HEX[(byte >> 4) as usize] as char); - key.push(HEX[(byte & 0x0f) as usize] as char); + text.push(HEX[(byte >> 4) as usize] as char); + text.push(HEX[(byte & 0x0f) as usize] as char); + } + text +} + +/// A refusal for an invocation whose request id is too long to echo: it names +/// the request by the hex SHA-256 of the id (64 bytes), so the reply fits the +/// smallest legal frame whatever the id's length. +fn unnamed_refusal(invocation: &TransportInvocation, reason: Refusal) -> TransportResult { + TransportResult { + request_id: hex(Sha256::digest(invocation.request_id().as_bytes()).into()), + ..TransportResult::refused(invocation, reason) } - key } fn replay_store_error(error: impl std::fmt::Display) -> ServiceError { @@ -441,6 +506,8 @@ mod tests { use crate::fixtures::*; use crate::{contract::*, envelope::*}; + const FRAME: usize = 1 << 20; + #[test] fn service_refuses_shared_keys_and_isolates_caller_request_identities() { let store = TestStore::new(); @@ -464,7 +531,7 @@ mod tests { .begin( &encrypt_invocation(&epiphany, &security("epiphany-distinct-test-key")) .unwrap(), - 1_000, + 1_000, FRAME, ) .unwrap(), TransportAdmission::Execute(_) @@ -474,7 +541,7 @@ mod tests { .begin( &encrypt_invocation(&ghostlight, &security("ghostlight-distinct-test-key")) .unwrap(), - 1_000, + 1_000, FRAME, ) .unwrap(), TransportAdmission::Execute(_) @@ -488,7 +555,7 @@ mod tests { let security = security("epiphany-distinct-test-key"); let mut service = open_service(&store, 1); let claim = match service - .begin(&encrypt_invocation(&invocation, &security).unwrap(), 1_000) + .begin(&encrypt_invocation(&invocation, &security).unwrap(), 1_000, FRAME) .unwrap() { TransportAdmission::Execute(claim) => claim, @@ -516,7 +583,7 @@ mod tests { let mut service = open_service(&store, 1); let replay = match service - .begin(&encrypt_invocation(&invocation, &security).unwrap(), 1_000) + .begin(&encrypt_invocation(&invocation, &security).unwrap(), 1_000, FRAME) .unwrap() { TransportAdmission::Reply(response) => response, @@ -535,7 +602,7 @@ mod tests { TransportInvocation::new("epiphany-yggdrasil", 2_000, [7; 32], changed_request) .unwrap(); let refusal = match service - .begin(&encrypt_invocation(&conflicting, &security).unwrap(), 1_000) + .begin(&encrypt_invocation(&conflicting, &security).unwrap(), 1_000, FRAME) .unwrap() { TransportAdmission::Reply(response) => response, @@ -558,13 +625,13 @@ mod tests { let mut service = open_service(&store, 1); let first_envelope = encrypt_invocation(&first, &security).unwrap(); assert!(matches!( - service.begin(&first_envelope, 1_000).unwrap(), + service.begin(&first_envelope, 1_000, FRAME).unwrap(), TransportAdmission::Execute(_) )); for (invocation, expected) in [(&first, Refusal::InFlight), (&second, Refusal::Capacity)] { let response = match service - .begin(&encrypt_invocation(invocation, &security).unwrap(), 1_000) + .begin(&encrypt_invocation(invocation, &security).unwrap(), 1_000, FRAME) .unwrap() { TransportAdmission::Reply(response) => response, @@ -588,7 +655,7 @@ mod tests { let mut service = open_service(&store, 1); assert!(matches!( service - .begin(&encrypt_invocation(&first, &security).unwrap(), 1_000) + .begin(&encrypt_invocation(&first, &security).unwrap(), 1_000, FRAME) .unwrap(), TransportAdmission::Execute(_) )); @@ -596,7 +663,7 @@ mod tests { let mut restarted = open_service(&store, 1); let ambiguous = match restarted - .begin(&encrypt_invocation(&first, &security).unwrap(), 1_000) + .begin(&encrypt_invocation(&first, &security).unwrap(), 1_000, FRAME) .unwrap() { TransportAdmission::Reply(response) => response, @@ -610,7 +677,7 @@ mod tests { ); assert!(matches!( restarted - .begin(&encrypt_invocation(&second, &security).unwrap(), 1_000) + .begin(&encrypt_invocation(&second, &security).unwrap(), 1_000, FRAME) .unwrap(), TransportAdmission::Execute(_) )); @@ -646,7 +713,7 @@ mod tests { edit(&mut request); let invocation = TransportInvocation::new(CALLER, 2_000, [7; 32], request).unwrap(); let envelope = encrypt_invocation(&invocation, &security(KEY)).unwrap(); - (service.begin(&envelope, 1_000).unwrap(), invocation) + (service.begin(&envelope, 1_000, FRAME).unwrap(), invocation) } fn refusal_of(admission: TransportAdmission, invocation: &TransportInvocation) -> Refusal { @@ -727,7 +794,7 @@ mod tests { .unwrap(); TransportService::open(&store.path, [admission], 5_000) .unwrap() - .begin(&envelope, 1_000) + .begin(&envelope, 1_000, FRAME) }; assert!(matches!(admit(size), Ok(TransportAdmission::Execute(_)))); let refused = admit(size - 1).unwrap(); @@ -741,12 +808,12 @@ mod tests { let mut expired = invocation_for(CALLER, "expired"); expired.expires_at_unix_ms = 10; - let reply = service.begin(&raw_envelope(&expired), 1_000).unwrap(); + let reply = service.begin(&raw_envelope(&expired), 1_000, FRAME).unwrap(); assert_eq!(refusal_of(reply, &expired), Refusal::Expired); let mut substituted = invocation_for(CALLER, "substituted"); substituted.provider_request_sha256 = [9; 32]; - let reply = service.begin(&raw_envelope(&substituted), 1_000).unwrap(); + let reply = service.begin(&raw_envelope(&substituted), 1_000, FRAME).unwrap(); assert_eq!( refusal_of(reply, &substituted), Refusal::ProviderDigestSubstitution @@ -754,7 +821,7 @@ mod tests { let mut malformed = invocation_for(CALLER, "malformed"); malformed.native_request_sha256 = [0; 32]; - let reply = service.begin(&raw_envelope(&malformed), 1_000).unwrap(); + let reply = service.begin(&raw_envelope(&malformed), 1_000, FRAME).unwrap(); assert_eq!(refusal_of(reply, &malformed), Refusal::Malformed); } @@ -880,14 +947,46 @@ mod tests { let mut service = open_service(&store, 8); let ok = invocation_for(CALLER, &at); assert!(matches!( - service.begin(&raw_envelope(&ok), 1_000).unwrap(), + service.begin(&raw_envelope(&ok), 1_000, FRAME).unwrap(), TransportAdmission::Execute(_) )); - let long = invocation_for(CALLER, &over); - assert_eq!( - service.begin(&raw_envelope(&long), 1_000).err(), - Some(ServiceError::Contract(ContractError::Invalid("request_id"))) - ); + } + + #[test] + fn a_request_id_past_the_bound_is_refused_malformed_in_a_reply_that_fits_the_smallest_frame() { + let smallest = frame_bytes_for_payload(4096); + for length in [ + MAX_IDENTITY_BYTES + 1, + MAX_IDENTITY_BYTES + 2, + 5_000, + 100_000, + 400_000, + ] { + let store = TestStore::new(); + let mut service = open_service(&store, 1); + let id = "i".repeat(length); + let long = invocation_for(CALLER, &id); + let TransportAdmission::Reply(reply) = service + .begin(&raw_envelope(&long), 1_000, smallest) + .unwrap() + else { + panic!("an over-long id must be refused, not executed ({length})"); + }; + assert!( + rmp_serde::to_vec(&reply).unwrap().len() <= smallest, + "{length}" + ); + let result: TransportResult = + rmp_serde::from_slice(&decrypt_message(&reply, &security(KEY)).unwrap()).unwrap(); + assert_eq!( + result.disposition, + TransportDisposition::Refused(Refusal::Malformed) + ); + assert_eq!(result.request_id, reply.request_id); + assert_eq!(result.request_id.len(), 64, "the id is named, not echoed"); + // Nothing was recorded or held: the caller's one slot is free. + let (_claim, _) = claim_for(&mut service, "after-the-long-one"); + } } #[test] @@ -904,7 +1003,7 @@ mod tests { request_id: &str, ) -> (Box, TransportInvocation) { let invocation = invocation_for(CALLER, request_id); - match service.begin(&raw_envelope(&invocation), 1_000).unwrap() { + match service.begin(&raw_envelope(&invocation), 1_000, FRAME).unwrap() { TransportAdmission::Execute(claim) => (claim, invocation), TransportAdmission::Reply(_) => panic!("expected an execution claim"), } @@ -955,7 +1054,7 @@ mod tests { // The recorded replay is the stand-in's exact bytes. let TransportAdmission::Reply(replay) = - service.begin(&raw_envelope(&invocation), 1_000).unwrap() + service.begin(&raw_envelope(&invocation), 1_000, FRAME).unwrap() else { panic!("expected a replay"); }; @@ -975,6 +1074,94 @@ mod tests { ); } + #[test] + fn a_frame_bound_failure_releases_the_callers_slot() { + let store = TestStore::new(); + let mut service = open_service(&store, 1); + let (claim, invocation) = claim_for(&mut service, "wedge"); + let big = text_result(&invocation, "y", "fixture-sse"); + assert_eq!( + service.complete(*claim, big, 10).unwrap_err(), + ServiceError::FrameBound + ); + // The caller's only slot is free again, and the dead request is no + // longer in flight. + let (_next, _) = claim_for(&mut service, "after-wedge"); + let TransportAdmission::Reply(reply) = service + .begin(&raw_envelope(&invocation), 1_000, FRAME) + .unwrap() + else { + panic!("expected a refusal"); + }; + assert_eq!( + decrypt_result(&reply, &security(KEY), &invocation) + .unwrap() + .disposition, + TransportDisposition::Refused(Refusal::Indeterminate) + ); + } + + #[test] + fn a_recorded_result_replayed_after_a_restart_at_a_smaller_frame_is_re_measured() { + let store = TestStore::new(); + let mut service = open_service(&store, 8); + let (claim, invocation) = claim_for(&mut service, "restart"); + let big = text_result(&invocation, &"y".repeat(4000), "fixture-sse"); + let recorded = service.complete(*claim, big, FRAME).unwrap(); + assert!(rmp_serde::to_vec(&recorded).unwrap().len() > 2000); + drop(service); + + let mut service = open_service(&store, 8); + let TransportAdmission::Reply(same) = service + .begin(&raw_envelope(&invocation), 1_000, FRAME) + .unwrap() + else { + panic!("expected a replay"); + }; + assert_eq!(same, recorded, "a frame that still fits replays the record"); + drop(service); + + let mut service = open_service(&store, 8); + let TransportAdmission::Reply(smaller) = service + .begin(&raw_envelope(&invocation), 1_000, 2000) + .unwrap() + else { + panic!("expected a replay"); + }; + assert!(rmp_serde::to_vec(&smaller).unwrap().len() <= 2000); + assert_eq!( + decrypt_result(&smaller, &security(KEY), &invocation).unwrap(), + result_size_failure( + CALLER, + "restart", + &invocation.request.model, + invocation.native_request_sha256, + invocation.provider_request_sha256, + ) + ); + // The stand-in is what is recorded now, byte for byte, at any frame. + let TransportAdmission::Reply(again) = service + .begin(&raw_envelope(&invocation), 1_000, FRAME) + .unwrap() + else { + panic!("expected a replay"); + }; + assert_eq!(again, smaller); + } + + #[test] + fn a_replay_whose_stand_in_cannot_fit_either_is_a_frame_bound_failure() { + let store = TestStore::new(); + let mut service = open_service(&store, 8); + let (claim, invocation) = claim_for(&mut service, "no-room"); + let big = text_result(&invocation, "y", "fixture-sse"); + service.complete(*claim, big, FRAME).unwrap(); + assert_eq!( + service.begin(&raw_envelope(&invocation), 1_000, 10).err(), + Some(ServiceError::FrameBound) + ); + } + #[test] fn nothing_is_recorded_when_not_even_the_stand_in_fits() { let store = TestStore::new(); @@ -985,9 +1172,10 @@ mod tests { service.complete(*claim, big, 10).unwrap_err(), ServiceError::FrameBound ); - // Still active, never completed: the retry is told it is in flight. + // Still active, never completed, and no longer live: the retry is told the + // outcome is indeterminate. let TransportAdmission::Reply(reply) = - service.begin(&raw_envelope(&invocation), 1_000).unwrap() + service.begin(&raw_envelope(&invocation), 1_000, FRAME).unwrap() else { panic!("expected a refusal"); }; @@ -995,7 +1183,7 @@ mod tests { decrypt_result(&reply, &security(KEY), &invocation) .unwrap() .disposition, - TransportDisposition::Refused(Refusal::InFlight) + TransportDisposition::Refused(Refusal::Indeterminate) ); } }