diff --git a/.changeset/sandbox-unreachable-error.md b/.changeset/sandbox-unreachable-error.md new file mode 100644 index 0000000000..3ceed3b62f --- /dev/null +++ b/.changeset/sandbox-unreachable-error.md @@ -0,0 +1,12 @@ +--- +'@e2b/python-sdk': minor +'e2b': minor +--- + +Add `SandboxNotRunningError` / `SandboxNotRunningException` and `SandboxUnreachableError` / `SandboxUnreachableException` for envd requests that fail because the sandbox cannot be reached. Both extend `TimeoutError` / `TimeoutException`, which these cases raised before, so existing `instanceof TimeoutError` / `except TimeoutException` handling keeps working. + +- `SandboxNotRunning*` — the proxy answered `502 "The sandbox was not found"` (killed, paused or timed out), either directly or from the health probe the SDK runs after a dropped connection or an unexplained `Unavailable` (e.g. the sandbox was killed mid-command). Retrying will not help. +- `SandboxUnreachable*` — the sandbox is not confirmed stopped but cannot be reached: `502 "port is not open"`, other `502`s, or a connection failure whose health probe got no answer. The original error is the `cause` / `__cause__`. +- A connection failure whose health probe finds the sandbox running still raises the original transport error. The probe itself runs without connection retries. + +**Breaking (JS):** `commands.run` / `connect`, `pty.create` / `connect` and `files.watchDir` on a gone sandbox now throw `SandboxNotRunningError` instead of `SandboxNotFoundError`, like every other call. `SandboxNotFoundError` is only raised by the control plane API. diff --git a/packages/js-sdk/src/envd/api.ts b/packages/js-sdk/src/envd/api.ts index 673648c886..f92223c940 100644 --- a/packages/js-sdk/src/envd/api.ts +++ b/packages/js-sdk/src/envd/api.ts @@ -8,16 +8,19 @@ import { InvalidArgumentError, NotFoundError, NotEnoughSpaceError, - SandboxNotFoundError, - formatSandboxTimeoutError, + formatSandboxUnavailableError, + SandboxNotRunningError, AuthenticationError, RateLimitError, - TimeoutError, } from '../errors' import { StartResponse, ConnectResponse } from './process/process_pb' -import { Code, ConnectError } from '@connectrpc/connect' import { WatchDirResponse } from './filesystem/filesystem_pb' -import { isConnectionTerminatedMessage, SandboxHealthCheck } from './rpc' +import { + isConnectionTerminatedMessage, + resolveTransportFailure, + SandboxHealthCheck, +} from './rpc' +import { isNetworkError } from '../retry' type ApiError = { message?: string } | string @@ -27,7 +30,7 @@ const DEFAULT_ERROR_MAP: Record Error> = { 404: (message) => new NotFoundError(message), 429: (message) => new RateLimitError(`${message}: The requests are being rate limited.`), - 502: formatSandboxTimeoutError, + 502: formatSandboxUnavailableError, 507: (message) => new NotEnoughSpaceError(message), } @@ -37,59 +40,90 @@ const HEALTH_CHECK_TIMEOUT_MS = 5_000 * Probes the sandbox's envd health endpoint. * * @param envdApi - The envd API client of the sandbox. - * @returns `true` if the sandbox is running, `false` if it is not, `undefined` if its state could not be determined. + * @returns `true` if the sandbox is running, `false` if it is not, `undefined` if it answered but its state could not be determined. + * @throws The `fetch` failure when the sandbox could not be reached (no answer within the probe timeout), or the `SandboxUnreachableError` a request would get when the proxy reports the sandbox running but envd's port not open. */ export async function checkSandboxHealth( envdApi: EnvdApiClient ): Promise { - try { - const res = await envdApi.api.GET('/health', { - signal: AbortSignal.timeout(HEALTH_CHECK_TIMEOUT_MS), - }) + const res = await envdApi.api.GET('/health', { + signal: AbortSignal.timeout(HEALTH_CHECK_TIMEOUT_MS), + }) - if (res.response.status === 502) { + if (res.response.status === 502) { + const err = formatSandboxUnavailableError(await getEnvdApiErrorMessage(res)) + if (err instanceof SandboxNotRunningError) { return false } - if (res.response.ok) { - return true - } - - return undefined - } catch { - return undefined + throw err + } + if (res.response.ok) { + return true } + + return undefined +} + +/** + * Checks whether a `fetch` rejection is a transport-level failure of the request to + * the sandbox: the connection could not be established or was dropped mid-request + * (the latter surfaces with runtime-specific wording, e.g. undici's `terminated`). + */ +export function isFetchTransportFailure(err: unknown): err is Error { + return ( + err instanceof Error && + (isConnectionTerminatedMessage(err.message) || isNetworkError(err)) + ) } /** - * Handles transport-level fetch failures from envd API calls. When the connection was - * dropped mid-request, probes the sandbox health to tell apart the sandbox being killed - * from a transient network failure (e.g. a load balancer dropping the connection). + * Handles transport-level fetch failures from envd API calls. When the connection to + * the sandbox could not be established or was dropped mid-request, probes the sandbox + * health to tell apart the sandbox being killed from the sandbox being unreachable or + * a transient network failure (e.g. a load balancer dropping the connection). * * @param err - The caught error, expected to be a fetch transport failure. - * @param checkHealth - Probe returning whether the sandbox is running, or `undefined` when unknown. - * @returns A `TimeoutError` when the connection was terminated mid-request and the sandbox is confirmed gone, or the original error otherwise. + * @param checkHealth - Probe resolving to whether the sandbox is running (`undefined` when unknown) and rejecting when it cannot be reached. + * @returns A `SandboxNotRunningError` when the sandbox is confirmed gone, a `SandboxUnreachableError` when the probe got no answer either, or the original error otherwise. */ export async function handleEnvdApiFetchError( err: unknown, checkHealth?: SandboxHealthCheck ): Promise { - // A connection dropped mid-body surfaces as a fetch failure whose message varies - // by runtime (e.g. undici's 'terminated'); match every known variant. - if (err instanceof Error && isConnectionTerminatedMessage(err.message)) { - const running = checkHealth - ? await checkHealth().catch(() => undefined) - : undefined - - if (running === false) { - return new TimeoutError( - `${err.message}: The sandbox was killed or reached its end of life while the request was in flight.` - ) + if (checkHealth && isFetchTransportFailure(err)) { + const resolved = await resolveTransportFailure(err, checkHealth) + if (resolved) { + return resolved } } return err as Error } +/** + * Extracts the error message of a non-2xx envd API response: the `message` of its + * JSON body, else its text body, else its status text. + */ +export async function getEnvdApiErrorMessage(res: { + error?: ApiError + response: Response +}): Promise { + let message = + (typeof res.error === 'string' ? res.error : res.error?.message) ?? '' + + // openapi-fetch consumes the body when parsing the error, except for + // responses without content + if (!message && !res.response.bodyUsed) { + try { + message = await res.response.text() + } catch { + // ignore unreadable bodies + } + } + + return message || res.response.statusText +} + /** * Handles errors from envd API responses by mapping HTTP status codes to specific error types. * @@ -111,20 +145,7 @@ export async function handleEnvdApiError( return } - let message = - (typeof res.error === 'string' ? res.error : res.error?.message) ?? '' - - // openapi-fetch consumes the body when parsing the error, except for - // responses without content - if (!message && !res.response.bodyUsed) { - try { - message = await res.response.text() - } catch { - // ignore unreadable bodies - } - } - - message = message || res.response.statusText + const message = await getEnvdApiErrorMessage(res) // Check if a custom error mapping is provided for this error code if (errorMap && res.response.status in errorMap) { @@ -143,21 +164,9 @@ export async function handleEnvdApiError( export async function handleProcessStartEvent( events: AsyncIterable ) { - let startEvent: StartResponse | ConnectResponse - - try { - startEvent = (await events[Symbol.asyncIterator]().next()).value - } catch (err) { - if (err instanceof ConnectError) { - if (err.code === Code.Unavailable) { - throw new SandboxNotFoundError( - 'Sandbox is probably not running anymore' - ) - } - } - - throw err - } + const startEvent: StartResponse | ConnectResponse = ( + await events[Symbol.asyncIterator]().next() + ).value if (startEvent.event?.event.case !== 'start') { throw new Error('Expected start event') } @@ -168,21 +177,9 @@ export async function handleProcessStartEvent( export async function handleWatchDirStartEvent( events: AsyncIterable ) { - let startEvent: WatchDirResponse - - try { - startEvent = (await events[Symbol.asyncIterator]().next()).value - } catch (err) { - if (err instanceof ConnectError) { - if (err.code === Code.Unavailable) { - throw new SandboxNotFoundError( - 'Sandbox is probably not running anymore' - ) - } - } - - throw err - } + const startEvent: WatchDirResponse = ( + await events[Symbol.asyncIterator]().next() + ).value if (startEvent.event?.case !== 'start') { throw new Error('Expected start event') } diff --git a/packages/js-sdk/src/envd/rpc.ts b/packages/js-sdk/src/envd/rpc.ts index 6c4bb342cc..893af57caa 100644 --- a/packages/js-sdk/src/envd/rpc.ts +++ b/packages/js-sdk/src/envd/rpc.ts @@ -1,22 +1,29 @@ import { Code, ConnectError } from '@connectrpc/connect' +import { codeFromString } from '@connectrpc/connect/protocol-connect' import { runtime } from '../utils' import { compareVersions } from 'compare-versions' import { defaultUsername } from '../connectionConfig' import { AuthenticationError, - formatSandboxTimeoutError, + formatSandboxUnavailableError, InvalidArgumentError, + isSandboxNotFoundMessage, + isSandboxPortNotOpenMessage, NotFoundError, RateLimitError, SandboxError, + SandboxNotRunningError, + SandboxUnreachableError, TimeoutError, } from '../errors' +import { isNetworkError } from '../retry' import { ENVD_DEFAULT_USER } from './versions' /** - * Result of a sandbox health probe: `true` if the sandbox is running, `false` if it is not, - * `undefined` if its state could not be determined. + * Sandbox health probe: resolves to `true` if the sandbox is running, `false` if it + * is not, `undefined` if it answered but its state could not be determined, and + * rejects when the sandbox could not be reached at all (no answer). */ export type SandboxHealthCheck = () => Promise @@ -68,6 +75,92 @@ function isConnectionTerminatedError(err: unknown): boolean { ) } +/** + * Checks whether the error is a transport-level failure of the request to the sandbox: + * the connection could not be established or failed with a browser's opaque network + * error (connect wraps the `fetch` rejection as `Code.Unknown` with the failure as + * `cause`) or was dropped mid-request. Errors + * envd responded with are never transport failures. + */ +export function isTransportFailure(err: unknown): boolean { + if (isConnectionTerminatedError(err)) { + return true + } + + return ( + err instanceof ConnectError && + err.code === Code.Unknown && + isNetworkError(err.cause) + ) +} + +/** + * Checks whether the error is an `Unavailable` whose message the proxy's answers do + * not explain — neither the sandbox not found nor its port not open — e.g. envd + * ending a stream with "the connection to sandbox ... ended before the stream + * completed" when the sandbox is killed mid-command. Whether the sandbox is gone + * can only be told by probing it. + */ +export function isAmbiguousUnavailable(err: unknown): boolean { + return ( + err instanceof ConnectError && + err.code === Code.Unavailable && + !isSandboxNotFoundMessage(err.rawMessage) && + !isSandboxPortNotOpenMessage(err.rawMessage) + ) +} + +/** + * Builds the error for a request that failed at the connection level when the + * follow-up health probe failed too. A probe answered by the proxy with the sandbox + * running but envd's port not open is already a `SandboxUnreachableError` and is + * kept, with the failed request as its cause; otherwise the probe's failure is + * reported alongside the request's, which becomes the cause. + */ +export function formatSandboxUnreachableError( + err: Error, + probeErr: unknown +): Error { + if (probeErr instanceof SandboxUnreachableError) { + probeErr.cause = err + return probeErr + } + const probeMessage = + probeErr instanceof Error ? probeErr.message : String(probeErr) + return new SandboxUnreachableError( + `${err.message}: The sandbox could not be reached and its health probe failed too (${probeMessage}). It was not confirmed to be stopped — this is likely a network issue or envd inside the sandbox not being up yet; check the sandbox state with 'Sandbox.getInfo()'.`, + { cause: err } + ) +} + +/** + * Runs the health probe for a request that failed at the connection level and + * returns the typed error for the outcome: a `SandboxNotRunningError` when the + * sandbox is confirmed gone, a `SandboxUnreachableError` when the probe got no + * answer either, or `undefined` when the sandbox answered (the failure was + * transient and the original error applies). + */ +export async function resolveTransportFailure( + err: Error, + checkHealth: SandboxHealthCheck +): Promise { + let running: boolean | undefined + try { + running = await checkHealth() + } catch (probeErr) { + return formatSandboxUnreachableError(err, probeErr) + } + + if (running === false) { + return new SandboxNotRunningError( + `${err.message}: The sandbox was killed or reached its end of life while the request was in flight.`, + { cause: err } + ) + } + + return undefined +} + const DEFAULT_ERROR_MAP: Partial Error>> = { [Code.InvalidArgument]: (message) => new InvalidArgumentError(message), [Code.Unauthenticated]: (message) => new AuthenticationError(message), @@ -76,7 +169,7 @@ const DEFAULT_ERROR_MAP: Partial Error>> = { new RateLimitError( `${message}: Rate limit exceeded, please try again later.` ), - [Code.Unavailable]: formatSandboxTimeoutError, + [Code.Unavailable]: formatSandboxUnavailableError, [Code.Canceled]: (message) => new TimeoutError( `${message}: This error is likely due to exceeding 'requestTimeoutMs'. You can pass the request timeout value as an option when making the request.` @@ -117,14 +210,20 @@ export function handleRpcError( } /** - * Like {@link handleRpcError}, but when the connection to the sandbox was dropped - * mid-request it probes the sandbox health to tell apart the sandbox being killed - * from a transient network failure (e.g. a load balancer dropping the connection). - * When the probe confirms the sandbox is gone, a `TimeoutError` is returned — - * consistent with how requests to an already-dead sandbox surface. + * Like {@link handleRpcError}, but when the request failed at the connection level + * (the connection to the sandbox could not be established or was dropped + * mid-request), or envd/the proxy answered an `Unavailable` that does not say + * whether the sandbox is gone ({@link isAmbiguousUnavailable}), it probes the + * sandbox health to tell apart the sandbox being killed from the sandbox being + * unreachable or a transient network failure (e.g. a load balancer dropping the + * connection). When the probe confirms the sandbox is gone, a + * `SandboxNotRunningError` is returned — the same error a request answered by the + * proxy with the sandbox not found gets; when the probe gets no answer either, a + * `SandboxUnreachableError`; otherwise the error maps as usual (an ambiguous + * `Unavailable` to a `SandboxUnreachableError` with the sandbox state unknown). * * @param err - The caught error, expected to be a `ConnectError` from the gRPC transport. - * @param checkHealth - Probe returning whether the sandbox is running, or `undefined` when unknown. + * @param checkHealth - Probe resolving to whether the sandbox is running (`undefined` when unknown) and rejecting when it cannot be reached. * @param errorMap - Optional map of gRPC `Code` values to error factory functions that override the defaults. * @returns The corresponding `Error` instance. */ @@ -133,13 +232,13 @@ export async function handleRpcErrorWithHealthCheck( checkHealth?: SandboxHealthCheck, errorMap?: Partial Error>> ): Promise { - if (isConnectionTerminatedError(err) && checkHealth) { - const running = await checkHealth().catch(() => undefined) - - if (running === false) { - return new TimeoutError( - `${(err as ConnectError).message}: The sandbox was killed or reached its end of life while the request was in flight.` - ) + if (checkHealth && (isTransportFailure(err) || isAmbiguousUnavailable(err))) { + const resolved = await resolveTransportFailure( + err as ConnectError, + checkHealth + ) + if (resolved) { + return resolved } } @@ -180,3 +279,46 @@ export function authenticationHeader( return { Authorization: `Basic ${encoded}` } } + +/** + * Rejects a plain (non-Connect-encoded) 502 answered by the proxy in front of envd — + * the sandbox is gone or envd's port is not open — as a `ConnectError` carrying the + * proxy's message. connect reads that message for unary RPCs only; a streaming RPC + * would otherwise surface it as a bare `HTTP 502`. + */ +export async function rejectProxyUnavailableResponse( + res: Response +): Promise { + if (res.status !== 502) { + return res + } + + let message = '' + try { + message = await res.text() + } catch { + // unreadable body + } + try { + const body = JSON.parse(message) + if ( + typeof body?.code === 'string' && + codeFromString(body.code) !== undefined + ) { + // Connect-encoded error (declared string `code`, e.g. "permission_denied") — + // let the transport decode it with that code. + return new Response(message, { + status: res.status, + statusText: res.statusText, + headers: res.headers, + }) + } + if (typeof body?.message === 'string') { + message = body.message + } + } catch { + // not a JSON body: keep the proxy's plain-text message + } + + throw new ConnectError(message || `HTTP ${res.status}`, Code.Unavailable) +} diff --git a/packages/js-sdk/src/errors.ts b/packages/js-sdk/src/errors.ts index f148935067..176fed680f 100644 --- a/packages/js-sdk/src/errors.ts +++ b/packages/js-sdk/src/errors.ts @@ -1,7 +1,38 @@ -// This is the message for the sandbox timeout error when the response code is 502/Unavailable -export function formatSandboxTimeoutError(message: string) { - return new TimeoutError( - `${message}: This error is likely due to sandbox timeout. You can modify the sandbox timeout by passing 'timeoutMs' when starting the sandbox or calling '.setTimeout' on the sandbox with the desired timeout.` +// Messages of the 502 the proxy in front of envd answers with when the sandbox does +// not exist anymore (killed or reached its timeout) or when it is running but envd's +// port is not open (e.g. its network is down). +const SANDBOX_NOT_FOUND_MESSAGE = 'was not found' +const SANDBOX_PORT_NOT_OPEN_MESSAGE = 'port is not open' + +export function isSandboxNotFoundMessage(message: string): boolean { + return message.includes(SANDBOX_NOT_FOUND_MESSAGE) +} + +export function isSandboxPortNotOpenMessage(message: string): boolean { + return message.includes(SANDBOX_PORT_NOT_OPEN_MESSAGE) +} + +/** + * Maps a 502/Unavailable answered by the proxy in front of envd: the sandbox was not + * found (killed or reached its timeout) is a `SandboxNotRunningError`; the sandbox + * running but envd's port not open (e.g. its network is down) a + * `SandboxUnreachableError`; any other message (e.g. envd ending a stream because + * the connection to the sandbox was lost) a `SandboxUnreachableError` too, with the + * sandbox state reported as unknown. + */ +export function formatSandboxUnavailableError(message: string): Error { + if (isSandboxNotFoundMessage(message)) { + return new SandboxNotRunningError( + `${message}: The sandbox was killed or reached its timeout. You can modify the sandbox timeout by passing 'timeoutMs' when starting the sandbox or calling '.setTimeout' on the sandbox with the desired timeout.` + ) + } + if (isSandboxPortNotOpenMessage(message)) { + return new SandboxUnreachableError( + `${message}: The sandbox is running but envd inside it could not be reached (e.g. its network is down). Check the sandbox state with 'Sandbox.getInfo()'.` + ) + } + return new SandboxUnreachableError( + `${message}: The sandbox could not be reached and its state is unknown — it may have been killed or reached its timeout, or the connection to it was lost. Check the sandbox state with 'Sandbox.getInfo()'.` ) } @@ -95,6 +126,58 @@ export class SandboxNotFoundError extends NotFoundError { } } +/** + * Thrown when the sandbox is not running anymore: the proxy in front of it + * answered a request (or the health probe run after a request failed at the + * connection level) with the sandbox not found — it was killed, paused or + * reached its timeout. Retrying the request will not help; create or resume a + * sandbox. + * + * The split from `SandboxUnreachableError` is only as good as the proxy's answer + * and the health probe: a sandbox that is gone but could not be asked about is a + * `SandboxUnreachableError`. `SandboxNotFoundError` is only raised by the control + * plane API (e.g. `Sandbox.connect()` to a sandbox that does not exist). + * + * Subclass of `TimeoutError`, which this case surfaced as before. + */ +export class SandboxNotRunningError extends TimeoutError { + constructor(message: string, options?: ErrorOptions) { + super(message) + this.name = 'SandboxNotRunningError' + if (options?.cause !== undefined) { + this.cause = options.cause + } + } +} + +/** + * Thrown when the sandbox could not be reached while it is not confirmed to be + * stopped: the proxy in front of the sandbox reports it running but envd's port + * not open (e.g. the sandbox's network is down), the proxy or envd answered + * with an unavailability the health probe could not resolve (e.g. the connection + * to the sandbox was lost mid-stream while the probe found it running), or a + * request failed at the connection level (the connection could not be + * established or was dropped mid-request) and the follow-up health probe got + * no answer from the sandbox either. A sandbox confirmed stopped (or paused) is + * a `SandboxNotRunningError` — but only as far as the probe can tell: a sandbox + * that is gone and could not be asked about surfaces here. + * + * This usually means the sandbox's network is down, envd inside it is not up + * (yet), or a network issue between the client and the sandbox. Check the + * sandbox state with `Sandbox.getInfo()`. + * + * Subclass of `TimeoutError`, which these cases surfaced as before. + */ +export class SandboxUnreachableError extends TimeoutError { + constructor(message: string, options?: ErrorOptions) { + super(message) + this.name = 'SandboxUnreachableError' + if (options?.cause !== undefined) { + this.cause = options.cause + } + } +} + /** * Thrown when authentication fails. */ diff --git a/packages/js-sdk/src/index.ts b/packages/js-sdk/src/index.ts index afe6917fcf..1799643be3 100644 --- a/packages/js-sdk/src/index.ts +++ b/packages/js-sdk/src/index.ts @@ -18,6 +18,8 @@ export { NotFoundError, SandboxError, SandboxNotFoundError, + SandboxNotRunningError, + SandboxUnreachableError, TemplateError, TimeoutError, RateLimitError, diff --git a/packages/js-sdk/src/retry.ts b/packages/js-sdk/src/retry.ts index c0eff5f856..903edce8bb 100644 --- a/packages/js-sdk/src/retry.ts +++ b/packages/js-sdk/src/retry.ts @@ -25,6 +25,10 @@ const CONNECTION_ERROR_CODES = new Set([ const CONNECTION_ERROR_SYSCALLS = new Set(['connect', 'getaddrinfo']) // Deno reports hyper's connect-phase failures as `client error (Connect)`. const DENO_CONNECTION_ERROR = /client error \(Connect\)/ +// Browsers reject every network failure with an opaque `TypeError`: +// Chrome `Failed to fetch`, Firefox `NetworkError when attempting to fetch resource.`, Safari `Load failed` +const OPAQUE_NETWORK_ERROR = + /^(Failed to fetch|Load failed|NetworkError when attempting to fetch resource\.?)$/ // GET/PUT/PATCH/DELETE are idempotent by HTTP semantics. A POST is retried // after a 502 or a network error that may have occurred once the request was // written only if it is listed here: a replay is a no-op or fails with a @@ -89,6 +93,16 @@ export function isConnectionError(error: unknown, depth = 0): boolean { return isConnectionError(error.cause, depth + 1) } +/** + * Whether `error` (thrown by `fetch`) is a network failure: a connection error + * (see {@link isConnectionError}) or the opaque `TypeError` browsers raise for any + * network failure, which carries no code or syscall. + */ +export function isNetworkError(error: unknown): boolean { + if (isConnectionError(error)) return true + return error instanceof Error && OPAQUE_NETWORK_ERROR.test(error.message) +} + /** * Whether `request` may be sent again after a 502 or a network error that may * have occurred once the request was written (see `REPLAYABLE_OPERATIONS`). diff --git a/packages/js-sdk/src/sandbox/index.ts b/packages/js-sdk/src/sandbox/index.ts index 5f0c6f2f53..778c9859fd 100644 --- a/packages/js-sdk/src/sandbox/index.ts +++ b/packages/js-sdk/src/sandbox/index.ts @@ -7,6 +7,7 @@ import { Username, } from '../connectionConfig' import { EnvdApiClient, handleEnvdApiError } from '../envd/api' +import { rejectProxyUnavailableResponse } from '../envd/rpc' import { createEnvdFetch, createEnvdRpcFetch } from '../envd/http2' import { createRpcLogger } from '../logs' import { Commands, Pty } from './commands' @@ -194,7 +195,7 @@ export class Sandbox extends SandboxApi { redirect: 'follow', } - return envdRpcFetch(url, options) + return envdRpcFetch(url, options).then(rejectProxyUnavailableResponse) }, }) diff --git a/packages/js-sdk/tests/envd/handleEnvdApiError.test.ts b/packages/js-sdk/tests/envd/handleEnvdApiError.test.ts index 4140741946..10bfa0a7ea 100644 --- a/packages/js-sdk/tests/envd/handleEnvdApiError.test.ts +++ b/packages/js-sdk/tests/envd/handleEnvdApiError.test.ts @@ -1,5 +1,9 @@ import { assert, test, describe } from 'vitest' -import { handleEnvdApiError, handleEnvdApiFetchError } from '../../src/envd/api' +import { + checkSandboxHealth, + handleEnvdApiError, + handleEnvdApiFetchError, +} from '../../src/envd/api' import { AuthenticationError, InvalidArgumentError, @@ -7,9 +11,29 @@ import { NotFoundError, RateLimitError, SandboxError, + SandboxNotRunningError, + SandboxUnreachableError, TimeoutError, } from '../../src/errors' +// undici surfaces a refused connection as `fetch failed` with the syscall error as `cause` +function connectRefused(): TypeError { + return Object.assign(new TypeError('fetch failed'), { + cause: Object.assign(new Error('connect ECONNREFUSED'), { + code: 'ECONNREFUSED', + }), + }) +} + +// Body of the proxy's 502 when the sandbox is running but nothing listens on envd's port +const PORT_CLOSED = 'The sandbox is running but port is not open' + +function healthApi(status: number, error?: { message?: string } | string) { + return { + api: { GET: async () => createMockResponse(status, error) }, + } as unknown as Parameters[0] +} + function createMockResponse( status: number, error?: { message?: string } | string @@ -83,6 +107,30 @@ describe('handleEnvdApiError', () => { assert.include(err?.message, 'rate limited') }) + test('returns SandboxUnreachableError for 502 with the sandbox running but its port closed', async () => { + const res = createMockResponse(502, { message: PORT_CLOSED }) + const err = await handleEnvdApiError(res) + assert.instanceOf(err, SandboxUnreachableError) + assert.include(err!.message, 'could not be reached') + }) + + test('returns SandboxNotRunningError for 502 for the sandbox not found', async () => { + const res = createMockResponse(502, { + message: 'The sandbox was not found', + }) + const err = await handleEnvdApiError(res) + assert.instanceOf(err, SandboxNotRunningError) + assert.instanceOf(err, TimeoutError) + assert.notInstanceOf(err, SandboxUnreachableError) + assert.include(err!.message, 'The sandbox was not found') + }) + + test('returns SandboxNotRunningError for a plain-text 502 for the sandbox not found', async () => { + const res = createMockResponse(502, 'The sandbox was not found') + const err = await handleEnvdApiError(res) + assert.instanceOf(err, SandboxNotRunningError) + }) + test('returns TimeoutError for 502', async () => { const res = createMockResponse(502, { message: 'Bad gateway' }) const err = await handleEnvdApiError(res) @@ -104,19 +152,35 @@ describe('handleEnvdApiError', () => { }) describe('handleEnvdApiFetchError', () => { + test.each([ + ['Chrome', 'Failed to fetch'], + ['Firefox', 'NetworkError when attempting to fetch resource.'], + ['Safari', 'Load failed'], + ])( + "returns a SandboxUnreachableError for %s's opaque network error when the probe gets no answer", + async (_, message) => { + const original = new TypeError(message) + const err = await handleEnvdApiFetchError(original, async () => { + throw new TypeError(message) + }) + assert.instanceOf(err, SandboxUnreachableError) + assert.strictEqual(err.cause, original) + } + ) + test('returns the original error for terminated fetch without a health check', async () => { const original = new TypeError('terminated') const err = await handleEnvdApiFetchError(original) assert.strictEqual(err, original) }) - test('returns a TimeoutError when the health check says the sandbox is not running', async () => { - const err = await handleEnvdApiFetchError( - new TypeError('terminated'), - async () => false - ) - assert.instanceOf(err, TimeoutError) + test('returns a SandboxNotRunningError when the health check says the sandbox is not running', async () => { + const original = new TypeError('terminated') + const err = await handleEnvdApiFetchError(original, async () => false) + assert.instanceOf(err, SandboxNotRunningError) + assert.notInstanceOf(err, SandboxUnreachableError) assert.include(err.message, 'sandbox was killed or reached its end of life') + assert.strictEqual(err.cause, original) }) // Each JS runtime surfaces a dropped connection with different wording, and not @@ -132,7 +196,7 @@ describe('handleEnvdApiFetchError', () => { for (const [runtime, error] of Object.entries(runtimeTerminatedErrors)) { test(`treats the ${runtime} dropped-connection error as terminated`, async () => { const err = await handleEnvdApiFetchError(error, async () => false) - assert.instanceOf(err, TimeoutError) + assert.instanceOf(err, SandboxNotRunningError) assert.include( err.message, 'sandbox was killed or reached its end of life' @@ -151,4 +215,100 @@ describe('handleEnvdApiFetchError', () => { const err = await handleEnvdApiFetchError(original) assert.strictEqual(err, original) }) + + test('returns a SandboxUnreachableError when the health check itself fails', async () => { + const original = new TypeError('terminated') + const err = await handleEnvdApiFetchError(original, async () => { + throw new TypeError('fetch failed') + }) + assert.instanceOf(err, SandboxUnreachableError) + assert.include(err.message, 'could not be reached') + assert.include(err.message, 'fetch failed') + assert.strictEqual(err.cause, original) + }) + + test('runs the health check when the connection could not be established', async () => { + let called = false + const original = connectRefused() + const err = await handleEnvdApiFetchError(original, async () => { + called = true + return true + }) + assert.isTrue(called) + assert.strictEqual(err, original) + }) + + test('returns a SandboxNotRunningError for a refused connection when the sandbox is gone', async () => { + const err = await handleEnvdApiFetchError( + connectRefused(), + async () => false + ) + assert.instanceOf(err, SandboxNotRunningError) + }) + + test('returns a SandboxUnreachableError for a refused connection when the probe gets no answer', async () => { + const err = await handleEnvdApiFetchError(connectRefused(), async () => { + throw new TypeError('fetch failed') + }) + assert.instanceOf(err, SandboxUnreachableError) + }) + + test('does not run the health check for a non-connection fetch failure', async () => { + const original = new TypeError('fetch failed') + const err = await handleEnvdApiFetchError(original, async () => { + throw new Error('health check should not run') + }) + assert.strictEqual(err, original) + }) +}) + +describe('checkSandboxHealth', () => { + test('reports the sandbox running', async () => { + assert.strictEqual(await checkSandboxHealth(healthApi(204)), true) + }) + + test('reports the sandbox gone for 502', async () => { + const api = healthApi(502, { message: 'The sandbox was not found' }) + assert.strictEqual(await checkSandboxHealth(api), false) + }) + + test('reports the sandbox gone for a plain-text 502', async () => { + const api = healthApi(502, 'The sandbox was not found') + assert.strictEqual(await checkSandboxHealth(api), false) + }) + + test('is inconclusive for other statuses', async () => { + assert.strictEqual(await checkSandboxHealth(healthApi(500)), undefined) + }) + + test('throws SandboxUnreachableError for a 502 with an unrecognized message', async () => { + try { + await checkSandboxHealth(healthApi(502, { message: 'Bad gateway' })) + assert.fail('expected the probe to throw') + } catch (err) { + assert.instanceOf(err, SandboxUnreachableError) + assert.notInstanceOf(err, SandboxNotRunningError) + } + }) + + test('throws SandboxUnreachableError for 502 with the sandbox running but its port closed', async () => { + const api = healthApi(502, { message: PORT_CLOSED }) + try { + await checkSandboxHealth(api) + assert.fail('expected the probe to throw') + } catch (err) { + assert.instanceOf(err, SandboxUnreachableError) + assert.include((err as Error).message, PORT_CLOSED) + } + }) + + test('keeps the probe SandboxUnreachableError with the failed request as cause', async () => { + const original = connectRefused() + const err = await handleEnvdApiFetchError(original, () => + checkSandboxHealth(healthApi(502, { message: PORT_CLOSED })) + ) + assert.instanceOf(err, SandboxUnreachableError) + assert.include(err.message, PORT_CLOSED) + assert.strictEqual(err.cause, original) + }) }) diff --git a/packages/js-sdk/tests/envd/handleRpcError.test.ts b/packages/js-sdk/tests/envd/handleRpcError.test.ts index f90f893e64..a3ffcc8bf3 100644 --- a/packages/js-sdk/tests/envd/handleRpcError.test.ts +++ b/packages/js-sdk/tests/envd/handleRpcError.test.ts @@ -3,16 +3,124 @@ import { Code, ConnectError } from '@connectrpc/connect' import { handleRpcError, handleRpcErrorWithHealthCheck, + isAmbiguousUnavailable, + isTransportFailure, + rejectProxyUnavailableResponse, } from '../../src/envd/rpc' +import { + handleProcessStartEvent, + handleWatchDirStartEvent, +} from '../../src/envd/api' import { AuthenticationError, InvalidArgumentError, NotFoundError, RateLimitError, SandboxError, + SandboxNotFoundError, + SandboxNotRunningError, + SandboxUnreachableError, TimeoutError, } from '../../src/errors' +// connect wraps a `fetch` rejection as `Code.Unknown` with the failure as `cause` +function connectRefused(): ConnectError { + const cause = Object.assign(new TypeError('fetch failed'), { + cause: Object.assign(new Error('connect ECONNREFUSED'), { + code: 'ECONNREFUSED', + }), + }) + return new ConnectError( + cause.message, + Code.Unknown, + undefined, + undefined, + cause + ) +} + +// Body of the proxy's 502 when the sandbox is running but nothing listens on envd's port +const PORT_CLOSED = 'The sandbox is running but port is not open' +// envd ends a stream with this Unavailable when the sandbox is killed mid-command +const STREAM_ENDED = + 'the connection to sandbox ended before the stream completed' + +function neverProbe(): Promise { + throw new Error('health check should not run') +} + +describe('Unavailable with the sandbox not found', () => { + test('returns SandboxNotRunningError, not SandboxNotFoundError, from every call', () => { + const err = handleRpcError( + new ConnectError('The sandbox was not found', Code.Unavailable) + ) + assert.instanceOf(err, SandboxNotRunningError) + assert.notInstanceOf(err, SandboxNotFoundError) + }) +}) + +describe('ambiguous Unavailable', () => { + test('is an Unavailable that says neither not found nor port not open', () => { + assert.isTrue( + isAmbiguousUnavailable(new ConnectError(STREAM_ENDED, Code.Unavailable)) + ) + assert.isFalse( + isAmbiguousUnavailable( + new ConnectError('The sandbox was not found', Code.Unavailable) + ) + ) + assert.isFalse( + isAmbiguousUnavailable(new ConnectError(PORT_CLOSED, Code.Unavailable)) + ) + assert.isFalse( + isAmbiguousUnavailable(new ConnectError(STREAM_ENDED, Code.Unknown)) + ) + }) + + test('is probed and returns SandboxNotRunningError when the sandbox is gone', async () => { + const original = new ConnectError(STREAM_ENDED, Code.Unavailable) + const err = await handleRpcErrorWithHealthCheck(original, async () => false) + assert.instanceOf(err, SandboxNotRunningError) + assert.notInstanceOf(err, SandboxUnreachableError) + assert.strictEqual(err.cause, original) + }) + + test('returns SandboxUnreachableError with the state unknown when the sandbox is running', async () => { + const err = await handleRpcErrorWithHealthCheck( + new ConnectError(STREAM_ENDED, Code.Unavailable), + async () => true + ) + assert.instanceOf(err, SandboxUnreachableError) + assert.notInstanceOf(err, SandboxNotRunningError) + assert.include(err.message, 'state is unknown') + assert.notInclude(err.message, 'is running but') + }) + + test('returns SandboxUnreachableError when the probe itself fails', async () => { + const original = new ConnectError(STREAM_ENDED, Code.Unavailable) + const err = await handleRpcErrorWithHealthCheck(original, async () => { + throw new Error('health check failed') + }) + assert.instanceOf(err, SandboxUnreachableError) + assert.include(err.message, 'health check failed') + assert.strictEqual(err.cause, original) + }) + + test('is not probed for the sandbox not found or its port not open', async () => { + const notFound = await handleRpcErrorWithHealthCheck( + new ConnectError('The sandbox was not found', Code.Unavailable), + neverProbe + ) + assert.instanceOf(notFound, SandboxNotRunningError) + const portClosed = await handleRpcErrorWithHealthCheck( + new ConnectError(PORT_CLOSED, Code.Unavailable), + neverProbe + ) + assert.instanceOf(portClosed, SandboxUnreachableError) + assert.include(portClosed.message, 'is running but') + }) +}) + describe('handleRpcError', () => { test('returns InvalidArgumentError for InvalidArgument', () => { const err = handleRpcError(new ConnectError('bad', Code.InvalidArgument)) @@ -37,6 +145,28 @@ describe('handleRpcError', () => { assert.include(err.message, 'Rate limit') }) + test('returns SandboxUnreachableError for Unavailable with the sandbox running but its port closed', () => { + const err = handleRpcError(new ConnectError(PORT_CLOSED, Code.Unavailable)) + assert.instanceOf(err, SandboxUnreachableError) + assert.include(err.message, 'could not be reached') + }) + + test('returns SandboxNotRunningError for Unavailable with the sandbox not found', () => { + const err = handleRpcError( + new ConnectError('The sandbox was not found', Code.Unavailable) + ) + assert.instanceOf(err, SandboxNotRunningError) + assert.instanceOf(err, TimeoutError) + assert.notInstanceOf(err, SandboxUnreachableError) + assert.include(err.message, 'The sandbox was not found') + }) + + test('returns SandboxUnreachableError, not SandboxNotRunningError, for Unavailable with the port closed', () => { + const err = handleRpcError(new ConnectError(PORT_CLOSED, Code.Unavailable)) + assert.instanceOf(err, SandboxUnreachableError) + assert.notInstanceOf(err, SandboxNotRunningError) + }) + test('returns TimeoutError for Unavailable', () => { const err = handleRpcError(new ConnectError('gone', Code.Unavailable)) assert.instanceOf(err, TimeoutError) @@ -73,13 +203,27 @@ describe('handleRpcErrorWithHealthCheck', () => { Browser: 'network error', } - test('returns a TimeoutError when the health check says the sandbox is not running', async () => { + test('does not probe health on an Unavailable for the sandbox not found', async () => { + let probed = false const err = await handleRpcErrorWithHealthCheck( - terminated(), - async () => false + new ConnectError('The sandbox was not found', Code.Unavailable), + async () => { + probed = true + return true + } ) + assert.isFalse(probed) + assert.instanceOf(err, SandboxNotRunningError) + }) + + test('returns a SandboxNotRunningError when the health check says the sandbox is not running', async () => { + const original = terminated() + const err = await handleRpcErrorWithHealthCheck(original, async () => false) + assert.instanceOf(err, SandboxNotRunningError) assert.instanceOf(err, TimeoutError) + assert.notInstanceOf(err, SandboxUnreachableError) assert.include(err.message, 'sandbox was killed or reached its end of life') + assert.strictEqual(err.cause, original) }) for (const [runtime, message] of Object.entries(runtimeTerminatedMessages)) { @@ -88,7 +232,7 @@ describe('handleRpcErrorWithHealthCheck', () => { new ConnectError(message, Code.Unknown), async () => false ) - assert.instanceOf(err, TimeoutError) + assert.instanceOf(err, SandboxNotRunningError) assert.include( err.message, 'sandbox was killed or reached its end of life' @@ -116,13 +260,97 @@ describe('handleRpcErrorWithHealthCheck', () => { assert.notInclude(err.message, 'killed') }) - test('falls back to the generic mapping when the health check itself fails', async () => { - const err = await handleRpcErrorWithHealthCheck(terminated(), async () => { + test('keeps the probe SandboxUnreachableError with the failed request as cause', async () => { + const original = terminated() + const probeErr = new SandboxUnreachableError(PORT_CLOSED) + const err = await handleRpcErrorWithHealthCheck(original, async () => { + throw probeErr + }) + assert.strictEqual(err, probeErr) + assert.strictEqual(err.cause, original) + }) + + test('returns a SandboxUnreachableError when the health check itself fails', async () => { + const original = terminated() + const err = await handleRpcErrorWithHealthCheck(original, async () => { throw new Error('health check failed') }) + assert.instanceOf(err, SandboxUnreachableError) + assert.include(err.message, 'could not be reached') + assert.include(err.message, 'health check failed') + assert.strictEqual(err.cause, original) + }) + + test.each([ + ['Chrome', 'Failed to fetch'], + ['Firefox', 'NetworkError when attempting to fetch resource.'], + ['Safari', 'Load failed'], + ])( + "treats %s's opaque network error as a transport failure", + async (_, message) => { + const cause = new TypeError(message) + const err = new ConnectError( + message, + Code.Unknown, + undefined, + undefined, + cause + ) + assert.isTrue(isTransportFailure(err)) + + const resolved = await handleRpcErrorWithHealthCheck(err, async () => { + throw new TypeError(message) + }) + assert.instanceOf(resolved, SandboxUnreachableError) + assert.strictEqual(resolved.cause, err) + } + ) + + test('treats a refused connection as a transport failure', () => { + assert.isTrue(isTransportFailure(connectRefused())) + assert.isFalse(isTransportFailure(new ConnectError('nope', Code.NotFound))) + assert.isFalse( + isTransportFailure( + new ConnectError( + 'boom', + Code.Unknown, + undefined, + undefined, + new Error('boom') + ) + ) + ) + }) + + test('runs the health check when the connection could not be established', async () => { + let called = false + const original = connectRefused() + const err = await handleRpcErrorWithHealthCheck(original, async () => { + called = true + return true + }) + assert.isTrue(called) assert.instanceOf(err, SandboxError) assert.notInstanceOf(err, TimeoutError) - assert.notInclude(err.message, 'killed') + assert.notInstanceOf(err, SandboxUnreachableError) + }) + + test('returns a SandboxNotRunningError for a refused connection when the sandbox is gone', async () => { + const err = await handleRpcErrorWithHealthCheck( + connectRefused(), + async () => false + ) + assert.instanceOf(err, SandboxNotRunningError) + }) + + test('returns a SandboxUnreachableError for a refused connection when the probe gets no answer', async () => { + const err = await handleRpcErrorWithHealthCheck( + connectRefused(), + async () => { + throw new TypeError('fetch failed') + } + ) + assert.instanceOf(err, SandboxUnreachableError) }) test('does not run the health check for other errors', async () => { @@ -138,3 +366,116 @@ describe('handleRpcErrorWithHealthCheck', () => { assert.isFalse(called) }) }) + +describe('start event handlers', () => { + async function* failing(): AsyncGenerator { + throw new ConnectError('port is not open', Code.Unavailable) + } + + test('handleProcessStartEvent rethrows the ConnectError so the caller maps it', async () => { + await handleProcessStartEvent(failing()).then( + () => assert.fail('expected rejection'), + (err) => { + assert.instanceOf(err, ConnectError) + assert.equal(err.code, Code.Unavailable) + assert.instanceOf(handleRpcError(err), TimeoutError) + } + ) + }) + + test('handleWatchDirStartEvent rethrows the ConnectError so the caller maps it', async () => { + await handleWatchDirStartEvent(failing()).then( + () => assert.fail('expected rejection'), + (err) => { + assert.instanceOf(err, ConnectError) + assert.equal(err.code, Code.Unavailable) + assert.instanceOf(handleRpcError(err), TimeoutError) + } + ) + }) +}) + +describe('rejectProxyUnavailableResponse', () => { + test('passes other responses through', async () => { + const res = new Response('ok', { status: 200 }) + assert.strictEqual(await rejectProxyUnavailableResponse(res), res) + }) + + test('rejects a proxy 502 as Unavailable with the body message', async () => { + const res = new Response( + JSON.stringify({ message: PORT_CLOSED, port: 49983, code: 502 }), + { status: 502, headers: { 'Content-Type': 'application/json' } } + ) + try { + await rejectProxyUnavailableResponse(res) + assert.fail('expected a rejection') + } catch (err) { + assert.instanceOf(err, ConnectError) + assert.strictEqual((err as ConnectError).code, Code.Unavailable) + assert.strictEqual((err as ConnectError).rawMessage, PORT_CLOSED) + } + }) + + test('passes a Connect-encoded 502 error through for the transport to decode', async () => { + const body = JSON.stringify({ + code: 'permission_denied', + message: 'access denied', + }) + const res = new Response(body, { + status: 502, + headers: { 'Content-Type': 'application/json' }, + }) + const out = await rejectProxyUnavailableResponse(res) + assert.strictEqual(out.status, 502) + assert.strictEqual(out.headers.get('Content-Type'), 'application/json') + assert.deepEqual(await out.json(), JSON.parse(body)) + }) + + test('rejects a 502 whose string code is not a Connect code as Unavailable with the body message', async () => { + const res = new Response( + JSON.stringify({ + code: 'BAD_GATEWAY', + message: 'The sandbox was not found', + }), + { status: 502, headers: { 'Content-Type': 'application/json' } } + ) + try { + await rejectProxyUnavailableResponse(res) + assert.fail('expected a rejection') + } catch (err) { + assert.instanceOf(err, ConnectError) + assert.strictEqual((err as ConnectError).code, Code.Unavailable) + assert.strictEqual( + (err as ConnectError).rawMessage, + 'The sandbox was not found' + ) + } + }) + + test('keeps the plain-text body of a 502 as the message', async () => { + const res = new Response('The sandbox was not found', { status: 502 }) + try { + await rejectProxyUnavailableResponse(res) + assert.fail('expected a rejection') + } catch (err) { + assert.instanceOf(err, ConnectError) + assert.strictEqual((err as ConnectError).code, Code.Unavailable) + assert.strictEqual( + (err as ConnectError).rawMessage, + 'The sandbox was not found' + ) + assert.instanceOf(handleRpcError(err), SandboxNotRunningError) + } + }) + + test('falls back to the status for a 502 without a body', async () => { + const res = new Response(null, { status: 502 }) + try { + await rejectProxyUnavailableResponse(res) + assert.fail('expected a rejection') + } catch (err) { + assert.instanceOf(err, ConnectError) + assert.strictEqual((err as ConnectError).rawMessage, 'HTTP 502') + } + }) +}) diff --git a/packages/js-sdk/tests/retry.test.ts b/packages/js-sdk/tests/retry.test.ts index 7bd4f547ee..7f43e9f9e7 100644 --- a/packages/js-sdk/tests/retry.test.ts +++ b/packages/js-sdk/tests/retry.test.ts @@ -7,6 +7,7 @@ import { parseRetryAfter, resolveRetries, withRetry, + isNetworkError, } from '../src/retry' import { EnvdApiClient } from '../src/envd/api' import { InvalidArgumentError } from '../src/errors' @@ -487,6 +488,26 @@ const aborts = { timeout: new DOMException('timed out', 'TimeoutError'), } +describe('isNetworkError', () => { + test.each([ + ...Object.entries(connectFailures), + ['Chrome', opaque.Chrome], + ['Firefox', opaque.Firefox], + ['Safari', opaque.Safari], + ])('recognizes %s', (_, error) => { + expect(isNetworkError(error)).toBe(true) + }) + + test.each([ + ...Object.entries(terminated), + ...Object.entries(aborts), + ['other TypeError', new TypeError('Failed to parse URL')], + ['non-error', 'Failed to fetch'], + ])('rejects %s', (_, error) => { + expect(isNetworkError(error)).toBe(false) + }) +}) + describe('isConnectionError', () => { test.each(Object.entries(connectFailures))('recognizes %s', (_, error) => { expect(isConnectionError(error)).toBe(true) diff --git a/packages/python-sdk/e2b/__init__.py b/packages/python-sdk/e2b/__init__.py index f8371b895f..93c67b65bf 100644 --- a/packages/python-sdk/e2b/__init__.py +++ b/packages/python-sdk/e2b/__init__.py @@ -51,6 +51,8 @@ ServiceBusyException, SandboxException, SandboxNotFoundException, + SandboxNotRunningException, + SandboxUnreachableException, TemplateException, TimeoutException, VolumeException, @@ -184,6 +186,8 @@ "NotFoundException", "FileNotFoundException", "SandboxNotFoundException", + "SandboxNotRunningException", + "SandboxUnreachableException", "AuthenticationException", "GitAuthException", "GitUpstreamException", diff --git a/packages/python-sdk/e2b/api/client_async/__init__.py b/packages/python-sdk/e2b/api/client_async/__init__.py index af37aa5b81..e8449ec95b 100644 --- a/packages/python-sdk/e2b/api/client_async/__init__.py +++ b/packages/python-sdk/e2b/api/client_async/__init__.py @@ -153,8 +153,11 @@ async def execute(self, request: Request) -> Response: # they replaced, the transports are not bound to an event loop and the caches # are process-global rather than per-loop. _transports: Dict[_TransportKey, ConnectionRetryTransport] = {} +# The balancing transport under each retrying one, for callers that must not +# retry connecting (the health probe run after a request failed). +_pools: Dict[_TransportKey, BalancingTransport] = {} # The httpx adapter over each transport, shared by every httpx client on it. -_httpx_transports: Dict[_TransportKey, AsyncPyqwestTransport] = {} +_httpx_transports: Dict[Tuple[_TransportKey, bool], AsyncPyqwestTransport] = {} def get_pyqwest_transport( @@ -162,7 +165,8 @@ def get_pyqwest_transport( read_timeout: Optional[float] = None, *, http_version: HttpVersion = DEFAULT_HTTP_VERSION, -) -> ConnectionRetryTransport: + retry_connect: bool = True, +) -> Transport: """The shared pyqwest transport with the SDK's tuning — system CA certs (without which TLS through an intercepting proxy fails) and the httpx-equivalent pool limits — behind connect-only retries. It balances @@ -187,7 +191,10 @@ def get_pyqwest_transport( Requests are logged by pyqwest itself on the ``pyqwest.access`` and ``pyqwest`` loggers at ``DEBUG`` (off unless enabled) — the transport-level diagnostics httpcore used to provide. The SDK's own ``logger`` option is - separate and sits above this, on the httpx client.""" + separate and sits above this, on the httpx client. + + ``retry_connect=False`` returns the balancing transport under the retrying + one — the same pools, without the connection retries.""" key = (proxy, read_timeout, http_version) def build() -> HTTPTransport: @@ -209,11 +216,11 @@ def build() -> HTTPTransport: with _transport_lock: transport = _transports.get(key) if transport is None: - transport = ConnectionRetryTransport( - BalancingTransport(build), max_retries=connection_retries - ) + pool = BalancingTransport(build) + transport = ConnectionRetryTransport(pool, max_retries=connection_retries) + _pools[key] = pool _transports[key] = transport - return transport + return transport if retry_connect else _pools[key] def get_httpx_transport( @@ -221,15 +228,19 @@ def get_httpx_transport( read_timeout: Optional[float] = None, *, http_version: HttpVersion = DEFAULT_HTTP_VERSION, + retry_connect: bool = True, ) -> AsyncPyqwestTransport: """The httpx adapter over the shared transport of :func:`get_pyqwest_transport`, for the generated httpx clients (control plane, envd HTTP API, volume content). The adapter holds no state of its own and does not close the pools, so closing an httpx client leaves them - intact for the other clients on it.""" - key = (proxy, read_timeout, http_version) + intact for the other clients on it. ``retry_connect=False`` adapts the + same pools without the connection retries.""" + key = ((proxy, read_timeout, http_version), retry_connect) # Resolve the pool before taking the lock: it takes the same one. - pool = get_pyqwest_transport(proxy, read_timeout, http_version=http_version) + pool = get_pyqwest_transport( + proxy, read_timeout, http_version=http_version, retry_connect=retry_connect + ) with _transport_lock: transport = _httpx_transports.get(key) if transport is None: @@ -243,6 +254,7 @@ def get_transport( *, http_version: Optional[HttpVersion] = None, for_streaming: bool = False, + retry_connect: bool = True, ) -> AsyncPyqwestTransport: """The shared httpx transport factory for the control-plane REST API and envd HTTP API (file transfers, health checks). For TLS connections ALPN @@ -264,16 +276,26 @@ def get_transport( whole-request deadlines rather than idle bounds — so only streamed downloads take it, and they get their own pool (see :func:`get_pyqwest_transport`). + + ``retry_connect=False`` returns the adapter over the same pool without the + connection retries (``E2B_CONNECTION_RETRIES``): for the health probe run + after a request failed, so a sandbox that cannot be reached is not tried + all over again. """ return get_httpx_transport( proxy_to_config(config.proxy), READ_TIMEOUT if for_streaming else None, http_version=(config.http_version if http_version is None else http_version), + retry_connect=retry_connect, ) def get_envd_api( - config: ConnectionConfig, base_url: str, *, for_streaming: bool = False + config: ConnectionConfig, + base_url: str, + *, + for_streaming: bool = False, + retry_connect: bool = True, ) -> httpx.AsyncClient: """An httpx client for a sandbox's envd HTTP API (file transfers, health checks) on the shared transports. The client itself is a cheap stateless @@ -281,7 +303,9 @@ def get_envd_api( is shared and loop-independent.""" return httpx.AsyncClient( base_url=base_url, - transport=get_transport(config, for_streaming=for_streaming), + transport=get_transport( + config, for_streaming=for_streaming, retry_connect=retry_connect + ), headers=config.sandbox_headers, event_hooks=make_async_logging_event_hooks(config.logger), ) diff --git a/packages/python-sdk/e2b/api/client_sync/__init__.py b/packages/python-sdk/e2b/api/client_sync/__init__.py index 720cfdbbda..473625e110 100644 --- a/packages/python-sdk/e2b/api/client_sync/__init__.py +++ b/packages/python-sdk/e2b/api/client_sync/__init__.py @@ -158,8 +158,11 @@ def execute_sync(self, request: SyncRequest) -> SyncResponse: # pyqwest transports are thread-safe, so unlike the httpx transports they # replaced, the caches are process-global rather than per-thread. _transports: Dict[_TransportKey, ConnectionRetryTransport] = {} +# The balancing transport under each retrying one, for callers that must not +# retry connecting (the health probe run after a request failed). +_pools: Dict[_TransportKey, BalancingTransport] = {} # The httpx adapter over each transport, shared by every httpx client on it. -_httpx_transports: Dict[_TransportKey, PyqwestTransport] = {} +_httpx_transports: Dict[Tuple[_TransportKey, bool], PyqwestTransport] = {} def get_pyqwest_transport( @@ -167,7 +170,8 @@ def get_pyqwest_transport( read_timeout: Optional[float] = None, *, http_version: HttpVersion = DEFAULT_HTTP_VERSION, -) -> ConnectionRetryTransport: + retry_connect: bool = True, +) -> SyncTransport: """The shared pyqwest transport with the SDK's tuning — system CA certs (without which TLS through an intercepting proxy fails) and the httpx-equivalent pool limits — behind connect-only retries. It balances @@ -192,7 +196,10 @@ def get_pyqwest_transport( Requests are logged by pyqwest itself on the ``pyqwest.access`` and ``pyqwest`` loggers at ``DEBUG`` (off unless enabled) — the transport-level diagnostics httpcore used to provide. The SDK's own ``logger`` option is - separate and sits above this, on the httpx client.""" + separate and sits above this, on the httpx client. + + ``retry_connect=False`` returns the balancing transport under the retrying + one — the same pools, without the connection retries.""" key = (proxy, read_timeout, http_version) def build() -> SyncHTTPTransport: @@ -214,11 +221,11 @@ def build() -> SyncHTTPTransport: with _transport_lock: transport = _transports.get(key) if transport is None: - transport = ConnectionRetryTransport( - BalancingTransport(build), max_retries=connection_retries - ) + pool = BalancingTransport(build) + transport = ConnectionRetryTransport(pool, max_retries=connection_retries) + _pools[key] = pool _transports[key] = transport - return transport + return transport if retry_connect else _pools[key] def get_httpx_transport( @@ -226,15 +233,19 @@ def get_httpx_transport( read_timeout: Optional[float] = None, *, http_version: HttpVersion = DEFAULT_HTTP_VERSION, + retry_connect: bool = True, ) -> PyqwestTransport: """The httpx adapter over the shared transport of :func:`get_pyqwest_transport`, for the generated httpx clients (control plane, envd HTTP API, volume content). The adapter holds no state of its own and does not close the pools, so closing an httpx client leaves them - intact for the other clients on it.""" - key = (proxy, read_timeout, http_version) + intact for the other clients on it. ``retry_connect=False`` adapts the + same pools without the connection retries.""" + key = ((proxy, read_timeout, http_version), retry_connect) # Resolve the pool before taking the lock: it takes the same one. - pool = get_pyqwest_transport(proxy, read_timeout, http_version=http_version) + pool = get_pyqwest_transport( + proxy, read_timeout, http_version=http_version, retry_connect=retry_connect + ) with _transport_lock: transport = _httpx_transports.get(key) if transport is None: @@ -248,6 +259,7 @@ def get_transport( *, http_version: Optional[HttpVersion] = None, for_streaming: bool = False, + retry_connect: bool = True, ) -> PyqwestTransport: """The shared httpx transport factory for the control-plane REST API and envd HTTP API (file transfers, health checks). For TLS connections ALPN @@ -269,16 +281,26 @@ def get_transport( whole-request deadlines rather than idle bounds — so only streamed downloads take it, and they get their own pool (see :func:`get_pyqwest_transport`). + + ``retry_connect=False`` returns the adapter over the same pool without the + connection retries (``E2B_CONNECTION_RETRIES``): for the health probe run + after a request failed, so a sandbox that cannot be reached is not tried + all over again. """ return get_httpx_transport( proxy_to_config(config.proxy), READ_TIMEOUT if for_streaming else None, http_version=(config.http_version if http_version is None else http_version), + retry_connect=retry_connect, ) def get_envd_api( - config: ConnectionConfig, base_url: str, *, for_streaming: bool = False + config: ConnectionConfig, + base_url: str, + *, + for_streaming: bool = False, + retry_connect: bool = True, ) -> httpx.Client: """An httpx client for a sandbox's envd HTTP API (file transfers, health checks) on the shared transports. The client itself is a cheap stateless @@ -286,7 +308,9 @@ def get_envd_api( is shared and thread-safe.""" return httpx.Client( base_url=base_url, - transport=get_transport(config, for_streaming=for_streaming), + transport=get_transport( + config, for_streaming=for_streaming, retry_connect=retry_connect + ), headers=config.sandbox_headers, event_hooks=make_logging_event_hooks(config.logger), ) diff --git a/packages/python-sdk/e2b/envd/api.py b/packages/python-sdk/e2b/envd/api.py index 811d58d592..21a0ecb7ec 100644 --- a/packages/python-sdk/e2b/envd/api.py +++ b/packages/python-sdk/e2b/envd/api.py @@ -1,9 +1,11 @@ import httpx -import json from typing import Callable, Optional -from e2b.envd.rpc import format_terminated_exception +from e2b.envd.rpc import ( + format_sandbox_unreachable_exception, + format_terminated_exception, +) from e2b.exceptions import ( SandboxException, NotFoundException, @@ -11,7 +13,8 @@ InvalidArgumentException, NotEnoughSpaceException, RateLimitException, - format_sandbox_timeout_exception, + SandboxNotRunningException, + format_sandbox_unavailable_exception, ) @@ -25,41 +28,66 @@ 429: lambda message: RateLimitException( f"{message}: The requests are being rate limited." ), - 502: format_sandbox_timeout_exception, + 502: format_sandbox_unavailable_exception, 507: NotEnoughSpaceException, } HEALTH_CHECK_TIMEOUT = 5 # seconds +# Raised by the pyqwest httpx adapter when a request fails at the connection +# level: the connection could not be established (``ConnectError`` / +# ``ConnectTimeout``, after the connection layer's own retries) or was dropped +# mid-request (``ReadError`` / ``WriteError`` / ``RemoteProtocolError``, e.g. an +# HTTP/2 stream reset). Read timeouts are excluded: the request reached the +# sandbox and exhausted its budget, which is a definitive result. +ENVD_API_TRANSPORT_ERRORS = ( + httpx.ConnectError, + httpx.ConnectTimeout, + httpx.ReadError, + httpx.WriteError, + httpx.RemoteProtocolError, +) + def check_sandbox_health(envd_api: httpx.Client) -> Optional[bool]: """Probe the sandbox's envd health endpoint. - :return: ``True`` if the sandbox is running, ``False`` if it is not, ``None`` if its state could not be determined. + :param envd_api: The sandbox's envd HTTP client -- the one built with ``retry_connect=False`` when probing after a failed request, so a sandbox that cannot be reached is not tried all over again. + + :return: ``True`` if the sandbox is running, ``False`` if it is not, ``None`` if it answered but its state could not be determined. + :raises httpx.TransportError: When the sandbox could not be reached (no answer within the probe timeout). + :raises SandboxUnreachableException: When the proxy reports the sandbox running but envd's port not open -- the exception a request would get. """ try: r = envd_api.get(ENVD_API_HEALTH_ROUTE, timeout=HEALTH_CHECK_TIMEOUT) - if r.status_code == 502: - return False - if r.is_success: - return True - return None + except httpx.TransportError: + raise except Exception: return None + return _health_status(r) async def acheck_sandbox_health(envd_api: httpx.AsyncClient) -> Optional[bool]: """Async version of :func:`check_sandbox_health`.""" try: r = await envd_api.get(ENVD_API_HEALTH_ROUTE, timeout=HEALTH_CHECK_TIMEOUT) - if r.status_code == 502: - return False - if r.is_success: - return True - return None + except httpx.TransportError: + raise except Exception: return None + return _health_status(r) + + +def _health_status(r: httpx.Response) -> Optional[bool]: + if r.status_code == 502: + err = format_sandbox_unavailable_exception(get_message(r)) + if isinstance(err, SandboxNotRunningException): + return False + raise err + if r.is_success: + return True + return None def handle_envd_api_transport_exception( @@ -69,12 +97,10 @@ def handle_envd_api_transport_exception( """Handle transport-level errors from envd API requests. :param e: The caught exception, expected to be a transport-level ``httpx`` error. - :param sandbox_running: Result of a sandbox health probe (``None`` when unknown), used to disambiguate a connection dropped mid-request. - :return: A ``TimeoutException`` when the connection dropped mid-request and the sandbox is confirmed gone, or the original exception unchanged otherwise. + :param sandbox_running: Result of a sandbox health probe (``None`` when unknown), used to disambiguate a request that failed at the connection level. + :return: A ``SandboxNotRunningException`` when the request failed at the connection level and the sandbox is confirmed gone, or the original exception unchanged otherwise. """ - # A remote protocol error (e.g. an HTTP/2 stream reset) means the connection to the - # sandbox was dropped mid-request — either the sandbox died or the network failed - if isinstance(e, httpx.RemoteProtocolError): + if isinstance(e, ENVD_API_TRANSPORT_ERRORS): return format_terminated_exception(e, sandbox_running) return e @@ -84,15 +110,19 @@ def handle_envd_api_transport_exception_with_health( e: Exception, envd_api: httpx.Client, ) -> Exception: - """Like :func:`handle_envd_api_transport_exception`, but when the connection to the - sandbox was dropped mid-request it probes the sandbox health to tell apart the sandbox - being killed from a transient network failure (e.g. a load balancer dropping the connection). + """Like :func:`handle_envd_api_transport_exception`, but when the request failed at + the connection level (the connection to the sandbox could not be established or was + dropped mid-request) it probes the sandbox health to tell apart the sandbox being + killed from the sandbox being unreachable or a transient network failure (e.g. a + load balancer dropping the connection). When the probe gets no answer either, a + ``SandboxUnreachableException`` is returned. """ - sandbox_running = ( - check_sandbox_health(envd_api) - if isinstance(e, httpx.RemoteProtocolError) - else None - ) + if not isinstance(e, ENVD_API_TRANSPORT_ERRORS): + return e + try: + sandbox_running = check_sandbox_health(envd_api) + except Exception as probe_error: + return format_sandbox_unreachable_exception(e, probe_error) return handle_envd_api_transport_exception(e, sandbox_running) @@ -101,21 +131,25 @@ async def ahandle_envd_api_transport_exception_with_health( envd_api: httpx.AsyncClient, ) -> Exception: """Async version of :func:`handle_envd_api_transport_exception_with_health`.""" - sandbox_running = ( - await acheck_sandbox_health(envd_api) - if isinstance(e, httpx.RemoteProtocolError) - else None - ) + if not isinstance(e, ENVD_API_TRANSPORT_ERRORS): + return e + try: + sandbox_running = await acheck_sandbox_health(envd_api) + except Exception as probe_error: + return format_sandbox_unreachable_exception(e, probe_error) return handle_envd_api_transport_exception(e, sandbox_running) def get_message(e: httpx.Response) -> str: + """The error message of a response: the string ``message`` of its JSON object + body, else its text.""" try: - message = e.json().get("message", e.text) - except json.JSONDecodeError: - message = e.text - - return message + body = e.json() + except ValueError: + return e.text + if isinstance(body, dict) and isinstance(body.get("message"), str): + return body["message"] + return e.text def handle_envd_api_exception( diff --git a/packages/python-sdk/e2b/envd/rpc.py b/packages/python-sdk/e2b/envd/rpc.py index 7caca5c70d..bb9d7dfa5c 100644 --- a/packages/python-sdk/e2b/envd/rpc.py +++ b/packages/python-sdk/e2b/envd/rpc.py @@ -9,8 +9,12 @@ SandboxException, InvalidArgumentException, NotFoundException, + SandboxNotRunningException, + SandboxUnreachableException, TimeoutException, - format_sandbox_timeout_exception, + format_sandbox_unavailable_exception, + is_sandbox_not_found_message, + is_sandbox_port_not_open_message, AuthenticationException, RateLimitException, ) @@ -19,7 +23,7 @@ Code.INVALID_ARGUMENT: InvalidArgumentException, Code.UNAUTHENTICATED: AuthenticationException, Code.NOT_FOUND: NotFoundException, - Code.UNAVAILABLE: format_sandbox_timeout_exception, + Code.UNAVAILABLE: format_sandbox_unavailable_exception, Code.RESOURCE_EXHAUSTED: lambda message: RateLimitException( f"{message}: Rate limit exceeded, please try again later." ), @@ -59,20 +63,55 @@ def is_transport_failure(e: Exception) -> bool: ) +def is_ambiguous_unavailable(e: Exception) -> bool: + """Whether the error is an ``UNAVAILABLE`` whose message the proxy's answers do + not explain -- neither the sandbox not found nor its port not open -- e.g. envd + ending a stream with "the connection to sandbox ... ended before the stream + completed" when the sandbox is killed mid-command. Whether the sandbox is gone + can only be told by probing it.""" + return ( + isinstance(e, ConnectError) + and e.code is Code.UNAVAILABLE + and not is_sandbox_not_found_message(e.message) + and not is_sandbox_port_not_open_message(e.message) + ) + + def format_terminated_exception( e: Exception, sandbox_running: Optional[bool], ) -> Exception: - """Handle an exception for a connection to the sandbox dropped mid-request: when a + """Handle an exception for a request that failed at the connection level: when a sandbox health probe confirmed the sandbox is gone (``sandbox_running is False``), - return a ``TimeoutException``; otherwise return the original error unchanged.""" + return a ``SandboxNotRunningException``; otherwise return the original error unchanged.""" if sandbox_running is False: - return TimeoutException( + err = SandboxNotRunningException( f"{e}: The sandbox was killed or reached its end of life while the request was in flight." ) + err.__cause__ = e + return err return e +def format_sandbox_unreachable_exception( + e: Exception, probe_error: Exception +) -> Exception: + """Build the exception for a request that failed at the connection level when the + follow-up sandbox health probe failed too. A probe answered by the proxy with the + sandbox running but envd's port not open is already a + ``SandboxUnreachableException`` and is kept, with the failed request as its + cause; otherwise the probe's failure is reported alongside the request's, which + becomes the cause.""" + if isinstance(probe_error, SandboxUnreachableException): + probe_error.__cause__ = e + return probe_error + err = SandboxUnreachableException( + f"{e}: The sandbox could not be reached and its health probe failed too ({probe_error}). It was not confirmed to be stopped — this is likely a network issue or envd inside the sandbox not being up yet; check the sandbox state with 'Sandbox.get_info()'." + ) + err.__cause__ = e + return err + + def handle_rpc_exception( e: Exception, error_map: Optional[dict[Code, Callable[[str], Exception]]] = None, @@ -82,8 +121,8 @@ def handle_rpc_exception( :param e: The caught exception, expected to be a ``ConnectError``. :param error_map: Optional map of gRPC codes to exception factories that override the defaults. - :param sandbox_running: Result of a sandbox health probe (``None`` when unknown), used to disambiguate a connection dropped mid-request. - :return: The corresponding exception. A connection dropped mid-request with the sandbox confirmed gone becomes a ``TimeoutException``; non-``ConnectError`` errors are otherwise returned as-is. + :param sandbox_running: Result of a sandbox health probe (``None`` when unknown), used to disambiguate a connection dropped mid-request or an ambiguous ``UNAVAILABLE``. + :return: The corresponding exception. A connection dropped mid-request, or an ambiguous ``UNAVAILABLE``, with the sandbox confirmed gone becomes a ``SandboxNotRunningException``; non-``ConnectError`` errors are otherwise returned as-is. """ if isinstance(e, ConnectError): # connectrpc converts asyncio cancellation into a ConnectError with @@ -97,7 +136,9 @@ def handle_rpc_exception( # connection to the sandbox was dropped mid-request — either the # sandbox died or the network failed — so the code mapping below, # which describes envd responses, doesn't apply. - if is_transport_failure(e): + if is_transport_failure(e) or ( + sandbox_running is False and is_ambiguous_unavailable(e) + ): return format_terminated_exception(e, sandbox_running) # Everything else maps by code; classifiable client-side failures @@ -120,16 +161,26 @@ def handle_rpc_exception_with_health( check_health: Optional[Callable[[], Optional[bool]]] = None, error_map: Optional[dict[Code, Callable[[str], Exception]]] = None, ): - """Like :func:`handle_rpc_exception`, but when the connection to the sandbox was - dropped mid-request it probes the sandbox health to tell apart the sandbox being - killed from a transient network failure (e.g. a load balancer dropping the connection). + """Like :func:`handle_rpc_exception`, but when the request failed at the connection + level (the connection to the sandbox could not be established or was dropped + mid-request), or envd/the proxy answered an ``UNAVAILABLE`` that does not say + whether the sandbox is gone (:func:`is_ambiguous_unavailable`), it probes the + sandbox health to tell apart the sandbox being killed from the sandbox being + unreachable or a transient network failure (e.g. a load balancer dropping the + connection). When the probe confirms the sandbox is gone, a + ``SandboxNotRunningException`` is returned; when the probe gets no answer either + (the health check raises), a ``SandboxUnreachableException``; otherwise the error + maps as usual (an ambiguous ``UNAVAILABLE`` to a ``SandboxUnreachableException`` + with the sandbox state unknown). """ sandbox_running = None - if check_health is not None and is_transport_failure(e): + if check_health is not None and ( + is_transport_failure(e) or is_ambiguous_unavailable(e) + ): try: sandbox_running = check_health() - except Exception: - sandbox_running = None + except Exception as probe_error: + return format_sandbox_unreachable_exception(e, probe_error) return handle_rpc_exception(e, error_map, sandbox_running) @@ -140,9 +191,11 @@ async def ahandle_rpc_exception_with_health( ): """Async version of :func:`handle_rpc_exception_with_health`.""" sandbox_running = None - if check_health is not None and is_transport_failure(e): + if check_health is not None and ( + is_transport_failure(e) or is_ambiguous_unavailable(e) + ): try: sandbox_running = await check_health() - except Exception: - sandbox_running = None + except Exception as probe_error: + return format_sandbox_unreachable_exception(e, probe_error) return handle_rpc_exception(e, error_map, sandbox_running) diff --git a/packages/python-sdk/e2b/exceptions.py b/packages/python-sdk/e2b/exceptions.py index 4cc3abb62e..d479e21313 100644 --- a/packages/python-sdk/e2b/exceptions.py +++ b/packages/python-sdk/e2b/exceptions.py @@ -1,9 +1,38 @@ from typing import Optional -def format_sandbox_timeout_exception(message: str): - return TimeoutException( - f"{message}: This error is likely due to sandbox timeout. You can modify the sandbox timeout by passing 'timeout' when starting the sandbox or calling '.set_timeout' on the sandbox with the desired timeout." +# Messages of the 502 the proxy in front of envd answers with when the sandbox does +# not exist anymore (killed or reached its timeout) or when it is running but envd's +# port is not open (e.g. its network is down). +SANDBOX_NOT_FOUND_MESSAGE = "was not found" +SANDBOX_PORT_NOT_OPEN_MESSAGE = "port is not open" + + +def is_sandbox_not_found_message(message: str) -> bool: + return SANDBOX_NOT_FOUND_MESSAGE in message + + +def is_sandbox_port_not_open_message(message: str) -> bool: + return SANDBOX_PORT_NOT_OPEN_MESSAGE in message + + +def format_sandbox_unavailable_exception(message: str) -> Exception: + """Map a 502/UNAVAILABLE answered by the proxy in front of envd: the sandbox was + not found (killed or reached its timeout) is a ``SandboxNotRunningException``; + the sandbox running but envd's port not open (e.g. its network is down) a + ``SandboxUnreachableException``; any other message (e.g. envd ending a stream + because the connection to the sandbox was lost) a ``SandboxUnreachableException`` + too, with the sandbox state reported as unknown.""" + if is_sandbox_not_found_message(message): + return SandboxNotRunningException( + f"{message}: The sandbox was killed or reached its timeout. You can modify the sandbox timeout by passing 'timeout' when starting the sandbox or calling '.set_timeout' on the sandbox with the desired timeout." + ) + if is_sandbox_port_not_open_message(message): + return SandboxUnreachableException( + f"{message}: The sandbox is running but envd inside it could not be reached (e.g. its network is down). Check the sandbox state with 'Sandbox.get_info()'." + ) + return SandboxUnreachableException( + f"{message}: The sandbox could not be reached and its state is unknown — it may have been killed or reached its timeout, or the connection to it was lost. Check the sandbox state with 'Sandbox.get_info()'." ) @@ -88,6 +117,49 @@ class SandboxNotFoundException(NotFoundException): pass +class SandboxNotRunningException(TimeoutException): + """ + Raised when the sandbox is not running anymore: the proxy in front of it + answered a request (or the health probe run after a request failed at the + connection level) with the sandbox not found -- it was killed, paused or + reached its timeout. Retrying the request will not help; create or resume a + sandbox. + + The split from ``SandboxUnreachableException`` is only as good as the proxy's + answer and the health probe: a sandbox that is gone but could not be asked + about is a ``SandboxUnreachableException``. Handlers that want every "sandbox + is gone" case should catch both this class and ``SandboxNotFoundException`` + (the control-plane API's answer for a sandbox that does not exist). + + Subclass of ``TimeoutException``, which this case surfaced as before. + """ + + pass + + +class SandboxUnreachableException(TimeoutException): + """ + Raised when the sandbox could not be reached while it is not confirmed to be + stopped: the proxy in front of the sandbox reports it running but envd's port + not open (e.g. the sandbox's network is down), the proxy or envd answered with + an unavailability the health probe could not resolve (e.g. the connection to + the sandbox was lost mid-stream while the probe found it running), or a + request failed at the connection level (the connection could not be + established or was dropped mid-request) and the follow-up health probe got no + answer from the sandbox either. A sandbox confirmed stopped (or paused) is a + ``SandboxNotRunningException`` -- but only as far as the probe can tell: a + sandbox that is gone and could not be asked about surfaces here. + + This usually means the sandbox's network is down, envd inside it is not up + (yet), or a network issue between the client and the sandbox. Check the + sandbox state with ``Sandbox.get_info()``. + + Subclass of ``TimeoutException``, which these cases surfaced as before. + """ + + pass + + class AuthenticationException(Exception): """ Raised when authentication fails. diff --git a/packages/python-sdk/e2b/sandbox_async/commands/command.py b/packages/python-sdk/e2b/sandbox_async/commands/command.py index f2ad186066..514add11af 100644 --- a/packages/python-sdk/e2b/sandbox_async/commands/command.py +++ b/packages/python-sdk/e2b/sandbox_async/commands/command.py @@ -1,6 +1,8 @@ from typing import Any, AsyncGenerator, Dict, List, Literal, Optional, Union, overload import httpx + +from e2b.api.client_async import get_envd_api from connectrpc.code import Code from connectrpc.errors import ConnectError from packaging.version import Version @@ -49,9 +51,15 @@ def __init__( connection_config, ) self._envd_api = envd_api + # The health probe run after a request failed goes without the + # connection retries, so a sandbox that cannot be reached is not tried + # all over again (see `get_transport`). + self._envd_health_api = get_envd_api( + connection_config, envd_api_url, retry_connect=False + ) async def _check_health(self) -> Optional[bool]: - return await acheck_sandbox_health(self._envd_api) + return await acheck_sandbox_health(self._envd_health_api) async def list( self, diff --git a/packages/python-sdk/e2b/sandbox_async/commands/pty.py b/packages/python-sdk/e2b/sandbox_async/commands/pty.py index dda9d2e6f3..1047a81140 100644 --- a/packages/python-sdk/e2b/sandbox_async/commands/pty.py +++ b/packages/python-sdk/e2b/sandbox_async/commands/pty.py @@ -2,6 +2,8 @@ import httpx +from e2b.api.client_async import get_envd_api + from connectrpc.code import Code from connectrpc.errors import ConnectError from packaging.version import Version @@ -50,9 +52,15 @@ def __init__( connection_config, ) self._envd_api = envd_api + # The health probe run after a request failed goes without the + # connection retries, so a sandbox that cannot be reached is not tried + # all over again (see `get_transport`). + self._envd_health_api = get_envd_api( + connection_config, envd_api_url, retry_connect=False + ) async def _check_health(self) -> Optional[bool]: - return await acheck_sandbox_health(self._envd_api) + return await acheck_sandbox_health(self._envd_health_api) async def kill( self, diff --git a/packages/python-sdk/e2b/sandbox_async/filesystem/filesystem.py b/packages/python-sdk/e2b/sandbox_async/filesystem/filesystem.py index 76ca51e329..e59612f926 100644 --- a/packages/python-sdk/e2b/sandbox_async/filesystem/filesystem.py +++ b/packages/python-sdk/e2b/sandbox_async/filesystem/filesystem.py @@ -68,7 +68,8 @@ async def _ahandle_filesystem_rpc_exception( - e: Exception, envd_api: httpx.AsyncClient + e: Exception, + envd_api: httpx.AsyncClient, ) -> Exception: return await ahandle_rpc_exception_with_health( e, lambda: acheck_sandbox_health(envd_api), _FILESYSTEM_RPC_ERROR_MAP @@ -104,6 +105,12 @@ def __init__( self._envd_api_streaming = get_envd_api( connection_config, envd_api_url, for_streaming=True ) + # The health probe run after a request failed goes without the + # connection retries, so a sandbox that cannot be reached is not tried + # all over again. + self._envd_health_api = get_envd_api( + connection_config, envd_api_url, retry_connect=False + ) @overload async def read( @@ -244,9 +251,9 @@ async def read( ) else: r = await client.send(request, stream=True) - except httpx.RemoteProtocolError as e: + except httpx.TransportError as e: raise await ahandle_envd_api_transport_exception_with_health( - e, self._envd_api + e, self._envd_health_api ) except asyncio.TimeoutError as e: # wait_for's expiry; keep the httpx exception the @@ -267,9 +274,9 @@ async def read( headers=headers, timeout=timeout, ) - except httpx.RemoteProtocolError as e: + except httpx.TransportError as e: raise await ahandle_envd_api_transport_exception_with_health( - e, self._envd_api + e, self._envd_health_api ) err = await _ahandle_filesystem_envd_api_exception(r) @@ -412,9 +419,9 @@ async def _upload_file(file): params=params, timeout=None if is_streamed else upload_timeout, ) - except httpx.RemoteProtocolError as e: + except httpx.TransportError as e: raise await ahandle_envd_api_transport_exception_with_health( - e, self._envd_api + e, self._envd_health_api ) err = await _ahandle_filesystem_envd_api_exception(r) @@ -461,9 +468,9 @@ async def _upload_file(file): None if multipart_body_is_streamed(files) else upload_timeout ), ) - except httpx.RemoteProtocolError as e: + except httpx.TransportError as e: raise await ahandle_envd_api_transport_exception_with_health( - e, self._envd_api + e, self._envd_health_api ) err = await _ahandle_filesystem_envd_api_exception(r) @@ -518,7 +525,7 @@ async def list( return entries except Exception as e: - raise await _ahandle_filesystem_rpc_exception(e, self._envd_api) + raise await _ahandle_filesystem_rpc_exception(e, self._envd_health_api) async def exists( self, @@ -550,7 +557,7 @@ async def exists( if isinstance(e, ConnectError): if e.code == Code.NOT_FOUND: return False - raise await _ahandle_filesystem_rpc_exception(e, self._envd_api) + raise await _ahandle_filesystem_rpc_exception(e, self._envd_health_api) async def get_info( self, @@ -578,7 +585,7 @@ async def get_info( return map_entry_info(r.entry or filesystem_pb.EntryInfo()) except Exception as e: - raise await _ahandle_filesystem_rpc_exception(e, self._envd_api) + raise await _ahandle_filesystem_rpc_exception(e, self._envd_health_api) async def remove( self, @@ -602,7 +609,7 @@ async def remove( headers=authentication_header(self._envd_version, user), ) except Exception as e: - raise await _ahandle_filesystem_rpc_exception(e, self._envd_api) + raise await _ahandle_filesystem_rpc_exception(e, self._envd_health_api) async def rename( self, @@ -635,7 +642,7 @@ async def rename( return map_entry_info(r.entry or filesystem_pb.EntryInfo()) except Exception as e: - raise await _ahandle_filesystem_rpc_exception(e, self._envd_api) + raise await _ahandle_filesystem_rpc_exception(e, self._envd_health_api) async def make_dir( self, @@ -666,7 +673,7 @@ async def make_dir( if isinstance(e, ConnectError): if e.code == Code.ALREADY_EXISTS: return False - raise await _ahandle_filesystem_rpc_exception(e, self._envd_api) + raise await _ahandle_filesystem_rpc_exception(e, self._envd_health_api) async def watch_dir( self, @@ -749,11 +756,11 @@ async def watch_dir( events=events, on_event=on_event, on_exit=on_exit, - check_health=lambda: acheck_sandbox_health(self._envd_api), + check_health=lambda: acheck_sandbox_health(self._envd_health_api), ) except Exception as e: try: await events.aclose() except Exception: pass - raise await _ahandle_filesystem_rpc_exception(e, self._envd_api) + raise await _ahandle_filesystem_rpc_exception(e, self._envd_health_api) diff --git a/packages/python-sdk/e2b/sandbox_sync/commands/command.py b/packages/python-sdk/e2b/sandbox_sync/commands/command.py index d929301e4e..a46a84dc98 100644 --- a/packages/python-sdk/e2b/sandbox_sync/commands/command.py +++ b/packages/python-sdk/e2b/sandbox_sync/commands/command.py @@ -11,6 +11,8 @@ ) import httpx + +from e2b.api.client_sync import get_envd_api from connectrpc.code import Code from connectrpc.errors import ConnectError from packaging.version import Version @@ -58,9 +60,15 @@ def __init__( connection_config, ) self._envd_api = envd_api + # The health probe run after a request failed goes without the + # connection retries, so a sandbox that cannot be reached is not tried + # all over again (see `get_transport`). + self._envd_health_api = get_envd_api( + connection_config, envd_api_url, retry_connect=False + ) def _check_health(self) -> Optional[bool]: - return check_sandbox_health(self._envd_api) + return check_sandbox_health(self._envd_health_api) def list( self, diff --git a/packages/python-sdk/e2b/sandbox_sync/commands/pty.py b/packages/python-sdk/e2b/sandbox_sync/commands/pty.py index 9465007759..5ea770d85c 100644 --- a/packages/python-sdk/e2b/sandbox_sync/commands/pty.py +++ b/packages/python-sdk/e2b/sandbox_sync/commands/pty.py @@ -2,6 +2,8 @@ import httpx +from e2b.api.client_sync import get_envd_api + from connectrpc.code import Code from connectrpc.errors import ConnectError from packaging.version import Version @@ -46,9 +48,15 @@ def __init__( connection_config, ) self._envd_api = envd_api + # The health probe run after a request failed goes without the + # connection retries, so a sandbox that cannot be reached is not tried + # all over again (see `get_transport`). + self._envd_health_api = get_envd_api( + connection_config, envd_api_url, retry_connect=False + ) def _check_health(self) -> Optional[bool]: - return check_sandbox_health(self._envd_api) + return check_sandbox_health(self._envd_health_api) def kill( self, diff --git a/packages/python-sdk/e2b/sandbox_sync/filesystem/filesystem.py b/packages/python-sdk/e2b/sandbox_sync/filesystem/filesystem.py index 752d4999a7..a61936d58b 100644 --- a/packages/python-sdk/e2b/sandbox_sync/filesystem/filesystem.py +++ b/packages/python-sdk/e2b/sandbox_sync/filesystem/filesystem.py @@ -63,7 +63,10 @@ } -def _handle_filesystem_rpc_exception(e: Exception, envd_api: httpx.Client) -> Exception: +def _handle_filesystem_rpc_exception( + e: Exception, + envd_api: httpx.Client, +) -> Exception: return handle_rpc_exception_with_health( e, lambda: check_sandbox_health(envd_api), _FILESYSTEM_RPC_ERROR_MAP ) @@ -100,6 +103,12 @@ def __init__( self._envd_api_streaming = get_envd_api( connection_config, envd_api_url, for_streaming=True ) + # The health probe run after a request failed goes without the + # connection retries, so a sandbox that cannot be reached is not tried + # all over again. + self._envd_health_api = get_envd_api( + connection_config, envd_api_url, retry_connect=False + ) @overload def read( @@ -223,8 +232,10 @@ def read( ) try: r = self._envd_api_streaming.send(request, stream=True) - except httpx.RemoteProtocolError as e: - raise handle_envd_api_transport_exception_with_health(e, self._envd_api) + except httpx.TransportError as e: + raise handle_envd_api_transport_exception_with_health( + e, self._envd_health_api + ) err = _handle_filesystem_envd_api_exception(r) if err: @@ -240,8 +251,10 @@ def read( headers=headers, timeout=timeout, ) - except httpx.RemoteProtocolError as e: - raise handle_envd_api_transport_exception_with_health(e, self._envd_api) + except httpx.TransportError as e: + raise handle_envd_api_transport_exception_with_health( + e, self._envd_health_api + ) err = _handle_filesystem_envd_api_exception(r) if err: @@ -382,9 +395,9 @@ def write_files( params=params, timeout=None if is_streamed else upload_timeout, ) - except httpx.RemoteProtocolError as e: + except httpx.TransportError as e: raise handle_envd_api_transport_exception_with_health( - e, self._envd_api + e, self._envd_health_api ) err = _handle_filesystem_envd_api_exception(r) @@ -425,8 +438,10 @@ def write_files( None if multipart_body_is_streamed(files) else upload_timeout ), ) - except httpx.RemoteProtocolError as e: - raise handle_envd_api_transport_exception_with_health(e, self._envd_api) + except httpx.TransportError as e: + raise handle_envd_api_transport_exception_with_health( + e, self._envd_health_api + ) err = _handle_filesystem_envd_api_exception(r) if err: @@ -480,7 +495,7 @@ def list( return entries except Exception as e: - raise _handle_filesystem_rpc_exception(e, self._envd_api) + raise _handle_filesystem_rpc_exception(e, self._envd_health_api) def exists( self, @@ -511,7 +526,7 @@ def exists( if isinstance(e, ConnectError): if e.code == Code.NOT_FOUND: return False - raise _handle_filesystem_rpc_exception(e, self._envd_api) + raise _handle_filesystem_rpc_exception(e, self._envd_health_api) def get_info( self, @@ -539,7 +554,7 @@ def get_info( return map_entry_info(r.entry or filesystem_pb.EntryInfo()) except Exception as e: - raise _handle_filesystem_rpc_exception(e, self._envd_api) + raise _handle_filesystem_rpc_exception(e, self._envd_health_api) def remove( self, @@ -563,7 +578,7 @@ def remove( headers=authentication_header(self._envd_version, user), ) except Exception as e: - raise _handle_filesystem_rpc_exception(e, self._envd_api) + raise _handle_filesystem_rpc_exception(e, self._envd_health_api) def rename( self, @@ -596,7 +611,7 @@ def rename( return map_entry_info(r.entry or filesystem_pb.EntryInfo()) except Exception as e: - raise _handle_filesystem_rpc_exception(e, self._envd_api) + raise _handle_filesystem_rpc_exception(e, self._envd_health_api) def make_dir( self, @@ -627,7 +642,7 @@ def make_dir( if isinstance(e, ConnectError): if e.code == Code.ALREADY_EXISTS: return False - raise _handle_filesystem_rpc_exception(e, self._envd_api) + raise _handle_filesystem_rpc_exception(e, self._envd_health_api) def watch_dir( self, @@ -685,7 +700,7 @@ def watch_dir( }, ) except Exception as e: - raise _handle_filesystem_rpc_exception(e, self._envd_api) + raise _handle_filesystem_rpc_exception(e, self._envd_health_api) return WatchHandle( self._rpc, @@ -693,5 +708,5 @@ def watch_dir( self._connection_config, self._envd_version, user, - lambda: check_sandbox_health(self._envd_api), + lambda: check_sandbox_health(self._envd_health_api), ) diff --git a/packages/python-sdk/tests/test_envd_api_exception.py b/packages/python-sdk/tests/test_envd_api_exception.py index 7c0694da7c..59da690806 100644 --- a/packages/python-sdk/tests/test_envd_api_exception.py +++ b/packages/python-sdk/tests/test_envd_api_exception.py @@ -1,8 +1,12 @@ import httpx from e2b.envd.api import ( + acheck_sandbox_health, + ahandle_envd_api_transport_exception_with_health, + check_sandbox_health, format_envd_api_exception, handle_envd_api_transport_exception, + handle_envd_api_transport_exception_with_health, ) from e2b.exceptions import ( AuthenticationException, @@ -11,6 +15,8 @@ NotFoundException, RateLimitException, SandboxException, + SandboxNotRunningException, + SandboxUnreachableException, TimeoutException, ) @@ -36,6 +42,22 @@ def test_maps_429_to_rate_limit(): assert "rate limited" in str(err) +# Body of the proxy's 502 when the sandbox is running but nothing listens on envd's port +PORT_CLOSED = "The sandbox is running but port is not open" + + +def _port_closed(request: httpx.Request): + return httpx.Response( + 502, json={"message": PORT_CLOSED, "port": 49983, "code": 502} + ) + + +def test_maps_502_with_port_closed_to_unreachable(): + err = format_envd_api_exception(502, PORT_CLOSED) + assert isinstance(err, SandboxUnreachableException) + assert "could not be reached" in str(err) + + def test_maps_502_to_timeout(): err = format_envd_api_exception(502, "Bad gateway") assert isinstance(err, TimeoutException) @@ -70,11 +92,19 @@ def test_returns_original_when_not_transport_error(): assert err is original +def test_502_sandbox_not_found_returns_not_running(): + err = format_envd_api_exception(502, "The sandbox was not found") + assert isinstance(err, SandboxNotRunningException) + assert isinstance(err, TimeoutException) + assert not isinstance(err, SandboxUnreachableException) + assert "The sandbox was not found" in str(err) + + def test_health_result_confirms_sandbox_killed(): err = handle_envd_api_transport_exception( httpx.RemoteProtocolError("peer closed connection"), sandbox_running=False ) - assert isinstance(err, TimeoutException) + assert isinstance(err, SandboxNotRunningException) assert "sandbox was killed or reached its end of life" in str(err) @@ -82,3 +112,184 @@ def test_health_result_running_returns_raw_error(): original = httpx.RemoteProtocolError("peer closed connection") err = handle_envd_api_transport_exception(original, sandbox_running=True) assert err is original + + +def _health_client(handler) -> httpx.Client: + return httpx.Client( + base_url="http://sandbox", transport=httpx.MockTransport(handler) + ) + + +def _ahealth_client(handler) -> httpx.AsyncClient: + return httpx.AsyncClient( + base_url="http://sandbox", transport=httpx.MockTransport(handler) + ) + + +def _refuse(request: httpx.Request): + raise httpx.ConnectError("connection refused", request=request) + + +def _not_found(request: httpx.Request): + return httpx.Response(502, json={"message": "The sandbox was not found"}) + + +def test_health_check_reports_running(): + with _health_client(lambda request: httpx.Response(204)) as client: + assert check_sandbox_health(client) is True + + +def test_health_check_reports_sandbox_gone(): + with _health_client(_not_found) as client: + assert check_sandbox_health(client) is False + + +def test_health_check_raises_unreachable_for_other_502(): + with _health_client(lambda request: httpx.Response(502)) as client: + try: + check_sandbox_health(client) + except SandboxUnreachableException: + pass + else: + raise AssertionError("expected the probe to raise") + + +def test_health_check_tolerates_non_object_json_502_bodies(): + for body in (["Bad gateway"], 1, {"message": 5}): + with _health_client( + lambda request, body=body: httpx.Response(502, json=body) + ) as client: + try: + check_sandbox_health(client) + except SandboxUnreachableException: + pass + else: + raise AssertionError("expected the probe to raise") + + +def test_health_check_reports_sandbox_gone_for_plain_text_body(): + with _health_client( + lambda request: httpx.Response(502, text="The sandbox was not found") + ) as client: + assert check_sandbox_health(client) is False + + +def test_health_check_raises_unreachable_when_port_closed(): + with _health_client(_port_closed) as client: + try: + check_sandbox_health(client) + except SandboxUnreachableException as e: + assert PORT_CLOSED in str(e) + else: + raise AssertionError("expected the probe to raise") + + +def test_connect_failure_with_port_closed_returns_unreachable(): + original = httpx.ConnectError("connection refused") + with _health_client(_port_closed) as client: + err = handle_envd_api_transport_exception_with_health(original, client) + assert isinstance(err, SandboxUnreachableException) + assert PORT_CLOSED in str(err) + assert err.__cause__ is original + + +def test_health_check_unknown_status_is_inconclusive(): + with _health_client(lambda request: httpx.Response(500)) as client: + assert check_sandbox_health(client) is None + + +def test_health_check_raises_when_sandbox_unreachable(): + with _health_client(_refuse) as client: + try: + check_sandbox_health(client) + except httpx.ConnectError: + pass + else: + raise AssertionError("expected the probe to raise") + + +def test_connect_failure_with_sandbox_killed_returns_not_running(): + with _health_client(_not_found) as client: + err = handle_envd_api_transport_exception_with_health( + httpx.ConnectError("connection refused"), client + ) + assert isinstance(err, SandboxNotRunningException) + + +def test_connect_failure_with_running_sandbox_returns_raw_error(): + original = httpx.ConnectError("connection refused") + with _health_client(lambda request: httpx.Response(204)) as client: + err = handle_envd_api_transport_exception_with_health(original, client) + assert err is original + + +def test_connect_failure_with_unreachable_health_returns_unreachable(): + original = httpx.ConnectError("connection refused") + with _health_client(_refuse) as client: + err = handle_envd_api_transport_exception_with_health(original, client) + assert isinstance(err, SandboxUnreachableException) + assert "could not be reached" in str(err) + assert "connection refused" in str(err) + assert err.__cause__ is original + + +def test_dropped_connection_with_unreachable_health_returns_unreachable(): + original = httpx.RemoteProtocolError("stream reset") + with _health_client(_refuse) as client: + err = handle_envd_api_transport_exception_with_health(original, client) + assert isinstance(err, SandboxUnreachableException) + + +def test_non_transport_error_skips_health_check(): + def fail(request: httpx.Request): + raise AssertionError("health check should not run") + + original = httpx.ReadTimeout("timed out") + with _health_client(fail) as client: + err = handle_envd_api_transport_exception_with_health(original, client) + assert err is original + + +async def test_async_health_check_raises_when_sandbox_unreachable(): + async with _ahealth_client(_refuse) as client: + try: + await acheck_sandbox_health(client) + except httpx.ConnectError: + pass + else: + raise AssertionError("expected the probe to raise") + + +async def test_async_connect_failure_with_unreachable_health_returns_unreachable(): + original = httpx.ConnectError("connection refused") + async with _ahealth_client(_refuse) as client: + err = await ahandle_envd_api_transport_exception_with_health(original, client) + assert isinstance(err, SandboxUnreachableException) + assert err.__cause__ is original + + +async def test_async_connect_failure_with_sandbox_killed_returns_not_running(): + async with _ahealth_client(_not_found) as client: + err = await ahandle_envd_api_transport_exception_with_health( + httpx.ConnectError("connection refused"), client + ) + assert isinstance(err, SandboxNotRunningException) + + +async def test_async_health_check_raises_unreachable_when_port_closed(): + async with _ahealth_client(_port_closed) as client: + try: + await acheck_sandbox_health(client) + except SandboxUnreachableException as e: + assert PORT_CLOSED in str(e) + else: + raise AssertionError("expected the probe to raise") + + +async def test_async_connect_failure_with_port_closed_returns_unreachable(): + original = httpx.ConnectError("connection refused") + async with _ahealth_client(_port_closed) as client: + err = await ahandle_envd_api_transport_exception_with_health(original, client) + assert isinstance(err, SandboxUnreachableException) + assert PORT_CLOSED in str(err) + assert err.__cause__ is original diff --git a/packages/python-sdk/tests/test_envd_client_transport.py b/packages/python-sdk/tests/test_envd_client_transport.py index 5c74c5aa09..4aaee65783 100644 --- a/packages/python-sdk/tests/test_envd_client_transport.py +++ b/packages/python-sdk/tests/test_envd_client_transport.py @@ -110,6 +110,43 @@ def test_sync_pool_is_cached_per_proxy(): assert pool_a is not pool_c +def test_sync_retry_connect_false_returns_the_pool_under_the_retrying_transport(): + key = (None, None, api_client_sync.DEFAULT_HTTP_VERSION) + retrying = api_client_sync.get_pyqwest_transport(None) + plain = api_client_sync.get_pyqwest_transport(None, retry_connect=False) + + assert isinstance(retrying, api_client_sync.ConnectionRetryTransport) + assert isinstance(plain, api_client_sync.BalancingTransport) + assert plain is api_client_sync._pools[key] + assert plain is api_client_sync.get_pyqwest_transport(None, retry_connect=False) + + # The httpx adapters are cached per flag, each over its own transport. + adapter = api_client_sync.get_httpx_transport(None) + plain_adapter = api_client_sync.get_httpx_transport(None, retry_connect=False) + assert adapter is api_client_sync.get_httpx_transport(None) + assert plain_adapter is api_client_sync.get_httpx_transport( + None, retry_connect=False + ) + assert adapter is not plain_adapter + + +def test_async_retry_connect_false_returns_the_pool_under_the_retrying_transport(): + key = (None, None, api_client_async.DEFAULT_HTTP_VERSION) + retrying = api_client_async.get_pyqwest_transport(None) + plain = api_client_async.get_pyqwest_transport(None, retry_connect=False) + + assert isinstance(retrying, api_client_async.ConnectionRetryTransport) + assert isinstance(plain, api_client_async.BalancingTransport) + assert plain is api_client_async._pools[key] + + adapter = api_client_async.get_httpx_transport(None) + plain_adapter = api_client_async.get_httpx_transport(None, retry_connect=False) + assert plain_adapter is api_client_async.get_httpx_transport( + None, retry_connect=False + ) + assert adapter is not plain_adapter + + def test_sync_pool_is_not_shared_across_proxy_credentials(): # Same proxy URL, different credentials or headers: separate pools, since # the proxy configuration is fixed per transport. diff --git a/packages/python-sdk/tests/test_envd_rpc_exception.py b/packages/python-sdk/tests/test_envd_rpc_exception.py index 5a911ec28c..9208e9ba64 100644 --- a/packages/python-sdk/tests/test_envd_rpc_exception.py +++ b/packages/python-sdk/tests/test_envd_rpc_exception.py @@ -16,6 +16,8 @@ NotFoundException, RateLimitException, SandboxException, + SandboxNotRunningException, + SandboxUnreachableException, TimeoutException, ) @@ -41,6 +43,16 @@ def test_maps_resource_exhausted_to_rate_limit(): assert "Rate limit" in str(err) +# Body of the proxy's 502 when the sandbox is running but nothing listens on envd's port +PORT_CLOSED = "The sandbox is running but port is not open" + + +def test_maps_unavailable_with_port_closed_to_unreachable(): + err = handle_rpc_exception(ConnectError(Code.UNAVAILABLE, PORT_CLOSED)) + assert isinstance(err, SandboxUnreachableException) + assert "could not be reached" in str(err) + + def test_maps_unavailable_to_timeout(): err = handle_rpc_exception(ConnectError(Code.UNAVAILABLE, "gone")) assert isinstance(err, TimeoutException) @@ -76,6 +88,12 @@ def _stream_reset() -> ConnectError: return e +def _connect_refused() -> ConnectError: + err = ConnectError(Code.UNKNOWN, "connection refused") + err.__cause__ = ConnectionError("connection refused") + return err + + def _cancelled() -> ConnectError: # connectrpc converts asyncio cancellation into ConnectError(CANCELED) # with the CancelledError as __cause__. @@ -162,10 +180,124 @@ def test_returns_raw_transport_failure_without_health_result(): assert err is original -def test_health_check_confirms_sandbox_killed(): - err = handle_rpc_exception_with_health(_stream_reset(), lambda: False) +def test_unavailable_sandbox_not_found_returns_not_running(): + err = handle_rpc_exception( + ConnectError(Code.UNAVAILABLE, "The sandbox was not found") + ) + assert isinstance(err, SandboxNotRunningException) assert isinstance(err, TimeoutException) + assert not isinstance(err, SandboxUnreachableException) + assert "The sandbox was not found" in str(err) + + +def test_unavailable_port_closed_is_not_not_running(): + err = handle_rpc_exception(ConnectError(Code.UNAVAILABLE, PORT_CLOSED)) + assert isinstance(err, SandboxUnreachableException) + assert not isinstance(err, SandboxNotRunningException) + + +def test_unavailable_sandbox_not_found_does_not_probe(): + probed = [] + + def check(): + probed.append(True) + return True + + err = handle_rpc_exception_with_health( + ConnectError(Code.UNAVAILABLE, "The sandbox was not found"), check + ) + assert not probed + assert isinstance(err, SandboxNotRunningException) + + +# envd ends a stream with this UNAVAILABLE when the sandbox is killed mid-command +STREAM_ENDED = "the connection to sandbox ended before the stream completed" + + +def test_ambiguous_unavailable_probes_health_and_returns_not_running(): + original = ConnectError(Code.UNAVAILABLE, STREAM_ENDED) + err = handle_rpc_exception_with_health(original, lambda: False) + assert isinstance(err, SandboxNotRunningException) + assert not isinstance(err, SandboxUnreachableException) + assert err.__cause__ is original + + +def test_ambiguous_unavailable_with_sandbox_running_is_unreachable_with_unknown_state(): + err = handle_rpc_exception_with_health( + ConnectError(Code.UNAVAILABLE, STREAM_ENDED), lambda: True + ) + assert isinstance(err, SandboxUnreachableException) + assert not isinstance(err, SandboxNotRunningException) + assert "state is unknown" in str(err) + assert "is running but" not in str(err) + + +def test_ambiguous_unavailable_with_failing_probe_is_unreachable(): + def check(): + raise RuntimeError("health check failed") + + original = ConnectError(Code.UNAVAILABLE, STREAM_ENDED) + err = handle_rpc_exception_with_health(original, check) + assert isinstance(err, SandboxUnreachableException) + assert "health check failed" in str(err) + assert err.__cause__ is original + + +def test_ambiguous_unavailable_without_probe_is_unreachable_with_unknown_state(): + err = handle_rpc_exception(ConnectError(Code.UNAVAILABLE, STREAM_ENDED)) + assert isinstance(err, SandboxUnreachableException) + assert "state is unknown" in str(err) + + +def test_port_closed_unavailable_does_not_probe(): + def check(): + raise AssertionError("health check should not run") + + err = handle_rpc_exception_with_health( + ConnectError(Code.UNAVAILABLE, PORT_CLOSED), check + ) + assert isinstance(err, SandboxUnreachableException) + assert "is running but" in str(err) + + +async def test_async_ambiguous_unavailable_probes_health_and_returns_not_running(): + async def check(): + return False + + original = ConnectError(Code.UNAVAILABLE, STREAM_ENDED) + err = await ahandle_rpc_exception_with_health(original, check) + assert isinstance(err, SandboxNotRunningException) + assert err.__cause__ is original + + +async def test_async_ambiguous_unavailable_with_sandbox_running_is_unreachable(): + async def check(): + return True + + err = await ahandle_rpc_exception_with_health( + ConnectError(Code.UNAVAILABLE, STREAM_ENDED), check + ) + assert isinstance(err, SandboxUnreachableException) + assert "state is unknown" in str(err) + + +async def test_async_unavailable_sandbox_not_found_returns_not_running(): + async def check(): + raise AssertionError("health check should not run") + + err = await ahandle_rpc_exception_with_health( + ConnectError(Code.UNAVAILABLE, "The sandbox was not found"), check + ) + assert isinstance(err, SandboxNotRunningException) + + +def test_health_check_confirms_sandbox_killed(): + original = _stream_reset() + err = handle_rpc_exception_with_health(original, lambda: False) + assert isinstance(err, SandboxNotRunningException) + assert not isinstance(err, SandboxUnreachableException) assert "sandbox was killed or reached its end of life" in str(err) + assert err.__cause__ is original def test_health_check_running_returns_raw_error(): @@ -180,15 +312,61 @@ def test_health_check_unknown_returns_raw_error(): assert err is original -def test_health_check_failure_returns_raw_error(): +def test_health_check_port_closed_keeps_probe_exception(): + probe_error = SandboxUnreachableException(PORT_CLOSED) + + def check(): + raise probe_error + + original = _stream_reset() + err = handle_rpc_exception_with_health(original, check) + assert err is probe_error + assert err.__cause__ is original + + +def test_health_check_failure_returns_unreachable(): def check(): raise RuntimeError("health check failed") original = _stream_reset() err = handle_rpc_exception_with_health(original, check) + assert isinstance(err, SandboxUnreachableException) + assert "could not be reached" in str(err) + assert "health check failed" in str(err) + assert err.__cause__ is original + + +def test_connect_failure_is_transport_failure(): + assert is_transport_failure(_connect_refused()) + + +def test_connect_failure_runs_health_check(): + called = False + + def check(): + nonlocal called + called = True + return True + + original = _connect_refused() + err = handle_rpc_exception_with_health(original, check) + assert called assert err is original +def test_connect_failure_with_sandbox_killed_returns_not_running(): + err = handle_rpc_exception_with_health(_connect_refused(), lambda: False) + assert isinstance(err, SandboxNotRunningException) + + +def test_connect_failure_with_unreachable_health_returns_unreachable(): + def check(): + raise ConnectionError("connection refused") + + err = handle_rpc_exception_with_health(_connect_refused(), check) + assert isinstance(err, SandboxUnreachableException) + + def test_health_check_not_run_for_other_exceptions(): def fail(): raise AssertionError("health check should not run") @@ -204,14 +382,23 @@ async def check(): return False err = await ahandle_rpc_exception_with_health(_stream_reset(), check) - assert isinstance(err, TimeoutException) + assert isinstance(err, SandboxNotRunningException) assert "sandbox was killed or reached its end of life" in str(err) -async def test_async_health_check_failure_returns_raw_error(): +async def test_async_health_check_failure_returns_unreachable(): async def check(): raise RuntimeError("health check failed") original = _stream_reset() err = await ahandle_rpc_exception_with_health(original, check) - assert err is original + assert isinstance(err, SandboxUnreachableException) + assert err.__cause__ is original + + +async def test_async_connect_failure_with_unreachable_health_returns_unreachable(): + async def check(): + raise ConnectionError("connection refused") + + err = await ahandle_rpc_exception_with_health(_connect_refused(), check) + assert isinstance(err, SandboxUnreachableException) diff --git a/packages/python-sdk/tests/transport_caches.py b/packages/python-sdk/tests/transport_caches.py index 0ff5ce5f55..ae4a861df8 100644 --- a/packages/python-sdk/tests/transport_caches.py +++ b/packages/python-sdk/tests/transport_caches.py @@ -18,4 +18,5 @@ def reset_transport_caches() -> None: for module in (api_client_sync, api_client_async): module._transports.clear() + module._pools.clear() module._httpx_transports.clear()