From 1a2236d61a999cee6dc7602bc2246e2fc33731bd Mon Sep 17 00:00:00 2001 From: Matt Johnston Date: Mon, 31 Aug 2026 15:45:54 -0300 Subject: [PATCH 1/5] feat(workers logs): add `supabase workers logs` MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit `status` reports the deployment; nothing reported the runtime. Once `push` succeeded and `status` said `active`, a misbehaving worker was a black box from the CLI. Reads the project's unified logs stream rather than a worker route — there is no worker-scoped log endpoint — via `v1GetProjectLogs`, which the generated client already carries. `--source app|requests|builds` narrows to one of the three streams; without it all three are returned. `--tail` caps the rows. Three things about that endpoint are load-bearing and non-obvious, so they are documented at each site: - **The filter is `log_attributes`, not the `source` column.** Worker rows carry an empty top-level `source`, because the Workers Logflare source is not enrolled as a category in the generic logs path. `where source = 'worker_guest_logs'` matches nothing. The `in (...)` list over the three known streams is therefore a tenancy guard, not a convenience — with `source` empty it is the only thing excluding a non-worker row that happens to carry a `worker` attribute. - **Both timestamp bounds are always sent, spanning under 24h.** One bound alone yields a one-minute window, silently; neither is an outright error; and a span over 24h is clamped to `start + 24h`, which returns an *older* slice than the one asked for rather than a truncated one. - **A failed query can arrive as HTTP 200** with a populated `error`, so the envelope is checked before `result`. The response is decoded against a local schema rather than the generated `V1GetProjectLogsOutput`: that schema marks `result`/`error` optional but allows neither to be `null`, while the endpoint always sends one of them as an explicit `null`, so decoding any real response against it fails. Rendering is per-stream, because `event_message` differs in kind — on the request stream it is only `"GET /"`, with status and duration in `log_attributes`, so the request line is composed. `severity_text` is ignored: it is `INFO` on every row of every stream, so the level is derived, and guest lines report none rather than a guess. A guest message is tenant-controlled bytes, so escape sequences are stripped before it reaches a terminal while a stack trace's newlines and indentation survive. `mapRequestError`/`unexpectedStatus`/`decodeBody` move out of `workers-api.ts` into `workers-api-status.ts`, unchanged, now that a second seam needs them. The test helper records `urlParams`: `HttpClientRequest` keeps them off the URL, so without this no test could assert the emitted SQL or window. --- .../experimental/workers/logs/SIDE_EFFECTS.md | 114 ++++ .../experimental/workers/logs/logs.command.ts | 73 +++ .../experimental/workers/logs/logs.handler.ts | 158 +++++ .../workers/logs/logs.integration.test.ts | 561 ++++++++++++++++++ .../workers/workers-logs.format.ts | 160 +++++ .../workers/workers-logs.format.unit.test.ts | 199 +++++++ .../experimental/workers/workers.command.ts | 2 + .../legacy/shared/legacy-db-target-flags.ts | 1 + .../cli/src/shared/workers/worker-logs-api.ts | 200 +++++++ .../cli/src/shared/workers/worker-logs.sql.ts | 131 ++++ .../workers/worker-logs.sql.unit.test.ts | 119 ++++ .../src/shared/workers/workers-api-status.ts | 79 +++ apps/cli/src/shared/workers/workers-api.ts | 68 +-- apps/cli/src/shared/workers/workers.errors.ts | 46 ++ apps/cli/tests/helpers/legacy-workers.ts | 98 +++ 15 files changed, 1942 insertions(+), 67 deletions(-) create mode 100644 apps/cli/src/legacy/commands/experimental/workers/logs/SIDE_EFFECTS.md create mode 100644 apps/cli/src/legacy/commands/experimental/workers/logs/logs.command.ts create mode 100644 apps/cli/src/legacy/commands/experimental/workers/logs/logs.handler.ts create mode 100644 apps/cli/src/legacy/commands/experimental/workers/logs/logs.integration.test.ts create mode 100644 apps/cli/src/legacy/commands/experimental/workers/workers-logs.format.ts create mode 100644 apps/cli/src/legacy/commands/experimental/workers/workers-logs.format.unit.test.ts create mode 100644 apps/cli/src/shared/workers/worker-logs-api.ts create mode 100644 apps/cli/src/shared/workers/worker-logs.sql.ts create mode 100644 apps/cli/src/shared/workers/worker-logs.sql.unit.test.ts create mode 100644 apps/cli/src/shared/workers/workers-api-status.ts diff --git a/apps/cli/src/legacy/commands/experimental/workers/logs/SIDE_EFFECTS.md b/apps/cli/src/legacy/commands/experimental/workers/logs/SIDE_EFFECTS.md new file mode 100644 index 0000000000..613c88e4ce --- /dev/null +++ b/apps/cli/src/legacy/commands/experimental/workers/logs/SIDE_EFFECTS.md @@ -0,0 +1,114 @@ +# `supabase experimental workers logs ` + +> **No live test yet.** The other `workers` commands skip live coverage because they +> run against the v2 Management API, which the supabase/cli-e2e-ci supabox stack is +> not expected to serve. This one reads the v1 analytics endpoint, which that stack +> may well serve — but a meaningful assertion needs a deployed worker that has +> actually emitted log lines, which the stack cannot provide. Revisit alongside the +> rest of the family. + +## Files Read + +| Path | Format | When | +| --------------------------------------------- | ---------- | ----------------------------------------------------------------------------------------------------------- | +| `/access-token` | plain text | when `SUPABASE_ACCESS_TOKEN` is unset and the keyring holds no credential | +| `/supabase/.temp/project-ref` | plain text | when neither `--project-ref` nor `SUPABASE_PROJECT_ID` is set — names the linked project | +| `/profile` | plain text | when neither `--profile` nor `SUPABASE_PROFILE` is set — names the profile, defaulting to `supabase` | +| `` (YAML) | YAML | when `SUPABASE_PROFILE` is a filesystem path rather than a built-in name; a read failure aborts the command | + +The project config is **not** read. Unlike `status` and `delete`, nothing in this +command's output depends on local state — there is no source path to report — so +`config.toml` is never opened and an unparseable one cannot block a log read. + +## Files Written + +| Path | Format | When | +| ----------------------------------------------- | ------ | --------------------------------------------------------------- | +| `/telemetry.json` | JSON | always — flushed on success and on failure | +| `/supabase/.temp/linked-project.json` | JSON | after the project ref resolves, when the cache does not hold it | + +## API Routes + +| Method | Path | Auth | Request | Response (used fields) | +| ------ | --------------------------------------------- | ------------ | ----------------------------------------------------------------------------------------------------- | ---------------------- | +| `GET` | `/v1/projects/{ref}/analytics/endpoints/logs` | Bearer token | `sql`, `iso_timestamp_start`, `iso_timestamp_end` as query parameters | `result[]`, `error` | +| `GET` | `/v2/projects/{ref}/workers/{name}` | Bearer token | none — **only when the log query returned no rows**, to tell "not deployed" from "deployed and quiet" | presence only | +| `GET` | `/v1/projects` | Bearer token | none — only when no ref resolved and the session is interactive | project picker | + +Requires the `analytics_logs_read` permission, and the project must be on the +Workers private-alpha allow-list — an unenrolled project answers 404. + +### The query + +SQL in **ClickHouse dialect** against the project's unified `logs` table, filtered +on `log_attributes['worker']` and `log_attributes['source']`. It does **not** filter +the top-level `source` column: worker rows carry an empty string there, because the +Workers Logflare source is not enrolled as a category in the generic logs path. + +### The window + +Both `iso_timestamp_start` and `iso_timestamp_end` are always sent, spanning just +under 24 hours. This is not optional: + +- one bound alone yields a **one-minute** window, server-side and silently; +- neither bound is an outright error; +- a span over 24 hours is **silently clamped** to `start + 24h`, which returns an + older slice than the one requested rather than a truncated one. + +### Rate limits + +The v1 analytics endpoints allow **10 requests per 60 seconds**, and the server +applies a 30-second query timeout. One invocation spends one request, or two when +the result is empty. + +## Exit Codes + +| Code | Condition | +| ---- | ------------------------------------------------------------ | +| `0` | success, including "no logs in the last 24 hours" | +| `1` | invalid worker name | +| `1` | nothing deployed under that name | +| `1` | the log query failed (rejected, or the server's 30s timeout) | +| `1` | log usage exceeded (402), or rate limited (429) | +| `1` | API error, or project not enrolled in the alpha | + +## Environment Variables + +| Variable | Purpose | Required? | +| ----------------------- | ---------------------------------------------------- | ---------------------------------------------------------------- | +| `SUPABASE_ACCESS_TOKEN` | auth token (bypasses credential file/keyring lookup) | no (falls back to keyring → `~/.supabase/access-token`) | +| `SUPABASE_PROFILE` | built-in profile name or YAML file path | no (falls back to `~/.supabase/profile` -> `supabase`) | +| `SUPABASE_PROJECT_ID` | project ref, consulted after `--project-ref` | no (falls back to `supabase/.temp/project-ref`, then the picker) | +| `SUPABASE_WORKDIR` | project directory the command acts on | no (falls back to `--workdir`, then the ancestor walk) | +| `SUPABASE_HOME` | directory holding `telemetry.json` | no (falls back to `~/.supabase`) | + +## Telemetry Events Fired + +| Event | When | Notable properties / groups | +| ---------------------- | ------------------------------------------ | ----------------------------------- | +| `cli_command_executed` | post-run, success or failure (via wrapper) | `exit_code`, `duration_ms`, `flags` | + +`--source` is a choice flag, so its value is logged verbatim (a closed enum carries +no user data). `--project-ref` is not on this command's safe list, so its value is +redacted. No custom events. + +## Output Formats + +| Mode | stdout | stderr | +| ----------------------------- | --------------------------------------------------------------------------------------------------- | --------------------------------------------------------- | +| text (default) | one line per entry, oldest first, per-stream layout; severity as colour | the spinner, and the `status` hint when there are no logs | +| `--output-format json` | one structured result carrying every entry | as above | +| `--output-format stream-json` | the same result as a single terminal event | as above | +| `-o json` / `yaml` / `toml` | the same payload in that encoding, and nothing else | as above | +| `-o pretty` / `table` / `csv` | the text rendering — these fall through rather than encoding | as above | +| `-o env` | refused before any request; the payload nests a `logs` array a flat `KEY=value` list cannot express | the error | + +Machine payloads carry each entry's `id`, both `timestamp` (ISO) and +`timestamp_ms`, `stream`, `message`, the derived `level` when one exists, and the +raw `attributes` map — whose values are all strings, since the column is a +`Map(String, String)`. + +A `worker_guest_logs` message is bytes the tenant's own code printed. Control and +escape sequences are stripped before it reaches a terminal, so a worker cannot +reposition the cursor or forge CLI output; interior newlines and indentation are +preserved so a stack trace survives intact. diff --git a/apps/cli/src/legacy/commands/experimental/workers/logs/logs.command.ts b/apps/cli/src/legacy/commands/experimental/workers/logs/logs.command.ts new file mode 100644 index 0000000000..44b7ef71fe --- /dev/null +++ b/apps/cli/src/legacy/commands/experimental/workers/logs/logs.command.ts @@ -0,0 +1,73 @@ +import { Argument, Command, Flag } from "effect/unstable/cli"; +import type * as CliCommand from "effect/unstable/cli/Command"; +import { withJsonErrorHandling } from "../../../../../shared/output/json-error-handling.ts"; +import { legacyManagementApiRuntimeLayer } from "../../../../shared/legacy-management-api-runtime.layer.ts"; +import { withLegacyCommandInstrumentation } from "../../../../telemetry/legacy-command-instrumentation.ts"; +import { legacyWorkersLogs } from "./logs.handler.ts"; + +/** The `--source` words, mapped to stream names in `worker-logs.sql.ts`. */ +const SOURCE_VALUES = ["app", "requests", "builds"] as const; + +/** + * The endpoint's own ceiling is the SQL `LIMIT`, so this bound is the CLI's + * choice. 1000 is high enough to be a non-issue in practice and low enough that a + * typo cannot ask for a payload nobody wants. + * + * 0 is allowed and means "no history", which only becomes useful alongside + * `--follow`; on its own it prints nothing and makes no request. + */ +const MAX_TAIL = 1000; + +const config = { + name: Argument.string("name").pipe(Argument.withDescription("Worker to read logs for.")), + projectRef: Flag.string("project-ref").pipe( + Flag.withDescription("Project ref of the Supabase project."), + Flag.optional, + ), + source: Flag.choice("source", SOURCE_VALUES).pipe( + Flag.withDescription( + "Limit to one log stream: app (the worker's own output), requests (HTTP access), " + + "builds (deploy lifecycle). Defaults to all three.", + ), + Flag.optional, + ), + tail: Flag.integer("tail").pipe( + Flag.filter( + (tail) => tail >= 0 && tail <= MAX_TAIL, + (tail) => `Expected --tail between 0 and ${MAX_TAIL}, got ${tail}`, + ), + Flag.withDescription("Number of log lines to print."), + Flag.withDefault(100), + ), +} as const; + +export type LegacyWorkersLogsFlags = CliCommand.Command.Config.Infer; + +export const legacyWorkersLogsCommand = Command.make("logs", config).pipe( + Command.withDescription( + "Print a worker's recent logs: its own output, the HTTP requests it served, and its " + + "deploy lifecycle events.\n\n" + + "Covers the last 24 hours, which is the longest window the logs API will answer in one " + + "query. Lines are printed oldest first.", + ), + Command.withShortDescription("Show a worker's logs"), + Command.withExamples([ + { + command: "supabase experimental workers logs api", + description: "Print the last 100 log lines across all streams", + }, + { + command: "supabase experimental workers logs api --source requests --tail 20", + description: "Print the 20 most recent HTTP requests the worker served", + }, + ]), + Command.withHandler((flags) => + legacyWorkersLogs(flags).pipe( + // `config` as well as `flags`: `--source` is a choice flag, and the wrapper + // treats a command's own declared choices as safe to log verbatim. + withLegacyCommandInstrumentation({ flags, config }), + withJsonErrorHandling, + ), + ), + Command.provide(legacyManagementApiRuntimeLayer(["experimental", "workers", "logs"])), +); diff --git a/apps/cli/src/legacy/commands/experimental/workers/logs/logs.handler.ts b/apps/cli/src/legacy/commands/experimental/workers/logs/logs.handler.ts new file mode 100644 index 0000000000..c284fbdc9c --- /dev/null +++ b/apps/cli/src/legacy/commands/experimental/workers/logs/logs.handler.ts @@ -0,0 +1,158 @@ +import { Effect, Option } from "effect"; +import { Output } from "../../../../../shared/output/output.service.ts"; +import { emitSuccessTrailer } from "../../../../../shared/cli/success-trailer.ts"; +import { legacyAqua } from "../../../../shared/legacy-colors.ts"; +import { + legacyEmitWorkersMachineOutput, + legacyRejectWorkersEnvOutput, + legacyWorkersProjectRefSuffix, +} from "../workers.output.ts"; +import { legacyRenderWorkerLogLine, legacyWorkerLogLevel } from "../workers-logs.format.ts"; +import { LegacyPlatformApi } from "../../../../auth/legacy-platform-api.service.ts"; +import { + fetchWorkerLogs, + type WorkerLogEntry, +} from "../../../../../shared/workers/worker-logs-api.ts"; +import { + ALL_WORKER_LOG_STREAMS, + logWindow, + WORKER_LOG_STREAMS, + type WorkerLogSourceChoice, +} from "../../../../../shared/workers/worker-logs.sql.ts"; +import { getWorker } from "../../../../../shared/workers/workers-api.ts"; +import { WorkerNotDeployedError } from "../../../../../shared/workers/workers.errors.ts"; +import { LegacyProjectRefResolver } from "../../../../config/legacy-project-ref.service.ts"; +import { LegacyLinkedProjectCache } from "../../../../telemetry/legacy-linked-project-cache.service.ts"; +import { LegacyTelemetryState } from "../../../../telemetry/legacy-telemetry-state.service.ts"; +import { legacyValidateWorkerName } from "../workers.shared.ts"; +import type { LegacyWorkersLogsFlags } from "./logs.command.ts"; + +/** + * `supabase experimental workers logs ` — what the worker has actually been doing. + * + * `status` reports the deployment; this reports the runtime. Between them they + * cover the two questions a deployed worker raises, and neither answers the + * other's. + * + * Unlike the rest of the family this does not talk to `/v2/.../workers` — there is + * no worker-scoped log route — but to the project's unified logs stream. See + * `worker-logs.sql.ts` for the query and why it filters on `log_attributes` + * rather than the `source` column. + */ + +/** The machine-format row for one line. */ +function toPayloadEntry(entry: WorkerLogEntry) { + const level = legacyWorkerLogLevel(entry); + return { + id: entry.id, + // Both forms: the ISO string is what a human or `jq` wants to read, the raw + // epoch value is what a script sorts or diffs on without reparsing. + timestamp: new Date(entry.timestampMs).toISOString(), + timestamp_ms: entry.timestampMs, + stream: entry.stream, + message: entry.message, + ...(level === undefined ? {} : { level }), + attributes: entry.attributes, + }; +} + +export const legacyWorkersLogs = Effect.fn("legacy.experimental.workers.logs")(function* ( + flags: LegacyWorkersLogsFlags, +) { + const output = yield* Output; + const api = yield* LegacyPlatformApi; + const resolver = yield* LegacyProjectRefResolver; + const linkedProjectCache = yield* LegacyLinkedProjectCache; + const telemetryState = yield* LegacyTelemetryState; + + // The ref is resolved outside the finalizers because caching it is one of them; + // everything that can fail on its own belongs inside, so those failures still + // flush telemetry. Same shape as the rest of the family. + const projectRef = yield* resolver.resolve(flags.projectRef); + const refSuffix = legacyWorkersProjectRefSuffix(flags.projectRef); + + yield* Effect.gen(function* () { + const name = yield* legacyValidateWorkerName(flags.name); + + // Up front, like the rest of the family: this payload always carries a `logs` + // array, so `-o env` can never encode it, and finding that out at emit time + // means failing after the query has already been paid for. + yield* legacyRejectWorkersEnvOutput(); + + const streams = Option.isSome(flags.source) + ? [WORKER_LOG_STREAMS[flags.source.value as WorkerLogSourceChoice]] + : ALL_WORKER_LOG_STREAMS; + + // `--tail 0` is "no history". On its own that is a no-op, but it is the shape + // `--follow` will want, and issuing a `limit 0` query would be a 400. + const entries = + flags.tail === 0 + ? [] + : yield* Effect.gen(function* () { + const fetching = yield* output.task("Fetching logs..."); + const rows = yield* fetchWorkerLogs(api, projectRef, { + name, + streams, + tail: flags.tail, + window: logWindow(new Date()), + }).pipe(Effect.tapError(() => fetching.fail())); + yield* fetching.clear(); + return rows; + }); + + // Nothing came back, which is two different situations wearing the same face: + // a worker that is not deployed at all, and one that is deployed and quiet. + // Only worth one extra request, and only in this branch. + if (entries.length === 0) { + const deployed = yield* getWorker(api, projectRef, name); + if (Option.isNone(deployed)) { + return yield* Effect.fail( + new WorkerNotDeployedError({ + detail: `Nothing is deployed for "${name}" in project ${projectRef}.`, + suggestion: `Deploy it with \`supabase experimental workers push ${name}${refSuffix}\`.`, + }), + ); + } + } + + const payload = { + worker_name: name, + project_ref: projectRef, + ...(Option.isSome(flags.source) ? { source: flags.source.value } : {}), + logs: entries.map(toPayloadEntry), + }; + + // `-o` asks for a machine-readable stdout, so nothing human may be written to + // it — `output.success` logs to stdout in text mode. + if (yield* legacyEmitWorkersMachineOutput(payload)) { + return; + } + + // One structured emission, in the structured branch only. Emitting before the + // check above put the payload on stdout twice. + if (output.format !== "text") { + yield* output.success("", payload); + return; + } + + if (entries.length === 0) { + // Deployed (the check above would have failed otherwise) and silent. + yield* output.raw(`No logs for "${name}" in the last 24 hours.\n`); + yield* emitSuccessTrailer( + `Check it is running with ${legacyAqua(`supabase experimental workers status ${name}${refSuffix}`)}.\n`, + ); + return; + } + + // Oldest first: the query orders newest-first so `limit` means "most recent", + // but a reader scrolls forwards through time, and a stack trace only makes + // sense in the order it was printed. + // No TTY check here: `legacyRenderWorkerLogLine` defaults to `process.stdout` + // and the colour helpers gate on it themselves, honouring NO_COLOR, CLICOLOR, + // CLICOLOR_FORCE and CI as well as the stream. + yield* output.raw(`${entries.map((entry) => legacyRenderWorkerLogLine(entry)).join("\n")}\n`); + }).pipe( + Effect.ensuring(linkedProjectCache.cache(projectRef)), + Effect.ensuring(telemetryState.flush), + ); +}); diff --git a/apps/cli/src/legacy/commands/experimental/workers/logs/logs.integration.test.ts b/apps/cli/src/legacy/commands/experimental/workers/logs/logs.integration.test.ts new file mode 100644 index 0000000000..cd1e9ba655 --- /dev/null +++ b/apps/cli/src/legacy/commands/experimental/workers/logs/logs.integration.test.ts @@ -0,0 +1,561 @@ +import { rmSync } from "node:fs"; +import { describe, expect, it } from "@effect/vitest"; +import { Effect, Option } from "effect"; +import { + makeWorkersProject, + setupLegacyWorkers, + workerApiLogRow, + workerIngressLogRow, + workerLogRow, + workerLogsRoute, + workerResource, + workersRoute, + WORKERS_PROJECT_REF, +} from "../../../../../../tests/helpers/legacy-workers.ts"; +import { + InvalidWorkerNameError, + WorkerLogsQueryFailedError, + WorkerLogsRateLimitedError, + WorkerLogsUsageExceededError, + WorkerNotDeployedError, + WorkersApiNetworkError, + WorkersApiUnexpectedStatusError, + WorkersUnavailableError, +} from "../../../../../shared/workers/workers.errors.ts"; +import { LegacyWorkersEnvNotSupportedError } from "../workers.errors.ts"; +import { legacyWorkersLogs } from "./logs.handler.ts"; + +const ESCAPE = "\u001b"; +const CONFIG = 'project_id = "demo"\n\n[workers.api]\nruntime = "node"\n'; +const LOGS_ROUTE = `GET ${workerLogsRoute()}`; +const GET_WORKER_ROUTE = `GET ${workersRoute("/api")}`; + +const T1 = 1_788_187_525_212; +const T2 = 1_788_187_531_671; +const T3 = 1_788_187_532_576; + +function project() { + const created = makeWorkersProject({ + "supabase/config.toml": CONFIG, + "supabase/workers/api/index.js": "export default {};\n", + }); + return { dir: created.dir, cleanup: () => rmSync(created.dir, { recursive: true, force: true }) }; +} + +/** The default flag set; every test overrides only what it is about. */ +function flags(overrides: Record = {}) { + return { + name: "api", + projectRef: Option.none(), + source: Option.none(), + tail: 100, + ...overrides, + } as Parameters[0]; +} + +function logsResponse(rows: ReadonlyArray) { + return { status: 200, body: { result: rows, error: null } }; +} + +/** + * The query parameters the handler actually sent. + * + * Read off the recorded request rather than the URL: `HttpClientRequest` keeps + * `urlParams` beside the URL rather than appended to it. + */ +function sentQuery(request: { readonly urlParams: Readonly> }) { + return request.urlParams; +} + +describe("legacy workers logs", () => { + it.live("prints a worker's own output oldest first", () => { + const repo = project(); + const { layer, out } = setupLegacyWorkers({ + workdir: repo.dir, + routes: { + [LOGS_ROUTE]: logsResponse([ + workerLogRow({ id: "c", tsMs: T3, message: "app drained" }), + workerLogRow({ id: "a", tsMs: T1, message: "listening on :8080" }), + workerLogRow({ id: "b", tsMs: T2, message: "terminate hook" }), + ]), + }, + }); + + return Effect.gen(function* () { + yield* legacyWorkersLogs(flags()); + + const lines = out.stdoutText.trimEnd().split("\n"); + expect(lines.map((line) => line.split(" ")[1])).toEqual([ + "listening on :8080", + "terminate hook", + "app drained", + ]); + }).pipe(Effect.provide(layer), Effect.ensuring(Effect.sync(repo.cleanup))); + }); + + it.live("composes a request line from attributes rather than the message", () => { + const repo = project(); + const { layer, out } = setupLegacyWorkers({ + workdir: repo.dir, + routes: { + [LOGS_ROUTE]: logsResponse([ + workerIngressLogRow({ + tsMs: T1, + status: "500", + method: "POST", + path: "/checkout", + durationMs: "7", + }), + ]), + }, + }); + + return Effect.gen(function* () { + yield* legacyWorkersLogs(flags()); + + expect(out.stdoutText).toContain("500 POST /checkout 7ms"); + }).pipe(Effect.provide(layer), Effect.ensuring(Effect.sync(repo.cleanup))); + }); + + it.live("prints a build event with its reason", () => { + const repo = project(); + const { layer, out } = setupLegacyWorkers({ + workdir: repo.dir, + routes: { + [LOGS_ROUTE]: logsResponse([ + workerApiLogRow({ tsMs: T1, event: "build_failed", reason: "exit status 1" }), + ]), + }, + }); + + return Effect.gen(function* () { + yield* legacyWorkersLogs(flags()); + + expect(out.stdoutText).toContain("build_failed exit status 1"); + }).pipe(Effect.provide(layer), Effect.ensuring(Effect.sync(repo.cleanup))); + }); + + it.live("always sends both timestamp bounds, under a 24 hour span", () => { + const repo = project(); + const { layer, http } = setupLegacyWorkers({ + workdir: repo.dir, + routes: { [LOGS_ROUTE]: logsResponse([workerLogRow({})]) }, + }); + + return Effect.gen(function* () { + yield* legacyWorkersLogs(flags()); + + const query = sentQuery(http.requests[0]!); + const start = query.iso_timestamp_start; + const end = query.iso_timestamp_end; + + // A lone bound yields a one-minute window server-side and sending neither + // is an outright error, so both must always be present. + expect(start).toBeTruthy(); + expect(end).toBeTruthy(); + expect(start!.endsWith("Z")).toBe(true); + expect(end!.endsWith("Z")).toBe(true); + // Over 24h the server silently clamps to start+24h, returning an older + // slice than the one asked for. + expect(Date.parse(end!) - Date.parse(start!)).toBeLessThan(24 * 60 * 60 * 1000); + }).pipe(Effect.provide(layer), Effect.ensuring(Effect.sync(repo.cleanup))); + }); + + it.live("filters on log_attributes, never the empty source column", () => { + const repo = project(); + const { layer, http } = setupLegacyWorkers({ + workdir: repo.dir, + routes: { [LOGS_ROUTE]: logsResponse([workerLogRow({})]) }, + }); + + return Effect.gen(function* () { + yield* legacyWorkersLogs(flags()); + + const sql = sentQuery(http.requests[0]!).sql ?? ""; + expect(sql).toContain("log_attributes['worker'] = 'api'"); + expect(sql).toContain("log_attributes['source'] in ("); + expect(sql).not.toMatch(/where source =/); + }).pipe(Effect.provide(layer), Effect.ensuring(Effect.sync(repo.cleanup))); + }); + + it.live("narrows to one stream for --source, and to all three without it", () => { + const repo = project(); + const { layer, http } = setupLegacyWorkers({ + workdir: repo.dir, + routes: { [LOGS_ROUTE]: logsResponse([workerLogRow({})]) }, + }); + + return Effect.gen(function* () { + yield* legacyWorkersLogs(flags({ source: Option.some("requests") })); + const narrowed = sentQuery(http.requests[0]!).sql ?? ""; + expect(narrowed).toContain("in ('worker_ingress_logs')"); + + yield* legacyWorkersLogs(flags()); + const all = sentQuery(http.requests[1]!).sql ?? ""; + expect(all).toContain("'worker_guest_logs'"); + expect(all).toContain("'worker_ingress_logs'"); + expect(all).toContain("'worker_api_logs'"); + }).pipe(Effect.provide(layer), Effect.ensuring(Effect.sync(repo.cleanup))); + }); + + it.live("renders a stream it has never heard of rather than failing", () => { + const repo = project(); + const { layer, out } = setupLegacyWorkers({ + workdir: repo.dir, + routes: { + [LOGS_ROUTE]: logsResponse([ + workerLogRow({ tsMs: T1, stream: "worker_future_logs", message: "from the future" }), + ]), + }, + }); + + return Effect.gen(function* () { + // The log contract is additive-only: unknown streams must be ignored, not + // rejected. + yield* legacyWorkersLogs(flags()); + + expect(out.stdoutText).toContain("from the future"); + }).pipe(Effect.provide(layer), Effect.ensuring(Effect.sync(repo.cleanup))); + }); + + it.live("strips escape sequences a worker printed", () => { + const repo = project(); + const { layer, out } = setupLegacyWorkers({ + workdir: repo.dir, + routes: { + [LOGS_ROUTE]: logsResponse([ + workerLogRow({ tsMs: T1, message: `${ESCAPE}[31mERROR: not really${ESCAPE}[0m` }), + ]), + }, + }); + + return Effect.gen(function* () { + yield* legacyWorkersLogs(flags()); + + expect(out.stdoutText).toContain("ERROR: not really"); + expect(out.stdoutText).not.toContain(ESCAPE); + }).pipe(Effect.provide(layer), Effect.ensuring(Effect.sync(repo.cleanup))); + }); + + it.live("keeps a blank guest line as a line", () => { + const repo = project(); + const { layer, out } = setupLegacyWorkers({ + workdir: repo.dir, + routes: { + [LOGS_ROUTE]: logsResponse([ + workerLogRow({ id: "a", tsMs: T1, message: "before" }), + workerLogRow({ id: "b", tsMs: T2, message: "" }), + workerLogRow({ id: "c", tsMs: T3, message: "after" }), + ]), + }, + }); + + return Effect.gen(function* () { + yield* legacyWorkersLogs(flags()); + + expect(out.stdoutText.trimEnd().split("\n")).toHaveLength(3); + }).pipe(Effect.provide(layer), Effect.ensuring(Effect.sync(repo.cleanup))); + }); + + it.live("makes no request at all for --tail 0", () => { + const repo = project(); + const { layer, http } = setupLegacyWorkers({ + workdir: repo.dir, + routes: { + [GET_WORKER_ROUTE]: { status: 200, body: { data: workerResource({ name: "api" }) } }, + }, + }); + + return Effect.gen(function* () { + // `limit 0` would be a 400, so no-history has to mean no query. + yield* legacyWorkersLogs(flags({ tail: 0 })); + + expect(http.routeKeys).not.toContain(workerLogsRoute()); + }).pipe(Effect.provide(layer), Effect.ensuring(Effect.sync(repo.cleanup))); + }); + + it.live("passes --tail through as the row limit", () => { + const repo = project(); + const { layer, http } = setupLegacyWorkers({ + workdir: repo.dir, + routes: { [LOGS_ROUTE]: logsResponse([workerLogRow({})]) }, + }); + + return Effect.gen(function* () { + yield* legacyWorkersLogs(flags({ tail: 7 })); + + expect(sentQuery(http.requests[0]!).sql).toContain("limit 7"); + }).pipe(Effect.provide(layer), Effect.ensuring(Effect.sync(repo.cleanup))); + }); + + it.live("reports a worker that is not deployed rather than an empty screen", () => { + const repo = project(); + const { layer } = setupLegacyWorkers({ + workdir: repo.dir, + routes: { + [LOGS_ROUTE]: logsResponse([]), + [GET_WORKER_ROUTE]: { status: 404 }, + }, + }); + + return Effect.gen(function* () { + const error = yield* legacyWorkersLogs(flags()).pipe(Effect.flip); + + expect(error).toBeInstanceOf(WorkerNotDeployedError); + const suggestion = error instanceof WorkerNotDeployedError ? error.suggestion : ""; + expect(suggestion).toContain("supabase experimental workers push api"); + }).pipe(Effect.provide(layer), Effect.ensuring(Effect.sync(repo.cleanup))); + }); + + it.live("says so when a deployed worker has simply been quiet", () => { + const repo = project(); + const { layer, out } = setupLegacyWorkers({ + workdir: repo.dir, + routes: { + [LOGS_ROUTE]: logsResponse([]), + [GET_WORKER_ROUTE]: { status: 200, body: { data: workerResource({ name: "api" }) } }, + }, + }); + + return Effect.gen(function* () { + yield* legacyWorkersLogs(flags()); + + expect(out.stdoutText).toContain('No logs for "api" in the last 24 hours.'); + }).pipe(Effect.provide(layer), Effect.ensuring(Effect.sync(repo.cleanup))); + }); + + it.live("treats absent, null and empty result identically", () => { + const repo = project(); + const { layer, out } = setupLegacyWorkers({ + workdir: repo.dir, + routes: { + [LOGS_ROUTE]: [ + { status: 200, body: {} }, + { status: 200, body: { result: null } }, + { status: 200, body: { result: [] } }, + ], + [GET_WORKER_ROUTE]: { status: 200, body: { data: workerResource({ name: "api" }) } }, + }, + }); + + return Effect.gen(function* () { + for (const _ of [0, 1, 2]) { + yield* legacyWorkersLogs(flags()); + } + + expect(out.stdoutText.match(/No logs for/gu)).toHaveLength(3); + }).pipe(Effect.provide(layer), Effect.ensuring(Effect.sync(repo.cleanup))); + }); + + it.live("fails on a 200 that carries a query error", () => { + const repo = project(); + const { layer } = setupLegacyWorkers({ + workdir: repo.dir, + routes: { + [LOGS_ROUTE]: { status: 200, body: { result: null, error: "query timed out" } }, + }, + }); + + return Effect.gen(function* () { + // The endpoint reports a failed query with a 200, so reading `result` + // first would report success. + const error = yield* legacyWorkersLogs(flags()).pipe(Effect.flip); + + expect(error).toBeInstanceOf(WorkerLogsQueryFailedError); + const detail = error instanceof WorkerLogsQueryFailedError ? error.detail : ""; + expect(detail).toContain("query timed out"); + }).pipe(Effect.provide(layer), Effect.ensuring(Effect.sync(repo.cleanup))); + }); + + it.live("reads the structured form of a query error too", () => { + const repo = project(); + const { layer } = setupLegacyWorkers({ + workdir: repo.dir, + routes: { + [LOGS_ROUTE]: { + status: 200, + body: { + result: null, + error: { code: 400, message: "Unknown expression", status: "INVALID", errors: [] }, + }, + }, + }, + }); + + return Effect.gen(function* () { + const error = yield* legacyWorkersLogs(flags()).pipe(Effect.flip); + + const detail = error instanceof WorkerLogsQueryFailedError ? error.detail : ""; + expect(detail).toContain("Unknown expression"); + }).pipe(Effect.provide(layer), Effect.ensuring(Effect.sync(repo.cleanup))); + }); + + it.live("maps 402 to a usage error and 429 to a rate limit error", () => { + const repo = project(); + const { layer } = setupLegacyWorkers({ + workdir: repo.dir, + routes: { [LOGS_ROUTE]: [{ status: 402 }, { status: 429 }] }, + }); + + return Effect.gen(function* () { + const usage = yield* legacyWorkersLogs(flags()).pipe(Effect.flip); + const limited = yield* legacyWorkersLogs(flags()).pipe(Effect.flip); + + expect(usage).toBeInstanceOf(WorkerLogsUsageExceededError); + expect(limited).toBeInstanceOf(WorkerLogsRateLimitedError); + }).pipe(Effect.provide(layer), Effect.ensuring(Effect.sync(repo.cleanup))); + }); + + it.live("reports a project outside the alpha for a 404", () => { + const repo = project(); + const { layer } = setupLegacyWorkers({ + workdir: repo.dir, + routes: { [LOGS_ROUTE]: { status: 404 } }, + }); + + return Effect.gen(function* () { + const error = yield* legacyWorkersLogs(flags()).pipe(Effect.flip); + + expect(error).toBeInstanceOf(WorkersUnavailableError); + }).pipe(Effect.provide(layer), Effect.ensuring(Effect.sync(repo.cleanup))); + }); + + it.live("reports an unexpected status, which is where a rejected query lands", () => { + const repo = project(); + const { layer } = setupLegacyWorkers({ + workdir: repo.dir, + routes: { [LOGS_ROUTE]: { status: 500, body: { message: "query rejected" } } }, + }); + + return Effect.gen(function* () { + const error = yield* legacyWorkersLogs(flags()).pipe(Effect.flip); + + expect(error).toBeInstanceOf(WorkersApiUnexpectedStatusError); + }).pipe(Effect.provide(layer), Effect.ensuring(Effect.sync(repo.cleanup))); + }); + + it.live("reports a transport failure", () => { + const repo = project(); + const { layer } = setupLegacyWorkers({ + workdir: repo.dir, + routes: { [LOGS_ROUTE]: { transportError: "connection reset" } }, + }); + + return Effect.gen(function* () { + const error = yield* legacyWorkersLogs(flags()).pipe(Effect.flip); + + expect(error).toBeInstanceOf(WorkersApiNetworkError); + }).pipe(Effect.provide(layer), Effect.ensuring(Effect.sync(repo.cleanup))); + }); + + it.live("rejects an impossible worker name before any request", () => { + const repo = project(); + const { layer, http } = setupLegacyWorkers({ workdir: repo.dir, routes: {} }); + + return Effect.gen(function* () { + const error = yield* legacyWorkersLogs(flags({ name: "Not A Name" })).pipe(Effect.flip); + + expect(error).toBeInstanceOf(InvalidWorkerNameError); + expect(http.requests).toHaveLength(0); + }).pipe(Effect.provide(layer), Effect.ensuring(Effect.sync(repo.cleanup))); + }); + + it.live("refuses -o env before spending the query", () => { + const repo = project(); + const { layer, http } = setupLegacyWorkers({ + workdir: repo.dir, + goOutput: "env", + routes: { [LOGS_ROUTE]: logsResponse([workerLogRow({})]) }, + }); + + return Effect.gen(function* () { + const error = yield* legacyWorkersLogs(flags()).pipe(Effect.flip); + + expect(error).toBeInstanceOf(LegacyWorkersEnvNotSupportedError); + expect(http.requests).toHaveLength(0); + }).pipe(Effect.provide(layer), Effect.ensuring(Effect.sync(repo.cleanup))); + }); + + it.live("emits only the payload on stdout for -o json", () => { + const repo = project(); + const { layer, out } = setupLegacyWorkers({ + workdir: repo.dir, + goOutput: "json", + routes: { + [LOGS_ROUTE]: logsResponse([ + workerIngressLogRow({ tsMs: T1, status: "503", durationMs: "12" }), + ]), + }, + }); + + return Effect.gen(function* () { + yield* legacyWorkersLogs(flags()); + + const payload = JSON.parse(out.stdoutText) as { + worker_name: string; + logs: ReadonlyArray>; + }; + expect(payload.worker_name).toBe("api"); + expect(payload.logs[0]?.level).toBe("error"); + // Both timestamp forms, and the raw attributes. + expect(payload.logs[0]?.timestamp).toBe(new Date(T1).toISOString()); + expect(payload.logs[0]?.timestamp_ms).toBe(T1); + const attributes = payload.logs[0]?.attributes as Record | undefined; + expect(attributes?.duration_ms).toBe("12"); + }).pipe(Effect.provide(layer), Effect.ensuring(Effect.sync(repo.cleanup))); + }); + + it.live("emits exactly one structured result for --output-format json", () => { + const repo = project(); + const { layer, out } = setupLegacyWorkers({ + workdir: repo.dir, + format: "json", + routes: { [LOGS_ROUTE]: logsResponse([workerLogRow({ tsMs: T1 })]) }, + }); + + return Effect.gen(function* () { + yield* legacyWorkersLogs(flags()); + + expect(out.stdoutText).toBe(""); + const results = out.messages.filter((message) => message.type === "success"); + expect(results).toHaveLength(1); + }).pipe(Effect.provide(layer), Effect.ensuring(Effect.sync(repo.cleanup))); + }); + + it.live("flushes telemetry even when the query fails", () => { + const repo = project(); + const { layer, telemetry } = setupLegacyWorkers({ + workdir: repo.dir, + routes: { [LOGS_ROUTE]: { status: 500 } }, + }); + + return Effect.gen(function* () { + yield* legacyWorkersLogs(flags()).pipe(Effect.ignore); + + expect(telemetry.flushed).toBe(true); + }).pipe(Effect.provide(layer), Effect.ensuring(Effect.sync(repo.cleanup))); + }); + + it.live("uses the project ref from the flag and echoes it in suggestions", () => { + const repo = project(); + const { layer } = setupLegacyWorkers({ + workdir: repo.dir, + linked: false, + routes: { + [`GET /v1/projects/${WORKERS_PROJECT_REF}/analytics/endpoints/logs`]: logsResponse([]), + [GET_WORKER_ROUTE]: { status: 404 }, + }, + }); + + return Effect.gen(function* () { + const error = yield* legacyWorkersLogs( + flags({ projectRef: Option.some(WORKERS_PROJECT_REF) }), + ).pipe(Effect.flip); + + // A copy-pasted suggestion must not silently re-resolve to whatever this + // checkout happens to be linked to. + const suggestion = error instanceof WorkerNotDeployedError ? error.suggestion : ""; + expect(suggestion).toContain(`--project-ref ${WORKERS_PROJECT_REF}`); + }).pipe(Effect.provide(layer), Effect.ensuring(Effect.sync(repo.cleanup))); + }); +}); diff --git a/apps/cli/src/legacy/commands/experimental/workers/workers-logs.format.ts b/apps/cli/src/legacy/commands/experimental/workers/workers-logs.format.ts new file mode 100644 index 0000000000..e948cf5e36 --- /dev/null +++ b/apps/cli/src/legacy/commands/experimental/workers/workers-logs.format.ts @@ -0,0 +1,160 @@ +import { legacyRed, legacyYellow, type LegacyColorStream } from "../../../shared/legacy-colors.ts"; +import { WORKER_LOG_STREAMS } from "../../../../shared/workers/worker-logs.sql.ts"; +import type { WorkerLogEntry } from "../../../../shared/workers/worker-logs-api.ts"; + +/** + * Text rendering for `supabase experimental workers logs`. + * + * Pure, like `workers.format.ts` beside it: no Effect, no services, so the line + * shapes and the level derivation are unit-testable directly. + * + * Kept apart from `workers.format.ts` because that file renders *resources* - a + * worker's details as a key/value block - while this renders a stream of events, + * one line each, with a different layout per stream. + */ + +export type WorkerLogLevel = "info" | "warn" | "error"; + +/** + * The level for one line, derived rather than read. + * + * `severity_text` on the row is not usable: every observed row of every stream + * carries `INFO`, including a 200 request log, so it is a pipeline default rather + * than a signal. Platform's own log presets do the same thing - they derive level + * from `log_attributes`, and no platform code branches on `severity_text`. + * + * Guest output has no level available without parsing tenant text, so it is + * reported absent rather than guessed at. + */ +export function legacyWorkerLogLevel(entry: WorkerLogEntry): WorkerLogLevel | undefined { + if (entry.stream === WORKER_LOG_STREAMS.requests) { + // `log_attributes` is a Map(String, String), so this is "200", not 200. + const status = Number(entry.attributes.status); + if (!Number.isFinite(status)) { + return undefined; + } + if (status >= 500) { + return "error"; + } + return status >= 400 ? "warn" : "info"; + } + if (entry.stream === WORKER_LOG_STREAMS.builds) { + return entry.attributes.event === "build_failed" ? "error" : "info"; + } + return undefined; +} + +/** + * The escape-sequence and control-character patterns stripped from a guest line. + * + * Module constants so they compile once rather than per line, and so the + * `no-control-regex` suppression sits in one place: matching control characters is + * the entire purpose here, and every pattern is written with Unicode escapes so + * the source itself holds no raw control bytes. + */ +/* oxlint-disable no-control-regex */ +/** OSC: ESC ] ... terminated by BEL or ESC backslash. */ +const OSC_SEQUENCE = /\u001b\][^\u0007\u001b]*(?:\u0007|\u001b\\)?/gu; +/** CSI: ESC [ parameters intermediates final. */ +const CSI_SEQUENCE = /\u001b\[[0-9;?]*[ -/]*[@-~]/gu; +/** Remaining two-character escape sequences. */ +const ESCAPE_SEQUENCE = /\u001b[@-Z\\-_]/gu; +/** Leftover C0 controls and DEL, keeping tab, newline and carriage return. */ +const C0_CONTROLS = /[\u0000-\u0008\u000b\u000c\u000e-\u001f\u007f]/gu; +/* oxlint-enable no-control-regex */ + +/** + * Control characters stripped from a line before it reaches a terminal. + * + * A `worker_guest_logs` message is bytes the tenant's own code printed, so it is + * the one untrusted string this CLI displays: left alone, a worker could emit + * ANSI escapes that reposition the cursor, recolour later output, or forge a line + * that looks like the CLI's own. + * + * Deliberately narrow. Tabs, and the interior newlines and indentation of a stack + * trace, are content the reader needs - only escape sequences and the other C0 + * controls go. + */ +function stripControlSequences(message: string): string { + return message + .replaceAll(OSC_SEQUENCE, "") + .replaceAll(CSI_SEQUENCE, "") + .replaceAll(ESCAPE_SEQUENCE, "") + .replaceAll(C0_CONTROLS, ""); +} + +/** + * Colour for a level, or plain text. + * + * The stream is threaded through rather than a boolean because + * `legacyAqua`/`legacyRed`/... already own the colour decision: they consult + * `NO_COLOR`, `CLICOLOR`, `CLICOLOR_FORCE`, `CI` and the stream's own + * `hasColors()`. Deciding here - from `isTTY`, say - would both duplicate that + * gate and get it wrong, since `CLICOLOR_FORCE=1` deliberately styles a piped + * stream. + * + * Only `warn` and `error` are coloured. `info` is the overwhelming majority of + * lines, and tinting all of them would make the exceptions harder to spot, not + * easier. + */ +function colourise( + text: string, + level: WorkerLogLevel | undefined, + stream: LegacyColorStream, +): string { + if (level === "error") { + return legacyRed(text, stream); + } + return level === "warn" ? legacyYellow(text, stream) : text; +} + +/** + * `HH:MM:SS` in UTC. + * + * Time only, not a full timestamp: every line in one invocation falls inside a + * window of at most a day, so repeating the date on all hundred of them costs + * width the message needs. The machine payload carries the full ISO string and + * the raw epoch value. + */ +function formatLogTime(timestampMs: number): string { + return new Date(timestampMs).toISOString().slice(11, 19); +} + +/** + * One rendered line. + * + * Per-stream layouts rather than one shared format, because `event_message` means + * something different in each. On the request stream it is only `"GET /"` - the + * status and duration live in `log_attributes` - so the useful line has to be + * *composed*, and a single format wide enough for all three would be mostly empty + * for each of them. + * + * An unrecognised stream falls back to the bare message: the log contract is + * additive-only, so a stream this CLI has not heard of must still print. + */ +export function legacyRenderWorkerLogLine( + entry: WorkerLogEntry, + stream: LegacyColorStream = process.stdout, +): string { + const time = formatLogTime(entry.timestampMs); + const level = legacyWorkerLogLevel(entry); + + if (entry.stream === WORKER_LOG_STREAMS.requests) { + const { status, method, path, duration_ms: duration } = entry.attributes; + const request = [status, method, path].filter((part) => part !== undefined).join(" "); + const suffix = duration === undefined ? "" : ` ${duration}ms`; + return `${time} ${colourise(`${request}${suffix}`, level, stream)}`; + } + + if (entry.stream === WORKER_LOG_STREAMS.builds) { + const { event, reason } = entry.attributes; + const described = [event ?? entry.message, reason] + .filter((part) => part !== undefined) + .join(" "); + return `${time} ${colourise(described, level, stream)}`; + } + + // Guest output, and anything newer. The message is the payload, and it is the + // untrusted one. + return `${time} ${colourise(stripControlSequences(entry.message), level, stream)}`; +} diff --git a/apps/cli/src/legacy/commands/experimental/workers/workers-logs.format.unit.test.ts b/apps/cli/src/legacy/commands/experimental/workers/workers-logs.format.unit.test.ts new file mode 100644 index 0000000000..5905c0e4db --- /dev/null +++ b/apps/cli/src/legacy/commands/experimental/workers/workers-logs.format.unit.test.ts @@ -0,0 +1,199 @@ +import { describe, expect, it } from "@effect/vitest"; +import type { WorkerLogEntry } from "../../../../shared/workers/worker-logs-api.ts"; +import { legacyRenderWorkerLogLine, legacyWorkerLogLevel } from "./workers-logs.format.ts"; + +const ESCAPE = "\u001b"; + +/** + * Colour is decided by the stream, so the tests supply one. This keeps them + * independent of ambient NO_COLOR / CI / TTY state, and lets the coloured + * rendering be asserted at all rather than only its absence. + */ +const PLAIN = { hasColors: () => false }; +const COLOURED = { hasColors: () => true }; +const AT = Date.parse("2026-08-31T14:45:32.576Z"); + +function entry(overrides: Partial = {}): WorkerLogEntry { + return { + id: "row-1", + timestampMs: AT, + message: "workers shim: listening on :8080 (serving)", + stream: "worker_guest_logs", + attributes: { source: "worker_guest_logs", worker: "api" }, + ...overrides, + }; +} + +describe("legacyWorkerLogLevel", () => { + it("derives the level from a request status, which arrives as a string", () => { + const at = (status: string) => + legacyWorkerLogLevel(entry({ stream: "worker_ingress_logs", attributes: { status } })); + + expect(at("200")).toBe("info"); + expect(at("301")).toBe("info"); + expect(at("404")).toBe("warn"); + expect(at("499")).toBe("warn"); + expect(at("500")).toBe("error"); + expect(at("503")).toBe("error"); + }); + + it("reports no level for a request row with an unusable status", () => { + expect( + legacyWorkerLogLevel(entry({ stream: "worker_ingress_logs", attributes: {} })), + ).toBeUndefined(); + expect( + legacyWorkerLogLevel(entry({ stream: "worker_ingress_logs", attributes: { status: "wat" } })), + ).toBeUndefined(); + }); + + it("marks a failed build as an error and other events as info", () => { + expect( + legacyWorkerLogLevel( + entry({ stream: "worker_api_logs", attributes: { event: "build_failed" } }), + ), + ).toBe("error"); + expect( + legacyWorkerLogLevel( + entry({ stream: "worker_api_logs", attributes: { event: "deploy_accepted" } }), + ), + ).toBe("info"); + }); + + it("reports no level for guest output rather than guessing one", () => { + // Nothing short of parsing tenant text could tell, so absent is the honest + // answer. + expect(legacyWorkerLogLevel(entry())).toBeUndefined(); + }); + + it("reports no level for an unknown stream", () => { + expect(legacyWorkerLogLevel(entry({ stream: "worker_future_logs" }))).toBeUndefined(); + }); +}); + +describe("legacyRenderWorkerLogLine", () => { + it("prints the time and message for guest output", () => { + expect(legacyRenderWorkerLogLine(entry(), PLAIN)).toBe( + "14:45:32 workers shim: listening on :8080 (serving)", + ); + }); + + it("composes the request line from attributes, not the message", () => { + // On the wire `event_message` is only "GET /" - status and duration live in + // log_attributes, so the useful line has to be assembled. + const line = legacyRenderWorkerLogLine( + entry({ + stream: "worker_ingress_logs", + message: "GET /", + attributes: { status: "200", method: "GET", path: "/", duration_ms: "23" }, + }), + PLAIN, + ); + + expect(line).toBe("14:45:32 200 GET / 23ms"); + }); + + it("prints the event and reason for a build line", () => { + const line = legacyRenderWorkerLogLine( + entry({ + stream: "worker_api_logs", + message: "build_failed ref/api", + attributes: { event: "build_failed", reason: "exit status 1" }, + }), + PLAIN, + ); + + expect(line).toBe("14:45:32 build_failed exit status 1"); + }); + + it("falls back to the message for an unknown stream", () => { + // The log contract is additive-only, so a new stream must still print. + const line = legacyRenderWorkerLogLine( + entry({ stream: "worker_future_logs", message: "something new" }), + PLAIN, + ); + + expect(line).toBe("14:45:32 something new"); + }); + + it("renders a blank guest line as a blank line, not a dropped entry", () => { + expect(legacyRenderWorkerLogLine(entry({ message: "" }), PLAIN)).toBe("14:45:32 "); + }); + + it("strips ANSI escapes a worker printed, so it cannot forge output", () => { + const line = legacyRenderWorkerLogLine( + entry({ message: `${ESCAPE}[31mfake error${ESCAPE}[0m` }), + PLAIN, + ); + + expect(line).toBe("14:45:32 fake error"); + expect(line).not.toContain(ESCAPE); + }); + + it("strips a cursor-repositioning sequence", () => { + const line = legacyRenderWorkerLogLine( + entry({ message: `${ESCAPE}[2A${ESCAPE}[1Goverwritten` }), + PLAIN, + ); + + expect(line).toBe("14:45:32 overwritten"); + }); + + it("strips an OSC window-title sequence", () => { + const line = legacyRenderWorkerLogLine( + entry({ message: `${ESCAPE}]0;title${ESCAPE}\\kept` }), + PLAIN, + ); + + expect(line).toBe("14:45:32 kept"); + }); + + it("keeps a stack trace's newlines and indentation intact", () => { + const trace = "TypeError: boom\n at handler (index.js:3:11)\n\tat run (index.js:9:2)"; + + expect(legacyRenderWorkerLogLine(entry({ message: trace }), PLAIN)).toBe(`14:45:32 ${trace}`); + }); + + it("tints an error line red and a warning yellow, on the message only", () => { + const server = entry({ + stream: "worker_ingress_logs", + attributes: { status: "500", method: "GET", path: "/" }, + }); + const client = entry({ + stream: "worker_ingress_logs", + attributes: { status: "404", method: "GET", path: "/" }, + }); + + const errorLine = legacyRenderWorkerLogLine(server, COLOURED); + const warnLine = legacyRenderWorkerLogLine(client, COLOURED); + + // The timestamp stays plain so nothing a script greps on changes colour. + expect(errorLine.startsWith("14:45:32 ")).toBe(true); + expect(warnLine.startsWith("14:45:32 ")).toBe(true); + expect(errorLine).toContain(`${ESCAPE}[31m`); + expect(warnLine).toContain(`${ESCAPE}[33m`); + }); + + it("leaves an info line untinted, so the exceptions stand out", () => { + const line = legacyRenderWorkerLogLine( + entry({ + stream: "worker_ingress_logs", + attributes: { status: "200", method: "GET", path: "/" }, + }), + COLOURED, + ); + + expect(line).not.toContain(ESCAPE); + }); + + it("emits no escapes at all for a stream that cannot colour", () => { + const line = legacyRenderWorkerLogLine( + entry({ + stream: "worker_ingress_logs", + attributes: { status: "500", method: "GET", path: "/" }, + }), + PLAIN, + ); + + expect(line).not.toContain(ESCAPE); + }); +}); diff --git a/apps/cli/src/legacy/commands/experimental/workers/workers.command.ts b/apps/cli/src/legacy/commands/experimental/workers/workers.command.ts index b5b536fdb7..dc4601a582 100644 --- a/apps/cli/src/legacy/commands/experimental/workers/workers.command.ts +++ b/apps/cli/src/legacy/commands/experimental/workers/workers.command.ts @@ -1,6 +1,7 @@ import { Command } from "effect/unstable/cli"; import { legacyWorkersDeleteCommand } from "./delete/delete.command.ts"; import { legacyWorkersListCommand } from "./list/list.command.ts"; +import { legacyWorkersLogsCommand } from "./logs/logs.command.ts"; import { legacyWorkersNewCommand } from "./new/new.command.ts"; import { legacyWorkersPushCommand } from "./push/push.command.ts"; import { legacyWorkersStatusCommand } from "./status/status.command.ts"; @@ -15,6 +16,7 @@ export const legacyWorkersCommand = Command.make("workers").pipe( legacyWorkersPushCommand, legacyWorkersListCommand, legacyWorkersStatusCommand, + legacyWorkersLogsCommand, legacyWorkersDeleteCommand, ]), ); diff --git a/apps/cli/src/legacy/shared/legacy-db-target-flags.ts b/apps/cli/src/legacy/shared/legacy-db-target-flags.ts index 963e9b295e..2f5a55fa03 100644 --- a/apps/cli/src/legacy/shared/legacy-db-target-flags.ts +++ b/apps/cli/src/legacy/shared/legacy-db-target-flags.ts @@ -146,6 +146,7 @@ export const VALUE_CONSUMING_LONG_FLAGS = new Set([ "status", "sub", "swift-access-control", + "tail", "template", "timestamp", "to", diff --git a/apps/cli/src/shared/workers/worker-logs-api.ts b/apps/cli/src/shared/workers/worker-logs-api.ts new file mode 100644 index 0000000000..21ee21f76c --- /dev/null +++ b/apps/cli/src/shared/workers/worker-logs-api.ts @@ -0,0 +1,200 @@ +import { operationDefinitions, type ApiClient } from "@supabase/api/effect"; +import { Effect, Option, Predicate, Schema } from "effect"; +import { decodeBody, mapRequestError, unexpectedStatus } from "./workers-api-status.ts"; +import { + WorkerLogsQueryFailedError, + WorkerLogsRateLimitedError, + WorkerLogsUsageExceededError, + WorkersUnavailableError, +} from "./workers.errors.ts"; +import { workerLogsQuery } from "./worker-logs.sql.ts"; + +/** + * Reading a worker's logs, over the project's unified logs stream: + * `GET /v1/projects/{ref}/analytics/endpoints/logs`. + * + * Not `/v2/projects/{ref}/workers/...` like the rest of the family — there is no + * worker-scoped log route — so this is the one Workers seam that talks to the + * analytics API, and the one that has to reckon with its two quirks: the query is + * SQL this CLI writes, and a failed query can arrive as **HTTP 200 with an + * `error` field**. + */ + +/** One log line, flattened out of the untyped `result` array. */ +export interface WorkerLogEntry { + /** Logflare-minted. The dedupe key for an overlapping-window poll. */ + readonly id: string; + /** Epoch milliseconds. Order on this — guest lines arrive out of order. */ + readonly timestampMs: number; + /** + * Display text only, never a parse target. On the guest stream it is + * tenant-controlled bytes and must be escaped before it reaches a terminal. + */ + readonly message: string; + /** `worker_guest_logs` | `worker_ingress_logs` | `worker_api_logs`, or newer. */ + readonly stream: string; + /** + * `log_attributes`, which is a `Map(String, String)` — so `status` arrives as + * `"200"` and `duration_ms` as `"23"`. Coerce before comparing. + */ + readonly attributes: Readonly>; +} + +/** + * The row shape the projection in `workerLogsQuery` produces. + * + * `stream` stays a plain string and `log_attributes` an open record because the + * log contract is additive-only: a new stream or a new attribute must render, + * not fail the whole read. A closed `Schema.Literal` union here would break the + * command the next time a stream is added. + */ +const WorkerLogRow = Schema.Struct({ + id: Schema.String, + ts_ms: Schema.Number, + stream: Schema.String, + event_message: Schema.String, + log_attributes: Schema.Record(Schema.String, Schema.String), +}); + +/** The endpoint's structured error shape, when it is not a bare string. */ +const StructuredLogError = Schema.Struct({ + message: Schema.String, +}); + +/** + * The response envelope, declared here rather than reusing the generated + * `V1GetProjectLogsOutput`. + * + * The generated schema is `optionalKey` on both fields but allows neither to be + * `null` — while the endpoint sends exactly `{"result":[...],"error":null}` on + * success and `{"result":null,"error":"..."}` on failure, because + * `getAnalyticsResponse` normalises the unused half to an explicit `null`. Decoding + * a real response against the generated schema therefore always fails. + * + * `error` stays `Unknown` so the string and structured forms are both accepted and + * narrowed at the point of use; the generated struct also marks fields required + * that real bodies omit. + */ +const LogsResponse = Schema.Struct({ + result: Schema.optionalKey(Schema.NullOr(Schema.Array(Schema.Unknown))), + error: Schema.optionalKey(Schema.NullOr(Schema.Unknown)), +}); + +/** + * Renders the endpoint's `error` field, which is `string | {code, errors[], message, status}`. + * + * Narrowed through the schema rather than a `typeof` chain so the structured + * shape is checked rather than assumed. + */ +const describeLogError = Effect.fnUntraced(function* (error: unknown) { + if (Predicate.isString(error)) { + return error; + } + const structured = yield* Schema.decodeUnknownEffect(StructuredLogError)(error).pipe( + Effect.option, + ); + return Option.isSome(structured) ? structured.value.message : JSON.stringify(error); +}); + +export const fetchWorkerLogs = Effect.fnUntraced(function* ( + api: ApiClient, + projectRef: string, + options: { + readonly name: string; + readonly streams: ReadonlyArray; + readonly tail: number; + /** Both bounds, always — see `logWindow`. */ + readonly window: { readonly start: string; readonly end: string }; + }, +) { + const operation = `read logs for worker "${options.name}"`; + const sql = workerLogsQuery({ + name: options.name, + streams: options.streams, + tail: options.tail, + }); + + const response = yield* api + .executeRaw(operationDefinitions.v1GetProjectLogs, { + ref: projectRef, + sql, + iso_timestamp_start: options.window.start, + iso_timestamp_end: options.window.end, + }) + .pipe(Effect.mapError(mapRequestError(operation))); + + if (response.status === 402) { + return yield* Effect.fail( + new WorkerLogsUsageExceededError({ + detail: `The log query allowance for project ${projectRef} is exhausted.`, + suggestion: "Enable additional usage for this project in the dashboard, then retry.", + }), + ); + } + // The analytics endpoints allow 10 requests per 60 seconds, which is well + // inside what a tight `--follow` poll would spend. + if (response.status === 429) { + return yield* Effect.fail( + new WorkerLogsRateLimitedError({ + detail: "The logs API is rate limiting this project (10 requests per minute).", + suggestion: "Wait a minute before retrying, and avoid running several tails at once.", + }), + ); + } + // The route gates on the same private-alpha allow-list as the rest of the + // family, and answers 404 for a project outside it. + if (response.status === 404) { + return yield* Effect.fail( + new WorkersUnavailableError({ + detail: `Logs are not available for project ${projectRef}.`, + suggestion: + "Workers are in private alpha. Ask in the Supabase dashboard to have this project enrolled.", + }), + ); + } + if (response.status !== 200) { + // A rejected query or the server's 30-second timeout lands here rather than + // in the 200-with-`error` branch below, so both paths have to exist. + return yield* unexpectedStatus({ + operation, + status: response.status, + body: yield* response.text.pipe(Effect.orElseSucceed(() => "")), + }); + } + + const body = yield* response.json.pipe(Effect.mapError(mapRequestError(operation))); + const decoded = yield* decodeBody(LogsResponse, operation, body, response.status); + + // Checked before `result`: this endpoint reports a failed query with a 200 and + // a populated `error`, so reading `result` first reports success on a failure. + if (decoded.error !== undefined && decoded.error !== null) { + const described = yield* describeLogError(decoded.error); + return yield* Effect.fail( + new WorkerLogsQueryFailedError({ + detail: `The logs API could not run this query: ${described}.`, + suggestion: "Retry shortly; if it persists, report it with `supabase issue`.", + }), + ); + } + + // `result` is optional in the contract, so absent, null and [] all mean "no + // rows" and must not be told apart. + const rows = decoded.result ?? []; + const entries: Array = []; + for (const row of rows) { + const parsed = yield* decodeBody(WorkerLogRow, operation, row, response.status); + entries.push({ + id: parsed.id, + timestampMs: parsed.ts_ms, + message: parsed.event_message, + stream: parsed.stream, + attributes: parsed.log_attributes, + }); + } + + // The query orders `desc` to make `limit` mean "the most recent N". Sorting + // here rather than trusting that order: guest lines are ingested late and out + // of order, and once `--follow` merges overlapping windows the server's order + // stops being meaningful at all. + return entries.sort((left, right) => left.timestampMs - right.timestampMs); +}); diff --git a/apps/cli/src/shared/workers/worker-logs.sql.ts b/apps/cli/src/shared/workers/worker-logs.sql.ts new file mode 100644 index 0000000000..855bdacc31 --- /dev/null +++ b/apps/cli/src/shared/workers/worker-logs.sql.ts @@ -0,0 +1,131 @@ +/** + * The ClickHouse query `supabase experimental workers logs` sends, and the two literals it + * turns on. + * + * Pure — no Effect, no services — so the query text and the window arithmetic are + * unit-testable without a stubbed API, and so the literals have exactly one home + * if the log pipeline ever renames a stream. + * + * Everything here was verified against a real project; see + * `scratch/FINDINGS-worker-logs.md` for the captured rows. + */ + +/** + * The three streams that share the Workers Logflare source, keyed by the word the + * `--source` flag exposes. + * + * `worker_guest_logs` is an internal name; `app` is what a user means. The + * mapping lives here rather than in the command so the flag and the query cannot + * drift apart. + */ +export const WORKER_LOG_STREAMS = { + app: "worker_guest_logs", + requests: "worker_ingress_logs", + builds: "worker_api_logs", +} as const; + +export type WorkerLogSourceChoice = keyof typeof WORKER_LOG_STREAMS; + +/** Every stream, for an invocation that named none. */ +export const ALL_WORKER_LOG_STREAMS: ReadonlyArray = Object.values(WORKER_LOG_STREAMS); + +/** + * Which `log_attributes` key carries the worker name. + * + * The Logflare writer stamps `metadata.worker`, and the whole metadata map lands + * in `log_attributes` on the ClickHouse side. + */ +const WORKER_LOG_NAME_ATTRIBUTE = "worker"; + +/** Which key carries the stream name. See {@link workerLogsQuery} for why. */ +const WORKER_LOG_STREAM_ATTRIBUTE = "source"; + +/** + * The server clamps a span of *more than* 24 hours, so the default window sits + * just under the boundary rather than on it. + * + * Being clamped is worse than being rejected: the server rewrites `end` to + * `start + 24h`, so an over-wide request silently returns an *older* slice than + * the one asked for. + */ +export const WORKER_LOG_WINDOW_MINUTES = 23 * 60 + 59; + +/** + * Timestamps for the endpoint's `iso_timestamp_start`/`iso_timestamp_end`. + * + * The v1 DTO validates these with `z.string().datetime()`, which requires a + * trailing `Z` and rejects numeric offsets — so this is `toISOString()` and must + * stay that way. + */ +export function isoLogTimestamp(date: Date): string { + return date.toISOString(); +} + +/** + * A closed window ending at `now`. + * + * Both bounds, always. Sending only a start yields a **one-minute** window + * server-side (the lone bound is minute-rounded and the other derived from it), + * and sending neither is an outright error — so there is no valid single-bound + * call to make. + */ +export function logWindow( + now: Date, + spanMinutes: number = WORKER_LOG_WINDOW_MINUTES, +): { readonly start: string; readonly end: string } { + return { + start: isoLogTimestamp(new Date(now.getTime() - spanMinutes * 60_000)), + end: isoLogTimestamp(now), + }; +} + +/** + * Single-quoted SQL string literal. + * + * Every value this module interpolates is either an internal constant or a name + * `legacyValidateWorkerName` has already reduced to a DNS label, so this is a + * backstop rather than the guard. It exists so the guarantee does not rest on a + * caller remembering to validate first. + */ +function quote(value: string): string { + return `'${value.replaceAll("'", "''")}'`; +} + +/** + * The logs query for one worker. + * + * Two things about the projection are load-bearing: + * + * - **The filter is `log_attributes`, not the `source` column.** Worker rows carry + * an empty top-level `source`, because the Workers Logflare source is not + * enrolled as a category in the generic logs path — so `where source = + * 'worker_guest_logs'` matches nothing. The stream survives only in + * `log_attributes['source']`. + * - **The `in (...)` list is a tenancy guard, not a convenience.** With `source` + * empty there is nothing else keeping a non-worker row that happens to carry a + * `worker` attribute out of the result. + * + * `toUnixTimestamp64Milli` rather than a formatter: ClickHouse's `%M` is the + * *month name*, and bare `toString(timestamp)` yields + * `2026-08-31 14:45:32.576000000` — space-separated, nine decimals, no zone. + * Epoch milliseconds have no such trap and sort as a number. + */ +export function workerLogsQuery(options: { + readonly name: string; + readonly streams: ReadonlyArray; + readonly tail: number; +}): string { + const streams = options.streams.map(quote).join(", "); + return ( + `select id, ` + + `toUnixTimestamp64Milli(timestamp) as ts_ms, ` + + `log_attributes['${WORKER_LOG_STREAM_ATTRIBUTE}'] as stream, ` + + `event_message, ` + + `log_attributes ` + + `from logs ` + + `where log_attributes['${WORKER_LOG_NAME_ATTRIBUTE}'] = ${quote(options.name)} ` + + `and log_attributes['${WORKER_LOG_STREAM_ATTRIBUTE}'] in (${streams}) ` + + `order by timestamp desc ` + + `limit ${options.tail}` + ); +} diff --git a/apps/cli/src/shared/workers/worker-logs.sql.unit.test.ts b/apps/cli/src/shared/workers/worker-logs.sql.unit.test.ts new file mode 100644 index 0000000000..3123a8b1c5 --- /dev/null +++ b/apps/cli/src/shared/workers/worker-logs.sql.unit.test.ts @@ -0,0 +1,119 @@ +import { describe, expect, it } from "@effect/vitest"; +import { validateWorkerNameMessage } from "./worker-runtimes.ts"; +import { + ALL_WORKER_LOG_STREAMS, + isoLogTimestamp, + logWindow, + WORKER_LOG_STREAMS, + WORKER_LOG_WINDOW_MINUTES, + workerLogsQuery, +} from "./worker-logs.sql.ts"; + +describe("workerLogsQuery", () => { + it("filters on log_attributes, never the source column", () => { + const sql = workerLogsQuery({ name: "api", streams: ALL_WORKER_LOG_STREAMS, tail: 100 }); + + expect(sql).toContain("log_attributes['worker'] = 'api'"); + expect(sql).toContain("log_attributes['source'] in ("); + // The load-bearing negative: worker rows carry an empty top-level `source`, + // so a predicate on that column matches nothing at all. + expect(sql).not.toMatch(/(?:^|\s)where source =/); + expect(sql).not.toMatch(/(?:^|\s)and source =/); + }); + + it("constrains to the known streams even when none was requested", () => { + const sql = workerLogsQuery({ name: "api", streams: ALL_WORKER_LOG_STREAMS, tail: 10 }); + + // With `source` empty this list is the only thing keeping a non-worker row + // that happens to carry a `worker` attribute out of the results. + expect(sql).toContain("'worker_guest_logs'"); + expect(sql).toContain("'worker_ingress_logs'"); + expect(sql).toContain("'worker_api_logs'"); + }); + + it("narrows to a single stream when one was requested", () => { + const sql = workerLogsQuery({ + name: "api", + streams: [WORKER_LOG_STREAMS.requests], + tail: 10, + }); + + expect(sql).toContain("in ('worker_ingress_logs')"); + expect(sql).not.toContain("worker_guest_logs"); + }); + + it("projects epoch milliseconds rather than a formatted timestamp", () => { + const sql = workerLogsQuery({ name: "api", streams: ALL_WORKER_LOG_STREAMS, tail: 1 }); + + expect(sql).toContain("toUnixTimestamp64Milli(timestamp) as ts_ms"); + // `%M` is ClickHouse's month name, and bare toString has no zone — neither + // belongs in this query. + expect(sql).not.toContain("formatDateTime"); + expect(sql).not.toContain("toString(timestamp)"); + }); + + it("orders newest first so limit means the most recent lines", () => { + const sql = workerLogsQuery({ name: "api", streams: ALL_WORKER_LOG_STREAMS, tail: 42 }); + + expect(sql).toContain("order by timestamp desc"); + expect(sql).toContain("limit 42"); + }); + + it("escapes a quote in the worker name", () => { + // Unreachable in practice — the handler validates first, see below — so this + // pins the backstop rather than the guard. + const sql = workerLogsQuery({ name: "a'b", streams: ALL_WORKER_LOG_STREAMS, tail: 1 }); + + expect(sql).toContain("log_attributes['worker'] = 'a''b'"); + }); +}); + +describe("worker name validation is the injection guard", () => { + it.each(["a'--", "a' or '1'='1", 'a"b', "a;drop", "a b"])("rejects %j", (name) => { + expect(validateWorkerNameMessage(name)).toBeDefined(); + }); + + it("accepts an ordinary DNS label", () => { + expect(validateWorkerNameMessage("say-hello")).toBeUndefined(); + }); +}); + +describe("logWindow", () => { + it("always returns both bounds", () => { + const window = logWindow(new Date("2026-08-31T12:00:00.000Z")); + + // A lone bound yields a one-minute window server-side, and sending neither is + // an outright error, so there is no valid single-bound call. + expect(window.start).toBeDefined(); + expect(window.end).toBeDefined(); + }); + + it("stays under the 24 hour span the server clamps at", () => { + const now = new Date("2026-08-31T12:00:00.000Z"); + const window = logWindow(now); + const spanMs = Date.parse(window.end) - Date.parse(window.start); + + // Being clamped is worse than being rejected: the server rewrites `end` to + // `start + 24h`, returning an older slice than the one asked for. + expect(spanMs).toBeLessThan(24 * 60 * 60 * 1000); + expect(WORKER_LOG_WINDOW_MINUTES).toBeLessThan(24 * 60); + }); + + it("ends at the given instant", () => { + const now = new Date("2026-08-31T12:00:00.000Z"); + + expect(logWindow(now).end).toBe("2026-08-31T12:00:00.000Z"); + }); +}); + +describe("isoLogTimestamp", () => { + it("emits a Z suffix and no numeric offset", () => { + // The v1 DTO validates with `z.string().datetime()`, which requires the Z and + // rejects `+00:00`. + const formatted = isoLogTimestamp(new Date("2026-08-31T12:00:00.000Z")); + + expect(formatted).toBe("2026-08-31T12:00:00.000Z"); + expect(formatted.endsWith("Z")).toBe(true); + expect(formatted).not.toContain("+"); + }); +}); diff --git a/apps/cli/src/shared/workers/workers-api-status.ts b/apps/cli/src/shared/workers/workers-api-status.ts new file mode 100644 index 0000000000..170f11c47a --- /dev/null +++ b/apps/cli/src/shared/workers/workers-api-status.ts @@ -0,0 +1,79 @@ +import { markSupabaseApiInputErrorAsUserInput, SupabaseApiInputError } from "@supabase/api/effect"; +import { Effect, Schema } from "effect"; +import * as HttpClientError from "effect/unstable/http/HttpClientError"; +import { WorkersApiNetworkError, WorkersApiUnexpectedStatusError } from "./workers.errors.ts"; + +/** + * Status handling shared by every Workers API seam. + * + * Hoisted out of `workers-api.ts` when `worker-logs-api.ts` needed the same + * three helpers verbatim: the worker routes and the analytics logs endpoint sit + * on different API families but fail the same three ways — the request never + * left, the server answered something unexpected, or the body could not be read. + * + * Route-specific status meaning stays with its route. `projectScoped404` is the + * example: it disambiguates a `/v2/workers` 404 by response body and means + * nothing anywhere else, so it did not come along. + */ + +/** + * Everything that can go wrong before a status code exists: the generated input + * schema rejecting the request, or the transport failing outright. + */ +export function mapRequestError(operation: string) { + return (error: unknown) => { + if (error instanceof SupabaseApiInputError) { + // The only inputs these operations take are the resolved project ref and + // the prevalidated worker name, so a schema rejection is user-derived. + return markSupabaseApiInputErrorAsUserInput(error); + } + if (HttpClientError.isHttpClientError(error)) { + // `message` is the library's own rendering of the reason — its label, the + // description when there is one, and the method and URL that failed. + // These requests all go to the Management API, so that URL is safe to + // show and is the most useful thing in the sentence. + return new WorkersApiNetworkError({ + detail: `Could not reach the Workers API while trying to ${operation}: ${error.message}.`, + suggestion: "Check your network connection and retry.", + }); + } + return new WorkersApiNetworkError({ + detail: `Could not reach the Workers API while trying to ${operation}: ${String(error)}.`, + suggestion: "Check your network connection and retry.", + }); + }; +} + +export const unexpectedStatus = Effect.fnUntraced(function* (options: { + readonly operation: string; + readonly status: number; + readonly body: string; +}) { + const trimmed = options.body.trim(); + return yield* Effect.fail( + new WorkersApiUnexpectedStatusError({ + status: options.status, + detail: `The Workers API answered ${options.status} while trying to ${options.operation}${ + trimmed === "" ? "" : `: ${trimmed}` + }.`, + suggestion: "Retry shortly; if it persists, report it with `supabase issue`.", + }), + ); +}); + +export const decodeBody = ( + schema: Schema.Codec, + operation: string, + body: unknown, + status: number, +) => + Schema.decodeUnknownEffect(schema)(body).pipe( + Effect.mapError( + (error) => + new WorkersApiUnexpectedStatusError({ + status, + detail: `The Workers API returned a response this CLI could not read while trying to ${operation}: ${error.message}.`, + suggestion: "Update the CLI with `supabase update`, then retry.", + }), + ), + ); diff --git a/apps/cli/src/shared/workers/workers-api.ts b/apps/cli/src/shared/workers/workers-api.ts index 4bdec8bfc4..e557cde852 100644 --- a/apps/cli/src/shared/workers/workers-api.ts +++ b/apps/cli/src/shared/workers/workers-api.ts @@ -1,7 +1,5 @@ import { - markSupabaseApiInputErrorAsUserInput, operationDefinitions, - SupabaseApiInputError, V2CreateWorkerUploadOutput, V2DeployAWorkerOutput, V2GetAWorkerOutput, @@ -10,13 +8,11 @@ import { } from "@supabase/api/effect"; import { Effect, Option, Schedule, Schema } from "effect"; import * as HttpClient from "effect/unstable/http/HttpClient"; -import * as HttpClientError from "effect/unstable/http/HttpClientError"; import * as HttpClientRequest from "effect/unstable/http/HttpClientRequest"; +import { decodeBody, mapRequestError, unexpectedStatus } from "./workers-api-status.ts"; import { WorkerBuildTimeoutError, - WorkersApiNetworkError, WorkerProjectNotFoundError, - WorkersApiUnexpectedStatusError, WorkersUnavailableError, WorkerUploadFailedError, } from "./workers.errors.ts"; @@ -136,68 +132,6 @@ const projectScoped404 = Effect.fnUntraced(function* (options: { }); }); -/** - * Everything that can go wrong before a status code exists: the generated input - * schema rejecting the request, or the transport failing outright. - */ -function mapRequestError(operation: string) { - return (error: unknown) => { - if (error instanceof SupabaseApiInputError) { - // The only inputs these operations take are the resolved project ref and - // the prevalidated worker name, so a schema rejection is user-derived. - return markSupabaseApiInputErrorAsUserInput(error); - } - if (HttpClientError.isHttpClientError(error)) { - // `message` is the library's own rendering of the reason — its label, the - // description when there is one, and the method and URL that failed. - // These requests all go to the Management API, so that URL is safe to - // show and is the most useful thing in the sentence. - return new WorkersApiNetworkError({ - detail: `Could not reach the Workers API while trying to ${operation}: ${error.message}.`, - suggestion: "Check your network connection and retry.", - }); - } - return new WorkersApiNetworkError({ - detail: `Could not reach the Workers API while trying to ${operation}: ${String(error)}.`, - suggestion: "Check your network connection and retry.", - }); - }; -} - -const unexpectedStatus = Effect.fnUntraced(function* (options: { - readonly operation: string; - readonly status: number; - readonly body: string; -}) { - const trimmed = options.body.trim(); - return yield* Effect.fail( - new WorkersApiUnexpectedStatusError({ - status: options.status, - detail: `The Workers API answered ${options.status} while trying to ${options.operation}${ - trimmed === "" ? "" : `: ${trimmed}` - }.`, - suggestion: "Retry shortly; if it persists, report it with `supabase issue`.", - }), - ); -}); - -const decodeBody = ( - schema: Schema.Codec, - operation: string, - body: unknown, - status: number, -) => - Schema.decodeUnknownEffect(schema)(body).pipe( - Effect.mapError( - (error) => - new WorkersApiUnexpectedStatusError({ - status, - detail: `The Workers API returned a response this CLI could not read while trying to ${operation}: ${error.message}.`, - suggestion: "Update the CLI with `supabase update`, then retry.", - }), - ), - ); - export const listWorkers = Effect.fnUntraced(function* (api: ApiClient, projectRef: string) { const operation = "list workers"; const response = yield* api diff --git a/apps/cli/src/shared/workers/workers.errors.ts b/apps/cli/src/shared/workers/workers.errors.ts index eda518866c..8d9ba34515 100644 --- a/apps/cli/src/shared/workers/workers.errors.ts +++ b/apps/cli/src/shared/workers/workers.errors.ts @@ -263,3 +263,49 @@ export class WorkerDeleteConfirmationRequiredError extends Data.TaggedError( return actionability.provideFlags; } } + +/** + * The logs query itself failed. + * + * The analytics endpoint can answer **HTTP 200** with a populated `error` field, + * so this is not reachable from a status code alone. It also covers the server's + * 30-second query timeout, which arrives as a non-2xx. + * + * Classified apart from {@link WorkersApiUnexpectedStatusError} on purpose: the + * SQL is this CLI's, not the user's input, so a rejected query means a projection + * or a filter here is wrong. Its own fingerprint keeps that visible in telemetry + * instead of grouped with transport noise from every other Workers route. + */ +export class WorkerLogsQueryFailedError extends Data.TaggedError("WorkerLogsQueryFailedError")<{ + readonly detail: string; + readonly suggestion: string; +}> { + get [ErrorActionabilityId](): CliErrorActionabilityDeclaration { + return { ...actionability.apiStatus, fingerprint_suffix: "query" }; + } +} + +/** The project has exhausted its log query allowance (402). */ +export class WorkerLogsUsageExceededError extends Data.TaggedError("WorkerLogsUsageExceededError")<{ + readonly detail: string; + readonly suggestion: string; +}> { + get [ErrorActionabilityId](): CliErrorActionabilityDeclaration { + return { ...actionability.planLimit, fingerprint_suffix: "plan_limit" }; + } +} + +/** + * The analytics endpoints allow 10 requests per 60 seconds, which `--follow` + * polls against — so 429 is an ordinary outcome here rather than an edge case, + * and it gets its own error so the suggestion can name the poll interval as the + * thing to slow down. + */ +export class WorkerLogsRateLimitedError extends Data.TaggedError("WorkerLogsRateLimitedError")<{ + readonly detail: string; + readonly suggestion: string; +}> { + get [ErrorActionabilityId](): CliErrorActionabilityDeclaration { + return { ...actionability.apiStatus, fingerprint_suffix: "api_status" }; + } +} diff --git a/apps/cli/tests/helpers/legacy-workers.ts b/apps/cli/tests/helpers/legacy-workers.ts index ded297864d..e308f6f576 100644 --- a/apps/cli/tests/helpers/legacy-workers.ts +++ b/apps/cli/tests/helpers/legacy-workers.ts @@ -33,6 +33,14 @@ export const WORKERS_PROJECT_REF = "abcdefghijklmnopqrst"; export interface RecordedRequest { readonly method: string; readonly url: string; + /** + * Query parameters, which `url` does not carry. + * + * `HttpClientRequest` keeps `urlParams` beside the URL rather than appended to + * it, so a test asserting what a GET actually asked for has to read this. The + * analytics logs endpoint puts the whole SQL query here. + */ + readonly urlParams: Readonly>; /** The request body decoded as UTF-8 — meaningful for the JSON requests. */ readonly body: string; /** Byte length of the body, which is what matters for the binary upload. */ @@ -107,6 +115,8 @@ export function mockWorkersHttp(routes: WorkersHttpRoutes) { requests.push({ method: request.method, url: request.url, + // UrlParams is iterable over [key, value] pairs, not an array. + urlParams: Object.fromEntries(request.urlParams), body: new TextDecoder().decode(bytes), byteLength: bytes.length, }); @@ -200,6 +210,94 @@ export function workerResource(options: { export const workersRoute = (suffix = "") => `/v2/projects/${WORKERS_PROJECT_REF}/workers${suffix}`; +/** + * The unified logs endpoint `workers logs` queries. Not under `/v2/.../workers` — + * there is no worker-scoped log route. + */ +export const workerLogsRoute = () => `/v1/projects/${WORKERS_PROJECT_REF}/analytics/endpoints/logs`; + +/** + * One row as the logs endpoint returns it, matching the projection in + * `workerLogsQuery`. + * + * Shaped from rows captured off a real project (see + * `scratch/FINDINGS-worker-logs.md`), which is why `log_attributes` values are + * all strings: the column is a `Map(String, String)`, so `status` really does + * arrive as `"200"`. + */ +export function workerLogRow(options: { + readonly id?: string; + readonly tsMs?: number; + readonly stream?: string; + readonly message?: string; + readonly worker?: string; + readonly attributes?: Readonly>; +}) { + const stream = options.stream ?? "worker_guest_logs"; + return { + id: options.id ?? "row-1", + ts_ms: options.tsMs ?? 1_788_187_532_576, + stream, + event_message: options.message ?? "workers shim: listening on :8080 (serving)", + log_attributes: { + source: stream, + worker: options.worker ?? "api", + project: WORKERS_PROJECT_REF, + ...options.attributes, + }, + }; +} + +/** An HTTP access log row, whose fields live in `log_attributes`, not the message. */ +export function workerIngressLogRow(options: { + readonly id?: string; + readonly tsMs?: number; + readonly worker?: string; + readonly status?: string; + readonly method?: string; + readonly path?: string; + readonly durationMs?: string; +}) { + const method = options.method ?? "GET"; + const path = options.path ?? "/"; + return workerLogRow({ + ...(options.id === undefined ? {} : { id: options.id }), + ...(options.tsMs === undefined ? {} : { tsMs: options.tsMs }), + ...(options.worker === undefined ? {} : { worker: options.worker }), + stream: "worker_ingress_logs", + // Only method and path — status and duration are deliberately absent, as + // they are on the wire. + message: `${method} ${path}`, + attributes: { + method, + path, + status: options.status ?? "200", + duration_ms: options.durationMs ?? "23", + instance_id: "microvm-3f4b0c03-9310-3f72-940d-f56deeef795e", + }, + }); +} + +/** A build/deploy lifecycle row. */ +export function workerApiLogRow(options: { + readonly id?: string; + readonly tsMs?: number; + readonly worker?: string; + readonly event?: string; + readonly reason?: string; +}) { + const worker = options.worker ?? "api"; + const event = options.event ?? "deploy_accepted"; + return workerLogRow({ + ...(options.id === undefined ? {} : { id: options.id }), + ...(options.tsMs === undefined ? {} : { tsMs: options.tsMs }), + worker, + stream: "worker_api_logs", + message: `${event} ${WORKERS_PROJECT_REF}/${worker}`, + attributes: { event, ...(options.reason === undefined ? {} : { reason: options.reason }) }, + }); +} + /** A per-test temp project, optionally pre-seeded with files. */ export function makeWorkersProject(files: Readonly> = {}): { readonly dir: string; From c4080deffdd6a66964b76cf9ba6bc3c975a6c428 Mon Sep 17 00:00:00 2001 From: Matt Johnston Date: Mon, 31 Aug 2026 15:54:04 -0300 Subject: [PATCH 2/5] feat(workers logs): print timestamps in local time MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Matches the only other log-line format this shell prints — the `--debug` HTTP logger, which uses Go's `log.LstdFlags` (`legacy-debug-logger.layer.ts`). Someone reading a tail is asking "what just happened", and the answer gets compared against their own clock. Text output only. The machine payload keeps both unambiguous forms, so nothing that is parsed, sorted, or pasted into an issue depends on the reader's zone: `timestamp` stays ISO-8601 UTC and `timestamp_ms` the raw epoch value. The unit tests derive their expected prefix from the same instant with the same field accessors, rather than hardcoding one: a literal `"14:45:32"` would have passed only on a UTC machine. One case additionally pins the zone choice itself — asserting the output is *not* the UTC rendering — guarded so it stays meaningful on a UTC machine, where the two coincide. Verified green under `TZ=Asia/Tokyo`, `TZ=UTC`, and the ambient zone. --- .../experimental/workers/logs/SIDE_EFFECTS.md | 9 ++-- .../workers/workers-logs.format.ts | 19 +++++-- .../workers/workers-logs.format.unit.test.ts | 51 +++++++++++++++---- 3 files changed, 60 insertions(+), 19 deletions(-) diff --git a/apps/cli/src/legacy/commands/experimental/workers/logs/SIDE_EFFECTS.md b/apps/cli/src/legacy/commands/experimental/workers/logs/SIDE_EFFECTS.md index 613c88e4ce..e31e51fe67 100644 --- a/apps/cli/src/legacy/commands/experimental/workers/logs/SIDE_EFFECTS.md +++ b/apps/cli/src/legacy/commands/experimental/workers/logs/SIDE_EFFECTS.md @@ -96,16 +96,17 @@ redacted. No custom events. | Mode | stdout | stderr | | ----------------------------- | --------------------------------------------------------------------------------------------------- | --------------------------------------------------------- | -| text (default) | one line per entry, oldest first, per-stream layout; severity as colour | the spinner, and the `status` hint when there are no logs | +| text (default) | one line per entry, oldest first, per-stream layout; `HH:MM:SS` local time; severity as colour | the spinner, and the `status` hint when there are no logs | | `--output-format json` | one structured result carrying every entry | as above | | `--output-format stream-json` | the same result as a single terminal event | as above | | `-o json` / `yaml` / `toml` | the same payload in that encoding, and nothing else | as above | | `-o pretty` / `table` / `csv` | the text rendering — these fall through rather than encoding | as above | | `-o env` | refused before any request; the payload nests a `logs` array a flat `KEY=value` list cannot express | the error | -Machine payloads carry each entry's `id`, both `timestamp` (ISO) and -`timestamp_ms`, `stream`, `message`, the derived `level` when one exists, and the -raw `attributes` map — whose values are all strings, since the column is a +Text output prints the time in the reader's own timezone, matching the `--debug` +HTTP logger. Machine payloads carry the unambiguous forms instead — each entry's +`id`, both `timestamp` (ISO-8601 UTC) and `timestamp_ms` (raw epoch), `stream`, +`message`, the derived `level` when one exists, and the raw `attributes` map — whose values are all strings, since the column is a `Map(String, String)`. A `worker_guest_logs` message is bytes the tenant's own code printed. Control and diff --git a/apps/cli/src/legacy/commands/experimental/workers/workers-logs.format.ts b/apps/cli/src/legacy/commands/experimental/workers/workers-logs.format.ts index e948cf5e36..d2c36b5a46 100644 --- a/apps/cli/src/legacy/commands/experimental/workers/workers-logs.format.ts +++ b/apps/cli/src/legacy/commands/experimental/workers/workers-logs.format.ts @@ -108,16 +108,27 @@ function colourise( return level === "warn" ? legacyYellow(text, stream) : text; } +const pad2 = (value: number): string => String(value).padStart(2, "0"); + /** - * `HH:MM:SS` in UTC. + * `HH:MM:SS` in the reader's own timezone. + * + * Local rather than UTC, matching the only other log-line format this shell + * prints - the `--debug` HTTP logger, which uses Go's `log.LstdFlags` + * (`legacy-debug-logger.layer.ts`). Someone reading a tail is asking "what just + * happened", and the answer is compared against their own clock. + * + * Machine output keeps the unambiguous forms, so nothing that gets parsed, + * sorted, or pasted into an issue depends on the reader's zone: the payload + * carries an ISO-8601 UTC `timestamp` alongside the raw epoch `timestamp_ms`. * * Time only, not a full timestamp: every line in one invocation falls inside a * window of at most a day, so repeating the date on all hundred of them costs - * width the message needs. The machine payload carries the full ISO string and - * the raw epoch value. + * width the message needs. */ function formatLogTime(timestampMs: number): string { - return new Date(timestampMs).toISOString().slice(11, 19); + const at = new Date(timestampMs); + return `${pad2(at.getHours())}:${pad2(at.getMinutes())}:${pad2(at.getSeconds())}`; } /** diff --git a/apps/cli/src/legacy/commands/experimental/workers/workers-logs.format.unit.test.ts b/apps/cli/src/legacy/commands/experimental/workers/workers-logs.format.unit.test.ts index 5905c0e4db..69b050a31a 100644 --- a/apps/cli/src/legacy/commands/experimental/workers/workers-logs.format.unit.test.ts +++ b/apps/cli/src/legacy/commands/experimental/workers/workers-logs.format.unit.test.ts @@ -13,6 +13,22 @@ const PLAIN = { hasColors: () => false }; const COLOURED = { hasColors: () => true }; const AT = Date.parse("2026-08-31T14:45:32.576Z"); +/** + * The expected `HH:MM:SS` prefix for an instant, in this machine's zone. + * + * Derived rather than hardcoded: the renderer prints local time, so a literal + * `"14:45:32"` would pass only on a UTC machine and fail everywhere else. Written + * with the same field accessors the renderer uses, so what it pins is the format + * and the zone choice, not an arithmetic that could drift with the clock. + */ +function localTime(timestampMs: number): string { + const at = new Date(timestampMs); + const pad = (value: number) => String(value).padStart(2, "0"); + return `${pad(at.getHours())}:${pad(at.getMinutes())}:${pad(at.getSeconds())}`; +} + +const T = localTime(AT); + function entry(overrides: Partial = {}): WorkerLogEntry { return { id: "row-1", @@ -73,7 +89,7 @@ describe("legacyWorkerLogLevel", () => { describe("legacyRenderWorkerLogLine", () => { it("prints the time and message for guest output", () => { expect(legacyRenderWorkerLogLine(entry(), PLAIN)).toBe( - "14:45:32 workers shim: listening on :8080 (serving)", + `${T} workers shim: listening on :8080 (serving)`, ); }); @@ -89,7 +105,7 @@ describe("legacyRenderWorkerLogLine", () => { PLAIN, ); - expect(line).toBe("14:45:32 200 GET / 23ms"); + expect(line).toBe(`${T} 200 GET / 23ms`); }); it("prints the event and reason for a build line", () => { @@ -102,7 +118,7 @@ describe("legacyRenderWorkerLogLine", () => { PLAIN, ); - expect(line).toBe("14:45:32 build_failed exit status 1"); + expect(line).toBe(`${T} build_failed exit status 1`); }); it("falls back to the message for an unknown stream", () => { @@ -112,11 +128,24 @@ describe("legacyRenderWorkerLogLine", () => { PLAIN, ); - expect(line).toBe("14:45:32 something new"); + expect(line).toBe(`${T} something new`); + }); + + it("renders local time, not UTC", () => { + // Pinned against a fixed offset rather than the ambient zone, so the choice is + // asserted on a UTC machine too, where local and UTC would otherwise coincide. + const utc = new Date(AT).toISOString().slice(11, 19); + const offsetMinutes = new Date(AT).getTimezoneOffset(); + const line = legacyRenderWorkerLogLine(entry(), PLAIN); + + expect(line.startsWith(`${T} `)).toBe(true); + if (offsetMinutes !== 0) { + expect(line.startsWith(`${utc} `)).toBe(false); + } }); it("renders a blank guest line as a blank line, not a dropped entry", () => { - expect(legacyRenderWorkerLogLine(entry({ message: "" }), PLAIN)).toBe("14:45:32 "); + expect(legacyRenderWorkerLogLine(entry({ message: "" }), PLAIN)).toBe(`${T} `); }); it("strips ANSI escapes a worker printed, so it cannot forge output", () => { @@ -125,7 +154,7 @@ describe("legacyRenderWorkerLogLine", () => { PLAIN, ); - expect(line).toBe("14:45:32 fake error"); + expect(line).toBe(`${T} fake error`); expect(line).not.toContain(ESCAPE); }); @@ -135,7 +164,7 @@ describe("legacyRenderWorkerLogLine", () => { PLAIN, ); - expect(line).toBe("14:45:32 overwritten"); + expect(line).toBe(`${T} overwritten`); }); it("strips an OSC window-title sequence", () => { @@ -144,13 +173,13 @@ describe("legacyRenderWorkerLogLine", () => { PLAIN, ); - expect(line).toBe("14:45:32 kept"); + expect(line).toBe(`${T} kept`); }); it("keeps a stack trace's newlines and indentation intact", () => { const trace = "TypeError: boom\n at handler (index.js:3:11)\n\tat run (index.js:9:2)"; - expect(legacyRenderWorkerLogLine(entry({ message: trace }), PLAIN)).toBe(`14:45:32 ${trace}`); + expect(legacyRenderWorkerLogLine(entry({ message: trace }), PLAIN)).toBe(`${T} ${trace}`); }); it("tints an error line red and a warning yellow, on the message only", () => { @@ -167,8 +196,8 @@ describe("legacyRenderWorkerLogLine", () => { const warnLine = legacyRenderWorkerLogLine(client, COLOURED); // The timestamp stays plain so nothing a script greps on changes colour. - expect(errorLine.startsWith("14:45:32 ")).toBe(true); - expect(warnLine.startsWith("14:45:32 ")).toBe(true); + expect(errorLine.startsWith(`${T} `)).toBe(true); + expect(warnLine.startsWith(`${T} `)).toBe(true); expect(errorLine).toContain(`${ESCAPE}[31m`); expect(warnLine).toContain(`${ESCAPE}[33m`); }); From b7c98642ef628a0c30d34ac3408bccbb50372582 Mon Sep 17 00:00:00 2001 From: Matt Johnston Date: Mon, 31 Aug 2026 16:02:17 -0300 Subject: [PATCH 3/5] feat(workers logs): add `--follow` to keep printing new lines MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Opt-in, matching `workers push --wait`: long-running behaviour in this family is asked for, never defaulted. **The poll interval is set by the rate limit, not by responsiveness.** The v1 analytics endpoints allow 10 requests per 60 seconds, so the two-second poll a live tail suggests would spend the whole allowance in ten seconds. Six seconds is the arithmetic floor; ten leaves room for the history query, the deployed-worker check, and a retry in the same window. Measured at ~7 requests in the worst 60-second window. The interval is in `--follow`'s help text, because a 10-second tail is visibly not a live stream and would otherwise look broken. The cursor deliberately lags 60 seconds behind the newest line printed. Guest lines are relayed CloudWatch -> subscription filter -> Lambda -> Logflare and arrive late and out of order, so a cursor sitting on the newest timestamp would drop every straggler permanently. Overlap is therefore guaranteed and expected; dedupe on the Logflare-minted `id` is what makes it invisible, bounded so a long tail does not grow the set forever. `followWindow` clamps to the same sub-24h span as a bounded read, so a tail resumed after a laptop suspend cannot ask for a wider window — the server answers those by returning an *older* slice. Every poll sends both timestamp bounds. Advancing only `iso_timestamp_start` is the obvious implementation and is wrong: it yields a one-minute window. Output: - `-o json|yaml|toml` and `--output-format json` are refused up front, beside the `-o env` refusal and for the same reason — each promises one terminal payload and a tail has no last element. - `--output-format stream-json` emits one `log-entry` event per line instead of a single `result`, reusing the existing variant. `stream` splits error/warn to `stderr`; `source` separates backlog from live. - SIGINT exits 130, matching the local `supabase logs` command. - `--tail 0` skips the backlog and makes no history request, since the endpoint rejects `limit 0`. It also suppresses the not-deployed check, which would otherwise read "no rows" as "no worker" when no query was made at all. Both schedules are injectable, as `awaitWorkerBuild`'s are, so the cursor, dedupe and retry paths are tested without a wall clock. The SIGINT test forks the handler and synchronises on the mock's `awaitExit` — `exit` never returns, so the handler cannot be awaited. Stressed over five consecutive runs. --- .../experimental/workers/logs/SIDE_EFFECTS.md | 21 +- .../experimental/workers/logs/logs.command.ts | 28 ++- .../experimental/workers/logs/logs.handler.ts | 195 ++++++++++++++++-- .../workers/logs/logs.integration.test.ts | 185 ++++++++++++++++- .../experimental/workers/workers.errors.ts | 22 ++ .../cli/src/shared/workers/worker-logs.sql.ts | 51 +++++ apps/cli/tests/helpers/legacy-workers.ts | 13 +- 7 files changed, 497 insertions(+), 18 deletions(-) diff --git a/apps/cli/src/legacy/commands/experimental/workers/logs/SIDE_EFFECTS.md b/apps/cli/src/legacy/commands/experimental/workers/logs/SIDE_EFFECTS.md index e31e51fe67..8c12df3ba4 100644 --- a/apps/cli/src/legacy/commands/experimental/workers/logs/SIDE_EFFECTS.md +++ b/apps/cli/src/legacy/commands/experimental/workers/logs/SIDE_EFFECTS.md @@ -58,8 +58,18 @@ under 24 hours. This is not optional: ### Rate limits The v1 analytics endpoints allow **10 requests per 60 seconds**, and the server -applies a 30-second query timeout. One invocation spends one request, or two when -the result is empty. +applies a 30-second query timeout. One bounded invocation spends one request, or +two when the result is empty. + +`--follow` polls every **10 seconds** — 6 requests a minute, leaving room for the +history query, the deployed-worker check, and a retry inside the same window. The +interval is set by that limit, not by responsiveness: a 2-second poll would spend +the allowance in ten seconds. A 429 mid-tail is retried on a spaced schedule +rather than ending the tail. + +Each poll re-asks for a window starting 60 seconds behind the newest line already +printed, because guest lines arrive late and out of order. Overlap is therefore +guaranteed and is deduplicated on the Logflare-minted `id`. ## Exit Codes @@ -103,6 +113,13 @@ redacted. No custom events. | `-o pretty` / `table` / `csv` | the text rendering — these fall through rather than encoding | as above | | `-o env` | refused before any request; the payload nests a `logs` array a flat `KEY=value` list cannot express | the error | +With `--follow`, `stream-json` emits a `log-entry` event per line rather than one +terminal `result` — a tail has no last element. Its `stream` field is `stderr` when +the derived level is error or warn and `stdout` otherwise, and `source` separates +the initial backlog (`history`) from lines that arrived afterwards (`live`). +`--tail 0 --follow` skips the backlog entirely and makes no history request, since +the endpoint rejects `limit 0`. + Text output prints the time in the reader's own timezone, matching the `--debug` HTTP logger. Machine payloads carry the unambiguous forms instead — each entry's `id`, both `timestamp` (ISO-8601 UTC) and `timestamp_ms` (raw epoch), `stream`, diff --git a/apps/cli/src/legacy/commands/experimental/workers/logs/logs.command.ts b/apps/cli/src/legacy/commands/experimental/workers/logs/logs.command.ts index 44b7ef71fe..a3ffa25ffa 100644 --- a/apps/cli/src/legacy/commands/experimental/workers/logs/logs.command.ts +++ b/apps/cli/src/legacy/commands/experimental/workers/logs/logs.command.ts @@ -2,6 +2,7 @@ import { Argument, Command, Flag } from "effect/unstable/cli"; import type * as CliCommand from "effect/unstable/cli/Command"; import { withJsonErrorHandling } from "../../../../../shared/output/json-error-handling.ts"; import { legacyManagementApiRuntimeLayer } from "../../../../shared/legacy-management-api-runtime.layer.ts"; +import { WORKER_LOG_POLL_SECONDS } from "../../../../../shared/workers/worker-logs.sql.ts"; import { withLegacyCommandInstrumentation } from "../../../../telemetry/legacy-command-instrumentation.ts"; import { legacyWorkersLogs } from "./logs.handler.ts"; @@ -31,12 +32,24 @@ const config = { ), Flag.optional, ), + follow: Flag.boolean("follow").pipe( + Flag.withAlias("f"), + Flag.withDescription( + `Keep printing new lines until interrupted, polling every ${WORKER_LOG_POLL_SECONDS} seconds.`, + ), + // Required: `legacy-boolean-flag-defaults.unit.test.ts` walks the whole + // command tree and fails any bare boolean. `workers push --wait` shipped + // without one and broke plain `push` parsing. + Flag.withDefault(false), + ), tail: Flag.integer("tail").pipe( Flag.filter( (tail) => tail >= 0 && tail <= MAX_TAIL, (tail) => `Expected --tail between 0 and ${MAX_TAIL}, got ${tail}`, ), - Flag.withDescription("Number of log lines to print."), + Flag.withDescription( + "Number of log lines to print. Use 0 with --follow to skip history and print only new lines.", + ), Flag.withDefault(100), ), } as const; @@ -48,7 +61,10 @@ export const legacyWorkersLogsCommand = Command.make("logs", config).pipe( "Print a worker's recent logs: its own output, the HTTP requests it served, and its " + "deploy lifecycle events.\n\n" + "Covers the last 24 hours, which is the longest window the logs API will answer in one " + - "query. Lines are printed oldest first.", + "query. Lines are printed oldest first.\n\n" + + `Use --follow to keep printing new lines as they arrive. The logs API is rate limited, so ` + + `following polls every ${WORKER_LOG_POLL_SECONDS} seconds rather than continuously; new ` + + "lines can take that long to appear.", ), Command.withShortDescription("Show a worker's logs"), Command.withExamples([ @@ -60,6 +76,14 @@ export const legacyWorkersLogsCommand = Command.make("logs", config).pipe( command: "supabase experimental workers logs api --source requests --tail 20", description: "Print the 20 most recent HTTP requests the worker served", }, + { + command: "supabase experimental workers logs api --follow", + description: "Print recent logs, then keep printing new lines until interrupted", + }, + { + command: "supabase experimental workers logs api --tail 0 --follow", + description: "Skip the backlog and print only lines that arrive from now on", + }, ]), Command.withHandler((flags) => legacyWorkersLogs(flags).pipe( diff --git a/apps/cli/src/legacy/commands/experimental/workers/logs/logs.handler.ts b/apps/cli/src/legacy/commands/experimental/workers/logs/logs.handler.ts index c284fbdc9c..be94607ca0 100644 --- a/apps/cli/src/legacy/commands/experimental/workers/logs/logs.handler.ts +++ b/apps/cli/src/legacy/commands/experimental/workers/logs/logs.handler.ts @@ -1,4 +1,4 @@ -import { Effect, Option } from "effect"; +import { Effect, Option, Ref, Schedule } from "effect"; import { Output } from "../../../../../shared/output/output.service.ts"; import { emitSuccessTrailer } from "../../../../../shared/cli/success-trailer.ts"; import { legacyAqua } from "../../../../shared/legacy-colors.ts"; @@ -8,6 +8,8 @@ import { legacyWorkersProjectRefSuffix, } from "../workers.output.ts"; import { legacyRenderWorkerLogLine, legacyWorkerLogLevel } from "../workers-logs.format.ts"; +import { ProcessControl } from "../../../../../shared/runtime/process-control.service.ts"; +import { LegacyWorkersFollowNotSupportedError } from "../workers.errors.ts"; import { LegacyPlatformApi } from "../../../../auth/legacy-platform-api.service.ts"; import { fetchWorkerLogs, @@ -15,7 +17,9 @@ import { } from "../../../../../shared/workers/worker-logs-api.ts"; import { ALL_WORKER_LOG_STREAMS, + followWindow, logWindow, + WORKER_LOG_POLL_SECONDS, WORKER_LOG_STREAMS, type WorkerLogSourceChoice, } from "../../../../../shared/workers/worker-logs.sql.ts"; @@ -25,6 +29,7 @@ import { LegacyProjectRefResolver } from "../../../../config/legacy-project-ref. import { LegacyLinkedProjectCache } from "../../../../telemetry/legacy-linked-project-cache.service.ts"; import { LegacyTelemetryState } from "../../../../telemetry/legacy-telemetry-state.service.ts"; import { legacyValidateWorkerName } from "../workers.shared.ts"; +import { legacyWorkersMachineOutputRequested } from "../workers.output.ts"; import type { LegacyWorkersLogsFlags } from "./logs.command.ts"; /** @@ -40,6 +45,38 @@ import type { LegacyWorkersLogsFlags } from "./logs.command.ts"; * rather than the `source` column. */ +/** + * How many printed ids the follow loop remembers. + * + * Only lines inside the cursor's grace window can still be re-offered by a later + * poll, so a bound well above one window's worth cannot cause a repeat while + * keeping the set from growing for the lifetime of a long tail. + */ +const SEEN_ID_LIMIT = 5000; + +/** + * How long one poll may keep failing before the tail gives up. + * + * Bounded by elapsed time rather than attempts, and spaced, so a 429 or a + * momentary blip is ridden out without spending the rate limit on retries. Same + * reasoning as `awaitWorkerBuild`'s read retry. + */ +const FOLLOW_READ_RETRY = Schedule.spaced("5 seconds").pipe( + Schedule.upTo({ duration: "1 minute" }), +); + +/** + * Test seams for the follow loop. + * + * Both schedules are parameters for the same reason `awaitWorkerBuild`'s are: the + * real ones are spaced in seconds, and a test exercising the cursor, the dedupe, + * or the retry path should not wait on a wall clock to do it. + */ +export interface LegacyWorkersLogsOptions { + readonly pollSchedule?: Schedule.Schedule; + readonly retrySchedule?: Schedule.Schedule; +} + /** The machine-format row for one line. */ function toPayloadEntry(entry: WorkerLogEntry) { const level = legacyWorkerLogLevel(entry); @@ -58,12 +95,14 @@ function toPayloadEntry(entry: WorkerLogEntry) { export const legacyWorkersLogs = Effect.fn("legacy.experimental.workers.logs")(function* ( flags: LegacyWorkersLogsFlags, + options: LegacyWorkersLogsOptions = {}, ) { const output = yield* Output; const api = yield* LegacyPlatformApi; const resolver = yield* LegacyProjectRefResolver; const linkedProjectCache = yield* LegacyLinkedProjectCache; const telemetryState = yield* LegacyTelemetryState; + const processControl = yield* ProcessControl; // The ref is resolved outside the finalizers because caching it is one of them; // everything that can fail on its own belongs inside, so those failures still @@ -79,6 +118,60 @@ export const legacyWorkersLogs = Effect.fn("legacy.experimental.workers.logs")(f // means failing after the query has already been paid for. yield* legacyRejectWorkersEnvOutput(); + // Also up front: a tail has no single terminal payload, so the bounded + // machine formats cannot express it. `stream-json` can, and is allowed. + if (flags.follow) { + const machineOutput = yield* legacyWorkersMachineOutputRequested(); + if (machineOutput || output.format === "json") { + return yield* new LegacyWorkersFollowNotSupportedError({ + message: + "--follow cannot be combined with a single-payload output format. " + + "Use --output-format stream-json to stream, or drop --follow.", + }); + } + } + + const pollSchedule = + options.pollSchedule ?? Schedule.spaced(`${WORKER_LOG_POLL_SECONDS} seconds`); + const readRetrySchedule = options.retrySchedule ?? FOLLOW_READ_RETRY; + // A poll asks for whatever arrived since the cursor, not for `--tail` lines; + // `--tail 0` means "no history", not "no new lines". + const pollTail = Math.max(flags.tail, 1); + + /** + * Write a batch of lines out, in whichever form the format calls for. + * + * `stream-json` emits the existing `log-entry` event per line rather than one + * terminal `result`: a tail has no terminal element, and that variant already + * carries the field set this needs. `stream` is derived from the level so a + * consumer can split diagnostics from ordinary output the way it would for a + * real process; `source` distinguishes the backlog from what arrived after. + */ + const emitLines = ( + batch: ReadonlyArray, + origin: "history" | "live" = "history", + ) => + Effect.gen(function* () { + if (batch.length === 0) { + return; + } + if (output.format === "stream-json") { + for (const entry of batch) { + const level = legacyWorkerLogLevel(entry); + yield* output.event({ + type: "log-entry", + timestamp: new Date(entry.timestampMs).toISOString(), + service: name, + stream: level === "error" || level === "warn" ? "stderr" : "stdout", + line: entry.message, + source: origin, + }); + } + return; + } + yield* output.raw(`${batch.map((entry) => legacyRenderWorkerLogLine(entry)).join("\n")}\n`); + }); + const streams = Option.isSome(flags.source) ? [WORKER_LOG_STREAMS[flags.source.value as WorkerLogSourceChoice]] : ALL_WORKER_LOG_STREAMS; @@ -103,7 +196,10 @@ export const legacyWorkersLogs = Effect.fn("legacy.experimental.workers.logs")(f // Nothing came back, which is two different situations wearing the same face: // a worker that is not deployed at all, and one that is deployed and quiet. // Only worth one extra request, and only in this branch. - if (entries.length === 0) { + // + // Skipped when `--tail 0` asked for no history: no query was made, so zero + // rows says nothing about whether the worker exists. + if (entries.length === 0 && flags.tail > 0) { const deployed = yield* getWorker(api, projectRef, name); if (Option.isNone(deployed)) { return yield* Effect.fail( @@ -123,19 +219,21 @@ export const legacyWorkersLogs = Effect.fn("legacy.experimental.workers.logs")(f }; // `-o` asks for a machine-readable stdout, so nothing human may be written to - // it — `output.success` logs to stdout in text mode. - if (yield* legacyEmitWorkersMachineOutput(payload)) { + // it — `output.success` logs to stdout in text mode. Unreachable while + // following, which refuses these formats up front. + if (!flags.follow && (yield* legacyEmitWorkersMachineOutput(payload))) { return; } - // One structured emission, in the structured branch only. Emitting before the - // check above put the payload on stdout twice. - if (output.format !== "text") { + // One structured emission, in the structured branch only, and only for a + // bounded read. A tail has no terminal payload to put here — it emits a + // `log-entry` event per line through `emitLines` instead. + if (!flags.follow && output.format !== "text") { yield* output.success("", payload); return; } - if (entries.length === 0) { + if (entries.length === 0 && !flags.follow) { // Deployed (the check above would have failed otherwise) and silent. yield* output.raw(`No logs for "${name}" in the last 24 hours.\n`); yield* emitSuccessTrailer( @@ -147,10 +245,83 @@ export const legacyWorkersLogs = Effect.fn("legacy.experimental.workers.logs")(f // Oldest first: the query orders newest-first so `limit` means "most recent", // but a reader scrolls forwards through time, and a stack trace only makes // sense in the order it was printed. - // No TTY check here: `legacyRenderWorkerLogLine` defaults to `process.stdout` - // and the colour helpers gate on it themselves, honouring NO_COLOR, CLICOLOR, - // CLICOLOR_FORCE and CI as well as the stream. - yield* output.raw(`${entries.map((entry) => legacyRenderWorkerLogLine(entry)).join("\n")}\n`); + yield* emitLines(entries); + + // A tail with nothing to show yet would otherwise look like a hang. On stderr, + // so it never lands in piped output. + if (flags.follow && entries.length === 0 && output.format === "text") { + yield* output.raw(`Waiting for new logs from "${name}". Press Ctrl+C to stop.\n`, "stderr"); + } + + if (!flags.follow) { + return; + } + + // --- follow --------------------------------------------------------------- + // + // The cursor is the newest timestamp printed, and the set of ids already + // printed. Both live inside this generator rather than being captured while + // the Effect was built: an Effect is a reusable description and may run more + // than once, and shared cursor state across runs would drop lines. + const seenIds = yield* Ref.make(new Set(entries.map((entry) => entry.id))); + const newestSeenMs = yield* Ref.make( + entries.length === 0 ? Date.now() : entries[entries.length - 1]!.timestampMs, + ); + + const pollOnce = Effect.gen(function* () { + const cursor = yield* Ref.get(newestSeenMs); + const rows = yield* fetchWorkerLogs(api, projectRef, { + name, + streams, + tail: pollTail, + window: followWindow(new Date(), cursor), + }); + + // Windows always overlap - the server rounds them to the minute and the + // cursor deliberately lags - so dedupe is what makes the overlap invisible + // rather than a source of repeats. + const printed = yield* Ref.get(seenIds); + const fresh = rows.filter((row) => !printed.has(row.id)); + if (fresh.length === 0) { + return; + } + + yield* emitLines(fresh, "live"); + yield* Ref.update(seenIds, (previous) => { + const next = new Set(previous); + for (const row of fresh) { + next.add(row.id); + } + // Bounded so a tail left running for hours does not grow it without + // limit. Only ids inside the grace window can still be re-offered, so + // forgetting the oldest cannot resurrect them. + if (next.size <= SEEN_ID_LIMIT) { + return next; + } + return new Set([...next].slice(next.size - SEEN_ID_LIMIT)); + }); + yield* Ref.set( + newestSeenMs, + fresh.reduce((newest, row) => Math.max(newest, row.timestampMs), cursor), + ); + }); + + // A 429 or a blip should not end a tail the user is watching; the schedule is + // spaced in seconds, so retrying rides out a transient failure without + // spending the rate limit. + const poll = pollOnce.pipe(Effect.retry({ schedule: readRetrySchedule })); + + // `repeat` runs the body before applying the schedule, so the first poll is + // immediate. That is wanted: it catches anything that landed while the history + // query was in flight, and the rows it repeats are discarded by the id dedupe. + // Measured cost is ~7 requests in the worst 60-second window, against a limit + // of 10. + yield* Effect.raceFirst( + poll.pipe(Effect.repeat({ schedule: pollSchedule })), + processControl + .awaitSignal() + .pipe(Effect.flatMap((signal) => processControl.exit(signal === "SIGINT" ? 130 : 0))), + ); }).pipe( Effect.ensuring(linkedProjectCache.cache(projectRef)), Effect.ensuring(telemetryState.flush), diff --git a/apps/cli/src/legacy/commands/experimental/workers/logs/logs.integration.test.ts b/apps/cli/src/legacy/commands/experimental/workers/logs/logs.integration.test.ts index cd1e9ba655..ec17ede346 100644 --- a/apps/cli/src/legacy/commands/experimental/workers/logs/logs.integration.test.ts +++ b/apps/cli/src/legacy/commands/experimental/workers/logs/logs.integration.test.ts @@ -1,6 +1,6 @@ import { rmSync } from "node:fs"; import { describe, expect, it } from "@effect/vitest"; -import { Effect, Option } from "effect"; +import { Effect, Option, Schedule } from "effect"; import { makeWorkersProject, setupLegacyWorkers, @@ -12,6 +12,7 @@ import { workersRoute, WORKERS_PROJECT_REF, } from "../../../../../../tests/helpers/legacy-workers.ts"; +import { LegacyWorkersFollowNotSupportedError } from "../workers.errors.ts"; import { InvalidWorkerNameError, WorkerLogsQueryFailedError, @@ -53,6 +54,19 @@ function flags(overrides: Record = {}) { } as Parameters[0]; } +/** + * Follow options that drive the loop instantly and stop after N polls. + * + * The real schedule is spaced in seconds; `recurs` also gives the tail an end, so + * a test does not have to deliver a signal just to finish. + */ +function followFor(polls: number) { + return { + pollSchedule: Schedule.recurs(polls), + retrySchedule: Schedule.recurs(0), + }; +} + function logsResponse(rows: ReadonlyArray) { return { status: 200, body: { result: rows, error: null } }; } @@ -558,4 +572,173 @@ describe("legacy workers logs", () => { expect(suggestion).toContain(`--project-ref ${WORKERS_PROJECT_REF}`); }).pipe(Effect.provide(layer), Effect.ensuring(Effect.sync(repo.cleanup))); }); + + it.live("keeps printing new lines, sending both bounds on every poll", () => { + const repo = project(); + const { layer, out, http } = setupLegacyWorkers({ + workdir: repo.dir, + routes: { + [LOGS_ROUTE]: [ + logsResponse([workerLogRow({ id: "a", tsMs: T1, message: "first" })]), + logsResponse([workerLogRow({ id: "b", tsMs: T2, message: "second" })]), + logsResponse([workerLogRow({ id: "c", tsMs: T3, message: "third" })]), + ], + }, + }); + + return Effect.gen(function* () { + yield* legacyWorkersLogs(flags({ follow: true }), followFor(2)); + + expect(out.stdoutText).toContain("first"); + expect(out.stdoutText).toContain("second"); + expect(out.stdoutText).toContain("third"); + + // Every request, history and polls alike, must carry both bounds: a lone + // start silently yields a one-minute window and neither is an error. + for (const request of http.requests) { + const query = sentQuery(request); + expect(query.iso_timestamp_start).toBeTruthy(); + expect(query.iso_timestamp_end).toBeTruthy(); + expect( + Date.parse(query.iso_timestamp_end!) - Date.parse(query.iso_timestamp_start!), + ).toBeLessThan(24 * 60 * 60 * 1000); + } + }).pipe(Effect.provide(layer), Effect.ensuring(Effect.sync(repo.cleanup))); + }); + + it.live("does not reprint a line an overlapping window returns again", () => { + const repo = project(); + const { layer, out } = setupLegacyWorkers({ + workdir: repo.dir, + routes: { + [LOGS_ROUTE]: [ + logsResponse([workerLogRow({ id: "a", tsMs: T1, message: "only once" })]), + // The cursor lags deliberately, so the same row comes back. + logsResponse([ + workerLogRow({ id: "a", tsMs: T1, message: "only once" }), + workerLogRow({ id: "b", tsMs: T2, message: "and this" }), + ]), + ], + }, + }); + + return Effect.gen(function* () { + yield* legacyWorkersLogs(flags({ follow: true }), followFor(1)); + + expect(out.stdoutText.match(/only once/gu)).toHaveLength(1); + expect(out.stdoutText).toContain("and this"); + }).pipe(Effect.provide(layer), Effect.ensuring(Effect.sync(repo.cleanup))); + }); + + it.live("still emits a line that arrived late, inside the grace window", () => { + const repo = project(); + const { layer, out } = setupLegacyWorkers({ + workdir: repo.dir, + routes: { + [LOGS_ROUTE]: [ + logsResponse([workerLogRow({ id: "a", tsMs: T3, message: "newest first" })]), + // Older than the cursor, which is why the cursor lags at all. + logsResponse([workerLogRow({ id: "late", tsMs: T1, message: "arrived late" })]), + ], + }, + }); + + return Effect.gen(function* () { + yield* legacyWorkersLogs(flags({ follow: true }), followFor(1)); + + expect(out.stdoutText).toContain("arrived late"); + }).pipe(Effect.provide(layer), Effect.ensuring(Effect.sync(repo.cleanup))); + }); + + it.live("skips history for --tail 0 but still follows", () => { + const repo = project(); + const { layer, out, http } = setupLegacyWorkers({ + workdir: repo.dir, + routes: { + [LOGS_ROUTE]: logsResponse([workerLogRow({ id: "new", tsMs: T2, message: "brand new" })]), + }, + }); + + return Effect.gen(function* () { + yield* legacyWorkersLogs(flags({ follow: true, tail: 0 }), followFor(1)); + + expect(out.stdoutText).toContain("brand new"); + // No history request; every request belongs to the poll loop, and none may + // ask for `limit 0`, which the endpoint rejects. + for (const request of http.requests) { + expect(sentQuery(request).sql).not.toContain("limit 0"); + } + }).pipe(Effect.provide(layer), Effect.ensuring(Effect.sync(repo.cleanup))); + }); + + it.live("emits a log-entry event per line under stream-json", () => { + const repo = project(); + const { layer, out } = setupLegacyWorkers({ + workdir: repo.dir, + format: "stream-json", + routes: { + [LOGS_ROUTE]: [ + logsResponse([workerIngressLogRow({ id: "a", tsMs: T1, status: "500" })]), + logsResponse([workerLogRow({ id: "b", tsMs: T2, message: "app line" })]), + ], + }, + }); + + return Effect.gen(function* () { + yield* legacyWorkersLogs(flags({ follow: true }), followFor(1)); + + const entries = out.events.filter((event) => event.type === "log-entry"); + expect(entries).toHaveLength(2); + // A tail has no terminal payload, so no single `result` is emitted. + expect(out.events.filter((event) => event.type === "result")).toHaveLength(0); + // An error line is routed to stderr so a consumer can split diagnostics. + expect(entries[0]).toMatchObject({ stream: "stderr", source: "history" }); + expect(entries[1]).toMatchObject({ stream: "stdout", source: "live" }); + }).pipe(Effect.provide(layer), Effect.ensuring(Effect.sync(repo.cleanup))); + }); + + it.live("refuses --follow for the single-payload output formats", () => { + const repo = project(); + + return Effect.gen(function* () { + for (const setup of [ + setupLegacyWorkers({ workdir: repo.dir, goOutput: "json", routes: {} }), + setupLegacyWorkers({ workdir: repo.dir, format: "json", routes: {} }), + ]) { + const error = yield* legacyWorkersLogs(flags({ follow: true })).pipe( + Effect.flip, + Effect.provide(setup.layer), + ); + + expect(error).toBeInstanceOf(LegacyWorkersFollowNotSupportedError); + // Refused before any query is paid for. + expect(setup.http.requests).toHaveLength(0); + } + }).pipe(Effect.ensuring(Effect.sync(repo.cleanup))); + }); + + it.live("exits 130 when interrupted with SIGINT", () => { + const repo = project(); + const { layer, processControl } = setupLegacyWorkers({ + workdir: repo.dir, + signal: "SIGINT", + routes: { [LOGS_ROUTE]: logsResponse([workerLogRow({ id: "a", tsMs: T1 })]) }, + }); + + return Effect.gen(function* () { + // `exit` never returns - in production the process is gone - so the handler + // cannot be awaited here. Fork it and synchronise on the exit itself, which + // is the observable condition, rather than on a delay. + yield* Effect.forkChild( + legacyWorkersLogs(flags({ follow: true }), { + pollSchedule: Schedule.forever, + retrySchedule: Schedule.recurs(0), + }).pipe(Effect.ignore), + ); + + const code = yield* processControl.awaitExit; + + expect(code).toBe(130); + }).pipe(Effect.provide(layer), Effect.ensuring(Effect.sync(repo.cleanup))); + }); }); diff --git a/apps/cli/src/legacy/commands/experimental/workers/workers.errors.ts b/apps/cli/src/legacy/commands/experimental/workers/workers.errors.ts index 65e2b749a2..c0a5c0a3b8 100644 --- a/apps/cli/src/legacy/commands/experimental/workers/workers.errors.ts +++ b/apps/cli/src/legacy/commands/experimental/workers/workers.errors.ts @@ -23,3 +23,25 @@ export class LegacyWorkersEnvNotSupportedError extends Data.TaggedError( return actionability.invalidInput; } } + +/** + * `--follow` was asked for alongside an output format that cannot express a + * stream. + * + * `-o json|yaml|toml` and `--output-format json` each promise exactly one + * terminal payload, and an unbounded tail has no last element to put in it. + * Refused up front rather than at the first emission, for the same reason + * {@link LegacyWorkersEnvNotSupportedError} is: discovering it later means + * failing after the first query has been paid for. + * + * `--output-format stream-json` is the streaming machine format and is allowed. + */ +export class LegacyWorkersFollowNotSupportedError extends Data.TaggedError( + "LegacyWorkersFollowNotSupportedError", +)<{ + readonly message: string; +}> { + get [ErrorActionabilityId](): CliErrorActionabilityDeclaration { + return actionability.provideFlags; + } +} diff --git a/apps/cli/src/shared/workers/worker-logs.sql.ts b/apps/cli/src/shared/workers/worker-logs.sql.ts index 855bdacc31..98d2845c04 100644 --- a/apps/cli/src/shared/workers/worker-logs.sql.ts +++ b/apps/cli/src/shared/workers/worker-logs.sql.ts @@ -50,6 +50,30 @@ const WORKER_LOG_STREAM_ATTRIBUTE = "source"; */ export const WORKER_LOG_WINDOW_MINUTES = 23 * 60 + 59; +/** + * How often `--follow` re-queries. + * + * **Set by the rate limit, not by responsiveness.** The v1 analytics endpoints + * allow 10 requests per 60 seconds, so the two-second poll a live tail suggests + * would 429 within the first ten seconds. Six seconds is the arithmetic floor; + * ten leaves room for the initial history query, the deployed-worker check, and a + * retry inside the same window. + */ +export const WORKER_LOG_POLL_SECONDS = 10; + +/** + * How far behind the newest line seen the next window starts. + * + * Guest lines are relayed CloudWatch -> subscription filter -> Lambda -> Logflare + * and arrive **late and out of order**, so a cursor sitting exactly on the newest + * timestamp drops every straggler permanently. The window is deliberately + * re-asked for ground it has already covered; `id` dedupe absorbs the overlap. + * + * Wider than one poll interval, so a line delayed by a full cycle is still + * inside the next window. + */ +export const WORKER_LOG_CURSOR_GRACE_SECONDS = 60; + /** * Timestamps for the endpoint's `iso_timestamp_start`/`iso_timestamp_end`. * @@ -79,6 +103,33 @@ export function logWindow( }; } +/** + * The window for one `--follow` poll: from just before the newest line seen, up + * to now. + * + * Clamped to the same sub-24h span as {@link logWindow}. That matters when a tail + * is left running past a laptop suspend: without the clamp the resumed poll would + * ask for a wider span, and the server answers an over-wide request by rewriting + * `end` to `start + 24h` — returning an *older* slice rather than a truncated one, + * so a resumed tail would silently start replaying yesterday. + */ +export function followWindow( + now: Date, + newestSeenMs: number, + options: { + readonly graceSeconds?: number; + readonly spanMinutes?: number; + } = {}, +): { readonly start: string; readonly end: string } { + const grace = (options.graceSeconds ?? WORKER_LOG_CURSOR_GRACE_SECONDS) * 1000; + const spanMs = (options.spanMinutes ?? WORKER_LOG_WINDOW_MINUTES) * 60_000; + const earliest = now.getTime() - spanMs; + return { + start: isoLogTimestamp(new Date(Math.max(newestSeenMs - grace, earliest))), + end: isoLogTimestamp(now), + }; +} + /** * Single-quoted SQL string literal. * diff --git a/apps/cli/tests/helpers/legacy-workers.ts b/apps/cli/tests/helpers/legacy-workers.ts index e308f6f576..98c91e846d 100644 --- a/apps/cli/tests/helpers/legacy-workers.ts +++ b/apps/cli/tests/helpers/legacy-workers.ts @@ -17,7 +17,7 @@ import { randomLayer } from "../../src/shared/runtime/random.layer.ts"; import { LegacyProjectNotLinkedError } from "../../src/legacy/config/legacy-project-ref.errors.ts"; import { mockLegacyLinkedProjectCacheLayer } from "./legacy-mocks.ts"; import { LegacyTelemetryState } from "../../src/legacy/telemetry/legacy-telemetry-state.service.ts"; -import { mockOutput, mockRuntimeInfo, mockTty } from "./mocks.ts"; +import { mockOutput, mockProcessControl, mockRuntimeInfo, mockTty } from "./mocks.ts"; /** * Shared scaffolding for the `supabase experimental workers` command integration tests. @@ -372,6 +372,12 @@ export interface WorkersSetupOptions { readonly yes?: boolean; /** Raw argv, which `legacyResolveYes` scans for an explicit `--yes=false`. */ readonly cliArgs?: ReadonlyArray; + /** + * The signal `awaitSignal` resolves with. `logs --follow` races its poll loop + * against this, so a test that wants the tail to end supplies one; the default + * never fires, modelling a terminal nobody has interrupted. + */ + readonly signal?: "SIGINT" | "SIGTERM" | "SIGHUP"; } /** @@ -412,11 +418,15 @@ export function setupLegacyWorkers(options: WorkersSetupOptions) { }); const http = mockWorkersHttp(options.routes ?? {}); const telemetry = mockWorkersTelemetryState(); + const processControl = mockProcessControl( + options.signal === undefined ? {} : { signal: options.signal }, + ); return { out, http, telemetry, + processControl, layer: Layer.mergeAll( out.layer, http.layer, @@ -433,6 +443,7 @@ export function setupLegacyWorkers(options: WorkersSetupOptions) { ), Layer.succeed(LegacyYesFlag, options.yes ?? false), Layer.succeed(CliArgs, { args: options.cliArgs ?? [] }), + processControl.layer, BunServices.layer, ), }; From 2f68ad6d92c9875ea1bdd8acc52f4e96db5747b9 Mon Sep 17 00:00:00 2001 From: Matt Johnston Date: Mon, 31 Aug 2026 16:06:52 -0300 Subject: [PATCH 4/5] feat: show stream tags in worker logs when multiple sources --- .../experimental/workers/logs/logs.handler.ts | 8 +- .../workers/logs/logs.integration.test.ts | 13 +-- .../workers/workers-logs.format.ts | 47 +++++++++- .../workers/workers-logs.format.unit.test.ts | 89 +++++++++++++++---- 4 files changed, 129 insertions(+), 28 deletions(-) diff --git a/apps/cli/src/legacy/commands/experimental/workers/logs/logs.handler.ts b/apps/cli/src/legacy/commands/experimental/workers/logs/logs.handler.ts index be94607ca0..df6869fce0 100644 --- a/apps/cli/src/legacy/commands/experimental/workers/logs/logs.handler.ts +++ b/apps/cli/src/legacy/commands/experimental/workers/logs/logs.handler.ts @@ -138,6 +138,10 @@ export const legacyWorkersLogs = Effect.fn("legacy.experimental.workers.logs")(f // `--tail 0` means "no history", not "no new lines". const pollTail = Math.max(flags.tail, 1); + // The stream tag only earns its width when streams are actually mixed; with + // `--source` every line would carry the same one. + const showStream = Option.isNone(flags.source); + /** * Write a batch of lines out, in whichever form the format calls for. * @@ -169,7 +173,9 @@ export const legacyWorkersLogs = Effect.fn("legacy.experimental.workers.logs")(f } return; } - yield* output.raw(`${batch.map((entry) => legacyRenderWorkerLogLine(entry)).join("\n")}\n`); + yield* output.raw( + `${batch.map((entry) => legacyRenderWorkerLogLine(entry, { showStream })).join("\n")}\n`, + ); }); const streams = Option.isSome(flags.source) diff --git a/apps/cli/src/legacy/commands/experimental/workers/logs/logs.integration.test.ts b/apps/cli/src/legacy/commands/experimental/workers/logs/logs.integration.test.ts index ec17ede346..e0900bf8d8 100644 --- a/apps/cli/src/legacy/commands/experimental/workers/logs/logs.integration.test.ts +++ b/apps/cli/src/legacy/commands/experimental/workers/logs/logs.integration.test.ts @@ -98,12 +98,13 @@ describe("legacy workers logs", () => { return Effect.gen(function* () { yield* legacyWorkersLogs(flags()); - const lines = out.stdoutText.trimEnd().split("\n"); - expect(lines.map((line) => line.split(" ")[1])).toEqual([ - "listening on :8080", - "terminate hook", - "app drained", - ]); + // `