Fix pycrdt cross-thread GC crash in collab worker and server

pycrdt's Rust objects are !Send; the automatic, threshold-triggered
GC could collect a Doc/Subscription on a different thread than the
one that created it, crashing with a pyo3 unsendable-object error.

- gunicorn (business/collab.py): route all pycrdt calls through a
  single dedicated worker thread (iris_engine/collab/pycrdt_worker.py)
  and disable automatic GC process-wide, collecting explicitly only
  on that thread.
- collab_server.py (uvicorn, single event loop thread): add a
  periodic gc.collect() on the event-loop thread itself, since all
  pycrdt objects there are already created/destroyed on that one
  thread.

Author: Che4ter

source/app/business/collab.py

@@ -39,6 +39,7 @@
 from app.db import db
 from app.iris_engine.collab.render import markdown_to_ydoc_update
 from app.iris_engine.collab.render import ydoc_update_to_markdown
+from app.iris_engine.collab.pycrdt_worker import run as _run_on_pycrdt_thread
 from app.iris_engine.utils.tracker import track_activity
 from app.models.authorization import CaseAccessLevel
 from app.models.authorization import WarRoomAccessLevel
@@ -259,7 +260,7 @@ def ensure_snapshot(doc_name, current_content):
     # it would leave users unable to see content they know is in
     # `note.note_content` / `case.description` / etc.
     seed_md = current_content or ''
-    seed_bytes = markdown_to_ydoc_update(seed_md)
+    seed_bytes = _run_on_pycrdt_thread(markdown_to_ydoc_update, seed_md)
 
     if row is None:
         row = CollabDoc(
@@ -312,7 +313,12 @@ def apply_wire_update(doc_name, update_b64, user_id):
         # combining update blobs without needing to materialise a full
         # Doc — much cheaper than apply+get_update on hot paths. It's
         # a variadic (not a list), hence the star-splat.
-        new_state = merge_updates(bytes(row.y_state), update_bytes)
+        #
+        # Routed through the dedicated pycrdt worker thread: see
+        # `iris_engine.collab.pycrdt_worker` for why — the transient
+        # Doc/Subscription objects merge_updates allocates internally
+        # are `!Send` and must not be GC'd off-thread.
+        new_state = _run_on_pycrdt_thread(merge_updates, bytes(row.y_state), update_bytes)
     except Exception:
         # Malformed update — drop it. Peers won't see the broadcast
         # either (caller checks the return value).
@@ -348,7 +354,7 @@ def flush_to_source(doc_name):
         return
 
     try:
-        new_content = ydoc_update_to_markdown(bytes(row.y_state))
+        new_content = _run_on_pycrdt_thread(ydoc_update_to_markdown, bytes(row.y_state))
     except Exception:
         # Never let a bad snapshot break the audit path — keep the
         # source column as-is until a subsequent flush succeeds.

source/app/collab_server.py

@@ -9,6 +9,8 @@
 
 from __future__ import annotations
 
+import asyncio
+import gc
 import json
 import os
 import re
@@ -46,6 +48,15 @@
 )
 COLLAB_STORE_DB = COLLAB_STORE_DIR / "yjs.sqlite3"
 
+# `gc.disable()` (see iris_engine.collab.pycrdt_worker, imported transitively via
+# the collab blueprint) is process-wide, so this ASGI process also needs its own
+# explicit collection. Safe to run directly on the event-loop thread here: unlike
+# the gunicorn worker, every pycrdt Doc/XmlFragment in this process is created and
+# torn down on the single asyncio event-loop thread (the `to_thread` calls below
+# only ever touch plain DB objects), so there is no cross-thread drop to guard
+# against with a dedicated worker thread.
+GC_INTERVAL_SECONDS = 60
+
 
 class AuthorizedRoom(NamedTuple):
     room_name: str
@@ -99,6 +110,7 @@ class IrisCollabASGIApp:
     def __init__(self, websocket_server: WebsocketServer):
         self._websocket_server = websocket_server
         self._server_task: Task | None = None
+        self._gc_task: Task | None = None
 
     async def __call__(
         self,
@@ -141,8 +153,12 @@ async def _handle_lifespan(
                 COLLAB_STORE_DIR.mkdir(parents=True, exist_ok=True)
                 self._server_task = create_task(self._websocket_server.start())
                 await self._websocket_server.started.wait()
+                self._gc_task = create_task(_periodic_gc())
                 await send({"type": "lifespan.startup.complete"})
             elif message["type"] == "lifespan.shutdown":
+                if self._gc_task is not None:
+                    self._gc_task.cancel()
+                    self._gc_task = None
                 await self._websocket_server.stop()
                 if self._server_task is not None:
                     await self._server_task
@@ -183,6 +199,15 @@ async def _handle_http(
         await send({"type": "http.response.body", "body": body})
 
 
+async def _periodic_gc() -> None:
+    try:
+        while True:
+            await asyncio.sleep(GC_INTERVAL_SECONDS)
+            gc.collect()
+    except asyncio.CancelledError:
+        pass
+
+
 def authorize_scope(scope: dict[str, Any]) -> AuthorizedRoom | None:
     if not _check_origin(scope):
         origin = _get_header(scope, b"origin")

source/app/iris_engine/collab/pycrdt_worker.py

@@ -0,0 +1,64 @@
+#  IRIS Source Code
+#  Copyright (C) 2026 - DFIR-IRIS
+#  contact@dfir-iris.org
+#
+#  This program is free software; you can redistribute it and/or
+#  modify it under the terms of the GNU Lesser General Public
+#  License as published by the Free Software Foundation; either
+#  version 3 of the License, or (at your option) any later version.
+
+"""Confines all pycrdt (Yrs) work to a single dedicated thread.
+
+pycrdt's Rust objects (`Doc`, `XmlFragment`, and the transaction/
+subscription handles pyo3 wraps around them) are `!Send`: yrs uses
+`Rc<RefCell<...>>` internally, so if Python's cyclic garbage collector
+drops one of these objects on a thread other than the one that created
+it, pyo3 raises:
+
+    RuntimeError: pycrdt::subscription::Subscription is unsendable,
+    but is being dropped on another thread
+
+Gunicorn runs this app under a `gthread` worker (many request-handling
+threads sharing one heap/GIL). CPython's *automatic* generational
+collector can fire from any thread the moment its allocation counter
+crosses a threshold — so a `Doc` created while resolving a note on
+thread A can get collected, and crash, while thread B is busy doing
+something completely unrelated (e.g. a SQLAlchemy query). That's why
+the traceback in production never mentions collab code.
+
+Fix: disable automatic collection process-wide and route every pycrdt
+call, plus the `gc.collect()` that reclaims it, through this one
+worker thread. Objects are then always created and destroyed on the
+same thread, which is what pyo3 requires.
+"""
+from __future__ import annotations
+
+import gc
+from concurrent.futures import ThreadPoolExecutor
+from typing import Callable, TypeVar
+
+T = TypeVar("T")
+
+# Automatic, threshold-triggered collection can run on any thread. Disabling
+# it process-wide is what makes the explicit gc.collect() below in
+# `_run_and_collect` (always on `_executor`'s single thread) the only place
+# cyclic garbage collection ever happens.
+gc.disable()
+
+_executor = ThreadPoolExecutor(max_workers=1, thread_name_prefix="pycrdt-worker")
+
+
+def run(fn: Callable[..., T], *args, **kwargs) -> T:
+    """Run `fn(*args, **kwargs)` on the dedicated pycrdt thread and return its result.
+
+    Every pycrdt `Doc`/`XmlFragment` created inside `fn` is guaranteed to be
+    garbage-collected on this same thread before `run()` returns.
+    """
+    return _executor.submit(_run_and_collect, fn, args, kwargs).result()
+
+
+def _run_and_collect(fn, args, kwargs):
+    try:
+        return fn(*args, **kwargs)
+    finally:
+        gc.collect()