feat: snappy framing format (decompression) (#710)

Co-authored-by: Akash SM <akashsm@ce.iitr.ac.in>

Author: MegaRedHand

lib/snappy_ex.ex

@@ -0,0 +1,126 @@
+defmodule SnappyEx do
+  @moduledoc """
+    Encoder/decoder implementation for the [Snappy framing format](https://github.com/google/snappy/blob/main/framing_format.txt)
+  """
+  import Bitwise
+
+  @bit_mask_32 2 ** 32 - 1
+  @chunk_size_limit 65_540
+
+  @stream_identifier "sNaPpY"
+
+  @id_compressed_data 0x00
+  @id_uncompressed_data 0x01
+  @id_padding 0xFE
+  @id_stream_identifier 0xFF
+
+  @ids_payload_chunks [@id_compressed_data, @id_uncompressed_data]
+  @ids_reserved_unskippable_chunks 0x02..0x7F
+  @ids_reserved_skippable_chunks 0x80..0xFD
+
+  ##########################
+  ### Public API
+  ##########################
+
+  @doc """
+  Compresses the given data.
+  Returns the compressed data.
+
+  ## Examples
+
+      iex> SnappyEx.compress("")
+      <<0xFF, 6::little-size(24)>> <> "sNaPpY"
+  """
+  @spec compress(binary()) :: binary()
+  def compress(data) when is_binary(data) do
+    # TODO: implement
+    <<@id_stream_identifier, 6::little-size(24)>> <> @stream_identifier
+  end
+
+  @doc """
+  Uncompresses a given stream.
+  Returns a result tuple with the uncompressed data or an error message.
+
+  ## Examples
+
+      iex> SnappyEx.decompress(<<0xFF, 6::little-size(24)>> <> "sNaPpY")
+      {:ok, ""}
+  """
+  @spec decompress(nonempty_binary()) :: {:ok, binary()} | {:error, String.t()}
+  def decompress(<<@id_stream_identifier>> <> _ = chunks), do: decompress_frames(chunks, <<>>)
+
+  @spec decompress(<<>>) :: {:error, String.t()}
+  def decompress(chunks) when is_binary(chunks), do: {:error, "no stream identifier at beginning"}
+
+  @spec compute_checksum(binary()) :: Types.uint32()
+  def compute_checksum(data) when is_binary(data) do
+    checksum = Crc32c.calc!(data)
+
+    # the crc32c checksum of the uncompressed data is masked before inserted into the
+    # frame using masked_checksum = ((checksum >> 15) | (checksum << 17)) + 0xa282ead8
+    (checksum >>> 15 ||| checksum <<< 17) + 0xA282EAD8 &&& @bit_mask_32
+  end
+
+  ##########################
+  ### Private Functions
+  ##########################
+
+  defp decompress_frames("", acc), do: {:ok, acc}
+
+  defp decompress_frames(chunks, acc) do
+    with {:ok, {id, data, remaining_chunks}} <- process_chunk_metadata(chunks),
+         {:ok, new_acc} <- parse_chunk(acc, id, data) do
+      decompress_frames(remaining_chunks, new_acc)
+    end
+  end
+
+  # chunk layout: 1-byte chunk_id, 3-bytes chunk_size, remaining chunks if any data present.
+  defp process_chunk_metadata(chunks) when byte_size(chunks) < 4,
+    do: {:error, "header too small"}
+
+  defp process_chunk_metadata(<<_id::size(8), size::little-size(24), rest::binary>>)
+       when byte_size(rest) < size,
+       do: {:error, "missing data in chunk. expected: #{byte_size(rest)}. got: #{size}"}
+
+  defp process_chunk_metadata(<<id::size(8), size::little-size(24), rest::binary>>) do
+    <<chunk::binary-size(size), remaining_chunks::binary>> = rest
+    {:ok, {id, chunk, remaining_chunks}}
+  end
+
+  # Stream identifier
+  # NOTE: it can appear more than once, and must be validated each time
+  defp parse_chunk(acc, @id_stream_identifier, @stream_identifier), do: {:ok, acc}
+  defp parse_chunk(_, @id_stream_identifier, _), do: {:error, "invalid stream identifier"}
+
+  # Data-carrying chunks (compressed or uncompressed)
+  defp parse_chunk(_acc, id, data)
+       when id in @ids_payload_chunks and byte_size(data) > @chunk_size_limit,
+       do: {:error, "chunk is bigger than limit"}
+
+  defp parse_chunk(acc, id, data) when id in @ids_payload_chunks do
+    <<checksum::little-size(32), compressed_data::binary>> = data
+
+    with {:ok, uncompressed_data} <- decompress_payload(id, compressed_data),
+         :ok <- verify_checksum(uncompressed_data, checksum) do
+      {:ok, <<acc::binary, uncompressed_data::binary>>}
+    end
+  end
+
+  # Skippable chunks (padding or reserved)
+  defp parse_chunk(acc, id, _data)
+       when id == @id_padding or id in @ids_reserved_skippable_chunks,
+       do: {:ok, acc}
+
+  # Reserved unskippable chunks
+  defp parse_chunk(_acc, id, _data) when id in @ids_reserved_unskippable_chunks,
+    do: {:error, "unskippable chunk of type: #{id}"}
+
+  defp decompress_payload(@id_compressed_data, data), do: :snappyer.decompress(data)
+  defp decompress_payload(@id_uncompressed_data, data), do: {:ok, data}
+
+  defp verify_checksum(data, checksum) do
+    if checksum == compute_checksum(data),
+      do: :ok,
+      else: {:error, "invalid checksum"}
+  end
+end

lib/types/mod.ex

@@ -7,6 +7,7 @@ defmodule Types do
   ## Integer types
   @type uint8 :: 0..unquote(2 ** 8 - 1)
   @type uint16 :: 0..unquote(2 ** 16 - 1)
+  @type uint32 :: 0..unquote(2 ** 32 - 1)
   @type uint64 :: 0..unquote(2 ** 64 - 1)
   @type uint256 :: 0..unquote(2 ** 256 - 1)
 

mix.exs

@@ -54,11 +54,12 @@ defmodule LambdaEthereumConsensus.MixProject do
       {:ex2ms, "~> 1.6", runtime: false},
       {:eflambe, "~> 0.3.1"},
       {:patch, "~> 0.13.0", only: [:test]},
-      {:stream_data, "~> 0.5", only: [:test]},
+      {:stream_data, "~> 0.6", only: [:test]},
       {:benchee, "~> 1.2", only: [:dev]},
       {:dialyxir, "~> 1.1", only: [:dev, :test], runtime: false},
       {:credo, "~> 1.7", only: [:dev, :test], runtime: false},
-      {:open_api_spex, "~> 3.18"}
+      {:open_api_spex, "~> 3.18"},
+      {:crc32c, git: "https://github.com/lambdaclass/crc32c", branch: "bump-rustler-to-29"}
     ]
   end
 

mix.lock

@@ -9,6 +9,7 @@
   "cowboy": {:hex, :cowboy, "2.10.0", "ff9ffeff91dae4ae270dd975642997afe2a1179d94b1887863e43f681a203e26", [:make, :rebar3], [{:cowlib, "2.12.1", [hex: :cowlib, repo: "hexpm", optional: false]}, {:ranch, "1.8.0", [hex: :ranch, repo: "hexpm", optional: false]}], "hexpm", "3afdccb7183cc6f143cb14d3cf51fa00e53db9ec80cdcd525482f5e99bc41d6b"},
   "cowboy_telemetry": {:hex, :cowboy_telemetry, "0.4.0", "f239f68b588efa7707abce16a84d0d2acf3a0f50571f8bb7f56a15865aae820c", [:rebar3], [{:cowboy, "~> 2.7", [hex: :cowboy, repo: "hexpm", optional: false]}, {:telemetry, "~> 1.0", [hex: :telemetry, repo: "hexpm", optional: false]}], "hexpm", "7d98bac1ee4565d31b62d59f8823dfd8356a169e7fcbb83831b8a5397404c9de"},
   "cowlib": {:hex, :cowlib, "2.12.1", "a9fa9a625f1d2025fe6b462cb865881329b5caff8f1854d1cbc9f9533f00e1e1", [:make, :rebar3], [], "hexpm", "163b73f6367a7341b33c794c4e88e7dbfe6498ac42dcd69ef44c5bc5507c8db0"},
+  "crc32c": {:git, "https://github.com/lambdaclass/crc32c", "e4207478aa5b01c7e0326d89a399a76dc8215daf", [branch: "bump-rustler-to-29"]},
   "credo": {:hex, :credo, "1.7.3", "05bb11eaf2f2b8db370ecaa6a6bda2ec49b2acd5e0418bc106b73b07128c0436", [:mix], [{:bunt, "~> 0.2.1 or ~> 1.0", [hex: :bunt, repo: "hexpm", optional: false]}, {:file_system, "~> 0.2 or ~> 1.0", [hex: :file_system, repo: "hexpm", optional: false]}, {:jason, "~> 1.0", [hex: :jason, repo: "hexpm", optional: false]}], "hexpm", "35ea675a094c934c22fb1dca3696f3c31f2728ae6ef5a53b5d648c11180a4535"},
   "deep_merge": {:hex, :deep_merge, "1.0.0", "b4aa1a0d1acac393bdf38b2291af38cb1d4a52806cf7a4906f718e1feb5ee961", [:mix], [], "hexpm", "ce708e5f094b9cd4e8f2be4f00d2f4250c4095be93f8cd6d018c753894885430"},
   "dialyxir": {:hex, :dialyxir, "1.4.3", "edd0124f358f0b9e95bfe53a9fcf806d615d8f838e2202a9f430d59566b6b53b", [:mix], [{:erlex, ">= 0.2.6", [hex: :erlex, repo: "hexpm", optional: false]}], "hexpm", "bf2cfb75cd5c5006bec30141b131663299c661a864ec7fbbc72dfa557487a986"},

test/unit/snappy_test.exs

@@ -3,7 +3,7 @@ defmodule Unit.SnappyTest do
   use ExUnitProperties
   doctest Snappy
 
-  def assert_snappy_decompress(compressed, uncompressed) do
+  defp assert_snappy_decompress(compressed, uncompressed) do
     {:ok, ^uncompressed} =
       compressed
       |> Base.decode16!()

test/unit/snappyex_test.exs

@@ -0,0 +1,120 @@
+defmodule Unit.SnappyExTest do
+  use ExUnit.Case
+  use ExUnitProperties
+  doctest SnappyEx
+
+  @empty_stream <<0xFF, 6::little-size(24)>> <> "sNaPpY"
+
+  defp assert_snappy_decompress(compressed, uncompressed) do
+    assert compressed |> Base.decode16!() |> SnappyEx.decompress() ==
+             {:ok, Base.decode16!(uncompressed)}
+  end
+
+  test "empty stream w/o stream identifier" do
+    assert {:error, _} = SnappyEx.decompress("")
+  end
+
+  test "empty stream w/ stream identifier" do
+    assert {:ok, ""} = SnappyEx.decompress(@empty_stream)
+  end
+
+  test "uncompressed stream" do
+    data = "some uncompressed data"
+    checksum = <<SnappyEx.compute_checksum(data)::little-size(32)>>
+
+    size = byte_size(data) + byte_size(checksum)
+
+    header = <<0x01, size::little-size(24)>>
+    stream = Enum.join([@empty_stream, header, checksum, data])
+
+    assert SnappyEx.decompress(stream) == {:ok, data}
+  end
+
+  test "compressed stream" do
+    data = "some compressed data"
+    checksum = <<SnappyEx.compute_checksum(data)::little-size(32)>>
+
+    {:ok, compressed_data} = :snappyer.compress(data)
+    size = byte_size(compressed_data) + byte_size(checksum)
+
+    header = <<0x00, size::little-size(24)>>
+    stream = Enum.join([@empty_stream, header, checksum, compressed_data])
+
+    assert SnappyEx.decompress(stream) == {:ok, data}
+  end
+
+  test "decompress GetMetadata response uncompressed 0" do
+    assert_snappy_decompress(
+      "FF060000734E6150705901150000F1D17CFF0008000000000000FFFFFFFFFFFFFFFF0F",
+      "0008000000000000FFFFFFFFFFFFFFFF0F"
+    )
+  end
+
+  test "decompress GetMetadata response uncompressed 1" do
+    assert_snappy_decompress(
+      "FF060000734E6150705901150000CD11E7D53A03000000000000FFFFFFFFFFFFFFFF0F",
+      "3A03000000000000FFFFFFFFFFFFFFFF0F"
+    )
+  end
+
+  test "decompress GetMetadata response compressed" do
+    assert_snappy_decompress(
+      "FF060000734E61507059000A0000B3A056EA1100003E0100",
+      "0000000000000000000000000000000000"
+    )
+  end
+
+  test "decompress Ping response 0" do
+    assert_snappy_decompress(
+      "FF060000734E61507059010C0000B18525A04300000000000000",
+      "4300000000000000"
+    )
+  end
+
+  test "decompress Ping response 1" do
+    assert_snappy_decompress(
+      "FF060000734E61507059010C00000175DE410100000000000000",
+      "0100000000000000"
+    )
+  end
+
+  test "decompress Ping response 2" do
+    assert_snappy_decompress(
+      "FF060000734E61507059010C0000EAB2043E0500000000000000",
+      "0500000000000000"
+    )
+  end
+
+  test "decompress Ping response 3" do
+    assert_snappy_decompress(
+      "FF060000734E61507059010C0000290398070000000000000000",
+      "0000000000000000"
+    )
+  end
+
+  test "decompress error response" do
+    assert_snappy_decompress(
+      "FF060000734E6150705900220000EF99F84B1C6C4661696C656420746F20756E636F6D7072657373206D657373616765",
+      Base.encode16("Failed to uncompress message")
+    )
+  end
+
+  property "SnappyEx == Snappy: random stream" do
+    check all(stream <- binary(min_length: 1)) do
+      expected = Snappy.decompress(stream)
+
+      case SnappyEx.decompress(stream) do
+        {:ok, result} -> assert expected == {:ok, result}
+        {:error, reason} -> assert {:error, _} = expected, reason
+      end
+    end
+  end
+
+  property "SnappyEx == Snappy: random valid stream" do
+    check all(payload <- binary(min_length: 1)) do
+      {:ok, stream} = Snappy.compress(payload)
+
+      assert SnappyEx.decompress(stream) == {:ok, payload}
+    end
+  end
+end