reliar-store-postgres 0.7.0

//! `claim`: the three-statement claim body (inbox contract §3.1) — the in-flight advisory-lock
//! guard, the `INSERT … ON CONFLICT DO NOTHING` claim (with the trace columns, ADR 0042 A.2.3),
//! and (only when nothing was inserted) the state read that decides `AlreadyCompleted`/`Dead` vs.
//! `Claimed`.

use reliar_inbox::{InboxClaim, InboxMessage, InboxRecordId, InboxScope};
use sqlx::{Postgres, Transaction};

use super::error::PostgresInboxError;
use crate::connection::schema::{restore_search_path, set_search_path};

use super::PostgresInboxStore;

/// The fixed Reliar advisory-lock "class" for the inbox's in-flight guard (inbox contract §3.1):
/// `i32::from_be_bytes(*b"RELI")`, the first argument to the two-argument
/// `pg_try_advisory_xact_lock`, which PostgreSQL documents as a distinct key space from the
/// one-argument `bigint` form — so this can never collide with a key any other Reliar or host
/// code chooses in that space. Released by the caller's commit or rollback; nothing to clean up.
const ADVISORY_LOCK_CLASS: i32 = i32::from_be_bytes(*b"RELI");

/// [`reliar_inbox::InboxStore::claim`]'s body — see that trait method's rustdoc for the full
/// contract.
pub(super) async fn claim(
    store: &PostgresInboxStore,
    tx: &mut Transaction<'_, Postgres>,
    scope: &InboxScope,
    message: InboxMessage<'_>,
) -> Result<InboxClaim, PostgresInboxError> {
    let restore = if store.settings.claim_sets_search_path {
        Some(set_search_path(tx, &store.settings.schema).await?)
    } else {
        None
    };

    let result = claim_locked(tx, scope.as_str(), message).await;

    // Only restore on success, exactly like `enqueue`'s wrap (`store::enqueue`): a failed
    // statement already aborts the transaction, so issuing another on it would mask the real
    // error behind "current transaction is aborted" instead. The caller's own
    // rollback/abandonment is what actually undoes the transaction-local `search_path`.
    if result.is_ok()
        && let Some(previous) = restore
    {
        restore_search_path(tx, &previous).await?;
    }

    result
}

async fn claim_locked(
    tx: &mut Transaction<'_, Postgres>,
    scope: &str,
    message: InboxMessage<'_>,
) -> Result<InboxClaim, PostgresInboxError> {
    let message_id = message.id.as_uuid();

    // 1. the in-flight guard (inbox contract §3.1/ADR 0042 §3). `false` ⇒ `InProgress`, return
    // now, no write, tx still usable. A collision between two *concurrently claimed* keys can
    // report this spuriously — a redelivery, never a lost or doubled effect. The uuid is bound
    // directly and cast in SQL (`$3::uuid::text`) rather than allocated here with `.to_string()`.
    // `hashtext` is a built-in PostgreSQL function (a 32-bit hash of its `text` argument) used
    // here only to fold `scope || '/' || message_id` into the single `int4` the two-argument
    // `pg_try_advisory_xact_lock` takes as its key — not a cryptographic hash, and not indexed.
    let acquired = sqlx::query_scalar!(
        r#"SELECT pg_try_advisory_xact_lock($1, hashtext($2 || '/' || $3::uuid::text)) AS "acquired!""#,
        ADVISORY_LOCK_CLASS,
        scope,
        message_id,
    )
    .fetch_one(&mut **tx)
    .await?;

    if !acquired {
        return Ok(InboxClaim::InProgress);
    }

    // 2. the claim. DO NOTHING, never DO UPDATE: the AlreadyCompleted path is the hot path of a
    // redelivery storm and must not write, WAL or bloat a row it only reads. A returned row means
    // we inserted ⇒ read its own `attempts` back rather than assume 0, so this stays correct even
    // if a future migration ever gives the row a non-zero starting value.
    let id = InboxRecordId::new();

    if let Some(attempts) = insert_claim_row(tx, id, scope, message).await? {
        return Ok(InboxClaim::Claimed {
            attempt: claimed_attempt(attempts),
        });
    }

    // 3. only when 2 returned nothing: the key was committed a moment ago, so decide from its
    // state. `fetch_optional`, not `fetch_one`: a concurrent `purge` can delete the row between
    // step 2's conflict and this read's own READ COMMITTED snapshot — the advisory lock held
    // since step 1 serializes *claims* only, so a deleted-then-recreated key is a real
    // possibility here, handled below.
    let row = sqlx::query!(
        r#"SELECT id, attempts, completed_at, dead_at FROM inbox WHERE scope = $1 AND message_id = $2"#,
        scope,
        message_id,
    )
    .fetch_optional(&mut **tx)
    .await?;

    let Some(row) = row else {
        // The advisory lock held since step 1 serializes *claims* only — `fail` inserts this
        // same `(scope, message_id)` (ADR 0042 §4's `ON CONFLICT … DO UPDATE`) as a plain pool
        // statement, without ever taking it, precisely for the case where the claiming
        // transaction has rolled back. So a concurrent `purge` deleting the row this session's
        // `SELECT` just missed, followed by a concurrent `fail` recreating it before this
        // session's own re-insert runs, is a real race — not a corrupt-row scenario. Rather than
        // a plain `INSERT … DO NOTHING` that could still return `None` here, the re-insert
        // upserts and reads in one statement (ADR 0042 Amendment C.8): PostgreSQL guarantees an
        // atomic insert-or-update outcome for `ON CONFLICT DO UPDATE` with no `WHERE` clause, so
        // `fetch_one` is correct — this statement can never return zero rows.
        return upsert_claim_row(tx, InboxRecordId::new(), scope, message).await;
    };

    if let Some(completed_at) = row.completed_at {
        return Ok(InboxClaim::AlreadyCompleted { completed_at });
    }

    if let Some(dead_at) = row.dead_at {
        return Ok(InboxClaim::Dead {
            id: InboxRecordId::from_uuid(row.id),
            attempts: claimed_attempts_recorded(row.attempts),
            dead_at,
        });
    }

    Ok(InboxClaim::Claimed {
        attempt: claimed_attempt(row.attempts),
    })
}

/// `u32::try_from`/`saturating_add`, never `as` (inbox contract §3.1): an out-of-range
/// `attempts` is a corrupt row, not a panic — it saturates instead.
fn claimed_attempt(attempts: i32) -> u32 {
    u32::try_from(attempts)
        .unwrap_or(u32::MAX)
        .saturating_add(1)
}

/// Same conversion as [`claimed_attempt`], without the `+ 1`: [`reliar_inbox::InboxClaim::Dead`]
/// reports the recorded count as-is, not the next attempt ordinal.
fn claimed_attempts_recorded(attempts: i32) -> u32 {
    u32::try_from(attempts).unwrap_or(u32::MAX)
}

/// Step 2's `INSERT … ON CONFLICT DO NOTHING` — the redelivery hot path, never reused by step 3
/// (ADR 0042 Amendment C.8): a redelivery storm conflicts here and answers `AlreadyCompleted` with
/// no further write, so paying `DO NOTHING`'s conflict cost on every redelivery is the trade §3
/// makes deliberately. Returns the inserted row's `attempts` on success, `None` on a conflict.
/// `id` is client-minted (ADR 0042 A.2.1); `ON CONFLICT (scope, message_id)` infers from
/// `ix_inbox_scope_message_id`.
async fn insert_claim_row(
    tx: &mut Transaction<'_, Postgres>,
    id: InboxRecordId,
    scope: &str,
    message: InboxMessage<'_>,
) -> Result<Option<i32>, PostgresInboxError> {
    let id = id.as_uuid();
    let message_id = message.id.as_uuid();
    let message_type = message.message_type.name();
    let message_version = i32::from(message.message_type.version());
    let conversation_id = message.conversation_id.as_uuid();
    let correlation_id = message
        .correlation_id
        .map(reliar_core::CorrelationId::as_str);
    let causation_id = message.causation_id.map(|c| c.as_uuid());

    let inserted = sqlx::query_scalar!(
        r#"INSERT INTO inbox (id, scope, message_id, message_type, message_version,
                               conversation_id, correlation_id, causation_id)
           VALUES ($1, $2, $3, $4, $5, $6, $7, $8)
           ON CONFLICT (scope, message_id) DO NOTHING
           RETURNING attempts"#,
        id,
        scope,
        message_id,
        message_type,
        message_version,
        conversation_id,
        correlation_id,
        causation_id,
    )
    .fetch_optional(&mut **tx)
    .await?;

    Ok(inserted)
}

/// Step 3's fallback re-insert (ADR 0042 Amendment C.8), reached only when step 2 conflicted and
/// step 3's own `SELECT` then found the key gone — a concurrent `purge` deleted it after the
/// conflict but before this session's read. Upserts and reads in one statement, `DO UPDATE` with
/// no `WHERE` clause so PostgreSQL's documented atomic insert-or-update guarantee applies:
/// exactly one row is always returned, so `fetch_one`, never `fetch_optional`. `SET updated_at =
/// inbox.updated_at` is an identity assignment — this path transitions nothing, but `DO UPDATE`
/// requires a `SET` — so it costs a HOT update on a row already in the buffer pool, on a path that
/// has already lost a race to `purge`.
///
/// Decides the same three-way precedence step 3's `SELECT` does (`completed_at` → `dead_at` →
/// `attempts`); the identity `SET` leaves `attempts` untouched, so `Claimed { attempt: attempts +
/// 1 }` reports the same ordinal either the insert or the conflict branch produces. On the
/// conflict branch `RETURNING id` is the **existing** row's id, never the `id` this call minted —
/// `InboxClaim::Dead` is built from the returned id, not the parameter.
async fn upsert_claim_row(
    tx: &mut Transaction<'_, Postgres>,
    id: InboxRecordId,
    scope: &str,
    message: InboxMessage<'_>,
) -> Result<InboxClaim, PostgresInboxError> {
    let id = id.as_uuid();
    let message_id = message.id.as_uuid();
    let message_type = message.message_type.name();
    let message_version = i32::from(message.message_type.version());
    let conversation_id = message.conversation_id.as_uuid();
    let correlation_id = message
        .correlation_id
        .map(reliar_core::CorrelationId::as_str);
    let causation_id = message.causation_id.map(|c| c.as_uuid());

    let row = sqlx::query!(
        r#"INSERT INTO inbox (id, scope, message_id, message_type, message_version,
                               conversation_id, correlation_id, causation_id)
           VALUES ($1, $2, $3, $4, $5, $6, $7, $8)
           ON CONFLICT (scope, message_id) DO UPDATE
              SET updated_at = inbox.updated_at
           RETURNING id, attempts, completed_at, dead_at"#,
        id,
        scope,
        message_id,
        message_type,
        message_version,
        conversation_id,
        correlation_id,
        causation_id,
    )
    .fetch_one(&mut **tx)
    .await?;

    if let Some(completed_at) = row.completed_at {
        return Ok(InboxClaim::AlreadyCompleted { completed_at });
    }

    if let Some(dead_at) = row.dead_at {
        return Ok(InboxClaim::Dead {
            id: InboxRecordId::from_uuid(row.id),
            attempts: claimed_attempts_recorded(row.attempts),
            dead_at,
        });
    }

    Ok(InboxClaim::Claimed {
        attempt: claimed_attempt(row.attempts),
    })
}