diff --git a/backend/app/services/ingest_core.py b/backend/app/services/ingest_core.py index 5cfaf5c..8bc9772 100644 --- a/backend/app/services/ingest_core.py +++ b/backend/app/services/ingest_core.py @@ -240,6 +240,15 @@ class Ingester: ), ) + # #899 L1: emit run milestones through the real logger (not only the + # in-memory log_lines → DownloadResult.stdout, which is persisted to the + # DownloadEvent ONLY at phase 3). A worker SIGKILL/OOM/hard-time-limit + # mid-walk would otherwise leave NO trace; these land in the container log + # in real time regardless of whether the event gets finalized. + log.info( + "%s ingest START (%s): source=%s campaign=%s resume_cursor=%s", + self._platform, mode, source_id, campaign_id, resume_cursor, + ) try: for post, included, page_cursor in self.client.iter_posts( campaign_id, cursor=resume_cursor @@ -249,6 +258,15 @@ class Ingester: # after it. Carried as DownloadResult.cursor (plan #704). if page_cursor and page_cursor != emitted_cursor: emitted_cursor = page_cursor + # #899 L1: a per-page breadcrumb in the container log (pages can + # be minutes apart on image-dense backfills) — survives a worker + # kill so the operator sees how far a since-died walk got. + log.info( + "%s ingest progress (%s, source=%s): posts=%d downloaded=%d " + "skipped=%d errors=%d quarantined=%d gated=%d cursor=%s", + self._platform, mode, source_id, posts_processed, downloaded, + skipped_count, errors, quarantined, gated_skipped, emitted_cursor, + ) # plan #705 #6: persist the cursor at each page boundary so a # worker SIGKILL mid-chunk resumes near the crash, not the # chunk start. (phase 3 still writes the final cursor — same @@ -450,6 +468,10 @@ class Ingester: # posts), so don't re-write them — return PARTIAL (reads as "ok/progress", # the lifecycle no-ops since state is gone) instead of a false "complete". if stopped: + log.info( + "%s ingest STOPPED by operator (%s, source=%s): %d file(s) this chunk", + self._platform, mode, source_id, downloaded, + ) return _result( success=False, return_code=-1, error_type=ErrorType.PARTIAL, @@ -467,7 +489,7 @@ class Ingester: log_lines.append(f"{quarantined} media item(s) quarantined (invalid)") if dead_lettered: log_lines.append(f"{dead_lettered} media item(s) skipped (dead-lettered)") - log_lines.append( + summary = ( f"{self._platform} ingest ({mode}): {downloaded} downloaded, " f"{skipped_count} skipped, {quarantined} quarantined, " f"{dead_lettered} dead-lettered, {errors} error(s), " @@ -481,6 +503,9 @@ class Ingester: + (", reached end" if reached_bottom else "") + (", time-boxed" if budget_hit else "") ) + log_lines.append(summary) + # #899 L1: also to the container log (survives event-finalization failure). + log.info("%s (source=%s)", summary, source_id) if budget_hit: # A chunk that hit its time-box but made forward progress is a