From 19e90278e759a36bdf6d95a9491cc3effb72936f Mon Sep 17 00:00:00 2001 From: Adam Shiervani Date: Sat, 19 Sep 2026 15:04:01 +0200 Subject: [PATCH] feat(releases): POST /releases/sync for the upload script The scheduled tick found a new version at most 30 minutes plus the settle window after upload. The upload script knows when its last object is written, so it can trigger the sync itself. One runner now owns the in-progress flag and serves both the timer and the endpoint. The endpoint is guarded by RELEASE_SYNC_TOKEN as a bearer token, compared in constant time, and is not registered without it. It runs with the settle window off and answers with the per-outcome counts, or 409 while a run is in progress. --- .env.example | 4 +++ src/auth.ts | 15 +++++++++++ src/errors.ts | 8 ++++++ src/index.ts | 18 ++++++++++--- src/release-sync.ts | 54 ++++++++++++++++++++++++++------------ src/releases.ts | 19 +++++++++++++- test/auth.test.ts | 31 ++++++++++++++++++++++ test/sync-releases.test.ts | 53 ++++++++++++++++++++++++++++++------- 8 files changed, 171 insertions(+), 31 deletions(-) create mode 100644 test/auth.test.ts diff --git a/.env.example b/.env.example index a7a246a..0ad2520 100644 --- a/.env.example +++ b/.env.example @@ -40,3 +40,7 @@ REAL_IP_HEADER= # ICE Servers for WebRTC, split by comma (e.g. stun:stun.l.google.com:19302,stun:stun1.l.google.com:19302) ICE_SERVERS= + +# Bearer token for POST /releases/sync (the upload script calls it after its last write). +# Leave empty to disable the endpoint. +RELEASE_SYNC_TOKEN= diff --git a/src/auth.ts b/src/auth.ts index 8985196..5856cca 100644 --- a/src/auth.ts +++ b/src/auth.ts @@ -1,3 +1,4 @@ +import { createHash, timingSafeEqual } from "crypto"; import { type NextFunction, type Request, type Response } from "express"; import * as jose from "jose"; import { UnauthorizedError } from "./errors"; @@ -56,3 +57,17 @@ export const authenticated = async (req: Request, res: Response, next: NextFunct next(); }; + +const sha256 = (value: string) => createHash("sha256").update(value).digest(); + +/** Guards a route with one static bearer token, compared in constant time. */ +export const bearerToken = (expected: string) => { + const expectedDigest = sha256(expected); + return (req: Request, res: Response, next: NextFunction) => { + const presented = req.headers.authorization?.match(/^Bearer (.+)$/)?.[1]; + if (!presented || !timingSafeEqual(sha256(presented), expectedDigest)) { + throw new UnauthorizedError("Invalid bearer token"); + } + next(); + }; +}; diff --git a/src/errors.ts b/src/errors.ts index 07f868b..936747d 100644 --- a/src/errors.ts +++ b/src/errors.ts @@ -31,6 +31,14 @@ export class ForbiddenError extends HttpError { } } +export class ConflictError extends HttpError { + constructor(message?: string, code?: string) { + super(409, message); + this.name = "ConflictError"; + this.code = code; + } +} + export class NotFoundError extends HttpError { constructor(message?: string, code?: string) { super(404, message); diff --git a/src/index.ts b/src/index.ts index 2a235c1..5d677e9 100644 --- a/src/index.ts +++ b/src/index.ts @@ -11,10 +11,10 @@ import * as Webrtc from "./webrtc"; import * as Releases from "./releases"; import { HttpError } from "./errors"; -import { authenticated } from "./auth"; +import { authenticated, bearerToken } from "./auth"; import { prisma } from "./db"; import { baseUrl, bucketName, s3Client } from "./s3"; -import { scheduleReleaseSync } from "./release-sync"; +import { createReleaseSyncRunner, scheduleReleaseSync } from "./release-sync"; import { initializeWebRTCSignaling } from "./webrtc-signaling"; declare global { @@ -49,6 +49,7 @@ declare global { ICE_SERVERS: string; ALLOWED_IDENTITIES?: string; + RELEASE_SYNC_TOKEN?: string; } } } @@ -116,12 +117,23 @@ app.get( }, ); +// One sync at a time, shared by the scheduled tick below and the upload +// script's POST /releases/sync. The route exists only with a token configured. +const releaseSync = createReleaseSyncRunner({ prisma, s3Client }, { bucketName, baseUrl }); + app.get("/releases", Releases.Retrieve); app.get( "/releases/system_recovery/latest", Releases.RetrieveLatestSystemRecovery, ); app.get("/releases/app/latest", Releases.RetrieveLatestApp); +if (process.env.RELEASE_SYNC_TOKEN) { + app.post( + "/releases/sync", + bearerToken(process.env.RELEASE_SYNC_TOKEN), + Releases.Sync(releaseSync), + ); +} app.get("/devices", authenticated, Devices.List); app.get("/devices/:id", authenticated, Devices.Retrieve); @@ -222,4 +234,4 @@ const server = app.listen(PORT, () => { initializeWebRTCSignaling(server); // Register new R2 releases at the default rollout, now and every 30 minutes. -scheduleReleaseSync({ prisma, s3Client }, { bucketName, baseUrl }); +scheduleReleaseSync(releaseSync); diff --git a/src/release-sync.ts b/src/release-sync.ts index 7a2f647..960c6c8 100644 --- a/src/release-sync.ts +++ b/src/release-sync.ts @@ -44,6 +44,8 @@ export type ReleaseOutcome = | "skipped" | "aborted"; +export type SyncStats = Record; + /** Settle window for unattended runs; shorter than the interval, so it costs at most one tick. */ export const UPLOAD_SETTLE_MS = 10 * 60 * 1000; @@ -322,8 +324,8 @@ export async function syncReleases( clients: SyncClients, config: SyncConfig, decide: ReleaseDecider, -): Promise> { - const stats: Record = { +): Promise { + const stats: SyncStats = { created: 0, "already-synced": 0, uploading: 0, @@ -369,38 +371,56 @@ export async function syncReleases( const RELEASE_SYNC_INTERVAL_MS = 30 * 60 * 1000; +/** Runs one unattended sync, or resolves to "busy" while another run is in progress. */ +export type ReleaseSyncRunner = ( + options?: Pick, +) => Promise; + /** - * Runs the sync now and then every `intervalMs`, inside the API process. - * A tick that fires while the previous run is still going is skipped, not - * queued. A failed run is logged and the schedule continues, so one bad R2 - * or DB response never stops future syncs. - * Returns a function that stops the schedule. + * One in-process sync at a time, shared by every trigger (the timer and the + * HTTP endpoint): a second run would only walk the bucket again and lose the + * (version, type) race to the first. */ -export function scheduleReleaseSync( +export function createReleaseSyncRunner( clients: SyncClients, config: SyncConfig, - intervalMs: number = RELEASE_SYNC_INTERVAL_MS, -): () => void { +): ReleaseSyncRunner { let running = false; - const run = async () => { + return async (options = {}) => { if (running) { - console.warn("[sync-releases] previous run still in progress, skipping this tick"); - return; + return "busy"; } running = true; try { - await syncReleases( + return await syncReleases( clients, - { uploadSettleMs: UPLOAD_SETTLE_MS, ...config }, + { uploadSettleMs: UPLOAD_SETTLE_MS, ...config, ...options }, createAtDefaultRollout, ); - } catch (error) { - console.error("[sync-releases] scheduled run failed", error); } finally { running = false; } }; +} + +/** + * Runs the sync now and then every `intervalMs`. A failed run is logged and + * the schedule continues. Returns a function that stops the schedule. + */ +export function scheduleReleaseSync( + runner: ReleaseSyncRunner, + intervalMs: number = RELEASE_SYNC_INTERVAL_MS, +): () => void { + const run = async () => { + try { + if ((await runner()) === "busy") { + console.warn("[sync-releases] previous run still in progress, skipping this tick"); + } + } catch (error) { + console.error("[sync-releases] scheduled run failed", error); + } + }; void run(); const timer = setInterval(run, intervalMs); diff --git a/src/releases.ts b/src/releases.ts index 33d5323..1763513 100644 --- a/src/releases.ts +++ b/src/releases.ts @@ -1,6 +1,7 @@ import { Request, Response } from "express"; import { prisma } from "./db"; -import { BadRequestError, InternalServerError, NotFoundError } from "./errors"; +import { BadRequestError, ConflictError, InternalServerError, NotFoundError } from "./errors"; +import type { ReleaseSyncRunner } from "./release-sync"; import semver from "semver"; import { GetObjectCommand, ListObjectsV2Command } from "@aws-sdk/client-s3"; @@ -710,3 +711,19 @@ function latestArtifactRedirect(kind: OtaKind) { } export const RetrieveLatestApp = latestArtifactRedirect("app"); + +/** + * POST /releases/sync: register every stable R2 version missing from the DB + * and answer with the per-outcome counts. Made for the upload script's last + * step, so the settle window is off: the caller vouches its last object is + * written. + */ +export function Sync(runner: ReleaseSyncRunner) { + return async (req: Request, res: Response) => { + const stats = await runner({ uploadSettleMs: 0 }); + if (stats === "busy") { + throw new ConflictError("A release sync is already in progress"); + } + return res.json(stats); + }; +} diff --git a/test/auth.test.ts b/test/auth.test.ts new file mode 100644 index 0000000..2adcc51 --- /dev/null +++ b/test/auth.test.ts @@ -0,0 +1,31 @@ +import type { Request, Response } from "express"; +import { describe, expect, it, vi } from "vitest"; + +import { bearerToken } from "../src/auth"; +import { UnauthorizedError } from "../src/errors"; + +describe("bearerToken", () => { + const guard = bearerToken("release-sync-secret"); + const res = {} as Response; + + function request(authorization?: string): Request { + return { headers: { authorization } } as unknown as Request; + } + + it("calls next for the configured token", () => { + const next = vi.fn(); + guard(request("Bearer release-sync-secret"), res, next); + expect(next).toHaveBeenCalledOnce(); + }); + + it.each([ + ["no header", undefined], + ["wrong token", "Bearer nope"], + ["missing scheme", "release-sync-secret"], + ["prefix of the token", "Bearer release-sync"], + ])("rejects %s without calling next", (_label, authorization) => { + const next = vi.fn(); + expect(() => guard(request(authorization), res, next)).toThrow(UnauthorizedError); + expect(next).not.toHaveBeenCalled(); + }); +}); diff --git a/test/sync-releases.test.ts b/test/sync-releases.test.ts index aa29e18..fd55f52 100644 --- a/test/sync-releases.test.ts +++ b/test/sync-releases.test.ts @@ -5,17 +5,21 @@ import { S3Client, } from "@aws-sdk/client-s3"; import { Prisma, PrismaClient } from "@prisma/client"; +import type { Request, Response } from "express"; import { afterEach, describe, expect, beforeEach, it, vi } from "vitest"; import { collectReleaseArtifacts, createAtDefaultRollout, + createReleaseSyncRunner, scheduleReleaseSync, syncReleases, UPLOAD_SETTLE_MS, type ReleaseDecider, type ReleaseType, } from "../src/release-sync"; +import { Sync } from "../src/releases"; +import { ConflictError } from "../src/errors"; import { otaFileForPrefix } from "../src/skus"; import { createAsyncIterable, s3Mock, testPrisma } from "./setup"; @@ -92,6 +96,13 @@ beforeEach(() => { s3Mock.on(ListObjectsV2Command).resolves({ Contents: [] }); }); +function newRunner() { + return createReleaseSyncRunner( + { prisma: testPrisma, s3Client: syncS3Client }, + { bucketName: SYNC_BUCKET, baseUrl: SYNC_BASE_URL }, + ); +} + describe("syncReleases", () => { it("marks legacy app artifacts compatible with the default SKU only", async () => { mockS3HashFile("app", "9.9.1", "legacy-app-hash"); @@ -454,11 +465,7 @@ describe("scheduleReleaseSync", () => { mockS3HashFile("app", version, "app-hash"); const errorLog = vi.spyOn(console, "error").mockImplementation(() => {}); - stop = scheduleReleaseSync( - { prisma: testPrisma, s3Client: syncS3Client }, - { bucketName: SYNC_BUCKET, baseUrl: SYNC_BASE_URL }, - INTERVAL_MS, - ); + stop = scheduleReleaseSync(newRunner(), INTERVAL_MS); await vi.waitFor(() => expect(errorLog).toHaveBeenCalledOnce()); expect( @@ -488,11 +495,7 @@ describe("scheduleReleaseSync", () => { .resolves({ CommonPrefixes: [] }); const warnLog = vi.spyOn(console, "warn").mockImplementation(() => {}); - stop = scheduleReleaseSync( - { prisma: testPrisma, s3Client: syncS3Client }, - { bucketName: SYNC_BUCKET, baseUrl: SYNC_BASE_URL }, - INTERVAL_MS, - ); + stop = scheduleReleaseSync(newRunner(), INTERVAL_MS); // Two ticks fire while the first run is still waiting on R2. await vi.advanceTimersByTimeAsync(INTERVAL_MS * 2); @@ -515,3 +518,33 @@ describe("scheduleReleaseSync", () => { expect(warnLog).toHaveBeenCalledTimes(2); }); }); + +describe("Sync handler", () => { + const request = {} as Request; + + it("registers a version uploaded moments ago and answers with the counts", async () => { + const version = "9.9.12"; + mockS3ListVersions("app", [version]); + mockS3HashFile("app", version, "fresh-hash"); + // Inside the settle window: the scheduled tick would defer this version, + // the endpoint trusts the caller and registers it. + mockS3UploadedAt("app", version, new Date(Date.now() - 60 * 1000)); + const res = { json: vi.fn() } as unknown as Response; + + await Sync(newRunner())(request, res); + + expect(res.json).toHaveBeenCalledWith(expect.objectContaining({ created: 1, uploading: 0 })); + const release = await testPrisma.release.findUniqueOrThrow({ + where: { version_type: { version, type: "app" } }, + }); + expect(release.rolloutPercentage).toBe(10); + }); + + it("answers 409 while a run is already in progress", async () => { + const busyRunner = vi.fn().mockResolvedValue("busy"); + + await expect(Sync(busyRunner)(request, { json: vi.fn() } as unknown as Response)).rejects.toThrow( + ConflictError, + ); + }); +});