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
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122 | (ns ird.p2p.libp2p-node-test
"Two real in-process libp2p nodes (gossipsub over loopback tcp), always
on: it dials one node from the other, joins the same room from both over
the libp2p transport, runs the hello/verify handshake and exchanges a
message through real pubsub. This is the empirical proof that the pinned
interface-v3 libp2p stack (libp2p 3.x + @libp2p/gossipsub 17.x) actually
delivers frames between nodes (two-node exchange settles in well under a
second locally). A second always-on test puts a third node between them
as a circuit-relay hop: both peers reserve a slot on it, one dials the
other's /p2p-circuit address, and the same handshake and messages flow
through the hop. If your
environment cannot bind loopback tcp, set IRD_P2P_NO_LIBP2P_NODE=1 to skip."
(:require [clojure.string :as str]
[cljs.test :refer [deftest is]]
[cljs.core.async :as async :refer [go-loop <!]]
[ird.p2p :as p2p]
[ird.p2p.crypto :as crypto]
[ird.p2p.descriptor :as desc]
[ird.p2p.impl.libp2p :as impl]
[ird.p2p.test-util :as tu :refer [deftest-async]]
[ird.p2p.transport.libp2p :as lp]))
(def low "00000000000000000000000000000001")
(def high "0f1e2d3c4b5a69788796a5b4c3d2e1f0")
(deftest-async two-real-nodes-signal-through-gossipsub
{:timeout-ms 30000}
(if (tu/env "IRD_P2P_NO_LIBP2P_NODE")
(is true "IRD_P2P_NO_LIBP2P_NODE set — skipping the real libp2p node test")
(let [na (<! (impl/create-node {}))
nb (<! (impl/create-node {}))]
(is (not (tu/error? na)) (some-> na ex-message))
(is (not (tu/error? nb)) (some-> nb ex-message))
;; dial b from a so the two nodes are connected before joining
(is (true? (<! ((:dial na) (<! ((:multiaddrs nb)))))))
(let [secret (crypto/random-secret)
app (fn [n] (p2p/make-app {:host "h" :app-salt "node/v1"
:descriptor (desc/fallback "h")
:transport (lp/transport {:node n})}))
ra (<! (p2p/join (app na) secret {:peer-id low :options {:pacing? false}}))
ca (tu/collector (p2p/events ra))
rb (<! (p2p/join (app nb) secret {:peer-id high :options {:pacing? false}}))
cb (tu/collector (p2p/events rb))]
(is (not (tu/error? ra)) (some-> ra ex-message))
(is (not (tu/error? rb)) (some-> rb ex-message))
;; both peers prove membership over real gossipsub
(is (some? (<! (tu/wait-for cb #(= {:type :peer-joined :peer low} %) 20000))))
(is (some? (<! (tu/wait-for ca #(= {:type :peer-joined :peer high} %) 20000))))
(is (some? (<! (tu/wait-for ca #(= {:type :peer-ready :peer high :via :relayed} %) 20000))))
;; a directed message flows both ways
(is (true? (<! (p2p/send ra high (p2p/json {:text "over real gossip"})))))
(is (some? (<! (tu/wait-for cb #(= "over real gossip" (get-in % [:message :value :text])) 20000))))
(is (true? (<! (p2p/broadcast rb (p2p/json {:reply "ack"})))))
(is (some? (<! (tu/wait-for ca #(= "ack" (get-in % [:message :value :reply])) 20000))))
(<! (p2p/leave ra)) (<! (p2p/leave rb))
(<! ((:stop na))) (<! ((:stop nb)))))))
;; ---- the same exchange through a circuit-relay hop ---------------------------------
(def relayed-low "000000000000000000000000000000a1")
(def relayed-high "f0e1d2c3b4a5968778695a4b3c2d1e0f")
(def circuit-listen ["/ip4/127.0.0.1/tcp/0" "/ip4/127.0.0.1/tcp/0/ws" "/p2p-circuit"])
(defn- circuit-addr [hop-addr n] (str hop-addr "/p2p-circuit/p2p/" (:peer-id n)))
(defn- advertises?
"Channel of true once node N lists multiaddr MA (the reservation lands
asynchronously after the dial), polling up to TIMEOUT-MS; false otherwise."
[n ma timeout-ms]
(let [deadline (+ (js/Date.now) timeout-ms)]
(go-loop []
(cond
(some #{ma} (<! ((:multiaddrs n)))) true
(>= (js/Date.now) deadline) false
:else (do (<! (async/timeout 20)) (recur))))))
(deftest-async relay-hop-two-nodes-signal-through-circuit-relay
{:timeout-ms 30000}
(if (tu/env "IRD_P2P_NO_LIBP2P_NODE")
(is true "IRD_P2P_NO_LIBP2P_NODE set — skipping the real libp2p relay-hop test")
(let [hop (<! (impl/create-node {:relay-server? true}))
na (<! (impl/create-node {:listen circuit-listen}))
nb (<! (impl/create-node {:listen circuit-listen}))]
(is (not (tu/error? hop)) (some-> hop ex-message))
(is (not (tu/error? na)) (some-> na ex-message))
(is (not (tu/error? nb)) (some-> nb ex-message))
(let [hop-addr (first (remove #(str/includes? % "/ws") (<! ((:multiaddrs hop)))))]
;; each peer connects to the hop only, and its reservation shows up as
;; its own relayed address
(is (true? (<! ((:dial na) [hop-addr]))))
(is (true? (<! ((:dial nb) [hop-addr]))))
(is (true? (<! (advertises? na (circuit-addr hop-addr na) 10000))))
(is (true? (<! (advertises? nb (circuit-addr hop-addr nb) 10000))))
;; a reaches b through the hop: the circuit address is the only address
;; of b that a is ever given, and the hop subscribes to no topic, so
;; every frame below crosses the relayed connection
(is (true? (<! ((:dial na) [(circuit-addr hop-addr nb)]))))
(let [secret (crypto/random-secret)
app (fn [n] (p2p/make-app {:host "h" :app-salt "node/v1"
:descriptor (desc/fallback "h")
:transport (lp/transport {:node n})}))
ra (<! (p2p/join (app na) secret {:peer-id relayed-low :options {:pacing? false}}))
ca (tu/collector (p2p/events ra))
rb (<! (p2p/join (app nb) secret {:peer-id relayed-high :options {:pacing? false}}))
cb (tu/collector (p2p/events rb))]
(is (not (tu/error? ra)) (some-> ra ex-message))
(is (not (tu/error? rb)) (some-> rb ex-message))
;; both peers prove membership over gossipsub through the hop
(is (some? (<! (tu/wait-for cb #(= {:type :peer-joined :peer relayed-low} %) 20000))))
(is (some? (<! (tu/wait-for ca #(= {:type :peer-joined :peer relayed-high} %) 20000))))
(is (some? (<! (tu/wait-for ca #(= {:type :peer-ready :peer relayed-high :via :relayed} %) 20000))))
(is (some? (<! (tu/wait-for cb #(= {:type :peer-ready :peer relayed-low :via :relayed} %) 20000))))
;; a directed message flows both ways, then a broadcast
(is (true? (<! (p2p/send ra relayed-high (p2p/json {:text "through the hop"})))))
(is (some? (<! (tu/wait-for cb #(= "through the hop" (get-in % [:message :value :text])) 20000))))
(is (true? (<! (p2p/send rb relayed-low (p2p/json {:text "and back"})))))
(is (some? (<! (tu/wait-for ca #(= "and back" (get-in % [:message :value :text])) 20000))))
(is (true? (<! (p2p/broadcast rb (p2p/json {:reply "ack"})))))
(is (some? (<! (tu/wait-for ca #(= "ack" (get-in % [:message :value :reply])) 20000))))
(<! (p2p/leave ra)) (<! (p2p/leave rb))
(<! ((:stop na))) (<! ((:stop nb))) (<! ((:stop hop))))))))
|