Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 1 addition & 1 deletion DESIGN.md
Original file line number Diff line number Diff line change
Expand Up @@ -12,7 +12,7 @@ This file collects architecture and design notes for studio that aren't otherwis

- [`src/features/instance/databases/DESIGN.md`](src/features/instance/databases/DESIGN.md) — when the browse grid may show its spinner, and why a table without a primary key can't be listed.

- [`src/features/instance/applications/DESIGN.md`](src/features/instance/applications/DESIGN.md) — when a rolling deploy is done: following its restart job, and why a job Studio can't follow is never a success.
- [`src/features/instance/applications/DESIGN.md`](src/features/instance/applications/DESIGN.md) — when a rolling deploy is done: following its restart job, and why a job Studio can't follow is never a success; and when a live deploy may be sent again.
- [`src/features/instance/config/roles/DESIGN.md`](src/features/instance/config/roles/DESIGN.md) — why the role editor's JSON text is the only source of truth, what forces the JSON view, and what Save changes on purpose.
- [`src/router/DESIGN.md`](src/router/DESIGN.md) — why a failed route load is reported and shown only by its route, and why a preload still toasts.

Expand Down
12 changes: 12 additions & 0 deletions src/features/instance/applications/DESIGN.md
Original file line number Diff line number Diff line change
Expand Up @@ -43,3 +43,15 @@ All three kinds of request are built in one place, [`deployRequestBody`](../../.
because Harper's validator refuses the wrong mix outright: a stage with `restart`, or an activation carrying `package`,
install options or `activate`. A stage succeeds only once every other node confirmed staging it, so a stage restarts
nothing and is reported without a health check. Rollback to a kept release (#1232) is the same activation call.

## A live deploy is sent once

A live (SSE) deploy that gets no event stream goes out again as the buffered POST only when its status proves Harper
never started it: a 4xx. Harper refuses a request before the operation starts (authentication, permissions,
validation) with a 4xx, and answers one it started with a 200 event stream at once. Everything else can follow a deploy
that is already running: a gateway's 5xx after forwarding, a 2xx that is not a stream, a dropped connection, the idle
watchdog firing before headers. Harper's own request-queue 503 never started anything either, but status alone cannot
tell it from a gateway's. So [`onDeployComponentSubmit`](../../../integrations/api/instance/applications/deployComponent.ts)
turns those into `SSEInconclusiveError`, which every deploy caller already answers by checking rather than resending.
`streamOperation` stays neutral: it puts the `status` on `SSEUnsupportedError` and leaves the decision to the caller,
because a read (`get_deployment`) may fall back on anything.
Original file line number Diff line number Diff line change
@@ -0,0 +1,103 @@
import { afterEach, describe, expect, it, vi } from 'vitest';

vi.mock('@/integrations/api/sse/resolveInstanceConnection', () => ({
resolveInstanceConnection: () => ({
url: 'https://ins-1.example.com:9925/',
headers: { 'Content-Type': 'application/json', Accept: 'text/event-stream' },
credentials: 'same-origin' as RequestCredentials,
mode: 'direct',
}),
}));

import { SSEInconclusiveError } from '@/integrations/api/sse/errors';
import { onDeployComponentSubmit } from './deployComponent';

const live = {
request: { kind: 'deploy' as const, project: 'my-app', package: 'npm:my-app' },
entityId: 'ins-1' as const,
entityType: 'instance' as const,
useSSE: true,
};

function bufferedClient() {
return { post: vi.fn().mockResolvedValue({ data: { message: 'Successfully deployed: my-app' } }) };
}

function deploy(client: ReturnType<typeof bufferedClient>) {
return onDeployComponentSubmit({ ...live, instanceClient: client as never });
}

afterEach(() => {
vi.unstubAllGlobals();
vi.useRealTimers();
});

describe('the live deploy, through the real stream', () => {
it.each([502, 503, 504])(
'is not sent again after a %i from a gateway, which may have forwarded it',
async (status) => {
vi.stubGlobal(
'fetch',
vi.fn().mockResolvedValue(
new Response('<html>Bad Gateway</html>', {
status,
headers: { 'content-type': 'text/html' },
}),
),
);
const client = bufferedClient();

await expect(deploy(client)).rejects.toBeInstanceOf(SSEInconclusiveError);
expect(client.post).not.toHaveBeenCalled();
},
);

it('is not sent again after a 2xx that is not a stream, which means it ran', async () => {
vi.stubGlobal(
'fetch',
vi.fn().mockResolvedValue(Response.json({ message: 'Successfully deployed: my-app' })),
);
const client = bufferedClient();

await expect(deploy(client)).rejects.toBeInstanceOf(SSEInconclusiveError);
expect(client.post).not.toHaveBeenCalled();
});

it('is not sent again when the connection drops after the request went out', async () => {
vi.stubGlobal('fetch', vi.fn().mockRejectedValue(new TypeError('Failed to fetch')));
const client = bufferedClient();

await expect(deploy(client)).rejects.toBeInstanceOf(SSEInconclusiveError);
expect(client.post).not.toHaveBeenCalled();
});

it('is not sent again when no answer arrives within the idle window', async () => {
vi.useFakeTimers();
vi.stubGlobal(
'fetch',
vi.fn((_url: string, init: RequestInit) =>
new Promise((_resolve, reject) => {
init.signal?.addEventListener('abort', () => reject(new DOMException('aborted', 'AbortError')));
})
),
);
const client = bufferedClient();

const outcome = deploy(client).catch((e: unknown) => e);
await vi.advanceTimersByTimeAsync(120_000);

expect(await outcome).toBeInstanceOf(SSEInconclusiveError);
expect(client.post).not.toHaveBeenCalled();
});

it.each([401, 403])('goes out on the buffered path when Harper refused the stream with a %i', async (status) => {
vi.stubGlobal(
'fetch',
vi.fn().mockResolvedValue(Response.json({ error: 'Login failed' }, { status })),
);
const client = bufferedClient();

await expect(deploy(client)).resolves.toEqual({ message: 'Successfully deployed: my-app' });
expect(client.post).toHaveBeenCalledOnce();
});
});
Original file line number Diff line number Diff line change
Expand Up @@ -4,7 +4,7 @@ vi.mock('./deployComponentStream', () => ({
deployComponentStream: vi.fn(),
}));

import { SSEOperationError, SSEUnsupportedError } from '@/integrations/api/sse/errors';
import { SSEInconclusiveError, SSEOperationError, SSEUnsupportedError } from '@/integrations/api/sse/errors';
import { deployAndFollowComponent, deployRequestBody, onDeployComponentSubmit } from './deployComponent';
import { deployComponentStream } from './deployComponentStream';
import { DeployRolloutError } from './restartJob';
Expand Down Expand Up @@ -123,17 +123,44 @@ describe('onDeployComponentSubmit', () => {
expect(client.post).not.toHaveBeenCalled();
});

it('falls back to the buffered path when the stream is unsupported', async () => {
mockStream.mockRejectedValue(new SSEUnsupportedError('no stream'));
it.each([400, 401, 403, 404, 406, 413, 429])(
'falls back to the buffered path when the stream was refused with a %i',
async (status) => {
mockStream.mockRejectedValue(new SSEUnsupportedError('no stream', { status }));
const client = fakeClient({ message: 'buffered' });
const result = await onDeployComponentSubmit({
...baseArgs,
useSSE: true,
// eslint-disable-next-line @typescript-eslint/no-explicit-any
instanceClient: client as any,
});
expect(result).toEqual({ message: 'buffered' });
expect(client.post).toHaveBeenCalledOnce();
},
);

it.each([
['a gateway 502', 502],
['a gateway 503', 503],
['a gateway 504', 504],
['a Harper 500', 500],
['a 2xx that is not a stream', 200],
['no response at all', undefined],
])('never sends the deploy again after %s, which can follow a deploy already running', async (_, status) => {
const unsupported = new SSEUnsupportedError('no stream', { status });
mockStream.mockRejectedValue(unsupported);
const client = fakeClient({ message: 'buffered' });
const result = await onDeployComponentSubmit({

const error = await onDeployComponentSubmit({
...baseArgs,
useSSE: true,
// eslint-disable-next-line @typescript-eslint/no-explicit-any
instanceClient: client as any,
});
expect(result).toEqual({ message: 'buffered' });
expect(client.post).toHaveBeenCalledOnce();
}).catch((e: unknown) => e);

expect(error).toBeInstanceOf(SSEInconclusiveError);
expect((error as Error).cause).toBe(unsupported);
expect(client.post).not.toHaveBeenCalled();
});

it('rethrows a deploy failure without re-POSTing (no double deploy)', async () => {
Expand Down
15 changes: 12 additions & 3 deletions src/integrations/api/instance/applications/deployComponent.ts
Original file line number Diff line number Diff line change
@@ -1,7 +1,7 @@
import { InstanceClientConfig, InstanceTypeConfig } from '@/config/instanceClientConfig';
import { EntityIds } from '@/features/auth/store/authStore';
import { ReplicatedResponse } from '@/integrations/api/replication';
import { SSEUnsupportedError } from '@/integrations/api/sse/errors';
import { SSEInconclusiveError, SSEUnsupportedError } from '@/integrations/api/sse/errors';
import { useMutation } from '@tanstack/react-query';
import { DeployComponentResult, deployComponentStream, DeployStreamEvent } from './deployComponentStream';
import { DeployRolloutError, followRestartJob } from './restartJob';
Expand Down Expand Up @@ -49,6 +49,11 @@ export interface DeployComponentArgs extends InstanceClientConfig, InstanceTypeC
signal?: AbortSignal;
}

/** Only a 4xx proves Harper never started the deploy: applications/DESIGN.md, "A live deploy is sent once". */
function refusedBeforeRunning({ status }: SSEUnsupportedError): boolean {
return status !== undefined && status >= 400 && status < 500;
}

export async function onDeployComponentSubmit({
request,
entityType,
Expand All @@ -64,14 +69,18 @@ export async function onDeployComponentSubmit({
try {
return await deployComponentStream({ connection: { id: entityId }, body, signal, onEvent });
} catch (error) {
// Couldn't establish a stream (bad response/content-type/connection) and no deploy
// has started yet — safe to fall back to the proven buffered path below.
if (!(error instanceof SSEUnsupportedError)) {
// A terminal error, an inconclusive stream, or a deliberate abort. The deploy has
// already started server-side, so we must NOT re-POST it (that would deploy
// twice). Let the caller handle it (show the error, or poll on inconclusive).
throw error;
}
if (!refusedBeforeRunning(error)) {
throw new SSEInconclusiveError(
'The live deploy did not stream and may already have started, so it was not sent again.',
{ cause: error },
);
}
}
}

Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -92,7 +92,8 @@ export interface DeployComponentStreamParams {
* and resolving with the terminal `done` result.
*
* Throws the transport errors from `@/integrations/api/sse/errors`:
* - `SSEUnsupportedError` — the server/proxy didn't stream; caller should retry buffered.
* - `SSEUnsupportedError` — the server/proxy didn't stream. Resend buffered only when its `status` is a 4xx
* refusal (`onDeployComponentSubmit` decides); anything else may follow a deploy already running.
* - `SSEOperationError` — the deploy failed (terminal `error` event).
* - `SSEInconclusiveError` — the stream ended without a verdict; caller must poll.
*/
Expand Down
8 changes: 6 additions & 2 deletions src/integrations/api/sse/errors.ts
Original file line number Diff line number Diff line change
Expand Up @@ -8,12 +8,16 @@
* The server did not (or could not) speak SSE for this request: the response was not OK,
* arrived without a `text/event-stream` content-type (e.g. an older Harper that ignores
* the `Accept` header, or a proxy that rewrote it), or the connection failed before any
* bytes streamed. The caller should fall back to the classic buffered operation.
* bytes streamed. A read can fall back to the classic buffered operation; a write only when
* `status` shows the server refused it (see `onDeployComponentSubmit`).
*/
export class SSEUnsupportedError extends Error {
constructor(message: string, options?: { cause?: unknown }) {
readonly status?: number;

constructor(message: string, options?: { cause?: unknown; status?: number }) {
super(message, options);
this.name = 'SSEUnsupportedError';
this.status = options?.status;
}
}

Expand Down
32 changes: 24 additions & 8 deletions src/integrations/api/sse/streamOperation.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -109,23 +109,39 @@ describe('streamOperation', () => {
).rejects.toMatchObject({ name: 'SSEOperationError', message: 'The operation failed.' });
});

it('falls back (SSEUnsupportedError) when the response is not an event stream', async () => {
it('falls back (SSEUnsupportedError) when the response is not an event stream, with its status', async () => {
vi.stubGlobal(
'fetch',
vi.fn().mockResolvedValue(sseResponse('{"message":"ok"}', { contentType: 'application/json' })),
);

await expect(
streamOperation({ connection: { id: 'ins-1' }, body: { operation: 'deploy_component' } }),
).rejects.toBeInstanceOf(SSEUnsupportedError);
const error = await streamOperation({ connection: { id: 'ins-1' }, body: { operation: 'deploy_component' } })
.catch((e: unknown) => e);

expect(error).toBeInstanceOf(SSEUnsupportedError);
expect((error as SSEUnsupportedError).status).toBe(200);
});

it('falls back (SSEUnsupportedError) when fetch rejects before streaming', async () => {
it('reports the status of a response that refused the stream', async () => {
vi.stubGlobal(
'fetch',
vi.fn().mockResolvedValue(sseResponse('{"error":"Gateway Timeout"}', { status: 504, contentType: 'text/html' })),
);

const error = await streamOperation({ connection: { id: 'ins-1' }, body: { operation: 'deploy_component' } })
.catch((e: unknown) => e);

expect((error as SSEUnsupportedError).status).toBe(504);
});

it('falls back (SSEUnsupportedError) when fetch rejects before streaming, with no status', async () => {
vi.stubGlobal('fetch', vi.fn().mockRejectedValue(new TypeError('network down')));

await expect(
streamOperation({ connection: { id: 'ins-1' }, body: { operation: 'deploy_component' } }),
).rejects.toBeInstanceOf(SSEUnsupportedError);
const error = await streamOperation({ connection: { id: 'ins-1' }, body: { operation: 'deploy_component' } })
.catch((e: unknown) => e);

expect(error).toBeInstanceOf(SSEUnsupportedError);
expect((error as SSEUnsupportedError).status).toBeUndefined();
});

it('throws SSEInconclusiveError when the stream ends without a terminal event', async () => {
Expand Down
9 changes: 5 additions & 4 deletions src/integrations/api/sse/streamOperation.ts
Original file line number Diff line number Diff line change
Expand Up @@ -25,8 +25,8 @@ export interface StreamOperationOptions {
*
* Failure modes are explicit (see `./errors`):
* - {@link SSEUnsupportedError} — couldn't establish a stream (bad response, wrong
* content-type, or connection error before any bytes). Caller should fall back to the
* buffered axios path.
* content-type, or connection error before any bytes). A read can fall back to the buffered
* axios path; the request may still have reached the server, so a write must check `status`.
* - {@link SSEOperationError} — a terminal `error` event arrived.
* - {@link SSEInconclusiveError} — the stream ended (or idled out) without a terminal
* event. The outcome is unknown; caller must poll to determine it, never assume success.
Expand Down Expand Up @@ -70,8 +70,8 @@ export async function streamOperation({
if (signal?.aborted) {
throw error;
}
// Anything else (network error, DNS, idle before headers) means we never got a
// stream going; let the caller fall back to the buffered request.
// Anything else (network error, DNS, idle before headers) means we never got a stream
// going, though the request may have been sent: there is no `status` to say otherwise.
throw new SSEUnsupportedError('Failed to open the SSE stream.', { cause: error });
}

Expand All @@ -81,6 +81,7 @@ export async function streamOperation({
await response.body?.cancel().catch(() => {});
throw new SSEUnsupportedError(
`Server did not return an event stream (status ${response.status}, content-type "${contentType}").`,
{ status: response.status },
);
}

Expand Down
Loading