Files
Oskar Otwinowski 8b0385c429 feat(run-engine): trigger tasks pinned to an external deployment id (#4664)
The SDK discovers an external deployment id at runtime (explicit
TRIGGER_EXTERNAL_DEPLOYMENT_ID always; platform commit-SHA variables and
generic fallbacks when TRIGGER_AUTOMATIC_SKEW_VERSION_PROTECTION=1) and
sends it alongside lockToVersion; the server resolves precedence
(version > external id > current). An id held by a deployed deployment
pins the run to that worker; an in-flight or unknown id parks the run in
PENDING_VERSION with the id in TaskRun.annotations, wakes it pinned when
a deployment carrying the id finalizes (ClickHouse candidates, Postgres
authoritative), and expires it after a deadline that re-checks Postgres
before acting. Parking outranks delaying and preserves delayUntil. The
id is projected to ClickHouse task_runs_v2.external_deployment_id during
replication. Redis cache for id-to-worker resolution, guarded
version-aware writes.

Ids are not unique. Several deployments can hold one id - a --force
rebuild is the ordinary way to get there - so resolution always picks
the highest version among the candidates, never the newest by timestamp.
The rule is applied identically on both paths that can bind a run to a
worker: resolveExternalDeployment at trigger time, and
PendingVersionSystem when a landing deployment wakes a parked run.
Version comparison is numeric on the counter half, so 20260807.10
outranks 20260807.9.

A run whose id never lands expires at the deadline with
EXTERNAL_DEPLOYMENT_NOT_FOUND and an error naming the id it waited for,
which is what a failed build or a typo looks like from the caller.
Default deadline is one hour (EXTERNAL_DEPLOYMENT_PARK_DEADLINE_MS).

Debounce registration happens in both the parked and the delayed branch
through one helper, so a debounced run that parks still binds its
debounce key; without it every later trigger for the same key created
another parked run, and all of them executed when the deployment landed.
The two DELAYED-only status checks in DebounceSystem also accept
PENDING_VERSION, without which the lock-contention fallback would
rethrow a 5xx the SDK retries and amplifies, and the fast path would
push every trigger on a parked key through the redlock.

Resolution is skipped in development. A dev environment cannot hold a
WorkerDeployment - trigger dev registers a BackgroundWorker with nothing
behind it, and deploy --env refuses dev - so an external deployment id
there could only ever park, and the parked run then expired against the
dev TTL while a connected dev worker sat idle. The id is still annotated
so the dashboard shows what the app sent (TRI-13000).
2026-08-19 17:43:53 +02:00

189 lines
6.6 KiB
TypeScript

import { redisTest } from "@internal/testcontainers";
import { Redis } from "ioredis";
import { describe, expect } from "vitest";
import { RedisExternalDeploymentCache } from "~/services/externalDeploymentCache.server";
vi.setConfig({ testTimeout: 30_000 });
function entry(version: string, workerId = `worker_${version}`) {
return { workerId, version, sdkVersion: "4.0.0", cliVersion: "4.0.0" };
}
function deployed(version: string, workerId = `worker_${version}`) {
return { outcome: "deployed", entry: entry(version, workerId) };
}
describe("RedisExternalDeploymentCache", () => {
redisTest("stores and reads back a resolution", async ({ redisOptions }) => {
const redis = new Redis({ ...redisOptions, maxRetriesPerRequest: null });
const cache = new RedisExternalDeploymentCache({ redis });
try {
expect(await cache.get("env_1", "commit-a")).toBeNull();
await cache.setIfNewer("env_1", "commit-a", entry("20260807.1"));
expect(await cache.get("env_1", "commit-a")).toEqual(deployed("20260807.1"));
} finally {
await redis.quit();
}
});
redisTest("scopes entries by environment", async ({ redisOptions }) => {
const redis = new Redis({ ...redisOptions, maxRetriesPerRequest: null });
const cache = new RedisExternalDeploymentCache({ redis });
try {
await cache.setIfNewer("env_1", "commit-a", entry("20260807.1"));
expect(await cache.get("env_2", "commit-a")).toBeNull();
} finally {
await redis.quit();
}
});
redisTest("overwrites when the finalising version is higher", async ({ redisOptions }) => {
const redis = new Redis({ ...redisOptions, maxRetriesPerRequest: null });
const cache = new RedisExternalDeploymentCache({ redis });
try {
await cache.setIfNewer("env_1", "commit-a", entry("20260807.1"));
await cache.setIfNewer("env_1", "commit-a", entry("20260807.2"));
expect(await cache.get("env_1", "commit-a")).toEqual(deployed("20260807.2"));
} finally {
await redis.quit();
}
});
redisTest(
"refuses a lower version — the slower earlier build finalising last must not reinstate the version the operator was replacing",
async ({ redisOptions }) => {
const redis = new Redis({ ...redisOptions, maxRetriesPerRequest: null });
const cache = new RedisExternalDeploymentCache({ redis });
try {
await cache.setIfNewer("env_1", "commit-a", entry("20260807.2"));
await cache.setIfNewer("env_1", "commit-a", entry("20260807.1"));
expect(await cache.get("env_1", "commit-a")).toEqual(deployed("20260807.2"));
} finally {
await redis.quit();
}
}
);
redisTest("refuses an equal version", async ({ redisOptions }) => {
const redis = new Redis({ ...redisOptions, maxRetriesPerRequest: null });
const cache = new RedisExternalDeploymentCache({ redis });
try {
await cache.setIfNewer("env_1", "commit-a", entry("20260807.1", "worker_first"));
await cache.setIfNewer("env_1", "commit-a", entry("20260807.1", "worker_second"));
expect(await cache.get("env_1", "commit-a")).toEqual(deployed("20260807.1", "worker_first"));
} finally {
await redis.quit();
}
});
redisTest(
"compares the counter numerically, not lexicographically — .10 beats .9",
async ({ redisOptions }) => {
const redis = new Redis({ ...redisOptions, maxRetriesPerRequest: null });
const cache = new RedisExternalDeploymentCache({ redis });
try {
await cache.setIfNewer("env_1", "commit-a", entry("20260807.9"));
await cache.setIfNewer("env_1", "commit-a", entry("20260807.10"));
expect(await cache.get("env_1", "commit-a")).toEqual(deployed("20260807.10"));
await cache.setIfNewer("env_1", "commit-a", entry("20260807.9"));
expect(await cache.get("env_1", "commit-a")).toEqual(deployed("20260807.10"));
} finally {
await redis.quit();
}
}
);
redisTest("a later date wins regardless of counter", async ({ redisOptions }) => {
const redis = new Redis({ ...redisOptions, maxRetriesPerRequest: null });
const cache = new RedisExternalDeploymentCache({ redis });
try {
await cache.setIfNewer("env_1", "commit-a", entry("20260807.50"));
await cache.setIfNewer("env_1", "commit-a", entry("20260808.1"));
expect(await cache.get("env_1", "commit-a")).toEqual(deployed("20260808.1"));
await cache.setIfNewer("env_1", "commit-a", entry("20260806.99"));
expect(await cache.get("env_1", "commit-a")).toEqual(deployed("20260808.1"));
} finally {
await redis.quit();
}
});
redisTest("caches a missing resolution with a short TTL", async ({ redisOptions }) => {
const redis = new Redis({ ...redisOptions, maxRetriesPerRequest: null });
const cache = new RedisExternalDeploymentCache({ redis, missingTtlSeconds: 30 });
try {
await cache.setMissing("env_1", "commit-a");
expect(await cache.get("env_1", "commit-a")).toEqual({ outcome: "missing" });
const ttl = await redis.ttl("skewid:env_1:commit-a");
expect(ttl).toBeGreaterThan(0);
expect(ttl).toBeLessThanOrEqual(30);
} finally {
await redis.quit();
}
});
redisTest("a landed deployment replaces a missing marker", async ({ redisOptions }) => {
const redis = new Redis({ ...redisOptions, maxRetriesPerRequest: null });
const cache = new RedisExternalDeploymentCache({ redis });
try {
await cache.setMissing("env_1", "commit-a");
await cache.setIfNewer("env_1", "commit-a", entry("20260807.1"));
expect(await cache.get("env_1", "commit-a")).toEqual(deployed("20260807.1"));
} finally {
await redis.quit();
}
});
redisTest("a missing marker never clobbers a resolution", async ({ redisOptions }) => {
const redis = new Redis({ ...redisOptions, maxRetriesPerRequest: null });
const cache = new RedisExternalDeploymentCache({ redis });
try {
await cache.setIfNewer("env_1", "commit-a", entry("20260807.1"));
await cache.setMissing("env_1", "commit-a");
expect(await cache.get("env_1", "commit-a")).toEqual(deployed("20260807.1"));
} finally {
await redis.quit();
}
});
redisTest("sets a TTL on the entry", async ({ redisOptions }) => {
const redis = new Redis({ ...redisOptions, maxRetriesPerRequest: null });
const cache = new RedisExternalDeploymentCache({ redis, ttlSeconds: 120 });
try {
await cache.setIfNewer("env_1", "commit-a", entry("20260807.1"));
const ttl = await redis.ttl("skewid:env_1:commit-a");
expect(ttl).toBeGreaterThan(0);
expect(ttl).toBeLessThanOrEqual(120);
} finally {
await redis.quit();
}
});
});