fix: sequential batch indicator processing with 40-min per-indicator timeout

Generated with [Claude Code](https://claude.ai/code)
via [Happy](https://happy.engineering)

Co-Authored-By: Claude <noreply@anthropic.com>
Co-Authored-By: Happy <yesreply@happy.engineering>

app/worker.py

@@ -19,7 +19,7 @@ from app.core.email import send_completion_email
 logger = logging.getLogger(__name__)
 
 BATCH_POLL_INTERVAL = 30  # seconds between status checks
-BATCH_TIMEOUT = 1200  # 20 minutes maximum wait
+BATCH_TIMEOUT = 2400  # 40 minutes per indicator
 
 
 def _save_spatial_json(spatial, status_value: str, path: str) -> None:
@@ -67,10 +67,11 @@ async def process_job(job_id: str, db: Database, registry: IndicatorRegistry) ->
             else:
                 process_indicators.append((indicator_id, indicator))
 
-        # -- Phase 1: Submit batch jobs --
-        batch_submissions = {}
-        fallback_ids = set()
+        # -- Process batch indicators sequentially --
         for indicator_id, indicator in batch_indicators.items():
+            result = None
+
+            # Submit
             await db.update_job_progress(job_id, indicator_id, "submitting")
             try:
                 jobs = await indicator.submit_batch(
@@ -78,71 +79,52 @@ async def process_job(job_id: str, db: Database, registry: IndicatorRegistry) ->
                     job.request.time_range,
                     season_months=job.request.season_months(),
                 )
-                batch_submissions[indicator_id] = jobs
                 job_ids = [getattr(j, 'job_id', '?') for j in jobs]
                 print(f"[Aperture] Submitted {indicator_id} batch jobs: {job_ids}")
                 await db.update_job_progress(job_id, indicator_id, "processing on CDSE")
             except Exception as exc:
                 logger.warning("Batch submit failed for %s, will use fallback: %s", indicator_id, exc)
-                fallback_ids.add(indicator_id)
-
-        # -- Phase 2: Poll until all batch jobs finish --
-        poll_start = time.monotonic()
-        pending = dict(batch_submissions)
-
-        while pending:
-            # Check current statuses before sleeping
-            elapsed = time.monotonic() - poll_start
-            for indicator_id in list(pending.keys()):
-                jobs = pending[indicator_id]
-                statuses = [j.status() for j in jobs]
-                job_ids = [getattr(j, 'job_id', '?') for j in jobs]
-                print(f"[Aperture] Poll {indicator_id} ({elapsed:.0f}s): {list(zip(job_ids, statuses))}")
-                if all(s == "finished" for s in statuses):
-                    logger.info("Batch jobs finished for %s", indicator_id)
-                    del pending[indicator_id]
-                elif any(s in ("error", "canceled") for s in statuses):
-                    logger.warning("Batch job failed for %s: %s", indicator_id, statuses)
-                    del pending[indicator_id]
-
-            if not pending:
-                break
-
-            if elapsed >= BATCH_TIMEOUT:
-                logger.warning("Batch poll timeout after %.0fs, remaining: %s", elapsed, list(pending.keys()))
-                fallback_ids.update(pending.keys())
-                break
-
-            await asyncio.sleep(BATCH_POLL_INTERVAL)
-
-        # -- Phase 3: Harvest batch results + process non-batch indicators --
-        for indicator_id in job.request.indicator_ids:
-            indicator = registry.get(indicator_id)
-
-            if indicator_id in fallback_ids:
-                await db.update_job_progress(job_id, indicator_id, "processing")
-                result = await indicator.process(
-                    job.request.aoi,
-                    job.request.time_range,
-                    season_months=job.request.season_months(),
-                )
-            elif indicator_id in batch_submissions:
-                await db.update_job_progress(job_id, indicator_id, "downloading")
-                try:
-                    result = await indicator.harvest(
-                        job.request.aoi,
-                        job.request.time_range,
-                        season_months=job.request.season_months(),
-                        batch_jobs=batch_submissions[indicator_id],
-                    )
-                except Exception as exc:
-                    logger.warning("Harvest failed for %s, using fallback: %s", indicator_id, exc)
-                    result = await indicator.process(
-                        job.request.aoi,
-                        job.request.time_range,
-                        season_months=job.request.season_months(),
-                    )
-            else:
+                jobs = None
+
+            # Poll
+            if jobs is not None:
+                poll_start = time.monotonic()
+                finished = False
+                while True:
+                    elapsed = time.monotonic() - poll_start
+                    statuses = [j.status() for j in jobs]
+                    job_ids = [getattr(j, 'job_id', '?') for j in jobs]
+                    print(f"[Aperture] Poll {indicator_id} ({elapsed:.0f}s): {list(zip(job_ids, statuses))}")
+
+                    if all(s == "finished" for s in statuses):
+                        logger.info("Batch jobs finished for %s", indicator_id)
+                        finished = True
+                        break
+                    elif any(s in ("error", "canceled") for s in statuses):
+                        logger.warning("Batch job failed for %s: %s", indicator_id, statuses)
+                        break
+
+                    if elapsed >= BATCH_TIMEOUT:
+                        logger.warning("Batch poll timeout after %.0fs for %s", elapsed, indicator_id)
+                        break
+
+                    await asyncio.sleep(BATCH_POLL_INTERVAL)
+
+                # Harvest
+                if finished:
+                    await db.update_job_progress(job_id, indicator_id, "downloading")
+                    try:
+                        result = await indicator.harvest(
+                            job.request.aoi,
+                            job.request.time_range,
+                            season_months=job.request.season_months(),
+                            batch_jobs=jobs,
+                        )
+                    except Exception as exc:
+                        logger.warning("Harvest failed for %s, using fallback: %s", indicator_id, exc)
+
+            # Fallback if submit failed, poll timed out, jobs errored, or harvest failed
+            if result is None:
                 await db.update_job_progress(job_id, indicator_id, "processing")
                 result = await indicator.process(
                     job.request.aoi,
@@ -153,7 +135,20 @@ async def process_job(job_id: str, db: Database, registry: IndicatorRegistry) ->
             spatial = indicator.get_spatial_data()
             if spatial is not None:
                 spatial_cache[indicator_id] = spatial
+            await db.save_job_result(job_id, result)
+            await db.update_job_progress(job_id, indicator_id, "complete")
 
+        # -- Process non-batch indicators --
+        for indicator_id, indicator in process_indicators:
+            await db.update_job_progress(job_id, indicator_id, "processing")
+            result = await indicator.process(
+                job.request.aoi,
+                job.request.time_range,
+                season_months=job.request.season_months(),
+            )
+            spatial = indicator.get_spatial_data()
+            if spatial is not None:
+                spatial_cache[indicator_id] = spatial
             await db.save_job_result(job_id, result)
             await db.update_job_progress(job_id, indicator_id, "complete")