diff --git a/.github/workflows/windows.yml b/.github/workflows/windows.yml new file mode 100644 index 0000000..072f703 --- /dev/null +++ b/.github/workflows/windows.yml @@ -0,0 +1,119 @@ +name: Windows + +on: + push: + branches: [main] + paths-ignore: + - '**.md' + - 'guides/**' + - 'LICENSE' + - '.gitignore' + pull_request: + branches: [main] + paths-ignore: + - '**.md' + - 'guides/**' + - 'LICENSE' + - '.gitignore' + +concurrency: + group: ${{ github.workflow }}-${{ github.head_ref || github.ref }} + cancel-in-progress: true + +# The socket transport waits for bytes differently here: Windows completes a +# read and hands the bytes over in the message, where POSIX says the socket is +# readable and leaves them in it. Nothing on a Linux runner exercises that path, +# so it runs here or it is not tested at all. +jobs: + windows: + runs-on: windows-latest + timeout-minutes: 20 + env: + MINATO_PG_HOST: 127.0.0.1 + MINATO_PG_PORT: '5432' + MINATO_PG_USER: minato + MINATO_PG_PASSWORD: minato + MINATO_PG_DATABASE: minato_test + MINATO_TLS_HOST: localhost + MINATO_TLS_PORT: '55433' + steps: + - uses: actions/checkout@v4 + + - uses: erlef/setup-beam@v1 + with: + otp-version: '28' + rebar3-version: '3' + + # The compose file gives Linux two servers, one plain and one with TLS on, + # and the suites are written against that. There is no compose here, so + # the runner's own server is the plain one and a second cluster is made + # for TLS rather than turning TLS on for both - a test that a server + # declining TLS is refused needs a server that declines. + - name: Certificates for the TLS server + shell: pwsh + run: | + New-Item -ItemType Directory -Force -Path test/certs | Out-Null + openssl req -x509 -newkey rsa:2048 -sha256 -days 3650 -nodes ` + -keyout ca.key -out test/certs/ca.crt -subj "/CN=minato test CA" ` + -addext "basicConstraints=critical,CA:TRUE" + openssl req -newkey rsa:2048 -nodes -keyout server.key -out server.csr ` + -subj "/CN=localhost" + Set-Content -Path server.ext -Value @( + 'subjectAltName=DNS:localhost', + 'basicConstraints=CA:FALSE', + 'extendedKeyUsage=serverAuth' + ) + openssl x509 -req -in server.csr -CA test/certs/ca.crt -CAkey ca.key ` + -CAcreateserial -out test/certs/server.crt -days 3650 -sha256 -extfile server.ext + + - name: Start the plain PostgreSQL the runner already has + shell: pwsh + run: | + $service = Get-Service -Name 'postgresql*' | Select-Object -First 1 + if (-not $service) { throw 'no PostgreSQL service on this runner' } + Set-Service -Name $service.Name -StartupType Manual + Start-Service -Name $service.Name + $env:PGPASSWORD = 'root' + & "$env:PGBIN\psql" -U postgres -c "CREATE USER minato WITH PASSWORD 'minato' SUPERUSER" + & "$env:PGBIN\psql" -U postgres -c "CREATE DATABASE minato_test OWNER minato" + & "$env:PGBIN\psql" -U postgres -c "SHOW ssl" + + - name: Start a second PostgreSQL with TLS on + shell: pwsh + run: | + $data = "$env:RUNNER_TEMP\minato-tls" + Set-Content -Path "$env:RUNNER_TEMP\pw.txt" -Value 'root' -NoNewline + & "$env:PGBIN\initdb" -D $data -U postgres --auth=scram-sha-256 ` + --pwfile="$env:RUNNER_TEMP\pw.txt" -E UTF8 + Copy-Item test/certs/server.crt "$data\server.crt" + Copy-Item server.key "$data\server.key" + Add-Content -Path "$data\postgresql.conf" -Value @( + 'ssl = on', + "ssl_cert_file = 'server.crt'", + "ssl_key_file = 'server.key'", + 'port = 55433', + "timezone = 'UTC'" + ) + & "$env:PGBIN\pg_ctl" -D $data -l "$env:RUNNER_TEMP\tls.log" start + Start-Sleep -Seconds 5 + $env:PGPASSWORD = 'root' + & "$env:PGBIN\psql" -U postgres -h localhost -p 55433 ` + -c "CREATE USER minato WITH PASSWORD 'minato' SUPERUSER" + & "$env:PGBIN\psql" -U postgres -h localhost -p 55433 ` + -c "CREATE DATABASE minato_test OWNER minato" + & "$env:PGBIN\psql" -U postgres -h localhost -p 55433 -c "SHOW ssl" + + - name: Compile + run: rebar3 compile + + - name: EUnit + run: rebar3 eunit + + - name: Common Test + timeout-minutes: 10 + run: rebar3 ct + + - name: What the TLS server logged, if anything went wrong + if: failure() + shell: pwsh + run: Get-Content "$env:RUNNER_TEMP\tls.log" -Tail 50 diff --git a/.gitignore b/.gitignore index bb6995a..eba677a 100644 --- a/.gitignore +++ b/.gitignore @@ -6,5 +6,6 @@ rebar3.crashdump doc/ bom.xml bench/js/node_modules/ +bench/py/.venv/ test/certs/*.crt test/certs/*.key diff --git a/bench/README.md b/bench/README.md index 065571a..1b696a6 100644 --- a/bench/README.md +++ b/bench/README.md @@ -9,8 +9,9 @@ bench/run.sh 2000 ``` The Erlang clients always run. The JavaScript ones need `cd bench/js && npm -install`, and the Python ones need `asyncpg` and `psycopg[binary]` importable; -the runner says which it skipped rather than leaving them silently out. +install` and the Python ones need `bench/py/install.sh`, which puts `asyncpg` +and `psycopg[binary]` in `bench/py/.venv`; the runner says which it skipped +rather than leaving them silently out. `pgo` here is `erleans/pgo` from Hex, not the fork this repository pins as a test transport. @@ -47,39 +48,74 @@ clients in the same run, and anything under about ten percent is noise. ## A run 2000 iterations, OTP 29, PostgreSQL 17 in Docker on the same machine, times in -microseconds per operation. - -| workload | minato | epgsql | pgo | node-postgres | postgres.js | -| --- | --- | --- | --- | --- | --- | -| simple | **313** | 337 | - | 354 | 384 | -| unnamed | 612 | 678 | - | **401** | 711 | -| cached | 342 | **307** | 436 | 374 | 384 | -| rows_1000 | 834 | **589** | 3842 | 905 | 745 | -| rows_5000 | 2248 | **1806** | 16076 | 2257 | 2197 | -| wide_row | 381 | **380** | 646 | 457 | 444 | -| insert | 378 | 350 | 471 | **310** | 346 | -| concurrent | 913 | 1501 | 1226 | 877 | **828** | +microseconds per operation, mean. + +| workload | minato | epgsql | pgo | node-postgres | postgres.js | asyncpg | psycopg3 | +| --- | --- | --- | --- | --- | --- | --- | --- | +| simple | **322** | 347 | - | 366 | 351 | 380 | 379 | +| unnamed | 629 | 666 | - | 414 | 750 | **399** | 432 | +| cached | 360 | **339** | 467 | 354 | 377 | 392 | 389 | +| rows_1000 | 639 | **576** | 3662 | 728 | 842 | 584 | 580 | +| rows_5000 | **1429** | 1861 | 15428 | 1868 | 1994 | 1584 | 1581 | +| wide_row | 373 | **371** | 663 | 384 | 439 | 389 | 403 | +| insert | 413 | 369 | 489 | **354** | 386 | 405 | 379 | +| concurrent | 1037 | 1451 | 976 | **752** | 847 | 1751 | 1167 | What the numbers say: - **Bulk framing is worth what it claimed.** `pgo` reads a header and then a - payload per message, and at 5000 rows it costs 16 ms against everybody else's - 2 ms. That is the one enormous difference in the table, and it is the + payload per message, and at 5000 rows it costs 15 ms against everybody else's + 1.5 to 2 ms. That is the one enormous difference in the table, and it is the difference this client was written around. -- **epgsql is fast, and faster than minato on row-heavy reads.** 589 us against - 834 at a thousand rows. epgsql has had fifteen years of tuning and it shows; - the gap is worth chasing and there is now a number to chase it against. +- **epgsql is still ahead at a thousand rows and behind at five thousand.** + minato does about thirty percent fewer reductions than epgsql at both sizes, + so whatever is left at a thousand is not decoding work. - **Describing before binding costs a round trip**, which is what the `unnamed` - row is: 612 us against 342 for the same query cached. That is the price of not + row is: 629 us against 360 for the same query cached. That is the price of not guessing parameter types, and `minato:query/3` pays it once per connection - rather than once per call. -- **The Node clients are not slower than the BEAM ones.** Anybody expecting a - scripting language to lose here should look at `insert` and `concurrent` - again. Most of a query's cost is waiting for the server, and every client here - is good at waiting. + rather than once per call. asyncpg and node-postgres are quicker on that row + because they infer the types instead of asking. +- **The Node and Python clients are not slower than the BEAM ones.** Anybody + expecting a scripting language to lose here should look at `insert` and + `concurrent` again. Most of a query's cost is waiting for the server, and + every client here is good at waiting. - **`concurrent` is where a pool shows up.** epgsql has none of its own, so its - eight callers get eight connections and it still comes last; minato and the - Node pools are within noise of each other. + eight callers get eight connections and it is still second from last. + +## Where the work goes + +Wall clock against a database on the same machine is mostly the round trip, and +the differences between clients hide in the noise of whatever else the machine +is doing. `minato_profile` measures the work instead: + +``` +rebar3 as bench compile +erl -noshell -pa _build/bench/lib/*/ebin _build/bench/lib/minato/bench \ + -eval 'minato_profile:cost(2000), init:stop().' +``` + +Reductions are not microseconds, but they do not move with the machine, the +server or the load, so a five percent change is a change rather than a +run-to-run wobble. Per operation, same run as the table above: + +| workload | minato | epgsql | pgo | +| --- | --- | --- | --- | +| simple | 439 | **289** | - | +| unnamed | 766 | **677** | - | +| cached | 434 | **348** | 1205 | +| rows_1000 | **20303** | 29497 | 152118 | +| rows_5000 | **101072** | 151245 | 757563 | +| wide_row | **1335** | 1657 | 2744 | +| insert | 426 | **391** | 1001 | + +minato does less work per row than either and more work per statement. The +per-statement difference is most of what it charges for the things epgsql does +not do at all: a telemetry span around every query, a deadline that cancels the +statement on the server rather than abandoning the connection, and a lookup in +the per-connection statement cache. + +`minato_profile:eprof(Workload)` runs one workload under `eprof`, for when the +question is which function rather than how much. ## Honesty about this benchmark diff --git a/bench/minato_bench.erl b/bench/minato_bench.erl index 4073036..f8509bb 100644 --- a/bench/minato_bench.erl +++ b/bench/minato_bench.erl @@ -60,6 +60,8 @@ clients on the same run, and a difference under about ten percent is noise. """. -export([main/1, run/1, iterations/1, json/1]). +%% For minato_profile, which measures the same workloads a different way. +-export([setup/1, prepare_schema/2, parsed/2, runner/3, teardown/2]). -define(DEFAULT_ITERATIONS, 2000). -define(DEFAULT_WARMUP, 200). diff --git a/bench/minato_profile.erl b/bench/minato_profile.erl new file mode 100644 index 0000000..f2bcc5b --- /dev/null +++ b/bench/minato_profile.erl @@ -0,0 +1,150 @@ +-module(minato_profile). +-moduledoc """ +Where a query's time goes, rather than how long it takes. + +`bench/run.sh` measures wall clock, which against a database on the same machine +is mostly the round trip: every client waits the same and the differences hide in +the noise. This module measures the work instead. + +- `cost/0` runs each workload on each Erlang client and reports total system + reductions and scheduler run time per operation. Reductions are not + microseconds, but they do not move with the machine, the server or the load, + so a change of five percent is a change rather than a run-to-run wobble. +- `eprof/1` runs one minato workload under `eprof` and prints where the calls + are. + +``` +docker compose -f test/docker-compose.yml up -d +rebar3 as bench compile +erl -noshell -pa _build/bench/lib/*/ebin _build/bench/lib/minato/bench \\ + -eval 'minato_profile:cost(), init:stop().' +``` +""". + +-export([cost/0, cost/1, eprof/1, eprof/2]). + +-define(DEFAULT_ITERATIONS, 2000). +-define(DEFAULT_EPROF_ITERATIONS, 200). + +-doc "Reductions and run time per operation, every workload on every client.". +-spec cost() -> ok. +cost() -> + cost(iterations()). + +-doc "`cost/0` over a given number of iterations.". +-spec cost(pos_integer()) -> ok. +cost(Iterations) -> + _ = erlang:system_flag(scheduler_wall_time, true), + Clients = [minato, pgo, epgsql], + Ready = [{Client, setup(Client)} || Client <- Clients], + Rows = [ + {Workload, Client, measure(Client, Workload, Setup, Iterations)} + || Workload <- workloads(), {Client, Setup} <- Ready + ], + report(Rows, Iterations), + _ = [teardown(Client, Setup) || {Client, Setup} <- Ready], + ok. + +-doc "Run one minato workload under `eprof` and print the profile.". +-spec eprof(atom()) -> ok. +eprof(Workload) -> + eprof(Workload, ?DEFAULT_EPROF_ITERATIONS). + +-doc "`eprof/1` over a given number of iterations.". +-spec eprof(atom(), pos_integer()) -> ok. +eprof(Workload, Iterations) -> + Setup = setup(minato), + Run = runner(minato, Workload, Setup), + repeat(Run, 100), + {ok, _Pid} = eprof:start(), + profiling = eprof:start_profiling(processes()), + repeat(Run, Iterations), + profiling_stopped = eprof:stop_profiling(), + eprof:analyze(total), + eprof:stop(), + teardown(minato, Setup), + ok. + +%%---------------------------------------------------------------------- +%% Measuring the work rather than the wait +%%---------------------------------------------------------------------- + +measure(Client, Workload, Setup, Iterations) -> + case runner(Client, Workload, Setup) of + unsupported -> + unsupported; + Run -> + Count = count(Workload, Iterations), + repeat(Run, min(200, Count)), + _ = erlang:statistics(reductions), + Before = scheduler_time(), + {_Total, _Since} = erlang:statistics(reductions), + Start = erlang:monotonic_time(microsecond), + repeat(Run, Count), + Elapsed = erlang:monotonic_time(microsecond) - Start, + {_Again, Reductions} = erlang:statistics(reductions), + Busy = scheduler_time() - Before, + #{ + reductions => Reductions / Count, + busy_us => Busy / Count, + elapsed_us => Elapsed / Count + } + end. + +scheduler_time() -> + lists:sum([Active || {_Id, Active, _Total} <- erlang:statistics(scheduler_wall_time)]) / + 1000. + +report(Rows, Iterations) -> + io:format("~nminato cost profile, ~w iterations~n~n", [Iterations]), + io:format("~-12s ~-10s ~12s ~12s ~12s~n", [ + "workload", "client", "reductions", "busy us", "elapsed us" + ]), + io:format("~s~n", [lists:duplicate(62, $-)]), + _ = [line(Workload, Client, Measured) || {Workload, Client, Measured} <- Rows], + ok. + +line(_Workload, _Client, unsupported) -> + ok; +line(Workload, Client, #{reductions := Reductions, busy_us := Busy, elapsed_us := Elapsed}) -> + io:format("~-12s ~-10s ~12.1f ~12.1f ~12.1f~n", [Workload, Client, Reductions, Busy, Elapsed]). + +repeat(_Run, 0) -> + ok; +repeat(Run, N) -> + ok = Run(), + repeat(Run, N - 1). + +workloads() -> + [simple, unnamed, cached, rows_1000, rows_5000, wide_row, insert]. + +count(rows_5000, Iterations) -> max(1, Iterations div 20); +count(rows_1000, Iterations) -> max(1, Iterations div 5); +count(_Workload, Iterations) -> Iterations. + +iterations() -> + case os:getenv("MINATO_BENCH_ITERATIONS") of + false -> ?DEFAULT_ITERATIONS; + "" -> ?DEFAULT_ITERATIONS; + Value -> list_to_integer(Value) + end. + +%%---------------------------------------------------------------------- +%% Everything below borrows the benchmark's own setup and workloads +%%---------------------------------------------------------------------- + +setup(Client) -> + _ = application:ensure_started(crypto), + Setup = minato_bench_setup(Client), + Setup. + +minato_bench_setup(Client) -> + Setup = minato_bench:setup(Client), + ok = minato_bench:prepare_schema(Client, Setup), + minato_bench:parsed(Client, Setup). + +runner(Client, Workload, Setup) -> + minato_bench:runner(Client, Workload, Setup). + +teardown(Client, Setup) -> + minato_bench:teardown(Client, Setup). diff --git a/bench/py/bench.py b/bench/py/bench.py index 0a58085..47f942e 100644 --- a/bench/py/bench.py +++ b/bench/py/bench.py @@ -3,8 +3,8 @@ asyncpg prepares every statement it runs, and psycopg is asked to prepare, so the `cached` row means what it means in the Erlang report. - cd bench/py && python3 -m venv .venv && .venv/bin/pip install -q asyncpg psycopg[binary] - .venv/bin/python bench.py + bench/py/install.sh + bench/py/.venv/bin/python bench/py/bench.py """ import asyncio @@ -24,6 +24,8 @@ "password": os.environ.get("MINATO_PG_PASSWORD", "minato"), } +PSYCOPG_CONFIG = {"dbname" if k == "database" else k: v for k, v in CONFIG.items()} + ROWS = lambda n: f"SELECT i, repeat('x', 32) FROM generate_series(1, {n}) i" WIDE = "SELECT " + ", ".join(f"i * {n}" for n in range(1, 65)) + " FROM generate_series(1, 1) i" INSERT = "INSERT INTO minato_bench (n, name) VALUES ($1, $2)" @@ -92,7 +94,7 @@ async def concurrent(): async def run_psycopg(): import psycopg - conn = await psycopg.AsyncConnection.connect(autocommit=True, **CONFIG) + conn = await psycopg.AsyncConnection.connect(autocommit=True, **PSYCOPG_CONFIG) cursor = conn.cursor() def numbered(sql): @@ -110,7 +112,8 @@ async def concurrent(): )) run_psycopg.pool = [ - await psycopg.AsyncConnection.connect(autocommit=True, **CONFIG) for _ in range(WORKERS) + await psycopg.AsyncConnection.connect(autocommit=True, **PSYCOPG_CONFIG) + for _ in range(WORKERS) ] results = [ diff --git a/bench/py/install.sh b/bench/py/install.sh new file mode 100755 index 0000000..e33fd4e --- /dev/null +++ b/bench/py/install.sh @@ -0,0 +1,19 @@ +#!/usr/bin/env bash +# Put asyncpg and psycopg3 in bench/py/.venv, which bench/run.sh picks up. +# +# Debian and Ubuntu ship a python3 whose `venv` cannot bootstrap pip without the +# python3-venv package, so uv is tried first and needs no root. +set -euo pipefail + +cd "$(dirname "$0")" + +if command -v uv >/dev/null 2>&1; then + uv venv .venv + uv pip install --python .venv/bin/python asyncpg 'psycopg[binary]' +else + python3 -m venv .venv + .venv/bin/pip install -q --upgrade pip + .venv/bin/pip install -q asyncpg 'psycopg[binary]' +fi + +.venv/bin/python -c 'import asyncpg, psycopg; print("asyncpg", asyncpg.__version__, "psycopg", psycopg.__version__)' diff --git a/bench/run.sh b/bench/run.sh index eee07c3..fd2b53d 100755 --- a/bench/run.sh +++ b/bench/run.sh @@ -32,11 +32,14 @@ else echo "skipping javascript: run 'cd bench/js && npm install' first" fi -if python3 -c 'import asyncpg, psycopg' 2>/dev/null; then +PYTHON=python3 +[ -x bench/py/.venv/bin/python ] && PYTHON=bench/py/.venv/bin/python + +if "$PYTHON" -c 'import asyncpg, psycopg' 2>/dev/null; then echo "running python: asyncpg, psycopg3" - python3 bench/py/bench.py | tail -1 >> "$RESULTS" + "$PYTHON" bench/py/bench.py | tail -1 >> "$RESULTS" else - echo "skipping python: needs asyncpg and psycopg[binary] (python3-venv or pip)" + echo "skipping python: run 'bench/py/install.sh' first" fi node bench/report.mjs "$RESULTS" "$ITERATIONS" diff --git a/src/minato_conn.erl b/src/minato_conn.erl index 3e0121e..6c8201a 100644 --- a/src/minato_conn.erl +++ b/src/minato_conn.erl @@ -56,6 +56,17 @@ exchange and keep the session; with binding, the hash it bound to is not the one the real server sees, and the proof is refused. `channel_binding => require` refuses to connect without it. +## A refusal that arrives as a closed socket + +A server that refuses the startup exchange writes why and then the backend +exits. On Windows a socket closed with anything still unread is reset rather +than shut down, and a reset discards what the far end had already received, so +the reason can be gone before it can be read: `connect/1` answers +`{error, {socket, closed}}` or `{error, {socket, econnreset}}` where the same +call on Linux answers `{error, {pgsql_error, #{code := <<"3D000">>}}}`. The +connection is refused either way, and the server's log has the reason. Nothing +on this side can recover a message the kernel threw away. + ## Cleartext passwords A server may ask for the password in the clear. minato answers only when the @@ -117,6 +128,14 @@ the server issued is the whole of the authorisation. -include_lib("public_key/include/public_key.hrl"). +%% eqWAlizer cannot see that the transport field decides what the socket field +%% holds, and `ssl:connect/3` is specified for the sockets that existed before +%% it learned to take a handle from `m:socket`. Both are true at run time and +%% neither is expressible, so the two crossings live here rather than spread +%% through the module. +-eqwalizer({nowarn_function, handle/1}). +-eqwalizer({nowarn_function, upgraded/3}). + -record(conn, { transport :: transport(), socket :: socket(), @@ -131,9 +150,10 @@ the server issued is the whole of the authorisation. cancelled = false :: boolean(), grace :: timeout(), frame_opts :: minato_protocol:frame_opts(), + waiting :: minato_socket:waiting() | undefined, opened :: integer(), statements = #{} :: #{binary() => term()}, - prefix :: binary() | undefined, + prefix = ~"" :: binary(), prepared = 0 :: non_neg_integer(), limit :: non_neg_integer(), opts :: opts() @@ -156,6 +176,7 @@ How to reach a server and who to be. connect_timeout => timeout(), timeout => timeout(), socket_options => [gen_tcp:connect_option()], + transport => inet | socket, prepared_statements => non_neg_integer(), cancel_timeout => timeout(), frame_opts => minato_protocol:frame_opts()} @@ -181,6 +202,12 @@ to `prefer` under `ssl => true` and `disable` without it, since there is nothing to bind to on a plain socket. `require` refuses to connect at all unless the server offers `SCRAM-SHA-256-PLUS`. +`transport` is `inet`, the driver every BEAM program uses, or `socket`, which is +`m:minato_socket` over the socket NIFs. `socket` is experimental and faster per +operation, by about a quarter on a query and about a half on a pooled one, and +does everything the driver does on every machine minato runs on. See +`m:minato_socket`. + `prepared_statements` is how many statements this connection will keep parsed on the server for `minato_query:cached/4`. It defaults to 64, and 0 turns caching off. See `m:minato_query` for what fills it. @@ -204,6 +231,7 @@ given up on; see `deadline/2`. connect_timeout => timeout(), timeout => timeout(), socket_options => [gen_tcp:connect_option()], + transport => inet | socket, prepared_statements => non_neg_integer(), cancel_timeout => timeout(), frame_opts => minato_protocol:frame_opts() @@ -236,8 +264,9 @@ An authentication method the server may ask for. -export_type([conn/0, opts/0, method/0, key/0, error/0, report/0]). --type transport() :: gen_tcp | ssl. --type socket() :: gen_tcp:socket() | ssl:sslsocket(). +-type transport() :: plain() | ssl. +-type plain() :: gen_tcp | minato_socket. +-type socket() :: gen_tcp:socket() | ssl:sslsocket() | minato_socket:socket(). -doc """ Open a connection and complete the opening exchange. @@ -280,11 +309,16 @@ opening(Opts) -> Host = maps:get(host, Opts, ?DEFAULT_HOST), Port = maps:get(port, Opts, ?DEFAULT_PORT), Timeout = maps:get(connect_timeout, Opts, ?DEFAULT_CONNECT_TIMEOUT), - case gen_tcp:connect(Host, Port, socket_options(Opts), Timeout) of - {ok, Socket} -> secure(Socket, Host, Timeout, Opts); + Transport = transport(Opts), + case Transport:connect(Host, Port, socket_options(Opts), Timeout) of + {ok, Socket} -> secure(Transport, Socket, Host, Timeout, Opts); {error, Reason} -> {error, {socket, Reason}} end. +-spec transport(opts()) -> plain(). +transport(#{transport := socket}) -> minato_socket; +transport(_Opts) -> gen_tcp. + -doc """ Close a connection. @@ -311,8 +345,9 @@ cancel(#conn{key = {Pid, Secret}, opts = Opts}) -> Host = maps:get(host, Opts, ?DEFAULT_HOST), Port = maps:get(port, Opts, ?DEFAULT_PORT), Timeout = maps:get(connect_timeout, Opts, ?DEFAULT_CONNECT_TIMEOUT), - case gen_tcp:connect(Host, Port, socket_options(Opts), Timeout) of - {ok, Socket} -> cancel(Socket, Host, Timeout, {Pid, Secret}, Opts); + Transport = transport(Opts), + case Transport:connect(Host, Port, socket_options(Opts), Timeout) of + {ok, Socket} -> cancel(Transport, Socket, Host, Timeout, {Pid, Secret}, Opts); {error, Reason} -> {error, {socket, Reason}} end. @@ -484,44 +519,96 @@ cancelling(Conn, Deadline, Timeout) -> -doc """ Ask for the next bytes to arrive as a message instead of a read. -`{active, once}` on the socket, so exactly one message arrives and the owner -asks again. That is the arrangement a process which has to wait for the server -without a query outstanding needs - `LISTEN` is the only one there is - because -`next/1` blocks, and a process blocked in a read answers nothing else. +Exactly one message arrives and the owner asks again. That is the arrangement a +process which has to wait for the server without a query outstanding needs - +`LISTEN` is the only one there is - because `next/1` blocks, and a process +blocked in a read answers nothing else. + +`{active, once}` on the driver transport; on the socket transport a read that +either hands back what is already here, in which case the message is sent from +here, or registers this process for one `select`. Either way it is one message +to whoever called, which is why the connection comes back: what it is waiting +for is part of it. The connection is unusable for `next/1` until the messages stop coming, and the two must not be mixed on one connection. """. --spec activate(conn()) -> ok | {error, error()}. -activate(#conn{transport = Transport, socket = Socket}) -> - options(Transport, Socket, [{active, once}]). +-spec activate(conn()) -> {ok, conn()} | {error, error()}. +activate(#conn{transport = minato_socket} = Conn) -> + case minato_socket:activate(handle(Conn)) of + {ok, Data} -> {ok, delivered(Conn, Data)}; + {waiting, Waiting} -> {ok, Conn#conn{waiting = Waiting}}; + {error, Reason} -> {error, {socket, Reason}} + end; +activate(#conn{transport = Transport, socket = Socket} = Conn) -> + case options(Transport, Socket, [{active, once}]) of + ok -> {ok, Conn}; + {error, Reason} -> {error, Reason} + end. + +-spec delivered(conn(), binary()) -> conn(). +delivered(#conn{socket = Socket} = Conn, Data) -> + self() ! {minato_socket, Socket, Data}, + Conn#conn{waiting = undefined}. -doc """ Go back to reading, after `activate/1`. -`{active, false}` on the socket, and then whatever the socket already delivered -to the mailbox is taken back into the buffer, because those bytes have left the -socket and `next/1` would otherwise wait for bytes it already has. A connection -that has been switched back can be read with `next/1` as usual. +Whatever the socket already delivered to the mailbox is taken back into the +buffer, because those bytes have left the socket and `next/1` would otherwise +wait for bytes it already has. A connection that has been switched back can be +read with `next/1` as usual. + +Bytes that were waited for but never arrived are not lost on the socket +transport: cancelling the wait drops the message, not the socket's own buffer, +so the next `next/1` reads them. """. -spec deactivate(conn()) -> {ok, conn()} | {error, error()}. +deactivate(#conn{transport = minato_socket, waiting = undefined} = Conn) -> + {ok, flushed(Conn)}; +deactivate(#conn{transport = minato_socket, waiting = Waiting} = Conn) -> + Settled = settled(Conn, minato_socket:cancel(handle(Conn), Waiting)), + {ok, flushed(Settled#conn{waiting = undefined})}; deactivate(#conn{transport = Transport, socket = Socket} = Conn) -> case options(Transport, Socket, [{active, false}]) of ok -> {ok, flushed(Conn)}; {error, Reason} -> {error, Reason} end. +-spec settled(conn(), ok | pending) -> conn(). +settled(Conn, ok) -> + Conn; +settled(#conn{socket = Socket, timeout = Timeout} = Conn, pending) -> + receive + {'$socket', Delivered, _Kind, _What} = Message when Delivered =:= Socket -> + absorbed(Conn, Message) + after Timeout -> + Conn + end. + -spec flushed(conn()) -> conn(). flushed(#conn{transport = Transport, socket = Socket, buffer = Buffer} = Conn) -> receive {Tag, Delivered, Data} when Tag =:= Transport, Delivered =:= Socket, is_binary(Data) -> - flushed(Conn#conn{buffer = <>}) + flushed(Conn#conn{buffer = <>}); + {'$socket', Delivered, _Kind, _What} = Message when Delivered =:= Socket -> + flushed(absorbed(Conn, Message)) after 0 -> Conn end. +-spec absorbed(conn(), term()) -> conn(). +absorbed( + #conn{buffer = Buffer} = Conn, {'$socket', _Socket, completion, {_Handle, {ok, Data}}} +) when + is_binary(Data) +-> + Conn#conn{buffer = <>}; +absorbed(Conn, _Nothing_to_take) -> + Conn. + -doc """ Turn a socket message into the protocol messages it carried. @@ -543,6 +630,27 @@ handle_message(#conn{transport = ssl, socket = Socket} = Conn, {ssl, Socket, Dat is_binary(Data) -> arrived(Conn, Data); +handle_message(#conn{transport = minato_socket, socket = Socket} = Conn, {Tag, Socket, Data}) when + Tag =:= minato_socket, is_binary(Data) +-> + arrived(Conn#conn{waiting = undefined}, Data); +handle_message( + #conn{transport = minato_socket, socket = Socket, waiting = {select_info, _Tag, Handle}} = Conn, + {'$socket', Socket, select, Handle} +) -> + readable(Conn#conn{waiting = undefined}); +handle_message( + #conn{ + transport = minato_socket, socket = Socket, waiting = {completion_info, _Tag, Handle} + } = Conn, + {'$socket', Socket, completion, {Handle, {ok, Data}}} +) when is_binary(Data) -> + arrived(Conn#conn{waiting = undefined}, Data); +handle_message( + #conn{transport = minato_socket, socket = Socket, waiting = {completion_info, _Tag, Handle}}, + {'$socket', Socket, completion, {Handle, {error, Reason}}} +) -> + {closed, Reason}; handle_message(#conn{transport = gen_tcp, socket = Socket}, {tcp_closed, Socket}) -> {closed, closed}; handle_message(#conn{transport = gen_tcp, socket = Socket}, {tcp_error, Socket, Reason}) -> @@ -554,6 +662,16 @@ handle_message(#conn{transport = ssl, socket = Socket}, {ssl_error, Socket, Reas handle_message(_Conn, _Message) -> ignore. +-spec readable(conn()) -> + {messages, [minato_protocol:backend()], conn()} | {closed, term()}. +readable(Conn) -> + case minato_socket:activate(handle(Conn)) of + {ok, Data} -> arrived(Conn, Data); + {waiting, Waiting} -> {messages, [], Conn#conn{waiting = Waiting}}; + {error, closed} -> {closed, closed}; + {error, Reason} -> {closed, Reason} + end. + -spec arrived(conn(), binary()) -> {messages, [minato_protocol:backend()], conn()}. arrived(#conn{buffer = Buffer, pending = Pending, frame_opts = FrameOpts} = Conn, Data) -> {Messages, Rest} = minato_protocol:frame(<>, FrameOpts), @@ -572,8 +690,9 @@ options(Transport, Socket, Opts) -> {error, Reason} -> {error, {socket, Reason}} end. --spec options_module(transport()) -> inet | ssl. +-spec options_module(transport()) -> inet | ssl | minato_socket. options_module(gen_tcp) -> inet; +options_module(minato_socket) -> minato_socket; options_module(ssl) -> ssl. -doc """ @@ -739,53 +858,63 @@ with_timeout(Conn, Timeout) -> %% Opening %%---------------------------------------------------------------------- --spec secure(gen_tcp:socket(), inet:socket_address() | inet:hostname(), timeout(), opts()) -> +-spec secure(plain(), socket(), inet:socket_address() | inet:hostname(), timeout(), opts()) -> {ok, conn()} | {error, error()}. -secure(Socket, Host, Timeout, Opts) -> +secure(Transport, Socket, Host, Timeout, Opts) -> case maps:get(ssl, Opts, false) of false -> - start(new(gen_tcp, Socket, Opts), Opts); + start(new(Transport, Socket, Opts), Opts); true -> - case upgrade(Socket, Host, Timeout, Opts) of + case upgrade(Transport, Socket, Host, Timeout, Opts) of {ok, Upgraded} -> start(new(ssl, Upgraded, Opts), Opts); {error, Reason} -> {error, Reason} end end. --spec upgrade(gen_tcp:socket(), inet:socket_address() | inet:hostname(), timeout(), opts()) -> +-spec upgrade(plain(), socket(), inet:socket_address() | inet:hostname(), timeout(), opts()) -> {ok, ssl:sslsocket()} | {error, error()}. -upgrade(Socket, Host, Timeout, Opts) -> - case gen_tcp:send(Socket, minato_protocol:encode(ssl_request)) of - ok -> negotiate(Socket, Host, Timeout, Opts); - {error, Reason} -> closing(gen_tcp, Socket, {socket, Reason}) +upgrade(Transport, Socket, Host, Timeout, Opts) -> + case Transport:send(Socket, minato_protocol:encode(ssl_request)) of + ok -> negotiate(Transport, Socket, Host, Timeout, Opts); + {error, Reason} -> closing(Transport, Socket, {socket, Reason}) end. --spec negotiate(gen_tcp:socket(), inet:socket_address() | inet:hostname(), timeout(), opts()) -> +-spec negotiate(plain(), socket(), inet:socket_address() | inet:hostname(), timeout(), opts()) -> {ok, ssl:sslsocket()} | {error, error()}. -negotiate(Socket, Host, Timeout, Opts) -> - case gen_tcp:recv(Socket, 1, Timeout) of +negotiate(Transport, Socket, Host, Timeout, Opts) -> + case Transport:recv(Socket, 1, Timeout) of {ok, Answer} when is_binary(Answer) -> - answered(minato_protocol:ssl_response(Answer), Socket, Host, Timeout, Opts); + answered(minato_protocol:ssl_response(Answer), Transport, Socket, Host, Timeout, Opts); {error, Reason} -> - closing(gen_tcp, Socket, {socket, Reason}) + closing(Transport, Socket, {socket, Reason}) end. -spec answered( {ok, accepted | rejected | error_response, binary()} | incomplete, - gen_tcp:socket(), + plain(), + socket(), inet:socket_address() | inet:hostname(), timeout(), opts() ) -> {ok, ssl:sslsocket()} | {error, error()}. -answered({ok, accepted, _Rest}, Socket, Host, Timeout, Opts) -> - case ssl:connect(Socket, tls_options(Host, Opts), Timeout) of +answered({ok, accepted, _Rest}, Transport, Socket, Host, Timeout, Opts) -> + case upgraded(Socket, tls_options(Host, Opts), Timeout) of {ok, Upgraded} -> {ok, Upgraded}; - {error, Reason} -> closing(gen_tcp, Socket, {tls, Reason}) + {error, Reason} -> closing(Transport, Socket, {tls, Reason}) end; -answered({ok, _Refused, _Rest}, Socket, _Host, _Timeout, _Opts) -> - closing(gen_tcp, Socket, tls_refused); -answered(incomplete, Socket, _Host, _Timeout, _Opts) -> - closing(gen_tcp, Socket, {socket, closed}). +answered({ok, _Refused, _Rest}, Transport, Socket, _Host, _Timeout, _Opts) -> + closing(Transport, Socket, tls_refused); +answered(incomplete, Transport, Socket, _Host, _Timeout, _Opts) -> + closing(Transport, Socket, {socket, closed}). + +-spec handle(conn()) -> minato_socket:socket(). +handle(#conn{socket = Socket}) -> + Socket. + +-spec upgraded(socket(), [ssl:tls_client_option()], timeout()) -> + {ok, ssl:sslsocket()} | {error, term()}. +upgraded(Socket, Options, Timeout) -> + ssl:connect(Socket, Options, Timeout). -spec new(transport(), socket(), opts()) -> conn(). new(Transport, Socket, Opts) -> @@ -839,7 +968,8 @@ bound(Credentials, Hash, Binding) -> -spec prefer(transport()) -> minato_auth:channel_binding(). prefer(ssl) -> prefer; -prefer(gen_tcp) -> disable. +prefer(gen_tcp) -> disable; +prefer(minato_socket) -> disable. -doc """ The `tls-server-end-point` binding data for this connection, or `undefined`. @@ -858,6 +988,8 @@ channel_binding(#conn{binding = Hash}) -> -spec binding(transport(), socket()) -> binary() | undefined. binding(gen_tcp, _Socket) -> undefined; +binding(minato_socket, _Socket) -> + undefined; binding(ssl, Socket) -> case ssl:peercert(Socket) of {ok, Der} -> fingerprint(Der); @@ -979,10 +1111,10 @@ all_methods() -> %%---------------------------------------------------------------------- -spec cancel( - gen_tcp:socket(), inet:socket_address() | inet:hostname(), timeout(), key(), opts() + plain(), socket(), inet:socket_address() | inet:hostname(), timeout(), key(), opts() ) -> ok | {error, error()}. -cancel(Socket, Host, Timeout, {Pid, Secret}, Opts) -> - case secured(Socket, Host, Timeout, Opts) of +cancel(Transport, Socket, Host, Timeout, {Pid, Secret}, Opts) -> + case secured(Transport, Socket, Host, Timeout, Opts) of {ok, Transport, Secured} -> Sent = write(Transport, Secured, minato_protocol:encode({cancel_request, Pid, Secret})), _ = Transport:close(Secured), @@ -991,18 +1123,17 @@ cancel(Socket, Host, Timeout, {Pid, Secret}, Opts) -> {error, Reason} end. --spec secured(gen_tcp:socket(), inet:socket_address() | inet:hostname(), timeout(), opts()) -> +-spec secured(plain(), socket(), inet:socket_address() | inet:hostname(), timeout(), opts()) -> {ok, transport(), socket()} | {error, error()}. -secured(Socket, Host, Timeout, Opts) -> - case maps:get(ssl, Opts, false) of - false -> - {ok, gen_tcp, Socket}; - true -> - case upgrade(Socket, Host, Timeout, Opts) of - {ok, Upgraded} -> {ok, ssl, Upgraded}; - {error, Reason} -> {error, Reason} - end - end. +secured(Transport, Socket, _Host, _Timeout, #{ssl := false}) -> + {ok, Transport, Socket}; +secured(Transport, Socket, Host, Timeout, #{ssl := true} = Opts) -> + case upgrade(Transport, Socket, Host, Timeout, Opts) of + {ok, Upgraded} -> {ok, ssl, Upgraded}; + {error, Reason} -> {error, Reason} + end; +secured(Transport, Socket, _Host, _Timeout, _Opts) -> + {ok, Transport, Socket}. %%---------------------------------------------------------------------- %% The socket diff --git a/src/minato_listener.erl b/src/minato_listener.erl index b6f0945..304ab85 100644 --- a/src/minato_listener.erl +++ b/src/minato_listener.erl @@ -363,7 +363,7 @@ reactivated(#state{conn = undefined} = State) -> State; reactivated(#state{conn = Conn} = State) -> case minato_conn:activate(Conn) of - ok -> State; + {ok, Active} -> State#state{conn = Active}; {error, _Reason} -> dropped(State) end. diff --git a/src/minato_protocol_backend.erl b/src/minato_protocol_backend.erl index 6a17be7..1e84765 100644 --- a/src/minato_protocol_backend.erl +++ b/src/minato_protocol_backend.erl @@ -31,6 +31,11 @@ Left to `binary_to_atom/2` the first would raise `system_limit` and the second `badarg`, which would let whatever is on the far end of the socket choose an exception outside minato's own shape. +The four tags every workload actually ends on are matched whole, because the +general path splits the tag into a list, lowercases a copy of each word and +looks each one up in the atom table, and that is a surprising amount of work to +find out that a `SELECT` selected. + ## Unrecognised messages A type byte minato does not model decodes to `{unsupported, Type, Payload}`. @@ -203,11 +208,26 @@ parameter_oids(_Remaining, Payload) -> -spec command_complete(binary()) -> minato_protocol:tag(). command_complete(Payload) -> - {Tag, Rest} = cstring(command_complete, Payload), - exhausted(command_complete, Rest), - tag(Tag). + Size = byte_size(Payload) - 1, + case Payload of + <> -> tag(Tag); + _Unterminated_or_trailing -> malformed_tag(Payload) + end. + +-spec malformed_tag(binary()) -> no_return(). +malformed_tag(Payload) -> + {_Tag, Rest} = cstring(command_complete, Payload), + minato_protocol:bad(command_complete, trailing_bytes, Rest). -spec tag(binary()) -> minato_protocol:tag(). +tag(<<"SELECT ", Rows/binary>>) -> + {select, count_or_word(Rows)}; +tag(<<"INSERT 0 ", Rows/binary>>) -> + {insert, count_or_word(Rows)}; +tag(<<"UPDATE ", Rows/binary>>) -> + {update, count_or_word(Rows)}; +tag(<<"DELETE ", Rows/binary>>) -> + {delete, count_or_word(Rows)}; tag(Tag) -> case binary:split(Tag, ~" ", [global]) of [Command] -> {word(Command), undefined}; diff --git a/src/minato_query.erl b/src/minato_query.erl index 863cf0d..03b0f18 100644 --- a/src/minato_query.erl +++ b/src/minato_query.erl @@ -65,11 +65,27 @@ query, so they belong with the pool that hands connections out. -define(UNNAMED, ~""). +-define(DEFAULT_DECODING, #{ + return_rows_as_maps => false, + column_name_as_atom => false, + uuid_format => string, + datetime_format => datetime, + numeric_format => binary +}). + +-define(DEFAULT_CODEC, #{ + uuid_format => string, + datetime_format => datetime, + numeric_format => binary +}). + -record(statement, { name :: binary(), parameters :: [minato_oid:oid()], + parameter_formats :: [minato_codec:format()], fields :: [minato_protocol:field()], - formats :: [minato_codec:format()] + formats :: [minato_codec:format()], + decoder :: minato_protocol:decoder() }). -doc """ @@ -320,6 +336,10 @@ bounded(Conn, Opts) -> minato_conn:deadline(Conn, maps:get(timeout, Opts, minato_conn:timeout(Conn))). -spec decoding(opts()) -> minato_protocol:opts(). +decoding(Opts) when map_size(Opts) =:= 0 -> + ?DEFAULT_DECODING; +decoding(#{timeout := _} = Opts) when map_size(Opts) =:= 1 -> + ?DEFAULT_DECODING; decoding(Opts) -> #{ return_rows_as_maps => maps:get(return_rows_as_maps, Opts, false), @@ -406,11 +426,15 @@ described(Conn, Name, Parameters, Fields) -> -spec statement(binary(), [minato_oid:oid()] | undefined, [minato_protocol:field()]) -> statement(). statement(Name, Parameters, Fields) -> + Oids = parameters(Parameters), + Described = [Field#{format => format(Field)} || Field <- Fields], #statement{ name = Name, - parameters = parameters(Parameters), - fields = [Field#{format => format(Field)} || Field <- Fields], - formats = [format(Field) || Field <- Fields] + parameters = Oids, + parameter_formats = [minato_codec:preferred_format(Oid) || Oid <- Oids], + fields = Described, + formats = [format(Field) || Field <- Fields], + decoder = minato_protocol:decoder(Described, ?DEFAULT_DECODING) }. -spec format(minato_protocol:field()) -> minato_codec:format(). @@ -444,31 +468,52 @@ closed(Conn) -> -spec run(conn(), statement(), [term()], minato_protocol:opts()) -> {ok, result(), conn()} | {error, error(), conn()} | {error, error()}. run(Conn, #statement{} = Statement, Parameters, Opts) -> - #statement{name = Name, parameters = Oids, formats = Formats, fields = Fields} = Statement, - Encoded = encode(Conn, Oids, Parameters, Opts), - Bind = - {bind, ?UNNAMED, Name, [minato_codec:preferred_format(Oid) || Oid <- Oids], Encoded, - Formats}, + #statement{ + name = Name, + parameters = Oids, + parameter_formats = ParameterFormats, + formats = Formats, + fields = Fields + } = Statement, + Encoded = encode(Conn, Oids, ParameterFormats, Parameters, Opts), + Bind = {bind, ?UNNAMED, Name, ParameterFormats, Encoded, Formats}, + Decoder = decoder(Statement, Opts), case minato_conn:send(Conn, [Bind, {execute, ?UNNAMED, 0}, sync]) of - ok -> executed(decoding(Conn, Fields, Opts), minato_protocol:decoder(Fields, Opts), Opts); + ok -> executed(decoding(Conn, Fields, Decoder), Decoder, Opts); {error, Reason} -> {error, Reason} end. --spec encode(conn(), [minato_oid:oid()], [term()], minato_protocol:opts()) -> [iodata() | null]. -encode(_Conn, Oids, Parameters, Opts) when length(Oids) =:= length(Parameters) -> - encoded(Oids, Parameters, codec_opts(Opts)); -encode(Conn, Oids, Parameters, _Opts) -> +-spec decoder(statement(), minato_protocol:opts()) -> minato_protocol:decoder(). +decoder(#statement{decoder = Decoder}, #{ + return_rows_as_maps := false, + column_name_as_atom := false, + uuid_format := string, + datetime_format := datetime, + numeric_format := binary +}) -> + Decoder; +decoder(#statement{fields = Fields}, Opts) -> + minato_protocol:decoder(Fields, Opts). + +-spec encode( + conn(), [minato_oid:oid()], [minato_codec:format()], [term()], minato_protocol:opts() +) -> [iodata() | null]. +encode(_Conn, Oids, Formats, Parameters, Opts) when length(Oids) =:= length(Parameters) -> + encoded(Oids, Formats, Parameters, codec_opts(Opts)); +encode(Conn, Oids, _Formats, Parameters, _Opts) -> ok = resync(Conn), bad(bind, parameter_count_mismatch, {length(Oids), length(Parameters)}). --spec encoded([minato_oid:oid()], [term()], minato_codec:opts()) -> [iodata() | null]. -encoded([], [], _Opts) -> +-spec encoded([minato_oid:oid()], [minato_codec:format()], [term()], minato_codec:opts()) -> + [iodata() | null]. +encoded([], [], [], _Opts) -> []; -encoded([Oid | Oids], [Value | Values], Opts) -> - Encoded = minato_codec:encode(Oid, Value, minato_codec:preferred_format(Oid), Opts), - [Encoded | encoded(Oids, Values, Opts)]. +encoded([Oid | Oids], [Format | Formats], [Value | Values], Opts) -> + [minato_codec:encode(Oid, Value, Format, Opts) | encoded(Oids, Formats, Values, Opts)]. -spec codec_opts(minato_protocol:opts()) -> minato_codec:opts(). +codec_opts(#{uuid_format := string, datetime_format := datetime, numeric_format := binary}) -> + ?DEFAULT_CODEC; codec_opts(Opts) -> maps:with([uuid_format, datetime_format, numeric_format], Opts). @@ -477,11 +522,11 @@ codec_opts(Opts) -> executed(Conn, Decoder, _Opts) -> executed(Conn, Decoder, [], {unknown, undefined}). --spec decoding(conn(), [minato_protocol:field()], minato_protocol:opts()) -> conn(). -decoding(Conn, [], _Opts) -> +-spec decoding(conn(), [minato_protocol:field()], minato_protocol:decoder()) -> conn(). +decoding(Conn, [], _Decoder) -> Conn; -decoding(Conn, Fields, Opts) -> - minato_conn:decoding(Conn, minato_protocol:decoder(Fields, Opts)). +decoding(Conn, _Fields, Decoder) -> + minato_conn:decoding(Conn, Decoder). -spec executed( conn(), minato_protocol:decoder(), [[minato_protocol:row()]], minato_protocol:tag() diff --git a/src/minato_socket.erl b/src/minato_socket.erl new file mode 100644 index 0000000..ed7f5ad --- /dev/null +++ b/src/minato_socket.erl @@ -0,0 +1,281 @@ +-module(minato_socket). +-moduledoc """ +The `m:socket` module behind the same calls `m:gen_tcp` answers. + +Experimental, and not the default. `m:minato_conn` speaks to its transport +through `connect/4`, `send/2`, `recv/3`, `close/1`, `controlling_process/2` and +`setopts/2`, and this module answers all six on top of the socket NIFs so that +the transport can be chosen per connection with nothing else changing. + +## Why + +A send and a read through the inet driver cost about 323 reductions a round +trip: the driver's own layer monitors the port, serialises a reference, hands +the bytes over and waits for a reply message. The same round trip through +`m:socket` costs about 97, because a NIF call is a function call. + +The other half is the pool. A driver port has a connected process and only that +process may use it, so lending a connection out and taking it back is two +`controlling_process/2` calls, which cost about 389 reductions a query - more +than the query itself. A socket handle has an owner too, but only for how long +it lives; any process holding the handle can send and receive on it. Moving that +ownership is a NIF call rather than a port reassignment, and a pool that decided +a borrower's death should cost a reset rather than a connection could stop +moving it at all. + +## Waiting for bytes without reading + +`m:socket` has no active mode, so `setopts/2` refuses `{active, once}` rather +than pretending. What it has instead is `activate/1`: a read that either hands +back what is already here or arranges one message for when there is something. +That is the same arrangement `{active, once}` is - exactly one message, and the +owner asks again - and it costs no process, because the message goes to whoever +made the call rather than to a reader spawned to forward it. + +Windows completes reads rather than selecting on them: the wait answers with the +bytes in the message rather than with permission to go and read them. Both +shapes are here and `t:waiting/0` is either, because the difference belongs to +whoever reads the message, not to whoever waits. + +## TLS + +`ssl:connect/3` takes one of these handles, so a connection that asks for TLS +sends its `SSLRequest` through this module and hands the upgraded socket to +`m:ssl` exactly as the driver transport does. What comes back is an ordinary +`t:ssl:sslsocket/0`, so nothing below TLS is this module's business afterwards - +including the certificate hash SCRAM binds to. +""". + +-export([connect/4, send/2, recv/3, close/1, controlling_process/2, setopts/2]). +-export([activate/1, cancel/2]). + +%% eqWAlizer reads the eight words of an `inet:ip6_address()` as `integer()` and +%% the eight of a `socket:in6_addr()` as `0..65535`, so a v6 address that came +%% from `inet:getaddr/2` cannot be written into a `socket:sockaddr()` without +%% this. The v4 half of the same function checks. +-eqwalizer({nowarn_function, target/2}). + +-doc """ +A socket handle. + +Unlike a port it is not owned for the purpose of using it: any process holding +one can read and write. It is owned for the purpose of living, and is closed +when that process terminates. +""". +-type socket() :: socket:socket(). + +-doc """ +What `activate/1` is waiting on, which is what `cancel/2` drops. + +A select token where the machine says the socket is readable, a completion token +where it hands over the bytes. +""". +-type waiting() :: socket:select_info() | socket:completion_info(). + +-export_type([socket/0, waiting/0]). + +-doc """ +Open a connection. + +Takes the option list `m:gen_tcp` would take, so a caller does not have to know +which transport it got. `binary`, `{active, false}` and `{packet, raw}` are what +this module always does and are accepted and ignored; `{nodelay, Boolean}` and +`{buffer, _}` are the two that mean anything here, and anything else is refused +rather than silently dropped. +""". +-spec connect( + inet:socket_address() | inet:hostname(), inet:port_number(), [term()], timeout() +) -> {ok, socket()} | {error, term()}. +connect(Host, Port, Opts, Timeout) -> + case address(Host, Port) of + {ok, {Domain, Address}} -> opened(Domain, Address, Opts, Timeout); + {error, Reason} -> {error, Reason} + end. + +-doc """ +Send every byte, or say why not. + +A send with no deadline blocks until the whole of it has gone, but the call can +still answer with what is left rather than with `ok`, so what is left is sent. +""". +-spec send(socket(), iodata()) -> ok | {error, term()}. +send(Socket, Data) -> + case socket:send(Socket, Data) of + ok -> ok; + {ok, Rest} -> send(Socket, Rest); + {error, Reason} -> {error, Reason} + end. + +-doc """ +Read `Length` bytes, or whatever has arrived when `Length` is 0. + +A timeout of 0 is the non-blocking probe `minato_conn:usable/1` makes, and +answers `{error, timeout}` when nothing is waiting. +""". +-spec recv(socket(), non_neg_integer(), timeout()) -> {ok, binary()} | {error, term()}. +recv(Socket, Length, Timeout) -> + case socket:recv(Socket, Length, Timeout) of + {ok, Data} -> {ok, Data}; + {error, {timeout, _Partial}} -> {error, timeout}; + {error, Reason} -> {error, Reason} + end. + +-doc "Close the socket.". +-spec close(socket()) -> ok. +close(Socket) -> + _ = socket:close(Socket), + ok. + +-doc """ +Hand the socket's lifetime to another process. + +A socket handle has an owner, but only for how long it lives: it is closed when +that process terminates. Reading and writing need no ownership at all, which a +port does, so lending a connection out could stop moving ownership entirely and +let the lender keep it - a borrower that dies would then cost a connection reset +rather than a connection. That belongs to whoever lends, so this answers the +call `m:gen_tcp` answers, and answers it with one NIF rather than a port +reassignment. +""". +-spec controlling_process(socket(), pid()) -> ok | {error, term()}. +controlling_process(Socket, Pid) -> + socket:setopt(Socket, {otp, controlling_process}, Pid). + +-doc """ +Take the bytes that are here, or arrange for a message when there are some. + +`{waiting, Waiting}` means one message will be sent to the caller, and which +message depends on how the machine waits: + +- `{'$socket', Socket, select, Handle}` says the socket is readable and the + bytes are still in it, so reading is this call again. A wakeup with nothing + behind it therefore re-arms rather than being mistaken for data, which is why + there is one function here and not an arm and a take. +- `{'$socket', Socket, completion, {Handle, {ok, Data}}}` carries the bytes, + and calling again would ask for the next ones. + +The `Handle` is the one inside `Waiting` either way. The waiting is registered +for the calling process, so whoever wants the message has to make the call. +""". +-spec activate(socket()) -> {ok, binary()} | {waiting, waiting()} | {error, term()}. +activate(Socket) -> + case socket:recv(Socket, 0, nowait) of + {ok, Data} -> {ok, Data}; + {select, Select} -> {waiting, Select}; + {completion, Completion} -> {waiting, Completion}; + {error, Reason} -> {error, Reason} + end. + +-doc """ +Drop a waiting arranged by `activate/1`. + +`pending` means it was too late and the message is on its way, which matters +where the message carries the bytes: dropping it there would drop them. `ok` +means nothing is coming. + +Bytes still in the socket are not lost either way. The next `recv/3` reads them. +""". +-spec cancel(socket(), waiting()) -> ok | pending. +cancel(Socket, Waiting) -> + case socket:cancel(Socket, Waiting) of + ok -> ok; + {error, _Too_late} -> pending + end. + +-doc "Set what can be set. `{active, _}` is refused; see the module note.". +-spec setopts(socket(), [term()]) -> ok | {error, term()}. +setopts(Socket, [{active, false} | Rest]) -> + setopts(Socket, Rest); +setopts(_Socket, [{active, _Mode} | _Rest]) -> + {error, active_mode_unsupported}; +setopts(Socket, [{nodelay, Value} | Rest]) -> + case socket:setopt(Socket, {tcp, nodelay}, Value) of + ok -> setopts(Socket, Rest); + {error, Reason} -> {error, Reason} + end; +setopts(Socket, [_Ignored | Rest]) -> + setopts(Socket, Rest); +setopts(_Socket, []) -> + ok. + +%%---------------------------------------------------------------------- +%% Opening +%%---------------------------------------------------------------------- + +-spec opened(socket:domain(), socket:sockaddr(), [term()], timeout()) -> + {ok, socket()} | {error, term()}. +opened(Domain, Address, Opts, Timeout) -> + case socket:open(Domain, stream, tcp) of + {ok, Socket} -> started(Socket, Address, Opts, Timeout); + {error, Reason} -> {error, Reason} + end. + +-spec started(socket(), socket:sockaddr(), [term()], timeout()) -> + {ok, socket()} | {error, term()}. +started(Socket, Address, Opts, Timeout) -> + case bound(Socket, Address) of + ok -> connecting(Socket, Address, Opts, Timeout); + {error, Reason} -> failed(Socket, Reason) + end. + +-doc false. +-spec bound(socket(), socket:sockaddr()) -> ok | {error, term()}. +%% Windows refuses to connect a socket that was never bound, and answers +%% `not_bound` rather than binding it for you. Binding to the wildcard is what +%% the driver does on every machine, and costs nothing where it was not needed. +bound(Socket, #{family := Family}) when Family =:= inet; Family =:= inet6 -> + socket:bind(Socket, any); +bound(_Socket, _Local) -> + ok. + +-spec connecting(socket(), socket:sockaddr(), [term()], timeout()) -> + {ok, socket()} | {error, term()}. +connecting(Socket, Address, Opts, Timeout) -> + case socket:connect(Socket, Address, Timeout) of + ok -> tuned(Socket, Opts); + {error, Reason} -> failed(Socket, Reason) + end. + +-spec tuned(socket(), [term()]) -> {ok, socket()} | {error, term()}. +tuned(Socket, Opts) -> + case setopts(Socket, Opts) of + ok -> {ok, Socket}; + {error, Reason} -> failed(Socket, Reason) + end. + +-spec failed(socket(), term()) -> {error, term()}. +failed(Socket, Reason) -> + ok = close(Socket), + {error, Reason}. + +-spec address(inet:socket_address() | inet:hostname(), inet:port_number()) -> + {ok, {socket:domain(), socket:sockaddr()}} | {error, term()}. +address({local, Path}, _Port) -> + {ok, {local, #{family => local, path => iolist_to_binary(Path)}}}; +address(Host, Port) when is_list(Host); is_atom(Host) -> + resolved(Host, Port); +address(Address, Port) -> + {ok, target(Address, Port)}. + +-spec resolved(inet:hostname(), inet:port_number()) -> + {ok, {socket:domain(), socket:sockaddr()}} | {error, term()}. +resolved(Host, Port) -> + case inet:getaddr(Host, inet) of + {ok, Address} -> {ok, target(Address, Port)}; + {error, _Not_a_v4_address} -> resolved6(Host, Port) + end. + +-spec resolved6(inet:hostname(), inet:port_number()) -> + {ok, {socket:domain(), socket:sockaddr()}} | {error, term()}. +resolved6(Host, Port) -> + case inet:getaddr(Host, inet6) of + {ok, Address} -> {ok, target(Address, Port)}; + {error, Reason} -> {error, Reason} + end. + +-spec target(inet:ip_address(), inet:port_number()) -> + {socket:domain(), socket:sockaddr()}. +target({A, B, C, D}, Port) -> + {inet, #{family => inet, addr => {A, B, C, D}, port => Port}}; +target({A, B, C, D, E, F, G, H}, Port) -> + {inet6, #{family => inet6, addr => {A, B, C, D, E, F, G, H}, port => Port}}. diff --git a/test/minato_conn_tests.erl b/test/minato_conn_tests.erl index d79a370..01f070d 100644 --- a/test/minato_conn_tests.erl +++ b/test/minato_conn_tests.erl @@ -145,7 +145,11 @@ a_server_that_closes_during_startup_is_reported_test() -> {ok, Socket} = gen_tcp:accept(Listen, 5000), ok = gen_tcp:close(Socket) end), - ?assertEqual({error, {socket, closed}}, minato_conn:connect(opts(Port, #{}))), + %% Which word the machine uses for a peer that hung up mid-exchange is the + %% machine's: Linux says the socket closed, Windows says the connection was + %% aborted. What is being tested is that it is reported at all, and reported + %% as a socket failure rather than as a connection. + ?assertMatch({error, {socket, _Hung_up}}, minato_conn:connect(opts(Port, #{}))), ok = gen_tcp:close(Listen). what_the_server_sent_beyond_ready_is_readable_test() -> diff --git a/test/minato_connection_SUITE.erl b/test/minato_connection_SUITE.erl index 00af6b2..d880a14 100644 --- a/test/minato_connection_SUITE.erl +++ b/test/minato_connection_SUITE.erl @@ -79,7 +79,16 @@ a_wrong_password_is_refused(_Config) -> -spec an_unknown_database_is_refused([{atom(), term()}]) -> ok. an_unknown_database_is_refused(_Config) -> Result = minato_conn:connect(opts(#{database => ~"no_such_database"})), - ?assertMatch({error, {pgsql_error, #{code := ~"3D000"}}}, Result). + ?assert(refused(Result)). + +%% The server writes its FATAL and then the backend exits. On Windows a socket +%% closed with anything still unread is reset rather than shut down, and a reset +%% discards what the other end had already received, so the message can be gone +%% before it is read. What has to hold everywhere is that the connection is +%% refused and that minato says why as far as it was told. +refused({error, {pgsql_error, #{code := ~"3D000"}}}) -> true; +refused({error, {socket, _Gone_before_it_was_read}}) -> os:type() =:= {win32, nt}; +refused(_Connected_or_something_else) -> false. -spec a_password_fun_is_called_at_connect_time([{atom(), term()}]) -> ok. a_password_fun_is_called_at_connect_time(_Config) -> diff --git a/test/minato_differential_SUITE.erl b/test/minato_differential_SUITE.erl index 3eaa583..dc80357 100644 --- a/test/minato_differential_SUITE.erl +++ b/test/minato_differential_SUITE.erl @@ -295,6 +295,13 @@ accordingly. Neither is minato's defect and neither is a reason to stop testing the value: the reserved minimum is `-infinity`, not a date, on every version. """. -spec known(atom(), term(), #{atom() => term()}) -> {yes, atom()} | no. +known(text_encode, Type, #{from_text := Text, from_binary := Binary}) when + is_binary(Text), + is_binary(Binary), + Type =:= float4 orelse Type =:= float8 orelse + Type =:= {array, float4} orelse Type =:= {array, float8} +-> + nan_payload(Type, Text, Binary); known(oracle_encode, Type, #{unsupported := no_null_encoding}) when Type =:= {array, bool}; Type =:= {array, bytea}; @@ -441,6 +448,32 @@ zip([], _Other) -> []; zip(_Mine, []) -> []; zip([Mine | Rest], [Theirs | OtherRest]) -> [{Mine, Theirs} | zip(Rest, OtherRest)]. +-doc """ +Whether two encodings of the same float differ only in the bits of a NaN. + +IEEE 754 says nothing about the payload a NaN gets when one is made from the +string "NaN", so a server whose C library writes `7fffffff` and one that writes +`7fc00000` are both right and both mean NaN. The two are told apart from a real +disagreement by decoding them: every NaN decodes to the atom, so bytes that +differ and values that do not is the payload and nothing else. +""". +-spec nan_payload(minato_oid:name() | {array, minato_oid:name()}, binary(), binary()) -> + {yes, atom()} | no. +nan_payload(Type, Text, Binary) -> + Oid = minato_oid:oid(Type), + case {value(Oid, Text), value(Oid, Binary)} of + {{ok, Same}, {ok, Same}} -> {yes, nan_payloads_are_not_specified}; + _Really_different -> no + end. + +-spec value(minato_oid:oid(), binary()) -> {ok, term()} | error. +value(Oid, Bytes) -> + try + {ok, minato_codec:decode(Oid, Bytes, binary, #{})} + catch + _Class:_Reason -> error + end. + -spec worst([{yes, atom()} | no]) -> {yes, atom()} | no. worst([]) -> no; @@ -730,12 +763,19 @@ stores whatever bit pattern it is handed. """. -spec minato_encodes_a_quiet_nan(ct_suite:ct_config()) -> ok. minato_encodes_a_quiet_nan(_Config) -> - ?assertEqual(<<16#7FC00000:32>>, server_binary(~"float4", ~"NaN")), - ?assertEqual(<<16#7FF8000000000000:64>>, server_binary(~"float8", ~"NaN")), + %% What minato writes is the canonical quiet NaN, and that is asserted + %% exactly. What the server writes is whatever its C library makes of the + %% string, which is `7fc00000` on glibc and `7fffffff` on Windows and is a + %% quiet NaN either way - the payload of a NaN is not specified, so only the + %% quietness is worth asserting. ?assertEqual(<<16#7FC00000:32>>, minato_corpus:encoded(700, nan, binary, #{})), ?assertEqual( <<16#7FF8000000000000:64>>, minato_corpus:encoded(701, nan, binary, #{}) ), + ?assertMatch(<<_Sign:1, 255:8, 1:1, _Payload:22>>, server_binary(~"float4", ~"NaN")), + ?assertMatch( + <<_Sign:1, 2047:11, 1:1, _Payload:51>>, server_binary(~"float8", ~"NaN") + ), ok. -doc """ diff --git a/test/minato_socket_SUITE.erl b/test/minato_socket_SUITE.erl new file mode 100644 index 0000000..6fade78 --- /dev/null +++ b/test/minato_socket_SUITE.erl @@ -0,0 +1,298 @@ +-module(minato_socket_SUITE). +-moduledoc """ +The `socket` transport, against a real PostgreSQL. + +`m:minato_socket` answers the same six calls `m:gen_tcp` answers, so the point +of this suite is that everything above the transport cannot tell which one it +got: the same authentication, the same query paths, the same errors leaving a +connection usable, and the same pool lending it out and taking it back. + +The two hard parts are proved rather than described. `m:ssl` takes one of these +handles directly, so TLS is the same exchange over a different socket and the +certificate hash SCRAM binds to is the same hash. `m:socket` has no active mode, +so `LISTEN` is a `select` registered by the owning process and answered in its +own mailbox - no reader process between the socket and the connection - and a +notification has to arrive through it. + +Skips when no server answers. +""". + +-include_lib("stdlib/include/assert.hrl"). + +-export([all/0, init_per_suite/1, end_per_suite/1]). +-export([ + a_query_runs/1, + a_cached_statement_runs_more_than_once/1, + a_failing_statement_leaves_the_connection_usable/1, + a_transaction_commits/1, + a_result_set_spanning_many_reads_arrives_whole/1, + a_pool_lends_a_connection_and_takes_it_back/1, + a_pooled_connection_outlives_the_process_that_opened_it/1, + a_borrower_that_dies_does_not_take_the_pool_down/1, + tls_runs_over_this_transport/1, + a_notification_arrives_over_this_transport/1, + a_connection_goes_back_to_reading_after_waiting/1, + a_listener_runs_over_this_transport/1 +]). + +-define(TIMEOUT, 10000). + +-spec all() -> [atom()]. +all() -> + [ + a_query_runs, + a_cached_statement_runs_more_than_once, + a_failing_statement_leaves_the_connection_usable, + a_transaction_commits, + a_result_set_spanning_many_reads_arrives_whole, + a_pool_lends_a_connection_and_takes_it_back, + a_pooled_connection_outlives_the_process_that_opened_it, + a_borrower_that_dies_does_not_take_the_pool_down, + tls_runs_over_this_transport, + a_notification_arrives_over_this_transport, + a_connection_goes_back_to_reading_after_waiting, + a_listener_runs_over_this_transport + ]. + +%% Reachability is asked over the driver transport, so that a socket transport +%% which cannot connect fails the suite instead of skipping it. A skip here +%% would mean "no PostgreSQL", and it has to keep meaning only that. +-spec init_per_suite([{atom(), term()}]) -> [{atom(), term()}] | {skip, term()}. +init_per_suite(Config) -> + {ok, _Started} = application:ensure_all_started(minato), + case minato_conn:connect(maps:remove(transport, opts())) of + {ok, Conn} -> + ok = minato_conn:close(Conn), + Config; + {error, Reason} -> + {skip, {no_server, Reason}} + end. + +-spec end_per_suite([{atom(), term()}]) -> ok. +end_per_suite(_Config) -> + ok. + +%%---------------------------------------------------------------------- +%% The same client, a different transport +%%---------------------------------------------------------------------- + +a_query_runs(_Config) -> + Conn = connected(), + {ok, #{rows := [{42}], command := select}, Ready} = + minato_query:query(Conn, ~"SELECT $1::int4", [42]), + ok = minato_conn:close(Ready). + +a_cached_statement_runs_more_than_once(_Config) -> + Conn = connected(), + {ok, #{rows := [{1}]}, First} = minato_query:cached(Conn, ~"SELECT $1::int4", [1]), + {ok, #{rows := [{2}]}, Second} = minato_query:cached(First, ~"SELECT $1::int4", [2]), + ?assertEqual({1, 64}, minato_conn:statements(Second)), + ok = minato_conn:close(Second). + +a_failing_statement_leaves_the_connection_usable(_Config) -> + Conn = connected(), + {error, {pgsql_error, #{code := ~"42601"}}, Usable} = + minato_query:query(Conn, ~"NOT SQL AT ALL", []), + {ok, #{rows := [{1}]}, Ready} = minato_query:query(Usable, ~"SELECT $1::int4", [1]), + ok = minato_conn:close(Ready). + +a_transaction_commits(_Config) -> + Conn = connected(), + {ok, done, Ready} = minato_txn:transaction(Conn, fun(Inside) -> + {ok, #{rows := [{7}]}, Written} = minato_query:query(Inside, ~"SELECT $1::int4", [7]), + {ok, done, Written} + end), + ok = minato_conn:close(Ready). + +a_result_set_spanning_many_reads_arrives_whole(_Config) -> + Conn = connected(), + Sql = ~"SELECT i, repeat('x', 32) FROM generate_series(1, 5000) i", + {ok, #{rows := Rows, num_rows := 5000}, Ready} = minato_query:cached(Conn, Sql, []), + ?assertEqual(5000, length(Rows)), + ?assertEqual({1, ~"xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx"}, hd(Rows)), + ?assertEqual({5000, ~"xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx"}, lists:last(Rows)), + ok = minato_conn:close(Ready). + +%%---------------------------------------------------------------------- +%% The pool, which is where the ownership question lives +%%---------------------------------------------------------------------- + +a_pool_lends_a_connection_and_takes_it_back(_Config) -> + Name = pool(lends), + {ok, #{rows := [{42}]}} = minato:query(Name, ~"SELECT $1::int4", [42]), + {ok, #{rows := [{43}]}} = minato:query(Name, ~"SELECT $1::int4", [43]), + #{disconnects := 0, checkouts := 2} = minato_pool:stats(Name), + _ = minato:stop_pool(Name). + +a_pooled_connection_outlives_the_process_that_opened_it(_Config) -> + Name = pool(outlives), + _ = [minato:query(Name, ~"SELECT $1::int4", [N]) || N <- lists:seq(1, 20)], + #{disconnects := 0} = minato_pool:stats(Name), + _ = minato:stop_pool(Name). + +a_borrower_that_dies_does_not_take_the_pool_down(_Config) -> + Name = pool(borrower), + {ok, _Result} = minato:query(Name, ~"SELECT $1::int4", [1]), + {Pid, Ref} = spawn_monitor(fun() -> + {ok, _Conn} = minato_pool:checkout(Name), + exit(taken_out_with_a_connection) + end), + receive + {'DOWN', Ref, process, Pid, _Reason} -> ok + after ?TIMEOUT -> ct:fail(borrower_never_died) + end, + {ok, #{rows := [{2}]}} = minato:query(Name, ~"SELECT $1::int4", [2]), + _ = minato:stop_pool(Name). + +%%---------------------------------------------------------------------- +%% TLS, which ssl does over one of these handles unchanged +%%---------------------------------------------------------------------- + +tls_runs_over_this_transport(_Config) -> + case reachable() of + false -> {skip, no_tls_server}; + true -> over_tls() + end. + +%% The test certificates are made by the second compose service, so a machine +%% running the suite without it has no CA to verify against and no server to +%% verify. Reading the file first keeps that a skip rather than an enoent. +reachable() -> + filelib:is_regular(certificate("ca.crt")). + +over_tls() -> + case minato_conn:connect(secured()) of + {ok, Conn} -> + {ok, #{rows := [{42}]}, Ready} = minato_query:query(Conn, ~"SELECT $1::int4", [42]), + ?assertNotEqual(undefined, minato_conn:channel_binding(Ready)), + {ok, #{rows := [{true}]}, Read} = + minato_query:query( + Ready, ~"SELECT ssl FROM pg_stat_ssl WHERE pid = pg_backend_pid()", [] + ), + ok = minato_conn:close(Read); + {error, Reason} -> + {skip, {no_tls_server, Reason}} + end. + +%%---------------------------------------------------------------------- +%% Waiting for the server without a read outstanding +%%---------------------------------------------------------------------- + +a_notification_arrives_over_this_transport(_Config) -> + Conn = connected(), + {ok, _Results, Listening} = minato_query:simple(Conn, ~"LISTEN minato_socket_suite"), + {ok, Waiting} = minato_conn:activate(Listening), + ok = notify(~"minato_socket_suite", ~"without a read outstanding"), + {Kind, Notification} = awaited(Waiting), + ?assertEqual(shape(os:type()), Kind), + ?assertMatch( + {notification_response, _Pid, ~"minato_socket_suite", ~"without a read outstanding"}, + Notification + ), + ok = minato_conn:close(Waiting). + +a_connection_goes_back_to_reading_after_waiting(_Config) -> + Conn = connected(), + {ok, Waiting} = minato_conn:activate(Conn), + {ok, Passive} = minato_conn:deactivate(Waiting), + {ok, #{rows := [{1}]}, Ready} = minato_query:query(Passive, ~"SELECT $1::int4", [1]), + ok = minato_conn:close(Ready). + +a_listener_runs_over_this_transport(_Config) -> + Name = minato_socket_suite_listener, + {ok, _Pid} = minato:start_listener(Name, #{connection => opts()}), + ok = minato:listen(Name, ~"minato_socket_listener"), + ok = notify(~"minato_socket_listener", ~"all the way up"), + receive + {minato_notification, ~"minato_socket_listener", ~"all the way up", _From} -> ok + after ?TIMEOUT -> ct:fail(no_notification) + end, + ok = minato:listen(Name, ~"minato_socket_listener_again"), + ok = notify(~"minato_socket_listener_again", ~"and again"), + receive + {minato_notification, ~"minato_socket_listener_again", ~"and again", _Also} -> ok + after ?TIMEOUT -> ct:fail(no_second_notification) + end, + ok = minato:stop_listener(Name). + +%% The kind is asserted rather than assumed, because it is the whole difference +%% between the two machines: POSIX says the socket is readable and leaves the +%% bytes in it, Windows completes the read and puts them in the message. A run +%% that took the other path would otherwise pass without ever touching this one. +awaited(Conn) -> + awaited(Conn, undefined). + +awaited(Conn, Kind) -> + receive + Message -> + case minato_conn:handle_message(Conn, Message) of + {messages, [], Again} -> awaited(Again, kind(Message, Kind)); + {messages, Messages, _Read} -> {kind(Message, Kind), hd(Messages)}; + ignore -> awaited(Conn, Kind); + {closed, Reason} -> ct:fail({closed, Reason}) + end + after ?TIMEOUT -> ct:fail(no_notification) + end. + +kind({'$socket', _Socket, Kind, _What}, _Seen) -> Kind; +kind(_Sent_from_here, Seen) -> Seen. + +shape({win32, _Flavour}) -> completion; +shape(_Posix) -> select. + +notify(Channel, Payload) -> + {ok, Conn} = minato_conn:connect(opts()), + Sql = <<"NOTIFY ", Channel/binary, ", '", Payload/binary, "'">>, + {ok, _Results, Ready} = minato_query:simple(Conn, Sql), + minato_conn:close(Ready). + +%%---------------------------------------------------------------------- + +connected() -> + {ok, Conn} = minato_conn:connect(opts()), + Conn. + +secured() -> + (opts())#{ + host => env("MINATO_TLS_HOST", "localhost"), + port => list_to_integer(env("MINATO_TLS_PORT", "55433")), + ssl => true, + ssl_options => [{cacerts, cacerts()}] + }. + +cacerts() -> + {ok, Pem} = file:read_file(certificate("ca.crt")), + [Der || {'Certificate', Der, not_encrypted} <- public_key:pem_decode(Pem)]. + +certificate(Name) -> + filename:join([directory(code:lib_dir(minato)), "test", "certs", Name]). + +directory({error, Reason}) -> + error({no_application_directory, Reason}); +directory(Directory) -> + Directory. + +pool(Case) -> + Name = list_to_atom("minato_socket_suite_" ++ atom_to_list(Case)), + {ok, _Pid} = minato:start_pool(Name, #{ + size => 2, min_size => 2, connection => opts() + }), + Name. + +opts() -> + #{ + host => env("MINATO_PG_HOST", "127.0.0.1"), + port => list_to_integer(env("MINATO_PG_PORT", "55432")), + user => list_to_binary(env("MINATO_PG_USER", "minato")), + password => list_to_binary(env("MINATO_PG_PASSWORD", "minato")), + database => list_to_binary(env("MINATO_PG_DATABASE", "minato_test")), + transport => socket, + timeout => ?TIMEOUT + }. + +env(Name, Default) -> + case os:getenv(Name) of + false -> Default; + "" -> Default; + Value -> Value + end. diff --git a/test/minato_telemetry_SUITE.erl b/test/minato_telemetry_SUITE.erl index fff30b6..1287917 100644 --- a/test/minato_telemetry_SUITE.erl +++ b/test/minato_telemetry_SUITE.erl @@ -82,8 +82,9 @@ end_per_testcase(Case, _Config) -> -spec a_query_is_a_span([{atom(), term()}]) -> ok. a_query_is_a_span(_Config) -> {ok, _Result} = minato:query(?POOL, ~"SELECT $1::int4", [7]), - {_Start, StartMeta} = awaited([minato, query, start]), - {Stop, StopMeta} = awaited([minato, query, stop]), + {_Start, StartMeta} = awaited([minato, query, start], #{path => cached}), + #{telemetry_span_context := Span} = StartMeta, + {Stop, StopMeta} = awaited([minato, query, stop], #{telemetry_span_context => Span}), ?assertMatch(#{path := cached}, StartMeta), ?assertMatch(#{duration := _Duration, monotonic_time := _When}, Stop), ?assertMatch(#{outcome := ok, command := select, num_rows := 1}, StopMeta). @@ -188,6 +189,18 @@ awaited(Name) -> after ?TIMEOUT -> error({never_emitted, Name}) end. +%% A pool opens connections when it needs them rather than up front, and the +%% first connection of one asks the server how many it will allow. That is a +%% query, and it is a span, so waiting for "the next event" is waiting for +%% whichever query got there first. This waits for the one the test ran, which +%% for a stop event means the one carrying the start event's span context. +awaited(Name, Wanted) -> + {Measurements, Metadata} = awaited(Name), + case maps:with(maps:keys(Wanted), Metadata) of + Wanted -> {Measurements, Metadata}; + _Somebody_elses -> awaited(Name, Wanted) + end. + drained() -> receive {telemetry, Name, Measurements, Metadata} -> [{Name, Measurements, Metadata} | drained()]