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