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..6fc276ac7d --- /dev/null +++ b/apps/cli/src/legacy/commands/experimental/workers/logs/SIDE_EFFECTS.md @@ -0,0 +1,155 @@ +# `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 bounded invocation spends one request, or +two when the result is empty. + +`--follow` polls every **10 seconds**. A quiet tail spends one request per poll — +6 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 poll that finds more than one page of new rows drains the burst over up to +**5** requests, so a sustained backlog can reach **30 requests a minute** and will +be rate limited. That is deliberate rather than budgeted for: a 429 mid-tail is +retried on a spaced schedule rather than ending the tail, so the effect is a +throttled tail, not a dropped one. + +The drain bound is also a **lossy** one. It walks backwards from the newest rows, +so exhausting it leaves the oldest part of the burst unfetched while the cursor +advances past it — a burst above **5000 lines in one poll interval** loses its +middle. The run says so once on stderr, in every output format, since a +`stream-json` consumer cannot infer the hole from the events it receives. + +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 + +| 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 | +| `130` | `--follow` interrupted with SIGINT | + +## 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` | + +`--kind` 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; `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 | neither — both are text-only | +| `--output-format stream-json` | the same result as a single terminal event | neither — both are text-only | +| `-o json` / `yaml` / `toml` | the same payload in that encoding, and nothing else | neither — both are text-only | +| `-o pretty` / `table` / `csv` | the text rendering — these fall through rather than encoding | both, as in text | +| `-o env` | refused before any request; the payload nests a `logs` array a flat `KEY=value` list cannot express | the error | + +A structured emission is the end of a bounded read: the handler returns at +`legacyEmitWorkersMachineOutput` or at `output.success`, so the no-logs line and +its `status` trailer below them are never reached, and `output.task` is a no-op +in those modes. `-o pretty`, `table` and `csv` are the exception, since they +encode nothing and fall through to the same text branch. + +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`. + +A bounded read echoes `--kind` back as a top-level `kind` key when the flag was +given. That is a different axis from the per-line `source` above — `kind` is which +stream was asked for, `source` is whether the line came from the backlog or the +tail — so the two never mean the same thing. + +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 +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..d6f98556a2 --- /dev/null +++ b/apps/cli/src/legacy/commands/experimental/workers/logs/logs.command.ts @@ -0,0 +1,97 @@ +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_KINDS, + 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"; + +/** + * 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, + ), + kind: Flag.choice("kind", WORKER_LOG_KINDS).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, + ), + follow: Flag.boolean("follow").pipe( + Flag.withAlias("f"), + Flag.withDescription( + `Keep printing new lines until interrupted, polling every ${WORKER_LOG_POLL_SECONDS} seconds.`, + ), + // Required: `Flag.boolean` alone builds a *required* param, which breaks + // invocations that omit the flag. `legacy-boolean-flag-defaults.unit.test.ts` + // walks the command tree and fails any bare boolean. + 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. Use 0 with --follow to skip history and print only new lines.", + ), + 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.\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([ + { + command: "supabase experimental workers logs api", + description: "Print the last 100 log lines across all streams", + }, + { + command: "supabase experimental workers logs api --kind 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( + // `config` as well as `flags`: `--kind` 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..bc97de5cdc --- /dev/null +++ b/apps/cli/src/legacy/commands/experimental/workers/logs/logs.handler.ts @@ -0,0 +1,493 @@ +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"; +import { + legacyEmitWorkersMachineOutput, + legacyRejectWorkersEnvOutput, + legacyWorkersProjectRefSuffix, +} from "../workers.output.ts"; +import { + legacyRenderWorkerLogLine, + legacyWorkerLogLevel, + legacyWorkerLogText, +} 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, + type WorkerLogEntry, +} from "../../../../../shared/workers/worker-logs-api.ts"; +import { + ALL_WORKER_LOG_STREAMS, + followWindow, + logWindow, + WORKER_LOG_POLL_SECONDS, + WORKER_LOG_STREAMS, +} from "../../../../../shared/workers/worker-logs.sql.ts"; +import { getWorker } from "../../../../../shared/workers/workers-api.ts"; +import { + WorkerLogsQueryFailedError, + WorkerLogsRateLimitedError, + WorkerNotDeployedError, + WorkersApiNetworkError, + WorkersApiUnexpectedStatusError, +} 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 { + legacyWorkersMachineOutputRequested, + legacyWorkersRenderFormat, +} from "../workers.output.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. + */ + +/** + * 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 many rows one poll asks for per request. + * + * Independent of `--tail`, which bounds only the history a run opens with. + * Sharing them meant `--tail 1 --follow` polled with `limit 1`: the query orders + * newest-first, so a burst came back as its newest row alone and the cursor then + * advanced past the rest, dropping them for good. The default `--tail 100` had + * the same hole above 100 rows in a polling interval. + */ +const FOLLOW_PAGE_SIZE = 1000; + +/** + * How many requests one poll may spend draining a burst. + * + * A bound rather than an open loop: the endpoint allows 10 requests a minute, so + * an unbounded drain could spend a whole window's allowance on one poll. + * + * **Rows past the bound are dropped, not deferred.** The drain walks `end` + * backwards, so the pages it did fetch are the *newest* ones; the cursor then + * advances to the newest row printed, past the region it never reached. Only the + * part of that region inside the next window's grace is picked up again. Nothing + * here can fix that — `followWindow` moves the window's floor, not its ceiling, + * so lowering the cursor just re-fetches the same newest pages and never walks + * down to the gap. A burst above {@link FOLLOW_PAGE_SIZE} × this bound in one + * poll interval therefore loses its middle, and says so on stderr. + */ +const FOLLOW_MAX_PAGES = 5; + +/** + * 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" }), +); + +/** + * Which poll failures are worth spending another request on. + * + * A tail should ride out a 429 or a momentary blip, but 401, 402 and 404 answer + * the same way every time. Retrying those held the error back for a minute and + * spent most of the endpoint's ten-requests-per-minute allowance getting nowhere, + * so the reader waited longer and then hit a rate limit on top of the real cause. + * + * Server-side statuses are retried and client-side ones are not, with the + * exception of 408 and 429, which are the server asking for exactly that. A + * decode failure carries the response's own status, so a malformed 200 body is + * correctly read as terminal: it will not parse any better on a second attempt. + */ +function isRetryableFollowFailure(error: unknown): boolean { + if (error instanceof WorkersApiUnexpectedStatusError) { + return error.status >= 500 || error.status === 408 || error.status === 429; + } + return ( + error instanceof WorkerLogsRateLimitedError || + error instanceof WorkersApiNetworkError || + // The endpoint reports a rejected or timed-out query this way, and its own + // suggestion is to retry shortly. + error instanceof WorkerLogsQueryFailedError + ); +} + +/** + * 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); + 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, + 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; + + // Telemetry wraps the ref resolution as well: an unlinked non-interactive + // checkout fails inside `resolve`, and by then the command has run. Only the + // linked-project cache stays under the ref, since it has nothing to write + // without one. + yield* Effect.gen(function* () { + 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(); + + // Resolved once, before anything branches: `-o` outranks `--output-format`, + // so `output.format` on its own is not what this run renders in. + const renderFormat = yield* legacyWorkersRenderFormat(); + + // 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 || renderFormat === "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; + + // The stream tag only earns its width when streams are actually mixed; with + // `--kind` every line would carry the same one. + const showStream = Option.isNone(flags.kind); + + /** + * 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 (renderFormat === "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", + // The composed sentence, the same one text mode renders: a + // request line's status and duration and a build's reason live + // in `log_attributes`, and `log-entry` has no field to carry + // them separately. + line: legacyWorkerLogText(entry), + source: origin, + }); + } + return; + } + yield* output.raw( + `${batch.map((entry) => legacyRenderWorkerLogLine(entry, { showStream })).join("\n")}\n`, + ); + }); + + const streams = Option.isSome(flags.kind) + ? [WORKER_LOG_STREAMS[flags.kind.value]] + : ALL_WORKER_LOG_STREAMS; + + // Before any request, so a slow history query or deployed-worker check + // cannot widen what `followFloorMs` below treats as "already there". + const startedAtMs = Date.now(); + + // `--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. + // + // `--tail 0` makes no history query, so zero rows says nothing either way — + // but a tail still has to know the worker exists, or a typo waits forever on + // logs that can never arrive. A bounded `--tail 0` run prints nothing by + // definition and is left alone. + if (entries.length === 0 && (flags.tail > 0 || flags.follow)) { + // Its own task: with `--tail 0` there is no "Fetching logs..." to inherit, + // and clearing that one before this request left text mode silent across + // a call that can take a moment. + const checking = yield* output.task("Checking worker..."); + const deployed = yield* getWorker(api, projectRef, name).pipe( + Effect.tapError(() => checking.fail()), + ); + yield* checking.clear(); + 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.kind) ? { kind: flags.kind.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. Unreachable while + // following, which refuses these formats up front. + if (!flags.follow && (yield* legacyEmitWorkersMachineOutput(payload))) { + return; + } + + // 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 && renderFormat !== "text") { + yield* output.success("", payload); + return; + } + + 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( + `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. + 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 && renderFormat === "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))); + // Same reason these live in the generator rather than the closure: an + // Effect is a reusable description, and a notice already "shown" on a + // previous run would stay silent on the next one. + const skipNoticeShown = yield* Ref.make(false); + const newestSeenMs = yield* Ref.make( + entries.length === 0 ? Date.now() : entries[entries.length - 1]!.timestampMs, + ); + + // `--tail 0` asked for no history, and `followWindow` deliberately reaches + // a grace period behind the cursor so a line relayed late is still caught. + // Both are wanted, and together they let pre-invocation lines through — so + // keep the wide window and filter on when the line was actually written. + const followFloorMs = flags.tail === 0 ? startedAtMs : Number.NEGATIVE_INFINITY; + + const pollOnce = Effect.gen(function* () { + const cursor = yield* Ref.get(newestSeenMs); + + // One request only ever answers with the newest page of its window, so a + // burst bigger than a page needs several. Walk `end` backwards while + // pages come back full; a short page means the window is drained. + const collected: Array = []; + let end = new Date(); + // A short page is the only proof the window is empty below this point. + // Both other exits — the page budget running out, and a full page too + // narrow to walk past — leave rows unfetched underneath. + let drained = false; + for (let page = 0; page < FOLLOW_MAX_PAGES; page += 1) { + const rows = yield* fetchWorkerLogs(api, projectRef, { + name, + streams, + tail: FOLLOW_PAGE_SIZE, + window: followWindow(end, cursor), + }); + collected.push(...rows); + if (rows.length < FOLLOW_PAGE_SIZE) { + drained = true; + break; + } + // Rows arrive oldest-first, so the next page ends where this one began. + const nextEnd = new Date(rows[0]!.timestampMs); + // A full page whose rows all share one timestamp cannot narrow the + // window: re-requesting it would return the same page forever. + if (nextEnd.getTime() >= end.getTime()) { + break; + } + end = nextEnd; + } + + // Once per run, not once per poll: a sustained burst would otherwise + // repeat this every interval and bury the lines it is warning about. + // + // Emitted in **every** format, unlike the "Waiting for new logs" notice + // above — same reasoning as `push`'s `reportUnattempted`. That one is + // progress, which a machine consumer did not ask for; this one says the + // stream it is reading has a hole in it, which it cannot infer from the + // events themselves. + if (!drained && !(yield* Ref.get(skipNoticeShown))) { + yield* Ref.set(skipNoticeShown, true); + yield* output.raw( + `Skipped part of a burst larger than ${FOLLOW_MAX_PAGES * FOLLOW_PAGE_SIZE} lines: ` + + `some lines older than the ones below were not printed. ` + + `Narrow the stream with --kind, or read the full range in the dashboard.\n`, + "stderr", + ); + } + + // 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 = collected + .filter((row) => !printed.has(row.id) && row.timestampMs >= followFloorMs) + // Each page is oldest-first but the pages themselves walk backwards, so + // the concatenation is not ordered until this runs. + .sort((left, right) => left.timestampMs - right.timestampMs); + 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. Anything definitive surfaces on the first + // attempt — see `isRetryableFollowFailure`. + const poll = pollOnce.pipe( + Effect.retry({ schedule: readRetrySchedule, while: isRetryableFollowFailure }), + ); + + // `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. + // + // Cost against the endpoint's limit of 10 requests per 60 seconds: a quiet + // tail spends one request per poll, so 6 a minute, plus the opening history + // query and the deployed-worker check. A poll draining a burst spends up to + // `FOLLOW_MAX_PAGES`, so a sustained backlog can reach 30 a minute and will + // be rate limited — `isRetryableFollowFailure` treats the 429 as transient + // and the spaced retry rides it out, which throttles the tail rather than + // ending it. + yield* Effect.raceFirst( + poll.pipe(Effect.repeat({ schedule: pollSchedule })), + // `setExitCode`, not `exit`: the production `exit` calls `process.exit` + // synchronously, which tears the runtime down from inside this race + // branch — before the linked-project cache is written, before telemetry + // is flushed, and before the instrumentation wrapper emits its post-run + // event. Recording the code lets the race complete normally so the + // finalizers run, and `runCli` exits with it once they have. + processControl + .awaitSignal() + .pipe( + Effect.flatMap((signal) => processControl.setExitCode(signal === "SIGINT" ? 130 : 0)), + ), + ); + }).pipe(Effect.ensuring(linkedProjectCache.cache(projectRef))); + }).pipe(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..fb8af78022 --- /dev/null +++ b/apps/cli/src/legacy/commands/experimental/workers/logs/logs.integration.test.ts @@ -0,0 +1,1057 @@ +import { rmSync } from "node:fs"; +import { describe, expect, it } from "@effect/vitest"; +import { Effect, Exit, Option, Schedule } from "effect"; +import { + makeWorkersProject, + setupLegacyWorkers, + workerApiLogRow, + workerIngressLogRow, + workerLogRow, + workerLogsRoute, + workerResource, + workersRoute, + WORKERS_PROJECT_REF, +} from "../../../../../../tests/helpers/legacy-workers.ts"; +import { LegacyWorkersFollowNotSupportedError } from "../workers.errors.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(), + kind: Option.none(), + tail: 100, + ...overrides, + } 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 } }; +} + +/** + * 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()); + + // `