27fffe3
use std::time::Duration;
use sea_orm::{ConnectionTrait, DatabaseConnection, DbBackend, DbErr, Statement};
use serde_json::Value;
use tokio_util::sync::CancellationToken;
use uuid::Uuid;
use crate::envelope::Envelope;
use crate::error::QueueError;
use crate::publisher::Publisher;
use crate::route;
const BATCH_SIZE: i64 = 16;
const POLL_INTERVAL: Duration = Duration::from_millis(500);
const PUBLISH_TIMEOUT: Duration = Duration::from_secs(10);
const MAX_PUBLISH_ATTEMPTS: i32 = 50;
const PURGE_INTERVAL: Duration = Duration::from_secs(600);
const PURGED_RETENTION_SECS: i64 = 7 * 24 * 60 * 60;
const CLAIM: &str = "
WITH pending AS (
SELECT candidate.id FROM event_outbox AS candidate
WHERE candidate.published_at IS NULL AND candidate.available_at <= CURRENT_TIMESTAMP
AND (candidate.leased_until IS NULL OR candidate.leased_until <= CURRENT_TIMESTAMP)
AND NOT EXISTS (
SELECT 1 FROM event_outbox AS earlier
WHERE earlier.ordering_key = candidate.ordering_key AND earlier.published_at IS NULL
AND (earlier.occurred_at, earlier.id) < (candidate.occurred_at, candidate.id)
)
ORDER BY candidate.occurred_at, candidate.id
LIMIT $1 FOR UPDATE OF candidate SKIP LOCKED
)
UPDATE event_outbox AS events
SET lease_id = $2, leased_until = CURRENT_TIMESTAMP + INTERVAL '60 seconds',
attempts = LEAST(events.attempts, 2147483646) + 1
FROM pending WHERE events.id = pending.id
RETURNING events.id, events.kind, events.envelope, events.attempts
";
struct PendingEvent {
id: Uuid,
kind: String,
envelope: Value,
attempts: i32,
}
pub async fn run(db: DatabaseConnection, publisher: Publisher, cancel: CancellationToken) {
let mut purge_at = tokio::time::Instant::now() + PURGE_INTERVAL;
loop {
if cancel.is_cancelled() {
return;
}
if tokio::time::Instant::now() >= purge_at {
purge_at = tokio::time::Instant::now() + PURGE_INTERVAL;
purge_published(&db).await;
}
match run_batch(&db, &publisher).await {
Ok(count) if count > 0 => continue,
Ok(_) => {}
Err(error) => tracing::error!(%error, "event outbox relay failed"),
}
tokio::select! {
_ = cancel.cancelled() => return,
_ = tokio::time::sleep(POLL_INTERVAL) => {}
}
}
}
async fn run_batch(db: &DatabaseConnection, publisher: &Publisher) -> Result<usize, DbErr> {
let lease_id = Uuid::now_v7();
let claimed = db
.query_all_raw(Statement::from_sql_and_values(
DbBackend::Postgres,
CLAIM,
[BATCH_SIZE.into(), lease_id.into()],
))
.await?;
let mut pending = Vec::with_capacity(claimed.len());
for row in claimed {
pending.push(PendingEvent {
id: row.try_get("", "id")?,
kind: row.try_get("", "kind")?,
envelope: row.try_get("", "envelope")?,
attempts: row.try_get("", "attempts")?,
});
}
let count = pending.len();
for row in pending {
publish(db, publisher, lease_id, row).await?;
}
Ok(count)
}
async fn publish(
db: &DatabaseConnection,
publisher: &Publisher,
lease_id: Uuid,
row: PendingEvent,
) -> Result<(), DbErr> {
let result = tokio::time::timeout(PUBLISH_TIMEOUT, publish_body(publisher, &row))
.await
.unwrap_or_else(|_| Err(QueueError::publish("outbox publish timed out")));
let statement = match result {
Ok(()) => Statement::from_sql_and_values(
DbBackend::Postgres,
"UPDATE event_outbox SET published_at = CURRENT_TIMESTAMP, lease_id = NULL, leased_until = NULL WHERE id = $1 AND lease_id = $2 AND published_at IS NULL",
[row.id.into(), lease_id.into()],
),
Err(error) => {
if row.attempts >= MAX_PUBLISH_ATTEMPTS {
tracing::error!(
event_id = %row.id,
kind = %row.kind,
attempt = row.attempts,
%error,
"outbox event exceeded the retry budget and was discarded"
);
Statement::from_sql_and_values(
DbBackend::Postgres,
"UPDATE event_outbox SET published_at = CURRENT_TIMESTAMP, lease_id = NULL, leased_until = NULL WHERE id = $1 AND lease_id = $2 AND published_at IS NULL",
[row.id.into(), lease_id.into()],
)
} else {
tracing::warn!(
event_id = %row.id,
kind = %row.kind,
attempt = row.attempts,
%error,
"outbox publish will retry"
);
let delay = retry_delay(row.attempts);
Statement::from_sql_and_values(
DbBackend::Postgres,
"UPDATE event_outbox SET available_at = CURRENT_TIMESTAMP + ($3 * INTERVAL '1 second'), lease_id = NULL, leased_until = NULL WHERE id = $1 AND lease_id = $2 AND published_at IS NULL",
[row.id.into(), lease_id.into(), delay.into()],
)
}
}
};
let updated = db.execute_raw(statement).await?;
if updated.rows_affected() == 0 {
tracing::warn!(event_id = %row.id, "outbox lease was reclaimed by another relay");
}
Ok(())
}
async fn publish_body(publisher: &Publisher, row: &PendingEvent) -> Result<(), QueueError> {
let envelope: Envelope<Value> = serde_json::from_value(row.envelope.clone())?;
if envelope.id != row.id || envelope.kind != row.kind {
return Err(QueueError::payload("outbox event identity is invalid"));
}
let route = route::route(&envelope.kind)?;
let topic = publisher.topic(&route.area);
let body = envelope.summary_body(&route.tag)?;
let properties = envelope.properties();
publisher
.publish(
&topic,
&route.tag,
vec![row.id.to_string()],
properties,
body,
)
.await
}
fn retry_delay(attempts: i32) -> f64 {
f64::from(2_u32.pow(attempts.clamp(1, 8) as u32))
}
async fn purge_published(db: &DatabaseConnection) {
match db
.execute_unprepared(&format!(
"DELETE FROM event_outbox WHERE published_at < CURRENT_TIMESTAMP - INTERVAL '{PURGED_RETENTION_SECS} seconds'"
))
.await
{
Ok(result) if result.rows_affected() > 0 => {
tracing::info!(
rows = result.rows_affected(),
"purged published outbox events"
);
}
Ok(_) => {}
Err(error) => tracing::warn!(%error, "outbox purge failed"),
}
}