Flow Hardening (Phase 6.6)

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.

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:

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 `<table>_backup_<runId>` 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 <backup>`.
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

Get Started

Tutorials

Use Cases

Decisioning

Studio

Data Pipelines

AI & ML

Operations & Reporting

Governance & Security

Integrations

Reference

Flow Hardening (Phase 6.6)

1. Advisory PG lock on scheduler tick

2. Per-tenant scheduler dashboard at `/data/scheduler`

3. k6 baseline perf script at `k6/flow/baseline.js`

4. Runtime correctness & recovery guardrails

File-arrival deadline enforcement (shipped)

Honest limits deferred beyond Phase 6.6

​1. Advisory PG lock on scheduler tick

​2. Per-tenant scheduler dashboard at /data/scheduler

​3. k6 baseline perf script at k6/flow/baseline.js

​4. Runtime correctness & recovery guardrails

​File-arrival deadline enforcement (shipped)

​Honest limits deferred beyond Phase 6.6

1. Advisory PG lock on scheduler tick

2. Per-tenant scheduler dashboard at `/data/scheduler`

3. k6 baseline perf script at `k6/flow/baseline.js`

4. Runtime correctness & recovery guardrails

File-arrival deadline enforcement (shipped)

Honest limits deferred beyond Phase 6.6