diff --git a/packages/server/src/service/materialization_scheduler.spec.ts b/packages/server/src/service/materialization_scheduler.spec.ts index 7460c0f9..feb2b051 100644 --- a/packages/server/src/service/materialization_scheduler.spec.ts +++ b/packages/server/src/service/materialization_scheduler.spec.ts @@ -257,4 +257,33 @@ describe("MaterializationScheduler", () => { await sched.tick(dueLater + 1); expect(service.calls.length).toBe(3); }); + + it("prunes arming state when a package unloads, then re-anchors on reload without a stale fire", async () => { + // A never-fired schedule (fakeService default lastFireAt = null), so a + // fresh arm anchors from `now` and does not catch up. + const service = fakeService(); + const p = fakePackage("p", { schedule: "* * * * *" }); + // makeScheduler closes over this array via getLoadedPackages, so mutating + // it simulates the package leaving and re-entering the loaded set. + const loaded = [p]; + const sched = makeScheduler(loaded, service); + + await sched.tick(t0); // arm: nextFire = t0 + 60s + + // Unload: the package leaves the loaded set, so the sweep prunes its state + // (tick's not-seen cleanup) instead of leaving a stale nextFire behind. + loaded.length = 0; + await sched.tick(dueLater); // past the old nextFire, but the package is gone + expect(service.calls.length).toBe(0); + + // Reload: with the state pruned, arm is fresh — anchored from `now`, so the + // pre-unload nextFire can't resurface as a spurious catch-up. + loaded.push(p); + await sched.tick(dueLater + 1); // now < fresh nextFire -> no fire + expect(service.calls.length).toBe(0); + + // It is genuinely re-armed (not inert): the next occurrence still fires. + await sched.tick(dueLater + 1 + 60_001); + expect(service.calls.length).toBe(1); + }); }); diff --git a/packages/server/src/service/materialization_service.spec.ts b/packages/server/src/service/materialization_service.spec.ts index 80b62490..781108be 100644 --- a/packages/server/src/service/materialization_service.spec.ts +++ b/packages/server/src/service/materialization_service.spec.ts @@ -1,6 +1,6 @@ import type { Connection as MalloyConnection } from "@malloydata/malloy"; import { Manifest } from "@malloydata/malloy"; -import { beforeEach, describe, expect, it } from "bun:test"; +import { afterEach, beforeEach, describe, expect, it } from "bun:test"; import * as sinon from "sinon"; import { BadRequestError, @@ -27,6 +27,11 @@ import { MaterializationService, stagingSuffix, } from "./materialization_service"; +import { resetMaterializationTelemetryForTesting } from "../materialization_metrics"; +import { + startMetricsHarness, + type MetricsHarness, +} from "../test_helpers/metrics_harness"; type MockRepo = sinon.SinonStubbedInstance; @@ -500,6 +505,86 @@ describe("MaterializationService", () => { }); }); +describe("deleteMaterialization telemetry", () => { + let ctx: ReturnType; + let harness: MetricsHarness; + + beforeEach(async () => { + ctx = createMocks(); + harness = await startMetricsHarness(); + // Drop cached instruments so recordDropTables re-binds to this provider. + resetMaterializationTelemetryForTesting(); + }); + + afterEach(async () => { + resetMaterializationTelemetryForTesting(); + await harness.shutdown(); + }); + + const DROP_COUNTER = "publisher_materialization_drop_tables_total"; + + // A terminal materialization whose manifest names one physical table, so + // dropMaterializedTables issues exactly one DROP — the emission under test. + function dropTablesFixture(runSQL: sinon.SinonStub) { + const getMalloyConnection = sinon + .stub() + .resolves({ runSQL, dialectName: "duckdb" }); + (ctx.environmentStore.getEnvironment as sinon.SinonStub).resolves({ + getPackage: sinon.stub().resolves({ getMalloyConnection }), + withPackageLock: async (_n: string, fn: () => Promise) => + fn(), + }); + ctx.repository.getMaterializationById.resolves( + makeMaterialization({ + status: "MANIFEST_FILE_READY", + manifest: { + builtAt: new Date().toISOString(), + strict: false, + entries: { + b1: { + sourceEntityId: "b1", + physicalTableName: "orders_mz", + connectionName: "duckdb", + }, + }, + }, + }), + ); + } + + it("meters a successful physical-table drop as outcome=success", async () => { + dropTablesFixture(sinon.stub().resolves()); + + await ctx.service.deleteMaterialization("my-env", "pkg", "mat-1", { + dropTables: true, + }); + + expect( + await harness.collectCounter(DROP_COUNTER, { outcome: "success" }), + ).toBe(1); + expect( + await harness.collectCounter(DROP_COUNTER, { outcome: "failure" }), + ).toBe(0); + }); + + it("meters a failed drop as outcome=failure and still deletes the record", async () => { + dropTablesFixture(sinon.stub().rejects(new Error("drop boom"))); + + // The drop is best-effort: a failure is metered but never surfaces to the + // caller, and the record is still removed. + await ctx.service.deleteMaterialization("my-env", "pkg", "mat-1", { + dropTables: true, + }); + + expect( + await harness.collectCounter(DROP_COUNTER, { outcome: "failure" }), + ).toBe(1); + expect(ctx.repository.deleteMaterialization.calledOnceWith("mat-1")).toBe( + true, + ); + }); +}); + describe("stagingSuffix", () => { it("derives a short, stable suffix from the sourceEntityId", () => { expect(stagingSuffix("abcdef1234567890")).toBe("_abcdef123456"); @@ -897,6 +982,39 @@ describe("buildOneSource", () => { ); }); + it("rethrows the original build error when staging cleanup also fails", async () => { + const runSQL = sinon.stub(); + runSQL.onCall(0).resolves(); // initial staging drop + runSQL.onCall(1).rejects(new Error("create boom")); // CREATE TABLE AS + runSQL.onCall(2).rejects(new Error("cleanup boom")); // cleanup drop fails + // The cleanup failure is swallowed (logged as a physical leak); the + // original build error is what propagates, so the run is classified by its + // real cause rather than the cleanup noise. + await expect(callBuildOneSource({ runSQL }, "orders_v1")).rejects.toThrow( + "create boom", + ); + expect(runSQL.callCount).toBe(3); + expect(runSQL.lastCall.args[0]).toBe( + 'DROP TABLE IF EXISTS "orders_v1_abcdef123456"', + ); + }); + + it("drops staging when the build is cancelled mid-run (no _staging orphan)", async () => { + // buildOneSource is cancellation-agnostic: a cancel surfaces as a thrown + // build SQL and takes the same staging-cleanup path as a failure, so a + // cancelled build leaves no orphaned _staging table behind. + const runSQL = sinon.stub(); + runSQL.onCall(0).resolves(); // initial staging drop + runSQL.onCall(1).rejects(new Error("Build cancelled")); // aborted mid-CREATE + runSQL.onCall(2).resolves(); // cleanup staging drop + await expect(callBuildOneSource({ runSQL }, "orders_v1")).rejects.toThrow( + "Build cancelled", + ); + expect(runSQL.lastCall.args[0]).toBe( + 'DROP TABLE IF EXISTS "orders_v1_abcdef123456"', + ); + }); + it("quotes each segment of a container path for a backtick dialect", async () => { const runSQL = sinon.stub().resolves(); // Container path (dataset.table) on a backtick dialect (BigQuery): each diff --git a/packages/server/src/storage/duckdb/MaterializationRepository.spec.ts b/packages/server/src/storage/duckdb/MaterializationRepository.spec.ts index 5bd2fb22..005270bc 100644 --- a/packages/server/src/storage/duckdb/MaterializationRepository.spec.ts +++ b/packages/server/src/storage/duckdb/MaterializationRepository.spec.ts @@ -3,16 +3,21 @@ import { DuckDBConnection } from "./DuckDBConnection"; import { MaterializationRepository } from "./MaterializationRepository"; import { initializeSchema } from "./schema"; -// Capture the SQL + params of the last query without touching a real DuckDB. +// Capture the SQL + params of reads (all) and writes (run) without touching a +// real DuckDB. function fakeRepo() { const calls: Array<{ sql: string; params: unknown[] }> = []; + const runCalls: Array<{ sql: string; params: unknown[] }> = []; const db = { all: async (sql: string, params: unknown[]) => { calls.push({ sql, params }); return []; }, + run: async (sql: string, params: unknown[]) => { + runCalls.push({ sql, params }); + }, } as unknown as DuckDBConnection; - return { repo: new MaterializationRepository(db), calls }; + return { repo: new MaterializationRepository(db), calls, runCalls }; } describe("MaterializationRepository list bounds", () => { @@ -152,3 +157,30 @@ describe("MaterializationRepository.getLatestScheduledFireAt (real DuckDB)", () expect(await repo.getLatestScheduledFireAt(ENV_ID, PKG)).toBeNull(); }); }); + +// The publisher tracks materialization *records*; dropping the physical tables +// is opt-in (deleteMaterialization({ dropTables })) and otherwise the caller's +// responsibility. These cascade deletes therefore remove records only — they +// must never issue DDL — so a materialized table intentionally outlives the +// record when an environment or package is deleted. Locking that in guards the +// contract against a future refactor that "helpfully" adds a DROP here (which +// would fire against a possibly-already-torn-down connection on env delete). +describe("MaterializationRepository cascade deletes are records-only", () => { + it("deleteByEnvironmentId issues a single DELETE and no DDL", async () => { + const { repo, runCalls } = fakeRepo(); + await repo.deleteByEnvironmentId("env-1"); + expect(runCalls.length).toBe(1); + expect(runCalls[0].sql).toContain("DELETE FROM materializations"); + expect(runCalls[0].sql).not.toContain("DROP"); + expect(runCalls[0].params).toEqual(["env-1"]); + }); + + it("deleteByPackage issues a single DELETE and no DDL", async () => { + const { repo, runCalls } = fakeRepo(); + await repo.deleteByPackage("env-1", "pkg-a"); + expect(runCalls.length).toBe(1); + expect(runCalls[0].sql).toContain("DELETE FROM materializations"); + expect(runCalls[0].sql).not.toContain("DROP"); + expect(runCalls[0].params).toEqual(["env-1", "pkg-a"]); + }); +}); diff --git a/packages/server/src/storage/duckdb/MaterializationRepository.ts b/packages/server/src/storage/duckdb/MaterializationRepository.ts index 85b8ca57..0a9ffa19 100644 --- a/packages/server/src/storage/duckdb/MaterializationRepository.ts +++ b/packages/server/src/storage/duckdb/MaterializationRepository.ts @@ -263,6 +263,12 @@ export class MaterializationRepository { await this.db.run("DELETE FROM materializations WHERE id = ?", [id]); } + // Cascade deletes remove the publisher's records only; they intentionally + // issue no DDL. Dropping the physical tables is opt-in + // (deleteMaterialization({ dropTables })) and otherwise the caller's + // responsibility, so a materialized table outlives its record when an + // environment or package is deleted. Adding a DROP here would also risk + // firing against a connection already torn down by the delete path. async deleteByEnvironmentId(environmentId: string): Promise { await this.db.run( "DELETE FROM materializations WHERE environment_id = ?", diff --git a/packages/server/tests/integration/materialization/scheduler_transitions.integration.spec.ts b/packages/server/tests/integration/materialization/scheduler_transitions.integration.spec.ts new file mode 100644 index 00000000..2c01ffe3 --- /dev/null +++ b/packages/server/tests/integration/materialization/scheduler_transitions.integration.spec.ts @@ -0,0 +1,256 @@ +/// + +import { afterAll, afterEach, beforeAll, describe, expect, it } from "bun:test"; +import path from "path"; +import { fileURLToPath } from "url"; +import { MaterializationScheduler } from "../../../src/service/materialization_scheduler"; +import { MaterializationService } from "../../../src/service/materialization_service"; +import { environmentStore } from "../../../src/server"; +import { RestE2EEnv, startRestE2E } from "../../harness/rest_e2e"; + +const __filename = fileURLToPath(import.meta.url); +const __dirname = path.dirname(__filename); + +const FIXTURES = path.resolve(__dirname, "../../fixtures"); +const SCHEDULED_FIXTURE = path.join(FIXTURES, "persist-schedule-test"); // cron "0 6 * * *", scope: version +const PERSIST_FIXTURE = path.join(FIXTURES, "persist-test"); // same persist source, no schedule + +// Deterministic clock: tick(nowMs) is called explicitly so the daily +// "0 6 * * *" cron fires without waiting on a wall-clock minute. +const T0 = Date.parse("2026-01-01T00:00:00.000Z"); +const MIN = 60_000; +const H = 60 * MIN; +const D = 24 * H; + +const TERMINAL = ["MANIFEST_FILE_READY", "FAILED", "CANCELLED"]; + +// Integration coverage for the scheduler's state-machine transitions against the +// real store (#895): the unit spec covers the per-tick guards; these drive the +// transitions end to end through the REST-created packages the scheduler sweeps. +// Each test uses its own environment and tears it down in afterEach — the +// scheduler sweeps every loaded environment, so a leftover env would leak fires +// into another test (and steal the per-tick cap in the carryover case). +describe("MaterializationScheduler transitions (integration, real store)", () => { + let e2e: (RestE2EEnv & { stop(): Promise }) | null = null; + let baseUrl: string; + let currentEnv: string | null = null; + + beforeAll(async () => { + e2e = await startRestE2E(); + baseUrl = e2e.baseUrl; + }); + + afterEach(async () => { + if (currentEnv) { + try { + await fetch(`${baseUrl}/api/v0/environments/${currentEnv}`, { + method: "DELETE", + }); + } catch { + // best-effort teardown + } + currentEnv = null; + } + }); + + afterAll(async () => { + await e2e?.stop(); + e2e = null; + }); + + // ── helpers ────────────────────────────────────────────────────────── + + function newScheduler(maxFiresPerTick = 10): MaterializationScheduler { + return new MaterializationScheduler( + environmentStore, + new MaterializationService(environmentStore), + { tickIntervalMs: 60_000, maxFiresPerTick }, + ); + } + + async function createEnv( + name: string, + packages: Array<{ name: string; location: string }>, + ): Promise { + const res = await fetch(`${baseUrl}/api/v0/environments`, { + method: "POST", + headers: { "Content-Type": "application/json" }, + body: JSON.stringify({ name, packages, connections: [] }), + }); + if (!res.ok) { + throw new Error( + `create env ${name} failed (${res.status}): ${await res.text()}`, + ); + } + currentEnv = name; + } + + async function waitForPackage(env: string, pkg: string): Promise { + const deadline = Date.now() + 30_000; + while (Date.now() < deadline) { + const res = await fetch( + `${baseUrl}/api/v0/environments/${env}/packages/${pkg}`, + ); + if (res.ok) return; + await new Promise((r) => setTimeout(r, 500)); + } + throw new Error(`package ${env}/${pkg} did not load in time`); + } + + // Set scope: version + a cron schedule on a package (makes persist-test + // standalone-schedulable). editingPolicy is true, so the publish-gate rules + // run and must pass (version scope + valid cron, no freshness). + async function setSchedule( + env: string, + pkg: string, + schedule: string, + ): Promise { + const res = await fetch( + `${baseUrl}/api/v0/environments/${env}/packages/${pkg}`, + { + method: "PATCH", + headers: { "Content-Type": "application/json" }, + body: JSON.stringify({ + name: pkg, + scope: "version", + materialization: { schedule }, + }), + }, + ); + if (!res.ok) { + throw new Error( + `set schedule on ${pkg} failed (${res.status}): ${await res.text()}`, + ); + } + } + + type Row = Record; + + async function envMaterializations(env: string): Promise { + const res = await fetch( + `${baseUrl}/api/v0/environments/${env}/packages/materializations`, + ); + expect(res.status).toBe(200); + return (await res.json()) as Row[]; + } + + function triggerOf(m: Row): string | undefined { + return (m.metadata as { trigger?: string } | null)?.trigger; + } + + async function scheduledFireCount(env: string): Promise { + return (await envMaterializations(env)).filter( + (m) => triggerOf(m) === "SCHEDULER", + ).length; + } + + // Let every fired build settle so env teardown doesn't race an in-flight one. + async function waitAllTerminal(env: string): Promise { + const deadline = Date.now() + 90_000; + while (Date.now() < deadline) { + const rows = await envMaterializations(env); + if (rows.every((m) => TERMINAL.includes(m.status as string))) return; + await new Promise((r) => setTimeout(r, 250)); + } + throw new Error(`materializations in ${env} did not all settle`); + } + + // ── transitions ───────────────────────────────────────────────────── + + it( + "a cron edit between ticks re-arms from now (no stale fire from the old cadence)", + async () => { + const ENV = "sched-tx-cron-edit"; + await createEnv(ENV, [ + { name: "persist-schedule-test", location: SCHEDULED_FIXTURE }, + ]); + await waitForPackage(ENV, "persist-schedule-test"); + const sched = newScheduler(); + + // Arm on the original "0 6 * * *" cadence: nextFire = Jan-01 06:00. + await sched.tick(T0); + + // Edit the cron to midnight before the old 06:00 occurrence fires. + await setSchedule(ENV, "persist-schedule-test", "0 0 * * *"); + + // Tick just past the OLD 06:00 occurrence. If the edit didn't re-arm, + // the stale nextFire would fire here; instead arm() re-anchors from now + // on the new cron (next = Jan-02 00:00), so nothing fires. + await sched.tick(T0 + 6 * H + MIN); + expect(await scheduledFireCount(ENV)).toBe(0); + + // Past the NEW cron's next occurrence (Jan-02 00:00): it fires, proving + // the edit re-armed rather than silently dropping the schedule. + await sched.tick(T0 + D + MIN); + expect(await scheduledFireCount(ENV)).toBe(1); + + await waitAllTerminal(ENV); + }, + { timeout: 120_000 }, + ); + + // The unload/reload prune + re-anchor transition is covered as a deterministic + // unit test (materialization_scheduler.spec.ts): it is pure in-memory arming + // state, and driving it here via a package delete + same-name re-add races the + // package-install staging rename on Windows (POSIX is fine), which is unrelated + // to the scheduler behavior under test. + + it( + "a fired occurrence advances to the next occurrence (fires once per occurrence, not every tick)", + async () => { + const ENV = "sched-tx-cadence"; + await createEnv(ENV, [ + { name: "persist-schedule-test", location: SCHEDULED_FIXTURE }, + ]); + await waitForPackage(ENV, "persist-schedule-test"); + const sched = newScheduler(); + + await sched.tick(T0); // arm + await sched.tick(T0 + 6 * H + MIN); // due -> fire once + expect(await scheduledFireCount(ENV)).toBe(1); + await waitAllTerminal(ENV); + + // Immediately after, still inside the same daily window: the scheduler + // advanced nextFire to the next occurrence, so it does NOT re-fire every + // tick. (The advance is outcome-independent — a FAILED fire advances the + // same way; that isolation is covered in the scheduler unit spec.) + await sched.tick(T0 + 6 * H + 2 * MIN); + expect(await scheduledFireCount(ENV)).toBe(1); + + // The next day's 06:00 occurrence fires again. + await sched.tick(T0 + D + 6 * H + MIN); + expect(await scheduledFireCount(ENV)).toBe(2); + await waitAllTerminal(ENV); + }, + { timeout: 180_000 }, + ); + + it( + "maxFiresPerTick caps a tick and the capped package fires on the next tick", + async () => { + const ENV = "sched-tx-cap"; + await createEnv(ENV, [ + { name: "persist-schedule-test", location: SCHEDULED_FIXTURE }, + { name: "persist-test", location: PERSIST_FIXTURE }, + ]); + await waitForPackage(ENV, "persist-schedule-test"); + await waitForPackage(ENV, "persist-test"); + // Make the second package standalone-schedulable on the same cadence. + await setSchedule(ENV, "persist-test", "0 6 * * *"); + + const sched = newScheduler(1); // cap = 1 fire per tick + + await sched.tick(T0); // arm both + // Both due at 06:00, but the cap lets only one fire this tick. + await sched.tick(T0 + 6 * H + MIN); + expect(await scheduledFireCount(ENV)).toBe(1); + + // The capped package is still due and fires on the following tick. + await sched.tick(T0 + 6 * H + 2 * MIN); + expect(await scheduledFireCount(ENV)).toBe(2); + + await waitAllTerminal(ENV); + }, + { timeout: 180_000 }, + ); +});