agent_block_core/bridge/knl.rs
1//! `knl.*` — Lua surface of the kernel syscall layer.
2//!
3//! This module is an adapter, nothing more. The domain rules live in
4//! [`crate::knl`] (pure Rust, unit-tested without a VM) and are stated in
5//! that module's doc; here we only:
6//!
7//! 1. define the `Session` userdata and bind its methods,
8//! 2. convert Lua tables ⇄ `serde_json::Value`,
9//! 3. attribute failures as `knl: <method>: <kind>: <reason>`.
10//!
11//! Keeping the conversion in one place is what makes the re-entrancy
12//! discipline checkable: walking a Lua table can call back into Lua, so
13//! every conversion happens *outside* an active borrow of the session,
14//! and the kernel core never sees a Lua value at all.
15//!
16//! # The Lua surface
17//!
18//! Five module functions and one userdata.
19//!
20//! - `knl.open(opts?) -> session` — `{ owner?, budget? = { amount, tag?,
21//! desc? }, store?, parent? }`. State only: the policy half a beat runs
22//! against is built by the Lua kernel's own constructor, never here.
23//! `parent` opens the session *from* another one, on that one's database
24//! and out of its balance — `budget = { from_parent = n, tag? }` — which is
25//! one write: the child's opening and grant, and the parent's reservation.
26//! A balance that will not cover it records a refusal on the parent and
27//! raises `refused`, and no session is returned.
28//! - `knl.resume(opts) -> session` — `{ session, store?, budget? }`: reopen a
29//! stream and re-fold it. An absent `store` means what it means on open,
30//! the host's database — so `knl.resume{ session = id }` reopens what
31//! `knl.open{}` wrote.
32//! - `knl.new_beat_id() -> string` — a time-ordered, session-free id for the
33//! caller to stamp on the events of one beat.
34//! - `knl.error(e) -> { kind, method, retryable, message }` — read a raised
35//! failure back as data (see *Failures carry their class*).
36//! - `knl.api() -> { session, module, errors, schema, types }` — the declared
37//! surface, the failure vocabulary, the columns a query may name, and the
38//! generated shapes of every argument and return (see *One declaration*).
39//!
40//! The session userdata answers `id`, `scope_id`, `owner`, `append`,
41//! `events`, `len`, `view`, `query`, `reserve`, `spend`, `remaining`,
42//! `exhausted` and `close`, plus the `__close` metamethod. [`SESSION_API`]
43//! and [`MODULE_API`] hold each one's contract in a line — including the
44//! classes it can raise — and are what `knl.api()` hands back.
45//!
46//! # Everything that reaches the store yields
47//!
48//! `knl.open` / `knl.resume` and every session method that touches the log are
49//! bound as **async** functions, so calling one suspends the coroutine rather
50//! than stopping the VM. That is not an optimization: the Lua VM's thread is
51//! the only worker of its own runtime, so a syscall that waited there would
52//! also stop every other coroutine on that VM, its timers and its
53//! cancellation ([`crate::knl`] § Async).
54//!
55//! **Nothing about the Lua changes.** A yield inside `pcall` / `xpcall`, a
56//! `<close>` scope ending (cleanly or by error) and a `for` loop over beats
57//! are all yieldable in Lua 5.4, so `s:append(...)` reads and behaves exactly
58//! as it did. The one requirement is the one the shell already meets: these
59//! methods are reachable from inside a coroutine, which the main chunk and
60//! every bus handler are.
61//!
62//! Three methods stay synchronous — `id`, `scope_id`, `owner` — because they
63//! answer out of the value and wait for nothing, and so do `knl.new_beat_id`,
64//! `knl.error` and `knl.api`.
65//!
66//! # The invariants the surface enforces
67//!
68//! The shell reaches kernel state only through the methods of the
69//! userdata returned by `knl.open(opts?)`, so the invariants are enforced
70//! by the shape of the API rather than by convention:
71//!
72//! - **An event is envelope + meta + data.** What you append is
73//! `{ kind = …, beat? = …, meta? = { … }, data? = { … } }` and nothing
74//! else: a top-level key outside that set is refused, because a kind's own
75//! fields belong under `data`. `meta` is shallow — string, number or
76//! boolean values — so a reader can group or filter on it without knowing
77//! the kind; `data` is yours, at any depth, and the shape under it is
78//! declared where the kind is written (`knl.shapes`). The kernel checks
79//! the `data` of its own six kinds (`session_*` / `budget_*`) and of no
80//! others. Both come back the way they went in, `data` defaulting to an
81//! empty table when you write none.
82//! - **I1 append-only.** There is deliberately no `update`, `delete` or
83//! `replace`. `events()` / `view()` hand back freshly built tables, so
84//! a caller that mutates a returned value cannot reach recorded state.
85//! `seq` and `epoch_ms` are assigned by the kernel and overwrite any
86//! caller-supplied field of the same name. Nothing else is added: a
87//! `beat` the caller declared is stored exactly as given.
88//! - **A session has a scope.** The two are different things sharing one
89//! lifetime: the session is the stream (`s:id()`), the scope is the
90//! authority it is written under — a kernel-issued scope id (`s:scope_id()`)
91//! and the principal it belongs to (`s:owner()`, a real id or the reserved
92//! "anon" / "system"). Neither id is a caller's to choose, and they are
93//! not each other: `s:id()` names the stream a `knl.resume` reopens. The
94//! scope id is recorded on `session_opened` and on every `budget_*` event,
95//! so the boundary is readable — and unforgeable — from the log alone.
96//! There is no per-event author: a session holds only its own events, so
97//! an accounting of what a run consumed keys on the `kind` alone — every
98//! `llm_response` in the log is a call this run made.
99//! - **Beats are yours.** `knl.new_beat_id()` mints a time-ordered id; you
100//! stamp it on the `llm_response`, `tool_call` and `tool_result` events
101//! that belong to one beat. The kernel does not number beats, does not
102//! require the field, and asks only that a `beat` you do write is a
103//! string.
104//! - **I3 budget monotonicity.** The budget is a quota an owner grants
105//! the session, not a ledger of what it used. Two deductions take from it
106//! and neither holds anything for the other: `reserve(n)` is the deduction
107//! that *asks* — it takes `n` off and returns `true`, or refuses with
108//! `false, tag` and leaves the balance exactly as it was — and `spend(n)`
109//! is the deduction that does not ask, returning nothing: the write landing
110//! *is* the answer, and `remaining()` is the separate question of what is
111//! left. There is no hold and nothing to release, so **a beat that calls
112//! both deducts twice**; which of the two a beat uses is yours to pick.
113//! Both take non-negative whole amounts and the balance can only decrease;
114//! there is no API to raise or reset it, and no `append` moves it. What a
115//! session actually consumed is read off the responses it recorded
116//! (`knl.views.usage`, a query view in Lua), independent of the balance.
117//! - **I6 session lifecycle.** All state lives inside the userdata — no
118//! module-level statics, no Lua globals — so two sessions are fully
119//! independent. There is no "run" inside a session: `knl.open()` records
120//! `session_opened` and the grant, so the log says what was allowed, and
121//! `close` records `session_closed`. Both events are the kernel's alone —
122//! appending either by hand is an error — so the lifecycle in the log is
123//! the lifecycle that happened. Three paths reach the closing boundary
124//! and the log never loses it: `close(reason?)` said
125//! explicitly; a `local s <close> = knl.open{...}` scope ending, which
126//! records `scope_exit` — or `error` with the message in `detail` when the
127//! block raised; and the drop backstop, `dropped`, for a handle nobody
128//! closed. Whichever runs first wins and the rest are no-ops, because
129//! `close` is idempotent — so an explicit reason is never overwritten by
130//! the scope or the collector that follows it.
131//! - **Closed is the handle's, not the stream's.** `closed` is a flag on
132//! *this* userdata: after it, this handle's `append` / `reserve` / `spend`
133//! are errors while its reads keep working. The log turns nothing away.
134//! A write from another handle that never saw the ending is recorded after
135//! the `session_closed`, because that is what happened and it is exactly
136//! what an audit is reading for; two handles that both close leave two
137//! endings, not one. One reader consults `session_closed` at all —
138//! `knl.resume`, which refuses a stream whose ending is already in the log,
139//! because a session is disposable.
140//! - **A child is paid for, and then it is on its own.** `knl.open{ parent =
141//! s, budget = { from_parent = n } }` moves `n` out of `s`'s balance and
142//! opens a session on `s`'s database with `n` of its own, in one write:
143//! `s`'s ledger gains a `budget_reserved` naming the child, and the child's
144//! log opens with `parent` recorded on it. Nothing comes back when the
145//! child closes — an allocation is a spend — and `s` holds no handle on it:
146//! the structure is in the log, and `knl.views.tree` reads it. Closing a
147//! parent whose children are still open is not refused; the boundary
148//! records them (`session_closed.data.open_children`).
149//! - **K2 model call.** There is no composite call and the session keeps
150//! no backend of its own. The driver takes the beat's cost off the quota
151//! itself — `reserve` before the call to be stopped when it will not fit,
152//! or `spend` after it to meter what happened — calls the backend, and
153//! appends the `llm_response`. Doing both charges the beat twice.
154//!
155//! # Driving a beat from Lua
156//!
157//! ```lua
158//! local s = knl.open({
159//! owner = "user-42", -- default: the reserved "anon"
160//! budget = { amount = 10000, tag = "tokens" },
161//! })
162//! s:append({ kind = "msg_user", data = { content = "hi" } })
163//! -- Drive the beat yourself: name it, take the cost off the quota, call
164//! -- the backend, append the response. This beat asks first, so it is
165//! -- stopped before it spends; a beat that only meters what happened calls
166//! -- `s:spend(actual)` after the response instead. Not both: the two
167//! -- deductions are independent, so a beat that uses both is charged twice.
168//! local beat = knl.new_beat_id()
169//! local ok, tag = s:reserve(est)
170//! if not ok then return { budget_stopped = true, tag = tag } end
171//! s:append({ kind = "llm_response", beat = beat,
172//! data = { content = blocks, usage = u } })
173//! local events = s:events(from) -- the record, from `from` on
174//! local tail = s:view("tail", { n = 5 }) -- the last events, verbatim
175//! s:close("done")
176//! ```
177//!
178//! # Reading the log
179//!
180//! Two named read faces and no more: the events, and the last of the record.
181//! Turning events into a request for a provider — which role a kind takes,
182//! whether a system message goes in front, where to cut the history off — is
183//! the shell's policy, so it is written in Lua over `events(from)` rather
184//! than named as a view here. The token account went the same way: it reads
185//! the `usage` the adapter recorded on each `llm_response`, which is the
186//! shell's vocabulary, so it is a query view in Lua (`knl.views.usage`) and
187//! not a name the kernel answers to.
188//!
189//! The third face is not a name but a language: `s:query(sql, params?,
190//! opts?)` reads the log with one `SELECT` / `WITH` over the table the events
191//! live in, whose columns `knl.api().schema` publishes. That is what keeps
192//! the list of names short — a fold the kernel has no opinion about is a
193//! query, not a name it had to be taught. `$stream` binds to this session
194//! and `$sessions` to the set in `opts.sessions`, so reading across a tree of
195//! sessions is one statement. Values are bound, never pasted; the connection
196//! it runs on cannot write; and it returns `rows, truncated`, so a page can
197//! be told from a complete answer.
198//!
199//! # Storage backend
200//!
201//! **A real session is a file.** `knl.open` takes an optional `store`, and
202//! leaving it out is the host's database — one file per project,
203//! `{base_dir}/projects/<slug>/knl.sqlite` unless `AGENT_BLOCK_KNL_PATH` says
204//! otherwise ([`crate::bridge::config::knl_path`]) — so every session a script
205//! opens is a stream in one database, and a tree opened from a default parent
206//! is a tree in one file.
207//!
208//! `"mem"` is the other choice and has to be asked for by name: an in-memory
209//! SQLite database that lives as long as the session does, **for tests and
210//! mocks — one session, one process, nothing shared**. It is not a lighter
211//! version of the default. A shared-cache URI is how a second connection
212//! reaches one, and shared cache locks per *table*, so a second writer meets
213//! `SQLITE_LOCKED` at once and no busy timeout waits it out; opening a child
214//! on a parent that is on one is therefore a `validation` refusal rather than
215//! a wait. `{ sqlite = "<path>" }` is a file the caller picked.
216//!
217//! One backend, two kinds of database — the log is a table either way, which
218//! is what `s:query` reads. `knl.resume({ session = "<id>", store?,
219//! budget? })` reopens a stream and re-folds it, so a resumed session's
220//! accounting continues from the recorded state — it behaves exactly like a
221//! fresh session, only pre-loaded. A file survives the process; an in-memory
222//! database does not, so resuming one is only possible while another handle
223//! still holds it open.
224//!
225//! # Failures carry their class
226//!
227//! A failure is raised as a message, because mlua raises every error a Rust
228//! callback returns as its own userdata and offers no way to make a Lua table
229//! *be* the raised value. So the message is given a shape instead:
230//!
231//! ```text
232//! knl: <method>: <kind>: <reason>
233//! │ │ │ └── prose, and the only part that is
234//! │ │ └── one of knl.api().errors (knl::KnlError::KINDS)
235//! │ └── the method that raised
236//! └── the fixed prefix
237//! ```
238//!
239//! The first three fields are a closed vocabulary ([`knl::KnlError::KINDS`])
240//! and only the fourth is prose. `knl.error(e)` reads it back as
241//! `{ kind, method, retryable, message }` — with `retryable` the kernel's own
242//! judgement, true for `busy` alone — and an unattributed raise comes back
243//! whole, `kind` absent and the entire text as `message`, so the reader never
244//! fails on input it does not recognise. `knl.api().errors` publishes the
245//! class list, so the shell's own declaration of it is checked rather than
246//! trusted. A caller that only wants to print keeps working: the table
247//! renders as the message it came from, and the message still contains what
248//! it always did.
249//!
250//! # The declared surface, and the check that holds it
251//!
252//! What Lua can reach is [`SESSION_API`] plus [`MODULE_API`] and nothing
253//! else, and that is a checked claim rather than a documented intention: a
254//! test reflects over a live userdata and over the `knl` table and fails on a
255//! method bound without an entry, and on an entry with no method. A second
256//! test holds `knl.api().schema` against the columns the store actually has.
257//! The Lua kernel runs the mirror image of both, so a syscall added on one
258//! side and not the other goes red rather than drifting.
259//!
260//! # One declaration: the Rust types
261//!
262//! The two tables above say what the methods are *called*; [`types`] says what
263//! each one takes and answers, and it is the only place either is written
264//! down. Every argument is deserialized into one of those types
265//! ([`from_lua`]) and every table a syscall answers with is built from one
266//! ([`as_table`]), so the check runs on every call in both modes — including a
267//! direct `s:append(...)`, which never passes through the Lua kernel's own dev
268//! gate.
269//!
270//! The same types are rendered as lshape ([`lshape_module_source`], via
271//! `schema-bridge`) and embedded as the Lua module `knl_types` at host start.
272//! `knl.shapes.session` / `knl.shapes.module` point at that module rather than
273//! restating it, and `knl.api().types` hands the source text back for tooling.
274//! What this replaced was two declarations of one interface — these
275//! signatures, and a hand-written lshape table beside them — held together by
276//! a test that compared *names*, which is a test a renamed field walks
277//! straight past. The generated module is built at start rather than checked
278//! in for the same reason: a generated file in the tree is one somebody can
279//! edit, and an edited one is the second declaration all over again.
280
281use mlua::prelude::*;
282use serde_json::{Map, Value};
283use tokio::sync::Mutex;
284
285use super::{json_to_lua, lua_to_json};
286use crate::knl;
287
288/// Every method the session userdata answers to, with the contract it holds
289/// in one line.
290///
291/// The declared surface: what Lua can reach on a session is this table and
292/// nothing else, and a test below reflects over a live userdata to hold that
293/// (a method registered without an entry here fails it). `knl.api()` hands
294/// the same list to Lua, so a caller can ask what a session offers instead of
295/// guessing.
296/// Each doc names the classes that method can raise (`knl.error(e).kind`,
297/// one of `knl.api().errors`), so a caller reads what it has to handle from
298/// the same table it reads the signature from. `busy` / `storage` /
299/// `corruption` are the durable backend's and never occur on the in-memory
300/// one; a method that only reads its own value raises nothing at all.
301pub const SESSION_API: &[(&str, &str)] = &[
302 ("id", "id() -> string — the stream this session writes"),
303 (
304 "scope_id",
305 "scope_id() -> string — the authority the stream is written under",
306 ),
307 (
308 "owner",
309 "owner() -> string — the principal the scope belongs to (or \"anon\" / \"system\")",
310 ),
311 (
312 "append",
313 "append(event) -> seq — record a fact: { kind, beat?, meta? (shallow), data? }; a key \
314 outside that envelope, a kernel-only kind and a nested meta are refused, and the budget \
315 does not move [raises: validation, closed, busy, storage]",
316 ),
317 (
318 "events",
319 "events(from?) -> rows, truncated — the record from `from` on, as fresh tables in the \
320 shape it was written in (kind / beat / meta / data, plus the kernel's stamps), capped at \
321 the kernel's row limit; `truncated` says the cap cut the read short, and the rest is read \
322 by paging on `from` [raises: busy, storage, corruption]",
323 ),
324 (
325 "len",
326 "len() -> integer — how many events are recorded [raises: busy, storage]",
327 ),
328 (
329 "view",
330 "view(name, opts?) -> table — the one named fold: \"tail\" { n }; anything else is a \
331 query [raises: validation, busy, storage, corruption]",
332 ),
333 (
334 "query",
335 "query(sql, params?, opts?) -> rows, truncated — read the log with one SELECT / WITH; \
336 $stream is this session, $sessions is opts.sessions (default { this session }) \
337 [raises: validation, busy, storage, corruption, timeout]",
338 ),
339 (
340 "reserve",
341 "reserve(n) -> true | false, tag — the deduction that asks: refuse if remaining < n, \
342 atomic, both answers recorded [raises: validation, closed, busy, storage, corruption]",
343 ),
344 (
345 "spend",
346 "spend(n) -> nil — the deduction that does not ask; independent of reserve, so calling \
347 both for one beat deducts twice; the write is the answer, read the balance \
348 with remaining() [raises: validation, closed, busy, storage]",
349 ),
350 (
351 "remaining",
352 "remaining() -> integer | nil — the balance, nil without a budget; a store that cannot be \
353 read raises rather than reporting a stale one [raises: busy, storage, corruption]",
354 ),
355 (
356 "exhausted",
357 "exhausted() -> boolean — whether the budget is used up (false without one) \
358 [raises: busy, storage, corruption]",
359 ),
360 (
361 "close",
362 "close(reason?, detail?) -> nil — record session_closed and end the session; idempotent \
363 [raises: validation, busy, storage]",
364 ),
365 (
366 "__close",
367 "__close(err) — the <close> scope boundary: scope_exit, or error with the message as detail \
368 [raises: busy, storage — only on a clean exit; an unwinding one is logged]",
369 ),
370];
371
372/// Every function the `knl` global carries, with its one-line contract.
373///
374/// The module half of the declared surface, held by the same test, and
375/// annotated with the error classes each can raise like [`SESSION_API`].
376pub const MODULE_API: &[(&str, &str)] = &[
377 (
378 "open",
379 "open(opts?) -> session — owner? / budget? / store? (absent is the host's database, one \
380 file per project; \"mem\" is an in-memory database for tests and mocks — one session, \
381 one process, nothing shared; or { sqlite = path }); parent? opens a child on the \
382 parent's database with budget = { from_parent = n, tag? }, moving n out of the parent's \
383 balance in one write, and a parent on \"mem\" is refused because a tree needs a file \
384 store [raises: validation, refused, closed, busy, storage]",
385 ),
386 (
387 "resume",
388 "resume(opts) -> session — reopen a stream and re-fold it; an absent store means the \
389 host's database, as on open; a closed session is not resumable \
390 [raises: validation, closed, busy, storage, corruption]",
391 ),
392 (
393 "new_beat_id",
394 "new_beat_id() -> string — mint a time-ordered beat id for the caller to stamp",
395 ),
396 (
397 "error",
398 "error(err) -> { kind, method, retryable, message } — read a raised failure as a table; \
399 an unrecognised one comes back with kind = nil and the whole text as message",
400 ),
401 (
402 "api",
403 "api() -> { session = …, module = …, errors = { kind }, schema = { table, columns }, \
404 fields = { amount, tag, … } } — the declared surface, the columns a query may name, and \
405 the `data` paths a view reaches into",
406 ),
407];
408
409/// The declared surface as Rust types — the single source both sides read.
410///
411/// # Why the types are the declaration
412///
413/// [`SESSION_API`] and [`MODULE_API`] say what the methods are *called* and
414/// what each one is for; this module says what each one *takes and answers*.
415/// Every entry above has its argument and return types here, and they are the
416/// only place either is written down: the Lua kernel's own registry
417/// (`knl.shapes.session` / `knl.shapes.module`) is built by pointing at
418/// [`lshape_module_source`], which is generated from exactly these types at
419/// host start. Before this there were two declarations of one interface —
420/// these signatures, and a hand-written lshape table beside them — and the
421/// test that held them together compared *names*. A misspelt field on either
422/// side went unnoticed until a caller hit it.
423///
424/// # What that buys, beyond one declaration
425///
426/// The types are also the parser. [`from_lua`] deserializes a caller's table
427/// into them, so the check runs on every call in both modes rather than only
428/// under the Lua dev gate, and a direct `s:append(...)` — which never passes
429/// through that gate — is checked exactly like a call the shell made.
430///
431/// # The three shapes serde cannot state, and why
432///
433/// - [`Json`] / [`Meta`] — `data`, and a query's parameters, are opaque to the
434/// kernel: their shape belongs to whoever writes the kind. They map to
435/// lshape's `any` and to a map of labels.
436/// - [`StoreSpec`] / [`BudgetOpt`] — a string *or* a table, and two forms that
437/// exclude each other. `#[derive(SchemaBridge)]` renders an enum as its
438/// variant names, which is right for [`ViewName`] and wrong for a union, so
439/// these carry the schema by hand and the derive stays where it is honest.
440/// - [`ViewName`] — the vocabulary is [`knl::VIEW_TAIL`] and stays the
441/// kernel's: the schema is built *from* that constant rather than repeating
442/// it, and an unknown name is still refused by the kernel, in its own words.
443pub mod types {
444 use schema_bridge::{Field, Schema, SchemaBridge};
445 use serde::{Deserialize, Serialize};
446 use serde_json::{Map, Value};
447 use std::collections::BTreeMap;
448
449 /// Opaque JSON: what a kind's `data` is about, and what a query binds.
450 ///
451 /// `Any` rather than a shape, and deliberately — the kernel records
452 /// `data` as written and judges only its own six kinds, so a schema here
453 /// would be this layer inventing a contract it does not hold anyone to.
454 #[derive(Debug, Clone, Default, PartialEq, Serialize, Deserialize)]
455 #[serde(transparent)]
456 pub struct Json(pub Value);
457
458 impl SchemaBridge for Json {
459 fn to_ts() -> String {
460 "unknown".to_string()
461 }
462 fn to_schema() -> Schema {
463 Schema::Any
464 }
465 }
466
467 /// One label in an event's `meta`: a string, a number or a flag.
468 ///
469 /// The whole of the vocabulary. `meta` is the half of the envelope a view
470 /// can read without ever being broken by a change to a kind, and a nested
471 /// value would make it a second `data` with none of that promise.
472 #[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
473 #[serde(
474 untagged,
475 expecting = "a label: meta is shallow (a string, a number or a boolean)"
476 )]
477 pub enum MetaValue {
478 /// A word.
479 Text(String),
480 /// A count or a measurement.
481 Number(f64),
482 /// A flag.
483 Flag(bool),
484 }
485
486 impl SchemaBridge for MetaValue {
487 fn to_ts() -> String {
488 "string | number | boolean".to_string()
489 }
490 fn to_schema() -> Schema {
491 Schema::Union(vec![Schema::String, Schema::Number, Schema::Boolean])
492 }
493 }
494
495 /// An event's `meta`: labels, and only labels.
496 pub type Meta = BTreeMap<String, MetaValue>;
497
498 // -- scalars ------------------------------------------------------------
499 //
500 // A newtype rather than a bare `String` / `u64`, because the registry
501 // entry that names one is what makes the surface readable: `id()` answers
502 // a `SessionId`, not "a string".
503
504 /// The stream a session writes — what `knl.resume` reopens.
505 #[derive(Debug, Clone, PartialEq, Serialize, Deserialize, SchemaBridge)]
506 #[serde(transparent)]
507 pub struct SessionId(pub String);
508
509 /// The kernel-issued authority a stream is written under.
510 #[derive(Debug, Clone, PartialEq, Serialize, Deserialize, SchemaBridge)]
511 #[serde(transparent)]
512 pub struct ScopeId(pub String);
513
514 /// The principal a scope belongs to (a real id, or `anon` / `system`).
515 #[derive(Debug, Clone, PartialEq, Serialize, Deserialize, SchemaBridge)]
516 #[serde(transparent)]
517 pub struct Owner(pub String);
518
519 /// A beat id: time-ordered, session-free, and opaque to the kernel.
520 #[derive(Debug, Clone, PartialEq, Serialize, Deserialize, SchemaBridge)]
521 #[serde(transparent)]
522 pub struct BeatId(pub String);
523
524 /// An event's position in its stream — assigned by the kernel, and the
525 /// `from` a read starts at.
526 #[derive(Debug, Clone, PartialEq, Serialize, Deserialize, SchemaBridge)]
527 #[serde(transparent)]
528 pub struct Seq(pub u64);
529
530 /// How many events are recorded.
531 #[derive(Debug, Clone, PartialEq, Serialize, Deserialize, SchemaBridge)]
532 #[serde(transparent)]
533 pub struct Count(pub u64);
534
535 /// What `reserve` / `spend` move: a whole number of budget units.
536 ///
537 /// Signed, so a caller's negative lands here as a value the kernel refuses
538 /// rather than as a deserializer's type error about `u64` — the amount is
539 /// a number the kernel has a rule about, and the rule is the kernel's.
540 #[derive(Debug, Clone, Copy, PartialEq, Serialize, Deserialize, SchemaBridge)]
541 #[serde(transparent)]
542 pub struct Amount(pub i64);
543
544 /// The balance, or nothing at all when the session was granted no budget.
545 #[derive(Debug, Clone, PartialEq, Serialize, Deserialize, SchemaBridge)]
546 #[serde(transparent)]
547 pub struct Remaining(pub Option<i64>);
548
549 /// Whether the budget is used up (`false` without one).
550 #[derive(Debug, Clone, Copy, PartialEq, Serialize, Deserialize, SchemaBridge)]
551 #[serde(transparent)]
552 pub struct Exhausted(pub bool);
553
554 /// The statement a read is written as: one `SELECT` or `WITH`.
555 #[derive(Debug, Clone, PartialEq, Serialize, Deserialize, SchemaBridge)]
556 #[serde(transparent)]
557 pub struct Sql(pub String);
558
559 /// Which kind of ending a close was — a short word a reader can fold on.
560 #[derive(Debug, Clone, PartialEq, Serialize, Deserialize, SchemaBridge)]
561 #[serde(transparent)]
562 pub struct CloseReason(pub String);
563
564 /// The sentence only this close can tell: the message of the error a
565 /// caller's own bracket caught, say. Truncated before it is recorded.
566 #[derive(Debug, Clone, PartialEq, Serialize, Deserialize, SchemaBridge)]
567 #[serde(transparent)]
568 pub struct CloseDetail(pub String);
569
570 /// Whatever a raise handed over — the argument of `knl.error`, and what
571 /// `__close` is given when its block is unwinding.
572 ///
573 /// `any`, because it is: the bridge's own attributed message, a Lua-side
574 /// `error("...")`, or a value from somewhere else entirely.
575 #[derive(Debug, Clone, Copy, PartialEq)]
576 pub struct Raised;
577
578 impl SchemaBridge for Raised {
579 fn to_ts() -> String {
580 "unknown".to_string()
581 }
582 fn to_schema() -> Schema {
583 Schema::Any
584 }
585 }
586
587 /// The one named fold: `tail`.
588 ///
589 /// The vocabulary is [`crate::knl::projection::VIEW_TAIL`] and the schema is built
590 /// from it, so there is no second list to keep in step. A name that is
591 /// not in it reaches the kernel and is refused there, which is where the
592 /// vocabulary lives.
593 #[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
594 #[serde(transparent)]
595 pub struct ViewName(pub String);
596
597 impl SchemaBridge for ViewName {
598 fn to_ts() -> String {
599 format!("{:?}", crate::knl::projection::VIEW_TAIL)
600 }
601 fn to_schema() -> Schema {
602 Schema::Enum(vec![crate::knl::projection::VIEW_TAIL.to_string()])
603 }
604 }
605
606 /// What a named fold takes: `tail`'s `n`, and nothing else.
607 #[derive(Debug, Clone, Default, PartialEq, Serialize, Deserialize, SchemaBridge)]
608 #[serde(deny_unknown_fields)]
609 pub struct ViewOpts {
610 /// How many events from the end. Absent is the kernel's default.
611 #[serde(default, skip_serializing_if = "Option::is_none")]
612 pub n: Option<u64>,
613 }
614
615 /// The backend a session's log lives in, when the caller names one.
616 ///
617 /// `"mem"` is an in-memory database that lives as long as the session
618 /// does — tests and mocks: one session, one process, nothing shared;
619 /// `{ sqlite = "<path>" }` is a stream in a file the caller picked.
620 /// Naming neither is the host's database, which is where a session that
621 /// is not a test belongs (see the module header).
622 #[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
623 #[serde(
624 untagged,
625 expecting = r#"a store: "mem", or a table { sqlite = <path> }"#
626 )]
627 pub enum StoreSpec {
628 /// A backend named by a word — `"mem"`, and nothing else.
629 Named(String),
630 /// A durable stream at a path.
631 File(SqliteStore),
632 }
633
634 impl SchemaBridge for StoreSpec {
635 fn to_ts() -> String {
636 r#""mem" | { sqlite: string }"#.to_string()
637 }
638 fn to_schema() -> Schema {
639 Schema::Union(vec![
640 Schema::Enum(vec![MEM_STORE.to_string()]),
641 SqliteStore::to_schema(),
642 ])
643 }
644 }
645
646 /// The in-memory backend, by name.
647 pub const MEM_STORE: &str = "mem";
648
649 /// `{ sqlite = "<path>" }` — the durable half of [`StoreSpec`].
650 #[derive(Debug, Clone, PartialEq, Serialize, Deserialize, SchemaBridge)]
651 #[serde(deny_unknown_fields)]
652 pub struct SqliteStore {
653 /// Where the database file is.
654 pub sqlite: String,
655 }
656
657 /// What `opts.budget` asked for.
658 ///
659 /// One table with both forms in it, because that is what a caller writes
660 /// and because the refusal for writing both has to name both. Which of
661 /// the two a given table *is* — an owner's grant (`amount`) or an
662 /// allocation out of a parent's balance (`from_parent`) — is decided
663 /// after the parse, where the two can be named against each other.
664 #[derive(Debug, Clone, Default, PartialEq, Serialize, Deserialize)]
665 #[serde(deny_unknown_fields)]
666 pub struct BudgetOpt {
667 /// What an owner allows this session, out of nothing.
668 #[serde(default, skip_serializing_if = "Option::is_none")]
669 pub amount: Option<i64>,
670 /// What the unit is called. The kernel reads the number and this
671 /// rides onto `budget_granted` verbatim.
672 #[serde(default, skip_serializing_if = "Option::is_none")]
673 pub tag: Option<String>,
674 /// What was allowed and why. A grant's alone: an allocation records
675 /// the parent it came from, which is the whole of what the kernel
676 /// knows about why it happened.
677 #[serde(default, skip_serializing_if = "Option::is_none")]
678 pub desc: Option<String>,
679 /// What the parent named in `opts.parent` hands over out of its own
680 /// balance.
681 #[serde(default, skip_serializing_if = "Option::is_none")]
682 pub from_parent: Option<i64>,
683 }
684
685 impl SchemaBridge for BudgetOpt {
686 fn to_ts() -> String {
687 format!("{} | {}", BudgetGrant::to_ts(), BudgetAllocation::to_ts())
688 }
689 /// The union the field really is, rather than the flat table it is
690 /// parsed as: a reader of the schema should see that the two forms
691 /// exclude each other, which the parse form cannot say.
692 fn to_schema() -> Schema {
693 Schema::Union(vec![
694 BudgetGrant::to_schema(),
695 BudgetAllocation::to_schema(),
696 ])
697 }
698 }
699
700 /// `{ amount, tag?, desc? }` — a balance appearing, which only an owner
701 /// may do.
702 #[derive(Debug, Clone, PartialEq, Serialize, Deserialize, SchemaBridge)]
703 pub struct BudgetGrant {
704 /// A whole number of units.
705 pub amount: i64,
706 /// What the unit is called.
707 #[serde(default, skip_serializing_if = "Option::is_none")]
708 pub tag: Option<String>,
709 /// What was allowed and why.
710 #[serde(default, skip_serializing_if = "Option::is_none")]
711 pub desc: Option<String>,
712 }
713
714 /// `{ from_parent, tag? }` — a balance changing hands: the parent's falls
715 /// by exactly what the child's rises by, in one write.
716 #[derive(Debug, Clone, PartialEq, Serialize, Deserialize, SchemaBridge)]
717 pub struct BudgetAllocation {
718 /// A whole number of units, out of the parent's balance.
719 pub from_parent: i64,
720 /// What the unit is called (the parent's, when it is left out).
721 #[serde(default, skip_serializing_if = "Option::is_none")]
722 pub tag: Option<String>,
723 }
724
725 /// `knl.open(opts?)` — state only. Policy has its own constructor.
726 #[derive(Debug, Clone, Default, PartialEq, Serialize, Deserialize, SchemaBridge)]
727 #[serde(deny_unknown_fields)]
728 pub struct OpenOpts {
729 /// The principal the session belongs to. Absent is the reserved
730 /// anonymous id, so the layer above always has a real key to read.
731 #[serde(default, skip_serializing_if = "Option::is_none")]
732 pub owner: Option<String>,
733 /// The quota, and where it came from.
734 #[serde(default, skip_serializing_if = "Option::is_none")]
735 pub budget: Option<BudgetOpt>,
736 /// Where the log lives. Absent is the host's database — except
737 /// for a child, which goes where its parent already is.
738 #[serde(default, skip_serializing_if = "Option::is_none")]
739 pub store: Option<StoreSpec>,
740 /// The session this one is opened *from*.
741 ///
742 /// Declared, never read here. The value is the kernel's own
743 /// userdata — a live handle rather than data — so the key is taken
744 /// off the table before the rest is deserialized (`without_parent`)
745 /// and the handle itself is read directly (`parse_parent`). The
746 /// field stays because the *shape* has to say that a parent is part
747 /// of `open`; it is `any` because no data schema can describe a
748 /// handle.
749 #[serde(default, skip_serializing_if = "Option::is_none")]
750 pub parent: Option<Json>,
751 }
752
753 /// `knl.resume(opts)` — reopen a stream and re-fold it.
754 #[derive(Debug, Clone, PartialEq, Serialize, Deserialize, SchemaBridge)]
755 #[serde(deny_unknown_fields)]
756 pub struct ResumeOpts {
757 /// Where the stream lives. Absent means the same thing it does on
758 /// open — the host's database — so a session opened without a store
759 /// is resumed by its id alone.
760 #[serde(default, skip_serializing_if = "Option::is_none")]
761 pub store: Option<StoreSpec>,
762 /// The stream to reopen.
763 pub session: String,
764 /// The owner granting *again*: recorded and added to the balance the
765 /// log already carries, rather than replacing it.
766 #[serde(default, skip_serializing_if = "Option::is_none")]
767 pub budget: Option<BudgetOpt>,
768 }
769
770 /// What `s:append` records: the envelope, and nothing beside it.
771 ///
772 /// Open, and deliberately: the kernel stamps `seq` / `epoch_ms` /
773 /// `_schema_version` on a stored event, so what comes back out of
774 /// `events()` carries more keys than what went in. The closure — no
775 /// other top-level key — is the kernel's, enforced at the syscall where
776 /// the stamps are known.
777 #[derive(Debug, Clone, PartialEq, Serialize, Deserialize, SchemaBridge)]
778 pub struct AppendEvent {
779 /// What happened.
780 pub kind: String,
781 /// The beat this event belongs to, when the caller declared one.
782 #[serde(default, skip_serializing_if = "Option::is_none")]
783 pub beat: Option<String>,
784 /// Shallow labels a view can group or filter on.
785 #[serde(default, skip_serializing_if = "Option::is_none")]
786 pub meta: Option<Meta>,
787 /// What the kind is about. An empty table when none was written.
788 #[serde(default, skip_serializing_if = "Option::is_none")]
789 pub data: Option<Json>,
790 }
791
792 /// One recorded event, as it comes back out.
793 #[derive(Debug, Clone, PartialEq, Serialize, Deserialize, SchemaBridge)]
794 pub struct EventRow {
795 /// What happened.
796 pub kind: String,
797 /// Where in the stream, assigned by the kernel.
798 pub seq: u64,
799 /// When, assigned by the kernel.
800 pub epoch_ms: u64,
801 /// Which revision of the event vocabulary this was read through.
802 pub _schema_version: u64,
803 /// The beat this event belongs to, when one was declared.
804 #[serde(default, skip_serializing_if = "Option::is_none")]
805 pub beat: Option<String>,
806 /// The shallow labels it was written with.
807 #[serde(default, skip_serializing_if = "Option::is_none")]
808 pub meta: Option<Meta>,
809 /// What the kind is about, as written.
810 pub data: Json,
811 }
812
813 /// The record from a position on, as far as one read goes.
814 #[derive(Debug, Clone, PartialEq, Serialize, Deserialize, SchemaBridge)]
815 #[serde(transparent)]
816 pub struct EventRows(pub Vec<EventRow>);
817
818 /// What `s:events(from?)` answers: the rows, and whether the row cap cut
819 /// the read short.
820 ///
821 /// A pair for the same reason [`QueryResult`] is one — that is what the
822 /// call returns, two values — and it carries the same second value for the
823 /// same reason: a read that stopped at the cap and a read that reached the
824 /// end of the stream are different facts, and a caller folding a request
825 /// out of the rows has to be able to tell them apart.
826 #[derive(Debug, Clone, PartialEq, Serialize, Deserialize, SchemaBridge)]
827 pub struct EventsResult(pub EventRows, pub bool);
828
829 /// The values a statement binds.
830 ///
831 /// A list is the values for the `?` parameters, in order; a table with
832 /// names is the values for `:name` / `@name` / `$name`. A statement is
833 /// written one way or the other and the two are not mixed.
834 #[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
835 #[serde(
836 untagged,
837 expecting = "the values a statement binds: a list for `?`, or a table of names"
838 )]
839 pub enum QueryParams {
840 /// Values for the anonymous `?` parameters.
841 Positional(Vec<Value>),
842 /// Values for the named ones.
843 Named(Map<String, Value>),
844 }
845
846 impl SchemaBridge for QueryParams {
847 fn to_ts() -> String {
848 "unknown[] | Record<string, unknown>".to_string()
849 }
850 fn to_schema() -> Schema {
851 Schema::Union(vec![
852 Schema::Array(Box::new(Schema::Any)),
853 Schema::Record {
854 key: Box::new(Schema::String),
855 value: Box::new(Schema::Any),
856 },
857 ])
858 }
859 }
860
861 /// What a caller asks for beyond the SQL itself.
862 ///
863 /// Closed: an option the kernel does not know must not quietly do
864 /// nothing, which is exactly what a misspelt `limit` or `timeout_ms`
865 /// would do.
866 #[derive(Debug, Clone, Default, PartialEq, Serialize, Deserialize, SchemaBridge)]
867 #[serde(deny_unknown_fields)]
868 pub struct QueryOpts {
869 /// The streams `$sessions` expands to. Omitted is this session's own
870 /// stream and nothing else.
871 #[serde(default, skip_serializing_if = "Option::is_none")]
872 pub sessions: Option<Vec<String>>,
873 /// How long the read may run. Absent is the kernel's default.
874 #[serde(default, skip_serializing_if = "Option::is_none")]
875 pub timeout_ms: Option<u64>,
876 /// How many rows before the rest are cut off. Absent is the
877 /// kernel's default.
878 #[serde(default, skip_serializing_if = "Option::is_none")]
879 pub limit: Option<u64>,
880 }
881
882 /// What `s:query` answers: the rows, and whether the cap cut any off.
883 ///
884 /// A pair rather than a table, because that is what the call returns —
885 /// two values, so a page can be told from a complete answer without
886 /// unwrapping anything.
887 #[derive(Debug, Clone, PartialEq, Serialize, Deserialize, SchemaBridge)]
888 pub struct QueryResult(pub Vec<Json>, pub bool);
889
890 /// A raised kernel failure, read back as data (`knl.error(e)`).
891 ///
892 /// `kind` and `method` are optional because a raise that carried no
893 /// attribution is reported whole rather than rejected: `message` then
894 /// holds the entire text. So `message` is the field a reader can always
895 /// count on, and `kind` is the one it must ask for.
896 #[derive(Debug, Clone, PartialEq, Serialize, Deserialize, SchemaBridge)]
897 pub struct ErrorTable {
898 /// The class, when the raise carried one.
899 #[serde(default, skip_serializing_if = "Option::is_none")]
900 pub kind: Option<String>,
901 /// The method that raised, when the raise carried one.
902 #[serde(default, skip_serializing_if = "Option::is_none")]
903 pub method: Option<String>,
904 /// The kernel's own judgement, true for contention alone.
905 pub retryable: bool,
906 /// What went wrong, in prose.
907 pub message: String,
908 }
909
910 /// One entry of the declared surface: a name and the contract it holds.
911 #[derive(Debug, Clone, PartialEq, Serialize, Deserialize, SchemaBridge)]
912 pub struct ApiEntry {
913 /// What Lua calls it.
914 pub name: String,
915 /// The contract, in a line.
916 pub doc: String,
917 }
918
919 /// One column of the table a query reads.
920 ///
921 /// The one type here whose schema is written out rather than derived.
922 /// `type` is a Rust keyword, so the field is `declared_type` and serde is
923 /// told to rename it — and `#[derive(SchemaBridge)]` reads
924 /// `serde(rename_all)` but not a per-field `serde(rename)`, so the derive
925 /// would declare `declared_type` while the value carries `type`. That is
926 /// exactly the drift these types exist to remove, and the generated-shape
927 /// test caught it, so the schema says what serde does.
928 #[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
929 pub struct ApiColumn {
930 /// What SQL calls it.
931 pub name: String,
932 /// How SQLite declared it.
933 #[serde(rename = "type")]
934 pub declared_type: String,
935 /// Whether it is part of the primary key.
936 pub pk: bool,
937 }
938
939 impl SchemaBridge for ApiColumn {
940 fn to_ts() -> String {
941 "{ name: string; type: string; pk: boolean; }".to_string()
942 }
943 fn to_schema() -> Schema {
944 Schema::Object(vec![
945 Field::new("name", Schema::String),
946 Field::new("type", Schema::String),
947 Field::new("pk", Schema::Boolean),
948 ])
949 }
950 }
951
952 /// The read contract: the table a query names, and the columns it has.
953 #[derive(Debug, Clone, PartialEq, Serialize, Deserialize, SchemaBridge)]
954 pub struct ApiSchema {
955 /// The table the events live in.
956 pub table: String,
957 /// Its columns, as SQLite reports them.
958 pub columns: Vec<ApiColumn>,
959 }
960
961 /// The `data` field names of the kinds the kernel writes itself.
962 ///
963 /// The columns are published above ([`ApiSchema`]) and they are only half
964 /// of what a view has to spell: everything a `budget_*` or `session_*`
965 /// event is *about* lives inside the `data` column, and a Lua view reaches
966 /// it with a `json_extract` path — `knl.views.ledger` reads `$.amount` and
967 /// `$.tag`, `knl.views.tree` reads `$.parent` and `$.open_children`. Those
968 /// paths are the Rust `FIELD_*` constants spelled out in SQL, in another
969 /// language, in another file; nothing held them together, so a rename here
970 /// would have left the view answering NULL for every row.
971 ///
972 /// So the names are published from the constants themselves, and the view
973 /// is held against them where a store exists
974 /// (`tests/fixtures/knl_beat_test.lua`, inv11) — the same two-sided
975 /// arrangement the columns and the error classes already have.
976 #[derive(Debug, Clone, PartialEq, Serialize, Deserialize, SchemaBridge)]
977 pub struct ApiFields {
978 /// `budget_*`: how much, in the grant's unit.
979 pub amount: String,
980 /// `budget_*`: the grant's unit, when it named one.
981 pub tag: String,
982 /// `budget_granted`: the owner's free-text note.
983 pub desc: String,
984 /// `budget_refused`: the balance the refusal was measured against.
985 pub remaining: String,
986 /// `session_opened` / `budget_*`: the authority it was written under.
987 pub scope_id: String,
988 /// `session_opened`: the principal the scope belongs to.
989 pub owner: String,
990 /// `session_opened` / `budget_granted`: the stream this was opened
991 /// from.
992 pub parent: String,
993 /// `budget_reserved` / `budget_refused`: the stream the units went to.
994 pub child: String,
995 /// `session_closed`: which kind of ending it was.
996 pub reason: String,
997 /// `session_closed`: the sentence only that close could tell.
998 pub detail: String,
999 /// `session_closed`: the children that had not ended when it did.
1000 pub open_children: String,
1001 }
1002
1003 /// What `knl.api()` answers: the whole declared surface, as data.
1004 #[derive(Debug, Clone, PartialEq, Serialize, Deserialize, SchemaBridge)]
1005 pub struct ApiReport {
1006 /// Every method the session userdata answers to.
1007 pub session: Vec<ApiEntry>,
1008 /// Every function the `knl` global carries.
1009 pub module: Vec<ApiEntry>,
1010 /// The closed list of classes `knl.error(e).kind` can report.
1011 pub errors: Vec<String>,
1012 /// The columns a query may name.
1013 pub schema: ApiSchema,
1014 /// The `data` paths a view reaches into, as the kernel spells them.
1015 pub fields: ApiFields,
1016 /// The generated `knl_types` module, as source text — the same one
1017 /// the host embeds, for a tool that wants to read the surface
1018 /// without loading it.
1019 pub types: String,
1020 }
1021
1022 /// Whether a stray key in a table is a violation or a pass-through.
1023 ///
1024 /// lshape's `T.shape` is open by default, which is right for the tables
1025 /// a caller writes and wrong for the two that are contracts: an option
1026 /// the kernel does not know must not quietly do nothing.
1027 #[derive(Debug, Clone, Copy, PartialEq, Eq)]
1028 pub enum Strict {
1029 /// Extra keys pass (lshape's default).
1030 Open,
1031 /// Extra keys are a violation.
1032 Closed,
1033 }
1034
1035 /// Every type the syscall surface is declared in, with the name Lua sees.
1036 ///
1037 /// This list *is* the module `knl_types` — nothing is generated that is
1038 /// not here, and a test holds every entry of it against a reference from
1039 /// the Lua registry, so a type nobody declares and a declaration with no
1040 /// type are both failures.
1041 pub fn declared() -> Vec<(&'static str, Schema, Strict)> {
1042 vec![
1043 ("SessionId", SessionId::to_schema(), Strict::Open),
1044 ("ScopeId", ScopeId::to_schema(), Strict::Open),
1045 ("Owner", Owner::to_schema(), Strict::Open),
1046 ("BeatId", BeatId::to_schema(), Strict::Open),
1047 ("Seq", Seq::to_schema(), Strict::Open),
1048 ("Count", Count::to_schema(), Strict::Open),
1049 ("Amount", Amount::to_schema(), Strict::Open),
1050 ("Remaining", Remaining::to_schema(), Strict::Open),
1051 ("Exhausted", Exhausted::to_schema(), Strict::Open),
1052 ("Sql", Sql::to_schema(), Strict::Open),
1053 ("CloseReason", CloseReason::to_schema(), Strict::Open),
1054 ("CloseDetail", CloseDetail::to_schema(), Strict::Open),
1055 ("Raised", Raised::to_schema(), Strict::Open),
1056 ("ViewName", ViewName::to_schema(), Strict::Open),
1057 ("ViewOpts", ViewOpts::to_schema(), Strict::Closed),
1058 ("OpenOpts", OpenOpts::to_schema(), Strict::Open),
1059 ("ResumeOpts", ResumeOpts::to_schema(), Strict::Open),
1060 ("AppendEvent", AppendEvent::to_schema(), Strict::Open),
1061 ("EventsResult", EventsResult::to_schema(), Strict::Open),
1062 ("QueryParams", QueryParams::to_schema(), Strict::Open),
1063 ("QueryOpts", QueryOpts::to_schema(), Strict::Closed),
1064 ("QueryResult", QueryResult::to_schema(), Strict::Open),
1065 ("ErrorTable", ErrorTable::to_schema(), Strict::Open),
1066 ("ApiReport", ApiReport::to_schema(), Strict::Open),
1067 ]
1068 }
1069}
1070
1071/// The `knl_types` Lua module, as source text.
1072///
1073/// Generated from [`types::declared`] at every host start rather than checked
1074/// in: a generated file in the tree is a file that can be edited, and one that
1075/// has been edited is a second declaration wearing the first one's name. The
1076/// host adds it to the embedded module set ([`crate::host`]) so the Lua kernel
1077/// can `require("knl_types")`, and `knl.api().types` hands back this same text
1078/// for tooling that wants to read the surface without loading it.
1079///
1080/// `schema_bridge_lshape::generate_lshape_file` would do all of this in one
1081/// call, except that it has no way to emit lshape's strict mode
1082/// ([`types::Strict::Closed`]) — so the module is assembled here from that
1083/// crate's per-schema renderer, in the same layout, and the two tables whose
1084/// extra keys are violations get the option appended.
1085///
1086/// **Built once per process.** It is a pure function of types that are fixed
1087/// at compile time, so the text it answers can only be one text; rendering
1088/// every declared schema again on each call — the host's start, every
1089/// `knl.api()` — was work whose result was known. The `OnceLock` holds it and
1090/// the call hands back a copy, which keeps the signature (and the caller that
1091/// hands the source to `Lua::load`) exactly as it was.
1092pub fn lshape_module_source() -> String {
1093 static SOURCE: std::sync::OnceLock<String> = std::sync::OnceLock::new();
1094 SOURCE.get_or_init(build_lshape_module_source).clone()
1095}
1096
1097/// The body of [`lshape_module_source`], run once behind its `OnceLock`.
1098fn build_lshape_module_source() -> String {
1099 #[cfg(test)]
1100 TYPES_BUILDS.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
1101 let mut out = String::from(
1102 "-- Generated at host start by agent-block-core from the argument and\n\
1103 -- return types of `bridge/knl.rs` (schema-bridge -> lshape). Not a file\n\
1104 -- in the tree: there is nothing here to edit, and so nothing to drift.\n\
1105 local T = require(\"lshape\").t\n\nlocal M = {}\n\n",
1106 );
1107 for (name, schema, strict) in types::declared() {
1108 let body = schema_bridge_lshape::schema_to_lshape(&schema)
1109 .unwrap_or_else(|e| unreachable!("knl_types {name} does not map to lshape: {e}"));
1110 let body = match strict {
1111 types::Strict::Open => body,
1112 types::Strict::Closed => close_shape(name, body),
1113 };
1114 out.push_str(&format!("M.{name} = {}\n\n", named_scalar(name, body)));
1115 }
1116 out.push_str("return M\n");
1117 out
1118}
1119
1120/// A type whose schema is a bare primitive, given its name back.
1121///
1122/// `SessionId` and `Owner` are both strings, and lshape's `T.string` is one
1123/// table: rendered as they stand, the two names would be the same value, and
1124/// a registry that says `id() -> SessionId` and `owner() -> Owner` would be
1125/// saying one thing twice — with no way left to tell a type nobody references
1126/// from one that is referenced under another name. `:describe` wraps the
1127/// primitive in a node of its own carrying the name, which `check` passes
1128/// straight through and `reflect` reads back, so the declaration keeps the
1129/// distinction the Rust type made.
1130///
1131/// Only the bare case: anything with a combinator in it (`T.one_of({…})`,
1132/// `T.shape({…})`, `T.integer:is_optional()`) is already a fresh table.
1133fn named_scalar(name: &str, body: String) -> String {
1134 let bare = body
1135 .strip_prefix("T.")
1136 .is_some_and(|rest| !rest.is_empty() && rest.chars().all(|c| c.is_ascii_lowercase()));
1137 if bare {
1138 format!("{body}:describe({name:?})")
1139 } else {
1140 body
1141 }
1142}
1143
1144/// `T.shape({ … })` with lshape's strict mode turned on.
1145///
1146/// A text edit rather than a generator option because the generator has none
1147/// (schema-bridge-lshape 0.2 renders `T.shape(fields)` and stops). It is
1148/// exact rather than approximate: the renderer's output for an object ends in
1149/// `})` and in nothing else, so the tail is replaced rather than searched for,
1150/// and a schema that did not render as a shape is a mistake in
1151/// [`types::declared`] rather than something to paper over.
1152fn close_shape(name: &str, body: String) -> String {
1153 let Some(fields) = body.strip_suffix("})") else {
1154 unreachable!("knl_types {name} is marked strict but did not render as a T.shape: {body}");
1155 };
1156 format!("{fields}}}, {{ open = false }})")
1157}
1158
1159/// Read a Lua value as `T`, attributing a refusal to `method`.
1160///
1161/// The type is the check. What a syscall accepts used to be a hand-written
1162/// walk of the table — one per argument, each with its own idea of how to say
1163/// "that is not a string" — and this is the whole of it now: the same types
1164/// the surface is declared in are the ones a caller's table is read into, so
1165/// the check and the declaration cannot disagree.
1166///
1167/// Three things make the refusal readable:
1168///
1169/// - `noun` names the argument (`opts`, `event`, `budget`), and
1170/// `serde_path_to_error` adds the field the deserializer was at, so a caller
1171/// gets `budget.tag: invalid type: number, expected a string` rather than
1172/// the leaf message alone;
1173/// - the class is [`knl::KnlError::VALIDATION`], the same one the kernel's own
1174/// validator uses, so the shell sees one vocabulary either side of the
1175/// boundary;
1176/// - nothing is turned off. A value serde has no representation for — a
1177/// function where a string belonged — is refused rather than skipped, which
1178/// is what `lua_to_json` has always done on the way to the store. The one
1179/// value that legitimately cannot cross is `opts.parent`, and it is lifted
1180/// off the table before this runs ([`without_parent`]) rather than bought
1181/// with a deserializer that ignores every other one too.
1182fn from_lua<T: serde::de::DeserializeOwned>(
1183 method: &str,
1184 noun: &str,
1185 value: LuaValue,
1186) -> LuaResult<T> {
1187 let de = mlua::serde::Deserializer::new(value);
1188 serde_path_to_error::deserialize(de).map_err(|error| {
1189 let path = error.path().to_string();
1190 let at = if path.is_empty() || path == "." {
1191 noun.to_string()
1192 } else {
1193 format!("{noun}.{path}")
1194 };
1195 // mlua renders every deserializer failure as `deserialize error: …`.
1196 // The class is already the third field of the attribution, so the
1197 // prefix would be the message saying twice what it is and once what
1198 // went wrong.
1199 let reason = error.into_inner().to_string();
1200 let reason = reason
1201 .strip_prefix("deserialize error: ")
1202 .unwrap_or(&reason)
1203 .to_string();
1204 err(method, format!("{at}: {reason}"))
1205 })
1206}
1207
1208/// K5 session: the only handle the Lua side has on kernel state.
1209struct Session {
1210 // A `tokio::sync::Mutex`, and not the `RefCell` this used to be. Every
1211 // method that reaches the store yields now, so a second coroutine on the
1212 // same VM can call one while the first is suspended in the middle of
1213 // another — which a `RefCell` answers by panicking. An async lock answers
1214 // it by making the second call wait for the first, which is the same
1215 // serialization the kernel already promises per stream.
1216 //
1217 // Holding the guard across the store's `.await` is the point rather than a
1218 // hazard: a session's own calls are meant to be one at a time, and the
1219 // lock's whole job is to say so.
1220 state: Mutex<knl::Session>,
1221 /// The three identity reads, copied out at construction.
1222 ///
1223 /// `id` / `scope_id` / `owner` are immutable once this userdata exists —
1224 /// the stream is adopted before the value is built (`open_sqlite`,
1225 /// `resume_on`, `Session::open_child`, and `knl::Session::new` inside
1226 /// itself), and neither the scope id nor the owner has a setter at all —
1227 /// so the answer is a field here rather than a read behind the lock.
1228 ///
1229 /// That is what makes those three methods *never raise*, which is what
1230 /// [`SESSION_API`] says about them. Behind the lock they could not:
1231 /// a `try_lock` has an answer for "somebody is mid-call" and every answer
1232 /// to that is wrong for an identity read — raising turns `s:id()` into a
1233 /// call a caller has to handle, and waiting is the one thing a
1234 /// synchronous method on the VM's thread must not do.
1235 identity: Identity,
1236}
1237
1238/// What a session answers about itself without touching the store.
1239///
1240/// Fixed at construction and never written again, so the reads are plain
1241/// field reads and the lock stays for the calls that actually reach the log.
1242struct Identity {
1243 /// The stream this session writes (`s:id()`).
1244 id: String,
1245 /// The authority the stream is written under (`s:scope_id()`).
1246 scope_id: String,
1247 /// The principal the scope belongs to (`s:owner()`).
1248 owner: String,
1249}
1250
1251impl Session {
1252 /// Wrap a kernel session as the Lua userdata.
1253 ///
1254 /// The identity is read *here*, which is why every caller adopts the
1255 /// stream id before handing the session over: after this line the three
1256 /// values are the userdata's own, and the kernel session's copy of them
1257 /// can no longer be reached from Lua.
1258 fn from_state(state: knl::Session) -> Self {
1259 let identity = Identity {
1260 id: state.id().to_string(),
1261 scope_id: state.scope_id().to_string(),
1262 owner: state.owner().to_string(),
1263 };
1264 Self {
1265 state: Mutex::new(state),
1266 identity,
1267 }
1268 }
1269
1270 /// Open a session for `owner` with an optional budget grant, on the
1271 /// in-memory store.
1272 async fn new(
1273 owner: String,
1274 grant: Option<knl::BudgetGrant>,
1275 drivers: &knl::IsleDrivers,
1276 ) -> LuaResult<Self> {
1277 let state = knl::Session::new(owner, grant, drivers)
1278 .await
1279 .map_err(|e| knl_err("open", &e))?;
1280 Ok(Self::from_state(state))
1281 }
1282}
1283
1284/// The backstop under `close` and `<close>`: a handle nobody ended still
1285/// records the session's boundary, here, where the value dies.
1286///
1287/// A dropped handle is the one close path with no caller left to tell, and now
1288/// also the one with nowhere to wait: `Drop` cannot be `async`, and this runs
1289/// on the VM's own thread inside a Lua collection cycle, where blocking on
1290/// SQLite would stop every other coroutine, timer and cancellation that VM
1291/// owns. So the boundary is *submitted* rather than awaited
1292/// ([`knl::Session::close_detached`]): the event goes to the connection
1293/// thread, whose driver the host holds, and lands there while nothing waits
1294/// for it.
1295///
1296/// The lock is taken with `try_lock`, not awaited: a session still borrowed by
1297/// a suspended call has an owner, and this collection cycle is not it. A
1298/// failure is a `warn!` and nothing else — panicking in `drop` would abort the
1299/// process, and a session already past its last reader is not worth that.
1300impl Drop for Session {
1301 fn drop(&mut self) {
1302 // `get_mut` rather than a lock at all: `Drop` has `&mut self`, so no
1303 // other holder of the session can exist by definition.
1304 let state = self.state.get_mut();
1305 if state.is_closed() {
1306 return;
1307 }
1308 state.close_detached(knl::CLOSE_REASON_DROPPED);
1309 }
1310}
1311
1312/// The longest `detail` a close records, in characters.
1313///
1314/// A `session_closed` says why a session ended, and an error message can be
1315/// a whole traceback; the cap keeps one bad turn from putting a page into
1316/// the log. Counted in `chars` so the cut never lands inside one.
1317const DETAIL_MAX_CHARS: usize = 200;
1318
1319/// `text` cut to [`DETAIL_MAX_CHARS`], with an ellipsis when it was cut.
1320///
1321/// One rule for both close paths: what `<close>` records off a raised error
1322/// and what a caller passes to `close(reason, detail)` are capped the same
1323/// way, so the log cannot grow a page-long entry from either side.
1324fn truncated(text: &str) -> String {
1325 if text.chars().count() <= DETAIL_MAX_CHARS {
1326 return text.to_string();
1327 }
1328 text.chars().take(DETAIL_MAX_CHARS).collect::<String>() + "..."
1329}
1330
1331/// The error a `<close>` scope was unwinding with, as `detail` text.
1332///
1333/// Read without re-entering Lua (no `__tostring` call): the value arrives
1334/// while the VM is already unwinding, and a metamethod raising there would
1335/// replace the error the log is trying to record.
1336fn error_detail(error: &LuaValue) -> String {
1337 let text = match error {
1338 LuaValue::String(s) => s.to_string_lossy(),
1339 LuaValue::Error(e) => e.to_string(),
1340 LuaValue::Integer(i) => i.to_string(),
1341 LuaValue::Number(n) => n.to_string(),
1342 other => format!("<{}>", other.type_name()),
1343 };
1344 truncated(&text)
1345}
1346
1347/// The `knl: <method>: <kind>: <reason>` attribution, as text.
1348///
1349/// Four fields in a fixed order, and the first three are a closed
1350/// vocabulary: the prefix, the method the caller invoked, and the class of
1351/// the failure ([`knl::KnlError::KINDS`]). Only the fourth is prose. That
1352/// is what lets [`error_table`] hand the same four fields back as a table
1353/// without the Lua side matching on a sentence that is meant to change.
1354fn attributed(method: &str, kind: &str, reason: impl std::fmt::Display) -> String {
1355 format!("knl: {method}: {kind}: {reason}")
1356}
1357
1358/// Build a `knl:`-attributed error of `kind` for `method`.
1359fn err_of(method: &str, kind: &str, reason: impl std::fmt::Display) -> LuaError {
1360 LuaError::external(attributed(method, kind, reason))
1361}
1362
1363/// The bridge refusing what it was handed: a `validation` failure.
1364///
1365/// Every refusal raised on this side of the boundary — a non-table event, a
1366/// misspelt budget field, an amount that is not a whole number — is the
1367/// caller's arguments not holding up, which is the same class the kernel
1368/// gives its own validator's refusals. So the shell sees one vocabulary,
1369/// whether the check ran in Rust's kernel or in its adapter.
1370fn err(method: &str, reason: impl std::fmt::Display) -> LuaError {
1371 err_of(method, knl::KnlError::VALIDATION, reason)
1372}
1373
1374/// A kernel failure, carrying the kernel's own classification outwards.
1375///
1376/// The bridge does not re-decide what went wrong: [`knl::KnlError::kind`]
1377/// already said, and this only renders it.
1378fn knl_err(method: &str, error: &knl::KnlError) -> LuaError {
1379 err_of(method, error.kind(), error.reason())
1380}
1381
1382/// Convert the Lua table `noun` into a JSON object for `method`.
1383///
1384/// Runs before any borrow of the session: the walk may re-enter Lua,
1385/// which must not observe a held borrow.
1386fn table_to_object(
1387 lua: &Lua,
1388 method: &str,
1389 noun: &str,
1390 value: LuaValue,
1391) -> LuaResult<Map<String, Value>> {
1392 if !matches!(value, LuaValue::Table(_)) {
1393 return Err(err(
1394 method,
1395 format!("{noun} must be a table, got {}", value.type_name()),
1396 ));
1397 }
1398 match lua_to_json(lua, value).map_err(|e| err(method, e))? {
1399 Value::Object(obj) => Ok(obj),
1400 _ => Err(err(
1401 method,
1402 format!("{noun} must be a table with string keys"),
1403 )),
1404 }
1405}
1406
1407impl LuaUserData for Session {
1408 fn add_methods<M: LuaUserDataMethods<Self>>(methods: &mut M) {
1409 // The three identity reads below are the only synchronous methods on
1410 // the session: each answers out of a field of the userdata itself
1411 // ([`Identity`]), touching neither the store nor the lock, which is
1412 // the work a sync `add_method` is still for. They cannot fail, and
1413 // that is a property of where the value is kept rather than a promise
1414 // made about a lock nobody was supposed to be holding. Everything
1415 // after them reaches the store, so everything after them yields.
1416
1417 // s:id() -> string
1418 methods.add_method("id", |_, this, ()| Ok(this.identity.id.clone()));
1419
1420 // s:scope_id() -> string
1421 //
1422 // The kernel-issued id of the scope this session is written under,
1423 // as recorded on `session_opened` and on every `budget_*` event.
1424 // Not `s:id()`: that names the stream, this names the authority the
1425 // stream is written under, and neither is a caller's to choose.
1426 methods.add_method("scope_id", |_, this, ()| Ok(this.identity.scope_id.clone()));
1427
1428 // s:owner() -> string
1429 //
1430 // The principal the scope belongs to (a real id, or the reserved
1431 // "anon" / "system"). Total — never nil.
1432 methods.add_method("owner", |_, this, ()| Ok(this.identity.owner.clone()));
1433
1434 // s:append(event) -> seq
1435 //
1436 // K1: the only way to add to the history, and there is no way to
1437 // change what is already in it. The event is the envelope
1438 // (`kind` / `beat?` / `meta?` / `data?`) and nothing beside it; what
1439 // is under `data` is recorded as written, `beat` included, and the
1440 // kernel adds `seq` / `epoch_ms` and an empty `data` when there was
1441 // none. No append touches the budget — that is `reserve` before the
1442 // call and `spend` after it — and the two `session_*` kinds are
1443 // refused here, since only `knl.open` / `close` write those.
1444 methods.add_async_method("append", |lua, this, event: LuaValue| async move {
1445 // Two readings of one table, and they answer different questions.
1446 //
1447 // The first is the type: `kind` is a string, `beat` is a string,
1448 // `meta` holds labels and nothing deeper — the contract
1449 // `knl_types.AppendEvent` publishes, checked here on every call in
1450 // both modes rather than only under the Lua dev gate, which a
1451 // direct `s:append(...)` never passes through.
1452 let _: types::AppendEvent = from_lua("append", "event", event.clone())?;
1453 // The second is the object the kernel records. It is the table
1454 // itself rather than the parse above, because the envelope's
1455 // closure is the kernel's rule and it is stated where the stamps
1456 // are known: a stray top-level key is refused there, and `seq` /
1457 // `epoch_ms` given by a caller are overwritten rather than
1458 // rejected. Both conversions run before the session is reached —
1459 // walking a Lua table can call back into Lua.
1460 let obj = table_to_object(&lua, "append", "event", event)?;
1461 this.state
1462 .lock()
1463 .await
1464 .append(obj)
1465 .await
1466 .map_err(|e| knl_err("append", &e))
1467 });
1468
1469 // s:events(from?) -> rows, truncated
1470 //
1471 // K1: the returned tables are freshly built from the stored JSON
1472 // on every call, so mutating them cannot reach kernel state.
1473 //
1474 // Bounded, and it says when it cut. A stream grows without bound and
1475 // every row of a read is decoded, upcasted and built into a Lua table
1476 // on the VM's own thread, so "the whole log" is not a size this call
1477 // can promise. It reads at most [`knl::DEFAULT_LIMIT`] rows — the
1478 // same knob `query` caps on — and answers the pair `query` answers:
1479 // the rows, and whether there were more. A caller that wants the rest
1480 // pages with `from`, and one that folds a request refuses a `true`
1481 // rather than folding a history whose newest events are missing (the
1482 // Lua kernel's `beat` does exactly that).
1483 methods.add_async_method("events", |lua, this, from: Option<u64>| async move {
1484 let selected = {
1485 let state = this.state.lock().await;
1486 // One more than the cap, so "there were more" is something the
1487 // read answered rather than something a round count implies.
1488 state
1489 .events(from.unwrap_or(0), knl::DEFAULT_LIMIT.saturating_add(1))
1490 .await
1491 .map_err(|e| knl_err("events", &e))?
1492 // The guard is released here, before the conversion below.
1493 };
1494 let truncated = selected.len() > knl::DEFAULT_LIMIT;
1495 // The events come out of the kernel as `Current` — the proof that
1496 // they were read through the upcaster seam — and that proof stops
1497 // at this boundary: what Lua gets is a table, so the objects are
1498 // taken back out here, at the one place they leave the kernel.
1499 let selected: Vec<Value> = selected
1500 .into_iter()
1501 .take(knl::DEFAULT_LIMIT)
1502 .map(|event| Value::Object(event.into_inner()))
1503 .collect();
1504 // The session is released above: json_to_lua re-enters Lua.
1505 let rows = json_to_lua(&lua, Value::Array(selected))?;
1506 Ok((rows, truncated))
1507 });
1508
1509 // s:len() -> number of recorded events
1510 methods.add_async_method("len", |_, this, ()| async move {
1511 let n = this
1512 .state
1513 .lock()
1514 .await
1515 .len()
1516 .await
1517 .map_err(|e| knl_err("len", &e))?;
1518 Ok(n as u64)
1519 });
1520
1521 // s:view(name, opts?) -> projection (fresh table each call)
1522 //
1523 // `tail` (`opts.n` events from the end), and that is the whole
1524 // vocabulary: an unknown name is an error, because a projection the
1525 // kernel does not name is the shell's to build — from
1526 // `events(from)`, or as a query view over the published schema.
1527 methods.add_async_method(
1528 "view",
1529 |lua, this, (name, opts): (LuaValue, LuaValue)| async move {
1530 // The name's *type* is settled here and its vocabulary is
1531 // not: which folds exist is the kernel's, and an unknown one
1532 // is refused there, in its own words.
1533 let types::ViewName(name) = from_lua("view", "name", name)?;
1534 let opts = view_opts(from_lua("view", "opts", opts)?);
1535 let value = {
1536 let mut state = this.state.lock().await;
1537 state
1538 .view(&name, opts.as_ref())
1539 .await
1540 .map_err(|e| knl_err("view", &e))?
1541 };
1542 // The session is released above: json_to_lua re-enters Lua.
1543 json_to_lua(&lua, value)
1544 },
1545 );
1546
1547 // s:query(sql, params?, opts?) -> rows, truncated
1548 //
1549 // The log read with SQL. `view` names the folds whose consumer is
1550 // the kernel's own; everything else — beats grouped, tool calls
1551 // paired with their results, a ledger — is a SELECT over the table
1552 // the events live in, whose columns `knl.api().schema` publishes.
1553 //
1554 // What the kernel keeps around it: one statement and it reads, a
1555 // connection that cannot write, values bound rather than pasted, a
1556 // deadline, a row cap. The second return says whether the cap cut
1557 // anything off, so a caller can tell a complete answer from a page.
1558 methods.add_async_method(
1559 "query",
1560 |lua, this, (sql, params, opts): (LuaValue, LuaValue, LuaValue)| async move {
1561 // All three are read before the session is reached: walking a
1562 // Lua table can re-enter Lua.
1563 let types::Sql(sql) = from_lua("query", "sql", sql)?;
1564 let params = query_params(from_lua("query", "params", params)?);
1565 let opts = query_opts(from_lua("query", "opts", opts)?);
1566
1567 let found = {
1568 let state = this.state.lock().await;
1569 state
1570 .query(&sql, params, &opts)
1571 .await
1572 .map_err(|e| knl_err("query", &e))?
1573 };
1574 let rows: Vec<Value> = found.rows.into_iter().map(Value::Object).collect();
1575 // The session is released above: json_to_lua re-enters Lua.
1576 let rows = json_to_lua(&lua, Value::Array(rows))?;
1577 Ok((rows, found.truncated))
1578 },
1579 );
1580
1581 // s:reserve(n) -> true | false, tag
1582 //
1583 // K4, the decision point: ask before spending. `true` means the
1584 // amount was taken off the balance; `false` means it would not fit
1585 // and *nothing* was taken, with the grant's `tag` as the second
1586 // return so a caller can name the allowance that stopped it
1587 // without reading the log. Always `true` without a budget.
1588 methods.add_async_method("reserve", |_, this, amount: LuaValue| async move {
1589 // Whole is the type's business, non-negative is the kernel's:
1590 // the balance rule belongs where the balance is.
1591 let types::Amount(amount) = from_lua("reserve", "amount", amount)?;
1592 let mut state = this.state.lock().await;
1593 let granted = state
1594 .reserve(amount)
1595 .await
1596 .map_err(|e| knl_err("reserve", &e))?;
1597 // The tag rides along only on a refusal: it answers "which
1598 // budget stopped you", which is a question only then.
1599 let tag = if granted {
1600 None
1601 } else {
1602 state.grant().and_then(|grant| grant.tag.clone())
1603 };
1604 Ok((granted, tag))
1605 });
1606
1607 // s:spend(n) — the deduction that does not ask.
1608 //
1609 // K4. Non-negative amounts only, and the balance never rises. It
1610 // holds nothing for `reserve` to release and releases nothing
1611 // `reserve` held: a beat that calls both is charged twice.
1612 //
1613 // It returns nothing. It used to hand back the balance it read
1614 // afterwards, which made a deduction that landed and then failed its
1615 // read-back indistinguishable from one that never landed — the caller
1616 // saw an error either way and could not tell whether the `budget_spent`
1617 // was in the log. Two questions, two calls: this one raises only if
1618 // the write itself failed, and `s:remaining()` answers the other.
1619 methods.add_async_method("spend", |_, this, amount: LuaValue| async move {
1620 let types::Amount(amount) = from_lua("spend", "amount", amount)?;
1621 this.state
1622 .lock()
1623 .await
1624 .spend(amount)
1625 .await
1626 .map_err(|e| knl_err("spend", &e))
1627 });
1628
1629 // s:remaining() -> number or nil (no budget)
1630 //
1631 // Raises when the ledger cannot be read: a store that is down has no
1632 // balance to report, and both values this could otherwise return —
1633 // a stale number, or the nil that means "no budget here" — read as
1634 // facts a run would carry on spending against.
1635 methods.add_async_method("remaining", |_, this, ()| async move {
1636 this.state
1637 .lock()
1638 .await
1639 .remaining()
1640 .await
1641 .map_err(|e| knl_err("remaining", &e))
1642 });
1643
1644 // s:exhausted() -> boolean (always false without a budget)
1645 //
1646 // Raises for the same reason `remaining` does: a `false` that meant
1647 // "the store could not be read" is the one answer a run must never
1648 // be handed, because it reads as "carry on".
1649 methods.add_async_method("exhausted", |_, this, ()| async move {
1650 this.state
1651 .lock()
1652 .await
1653 .exhausted()
1654 .await
1655 .map_err(|e| knl_err("exhausted", &e))
1656 });
1657
1658 // s:close(reason?, detail?) — records `session_closed` and ends the
1659 // session. Idempotent.
1660 //
1661 // The reason says *which kind of ending* this was and stays a short
1662 // vocabulary a reader can fold on; the optional `detail` is the
1663 // sentence only this close can tell — the message of the error a
1664 // caller's own bracket caught, say. Keeping them apart is what stops
1665 // every distinct error message from becoming its own reason, and it
1666 // is the same split the `<close>` path records. `detail` is truncated
1667 // exactly as that path truncates it.
1668 methods.add_async_method(
1669 "close",
1670 |_, this, (reason, detail): (LuaValue, LuaValue)| async move {
1671 let reason: Option<types::CloseReason> = from_lua("close", "reason", reason)?;
1672 let reason = reason.map(|types::CloseReason(text)| text);
1673 let detail: Option<types::CloseDetail> = from_lua("close", "detail", detail)?;
1674 let detail = detail.map(|types::CloseDetail(text)| truncated(&text));
1675 // A close whose `session_closed` append fails (a database
1676 // contended past its retries, a store that is gone) surfaces
1677 // here: the session stays open and the caller knows the
1678 // boundary was not recorded, instead of a silent closed=true
1679 // with no record.
1680 this.state
1681 .lock()
1682 .await
1683 .close_with(reason.as_deref(), detail.as_deref())
1684 .await
1685 .map_err(|e| knl_err("close", &e))?;
1686 Ok(())
1687 },
1688 );
1689
1690 // __close(self, err) — the Lua 5.4 to-be-closed variable:
1691 //
1692 // do
1693 // local s <close> = knl.open({ owner = "u" })
1694 // ...
1695 // end -- the session's boundary is recorded here
1696 //
1697 // The reason says how the scope ended, not what went wrong: a clean
1698 // exit is "scope_exit", an unwinding one "error", with the message
1699 // in `detail`. Folding the message into the reason would make every
1700 // distinct failure its own reason and the vocabulary unreadable.
1701 //
1702 // An explicit `close` earlier in the block already ended the session,
1703 // and this is a no-op then: the caller's reason is the one in the log.
1704 //
1705 // What a failed append does depends on whether there is already an
1706 // error on its way out:
1707 //
1708 // - clean exit (`err` is nil): raise, exactly as `close` does, since
1709 // a close that reports success with no `session_closed` recorded
1710 // is the one outcome the boundary exists to rule out;
1711 // - unwinding (`err` is non-nil): do *not* raise. Lua would replace
1712 // the body's error with this one, and the body's error is what the
1713 // caller is trying to diagnose — a bookkeeping failure must not
1714 // overwrite the failure it is bookkeeping for. It goes to the log
1715 // as a `warn!` and the original error propagates unchanged.
1716 methods.add_async_meta_method(
1717 LuaMetaMethod::Close,
1718 |_, this, error: LuaValue| async move {
1719 // Computed before the session is reached: nothing about the
1720 // error value is read while it is held.
1721 let unwinding = !matches!(error, LuaValue::Nil);
1722 let (reason, detail) = match error {
1723 LuaValue::Nil => (knl::CLOSE_REASON_SCOPE_EXIT, None),
1724 error => (knl::CLOSE_REASON_ERROR, Some(error_detail(&error))),
1725 };
1726 let mut state = this.state.lock().await;
1727 if state.is_closed() {
1728 return Ok(());
1729 }
1730 let outcome = state.close_with(Some(reason), detail.as_deref()).await;
1731 match outcome {
1732 Ok(()) => Ok(()),
1733 Err(e) if unwinding => {
1734 tracing::warn!(
1735 session = %state.id(),
1736 error = %e,
1737 "knl: session_closed was not recorded; \
1738 the block's own error is propagating instead"
1739 );
1740 Ok(())
1741 }
1742 Err(e) => Err(knl_err("close", &e)),
1743 }
1744 },
1745 );
1746 }
1747}
1748
1749/// What `opts.budget` asked for, once the two forms have been told apart.
1750///
1751/// Two things a caller can mean by "this session's budget", and they are not
1752/// interchangeable: `amount` is an owner *granting* — a balance out of
1753/// nothing the kernel can account for, which only an owner may do — while
1754/// `from_parent` is an *allocation*, units moved out of the balance a parent
1755/// session already holds. One is a quota appearing, the other is a quota
1756/// changing hands, so they are separated here and refused together
1757/// ([`budget_source`]).
1758enum BudgetSource {
1759 /// `{ amount, tag?, desc? }` — what an owner allows this session.
1760 Grant(knl::BudgetGrant),
1761 /// `{ from_parent, tag? }` — what the parent named in `opts.parent`
1762 /// hands over out of its own balance.
1763 FromParent(knl::Allocation),
1764}
1765
1766/// Decide which of the two `budget` forms a caller wrote.
1767///
1768/// The table's *shape* was settled by the deserializer ([`types::BudgetOpt`]),
1769/// including the refusal of a misspelt field — a misspelt cap that reads as
1770/// "no cap" is exactly the failure a budget exists to prevent. What is left
1771/// is the part no schema states: the two amounts exclude each other, and
1772/// naming both is refused rather than resolved by precedence, because "the
1773/// owner allows 100" and "the parent hands over 100" are different claims
1774/// about where a balance came from and a table that makes both says neither.
1775/// `desc` belongs to a grant alone — an allocation records the parent it came
1776/// from, which is the whole of what the kernel knows about why it happened.
1777fn budget_source(
1778 method: &str,
1779 budget: Option<types::BudgetOpt>,
1780) -> LuaResult<Option<BudgetSource>> {
1781 let Some(budget) = budget else {
1782 return Ok(None);
1783 };
1784 let types::BudgetOpt {
1785 amount,
1786 tag,
1787 desc,
1788 from_parent,
1789 } = budget;
1790
1791 if let Some(from_parent) = from_parent {
1792 if amount.is_some() {
1793 return Err(err(
1794 method,
1795 "budget names both amount and from_parent: an owner's grant and an allocation \
1796 out of a parent's balance are different claims about where the quota came from",
1797 ));
1798 }
1799 if desc.is_some() {
1800 return Err(err(
1801 method,
1802 "budget.desc belongs to an owner's grant; an allocation records the parent it \
1803 came from instead",
1804 ));
1805 }
1806 if from_parent < 0 {
1807 return Err(err(
1808 method,
1809 format!(
1810 "budget.from_parent must be a non-negative whole number, got {from_parent}"
1811 ),
1812 ));
1813 }
1814 return Ok(Some(BudgetSource::FromParent(knl::Allocation {
1815 amount: from_parent,
1816 tag,
1817 })));
1818 }
1819
1820 let Some(amount) = amount else {
1821 return Err(err(
1822 method,
1823 "budget.amount is required (non-negative whole number), or budget.from_parent to \
1824 allocate out of a parent's balance",
1825 ));
1826 };
1827 if amount < 0 {
1828 return Err(err(
1829 method,
1830 format!("budget.amount must be a non-negative whole number, got {amount}"),
1831 ));
1832 }
1833
1834 Ok(Some(BudgetSource::Grant(knl::BudgetGrant {
1835 amount,
1836 tag,
1837 desc,
1838 })))
1839}
1840
1841/// Read `opts.budget` where only an owner's grant makes sense.
1842///
1843/// `knl.resume` reopens a stream that already exists, so there is no parent
1844/// to allocate from and no child being opened: an allocation there is a
1845/// caller reaching for the wrong call, and it is named as such rather than
1846/// silently read as a grant of the same size.
1847fn grant_only(
1848 method: &str,
1849 budget: Option<types::BudgetOpt>,
1850) -> LuaResult<Option<knl::BudgetGrant>> {
1851 match budget_source(method, budget)? {
1852 None => Ok(None),
1853 Some(BudgetSource::Grant(grant)) => Ok(Some(grant)),
1854 Some(BudgetSource::FromParent(_)) => Err(err(
1855 method,
1856 "budget.from_parent allocates from a parent's balance, which is what \
1857 open{ parent = … } does; this call takes an owner's grant (amount)",
1858 )),
1859 }
1860}
1861
1862/// Read `opts.owner`: the principal the session belongs to.
1863///
1864/// Total: an absent owner is the reserved anonymous id rather than `nil`, so
1865/// the policy layer above the kernel always has a real key to read. The
1866/// reserved ids are the kernel's own namespace and an untrusted Lua caller
1867/// must not claim one, or it could impersonate a reserved principal on
1868/// `session_opened`. Compared against the consts, not literal strings, so the
1869/// guard tracks the kernel's definition.
1870fn owner_of(owner: Option<String>) -> LuaResult<String> {
1871 let Some(owner) = owner else {
1872 return Ok(knl::ANON.to_string());
1873 };
1874 if owner == knl::ANON || owner == knl::SYSTEM {
1875 return Err(err("open", format!("owner {owner:?} is reserved")));
1876 }
1877 Ok(owner)
1878}
1879
1880/// The `params` of `s:query`, in the kernel's terms.
1881///
1882/// A list is the values for the `?` parameters, in order; a table with names
1883/// is the values for `:name` / `@name` / `$name`. An absent or empty table is
1884/// neither: the statement is expected to have no parameters of its own.
1885fn query_params(params: Option<types::QueryParams>) -> knl::QueryParams {
1886 match params {
1887 None => knl::QueryParams::None,
1888 Some(types::QueryParams::Positional(values)) if values.is_empty() => knl::QueryParams::None,
1889 Some(types::QueryParams::Positional(values)) => knl::QueryParams::Positional(values),
1890 Some(types::QueryParams::Named(named)) if named.is_empty() => knl::QueryParams::None,
1891 Some(types::QueryParams::Named(named)) => knl::QueryParams::Named(named),
1892 }
1893}
1894
1895/// The `opts` of `s:query`, in the kernel's terms.
1896///
1897/// An option the caller left out is the kernel's own default rather than a
1898/// value this layer picks: the deadline and the row cap are the store's
1899/// policy, and a second copy of either here would be a second thing to change.
1900/// An empty `sessions` list is passed through as the empty set, which the
1901/// kernel refuses in its own words rather than being read as "all of them".
1902fn query_opts(opts: Option<types::QueryOpts>) -> knl::QueryOpts {
1903 let Some(opts) = opts else {
1904 return knl::QueryOpts::default();
1905 };
1906 knl::QueryOpts {
1907 sessions: opts.sessions,
1908 timeout_ms: opts.timeout_ms.unwrap_or(knl::DEFAULT_TIMEOUT_MS),
1909 limit: opts.limit.map_or(knl::DEFAULT_LIMIT, |n| n as usize),
1910 }
1911}
1912
1913/// The `opts` of `s:view`, as the object the kernel's projections read.
1914///
1915/// Built field by field rather than serialized, so the map holds exactly what
1916/// the caller named: an absent `n` is absent, and `tail` falls back to its own
1917/// default instead of being handed a null to interpret.
1918fn view_opts(opts: Option<types::ViewOpts>) -> Option<Map<String, Value>> {
1919 let opts = opts?;
1920 let mut out = Map::new();
1921 if let Some(n) = opts.n {
1922 out.insert("n".to_string(), Value::from(n));
1923 }
1924 Some(out)
1925}
1926
1927/// The storage backend a session's log goes in.
1928enum StoreTarget {
1929 /// The in-memory store, asked for by name: `"mem"`, for tests and mocks.
1930 Mem,
1931 /// A durable SQLite stream at the given path.
1932 Sqlite(String),
1933}
1934
1935/// Read `opts.store`: `"mem"` → in-memory, `{ sqlite = "<path>" }` → durable.
1936///
1937/// The two forms are the deserializer's ([`types::StoreSpec`]); what is left
1938/// here is the one word the union cannot state — that the only *named* store
1939/// is the in-memory one. `method` names the caller (`open` / `resume`) for
1940/// attribution.
1941fn store_target(method: &str, spec: types::StoreSpec) -> LuaResult<StoreTarget> {
1942 match spec {
1943 types::StoreSpec::Named(name) if name == types::MEM_STORE => Ok(StoreTarget::Mem),
1944 types::StoreSpec::Named(name) => Err(err(
1945 method,
1946 format!(r#"unknown store {name:?} (expected "mem" or {{ sqlite = <path> }})"#),
1947 )),
1948 types::StoreSpec::File(file) => Ok(StoreTarget::Sqlite(file.sqlite)),
1949 }
1950}
1951
1952/// Read `opts.parent`: the session this one is opened from, if any.
1953///
1954/// Taken off the table by hand, and before the rest of it is read as data: a
1955/// parent is a live handle rather than a value, and serde has no
1956/// representation for a userdata (the deserializer is told to drop it, which
1957/// is what lets an otherwise closed `opts` carry one). Only its presence and
1958/// its type are settled here — whether it really is a kernel session, and
1959/// whether its balance covers what is being asked for, is answered where the
1960/// allocation runs, with the parent borrowed.
1961/// `opts` with its `parent` taken out, so the rest can be read strictly.
1962///
1963/// The companion of [`parse_parent`], and the reason [`from_lua`] can leave
1964/// the deserializer's defaults alone. A parent is a live session userdata,
1965/// which serde cannot carry; the way to let one through would be to tell the
1966/// deserializer to skip every value it has no representation for, and that
1967/// would skip a function a caller wrote where a string belonged as well. So
1968/// the one key that cannot cross is lifted out here — `parse_parent` already
1969/// holds the real handle — and what is left is read with nothing turned off.
1970///
1971/// A shallow copy, because `parent` is a top-level key: everything nested is
1972/// shared with the caller's table and read from it exactly as before.
1973fn without_parent(lua: &Lua, opts: &LuaValue) -> LuaResult<LuaValue> {
1974 let LuaValue::Table(table) = opts else {
1975 return Ok(opts.clone());
1976 };
1977 let rest = lua.create_table()?;
1978 for pair in table.clone().pairs::<LuaValue, LuaValue>() {
1979 let (key, value) = pair?;
1980 if let LuaValue::String(name) = &key {
1981 if name.to_str()? == "parent" {
1982 continue;
1983 }
1984 }
1985 rest.set(key, value)?;
1986 }
1987 Ok(LuaValue::Table(rest))
1988}
1989
1990fn parse_parent(opts: &LuaValue) -> LuaResult<Option<LuaAnyUserData>> {
1991 let LuaValue::Table(opts) = opts else {
1992 return Ok(None);
1993 };
1994 match opts.get::<LuaValue>("parent")? {
1995 LuaValue::Nil => Ok(None),
1996 LuaValue::UserData(parent) => Ok(Some(parent)),
1997 other => Err(err(
1998 "open",
1999 format!(
2000 "parent must be a session (the userdata knl.open returns), got {}",
2001 other.type_name()
2002 ),
2003 )),
2004 }
2005}
2006
2007/// Open a NEW durable session on the SQLite stream at `path`.
2008///
2009/// The stream id is minted here and adopted as the session's own id, so the
2010/// id `knl.open` reports (`s:id()`) is exactly the stream a later
2011/// `knl.resume` reopens — the durable identity is one string, not two.
2012async fn open_sqlite(
2013 owner: String,
2014 grant: Option<knl::BudgetGrant>,
2015 path: &std::path::Path,
2016 drivers: &knl::IsleDrivers,
2017) -> LuaResult<Session> {
2018 let stream = uuid::Uuid::new_v4().to_string();
2019 let store = knl::SqliteEventStore::open(path, stream.clone(), drivers)
2020 .await
2021 .map_err(|e| knl_err("open", &e))?;
2022 let mut state = knl::Session::open_on(owner, grant, Box::new(store))
2023 .await
2024 .map_err(|e| knl_err("open", &e))?;
2025 state.adopt_id(stream);
2026 Ok(Session::from_state(state))
2027}
2028
2029/// Reopen the stream `session_id` and resume it.
2030///
2031/// `store` is the backend the stream lives in — a file, or the in-memory
2032/// database of that name while some handle still holds it open.
2033/// `Session::resume` re-folds the log; the reopened stream's id is adopted so
2034/// `s:id()` matches the stream the caller named.
2035async fn resume_on(
2036 grant: Option<knl::BudgetGrant>,
2037 store: knl::SqliteEventStore,
2038 session_id: String,
2039) -> LuaResult<Session> {
2040 // Resumed with no grant, so nothing has been written yet when the check
2041 // below runs: a refused resume must leave the stream exactly as it found
2042 // it, and a `budget_granted` recorded before the refusal would be the
2043 // caller writing into a stream it was not allowed to touch.
2044 let mut state = knl::Session::resume(None, Box::new(store))
2045 .await
2046 .map_err(|e| knl_err("resume", &e))?;
2047 // The open path refuses an untrusted caller claiming a reserved
2048 // principal (parse_owner); resume must hold the same line, or Lua could
2049 // reopen a SYSTEM-owned stream and write into the reserved namespace.
2050 // ANON streams stay resumable — they are what unspecified-owner Lua
2051 // sessions (and pre-owner logs) record as.
2052 if state.owner() == knl::SYSTEM {
2053 return Err(err(
2054 "resume",
2055 format!("stream owner {:?} is reserved", knl::SYSTEM),
2056 ));
2057 }
2058 // The stream passed: now the owner's fresh grant is recorded, adding to
2059 // what the ledger already carried.
2060 if let Some(grant) = grant {
2061 state
2062 .grant_more(grant)
2063 .await
2064 .map_err(|e| knl_err("resume", &e))?;
2065 }
2066 state.adopt_id(session_id);
2067 Ok(Session::from_state(state))
2068}
2069
2070/// Open the store a child's stream will live in.
2071///
2072/// `named` is what the caller asked for and `parent_db` is where the parent
2073/// is; an absent `store` means the child goes where its parent already is,
2074/// which is the only answer that always works. A store the caller *did*
2075/// name is opened as asked and handed to the kernel, which refuses it if it
2076/// turns out to be a different database — the check belongs there, next to
2077/// the transaction that would have to span both.
2078async fn open_child_store(
2079 named: Option<StoreTarget>,
2080 parent_db: &str,
2081 stream: &str,
2082 drivers: &knl::IsleDrivers,
2083) -> LuaResult<Box<dyn knl::EventStore>> {
2084 let store = match named {
2085 // The parent's database, addressed exactly as it was opened (a path,
2086 // or the in-memory database's shared-cache URI).
2087 None => knl::SqliteEventStore::open(std::path::Path::new(parent_db), stream, drivers).await,
2088 Some(StoreTarget::Sqlite(path)) => {
2089 knl::SqliteEventStore::open(std::path::Path::new(&path), stream, drivers).await
2090 }
2091 Some(StoreTarget::Mem) => knl::SqliteEventStore::open_memory(stream, drivers).await,
2092 };
2093 Ok(Box::new(store.map_err(|e| knl_err("open", &e))?))
2094}
2095
2096/// Open a session from `parent`, paying for it out of the parent's balance —
2097/// the `knl.open{ parent = … }` path.
2098///
2099/// The parent is held for the whole of it: the allocation is one transaction
2100/// on the parent's store, so the parent's own calls wait for it exactly as
2101/// they wait for any other syscall of its own.
2102///
2103/// A parent on the in-memory database is refused before any of that
2104/// ([`knl::is_memory_database`]). A tree writes to one database and that one
2105/// is addressed by a shared-cache URI, whose locks are per table: the child's
2106/// first write while the parent holds the table meets `SQLITE_LOCKED`, which
2107/// no busy timeout waits out. So the refusal is the honest answer, and the
2108/// alternative — teaching the kernel to wait on that lock — is machinery for a
2109/// store that exists for tests and mocks.
2110async fn open_child_session(
2111 lua: Lua,
2112 parent: LuaAnyUserData,
2113 owner: String,
2114 allocation: knl::Allocation,
2115 named_store: Option<StoreTarget>,
2116 drivers: knl::IsleDrivers,
2117) -> LuaResult<LuaAnyUserData> {
2118 let handle = parent.borrow::<Session>().map_err(|_| {
2119 err(
2120 "open",
2121 "parent must be a session returned by knl.open / knl.resume",
2122 )
2123 })?;
2124 let child = {
2125 let mut state = handle.state.lock().await;
2126 let parent_db = state
2127 .database()
2128 .ok_or_else(|| {
2129 err(
2130 "open",
2131 "the parent's store keeps a single stream, so there is no database to open a \
2132 child on",
2133 )
2134 })?
2135 .to_string();
2136 if knl::is_memory_database(&parent_db) {
2137 return Err(err(
2138 "open",
2139 "a session tree needs a file store: the parent is on the in-memory database \
2140 (store = \"mem\"), whose shared cache locks per table, and the kernel does not \
2141 wait on that lock. Open the parent without a store (the host's database) or on \
2142 { sqlite = <path> }",
2143 ));
2144 }
2145 // Minted here and adopted by the child below, so the id `s:id()`
2146 // reports is the stream the parent's log names as its child.
2147 let stream = uuid::Uuid::new_v4().to_string();
2148 let store = open_child_store(named_store, &parent_db, &stream, &drivers).await?;
2149 state
2150 .open_child(stream, owner, allocation, store)
2151 .await
2152 .map_err(|e| knl_err("open", &e))?
2153 };
2154 // The parent is released before Lua is re-entered to build the userdata.
2155 drop(handle);
2156 lua.create_userdata(Session::from_state(child))
2157}
2158
2159/// Build a session userdata from `opts` — the body of `knl.open`.
2160///
2161/// `opts.owner` is the principal (default the reserved anonymous id),
2162/// `opts.budget` the grant (`{ amount, tag?, desc? }`), and `opts.store`
2163/// the backend — absent is `default_store`, the file the host owns; `"mem"`
2164/// asks for the in-memory database by name; `{ sqlite = "<path>" }` is a file
2165/// the caller picked.
2166///
2167/// `opts.parent` is the other way to open: a session this one is opened
2168/// *from*, with `budget = { from_parent = n, tag? }` moving `n` out of that
2169/// session's balance in the same write that opens this one. The two forms
2170/// are exclusive in both directions — a parent with an owner's grant would be
2171/// a child whose quota nobody paid for, and `from_parent` with no parent has
2172/// nowhere to take it from — so each is refused with the other named.
2173async fn open_session(
2174 lua: Lua,
2175 opts: LuaValue,
2176 drivers: knl::IsleDrivers,
2177 default_store: std::path::PathBuf,
2178) -> LuaResult<LuaAnyUserData> {
2179 // The parent comes off the table first and by hand: it is a live session
2180 // handle, which serde cannot carry (see `parse_parent`).
2181 let parent = parse_parent(&opts)?;
2182 // Everything else is read as data, in one step, by the same types the
2183 // surface is declared in. The Lua value is consumed here and nothing of
2184 // it survives into the awaits below, which is the rule a `LuaTable` held
2185 // across a suspension point would break.
2186 let opts: types::OpenOpts = match opts {
2187 LuaValue::Nil => types::OpenOpts::default(),
2188 value => from_lua("open", "opts", without_parent(&lua, &value)?)?,
2189 };
2190 let owner = owner_of(opts.owner)?;
2191 let budget = budget_source("open", opts.budget)?;
2192 // An absent `store` is *not* the default one here: a child with no store
2193 // goes where its parent already is, and a child that asked for "mem"
2194 // asked for a different database and is refused. Telling the two apart
2195 // is why the question is asked as an Option.
2196 let named_store = opts
2197 .store
2198 .map(|spec| store_target("open", spec))
2199 .transpose()?;
2200
2201 let Some(parent) = parent else {
2202 let grant = match budget {
2203 None => None,
2204 Some(BudgetSource::Grant(grant)) => Some(grant),
2205 Some(BudgetSource::FromParent(_)) => {
2206 return Err(err(
2207 "open",
2208 "budget.from_parent allocates out of a parent's balance, so it needs \
2209 opts.parent: the session to open this one from",
2210 ));
2211 }
2212 };
2213 // No store named: the host's database, which is where a real session
2214 // belongs. `"mem"` is the other answer and it has to be asked for.
2215 let session = match named_store {
2216 None => open_sqlite(owner, grant, &default_store, &drivers).await?,
2217 Some(StoreTarget::Mem) => Session::new(owner, grant, &drivers).await?,
2218 Some(StoreTarget::Sqlite(path)) => {
2219 open_sqlite(owner, grant, std::path::Path::new(&path), &drivers).await?
2220 }
2221 };
2222 return lua.create_userdata(session);
2223 };
2224
2225 let allocation = match budget {
2226 Some(BudgetSource::FromParent(allocation)) => allocation,
2227 _ => {
2228 return Err(err(
2229 "open",
2230 "a child's quota comes out of its parent's: opts.parent needs \
2231 budget = { from_parent = n, tag? }",
2232 ));
2233 }
2234 };
2235 open_child_session(lua, parent, owner, allocation, named_store, drivers).await
2236}
2237
2238/// Resume a persisted session — the body of `knl.resume`.
2239///
2240/// Requires `opts.session = "<stream id>"`. `opts.store` says where the
2241/// stream lives and means what it means on open when it is left out: the file
2242/// the host owns, which is where a session opened without a store went. So
2243/// the round trip needs one argument — `knl.resume{ session = id }` reopens
2244/// what `knl.open{}` wrote. `"mem"` reopens an in-memory stream, for exactly
2245/// as long as some handle is still holding it, and `{ sqlite = "<path>" }` a
2246/// file the caller named.
2247/// `opts.budget` is optional and means the owner grants
2248/// *again*: it is recorded and added to the balance the log already
2249/// carries, rather than replacing it. The returned userdata is the same
2250/// one `knl.open` returns, only pre-loaded with the balance folded from the
2251/// ledger.
2252async fn resume_session(
2253 lua: Lua,
2254 opts: LuaValue,
2255 drivers: knl::IsleDrivers,
2256 default_store: std::path::PathBuf,
2257) -> LuaResult<LuaAnyUserData> {
2258 if matches!(opts, LuaValue::Nil) {
2259 return Err(err("resume", "opts must be a table with store and session"));
2260 }
2261 // Read as data, in one step, and consumed here: nothing of the Lua table
2262 // survives into the awaits below.
2263 let opts: types::ResumeOpts = from_lua("resume", "opts", opts)?;
2264 let grant = grant_only("resume", opts.budget)?;
2265 let store = opts
2266 .store
2267 .map(|spec| store_target("resume", spec))
2268 .transpose()?;
2269 let session_id = opts.session;
2270 let store = match store {
2271 // The host's database, as on open: a session that named no store
2272 // went there, so a resume that names none looks there.
2273 None => knl::SqliteEventStore::open(&default_store, session_id.clone(), &drivers).await,
2274 Some(StoreTarget::Sqlite(path)) => {
2275 knl::SqliteEventStore::open(std::path::Path::new(&path), session_id.clone(), &drivers)
2276 .await
2277 }
2278 // An in-memory stream is reopenable too, for as long as it exists:
2279 // the database is named after the stream, so a second handle on a
2280 // live one finds the same log. It cannot outlive the process, and it
2281 // does not pretend to — a name nobody is holding open resumes as an
2282 // empty stream, which is refused for having no session in it.
2283 Some(StoreTarget::Mem) => {
2284 knl::SqliteEventStore::open_memory(session_id.clone(), &drivers).await
2285 }
2286 }
2287 .map_err(|e| knl_err("resume", &e))?;
2288 let state = resume_on(grant, store, session_id).await?;
2289 lua.create_userdata(state)
2290}
2291
2292/// Take an attributed message apart into `{ kind, method, retryable,
2293/// message }` — the body of `knl.error`.
2294///
2295/// # Why a function and not the raised value
2296///
2297/// The error a caller wants is a table. It cannot be one: mlua raises every
2298/// failure a Rust callback returns as its own `WrappedFailure` userdata
2299/// ([`LuaError`] has no variant that carries a Lua value), so a bridge method
2300/// has no way to make a table *be* the raised object. What it can do is
2301/// raise a message with a shape — [`attributed`] fixes the first three fields
2302/// as a closed vocabulary — and hand the shell a reader for it. So
2303/// `knl.error(err)` is that reader: `pcall`, pass what was caught, and get
2304/// the table.
2305///
2306/// The argument is anything the raise handed over: the userdata, or a string
2307/// somebody already rendered. Either way it is read as text.
2308///
2309/// An unrecognised message is not an error. A raise that did not come from
2310/// this bridge (a Lua-side `error("...")`, a message from another module) is
2311/// reported as it is — `method = nil`, `kind = nil`, `retryable = false`, and
2312/// the whole text as `message` — because a reader that raised on unfamiliar
2313/// input would turn every unrelated failure into a second one, inside the
2314/// handler that was trying to describe the first.
2315///
2316/// The returned table carries a `__tostring` that gives the original message
2317/// back, so `tostring(knl.error(e))` is `tostring(e)` and a caller that only
2318/// wants to print or `find` in it does not have to know which it is holding.
2319fn error_table(lua: &Lua, raised: LuaValue) -> LuaResult<LuaTable> {
2320 let text = match &raised {
2321 LuaValue::String(text) => text.to_str()?.to_string(),
2322 // Anything else is rendered by Lua's own rules: the raised value is
2323 // a userdata whose `__tostring` is the message.
2324 other => other.to_string()?,
2325 };
2326
2327 let mut read = types::ErrorTable {
2328 kind: None,
2329 method: None,
2330 retryable: false,
2331 message: text.clone(),
2332 };
2333
2334 // `knl: <method>: <kind>: <message>` — read off the line that carries
2335 // it, since a raise that crossed a callback boundary arrives with a
2336 // traceback on the lines after it. Split on the first two separators
2337 // only: the message is whatever is left, colons and all.
2338 let attributed = text
2339 .lines()
2340 .find_map(|line| line.split_once("knl: ").map(|(_, rest)| rest));
2341 if let Some((method, rest)) = attributed.and_then(|rest| rest.split_once(": ")) {
2342 if let Some((kind, message)) = rest.split_once(": ") {
2343 // Only a kind the kernel actually publishes is taken as one, so
2344 // a message that merely looks like the shape is left as prose.
2345 if knl::KnlError::KINDS.contains(&kind) {
2346 read.method = Some(method.to_string());
2347 read.kind = Some(kind.to_string());
2348 read.retryable = knl::KnlError::kind_is_retryable(kind);
2349 read.message = message.to_string();
2350 }
2351 }
2352 }
2353
2354 // Built from the declared type rather than field by field, so the table a
2355 // caller reads is the one `knl_types.ErrorTable` describes.
2356 let out = as_table(lua, "error", &read)?;
2357
2358 // The table renders as the message it was read from, so it can stand in
2359 // for the raised value wherever one was being printed or searched.
2360 let meta = lua.create_table()?;
2361 meta.set(
2362 "__tostring",
2363 lua.create_function(move |_, _: LuaValue| Ok(text.clone()))?,
2364 )?;
2365 out.set_metatable(Some(meta))?;
2366 Ok(out)
2367}
2368
2369/// Mint a beat id — the body of `knl.new_beat_id`.
2370///
2371/// A UUID v7: random, but with its timestamp in the leading bits, so beat
2372/// ids of one session sort in the order they were minted. That ordering is
2373/// the only property the id carries; the kernel treats it as opaque.
2374///
2375/// Session-free on purpose, like a sequence generator: a beat is declared by
2376/// the layer that drives the loop, and asking the kernel for one would put
2377/// the numbering back where this round took it out of.
2378fn new_beat_id(_: &Lua, _: ()) -> LuaResult<String> {
2379 Ok(uuid::Uuid::now_v7().to_string())
2380}
2381
2382/// Build a Lua table from a declared type.
2383///
2384/// The other half of [`from_lua`]: what a syscall answers is built by
2385/// serializing the type the registry names, so a return cannot grow a field
2386/// the declaration does not have. A failure here is the bridge's own bug
2387/// rather than the caller's, and it is attributed as `validation` all the same
2388/// — the vocabulary is closed and there is no class for "the kernel could not
2389/// describe itself".
2390fn as_table<T: serde::Serialize>(lua: &Lua, method: &str, value: &T) -> LuaResult<LuaTable> {
2391 match lua.to_value(value).map_err(|e| err(method, e))? {
2392 LuaValue::Table(table) => Ok(table),
2393 other => Err(err(
2394 method,
2395 format!(
2396 "the answer did not serialize as a table, got {}",
2397 other.type_name()
2398 ),
2399 )),
2400 }
2401}
2402
2403/// The declared surface as a Lua table — the body of `knl.api()`.
2404///
2405/// [`types::ApiReport`], built from [`SESSION_API`], [`MODULE_API`],
2406/// [`knl::KnlError::KINDS`], the events table itself and
2407/// [`lshape_module_source`], so a caller reads what the kernel offers from the
2408/// same places the reflection test holds the registration to.
2409///
2410/// `errors` is the closed list of classes `knl.error(e).kind` can report. It
2411/// is published for the same reason the two method lists are: the shell keeps
2412/// its own declaration of the vocabulary, and a declaration nobody can check
2413/// is one that drifts. `fields` is the other half of the read contract: the
2414/// `data` paths a Lua view spells in `json_extract`, taken from the kernel's
2415/// own [`knl::FIELD_AMOUNT`] and friends. `types` is the generated
2416/// `knl_types` module as source text — the same one the host embeds — so a
2417/// tool can read the argument and return shapes without loading them.
2418fn api(lua: &Lua, _: ()) -> LuaResult<LuaTable> {
2419 // Built once per process, like the types module it carries: every field of
2420 // it is a pure function of compile-time constants — two `&'static` lists,
2421 // the error vocabulary, the kernel's field names — and `schema` is a
2422 // `PRAGMA table_info` against a private in-memory database created from a
2423 // `const` DDL, so the answer cannot differ between two calls. What was
2424 // paid on each call was a SQLite open + CREATE TABLE + pragma and a full
2425 // re-render of the types module.
2426 //
2427 // Only the *value* is cached. A Lua table belongs to one VM, so the
2428 // conversion below still runs per call — and it must, since a caller may
2429 // mutate what it is handed.
2430 //
2431 // A failure is not cached: `events_schema` is fallible, and a store that
2432 // could not be opened once is not a permanent answer about the surface.
2433 let report = match API_REPORT.get() {
2434 Some(report) => report,
2435 None => {
2436 let built = build_api_report()?;
2437 API_REPORT.get_or_init(|| built)
2438 }
2439 };
2440 as_table(lua, "api", report)
2441}
2442
2443/// The declared surface, built once and held by [`api`].
2444static API_REPORT: std::sync::OnceLock<types::ApiReport> = std::sync::OnceLock::new();
2445
2446/// How many times [`build_api_report`] actually ran, so a test can hold the
2447/// cache to its promise rather than to a stopwatch.
2448#[cfg(test)]
2449static API_BUILDS: std::sync::atomic::AtomicUsize = std::sync::atomic::AtomicUsize::new(0);
2450
2451/// How many times [`build_lshape_module_source`] actually ran, for the same
2452/// reason.
2453#[cfg(test)]
2454static TYPES_BUILDS: std::sync::atomic::AtomicUsize = std::sync::atomic::AtomicUsize::new(0);
2455
2456/// Build [`types::ApiReport`] from the places the reflection test holds the
2457/// registration to.
2458fn build_api_report() -> LuaResult<types::ApiReport> {
2459 #[cfg(test)]
2460 API_BUILDS.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
2461
2462 /// One `{ name, doc }` list, in declaration order.
2463 fn listed(entries: &[(&str, &str)]) -> Vec<types::ApiEntry> {
2464 entries
2465 .iter()
2466 .map(|(name, doc)| types::ApiEntry {
2467 name: (*name).to_string(),
2468 doc: (*doc).to_string(),
2469 })
2470 .collect()
2471 }
2472
2473 // The read contract: the table a query names and the columns it has.
2474 // Read off SQLite itself (`PRAGMA table_info`) rather than written out
2475 // here, so the published schema is the schema — a caller's SQL is written
2476 // against columns that exist, and the shell's own declaration of them can
2477 // be checked instead of trusted.
2478 let schema = types::ApiSchema {
2479 table: knl::EVENTS_TABLE.to_string(),
2480 columns: knl::events_schema()
2481 .map_err(|e| knl_err("api", &e))?
2482 .into_iter()
2483 .map(|column| types::ApiColumn {
2484 name: column.name,
2485 declared_type: column.declared_type,
2486 pk: column.pk,
2487 })
2488 .collect(),
2489 };
2490
2491 // The other half of the read contract: the `data` paths a view reaches
2492 // into, taken from the constants the writers use rather than retyped, so
2493 // a Lua `json_extract('$.amount')` can be held against the field the
2494 // kernel actually wrote.
2495 let fields = types::ApiFields {
2496 amount: knl::FIELD_AMOUNT.to_string(),
2497 tag: knl::FIELD_TAG.to_string(),
2498 desc: knl::FIELD_DESC.to_string(),
2499 remaining: knl::FIELD_REMAINING.to_string(),
2500 scope_id: knl::FIELD_SCOPE_ID.to_string(),
2501 owner: knl::FIELD_OWNER.to_string(),
2502 parent: knl::FIELD_PARENT.to_string(),
2503 child: knl::FIELD_CHILD.to_string(),
2504 reason: knl::FIELD_REASON.to_string(),
2505 detail: knl::FIELD_DETAIL.to_string(),
2506 open_children: knl::FIELD_OPEN_CHILDREN.to_string(),
2507 };
2508
2509 Ok(types::ApiReport {
2510 session: listed(SESSION_API),
2511 module: listed(MODULE_API),
2512 errors: knl::KnlError::KINDS.iter().map(|k| k.to_string()).collect(),
2513 schema,
2514 fields,
2515 types: lshape_module_source(),
2516 })
2517}
2518
2519/// Register the `knl` global. Two things come from the host and nothing else
2520/// does: the connection threads, and the file a session with no `store` of its
2521/// own lands in. All session state stays inside the userdata.
2522///
2523/// The functions are exactly [`MODULE_API`]: `knl.open(opts?)` is the
2524/// constructor (owner- and store-aware), `knl.resume(opts)` reopens a
2525/// persisted SQLite session, `knl.new_beat_id()` mints a beat id for the
2526/// caller to stamp on events, `knl.error(e)` reads a raised failure back as
2527/// a table, and `knl.api()` reports the declared surface.
2528/// Each is bound by hand — a `create_function` needs its own signature — and
2529/// a test below checks the set of bound names against the table.
2530///
2531/// `drivers` is where the connection threads a session opens are kept. They
2532/// cannot belong to the session: the drop backstop hands its closing event to
2533/// the thread *after* the handle is gone, so a thread the store had already
2534/// stopped could not take it. The host owns them for the length of a run and
2535/// drains them once at the end of it ([`knl::IsleDrivers`]).
2536///
2537/// `default_store` is that file — `{base_dir}/projects/<slug>/knl.sqlite`
2538/// unless `AGENT_BLOCK_KNL_PATH` says otherwise
2539/// ([`crate::bridge::config::knl_path`]). It is a path rather than an open
2540/// store because a session opens its own stream: one file, one stream per
2541/// `knl.open{}`, which is what lets a tree opened from a default parent live
2542/// in one database. Where it goes is the host's answer, so it is decided
2543/// there and carried here.
2544///
2545/// `open` and `resume` are async because opening a stream waits — for the
2546/// connection thread to start, for the schema, for the opening events to land
2547/// — and this is called on the Lua VM's own thread, which must not wait for
2548/// anything. Both are reachable only from inside a coroutine, which the main
2549/// chunk and every bus handler already are.
2550pub fn register(
2551 lua: &Lua,
2552 drivers: knl::IsleDrivers,
2553 default_store: std::path::PathBuf,
2554) -> LuaResult<()> {
2555 let knl_tbl = lua.create_table()?;
2556
2557 // knl.open(opts?) -> Session userdata
2558 {
2559 let drivers = drivers.clone();
2560 let default_store = default_store.clone();
2561 knl_tbl.set(
2562 "open",
2563 lua.create_async_function(move |lua, opts: LuaValue| {
2564 let drivers = drivers.clone();
2565 let default_store = default_store.clone();
2566 open_session(lua, opts, drivers, default_store)
2567 })?,
2568 )?;
2569 }
2570
2571 // knl.resume(opts) -> Session userdata (a recorded stream re-folded)
2572 knl_tbl.set(
2573 "resume",
2574 lua.create_async_function(move |lua, opts: LuaValue| {
2575 let drivers = drivers.clone();
2576 let default_store = default_store.clone();
2577 resume_session(lua, opts, drivers, default_store)
2578 })?,
2579 )?;
2580
2581 // knl.new_beat_id() -> string (time-ordered, session-free)
2582 knl_tbl.set("new_beat_id", lua.create_function(new_beat_id)?)?;
2583
2584 // knl.error(err) -> { kind, method, retryable, message }
2585 knl_tbl.set("error", lua.create_function(error_table)?)?;
2586
2587 // knl.api() -> the declared surface, session methods and module functions
2588 knl_tbl.set("api", lua.create_function(api)?)?;
2589
2590 lua.globals().set("knl", knl_tbl)?;
2591 Ok(())
2592}
2593
2594// ---------------------------------------------------------------------------
2595// Tests
2596// ---------------------------------------------------------------------------
2597
2598/// The generated declaration, held against the types it was generated from.
2599///
2600/// The claim this round makes is that there is one declaration of the syscall
2601/// surface and it is the Rust types. These are the tests that make it a claim
2602/// rather than an intention: the module has to *load* under the lshape the
2603/// host ships, and a value built from each Rust type has to *pass* the shape
2604/// generated for it. A field renamed on one side and not the other cannot
2605/// survive both.
2606#[cfg(test)]
2607mod generated_types {
2608 use super::types::*;
2609 use super::*;
2610 use serde_json::json;
2611
2612 /// The vendored lshape, in dependency order: `luacats` needs `reflect`,
2613 /// and the aggregate needs all four. The same sources the host embeds.
2614 const LSHAPE_PARTS: [(&str, &str); 4] = [
2615 ("lshape.t", include_str!("../../blocks/lib/lshape/t.lua")),
2616 (
2617 "lshape.reflect",
2618 include_str!("../../blocks/lib/lshape/reflect.lua"),
2619 ),
2620 (
2621 "lshape.check",
2622 include_str!("../../blocks/lib/lshape/check.lua"),
2623 ),
2624 (
2625 "lshape.luacats",
2626 include_str!("../../blocks/lib/lshape/luacats.lua"),
2627 ),
2628 ];
2629
2630 const LSHAPE_ROOT: &str = include_str!("../../blocks/lib/lshape/init.lua");
2631
2632 /// A VM with the vendored lshape on it and the generated module loaded,
2633 /// arranged the way the host arranges them.
2634 fn types_vm() -> (Lua, LuaTable, LuaFunction) {
2635 let lua = Lua::new();
2636 let package: LuaTable = lua.globals().get("package").expect("package");
2637 let loaded: LuaTable = package.get("loaded").expect("package.loaded");
2638 for (name, source) in LSHAPE_PARTS {
2639 let module: LuaValue = lua
2640 .load(source)
2641 .set_name(name)
2642 .eval()
2643 .unwrap_or_else(|e| panic!("{name}: {e}"));
2644 loaded.set(name, module).expect("preload");
2645 }
2646 let root: LuaValue = lua
2647 .load(LSHAPE_ROOT)
2648 .set_name("lshape")
2649 .eval()
2650 .expect("lshape");
2651 loaded.set("lshape", root.clone()).expect("preload lshape");
2652
2653 let module: LuaTable = lua
2654 .load(lshape_module_source())
2655 .set_name("knl_types")
2656 .eval()
2657 .expect("the generated module must load under the vendored lshape");
2658 let check: LuaFunction = lua
2659 .load(r#"return require("lshape").check.check"#)
2660 .eval()
2661 .expect("lshape.check.check");
2662 (lua, module, check)
2663 }
2664
2665 /// The declared types, each with a value built from the Rust type it was
2666 /// generated from.
2667 ///
2668 /// One entry per name in [`types::declared`] — the test below holds the
2669 /// two lists against each other, so a type added there without a sample
2670 /// here is a failure rather than a gap nobody notices.
2671 fn samples(lua: &Lua) -> Vec<(&'static str, LuaValue)> {
2672 fn to(lua: &Lua, value: impl serde::Serialize) -> LuaValue {
2673 lua.to_value(&value).expect("a declared type serializes")
2674 }
2675 vec![
2676 ("SessionId", to(lua, SessionId("s-1".into()))),
2677 ("ScopeId", to(lua, ScopeId("scope-1".into()))),
2678 ("Owner", to(lua, Owner("user-42".into()))),
2679 ("BeatId", to(lua, BeatId("beat-1".into()))),
2680 ("Seq", to(lua, Seq(7))),
2681 ("Count", to(lua, Count(3))),
2682 ("Amount", to(lua, Amount(10))),
2683 ("Remaining", to(lua, Remaining(Some(90)))),
2684 ("Exhausted", to(lua, Exhausted(false))),
2685 ("Sql", to(lua, Sql("SELECT 1".into()))),
2686 ("CloseReason", to(lua, CloseReason("done".into()))),
2687 (
2688 "CloseDetail",
2689 to(lua, CloseDetail("the block raised".into())),
2690 ),
2691 // `Raised` is whatever a raise handed over, so the sample is a
2692 // value no other type would take.
2693 ("Raised", to(lua, json!({ "anything": [1, "at", true] }))),
2694 (
2695 "ViewName",
2696 to(lua, ViewName(crate::knl::projection::VIEW_TAIL.into())),
2697 ),
2698 ("ViewOpts", to(lua, ViewOpts { n: Some(5) })),
2699 (
2700 "OpenOpts",
2701 to(
2702 lua,
2703 OpenOpts {
2704 owner: Some("user-42".into()),
2705 budget: Some(BudgetOpt {
2706 amount: Some(1000),
2707 tag: Some("tokens".into()),
2708 desc: Some("one nightly run".into()),
2709 from_parent: None,
2710 }),
2711 store: Some(StoreSpec::File(SqliteStore {
2712 sqlite: "/tmp/knl.db".into(),
2713 })),
2714 parent: None,
2715 },
2716 ),
2717 ),
2718 (
2719 "ResumeOpts",
2720 to(
2721 lua,
2722 ResumeOpts {
2723 store: Some(StoreSpec::Named(MEM_STORE.into())),
2724 session: "s-1".into(),
2725 budget: Some(BudgetOpt {
2726 from_parent: Some(25),
2727 tag: Some("tokens".into()),
2728 amount: None,
2729 desc: None,
2730 }),
2731 },
2732 ),
2733 ),
2734 (
2735 "AppendEvent",
2736 to(
2737 lua,
2738 AppendEvent {
2739 kind: "msg_user".into(),
2740 beat: Some("beat-1".into()),
2741 meta: Some(Meta::from([
2742 ("label".to_string(), MetaValue::Text("seed".into())),
2743 ("n".to_string(), MetaValue::Number(1.0)),
2744 ("on".to_string(), MetaValue::Flag(true)),
2745 ])),
2746 data: Some(Json(json!({ "content": "hi" }))),
2747 },
2748 ),
2749 ),
2750 (
2751 "EventsResult",
2752 to(
2753 lua,
2754 EventsResult(
2755 EventRows(vec![EventRow {
2756 kind: "msg_user".into(),
2757 seq: 2,
2758 epoch_ms: 1_700_000_000_000,
2759 _schema_version: 1,
2760 beat: None,
2761 meta: None,
2762 data: Json(json!({ "content": "hi" })),
2763 }]),
2764 false,
2765 ),
2766 ),
2767 ),
2768 (
2769 "QueryParams",
2770 to(lua, QueryParams::Positional(vec![json!("note")])),
2771 ),
2772 (
2773 "QueryOpts",
2774 to(
2775 lua,
2776 QueryOpts {
2777 sessions: Some(vec!["s-1".into(), "s-2".into()]),
2778 timeout_ms: Some(250),
2779 limit: Some(10),
2780 },
2781 ),
2782 ),
2783 (
2784 "QueryResult",
2785 to(
2786 lua,
2787 QueryResult(vec![Json(json!({ "kind": "msg_user" }))], true),
2788 ),
2789 ),
2790 (
2791 "ErrorTable",
2792 to(
2793 lua,
2794 ErrorTable {
2795 kind: Some("closed".into()),
2796 method: Some("append".into()),
2797 retryable: false,
2798 message: "the session is closed".into(),
2799 },
2800 ),
2801 ),
2802 (
2803 "ApiReport",
2804 to(
2805 lua,
2806 ApiReport {
2807 session: vec![ApiEntry {
2808 name: "append".into(),
2809 doc: "append(event) -> seq".into(),
2810 }],
2811 module: vec![ApiEntry {
2812 name: "open".into(),
2813 doc: "open(opts?) -> session".into(),
2814 }],
2815 errors: vec!["busy".into()],
2816 schema: ApiSchema {
2817 table: "events".into(),
2818 columns: vec![ApiColumn {
2819 name: "seq".into(),
2820 declared_type: "INTEGER".into(),
2821 pk: true,
2822 }],
2823 },
2824 fields: ApiFields {
2825 amount: "amount".into(),
2826 tag: "tag".into(),
2827 desc: "desc".into(),
2828 remaining: "remaining".into(),
2829 scope_id: "scope_id".into(),
2830 owner: "owner".into(),
2831 parent: "parent".into(),
2832 child: "child".into(),
2833 reason: "reason".into(),
2834 detail: "detail".into(),
2835 open_children: "open_children".into(),
2836 },
2837 types: "-- generated".into(),
2838 },
2839 ),
2840 ),
2841 ]
2842 }
2843
2844 /// (c) The module loads under the lshape the host ships, and exports one
2845 /// shape per declared type and nothing else.
2846 #[test]
2847 fn the_generated_module_exports_exactly_the_declared_types() {
2848 let (_lua, module, _check) = types_vm();
2849
2850 let mut exported: Vec<String> = module
2851 .pairs::<String, LuaValue>()
2852 .map(|pair| pair.expect("a module entry").0)
2853 .collect();
2854 exported.sort();
2855
2856 let mut expected: Vec<String> = declared()
2857 .into_iter()
2858 .map(|(name, _, _)| name.to_string())
2859 .collect();
2860 expected.sort();
2861
2862 assert_eq!(exported, expected, "the generated module drifted");
2863 }
2864
2865 /// (c) A value built from each Rust type passes the shape generated for
2866 /// it. This is the test that the two agree: `to_value` of the fixture is
2867 /// what a syscall would hand Lua, and `check.check` is what the Lua
2868 /// kernel's dev gate runs.
2869 #[test]
2870 fn every_declared_type_accepts_a_value_built_from_its_rust_type() {
2871 let (lua, module, check) = types_vm();
2872
2873 let samples = samples(&lua);
2874 let mut sampled: Vec<&str> = samples.iter().map(|(name, _)| *name).collect();
2875 sampled.sort_unstable();
2876 let mut expected: Vec<&str> = declared().into_iter().map(|(name, _, _)| name).collect();
2877 expected.sort_unstable();
2878 assert_eq!(
2879 sampled, expected,
2880 "every declared type needs a sample built from it"
2881 );
2882
2883 for (name, value) in samples {
2884 let shape: LuaValue = module.get(name).expect("the shape of a declared type");
2885 let (ok, why): (bool, Option<String>) = check
2886 .call((value, shape))
2887 .unwrap_or_else(|e| panic!("{name}: {e}"));
2888 assert!(ok, "{name}: {}", why.unwrap_or_default());
2889 }
2890 }
2891
2892 /// (c) And the shapes refuse: a closed options table is closed on the Lua
2893 /// side too, which is what makes the generated declaration worth running.
2894 #[test]
2895 fn the_generated_shapes_refuse_what_the_rust_types_refuse() {
2896 let (lua, module, check) = types_vm();
2897
2898 let refused: [(&str, LuaValue); 3] = [
2899 // An option the kernel does not know must not quietly do nothing.
2900 (
2901 "QueryOpts",
2902 lua.to_value(&json!({ "rows": 10 })).expect("value"),
2903 ),
2904 // `n` counts events.
2905 (
2906 "ViewOpts",
2907 lua.to_value(&json!({ "count": 2 })).expect("value"),
2908 ),
2909 // `meta` is shallow.
2910 (
2911 "AppendEvent",
2912 lua.to_value(&json!({ "kind": "note", "meta": { "deep": { "no": 1 } } }))
2913 .expect("value"),
2914 ),
2915 ];
2916
2917 for (name, value) in refused {
2918 let shape: LuaValue = module.get(name).expect("the shape of a declared type");
2919 let (ok, _why): (bool, Option<String>) = check
2920 .call((value, shape))
2921 .unwrap_or_else(|e| panic!("{name}: {e}"));
2922 assert!(!ok, "{name} accepted a value its Rust type refuses");
2923 }
2924 }
2925
2926 /// (c) `knl.api().types` is the same text the host embeds, so a tool that
2927 /// asks the kernel what it takes reads the module that is actually loaded.
2928 #[test]
2929 fn the_api_publishes_the_module_it_generated() {
2930 let lua = Lua::new();
2931 let dir = tempfile::tempdir().expect("tempdir");
2932 register(&lua, knl::IsleDrivers::new(), dir.path().join("knl.sqlite"))
2933 .expect("register knl");
2934 let published: String = lua
2935 .load(r#"return knl.api().types"#)
2936 .eval()
2937 .expect("knl.api().types");
2938 assert_eq!(published, lshape_module_source());
2939 }
2940}
2941
2942#[cfg(test)]
2943mod tests {
2944 use super::*;
2945
2946 /// Test helpers loaded into every VM.
2947 const FIXTURES: &str = r#"
2948 -- The recorded kinds in order, as one comparable string.
2949 function kinds_of(s)
2950 local names = {}
2951 for _, e in ipairs(s:events()) do
2952 table.insert(names, e.kind)
2953 end
2954 return table.concat(names, ",")
2955 end
2956
2957 -- The classified failure of a call that is supposed to fail, plus
2958 -- the raised value itself for the tests that check how it reads.
2959 function failure(fn, ...)
2960 local ok, raised = pcall(fn, ...)
2961 assert(not ok, "the call was supposed to fail")
2962 return knl.error(raised), raised
2963 end
2964 "#;
2965
2966 /// A Lua VM with the `knl` bridge on it, and the two things a session
2967 /// now needs around it: a runtime to yield into, and somewhere for its
2968 /// connection threads to be owned.
2969 ///
2970 /// Every session method that reaches the store suspends, so a chunk that
2971 /// calls one has to run as a coroutine on a runtime — [`Vm::exec`] is
2972 /// that, and it is why the chunks below say `vm.exec(...)` where they
2973 /// used to say `lua.load(...).exec()`. The assertions inside them are
2974 /// unchanged.
2975 struct Vm {
2976 lua: Lua,
2977 /// The connection threads of every session the chunks open, held for
2978 /// the test's lifetime exactly as the host holds them for a run's.
2979 drivers: knl::IsleDrivers,
2980 rt: tokio::runtime::Runtime,
2981 /// The default store's directory, held so it outlives the VM.
2982 ///
2983 /// A session opened without a `store` goes into the file the host
2984 /// owns; here that is one temp file per VM, which is what keeps a
2985 /// test's sessions out of the developer's own database and out of
2986 /// every other test's.
2987 dir: tempfile::TempDir,
2988 }
2989
2990 impl Vm {
2991 /// Fresh VM with only the `knl` bridge registered.
2992 fn new() -> Self {
2993 let lua = Lua::new();
2994 let drivers = knl::IsleDrivers::new();
2995 let dir = tempfile::tempdir().expect("tempdir");
2996 register(&lua, drivers.clone(), dir.path().join("knl.sqlite")).expect("register knl");
2997 let rt = tokio::runtime::Builder::new_current_thread()
2998 .enable_all()
2999 .build()
3000 .expect("a runtime for the VM to yield into");
3001 rt.block_on(async { lua.load(FIXTURES).exec_async().await })
3002 .expect("fixtures");
3003 Self {
3004 lua,
3005 drivers,
3006 rt,
3007 dir,
3008 }
3009 }
3010
3011 /// The file a session with no `store` of its own lands in.
3012 fn default_store(&self) -> std::path::PathBuf {
3013 self.dir.path().join("knl.sqlite")
3014 }
3015
3016 /// Run `chunk` to completion, as a coroutine.
3017 fn exec(&self, chunk: &str) -> LuaResult<()> {
3018 self.rt
3019 .block_on(async { self.lua.load(chunk).exec_async().await })
3020 }
3021
3022 /// Run `chunk` and take what it returns.
3023 fn eval<R: mlua::FromLuaMulti>(&self, chunk: &str) -> LuaResult<R> {
3024 self.rt
3025 .block_on(async { self.lua.load(chunk).eval_async::<R>().await })
3026 }
3027
3028 /// Run a chunk that is expected to fail, returning the message.
3029 fn expect_err(&self, chunk: &str) -> String {
3030 self.exec(chunk)
3031 .expect_err("chunk was expected to fail")
3032 .to_string()
3033 }
3034
3035 /// Drive `f` on this VM's runtime — for the assertions that read a
3036 /// store directly rather than through Lua.
3037 fn block_on<F: std::future::Future>(&self, f: F) -> F::Output {
3038 self.rt.block_on(f)
3039 }
3040
3041 /// Let go of the VM and drain its connection threads.
3042 ///
3043 /// Dropping the Lua state collects every session userdata, which is
3044 /// where a handle nobody closed submits its boundary without waiting;
3045 /// shutting the drivers down is what waits for those writes to land.
3046 /// A test that reads the database afterwards calls this first.
3047 fn finish(self) {
3048 drop(self.finish_keeping_the_store());
3049 }
3050
3051 /// [`Vm::finish`], handing the default store's directory back.
3052 ///
3053 /// For the test that reads the default database itself: the temp dir
3054 /// is deleted when it drops, so a caller that wants to open the file
3055 /// afterwards has to hold it.
3056 fn finish_keeping_the_store(self) -> tempfile::TempDir {
3057 let Self {
3058 lua,
3059 drivers,
3060 rt,
3061 dir,
3062 } = self;
3063 drop(lua);
3064 let failures = rt.block_on(drivers.shutdown());
3065 assert!(
3066 failures.is_empty(),
3067 "the connection threads did not shut down cleanly: {failures:?}"
3068 );
3069 dir
3070 }
3071 }
3072
3073 /// Fresh Lua VM with only the `knl` bridge registered.
3074 fn vm() -> Vm {
3075 Vm::new()
3076 }
3077
3078 /// (Happy path) append assigns strictly increasing seq numbers, `len`
3079 /// tracks them, and `events()` exposes the caller fields plus the
3080 /// kernel-owned `seq` / `epoch_ms`. Seq 1 is the kernel's own
3081 /// `session_opened`.
3082 #[test]
3083 fn append_assigns_monotonic_seq_and_len_tracks() {
3084 let vm = vm();
3085 vm.exec(
3086 r#"
3087 local s = knl.open()
3088 assert(s:len() == 1, "a fresh session holds session_opened")
3089 local a = s:append({ kind = "user_msg", data = { text = "hi" } })
3090 local b = s:append({ kind = "note", data = { name = "sh" } })
3091 assert(a == 2, "first caller seq: " .. tostring(a))
3092 assert(b == 3, "second caller seq: " .. tostring(b))
3093 assert(s:len() == 3, "len: " .. tostring(s:len()))
3094
3095 local evs = s:events()
3096 assert(#evs == 3, "events len: " .. tostring(#evs))
3097 assert(evs[1].kind == "session_opened")
3098 assert(evs[2].kind == "user_msg")
3099 assert(evs[2].data.text == "hi")
3100 assert(evs[2].seq == 2)
3101 assert(type(evs[2].epoch_ms) == "number", "epoch_ms must be a number")
3102 assert(evs[3].kind == "note")
3103 assert(evs[3].seq == 3)
3104
3105 -- The envelope is closed: a kind's own field at the top level is
3106 -- refused, with the place it belongs in the message.
3107 local err = failure(function() s:append({ kind = "note", text = "hi" }) end)
3108 assert(err.kind == "validation", "kind: " .. tostring(err.kind))
3109 assert(err.message:find("under data"), "message: " .. err.message)
3110 "#,
3111 )
3112 .expect("happy path chunk");
3113 }
3114
3115 /// (I1) No mutation API is reachable on the session userdata.
3116 #[test]
3117 fn session_exposes_no_mutation_api() {
3118 let vm = vm();
3119 vm.exec(
3120 r#"
3121 local s = knl.open()
3122 s:append({ kind = "user_msg" })
3123 for _, name in ipairs({ "update", "delete", "replace", "set", "insert",
3124 "remove", "clear", "truncate", "pop" }) do
3125 local ok, v = pcall(function() return s[name] end)
3126 assert(not ok or v == nil, "mutation API must not exist: " .. name)
3127 end
3128 "#,
3129 )
3130 .expect("mutation-surface chunk");
3131 }
3132
3133 /// (I1) The table returned by `events()` is a deep copy: mutating it,
3134 /// including nested tables and the array itself, leaves the history
3135 /// untouched.
3136 #[test]
3137 fn events_returns_a_deep_copy() {
3138 let vm = vm();
3139 vm.exec(
3140 r#"
3141 local s = knl.open()
3142 s:append({ kind = "user_msg", meta = { tag = "a" },
3143 data = { text = "hi", blocks = { { type = "text" } } } })
3144
3145 local evs = s:events()
3146 evs[2].kind = "TAMPERED"
3147 evs[2].data.text = nil
3148 evs[2].data.extra = "injected"
3149 evs[2].meta.tag = "b"
3150 evs[2].data.blocks[1].type = "tampered"
3151 table.insert(evs, { kind = "ghost" })
3152
3153 local again = s:events()
3154 assert(#again == 2, "history length changed: " .. tostring(#again))
3155 assert(again[2].kind == "user_msg", "kind changed: " .. tostring(again[2].kind))
3156 assert(again[2].data.text == "hi", "data changed")
3157 assert(again[2].data.extra == nil, "field injected into history")
3158 assert(again[2].meta.tag == "a", "meta changed")
3159 assert(again[2].data.blocks[1].type == "text", "nested table changed")
3160 "#,
3161 )
3162 .expect("deep copy chunk");
3163 }
3164
3165 /// (I1) `seq` / `epoch_ms` are kernel-owned: a caller-supplied value is
3166 /// overwritten rather than trusted. There is no `author` field.
3167 #[test]
3168 fn kernel_owned_fields_override_caller_values() {
3169 let vm = vm();
3170 vm.exec(
3171 r#"
3172 local s = knl.open()
3173 local seq = s:append({ kind = "user_msg", seq = 999, epoch_ms = 1 })
3174 assert(seq == 2, "returned seq: " .. tostring(seq))
3175 local e = s:events(2)[1]
3176 assert(e.seq == 2, "stored seq: " .. tostring(e.seq))
3177 assert(e.epoch_ms ~= 1, "epoch_ms must be kernel-assigned")
3178 assert(e.author == nil, "there is no per-event author anymore")
3179 "#,
3180 )
3181 .expect("kernel-owned field chunk");
3182 }
3183
3184 /// `events(from)` returns the tail with `seq >= from`.
3185 #[test]
3186 fn events_from_filters_by_seq() {
3187 let vm = vm();
3188 vm.exec(
3189 r#"
3190 local s = knl.open()
3191 for i = 1, 3 do s:append({ kind = "e" .. i }) end
3192 local tail = s:events(3)
3193 assert(#tail == 2, "tail len: " .. tostring(#tail))
3194 assert(tail[1].seq == 3 and tail[2].seq == 4)
3195 assert(#s:events(5) == 0, "past-the-end filter must be empty")
3196 assert(#s:events(0) == 4, "from=0 must return everything")
3197
3198 -- A read that reached the end of the stream says so.
3199 local rows, truncated = s:events()
3200 assert(#rows == 4 and truncated == false, "nothing was cut off")
3201 "#,
3202 )
3203 .expect("events(from) chunk");
3204 }
3205
3206 /// `events` is bounded, and it says when the bound bit.
3207 ///
3208 /// The regression this pins: the read used to be `usize::MAX` — every
3209 /// event of a stream decoded, upcasted and built into a Lua table on the
3210 /// VM's own thread, with nothing but the caller's discretion between a
3211 /// long-running session and the whole of its log in memory at once. It
3212 /// now stops at the kernel's row cap and answers the pair `query`
3213 /// answers, so "there is more" is a fact the read reported rather than one
3214 /// a caller has to infer from a suspiciously round count.
3215 #[test]
3216 fn events_stops_at_the_row_cap_and_says_it_cut() {
3217 let vm = vm();
3218 // The cap is the kernel's, so the chunk is written against it rather
3219 // than against a number typed in twice.
3220 let chunk = format!(
3221 r#"
3222 local cap = {cap}
3223 -- An in-memory database on purpose: this seeds a thousand events
3224 -- and the default store is a file, whose every commit is an fsync.
3225 local s = knl.open({{ store = "mem" }})
3226 -- The opening is already an event, so this is one past the cap.
3227 for i = 1, cap do s:append({{ kind = "e" .. i }}) end
3228 assert(s:len() == cap + 1, "seeded: " .. tostring(s:len()))
3229
3230 local rows, truncated = s:events()
3231 assert(#rows == cap, "the read is capped: " .. tostring(#rows))
3232 assert(truncated == true, "and it says the cap cut the read short")
3233 assert(rows[1].seq == 1 and rows[cap].seq == cap, "the page is the front of the log")
3234
3235 -- The rest is read by paging on `from`, and that page is whole.
3236 local rest, more = s:events(cap + 1)
3237 assert(#rest == 1, "the remainder: " .. tostring(#rest))
3238 assert(more == false, "and nothing is left after it")
3239 assert(rest[1].seq == cap + 1, "seq: " .. tostring(rest[1].seq))
3240 "#,
3241 cap = knl::DEFAULT_LIMIT,
3242 );
3243 vm.exec(&chunk).expect("events cap chunk");
3244 }
3245
3246 /// (attribution) `append` rejects a missing / non-string `kind` and a
3247 /// non-table event, with `knl: append:` in the message.
3248 #[test]
3249 fn append_validates_event_shape_with_attributed_errors() {
3250 let vm = vm();
3251
3252 let msg = vm.expect_err(r#"knl.open():append({ text = "no kind" })"#);
3253 assert!(msg.contains("knl: append:"), "missing attribution: {msg}");
3254 assert!(msg.contains("missing field `kind`"), "{msg}");
3255
3256 let msg = vm.expect_err(r#"knl.open():append({ kind = 42 })"#);
3257 assert!(msg.contains("knl: append:"), "missing attribution: {msg}");
3258 assert!(msg.contains("event.kind"), "{msg}");
3259 assert!(msg.contains("expected a string"), "{msg}");
3260
3261 let msg = vm.expect_err(r#"knl.open():append("not a table")"#);
3262 assert!(msg.contains("knl: append:"), "missing attribution: {msg}");
3263 assert!(msg.contains("event:"), "{msg}");
3264 assert!(msg.contains("expected table"), "{msg}");
3265
3266 // A rejected append leaves no trace in the history.
3267 vm.exec(
3268 r#"
3269 local s = knl.open()
3270 pcall(function() s:append({ text = "no kind" }) end)
3271 assert(s:len() == 1, "rejected append was recorded")
3272 assert(s:append({ kind = "ok" }) == 2, "seq must not be consumed by a failure")
3273 "#,
3274 )
3275 .expect("rejected-append chunk");
3276 }
3277
3278 /// (I3) A negative `spend` is an error, attributed to `knl: spend:`,
3279 /// and leaves the balance untouched.
3280 #[test]
3281 fn spend_rejects_negative_amounts() {
3282 let vm = vm();
3283 let msg = vm.expect_err(
3284 r#"
3285 local s = knl.open({ budget = { amount = 100, tag = "beats" } })
3286 s:spend(-1)
3287 "#,
3288 );
3289 assert!(msg.contains("knl: spend:"), "missing attribution: {msg}");
3290 assert!(msg.contains("non-negative"), "{msg}");
3291
3292 vm.exec(
3293 r#"
3294 local s = knl.open({ budget = { amount = 100, tag = "beats" } })
3295 pcall(function() s:spend(-1) end)
3296 assert(s:remaining() == 100, "balance changed: " .. tostring(s:remaining()))
3297 -- A negative spend is rejected even without a budget.
3298 local ok = pcall(function() knl.open():spend(-1) end)
3299 assert(not ok, "negative spend must be rejected without a budget too")
3300 -- So is a non-numeric amount.
3301 local ok2 = pcall(function() knl.open():spend("many") end)
3302 assert(not ok2, "a non-numeric amount must be rejected")
3303 "#,
3304 )
3305 .expect("negative-spend chunk");
3306 }
3307
3308 /// (I3) `remaining` is non-increasing across a call sequence, is
3309 /// floored at zero, and `exhausted()` flips once the budget is used
3310 /// up. `spend` itself answers nothing: the balance is `remaining()`.
3311 #[test]
3312 fn spend_is_monotonic_and_flips_exhausted() {
3313 let vm = vm();
3314 vm.exec(
3315 r#"
3316 local s = knl.open({ budget = { amount = 1000, tag = "beats" } })
3317 assert(s:remaining() == 1000)
3318 assert(s:exhausted() == false)
3319
3320 local prev = s:remaining()
3321 for _, n in ipairs({ 120, 0, 300, 80 }) do
3322 assert(s:spend(n) == nil, "spend answers with the write, not a number")
3323 local r = s:remaining()
3324 assert(r <= prev, "remaining rose: " .. tostring(prev) .. " -> " .. tostring(r))
3325 prev = r
3326 end
3327 assert(s:remaining() == 500, "remaining: " .. tostring(s:remaining()))
3328 assert(s:exhausted() == false)
3329
3330 -- Overspending floors at zero and never goes negative.
3331 s:spend(9999)
3332 assert(s:remaining() == 0, "floor: " .. tostring(s:remaining()))
3333 assert(s:exhausted() == true, "exhausted must flip after overspending")
3334 s:spend(1)
3335 assert(s:remaining() == 0, "spending past zero stays at zero")
3336 "#,
3337 )
3338 .expect("budget monotonicity chunk");
3339 }
3340
3341 /// (I3) Without a budget, `remaining()` is nil, `spend` records nothing
3342 /// and the session is never exhausted.
3343 #[test]
3344 fn session_without_budget_reports_nil() {
3345 let vm = vm();
3346 vm.exec(
3347 r#"
3348 local s = knl.open()
3349 assert(s:remaining() == nil, "remaining must be nil without a budget")
3350 assert(s:spend(50) == nil, "spend answers nothing")
3351 assert(s:len() == 1, "a settlement without a budget records nothing")
3352 assert(s:exhausted() == false, "no budget can never be exhausted")
3353
3354 -- An empty opts table behaves the same way.
3355 local s2 = knl.open({})
3356 assert(s2:remaining() == nil)
3357 "#,
3358 )
3359 .expect("no-budget chunk");
3360 }
3361
3362 /// (attribution) Malformed `budget` options are rejected by
3363 /// `knl.open` itself.
3364 ///
3365 /// Two kinds of refusal meet here and the messages say which is which.
3366 /// The *shape* is the declared type's ([`types::BudgetOpt`], read by
3367 /// [`from_lua`]), so a misspelt field or a mistyped one names the path it
3368 /// was at — `opts.budget.tag` — and the whole set of fields it could have
3369 /// been. The *rule* is the bridge's: a quota has to be there, it cannot
3370 /// be negative, and the two forms exclude each other. No schema states
3371 /// any of those three, so they are checked after the parse and keep their
3372 /// own words.
3373 #[test]
3374 fn session_validates_budget_options() {
3375 let vm = vm();
3376
3377 let msg = vm.expect_err(r#"knl.open({ budget = { amount = -1 } })"#);
3378 assert!(msg.contains("knl: open:"), "missing attribution: {msg}");
3379 assert!(msg.contains("budget.amount"), "{msg}");
3380
3381 let msg = vm.expect_err(r#"knl.open({ budget = {} })"#);
3382 assert!(msg.contains("knl: open:"), "missing attribution: {msg}");
3383 assert!(msg.contains("required"), "{msg}");
3384
3385 // A misspelt field is an error, not a silently ignored cap: the
3386 // failure a budget exists to prevent is exactly "the limit I set
3387 // was not read". The refusal names the field and the set it is not
3388 // in, which is the deserializer reading the declared type.
3389 let msg = vm.expect_err(r#"knl.open({ budget = { tokens = 100 } })"#);
3390 assert!(msg.contains("knl: open:"), "missing attribution: {msg}");
3391 assert!(msg.contains("unknown field `tokens`"), "{msg}");
3392 assert!(msg.contains("`amount`"), "{msg}");
3393
3394 let msg = vm.expect_err(r#"knl.open({ budget = { amount = 10, tag = 7 } })"#);
3395 assert!(msg.contains("opts.budget.tag"), "{msg}");
3396 assert!(msg.contains("expected a string"), "{msg}");
3397
3398 let msg = vm.expect_err(r#"knl.open({ budget = { amount = 1.5 } })"#);
3399 assert!(msg.contains("opts.budget.amount"), "{msg}");
3400
3401 // The words are optional, and carried verbatim when given.
3402 vm.exec(
3403 r#"
3404 local s = knl.open({ budget = { amount = 42, tag = "tokens",
3405 desc = "one nightly run" } })
3406 assert(s:remaining() == 42, "remaining: " .. tostring(s:remaining()))
3407 local granted = s:events()[2]
3408 assert(granted.kind == "budget_granted", "kind: " .. tostring(granted.kind))
3409 assert(granted.data.amount == 42 and granted.data.tag == "tokens")
3410 assert(granted.data.desc == "one nightly run",
3411 "desc: " .. tostring(granted.data.desc))
3412
3413 local bare = knl.open({ budget = { amount = 7 } })
3414 local g2 = bare:events()[2].data
3415 assert(g2.amount == 7 and g2.tag == nil and g2.desc == nil,
3416 "a grant with no words must invent none")
3417 "#,
3418 )
3419 .expect("grant options chunk");
3420
3421 let msg = vm.expect_err(r#"knl.open({ budget = 100 })"#);
3422 assert!(msg.contains("knl: open:"), "missing attribution: {msg}");
3423 assert!(msg.contains("opts.budget"), "{msg}");
3424 assert!(msg.contains("expected table"), "{msg}");
3425
3426 let msg = vm.expect_err(r#"knl.open("nope")"#);
3427 assert!(msg.contains("knl: open:"), "missing attribution: {msg}");
3428 assert!(msg.contains("opts:"), "{msg}");
3429 assert!(msg.contains("expected table"), "{msg}");
3430 }
3431
3432 /// (I6) Two sessions share nothing: ids differ and history / budget
3433 /// of one is invisible to the other.
3434 #[test]
3435 fn two_sessions_are_independent() {
3436 let vm = vm();
3437 vm.exec(
3438 r#"
3439 local a = knl.open({ budget = { amount = 100, tag = "beats" } })
3440 local b = knl.open({ budget = { amount = 100, tag = "beats" } })
3441
3442 assert(type(a:id()) == "string" and #a:id() > 0, "id must be a non-empty string")
3443 assert(a:id() ~= b:id(), "session ids must be unique")
3444
3445 a:append({ kind = "only_in_a" })
3446 a:spend(60)
3447
3448 -- a: session_opened, budget_granted, only_in_a, budget_spent.
3449 -- b: session_opened, budget_granted.
3450 assert(a:len() == 4 and b:len() == 2, "history leaked between sessions")
3451 assert(#b:events(3) == 0, "b holds only its own opening")
3452 assert(a:remaining() == 40 and b:remaining() == 100, "budget leaked between sessions")
3453
3454 -- Closing one leaves the other usable.
3455 a:close()
3456 assert(b:append({ kind = "still_open" }) == 3)
3457 "#,
3458 )
3459 .expect("session independence chunk");
3460 }
3461
3462 /// (I6) After `close()`, `append` and `spend` are errors; read-only
3463 /// methods keep working and `close()` is idempotent.
3464 #[test]
3465 fn closed_session_rejects_append_and_spend() {
3466 let vm = vm();
3467
3468 let msg = vm.expect_err(
3469 r#"
3470 local s = knl.open()
3471 s:close()
3472 s:append({ kind = "after_close" })
3473 "#,
3474 );
3475 assert!(msg.contains("knl: append:"), "missing attribution: {msg}");
3476 assert!(msg.contains("session is closed"), "{msg}");
3477
3478 let msg = vm.expect_err(
3479 r#"
3480 local s = knl.open({ budget = { amount = 10, tag = "beats" } })
3481 s:close()
3482 s:spend(1)
3483 "#,
3484 );
3485 assert!(msg.contains("knl: spend:"), "missing attribution: {msg}");
3486 assert!(msg.contains("session is closed"), "{msg}");
3487
3488 // A closed session cannot be granted more either.
3489 let msg = vm.expect_err(
3490 r#"
3491 local s = knl.open({ budget = { amount = 10, tag = "beats" } })
3492 s:close()
3493 s:reserve(1)
3494 "#,
3495 );
3496 assert!(msg.contains("knl: reserve:"), "missing attribution: {msg}");
3497 assert!(msg.contains("session is closed"), "{msg}");
3498
3499 vm.exec(
3500 r#"
3501 local s = knl.open({ budget = { amount = 10, tag = "beats" } })
3502 s:append({ kind = "before_close" })
3503 s:spend(4)
3504 s:close()
3505 s:close() -- idempotent
3506
3507 -- Reads still work after the session ends.
3508 assert(s:len() == 5,
3509 "session_opened + budget_granted + before_close + budget_spent + session_closed")
3510 assert(s:events()[3].kind == "before_close")
3511 assert(s:remaining() == 6)
3512 assert(s:exhausted() == false)
3513 assert(type(s:id()) == "string")
3514 "#,
3515 )
3516 .expect("closed-session read chunk");
3517 }
3518
3519 /// (I6) The bridge installs exactly one global and keeps no state
3520 /// there: a second VM starts with its own fresh session.
3521 #[test]
3522 fn state_lives_in_the_userdata_not_in_globals() {
3523 let vm_a = vm();
3524 vm_a.exec(
3525 r#"
3526 local s = knl.open()
3527 s:append({ kind = "in_vm_a" })
3528 assert(s:len() == 2)
3529 -- `knl` itself carries no session state.
3530 assert(knl.events == nil and knl.append == nil and knl.spend == nil)
3531 "#,
3532 )
3533 .expect("vm a chunk");
3534
3535 let vm_b = vm();
3536 vm_b.exec(
3537 r#"
3538 local s = knl.open()
3539 assert(s:len() == 1, "a second VM starts with only its own session_opened")
3540 assert(s:events()[1].kind == "session_opened")
3541 "#,
3542 )
3543 .expect("vm b chunk");
3544 }
3545
3546 /// The kernel brackets the session: `session_opened` on open,
3547 /// `session_closed` on close, with the caller's reason (or the default).
3548 #[test]
3549 fn session_boundaries_are_recorded_by_the_kernel() {
3550 let vm = vm();
3551 vm.exec(
3552 r#"
3553 local s = knl.open()
3554 local opened = s:events()[1]
3555 assert(opened.kind == "session_opened", "kind: " .. tostring(opened.kind))
3556 assert(opened.seq == 1)
3557
3558 s:close("budget_exhausted")
3559 local evs = s:events()
3560 assert(#evs == 2, "close must record session_closed")
3561 assert(evs[2].kind == "session_closed")
3562 assert(evs[2].data.reason == "budget_exhausted")
3563
3564 s:close("ignored")
3565 assert(s:len() == 2, "close is idempotent")
3566
3567 -- Without a reason the kernel records its default.
3568 local d = knl.open()
3569 d:close()
3570 assert(d:events()[2].data.reason == "closed", "default reason")
3571 "#,
3572 )
3573 .expect("session boundary chunk");
3574
3575 let msg = vm.expect_err(r#"knl.open():close({ not_a = "string" })"#);
3576 assert!(msg.contains("knl: close:"), "missing attribution: {msg}");
3577 assert!(msg.contains("reason:"), "{msg}");
3578 assert!(msg.contains("expected a string"), "{msg}");
3579 }
3580
3581 /// The kernel checks the *envelope* of every event — the closed set of
3582 /// top-level keys, a string beat, a shallow meta, a table data — and the
3583 /// shape of a kind's own `data` is the writer's business, not its.
3584 #[test]
3585 fn the_envelope_is_validated_and_a_kinds_own_data_is_not() {
3586 let vm = vm();
3587
3588 // A kind's own field at the top level: refused, and the message says
3589 // where it goes.
3590 let msg = vm.expect_err(r#"knl.open():append({ kind = "msg_user", content = "hi" })"#);
3591 assert!(msg.contains("knl: append:"), "missing attribution: {msg}");
3592 assert!(msg.contains("content"), "{msg}");
3593 assert!(msg.contains("under data"), "{msg}");
3594
3595 // `meta` is shallow: nesting belongs under `data`.
3596 let msg =
3597 vm.expect_err(r#"knl.open():append({ kind = "note", meta = { deep = { a = 1 } } })"#);
3598 assert!(msg.contains("knl: append:"), "missing attribution: {msg}");
3599 assert!(msg.contains("meta is shallow"), "{msg}");
3600
3601 let msg = vm.expect_err(r#"knl.open():append({ kind = "note", data = 7 })"#);
3602 assert!(msg.contains("data must be a table"), "{msg}");
3603
3604 vm.exec(
3605 r#"
3606 local s = knl.open()
3607 pcall(function() s:append({ kind = "note", text = "hi" }) end)
3608 assert(s:len() == 1, "a rejected event was recorded")
3609
3610 -- The kinds of a turn are the Lua kernel's, shape and all: the
3611 -- Rust side takes whatever `data` says, at any depth.
3612 local beat = knl.new_beat_id()
3613 s:append({ kind = "msg_user", data = { content = "hi" } })
3614 s:append({ kind = "tool_call", beat = beat,
3615 data = { call_id = "c1", name = "sh", args = { cmd = "ls" } } })
3616 s:append({ kind = "tool_result", beat = beat,
3617 data = { call_id = "c1", ok = false, result = "boom" } })
3618 -- …including an empty one.
3619 s:append({ kind = "tool_call" })
3620 assert(s:len() == 5)
3621 assert(s:events()[3].beat == beat, "the declared beat is recorded")
3622 assert(s:events()[5].beat == nil, "an undeclared beat stays absent")
3623 assert(s:events()[3].data.args.cmd == "ls", "data comes back at any depth")
3624 assert(next(s:events()[5].data) == nil, "an absent data reads as empty")
3625
3626 -- meta takes scalars, and comes back as it was written.
3627 s:append({ kind = "note", meta = { label = "a", attempt = 2, retried = true } })
3628 local m = s:events()[6].meta
3629 assert(m.label == "a" and m.attempt == 2 and m.retried == true,
3630 "meta round-trips")
3631
3632 -- A numbered beat is refused, on any kind.
3633 local ok = pcall(function() s:append({ kind = "note", beat = 1 }) end)
3634 assert(not ok, "a numeric beat was accepted")
3635 "#,
3636 )
3637 .expect("envelope chunk");
3638 }
3639
3640 /// The budget ledger is the kernel's: Lua can read those events but not
3641 /// write them. Appending one by hand would be granting yourself the
3642 /// quota your owner set, so it is refused and the balance does not
3643 /// move.
3644 #[test]
3645 fn lua_cannot_append_the_budget_kinds_by_hand() {
3646 let vm = vm();
3647
3648 let msg = vm.expect_err(
3649 r#"
3650 local s = knl.open({ budget = { amount = 10, tag = "beats" } })
3651 s:append({ kind = "budget_reserved", data = { amount = 5 } })
3652 "#,
3653 );
3654 assert!(msg.contains("knl: append:"), "missing attribution: {msg}");
3655 assert!(msg.contains("kernel only"), "{msg}");
3656 assert!(msg.contains("budget_reserved"), "{msg}");
3657
3658 vm.exec(
3659 r#"
3660 local s = knl.open({ budget = { amount = 10, tag = "beats" } })
3661 for _, ev in ipairs({
3662 { kind = "budget_granted", data = { amount = 1000000 } },
3663 { kind = "budget_reserved", data = { amount = 5 } },
3664 { kind = "budget_refused", data = { amount = 5, remaining = 0 } },
3665 { kind = "budget_spent", data = { amount = 5 } },
3666 }) do
3667 local ok = pcall(function() s:append(ev) end)
3668 assert(not ok, "a caller wrote " .. ev.kind)
3669 end
3670
3671 assert(s:len() == 2, "a rejected budget event was recorded: " .. tostring(s:len()))
3672 assert(s:remaining() == 10, "a forged event moved the balance")
3673
3674 -- Reading them is fine: the kernel's own writes are in the log
3675 -- like everything else.
3676 s:reserve(4)
3677 local evs = s:events()
3678 assert(evs[3].kind == "budget_reserved" and evs[3].data.amount == 4,
3679 "the kernel's own reservation is readable")
3680 "#,
3681 )
3682 .expect("kernel-only kind chunk");
3683 }
3684
3685 /// (K4) `reserve` is the decision point: it takes what fits, refuses
3686 /// what does not without moving the balance, and names the grant when
3687 /// it refuses. Every answer is a fact in the log.
3688 #[test]
3689 fn reserve_grants_refuses_and_records_both() {
3690 let vm = vm();
3691 vm.exec(
3692 r#"
3693 local s = knl.open({ budget = { amount = 100, tag = "beats" } })
3694
3695 local ok, tag = s:reserve(30)
3696 assert(ok == true, "a covered reservation must be granted")
3697 assert(tag == nil, "a granted reservation names no budget")
3698 assert(s:remaining() == 70, "remaining: " .. tostring(s:remaining()))
3699
3700 local ok2, tag2 = s:reserve(1000)
3701 assert(ok2 == false, "an uncovered reservation must be refused")
3702 assert(tag2 == "beats", "a refusal must name the budget: " .. tostring(tag2))
3703 assert(s:remaining() == 70, "a refusal must not deduct")
3704 assert(s:exhausted() == false, "a refusal must not exhaust")
3705
3706 -- Both answers are in the log, with what was asked for.
3707 local evs = s:events()
3708 assert(evs[3].kind == "budget_reserved" and evs[3].data.amount == 30)
3709 assert(evs[3].data.tag == "beats")
3710 assert(evs[4].kind == "budget_refused" and evs[4].data.amount == 1000)
3711 assert(evs[4].data.remaining == 70, "the refusal records what there was")
3712
3713 -- Exactly the balance is coverable, and zero always is.
3714 assert(s:reserve(70) == true)
3715 assert(s:remaining() == 0 and s:exhausted() == true)
3716 assert(s:reserve(0) == true, "zero fits even at zero")
3717 assert(s:reserve(1) == false, "nothing fits past zero")
3718 "#,
3719 )
3720 .expect("reserve chunk");
3721
3722 // Without a budget every reservation is granted, and nothing is
3723 // recorded: a session with no quota keeps no ledger.
3724 vm.exec(
3725 r#"
3726 local s = knl.open()
3727 local ok, tag = s:reserve(999999)
3728 assert(ok == true and tag == nil, "no budget must grant everything")
3729 assert(s:len() == 1, "a session with no quota recorded a ledger event")
3730 "#,
3731 )
3732 .expect("no-budget reserve chunk");
3733
3734 let msg = vm.expect_err(r#"knl.open({ budget = { amount = 10 } }):reserve(-1)"#);
3735 assert!(msg.contains("knl: reserve:"), "missing attribution: {msg}");
3736 assert!(msg.contains("non-negative"), "{msg}");
3737
3738 let msg = vm.expect_err(r#"knl.open():reserve("many")"#);
3739 assert!(msg.contains("knl: reserve:"), "missing attribution: {msg}");
3740 }
3741
3742 /// The counter is a cache of the log: folding `granted − reserved −
3743 /// spent` over what Lua can read reproduces `remaining()` exactly.
3744 #[test]
3745 fn the_balance_lua_reads_is_the_fold_of_the_ledger() {
3746 let vm = vm();
3747 vm.exec(
3748 r#"
3749 local function folded(s)
3750 local balance = nil
3751 for _, ev in ipairs(s:events()) do
3752 if ev.kind == "budget_granted" then
3753 balance = (balance or 0) + ev.data.amount
3754 elseif ev.kind == "budget_reserved" or ev.kind == "budget_spent" then
3755 balance = math.max(0, balance - ev.data.amount)
3756 end
3757 end
3758 return balance
3759 end
3760
3761 local s = knl.open({ budget = { amount = 500, tag = "beats" } })
3762 assert(folded(s) == s:remaining())
3763
3764 s:reserve(120)
3765 s:append({ kind = "llm_response",
3766 data = { content = { { type = "text", text = "hi" } },
3767 usage = { input_tokens = 100, output_tokens = 50 } } })
3768 s:spend(30) -- the call overran its estimate
3769 s:reserve(10000) -- refused, and moves nothing
3770 s:spend(0)
3771
3772 assert(s:remaining() == 350, "remaining: " .. tostring(s:remaining()))
3773 assert(folded(s) == s:remaining(), "the fold and the counter disagree")
3774
3775 -- What the call consumed is the other, independent reading, and
3776 -- it is in the log rather than in the ledger: the counts sit on
3777 -- the response, for a query view to sum.
3778 local r = s:events()[4] -- opened, granted, reserved, the response
3779 assert(r.kind == "llm_response", "kind: " .. tostring(r.kind))
3780 assert(r.data.usage.input_tokens == 100 and r.data.usage.output_tokens == 50)
3781 "#,
3782 )
3783 .expect("fold chunk");
3784 }
3785
3786 /// The session's own boundaries are the kernel's: Lua can read them but
3787 /// not write them. Hand-appending either would be claiming an opening
3788 /// the stream never had, or an ending it never reached, so both are
3789 /// refused and the session stays open.
3790 #[test]
3791 fn lua_cannot_append_the_session_boundary_kinds_by_hand() {
3792 let vm = vm();
3793
3794 let msg = vm.expect_err(
3795 r#"
3796 local s = knl.open()
3797 s:append({ kind = "session_closed", data = { reason = "carried over" } })
3798 "#,
3799 );
3800 assert!(msg.contains("knl: append:"), "missing attribution: {msg}");
3801 assert!(msg.contains("kernel only"), "{msg}");
3802 assert!(msg.contains("session_closed"), "{msg}");
3803
3804 vm.exec(
3805 r#"
3806 local s = knl.open({ budget = { amount = 100, tag = "beats" } })
3807 for _, ev in ipairs({
3808 { kind = "session_opened", data = { scope_id = "s", owner = "me" } },
3809 { kind = "session_closed", data = { reason = "carried over" } },
3810 }) do
3811 local ok = pcall(function() s:append(ev) end)
3812 assert(not ok, "a caller wrote " .. ev.kind)
3813 end
3814
3815 -- Still open, and nothing was recorded.
3816 assert(s:len() == 2, "a rejected boundary was recorded: " .. tostring(s:len()))
3817 assert(s:append({ kind = "note" }) == 3, "the refusal ended the session")
3818 s:spend(10)
3819 assert(s:remaining() == 90)
3820
3821 -- Only close writes the boundary, and it writes exactly one.
3822 s:close("done")
3823 assert(kinds_of(s) ==
3824 "session_opened,budget_granted,note,budget_spent,session_closed",
3825 "recorded: " .. kinds_of(s))
3826 local evs = s:events()
3827 assert(evs[5].data.reason == "done",
3828 "reason: " .. tostring(evs[5].data.reason))
3829
3830 local ok = pcall(function() s:append({ kind = "note" }) end)
3831 assert(not ok, "a closed session took a write")
3832 "#,
3833 )
3834 .expect("session boundary kind chunk");
3835 }
3836
3837 /// `knl.new_beat_id()` mints the beat id the shell stamps on its events:
3838 /// a fresh non-empty string every call, needing no session, and ordered
3839 /// by the time it was minted (UUID v7) so a stream's beats sort the way
3840 /// they happened.
3841 #[test]
3842 fn new_beat_id_mints_distinct_time_ordered_ids() {
3843 let vm = vm();
3844 vm.exec(
3845 r#"
3846 local a = knl.new_beat_id()
3847 local b = knl.new_beat_id()
3848 assert(type(a) == "string" and #a > 0, "a beat id must be a non-empty string")
3849 assert(a ~= b, "two beats must be two ids")
3850 assert(a < b, "beat ids must sort in the order they were minted: " .. a .. " " .. b)
3851
3852 -- Version 7: the 13th hex digit of a UUID is the version nibble.
3853 assert(a:sub(15, 15) == "7", "not a v7 uuid: " .. a)
3854
3855 -- It is a module function, not a session method: no session is
3856 -- needed to name a beat.
3857 local s = knl.open()
3858 assert(s.new_beat_id == nil, "the beat id is not the session's to mint")
3859
3860 -- And it is what the kernel accepts as a beat.
3861 s:append({ kind = "llm_response", beat = a,
3862 data = { content = { { type = "text", text = "ok" } },
3863 usage = { input_tokens = 1 } } })
3864 assert(s:events()[2].beat == a, "the minted beat is recorded verbatim")
3865 "#,
3866 )
3867 .expect("new_beat_id chunk");
3868 }
3869
3870 /// `view("tail", { n = k })` returns the last k events verbatim.
3871 #[test]
3872 fn view_tail_returns_the_last_events() {
3873 let vm = vm();
3874 vm.exec(
3875 r#"
3876 local s = knl.open()
3877 for i = 1, 5 do s:append({ kind = "e" .. i }) end
3878
3879 local t = s:view("tail", { n = 2 })
3880 assert(#t == 2, "tail len: " .. tostring(#t))
3881 assert(t[1].kind == "e4" and t[2].kind == "e5")
3882 assert(t[2].seq == 6, "tail keeps the envelope")
3883
3884 assert(#s:view("tail", { n = 99 }) == 6, "n larger than the history")
3885 assert(#s:view("tail", { n = 0 }) == 0)
3886 assert(#s:view("tail") == 6, "n defaults to 20")
3887 "#,
3888 )
3889 .expect("tail view chunk");
3890
3891 // `n` counts events, so a negative one is not a value the fold has to
3892 // have an opinion about: the declared type says whole and unsigned
3893 // and the refusal names the field it was reading.
3894 let msg = vm.expect_err(r#"knl.open():view("tail", { n = -1 })"#);
3895 assert!(msg.contains("knl: view:"), "missing attribution: {msg}");
3896 assert!(msg.contains("opts.n"), "{msg}");
3897
3898 // And an option `tail` does not have is a typo rather than a knob
3899 // that quietly does nothing.
3900 let msg = vm.expect_err(r#"knl.open():view("tail", { count = 2 })"#);
3901 assert!(msg.contains("knl: view:"), "missing attribution: {msg}");
3902 assert!(msg.contains("unknown field `count`"), "{msg}");
3903 }
3904
3905 /// (attribution) The view vocabulary is closed: an unknown name is an
3906 /// error, and so is a non-string name or non-table opts.
3907 #[test]
3908 fn view_rejects_unknown_names_and_bad_arguments() {
3909 let vm = vm();
3910
3911 let msg = vm.expect_err(r#"knl.open():view("dialog")"#);
3912 assert!(msg.contains("knl: view:"), "missing attribution: {msg}");
3913 assert!(msg.contains(r#"unknown view "dialog""#), "{msg}");
3914
3915 // The token account is one of the names the kernel does not have:
3916 // it reads the `data` of an `llm_response`, so it is a query view in
3917 // Lua (`knl.views.usage`) over the published schema.
3918 let msg = vm.expect_err(r#"knl.open():view("usage")"#);
3919 assert!(msg.contains("knl: view:"), "missing attribution: {msg}");
3920 assert!(msg.contains(r#"unknown view "usage""#), "{msg}");
3921
3922 let msg = vm.expect_err(r#"knl.open():view(42)"#);
3923 assert!(msg.contains("knl: view:"), "missing attribution: {msg}");
3924 assert!(msg.contains("name:"), "{msg}");
3925 assert!(msg.contains("expected a string"), "{msg}");
3926
3927 let msg = vm.expect_err(r#"knl.open():view("tail", "n=2")"#);
3928 assert!(msg.contains("knl: view:"), "missing attribution: {msg}");
3929 assert!(msg.contains("opts:"), "{msg}");
3930 assert!(msg.contains("expected table"), "{msg}");
3931 }
3932
3933 /// (I1) A view is a fresh table every call: mutating it cannot reach
3934 /// the history.
3935 #[test]
3936 fn view_returns_a_fresh_table_each_call() {
3937 let vm = vm();
3938 vm.exec(
3939 r#"
3940 local s = knl.open()
3941 s:append({ kind = "msg_user", data = { content = "hi" } })
3942
3943 local t = s:view("tail", { n = 1 })
3944 t[1].kind = "TAMPERED"
3945 t[1].data.content = nil
3946 table.insert(t, { kind = "ghost" })
3947
3948 local again = s:view("tail", { n = 1 })
3949 assert(#again == 1, "tail length changed: " .. tostring(#again))
3950 assert(again[1].kind == "msg_user", "kind changed: " .. tostring(again[1].kind))
3951 assert(again[1].data.content == "hi", "content changed")
3952
3953 -- …and the history itself is untouched by any of it.
3954 assert(s:len() == 2, "len: " .. tostring(s:len()))
3955 assert(s:events()[2].kind == "msg_user", "the record was reachable")
3956 "#,
3957 )
3958 .expect("view copy chunk");
3959 }
3960
3961 /// `store = "mem"` is the in-memory database asked for by name — it opens
3962 /// like any other session; an unknown store string is a `knl: open:`
3963 /// error.
3964 #[test]
3965 fn store_mem_is_asked_for_by_name_and_unknown_stores_are_rejected() {
3966 let vm = vm();
3967 vm.exec(
3968 r#"
3969 local s = knl.open({ store = "mem", owner = "x", budget = { amount = 10, tag = "beats" } })
3970 assert(s:len() == 2, "a mem session opens like any other: session_opened + the grant")
3971 assert(s:owner() == "x")
3972 assert(s:append({ kind = "note" }) == 3)
3973 "#,
3974 )
3975 .expect("mem store chunk");
3976
3977 let msg = vm.expect_err(r#"knl.open({ store = "postgres" })"#);
3978 assert!(msg.contains("knl: open:"), "missing attribution: {msg}");
3979 assert!(msg.contains("unknown store"), "{msg}");
3980
3981 // A table that is not the durable form is refused with both forms
3982 // named, which is the union the declared type states.
3983 let msg = vm.expect_err(r#"knl.open({ store = { redis = "x" } })"#);
3984 assert!(msg.contains("knl: open:"), "missing attribution: {msg}");
3985 assert!(msg.contains("opts.store"), "{msg}");
3986 assert!(msg.contains("sqlite"), "{msg}");
3987 }
3988
3989 /// The parent is the one thing in `opts` that is a handle rather than a
3990 /// value, and it is the *only* one: everything else is read as data, with
3991 /// nothing turned off.
3992 ///
3993 /// The two halves are one decision. Letting a userdata through by telling
3994 /// the deserializer to skip what it cannot represent would also skip a
3995 /// function written where a string belonged — the option would read as
3996 /// absent and the session would open with the default. So `parent` comes
3997 /// off the table by hand and the rest is read strictly.
3998 #[test]
3999 fn the_parent_is_the_only_value_read_as_a_handle() {
4000 let vm = vm();
4001
4002 vm.exec(
4003 r#"
4004 local p = knl.open({ owner = "p", budget = { amount = 10, tag = "beats" } })
4005 local c = knl.open({ owner = "c", parent = p, budget = { from_parent = 4 } })
4006 assert(c:remaining() == 4, "the child's balance: " .. tostring(c:remaining()))
4007 assert(p:remaining() == 6, "the parent paid: " .. tostring(p:remaining()))
4008 "#,
4009 )
4010 .expect("parent chunk");
4011
4012 // A function where a string belonged is refused, naming the field —
4013 // not read as "no owner given".
4014 let msg = vm.expect_err(r#"knl.open({ owner = function() end })"#);
4015 assert!(msg.contains("knl: open:"), "missing attribution: {msg}");
4016 assert!(msg.contains("opts.owner"), "{msg}");
4017
4018 // And `parent` is still only accepted as a session.
4019 let msg = vm.expect_err(r#"knl.open({ parent = "s-1", budget = { from_parent = 1 } })"#);
4020 assert!(msg.contains("knl: open:"), "missing attribution: {msg}");
4021 assert!(msg.contains("must be a session"), "{msg}");
4022 }
4023
4024 /// (Fix 6) The reserved owner ids are the kernel's own namespace: an
4025 /// untrusted Lua caller cannot claim "system" or "anon" (a spoofing hole
4026 /// for the future permission layer), each rejected as a `knl: open:` error.
4027 /// An unspecified owner still defaults to the kernel-assigned "anon", and a
4028 /// real principal id is accepted verbatim.
4029 #[test]
4030 fn open_rejects_reserved_owner_ids_from_the_caller() {
4031 let vm = vm();
4032
4033 let msg = vm.expect_err(r#"knl.open({ owner = "system" })"#);
4034 assert!(msg.contains("knl: open:"), "missing attribution: {msg}");
4035 assert!(msg.contains("reserved"), "{msg}");
4036 assert!(msg.contains("system"), "{msg}");
4037
4038 let msg = vm.expect_err(r#"knl.open({ owner = "anon" })"#);
4039 assert!(msg.contains("knl: open:"), "missing attribution: {msg}");
4040 assert!(msg.contains("reserved"), "{msg}");
4041 assert!(msg.contains("anon"), "{msg}");
4042
4043 vm.exec(
4044 r#"
4045 -- Unspecified owner is the kernel-assigned reserved anon.
4046 assert(knl.open():owner() == "anon", "default owner must be anon")
4047 assert(knl.open({}):owner() == "anon", "empty opts default owner must be anon")
4048 -- A real principal id is accepted verbatim.
4049 assert(knl.open({ owner = "alice" }):owner() == "alice", "owner not carried")
4050 "#,
4051 )
4052 .expect("reserved-owner chunk");
4053 }
4054
4055 /// (scope) A session has a scope: `s:scope_id()` is a real kernel-issued
4056 /// string, it is not the session id, and it is what `session_opened` and
4057 /// every `budget_*` event were written under. Two runs are two scopes.
4058 #[test]
4059 fn a_session_reports_its_scope_id_and_records_it_on_the_log() {
4060 let vm = vm();
4061 vm.exec(
4062 r#"
4063 local s = knl.open({ owner = "alice", budget = { amount = 100, tag = "beats" } })
4064 local scope = s:scope_id()
4065 assert(type(scope) == "string" and #scope > 0, "scope_id must be a non-empty string")
4066 assert(scope ~= s:id(), "the scope names the authority, the id names the stream")
4067
4068 s:reserve(30) -- budget_reserved
4069 s:spend(10) -- budget_spent
4070 s:reserve(10000) -- budget_refused
4071
4072 local seen = 0
4073 for _, e in ipairs(s:events()) do
4074 if e.kind == "session_opened" then
4075 assert(e.data.scope_id == scope,
4076 "session_opened scope_id: " .. tostring(e.data.scope_id))
4077 assert(e.data.owner == "alice", "the owner rides beside it")
4078 seen = seen + 1
4079 elseif e.kind:sub(1, 7) == "budget_" then
4080 assert(e.data.scope_id == scope,
4081 e.kind .. " scope_id: " .. tostring(e.data.scope_id))
4082 seen = seen + 1
4083 end
4084 end
4085 assert(seen == 5,
4086 "session_opened + granted + reserved + spent + refused: " .. tostring(seen))
4087
4088 -- A caller's own event carries no scope id: the field is on the
4089 -- kinds only the kernel writes.
4090 s:append({ kind = "note" })
4091 local evs = s:events()
4092 assert(evs[#evs].data.scope_id == nil,
4093 "a caller's event must not carry a scope id")
4094
4095 assert(knl.open({ owner = "bob" }):scope_id() ~= scope,
4096 "two runs must be two scopes")
4097 "#,
4098 )
4099 .expect("scope id chunk");
4100 }
4101
4102 /// (scope, durable) The scope outlives the process: a reopened stream
4103 /// resumes under the id its `session_opened` recorded — not a fresh one —
4104 /// and the ledger it goes on writing names that same scope.
4105 #[test]
4106 fn a_resumed_session_keeps_the_scope_id_the_log_recorded() {
4107 let vm = vm();
4108 let dir = tempfile::tempdir().expect("tempdir");
4109 let path = dir.path().join("knl.db");
4110 let path = path.to_str().expect("utf-8 path");
4111
4112 vm.exec(&format!(
4113 r#"
4114 local path = "{path}"
4115 local s = knl.open({{ store = {{ sqlite = path }}, owner = "scoped-user",
4116 budget = {{ amount = 100, tag = "beats" }} }})
4117 local id, scope = s:id(), s:scope_id()
4118 assert(scope ~= id, "the scope id is not the stream id")
4119 s:reserve(20)
4120
4121 -- Resumed while the stream is still open: a session is disposable,
4122 -- so a closed one is never reopened.
4123 local r = knl.resume({{ store = {{ sqlite = path }}, session = id }})
4124 assert(r:id() == id, "resumed id is the stream it reopened")
4125 assert(r:scope_id() == scope, "resumed scope: " .. tostring(r:scope_id()))
4126 assert(r:owner() == "scoped-user", "resumed owner: " .. tostring(r:owner()))
4127
4128 r:reserve(5)
4129 local evs = r:events()
4130 local last = evs[#evs]
4131 assert(last.kind == "budget_reserved", "last kind: " .. tostring(last.kind))
4132 assert(last.data.scope_id == scope,
4133 "continued scope_id: " .. tostring(last.data.scope_id))
4134 "#
4135 ))
4136 .expect("durable scope chunk");
4137 }
4138
4139 /// (durable) `knl.open({ store = { sqlite = path } })` writes to a
4140 /// persisted stream, and `knl.resume` reopens it and re-folds the
4141 /// record: the owner, the balance the budget ledger implies and the
4142 /// recorded events all come back, and the resumed session carries on
4143 /// from there. A `budget` on resume is the owner granting again:
4144 /// recorded, and added to what was left.
4145 #[test]
4146 fn open_and_resume_a_durable_sqlite_session() {
4147 let vm = vm();
4148 let dir = tempfile::tempdir().expect("tempdir");
4149 let path = dir.path().join("knl.db");
4150 let path = path.to_str().expect("utf-8 path");
4151
4152 vm.exec(&format!(
4153 r#"
4154 local path = "{path}"
4155 -- The responses in a stream, for the counts a query view sums.
4156 local function responses(s)
4157 local out = {{}}
4158 for _, ev in ipairs(s:events()) do
4159 if ev.kind == "llm_response" then out[#out + 1] = ev end
4160 end
4161 return out
4162 end
4163
4164 local s = knl.open({{ store = {{ sqlite = path }}, owner = "durable-user",
4165 budget = {{ amount = 100, tag = "beats" }} }})
4166 s:reserve(30)
4167 s:append({{ kind = "llm_response",
4168 data = {{ content = {{ {{ type = "text", text = "a" }} }},
4169 usage = {{ input_tokens = 30 }} }} }})
4170 s:append({{ kind = "msg_user", data = {{ content = "more" }} }})
4171 s:reserve(15)
4172 s:append({{ kind = "llm_response",
4173 data = {{ content = {{ {{ type = "text", text = "b" }} }},
4174 usage = {{ input_tokens = 20 }} }} }})
4175 s:spend(5) -- the second call overran its estimate
4176 assert(s:remaining() == 50, "open remaining: " .. tostring(s:remaining()))
4177 local id = s:id()
4178
4179 -- Reopen the same stream and continue where it left off. No new
4180 -- grant: the balance is what the ledger says was left. The stream
4181 -- is still open: a session is disposable, so a closed one is not
4182 -- reopened.
4183 local r = knl.resume({{ store = {{ sqlite = path }}, session = id }})
4184 assert(r:owner() == "durable-user", "resumed owner: " .. tostring(r:owner()))
4185 assert(r:remaining() == 50, "resumed remaining: " .. tostring(r:remaining()))
4186 assert(r:id() == id, "resumed id is the stream it reopened")
4187 -- The record came back whole: the counts are on the responses,
4188 -- where a query view reads them.
4189 local rs = responses(r)
4190 assert(#rs == 2, "resumed responses: " .. tostring(#rs))
4191 assert(rs[1].data.usage.input_tokens == 30
4192 and rs[2].data.usage.input_tokens == 20,
4193 "the counts came back with the record")
4194
4195 -- The grant's words came back too: a refusal still names it.
4196 local ok, tag = r:reserve(1000)
4197 assert(ok == false and tag == "beats", "refused tag: " .. tostring(tag))
4198
4199 -- The record and the ledger continue on the resumed session.
4200 r:reserve(5)
4201 r:append({{ kind = "llm_response", beat = knl.new_beat_id(),
4202 data = {{ content = {{ {{ type = "text", text = "c" }} }},
4203 usage = {{ input_tokens = 5 }} }} }})
4204 assert(#responses(r) == 3, "continued responses: " .. tostring(#responses(r)))
4205 assert(r:remaining() == 45, "continued remaining: " .. tostring(r:remaining()))
4206
4207 -- Granting again on resume adds to what is left, and is recorded.
4208 local g = knl.resume({{ store = {{ sqlite = path }}, session = id,
4209 budget = {{ amount = 100, tag = "beats",
4210 desc = "a second grant" }} }})
4211 assert(g:remaining() == 145, "re-granted remaining: " .. tostring(g:remaining()))
4212 local evs = g:events()
4213 local last = evs[#evs]
4214 assert(last.kind == "budget_granted", "last event: " .. tostring(last.kind))
4215 assert(last.data.amount == 100 and last.data.desc == "a second grant")
4216 "#
4217 ))
4218 .expect("durable open/resume chunk");
4219 }
4220
4221 /// (owner namespace) resume holds the same reserved-principal line as
4222 /// open: a stream the host opened as SYSTEM cannot be reopened from
4223 /// Lua, or an untrusted caller could write into the reserved namespace
4224 /// through the resume side door.
4225 #[test]
4226 fn resume_rejects_a_reserved_system_owned_stream() {
4227 let vm = vm();
4228 let dir = tempfile::tempdir().expect("tempdir");
4229 let path = dir.path().join("knl.db");
4230 let path_str = path.to_str().expect("utf-8 path");
4231
4232 // The host side (Rust) legitimately opens a SYSTEM-owned stream, on
4233 // the same collection of connection threads the VM's sessions use.
4234 let stream = "system-stream".to_string();
4235 let drivers = vm.drivers.clone();
4236 vm.block_on(async {
4237 let store = crate::knl::SqliteEventStore::open(&path, stream.clone(), &drivers)
4238 .await
4239 .expect("open store");
4240 let state =
4241 crate::knl::Session::open_on(crate::knl::SYSTEM.to_string(), None, Box::new(store))
4242 .await
4243 .expect("open system session");
4244 drop(state);
4245 });
4246
4247 // Lua resuming it is refused, exactly as claiming SYSTEM at open is.
4248 let msg = vm.expect_err(&format!(
4249 r#"knl.resume({{ store = {{ sqlite = "{path_str}" }}, session = "{stream}" }})"#
4250 ));
4251 assert!(
4252 msg.contains("reserved"),
4253 "must name the reserved owner: {msg}"
4254 );
4255 }
4256
4257 /// (attribution) resume needs a session id, and a stream that is one:
4258 /// each missing piece is a `knl: resume:` error.
4259 #[test]
4260 fn resume_requires_a_session_id_and_a_stream_that_holds_a_session() {
4261 let vm = vm();
4262
4263 let msg = vm.expect_err(r#"knl.resume()"#);
4264 assert!(msg.contains("knl: resume:"), "missing attribution: {msg}");
4265
4266 let msg = vm.expect_err(r#"knl.resume({ store = { sqlite = "/tmp/x.db" } })"#);
4267 assert!(msg.contains("knl: resume:"), "missing attribution: {msg}");
4268 assert!(msg.contains("missing field `session`"), "{msg}");
4269
4270 // A name nobody is holding open is an empty stream, not a session:
4271 // an in-memory database exists only while a handle does, so resuming
4272 // one that has gone is refused for having no opening in it — the same
4273 // answer a fresh file gives.
4274 let msg = vm.expect_err(r#"knl.resume({ session = "never-opened" })"#);
4275 assert!(msg.contains("knl: resume:"), "missing attribution: {msg}");
4276 assert!(msg.contains("no session to resume"), "{msg}");
4277 }
4278
4279 /// (mem) An in-memory session is a session: it is resumable while it is
4280 /// alive, by the id it reports, and the resumed handle reads the same log
4281 /// and continues the same ledger. What it cannot do is outlive the
4282 /// process, and nothing here pretends otherwise.
4283 ///
4284 /// The store is named on both calls, and that is the point of the last
4285 /// paragraph: an absent one is the host's database, so a `mem` stream is
4286 /// only reachable by asking for `mem`.
4287 #[test]
4288 fn an_in_memory_stream_is_resumable_while_it_is_open() {
4289 let vm = vm();
4290 vm.exec(
4291 r#"
4292 local s = knl.open({ store = "mem", owner = "mem-user",
4293 budget = { amount = 100, tag = "beats" } })
4294 local id = s:id()
4295 s:reserve(30)
4296 s:append({ kind = "note", data = { text = "in memory" } })
4297
4298 -- The writer is still alive, so the database is still there.
4299 local r = knl.resume({ store = "mem", session = id })
4300 assert(r:id() == id, "resumed id: " .. tostring(r:id()))
4301 assert(r:owner() == "mem-user", "resumed owner: " .. tostring(r:owner()))
4302 assert(r:remaining() == 70, "resumed remaining: " .. tostring(r:remaining()))
4303 assert(r:len() == 4, "session_opened + granted + reserved + note")
4304
4305 -- And the resumed handle writes into the same log.
4306 r:spend(20)
4307 assert(s:remaining() == 50, "the writer sees it: " .. tostring(s:remaining()))
4308
4309 -- An absent store means the same thing on resume as it does on
4310 -- open — the host's database — so it does not find this stream:
4311 -- nothing about a `mem` session is in that file.
4312 local missed = failure(knl.resume, { session = id })
4313 assert(missed.kind == "validation", missed.kind)
4314 assert(missed.message:find("no session to resume", 1, true), missed.message)
4315 "#,
4316 )
4317 .expect("in-memory resume chunk");
4318 }
4319
4320 // -- the default store: the file the host owns -------------------------
4321
4322 /// A session opened without a `store` lands in the host's database, and
4323 /// two of them are two streams in that one file.
4324 ///
4325 /// The file is what a project's sessions share — which is what a tree
4326 /// opened from a default parent needs — so both halves are asserted: the
4327 /// path the host handed to [`register`] is the one that exists, and the
4328 /// two sessions are told apart inside it by their stream ids.
4329 #[test]
4330 fn a_session_with_no_store_lands_in_the_hosts_database() {
4331 let vm = vm();
4332 let store = vm.default_store();
4333 assert!(
4334 !store.exists(),
4335 "the file is SQLite's to create, on the first session that needs it"
4336 );
4337
4338 let (first, second): (String, String) = vm
4339 .eval(
4340 r#"
4341 local a = knl.open({ owner = "u" })
4342 local b = knl.open({ owner = "u" })
4343 a:append({ kind = "note", data = { text = "a" } })
4344 b:append({ kind = "note", data = { text = "b" } })
4345 a:close("done")
4346 b:close("done")
4347 return a:id(), b:id()
4348 "#,
4349 )
4350 .expect("two default sessions");
4351 assert_ne!(first, second, "two opens are two streams");
4352
4353 // Held, so the file is still there to be read.
4354 let _dir = vm.finish_keeping_the_store();
4355 assert!(
4356 store.exists(),
4357 "the default store was never created: {}",
4358 store.display()
4359 );
4360 // One file, both logs: each stream is complete, and reading one does
4361 // not turn up the other's events.
4362 for stream in [&first, &second] {
4363 let log = persisted(&store, stream);
4364 let kinds: Vec<&str> = log
4365 .iter()
4366 .map(|e| e["kind"].as_str().expect("a kind"))
4367 .collect();
4368 assert_eq!(
4369 kinds,
4370 ["session_opened", "note", "session_closed"],
4371 "stream {stream}"
4372 );
4373 }
4374 }
4375
4376 /// A session opened without a `store` is resumed by its id alone: the
4377 /// resume looks in the same file the open wrote to.
4378 #[test]
4379 fn a_default_session_resumes_by_id_alone() {
4380 let vm = vm();
4381 vm.exec(
4382 r#"
4383 local s = knl.open({ owner = "u", budget = { amount = 100, tag = "beats" } })
4384 local id = s:id()
4385 s:reserve(30)
4386 s:append({ kind = "note", data = { text = "recorded" } })
4387
4388 local r = knl.resume({ session = id })
4389 assert(r:id() == id, "resumed id: " .. tostring(r:id()))
4390 assert(r:owner() == "u", "resumed owner: " .. tostring(r:owner()))
4391 assert(r:remaining() == 70, "the ledger came back: " .. tostring(r:remaining()))
4392 assert(r:len() == 4, "session_opened + granted + reserved + note")
4393 "#,
4394 )
4395 .expect("resume by id alone");
4396 vm.finish();
4397 }
4398
4399 /// A tree needs a file store, so a parent on `"mem"` is refused — with a
4400 /// message that says which store it is and what to open instead.
4401 ///
4402 /// The refusal is the whole answer: an in-memory database locks per table
4403 /// under its shared cache, and the alternative to refusing would be
4404 /// teaching the kernel to wait on that lock.
4405 #[test]
4406 fn a_child_of_a_mem_parent_is_refused() {
4407 let vm = vm();
4408 vm.exec(
4409 r#"
4410 local parent = knl.open({ store = "mem", owner = "u",
4411 budget = { amount = 100, tag = "tokens" } })
4412 local refused = failure(knl.open, {
4413 owner = "w", parent = parent, budget = { from_parent = 10 },
4414 })
4415 assert(refused.kind == "validation", refused.kind)
4416 assert(refused.message:find("a session tree needs a file store", 1, true),
4417 refused.message)
4418 assert(refused.message:find("mem", 1, true), refused.message)
4419
4420 -- Nothing was opened and nothing moved.
4421 assert(parent:remaining() == 100, tostring(parent:remaining()))
4422
4423 -- The same parent on the host's database takes a child.
4424 local ok_parent = knl.open({ owner = "u", budget = { amount = 100, tag = "tokens" } })
4425 local child = knl.open({ owner = "w", parent = ok_parent,
4426 budget = { from_parent = 10 } })
4427 assert(ok_parent:remaining() == 90, tostring(ok_parent:remaining()))
4428 child:close("done")
4429 ok_parent:close("done")
4430 parent:close("done")
4431 "#,
4432 )
4433 .expect("the refusal");
4434 vm.finish();
4435 }
4436
4437 // -- session lifecycle: `<close>` and the drop backstop ----------------
4438 //
4439 // Every one of these reads the boundary back out of a *reopened* SQLite
4440 // stream rather than off the handle that wrote it: the question is
4441 // whether the record landed, and only the durable log answers that.
4442
4443 /// The persisted events of `stream`, read through a fresh connection.
4444 ///
4445 /// A runtime of its own, and a collection of its own that is drained
4446 /// before the rows are handed back: this is a plain read, and it should
4447 /// leave nothing running behind it.
4448 fn persisted(path: &std::path::Path, stream: &str) -> Vec<Value> {
4449 use crate::knl::EventStore;
4450
4451 let rt = tokio::runtime::Builder::new_current_thread()
4452 .enable_all()
4453 .build()
4454 .expect("a runtime to read on");
4455 rt.block_on(async {
4456 let drivers = knl::IsleDrivers::new();
4457 let store = crate::knl::SqliteEventStore::open(path, stream, &drivers)
4458 .await
4459 .expect("reopen the stream");
4460 let log = store.read(0, usize::MAX).await.expect("read the stream");
4461 drop(store);
4462 assert!(drivers.shutdown().await.is_empty(), "the reader joined");
4463 log
4464 })
4465 }
4466
4467 /// Run `chunk` in a fresh VM and return the session id it yields.
4468 ///
4469 /// The VM is dropped *and its connection threads drained* before the
4470 /// caller reads the stream: collecting the Lua state is what makes the
4471 /// drop backstop submit its boundary, and draining the threads is what
4472 /// waits for that submitted write to land. Only then is the log
4473 /// inspected.
4474 fn stream_id_from(chunk: String) -> String {
4475 let vm = vm();
4476 let id = vm.eval::<String>(&chunk).expect("close scope chunk");
4477 vm.finish();
4478 id
4479 }
4480
4481 /// (I6) A `<close>` scope that ends cleanly records the session's
4482 /// boundary with `scope_exit`: the shell no longer has to remember to
4483 /// close.
4484 #[test]
4485 fn a_close_scope_records_the_boundary_on_the_way_out() {
4486 let dir = tempfile::tempdir().expect("tempdir");
4487 let path = dir.path().join("knl.db");
4488 let path_str = path.to_str().expect("utf-8 path");
4489
4490 let id = stream_id_from(format!(
4491 r#"
4492 local id
4493 do
4494 local s <close> = knl.open({{ store = {{ sqlite = "{path_str}" }}, owner = "t" }})
4495 id = s:id()
4496 s:append({{ kind = "note" }})
4497 assert(s:len() == 2, "inside the scope: session_opened + note")
4498 end
4499 return id
4500 "#
4501 ));
4502
4503 let log = persisted(&path, &id);
4504 let last = log.last().expect("the stream is not empty");
4505 assert_eq!(last["kind"], Value::from("session_closed"), "{last}");
4506 assert_eq!(last["data"]["reason"], Value::from("scope_exit"), "{last}");
4507 assert_eq!(
4508 last["data"].get("detail"),
4509 None,
4510 "a clean exit has nothing to say"
4511 );
4512 assert_eq!(log.len(), 3, "session_opened + note + session_closed");
4513 }
4514
4515 /// (I6) A block that raises closes its session too, with `error` as the
4516 /// reason and the message as `detail` — so the log says the session
4517 /// ended badly without the reason vocabulary growing a member per
4518 /// failure.
4519 #[test]
4520 fn a_close_scope_that_raises_records_the_error_and_its_message() {
4521 let dir = tempfile::tempdir().expect("tempdir");
4522 let path = dir.path().join("knl.db");
4523 let path_str = path.to_str().expect("utf-8 path");
4524
4525 let id = stream_id_from(format!(
4526 r#"
4527 local id
4528 local ok, msg = pcall(function()
4529 local s <close> = knl.open({{ store = {{ sqlite = "{path_str}" }}, owner = "t" }})
4530 id = s:id()
4531 error("boom")
4532 end)
4533 assert(not ok, "the block was supposed to fail")
4534 assert(tostring(msg):find("boom"), "the error is still the caller's: " .. tostring(msg))
4535 return id
4536 "#
4537 ));
4538
4539 let log = persisted(&path, &id);
4540 let last = log.last().expect("the stream is not empty");
4541 assert_eq!(last["kind"], Value::from("session_closed"), "{last}");
4542 assert_eq!(last["data"]["reason"], Value::from("error"), "{last}");
4543 let detail = last["data"]["detail"].as_str().expect("detail text");
4544 assert!(detail.contains("boom"), "detail: {detail}");
4545 }
4546
4547 // -- a store that fails, so a failed close can be driven from Lua -------
4548 //
4549 // An append is serialized and lands, so a close can no longer be made to
4550 // fail by racing another handle. What is left is a backend that reports
4551 // a failure, which is a real thing a durable store does (a database gone,
4552 // or contended past its retries). The store below is the smallest honest
4553 // stand-in: it fails the append at a chosen position and no other.
4554
4555 /// A [`knl::EventStore`] that fails its `nth` append and serves the rest
4556 /// from an in-memory log the test keeps a handle on.
4557 /// The shared log a [`FlakyStore`] writes to.
4558 ///
4559 /// `Arc<tokio::sync::Mutex<_>>` rather than the `Rc<RefCell<_>>` it used
4560 /// to be: an [`knl::EventStore`] is `Send + Sync` now (the durable one's
4561 /// calls travel to a connection thread), and the SPI is `async`, so the
4562 /// lock has to be one that may be held across a suspension point.
4563 type SharedLog = std::sync::Arc<Mutex<knl::MemEventStore>>;
4564
4565 struct FlakyStore {
4566 /// The real log, shared with the test so it can be read after the
4567 /// session that owned it is gone.
4568 inner: SharedLog,
4569 /// Which append (1-based) fails; `0` fails none.
4570 fails_on: usize,
4571 /// How many appends have been attempted.
4572 attempts: std::sync::atomic::AtomicUsize,
4573 }
4574
4575 impl FlakyStore {
4576 /// A store whose `fails_on`-th append reports a failure, plus the
4577 /// handle on the log it writes to.
4578 fn new(fails_on: usize) -> (Self, SharedLog) {
4579 let inner: SharedLog = std::sync::Arc::default();
4580 let store = Self {
4581 inner: std::sync::Arc::clone(&inner),
4582 fails_on,
4583 attempts: std::sync::atomic::AtomicUsize::new(0),
4584 };
4585 (store, inner)
4586 }
4587
4588 /// Whether this attempt is the one that fails.
4589 fn fails_now(&self) -> bool {
4590 let attempt = self
4591 .attempts
4592 .fetch_add(1, std::sync::atomic::Ordering::Relaxed)
4593 + 1;
4594 attempt == self.fails_on
4595 }
4596
4597 /// The log, borrowed.
4598 async fn log(&self) -> tokio::sync::MutexGuard<'_, knl::MemEventStore> {
4599 self.inner.lock().await
4600 }
4601 }
4602
4603 #[async_trait::async_trait]
4604 impl knl::EventStore for FlakyStore {
4605 async fn append(&mut self, event: Map<String, Value>) -> knl::KnlResult<knl::Committed> {
4606 if self.fails_now() {
4607 return Err(knl::KnlError::Storage("the store is down".to_string()));
4608 }
4609 self.log().await.append(event).await
4610 }
4611
4612 /// A batch is *one* write, as it is on the durable backend: it counts
4613 /// as one attempt, and when that attempt is the failing one nothing
4614 /// in the batch is recorded. A stand-in that let half a batch land
4615 /// would be modelling a store the SPI does not allow.
4616 async fn append_many(
4617 &mut self,
4618 events: Vec<Map<String, Value>>,
4619 ) -> knl::KnlResult<Vec<knl::Committed>> {
4620 if self.fails_now() {
4621 return Err(knl::KnlError::Storage("the store is down".to_string()));
4622 }
4623 let mut log = self.log().await;
4624 let mut committed = Vec::with_capacity(events.len());
4625 for event in events {
4626 committed.push(log.append(event).await?);
4627 }
4628 Ok(committed)
4629 }
4630
4631 async fn append_if(
4632 &mut self,
4633 kinds: Option<&[&str]>,
4634 decide: knl::Decision,
4635 ) -> knl::KnlResult<Option<knl::Committed>> {
4636 if self.fails_now() {
4637 return Err(knl::KnlError::Storage("the store is down".to_string()));
4638 }
4639 self.log().await.append_if(kinds, decide).await
4640 }
4641
4642 async fn read_kinds(
4643 &self,
4644 kinds: Option<&[&str]>,
4645 from_seq: u64,
4646 limit: usize,
4647 ) -> knl::KnlResult<Vec<Value>> {
4648 self.log().await.read_kinds(kinds, from_seq, limit).await
4649 }
4650
4651 async fn head(&self) -> knl::KnlResult<Option<u64>> {
4652 self.log().await.head().await
4653 }
4654
4655 async fn len(&self) -> knl::KnlResult<usize> {
4656 self.log().await.len().await
4657 }
4658 }
4659
4660 /// The kinds an in-memory log holds, in order.
4661 fn kinds_in(log: &SharedLog) -> Vec<String> {
4662 use crate::knl::EventStore;
4663
4664 let rt = tokio::runtime::Builder::new_current_thread()
4665 .build()
4666 .expect("a runtime to read on");
4667 rt.block_on(async { log.lock().await.read(0, usize::MAX).await })
4668 .expect("read the log")
4669 .iter()
4670 .map(|e| e["kind"].as_str().unwrap_or("").to_string())
4671 .collect()
4672 }
4673
4674 /// A VM where `open_failing(n)` opens a session on a [`FlakyStore`] whose
4675 /// `n`-th append fails, and the log it writes to.
4676 ///
4677 /// The hook is test-only and lives here rather than in `register`: the
4678 /// Lua surface a caller sees is [`MODULE_API`] and nothing else. It
4679 /// builds the same userdata `knl.open` builds, so `<close>`, the drop
4680 /// backstop and every method behave exactly as they do in production.
4681 fn vm_with_a_failing_store(fails_on: usize) -> (Vm, SharedLog) {
4682 let vm = vm();
4683 let (store, log) = FlakyStore::new(fails_on);
4684 // Handed over once, from inside an async function like `knl.open`
4685 // itself: opening a session is a write, so it suspends.
4686 let store = std::sync::Arc::new(Mutex::new(Some(store)));
4687 let open_failing =
4688 vm.lua
4689 .create_async_function(move |lua, ()| {
4690 let store = std::sync::Arc::clone(&store);
4691 async move {
4692 let store = store.lock().await.take().ok_or_else(|| {
4693 err("open", "the failing store can only be opened once")
4694 })?;
4695 let state = knl::Session::open_on("t".to_string(), None, Box::new(store))
4696 .await
4697 .map_err(|e| knl_err("open", &e))?;
4698 lua.create_userdata(Session::from_state(state))
4699 }
4700 })
4701 .expect("create open_failing");
4702 vm.lua
4703 .globals()
4704 .set("open_failing", open_failing)
4705 .expect("register open_failing");
4706 (vm, log)
4707 }
4708
4709 /// (I6) The block's own error wins over a close that could not be
4710 /// recorded.
4711 ///
4712 /// The store fails the `session_closed` append, so `__close` has a real
4713 /// failure to report while an error is already on its way out of the
4714 /// block. Lua would let `__close` replace that error; it must not,
4715 /// because the bookkeeping failure is not what the caller is trying to
4716 /// diagnose. It goes to the tracing log instead.
4717 #[test]
4718 fn a_failed_close_does_not_replace_the_error_the_block_raised() {
4719 // 1: session_opened (no grant, so the close is the second append).
4720 let (vm, log) = vm_with_a_failing_store(2);
4721
4722 vm.exec(
4723 r#"
4724 local kept
4725 local ok, msg = pcall(function()
4726 local s <close> = open_failing()
4727 kept = s
4728 error("boom")
4729 end)
4730 assert(not ok, "the block was supposed to fail")
4731 assert(tostring(msg):find("boom"),
4732 "the close replaced the block's error: " .. tostring(msg))
4733 assert(not tostring(msg):find("the store is down"),
4734 "the close's own failure surfaced instead: " .. tostring(msg))
4735 -- The session stayed open: the boundary was not recorded, and the
4736 -- handle says so rather than pretending otherwise.
4737 assert(kept:len() == 1, "len after the failed close: " .. tostring(kept:len()))
4738 "#,
4739 )
4740 .expect("failing close chunk");
4741
4742 assert_eq!(
4743 kinds_in(&log),
4744 ["session_opened"],
4745 "the boundary really was not recorded"
4746 );
4747 }
4748
4749 /// (I6) A clean scope exit keeps raising when the boundary cannot be
4750 /// recorded: there is no body error to preserve, so silence would be a
4751 /// close reporting success with nothing in the log.
4752 #[test]
4753 fn a_failed_close_on_a_clean_scope_exit_still_raises() {
4754 let (vm, log) = vm_with_a_failing_store(2);
4755
4756 let msg = vm.expect_err(
4757 r#"
4758 do
4759 local s <close> = open_failing()
4760 end
4761 "#,
4762 );
4763 assert!(msg.contains("knl: close:"), "missing attribution: {msg}");
4764 assert!(msg.contains("the store is down"), "{msg}");
4765
4766 // Nothing but the opening is in the log: the raise is the only thing
4767 // that says the session ended, which is why it must be raised.
4768 assert_eq!(kinds_in(&log), ["session_opened"]);
4769 }
4770
4771 /// (F4) An open that cannot be recorded leaves the stream *empty*.
4772 ///
4773 /// The opening and the grant are one write now (`append_many`), so there
4774 /// is no window where a reader could see a session that began without the
4775 /// quota it began under — and nothing to close on the way out either.
4776 /// This replaces the earlier behaviour, where the two were separate
4777 /// appends and a failed second one had to be patched over with a
4778 /// best-effort `session_closed`.
4779 #[tokio::test]
4780 async fn an_open_that_cannot_be_recorded_leaves_the_stream_empty() {
4781 use crate::knl::EventStore;
4782
4783 // The whole opening is one write, so it is the first attempt.
4784 let (store, log) = FlakyStore::new(1);
4785 let err = knl::Session::open_on(
4786 "t".to_string(),
4787 Some(knl::BudgetGrant::new(100)),
4788 Box::new(store),
4789 )
4790 .await
4791 .expect_err("the open must fail");
4792 assert_eq!(err.reason(), "the store is down");
4793
4794 let recorded = log.lock().await.read(0, usize::MAX).await.expect("read");
4795 assert!(
4796 recorded.is_empty(),
4797 "a failed open records nothing at all: {recorded:?}"
4798 );
4799 }
4800
4801 /// The other side of it: an open that *does* land records both events, in
4802 /// order, from the one write.
4803 #[tokio::test]
4804 async fn an_open_records_its_boundary_and_its_grant_together() {
4805 use crate::knl::{event::kind_of, EventStore};
4806
4807 let (store, log) = FlakyStore::new(0);
4808 let session = knl::Session::open_on(
4809 "t".to_string(),
4810 Some(knl::BudgetGrant::new(100)),
4811 Box::new(store),
4812 )
4813 .await
4814 .expect("the open lands");
4815 let recorded = log.lock().await.read(0, usize::MAX).await.expect("read");
4816 let kinds: Vec<&str> = recorded.iter().map(kind_of).collect();
4817 assert_eq!(kinds, ["session_opened", "budget_granted"]);
4818 assert_eq!(session.remaining().await, Ok(Some(100)));
4819 }
4820
4821 /// `close(reason, detail)` records both: the reason stays the short word
4822 /// a reader folds on, and the sentence only this close can tell goes to
4823 /// `detail` — which is what lets a Lua-side bracket record the message of
4824 /// the error its body raised. `close(reason)` and `close()` are
4825 /// unchanged.
4826 #[test]
4827 fn close_records_an_optional_detail_beside_the_reason() {
4828 let dir = tempfile::tempdir().expect("tempdir");
4829 let path = dir.path().join("knl.db");
4830 let path_str = path.to_str().expect("utf-8 path");
4831
4832 let id = stream_id_from(format!(
4833 r#"
4834 local id
4835 do
4836 local s = knl.open({{ store = {{ sqlite = "{path_str}" }}, owner = "t" }})
4837 id = s:id()
4838 s:close("error", "the body raised: boom")
4839 end
4840 return id
4841 "#
4842 ));
4843
4844 let log = persisted(&path, &id);
4845 let last = log.last().expect("the stream is not empty");
4846 assert_eq!(last["kind"], Value::from("session_closed"), "{last}");
4847 assert_eq!(last["data"]["reason"], Value::from("error"), "{last}");
4848 assert_eq!(
4849 last["data"]["detail"],
4850 Value::from("the body raised: boom"),
4851 "{last}"
4852 );
4853
4854 // The one- and no-argument forms still work, and a detail is never
4855 // invented for them.
4856 let vm = vm();
4857 vm.exec(
4858 r#"
4859 local a = knl.open({ owner = "t" })
4860 a:close("done")
4861 local last = a:events()[a:len()].data
4862 assert(last.reason == "done", "reason: " .. tostring(last.reason))
4863 assert(last.detail == nil, "a close with no detail must record none")
4864
4865 local b = knl.open({ owner = "t" })
4866 b:close()
4867 local closed = b:events()[b:len()].data
4868 assert(closed.reason == "closed", "default reason: " .. tostring(closed.reason))
4869 assert(closed.detail == nil)
4870 "#,
4871 )
4872 .expect("close forms chunk");
4873
4874 // And a non-string detail is refused, naming which argument it was.
4875 let msg = vm.expect_err(r#"knl.open({ owner = "t" }):close("error", 7)"#);
4876 assert!(msg.contains("knl: close:"), "missing attribution: {msg}");
4877 assert!(msg.contains("detail:"), "{msg}");
4878 assert!(msg.contains("expected a string"), "{msg}");
4879 }
4880
4881 /// A long `detail` is cut to the cap, exactly as the `<close>` path cuts
4882 /// the message of a raised error: one bad turn must not put a page into
4883 /// the log, whichever side records it.
4884 #[test]
4885 fn a_long_close_detail_is_truncated() {
4886 let vm = vm();
4887 vm.exec(
4888 r#"
4889 local s = knl.open({ owner = "t" })
4890 s:close("error", string.rep("x", 500))
4891 local last = s:events()[s:len()].data
4892 assert(#last.detail == 203, "detail length: " .. tostring(#last.detail))
4893 assert(last.detail:sub(-3) == "...", "a cut detail says it was cut")
4894 "#,
4895 )
4896 .expect("long detail chunk");
4897 }
4898
4899 /// (disposable) A closed stream is not reopened. The session ended; what
4900 /// comes after an ending is a new session, and `knl.resume` says so
4901 /// instead of handing back a handle onto a finished log.
4902 #[test]
4903 fn resume_refuses_a_closed_stream() {
4904 let vm = vm();
4905 let dir = tempfile::tempdir().expect("tempdir");
4906 let path = dir.path().join("knl.db");
4907 let path_str = path.to_str().expect("utf-8 path");
4908
4909 let msg = vm.expect_err(&format!(
4910 r#"
4911 local s = knl.open({{ store = {{ sqlite = "{path_str}" }}, owner = "t" }})
4912 local id = s:id()
4913 s:close("done")
4914 knl.resume({{ store = {{ sqlite = "{path_str}" }}, session = id }})
4915 "#
4916 ));
4917 assert!(msg.contains("knl: resume:"), "missing attribution: {msg}");
4918 assert!(msg.contains("session is closed"), "{msg}");
4919 assert!(msg.contains("disposable"), "{msg}");
4920 }
4921
4922 /// (F3) A resume that is refused writes nothing. The reserved-owner
4923 /// check runs before any append, so a caller cannot leave a
4924 /// `budget_granted` in a stream it was not allowed to reopen.
4925 #[test]
4926 fn a_refused_resume_records_no_grant() {
4927 let vm = vm();
4928 let dir = tempfile::tempdir().expect("tempdir");
4929 let path = dir.path().join("knl.db");
4930 let path_str = path.to_str().expect("utf-8 path");
4931
4932 // The host side legitimately opens a SYSTEM-owned stream.
4933 let stream = "system-grant-stream".to_string();
4934 let drivers = vm.drivers.clone();
4935 vm.block_on(async {
4936 let store = crate::knl::SqliteEventStore::open(&path, stream.clone(), &drivers)
4937 .await
4938 .expect("open store");
4939 let state =
4940 crate::knl::Session::open_on(crate::knl::SYSTEM.to_string(), None, Box::new(store))
4941 .await
4942 .expect("open system session");
4943 drop(state);
4944 });
4945 let before = persisted(&path, &stream).len();
4946
4947 let msg = vm.expect_err(&format!(
4948 r#"knl.resume({{ store = {{ sqlite = "{path_str}" }}, session = "{stream}",
4949 budget = {{ amount = 100, tag = "beats" }} }})"#
4950 ));
4951 assert!(msg.contains("reserved"), "{msg}");
4952
4953 let log = persisted(&path, &stream);
4954 assert!(
4955 !log.iter().any(|e| e["kind"] == "budget_granted"),
4956 "a refused resume wrote its grant anyway: {log:?}"
4957 );
4958 assert_eq!(log.len(), before, "a refused resume wrote nothing at all");
4959 }
4960
4961 /// The Lua surface is exactly what is declared: the methods a session
4962 /// answers to are [`SESSION_API`], the functions on the `knl` global are
4963 /// [`MODULE_API`], and `knl.api()` reports both. A method registered
4964 /// without an entry in the table fails here.
4965 #[test]
4966 fn the_lua_surface_is_exactly_what_is_declared() {
4967 let vm = vm();
4968
4969 // The session's methods, read off the live userdata's metatable.
4970 let mut declared: Vec<&str> = SESSION_API
4971 .iter()
4972 .map(|(name, _)| *name)
4973 .filter(|name| *name != "__close")
4974 .collect();
4975 declared.sort_unstable();
4976
4977 // Read off the live userdata's metatable. Lua cannot reach it (mlua
4978 // protects it with `__metatable`), so the reflection is done from
4979 // here — it is still the registration itself that is being read, not
4980 // a second list of names.
4981 let session: LuaAnyUserData = vm
4982 .eval(r#"return knl.open({ owner = "t" })"#)
4983 .expect("open a session to reflect over");
4984 let meta = session.metatable().expect("the session's metatable");
4985 let index: LuaTable = meta.get("__index").expect("the methods table");
4986 let mut reflected: Vec<String> = index
4987 .pairs::<String, LuaValue>()
4988 .map(|pair| pair.expect("a method entry").0)
4989 .collect();
4990 reflected.sort();
4991 assert_eq!(reflected, declared, "the session surface is SESSION_API");
4992
4993 // The `<close>` metamethod is on the metatable itself, not in
4994 // `__index`, and it is declared too.
4995 assert!(
4996 SESSION_API.iter().any(|(name, _)| *name == "__close"),
4997 "the scope boundary belongs to the declared surface"
4998 );
4999 assert!(
5000 !matches!(
5001 meta.get::<LuaValue>("__close").expect("read __close"),
5002 LuaValue::Nil
5003 ),
5004 "a session must carry the <close> metamethod"
5005 );
5006
5007 // The module's functions.
5008 let mut module: Vec<&str> = MODULE_API.iter().map(|(name, _)| *name).collect();
5009 module.sort_unstable();
5010 let mut bound: Vec<String> = vm
5011 .eval::<Vec<String>>(
5012 r#"
5013 local names = {}
5014 for name, value in pairs(knl) do
5015 if type(value) == "function" then table.insert(names, name) end
5016 end
5017 return names
5018 "#,
5019 )
5020 .expect("reflect over the knl global");
5021 bound.sort();
5022 assert_eq!(bound, module, "the module surface is MODULE_API");
5023
5024 // And `knl.api()` hands the same two lists to Lua, each entry with
5025 // the name and its one-line contract.
5026 vm.exec(
5027 r#"
5028 local api = knl.api()
5029 assert(#api.session > 0 and #api.module > 0, "api() must list both halves")
5030 for _, half in ipairs({ api.session, api.module }) do
5031 for _, entry in ipairs(half) do
5032 assert(type(entry.name) == "string" and #entry.name > 0, "an entry needs a name")
5033 assert(type(entry.doc) == "string" and #entry.doc > 0, "an entry needs a doc")
5034 end
5035 end
5036 assert(api.session[1].name == "id", "first: " .. tostring(api.session[1].name))
5037 "#,
5038 )
5039 .expect("api() chunk");
5040
5041 let counted: usize = vm
5042 .eval(r#"local a = knl.api() return #a.session + #a.module"#)
5043 .expect("count the api entries");
5044 assert_eq!(counted, SESSION_API.len() + MODULE_API.len());
5045 }
5046
5047 /// A raised failure carries its class, and `knl.error` hands it back as
5048 /// a table: what a caller branches on is a word from a closed list, not
5049 /// a sentence that is free to be reworded.
5050 #[test]
5051 fn a_raised_failure_reports_its_class_through_knl_error() {
5052 let vm = vm();
5053 vm.exec(
5054 r#"
5055 -- A closed handle refusing its own write. The session is over,
5056 -- and asking again is not what fixes that.
5057 local s = knl.open({ owner = "t" })
5058 s:close()
5059 local e = failure(function() s:append({ kind = "note" }) end)
5060 assert(e.kind == "closed", "kind: " .. tostring(e.kind))
5061 assert(e.method == "append", "method: " .. tostring(e.method))
5062 assert(e.retryable == false, "a closed session is not a retry")
5063 assert(e.message == "session is closed", "message: " .. tostring(e.message))
5064
5065 local t = knl.open({ owner = "t", budget = { amount = 10 } })
5066
5067 -- A kernel-only kind: the caller asked for something the kernel
5068 -- will not record from it.
5069 local k = failure(function() t:append({ kind = "budget_granted", amount = 1 }) end)
5070 assert(k.kind == "validation", "kind: " .. tostring(k.kind))
5071 assert(k.method == "append", "method: " .. tostring(k.method))
5072 assert(k.retryable == false)
5073
5074 -- A negative reserve, refused before anything moves.
5075 local n = failure(function() t:reserve(-1) end)
5076 assert(n.kind == "validation", "kind: " .. tostring(n.kind))
5077 assert(n.method == "reserve", "method: " .. tostring(n.method))
5078
5079 -- An unknown view: the kernel's own validator, same class.
5080 local v = failure(function() t:view("nope") end)
5081 assert(v.kind == "validation", "kind: " .. tostring(v.kind))
5082 assert(v.method == "view", "method: " .. tostring(v.method))
5083
5084 -- A refusal raised on the bridge side, before the kernel is
5085 -- reached, is the same class: one vocabulary either way.
5086 local b = failure(function() t:append(7) end)
5087 assert(b.kind == "validation", "kind: " .. tostring(b.kind))
5088 assert(b.method == "append", "method: " .. tostring(b.method))
5089 "#,
5090 )
5091 .expect("classified failures chunk");
5092 }
5093
5094 /// The class did not cost the message. A caller that only prints, or
5095 /// searches the text it caught, reads exactly what it read before — and
5096 /// the table stands in for the raised value wherever one was.
5097 #[test]
5098 fn a_classified_failure_still_reads_as_a_message() {
5099 let vm = vm();
5100 vm.exec(
5101 r#"
5102 local s = knl.open({ owner = "t" })
5103 s:close()
5104 local e, raised = failure(function() s:append({ kind = "note" }) end)
5105
5106 local text = tostring(raised)
5107 assert(text:find("knl: append:", 1, true), "attribution: " .. text)
5108 assert(text:find("session is closed", 1, true), "reason: " .. text)
5109 assert(tostring(e) == text, "the table must render as its message")
5110
5111 -- A raise that did not come from the kernel is reported whole
5112 -- rather than raising a second failure inside the handler.
5113 local other = knl.error("something else entirely")
5114 assert(other.kind == nil, "kind: " .. tostring(other.kind))
5115 assert(other.method == nil, "method: " .. tostring(other.method))
5116 assert(other.retryable == false)
5117 assert(other.message == "something else entirely",
5118 "message: " .. tostring(other.message))
5119
5120 -- …including one that merely looks like the shape. Only a class
5121 -- the kernel publishes is read as one.
5122 local fake = knl.error("knl: append: nonsense: hello")
5123 assert(fake.kind == nil, "kind: " .. tostring(fake.kind))
5124 assert(fake.message == "knl: append: nonsense: hello")
5125 "#,
5126 )
5127 .expect("message compatibility chunk");
5128 }
5129
5130 /// `knl.api().errors` is the kernel's class list itself, so the shell can
5131 /// hold its own declaration of the vocabulary against it instead of
5132 /// against a list somebody retyped.
5133 #[test]
5134 fn api_publishes_the_error_vocabulary() {
5135 let vm = vm();
5136 let published: Vec<String> = vm
5137 .eval(r#"return knl.api().errors"#)
5138 .expect("read knl.api().errors");
5139 let declared: Vec<String> = knl::KnlError::KINDS
5140 .iter()
5141 .map(|kind| (*kind).to_string())
5142 .collect();
5143 assert_eq!(published, declared);
5144
5145 // And every class a method's doc names is one of them, so the two
5146 // halves of the declaration cannot drift apart.
5147 for (name, doc) in SESSION_API.iter().chain(MODULE_API.iter()) {
5148 let Some((_, raises)) = doc.split_once("[raises: ") else {
5149 continue;
5150 };
5151 let raises = raises.split(']').next().unwrap_or("");
5152 for kind in raises.split(',').map(str::trim).filter(|k| !k.is_empty()) {
5153 // A doc may add a clause after the list ("— only on a clean
5154 // exit"); the class is the first word of the entry.
5155 let kind = kind.split_whitespace().next().unwrap_or("");
5156 assert!(
5157 knl::KnlError::KINDS.contains(&kind),
5158 "{name} names a class the kernel does not publish: {kind:?}"
5159 );
5160 }
5161 }
5162 }
5163
5164 /// The declared surface is built once and answered from there.
5165 ///
5166 /// `knl.api()` used to open an in-memory SQLite database, create the
5167 /// events table in it, run `PRAGMA table_info` and re-render every
5168 /// declared type into Lua source — on every call, for an answer that is a
5169 /// pure function of types fixed at compile time. What a caller gets must
5170 /// still be its own table (a Lua value belongs to one VM, and a caller may
5171 /// write to what it is handed), so this holds two things: the values are
5172 /// equal, and the second call did not rebuild.
5173 ///
5174 /// Counted rather than timed, and read *after* a first call rather than
5175 /// from zero: these statics are per-process, so another test in this
5176 /// binary may have primed them already.
5177 #[test]
5178 fn the_declared_surface_is_built_once_and_the_table_is_fresh() {
5179 use std::sync::atomic::Ordering;
5180
5181 let vm = vm();
5182 let first: Vec<String> = vm
5183 .eval(r#"local a = knl.api() return { a.types, a.schema.table }"#)
5184 .expect("the first api() call");
5185 let (api_builds, type_builds) = (
5186 API_BUILDS.load(Ordering::Relaxed),
5187 TYPES_BUILDS.load(Ordering::Relaxed),
5188 );
5189
5190 let second: Vec<String> = vm
5191 .eval(r#"local a = knl.api() return { a.types, a.schema.table }"#)
5192 .expect("the second api() call");
5193 assert_eq!(first, second, "two calls answer the same surface");
5194 assert_eq!(
5195 API_BUILDS.load(Ordering::Relaxed),
5196 api_builds,
5197 "the report was rebuilt on the second call"
5198 );
5199 assert_eq!(
5200 TYPES_BUILDS.load(Ordering::Relaxed),
5201 type_builds,
5202 "the types module was re-rendered on the second call"
5203 );
5204 assert!(
5205 !first[0].is_empty() && first[1] == knl::EVENTS_TABLE,
5206 "and the answer is the real one: {first:?}"
5207 );
5208
5209 // A caller writing to what it was handed does not reach the next
5210 // caller: the cache is the value, and the table is made per call.
5211 vm.exec(
5212 r#"
5213 local a = knl.api()
5214 a.schema.table = "scribbled"
5215 assert(knl.api().schema.table ~= "scribbled", "api() handed out a shared table")
5216 "#,
5217 )
5218 .expect("api table freshness chunk");
5219 }
5220
5221 /// A backend that is down surfaces as `storage`, not as the caller
5222 /// having done something wrong: the arguments were fine and the store
5223 /// could not do the work.
5224 #[test]
5225 fn a_store_that_is_down_surfaces_as_storage() {
5226 let (vm, _log) = vm_with_a_failing_store(2);
5227 vm.exec(
5228 r#"
5229 local e = failure(function()
5230 do local s <close> = open_failing() end
5231 end)
5232 assert(e.kind == "storage", "kind: " .. tostring(e.kind))
5233 assert(e.method == "close", "method: " .. tostring(e.method))
5234 assert(e.retryable == false, "a store that is down is not a retry")
5235 assert(e.message == "the store is down", "message: " .. tostring(e.message))
5236 "#,
5237 )
5238 .expect("failing store chunk");
5239 }
5240
5241 /// (I6) An explicit `close` wins: the scope exit that follows it is a
5242 /// no-op, so the reason in the log is the caller's and there is exactly
5243 /// one boundary.
5244 #[test]
5245 fn an_explicit_close_wins_over_the_scope_exit() {
5246 let dir = tempfile::tempdir().expect("tempdir");
5247 let path = dir.path().join("knl.db");
5248 let path_str = path.to_str().expect("utf-8 path");
5249
5250 let id = stream_id_from(format!(
5251 r#"
5252 local id
5253 do
5254 local s <close> = knl.open({{ store = {{ sqlite = "{path_str}" }}, owner = "t" }})
5255 id = s:id()
5256 s:close("done")
5257 end
5258 return id
5259 "#
5260 ));
5261
5262 let log = persisted(&path, &id);
5263 let finished: Vec<&Value> = log
5264 .iter()
5265 .filter(|e| e["kind"] == "session_closed")
5266 .collect();
5267 assert_eq!(finished.len(), 1, "exactly one boundary: {log:?}");
5268 assert_eq!(finished[0]["data"]["reason"], Value::from("done"));
5269 }
5270
5271 /// (I6) The backstop: a handle that goes out of scope with no `<close>`
5272 /// and no explicit close still records the boundary when the collector
5273 /// reclaims it. A session that ends by being forgotten is still an
5274 /// ended session, and a reader of the stream must not see it as open
5275 /// forever.
5276 #[test]
5277 fn a_collected_handle_records_the_boundary_as_dropped() {
5278 let dir = tempfile::tempdir().expect("tempdir");
5279 let path = dir.path().join("knl.db");
5280 let path_str = path.to_str().expect("utf-8 path");
5281
5282 let id = stream_id_from(format!(
5283 r#"
5284 -- Opened inside a function so the handle is unreachable the
5285 -- moment it returns: nothing holds the userdata but the
5286 -- collector.
5287 local function run()
5288 local s = knl.open({{ store = {{ sqlite = "{path_str}" }}, owner = "t" }})
5289 s:append({{ kind = "note" }})
5290 return s:id()
5291 end
5292 local id = run()
5293 collectgarbage("collect")
5294 collectgarbage("collect")
5295 return id
5296 "#
5297 ));
5298
5299 let log = persisted(&path, &id);
5300 let last = log.last().expect("the stream is not empty");
5301 assert_eq!(
5302 last["kind"],
5303 Value::from("session_closed"),
5304 "the collector left the session open: {log:?}"
5305 );
5306 assert_eq!(last["data"]["reason"], Value::from("dropped"), "{last}");
5307 }
5308
5309 // -- reading the log with SQL ------------------------------------------
5310
5311 /// The fourth read face: one `SELECT` over the table the events live in.
5312 /// `$stream` is this session without the caller naming it, values are
5313 /// bound, and the second return says whether the cap cut anything off.
5314 #[test]
5315 fn query_reads_the_log_with_sql() {
5316 let vm = vm();
5317 vm.exec(
5318 r#"
5319 local s = knl.open({ owner = "q" })
5320 s:append({ kind = "msg_user", beat = "b1", data = { content = "hi" } })
5321 s:append({ kind = "note", meta = { label = "a" }, data = { text = "a note" } })
5322
5323 local rows, truncated = s:query(
5324 "SELECT seq, kind FROM events WHERE stream = $stream ORDER BY seq")
5325 assert(#rows == 3, "rows: " .. tostring(#rows))
5326 assert(truncated == false, "nothing was cut off")
5327 assert(rows[1].kind == "session_opened", "first: " .. tostring(rows[1].kind))
5328 assert(rows[2].kind == "msg_user" and rows[2].seq == 2)
5329 assert(rows[3].kind == "note")
5330
5331 -- A fold the kernel does not name is a query, not a view it had
5332 -- to be taught.
5333 local counted = s:query([[
5334 SELECT kind, COUNT(*) AS n FROM events
5335 WHERE stream = $stream GROUP BY kind ORDER BY kind]])
5336 assert(#counted == 3, "kinds: " .. tostring(#counted))
5337
5338 -- The envelope is columns, so grouping a run by beat is a
5339 -- GROUP BY rather than a json path…
5340 local beats = s:query([[
5341 SELECT beat, COUNT(*) AS n FROM events
5342 WHERE stream = $stream AND beat IS NOT NULL GROUP BY beat]])
5343 assert(#beats == 1 and beats[1].beat == "b1" and beats[1].n == 1,
5344 "beat is a column of its own")
5345
5346 -- …while a kind's own shape is read out of `data`, and `meta`
5347 -- can be read without knowing the kind at all.
5348 local read = s:query([[
5349 SELECT json_extract(data, '$.content') AS content,
5350 json_extract(meta, '$.label') AS label
5351 FROM events WHERE stream = $stream AND kind = 'msg_user']])
5352 assert(read[1].content == "hi", "data path: " .. tostring(read[1].content))
5353 assert(read[1].label == nil, "this one carried no meta")
5354
5355 -- Values are bound: positionally…
5356 local one = s:query("SELECT kind FROM events WHERE kind = ?", { "note" })
5357 assert(#one == 1 and one[1].kind == "note", "positional bind")
5358 -- …and by name, with the prefix character left to SQLite.
5359 local named = s:query("SELECT kind FROM events WHERE kind = :kind",
5360 { kind = "msg_user" })
5361 assert(#named == 1 and named[1].kind == "msg_user", "named bind")
5362
5363 -- A quote in a value is a character, not the end of a string, and
5364 -- a value that would be SQL if it were pasted in matches nothing.
5365 s:append({ kind = "it's odd" })
5366 local quoted = s:query("SELECT kind FROM events WHERE kind = ?", { "it's odd" })
5367 assert(#quoted == 1, "a quote in a bound value: " .. tostring(#quoted))
5368 local injected = s:query("SELECT kind FROM events WHERE kind = ?",
5369 { "x' OR 1=1 --" })
5370 assert(#injected == 0, "a bound value is never SQL: " .. tostring(#injected))
5371
5372 -- The SQLite types come back as themselves, and a NULL column is
5373 -- absent rather than present-and-null, so it reads as nil.
5374 local typed = s:query(
5375 "SELECT 1 AS whole, 1.5 AS fraction, 'text' AS words, NULL AS absent")
5376 assert(typed[1].whole == 1 and typed[1].fraction == 1.5)
5377 assert(typed[1].words == "text")
5378 assert(typed[1].absent == nil, "a NULL column reads as nil")
5379
5380 -- Reads keep working after the handle closed.
5381 s:close()
5382 assert(#s:query("SELECT 1 AS one") == 1, "a closed handle still reads")
5383 "#,
5384 )
5385 .expect("query chunk");
5386 }
5387
5388 /// `$sessions` reads across the set it was given: two streams in one
5389 /// database, one statement. This is what a session tree reads with.
5390 #[test]
5391 fn query_reads_across_the_session_set() {
5392 let vm = vm();
5393 let dir = tempfile::tempdir().expect("tempdir");
5394 let path = dir.path().join("knl.db");
5395 let path = path.to_str().expect("utf-8 path");
5396
5397 vm.exec(&format!(
5398 r#"
5399 local path = "{path}"
5400 local a = knl.open({{ store = {{ sqlite = path }}, owner = "a" }})
5401 local b = knl.open({{ store = {{ sqlite = path }}, owner = "b" }})
5402 a:append({{ kind = "from_a" }})
5403 b:append({{ kind = "from_b" }})
5404
5405 local sql = "SELECT stream, kind FROM events WHERE stream IN $sessions \
5406 AND kind LIKE 'from_%' ORDER BY kind"
5407
5408 -- Both streams, one statement.
5409 local both = a:query(sql, nil, {{ sessions = {{ a:id(), b:id() }} }})
5410 assert(#both == 2, "both streams: " .. tostring(#both))
5411 assert(both[1].kind == "from_a" and both[2].kind == "from_b")
5412
5413 -- Left out, the set is the asking session's own stream.
5414 local mine = a:query(sql)
5415 assert(#mine == 1 and mine[1].kind == "from_a", "own stream only")
5416
5417 -- An empty set is a mistake, not "all of them".
5418 local e = failure(function() a:query(sql, nil, {{ sessions = {{}} }}) end)
5419 assert(e.kind == "validation", "kind: " .. tostring(e.kind))
5420 assert(e.method == "query", "method: " .. tostring(e.method))
5421 "#
5422 ))
5423 .expect("session set chunk");
5424 }
5425
5426 /// A query reads. Anything that writes, and anything that is two
5427 /// statements, is refused as the caller's mistake — before the connection
5428 /// is reached, and on a connection that could not do it anyway.
5429 #[test]
5430 fn query_refuses_everything_that_is_not_one_read() {
5431 let vm = vm();
5432 vm.exec(
5433 r#"
5434 local s = knl.open({ owner = "q" })
5435 s:append({ kind = "note" })
5436
5437 for _, sql in ipairs({
5438 "INSERT INTO events (stream) VALUES ('x')",
5439 "UPDATE events SET kind = 'x'",
5440 "DELETE FROM events",
5441 "DROP TABLE events",
5442 "PRAGMA table_info(events)",
5443 "ATTACH DATABASE '/tmp/other.db' AS other",
5444 "SELECT 1; DROP TABLE events",
5445 }) do
5446 local e = failure(function() s:query(sql) end)
5447 assert(e.kind == "validation", sql .. " -> " .. tostring(e.kind))
5448 assert(e.method == "query", sql .. " -> " .. tostring(e.method))
5449 end
5450
5451 -- The log is exactly as it was.
5452 assert(s:len() == 2, "len after the refusals: " .. tostring(s:len()))
5453
5454 -- And the arguments are checked too: a misspelt option is an
5455 -- error rather than a limit nobody applied.
5456 local e = failure(function() s:query("SELECT 1", nil, { rows = 10 }) end)
5457 assert(e.kind == "validation", "kind: " .. tostring(e.kind))
5458 local m = failure(function() s:query(42) end)
5459 assert(m.message:find("sql:", 1, true), m.message)
5460 assert(m.message:find("expected a string", 1, true), m.message)
5461 "#,
5462 )
5463 .expect("refusal chunk");
5464 }
5465
5466 /// The row cap is reported, so a page can be told from a whole answer.
5467 #[test]
5468 fn query_caps_the_rows_and_says_when_it_cut() {
5469 let vm = vm();
5470 vm.exec(
5471 r#"
5472 local s = knl.open({ owner = "q" })
5473 for i = 1, 5 do s:append({ kind = "e" .. i }) end
5474
5475 local rows, truncated = s:query(
5476 "SELECT kind FROM events ORDER BY seq", nil, { limit = 2 })
5477 assert(#rows == 2, "capped rows: " .. tostring(#rows))
5478 assert(truncated == true, "the cap cut rows off")
5479
5480 local all, whole = s:query("SELECT kind FROM events ORDER BY seq", nil, { limit = 6 })
5481 assert(#all == 6 and whole == false, "nothing was cut off")
5482 "#,
5483 )
5484 .expect("limit chunk");
5485 }
5486
5487 /// A query that will not finish is cut short and says so in its own
5488 /// class — "ask again" would be the wrong advice for a slow read.
5489 #[test]
5490 fn query_that_runs_too_long_reports_a_timeout() {
5491 let vm = vm();
5492 vm.exec(
5493 r#"
5494 local s = knl.open({ owner = "q" })
5495 local e = failure(function()
5496 s:query([[WITH RECURSIVE forever(x) AS (
5497 SELECT 1 UNION ALL SELECT x + 1 FROM forever)
5498 SELECT COUNT(*) FROM forever]], nil, { timeout_ms = 50 })
5499 end)
5500 assert(e.kind == "timeout", "kind: " .. tostring(e.kind))
5501 assert(e.method == "query", "method: " .. tostring(e.method))
5502 assert(e.retryable == false, "a slow query is not a retry")
5503
5504 -- The session is fine afterwards: a statement ended, not the
5505 -- reader.
5506 assert(#s:query("SELECT 1 AS one") == 1)
5507 "#,
5508 )
5509 .expect("timeout chunk");
5510 }
5511
5512 /// `knl.api().schema` is the read contract: the table a query names, and
5513 /// its columns as SQLite reports them — including which two are the key.
5514 #[test]
5515 fn api_publishes_the_events_schema() {
5516 let vm = vm();
5517 vm.exec(
5518 r#"
5519 local schema = knl.api().schema
5520 assert(schema.table == "events", "table: " .. tostring(schema.table))
5521
5522 local names, keyed = {}, {}
5523 for _, column in ipairs(schema.columns) do
5524 assert(type(column.name) == "string" and #column.name > 0)
5525 assert(type(column.type) == "string" and #column.type > 0)
5526 table.insert(names, column.name)
5527 if column.pk then table.insert(keyed, column.name) end
5528 end
5529 assert(table.concat(names, ",")
5530 == "stream,seq,epoch_ms,kind,schema_version,beat,meta,data",
5531 "columns: " .. table.concat(names, ","))
5532 assert(table.concat(keyed, ",") == "stream,seq",
5533 "primary key: " .. table.concat(keyed, ","))
5534
5535 -- Every published column is one a query may actually name.
5536 local s = knl.open({ owner = "q" })
5537 local rows = s:query("SELECT " .. table.concat(names, ", ") ..
5538 " FROM " .. schema.table .. " WHERE stream = $stream")
5539 assert(#rows == 1, "the opening event: " .. tostring(#rows))
5540 assert(rows[1].kind == "session_opened")
5541 assert(rows[1].schema_version == 1, "the stored version is a column")
5542 assert(rows[1].beat == nil, "an undeclared beat is NULL")
5543 assert(type(rows[1].meta) == "string", "meta stays the stored text")
5544 assert(type(rows[1].data) == "string", "and so does data")
5545 "#,
5546 )
5547 .expect("schema chunk");
5548 }
5549
5550 // -- the rule this round exists for -------------------------------------
5551
5552 /// **A slow write does not stop the VM.**
5553 ///
5554 /// This is the property the whole round is about, so it is asserted
5555 /// directly rather than inferred from the shape of the code: a second
5556 /// coroutine on the same Lua state goes on running — advancing a counter
5557 /// through an async function of its own — for the *whole* time an
5558 /// `s:append` is waiting on a write lock another connection is holding.
5559 ///
5560 /// Before this round the session's methods were synchronous, so the
5561 /// append would have parked the VM's one thread and the ticker would have
5562 /// counted nothing until the lock was released.
5563 #[test]
5564 fn a_slow_write_does_not_block_another_coroutine_on_the_same_vm() {
5565 use std::sync::atomic::{AtomicUsize, Ordering};
5566 use std::sync::Arc;
5567 use std::time::Duration;
5568
5569 /// How long the blocker holds the write lock.
5570 const HELD: Duration = Duration::from_millis(300);
5571 /// How long each tick takes, so ~60 fit inside `HELD`.
5572 const TICK: Duration = Duration::from_millis(5);
5573 /// The floor the assertion uses. Far below what should actually
5574 /// happen (~60), because the point is "the VM kept running", not a
5575 /// measurement of how fast it ran.
5576 const AT_LEAST: usize = 5;
5577
5578 let vm = vm();
5579 let dir = tempfile::tempdir().expect("tempdir");
5580 let path = dir.path().join("knl.db");
5581 let path_str = path.to_str().expect("utf-8 path").to_string();
5582
5583 // `tick()` waits like any async bridge function does; `ticks()` reads
5584 // the counter without waiting for anything.
5585 let ticks = Arc::new(AtomicUsize::new(0));
5586 let counter = Arc::clone(&ticks);
5587 let tick = vm
5588 .lua
5589 .create_async_function(move |_, ()| {
5590 let counter = Arc::clone(&counter);
5591 async move {
5592 tokio::time::sleep(TICK).await;
5593 counter.fetch_add(1, Ordering::Relaxed);
5594 Ok(())
5595 }
5596 })
5597 .expect("create tick");
5598 vm.lua.globals().set("tick", tick).expect("set tick");
5599 let counter = Arc::clone(&ticks);
5600 let read_ticks = vm
5601 .lua
5602 .create_function(move |_, ()| Ok(counter.load(Ordering::Relaxed)))
5603 .expect("create ticks");
5604 vm.lua
5605 .globals()
5606 .set("ticks", read_ticks)
5607 .expect("set ticks");
5608
5609 // The session is opened before the lock is taken, so the only thing
5610 // waiting on it is the append below.
5611 vm.exec(&format!(
5612 r#"session = knl.open({{ store = {{ sqlite = "{path_str}" }}, owner = "t" }})"#
5613 ))
5614 .expect("open the durable session");
5615
5616 // A second connection holds the write lock for `HELD`, on a thread of
5617 // its own so the test can go on driving the VM.
5618 let (locked_tx, locked_rx) = std::sync::mpsc::channel();
5619 let blocker_path = path.clone();
5620 let blocker = std::thread::spawn(move || {
5621 let conn = rusqlite::Connection::open(&blocker_path).expect("open the blocker");
5622 conn.busy_timeout(HELD).expect("busy timeout");
5623 conn.execute_batch("BEGIN EXCLUSIVE")
5624 .expect("take the write lock");
5625 locked_tx.send(()).expect("announce the lock");
5626 std::thread::sleep(HELD);
5627 conn.execute_batch("ROLLBACK").expect("release the lock");
5628 });
5629 locked_rx.recv().expect("the lock was taken");
5630
5631 // Two coroutines, driven together on the VM's runtime: one blocked on
5632 // the write, one counting. `during` is the number of ticks that
5633 // landed while the append was waiting.
5634 let during: usize = vm.block_on(async {
5635 let writer = vm
5636 .lua
5637 .load(
5638 r#"
5639 local before = ticks()
5640 session:append({ kind = "slow" })
5641 return ticks() - before
5642 "#,
5643 )
5644 .eval_async::<usize>();
5645 let ticker = vm.lua.load(r#"for _ = 1, 200 do tick() end"#).exec_async();
5646 // Both futures poll the same Lua state on this one thread, which
5647 // is exactly what the VM's own LocalSet does with its coroutines.
5648 let (written, _ticked) = tokio::join!(writer, ticker);
5649 written.expect("the append eventually lands")
5650 });
5651
5652 blocker.join().expect("the blocker thread");
5653
5654 assert!(
5655 during >= AT_LEAST,
5656 "the VM stopped while the write was waiting: only {during} tick(s) ran"
5657 );
5658
5659 // And the write itself landed once the lock was released.
5660 vm.exec(r#"assert(kinds_of(session) == "session_opened,slow", kinds_of(session))"#)
5661 .expect("the slow append landed");
5662 }
5663
5664 /// **An identity read never waits and never raises.**
5665 ///
5666 /// `id` / `scope_id` / `owner` are declared as methods that answer out of
5667 /// the value, and [`SESSION_API`] lists no class for them. They used to
5668 /// reach the session behind a `try_lock`, which has exactly one answer for
5669 /// "somebody else is mid-call" and it is a raise — so a second coroutine
5670 /// asking a session its own id while the first was suspended inside
5671 /// `s:append` got a `validation` failure instead of a string. The three
5672 /// values are copied into the userdata at construction now, and this is
5673 /// the case that says so: same shape as the non-blocking write above, with
5674 /// the identity read taking the ticker's place.
5675 #[test]
5676 fn an_identity_read_answers_while_another_coroutine_holds_the_session() {
5677 use std::time::Duration;
5678
5679 /// How long the blocker holds the write lock — long enough that the
5680 /// append below is certainly still suspended when the reader runs.
5681 const HELD: Duration = Duration::from_millis(300);
5682
5683 let vm = vm();
5684 let dir = tempfile::tempdir().expect("tempdir");
5685 let path = dir.path().join("knl.db");
5686 let path_str = path.to_str().expect("utf-8 path").to_string();
5687
5688 // One yield for the reader, so it asks from inside the same suspension
5689 // the writer is parked in rather than before the writer got there.
5690 let pause = vm
5691 .lua
5692 .create_async_function(|_, ()| async move {
5693 tokio::time::sleep(Duration::from_millis(20)).await;
5694 Ok(())
5695 })
5696 .expect("create pause");
5697 vm.lua.globals().set("pause", pause).expect("set pause");
5698
5699 vm.exec(&format!(
5700 r#"
5701 session = knl.open({{ store = {{ sqlite = "{path_str}" }}, owner = "u-7" }})
5702 -- What the reads must still answer while the session is held.
5703 expected_id, expected_scope, expected_owner =
5704 session:id(), session:scope_id(), session:owner()
5705 "#
5706 ))
5707 .expect("open the durable session");
5708
5709 // A second connection holds the write lock, so the append below is
5710 // parked inside the store with the session's own lock held.
5711 let (locked_tx, locked_rx) = std::sync::mpsc::channel();
5712 let blocker_path = path.clone();
5713 let blocker = std::thread::spawn(move || {
5714 let conn = rusqlite::Connection::open(&blocker_path).expect("open the blocker");
5715 conn.busy_timeout(HELD).expect("busy timeout");
5716 conn.execute_batch("BEGIN EXCLUSIVE")
5717 .expect("take the write lock");
5718 locked_tx.send(()).expect("announce the lock");
5719 std::thread::sleep(HELD);
5720 conn.execute_batch("ROLLBACK").expect("release the lock");
5721 });
5722 locked_rx.recv().expect("the lock was taken");
5723
5724 let read: String = vm.block_on(async {
5725 let writer = vm
5726 .lua
5727 .load(r#"session:append({ kind = "slow" })"#)
5728 .exec_async();
5729 // The reader yields once first, so the writer is inside the store
5730 // — and therefore holding the session — before it asks.
5731 let reader = vm
5732 .lua
5733 .load(
5734 r#"
5735 pause()
5736 local id, scope, owner = session:id(), session:scope_id(), session:owner()
5737 assert(id == expected_id, "id: " .. tostring(id))
5738 assert(scope == expected_scope, "scope_id: " .. tostring(scope))
5739 assert(owner == expected_owner, "owner: " .. tostring(owner))
5740 assert(owner == "u-7", "owner: " .. tostring(owner))
5741 return id
5742 "#,
5743 )
5744 .eval_async::<String>();
5745 let (written, read) = tokio::join!(writer, reader);
5746 written.expect("the append eventually lands");
5747 read.expect("an identity read must answer while the session is held")
5748 });
5749
5750 blocker.join().expect("the blocker thread");
5751 assert!(!read.is_empty(), "the read answered the stream's own id");
5752 }
5753
5754 /// `knl.open{ parent = s, budget = { from_parent = n } }` opens a session
5755 /// on the parent's database and out of its balance, in one write: the
5756 /// child's log names the parent and carries the grant, and the parent's
5757 /// ledger carries the reservation naming the child.
5758 #[test]
5759 fn open_with_a_parent_allocates_out_of_the_parents_balance() {
5760 let vm = vm();
5761 vm.exec(
5762 r#"
5763 local parent = knl.open({ owner = "u", budget = { amount = 100, tag = "tokens" } })
5764 local child = knl.open({
5765 owner = "worker",
5766 parent = parent,
5767 budget = { from_parent = 40 },
5768 })
5769
5770 assert(child:id() ~= parent:id(), "a child is its own stream")
5771 assert(child:owner() == "worker", child:owner())
5772 assert(parent:remaining() == 60, "parent: " .. tostring(parent:remaining()))
5773 assert(child:remaining() == 40, "child: " .. tostring(child:remaining()))
5774
5775 -- the child's own log: opened, and opened with the grant
5776 assert(kinds_of(child) == "session_opened,budget_granted", kinds_of(child))
5777 local opened = child:events()[1]
5778 assert(opened.data.parent == parent:id(), tostring(opened.data.parent))
5779 local granted = child:events()[2]
5780 assert(granted.data.parent == parent:id(), tostring(granted.data.parent))
5781 assert(granted.data.amount == 40, tostring(granted.data.amount))
5782 assert(granted.data.tag == "tokens", "the parent's unit by default")
5783
5784 -- the parent's side: a reservation naming where the units went
5785 local ledger = parent:events()
5786 local reserved = ledger[#ledger]
5787 assert(reserved.kind == "budget_reserved", reserved.kind)
5788 assert(reserved.data.child == child:id(), tostring(reserved.data.child))
5789
5790 -- and closing the child gives nothing back
5791 child:close("done")
5792 assert(parent:remaining() == 60, "an allocation is a spend")
5793 parent:close("done")
5794 "#,
5795 )
5796 .expect("the allocation");
5797 vm.finish();
5798 }
5799
5800 /// A balance that will not cover the allocation raises `refused` — the
5801 /// class that reports a decision — with the refusal in the parent's log
5802 /// and no session handed back.
5803 #[test]
5804 fn a_child_the_parent_cannot_pay_for_is_refused() {
5805 let vm = vm();
5806 vm.exec(
5807 r#"
5808 local parent = knl.open({ owner = "u", budget = { amount = 10, tag = "tokens" } })
5809 local read, raised = failure(knl.open, {
5810 owner = "worker", parent = parent, budget = { from_parent = 40 },
5811 })
5812 assert(read.kind == "refused", "kind: " .. tostring(read.kind))
5813 assert(read.method == "open", "method: " .. tostring(read.method))
5814 assert(read.retryable == false, "the same balance answers the same")
5815 assert(tostring(raised):find("40", 1, true), tostring(raised))
5816
5817 -- recorded on the parent, and the balance did not move
5818 assert(parent:remaining() == 10, tostring(parent:remaining()))
5819 local ledger = parent:events()
5820 local refused = ledger[#ledger]
5821 assert(refused.kind == "budget_refused", refused.kind)
5822 assert(refused.data.remaining == 10, tostring(refused.data.remaining))
5823 assert(type(refused.data.child) == "string", "the refusal names the child")
5824 parent:close("done")
5825 "#,
5826 )
5827 .expect("the refusal");
5828 vm.finish();
5829 }
5830
5831 /// The two forms of `budget` are exclusive in both directions, and a
5832 /// child on another store is refused as the validation it is. A quota
5833 /// nobody paid for and a tree spread over two logs are the two shapes
5834 /// this rules out.
5835 #[test]
5836 fn a_parent_and_a_grant_are_not_mixed() {
5837 let vm = vm();
5838 vm.exec(
5839 r#"
5840 local parent = knl.open({ owner = "u", budget = { amount = 100, tag = "tokens" } })
5841
5842 -- from_parent with nobody to take it from
5843 local orphan = failure(knl.open, { owner = "w", budget = { from_parent = 5 } })
5844 assert(orphan.kind == "validation", orphan.kind)
5845 assert(orphan.message:find("opts.parent", 1, true), orphan.message)
5846
5847 -- a parent, and an owner's grant instead of an allocation
5848 local granted = failure(knl.open, {
5849 owner = "w", parent = parent, budget = { amount = 5 },
5850 })
5851 assert(granted.kind == "validation", granted.kind)
5852 assert(granted.message:find("from_parent", 1, true), granted.message)
5853
5854 -- both at once says neither
5855 local both = failure(knl.open, {
5856 owner = "w", parent = parent, budget = { amount = 5, from_parent = 5 },
5857 })
5858 assert(both.kind == "validation", both.kind)
5859
5860 -- a parent that is not a session
5861 local nonsense = failure(knl.open, {
5862 owner = "w", parent = "s-1", budget = { from_parent = 5 },
5863 })
5864 assert(nonsense.kind == "validation", nonsense.kind)
5865 assert(nonsense.message:find("must be a session", 1, true), nonsense.message)
5866
5867 -- a child on a store of its own is a second log, and a tree is one
5868 local split = failure(knl.open, {
5869 owner = "w", parent = parent, budget = { from_parent = 5 }, store = "mem",
5870 })
5871 assert(split.kind == "validation", split.kind)
5872 assert(split.message:find("one log", 1, true), split.message)
5873
5874 -- none of it moved the balance
5875 assert(parent:remaining() == 100, tostring(parent:remaining()))
5876 parent:close("done")
5877 "#,
5878 )
5879 .expect("the refusals");
5880 vm.finish();
5881 }
5882
5883 /// Closing a parent whose children are still open is not refused: the
5884 /// boundary records them and lands.
5885 #[test]
5886 fn a_close_records_the_children_that_were_still_open() {
5887 let vm = vm();
5888 vm.exec(
5889 r#"
5890 local parent = knl.open({ owner = "u", budget = { amount = 100, tag = "tokens" } })
5891 local running = knl.open({ owner = "w", parent = parent, budget = { from_parent = 10 } })
5892 local done = knl.open({ owner = "w", parent = parent, budget = { from_parent = 10 } })
5893 done:close("done")
5894
5895 parent:close("done")
5896 local events = parent:events()
5897 local boundary = events[#events]
5898 assert(boundary.kind == "session_closed", boundary.kind)
5899 local open_children = boundary.data.open_children
5900 assert(type(open_children) == "table", type(open_children))
5901 assert(#open_children == 1, "one child was still open, got " .. #open_children)
5902 assert(open_children[1] == running:id(), tostring(open_children[1]))
5903 running:close("done")
5904 "#,
5905 )
5906 .expect("the close");
5907 vm.finish();
5908 }
5909
5910 /// A durable tree: the child goes into the parent's file without being
5911 /// told where that is, and one recursive statement reads the shape back
5912 /// out of the log.
5913 #[test]
5914 fn a_childs_stream_lands_in_the_parents_database() {
5915 let vm = vm();
5916 let dir = tempfile::tempdir().expect("tempdir");
5917 let path = dir.path().join("tree.db");
5918 let path_str = path.to_str().expect("utf-8 path").to_string();
5919
5920 let rows: usize = vm
5921 .eval(&format!(
5922 r#"
5923 local parent = knl.open({{
5924 owner = "u",
5925 budget = {{ amount = 100, tag = "tokens" }},
5926 store = {{ sqlite = "{path_str}" }},
5927 }})
5928 local child = knl.open({{
5929 owner = "w", parent = parent, budget = {{ from_parent = 25 }},
5930 }})
5931 -- The child was never told where the log is, and it is in it:
5932 -- one statement over the parent's own store reaches both.
5933 local found = parent:query(
5934 "SELECT stream FROM events WHERE kind = 'session_opened' ORDER BY stream"
5935 )
5936 child:close("done")
5937 parent:close("done")
5938 return #found
5939 "#
5940 ))
5941 .expect("the durable tree");
5942 assert_eq!(rows, 2, "the parent and its child are in one database");
5943 vm.finish();
5944 }
5945}