Substrate: qexec executor & qvalue model

What the substrate is#

Qirava's stdlib (the qpkgs tree) is organized as one substrate plus one isolated function package per area.* namespace (array.*, string.*, math.*, crypto.*, and so on). The substrate is two crates:

qexec

The bounded worker executor and function runtime. It owns a thread-backed worker pool with per-worker load caps, a resource budget derived from the device's real capacities, FIFO route affinity, optional metrics, and a scoped registry of built-in and user-defined functions. It is the thing every function package registers against. Pure std; depends on nothing else.

qvalue

The shared dynamic value model: arbitrary-precision integers and decimals (bignum), a Value enum and Record type, and a compact length-prefixed binary codec. It is the common ABI for arguments and results across every function package and the DMS. Pure std; does not depend on qexec.

Crate names are qexec and qvalue (the older tq* prefix has been renamed away). A consumer wires them by path dependency; there is no top-level workspace manifest, so the de-facto member list is whatever a consumer declares. The DMS (qdms) path-deps both, which is where cargo check runs green.

The qexec executor: configuration and the 80% budget#

The executor never guesses how much of the machine it may use. It starts from device capacities (raw totals) and applies a safety budget on top — by default 80% — and it never exceeds the physical capacity reported. You describe capacities with SystemResources and turn them into an ExecConfig.

SystemResources

Raw device totals: cpu_threads: usize, memory_bytes: Option<u64>, storage_bytes: Option<u64>. None means unknown/unlimited for that dimension.

SystemResources::scan()

Best-effort, zero-dependency scan of the host. CPU thread count is always resolved (via std available_parallelism, floored at 1). On 64-bit Linux it reads total memory from /proc/meminfo and total storage via statvfs at "/"; on other platforms memory and storage are left None so the executor degrades to unbounded for those dimensions instead of guessing.

SystemResources::scan_at(path)

Like scan(), but measures storage at a specific data path instead of "/".

ExecConfig is the executor's full configuration. Its fields and their defaults:

Three constructors cover the common cases. ExecConfig::auto() is the production entrypoint — it scans the host and applies the default 80% system budget. ExecConfig::from_system(resources, system_budget_percent) builds from capacities you supply (scanned or from user config). ExecConfig::default_for(available_threads) builds a thread-only config with capacities unknown (unbounded memory/storage). ExecConfig::default() uses default_for with the detected parallelism.

use qexec::{ExecConfig, SystemResources};

// Production: scan the host, apply the default 80% safety budget.
let config = ExecConfig::auto();

// Equivalent, explicit:
let config = ExecConfig::from_system(SystemResources::scan(), 80);

// User-supplied capacities (e.g. a container quota) instead of a scan:
let config = ExecConfig::from_system(
    SystemResources { cpu_threads: 8, memory_bytes: Some(16 << 30), storage_bytes: Some(512 << 30) },
    80,
);

How the worker thread count is derived when worker_threads is None: it is system_budget_percent OF worker_budget_percent of available_threads — i.e. 80% of 80% by default — floored at 1. On a 16-thread host that is 16 * 0.8 * 0.8 = 10 workers.

How memory and storage ceilings are resolved: the budgeted ceiling is capacity * system_budget_percent (the 80% margin — NOT 80%*80%). An explicit max_* override replaces the budget but is still clamped down to physical capacity, so the executor stays at or below the device. If capacity is unknown and there is no explicit cap, that dimension is unbounded (None). Inspect the resolved values with resolved_worker_threads(), resolved_memory_limit(), resolved_storage_limit(), and resolved_operation_threads_hint() (the helper-pool size that fills out the rest of the system budget so workers + helpers stays within the 80% target, never starved below 1).

let resources = SystemResources { cpu_threads: 16, memory_bytes: Some(1_000), storage_bytes: Some(2_000) };
let config = ExecConfig::from_system(resources, 80);

config.available_threads               // raw threads
config.resolved_worker_threads()       // 80% of 80% of 16
config.resolved_operation_threads_hint() // helpers fill the rest of the 80% budget
config.resolved_memory_limit()         // 80% of 1000
config.resolved_storage_limit()        // 80% of 2000

available_threads                = 16
resolved_worker_threads()        = Ok(10)
resolved_operation_threads_hint()= Ok(2)   // budget(12) - workers(10) = 2
resolved_memory_limit()          = Some(800)
resolved_storage_limit()         = Some(1_600)

These are the exact values asserted by qexec's own tests (memory_and_storage_budget_use_system_percent_not_double, default_uses_80_percent_of_80_percent_for_workers).

Routing: ExecutionRoute, ResourceRequest, and Guards#

Every submission to the executor carries a route (where and how to place it) and a resource request (what it needs to reserve). Three small types describe this.

FunctionId(u64) / ExecutionKey(u64)

Opaque identifiers. FunctionId names the function; ExecutionKey identifies the logical stream of work (for record mutations the runtime passes the record's key so writes to the same record can serialize).

ExecutionRoute { function_id, execution_key, fifo }

How a job is placed. Construct with ExecutionRoute::new(function_id, execution_key, fifo). When fifo is true, all jobs sharing the same (function_id, execution_key) are pinned to one worker and serialized while that route is active; when it drains to zero active jobs the affinity is released. When fifo is false the job goes to the lowest-loaded worker.

ResourceRequest { memory_bytes, storage_bytes }

What to reserve from the shared budget for this job. Helpers: ResourceRequest::empty() (reserve nothing), ::new(mem, storage), ::memory(bytes), ::storage(bytes). An empty request is free — it touches no atomics and allocates nothing — so declaring a zero budget on a hot path costs nothing.

Reservation is admission control. The executor reserves the requested bytes BEFORE it enqueues the job; if the reservation would exceed a configured ceiling the submission is rejected immediately with ResourceLimitExceeded and no worker slot is consumed. A dimension with no configured cap is never accounted, so reservations against an uncapped config are always free no-ops. Reservations are RAII: the budget is returned automatically when the job finishes (or its lease is dropped). The public guard type is ResourceGuard, obtained from a ResourcePool via try_reserve, or from inside a job via WorkerContext::reserve_memory / reserve_storage / reserve_resources.

A job is a boxed closure FnOnce(&mut WorkerContext, &mut Vec<u8>) -> Result<(), JobError>. It writes its result bytes into the supplied output buffer and returns Ok(()) on success, or a JobError { code: u16, message: String } on failure. WorkerContext exposes worker_id(), route(), and the reserve_* helpers. A panic inside a job is caught and surfaced as ExecError::JobPanicked rather than crashing the process.

Submitting and awaiting jobs#

Executor::new(config) spawns the worker threads and returns a cheap-to-clone handle. Submit with try_submit(route, resources, job), which is non-blocking: it returns Err immediately if all workers are at their load cap (WorkersBusy), if the reservation fails (ResourceLimitExceeded), or if a worker thread has stopped (WorkerStopped). On success you get a JobHandle; call wait() to block for the result bytes, or try_wait() to poll without blocking.

use qexec::{ExecConfig, Executor, ExecutionRoute, FunctionId, ExecutionKey, ResourceRequest};

let executor = Executor::new(ExecConfig {
    available_threads: 2,
    worker_threads: Some(1),
    ..ExecConfig::default_for(2)
}).unwrap();

let handle = executor.try_submit(
    ExecutionRoute::new(FunctionId(1), ExecutionKey(1), false),
    ResourceRequest::empty(),
    Box::new(|context, output| {
        output.extend_from_slice(format!("worker={}", context.worker_id()).as_bytes());
        Ok(())
    }),
).unwrap();

let response = handle.wait().unwrap();

response == b"worker=0"

Transcribed from qexec's executes_job_and_returns_response test. handle.wait() returns Result<Vec<u8>, ExecError>; here it is Ok(b"worker=0").

Introspection on a live executor: worker_count(), worker_loads() (current in-flight count per worker), resources() (a ResourceSnapshot of used vs. max bytes), resource_pool() (a clonable ResourcePool you can reserve against off-worker), operation_threads_hint(), and metrics() (a MetricsSnapshot — enabled, submitted, completed, failed, busy, avg_latency_ns, max_latency_ns; all zero unless metrics_enabled was set).

The function runtime: registry, scopes, and execute()#

Most callers do not use the raw executor directly; they use the function runtime layered on top of it (re-exported from qexec). The runtime owns one always-ready executor plus a registry that holds two origins of functions, dispatched identically:

FunctionOrigin::Builtin

Hardcoded functions shipped with the DMS / stdlib (e.g. the area.* packages).

FunctionOrigin::UserDefined

Functions registered by the application on top of the DMS, including ones added at runtime.

Each function has a scope that controls who may call it:

FunctionScope::Public

Callable from the outside via execute / execute_detached / call_inline. Every runtime must register at least one public function or build() fails with NoPublicFunctions.

FunctionScope::InternalGlobal

A helper callable from inside ANY function via ctx.call / chain / all.

FunctionScope::InternalScoped { owner }

A helper callable only within the call tree rooted at owner (root_key must equal owner). owner must be a registered public function or build fails with MissingOwner.

FunctionScope::InternalPrivate { owner }

A helper callable only by its immediate owner (current_key must equal owner). owner must be a registered function.

You build a runtime with FunctionRuntime::builder(config), register functions, then build(). Public registrars take (key, resources, fifo, body); internal registrars take (key, body) and reserve nothing by default (with _with_resources variants to declare a budget). The body is Fn(&mut FunctionContext, &[u8]) -> FunctionResponse + Send + Sync + 'static. Building assigns ids (public ids from 1, internal from 1_000_000) and validates owner references.

execute(function_key, input, fifo, execution_key) is the one and only way to run a registered PUBLIC function — from a request, a scheduler, or another caller. fifo defaults to false (concurrent); true serializes this call on its execution_key, and a function registered fifo always serializes regardless. execution_key is Option: None makes the runtime generate a unique key (only matters under fifo). It returns a FunctionCall; call wait() for the FunctionResponse (or try_wait() to poll).

let mut b = FunctionRuntime::builder(ExecConfig::default_for(2));
b.register_public("seed", ResourceRequest::empty(), false, |_c, _i| {
    FunctionResponse::ok(b"seed".to_vec())
}).unwrap();
let rt = b.build().unwrap();

rt.register_dynamic(
    "dyn.echo",
    FunctionScope::Public,
    ResourceRequest::empty(),
    false,
    |_c, input| FunctionResponse::ok(input.to_vec()),
).unwrap();

let resp = rt.execute("dyn.echo", b"hi", false, Some(ExecutionKey(1))).wait();

resp.ok        == true
resp.code      == ResponseCode::Ok
resp.data      == b"hi"
resp.meta.root_key == "dyn.echo"

Transcribed from the dynamic_function_registered_after_build test. The runtime also exposes register_dynamic (errors on a duplicate key), set_dynamic (register-or-replace atomically, no missing window), unregister_dynamic (returns whether it existed), and is_registered.

Other run paths: execute_detached(key, input, fifo, execution_key, on_complete) submits without a waitable handle and invokes the callback exactly once when done — used by event-loop callers that must not block a thread on wait(). call_inline(key, input) runs a public function on the CALLING thread under the same shared reservation, for re-entrant calls (e.g. an inline QQL call from inside a query job) without submitting a second pool job; it is bounded to 64 levels of nesting to prevent a recursive function from overflowing the worker stack.

FunctionResponse, ResponseCode, and helper composition#

Every function returns a FunctionResponse with fields ok: bool, code: ResponseCode, message: String, data: Vec<u8>, error: Option<FunctionErrorInfo>, and meta: ResponseMeta. Construct success with FunctionResponse::ok(data) or ok_message(message, data); construct failure with FunctionResponse::error(code, message). The runtime stamps meta on the way out: root_key (the public function that started the call), worker_id (Some on a worker, None inline), and helper_calls (how many helper invocations the call made).

Inside a function body, FunctionContext is the only way to invoke helpers — every path reserves the helper's declared budget. ctx.call(key, input) runs one helper. ctx.chain(&[keys], input) runs helpers in sequence, feeding each one's data into the next and stopping at the first non-ok. ctx.all(&[HelperRequest]) runs a batch sequentially; ctx.all_concurrent(&[HelperRequest]) fans the batch out across the helper pool (the first call always runs inline on the worker so it participates rather than idling), falling back to sequential when there is one call, no helper pool, or any call is fifo. A HelperRequest is HelperRequest::new(key, input), with .fifo() and .with_resources(req) builders; a per-call .with_resources override takes precedence over the helper's registered budget on every path. ctx also offers reserve_memory / reserve_storage / reserve_resources (returning a ResourceGuard or ResponseCode::ResourceLimit) and the accessors worker_id(), root_key(), current_key(), helper_calls().

There is also an off-node wire format for FunctionResponse (magic bytes TQFR1, little-endian, length-prefixed) retained for shipping responses across machines. The in-process fast path does NOT use it — the typed FunctionResponse is handed back through the FunctionCall's slot, skipping the encode/decode round-trip entirely.

The qvalue model: Value, Record, and types#

qvalue is the common vocabulary for structured arguments and results. Its central type is the Value enum — a dynamically-typed value covering everything a function or a database row needs, including arbitrary-precision numbers and a first-class vector for AI/embedding workloads.

Helpers on Value: decimal_from_str(s) and bigint_from_str(s) parse arbitrary-precision numbers from text (None on malformed input); as_bigdecimal() coerces any numeric Value to an exact BigDecimal (Int/BigInt/Decimal exact, Float via its decimal text, Timestamp as its nanosecond integer); is_number() tests numeric variants; and compare(&other) gives the total-ish ordering used by WHERE/SORT/MIN/MAX — numerics compare across types via exact BigDecimal (with native fast paths for plain Int/Float/Timestamp), strings and bools compare natively, and incomparable types yield None.

A Record is an ordered set of named fields: pub fields: Vec<(String, Value)>. Build fluently with Record::new().with(name, value); read with get(name) or get_path("addr.city") which navigates nested Maps (an exact flat field named literally "a.b" still wins first). Records are the unit the codec serializes.

Schema support lives alongside the model for the database's hybrid schema/schemaless tables: ColumnType (Int, BigInt, Decimal, Float, Timestamp, Bool, Str, Bytes, Vector(u32), List, Map, Any), Column (with .nullable() and .unique() builders), TableSchema (Schema(Vec<Column>) or Schemaless), and validate(record, schema). Schemaless tables always pass; schema tables require every declared column present (or nullable), matching its type and vector dimension, and reject unknown fields — returning a SchemaError (MissingField, TypeMismatch, VectorDim, UnknownField).

The qvalue binary codec#

qvalue ships a compact, deterministic, zero-dependency binary codec for Records: encode_record(&Record) -> Vec<u8> and decode_record(&[u8]) -> Option<Record>. The format is length-prefixed and little-endian. A record is encoded as a u32 field count, then for each field a u32 name length, the UTF-8 name bytes, and the encoded value. Each value begins with a 1-byte type tag, then its tag-specific payload.

use qvalue::model::{Record, Value, encode_record, decode_record};

let r = Record::new()
    .with("id", Value::Int(42))
    .with("supply", Value::bigint_from_str("170141183460469231731687303715884105727000").unwrap())
    .with("price", Value::decimal_from_str("1234.000000000000000000000000001").unwrap())
    .with("ts", Value::Timestamp(1_700_000_000_123_456_789))
    .with("name", Value::Str("alice".into()))
    .with("active", Value::Bool(true))
    .with("score", Value::Float(9.5))
    .with("embedding", Value::Vector(vec![0.1, 0.2, 0.3]))
    .with("tags", Value::List(vec![Value::Str("a".into()), Value::Int(2)]))
    .with("meta", Value::Map(vec![("k".into(), Value::Str("v".into()))]));

let bytes = encode_record(&r);
let back  = decode_record(&bytes).expect("decode");

back == r            // exact round-trip, including BigInt/Decimal of any range
decode_record(&bytes[..bytes.len() - 3]) == None   // truncation rejected

Transcribed from qvalue's record_codec_round_trips and decode_rejects_truncated_buffer tests. The arbitrary-precision supply and price values survive byte-for-byte because they are encoded as exact decimal text.

Putting it together: a function package#

An area.* function package is the canonical consumer of the substrate. It depends on exactly qexec and qvalue (crypto is the one exception, also depending on encoding). It registers its functions against a FunctionRuntimeBuilder; each body decodes its &[u8] input into qvalue Values (often via decode_record), computes a result, and encodes it back to bytes inside a FunctionResponse. Because args and results are always the qvalue ABI and the executor only ever moves bytes, any package speaks to any other through the same shared vocabulary.

caller
  |  execute("array.map", input_bytes, fifo, key)
  v
FunctionRuntime ---- resolve public fn ----> FunctionDef (id, scope, resources, body)
  |  reserve ResourceRequest from shared budget (RAII)
  |  build ExecutionRoute(function_id, execution_key, fifo)
  v
Executor.try_submit  --> lowest-loaded (or FIFO-pinned) worker
  |                       runs body(&mut FunctionContext, &input)
  |                         body: decode qvalue -> compute -> encode qvalue
  |                         (ctx.call / chain / all_concurrent for helpers)
  v
FunctionResponse  <-- typed result handed back via FunctionCall slot
  ^  release reservation + worker slot, record metrics
  |
caller .wait()

1. Build an ExecConfig — ExecConfig::auto() in production, or from_system/default_for for explicit control.
2. FunctionRuntime::builder(config), then register at least one public function (plus any helpers).
3. build() — spawns the worker pool and helper pool and validates the registry.
4. Call execute(key, input, fifo, execution_key) and .wait() for the FunctionResponse, or execute_detached for callback-style completion.
5. Encode inputs and decode outputs with qvalue's encode_record/decode_record so every function shares one wire shape.