686 lines
27 KiB
Rust
686 lines
27 KiB
Rust
//! The transactional seam's pg arm (ADR-007) — `begin_tx`, the
|
|
//! [`PgTxHandle`] (the pooled-object handle), and all eleven `*_tx`
|
|
//! methods (`pg-engine-seam-tx`).
|
|
//!
|
|
//! # No bridge, no `spawn_blocking`
|
|
//!
|
|
//! The handle holds the pooled object directly: tokio-postgres's
|
|
//! client is `Send + Sync` (POC #2 compile-probe — ADR-007's pg arm),
|
|
//! so every op is a straight `.await` through the held
|
|
//! [`deadpool_postgres::Object`], no hop cost, no thread rigging.
|
|
//! `begin_tx` checks the pooled client out, issues `BEGIN`, and hands
|
|
//! the caller the boxed handle; `commit` and the drop path conclude
|
|
//! the transaction and return (or, when the state is unknowable,
|
|
//! discard) the object — the pool accounting stays exact on every
|
|
//! arm.
|
|
//!
|
|
//! # The unknowable-state disposition
|
|
//!
|
|
//! A failed `BEGIN`/`COMMIT`/`ROLLBACK` leaves the client's session
|
|
//! state unknowable: the object is **discarded, not re-pooled** —
|
|
//! [`Object::take`](deadpool_postgres::Object::take) takes it from
|
|
//! the pool permanently (the size accounting stays exact; the next
|
|
//! checkout draws a fresh object). Same posture for an already-closed
|
|
//! client (`is_closed`).
|
|
//!
|
|
//! # Drop = rollback (ADR-021 §4)
|
|
//!
|
|
//! A handle dropped without `commit` rolls back: `Drop` schedules the
|
|
//! rollback on the construction runtime (async-native drop posture —
|
|
//! `Drop` cannot `.await`; a detached task carries the `ROLLBACK`
|
|
//! then re-pools the object, and a failed rollback discards it). The
|
|
//! detached task never races ghosts: `tokio::spawn`'s failure arm
|
|
//! (runtime shut down before the task runs — the one case the
|
|
//! rollback does not run in) drops the held object, whose connection
|
|
//! closes server-side and the server itself aborts the open
|
|
//! transaction (POC ground: no ghost survives a closed session). Drop
|
|
//! is panic-safe (`Handle::spawn` never panics — the handle is
|
|
//! captured, not current-context resolved) and never blocks.
|
|
//!
|
|
//! Ops after `commit`/drop-consumption fail closed with
|
|
//! [`Error::Closed`] (the wave-3 fail-closed guard shape) — one-shot
|
|
//! handles cannot op past their terminal transition.
|
|
//!
|
|
//! # Entry-point validation + payload encoding
|
|
//!
|
|
//! Every name-bearing method validates via core's `validation`
|
|
//! helpers (shared-namespace kinds vs consumer-local — ADR-008 §4)
|
|
//! **before any round trip**; payloads encode through core's fallible
|
|
//! [`alkstore::encode_payload`] with the typed `Codec` error `?`-ed
|
|
//! at the seam (ADR-023 §1). `notify_tx` additionally checks the
|
|
//! 8000-byte payload boundary client-side, before any round trip
|
|
//! (`PayloadTooLarge { limit: 8000 }` — ADR-016 §5); the measured
|
|
//! quantity is the serde_json serialization of the payload `Value` —
|
|
//! the same byte string ADR-020 §4 pins as the stored row bytes —
|
|
//! which is valid UTF-8 by construction.
|
|
//!
|
|
//! # The tx wakes
|
|
//!
|
|
//! The producer paths (`publish_with_key_tx` — which `publish_tx`
|
|
//! rides, `enqueue_tx`, `outbox_enqueue_tx`) issue `pg_notify` on the
|
|
//! mechanism's wake channel (the stream name / the queue name / the
|
|
//! outbox's derived backing-queue name — the mechanism-name-is-the-
|
|
//! channel realization) **inside the caller's transaction** ([`wake_tx`]).
|
|
//! NOTIFY's native transactional delivery makes the wake commit-atomic:
|
|
//! it delivers at the caller's COMMIT, and a rollback discards the row
|
|
//! *with* its wake (the tx/no-ghosts rows' pin). The scheduler's tick
|
|
//! fire path rides the same [`wake_tx`] (one coalesced wake per firing
|
|
//! schedule per tick, on the fired queue's name — the scheduler's
|
|
//! docs). Wakes are best-effort
|
|
//! (ADR-006; the engine-wide posture — a wake must never fail a write):
|
|
//! a failed statement is logged and swallowed, and since a failed
|
|
//! statement aborts the tx server-side the truth still surfaces typed
|
|
//! at COMMIT. `run_once` is a pull op, consumer-driven — no wake there.
|
|
//!
|
|
//! # Engine-side arithmetic
|
|
//!
|
|
//! The stamp/derivation arithmetic lives in [`crate::resolution`]
|
|
//! (one owner per formula, ADR-012 §2) — the plain-queue 300/3/5/none
|
|
//! set on `enqueue_tx`, the outbox's 60/5/5 set and the
|
|
//! `__alkstore_outbox:{name}` derivation on `outbox_enqueue_tx`, and
|
|
//! `resolve_enqueue_opts`' delay-over-`run_at` + relative-`expires`
|
|
//! resolution on both enqueue shapes, one clock read per enqueue
|
|
//! (the resolution's `now` is the std-time read, document there).
|
|
|
|
use deadpool_postgres::Object;
|
|
use serde_json::Value;
|
|
use tokio::runtime::Handle;
|
|
|
|
use alkstore::{
|
|
BoxedFuture, EnqueueOpts, Error, Job, Result, StreamEvent, TxHandle, validate_local_name,
|
|
validate_shared_name,
|
|
};
|
|
|
|
use crate::queue::{dead_columns, job_from_row, live_columns};
|
|
use crate::resolution::{
|
|
ResolvedEnqueue, outbox_backing_queue_name, outbox_default_stamps, plain_queue_default_stamps,
|
|
resolve_enqueue_opts, stamps_with_override,
|
|
};
|
|
use crate::schema::{QualifiedTable, tables};
|
|
use crate::seam::{pg_error, pool_error};
|
|
use crate::stream::stream_events_from_rows;
|
|
|
|
/// The notification payload boundary (the contract's pinned limit —
|
|
/// client-side checked, typed before any round trip; the predicate
|
|
/// rejects at `len >= limit`, the server's NUL-inclusive wire budget).
|
|
/// One owner of the limit: both `notify_tx` (below) and the
|
|
/// auto-commit `notify` (the notify task's) check through it.
|
|
pub(crate) const NOTIFY_PAYLOAD_LIMIT: usize = 8000;
|
|
|
|
/// The caller-held pg transaction handle: the pooled-object handle.
|
|
/// Constructed only through [`crate::PgStore`]'s `begin_tx` — never
|
|
/// held across an engine `Store` trait object (the trait is what
|
|
/// consumers see).
|
|
pub struct PgTxHandle {
|
|
object: Option<Object>,
|
|
schema: String,
|
|
runtime: Handle,
|
|
}
|
|
|
|
impl std::fmt::Debug for PgTxHandle {
|
|
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
|
|
f.debug_struct("PgTxHandle")
|
|
.field("open", &self.object.is_some())
|
|
.finish_non_exhaustive()
|
|
}
|
|
}
|
|
|
|
impl PgTxHandle {
|
|
/// `begin_tx`: pool checkout + `BEGIN`, returning the boxed
|
|
/// handle. Pool exhaustion blocks (deadpool's acquire semantics);
|
|
/// a closed pool (or store) fails its checkout. A failed `BEGIN`
|
|
/// discards the object (session state unknowable) — checkout +
|
|
/// begin fail as one typed `Database` error, no stranded slot.
|
|
pub(crate) fn begin(
|
|
pool: deadpool_postgres::Pool,
|
|
schema: String,
|
|
) -> BoxedFuture<'static, Result<Box<dyn TxHandle + Send>>> {
|
|
Box::pin(async move {
|
|
let object = pool.get().await.map_err(pool_error)?;
|
|
let client: &tokio_postgres::Client = &object;
|
|
if let Err(e) = client.batch_execute("BEGIN").await {
|
|
drop(Object::take(object));
|
|
return Err(pg_error(e));
|
|
}
|
|
Ok(Box::new(PgTxHandle {
|
|
object: Some(object),
|
|
schema,
|
|
runtime: Handle::current(),
|
|
}) as Box<dyn TxHandle + Send>)
|
|
})
|
|
}
|
|
|
|
/// The concrete-typed constructor — the tests' fault-injection
|
|
/// surface (poisoning the tx server-side needs raw in-tx SQL, and
|
|
/// the boxed trait object cannot downcast, ADR-007). Never
|
|
/// consumer API.
|
|
#[cfg(test)]
|
|
pub(crate) fn begin_concrete(
|
|
pool: deadpool_postgres::Pool,
|
|
schema: String,
|
|
) -> BoxedFuture<'static, Result<PgTxHandle>> {
|
|
Box::pin(async move {
|
|
let object = pool.get().await.map_err(pool_error)?;
|
|
let client: &tokio_postgres::Client = &object;
|
|
if let Err(e) = client.batch_execute("BEGIN").await {
|
|
drop(Object::take(object));
|
|
return Err(pg_error(e));
|
|
}
|
|
Ok(PgTxHandle {
|
|
object: Some(object),
|
|
schema,
|
|
runtime: Handle::current(),
|
|
})
|
|
})
|
|
}
|
|
|
|
/// The private op bridge: borrow the held object for one `.await`
|
|
/// op; ops after a terminal disposition fail closed with
|
|
/// [`Error::Closed`].
|
|
fn client(&self) -> Result<&tokio_postgres::Client> {
|
|
self.object
|
|
.as_ref()
|
|
.map(|o| {
|
|
let client: &tokio_postgres::Client = o;
|
|
client
|
|
})
|
|
.ok_or(Error::Closed)
|
|
}
|
|
|
|
/// The held pooled client — the fault-injection surface the
|
|
/// tests exercise (server-side tx poisoning); never consumer API.
|
|
#[cfg(test)]
|
|
pub(crate) fn pooled_client(&self) -> &tokio_postgres::Client {
|
|
self.client().expect("tests drive open handles")
|
|
}
|
|
|
|
/// Drain the held object, simulating a consumed (committed/dropped)
|
|
/// shell — the tests' entry into the fail-closed guard arm.
|
|
#[cfg(test)]
|
|
pub(crate) fn consume_shell(&mut self) {
|
|
self.object = None;
|
|
}
|
|
|
|
/// Schema-qualify a table reference for this handle's engine-owned
|
|
/// schema.
|
|
fn table(&self, name: &str) -> String {
|
|
QualifiedTable {
|
|
schema: &self.schema,
|
|
name,
|
|
}
|
|
.to_string()
|
|
}
|
|
|
|
/// The plain-queue enqueue's shared write (the `enqueue_tx` /
|
|
/// `outbox_enqueue_tx` INSERT), called with this handle's
|
|
/// qualified job table — the module-level [`enqueue_row`]'s
|
|
/// handle-side bridge.
|
|
async fn enqueue_row(
|
|
&self,
|
|
client: &tokio_postgres::Client,
|
|
queue: &str,
|
|
bytes: Vec<u8>,
|
|
opts: &EnqueueOpts,
|
|
base: crate::resolution::Stamps,
|
|
) -> Result<i64> {
|
|
enqueue_row(client, &self.schema, queue, bytes, opts, base).await
|
|
}
|
|
}
|
|
|
|
/// The enqueue INSERT — the one owner of the plain-queue row write.
|
|
/// The tx paths (`enqueue_tx` over the plain-queue defaults,
|
|
/// `outbox_enqueue_tx` over the outbox's 60/5/5 set) and the auto-commit
|
|
/// `Queue::enqueue` over the handle's `QueueOpts` ride it; one clock
|
|
/// read per enqueue (the `now` the resolution and the write share).
|
|
pub(crate) async fn enqueue_row(
|
|
client: &tokio_postgres::Client,
|
|
schema: &str,
|
|
queue: &str,
|
|
bytes: Vec<u8>,
|
|
opts: &EnqueueOpts,
|
|
base: crate::resolution::Stamps,
|
|
) -> Result<i64> {
|
|
let now = crate::resolution::now_unix();
|
|
let ResolvedEnqueue { run_at, expires_at } = resolve_enqueue_opts(now, opts);
|
|
let stamps = stamps_with_override(base, opts);
|
|
let row = client
|
|
.query_one(
|
|
&enqueue_row_sql(schema),
|
|
&[
|
|
&queue,
|
|
&bytes,
|
|
&run_at,
|
|
&opts.priority,
|
|
&stamps.max_attempts,
|
|
&expires_at,
|
|
&now,
|
|
&stamps.visibility_timeout_s,
|
|
&stamps.backoff_base_s,
|
|
&stamps.dead_letter_retention_s,
|
|
],
|
|
)
|
|
.await
|
|
.map_err(pg_error)?;
|
|
Ok(row.get(0))
|
|
}
|
|
|
|
/// The enqueue INSERT's SQL (schema-qualified job table).
|
|
fn enqueue_row_sql(schema: &str) -> String {
|
|
format!(
|
|
"INSERT INTO {} (queue, payload, run_at, priority, max_attempts,
|
|
expires_at, created_at, visibility_timeout_s,
|
|
backoff_base_s, dead_letter_retention_s)
|
|
VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10)
|
|
RETURNING id",
|
|
QualifiedTable {
|
|
schema,
|
|
name: tables::JOB
|
|
}
|
|
)
|
|
}
|
|
|
|
/// The in-tx wake statement (empty payload — `Wake { channel }` only,
|
|
/// ADR-008 §3): issued inside the caller's transaction, so NOTIFY's
|
|
/// native transactional delivery makes it commit-atomic — delivers at
|
|
/// the caller's COMMIT, discarded with the row by a rollback. Best-
|
|
/// effort at the op (the engine-wide posture — a wake must never fail
|
|
/// a write): a failed statement is logged and swallowed; because a
|
|
/// failed statement aborts the tx server-side, the truth still
|
|
/// surfaces typed at COMMIT. Shared by the tx producer paths and the
|
|
/// scheduler's tick fire path (the tick's in-tx fire wake — the same
|
|
/// commit-atomic semantics at the tick's own `in_tx` frame).
|
|
pub(crate) async fn wake_tx(client: &tokio_postgres::Client, channel: &str) {
|
|
if let Err(wake_err) = client
|
|
.query_one("SELECT pg_notify($1, '')", &[&channel])
|
|
.await
|
|
{
|
|
eprintln!(
|
|
"alkstore-postgres: tx wake failed for {channel:?} \
|
|
(best-effort: the durable row lands at the caller's commit): {wake_err}"
|
|
);
|
|
}
|
|
}
|
|
|
|
/// Encode a trait-crossing payload `Value` into the exact stored bytes
|
|
/// (the serde_json serialization — ADR-020 §4). The notify carriage
|
|
/// reuses it: the JSON serialization is valid UTF-8 text by
|
|
/// construction (ADR-016 §5).
|
|
pub(crate) fn encode_payload_bytes(payload: &Value) -> Result<Vec<u8>> {
|
|
alkstore::encode_payload(payload)
|
|
}
|
|
|
|
impl TxHandle for PgTxHandle {
|
|
fn enqueue_tx<'a>(
|
|
&'a mut self,
|
|
queue: &str,
|
|
opts: EnqueueOpts,
|
|
payload: Value,
|
|
) -> BoxedFuture<'a, Result<i64>> {
|
|
if let Err(e) = validate_shared_name(queue) {
|
|
return Box::pin(async move { Err(e) });
|
|
}
|
|
let queue = queue.to_string();
|
|
Box::pin(async move {
|
|
let client = self.client()?;
|
|
let bytes = encode_payload_bytes(&payload)?;
|
|
let id = self
|
|
.enqueue_row(client, &queue, bytes, &opts, plain_queue_default_stamps())
|
|
.await?;
|
|
// Posture-parity wake (the contract obligation is the
|
|
// stream path's): pg_notify on the queue's wake channel
|
|
// inside the caller's tx — commit-atomic like every
|
|
// in-tx notify, matching SQLite's watcher firing on the
|
|
// committed write. Best-effort (the queues row's pinned
|
|
// re-poll safety net covers a gap; a failed statement
|
|
// aborts the tx server-side, so the truth surfaces typed
|
|
// at COMMIT).
|
|
wake_tx(client, &queue).await;
|
|
Ok(id)
|
|
})
|
|
}
|
|
|
|
fn publish_tx<'a>(&'a mut self, stream: &str, payload: Value) -> BoxedFuture<'a, Result<i64>> {
|
|
let stream = stream.to_string();
|
|
Box::pin(async move { self.publish_with_key_tx(&stream, None, payload).await })
|
|
}
|
|
|
|
fn publish_with_key_tx<'a>(
|
|
&'a mut self,
|
|
stream: &str,
|
|
key: Option<String>,
|
|
payload: Value,
|
|
) -> BoxedFuture<'a, Result<i64>> {
|
|
if let Err(e) = validate_shared_name(stream) {
|
|
return Box::pin(async move { Err(e) });
|
|
}
|
|
let key = match key {
|
|
Some(k) if k.is_empty() || k.trim().is_empty() => {
|
|
return Box::pin(async move { Err(Error::InvalidName { name: k }) });
|
|
}
|
|
other => other,
|
|
};
|
|
let stream = stream.to_string();
|
|
Box::pin(async move {
|
|
let client = self.client()?;
|
|
let bytes = encode_payload_bytes(&payload)?;
|
|
let created_at = crate::resolution::now_unix();
|
|
let row = client
|
|
.query_one(
|
|
&format!(
|
|
"INSERT INTO {} (stream, \"key\", payload, created_at)
|
|
VALUES ($1, $2, $3, $4)
|
|
RETURNING id",
|
|
self.table(tables::EVENTS)
|
|
),
|
|
&[&stream, &key, &bytes, &created_at],
|
|
)
|
|
.await
|
|
.map_err(pg_error)?;
|
|
// Then wake, inside the caller's tx (the contract
|
|
// obligation — engine-postgres.md pins `pg_notify` as the
|
|
// streams wake trigger, and core-contract.md pins "events
|
|
// never require polling to become visible" with rollback
|
|
// dropping the event row *with* its wake): the stream's
|
|
// wake channel, the mechanism-name-is-the-channel
|
|
// realization — the auto-commit publish's twin.
|
|
wake_tx(client, &stream).await;
|
|
Ok(row.get(0))
|
|
})
|
|
}
|
|
|
|
fn notify_tx<'a>(&'a mut self, channel: &str, payload: Value) -> BoxedFuture<'a, Result<()>> {
|
|
if let Err(e) = validate_shared_name(channel) {
|
|
return Box::pin(async move { Err(e) });
|
|
}
|
|
let channel = channel.to_string();
|
|
Box::pin(async move {
|
|
let bytes = encode_payload_bytes(&payload)?;
|
|
// The 8000-byte boundary: client-side, typed, before any
|
|
// round trip (ADR-016 §5). The pinned quantity is the
|
|
// serde_json serialization — the same byte string the
|
|
// stored-row bytes pin; the predicate mirrors the server
|
|
// exactly (the pg payload budget is the wire's 8000 bytes
|
|
// counting the NUL terminator — probe-pinned: ≤ 7999-byte
|
|
// texts deliver, 8000 is server-rejected) so every
|
|
// too-big payload stays typed, never an opaque
|
|
// server-side reject.
|
|
if bytes.len() >= NOTIFY_PAYLOAD_LIMIT {
|
|
return Err(Error::PayloadTooLarge {
|
|
limit: NOTIFY_PAYLOAD_LIMIT,
|
|
});
|
|
}
|
|
let client = self.client()?;
|
|
let text = String::from_utf8(bytes)
|
|
.map_err(|e| Error::Codec(format!("notify payload must be utf-8 text: {e}")))?;
|
|
client
|
|
.query_one("SELECT pg_notify($1, $2)", &[&channel, &text])
|
|
.await
|
|
.map_err(pg_error)?;
|
|
Ok(())
|
|
})
|
|
}
|
|
|
|
fn save_offset_tx<'a>(
|
|
&'a mut self,
|
|
stream: &str,
|
|
consumer: &str,
|
|
offset: i64,
|
|
) -> BoxedFuture<'a, Result<()>> {
|
|
if let Err(e) = validate_shared_name(stream) {
|
|
return Box::pin(async move { Err(e) });
|
|
}
|
|
if let Err(e) = validate_local_name(consumer) {
|
|
return Box::pin(async move { Err(e) });
|
|
}
|
|
let stream = stream.to_string();
|
|
let consumer = consumer.to_string();
|
|
Box::pin(async move {
|
|
let client = self.client()?;
|
|
// Monotone upsert (one owner): a first save seeds the
|
|
// checkpoint — clamped at 0, the absent-consumer=0 rule
|
|
// (ADR-019 §1) — and a save at-or-below the stored one
|
|
// updates nothing (ADR-019 §6's silent no-op).
|
|
client
|
|
.execute(
|
|
&format!(
|
|
"INSERT INTO {} AS o (stream, consumer, \"offset\")
|
|
VALUES ($1, $2, GREATEST($3::bigint, 0))
|
|
ON CONFLICT (stream, consumer) DO UPDATE
|
|
SET \"offset\" = EXCLUDED.\"offset\"
|
|
WHERE EXCLUDED.\"offset\" > o.\"offset\"",
|
|
self.table(tables::OFFSETS)
|
|
),
|
|
&[&stream, &consumer, &offset],
|
|
)
|
|
.await
|
|
.map_err(pg_error)?;
|
|
Ok(())
|
|
})
|
|
}
|
|
|
|
fn get_job_tx<'a>(
|
|
&'a mut self,
|
|
queue: &str,
|
|
job_id: i64,
|
|
) -> BoxedFuture<'a, Result<Option<Job>>> {
|
|
if let Err(e) = validate_shared_name(queue) {
|
|
return Box::pin(async move { Err(e) });
|
|
}
|
|
let queue = queue.to_string();
|
|
Box::pin(async move {
|
|
let client = self.client()?;
|
|
// Queue-scoped (the seam's scoping rule): the id
|
|
// namespace is global across queues — the queue column
|
|
// must match the argument or the row does not surface.
|
|
// Dead rows are get-job-visible too (the auto-commit
|
|
// `get_job` shape; the diagnosis columns ride).
|
|
let row = client
|
|
.query_opt(
|
|
&format!(
|
|
"SELECT {dead} FROM {deadt} WHERE id = $1 AND queue = $2",
|
|
dead = dead_columns(),
|
|
deadt = self.table(tables::DEAD)
|
|
),
|
|
&[&job_id, &queue],
|
|
)
|
|
.await
|
|
.map_err(pg_error)?;
|
|
if let Some(row) = row {
|
|
return Ok(Some(job_from_row(&row)?));
|
|
}
|
|
let row = client
|
|
.query_opt(
|
|
&format!(
|
|
"SELECT {live} FROM {jobt} WHERE id = $1 AND queue = $2",
|
|
live = live_columns(),
|
|
jobt = self.table(tables::JOB)
|
|
),
|
|
&[&job_id, &queue],
|
|
)
|
|
.await
|
|
.map_err(pg_error)?;
|
|
match row {
|
|
Some(row) => Ok(Some(job_from_row(&row)?)),
|
|
None => Ok(None),
|
|
}
|
|
})
|
|
}
|
|
|
|
fn get_offset_tx<'a>(
|
|
&'a mut self,
|
|
stream: &str,
|
|
consumer: &str,
|
|
) -> BoxedFuture<'a, Result<i64>> {
|
|
if let Err(e) = validate_shared_name(stream) {
|
|
return Box::pin(async move { Err(e) });
|
|
}
|
|
if let Err(e) = validate_local_name(consumer) {
|
|
return Box::pin(async move { Err(e) });
|
|
}
|
|
let stream = stream.to_string();
|
|
let consumer = consumer.to_string();
|
|
Box::pin(async move {
|
|
let client = self.client()?;
|
|
let row = client
|
|
.query_opt(
|
|
&format!(
|
|
"SELECT \"offset\" FROM {} WHERE stream = $1 AND consumer = $2",
|
|
self.table(tables::OFFSETS)
|
|
),
|
|
&[&stream, &consumer],
|
|
)
|
|
.await
|
|
.map_err(pg_error)?;
|
|
Ok(row.map(|r| r.get(0)).unwrap_or(0))
|
|
})
|
|
}
|
|
|
|
fn read_since_tx<'a>(
|
|
&'a mut self,
|
|
stream: &str,
|
|
offset: i64,
|
|
limit: i64,
|
|
) -> BoxedFuture<'a, Result<Vec<StreamEvent>>> {
|
|
if let Err(e) = validate_shared_name(stream) {
|
|
return Box::pin(async move { Err(e) });
|
|
}
|
|
let stream = stream.to_string();
|
|
Box::pin(async move {
|
|
// The extent guard (ADR-023 §2): `limit <= 0` reads
|
|
// nothing — a value, never an error — at trait-impl
|
|
// entry, before any round trip.
|
|
if limit <= 0 {
|
|
return Ok(Vec::new());
|
|
}
|
|
let client = self.client()?;
|
|
let rows = client
|
|
.query(
|
|
&format!(
|
|
"SELECT id, \"key\", payload, created_at
|
|
FROM {}
|
|
WHERE stream = $1 AND id > $2
|
|
ORDER BY id ASC
|
|
LIMIT $3",
|
|
self.table(tables::EVENTS)
|
|
),
|
|
&[&stream, &offset, &limit],
|
|
)
|
|
.await
|
|
.map_err(pg_error)?;
|
|
stream_events_from_rows(&rows, stream)
|
|
})
|
|
}
|
|
|
|
fn read_from_consumer_tx<'a>(
|
|
&'a mut self,
|
|
stream: &str,
|
|
consumer: &str,
|
|
limit: i64,
|
|
) -> BoxedFuture<'a, Result<Vec<StreamEvent>>> {
|
|
if let Err(e) = validate_shared_name(stream) {
|
|
return Box::pin(async move { Err(e) });
|
|
}
|
|
if let Err(e) = validate_local_name(consumer) {
|
|
return Box::pin(async move { Err(e) });
|
|
}
|
|
let stream = stream.to_string();
|
|
let consumer = consumer.to_string();
|
|
Box::pin(async move {
|
|
if limit <= 0 {
|
|
return Ok(Vec::new());
|
|
}
|
|
let from = self.get_offset_tx(&stream, &consumer).await?;
|
|
self.read_since_tx(&stream, from, limit).await
|
|
})
|
|
}
|
|
|
|
fn outbox_enqueue_tx<'a>(
|
|
&'a mut self,
|
|
outbox: &str,
|
|
opts: EnqueueOpts,
|
|
payload: Value,
|
|
) -> BoxedFuture<'a, Result<i64>> {
|
|
if let Err(e) = validate_shared_name(outbox) {
|
|
return Box::pin(async move { Err(e) });
|
|
}
|
|
let outbox = outbox.to_string();
|
|
Box::pin(async move {
|
|
let client = self.client()?;
|
|
let bytes = encode_payload_bytes(&payload)?;
|
|
let queue = outbox_backing_queue_name(&outbox);
|
|
let id = self
|
|
.enqueue_row(client, &queue, bytes, &opts, outbox_default_stamps())
|
|
.await?;
|
|
// The outbox's posture-parity twin of `enqueue_tx`'s wake:
|
|
// the derived backing-queue name is the channel. No other
|
|
// path targets it — `schedule()` rejects reserved names, so
|
|
// the outbox's backing queue stays out of the scheduler's
|
|
// reach (ADR-021 §3).
|
|
wake_tx(client, &queue).await;
|
|
Ok(id)
|
|
})
|
|
}
|
|
|
|
/// Commit: `COMMIT` + re-pool. A failed COMMIT discards the object
|
|
/// — its state is unknowable (the wave-3 review's finding (c),
|
|
/// pre-empted here; the pool accounting stays exact either way).
|
|
/// One-shot; later ops on the consumed shell fail `Closed`.
|
|
fn commit(mut self: Box<Self>) -> BoxedFuture<'static, Result<()>> {
|
|
let Some(object) = self.object.take() else {
|
|
return Box::pin(async { Err(Error::Closed) });
|
|
};
|
|
Box::pin(async move {
|
|
let client: &tokio_postgres::Client = &object;
|
|
match client.batch_execute("COMMIT").await {
|
|
Ok(()) => {
|
|
drop(object);
|
|
Ok(())
|
|
}
|
|
Err(e) => {
|
|
drop(Object::take(object));
|
|
Err(pg_error(e))
|
|
}
|
|
}
|
|
})
|
|
}
|
|
}
|
|
|
|
impl Drop for PgTxHandle {
|
|
/// Drop = rollback (ADR-021 §4): the rollback runs on the
|
|
/// construction runtime as a detached task — `ROLLBACK`, then
|
|
/// re-pool; a failed rollback (or an already-closed client)
|
|
/// discards the object (unknowable state). Panic-safe: no panic
|
|
/// escapes drop (`Handle::spawn` is captured-handle-bound, never
|
|
/// current-context-bound), never blocking; the dropped-future arm
|
|
/// (runtime shut down before the task runs) closes the session —
|
|
/// the server aborts the open transaction, so no ghost survives
|
|
/// even there.
|
|
fn drop(&mut self) {
|
|
let Some(object) = self.object.take() else {
|
|
return;
|
|
};
|
|
self.runtime.spawn(async move {
|
|
if object.is_closed() {
|
|
drop(Object::take(object));
|
|
return;
|
|
}
|
|
let client: &tokio_postgres::Client = &object;
|
|
if let Err(e) = client.batch_execute("ROLLBACK").await {
|
|
drop(Object::take(object));
|
|
eprintln!(
|
|
"alkstore: pg tx handle dropped with a failed ROLLBACK; \
|
|
the pooled client is discarded (state unknowable): {e}"
|
|
);
|
|
return;
|
|
}
|
|
drop(object);
|
|
});
|
|
}
|
|
}
|
|
|
|
// The row decodes ride one owner each: the stream-page decode
|
|
// (`stream_events_from_rows`) lives in `stream.rs`, the job-row
|
|
// decode (`job_from_row` + the select column lists) in `queue.rs` —
|
|
// the one decode owners the tx reads and the auto-commit reads share
|
|
// (ADR-012 §2's one-owner rule).
|