Deterministic cleanup for collaborative editing documents

.claude/agents/phoenix-elixir-expert.md

@@ -29,6 +29,15 @@ You are a **battle-tested Elixir/Phoenix architect** with deep expertise in the
 
 - For transaction and prelim-type rules when touching y_ex from Elixir, see `.claude/guidelines/yex-guidelines.md §Transaction Deadlock Rules` and `§Prelim Types`.
 
+## OTP / supervision trees
+
+- When writing or reviewing a supervisor, GenServer, or named process, see
+  `.claude/guidelines/testable-supervision-trees.md §0. The principle` and
+  `§4. Anti-patterns checklist`. Names and collaborators are *parameters*, not
+  constants — don't bake `name: __MODULE__` into a process or resolve
+  dependencies from global state, or you force the suite serial (breaks
+  `async: true`).
+
 ## Lightning Project Context
 
 **Architecture Awareness:**

CHANGELOG.md

@@ -24,6 +24,13 @@ and this project adheres to
   `WorkOrder.active_states/0` and replacing all hardcoded state lists across the
   codebase
   [#4589](https://github.com/OpenFn/lightning/issues/4589)
+- Collaborative editing documents now shut down deterministically. A document
+  tree can be handed an `owner` process to monitor, and when that owner exits it
+  stops cleanly with a final persistence flush; `Lightning.Collaborate` gains a
+  synchronous, idempotent `stop_document/1`. Production behaviour is unchanged
+  (documents started by a LiveView still outlive it), but tests can now bind a
+  document's lifetime to the test that starts it, fixing intermittent failures
+  caused by document processes leaking between runs.
 
 ### Fixed
 

CLAUDE.md

@@ -132,6 +132,11 @@ in development.
 - Use `{}` brace syntax in HEEx templates
 - `warnings_as_errors: true` - code must compile without warnings
 
+> Before writing or reviewing any supervisor, GenServer, or named process, see
+> `.claude/guidelines/testable-supervision-trees.md`. Don't bake `name:
+> __MODULE__` into a process or resolve collaborators from global state — it
+> forces the test suite serial.
+
 ### React/TypeScript
 
 - Props from LiveView are **underscore_cased** (not camelCase)
@@ -209,6 +214,7 @@ Detailed guidelines in `.claude/guidelines/`:
 - `testing-essentials.md` - Unit testing patterns and anti-patterns
 - `e2e-testing.md` - Playwright E2E testing
 - `yex-guidelines.md` - Critical Yex (Yjs/Elixir) usage rules
+- `testable-supervision-trees.md` - OTP processes addressable in tests (async), nameless in prod
 - `toast-notifications.md` - Notification patterns
 - `logging.md` - Logger level conventions and Sentry noise (info/warning vs error)
 - `ui-patterns.md` - Button variants, disabled states, Tailwind conventions

config/test.exs

@@ -165,3 +165,18 @@ config :lightning, :github_app,
 config :lightning, LightningWeb.CollectionsController,
   default_stream_limit: 25,
   max_database_limit: 15
+
+# Under test, collaboration document children are spawned by an internal
+# GenServer rather than the test process. When a test owns the database
+# connection (and any per-test mocks), those children need to be granted access
+# explicitly. This callback runs synchronously as each document tree starts up,
+# so the children can talk to the database and resolve mocks via the owning test
+# process. Outside the test environment this config is absent and the supervisor
+# falls back to a no-op.
+config :lightning,
+       :collaboration_process_allow,
+       fn owner, pid ->
+         Ecto.Adapters.SQL.Sandbox.allow(Lightning.Repo, owner, pid)
+         Mox.allow(LightningMock, owner, pid)
+         :ok
+       end

lib/lightning/collaboration.ex

@@ -26,16 +26,16 @@ defmodule Lightning.Collaborate do
       Collaborate.start(user: user, workflow: workflow)
   """
   alias Lightning.Collaboration.DocumentSupervisor
+  alias Lightning.Collaboration.Instance
   alias Lightning.Collaboration.Registry
   alias Lightning.Collaboration.Session
   alias Lightning.Collaboration.Supervisor, as: SessionSupervisor
 
   require Logger
 
-  @pg_scope :workflow_collaboration
-
-  @spec start(opts :: Keyword.t()) :: GenServer.on_start()
-  def start(opts) do
+  @spec start(instance :: Instance.t(), opts :: Keyword.t()) ::
+          GenServer.on_start()
+  def start(instance \\ Instance.default(), opts) do
     session_id = Ecto.UUID.generate()
     parent_pid = Keyword.get(opts, :parent_pid, self())
 
@@ -52,45 +52,160 @@ defmodule Lightning.Collaborate do
       "Starting collaboration for document: #{document_name} (workflow: #{workflow.id})"
     )
 
-    # Ensure document supervisor exists for this document
-    case lookup_shared_doc(document_name) do
-      nil ->
-        Logger.info("Starting document for #{document_name}")
-        {:ok, _doc_supervisor_pid} = start_document(workflow, document_name)
+    # Ensure document supervisor exists for this document. Track whether THIS
+    # call started the document, so we only tear down a doc we orphaned.
+    started_here? =
+      case lookup_shared_doc(instance, document_name) do
+        nil ->
+          Logger.info("Starting document for #{document_name}")
+
+          {:ok, _doc_supervisor_pid} =
+            start_document(instance, workflow, document_name)
+
+          true
 
-      _shared_doc_pid ->
-        Logger.info("Found existing document for #{document_name}")
+        _shared_doc_pid ->
+          Logger.info("Found existing document for #{document_name}")
+          false
+      end
+
+    # Start session for this user
+    result =
+      SessionSupervisor.start_child(
+        instance.dynamic_supervisor,
+        {
+          Session,
+          workflow: workflow,
+          user: user,
+          parent_pid: parent_pid,
+          document_name: document_name,
+          registry: instance.registry,
+          pg_scope: instance.pg_scope,
+          name:
+            Registry.via(
+              instance.registry,
+              {:session, "#{document_name}:#{session_id}", user.id}
+            )
+        }
+      )
+
+    case result do
+      {:ok, _session_pid} ->
+        result
+
+      _error ->
+        if started_here?, do: stop_document(instance, document_name)
+        result
+    end
+  end
+
+  @doc """
+  Deterministically stops the collaborative document for `document_name`.
+
+  Tears down its DocumentSupervisor, SharedDoc and PersistenceWriter (with a
+  final flush). Synchronous and idempotent: returns `:ok` whether or not a
+  document is running. The symmetric partner to `start_document/2`.
+  """
+  @spec stop_document(instance :: Instance.t(), document_name :: String.t()) ::
+          :ok
+  def stop_document(instance \\ Instance.default(), document_name) do
+    case Registry.whereis(instance.registry, {:doc_supervisor, document_name}) do
+      nil ->
         :ok
+
+      pid ->
+        try do
+          DocumentSupervisor.stop(pid)
+          :ok
+        catch
+          :exit, _ -> :ok
+        end
     end
+  end
 
-    # Start session for this user
-    SessionSupervisor.start_child({
-      Session,
-      workflow: workflow,
-      user: user,
-      parent_pid: parent_pid,
-      document_name: document_name,
-      name: Registry.via({:session, "#{document_name}:#{session_id}", user.id})
-    })
+  @doc """
+  Starts the collaborative document tree for `document_name`.
+
+  `document_name` is a positional payload (domain identity); the registered name
+  and the optional `owner` are process configuration and live in trailing `opts`,
+  per `.claude/guidelines/testable-supervision-trees.md` §1.
+
+  ## Options
+
+  - `:owner` — a pid the document tree monitors. When that pid goes `:DOWN`, the
+    tree stops `:normal` (running its flush via `terminate/2`; `:transient` means
+    no restart). Lets any caller — a test, a request — get deterministic cleanup
+    by passing `owner: self()`, with no wrapper. Defaults to `nil` (no monitor),
+    so production documents outlive the LiveView that starts them.
+  """
+  @spec start_document(
+          workflow :: Lightning.Workflows.Workflow.t(),
+          document_name :: String.t()
+        ) :: {:ok, pid()}
+  @spec start_document(
+          workflow :: Lightning.Workflows.Workflow.t(),
+          document_name :: String.t(),
+          opts :: Keyword.t()
+        ) :: {:ok, pid()}
+  @spec start_document(
+          instance :: Instance.t(),
+          workflow :: Lightning.Workflows.Workflow.t(),
+          document_name :: String.t(),
+          opts :: Keyword.t()
+        ) :: {:ok, pid()}
+  def start_document(%Lightning.Workflows.Workflow{} = workflow, document_name) do
+    start_document(Instance.default(), workflow, document_name, [])
+  end
+
+  def start_document(%Instance{} = instance, workflow, document_name) do
+    start_document(instance, workflow, document_name, [])
+  end
+
+  def start_document(
+        %Lightning.Workflows.Workflow{} = workflow,
+        document_name,
+        opts
+      )
+      when is_list(opts) do
+    start_document(Instance.default(), workflow, document_name, opts)
   end
 
   def start_document(
+        %Instance{} = instance,
         %Lightning.Workflows.Workflow{} = workflow,
-        document_name
-      ) do
+        document_name,
+        opts
+      )
+      when is_list(opts) do
+    doc_opts =
+      [
+        workflow: workflow,
+        document_name: document_name,
+        owner: Keyword.get(opts, :owner),
+        registry: instance.registry,
+        pg_scope: instance.pg_scope,
+        name: Registry.via(instance.registry, {:doc_supervisor, document_name})
+      ]
+      |> then(fn base ->
+        # Only forward auto_exit when a caller supplied it, so the production
+        # default (DocumentSupervisor's own `auto_exit: true`) is preserved.
+        case Keyword.fetch(opts, :auto_exit) do
+          {:ok, auto_exit} -> Keyword.put(base, :auto_exit, auto_exit)
+          :error -> base
+        end
+      end)
+
     case SessionSupervisor.start_child(
-           {DocumentSupervisor,
-            workflow: workflow,
-            document_name: document_name,
-            name: Registry.via({:doc_supervisor, document_name})}
+           instance.dynamic_supervisor,
+           {DocumentSupervisor, doc_opts}
          ) do
       {:ok, pid} -> {:ok, pid}
       {:error, {:already_started, pid}} -> {:ok, pid}
     end
   end
 
-  defp lookup_shared_doc(document_name) do
-    case :pg.get_members(@pg_scope, document_name) do
+  defp lookup_shared_doc(%Instance{} = instance, document_name) do
+    case :pg.get_members(instance.pg_scope, document_name) do
       [] -> nil
       [shared_doc_pid | _] -> shared_doc_pid
     end