diff --git a/docs/decisions.md b/docs/decisions.md index 14e5326..0894766 100644 --- a/docs/decisions.md +++ b/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 diff --git a/lib/mysqlex/packet.ex b/lib/mysqlex/packet.ex new file mode 100644 index 0000000..5de2362 --- /dev/null +++ b/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(<>), + 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 + <> = payload + + [ + <<@max_payload::little-24, sequence_id>>, + frame | encode(rest, rem(sequence_id + 1, 256)) + ] + end + + def encode(payload, sequence_id) do + [<>, payload] + end +end diff --git a/mix.exs b/mix.exs index 0bb76ae..e51fc85 100644 --- a/mix.exs +++ b/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} diff --git a/mix.lock b/mix.lock index 59b3044..73468d2 100644 --- a/mix.lock +++ b/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"}, } diff --git a/test/mysqlex/packet_test.exs b/test/mysqlex/packet_test.exs new file mode 100644 index 0000000..76ea092 --- /dev/null +++ b/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