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

Call record: ledger carries cap and starting world; replay refuses another world

Browse source 
The run ledger is the one owner of what a run began as: its cap and the
world id and state digest it started from. Replay takes the store, the run
and a restored world and refuses a different world up front with a typed
error naming the field. Restores the decode checks the ledger rules do not
imply, makes the replay key an exhaustive destructure, and pins the
over-bound edge. Not yet built or tested at this commit.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
Meta Vens 2026-10-10 20:32:17 +02:00
commit 2c5eb95da3
4 changed files with 499 additions and 99 deletions
Download patch file Download diff file Expand all files Collapse all files

2
crates/ghostlight/src/lib.rs
View file

@ -133,7 +133,7 @@ pub use table::{
pub use vault::{VaultError, VaultEvidenceSource};
pub use verse::{
CALL_RECORD_SCHEMA, CallRecordError, CallRecordStore, ProviderCallRecord, RecordingInferencePort,
ReplayInferencePort, RunId,
ReplayInferencePort, RunField, RunId, RunStart,
};
use patch::{EdgeRecord, EntityRecord, LedgerDelta, ResolvedOp, ResolvedPatch};

264
crates/ghostlight/src/verse/calls.rs
View file

@ -38,9 +38,9 @@ use std::{
use thiserror::Error;
const CALL_RECORD_ROW: &str = "verse_provider_call.v1";
const RUN_SPEND_ROW: &str = "verse_run_spend.v1";
const RUN_LEDGER_ROW: &str = "verse_run_ledger.v1";
/// The run ledger document's schema.
const RUN_SPEND_SCHEMA: &str = "ghostlight.verse_run_spend.v1";
const RUN_LEDGER_SCHEMA: &str = "ghostlight.verse_run_ledger.v1";
/// The call record document's schema. A store holding any other row type or
/// schema is refused at open: this store is not migrated from, or shared with,
/// another document.
@ -62,6 +62,46 @@ pub enum CallRecordError {
RunInUse,
#[error("the call could pass the run's token cap")]
OverCap,
#[error("the run was recorded with a different {0}")]
RunMismatch(RunField),
}
/// What a run is bound to when it begins, named by the field a mismatch refuses.
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub enum RunField {
/// The token cap the run was opened with.
Cap,
/// The identity of the world the run started from.
WorldId,
/// The state digest of the world the run started from.
StateDigest,
}
impl std::fmt::Display for RunField {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
f.write_str(match self {
Self::Cap => "token cap",
Self::WorldId => "world id",
Self::StateDigest => "world state digest",
})
}
}
/// The world a run began from: its identity and state digest. Replay is of a
/// restored copy of this world, so it takes the same two values.
#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
pub struct RunStart {
world_id: String,
state_digest: String,
}
impl RunStart {
pub fn of(world: &crate::WorldSnapshot) -> Self {
Self {
world_id: world.world_id.text(),
state_digest: world.state_digest.clone(),
}
}
}
/// A run's identity as a key segment.
@ -158,37 +198,62 @@ fn hex(digest: &[u8]) -> String {
/// one world's random identity. Everything else the provider is sent
/// (model, instructions, input, tools, shape) is in the digest.
fn content_digest(request: &CodexProviderRequest) -> Result<String, InferenceFault> {
// Exhaustive on purpose: a field added upstream must be placed in the key
// or left out here by name before this compiles.
let CodexProviderRequest {
schema_id,
request_id: _request_id, // names the call, not what it asks
conversation_id: _conversation_id, // names the call, not what it asks
model,
instructions,
input,
reasoning_effort,
reasoning_summary,
service_tier,
output_format_name,
previous_response_id,
tools,
tool_choice,
parallel_tool_calls,
output_schema_json,
max_output_tokens,
prompt_cache_key,
} = request;
let bytes = rmp_serde::to_vec(&(
&request.schema_id,
&request.model,
&request.instructions,
&request.input,
&request.reasoning_effort,
&request.reasoning_summary,
&request.service_tier,
&request.output_format_name,
&request.previous_response_id,
&request.tools,
&request.tool_choice,
request.parallel_tool_calls,
&request.output_schema_json,
request.max_output_tokens,
&request.prompt_cache_key,
schema_id,
model,
instructions,
input,
reasoning_effort,
reasoning_summary,
service_tier,
output_format_name,
previous_response_id,
tools,
tool_choice,
parallel_tool_calls,
output_schema_json,
max_output_tokens,
prompt_cache_key,
))
.map_err(|_| InferenceFault::new("a provider request could not be measured"))?;
Ok(hex(&Sha256::digest(bytes)))
}
/// A run's persisted ledger: tokens reserved or charged against its cap.
/// A run's persisted ledger, the one owner of what the run began as: the cap
/// it was opened under, the world it started from, and the tokens reserved or
/// charged so far.
#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
struct RunSpend {
struct RunLedger {
run_id: String,
cap: u64,
start: RunStart,
spent: u64,
}
enum DecodedRow {
Call(ProviderCallRecord),
Spend(RunSpend),
Ledger(RunLedger),
}
fn canonical<T: Serialize + for<'de> Deserialize<'de>>(
@ -207,20 +272,22 @@ fn decode_row(row: &CultCacheEnvelope) -> Result<DecodedRow, CallRecordError> {
match (row.r#type.as_str(), row.schema_id.as_deref()) {
(CALL_RECORD_ROW, Some(CALL_RECORD_SCHEMA)) => {
let record: ProviderCallRecord = canonical(row)?;
// The run id and a sequence of zero need no check here: a run id
// that is no key segment has no readable ledger, and a zero
// sequence fails the density rule in `open`.
if row.key != row_key(&record.run_id, record.sequence) {
// A run id that is no whole key segment would key into another
// run's records, and sequences count from 1.
if RunId::new(&record.run_id).is_err()
|| record.sequence == 0
|| row.key != row_key(&record.run_id, record.sequence)
{
return Err(CallRecordError::Corrupt);
}
Ok(DecodedRow::Call(record))
}
(RUN_SPEND_ROW, Some(RUN_SPEND_SCHEMA)) => {
let spend: RunSpend = canonical(row)?;
if RunId::new(&spend.run_id).is_err() || row.key != spend.run_id {
(RUN_LEDGER_ROW, Some(RUN_LEDGER_SCHEMA)) => {
let ledger: RunLedger = canonical(row)?;
if RunId::new(&ledger.run_id).is_err() || row.key != ledger.run_id {
return Err(CallRecordError::Corrupt);
}
Ok(DecodedRow::Spend(spend))
Ok(DecodedRow::Ledger(ledger))
}
_ => Err(CallRecordError::Corrupt),
}
@ -229,8 +296,8 @@ fn decode_row(row: &CultCacheEnvelope) -> Result<DecodedRow, CallRecordError> {
#[derive(Default)]
struct RunTally {
last_sequence: u64,
/// Tokens reserved or charged: the persisted ledger, read at open.
spent: u64,
/// The persisted ledger, read at open and written where it is spent.
ledger: Option<RunLedger>,
}
struct StoreState {
@ -271,18 +338,23 @@ impl CallRecordStore {
let tally = runs.entry(record.run_id).or_default();
tally.last_sequence = tally.last_sequence.max(record.sequence);
}
DecodedRow::Spend(spend) => {
runs.entry(spend.run_id).or_default().spent = spend.spent;
DecodedRow::Ledger(ledger) => {
runs.entry(ledger.run_id.clone()).or_default().ledger = Some(ledger);
}
}
}
// Keys are unique, so a run is dense from 1 exactly when its rows
// number its highest sequence. Every record was reserved before it was
// made, so the ledger covers what the records charged.
// Sequences start at 1 and keys are unique, so a run is dense exactly
// when its rows number its highest sequence. Every record was reserved
// in a ledger before it was made, so a run with records has a ledger
// that covers what they charged.
for (run, tally) in &runs {
let rows = counts.get(run).copied().unwrap_or(0);
let recorded = charged.get(run).copied().unwrap_or(0);
if rows != tally.last_sequence || tally.spent < recorded {
let covered = tally
.ledger
.as_ref()
.is_some_and(|ledger| ledger.spent >= recorded);
if rows != tally.last_sequence || !covered {
return Err(CallRecordError::Corrupt);
}
}
@ -315,16 +387,46 @@ impl CallRecordStore {
.lock()?
.runs
.get(run.as_str())
.map_or(0, |tally| tally.spent))
.and_then(|tally| tally.ledger.as_ref())
.map_or(0, |ledger| ledger.spent))
}
/// Takes the run's ledger for one port, or refuses when another holds it.
fn own(&self, run: &RunId) -> Result<(), CallRecordError> {
if self.lock()?.owned.insert(run.as_str().to_owned()) {
Ok(())
} else {
Err(CallRecordError::RunInUse)
/// The first claim of a run writes its ledger: the cap it is opened under
/// and the world it starts from. A later claim must name the same cap; the
/// start is not compared, because a resumed world has moved on.
fn claim(&self, run: &RunId, cap: u64, start: &RunStart) -> Result<(), CallRecordError> {
let mut state = self.lock()?;
if !state.owned.insert(run.as_str().to_owned()) {
return Err(CallRecordError::RunInUse);
}
let begun = match state.runs.get(run.as_str()).and_then(|t| t.ledger.as_ref()) {
Some(ledger) if ledger.cap == cap => Ok(()),
Some(_) => Err(CallRecordError::RunMismatch(RunField::Cap)),
None => Self::write_ledger(
&mut state,
RunLedger {
run_id: run.as_str().to_owned(),
cap,
start: start.clone(),
spent: 0,
},
),
};
if begun.is_err() {
state.owned.remove(run.as_str());
}
begun
}
/// The world a run began from, when it has begun.
fn start_of(&self, run: &RunId) -> Result<Option<RunStart>, CallRecordError> {
Ok(self
.lock()?
.runs
.get(run.as_str())
.and_then(|tally| tally.ledger.as_ref())
.map(|ledger| ledger.start.clone()))
}
fn disown(&self, run: &RunId) {
@ -333,32 +435,46 @@ impl CallRecordStore {
}
}
/// Persists `spent` as the run's ledger, then adopts it in memory. A write
/// that fails leaves the previous figure.
fn write_spent(state: &mut StoreState, run: &str, spent: u64) -> Result<(), CallRecordError> {
let spend = RunSpend {
run_id: run.to_owned(),
spent,
};
/// Persists the ledger, then adopts it in memory. A write that fails
/// leaves the previous ledger.
fn write_ledger(state: &mut StoreState, ledger: RunLedger) -> Result<(), CallRecordError> {
let row = CultCacheEnvelope {
key: run.to_owned(),
r#type: RUN_SPEND_ROW.into(),
payload: rmp_serde::to_vec_named(&spend).map_err(|_| CallRecordError::Store)?,
key: ledger.run_id.clone(),
r#type: RUN_LEDGER_ROW.into(),
payload: rmp_serde::to_vec_named(&ledger).map_err(|_| CallRecordError::Store)?,
stored_at: Utc::now().to_rfc3339(),
schema_id: Some(RUN_SPEND_SCHEMA.into()),
schema_id: Some(RUN_LEDGER_SCHEMA.into()),
};
state.store.push(&row).map_err(|_| CallRecordError::Store)?;
state.runs.entry(run.to_owned()).or_default().spent = spent;
state.runs.entry(ledger.run_id.clone()).or_default().ledger = Some(ledger);
Ok(())
}
/// Persists `spent` as the run's spend; the run must have been claimed.
fn write_spent(state: &mut StoreState, run: &str, spent: u64) -> Result<(), CallRecordError> {
let mut ledger = state
.runs
.get(run)
.and_then(|tally| tally.ledger.clone())
.ok_or(CallRecordError::Store)?;
ledger.spent = spent;
Self::write_ledger(state, ledger)
}
fn held(state: &StoreState, run: &RunId) -> u64 {
state
.runs
.get(run.as_str())
.and_then(|tally| tally.ledger.as_ref())
.map_or(0, |ledger| ledger.spent)
}
/// Charges `bound` to the run's persisted ledger, or refuses with
/// `OverCap` when the ledger plus the bound would pass `cap`. The check and
/// the write hold one lock, so calls on the wire hold their bounds.
fn reserve(&self, run: &RunId, bound: u64, cap: u64) -> Result<(), CallRecordError> {
let mut state = self.lock()?;
let held = state.runs.get(run.as_str()).map_or(0, |tally| tally.spent);
let after = held.saturating_add(bound);
let after = Self::held(&state, run).saturating_add(bound);
if after > cap {
return Err(CallRecordError::OverCap);
}
@ -379,8 +495,9 @@ impl CallRecordStore {
if let Some(record) = record {
Self::push_record(&mut state, record)?;
}
let held = state.runs.get(run.as_str()).map_or(0, |tally| tally.spent);
let settled = held.saturating_sub(bound).saturating_add(charged);
let settled = Self::held(&state, run)
.saturating_sub(bound)
.saturating_add(charged);
Self::write_spent(&mut state, run.as_str(), settled)
}
@ -446,15 +563,18 @@ pub struct RecordingInferencePort {
}
impl RecordingInferencePort {
/// Takes ownership of the run's ledger, which resumes from the store. A
/// second port on a run whose first is alive is refused with `RunInUse`.
/// Takes ownership of the run's ledger, which resumes from the store. The
/// first open of a run records its cap and the world it starts from; a
/// reopen with another cap is refused with `RunMismatch(Cap)`. A second
/// port on a run whose first is alive is refused with `RunInUse`.
pub fn open(
inner: Arc<dyn InferencePort>,
store: Arc<CallRecordStore>,
run: RunId,
cap_tokens: u64,
start: &RunStart,
) -> Result<Self, CallRecordError> {
store.own(&run)?;
store.claim(&run, cap_tokens, start)?;
Ok(Self {
inner,
store,
@ -567,20 +687,26 @@ pub struct ReplayInferencePort {
}
impl ReplayInferencePort {
/// `records` are one run's, as `CallRecordStore::records` returns them.
/// Replays `run` from the store against `world`, a restored copy of the
/// world the run started from. Any other world is refused here, before a
/// call, with `RunMismatch` naming the field; a run the store never began
/// is `NotReplayable`.
pub fn new(
caller_runtime_id: impl Into<String>,
records: Vec<ProviderCallRecord>,
store: &CallRecordStore,
run: &RunId,
world: &RunStart,
) -> Result<Self, CallRecordError> {
let dense = records.iter().enumerate().all(|(index, record)| {
record.sequence == index as u64 + 1 && record.run_id == records[0].run_id
});
if !dense {
return Err(CallRecordError::NotReplayable);
let began = store.start_of(run)?.ok_or(CallRecordError::NotReplayable)?;
if began.world_id != world.world_id {
return Err(CallRecordError::RunMismatch(RunField::WorldId));
}
if began.state_digest != world.state_digest {
return Err(CallRecordError::RunMismatch(RunField::StateDigest));
}
Ok(Self {
caller_runtime_id: caller_runtime_id.into(),
pending: Mutex::new(records.into()),
pending: Mutex::new(store.records(run)?.into()),
})
}
}

330
crates/ghostlight/src/verse/calls/tests.rs
View file

@ -129,14 +129,28 @@ fn store_in(directory: &tempfile::TempDir) -> (Arc<CallRecordStore>, std::path::
)
}
/// A world identity and digest for tests that need a run to start somewhere.
fn start() -> RunStart {
RunStart {
world_id: "world-a".into(),
state_digest: "sha256:a".into(),
}
}
fn recording(
inner: &Arc<ScriptedPort>,
store: &Arc<CallRecordStore>,
run: &str,
cap: u64,
) -> RecordingInferencePort {
RecordingInferencePort::open(inner.clone(), store.clone(), RunId::new(run).unwrap(), cap)
.expect("the port opens")
RecordingInferencePort::open(
inner.clone(),
store.clone(),
RunId::new(run).unwrap(),
cap,
&start(),
)
.expect("the port opens")
}
fn json(output: &InferenceOutput) -> Vec<u8> {
@ -502,6 +516,7 @@ async fn one_run_has_one_ledger_and_one_owner() {
store.clone(),
RunId::new("run-own").unwrap(),
cap,
&start(),
);
assert_eq!(second.err(), Some(CallRecordError::RunInUse));
// Another run is another ledger.
@ -583,7 +598,7 @@ async fn a_reopened_run_resumes_its_spend() {
let store = Arc::new(CallRecordStore::open(&path).unwrap());
let inner = ScriptedPort::new(vec![Ok(reply("two", "r2"))]);
let resumed = recording(&inner, &store, "run-resume", bound * 2 - 1);
let resumed = recording(&inner, &store, "run-resume", bound * 2);
assert_eq!(resumed.spent(), bound);
let fault = resumed
.infer(probe)
@ -619,8 +634,7 @@ async fn replay_serves_stored_responses_and_refuses_strangers() {
}
let calls_made = inner.calls();
let stored = store.records(&run).unwrap();
let replay = ReplayInferencePort::new("verse-replay", stored).unwrap();
let replay = ReplayInferencePort::new("verse-replay", &store, &run, &start()).unwrap();
// A request the run never made is a fault and consumes nothing.
let stranger = replay
.prepare(request_for(CommandId::new(), "z", 1_200))
@ -665,20 +679,14 @@ fn sample_record(run: &str, sequence: u64) -> ProviderCallRecord {
}
#[test]
fn replay_needs_one_runs_dense_records() {
assert!(
ReplayInferencePort::new("x", vec![sample_record("a", 1), sample_record("a", 2)]).is_ok()
fn replay_needs_a_run_the_store_began() {
let directory = tempfile::tempdir().unwrap();
let (store, _) = store_in(&directory);
let run = RunId::new("never-begun").unwrap();
assert_eq!(
ReplayInferencePort::new("x", &store, &run, &start()).err(),
Some(CallRecordError::NotReplayable)
);
for records in [
vec![sample_record("a", 2)],
vec![sample_record("a", 1), sample_record("a", 3)],
vec![sample_record("a", 1), sample_record("b", 2)],
] {
assert_eq!(
ReplayInferencePort::new("x", records).err(),
Some(CallRecordError::NotReplayable)
);
}
}
#[tokio::test]
@ -710,11 +718,13 @@ async fn a_record_holds_no_request_text() {
fn spend_row(run: &str, spent: u64) -> CultCacheEnvelope {
envelope(
RUN_SPEND_ROW,
Some(RUN_SPEND_SCHEMA),
RUN_LEDGER_ROW,
Some(RUN_LEDGER_SCHEMA),
run,
rmp_serde::to_vec_named(&RunSpend {
rmp_serde::to_vec_named(&RunLedger {
run_id: run.into(),
cap: u64::MAX,
start: start(),
spent,
})
.unwrap(),
@ -800,11 +810,13 @@ fn the_store_refuses_any_row_that_is_not_a_canonical_v1_record() {
(
"a ledger at the wrong key",
envelope(
RUN_SPEND_ROW,
Some(RUN_SPEND_SCHEMA),
RUN_LEDGER_ROW,
Some(RUN_LEDGER_SCHEMA),
"other",
rmp_serde::to_vec_named(&RunSpend {
rmp_serde::to_vec_named(&RunLedger {
run_id: "run-x".into(),
cap: u64::MAX,
start: start(),
spent: 1,
})
.unwrap(),
@ -813,12 +825,14 @@ fn the_store_refuses_any_row_that_is_not_a_canonical_v1_record() {
(
"a ledger that is not canonical",
envelope(
RUN_SPEND_ROW,
Some(RUN_SPEND_SCHEMA),
RUN_LEDGER_ROW,
Some(RUN_LEDGER_SCHEMA),
"run-x",
[
rmp_serde::to_vec_named(&RunSpend {
rmp_serde::to_vec_named(&RunLedger {
run_id: "run-x".into(),
cap: u64::MAX,
start: start(),
spent: 1,
})
.unwrap(),
@ -909,7 +923,7 @@ async fn replay_keys_the_request_content_not_the_command_identity() {
.unwrap();
}
// A fresh genesis names every command anew; the content is what repeats.
let replay = ReplayInferencePort::new("verse-replay", store.records(&run).unwrap()).unwrap();
let replay = ReplayInferencePort::new("verse-replay", &store, &run, &start()).unwrap();
let under_new_names = |prompt: &str, max: u32| {
replay
.prepare(request_for(CommandId::new(), prompt, max))
@ -928,3 +942,263 @@ async fn replay_keys_the_request_content_not_the_command_identity() {
assert_eq!(json(&one), json(&reply("one", "r1")));
assert_eq!(json(&two), json(&reply("two", "r2")));
}
#[tokio::test]
async fn usage_one_token_over_the_bound_faults_and_stores_no_record() {
let directory = tempfile::tempdir().unwrap();
let (store, _) = store_in(&directory);
let prepared = prepared_for("one over", 1_200);
let bound = bound_of(&prepared);
let inner = ScriptedPort::new(vec![Ok(
reply("edge", "r1").with_usage(usage(bound, 1, None))
)]);
let port = recording(&inner, &store, "run-edge", bound + 200);
let fault = port.infer(prepared).await.expect_err("one over");
assert_eq!(fault.class(), InferenceFaultClass::BudgetExhausted);
assert_eq!(port.spent(), bound + 1, "charged in full");
assert!(
store
.records(&RunId::new("run-edge").unwrap())
.unwrap()
.is_empty()
);
}
/// Every field of the provider request is in the replay key except the two
/// ids that name the call.
#[test]
fn the_replay_key_covers_what_the_request_asks_and_not_what_names_it() {
let base = prepared_for("key", 1_200).invocation.request.clone();
let digest = |edit: &dyn Fn(&mut CodexProviderRequest)| {
let mut request = base.clone();
edit(&mut request);
content_digest(&request).unwrap()
};
let same = digest(&|_| {});
assert_eq!(same, content_digest(&base).unwrap());
let changes: [(&str, Box<dyn Fn(&mut CodexProviderRequest)>); 10] = [
("model", Box::new(|r| r.model.push('x'))),
("instructions", Box::new(|r| r.instructions.push('x'))),
(
"tools",
Box::new(|r| {
r.tools.push(CodexToolDefinition {
name: "t".into(),
description: "d".into(),
parameters_json: "{}".into(),
})
}),
),
(
"input",
Box::new(|r| r.input.push(CodexInputItem::UserText { text: "x".into() })),
),
(
"max_output_tokens",
Box::new(|r| r.max_output_tokens = Some(1)),
),
(
"reasoning_effort",
Box::new(|r| r.reasoning_effort = Some("low".into())),
),
(
"service_tier",
Box::new(|r| r.service_tier = Some("flex".into())),
),
(
"output_schema_json",
Box::new(|r| r.output_schema_json = Some("{}".into())),
),
(
"previous_response_id",
Box::new(|r| r.previous_response_id = Some("p".into())),
),
(
"parallel_tool_calls",
Box::new(|r| r.parallel_tool_calls = !r.parallel_tool_calls),
),
];
for (field, change) in changes {
assert_ne!(digest(&*change), same, "{field} is not in the replay key");
}
assert_eq!(digest(&|r| r.request_id.push('x')), same);
assert_eq!(digest(&|r| r.conversation_id.push('x')), same);
}
fn open_with(
rows: Vec<CultCacheEnvelope>,
) -> (tempfile::TempDir, Result<CallRecordStore, CallRecordError>) {
let directory = tempfile::tempdir().unwrap();
let path = directory.path().join("calls.redb");
let mut backing = OwnedRedbMessagePackBackingStore::new(&path).unwrap();
for row in &rows {
backing.push(row).unwrap();
}
drop(backing);
let opened = CallRecordStore::open(&path);
(directory, opened)
}
fn record_row(record: &ProviderCallRecord) -> CultCacheEnvelope {
envelope(
CALL_RECORD_ROW,
Some(CALL_RECORD_SCHEMA),
&row_key(&record.run_id, record.sequence),
rmp_serde::to_vec_named(record).unwrap(),
)
}
#[test]
fn a_run_with_sequence_zero_beside_a_hole_is_refused() {
let (_directory, opened) = open_with(vec![
record_row(&sample_record("run-x", 0)),
record_row(&sample_record("run-x", 2)),
spend_row("run-x", 6),
]);
assert_eq!(opened.err(), Some(CallRecordError::Corrupt));
}
#[test]
fn a_record_charged_nothing_still_needs_its_ledger() {
let mut record = sample_record("run-y", 1);
record.charged = 0;
let (_directory, opened) = open_with(vec![record_row(&record)]);
assert_eq!(opened.err(), Some(CallRecordError::Corrupt));
}
#[test]
fn a_record_whose_run_id_is_no_whole_key_segment_is_refused() {
let mut stray = sample_record("run-z/b", 1);
stray.charged = 0;
let (_directory, opened) = open_with(vec![
record_row(&stray),
record_row(&sample_record("run-z", 1)),
spend_row("run-z", 3),
]);
assert_eq!(opened.err(), Some(CallRecordError::Corrupt));
// A neighbouring run id that merely shares the prefix keeps its own records.
let (_directory, opened) = open_with(vec![
record_row(&sample_record("run-z", 1)),
spend_row("run-z", 3),
spend_row("run-z-b", 0),
]);
let store = opened.unwrap();
assert_eq!(
store.records(&RunId::new("run-z").unwrap()).unwrap().len(),
1
);
}
#[tokio::test]
async fn a_run_reopens_only_under_the_cap_it_began_with() {
let directory = tempfile::tempdir().unwrap();
let (store, path) = store_in(&directory);
let inner = ScriptedPort::new(Vec::new());
drop(recording(&inner, &store, "run-cap", 5_000));
let run = RunId::new("run-cap").unwrap();
let open = |store: &Arc<CallRecordStore>, cap: u64| {
RecordingInferencePort::open(inner.clone(), store.clone(), run.clone(), cap, &start())
};
for other in [4_999, 5_001, u64::MAX] {
assert_eq!(
open(&store, other).err(),
Some(CallRecordError::RunMismatch(RunField::Cap))
);
}
// A refused reopen does not hold the run, and the cap outlives the process.
drop(open(&store, 5_000).expect("the same cap reopens"));
drop(store);
let store = Arc::new(CallRecordStore::open(&path).unwrap());
assert_eq!(
open(&store, 5_001).err(),
Some(CallRecordError::RunMismatch(RunField::Cap))
);
open(&store, 5_000).expect("the same cap reopens after a restart");
let message = format!("{}", CallRecordError::RunMismatch(RunField::Cap));
assert!(message.contains("token cap") && !message.contains("5000"));
}
/// A world made by the production path: one genesis command in a fresh world
/// file, its snapshot, and the file's path.
async fn genesis(
directory: &std::path::Path,
command: CommandId,
) -> (crate::WorldSnapshot, std::path::PathBuf) {
let path = directory.join("world.cc");
let (mailbox, task) = crate::WorldMailbox::open(&path).expect("an empty world");
mailbox
.create_fixture(
crate::tests::creation(command, "Replayed"),
&crate::tests::auth_principal(crate::tests::owner()),
)
.await
.expect("a created world");
let snapshot = mailbox.snapshot().await.unwrap();
drop(mailbox);
task.await.unwrap();
(snapshot, path)
}
async fn reopened(path: &std::path::Path) -> crate::WorldSnapshot {
let (mailbox, task) = crate::WorldMailbox::open(path).expect("a restored world");
let snapshot = mailbox.snapshot().await.unwrap();
drop(mailbox);
task.await.unwrap();
snapshot
}
#[tokio::test]
async fn replay_runs_against_a_restored_copy_and_refuses_any_other_world_up_front() {
let command = CommandId::new();
let first = tempfile::tempdir().unwrap();
let (world, world_file) = genesis(first.path(), command).await;
let began = RunStart::of(&world);
let directory = tempfile::tempdir().unwrap();
let (store, _) = store_in(&directory);
let run = RunId::new("run-world").unwrap();
let inner = ScriptedPort::new(ok_all(&[reply("one", "r1")]));
let live =
RecordingInferencePort::open(inner.clone(), store.clone(), run.clone(), u64::MAX, &began)
.unwrap();
live.infer(live.prepare(request_for(command, "go", 1_200)).unwrap())
.await
.unwrap();
drop(live);
// A byte copy of the starting world, restored, replays the run.
let restored_dir = tempfile::tempdir().unwrap();
let restored_file = restored_dir.path().join("world.cc");
std::fs::copy(&world_file, &restored_file).unwrap();
let restored = RunStart::of(&reopened(&restored_file).await);
assert_eq!(restored, began);
let replay = ReplayInferencePort::new("verse-replay", &store, &run, &restored)
.expect("a restored copy replays");
let served = replay
.infer(replay.prepare(request_for(command, "go", 1_200)).unwrap())
.await
.unwrap();
assert_eq!(json(&served), json(&reply("one", "r1")));
// The same genesis created again is another world and is refused before
// any call, naming the field and nothing it held.
let again = tempfile::tempdir().unwrap();
let (recreated, _) = genesis(again.path(), command).await;
let refusal = ReplayInferencePort::new("verse-replay", &store, &run, &RunStart::of(&recreated))
.err()
.expect("a re-created world");
assert_eq!(refusal, CallRecordError::RunMismatch(RunField::WorldId));
let text = format!("{refusal} {refusal:?}");
assert!(!text.contains(&began.world_id) && !text.contains(&began.state_digest));
// The same id with another digest is refused too.
let drifted = RunStart {
world_id: began.world_id.clone(),
state_digest: format!("{}x", began.state_digest),
};
assert_eq!(
ReplayInferencePort::new("verse-replay", &store, &run, &drifted).err(),
Some(CallRecordError::RunMismatch(RunField::StateDigest))
);
assert_eq!(inner.calls(), 1, "replay never reached the provider");
}

2
crates/ghostlight/src/verse/mod.rs
View file

@ -6,5 +6,5 @@ mod calls;
pub use calls::{
CALL_RECORD_SCHEMA, CallRecordError, CallRecordStore, ProviderCallRecord,
RecordingInferencePort, ReplayInferencePort, RunId,
RecordingInferencePort, ReplayInferencePort, RunField, RunId, RunStart,
};