fix(warehouse): write Snowflake reconnects to the log

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"
 

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<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<Listener> = ((globalThis as Record<symbol, unknown>)[LISTENERS_KEY] as Set<Listener>) ??
+  ((globalThis as Record<symbol, unknown>)[LISTENERS_KEY] = new Set<Listener>())
+
+/** 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
+    }
+  }
+}

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<any> {
     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<void> {
+  function reconnect(reason: ReconnectEvent["reason"]): Promise<void> {
     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<void> {
     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)

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<void>) {
+    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<void>((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)
+  })
+})

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<string, Set<string>>()
+
+/** 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<string>()
+  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()
+}