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 <noreply@anthropic.com>
This commit is contained in:
Meta Vens
parent
bf07b2cd38
commit
2d383d7818
4 changed files with 372 additions and 84 deletions
|
|
@ -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.
|
||||
|
|
|
|||
|
|
@ -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<usize> {
|
||||
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}");
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -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)
|
||||
}
|
||||
|
|
|
|||
|
|
@ -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<TransportAdmission, ServiceError> {
|
||||
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<TransportEnvelope, ServiceError> {
|
||||
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<ExecutionClaim>, 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)
|
||||
);
|
||||
}
|
||||
}
|
||||
|
|
|
|||
Loading…
Reference in a new issue