Struct Postgres 

pub struct Postgres;

Postgres: double-quoted names, $1 placeholders, rows claimed with FOR UPDATE SKIP LOCKED.

It builds the statements of the row lock form (RowLock), of the lease form (Lease), of the advisory lock form (Advisory), and the insert. A lease claim is one statement: it locks the claimable rows, writes their lease, and returns them as they were. Every name is quoted, so a name keeps its case and may hold any character; a name over 63 bytes, which Postgres would cut short without a word, is refused. A table on the database’s clock reads statement_timestamp().

In the advisory lock form a row’s key is the text concat renders from the key’s parts, a column without a value read as empty text. A session lock on the key’s 64-bit hash (hashtextextended, Postgres 11 or later) holds the row, so two keys with one hash wait for each other: a delay, never a double delivery. The claim leaves out the keys another session holds: it probes each candidate’s key with a shared lock that ends with the claim’s own transaction, in claim order and only until it has its rows. The take counts the attempt and returns the row in one statement.

In a table with FIFO groups a claim’s transaction takes its group (fifo_guard) with a lock the transaction holds on the 64-bit hash of the table’s name and the group. Two groups with one hash wait for each other: a delay, never two rows of one group in work.

A table’s transactions open with BEGIN, or at the isolation level it names (begin): READ COMMITTED, REPEATABLE READ or SERIALIZABLE. Postgres runs READ UNCOMMITTED as READ COMMITTED, so a table that names it is refused: a level the database does not keep is a level it lacks.

Examples

A subscription asks Postgres for the statements of its table, as ruststream-sqlx does when the subscription starts:

use ruststream_sqlx_dialect::{
    ClaimShape, Column, Dialect, Form, Postgres, RowLock, Statement, StatementError, TableSpec,
};

// The ledger as `#[derive(Inbox)]` describes it: one entry of an account in work at a time.
const LEDGER: TableSpec<'static> =
    TableSpec::new("ledger", Column::new("entry_id"), Form::RowLock)
        .fifo_group(Column::new("account"));

// What a claim's transaction runs on a row lock table: the guard of its group, where the
// table keeps groups in order, then the claim.
fn claim_statements(
    dialect: &impl RowLock,
    spec: &TableSpec<'_>,
) -> Result<Vec<Statement>, StatementError> {
    let mut statements: Vec<Statement> = dialect.fifo_guard(spec)?.into_iter().collect();
    statements.push(dialect.lock_claim(spec, ClaimShape::Rows)?);
    Ok(statements)
}

fn main() -> Result<(), StatementError> {
    // Postgres takes the account's group with a lock the transaction holds, then claims the
    // group's head.
    let statements = claim_statements(&Postgres, &LEDGER)?;
    assert_eq!(statements.len(), 2);
    Ok(())
}

Trait Implementations

impl Advisory for Postgres

fn advisory_claim( &self, spec: &TableSpec<'_>, ) -> Result<Statement, StatementError>

The candidates: up to Param::Limit claimable rows of the subscription’s group in claim order, two columns each, the id and the lock key as text (named __lock). It takes no lock and writes nothing; where the database can tell, it leaves out the rows whose key another session holds.

fn lock(&self) -> Option<Statement>

Tries the lock on Param::Key for the session, without waiting: one row whose first column is a 64-bit integer, nonzero when the lock was taken. None when the process keeps the locks.

fn unlock(&self) -> Option<Statement>

Releases the lock on Param::Key: one row whose first column is a 64-bit integer, nonzero when the session held it. None when the process keeps the locks.

fn take( &self, spec: &TableSpec<'_>, shape: ClaimShape, ) -> Result<Vec<Statement>, StatementError>

The statements that take a candidate whose lock the session holds: they count its attempt where the table has one and read the row again in shape, as it was before the count, only while it is still claimable. The last returns the row, or none when the row is gone or no longer claimable; a first statement, where there are two, counts the attempt and changes no row in that case.

impl Clone for Postgres

fn clone(&self) -> Self

Returns a duplicate of the value.

fn clone_from(&mut self, source: &Self)

Performs copy-assignment from source.

impl Copy for Postgres

impl Debug for Postgres

fn fmt(&self, f: &mut Formatter<'_>) -> Result

Formats the value using the given formatter.

impl Default for Postgres

fn default() -> Self

Returns the “default value” for a type.

impl Dialect for Postgres

fn name(&self) -> &'static str

The dialect’s name, for messages.

fn quote_into(&self, ident: &str, out: &mut String)

Appends ident to out, quoted as a name of this database.

fn placeholder_into(&self, index: NonZeroUsize, out: &mut String)

Appends the placeholder of parameter number index (counted from 1) to out.

fn fetch(&self, spec: &TableSpec<'_>) -> Result<Statement, StatementError>

The statement that reads the rows of claimed ids, bound as one list.

fn ack(&self, spec: &TableSpec<'_>) -> Result<Statement, StatementError>

The statement that acknowledges a row: it deletes the row, or sets processed_at when the table has that column.

fn retry( &self, spec: &TableSpec<'_>, ) -> Result<Option<Statement>, StatementError>

The statement that releases a row for another attempt at once, or None when releasing the row needs no statement.

fn retry_after(&self, spec: &TableSpec<'_>) -> Result<Statement, StatementError>

The statement that releases a row for another attempt after a delay, bound as Param::RetryAfter.

fn discard(&self, spec: &TableSpec<'_>) -> Result<Statement, StatementError>

The statement that drops a row: it deletes the row, or sets processed_at when the table has that column.

fn dead_letter_group( &self, spec: &TableSpec<'_>, ) -> Result<Statement, StatementError>

The statement that moves a row whose attempts are spent to another group, bound as Param::Destination.

fn dead_letter_table( &self, spec: &TableSpec<'_>, target: TableName<'_>, ) -> Result<Vec<Statement>, StatementError>

The statements that move a row whose attempts are spent to target, a table with the same columns; they run in one transaction.

fn insert(&self, spec: &TableSpec<'_>) -> Result<Statement, StatementError>

The statement that inserts a row: every column the database does not fill, in the order of TableSpec::columns, each bound as Param::Column.

fn begin( &self, opening: Opening, ) -> Result<Option<&'static str>, StatementError>

The statement that opens a transaction at opening in place of BEGIN, or None when BEGIN opens it.

fn fifo_guard( &self, spec: &TableSpec<'_>, ) -> Result<Option<Statement>, StatementError>

The statement a claim of a table with FIFO groups runs first in its transaction, to take the subscription’s group for it; None for a table without FIFO groups, or a dialect that takes no group.

fn server_version(&self) -> Option<&'static str>

The query that reads the server’s version as one text column, or None when the dialect’s statements run on every version of its server.

fn check_server( &self, spec: &TableSpec<'_>, version: &str, ) -> Result<(), StatementError>

Refuses a server older than the statements of spec need; version is what the server_version query returned.

fn savepoint(&self) -> &'static str

The statement that sets the savepoint after a claim, inside the claim’s transaction: what a handler writes after it can be discarded while the claim’s own work stays.

fn rollback_to_savepoint(&self) -> &'static str

The statement that discards what the transaction did after savepoint, and keeps the transaction open.

impl Eq for Postgres

impl Hash for Postgres

fn hash<__H: Hasher>(&self, state: &mut __H)

Feeds this value into the given Hasher.

fn hash_slice<H>(data: &[Self], state: &mut H)

where H: Hasher, Self: Sized,

Feeds a slice of this type into the given Hasher.

impl Lease for Postgres

fn lease_claim( &self, spec: &TableSpec<'_>, shape: ClaimShape, ) -> Result<Statement, StatementError>

The statement that claims up to Param::Limit rows of the subscription’s group, in claim order, among the rows whose lease ended by Param::LeaseNow.

fn extend(&self, spec: &TableSpec<'_>) -> Result<Statement, StatementError>

The statement that extends a delivery’s lease: it writes the new expiry (Param::Lease) while the row still holds the delivery’s token (Param::Held), so a handler that runs longer than one lease keeps its row.

fn stamp(&self, spec: &TableSpec<'_>) -> Result<Statement, StatementError>

The statement that leases one claimed row: it writes the expiry (Param::Lease) and counts the attempt, while no lease holds the row at Param::LeaseNow.

fn claim_writes_lease(&self) -> bool

Whether the lease claim writes the lease itself. A dialect whose lease claim only selects the rows answers false, and the claim’s transaction then runs stamp for each claimed row before it commits.

fn claim_counts_attempt(&self, spec: &TableSpec<'_>) -> bool

Whether the rows the lease claim of spec returns carry the attempt the claim counted. A dialect whose claim returns the rows as they were before answers false; where it answers true, a delivery reports one less than its row carries.

fn begin_lease_claim(&self) -> Option<&'static str>

The statement that opens the transaction of a claim that only selects its rows, in place of BEGIN, or None when BEGIN opens it.

impl Opens<ReadCommitted> for Postgres

impl Opens<RepeatableRead> for Postgres

impl Opens<Serializable> for Postgres

impl OutboxDialect for Postgres

fn outbox_fetch( &self, spec: &TableSpec<'_>, ) -> Result<Statement, StatementError>

The statement that reads the record Param::Id names while it is unprocessed, with every column: what the consumer of a tracked message takes into work.

fn outbox_mark(&self, spec: &TableSpec<'_>) -> Result<Statement, StatementError>

The statement that marks the record Param::Id names processed: it sets processed_at to now, or deletes the record in a table without that column.

fn outbox_recover( &self, spec: &TableSpec<'_>, ) -> Result<Statement, StatementError>

The statement that reads the unprocessed records published under the name Param::Group binds, with every column: what the republish at startup sends again.

impl PartialEq for Postgres

fn eq(&self, other: &Self) -> bool

Equality operator ==.

fn ne(&self, other: &Rhs) -> bool

Inequality operator !=.

impl RowLock for Postgres

fn lock_claim( &self, spec: &TableSpec<'_>, shape: ClaimShape, ) -> Result<Statement, StatementError>

The statement that claims up to Param::Limit rows of the subscription’s group, in claim order, and locks them until the claim’s transaction ends.

impl StructuralPartialEq for Postgres