test(materialization): GC + scheduler state-machine transition coverage

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);
+   });
 });

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<ResourceRepository>;
 
@@ -500,6 +505,86 @@ describe("MaterializationService", () => {
    });
 });
 
+describe("deleteMaterialization telemetry", () => {
+   let ctx: ReturnType<typeof createMocks>;
+   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<unknown>) =>
+            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

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"]);
+   });
+});

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<void> {
       await this.db.run(
          "DELETE FROM materializations WHERE environment_id = ?",