feat(outbox): introduce status management for event delivery outcomes and update related schema

Modified lib/service/auth/src/migrate.rs +0 −2 View lib/service/auth/src/migrate.rs

@@ -18,7 +18,5 @@

1818 ])

1919 .run(db.write())

2020 .await?;

21 // 事件 outbox 与业务表一起在启动时建好，relay 随后才能领取。

22 crate::events::outbox::sync_schema(db.write()).await?;

2321 Ok(())

2422}

Modified lib/share/events/src/outbox.rs +40 −37 View lib/share/events/src/outbox.rs

@@ -24,14 +24,14 @@

2424const SHUTDOWN_TIMEOUT: Duration = Duration::from_secs(30);

2525const POLL_INTERVAL: Duration = Duration::from_millis(500);

2626const BACKLOG_SAMPLE_INTERVAL: Duration = Duration::from_secs(10);

27/// 单条事件的重试上限。超过后丢弃并打 ERROR：否则一条毒消息会永久堵住它所在的

27/// 单条事件的重试上限。超过后标记为 discarded 并打 ERROR：否则一条毒消息会永久堵住它所在的

2828/// ordering_key（`auth:{user_id}` 之类），后面的事件永远发不出去。

2929const MAX_PUBLISH_ATTEMPTS: i32 = 50;

3030/// 已投递行的保留时长与清理间隔。outbox 只是投递队列，不承担归档职责。

31/// discarded 行保留更久，便于事后回答「哪些事件丢了、影响谁」。

3132const PURGE_INTERVAL: Duration = Duration::from_secs(600);

3233const PURGED_RETENTION_SECS: i64 = 7 * 24 * 60 * 60;

33

34const OUTBOX_ENTITY_PREFIX: &str = "models::event_outbox";

34const DISCARDED_RETENTION_SECS: i64 = 90 * 24 * 60 * 60;

3535

3636static BACKLOG_AGE: LazyLock<Gauge<f64>> = LazyLock::new(|| {

3737 global::meter("events")

@@ -68,17 +68,21 @@

6868

6969const ENQUEUE_ATTEMPTS: usize = 3;

7070

71// `published_at IS NULL` 只是滚动升级期间的兼容条件：旧 relay 领取/投递后只写

72// published_at，不会写 status；这些行必须视为已终结，否则会被重投或永久堵住 key。

7173const CLAIM: &str = "

7274WITH pending AS (

7375 SELECT candidate.id FROM event_outbox AS candidate

74 WHERE candidate.published_at IS NULL AND candidate.available_at <= CURRENT_TIMESTAMP

76 WHERE candidate.status = 0 AND candidate.published_at IS NULL

77 AND candidate.available_at <= CURRENT_TIMESTAMP

7578 AND (candidate.leased_until IS NULL OR candidate.leased_until <= CURRENT_TIMESTAMP)

7679 AND NOT EXISTS (

7780 SELECT 1 FROM event_outbox AS earlier

78 WHERE earlier.ordering_key = candidate.ordering_key AND earlier.published_at IS NULL

79 AND (earlier.occurred_at, earlier.id) < (candidate.occurred_at, candidate.id)

81 WHERE earlier.ordering_key = candidate.ordering_key

82 AND earlier.status = 0 AND earlier.published_at IS NULL

83 AND earlier.seq < candidate.seq

8084 )

81 ORDER BY candidate.occurred_at, candidate.id

85 ORDER BY candidate.available_at, candidate.seq

8286 LIMIT $1 FOR UPDATE OF candidate SKIP LOCKED

8387)

8488UPDATE event_outbox AS events

@@ -88,19 +92,6 @@

8892RETURNING events.id, events.subject, events.kind, events.envelope, events.attempts

8993";

9094

91/// 建表与索引。走 `EntityFirst` 是为了复用启动迁移的 advisory lock：

92/// 多实例同时启动时不会并发 `CREATE TABLE`。

93pub async fn sync_schema(db: &DatabaseConnection) -> Result<(), DbErr> {

94 db::migrate::EntityFirst::new(&[OUTBOX_ENTITY_PREFIX])

95 .after([

96 "CREATE INDEX IF NOT EXISTS idx_event_outbox_pending ON event_outbox (available_at, occurred_at, id) WHERE published_at IS NULL",

97 "CREATE INDEX IF NOT EXISTS idx_event_outbox_ordering ON event_outbox (ordering_key, occurred_at, id) WHERE published_at IS NULL",

98 "CREATE INDEX IF NOT EXISTS idx_event_outbox_published ON event_outbox (published_at) WHERE published_at IS NOT NULL",

99 ])

100 .run(db)

101 .await

102}

103

10495fn require_durable(kind: EventKind) -> Result<(), DbErr> {

10596 if kind.durable() {

10697 return Ok(());

@@ -330,7 +321,7 @@

330321 Ok(()) => (

331322 Statement::from_sql_and_values(

332323 DbBackend::Postgres,

333 "UPDATE event_outbox SET published_at = CURRENT_TIMESTAMP, lease_id = NULL, leased_until = NULL, last_error = NULL WHERE id = $1 AND lease_id = $2 AND published_at IS NULL",

324 "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",

334325 [row.id.into(), lease_id.into()],

335326 ),

336327 PublishOutcome::Published,

@@ -351,7 +342,7 @@

351342 (

352343 Statement::from_sql_and_values(

353344 DbBackend::Postgres,

354 "UPDATE event_outbox SET published_at = CURRENT_TIMESTAMP, lease_id = NULL, leased_until = NULL, last_error = $3 WHERE id = $1 AND lease_id = $2 AND published_at IS NULL",

345 "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",

355346 [row.id.into(), lease_id.into(), error_text.into()],

356347 ),

357348 PublishOutcome::Discarded(if permanent {

@@ -366,7 +357,7 @@

366357 (

367358 Statement::from_sql_and_values(

368359 DbBackend::Postgres,

369 "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 published_at IS NULL",

360 "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",

370361 [

371362 row.id.into(),

372363 lease_id.into(),

@@ -446,24 +437,36 @@

446437 Ok(count)

447438}

448439

449async fn purge_published(db: &DatabaseConnection) {

450 match db

451 .execute_unprepared(&format!(

452 "DELETE FROM event_outbox WHERE id IN (SELECT id FROM event_outbox WHERE published_at < CURRENT_TIMESTAMP - INTERVAL '{PURGED_RETENTION_SECS} seconds' LIMIT 10000)"

453 ))

454 .await

455 {

440async fn purge_rows(db: &DatabaseConnection, state: &'static str, sql: &str) {

441 match db.execute_unprepared(sql).await {

456442 Ok(result) if result.rows_affected() > 0 => {

457 tracing::info!(

458 rows = result.rows_affected(),

459 "purged published outbox events"

460 );

443 tracing::info!(rows = result.rows_affected(), state, "purged outbox events");

461444 }

462445 Ok(_) => {}

463 Err(error) => tracing::warn!(%error, "outbox purge failed"),

446 Err(error) => tracing::warn!(%error, state, "outbox purge failed"),

464447 }

465448}

466449

450async fn purge_finished(db: &DatabaseConnection) {

451 purge_rows(

452 db,

453 "published",

454 &format!(

455 // status 0 + published_at set 只会来自滚动升级期间的旧 relay，按 published 清掉。

456 "DELETE FROM event_outbox WHERE id IN (SELECT id FROM event_outbox WHERE published_at < CURRENT_TIMESTAMP - INTERVAL '{PURGED_RETENTION_SECS} seconds' LIMIT 10000)"

457 ),

458 )

459 .await;

460 purge_rows(

461 db,

462 "discarded",

463 &format!(

464 "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)"

465 ),

466 )

467 .await;

468}

469

467470async fn run_database(

468471 domain: &'static str,

469472 db: DatabaseConnection,

@@ -479,13 +482,13 @@

479482 }

480483 if Instant::now() >= purge_at {

481484 purge_at = Instant::now() + PURGE_INTERVAL;

482 purge_published(&db).await;

485 purge_finished(&db).await;

483486 }

484487 if Instant::now() >= sample_at {

485488 sample_at = Instant::now() + BACKLOG_SAMPLE_INTERVAL;

486489 let sample = db.query_one_raw(Statement::from_string(

487490 DbBackend::Postgres,

488 "SELECT COALESCE(EXTRACT(EPOCH FROM (CURRENT_TIMESTAMP - (SELECT available_at FROM event_outbox WHERE published_at IS NULL AND available_at <= CURRENT_TIMESTAMP ORDER BY available_at, occurred_at, id LIMIT 1))), 0)::double precision AS age",

491 "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",

489492 )).await;

490493 match sample {

491494 Ok(Some(row)) => match row.try_get::<f64>("", "age") {

Modified lib/share/events/Cargo.toml +0 −1 View lib/share/events/Cargo.toml

@@ -5,7 +5,6 @@

55

66[dependencies]

77config = { workspace = true }

8db = { workspace = true }

98envelope = { workspace = true }

109models = { workspace = true }

1110queue = { workspace = true }

Modified lib/share/models/src/event_outbox.rs +17 −0 View lib/share/models/src/event_outbox.rs

@@ -1,6 +1,18 @@

11use sea_orm::JsonValue;

22use sea_orm::entity::prelude::*;

33

4#[derive(Clone, Copy, Debug, Default, PartialEq, Eq, EnumIter, DeriveActiveEnum)]

5#[sea_orm(rs_type = "i16", db_type = "SmallInteger")]

6pub enum OutboxStatus {

7 #[default]

8 #[sea_orm(num_value = 0)]

9 Pending,

10 #[sea_orm(num_value = 1)]

11 Published,

12 #[sea_orm(num_value = 2)]

13 Discarded,

14}

15

416#[sea_orm::model]

517#[derive(Clone, Debug, PartialEq, Eq, DeriveEntityModel)]

618#[sea_orm(table_name = "event_outbox")]

@@ -12,7 +24,12 @@

1224 pub ordering_key: String,

1325 pub envelope: JsonValue,

1426 pub occurred_at: DateTimeUtc,

27 // `seq` 是 DB 拥有的 identity 列，用于单列排序与判头；不映射到实体，

28 // 默认值由 schema_extras.sql 维护。

29 #[sea_orm(default_value = 0)]

30 pub status: OutboxStatus,

1531 pub published_at: Option<DateTimeUtc>,

32 pub discarded_at: Option<DateTimeUtc>,

1633 #[sea_orm(default_expr = "Expr::current_timestamp()")]

1734 pub available_at: DateTimeUtc,

1835 pub lease_id: Option<Uuid>,

Modified lib/share/models/schema_extras.sql +54 −9 View lib/share/models/schema_extras.sql

@@ -1,9 +1,9 @@

11-- schema_extras.sql

22-- Idempotent PostgreSQL DDL for constraints that SeaORM entity-first `sync`

3-- cannot express via entity attributes. Applies after every `sync` when

4-- auto_migrate is enabled (historical convention: lib/db/schema.rs; the db

5-- crate is currently removed and must apply this file when restored).

6-- Safe to re-run on every startup; every statement is guarded with

3-- cannot express via entity attributes. Applied by models::migrate::sync_schema

4-- after each successful entity sync (fingerprint-gated), so it only runs when

5-- the entity schema or one of the two SQL files changed.

6-- Safe to re-run on every schema change; every statement is guarded with

77-- IF NOT EXISTS / DO-block checks.

88

99-- ---------------------------------------------------------------------------

@@ -250,15 +250,60 @@

250250END $$;

251251

252252-- ---------------------------------------------------------------------------

253-- event_outbox: relay claims un-published rows; the partial indexes keep the

254-- claim query and per-ordering-key head lookup cheap as the table grows.

253-- event_outbox: the relay orders per ordering_key by the DB-generated `seq`

254-- (monotonic, independent of host wall clocks) and records the delivery

255-- outcome in `status` (0 pending / 1 published / 2 discarded). `seq` is

256-- intentionally not mapped on the entity: it is DB-owned, and only pending

257-- rows need a value (already-published rows keep NULL until purged).

255258-- ---------------------------------------------------------------------------

259DO $$

260BEGIN

261 IF to_regclass('event_outbox') IS NOT NULL

262 AND NOT EXISTS (

263 SELECT 1 FROM information_schema.columns

264 WHERE table_schema = current_schema()

265 AND table_name = 'event_outbox'

266 AND column_name = 'seq'

267 )

268 THEN

269 ALTER TABLE event_outbox ADD COLUMN seq BIGINT;

270 WITH ordered AS (

271 SELECT id, row_number() OVER (ORDER BY occurred_at, id) AS n

272 FROM event_outbox WHERE status = 0

273 )

274 UPDATE event_outbox e SET seq = ordered.n FROM ordered WHERE e.id = ordered.id;

275 CREATE SEQUENCE IF NOT EXISTS event_outbox_seq;

276 PERFORM setval(

277 'event_outbox_seq',

278 GREATEST(COALESCE((SELECT max(seq) FROM event_outbox), 0) + 1, 1),

279 false

280 );

281 ALTER TABLE event_outbox ALTER COLUMN seq SET DEFAULT nextval('event_outbox_seq');

282 ALTER SEQUENCE event_outbox_seq OWNED BY event_outbox.seq;

283 END IF;

284END $$;

285

286-- The index definitions changed, so IF NOT EXISTS is not enough: drop the old

287-- column lists first.

288DROP INDEX IF EXISTS idx_event_outbox_pending;

289DROP INDEX IF EXISTS idx_event_outbox_ordering;

290DROP INDEX IF EXISTS idx_event_outbox_published;

256291CREATE INDEX IF NOT EXISTS idx_event_outbox_pending

257 ON event_outbox (available_at, occurred_at, id) WHERE published_at IS NULL;

292 ON event_outbox (available_at, seq) WHERE status = 0;

258293CREATE INDEX IF NOT EXISTS idx_event_outbox_ordering

259 ON event_outbox (ordering_key, occurred_at, id) WHERE published_at IS NULL;

294 ON event_outbox (ordering_key, seq) WHERE status = 0;

260295CREATE INDEX IF NOT EXISTS idx_event_outbox_published

261 ON event_outbox (published_at) WHERE published_at IS NOT NULL;

296 ON event_outbox (published_at) WHERE status = 1;

297CREATE INDEX IF NOT EXISTS idx_event_outbox_discarded

298 ON event_outbox (discarded_at) WHERE status = 2;

299

300DO $$ BEGIN

301 IF NOT EXISTS (SELECT 1 FROM pg_constraint WHERE conname = 'event_outbox_status_valid') THEN

302 ALTER TABLE event_outbox

303 ADD CONSTRAINT event_outbox_status_valid

304 CHECK (status IN (0, 1, 2));

305 END IF;

306END $$;

262307

263308-- ---------------------------------------------------------------------------

264309-- 2026-10-30: KenOAuth becomes the only authority for organizations,

Modified lib/share/models/schema_pre_sync.sql +11 −0 View lib/share/models/schema_pre_sync.sql

@@ -139,6 +139,17 @@

139139 END IF;

140140END $$;

141141

142-- event_outbox: `status` splits the delivery outcome from the legacy

143-- `published_at` marker. Existing rows start as terminal (1) so they are never

144-- re-delivered; only rows still pending flip to 0. New inserts default to 0.

145DO $$ BEGIN

146 IF to_regclass('event_outbox') IS NOT NULL THEN

147 ALTER TABLE event_outbox ADD COLUMN IF NOT EXISTS status SMALLINT NOT NULL DEFAULT 1;

148 UPDATE event_outbox SET status = 0 WHERE status = 1 AND published_at IS NULL;

149 ALTER TABLE event_outbox ALTER COLUMN status SET DEFAULT 0;

150 END IF;

151END $$;

152

142153-- pull_request domain (PR API): the sync layer cannot relax NOT NULL nor add

143154-- NOT NULL DEFAULT columns on populated tables, so both are applied here.

144155DO $$ BEGIN

Modified Cargo.lock +0 −1 View Cargo.lock

@@ -2433,7 +2433,6 @@

24332433 "base64 0.23.1",

24342434 "chrono",

24352435 "config 1.0.0",

2436 "db",

24372436 "envelope",

24382437 "futures-util",

24392438 "internal",