From cb9116f8ba2ecb71456e4fc43880190c7b66ef3f Mon Sep 17 00:00:00 2001 From: Kazuno Fukuda <4kz12zz@gmail.com> Date: Fri, 7 Aug 2026 21:24:18 +0900 Subject: [PATCH] Handle live stream quota errors --- README.md | 8 ++ docs/cli-design.md | 37 +++++ packages/protocol/src/errors.ts | 2 + src/application/event-stream.ts | 26 +++- src/application/result.ts | 14 +- .../commands/event-stream.test.ts | 133 ++++++++++++++++++ src/presentation/output/errors.ts | 19 ++- src/presentation/output/result.ts | 4 +- 8 files changed, 238 insertions(+), 5 deletions(-) diff --git a/README.md b/README.md index 7a2d4fa..44dabe8 100644 --- a/README.md +++ b/README.md @@ -75,6 +75,14 @@ barestash events stream --endpoint ep_abc123 | jq . Live streaming requires an authenticated private endpoint. For temporary endpoints, use `barestash events tail --endpoint ep_abc123`. +Private live streams are subject to service concurrency and daily quotas. If +the concurrency limit is reached, close another live stream before retrying. +If the daily quota is reached, the CLI prints the API's UTC reset guidance and +`Retry-After` delay to stderr. An admission rejection leaves JSONL stdout empty +and exits non-zero without reconnecting. If an established stream closes at +the daily limit, the CLI makes its normal reconnect attempt and exits without +another reconnect when that attempt receives the quota rejection. + Press `Ctrl+C` to stop streaming. The command exits successfully without adding a non-JSONL line to stdout or a diagnostic to stderr. diff --git a/docs/cli-design.md b/docs/cli-design.md index 3b71382..28406fd 100644 --- a/docs/cli-design.md +++ b/docs/cli-design.md @@ -861,6 +861,13 @@ Expected behavior: - Supports reconnect with `Last-Event-ID` (handled by the CLI SSE client) - For a temporary endpoint, prints the backend policy error once to stderr, writes no JSONL record, and exits without reconnecting +- For an initial concurrency or daily-quota rejection, prints the backend + error and actionable guidance to stderr, writes no JSONL record, and exits + non-zero without reconnecting +- Displays the daily quota's UTC reset guidance and numeric `Retry-After` + delay in seconds +- If a clean stream EOF is followed by a daily-quota rejection on reconnect, + exits non-zero without making another reconnect attempt Implementation sketch: @@ -1094,6 +1101,34 @@ The CLI treats `temporary_endpoint_stream_not_supported` as non-retryable. It prints the error once to stderr, writes nothing to stdout, and exits with code `1` without sleeping or reconnecting. +### Live stream concurrency limit + +```text +Live stream subscriber limit reached for this account (5). + +Close another live stream before retrying. +``` + +The CLI treats `stream_concurrency_limit_exceeded` as non-retryable. It prints +the backend message and guidance to stderr, writes nothing to stdout, and exits +with code `1` without sleeping or reconnecting. + +### Live stream daily quota + +```text +Daily live stream quota reached. Try again after 2026-08-08T00:00:00.000Z. + +Retry-After: 10800 seconds. +``` + +The backend message supplies the UTC reset time, and the CLI displays a valid +numeric `Retry-After` header as seconds. An initial +`stream_daily_quota_exceeded` response is non-retryable and leaves stdout +empty. If an established stream ends cleanly because the quota is exhausted, +the existing reconnect attempt is allowed; when it receives the 429 response, +the CLI prints the error once and exits with code `1` without reconnecting +again. + ### Cannot delete temporary endpoint ```text @@ -1172,6 +1207,8 @@ user to authenticate again. | `temporary_endpoint_delete_not_supported` | 400 | Cannot delete temporary endpoint. Explain that temporary endpoints expire automatically and suggest creating a new temporary endpoint if needed. | | `temporary_endpoint_stream_not_supported` | 400 | Live streaming is unavailable for temporary endpoints. Suggest `events tail` or creating a private endpoint; do not reconnect. | | `event_limit_exceeded` | 429 | Endpoint has reached its configured event limit. Suggest creating and setting a new default endpoint. | +| `stream_concurrency_limit_exceeded` | 429 | Show the backend message and advise closing another live stream before retrying. Keep stdout empty and do not reconnect. | +| `stream_daily_quota_exceeded` | 429 | Show the backend UTC reset guidance and numeric `Retry-After` delay. Keep stdout empty and stop reconnecting. | | `event_not_found` | 404 | Use the Invalid event ID message above. | | `body_not_found` | 404 | Body unavailable for event `{event_id}`. | | `internal_error` | 500 | An unexpected error occurred. Retry later. | diff --git a/packages/protocol/src/errors.ts b/packages/protocol/src/errors.ts index 8c06bce..ae8a77b 100644 --- a/packages/protocol/src/errors.ts +++ b/packages/protocol/src/errors.ts @@ -31,6 +31,8 @@ export type RestErrorCode = | "event_limit_exceeded" | "rate_limit_exceeded" | "rate_limit_unavailable" + | "stream_concurrency_limit_exceeded" + | "stream_daily_quota_exceeded" | "event_not_found" | "body_not_found" | "internal_error"; diff --git a/src/application/event-stream.ts b/src/application/event-stream.ts index 1869ab4..593f387 100644 --- a/src/application/event-stream.ts +++ b/src/application/event-stream.ts @@ -7,6 +7,7 @@ import { } from "../infrastructure/sse.js"; import { type AuthDeps, authHeaders } from "./auth.js"; import { + apiError, CliApiErrorException, type CliResult, fromApiCall, @@ -22,6 +23,18 @@ export type StreamEventsDeps = AuthDeps & { onPayload: (payload: unknown) => void; }; +function retryAfterSeconds(response: Response): number | undefined { + const value = response.headers.get("retry-after"); + + if (value === null) { + return undefined; + } + + const seconds = Number(value); + + return Number.isSafeInteger(seconds) && seconds >= 0 ? seconds : undefined; +} + /** @public */ export async function streamEvents( deps: StreamEventsDeps, @@ -64,7 +77,18 @@ export async function streamEvents( } if (!response.ok) { - return fromApiCall(await deps.apiClient.resultFromResponse(response)); + const result = await deps.apiClient.resultFromResponse(response); + + if ( + result.kind === "error" && + result.error.error.code === "stream_daily_quota_exceeded" + ) { + return apiError(result.error, { + retryAfterSeconds: retryAfterSeconds(response), + }); + } + + return fromApiCall(result); } try { diff --git a/src/application/result.ts b/src/application/result.ts index 3a702ab..79e4f28 100644 --- a/src/application/result.ts +++ b/src/application/result.ts @@ -3,6 +3,7 @@ import type { RestErrorResponse } from "@barestash/cli-protocol/errors"; export type CliApiError = { kind: "api-error"; error: RestErrorResponse; + retryAfterSeconds?: number; }; /** @public */ @@ -31,8 +32,17 @@ export function ok(value: T): CliResult { return { kind: "ok", value }; } -export function apiError(error: RestErrorResponse): CliApiError { - return { kind: "api-error", error }; +export function apiError( + error: RestErrorResponse, + options: { retryAfterSeconds?: number } = {}, +): CliApiError { + return { + kind: "api-error", + error, + ...(options.retryAfterSeconds === undefined + ? {} + : { retryAfterSeconds: options.retryAfterSeconds }), + }; } export function localError(message: string): CliLocalError { diff --git a/src/presentation/commands/event-stream.test.ts b/src/presentation/commands/event-stream.test.ts index 1151104..867ff93 100644 --- a/src/presentation/commands/event-stream.test.ts +++ b/src/presentation/commands/event-stream.test.ts @@ -589,6 +589,139 @@ describe("event stream commands", () => { expect(stdout).toEqual([]); }); + it("reports stream concurrency rejection with actionable guidance", async () => { + const { io, stderr, stdout } = makeIo(); + const requests: Request[] = []; + const sleeps: number[] = []; + + const exitCode = await runCli( + ["events", "stream", "--endpoint", "ep_01JDEF"], + io, + { + env: { + BARESTASH_API_URL: "https://api.example.com", + }, + sleep: async (milliseconds) => { + sleeps.push(milliseconds); + }, + fetch: async (input: string | URL | Request, init?: RequestInit) => { + requests.push(new Request(input, init)); + + return Response.json( + { + error: { + code: "stream_concurrency_limit_exceeded", + message: + "Live stream subscriber limit reached for this account (5).", + }, + }, + { status: 429 }, + ); + }, + }, + ); + + expect(exitCode).toBe(1); + expect(stdout).toEqual([]); + expect(requests).toHaveLength(1); + expect(sleeps).toEqual([]); + expect(stderr).toEqual([ + "Live stream subscriber limit reached for this account (5).", + "", + "Close another live stream before retrying.", + ]); + }); + + it("reports the daily stream quota reset and Retry-After delay", async () => { + const { io, stderr, stdout } = makeIo(); + const requests: Request[] = []; + const sleeps: number[] = []; + + const exitCode = await runCli( + ["events", "stream", "--endpoint", "ep_01JDEF"], + io, + { + env: { + BARESTASH_API_URL: "https://api.example.com", + }, + sleep: async (milliseconds) => { + sleeps.push(milliseconds); + }, + fetch: async (input: string | URL | Request, init?: RequestInit) => { + requests.push(new Request(input, init)); + + return Response.json( + { + error: { + code: "stream_daily_quota_exceeded", + message: + "Daily live stream quota reached. Try again after 2026-08-08T00:00:00.000Z.", + }, + }, + { status: 429, headers: { "retry-after": "10800" } }, + ); + }, + }, + ); + + expect(exitCode).toBe(1); + expect(stdout).toEqual([]); + expect(requests).toHaveLength(1); + expect(sleeps).toEqual([]); + expect(stderr).toEqual([ + "Daily live stream quota reached. Try again after 2026-08-08T00:00:00.000Z.", + "", + "Retry-After: 10800 seconds.", + ]); + }); + + it("reconnects once after clean EOF and stops on daily quota rejection", async () => { + const { io, stderr, stdout } = makeIo(); + const requests: Request[] = []; + const sleeps: number[] = []; + + const exitCode = await runCli( + ["events", "stream", "--endpoint", "ep_01JDEF"], + io, + { + env: { + BARESTASH_API_URL: "https://api.example.com", + }, + sleep: async (milliseconds) => { + sleeps.push(milliseconds); + }, + fetch: async (input: string | URL | Request, init?: RequestInit) => { + requests.push(new Request(input, init)); + + if (requests.length === 1) { + return sseResponse([]); + } + + return Response.json( + { + error: { + code: "stream_daily_quota_exceeded", + message: + "Daily live stream quota reached. Try again after 2026-08-08T00:00:00.000Z.", + }, + }, + { status: 429, headers: { "retry-after": "10800" } }, + ); + }, + }, + ); + + expect(exitCode).toBe(1); + expect(stdout).toEqual([]); + expect(requests).toHaveLength(2); + expect(sleeps).toEqual([1000]); + expect(stderr).toEqual([ + "Daily live stream quota reached. Try again after 2026-08-08T00:00:00.000Z.", + "", + "Retry-After: 10800 seconds.", + ]); + }); + it("reports stream API errors from the first failed response without retrying", async () => { const { io, stderr, stdout } = makeIo(); const requests: Request[] = []; diff --git a/src/presentation/output/errors.ts b/src/presentation/output/errors.ts index 4c821dc..308242b 100644 --- a/src/presentation/output/errors.ts +++ b/src/presentation/output/errors.ts @@ -14,9 +14,26 @@ export function printNoEndpointSelected(io: CliIo): void { } /** @public */ -export function printApiError(io: CliIo, error: RestErrorResponse): void { +export function printApiError( + io: CliIo, + error: RestErrorResponse, + options: { retryAfterSeconds?: number } = {}, +): void { io.stderr(error.error.message); + if (error.error.code === "stream_concurrency_limit_exceeded") { + io.stderr(""); + io.stderr("Close another live stream before retrying."); + } + + if ( + error.error.code === "stream_daily_quota_exceeded" && + options.retryAfterSeconds !== undefined + ) { + io.stderr(""); + io.stderr(`Retry-After: ${options.retryAfterSeconds} seconds.`); + } + if (error.error.code === "endpoint_expired") { io.stderr(""); io.stderr("Create and set a new default endpoint:"); diff --git a/src/presentation/output/result.ts b/src/presentation/output/result.ts index d73b8f6..ed143d9 100644 --- a/src/presentation/output/result.ts +++ b/src/presentation/output/result.ts @@ -9,7 +9,9 @@ export function handleCliResult(result: CliResult, io: CliIo): T | null { } if (result.kind === "api-error") { - printApiError(io, result.error); + printApiError(io, result.error, { + retryAfterSeconds: result.retryAfterSeconds, + }); return null; }