Skip to main content

max / makenotwork

mt: supervise background sweeps + exclude mod-removed posts from profile counts (ultra-fuzz P-sup, B1/B2) P-sup: the session-expiry and image-purge sweeps were bare spawns, so a panic inside either silently killed the worker for the process lifetime (S3 orphans pile up / session table grows) while the server kept serving. Add a supervise() helper that runs each loop in its own joinable task, catches a panic as a JoinError, and restarts after a 5s backoff. B1/B2: profile post/endorsement counts, the membership summary, and the activity feed filtered only deleted_at (dead for posts since m007), not removed_at (the live mod-removal column), so mod-removed posts still inflated tallies and showed in activity. Add AND p.removed_at IS NULL to all four subqueries; fix the stale create_post comment. New profile_counts_exclude_mod_removed_posts test; 248 integration + 135 lib unit green; clippy clean. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
Author: Max Johnson <me@maxj.phd> · 2026-06-19 02:45 UTC
Commit: 9f0d49753ae2686f04e4d7350eb665763a1514f9
Parent: 7f0555e
4 files changed, +105 insertions, -11 deletions
M multithreaded/crates/mt-db/src/mutations.rs +3 -2
@@ -208,8 +208,9 @@ pub async fn create_thread_with_op(
208 208 /// Insert a reply and bump the thread's last_activity_at atomically.
209 209 ///
210 210 /// Opening posts are created with the thread via [`create_thread_with_op`];
211 - /// this is for replies only. Reply counts are computed live from non-removed
212 - /// posts at read time (no denormalized counter to maintain).
211 + /// this is for replies only. The display reply count derives from the
212 + /// denormalized `threads.post_count`, which the m029 trigger maintains on every
213 + /// post INSERT/DELETE/removed_at change — no counter to update by hand here.
213 214 #[tracing::instrument(skip_all)]
214 215 pub async fn create_post(
215 216 pool: &PgPool,
M multithreaded/crates/mt-db/src/queries.rs +4
@@ -884,6 +884,7 @@ pub async fn get_user_profile_in_community(
884 884 JOIN categories c ON c.id = t.category_id
885 885 WHERE p.author_id = u.mnw_account_id
886 886 AND c.community_id = co.id
887 + AND p.removed_at IS NULL
887 888 AND p.deleted_at IS NULL
888 889 AND t.deleted_at IS NULL) AS post_count,
889 890 (SELECT COUNT(*) FROM post_endorsements pe
@@ -892,6 +893,7 @@ pub async fn get_user_profile_in_community(
892 893 JOIN categories c ON c.id = t.category_id
893 894 WHERE p.author_id = u.mnw_account_id
894 895 AND c.community_id = co.id
896 + AND p.removed_at IS NULL
895 897 AND p.deleted_at IS NULL
896 898 AND t.deleted_at IS NULL) AS endorsement_count
897 899 FROM users u
@@ -935,6 +937,7 @@ pub async fn get_user_activity_in_community(
935 937 JOIN categories c ON c.id = t.category_id
936 938 WHERE c.community_id = $1
937 939 AND p.author_id = $2
940 + AND p.removed_at IS NULL
938 941 AND p.deleted_at IS NULL
939 942 AND t.deleted_at IS NULL
940 943 ORDER BY p.created_at DESC
@@ -972,6 +975,7 @@ pub async fn get_user_membership_summary(
972 975 JOIN categories c ON c.id = t.category_id
973 976 WHERE p.author_id = $1
974 977 AND c.community_id = co.id
978 + AND p.removed_at IS NULL
975 979 AND p.deleted_at IS NULL
976 980 AND t.deleted_at IS NULL) AS post_count
977 981 FROM memberships m
M multithreaded/src/main.rs +49 -9
@@ -81,20 +81,29 @@ async fn main() {
81 81 let session_store = PostgresStore::new(pool);
82 82 session_store.migrate().await.expect("failed to migrate session store");
83 83
84 - let deletion_task = tokio::task::spawn(
85 - session_store
84 + // Both background loops run under panic supervision: a panic inside a sweep
85 + // would otherwise silently kill the worker for the life of the process
86 + // (S3 orphans pile up / the session table grows unbounded) while the server
87 + // keeps serving. `supervise` restarts the worker after a panic.
88 + let deletion_store = session_store.clone();
89 + let deletion_task = supervise("session-expiry-sweep", move || {
90 + deletion_store
86 91 .clone()
87 - .continuously_delete_expired(tokio::time::Duration::from_secs(3600)),
88 - );
92 + .continuously_delete_expired(tokio::time::Duration::from_secs(3600))
93 + });
89 94
90 95 // Reconcile sweep: purge S3 objects for removed images (backlog + retries).
91 96 // Only meaningful when S3 is configured.
92 97 let purge_task = state.s3.as_ref().map(|s3| {
93 - tokio::task::spawn(multithreaded::maintenance::continuously_purge_removed_images(
94 - state.db.clone(),
95 - s3.clone(),
96 - tokio::time::Duration::from_secs(6 * 3600),
97 - ))
98 + let db = state.db.clone();
99 + let s3 = s3.clone();
100 + supervise("image-purge-sweep", move || {
101 + multithreaded::maintenance::continuously_purge_removed_images(
102 + db.clone(),
103 + s3.clone(),
104 + tokio::time::Duration::from_secs(6 * 3600),
105 + )
106 + })
98 107 });
99 108
100 109 let session_layer = SessionManagerLayer::new(session_store)
@@ -162,6 +171,37 @@ async fn main() {
162 171 }
163 172 }
164 173
174 + /// Spawn a never-returning background loop under panic supervision.
175 + ///
176 + /// The inner future is run in its own task so a panic surfaces here as a
177 + /// `JoinError` instead of silently killing the worker; on a panic — or an
178 + /// unexpected normal return — it is logged and restarted after a short backoff.
179 + /// `factory` is `Fn` so it can rebuild the future (cloning any captured handles)
180 + /// on each restart. Aborting the returned handle (at shutdown) stops the loop.
181 + fn supervise<F, Fut>(name: &'static str, factory: F) -> tokio::task::JoinHandle<()>
182 + where
183 + F: Fn() -> Fut + Send + 'static,
184 + Fut: std::future::Future + Send + 'static,
185 + Fut::Output: Send,
186 + {
187 + const RESTART_BACKOFF: std::time::Duration = std::time::Duration::from_secs(5);
188 + tokio::task::spawn(async move {
189 + loop {
190 + match tokio::task::spawn(factory()).await {
191 + Ok(_) => {
192 + tracing::error!(task = name, "background task returned unexpectedly; restarting");
193 + }
194 + Err(e) if e.is_panic() => {
195 + tracing::error!(task = name, "background task panicked; restarting");
196 + }
197 + // Cancelled: the supervisor itself is being aborted at shutdown.
198 + Err(_) => return,
199 + }
200 + tokio::time::sleep(RESTART_BACKOFF).await;
201 + }
202 + })
203 + }
204 +
165 205 async fn shutdown_signal() {
166 206 use tokio::signal;
167 207 let ctrl_c = async { signal::ctrl_c().await.expect("failed to install Ctrl+C handler") };
M multithreaded/tests/workflows/admin_queries.rs +49
@@ -198,6 +198,55 @@ async fn test_membership_summary_post_count() {
198 198 }
199 199
200 200 #[tokio::test]
201 + async fn profile_counts_exclude_mod_removed_posts() {
202 + // Mod-removed posts (removed_at set) must not inflate profile/membership
203 + // post counts or the activity feed — the live hide column is removed_at,
204 + // not the (unused-for-posts) deleted_at (audit B1/B2).
205 + let mut h = TestHarness::new().await;
206 + let user_id = h.login_as("removee").await;
207 + let mod_id = h.login_as("themod").await;
208 + let comm_id = h.create_community("Mod Test", "modtest").await;
209 + let cat_id = h.create_category(comm_id, "General", "general").await;
210 + h.add_membership(user_id, comm_id, "member").await;
211 +
212 + // OP + two replies = 3 posts; then mod-remove one reply.
213 + let thread_id = h
214 + .create_thread_with_post(cat_id, user_id, "Thread", "OP body")
215 + .await;
216 + mt_db::mutations::create_post(&h.db, thread_id, user_id, "kept reply", "<p>kept</p>")
217 + .await
218 + .unwrap();
219 + let removed_post =
220 + mt_db::mutations::create_post(&h.db, thread_id, user_id, "bad reply", "<p>bad</p>")
221 + .await
222 + .unwrap();
223 + assert!(
224 + mt_db::mutations::mod_remove_post(&h.db, removed_post, mod_id)
225 + .await
226 + .unwrap()
227 + );
228 +
229 + // Membership summary: 3 created, 1 removed → 2.
230 + let summaries = mt_db::queries::get_user_membership_summary(&h.db, user_id)
231 + .await
232 + .unwrap();
233 + assert_eq!(summaries[0].post_count, 2, "removed post excluded from summary");
234 +
235 + // Profile counts: same exclusion on the per-community post_count.
236 + let profile = mt_db::queries::get_user_profile_in_community(&h.db, "modtest", "removee")
237 + .await
238 + .unwrap()
239 + .expect("profile exists");
240 + assert_eq!(profile.post_count, 2, "removed post excluded from profile count");
241 +
242 + // Activity feed must not surface the removed post.
243 + let activity = mt_db::queries::get_user_activity_in_community(&h.db, comm_id, user_id, 50)
244 + .await
245 + .unwrap();
246 + assert_eq!(activity.len(), 2, "removed post excluded from activity feed");
247 + }
248 +
249 + #[tokio::test]
201 250 async fn test_membership_summary_excludes_suspended() {
202 251 let mut h = TestHarness::new().await;
203 252