♻️ Drive websocket recheck from the relay loop

Fold the subscription re-check into the relay loop itself: wait on
the subscription channel and a one-shot tick channel with `alts!`
and priority on the tick. A single tick channel covers a whole
cadence and message turns reuse it, so the cadence holds under load
without churning timers. Drops `schedule-revalidation` and the
`promesa.exec` require. No behavior change; the focused suite
passes untouched.

AI-assisted-by: muse-spark-1.3-contributor

.serena/memories/backend/websocket-subscriptions.md

@@ -12,7 +12,7 @@
 A subscription is authorized once, at `:subscribe-file`/`:subscribe-team`, and that decision used to be trusted for the whole connection life (GHSA-m53j-2766-6jqw, medium). Two independent mechanisms now bound it:
 
 - **Announcement (primary).** A mutation that can change access publishes on `app.rpc.notifications/internal-revocation-topic` (`notify-permissions-changed` for one profile, `notify-team-permissions-changed` for one team id — the watcher resolves the members itself, so a large team is one event, not N). The `::revocation-watcher` Integrant component in `app.http.websocket` subscribes and calls `revalidate-profile-subscriptions`, which looks the profile up in `:by-profile` and re-checks every subscription with the same `has-read-permissions?` predicates the subscribe handlers authorize with — and `sql:file-permissions` additionally filters `deleted_at` on file, project and team, so a soft-deleted resource grants nothing. The event is only the trigger; the fresh check is the decision, which is what keeps a downgrade to `viewer` (still readable) and a non-member organization owner (read-only via `get-organization-owner-permissions`) from being mistaken for a revocation. Both publishers defer through `db/after-commit`, because the watcher checks on another connection and would otherwise see the pre-commit state.
-- **Interval (safety net).** `start-relay` arms a `px/schedule` task per subscription that re-checks every `subscription-revalidation-interval` (default `default-revalidation-interval`, 5 min, a code constant in `app.http.websocket` and **not** in the `config.clj` default map, same rule as session lifetimes). It covers what no announcement can: Nitrate transferring organization ownership (an HTTP call to another service) and rows deleted straight over SQL. The re-check is a scheduled task, not a timer in the relay loop, because a loop-driven timer would be pushed away forever by steady edit traffic — exactly the busiest files. The task stops when the channel closes: every way a subscription can end closes it, so a finished relay never keeps polling. Cost is two or three permission queries per subscription per interval, plus one Nitrate call per team per interval when the `:admin-console` flag is on (cached 30 s); ticks run on promesa's default scheduler (one or two threads shared with every other scheduled task), so do not shorten the interval without sizing that. Two places fail open, deliberately: `still-authorized?` (a transient DB error logs and keeps the subscription; `InterruptedException` is rethrown) and the `catch Throwable` in `watch-revocations`.
+- **Interval (safety net).** `start-relay` runs the relay as a `go-loop` that waits on the subscription channel and a one-shot tick channel together via `sp/alts!` with `:priority true` on the tick: when both hold something, the tick wins, so a steady stream of messages can never push the re-check away. The cadence is `subscription-revalidation-interval` (default `default-revalidation-interval`, 5 min, a code constant in `app.http.websocket` and **not** in the `config.clj` default map, same rule as session lifetimes). The deadline is fixed per cadence: a single tick channel covers a whole cadence and message turns reuse it (minting a fresh timeout per turn would both churn timers and starve under load exactly like an in-loop timer does); a new one is minted only after a tick fires. It covers what no announcement can: Nitrate transferring organization ownership (an HTTP call to another service) and rows deleted straight over SQL. Closing the channel ends the loop either way, because a take on a closed channel is always ready with nil — every way a subscription can end closes it, so a finished relay never keeps polling. Cost is two or three permission queries per subscription per interval, plus one Nitrate call per team per interval when the `:admin-console` flag is on (cached 30 s); one `timeout-chan` per subscription per cadence on promesa's default scheduler (one or two threads shared with every other scheduled task) — do not shorten the interval without sizing that. A tick abandoned at teardown (the subscription closed while it was pending) still fires its scheduled close harmlessly. Two places fail open, deliberately: `still-authorized?` (a transient DB error logs and keeps the subscription; `InterruptedException` is rethrown) and the `catch Throwable` in `watch-revocations`.
 
 Which mutations announce: `delete-team-member`, `leave-team`, `update-team-member-role` (member only), `delete-team`, `delete-file`, `delete-project`, `move-files`, `move-project` (one team event each). Anything not listed here is covered only by the interval. `delete-share-link` deliberately does not: a share-link visitor never passes the websocket subscribe check, because `check-read-permissions!` is called without a `share-id` and that arity only considers team membership.
 
@@ -25,14 +25,14 @@ Which mutations announce: `delete-team-member`, `leave-team`, `update-team-membe
 
 ## Time and channel traps in this namespace
 
-- `take!` is the proven relay path here. `msgbus.clj` does use `sp/alts!` over data channels in production, so there is no general defect — but the relay keeps the simple blocking take plus an independent scheduled task instead.
+- The relay waits with `sp/alts!` over `[tick-ch channel]` with `:priority true`, following the `msgbus.clj` pattern (`start-io-loop` destructures the same `[val port]` shape and compares ports with `identical?`). A take on a closed channel is always ready with nil, which is what ends the loop — and what makes closing the channel from any teardown path prompt even while parked in `alts!`. `sp/timeout-chan` is one-shot: it closes once after the delay, and one abandoned at teardown still fires its scheduled close harmlessly.
 - `ct/diff a b` returns **b − a** (`Duration/between`), so "time until a future instant" is `(inst-ms (ct/diff (ct/now) future))`.
 - `ct/duration` unit keywords are singular: `{:millis 150}`, `{:minutes 5}`, never `:milliseconds`/`:days`.
 - `still-authorized?` takes the check itself: `(boolean (check-access))`. Calling `(boolean pred)` on the closure is always true and silently disables the interval — this exact bug shipped once and no test caught it until the resubscribe-counting test existed. Naming rule here: no `?` for the action that queries and can fail (`check-access`), `?` only for the safe yes/no answer (`still-authorized?`).
 
 ## Tests
 
-`backend/test/backend_tests/http_websocket_test.clj` drives the real Redis-backed msgbus, so the revocation tests are end-to-end: register the connection in the real registry, run the real command, wait for the subscription to drop, then assert no content arrives. A connection built without `register-connection` is invisible to the watcher, so the watcher-based tests must go through `open-registered-connection`. Interval tests shorten the config via `with-redefs` on `cf/get` rather than waiting minutes. `with-clean-registry` runs as an `:each` fixture, so no test can leak a live relay (plus its revalidation task) into the next one. `captured-events` forwards the keyword args through to the real `mbus/pub!` — rebuilding them positionally publishes to the topic `:message` with a `nil` payload and silently untests every other notice. Interval tests take access away with a direct row delete, never with a command: a command announces, so the test would pass even with the interval removed. The resubscribe-counting test is the one that proves finished relays stop polling.
+`backend/test/backend_tests/http_websocket_test.clj` drives the real Redis-backed msgbus, so the revocation tests are end-to-end: register the connection in the real registry, run the real command, wait for the subscription to drop, then assert no content arrives. A connection built without `register-connection` is invisible to the watcher, so the watcher-based tests must go through `open-registered-connection`. Interval tests shorten the config via `with-redefs` on `cf/get` rather than waiting minutes. `with-clean-registry` runs as an `:each` fixture, so no test can leak a live relay loop into the next one. `captured-events` forwards the keyword args through to the real `mbus/pub!` — rebuilding them positionally publishes to the topic `:message` with a `nil` payload and silently untests every other notice. Interval tests take access away with a direct row delete, never with a command: a command announces, so the test would pass even with the interval removed. The resubscribe-counting test is the one that proves finished relays stop polling.
 
 ## See also
 

backend/src/app/http/websocket.clj

@@ -26,7 +26,6 @@
    [app.rpc.notifications :as notifications]
    [app.util.websocket :as ws]
    [integrant.core :as ig]
-   [promesa.exec :as px]
    [promesa.exec.csp :as sp]
    [yetti.websocket :as yws]))
 
@@ -223,46 +222,48 @@
         (swap! state dissoc ::team-subscription)))))
 
 
-(defn- schedule-revalidation
-  "Rearms `task` every `interval-ms` while `alive?` holds.
-
-  Kept independent of the relay loop on purpose: a re-check driven by
-  the same loop would be pushed away by a steady stream of messages, so
-  a busy file would never be re-checked at all. The closed channel is
-  what ends `alive?`, because every way a subscription can end closes
-  it."
-  [alive? interval-ms task]
-  (when (alive?)
-    (px/schedule interval-ms
-                 (fn []
-                   (when (alive?)
-                     (task)
-                     (schedule-revalidation alive? interval-ms task))))))
-
 (defn- start-relay
   "Forwards `channel` into the client output channel of `wsp`, and keeps
-  a re-check of `check-access` running on the interval. When access no
+  a re-check of `check-access` running on a fixed cadence. When access no
   longer holds, `close-fn` tears the subscription down, which closes the
   channel and ends the loop.
 
+  The loop waits on the subscription channel and a one-shot tick channel
+  together, with priority on the tick: when both hold something, the tick
+  wins, so a steady stream of messages can never push the re-check away.
+  A single tick channel covers a whole cadence and message turns reuse
+  it, so busy files don't churn timers; the cadence holds under load
+  because nothing moves the tick. Closing the channel ends the loop
+  either way, because a take on a closed channel is always ready
+  with nil.
+
   `on-forward` runs for every forwarded message, which is how the file
   relay announces presence."
   [{:keys [::ws/output-ch] :as wsp} channel check-access close-fn on-forward]
-  (schedule-revalidation #(not (sp/closed? channel))
-                         (inst-ms (revalidation-interval))
-                         (fn []
-                           (when-not (still-authorized? check-access)
-                             (l/info :hint
-                                     "closing websocket subscription on re-check"
-                                     :profile-id (::profile-id wsp))
-                             (close-fn))))
-  (sp/go-loop []
-    ;; nil means the channel was closed, which means the subscription
-    ;; is gone and the relay ends.
-    (when-let [message (sp/take! channel)]
-      (sp/put! output-ch message)
-      (when on-forward (on-forward message))
-      (recur))))
+  (let [interval-ms (inst-ms (revalidation-interval))]
+    (sp/go-loop [tick-ch (sp/timeout-chan interval-ms)]
+      (let [[message port] (sp/alts! [tick-ch channel] :priority true)]
+        (cond
+          (identical? port tick-ch)
+          (do
+            (when-not (still-authorized? check-access)
+              (l/info :hint
+                      "closing websocket subscription on re-check"
+                      :profile-id (::profile-id wsp))
+              (close-fn))
+            (when-not (sp/closed? channel)
+              (recur (sp/timeout-chan interval-ms))))
+
+          (nil? message)
+          ;; The channel was closed, which means the subscription is gone
+          ;; and the relay ends.
+          nil
+
+          :else
+          (do
+            (sp/put! output-ch message)
+            (when on-forward (on-forward message))
+            (recur tick-ch)))))))
 
 ;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;
 ;; WEBSOCKET HANDLER