dasemu/Synapse
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Fase 4: migrate the daemon event channel from SSE to WebSocket with resume-by-revision

Browse Source 
Adds a WS upgrade handler alongside the existing SSE stream on /v1/events
(kept alive for instant rollback), authenticated via Sec-WebSocket-Protocol
instead of a query-string token. The daemon now keeps a ~500-entry circular
buffer of broadcast sync-db-changes deltas indexed by renderer revision; a
client reconnecting sends {type: subscribe, graphId, sinceRev} and either
gets a gap-free replay + ack, or a {type: resync} instructing it to discard
all cached renderer state and reload (frontend.db.subs/reset-graph!, wired
through a new :db/resync event). The web/Electron client transport
(frontend.persist-db.remote) now opens the WS channel by default; the SSE
path (connect-events!) is untouched and still fully wired.

Also fixes new/pre-existing test breakage surfaced by switching the default
transport: persist_db_test.cljs's shared Electron-failover fixture now mocks
open-ws-fn (not just open-sse-fn), and a `new` applied to an inline (or ...)
expression in make-server's WebSocketServer construction is replaced with
the same let-bound-symbol pattern deps/db-sync's node/server.cljs already
uses -- inlining it silently dropped `new`, breaking every daemon test.

Confirmed 7 unrelated persist_db_test.cljs failures (electron-ensure-remote-*
IPC bookkeeping) already fail identically on the pre-Fase-4 baseline -- left
untouched, out of scope here.

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
This commit is contained in:
Darío Sevilla
2026-08-26 10:12:53 +02:00
parent be7abd52a3
commit cf61c46c35
10 changed files with 636 additions and 14 deletions
Download Patch File Download Diff File Expand all files Collapse all files

51
src/main/frontend/worker/db_worker_node.cljs
View File

@@ -1,10 +1,12 @@
(ns frontend.worker.db-worker-node
"Node.js daemon entrypoint for db-worker."
(:require ["http" :as http]
["ws" :as ws]
[clojure.string :as string]
[frontend.worker.db-core :as db-core]
[frontend.worker.db-worker-node-auth :as auth]
[frontend.worker.db-worker-node-lock :as db-lock]
[frontend.worker.db-worker-node-ws :as db-ws]
[frontend.worker.platform.node :as platform-node]
[frontend.worker.state :as worker-state]
[lambdaisland.glogi :as log]
@@ -158,6 +160,17 @@
(nil? method) nil
:else (str method)))
(defn- sync-db-changes-rev
"Peeks the transit-encoded [:sync-db-changes {...:delta {:rev}}] tuple
shared-service/broadcast-to-clients! always sends for this event type (see
db_listener.cljs), so the WS resume buffer can index by renderer revision
without re-deriving it. Returns nil for any other event type."
[type payload]
(when (and (= type :sync-db-changes) (string? payload))
(try
(get-in (ldb/read-transit-str payload) [1 :delta :rev])
(catch :default _ nil))))
(defn- handle-event!
[type payload]
(let [event (js/JSON.stringify (clj->js {:type (encode-event-type type)
@@ -167,7 +180,10 @@
(try
(.write res message)
(catch :default e
(log/error :sse-write-failed e))))))
(log/error :sse-write-failed e))))
(when-let [rev (sync-db-changes-rev type payload)]
(db-ws/record-delta! rev event))
(db-ws/broadcast! event)))
(defn- sse-handler
[^js req ^js res]
@@ -380,11 +396,32 @@
(str "Bearer ")))
(auth/authorized? (.. req -headers -authorization))))
(defn- ws-upgrade-handler
[^js wss bound-repo port extra-allowed-hosts]
(fn [^js req ^js socket head]
(let [allowed-hosts (allowed-hosts-for {:port port :extra-allowed-hosts extra-allowed-hosts})
request-path (.-pathname (js/URL. (.-url req) "http://127.0.0.1"))]
(cond
(not (host-allowed? req allowed-hosts))
(db-ws/reject-upgrade! socket 403 "Host header not allowed")
(not= request-path "/v1/events")
(db-ws/reject-upgrade! socket 404 "not found")
(not (auth/authorized-token? (db-ws/token-from-upgrade-req req)))
(db-ws/reject-upgrade! socket 401 "missing or invalid credentials")
:else
(.handleUpgrade wss req socket head
(fn [ws-socket]
(db-ws/handle-connection! ws-socket bound-repo)))))))
(defn- make-server
[proxy {:keys [bound-repo stop-fn host port owner-source root-dir
allowed-origins extra-allowed-hosts]}]
(http/createServer
(fn [^js req ^js res]
(let [server
(http/createServer
(fn [^js req ^js res]
(let [url (.-url req)
parsed-url (js/URL. url "http://127.0.0.1")
request-path (.-pathname parsed-url)
@@ -515,7 +552,11 @@
(send-text! res 405 "method-not-allowed"))
:else
(send-text! res 404 "not-found"))))))
(send-text! res 404 "not-found")))))
WSS (or (.-WebSocketServer ws) (.-Server ws))
^js wss (new WSS #js {:noServer true})]
(.on server "upgrade" (ws-upgrade-handler wss bound-repo port extra-allowed-hosts))
server))
(defn- show-help!
[]
@@ -584,6 +625,8 @@
(.end res)
(catch :default _)))
(reset! *sse-clients #{})
(db-ws/close-all!)
(db-ws/reset-buffer!)
(when-let [lock-path (:path @*lock-info)]
(db-lock/remove-lock! lock-path)))

18
src/main/frontend/worker/db_worker_node_auth.cljs
View File

@@ -146,16 +146,26 @@
(do (record-failed-attempt! username)
{:error :invalid-credentials})))))
(defn authorized-token?
"Given a bare session token (no `Bearer ` prefix), returns true and
refreshes :last-seen-at when it matches an active session. Shared by the
Authorization-header path (authorized?) and the WebSocket upgrade path,
which carries the token via Sec-WebSocket-Protocol instead -- see
db_worker_node_ws.cljs."
[token]
(when (string? token)
(when-let [session (some #(when (timing-safe-equal? token (:token %)) %)
(vals @*sessions))]
(swap! *sessions assoc-in [(:id session) :last-seen-at] (now-iso))
true)))
(defn authorized?
"Given an `Authorization: Bearer <token>` header value (or nil), returns
true and refreshes :last-seen-at when it matches an active session."
[authorization-header]
(when-let [token (some-> authorization-header
(as-> h (when (string/starts-with? h "Bearer ") (subs h 7))))]
(when-let [session (some #(when (timing-safe-equal? token (:token %)) %)
(vals @*sessions))]
(swap! *sessions assoc-in [(:id session) :last-seen-at] (now-iso))
true)))
(authorized-token? token)))
(defn list-sessions
[]

171
src/main/frontend/worker/db_worker_node_ws.cljs Normal file
View File

@@ -0,0 +1,171 @@
(ns frontend.worker.db-worker-node-ws
"WebSocket event channel for db-worker-node -- kept alongside the SSE
channel (/v1/events) during the Fase 4 migration so either transport can
be rolled back independently; see docs/agent-guide.
Wire format for domain events mirrors SSE exactly: {\"type\" ... \"payload\"
<transit-string>} -- db_worker_node.cljs builds that message once (for
both transports) and hands it to broadcast!/record-delta! here.
Client -> server control message: {\"type\" \"subscribe\" \"graphId\" repo
\"sinceRev\" rev-or-null}. Server -> client control messages:
{\"type\" \"subscribed\" \"rev\" head} (caught up, or replay was just sent),
{\"type\" \"resync\" \"rev\" head} (client must discard cached state and
reload everything -- see frontend.handler.events :db/resync),
{\"type\" \"error\" \"message\" ...}.
The bearer token can't travel in a header on the WS handshake the way
fetch requests use Authorization -- carried via Sec-WebSocket-Protocol
instead (see token-from-upgrade-req), which doesn't leak into proxy/access
logs the way the SSE interim query-param token does."
(:require [clojure.string :as string]
[lambdaisland.glogi :as log]))
(def ^:private buffer-capacity 500)
(def ^:private heartbeat-interval-ms 30000)
(defonce ^:private *clients (atom #{}))
(defonce ^:private *delta-buffer (atom (sorted-map)))
(defonce ^:private *head-rev (atom 0))
(defn- ws-open?
[^js socket]
(= 1 (.-readyState socket)))
(defn- ws-send!
[^js socket text]
(when (ws-open? socket)
(try
(.send socket text)
(catch :default e
(log/error :db-worker-node-ws-send-failed e)))))
(defn broadcast!
"Sends a pre-built wire message (see db_worker_node.cljs handle-event!) to
every currently connected WS client, live-tailing style. Independent of
record-delta! -- non-graph events (notifications, logs, ...) broadcast
without ever entering the resume buffer."
[message]
(doseq [^js socket @*clients]
(ws-send! socket message)))
(defn record-delta!
"Keeps the last `buffer-capacity` :sync-db-changes wire messages, indexed
by renderer revision, so a client reconnecting within that window can
resume via replay instead of a full reload."
[rev message]
(when (and (integer? rev) (not (neg? rev)))
(swap! *delta-buffer
(fn [buf]
(let [buf' (assoc buf rev message)]
(if (> (count buf') buffer-capacity)
(dissoc buf' (first (keys buf')))
buf'))))
(swap! *head-rev max rev)))
(defn- oldest-buffered-rev
[]
(some-> @*delta-buffer first key))
(defn- can-replay-from?
[since-rev]
(let [oldest (oldest-buffered-rev)]
(or (nil? oldest) (<= oldest (inc since-rev)))))
(defn- replay-messages
[since-rev]
(->> @*delta-buffer
(drop-while (fn [[rev _message]] (<= rev since-rev)))
(map second)))
(defn- send-control!
[^js socket payload]
(ws-send! socket (js/JSON.stringify (clj->js payload))))
(defn- handle-subscribe!
[^js socket bound-repo {:strs [graphId sinceRev]}]
(if (and (string? graphId) (not= graphId bound-repo))
(send-control! socket {:type "error" :message "graph-id does not match bound repo"})
(let [since-rev (if (integer? sinceRev) sinceRev -1)
head @*head-rev]
(if (can-replay-from? since-rev)
(do
(doseq [message (replay-messages since-rev)]
(ws-send! socket message))
(send-control! socket {:type "subscribed" :rev head}))
(send-control! socket {:type "resync" :rev head})))))
(defn- handle-message!
[^js socket bound-repo text]
(try
(let [data (js->clj (js/JSON.parse text))]
(case (get data "type")
"subscribe" (handle-subscribe! socket bound-repo data)
"ping" (send-control! socket {:type "pong"})
(send-control! socket {:type "error" :message "unknown message type"})))
(catch :default e
(log/error :db-worker-node-ws-message-failed e)
(send-control! socket {:type "error" :message "invalid message"}))))
(defn- start-heartbeat!
"Terminates sockets that stop answering pings -- mobile networks in
particular drop connections silently, and a dead entry in *clients would
otherwise sit there forever, still counted by broadcast!."
[^js socket]
(let [alive? (atom true)
interval (js/setInterval
(fn []
(if @alive?
(do (reset! alive? false)
(try (.ping socket) (catch :default _ nil)))
(try (.terminate socket) (catch :default _ nil))))
heartbeat-interval-ms)]
(.on socket "pong" (fn [] (reset! alive? true)))
interval))
(defn handle-connection!
[^js socket bound-repo]
(swap! *clients conj socket)
(let [heartbeat (start-heartbeat! socket)
cleanup! (fn []
(js/clearInterval heartbeat)
(swap! *clients disj socket))]
(.on socket "message"
(fn [data]
(handle-message! socket bound-repo (if (string? data) data (.toString data)))))
(.on socket "close" cleanup!)
(.on socket "error" (fn [error]
(log/error :db-worker-node-ws-error error)
(cleanup!)))))
(defn token-from-upgrade-req
"Extracts the bare session token from Sec-WebSocket-Protocol. The client
sends a single subprotocol string \"bearer.<token>\" (see
frontend.persist-db.remote/open-websocket)."
[^js req]
(when-let [header (some-> req .-headers (aget "sec-websocket-protocol"))]
(let [first-protocol (-> header (string/split #",") first string/trim)]
(when (string/starts-with? first-protocol "bearer.")
(subs first-protocol (count "bearer."))))))
(defn reject-upgrade!
[^js socket status reason]
(try
(.write socket
(str "HTTP/1.1 " status " " reason "\r\n"
"Connection: close\r\n"
"Content-Type: application/json\r\n\r\n"
(js/JSON.stringify (clj->js {:ok false :error {:message reason}}))))
(catch :default _ nil))
(try (.destroy socket) (catch :default _ nil)))
(defn close-all!
[]
(doseq [^js socket @*clients]
(try (.terminate socket) (catch :default _ nil)))
(reset! *clients #{}))
(defn reset-buffer!
[]
(reset! *delta-buffer (sorted-map))
(reset! *head-rev 0))