From f73541ba910ef946fb99d2c5028ed822982d09cb Mon Sep 17 00:00:00 2001 From: anandgupta42 Date: Tue, 6 Oct 2026 14:48:30 -0700 Subject: [PATCH 1/7] fix(dbt-tools): install packages only when needed, under a lock, and re-read the manifest `altimate-dbt` ran `dbt deps` on every start, and concurrent invocations corrupted `dbt_packages`. Packages are now installed only when missing, out of date or half-written, under a cross-process lock with a heartbeat and stale-lock handling. `deps` and `add-packages` take the same lock. With the install no longer staggering start-up, concurrent processes could read `target/manifest.json` while dbt rewrote it, so reading the manifest now retries with jittered backoff. `schema-verify` now fails only on what dbt treats as an error (an enforced contract the built relation does not match, a declared column with a test that the model does not produce) and never tells the agent to remove columns. Adds `yaml` as a direct dependency of `dbt-tools` (already in the lockfile through other workspaces) for reading `packages.yml` and `dbt_project.yml`. Co-Authored-By: Claude Opus 5.5 (1M context) --- bun.lock | 1 + packages/dbt-tools/package.json | 5 +- packages/dbt-tools/src/adapter.ts | 10 +- packages/dbt-tools/src/commands/deps.ts | 11 +- .../dbt-tools/src/commands/schema-verify.ts | 150 ++++-- packages/dbt-tools/src/index.ts | 15 +- packages/dbt-tools/src/manifest-retry.ts | 36 ++ packages/dbt-tools/src/packages.ts | 428 ++++++++++++++++++ packages/dbt-tools/test/build.test.ts | 28 +- .../dbt-tools/test/fixture/packages-worker.ts | 33 ++ .../dbt-tools/test/manifest-retry.test.ts | 35 ++ .../dbt-tools/test/packages-real-dbt.test.ts | 142 ++++++ packages/dbt-tools/test/packages.test.ts | 345 ++++++++++++++ packages/dbt-tools/test/schema-verify.test.ts | 195 ++++++-- 14 files changed, 1358 insertions(+), 76 deletions(-) create mode 100644 packages/dbt-tools/src/manifest-retry.ts create mode 100644 packages/dbt-tools/src/packages.ts create mode 100644 packages/dbt-tools/test/fixture/packages-worker.ts create mode 100644 packages/dbt-tools/test/manifest-retry.test.ts create mode 100644 packages/dbt-tools/test/packages-real-dbt.test.ts create mode 100644 packages/dbt-tools/test/packages.test.ts diff --git a/bun.lock b/bun.lock index 0495f7f8cd..7d50846917 100644 --- a/bun.lock +++ b/bun.lock @@ -151,6 +151,7 @@ }, "dependencies": { "@altimateai/dbt-integration": "^0.2.2", + "yaml": "2.8.3", }, "devDependencies": { "@tsconfig/bun": "catalog:", diff --git a/packages/dbt-tools/package.json b/packages/dbt-tools/package.json index 88801d319b..8b3a1e7890 100644 --- a/packages/dbt-tools/package.json +++ b/packages/dbt-tools/package.json @@ -9,11 +9,12 @@ "scripts": { "build": "bun build src/index.ts --outdir dist --target node --format esm && bun run script/copy-python.ts", "typecheck": "tsc --noEmit", - "test": "bun test test/cli.test.ts test/config.test.ts test/dbt-cli.test.ts test/dbt-resolve.test.ts test/build-integrity.test.ts --timeout 30000", + "test": "bun test test/cli.test.ts test/config.test.ts test/dbt-cli.test.ts test/dbt-resolve.test.ts test/build-integrity.test.ts test/packages.test.ts test/manifest-retry.test.ts --timeout 30000", "test:e2e": "bun test test/e2e/ --timeout 300000" }, "dependencies": { - "@altimateai/dbt-integration": "^0.2.2" + "@altimateai/dbt-integration": "^0.2.2", + "yaml": "2.8.3" }, "devDependencies": { "@tsconfig/bun": "catalog:", diff --git a/packages/dbt-tools/src/adapter.ts b/packages/dbt-tools/src/adapter.ts index ac906b8804..72d3e360af 100644 --- a/packages/dbt-tools/src/adapter.ts +++ b/packages/dbt-tools/src/adapter.ts @@ -1,5 +1,6 @@ import type { Config } from "./config" import { bufferLog } from "./log-buffer" +import { retryParseManifest } from "./manifest-retry" export { getRecentDbtLogs, clearDbtLogs } from "./log-buffer" import { DBTProjectIntegrationAdapter, @@ -48,7 +49,10 @@ function configuration(cfg: Config): DBTConfiguration { getQueryLimit: () => cfg.queryLimit ?? DEFAULT_CONFIGURATION_VALUES.queryLimit, getEnableNotebooks: () => DEFAULT_CONFIGURATION_VALUES.enableNotebooks, getDisableQueryHistory: () => DEFAULT_CONFIGURATION_VALUES.disableQueryHistory, - getInstallDepsOnProjectInitialization: () => DEFAULT_CONFIGURATION_VALUES.installDepsOnProjectInitialization, + // altimate-dbt installs packages itself (packages.ts: only when missing or out of date, + // under a cross-process lock). The library's own install runs `dbt deps` unconditionally + // from every process, and concurrent runs corrupt dbt_packages. + getInstallDepsOnProjectInitialization: () => false, getDisableDepthsCalculation: () => true, getWorkingDirectory: () => cfg.projectRoot, getAltimateUrl: () => "", @@ -208,5 +212,9 @@ export async function create(cfg: Config): Promise ) await adapter.initialize() + // Another altimate-dbt process may have been rewriting target/manifest.json while the + // library read it; re-read until it is whole, and keep the result for later commands. + retryParseManifest(adapter) + await adapter.parseManifest() return adapter } diff --git a/packages/dbt-tools/src/commands/deps.ts b/packages/dbt-tools/src/commands/deps.ts index feda088730..7e51dd15ce 100644 --- a/packages/dbt-tools/src/commands/deps.ts +++ b/packages/dbt-tools/src/commands/deps.ts @@ -1,14 +1,17 @@ import type { DBTProjectIntegrationAdapter, CommandProcessResult } from "@altimateai/dbt-integration" +import { installPackagesLocked } from "../packages" -export async function deps(adapter: DBTProjectIntegrationAdapter) { - const result = await adapter.installDeps() +// Explicit requests always run `dbt deps`, but under the same lock the automatic +// install uses so they cannot interleave with another process's install. +export async function deps(adapter: DBTProjectIntegrationAdapter, projectRoot: string) { + const result = await installPackagesLocked(projectRoot, () => adapter.installDeps()) return format(result) } -export async function add(adapter: DBTProjectIntegrationAdapter, args: string[]) { +export async function add(adapter: DBTProjectIntegrationAdapter, args: string[], projectRoot: string) { const raw = flag(args, "packages") if (!raw) return { error: "Missing --packages" } - const result = await adapter.installDbtPackages(raw.split(",")) + const result = await installPackagesLocked(projectRoot, () => adapter.installDbtPackages(raw.split(","))) return format(result) } diff --git a/packages/dbt-tools/src/commands/schema-verify.ts b/packages/dbt-tools/src/commands/schema-verify.ts index 4ed117e04b..b946cac487 100644 --- a/packages/dbt-tools/src/commands/schema-verify.ts +++ b/packages/dbt-tools/src/commands/schema-verify.ts @@ -1,27 +1,34 @@ -import type { ColumnMetaData, DBTProjectIntegrationAdapter } from "@altimateai/dbt-integration" +import type { ColumnMetaData, DBTProjectIntegrationAdapter, TestMetaData } from "@altimateai/dbt-integration" /** - * Verify that a model's actual produced columns match the spec declared in - * `schema.yml` (compiled into manifest.json as `node.columns`). + * Compare a model's materialized columns with the columns its YAML declares + * (`node.columns` in the manifest) and report only what dbt's own semantics + * establish. * - * Spec source: `adapter.nodeMetaMap.lookupByBaseName(model).columns` — these - * are the columns the schema.yml entry promised. Object insertion order is - * preserved through manifest parsing, so it carries the spec's column order. + * What YAML `columns:` means in dbt: + * - Without an enforced contract it is documentation plus the anchor for + * column tests. It is NOT an exhaustive list: models routinely produce + * columns the YAML never mentions, so "produced but not listed" says + * nothing about whether a column is wanted. + * - With `config.contract.enforced: true`, dbt itself requires the model to + * return exactly the declared columns and rejects the build otherwise. + * - A declared column the model does not produce is a real problem when a + * test is attached to it (the test reads a column that is not there), and + * when a contract is enforced. With neither, it is a stale or aspirational + * entry: worth knowing, not a failure. + * - Column order in YAML never constrains the model (dbt aligns it itself + * under a contract), and `data_type` is only checked by dbt under a contract + * (adapters spell types differently, so string comparison is not evidence). * - * Actual source: `adapter.getColumnsOfModel(model)` — the columns the - * warehouse / catalog reports the materialized table actually has. - * - * Returns four lists the agent must treat as the contract: - * - columns_extra: in actual, not in spec → REMOVE from SELECT - * - columns_missing: in spec, not in actual → ADD to SELECT - * - columns_reordered: in both, wrong position → REORDER the SELECT - * - type_mismatches: same name, different declared types - * - * `verdict` is "match" iff all four lists are empty. - * - * Skip cases: - * - "no-spec": schema.yml doesn't declare columns for this model — nothing - * to verify; agent has no contract to fail against. + * Output: + * - verdict "mismatch" iff `findings` is non-empty. Each finding names what is + * wrong, the YAML file that declares it, and why it is established. + * - `notes` carry true observations that are not problems (unlisted columns, + * declared-but-absent columns nothing depends on, type differences). + * - `columns_extra` / `columns_missing` / `columns_reordered` / + * `type_mismatches` are the raw diff against the YAML, for transparency. They + * are not instructions: `columns_extra` means "not listed in YAML". + * - verdict "no-spec": the model has no columns declared in YAML. */ export async function schemaVerify(adapter: DBTProjectIntegrationAdapter, args: string[]) { const model = flag(args, "model") @@ -58,7 +65,7 @@ export async function schemaVerify(adapter: DBTProjectIntegrationAdapter, args: return { model, verdict: "no-spec" as const, - message: `Model '${model}' has no columns declared in schema.yml. There is no spec to verify against; the agent's column choices are unconstrained.`, + message: `Model '${model}' has no columns declared in YAML. There is no declaration to compare against.`, actual_columns: actual.map((c) => c.column), } } @@ -127,24 +134,111 @@ export async function schemaVerify(adapter: DBTProjectIntegrationAdapter, args: } } - const verdict = - columns_extra.length === 0 && - columns_missing.length === 0 && - columns_reordered.length === 0 && - type_mismatches.length === 0 - ? ("match" as const) - : ("mismatch" as const) + // 5. Decide what is established (see header). Raw diff lists above are kept as-is. + const contractEnforced = isContractEnforced(node as unknown as Record) + const { file: declaredIn, pkg: declaredPackage } = splitPatchPath(node.patch_path) + const where = describeSource(declaredIn, declaredPackage, node.package_name) + const testsByColumn = columnTests(parsed?.testMetaMap, node.unique_id) + + const findings: Finding[] = [] + const notes: string[] = [] + + const listed = (names: string[]) => names.map((n) => `\`${n}\``).join(", ") + + if (contractEnforced) { + if (columns_missing.length > 0) + findings.push({ + kind: "contract-missing-columns", + columns: columns_missing, + evidence: `${where} declares an enforced contract that lists ${listed(columns_missing)}, but the built table does not have ${columns_missing.length === 1 ? "it" : "them"}. dbt rejects a contract-enforced model whose columns differ from the contract.`, + }) + if (columns_extra.length > 0) + findings.push({ + kind: "contract-extra-columns", + columns: columns_extra, + evidence: `${where} declares an enforced contract that does not list ${listed(columns_extra)}, but the built table has ${columns_extra.length === 1 ? "it" : "them"}. dbt rejects a contract-enforced model whose columns differ from the contract.`, + }) + } else { + const missingWithTests = columns_missing.filter((c) => (testsByColumn.get(c.toLowerCase()) ?? []).length > 0) + const missingUntested = columns_missing.filter((c) => !missingWithTests.includes(c)) + if (missingWithTests.length > 0) { + const detail = missingWithTests + .map((c) => `\`${c}\` (tests: ${(testsByColumn.get(c.toLowerCase()) ?? []).join(", ")})`) + .join("; ") + findings.push({ + kind: "tested-column-missing", + columns: missingWithTests, + evidence: `${where} declares column(s) with tests attached that the built table does not have: ${detail}. Those tests read a column that does not exist.`, + }) + } + if (missingUntested.length > 0) + notes.push( + `${where} declares ${listed(missingUntested)}, which the built table does not have. No test or contract depends on ${missingUntested.length === 1 ? "it" : "them"}, so this may be a stale or aspirational YAML entry rather than a defect in the model.`, + ) + if (columns_extra.length > 0) + notes.push( + `The built table has ${listed(columns_extra)}, not listed in ${where}. dbt does not require YAML to list every column, so this is not an error.`, + ) + } + if (type_mismatches.length > 0 && contractEnforced) + notes.push( + `Declared data_type differs from the warehouse type for ${listed(type_mismatches.map((t) => t.column))}; dbt compares types itself when it builds a contract-enforced model and rejects a real violation, so confirm with a build rather than from this string comparison (adapters spell types differently).`, + ) + + const verdict = findings.length === 0 ? ("match" as const) : ("mismatch" as const) return { model, verdict, + spec: { declared_in: declaredIn, package: declaredPackage ?? node.package_name, contract_enforced: contractEnforced }, expected_columns: expectedNames, actual_columns: actualNames, columns_extra, columns_missing, columns_reordered, type_mismatches, + findings, + notes, + } +} + +export interface Finding { + kind: "contract-missing-columns" | "contract-extra-columns" | "tested-column-missing" + columns: string[] + /** Which YAML declares what, and why this is a problem. */ + evidence: string +} + +function isContractEnforced(node: Record): boolean { + const fromConfig = (node.config as { contract?: { enforced?: unknown } } | undefined)?.contract?.enforced + const fromNode = (node.contract as { enforced?: unknown } | undefined)?.enforced + return fromConfig === true || fromNode === true +} + +/** dbt's `patch_path` looks like `://`. */ +function splitPatchPath(patchPath: string | undefined): { file?: string; pkg?: string } { + if (!patchPath) return {} + const m = /^([^:/]+):\/\/(.*)$/.exec(patchPath) + return m ? { pkg: m[1], file: m[2] } : { file: patchPath } +} + +function describeSource(file: string | undefined, declaringPackage: string | undefined, modelPackage: string | undefined): string { + const f = file ? `\`${file}\`` : "the YAML entry" + if (declaringPackage && modelPackage && declaringPackage !== modelPackage) + return `${f} in package \`${declaringPackage}\` (not this project)` + return f +} + +/** Lower-cased column name -> names of tests attached to that column of the model. */ +function columnTests(testMetaMap: Map | undefined, modelUniqueId: string): Map { + const out = new Map() + if (!testMetaMap || typeof testMetaMap.entries !== "function") return out + for (const [name, t] of testMetaMap.entries()) { + if (t.attached_node !== modelUniqueId || !t.column_name) continue + const key = t.column_name.toLowerCase() + out.set(key, [...(out.get(key) ?? []), name]) } + return out } function flag(args: string[], name: string): string | undefined { diff --git a/packages/dbt-tools/src/index.ts b/packages/dbt-tools/src/index.ts index 6a84df73f7..0e742060cf 100644 --- a/packages/dbt-tools/src/index.ts +++ b/packages/dbt-tools/src/index.ts @@ -142,6 +142,17 @@ async function main() { const { configure } = await import("./dbt-cli") configure({ pythonPath: cfg.pythonPath, projectRoot: cfg.projectRoot }) + // Install declared dbt packages only when missing or out of date, serialised across + // processes. Failures are not fatal here: the manifest parse below reports a missing + // package with dbt's own error, and the next invocation retries the install. + try { + const { ensurePackages, dbtDepsInstaller } = await import("./packages") + await ensurePackages(cfg.projectRoot, dbtDepsInstaller(cfg)) + } catch (err) { + const { bufferLog } = await import("./log-buffer") + bufferLog(`[dbt-tools] package install check failed: ${err instanceof Error ? err.message : String(err)}`) + } + // Lazy import to avoid loading python-bridge until needed let adapter try { @@ -194,10 +205,10 @@ async function main() { result = await (await import("./commands/graph")).parents(adapter, rest) break case "deps": - result = await (await import("./commands/deps")).deps(adapter) + result = await (await import("./commands/deps")).deps(adapter, cfg.projectRoot) break case "add-packages": - result = await (await import("./commands/deps")).add(adapter, rest) + result = await (await import("./commands/deps")).add(adapter, rest, cfg.projectRoot) break default: result = { error: `Unknown command: ${cmd}`, usage: USAGE } diff --git a/packages/dbt-tools/src/manifest-retry.ts b/packages/dbt-tools/src/manifest-retry.ts new file mode 100644 index 0000000000..2ddcc3fca2 --- /dev/null +++ b/packages/dbt-tools/src/manifest-retry.ts @@ -0,0 +1,36 @@ +/** + * Re-read the manifest when another process is rewriting it. + * + * Every `altimate-dbt` process runs `dbt parse` when it starts, and dbt rewrites + * `target/manifest.json` in place. A process that reads the file while a sibling + * is rewriting it gets nothing back: the library's `parseManifest` returns + * `undefined` and later calls fail with "not found in manifest" or "No manifest + * has been generated". The window is a few milliseconds, but processes started + * together move in lockstep (previously the unconditional `dbt deps` of each + * process spread them out), so it was hit in about 1 call in 9 when four ran at once. + * The file is complete again moments later, so reading again is the repair. + */ + +export interface ManifestSource { + parseManifest(): Promise +} + +const sleep = (ms: number) => new Promise((r) => setTimeout(r, ms)) + +/** Replace `source.parseManifest` with a version that retries while the result is missing. */ +export function retryParseManifest( + source: ManifestSource, + opts: { attempts?: number; baseDelayMs?: number } = {}, +): void { + const attempts = opts.attempts ?? 6 + const base = opts.baseDelayMs ?? 150 + const original = source.parseManifest.bind(source) + source.parseManifest = async () => { + let result = await original() + for (let i = 1; result === undefined && i < attempts; i++) { + await sleep(base * i + Math.floor(Math.random() * base)) + result = await original() + } + return result + } +} diff --git a/packages/dbt-tools/src/packages.ts b/packages/dbt-tools/src/packages.ts new file mode 100644 index 0000000000..b10b99cfbb --- /dev/null +++ b/packages/dbt-tools/src/packages.ts @@ -0,0 +1,428 @@ +/** + * dbt package installation that is safe to run from many `altimate-dbt` + * processes at once. + * + * Why this exists: `@altimateai/dbt-integration` runs `dbt deps` unconditionally + * the first time an adapter parses the manifest (its + * `installDepsOnProjectInitialization` setting, which `adapter.ts` now turns + * off). `dbt deps` deletes and re-extracts every package directory, so two + * `altimate-dbt` processes started together (the finish-time validators start + * several) could leave `dbt_packages//dbt_project.yml` missing or half + * written, which makes every later dbt command fail with "No dbt_project.yml + * found". Installation is now (1) skipped when the installed packages already + * satisfy what the project declares, and (2) serialised by a cross-process + * lock with a stale-owner story when it is needed. + * + * State lives under `/target/.altimate-dbt/`: + * packages.lock/ mkdir-based lock; owner.json inside; mtime = heartbeat + * packages.lock.break/ short-lived guard so only one process breaks a stale lock + * packages.stamp.json fingerprint of the declaration files + package dirs of the last good install + * packages.dirty present while an install is running or after one did not finish + */ + +import { execFile } from "child_process" +import { createHash, randomUUID } from "crypto" +import { + existsSync, + mkdirSync, + readFileSync, + readdirSync, + rmSync, + statSync, + utimesSync, + writeFileSync, +} from "fs" +import { hostname } from "os" +import { join, resolve } from "path" +import { parse as parseYaml } from "yaml" +import { bufferLog } from "./log-buffer" +import { buildDbtEnv, resolveDbt } from "./dbt-resolve" + +const STATE_DIR = join("target", ".altimate-dbt") +const DECLARATION_FILES = ["packages.yml", "dependencies.yml", "package-lock.yml"] as const +const DEFAULT_INSTALL_PATH = "dbt_packages" + +/** A lock whose heartbeat is older than this (or whose owner process is gone) may be broken. */ +export const DEFAULT_STALE_MS = 30_000 +const DEFAULT_HEARTBEAT_MS = 2_000 +/** Upper bound on waiting for another process's install before giving up on the lock. */ +const DEFAULT_MAX_WAIT_MS = 10 * 60_000 +const DEPS_TIMEOUT_MS = 5 * 60_000 +const BREAK_GUARD_STALE_MS = 30_000 + +export interface LockOptions { + staleMs?: number + heartbeatMs?: number + maxWaitMs?: number +} + +export type EnsureAction = "none" | "skipped" | "skipped-after-wait" | "installed" + +export interface EnsureResult { + action: EnsureAction + /** Why an install was (or was not) needed. */ + reason: string +} + +const sleep = (ms: number) => new Promise((r) => setTimeout(r, ms)) +const jitter = (base: number) => base + Math.floor(Math.random() * base) + +// --------------------------------------------------------------------------- +// Declared vs installed +// --------------------------------------------------------------------------- + +interface Declared { + /** Hub packages carry a predictable install directory name; others do not. */ + hubDir?: string + version?: string +} + +function readYaml(path: string): unknown { + try { + return parseYaml(readFileSync(path, "utf-8")) + } catch { + return undefined + } +} + +function projectInstallPath(root: string): string { + const cfg = readYaml(join(root, "dbt_project.yml")) as Record | undefined + const raw = cfg?.["packages-install-path"] + if (typeof raw === "string" && raw.length > 0 && !raw.includes("{{")) return resolve(root, raw) + return resolve(root, DEFAULT_INSTALL_PATH) +} + +/** Entries the project asks dbt to install. Prefers the lock file (exact versions). */ +function declaredPackages(root: string): Declared[] { + const present = DECLARATION_FILES.filter((f) => existsSync(join(root, f))) + if (present.length === 0) return [] + const source = present.includes("package-lock.yml") ? "package-lock.yml" : present.find((f) => f !== "package-lock.yml") + const doc = readYaml(join(root, source!)) as { packages?: unknown } | undefined + const entries = Array.isArray(doc?.packages) ? (doc!.packages as unknown[]) : [] + const out: Declared[] = [] + for (const e of entries) { + if (!e || typeof e !== "object") continue + const rec = e as Record + if (typeof rec.package === "string") { + const name = rec.package.split("/").pop() ?? rec.package + // Only the lock file pins an exact version; a packages.yml entry may be a range. + const version = source === "package-lock.yml" && rec.version !== undefined ? String(rec.version) : undefined + out.push({ hubDir: name, version }) + } else { + out.push({}) + } + } + return out +} + +/** Directories under the install path that look like a complete package. */ +function completePackageDirs(installPath: string): Map { + const out = new Map() + if (!existsSync(installPath)) return out + for (const name of readdirSync(installPath)) { + if (name.startsWith(".")) continue + const dir = join(installPath, name) + try { + if (!statSync(dir).isDirectory()) continue + } catch { + continue // dangling symlink + } + const projectFile = join(dir, "dbt_project.yml") + if (!existsSync(projectFile)) continue + const v = (readYaml(projectFile) as { version?: unknown } | undefined)?.version + out.set(name, v === undefined ? undefined : String(v)) + } + return out +} + +/** Every directory directly under the install path, complete or not. */ +function allPackageDirs(installPath: string): string[] { + if (!existsSync(installPath)) return [] + return readdirSync(installPath).filter((n) => !n.startsWith(".")) +} + +function fingerprint(root: string): string { + const h = createHash("sha256") + for (const f of DECLARATION_FILES) { + const p = join(root, f) + if (existsSync(p)) h.update(f).update("\0").update(readFileSync(p)).update("\0") + } + return h.digest("hex") +} + +function stateDir(root: string): string { + return join(root, STATE_DIR) +} + +interface Stamp { + fingerprint: string + packages: string[] +} + +function readStamp(root: string): Stamp | undefined { + try { + return JSON.parse(readFileSync(join(stateDir(root), "packages.stamp.json"), "utf-8")) as Stamp + } catch { + return undefined + } +} + +export interface SatisfiedResult { + ok: boolean + reason: string +} + +/** + * Do the installed packages satisfy what the project declares? Read-only and + * cheap; this is what lets the common case skip `dbt deps` entirely. + */ +export function packagesSatisfied(root: string): SatisfiedResult { + const declared = declaredPackages(root) + if (declared.length === 0) return { ok: true, reason: "no packages declared" } + if (existsSync(join(stateDir(root), "packages.dirty"))) + return { ok: false, reason: "a previous or concurrent install has not finished" } + + const installPath = projectInstallPath(root) + const installed = completePackageDirs(installPath) + const incomplete = allPackageDirs(installPath).filter((n) => !installed.has(n)) + if (incomplete.length > 0) return { ok: false, reason: `incomplete package directory: ${incomplete.join(", ")}` } + + const stamp = readStamp(root) + if (stamp) { + if (stamp.fingerprint !== fingerprint(root)) + return { ok: false, reason: "package declarations changed since the last install" } + const gone = stamp.packages.filter((p) => !installed.has(p)) + if (gone.length > 0) return { ok: false, reason: `installed package missing: ${gone.join(", ")}` } + return { ok: true, reason: "installed packages match the last install of these declarations" } + } + + // A lock older than packages.yml was not resolved from the current declarations. + try { + const lockM = statSync(join(root, "package-lock.yml")).mtimeMs + const declM = Math.max( + ...["packages.yml", "dependencies.yml"].filter((f) => existsSync(join(root, f))).map((f) => statSync(join(root, f)).mtimeMs), + ) + if (declM > lockM + 1000) return { ok: false, reason: "packages.yml is newer than package-lock.yml" } + } catch {} + + // No stamp: packages were installed by something else (image build, `dbt deps` by hand). + // Hub packages can be verified by name and, when a lock pins it, by version. + let unverifiable = 0 + for (const d of declared) { + if (!d.hubDir) { + unverifiable++ + continue + } + if (!installed.has(d.hubDir)) return { ok: false, reason: `package ${d.hubDir} is not installed` } + const have = installed.get(d.hubDir) + if (d.version !== undefined && have !== d.version) + return { ok: false, reason: `package ${d.hubDir} is ${have ?? "unversioned"}, lock requires ${d.version}` } + } + const nonHubInstalled = [...installed.keys()].filter((n) => !declared.some((d) => d.hubDir === n)).length + if (nonHubInstalled < unverifiable) return { ok: false, reason: "fewer non-hub packages installed than declared" } + return { ok: true, reason: "installed packages satisfy the declarations" } +} + +// --------------------------------------------------------------------------- +// Cross-process lock +// --------------------------------------------------------------------------- + +interface Owner { + pid: number + host: string + token: string + startedAt: number +} + +function pidAlive(pid: number): boolean { + try { + process.kill(pid, 0) + return true + } catch (e) { + return (e as NodeJS.ErrnoException).code === "EPERM" + } +} + +function readOwner(lockDir: string): Owner | undefined { + try { + return JSON.parse(readFileSync(join(lockDir, "owner.json"), "utf-8")) as Owner + } catch { + return undefined + } +} + +function lockIsStale(lockDir: string, staleMs: number): boolean { + let mtime: number + try { + mtime = statSync(lockDir).mtimeMs + } catch { + return false // already gone + } + const owner = readOwner(lockDir) + if (owner && owner.host === hostname() && owner.pid !== process.pid && !pidAlive(owner.pid)) return true + return Date.now() - mtime > staleMs +} + +/** Remove a stale lock. A guard directory makes sure only one process does it at a time. */ +function breakStaleLock(lockDir: string, staleMs: number, observedToken: string | undefined): void { + const guard = `${lockDir}.break` + try { + mkdirSync(guard) + } catch { + try { + if (Date.now() - statSync(guard).mtimeMs > BREAK_GUARD_STALE_MS) rmSync(guard, { recursive: true, force: true }) + } catch {} + return + } + try { + // Re-check under the guard: the lock may have been released and re-taken since we looked. + // A new owner has a new token, so only the lock we judged stale is removed. + if (readOwner(lockDir)?.token === observedToken && lockIsStale(lockDir, staleMs)) { + bufferLog(`[dbt-tools] breaking stale package lock ${lockDir}`) + rmSync(lockDir, { recursive: true, force: true }) + } + } finally { + rmSync(guard, { recursive: true, force: true }) + } +} + +async function acquire(root: string, opts: Required): Promise<() => void> { + const dir = stateDir(root) + const lockDir = join(dir, "packages.lock") + mkdirSync(dir, { recursive: true }) + const deadline = Date.now() + opts.maxWaitMs + for (;;) { + try { + mkdirSync(lockDir) + break + } catch (e) { + if ((e as NodeJS.ErrnoException).code !== "EEXIST") throw e + if (lockIsStale(lockDir, opts.staleMs)) breakStaleLock(lockDir, opts.staleMs, readOwner(lockDir)?.token) + if (Date.now() > deadline) throw new Error(`timed out after ${opts.maxWaitMs}ms waiting for the package install lock at ${lockDir}`) + await sleep(jitter(50)) + } + } + const owner: Owner = { pid: process.pid, host: hostname(), token: randomUUID(), startedAt: Date.now() } + writeFileSync(join(lockDir, "owner.json"), JSON.stringify(owner)) + const beat = setInterval(() => { + try { + if (readOwner(lockDir)?.token !== owner.token) { + bufferLog("[dbt-tools] package install lock was taken over; continuing without it") + clearInterval(beat) + return + } + const now = new Date() + utimesSync(lockDir, now, now) + } catch {} + }, opts.heartbeatMs) + beat.unref() + return () => { + clearInterval(beat) + // Only remove the lock if it is still ours. + if (readOwner(lockDir)?.token === owner.token) rmSync(lockDir, { recursive: true, force: true }) + } +} + +function withDefaults(o?: LockOptions): Required { + return { + staleMs: o?.staleMs ?? DEFAULT_STALE_MS, + heartbeatMs: o?.heartbeatMs ?? DEFAULT_HEARTBEAT_MS, + maxWaitMs: o?.maxWaitMs ?? DEFAULT_MAX_WAIT_MS, + } +} + +/** Run `fn` while holding the project's package lock. */ +export async function withPackagesLock(root: string, fn: () => Promise, opts?: LockOptions): Promise { + const release = await acquire(root, withDefaults(opts)) + try { + return await fn() + } finally { + release() + } +} + +// --------------------------------------------------------------------------- +// Install +// --------------------------------------------------------------------------- + +/** + * Run an install while holding the lock, bracketed by the `dirty` marker so a + * crash or failure mid-install is noticed (and repaired) by the next caller. + * Caller must already hold the lock. + */ +async function installUnderLock(root: string, install: () => Promise, record: boolean): Promise { + const dir = stateDir(root) + const dirty = join(dir, "packages.dirty") + writeFileSync(dirty, JSON.stringify({ pid: process.pid, startedAt: Date.now() })) + const result = await install() + const installPath = projectInstallPath(root) + const complete = [...completePackageDirs(installPath).keys()] + const broken = allPackageDirs(installPath).filter((n) => !complete.includes(n)) + if (declaredPackages(root).length > 0 && (complete.length === 0 || broken.length > 0)) { + throw new Error(`dbt deps finished but the package directory is incomplete${broken.length ? `: ${broken.join(", ")}` : ""}`) + } + const stampFile = join(dir, "packages.stamp.json") + if (record) { + writeFileSync(stampFile, JSON.stringify({ fingerprint: fingerprint(root), packages: complete } satisfies Stamp)) + } else { + // The caller's installer reports failure as a result, not a throw, so success is not + // known. Drop the old stamp: the next check verifies the directories against the + // declarations instead of trusting this run. + rmSync(stampFile, { force: true }) + } + rmSync(dirty, { force: true }) + return result +} + +/** + * Make sure the declared packages are installed. Installs at most once per + * change of declarations, however many processes ask at the same time. + */ +export async function ensurePackages( + root: string, + install: () => Promise, + opts?: LockOptions, +): Promise { + if (declaredPackages(root).length === 0) return { action: "none", reason: "no packages declared" } + const fast = packagesSatisfied(root) + if (fast.ok) return { action: "skipped", reason: fast.reason } + + const waitedFor = existsSync(join(stateDir(root), "packages.lock")) + return withPackagesLock( + root, + async () => { + // Another process may have finished the install while we waited. + const again = packagesSatisfied(root) + if (again.ok) return { action: "skipped-after-wait" as const, reason: waitedFor ? again.reason : `${again.reason} (after waiting for the lock)` } + await installUnderLock(root, install, true) + return { action: "installed" as const, reason: again.reason } + }, + opts, + ) +} + +/** + * Run an explicit install request (`altimate-dbt deps`, `add-packages`) under the lock. The + * library reports a failed `dbt deps` as a result rather than throwing, so this does not + * claim success (no stamp); later checks verify the directories themselves. + */ +export async function installPackagesLocked(root: string, install: () => Promise, opts?: LockOptions): Promise { + return withPackagesLock(root, () => installUnderLock(root, install, false), opts) +} + +/** An installer that shells out to `dbt deps` with the project's resolved dbt binary. */ +export function dbtDepsInstaller(cfg: { pythonPath?: string; projectRoot: string }): () => Promise { + return () => + new Promise((res, rej) => { + const dbt = resolveDbt(cfg.pythonPath, cfg.projectRoot) + execFile( + dbt.path, + ["deps"], + { cwd: cfg.projectRoot, env: buildDbtEnv(dbt), timeout: DEPS_TIMEOUT_MS, maxBuffer: 10 * 1024 * 1024 }, + (err, _stdout, stderr) => { + if (err) rej(new Error(`dbt deps failed: ${String(stderr || err.message).slice(-500)}`)) + else res() + }, + ) + }) +} diff --git a/packages/dbt-tools/test/build.test.ts b/packages/dbt-tools/test/build.test.ts index 1067870e3c..68db631533 100644 --- a/packages/dbt-tools/test/build.test.ts +++ b/packages/dbt-tools/test/build.test.ts @@ -71,7 +71,8 @@ describe("build command", () => { }) test("project-wide build collects per-model schema-verify mismatches", async () => { - // Mock manifest with 3 models: one matching spec, one mismatch (extra col), one no-spec. + // Mock manifest with 3 models: one matching spec, one mismatch (enforced contract violated), one no-spec. + // A column the YAML does not list is NOT a mismatch on its own (see schema-verify.test.ts). const matchingNode = { resource_type: "model", name: "users_dim", @@ -80,6 +81,7 @@ describe("build command", () => { const mismatchNode = { resource_type: "model", name: "products_dim", + config: { contract: { enforced: true } }, columns: { id: { name: "id", description: "", data_type: "INT" } }, } const nospecNode = { resource_type: "model", name: "legacy_facts", columns: {} } @@ -114,6 +116,30 @@ describe("build command", () => { expect(summary.errored).toBe(0) expect(summary.mismatches[0]?.model).toBe("products_dim") expect(summary.mismatches[0]?.columns_extra).toContain("extra_col") + expect((summary.mismatches[0] as unknown as { findings: Array<{ kind: string }> }).findings[0]?.kind).toBe("contract-extra-columns") + }) + + test("project-wide build does not report a model whose only difference is columns the YAML does not list", async () => { + const node = { + resource_type: "model", + name: "orders", + columns: { id: { name: "id", description: "", data_type: "" } }, + } + const adapter = makeAdapter({ + parseManifest: mock(() => Promise.resolve({ + nodeMetaMap: { + lookupByBaseName: mock(() => node), + lookupByUniqueId: mock(() => undefined), + nodes: mock(() => [node][Symbol.iterator]()), + }, + } as never)), + getColumnsOfModel: mock(() => Promise.resolve([{ column: "id", dtype: "INT" }, { column: "total", dtype: "INT" }])), + }) + const result = await build(adapter, []) + const summary = (result as unknown as { schema_verify_summary: { match: number; mismatch: number; mismatches: unknown[] } }).schema_verify_summary + expect(summary.match).toBe(1) + expect(summary.mismatch).toBe(0) + expect(summary.mismatches).toEqual([]) }) test("build surfaces stderr as error", async () => { diff --git a/packages/dbt-tools/test/fixture/packages-worker.ts b/packages/dbt-tools/test/fixture/packages-worker.ts new file mode 100644 index 0000000000..cf24f243d9 --- /dev/null +++ b/packages/dbt-tools/test/fixture/packages-worker.ts @@ -0,0 +1,33 @@ +// Child process used by test/packages.test.ts: runs ensurePackages against a +// project with a deliberately slow, non-atomic fake `dbt deps`, then reports +// what it observed once ensurePackages returned. +// +// usage: bun packages-worker.ts [crash] +import { appendFileSync, existsSync, mkdirSync, rmSync, writeFileSync } from "fs" +import { join } from "path" +import { ensurePackages } from "../../src/packages" + +const [root, log, pkg, mode] = process.argv.slice(2) as [string, string, string, string | undefined] +const sleep = (ms: number) => new Promise((r) => setTimeout(r, ms)) + +// What dbt deps does to a package: delete it, re-create the directory, then +// write its files. A reader that looks in between sees a missing or empty package. +async function slowInstall() { + appendFileSync(log, `install ${process.pid}\n`) + const dir = join(root, "dbt_packages", pkg) + rmSync(dir, { recursive: true, force: true }) + await sleep(150) + mkdirSync(dir, { recursive: true }) + if (mode === "crash") { + // Die with the package directory present but incomplete. + process.kill(process.pid, "SIGKILL") + } + await sleep(150) + writeFileSync(join(dir, "dbt_project.yml"), `name: ${pkg}\nversion: '1.0.0'\n`) + await sleep(50) +} + +await ensurePackages(root, slowInstall, { staleMs: 1500, heartbeatMs: 100 }) +// This is the moment a real command would go on to read the packages. +const ok = existsSync(join(root, "dbt_packages", pkg, "dbt_project.yml")) +appendFileSync(log, `${ok ? "ready" : "BROKEN"} ${process.pid}\n`) diff --git a/packages/dbt-tools/test/manifest-retry.test.ts b/packages/dbt-tools/test/manifest-retry.test.ts new file mode 100644 index 0000000000..77fa5c2c21 --- /dev/null +++ b/packages/dbt-tools/test/manifest-retry.test.ts @@ -0,0 +1,35 @@ +import { describe, expect, test } from "bun:test" +import { retryParseManifest } from "../src/manifest-retry" + +function source(results: Array) { + let calls = 0 + return { + get calls() { + return calls + }, + parseManifest: async () => results[Math.min(calls++, results.length - 1)] as { nodes: number } | undefined, + } +} + +describe("retryParseManifest", () => { + test("a manifest that is whole on the first read is read once", async () => { + const s = source([{ nodes: 1 }]) + retryParseManifest(s, { baseDelayMs: 1 }) + expect(await s.parseManifest()).toEqual({ nodes: 1 }) + expect(s.calls).toBe(1) + }) + + test("a manifest caught mid-rewrite is read again until it is whole", async () => { + const s = source([undefined, undefined, { nodes: 2 }]) + retryParseManifest(s, { baseDelayMs: 1 }) + expect(await s.parseManifest()).toEqual({ nodes: 2 }) + expect(s.calls).toBe(3) + }) + + test("gives up after the attempt limit and reports the absence", async () => { + const s = source([undefined]) + retryParseManifest(s, { attempts: 4, baseDelayMs: 1 }) + expect(await s.parseManifest()).toBeUndefined() + expect(s.calls).toBe(4) + }) +}) diff --git a/packages/dbt-tools/test/packages-real-dbt.test.ts b/packages/dbt-tools/test/packages-real-dbt.test.ts new file mode 100644 index 0000000000..b8fbde73cb --- /dev/null +++ b/packages/dbt-tools/test/packages-real-dbt.test.ts @@ -0,0 +1,142 @@ +// Real dbt + real built altimate-dbt (dist/index.js): concurrent invocations must +// not each run `dbt deps`. Skipped when dist/ is not built or dbt-duckdb is not +// installed. Uses a local package, so it needs no network. +import { afterAll, beforeAll, describe, expect, test } from "bun:test" +import { spawn, spawnSync } from "child_process" +import { chmodSync, copyFileSync, existsSync, mkdirSync, mkdtempSync, readFileSync, rmSync, writeFileSync } from "fs" +import { tmpdir } from "os" +import { dirname, join } from "path" + +const DIST = process.env.ALTIMATE_DBT_DIST ?? join(import.meta.dir, "..", "dist", "index.js") + +function findDbt(): string | undefined { + const r = spawnSync("dbt", ["--version"], { encoding: "utf-8", timeout: 20_000 }) + if (r.status !== 0 || !/duckdb/i.test(`${r.stdout}${r.stderr}`)) return undefined + const which = spawnSync(process.platform === "win32" ? "where" : "which", ["dbt"], { encoding: "utf-8" }) + return which.stdout.trim().split(/\r?\n/)[0] || undefined +} + +const REAL_DBT = existsSync(DIST) ? findDbt() : undefined +const suite = REAL_DBT ? describe : describe.skip + +let work = "" +let proj = "" +let shim = "" +let log = "" +let env: Record = {} + +const CONCURRENCY = 4 + +function dbtSync(args: string[]) { + return spawnSync(REAL_DBT!, args, { cwd: proj, encoding: "utf-8", env: { ...process.env, ...env }, timeout: 120_000 }) +} + +function runAltimateDbt(i: number) { + return new Promise<{ code: number | null; out: string }>((resolve) => { + const c = spawn( + "node", + [DIST, "schema-verify", "--model", "m"], + { cwd: proj, env: { ...process.env, ...env, E2E_DB: join(proj, `db${i}.duckdb`) } }, + ) + let out = "" + c.stdout.on("data", (d) => (out += d)) + c.stderr.on("data", (d) => (out += d)) + c.on("close", (code) => resolve({ code, out })) + }) +} + +const depsCalls = () => + existsSync(log) ? readFileSync(log, "utf-8").split("\n").filter((l) => l.split(" ")[0] === "deps").length : 0 + +const packageHealthy = () => existsSync(join(proj, "dbt_packages", "localpkg", "dbt_project.yml")) + +suite("real dbt: concurrent altimate-dbt invocations and `dbt deps`", () => { + beforeAll(() => { + work = mkdtempSync(join(tmpdir(), "dbt-pkgs-real-")) + proj = join(work, "proj") + const pkg = join(work, "localpkg") + mkdirSync(join(pkg, "macros"), { recursive: true }) + writeFileSync(join(pkg, "dbt_project.yml"), "name: localpkg\nversion: '0.1.0'\nconfig-version: 2\n") + writeFileSync(join(pkg, "macros", "one.sql"), "{% macro one() %}1{% endmacro %}\n") + mkdirSync(join(proj, "models"), { recursive: true }) + mkdirSync(join(proj, "seeds"), { recursive: true }) + writeFileSync(join(proj, "dbt_project.yml"), "name: proj\nversion: '1.0'\nconfig-version: 2\nprofile: proj\n") + writeFileSync( + join(proj, "profiles.yml"), + "proj:\n target: dev\n outputs:\n dev:\n type: duckdb\n path: \"{{ env_var('E2E_DB', 'e2e.duckdb') }}\"\n threads: 1\n", + ) + writeFileSync(join(proj, "packages.yml"), "packages:\n - local: ../localpkg\n") + writeFileSync(join(proj, "seeds", "raw.csv"), "id,name\n1,a\n2,b\n") + writeFileSync(join(proj, "models", "m.sql"), "select id, name, {{ localpkg.one() }} as one from {{ ref('raw') }}\n") + writeFileSync(join(proj, "models", "m.yml"), "version: 2\nmodels:\n - name: m\n columns:\n - name: id\n") + + // A `dbt` that records its subcommand, then runs the real one. Both the dbt-tools resolver + // (ALTIMATE_DBT_PATH) and the integration library (PATH) will find it. + log = join(work, "dbt-calls.log") + const bin = join(work, "bin") + mkdirSync(bin) + shim = join(bin, "dbt") + writeFileSync(shim, `#!/bin/sh\necho "$@" >> "${log}"\nexec "${REAL_DBT}" "$@"\n`) + chmodSync(shim, 0o755) + const home = join(work, "home") + mkdirSync(home) + env = { + HOME: home, + XDG_CONFIG_HOME: join(home, ".config"), + PATH: `${bin}${process.platform === "win32" ? ";" : ":"}${process.env.PATH ?? ""}`, + ALTIMATE_DBT_PATH: shim, + } + // The library looks for dbt next to the configured python, so point it at a directory + // whose `dbt` is the recording shim. + const venvBin = join(work, "venv", "bin") + mkdirSync(venvBin, { recursive: true }) + copyFileSync(shim, join(venvBin, "dbt")) + chmodSync(join(venvBin, "dbt"), 0o755) + const realPython = join(dirname(REAL_DBT!), "python3") + // A wrapper, not a symlink: python finds its site-packages relative to the real executable. + writeFileSync(join(venvBin, "python3"), `#!/bin/sh\nexec "${existsSync(realPython) ? realPython : "python3"}" "$@"\n`) + chmodSync(join(venvBin, "python3"), 0o755) + const init = spawnSync("node", [DIST, "init", "--project-root", proj, "--python-path", join(venvBin, "python3")], { + env: { ...process.env, ...env }, + encoding: "utf-8", + }) + expect(init.status).toBe(0) + + // Prepare a built project (the shim records these calls too; the log is cleared below). + expect(dbtSync(["deps"]).status).toBe(0) + expect(dbtSync(["build"]).status).toBe(0) + for (let i = 0; i < CONCURRENCY; i++) copyFileSync(join(proj, "e2e.duckdb"), join(proj, `db${i}.duckdb`)) + }, 180_000) + + afterAll(() => { + if (work) rmSync(work, { recursive: true, force: true }) + }) + + test("packages already installed: concurrent invocations run `dbt deps` zero times", async () => { + writeFileSync(log, "") + expect(packageHealthy()).toBe(true) + const runs = await Promise.all(Array.from({ length: CONCURRENCY }, (_, i) => runAltimateDbt(i))) + expect(runs.every((r) => !/No dbt_project\.yml found/.test(r.out))).toBe(true) + expect(depsCalls()).toBe(0) + expect(packageHealthy()).toBe(true) + }, 120_000) + + test("packages missing: concurrent invocations install exactly once and leave a healthy directory", async () => { + rmSync(join(proj, "dbt_packages"), { recursive: true, force: true }) + rmSync(join(proj, "target", ".altimate-dbt"), { recursive: true, force: true }) + writeFileSync(log, "") + const runs = await Promise.all(Array.from({ length: CONCURRENCY }, (_, i) => runAltimateDbt(i))) + expect(runs.every((r) => !/No dbt_project\.yml found/.test(r.out))).toBe(true) + expect(depsCalls()).toBe(1) + expect(packageHealthy()).toBe(true) + }, 120_000) + + test("a package directory left without dbt_project.yml is repaired by the next invocation", async () => { + rmSync(join(proj, "dbt_packages", "localpkg"), { recursive: true, force: true }) + mkdirSync(join(proj, "dbt_packages", "localpkg"), { recursive: true }) // present but empty + writeFileSync(log, "") + await runAltimateDbt(0) + expect(depsCalls()).toBe(1) + expect(packageHealthy()).toBe(true) + }, 120_000) +}) diff --git a/packages/dbt-tools/test/packages.test.ts b/packages/dbt-tools/test/packages.test.ts new file mode 100644 index 0000000000..199358ec48 --- /dev/null +++ b/packages/dbt-tools/test/packages.test.ts @@ -0,0 +1,345 @@ +import { afterEach, beforeEach, describe, expect, test } from "bun:test" +import { spawn } from "child_process" +import { existsSync, mkdirSync, mkdtempSync, readFileSync, rmSync, utimesSync, writeFileSync } from "fs" +import { tmpdir } from "os" +import { join } from "path" +import { + ensurePackages, + installPackagesLocked, + packagesSatisfied, + withPackagesLock, +} from "../src/packages" + +const WORKER = join(import.meta.dir, "fixture", "packages-worker.ts") +const PKG = "mypkg" + +let root = "" + +function writeProject(opts: { lock?: string; packages?: string } = {}) { + writeFileSync(join(root, "dbt_project.yml"), "name: proj\nversion: '1.0'\nconfig-version: 2\nprofile: proj\n") + writeFileSync(join(root, "packages.yml"), opts.packages ?? `packages:\n - package: acme/${PKG}\n version: 1.0.0\n`) + if (opts.lock) writeFileSync(join(root, "package-lock.yml"), opts.lock) +} + +function lockFor(version: string) { + return `packages:\n- package: acme/${PKG}\n version: ${version}\nsha1_hash: abc\n` +} + +/** What `dbt deps` leaves behind for a healthy install. */ +function installPackage(version = "1.0.0", name = PKG) { + const dir = join(root, "dbt_packages", name) + mkdirSync(dir, { recursive: true }) + writeFileSync(join(dir, "dbt_project.yml"), `name: ${name}\nversion: '${version}'\n`) +} + +function counter() { + let calls = 0 + return { + install: async () => { + calls++ + installPackage() + }, + get calls() { + return calls + }, + } +} + +const stateDir = () => join(root, "target", ".altimate-dbt") + +beforeEach(() => { + root = mkdtempSync(join(tmpdir(), "dbt-pkgs-")) +}) +afterEach(() => { + rmSync(root, { recursive: true, force: true }) +}) + +describe("packagesSatisfied / ensurePackages: when to install", () => { + test("no packages declared: nothing to do, installer never called", async () => { + writeFileSync(join(root, "dbt_project.yml"), "name: proj\n") + const c = counter() + expect((await ensurePackages(root, c.install)).action).toBe("none") + expect(c.calls).toBe(0) + }) + + test("declared but not installed: installs once, then skips", async () => { + writeProject() + const c = counter() + expect((await ensurePackages(root, c.install)).action).toBe("installed") + expect((await ensurePackages(root, c.install)).action).toBe("skipped") + expect((await ensurePackages(root, c.install)).action).toBe("skipped") + expect(c.calls).toBe(1) + }) + + test("packages already installed by something else and matching the lock: not reinstalled", async () => { + // e.g. an image built with `dbt deps`: no stamp from us, but the install is current. + writeProject({ lock: lockFor("1.0.0") }) + installPackage("1.0.0") + const c = counter() + expect((await ensurePackages(root, c.install)).action).toBe("skipped") + expect(c.calls).toBe(0) + }) + + test("installed version differs from the lock: reinstalls", async () => { + writeProject({ lock: lockFor("2.0.0") }) + installPackage("1.0.0") + const c = counter() + expect((await ensurePackages(root, c.install)).action).toBe("installed") + expect(c.calls).toBe(1) + }) + + test("a declared hub package whose directory is missing: reinstalls", async () => { + writeProject({ lock: lockFor("1.0.0") }) + installPackage("1.0.0", "other") + expect(packagesSatisfied(root).ok).toBe(false) + }) + + test("package directory without dbt_project.yml (the corruption seen in the benchmark): reinstalls", async () => { + writeProject({ lock: lockFor("1.0.0") }) + installPackage() + const c = counter() + await ensurePackages(root, c.install) // stamps a good install + rmSync(join(root, "dbt_packages", PKG, "dbt_project.yml")) + expect(packagesSatisfied(root).ok).toBe(false) + expect((await ensurePackages(root, c.install)).action).toBe("installed") + expect(existsSync(join(root, "dbt_packages", PKG, "dbt_project.yml"))).toBe(true) + }) + + test("declarations changed since the last install: reinstalls", async () => { + writeProject() + const c = counter() + await ensurePackages(root, c.install) + writeFileSync(join(root, "packages.yml"), `packages:\n - package: acme/${PKG}\n version: 1.1.0\n`) + expect((await ensurePackages(root, c.install)).action).toBe("installed") + expect(c.calls).toBe(2) + }) + + test("non-hub (local/git) packages: verified by count without a stamp, by stamp afterwards", async () => { + writeProject({ packages: "packages:\n - local: ../elsewhere\n" }) + const c = counter() + expect(packagesSatisfied(root).ok).toBe(false) // nothing installed + installPackage("0.1.0", "elsewhere") + expect(packagesSatisfied(root).ok).toBe(true) + await ensurePackages(root, c.install) + expect(c.calls).toBe(0) + }) + + test("honours packages-install-path from dbt_project.yml", async () => { + writeProject() + writeFileSync( + join(root, "dbt_project.yml"), + "name: proj\nconfig-version: 2\nprofile: proj\npackages-install-path: deps_here\n", + ) + const dir = join(root, "deps_here", PKG) + mkdirSync(dir, { recursive: true }) + writeFileSync(join(dir, "dbt_project.yml"), "name: mypkg\nversion: '1.0.0'\n") + expect(packagesSatisfied(root).ok).toBe(true) + }) +}) + +describe("crash and failure recovery", () => { + test("an install that throws leaves the project marked unfinished; the next call repairs it", async () => { + writeProject() + await expect( + ensurePackages(root, async () => { + mkdirSync(join(root, "dbt_packages", PKG), { recursive: true }) // half-installed + throw new Error("network down") + }), + ).rejects.toThrow("network down") + expect(existsSync(join(stateDir(), "packages.dirty"))).toBe(true) + expect(packagesSatisfied(root).ok).toBe(false) + + const c = counter() + expect((await ensurePackages(root, c.install)).action).toBe("installed") + expect(existsSync(join(stateDir(), "packages.dirty"))).toBe(false) + expect(packagesSatisfied(root).ok).toBe(true) + }) + + test("dirty marker forces a reinstall even when every package directory looks complete", async () => { + writeProject({ lock: lockFor("1.0.0") }) + installPackage("1.0.0") + mkdirSync(stateDir(), { recursive: true }) + writeFileSync(join(stateDir(), "packages.dirty"), "{}") + const c = counter() + expect((await ensurePackages(root, c.install)).action).toBe("installed") + expect(c.calls).toBe(1) + }) + + test("a deps run that finishes with an empty or incomplete package directory is not recorded as good", async () => { + writeProject() + await expect(ensurePackages(root, async () => {})).rejects.toThrow("incomplete") + expect(existsSync(join(stateDir(), "packages.stamp.json"))).toBe(false) + expect(existsSync(join(stateDir(), "packages.dirty"))).toBe(true) + }) +}) + +describe("lock", () => { + test("is exclusive and released afterwards", async () => { + writeProject() + const order: string[] = [] + const a = withPackagesLock(root, async () => { + order.push("a-start") + await new Promise((r) => setTimeout(r, 200)) + order.push("a-end") + }) + await new Promise((r) => setTimeout(r, 50)) + const b = withPackagesLock(root, async () => { + order.push("b-start") + }) + await Promise.all([a, b]) + expect(order).toEqual(["a-start", "a-end", "b-start"]) + expect(existsSync(join(stateDir(), "packages.lock"))).toBe(false) + }) + + test("a lock left by a process that no longer exists is broken immediately", async () => { + writeProject() + const lockDir = join(stateDir(), "packages.lock") + mkdirSync(lockDir, { recursive: true }) + // A pid that is certainly not running: spawn and reap a child. + const dead = spawn(process.execPath, ["-e", "0"]) + await new Promise((r) => dead.on("close", r)) + writeFileSync( + join(lockDir, "owner.json"), + JSON.stringify({ pid: dead.pid, host: require("os").hostname(), token: "x", startedAt: Date.now() }), + ) + const started = Date.now() + await withPackagesLock(root, async () => {}, { staleMs: 60_000 }) + expect(Date.now() - started).toBeLessThan(2_000) + }) + + test("a lock whose owner is alive but has stopped heartbeating is broken after staleMs", async () => { + writeProject() + const lockDir = join(stateDir(), "packages.lock") + mkdirSync(lockDir, { recursive: true }) + writeFileSync( + join(lockDir, "owner.json"), + JSON.stringify({ pid: process.pid + 1, host: "some-other-host", token: "x", startedAt: 0 }), + ) + const old = new Date(Date.now() - 10_000) + utimesSync(lockDir, old, old) + await withPackagesLock(root, async () => {}, { staleMs: 1_000 }) + }) + + test("a live, heartbeating owner is waited for, not broken", async () => { + writeProject() + let holderDone = false + const holder = withPackagesLock( + root, + async () => { + await new Promise((r) => setTimeout(r, 600)) // longer than staleMs: only the heartbeat keeps it alive + holderDone = true + }, + { staleMs: 300, heartbeatMs: 50 }, + ) + await new Promise((r) => setTimeout(r, 100)) + let waiterSawHolderDone = false + await withPackagesLock(root, async () => (waiterSawHolderDone = holderDone), { staleMs: 300, heartbeatMs: 50 }) + await holder + expect(waiterSawHolderDone).toBe(true) + }) + + test("gives up with a clear error when the lock is never released", async () => { + writeProject() + const lockDir = join(stateDir(), "packages.lock") + mkdirSync(lockDir, { recursive: true }) + writeFileSync(join(lockDir, "owner.json"), JSON.stringify({ pid: process.pid, host: "h", token: "x", startedAt: 0 })) + await expect(withPackagesLock(root, async () => {}, { staleMs: 60_000, maxWaitMs: 300 })).rejects.toThrow( + "timed out", + ) + }) + + test("explicit install (altimate-dbt deps) runs under the lock, clears the dirty marker and does not claim success", async () => { + writeProject({ lock: lockFor("1.0.0") }) + await installPackagesLocked(root, async () => installPackage()) + expect(packagesSatisfied(root).ok).toBe(true) + expect(existsSync(join(stateDir(), "packages.dirty"))).toBe(false) + expect(existsSync(join(stateDir(), "packages.stamp.json"))).toBe(false) + }) + + test("an explicit install whose dbt deps failed (result, not throw) leaves old packages to be verified, not trusted", async () => { + writeProject({ lock: lockFor("2.0.0") }) + installPackage("1.0.0") // stale package left over from a failed upgrade + await installPackagesLocked(root, async () => ({ stderr: "network error" })) + expect(packagesSatisfied(root).ok).toBe(false) // lock wants 2.0.0 + }) + + test("a packages.yml edited after the lock was written is not satisfied by the old install", () => { + writeProject({ lock: lockFor("1.0.0") }) + installPackage("1.0.0") + expect(packagesSatisfied(root).ok).toBe(true) + const later = new Date(Date.now() + 10_000) + utimesSync(join(root, "packages.yml"), later, later) + expect(packagesSatisfied(root)).toMatchObject({ ok: false, reason: expect.stringContaining("newer") }) + }) +}) + +describe("concurrent callers", () => { + test("many in-process callers install exactly once and all return with packages complete", async () => { + writeProject() + let installs = 0 + const seen: boolean[] = [] + const slow = async () => { + installs++ + const dir = join(root, "dbt_packages", PKG) + mkdirSync(dir, { recursive: true }) + await new Promise((r) => setTimeout(r, 150)) + writeFileSync(join(dir, "dbt_project.yml"), "name: mypkg\nversion: '1.0.0'\n") + } + await Promise.all( + Array.from({ length: 8 }, async () => { + await ensurePackages(root, slow) + seen.push(existsSync(join(root, "dbt_packages", PKG, "dbt_project.yml"))) + }), + ) + expect(installs).toBe(1) + expect(seen).toEqual(Array(8).fill(true)) + }) + + function runWorker(mode?: "crash") { + const logFile = join(root, "worker.log") + return new Promise<{ code: number | null; signal: NodeJS.Signals | null }>((resolve) => { + const c = spawn(process.execPath, [WORKER, root, logFile, PKG, ...(mode ? [mode] : [])], { stdio: "ignore" }) + c.on("close", (code, signal) => resolve({ code, signal })) + }) + } + const readLog = () => (existsSync(join(root, "worker.log")) ? readFileSync(join(root, "worker.log"), "utf-8").trim().split("\n") : []) + + test("separate processes started together: one install, nobody sees a broken package directory", async () => { + writeProject() + const results = await Promise.all(Array.from({ length: 8 }, () => runWorker())) + expect(results.every((r) => r.code === 0)).toBe(true) + const lines = readLog() + expect(lines.filter((l) => l.startsWith("install")).length).toBe(1) + expect(lines.filter((l) => l.startsWith("ready")).length).toBe(8) + expect(lines.filter((l) => l.startsWith("BROKEN")).length).toBe(0) + }, 30_000) + + test("a process killed in the middle of an install does not poison the project", async () => { + writeProject() + const crashed = await runWorker("crash") + expect(crashed.signal).toBe("SIGKILL") + // Package directory exists but is incomplete, lock and dirty marker are left behind. + expect(existsSync(join(root, "dbt_packages", PKG))).toBe(true) + expect(existsSync(join(root, "dbt_packages", PKG, "dbt_project.yml"))).toBe(false) + expect(existsSync(join(stateDir(), "packages.lock"))).toBe(true) + + // Recovery: several processes arrive at once; the dead owner's lock is broken, the install repeats once. + const results = await Promise.all(Array.from({ length: 4 }, () => runWorker())) + expect(results.every((r) => r.code === 0)).toBe(true) + const lines = readLog() + expect(lines.filter((l) => l.startsWith("BROKEN")).length).toBe(0) + expect(lines.filter((l) => l.startsWith("ready")).length).toBe(4) + expect(lines.filter((l) => l.startsWith("install")).length).toBe(2) // the crashed one + one repair + expect(packagesSatisfied(root).ok).toBe(true) + }, 30_000) +}) + +describe("adapter wiring", () => { + test("the dbt integration library is told not to run `dbt deps` itself", () => { + // adapter.ts cannot be imported under bun (python-bridge), so assert on the source. The + // library default is true; leaving it on reintroduces the unsynchronised install from + // every process. The behavioural check is test/packages-real-dbt.test.ts. + const src = readFileSync(join(import.meta.dir, "..", "src", "adapter.ts"), "utf-8") + expect(src).toMatch(/getInstallDepsOnProjectInitialization:\s*\(\)\s*=>\s*false/) + }) +}) diff --git a/packages/dbt-tools/test/schema-verify.test.ts b/packages/dbt-tools/test/schema-verify.test.ts index 04094c31ce..e634999dab 100644 --- a/packages/dbt-tools/test/schema-verify.test.ts +++ b/packages/dbt-tools/test/schema-verify.test.ts @@ -8,6 +8,12 @@ type AdapterOverrides = { nodeFound?: boolean parseManifestError?: Error getColumnsError?: Error + /** `config.contract.enforced` on the model. */ + contractEnforced?: boolean + /** Tests attached to columns of the model: column name -> test names. */ + columnTests?: Record + patchPath?: string + packageName?: string } function makeAdapter(o: AdapterOverrides = {}): DBTProjectIntegrationAdapter { @@ -20,11 +26,11 @@ function makeAdapter(o: AdapterOverrides = {}): DBTProjectIntegrationAdapter { schema: "main", alias: "target", name: "target", - package_name: "proj", + package_name: o.packageName ?? "proj", description: "", - patch_path: "schema.yml", + patch_path: o.patchPath ?? "schema.yml", columns: o.expectedColumns ?? {}, - config: {} as never, + config: (o.contractEnforced === undefined ? {} : { contract: { enforced: o.contractEnforced } }) as never, resource_type: "model", depends_on: { nodes: [], macros: [] } as never, is_external_project: false, @@ -32,6 +38,12 @@ function makeAdapter(o: AdapterOverrides = {}): DBTProjectIntegrationAdapter { meta: {}, } as unknown as NodeMetaData) + const testMetaMap = new Map() + for (const [column, names] of Object.entries(o.columnTests ?? {})) + for (const name of names) testMetaMap.set(name, { attached_node: "model.proj.target", column_name: column }) + // A test attached to some other model must never count. + testMetaMap.set("not_null_other_email", { attached_node: "model.proj.other", column_name: "email" }) + const parseManifest = o.parseManifestError ? mock(() => Promise.reject(o.parseManifestError)) : mock(() => Promise.resolve({ @@ -40,6 +52,7 @@ function makeAdapter(o: AdapterOverrides = {}): DBTProjectIntegrationAdapter { lookupByUniqueId: mock(() => node), nodes: mock(() => []), }, + testMetaMap, } as never)) const getColumnsOfModel = o.getColumnsError @@ -95,29 +108,134 @@ describe("schema-verify command", () => { expect(result.columns_reordered).toEqual([]) }) - test("detects extra columns in actual not in spec", async () => { + test("extra columns the YAML does not list are reported in the raw diff but are not a mismatch", async () => { const adapter = makeAdapter({ expectedColumns: { id: col("id"), name: col("name") }, actualColumns: [db("id"), db("name"), db("extra1"), db("extra2")], }) const result = await schemaVerify(adapter, ["--model", "target"]) as Record - expect(result.verdict).toBe("mismatch") + expect(result.verdict).toBe("match") + expect(result.findings).toEqual([]) expect(result.columns_extra).toEqual(["extra1", "extra2"]) - expect(result.columns_missing).toEqual([]) + expect((result.notes as string[]).join(" ")).toContain("not an error") }) - test("detects missing columns in actual that spec requires", async () => { + test("YAML that documents only the tested columns: undocumented columns are never an error (airbnb001 shape)", async () => { + // monthly_agg_reviews: YAML lists DATE_SENTIMENT_ID and REVIEW_SENTIMENT, the model produces six columns. const adapter = makeAdapter({ - expectedColumns: { id: col("id"), name: col("name"), email: col("email") }, - actualColumns: [db("id"), db("name")], + expectedColumns: { DATE_SENTIMENT_ID: col("DATE_SENTIMENT_ID"), REVIEW_SENTIMENT: col("REVIEW_SENTIMENT") }, + columnTests: { DATE_SENTIMENT_ID: ["unique_x", "not_null_x"], REVIEW_SENTIMENT: ["accepted_values_x"] }, + actualColumns: [ + db("REVIEW_TOTALS"), db("REVIEW_SENTIMENT"), db("MONTH_YEAR"), db("MONTH"), db("YEAR"), db("DATE_SENTIMENT_ID"), + ], }) - const result = await schemaVerify(adapter, ["--model", "target"]) as Record + const result = await schemaVerify(adapter, ["--model", "monthly_agg_reviews"]) as Record + expect(result.verdict).toBe("match") + expect(result.findings).toEqual([]) + // Column order differs from the YAML too; that is not an error either. + expect((result.columns_reordered as unknown[]).length).toBeGreaterThan(0) + }) + + test("a declared column the model does not produce, with a test attached, is a finding naming the file and the test", async () => { + const adapter = makeAdapter({ + expectedColumns: { id: col("id"), email: col("email") }, + columnTests: { email: ["not_null_target_email"] }, + patchPath: "proj://models/schema.yml", + actualColumns: [db("id")], + }) + const result = await schemaVerify(adapter, ["--model", "target"]) as Record expect(result.verdict).toBe("mismatch") expect(result.columns_missing).toEqual(["email"]) - expect(result.columns_extra).toEqual([]) + expect(result.findings).toHaveLength(1) + expect(result.findings[0].kind).toBe("tested-column-missing") + expect(result.findings[0].columns).toEqual(["email"]) + expect(result.findings[0].evidence).toContain("models/schema.yml") + expect(result.findings[0].evidence).toContain("not_null_target_email") + expect(result.spec).toEqual({ declared_in: "models/schema.yml", package: "proj", contract_enforced: false }) + }) + + test("a declared column the model does not produce, with nothing attached, is only a note (asana001 shape)", async () => { + // Package YAML declares assignee_status (description only); the model never produced it. + const adapter = makeAdapter({ + expectedColumns: { task_id: col("task_id"), assignee_status: col("assignee_status") }, + columnTests: { task_id: ["unique_x"] }, + patchPath: "asana_source://models/stg_asana.yml", + packageName: "proj", + actualColumns: [db("task_id"), db("name")], + }) + const result = await schemaVerify(adapter, ["--model", "target"]) as Record + expect(result.verdict).toBe("match") + expect(result.findings).toEqual([]) + const notes = (result.notes as string[]).join("\n") + expect(notes).toContain("assignee_status") + expect(notes).toContain("models/stg_asana.yml") + expect(notes).toContain("package `asana_source`") + expect(notes).toContain("stale or aspirational") + }) + + test("tests attached to a different model do not make a missing column a finding", async () => { + const adapter = makeAdapter({ + expectedColumns: { id: col("id"), email: col("email") }, + columnTests: {}, + actualColumns: [db("id")], + }) + // makeAdapter always adds not_null_other_email on model.proj.other; this model's `email` is missing. + const result = await schemaVerify(adapter, ["--model", "target"]) as Record + expect(result.verdict).toBe("match") + }) + + test("enforced contract: a missing column is a finding", async () => { + const adapter = makeAdapter({ + contractEnforced: true, + expectedColumns: { id: col("id", "integer"), name: col("name", "varchar") }, + actualColumns: [db("id", "INTEGER")], + }) + const result = await schemaVerify(adapter, ["--model", "target"]) as Record + expect(result.verdict).toBe("mismatch") + expect(result.findings.map((f: any) => f.kind)).toEqual(["contract-missing-columns"]) + expect(result.findings[0].evidence).toContain("enforced contract") + expect(result.spec.contract_enforced).toBe(true) + }) + + test("enforced contract: an extra column is a finding (dbt rejects it)", async () => { + const adapter = makeAdapter({ + contractEnforced: true, + expectedColumns: { id: col("id"), name: col("name") }, + actualColumns: [db("id"), db("name"), db("amount")], + }) + const result = await schemaVerify(adapter, ["--model", "target"]) as Record + expect(result.verdict).toBe("mismatch") + expect(result.findings.map((f: any) => f.kind)).toEqual(["contract-extra-columns"]) + expect(result.findings[0].columns).toEqual(["amount"]) + }) + + test("enforced contract that the table satisfies: match, even when the order differs", async () => { + const adapter = makeAdapter({ + contractEnforced: true, + expectedColumns: { id: col("id", "integer"), name: col("name", "varchar") }, + actualColumns: [db("name", "VARCHAR"), db("id", "INTEGER")], + }) + const result = await schemaVerify(adapter, ["--model", "target"]) as Record + expect(result.verdict).toBe("match") + }) + + test("contract = false is not a contract", async () => { + const adapter = makeAdapter({ + contractEnforced: false, + expectedColumns: { id: col("id") }, + actualColumns: [db("id"), db("extra")], + }) + expect(((await schemaVerify(adapter, ["--model", "target"])) as Record).verdict).toBe("match") + }) + + test("model without any YAML columns: no-spec, whatever the table looks like", async () => { + const adapter = makeAdapter({ expectedColumns: {}, actualColumns: [db("a"), db("b"), db("c")], patchPath: "" }) + const result = await schemaVerify(adapter, ["--model", "target"]) as Record + expect(result.verdict).toBe("no-spec") + expect(result.findings).toBeUndefined() }) - test("detects column reordering when same set but different position", async () => { + test("column order that differs from the YAML is reported in the raw diff, never as a mismatch", async () => { const adapter = makeAdapter({ // schema.yml order: id, name, email expectedColumns: { id: col("id"), name: col("name"), email: col("email") }, @@ -125,13 +243,11 @@ describe("schema-verify command", () => { actualColumns: [db("name"), db("id"), db("email")], }) const result = await schemaVerify(adapter, ["--model", "target"]) as Record - expect(result.verdict).toBe("mismatch") + expect(result.verdict).toBe("match") expect(result.columns_extra).toEqual([]) expect(result.columns_missing).toEqual([]) const reordered = result.columns_reordered as Array<{ column: string }> - expect(reordered.length).toBeGreaterThan(0) - const reorderedNames = reordered.map((r) => r.column) - expect(reorderedNames).toContain("id") + expect(reordered.map((r) => r.column)).toContain("id") }) test("case-insensitive name comparison (dbt convention)", async () => { @@ -143,16 +259,26 @@ describe("schema-verify command", () => { expect(result.verdict).toBe("match") }) - test("detects type mismatch when spec declares a different data_type", async () => { + test("declared data_type that differs without a contract: raw diff only, no mismatch", async () => { const adapter = makeAdapter({ expectedColumns: { id: col("id", "INTEGER"), name: col("name", "VARCHAR") }, actualColumns: [db("id", "BIGINT"), db("name", "VARCHAR")], }) const result = await schemaVerify(adapter, ["--model", "target"]) as Record - expect(result.verdict).toBe("mismatch") - const mm = result.type_mismatches as Array<{ column: string; actual_type: string; expected_type: string }> - expect(mm.length).toBe(1) - expect(mm[0]?.column).toBe("id") + expect(result.verdict).toBe("match") + const mm = result.type_mismatches as Array<{ column: string }> + expect(mm.map((t) => t.column)).toEqual(["id"]) + }) + + test("declared data_type that differs under a contract: noted, because dbt checks types itself at build", async () => { + const adapter = makeAdapter({ + contractEnforced: true, + expectedColumns: { id: col("id", "INTEGER") }, + actualColumns: [db("id", "BIGINT")], + }) + const result = await schemaVerify(adapter, ["--model", "target"]) as Record + expect(result.verdict).toBe("match") + expect(result.notes.join(" ")).toContain("data_type") }) test("ignores type mismatch when spec does not declare data_type", async () => { @@ -175,9 +301,9 @@ describe("schema-verify command", () => { expect((result as { error: string }).error).toContain("Build the model first") }) - test("realistic ade-bench f1002 pattern — extra rank-breakdown columns", async () => { - // Spec: just rank, driver_full_name, podiums - // Actual: agent helpfully added p1, p2, p3 breakdowns + test("f1002 shape: extra rank-breakdown columns beyond the YAML are not an error by themselves", async () => { + // YAML: rank, driver_full_name, podiums. Model also returns p1, p2, p3. dbt is happy with that; + // whether the extras are wanted is a question for the task, not for the YAML. const adapter = makeAdapter({ expectedColumns: { rank: col("rank"), @@ -187,25 +313,18 @@ describe("schema-verify command", () => { actualColumns: [db("rank"), db("driver_full_name"), db("podiums"), db("p1"), db("p2"), db("p3")], }) const result = await schemaVerify(adapter, ["--model", "most_podiums"]) as Record - expect(result.verdict).toBe("mismatch") + expect(result.verdict).toBe("match") expect(result.columns_extra).toEqual(["p1", "p2", "p3"]) expect(result.columns_missing).toEqual([]) }) - test("realistic ade-bench pattern — column-order divergence (product_id-first vs inventory_id-first)", async () => { + test("no output of schema-verify tells the reader to remove, add or reorder anything", async () => { const adapter = makeAdapter({ - // Spec leads with product_id - expectedColumns: { - product_id: col("product_id"), - product_code: col("product_code"), - inventory_id: col("inventory_id"), - }, - // Actual leads with inventory_id - actualColumns: [db("inventory_id"), db("product_id"), db("product_code")], + expectedColumns: { id: col("id"), gone: col("gone") }, + columnTests: { gone: ["not_null_target_gone"] }, + actualColumns: [db("name"), db("id")], }) - const result = await schemaVerify(adapter, ["--model", "obt_product_inventory"]) as Record - expect(result.verdict).toBe("mismatch") - const reordered = result.columns_reordered as Array<{ column: string }> - expect(reordered.length).toBeGreaterThan(0) + const text = JSON.stringify(await schemaVerify(adapter, ["--model", "target"])) + expect(text).not.toMatch(/\b(REMOVE|ADD|REORDER|CAST)\b/) }) }) From 8e09f7f5c011789df24f65b8f6426abc3b5a8374 Mon Sep 17 00:00:00 2001 From: anandgupta42 Date: Tue, 6 Oct 2026 14:48:30 -0700 Subject: [PATCH 2/7] fix(validators): stop false alarms from `dbt-schema-verify` and `dbt-tests-pass` - `dbt-schema-verify` no longer instructs the agent to remove or add columns so a model matches its YAML; it reports only real dbt errors, names the file that declares the column, and suggests no destructive change. - `dbt-tests-pass` treats dbt's "Nothing to do" (a model with no tests) as "nothing to check" rather than "could not be tested". Failing tests and real run errors still fail. - An `altimate-dbt` check that errors is retried once on its own, because the validators run several processes against one single-writer DuckDB file. - Rewrites the `dbt-schema-verify` and `dbt-develop` skills that taught the old rule, and updates the validator and dbt-tools docs. Co-Authored-By: Claude Opus 5.5 (1M context) --- .opencode/skills/dbt-develop/SKILL.md | 6 +- .opencode/skills/dbt-schema-verify/SKILL.md | 95 ++++---- docs/docs/data-engineering/tools/dbt-tools.md | 2 + docs/docs/data-engineering/validators.md | 75 +++++- .../altimate/validators/dbt-schema-verify.ts | 82 +++++-- .../src/altimate/validators/dbt-tests-pass.ts | 75 +++++- .../altimate/validators/validator-utils.ts | 29 +++ .../validators/dbt-schema-verify.test.ts | 227 ++++++++++++++++++ .../validators/dbt-tests-pass-check.test.ts | 177 ++++++++++++++ .../validators/e2e-real-dbt-5.test.ts | 28 ++- .../validators/fake-altimate-dbt.helper.ts | 84 +++++++ 11 files changed, 786 insertions(+), 94 deletions(-) create mode 100644 packages/opencode/test/altimate/validators/dbt-schema-verify.test.ts create mode 100644 packages/opencode/test/altimate/validators/dbt-tests-pass-check.test.ts create mode 100644 packages/opencode/test/altimate/validators/fake-altimate-dbt.helper.ts diff --git a/.opencode/skills/dbt-develop/SKILL.md b/.opencode/skills/dbt-develop/SKILL.md index 32b4254420..4384bf1ac1 100644 --- a/.opencode/skills/dbt-develop/SKILL.md +++ b/.opencode/skills/dbt-develop/SKILL.md @@ -168,7 +168,9 @@ altimate-dbt info # confirm every req **Diff column shape against the spec — use the `dbt-schema-verify` skill.** For each model the task touched, run `altimate-dbt schema-verify --model -` and treat any `mismatch` verdict as "not done." Full procedure, +` and treat a `mismatch` verdict (an enforced contract the table does +not match, or a declared column with tests that the table lacks) as "not +done." Columns the YAML does not list are not a mismatch. Full procedure, output interpretation, and fallback (when `altimate-dbt` is missing) live in the dedicated **dbt-schema-verify** skill, which auto-loads alongside this one. @@ -204,7 +206,7 @@ Use `altimate-dbt children` and `altimate-dbt parents` to verify the DAG is inta 5. **Fix ALL errors, not just yours.** After creating/modifying models, run a full `dbt build`. If ANY model fails — even pre-existing ones you didn't touch — fix them. Your job is to leave the project in a fully working state. 6. **Verify transformation correctness, not just mechanics.** For non-trivial models, generate and run dbt unit tests as part of the validate step (use the `dbt-unit-tests` skill). Passing `dbt build` only proves the SQL is syntactically valid — it doesn't prove the *values* are right. 7. **Enumerate deliverables, then check them off.** The task is not done until every model, column, test, and config change explicitly requested exists on disk and in the manifest. Re-read the prompt at the end and verify each requested item — don't trust your own intermediate "done" feeling. -8. **Match the column spec exactly — and verify it mechanically, not by inspection.** Use the dedicated **dbt-schema-verify** skill. Before declaring any model task done, run `altimate-dbt schema-verify --model ` and treat any `mismatch` verdict as "not done." Adding "helpful" extras (rank breakdowns, name-resolved fields, lineage metadata), reordering columns "more logically", or substituting synonyms (`supplier_id` for `supplier_company`, `transaction_type_name` for `transaction_type`) all break equality tests. The contract is what the spec says, not what you think would be useful. +8. **Produce the columns the task asks for — and check declared columns mechanically, not by inspection.** Use the dedicated **dbt-schema-verify** skill. Before declaring any model task done, run `altimate-dbt schema-verify --model ` and treat a `mismatch` verdict as "not done." Adding "helpful" extras (rank breakdowns, name-resolved fields, lineage metadata) or substituting synonyms (`supplier_id` for `supplier_company`) can break the task's own expectations: follow the columns the task and the existing consumers of the model call for. YAML that lists only some columns does not mean the others should be removed, and a YAML column nothing tests or enforces may be stale rather than required. ## Common Pitfalls in Transformation Logic diff --git a/.opencode/skills/dbt-schema-verify/SKILL.md b/.opencode/skills/dbt-schema-verify/SKILL.md index 80b7dab31d..5a24e6bea3 100644 --- a/.opencode/skills/dbt-schema-verify/SKILL.md +++ b/.opencode/skills/dbt-schema-verify/SKILL.md @@ -4,21 +4,17 @@ applyPaths: - "dbt_project.yml" - "**/dbt_project.yml" description: | - REQUIRED after building or modifying ANY dbt model that has columns declared - in `schema.yml` / `_models.yml`. Run `altimate-dbt schema-verify --model - ` to diff actual columns against the spec, and treat any `mismatch` - verdict as "not done." - - The most common reason "the build is green but the tests still fail" is - that the model produces the right *data values* in the wrong *column - shape* — extra columns, missing columns, wrong order, wrong types. Many - dbt equality tests grade the column tuple `(name, type, position)` - exactly, and the agent's prior bias is to add "helpful" extras - (`p1`/`p2`/`p3` rank breakdowns, name-resolved variants, lineage - metadata) or reorder columns "more logically." Both break the contract. - - This skill enforces the mechanical check that catches those bugs before - declaring done. Use it before declaring any model task complete. + Use after building or modifying a dbt model whose columns are declared in + YAML (`schema.yml` / `_models.yml`). Run `altimate-dbt schema-verify --model + ` to compare the built table with the declared columns, and treat a + `mismatch` verdict (a non-empty `findings` list) as "not done" until you have + decided whether the model or the YAML is wrong. + + YAML `columns:` is documentation plus the anchor for column tests. It is not + an exhaustive list of what a model may produce, so columns the YAML does not + mention are not an error unless a contract is enforced. What does count: an + enforced contract the table does not match, and a declared column that has + tests attached but is not produced. --- # dbt schema-verify @@ -72,23 +68,33 @@ Returns a structured JSON result: | verdict | meaning | what to do | |---|---|---| -| `match` | actual columns match the spec exactly (case-insensitive on names) | DONE — proceed | -| `mismatch` | one or more of `columns_extra`, `columns_missing`, `columns_reordered`, `type_mismatches` is non-empty | NOT DONE — read the diff, fix the model SQL, rebuild, re-run schema-verify | -| `no-spec` | the model has no columns declared in `schema.yml` | DONE for shape-fidelity purposes — no contract to verify against | - -## How to act on a `mismatch` - -For each non-empty list, the fix is mechanical: - -| Field | What it means | What to change in the model SQL | -|---|---|---| -| `columns_extra` | columns in your model NOT in the spec | REMOVE them from the `SELECT` | -| `columns_missing` | columns in the spec NOT in your model | ADD them to the `SELECT` (compute them, or rename an existing column if you used a synonym) | -| `columns_reordered` | columns present in both but at different positions | REORDER the columns in your `SELECT` to match the spec's order | -| `type_mismatches` | declared `data_type` in spec disagrees with the warehouse's reported type | CAST in the `SELECT` or change the upstream source | - -Then run `altimate-dbt build --model ` again, then re-run -`altimate-dbt schema-verify --model ` until verdict is `match`. +| `match` | nothing dbt treats as an error was found (`notes` may still hold true observations) | proceed | +| `mismatch` | `findings` is non-empty: an enforced contract the table does not match, or a declared column with tests attached that the table lacks | NOT DONE until you decide which side is wrong | +| `no-spec` | the model has no columns declared in YAML | nothing to compare | + +`spec` says which YAML file declares the columns, in which package, and +whether a contract is enforced. Each entry in `findings` carries the +evidence. `columns_extra` / `columns_missing` / `columns_reordered` / +`type_mismatches` are the raw diff against the YAML, not instructions: +`columns_extra` only means "not listed in the YAML". + +## How to act on a finding + +A finding says two things disagree. It does not say which one is wrong. + +- **Contract enforced.** dbt will refuse to build a model whose columns differ + from its contract. Either the model should produce exactly the contract's + columns, or the contract is out of date. Decide from the task. +- **Declared column with tests, not produced.** The tests read a column that + is not there and will fail or error. Either the model should produce the + column, or the YAML entry (and its tests) is stale. Decide from the task. +- **Never** delete or invent a column just because the YAML does or does not + list it. Undocumented columns are normal, and a declared column with no + tests and no contract (a `note`) may simply be stale. If the task states + which columns the output must have, follow the task. + +Then run `altimate-dbt build --model ` again and re-run +`altimate-dbt schema-verify --model `. ## Iron Rules @@ -96,18 +102,14 @@ Then run `altimate-dbt build --model ` again, then re-run columns yourself and concluding "looks right to me" does not count. Run the command and read its output. 2. **A `mismatch` is "not done", even if the build is green.** dbt build - only proves the SQL compiled and ran without errors. It does not prove - the column shape is correct. Equality tests grade shape AND values. -3. **Do not reinterpret the spec to make the model right.** The spec is - the contract. If the spec lists `supplier_company` and your model has - `supplier_id`, the answer is to fix your model, not to argue that - `supplier_id` is more useful. + does not check declared columns unless a contract is enforced. +3. **Do not change the model to satisfy YAML that only documents some + columns.** The YAML is evidence about intent, not the spec of the whole + table. The task and the existing consumers of the model decide which + columns are wanted. 4. **Run schema-verify on every model touched, not just the last one.** - The most common "almost-pass" is N-1 models passing and the Nth one - silently failing on column shape. Walk the list. -5. **Skip only on `no-spec`.** Do not skip on the grounds that the model - is small, or trivial, or "obvious." The spec is small only because - the dbt project author already curated it. +5. **`no-spec` and `match` with notes need no action.** Read the notes; act + on them only if they match what the task asked for. ## Fallback when altimate-dbt is unavailable @@ -123,10 +125,9 @@ cat models/**/*.yml | grep -A 50 "name: " # or: yq eval '...' models/**/ dbt show --select --limit 0 ``` -Compare the two ordered lists. Produce the same four-bucket diff -(`columns_extra`, `columns_missing`, `columns_reordered`, -`type_mismatches`) in your head, and apply the same fix logic. The -mechanics don't change; only the tool name does. +Compare the two lists and apply the same rules: only an enforced +contract, or a declared column with tests that the table lacks, is a +problem. Columns the YAML does not list are not. ## What this skill does NOT cover diff --git a/docs/docs/data-engineering/tools/dbt-tools.md b/docs/docs/data-engineering/tools/dbt-tools.md index 79714e1bb5..096ae155dd 100644 --- a/docs/docs/data-engineering/tools/dbt-tools.md +++ b/docs/docs/data-engineering/tools/dbt-tools.md @@ -168,6 +168,8 @@ All commands provide friendly error diagnostics with actionable fix suggestions > **Tip:** In builder mode, the agent prefers `altimate-dbt` over the raw `dbt_run` tool for better error handling and auto-detection. +**Packages.** If the project declares packages (`packages.yml`, `dependencies.yml`), `altimate-dbt` installs them when a command starts, but only when they are missing or out of date (a package directory is absent or has no `dbt_project.yml`, the lock file pins a different version, or the declarations changed since the last install). Installation is serialised across processes with a lock under `target/.altimate-dbt/`, so several `altimate-dbt` commands started together install at most once and cannot corrupt `dbt_packages/`. A lock left by a crashed process is broken automatically (owner process gone, or no heartbeat for 30 seconds) and the interrupted install is repeated. `altimate-dbt deps` always runs `dbt deps`, under the same lock. Because every command re-parses the project and dbt rewrites `target/manifest.json` in place, a command that reads the manifest while a sibling rewrites it re-reads it (up to six times, with backoff) before reporting it missing. + --- ## dbt Skills diff --git a/docs/docs/data-engineering/validators.md b/docs/docs/data-engineering/validators.md index 0aaab133ef..b27a36804a 100644 --- a/docs/docs/data-engineering/validators.md +++ b/docs/docs/data-engineering/validators.md @@ -53,7 +53,8 @@ deterministic-validators work and are currently soak-tested in shadow mode After the agent declares done, runs `altimate-dbt test --model ` against every dbt model the agent modified during this session. Refuses -to terminate if any model's tests fail or error. +to terminate if a model's tests fail, or if tests could not be run +because dbt itself errored. **Catches**: row-data correctness errors (`relationships`, `unique`, `not_null`, `accepted_values`, `AUTO_*_equality` tests) — the kind of @@ -61,16 +62,59 @@ bug that column-shape verification cannot detect because the schema can be green while the SELECT logic produces wrong values or wrong row counts. +**What counts as what** + +| dbt result for the model | Validator | +|---|---| +| Tests ran and all passed | passes | +| Tests ran and some failed or errored | **fails**, naming the failing tests | +| dbt selected no tests (it prints `Nothing to do` and no summary) | passes, and sends nothing. A model with no tests has nothing to check. Recorded as `no_tests` in telemetry. | +| The selector matched no enabled node | passes (`unmatched_selector_models` in telemetry) | +| dbt aborted (connection, lock, compile or runtime error) before or while running tests | **fails**, with dbt's error text | + +An error from a run that happened in parallel with other `altimate-dbt` +processes is retried once, alone, before it is reported, because several +processes sharing a single-writer warehouse such as DuckDB can fail each +other without anything being wrong with the model (`retried_serially` in +telemetry). + ### `dbt-schema-verify` After the agent declares done, runs `altimate-dbt schema-verify --model -` on every modified model. Reports a mismatch if the produced -column shape diverges from the `schema.yml` spec (extra, missing, -reordered, or type-mismatched columns). - -**Catches**: column-level drift that wouldn't be caught by `dbt build` -alone — equality tests against the spec would fail later but the -agent has already declared done. +` on every modified model and compares the built table with the +columns its YAML declares. It reports only what dbt's own semantics +establish as a problem, and says which file declares what: + +- **Enforced contract.** The model has `contract: {enforced: true}` and + the built table has columns the contract does not list, or lacks columns + it lists. dbt itself rejects such a model. +- **Declared column with tests, not produced.** The YAML declares a column + that has at least one test attached, but the built table does not have it. + The test reads a column that does not exist. + +It does **not** report, and never tells the agent to remove, add or +reorder anything because of: + +- columns the model produces that the YAML does not list. YAML commonly + documents only some columns (those with tests or descriptions); that is + not an exhaustive list unless a contract is enforced; +- a declared column with no tests and no contract that the model lacks. + This may be a stale or aspirational entry, so it is only a note in the + `altimate-dbt schema-verify` output; +- column order, or a declared `data_type` that differs from the warehouse + type outside a contract (dbt does not enforce either, and adapters spell + types differently); +- a model with no YAML entry or no declared columns (`no-spec`). + +The message states the evidence and leaves the decision with the agent +(change the model if the YAML is right, or the YAML if its entry is +stale). The same classification appears in `altimate-dbt schema-verify`, +`altimate-dbt build --model` and the project-wide `altimate-dbt build` +output as `findings` (established problems), `notes` (true observations) +and `spec` (declaring file, package, contract flag); `columns_extra`, +`columns_missing`, `columns_reordered` and `type_mismatches` remain as the +raw diff against the YAML and are not instructions. A tool error is still a +failure (drift cannot be ruled out) and now includes the first error text. ## Completion gates (shadow mode) @@ -234,6 +278,10 @@ When a validator runs, it returns: // validator-specific extras: // dbt-tests-pass: passed?: number + no_tests?: number + no_tests_models?: string[] + unmatched_selector_models?: string[] + retried_serially?: number failed?: number errored?: number spawn_failures?: number @@ -244,13 +292,14 @@ When a validator runs, it returns: match?: number mismatch?: number no_spec?: number + diff_not_established?: number mismatch_models?: string[] } } ``` `reason` names the failing models inline (e.g. `"2 of 3 models you -edited have a column-shape mismatch against schema.yml: foo, bar"`). +edited contradict the YAML that declares their columns: foo, bar"`). ## Phased rollout plan @@ -284,9 +333,11 @@ default-on. Track progress in deeper (workspace layouts) are not detected. - Multiple `dbt_project.yml` candidates pick the alphabetically-first match deterministically. -- The validator surfaces "schema mismatch" even when the real cause - is "model never materialized" — distinguishing these requires - changes inside `altimate-dbt`. +- A model that was never materialized is reported as a tool error + ("has no warehouse table. Build it first"), not as a column finding. +- Retries consume the shared validator budget (three by default) whether + or not the failure text changed. A failure that is re-sent verbatim on + every retry spends the budget without giving the agent new information. ## Writing custom validators diff --git a/packages/opencode/src/altimate/validators/dbt-schema-verify.ts b/packages/opencode/src/altimate/validators/dbt-schema-verify.ts index 5825734f87..2c04efc0a5 100644 --- a/packages/opencode/src/altimate/validators/dbt-schema-verify.ts +++ b/packages/opencode/src/altimate/validators/dbt-schema-verify.ts @@ -4,8 +4,15 @@ * * Fires after the agent declares done. Detects whether the session touched * any dbt models, runs `altimate-dbt schema-verify` against each touched - * model, and reports a mismatch if the produced column shape diverges from - * the schema.yml spec. + * model, and reports the disagreements between the built table and the YAML + * that declares its columns that dbt's own semantics establish as problems: + * an enforced contract that the table does not match, or a declared column + * that has tests attached but is not produced. + * + * It deliberately says nothing about columns the model produces that the YAML + * does not list (YAML commonly documents only some columns), and never tells + * the agent to remove or add columns. See `packages/dbt-tools/src/commands/ + * schema-verify.ts` for the rule. * * The agent does not see this validator existing — it runs in the harness * AFTER `finishReason === "stop"`. Its output is surfaced to the agent only @@ -24,11 +31,22 @@ import { modelNameFromPath, extractLastJsonObject, runWithConcurrencyLimit, + retryErroredSerially, } from "./validator-utils" +interface SchemaVerifyFinding { + kind?: string + columns?: string[] + /** Which YAML declares what, and why it is a problem. Written by altimate-dbt. */ + evidence?: string +} + interface SchemaVerifyOutput { model?: string verdict?: "match" | "mismatch" | "no-spec" + /** Present from the altimate-dbt that classifies findings; absent from older builds. */ + findings?: SchemaVerifyFinding[] + notes?: string[] columns_extra?: string[] columns_missing?: string[] columns_reordered?: unknown[] @@ -36,6 +54,17 @@ interface SchemaVerifyOutput { error?: string } +/** + * Established problems for one result. Only `findings` count: the raw + * extra/missing/reordered/type lists are an unfiltered diff against the YAML + * and prove nothing on their own. An older altimate-dbt that does not emit + * `findings` cannot say whether a contract or a test is involved, so its + * "mismatch" verdict is not treated as established. + */ +function establishedFindings(r: SchemaVerifyOutput): SchemaVerifyFinding[] { + return r.verdict === "mismatch" && Array.isArray(r.findings) ? r.findings.filter((f) => f.evidence) : [] +} + /** * Extract a SchemaVerifyOutput JSON object from mixed stdout. * `altimate-dbt schema-verify` may emit dbt log noise (ANSI codes, parser @@ -116,24 +145,17 @@ async function runSchemaVerify(model: string, cwd: string): Promise 0) { - lines.push(` • Columns in your model NOT in spec — REMOVE: ${m.columns_extra.join(", ")}`) - } - if (m.columns_missing && m.columns_missing.length > 0) { - lines.push(` • Columns in spec NOT in your model — ADD: ${m.columns_missing.join(", ")}`) - } - if (m.columns_reordered && m.columns_reordered.length > 0) { - lines.push(` • Columns in wrong order — REORDER the SELECT to match schema.yml`) - } - if (m.type_mismatches && m.type_mismatches.length > 0) { - lines.push(` • Type mismatches — CAST or change the upstream source`) - } + for (const f of establishedFindings(m)) lines.push(` • ${f.evidence}`) } return lines.join("\n") } @@ -141,7 +163,7 @@ function formatFixHint(mismatches: SchemaVerifyOutput[]): string { export const DbtSchemaVerifyValidator: Validator = { name: "dbt-schema-verify", description: - "After the agent declares done, runs `altimate-dbt schema-verify` on every dbt model the agent modified during this session and refuses to terminate if any model's actual columns diverge from the schema.yml spec (extra, missing, reordered, or type-mismatched).", + "After the agent declares done, runs `altimate-dbt schema-verify` on every dbt model the agent modified during this session and refuses to terminate if the built table contradicts its YAML in a way dbt itself treats as an error: an enforced contract the columns do not match, or a declared column with tests attached that the model does not produce. Columns the model produces that the YAML does not list are not reported.", async appliesTo(ctx: ValidatorContext): Promise { // Only run for sessions that took place inside a dbt project. Quick check. @@ -179,11 +201,19 @@ export const DbtSchemaVerifyValidator: Validator = { // Run schema-verify calls with a bounded concurrency limit to prevent // resource contention from too many simultaneous dbt processes. let spawnFailures = 0 - const outputs = await runWithConcurrencyLimit( + const parallel = await runWithConcurrencyLimit( touched, (path) => runSchemaVerify(modelNameFromPath(path), dbtRoot), VALIDATOR_CONCURRENCY, ) + // An error from a parallel run may be the processes contending for the warehouse + // (single-writer DuckDB), not a fact about the model: retry those one at a time. + const { outputs, retried } = await retryErroredSerially( + touched, + parallel, + (path) => runSchemaVerify(modelNameFromPath(path), dbtRoot), + (o) => (o.verdict ? undefined : o.error), + ) const results: SchemaVerifyOutput[] = [] for (let i = 0; i < outputs.length; i++) { const out = outputs[i]! @@ -198,9 +228,11 @@ export const DbtSchemaVerifyValidator: Validator = { } } - const mismatches = results.filter((r) => r.verdict === "mismatch") + const mismatches = results.filter((r) => establishedFindings(r).length > 0) const noSpec = results.filter((r) => r.verdict === "no-spec").length - const matches = results.filter((r) => r.verdict === "match").length + // "match" or a mismatch verdict with nothing established (nothing to act on). + const matches = results.filter((r) => !r.error && (r.verdict === "match" || r.verdict === "mismatch") && establishedFindings(r).length === 0).length + const unestablished = results.filter((r) => !r.error && r.verdict === "mismatch" && establishedFindings(r).length === 0).length const errored = results.filter((r) => r.error).length const baseDetails = { @@ -208,7 +240,11 @@ export const DbtSchemaVerifyValidator: Validator = { verified: results.length, match: matches, no_spec: noSpec, + // Results whose raw diff differs from the YAML but where nothing is established + // (unlisted columns, untested declared-but-absent columns). Telemetry only. + diff_not_established: unestablished, errored, + retried_serially: retried, spawn_failures: spawnFailures, dbt_root: dbtRoot, session_id: ctx.sessionID, @@ -229,10 +265,12 @@ export const DbtSchemaVerifyValidator: Validator = { // (set at push time), so the operator can see WHICH models couldn't be verified // instead of an anonymous count. const erroredNames = results.filter((r) => r.error).map((r) => r.model).filter(Boolean) as string[] + // The cause, not just a count: the first error, trimmed to one line. + const firstError = results.find((r) => r.error)?.error?.replace(/\s+/g, " ").trim().slice(0, 240) const reason = mismatches.length > 0 - ? `${mismatches.length} of ${results.length} models you edited have a column-shape mismatch against schema.yml${mismatchNames.length ? `: ${mismatchNames.join(", ")}` : ""}. The build may be green, but equality tests will fail.` - : `${errored} model(s) could not be schema-verified (spawn or tool errors)${erroredNames.length ? `: ${erroredNames.join(", ")}` : ""} — schema drift cannot be ruled out. Investigate before declaring done.` + ? `${mismatches.length} of ${results.length} models you edited contradict the YAML that declares their columns${mismatchNames.length ? `: ${mismatchNames.join(", ")}` : ""}.` + : `${errored} model(s) could not be schema-verified (spawn or tool errors)${erroredNames.length ? `: ${erroredNames.join(", ")}` : ""} — schema drift cannot be ruled out${firstError ? ` (first error: ${firstError})` : ""}. Investigate before declaring done.` // altimate_change end return { @@ -241,7 +279,7 @@ export const DbtSchemaVerifyValidator: Validator = { fixHint: mismatches.length > 0 ? formatFixHint(mismatches) + - `\n\nFix the model SQL to match the schema.yml spec (do not edit the spec), rebuild, and the harness will re-check before declaring done.` + `\n\nThese are facts about the built table and the YAML, not a verdict on which side is wrong: change the model if the YAML is right, or the YAML if its entry is stale. A column the model produces that the YAML does not list is not an error unless a contract is enforced.` : `Run \`altimate-dbt schema-verify \` manually to diagnose the error. Check that altimate-dbt is on PATH and that the dbt project compiles cleanly.`, details: { ...baseDetails, diff --git a/packages/opencode/src/altimate/validators/dbt-tests-pass.ts b/packages/opencode/src/altimate/validators/dbt-tests-pass.ts index ecc9282908..651eb039da 100644 --- a/packages/opencode/src/altimate/validators/dbt-tests-pass.ts +++ b/packages/opencode/src/altimate/validators/dbt-tests-pass.ts @@ -27,6 +27,7 @@ import { modelNameFromPath, extractLastJsonObject, runWithConcurrencyLimit, + retryErroredSerially, } from "./validator-utils" export interface TestSummary { @@ -47,6 +48,13 @@ interface TestRunOutput { summary?: TestSummary /** Top-level error from altimate-dbt (manifest missing, compile error, etc.). */ error?: string + /** + * dbt ran and selected no tests for this model (it printed "Nothing to do" + * and no summary). That is "nothing to check", not a failed check. + */ + noTests?: boolean + /** Subset of `noTests`: dbt said the selector matched no enabled node at all. */ + unmatchedSelector?: boolean } /** @@ -116,6 +124,25 @@ export function parseDbtTestOutput(stdout: string): TestSummary | null { return { total, pass, error, failingTests } } +// dbt prints these instead of a `Done. PASS=...` summary when the selection has +// no tests (every dbt 1.x we have seen; a selector that matches no node adds +// "does not match any enabled nodes" before it). +const NOTHING_TO_DO_RE = /\bNothing to do\b|\bNo nodes selected\b/i +const UNMATCHED_SELECTOR_RE = /does not match any (?:enabled )?nodes?/i +// Markers of a run that failed before or while selecting. If any is present +// "Nothing to do" is not trustworthy, so the run stays an error. +const DBT_ABORT_RE = /Encountered an error|Compilation Error|Parsing Error|Runtime Error|Database Error|Could not set lock/i + +/** + * Did dbt run and find no tests for the selection? Only true when dbt printed + * its "Nothing to do" notice, no run summary, and no sign of an abort. + */ +export function isNothingToTest(dbtLog: string): { noTests: boolean; unmatchedSelector: boolean } { + const cleaned = stripAnsi(dbtLog ?? "") + const noTests = NOTHING_TO_DO_RE.test(cleaned) && !DBT_ABORT_RE.test(cleaned) && parseDbtTestOutput(cleaned) === null + return { noTests, unmatchedSelector: noTests && UNMATCHED_SELECTOR_RE.test(cleaned) } +} + /** Strip ANSI CSI/colour escape sequences from a string. */ function stripAnsi(s: string): string { // Matches CSI sequences (most common: \x1b[...m for colours). @@ -166,13 +193,23 @@ async function runDbtTest(model: string, cwd: string): Promise