Skip to main content

max / makenotwork

server: fold in outstanding working-tree changes Scanning (worker/spool/archive), storage routes (media/gallery/uploads/ versions), stripe subscription webhook, totp/crypto, db modules, and migration 137 (pending_uploads bucket uniqueness). Committed as-is from the working tree.
Co-Authored-By
Claude Opus 4.8 (1M context) <noreply@anthropic.com>
Author: Max Johnson <me@maxj.phd> · 2026-06-12 21:29 UTC
Signed with PGP, not checked
Commit: a27ad4d353d0a6c6dacf0b82dd90579930112a54
Parent: bd89948
36 files changed, +995 insertions, -209 deletions
M server/Cargo.lock +64
@@ -17,6 +17,16 @@
17 17 source = "registry+https://github.com/rust-lang/crates.io-index"
18 18 checksum = "320119579fcad9c21884f5c4861d16174d0e06250625266f50fe6898340abefa"
19 19
20 + [[package]]
21 + name = "aead"
22 + version = "0.5.2"
23 + source = "registry+https://github.com/rust-lang/crates.io-index"
24 + checksum = "d122413f284cf2d62fb1b7db97e02edb8cda96d769b16e443a4f6195e35662b0"
25 + dependencies = [
26 + "crypto-common 0.1.6",
27 + "generic-array",
28 + ]
29 +
20 30 [[package]]
21 31 name = "aes"
22 32 version = "0.8.4"
@@ -1559,6 +1569,30 @@
1559 1569 source = "registry+https://github.com/rust-lang/crates.io-index"
1560 1570 checksum = "613afe47fcd5fac7ccf1db93babcb082c5994d996f20b8b159f2ad1658eb5724"
1561 1571
1572 + [[package]]
1573 + name = "chacha20"
1574 + version = "0.9.1"
1575 + source = "registry+https://github.com/rust-lang/crates.io-index"
1576 + checksum = "c3613f74bd2eac03dad61bd53dbe620703d4371614fe0bc3b9f04dd36fe4e818"
1577 + dependencies = [
1578 + "cfg-if",
1579 + "cipher",
1580 + "cpufeatures 0.2.17",
1581 + ]
1582 +
1583 + [[package]]
1584 + name = "chacha20poly1305"
1585 + version = "0.10.1"
1586 + source = "registry+https://github.com/rust-lang/crates.io-index"
1587 + checksum = "10cd79432192d1c0f4e1a0fef9527696cc039165d729fb41b3f4f4f354c2dc35"
1588 + dependencies = [
1589 + "aead",
1590 + "chacha20",
1591 + "cipher",
1592 + "poly1305",
1593 + "zeroize",
1594 + ]
1595 +
1562 1596 [[package]]
1563 1597 name = "chrono"
1564 1598 version = "0.4.44"
@@ -1581,6 +1615,7 @@
1581 1615 dependencies = [
1582 1616 "crypto-common 0.1.6",
1583 1617 "inout",
1618 + "zeroize",
1584 1619 ]
1585 1620
1586 1621 [[package]]
@@ -2069,6 +2104,7 @@
2069 2104 checksum = "1bfb12502f3fc46cca1bb51ac28df9d618d813cdc3d2f25b9fe775a34af26bb3"
2070 2105 dependencies = [
2071 2106 "generic-array",
2107 + "rand_core 0.6.4",
2072 2108 "typenum",
2073 2109 ]
2074 2110
@@ -4161,6 +4197,7 @@
4161 4197 "axum-extra",
4162 4198 "base64 0.22.1",
4163 4199 "bzip2 0.4.4",
4200 + "chacha20poly1305",
4164 4201 "chrono",
4165 4202 "clap",
4166 4203 "const-oid 0.9.6",
@@ -4678,6 +4715,12 @@
4678 4715 source = "registry+https://github.com/rust-lang/crates.io-index"
4679 4716 checksum = "384b8ab6d37215f3c5301a95a4accb5d64aa607f1fcb26a11b5303878451b4fe"
4680 4717
4718 + [[package]]
4719 + name = "opaque-debug"
4720 + version = "0.3.1"
4721 + source = "registry+https://github.com/rust-lang/crates.io-index"
4722 + checksum = "c08d65885ee38876c4f86fa503fb49d7b507c2b62552df7c70b2fce627e06381"
4723 +
4681 4724 [[package]]
4682 4725 name = "openssl"
4683 4726 version = "0.10.76"
@@ -5065,6 +5108,17 @@
5065 5108 "miniz_oxide",
5066 5109 ]
5067 5110
5111 + [[package]]
5112 + name = "poly1305"
5113 + version = "0.8.0"
5114 + source = "registry+https://github.com/rust-lang/crates.io-index"
5115 + checksum = "8159bd90725d2df49889a078b54f4f79e87f1f8a8444194cdca81d38f5393abf"
5116 + dependencies = [
5117 + "cpufeatures 0.2.17",
5118 + "opaque-debug",
5119 + "universal-hash",
5120 + ]
5121 +
5068 5122 [[package]]
5069 5123 name = "pom-contract"
5070 5124 version = "0.1.0"
@@ -7607,6 +7661,16 @@
7607 7661 source = "registry+https://github.com/rust-lang/crates.io-index"
7608 7662 checksum = "ebc1c04c71510c7f702b52b7c350734c9ff1295c464a03335b00bb84fc54f853"
7609 7663
7664 + [[package]]
7665 + name = "universal-hash"
7666 + version = "0.5.1"
7667 + source = "registry+https://github.com/rust-lang/crates.io-index"
7668 + checksum = "fc1de2c688dc15305988b563c3854064043356019f97a4b46276fe734c4f07ea"
7669 + dependencies = [
7670 + "crypto-common 0.1.6",
7671 + "subtle",
7672 + ]
7673 +
7610 7674 [[package]]
7611 7675 name = "unsafe-libyaml"
7612 7676 version = "0.2.11"
M server/Cargo.toml +1
@@ -69,6 +69,7 @@
69 69 sha1 = "0.10.6"
70 70 sha2 = "0.10.9"
71 71 subtle = "2.6"
72 + chacha20poly1305 = "0.10.1"
72 73 hex = "0.4.3"
73 74 base64 = "0.22.1"
74 75
M server/src/config.rs +54 -2
@@ -217,8 +217,20 @@
217 217 .ok()
218 218 .and_then(|s| s.parse::<UserId>().ok());
219 219
220 - // SyncKit JWT secret - optional, sync endpoints return 503 if unset
221 - let synckit_jwt_secret = std::env::var("SYNCKIT_JWT_SECRET").ok();
220 + // SyncKit JWT secret - optional, sync endpoints return 503 if unset.
221 + // When set it IS the HS256 symmetric signing key for SyncKit/OAuth
222 + // bearer tokens, so enforce the same >=32-char floor as SIGNING_SECRET:
223 + // a short value is offline-brute-forceable into token forgery. Fail
224 + // closed (refuse to start) rather than silently accepting a weak key.
225 + let synckit_jwt_secret = match std::env::var("SYNCKIT_JWT_SECRET") {
226 + Ok(secret) => {
227 + if secret.len() < 32 {
228 + return Err(ConfigError::WeakSynckitJwtSecret);
229 + }
230 + Some(secret)
231 + }
232 + Err(_) => None,
233 + };
222 234
223 235 // File scanning - enabled by default, set SCAN_ENABLED=false to disable
224 236 let scan = ScanConfig::from_env();
@@ -592,6 +604,8 @@
592 604 MissingSigningSecret,
593 605 #[error("SIGNING_SECRET must be at least 32 characters long")]
594 606 WeakSigningSecret,
607 + #[error("SYNCKIT_JWT_SECRET must be at least 32 characters long")]
608 + WeakSynckitJwtSecret,
595 609 }
596 610
597 611 #[cfg(test)]
@@ -792,6 +806,44 @@
792 806 drop(guard);
793 807 }
794 808
809 + #[test]
810 + fn from_env_fails_with_short_synckit_jwt_secret() {
811 + let guard = EnvGuard::new();
812 + guard.clear_all();
813 +
814 + // SAFETY: test-only, serialized by EnvGuard mutex
815 + unsafe {
816 + std::env::set_var("DATABASE_URL", "postgres://localhost/test_db");
817 + std::env::set_var("SIGNING_SECRET", "x".repeat(32));
818 + // 31 chars — one under the floor.
819 + std::env::set_var("SYNCKIT_JWT_SECRET", "x".repeat(31));
820 + }
821 +
822 + let err = Config::from_env().unwrap_err();
823 + assert!(
824 + matches!(err, ConfigError::WeakSynckitJwtSecret),
825 + "expected WeakSynckitJwtSecret, got: {err}"
826 + );
827 + drop(guard);
828 + }
829 +
830 + #[test]
831 + fn from_env_accepts_strong_synckit_jwt_secret() {
832 + let guard = EnvGuard::new();
833 + guard.clear_all();
834 +
835 + // SAFETY: test-only, serialized by EnvGuard mutex
836 + unsafe {
837 + std::env::set_var("DATABASE_URL", "postgres://localhost/test_db");
838 + std::env::set_var("SIGNING_SECRET", "x".repeat(32));
839 + std::env::set_var("SYNCKIT_JWT_SECRET", "y".repeat(32));
840 + }
841 +
842 + let config = Config::from_env().expect("32-char JWT secret should be accepted");
843 + assert_eq!(config.synckit_jwt_secret.as_deref(), Some("y".repeat(32).as_str()));
844 + drop(guard);
845 + }
846 +
795 847 #[test]
796 848 fn from_env_uses_random_dev_secret_when_not_production() {
797 849 let guard = EnvGuard::new();
M server/src/constants.rs +22 -1
@@ -171,7 +171,14 @@
171 171
172 172 // -- File scanning --
173 173 pub const SCAN_MAX_MEMORY_BYTES: usize = 100 * 1024 * 1024; // 100 MB in-memory threshold
174 - pub const SCAN_MAX_CONCURRENT: usize = 4; // Max concurrent file scans (each can use up to 100 MB RAM)
174 + // Ceiling on in-flight scans, enforced by a semaphore around the CPU/clamd
175 + // phase. NOTE: with `SCAN_WORKER_COUNT` workers each scanning one file at a
176 + // time, the real concurrency is `min(SCAN_MAX_CONCURRENT, SCAN_WORKER_COUNT)` —
177 + // today that's 2 (~200 MB peak), so this semaphore only begins to bind if the
178 + // worker count is raised above it. Kept as an explicit ceiling so that raising
179 + // `SCAN_WORKER_COUNT` can't silently blow past the memory budget. The
180 + // assertion below documents that intent.
181 + pub const SCAN_MAX_CONCURRENT: usize = 4; // Memory-budget ceiling on concurrent scans
175 182 pub const SCAN_WORKER_COUNT: usize = 2; // Background worker tasks draining scan_jobs queue
176 183 /// Retention window for terminal-state (`done`, `failed`) `scan_jobs` rows.
177 184 /// Queued/running rows are operational queue state and not affected.
@@ -204,8 +211,18 @@
204 211 pub const SCAN_ZIP_MAX_RATIO: f64 = 100.0; // Max compression ratio before ZIP bomb
205 212 pub const SCAN_ZIP_MAX_DEPTH: u32 = 2; // Max nested archives (detection is 1 level deep; decompressed size limit is the primary defense)
206 213 pub const SCAN_ZIP_MAX_UNCOMPRESSED: u64 = 2 * 1024 * 1024 * 1024; // 2 GB uncompressed limit
214 + // Cap the number of ZIP entries inspected. Depth/ratio/uncompressed-size are
215 + // already bounded, but a ZIP with millions of tiny entries forces a full
216 + // per-entry decompression pass bounded only by the 2 GB total. 100k entries is
217 + // far past any legitimate sample pack / content bundle; beyond it, fail closed.
218 + pub const SCAN_ZIP_MAX_ENTRIES: usize = 100_000;
207 219 pub const SCAN_MALWAREBAZAAR_TIMEOUT_SECS: u64 = 5;
208 220 pub const SCAN_CLAMAV_TIMEOUT_SECS: u64 = 30;
221 + // How often the background probe pings clamd to maintain the runtime liveness
222 + // flag. The boot gate proves clamd was up at startup; this catches a death
223 + // AFTER boot so a configured-but-erroring clamav scan holds otherwise-clean
224 + // uploads for review instead of passing them on zero AV coverage (FailOpen).
225 + pub const SCAN_CLAMAV_HEALTH_PROBE_SECS: u64 = 30;
209 226
210 227 // -- Invite system --
211 228 pub const INVITES_ENABLED: bool = true;
@@ -358,6 +375,10 @@
358 375 const _: () = assert!(SCAN_SPOOL_FREE_RESERVE_BYTES < SCAN_SPOOL_MAX_BYTES);
359 376 const _: () = assert!(SCAN_SPOOL_MAX_BYTES > SCAN_MAX_MEMORY_BYTES as u64);
360 377 const _: () = assert!(SCAN_JOB_RETENTION_DAYS >= 7); // no same-day purge race
378 + // The concurrency ceiling must not sit below the worker count, or the memory
379 + // budget it's meant to enforce is unenforceable (workers would exceed it).
380 + const _: () = assert!(SCAN_MAX_CONCURRENT >= SCAN_WORKER_COUNT);
381 + const _: () = assert!(SCAN_ZIP_MAX_ENTRIES > 0);
361 382 const _: () = assert!(BROADCAST_PARALLELISM > 0 && BROADCAST_PARALLELISM <= 64);
362 383 const _: () = assert!(SCAN_ZIP_MAX_UNCOMPRESSED > SCAN_MAX_MEMORY_BYTES as u64);
363 384 const _: () = assert!(GIT_RAW_MAX_BYTES > GIT_MAX_FILE_SIZE_BYTES);
M server/src/crypto.rs +145 -1
@@ -1,4 +1,101 @@
1 - //! Cryptographic utilities: constant-time comparison, key generation, feed signing.
1 + //! Cryptographic utilities: constant-time comparison, key generation, feed
2 + //! signing, and secret encryption at rest.
3 +
4 + use crate::error::{AppError, Result};
5 +
6 + /// Version-tagged prefix on an encrypted-at-rest TOTP secret. Its presence is
7 + /// how [`decrypt_totp_secret`] distinguishes a ciphertext from a legacy
8 + /// plaintext base32 seed during the dual-read migration window.
9 + const TOTP_ENC_PREFIX: &str = "enc:v1:";
10 +
11 + /// Derive a domain-separated 32-byte key for TOTP-secret encryption from the
12 + /// global signing secret, using HMAC-SHA256 as a PRF. The label keeps this key
13 + /// independent of every other use of the signing secret (feed signing,
14 + /// backup-code HMAC, session tokens), so reuse in one context can't weaken
15 + /// another.
16 + fn totp_encryption_key(signing_secret: &str) -> [u8; 32] {
17 + use hmac::{Hmac, Mac};
18 + use sha2::Sha256;
19 +
20 + let mut mac = Hmac::<Sha256>::new_from_slice(signing_secret.as_bytes())
21 + .expect("HMAC-SHA256 accepts any key length");
22 + mac.update(b"mnw-totp-secret-encryption-v1");
23 + mac.finalize().into_bytes().into()
24 + }
25 +
26 + /// Encrypt a TOTP secret for storage at rest with ChaCha20-Poly1305 (AEAD).
27 + ///
28 + /// The on-disk form is `enc:v1:` + base64(`nonce(12) || ciphertext+tag`). A
29 + /// fresh random nonce is drawn per call, so encrypting the same seed twice
30 + /// yields distinct ciphertexts. A database read alone (snapshot, replica, SQL
31 + /// injection elsewhere) no longer yields a usable second factor — the attacker
32 + /// also needs `SIGNING_SECRET`.
33 + pub fn encrypt_totp_secret(plaintext: &str, signing_secret: &str) -> String {
34 + use base64::Engine;
35 + use chacha20poly1305::{aead::Aead, ChaCha20Poly1305, KeyInit, Nonce};
36 +
37 + let key = totp_encryption_key(signing_secret);
38 + let cipher = ChaCha20Poly1305::new((&key).into());
39 +
40 + let mut nonce_bytes = [0u8; 12];
41 + rand::RngCore::fill_bytes(&mut rand::rng(), &mut nonce_bytes);
42 + let nonce = Nonce::from(nonce_bytes);
43 +
44 + let ciphertext = cipher
45 + .encrypt(&nonce, plaintext.as_bytes())
46 + // Encryption of an in-memory plaintext with a valid key/nonce cannot
47 + // fail; the only error variant is for buffer-size issues we don't hit.
48 + .expect("ChaCha20-Poly1305 encryption is infallible here");
49 +
50 + let mut payload = Vec::with_capacity(nonce_bytes.len() + ciphertext.len());
51 + payload.extend_from_slice(&nonce_bytes);
52 + payload.extend_from_slice(&ciphertext);
53 +
54 + format!(
55 + "{TOTP_ENC_PREFIX}{}",
56 + base64::engine::general_purpose::STANDARD.encode(payload)
57 + )
58 + }
59 +
60 + /// Decrypt a stored TOTP secret produced by [`encrypt_totp_secret`].
61 + ///
62 + /// Every stored secret MUST carry the `enc:v1:` prefix. There is no legacy
63 + /// plaintext fallback — a value without the prefix is rejected as malformed
64 + /// (backwards compatibility with pre-encryption plaintext seeds was cut, so
65 + /// any such user re-enrolls their authenticator). Also errors when a tagged
66 + /// ciphertext fails to decode or authenticate (wrong key or tampering).
67 + pub fn decrypt_totp_secret(stored: &str, signing_secret: &str) -> Result<String> {
68 + use base64::Engine;
69 + use chacha20poly1305::{aead::Aead, ChaCha20Poly1305, KeyInit, Nonce};
70 +
71 + let Some(b64) = stored.strip_prefix(TOTP_ENC_PREFIX) else {
72 + return Err(AppError::Internal(anyhow::anyhow!(
73 + "totp secret is not encrypted (missing {TOTP_ENC_PREFIX} prefix)"
74 + )));
75 + };
76 +
77 + let payload = base64::engine::general_purpose::STANDARD
78 + .decode(b64)
79 + .map_err(|e| AppError::Internal(anyhow::anyhow!("totp secret base64 decode: {e}")))?;
80 + if payload.len() < 12 {
81 + return Err(AppError::Internal(anyhow::anyhow!(
82 + "totp secret ciphertext too short"
83 + )));
84 + }
85 + let (nonce_bytes, ciphertext) = payload.split_at(12);
86 + let nonce_arr: [u8; 12] = nonce_bytes
87 + .try_into()
88 + .expect("split_at(12) on a >=12-byte payload yields exactly 12 bytes");
89 +
90 + let key = totp_encryption_key(signing_secret);
91 + let cipher = ChaCha20Poly1305::new((&key).into());
92 + let plaintext = cipher
93 + .decrypt(&Nonce::from(nonce_arr), ciphertext)
94 + .map_err(|_| AppError::Internal(anyhow::anyhow!("totp secret decryption failed")))?;
95 +
96 + String::from_utf8(plaintext)
97 + .map_err(|e| AppError::Internal(anyhow::anyhow!("totp secret utf8: {e}")))
98 + }
2 99
3 100 /// Constant-time byte comparison for tokens, MACs, and other fixed-shape
4 101 /// secrets. Backed by [`subtle::ConstantTimeEq`] (audited reference impl)
@@ -86,6 +183,53 @@
86 183 mod tests {
87 184 use super::*;
88 185
186 + // ── TOTP secret encryption at rest ──
187 +
188 + #[test]
189 + fn totp_secret_round_trips() {
190 + let secret = "JBSWY3DPEHPK3PXP";
191 + let key = "a-stable-signing-secret-at-least-32c";
192 + let enc = encrypt_totp_secret(secret, key);
193 + assert!(enc.starts_with("enc:v1:"), "ciphertext must be version-tagged");
194 + assert_ne!(enc, secret, "ciphertext must not be the plaintext");
195 + assert_eq!(decrypt_totp_secret(&enc, key).unwrap(), secret);
196 + }
197 +
198 + #[test]
199 + fn totp_secret_nonce_is_random() {
200 + let secret = "JBSWY3DPEHPK3PXP";
201 + let key = "a-stable-signing-secret-at-least-32c";
202 + // Same plaintext + key encrypted twice must differ (fresh nonce each time).
203 + assert_ne!(encrypt_totp_secret(secret, key), encrypt_totp_secret(secret, key));
204 + }
205 +
206 + #[test]
207 + fn totp_secret_wrong_key_fails_to_decrypt() {
208 + let secret = "JBSWY3DPEHPK3PXP";
209 + let enc = encrypt_totp_secret(secret, "a-stable-signing-secret-at-least-32c");
210 + assert!(decrypt_totp_secret(&enc, "a-different-signing-secret-32-chars!").is_err());
211 + }
212 +
213 + #[test]
214 + fn totp_secret_tampered_ciphertext_fails() {
215 + let key = "a-stable-signing-secret-at-least-32c";
216 + let enc = encrypt_totp_secret("JBSWY3DPEHPK3PXP", key);
217 + // Flip a character in the base64 body — the AEAD tag must reject it.
218 + let mut bytes: Vec<char> = enc.chars().collect();
219 + let last = bytes.len() - 1;
220 + bytes[last] = if bytes[last] == 'A' { 'B' } else { 'A' };
221 + let tampered: String = bytes.into_iter().collect();
222 + assert!(decrypt_totp_secret(&tampered, key).is_err());
223 + }
224 +
225 + #[test]
226 + fn totp_secret_unprefixed_plaintext_is_rejected() {
227 + // Backwards compat was cut: a bare (pre-encryption) plaintext seed has
228 + // no `enc:v1:` prefix and must be rejected, not trusted.
229 + let key = "a-stable-signing-secret-at-least-32c";
230 + assert!(decrypt_totp_secret("JBSWY3DPEHPK3PXP", key).is_err());
231 + }
232 +
89 233 // ── constant_time_compare ──
90 234
91 235 #[test]
M server/src/main.rs +16
@@ -363,18 +363,34 @@
363 363 // Start scan worker pool. Only meaningful if a scanner is configured;
364 364 // otherwise enqueue_scan_for never enqueues (trust-gate fast path).
365 365 if let (Some(scanner), Some(s3_for_workers)) = (state.scanner.clone(), state.s3.clone()) {
366 + // Runtime ClamAV liveness: starts healthy (the boot gate already proved
367 + // clamd was up), maintained by the probe spawned below.
368 + let clamav_healthy = std::sync::Arc::new(std::sync::atomic::AtomicBool::new(true));
369 + let clamav_socket = scanner.clamav_socket().map(str::to_string);
366 370 let scan_ctx = std::sync::Arc::new(makenotwork::scanning::worker::WorkerContext {
367 371 db: state.db.clone(),
368 372 s3: s3_for_workers,
369 373 pipeline: scanner,
370 374 scan_semaphore: state.scan_semaphore.clone(),
371 375 wam: state.wam.clone(),
376 + clamav_healthy: clamav_healthy.clone(),
372 377 });
373 378 let worker_count = makenotwork::constants::SCAN_WORKER_COUNT;
374 379 let worker_shutdown_rx = shutdown_tx.subscribe();
375 380 makenotwork::scanning::worker::spawn_pool(worker_count, scan_ctx, worker_shutdown_rx);
376 381 tracing::info!(worker_count, "scan worker pool started");
377 382
383 + // Post-boot clamd liveness probe — flips clean verdicts to held-for-review
384 + // if clamd dies after startup (the boot gate can't catch that).
385 + if let Some(socket) = clamav_socket {
386 + makenotwork::scanning::worker::spawn_clamav_health_probe(
387 + socket,
388 + clamav_healthy,
389 + shutdown_tx.subscribe(),
390 + );
391 + tracing::info!("clamav runtime health probe started");
392 + }
393 +
378 394 let report = makenotwork::scanning::spool::reap_all(
379 395 std::path::Path::new(makenotwork::constants::SCAN_SPOOL_DIR),
380 396 );
M server/src/monitor.rs +9 -3
@@ -208,7 +208,11 @@
208 208 }
209 209
210 210 // Notify opted-in users of status changes (fire-and-forget).
211 - // Paced at ~10/sec to stay under Postmark's default send rate.
211 + // Batched pacing: pause 1s every 50 sends (matching the
212 + // scheduler's announcement fan-out) instead of a hard 100ms
213 + // between every send, which turned a 1000-subscriber notify
214 + // into a ~100s single task holding the email-client clone.
215 + // The subscriber list is bounded by the query's LIMIT.
212 216 {
213 217 let pool = state.db.clone();
214 218 let email_client = state.email.clone();
@@ -220,7 +224,10 @@
220 224 match db::users::get_status_alert_subscribers(&pool).await {
221 225 Ok(subscribers) if !subscribers.is_empty() => {
222 226 tracing::info!(count = subscribers.len(), "sending status notifications to opted-in users");
223 - for sub in &subscribers {
227 + for (i, sub) in subscribers.iter().enumerate() {
228 + if i > 0 && i % 50 == 0 {
229 + tokio::time::sleep(std::time::Duration::from_secs(1)).await;
230 + }
224 231 let unsub_url = crate::email::generate_unsubscribe_url(
225 232 &host_url, sub.id, crate::email::UnsubscribeAction::Status, &sub.id.to_string(), &signing_secret,
226 233 );
@@ -231,7 +238,6 @@
231 238 &prev_status,
232 239 &unsub_url,
233 240 ).await;
234 - tokio::time::sleep(std::time::Duration::from_millis(100)).await;
235 241 }
236 242 }
237 243 Err(e) => {
M server/src/db/creator_tiers.rs +7
@@ -425,6 +425,11 @@
425 425 /// whose items have not yet been hidden.
426 426 #[tracing::instrument(skip_all)]
427 427 pub async fn get_expired_grace_creators(pool: &PgPool) -> Result<Vec<UserId>> {
428 + // Bounded batch per call. The scheduler enforces these inline on the tick
429 + // (two DB round-trips per creator), so an unbounded result set would let a
430 + // backlog stall the tick. `grace_enforced_at` is set as each creator is
431 + // processed, so successive ticks drain the rest; ORDER BY oldest-first keeps
432 + // it deterministic and starvation-free.
428 433 let ids: Vec<UserId> = sqlx::query_scalar(
429 434 r#"
430 435 SELECT user_id FROM creator_subscriptions
@@ -432,6 +437,8 @@
432 437 AND canceled_at IS NOT NULL
433 438 AND canceled_at < NOW() - INTERVAL '30 days'
434 439 AND grace_enforced_at IS NULL
440 + ORDER BY canceled_at ASC
441 + LIMIT 200
435 442 "#,
436 443 )
437 444 .fetch_all(pool)
M server/src/db/pending_s3_deletions.rs +6 -3
@@ -16,8 +16,8 @@
16 16
17 17 /// Enqueue S3 keys for deletion. Each key is (s3_key, bucket).
18 18 #[tracing::instrument(skip_all)]
19 - pub async fn enqueue_deletions(
20 - pool: &PgPool,
19 + pub async fn enqueue_deletions<'e>(
20 + executor: impl sqlx::PgExecutor<'e>,
21 21 keys: &[(String, String)],
22 22 source: &str,
23 23 ) -> Result<()> {
@@ -26,13 +26,16 @@
26 26 }
27 27 let s3_keys: Vec<&str> = keys.iter().map(|(k, _)| k.as_str()).collect();
28 28 let buckets: Vec<&str> = keys.iter().map(|(_, b)| b.as_str()).collect();
29 + // Takes any `PgExecutor` so callers can enqueue inside a transaction —
30 + // letting a row delete, its storage refund, and the S3-delete enqueue commit
31 + // atomically (see `versions::delete_version`).
29 32 sqlx::query(
30 33 "INSERT INTO pending_s3_deletions (s3_key, bucket, source) SELECT * FROM unnest($1::text[], $2::text[], $3::text[]) ON CONFLICT DO NOTHING",
31 34 )
32 35 .bind(&s3_keys)
33 36 .bind(&buckets)
34 37 .bind(vec![source; keys.len()])
35 - .execute(pool)
38 + .execute(executor)
36 39 .await?;
37 40 Ok(())
38 41 }
M server/src/db/pending_uploads.rs +1 -1
M server/src/db/scan_jobs.rs +54 -9
M server/src/db/totp.rs +28 -6
M server/src/db/validated_types.rs +10 -6
M server/src/db/versions.rs +21 -17
M server/src/scanning/archive.rs +17
M server/src/scanning/mod.rs +11 -1
M server/src/scanning/spool.rs +15
M server/src/scanning/worker.rs +167 -35
M server/templates/pages/forgot_password.html +1
M server/tests/harness/mod.rs +3
M server/src/db/items/media.rs +75 -1
M server/src/routes/api/totp.rs +4 -3
M server/src/routes/pages/blog.rs +9 -8
M server/src/routes/pages/feeds.rs +11 -9
M server/src/routes/pages/mod.rs +14 -4
M server/src/routes/pages/sandbox.rs +12 -6
M server/src/routes/storage/gallery.rs +8 -1
M server/src/routes/storage/media.rs +11 -1
M server/src/routes/storage/mod.rs +44 -1
M server/src/routes/storage/uploads.rs +44 -60
M server/src/routes/storage/versions.rs +9 -1
M server/src/routes/pages/email_actions/mod.rs +41 -16
M server/src/routes/pages/public/two_factor.rs +6 -3
M server/src/routes/stripe/webhook/subscriptions.rs +16 -4
M server/src/routes/pages/dashboard/wizards/item/mod.rs +28 -6
A server/migrations/137_pending_uploads_bucket_unique.sql +11