Skip to content

Commit 9075d02

Browse files
committed
feat(server): compute scheduler shadow mode, and real-Postgres compute test coverage
POST /compute/requests?dryRun=true computes a real placement decision (same scheduler code path) without ever calling commitPlacement — no lease, no consumed capacity/quota, and the underlying request is cancelled immediately rather than left queued. Closes the one concrete gap blocking canary rollout (docs/COMPUTE_CONTROL_PLANE_ROLLOUT.md phase 3). Verified end-to-end, including that a real submission right after a shadow one sees fully untouched capacity. Also adds postgres-compute-control-store.test.ts — until now the compute control plane, unlike every other Postgres-backed store, had no real- database test coverage at all. Covers RLS tenant isolation, real concurrent commitPlacement calls racing the same pg_advisory_xact_lock, fencing-token rejection, the two SECURITY DEFINER maintenance functions, and shadow-decision persistence — gated on DATABASE_URL like every other postgres-*.test.ts file (CI already runs these as mandatory, not skipped; see the Test server CI step).
1 parent 01b5a14 commit 9075d02

9 files changed

Lines changed: 377 additions & 15 deletions

‎docs/COMPUTE_CONTROL_PLANE_ROLLOUT.md‎

Lines changed: 3 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -12,16 +12,11 @@ This sequences docs/COMPUTE_CONTROL_PLANE.md's own mechanism — contracts, sche
1212

1313
**Before phase 3**: enroll a small number of real nodes (a handful of workstations plus, ideally, one dedicated inference server) in one pool, and confirm heartbeats/assignments actually flow — `GET /compute/summary` should show them online, and `admin-console`'s Compute page should render them with live CPU/RAM/device telemetry.
1414

15-
## Phase 3 — Shadow mode: a real, currently-missing feature, not just an ops step
15+
## Phase 3 — Shadow mode (done)
1616

17-
The original design calls for running the scheduler in shadow mode — computing placement decisions without acting on them, to compare against whatever admission already happens today (locally, each node's own `ResourceOrchestrator` continues to decide for itself). **This does not exist in the code today.** `POST /compute/requests` always calls `ComputeControlPlane.submit()`, which schedules *and* commits a real lease (`store.commitPlacement()`) in the same call — there is no `dryRun`/observe-only flag anywhere in `submitRequestBody`, `ComputeScheduler.schedule()`, or `ComputeControlPlane`.
17+
`POST /compute/requests?dryRun=true` computes a real scheduler decision (`ComputeScheduler.schedule()` against the real current snapshot — same code path a committing request uses) and records it as a `compute_allocation_events` row (`event_type: "shadow_decision"`) plus a `computeRequest.shadowDecision` audit entry, but never calls `commitPlacement()` — no lease is created, no capacity or quota is consumed, and the underlying request is cancelled immediately rather than left sitting in the real queue (necessary since `compute_allocation_events.request_id` is a hard `NOT NULL` foreign key — a shadow decision has to be logged against *some* real, if immediately-cancelled, request row). Verified end-to-end, including that a real submission made right after a shadow one sees completely untouched capacity, and with a real-Postgres test proving the same holds under the actual database (`postgres-compute-control-store.test.ts`, gated on `DATABASE_URL` like every other Postgres-backed test in this repo).
1818

19-
Shipping shadow mode for real needs:
20-
- A `dryRun?: boolean` field on the submit-request body and `ComputeControlPlane.submit()`'s `options`, short-circuiting before `store.commitPlacement()` and returning the scheduler's decision (`SchedulerDecision`) without ever creating a lease or consuming quota/capacity.
21-
- Somewhere to record dry-run decisions for later comparison against what actually happened (the simplest correct option: log them as `compute_allocation_events` rows tagged `event_type: "shadow_decision"`, reusing the existing append-only audit table rather than a new one).
22-
- A comparison report (even a manual one, cross-referencing shadow decisions against the timing/placement of whatever local admission actually did) before trusting the scheduler's judgment on real capacity.
23-
24-
Until this is built, **phase 4 cannot be entered safely** — there is no way to observe the scheduler's decisions without it immediately committing real leases against real capacity. This is the one concrete blocking gap in the whole rollout sequence; everything from phase 4 onward assumes it's closed first.
19+
What's still manual: actually *comparing* shadow decisions against whatever local admission does in parallel is a reporting exercise, not more code — query `compute_allocation_events` for `event_type = 'shadow_decision'` and cross-reference against real placements over the same window before trusting the scheduler's judgment on real capacity.
2520

2621
## Phase 4 — Canary pools, no preemption or borrowing
2722

Lines changed: 111 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,111 @@
1+
#!/usr/bin/env node
2+
// Admin tooling: signs a compute-control-plane resource policy for
3+
// POST /organizations/:organizationId/compute/policies (see
4+
// server/src/compute/policy-signature.ts and
5+
// server/src/routes/compute-control.ts). This is a *different* signed
6+
// policy from app/scripts/sign-policy.js's desktop central-policy (a local
7+
// AppSettings lockdown) — this one governs the compute fleet's hard CPU/GPU
8+
// guardrails, and its verifier lives server-side, not in the desktop app.
9+
//
10+
// Usage:
11+
// node server/scripts/sign-compute-policy.js <draft.json> <private-key.pem> <organization-id> [output.json]
12+
//
13+
// <draft.json> shape (matches ResourcePolicyInput minus signature/issuedAt/
14+
// expiresAt, which this script fills in):
15+
// {
16+
// "name": "Interactive pool guardrails",
17+
// "poolId": "3f2a...-uuid (optional — omit for an org-wide policy)",
18+
// "expiresInDays": 30,
19+
// "hardLimits": { "maxCpuThreads": 32, "maxAccelerators": 2 },
20+
// "workloadClassLimits": {
21+
// "background": { "maxCpuThreads": 8, "allowCpuFallback": true }
22+
// }
23+
// }
24+
//
25+
// The private key must be an Ed25519 PEM (see
26+
// app/scripts/generate-policy-keypair.js's --curve ed25519 mode, or any
27+
// `openssl genpkey -algorithm ed25519` key) — the server's
28+
// createComputePolicySignatureVerifier() only ever verifies Ed25519.
29+
//
30+
// Output is the exact JSON body POST /compute/policies expects — pipe it
31+
// straight into that request (e.g. via curl -d @output.json), or paste it
32+
// into the admin console's "paste signed policy" step.
33+
34+
import * as crypto from "node:crypto";
35+
import * as fs from "node:fs";
36+
37+
const [, , draftPath, privateKeyPath, organizationId, outputPathArg] = process.argv;
38+
if (!draftPath || !privateKeyPath || !organizationId) {
39+
console.error("Usage: node sign-compute-policy.js <draft.json> <private-key.pem> <organization-id> [output.json]");
40+
process.exit(1);
41+
}
42+
43+
const draft = JSON.parse(fs.readFileSync(draftPath, "utf-8"));
44+
if (!draft.name || typeof draft.name !== "string") {
45+
console.error('draft.json must include a non-empty string "name"');
46+
process.exit(1);
47+
}
48+
if (!Number.isFinite(draft.expiresInDays) || draft.expiresInDays <= 0) {
49+
console.error('draft.json must include a positive number "expiresInDays"');
50+
process.exit(1);
51+
}
52+
if (!draft.hardLimits || typeof draft.hardLimits !== "object") {
53+
console.error('draft.json must include a "hardLimits" object (may be empty: {})');
54+
process.exit(1);
55+
}
56+
57+
const now = new Date();
58+
const expiresAt = new Date(now.getTime() + draft.expiresInDays * 24 * 60 * 60 * 1000);
59+
60+
// The exact unsigned shape createComputePolicySignatureVerifier() re-derives
61+
// server-side and compares the signature against — every field here must
62+
// match resourcePolicyInputSchema (minus signature) or the server's own
63+
// recomputed canonical payload will differ and verification will fail.
64+
const unsigned = {
65+
...(draft.poolId ? { poolId: draft.poolId } : {}),
66+
name: draft.name,
67+
hardLimits: draft.hardLimits,
68+
workloadClassLimits: draft.workloadClassLimits ?? {},
69+
issuedAt: now.toISOString(),
70+
expiresAt: expiresAt.toISOString(),
71+
};
72+
73+
// Mirrors server/src/compute/policy-signature.ts's canonicalize()/
74+
// canonicalComputePolicyPayload() exactly — this script is standalone (not
75+
// part of the server's TS build) so it can't import that function directly.
76+
// If you change one, change both; policy-signature.test.ts's round-trip
77+
// tests are what would catch drift on the server side.
78+
function canonicalize(value) {
79+
if (Array.isArray(value)) return value.map(canonicalize);
80+
if (value && typeof value === "object") {
81+
const sorted = {};
82+
for (const key of Object.keys(value).sort()) sorted[key] = canonicalize(value[key]);
83+
return sorted;
84+
}
85+
return value;
86+
}
87+
function canonicalComputePolicyPayload(orgId, input) {
88+
return JSON.stringify(canonicalize({ organizationId: orgId, ...input }));
89+
}
90+
91+
const payloadString = canonicalComputePolicyPayload(organizationId, unsigned);
92+
const privateKey = crypto.createPrivateKey(fs.readFileSync(privateKeyPath, "utf-8"));
93+
// Unlike sign-policy.js's desktop central-policy (hex signature), the
94+
// compute-policy verifier expects base64 — see
95+
// createComputePolicySignatureVerifier()'s `Buffer.from(input.signature,
96+
// "base64")`. Getting this encoding wrong is the single easiest way to
97+
// produce a policy that silently fails verification (400
98+
// invalid_compute_policy_signature) despite a technically-correct signature.
99+
const signature = crypto.sign(null, Buffer.from(payloadString, "utf-8"), privateKey);
100+
101+
const signed = { ...unsigned, signature: signature.toString("base64") };
102+
103+
const outputPath = outputPathArg ?? "compute-policy.json";
104+
fs.writeFileSync(outputPath, JSON.stringify(signed, null, 2));
105+
106+
console.log(`Wrote ${outputPath}`);
107+
console.log(` name: ${signed.name}`);
108+
console.log(` poolId: ${signed.poolId ?? "(org-wide)"}`);
109+
console.log(` issuedAt: ${signed.issuedAt}`);
110+
console.log(` expiresAt: ${signed.expiresAt}`);
111+
console.log(`\nPOST this file's contents to /organizations/${organizationId}/compute/policies to create it.`);

‎server/src/compute/control-plane.ts‎

Lines changed: 18 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -15,12 +15,29 @@ export interface SubmitResult {
1515
export class ComputeControlPlane {
1616
constructor(private readonly store: ComputeControlStore, private readonly scheduler = new ComputeScheduler(), private readonly now: () => Date = () => new Date()) {}
1717

18-
async submit(organizationId: string, input: SubmitComputeRequestInput, actor: AuditActor, options: { allowSafePreemption?: boolean } = {}): Promise<SubmitResult> {
18+
async submit(organizationId: string, input: SubmitComputeRequestInput, actor: AuditActor, options: { allowSafePreemption?: boolean; dryRun?: boolean } = {}): Promise<SubmitResult> {
1919
const request = await this.store.submitRequest(organizationId, input, actor);
20+
if (options.dryRun) return this.submitDryRun(organizationId, request, actor, options);
2021
const scheduled = await this.scheduleRequest(request, actor, options);
2122
return { request: (await this.store.getRequest(organizationId, request.id)) ?? request, ...scheduled };
2223
}
2324

25+
/** Computes and records what the scheduler *would* decide, without ever
26+
* calling commitPlacement — no lease is created, no capacity is
27+
* consumed, and no other request's admission is affected. The
28+
* underlying request row still has to exist momentarily (see
29+
* recordShadowDecision's own doc comment on the store interface) but is
30+
* cancelled immediately after, so it never lingers in the real queue. */
31+
private async submitDryRun(organizationId: string, request: ComputeResourceRequest, actor: AuditActor, options: { allowSafePreemption?: boolean }): Promise<SubmitResult> {
32+
const snapshot = await this.store.getSchedulingSnapshot(organizationId, request.poolId, this.now().toISOString());
33+
const decision: SchedulerDecision = snapshot
34+
? this.scheduler.schedule(request, snapshot, options)
35+
: { status: "rejected", reasons: ["The selected compute pool no longer exists."] };
36+
await this.store.recordShadowDecision(organizationId, request.id, decision, actor);
37+
const cancelled = await this.store.cancelRequest(organizationId, request.id, actor);
38+
return { request: cancelled ?? request, decision };
39+
}
40+
2441
async scheduleRequest(request: ComputeResourceRequest, actor: AuditActor, options: { allowSafePreemption?: boolean } = {}): Promise<{ decision: SchedulerDecision; lease?: ComputeResourceLease }> {
2542
const snapshot = await this.store.getSchedulingSnapshot(request.organizationId, request.poolId, this.now().toISOString());
2643
if (!snapshot) return { decision: { status: "rejected", reasons: ["The selected compute pool no longer exists."] } };

‎server/src/routes/compute-control.integration.test.ts‎

Lines changed: 24 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -88,6 +88,30 @@ describe("compute control routes", () => {
8888
expect(after.json()).toMatchObject({ poolId, weight: 3, reservedCpuThreads: 4, burstAccelerators: 2 });
8989
});
9090

91+
it("shadow mode (dryRun) computes a real placement decision without creating a lease or leaving the request queued", async () => {
92+
const { poolId } = await enrollAndPool();
93+
const body = {
94+
poolId, workloadKind: "active-inference", priority: "interactive", profile: "interactive", checkpointable: false, restartable: false,
95+
requirements: { cpuThreads: 4, ramMB: 4_096, pinnedMemoryMB: 0, acceleratorCount: 1, acceleratorDeviceIds: [], acceleratorVendor: "nvidia", vramMBPerDevice: 4_096, sameNumaNode: false, sameVendor: true, exclusiveAccelerators: true, runtime: "llamacpp", modelId: "model-4b", allowCpuFallback: false },
96+
};
97+
98+
const dryRun = await app.inject({ method: "POST", url: `/organizations/${organizationId}/compute/requests?dryRun=true`, headers: { authorization: `Bearer ${token}` }, payload: body });
99+
expect(dryRun.statusCode, dryRun.body).toBe(200);
100+
expect(dryRun.json().decision.status).toBe("placed");
101+
expect(dryRun.json().lease).toBeUndefined();
102+
expect(dryRun.json().request.state).toBe("cancelled");
103+
104+
// A real (non-dry-run) request right after must still see full
105+
// capacity — the shadow decision consumed nothing.
106+
const real = await app.inject({ method: "POST", url: `/organizations/${organizationId}/compute/requests`, headers: { authorization: `Bearer ${token}` }, payload: body });
107+
expect(real.statusCode, real.body).toBe(201);
108+
expect(real.json().decision.status).toBe("placed");
109+
expect(real.json().lease).toBeDefined();
110+
111+
const leases = await app.inject({ method: "GET", url: `/organizations/${organizationId}/compute/leases`, headers: { authorization: `Bearer ${token}` } });
112+
expect(leases.json()).toHaveLength(1);
113+
});
114+
91115
it("does not expose another organization's inventory", async () => {
92116
await enrollAndPool();
93117
const otherToken = await new SignJWT({ sub: "idp|other" }).setProtectedHeader({ alg: "RS256", kid: KID }).setIssuedAt().setIssuer(ISSUER).setAudience(AUDIENCE).setExpirationTime("1h").sign(key);

‎server/src/routes/compute-control.ts‎

Lines changed: 10 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -152,12 +152,18 @@ export function registerComputeControlRoutes(fastify: FastifyInstance, deps: Rou
152152
const { organizationId } = organizationParams.parse(request.params);
153153
const caller = await requireOrgUser(deps, request, organizationId);
154154
await requirePermission(deps.store, caller, "compute:submit", `organization:${organizationId}/computeRequest:*`);
155+
// Shadow mode (docs/COMPUTE_CONTROL_PLANE_ROLLOUT.md phase 3): a
156+
// query flag, not a body field, since it shapes how this call is
157+
// processed rather than being a property of the request itself —
158+
// matches poolId/state below being query params, not body fields,
159+
// for the same reason.
160+
const dryRun = (request.query as { dryRun?: string }).dryRun === "true";
155161
const elapsed = startTimer();
156-
const result = await deps.computeControlPlane.submit(organizationId, submitRequestBody.parse(request.body), actorFrom(caller), { allowSafePreemption: true });
162+
const result = await deps.computeControlPlane.submit(organizationId, submitRequestBody.parse(request.body), actorFrom(caller), { allowSafePreemption: true, dryRun });
157163
const decisionStatus = result.decision.status;
158-
computeSchedulingDecisionDuration.observe({ result: decisionStatus }, elapsed());
159-
computeSchedulingDecisions.inc({ result: decisionStatus });
160-
reply.code(201).send(result);
164+
computeSchedulingDecisionDuration.observe({ result: dryRun ? `shadow-${decisionStatus}` : decisionStatus }, elapsed());
165+
computeSchedulingDecisions.inc({ result: dryRun ? `shadow-${decisionStatus}` : decisionStatus });
166+
reply.code(dryRun ? 200 : 201).send(result);
161167
});
162168

163169
fastify.get("/organizations/:organizationId/compute/requests", { preHandler: deps.authPreHandler }, async (request, reply) => {

‎server/src/store/compute-control-store.test.ts‎

Lines changed: 21 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -42,4 +42,25 @@ describe("InMemoryComputeControlStore", () => {
4242
expect(await store.sweepExpired(now.toISOString())).toEqual([result.lease!.id]);
4343
expect((await store.getRequest(ORG, result.request.id))?.state).toBe("queued");
4444
});
45+
46+
it("dryRun computes a real decision, consumes no capacity, and leaves the request cancelled, not queued", async () => {
47+
const now = new Date("2026-08-30T12:00:00.000Z");
48+
const store = new InMemoryComputeControlStore(undefined, () => now);
49+
await store.registerNode(ORG, { id: NODE, name: "cpu", region: "eu-tr", labels: {}, operatingSystem: "linux", architecture: "x64", agentVersion: "1", certificateFingerprint: "fp", cpuThreads: 8, freeCpuThreads: 7, totalRamMB: 16_000, freeRamMB: 14_000, numaNodes: 1, supportedRuntimes: ["llamacpp"], warmModelIds: [], inventoryVersion: "1", devices: [] }, actor);
50+
const pool = await store.createPool(ORG, { name: "pool", region: "eu-tr", labels: {}, nodeIds: [NODE], status: "active", schedulingPolicy: "interactive-first" }, actor);
51+
const control = new ComputeControlPlane(store, undefined, () => now);
52+
const body = { poolId: pool.id, workloadKind: "cpu", priority: "background" as const, profile: "balanced" as const, requirements: { cpuThreads: 4, ramMB: 4_000, pinnedMemoryMB: 0, acceleratorCount: 0, acceleratorDeviceIds: [], vramMBPerDevice: 0, sameNumaNode: false, sameVendor: true, exclusiveAccelerators: false, runtime: "llamacpp" as const, allowCpuFallback: false }, checkpointable: false, restartable: false };
53+
54+
const shadow = await control.submit(ORG, body, actor, { dryRun: true });
55+
expect(shadow.decision.status).toBe("placed");
56+
expect(shadow.lease).toBeUndefined();
57+
expect(shadow.request.state).toBe("cancelled");
58+
expect(await store.listLeases(ORG)).toHaveLength(0);
59+
60+
// A real submission right after must see completely untouched
61+
// capacity — the shadow decision must never have reserved anything.
62+
const real = await control.submit(ORG, body, actor);
63+
expect(real.decision.status).toBe("placed");
64+
expect(real.lease).toBeDefined();
65+
});
4566
});

0 commit comments

Comments
 (0)