diff --git a/crates/claude-connector/src/daemon.rs b/crates/claude-connector/src/daemon.rs index 0e3c5e8..d37d833 100644 --- a/crates/claude-connector/src/daemon.rs +++ b/crates/claude-connector/src/daemon.rs @@ -11,7 +11,8 @@ use std::time::Duration; use cultcache_rs::{CultCache, DatabaseEntry, SingleFileMessagePackBackingStore}; use model_connector::{ - CallerAdmission, CallerConfig, MIN_PAYLOAD_BYTES, ModelPrice, ProviderBackend, ServeError, + CallerAdmission, CallerConfig, LeaseTurn, MIN_PAYLOAD_BYTES, ModelPrice, ProviderBackend, + ServeError, ServeLimits, ServiceError, TransportService, load_admissions, serve_service, }; use serde::{Deserialize, Serialize}; @@ -29,14 +30,17 @@ use crate::{ const CONFIG_KEY: &str = "runtime"; const MODEL_PREFIX: &str = "claude-"; -/// A unix account that may lease the subscription token for a Self run, the -/// Self unit (`mind-self@.service`) its process must run in, -/// the configured caller whose spend window its leases reserve in, and what -/// each lease reserves there. +/// A unix account that may lease the subscription token for a Self run or a +/// Persona turn, the unit its process must run in (`mind-self@` +/// or `mind-persona@`, by `turn`), the configured caller whose +/// spend window its leases reserve in, and what each lease reserves there. #[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] pub struct LeaseCallerConfig { pub unix_user: String, pub caller_id: String, + /// The kind of turn this account leases for: it fixes the unit, `mind-self@` + /// or `mind-persona@`, its process must run in. + pub turn: LeaseTurn, pub unit_instance: String, pub reservation_micro_usd: u64, } @@ -504,6 +508,7 @@ fn lease_callers_of(config: &ClaudeConnectorConfig) -> Result, .map(|uid| LeaseCaller { uid, caller_id: lease.caller_id.clone(), + turn: lease.turn, unit_instance: lease.unit_instance.clone(), reservation_micro_usd: lease.reservation_micro_usd, }) diff --git a/crates/claude-connector/src/lease.rs b/crates/claude-connector/src/lease.rs index d8badf3..9235a17 100644 --- a/crates/claude-connector/src/lease.rs +++ b/crates/claude-connector/src/lease.rs @@ -28,7 +28,7 @@ use std::thread; use std::time::{Duration, Instant, SystemTime, UNIX_EPOCH}; use model_connector::{ - LeaseAdmission, LeaseGrant, LeaseRefusal, LeaseSettlement, LeaseUse, ModelUsage, + LeaseAdmission, LeaseGrant, LeaseRefusal, LeaseSettlement, LeaseTurn, LeaseUse, ModelUsage, TransportService, }; use serde::{Deserialize, Serialize}; @@ -179,13 +179,17 @@ fn read_message(stream: &mut impl Bounded, deadline: Instant) -> io::Result` of `mind-self@.service`. + /// Which kind of turn this account leases for; it picks the unit. + pub turn: LeaseTurn, + /// The `` of `mind-self@.service` or + /// `mind-persona@.service`. pub unit_instance: String, /// What each of this caller's leases reserves, whatever the run asks. pub reservation_micro_usd: u64, @@ -194,7 +198,10 @@ pub struct LeaseCaller { impl LeaseCaller { /// The one unit a lease to this caller may be requested from. pub fn unit(&self) -> String { - format!("mind-self@{}.service", self.unit_instance) + match self.turn { + LeaseTurn::SelfRun => format!("mind-self@{}.service", self.unit_instance), + LeaseTurn::PersonaTurn => format!("mind-persona@{}.service", self.unit_instance), + } } } @@ -400,7 +407,8 @@ struct Lender { /// Serves leases on `listener` until it fails. Each connection is one lease: /// request, then grant or refusal, then (after a grant) settle. A caller is /// the account whose uid the kernel reports for the connection, and only a -/// process in that caller's Self unit may lease, once per unit invocation; no +/// process in that caller's unit may lease, once per unit invocation for a Self +/// run and once per turn for a Persona; no /// field of any message names either. pub fn serve_leases( listener: &UnixListener, @@ -485,6 +493,7 @@ impl Lender { .map_err(|_| io::Error::other("service lock"))? .begin_lease( &caller.caller_id, + caller.turn, &invocation.invocation_id, caller.reservation_micro_usd, request.expires_at_unix_ms, diff --git a/crates/claude-connector/src/main.rs b/crates/claude-connector/src/main.rs index e7f28bc..ac435a1 100644 --- a/crates/claude-connector/src/main.rs +++ b/crates/claude-connector/src/main.rs @@ -18,7 +18,7 @@ use claude_connector::{ ClaudeConnectorConfig, ConfigError, LeaseCallerConfig, MIN_SECRET_BYTES, check_ready, load_config, write_config, }; -use model_connector::{CallerConfig, ModelPrice, Rates, frame_bytes_for_payload}; +use model_connector::{CallerConfig, LeaseTurn, ModelPrice, Rates, frame_bytes_for_payload}; fn main() -> ExitCode { let args: Vec = std::env::args().skip(1).collect(); @@ -95,7 +95,7 @@ const CALLER: [&str; 7] = [ "--max-output-tokens", ]; -const INIT: [&str; 21] = [ +const INIT: [&str; 22] = [ "--listen", "--replay-store", "--claude-binary", @@ -115,12 +115,14 @@ const INIT: [&str; 21] = [ "--socket-timeout-ms", "--lease-socket", "--lease-max-ms", + "--turn", "--lease-unit-instance", "--lease-reservation-micro-usd", ]; -const LEASE_CALLER: [&str; 3] = [ +const LEASE_CALLER: [&str; 4] = [ "--caller-id", + "--turn", "--lease-unit-instance", "--lease-reservation-micro-usd", ]; @@ -180,6 +182,15 @@ impl Flags { .map_err(|_| Failure::Malformed(name)) } + /// The kind of turn a leasing account serves: `self` or `persona`. + fn turn(&self) -> Result { + match self.text("--turn")?.as_str() { + "self" => Ok(LeaseTurn::SelfRun), + "persona" => Ok(LeaseTurn::PersonaTurn), + _ => Err(Failure::Malformed("--turn")), + } + } + fn number_or( &self, name: &'static str, @@ -328,6 +339,7 @@ fn init(path: &str, args: &[&str]) -> Result<(), Failure> { Ok(LeaseCallerConfig { unix_user: unix_user.clone(), caller_id: first.caller_runtime_id.clone(), + turn: flags.turn()?, unit_instance: flags.text("--lease-unit-instance")?, reservation_micro_usd: flags.number("--lease-reservation-micro-usd")?, }) @@ -398,6 +410,7 @@ fn add_lease_caller(path: &str, args: &[&str]) -> Result<(), Failure> { config.lease_callers.push(LeaseCallerConfig { unix_user: flags.text("--lease-user")?, caller_id: flags.text("--caller-id")?, + turn: flags.turn()?, unit_instance: flags.text("--lease-unit-instance")?, reservation_micro_usd: flags.number("--lease-reservation-micro-usd")?, }); diff --git a/crates/claude-connector/tests/config_document.rs b/crates/claude-connector/tests/config_document.rs index 3659113..8ca6828 100644 --- a/crates/claude-connector/tests/config_document.rs +++ b/crates/claude-connector/tests/config_document.rs @@ -15,7 +15,7 @@ use cultcache_rs::{ CacheBackingStore, CultCache, CultCacheEnvelope, DatabaseEntry, SingleFileMessagePackBackingStore, }; -use model_connector::{CallerConfig, MIN_PAYLOAD_BYTES, ModelPrice, Rates}; +use model_connector::{CallerConfig, LeaseTurn, MIN_PAYLOAD_BYTES, ModelPrice, Rates}; const CANARY: &str = "canary-9f3b7c1d"; const OPUS: &str = "claude-opus-5-1"; @@ -99,6 +99,7 @@ fn valid() -> ClaudeConnectorConfig { lease_callers: vec![LeaseCallerConfig { unix_user: "mind-eureka-self".to_string(), caller_id: "persona-a-caller".to_string(), + turn: LeaseTurn::SelfRun, unit_instance: "eureka".to_string(), reservation_micro_usd: 5_000_000, }], @@ -1106,6 +1107,8 @@ fn config_init_binds_each_lease_user_to_the_configured_caller() { "mind-eureka-self", "--lease-max-ms", "120000", + "--turn", + "self", "--lease-unit-instance", "eureka", "--lease-reservation-micro-usd", @@ -1120,6 +1123,7 @@ fn config_init_binds_each_lease_user_to_the_configured_caller() { [LeaseCallerConfig { unix_user: "mind-eureka-self".to_string(), caller_id: "persona-a-caller".to_string(), + turn: LeaseTurn::SelfRun, unit_instance: "eureka".to_string(), reservation_micro_usd: 5_000_000, }] @@ -1139,6 +1143,8 @@ fn config_check_refuses_a_lease_account_the_system_does_not_have() { socket.to_str().unwrap(), "--lease-user", "no-such-account-4f1c", + "--turn", + "self", "--lease-unit-instance", "eureka", "--lease-reservation-micro-usd", @@ -1185,6 +1191,8 @@ fn serve_binds_the_lease_socket_and_answers_a_stranger_with_a_refusal() { socket.to_str().unwrap(), "--lease-user", "nobody", + "--turn", + "self", "--lease-unit-instance", "eureka", "--lease-reservation-micro-usd", @@ -1304,6 +1312,7 @@ fn a_lease_caller_bills_one_configured_caller_within_that_callers_ceiling() { let lease = |user: &str, caller: &str, instance: &str, reservation: u64| LeaseCallerConfig { unix_user: user.to_string(), caller_id: caller.to_string(), + turn: LeaseTurn::SelfRun, unit_instance: instance.to_string(), reservation_micro_usd: reservation, }; @@ -1339,6 +1348,8 @@ fn add_caller_and_add_lease_caller_extend_the_document_and_grow_the_serve_limits "/run/claude-connector/lease.sock", "--lease-user", "mind-eureka-self", + "--turn", + "self", "--lease-unit-instance", "eureka", "--lease-reservation-micro-usd", @@ -1386,6 +1397,8 @@ fn add_caller_and_add_lease_caller_extend_the_document_and_grow_the_serve_limits "mind-heartbeat", "--caller-id", "heartbeat", + "--turn", + "self", "--lease-unit-instance", "heartbeat", "--lease-reservation-micro-usd", @@ -1446,6 +1459,8 @@ fn add_caller_and_add_lease_caller_extend_the_document_and_grow_the_serve_limits "mind-heartbeat", "--caller-id", "heartbeat", + "--turn", + "self", "--lease-unit-instance", "heartbeat", "--lease-reservation-micro-usd", @@ -1472,6 +1487,8 @@ fn add_lease_caller_needs_a_document_that_lends_a_token() { "mind-heartbeat", "--caller-id", "persona-a-caller", + "--turn", + "self", "--lease-unit-instance", "heartbeat", "--lease-reservation-micro-usd", @@ -1668,3 +1685,69 @@ fn config_init_keeps_an_existing_secret_file_only_if_it_is_usable() { assert_eq!(std::fs::read_to_string(&secret).unwrap(), "k".repeat(32)); assert!(scratch.path("connector.cc").exists()); } + +#[test] +fn a_persona_lease_row_validates_and_an_account_leases_for_one_turn_only() { + let mut config = valid(); + let mut persona = config.lease_callers[0].clone(); + persona.unix_user = "mind-eureka-persona".to_string(); + persona.turn = LeaseTurn::PersonaTurn; + config.lease_callers.push(persona.clone()); + config.validate().unwrap(); + // The same account under the other turn is a second row for one uid. + config.lease_callers[1].unix_user = config.lease_callers[0].unix_user.clone(); + assert_eq!(refused(&config), "lease_callers"); +} + +#[test] +fn add_lease_caller_takes_the_turn_and_refuses_an_unknown_one() { + let scratch = Scratch::new(); + let made = init( + &scratch, + &init_flags( + &scratch, + &[ + "--lease-socket", + "/run/claude-connector/lease.sock", + "--lease-user", + "mind-eureka-self", + "--turn", + "self", + "--lease-unit-instance", + "eureka", + "--lease-reservation-micro-usd", + "5000000", + ], + ), + ); + assert!(made.status.success(), "{}", shown(&made)); + let path = scratch.path("connector.cc"); + let add = |turn: &str| { + run( + &[ + "config", + "add-lease-caller", + path.to_str().unwrap(), + "--lease-user", + "mind-eureka-persona", + "--caller-id", + "persona-a-caller", + "--turn", + turn, + "--lease-unit-instance", + "eureka", + "--lease-reservation-micro-usd", + "1000000", + ], + &[], + ) + }; + let unknown = add("heartbeat"); + assert!(!unknown.status.success()); + assert_eq!(load_config(&path).unwrap().lease_callers.len(), 1); + let added = add("persona"); + assert!(added.status.success(), "{}", shown(&added)); + let config = load_config(&path).unwrap(); + assert_eq!(config.lease_callers[1].turn, LeaseTurn::PersonaTurn); + assert_eq!(config.lease_callers[0].turn, LeaseTurn::SelfRun); +} diff --git a/crates/claude-connector/tests/lease.rs b/crates/claude-connector/tests/lease.rs index 72fe188..f6a2b7e 100644 --- a/crates/claude-connector/tests/lease.rs +++ b/crates/claude-connector/tests/lease.rs @@ -22,7 +22,8 @@ use claude_connector::{ decode_frame, encode_frame, serve_leases, }; use model_connector::{ - CallerAdmission, LeaseAdmission, LeaseRefusal, ModelPrice, ModelUsage, Rates, SpendCeiling, + CallerAdmission, LeaseAdmission, LeaseRefusal, LeaseTurn, LeaseUse, ModelPrice, ModelUsage, Rates, + SpendCeiling, SpendWindowMs, TransportService, Usage, }; use zeroize::Zeroizing; @@ -147,6 +148,7 @@ struct Setup { ceiling: u64, reservation: u64, uid: u32, + turn: LeaseTurn, token: Option<&'static str>, max_connections: usize, } @@ -156,6 +158,7 @@ fn setup(ceiling: u64, reservation: u64) -> Setup { ceiling, reservation, uid: own_uid(), + turn: LeaseTurn::SelfRun, token: Some(TOKEN), max_connections: 8, } @@ -218,6 +221,7 @@ fn start_with( let caller = LeaseCaller { uid: setup.uid, caller_id: CALLER.to_string(), + turn: setup.turn, unit_instance: instance.to_string(), reservation_micro_usd: setup.reservation, }; @@ -411,7 +415,7 @@ fn probe_reserve(harness: &Harness, name: &str, cost: u64) -> LeaseAdmission { .service .lock() .unwrap() - .begin_lease(CALLER, name, cost, FAR, now_ms()) + .begin_lease(CALLER, LeaseTurn::SelfRun, name, cost, FAR, now_ms()) .unwrap() } @@ -539,7 +543,7 @@ fn the_reservation_is_claimed_before_the_token_is_read_or_lent() { let admission = service .lock() .unwrap() - .begin_lease(CALLER, "probe", 1, FAR, now_ms()) + .begin_lease(CALLER, LeaseTurn::SelfRun, "probe", 1, FAR, now_ms()) .unwrap(); *sink.lock().unwrap() = Some(admission); })); @@ -1164,3 +1168,161 @@ fn the_unit_is_asked_of_the_callers_unit_through_the_peers_handle() { vec![(std::process::id().to_string(), UNIT.to_string())] ); } + +const PERSONA_UNIT: &str = "mind-persona@eureka.service"; + +/// A listener whose one lease caller is a Persona account; peers start in +/// invocation 1 of the Persona unit. +fn start_persona(setup: Setup) -> Harness { + let units = Arc::new(FakeUnits { + peer: Mutex::new(None), + asked_about: Mutex::new(Vec::new()), + }); + let harness = start_with( + Setup { + turn: LeaseTurn::PersonaTurn, + ..setup + }, + units.clone(), + Some(units), + INSTANCE, + ); + harness.run_in(PERSONA_UNIT, 1); + harness +} + +fn lease_on( + harness: &Harness, + turn: LeaseTurn, + invocation: &str, +) -> LeaseAdmission { + harness + .service + .lock() + .unwrap() + .begin_lease(CALLER, turn, invocation, 100, FAR, now_ms()) + .unwrap() +} + +fn lease_id(admission: &LeaseAdmission) -> String { + match admission { + LeaseAdmission::Granted(grant) => grant.lease_id.clone(), + other => panic!("expected a grant, got {other:?}"), + } +} + +fn busy() -> LeaseAdmission { + LeaseAdmission::Refused(LeaseRefusal::Busy) +} + +#[test] +fn persona_and_self_of_one_caller_lease_together() { + let harness = start(setup(1_000_000, 100)); + let own = lease_on(&harness, LeaseTurn::SelfRun, "self-run"); + assert_eq!(lease_id(&own), "lease:self-run"); + let persona = lease_on(&harness, LeaseTurn::PersonaTurn, "persona-run"); + assert_eq!(lease_id(&persona), "lease:persona-run:1"); + // Each turn kind still holds one: a second Self run is busy while the + // first is open, a second Persona lease of the same invocation too. + assert_eq!(lease_on(&harness, LeaseTurn::SelfRun, "self-run-2"), busy()); + assert_eq!( + lease_on(&harness, LeaseTurn::PersonaTurn, "persona-run"), + busy() + ); + // Settling the Persona lease frees nothing of the Self lease's. + harness + .service + .lock() + .unwrap() + .settle_lease(CALLER, "lease:persona-run:1", LeaseUse::Reservation) + .unwrap(); + assert_eq!(lease_on(&harness, LeaseTurn::SelfRun, "self-run-2"), busy()); +} + +#[test] +fn persona_invocation_leases_sequentially() { + let harness = start(setup(1_000_000, 100)); + for expected in 1..=3 { + let granted = lease_on(&harness, LeaseTurn::PersonaTurn, "persona-run"); + let id = lease_id(&granted); + assert_eq!(id, format!("lease:persona-run:{expected}")); + harness + .service + .lock() + .unwrap() + .settle_lease(CALLER, &id, LeaseUse::Reservation) + .unwrap(); + } + // Another invocation counts from one: the sequence is per invocation. + let other = lease_on(&harness, LeaseTurn::PersonaTurn, "persona-other"); + assert_eq!(lease_id(&other), "lease:persona-other:1"); +} + +#[test] +fn persona_invocation_second_open_lease_busy() { + let harness = start_persona(setup(1_000_000, 100)); + let first = ask(&harness, 60_000); + assert!(granted_ok(&first)); + let second = ask(&harness, 60_000); + assert_eq!(refusal_of(&second), Some(LeaseRefusalKind::Busy)); + assert!(second.fds.is_empty()); + assert_eq!(harness.tokens.reads.load(Ordering::SeqCst), 1); +} + +#[test] +fn persona_turns_of_a_live_invocation_each_lend_a_token_over_the_socket() { + let harness = start_persona(setup(1_000_000, 100)); + for _ in 0..2 { + let mut answer = ask(&harness, 60_000); + assert!(granted_ok(&answer)); + settle(&mut answer, vec![used(MODEL, 10)]); + } + assert_eq!(harness.tokens.reads.load(Ordering::SeqCst), 2); +} + +#[test] +fn self_invocation_still_leases_once() { + let harness = start(setup(1_000_000, 100)); + let id = lease_id(&lease_on(&harness, LeaseTurn::SelfRun, "self-run")); + harness + .service + .lock() + .unwrap() + .settle_lease(CALLER, &id, LeaseUse::Reservation) + .unwrap(); + assert_eq!(lease_on(&harness, LeaseTurn::SelfRun, "self-run"), busy()); +} + +#[test] +fn persona_row_refuses_mind_self_unit() { + let harness = start_persona(setup(1_000_000, 100)); + harness.run_in(UNIT, 1); + let answer = ask(&harness, 60_000); + assert_eq!(refusal_of(&answer), Some(LeaseRefusalKind::UnknownCaller)); + assert!(answer.fds.is_empty()); + assert_eq!(harness.tokens.reads.load(Ordering::SeqCst), 0); + // And a Self row refuses the Persona unit. + let harness = start(setup(1_000_000, 100)); + harness.run_in(PERSONA_UNIT, 1); + let answer = ask(&harness, 60_000); + assert_eq!(refusal_of(&answer), Some(LeaseRefusalKind::UnknownCaller)); + assert_eq!(harness.tokens.reads.load(Ordering::SeqCst), 0); +} + +#[test] +fn a_persona_lease_reserves_and_settles_in_the_callers_window() { + let harness = start(setup(1_000, 600)); + let id = lease_id(&lease_on(&harness, LeaseTurn::PersonaTurn, "persona-run")); + // 100 reserved by the helper; a Self probe of 901 cannot fit, 900 can. + assert_eq!(probe_reserve(&harness, "p1", 901), ceiling_refused()); + harness + .service + .lock() + .unwrap() + .settle_lease(CALLER, &id, LeaseUse::NeverLent) + .unwrap(); + assert!(matches!( + probe_reserve(&harness, "p2", 1_000), + LeaseAdmission::Granted(_) + )); +} diff --git a/crates/model-connector/src/service.rs b/crates/model-connector/src/service.rs index 8adaf07..ea3dbeb 100644 --- a/crates/model-connector/src/service.rs +++ b/crates/model-connector/src/service.rs @@ -337,7 +337,7 @@ struct LiveRequest { reserved_micro_usd: u64, /// A Self run's credential lease, not a provider call: it holds its /// reservation in the window like a call does but takes no request slot. - lease: bool, + lease: Option, } #[derive(Debug, Clone, PartialEq, Eq, Hash)] @@ -846,7 +846,7 @@ impl TransportService { .live .iter() .filter(|(key, request)| { - !request.lease && key.caller_runtime_id == invocation.caller_runtime_id + request.lease.is_none() && key.caller_runtime_id == invocation.caller_runtime_id }) .count(); if active_for_caller >= max_concurrent_requests { @@ -909,7 +909,7 @@ impl TransportService { LiveRequest { window, reserved_micro_usd, - lease: false, + lease: None, }, ); Ok(TransportAdmission::Execute(Box::new(ExecutionClaim { @@ -1272,7 +1272,9 @@ fn replay_store_error(error: impl std::fmt::Display) -> ServiceError { } mod lease; -pub use lease::{LeaseAdmission, LeaseGrant, LeaseRefusal, LeaseSettlement, LeaseUse, ModelUsage}; +pub use lease::{ + LeaseAdmission, LeaseGrant, LeaseRefusal, LeaseSettlement, LeaseTurn, LeaseUse, ModelUsage, +}; fn contract_refusal(error: &ContractError) -> Refusal { match error { diff --git a/crates/model-connector/src/service/lease.rs b/crates/model-connector/src/service/lease.rs index 6e9be71..d6e01e5 100644 --- a/crates/model-connector/src/service/lease.rs +++ b/crates/model-connector/src/service/lease.rs @@ -1,4 +1,4 @@ -//! A credential lease: a Self run's reservation in the caller's spend window. +//! A credential lease: a Self run's or a Persona turn's reservation in the caller's spend window. //! The lease is one record of the same ledger a provider call reserves in, not //! a second ledger: it is claimed before anything is lent, counts against the //! window's ceiling while open, and settles to the true cost of what the run @@ -12,6 +12,22 @@ use super::*; /// The request-id namespace of leases in a caller's replay records. const LEASE_PREFIX: &str = "lease:"; +/// The most leases one Persona invocation may take in its life. +const MAX_PERSONA_LEASES: u32 = 1024; + +/// Room kept in an identity for `:`, `n` at most `MAX_PERSONA_LEASES`. +const SEQUENCE_SUFFIX_BYTES: usize = 5; + +/// Which kind of turn a lease serves. It is a property of the lease caller's +/// configured row, never of a request. A Self run leases once per unit +/// invocation; a Persona turn leases once per turn, so one live invocation +/// leases in sequence, one open lease at a time. +#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)] +pub enum LeaseTurn { + SelfRun, + PersonaTurn, +} + /// What an admitted lease holds. #[derive(Debug, Clone, PartialEq, Eq)] pub struct LeaseGrant { @@ -77,7 +93,8 @@ pub enum LeaseRefusal { /// free pass. CostBound, UnknownCaller, - /// The caller holds a lease already, or this run id was leased before. + /// A Self run: the caller holds a Self lease already, or this invocation + /// leased before. A Persona turn: this invocation holds a lease now. Busy, } @@ -89,11 +106,17 @@ pub enum LeaseAdmission { impl TransportService { /// Reserves `max_cost_micro_usd` in `caller_runtime_id`'s current window - /// for the unit invocation `invocation_id`, which leases once. A ledger - /// that cannot be written answers `Err` and reserves nothing. A lease takes no request slot. + /// for the unit invocation `invocation_id`. A Self run leases once per + /// invocation, as `lease:`, and one Self lease is open per + /// caller. A Persona turn takes the next free sequence of its invocation + /// as `lease::`, one open at a time; a Persona lease never + /// blocks or is blocked by the caller's Self lease. A ledger that cannot + /// be written answers `Err` and reserves nothing. A lease takes no + /// request slot. pub fn begin_lease( &mut self, caller_runtime_id: &str, + turn: LeaseTurn, invocation_id: &str, max_cost_micro_usd: u64, expires_at_unix_ms: u64, @@ -106,30 +129,60 @@ impl TransportService { let spend_window_ms = caller.spend_window_ms; let connection_key_epoch = caller.connection_key_epoch; require_id(invocation_id, "invocation_id")?; - if invocation_id.len() + LEASE_PREFIX.len() > MAX_IDENTITY_BYTES { + if invocation_id.len() + LEASE_PREFIX.len() + SEQUENCE_SUFFIX_BYTES > MAX_IDENTITY_BYTES { return Err(ContractError::Invalid("invocation_id").into()); } if max_cost_micro_usd == 0 { return Ok(LeaseAdmission::Refused(LeaseRefusal::CostBound)); } - let identity = RequestIdentity { - caller_runtime_id: caller_runtime_id.to_string(), - request_id: format!("{LEASE_PREFIX}{invocation_id}"), + let identity = match turn { + LeaseTurn::SelfRun => { + let identity = RequestIdentity { + caller_runtime_id: caller_runtime_id.to_string(), + request_id: format!("{LEASE_PREFIX}{invocation_id}"), + }; + let holding = self.live.iter().any(|(held, request)| { + request.lease == Some(LeaseTurn::SelfRun) + && held.caller_runtime_id == caller_runtime_id + }); + if holding + || self + .replay_store + .get::(&replay_key(&identity)) + .map_err(replay_store_error)? + .is_some() + { + return Ok(LeaseAdmission::Refused(LeaseRefusal::Busy)); + } + identity + } + LeaseTurn::PersonaTurn => { + let mut free = None; + for sequence in 1..=MAX_PERSONA_LEASES { + let identity = RequestIdentity { + caller_runtime_id: caller_runtime_id.to_string(), + request_id: format!("{LEASE_PREFIX}{invocation_id}:{sequence}"), + }; + if self.live.contains_key(&identity) { + return Ok(LeaseAdmission::Refused(LeaseRefusal::Busy)); + } + if self + .replay_store + .get::(&replay_key(&identity)) + .map_err(replay_store_error)? + .is_none() + { + free = Some(identity); + break; + } + } + let Some(identity) = free else { + return Ok(LeaseAdmission::Refused(LeaseRefusal::Busy)); + }; + identity + } }; let key = replay_key(&identity); - let holding = self - .live - .iter() - .any(|(held, request)| request.lease && held.caller_runtime_id == caller_runtime_id); - if holding - || self - .replay_store - .get::(&key) - .map_err(replay_store_error)? - .is_some() - { - return Ok(LeaseAdmission::Refused(LeaseRefusal::Busy)); - } let admitted_at_unix_ms = admission_clock( &self.window_spend, caller_runtime_id, @@ -185,7 +238,7 @@ impl TransportService { LiveRequest { window, reserved_micro_usd: max_cost_micro_usd, - lease: true, + lease: Some(turn), }, ); Ok(LeaseAdmission::Granted(LeaseGrant { @@ -213,7 +266,7 @@ impl TransportService { if !self .live .get(&identity) - .is_some_and(|request| request.lease) + .is_some_and(|request| request.lease.is_some()) { return Err(ServiceError::NoActiveRequest); }