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..62000debe --- /dev/null +++ b/packages/drivers/src/reconnect-events.ts @@ -0,0 +1,49 @@ +/** + * 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 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, 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`. Never carries the session's statements (a replayed setting can hold anything the user ran). */ + error?: string +} + +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. +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) { + // A broken listener, sync or async, must not fail the reconnect or leave an unhandled rejection. + try { + void Promise.resolve(listener(event)).catch(() => {}) + } catch { + // as above + } + } +} diff --git a/packages/drivers/src/snowflake.ts b/packages/drivers/src/snowflake.ts index 0284c8263..bb4156f1c 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 @@ -214,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 @@ -290,9 +294,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 = () => { @@ -314,13 +322,22 @@ 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}` }, ) } } 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 +363,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 || stateStatementsRunning > 0, + }) + }) + .catch((err) => { + emitReconnect({ + warehouse: "snowflake", + account, + phase: "failed", + reason, + durationMs: Date.now() - startedAt, + error: String((err as { eventMessage?: string })?.eventMessage ?? (err as Error)?.message ?? err), + }) + throw err }) .finally(() => { reconnecting = undefined @@ -372,7 +409,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") } } @@ -421,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 @@ -436,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() + 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 d7c92ef4a..37cf8566a 100644 --- a/packages/drivers/test/snowflake-reconnect.test.ts +++ b/packages/drivers/test/snowflake-reconnect.test.ts @@ -505,3 +505,126 @@ 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 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() + 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", "closed-during-statement"], + ["reconnected", "closed-during-statement"], + ]) + }) + + 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") + // 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 () => { + 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..c266cc923 --- /dev/null +++ b/packages/opencode/src/altimate/native/connections/reconnect-log.ts @@ -0,0 +1,76 @@ +// 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) +} + +/** 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 + 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..d2822eeaf 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({ @@ -754,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 new file mode 100644 index 000000000..2fcce4b51 --- /dev/null +++ b/packages/opencode/test/altimate/warehouse-reconnect-log.test.ts @@ -0,0 +1,93 @@ +// 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: "closed-during-statement", + 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)).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") + 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") + }) +})