main
use std::sync::LazyLock;
use std::time::{Duration, Instant};
use chrono::{DateTime, Utc};
use futures_util::future;
use models::event_outbox;
use opentelemetry::metrics::{Counter, Gauge};
use opentelemetry::{KeyValue, global};
use queue::{DeliveryClass, EVENT_PREFIX, Envelope, Publisher, QueueError, SCHEMA_VERSION};
use sea_orm::ActiveValue::Set;
use sea_orm::{ConnectionTrait, DatabaseConnection, DbBackend, DbErr, EntityTrait, Statement};
use serde::Serialize;
use serde_json::Value;
use tokio::task::{JoinHandle, JoinSet};
use tokio_util::sync::CancellationToken;
use tracing::Instrument;
use uuid::Uuid;
use crate::handler::concrete_kind;
use crate::{EventHandler, EventKind, Labels, label};
const BATCH_SIZE: i64 = 16;
const PUBLISH_TIMEOUT: Duration = Duration::from_secs(10);
const SHUTDOWN_TIMEOUT: Duration = Duration::from_secs(30);
const POLL_INTERVAL: Duration = Duration::from_millis(500);
const BACKLOG_SAMPLE_INTERVAL: Duration = Duration::from_secs(10);
/// 单条事件的重试上限。超过后标记为 discarded 并打 ERROR：否则一条毒消息会永久堵住它所在的
/// ordering_key（`auth:{user_id}` 之类），后面的事件永远发不出去。
const MAX_PUBLISH_ATTEMPTS: i32 = 50;
/// 已投递行的保留时长与清理间隔。outbox 只是投递队列，不承担归档职责。
/// discarded 行保留更久，便于事后回答「哪些事件丢了、影响谁」。
const PURGE_INTERVAL: Duration = Duration::from_secs(600);
const PURGED_RETENTION_SECS: i64 = 7 * 24 * 60 * 60;
const DISCARDED_RETENTION_SECS: i64 = 90 * 24 * 60 * 60;
static BACKLOG_AGE: LazyLock<Gauge<f64>> = LazyLock::new(|| {
global::meter("events")
.f64_gauge("event.outbox.oldest.ready.age")
.with_unit("s")
.build()
});
static PUBLISHED: LazyLock<Counter<u64>> = LazyLock::new(|| {
global::meter("events")
.u64_counter("event.outbox.published")
.build()
});
static RETRIED: LazyLock<Counter<u64>> = LazyLock::new(|| {
global::meter("events")
.u64_counter("event.outbox.retried")
.build()
});
static DISCARDED: LazyLock<Counter<u64>> = LazyLock::new(|| {
global::meter("events")
.u64_counter("event.outbox.discarded")
.build()
});
static LEASE_LOST: LazyLock<Counter<u64>> = LazyLock::new(|| {
global::meter("events")
.u64_counter("event.outbox.lease_lost")
.build()
});
static ENQUEUE_FAILED: LazyLock<Counter<u64>> = LazyLock::new(|| {
global::meter("events")
.u64_counter("event.outbox.enqueue_failed")
.build()
});
const ENQUEUE_ATTEMPTS: usize = 3;
// `published_at IS NULL` 只是滚动升级期间的兼容条件：旧 relay 领取/投递后只写
// published_at，不会写 status；这些行必须视为已终结，否则会被重投或永久堵住 key。
const CLAIM: &str = "
WITH pending AS (
SELECT candidate.id FROM event_outbox AS candidate
WHERE candidate.status = 0 AND 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.status = 0 AND earlier.published_at IS NULL
AND earlier.seq < candidate.seq
)
ORDER BY candidate.available_at, candidate.seq
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.subject, events.kind, events.envelope, events.attempts
";
fn require_durable(kind: EventKind) -> Result<(), DbErr> {
if kind.durable() {
return Ok(());
}
Err(DbErr::Custom(format!(
"event `{kind}` requires an outbox-enabled business domain"
)))
}
#[tracing::instrument(skip_all, fields(kind = %kind, request_id = %envelope::context::resolve_request_id(_request_id)))]
pub async fn enqueue_at<T: Serialize>(
conn: &impl ConnectionTrait,
kind: EventKind,
labels: Labels,
payload: T,
trace_id: &str,
available_at: DateTime<Utc>,
_request_id: Option<&str>,
) -> Result<Uuid, DbErr> {
require_durable(kind)?;
let (id, model) = build_model(kind, labels, payload, trace_id, available_at)?;
event_outbox::Entity::insert(model)
.exec_without_returning(conn)
.await?;
Ok(id)
}
pub async fn enqueue_now<T: Serialize>(
conn: &impl ConnectionTrait,
kind: EventKind,
labels: Labels,
payload: T,
trace_id: &str,
request_id: Option<&str>,
) -> Result<Uuid, DbErr> {
enqueue_at(
conn,
kind,
labels,
payload,
trace_id,
Utc::now(),
request_id,
)
.await
}
#[tracing::instrument(skip_all, fields(kind = %kind))]
pub async fn emit_retry<T: Serialize>(
conn: &impl ConnectionTrait,
kind: EventKind,
labels: Labels,
payload: &T,
) -> Result<Uuid, DbErr> {
let mut attempt = 1_usize;
loop {
match crate::emit::emit(conn, kind, labels.clone(), payload).await {
Ok(id) => return Ok(id),
Err(error) if attempt >= ENQUEUE_ATTEMPTS => {
ENQUEUE_FAILED.add(1, &[KeyValue::new("domain", kind.domain())]);
return Err(error);
}
Err(error) => {
tracing::warn!(kind = %kind, attempt, %error, "retrying the outbox enqueue");
tokio::time::sleep(Duration::from_millis(50) * attempt as u32).await;
attempt += 1;
}
}
}
}
fn build_model<T: Serialize>(
kind: EventKind,
labels: Labels,
payload: T,
trace_id: &str,
available_at: DateTime<Utc>,
) -> Result<(Uuid, event_outbox::ActiveModel), DbErr> {
require_durable(kind)?;
let ordering_key = kind.ordering_key(&labels)?;
let labels = labels.without(label::ORDERING_KEY);
let mut envelope = Envelope::new(kind.as_str(), payload)
.with_delivery(DeliveryClass::BusinessEvent)
.with_source(kind.domain())
.with_labels(labels);
if !trace_id.is_empty() {
envelope = envelope.with_trace_id(trace_id);
}
let subject = envelope
.bound_subject(EVENT_PREFIX)
.map_err(|error| DbErr::Custom(format!("invalid outbox event kind: {error}")))?;
let encoded = envelope
.encode()
.map_err(|error| DbErr::Custom(format!("encoding outbox event: {error}")))?;
if encoded.len() > queue::MAX_ENVELOPE_BYTES {
return Err(DbErr::Custom(
"outbox event exceeds the size limit".to_owned(),
));
}
let stored = serde_json::from_slice(&encoded)
.map_err(|error| DbErr::Custom(format!("storing outbox event: {error}")))?;
let model = event_outbox::ActiveModel {
id: Set(envelope.id),
subject: Set(subject),
kind: Set(envelope.kind.clone()),
ordering_key: Set(ordering_key),
envelope: Set(stored),
occurred_at: Set(envelope.occurred_at),
available_at: Set(available_at),
..Default::default()
};
Ok((envelope.id, model))
}
struct PendingEvent {
id: Uuid,
subject: String,
kind: String,
envelope: Value,
attempts: i32,
}
fn retry_delay(attempts: i32) -> f64 {
f64::from(2_u32.pow(attempts.clamp(1, 8) as u32))
}
async fn publish_identity(
handler: Option<&EventHandler>,
row: &PendingEvent,
) -> Result<(), QueueError> {
let handler =
handler.ok_or_else(|| QueueError::publish("identity event handler is not configured"))?;
let request_id = row
.envelope
.get("labels")
.and_then(|labels| labels.get(label::REQUEST_ID))
.or_else(|| {
row.envelope
.get("payload")
.and_then(|payload| payload.get(label::REQUEST_ID))
})
.and_then(Value::as_str)
.unwrap_or_default();
let mut envelope: Envelope<Value> = serde_json::from_value(row.envelope.clone())?;
if envelope.id != row.id || envelope.schema_version != SCHEMA_VERSION {
return Err(QueueError::payload(
"outbox event identity or schema is invalid",
));
}
envelope.attempt = row.attempts.max(1) as u32;
handler
.publish_durable(&row.subject, &envelope, Some(request_id))
.await
}
async fn publish_app(publisher: Option<&Publisher>, row: &PendingEvent) -> Result<(), QueueError> {
let publisher =
publisher.ok_or_else(|| QueueError::publish("app event publisher is not configured"))?;
let mut envelope: Envelope<Value> = serde_json::from_value(row.envelope.clone())?;
if envelope.id != row.id
|| envelope.kind != row.kind
|| envelope.schema_version != SCHEMA_VERSION
{
return Err(QueueError::payload("outbox event identity is invalid"));
}
envelope.verify_event_subject(&row.subject)?;
envelope.attempt = row.attempts.max(1) as u32;
let kind = concrete_kind(&envelope.kind)?;
let topic = publisher.topic(kind.topic());
let body = envelope.encode()?;
let payload = serde_json::to_value(&envelope.payload).ok();
let tag = kind.tag(payload.as_ref());
publisher
.publish(
&topic,
tag,
vec![row.id.to_string()],
envelope.properties(payload.as_ref(), None),
body,
)
.await
}
enum PublishOutcome {
Published,
Retried,
Discarded(&'static str),
}
async fn publish(
db: &DatabaseConnection,
identity: Option<&EventHandler>,
app: Option<&Publisher>,
lease_id: Uuid,
row: PendingEvent,
) -> Result<(), DbErr> {
let request_id = row
.envelope
.get("labels")
.and_then(|labels| labels.get(label::REQUEST_ID))
.or_else(|| {
row.envelope
.get("payload")
.and_then(|payload| payload.get(label::REQUEST_ID))
})
.and_then(Value::as_str)
.unwrap_or_default()
.to_owned();
let span = tracing::info_span!(
"outbox.publish",
event_id = %row.id,
subject = %row.subject,
kind = %row.kind,
request_id = %request_id,
);
let result = async {
match concrete_kind(&row.kind) {
Ok(kind) if kind.identity() => publish_identity(identity, &row).await,
_ => publish_app(app, &row).await,
}
}
.instrument(span);
let result = tokio::time::timeout(PUBLISH_TIMEOUT, result)
.await
.unwrap_or_else(|_| Err(QueueError::publish("outbox publish timed out")));
let (statement, outcome) = match result {
Ok(()) => (
Statement::from_sql_and_values(
DbBackend::Postgres,
"UPDATE event_outbox SET status = 1, published_at = CURRENT_TIMESTAMP, lease_id = NULL, leased_until = NULL, last_error = NULL WHERE id = $1 AND lease_id = $2 AND status = 0",
[row.id.into(), lease_id.into()],
),
PublishOutcome::Published,
),
Err(error) => {
let permanent = !error.is_retryable();
let error_text: String = error.to_string().chars().take(2048).collect();
if permanent || row.attempts >= MAX_PUBLISH_ATTEMPTS {
tracing::error!(
event_id = %row.id,
subject = %row.subject,
kind = %row.kind,
attempt = row.attempts,
permanent,
error = %error_text,
"outbox event was discarded"
);
(
Statement::from_sql_and_values(
DbBackend::Postgres,
"UPDATE event_outbox SET status = 2, discarded_at = CURRENT_TIMESTAMP, lease_id = NULL, leased_until = NULL, last_error = $3 WHERE id = $1 AND lease_id = $2 AND status = 0",
[row.id.into(), lease_id.into(), error_text.into()],
),
PublishOutcome::Discarded(if permanent {
"permanent"
} else {
"retry_budget"
}),
)
} else {
tracing::warn!(event_id = %row.id, subject = %row.subject, kind = %row.kind, attempt = row.attempts, error = %error_text, "outbox publish will retry");
let delay = retry_delay(row.attempts) + f64::from(row.id.as_bytes()[15] % 5);
(
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, last_error = $4 WHERE id = $1 AND lease_id = $2 AND status = 0",
[
row.id.into(),
lease_id.into(),
delay.into(),
error_text.into(),
],
),
PublishOutcome::Retried,
)
}
}
};
let event_domain = concrete_kind(&row.kind)
.map(EventKind::domain)
.unwrap_or_default();
let updated = db.execute_raw(statement).await?;
if updated.rows_affected() == 0 {
LEASE_LOST.add(1, &[KeyValue::new("domain", event_domain)]);
tracing::warn!(event_id = %row.id, "outbox lease was reclaimed by another relay");
} else {
match outcome {
PublishOutcome::Published => {
PUBLISHED.add(1, &[KeyValue::new("domain", event_domain)]);
}
PublishOutcome::Retried => {
RETRIED.add(1, &[KeyValue::new("domain", event_domain)]);
}
PublishOutcome::Discarded(reason) => {
DISCARDED.add(
1,
&[
KeyValue::new("domain", event_domain),
KeyValue::new("reason", reason),
],
);
}
}
}
Ok(())
}
#[tracing::instrument(skip_all, fields(request_id = %envelope::context::resolve_request_id(_request_id)))]
async fn run_batch(
db: &DatabaseConnection,
identity: Option<&EventHandler>,
app: Option<&Publisher>,
_request_id: Option<&str>,
) -> 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")?,
subject: row.try_get("", "subject")?,
kind: row.try_get("", "kind")?,
envelope: row.try_get("", "envelope")?,
attempts: row.try_get("", "attempts")?,
});
}
let count = pending.len();
let results = future::join_all(
pending
.into_iter()
.map(|row| publish(db, identity, app, lease_id, row)),
)
.await;
for result in results {
result?;
}
Ok(count)
}
async fn purge_rows(db: &DatabaseConnection, state: &'static str, sql: &str) {
match db.execute_unprepared(sql).await {
Ok(result) if result.rows_affected() > 0 => {
tracing::info!(rows = result.rows_affected(), state, "purged outbox events");
}
Ok(_) => {}
Err(error) => tracing::warn!(%error, state, "outbox purge failed"),
}
}
async fn purge_finished(db: &DatabaseConnection) {
purge_rows(
db,
"published",
&format!(
// status 0 + published_at set 只会来自滚动升级期间的旧 relay，按 published 清掉。
"DELETE FROM event_outbox WHERE id IN (SELECT id FROM event_outbox WHERE published_at < CURRENT_TIMESTAMP - INTERVAL '{PURGED_RETENTION_SECS} seconds' LIMIT 10000)"
),
)
.await;
purge_rows(
db,
"discarded",
&format!(
"DELETE FROM event_outbox WHERE id IN (SELECT id FROM event_outbox WHERE status = 2 AND discarded_at < CURRENT_TIMESTAMP - INTERVAL '{DISCARDED_RETENTION_SECS} seconds' LIMIT 10000)"
),
)
.await;
}
async fn run_database(
domain: &'static str,
db: DatabaseConnection,
identity: Option<EventHandler>,
app: Option<Publisher>,
cancel: CancellationToken,
) {
let mut sample_at = Instant::now();
let mut purge_at = Instant::now() + PURGE_INTERVAL;
loop {
if cancel.is_cancelled() {
return;
}
if Instant::now() >= purge_at {
purge_at = Instant::now() + PURGE_INTERVAL;
purge_finished(&db).await;
}
if Instant::now() >= sample_at {
sample_at = Instant::now() + BACKLOG_SAMPLE_INTERVAL;
let sample = db.query_one_raw(Statement::from_string(
DbBackend::Postgres,
"SELECT COALESCE(EXTRACT(EPOCH FROM (CURRENT_TIMESTAMP - (SELECT available_at FROM event_outbox WHERE status = 0 AND published_at IS NULL AND available_at <= CURRENT_TIMESTAMP ORDER BY available_at, seq LIMIT 1))), 0)::double precision AS age",
)).await;
match sample {
Ok(Some(row)) => match row.try_get::<f64>("", "age") {
Ok(age) => BACKLOG_AGE.record(age.max(0.0), &[KeyValue::new("domain", domain)]),
Err(error) => {
tracing::warn!(%domain, %error, "outbox backlog sample is invalid")
}
},
Ok(None) => {}
Err(error) => tracing::warn!(%domain, %error, "outbox backlog sample failed"),
}
}
match run_batch(&db, identity.as_ref(), app.as_ref(), None).await {
Ok(count) if count > 0 => continue,
Ok(_) => {}
Err(error) => {
tracing::error!(%domain, %error, "outbox relay database operation failed")
}
}
tokio::select! {
_ = cancel.cancelled() => return,
_ = tokio::time::sleep(POLL_INTERVAL) => {}
}
}
}
pub struct OutboxRelay {
cancel: CancellationToken,
join: Option<JoinHandle<Result<(), DbErr>>>,
}
impl OutboxRelay {
pub fn start(
identity: Option<EventHandler>,
app: Option<Publisher>,
databases: Vec<(&'static str, DatabaseConnection)>,
) -> Self {
let cancel = CancellationToken::new();
let join = if identity.is_some() || app.is_some() {
let cancel = cancel.clone();
Some(tokio::spawn(async move {
let mut workers = JoinSet::new();
for (domain, db) in databases {
workers.spawn(run_database(
domain,
db,
identity.clone(),
app.clone(),
cancel.clone(),
));
}
while let Some(result) = workers.join_next().await {
result.map_err(|error| {
DbErr::Custom(format!("outbox relay task failed: {error}"))
})?;
if !cancel.is_cancelled() {
return Err(DbErr::Custom(
"outbox relay stopped unexpectedly".to_owned(),
));
}
}
Ok(())
}))
} else {
None
};
if join.is_none() {
tracing::warn!(
"event forwarding is disabled; reliable events remain pending in the database outbox"
);
}
Self { cancel, join }
}
pub async fn wait(&mut self) -> Result<(), DbErr> {
let Some(join) = self.join.as_mut() else {
return std::future::pending().await;
};
let result = join.await;
self.join.take();
result.map_err(|error| DbErr::Custom(format!("outbox relay task failed: {error}")))??;
Err(DbErr::Custom(
"outbox relay stopped unexpectedly".to_owned(),
))
}
pub async fn shutdown(mut self) -> Result<(), DbErr> {
self.cancel.cancel();
let Some(mut join) = self.join.take() else {
return Ok(());
};
match tokio::time::timeout(SHUTDOWN_TIMEOUT, &mut join).await {
Ok(result) => result
.map_err(|error| DbErr::Custom(format!("outbox relay task failed: {error}")))?,
Err(_) => {
join.abort();
let _ = join.await;
Err(DbErr::Custom(
"outbox relay shutdown timed out; pending leases will recover".to_owned(),
))
}
}
}
}
impl Drop for OutboxRelay {
fn drop(&mut self) {
self.cancel.cancel();
}
}
#[cfg(test)]
mod tests {
use crate::{Labels, label};
#[test]
fn auth_events_serialize_per_user_and_fall_back_to_request_id() {
let user = Labels::new()
.with(label::USER_ID, 42)
.with(label::REQUEST_ID, "req-1");
assert_eq!(
label::AUTH_LOGIN_CREATE.ordering_key(&user).unwrap(),
"auth:42"
);
let anonymous = Labels::new().with(label::REQUEST_ID, "req-2");
assert_eq!(
label::AUTH_LOGIN_CREATE.ordering_key(&anonymous).unwrap(),
"auth:req-2"
);
}
#[test]
fn organization_and_email_use_their_own_resource_ids() {
let org = Labels::new()
.with(label::ORGANIZATION_ID, "org-1")
.with(label::ACTOR_USER_ID, 7);
assert_eq!(
label::ORGANIZATION_LIFE_CYCLE_CREATE
.ordering_key(&org)
.unwrap(),
"organization:org-1"
);
let mail = Labels::new().with(label::REQUEST_ID, "mail-1");
assert_eq!(
label::EMAIL_DELIVERY_CREATE.ordering_key(&mail).unwrap(),
"email:mail-1"
);
}
#[test]
fn events_without_any_correlation_label_are_rejected() {
let error = label::AUTH_LOGIN_CREATE
.ordering_key(&Labels::new())
.unwrap_err();
assert!(
error
.to_string()
.contains("requires a user, organization or request id")
);
}
}