Postgres internals

ioxide.pg is a compact Postgres v3 driver built directly on the core's RingSocket: connect, handshake, and every query run on the per-reactor ring and resume inline. It pipelines, auto-prepares, and times out - here is every class.

PgConnection · PgConnection.cs

One Postgres connection over one RingSocket. Opening it never blocks the reactor (connect and the startup handshake are ring ops). The class is the pipelining engine.

Buffers

Two native buffers, held as nint so the async flow can do address arithmetic without an unsafe context (only small parse/format helpers go unsafe). _send is a fixed 512 KB block that is never realloc'd - a pipelined send holds its address across an await, so growth would dangle; overflow throws instead. _recv starts at 64 KB and grows by doubling, reclaiming the consumed prefix first, up to PgOptions.MaxReceiveBytes (default 64 MB) - a single backend message larger than that breaks the connection.

PgConnection.cs · staging into the fixed send buffer - overflow throws, never realloc

private unsafe void WriteQueryAt(string sql)
{
    int needed = PgProtocol.QueryLength(sql);
    if (needed > _sendCapacity)
        throw new PgException($"query exceeds send buffer ({_sendCapacity} bytes)");
    if (_sendEnd + needed > _sendCapacity)
        throw new PgException("pipelined send buffer full");   // _send's address is held across awaits
    int written = PgProtocol.WriteQuery(
        new Span<byte>((void*)(_send + _sendEnd), _sendCapacity - _sendEnd), sql);
    _sendEnd += written;
}

PgConnection.cs · the recv buffer compacts before it grows, and is bounded by MaxReceiveBytes

private unsafe void EnsureRecvSpace()
{
    if (_received < _recvCapacity)
        return;

    // Full, with a partial message at the end: reclaim the consumed prefix before growing.
    if (_scan > 0)
    {
        Buffer.MemoryCopy((void*)(_recv + _scan), (void*)_recv, _recvCapacity, _received - _scan);
        _received -= _scan;
        _scan = 0;
        return;
    }

    if (_recvCapacity >= _maxBufferSize)
    {
        IsBroken = true;
        throw new PgException($"backend message exceeds {_maxBufferSize} bytes");
    }

    _recvCapacity *= 2;   // only when a single message outsizes the whole buffer
    _recv = (nint)NativeMemory.Realloc((void*)_recv, (nuint)_recvCapacity);
}

Pipelining

Each query is fire-and-stage: the message is formatted synchronously into _send, a Pending (an IValueTaskSource<PgResult>) is enqueued on _inflight, and two self-perpetuating loops are kicked if not already running. SenderLoopAsync drains [_sendOffset, _sendEnd), snapshotting _sendEnd so commands appended mid-send are picked up next cycle, and compacts the consumed prefix each cycle so the fixed buffer doesn't march to capacity under sustained load. ReaderLoopAsync reads one message at a time and routes it to _inflight.Peek(), completing and dequeuing the front waiter at ReadyForQuery. FIFO routing is correct because Postgres guarantees in-order replies on one connection. Everything is single-threaded per reactor, so the queue and the _sending/_reading flags need no locks.

PgConnection.cs · the sender drains a snapshot of _sendEnd so mid-send appends are picked up next cycle

private async Task SenderLoopAsync()
{
    try
    {
        while (_sendOffset < _sendEnd)
        {
            int end = _sendEnd;                 // snapshot; more may be appended while we send
            while (_sendOffset < end)
            {
                int n = await _socket.SendAsync(_send + _sendOffset, end - _sendOffset);
                if (n <= 0) { IsBroken = true; throw PgException.Transport("send", n); }
                _sendOffset += n;
            }
            // ...reclaim the sent prefix, slide any appended tail to the front...
        }
        _sending = false;
    }
    catch (Exception ex) { IsBroken = true; FailAll(ex); }
}

PgConnection.cs · the reader routes each message to the FIFO front, completing it at ReadyForQuery

private async Task ReaderLoopAsync()
{
    while (_inflight.Count > 0)
    {
        Message message = await ReceiveMessageAsync();
        Pending front = _inflight.Peek();

        switch (message.Tag)
        {
            case PgProtocol.RowDescription:
                if (front.OnRow != null)
                    front.Columns = PgProtocol.ParseRowDescription(Body(message));
                break;
            case PgProtocol.DataRow:
                front.Rows++;
                if (front.OnRow != null)
                    InvokeRow(front.OnRow, front.Columns, in message);
                break;
            case PgProtocol.ReadyForQuery:
                _inflight.Dequeue();
                front.Complete();               // resume the awaiter inline, in order
                break;
        }
    }
}

Simple vs extended (prepared)

QueryAsync(sql) sends a simple Query ('Q'). The parameterized overload QueryAsync(sql, args) uses the extended protocol: SubmitParamAsync auto-prepares - SQL text maps to a server statement name (s0, s1, …) cached per connection in _prepared; first use emits Parse + Bind + Execute + Sync, reuse skips straight to Bind/Execute. The trailing Sync makes each command produce exactly one ReadyForQuery, so it routes through the same FIFO machinery. The parameter span is consumed into _send before the first await, so it never crosses one. The cache is bounded (MaxPreparedStatements): past the cap, new SQL uses the unnamed statement (re-parsed each call, never accumulated) so a connection can't leak server-side statements; a cached statement invalidated server-side (SQLSTATE 0A000/26000) is evicted so the next call re-parses instead of failing forever.

PgConnection.cs · a server-invalidated statement is evicted so the next call re-parses

// In the reader, on an ErrorResponse for a prepared statement:
else if (front.Sql != null && IsStaleStatement(front.Error))
{
    _prepared.Remove(front.Sql);   // e.g. a schema change invalidated it - drop and re-Parse next time
}

private static bool IsStaleStatement(PgException error) =>
    error.SqlState is "0A000" or "26000";   // feature-not-supported / invalid-schema-name

Row results

For the default single-value path, the reader captures the first field of the first row (zero-copy via a ref struct view) into PgResult.Value. For the row-streaming overloads (an onRow handler), each DataRow is handed to the callback inline as a PgRow; and the RowDescription is parsed into PgColumn[] (names + type OIDs) only when streaming - the single-value path skips it entirely, so it stays allocation-free.

Timeouts and breakage

A server ErrorResponse throws a PgException but leaves the connection usable - the stream resyncs at the next ReadyForQuery. A transport failure (or malformed framing) sets IsBroken and the pool evicts it. Each Pending stamps EnqueuedAtMs; CheckTimeout (called by the pool's reactor-thread ticker) tears the connection down with a diagnostic PgException if its oldest in-flight command has aged past CommandTimeoutMs - closing the fd cancels the stuck ring recv, so a silent backend can't park a query forever.

PgConnection.cs · the reactor ticker checks the age of the oldest in-flight command

internal bool CheckTimeout(long nowMs, int timeoutMs, string host, ushort port)
{
    if (timeoutMs <= 0 || _inflight.Count == 0)
        return false;
    long age = nowMs - _inflight.Peek().EnqueuedAtMs;
    if (age <= timeoutMs)
        return false;

    int inflight = _inflight.Count;
    IsBroken = true;
    FailAll(new PgException(
        $"pg command timed out after {timeoutMs} ms ({inflight} in flight on {host}:{port}, oldest sent ~{age} ms ago)"));
    return true;   // the pool disposes us; closing the fd cancels the stuck recv
}

PgProtocol · PgProtocol.cs

A pure (no I/O) format/parse layer - "knows nothing about io_uring, reactors, or sockets." WriteStartup sends protocol 3.0 with user + database. WriteQuery frames a simple query. WriteExtended frames the optional Parse + Bind (text-format params and results) + Execute (all rows) + Sync; ExtendedLength sizes it up front. TryReadMessage walks the self-describing length to frame one backend message (throwing on malformed framing). ParseRowDescription turns a 'T' body into PgColumn[]; TryReadFirstField locates the first DataRow field; ReadError parses an ErrorResponse into severity + SQLSTATE + message (folding in DETAIL/HINT, often the most useful part). The SASL helpers (WriteSaslInitialResponse / WriteSaslResponse / OffersScramSha256) bridge to PgScram.

PgPool · PgPool.cs

N connections on one reactor (default 8), round-robin across them so each multiplexes many in-flight commands - the connections stay busy instead of one round-trip at a time. One pool per reactor, created from OnStart; not thread-safe by design (every member runs on the owning reactor thread).

PgScram · PgScram.cs

A correct RFC 7677 SCRAM-SHA-256 implementation (no channel binding): the n,, gs2 header, a 144-bit client nonce, PBKDF2 salted password, ClientKey/StoredKey/ClientSignature/proof, and a constant-time server-signature verify (FixedTimeEquals). Trust auth (code 0) is the other supported path; MD5, cleartext, GSS, and SASL channel binding are explicitly unsupported and fail with a clear error rather than a silent hang.

PgScram.cs · the proof is ClientKey XOR ClientSignature; the server's signature is checked in constant time

// client-final: prove knowledge of the password without ever sending it
byte[] clientKey       = HMACSHA256.HashData(_saltedPassword, "Client Key"u8.ToArray());
byte[] storedKey       = SHA256.HashData(clientKey);
byte[] clientSignature = HMACSHA256.HashData(storedKey, Encoding.UTF8.GetBytes(authMessage));

var proof = new byte[32];
for (int i = 0; i < 32; i++)
    proof[i] = (byte)(clientKey[i] ^ clientSignature[i]);

// server-final: verify the server proved it too
byte[] serverKey = HMACSHA256.HashData(_saltedPassword, "Server Key"u8.ToArray());
byte[] expected  = HMACSHA256.HashData(serverKey, Encoding.UTF8.GetBytes(_authMessage));
if (!CryptographicOperations.FixedTimeEquals(expected, Convert.FromBase64String(v)))
    throw new PgException("SCRAM: server signature verification failed");

The value types