fix: batch account-maintenance per cycle + reclaim stuck-Running tasks

nt-be/src/handlers/balance_changes/account_monitor.rs

@@ -104,18 +104,31 @@ pub async fn run_maintenance_cycle(
     app_state: &AppState,
     up_to_block: i64,
 ) -> Result<(), Box<dyn std::error::Error>> {
-    // Process all enabled, non-confidential accounts; dirty accounts first.
-    // Confidential DAOs are handled by a dedicated 5-minute poll worker
+    // Process a bounded batch of enabled, non-confidential accounts per cycle:
+    // dirty accounts first, then the least-recently-synced ones. Confidential
+    // DAOs are handled by a dedicated 5-minute poll worker
     // (`run_confidential_poll_cycle`) and have no on-chain pipeline to run.
+    //
+    // The batch bound is what keeps a cycle finishing well within the handler
+    // `job_timeout`. Processing *every* account each cycle made the total time
+    // O(accounts); once that exceeded 30 min the handler timed out mid-cycle,
+    // and the timed-out task was left orphaned in `Running` (its ack never
+    // landed) — those phantom rows piled up. Dirty accounts still get processed
+    // promptly (they sort first); non-dirty accounts rotate by `last_synced_at`
+    // across cycles (the cron fires every 60s).
+    let batch_size = maintenance_batch_size();
     let accounts: Vec<(String, Option<DateTime<Utc>>)> = sqlx::query_as(
         r#"
         SELECT account_id, dirty_at
         FROM monitored_accounts
         WHERE enabled = true
           AND is_confidential_account = false
-        ORDER BY dirty_at DESC NULLS LAST
+        ORDER BY dirty_at DESC NULLS LAST,
+                 last_synced_at ASC NULLS FIRST
+        LIMIT $1
         "#,
     )
+    .bind(batch_size)
     .fetch_all(&app_state.db_pool)
     .await?;
 
@@ -178,6 +191,21 @@ pub async fn run_maintenance_cycle(
     Ok(())
 }
 
+/// Max accounts processed per maintenance cycle
+/// (`ACCOUNT_MAINTENANCE_BATCH_SIZE`, default 25). Bounds the cycle's total time
+/// so it finishes well within the handler `job_timeout` (default 30 min): worst
+/// case is `batch / concurrency × per_account_timeout`, i.e. 25/4×120s ≈ 12.5 min
+/// with the defaults. Dirty accounts sort first so they're never starved by the
+/// bound; the rest rotate by `last_synced_at`. Raise it only while keeping that
+/// worst case under `job_timeout`.
+fn maintenance_batch_size() -> i64 {
+    std::env::var("ACCOUNT_MAINTENANCE_BATCH_SIZE")
+        .ok()
+        .and_then(|s| s.parse::<i64>().ok())
+        .filter(|&n| n >= 1)
+        .unwrap_or(25)
+}
+
 /// Bounded parallelism for per-account maintenance within one cycle
 /// (`ACCOUNT_MAINTENANCE_CONCURRENCY`, default 4). Kept modest so parallel
 /// accounts don't exhaust the shared DB pool (max 20) or upstream RPC budgets.

nt-be/src/jobs/handlers.rs

@@ -544,6 +544,42 @@ pub async fn apalis_prune(_t: Tick, state: Data<Arc<AppState>>) -> Result<String
         .unwrap_or(48)
         .clamp(0, 3650 * 24);
 
+    // Reclaim tasks stuck in `Running` well past the handler timeout. apalis's
+    // own orphan-reclaim only fires when the *worker's* heartbeat is stale, so a
+    // task whose ack never landed (e.g. the handler hit `job_timeout` during DB
+    // instability, or the ack UPDATE failed) is left `Running` forever while the
+    // worker stays alive — these phantom rows accumulate (one per timeout) and
+    // are never reclaimed or pruned. The threshold is at least 2× `job_timeout`,
+    // so a genuinely-running task (bounded by `job_timeout`) is never touched.
+    // Reclaimed rows become `Killed` (terminal, not retried) and are then
+    // removed by the retention sweep below once they age out.
+    let reclaim_secs: i64 = std::env::var("APALIS_STUCK_RUNNING_RECLAIM_SECONDS")
+        .ok()
+        .and_then(|s| s.parse::<i64>().ok())
+        .filter(|&n| n >= 1)
+        .unwrap_or(3600)
+        .max((crate::jobs::job_timeout().as_secs() as i64).saturating_mul(2));
+    let reclaimed = sqlx::query(
+        r#"UPDATE apalis.jobs
+           SET status = 'Killed',
+               done_at = NOW(),
+               last_result = '{"Err": "Reclaimed: stuck in Running past threshold (ack likely failed)"}'::jsonb
+           WHERE status = 'Running'
+             AND lock_at IS NOT NULL
+             AND lock_at < now() - make_interval(secs => $1::double precision)"#,
+    )
+    .bind(reclaim_secs)
+    .execute(&state.db_pool)
+    .await?;
+    let reclaimed = reclaimed.rows_affected();
+    if reclaimed > 0 {
+        tracing::warn!(
+            reclaimed,
+            reclaim_secs,
+            "reclaimed apalis tasks stuck in Running (orphaned locks)"
+        );
+    }
+
     const BATCH_SIZE: i64 = 10_000;
     // Backstop against a runaway loop; 100 batches = up to 1M rows per run.
     const MAX_BATCHES: usize = 100;
@@ -576,6 +612,6 @@ pub async fn apalis_prune(_t: Tick, state: Data<Arc<AppState>>) -> Result<String
     }
 
     Ok(format!(
-        "pruned {total} terminal tasks older than {retention_hours}h"
+        "reclaimed {reclaimed} stuck-Running, pruned {total} terminal tasks older than {retention_hours}h"
     ))
 }