feat(deletion): implement lease management for account and space deletions

Modified app/src/main.rs +42 −2 View app/src/main.rs

@@ -98,8 +98,16 @@

9898

9999 spawn_dependency_health_sampler(state.clone());

100100

101 tokio::spawn(chat::worker::run_background_tasks(deps.chat.clone()));

102 tokio::spawn(deletion::run_worker(state.clone(), deps.accounts.clone()));

101 let mut workers = tokio::task::JoinSet::new();

102 workers.spawn(chat::worker::run_background_tasks(

103 deps.chat.clone(),

104 cancel.clone(),

105 ));

106 workers.spawn(deletion::run_worker(

107 state.clone(),

108 deps.accounts.clone(),

109 cancel.clone(),

110 ));

103111

104112 let relay = events::outbox::OutboxRelay::start(

105113 Some(auth_events),

@@ -206,6 +214,7 @@

206214 for task in push_tasks {

207215 let _ = task.await;

208216 }

217 drain_workers(&mut workers, &cancel).await;

209218 conn.shutdown().await;

210219 api_result?;

211220

@@ -395,6 +404,37 @@

395404 })))

396405}

397406

407const WORKER_SHUTDOWN_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(15);

408

409/// 等待后台 worker 收尾；超时则中止，未完成的删除行由租约过期后重新领取。

410async fn drain_workers(

411 workers: &mut tokio::task::JoinSet<()>,

412 cancel: &tokio_util::sync::CancellationToken,

413) {

414 let drained = tokio::time::timeout(WORKER_SHUTDOWN_TIMEOUT, async {

415 while let Some(result) = workers.join_next().await {

416 match result {

417 Ok(()) => {

418 if !cancel.is_cancelled() {

419 tracing::warn!("a background worker stopped unexpectedly");

420 }

421 }

422 Err(error) if error.is_panic() => {

423 tracing::error!(%error, "background worker panicked");

424 }

425 Err(error) => {

426 tracing::error!(%error, "background worker failed");

427 }

428 }

429 }

430 })

431 .await;

432 if drained.is_err() {

433 tracing::warn!("background workers did not stop in time; aborting them");

434 workers.shutdown().await;

435 }

436}

437

398438const QUEUE_DEPTH_INTERVAL: std::time::Duration = std::time::Duration::from_secs(30);

399439

400440const HEALTH_SAMPLE_INTERVAL: std::time::Duration = std::time::Duration::from_secs(10);

Modified lib/service/chat/src/worker.rs +9 −2 View lib/service/chat/src/worker.rs

@@ -22,12 +22,18 @@

2222 }

2323}

2424

25pub async fn run_background_tasks(service: ChatService) {

25pub async fn run_background_tasks(

26 service: ChatService,

27 cancel: tokio_util::sync::CancellationToken,

28) {

2629 let mut interval = tokio::time::interval(Duration::from_secs(30));

2730 interval.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Delay);

2831 let mut tick: u32 = 0;

2932 loop {

30 interval.tick().await;

33 tokio::select! {

34 _ = cancel.cancelled() => break,

35 _ = interval.tick() => {}

36 }

3137 let request_id = uuid::Uuid::now_v7().to_string();

3238 let span = tracing::info_span!("chat.background", request_id = %request_id);

3339 if let Err(error) = api::middleware::with_request_id(

@@ -67,4 +73,5 @@

6773 }

6874 }

6975 }

76 tracing::info!("chat background tasks stopped");

7077}

Modified lib/service/deletion/src/lib.rs +237 −52 View lib/service/deletion/src/lib.rs

@@ -17,17 +17,62 @@

1717 user_recent_mentions, user_saved_messages,

1818};

1919use proto::KenShellClient;

20use sea_orm::sea_query::Expr;

2021use sea_orm::{

21 ActiveModelTrait, ColumnTrait, EntityTrait, PaginatorTrait, QueryFilter, QuerySelect, Set,

22 ActiveModelTrait, ColumnTrait, ConnectionTrait, DbBackend, EntityTrait, PaginatorTrait,

23 QueryFilter, QuerySelect, Set, Statement,

2224};

2325use serde::Serialize;

2426use time::OffsetDateTime;

27use tokio_util::sync::CancellationToken;

2528use tracing::Instrument;

2629use uuid::Uuid;

2730

2831const RETRY_BASE_SECS: u64 = 30;

2932const RETRY_MAX_SECS: u64 = 3600;

33/// 一次只领取一行：purge 会跨 KenShell / KenAuth 多步执行，逐行领取能让关停

34/// 在行与行之间生效，而不是把已领取的其余行晾到租约过期。

35const CLAIM_BATCH: i64 = 1;

36const CLAIM_LEASE_SECS: f64 = 600.0;

3037

38const CLAIM_SPACES: &str = "

39WITH pending AS (

40 SELECT candidate.id FROM space_deletions AS candidate

41 WHERE candidate.status = 'pending'

42 AND candidate.available_at <= CURRENT_TIMESTAMP

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

44 ORDER BY candidate.available_at

45 LIMIT $1

46 FOR UPDATE OF candidate SKIP LOCKED

47)

48UPDATE space_deletions AS d

49SET lease_id = $2,

50 leased_until = CURRENT_TIMESTAMP + make_interval(secs => $3),

51 attempts = LEAST(d.attempts, 2147483646) + 1

52FROM pending

53WHERE d.id = pending.id

54RETURNING d.id

55";

56

57const CLAIM_ACCOUNTS: &str = "

58WITH pending AS (

59 SELECT candidate.id FROM account_deletions AS candidate

60 WHERE candidate.status = 'pending'

61 AND candidate.available_at <= CURRENT_TIMESTAMP

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

63 ORDER BY candidate.available_at

64 LIMIT $1

65 FOR UPDATE OF candidate SKIP LOCKED

66)

67UPDATE account_deletions AS d

68SET lease_id = $2,

69 leased_until = CURRENT_TIMESTAMP + make_interval(secs => $3),

70 attempts = LEAST(d.attempts, 2147483646) + 1

71FROM pending

72WHERE d.id = pending.id

73RETURNING d.id

74";

75

3176#[derive(Debug, thiserror::Error)]

3277pub enum DeletionError {

3378 #[error("database: {0}")]

@@ -157,6 +202,8 @@

157202 active.available_at = Set(purge_after);

158203 active.last_error = Set(None);

159204 active.attempts = Set(0);

205 active.lease_id = Set(None);

206 active.leased_until = Set(None);

160207 active.updated_at = Set(now);

161208 active.update(db).await?

162209 }

@@ -170,6 +217,8 @@

170217 last_error: Set(None),

171218 attempts: Set(0),

172219 available_at: Set(purge_after),

220 lease_id: Set(None),

221 leased_until: Set(None),

173222 created_at: Set(now),

174223 updated_at: Set(now),

175224 }

@@ -217,52 +266,114 @@

217266 }

218267}

219268

220pub async fn run_worker(state: AppState, accounts: auth::Accounts) {

269pub async fn run_worker(state: AppState, accounts: auth::Accounts, cancel: CancellationToken) {

221270 tracing::info!("deletion worker started");

222271 let mut interval = tokio::time::interval(Duration::from_secs(30));

223272 interval.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Delay);

224273 loop {

225 interval.tick().await;

274 tokio::select! {

275 _ = cancel.cancelled() => break,

276 _ = interval.tick() => {}

277 }

226278 let request_id = Uuid::now_v7().to_string();

227279 let span = tracing::info_span!("deletion.worker", request_id = %request_id);

228280 let result = api::middleware::with_request_id(

229281 Some(&request_id),

230 purge_due(&state, &accounts).instrument(span),

282 purge_due(&state, &accounts, &cancel).instrument(span),

231283 )

232284 .await;

233285 if let Err(error) = result {

234286 tracing::warn!(%error, "deletion worker pass failed");

235287 }

236288 }

289 tracing::info!("deletion worker stopped");

237290}

238291

239async fn purge_due(state: &AppState, accounts: &auth::Accounts) -> Result<(), DeletionError> {

240 let db = state.database.read();

241 let cutoff = now();

242 let spaces = space_deletion::Entity::find()

243 .filter(space_deletion::Column::Status.eq("pending"))

244 .filter(space_deletion::Column::AvailableAt.lte(cutoff))

245 .all(db)

246 .await?;

247 for row in spaces {

248 if let Err(error) = purge_space(state, &row).await {

249 record_space_failure(state, &row, &error).await?;

292async fn purge_due(

293 state: &AppState,

294 accounts: &auth::Accounts,

295 cancel: &CancellationToken,

296) -> Result<(), DeletionError> {

297 while !cancel.is_cancelled() {

298 let lease_id = Uuid::now_v7();

299 let batch = claim_spaces(state, lease_id).await?;

300 if batch.is_empty() {

301 break;

302 }

303 for row in batch {

304 if let Err(error) = purge_space(state, &row).await {

305 record_space_failure(state, &row, &error).await?;

306 }

250307 }

251308 }

252309

253 let pending_accounts = account_deletion::Entity::find()

254 .filter(account_deletion::Column::Status.eq("pending"))

255 .filter(account_deletion::Column::AvailableAt.lte(cutoff))

256 .all(db)

257 .await?;

258 for row in pending_accounts {

259 if let Err(error) = purge_account(state, accounts, &row).await {

260 record_account_failure(state, &row, &error).await?;

310 while !cancel.is_cancelled() {

311 let lease_id = Uuid::now_v7();

312 let batch = claim_accounts(state, lease_id).await?;

313 if batch.is_empty() {

314 break;

315 }

316 for row in batch {

317 if let Err(error) = purge_account(state, accounts, &row).await {

318 record_account_failure(state, &row, &error).await?;

319 }

261320 }

262321 }

263322 Ok(())

264323}

265324

325async fn claim_spaces(

326 state: &AppState,

327 lease_id: Uuid,

328) -> Result<Vec<space_deletion::Model>, DeletionError> {

329 let db = state.database.write();

330 let claimed = db

331 .query_all_raw(Statement::from_sql_and_values(

332 DbBackend::Postgres,

333 CLAIM_SPACES,

334 [CLAIM_BATCH.into(), lease_id.into(), CLAIM_LEASE_SECS.into()],

335 ))

336 .await?;

337 let mut ids = Vec::with_capacity(claimed.len());

338 for row in claimed {

339 let id: Uuid = row.try_get("", "id")?;

340 ids.push(id);

341 }

342 if ids.is_empty() {

343 return Ok(Vec::new());

344 }

345 Ok(space_deletion::Entity::find()

346 .filter(space_deletion::Column::Id.is_in(ids))

347 .all(db)

348 .await?)

349}

350

351async fn claim_accounts(

352 state: &AppState,

353 lease_id: Uuid,

354) -> Result<Vec<account_deletion::Model>, DeletionError> {

355 let db = state.database.write();

356 let claimed = db

357 .query_all_raw(Statement::from_sql_and_values(

358 DbBackend::Postgres,

359 CLAIM_ACCOUNTS,

360 [CLAIM_BATCH.into(), lease_id.into(), CLAIM_LEASE_SECS.into()],

361 ))

362 .await?;

363 let mut ids = Vec::with_capacity(claimed.len());

364 for row in claimed {

365 let id: Uuid = row.try_get("", "id")?;

366 ids.push(id);

367 }

368 if ids.is_empty() {

369 return Ok(Vec::new());

370 }

371 Ok(account_deletion::Entity::find()

372 .filter(account_deletion::Column::Id.is_in(ids))

373 .all(db)

374 .await?)

375}

376

266377async fn purge_space(state: &AppState, row: &space_deletion::Model) -> Result<(), DeletionError> {

267378 // Never delete a space whose ownership changed after the request.

268379 let space = {

@@ -270,9 +381,7 @@

270381 spaces::Entity::find_by_id(row.space_id).one(db).await?

271382 };

272383 let Some(space) = space else {

273 space_deletion::Entity::delete_by_id(row.id)

274 .exec(state.database.write())

275 .await?;

384 finish_space_deletion(state, row).await?;

276385 return Ok(());

277386 };

278387 if space.owner_user_id != row.requested_by {

@@ -281,9 +390,7 @@

281390 active.status = Set(SpaceStatus::Active);

282391 active.archived_at = Set(None);

283392 active.update(db).await?;

284 space_deletion::Entity::delete_by_id(row.id)

285 .exec(db)

286 .await?;

393 finish_space_deletion(state, row).await?;

287394 return Ok(());

288395 }

289396

@@ -314,14 +421,33 @@

314421 archive_local_space(state, row.space_id).await?;

315422

316423 // 3. The space row and its members/roles/invites cascade away.

317 let db = state.database.write();

318 spaces::Entity::delete_by_id(row.space_id).exec(db).await?;

424 spaces::Entity::delete_by_id(row.space_id)

425 .exec(state.database.write())

426 .await?;

427

428 if finish_space_deletion(state, row).await? {

429 tracing::info!(space_id = %row.space_id, "space purged");

430 }

431 Ok(())

432}

319433

320 space_deletion::Entity::delete_by_id(row.id)

321 .exec(db)

434async fn finish_space_deletion(

435 state: &AppState,

436 row: &space_deletion::Model,

437) -> Result<bool, DeletionError> {

438 let deleted = space_deletion::Entity::delete_many()

439 .filter(space_deletion::Column::Id.eq(row.id))

440 .filter(space_deletion::Column::LeaseId.eq(row.lease_id))

441 .exec(state.database.write())

322442 .await?;

323 tracing::info!(space_id = %row.space_id, "space purged");

324 Ok(())

443 if deleted.rows_affected == 0 {

444 tracing::warn!(

445 space_id = %row.space_id,

446 "space deletion lease was reclaimed; skipping the completion write"

447 );

448 return Ok(false);

449 }

450 Ok(true)

325451}

326452

327453async fn archive_local_space(state: &AppState, space_id: Uuid) -> Result<(), DeletionError> {

@@ -535,9 +661,18 @@

535661 Ok(()) | Err(KenAuthError::NotFound) => {}

536662 Err(error) => return Err(ken_auth(error)),

537663 }

538 account_deletion::Entity::delete_by_id(row.id)

664 let deleted = account_deletion::Entity::delete_many()

665 .filter(account_deletion::Column::Id.eq(row.id))

666 .filter(account_deletion::Column::LeaseId.eq(row.lease_id))

539667 .exec(db)

540668 .await?;

669 if deleted.rows_affected == 0 {

670 tracing::warn!(

671 user_id = %row.user_id,

672 "account deletion lease was reclaimed; skipping the completion write"

673 );

674 return Ok(());

675 }

541676 tracing::info!(user_id = %row.user_id, "account purged");

542677 Ok(())

543678}

@@ -547,14 +682,39 @@

547682 row: &space_deletion::Model,

548683 error: &DeletionError,

549684) -> Result<(), DeletionError> {

550 let attempts = row.attempts.saturating_add(1);

551 let mut active: space_deletion::ActiveModel = row.clone().into();

552 active.attempts = Set(attempts);

553 active.available_at = Set(retry_at(attempts));

554 active.last_error = Set(Some(error.to_string()));

555 active.status = Set("pending".into());

556 active.updated_at = Set(now());

557 active.update(state.database.write()).await?;

685 let attempts = row.attempts.max(1);

686 let updated = space_deletion::Entity::update_many()

687 .col_expr(space_deletion::Column::Attempts, Expr::value(attempts))

688 .col_expr(

689 space_deletion::Column::AvailableAt,

690 Expr::value(retry_at(attempts)),

691 )

692 .col_expr(

693 space_deletion::Column::LastError,

694 Expr::value(Some(error.to_string())),

695 )

696 .col_expr(space_deletion::Column::Status, Expr::value("pending"))

697 .col_expr(

698 space_deletion::Column::LeaseId,

699 Expr::value(Option::<Uuid>::None),

700 )

701 .col_expr(

702 space_deletion::Column::LeasedUntil,

703 Expr::value(Option::<OffsetDateTime>::None),

704 )

705 .col_expr(space_deletion::Column::UpdatedAt, Expr::value(now()))

706 .filter(space_deletion::Column::Id.eq(row.id))

707 .filter(space_deletion::Column::LeaseId.eq(row.lease_id))

708 .exec(state.database.write())

709 .await?;

710 if updated.rows_affected == 0 {

711 tracing::warn!(

712 space_id = %row.space_id,

713 %error,

714 "space purge failed after its lease was reclaimed"

715 );

716 return Ok(());

717 }

558718 tracing::warn!(space_id = %row.space_id, %error, attempts, "space purge failed; will retry");

559719 Ok(())

560720}

@@ -564,14 +724,39 @@

564724 row: &account_deletion::Model,

565725 error: &DeletionError,

566726) -> Result<(), DeletionError> {

567 let attempts = row.attempts.saturating_add(1);

568 let mut active: account_deletion::ActiveModel = row.clone().into();

569 active.attempts = Set(attempts);

570 active.available_at = Set(retry_at(attempts));

571 active.last_error = Set(Some(error.to_string()));

572 active.status = Set("pending".into());

573 active.updated_at = Set(now());

574 active.update(state.database.write()).await?;

727 let attempts = row.attempts.max(1);

728 let updated = account_deletion::Entity::update_many()

729 .col_expr(account_deletion::Column::Attempts, Expr::value(attempts))

730 .col_expr(

731 account_deletion::Column::AvailableAt,

732 Expr::value(retry_at(attempts)),

733 )

734 .col_expr(

735 account_deletion::Column::LastError,

736 Expr::value(Some(error.to_string())),

737 )

738 .col_expr(account_deletion::Column::Status, Expr::value("pending"))

739 .col_expr(

740 account_deletion::Column::LeaseId,

741 Expr::value(Option::<Uuid>::None),

742 )

743 .col_expr(

744 account_deletion::Column::LeasedUntil,

745 Expr::value(Option::<OffsetDateTime>::None),

746 )

747 .col_expr(account_deletion::Column::UpdatedAt, Expr::value(now()))

748 .filter(account_deletion::Column::Id.eq(row.id))

749 .filter(account_deletion::Column::LeaseId.eq(row.lease_id))

750 .exec(state.database.write())

751 .await?;

752 if updated.rows_affected == 0 {

753 tracing::warn!(

754 user_id = %row.user_id,

755 %error,

756 "account purge failed after its lease was reclaimed"

757 );

758 return Ok(());

759 }

575760 tracing::warn!(user_id = %row.user_id, %error, attempts, "account purge failed; will retry");

576761 Ok(())

577762}

Modified lib/service/deletion/Cargo.toml +1 −0 View lib/service/deletion/Cargo.toml

@@ -15,5 +15,6 @@

1515thiserror = { workspace = true }

1616time = { workspace = true }

1717tokio = { workspace = true, features = ["time"] }

18tokio-util = { workspace = true }

1819tracing = { workspace = true }

1920uuid = { workspace = true }

Modified lib/share/models/src/system/account_deletion.rs +2 −0 View lib/share/models/src/system/account_deletion.rs

@@ -18,6 +18,8 @@

1818 #[sea_orm(default_value = 0)]

1919 pub attempts: i32,

2020 pub available_at: OffsetDateTime,

21 pub lease_id: Option<Uuid>,

22 pub leased_until: Option<OffsetDateTime>,

2123 pub created_at: OffsetDateTime,

2224 pub updated_at: OffsetDateTime,

2325}

Modified lib/share/models/src/system/space_deletion.rs +2 −0 View lib/share/models/src/system/space_deletion.rs

@@ -19,6 +19,8 @@

1919 #[sea_orm(default_value = 0)]

2020 pub attempts: i32,

2121 pub available_at: OffsetDateTime,

22 pub lease_id: Option<Uuid>,

23 pub leased_until: Option<OffsetDateTime>,

2224 pub created_at: OffsetDateTime,

2325 pub updated_at: OffsetDateTime,

2426}

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

@@ -1973,6 +1973,7 @@

19731973 "thiserror 2.0.21",

19741974 "time 0.3.55",

19751975 "tokio",

1976 "tokio-util",

19761977 "tracing",

19771978 "uuid",

19781979]