diff --git a/evidence/bi-p002-p012/bi-p005-arrow-roundtrip.json b/evidence/bi-p002-p012/bi-p005-arrow-roundtrip.json index 6e065aa..91cfd8b 100644 --- a/evidence/bi-p002-p012/bi-p005-arrow-roundtrip.json +++ b/evidence/bi-p002-p012/bi-p005-arrow-roundtrip.json @@ -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"] ] } }, diff --git a/evidence/bi-p019-p021/cancellation-and-budget.json b/evidence/bi-p019-p021/cancellation-and-budget.json new file mode 100644 index 0000000..d45ec0a --- /dev/null +++ b/evidence/bi-p019-p021/cancellation-and-budget.json @@ -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 + } +} diff --git a/src/analytics/cost/query-budget.ts b/src/analytics/cost/query-budget.ts new file mode 100644 index 0000000..5d1a5e9 --- /dev/null +++ b/src/analytics/cost/query-budget.ts @@ -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; +} + +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; + get(tenantId: string): Promise; +} + +export class InMemoryWindowStore implements SharedWindowStore { + private readonly entries = new Map(); + 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 { + 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 { + 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; +} + +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; + private readonly now: () => number; + private readonly sleep: (ms: number) => Promise; + + 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 { + 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 { + 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 }; + } +} diff --git a/src/analytics/execution/cancellation.ts b/src/analytics/execution/cancellation.ts new file mode 100644 index 0000000..3c0d61f --- /dev/null +++ b/src/analytics/execution/cancellation.ts @@ -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; + /** Ask the engine to stop it. Returning true means "accepted", not "stopped". */ + requestKill(queryId: string): Promise; + /** Live progress for the running query, when the engine exposes it (BI-P021). */ + progress?(queryId: string): Promise; +} + +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; +} + +export const DEFAULT_CANCELLATION_TIMEOUT_MS = 5_000; +const DEFAULT_POLL_INTERVAL_MS = 100; + +const defaultSleep = (ms: number): Promise => new Promise((r) => setTimeout(r, ms)); + +export async function cancelAndVerify( + control: EngineProcessControl, + queryId: string, + opts: CancellationOptions = {}, +): Promise { + 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 { + 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; +} diff --git a/src/analytics/execution/clickhouse-process-control.ts b/src/analytics/execution/clickhouse-process-control.ts new file mode 100644 index 0000000..5476497 --- /dev/null +++ b/src/analytics/execution/clickhouse-process-control.ts @@ -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 { + 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 { + 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 { + 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> }).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, expectJson = true): Promise { + 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 = {}; + 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; + } +} diff --git a/src/analytics/execution/query-execution-record.ts b/src/analytics/execution/query-execution-record.ts new file mode 100644 index 0000000..f847cc6 --- /dev/null +++ b/src/analytics/execution/query-execution-record.ts @@ -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 { + 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 = { + 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; +} + +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 { + await this.sink.write(this.current); + } + + async started(): Promise { + 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 { + 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 { + 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`, + ); + } + } +} diff --git a/tests/analytics/query-execution-model.test.ts b/tests/analytics/query-execution-model.test.ts new file mode 100644 index 0000000..3f1a8ab --- /dev/null +++ b/tests/analytics/query-execution-model.test.ts @@ -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); + }); +}); diff --git a/tests/integration/analytics-arrow-clickhouse.integration.test.ts b/tests/integration/analytics-arrow-clickhouse.integration.test.ts index 648ed06..b87e38f 100644 --- a/tests/integration/analytics-arrow-clickhouse.integration.test.ts +++ b/tests/integration/analytics-arrow-clickhouse.integration.test.ts @@ -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 { - 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 = {}): Promise { - 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 { - 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); diff --git a/tests/integration/analytics-clickhouse.integration.test.ts b/tests/integration/analytics-clickhouse.integration.test.ts index 8d072cb..64e9f00 100644 --- a/tests/integration/analytics-clickhouse.integration.test.ts +++ b/tests/integration/analytics-clickhouse.integration.test.ts @@ -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' }); }); }); diff --git a/tests/integration/analytics-engine-clickhouse.integration.test.ts b/tests/integration/analytics-engine-clickhouse.integration.test.ts index d798ca9..db88ff5 100644 --- a/tests/integration/analytics-engine-clickhouse.integration.test.ts +++ b/tests/integration/analytics-engine-clickhouse.integration.test.ts @@ -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 { - 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 = {}; - 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 }); }); diff --git a/tests/integration/analytics-resource-governance.integration.test.ts b/tests/integration/analytics-resource-governance.integration.test.ts index a0b5094..a4451da 100644 --- a/tests/integration/analytics-resource-governance.integration.test.ts +++ b/tests/integration/analytics-resource-governance.integration.test.ts @@ -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 = {}, 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 = {}; - 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); diff --git a/tests/integration/analytics-tenant-isolation.integration.test.ts b/tests/integration/analytics-tenant-isolation.integration.test.ts index 3a1f9c9..4d1f60f 100644 --- a/tests/integration/analytics-tenant-isolation.integration.test.ts +++ b/tests/integration/analytics-tenant-isolation.integration.test.ts @@ -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 = {}, 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 = {}; - 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); diff --git a/tests/integration/query-cancellation-budget.integration.test.ts b/tests/integration/query-cancellation-budget.integration.test.ts new file mode 100644 index 0000000..8c8a121 --- /dev/null +++ b/tests/integration/query-cancellation-budget.integration.test.ts @@ -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 { + 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 { + 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 = {}; + +async function waitUntilRunning(queryId: string, timeoutMs = 10_000): Promise { + 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); + }); +}); diff --git a/tests/integration/search-opensearch.integration.test.ts b/tests/integration/search-opensearch.integration.test.ts index d8f20e6..6ab96a4 100644 --- a/tests/integration/search-opensearch.integration.test.ts +++ b/tests/integration/search-opensearch.integration.test.ts @@ -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); }); });