1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106 | (ns ird.p2p.db.soak-test
"A bounded long session on two real persistent nodes, always on: A and B
take turns appending batches of records, each batch awaited on the other
side; B is then closed, A keeps appending, and B comes back (same :path,
same peer id), dials A again and both converge on the full history — the
exact records, in the exact canonical order — and still replicate live.
If your environment cannot bind loopback tcp, set IRD_P2P_DB_NO_NODE=1 to
skip.
This is churn inside one process over loopback. Real network churn —
NATs, lossy links, long partitions, many peers — is out of scope here."
(:require [cljs.test :refer [deftest is testing]]
[cljs.core.async :as async :refer [<!]]
[ird.p2p.db.core :as db]
[ird.p2p.db.test-util :as tu :refer [deftest-async]]))
(def ^:private batches 6)
(def ^:private batch-size 100)
(def ^:private away-size 20)
(defn- batch-payloads [j n] (mapv (fn [i] {:n i :batch j}) (range n)))
(defn- append-all
"Append PAYLOADS to H one after the other. Channel of the records, or of
the first error."
[h payloads]
(async/go-loop [ps payloads acc []]
(if-let [p (first ps)]
(let [r (<! (db/append h p))]
(if (tu/error? r) r (recur (rest ps) (conj acc r))))
acc)))
(defn- canonical? [es]
(and (= es (db/canonical-order es))
(= (range 1 (inc (count es))) (map #(get-in % [:clock :time]) es))))
(deftest-async long-session-converges-across-a-restart
{:timeout-ms 180000}
(if (tu/skip-nodes?)
(is true "IRD_P2P_DB_NO_NODE set — skipping the real-node soak test")
(let [dir-a (tu/tmp-dir "ird-p2p-db-soak-a")
dir-b (tu/tmp-dir "ird-p2p-db-soak-b")
secret (db/random-secret)
opts {:secret secret :app-salt "test/v1" :name "soak"}
a (<! (db/open (assoc opts :path dir-a)))
_ (is (not (tu/error? a)) (some-> a ex-message))
a-addrs (<! (db/multiaddrs a))
b (<! (db/open (assoc opts :path dir-b :dial a-addrs)))
_ (is (not (tu/error? b)) (some-> b ex-message))
ca (tu/collector (db/entries-stream a))
cb (tu/collector (db/entries-stream b))
pa (tu/collector (db/peers-stream a))
pb (tu/collector (db/peers-stream b))
expected (into [] (mapcat #(batch-payloads % batch-size)) (range batches))
away (batch-payloads batches away-size)
last-of? (fn [j n] #(= {:n (dec n) :batch j} (:payload %)))]
(is (some? (<! (tu/wait-for pb #(= :peer-joined (:type %)) 20000))))
(testing "interleaved batches, each one awaited on the other side"
(loop [j 0]
(when (< j batches)
(let [[writer reader-seen] (if (even? j) [a cb] [b ca])
rs (<! (append-all writer (batch-payloads j batch-size)))]
(is (and (not (tu/error? rs)) (= batch-size (count rs))) (str "batch " j " appended"))
(is (some? (<! (tu/wait-for reader-seen (last-of? j batch-size) 20000))) (str "batch " j " replicated"))
(recur (inc j)))))
(is (true? (<! (tu/eventually #(= (count expected) (count @ca) (count @cb)) 20000))) "both streams delivered every record once")
(let [ea (<! (db/entries a)) eb (<! (db/entries b))]
(is (= expected (mapv :payload ea)))
(is (= ea eb))))
(testing "B goes away, A carries on, B comes back and both hold the full history"
(let [pid (db/peer-id b)
_ (<! (db/close b))
rs (<! (append-all a away))
_ (is (and (not (tu/error? rs)) (= away-size (count rs))))
b2 (<! (db/open (assoc opts :path dir-b :dial a-addrs)))
_ (is (not (tu/error? b2)) (some-> b2 ex-message))
cb2 (tu/collector (db/entries-stream b2))
pb2 (tu/collector (db/peers-stream b2))
full (into expected away)]
(is (= pid (db/peer-id b2)))
(is (some? (<! (tu/wait-for cb2 (last-of? batches away-size) 20000))) "B caught up with what it missed")
(is (true? (<! (tu/eventually #(= (count full) (count @cb2)) 20000))) "B's stream replayed and completed the history, each record once")
(let [ea (<! (db/entries a)) eb (<! (db/entries b2))]
(is (= full (mapv :payload ea)) "the exact records in the exact order")
(is (= (mapv :hash ea) (mapv :hash eb)))
(is (= ea eb))
(is (canonical? ea))
(is (canonical? eb)))
(testing "and the session goes on, both ways"
(let [rb (<! (db/append b2 {:n 0 :batch "after-b"}))
_ (is (not (tu/error? rb)) (some-> rb ex-message))
_ (is (some? (<! (tu/wait-for ca #(= "after-b" (get-in % [:payload :batch])) 20000))) "A received B's live record")
ra (<! (db/append a {:n 0 :batch "after-a"}))
_ (is (not (tu/error? ra)) (some-> ra ex-message))
_ (is (some? (<! (tu/wait-for cb2 #(= "after-a" (get-in % [:payload :batch])) 20000))) "B received A's live record")
ea (<! (db/entries a)) eb (<! (db/entries b2))]
(is (= (+ 2 (count full)) (count ea)))
(is (= [(:hash rb) (:hash ra)] (mapv :hash (take-last 2 ea))))
(is (= ea eb))
(is (canonical? ea))))
(is (not-any? #(= :sync-error (:type %)) @pb2))
(<! (db/close b2))))
(is (not-any? #(= :sync-error (:type %)) (concat @pa @pb)))
(<! (db/close a))
(tu/rm-rf dir-a)
(tu/rm-rf dir-b))))
|