fix docker build

package.json

@@ -38,5 +38,13 @@
   "license": "MIT",
   "workspaces": [
     "packages/*"
-  ]
+  ],
+  "resolutions": {
+    "@libp2p/interface-connection-manager": "1.3.8"
+  },
+  "overrides": {
+    "libp2p": {
+      "@libp2p/interface-connection-manager": "1.3.8"
+    }
+  }
 }

packages/cli/package.json

@@ -43,7 +43,7 @@
     "@chainsafe/discv5": "3.0.0",
     "@multiformats/multiaddr": "12.1.3",
     "@libp2p/interface-connection": "3.0.2",
-    "@libp2p/interface-connection-manager": "1.3.0",
+    "@libp2p/interface-connection-manager": "1.3.8",
     "@libp2p/interface-peer-id": "2.0.1",
     "@libp2p/peer-id-factory": "2.0.1",
     "@libp2p/prometheus-metrics": "1.1.3"

packages/node/package.json

@@ -31,20 +31,21 @@
     "check-readme": "typescript-docs-verifier"
   },
   "dependencies": {
+    "@chainsafe/fast-crc32c": "4.1.1",
     "@chainsafe/as-chacha20poly1305": "0.1.0",
     "@chainsafe/as-sha256": "0.3.1",
     "@chainsafe/bls": "7.1.1",
     "@chainsafe/discv5": "3.0.0",
     "@chainsafe/libp2p-gossipsub": "6.2.0",
     "@chainsafe/libp2p-noise": "11.0.4",
     "@chainsafe/persistent-merkle-tree": "0.5.0",
-    "@chainsafe/snappy-stream": "5.1.2",
+    "snappy": "7.2.2",
     "@chainsafe/ssz": "0.10.2",
     "@chainsafe/threads": "1.10.0",
     "@ethersproject/abi": "5.7.0",
     "@libp2p/bootstrap": "6.0.3",
     "@libp2p/interface-connection": "3.0.2",
-    "@libp2p/interface-connection-manager": "1.3.0",
+    "@libp2p/interface-connection-manager": "1.3.8",
     "@libp2p/interface-peer-id": "2.0.1",
     "@libp2p/interface-pubsub": "3.0.0",
     "@libp2p/interface-registrar": "2.0.8",
@@ -80,5 +81,10 @@
     "@types/abstract-leveldown": "7.2.1",
     "@types/rocksdb": "3.0.1",
     "@types/varint": "6.0.1"
+  },
+  "overrides": {
+    "libp2p": {
+      "@libp2p/interface-connection-manager": "1.3.8"
+    }
   }
 }

packages/node/src/network/gossip/handler.ts

@@ -32,7 +32,7 @@ export class BundlerGossipsub extends GossipSub {
       {
         peerId: modules.libp2p.peerId,
         peerStore: modules.libp2p.peerStore,
-        registrar: modules.libp2p.registrar,
+        registrar: modules.libp2p.registrar as any,
         connectionManager: modules.libp2p.connectionManager as any,
       },
       {}

packages/node/src/reqresp/encodingStrategies/sszSnappy/encode.ts

@@ -1,12 +1,11 @@
-import varint from "varint";
-import { source } from "stream-to-it";
-import snappy from "@chainsafe/snappy-stream";
+import { encode as varintEncode } from "uint8-varint";
 import {
   EncodedPayload,
   EncodedPayloadType,
   TypeSerializer,
 } from "../../types";
 import { SszSnappyError, SszSnappyErrorCode } from "./errors";
+import { encodeSnappy } from "./snappyFrames/compress";
 
 /**
  * ssz_snappy encoding strategy writer.
@@ -29,23 +28,11 @@ export async function* writeSszSnappyPayload<T>(
  */
 export async function* encodeSszSnappy(bytes: Buffer): AsyncGenerator<Buffer> {
   // MUST encode the length of the raw SSZ bytes, encoded as an unsigned protobuf varint
-  yield Buffer.from(varint.encode(bytes.length));
+  yield Buffer.from(varintEncode(bytes.length));
 
   // By first computing and writing the SSZ byte length, the SSZ encoder can then directly
   // write the chunk contents to the stream. Snappy writer compresses frame by frame
-
-  /**
-   * Use sync version (default) for compress as it is almost 2x faster than async
-   * one and most payloads are "1 chunk" and 100kb payloads (which would mostly be
-   * big bellatrix blocks with transactions) are just 2 chunks
-   *
-   * To use async version (for e.g. on big payloads) instantiate the stream with
-   * `createCompressStream({asyncCompress: true})`
-   */
-  const stream = snappy.createCompressStream();
-  stream.write(bytes);
-  stream.end();
-  yield* source<Buffer>(stream);
+  yield* encodeSnappy(bytes);
 }
 
 /**

packages/node/src/reqresp/encodingStrategies/sszSnappy/snappyFrames/common.ts

@@ -0,0 +1,11 @@
+export enum ChunkType {
+  IDENTIFIER = 0xff,
+  COMPRESSED = 0x00,
+  UNCOMPRESSED = 0x01,
+  PADDING = 0xfe,
+}
+
+export const IDENTIFIER = Buffer.from([0x73, 0x4e, 0x61, 0x50, 0x70, 0x59]);
+export const IDENTIFIER_FRAME = Buffer.from([
+  0xff, 0x06, 0x00, 0x00, 0x73, 0x4e, 0x61, 0x50, 0x70, 0x59,
+]);

packages/node/src/reqresp/encodingStrategies/sszSnappy/snappyFrames/compress.ts

@@ -0,0 +1,57 @@
+import snappy from "snappy";
+import crc32c from "@chainsafe/fast-crc32c";
+import { ChunkType, IDENTIFIER_FRAME } from "./common.js";
+
+// The logic in this file is largely copied (in simplified form) from https://github.com/ChainSafe/node-snappy-stream/
+
+/**
+ * As per the snappy framing format for streams, the size of any uncompressed chunk can be
+ * no longer than 65536 bytes.
+ *
+ * From: https://github.com/google/snappy/blob/main/framing_format.txt#L90:L92
+ */
+const UNCOMPRESSED_CHUNK_SIZE = 65536;
+
+function checksum(value: Buffer): Buffer {
+  const x = crc32c.calculate(value);
+  const result = Buffer.allocUnsafe?.(4) ?? Buffer.alloc(4);
+
+  // As defined in section 3 of https://github.com/google/snappy/blob/master/framing_format.txt
+  // And other implementations for reference:
+  // Go: https://github.com/golang/snappy/blob/2e65f85255dbc3072edf28d6b5b8efc472979f5a/snappy.go#L97
+  // Python: https://github.com/andrix/python-snappy/blob/602e9c10d743f71bef0bac5e4c4dffa17340d7b3/snappy/snappy.py#L70
+  // Mask the right hand to (32 - 17) = 15 bits -> 0x7fff, to keep correct 32 bit values.
+  // Shift the left hand with >>> for correct 32 bit intermediate result.
+  // Then final >>> 0 for 32 bits output
+  result.writeUInt32LE(
+    (((x >>> 15) | ((x & 0x7fff) << 17)) + 0xa282ead8) >>> 0,
+    0
+  );
+
+  return result;
+}
+
+export async function* encodeSnappy(bytes: Buffer): AsyncGenerator<Buffer> {
+  yield IDENTIFIER_FRAME;
+
+  for (let i = 0; i < bytes.length; i += UNCOMPRESSED_CHUNK_SIZE) {
+    const chunk = bytes.subarray(i, i + UNCOMPRESSED_CHUNK_SIZE);
+    const compressed = snappy.compressSync(chunk);
+    if (compressed.length < chunk.length) {
+      const size = compressed.length + 4;
+      yield Buffer.concat([
+        Buffer.from([ChunkType.COMPRESSED, size, size >> 8, size >> 16]),
+        checksum(chunk),
+        compressed,
+      ]);
+    } else {
+      const size = chunk.length + 4;
+      yield Buffer.concat([
+        //
+        Buffer.from([ChunkType.UNCOMPRESSED, size, size >> 8, size >> 16]),
+        checksum(chunk),
+        chunk,
+      ]);
+    }
+  }
+}