From f6843d01c641517deec0e0cf89414f4ebb94df08 Mon Sep 17 00:00:00 2001 From: Haider Date: Thu, 8 Oct 2026 04:03:46 +0530 Subject: [PATCH 1/3] 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 0c35bb4d82..d89aa73ced 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 0000000000..9726c906b7 --- /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 0284c8263f..13587c3526 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 d7c92ef4a6..06c804260e 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 0000000000..daf2d637fd --- /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 aa2c2f9128..38a752a8a5 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 0000000000..889545b154 --- /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/3] 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 9726c906b7..62000debe1 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 13587c3526..bb4156f1c9 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 06c804260e..37cf8566ab 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 daf2d637fd..c266cc9230 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 38a752a8a5..d2822eeaf9 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 889545b154..2fcce4b51b 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") From 3585f359912c6f45fae8ffa53f78d5f41eac6091 Mon Sep 17 00:00:00 2001 From: Haider Date: Mon, 12 Oct 2026 21:27:25 +0530 Subject: [PATCH 3/3] fix(warehouse): count only what ran on the replaced session; keep SQL out of reconnect events Second review round on the reconnect log. - A statement that may hold session state is counted against the session it was sent to, from the moment it is sent. One queued behind a reconnect no longer marks that reconnect as losing state, and one still running on an older session no longer marks a later reconnect. - The test is the one `noteSession` applies, so `SET v = ?` with binds (not replayable) is counted too. - The `failed` event carries only the Snowflake error code for a setting that could not be restored. The SDK's message can quote the setting; the caller's own error still has both. - `registry.remove()` keeps the name of a connection whose connect is still in flight, so a reconnect on a shared account is not put down to the other. Co-Authored-By: Claude Opus 5.5 --- packages/drivers/src/snowflake.ts | 56 +++++++++++------ .../drivers/test/snowflake-reconnect.test.ts | 63 ++++++++++++++++++- .../altimate/native/connections/registry.ts | 4 +- .../altimate/warehouse-reconnect-log.test.ts | 24 ++++++- 4 files changed, 124 insertions(+), 23 deletions(-) diff --git a/packages/drivers/src/snowflake.ts b/packages/drivers/src/snowflake.ts index bb4156f1c9..ec1caf2277 100644 --- a/packages/drivers/src/snowflake.ts +++ b/packages/drivers/src/snowflake.ts @@ -215,9 +215,10 @@ 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 + /** Statements that may create temporary objects or open a transaction, by the generation of the session they were + * sent to, while they run: one that finishes after a reconnect is reported to its caller then, so the reconnect + * cannot yet say the replaced session lost nothing. */ + const stateStatementsRunning = new Map() function openConnection(): Promise { const account = connectOptions?.account @@ -324,12 +325,19 @@ export async function connect( discard() // The caller ran this setting and is told which; the reconnect event, which any subscriber in the // process receives, gets the message without it. + // The SDK's message can quote the setting (a schema or value it names), so the event carries only its + // error code. const cause = (err as Error)?.message ?? err + const code = (err as { code?: unknown })?.code 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}` }, + { + eventMessage: + "Snowflake closed the session and its settings could not be restored on a new one" + + (code !== undefined ? ` (Snowflake error ${String(code)})` : ""), + }, ) } } @@ -350,6 +358,8 @@ export async function connect( discard() throw new Error("Snowflake connection was closed while reconnecting") } + // Read before the generation moves on: statements still running on the session being replaced. + const stateStillRunning = (stateStatementsRunning.get(generation) ?? 0) > 0 generation++ if (tempIn !== undefined || txnIn !== undefined) { stateLostAt = generation @@ -370,7 +380,7 @@ export async function connect( reason, durationMs: Date.now() - startedAt, settingsRestored, - sessionStateLost: stateLostAt === generation || stateStatementsRunning > 0, + sessionStateLost: stateLostAt === generation || stateStillRunning, }) }) .catch((err) => { @@ -458,17 +468,6 @@ 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 @@ -477,7 +476,7 @@ export async function connect( const used = connection const ranIn = generation try { - const result = await runQuery(used, sql, binds) + const result = await runTracked(used, sql, binds, ranIn) await noteSession(sql, binds, ranIn, startedInEpoch) return result } catch (err) { @@ -496,12 +495,33 @@ export async function connect( ) } const ranAgainIn = generation - const result = await runQuery(connection, sql, binds) + const result = await runTracked(connection, sql, binds, ranAgainIn) await noteSession(sql, binds, ranAgainIn, startedInEpoch) return result } } + /** Whether a statement may create temporary objects or open a transaction: the same test `noteSession` applies + * when it finishes, so a statement it would report as lost is also counted while it runs. */ + function mayHoldState(sql: string, binds?: any[]): boolean { + const replayable = isSessionSetting(sql) && !(binds && binds.length) + return opensTransaction(sql) || holdsSessionState(sql) || (looksLikeSessionChange(sql) && !replayable) + } + + /** `runQuery`, counting a statement that may hold session state against the session it was sent to (`ran`) for + * as long as it runs. Counted only once sent: one still waiting on a reconnect will run on the new session. */ + async function runTracked(conn: any, sql: string, binds: any[] | undefined, ran: number) { + if (!mayHoldState(sql, binds)) return runQuery(conn, sql, binds) + stateStatementsRunning.set(ran, (stateStatementsRunning.get(ran) ?? 0) + 1) + try { + return await runQuery(conn, sql, binds) + } finally { + const left = (stateStatementsRunning.get(ran) ?? 1) - 1 + if (left > 0) stateStatementsRunning.set(ran, left) + else stateStatementsRunning.delete(ran) + } + } + function runQuery(conn: any, sql: string, binds?: any[]): Promise<{ columns: string[]; rows: any[][] }> { return new Promise((resolve, reject) => { const options: Record = { diff --git a/packages/drivers/test/snowflake-reconnect.test.ts b/packages/drivers/test/snowflake-reconnect.test.ts index 37cf8566ab..e877b1b87f 100644 --- a/packages/drivers/test/snowflake-reconnect.test.ts +++ b/packages/drivers/test/snowflake-reconnect.test.ts @@ -563,9 +563,10 @@ describe("reconnect events", () => { 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") - // 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") + // Every subscriber in the process gets this event. The setting the user ran, and the SDK's message (which + // quotes the schema it names), stay with the caller; the event says which Snowflake error it was. + expect(events[1].error).toContain("Snowflake error 2003") + expect(events[1].error).not.toContain("ANALYTICS") }) test("the caller still learns which setting could not be restored", async () => { @@ -595,6 +596,62 @@ describe("reconnect events", () => { expect(events.find((e) => e.phase === "reconnected")?.sessionStateLost).toBe(true) }) + test("a stateful statement queued behind a reconnect is not counted against the replaced session", async () => { + const { sdk, created } = fakeSdk() + const c = await connect(config, sdk) + await c.connect() + created[0].state.up = false + // The reconnect opens slowly; BEGIN arrives while it is under way and runs only on the new session. + sdk.hold = true + const events = await collect(async () => { + const first = c.execute("SELECT 1") + await Bun.sleep(5) + const begin = c.execute("BEGIN") + await Bun.sleep(5) + sdk.release() + await Promise.all([first, begin]) + }) + expect(events.find((e) => e.phase === "reconnected")?.sessionStateLost).toBe(false) + }) + + test("a session setting with binds, 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 () => { + // Not replayable (it has binds), so finishing on the replaced session loses it, as `noteSession` reports. + const setting = c.execute("SET V = ?", undefined, [1]).catch(() => {}) + await Bun.sleep(5) + created[0].state.up = false + await c.execute("SELECT 1").catch(() => {}) + release() + await setting + }) + expect(events.find((e) => e.phase === "reconnected")?.sessionStateLost).toBe(true) + }) + + test("a stateful statement left on an older session does not mark a later reconnect", 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(() => {}) // first reconnect: replaces the session T1 is running on + created[1].state.up = false + await c.execute("SELECT 1").catch(() => {}) // second: replaces a session nothing stateful ran on + release() + await temp + }) + const reconnected = events.filter((e) => e.phase === "reconnected") + expect(reconnected.map((e) => e.sessionStateLost)).toEqual([true, false]) + }) + test("an async listener that rejects does not leave an unhandled rejection", async () => { const { sdk, created } = fakeSdk() const c = await connect(config, sdk) diff --git a/packages/opencode/src/altimate/native/connections/registry.ts b/packages/opencode/src/altimate/native/connections/registry.ts index d2822eeaf9..930c6a46ce 100644 --- a/packages/opencode/src/altimate/native/connections/registry.ts +++ b/packages/opencode/src/altimate/native/connections/registry.ts @@ -757,7 +757,9 @@ export async function remove(name: string): Promise<{ success: boolean; error?: // Close SSH tunnel if active closeTunnel(name) - ReconnectLog.forget(name) + // A connect still in flight for this name will finish and keep its name in use; forgetting it now would let a + // later reconnect on that account be put down to another connection that shares it. + if (!pending.has(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 2fcce4b51b..c48163296c 100644 --- a/packages/opencode/test/altimate/warehouse-reconnect-log.test.ts +++ b/packages/opencode/test/altimate/warehouse-reconnect-log.test.ts @@ -3,8 +3,9 @@ * 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 { afterEach, beforeEach, describe, expect, spyOn, test } from "bun:test" import * as ReconnectLog from "../../src/altimate/native/connections/reconnect-log" +import * as Registry from "../../src/altimate/native/connections/registry" import { emitReconnect } from "../../../drivers/src/reconnect-events" function recorder() { @@ -90,4 +91,25 @@ describe("reconnect log", () => { expect(r.lines[0].fields.name).toBeUndefined() expect(r.lines[0].fields.account).toBe("shared-acct") }) + + // `get()` records the name before it connects; a `remove()` that lands while that connect is still in flight + // must not forget it, or a later reconnect on a shared account is put down to the other connection. + test("removing a connection whose connect is still in flight keeps its name", async () => { + const forget = spyOn(ReconnectLog, "forget") + try { + Registry.setConfigs({ in_flight: { type: "duckdb", path: ":memory:" }, settled: { type: "duckdb", path: ":memory:" } }) + const connecting = Registry.get("in_flight") + await Registry.remove("in_flight") + expect(forget).not.toHaveBeenCalledWith("in_flight") + await connecting.catch(() => {}) + + // With nothing in flight, the name is forgotten. + await Registry.get("settled") + await Registry.remove("settled") + expect(forget).toHaveBeenCalledWith("settled") + } finally { + forget.mockRestore() + Registry.reset() + } + }) })