Tolerate slow or unreachable cross-node SharedDoc during session cleanup (#4845)

The collaborative editor Session calls SharedDoc.unobserve/1 during
cleanup, which issues a cross-node GenServer.call to a SharedDoc that may
live on another node. When that node is unreachable (:noconnection) or
slow to reply (:timeout), the call exited and crashed the Session; a
Phoenix channel blocked in save_workflow then inherited the exit and
dropped the client's editor connection.

Wrap both unobserve call sites (terminate/2 and the parent :DOWN handler)
in safe_unobserve/1, which catches the exit and logs at warning instead
of crashing. Skipping the unobserve is safe: the SharedDoc monitors each
observer and runs the same cleanup from its own :DOWN handler when the
Session dies.

Closes #4817

Author: midigofrank

CHANGELOG.md

@@ -42,6 +42,12 @@ and this project adheres to
 
 ### Fixed
 
+- Stop the collaborative editor's Session (and the Phoenix channel calling it)
+  from crashing when the cross-node `SharedDoc.unobserve/1` during cleanup hits
+  a SharedDoc on a node that is unreachable (`:noconnection`) or slow to reply
+  (`:timeout`); the failed unobserve is now tolerated as a no-op since the
+  SharedDoc cleans up observers via its own monitor.
+  [#4817](https://github.com/OpenFn/lightning/issues/4817)
 - Fix email format validation not displaying in the Add Collaborators and Invite
   Collaborator modal. [#4765](https://github.com/OpenFn/lightning/issues/4765)
 - Fix a `workflows_pkey` duplicate-key crash when reconnecting to the

lib/lightning/collaboration/session.ex

@@ -139,11 +139,11 @@ defmodule Lightning.Collaboration.Session do
       Process.demonitor(state.parent_ref)
     end
 
-    # Don't check Process.alive? - it only works for local PIDs
-    # and shared_doc_pid can be on another node in a distributed cluster.
-    # Sending to a dead process is safe (message is discarded).
+    # Don't check Process.alive? - it only works for local PIDs and
+    # shared_doc_pid can be on another node in a distributed cluster.
+    # safe_unobserve/1 tolerates the remote being slow or gone (see #4817).
     if shared_doc_pid do
-      SharedDoc.unobserve(shared_doc_pid)
+      safe_unobserve(shared_doc_pid)
     end
 
     Presence.untrack_user_presence(
@@ -155,6 +155,17 @@ defmodule Lightning.Collaboration.Session do
     :ok
   end
 
+  defp safe_unobserve(shared_doc_pid) do
+    SharedDoc.unobserve(shared_doc_pid)
+  catch
+    :exit, reason ->
+      Logger.warning(
+        "SharedDoc.unobserve skipped during session cleanup: #{inspect(reason)}"
+      )
+
+      :ok
+  end
+
   def lookup_shared_doc(document_name) do
     case :pg.get_members(@pg_scope, document_name) do
       [] -> nil
@@ -451,10 +462,11 @@ defmodule Lightning.Collaboration.Session do
     if ref == parent_ref do
       Process.demonitor(parent_ref)
 
-      # Don't check Process.alive? - it only works for local PIDs
-      # and shared_doc_pid can be on another node in a distributed cluster.
+      # Don't check Process.alive? - it only works for local PIDs and
+      # shared_doc_pid can be on another node in a distributed cluster.
+      # safe_unobserve/1 tolerates the remote being slow or gone (see #4817).
       if shared_doc_pid do
-        SharedDoc.unobserve(shared_doc_pid)
+        safe_unobserve(shared_doc_pid)
       end
 
       {:stop, :normal, %{state | parent_ref: nil, shared_doc_pid: nil}}

test/lightning/collaboration/session_test.exs

@@ -655,6 +655,27 @@ defmodule Lightning.SessionTest do
       #     %{}
       # )
     end
+
+    @tag :capture_log
+    test "terminate tolerates an unreachable SharedDoc", %{user: user} do
+      # The cross-node `unobserve` issued during cleanup can exit when the
+      # SharedDoc's node is gone (:noconnection) or slow (:timeout). terminate/2
+      # must swallow that and still finish cleanup rather than crash (#4817).
+      # A dead pid stands in for the unreachable remote: the GenServer.call to
+      # it exits :noproc, the same :exit class caught by safe_unobserve/1.
+      dead_pid = spawn(fn -> :ok end)
+      refute_eventually(Process.alive?(dead_pid))
+
+      state = %Session{
+        parent_ref: nil,
+        shared_doc_pid: dead_pid,
+        user: user,
+        workflow: %Workflow{id: Ecto.UUID.generate()},
+        document_name: "workflow:#{Ecto.UUID.generate()}"
+      }
+
+      assert :ok = Session.terminate(:shutdown, state)
+    end
   end
 
   defp get_expected_value(model, key) do