fix: keep ETS storage tables supervised

Route `Jido.Storage.ETS` table creation through a supervised owner process and give fallback-created tables a supervisor heir so short-lived callers do not drop storage tables.

Fixes #302.

Author: mikehostetler

lib/jido/application.ex

@@ -6,7 +6,9 @@ defmodule Jido.Application do
   def start(_type, _args) do
     Jido.Telemetry.setup()
 
-    children = []
+    children = [
+      Jido.Storage.ETS.Owner
+    ]
 
     register_signal_extensions()
     Jido.Discovery.init_async()

lib/jido/storage/ets.ex

@@ -218,16 +218,31 @@ defmodule Jido.Storage.ETS do
     :"#{base}_thread_meta"
   end
 
-  defp ensure_tables(opts) do
+  @doc false
+  @spec create_tables(opts()) :: :ok
+  def create_tables(opts) do
     ensure_table(checkpoint_table(opts), [:set])
     ensure_table(threads_table(opts), [:ordered_set])
     ensure_table(meta_table(opts), [:set])
   end
 
+  defp ensure_tables(opts) do
+    case Jido.Storage.ETS.Owner.ensure_tables(opts) do
+      :ok -> :ok
+      {:error, _reason} -> create_tables(opts)
+    end
+  end
+
   defp ensure_table(name, extra_opts) do
     case :ets.whereis(name) do
       :undefined ->
-        :ets.new(name, [:named_table, :public, read_concurrency: true] ++ extra_opts)
+        _ =
+          :ets.new(
+            name,
+            [:named_table, :public, read_concurrency: true] ++ heir_opts(name) ++ extra_opts
+          )
+
+        :ok
 
       _ref ->
         :ok
@@ -236,6 +251,16 @@ defmodule Jido.Storage.ETS do
     ArgumentError -> :ok
   end
 
+  defp heir_opts(name) do
+    case Process.whereis(Jido.Supervisor) do
+      pid when is_pid(pid) and pid != self() ->
+        [{:heir, pid, {:jido_storage_ets, name}}]
+
+      _other ->
+        []
+    end
+  end
+
   defp get_current_rev(table, thread_id) do
     case :ets.select_reverse(table, [{{{thread_id, :"$1"}, :_}, [], [:"$1"]}], 1) do
       {[seq], _cont} -> seq + 1

lib/jido/storage/ets/owner.ex

@@ -0,0 +1,47 @@
+defmodule Jido.Storage.ETS.Owner do
+  @moduledoc false
+
+  use GenServer
+
+  @name __MODULE__
+  @call_timeout 5_000
+
+  @doc false
+  @spec child_spec(keyword()) :: Supervisor.child_spec()
+  def child_spec(opts) do
+    %{
+      id: @name,
+      start: {@name, :start_link, [opts]},
+      type: :worker,
+      restart: :permanent,
+      shutdown: 5_000
+    }
+  end
+
+  @doc false
+  @spec start_link(keyword()) :: GenServer.on_start()
+  def start_link(opts \\ []) do
+    GenServer.start_link(__MODULE__, opts, name: @name)
+  end
+
+  @doc false
+  @spec ensure_tables(keyword()) :: :ok | {:error, term()}
+  def ensure_tables(opts) when is_list(opts) do
+    case GenServer.whereis(@name) do
+      nil -> {:error, :not_started}
+      pid -> GenServer.call(pid, {:ensure_tables, opts}, @call_timeout)
+    end
+  catch
+    :exit, reason -> {:error, {:owner_unavailable, reason}}
+  end
+
+  @impl true
+  def init(_opts) do
+    {:ok, %{}}
+  end
+
+  @impl true
+  def handle_call({:ensure_tables, opts}, _from, state) do
+    {:reply, Jido.Storage.ETS.create_tables(opts), state}
+  end
+end

test/jido/storage/ets_test.exs

@@ -320,7 +320,7 @@ defmodule JidoTest.Storage.ETSTest do
       thread_id = "thread_#{System.unique_integer([:positive])}"
       total_appends = 40
 
-      # Ensure tables are created/owned by the test process.
+      # Ensure tables are created before concurrent appends start.
       assert {:ok, _} = ETS.append_thread(thread_id, [%{kind: :note, payload: %{n: 0}}], opts)
 
       results =
@@ -400,4 +400,51 @@ defmodule JidoTest.Storage.ETSTest do
       assert {:ok, %Thread{}} = ETS.load_thread("key1", opts)
     end
   end
+
+  describe "table ownership" do
+    test "tables survive when first used by a short-lived process" do
+      opts = [table: unique_table(:ephemeral_owner)]
+      table = :"#{Keyword.fetch!(opts, :table)}_checkpoints"
+      owner = Process.whereis(Jido.Storage.ETS.Owner)
+
+      assert is_pid(owner)
+
+      parent = self()
+
+      {caller, ref} =
+        spawn_monitor(fn ->
+          assert :ok = ETS.put_checkpoint(:ephemeral_key, :value, opts)
+          send(parent, {:owner, :ets.info(table, :owner), self()})
+        end)
+
+      assert_receive {:owner, ^owner, ^caller}
+      assert_receive {:DOWN, ^ref, :process, ^caller, _reason}
+
+      assert :ets.whereis(table) != :undefined
+      assert {:ok, :value} = ETS.get_checkpoint(:ephemeral_key, opts)
+    end
+
+    test "fallback-created tables survive a short-lived creator" do
+      opts = [table: unique_table(:fallback_ephemeral_owner)]
+      table = :"#{Keyword.fetch!(opts, :table)}_checkpoints"
+      supervisor = Process.whereis(Jido.Supervisor)
+
+      assert is_pid(supervisor)
+
+      parent = self()
+
+      {caller, ref} =
+        spawn_monitor(fn ->
+          assert :ok = ETS.create_tables(opts)
+          true = :ets.insert(table, {:ephemeral_key, :value})
+          send(parent, {:owner, :ets.info(table, :owner), self()})
+        end)
+
+      assert_receive {:owner, ^caller, ^caller}
+      assert_receive {:DOWN, ^ref, :process, ^caller, _reason}
+
+      assert :ets.info(table, :owner) == supervisor
+      assert {:ok, :value} = ETS.get_checkpoint(:ephemeral_key, opts)
+    end
+  end
 end