diff --git a/crates/model-connector/src/contract.rs b/crates/model-connector/src/contract.rs index 5953b8e..f22b8f9 100644 --- a/crates/model-connector/src/contract.rs +++ b/crates/model-connector/src/contract.rs @@ -573,6 +573,12 @@ pub enum Refusal { // exchange, where dropping it would break the provider's own rules. A plain // comment: a doc comment would split the published schema into a union. ForeignReasoningState, + // The caller's newest admission window starts after the clock's reading, + // and the request does not fit what that window holds: the host's clock + // is behind the window and the caller stays refused until it catches up. + // A plain comment: a doc comment would split the published schema into a + // union. + ClockBehindAdmissionWindow, } /// The one wire codec of every v3 document: named MessagePack (struct fields diff --git a/crates/model-connector/src/service.rs b/crates/model-connector/src/service.rs index 76aebc1..2cb85d5 100644 --- a/crates/model-connector/src/service.rs +++ b/crates/model-connector/src/service.rs @@ -735,7 +735,14 @@ impl TransportService { .saturating_add(reserved_micro_usd) > spend_ceiling { - return self.reply_refusal(&invocation, Refusal::SpendCeiling, &security); + // A clock reading behind the caller's newest window is named: the + // refusal holds until the clock reaches that window's start. + let refusal = if now_unix_ms < window.saturating_mul(spend_window_ms) { + Refusal::ClockBehindAdmissionWindow + } else { + Refusal::SpendCeiling + }; + return self.reply_refusal(&invocation, refusal, &security); } // Reserving and claiming the request are one write: a ledger that @@ -974,6 +981,19 @@ impl TransportService { && receipt.overrun_micro_usd == claim.call.overrun_of(receipt.cost_micro_usd) } + /// Where `caller_runtime_id`'s newest admission window starts, in unix + /// milliseconds, or `None` when the caller has none: the instant the host + /// clock must reach before a `ClockBehindAdmissionWindow` refusal can end. + pub fn admission_window_start_unix_ms(&self, caller_runtime_id: &str) -> Option { + let spend_window_ms = self.callers.get(caller_runtime_id)?.spend_window_ms; + let newest = self + .window_spend + .get(caller_runtime_id)? + .keys() + .next_back()?; + Some(newest.saturating_mul(spend_window_ms)) + } + fn reply_refusal( &self, invocation: &TransportInvocation, @@ -3242,7 +3262,7 @@ mod tests { assert_eq!(run_window(&mut service, "again", 20_100), 0); assert_eq!( refusal_at(&mut service, "last", 20_200, MAX_COST), - Refusal::SpendCeiling + Refusal::ClockBehindAdmissionWindow ); } @@ -3479,10 +3499,53 @@ mod tests { // reservation), window 9 does not: the caller's window is the newest. assert_eq!( refusal_at(&mut service, "back", 20_500, MAX_COST), - Refusal::SpendCeiling + Refusal::ClockBehindAdmissionWindow ); assert_eq!(service.window_spend(CALLER, 2), reservation()); assert_eq!(service.window_spend(CALLER, 9), 3 * FIXTURE_USAGE_COST); drop(old); } + + #[test] + fn a_clock_behind_the_callers_window_is_refused_by_name_until_it_catches_up() { + let window = 10_000; + let ceiling = reservation() + 2 * FIXTURE_USAGE_COST; + let store = TestStore::new(); + let mut service = open_limited(&store, ceiling, window); + assert_eq!(service.admission_window_start_unix_ms(CALLER), None); + assert_eq!(run_window(&mut service, "fill", 50_000), 3); + assert_eq!( + service.admission_window_start_unix_ms(CALLER), + Some(50_000) + ); + // Behind the full window: the cause is named, no slot is taken and + // nothing is reserved. + assert_eq!( + refusal_at(&mut service, "behind", 20_000, MAX_COST), + Refusal::ClockBehindAdmissionWindow + ); + assert!(service.live.is_empty()); + assert_eq!(service.window_spend(CALLER, 5), 3 * FIXTURE_USAGE_COST); + // Inside the full window the clock is not behind: an honest ceiling. + assert_eq!( + refusal_at(&mut service, "inside", 50_500, MAX_COST), + Refusal::SpendCeiling + ); + // Once the clock reaches the next window the caller is admitted. + assert_eq!(run_window(&mut service, "later", 60_000), 3); + } + + #[test] + fn the_clock_behind_refusal_is_no_wider_than_the_widest_other_refusal() { + let invocation = invocation_at("request-1", 1_000, MAX_COST); + let sized = |refusal| { + wire_encode(&TransportResult::refused(&invocation, refusal)) + .unwrap() + .len() + }; + assert!( + sized(Refusal::ClockBehindAdmissionWindow) + <= sized(Refusal::ProviderDigestSubstitution) + ); + } } diff --git a/crates/model-connector/tests/published_schemas.rs b/crates/model-connector/tests/published_schemas.rs index afe65e2..15ce197 100644 --- a/crates/model-connector/tests/published_schemas.rs +++ b/crates/model-connector/tests/published_schemas.rs @@ -110,7 +110,8 @@ fn the_refusal_schema_is_one_flat_string_enum() { .iter() .map(|value| value.as_str().unwrap()) .collect(); - assert_eq!(variants.len(), 14); + assert_eq!(variants.len(), 15); + assert!(variants.contains(&"ClockBehindAdmissionWindow")); assert!(variants.contains(&"ForeignReasoningState")); assert!(variants.contains(&"CostBound")); assert!(variants.contains(&"SpendCeiling")); diff --git a/schemas/transport_result.json b/schemas/transport_result.json index e85fd51..39e010f 100644 --- a/schemas/transport_result.json +++ b/schemas/transport_result.json @@ -53,7 +53,8 @@ "Indeterminate", "ReplayConflict", "Malformed", - "ForeignReasoningState" + "ForeignReasoningState", + "ClockBehindAdmissionWindow" ] }, "StopReason": {