Expire old wants after 5 minutes

.github/workflows/maven.yml

@@ -17,8 +17,6 @@ jobs:
         java-version: 11
     - name: Install and run ipfs
       run: ./install-run-ipfs.sh
-    - name: Pin data in kubo
-      run: ./kubo-pin.sh
     - name: Build and Package
       run: mvn package -Dmaven.test.skip=true
     - name: Run tests

src/main/java/org/peergos/protocol/bitswap/BitswapEngine.java

@@ -23,7 +23,7 @@ public class BitswapEngine {
     private static final Logger LOG = Logging.LOG();
 
     private final Blockstore store;
-    private final ConcurrentHashMap<Want, CompletableFuture<HashedBlock>> localWants = new ConcurrentHashMap<>();
+    private final ConcurrentHashMap<Want, WantResult> localWants = new ConcurrentHashMap<>();
     private final ConcurrentHashMap<Want, Boolean> persistBlocks = new ConcurrentHashMap<>();
     private final ConcurrentHashMap<Want, PeerId> blockHaves = new ConcurrentHashMap<>();
     private final Map<Want, Boolean> deniedWants = Collections.synchronizedMap(new LRUCache<>(10_000));
@@ -47,14 +47,14 @@ public synchronized void addConnection(PeerId peer, Multiaddr addr) {
     }
 
     public CompletableFuture<HashedBlock> getWant(Want w, boolean addToBlockstore) {
-        CompletableFuture<HashedBlock> existing = localWants.get(w);
+        WantResult existing = localWants.get(w);
         if (existing != null)
-            return existing;
-        CompletableFuture<HashedBlock> res = new CompletableFuture<>();
+            return existing.result;
         if (addToBlockstore)
             persistBlocks.put(w, true);
+        WantResult res = new WantResult(System.currentTimeMillis());
         localWants.put(w, res);
-        return res;
+        return res.result;
     }
 
     public boolean hasWants() {
@@ -69,6 +69,15 @@ public Set<PeerId> getConnected() {
         return connected;
     }
 
+    private static final class WantResult {
+        public final CompletableFuture<HashedBlock> result = new CompletableFuture<>();
+        public final long creationTime;
+
+        public WantResult(long creationTime) {
+            this.creationTime = creationTime;
+        }
+    }
+
     private Map<Want, Long> recentSentWants(PeerId peer) {
         Map<Want, Long> recent = recentWantsSent.get(peer);
         if (recent == null) {
@@ -85,8 +94,10 @@ public Set<Want> getWants(Set<PeerId> peers) {
 
             long now = System.currentTimeMillis();
             long minResendWait = 5_000;
-            Set<Want> res = localWants.keySet().stream()
-                    .filter(w -> !recent.containsKey(w) || recent.get(w) < now - minResendWait)
+            Set<Want> res = localWants.entrySet().stream()
+                    .filter(e -> e.getValue().creationTime > now - 5*60*1000)
+                    .map(e -> e.getKey())
+                    .filter(w -> ! recent.containsKey(w) || recent.get(w) < now - minResendWait)
                     .collect(Collectors.toSet());
             res.forEach(w -> recent.put(w, now));
             return res;
@@ -220,13 +231,13 @@ public void receiveMessage(MessageOuterClass.Message msg, Stream source) {
                     byte[] hash = Hash.sha256(data);
                     Cid c = new Cid(version, codec, type, hash);
                     Want w = new Want(c, auth);
-                    CompletableFuture<HashedBlock> waiter = localWants.get(w);
+                    WantResult waiter = localWants.get(w);
                     if (waiter != null) {
                         if (persistBlocks.containsKey(w)) {
                             store.put(data, codec);
                             persistBlocks.remove(w);
                         }
-                        waiter.complete(new HashedBlock(c, data));
+                        waiter.result.complete(new HashedBlock(c, data));
                         localWants.remove(w);
                     } else
                         LOG.info("Received block we don't want: z" + c.toBase58() + " from " + sourcePeerId.bareMultihash());

src/test/java/org/peergos/BitswapMirrorTest.java

@@ -29,6 +29,7 @@ public void mirrorTree() throws IOException {
         IdentifyBuilder.addIdentifyProtocol(node1);
         IPFS kubo = new IPFS("localhost", 5001);
         Multiaddr kuboAddress = Multiaddr.fromString("/ip4/127.0.0.1/tcp/4001/p2p/" + kubo.id().get("ID"));
+        preloadBlocksToKubo(kubo);
 //        Multiaddr kuboAddress = Multiaddr.fromString("/ip4/172.104.157.121/tcp/4001/p2p/QmVdFZgHnEgcedCS2G2ZNiEN59LuVrnRm7z3yXtEBv2XiF");
         node1.getAddressBook().setAddrs(kuboAddress.getPeerId(), 0, kuboAddress).join();
         Kademlia dht1 = builder1.getWanDht().get();
@@ -43,19 +44,26 @@ public void mirrorTree() throws IOException {
             Set<Want> rawToGet = new HashSet<>();
             toGet.add(new Want(Cid.decode("zdpuAwfJrGYtiGFDcSV3rDpaUrqCtQZRxMjdC6Eq9PNqLqTGg")));
             long t1 = System.currentTimeMillis();
+            Map<Cid, byte[]> blocks = new HashMap<>();
             while (true) {
                 Bitswap bitswap1 = builder1.getBitswap().get();
 //                BitswapController bc1 = bitswap1.dial(node1, kuboAddress).getController().join();
                 List<CborObject> cborBlocks = bitswap1
                         .get(new ArrayList<>(toGet), node1, Set.of(kuboAddress.getPeerId()), false).stream()
                         .map(f -> f.join())
-                        .map(h -> h.block)
+                        .map(h -> {
+                            blocks.put(h.hash, h.block);
+                            return h.block;
+                        })
                         .map(CborObject::fromByteArray)
                         .collect(Collectors.toList());
                 List<byte[]> rawBlocks = bitswap1
                         .get(new ArrayList<>(rawToGet), node1, Set.of(kuboAddress.getPeerId()), false).stream()
                         .map(f -> f.join())
-                        .map(h -> h.block)
+                        .map(h -> {
+                            blocks.put(h.hash, h.block);
+                            return h.block;
+                        })
                         .collect(Collectors.toList());
                 toGet.clear();
                 rawToGet.clear();
@@ -72,8 +80,28 @@ public void mirrorTree() throws IOException {
             }
             long t2 = System.currentTimeMillis();
             System.out.println("Mirror took " + (t2-t1) + "mS");
+            Assert.assertTrue(blocks.size() == 6745);
         } finally {
             node1.stop();
         }
     }
+
+    private static void preloadBlocksToKubo(IPFS kubo) throws IOException {
+        DataInputStream din = new DataInputStream(new FileInputStream("blocks.bin"));
+        Map<Cid, byte[]> blocks = new HashMap<>();
+        while (true) {
+            try {
+                int cidSize = din.readInt();
+                byte[] rawCid = din.readNBytes(cidSize);
+                int len = din.readInt();
+                byte[] b = din.readNBytes(len);
+                blocks.put(Cid.cast(rawCid), b);
+            } catch (IOException e) {
+                break;
+            }
+        }
+        for (Map.Entry<Cid, byte[]> e : blocks.entrySet()) {
+            kubo.block.put(e.getValue(), Optional.of(e.getKey().codec == Cid.Codec.Raw ? "raw" : "dag-cbor"));
+        }
+    }
 }