minato_conn (minato v0.18.6)

View Source

One connection to a PostgreSQL server, owned by the process that opened it.

connect/1 performs the whole opening exchange - TCP, optionally TLS, the StartupMessage, authentication through minato_auth, and then the parameters and the BackendKeyData the server volunteers - and hands back a connection the calling process owns outright.

There is no connection process. The socket is {active, false} and belongs to the caller, so a query is a send and a read on the caller's own scheduler slot rather than a message to a gen_server that puts every caller behind the slowest one. The PostgreSQL protocol is FIFO per connection, so a single owning process is an unpreemptable queue: one slow query stalls every other caller of that process, and no timeout on the caller's side can take the connection back. Caller ownership also gives session pinning, which transactions, advisory locks, SET ROLE and LISTEN all need to be correct rather than merely fast.

That is why conn/0 is threaded through next/1 rather than being a handle with hidden state: bytes that have arrived but have not been read yet live in the connection, so a caller that drops the returned connection drops the buffer with it.

Reading

next/1 answers with the next message, reading from the socket only when nothing already decoded is waiting. A read frames everything that arrived, so one recv yields as many messages as the server sent - which is the whole point of minato_protocol:frame/1 - without the caller ever seeing the buffer.

Every read passes an explicit timeout. A read that times out leaves the connection with an unknown number of bytes still to come, so the connection is no longer usable and next/1 says so with {error, {socket, timeout}}; the owner closes it. Recovering the stream would mean knowing how far through a message the far end was, which is exactly what a timeout means we do not know.

TLS

ssl => true upgrades the socket before the startup message, and refuses to continue when the server answers that it will not do TLS. There is no prefer mode. A mode that continues in the clear when the server declines is a mode the man in the middle chooses, because declining is the one message an attacker sitting on the connection can always produce.

Verification is on by default: verify_peer against the OS trust store, with the hostname checked and server_name_indication set, which OTP needs to be told explicitly on a socket upgrade. Anything passed in ssl_options replaces the default for that option and nothing else, so turning verification off is possible and has to be written down.

Under TLS the SCRAM exchange is bound to the session by default: SCRAM-SHA-256-PLUS with tls-server-end-point, which puts the hash of the server's certificate inside the material the client proof covers. A man in the middle holding a certificate the client accepts can otherwise relay the whole 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.

Cleartext passwords

A server may ask for the password in the clear. minato answers only when the transport is TLS, and refuses otherwise with {error, cleartext_password_refused}. The password is the credential for every future connection too, so handing it to whoever is on the wire is worse than failing to connect. auth => [Method] narrows what is acceptable further, down to SCRAM alone.

Cancelling

cancel/1 opens a second connection and sends CancelRequest with the BackendKeyData from this one, because that is the only way the protocol offers to interrupt a running query: the connection running it is busy. Cancellation is a request. The server may finish first, and a caller still has to read whatever the query produced.

The password is not kept after the exchange that used it, so cancel/1 works without one, which is what CancelRequest documents: possession of the secret the server issued is the whole of the authorisation.

Summary

Types

An open connection. Owned by the process that called connect/1.

What went wrong. pgsql_error is the server refusing, everything else is local.

What CancelRequest needs: the process id and secret the server issued.

An authentication method the server may ask for.

How to reach a server and who to be.

The map carried by an error({minato_conn, Report}) exception.

Functions

Ask for the next bytes to arrive as a message instead of a read.

How long this connection has been open, in milliseconds.

The key cancel/1 needs, for a caller that wants to cancel from elsewhere.

Ask the server to cancel whatever this connection is running.

The tls-server-end-point binding data for this connection, or undefined.

Take the next statement name this connection has room for, or undefined.

Close a connection.

Open a connection and complete the opening exchange.

Go back to reading, after activate/1.

Give the statement that runs next this long, then cancel it.

Decode data rows as they are framed, with Decoder, or stop doing so.

Drop what was kept under this SQL.

Hand the socket to another process.

Turn a socket message into the protocol messages it carried.

next/2 with the connection's own timeout.

Read the next message the server sent.

One reported parameter, or undefined.

Every parameter the server has reported, by name.

What this connection has parsed on the server under this SQL, or undefined.

Keep a parsed statement under its SQL, if there is room.

Write one frontend message, or a list of them in order.

How many statements this connection is keeping, and how many names it may claim.

The per read timeout this connection was opened with.

What the server last said about the transaction on this connection.

The same connection with a different read timeout.

Types

conn()

-opaque conn()

An open connection. Owned by the process that called connect/1.

error()

-type error() ::
          {pgsql_error, minato_protocol:fields()} |
          {socket, term()} |
          {tls, term()} |
          tls_refused | cleartext_password_refused |
          {auth_method_refused, method()}.

What went wrong. pgsql_error is the server refusing, everything else is local.

key()

-type key() :: {Pid :: integer(), Secret :: integer()}.

What CancelRequest needs: the process id and secret the server issued.

method()

-type method() :: scram_sha_256 | md5 | cleartext | none.

An authentication method the server may ask for.

none is a server that asks for nothing, which is trust in pg_hba.conf.

opts()

-type opts() ::
          #{host => inet:socket_address() | inet:hostname(),
            port => inet:port_number(),
            user := binary(),
            password => binary() | fun(() -> binary()),
            database => binary(),
            parameters => #{binary() => binary()},
            ssl => boolean(),
            ssl_options => [ssl:tls_client_option()],
            channel_binding => minato_auth:channel_binding(),
            auth => [method()],
            connect_timeout => timeout(),
            timeout => timeout(),
            socket_options => [gen_tcp:connect_option()],
            prepared_statements => non_neg_integer(),
            cancel_timeout => timeout(),
            frame_opts => minato_protocol:frame_opts()}.

How to reach a server and who to be.

#{host => inet:socket_address() | inet:hostname(),
  port => inet:port_number(),
  user := binary(),
  password => binary() | fun(() -> binary()),
  database => binary(),
  parameters => #{binary() => binary()},
  ssl => boolean(),
  ssl_options => [ssl:tls_client_option()],
  channel_binding => minato_auth:channel_binding(),
  auth => [method()],
  connect_timeout => timeout(),
  timeout => timeout(),
  socket_options => [gen_tcp:connect_option()],
  prepared_statements => non_neg_integer(),
  cancel_timeout => timeout(),
  frame_opts => minato_protocol:frame_opts()}

user is the only one that has to be given. database defaults to user, as PostgreSQL itself does.

password may be a fun, which is called once per connection attempt and never stored. That is what a rotating credential needs: a password read at start up and held in a pool's configuration for a month is a password that expires while the pool holds it.

parameters are added to the StartupMessage and are how application_name gets set. Two are sent unless they are given here: client_encoding is UTF8, because the codecs decode text as UTF-8 and a server sending LATIN1 would put mojibake in a binary rather than raise; and DateStyle is ISO, MDY, because the text format for dates and timestamps is only unambiguous under ISO. Neither affects the binary format.

channel_binding decides whether SCRAM is tied to the TLS session. It defaults 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.

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 minato_query for what fills it.

timeout is the per read timeout and applies for the life of the connection. connect_timeout covers the TCP connect and the TLS handshake. cancel_timeout is how long a cancelled statement is given to come back before the connection is given up on; see deadline/2.

report()

-type report() :: #{message := term(), reason := term(), data := term()}.

The map carried by an error({minato_conn, Report}) exception.

Functions

activate/1

-spec activate(conn()) -> ok | {error, error()}.

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.

The connection is unusable for next/1 until the messages stop coming, and the two must not be mixed on one connection.

age/1

-spec age(conn()) -> non_neg_integer().

How long this connection has been open, in milliseconds.

For a pool deciding whether to keep lending it. A connection to PostgreSQL itself can live for months; one that goes through a proxy, a load balancer or a NAT is on a clock somebody else set, and finding out where that clock ran out by having a query fail is the expensive way.

backend_key/1

-spec backend_key(conn()) -> key() | undefined.

The key cancel/1 needs, for a caller that wants to cancel from elsewhere.

cancel/1

-spec cancel(conn()) -> ok | {error, error()}.

Ask the server to cancel whatever this connection is running.

Opens a second connection to the same server, sends CancelRequest and closes it. No answer is defined by the protocol: the server either cancels the query or does not, and this connection finds out by reading what its query returned.

channel_binding/1

-spec channel_binding(conn()) -> binary() | undefined.

The tls-server-end-point binding data for this connection, or undefined.

The hash of the server's certificate, taken with the hash the certificate was signed with, which is what RFC 5929 defines and what PostgreSQL computes on its side. SHA-1 and MD5 signatures are hashed with SHA-256 instead, because the binding must not be weaker than the mechanism it is bound to.

There is nothing to bind to on a plain socket, so it is undefined there.

claim/1

-spec claim(conn()) -> {binary() | undefined, conn()}.

Take the next statement name this connection has room for, or undefined.

Names never repeat, even after forget/2, because the statement a forgotten name refers to may still be parsed on the server, and parsing a second one under the same name is an error rather than a replacement.

close/1

-spec close(conn()) -> ok.

Close a connection.

Terminate is sent first so the server tears the backend down rather than discovering a closed socket, and a failure to send it is not reported: the socket is being closed either way.

connect/1

-spec connect(opts()) -> {ok, conn()} | {error, error()}.

Open a connection and complete the opening exchange.

Answers once the server has sent ReadyForQuery, so a connection that is returned is one that can be used, not merely one that is open.

Raises error({minato_auth, _}) for a server that asks for something it cannot be answered with, and error({minato_scram, _}) for a server that fails to prove itself. The socket is closed before either is re-raised.

{ok, Conn} = minato_conn:connect(#{user => ~"minato", password => ~"minato",
                                   database => ~"minato_test"}).

deactivate/1

-spec deactivate(conn()) -> {ok, conn()} | {error, error()}.

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.

deadline/2

-spec deadline(conn(), timeout()) -> conn().

Give the statement that runs next this long, then cancel it.

A read timeout on its own can only end the wait, not the statement: the server carries on running the query, and the connection is left with an answer coming that nobody is reading, which makes it fit for closing and nothing else. A deadline sends CancelRequest when it expires and then goes on reading, so what arrives is the server's own 57014 and the ReadyForQuery after it. A cancelled statement therefore comes back as an ordinary failed statement with the connection still usable, which is the difference between a slow query costing one query and costing a connection.

The cancelled statement gets cancel_timeout to acknowledge it. A server that does not answer even that has stopped talking, and the connection really is finished: {error, {socket, timeout}}, with no connection handed back.

It lasts until it is cleared with infinity, which the query layer does before handing a connection back, so a deadline belongs to one call rather than to the connection. It covers everything that call does, including the Describe round trip a first-time statement needs, because a caller asking for 300 milliseconds means the answer, not one leg of it.

decoding/2

-spec decoding(conn(), minato_protocol:decoder() | undefined) -> conn().

Decode data rows as they are framed, with Decoder, or stop doing so.

A result set arrives as a run of DataRow messages, and framing them into a list only for the layer above to walk that list again costs an allocation and a traversal per row. With a decoder set, a contiguous run comes back as one {rows, Rows} message with the rows already decoded.

The decoder belongs to one statement's result, so it must be cleared before the connection is used for anything else. A decoder left behind would be applied to the next statement's rows, which is wrong data rather than an error.

forget/2

-spec forget(conn(), binary()) -> conn().

Drop what was kept under this SQL.

For a statement the server has stopped accepting, which is what a schema change under a cached plan looks like.

give_away/2

-spec give_away(conn(), pid()) -> ok | {error, error()}.

Hand the socket to another process.

The connection term is data and can be sent anywhere; the socket underneath it cannot. It belongs to one process, and a process that is handed a connection without being handed the socket gets {error, {socket, einval}} on its first read. A pool checking a connection out has to call this, and so does the borrower checking it back in.

handle_message/2

-spec handle_message(conn(), term()) ->
                        {messages, [minato_protocol:backend()], conn()} | {closed, term()} | ignore.

Turn a socket message into the protocol messages it carried.

For a connection under activate/1. Anything that is not this connection's socket answers ignore, so an owner with other messages of its own can pass everything through.

{closed, Reason} is the socket going away, and there is no connection in it: the socket is gone, and what is left of the buffer belongs to a message that will never finish arriving.

next/1

-spec next(conn()) -> {ok, minato_protocol:backend(), conn()} | {error, error()}.

next/2 with the connection's own timeout.

next/2

-spec next(conn(), timeout()) -> {ok, minato_protocol:backend(), conn()} | {error, error()}.

Read the next message the server sent.

Reads from the socket only when no already decoded message is waiting, so a result set of five thousand rows costs as many reads as the network chose to split it into rather than one per row.

ParameterStatus is not handled here even though the server may send it at any time. It is a message like any other and the layer above decides what to do with it, because a connection that quietly swallowed SET TIME ZONE taking effect would be hiding the one thing the message exists to announce.

parameter/2

-spec parameter(conn(), binary()) -> binary() | undefined.

One reported parameter, or undefined.

server_version, client_encoding and integer_datetimes are always among them; which others a server volunteers is the server's choice.

parameters/1

-spec parameters(conn()) -> #{binary() => binary()}.

Every parameter the server has reported, by name.

recall/2

-spec recall(conn(), binary()) -> dynamic().

What this connection has parsed on the server under this SQL, or undefined.

The value is whatever the query layer put there, which is why it is untyped here: the connection keeps statements without knowing what a statement is.

The cache belongs to the connection because the statement does: a name is meaningful only to the session that parsed it, so a cache anywhere else would hand out names another session does not have.

remember/3

-spec remember(conn(), binary(), term()) -> conn().

Keep a parsed statement under its SQL, if there is room.

Room is claim/1's to give: a connection that has handed out every name it is allowed answers undefined there, and the caller runs the statement unnamed. An unbounded workload therefore gets the uncached path rather than an unbounded pile of parsed statements in the backend.

send/2

-spec send(conn(), minato_protocol:frontend() | [minato_protocol:frontend()]) -> ok | {error, error()}.

Write one frontend message, or a list of them in order.

A list is written as a single send, which is what the extended query protocol wants: Parse, Bind, Execute and Sync in one write is one round trip rather than four.

statements/1

-spec statements(conn()) -> {non_neg_integer(), non_neg_integer()}.

How many statements this connection is keeping, and how many names it may claim.

timeout/1

-spec timeout(conn()) -> timeout().

The per read timeout this connection was opened with.

transaction_status/1

-spec transaction_status(conn()) -> minato_protocol:transaction_status().

What the server last said about the transaction on this connection.

Taken from the status byte of every ReadyForQuery as it is read, so it is only as current as the last message the caller read. failed is a transaction in which a statement has already errored, and everything further is rejected until it is rolled back.

with_timeout(Conn, Timeout)

-spec with_timeout(conn(), timeout()) -> conn().

The same connection with a different read timeout.

For work that should be given less time than the connection's own budget - a health check that has to answer inside a probe interval - or more, for a query that is known to be slow. The change lasts as long as the connection term the caller keeps.