Phase 1: packet framing

docs/decisions.md

@@ -11,7 +11,7 @@ there is evidence), **open** (not decided yet).
 
 ### 1. Keep `db_connection` rather than a plain GenServer
 
-**2026-10-20 · settled**
+**2026-10-20 · settled** · clarified by #15
 
 The driver stays on the `DBConnection` behaviour.
 
@@ -192,6 +192,47 @@ Integration tests against a real MySQL come later and stay separate.
 The placeholder test asserting `Mysqlex.hello() == :world` was deleted rather than
 made to pass; the suite is empty until the first real test.
 
+### 15. Keep `db_connection` now, hand-build a replacement later
+
+**2026-10-28 · settled**
+
+Keep `db_connection` while building the protocol. Once the driver works end to
+end, replacing it with a hand-written connection and pool becomes an option —
+and a deliberate exercise rather than a prerequisite.
+
+**Correcting #1:** that entry said the Ecto goal settled it. More precisely, Ecto
+itself does *not* require DBConnection — `etso`, an ETS-backed adapter, depends on
+`ecto` alone and implements `Ecto.Adapter`, `Ecto.Adapter.Schema` and
+`Ecto.Adapter.Queryable` directly. What requires DBConnection is
+`Ecto.Adapters.SQL`, the helper that supplies most of a SQL adapter. So dropping
+DBConnection costs `Ecto.Adapters.SQL`, not Ecto, and the swap stays available in
+either direction.
+
+Reason for keeping it now is momentum, not architecture. This project was
+abandoned once already. Owning the protocol and a hand-built pool at the same
+time means that when a query returns wrong rows, it is unclear whether the row
+decoder is wrong or a connection was checked back in with bytes still unread in
+its socket buffer — two unproven layers debugged against each other. That is a
+good way to stall again.
+
+What DBConnection is worth, concretely: 4,259 lines, of which `holder.ex` (474)
+exists to guarantee that a connection whose owner process died is never silently
+reused. The protocol-level production problems stay this project's own either
+way — error classification, draining half-read resultsets, `max_allowed_packet`,
+prepared statement invalidation across reconnects, TLS hostname verification,
+sub-binaries holding whole packets alive on the heap.
+
+Consequences:
+
+- The protocol layer must not depend on DBConnection, following MyXQL's split: a
+  standalone client that works on its own, with the behaviour as a thin wrapper
+  over it. That keeps the swap cheap and keeps the protocol testable without any
+  pooling in the picture.
+- Start with a single connection and no concurrency, so that every bug during the
+  protocol phases is a protocol bug by construction. Pooling becomes its own phase
+  afterwards.
+- The stubbed callbacks from #8 stay — they compile, and they document the gap.
+
 ---
 
 ## Open questions

lib/mysqlex/packet.ex

@@ -0,0 +1,67 @@
+defmodule Mysqlex.Packet do
+  @moduledoc """
+  Packet framing.
+
+  Every MySQL packet is a 3-byte little-endian payload length, a 1-byte
+  sequence id, and the payload itself. See
+  `docs/mysql-protocol/page_protocol_basic_packets.md`.
+  """
+
+  @max_payload 0xFFFFFF
+
+  @doc """
+  Decodes one packet from the front of `data`.
+
+  Returns `{:ok, sequence_id, payload, rest}`, or `:more` when `data` does not
+  hold a whole packet yet.
+  """
+  def decode(<<len::little-24, sequence_id, payload::binary-size(len), rest::binary>>),
+    do: {:ok, sequence_id, payload, rest}
+
+  def decode(_data), do: :more
+
+  @doc """
+  Decodes one logical message, rejoining any continuation frames.
+
+  A frame carrying exactly #{@max_payload} bytes means another frame follows, so
+  frames are collected until a shorter one ends the message. Returns
+  `{:ok, sequence_id, message, rest}` with the sequence id of the final frame —
+  the next frame the connection expects is that id plus one — or `:more` while
+  any frame of the message is still outstanding.
+  """
+  def decode_message(data), do: decode_message(data, [])
+
+  defp decode_message(data, frames) do
+    case decode(data) do
+      {:ok, _sequence_id, payload, rest} when byte_size(payload) == @max_payload ->
+        decode_message(rest, [payload | frames])
+
+      {:ok, sequence_id, payload, rest} ->
+        {:ok, sequence_id, IO.iodata_to_binary(Enum.reverse([payload | frames])), rest}
+
+      :more ->
+        :more
+    end
+  end
+
+  @doc """
+  Encodes `payload` as a packet carrying `sequence_id`, returning iodata.
+
+  A payload of #{@max_payload} bytes or more does not fit in the 3-byte length,
+  so it is split across continuation frames with the sequence id incrementing
+  per frame. A payload that is an exact multiple of the maximum is followed by
+  an empty frame, which is how the reader knows the message ended.
+  """
+  def encode(payload, sequence_id) when byte_size(payload) >= @max_payload do
+    <<frame::binary-size(@max_payload), rest::binary>> = payload
+
+    [
+      <<@max_payload::little-24, sequence_id>>,
+      frame | encode(rest, rem(sequence_id + 1, 256))
+    ]
+  end
+
+  def encode(payload, sequence_id) do
+    [<<byte_size(payload)::little-24, sequence_id>>, payload]
+  end
+end

mix.exs

@@ -29,6 +29,7 @@ defmodule Mysqlex.MixProject do
   defp deps do
     [
       {:db_connection, "~> 2.10"},
+      {:stream_data, "~> 1.4", only: [:dev, :test], runtime: false},
       {:credo, "~> 1.7", only: [:dev, :test], runtime: false},
       {:dialyxir, "~> 1.4", only: [:dev, :test], runtime: false},
       {:ex_doc, "~> 0.40", only: :dev, runtime: false}

mix.lock

@@ -12,5 +12,6 @@
   "makeup_elixir": {:hex, :makeup_elixir, "1.0.1", "e928a4f984e795e41e3abd27bfc09f51db16ab8ba1aebdba2b3a575437efafc2", [:mix], [{:makeup, "~> 1.0", [hex: :makeup, repo: "hexpm", optional: false]}, {:nimble_parsec, "~> 1.2.3 or ~> 1.3", [hex: :nimble_parsec, repo: "hexpm", optional: false]}], "hexpm", "7284900d412a3e5cfd97fdaed4f5ed389b8f2b4cb49efc0eb3bd10e2febf9507"},
   "makeup_erlang": {:hex, :makeup_erlang, "1.1.0", "835f7e60792e08824cda445639555d7bf1bbbddb1b60b306e33cb6f6db24dc74", [:mix], [{:makeup, "~> 1.0", [hex: :makeup, repo: "hexpm", optional: false]}], "hexpm", "1cd6780fb1dd1a03979abaed0fe82712b0625118fd5257d3ebbf73f960c73c3c"},
   "nimble_parsec": {:hex, :nimble_parsec, "1.4.2", "8efba0122db06df95bfaa78f791344a89352ba04baedd3849593bfce4d0dc1c6", [:mix], [], "hexpm", "4b21398942dda052b403bbe1da991ccd03a053668d147d53fb8c4e0efe09c973"},
+  "stream_data": {:hex, :stream_data, "1.4.0", "026f929db613aabea6208012ae9b8970d3fd5f88b3bdf26831bc536f98c42036", [:mix], [], "hexpm", "2b0ee3a340dcce1c8cf6302a763ee757d1e01c54d6e16d9069062509d68b1dc9"},
   "telemetry": {:hex, :telemetry, "1.4.2", "a0cb522801dffb1c49fe6e30561badffc7b6d0e180db1300df759faa22062855", [:rebar3], [], "hexpm", "928f6495066506077862c0d1646609eed891a4326bee3126ba54b60af61febb1"},
 }

test/mysqlex/packet_test.exs

@@ -0,0 +1,64 @@
+defmodule Mysqlex.PacketTest do
+  use ExUnit.Case, async: true
+  use ExUnitProperties
+
+  alias Mysqlex.Packet
+
+  test "decodes a packet into its sequence id and payload" do
+    assert {:ok, 0, "hello", ""} = Packet.decode(<<5::little-24, 0, "hello">>)
+  end
+
+  test "returns :more when the payload has not fully arrived" do
+    assert :more = Packet.decode(<<5::little-24, 0, "hel">>)
+  end
+
+  test "returns the bytes following the packet as rest" do
+    second = <<3::little-24, 1, "bye">>
+
+    assert {:ok, 0, "hello", ^second} = Packet.decode(<<5::little-24, 0, "hello">> <> second)
+  end
+
+  test "encodes a payload into a packet" do
+    assert IO.iodata_to_binary(Packet.encode("hello", 0)) == <<5::little-24, 0, "hello">>
+  end
+
+  property "decode/1 inverts encode/2" do
+    check all(payload <- binary(), sequence_id <- integer(0..255)) do
+      encoded = IO.iodata_to_binary(Packet.encode(payload, sequence_id))
+
+      assert {:ok, ^sequence_id, ^payload, ""} = Packet.decode(encoded)
+    end
+  end
+
+  describe "payloads at or beyond the maximum frame size" do
+    @max_payload 0xFFFFFF
+
+    test "encode/2 splits into continuation frames" do
+      head = :binary.copy(<<0>>, @max_payload)
+
+      assert IO.iodata_to_binary(Packet.encode(head <> "tail", 0)) ==
+               <<@max_payload::little-24, 0>> <> head <> <<4::little-24, 1, "tail">>
+    end
+
+    test "encode/2 appends an empty frame when the payload is exactly the maximum" do
+      payload = :binary.copy(<<0>>, @max_payload)
+
+      assert IO.iodata_to_binary(Packet.encode(payload, 0)) ==
+               <<@max_payload::little-24, 0>> <> payload <> <<0::little-24, 1>>
+    end
+
+    test "decode_message/1 reassembles continuation frames" do
+      head = :binary.copy(<<0>>, @max_payload)
+      data = IO.iodata_to_binary(Packet.encode(head <> "tail", 0))
+
+      assert {:ok, 1, message, ""} = Packet.decode_message(data)
+      assert message == head <> "tail"
+    end
+
+    test "decode_message/1 returns :more until the final frame arrives" do
+      head = :binary.copy(<<0>>, @max_payload)
+
+      assert :more = Packet.decode_message(<<@max_payload::little-24, 0>> <> head)
+    end
+  end
+end