Deploying to gh-pages from @ 8f53e8b 🚀

src/subspace_farmer/utils/farmer_piece_getter.rs.html

@@ -129,39 +129,23 @@
 <a href="#129" id="129">129</a>
 <a href="#130" id="130">130</a>
 <a href="#131" id="131">131</a>
-<a href="#132" id="132">132</a>
-<a href="#133" id="133">133</a>
-<a href="#134" id="134">134</a>
-<a href="#135" id="135">135</a>
-<a href="#136" id="136">136</a>
-<a href="#137" id="137">137</a>
-<a href="#138" id="138">138</a>
-<a href="#139" id="139">139</a>
-<a href="#140" id="140">140</a>
-<a href="#141" id="141">141</a>
-<a href="#142" id="142">142</a>
-<a href="#143" id="143">143</a>
-<a href="#144" id="144">144</a>
-<a href="#145" id="145">145</a>
 </pre></div><pre class="rust"><code><span class="kw">use </span><span class="kw">crate</span>::piece_cache::PieceCache;
 <span class="kw">use </span><span class="kw">crate</span>::utils::readers_and_pieces::ReadersAndPieces;
 <span class="kw">use </span><span class="kw">crate</span>::NodeClient;
 <span class="kw">use </span>async_trait::async_trait;
 <span class="kw">use </span>parking_lot::Mutex;
-<span class="kw">use </span>std::collections::HashSet;
 <span class="kw">use </span>std::error::Error;
 <span class="kw">use </span>std::sync::Arc;
 <span class="kw">use </span>subspace_core_primitives::{Piece, PieceIndex};
 <span class="kw">use </span>subspace_farmer_components::plotting::{PieceGetter, PieceGetterRetryPolicy};
 <span class="kw">use </span>subspace_networking::libp2p::kad::RecordKey;
-<span class="kw">use </span>subspace_networking::libp2p::PeerId;
 <span class="kw">use </span>subspace_networking::utils::multihash::ToMultihash;
 <span class="kw">use </span>subspace_networking::utils::piece_provider::{PieceProvider, PieceValidator, RetryPolicy};
-<span class="kw">use </span>subspace_networking::Node;
 <span class="kw">use </span>tracing::{debug, error, trace};
 
+<span class="kw">const </span>MAX_RANDOM_WALK_ROUNDS: usize = <span class="number">35</span>;
+
 <span class="kw">pub struct </span>FarmerPieceGetter&lt;PV, NC&gt; {
-    node: Node,
     piece_provider: PieceProvider&lt;PV&gt;,
     piece_cache: PieceCache,
     node_client: NC,
@@ -170,14 +154,12 @@
 
 <span class="kw">impl</span>&lt;PV, NC&gt; FarmerPieceGetter&lt;PV, NC&gt; {
     <span class="kw">pub fn </span>new(
-        node: Node,
         piece_provider: PieceProvider&lt;PV&gt;,
         piece_cache: PieceCache,
         node_client: NC,
         readers_and_pieces: Arc&lt;Mutex&lt;<span class="prelude-ty">Option</span>&lt;ReadersAndPieces&gt;&gt;&gt;,
     ) -&gt; <span class="self">Self </span>{
         <span class="self">Self </span>{
-            node,
             piece_provider,
             piece_cache,
             node_client,
@@ -216,7 +198,7 @@
         </span><span class="macro">trace!</span>(%piece_index, <span class="string">&quot;Getting piece from DSN L2 cache&quot;</span>);
         <span class="kw">let </span>maybe_piece = <span class="self">self
             </span>.piece_provider
-            .get_piece(piece_index, <span class="self">Self</span>::convert_retry_policy(retry_policy))
+            .get_piece_from_dsn_cache(piece_index, <span class="self">Self</span>::convert_retry_policy(retry_policy))
             .<span class="kw">await</span><span class="question-mark">?</span>;
 
         <span class="kw">if </span>maybe_piece.is_some() {
@@ -258,31 +240,21 @@
         }
 
         <span class="comment">// L1 piece acquisition
-        // TODO: consider using retry policy for L1 lookups as well.
-        </span><span class="macro">trace!</span>(%piece_index, <span class="string">&quot;Getting piece from DSN L1&quot;</span>);
-        <span class="kw">let </span>connected_peers = HashSet::&lt;PeerId&gt;::from_iter(<span class="self">self</span>.node.connected_peers().<span class="kw">await</span><span class="question-mark">?</span>);
-        <span class="kw">if </span>connected_peers.is_empty() {
-            <span class="macro">debug!</span>(%piece_index, <span class="string">&quot;Cannot acquire piece from DSN L1: no connected peers&quot;</span>);
-
-            <span class="kw">return </span><span class="prelude-val">Ok</span>(<span class="prelude-val">None</span>);
-        }
+        </span><span class="macro">trace!</span>(%piece_index, <span class="string">&quot;Getting piece from DSN L1.&quot;</span>);
 
-        <span class="kw">for </span>peer_id <span class="kw">in </span>connected_peers.iter() {
-            <span class="kw">let </span>maybe_piece = <span class="self">self
-                </span>.piece_provider
-                .get_piece_from_peer(<span class="kw-2">*</span>peer_id, piece_index)
-                .<span class="kw">await</span>;
+        <span class="kw">let </span>archival_storage_search_result = <span class="self">self
+            </span>.piece_provider
+            .get_piece_from_archival_storage(piece_index, MAX_RANDOM_WALK_ROUNDS)
+            .<span class="kw">await</span>;
 
-            <span class="kw">if </span>maybe_piece.is_some() {
-                <span class="macro">trace!</span>(%piece_index, %peer_id, <span class="string">&quot;DSN L1 lookup succeeded&quot;</span>);
+        <span class="kw">if </span>archival_storage_search_result.is_some() {
+            <span class="macro">trace!</span>(%piece_index, <span class="string">&quot;DSN L1 lookup succeeded&quot;</span>);
 
-                <span class="kw">return </span><span class="prelude-val">Ok</span>(maybe_piece);
-            }
+            <span class="kw">return </span><span class="prelude-val">Ok</span>(archival_storage_search_result);
         }
 
         <span class="macro">debug!</span>(
             %piece_index,
-            connected_peers=%connected_peers.len(),
             <span class="string">&quot;Cannot acquire piece: all methods yielded empty result&quot;
         </span>);
         <span class="prelude-val">Ok</span>(<span class="prelude-val">None</span>)

src/subspace_farmer/utils/piece_validator.rs.html

@@ -109,10 +109,13 @@
 <a href="#109" id="109">109</a>
 <a href="#110" id="110">110</a>
 <a href="#111" id="111">111</a>
+<a href="#112" id="112">112</a>
+<a href="#113" id="113">113</a>
 </pre></div><pre class="rust"><code><span class="kw">use </span><span class="kw">crate</span>::NodeClient;
 <span class="kw">use </span>async_trait::async_trait;
 <span class="kw">use </span>lru::LruCache;
 <span class="kw">use </span>parking_lot::Mutex;
+<span class="kw">use </span>std::sync::Arc;
 <span class="kw">use </span>subspace_archiving::archiver::is_piece_valid;
 <span class="kw">use </span>subspace_core_primitives::crypto::kzg::Kzg;
 <span class="kw">use </span>subspace_core_primitives::{Piece, PieceIndex, SegmentCommitment, SegmentIndex};
@@ -121,19 +124,20 @@
 <span class="kw">use </span>subspace_networking::Node;
 <span class="kw">use </span>tracing::{error, warn};
 
-<span class="kw">pub struct </span>SegmentCommitmentPieceValidator&lt;NC&gt; {
+<span class="attr">#[derive(Clone)]
+</span><span class="kw">pub struct </span>SegmentCommitmentPieceValidator&lt;NC&gt; {
     dsn_node: Node,
     node_client: NC,
     kzg: Kzg,
-    segment_commitment_cache: Mutex&lt;LruCache&lt;SegmentIndex, SegmentCommitment&gt;&gt;,
+    segment_commitment_cache: Arc&lt;Mutex&lt;LruCache&lt;SegmentIndex, SegmentCommitment&gt;&gt;&gt;,
 }
 
 <span class="kw">impl</span>&lt;NC&gt; SegmentCommitmentPieceValidator&lt;NC&gt; {
     <span class="kw">pub fn </span>new(
         dsn_node: Node,
         node_client: NC,
         kzg: Kzg,
-        segment_commitment_cache: Mutex&lt;LruCache&lt;SegmentIndex, SegmentCommitment&gt;&gt;,
+        segment_commitment_cache: Arc&lt;Mutex&lt;LruCache&lt;SegmentIndex, SegmentCommitment&gt;&gt;&gt;,
     ) -&gt; <span class="self">Self </span>{
         <span class="self">Self </span>{
             dsn_node,

src/subspace_networking/node.rs.html

@@ -602,6 +602,28 @@
 <a href="#602" id="602">602</a>
 <a href="#603" id="603">603</a>
 <a href="#604" id="604">604</a>
+<a href="#605" id="605">605</a>
+<a href="#606" id="606">606</a>
+<a href="#607" id="607">607</a>
+<a href="#608" id="608">608</a>
+<a href="#609" id="609">609</a>
+<a href="#610" id="610">610</a>
+<a href="#611" id="611">611</a>
+<a href="#612" id="612">612</a>
+<a href="#613" id="613">613</a>
+<a href="#614" id="614">614</a>
+<a href="#615" id="615">615</a>
+<a href="#616" id="616">616</a>
+<a href="#617" id="617">617</a>
+<a href="#618" id="618">618</a>
+<a href="#619" id="619">619</a>
+<a href="#620" id="620">620</a>
+<a href="#621" id="621">621</a>
+<a href="#622" id="622">622</a>
+<a href="#623" id="623">623</a>
+<a href="#624" id="624">624</a>
+<a href="#625" id="625">625</a>
+<a href="#626" id="626">626</a>
 </pre></div><pre class="rust"><code><span class="kw">use </span><span class="kw">crate</span>::protocols::request_response::handlers::generic_request_handler::GenericRequest;
 <span class="kw">use </span><span class="kw">crate</span>::protocols::request_response::request_response_factory;
 <span class="kw">use </span><span class="kw">crate</span>::shared::{Command, CreatedSubscription, PeerDiscovered, Shared};
@@ -1004,7 +1026,21 @@
         <span class="kw-2">&amp;</span><span class="self">self</span>,
         key: Multihash,
     ) -&gt; <span class="prelude-ty">Result</span>&lt;<span class="kw">impl </span>Stream&lt;Item = PeerId&gt;, GetClosestPeersError&gt; {
-        <span class="kw">let </span>permit = <span class="self">self</span>.shared.rate_limiter.acquire_kademlia_permit().<span class="kw">await</span>;
+        <span class="self">self</span>.get_closest_peers_internal(key, <span class="bool-val">true</span>).<span class="kw">await
+    </span>}
+
+    <span class="doccomment">/// Get closest peers by multihash key using Kademlia DHT.
+    </span><span class="kw">async fn </span>get_closest_peers_internal(
+        <span class="kw-2">&amp;</span><span class="self">self</span>,
+        key: Multihash,
+        acquire_permit: bool,
+    ) -&gt; <span class="prelude-ty">Result</span>&lt;<span class="kw">impl </span>Stream&lt;Item = PeerId&gt;, GetClosestPeersError&gt; {
+        <span class="kw">let </span>permit = <span class="kw">if </span>acquire_permit {
+            <span class="prelude-val">Some</span>(<span class="self">self</span>.shared.rate_limiter.acquire_kademlia_permit().<span class="kw">await</span>)
+        } <span class="kw">else </span>{
+            <span class="prelude-val">None
+        </span>};
+
         <span class="macro">trace!</span>(<span class="question-mark">?</span>key, <span class="string">&quot;Starting &#39;GetClosestPeers&#39; request.&quot;</span>);
 
         <span class="kw">let </span>(result_sender, result_receiver) = mpsc::unbounded();
@@ -1192,6 +1228,14 @@
     ) -&gt; <span class="prelude-ty">Result</span>&lt;<span class="kw">impl </span>Stream&lt;Item = PeerId&gt;, GetProvidersError&gt; {
         <span class="self">self</span>.node.get_providers_internal(key, <span class="bool-val">false</span>).<span class="kw">await
     </span>}
+
+    <span class="doccomment">/// Get closest peers by key. Initiate &#39;find_node&#39; Kademlia operation.
+    </span><span class="kw">pub async fn </span>get_closest_peers(
+        <span class="kw-2">&amp;mut </span><span class="self">self</span>,
+        key: Multihash,
+    ) -&gt; <span class="prelude-ty">Result</span>&lt;<span class="kw">impl </span>Stream&lt;Item = PeerId&gt;, GetClosestPeersError&gt; {
+        <span class="self">self</span>.node.get_closest_peers_internal(key, <span class="bool-val">false</span>).<span class="kw">await
+    </span>}
     <span class="doccomment">/// Sends the generic request to the peer and awaits the result.
     </span><span class="kw">pub async fn </span>send_generic_request&lt;Request&gt;(
         <span class="kw-2">&amp;mut </span><span class="self">self</span>,

src/subspace_networking/node_runner.rs.html

@@ -1567,7 +1567,7 @@
     ClosestPeers {
         sender: mpsc::UnboundedSender<PeerId>,
         <span class="comment">// Just holding onto permit while data structure is not dropped
-        </span>_permit: RateLimiterPermit,
+        </span>_permit: <span class="prelude-ty">Option</span>&lt;RateLimiterPermit&gt;,
     },
     Providers {
         sender: mpsc::UnboundedSender&lt;PeerId&gt;,

src/subspace_networking/shared.rs.html

@@ -226,7 +226,7 @@
     GetClosestPeers {
         key: Multihash,
         result_sender: mpsc::UnboundedSender<PeerId>,
-        permit: RateLimiterPermit,
+        permit: <span class="prelude-ty">Option</span>&lt;RateLimiterPermit&gt;,
     },
     GenericRequest {
         peer_id: PeerId,