> ## Documentation Index > Fetch the complete documentation index at: https://docs.kaireonai.com/llms.txt > Use this file to discover all available pages before exploring further. # Flow Hardening (Phase 6.6) > Operational guardrails shipped in Phase 6.6 — advisory-locked scheduler ticks, per-tenant scheduler dashboard, and k6 baseline perf script. Phase 6.6 closes the operational gaps identified across Phases 6.0–6.5. Each item is a real correctness or observability fix, not a stylistic sweep. ## 1. Advisory PG lock on scheduler tick **Problem:** `/api/v1/cron/flow-scheduler-tick` was lock-free. An external orchestrator that double-fires the endpoint (e.g. a Vercel Cron retry on a transient timeout) would dispatch every due pipeline twice. **Fix:** Wrap the tick in `pg_try_advisory_lock(0x666c6f77)`. When the lock is already held, the route returns 200 with `skipped: true` so the orchestrator doesn't treat it as an error. ```ts theme={null} const lockRows = await prisma.$queryRawUnsafe( `SELECT pg_try_advisory_lock($1::bigint) AS locked`, ADVISORY_KEY, ); if (!lockRows[0]?.locked) { return NextResponse.json({ skipped: true, ... }, { status: 200 }); } try { /* sweep + dispatch */ } finally { await prisma.$queryRawUnsafe(`SELECT pg_advisory_unlock($1::bigint)`, ADVISORY_KEY); } ``` The advisory lock is process-wide on the PG instance, so even multi-replica deployments are safe. ## 2. Per-tenant scheduler dashboard at `/data/scheduler` A read-only page (no write controls) showing every IR-native pipeline with `ir.schedule`: | Column | Source | | ------------- | --------------------------------------------------------------------------------------------------- | | Pipeline name | `pipelines.name` | | Schedule | `ir.schedule` (cron / interval / rrule) | | Last run | `pipelines.lastRunAt` + `lastRunStatus` pill | | **Lag** | Minutes between most-recent expected fire and actual `lastRunAt` — green if \<5min, amber otherwise | | Next 5 fires | Computed via the next-fire-times helper | API surface: `GET /api/v1/scheduler-status`. Tenant-scoped via `requireTenant`. ## 3. k6 baseline perf script at `k6/flow/baseline.js` Run with: ```bash theme={null} k6 run k6/flow/baseline.js \ --env BASE_URL=http://localhost:3000 \ --env API_KEY=$KEY \ --env TENANT_ID=$TENANT \ --env PIPELINE_ID=$PIPELINE ``` SLO thresholds enforced in the script: | Endpoint | p95 | | --------------------------------------- | ---------------------------------- | | `GET /api/v1/pipelines?limit=20` | \<400ms | | `GET /api/v1/pipelines/:id/ir` | \<400ms | | `GET /api/v1/pipelines/:id/sql-preview` | \<800ms (info\_schema lookup) | | `GET /api/v1/lineage` | \<800ms (hits actual target table) | Global threshold: `http_req_failed: rate<0.01` (under 1% errors). ## 4. Runtime correctness & recovery guardrails These pipeline-runtime guardrails underlie the scheduler hardening above. Each is opt-in on the IR (or on the target/validate node) and each is best-effort where noted, so enabling one never turns a healthy run into a failure. | Guardrail | Where it lives | Behavior | | -------------------------- | ---------------------------------------------------------------- | -------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | | **Pre-load backup** | `lib/flow/runtime/load-modes/backup.ts` | Before a `truncate` or `blue_green` load, snapshots the target to `_backup_` via `CREATE TABLE … AS TABLE`. After a clean load it prunes to `retainCount` newest (ordered by `pg_class.oid`, since the run-id suffix is not time-sortable). On a **failed** load the backup is left in place and its name is recorded in run metadata so an operator can recover with `INSERT INTO target SELECT * FROM `. | | **Row-count anomaly band** | `lib/flow/runtime/executors/target.ts` (`expectedRowCountDelta`) | Opt-in per target: `{ minPct, maxPct, windowRuns }`. Compares this run's `rowsLoaded` against the mean of the last N completed runs (default window 7, capped 60; needs ≥3 runs of history). Outside the band emits a **warning** system-health alert tagged `spike`/`drop`. Never fails the run. | | **Resumable DAG** | `lib/flow/runtime/batch-interpreter.ts` + `resumable-runner.ts` | When a `checkpointStore` is supplied, `runResumableDag` skips nodes whose prior checkpoint is `status: "completed"` and reloads their cached output, so a retried run resumes from the last successful node instead of re-running the whole DAG. | | **DLQ / quarantine** | `lib/flow/runtime/dlq.ts` | A validate node with quarantine enabled routes failed rows to a fixed-shape table (`id`, `pipelineId`, `runId`, `nodeId`, `ruleId`, `row` JSONB, `createdAt`), created idempotently via `CREATE TABLE IF NOT EXISTS` + a `(pipelineId, runId)` index. | | **Retry policy** | `lib/flow/ir/pipeline.ts` (`errorHandling`) | Per-pipeline: `retry { maxAttempts 0–20, backoff: exponential \| linear \| fixed }`, `dlq { enabled, destination? }`, and `onFailure: abort \| continue \| alert`. | | **Staging idempotency** | `lib/flow/runtime/staging-materializer.ts` | Source/transform/branch staging tables (`_flow_src_*` / `_flow_xform_*` / `_flow_branch_*`) are `DROP TABLE IF EXISTS` then re-`CREATE`d, so re-running the same `runId` is idempotent. Regular (not `TEMPORARY`) tables, because Prisma pool routing can send follow-up statements to a different session. | ### File-arrival deadline enforcement (shipped) `trigger.file_arrival.deadline` **is** enforced — `maybeFireDeadlineMiss` in `lib/flow/scheduler/run-tick.ts` compares `now` against `anchor + windowMinutes` (anchor = `lastRunAt`, or the pipeline's `createdAt` when it has never run) and, once the window has elapsed, takes the `onMiss` action: `alert` (warning system-health alert), `fail` (a synthetic failed `PipelineRun` so health dashboards count the missed SLA), or `skip` (log + continue). It is idempotent per `(pipeline, anchor)` so a stuck pipeline alerts once, not every tick. ## Honest limits deferred beyond Phase 6.6 | Item | Why deferred | | ---------------------------------------------------------------- | --------------------------------------------------------- | | Per-partition lag metrics for streaming consumers | Requires a real broker, gated by `FLOW_STREAMING_ENABLED` | | Run-over-run trend chart on the per-node metrics drawer | Frontend chart component, not yet picked | | Selective replay-from-DLQ runtime (`replayDlqOnly` flag) | Source executor needs row-key filter capability | | Visual DAG highlight on lineage row click | React Flow custom edge highlighting — polish plan | | Drag-add-edge / right-click delete on Visual canvas | Polish plan | | Editable per-node config form in right pane | Polish plan | | DLQ surfacing for non-validate failures (`.failed/` folder rows) | Source executor needs to emit failure-folder index |