max / makenotwork

6.5 KB · 170 lines History Blame Raw

1	//! Background maintenance tasks.
2
3	use std::collections::HashSet;
4	use std::sync::Arc;
5	use std::time::Duration;
6
7	use sqlx::PgPool;
8
9	use crate::storage::S3Storage;
10
11	/// One batch of removed-image objects to delete per round. S3's batch-delete
12	/// caps at 1000; staying well under keeps each request small.
13	const PURGE_BATCH: i64 = 500;
14
15	/// Periodically delete the S3 objects backing removed images whose object has
16	/// not been purged yet.
17	///
18	/// Closes two gaps left by the inline best-effort delete in
19	/// `remove_image_handler`: images removed before purge-tracking existed (the
20	/// backlog) and removals whose inline delete failed transiently. The sweep is
21	/// convergent: every successfully deleted object's image is marked
22	/// `s3_purged_at` and never revisited, so steady-state work is zero.
23	///
24	/// Runs once at startup, then every `interval`. Cancel by aborting the task.
25	pub async fn continuously_purge_removed_images(db: PgPool, s3: Arc<S3Storage>, interval: Duration) {
26	loop {
27	match purge_removed_image_objects(&db, &s3).await {
28	Ok(0) => {}
29	Ok(n) => tracing::info!(purged = n, "reconcile: purged orphaned image objects"),
30	Err(e) => tracing::error!(error = %e, "reconcile: image purge sweep failed"),
31	}
32	tokio::time::sleep(interval).await;
33	}
34	}
35
36	/// Delete the S3 objects for all removed-but-unpurged images, in batches.
37	/// Returns the number of image objects purged. A batch that fails to make any
38	/// progress (every key errored) stops the sweep so it retries next interval
39	/// rather than spinning on the same failures.
40	async fn purge_removed_image_objects(db: &PgPool, s3: &S3Storage) -> Result<usize, String> {
41	let mut purged = 0usize;
42	loop {
43	let pending = mt_db::queries::list_images_pending_s3_purge(db, PURGE_BATCH)
44	.await
45	.map_err(\|e\| format!("db error listing images to purge: {e}"))?;
46	if pending.is_empty() {
47	break;
48	}
49
50	let keys: Vec<String> = pending.iter().map(\|p\| p.s3_key.clone()).collect();
51	let failures = s3.delete_objects(&keys).await?;
52	let failed: HashSet<&str> = failures.iter().map(\|(k, _)\| k.as_str()).collect();
53	for (key, msg) in &failures {
54	tracing::warn!(s3_key = %key, error = %msg, "reconcile: failed to delete image object");
55	}
56
57	let purged_ids: Vec<uuid::Uuid> = pending
58	.iter()
59	.filter(\|p\| !failed.contains(p.s3_key.as_str()))
60	.map(\|p\| p.id)
61	.collect();
62
63	if purged_ids.is_empty() {
64	// No progress this round. Leave the rest for the next interval.
65	break;
66	}
67
68	purged += purged_ids.len();
69	mt_db::mutations::mark_images_s3_purged(db, &purged_ids)
70	.await
71	.map_err(\|e\| format!("db error marking images purged: {e}"))?;
72
73	if pending.len() < PURGE_BATCH as usize {
74	break;
75	}
76	}
77	Ok(purged)
78	}
79
80	// Chat
81
82	/// How often expired chat messages are swept.
83	///
84	/// Sets how long an expired message lingers, which is the only thing the
85	/// interval controls: expiry is stamped on the row at insert, so nothing is
86	/// kept alive by a late sweep. Fifteen minutes keeps each delete small and
87	/// bounds the lag well under the granularity anyone perceives in a window
88	/// measured in days.
89	pub const CHAT_SWEEP_INTERVAL: Duration = Duration::from_mins(15);
90
91	/// Periodically enforce both halves of chat retention.
92	///
93	/// Age and count are separate statements because they answer different
94	/// questions: age expires a quiet room, and the count cap bounds a busy one
95	/// that would reach its age limit holding far more than its owner allowed.
96	/// Whichever bites first wins, which is what `livechat::Retention` documents.
97	///
98	/// Both run every round even when the first finds nothing. Skipping the trim
99	/// when the age sweep is empty would be wrong in exactly the case that matters:
100	/// a room busy enough to be over its cap is usually one whose messages are all
101	/// too new to have expired.
102	///
103	/// Runs once at startup, then every `interval`. Cancel by aborting the task.
104	pub async fn continuously_sweep_chat(db: PgPool, interval: Duration) {
105	loop {
106	sweep_chat_once(&db).await;
107	tokio::time::sleep(interval).await;
108	}
109	}
110
111	/// One round of both retention statements. Split out of the loop so it can be
112	/// tested without waiting an interval.
113	///
114	/// Returns `(expired, trimmed)`. Errors are logged rather than returned: a
115	/// sweep is convergent and the next round retries, so a transient failure is
116	/// not worth propagating into a supervisor restart.
117	pub async fn sweep_chat_once(db: &PgPool) -> (u64, u64) {
118	let expired = match mt_db::mutations::sweep_expired_chat_messages(db).await {
119	Ok(n) => {
120	if n > 0 {
121	tracing::info!(expired = n, "chat: swept expired messages");
122	}
123	n
124	}
125	Err(e) => {
126	tracing::error!(error = %e, "chat: expiry sweep failed");
127	0
128	}
129	};
130
131	// Runs regardless of the result above: a failed expiry sweep is no reason
132	// to let a room grow past its cap as well.
133	let trimmed = match mt_db::mutations::trim_chat_rooms_to_cap(db).await {
134	Ok(n) => {
135	if n > 0 {
136	tracing::info!(trimmed = n, "chat: trimmed rooms to their message cap");
137	}
138	n
139	}
140	Err(e) => {
141	tracing::error!(error = %e, "chat: room cap trim failed");
142	0
143	}
144	};
145
146	(expired, trimmed)
147	}
148
149	/// Periodically drop chat send-rate buckets nobody has touched recently.
150	///
151	/// Not optional and not the same job as the retention sweep. Bucket state is
152	/// keyed by (user, room), which is unbounded and attacker-influenced: anyone
153	/// who can reach a room can mint a bucket, and both units run under a hard 512M
154	/// cgroup cap where an OOM restarts the whole site rather than degrading chat.
155	///
156	/// The crate picks the interval, and the choice is a correctness one rather
157	/// than a tuning one: evicting a bucket resets its budget, so sweeping faster
158	/// than a bucket can refill would turn eviction into a way to skip the queue.
159	/// Hence `Chat::sweep_interval` rather than a number chosen here.
160	pub async fn continuously_sweep_chat_rate_limits(chat: Arc<livechat::Chat>) {
161	let interval = chat.sweep_interval();
162	loop {
163	tokio::time::sleep(interval).await;
164	let dropped = chat.sweep(std::time::Instant::now());
165	if dropped > 0 {
166	tracing::debug!(dropped, "chat: evicted idle rate-limit buckets");
167	}
168	}
169	}
170