testBackgroundJobs

tcmts.cli.test.e2e.jobs.testBackgroundJobs · #u7tn7lmg · term
background jobs: durable, retried, pinned to the code that enqueued them; schedules run once per slot
import { okIn, plainEnv, tcmBin, tempDir, withChildren } from "#codebase/tcmts/cli/test/e2e/harness";
import { call, SHOP } from "#codebase/tcmts/cli/test/e2e/jobs";
import assert from "node:assert/strict";
import { spawn } from "node:child_process";
import fs from "node:fs";
import path from "node:path";
import { DatabaseSync } from "node:sqlite";

export const testBackgroundJobs = () =>
  withChildren(async (keep) => {
    const dir = tempDir("jobs");
    const outbox = path.join(dir, "outbox.log");
    const tcm = (...args: string[]) => okIn(dir, ...args);
    const scratch = (text: string) => fs.writeFileSync(path.join(dir, "scratch.ts"), text);
    const lines = () =>
      fs.existsSync(outbox) ? fs.readFileSync(outbox, "utf8").trim().split("\n") : [];
    const waitFor = async (cond: () => boolean | Promise<boolean>, ms = 20000) => {
      for (
        const end = Date.now() + ms;
        Date.now() < end;
        await new Promise((r) => setTimeout(r, 100))
      )
        if (await cond()) return;
      throw new Error(`timed out; outbox:\n${lines().join("\n")}`);
    };

    tcm("init");
    tcm("lib", "add", "tcm", "server", "--at", "lib.server");
    fs.rmSync(path.join(dir, "scratch.ts"));
    scratch(SHOP(outbox, "v1"));
    tcm("add");
    assert.match(
      tcm("test", "shop"),
      /1 passed, 0 failed/,
      "runJobs() drains the queue inside tcm test",
    );

    const child = keep(
      spawn(process.execPath, [tcmBin(), "start", "shop.main", "--port", "0"], {
        cwd: dir,
        env: plainEnv(),
      }),
    );
    let out = "";
    child.stdout.setEncoding("utf8").on("data", (d: string) => (out += d));
    child.stderr.setEncoding("utf8").on("data", (d: string) => (out += d));
    await waitFor(() => /url\s+http:\/\/\S+/.test(out));
    const url = /url\s+(http:\/\/\S+)/.exec(out)![1]!;
    // printed once the worker is up, a moment after the url: wait for it rather than racing it
    await waitFor(() => /jobs\s+recurring: shop\.heartbeat \(1s\)/.test(out));

    // a server function returns at once; the work happens in the background
    assert.match((await call(dir, url, "shop.signup", ["ada"])).body, /^"[\w-]{16}"$/);
    await waitFor(() => lines().includes("welcome ada (v1)"));

    // failures are retried with backoff
    await call(dir, url, "shop.startFlaky", []);
    await waitFor(() => lines().includes("flaky x ok after 2"));

    // a job runs the code version that enqueued it, even if a deploy happens before it runs
    await call(dir, url, "shop.signupLater", ["grace"]);
    scratch(SHOP(outbox, "v2"));
    tcm("update");
    await waitFor(() => lines().includes("welcome grace (v1)"));
    // the new signup is callable once the server has rebuilt for the update
    await waitFor(async () => (await call(dir, url, "shop.signup", ["linus"])).status === 200);
    await waitFor(() => lines().includes("welcome linus (v2)"));

    // jobs name their function by hash, so an inline function can't be enqueued
    const inline = await call(dir, url, "shop.enqueueInline", []);
    assert.equal(inline.status, 500);
    assert.match(out, /enqueue\(\) needs a codebase function/);

    // recurring: one run per slot
    await waitFor(() => lines().filter((l) => l === "tick").length >= 2);
    const db = new DatabaseSync(path.join(dir, ".tcm", "data", "jobs.sqlite"), { readOnly: true });
    const dupes = db
      .prepare(
        "SELECT count(*) AS n FROM (SELECT slot FROM jobs WHERE slot IS NOT NULL GROUP BY slot HAVING count(*) > 1)",
      )
      .get() as { n: number };
    db.close();
    assert.equal(dupes.n, 0);

    // tcm jobs: the queue, and acting on it
    const listing = tcm("jobs", "50");
    assert.match(listing, /done\s+flaky\s+2\/5/);
    assert.match(listing, /done\s+shop\.heartbeat/);
    const id = /^(\S{8})\s+done\s+sendWelcome/m.exec(listing)![1]!;
    assert.match(tcm("jobs", "retry", id), /queued again/);
    await waitFor(() => lines().filter((l) => l.startsWith("welcome")).length >= 4, 15000);
  })
testBackgroundJobs

uses
