mirror of
https://github.com/penpot/penpot.git
synced 2026-10-29 14:00:37 +00:00
🐛 Re-check websocket subscriptions on an interval
The announcement path is the primary mechanism, but it cannot see an access change that fires no command here: Nitrate transferring the ownership of an organization is an HTTP call to another service, and rows deleted straight over SQL publish nothing. Both keep a subscription alive until the connection happens to close. Each relay now re-checks access every `subscription-revalidation-interval` (5 minutes by default, overridable) and drops the subscription when it no longer holds. The default is a code constant in `app.http.websocket` rather than the `config.clj` default map, matching how session lifetimes are configured. The re-check is a scheduled task rather than a timer inside the relay loop, on purpose: driven by the loop, a steady stream of edits would push the check away forever and a busy file would never be re-checked at all. `start-relay` also replaces the `sp/pipe` of the team subscription and the ad-hoc `go-loop` of the file one, so both now share the same forwarding and re-check path. `still-authorized?` is the one place that fails open: a transient database error logs and keeps the subscription rather than kicking a user out of an editor, and the next tick tries again. Closes #12155 AI-assisted-by: space-bunny-free
This commit is contained in:
Andrey Antukh
parent
e6e6b20cbb
commit
d608f3b898
@ -253,6 +253,10 @@
|
||||
[:auth-token-cookie-max-age {:optional true} ::ct/duration]
|
||||
[:auth-token-cookie-max-age-absolute {:optional true} ::ct/duration]
|
||||
|
||||
;; How often a websocket subscription re-checks that its access still
|
||||
;; holds. The default lives in `app.http.websocket`, not here.
|
||||
[:subscription-revalidation-interval {:optional true} ::ct/duration]
|
||||
|
||||
[:registration-domain-whitelist {:optional true} [::sm/set :string]]
|
||||
[:email-verify-threshold {:optional true} ::ct/duration]
|
||||
|
||||
|
||||
@ -15,6 +15,7 @@
|
||||
[app.common.schema :as sm]
|
||||
[app.common.time :as ct]
|
||||
[app.common.uuid :as uuid]
|
||||
[app.config :as cf]
|
||||
[app.db :as db]
|
||||
[app.http.session :as session]
|
||||
[app.metrics :as mtx]
|
||||
@ -25,6 +26,7 @@
|
||||
[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]))
|
||||
|
||||
@ -139,6 +141,133 @@
|
||||
(pp/pprint)))
|
||||
|
||||
;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;
|
||||
;; SUBSCRIPTIONS
|
||||
;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;
|
||||
|
||||
;; A subscription is authorized once, when it is opened, and that
|
||||
;; single decision is then trusted for as long as the connection
|
||||
;; lives. Two things bound that: a mutation that changes access
|
||||
;; announces it (see the revocation section below), and the relay
|
||||
;; re-checks on an interval.
|
||||
|
||||
(def default-revalidation-interval
|
||||
"How long a subscription may rely on the authorization it was granted
|
||||
with before its relay loop re-checks it.
|
||||
|
||||
This is the safety net for what the announcement cannot cover: an
|
||||
access change made outside a Penpot command. Nitrate transferring the
|
||||
ownership of an organization is an HTTP call to another service that
|
||||
fires no RPC here, and rows deleted straight over SQL fire nothing
|
||||
either. Both are rare, so the interval is generous; the cost is one
|
||||
permission query per subscription per interval."
|
||||
(ct/duration {:minutes 5}))
|
||||
|
||||
(defn revalidation-interval
|
||||
[]
|
||||
(cf/get :subscription-revalidation-interval default-revalidation-interval))
|
||||
|
||||
(defn- still-authorized?
|
||||
"Evaluates the `pred` permission predicate, reporting instead of
|
||||
throwing when the lookup itself fails.
|
||||
|
||||
A transient database error must not close a subscription that is still
|
||||
legitimate, and the next tick tries again. This is the one place that
|
||||
fails open, deliberately: the announcement path is the primary
|
||||
mechanism and stays strict."
|
||||
[pred]
|
||||
(try
|
||||
(boolean pred)
|
||||
(catch Throwable cause
|
||||
(l/error :hint "cannot re-check a websocket subscription"
|
||||
:cause cause)
|
||||
true)))
|
||||
|
||||
(defn close-file-subscription
|
||||
"Tears down the file subscription held by the connection `wsp`, if it
|
||||
is subscribed to `file-id`.
|
||||
|
||||
Announces the departure so the remaining participants drop the
|
||||
presence of this session, closes the relay channel (which in turn
|
||||
stops the `:subscribe-file` go-loop, because `take!` on a closed
|
||||
channel returns nil) and removes the subscription from the bus.
|
||||
|
||||
Does nothing when the connection is subscribed to a different file,
|
||||
so it is safe to call for a subscription that is already gone."
|
||||
[{:keys [::mbus/msgbus]} {:keys [::ws/state ::session-id ::profile-id]} file-id]
|
||||
(let [subs (::file-subscription @state)]
|
||||
(when (= (:file-id subs) file-id)
|
||||
(mbus/pub! msgbus
|
||||
:topic file-id
|
||||
:message {:type :leave-file
|
||||
:file-id file-id
|
||||
:session-id session-id
|
||||
:profile-id profile-id})
|
||||
(let [ch (:channel subs)]
|
||||
(sp/close! ch)
|
||||
(mbus/purge! msgbus [ch])
|
||||
(swap! state dissoc ::file-subscription)))))
|
||||
|
||||
(defn close-team-subscription
|
||||
"Tears down the team subscription held by the connection `wsp`, if it
|
||||
is subscribed to `team-id`.
|
||||
|
||||
Closing the channel is what stops the relay, so no further team or
|
||||
organization traffic reaches this connection.
|
||||
|
||||
Does nothing when the connection is subscribed to a different team,
|
||||
so it is safe to call for a subscription that is already gone."
|
||||
[{:keys [::mbus/msgbus]} {:keys [::ws/state]} team-id]
|
||||
(let [subs (::team-subscription @state)]
|
||||
(when (= (:team-id subs) team-id)
|
||||
(let [ch (:channel subs)]
|
||||
(sp/close! ch)
|
||||
(mbus/purge! msgbus [ch])
|
||||
(swap! state dissoc ::team-subscription)))))
|
||||
|
||||
|
||||
(defn- schedule-revalidation
|
||||
"Rearms `task` every `interval-ms` for as long as `running?` 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. A task whose `running?`
|
||||
turned false stops rearming."
|
||||
[running? interval-ms task]
|
||||
(when @running?
|
||||
(px/schedule interval-ms
|
||||
(fn []
|
||||
(when @running?
|
||||
(task)
|
||||
(schedule-revalidation running? interval-ms task))))))
|
||||
|
||||
(defn- start-relay
|
||||
"Forwards `channel` into the client output channel of `wsp`, and keeps
|
||||
a re-check of `authorized?` running on the interval. When access no
|
||||
longer holds, `close!` tears the subscription down, which closes the
|
||||
channel and ends the loop.
|
||||
|
||||
`on-forward` runs for every forwarded message, which is how the file
|
||||
relay announces presence."
|
||||
[_ {:keys [::ws/output-ch] :as wsp} channel authorized? close! on-forward]
|
||||
(let [running? (atom true)]
|
||||
(schedule-revalidation running?
|
||||
(inst-ms (revalidation-interval))
|
||||
(fn []
|
||||
(when-not (still-authorized? authorized?)
|
||||
(l/info :hint
|
||||
"closing websocket subscription on re-check"
|
||||
:profile-id (::profile-id wsp))
|
||||
(reset! running? false)
|
||||
(close!))))
|
||||
(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)))))
|
||||
|
||||
;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;
|
||||
;; WEBSOCKET HANDLER
|
||||
;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;
|
||||
|
||||
@ -205,10 +334,10 @@
|
||||
nil)))
|
||||
|
||||
(defmethod handle-message :subscribe-team
|
||||
[cfg {:keys [::ws/id ::ws/state ::ws/output-ch ::session-id ::profile-id]} {:keys [team-id] :as params}]
|
||||
[cfg {:keys [::ws/id ::ws/state ::session-id ::profile-id] :as wsp} {:keys [team-id] :as params}]
|
||||
(l/trace :fn "handle-message" :event "subscribe-team" :team-id team-id :conn-id id)
|
||||
(teams/check-read-permissions! cfg profile-id team-id)
|
||||
(let [prev-subs (get @state ::team-subscription)
|
||||
(let [prev-subs (::team-subscription @state)
|
||||
organization-id (get-team-organization-id cfg team-id)
|
||||
;; Resolved server-side so a client only hears its readable team's org
|
||||
topics (cond-> [team-id]
|
||||
@ -217,8 +346,13 @@
|
||||
channel (sp/chan :buf (sp/dropping-buffer 64)
|
||||
:xf (remove #(= (:session-id %) session-id)))]
|
||||
|
||||
(sp/pipe channel output-ch false)
|
||||
(mbus/sub! (::mbus/msgbus cfg) :topics topics :chan channel)
|
||||
;; The team topic also carries library change diffs, so a
|
||||
;; subscription that outlives access to the team leaks content, not
|
||||
;; just presence.
|
||||
(start-relay cfg wsp channel
|
||||
#(teams/has-read-permissions? cfg profile-id team-id)
|
||||
#(close-team-subscription cfg wsp team-id)
|
||||
nil)
|
||||
|
||||
(let [subs {:team-id team-id
|
||||
:organization-id organization-id
|
||||
@ -226,14 +360,15 @@
|
||||
:topic team-id}]
|
||||
(swap! state assoc ::team-subscription subs))
|
||||
|
||||
(mbus/sub! (::mbus/msgbus cfg) :topics topics :chan channel)
|
||||
|
||||
;; Close previous subscription if exists
|
||||
(when-let [ch (:channel prev-subs)]
|
||||
(sp/close! ch)
|
||||
(mbus/purge! (::mbus/msgbus cfg) [ch]))))
|
||||
|
||||
|
||||
(defmethod handle-message :subscribe-file
|
||||
[cfg {:keys [::ws/id ::ws/state ::ws/output-ch ::session-id ::profile-id]} {:keys [file-id] :as params}]
|
||||
[cfg {:keys [::ws/id ::ws/state ::session-id ::profile-id] :as wsp} {:keys [file-id] :as params}]
|
||||
(l/trace :fn "handle-message" :event "subscribe-file" :file-id file-id :conn-id id)
|
||||
(bfc/check-file-exists cfg file-id)
|
||||
(files/check-read-permissions! cfg profile-id file-id)
|
||||
@ -244,77 +379,34 @@
|
||||
(let [subs {:file-id file-id :channel fch :topic file-id}]
|
||||
(swap! state assoc ::file-subscription subs))
|
||||
|
||||
(start-relay cfg wsp fch
|
||||
#(files/has-read-permissions? cfg profile-id file-id)
|
||||
#(close-file-subscription cfg wsp file-id)
|
||||
(fn [message]
|
||||
(when (contains? #{:join-file :leave-file :disconnect}
|
||||
(:type message))
|
||||
(mbus/pub! (::mbus/msgbus cfg)
|
||||
:topic file-id
|
||||
:message {:type :presence
|
||||
:file-id file-id
|
||||
:session-id session-id
|
||||
:profile-id profile-id}))))
|
||||
|
||||
;; Close previous subscription if exists
|
||||
(when-let [ch (:channel psub)]
|
||||
(sp/close! ch)
|
||||
(mbus/purge! (::mbus/msgbus cfg) [ch]))
|
||||
|
||||
(sp/go-loop []
|
||||
(when-let [{:keys [type] :as message} (sp/take! fch)]
|
||||
(sp/put! output-ch message)
|
||||
(when (or (= :join-file type)
|
||||
(= :leave-file type)
|
||||
(= :disconnect type))
|
||||
(let [message {:type :presence
|
||||
:file-id file-id
|
||||
:session-id session-id
|
||||
:profile-id profile-id}]
|
||||
(mbus/pub! (::mbus/msgbus cfg)
|
||||
:topic file-id
|
||||
:message message)))
|
||||
(recur)))
|
||||
|
||||
;; Subscribe to file topic
|
||||
(mbus/sub! (::mbus/msgbus cfg) :topic file-id :chan fch)
|
||||
|
||||
;; Notifify the rest of participants of the new connection.
|
||||
(let [message {:type :join-file
|
||||
:file-id file-id
|
||||
:session-id session-id
|
||||
:profile-id profile-id}]
|
||||
(mbus/pub! (::mbus/msgbus cfg) :topic file-id :message message))))
|
||||
|
||||
(defn close-file-subscription
|
||||
"Tears down the file subscription held by the connection `wsp`, if it
|
||||
is subscribed to `file-id`.
|
||||
|
||||
Announces the departure so the remaining participants drop the
|
||||
presence of this session, closes the relay channel (which in turn
|
||||
stops the `:subscribe-file` go-loop, because `take!` on a closed
|
||||
channel returns nil) and removes the subscription from the bus.
|
||||
|
||||
Does nothing when the connection is subscribed to a different file,
|
||||
so it is safe to call for a subscription that is already gone."
|
||||
[{:keys [::mbus/msgbus]} {:keys [::ws/state ::session-id ::profile-id]} file-id]
|
||||
(let [subs (::file-subscription @state)]
|
||||
(when (= (:file-id subs) file-id)
|
||||
(mbus/pub! msgbus
|
||||
:topic file-id
|
||||
:message {:type :leave-file
|
||||
:file-id file-id
|
||||
:session-id session-id
|
||||
:profile-id profile-id})
|
||||
(let [ch (:channel subs)]
|
||||
(sp/close! ch)
|
||||
(mbus/purge! msgbus [ch])
|
||||
(swap! state dissoc ::file-subscription)))))
|
||||
|
||||
(defn close-team-subscription
|
||||
"Tears down the team subscription held by the connection `wsp`, if it
|
||||
is subscribed to `team-id`.
|
||||
|
||||
Closing the channel is what stops the pipe into the client output, so
|
||||
no further team or organization traffic reaches this connection.
|
||||
|
||||
Does nothing when the connection is subscribed to a different team,
|
||||
so it is safe to call for a subscription that is already gone."
|
||||
[{:keys [::mbus/msgbus]} {:keys [::ws/state]} team-id]
|
||||
(let [subs (::team-subscription @state)]
|
||||
(when (= (:team-id subs) team-id)
|
||||
(let [ch (:channel subs)]
|
||||
(sp/close! ch)
|
||||
(mbus/purge! msgbus [ch])
|
||||
(swap! state dissoc ::team-subscription)))))
|
||||
(mbus/pub! (::mbus/msgbus cfg)
|
||||
:topic file-id
|
||||
:message {:type :join-file
|
||||
:file-id file-id
|
||||
:session-id session-id
|
||||
:profile-id profile-id})))
|
||||
|
||||
(defmethod handle-message :unsubscribe-file
|
||||
[cfg {:keys [::ws/id] :as wsp} {:keys [file-id] :as params}]
|
||||
|
||||
@ -6,7 +6,9 @@
|
||||
|
||||
(ns backend-tests.http-websocket-test
|
||||
(:require
|
||||
[app.common.time :as ct]
|
||||
[app.common.uuid :as uuid]
|
||||
[app.config :as cf]
|
||||
[app.db :as db]
|
||||
[app.http.websocket :as ws]
|
||||
[app.msgbus :as mbus]
|
||||
@ -742,3 +744,109 @@
|
||||
:id link-id}))]
|
||||
(t/is (some? link-id))
|
||||
(t/is (empty? events)))))
|
||||
|
||||
;; --- PERIODIC REVALIDATION
|
||||
;;
|
||||
;; The announcement path is the primary mechanism. This interval is the
|
||||
;; safety net for access changes that fire no command here, such as
|
||||
;; Nitrate transferring the ownership of an organization, or rows
|
||||
;; deleted straight over SQL.
|
||||
|
||||
(t/deftest revalidation-interval-defaults-to-five-minutes
|
||||
(t/is (= (ct/duration {:minutes 5}) ws/default-revalidation-interval))
|
||||
(t/testing "the default is a code constant, not a config.clj default"
|
||||
(t/is (not (contains? cf/default :subscription-revalidation-interval)))))
|
||||
|
||||
(t/deftest revalidation-is-configurable
|
||||
(with-redefs [cf/get (fn [k default] (if (= :subscription-revalidation-interval k)
|
||||
(ct/duration {:seconds 30})
|
||||
default))]
|
||||
(t/is (= (ct/duration {:seconds 30}) (ws/revalidation-interval)))))
|
||||
|
||||
(t/deftest interval-recheck-drops-a-subscription-without-any-announcement
|
||||
(let [{:keys [owner editor team file]} (editor-in-team! 1 2)]
|
||||
|
||||
(with-redefs [cf/get (fn [k default]
|
||||
(if (= :subscription-revalidation-interval k)
|
||||
(ct/duration {:millis 150})
|
||||
default))]
|
||||
(with-clean-registry
|
||||
(fn []
|
||||
(let [conn (-> (open-registered-connection (:id editor))
|
||||
(subscribe :subscribe-file {:file-id (:id file)}))
|
||||
out (:output conn)]
|
||||
|
||||
(t/testing "the subscription is live at first"
|
||||
(t/is (some? (subscribed-file conn))))
|
||||
|
||||
;; No command runs and no announcement is published: access is
|
||||
;; taken away the way an external change would take it.
|
||||
(th/command! {::th/type :delete-team-member
|
||||
::rpc/profile-id (:id owner)
|
||||
:team-id (:id team)
|
||||
:member-id (:id editor)})
|
||||
|
||||
(t/testing "the interval re-check drops it anyway"
|
||||
(t/is (wait-until #(nil? (subscribed-file conn)) 3000)))
|
||||
|
||||
(t/testing "and nothing reaches the client afterwards"
|
||||
(mbus/pub! (::mbus/msgbus th/*system*)
|
||||
:topic (:id file)
|
||||
:message {:type :file-change :file-id (:id file) :revn 9})
|
||||
(t/is (nil? (poll-msg! out :file-change
|
||||
revocation-timeout-ms))))))))))
|
||||
|
||||
(t/deftest interval-recheck-keeps-a-subscription-that-still-holds
|
||||
(let [{:keys [owner editor team file]} (editor-in-team! 1 2)]
|
||||
|
||||
(with-redefs [cf/get (fn [k default]
|
||||
(if (= :subscription-revalidation-interval k)
|
||||
(ct/duration {:millis 150})
|
||||
default))]
|
||||
(with-clean-registry
|
||||
(fn []
|
||||
(let [conn (-> (open-registered-connection (:id editor))
|
||||
(subscribe :subscribe-file {:file-id (:id file)}))
|
||||
out (:output conn)]
|
||||
|
||||
(t/testing "a downgrade to viewer still allows reading"
|
||||
(t/is (th/success? (th/command! {::th/type :update-team-member-role
|
||||
::rpc/profile-id (:id owner)
|
||||
:team-id (:id team)
|
||||
:member-id (:id editor)
|
||||
:role :viewer}))))
|
||||
|
||||
;; Wait for at least two interval ticks, so a re-check that
|
||||
;; wrongly closed the subscription would have run.
|
||||
(Thread/sleep 500)
|
||||
|
||||
(t/testing "the subscription survives the interval"
|
||||
(t/is (some? (subscribed-file conn))))
|
||||
|
||||
(t/testing "and traffic still flows"
|
||||
(mbus/pub! (::mbus/msgbus th/*system*)
|
||||
:topic (:id file)
|
||||
:message {:type :file-change :file-id (:id file) :revn 3})
|
||||
(t/is (some? (poll-msg! out :file-change
|
||||
revocation-timeout-ms))))))))))
|
||||
|
||||
(t/deftest interval-recheck-covers-the-team-subscription-too
|
||||
(let [{:keys [editor team]} (editor-in-team! 1 2)]
|
||||
|
||||
(with-redefs [cf/get (fn [k default]
|
||||
(if (= :subscription-revalidation-interval k)
|
||||
(ct/duration {:millis 150})
|
||||
default))]
|
||||
(with-clean-registry
|
||||
(fn []
|
||||
(let [conn (-> (open-registered-connection (:id editor))
|
||||
(subscribe :subscribe-team {:team-id (:id team)}))]
|
||||
|
||||
(t/is (some? (subscribed-team conn)))
|
||||
|
||||
(th/command! {::th/type :leave-team
|
||||
::rpc/profile-id (:id editor)
|
||||
:id (:id team)})
|
||||
|
||||
(t/testing "the team subscription is dropped by the interval"
|
||||
(t/is (wait-until #(nil? (subscribed-team conn)) 3000)))))))))
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user