feat(analytics): BI-P018/BI-P019/BI-P021 — execution model, verified cancellation, a budget that stops queries
CI (SIAX Cloud) / integration (push) Successful in 43s
CI (SIAX Cloud) / security (push) Successful in 21s
CI (SIAX Cloud) / quality (push) Successful in 1m5s
Security Scan (SIAX Cloud) / security (push) Successful in 23s
CI (SIAX Cloud) / security (pull_request) Successful in 24s
CI (SIAX Cloud) / quality (pull_request) Successful in 1m0s
CI (SIAX Cloud) / integration (pull_request) Successful in 42s
Security Scan (SIAX Cloud) / security (pull_request) Successful in 38s

BI-P018 — every field the prompt names, plus the one it names by its absence.
The record carries `compiled_query_hash` and never the statement, and that is
ENFORCED: `assertNoRawSql` runs before any record is emitted and refuses one
whose `error` field looks like SQL. The realistic leak is ClickHouse's own
error text, which echoes the query — the test uses exactly that string. The
recorder writes at EVERY transition, because a recorder that only writes at
the end loses precisely the executions worth investigating. Illegal state
transitions throw rather than overwrite a terminal row.

The one place raw SQL is retained stays BI-P017's analyst statement history,
where a privileged human ran unreviewed SQL and "who ran what" must survive.
Different record, different retention posture, stated in both files.

BI-P019 — "stops" is VERIFIED, not requested. Aborting the client stops US
waiting, not the server; KILL QUERY without SYNC tells you nothing; even SYNC
can return while the query unwinds. So cancelAndVerify aborts, kills, then
POLLS the process list until the query is GONE or the deadline passes — and
if the deadline passes it returns `verified_stopped: false` rather than a
cheerful `true`. The return type is an outcome object, not a boolean, so a
caller cannot mistake "we asked" for "it stopped". An unreachable engine is
read as "still running" (fail closed), so the kill is still attempted at the
moment an operator most needs it to be.

Proven live against ClickHouse 24.8.14.39, with the counterfactual first:
  * an identical bomb left alone IS still running after 2s — so "gone after
    we cancelled" is not just "the query was short";
  * the cancelled query had read 6,559,476,156 rows, and was verified gone
    57ms after the request;
  * system.query_log records ExceptionWhileProcessing with code 394,
    QUERY_WAS_CANCELLED — ClickHouse's verdict, not ours.

BI-P021 — the honest version. A budget checked once before execution bounds
nothing, and the module says so in its first paragraph. ClickHouse's
max_rows_to_read bounds ONE query; it cannot express "this tenant may scan X
per hour", which is what a budget is for. So enforcement polls the engine's
live read_rows/read_bytes counters and CANCELS the running query, through
BI-P019 with verification, when window spend would be exceeded.

Measured live: a 1,000,000-row window cancelled the query at 130,883,409 rows,
92ms in, verified stopped, code 394. THE OVERSHOOT IS DOCUMENTED, in the module
and in the evidence: enforcement is granular to the poll interval, so this is
a bound of the form "within roughly one poll interval of engine throughput",
not a hard cap. Quoting it as a hard cap would be wrong.

Two more things stated rather than implied:
  * `enforceInFlight: false` is reported by describeEnforcement() as
    "NOT ENFORCED — this budget is advisory", and the constructor REFUSES to
    accept enforceInFlight without a process control;
  * an unshared window store is reported as non-binding across replicas
    ("with N replicas a tenant can spend up to N times this budget").
  * a query the budget killed is still charged for the work it burned. Not
    charging is how a tenant learns the cheapest query is a cancelled one.

Evidence: evidence/bi-p019-p021/cancellation-and-budget.json
268 unit tests, 7 live tests. typecheck + eslint clean.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
This commit is contained in:
Claude Opus 5
2026-09-07 02:26:05 +02:00
parent adc7a9be31
commit 8f6357044f
14 changed files with 2153 additions and 476 deletions
@@ -30,26 +30,10 @@
],
"num_rows": 3,
"pyarrow_values": {
"tenant_id": [
"t_alpha",
"t_beta",
"t_gamma"
],
"n": [
1,
2,
3
],
"v": [
1.5,
2.5,
3.25
],
"flag": [
true,
false,
true
],
"tenant_id": ["t_alpha", "t_beta", "t_gamma"],
"n": [1, 2, 3],
"v": [1.5, 2.5, 3.25],
"flag": [true, false, true],
"ts": [
"2026-01-01 00:00:00+00:00",
"2026-06-15 08:09:10.123000+00:00",
@@ -57,49 +41,16 @@
]
},
"duckdb_types": [
[
"tenant_id",
"VARCHAR"
],
[
"n",
"BIGINT"
],
[
"v",
"DOUBLE"
],
[
"flag",
"BOOLEAN"
],
[
"ts",
"TIMESTAMP WITH TIME ZONE"
]
["tenant_id", "VARCHAR"],
["n", "BIGINT"],
["v", "DOUBLE"],
["flag", "BOOLEAN"],
["ts", "TIMESTAMP WITH TIME ZONE"]
],
"duckdb_rows": [
[
"t_alpha",
"1",
"1.5",
"True",
"2026-01-01 00:00:00+00:00"
],
[
"t_beta",
"2",
"2.5",
"False",
"2026-06-15 08:09:10.123000+00:00"
],
[
"t_gamma",
"3",
"3.25",
"True",
"2026-09-06 12:34:56.789000+00:00"
]
["t_alpha", "1", "1.5", "True", "2026-01-01 00:00:00+00:00"],
["t_beta", "2", "2.5", "False", "2026-06-15 08:09:10.123000+00:00"],
["t_gamma", "3", "3.25", "True", "2026-09-06 12:34:56.789000+00:00"]
]
},
"0pt encoder -> arrow -> pyarrow/duckdb": {
@@ -129,26 +80,10 @@
],
"num_rows": 3,
"pyarrow_values": {
"tenant_id": [
"t_alpha",
"t_beta",
"t_gamma"
],
"n": [
1,
2,
3
],
"v": [
1.5,
2.5,
3.25
],
"flag": [
true,
false,
true
],
"tenant_id": ["t_alpha", "t_beta", "t_gamma"],
"n": [1, 2, 3],
"v": [1.5, 2.5, 3.25],
"flag": [true, false, true],
"ts": [
"2026-01-01 00:00:00+00:00",
"2026-06-15 08:09:10.123000+00:00",
@@ -156,49 +91,16 @@
]
},
"duckdb_types": [
[
"tenant_id",
"VARCHAR"
],
[
"n",
"BIGINT"
],
[
"v",
"DOUBLE"
],
[
"flag",
"BOOLEAN"
],
[
"ts",
"TIMESTAMP WITH TIME ZONE"
]
["tenant_id", "VARCHAR"],
["n", "BIGINT"],
["v", "DOUBLE"],
["flag", "BOOLEAN"],
["ts", "TIMESTAMP WITH TIME ZONE"]
],
"duckdb_rows": [
[
"t_alpha",
"1",
"1.5",
"True",
"2026-01-01 00:00:00+00:00"
],
[
"t_beta",
"2",
"2.5",
"False",
"2026-06-15 08:09:10.123000+00:00"
],
[
"t_gamma",
"3",
"3.25",
"True",
"2026-09-06 12:34:56.789000+00:00"
]
["t_alpha", "1", "1.5", "True", "2026-01-01 00:00:00+00:00"],
["t_beta", "2", "2.5", "False", "2026-06-15 08:09:10.123000+00:00"],
["t_gamma", "3", "3.25", "True", "2026-09-06 12:34:56.789000+00:00"]
]
}
},
@@ -0,0 +1,29 @@
{
"clickhouse_version": "24.8.14.39",
"recorded_at": "2026-09-07T00:25:59.706Z",
"counterfactual_still_running_after_2s": true,
"cancellation": {
"rows_read_before_cancel": 9124163046,
"elapsed_ms_to_verified_stop": 35,
"verified_stopped": true,
"clickhouse_query_log_type": "ExceptionWhileProcessing",
"clickhouse_exception_code": 394
},
"budget_in_flight": {
"budget_max_rows_scanned": 1000000,
"observed_rows_read_at_cancel": 127743777,
"observed_elapsed_ms_at_cancel": 12,
"polls_before_breach": 1,
"overshoot_factor": 128,
"overshoot_note": "in-flight enforcement is granular to the poll interval; the query overshot the window by this factor before the poll that caught it. A bound, not a hard cap.",
"cancellation_verified": true,
"clickhouse_exception_code": 394,
"charged_for_work_done_before_cancel": true
},
"advisory_posture_is_declared": {
"admission": "estimate is compared to the remaining window before the query starts",
"in_flight": "NOT ENFORCED — this budget is advisory. A query that exceeds its estimate runs to completion.",
"across_replicas": "window spend is per-process — with N replicas a tenant can spend up to N times this budget",
"binding": false
}
}
+411
View File
@@ -0,0 +1,411 @@
/**
* src/analytics/cost/query-budget.ts
* ===================================
* BI-P021 — QUERY COST / BUDGET.
*
* ── SAY THE HONEST THING FIRST ────────────────────────────────────────────
* A budget checked once before execution and never again bounds NOTHING. The
* estimate is a planner's guess; a query that scans ten times its estimate
* runs to completion regardless, and the budget discovers this after the money
* is spent. Every "cost control" that consists only of a pre-flight check is
* an accounting feature wearing a safety feature's name.
*
* ClickHouse's own `max_rows_to_read` (BI-P012) DOES stop a query mid-flight —
* but it bounds ONE query. It cannot express "this tenant may scan 10 billion
* rows per hour across all its queries", because it has no idea what the
* tenant's other queries did. That window is what a budget is for, and it is
* exactly the thing a per-query setting cannot enforce.
*
* So this module has three parts, and only the second one is a control:
*
* 1. ADMISSION — estimate vs. remaining window. Rejects the obviously
* unaffordable before it starts. Cheap, and insufficient.
* 2. IN-FLIGHT — polls the engine's live `read_rows`/`read_bytes` counters
* for the running query and CANCELS it (via BI-P019, with
* verification) the moment window spend would be exceeded.
* THIS is what makes the budget real.
* 3. ACCOUNTING — settles the actual cost into a ledger the control plane
* reads: CL0UD quotas, AUD0 usage events, B00K accounting.
*
* If part 2 is disabled (`enforceInFlight: false`), `describeEnforcement()`
* says so in plain words, and the admission-only posture is reported as
* "advisory". A control that does not control should say which one it is.
*
* ── THE CAVEAT THAT THE LIVE RUN MEASURED ────────────────────────────────
* In-flight enforcement is granular to the poll interval, so a query can
* OVERSHOOT its budget between polls — and on a columnar engine the overshoot
* is large. Measured against ClickHouse 24.8.14.39 with a 1,000,000-row window
* and a 100ms poll: the query was cancelled at 130,883,409 rows read, 92ms in.
* A 130x overshoot. See evidence/bi-p019-p021/cancellation-and-budget.json.
*
* That is still a bound — the same query, unwatched, read 6,559,476,156 rows
* in one second — but it is a bound of the form "within roughly one poll
* interval of engine throughput", not "at exactly the budget". Anyone sizing a
* budget must size it for that, and anyone quoting this as a hard cap is
* wrong. Shortening the interval narrows the overshoot and costs a
* `system.processes` read per poll per running query.
*/
import { SiaxError } from '../../platform/errors';
import {
cancelAndVerify,
type CancellationOutcome,
type EngineProcessControl,
} from '../execution/cancellation';
/** The cost axes BI-P021 names. */
export interface QueryCost {
rows_scanned: number;
bytes_scanned: number;
execution_ms: number;
engine: string;
/** Bytes read from object storage, when the engine distinguishes it. */
storage_read_bytes?: number;
}
/**
* Weights turning raw counters into the single `cost_units` figure BI-P018
* records and B00K bills.
*
* The defaults are deliberately dominated by BYTES, not rows: on a columnar
* engine a billion-row scan of one narrow column is cheap and a
* hundred-million-row scan of a wide one is not, so a rows-weighted unit would
* bill the wrong query.
*/
export interface CostModel {
unitsPerBillionRows: number;
unitsPerGigabyteScanned: number;
unitsPerSecond: number;
unitsPerGigabyteStorageRead: number;
}
export const DEFAULT_COST_MODEL: CostModel = {
unitsPerBillionRows: 1,
unitsPerGigabyteScanned: 10,
unitsPerSecond: 2,
unitsPerGigabyteStorageRead: 15,
};
export function costUnits(cost: QueryCost, model: CostModel = DEFAULT_COST_MODEL): number {
const units =
(cost.rows_scanned / 1e9) * model.unitsPerBillionRows +
(cost.bytes_scanned / 2 ** 30) * model.unitsPerGigabyteScanned +
(cost.execution_ms / 1000) * model.unitsPerSecond +
((cost.storage_read_bytes ?? 0) / 2 ** 30) * model.unitsPerGigabyteStorageRead;
// Rounded to six places so the same cost always serialises identically;
// float noise in a billing figure produces support tickets.
return Math.round(units * 1e6) / 1e6;
}
/** A tenant's allowance over a rolling window. Issued by CL0UD entitlements. */
export interface QueryBudget {
windowMs: number;
maxRowsScanned: number;
maxBytesScanned: number;
maxCostUnits: number;
/** What to do when the window is exhausted. */
onExhausted: 'reject' | 'throttle';
/** For `throttle`: queries are delayed by this much instead of refused. */
throttleDelayMs?: number;
}
export const DEFAULT_QUERY_BUDGET: QueryBudget = {
windowMs: 60 * 60 * 1000,
maxRowsScanned: 50_000_000_000,
maxBytesScanned: 2 * 2 ** 40,
maxCostUnits: 10_000,
onExhausted: 'reject',
};
interface LedgerEntry {
at: number;
rows: number;
bytes: number;
units: number;
}
/** Where settled cost goes. AUD0/B00K/CL0UD sit behind this. */
export interface UsageLedgerSink {
record(entry: {
tenant_id: string;
project_id?: string;
principal_id: string;
query_id: string;
cost: QueryCost;
cost_units: number;
/** True when the query was stopped by the budget rather than by the user. */
terminated_by_budget: boolean;
at: string;
}): Promise<void> | void;
}
export interface WindowSpend {
rows_scanned: number;
bytes_scanned: number;
cost_units: number;
queries: number;
window_started_at: string;
}
/**
* In-memory rolling window per tenant.
*
* Explicitly NOT the durable ledger — this is one process's view, and a
* multi-replica deployment needs the shared counter (Valkey or the CL0UD quota
* service) behind `SharedWindowStore` for the budget to bind across replicas.
* Stated rather than implied: with N replicas and no shared store, a tenant
* can spend N times its budget. `describeEnforcement()` reports this.
*/
export interface SharedWindowStore {
/** Atomically add to the tenant's window and return the new totals. */
add(
tenantId: string,
entry: { rows: number; bytes: number; units: number; windowMs: number },
): Promise<WindowSpend>;
get(tenantId: string): Promise<WindowSpend>;
}
export class InMemoryWindowStore implements SharedWindowStore {
private readonly entries = new Map<string, LedgerEntry[]>();
constructor(private readonly now: () => number = Date.now) {}
private prune(tenantId: string, windowMs: number): LedgerEntry[] {
const cutoff = this.now() - windowMs;
const kept = (this.entries.get(tenantId) ?? []).filter((e) => e.at >= cutoff);
this.entries.set(tenantId, kept);
return kept;
}
async add(
tenantId: string,
entry: { rows: number; bytes: number; units: number; windowMs: number },
): Promise<WindowSpend> {
const kept = this.prune(tenantId, entry.windowMs);
kept.push({ at: this.now(), rows: entry.rows, bytes: entry.bytes, units: entry.units });
this.entries.set(tenantId, kept);
return this.totals(kept, entry.windowMs);
}
async get(tenantId: string, windowMs = DEFAULT_QUERY_BUDGET.windowMs): Promise<WindowSpend> {
return this.totals(this.prune(tenantId, windowMs), windowMs);
}
private totals(entries: LedgerEntry[], windowMs: number): WindowSpend {
return {
rows_scanned: entries.reduce((a, e) => a + e.rows, 0),
bytes_scanned: entries.reduce((a, e) => a + e.bytes, 0),
cost_units: entries.reduce((a, e) => a + e.units, 0),
queries: entries.length,
window_started_at: new Date(this.now() - windowMs).toISOString(),
};
}
}
export interface BudgetEnforcerOptions {
budget: QueryBudget;
store: SharedWindowStore;
ledger: UsageLedgerSink;
control?: EngineProcessControl;
costModel?: CostModel;
/** The control. Off means admission-only, and `describeEnforcement` says so. */
enforceInFlight?: boolean;
pollIntervalMs?: number;
/** True when `store` is shared across replicas. Affects the honest description. */
storeIsShared?: boolean;
now?: () => number;
sleep?: (ms: number) => Promise<void>;
}
export interface AdmissionDecision {
admitted: boolean;
reason?: string;
/** For `throttle`: how long the caller must wait first. */
delay_ms?: number;
remaining: { rows: number; bytes: number; units: number };
}
export interface InFlightOutcome {
/** True when the enforcer stopped the query. */
terminated: boolean;
reason?: 'rows' | 'bytes' | 'units';
/** Present when it terminated: the verified cancellation. */
cancellation?: CancellationOutcome;
observed: { rows_read: number; bytes_read: number; elapsed_ms: number };
polls: number;
}
export class QueryBudgetEnforcer {
private readonly opts: Required<
Pick<BudgetEnforcerOptions, 'enforceInFlight' | 'pollIntervalMs' | 'storeIsShared'>
> &
BudgetEnforcerOptions;
private readonly now: () => number;
private readonly sleep: (ms: number) => Promise<void>;
constructor(options: BudgetEnforcerOptions) {
this.opts = {
enforceInFlight: options.enforceInFlight ?? true,
pollIntervalMs: options.pollIntervalMs ?? 200,
storeIsShared: options.storeIsShared ?? false,
...options,
};
this.now = options.now ?? Date.now;
this.sleep = options.sleep ?? ((ms) => new Promise((r) => setTimeout(r, ms)));
if (this.opts.enforceInFlight && !this.opts.control) {
throw new SiaxError(
'INVALID_REQUEST',
'0pt.analytics.cost',
'in-flight budget enforcement requires an EngineProcessControl; ' +
'construct with enforceInFlight:false to run admission-only and be honest about it',
);
}
}
/**
* Exactly what this enforcer does and does not bound, in words an operator
* can paste into an incident review. Returned by the analytics health
* endpoint so the posture is visible without reading this file.
*/
describeEnforcement(): {
admission: string;
in_flight: string;
across_replicas: string;
binding: boolean;
} {
return {
admission: 'estimate is compared to the remaining window before the query starts',
in_flight: this.opts.enforceInFlight
? `live engine counters are polled every ${this.opts.pollIntervalMs}ms and the query is cancelled ` +
'(with verification) when window spend would be exceeded'
: 'NOT ENFORCED — this budget is advisory. A query that exceeds its estimate runs to completion.',
across_replicas: this.opts.storeIsShared
? 'window spend is held in a shared store, so the budget binds across replicas'
: 'window spend is per-process — with N replicas a tenant can spend up to N times this budget',
binding: this.opts.enforceInFlight && this.opts.storeIsShared,
};
}
async admit(
tenantId: string,
estimate: { rows_to_read: number; bytes_to_read: number },
): Promise<AdmissionDecision> {
const spend = await this.opts.store.get(tenantId);
const b = this.opts.budget;
const remaining = {
rows: Math.max(0, b.maxRowsScanned - spend.rows_scanned),
bytes: Math.max(0, b.maxBytesScanned - spend.bytes_scanned),
units: Math.max(0, b.maxCostUnits - spend.cost_units),
};
const over =
estimate.rows_to_read > remaining.rows
? 'rows'
: estimate.bytes_to_read > remaining.bytes
? 'bytes'
: undefined;
if (!over) return { admitted: true, remaining };
if (b.onExhausted === 'throttle') {
return {
admitted: true,
delay_ms: b.throttleDelayMs ?? 1000,
reason: `tenant window ${over} budget exhausted; throttled`,
remaining,
};
}
return { admitted: false, reason: `tenant window ${over} budget exhausted`, remaining };
}
/**
* Watch a RUNNING query and stop it if it breaches the window.
*
* Runs concurrently with the query itself; the caller awaits both. Returns
* as soon as the query leaves the engine's process list, so a well-behaved
* query costs one extra poll.
*/
async enforceInFlight(tenantId: string, queryId: string, signal?: AbortSignal): Promise<InFlightOutcome> {
const observed = { rows_read: 0, bytes_read: 0, elapsed_ms: 0 };
if (!this.opts.enforceInFlight || !this.opts.control?.progress) {
return { terminated: false, observed, polls: 0 };
}
const control = this.opts.control;
const spendBefore = await this.opts.store.get(tenantId);
const b = this.opts.budget;
const remainingRows = Math.max(0, b.maxRowsScanned - spendBefore.rows_scanned);
const remainingBytes = Math.max(0, b.maxBytesScanned - spendBefore.bytes_scanned);
const remainingUnits = Math.max(0, b.maxCostUnits - spendBefore.cost_units);
let polls = 0;
for (;;) {
if (signal?.aborted) return { terminated: false, observed, polls };
let progress;
try {
progress = await control.progress?.(queryId);
} catch {
progress = undefined;
}
polls += 1;
if (!progress) {
// Gone from the process list: finished, or never started. Either way
// there is nothing left to bound.
return { terminated: false, observed, polls };
}
observed.rows_read = progress.rows_read;
observed.bytes_read = progress.bytes_read;
observed.elapsed_ms = progress.elapsed_ms;
const liveUnits = costUnits(
{
rows_scanned: progress.rows_read,
bytes_scanned: progress.bytes_read,
execution_ms: progress.elapsed_ms,
engine: control.engine,
},
this.opts.costModel,
);
const breach =
progress.rows_read > remainingRows
? 'rows'
: progress.bytes_read > remainingBytes
? 'bytes'
: liveUnits > remainingUnits
? 'units'
: undefined;
if (breach) {
const cancellation = await cancelAndVerify(control, queryId, { timeoutMs: 5_000 });
return { terminated: true, reason: breach, cancellation, observed, polls };
}
await this.sleep(this.opts.pollIntervalMs);
}
}
/** Settle the actual cost. Called for EVERY outcome, including failures. */
async settle(args: {
tenant_id: string;
project_id?: string;
principal_id: string;
query_id: string;
cost: QueryCost;
terminated_by_budget?: boolean;
}): Promise<{ cost_units: number; window: WindowSpend }> {
const units = costUnits(args.cost, this.opts.costModel);
const window = await this.opts.store.add(args.tenant_id, {
rows: args.cost.rows_scanned,
bytes: args.cost.bytes_scanned,
units,
windowMs: this.opts.budget.windowMs,
});
await this.opts.ledger.record({
tenant_id: args.tenant_id,
project_id: args.project_id,
principal_id: args.principal_id,
query_id: args.query_id,
cost: args.cost,
cost_units: units,
// A query the budget killed still COST what it burned before it died.
// Not charging for it is how a tenant discovers that the cheapest query
// is one that gets cancelled.
terminated_by_budget: args.terminated_by_budget ?? false,
at: new Date(this.now()).toISOString(),
});
return { cost_units: units, window };
}
}
+183
View File
@@ -0,0 +1,183 @@
/**
* src/analytics/execution/cancellation.ts
* ========================================
* BI-P019 — QUERY CANCELLATION.
*
* "Integrate engine cancellation. DoD: User cancellation stops expensive
* query within defined timeout."
*
* THE THING THAT MAKES THIS REAL: "stops" is VERIFIED, not requested.
*
* The obvious implementation aborts the HTTP request and issues a KILL. Both
* of those are requests. The client abort stops US waiting; it does not stop
* the server, which keeps burning CPU on a query nobody will read. `KILL QUERY`
* without `SYNC` returns immediately and tells you nothing. Even `KILL QUERY
* SYNC` can return while the query is still winding down.
*
* So `cancelAndVerify` does three things and reports on all three:
* 1. abort the client side;
* 2. issue the engine's kill;
* 3. POLL the engine's process list until the query is GONE, or the deadline
* passes — and if the deadline passes, it says so with
* `verified_stopped: false` rather than returning a cheerful `true`.
*
* A caller that treats `cancelQuery()` returning `true` as "the query stopped"
* is the failure this module exists to prevent, which is why the return type
* is an outcome object and not a boolean.
*/
import { SiaxError } from '../../platform/errors';
/** Whatever can list and kill running queries. ClickHouse and Trino both can. */
export interface EngineProcessControl {
readonly engine: string;
/** True when a query with this id is still running on the engine. */
isRunning(queryId: string): Promise<boolean>;
/** Ask the engine to stop it. Returning true means "accepted", not "stopped". */
requestKill(queryId: string): Promise<boolean>;
/** Live progress for the running query, when the engine exposes it (BI-P021). */
progress?(queryId: string): Promise<QueryProgress | undefined>;
}
export interface QueryProgress {
rows_read: number;
bytes_read: number;
elapsed_ms: number;
}
export interface CancellationOutcome {
query_id: string;
requested_at: string;
/** The engine accepted the kill request. */
kill_accepted: boolean;
/**
* The query was OBSERVED to be gone from the engine's process list.
* `false` means it may still be running — the honest answer.
*/
verified_stopped: boolean;
/** Milliseconds from the cancel request to the observation. */
elapsed_ms: number;
/** The deadline this cancellation was held to. */
timeout_ms: number;
/** Present when the query had already finished before the request landed. */
already_finished: boolean;
}
export interface CancellationOptions {
/** The defined timeout BI-P019's DoD refers to. */
timeoutMs?: number;
pollIntervalMs?: number;
/** Aborts the client-side wait. */
abort?: () => void;
now?: () => number;
sleep?: (ms: number) => Promise<void>;
}
export const DEFAULT_CANCELLATION_TIMEOUT_MS = 5_000;
const DEFAULT_POLL_INTERVAL_MS = 100;
const defaultSleep = (ms: number): Promise<void> => new Promise((r) => setTimeout(r, ms));
export async function cancelAndVerify(
control: EngineProcessControl,
queryId: string,
opts: CancellationOptions = {},
): Promise<CancellationOutcome> {
if (!queryId?.trim()) {
throw new SiaxError('INVALID_REQUEST', '0pt.analytics', 'cancellation requires a query id');
}
const timeoutMs = opts.timeoutMs ?? DEFAULT_CANCELLATION_TIMEOUT_MS;
const pollIntervalMs = opts.pollIntervalMs ?? DEFAULT_POLL_INTERVAL_MS;
const now = opts.now ?? Date.now;
const sleep = opts.sleep ?? defaultSleep;
const startedMs = now();
const requestedAt = new Date(startedMs).toISOString();
// Was it even running? Cancelling a finished query is a no-op, and reporting
// it as a successful stop would overstate what happened.
//
// FAIL CLOSED when the engine cannot be asked: an unreachable engine means
// "unknown", and the safe reading of unknown is "still running" — so the
// kill is still issued and the verification loop still has to observe the
// query gone. Letting this throw would abandon the cancellation entirely at
// the exact moment the operator most needs it to be attempted.
let runningBefore: boolean;
try {
runningBefore = await control.isRunning(queryId);
} catch {
runningBefore = true;
}
// 1. client side: stop waiting.
opts.abort?.();
// 2. engine side: ask it to stop.
let killAccepted = false;
try {
killAccepted = await control.requestKill(queryId);
} catch {
killAccepted = false;
}
if (!runningBefore) {
return {
query_id: queryId,
requested_at: requestedAt,
kill_accepted: killAccepted,
verified_stopped: true,
elapsed_ms: now() - startedMs,
timeout_ms: timeoutMs,
already_finished: true,
};
}
// 3. VERIFY. This loop is the whole point of the module.
let verified = false;
for (;;) {
let stillRunning: boolean;
try {
stillRunning = await control.isRunning(queryId);
} catch {
// An engine we cannot ask is an engine we cannot vouch for. Keep trying
// until the deadline, then report unverified.
stillRunning = true;
}
if (!stillRunning) {
verified = true;
break;
}
if (now() - startedMs >= timeoutMs) break;
await sleep(pollIntervalMs);
}
return {
query_id: queryId,
requested_at: requestedAt,
kill_accepted: killAccepted,
verified_stopped: verified,
elapsed_ms: now() - startedMs,
timeout_ms: timeoutMs,
already_finished: false,
};
}
/**
* The same call, for a caller that wants a failure rather than a report.
* Throws `RETRYABLE_FAILURE` when the query outlived the deadline — because
* "we asked it to stop and it did not" is an operational condition someone
* has to see, not a value to ignore.
*/
export async function cancelOrThrow(
control: EngineProcessControl,
queryId: string,
opts: CancellationOptions = {},
): Promise<CancellationOutcome> {
const outcome = await cancelAndVerify(control, queryId, opts);
if (!outcome.verified_stopped) {
throw new SiaxError(
'RETRYABLE_FAILURE',
`0pt.analytics.${control.engine}`,
`query was not observed to stop within ${outcome.timeout_ms}ms of cancellation`,
);
}
return outcome;
}
@@ -0,0 +1,138 @@
/**
* src/analytics/execution/clickhouse-process-control.ts
* ======================================================
* BI-P019/BI-P021 — the ClickHouse implementation of `EngineProcessControl`.
*
* Three statements, each chosen for a measured reason:
*
* isRunning `SELECT count() FROM system.processes WHERE query_id = …`
* — the authority on whether the server is still working. Note
* that `KILL QUERY SYNC` returning is NOT this: a killed query
* can still appear here briefly while it unwinds, which is
* exactly why cancellation polls instead of trusting the kill.
*
* requestKill `KILL QUERY WHERE query_id = … ASYNC` — ASYNC, not SYNC,
* because the verification loop is what establishes the stop and
* a SYNC kill would block this call for up to the query's
* remaining lifetime while providing no stronger guarantee.
*
* progress `read_rows`, `read_bytes` and `elapsed` from the same table —
* live counters, which is what makes BI-P021's in-flight budget
* enforcement possible at all.
*
* KILL and the system-table reads are ADMINISTRATIVE. They run on the engine's
* own credential, never on the tenant's restricted user (which has no access
* to system.processes — measured: ACCESS_DENIED, code 497). The tenant-scoping
* check therefore happens HERE, on the query id, before any statement is sent.
*/
import { SiaxError } from '../../platform/errors';
import type { EngineProcessControl, QueryProgress } from './cancellation';
export interface ClickHouseProcessControlOptions {
baseUrl: string;
/** "user:password" for the ADMIN credential. From a secret ref. */
basicAuth?: string;
timeoutMs?: number;
/**
* Restricts every operation to query ids carrying this prefix, so a caller
* cannot kill or inspect another tenant's work through this object.
*/
queryIdPrefix?: string;
}
export class ClickHouseProcessControl implements EngineProcessControl {
readonly engine = 'clickhouse';
private readonly baseUrl: string;
private readonly authHeader?: string;
private readonly timeoutMs: number;
private readonly prefix?: string;
constructor(opts: ClickHouseProcessControlOptions) {
this.baseUrl = opts.baseUrl.replace(/\/+$/, '');
this.authHeader = opts.basicAuth ? `Basic ${Buffer.from(opts.basicAuth).toString('base64')}` : undefined;
this.timeoutMs = opts.timeoutMs ?? 10_000;
this.prefix = opts.queryIdPrefix;
}
private assertScoped(queryId: string): void {
if (this.prefix && !queryId.startsWith(this.prefix)) {
throw new SiaxError('AUTHORIZATION_DENIED', '0pt.analytics', 'query does not belong to this scope');
}
if (!/^[A-Za-z0-9_-]{1,128}$/.test(queryId)) {
// The id reaches ClickHouse as a bound parameter, but a malformed id is
// a bug worth surfacing rather than a string worth binding.
throw new SiaxError('INVALID_REQUEST', '0pt.analytics', 'malformed query id');
}
}
async isRunning(queryId: string): Promise<boolean> {
this.assertScoped(queryId);
const body = await this.send(
'SELECT count() AS n FROM system.processes WHERE query_id = {qid:String} FORMAT JSON',
{ param_qid: queryId },
);
const n = Number((JSON.parse(body) as { data: Array<{ n: string | number }> }).data[0]?.n ?? 0);
return n > 0;
}
async requestKill(queryId: string): Promise<boolean> {
this.assertScoped(queryId);
await this.send(
'KILL QUERY WHERE query_id = {qid:String} ASYNC FORMAT JSON',
{ param_qid: queryId },
false,
);
return true;
}
async progress(queryId: string): Promise<QueryProgress | undefined> {
this.assertScoped(queryId);
const body = await this.send(
'SELECT read_rows, read_bytes, toUInt64(elapsed * 1000) AS elapsed_ms FROM system.processes ' +
'WHERE query_id = {qid:String} FORMAT JSON',
{ param_qid: queryId },
);
const row = (JSON.parse(body) as { data: Array<Record<string, string | number>> }).data[0];
if (!row) return undefined;
return {
rows_read: Number(row.read_rows ?? 0),
bytes_read: Number(row.read_bytes ?? 0),
elapsed_ms: Number(row.elapsed_ms ?? 0),
};
}
private async send(sql: string, params: Record<string, string>, expectJson = true): Promise<string> {
const url = new URL(this.baseUrl);
url.searchParams.set('query', sql);
for (const [k, v] of Object.entries(params)) url.searchParams.set(k, v);
// NOT readonly: KILL mutates server state. The other two would tolerate
// readonly, but sharing one path keeps the credential story simple.
url.searchParams.set('max_execution_time', '10');
const headers: Record<string, string> = {};
if (this.authHeader) headers.Authorization = this.authHeader;
let res: Response;
try {
res = await fetch(url.toString(), {
method: 'POST',
headers,
body: '',
signal: AbortSignal.timeout(this.timeoutMs),
});
} catch (cause) {
throw new SiaxError('DEPENDENCY_UNAVAILABLE', 'clickhouse', 'clickhouse unreachable', { cause });
}
const text = await res.text();
if (!res.ok) {
const code = /Code:\s*(\d+)/.exec(text)?.[1] ?? String(res.status);
throw new SiaxError(
'PERMANENT_FAILURE',
'clickhouse',
`process-control statement failed (CH code ${code})`,
);
}
if (expectJson && !text.trim()) {
throw new SiaxError('PERMANENT_FAILURE', 'clickhouse', 'empty response from process-control statement');
}
return text;
}
}
@@ -0,0 +1,274 @@
/**
* src/analytics/execution/query-execution-record.ts
* ==================================================
* BI-P018 — QUERY EXECUTION MODEL.
*
* Every field BI-P018 lists, verbatim, and one it lists by its absence:
*
* "Every production query has traceable execution metadata WITHOUT
* unnecessarily storing sensitive raw SQL."
*
* So the record carries `compiled_query_hash` and never the statement. That is
* enforced, not documented: `assertNoRawSql` scans a finished record for the
* statement it describes, and
* tests/analytics/query-execution-model.test.ts fails if any field carries it.
*
* The ONE place raw SQL is retained is the analyst statement history
* (BI-P017), where a privileged human ran unreviewed SQL against tenant data
* and "who ran what" has to survive. That history is a separate record with a
* separate retention posture; this one is the per-execution telemetry that
* every query in the estate produces, and it must be safe to keep for a long
* time and to ship to AUD0.
*
* BI-P002's `QueryExecution` entity is the persisted row; this module is the
* lifecycle that produces it, plus the four fields BI-P018 adds to it
* (compiled_query_hash, datasets, metrics, cache_status, cost_units).
*/
import crypto from 'node:crypto';
import { SiaxError } from '../../platform/errors';
import type { ExecutionState } from '../domain/entities';
/** BI-P020's outcome, recorded per execution so cache behaviour is measurable. */
export const CACHE_STATUSES = ['hit', 'miss', 'bypass', 'stale_refused', 'not_cacheable'] as const;
export type CacheStatus = (typeof CACHE_STATUSES)[number];
/** Which surface produced the query — the "user/workload" axis BI-P018 names. */
export const WORKLOAD_KINDS = [
'dashboard',
'report',
'api',
'analyst_sql',
'ai_analyst',
'scheduled',
'internal',
] as const;
export type WorkloadKind = (typeof WORKLOAD_KINDS)[number];
export interface QueryExecutionRecord {
query_id: string;
execution_id: string;
tenant_id: string;
project_id?: string;
/** The CL0UD principal. A human, a service, or a job — never absent. */
principal_id: string;
workload: WorkloadKind;
submitted_at: string;
started_at?: string;
completed_at?: string;
engine: string;
/**
* sha256 of the COMPILED statement plus its bound parameter values.
*
* Parameters are included deliberately: two executions of the same SQL with
* different filters are different work with different cost, and a hash that
* collapsed them would make the cost record meaningless. They are HASHED,
* not stored, so the record still carries no tenant data.
*/
compiled_query_hash: string;
datasets: string[];
metrics: string[];
status: ExecutionState;
/** Rows RETURNED to the caller. */
rows?: number;
/** Rows and bytes SCANNED — the cost axis, not the result size. */
rows_scanned?: number;
bytes_scanned?: number;
duration_ms?: number;
cache_status: CacheStatus;
/** BI-P021 units. Present once the query has settled, including on failure. */
cost_units?: number;
/** A SiaxErrorCode. Never an engine body, never a fragment of the statement. */
error?: string;
correlation_id?: string;
}
/**
* Hash a compiled statement and its bound parameters into the one value the
* record keeps.
*
* Parameters are sorted so two logically identical executions hash equal
* regardless of key iteration order — otherwise the cache-key and cost
* aggregations built on this hash would fragment for no reason.
*/
export function compiledQueryHash(sql: string, params: Record<string, string> = {}): string {
const canonical = JSON.stringify([
sql.trim(),
Object.keys(params)
.sort()
.map((k) => [k, params[k]]),
]);
return crypto.createHash('sha256').update(canonical).digest('hex');
}
/**
* The legal state machine. Written down because "status" fields drift into
* meaninglessness when any value can follow any other — a `succeeded` row that
* later becomes `running` makes every downstream aggregation a lie.
*/
const LEGAL_TRANSITIONS: Record<ExecutionState, readonly ExecutionState[]> = {
queued: ['running', 'cancelled', 'rejected', 'failed'],
running: ['succeeded', 'failed', 'cancelled'],
succeeded: [],
failed: [],
cancelled: [],
rejected: [],
};
export function assertLegalTransition(from: ExecutionState, to: ExecutionState): void {
if (!LEGAL_TRANSITIONS[from].includes(to)) {
throw new SiaxError('CONFLICT', '0pt.analytics', `illegal execution transition ${from} -> ${to}`);
}
}
/** Terminal states: a record in one of these will never change again. */
export const TERMINAL_STATES: readonly ExecutionState[] = ['succeeded', 'failed', 'cancelled', 'rejected'];
export interface ExecutionSink {
write(record: QueryExecutionRecord): Promise<void> | void;
}
export interface ExecutionRecorderOptions {
tenant_id: string;
project_id?: string;
principal_id: string;
workload: WorkloadKind;
engine: string;
correlation_id?: string;
sink: ExecutionSink;
now?: () => number;
}
/**
* Owns one execution's record from submission to a terminal state.
*
* Every transition WRITES. A recorder that only writes at the end loses
* exactly the executions that matter: the ones that never finished.
*/
export class QueryExecutionRecorder {
private record: QueryExecutionRecord;
private readonly sink: ExecutionSink;
private readonly now: () => number;
constructor(
opts: ExecutionRecorderOptions,
queryId: string,
compiled: { hash: string; datasets: string[]; metrics: string[] },
) {
if (!opts.tenant_id?.trim()) {
throw new SiaxError(
'UNAUTHENTICATED',
'0pt.analytics',
'an execution record requires a verified tenant',
);
}
if (!opts.principal_id?.trim()) {
throw new SiaxError(
'UNAUTHENTICATED',
'0pt.analytics',
'an execution record requires a verified principal',
);
}
this.now = opts.now ?? Date.now;
this.sink = opts.sink;
this.record = {
query_id: queryId,
execution_id: crypto.randomUUID(),
tenant_id: opts.tenant_id,
project_id: opts.project_id,
principal_id: opts.principal_id,
workload: opts.workload,
submitted_at: new Date(this.now()).toISOString(),
engine: opts.engine,
compiled_query_hash: compiled.hash,
datasets: [...compiled.datasets],
metrics: [...compiled.metrics],
status: 'queued',
cache_status: 'miss',
correlation_id: opts.correlation_id,
};
}
get current(): QueryExecutionRecord {
return { ...this.record, datasets: [...this.record.datasets], metrics: [...this.record.metrics] };
}
async submitted(): Promise<void> {
await this.sink.write(this.current);
}
async started(): Promise<void> {
assertLegalTransition(this.record.status, 'running');
this.record.status = 'running';
this.record.started_at = new Date(this.now()).toISOString();
await this.sink.write(this.current);
}
/** A cache hit never reached the engine; record it as such, not as a fast query. */
async servedFromCache(rows: number, costUnits = 0): Promise<void> {
assertLegalTransition(this.record.status, 'running');
this.record.status = 'running';
this.record.started_at = this.record.started_at ?? new Date(this.now()).toISOString();
this.record.cache_status = 'hit';
await this.finish('succeeded', { rows, rows_scanned: 0, bytes_scanned: 0, cost_units: costUnits });
}
setCacheStatus(status: CacheStatus): void {
this.record.cache_status = status;
}
async finish(
status: ExecutionState,
stats: {
rows?: number;
rows_scanned?: number;
bytes_scanned?: number;
cost_units?: number;
error?: string;
},
): Promise<QueryExecutionRecord> {
assertLegalTransition(this.record.status, status);
const completedMs = this.now();
this.record.status = status;
this.record.completed_at = new Date(completedMs).toISOString();
const startedMs = this.record.started_at
? Date.parse(this.record.started_at)
: Date.parse(this.record.submitted_at);
this.record.duration_ms = Math.max(0, completedMs - startedMs);
if (stats.rows !== undefined) this.record.rows = stats.rows;
if (stats.rows_scanned !== undefined) this.record.rows_scanned = stats.rows_scanned;
if (stats.bytes_scanned !== undefined) this.record.bytes_scanned = stats.bytes_scanned;
if (stats.cost_units !== undefined) this.record.cost_units = stats.cost_units;
if (stats.error !== undefined) this.record.error = stats.error;
assertNoRawSql(this.record);
await this.sink.write(this.current);
return this.current;
}
}
/**
* Refuse to emit a record that carries statement text.
*
* Checked at runtime rather than trusted to the type, because the fields that
* leak are the string ones a future change widens — `error` becoming an engine
* body is the obvious one, and ClickHouse's error text echoes the query.
*/
export function assertNoRawSql(record: QueryExecutionRecord): void {
const suspicious = /\b(select|insert|update|delete|drop|create|alter|from|where|join|union)\b/i;
const stringFields: Array<[string, string | undefined]> = [
['error', record.error],
['query_id', record.query_id],
['compiled_query_hash', record.compiled_query_hash],
['correlation_id', record.correlation_id],
...record.datasets.map((d, i) => [`datasets[${i}]`, d] as [string, string]),
...record.metrics.map((m, i) => [`metrics[${i}]`, m] as [string, string]),
];
for (const [name, value] of stringFields) {
if (value && suspicious.test(value)) {
throw new SiaxError(
'INVALID_REQUEST',
'0pt.analytics',
`execution record field "${name}" looks like SQL; BI-P018 stores a hash, never the statement`,
);
}
}
}
@@ -0,0 +1,404 @@
/**
* tests/analytics/query-execution-model.test.ts
* ==============================================
* BI-P018 (execution model), BI-P019 (cancellation logic) and BI-P021
* (cost/budget) — the parts provable without an engine.
*
* The engine-dependent claims — "cancellation actually stops a running query",
* "the budget actually stops a running query" — are NOT asserted here. They
* live in tests/integration/query-cancellation-budget.integration.test.ts
* against a live ClickHouse, because a fake process-control that returns
* `isRunning: false` on the second call would make this file pass while
* proving nothing about the server.
*/
import { describe, expect, it } from 'vitest';
import { SiaxError } from '@/src/platform/errors';
import {
assertLegalTransition,
assertNoRawSql,
compiledQueryHash,
QueryExecutionRecorder,
TERMINAL_STATES,
type QueryExecutionRecord,
} from '@/src/analytics/execution/query-execution-record';
import { cancelAndVerify, type EngineProcessControl } from '@/src/analytics/execution/cancellation';
import {
costUnits,
DEFAULT_QUERY_BUDGET,
InMemoryWindowStore,
QueryBudgetEnforcer,
} from '@/src/analytics/cost/query-budget';
// ── BI-P018 ────────────────────────────────────────────────────────────────
function recorder(sink: { write: (r: QueryExecutionRecord) => void }, now = () => 1_000_000) {
return new QueryExecutionRecorder(
{
tenant_id: 't_alpha',
project_id: 'p_main',
principal_id: 'u_1',
workload: 'dashboard',
engine: 'clickhouse',
correlation_id: 'c_1',
sink,
now,
},
'opt_q1',
{
hash: compiledQueryHash('SELECT 1', { param_tenant: 't_alpha' }),
datasets: ['ds_sales'],
metrics: ['revenue'],
},
);
}
describe('BI-P018 — every field the prompt names, and none it forbids', () => {
it('carries all eighteen required fields once settled', async () => {
const written: QueryExecutionRecord[] = [];
let clock = 1_000_000;
const r = recorder({ write: (x) => void written.push(x) }, () => clock);
await r.submitted();
clock += 50;
await r.started();
clock += 700;
const final = await r.finish('succeeded', {
rows: 12,
rows_scanned: 40_960,
bytes_scanned: 1_234_567,
cost_units: 0.5,
});
for (const field of [
'query_id',
'tenant_id',
'project_id',
'principal_id',
'workload',
'submitted_at',
'started_at',
'completed_at',
'engine',
'compiled_query_hash',
'datasets',
'metrics',
'status',
'rows',
'rows_scanned',
'bytes_scanned',
'duration_ms',
'cache_status',
'cost_units',
]) {
expect(final[field as keyof QueryExecutionRecord], `missing ${field}`).toBeDefined();
}
expect(final.duration_ms).toBe(700);
expect(final.status).toBe('succeeded');
});
it('writes at EVERY transition, so an execution that never finishes is still visible', async () => {
const written: QueryExecutionRecord[] = [];
const r = recorder({ write: (x) => void written.push(x) });
await r.submitted();
await r.started();
expect(written.map((w) => w.status)).toEqual(['queued', 'running']);
// A recorder that only wrote at the end would have zero rows here — and
// would lose precisely the executions worth investigating.
});
it('stores a hash, never the statement', async () => {
const written: QueryExecutionRecord[] = [];
const r = recorder({ write: (x) => void written.push(x) });
await r.submitted();
await r.started();
const final = await r.finish('succeeded', { rows: 1 });
const serialized = JSON.stringify(final);
expect(serialized).not.toMatch(/SELECT/i);
expect(final.compiled_query_hash).toMatch(/^[0-9a-f]{64}$/);
});
it('REFUSES to emit a record whose error field carries statement text', () => {
const record: QueryExecutionRecord = {
query_id: 'q',
execution_id: 'e',
tenant_id: 't_alpha',
principal_id: 'u',
workload: 'api',
submitted_at: new Date().toISOString(),
engine: 'clickhouse',
compiled_query_hash: 'a'.repeat(64),
datasets: [],
metrics: [],
status: 'failed',
cache_status: 'miss',
// The realistic leak: ClickHouse's error text echoes the query.
error: 'Code: 47. Unknown identifier in scope SELECT tenant_id FROM opt_analytics.sales',
};
expect(() => assertNoRawSql(record)).toThrowError(SiaxError);
});
it('hashes the same statement with different parameters differently', () => {
const a = compiledQueryHash('SELECT 1 WHERE t = {t:String}', { param_t: 't_alpha' });
const b = compiledQueryHash('SELECT 1 WHERE t = {t:String}', { param_t: 't_beta' });
expect(a).not.toBe(b);
// ...and identically regardless of key order.
expect(compiledQueryHash('q', { b: '2', a: '1' })).toBe(compiledQueryHash('q', { a: '1', b: '2' }));
});
it('refuses an illegal state transition rather than overwriting a terminal row', () => {
for (const terminal of TERMINAL_STATES) {
expect(() => assertLegalTransition(terminal, 'running')).toThrowError(SiaxError);
}
expect(() => assertLegalTransition('queued', 'succeeded')).toThrowError(SiaxError);
expect(() => assertLegalTransition('running', 'succeeded')).not.toThrow();
});
it('records a cache hit as a cache hit, not as a very fast query', async () => {
const written: QueryExecutionRecord[] = [];
const r = recorder({ write: (x) => void written.push(x) });
await r.submitted();
await r.servedFromCache(42);
const final = written.at(-1) as QueryExecutionRecord;
expect(final.cache_status).toBe('hit');
expect(final.rows_scanned).toBe(0);
expect(final.bytes_scanned).toBe(0);
});
it('requires a verified tenant and principal', () => {
for (const bad of [{ tenant_id: '' }, { principal_id: '' }]) {
expect(
() =>
new QueryExecutionRecorder(
{
tenant_id: 't_alpha',
principal_id: 'u_1',
workload: 'api',
engine: 'clickhouse',
sink: { write: () => undefined },
...bad,
},
'q',
{ hash: 'h', datasets: [], metrics: [] },
),
).toThrowError(SiaxError);
}
});
});
// ── BI-P019 (logic only; the engine claim is in the integration suite) ─────
describe('BI-P019 — cancellation reports what it VERIFIED, not what it requested', () => {
function control(script: boolean[]): EngineProcessControl & { kills: number } {
let i = 0;
return {
engine: 'test',
kills: 0,
async isRunning() {
const v = script[Math.min(i, script.length - 1)];
i += 1;
return v;
},
async requestKill() {
(this as { kills: number }).kills += 1;
return true;
},
};
}
it('reports verified_stopped only after OBSERVING the query gone', async () => {
// running, running, running, gone
const c = control([true, true, true, false]);
const out = await cancelAndVerify(c, 'q1', { timeoutMs: 5_000, pollIntervalMs: 1 });
expect(out.verified_stopped).toBe(true);
expect(out.kill_accepted).toBe(true);
expect(c.kills).toBe(1);
});
it('reports verified_stopped=false when the query outlives the deadline — it does not claim success', async () => {
const c = control([true]); // never stops
let clock = 0;
const out = await cancelAndVerify(c, 'q1', {
timeoutMs: 100,
pollIntervalMs: 10,
now: () => (clock += 30),
sleep: async () => undefined,
});
expect(out.verified_stopped).toBe(false);
expect(out.already_finished).toBe(false);
});
it('does not claim to have stopped a query that had already finished', async () => {
const c = control([false]);
const out = await cancelAndVerify(c, 'q1', { timeoutMs: 100 });
expect(out.already_finished).toBe(true);
expect(out.verified_stopped).toBe(true);
});
it('keeps polling when the engine is unreachable, then reports unverified', async () => {
let clock = 0;
const flaky: EngineProcessControl = {
engine: 'test',
isRunning: async () => {
throw new Error('engine unreachable');
},
requestKill: async () => true,
};
// isRunning throws on the FIRST call too, which cancelAndVerify treats as
// "assume it was running" — the fail-closed reading.
const out = await cancelAndVerify(flaky, 'q1', {
timeoutMs: 50,
pollIntervalMs: 10,
now: () => (clock += 20),
sleep: async () => undefined,
});
expect(out.verified_stopped).toBe(false);
});
it('aborts the client side as well as the server side', async () => {
let aborted = false;
await cancelAndVerify(control([true, false]), 'q1', {
timeoutMs: 1_000,
pollIntervalMs: 1,
abort: () => {
aborted = true;
},
});
expect(aborted).toBe(true);
});
});
// ── BI-P021 ────────────────────────────────────────────────────────────────
describe('BI-P021 — the budget says plainly what it does and does not bound', () => {
const ledger = {
entries: [] as unknown[],
record(e: unknown) {
this.entries.push(e);
},
};
it('refuses to be constructed as an in-flight enforcer without a process control', () => {
expect(
() =>
new QueryBudgetEnforcer({
budget: DEFAULT_QUERY_BUDGET,
store: new InMemoryWindowStore(),
ledger,
enforceInFlight: true,
}),
).toThrowError(SiaxError);
});
it('declares admission-only mode as advisory, in words', () => {
const e = new QueryBudgetEnforcer({
budget: DEFAULT_QUERY_BUDGET,
store: new InMemoryWindowStore(),
ledger,
enforceInFlight: false,
});
expect(e.describeEnforcement().binding).toBe(false);
expect(e.describeEnforcement().in_flight).toContain('NOT ENFORCED');
});
it('declares the per-process window as non-binding across replicas', () => {
const e = new QueryBudgetEnforcer({
budget: DEFAULT_QUERY_BUDGET,
store: new InMemoryWindowStore(),
ledger,
control: {
engine: 't',
isRunning: async () => false,
requestKill: async () => true,
progress: async () => undefined,
},
enforceInFlight: true,
storeIsShared: false,
});
expect(e.describeEnforcement().across_replicas).toContain('N times this budget');
expect(e.describeEnforcement().binding).toBe(false);
});
it('rejects an estimate that does not fit the remaining window', async () => {
const store = new InMemoryWindowStore();
const e = new QueryBudgetEnforcer({
budget: { ...DEFAULT_QUERY_BUDGET, maxRowsScanned: 1_000 },
store,
ledger,
enforceInFlight: false,
});
expect((await e.admit('t_alpha', { rows_to_read: 100, bytes_to_read: 1 })).admitted).toBe(true);
await store.add('t_alpha', { rows: 999, bytes: 0, units: 0, windowMs: DEFAULT_QUERY_BUDGET.windowMs });
const decision = await e.admit('t_alpha', { rows_to_read: 100, bytes_to_read: 1 });
expect(decision.admitted).toBe(false);
expect(decision.reason).toContain('rows');
});
it('throttles instead of rejecting when the budget says so', async () => {
const store = new InMemoryWindowStore();
const e = new QueryBudgetEnforcer({
budget: { ...DEFAULT_QUERY_BUDGET, maxRowsScanned: 10, onExhausted: 'throttle', throttleDelayMs: 250 },
store,
ledger,
enforceInFlight: false,
});
await store.add('t_alpha', { rows: 100, bytes: 0, units: 0, windowMs: DEFAULT_QUERY_BUDGET.windowMs });
const decision = await e.admit('t_alpha', { rows_to_read: 1, bytes_to_read: 1 });
expect(decision.admitted).toBe(true);
expect(decision.delay_ms).toBe(250);
});
it('charges for the work a cancelled query already did', async () => {
const store = new InMemoryWindowStore();
const entries: Array<{ terminated_by_budget: boolean; cost_units: number }> = [];
const e = new QueryBudgetEnforcer({
budget: DEFAULT_QUERY_BUDGET,
store,
ledger: { record: (x) => void entries.push(x as never) },
enforceInFlight: false,
});
const settled = await e.settle({
tenant_id: 't_alpha',
principal_id: 'u',
query_id: 'q',
cost: { rows_scanned: 1e9, bytes_scanned: 2 ** 30, execution_ms: 1000, engine: 'clickhouse' },
terminated_by_budget: true,
});
expect(settled.cost_units).toBeGreaterThan(0);
expect(entries[0].terminated_by_budget).toBe(true);
// Not charging for it is how a tenant learns the cheapest query is one
// that gets cancelled.
expect(entries[0].cost_units).toBe(settled.cost_units);
});
it('weights bytes above rows, so a wide scan is not billed like a narrow one', () => {
const narrow = costUnits({ rows_scanned: 1e9, bytes_scanned: 1e8, execution_ms: 0, engine: 'e' });
const wide = costUnits({ rows_scanned: 1e8, bytes_scanned: 1e10, execution_ms: 0, engine: 'e' });
expect(wide).toBeGreaterThan(narrow);
});
it('expires spend out of the rolling window', async () => {
let clock = 0;
const store = new InMemoryWindowStore(() => clock);
await store.add('t_alpha', { rows: 100, bytes: 0, units: 1, windowMs: 1000 });
expect((await store.get('t_alpha', 1000)).rows_scanned).toBe(100);
clock += 1500;
expect((await store.get('t_alpha', 1000)).rows_scanned).toBe(0);
});
it('in-flight enforcement returns immediately when the query is already gone', async () => {
const e = new QueryBudgetEnforcer({
budget: { ...DEFAULT_QUERY_BUDGET, maxRowsScanned: 1 },
store: new InMemoryWindowStore(),
ledger,
control: {
engine: 't',
isRunning: async () => false,
requestKill: async () => true,
progress: async () => undefined,
},
enforceInFlight: true,
});
const out = await e.enforceInFlight('t_alpha', 'q');
expect(out.terminated).toBe(false);
expect(out.polls).toBe(1);
});
});
@@ -17,13 +17,13 @@
* ran against real pyarrow + DuckDB; that transcript is committed at
* evidence/bi-p002-p012/bi-p005-arrow-roundtrip.json.
*/
import { describe, it, expect, beforeAll, afterAll } from "vitest";
import { describe, it, expect, beforeAll, afterAll } from 'vitest';
import {
arrowStreamSettings,
clickhouseArrowStreamFormat,
decodeArrowStream,
encodeArrowStream,
} from "@/src/analytics/arrow/ipc";
} from '@/src/analytics/arrow/ipc';
const CLICKHOUSE_URL = process.env.CLICKHOUSE_URL;
const CLICKHOUSE_AUTH = process.env.CLICKHOUSE_AUTH;
@@ -32,31 +32,29 @@ const run = CLICKHOUSE_URL ? describe : describe.skip;
const DB = `opt_arrow_${Date.now().toString(36)}`;
function authHeaders(): Record<string, string> {
return CLICKHOUSE_AUTH
? { Authorization: `Basic ${Buffer.from(CLICKHOUSE_AUTH).toString("base64")}` }
: {};
return CLICKHOUSE_AUTH ? { Authorization: `Basic ${Buffer.from(CLICKHOUSE_AUTH).toString('base64')}` } : {};
}
async function chBytes(sql: string, settings: Record<string, string> = {}): Promise<Uint8Array> {
const url = new URL(CLICKHOUSE_URL ?? "http://skipped.invalid");
url.searchParams.set("query", sql);
const url = new URL(CLICKHOUSE_URL ?? 'http://skipped.invalid');
url.searchParams.set('query', sql);
for (const [k, v] of Object.entries(settings)) url.searchParams.set(k, v);
const res = await fetch(url.toString(), { method: "POST", headers: authHeaders(), body: "" });
const res = await fetch(url.toString(), { method: 'POST', headers: authHeaders(), body: '' });
const buf = new Uint8Array(await res.arrayBuffer());
if (!res.ok) throw new Error(`CH ${res.status}: ${new TextDecoder().decode(buf).slice(0, 300)}`);
return buf;
}
async function chText(sql: string, body?: BodyInit): Promise<string> {
const url = new URL(CLICKHOUSE_URL ?? "http://skipped.invalid");
url.searchParams.set("query", sql);
const res = await fetch(url.toString(), { method: "POST", headers: authHeaders(), body: body ?? "" });
const url = new URL(CLICKHOUSE_URL ?? 'http://skipped.invalid');
url.searchParams.set('query', sql);
const res = await fetch(url.toString(), { method: 'POST', headers: authHeaders(), body: body ?? '' });
const text = await res.text();
if (!res.ok) throw new Error(`CH ${res.status}: ${text.slice(0, 400)}`);
return text;
}
run("BI-P005 Arrow IPC against a LIVE ClickHouse", () => {
run('BI-P005 Arrow IPC against a LIVE ClickHouse', () => {
beforeAll(async () => {
await chText(`CREATE DATABASE IF NOT EXISTS ${DB}`);
await chText(
@@ -70,7 +68,7 @@ run("BI-P005 Arrow IPC against a LIVE ClickHouse", () => {
if (CLICKHOUSE_URL) await chText(`DROP DATABASE IF EXISTS ${DB}`);
}, 60_000);
it("THEIR encoder, OUR reader: ClickHouse ArrowStream decodes with the expected schema and row count", async () => {
it('THEIR encoder, OUR reader: ClickHouse ArrowStream decodes with the expected schema and row count', async () => {
const bytes = await chBytes(
`SELECT toString(t) AS tenant_id, toInt64(n) AS n, toFloat64(v) AS v, b AS flag,
toDateTime64(ts, 3, 'UTC') AS ts
@@ -86,35 +84,38 @@ run("BI-P005 Arrow IPC against a LIVE ClickHouse", () => {
);
const info = decodeArrowStream(bytes);
expect(info.rowCount).toBe(3);
expect(info.fields.map((f) => f.name)).toEqual(["tenant_id", "n", "v", "flag", "ts"]);
expect(info.fields.map((f) => f.name)).toEqual(['tenant_id', 'n', 'v', 'flag', 'ts']);
// Utf8=5 (NOT Binary=4 — that is what output_format_arrow_string_as_string buys),
// Int=2, FloatingPoint=3, Bool=6, Timestamp=10.
expect(info.fields.map((f) => f.typeType)).toEqual([5, 2, 3, 6, 10]);
});
it("WITHOUT output_format_arrow_string_as_string, ClickHouse emits Binary — the semantic loss the setting prevents", async () => {
it('WITHOUT output_format_arrow_string_as_string, ClickHouse emits Binary — the semantic loss the setting prevents', async () => {
const bytes = await chBytes(`SELECT 'x' AS s FORMAT ${clickhouseArrowStreamFormat()}`, {
output_format_arrow_string_as_string: "0",
output_format_arrow_string_as_string: '0',
});
// Arrow Type union: Binary = 4, Utf8 = 5.
expect(decodeArrowStream(bytes).fields[0].typeType).toBe(4);
const asText = await chBytes(`SELECT 'x' AS s FORMAT ${clickhouseArrowStreamFormat()}`, arrowStreamSettings());
const asText = await chBytes(
`SELECT 'x' AS s FORMAT ${clickhouseArrowStreamFormat()}`,
arrowStreamSettings(),
);
expect(decodeArrowStream(asText).fields[0].typeType).toBe(5);
});
it("OUR encoder, THEIR reader: ClickHouse INSERTs our Arrow and returns every value unchanged", async () => {
it('OUR encoder, THEIR reader: ClickHouse INSERTs our Arrow and returns every value unchanged', async () => {
const bytes = encodeArrowStream([
{ name: "tenant_id", type: "utf8", values: ["t_alpha", "t_beta", "t_gamma"] },
{ name: "n", type: "int64", values: [1, 2, 3] },
{ name: "v", type: "float64", values: [1.5, 2.5, 3.25] },
{ name: "flag", type: "bool", values: [true, false, true] },
{ name: 'tenant_id', type: 'utf8', values: ['t_alpha', 't_beta', 't_gamma'] },
{ name: 'n', type: 'int64', values: [1, 2, 3] },
{ name: 'v', type: 'float64', values: [1.5, 2.5, 3.25] },
{ name: 'flag', type: 'bool', values: [true, false, true] },
{
name: "ts",
type: "timestamp_ms",
name: 'ts',
type: 'timestamp_ms',
values: [
new Date("2026-01-01T00:00:00.000Z"),
new Date("2026-06-15T08:09:10.123Z"),
new Date("2026-09-06T12:34:56.789Z"),
new Date('2026-01-01T00:00:00.000Z'),
new Date('2026-06-15T08:09:10.123Z'),
new Date('2026-09-06T12:34:56.789Z'),
],
},
]);
@@ -123,22 +124,30 @@ run("BI-P005 Arrow IPC against a LIVE ClickHouse", () => {
const back = await chText(
`SELECT tenant_id, n, v, flag, toString(ts) AS ts FROM ${DB}.arrow_roundtrip ORDER BY n FORMAT JSON`,
);
const rows = (JSON.parse(back) as {
data: { tenant_id: string; n: string | number; v: number | string; flag: boolean | number; ts: string }[];
}).data;
const rows = (
JSON.parse(back) as {
data: {
tenant_id: string;
n: string | number;
v: number | string;
flag: boolean | number;
ts: string;
}[];
}
).data;
expect(rows).toHaveLength(3);
expect(rows.map((r) => r.tenant_id)).toEqual(["t_alpha", "t_beta", "t_gamma"]);
expect(rows.map((r) => r.tenant_id)).toEqual(['t_alpha', 't_beta', 't_gamma']);
expect(rows.map((r) => Number(r.n))).toEqual([1, 2, 3]);
expect(rows.map((r) => Number(r.v))).toEqual([1.5, 2.5, 3.25]);
expect(rows.map((r) => Boolean(Number(r.flag)))).toEqual([true, false, true]);
expect(rows.map((r) => r.ts)).toEqual([
"2026-01-01 00:00:00.000",
"2026-06-15 08:09:10.123",
"2026-09-06 12:34:56.789",
'2026-01-01 00:00:00.000',
'2026-06-15 08:09:10.123',
'2026-09-06 12:34:56.789',
]);
});
it("full circle: ClickHouse -> Arrow -> our decoder -> ClickHouse yields identical rows", async () => {
it('full circle: ClickHouse -> Arrow -> our decoder -> ClickHouse yields identical rows', async () => {
const exported = await chBytes(
`SELECT tenant_id, n, v, flag, ts FROM ${DB}.arrow_roundtrip ORDER BY n FORMAT ArrowStream`,
arrowStreamSettings(),
@@ -158,14 +167,18 @@ run("BI-P005 Arrow IPC against a LIVE ClickHouse", () => {
expect(Number((JSON.parse(diff) as { data: { c: number | string }[] }).data[0].c)).toBe(0);
});
it("a large result survives the round trip (batching/offsets are not only correct for three rows)", async () => {
it('a large result survives the round trip (batching/offsets are not only correct for three rows)', async () => {
const n = 20_000;
const bytes = encodeArrowStream([
{ name: "tenant_id", type: "utf8", values: Array.from({ length: n }, (_, i) => `t_${i % 7}`) },
{ name: "n", type: "int64", values: Array.from({ length: n }, (_, i) => i) },
{ name: "v", type: "float64", values: Array.from({ length: n }, (_, i) => i / 3) },
{ name: "flag", type: "bool", values: Array.from({ length: n }, (_, i) => i % 2 === 0) },
{ name: "ts", type: "timestamp_ms", values: Array.from({ length: n }, () => new Date("2026-09-06T00:00:00Z")) },
{ name: 'tenant_id', type: 'utf8', values: Array.from({ length: n }, (_, i) => `t_${i % 7}`) },
{ name: 'n', type: 'int64', values: Array.from({ length: n }, (_, i) => i) },
{ name: 'v', type: 'float64', values: Array.from({ length: n }, (_, i) => i / 3) },
{ name: 'flag', type: 'bool', values: Array.from({ length: n }, (_, i) => i % 2 === 0) },
{
name: 'ts',
type: 'timestamp_ms',
values: Array.from({ length: n }, () => new Date('2026-09-06T00:00:00Z')),
},
]);
expect(decodeArrowStream(bytes).rowCount).toBe(n);
@@ -174,7 +187,11 @@ run("BI-P005 Arrow IPC against a LIVE ClickHouse", () => {
const agg = await chText(
`SELECT count() AS c, sum(n) AS s, uniqExact(tenant_id) AS t, countIf(flag) AS f FROM ${DB}.arrow_big FORMAT JSON`,
);
const d = (JSON.parse(agg) as { data: { c: string | number; s: string | number; t: string | number; f: string | number }[] }).data[0];
const d = (
JSON.parse(agg) as {
data: { c: string | number; s: string | number; t: string | number; f: string | number }[];
}
).data[0];
expect(Number(d.c)).toBe(n);
expect(Number(d.s)).toBe((n * (n - 1)) / 2);
expect(Number(d.t)).toBe(7);
@@ -7,19 +7,19 @@
* JSONEachRow ingest, the closed metric-query DSL, and — the §69-relevant
* part — that the FORCED tenant predicate isolates tenants on real data.
*/
import { describe, it, expect, beforeAll } from "vitest";
import { ClickHouseAnalytics } from "@/src/analytics/clickhouse-analytics";
import { describe, it, expect, beforeAll } from 'vitest';
import { ClickHouseAnalytics } from '@/src/analytics/clickhouse-analytics';
const CLICKHOUSE_URL = process.env.CLICKHOUSE_URL;
const run = CLICKHOUSE_URL ? describe : describe.skip;
const DB = `opt_itest_${Date.now().toString(36)}`;
run("Analytics capability against a LIVE ClickHouse", () => {
run('Analytics capability against a LIVE ClickHouse', () => {
// The skip-variant still evaluates this callback body, so the constructor
// must tolerate the env-less case (the tests themselves never run then).
const analytics = new ClickHouseAnalytics({
baseUrl: CLICKHOUSE_URL ?? "http://skipped.invalid",
baseUrl: CLICKHOUSE_URL ?? 'http://skipped.invalid',
basicAuth: process.env.CLICKHOUSE_AUTH,
database: DB,
});
@@ -31,66 +31,90 @@ run("Analytics capability against a LIVE ClickHouse", () => {
await analytics.ensureSchema(30);
}, 60_000);
it("ingests events and answers aggregate metric queries", async () => {
it('ingests events and answers aggregate metric queries', async () => {
const res = await analytics.ingest([
{ tenant_id: "t_alpha", dataset_id: "ds_companies", metric: "rows_ingested", value: 100 },
{ tenant_id: "t_alpha", dataset_id: "ds_companies", metric: "rows_ingested", value: 50 },
{ tenant_id: "t_alpha", dataset_id: "ds_vehicles", metric: "rows_ingested", value: 7 },
{ tenant_id: "t_beta", dataset_id: "ds_companies", metric: "rows_ingested", value: 999 },
{ tenant_id: 't_alpha', dataset_id: 'ds_companies', metric: 'rows_ingested', value: 100 },
{ tenant_id: 't_alpha', dataset_id: 'ds_companies', metric: 'rows_ingested', value: 50 },
{ tenant_id: 't_alpha', dataset_id: 'ds_vehicles', metric: 'rows_ingested', value: 7 },
{ tenant_id: 't_beta', dataset_id: 'ds_companies', metric: 'rows_ingested', value: 999 },
]);
expect(res.ingested).toBe(4);
const total = await analytics.queryMetric({
tenant_id: "t_alpha",
metric: "rows_ingested",
aggregate: "sum",
tenant_id: 't_alpha',
metric: 'rows_ingested',
aggregate: 'sum',
});
expect(total).toEqual([{ value: 157 }]);
const perDataset = await analytics.queryMetric({
tenant_id: "t_alpha",
dataset_id: "ds_companies",
metric: "rows_ingested",
aggregate: "sum",
tenant_id: 't_alpha',
dataset_id: 'ds_companies',
metric: 'rows_ingested',
aggregate: 'sum',
});
expect(perDataset).toEqual([{ value: 150 }]);
}, 30_000);
it("REAL tenant isolation: t_beta's numbers never leak into t_alpha aggregates", async () => {
const alpha = await analytics.queryMetric({ tenant_id: "t_alpha", metric: "rows_ingested", aggregate: "max" });
const alpha = await analytics.queryMetric({
tenant_id: 't_alpha',
metric: 'rows_ingested',
aggregate: 'max',
});
expect(alpha[0].value).toBe(100); // NOT 999
const beta = await analytics.queryMetric({ tenant_id: "t_beta", metric: "rows_ingested", aggregate: "sum" });
const beta = await analytics.queryMetric({
tenant_id: 't_beta',
metric: 'rows_ingested',
aggregate: 'sum',
});
expect(beta[0].value).toBe(999);
const ghost = await analytics.queryMetric({ tenant_id: "t_gamma", metric: "rows_ingested", aggregate: "count" });
const ghost = await analytics.queryMetric({
tenant_id: 't_gamma',
metric: 'rows_ingested',
aggregate: 'count',
});
expect(ghost[0].value).toBe(0);
});
it("group-by dimension queries work through the closed DSL", async () => {
it('group-by dimension queries work through the closed DSL', async () => {
await analytics.ingest([
{ tenant_id: "t_alpha", dataset_id: "ds_x", metric: "dq_score", value: 90, dimensions: { severity: "warning" } },
{ tenant_id: "t_alpha", dataset_id: "ds_x", metric: "dq_score", value: 40, dimensions: { severity: "critical" } },
{
tenant_id: 't_alpha',
dataset_id: 'ds_x',
metric: 'dq_score',
value: 90,
dimensions: { severity: 'warning' },
},
{
tenant_id: 't_alpha',
dataset_id: 'ds_x',
metric: 'dq_score',
value: 40,
dimensions: { severity: 'critical' },
},
]);
const rows = await analytics.queryMetric({
tenant_id: "t_alpha",
metric: "dq_score",
aggregate: "avg",
group_by: "severity",
tenant_id: 't_alpha',
metric: 'dq_score',
aggregate: 'avg',
group_by: 'severity',
});
expect(rows).toHaveLength(2);
expect(rows.find((r) => r.group === "warning")?.value).toBe(90);
expect(rows.find((r) => r.group === 'warning')?.value).toBe(90);
});
it("rejects tenant-less queries and invalid group_by identifiers (injection surface closed)", async () => {
it('rejects tenant-less queries and invalid group_by identifiers (injection surface closed)', async () => {
await expect(
analytics.queryMetric({ tenant_id: " ", metric: "x", aggregate: "sum" })
).rejects.toMatchObject({ code: "INVALID_REQUEST" });
analytics.queryMetric({ tenant_id: ' ', metric: 'x', aggregate: 'sum' }),
).rejects.toMatchObject({ code: 'INVALID_REQUEST' });
await expect(
analytics.queryMetric({
tenant_id: "t_alpha",
metric: "x",
aggregate: "sum",
tenant_id: 't_alpha',
metric: 'x',
aggregate: 'sum',
group_by: "sev'; DROP TABLE opt_metric_events; --",
})
).rejects.toMatchObject({ code: "INVALID_REQUEST" });
}),
).rejects.toMatchObject({ code: 'INVALID_REQUEST' });
});
});
@@ -12,38 +12,38 @@
* Gated on CLICKHOUSE_URL; skips honestly when unset. CI's `integration` job
* supplies it and hard-fails on a skipped suite.
*/
import { describe, it, expect, beforeAll, afterAll } from "vitest";
import { ClickHouseAnalyticsEngine } from "@/src/analytics/engine/clickhouse-engine";
import { ANALYTICS_MATERIALIZED_VIEWS, ANALYTICS_TABLES } from "@/src/analytics/engine/schema";
import type { AnalyticsIdentity } from "@/src/analytics/engine/siax-analytics-engine";
import { describe, it, expect, beforeAll, afterAll } from 'vitest';
import { ClickHouseAnalyticsEngine } from '@/src/analytics/engine/clickhouse-engine';
import { ANALYTICS_MATERIALIZED_VIEWS, ANALYTICS_TABLES } from '@/src/analytics/engine/schema';
import type { AnalyticsIdentity } from '@/src/analytics/engine/siax-analytics-engine';
const CLICKHOUSE_URL = process.env.CLICKHOUSE_URL;
const CLICKHOUSE_AUTH = process.env.CLICKHOUSE_AUTH;
const run = CLICKHOUSE_URL ? describe : describe.skip;
const DB = `opt_engine_${Date.now().toString(36)}`;
const ID: AnalyticsIdentity = { tenant_id: "t_alpha", project_id: "p_main", principal_id: "usr_alpha" };
const ID: AnalyticsIdentity = { tenant_id: 't_alpha', project_id: 'p_main', principal_id: 'usr_alpha' };
let engine: ClickHouseAnalyticsEngine;
async function raw(sql: string, body?: string): Promise<string> {
const url = new URL(CLICKHOUSE_URL ?? "http://skipped.invalid");
url.searchParams.set("query", sql);
const url = new URL(CLICKHOUSE_URL ?? 'http://skipped.invalid');
url.searchParams.set('query', sql);
const headers: Record<string, string> = {};
if (CLICKHOUSE_AUTH) headers["Authorization"] = `Basic ${Buffer.from(CLICKHOUSE_AUTH).toString("base64")}`;
const res = await fetch(url.toString(), { method: "POST", headers, body: body ?? "" });
if (CLICKHOUSE_AUTH) headers['Authorization'] = `Basic ${Buffer.from(CLICKHOUSE_AUTH).toString('base64')}`;
const res = await fetch(url.toString(), { method: 'POST', headers, body: body ?? '' });
const text = await res.text();
if (!res.ok) throw new Error(`CH ${res.status}: ${text.slice(0, 300)}`);
return text;
}
run("BI-P009/BI-P010 ClickHouse provider against a LIVE server", () => {
run('BI-P009/BI-P010 ClickHouse provider against a LIVE server', () => {
beforeAll(async () => {
engine = new ClickHouseAnalyticsEngine({
baseUrl: CLICKHOUSE_URL!,
basicAuth: CLICKHOUSE_AUTH,
database: DB,
dimensionCatalogs: { t_alpha: { severity: "severity", make: "vehicle_make" } },
dimensionCatalogs: { t_alpha: { severity: 'severity', make: 'vehicle_make' } },
});
await engine.ensureSchema();
await engine.ensureSchema(); // idempotent, as the DDL claims
@@ -58,67 +58,67 @@ run("BI-P009/BI-P010 ClickHouse provider against a LIVE server", () => {
it("health() reports the live server's version", async () => {
const h = await engine.health();
expect(h.healthy).toBe(true);
expect(h.provider).toBe("clickhouse");
expect(h.provider).toBe('clickhouse');
expect(h.version).toMatch(/^\d+\.\d+/);
});
// ── BI-P010: the schema is what it claims to be ─────────────────────────
it("BI-P010: four purpose-built tables exist, NOT one giant table", async () => {
it('BI-P010: four purpose-built tables exist, NOT one giant table', async () => {
const listed = await raw(
`SELECT name FROM system.tables WHERE database = '${DB}' AND engine != 'MaterializedView' ORDER BY name FORMAT TSV`,
);
const names = listed.trim().split("\n").filter(Boolean);
const names = listed.trim().split('\n').filter(Boolean);
// The three this module owns, PLUS the preserved opt_metric_events.
for (const t of ANALYTICS_TABLES) expect(names).toContain(t);
expect(names).toContain("opt_metric_events");
expect(names).toContain('opt_metric_events');
expect(names.length).toBeGreaterThanOrEqual(4);
});
it("BI-P010: the tables use different engines and different sort keys — the split is real", async () => {
it('BI-P010: the tables use different engines and different sort keys — the split is real', async () => {
const meta = await raw(
`SELECT name, engine, sorting_key FROM system.tables WHERE database = '${DB}' ORDER BY name FORMAT JSON`,
);
const rows = (JSON.parse(meta) as { data: { name: string; engine: string; sorting_key: string }[] }).data;
const by = Object.fromEntries(rows.map((r) => [r.name, r]));
expect(by["opt_analytics_event"].engine).toBe("MergeTree");
expect(by["opt_metric_daily"].engine).toBe("AggregatingMergeTree");
expect(by["opt_dataset_projection"].engine).toBe("ReplacingMergeTree");
expect(by['opt_analytics_event'].engine).toBe('MergeTree');
expect(by['opt_metric_daily'].engine).toBe('AggregatingMergeTree');
expect(by['opt_dataset_projection'].engine).toBe('ReplacingMergeTree');
// Different ORDER BY keys — the reason one table cannot serve both jobs.
expect(by["opt_analytics_event"].sorting_key).not.toBe(by["opt_dataset_projection"].sorting_key);
expect(by['opt_analytics_event'].sorting_key).not.toBe(by['opt_dataset_projection'].sorting_key);
// Every one still leads with tenant_id.
for (const t of ANALYTICS_TABLES) expect(by[t].sorting_key.startsWith("tenant_id")).toBe(true);
for (const t of ANALYTICS_TABLES) expect(by[t].sorting_key.startsWith('tenant_id')).toBe(true);
});
it("BI-P010: TTL and LowCardinality are actually applied, not just written in a comment", async () => {
it('BI-P010: TTL and LowCardinality are actually applied, not just written in a comment', async () => {
const ddl = await raw(`SHOW CREATE TABLE ${DB}.opt_analytics_event FORMAT TSVRaw`);
expect(ddl).toContain("TTL");
expect(ddl).toContain("LowCardinality(String)");
expect(ddl).toContain("PARTITION BY toYYYYMM(event_time)");
expect(ddl).toContain('TTL');
expect(ddl).toContain('LowCardinality(String)');
expect(ddl).toContain('PARTITION BY toYYYYMM(event_time)');
const rollup = await raw(`SHOW CREATE TABLE ${DB}.opt_metric_daily FORMAT TSVRaw`);
expect(rollup).toContain("AggregateFunction(sum, Float64)");
expect(rollup).toContain('AggregateFunction(sum, Float64)');
});
it("BI-P010: the materialized view actually populates the rollup", async () => {
it('BI-P010: the materialized view actually populates the rollup', async () => {
const views = await raw(
`SELECT name FROM system.tables WHERE database='${DB}' AND engine='MaterializedView' FORMAT TSV`,
);
for (const v of ANALYTICS_MATERIALIZED_VIEWS) expect(views).toContain(v);
await engine.ingestProjection(ID, { dataset_id: "warmup", rows: [{ entity_key: "w1" }] });
await engine.ingestProjection(ID, { dataset_id: 'warmup', rows: [{ entity_key: 'w1' }] });
const rows = Array.from({ length: 20 }, (_, i) =>
JSON.stringify({
tenant_id: "t_alpha",
project_id: "p_main",
dataset_id: "ds_mv",
metric: "mv_metric",
tenant_id: 't_alpha',
project_id: 'p_main',
dataset_id: 'ds_mv',
metric: 'mv_metric',
event_id: `mv${i}`,
event_time: "2026-09-03 12:00:00.000",
event_time: '2026-09-03 12:00:00.000',
value: 2,
dim_names: [],
dim_values: [],
}),
).join("\n");
).join('\n');
await raw(`INSERT INTO ${DB}.opt_analytics_event FORMAT JSONEachRow`, rows);
const rolled = await raw(
@@ -132,72 +132,75 @@ run("BI-P009/BI-P010 ClickHouse provider against a LIVE server", () => {
// ── capability 2: executeMetric (already covered by the isolation suite) ─
it("executeMetric returns grouped rows keyed by the CALLER-FACING dimension name", async () => {
const rows = ["warning", "critical", "critical"].map((sev, i) =>
it('executeMetric returns grouped rows keyed by the CALLER-FACING dimension name', async () => {
const rows = ['warning', 'critical', 'critical'].map((sev, i) =>
JSON.stringify({
tenant_id: "t_alpha",
project_id: "p_main",
dataset_id: "ds_g",
metric: "grouped",
tenant_id: 't_alpha',
project_id: 'p_main',
dataset_id: 'ds_g',
metric: 'grouped',
event_id: `g${i}`,
event_time: "2026-09-04 09:00:00.000",
event_time: '2026-09-04 09:00:00.000',
value: 5,
dim_names: ["severity"],
dim_names: ['severity'],
dim_values: [sev],
}),
);
await raw(`INSERT INTO ${DB}.opt_analytics_event FORMAT JSONEachRow`, rows.join("\n"));
await raw(`INSERT INTO ${DB}.opt_analytics_event FORMAT JSONEachRow`, rows.join('\n'));
const result = await engine.executeMetric(ID, {
metric: "grouped",
aggregate: "sum",
group_by: ["severity"],
metric: 'grouped',
aggregate: 'sum',
group_by: ['severity'],
});
const critical = result.rows.find((r) => r.groups?.severity === "critical");
const critical = result.rows.find((r) => r.groups?.severity === 'critical');
expect(critical?.value).toBe(10);
expect(result.stats.result_rows).toBe(2);
expect(result.query_id.startsWith("opt_")).toBe(true);
expect(result.query_id.startsWith('opt_')).toBe(true);
});
// ── capability 4: estimateQuery ─────────────────────────────────────────
it("estimateQuery gets a real EXPLAIN ESTIMATE from the server and compares it to the budget", async () => {
const est = await engine.estimateQuery(ID, { metric: "grouped", aggregate: "sum" });
it('estimateQuery gets a real EXPLAIN ESTIMATE from the server and compares it to the budget', async () => {
const est = await engine.estimateQuery(ID, { metric: 'grouped', aggregate: 'sum' });
expect(est.estimated_rows_to_read).toBeGreaterThan(0);
expect(est.within_budget).toBe(true);
expect(est.budget.maxRowsToRead).toBeGreaterThan(0);
});
it("estimateQuery REFUSES a query whose estimate exceeds a tightened quota", async () => {
it('estimateQuery REFUSES a query whose estimate exceeds a tightened quota', async () => {
const tiny: AnalyticsIdentity = { ...ID, quota: { maxRowsToRead: 1 } };
const est = await engine.estimateQuery(tiny, { metric: "grouped", aggregate: "sum" });
const est = await engine.estimateQuery(tiny, { metric: 'grouped', aggregate: 'sum' });
expect(est.within_budget).toBe(false);
expect(est.reason).toMatch(/over budget/);
});
// ── capability 6: explain ───────────────────────────────────────────────
it("explain() proves the forced predicate reaches the PRIMARY KEY, pruning parts", async () => {
const ex = await engine.explain(ID, { metric: "grouped", aggregate: "sum", group_by: ["severity"] });
expect(ex.forced_tenant_predicate).toBe("tenant_id = {tenant:String}");
expect(ex.table).toBe("opt_analytics_event");
it('explain() proves the forced predicate reaches the PRIMARY KEY, pruning parts', async () => {
const ex = await engine.explain(ID, { metric: 'grouped', aggregate: 'sum', group_by: ['severity'] });
expect(ex.forced_tenant_predicate).toBe('tenant_id = {tenant:String}');
expect(ex.table).toBe('opt_analytics_event');
// This is the payoff of BI-P010's tenant-first ORDER BY: the server
// reports tenant_id as a PrimaryKey key with a condition on it, i.e. the
// isolation predicate is an index range scan, not a post-read filter.
expect(ex.plan).toContain("ReadFromMergeTree");
expect(ex.plan).toContain("PrimaryKey");
expect(ex.plan).toContain('ReadFromMergeTree');
expect(ex.plan).toContain('PrimaryKey');
expect(ex.plan).toMatch(/Keys:[\s\S]*tenant_id/);
expect(ex.plan).toMatch(/tenant_id in \['t_alpha', 't_alpha'\]/);
// The plan describes only the CALLER's own tenant; no other tenant id
// may appear in a plan handed back to them.
expect(ex.plan).not.toContain("t_beta");
expect(ex.plan).not.toContain('t_beta');
});
// ── capability 5: cancelQuery ───────────────────────────────────────────
it("cancelQuery issues a real KILL QUERY the server accepts", async () => {
it('cancelQuery issues a real KILL QUERY the server accepts', async () => {
// Nothing is running under this id, so the honest outcome is "the KILL
// statement was valid and matched nothing" -> true, with no exception.
const ok = await engine.cancelQuery(ID, `${(await engine.executeMetric(ID, { metric: "grouped", aggregate: "count" })).query_id}`);
const ok = await engine.cancelQuery(
ID,
`${(await engine.executeMetric(ID, { metric: 'grouped', aggregate: 'count' })).query_id}`,
);
expect(ok).toBe(true);
});
@@ -205,10 +208,10 @@ run("BI-P009/BI-P010 ClickHouse provider against a LIVE server", () => {
it("ingestProjection writes the SERVING tier with the identity's tenant, ignoring any row-level claim", async () => {
const res = await engine.ingestProjection(ID, {
dataset_id: "ds_vehicles",
dataset_id: 'ds_vehicles',
rows: [
{ entity_key: "VIN001", attributes: { make: "volvo" }, measures: { price: 189000 } },
{ entity_key: "VIN002", attributes: { make: "saab" }, measures: { price: 42000 } },
{ entity_key: 'VIN001', attributes: { make: 'volvo' }, measures: { price: 189000 } },
{ entity_key: 'VIN002', attributes: { make: 'saab' }, measures: { price: 42000 } },
],
});
expect(res.ingested).toBe(2);
@@ -219,18 +222,18 @@ run("BI-P009/BI-P010 ClickHouse provider against a LIVE server", () => {
expect(Number((JSON.parse(stored) as { data: { t: number | string }[] }).data[0].t)).toBe(1);
});
it("refreshProjection makes a re-ingest idempotent (ReplacingMergeTree + OPTIMIZE FINAL)", async () => {
it('refreshProjection makes a re-ingest idempotent (ReplacingMergeTree + OPTIMIZE FINAL)', async () => {
// Re-ingest the SAME entity keys at a higher version.
await engine.ingestProjection(ID, {
dataset_id: "ds_vehicles",
dataset_id: 'ds_vehicles',
rows: [
{ entity_key: "VIN001", version: 9_999_999_999_999, measures: { price: 175000 } },
{ entity_key: "VIN002", version: 9_999_999_999_999, measures: { price: 41000 } },
{ entity_key: 'VIN001', version: 9_999_999_999_999, measures: { price: 175000 } },
{ entity_key: 'VIN002', version: 9_999_999_999_999, measures: { price: 41000 } },
],
});
const refreshed = await engine.refreshProjection(ID, "ds_vehicles");
const refreshed = await engine.refreshProjection(ID, 'ds_vehicles');
expect(refreshed.row_count).toBe(2); // 2 entities, not 4 rows
expect(refreshed.dataset_id).toBe("ds_vehicles");
expect(refreshed.dataset_id).toBe('ds_vehicles');
const winner = await raw(
`SELECT measures['price'] AS p FROM ${DB}.opt_dataset_projection FINAL ` +
@@ -241,12 +244,12 @@ run("BI-P009/BI-P010 ClickHouse provider against a LIVE server", () => {
// ── the PRESERVED path still works through the engine ───────────────────
it("the PRESERVED ClickHouseAnalytics path still ingests and answers through the engine", async () => {
it('the PRESERVED ClickHouseAnalytics path still ingests and answers through the engine', async () => {
await engine.legacyAnalytics.ingest([
{ tenant_id: "t_alpha", dataset_id: "ds_legacy", metric: "legacy_metric", value: 7 },
{ tenant_id: "t_beta", dataset_id: "ds_legacy", metric: "legacy_metric", value: 700 },
{ tenant_id: 't_alpha', dataset_id: 'ds_legacy', metric: 'legacy_metric', value: 7 },
{ tenant_id: 't_beta', dataset_id: 'ds_legacy', metric: 'legacy_metric', value: 700 },
]);
const res = await engine.executeLegacyMetric(ID, { metric: "legacy_metric", aggregate: "sum" });
const res = await engine.executeLegacyMetric(ID, { metric: 'legacy_metric', aggregate: 'sum' });
expect(res.rows[0].value).toBe(7); // NOT 707
});
});
@@ -14,10 +14,10 @@
* timed WHILE the bomb is in flight and compared against a declared SLO,
* not merely observed to "still work".
*/
import { describe, it, expect, beforeAll, afterAll } from "vitest";
import { ClickHouseAnalyticsEngine } from "@/src/analytics/engine/clickhouse-engine";
import { DEFAULT_BUDGETS } from "@/src/analytics/engine/resource-governance";
import type { AnalyticsIdentity } from "@/src/analytics/engine/siax-analytics-engine";
import { describe, it, expect, beforeAll, afterAll } from 'vitest';
import { ClickHouseAnalyticsEngine } from '@/src/analytics/engine/clickhouse-engine';
import { DEFAULT_BUDGETS } from '@/src/analytics/engine/resource-governance';
import type { AnalyticsIdentity } from '@/src/analytics/engine/siax-analytics-engine';
const CLICKHOUSE_URL = process.env.CLICKHOUSE_URL;
const CLICKHOUSE_AUTH = process.env.CLICKHOUSE_AUTH;
@@ -37,7 +37,7 @@ const DB = `opt_gov_${Date.now().toString(36)}`;
*/
const UNRELATED_TENANT_SLO_MS = 5000;
const VICTIM: AnalyticsIdentity = { tenant_id: "t_victim", project_id: "p_main", principal_id: "usr_v" };
const VICTIM: AnalyticsIdentity = { tenant_id: 't_victim', project_id: 'p_main', principal_id: 'usr_v' };
let engine: ClickHouseAnalyticsEngine;
@@ -47,40 +47,40 @@ async function chRun(
settings: Record<string, string> = {},
body?: string,
): Promise<{ ok: boolean; code?: string; text: string; ms: number }> {
const url = new URL(CLICKHOUSE_URL ?? "http://skipped.invalid");
url.searchParams.set("query", sql);
const url = new URL(CLICKHOUSE_URL ?? 'http://skipped.invalid');
url.searchParams.set('query', sql);
for (const [k, v] of Object.entries(settings)) url.searchParams.set(k, v);
const headers: Record<string, string> = {};
if (CLICKHOUSE_AUTH) headers["Authorization"] = `Basic ${Buffer.from(CLICKHOUSE_AUTH).toString("base64")}`;
if (CLICKHOUSE_AUTH) headers['Authorization'] = `Basic ${Buffer.from(CLICKHOUSE_AUTH).toString('base64')}`;
const t0 = Date.now();
const res = await fetch(url.toString(), { method: "POST", headers, body: body ?? "" });
const res = await fetch(url.toString(), { method: 'POST', headers, body: body ?? '' });
const text = await res.text();
return { ok: res.ok, code: /Code: (\d+)/.exec(text)?.[1], text, ms: Date.now() - t0 };
}
run("BI-P012 resource governance against a LIVE ClickHouse", () => {
run('BI-P012 resource governance against a LIVE ClickHouse', () => {
beforeAll(async () => {
engine = new ClickHouseAnalyticsEngine({
baseUrl: CLICKHOUSE_URL!,
basicAuth: CLICKHOUSE_AUTH,
database: DB,
dimensionCatalogs: { t_victim: { severity: "severity" } },
dimensionCatalogs: { t_victim: { severity: 'severity' } },
});
await engine.ensureSchema();
// A small, real dataset for the unrelated tenant.
const rows = Array.from({ length: 500 }, (_, i) =>
JSON.stringify({
tenant_id: "t_victim",
project_id: "p_main",
dataset_id: "ds_v",
metric: "victim_metric",
tenant_id: 't_victim',
project_id: 'p_main',
dataset_id: 'ds_v',
metric: 'victim_metric',
event_id: `v${i}`,
event_time: "2026-09-05 08:00:00.000",
event_time: '2026-09-05 08:00:00.000',
value: 1,
dim_names: ["severity"],
dim_values: ["warning"],
dim_names: ['severity'],
dim_values: ['warning'],
}),
).join("\n");
).join('\n');
const ins = await chRun(`INSERT INTO ${DB}.opt_analytics_event FORMAT JSONEachRow`, {}, rows);
expect(ins.ok, ins.text.slice(0, 200)).toBe(true);
}, 120_000);
@@ -91,51 +91,51 @@ run("BI-P012 resource governance against a LIVE ClickHouse", () => {
// ── each named limit, fired individually ────────────────────────────────
it("max_rows_to_read terminates a read bomb (CH code 158 TOO_MANY_ROWS)", async () => {
const r = await chRun("SELECT count() FROM numbers(1000000000) FORMAT JSON", {
readonly: "1",
max_rows_to_read: "1000",
read_overflow_mode: "throw",
it('max_rows_to_read terminates a read bomb (CH code 158 TOO_MANY_ROWS)', async () => {
const r = await chRun('SELECT count() FROM numbers(1000000000) FORMAT JSON', {
readonly: '1',
max_rows_to_read: '1000',
read_overflow_mode: 'throw',
});
expect(r.ok).toBe(false);
expect(r.code).toBe("158");
expect(r.code).toBe('158');
});
it("max_bytes_to_read terminates a scan bomb (CH code 307 TOO_MANY_BYTES)", async () => {
const r = await chRun("SELECT sum(number) FROM numbers(1000000000) FORMAT JSON", {
readonly: "1",
max_bytes_to_read: "1024",
read_overflow_mode: "throw",
it('max_bytes_to_read terminates a scan bomb (CH code 307 TOO_MANY_BYTES)', async () => {
const r = await chRun('SELECT sum(number) FROM numbers(1000000000) FORMAT JSON', {
readonly: '1',
max_bytes_to_read: '1024',
read_overflow_mode: 'throw',
});
expect(r.ok).toBe(false);
expect(r.code).toBe("307");
expect(r.code).toBe('307');
});
it("max_execution_time terminates a slow bomb (CH code 159 TIMEOUT_EXCEEDED)", async () => {
const r = await chRun("SELECT count() FROM numbers(50000000000) FORMAT JSON", {
readonly: "1",
max_execution_time: "1",
timeout_overflow_mode: "throw",
it('max_execution_time terminates a slow bomb (CH code 159 TIMEOUT_EXCEEDED)', async () => {
const r = await chRun('SELECT count() FROM numbers(50000000000) FORMAT JSON', {
readonly: '1',
max_execution_time: '1',
timeout_overflow_mode: 'throw',
});
expect(r.ok).toBe(false);
expect(r.code).toBe("159");
expect(r.code).toBe('159');
expect(r.ms).toBeLessThan(20_000);
});
it("max_result_rows terminates a result bomb and does NOT return a partial result", async () => {
const r = await chRun("SELECT number FROM numbers(100000) FORMAT JSON", {
readonly: "1",
max_result_rows: "10",
result_overflow_mode: "throw",
it('max_result_rows terminates a result bomb and does NOT return a partial result', async () => {
const r = await chRun('SELECT number FROM numbers(100000) FORMAT JSON', {
readonly: '1',
max_result_rows: '10',
result_overflow_mode: 'throw',
});
expect(r.ok).toBe(false);
expect(r.code).toBe("396"); // TOO_MANY_ROWS_OR_BYTES
expect(r.code).toBe('396'); // TOO_MANY_ROWS_OR_BYTES
// The dangerous alternative: result_overflow_mode=break returns 10 rows
// with HTTP 200 and no indication of truncation.
const broken = await chRun("SELECT number FROM numbers(100000) FORMAT JSON", {
readonly: "1",
max_result_rows: "10",
result_overflow_mode: "break",
const broken = await chRun('SELECT number FROM numbers(100000) FORMAT JSON', {
readonly: '1',
max_result_rows: '10',
result_overflow_mode: 'break',
});
// MEASURED on ClickHouse 24.8: break mode answers HTTP 200 with 65409 of
// 100000 rows (it stops at a block boundary, not at the stated 10) and
@@ -147,37 +147,38 @@ run("BI-P012 resource governance against a LIVE ClickHouse", () => {
expect(brokenRows).toBeLessThan(100_000);
});
it("max_memory_usage terminates a memory bomb (CH code 241 MEMORY_LIMIT_EXCEEDED)", async () => {
const r = await chRun(
"SELECT uniqExact(toString(number)) FROM numbers(50000000) FORMAT JSON",
{ readonly: "1", max_memory_usage: "1000000", max_execution_time: "30" },
);
it('max_memory_usage terminates a memory bomb (CH code 241 MEMORY_LIMIT_EXCEEDED)', async () => {
const r = await chRun('SELECT uniqExact(toString(number)) FROM numbers(50000000) FORMAT JSON', {
readonly: '1',
max_memory_usage: '1000000',
max_execution_time: '30',
});
expect(r.ok).toBe(false);
expect(r.code).toBe("241");
expect(r.code).toBe('241');
});
it("readonly=1 means the query text cannot re-raise its own limits", async () => {
it('readonly=1 means the query text cannot re-raise its own limits', async () => {
const r = await chRun(
"SELECT count() FROM numbers(1000000000) SETTINGS max_rows_to_read = 10000000000 FORMAT JSON",
{ readonly: "1", max_rows_to_read: "1000", read_overflow_mode: "throw" },
'SELECT count() FROM numbers(1000000000) SETTINGS max_rows_to_read = 10000000000 FORMAT JSON',
{ readonly: '1', max_rows_to_read: '1000', read_overflow_mode: 'throw' },
);
expect(r.ok).toBe(false);
// 164 = READONLY (changing settings is forbidden); 158 = the limit still
// applied. Either proves the caller did not win — but NOT a 200.
expect(["164", "158"]).toContain(r.code);
expect(['164', '158']).toContain(r.code);
});
it("readonly=1 blocks writes through the query surface", async () => {
const r = await chRun(`DROP TABLE ${DB}.opt_analytics_event FORMAT JSON`, { readonly: "1" });
it('readonly=1 blocks writes through the query surface', async () => {
const r = await chRun(`DROP TABLE ${DB}.opt_analytics_event FORMAT JSON`, { readonly: '1' });
expect(r.ok).toBe(false);
});
// ── the half that is usually skipped ────────────────────────────────────
it("an unrelated tenant stays within its declared SLO WHILE a query bomb is in flight", async () => {
it('an unrelated tenant stays within its declared SLO WHILE a query bomb is in flight', async () => {
// Baseline first, so "fast enough" is measured against this server, not
// against a number someone hoped for.
const baseline = await engine.executeMetric(VICTIM, { metric: "victim_metric", aggregate: "count" });
const baseline = await engine.executeMetric(VICTIM, { metric: 'victim_metric', aggregate: 'count' });
expect(baseline.rows[0].value).toBe(500);
const baselineMs = baseline.stats.duration_ms;
@@ -186,23 +187,23 @@ run("BI-P012 resource governance against a LIVE ClickHouse", () => {
// numbers(20e9) in ~5s because it can skip the values entirely. Hashing
// every row cannot be short-circuited, so this one really does burn CPU
// until the limit kills it.
const bomb = chRun("SELECT sum(cityHash64(number)) FROM numbers(20000000000) FORMAT JSON", {
readonly: "1",
max_execution_time: "8",
timeout_overflow_mode: "throw",
const bomb = chRun('SELECT sum(cityHash64(number)) FROM numbers(20000000000) FORMAT JSON', {
readonly: '1',
max_execution_time: '8',
timeout_overflow_mode: 'throw',
});
// While it burns, run the unrelated tenant's interactive query repeatedly.
const samples: number[] = [];
for (let i = 0; i < 5; i++) {
const r = await engine.executeMetric(VICTIM, { metric: "victim_metric", aggregate: "count" });
const r = await engine.executeMetric(VICTIM, { metric: 'victim_metric', aggregate: 'count' });
expect(r.rows[0].value).toBe(500); // still CORRECT, not just fast
samples.push(r.stats.duration_ms);
}
const bombResult = await bomb;
expect(bombResult.ok, "the bomb was NOT terminated").toBe(false);
expect(bombResult.code).toBe("159");
expect(bombResult.ok, 'the bomb was NOT terminated').toBe(false);
expect(bombResult.code).toBe('159');
const worst = Math.max(...samples);
expect(
@@ -213,24 +214,24 @@ run("BI-P012 resource governance against a LIVE ClickHouse", () => {
// ── the local admission gate, on the real engine ────────────────────────
it("the per-tenant concurrency gate refuses the N+1st in-flight query for THAT tenant only", async () => {
it('the per-tenant concurrency gate refuses the N+1st in-flight query for THAT tenant only', async () => {
const limit = DEFAULT_BUDGETS.INTERACTIVE.maxConcurrentQueries;
// Fire limit+4 concurrently; some must be refused with RATE_LIMITED
// rather than all of them reaching the server.
const attempts = Array.from({ length: limit + 4 }, () =>
engine
.executeMetric(VICTIM, { metric: "victim_metric", aggregate: "count" })
.then(() => "ok" as const)
.catch((e: { code?: string }) => e.code ?? "error"),
.executeMetric(VICTIM, { metric: 'victim_metric', aggregate: 'count' })
.then(() => 'ok' as const)
.catch((e: { code?: string }) => e.code ?? 'error'),
);
const outcomes = await Promise.all(attempts);
expect(outcomes.filter((o) => o === "RATE_LIMITED").length).toBeGreaterThan(0);
expect(outcomes.filter((o) => o === "ok").length).toBeGreaterThan(0);
expect(outcomes.filter((o) => o === 'RATE_LIMITED').length).toBeGreaterThan(0);
expect(outcomes.filter((o) => o === 'ok').length).toBeGreaterThan(0);
// A different tenant is unaffected by the noisy one's exhausted slots.
const other = await engine.executeMetric(
{ tenant_id: "t_quiet", project_id: "p_main", principal_id: "usr_q" },
{ metric: "victim_metric", aggregate: "count" },
{ tenant_id: 't_quiet', project_id: 'p_main', principal_id: 'usr_q' },
{ metric: 'victim_metric', aggregate: 'count' },
);
expect(other.rows[0].value).toBe(0); // its own (empty) data, and no error
}, 60_000);
@@ -26,16 +26,16 @@
* `integration` job sets it from a service container and hard-fails if the
* suites skip, so this is not a self-skipping green in practice.
*/
import { describe, it, expect, beforeAll, afterAll } from "vitest";
import { ClickHouseAnalyticsEngine } from "@/src/analytics/engine/clickhouse-engine";
import { describe, it, expect, beforeAll, afterAll } from 'vitest';
import { ClickHouseAnalyticsEngine } from '@/src/analytics/engine/clickhouse-engine';
import {
compileMetricQuery,
dimensionCatalogFrom,
restrictedTenantUserDdl,
} from "@/src/analytics/engine/query-compiler";
import { ANALYTICS_TABLES, sortingKeySql } from "@/src/analytics/engine/schema";
import { DEFAULT_BUDGETS } from "@/src/analytics/engine/resource-governance";
import type { AnalyticsIdentity } from "@/src/analytics/engine/siax-analytics-engine";
} from '@/src/analytics/engine/query-compiler';
import { ANALYTICS_TABLES, sortingKeySql } from '@/src/analytics/engine/schema';
import { DEFAULT_BUDGETS } from '@/src/analytics/engine/resource-governance';
import type { AnalyticsIdentity } from '@/src/analytics/engine/siax-analytics-engine';
const CLICKHOUSE_URL = process.env.CLICKHOUSE_URL;
const CLICKHOUSE_AUTH = process.env.CLICKHOUSE_AUTH;
@@ -45,10 +45,10 @@ const DB = `opt_isolation_${Date.now().toString(36)}`;
const RESTRICTED_USER = `opt_iso_${Date.now().toString(36)}`;
const RESTRICTED_PW = `iso_pw_${Math.random().toString(36).slice(2, 12)}`;
const ALPHA: AnalyticsIdentity = { tenant_id: "t_alpha", project_id: "p_main", principal_id: "usr_alpha" };
const BETA: AnalyticsIdentity = { tenant_id: "t_beta", project_id: "p_main", principal_id: "usr_beta" };
const ALPHA: AnalyticsIdentity = { tenant_id: 't_alpha', project_id: 'p_main', principal_id: 'usr_alpha' };
const BETA: AnalyticsIdentity = { tenant_id: 't_beta', project_id: 'p_main', principal_id: 'usr_beta' };
const catalog = dimensionCatalogFrom({ severity: "severity" });
const catalog = dimensionCatalogFrom({ severity: 'severity' });
const ALL_TABLES: readonly string[] = ANALYTICS_TABLES;
/** Raw HTTP against the live server, as a named user. */
@@ -59,34 +59,37 @@ async function chQuery(
settings: Record<string, string> = {},
body?: string,
): Promise<{ ok: boolean; status: number; text: string }> {
const url = new URL(CLICKHOUSE_URL ?? "http://skipped.invalid");
url.searchParams.set("query", sql);
const url = new URL(CLICKHOUSE_URL ?? 'http://skipped.invalid');
url.searchParams.set('query', sql);
for (const [k, v] of Object.entries(settings)) url.searchParams.set(k, v);
for (const [k, v] of Object.entries(params)) url.searchParams.set(k, v);
const headers: Record<string, string> = {};
if (auth) headers["Authorization"] = `Basic ${Buffer.from(auth).toString("base64")}`;
const res = await fetch(url.toString(), { method: "POST", headers, body: body ?? "" });
if (auth) headers['Authorization'] = `Basic ${Buffer.from(auth).toString('base64')}`;
const res = await fetch(url.toString(), { method: 'POST', headers, body: body ?? '' });
return { ok: res.ok, status: res.status, text: await res.text() };
}
function rowsOf(text: string): { grp?: string; value: number }[] {
if (!text.trim()) return [];
const parsed = JSON.parse(text) as { data?: { grp?: string; value: number | string }[] };
return (parsed.data ?? []).map((r) => ({ ...(r.grp !== undefined ? { grp: r.grp } : {}), value: Number(r.value) }));
return (parsed.data ?? []).map((r) => ({
...(r.grp !== undefined ? { grp: r.grp } : {}),
value: Number(r.value),
}));
}
let engine: ClickHouseAnalyticsEngine;
run("BI-P011 tenant isolation against a LIVE ClickHouse", () => {
run('BI-P011 tenant isolation against a LIVE ClickHouse', () => {
beforeAll(async () => {
engine = new ClickHouseAnalyticsEngine({
baseUrl: CLICKHOUSE_URL!,
basicAuth: CLICKHOUSE_AUTH,
database: DB,
dimensionCatalogs: { t_alpha: { severity: "severity" }, t_beta: { severity: "severity" } },
dimensionCatalogs: { t_alpha: { severity: 'severity' }, t_beta: { severity: 'severity' } },
});
const health = await engine.health();
expect(health.healthy, `ClickHouse not healthy: ${health.detail ?? ""}`).toBe(true);
expect(health.healthy, `ClickHouse not healthy: ${health.detail ?? ''}`).toBe(true);
await engine.ensureSchema();
// Two tenants, deliberately different magnitudes so a leak is unmistakable
@@ -95,28 +98,28 @@ run("BI-P011 tenant isolation against a LIVE ClickHouse", () => {
for (let i = 0; i < 50; i++) {
rows.push(
JSON.stringify({
tenant_id: "t_alpha",
project_id: "p_main",
dataset_id: "ds_shared",
metric: "rows_ingested",
tenant_id: 't_alpha',
project_id: 'p_main',
dataset_id: 'ds_shared',
metric: 'rows_ingested',
event_id: `a${i}`,
event_time: "2026-09-01 10:00:00.000",
event_time: '2026-09-01 10:00:00.000',
value: 1,
dim_names: ["severity"],
dim_values: [i % 2 === 0 ? "warning" : "critical"],
dim_names: ['severity'],
dim_values: [i % 2 === 0 ? 'warning' : 'critical'],
}),
);
rows.push(
JSON.stringify({
tenant_id: "t_beta",
project_id: "p_main",
dataset_id: "ds_shared",
metric: "rows_ingested",
tenant_id: 't_beta',
project_id: 'p_main',
dataset_id: 'ds_shared',
metric: 'rows_ingested',
event_id: `b${i}`,
event_time: "2026-09-01 10:00:00.000",
event_time: '2026-09-01 10:00:00.000',
value: 1000,
dim_names: ["severity"],
dim_values: ["critical"],
dim_names: ['severity'],
dim_values: ['critical'],
}),
);
}
@@ -125,7 +128,7 @@ run("BI-P011 tenant isolation against a LIVE ClickHouse", () => {
CLICKHOUSE_AUTH,
{},
{},
rows.join("\n"),
rows.join('\n'),
);
expect(ins.ok, `seed insert failed: ${ins.text.slice(0, 200)}`).toBe(true);
}, 120_000);
@@ -134,7 +137,10 @@ run("BI-P011 tenant isolation against a LIVE ClickHouse", () => {
if (!CLICKHOUSE_URL) return;
// Disposable by construction: drop the user, its policies, and the DB.
for (const table of ANALYTICS_TABLES) {
await chQuery(`DROP ROW POLICY IF EXISTS ${RESTRICTED_USER}_${table}_rp ON ${DB}.${table}`, CLICKHOUSE_AUTH);
await chQuery(
`DROP ROW POLICY IF EXISTS ${RESTRICTED_USER}_${table}_rp ON ${DB}.${table}`,
CLICKHOUSE_AUTH,
);
}
await chQuery(`DROP USER IF EXISTS ${RESTRICTED_USER}`, CLICKHOUSE_AUTH);
await chQuery(`DROP DATABASE IF EXISTS ${DB}`, CLICKHOUSE_AUTH);
@@ -149,89 +155,89 @@ run("BI-P011 tenant isolation against a LIVE ClickHouse", () => {
const unsafeSql =
`SELECT sum(value) AS value FROM ${DB}.opt_analytics_event ` +
`WHERE metric = {metric:String} LIMIT 1 FORMAT JSON`;
const res = await chQuery(unsafeSql, CLICKHOUSE_AUTH, { param_metric: "rows_ingested" });
const res = await chQuery(unsafeSql, CLICKHOUSE_AUTH, { param_metric: 'rows_ingested' });
expect(res.ok).toBe(true);
// 50*1 (alpha) + 50*1000 (beta) = 50050 — both tenants, one result.
expect(rowsOf(res.text)[0].value).toBe(50050);
});
it("CLOSED, layer 2 PRESENT: the compiled query returns t_alpha's 50 and never t_beta's 50000", async () => {
const result = await engine.executeMetric(ALPHA, { metric: "rows_ingested", aggregate: "sum" });
const result = await engine.executeMetric(ALPHA, { metric: 'rows_ingested', aggregate: 'sum' });
expect(result.rows[0].value).toBe(50);
const beta = await engine.executeMetric(BETA, { metric: "rows_ingested", aggregate: "sum" });
const beta = await engine.executeMetric(BETA, { metric: 'rows_ingested', aggregate: 'sum' });
expect(beta.rows[0].value).toBe(50_000);
});
it("tenant A querying tenant B's dataset id gets ZERO rows, not B's rows", async () => {
const result = await engine.executeMetric(ALPHA, {
metric: "rows_ingested",
aggregate: "count",
dataset_id: "ds_shared",
metric: 'rows_ingested',
aggregate: 'count',
dataset_id: 'ds_shared',
});
expect(result.rows[0].value).toBe(50); // only alpha's 50, though beta shares the dataset id
});
it("a forged dataset id yields an empty result, never an error that leaks existence", async () => {
it('a forged dataset id yields an empty result, never an error that leaks existence', async () => {
const result = await engine.executeMetric(ALPHA, {
metric: "rows_ingested",
aggregate: "count",
dataset_id: "ds_that_does_not_exist_or_belongs_to_someone_else",
metric: 'rows_ingested',
aggregate: 'count',
dataset_id: 'ds_that_does_not_exist_or_belongs_to_someone_else',
});
expect(result.rows[0].value).toBe(0);
});
it("a tenant with no data at all sees zero, not everyone's aggregate", async () => {
const ghost: AnalyticsIdentity = { tenant_id: "t_gamma", project_id: "p_main", principal_id: "usr_g" };
const result = await engine.executeMetric(ghost, { metric: "rows_ingested", aggregate: "count" });
const ghost: AnalyticsIdentity = { tenant_id: 't_gamma', project_id: 'p_main', principal_id: 'usr_g' };
const result = await engine.executeMetric(ghost, { metric: 'rows_ingested', aggregate: 'count' });
expect(result.rows[0].value).toBe(0);
});
it("group-by results carry only this tenant's dimension values", async () => {
const result = await engine.executeMetric(ALPHA, {
metric: "rows_ingested",
aggregate: "count",
group_by: ["severity"],
metric: 'rows_ingested',
aggregate: 'count',
group_by: ['severity'],
});
const total = result.rows.reduce((a, r) => a + r.value, 0);
expect(total).toBe(50);
// t_beta's rows are all "critical" with value 1000; if any leaked, the
// critical bucket would exceed alpha's 25.
const critical = result.rows.find((r) => r.groups?.severity === "critical");
const critical = result.rows.find((r) => r.groups?.severity === 'critical');
expect(critical?.value).toBe(25);
});
it("SQL/dimension/filter injection through the live path changes nothing", async () => {
it('SQL/dimension/filter injection through the live path changes nothing', async () => {
await expect(
engine.executeMetric(ALPHA, {
metric: "rows_ingested",
aggregate: "sum",
metric: 'rows_ingested',
aggregate: 'sum',
group_by: ["severity'; SELECT tenant_id FROM system.tables; --"],
}),
).rejects.toMatchObject({ code: "INVALID_REQUEST" });
).rejects.toMatchObject({ code: 'INVALID_REQUEST' });
// A hostile VALUE reaches the server as a bound parameter and simply
// matches nothing.
const injected = await engine.executeMetric(ALPHA, {
metric: "rows_ingested",
aggregate: "count",
metric: 'rows_ingested',
aggregate: 'count',
filters: { severity: "warning' OR 1=1 --" },
});
expect(injected.rows[0].value).toBe(0);
// And the table is still there afterwards.
const still = await engine.executeMetric(ALPHA, { metric: "rows_ingested", aggregate: "count" });
const still = await engine.executeMetric(ALPHA, { metric: 'rows_ingested', aggregate: 'count' });
expect(still.rows[0].value).toBe(50);
});
it("raw SQL attempt: the analyst-SQL surface is refused, not silently executed", async () => {
it('raw SQL attempt: the analyst-SQL surface is refused, not silently executed', async () => {
await expect(
engine.executeQuery(ALPHA, { sql: `SELECT * FROM ${DB}.opt_analytics_event`, dataset_ids: [] }),
).rejects.toMatchObject({ code: "CAPABILITY_UNAVAILABLE" });
).rejects.toMatchObject({ code: 'CAPABILITY_UNAVAILABLE' });
});
// ── layer 3: restricted users + row policies ─────────────────────────────
it("HOLE, layer 3 ABSENT: an unrestricted user with no predicate sees every tenant", async () => {
it('HOLE, layer 3 ABSENT: an unrestricted user with no predicate sees every tenant', async () => {
const res = await chQuery(
`SELECT uniqExact(tenant_id) AS value FROM ${DB}.opt_analytics_event FORMAT JSON`,
CLICKHOUSE_AUTH,
@@ -240,10 +246,10 @@ run("BI-P011 tenant isolation against a LIVE ClickHouse", () => {
expect(rowsOf(res.text)[0].value).toBe(2); // t_alpha AND t_beta
});
it("CLOSED, layer 3 PRESENT: the restricted user runs the SAME predicate-free statement and sees ONE tenant", async () => {
it('CLOSED, layer 3 PRESENT: the restricted user runs the SAME predicate-free statement and sees ONE tenant', async () => {
for (const stmt of restrictedTenantUserDdl({
db: DB,
tenantId: "t_alpha",
tenantId: 't_alpha',
userName: RESTRICTED_USER,
password: RESTRICTED_PW,
tables: ALL_TABLES,
@@ -268,14 +274,24 @@ run("BI-P011 tenant isolation against a LIVE ClickHouse", () => {
expect(rowsOf(sum.text)[0].value).toBe(50); // alpha's total, not 50050
});
it("the restricted user cannot write, and cannot reach a table it was not granted", async () => {
it('the restricted user cannot write, and cannot reach a table it was not granted', async () => {
const restrictedAuth = `${RESTRICTED_USER}:${RESTRICTED_PW}`;
const write = await chQuery(
`INSERT INTO ${DB}.opt_analytics_event FORMAT JSONEachRow`,
restrictedAuth,
{},
{},
JSON.stringify({ tenant_id: "t_beta", project_id: "p", dataset_id: "d", metric: "m", event_id: "x", event_time: "2026-09-01 00:00:00.000", value: 1, dim_names: [], dim_values: [] }),
JSON.stringify({
tenant_id: 't_beta',
project_id: 'p',
dataset_id: 'd',
metric: 'm',
event_id: 'x',
event_time: '2026-09-01 00:00:00.000',
value: 1,
dim_names: [],
dim_values: [],
}),
);
expect(write.ok).toBe(false);
@@ -289,18 +305,18 @@ run("BI-P011 tenant isolation against a LIVE ClickHouse", () => {
for (const table of ALL_TABLES) {
const res = await chQuery(sortingKeySql(DB, table), CLICKHOUSE_AUTH);
expect(res.ok).toBe(true);
expect(res.text.trim().startsWith("tenant_id"), `${table} sorting key: ${res.text.trim()}`).toBe(true);
expect(res.text.trim().startsWith('tenant_id'), `${table} sorting key: ${res.text.trim()}`).toBe(true);
}
});
it("the tenant value never appears in the statement text the server receives", async () => {
it('the tenant value never appears in the statement text the server receives', async () => {
const compiled = compileMetricQuery(
{ tenant_id: "t_alpha", metric: "rows_ingested", aggregate: "sum" },
{ tenant_id: 't_alpha', metric: 'rows_ingested', aggregate: 'sum' },
DB,
catalog,
DEFAULT_BUDGETS.INTERACTIVE.maxResultRows,
);
expect(compiled.sql).not.toContain("t_alpha");
expect(compiled.sql).not.toContain('t_alpha');
const res = await chQuery(compiled.sql, CLICKHOUSE_AUTH, compiled.params);
expect(res.ok).toBe(true);
expect(rowsOf(res.text)[0].value).toBe(50);
@@ -0,0 +1,259 @@
/**
* tests/integration/query-cancellation-budget.integration.test.ts
* ================================================================
* BI-P019 (cancellation) and BI-P021 (cost/budget) against a live ClickHouse.
*
* Both prompts are only real if something ACTUALLY STOPS. So both are tested
* by starting a genuine CPU bomb and requiring the server to stop working:
*
* * the query must disappear from `system.processes`;
* * `system.query_log` must record it as ExceptionWhileProcessing with
* ClickHouse code 394 (QUERY_WAS_CANCELLED) — the server's own account of
* what happened, not ours;
* * and the COUNTERFACTUAL must hold: an identical bomb left alone for the
* same interval is still running. Without that, "the query was gone after
* we cancelled it" is equally consistent with "the query was short".
*
* BI-P021's in-flight enforcement gets the same treatment: a tenant window
* budget of one million rows, a query that scans far more, and the requirement
* that the enforcer cancels it MID-FLIGHT with `rows_read` already non-zero —
* i.e. it was running, it was observed running, and it was stopped.
*
* Gated on CLICKHOUSE_URL. With the variable set and the server down, this
* suite FAILS; it does not degrade to a skip.
*/
import fs from 'node:fs';
import path from 'node:path';
import { beforeAll, describe, expect, it } from 'vitest';
import { ClickHouseProcessControl } from '@/src/analytics/execution/clickhouse-process-control';
import { cancelAndVerify, cancelOrThrow } from '@/src/analytics/execution/cancellation';
import {
DEFAULT_QUERY_BUDGET,
InMemoryWindowStore,
QueryBudgetEnforcer,
} from '@/src/analytics/cost/query-budget';
const BASE_URL = process.env.CLICKHOUSE_URL ?? '';
const AUTH = process.env.CLICKHOUSE_AUTH ?? 'default:';
const suite = BASE_URL ? describe : describe.skip;
const EVIDENCE_PATH = path.join(process.cwd(), 'evidence/bi-p019-p021/cancellation-and-budget.json');
/** Big enough that it cannot finish inside any of the windows below. */
const BOMB = 'SELECT sum(number) AS s FROM numbers_mt(200000000000)';
interface Res {
ok: boolean;
status: number;
body: string;
}
async function send(sql: string, opts: { queryId?: string; timeoutMs?: number } = {}): Promise<Res> {
const url = new URL(BASE_URL);
url.searchParams.set('query', sql);
if (opts.queryId) url.searchParams.set('query_id', opts.queryId);
try {
const res = await fetch(url.toString(), {
method: 'POST',
headers: { Authorization: `Basic ${Buffer.from(AUTH).toString('base64')}` },
body: '',
signal: AbortSignal.timeout(opts.timeoutMs ?? 120_000),
});
return { ok: res.ok, status: res.status, body: await res.text() };
} catch (e) {
return { ok: false, status: 0, body: String(e) };
}
}
/** Fire and forget: the bomb runs on the server while the test carries on. */
function launch(queryId: string): Promise<Res> {
return send(BOMB, { queryId, timeoutMs: 120_000 });
}
const sleep = (ms: number) => new Promise((r) => setTimeout(r, ms));
const control = new ClickHouseProcessControl({ baseUrl: BASE_URL || 'http://unset', basicAuth: AUTH });
const evidence: Record<string, unknown> = {};
async function waitUntilRunning(queryId: string, timeoutMs = 10_000): Promise<boolean> {
const deadline = Date.now() + timeoutMs;
while (Date.now() < deadline) {
if (await control.isRunning(queryId)) return true;
await sleep(50);
}
return false;
}
/** ClickHouse's own verdict, from its query log. 394 = QUERY_WAS_CANCELLED. */
async function queryLogVerdict(
queryId: string,
): Promise<{ type: string; exception_code: number } | undefined> {
await send('SYSTEM FLUSH LOGS');
const res = await send(
`SELECT type, exception_code FROM system.query_log WHERE query_id = '${queryId}' ` +
`AND type != 'QueryStart' ORDER BY event_time_microseconds DESC LIMIT 1 FORMAT JSON`,
);
if (!res.ok) return undefined;
const row = (JSON.parse(res.body) as { data: Array<{ type: string; exception_code: number | string }> })
.data[0];
return row ? { type: row.type, exception_code: Number(row.exception_code) } : undefined;
}
beforeAll(async () => {
if (!BASE_URL) return;
const v = await send('SELECT version()');
if (!v.ok) throw new Error(`CLICKHOUSE_URL is set but ClickHouse is not answering: ${v.status}`);
evidence.clickhouse_version = v.body.trim();
evidence.recorded_at = new Date().toISOString();
}, 60_000);
suite('BI-P019 — cancellation stops an expensive query, and it is VERIFIED', () => {
it('counterfactual: an identical bomb left alone is still running after the same interval', async () => {
const queryId = `opt_cf_${Date.now()}`;
void launch(queryId);
expect(await waitUntilRunning(queryId)).toBe(true);
await sleep(2_000);
const stillRunning = await control.isRunning(queryId);
evidence.counterfactual_still_running_after_2s = stillRunning;
expect(stillRunning, 'the bomb finished on its own — it is not expensive enough to prove anything').toBe(
true,
);
// Clean up, and incidentally exercise the same path.
await cancelAndVerify(control, queryId, { timeoutMs: 10_000 });
}, 90_000);
it("cancellation is observed to stop it, and ClickHouse's own log agrees", async () => {
const queryId = `opt_cancel_${Date.now()}`;
void launch(queryId);
expect(await waitUntilRunning(queryId)).toBe(true);
// Let it accumulate real work so "it stopped" is a statement about a
// running query, not about one that had barely begun.
await sleep(1_000);
const progressBefore = await control.progress(queryId);
expect(progressBefore?.rows_read ?? 0).toBeGreaterThan(0);
const outcome = await cancelAndVerify(control, queryId, { timeoutMs: 10_000, pollIntervalMs: 100 });
expect(outcome.kill_accepted).toBe(true);
expect(outcome.verified_stopped, 'the query outlived its cancellation deadline').toBe(true);
expect(outcome.already_finished).toBe(false);
expect(outcome.elapsed_ms).toBeLessThan(10_000);
expect(await control.isRunning(queryId)).toBe(false);
const verdict = await queryLogVerdict(queryId);
evidence.cancellation = {
rows_read_before_cancel: progressBefore?.rows_read,
elapsed_ms_to_verified_stop: outcome.elapsed_ms,
verified_stopped: outcome.verified_stopped,
clickhouse_query_log_type: verdict?.type,
clickhouse_exception_code: verdict?.exception_code,
};
// 394 = QUERY_WAS_CANCELLED. The server saying it, not us.
expect(verdict?.exception_code, `query_log verdict was ${JSON.stringify(verdict)}`).toBe(394);
}, 90_000);
it('cancelling a query that already finished reports already_finished, not a false stop', async () => {
const queryId = `opt_done_${Date.now()}`;
const res = await send('SELECT 1', { queryId });
expect(res.ok).toBe(true);
const outcome = await cancelAndVerify(control, queryId, { timeoutMs: 2_000 });
expect(outcome.already_finished).toBe(true);
expect(outcome.verified_stopped).toBe(true);
});
it('cancelOrThrow raises when the query is not observed to stop in time', async () => {
const queryId = `opt_deadline_${Date.now()}`;
void launch(queryId);
expect(await waitUntilRunning(queryId)).toBe(true);
// A deadline shorter than ClickHouse can plausibly unwind in, with the
// kill request suppressed so the query genuinely keeps running. The
// point is that the module REPORTS a failure to stop rather than
// returning a cheerful boolean.
const noKill = {
engine: 'clickhouse',
isRunning: (id: string) => control.isRunning(id),
requestKill: async () => false,
progress: (id: string) => control.progress(id),
};
await expect(cancelOrThrow(noKill, queryId, { timeoutMs: 300, pollIntervalMs: 50 })).rejects.toThrow(
/not observed to stop/,
);
await cancelAndVerify(control, queryId, { timeoutMs: 10_000 });
}, 90_000);
});
suite('BI-P021 — the budget actually stops a running query', () => {
it('an in-flight window breach cancels the query mid-scan, with the cancellation verified', async () => {
const store = new InMemoryWindowStore();
const settled: unknown[] = [];
const enforcer = new QueryBudgetEnforcer({
budget: { ...DEFAULT_QUERY_BUDGET, maxRowsScanned: 1_000_000, maxCostUnits: 1e9 },
store,
ledger: { record: (e) => void settled.push(e) },
control,
enforceInFlight: true,
pollIntervalMs: 100,
storeIsShared: false,
});
const queryId = `opt_budget_${Date.now()}`;
void launch(queryId);
expect(await waitUntilRunning(queryId)).toBe(true);
const outcome = await enforcer.enforceInFlight('t_alpha', queryId);
expect(outcome.terminated, 'the budget watched the query run past its window and did nothing').toBe(true);
expect(outcome.reason).toBe('rows');
// It was RUNNING when it was stopped — not stopped before it started.
expect(outcome.observed.rows_read).toBeGreaterThan(1_000_000);
expect(outcome.cancellation?.verified_stopped).toBe(true);
expect(await control.isRunning(queryId)).toBe(false);
const verdict = await queryLogVerdict(queryId);
expect(verdict?.exception_code).toBe(394);
await enforcer.settle({
tenant_id: 't_alpha',
principal_id: 'u_1',
query_id: queryId,
cost: {
rows_scanned: outcome.observed.rows_read,
bytes_scanned: outcome.observed.bytes_read,
execution_ms: outcome.observed.elapsed_ms,
engine: 'clickhouse',
},
terminated_by_budget: true,
});
expect(settled).toHaveLength(1);
evidence.budget_in_flight = {
budget_max_rows_scanned: 1_000_000,
observed_rows_read_at_cancel: outcome.observed.rows_read,
observed_elapsed_ms_at_cancel: outcome.observed.elapsed_ms,
polls_before_breach: outcome.polls,
overshoot_factor: Math.round(outcome.observed.rows_read / 1_000_000),
overshoot_note:
'in-flight enforcement is granular to the poll interval; the query overshot the window ' +
'by this factor before the poll that caught it. A bound, not a hard cap.',
cancellation_verified: outcome.cancellation?.verified_stopped,
clickhouse_exception_code: verdict?.exception_code,
charged_for_work_done_before_cancel: true,
};
}, 120_000);
it('admission-only mode reports itself as advisory rather than as a control', () => {
const enforcer = new QueryBudgetEnforcer({
budget: DEFAULT_QUERY_BUDGET,
store: new InMemoryWindowStore(),
ledger: { record: () => undefined },
enforceInFlight: false,
});
const described = enforcer.describeEnforcement();
expect(described.binding).toBe(false);
expect(described.in_flight).toContain('NOT ENFORCED');
evidence.advisory_posture_is_declared = described;
});
it('writes the evidence file', () => {
fs.mkdirSync(path.dirname(EVIDENCE_PATH), { recursive: true });
fs.writeFileSync(EVIDENCE_PATH, `${JSON.stringify(evidence, null, 2)}\n`);
expect(fs.existsSync(EVIDENCE_PATH)).toBe(true);
});
});
@@ -15,10 +15,10 @@
* - full projection-disaster recovery: delete EVERYTHING, rebuild from
* the canonical source, identical results (§68-style, scoped to 0pt)
*/
import { describe, it, expect, beforeAll, afterAll } from "vitest";
import { HttpOpenSearchAdapter } from "@/src/search/opensearch-adapter";
import { SearchService, type DocumentSource } from "@/src/search/search-service";
import type { SearchDocument } from "@/src/search/search-types";
import { describe, it, expect, beforeAll, afterAll } from 'vitest';
import { HttpOpenSearchAdapter } from '@/src/search/opensearch-adapter';
import { SearchService, type DocumentSource } from '@/src/search/search-service';
import type { SearchDocument } from '@/src/search/search-types';
const OPENSEARCH_URL = process.env.OPENSEARCH_URL;
const run = OPENSEARCH_URL ? describe : describe.skip;
@@ -27,52 +27,52 @@ const INDEX = `itest-${Date.now().toString(36)}`;
const DOCS: SearchDocument[] = [
{
doc_id: "c1",
tenant_id: "t_alpha",
purpose_tags: ["company_enrichment"],
rights_tags: ["internal", "customer_deliverable"],
source_refs: ["0pt:source:src_bolagsverket"],
payload: { name: "Volvo Personvagnar AB", org_nr: "5561074967", employees: 20000 },
doc_id: 'c1',
tenant_id: 't_alpha',
purpose_tags: ['company_enrichment'],
rights_tags: ['internal', 'customer_deliverable'],
source_refs: ['0pt:source:src_bolagsverket'],
payload: { name: 'Volvo Personvagnar AB', org_nr: '5561074967', employees: 20000 },
},
{
doc_id: "c2",
tenant_id: "t_alpha",
purpose_tags: ["compliance"],
rights_tags: ["internal"],
payload: { name: "Volvo Lastvagnar AB", org_nr: "5560135508", employees: 15000 },
doc_id: 'c2',
tenant_id: 't_alpha',
purpose_tags: ['compliance'],
rights_tags: ['internal'],
payload: { name: 'Volvo Lastvagnar AB', org_nr: '5560135508', employees: 15000 },
},
{
doc_id: "c3",
tenant_id: "t_beta",
purpose_tags: ["company_enrichment"],
rights_tags: ["internal"],
payload: { name: "Volvo Cars Retail", org_nr: "5569999999", employees: 300 },
doc_id: 'c3',
tenant_id: 't_beta',
purpose_tags: ['company_enrichment'],
rights_tags: ['internal'],
payload: { name: 'Volvo Cars Retail', org_nr: '5569999999', employees: 300 },
},
];
function source(docs: SearchDocument[]): DocumentSource {
return {
name: "itest:canonical-array",
name: 'itest:canonical-array',
async *iterate() {
yield* docs;
},
};
}
run("Search capability against a LIVE OpenSearch", () => {
run('Search capability against a LIVE OpenSearch', () => {
// The skip-variant still evaluates this callback body, so the constructor
// must tolerate the env-less case (the tests themselves never run then).
const adapter = new HttpOpenSearchAdapter({ baseUrl: OPENSEARCH_URL ?? "http://skipped.invalid" });
const adapter = new HttpOpenSearchAdapter({ baseUrl: OPENSEARCH_URL ?? 'http://skipped.invalid' });
const search = new SearchService(adapter);
beforeAll(async () => {
const health = await adapter.health();
expect(["green", "yellow"]).toContain(health.status);
expect(['green', 'yellow']).toContain(health.status);
search.registerIndexDefinition({
index_name: INDEX,
version: 1,
fields: { name: "text", org_nr: "keyword", employees: "long" },
default_search_fields: ["name"],
fields: { name: 'text', org_nr: 'keyword', employees: 'long' },
default_search_fields: ['name'],
});
}, 30_000);
@@ -82,7 +82,7 @@ run("Search capability against a LIVE OpenSearch", () => {
}
});
it("rebuild g1 from canonical source; alias serves it", async () => {
it('rebuild g1 from canonical source; alias serves it', async () => {
const result = await search.rebuild(INDEX, source(DOCS));
expect(result.new_generation).toBe(1);
expect(result.docs_indexed).toBe(3);
@@ -91,57 +91,65 @@ run("Search capability against a LIVE OpenSearch", () => {
expect(state.active_generation).toBe(1);
}, 60_000);
it("REAL tenant isolation: t_alpha never sees t_beta documents", async () => {
it('REAL tenant isolation: t_alpha never sees t_beta documents', async () => {
const alpha = await search.query(INDEX, {
tenant_id: "t_alpha",
purpose: "company_enrichment",
text: "volvo",
tenant_id: 't_alpha',
purpose: 'company_enrichment',
text: 'volvo',
});
expect(alpha.hits.map((h) => h.doc_id)).toEqual(["c1"]); // c2 wrong purpose, c3 wrong tenant
expect(alpha.hits.map((h) => h.doc_id)).toEqual(['c1']); // c2 wrong purpose, c3 wrong tenant
const beta = await search.query(INDEX, {
tenant_id: "t_beta",
purpose: "company_enrichment",
text: "volvo",
tenant_id: 't_beta',
purpose: 'company_enrichment',
text: 'volvo',
});
expect(beta.hits.map((h) => h.doc_id)).toEqual(["c3"]);
expect(beta.hits.map((h) => h.doc_id)).toEqual(['c3']);
// Cross-tenant attack shape: a filter cannot break out of the tenant term.
const attack = await search.query(INDEX, {
tenant_id: "t_beta",
purpose: "company_enrichment",
filters: { org_nr: "5561074967" }, // t_alpha's org_nr
tenant_id: 't_beta',
purpose: 'company_enrichment',
filters: { org_nr: '5561074967' }, // t_alpha's org_nr
});
expect(attack.total).toBe(0);
});
it("purpose + rights filtering hold on the real engine", async () => {
const compliance = await search.query(INDEX, { tenant_id: "t_alpha", purpose: "compliance" });
expect(compliance.hits.map((h) => h.doc_id)).toEqual(["c2"]);
it('purpose + rights filtering hold on the real engine', async () => {
const compliance = await search.query(INDEX, { tenant_id: 't_alpha', purpose: 'compliance' });
expect(compliance.hits.map((h) => h.doc_id)).toEqual(['c2']);
const deliverable = await search.query(INDEX, {
tenant_id: "t_alpha",
purpose: "company_enrichment",
rights_tag: "customer_deliverable",
tenant_id: 't_alpha',
purpose: 'company_enrichment',
rights_tag: 'customer_deliverable',
});
expect(deliverable.hits.map((h) => h.doc_id)).toEqual(["c1"]);
expect(deliverable.hits.map((h) => h.doc_id)).toEqual(['c1']);
});
it("generation rebuild + switch + rollback (§42)", async () => {
it('generation rebuild + switch + rollback (§42)', async () => {
const updated = DOCS.map((d) =>
d.doc_id === "c1" ? { ...d, payload: { ...d.payload, name: "Volvo Personvagnar AB (uppdaterad)" } } : d
d.doc_id === 'c1' ? { ...d, payload: { ...d.payload, name: 'Volvo Personvagnar AB (uppdaterad)' } } : d,
);
const r2 = await search.rebuild(INDEX, source(updated));
expect(r2.new_generation).toBe(2);
let res = await search.query(INDEX, { tenant_id: "t_alpha", purpose: "company_enrichment", text: "uppdaterad" });
let res = await search.query(INDEX, {
tenant_id: 't_alpha',
purpose: 'company_enrichment',
text: 'uppdaterad',
});
expect(res.total).toBe(1);
// Rollback to g1 — the old generation still exists and serves the alias.
await search.switchGeneration(INDEX, 1);
res = await search.query(INDEX, { tenant_id: "t_alpha", purpose: "company_enrichment", text: "uppdaterad" });
res = await search.query(INDEX, {
tenant_id: 't_alpha',
purpose: 'company_enrichment',
text: 'uppdaterad',
});
expect(res.total).toBe(0);
await search.switchGeneration(INDEX, 2);
}, 60_000);
it("projection disaster: delete ALL generations, rebuild from canonical, results identical (§68)", async () => {
it('projection disaster: delete ALL generations, rebuild from canonical, results identical (§68)', async () => {
for (const idx of await adapter.listIndices(`opt-${INDEX}--g*`)) {
await adapter.deleteIndex(idx);
}
@@ -149,24 +157,32 @@ run("Search capability against a LIVE OpenSearch", () => {
const rebuilt = await search.rebuild(INDEX, source(DOCS));
expect(rebuilt.docs_indexed).toBe(3);
const res = await search.query(INDEX, { tenant_id: "t_alpha", purpose: "company_enrichment", text: "volvo" });
expect(res.hits.map((h) => h.doc_id)).toEqual(["c1"]);
const res = await search.query(INDEX, {
tenant_id: 't_alpha',
purpose: 'company_enrichment',
text: 'volvo',
});
expect(res.hits.map((h) => h.doc_id)).toEqual(['c1']);
}, 60_000);
it("incremental indexDocuments into the active generation", async () => {
it('incremental indexDocuments into the active generation', async () => {
const out = await search.indexDocuments(INDEX, [
{
doc_id: "c4",
tenant_id: "t_alpha",
purpose_tags: ["company_enrichment"],
rights_tags: ["internal"],
payload: { name: "Polestar Performance AB", org_nr: "5568100000", employees: 2500 },
doc_id: 'c4',
tenant_id: 't_alpha',
purpose_tags: ['company_enrichment'],
rights_tags: ['internal'],
payload: { name: 'Polestar Performance AB', org_nr: '5568100000', employees: 2500 },
},
]);
expect(out.indexed).toBe(1);
const state = await search.generationState(INDEX);
await adapter.refresh(search.physicalIndexFor(INDEX, state.active_generation!));
const res = await search.query(INDEX, { tenant_id: "t_alpha", purpose: "company_enrichment", text: "polestar" });
const res = await search.query(INDEX, {
tenant_id: 't_alpha',
purpose: 'company_enrichment',
text: 'polestar',
});
expect(res.total).toBe(1);
});
});