run more ops test

deps/db-sync/src/logseq/db_sync/order.cljs

@@ -5,7 +5,9 @@
 (defn fix-duplicate-orders! [conn tx-data tx-meta]
   (let [db @conn
         updates (->> tx-data
-                     (filter (fn [[e a v _tx added]] (and (= a :block/order) added e v))))
+                     (filter (fn [[e a v _tx added]]
+                               (and (= a :block/order) added e v
+                                    (:block/uuid (d/entity @conn e))))))
         groups (group-by (fn [{:keys [e v]}]
                            (let [parent (:block/parent (d/entity db e))]
                              [(:db/id parent) v]))

src/main/frontend/worker/db_sync.cljs

@@ -594,6 +594,9 @@
   (if-let [conn (worker-state/get-datascript-conn repo)]
     (let [tx-data (keep-last-update tx-data*)
           local-txs (pending-txs repo)
+          _ (prn :debug :repo repo
+                 :pending-txs-count (count local-txs)
+                 :pending-txs (map :tx-id local-txs))
           reversed-tx-data (get-reverse-tx-data local-txs)
           has-local-changes? (seq reversed-tx-data)
           *remote-tx-report (atom nil)
@@ -676,9 +679,9 @@
             ;; (prn :debug :remote-tx-data remote-tx-data-set)
             ;; (prn :debug :diff (data/diff remote-tx-data-set
             ;;                              (set normalized)))
-            (remove-pending-txs! repo (map :tx-id local-txs))
             (when (seq normalized-tx-data)
-              (persist-local-tx! repo normalized-tx-data reversed-datoms {:op :rtc-rebase})))))
+              (persist-local-tx! repo normalized-tx-data reversed-datoms {:op :rtc-rebase}))))
+        (remove-pending-txs! repo (map :tx-id local-txs)))
 
       (when tx-report
         (let [asset-uuids (asset-uuids-from-tx (:db-after remote-tx-report) (:tx-data remote-tx-report))]
@@ -845,28 +848,29 @@
 
 (defn enqueue-local-tx!
   [repo {:keys [tx-meta tx-data db-after db-before]}]
-  (let [conn (worker-state/get-datascript-conn repo)
-        db (some-> conn deref)
+  (when-not (:rtc-tx? tx-meta)
+    (let [conn (worker-state/get-datascript-conn repo)
+          db (some-> conn deref)
         ;; FIXME: all ignored properties
-        tx-data' (remove (fn [d] (contains? #{:logseq.property.embedding/hnsw-label-updated-at :block/tx-id} (:a d))) tx-data)]
-    (when (and db (seq tx-data'))
-      (let [normalized (normalize-tx-data db-after db-before tx-data')
-            reversed-datoms (reverse-tx-data tx-data)]
+          tx-data' (remove (fn [d] (contains? #{:logseq.property.embedding/hnsw-label-updated-at :block/tx-id} (:a d))) tx-data)]
+      (when (and db (seq tx-data'))
+        (let [normalized (normalize-tx-data db-after db-before tx-data')
+              reversed-datoms (reverse-tx-data tx-data)]
         ;; (prn :debug :tx-data tx-data'
         ;;      :normalized (normalize-tx-data db-after db-before tx-data'))
-        (persist-local-tx! repo normalized reversed-datoms tx-meta)
-        (when-let [client @worker-state/*db-sync-client]
-          (when (= repo (:repo client))
-            (let [send-queue (:send-queue client)]
-              (swap! send-queue
-                     (fn [prev]
-                       (p/then prev
-                               (fn [_]
-                                 (when-let [current @worker-state/*db-sync-client]
-                                   (when (= repo (:repo current))
-                                     (when-let [ws (:ws current)]
-                                       (when (ws-open? ws)
-                                         (flush-pending! repo current))))))))))))))))
+          (persist-local-tx! repo normalized reversed-datoms tx-meta)
+          (when-let [client @worker-state/*db-sync-client]
+            (when (= repo (:repo client))
+              (let [send-queue (:send-queue client)]
+                (swap! send-queue
+                       (fn [prev]
+                         (p/then prev
+                                 (fn [_]
+                                   (when-let [current @worker-state/*db-sync-client]
+                                     (when (= repo (:repo current))
+                                       (when-let [ws (:ws current)]
+                                         (when (ws-open? ws)
+                                           (flush-pending! repo current)))))))))))))))))
 
 (defn handle-local-tx!
   [repo {:keys [tx-data tx-meta] :as tx-report}]

src/test/frontend/worker/db_sync_sim_test.cljs

@@ -156,7 +156,8 @@
       (when (< local-tx server-t)
         (let [tx (->> (server-pull server local-tx)
                       (mapcat :tx))]
-          (prn :debug :apply-remote-tx :repo repo)
+          (prn :debug :apply-remote-tx :repo repo
+               :tx tx)
           (#'db-sync/apply-remote-tx! repo client tx)
           (client-op/update-local-tx repo server-t)
           (reset! progress? true)))
@@ -166,6 +167,7 @@
         (when (and (seq pending) (= local-tx' server-t'))
           (let [tx-data (build-upload-tx conn pending)
                 tx-ids (mapv :tx-id pending)]
+            (prn :debug :upload :repo repo :tx-data tx-data)
             (when (seq tx-data)
               (server-upload! server local-tx' tx-data)
               (#'db-sync/remove-pending-txs! repo tx-ids)
@@ -328,6 +330,7 @@
       (swap! state update :blocks disj (:block/uuid block))
       {:op :delete-block :uuid (:block/uuid block)})))
 
+;; TODO: add tag/property/migrate/undo/redo ops
 (def ^:private op-table
   [{:name :create-page :weight 6 :f op-create-page!}
    {:name :delete-page :weight 2 :f op-delete-page!}
@@ -449,8 +452,9 @@
                          {:repo repo-b :conn conn-b :client client-b :online? true}
                          {:repo repo-c :conn conn-c :client client-c :online? true}]
                 ;; FIXME:
-                ;; run-ops-opts {:pick-op-opts {:disable-ops #{:move-block}}}
-                run-ops-opts {}]
+                run-ops-opts {:pick-op-opts {:disable-ops #{:move-block}}}
+                ;; run-ops-opts {}
+                ]
             (prn :debug :phase-a)
             ;; Phase A: all online
             (dotimes [_ 60]
@@ -464,12 +468,12 @@
             (let [clients-phase-b [{:repo repo-a :conn conn-a :client client-a :online? true}
                                    {:repo repo-b :conn conn-b :client client-b :online? true}
                                    {:repo repo-c :conn conn-c :client client-c :online? false}]]
-              (dotimes [_ 40]
+              (dotimes [_ 100]
                 (let [client (rand-nth! rng (subvec (vec clients-phase-b) 0 2))
                       state (get repo->state (:repo client))]
                   (run-ops! rng (assoc client :base-uuid base-uuid :state state) 1 history run-ops-opts)
                   (sync-loop! server clients-phase-b)))
-              (dotimes [_ 20]
+              (dotimes [_ 100]
                 (run-ops! rng {:client client-c :conn conn-c :base-uuid base-uuid :state state-c} 1 history run-ops-opts)))
 
             ;; Phase C: reconnect C
@@ -481,12 +485,12 @@
             (let [clients-phase-d [{:repo repo-a :conn conn-a :client client-a :online? false}
                                    {:repo repo-b :conn conn-b :client client-b :online? true}
                                    {:repo repo-c :conn conn-c :client client-c :online? true}]]
-              (dotimes [_ 30]
+              (dotimes [_ 100]
                 (let [client (rand-nth! rng (subvec (vec clients-phase-d) 1 3))
                       state (get repo->state (:repo client))]
                   (run-ops! rng (assoc client :base-uuid base-uuid :state state) 1 history run-ops-opts)
                   (sync-loop! server clients-phase-d)))
-              (dotimes [_ 15]
+              (dotimes [_ 100]
                 (run-ops! rng {:conn conn-a :base-uuid base-uuid :state state-a} 1 history run-ops-opts)))
 
             ;; Final sync