<commit_before>(ns replikativ.stress-test
  (:require [clojure.test :refer :all]
            [clojure.core.async :as async :refer [close! timeout]]
            [superv.async :refer [<?? S]]
            [kabel.peer :refer [start stop]]
            [kabel.middleware.transit :refer [transit]]
            [konserve.memory :refer [new-mem-store]]
            [konserve.filestore :refer [new-fs-store]]
            [replikativ.crdt.materialize :refer [get-crdt]]
            [replikativ
             [peer :refer [server-peer]]
             [stage :refer [connect! create-stage! subscribe-crdts!]]]
            [replikativ.crdt.cdvcs.stage :as s]
            [replikativ.p2p
             [fetch :refer [fetch]]
             [hash :refer [ensure-hash]]
             [hooks :refer [hook]]]))


(require '[taoensso.timbre :as timbre])
(def prev-level (timbre/*config* :level))
(timbre/set-level! :warn)

(defn setup []
  ;; hooking map
  (def hooks (atom {[#".*"
                     #uuid "790f85e2-b48a-47be-b2df-6ad9ccbc73d6"]
                    [["mail:<EMAIL>"
                      #uuid "790f85e2-b48a-47be-b2df-6ad9ccbc73d6"]]}))

  ;; setup two peers with stores and a single commit in mail:<EMAIL> and mail:b@mail.com
  (def store-a (<?? S (new-mem-store)))

  (def store-b (<?? S (new-mem-store)))

  (def store-c (<?? S (new-mem-store)))

  (def peer-a (<?? S (server-peer S store-a "ws://127.0.0.1:9090"
                                  ;; include hooking middleware in peer-a
                                  :middleware (comp fetch
                                                    (partial hook hooks)
                                                    ensure-hash
                                                    )
                                  :id "PEER A")))

  (def log-b (atom {}))

  (def peer-b (<?? S (server-peer S store-b "ws://127.0.0.1:9091"
                                  :middleware fetch
                                  :id "PEER B")))


  (def peer-c (<?? S (server-peer S store-c "ws://127.0.0.1:9092"
                                  :middleware fetch
                                  :id "PEER C")))

  (start peer-a)
  (start peer-b)
  (start peer-c)

  (def stage-a (<?? S (create-stage! "mail:<EMAIL>" peer-a)))

  (<?? S (s/create-cdvcs! stage-a
                        :user "mail:<EMAIL>"
                        :id #uuid "790f85e2-b48a-47be-b2df-6ad9ccbc73d6"))
  (<?? S (s/create-cdvcs! stage-a
                        :user "mail:<EMAIL>"
                        :id #uuid "790f85e2-b48a-47be-b2df-6ad9ccbc73d6"))

  (<?? S (connect! stage-a "ws://127.0.0.1:9091" :retries 0))

  (<?? S (connect! stage-a "ws://127.0.0.1:9092" :retries 0))

  (def stage-b (<?? S (create-stage! "mail:<EMAIL>" peer-b)))

  (<?? S (s/create-cdvcs! stage-b
                        :user "mail:<EMAIL>"
                        :id #uuid "790f85e2-b48a-47be-b2df-6ad9ccbc73d6"))
  (<?? S (s/create-cdvcs! stage-b
                        :user "mail:<EMAIL>"
                        :id #uuid "790f85e2-b48a-47be-b2df-6ad9ccbc73d6"))

  (<?? S (connect! stage-b "ws://127.0.0.1:9092" :retries 0)))

(defn stop-all []
  (stop peer-a)
  (stop peer-b)
  (stop peer-c))


(deftest test-stress-two-peers
  (testing "Always block around transact and exert backpressure."
    (let [st (.getTime (java.util.Date.))]
      (doseq [i (range 100)]
        (<?? S (s/transact! stage-b
                            ["mail:<EMAIL>" #uuid "790f85e2-b48a-47be-b2df-6ad9ccbc73d6"]
                            [['+ i]])))
      (println "Time taken: " (- (.getTime (java.util.Date.)) st) " ms"))
    (<?? S (timeout 10000))
    (is (= (->
            (<?? S (get-crdt S store-a (<?? S (new-mem-store))
                             ["mail:<EMAIL>" #uuid "790f85e2-b48a-47be-b2df-6ad9ccbc73d6"]))
            (get-in [:state :commit-graph])
            count)
           101))
    (is (= (->
            (<?? S (get-crdt S store-a (<?? S (new-mem-store))
                             ["mail:<EMAIL>" #uuid "790f85e2-b48a-47be-b2df-6ad9ccbc73d6"]))
            (get-in [:state :heads])
            count)
           1))
    (is (= (->
            (<?? S (get-crdt S store-b (<?? S (new-mem-store))
                             ["mail:<EMAIL>" #uuid "790f85e2-b48a-47be-b2df-6ad9ccbc73d6"]))
            (get-in [:state :commit-graph])
            count)
           101))
    (is (=
         (->
          (<?? S (get-crdt S store-c (<?? S (new-mem-store))
                           ["mail:<EMAIL>" #uuid "790f85e2-b48a-47be-b2df-6ad9ccbc73d6"]))
          (get-in [:state :commit-graph])
          count))
        101)))



(deftest test-stress-thread-safety-stage
  (doseq [i (range 100)]
    (s/transact! stage-b
                 ["mail:<EMAIL>" #uuid "790f85e2-b48a-47be-b2df-6ad9ccbc73d6"]
                 [['+ i]]))
  (<?? S (timeout 10000))
  (is (=
       (->
        (<?? S (get-crdt S store-a (<?? S (new-mem-store))
                         ["mail:<EMAIL>" #uuid "790f85e2-b48a-47be-b2df-6ad9ccbc73d6"]))
        (get-in [:state :commit-graph])
        count)
       101))
  (is (=
       (->
        (<?? S (get-crdt S store-a (<?? S (new-mem-store))
                         ["mail:<EMAIL>" #uuid "790f85e2-b48a-47be-b2df-6ad9ccbc73d6"]))
        (get-in [:state :heads])
        count)
       1))
  (is (=
       (->
        (<?? S (get-crdt S store-b (<?? S (new-mem-store))
                         ["mail:<EMAIL>" #uuid "790f85e2-b48a-47be-b2df-6ad9ccbc73d6"]))
        (get-in [:state :commit-graph])
        count)
       101))
  (is (= (->
          (<?? S (get-crdt S store-c (<?? S (new-mem-store))
                           ["mail:<EMAIL>" #uuid "790f85e2-b48a-47be-b2df-6ad9ccbc73d6"]))
          (get-in [:state :commit-graph])
          count)
         101)))


(defn each-fixture [f]
  (try
    (setup)
    (f)
    (finally
      (stop-all))))

(use-fixtures :each each-fixture)

(timbre/set-level! prev-level)

(comment
  (->> (get-in @log-a [:pre-fetch :in])
       (filter #(= (:type %) :pub/downstream))
       #_(filter #(not= (:sender %) "STAGE-4f64"))
       (map (fn [{{{:keys [heads]} :op} :downstream
                  id :id :as op}]
              id #_[id op]))
       #_(filter (fn [[id h]] (= id 1))))



  ;; but even if you don't we must have robust behaviour
  (doseq [i (range 100)]
    (s/transact! stage-b
                 ["mail:<EMAIL>" #uuid "790f85e2-b48a-47be-b2df-6ad9ccbc73d6"]
                 [['+ i]]))

  (<?? (timeout 500))

  (require '[clojure.set :as set])

  (set/difference (set (keys @(:state store-b)))
                  (set (keys @(:state store-c)))))
<commit_msg>Move logging configuration in stress test.<commit_after>(ns replikativ.stress-test
  (:require [clojure.test :refer :all]
            [clojure.core.async :as async :refer [close! timeout]]
            [superv.async :refer [<?? S]]
            [kabel.peer :refer [start stop]]
            [kabel.middleware.transit :refer [transit]]
            [konserve.memory :refer [new-mem-store]]
            [konserve.filestore :refer [new-fs-store]]
            [replikativ.crdt.materialize :refer [get-crdt]]
            [replikativ
             [peer :refer [server-peer]]
             [stage :refer [connect! create-stage! subscribe-crdts!]]]
            [replikativ.crdt.cdvcs.stage :as s]
            [replikativ.p2p
             [fetch :refer [fetch]]
             [hash :refer [ensure-hash]]
             [hooks :refer [hook]]]
            [taoensso.timbre :as timbre]))


(defn setup []
  ;; hooking map
  (def hooks (atom {[#".*"
                     #uuid "790f85e2-b48a-47be-b2df-6ad9ccbc73d6"]
                    [["mail:<EMAIL>"
                      #uuid "790f85e2-b48a-47be-b2df-6ad9ccbc73d6"]]}))

  ;; setup two peers with stores and a single commit in mail:<EMAIL> and mail:<EMAIL>
  (def store-a (<?? S (new-mem-store)))

  (def store-b (<?? S (new-mem-store)))

  (def store-c (<?? S (new-mem-store)))

  (def peer-a (<?? S (server-peer S store-a "ws://127.0.0.1:9090"
                                  ;; include hooking middleware in peer-a
                                  :middleware (comp fetch
                                                    (partial hook hooks)
                                                    ensure-hash
                                                    )
                                  :id "PEER A")))

  (def log-b (atom {}))

  (def peer-b (<?? S (server-peer S store-b "ws://127.0.0.1:9091"
                                  :middleware fetch
                                  :id "PEER B")))


  (def peer-c (<?? S (server-peer S store-c "ws://127.0.0.1:9092"
                                  :middleware fetch
                                  :id "PEER C")))

  (start peer-a)
  (start peer-b)
  (start peer-c)

  (def stage-a (<?? S (create-stage! "mail:<EMAIL>" peer-a)))

  (<?? S (s/create-cdvcs! stage-a
                        :user "mail:<EMAIL>"
                        :id #uuid "790f85e2-b48a-47be-b2df-6ad9ccbc73d6"))
  (<?? S (s/create-cdvcs! stage-a
                        :user "mail:<EMAIL>"
                        :id #uuid "790f85e2-b48a-47be-b2df-6ad9ccbc73d6"))

  (<?? S (connect! stage-a "ws://127.0.0.1:9091" :retries 0))

  (<?? S (connect! stage-a "ws://127.0.0.1:9092" :retries 0))

  (def stage-b (<?? S (create-stage! "mail:<EMAIL>" peer-b)))

  (<?? S (s/create-cdvcs! stage-b
                        :user "mail:<EMAIL>"
                        :id #uuid "790f85e2-b48a-47be-b2df-6ad9ccbc73d6"))
  (<?? S (s/create-cdvcs! stage-b
                        :user "mail:<EMAIL>"
                        :id #uuid "790f85e2-b48a-47be-b2df-6ad9ccbc73d6"))

  (<?? S (connect! stage-b "ws://127.0.0.1:9092" :retries 0)))

(defn stop-all []
  (stop peer-a)
  (stop peer-b)
  (stop peer-c))


(deftest test-stress-two-peers
  (testing "Always block around transact and exert backpressure."
    (let [st (.getTime (java.util.Date.))]
      (doseq [i (range 100)]
        (<?? S (s/transact! stage-b
                            ["mail:<EMAIL>" #uuid "790f85e2-b48a-47be-b2df-6ad9ccbc73d6"]
                            [['+ i]])))
      (println "Time taken: " (- (.getTime (java.util.Date.)) st) " ms"))
    (<?? S (timeout 10000))
    (is (= (->
            (<?? S (get-crdt S store-a (<?? S (new-mem-store))
                             ["mail:<EMAIL>" #uuid "790f85e2-b48a-47be-b2df-6ad9ccbc73d6"]))
            (get-in [:state :commit-graph])
            count)
           101))
    (is (= (->
            (<?? S (get-crdt S store-a (<?? S (new-mem-store))
                             ["mail:<EMAIL>" #uuid "790f85e2-b48a-47be-b2df-6ad9ccbc73d6"]))
            (get-in [:state :heads])
            count)
           1))
    (is (= (->
            (<?? S (get-crdt S store-b (<?? S (new-mem-store))
                             ["mail:<EMAIL>" #uuid "790f85e2-b48a-47be-b2df-6ad9ccbc73d6"]))
            (get-in [:state :commit-graph])
            count)
           101))
    (is (=
         (->
          (<?? S (get-crdt S store-c (<?? S (new-mem-store))
                           ["mail:<EMAIL>" #uuid "790f85e2-b48a-47be-b2df-6ad9ccbc73d6"]))
          (get-in [:state :commit-graph])
          count))
        101)))



(deftest test-stress-thread-safety-stage
  (doseq [i (range 100)]
    (s/transact! stage-b
                 ["mail:<EMAIL>" #uuid "790f85e2-b48a-47be-b2df-6ad9ccbc73d6"]
                 [['+ i]]))
  (<?? S (timeout 10000))
  (is (=
       (->
        (<?? S (get-crdt S store-a (<?? S (new-mem-store))
                         ["mail:<EMAIL>" #uuid "790f85e2-b48a-47be-b2df-6ad9ccbc73d6"]))
        (get-in [:state :commit-graph])
        count)
       101))
  (is (=
       (->
        (<?? S (get-crdt S store-a (<?? S (new-mem-store))
                         ["mail:<EMAIL>" #uuid "790f85e2-b48a-47be-b2df-6ad9ccbc73d6"]))
        (get-in [:state :heads])
        count)
       1))
  (is (=
       (->
        (<?? S (get-crdt S store-b (<?? S (new-mem-store))
                         ["mail:<EMAIL>" #uuid "790f85e2-b48a-47be-b2df-6ad9ccbc73d6"]))
        (get-in [:state :commit-graph])
        count)
       101))
  (is (= (->
          (<?? S (get-crdt S store-c (<?? S (new-mem-store))
                           ["mail:<EMAIL>" #uuid "790f85e2-b48a-47be-b2df-6ad9ccbc73d6"]))
          (get-in [:state :commit-graph])
          count)
         101)))


(defn each-fixture [f]
  (try
    (def prev-level (timbre/*config* :level))
    (timbre/set-level! :warn)
    (setup)
    (f)
    (finally
      (stop-all)
      (timbre/set-level! prev-level))))

(use-fixtures :each each-fixture)


(comment
  (->> (get-in @log-a [:pre-fetch :in])
       (filter #(= (:type %) :pub/downstream))
       #_(filter #(not= (:sender %) "STAGE-4f64"))
       (map (fn [{{{:keys [heads]} :op} :downstream
                  id :id :as op}]
              id #_[id op]))
       #_(filter (fn [[id h]] (= id 1))))



  ;; but even if you don't we must have robust behaviour
  (doseq [i (range 100)]
    (s/transact! stage-b
                 ["mail:<EMAIL>" #uuid "790f85e2-b48a-47be-b2df-6ad9ccbc73d6"]
                 [['+ i]]))

  (<?? (timeout 500))

  (require '[clojure.set :as set])

  (set/difference (set (keys @(:state store-b)))
                  (set (keys @(:state store-c)))))

{'fraction_non_alphanumeric': 0.18236352729454108, 'fraction_numerical': 0.05728854229154169, 'mean_word_length': 1.551071155317521, 'pattern_counts': {'":': 0, '<': 114, '<?xml version=': 0, '>': 62, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 13, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 26, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '3287650', 'n_tokens_mistral': 5392, 'n_tokens_neox': 4876, 'n_words': 952}