GameCult/ModelConnector
Watch
9
0
Fork
You've already forked ModelConnector
 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

lease-persona-callers: a lease caller names its turn kind; Persona and Self of one caller lease side by side

Browse source 
LeaseTurn {SelfRun, PersonaTurn} on LeaseCaller and LeaseCallerConfig; the unit derives from the turn; Busy is keyed (caller, turn); a Persona invocation leases in sequence as lease:<invocation>:<n>, one open at a time. config add-lease-caller and init take --turn self|persona.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
Meta Vens 2026-10-31 01:56:34 +01:00
commit 13cbdaf23c
7 changed files with 372 additions and 45 deletions
Download patch file Download diff file Expand all files Collapse all files

15
crates/claude-connector/src/daemon.rs
View file

@ -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@<unit_instance>.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@<unit_instance>`
/// or `mind-persona@<unit_instance>`, 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<Vec<LeaseCaller>,
.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,
})

21
crates/claude-connector/src/lease.rs
View file

@ -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<Leas
/// The most a pipe takes in one atomic write.
const PIPE_BUF: usize = 4096;
/// A unix account allowed to lease, the Self unit its process must run in,
/// and the caller whose spend window its leases reserve in.
/// A unix account allowed to lease, the unit its process must run in (the
/// Self or the Persona unit of its instance, by `turn`), and the caller whose
/// spend window its leases reserve in.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct LeaseCaller {
pub uid: u32,
pub caller_id: String,
/// The `<instance>` of `mind-self@<instance>.service`.
/// Which kind of turn this account leases for; it picks the unit.
pub turn: LeaseTurn,
/// The `<instance>` of `mind-self@<instance>.service` or
/// `mind-persona@<instance>.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,

19
crates/claude-connector/src/main.rs
View file

@ -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<String> = 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<LeaseTurn, Failure> {
match self.text("--turn")?.as_str() {
"self" => Ok(LeaseTurn::SelfRun),
"persona" => Ok(LeaseTurn::PersonaTurn),
_ => Err(Failure::Malformed("--turn")),
}
}
fn number_or<T: std::str::FromStr>(
&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")?,
});

85
crates/claude-connector/tests/config_document.rs
View file

@ -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);
}

168
crates/claude-connector/tests/lease.rs
View file

@ -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(_)
));
}

10
crates/model-connector/src/service.rs
View file

@ -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<LeaseTurn>,
}
#[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 {

99
crates/model-connector/src/service/lease.rs
View file

@ -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>`, `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:<invocation>`, and one Self lease is open per
/// caller. A Persona turn takes the next free sequence of its invocation
/// as `lease:<invocation>:<n>`, 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::<ReplayRecord>(&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::<ReplayRecord>(&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::<ReplayRecord>(&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);
}