perf(query-engine): ClickStack-style skip indexes + token-aware log search

alchemy.run.ts

@@ -1,6 +1,7 @@
 import { appendFileSync } from "node:fs"
 import * as Alchemy from "alchemy"
 import * as Cloudflare from "alchemy/Cloudflare"
+import * as RemovalPolicy from "alchemy/RemovalPolicy"
 import * as Effect from "effect/Effect"
 import { formatMapleStage, parseMapleStage, resolveMapleDomains } from "@maple/infra/cloudflare"
 import { createAlertingWorker } from "./apps/alerting/alchemy.run.ts"
@@ -27,6 +28,52 @@ if (!process.env.CLOUDFLARE_ACCOUNT_ID && process.env.CLOUDFLARE_DEFAULT_ACCOUNT
 const resolveUrl = (domain: string | undefined, envKey: string, fallback = ""): string =>
 	domain ? `https://${domain}` : process.env[envKey]?.trim() || fallback
 
+const requireEnv = (key: string): string => {
+	const value = process.env[key]?.trim()
+	if (!value) {
+		throw new Error(`Missing required deployment env: ${key}`)
+	}
+	return value
+}
+
+const createProductionSharedResources = (stage: ReturnType<typeof parseMapleStage>) =>
+	Effect.gen(function* () {
+		// Bootstrap these account/zone-wide resources in production first. Other
+		// stages keep their existing bindings until the production state has been
+		// deployed, after which they can safely resolve these logical IDs via ref().
+		if (stage.kind !== "prd") {
+			return {}
+		}
+
+		const ingestEndpoint = (process.env.MAPLE_ENDPOINT?.trim() || "https://ingest.maple.dev").replace(
+			/\/+$/,
+			"",
+		)
+		const headers = { authorization: `Bearer ${requireEnv("MAPLE_OTEL_INGEST_KEY")}` }
+		const tracesDestination = yield* Cloudflare.Workers.ObservabilityDestination(
+			"workers-observability-traces",
+			{
+				name: "maple-workers-traces",
+				url: `${ingestEndpoint}/v1/traces`,
+				headers,
+				logpushDataset: "opentelemetry-traces",
+				enabled: true,
+			},
+		).pipe(RemovalPolicy.retain())
+		const logsDestination = yield* Cloudflare.Workers.ObservabilityDestination(
+			"workers-observability-logs",
+			{
+				name: "maple-workers-logs",
+				url: `${ingestEndpoint}/v1/logs`,
+				headers,
+				logpushDataset: "opentelemetry-logs",
+				enabled: true,
+			},
+		).pipe(RemovalPolicy.retain())
+
+		return { logsDestination, tracesDestination }
+	})
+
 export default Alchemy.Stack(
 	"maple",
 	{
@@ -40,6 +87,7 @@ export default Alchemy.Stack(
 	Effect.gen(function* () {
 		const stage = parseMapleStage(yield* Alchemy.Stage)
 		const domains = resolveMapleDomains(stage)
+		const shared = yield* createProductionSharedResources(stage)
 
 		const apiUrl = resolveUrl(domains.api, "MAPLE_API_BASE_URL")
 		const chatUrl = resolveUrl(domains.chat, "MAPLE_CHAT_BASE_URL")
@@ -48,11 +96,23 @@ export default Alchemy.Stack(
 		// back to a caller-supplied env var or the public Maple ingest endpoint.
 		const ingestUrl = resolveUrl(domains.ingest, "VITE_INGEST_URL", "https://ingest.maple.dev")
 
-		// chat-flue deploys before api so api can service-bind the real worker (the
-		// v1 WorkerStub cycle-breaker is gone — chat-flue's api URL is now static).
-		const chatFlue = yield* createChatFlueWorker({ stage, domains, mapleApiUrl: apiUrl })
-
-		const { worker: api, db: mapleDb } = yield* createMapleApi({ stage, domains, chatFlue })
+		// Deploy the API RPC surface before switching chat-flue to it. The reverse
+		// CHAT_FLUE binding is attached after chat deploys, which breaks the resource
+		// cycle without an HTTP fallback or a placeholder Worker.
+		const { worker: api, db: mapleDb } = yield* createMapleApi({
+			stage,
+			domains,
+		})
+		const chatFlue = yield* createChatFlueWorker({
+			stage,
+			domains,
+			mapleApiRpc: api,
+			logsDestination: shared.logsDestination,
+			tracesDestination: shared.tracesDestination,
+		})
+		yield* api.bind("CHAT_FLUE", {
+			bindings: [{ type: "service", name: "CHAT_FLUE", service: chatFlue.workerName }],
+		})
 
 		// Standalone ElectricSQL shape-proxy worker (DB-free); its public origin is
 		// baked into the web build (VITE_ELECTRIC_SYNC_URL).
@@ -71,7 +131,11 @@ export default Alchemy.Stack(
 
 		const localUi = yield* createLocalUiWorker({ stage, domains })
 
-		const alerting = yield* createAlertingWorker({ stage, domains, mapleDb })
+		const alerting = yield* createAlertingWorker({
+			stage,
+			domains,
+			mapleDb,
+		})
 
 		const summary = {
 			stage: formatMapleStage(stage),

apps/api/alchemy.run.ts

@@ -1,7 +1,9 @@
 import path from "node:path"
 import * as Cloudflare from "alchemy/Cloudflare"
+import type { Rpc } from "alchemy/Rpc"
 import * as Effect from "effect/Effect"
 import * as Redacted from "effect/Redacted"
+import type { MapleApiRpcShape } from "@maple/domain/internal-rpc"
 import type { MapleDomains, MapleStage } from "@maple/infra/cloudflare"
 import {
 	CLOUDFLARE_WORKER_PLACEMENT,
@@ -32,11 +34,12 @@ const optionalSecret = (key: string): Record<string, Redacted.Redacted<string>>
 export interface CreateMapleApiOptions {
 	stage: MapleStage
 	domains: MapleDomains
-	/** The chat-flue worker, service-bound as CHAT_FLUE (hosts the Flue `triage` workflow). */
-	chatFlue: Cloudflare.Worker
 }
 
-export const createMapleApi = ({ stage, domains, chatFlue }: CreateMapleApiOptions) =>
+/** Alchemy resource type carried across the chat-flue service binding. */
+export type MapleApiWorker = Cloudflare.Worker & Rpc<MapleApiRpcShape>
+
+export const createMapleApi = ({ stage, domains }: CreateMapleApiOptions) =>
 	Effect.gen(function* () {
 		// MAPLE_DB Hyperdrive comes in two flavors:
 		//
@@ -115,7 +118,7 @@ export const createMapleApi = ({ stage, domains, chatFlue }: CreateMapleApiOptio
 			name: resolveWorkerName("vcs-sync", stage),
 		})
 
-		const worker = yield* Cloudflare.Worker("api", {
+		const worker = (yield* Cloudflare.Worker("api", {
 			name: resolveWorkerName("api", stage),
 			main: path.join(import.meta.dirname, "src", "worker.ts"),
 			compatibility: { date: "2026-04-08", flags: ["nodejs_compat"] },
@@ -134,7 +137,6 @@ export const createMapleApi = ({ stage, domains, chatFlue }: CreateMapleApiOptio
 				VCS_SYNC_QUEUE: vcsSyncQueue,
 				CLICKHOUSE_SCHEMA_APPLY_WORKFLOW: schemaApplyWorkflow,
 				AI_TRIAGE_WORKFLOW: aiTriageWorkflow,
-				CHAT_FLUE: chatFlue,
 				EMAIL: Cloudflare.Email.SendEmail("email", {
 					allowedSenderAddresses: ["notifications@noreply.maple.dev"],
 				}),
@@ -198,7 +200,7 @@ export const createMapleApi = ({ stage, domains, chatFlue }: CreateMapleApiOptio
 				...optionalPlain("PLANETSCALE_OAUTH_TOKEN_INFO_URL"),
 				...optionalPlain("MAPLE_PLANETSCALE_API_BASE_URL"),
 			},
-		})
+		})) as MapleApiWorker
 
 		if (hyperdriveRefId) {
 			// v1 `HyperdriveRef` equivalent: bind the dashboard-managed config by ID

apps/api/package.json

@@ -20,6 +20,7 @@
 		"tinybird:build": "tinybird build",
 		"tinybird:deploy": "tinybird deploy",
 		"bench:fetch": "bun run scripts/bench-queries.ts fetch",
+		"bench:suite": "bun run scripts/bench-queries.ts suite",
 		"bench:run": "bun run scripts/bench-queries.ts run",
 		"bench:inspect": "bun run scripts/bench-queries.ts inspect",
 		"bench:compare": "bun run scripts/bench-queries.ts compare",

apps/api/scripts/BENCH.md

@@ -15,18 +15,65 @@ bun bench:fetch    [--context name] [--profile name] [--since 24h]
                    [--top 20] [--out path] [--org id]
                    # mine recent db.query.text spans from prod traces → JSON
 
+bun bench:suite    [--org id] [--since 24h] [--trace-id id] [--service name]
+                   [--search phrase] [--attr-key k] [--attr-value v]
+                   [--attr-scope span] [--table-map from=to,...] [--out path]
+                   # compile the committed DSL benchmark suite → JSON (same
+                   # shape as `fetch`, so `run`/`inspect`/`compare` replay it)
+
 bun bench:run      <file> [--runs 5] [--warmup 1] [--out path]
                    # replay each query N times and report aggregated stats
 
 bun bench:inspect  <file>
                    # run EXPLAIN and EXPLAIN PIPELINE for each query
 
-bun bench:compare  <a.json> <b.json>
-                   # diff two run outputs (p95 wall, read bytes, memory)
+bun bench:compare  <a.json> <b.json> [--max-regression-pct N]
+                   # diff two run outputs (p95 wall, read bytes, memory);
+                   # exits non-zero if any case's p95 regresses beyond N%
 ```
 
 Output JSONs land in `apps/api/scripts/.bench/` by default (gitignored).
 
+## Two ways to get a query set
+
+- **`fetch`** mines whatever prod actually ran for a context label — great for
+  chasing a specific slow query, but not repeatable (the fingerprints drift as
+  traffic changes).
+- **`suite`** compiles a **committed, repeatable** set of ~13 representative
+  cases (needle-in-haystack body search, trace point lookup, attribute
+  equality/contains filters, list scans, facets, autocomplete, plus canaries)
+  straight from the `@maple/query-engine` DSL, so the benchmark runs the exact
+  SQL production emits. Definitions live in
+  [scripts/bench/suite.ts](bench/suite.ts). Pass `--trace-id` / `--service` /
+  `--attr-key` / `--attr-value` that exist in the target window for
+  representative timing — the two lookup cases fall back to zero-row
+  placeholders (with a warning) otherwise.
+
+## A/B: schema-change spike (e.g. sort-key experiment)
+
+`suite --table-map` rewrites `FROM`/`JOIN` table names in the compiled SQL so
+you can benchmark the same queries against shadow tables without recompiling.
+Rules are whole-identifier and longest-first, so `traces=traces_t2` never
+touches `trace_detail_spans` or `traces_aggregates_hourly`.
+
+```
+# control vs candidate tables, same window/org
+bun bench:suite --org org_x --trace-id <id> --service <svc> \
+  --out .bench/suiteA.json
+bun bench:suite --org org_x --trace-id <id> --service <svc> \
+  --table-map "traces=traces_t2,logs=logs_l2" --out .bench/suiteB.json
+
+bun bench:run .bench/suiteA.json --out .bench/resA.json   # CLICKHOUSE_URL = dedicated CH
+bun bench:run .bench/suiteB.json --out .bench/resB.json
+bun bench:compare .bench/resA.json .bench/resB.json --max-regression-pct 25
+```
+
+Run A/B against a **dedicated ClickHouse** (docker/staging), not a Tinybird
+branch — Tinybird branches share prod compute, so wall-time is noisy (lean on
+`read rows`/`read bytes`, which are deterministic). The `inspect` command's
+`EXPLAIN PIPELINE` tells you whether `optimize_read_in_order` actually engaged
+for the candidate sort key.
+
 ## Implementation
 
 Built on Effect v4 end-to-end: