From f6843d01c641517deec0e0cf89414f4ebb94df08 Mon Sep 17 00:00:00 2001 From: Haider Date: Thu, 8 Oct 2026 04:03:46 +0530 Subject: [PATCH 1/2] fix(warehouse): write Snowflake reconnects to the log When Snowflake closes a session (idle timeout, VPN drop, laptop sleep), the driver opens a new one and the statement carries on, but nothing reached `opencode.log`. A session that kept dropping looked the same as one that never did, so support could not see it. - The driver emits a process-global reconnect event (`started`, `reconnected`, `failed`), following the browser sign-in notice pattern, with the reason (session reported down, or a statement refused), the duration, the session settings restored, and whether temporary objects or a transaction were lost - `reconnect-log.ts` writes them under `warehouse-connect` as `reconnecting`, `reconnected` and `reconnect failed`, naming the connection when its account maps to one; the error is masked - The registry installs it before each connect, next to the sign-in notice Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_018fJ3X7pcGT4R9yzjsJnqsV --- packages/drivers/src/index.ts | 4 + packages/drivers/src/reconnect-events.ts | 46 +++++++++++ packages/drivers/src/snowflake.ts | 36 ++++++++- .../drivers/test/snowflake-reconnect.test.ts | 73 ++++++++++++++++++ .../native/connections/reconnect-log.ts | 68 ++++++++++++++++ .../altimate/native/connections/registry.ts | 3 + .../altimate/warehouse-reconnect-log.test.ts | 77 +++++++++++++++++++ 7 files changed, 303 insertions(+), 4 deletions(-) create mode 100644 packages/drivers/src/reconnect-events.ts create mode 100644 packages/opencode/src/altimate/native/connections/reconnect-log.ts create mode 100644 packages/opencode/test/altimate/warehouse-reconnect-log.test.ts diff --git a/packages/drivers/src/index.ts b/packages/drivers/src/index.ts index 0c35bb4d8..d89aa73ce 100644 --- a/packages/drivers/src/index.ts +++ b/packages/drivers/src/index.ts @@ -8,6 +8,10 @@ export { normalizeConfig, sanitizeConnectionString } from "./normalize" export { onBrowserSignIn, redactSignInUrl } from "./sign-in" export type { BrowserSignInNotice } from "./sign-in" +// Re-export reconnect events (a closed session reopened by the driver) +export { onReconnect } from "./reconnect-events" +export type { ReconnectEvent } from "./reconnect-events" + // Re-export file-backed store guards export { allowsCreate, assertStoreExists, isLocalFilePath, requireStorePath } from "./file-store" diff --git a/packages/drivers/src/reconnect-events.ts b/packages/drivers/src/reconnect-events.ts new file mode 100644 index 000000000..9726c906b --- /dev/null +++ b/packages/drivers/src/reconnect-events.ts @@ -0,0 +1,46 @@ +/** + * A driver replacing a connection the warehouse closed (today: Snowflake after an idle + * timeout, a network drop or laptop sleep). The reopen is silent to the caller, which is + * the point, but without these events it was also missing from the log: a session that + * kept dropping looked exactly like one that never did. + */ +export interface ReconnectEvent { + warehouse: string + /** Identifies which account the connection belongs to, e.g. the Snowflake account locator. */ + account?: string + phase: "started" | "reconnected" | "failed" + /** How the closed session showed: the SDK reported it down before a statement, or a statement was refused. */ + reason: "connection-down" | "statement-refused" + /** From the start of the reconnect. On `reconnected` and `failed`. */ + durationMs?: number + /** Session settings (USE, SET, ALTER SESSION…) restored on the new session. On `reconnected`. */ + settingsRestored?: number + /** Temporary objects or an open transaction went with the old session. On `reconnected`. */ + sessionStateLost?: boolean + /** On `failed`. */ + error?: string +} + +type Listener = (event: ReconnectEvent) => void + +// Process-global, for the same reason as the sign-in notices: the driver and its +// subscriber can be loaded through different module graphs. +const LISTENERS_KEY = Symbol.for("altimate.drivers.reconnectListeners") +const listeners: Set = ((globalThis as Record)[LISTENERS_KEY] as Set) ?? + ((globalThis as Record)[LISTENERS_KEY] = new Set()) + +/** Subscribe to reconnect events from every driver. Returns the unsubscribe. */ +export function onReconnect(listener: Listener): () => void { + listeners.add(listener) + return () => listeners.delete(listener) +} + +export function emitReconnect(event: ReconnectEvent): void { + for (const listener of listeners) { + try { + listener(event) + } catch { + // a broken listener must not fail the reconnect + } + } +} diff --git a/packages/drivers/src/snowflake.ts b/packages/drivers/src/snowflake.ts index 0284c8263..13587c352 100644 --- a/packages/drivers/src/snowflake.ts +++ b/packages/drivers/src/snowflake.ts @@ -6,6 +6,7 @@ import * as fs from "fs" import type { ConnectionConfig, Connector, ConnectorResult, ExecuteOptions, SchemaColumn } from "./types" import { loadOptionalDriver } from "./resolve" import { emitBrowserSignIn, openInBrowser } from "./sign-in" +import { emitReconnect, type ReconnectEvent } from "./reconnect-events" /** * Run `fn` with stdout/stderr writes swallowed for the (synchronous) duration of @@ -290,9 +291,13 @@ export async function connect( } /** Replace a connection Snowflake has closed. Concurrent callers share one reconnect. */ - function reconnect(): Promise { + function reconnect(reason: ReconnectEvent["reason"]): Promise { if (!reconnecting) { const previous = connection + const account = String(connectOptions?.account ?? "") + const startedAt = Date.now() + let settingsRestored = 0 + emitReconnect({ warehouse: "snowflake", account, phase: "started", reason }) reconnecting = openConnection() .then(async (conn) => { const discard = () => { @@ -320,7 +325,10 @@ export async function connect( } } const unchanged = snapshot.length === sessionSettings.length && snapshot.every((v, i) => v === sessionSettings[i]) - if (unchanged) break + if (unchanged) { + settingsRestored = snapshot.length + break + } if (pass + 1 >= MAX_REPLAY_PASSES) { // Settings keep arriving: a connection installed now would run later queries with some of them missing. discard() @@ -346,6 +354,26 @@ export async function connect( } catch { // already terminated — nothing to release } + emitReconnect({ + warehouse: "snowflake", + account, + phase: "reconnected", + reason, + durationMs: Date.now() - startedAt, + settingsRestored, + sessionStateLost: stateLostAt === generation, + }) + }) + .catch((err) => { + emitReconnect({ + warehouse: "snowflake", + account, + phase: "failed", + reason, + durationMs: Date.now() - startedAt, + error: String((err as Error)?.message ?? err), + }) + throw err }) .finally(() => { reconnecting = undefined @@ -372,7 +400,7 @@ export async function connect( async function ensureLive(): Promise { if (reconnecting) return reconnecting if (connection && connectOptions && typeof connection.isUp === "function" && !connection.isUp()) { - await reconnect() + await reconnect("connection-down") } } @@ -436,7 +464,7 @@ export async function connect( if (!connectOptions || !isClosedConnectionError(err)) throw err // Reopen only if the failed connection is still the current one: a late error from a connection // another statement already replaced must not tear down its replacement. - if (connection === used) await reconnect() + if (connection === used) await reconnect("statement-refused") else if (reconnecting) await reconnecting const cause = String((err as Error)?.message ?? err) if (lostFor(started)) throw sessionLostError(cause) diff --git a/packages/drivers/test/snowflake-reconnect.test.ts b/packages/drivers/test/snowflake-reconnect.test.ts index d7c92ef4a..06c804260 100644 --- a/packages/drivers/test/snowflake-reconnect.test.ts +++ b/packages/drivers/test/snowflake-reconnect.test.ts @@ -505,3 +505,76 @@ describe("keepAliveSetting", () => { } }) }) + +describe("reconnect events", () => { + const { onReconnect } = require("../src/reconnect-events") + + async function collect(fn: (events: any[]) => Promise) { + const events: any[] = [] + const off = onReconnect((e: any) => void events.push(e)) + try { + await fn(events) + } finally { + off() + } + return events + } + + test("a session reported down is reported as reconnecting, then reconnected with its settings restored", async () => { + const { sdk, created } = fakeSdk() + const c = await connect(config, sdk) + await c.connect() + const events = await collect(async () => { + await c.execute("USE SCHEMA ANALYTICS") + created[0].state.up = false + await c.execute("SELECT 1") + }) + expect(events.map((e) => [e.phase, e.reason])).toEqual([ + ["started", "connection-down"], + ["reconnected", "connection-down"], + ]) + expect(events[1]).toMatchObject({ warehouse: "snowflake", account: "acct", settingsRestored: 1, sessionStateLost: false }) + expect(typeof events[1].durationMs).toBe("number") + }) + + test("a statement refused because the session closed is reported with that reason", async () => { + const { sdk, created } = fakeSdk() + const c = await connect(config, sdk) + await c.connect() + created[0].state.failNextWith = { code: 407002, message: "Unable to perform operation using terminated connection." } + const events = await collect(async () => { + await c.execute("SELECT 1") + }) + expect(events.map((e) => [e.phase, e.reason])).toEqual([ + ["started", "statement-refused"], + ["reconnected", "statement-refused"], + ]) + }) + + test("a reconnect that cannot restore the session is reported as failed, with the error", async () => { + const { sdk, created } = fakeSdk() + const c = await connect(config, sdk) + await c.connect() + await c.execute("USE SCHEMA ANALYTICS") + created[0].state.up = false + // The new session refuses the replayed setting. + sdk.nextState = { failNextWith: { code: 2003, message: "Schema 'ANALYTICS' does not exist" } } + const events = await collect(async () => { + await expect(c.execute("SELECT 1")).rejects.toThrow() + }) + expect(events.map((e) => e.phase)).toEqual(["started", "failed"]) + expect(events[1].error).toContain("does not exist") + }) + + test("losing temporary objects with the old session is reported", async () => { + const { sdk, created } = fakeSdk() + const c = await connect(config, sdk) + await c.connect() + await c.execute("CREATE TEMPORARY TABLE T1 (X INT)") + created[0].state.up = false + const events = await collect(async () => { + await c.execute("SELECT 1").catch(() => {}) + }) + expect(events.find((e) => e.phase === "reconnected")?.sessionStateLost).toBe(true) + }) +}) diff --git a/packages/opencode/src/altimate/native/connections/reconnect-log.ts b/packages/opencode/src/altimate/native/connections/reconnect-log.ts new file mode 100644 index 000000000..daf2d637f --- /dev/null +++ b/packages/opencode/src/altimate/native/connections/reconnect-log.ts @@ -0,0 +1,68 @@ +// altimate_change - new file +/** + * Writes a driver's reconnects to `opencode.log`. When Snowflake closes a session (idle + * timeout, VPN drop, laptop sleep) the driver opens a new one and the statement goes on, + * so the user sees nothing; without these lines the log did not show it either, and a + * session that kept dropping looked the same as one that never did. + */ +import { onReconnect, type ReconnectEvent } from "@altimateai/drivers" +import { fileLog } from "@/altimate/util/file-log" +import { Telemetry } from "@/altimate/telemetry" + +type Write = typeof fileLog + +/** Connection names by warehouse account: the driver knows the account, the log lines name the connection. */ +const names = new Map>() + +/** Called before each connect, so a later reconnect on that account can name its connection. */ +export function remember(account: string, name: string): void { + if (!account) return + const set = names.get(account) ?? new Set() + set.add(name) + names.set(account, set) +} + +/** One name only when the account maps to exactly one connection; two connections on one account are both possible. */ +function nameFor(account: string | undefined): string | undefined { + const set = account ? names.get(account) : undefined + return set && set.size === 1 ? [...set][0] : undefined +} + +export function handle(e: ReconnectEvent, write: Write = fileLog): void { + const base = { + ...(nameFor(e.account) ? { name: nameFor(e.account) } : {}), + type: e.warehouse, + ...(e.account ? { account: e.account } : {}), + reason: e.reason, + } + if (e.phase === "started") { + write("INFO", "warehouse-connect", "reconnecting", base) + } else if (e.phase === "reconnected") { + write("INFO", "warehouse-connect", "reconnected", { + ...base, + duration_ms: e.durationMs, + settings_restored: e.settingsRestored, + session_state_lost: e.sessionStateLost, + }) + } else { + write("WARN", "warehouse-connect", "reconnect failed", { + ...base, + duration_ms: e.durationMs, + error: Telemetry.maskString(String(e.error ?? "")).slice(0, 500), + }) + } +} + +let unsubscribe: (() => void) | undefined + +/** Idempotent; called before every warehouse connect. */ +export function install(write: Write = fileLog): void { + if (unsubscribe) return + unsubscribe = onReconnect((e) => handle(e, write)) +} + +export function resetForTests(): void { + unsubscribe?.() + unsubscribe = undefined + names.clear() +} diff --git a/packages/opencode/src/altimate/native/connections/registry.ts b/packages/opencode/src/altimate/native/connections/registry.ts index aa2c2f912..38a752a8a 100644 --- a/packages/opencode/src/altimate/native/connections/registry.ts +++ b/packages/opencode/src/altimate/native/connections/registry.ts @@ -30,6 +30,7 @@ import { startTunnel, extractSshConfig, closeTunnel } from "./ssh-tunnel" import type { WarehouseInfo } from "../types" import { Telemetry } from "../../../telemetry" import * as SignInNotice from "./sign-in-notice" +import * as ReconnectLog from "./reconnect-log" import { fileLog } from "@/altimate/util/file-log" /** In-memory config store. */ @@ -518,6 +519,8 @@ export async function get(name: string): Promise { // the outcome event below, so without this it leaves no trace at all. SignInNotice.install() SignInNotice.rememberOrigin(accountOf(config)) + ReconnectLog.install() + ReconnectLog.remember(accountOf(config), name) fileLog("INFO", "warehouse-connect", "connecting", { name, type: config.type, auth: authMethod }) try { Telemetry.track({ diff --git a/packages/opencode/test/altimate/warehouse-reconnect-log.test.ts b/packages/opencode/test/altimate/warehouse-reconnect-log.test.ts new file mode 100644 index 000000000..889545b15 --- /dev/null +++ b/packages/opencode/test/altimate/warehouse-reconnect-log.test.ts @@ -0,0 +1,77 @@ +// altimate_change - new file +/** + * A driver reopening a session the warehouse closed (idle timeout, VPN drop, sleep) is written to + * opencode.log. The statement carries on and the user sees nothing, so the log is the only trace of it. + */ +import { afterEach, beforeEach, describe, expect, test } from "bun:test" +import * as ReconnectLog from "../../src/altimate/native/connections/reconnect-log" +import { emitReconnect } from "../../../drivers/src/reconnect-events" + +function recorder() { + const lines: Array<{ level: string; service: string; message: string; fields: Record }> = [] + const write = (level: string, service: string, message: string, fields: Record = {}) => + void lines.push({ level, service, message, fields }) + return { lines, write: write as any } +} + +// Connects elsewhere in the run install the default writer; each test starts unsubscribed. +beforeEach(() => ReconnectLog.resetForTests()) +afterEach(() => ReconnectLog.resetForTests()) + +describe("reconnect log", () => { + test("a reconnect is written as reconnecting then reconnected, under the connection's name", () => { + const r = recorder() + ReconnectLog.install(r.write) + ReconnectLog.remember("acme-xy123", "prod_snowflake") + emitReconnect({ warehouse: "snowflake", account: "acme-xy123", phase: "started", reason: "connection-down" }) + emitReconnect({ + warehouse: "snowflake", + account: "acme-xy123", + phase: "reconnected", + reason: "connection-down", + durationMs: 812, + settingsRestored: 2, + sessionStateLost: false, + }) + expect(r.lines.map((l) => [l.level, l.service, l.message])).toEqual([ + ["INFO", "warehouse-connect", "reconnecting"], + ["INFO", "warehouse-connect", "reconnected"], + ]) + expect(r.lines[1].fields).toEqual({ + name: "prod_snowflake", + type: "snowflake", + account: "acme-xy123", + reason: "connection-down", + duration_ms: 812, + settings_restored: 2, + session_state_lost: false, + }) + }) + + test("a failed reconnect is a warning, with the error masked", () => { + const r = recorder() + ReconnectLog.handle( + { + warehouse: "snowflake", + account: "acme-xy123", + phase: "failed", + reason: "statement-refused", + durationMs: 120_000, + error: "login failed for jane@example.com", + }, + r.write, + ) + expect(r.lines[0].level).toBe("WARN") + expect(r.lines[0].message).toBe("reconnect failed") + expect(String(r.lines[0].fields.error)).not.toContain("jane@example.com") + }) + + test("an account shared by two connections is logged by account only", () => { + const r = recorder() + ReconnectLog.remember("shared-acct", "a") + ReconnectLog.remember("shared-acct", "b") + ReconnectLog.handle({ warehouse: "snowflake", account: "shared-acct", phase: "started", reason: "connection-down" }, r.write) + expect(r.lines[0].fields.name).toBeUndefined() + expect(r.lines[0].fields.account).toBe("shared-acct") + }) +}) From 6b73d107f369922ee640bd75cd94828d4a2ff8a0 Mon Sep 17 00:00:00 2001 From: Haider Date: Fri, 9 Oct 2026 03:02:34 +0530 Subject: [PATCH 2/2] fix(warehouse): address review on the reconnect log - The `failed` event no longer carries the replayed setting. It goes to every subscriber in the process, while the caller's own error still names it. - `emitReconnect` absorbs a rejecting async listener instead of leaving an unhandled rejection. - The reason `statement-refused` is now `closed-during-statement`: the SDK raises the same error while polling a statement Snowflake already accepted, so "refused" could send someone to re-run a write that ran. - `sessionStateLost` is also true while a statement that can create temporary objects or open a transaction is still running on the old session, since whether it did is not known yet. - `registry.remove()` drops the connection's name from the reconnect log, so a later reconnect on that account is not put down to it. - The masking test now also checks the error text survives. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_018fJ3X7pcGT4R9yzjsJnqsV --- packages/drivers/src/reconnect-events.ts | 17 +++--- packages/drivers/src/snowflake.ts | 30 ++++++++-- .../drivers/test/snowflake-reconnect.test.ts | 56 ++++++++++++++++++- .../native/connections/reconnect-log.ts | 8 +++ .../altimate/native/connections/registry.ts | 1 + .../altimate/warehouse-reconnect-log.test.ts | 18 +++++- 6 files changed, 114 insertions(+), 16 deletions(-) diff --git a/packages/drivers/src/reconnect-events.ts b/packages/drivers/src/reconnect-events.ts index 9726c906b..62000debe 100644 --- a/packages/drivers/src/reconnect-events.ts +++ b/packages/drivers/src/reconnect-events.ts @@ -9,19 +9,21 @@ export interface ReconnectEvent { /** Identifies which account the connection belongs to, e.g. the Snowflake account locator. */ account?: string phase: "started" | "reconnected" | "failed" - /** How the closed session showed: the SDK reported it down before a statement, or a statement was refused. */ - reason: "connection-down" | "statement-refused" + /** How the closed session showed: the SDK reported it down before a statement, or a statement failed because + * the session had closed. The second does not say the statement did not run: a write may already have. */ + reason: "connection-down" | "closed-during-statement" /** From the start of the reconnect. On `reconnected` and `failed`. */ durationMs?: number /** Session settings (USE, SET, ALTER SESSION…) restored on the new session. On `reconnected`. */ settingsRestored?: number - /** Temporary objects or an open transaction went with the old session. On `reconnected`. */ + /** Temporary objects or an open transaction went with the old session, or a statement that could create them + * was still running on it, so whether they did is not known yet. On `reconnected`. */ sessionStateLost?: boolean - /** On `failed`. */ + /** On `failed`. Never carries the session's statements (a replayed setting can hold anything the user ran). */ error?: string } -type Listener = (event: ReconnectEvent) => void +type Listener = (event: ReconnectEvent) => void | Promise // Process-global, for the same reason as the sign-in notices: the driver and its // subscriber can be loaded through different module graphs. @@ -37,10 +39,11 @@ export function onReconnect(listener: Listener): () => void { export function emitReconnect(event: ReconnectEvent): void { for (const listener of listeners) { + // A broken listener, sync or async, must not fail the reconnect or leave an unhandled rejection. try { - listener(event) + void Promise.resolve(listener(event)).catch(() => {}) } catch { - // a broken listener must not fail the reconnect + // as above } } } diff --git a/packages/drivers/src/snowflake.ts b/packages/drivers/src/snowflake.ts index 13587c352..bb4156f1c 100644 --- a/packages/drivers/src/snowflake.ts +++ b/packages/drivers/src/snowflake.ts @@ -215,6 +215,9 @@ export async function connect( /** The generation at which that state was lost: statements that started before it are not run, because they * may depend on it; statements issued after it run on the new session. Never cleared by a late error. */ let stateLostAt: number | undefined + /** Statements running now that may create temporary objects or open a transaction: one that finishes after a + * reconnect is reported to its caller then, so the reconnect cannot yet say the session lost nothing. */ + let stateStatementsRunning = 0 function openConnection(): Promise { const account = connectOptions?.account @@ -319,8 +322,14 @@ export async function connect( await runQuery(conn, setting) } catch (err) { discard() - throw new Error( - `Snowflake closed the session and its settings could not be restored on a new one (${setting}): ${(err as Error)?.message ?? err}`, + // The caller ran this setting and is told which; the reconnect event, which any subscriber in the + // process receives, gets the message without it. + const cause = (err as Error)?.message ?? err + throw Object.assign( + new Error( + `Snowflake closed the session and its settings could not be restored on a new one (${setting}): ${cause}`, + ), + { eventMessage: `Snowflake closed the session and its settings could not be restored on a new one: ${cause}` }, ) } } @@ -361,7 +370,7 @@ export async function connect( reason, durationMs: Date.now() - startedAt, settingsRestored, - sessionStateLost: stateLostAt === generation, + sessionStateLost: stateLostAt === generation || stateStatementsRunning > 0, }) }) .catch((err) => { @@ -371,7 +380,7 @@ export async function connect( phase: "failed", reason, durationMs: Date.now() - startedAt, - error: String((err as Error)?.message ?? err), + error: String((err as { eventMessage?: string })?.eventMessage ?? (err as Error)?.message ?? err), }) throw err }) @@ -449,6 +458,17 @@ export async function connect( * it may already have run (see `isRetrySafe`). */ async function executeQuery(sql: string, binds?: any[]): Promise<{ columns: string[]; rows: any[][] }> { + const mayHoldState = opensTransaction(sql) || holdsSessionState(sql) || (looksLikeSessionChange(sql) && !isSessionSetting(sql)) + if (!mayHoldState) return executeTracked(sql, binds) + stateStatementsRunning++ + try { + return await executeTracked(sql, binds) + } finally { + stateStatementsRunning-- + } + } + + async function executeTracked(sql: string, binds?: any[]): Promise<{ columns: string[]; rows: any[][] }> { // Captured before any wait: a statement queued behind a reconnect started on the old session's assumptions. const started = generation const startedInEpoch = epoch @@ -464,7 +484,7 @@ export async function connect( if (!connectOptions || !isClosedConnectionError(err)) throw err // Reopen only if the failed connection is still the current one: a late error from a connection // another statement already replaced must not tear down its replacement. - if (connection === used) await reconnect("statement-refused") + if (connection === used) await reconnect("closed-during-statement") else if (reconnecting) await reconnecting const cause = String((err as Error)?.message ?? err) if (lostFor(started)) throw sessionLostError(cause) diff --git a/packages/drivers/test/snowflake-reconnect.test.ts b/packages/drivers/test/snowflake-reconnect.test.ts index 06c804260..37cf8566a 100644 --- a/packages/drivers/test/snowflake-reconnect.test.ts +++ b/packages/drivers/test/snowflake-reconnect.test.ts @@ -537,7 +537,7 @@ describe("reconnect events", () => { expect(typeof events[1].durationMs).toBe("number") }) - test("a statement refused because the session closed is reported with that reason", async () => { + test("a statement that fails because the session closed is reported with that reason", async () => { const { sdk, created } = fakeSdk() const c = await connect(config, sdk) await c.connect() @@ -546,8 +546,8 @@ describe("reconnect events", () => { await c.execute("SELECT 1") }) expect(events.map((e) => [e.phase, e.reason])).toEqual([ - ["started", "statement-refused"], - ["reconnected", "statement-refused"], + ["started", "closed-during-statement"], + ["reconnected", "closed-during-statement"], ]) }) @@ -564,6 +564,56 @@ describe("reconnect events", () => { }) expect(events.map((e) => e.phase)).toEqual(["started", "failed"]) expect(events[1].error).toContain("does not exist") + // Every subscriber in the process gets this event; the setting the user ran stays with the caller. + expect(events[1].error).not.toContain("USE SCHEMA ANALYTICS") + }) + + test("the caller still learns which setting could not be restored", async () => { + const { sdk, created } = fakeSdk() + const c = await connect(config, sdk) + await c.connect() + await c.execute("USE SCHEMA ANALYTICS") + created[0].state.up = false + sdk.nextState = { failNextWith: { code: 2003, message: "Schema 'ANALYTICS' does not exist" } } + await expect(c.execute("SELECT 1")).rejects.toThrow("USE SCHEMA ANALYTICS") + }) + + test("a statement that may create session state, still running at the reconnect, is reported as possible loss", async () => { + const { sdk, created } = fakeSdk() + const c = await connect(config, sdk) + await c.connect() + let release!: () => void + created[0].state.holdNext = new Promise((r) => (release = r)) + const events = await collect(async () => { + const temp = c.execute("CREATE TEMPORARY TABLE T1 (X INT)").catch(() => {}) + await Bun.sleep(5) + created[0].state.up = false + await c.execute("SELECT 1").catch(() => {}) + release() + await temp + }) + expect(events.find((e) => e.phase === "reconnected")?.sessionStateLost).toBe(true) + }) + + test("an async listener that rejects does not leave an unhandled rejection", async () => { + const { sdk, created } = fakeSdk() + const c = await connect(config, sdk) + await c.connect() + const unhandled: unknown[] = [] + const onUnhandled = (e: unknown) => unhandled.push(e) + process.on("unhandledRejection", onUnhandled) + const off = onReconnect(async () => { + throw new Error("listener broke") + }) + try { + created[0].state.up = false + await c.execute("SELECT 1") + await Bun.sleep(20) + } finally { + off() + process.off("unhandledRejection", onUnhandled) + } + expect(unhandled).toEqual([]) }) test("losing temporary objects with the old session is reported", async () => { diff --git a/packages/opencode/src/altimate/native/connections/reconnect-log.ts b/packages/opencode/src/altimate/native/connections/reconnect-log.ts index daf2d637f..c266cc923 100644 --- a/packages/opencode/src/altimate/native/connections/reconnect-log.ts +++ b/packages/opencode/src/altimate/native/connections/reconnect-log.ts @@ -22,6 +22,14 @@ export function remember(account: string, name: string): void { names.set(account, set) } +/** Called when a connection is removed, so a later reconnect on its account is not put down to it. */ +export function forget(name: string): void { + for (const [account, set] of names) { + set.delete(name) + if (set.size === 0) names.delete(account) + } +} + /** One name only when the account maps to exactly one connection; two connections on one account are both possible. */ function nameFor(account: string | undefined): string | undefined { const set = account ? names.get(account) : undefined diff --git a/packages/opencode/src/altimate/native/connections/registry.ts b/packages/opencode/src/altimate/native/connections/registry.ts index 38a752a8a..d2822eeaf 100644 --- a/packages/opencode/src/altimate/native/connections/registry.ts +++ b/packages/opencode/src/altimate/native/connections/registry.ts @@ -757,6 +757,7 @@ export async function remove(name: string): Promise<{ success: boolean; error?: // Close SSH tunnel if active closeTunnel(name) + ReconnectLog.forget(name) // Remove from global config file const globalPath = globalConfigPath() diff --git a/packages/opencode/test/altimate/warehouse-reconnect-log.test.ts b/packages/opencode/test/altimate/warehouse-reconnect-log.test.ts index 889545b15..2fcce4b51 100644 --- a/packages/opencode/test/altimate/warehouse-reconnect-log.test.ts +++ b/packages/opencode/test/altimate/warehouse-reconnect-log.test.ts @@ -55,7 +55,7 @@ describe("reconnect log", () => { warehouse: "snowflake", account: "acme-xy123", phase: "failed", - reason: "statement-refused", + reason: "closed-during-statement", durationMs: 120_000, error: "login failed for jane@example.com", }, @@ -63,9 +63,25 @@ describe("reconnect log", () => { ) expect(r.lines[0].level).toBe("WARN") expect(r.lines[0].message).toBe("reconnect failed") + expect(String(r.lines[0].fields.error)).toContain("login failed") expect(String(r.lines[0].fields.error)).not.toContain("jane@example.com") }) + test("a removed connection is no longer named for its account", () => { + const r = recorder() + ReconnectLog.remember("acme-xy123", "old_conn") + ReconnectLog.forget("old_conn") + ReconnectLog.handle({ warehouse: "snowflake", account: "acme-xy123", phase: "started", reason: "connection-down" }, r.write) + expect(r.lines[0].fields.name).toBeUndefined() + + // Removing one of two connections on an account names the other again. + ReconnectLog.remember("shared-acct", "a") + ReconnectLog.remember("shared-acct", "b") + ReconnectLog.forget("a") + ReconnectLog.handle({ warehouse: "snowflake", account: "shared-acct", phase: "started", reason: "connection-down" }, r.write) + expect(r.lines[1].fields.name).toBe("b") + }) + test("an account shared by two connections is logged by account only", () => { const r = recorder() ReconnectLog.remember("shared-acct", "a")