The DuckLake catalog answers a call that exits as an error, so retention, GC and the compactor keep their state (T-464)

lib/smolquery/catalog/ducklake.ex

@@ -98,6 +98,19 @@ defmodule Smolquery.Catalog.DuckLake do
   way: a relative path returned as-is would match no store location, and GC would
   again see every committed segment as unreferenced.
 
+  ## A call that exits is an error here, never a crash upstream
+
+  Every statement this module runs goes through `Smolquery.Engine.try_query/4`
+  or `Smolquery.Engine.try_transaction/3`, so a call that times out on a busy
+  connection, or finds the connection gone, comes back as
+  `{:error, %Smolquery.Engine.CallExited{}}` like any other failure (T-464).
+  Every `Smolquery.Catalog` callback already promises `{:error, term()}`, and
+  the callers that matter are sweeps: the compactor, retention and GC each
+  visit every table in one long-lived process, and an exit from one table's
+  read used to take the whole sweep down and, for GC, its grace-period
+  bookkeeping with it. The abandoned statement keeps running on the
+  connection either way; what changes is that the caller keeps its state.
+
   `ducklake_merge_adjacent_files/2` must never be called on a smolquery table:
   over externally-registered files it crashes DuckDB fatally (ducklake
   `67480b1d`, format 0.4), and a fatal error invalidates the whole database.
@@ -426,7 +439,7 @@ defmodule Smolquery.Catalog.DuckLake do
         "INSERT INTO #{materialized_table(config.catalog)} VALUES (#{Enum.join(values, ", ")})"
       end)
 
-    Engine.transaction(config.engine, statements)
+    Engine.try_transaction(config.engine, statements)
   end
 
   @impl Catalog
@@ -676,7 +689,7 @@ defmodule Smolquery.Catalog.DuckLake do
   defp swap(config, ref, add, drop) do
     with {:ok, name} <- table_name(config, ref),
          :ok <-
-           Engine.transaction(
+           Engine.try_transaction(
              config.engine,
              [delete_statement(name, drop), add_statement(config, ref, add)],
              config.swap_timeout_ms
@@ -815,7 +828,7 @@ defmodule Smolquery.Catalog.DuckLake do
          :ok <- maybe_ensure_retention_table(config, options) do
       case option_statements(config, dataset, table, options) do
         [] -> :ok
-        statements -> Engine.transaction(config.engine, statements)
+        statements -> Engine.try_transaction(config.engine, statements)
       end
     end
   end
@@ -970,7 +983,7 @@ defmodule Smolquery.Catalog.DuckLake do
 
   defp transact(config, statements) do
     with_commit_retries(fn ->
-      case Engine.transaction(config.engine, statements) do
+      case Engine.try_transaction(config.engine, statements) do
         :ok -> {:ok, :committed}
         {:error, _error} = failure -> failure
       end
@@ -1123,7 +1136,7 @@ defmodule Smolquery.Catalog.DuckLake do
     now = System.system_time(:millisecond)
     created_at = connection.created_at || now
 
-    Engine.transaction(config.engine, [
+    Engine.try_transaction(config.engine, [
       delete_connection_sql(config, connection.name),
       "INSERT INTO #{connections_table(config.catalog)} " <>
         "(name, host, port, database_name, username, secret, sslmode, created_at, updated_at) " <>
@@ -1353,7 +1366,7 @@ defmodule Smolquery.Catalog.DuckLake do
   end
 
   defp query(config, sql, params \\ [], timeout \\ 30_000),
-    do: Engine.query(config.engine, sql, params, timeout)
+    do: Engine.try_query(config.engine, sql, params, timeout)
 
   defp engine_extensions do
     :smolquery

lib/smolquery/engine.ex

@@ -234,7 +234,24 @@ defmodule Smolquery.Engine do
   @spec try_query(handle(), String.t(), [term()], timeout()) ::
           {:ok, Result.t()} | {:error, Exception.t()}
   def try_query(handle, sql, params \\ [], timeout \\ 30_000) do
-    query(handle, sql, params, timeout)
+    catching_exit(fn -> query(handle, sql, params, timeout) end)
+  end
+
+  @doc """
+  Same as `transaction/3`, but an exit from the call comes back as an error —
+  `try_query/4`'s contract for a transaction (T-464).
+
+  A transaction whose call exits is still running on the connection, and
+  commits or rolls back on its own time; the caller learns only that it did
+  not hear the answer.
+  """
+  @spec try_transaction(handle(), [String.t()], timeout()) :: :ok | {:error, Exception.t()}
+  def try_transaction(handle, statements, timeout \\ 30_000) do
+    catching_exit(fn -> transaction(handle, statements, timeout) end)
+  end
+
+  defp catching_exit(call) do
+    call.()
   catch
     :exit, reason -> {:error, CallExited.new(reason)}
   end

lib/smolquery/storage_service/compactor.ex

@@ -183,13 +183,15 @@ defmodule Smolquery.StorageService.Compactor do
   the whole compactor down with it. A crash forgets the backoff, the row caps
   and the quarantine, so the same group was re-planned and re-merged next
   sweep and timed out again, forty-five times in ten hours, while each
-  abandoned transaction kept running on the catalog connection (T-460). Here
-  an exit is caught where it happens: the swap's transaction becomes
-  `{:swap_failed, %CallExited{}}`, any other exit during a table's
-  compaction — a catalog read, or a store put whose HTTP pool died —
-  `{:call_exited, %CallExited{}}`, and a listing that exits fails the sweep.
-  All three back the table off like any other failure and none recycles the
-  compaction engine, whose statement did not exit.
+  abandoned transaction kept running on the catalog connection (T-460). The
+  catalog now answers such an exit as `{:error, %CallExited{}}` itself
+  (`Smolquery.Catalog.DuckLake`, T-464), so a listing that exits fails the
+  sweep with that error and a read or the swap that exits fails its table
+  with it. One catch remains here, for an exit the catalog never sees: a
+  store put whose HTTP pool died mid-upload becomes
+  `{:call_exited, %CallExited{}}`. Both back the table off like any other
+  failure and neither recycles the compaction engine, whose statement did
+  not exit.
 
   The sweep also stops at the first such exit. The call that exited is still
   running on the catalog's compaction connection, which serializes its
@@ -311,7 +313,7 @@ defmodule Smolquery.StorageService.Compactor do
   defp run(state) do
     runtime = state.runtime
 
-    with {:ok, tables} <- catalog_call(:listing_failed, fn -> Catalog.tables(runtime.catalog) end) do
+    with {:ok, tables} <- Catalog.tables(runtime.catalog) do
       {cooling, due} = Enum.split_with(tables, &cooling_down?(state.cooldowns, &1))
       {outcomes, deferred} = sweep_due(runtime, state, due)
       swept = due -- deferred
@@ -758,16 +760,14 @@ defmodule Smolquery.StorageService.Compactor do
     end
   end
 
-  defp call_exited?({:failed, %{reason: {step, %CallExited{}}}})
-       when step in [:swap_failed, :call_exited],
-       do: true
-
+  defp call_exited?({:failed, %{reason: %CallExited{}}}), do: true
+  defp call_exited?({:failed, %{reason: {:call_exited, %CallExited{}}}}), do: true
   defp call_exited?(_outcome), do: false
 
   defp compact_table(runtime, quarantined_groups, table_ref) do
     started_at = System.monotonic_time(:microsecond)
 
-    case catalog_call(:call_exited, fn ->
+    case exit_safe(:call_exited, fn ->
            compact_listed(runtime, quarantined_groups, table_ref, started_at)
          end) do
       {:error, reason} -> failed(runtime, table_ref, reason, started_at)
@@ -799,7 +799,7 @@ defmodule Smolquery.StorageService.Compactor do
     end
   end
 
-  defp catalog_call(step, call) do
+  defp exit_safe(step, call) do
     call.()
   catch
     :exit, reason -> {:error, {step, CallExited.new(reason)}}
@@ -1041,10 +1041,7 @@ defmodule Smolquery.StorageService.Compactor do
   end
 
   defp swapped(runtime, table_ref, segment, paths) do
-    with {:ok, snapshot} <-
-           catalog_call(:swap_failed, fn ->
-             Catalog.replace_segments(runtime.catalog, table_ref, [segment], paths)
-           end),
+    with {:ok, snapshot} <- Catalog.replace_segments(runtime.catalog, table_ref, [segment], paths),
          :ok <- verify_retired(runtime, table_ref, paths) do
       {:ok, snapshot}
     end

lib/smolquery/storage_service/retention.ex

@@ -51,6 +51,7 @@ defmodule Smolquery.StorageService.Retention do
 
   alias Smolquery.Catalog
   alias Smolquery.Engine
+  alias Smolquery.Engine.CallExited
   alias Smolquery.StorageService.Routing
   alias Smolquery.StorageService.Runtime
 
@@ -72,19 +73,23 @@ defmodule Smolquery.StorageService.Retention do
   @doc """
   Sweeps now, without waiting for the interval.
 
-  Reports what was dropped per table, how many snapshots expired, and what
-  failed.
+  Reports what was dropped per table, how many snapshots expired, what
+  failed, and the tables the sweep left untouched behind a catalog call that
+  exited (`deferred`, T-464): the call that exited is still running on the
+  catalog connection, so every later table would queue behind it and fail
+  for a reason not its own.
   """
   @spec sweep(atom(), timeout()) :: {:ok, map()} | {:error, term()}
   def sweep(name, timeout \\ 60_000), do: GenServer.call(Runtime.retention(name), :sweep, timeout)
 
   defp run(state) do
     with {:ok, tables} <- Catalog.tables(state.runtime.catalog) do
-      outcomes = Enum.map(tables, &sweep_table(state.runtime, &1))
+      {outcomes, deferred} = sweep_tables(state.runtime, tables)
 
       report = %{
         dropped: for({:ok, drop} <- outcomes, do: drop),
         failed: for({:failed, failure} <- outcomes, do: failure),
+        deferred: deferred,
         expired_snapshots: expire(state.runtime)
       }
 
@@ -101,6 +106,24 @@ defmodule Smolquery.StorageService.Retention do
     end
   end
 
+  defp sweep_tables(_runtime, []), do: {[], []}
+
+  defp sweep_tables(runtime, [table_ref | rest]) do
+    case sweep_table(runtime, table_ref) do
+      {:failed, %{reason: %CallExited{}}} = outcome ->
+        Logger.warning(fn ->
+          "retention sweep stopped after a catalog call exited on #{inspect(table_ref)}: " <>
+            "#{length(rest)} table(s) deferred to the next sweep"
+        end)
+
+        {[outcome], rest}
+
+      outcome ->
+        {outcomes, deferred} = sweep_tables(runtime, rest)
+        {[outcome | outcomes], deferred}
+    end
+  end
+
   defp sweep_table(runtime, table_ref) do
     with true <- runtime.name |> Routing.resolve() |> Routing.own?(table_ref),
          {:ok, %{column: column, ttl_ms: ttl_ms}} <-

lib/smolquery_pg/pg_catalog.ex

@@ -50,6 +50,13 @@ defmodule SmolqueryPg.PgCatalog do
   every call runs through this server, so a refresh never interleaves with
   a read.
 
+  A rebuild that cannot read the catalog — a call that timed out behind a
+  busy connection, or found it gone, answers `{:error, %CallExited{}}`
+  since T-464 — is not a rebuild: the query answers
+  `{:error, {:pg_catalog_unavailable, _}}`, the generated tables keep their
+  last shape, and the next query tries again. An empty `pg_class` would be
+  a wrong answer a client acts on, not a failure it can retry.
+
   All calls take the edge's instance name; the server and its engine derive
   from it (`SmolqueryPg.Runtime.pg_catalog/1`).
   """
@@ -59,6 +66,7 @@ defmodule SmolqueryPg.PgCatalog do
   alias Explorer.DataFrame
   alias Smolquery.Catalog
   alias Smolquery.Engine
+  alias Smolquery.Engine.CallExited
   alias Smolquery.Engine.Frame
   alias Smolquery.Identifier
   alias SmolqueryPg.PgCatalog.Rewrite
@@ -174,9 +182,8 @@ defmodule SmolqueryPg.PgCatalog do
   end
 
   def handle_call({:query, sql, settings, params}, _from, state) do
-    state = ensure_fresh(state)
-
-    with {:ok, _ast, canonical} <- serialize(state.engine, Rewrite.pre(sql, settings)),
+    with {:ok, state} <- ensure_fresh(state),
+         {:ok, _ast, canonical} <- serialize(state.engine, Rewrite.pre(sql, settings)),
          {:ok, frame} <- run(state.engine, Rewrite.post(canonical), params) do
       {:reply, {:ok, columns(frame), rows(frame)}, state}
     else
@@ -289,11 +296,12 @@ defmodule SmolqueryPg.PgCatalog do
     now = System.monotonic_time(:millisecond)
 
     if now - state.refreshed_at > @refresh_ttl_ms or state.refreshed_at == 0 do
-      refresh(state.engine, state.runtime.catalog)
-
-      %{state | refreshed_at: now}
+      case refresh(state.engine, state.runtime.catalog) do
+        :ok -> {:ok, %{state | refreshed_at: now}}
+        {:error, reason} -> {:error, {:pg_catalog_unavailable, reason}}
+      end
     else
-      state
+      {:ok, state}
     end
   end
 
@@ -593,7 +601,12 @@ defmodule SmolqueryPg.PgCatalog do
   defp refresh(_engine, nil), do: :ok
 
   defp refresh(engine, catalog) do
-    tables = listed_tables(catalog)
+    with {:ok, tables} <- listed_tables(catalog) do
+      rebuild(engine, tables)
+    end
+  end
+
+  defp rebuild(engine, tables) do
     datasets = tables |> Enum.map(fn {dataset, _table, _schema} -> dataset end) |> Enum.uniq()
 
     Engine.transaction(engine, [
@@ -674,16 +687,24 @@ defmodule SmolqueryPg.PgCatalog do
   end
 
   defp listed_tables(catalog) do
-    case Catalog.tables(catalog) do
-      {:ok, refs} -> Enum.flat_map(refs, &table_entry(catalog, &1))
-      {:error, _reason} -> []
+    with {:ok, refs} <- Catalog.tables(catalog) do
+      Enum.reduce_while(refs, {:ok, []}, &collect_entry(catalog, &1, &2))
+    end
+  end
+
+  defp collect_entry(catalog, ref, {:ok, entries}) do
+    case table_entry(catalog, ref) do
+      {:ok, entry} -> {:cont, {:ok, [entry | entries]}}
+      :skip -> {:cont, {:ok, entries}}
+      {:error, reason} -> {:halt, {:error, reason}}
     end
   end
 
   defp table_entry(catalog, {dataset, table} = ref) do
     case Catalog.table_schema(catalog, ref) do
-      {:ok, schema} -> [{dataset, table, schema}]
-      {:error, _reason} -> []
+      {:ok, schema} -> {:ok, {dataset, table, schema}}
+      {:error, %CallExited{} = exited} -> {:error, exited}
+      {:error, _dropped_meanwhile} -> :skip
     end
   end
 

test/smolquery/catalog/ducklake_test.exs

@@ -544,6 +544,18 @@ defmodule Smolquery.Catalog.DuckLakeTest do
     end
   end
 
+  describe "a call that exits (T-464)" do
+    test "comes back as {:error, %CallExited{}} from a read and from a commit" do
+      catalog = DuckLake.new(engine: @engine)
+      Process.unregister(Engine.connection_name(@engine))
+
+      assert Catalog.tables(catalog) == {:error, %Smolquery.Engine.CallExited{reason: :noproc}}
+
+      assert Catalog.drop_segments(catalog, @table, ["/nowhere.parquet"]) ==
+               {:error, %Smolquery.Engine.CallExited{reason: :noproc}}
+    end
+  end
+
   describe "replace_segments/4" do
     test "one snapshot both adds the replacement and retires the inputs", %{
       catalog: catalog,

test/smolquery/engine_test.exs

@@ -189,6 +189,28 @@ defmodule Smolquery.EngineTest do
     end
   end
 
+  describe "try_transaction/3" do
+    test "commits like transaction/3 and returns its errors" do
+      assert Engine.try_transaction(@engine, ["CREATE TABLE try_txn (n INTEGER)"]) == :ok
+
+      assert {:error, %Adbc.Error{}} =
+               Engine.try_transaction(@engine, ["SELECT * FROM no_such_table"])
+    end
+
+    test "returns CallExited instead of exiting when the connection is down or busy (T-464)" do
+      assert {:error, %CallExited{reason: :noproc}} =
+               Engine.try_transaction(__MODULE__.Missing, ["SELECT 1"])
+
+      busy = Process.whereis(Engine.connection_name(@engine))
+      :ok = :sys.suspend(busy)
+
+      assert {:error, %CallExited{reason: :timeout}} =
+               Engine.try_transaction(@engine, ["SELECT 1"], 50)
+
+      :ok = :sys.resume(busy)
+    end
+  end
+
   describe "transaction/2" do
     test "commits every statement together" do
       assert Engine.transaction(@engine, [