diff --git a/backend/src/app/binfile/common.clj b/backend/src/app/binfile/common.clj index 327f02b104..c51b4a0351 100644 --- a/backend/src/app/binfile/common.clj +++ b/backend/src/app/binfile/common.clj @@ -308,6 +308,20 @@ AND deleted_at IS NULL ORDER BY revn ASC") +(def ^:private sql:get-branch-oplog-depth + "SELECT coalesce(sum(changes_count), 0) AS depth + FROM file_branch_change + WHERE file_id = ? + AND deleted_at IS NULL") + +(defn get-branch-oplog-depth + "The branch's op log depth: the number of changes the whole log holds, + which is the quantity a size gate limits. Every row stamps its own count + at write time, so this answers without decoding a change vector the + derive in the same request decodes again." + [{:keys [::db/conn] :as cfg} branch-file-id] + (:depth (db/exec-one! conn [sql:get-branch-oplog-depth branch-file-id]))) + (defn get-branch-changes "The branch's op log as decoded change batches, oldest first. This is the complete delta between the branch and its merge base, which is what diff --git a/backend/src/app/migrations.clj b/backend/src/app/migrations.clj index 4190815648..4ffc31dadf 100644 --- a/backend/src/app/migrations.clj +++ b/backend/src/app/migrations.clj @@ -11,6 +11,7 @@ [app.db :as db] [app.migrations.clj.migration-0023 :as mg0023] [app.migrations.clj.migration-0145 :as mg0145] + [app.migrations.clj.migration-0805 :as mg0805] [app.util.migrations :as mg] [integrant.core :as ig])) @@ -535,7 +536,13 @@ :fn (mg/resource "app/migrations/sql/0802-add-file-branch-change-table.sql")} {:name "0803-add-file-branch-change-media-id-map" - :fn (mg/resource "app/migrations/sql/0803-add-file-branch-change-media-id-map.sql")}]) + :fn (mg/resource "app/migrations/sql/0803-add-file-branch-change-media-id-map.sql")} + + {:name "0804-add-file-branch-change-changes-count" + :fn (mg/resource "app/migrations/sql/0804-add-file-branch-change-changes-count.sql")} + + {:name "0805-stamp-file-branch-change-changes-count" + :fn mg0805/migrate}]) (defn apply-migrations! [pool name migrations] diff --git a/backend/src/app/migrations/clj/migration_0805.clj b/backend/src/app/migrations/clj/migration_0805.clj new file mode 100644 index 0000000000..63159ba305 --- /dev/null +++ b/backend/src/app/migrations/clj/migration_0805.clj @@ -0,0 +1,38 @@ +;; This Source Code Form is subject to the terms of the Mozilla Public +;; License, v. 2.0. If a copy of the MPL was not distributed with this +;; file, You can obtain one at http://mozilla.org/MPL/2.0/. +;; +;; Copyright (c) KALEIDOS SUBSIDIARY SL + +(ns app.migrations.clj.migration-0805 + "Fill `file_branch_change.changes_count` from the stored change vectors + and make it required." + (:require + [app.common.logging :as l] + [app.db :as db] + [app.util.blob :as blob])) + +(def ^:private sql:get-rows + "SELECT id, changes FROM file_branch_change ORDER BY created_at") + +(defn- stamp-count! + [conn {:keys [id changes]}] + (db/update! conn :file-branch-change + {:changes-count (count (blob/decode changes))} + {:id id} + {::db/return-keys false})) + +(defn migrate + [conn] + (let [rows (db/plan conn [sql:get-rows]) + total (reduce (fn [total row] + (stamp-count! conn row) + (inc total)) + 0 + rows)] + (l/inf :hint "stamped op log row counts" :rows total) + ;; `NOT NULL` is the guard on the writers: an insert that omits the + ;; count fails rather than reading as a row that holds nothing. + (db/exec! conn ["ALTER TABLE file_branch_change + ALTER COLUMN changes_count SET NOT NULL + , ALTER COLUMN changes_count DROP DEFAULT"]))) diff --git a/backend/src/app/migrations/sql/0804-add-file-branch-change-changes-count.sql b/backend/src/app/migrations/sql/0804-add-file-branch-change-changes-count.sql new file mode 100644 index 0000000000..2704d12b6e --- /dev/null +++ b/backend/src/app/migrations/sql/0804-add-file-branch-change-changes-count.sql @@ -0,0 +1,18 @@ +-- The number of changes an op log row carries, so the depth gate can sum +-- the log without decoding it. +-- +-- The gate that bounds an operation by `branching-max-oplog-changes` runs +-- ahead of every branch integration, and the derive in the same request +-- replays the rows anyway, so reading the count off the row keeps the gate +-- from decoding a log the derive decodes a moment later. The count is not +-- recoverable from the row's bytes: the change vector is one compressed +-- blob, and its length in bytes says nothing about the number of changes +-- beside a squash that emits tens of thousands for one page. +-- +-- The migration that follows this one fills the column from the stored +-- blobs and then makes it NOT NULL, so a writer that omits the count fails +-- instead of recording an empty log. The live rows' sum is the log's +-- depth, the quantity `files_branch.clj::check-oplog-depth-limit!` limits. + +ALTER TABLE file_branch_change + ADD COLUMN IF NOT EXISTS changes_count integer; diff --git a/backend/src/app/rpc/commands/files_branch.clj b/backend/src/app/rpc/commands/files_branch.clj index cc1ed85931..656d15f51d 100644 --- a/backend/src/app/rpc/commands/files_branch.clj +++ b/backend/src/app/rpc/commands/files_branch.clj @@ -119,10 +119,12 @@ (defn- check-oplog-depth-limit! "Refuse an operation on a branch whose op log is deeper than the derive - was measured to survive. `changes` is the flattened log." - [changes branch-file-id operation] - (let [limit (branching-limit :branching-max-oplog-changes 100000) - depth (count changes)] + was measured to survive. `depth` is the number of changes the whole log + holds, as `bfc/get-branch-oplog-depth` reads it from the count each op + log row stamps: gating on that instead of the flattened log keeps the + gate from decoding rows the derive decodes a moment later." + [depth branch-file-id operation] + (let [limit (branching-limit :branching-max-oplog-changes 100000)] (when (> depth limit) (ex/raise :type :restriction :code :branching-oplog-limit-exceeded @@ -131,8 +133,7 @@ :operation operation :branch-file-id branch-file-id :limit limit - :actual depth)) - changes)) + :actual depth)))) (def ^:private schema:get-branching-limits [:map {:title "get-branching-limits"}]) @@ -889,8 +890,10 @@ dir (or direction :branch->main) main-file (-> (bfc/get-file cfg (:source-file-id branch) :realize? true) (check-file-size-limits! :compare)) - log (-> (branch-log cfg (:branch-file-id branch)) - (check-oplog-depth-limit! (:branch-file-id branch) :compare)) + _ (check-oplog-depth-limit! (bfc/get-branch-oplog-depth cfg (:branch-file-id branch)) + (:branch-file-id branch) + :compare) + log (branch-log cfg (:branch-file-id branch)) branch-file (bfc/get-file cfg (:branch-file-id branch) :realize? true :include-base-data? true) main-data (:data main-file) @@ -1027,8 +1030,9 @@ :branch-id branch-id)) main-file (-> (bfc/get-file cfg main-id :realize? true) (check-file-size-limits! :merge)) - _ (-> (branch-log cfg branch-file-id) - (check-oplog-depth-limit! branch-file-id :merge)) + _ (check-oplog-depth-limit! (bfc/get-branch-oplog-depth cfg branch-file-id) + branch-file-id + :merge) branch-file (bfc/get-file cfg branch-file-id :realize? true :include-base-data? true) base-data (branch-base-data cfg main-id branch-file) @@ -1254,6 +1258,7 @@ :file-id (:id file) :revn (:revn file) :changes (blob/encode (vec changes)) + :changes-count (count changes) :media-id-map (some-> (not-empty media-map) db/tjson) :data-version (fmg/data-version) :created-at ts @@ -1320,8 +1325,9 @@ (ex/raise :type :conflict :code :file-modified :hint "main was modified, recompute the diff and retry")) - _ (-> (branch-log cfg branch-file-id) - (check-oplog-depth-limit! branch-file-id :update-from-main)) + _ (check-oplog-depth-limit! (bfc/get-branch-oplog-depth cfg branch-file-id) + branch-file-id + :update-from-main) branch-file (bfc/get-file cfg branch-file-id :realize? true) base-raw (get-base-data cfg branch) diff --git a/backend/src/app/rpc/commands/files_update.clj b/backend/src/app/rpc/commands/files_update.clj index 857702b383..b7478a124d 100644 --- a/backend/src/app/rpc/commands/files_update.clj +++ b/backend/src/app/rpc/commands/files_update.clj @@ -440,6 +440,7 @@ :file-id (:id file) :revn (:revn file) :changes (blob/encode changes) + :changes-count (count changes) :media-id-map (media-id-map-from-index conn (:source-file-id branch) media-index) :data-version (fmg/data-version) diff --git a/backend/test/backend_tests/rpc_file_branch_test.clj b/backend/test/backend_tests/rpc_file_branch_test.clj index 37b98e43f3..a4d84a42dc 100644 --- a/backend/test/backend_tests/rpc_file_branch_test.clj +++ b/backend/test/backend_tests/rpc_file_branch_test.clj @@ -1540,6 +1540,81 @@ (t/is (str/includes? (:hint data) "materialise")) (t/is (not (str/includes? (:hint data) "merge"))))))))) +(t/deftest oplog-depth-gate-counts-the-changes-a-row-holds + ;; The gate bounds a log by the changes it holds, and every op log row stamps + ;; that number at write time, so the gate reads it rather than decoding a log + ;; the derive in the same command decodes again. One save carrying two + ;; changes is one row of depth two, which a row count cannot tell from one. + (with-redefs [cf/flags (conj cf/flags :branching)] + (let [profile (th/create-profile* 1 {:is-active true}) + proj-id (:default-project-id profile) + file (th/create-file* 1 {:profile-id (:id profile) + :project-id proj-id + :is-shared false}) + create (:result (th/command! {::th/type :create-file-branch + ::rpc/profile-id (:id profile) + :file-id (:id file) + :name "deep-log"})) + branch-id (:id create) + branch-file-id (:branch-file-id create) + derive! @#'bfc/branch-file-data + cfg {::db/conn th/*pool*} + depth #(bfc/get-branch-oplog-depth cfg %) + decoded-depth #(count (mapcat identity (bfc/get-branch-changes cfg %))) + rows #(th/db-query :file-branch-change {:file-id branch-file-id}) + reads (atom 0) + derives (atom 0) + compare! (fn [] + (with-redefs [bfc/get-branch-changes + (fn [& args] + (swap! reads inc) + (apply bfc/get-branch-changes args)) + bfc/branch-file-data + (fn [& args] + (swap! derives inc) + (apply derive! args))] + (th/command! {::th/type :get-branch-diff + ::rpc/profile-id (:id profile) + :branch-id branch-id})))] + + (let [f (th/db-get :file {:id branch-file-id})] + (th/command! {::th/type :update-file + ::rpc/profile-id (:id profile) + :id branch-file-id + :session-id (uuid/random) + :revn (:revn f) + :vern (:vern f) + :features cfeat/supported-features + :changes [{:type :add-page :id (uuid/random) :name "one"} + {:type :add-page :id (uuid/random) :name "two"}]})) + + (t/testing "a save stamps the changes its row holds" + (t/is (= 1 (count (rows)))) + (t/is (= 2 (count (mapcat identity (bfc/get-branch-changes cfg branch-file-id)))))) + + (t/testing "the stamped depth is the depth of the decoded log" + (t/is (= (decoded-depth branch-file-id) (depth branch-file-id)))) + + (t/testing "the gate refuses on the changes the row holds" + (with-redefs [cf/get (th/config-get-mock {:branching-max-oplog-changes 1})] + (let [out (compare!) + data (ex-data (:error out))] + (t/is (= :restriction (:type data))) + (t/is (= :branching-oplog-limit-exceeded (:code data))) + (t/is (= 2 (:actual data)) "the gate counted rows instead of the depth") + (t/is (zero? @reads) "the refused compare decoded the op log") + (t/is (zero? @derives) "the refused compare derived the branch first")))) + + (t/testing "an update from main stamps the squash it writes" + (apply-change* profile (:id file) {:type :add-page :id (uuid/random) :name "on-main"}) + (let [out (th/command! {::th/type :update-branch-from-main + ::rpc/profile-id (:id profile) + :branch-id branch-id})] + (t/is (nil? (:error out))) + (t/is (= 1 (count (rows))) "the squash is one row") + (t/is (= (decoded-depth branch-file-id) (depth branch-file-id))) + (t/is (= (depth branch-file-id) (:changes-count (first (rows)))))))))) + (t/deftest published-limits-are-the-ones-the-gates-enforce (let [profile (th/create-profile* 1 {:is-active true}) limits (fn [] (th/command! {::th/type :get-branching-limits