diff --git a/.claude/agents/phoenix-elixir-expert.md b/.claude/agents/phoenix-elixir-expert.md index f342a5bb02c..d32d4e8e205 100644 --- a/.claude/agents/phoenix-elixir-expert.md +++ b/.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:** diff --git a/.claude/guidelines/testable-supervision-trees.md b/.claude/guidelines/testable-supervision-trees.md new file mode 100644 index 00000000000..24769af7538 --- /dev/null +++ b/.claude/guidelines/testable-supervision-trees.md @@ -0,0 +1,505 @@ +--- +type: reference +status: active +date: 2026-05-19 +related: + - "[[elixir]]" +tags: + - elixir + - otp + - testing +--- + +# Testable Supervision Trees & Named Processes + +Guidance for building supervision trees, supervisors, GenServers and named +processes that are **uniquely addressable and isolated in tests** (so the suite +runs `async: true`) but **need no name in normal production use**. + +This exists because AI harnesses (and people in a hurry) reliably reach for the +shortcut — `name: __MODULE__` baked into the process, dependencies fished out +of global state — which works in dev, looks idiomatic, and quietly forces the +whole test suite serial. Point Claude at this doc when generating or reviewing +OTP code. + +--- + +## 0. The principle + +**A process's name, and its dependencies, are *parameters* — not constants.** + +Every failure mode here traces to one shortcut: hardcoding the name inside the +process, or resolving a collaborator from global state at call time, instead of +threading it through structure (supervision wiring, `start_link` opts, process +state, the caller signature). + +Per Gray & Tate's *Designing Elixir Systems with OTP*: push logic into a pure +functional core that needs no processes to test, so the GenServer is a thin +shell. Everything below is damage control for the thin shell that remains. + +> **Scope guard.** Often the cleanest fix is that a thing never needed to be a +> process or a stored value at all. That's worth one sentence at the design +> review, then move on. The objective of this document is the **caller +> signature and how information reaches child processes** — not relitigating +> whether something should be a GenServer. + +### One ownership seam, three payoffs + +Name-isolation (§1), deterministic teardown (§3), and async-safety (§2) are not +three problems — they are three payoffs of **one** seam. The root control is an +**injectable owning identity**: the process / atom / pid / name that owns the +thing. Make that a parameter and all three follow: + +- **Name-isolation** — an anonymous or per-test name means no + `{:already_started, _}` clash, so the suite runs `async: true`. +- **Deterministic teardown** — bind the owner to the test (it holds the pid, or + the process monitors a chosen owner) and the thing dies with its owner — no + manual cleanup. +- **Async-safety** — that same injected owner is the pid you scope `Mox.allow` + to (§2, Axis 2). Where the hop graph is deep you add the `Mox.allow` + strategies, but the seam is the same one. + +Lose the seam and you fight all three separately: a hardcoded name forces the +suite serial; a process with no symmetric `stop` and no owner to monitor leaks +past its test; a dependency fished from a global has no owner to lend a mock to. +The collaboration fix (§3) is the cautionary case — it solved *lifetime* in +test support rather than at the seam, because the production API has no owner +option yet. + +--- + +## 1. The 101 case: fixed children, no Registry + +A `Registry` is a lookup table from a **domain key → pid**. You need one only +when *all three* hold: + +1. an **arbitrary / open-ended** number of the process exists, +2. keyed by **runtime data** (a workflow id, a session id), and +3. the code that must talk to one **does not already hold its pid**. + +A supervisor with a **fixed, known set of children** — one of each — fails all +three. It needs nothing. The realisation: + +> For a constant set of children, **the registered module-atom name *is* your +> registry, and it's free.** A registered name already survives restarts — the +> supervisor brings the child back and it re-registers the same atom. That is +> the one feature people reach to `Registry` for. + +The whole 101 pattern, no Registry, fully async-isolatable: + +```elixir +defmodule MyApp.Cache do + use GenServer + + # name is an OPTION, defaulted — never hardcoded inside the module + def start_link(opts) do + {name, opts} = Keyword.pop(opts, :name, __MODULE__) + GenServer.start_link(__MODULE__, opts, name: name) + end + + # API takes the server ref FIRST, defaulted to the singleton + def fetch(server \\ __MODULE__, key), do: GenServer.call(server, {:fetch, key}) + + @impl true + def init(opts) do + # dependencies are injected, with prod defaults — never read from a global + {:ok, %{store: %{}, http: Keyword.get(opts, :http, MyApp.HTTP)}} + end +end +``` + +Production (in the app supervisor's fixed child list) gets the singleton for +free; callers write `MyApp.Cache.fetch(key)` and never name anything. The test +never *looks anything up* — it **holds the pid it just started**: + +```elixir +test "expires entries" do + pid = start_supervised!({MyApp.Cache, name: nil, http: HTTPMock}) + assert MyApp.Cache.fetch(pid, :missing) == nil +end +``` + +`name: nil` is the trick: it starts an anonymous, isolated instance even when +the app already booted a global `MyApp.Cache`, so there is no +`{:already_started, _}` clash, no need to gut `application.ex` in test config, +and the test stays `async: true`. (Only reach for config-driven "don't boot it +in `:test`" when something you *cannot* hand a pid — a Plug, a distant caller — +calls the API with the default name. That is the exception, not the default.) + +### Argument order convention: server ref first, with a default + +This is the OTP-wide convention and it is near-universal: +`GenServer.call(server, …)`, `Agent.get(agent, …)`, +`Registry.lookup(registry, …)`, `Phoenix.PubSub.broadcast(pubsub, …)`, +`Oban.insert(name \\ Oban, changeset)`, `Mox.allow(mock, …)`, +`Ecto.Adapters.SQL.Sandbox.allow(repo, …)` — the addressed thing is the +**subject**, so it leads. + +Critically, **first-with-default is the idiom that produces the `/1` + `/2` +pair cleanly**: + +```elixir +def fetch(server \\ __MODULE__, key) +# fetch(key) -> arity 1, server defaults to __MODULE__ (production) +# fetch(pid, key) -> arity 2, explicit instance (test) +``` + +Production never names anything; the test injects a pid through the *same* +function. **The signature is the injection seam.** That is why this convention +is load-bearing, not stylistic. + +The one principled exception: **pipeline-first APIs put the instance last** — +`Finch.build(:get, url) |> Finch.request(MyFinch)` — because the data being +transformed is the subject and the instance is configuration, so it reads in a +pipe. Rule of thumb: *callers piping data through it → instance last; callers +addressing a process → instance first (the common case).* + +### Constructor sub-rule: `start_link`-style functions put the name in trailing opts + +"Subject leads" governs functions that **address an already-running process**. +It does *not* govern **constructors** — `start_link` / `start_child` and +friends — and the constructor convention pulls the other way, just as +near-universally: the name goes in a **trailing `opts` keyword list**, never as +a leading positional. + +```elixir +GenServer.start_link(module, init_arg, opts) # name: in opts +Supervisor.start_link(children, opts) # name: in opts +Agent.start_link(fun, opts) # name: in opts +Registry.start_link(opts) # name: in opts +Oban.start_link(opts) / Finch.start_link(opts) # name: in opts +``` + +Every `start_link` in the standard library, and across Oban, Phoenix.PubSub and +Finch, puts `name:` in opts. So the two rules never actually collide — they +govern **mutually exclusive function shapes**: + +- a function that **creates** a process → name in trailing `opts` + (`def start_link(arg, opts \\ [])`); +- a function that **operates on** an existing one → instance leads positionally + (`def fetch(server \\ __MODULE__, key)`). + +You never call `start_link` on a running process, and you never need a +name-in-opts on `call` / `lookup`. Identity-vs-configuration doesn't change +this: whether the name is intrinsic identity (a `Registry` key) or an instance +selector (which `Oban`), the constructor still takes it in opts; only the call +side flips — `Oban.insert(name \\ Oban, changeset)` leads with the instance. + +One nuance worth stating, because it is the thing that confuses people: the +**process-registration name** (the `:via` tuple, the registered atom, the +`name:` opt) is what goes in trailing opts. **Domain data that happens to +identify the thing** (a workflow id, a `document_name`) is just an `init_arg` +payload and stays positional. `Lightning.Collaborate.start_document/2` already +gets this right — `document_name` is a positional payload, while the registered +name lives in opts inside the child spec: + +```elixir +def start_document(%Workflow{} = workflow, document_name) do + SessionSupervisor.start_child( + {DocumentSupervisor, + workflow: workflow, + document_name: document_name, + name: Registry.via({:doc_supervisor, document_name})} # name → opts + ) +end +``` + +So a new document entrypoint that also takes an **owner** (for owner-monitored +cleanup, §3) resolves cleanly: the owner is process configuration, so it joins +the registration name in opts — `start_document(workflow, document_name, opts \\ [])` +with `owner:` in `opts` — *not* threaded as a leading positional. Its +call/lookup partners (`stop_document/1`, `whereis/1`) keep the subject first, +per the main rule. + +--- + +## 2. The two axes (the centrepiece) + +People — and AI harnesses — reliably weld together two **independent** problems. +Welding them is what makes refactors thrash. Keep them apart: + +1. **Axis 1 — where does per-instance config / identity live?** +2. **Axis 2 — which process actually invokes the injected dependency?** + +`:persistent_term` (and runtime `Application.put_env`, and naked global ETS) is +the wrong answer to **Axis 1**. The `set_mox_global` / `async: false` / +`Sandbox.allow` pain is entirely **Axis 2**. Fixing Axis 1 does not fix Axis 2 +— but it makes Axis 2 *legible*, which is the precondition for fixing it. + +### `:persistent_term` is a smell + +Treat `:persistent_term`, runtime `Application.put_env`, and naked global ETS +for per-instance config as a **smell requiring explicit justification**. Its +presence means someone reached for *stored state* when *structure* is the +functional answer. The legitimacy litmus: + +> **Does this value ever need to differ between two tests running at the same +> time?** If yes, it cannot live in any global store — inject it. If no, and +> it is genuinely hot-read and fixed at boot, `:persistent_term` is fine +> (Phoenix/Ecto use it internally for exactly that). + +### Worked example: `Lightning.Adaptors.Supervisor` + +The adaptors supervisor is otherwise *exemplary* — its moduledoc states the +principle verbatim, it derives every child name via `Module.concat(name, …)` +(a fixed child set with full multi-instance async isolation and **zero +Registry**), and it injects the `:strategy` as an explicit opt with a +production default. One wart: + +```elixir +# supervisor.ex init/1 — strategy & source are LOCALS here… +strategy = Keyword.get(opts, :strategy, Config.strategy()) +:persistent_term.put(meta_key(name), %{strategy: strategy, source: source_for(strategy)}) +# …then the same init/1 injects cache/tasks/source_topic into child specs +# explicitly, two lines down — but routes strategy/source through a global. +``` + +`Scheduler` then re-fetches it from that global *at call time*: + +```elixir +# scheduler.ex — strategy materialises from nowhere, with no traceable owner +strategy = AdaptorsSupervisor.strategy(state.sup) # :persistent_term.get/1 +strategy.fetch_adaptor(name) +``` + +An investigation of every call site classified this **case (b): avoidable**. +Nothing reaching `Store` lacks the strategy/source at a point where it is +knowable; there is exactly one hardcoded production instance, so even the +stateless facade-from-web path collapses to "boot config for the one instance," +not a dynamic lookup. The generalisable tell: + +> **Global storage smuggling a value past a structural boundary that was +> already open two lines away and already carrying its siblings across.** + +The fix is Axis 1: inject `strategy`/`source` into the child specs the +supervisor is *already building* (it has them in scope), and let `Scheduler` +hold them in its state — exactly as it already does for `source` at `init/1`. + +### Why that makes Axis 2 legible + +Before: you cannot tell *which process* will call `StrategyMock`, because the +value appears from a global with no owner — so you reach for `set_mox_global` +and the suite goes serial. + +After injection: `Scheduler` visibly owns the strategy in its state, so the +mock's caller is obvious and you can scope the allowance: + +```elixir +# Axis 2 recipe — explicit allowance, async-safe +pid = start_supervised!({Lightning.Adaptors.Supervisor, + name: name, strategy: StrategyMock}) +Mox.allow(StrategyMock, self(), Process.whereis(scheduler_name(name))) +``` + +When the pid does not exist yet at setup time (leader election, lazy start), +use the **deferred-resolver form** of `Mox.allow/3` — the trick people forget: + +```elixir +Mox.allow(StrategyMock, self(), fn -> + :global.whereis_name(scheduler_name(name)) +end) +``` + +Mox resolves the pid lazily on first mock invocation, sidestepping the race. +Tasks started via `Task.Supervisor.async/async_nolink` carry `$callers`, so +Mox walks back to the allowed parent automatically; `start_child` +(fire-and-forget) does not propagate and needs its own allowance. + +`Mox.allow/3` and `Ecto.Adapters.SQL.Sandbox.allow/3` are **one concept** — +same signature, same ownership model (a test owns a resource and explicitly +lends it to processes it spawns). Mox's was modelled on Ecto's. + +> **When `set_mox_global` is legitimate, not a cop-out:** explicit allow is the +> default; its cost scales with how many *process hops* the mocked call +> traverses and how *dynamic* those pids are. `set_mox_global` (forcing +> `async: false` for that case) is the correct escape hatch when the hop graph +> is dynamic and deep — `HighlanderPG`-wrapped leader election is the textbook +> case. Try the deferred-resolver form *first*; most cases are not Highlander. + +--- + +## 3. Dynamic populations: when a Registry *is* earned + +When the population is genuinely open-ended and keyed by runtime data, and the +caller does not hold the pid — `DynamicSupervisor` + `Registry` + `:via`. +`Lightning.Collaborate` is the worked example: N sessions/documents keyed by +`document_name`, looked up by a LiveView that did not start them. + +```elixir +SessionSupervisor.start_child( + {Session, workflow: workflow, user: user, + name: Registry.via({:session, "#{document_name}:#{session_id}", user.id})} +) +``` + +Two valid shapes, document the trade-off: + +- **One Registry per top-level instance** (Oban): perfect isolation, but the + instance name threads through every public function + (`Oban.insert(MyApp.Oban, …)`). +- **One global Registry, key-namespaced** (Lightning collaboration): the public + API stays clean (`Collaborate.start/1` takes no name); isolation depends on + keys being genuinely unique. Correct when identity is naturally unique + (workflow + session); risky if a test can reuse a key. + +### The "earned name → config lookup" reference + +If a **stateless synchronous** caller genuinely holds only a name *and* the +population is dynamic, a name → config lookup is justified — and even then it +is **ETS-per-instance, never `:persistent_term`**. `commanded/eventstore` hit +this exact requirement and chose: one named ETS table (`read_concurrency: +true`) owned by a GenServer, holding `{name, pid, ref, store, config}`, the +owner pid monitored so the row **self-deletes on shutdown**. It explicitly +rejected `:persistent_term` because instances churn under async tests and +`:persistent_term` writes/erases trigger a global GC scan. The bonus: an +owner-monitored ETS table needs no manual cleanup function (cf. the adaptors +`forget/1` wart and its "we don't call this automatically because GC is +expensive" comment). + +### Lifetime/ownership: how a dynamic process gets torn down in a test + +§1's teardown story — `start_supervised!` hands the test the pid, ExUnit stops +it on exit — works *because that process is a fixed singleton owned by its +starter*. A §3 process is the opposite by design: started under a global +`DynamicSupervisor`, keyed by runtime data, **meant to outlive the caller** that +started it. ExUnit does not own it, so nothing tears it down when the test ends. +And if it touches a test-scoped resource — a DB write, say — it does so *after* +the test's Ecto-sandbox owner has exited, crashing with `owner ... exited` and +poisoning the next test. This is a real flake the Lightning collaboration suite +hit: a document started via `Collaborate.start/1` is owned by the global +`SessionSupervisor` DynamicSupervisor, outlived its test, and wrote to the DB +after teardown. + +Two recipes, in order of preference. + +**Option 1 (preferred) — owner-monitored self-cleanup on the production API.** +Generalise the eventstore owner-monitor above from a self-deleting config row to +a self-terminating process tree: let the dynamic `start` take an optional +`owner` pid, `Process.monitor/1` it, and stop the process when that owner goes +`:DOWN`. Then *any* caller — a test, a LiveView, a short-lived request — gets +deterministic cleanup for free by passing `owner: self()`, with no special +wrapper. This is the §0 seam landed in the API: one owner parameter buys +lifetime *and* (with a per-test name) async-isolation. + +**Option 2 (what the collaboration fix did) — public symmetric `stop` + a test +helper that binds it.** When the API has no owner option yet, add a +deterministic, idempotent public `stop` (the partner to the dynamic `start`), +then bind it to the test in test support with `on_exit`: + +```elixir +# lib/lightning/collaboration.ex — the symmetric public stop, idempotent +@spec stop_document(document_name :: String.t()) :: :ok +def stop_document(document_name) do + case Registry.whereis({:doc_supervisor, document_name}) do + nil -> + :ok + + pid -> + try do + DocumentSupervisor.stop(pid) + :ok + catch + :exit, _ -> :ok + end + end +end +``` + +```elixir +# lib/lightning/collaboration/document_supervisor.ex — synchronous graceful stop. +# GenServer.stop(:normal) guarantees terminate/2 runs the flush, unlike the +# DynamicSupervisor's :shutdown; :transient restart means a :normal exit is not +# restarted. +def stop(pid, timeout \\ 5_000) when is_pid(pid) do + GenServer.stop(pid, :normal, timeout) +end +``` + +```elixir +# test/support/collaboration_helpers.ex — reconstruct start_supervised's +# lifetime-binding at the test boundary +def start_collaboration_document( + %Lightning.Workflows.Workflow{} = workflow, + document_name + ) + when is_binary(document_name) do + on_exit(fn -> Lightning.Collaborate.stop_document(document_name) end) + Lightning.Collaborate.start_document(workflow, document_name) +end +``` + +The helper *is* the §0 seam, but living in test support rather than the API. It +works, and the symmetric `stop` is worth having regardless — production needs +deterministic teardown too (`Collaborate.start/1` calls `stop_document/1` to +clean up a document it orphaned when a session fails to attach). But every new +call site must *remember* to use the helper; the owner-monitored option (1) +needs no wrapper and protects every caller, which is why it is the fuller fix. +Either way, keep a blanket `stop_all_collaboration_documents/0` `on_exit` net as +belt-and-braces for serial (`async: false`) suites — it catches a single +un-bound leak before it corrupts the next test. + +--- + +## 4. Anti-patterns checklist (point the harness here) + +- ❌ `name: __MODULE__` hardcoded *inside* `start_link` on a worker. → Make + `:name` an option defaulting to `__MODULE__`. +- ❌ Public API calling `GenServer.call(__MODULE__, …)` with no server arg. → + `def fn(server \\ __MODULE__, …)`, server first. +- ❌ Putting the registered name as a leading positional on a constructor + "because the subject leads." → Constructors take `name:` (and `owner:`) in + trailing opts; "subject leads" is the rule for `call` / `lookup` only. Domain + data (a workflow id) can still be a positional payload. +- ❌ A dynamically-supervised process that must outlive its caller, with no + deterministic teardown — it outlives the test and crashes on a post-teardown + resource (e.g. a DB write after the Ecto-sandbox owner has exited). → Give it + an owner-monitored self-cleanup option (preferred), or a public symmetric + `stop` bound to the test via `on_exit`; keep a blanket sweep as + belt-and-braces for serial suites. +- ❌ Dependencies / per-instance config read from `:persistent_term` / + runtime `Application.put_env` / global ETS at call time. → Inject via + `start_link` opts → `init/1` → process state, or thread through the child + spec. Litmus: *does it vary per concurrent test?* +- ❌ Global storage used to pass a value the supervisor already had in scope + one child-spec away. → Inject it into the child spec. +- ❌ `set_mox_global` reached for reflexively to "fix" a boundary. → Try + `Mox.allow(mock, self(), pid)`, then the deferred-resolver form; reserve + global for genuinely dynamic/deep hop graphs. +- ❌ A `Registry` added to a *fixed* child set "to name things." → The + registered atom name already does this, free. +- ❌ Testing GenServer internals via raw `:"$gen_call"` / poking state. → + Test through the public API; test logic in a pure core module. +- ❌ App supervisor that cannot be started piecemeal in tests. → Children + startable independently via `start_supervised!`. + +--- + +## References + +- Gray & Tate, *Designing Elixir Systems with OTP* (Pragmatic Bookshelf) — + functional core / boundary layering. +- Saša Jurić, *Elixir in Action* — process registration as a parameter; `:via`. +- [Oban — instance & DB isolation](https://hexdocs.pm/oban/isolation.html) · + [`Oban.Registry`](https://hexdocs.pm/oban/Oban.Registry.html) +- [`Ecto.Adapters.SQL.Sandbox`](https://hexdocs.pm/ecto_sql/Ecto.Adapters.SQL.Sandbox.html) + — ownership / `allow/3`. +- [Mox docs](https://hexdocs.pm/mox/Mox.html) — `allow/3`, deferred resolver, + `set_mox_from_context`. +- [`commanded/eventstore`](https://github.com/commanded/eventstore) — + `lib/event_store/config/store.ex` (ETS-per-instance, owner-monitored). The + owner-monitor idea generalises beyond a config row: monitor a chosen owner and + have the whole *process tree* self-terminate on its `:DOWN` (§3, option 1). +- [`Registry`](https://hexdocs.pm/elixir/Registry.html) · + [Thoughtbot — dynamic process names](https://thoughtbot.com/blog/how-to-start-processes-with-dynamic-names-in-elixir) +- Constructor vs call/lookup argument order (§1): + [`GenServer`](https://hexdocs.pm/elixir/GenServer.html) · + [`Supervisor`](https://hexdocs.pm/elixir/Supervisor.html) · + [`Agent`](https://hexdocs.pm/elixir/Agent.html) · + [`Oban`](https://hexdocs.pm/oban/Oban.html) · + [`Phoenix.PubSub`](https://hexdocs.pm/phoenix_pubsub/Phoenix.PubSub.html) · + [`Finch`](https://hexdocs.pm/finch/Finch.html). +- In-repo: `lib/lightning/adaptors/supervisor.ex`, + `lib/lightning/collaboration.ex`, + `lib/lightning/collaboration/registry.ex`, + `lib/lightning/collaboration/document_supervisor.ex`, + `test/support/collaboration_helpers.ex`. diff --git a/CHANGELOG.md b/CHANGELOG.md index b3d0c5609b3..faed4131d5e 100644 --- a/CHANGELOG.md +++ b/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 diff --git a/CLAUDE.md b/CLAUDE.md index 59b9934e443..3b8ce54b89b 100644 --- a/CLAUDE.md +++ b/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 \ No newline at end of file diff --git a/config/test.exs b/config/test.exs index cb2d71914b7..3032eb87272 100644 --- a/config/test.exs +++ b/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 diff --git a/lib/lightning/collaboration.ex b/lib/lightning/collaboration.ex index a3a2f0af15c..2ddfdf52115 100644 --- a/lib/lightning/collaboration.ex +++ b/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 diff --git a/lib/lightning/collaboration/document_supervisor.ex b/lib/lightning/collaboration/document_supervisor.ex index c943884d21d..f86cbb45917 100644 --- a/lib/lightning/collaboration/document_supervisor.ex +++ b/lib/lightning/collaboration/document_supervisor.ex @@ -11,32 +11,65 @@ defmodule Lightning.Collaboration.DocumentSupervisor do Uses a transient restart strategy, only restarting if the supervisor itself crashes, not when child processes exit normally. Monitors both child processes and stops itself if either child crashes. + + Optionally monitors an `:owner` pid (passed through the child spec). When that + owner goes `:DOWN`, the supervisor stops `:normal` — running `terminate/2`'s + flush, and not restarting (transient). This is the owner-monitored + self-cleanup seam from + `.claude/guidelines/testable-supervision-trees.md` §3: any caller gets + deterministic teardown by passing `owner: self()`. With no owner (the + production default) the document outlives its starter as before. """ use GenServer - import Lightning.Collaboration.Registry, only: [via: 1] - alias Lightning.Collaboration.Persistence alias Lightning.Collaboration.PersistenceWriter + alias Lightning.Collaboration.Registry require Logger - @pg_scope :workflow_collaboration - def start_link(args, opts \\ []) do GenServer.start_link(__MODULE__, args, opts) end + @doc """ + Gracefully stops the DocumentSupervisor and its children. + + Synchronous: returns only once `terminate/2` has run, which flushes the + PersistenceWriter (via the SharedDoc) and stops both children. Because the + child spec uses `restart: :transient`, a `:normal` stop is not restarted by + the DynamicSupervisor. + """ + def stop(pid, timeout \\ 5_000) when is_pid(pid) do + GenServer.stop(pid, :normal, timeout) + end + @impl true def init(opts) do workflow = Keyword.fetch!(opts, :workflow) document_name = Keyword.fetch!(opts, :document_name) + registry = Keyword.get(opts, :registry, Registry) + pg_scope = Keyword.get(opts, :pg_scope, :workflow_collaboration) + + # Whether the SharedDoc exits on its own once its last observer leaves. + # Defaults to `true` (production behaviour). Callers that drive the document + # lifecycle explicitly — e.g. tests that want teardown sequenced solely + # through `terminate/2` rather than racing an asynchronous self-exit — can + # pass `auto_exit: false`. + auto_exit = Keyword.get(opts, :auto_exit, true) + + owner = Keyword.get(opts, :owner) + + # Optionally monitor an owner pid; when it goes :DOWN we stop :normal so the + # document tree dies with its owner. nil (production default) → no monitor. + owner_ref = if is_pid(owner), do: Process.monitor(owner), else: nil {:ok, persistence_writer_pid} = PersistenceWriter.start_link( document_name: document_name, workflow_id: workflow.id, - name: via({:persistence_writer, document_name}) + registry: registry, + name: Registry.via(registry, {:persistence_writer, document_name}) ) persistence_writer_ref = Process.monitor(persistence_writer_pid) @@ -45,25 +78,42 @@ defmodule Lightning.Collaboration.DocumentSupervisor do Yex.Sync.SharedDoc.start_link( [ doc_name: document_name, - auto_exit: true, + auto_exit: auto_exit, persistence: {Persistence, %{ workflow: workflow, - persistence_writer: persistence_writer_pid + persistence_writer: persistence_writer_pid, + registry: registry, + owner: owner }} ], - name: via({:shared_doc, document_name}) + name: Registry.via(registry, {:shared_doc, document_name}) ) # Register with :pg using document_name so versioned rooms are isolated - :ok = register_shared_doc_with_pg(document_name, shared_doc_pid) + :ok = register_shared_doc_with_pg(pg_scope, document_name, shared_doc_pid) shared_doc_ref = Process.monitor(shared_doc_pid) + # When started on behalf of an owner pid, grant that owner's process-scoped + # access (DB sandbox connection, mocks) to the two children, so they can + # reach the database and resolve mocks for the rest of their lives. The + # SharedDoc's *init-time* read is handled separately (the owner is threaded + # into its persistence state above, since `$callers` isn't propagated across + # `GenServer.start_link`); this grant covers everything after init. The + # callback defaults to a no-op, so production wiring is untouched; a real + # callback is only configured under the test environment. + if is_pid(owner) do + allow = grant_process_access_fun() + allow.(owner, persistence_writer_pid) + allow.(owner, shared_doc_pid) + end + {:ok, %{ workflow: workflow, + owner_ref: owner_ref, persistence_writer_pid: persistence_writer_pid, persistence_writer_ref: persistence_writer_ref, shared_doc_pid: shared_doc_pid, @@ -91,6 +141,10 @@ defmodule Lightning.Collaboration.DocumentSupervisor do @impl true def terminate(_reason, state) do + # Drop the owner monitor if it's still live (e.g. stopped via stop/2 while + # the owner is alive) so no stray :DOWN is delivered after we're gone. + if state[:owner_ref], do: Process.demonitor(state.owner_ref, [:flush]) + # Specifically stop the SharedDoc first, which sends a flush_and_stop # message to the PersistenceWriter. So we usually don't need to stop the # PersistenceWriter if the SharedDoc is exiting normally, but just in case @@ -124,6 +178,18 @@ defmodule Lightning.Collaboration.DocumentSupervisor do end @impl true + def handle_info( + {:DOWN, ref, :process, _pid, reason}, + %{owner_ref: ref} = state + ) do + Logger.debug("Owner DOWN, stopping document. reason: #{inspect(reason)}") + + Process.demonitor(ref, [:flush]) + + # Stop :normal so terminate/2 runs the flush; :transient => no restart. + {:stop, :normal, %{state | owner_ref: nil}} + end + def handle_info({:DOWN, ref, :process, _pid, reason}, state) do key = [:persistence_writer_ref, :shared_doc_ref] @@ -138,8 +204,15 @@ defmodule Lightning.Collaboration.DocumentSupervisor do {:stop, :normal, state |> Map.put(key, nil)} end - # Supervisor.start_link(children, strategy: :one_for_all) - defp register_shared_doc_with_pg(document_name, shared_doc_pid) do - :pg.join(@pg_scope, document_name, shared_doc_pid) + defp register_shared_doc_with_pg(pg_scope, document_name, shared_doc_pid) do + :pg.join(pg_scope, document_name, shared_doc_pid) + end + + defp grant_process_access_fun do + Application.get_env( + :lightning, + :collaboration_process_allow, + fn _owner, _pid -> :ok end + ) end end diff --git a/lib/lightning/collaboration/instance.ex b/lib/lightning/collaboration/instance.ex new file mode 100644 index 00000000000..8a2ef02a000 --- /dev/null +++ b/lib/lightning/collaboration/instance.ex @@ -0,0 +1,59 @@ +defmodule Lightning.Collaboration.Instance do + @moduledoc """ + Names the three pieces of collaboration infrastructure that a single + `Lightning.Collaboration.Supervisor` owns: its `Registry`, its + `DynamicSupervisor`, and its `:pg` scope. + + A plain value (no process), threaded from the public API down into child + specs so the whole tree can be addressed under a chosen base name. Production + uses `default/0`, which pins the three names to the application-wide + singletons started in `application.ex`. Tests can `derive/1` a fresh, isolated + set from a unique base, letting independent supervisor instances coexist. + """ + + @enforce_keys [:registry, :dynamic_supervisor, :pg_scope] + defstruct [:registry, :dynamic_supervisor, :pg_scope] + + @type t :: %__MODULE__{ + registry: atom(), + dynamic_supervisor: atom(), + pg_scope: atom() + } + + @doc """ + The production instance: the application-wide singletons. + + These are the literal names started by the collaboration supervision tree in + `application.ex`, so every defaulted public call resolves to the same global + infrastructure it always has. + """ + @spec default() :: t() + def default do + %__MODULE__{ + registry: Lightning.Collaboration.Registry, + dynamic_supervisor: Lightning.Collaboration.Supervisor.DocSupervisor, + pg_scope: :workflow_collaboration + } + end + + @doc """ + Derive an instance from a base name. + + Passing `Lightning.Collaboration.Supervisor` (the production base) yields the + exact `default/0` names. Any other base produces a distinct, isolated set, + which is how tests run independent supervisor instances side by side. + """ + @spec derive(atom()) :: t() + def derive(Lightning.Collaboration.Supervisor), do: default() + + def derive(base) when is_atom(base) do + %__MODULE__{ + registry: Module.concat(base, Registry), + dynamic_supervisor: Module.concat(base, DocSupervisor), + # The `:pg` scope must be a distinct atom from `base`: the supervisor + # registers itself under `base`, and `:pg.start_link/1` registers the + # scope process under the scope atom, so reusing `base` would clash. + pg_scope: Module.concat(base, PG) + } + end +end diff --git a/lib/lightning/collaboration/persistence.ex b/lib/lightning/collaboration/persistence.ex index 10ff16204be..f5c9432f79b 100644 --- a/lib/lightning/collaboration/persistence.ex +++ b/lib/lightning/collaboration/persistence.ex @@ -17,6 +17,21 @@ defmodule Lightning.Collaboration.Persistence do @impl true def bind(state, doc_name, doc) do + # `bind/3` runs inside the SharedDoc process and reads the database to load + # persisted state. When an owner is threaded through (tests running with an + # owner-scoped DB sandbox connection), register it as a caller of this + # process so the load can reach that connection. `$callers` is not + # propagated across `GenServer.start_link`, so it has to be set here, in the + # process that performs the read. In production no owner is present and this + # is a no-op. + case Map.get(state, :owner) do + owner when is_pid(owner) -> + Process.put(:"$callers", [owner | Process.get(:"$callers", [])]) + + _ -> + :ok + end + workflow = Map.fetch!(state, :workflow) if !workflow do @@ -46,7 +61,13 @@ defmodule Lightning.Collaboration.Persistence do @impl true def update_v1(state, update, doc_name, _doc) do - case PersistenceWriter.add_update(doc_name, update) do + # Resolve the writer through the instance's registry. Defaults to the global + # collaboration Registry when none was threaded in (production), so this is + # byte-for-byte the same lookup as before; an isolated instance finds its + # own writer rather than missing in the global registry. + registry = Map.get(state, :registry, Lightning.Collaboration.Registry) + + case PersistenceWriter.add_update(registry, doc_name, update) do :ok -> state diff --git a/lib/lightning/collaboration/persistence_writer.ex b/lib/lightning/collaboration/persistence_writer.ex index 40af51b8e70..a94d83d4bd2 100644 --- a/lib/lightning/collaboration/persistence_writer.ex +++ b/lib/lightning/collaboration/persistence_writer.ex @@ -38,6 +38,7 @@ defmodule Lightning.Collaboration.PersistenceWriter do defstruct [ :document_name, + :registry, :save_timer, :max_wait_timer, :last_save_at, @@ -61,9 +62,13 @@ defmodule Lightning.Collaboration.PersistenceWriter do @doc """ Adds a Yjs update to the pending batch for persistence. + + The writer is located through `registry` (defaulting to the global + collaboration Registry), so an isolated instance finds its own writer. """ - def add_update(document_name, update) when is_binary(update) do - if pid = Registry.whereis({:persistence_writer, document_name}) do + def add_update(registry \\ Registry, document_name, update) + when is_binary(update) do + if pid = Registry.whereis(registry, {:persistence_writer, document_name}) do GenServer.cast(pid, {:add_update, update}) :ok else @@ -77,8 +82,8 @@ defmodule Lightning.Collaboration.PersistenceWriter do This is called when the document is being unbound and ensures all updates are persisted before cleanup. """ - def flush_and_stop(document_name) do - case Registry.lookup({:persistence_writer, document_name}) do + def flush_and_stop(registry \\ Registry, document_name) do + case Registry.lookup(registry, {:persistence_writer, document_name}) do [{pid, _}] -> GenServer.call(pid, :flush_and_stop, 10_000) @@ -97,6 +102,7 @@ defmodule Lightning.Collaboration.PersistenceWriter do state = %__MODULE__{ document_name: document_name, + registry: Keyword.get(opts, :registry, Registry), last_save_at: DateTime.utc_now() } diff --git a/lib/lightning/collaboration/registry.ex b/lib/lightning/collaboration/registry.ex index 1723221f56a..afba355a184 100644 --- a/lib/lightning/collaboration/registry.ex +++ b/lib/lightning/collaboration/registry.ex @@ -39,11 +39,16 @@ defmodule Lightning.Collaboration.Registry do @doc """ Child specification for starting the Registry. + + The registered name is injectable via the `:name` option (defaulting to + `__MODULE__`) so a supervisor instance can run its own Registry. """ - def child_spec(_opts) do + def child_spec(opts) do + {name, _opts} = Keyword.pop(opts, :name, __MODULE__) + %{ - id: __MODULE__, - start: {Registry, :start_link, [[keys: :unique, name: __MODULE__]]}, + id: name, + start: {Registry, :start_link, [[keys: :unique, name: name]]}, type: :supervisor } end @@ -65,16 +70,17 @@ defmodule Lightning.Collaboration.Registry do # => {:error, {:already_registered, #PID<0.456.0>}} """ - @spec register(term()) :: {:ok, pid()} | {:error, {:already_registered, pid()}} - def register(key) do - case Registry.register(__MODULE__, key, nil) do + @spec register(registry :: atom(), term()) :: + {:ok, pid()} | {:error, {:already_registered, pid()}} + def register(registry \\ __MODULE__, key) do + case Registry.register(registry, key, nil) do {:ok, _pid} -> {:ok, self()} {:error, reason} -> {:error, reason} end end - def via(key) do - {:via, Registry, {__MODULE__, key}} + def via(registry \\ __MODULE__, key) do + {:via, Registry, {registry, key}} end @doc """ @@ -92,9 +98,9 @@ defmodule Lightning.Collaboration.Registry do # => [] """ - @spec lookup(term()) :: [{pid(), term()}] - def lookup(key) do - Registry.lookup(__MODULE__, key) + @spec lookup(registry :: atom(), term()) :: [{pid(), term()}] + def lookup(registry \\ __MODULE__, key) do + Registry.lookup(registry, key) end @doc """ @@ -112,22 +118,28 @@ defmodule Lightning.Collaboration.Registry do # => nil """ - @spec whereis(term()) :: pid() | nil - def whereis(key) do - case lookup(key) do + @spec whereis(registry :: atom(), term()) :: pid() | nil + def whereis(registry \\ __MODULE__, key) do + case lookup(registry, key) do [{pid, _value}] -> pid [] -> nil end end - def count(key \\ nil) + @doc """ + Count registered processes, optionally restricted to a key prefix. + + With no key (`count/0`, or `count/1` with a registry), counts every entry in + the registry. With a key, counts entries whose key starts with it. + """ + def count(registry \\ __MODULE__, key) - def count(nil) do - Registry.count(__MODULE__) + def count(registry, nil) do + Registry.count(registry) end - def count(key) do - select(key) |> length() + def count(registry, key) do + select(registry, key) |> length() end @doc """ @@ -139,10 +151,10 @@ defmodule Lightning.Collaboration.Registry do We do a select with any key that _starts with_ the given key. """ - @spec select(binary()) :: [{term(), pid()}] - def select(key) when is_binary(key) do + @spec select(registry :: atom(), binary()) :: [{term(), pid()}] + def select(registry \\ __MODULE__, key) when is_binary(key) do # We have two select specs, for keys with and without values. - Registry.select(__MODULE__, [ + Registry.select(registry, [ {{{:"$1", :"$2"}, :"$3", :"$4"}, [{:==, {:binary_part, :"$2", 0, byte_size(key)}, key}], [[:"$1", :"$3"]]}, {{{:"$1", :"$2", :"$5"}, :"$3", :"$4"}, @@ -150,8 +162,21 @@ defmodule Lightning.Collaboration.Registry do ]) end - def get_group(key) do - select(key) + @doc """ + List the `document_name`s of every registered DocumentSupervisor. + + The match spec lives here, beside the other key-shape knowledge this module + owns, rather than being hand-rolled by callers. + """ + @spec doc_supervisor_names(registry :: atom()) :: [binary()] + def doc_supervisor_names(registry \\ __MODULE__) do + Registry.select(registry, [ + {{{:doc_supervisor, :"$1"}, :_, :_}, [], [:"$1"]} + ]) + end + + def get_group(registry \\ __MODULE__, key) do + select(registry, key) |> Enum.reduce(%{}, fn [type, pid], acc -> case type do :session -> diff --git a/lib/lightning/collaboration/session.ex b/lib/lightning/collaboration/session.ex index 164d4b18578..67a9e4d9a5b 100644 --- a/lib/lightning/collaboration/session.ex +++ b/lib/lightning/collaboration/session.ex @@ -37,8 +37,6 @@ defmodule Lightning.Collaboration.Session do :document_name ] - @pg_scope :workflow_collaboration - @type start_opts :: [ workflow: Lightning.Workflows.Workflow.t(), user: User.t(), @@ -94,6 +92,7 @@ defmodule Lightning.Collaboration.Session do user = Keyword.fetch!(opts, :user) parent_pid = Keyword.fetch!(opts, :parent_pid) document_name = Keyword.fetch!(opts, :document_name) + pg_scope = Keyword.get(opts, :pg_scope, :workflow_collaboration) Logger.info("Starting session for document #{document_name}") @@ -108,7 +107,7 @@ defmodule Lightning.Collaboration.Session do document_name: document_name } - lookup_shared_doc(document_name) + lookup_shared_doc(pg_scope, document_name) |> case do nil -> {:stop, {:error, :shared_doc_not_found}} @@ -153,8 +152,8 @@ defmodule Lightning.Collaboration.Session do :ok end - def lookup_shared_doc(document_name) do - case :pg.get_members(@pg_scope, document_name) do + def lookup_shared_doc(pg_scope \\ :workflow_collaboration, document_name) do + case :pg.get_members(pg_scope, document_name) do [] -> nil [shared_doc_pid | _] -> shared_doc_pid end diff --git a/lib/lightning/collaboration/supervisor.ex b/lib/lightning/collaboration/supervisor.ex index a9fd97cb64d..1ac799a5ecd 100644 --- a/lib/lightning/collaboration/supervisor.ex +++ b/lib/lightning/collaboration/supervisor.ex @@ -10,19 +10,33 @@ defmodule Lightning.Collaboration.Supervisor do """ use Supervisor - def start_link(init_arg) do - Supervisor.start_link(__MODULE__, init_arg, name: __MODULE__) + alias Lightning.Collaboration.Instance + + @doc """ + Starts the collaboration supervisor. + + The registered name is taken from the `:name` option (defaulting to + `__MODULE__`). That name is the base from which `init/1` derives the + instance's Registry, `:pg` scope, and DynamicSupervisor names, so a test can + start a fully isolated tree by passing a unique `:name`. + """ + def start_link(opts) do + {name, opts} = Keyword.pop(opts, :name, __MODULE__) + Supervisor.start_link(__MODULE__, Keyword.put(opts, :name, name), name: name) end @impl true - def init(_opts) do + def init(opts) do + name = Keyword.fetch!(opts, :name) + instance = Instance.derive(name) + children = [ # Start Registry first - processes depend on it for registration - Lightning.Collaboration.Registry, + {Lightning.Collaboration.Registry, name: instance.registry}, # Start :pg for cluster-wide SharedDoc coordination %{ - id: :workflow_collaboration_pg, - start: {:pg, :start_link, [:workflow_collaboration]}, + id: :collaboration_pg, + start: {:pg, :start_link, [instance.pg_scope]}, type: :worker }, # Start the dynamic supervisor for collaboration processes @@ -33,7 +47,7 @@ defmodule Lightning.Collaboration.Supervisor do [ [ strategy: :one_for_one, - name: __MODULE__.DocSupervisor + name: instance.dynamic_supervisor ] ]}, type: :supervisor @@ -43,11 +57,12 @@ defmodule Lightning.Collaboration.Supervisor do Supervisor.init(children, strategy: :one_for_one) end - def start_child(child_spec) do - DynamicSupervisor.start_child(__MODULE__.DocSupervisor, child_spec) + def start_child(dynamic_supervisor \\ __MODULE__.DocSupervisor, child_spec) do + DynamicSupervisor.start_child(dynamic_supervisor, child_spec) end - def terminate_child(pid) when is_pid(pid) do - DynamicSupervisor.terminate_child(__MODULE__.DocSupervisor, pid) + def terminate_child(dynamic_supervisor \\ __MODULE__.DocSupervisor, pid) + when is_pid(pid) do + DynamicSupervisor.terminate_child(dynamic_supervisor, pid) end end diff --git a/test/lightning/collaborate_test.exs b/test/lightning/collaborate_test.exs index 16b2bf2350d..9192949086b 100644 --- a/test/lightning/collaborate_test.exs +++ b/test/lightning/collaborate_test.exs @@ -1,5 +1,5 @@ defmodule Lightning.CollaborateTest do - use Lightning.DataCase, async: false + use Lightning.DataCase, async: true alias Lightning.Collaborate alias Lightning.Collaboration.Registry @@ -8,31 +8,51 @@ defmodule Lightning.CollaborateTest do import Eventually import Lightning.CollaborationHelpers + # Each test drives its own isolated collaboration tree (Registry, + # DynamicSupervisor, and `:pg` scope), so concurrent tests can't see or + # collide with each other's documents, sessions, or process-group members. + # The instance is `start_supervised!`-owned, so the whole tree — including the + # DB-writing SharedDoc/PersistenceWriter children — is torn down before this + # test process (the sandbox owner) exits, avoiding a connection-checkin race. setup do + instance = start_collaboration_instance() user = insert(:user) workflow = insert(:workflow) - on_exit(fn -> - ensure_doc_supervisor_stopped(workflow.id) - end) - - {:ok, user: user, workflow: workflow} + {:ok, instance: instance, user: user, workflow: workflow} end - describe "start/1" do + describe "start/2" do test "starting a new collaboration with no existing SharedDoc", %{ + instance: instance, user: user, workflow: workflow } do + # Pre-start the document under an owner (self()) so its SharedDoc/ + # PersistenceWriter children can reach this test's sandbox connection (the + # SharedDoc reads the DB during init). Collaborate.start/2 then reuses the + # existing document rather than starting an unowned one. owner: self() also + # ties the tree to this test for deterministic, flush-inclusive teardown. + {:ok, _doc_supervisor} = + start_collaboration_document( + instance, + workflow, + "workflow:#{workflow.id}" + ) + assert {:ok, session_pid} = - Collaborate.start(user: user, workflow: workflow) + Collaborate.start(instance, user: user, workflow: workflow) assert Process.alive?(session_pid) assert shared_doc_pid = - Collaborate.whereis({:shared_doc, "workflow:#{workflow.id}"}) + Registry.whereis( + instance.registry, + {:shared_doc, "workflow:#{workflow.id}"} + ) - assert Collaborate.whereis( + assert Registry.whereis( + instance.registry, {:persistence_writer, "workflow:#{workflow.id}"} ) @@ -43,18 +63,30 @@ defmodule Lightning.CollaborateTest do end test "starting a new collaboration with an existing SharedDoc", %{ + instance: instance, user: user, workflow: workflow } do + # Pre-start the document under an owner (self()) so its children can reach + # this test's sandbox and the tree is torn down before the owner exits. + # Collaborate.start/2 then reuses this existing document. + {:ok, _doc_supervisor} = + start_collaboration_document( + instance, + workflow, + "workflow:#{workflow.id}" + ) + assert {:ok, session_1} = - Collaborate.start(user: user, workflow: workflow) + Collaborate.start(instance, user: user, workflow: workflow) assert {:ok, session_2} = - Collaborate.start(user: user, workflow: workflow) + Collaborate.start(instance, user: user, workflow: workflow) refute session_1 == session_2, "Same user and workflow get new sessions" - process_group = Registry.get_group("workflow:#{workflow.id}") + process_group = + Registry.get_group(instance.registry, "workflow:#{workflow.id}") for {key, _} <- process_group do assert key in [ @@ -65,12 +97,12 @@ defmodule Lightning.CollaborateTest do ] end - assert Registry.count("workflow:#{workflow.id}") == 5 + assert Registry.count(instance.registry, "workflow:#{workflow.id}") == 5 assert %{ persistence_writer: persistence_writer_pid, shared_doc: shared_doc_pid - } = Registry.get_group("workflow:#{workflow.id}") + } = Registry.get_group(instance.registry, "workflow:#{workflow.id}") assert Process.alive?(persistence_writer_pid) assert Process.alive?(shared_doc_pid) @@ -91,18 +123,93 @@ defmodule Lightning.CollaborateTest do refute_eventually(Process.alive?(shared_doc_pid)) end - test "start_document/2 is idempotent for the same document", %{ + test "start_document/3 is idempotent for the same document", %{ + instance: instance, workflow: workflow } do document_name = "workflow:#{workflow.id}" + # owner: self() (via the helper) ties the document tree to this test, so + # its DB-writing children are stopped before the sandbox owner exits. + assert {:ok, doc_supervisor_pid} = + start_collaboration_document(instance, workflow, document_name) + + assert Process.alive?(doc_supervisor_pid) + + assert {:ok, ^doc_supervisor_pid} = + Collaborate.start_document(instance, workflow, document_name) + end + + test "start_document/4 with an owner self-terminates when the owner exits", + %{instance: instance, workflow: workflow} do + document_name = "workflow:#{workflow.id}" + + # A separate owner process we control, so the document tree's shutdown is + # driven by this process exiting rather than by the test finishing. It is + # also the pid threaded into the SharedDoc's init-time DB read (via + # $callers), so grant it this test's sandbox connection first. + owner = spawn(fn -> Process.sleep(:infinity) end) + Ecto.Adapters.SQL.Sandbox.allow(Lightning.Repo, self(), owner) + assert {:ok, doc_supervisor_pid} = - Collaborate.start_document(workflow, document_name) + Collaborate.start_document(instance, workflow, document_name, + owner: owner + ) + doc_supervisor_ref = Process.monitor(doc_supervisor_pid) assert Process.alive?(doc_supervisor_pid) + assert Registry.whereis( + instance.registry, + {:doc_supervisor, document_name} + ) == + doc_supervisor_pid + + # Killing the owner stops the document tree with reason :normal, so + # terminate/2 runs the flush and the :transient child isn't restarted. + Process.exit(owner, :kill) + + assert_receive {:DOWN, ^doc_supervisor_ref, :process, ^doc_supervisor_pid, + :normal}, + 5000 + + refute_eventually( + Registry.whereis(instance.registry, {:doc_supervisor, document_name}) + ) + + refute_eventually( + Registry.whereis(instance.registry, {:shared_doc, document_name}) + ) + + refute_eventually( + Registry.whereis( + instance.registry, + {:persistence_writer, document_name} + ) + ) + end + + test "start_document monitors only the given owner (production default = none)", + %{instance: instance, workflow: workflow} do + document_name = "workflow:#{workflow.id}" + + # Start the document under this test as owner. The owner field is both the + # monitor target and (in tests) the pid the SharedDoc's init-time DB read + # is threaded through, so a real owner is what lets the tree start under an + # async sandbox at all. + {:ok, doc_supervisor_pid} = + start_collaboration_document(instance, workflow, document_name) + + # With an explicit owner, a monitor is set up keyed to that owner. + assert is_reference(:sys.get_state(doc_supervisor_pid).owner_ref) + + # The production default is the no-owner 3-arity: a second call without an + # owner is idempotent and reuses the existing tree — it does not impose a + # new monitor — so a production document outlives whoever re-requests it. assert {:ok, ^doc_supervisor_pid} = - Collaborate.start_document(workflow, document_name) + Collaborate.start_document(instance, workflow, document_name) + + assert is_reference(:sys.get_state(doc_supervisor_pid).owner_ref) end end end diff --git a/test/lightning/collaboration/document_supervisor_test.exs b/test/lightning/collaboration/document_supervisor_test.exs index 1b5e881a955..24ec941344b 100644 --- a/test/lightning/collaboration/document_supervisor_test.exs +++ b/test/lightning/collaboration/document_supervisor_test.exs @@ -1,5 +1,5 @@ defmodule Lightning.Collaboration.DocumentSupervisorTest do - use Lightning.DataCase, async: false + use Lightning.DataCase, async: true alias Lightning.Collaboration.DocumentState alias Lightning.Collaboration.DocumentSupervisor @@ -8,31 +8,59 @@ defmodule Lightning.Collaboration.DocumentSupervisorTest do import Eventually import Lightning.Factories + import Lightning.CollaborationHelpers import ExUnit.CaptureLog - # Common setup for most tests + # Common setup for most tests. + # + # Each test drives its own isolated collaboration tree (Registry, + # DynamicSupervisor, and `:pg` scope) so concurrent tests can't see or collide + # with each other's registrations or process-group members. The instance's + # registry and pg scope are threaded into every Registry/`:pg` call and into + # every DocumentSupervisor started below. setup do Process.flag(:trap_exit, true) + instance = start_collaboration_instance() workflow = insert(:workflow) workflow_id = workflow.id document_name = "workflow:#{workflow_id}" {:ok, - workflow: workflow, workflow_id: workflow_id, document_name: document_name} + instance: instance, + workflow: workflow, + workflow_id: workflow_id, + document_name: document_name} end # Setup for tests that need a running DocumentSupervisor defp setup_document_supervisor(context) do {:ok, doc_supervisor} = DocumentSupervisor.start_link( - [workflow: context.workflow, document_name: context.document_name], - name: Registry.via({:doc_supervisor, context.document_name}) + [ + workflow: context.workflow, + document_name: context.document_name, + registry: context.instance.registry, + pg_scope: context.instance.pg_scope, + owner: self() + ], + name: + Registry.via( + context.instance.registry, + {:doc_supervisor, context.document_name} + ) ) persistence_writer = - Registry.whereis({:persistence_writer, context.document_name}) + Registry.whereis( + context.instance.registry, + {:persistence_writer, context.document_name} + ) - shared_doc = Registry.whereis({:shared_doc, context.document_name}) + shared_doc = + Registry.whereis( + context.instance.registry, + {:shared_doc, context.document_name} + ) assert Process.alive?(doc_supervisor) assert Process.alive?(persistence_writer) @@ -53,7 +81,14 @@ defmodule Lightning.Collaboration.DocumentSupervisorTest do DocumentSupervisor.child_spec( workflow: context.workflow, document_name: context.document_name, - name: Registry.via({:doc_supervisor, context.document_name}) + registry: context.instance.registry, + pg_scope: context.instance.pg_scope, + owner: self(), + name: + Registry.via( + context.instance.registry, + {:doc_supervisor, context.document_name} + ) ) Map.merge(context, %{ @@ -63,15 +98,23 @@ defmodule Lightning.Collaboration.DocumentSupervisorTest do end # Helper function to verify cleanup after process termination - defp verify_cleanup(document_name, _workflow_id) do + defp verify_cleanup(instance, document_name, _workflow_id) do # Verify Registry is cleaned up eventually - refute_eventually(Registry.whereis({:doc_supervisor, document_name})) - refute_eventually(Registry.whereis({:persistence_writer, document_name})) - refute_eventually(Registry.whereis({:shared_doc, document_name})) + refute_eventually( + Registry.whereis(instance.registry, {:doc_supervisor, document_name}) + ) + + refute_eventually( + Registry.whereis(instance.registry, {:persistence_writer, document_name}) + ) + + refute_eventually( + Registry.whereis(instance.registry, {:shared_doc, document_name}) + ) # Verify process group is cleaned up eventually refute_eventually( - :pg.get_members(:workflow_collaboration, document_name) + :pg.get_members(instance.pg_scope, document_name) |> Enum.any?() ) end @@ -155,11 +198,16 @@ defmodule Lightning.Collaboration.DocumentSupervisorTest do refute Process.alive?(setup_data.persistence_writer) refute Process.alive?(setup_data.shared_doc) - verify_cleanup(setup_data.document_name, setup_data.workflow_id) + verify_cleanup( + setup_data.instance, + setup_data.document_name, + setup_data.workflow_id + ) end # Helper function to verify successful DocumentSupervisor initialization defp verify_initialization(%{ + instance: instance, workflow_id: workflow_id, document_name: document_name, doc_supervisor: doc_supervisor, @@ -167,11 +215,13 @@ defmodule Lightning.Collaboration.DocumentSupervisorTest do shared_doc: shared_doc }) do # Verify DocumentSupervisor is registered correctly - registered_supervisor = Registry.whereis({:doc_supervisor, document_name}) + registered_supervisor = + Registry.whereis(instance.registry, {:doc_supervisor, document_name}) + assert registered_supervisor == doc_supervisor # Verify SharedDoc is in process group (now keyed by document_name, not workflow_id) - members = :pg.get_members(:workflow_collaboration, document_name) + members = :pg.get_members(instance.pg_scope, document_name) assert shared_doc in members # Verify both processes are monitored by checking state @@ -183,13 +233,13 @@ defmodule Lightning.Collaboration.DocumentSupervisorTest do assert state.workflow.id == workflow_id # Verify all processes are grouped correctly in Registry - group = Registry.get_group(document_name) + group = Registry.get_group(instance.registry, document_name) assert Map.has_key?(group, :persistence_writer) assert Map.has_key?(group, :shared_doc) assert Map.has_key?(group, :doc_supervisor) # Count all registered processes for this workflow - assert Registry.count(document_name) == 3 + assert Registry.count(instance.registry, document_name) == 3 end # Helper function to test DocumentSupervisor startup failure @@ -292,7 +342,11 @@ defmodule Lightning.Collaboration.DocumentSupervisorTest do assert_all_processes_terminated(processes, monitor_refs) - verify_cleanup(context.document_name, context.workflow_id) + verify_cleanup( + context.instance, + context.document_name, + context.workflow_id + ) end test "3.2 - Termination with Already Dead Children", context do @@ -320,7 +374,11 @@ defmodule Lightning.Collaboration.DocumentSupervisorTest do refute Process.alive?(context.shared_doc) end) - verify_cleanup(context.document_name, context.workflow_id) + verify_cleanup( + context.instance, + context.document_name, + context.workflow_id + ) end test "3.3 - Termination Order Verification", context do @@ -354,7 +412,12 @@ defmodule Lightning.Collaboration.DocumentSupervisorTest do "PersistenceWriter: #{persistence_writer_time}ms" refute Process.alive?(context.doc_supervisor) - verify_cleanup(context.document_name, context.workflow_id) + + verify_cleanup( + context.instance, + context.document_name, + context.workflow_id + ) end end @@ -440,7 +503,10 @@ defmodule Lightning.Collaboration.DocumentSupervisorTest do # Verify it's alive and registered assert Process.alive?(doc_supervisor) - assert Registry.whereis({:doc_supervisor, context.document_name}) == + assert Registry.whereis( + context.instance.registry, + {:doc_supervisor, context.document_name} + ) == doc_supervisor # Test transient behavior - normal exit should NOT restart @@ -455,7 +521,10 @@ defmodule Lightning.Collaboration.DocumentSupervisorTest do # Verify no restart occurred for normal termination # Wait a reasonable time to ensure supervisor doesn't restart refute_eventually( - Registry.whereis({:doc_supervisor, context.document_name}) != nil, + Registry.whereis( + context.instance.registry, + {:doc_supervisor, context.document_name} + ) != nil, 1000 ) @@ -480,18 +549,23 @@ defmodule Lightning.Collaboration.DocumentSupervisorTest do :killed}, 5000 - # Wait for the supervisor to restart the child + # The transient strategy restarts the child after an abnormal exit. We + # already saw the killed process go down, but the BEAM can reuse its PID + # and the registry entry clears asynchronously, so poll for a registered, + # live replacement instead of comparing PIDs. assert_eventually( - Registry.whereis({:doc_supervisor, context.document_name}) != nil, + case Registry.whereis( + context.instance.registry, + {:doc_supervisor, context.document_name} + ) do + pid when is_pid(pid) -> Process.alive?(pid) + _ -> false + end, 2000 ) - # Verify the restarted process is different - restarted_supervisor = - Registry.whereis({:doc_supervisor, context.document_name}) - - assert restarted_supervisor != doc_supervisor2 - assert Process.alive?(restarted_supervisor) + assert %{active: 1} = + DynamicSupervisor.count_children(setup_data.test_supervisor) end) # Verify the test supervisor itself is still alive (didn't crash due to child failure) @@ -507,6 +581,7 @@ defmodule Lightning.Collaboration.DocumentSupervisorTest do setup :setup_document_supervisor test "5.1 - Registry Registration", %{ + instance: instance, workflow_id: workflow_id, document_name: document_name, doc_supervisor: doc_supervisor, @@ -518,17 +593,18 @@ defmodule Lightning.Collaboration.DocumentSupervisorTest do doc_supervisor: ^doc_supervisor, persistence_writer: ^persistence_writer, shared_doc: ^shared_doc - } = Registry.get_group(document_name) + } = Registry.get_group(instance.registry, document_name) # Test count function - assert Registry.count(document_name) == 3 + assert Registry.count(instance.registry, document_name) == 3 # Clean up GenServer.stop(doc_supervisor, :normal) - verify_cleanup(document_name, workflow_id) + verify_cleanup(instance, document_name, workflow_id) end test "5.2 - Process Group Registration", %{ + instance: instance, workflow_id: workflow_id, document_name: document_name, doc_supervisor: doc_supervisor, @@ -536,14 +612,14 @@ defmodule Lightning.Collaboration.DocumentSupervisorTest do } do # Test SharedDoc is the only member in workflow_collaboration process group assert [^shared_doc] = - :pg.get_members(:workflow_collaboration, document_name) + :pg.get_members(instance.pg_scope, document_name) assert [^shared_doc] = - :pg.get_local_members(:workflow_collaboration, document_name) + :pg.get_local_members(instance.pg_scope, document_name) # Clean up and verify process group is cleaned up GenServer.stop(doc_supervisor, :normal) - verify_cleanup(document_name, workflow_id) + verify_cleanup(instance, document_name, workflow_id) end end @@ -551,6 +627,7 @@ defmodule Lightning.Collaboration.DocumentSupervisorTest do setup :setup_document_supervisor test "6.1 - Rapid Child Restarts", %{ + instance: instance, workflow_id: workflow_id, document_name: document_name, doc_supervisor: doc_supervisor @@ -562,9 +639,13 @@ defmodule Lightning.Collaboration.DocumentSupervisorTest do # Rapidly kill the same child multiple times to simulate rapid failures # DocumentSupervisor should handle this gracefully and terminate cleanly persistence_writer = - Registry.whereis({:persistence_writer, document_name}) + Registry.whereis( + instance.registry, + {:persistence_writer, document_name} + ) - shared_doc = Registry.whereis({:shared_doc, document_name}) + shared_doc = + Registry.whereis(instance.registry, {:shared_doc, document_name}) # Kill persistence_writer first GenServer.stop(persistence_writer, :kill) @@ -580,10 +661,11 @@ defmodule Lightning.Collaboration.DocumentSupervisorTest do refute Process.alive?(shared_doc) end) - verify_cleanup(document_name, workflow_id) + verify_cleanup(instance, document_name, workflow_id) end test "6.2 - Concurrent Stop Requests", %{ + instance: instance, workflow_id: workflow_id, document_name: document_name, doc_supervisor: doc_supervisor @@ -633,7 +715,7 @@ defmodule Lightning.Collaboration.DocumentSupervisorTest do # Verify cleanup happened only once - no orphaned processes refute Process.alive?(doc_supervisor) - verify_cleanup(document_name, workflow_id) + verify_cleanup(instance, document_name, workflow_id) end) # Verify we captured the expected concurrent termination error @@ -641,6 +723,7 @@ defmodule Lightning.Collaboration.DocumentSupervisorTest do end test "6.3 - Process Already Stopping", %{ + instance: instance, workflow_id: workflow_id, document_name: document_name, doc_supervisor: doc_supervisor, @@ -672,22 +755,27 @@ defmodule Lightning.Collaboration.DocumentSupervisorTest do 5000 end) - verify_cleanup(document_name, workflow_id) + verify_cleanup(instance, document_name, workflow_id) end end describe "7. Integration Points" do - test "7.1 - With Lightning.Collaborate", %{workflow: workflow} do + test "7.1 - With Lightning.Collaborate", %{ + instance: instance, + workflow: workflow + } do workflow_id = workflow.id document_name = "workflow:#{workflow_id}" user = insert(:user) - # Test that Lightning.Collaborate.start/1 creates DocumentSupervisor through session + # Test that Lightning.Collaborate.start/2 creates DocumentSupervisor through session {:ok, session} = - Lightning.Collaborate.start(workflow: workflow, user: user) + Lightning.Collaborate.start(instance, workflow: workflow, user: user) # Verify DocumentSupervisor is created and registered - doc_supervisor = Registry.whereis({:doc_supervisor, document_name}) + doc_supervisor = + Registry.whereis(instance.registry, {:doc_supervisor, document_name}) + assert doc_supervisor != nil # Verify all expected processes are registered @@ -695,37 +783,50 @@ defmodule Lightning.Collaboration.DocumentSupervisorTest do doc_supervisor: ^doc_supervisor, persistence_writer: _persistence_writer, shared_doc: shared_doc - } = Registry.get_group(document_name) + } = Registry.get_group(instance.registry, document_name) # Verify SharedDoc is in process group assert [^shared_doc] = - :pg.get_members(:workflow_collaboration, document_name) + :pg.get_members(instance.pg_scope, document_name) # Clean up session which should clean up DocumentSupervisor # Capture potential race condition logs during process shutdown capture_log(fn -> GenServer.stop(session, :normal) - verify_cleanup(document_name, workflow_id) + verify_cleanup(instance, document_name, workflow_id) end) end setup :setup_document_supervisor test "7.2 - With Session Processes", %{ - workflow_id: _workflow_id, + instance: instance, + workflow_id: workflow_id, document_name: document_name, + doc_supervisor: doc_supervisor, shared_doc: shared_doc } do # Both discovery methods should find the same process - found_via_registry = Registry.whereis({:shared_doc, document_name}) - [found_via_pg] = :pg.get_members(:workflow_collaboration, document_name) + found_via_registry = + Registry.whereis(instance.registry, {:shared_doc, document_name}) + + [found_via_pg] = :pg.get_members(instance.pg_scope, document_name) assert found_via_registry == shared_doc assert found_via_pg == shared_doc assert found_via_registry == found_via_pg + + # Stop the tree synchronously while this test process is still alive. The + # children flush to the database under terminate/2; doing it here (rather + # than leaving it to the owner-DOWN path after the test process exits) + # guarantees no child is mid-query when the sandbox connection is checked + # back in. + GenServer.stop(doc_supervisor, :normal) + verify_cleanup(instance, document_name, workflow_id) end test "7.3 - Persistence Flow", %{ + instance: instance, workflow_id: workflow_id, document_name: document_name, shared_doc: shared_doc, @@ -747,12 +848,12 @@ defmodule Lightning.Collaboration.DocumentSupervisorTest do 5000 # Cleanup is handled by DocumentSupervisor monitoring - verify_cleanup(document_name, workflow_id) + verify_cleanup(instance, document_name, workflow_id) end end describe "8. Resource Management" do - test "8.1 - Memory Leaks", %{workflow: workflow} do + test "8.1 - Memory Leaks", %{instance: instance, workflow: workflow} do workflow_id = workflow.id document_name = "workflow:#{workflow_id}" @@ -761,8 +862,15 @@ defmodule Lightning.Collaboration.DocumentSupervisorTest do # Start DocumentSupervisor {:ok, doc_supervisor} = DocumentSupervisor.start_link( - [workflow: workflow, document_name: document_name], - name: Registry.via({:doc_supervisor, document_name}) + [ + workflow: workflow, + document_name: document_name, + registry: instance.registry, + pg_scope: instance.pg_scope, + owner: self() + ], + name: + Registry.via(instance.registry, {:doc_supervisor, document_name}) ) # Verify all processes are created and registered @@ -770,32 +878,39 @@ defmodule Lightning.Collaboration.DocumentSupervisorTest do doc_supervisor: ^doc_supervisor, persistence_writer: _persistence_writer, shared_doc: shared_doc - } = Registry.get_group(document_name) + } = Registry.get_group(instance.registry, document_name) # Verify process group membership assert [^shared_doc] = - :pg.get_members(:workflow_collaboration, document_name) + :pg.get_members(instance.pg_scope, document_name) # Stop DocumentSupervisor normally GenServer.stop(doc_supervisor, :normal) # Verify complete cleanup after each cycle - verify_cleanup(document_name, workflow_id) + verify_cleanup(instance, document_name, workflow_id) end end - test "8.2 - Timeout Handling", %{workflow: workflow} do + test "8.2 - Timeout Handling", %{instance: instance, workflow: workflow} do workflow_id = workflow.id document_name = "workflow:#{workflow_id}" # Start DocumentSupervisor {:ok, doc_supervisor} = DocumentSupervisor.start_link( - [workflow: workflow, document_name: document_name], - name: Registry.via({:doc_supervisor, document_name}) + [ + workflow: workflow, + document_name: document_name, + registry: instance.registry, + pg_scope: instance.pg_scope, + owner: self() + ], + name: Registry.via(instance.registry, {:doc_supervisor, document_name}) ) - shared_doc = Registry.whereis({:shared_doc, document_name}) + shared_doc = + Registry.whereis(instance.registry, {:shared_doc, document_name}) # Make SharedDoc unresponsive by suspending it :sys.suspend(shared_doc) @@ -811,12 +926,15 @@ defmodule Lightning.Collaboration.DocumentSupervisorTest do :shutdown}, 15_000 - verify_cleanup(document_name, workflow_id) + verify_cleanup(instance, document_name, workflow_id) end end describe "9. Checkpoint Creation" do - test "creates checkpoint from persisted updates", %{workflow: workflow} do + test "creates checkpoint from persisted updates", %{ + instance: instance, + workflow: workflow + } do document_name = "workflow:#{workflow.id}" # Create some initial document state with Y.Doc data @@ -853,18 +971,30 @@ defmodule Lightning.Collaboration.DocumentSupervisorTest do version: :update }) - # Start PersistenceWriter - {:ok, persistence_writer} = - PersistenceWriter.start_link( - document_name: document_name, - name: Registry.via({:persistence_writer, document_name}) + # Start PersistenceWriter under ExUnit so it is torn down before the + # sandbox owner is stopped, even if an assertion below fails. + persistence_writer = + start_supervised!( + {PersistenceWriter, + document_name: document_name, + registry: instance.registry, + name: + Registry.via( + instance.registry, + {:persistence_writer, document_name} + )} ) - # Trigger checkpoint creation by sending the message directly - send(persistence_writer, :create_checkpoint) + # Started directly (no DocumentSupervisor owner hook), so grant the writer + # this test's sandbox access before it writes the checkpoint. + allow_collaboration_process(persistence_writer) - # Wait for the checkpoint to be created - Process.sleep(100) + # Trigger checkpoint creation by sending the message directly, then drain + # the mailbox with a synchronous call. Messages are handled in order, so + # once :sys.get_state/1 returns, the :create_checkpoint DB insert has + # completed and no query is left in flight. + send(persistence_writer, :create_checkpoint) + :sys.get_state(persistence_writer) # Verify checkpoint was created checkpoint = @@ -890,11 +1020,13 @@ defmodule Lightning.Collaboration.DocumentSupervisorTest do assert Yex.Map.fetch!(checkpoint_workflow_map, "lock_version") == 1 assert Yex.Map.fetch!(checkpoint_workflow_map, "concurrency") == 10 - # Clean up + # Stop synchronously so the writer is gone before this test process exits; + # start_supervised! is the safety net if an assertion above raises first. GenServer.stop(persistence_writer, :normal) end test "creates checkpoint merging existing checkpoint with updates", %{ + instance: instance, workflow: workflow } do document_name = "workflow:#{workflow.id}" @@ -937,15 +1069,28 @@ defmodule Lightning.Collaboration.DocumentSupervisorTest do version: :update }) - # Start PersistenceWriter and trigger checkpoint - {:ok, persistence_writer} = - PersistenceWriter.start_link( - document_name: document_name, - name: Registry.via({:persistence_writer, document_name}) + # Start PersistenceWriter under ExUnit so it is torn down before the + # sandbox owner is stopped, even if an assertion below fails. + persistence_writer = + start_supervised!( + {PersistenceWriter, + document_name: document_name, + registry: instance.registry, + name: + Registry.via( + instance.registry, + {:persistence_writer, document_name} + )} ) + # Started directly (no DocumentSupervisor owner hook), so grant the writer + # this test's sandbox access before it writes the checkpoint. + allow_collaboration_process(persistence_writer) + + # Trigger the checkpoint, then drain the mailbox synchronously so the DB + # insert has finished (messages are handled in order) before we read back. send(persistence_writer, :create_checkpoint) - Process.sleep(100) + :sys.get_state(persistence_writer) # Get the new checkpoint (should be the most recent one) checkpoints = @@ -971,7 +1116,8 @@ defmodule Lightning.Collaboration.DocumentSupervisorTest do assert Yex.Map.fetch!(new_workflow_map, "name") == "Modified Name" assert Yex.Map.fetch!(new_workflow_map, "lock_version") == 2 - # Clean up + # Stop synchronously so the writer is gone before this test process exits; + # start_supervised! is the safety net if an assertion above raises first. GenServer.stop(persistence_writer, :normal) end end diff --git a/test/lightning/collaboration/no_change_snapshot_test.exs b/test/lightning/collaboration/no_change_snapshot_test.exs index d651209cda3..61a065d72e8 100644 --- a/test/lightning/collaboration/no_change_snapshot_test.exs +++ b/test/lightning/collaboration/no_change_snapshot_test.exs @@ -3,68 +3,61 @@ defmodule Lightning.Collaboration.NoChangeSnapshotTest do Tests to reproduce and verify the fix for phantom snapshot creation when no actual changes are made to a workflow. """ - use Lightning.DataCase, async: false + use Lightning.DataCase, async: true import Lightning.Factories + import Lightning.CollaborationHelpers - alias Lightning.Collaboration.{DocumentSupervisor, Session} + alias Lightning.Collaboration.Session alias Lightning.Workflows describe "saving without changes" do setup do - # Set global mode for the mock to allow cross-process calls - Mox.set_mox_global(LightningMock) - # Stub the broadcast calls that save_workflow makes + # Each test drives its own isolated collaboration tree (Registry, + # DynamicSupervisor, and `:pg` scope), so concurrent tests can't see each + # other's documents. + # + # Stub the broadcast calls that save_workflow makes. save_workflow runs in + # the Session process; tests that exercise it allow that process into this + # test's mocks/sandbox, and the DocumentSupervisor's spawned children are + # granted access by the owner-anchored startup hook via `owner: self()`. Mox.stub(LightningMock, :broadcast, fn _topic, _message -> :ok end) + instance = start_collaboration_instance() + user = insert(:user) project = insert(:project) workflow = insert(:workflow, name: "Test Workflow", project: project) job = insert(:job, workflow: workflow, name: "Test Job") - %{user: user, project: project, workflow: workflow, job: job} + %{ + instance: instance, + user: user, + project: project, + workflow: workflow, + job: job + } end test "does not create snapshot when saving Y.Doc with no changes", %{ + instance: instance, user: user, workflow: workflow } do - # Start document and session - start_supervised!( - {DocumentSupervisor, - workflow: workflow, document_name: "workflow:#{workflow.id}"} - ) - - session_pid = - start_supervised!( - {Session, - workflow: workflow, - user: user, - document_name: "workflow:#{workflow.id}"} - ) + session_pid = start_session(instance, workflow, user) # Get initial lock_version workflow = Workflows.get_workflow!(workflow.id) initial_lock_version = workflow.lock_version # Count initial snapshots - initial_snapshot_count = - Lightning.Repo.all( - from s in Lightning.Workflows.Snapshot, - where: s.workflow_id == ^workflow.id - ) - |> length() + initial_snapshot_count = snapshot_count(workflow.id) # Save without making any changes to the Y.Doc {:ok, saved_workflow} = Session.save_workflow(session_pid, user) # Verify no snapshot was created - final_snapshot_count = - Lightning.Repo.all( - from s in Lightning.Workflows.Snapshot, - where: s.workflow_id == ^workflow.id - ) - |> length() + final_snapshot_count = snapshot_count(workflow.id) assert saved_workflow.lock_version == initial_lock_version, "Lock version should not increment without changes" @@ -74,34 +67,18 @@ defmodule Lightning.Collaboration.NoChangeSnapshotTest do end test "creates snapshot when actually changing workflow data", %{ + instance: instance, user: user, workflow: workflow } do - # Start document and session - start_supervised!( - {DocumentSupervisor, - workflow: workflow, document_name: "workflow:#{workflow.id}"} - ) - - session_pid = - start_supervised!( - {Session, - workflow: workflow, - user: user, - document_name: "workflow:#{workflow.id}"} - ) + session_pid = start_session(instance, workflow, user) # Get initial lock_version workflow = Workflows.get_workflow!(workflow.id) initial_lock_version = workflow.lock_version # Count initial snapshots - initial_snapshot_count = - Lightning.Repo.all( - from s in Lightning.Workflows.Snapshot, - where: s.workflow_id == ^workflow.id - ) - |> length() + initial_snapshot_count = snapshot_count(workflow.id) # Make a real change to the workflow doc = Session.get_doc(session_pid) @@ -115,12 +92,7 @@ defmodule Lightning.Collaboration.NoChangeSnapshotTest do {:ok, saved_workflow} = Session.save_workflow(session_pid, user) # Verify snapshot WAS created - final_snapshot_count = - Lightning.Repo.all( - from s in Lightning.Workflows.Snapshot, - where: s.workflow_id == ^workflow.id - ) - |> length() + final_snapshot_count = snapshot_count(workflow.id) assert saved_workflow.lock_version == initial_lock_version + 1, "Lock version should increment with changes" @@ -155,4 +127,36 @@ defmodule Lightning.Collaboration.NoChangeSnapshotTest do "Round-trip serialization should not create phantom changes" end end + + # Start the document under the isolated instance (owner: self() ties its + # lifetime to this test) and a Session that joins it, granting that Session + # the per-test sandbox/mock access it needs to save. + defp start_session(instance, workflow, user) do + document_name = "workflow:#{workflow.id}" + + {:ok, _doc_supervisor} = + start_collaboration_document(instance, workflow, document_name) + + session_pid = + start_supervised!( + {Session, + workflow: workflow, + user: user, + document_name: document_name, + registry: instance.registry, + pg_scope: instance.pg_scope} + ) + + allow_collaboration_process(session_pid) + + session_pid + end + + defp snapshot_count(workflow_id) do + Lightning.Repo.all( + from s in Lightning.Workflows.Snapshot, + where: s.workflow_id == ^workflow_id + ) + |> length() + end end diff --git a/test/lightning/collaboration/persistence_test.exs b/test/lightning/collaboration/persistence_test.exs index 210a118e950..5bfa357e8eb 100644 --- a/test/lightning/collaboration/persistence_test.exs +++ b/test/lightning/collaboration/persistence_test.exs @@ -1,11 +1,12 @@ defmodule Lightning.Collaboration.PersistenceTest do - use Lightning.DataCase, async: false + use Lightning.DataCase, async: true alias Lightning.Collaboration.DocumentState alias Lightning.Collaboration.DocumentSupervisor alias Lightning.Collaboration.Registry import Lightning.Factories + import Lightning.CollaborationHelpers @moduledoc """ Tests for Lightning.Collaboration.Persistence module. @@ -14,9 +15,50 @@ defmodule Lightning.Collaboration.PersistenceTest do reconciling Y.Doc state from the database when a DocumentSupervisor starts. """ + # Each test drives its own isolated collaboration tree (Registry, + # DynamicSupervisor, and `:pg` scope), so concurrent tests can't collide on + # the application-wide singletons. Documents are started under that tree with + # `owner: self()`, which (a) lets the SharedDoc's init-time DB read and the + # children's later writes reach this test's sandbox connection, and (b) ties + # the tree's lifetime to this test. The instance supervisor is + # `start_supervised!`-owned, so its DB-writing children are stopped — flush + # included via DocumentSupervisor.terminate/2 — before this test process (the + # sandbox owner) exits, even if an assertion raises first. setup do Process.flag(:trap_exit, true) - :ok + instance = start_collaboration_instance() + {:ok, instance: instance} + end + + # Start a DocumentSupervisor under the test's isolated instance. owner: self() + # threads this test's sandbox/mock access into the spawned SharedDoc and + # PersistenceWriter (and into the SharedDoc's init-time read via the + # persistence state). + # + # Started under `start_supervised!` so ExUnit owns it: on test exit ExUnit + # synchronously stops it, running DocumentSupervisor.terminate/2 (which flushes + # the PersistenceWriter through the SharedDoc) to completion. That teardown is + # registered after DataCase's `stop_owner` and runs LIFO, so it executes while + # the sandbox owner is still alive — no DB-writing child is left mid-query when + # the connection is checked back in. + defp start_document(instance, workflow, document_name) do + start_supervised!( + {DocumentSupervisor, + workflow: workflow, + document_name: document_name, + registry: instance.registry, + pg_scope: instance.pg_scope, + owner: self(), + name: Registry.via(instance.registry, {:doc_supervisor, document_name})} + ) + end + + defp shared_doc_map(instance, document_name) do + shared_doc = + Registry.whereis(instance.registry, {:shared_doc, document_name}) + + doc = Yex.Sync.SharedDoc.get_doc(shared_doc) + Yex.Doc.get_map(doc, "workflow") end describe "reconcile_workflow_metadata/2" do @@ -30,6 +72,7 @@ defmodule Lightning.Collaboration.PersistenceTest do end test "converts deleted_at DateTime to string when reconciling", %{ + instance: instance, workflow: workflow, document_name: document_name } do @@ -59,18 +102,13 @@ defmodule Lightning.Collaboration.PersistenceTest do # Start DocumentSupervisor with workflow that has deleted_at # This triggers reconcile_workflow_metadata - {:ok, doc_supervisor} = - DocumentSupervisor.start_link( - [workflow: workflow_with_deleted, document_name: document_name], - name: Registry.via({:doc_supervisor, document_name}) - ) + doc_supervisor = + start_document(instance, workflow_with_deleted, document_name) assert Process.alive?(doc_supervisor) # Verify the deleted_at was properly converted to a string in Y.Doc - shared_doc = Registry.whereis({:shared_doc, document_name}) - doc = Yex.Sync.SharedDoc.get_doc(shared_doc) - reconciled_workflow_map = Yex.Doc.get_map(doc, "workflow") + reconciled_workflow_map = shared_doc_map(instance, document_name) deleted_at_value = Yex.Map.fetch!(reconciled_workflow_map, "deleted_at") @@ -80,12 +118,10 @@ defmodule Lightning.Collaboration.PersistenceTest do # Should match the original DateTime when parsed back assert {:ok, parsed_dt, _} = DateTime.from_iso8601(deleted_at_value) assert DateTime.compare(parsed_dt, workflow_with_deleted.deleted_at) == :eq - - # Clean up - GenServer.stop(doc_supervisor, :normal) end test "handles nil deleted_at correctly", %{ + instance: instance, workflow: workflow, document_name: document_name } do @@ -113,27 +149,20 @@ defmodule Lightning.Collaboration.PersistenceTest do }) # Start DocumentSupervisor - {:ok, doc_supervisor} = - DocumentSupervisor.start_link( - [workflow: workflow_without_deleted, document_name: document_name], - name: Registry.via({:doc_supervisor, document_name}) - ) + doc_supervisor = + start_document(instance, workflow_without_deleted, document_name) assert Process.alive?(doc_supervisor) # Verify deleted_at remains nil - shared_doc = Registry.whereis({:shared_doc, document_name}) - doc = Yex.Sync.SharedDoc.get_doc(shared_doc) - reconciled_workflow_map = Yex.Doc.get_map(doc, "workflow") + reconciled_workflow_map = shared_doc_map(instance, document_name) deleted_at_value = Yex.Map.fetch!(reconciled_workflow_map, "deleted_at") assert deleted_at_value == nil - - # Clean up - GenServer.stop(doc_supervisor, :normal) end test "reconciles lock_version when persisted state exists", %{ + instance: instance, workflow: workflow, document_name: document_name } do @@ -168,29 +197,21 @@ defmodule Lightning.Collaboration.PersistenceTest do }) # Start DocumentSupervisor - should reconcile to new lock_version - {:ok, doc_supervisor} = - DocumentSupervisor.start_link( - [workflow: updated_workflow, document_name: document_name], - name: Registry.via({:doc_supervisor, document_name}) - ) + doc_supervisor = start_document(instance, updated_workflow, document_name) assert Process.alive?(doc_supervisor) # Verify lock_version was reconciled to the current DB value - shared_doc = Registry.whereis({:shared_doc, document_name}) - doc = Yex.Sync.SharedDoc.get_doc(shared_doc) - reconciled_workflow_map = Yex.Doc.get_map(doc, "workflow") + reconciled_workflow_map = shared_doc_map(instance, document_name) reconciled_lock_version = Yex.Map.fetch!(reconciled_workflow_map, "lock_version") assert reconciled_lock_version == new_lock_version - - # Clean up - GenServer.stop(doc_supervisor, :normal) end test "reconciles workflow name when it changed", %{ + instance: instance, workflow: workflow, document_name: document_name } do @@ -221,61 +242,45 @@ defmodule Lightning.Collaboration.PersistenceTest do }) # Start DocumentSupervisor - {:ok, doc_supervisor} = - DocumentSupervisor.start_link( - [workflow: updated_workflow, document_name: document_name], - name: Registry.via({:doc_supervisor, document_name}) - ) + doc_supervisor = start_document(instance, updated_workflow, document_name) assert Process.alive?(doc_supervisor) # Verify name was reconciled - shared_doc = Registry.whereis({:shared_doc, document_name}) - doc = Yex.Sync.SharedDoc.get_doc(shared_doc) - reconciled_workflow_map = Yex.Doc.get_map(doc, "workflow") + reconciled_workflow_map = shared_doc_map(instance, document_name) reconciled_name = Yex.Map.fetch!(reconciled_workflow_map, "name") assert reconciled_name == "Updated Name" - - # Clean up - GenServer.stop(doc_supervisor, :normal) end end describe "bind/3 with no persisted state" do - test "initializes workflow document from database" do + test "initializes workflow document from database", %{instance: instance} do workflow = insert(:workflow) document_name = "workflow:#{workflow.id}" # Don't create any persisted state - fresh start # Start DocumentSupervisor - {:ok, doc_supervisor} = - DocumentSupervisor.start_link( - [workflow: workflow, document_name: document_name], - name: Registry.via({:doc_supervisor, document_name}) - ) + doc_supervisor = start_document(instance, workflow, document_name) assert Process.alive?(doc_supervisor) # Verify workflow was initialized from database - shared_doc = Registry.whereis({:shared_doc, document_name}) - doc = Yex.Sync.SharedDoc.get_doc(shared_doc) - workflow_map = Yex.Doc.get_map(doc, "workflow") + workflow_map = shared_doc_map(instance, document_name) assert Yex.Map.fetch!(workflow_map, "id") == workflow.id assert Yex.Map.fetch!(workflow_map, "name") == workflow.name assert Yex.Map.fetch!(workflow_map, "lock_version") == workflow.lock_version - - # Clean up - GenServer.stop(doc_supervisor, :normal) end end describe "bind/3 with stale persisted state" do - test "resets Y.Doc when persisted lock_version differs from database" do + test "resets Y.Doc when persisted lock_version differs from database", %{ + instance: instance + } do workflow = insert(:workflow, lock_version: 5) document_name = "workflow:#{workflow.id}" @@ -300,26 +305,17 @@ defmodule Lightning.Collaboration.PersistenceTest do }) # Start DocumentSupervisor - {:ok, doc_supervisor} = - DocumentSupervisor.start_link( - [workflow: workflow, document_name: document_name], - name: Registry.via({:doc_supervisor, document_name}) - ) + doc_supervisor = start_document(instance, workflow, document_name) assert Process.alive?(doc_supervisor) # Verify Y.Doc was reset to current database state - shared_doc = Registry.whereis({:shared_doc, document_name}) - doc = Yex.Sync.SharedDoc.get_doc(shared_doc) - workflow_map = Yex.Doc.get_map(doc, "workflow") + workflow_map = shared_doc_map(instance, document_name) # Should have current lock_version, not the old one assert Yex.Map.fetch!(workflow_map, "lock_version") == 5 # Should have current name, not the old one assert Yex.Map.fetch!(workflow_map, "name") == workflow.name - - # Clean up - GenServer.stop(doc_supervisor, :normal) end end end diff --git a/test/lightning/collaboration/registry_test.exs b/test/lightning/collaboration/registry_test.exs index ab281136758..82bf98648d7 100644 --- a/test/lightning/collaboration/registry_test.exs +++ b/test/lightning/collaboration/registry_test.exs @@ -1,64 +1,73 @@ defmodule Lightning.Collaboration.RegistryTest do - use Lightning.DataCase, async: false + use Lightning.DataCase, async: true alias Lightning.Collaboration.Registry setup do - # The Registry is started by the collaboration supervisor - # which should be running in test - :ok + # Each test runs against its own isolated Registry instance rather than the + # application-wide singleton, so concurrent tests can't see or collide with + # each other's registrations. The registry name is threaded into every + # Registry call below. + name = :"registry_test_#{System.unique_integer([:positive])}" + + start_supervised!(%{ + id: name, + start: {Elixir.Registry, :start_link, [[keys: :unique, name: name]]} + }) + + %{registry: name} end describe "Registry" do - test "registers and looks up processes" do + test "registers and looks up processes", %{registry: registry} do # Register current process with a session key workflow_id = Ecto.UUID.generate() user_id = Ecto.UUID.generate() session_key = {:session, workflow_id, user_id} - assert {:ok, pid} = Registry.register(session_key) + assert {:ok, pid} = Registry.register(registry, session_key) assert pid == self() # Verify lookup finds the process - assert [{^pid, nil}] = Registry.lookup(session_key) + assert [{^pid, nil}] = Registry.lookup(registry, session_key) # Verify whereis returns the pid - assert ^pid = Registry.whereis(session_key) + assert ^pid = Registry.whereis(registry, session_key) end - test "returns nil for non-existent keys" do + test "returns nil for non-existent keys", %{registry: registry} do non_existent_key = {:session, "nonexistent", "user"} - assert [] = Registry.lookup(non_existent_key) - assert nil == Registry.whereis(non_existent_key) + assert [] = Registry.lookup(registry, non_existent_key) + assert nil == Registry.whereis(registry, non_existent_key) end - test "prevents duplicate registrations" do + test "prevents duplicate registrations", %{registry: registry} do workflow_id = Ecto.UUID.generate() user_id = Ecto.UUID.generate() session_key = {:session, workflow_id, user_id} # First and second registration succeeds - assert {:ok, pid} = Registry.register(session_key) + assert {:ok, pid} = Registry.register(registry, session_key) assert {:error, {:already_registered, ^pid}} = - Registry.register(session_key) + Registry.register(registry, session_key) end - test "supports different key patterns" do + test "supports different key patterns", %{registry: registry} do workflow_id = Ecto.UUID.generate() user_id = Ecto.UUID.generate() doc_name = "workflow:#{workflow_id}" # Register with session key session_key = {:session, workflow_id, user_id} - assert {:ok, _} = Registry.register(session_key) + assert {:ok, _} = Registry.register(registry, session_key) # Can also register with different key type (though this would be different process) shared_doc_key = {:shared_doc, doc_name} # Since we can't register the same process twice, verify the key format works - assert [] = Registry.lookup(shared_doc_key) + assert [] = Registry.lookup(registry, shared_doc_key) end end end diff --git a/test/lightning/collaboration/session_test.exs b/test/lightning/collaboration/session_test.exs index 3d70bbb53c5..6270a791f99 100644 --- a/test/lightning/collaboration/session_test.exs +++ b/test/lightning/collaboration/session_test.exs @@ -1,11 +1,5 @@ defmodule Lightning.SessionTest do - # We assume that the WorkflowCollaboration supervisor is up - # that starts :pg with the :workflow_collaboration scope - # and a dynamic supervisor called Lightning.WorkflowCollaboration - - # Tests must be async: false, some of the processes we start are either - # not owned by the test process, or themselves start processes. - use Lightning.DataCase, async: false + use Lightning.DataCase, async: true import Eventually import Lightning.Factories @@ -14,7 +8,6 @@ defmodule Lightning.SessionTest do alias Lightning.Collaboration.DocumentState alias Lightning.Collaboration.DocumentSupervisor - # alias Lightning.Collaboration.PersistenceWriter alias Lightning.Collaboration.Registry alias Lightning.Collaboration.Session alias Lightning.Collaboration.TestClient @@ -22,15 +15,107 @@ defmodule Lightning.SessionTest do require Logger + # Each test drives its own isolated collaboration tree (Registry, + # DynamicSupervisor, and `:pg` scope), so concurrent tests can't see or + # collide with each other's documents, sessions, or process-group members. + # Documents and sessions are started under that instance with `owner: self()` + # and `start_supervised!`, so the DB-writing SharedDoc/PersistenceWriter + # children are flushed and stopped — via DocumentSupervisor.terminate/2 — before + # this test process (the sandbox owner) exits, even if an assertion raises. setup do + instance = start_collaboration_instance() user = insert(:user) - {:ok, user: user} + {:ok, instance: instance, user: user} end setup :verify_on_exit! + # Start a DocumentSupervisor under the test's isolated instance, owned by the + # test (owner: self()) and supervised by ExUnit for deterministic, + # flush-inclusive teardown before the sandbox owner exits. Extra opts (e.g. + # workflow overrides) are merged in. + defp start_document_supervisor(instance, workflow, document_name, opts \\ []) do + # Default `auto_exit: false` so the SharedDoc does not self-exit the instant + # its last observer leaves; teardown is driven by an explicit, in-body + # `drain_document/2` (`:normal`, flush-inclusive) rather than racing an + # asynchronous self-exit. Tests that assert the auto-exit-on-last-observer + # behaviour pass `auto_exit: true`. + # + # Started under `start_supervised!` so its slot is owned and bounded by the + # test, and `owner: self()` grants the children this test's sandbox/mock + # access. Each test drains the document explicitly before returning, so the + # tree is already gone (flush complete, connections checked in) before either + # ExUnit's supervised teardown or DataCase's `stop_owner` runs. + start_supervised!( + {DocumentSupervisor, + workflow: workflow, + document_name: document_name, + registry: instance.registry, + pg_scope: instance.pg_scope, + owner: self(), + auto_exit: Keyword.get(opts, :auto_exit, false), + name: Registry.via(instance.registry, {:doc_supervisor, document_name})} + ) + end + + # Synchronously stop a document's whole tree (SharedDoc flush + PersistenceWriter) + # in the test body, before this test process — the sandbox owner — returns. + # `Collaborate.stop_document/2` issues a `:normal` `GenServer.stop`, so + # `DocumentSupervisor.terminate/2` runs the final flush and every DB write has + # checked its connection back in by the time this returns. Idempotent. + defp drain_document(instance, document_name) do + # Synchronous, flush-inclusive `:normal` stop of the whole document tree. + Lightning.Collaborate.stop_document(instance, document_name) + + # `stop_document/2` is a no-op if the SharedDoc had already auto-exited; in + # that case the cascade (DocumentSupervisor + PersistenceWriter `:normal` + # stop, flush included) is in flight. Wait for the writer to be fully gone so + # its flush has completed and its connection is checked back in before this + # test — the sandbox owner — returns. + await_document_drained(instance, document_name) + end + + # For tests that intentionally let the SharedDoc auto-exit when its last + # observer leaves: stopping the SharedDoc cascades a `:normal` stop of the + # DocumentSupervisor and PersistenceWriter, whose final flush is asynchronous. + # Wait until the PersistenceWriter is fully gone from the registry — i.e. it + # has flushed and checked its connection back in — before the test (the sandbox + # owner) returns, so no DB write is in flight at teardown. + defp await_document_drained(instance, document_name) do + refute_eventually( + Registry.whereis(instance.registry, {:persistence_writer, document_name}) + ) + + refute_eventually( + Registry.whereis(instance.registry, {:doc_supervisor, document_name}) + ) + end + + # Start a Session under the test's isolated instance, pointed at that + # instance's `:pg` scope so it resolves the SharedDoc the DocumentSupervisor + # registered. Supervised by ExUnit. Extra opts are merged (e.g. parent_pid). + defp start_session_proc(instance, workflow, user, document_name, opts \\ []) do + base = [ + workflow: workflow, + user: user, + document_name: document_name, + registry: instance.registry, + pg_scope: instance.pg_scope, + name: + Registry.via( + instance.registry, + {:session, document_name, user.id, System.unique_integer([:positive])} + ) + ] + + start_supervised!({Session, Keyword.merge(base, opts)}, + id: {Session, document_name, user.id, System.unique_integer([:positive])} + ) + end + describe "start/1" do test "start_link/1 returns an error when the SharedDoc doesn't exist", %{ + instance: instance, user: user } do workflow_id = Ecto.UUID.generate() @@ -47,17 +132,22 @@ defmodule Lightning.SessionTest do {Session, user: user, workflow: workflow, + pg_scope: instance.pg_scope, document_name: "workflow:#{workflow.id}"} ) end - test "start/1 can join an existing shared doc", %{user: user1} do + test "start/1 can join an existing shared doc", %{ + instance: instance, + user: user1 + } do user2 = insert(:user) workflow = insert(:simple_workflow) - Lightning.Collaborate.start_document( - workflow, - "workflow:#{workflow.id}" + # auto_exit: true — this test asserts the SharedDoc self-exits once its + # last observer (session) leaves. + start_document_supervisor(instance, workflow, "workflow:#{workflow.id}", + auto_exit: true ) [parent1, parent2] = build_parents(2) @@ -69,6 +159,7 @@ defmodule Lightning.SessionTest do user: user, workflow: workflow, parent_pid: parent, + pg_scope: instance.pg_scope, document_name: "workflow:#{workflow.id}" ) @@ -92,7 +183,7 @@ defmodule Lightning.SessionTest do end) shared_doc_pid = - Registry.get_group("workflow:#{workflow.id}") + Registry.get_group(instance.registry, "workflow:#{workflow.id}") |> Map.get(:shared_doc) observer_processes = @@ -115,7 +206,7 @@ defmodule Lightning.SessionTest do document_name = "workflow:#{workflow.id}" assert_eventually( - length(:pg.get_members(:workflow_collaboration, document_name)) == 1 + length(:pg.get_members(instance.pg_scope, document_name)) == 1 ) Process.exit(parent2, :normal) @@ -131,31 +222,31 @@ defmodule Lightning.SessionTest do # But we might want to control the cleanup ourselves, in which case # this will be > 0 until we stop the SharedDoc ourselves. assert_eventually( - length(:pg.get_members(:workflow_collaboration, workflow.id)) == 0 + length(:pg.get_members(instance.pg_scope, workflow.id)) == 0 ) + + # The SharedDoc auto-exited above; wait for its PersistenceWriter to finish + # flushing and exit before this test (the sandbox owner) returns. + await_document_drained(instance, document_name) end end describe "workflow initialization" do - test "SharedDoc is initialized with workflow data", %{user: user} do + test "SharedDoc is initialized with workflow data", %{ + instance: instance, + user: user + } do # Create a workflow with jobs workflow = build(:complex_workflow, name: "Test Workflow") |> insert() - start_supervised!( - {DocumentSupervisor, - workflow: workflow, document_name: "workflow:#{workflow.id}"} - ) + document_name = "workflow:#{workflow.id}" + + start_document_supervisor(instance, workflow, document_name) # Start a session - this should initialize the SharedDoc with workflow data - session_pid = - start_supervised!( - {Session, - user: user, - workflow: workflow, - document_name: "workflow:#{workflow.id}"} - ) + session_pid = start_session_proc(instance, workflow, user, document_name) # Send a message to allow :handle_continue to finish shared_doc = Session.get_doc(session_pid) @@ -233,26 +324,27 @@ defmodule Lightning.SessionTest do "Trigger #{key} mismatch: expected #{expected_value |> inspect}, got #{doc_value |> inspect}" end) end + + drain_document(instance, document_name) end - test "existing SharedDoc is not reinitialized", %{user: user} do + test "existing SharedDoc is not reinitialized", %{ + instance: instance, + user: user + } do workflow = insert(:workflow, name: "Test Workflow") + document_name = "workflow:#{workflow.id}" insert(:job, workflow: workflow, name: "Original Job", body: "original") - start_supervised!( - {DocumentSupervisor, - workflow: workflow, document_name: "workflow:#{workflow.id}"} + # auto_exit: true — this test asserts the SharedDoc dies after both + # sessions stop. + start_document_supervisor(instance, workflow, document_name, + auto_exit: true ) # Start first session - session_1 = - start_supervised!( - {Session, - workflow: workflow, - user: user, - document_name: "workflow:#{workflow.id}"} - ) + session_1 = start_session_proc(instance, workflow, user, document_name) shared_doc_1 = Session.get_doc(session_1) @@ -262,13 +354,7 @@ defmodule Lightning.SessionTest do end) # Start second session - should connect to existing SharedDoc - session_2 = - start_supervised!( - {Session, - workflow: workflow, - user: user, - document_name: "workflow:#{workflow.id}"} - ) + session_2 = start_session_proc(instance, workflow, user, document_name) shared_doc_2 = Session.get_doc(session_2) @@ -280,16 +366,23 @@ defmodule Lightning.SessionTest do %Session{shared_doc_pid: shared_doc_pid} = :sys.get_state(session_1) - # # TODO: probably not needed anymore since we're using start_supervised! Session.stop(session_1) Session.stop(session_2) refute_eventually(Process.alive?(shared_doc_pid)) + + # The SharedDoc auto-exited; wait for its PersistenceWriter to flush and + # exit before this test (the sandbox owner) returns. + await_document_drained(instance, document_name) end - test "client can sync workflow data from SharedDoc", %{user: user} do + test "client can sync workflow data from SharedDoc", %{ + instance: instance, + user: user + } do # Create workflow with jobs workflow = insert(:workflow, name: "Sync Test Workflow") + document_name = "workflow:#{workflow.id}" job = insert(:job, @@ -298,19 +391,14 @@ defmodule Lightning.SessionTest do body: "console.log('sync')" ) - start_supervised!( - {DocumentSupervisor, - workflow: workflow, document_name: "workflow:#{workflow.id}"} + # auto_exit: true — this test asserts the SharedDoc dies after the session + # stops. + start_document_supervisor(instance, workflow, document_name, + auto_exit: true ) # Start session to initialize SharedDoc - session_pid = - start_supervised!( - {Session, - user: user, - workflow: workflow, - document_name: "workflow:#{workflow.id}"} - ) + session_pid = start_session_proc(instance, workflow, user, document_name) %Session{shared_doc_pid: shared_doc_pid} = :sys.get_state(session_pid) @@ -343,34 +431,29 @@ defmodule Lightning.SessionTest do Session.stop(session_pid) refute_eventually(Process.alive?(shared_doc_pid)) + + await_document_drained(instance, document_name) end end describe "persistence" do # @tag :pick - test "saves document state to the database", %{user: user} do + test "saves document state to the database", %{ + instance: instance, + user: user + } do workflow = insert(:simple_workflow) + document_name = "workflow:#{workflow.id}" - _document_supervisor = - start_supervised!( - {DocumentSupervisor, - workflow: workflow, document_name: "workflow:#{workflow.id}"} - ) + start_document_supervisor(instance, workflow, document_name) - session_pid = - start_supervised!( - {Session, - user: user, - workflow: workflow, - document_name: "workflow:#{workflow.id}"} - ) + session_pid = start_session_proc(instance, workflow, user, document_name) # This is an existing workflow, so when the session starts, it should # both initialize the workflow document and save the initial state # to the database. workflow_id = workflow.id - document_name = "workflow:#{workflow_id}" expected_workflow = %{ "id" => workflow_id, @@ -389,50 +472,58 @@ defmodule Lightning.SessionTest do # Lets find the PersistenceWriter and check it's state - persistence_writer = get_persistence_writer(document_name) + persistence_writer = get_persistence_writer(instance, document_name) # There should be 1 pending update - assert get_pending_updates(document_name) |> length() == 1 + assert get_pending_updates(instance, document_name) |> length() == 1 assert get_document_state(document_name) |> length() == 0, "Nothing is expected in the database yet" # Now lets add a job add_job(session_pid) - assert get_pending_updates(document_name) |> length() == 2 + assert get_pending_updates(instance, document_name) |> length() == 2 # And another job = string_params_for(:job) add_job(session_pid, job) - assert get_pending_updates(document_name) |> length() == 3 + assert get_pending_updates(instance, document_name) |> length() == 3 # And force saving the updates (this normally happens on a timer) send(persistence_writer, :force_save) - assert_eventually(get_pending_updates(document_name) |> length() == 0) + assert_eventually( + get_pending_updates(instance, document_name) |> length() == 0 + ) # And remove a job remove_job(session_pid, job) - assert get_pending_updates(document_name) |> length() == 1 + assert get_pending_updates(instance, document_name) |> length() == 1 send(persistence_writer, :force_save) - assert_eventually(get_pending_updates(document_name) |> length() == 0) + assert_eventually( + get_pending_updates(instance, document_name) |> length() == 0 + ) # And check that the document state is in the database assert get_document_state(document_name) |> length() == 2 # TODO: Recover from state without a checkpoint # TODO: Recover from state with a checkpoint + + # Synchronously drain so the PersistenceWriter's pending writes are flushed + # and its connection checked in before this test (the sandbox owner) exits. + Lightning.Collaborate.stop_document(instance, document_name) end - defp get_persistence_writer(document_name) do - Registry.get_group(document_name) + defp get_persistence_writer(instance, document_name) do + Registry.get_group(instance.registry, document_name) |> Map.get(:persistence_writer) end - defp get_pending_updates(document_name) do - persistence_writer = get_persistence_writer(document_name) + defp get_pending_updates(instance, document_name) do + persistence_writer = get_persistence_writer(instance, document_name) :sys.get_state(persistence_writer).pending_updates end @@ -469,29 +560,39 @@ defmodule Lightning.SessionTest do # from persistence. @tag :pick - test "client doc is still around", %{user: user} do + test "client doc is still around", %{instance: instance, user: user} do workflow = insert(:simple_workflow) + document_name = "workflow:#{workflow.id}" + # auto_exit: true — this test asserts the SharedDoc/PersistenceWriter/ + # DocumentSupervisor die after the session is killed and the test client + # unobserves. document_supervisor = - start_supervised!( - {DocumentSupervisor, - workflow: workflow, document_name: "workflow:#{workflow.id}"} + start_document_supervisor(instance, workflow, document_name, + auto_exit: true ) %{shared_doc: shared_doc, persistence_writer: persistence_writer} = - Registry.get_group("workflow:#{workflow.id}") + Registry.get_group(instance.registry, document_name) - session_pid = - start_supervised!( - {Session, - user: user, - workflow: workflow, - document_name: "workflow:#{workflow.id}"} - ) + session_pid = start_session_proc(instance, workflow, user, document_name) {:ok, client_pid} = GenServer.start(TestClient, shared_doc_pid: shared_doc) + on_exit(fn -> + # The TestClient holds its own Y.Doc and may have already lost its + # SharedDoc by the end of this reconnection test; tolerate its + # terminate-time unobserve failing against a dead doc. + if Process.alive?(client_pid) do + try do + GenServer.stop(client_pid) + catch + :exit, _ -> :ok + end + end + end) + # Ensure handle_continue has finished :sys.get_state(client_pid) @@ -517,7 +618,7 @@ defmodule Lightning.SessionTest do assert Process.alive?(client_pid), "Client should still be alive" - assert get_document_state("workflow:#{workflow.id}"), + assert get_document_state(document_name), "DocumentState should be saved in the database" # Client adds another job, while the SharedDoc is not around @@ -526,21 +627,13 @@ defmodule Lightning.SessionTest do # Starting a new document supervisor, like when the frontend reconnects # At this point, client is still running, and the SharedDoc should # pick up the existing document from the database. - start_supervised!( - {DocumentSupervisor, - workflow: workflow, document_name: "workflow:#{workflow.id}"} - ) + start_document_supervisor(instance, workflow, document_name) # Starting a new session - _session_pid = - start_supervised!( - {Session, - user: user, - workflow: workflow, - document_name: "workflow:#{workflow.id}"} - ) + _session_pid = start_session_proc(instance, workflow, user, document_name) - shared_doc_pid = Registry.get_group("workflow:#{workflow.id}").shared_doc + shared_doc_pid = + Registry.get_group(instance.registry, document_name).shared_doc GenServer.call(client_pid, {:observe, shared_doc_pid}) @@ -548,6 +641,10 @@ defmodule Lightning.SessionTest do assert length(jobs) == 3 assert get_jobs(shared_doc_pid) |> length() == 3 + + # Drain the reconnected document synchronously before returning so its + # PersistenceWriter flush completes while the sandbox owner is still alive. + Lightning.Collaborate.stop_document(instance, document_name) end end @@ -587,7 +684,7 @@ defmodule Lightning.SessionTest do describe "teardown" do @tag :capture_log - test "when a session is stopped", %{user: user1} do + test "when a session is stopped", %{instance: instance, user: user1} do workflow_id = Ecto.UUID.generate() workflow = %Lightning.Workflows.Workflow{ @@ -597,12 +694,15 @@ defmodule Lightning.SessionTest do positions: %{} } + document_name = "workflow:#{workflow.id}" + user2 = insert(:user) user3 = insert(:user) - start_supervised!( - {DocumentSupervisor, - workflow: workflow, document_name: "workflow:#{workflow.id}"} + # auto_exit: true — this test asserts the SharedDoc dies once the last + # session is stopped. + start_document_supervisor(instance, workflow, document_name, + auto_exit: true ) [{client1, parent1}, {client2, parent2}, {client3, _parent3}] = @@ -610,12 +710,8 @@ defmodule Lightning.SessionTest do parent = build_parent() client = - start_supervised!( - {Session, - user: user, - workflow: workflow, - parent_pid: parent, - document_name: "workflow:#{workflow.id}"} + start_session_proc(instance, workflow, user, document_name, + parent_pid: parent ) {client, parent} @@ -654,6 +750,10 @@ defmodule Lightning.SessionTest do # |> Map.get(:observer_process) == # %{} # ) + + # The SharedDoc auto-exited; wait for its PersistenceWriter to flush and + # exit before this test (the sandbox owner) returns. + await_document_drained(instance, document_name) end end @@ -725,45 +825,43 @@ defmodule Lightning.SessionTest do end describe "save_workflow/2" do - setup do - # Set global mode for the mock to allow cross-process calls - Mox.set_mox_global(LightningMock) - # Stub the broadcast calls that save_workflow makes + setup %{instance: instance} do + # Stub the broadcast calls that save_workflow makes. save_workflow runs in + # the Session process, which we allow into this test's mocks/sandbox below; + # the DocumentSupervisor's spawned children are granted access by the + # owner-anchored startup hook via `owner: self()`. Mox.stub(LightningMock, :broadcast, fn _topic, _message -> :ok end) user = insert(:user) project = insert(:project) workflow = insert(:workflow, name: "Original Name", project: project) + document_name = "workflow:#{workflow.id}" # Add a job so we have something to modify job = insert(:job, workflow: workflow, name: "Original Job") - start_supervised!( - {DocumentSupervisor, - workflow: workflow, document_name: "workflow:#{workflow.id}"} - ) + start_document_supervisor(instance, workflow, document_name) - session_pid = - start_supervised!( - {Session, - workflow: workflow, - user: user, - document_name: "workflow:#{workflow.id}"} - ) + session_pid = start_session_proc(instance, workflow, user, document_name) + + allow_collaboration_process(session_pid) %{ session: session_pid, user: user, workflow: workflow, job: job, - project: project + project: project, + document_name: document_name } end test "successfully saves workflow from Y.Doc", %{ session: session, user: user, - workflow: workflow + workflow: workflow, + instance: instance, + document_name: document_name } do # Modify Y.Doc via Session doc = Session.get_doc(session) @@ -784,9 +882,16 @@ defmodule Lightning.SessionTest do saved_from_db = Lightning.Workflows.get_workflow!(workflow.id) assert saved_from_db.name == "Updated Name" assert saved_from_db.lock_version == workflow.lock_version + 1 + + drain_document(instance, document_name) end - test "handles validation errors", %{session: session, user: user} do + test "handles validation errors", %{ + session: session, + user: user, + instance: instance, + document_name: document_name + } do # Set invalid data in Y.Doc (blank name) doc = Session.get_doc(session) @@ -800,12 +905,16 @@ defmodule Lightning.SessionTest do # Save should fail assert {:error, changeset} = Session.save_workflow(session, user) assert changeset.errors[:name] + + drain_document(instance, document_name) end test "handles workflow deleted error", %{ session: session, user: user, - workflow: workflow + workflow: workflow, + instance: instance, + document_name: document_name } do # Soft-delete the workflow Lightning.Repo.update!( @@ -816,13 +925,17 @@ defmodule Lightning.SessionTest do # Save should fail assert {:error, :workflow_deleted} = Session.save_workflow(session, user) + + drain_document(instance, document_name) end test "allows saving EXISTING workflow even when at activation limit", %{ session: session, user: user, project: project, - workflow: workflow + workflow: workflow, + instance: instance, + document_name: document_name } do # The workflow from setup was inserted via insert(), so it's :loaded (existing) assert workflow.__meta__.state == :loaded @@ -857,11 +970,15 @@ defmodule Lightning.SessionTest do # because their triggers are already counted in the limit assert {:ok, saved_workflow} = Session.save_workflow(session, user) assert saved_workflow.name == "Updated Name" + + drain_document(instance, document_name) end test "saves all workflow components correctly", %{ session: session, - user: user + user: user, + instance: instance, + document_name: document_name } do # Modify job name via Y.Doc doc = Session.get_doc(session) @@ -886,17 +1003,28 @@ defmodule Lightning.SessionTest do job_names = Enum.map(saved_from_db.jobs, & &1.name) assert "Modified Job" in job_names + + drain_document(instance, document_name) end - test "respects timeout for large workflows", %{session: session, user: user} do + test "respects timeout for large workflows", %{ + session: session, + user: user, + instance: instance, + document_name: document_name + } do # This test verifies the 10-second timeout is set # Actual timeout testing would require artificially slowing down the save assert {:ok, _workflow} = Session.save_workflow(session, user) + + drain_document(instance, document_name) end test "prevents circular reconciliation with skip_reconcile option", %{ session: session, - user: user + user: user, + instance: instance, + document_name: document_name } do # This test verifies that save_workflow passes skip_reconcile: true # to prevent WorkflowReconciler from updating the same Y.Doc @@ -904,12 +1032,16 @@ defmodule Lightning.SessionTest do # Mock or spy on Workflows.save_workflow to verify skip_reconcile is passed # For now, just verify save succeeds assert {:ok, _workflow} = Session.save_workflow(session, user) + + drain_document(instance, document_name) end test "handles concurrent saves with optimistic locking", %{ session: session, user: user, - workflow: workflow + workflow: workflow, + instance: instance, + document_name: document_name } do # Another process updates the workflow (simulating concurrent edit) {:ok, _updated} = @@ -929,14 +1061,18 @@ defmodule Lightning.SessionTest do {:ok, _} -> assert true {:error, _} -> assert true end + + drain_document(instance, document_name) end test "saves workflow with :built state and lock_version > 0", %{ + instance: instance, user: user, project: project } do # Create a new workflow struct (not yet saved) workflow_id = Ecto.UUID.generate() + document_name = "workflow:#{workflow_id}" new_workflow = %Workflow{ id: workflow_id, @@ -949,18 +1085,12 @@ defmodule Lightning.SessionTest do } # Start document and session with the new workflow - start_supervised!( - {DocumentSupervisor, - workflow: new_workflow, document_name: "workflow:#{workflow_id}"} - ) + start_document_supervisor(instance, new_workflow, document_name) session_pid = - start_supervised!( - {Session, - workflow: new_workflow, - user: user, - document_name: "workflow:#{workflow_id}"} - ) + start_session_proc(instance, new_workflow, user, document_name) + + allow_collaboration_process(session_pid) # First save - this creates the workflow in DB (lock_version becomes 1) assert {:ok, saved_workflow} = Session.save_workflow(session_pid, user) @@ -984,14 +1114,18 @@ defmodule Lightning.SessionTest do from_db = Lightning.Workflows.get_workflow!(workflow_id) assert from_db.name == "Updated After First Save" assert from_db.lock_version == 2 + + Lightning.Collaborate.stop_document(instance, document_name) end test "handles workflow deleted for :built workflow with lock_version > 0", %{ + instance: instance, user: user, project: project } do # Create a workflow and save it once to get lock_version > 0 workflow_id = Ecto.UUID.generate() + document_name = "workflow:#{workflow_id}" new_workflow = %Workflow{ id: workflow_id, @@ -1003,18 +1137,12 @@ defmodule Lightning.SessionTest do triggers: [] } - start_supervised!( - {DocumentSupervisor, - workflow: new_workflow, document_name: "workflow:#{workflow_id}"} - ) + start_document_supervisor(instance, new_workflow, document_name) session_pid = - start_supervised!( - {Session, - workflow: new_workflow, - user: user, - document_name: "workflow:#{workflow_id}"} - ) + start_session_proc(instance, new_workflow, user, document_name) + + allow_collaboration_process(session_pid) # First save to create in DB assert {:ok, _saved_workflow} = Session.save_workflow(session_pid, user) @@ -1031,13 +1159,17 @@ defmodule Lightning.SessionTest do # Try to save again - should get workflow_deleted error (covering line 475) assert {:error, :workflow_deleted} = Session.save_workflow(session_pid, user) + + Lightning.Collaborate.stop_document(instance, document_name) end @tag :capture_log test "malformed job id returns a changeset error without crashing the session", %{ session: session, - user: user + user: user, + instance: instance, + document_name: document_name } do # 16-byte unsubstituted placeholder: cast/1 accepts it, dump/1 rejects it. malformed_job = @@ -1058,13 +1190,17 @@ defmodule Lightning.SessionTest do # The contract from #4816: the GenServer (and thus the channel) stays up. assert Process.alive?(session) + + drain_document(instance, document_name) end @tag :capture_log test "cron cursor pointing at a job in another workflow returns a changeset error without crashing the session", %{ session: session, - user: user + user: user, + instance: instance, + document_name: document_name } do # A job in a DIFFERENT workflow — the compound same-workflow FK rejects a # cursor that points at it (finding #4: silent cross-workflow corruption). @@ -1091,6 +1227,8 @@ defmodule Lightning.SessionTest do assert {:error, %Ecto.Changeset{}} = Session.save_workflow(session, user) assert Process.alive?(session) + + drain_document(instance, document_name) end end @@ -1099,8 +1237,7 @@ defmodule Lightning.SessionTest do # we need workflows with :built state (not yet persisted to DB). # The main save_workflow/2 tests use insert() which creates :loaded workflows. describe "save_workflow/2 with NEW workflows" do - setup do - Mox.set_mox_global(LightningMock) + setup %{instance: instance} do Mox.stub(LightningMock, :broadcast, fn _topic, _message -> :ok end) user = insert(:user) @@ -1119,20 +1256,18 @@ defmodule Lightning.SessionTest do document_name = "workflow:new:#{workflow_id}" - start_supervised!( - {DocumentSupervisor, workflow: workflow, document_name: document_name} - ) + start_document_supervisor(instance, workflow, document_name) - session_pid = - start_supervised!( - {Session, workflow: workflow, user: user, document_name: document_name} - ) + session_pid = start_session_proc(instance, workflow, user, document_name) + + allow_collaboration_process(session_pid) %{ session: session_pid, user: user, workflow: workflow, - project: project + project: project, + document_name: document_name } end @@ -1140,7 +1275,9 @@ defmodule Lightning.SessionTest do session: session, user: user, project: project, - workflow: workflow + workflow: workflow, + document_name: document_name, + instance: instance } do # Verify workflow has :built state (new, not yet in DB) assert workflow.__meta__.state == :built @@ -1190,12 +1327,18 @@ defmodule Lightning.SessionTest do triggers_array = Yex.Doc.get_array(doc, "triggers") [ydoc_trigger] = Yex.Array.to_list(triggers_array) assert Yex.Map.fetch!(ydoc_trigger, "enabled") == false + + # Synchronously drain the document's batched writes before returning, so + # no PersistenceWriter write is in flight as the sandbox owner exits. + Lightning.Collaborate.stop_document(instance, document_name) end test "keeps triggers enabled when under activation limit", %{ session: session, user: user, - workflow: workflow + workflow: workflow, + document_name: document_name, + instance: instance } do assert workflow.__meta__.state == :built @@ -1228,44 +1371,43 @@ defmodule Lightning.SessionTest do # Verify trigger remains enabled saved_trigger = Enum.find(saved_workflow.triggers, &(&1.id == trigger_id)) assert saved_trigger.enabled == true + + Lightning.Collaborate.stop_document(instance, document_name) end end describe "save_workflow/2 validation errors" do - setup do - # Set global mode for the mock to allow cross-process calls - Mox.set_mox_global(LightningMock) - # Stub the broadcast calls that save_workflow makes + setup %{instance: instance} do + # Stub the broadcast calls that save_workflow makes. save_workflow runs in + # the Session process (allowed below); the DocumentSupervisor's spawned + # children are granted access by the owner-anchored startup hook. Mox.stub(LightningMock, :broadcast, fn _topic, _message -> :ok end) user = insert(:user) project = insert(:project) workflow = insert(:workflow, name: "Original Name", project: project) + document_name = "workflow:#{workflow.id}" - start_supervised!( - {DocumentSupervisor, - workflow: workflow, document_name: "workflow:#{workflow.id}"} - ) + start_document_supervisor(instance, workflow, document_name) - session_pid = - start_supervised!( - {Session, - workflow: workflow, - user: user, - document_name: "workflow:#{workflow.id}"} - ) + session_pid = start_session_proc(instance, workflow, user, document_name) + + allow_collaboration_process(session_pid) %{ session: session_pid, user: user, workflow: workflow, - project: project + project: project, + document_name: document_name } end test "writes validation errors to Y.Doc when workflow name is blank", %{ session: session, - user: user + user: user, + instance: instance, + document_name: document_name } do # Get Y.Doc and set blank name doc = Session.get_doc(session) @@ -1287,11 +1429,15 @@ defmodule Lightning.SessionTest do assert is_map(errors["workflow"]) assert is_list(errors["workflow"]["name"]) assert "This field can't be blank." in errors["workflow"]["name"] + + drain_document(instance, document_name) end test "clears errors from Y.Doc after successful save", %{ session: session, - user: user + user: user, + instance: instance, + document_name: document_name } do # Set up: Create errors in Y.Doc manually doc = Session.get_doc(session) @@ -1318,11 +1464,15 @@ defmodule Lightning.SessionTest do # Verify errors were cleared errors = Yex.Map.to_json(errors_map) assert errors == %{} + + drain_document(instance, document_name) end test "writes nested job validation errors to Y.Doc", %{ session: session, - user: user + user: user, + instance: instance, + document_name: document_name } do # Add a job with blank name doc = Session.get_doc(session) @@ -1363,11 +1513,15 @@ defmodule Lightning.SessionTest do errors["jobs"][job_id]["name"], &String.contains?(&1, "can't be blank") ) + + drain_document(instance, document_name) end test "nests workflow-level and entity errors correctly in Y.Doc", %{ session: session, - user: user + user: user, + instance: instance, + document_name: document_name } do # Create validation errors at both workflow and job level doc = Session.get_doc(session) @@ -1427,11 +1581,15 @@ defmodule Lightning.SessionTest do # Y.Doc has: workflow (map), jobs (array), triggers (array), edges (array) # Errors mirror this: workflow (map), jobs (map), triggers (map), edges (map) assert Map.keys(errors) |> Enum.sort() == ["jobs", "workflow"] + + drain_document(instance, document_name) end test "merges server errors with existing client errors", %{ session: session, - user: user + user: user, + instance: instance, + document_name: document_name } do # Simulate client-side validation errors already in Y.Doc doc = Session.get_doc(session) @@ -1495,11 +1653,15 @@ defmodule Lightning.SessionTest do assert Map.has_key?(errors["jobs"], job_id_2) assert is_list(errors["jobs"][job_id_2]["adaptor"]) assert "invalid adaptor" in errors["jobs"][job_id_2]["adaptor"] + + drain_document(instance, document_name) end test "server errors take precedence over client errors for same field", %{ session: session, - user: user + user: user, + instance: instance, + document_name: document_name } do # Add client error for workflow name doc = Session.get_doc(session) @@ -1535,11 +1697,15 @@ defmodule Lightning.SessionTest do ) refute "client side validation error" in errors["workflow"]["name"] + + drain_document(instance, document_name) end test "preserves client errors for entities not validated by server", %{ session: session, - user: user + user: user, + instance: instance, + document_name: document_name } do # Simulate multiple client errors across different entity types doc = Session.get_doc(session) @@ -1613,12 +1779,16 @@ defmodule Lightning.SessionTest do assert "client edge error" in errors["edges"][edge_id][ "condition_expression" ] + + drain_document(instance, document_name) end test "writes kafka_configuration validation errors correctly (not double-nested)", %{ session: session, - user: user + user: user, + instance: instance, + document_name: document_name } do # Add a kafka trigger with blank/invalid kafka_configuration doc = Session.get_doc(session) @@ -1668,6 +1838,8 @@ defmodule Lightning.SessionTest do # Should have actual validation errors assert Map.has_key?(kafka_errors, "hosts_string") assert Map.has_key?(kafka_errors, "topics_string") + + drain_document(instance, document_name) end test "returns error when existing workflow tries to activate trigger at limit", @@ -1675,7 +1847,9 @@ defmodule Lightning.SessionTest do session: session, user: user, workflow: workflow, - project: project + project: project, + instance: instance, + document_name: document_name } do # Verify workflow has :loaded state (existing, from DB) assert workflow.__meta__.state == :loaded @@ -1719,28 +1893,25 @@ defmodule Lightning.SessionTest do Session.save_workflow(session, user) assert text =~ "limit" + + drain_document(instance, document_name) end end describe "persistence reconciliation" do test "reconciles lock_version when loading persisted Y.Doc state", %{ + instance: instance, user: user } do workflow = insert(:simple_workflow) + document_name = "workflow:#{workflow.id}" # Start initial session and make some changes - {:ok, _doc_supervisor} = - Lightning.Collaborate.start_document( - workflow, - "workflow:#{workflow.id}" - ) + start_document_supervisor(instance, workflow, document_name) - {:ok, session1} = - Session.start_link( - user: user, - workflow: workflow, - parent_pid: self(), - document_name: "workflow:#{workflow.id}" + session1 = + start_session_proc(instance, workflow, user, document_name, + parent_pid: self() ) # Get the SharedDoc and verify initial lock_version @@ -1763,24 +1934,20 @@ defmodule Lightning.SessionTest do # Verify database has new lock_version assert updated_workflow.lock_version == new_lock_version - # Stop the session and document supervisor to simulate server restart + # Stop the session and document supervisor to simulate server restart. + # Both stops are synchronous, so the persisted state is flushed before we + # start a fresh tree; ExUnit tolerates terminating the now-dead children at + # teardown. Session.stop(session1) - ensure_doc_supervisor_stopped(workflow.id) + ensure_doc_supervisor_stopped(instance, workflow.id) # Start a new session - this will load persisted Y.Doc state # The persisted state has old lock_version, but fresh workflow has new one - {:ok, _doc_supervisor2} = - Lightning.Collaborate.start_document( - updated_workflow, - "workflow:#{workflow.id}" - ) + start_document_supervisor(instance, updated_workflow, document_name) - {:ok, session2} = - Session.start_link( - user: user, - workflow: updated_workflow, - parent_pid: self(), - document_name: "workflow:#{workflow.id}" + session2 = + start_session_proc(instance, updated_workflow, user, document_name, + parent_pid: self() ) # Get the SharedDoc and check lock_version was reconciled @@ -1796,27 +1963,24 @@ defmodule Lightning.SessionTest do reconciled_name = Yex.Map.fetch!(workflow_map2, "name") assert reconciled_name == "Updated Name" - Session.stop(session2) + # Synchronously drain (stops the session's SharedDoc + PersistenceWriter, + # flush included) before this test — the sandbox owner — returns. + Lightning.Collaborate.stop_document(instance, document_name) end test "discards stale persisted Y.Doc when lock_version changes", %{ + instance: instance, user: user } do workflow = insert(:simple_workflow) + document_name = "workflow:#{workflow.id}" # Start initial session with lock_version 0 - {:ok, _doc_supervisor} = - Lightning.Collaborate.start_document( - workflow, - "workflow:#{workflow.id}" - ) + start_document_supervisor(instance, workflow, document_name) - {:ok, session1} = - Session.start_link( - user: user, - workflow: workflow, - parent_pid: self(), - document_name: "workflow:#{workflow.id}" + session1 = + start_session_proc(instance, workflow, user, document_name, + parent_pid: self() ) # Get initial state @@ -1837,22 +2001,15 @@ defmodule Lightning.SessionTest do # Simulate server restart - persisted Y.Doc has old lock_version Session.stop(session1) - ensure_doc_supervisor_stopped(workflow.id) + ensure_doc_supervisor_stopped(instance, workflow.id) # Start new session with updated workflow # Persistence should detect stale lock_version and reload from DB - {:ok, _doc_supervisor2} = - Lightning.Collaborate.start_document( - updated_workflow, - "workflow:#{workflow.id}" - ) + start_document_supervisor(instance, updated_workflow, document_name) - {:ok, session2} = - Session.start_link( - user: user, - workflow: updated_workflow, - parent_pid: self(), - document_name: "workflow:#{workflow.id}" + session2 = + start_session_proc(instance, updated_workflow, user, document_name, + parent_pid: self() ) # Verify Y.Doc was reloaded from database @@ -1873,11 +2030,12 @@ defmodule Lightning.SessionTest do reconciled_name = Yex.Map.fetch!(workflow_map2, "name") assert reconciled_name == "Changed by another user" - Session.stop(session2) + Lightning.Collaborate.stop_document(instance, document_name) end test "handles persisted Y.Doc with nil lock_version when DB has real version", %{ + instance: instance, user: user } do # This tests the bug fix for issue #4164 @@ -1909,18 +2067,11 @@ defmodule Lightning.SessionTest do # Now start a session - this should NOT crash # The persistence layer should handle the nil lock_version and reset from DB - {:ok, _doc_supervisor} = - Lightning.Collaborate.start_document( - workflow, - doc_name - ) + start_document_supervisor(instance, workflow, doc_name) - {:ok, session} = - Session.start_link( - user: user, - workflow: workflow, - parent_pid: self(), - document_name: doc_name + session = + start_session_proc(instance, workflow, user, doc_name, + parent_pid: self() ) # Verify the session started and lock_version was reconciled from DB @@ -1932,10 +2083,11 @@ defmodule Lightning.SessionTest do assert reconciled_lock_version == workflow.lock_version, "Expected lock_version #{workflow.lock_version} but got #{reconciled_lock_version}" - Session.stop(session) + Lightning.Collaborate.stop_document(instance, doc_name) end test "merges delta updates with persisted state across save batches", %{ + instance: instance, user: user } do # This tests the fix for the merge_updates bug where delta updates @@ -2000,18 +2152,11 @@ defmodule Lightning.SessionTest do }) # Now start a session - this will load and reconstruct the full state - {:ok, _doc_supervisor} = - Lightning.Collaborate.start_document( - workflow, - doc_name - ) + start_document_supervisor(instance, workflow, doc_name) - {:ok, session} = - Session.start_link( - user: user, - workflow: workflow, - parent_pid: self(), - document_name: doc_name + session = + start_session_proc(instance, workflow, user, doc_name, + parent_pid: self() ) # Verify the session loaded the full state correctly @@ -2032,7 +2177,7 @@ defmodule Lightning.SessionTest do loaded_lock_version = Yex.Map.fetch!(loaded_workflow_map, "lock_version") assert loaded_lock_version == workflow.lock_version - Session.stop(session) + Lightning.Collaborate.stop_document(instance, doc_name) end end end diff --git a/test/lightning/collaboration/workflow_reconciler_test.exs b/test/lightning/collaboration/workflow_reconciler_test.exs index ffdee70fce4..f93d79f62c9 100644 --- a/test/lightning/collaboration/workflow_reconciler_test.exs +++ b/test/lightning/collaboration/workflow_reconciler_test.exs @@ -1,8 +1,5 @@ defmodule Lightning.Collaboration.WorkflowReconcilerTest do - # Tests must be async: false because we put a SharedDoc in a dynamic supervisor - # that isn't owned by the test process. So we need our Ecto sandbox to be - # in shared mode. - use Lightning.DataCase, async: false + use Lightning.DataCase, async: true import Lightning.Factories import Lightning.CollaborationHelpers @@ -11,14 +8,20 @@ defmodule Lightning.Collaboration.WorkflowReconcilerTest do alias Lightning.Collaboration.{Session, WorkflowReconciler} alias Lightning.Workflows - # we assume that the WorkflowCollaboration supervisor is up - # that starts :pg with the :workflow_collaboration scope - # and a dynamic supervisor called Lightning.WorkflowCollaboration + # WorkflowReconciler resolves its target SharedDoc through the default `:pg` + # scope (`Session.lookup_shared_doc/1`), exactly as the production save path + # does (`Workflows.after_commit/3`). So these tests run against the + # application-wide default instance rather than a per-test one. That is safe + # under async because every test inserts a fresh workflow with a unique id, so + # its document name (`workflow:`) never collides with another test's — + # the same isolation production already relies on between concurrent + # workflows. Determinism comes from ownership, not from a private registry: + # each document is started with `owner: self()` and torn down (flush included) + # before this test process — the sandbox owner — exits. setup do - # Set global mode for the mock to allow cross-process calls - Mox.set_mox_global(LightningMock) - # Stub the broadcast calls that WorkflowReconciler makes + # Stub the broadcast calls that the reconcile path makes from the test + # process (private-mode Mox: the stub applies to this process). Mox.stub(LightningMock, :broadcast, fn _topic, _message -> :ok end) user = insert(:user) @@ -28,21 +31,67 @@ defmodule Lightning.Collaboration.WorkflowReconcilerTest do describe "reconcile_workflow_changes/2" do setup do workflow = insert(:complex_workflow) + %{workflow: workflow} + end + # Start the collaboration document for this workflow, then open a session + # against it. `Collaborate.start/1` reuses the document we pre-start here + # rather than starting an unowned one. + # + # It targets the application-wide default registry/`:pg` scope because + # WorkflowReconciler resolves the SharedDoc through the default scope, exactly + # as production does; document-name uniqueness (a fresh workflow per test) + # keeps concurrent tests isolated. + # + # Teardown must run the document's flush `:normal` (so DocumentSupervisor's + # terminate/2 runs) rather than via ExUnit's supervised `:shutdown` (which a + # non-trapping DocumentSupervisor turns into an abrupt kill, skipping the + # flush and leaving its DB-writing children to be killed mid-query — a sandbox + # disconnect). So the document is started owner-monitored (not + # `start_supervised!`) and the `on_exit` below stops it `:normal` via + # `Collaborate.stop_document/1` (flush-inclusive). The default registry is the + # app global, alive throughout `on_exit`; the callback is registered after + # DataCase's `stop_owner` and runs LIFO before it, so the flush completes + # while this test — the sandbox owner — is still alive. + defp start_session(workflow, user) do + document_name = "workflow:#{workflow.id}" + + {:ok, _doc_supervisor} = + start_collaboration_document(workflow, document_name) + + {:ok, session_pid} = Collaborate.start(workflow: workflow, user: user) + + allow_collaboration_process(session_pid) + + # Sessions live under the default (non-ExUnit) dynamic supervisor. Stop the + # session first (so its terminate-time unobserve hits a live SharedDoc), + # then drain the document `:normal` (flush-inclusive). on_exit(fn -> + stop_session(session_pid) ensure_doc_supervisor_stopped(workflow.id) end) - %{workflow: workflow} + session_pid + end + + # Synchronously stop a session, tolerating the races inherent in teardown: it + # may already be gone, or exit :normal as we stop it. Returns only once the + # process is dead. + defp stop_session(session_pid) do + if Process.alive?(session_pid) do + try do + Session.stop(session_pid) + catch + :exit, _ -> :ok + end + end end test "job insert operations are applied to YDoc", %{ user: user, workflow: workflow } do - # Start a session to create the SharedDoc - {:ok, session_pid} = - Collaborate.start(workflow: workflow, user: user) + session_pid = start_session(workflow, user) # Create a new job changeset new_job = @@ -90,17 +139,15 @@ defmodule Lightning.Collaboration.WorkflowReconcilerTest do "adaptor" => "@openfn/language-http@latest" } = new_job_data - Session.stop(session_pid) - ensure_doc_supervisor_stopped(workflow.id) + # Teardown (session stop + synchronous document flush/stop) is handled by + # the `on_exit` registered in start_session/2. end test "job update operations are applied to YDoc", %{ user: user, workflow: workflow } do - # Start a session to create the SharedDoc - {:ok, session_pid} = - Collaborate.start(workflow: workflow, user: user) + session_pid = start_session(workflow, user) # Get the SharedDoc and verify initial state shared_doc = Session.get_doc(session_pid) @@ -157,17 +204,15 @@ defmodule Lightning.Collaboration.WorkflowReconcilerTest do } = updated_job_data - Session.stop(session_pid) - ensure_doc_supervisor_stopped(workflow.id) + # Teardown (session stop + synchronous document flush/stop) is handled by + # the `on_exit` registered in start_session/2. end test "job delete operations are applied to YDoc", %{ user: user, workflow: workflow } do - # Start a session to create the SharedDoc - {:ok, session_pid} = - Collaborate.start(workflow: workflow, user: user) + session_pid = start_session(workflow, user) # Get the first job to delete job_to_delete = Enum.at(workflow.jobs, 0) @@ -202,17 +247,15 @@ defmodule Lightning.Collaboration.WorkflowReconcilerTest do # Verify the deleted job is no longer in the YDoc refute find_in_ydoc_array(jobs_array, job_to_delete.id) - Session.stop(session_pid) - ensure_doc_supervisor_stopped(workflow.id) + # Teardown (session stop + synchronous document flush/stop) is handled by + # the `on_exit` registered in start_session/2. end test "edge insert operations are applied to YDoc", %{ user: user, workflow: workflow } do - # Start a session to create the SharedDoc - {:ok, session_pid} = - Collaborate.start(workflow: workflow, user: user) + session_pid = start_session(workflow, user) # Create a new edge between existing jobs %{id: source_job_id} = source_job = Enum.at(workflow.jobs, 1) @@ -268,9 +311,7 @@ defmodule Lightning.Collaboration.WorkflowReconcilerTest do user: user, workflow: workflow } do - # Start a session to create the SharedDoc - {:ok, session_pid} = - Collaborate.start(workflow: workflow, user: user) + session_pid = start_session(workflow, user) # Get the first edge to update edge_to_update = Enum.at(workflow.edges, 0) @@ -310,9 +351,7 @@ defmodule Lightning.Collaboration.WorkflowReconcilerTest do user: user, workflow: workflow } do - # Start a session to create the SharedDoc - {:ok, session_pid} = - Collaborate.start(workflow: workflow, user: user) + session_pid = start_session(workflow, user) # Get the first edge to delete edge_to_delete = Enum.at(workflow.edges, 0) @@ -355,9 +394,7 @@ defmodule Lightning.Collaboration.WorkflowReconcilerTest do user: user, workflow: workflow } do - # Start a session to create the SharedDoc - {:ok, session_pid} = - Collaborate.start(workflow: workflow, user: user) + session_pid = start_session(workflow, user) # Get the first trigger to update trigger_to_update = Enum.at(workflow.triggers, 0) @@ -403,9 +440,7 @@ defmodule Lightning.Collaboration.WorkflowReconcilerTest do user: user, workflow: workflow } do - # Start a session to create the SharedDoc - {:ok, session_pid} = - Collaborate.start(workflow: workflow, user: user) + session_pid = start_session(workflow, user) # Get the trigger to delete trigger_to_delete = Enum.at(workflow.triggers, 0) @@ -442,9 +477,7 @@ defmodule Lightning.Collaboration.WorkflowReconcilerTest do user: user, workflow: workflow } do - # Start a session to create the SharedDoc - {:ok, session_pid} = - Collaborate.start(workflow: workflow, user: user) + session_pid = start_session(workflow, user) # Create changeset for updating workflow properties workflow_changeset = @@ -470,9 +503,7 @@ defmodule Lightning.Collaboration.WorkflowReconcilerTest do user: user, workflow: workflow } do - # Start a session to create the SharedDoc - {:ok, session_pid} = - Collaborate.start(workflow: workflow, user: user) + session_pid = start_session(workflow, user) # Get initial lock_version from YDoc shared_doc = Session.get_doc(session_pid) @@ -505,17 +536,15 @@ defmodule Lightning.Collaboration.WorkflowReconcilerTest do assert Yex.Map.fetch!(updated_workflow_map, "lock_version") == new_lock_version - Session.stop(session_pid) - ensure_doc_supervisor_stopped(workflow.id) + # Teardown (session stop + synchronous document flush/stop) is handled by + # the `on_exit` registered in start_session/2. end test "positions updates are applied to YDoc", %{ user: user, workflow: workflow } do - # Start a session to create the SharedDoc - {:ok, session_pid} = - Collaborate.start(workflow: workflow, user: user) + session_pid = start_session(workflow, user) # Get some job IDs to create positions for job1_id = Enum.at(workflow.jobs, 0).id @@ -555,9 +584,7 @@ defmodule Lightning.Collaboration.WorkflowReconcilerTest do user: user, workflow: workflow } do - # Start a session to create the SharedDoc - {:ok, session_pid} = - Collaborate.start(workflow: workflow, user: user) + session_pid = start_session(workflow, user) # Get existing entities to modify job_to_update = Enum.at(workflow.jobs, 0) @@ -668,9 +695,7 @@ defmodule Lightning.Collaboration.WorkflowReconcilerTest do user: user, workflow: workflow } do - # Start a session to create the SharedDoc - {:ok, session_pid} = - Collaborate.start(workflow: workflow, user: user) + session_pid = start_session(workflow, user) # Create many new jobs at once (stress test) new_jobs = @@ -722,9 +747,9 @@ defmodule Lightning.Collaboration.WorkflowReconcilerTest do user: user, workflow: workflow } do - # Start multiple sessions to create multiple references to SharedDoc - {:ok, session1_pid} = - Collaborate.start(workflow: workflow, user: user) + # First session pre-starts the owned document (and registers teardown); + # the next two reuse that same document. + session1_pid = start_session(workflow, user) {:ok, session2_pid} = Collaborate.start(workflow: workflow, user: user) @@ -732,6 +757,11 @@ defmodule Lightning.Collaboration.WorkflowReconcilerTest do {:ok, session3_pid} = Collaborate.start(workflow: workflow, user: user) + for pid <- [session2_pid, session3_pid] do + allow_collaboration_process(pid) + on_exit(fn -> stop_session(pid) end) + end + # Verify all sessions share the same SharedDoc shared_doc1 = Session.get_doc(session1_pid) shared_doc2 = Session.get_doc(session2_pid) diff --git a/test/support/collaboration_helpers.ex b/test/support/collaboration_helpers.ex index d4521490c0e..49972f62003 100644 --- a/test/support/collaboration_helpers.ex +++ b/test/support/collaboration_helpers.ex @@ -1,4 +1,5 @@ defmodule Lightning.CollaborationHelpers do + alias Lightning.Collaboration.Instance alias Lightning.Collaboration.Session @doc """ @@ -15,31 +16,153 @@ defmodule Lightning.CollaborationHelpers do end @doc """ - Ensure the document supervisor is stopped for a given workflow id. + Start an isolated collaboration supervision tree for the calling test and + return its `%Lightning.Collaboration.Instance{}`. - This document supervisor has two children: - - PersistenceWriter - - SharedDoc + Each call builds a unique base name, so the started tree has its own Registry, + DynamicSupervisor, and `:pg` scope — independent of the application-wide + singletons and of any other test's tree. `start_supervised!/1` ties the tree's + lifetime to the test, so it is torn down automatically on exit. - When the parent process is stopped, the Session process is stopped. - The SharedDoc auto exits when there are no more Session processes observing it. + Pass the returned instance to `start_collaboration_document/3`, + `ensure_doc_supervisor_stopped/2`, and `stop_all_collaboration_documents/1` to + drive documents under this isolated tree. + """ + def start_collaboration_instance do + base = :"col_#{System.unique_integer([:positive])}" + + ExUnit.Callbacks.start_supervised!( + {Lightning.Collaboration.Supervisor, name: base} + ) + + Instance.derive(base) + end + + @doc """ + Start a collaboration document via the production entrypoint and tie its + lifetime to the calling test. + + Passes `owner: self()` to `Lightning.Collaborate.start_document/_`, so the + document tree monitors the test process and shuts itself down when the test + exits — the owner-monitored seam from + `.claude/guidelines/testable-supervision-trees.md` §3. No `on_exit` wrapper + needed. Returns whatever `Collaborate.start_document/_` returns. + + Pass an `%Instance{}` (from `start_collaboration_instance/0`) as the first + argument to drive the document under a test-owned, isolated tree; the + two-argument form uses the application-wide default instance. + + Prefer this over calling `Collaborate.start_document/_` directly in tests. + Documents live under a `DocSupervisor` that ExUnit doesn't own (unless you + started an isolated instance), so one started without an owner would outlive + the test that created it. + """ + def start_collaboration_document( + %Lightning.Workflows.Workflow{} = workflow, + document_name + ) + when is_binary(document_name) do + Lightning.Collaborate.start_document(workflow, document_name, owner: self()) + end + + def start_collaboration_document( + %Instance{} = instance, + %Lightning.Workflows.Workflow{} = workflow, + document_name + ) + when is_binary(document_name) do + Lightning.Collaborate.start_document(instance, workflow, document_name, + owner: self() + ) + end + + @doc """ + Grant a collaboration process the same per-test access the owner-anchored + startup hook grants the children it spawns. + + Tests that start a `Session` (or any other collaboration process) directly — + rather than letting `DocumentSupervisor.init/1` spawn it under `owner: self()` + — must call this on the returned pid so that process can reach the test's + database connection and resolve its mocks. + + Runs the configured `:collaboration_process_allow` callback (a no-op outside + the test env) for the database sandbox and the `Lightning` mock — the same + seam the startup hook uses — and additionally grants the per-test mocks a + `Session` resolves while saving (e.g. the usage limiter). Each `Mox.allow` is + guarded so a test that never stubbed a given mock is unaffected. + """ + def allow_collaboration_process(pid) when is_pid(pid) do + allow = + Application.get_env( + :lightning, + :collaboration_process_allow, + fn _owner, _pid -> :ok end + ) + + allow.(self(), pid) + + # The save path also resolves these mocks from inside the Session process. + # `set_mox_global` previously made every mock visible cross-process; under + # private Mox we allow them explicitly. A bare allow is harmless when the + # test set no expectations on the mock. + Enum.each( + [ + Lightning.Extensions.MockUsageLimiter, + Lightning.Extensions.MockProjectHook, + Lightning.Extensions.MockAccountHook, + Lightning.Extensions.MockCollectionHook, + Lightning.MockConfig + ], + fn mock -> safe_mox_allow(mock, pid) end + ) + + pid + end + + defp safe_mox_allow(mock, pid) do + Mox.allow(mock, self(), pid) + rescue + ArgumentError -> :ok + end + + @doc """ + Stop the document supervisor for a given workflow id. - When the SharedDoc process is stopped, the PersistenceWriter process does - it's own shutdown. This can take a millisecond or two, so to avoid - test errors where the PersistenceWriter tries to write to the database - after the test has finished, we wait for it's parent (DocumentSupervisor) - to be stopped. + Thin wrapper over `Lightning.Collaborate.stop_document/_` (synchronous, + idempotent, flush-inclusive). Safe in an `on_exit`: because the stop is + synchronous, the PersistenceWriter's final DB flush has completed by the time + it returns — see `Lightning.Collaborate.stop_document/_`. + + Pass an `%Instance{}` to stop a document under a test-owned isolated tree; the + single-argument form targets the application-wide default instance. """ def ensure_doc_supervisor_stopped(workflow_id) do - procs = - Lightning.Collaboration.Registry.get_group("workflow:#{workflow_id}") + Lightning.Collaborate.stop_document("workflow:#{workflow_id}") + end + + def ensure_doc_supervisor_stopped(%Instance{} = instance, workflow_id) do + Lightning.Collaborate.stop_document(instance, "workflow:#{workflow_id}") + end - if procs[:doc_supervisor] do - eventually_stop(procs.doc_supervisor) - end + @doc """ + Stop every collaboration document still running. + + A safety net for the `async: false` collaboration suites. Each document should + already be tied to its own test via `start_collaboration_document/_` (or + `ensure_doc_supervisor_stopped/_`); this only catches a stray document a call + site left running, keeping it out of the next test. + + Pass an `%Instance{}` to sweep only that test-owned tree. A test that drives + its own isolated instance does not need this sweep at all: the instance's + supervisor is `start_supervised!`-owned and `owner: self()` monitoring already + tears every document down on exit. + """ + def stop_all_collaboration_documents do + stop_all_collaboration_documents(Instance.default()) end - defp eventually_stop(pid) do - Eventually.eventually(fn -> Process.alive?(pid) end, false, 1000, 1) + def stop_all_collaboration_documents(%Instance{} = instance) do + Lightning.Collaboration.Registry.doc_supervisor_names(instance.registry) + |> Enum.each(&Lightning.Collaborate.stop_document(instance, &1)) end end