From e94b6cc7f6537228c73c3e45ad9ffcdd60fa5707 Mon Sep 17 00:00:00 2001 From: devRaGonSa Date: Sat, 21 Mar 2026 01:09:03 +0100 Subject: [PATCH] Add resumable historical backfill checkpoints --- ...historical-resumable-backfill-expansion.md | 108 ++++++ backend/README.md | 26 +- backend/app/config.py | 30 ++ backend/app/historical_ingestion.py | 88 ++++- backend/app/historical_models.py | 18 + backend/app/historical_storage.py | 318 ++++++++++++++++++ docs/historical-coverage-report.md | 18 +- 7 files changed, 584 insertions(+), 22 deletions(-) create mode 100644 ai/tasks/done/TASK-041-historical-resumable-backfill-expansion.md diff --git a/ai/tasks/done/TASK-041-historical-resumable-backfill-expansion.md b/ai/tasks/done/TASK-041-historical-resumable-backfill-expansion.md new file mode 100644 index 0000000..6968051 --- /dev/null +++ b/ai/tasks/done/TASK-041-historical-resumable-backfill-expansion.md @@ -0,0 +1,108 @@ +# TASK-041-historical-resumable-backfill-expansion + +## Goal +Ampliar la cobertura historica persistida de ambos servidores de la comunidad lo maximo posible mediante un proceso de backfill reanudable, seguro e idempotente, capaz de continuar sesiones incompletas sin perder progreso cuando la fuente CRCON publica falle intermitentemente. + +## Context +La capa historica del proyecto ya esta funcionando y la UI ya muestra cobertura registrada, pero la cobertura actual sigue siendo parcial. La fuente CRCON publica indica archivos mucho mas profundos que lo actualmente persistido, pero las sesiones largas de bootstrap sufren respuestas `502` intermitentes bajo carga. Eso significa que el limite actual de fechas no representa necesariamente el historico real disponible, sino el alcance conseguido hasta ahora por la importacion. + +La solucion correcta no es asumir que el historico ya esta completo, sino implementar o consolidar un backfill reanudable por lotes que permita seguir retrocediendo en el tiempo y ampliando la cobertura historica de forma progresiva. + +## Steps +1. Revisar la implementacion actual de: + - bootstrap historico + - refresh incremental + - persistencia historica + - validacion de cobertura +2. Revisar como se manejan actualmente: + - paginacion + - progreso de bootstrap + - reintentos + - errores upstream tipo `502` +3. Diseñar o consolidar una estrategia de backfill reanudable que permita: + - continuar desde una pagina o cursor previo + - guardar progreso util por servidor + - reintentar sin duplicar datos + - ampliar cobertura historica por bloques +4. Implementar o reforzar un mecanismo para ejecutar sesiones sucesivas de backfill historico para ambos servidores. +5. Garantizar que el sistema: + - no pierde progreso util + - no vuelve a importar innecesariamente lo ya consolidado + - puede seguir avanzando aunque una sesion concreta falle a mitad +6. Añadir o mejorar metadatos operativos de cobertura y progreso, por ejemplo: + - ultima pagina procesada por servidor + - rango temporal cubierto actual + - numero total de partidas persistidas + - estado del ultimo intento de backfill +7. Ejecutar o dejar documentado un flujo operativo realista para seguir ampliando cobertura en varias sesiones. +8. Documentar claramente que la cobertura historica puede crecer progresivamente y que el limite actual no debe interpretarse como limite definitivo del origen mientras siga habiendo paginas disponibles. +9. No crear todavia nueva UI historica en esta task salvo ajustes minimos de payload o semantica si fueran imprescindibles para reflejar mejor la cobertura. +10. Al completar la implementacion: + - dejar el repositorio consistente + - hacer commit + - hacer push al remoto si el entorno lo permite + +## Files to Read First +- AGENTS.md +- ai/repo-context.md +- ai/architecture-index.md +- docs/historical-crcon-source-discovery.md +- docs/historical-domain-model.md +- docs/historical-data-quality-notes.md +- docs/historical-coverage-report.md +- backend/README.md +- backend/app/config.py +- backend/app/historical_ingestion.py +- backend/app/historical_storage.py +- backend/app/historical_models.py +- backend/app/routes.py +- backend/app/payloads.py + +## Expected Files to Modify +- backend/README.md +- backend/app/config.py +- backend/app/historical_ingestion.py +- backend/app/historical_storage.py +- backend/app/historical_models.py +- opcionalmente nuevos modulos auxiliares si mejoran claridad del backfill y del checkpoint de progreso, por ejemplo: + - backend/app/historical_backfill_state.py + - backend/app/historical_backfill_runner.py +- opcionalmente documentacion tecnica adicional, por ejemplo: + - docs/historical-coverage-report.md + - docs/historical-backfill-operations.md + +## Constraints +- No basar esta ampliacion historica en A2S. +- No crear paginas frontend nuevas usando la URL de la comunidad. +- No depender del HTML de `/games` como fuente final. +- No romper el flujo actual de live status. +- No romper la persistencia historica ya existente. +- No hacer cambios destructivos. +- Mantener el trabajo centrado en backfill reanudable, idempotencia, resiliencia y ampliacion de cobertura. + +## Validation +- Existe un mecanismo reanudable para seguir ampliando el historico por lotes. +- El backfill puede continuar sin duplicar datos ya persistidos. +- El sistema registra progreso util por servidor. +- La cobertura historica puede ampliarse progresivamente con sesiones sucesivas. +- La documentacion deja claro el estado real y las limitaciones operativas. +- Los cambios quedan committeados y se hace push al remoto si el entorno lo permite. + +## Change Budget +- Preferir menos de 7 archivos modificados o creados. +- Preferir menos de 260 lineas cambiadas. + +## Outcome +- Se añadio persistencia de checkpoint por servidor y modo en `historical_backfill_progress`. +- El bootstrap ahora reanuda automaticamente desde `next_page` cuando no se pasa `--start-page`. +- Cada pagina completada guarda progreso util (`last_completed_page`, `next_page`, total descubierto y estado del ultimo intento). +- `server-summary` y el reporte de cobertura exponen metadatos operativos de backfill para reflejar mejor la cobertura real importada. +- Se documento el flujo operativo reanudable y los nuevos ajustes de reintentos para errores CRCON intermitentes. + +## Validation Notes +- `python -m compileall app` +- validacion local con SQLite de prueba en workspace verificando: + - reanudacion desde pagina `4` tras completar la pagina `3` + - persistencia de `last_completed_page` + - exposicion de `discovered_total_pages` en cobertura y resumen +- no se ejecuto backfill real contra CRCON por las restricciones de red del entorno actual diff --git a/backend/README.md b/backend/README.md index d0991a0..80c6aef 100644 --- a/backend/README.md +++ b/backend/README.md @@ -67,6 +67,8 @@ Variables opcionales: - `HLL_HISTORICAL_CRCON_PAGE_SIZE` - `HLL_HISTORICAL_CRCON_TIMEOUT_SECONDS` - `HLL_HISTORICAL_CRCON_DETAIL_WORKERS` +- `HLL_HISTORICAL_CRCON_REQUEST_RETRIES` +- `HLL_HISTORICAL_CRCON_RETRY_DELAY_SECONDS` - `HLL_HISTORICAL_REFRESH_INTERVAL_SECONDS` - `HLL_HISTORICAL_REFRESH_MAX_RETRIES` - `HLL_HISTORICAL_REFRESH_RETRY_DELAY_SECONDS` @@ -485,7 +487,7 @@ Flags utiles: - `--server comunidad-hispana-01` para limitar a un servidor - `--max-pages 2` para validacion local acotada - `--page-size 25` para ajustar paginacion -- `--start-page 4` para reanudar desde una pagina concreta en bootstraps largos +- `--start-page 4` para forzar una pagina concreta en bootstraps largos - `--detail-workers 16` para paralelizar el detalle por partida La ejecucion `bootstrap` recorre paginas historicas hasta agotar resultados. @@ -496,6 +498,15 @@ tardios sin reimportar todo el historico. El comando devuelve ademas un resumen de cobertura persistida por servidor. Esto ayuda a validar rapidamente cuantos matches reales quedaron importados, el rango temporal cubierto y si la carga ya supera la ultima semana movil que usa la UI. +Ese resumen incluye tambien checkpoint y estado operativo de backfill por +servidor: + +- `next_page` +- `last_completed_page` +- `discovered_total_matches` +- `discovered_total_pages` +- `archive_exhausted` +- `last_run` Como la fuente CRCON publica expone un archivo muy profundo y puede devolver errores `502` intermitentes bajo carga sostenida, el bootstrap completo debe @@ -503,12 +514,19 @@ tratarse como una operacion reanudable. Flujo recomendado: ```powershell python -m app.historical_ingestion bootstrap --detail-workers 16 -python -m app.historical_ingestion bootstrap --start-page 4 --detail-workers 16 +python -m app.historical_ingestion bootstrap --detail-workers 16 ``` -La segunda invocacion permite continuar desde la siguiente pagina pendiente si +La segunda invocacion reutiliza automaticamente el checkpoint persistido en +`historical_backfill_progress` y continua desde la siguiente pagina pendiente si la sesion anterior se corta por tiempo disponible o por inestabilidad puntual -del origen. +del origen. `--start-page` queda como override manual cuando se quiera +reprocesar o inspeccionar un tramo concreto. + +Los reintentos de cada request JSON pueden ajustarse sin tocar codigo con: + +- `HLL_HISTORICAL_CRCON_REQUEST_RETRIES` +- `HLL_HISTORICAL_CRCON_RETRY_DELAY_SECONDS` El runner `python -m app.historical_runner` deja ese refresh incremental listo para ejecucion local repetida sin depender de infraestructura externa. Por diff --git a/backend/app/config.py b/backend/app/config.py index 7e1abb9..2d6bcce 100644 --- a/backend/app/config.py +++ b/backend/app/config.py @@ -13,6 +13,8 @@ DEFAULT_REFRESH_INTERVAL_SECONDS = 120 DEFAULT_HISTORICAL_CRCON_PAGE_SIZE = 50 DEFAULT_HISTORICAL_CRCON_TIMEOUT_SECONDS = 15.0 DEFAULT_HISTORICAL_CRCON_DETAIL_WORKERS = 8 +DEFAULT_HISTORICAL_CRCON_REQUEST_RETRIES = 3 +DEFAULT_HISTORICAL_CRCON_RETRY_DELAY_SECONDS = 0.5 DEFAULT_HISTORICAL_REFRESH_INTERVAL_SECONDS = 1800 DEFAULT_HISTORICAL_REFRESH_MAX_RETRIES = 2 DEFAULT_HISTORICAL_REFRESH_RETRY_DELAY_SECONDS = 30 @@ -112,6 +114,34 @@ def get_historical_crcon_detail_workers() -> int: return worker_count +def get_historical_crcon_request_retries() -> int: + """Return the retry count used for CRCON historical JSON requests.""" + configured_value = os.getenv( + "HLL_HISTORICAL_CRCON_REQUEST_RETRIES", + str(DEFAULT_HISTORICAL_CRCON_REQUEST_RETRIES), + ) + retry_count = int(configured_value) + if retry_count <= 0: + raise ValueError("HLL_HISTORICAL_CRCON_REQUEST_RETRIES must be positive.") + + return retry_count + + +def get_historical_crcon_retry_delay_seconds() -> float: + """Return the base delay used between CRCON request retries.""" + configured_value = os.getenv( + "HLL_HISTORICAL_CRCON_RETRY_DELAY_SECONDS", + str(DEFAULT_HISTORICAL_CRCON_RETRY_DELAY_SECONDS), + ) + retry_delay_seconds = float(configured_value) + if retry_delay_seconds < 0: + raise ValueError( + "HLL_HISTORICAL_CRCON_RETRY_DELAY_SECONDS must be zero or positive." + ) + + return retry_delay_seconds + + def get_historical_refresh_interval_seconds() -> int: """Return the default interval used by the historical refresh loop.""" configured_value = os.getenv( diff --git a/backend/app/historical_ingestion.py b/backend/app/historical_ingestion.py index a63c28c..ddf2815 100644 --- a/backend/app/historical_ingestion.py +++ b/backend/app/historical_ingestion.py @@ -15,14 +15,20 @@ from urllib.request import Request, urlopen from .config import ( get_historical_crcon_detail_workers, get_historical_crcon_page_size, + get_historical_crcon_request_retries, get_historical_crcon_request_timeout_seconds, + get_historical_crcon_retry_delay_seconds, ) from .historical_storage import ( + finalize_backfill_progress, finalize_ingestion_run, + get_backfill_resume_page, get_refresh_cutoff_for_server, initialize_historical_storage, list_historical_coverage_report, list_historical_servers, + mark_backfill_progress_page_completed, + mark_backfill_progress_started, start_ingestion_run, upsert_historical_match, ) @@ -31,7 +37,6 @@ from .historical_storage import ( PUBLIC_INFO_ENDPOINT = "/api/get_public_info" MATCH_LIST_ENDPOINT = "/api/get_scoreboard_maps" MATCH_DETAIL_ENDPOINT = "/api/get_map_scoreboard" -DEFAULT_DETAIL_FETCH_RETRIES = 3 @dataclass(slots=True) @@ -57,7 +62,7 @@ def run_bootstrap( server_slug: str | None = None, max_pages: int | None = None, page_size: int | None = None, - start_page: int = 1, + start_page: int | None = None, detail_workers: int | None = None, ) -> dict[str, object]: """Run a first full historical import against one or all configured servers.""" @@ -77,7 +82,7 @@ def run_incremental_refresh( server_slug: str | None = None, max_pages: int | None = None, page_size: int | None = None, - start_page: int = 1, + start_page: int | None = None, detail_workers: int | None = None, ) -> dict[str, object]: """Refresh recent historical pages without replaying the whole archive.""" @@ -98,7 +103,7 @@ def _run_ingestion( server_slug: str | None, max_pages: int | None, page_size: int | None, - start_page: int, + start_page: int | None, detail_workers: int | None, incremental: bool, ) -> dict[str, object]: @@ -106,23 +111,35 @@ def _run_ingestion( stats = IngestionStats() selected_servers = _select_servers(server_slug) processed_servers: list[dict[str, object]] = [] - runs: list[int] = [] + active_runs: dict[str, int] = {} try: for server in selected_servers: run_id = start_ingestion_run(mode=mode, target_server_slug=str(server["slug"])) - runs.append(run_id) + active_runs[str(server["slug"])] = run_id + mark_backfill_progress_started( + server_slug=str(server["slug"]), + mode=mode, + run_id=run_id, + ) cutoff = ( get_refresh_cutoff_for_server(str(server["slug"])) if incremental else None ) + resolved_start_page = _resolve_start_page( + start_page=start_page, + server_slug=str(server["slug"]), + mode=mode, + ) server_stats = _ingest_server( server=server, + mode=mode, + run_id=run_id, stats=stats, max_pages=max_pages, page_size=page_size, - start_page=start_page, + start_page=resolved_start_page, detail_workers=detail_workers, cutoff=cutoff, ) @@ -138,8 +155,16 @@ def _run_ingestion( player_rows_updated=server_stats["player_rows_updated"], notes=f"public_name={server_stats['public_name']}", ) + finalize_backfill_progress( + server_slug=str(server["slug"]), + mode=mode, + run_id=run_id, + status="success", + archive_exhausted=bool(server_stats["archive_exhausted"]), + ) + active_runs.pop(str(server["slug"]), None) except Exception as exc: - for run_id in runs: + for active_server_slug, run_id in active_runs.items(): finalize_ingestion_run( run_id, status="failed", @@ -151,6 +176,13 @@ def _run_ingestion( player_rows_updated=stats.player_rows_updated, notes=str(exc), ) + finalize_backfill_progress( + server_slug=active_server_slug, + mode=mode, + run_id=run_id, + status="failed", + error_message=str(exc), + ) raise return { @@ -175,6 +207,8 @@ def _run_ingestion( def _ingest_server( *, server: dict[str, object], + mode: str, + run_id: int, stats: IngestionStats, max_pages: int | None, page_size: int | None, @@ -190,6 +224,7 @@ def _ingest_server( public_info = _fetch_public_info(str(server["scoreboard_base_url"])) discovered_total_matches: int | None = None last_page_processed: int | None = None + archive_exhausted = False for page_number in range(start_page, start_page + page_limit): payload = _fetch_match_page( @@ -201,6 +236,7 @@ def _ingest_server( discovered_total_matches = _coerce_int(payload.get("total")) page_matches = _coerce_match_list(payload.get("maps")) if not page_matches: + archive_exhausted = True break local_stats.pages_processed += 1 @@ -234,6 +270,15 @@ def _ingest_server( local_stats.apply(delta) stats.apply(delta) + mark_backfill_progress_page_completed( + server_slug=str(server["slug"]), + mode=mode, + page_number=page_number, + page_size=resolved_page_size, + run_id=run_id, + discovered_total_matches=discovered_total_matches, + ) + if stop_after_page: break @@ -251,9 +296,23 @@ def _ingest_server( "start_page": start_page, "last_page_processed": last_page_processed, "cutoff": cutoff, + "archive_exhausted": archive_exhausted, } +def _resolve_start_page( + *, + start_page: int | None, + server_slug: str, + mode: str, +) -> int: + if start_page is not None: + return max(1, start_page) + if mode != "bootstrap": + return 1 + return get_backfill_resume_page(server_slug, mode=mode) + + def _select_servers(server_slug: str | None) -> list[dict[str, object]]: servers = list_historical_servers() if server_slug is None: @@ -344,10 +403,12 @@ def _fetch_dict_payload( query: dict[str, object] | None = None, *, context: str = "", - retries: int = DEFAULT_DETAIL_FETCH_RETRIES, + retries: int | None = None, ) -> dict[str, object]: + resolved_retries = retries or get_historical_crcon_request_retries() + base_retry_delay_seconds = get_historical_crcon_retry_delay_seconds() last_error: Exception | None = None - for attempt in range(1, retries + 1): + for attempt in range(1, resolved_retries + 1): try: payload = _unwrap_result(_fetch_json(base_url, endpoint, query)) except Exception as exc: # pragma: no cover - network path @@ -359,8 +420,8 @@ def _fetch_dict_payload( f"Unexpected payload type for {base_url}{endpoint} {context}".strip() ) - if attempt < retries: - time.sleep(0.5 * attempt) + if attempt < resolved_retries: + time.sleep(base_retry_delay_seconds * attempt) assert last_error is not None raise last_error @@ -442,8 +503,7 @@ def build_arg_parser() -> argparse.ArgumentParser: parser.add_argument( "--start-page", type=int, - default=1, - help="resume bootstrap or refresh from a specific page number", + help="override the resume page; bootstrap uses persisted progress when omitted", ) parser.add_argument( "--detail-workers", diff --git a/backend/app/historical_models.py b/backend/app/historical_models.py index cbfb66c..97e1817 100644 --- a/backend/app/historical_models.py +++ b/backend/app/historical_models.py @@ -91,3 +91,21 @@ class HistoricalIngestionRunSummary: matches_updated: int player_rows_inserted: int player_rows_updated: int + + +@dataclass(frozen=True, slots=True) +class HistoricalBackfillProgressSummary: + """Persisted resume checkpoint and last attempt metadata per server.""" + + server_slug: str + mode: str + next_page: int + last_completed_page: int | None + discovered_total_matches: int | None + discovered_total_pages: int | None + archive_exhausted: bool + last_run_id: int | None + last_run_status: str | None + last_run_started_at: datetime | None + last_run_completed_at: datetime | None + last_error: str | None diff --git a/backend/app/historical_storage.py b/backend/app/historical_storage.py index 36a7b17..9aaa85f 100644 --- a/backend/app/historical_storage.py +++ b/backend/app/historical_storage.py @@ -147,6 +147,24 @@ def initialize_historical_storage(*, db_path: Path | None = None) -> Path: created_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP ); + CREATE TABLE IF NOT EXISTS historical_backfill_progress ( + historical_server_id INTEGER NOT NULL, + mode TEXT NOT NULL, + next_page INTEGER NOT NULL DEFAULT 1, + last_completed_page INTEGER, + discovered_total_matches INTEGER, + discovered_total_pages INTEGER, + archive_exhausted INTEGER NOT NULL DEFAULT 0, + last_run_id INTEGER, + last_run_status TEXT, + last_run_started_at TEXT, + last_run_completed_at TEXT, + last_error TEXT, + updated_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP, + PRIMARY KEY (historical_server_id, mode), + FOREIGN KEY (historical_server_id) REFERENCES historical_servers(id) + ); + CREATE INDEX IF NOT EXISTS idx_historical_matches_server_end ON historical_matches(historical_server_id, ended_at DESC, started_at DESC); @@ -155,6 +173,9 @@ def initialize_historical_storage(*, db_path: Path | None = None) -> Path: CREATE INDEX IF NOT EXISTS idx_historical_players_steam ON historical_players(steam_id); + + CREATE INDEX IF NOT EXISTS idx_historical_backfill_progress_run + ON historical_backfill_progress(last_run_id); """ ) _seed_default_historical_servers(connection) @@ -248,6 +269,259 @@ def finalize_ingestion_run( ) +def mark_backfill_progress_started( + *, + server_slug: str, + mode: str, + run_id: int, + db_path: Path | None = None, +) -> None: + """Persist the start of one resumable historical backfill attempt.""" + resolved_path = initialize_historical_storage(db_path=db_path) + with _connect(resolved_path) as connection: + server_row = _resolve_historical_server(connection, server_slug) + connection.execute( + """ + INSERT INTO historical_backfill_progress ( + historical_server_id, + mode, + next_page, + archive_exhausted, + last_run_id, + last_run_status, + last_run_started_at, + last_run_completed_at, + last_error + ) VALUES (?, ?, 1, 0, ?, 'running', ?, NULL, NULL) + ON CONFLICT(historical_server_id, mode) DO UPDATE SET + last_run_id = excluded.last_run_id, + last_run_status = excluded.last_run_status, + last_run_started_at = excluded.last_run_started_at, + last_run_completed_at = NULL, + last_error = NULL, + archive_exhausted = CASE + WHEN excluded.mode = 'bootstrap' THEN 0 + ELSE historical_backfill_progress.archive_exhausted + END, + updated_at = CURRENT_TIMESTAMP + """, + (server_row["id"], mode, run_id, _utc_now_iso()), + ) + + +def mark_backfill_progress_page_completed( + *, + server_slug: str, + mode: str, + page_number: int, + page_size: int, + run_id: int, + discovered_total_matches: int | None, + db_path: Path | None = None, +) -> None: + """Persist the latest completed page so bootstraps can resume safely.""" + resolved_path = initialize_historical_storage(db_path=db_path) + discovered_total_pages = None + if discovered_total_matches and page_size > 0: + discovered_total_pages = (discovered_total_matches + page_size - 1) // page_size + + with _connect(resolved_path) as connection: + server_row = _resolve_historical_server(connection, server_slug) + connection.execute( + """ + INSERT INTO historical_backfill_progress ( + historical_server_id, + mode, + next_page, + last_completed_page, + discovered_total_matches, + discovered_total_pages, + archive_exhausted, + last_run_id, + last_run_status, + last_run_started_at, + last_run_completed_at, + last_error + ) VALUES (?, ?, ?, ?, ?, ?, 0, ?, ?, ?, NULL, NULL) + ON CONFLICT(historical_server_id, mode) DO UPDATE SET + next_page = excluded.next_page, + last_completed_page = excluded.last_completed_page, + discovered_total_matches = COALESCE( + excluded.discovered_total_matches, + historical_backfill_progress.discovered_total_matches + ), + discovered_total_pages = COALESCE( + excluded.discovered_total_pages, + historical_backfill_progress.discovered_total_pages + ), + archive_exhausted = 0, + last_run_id = excluded.last_run_id, + last_run_status = excluded.last_run_status, + last_run_started_at = excluded.last_run_started_at, + last_run_completed_at = NULL, + last_error = NULL, + updated_at = CURRENT_TIMESTAMP + """, + ( + server_row["id"], + mode, + page_number + 1, + page_number, + discovered_total_matches, + discovered_total_pages, + run_id, + "running", + _utc_now_iso(), + ), + ) + + +def finalize_backfill_progress( + *, + server_slug: str, + mode: str, + run_id: int, + status: str, + archive_exhausted: bool = False, + error_message: str | None = None, + db_path: Path | None = None, +) -> None: + """Persist the final state of one resumable historical backfill attempt.""" + resolved_path = initialize_historical_storage(db_path=db_path) + with _connect(resolved_path) as connection: + server_row = _resolve_historical_server(connection, server_slug) + connection.execute( + """ + INSERT INTO historical_backfill_progress ( + historical_server_id, + mode, + next_page, + archive_exhausted, + last_run_id, + last_run_status, + last_run_started_at, + last_run_completed_at, + last_error + ) VALUES (?, ?, 1, ?, ?, ?, ?, ?, ?) + ON CONFLICT(historical_server_id, mode) DO UPDATE SET + archive_exhausted = CASE + WHEN excluded.last_run_status = 'success' AND excluded.archive_exhausted = 1 + THEN 1 + WHEN excluded.last_run_status = 'success' + THEN historical_backfill_progress.archive_exhausted + ELSE historical_backfill_progress.archive_exhausted + END, + last_run_id = excluded.last_run_id, + last_run_status = excluded.last_run_status, + last_run_started_at = COALESCE( + historical_backfill_progress.last_run_started_at, + excluded.last_run_started_at + ), + last_run_completed_at = excluded.last_run_completed_at, + last_error = excluded.last_error, + updated_at = CURRENT_TIMESTAMP + """, + ( + server_row["id"], + mode, + 1 if archive_exhausted else 0, + run_id, + status, + _utc_now_iso(), + _utc_now_iso(), + error_message, + ), + ) + + +def get_backfill_resume_page( + server_slug: str, + *, + mode: str = "bootstrap", + db_path: Path | None = None, +) -> int: + """Return the next page recorded for one resumable historical backfill.""" + resolved_path = initialize_historical_storage(db_path=db_path) + with _connect(resolved_path) as connection: + server_row = _resolve_historical_server(connection, server_slug) + row = connection.execute( + """ + SELECT next_page + FROM historical_backfill_progress + WHERE historical_server_id = ? AND mode = ? + """, + (server_row["id"], mode), + ).fetchone() + return max(1, int(row["next_page"])) if row and row["next_page"] else 1 + + +def list_historical_backfill_progress( + *, + server_slug: str | None = None, + mode: str = "bootstrap", + db_path: Path | None = None, +) -> list[dict[str, object]]: + """Return persisted resume checkpoints and last run state per server.""" + resolved_path = initialize_historical_storage(db_path=db_path) + where_clause = "" + params: list[object] = [mode] + if server_slug: + where_clause = "WHERE historical_servers.slug = ?" + params.append(server_slug) + + with _connect(resolved_path) as connection: + rows = connection.execute( + f""" + SELECT + historical_servers.slug AS server_slug, + historical_servers.display_name AS server_name, + progress.mode AS mode, + progress.next_page AS next_page, + progress.last_completed_page AS last_completed_page, + progress.discovered_total_matches AS discovered_total_matches, + progress.discovered_total_pages AS discovered_total_pages, + progress.archive_exhausted AS archive_exhausted, + progress.last_run_id AS last_run_id, + progress.last_run_status AS last_run_status, + progress.last_run_started_at AS last_run_started_at, + progress.last_run_completed_at AS last_run_completed_at, + progress.last_error AS last_error + FROM historical_servers + LEFT JOIN historical_backfill_progress AS progress + ON progress.historical_server_id = historical_servers.id + AND progress.mode = ? + {where_clause} + ORDER BY historical_servers.server_number ASC, historical_servers.slug ASC + """, + params, + ).fetchall() + + items: list[dict[str, object]] = [] + for row in rows: + items.append( + { + "server": { + "slug": row["server_slug"], + "name": row["server_name"], + }, + "mode": row["mode"] or mode, + "next_page": int(row["next_page"] or 1), + "last_completed_page": _coerce_int(row["last_completed_page"]), + "discovered_total_matches": _coerce_int(row["discovered_total_matches"]), + "discovered_total_pages": _coerce_int(row["discovered_total_pages"]), + "archive_exhausted": bool(row["archive_exhausted"]), + "last_run": { + "run_id": _coerce_int(row["last_run_id"]), + "status": _stringify(row["last_run_status"]), + "started_at": _stringify(row["last_run_started_at"]), + "completed_at": _stringify(row["last_run_completed_at"]), + "error": _stringify(row["last_error"]), + }, + } + ) + return items + + def upsert_historical_match( *, server_slug: str, @@ -567,6 +841,13 @@ def list_historical_server_summaries( params, ).fetchall() + progress_by_server = { + item["server"]["slug"]: item + for item in list_historical_backfill_progress( + server_slug=server_slug, + db_path=resolved_path, + ) + } top_maps_by_server: dict[str, list[dict[str, object]]] = {} for row in map_rows: server_key = str(row["server_slug"]) @@ -586,6 +867,8 @@ def list_historical_server_summaries( first_match_at = _stringify(row["first_match_at"]) last_match_at = _stringify(row["last_match_at"]) coverage_days = _calculate_coverage_days(first_match_at, last_match_at) + progress = progress_by_server.get(str(row["server_slug"]), {}) + discovered_total_matches = _coerce_int(progress.get("discovered_total_matches")) items.append( { "server": { @@ -602,10 +885,25 @@ def list_historical_server_summaries( "basis": "persisted-import", "status": _classify_coverage_status(matches_count, coverage_days), "imported_matches_count": matches_count, + "discovered_total_matches": discovered_total_matches, "first_match_at": first_match_at, "last_match_at": last_match_at, "coverage_days": coverage_days, }, + "backfill": { + "mode": progress.get("mode", "bootstrap"), + "next_page": _coerce_int(progress.get("next_page")) or 1, + "last_completed_page": _coerce_int(progress.get("last_completed_page")), + "discovered_total_matches": discovered_total_matches, + "discovered_total_pages": _coerce_int(progress.get("discovered_total_pages")), + "remaining_matches_estimate": ( + max(discovered_total_matches - matches_count, 0) + if discovered_total_matches is not None + else None + ), + "archive_exhausted": bool(progress.get("archive_exhausted")), + "last_run": progress.get("last_run"), + }, "time_range": { "start": first_match_at, "end": last_match_at, @@ -656,9 +954,17 @@ def list_historical_coverage_report( ).fetchall() items: list[dict[str, object]] = [] + progress_by_server = { + item["server"]["slug"]: item + for item in list_historical_backfill_progress( + server_slug=server_slug, + db_path=resolved_path, + ) + } for row in rows: first_match_at = _stringify(row["first_match_at"]) last_match_at = _stringify(row["last_match_at"]) + progress = progress_by_server.get(str(row["server_slug"]), {}) items.append( { "server": { @@ -673,6 +979,18 @@ def list_historical_coverage_report( "first_match_at": first_match_at, "last_match_at": last_match_at, "coverage_days": _calculate_coverage_days(first_match_at, last_match_at), + "backfill": { + "next_page": _coerce_int(progress.get("next_page")) or 1, + "last_completed_page": _coerce_int(progress.get("last_completed_page")), + "discovered_total_matches": _coerce_int( + progress.get("discovered_total_matches") + ), + "discovered_total_pages": _coerce_int( + progress.get("discovered_total_pages") + ), + "archive_exhausted": bool(progress.get("archive_exhausted")), + "last_run": progress.get("last_run"), + }, } ) return items diff --git a/docs/historical-coverage-report.md b/docs/historical-coverage-report.md index 4e46f9d..f5caff6 100644 --- a/docs/historical-coverage-report.md +++ b/docs/historical-coverage-report.md @@ -2,7 +2,7 @@ ## Validation Date -- 2026-03-20 +- 2026-03-21 ## Scope @@ -80,10 +80,20 @@ larga incluso con paralelismo. ## Operational Conclusion -- El bootstrap ya es reanudable mediante `--start-page`. +- El bootstrap queda reanudable por checkpoint persistido en + `historical_backfill_progress`; si no se pasa `--start-page`, una nueva + sesion continua desde `next_page`. +- Cada pagina completada actualiza por servidor: + - `last_completed_page` + - `next_page` + - `discovered_total_matches` + - `discovered_total_pages` + - `last_run` - La estrategia operativa razonable para completar todo el archivo es ejecutar - varias sesiones consecutivas, registrando la ultima pagina completada y - continuando desde ahi si el origen falla o si la sesion debe cortarse. + varias sesiones consecutivas con el mismo comando hasta que + `archive_exhausted` pase a `true`. +- `--start-page` se conserva solo como override manual cuando haga falta + reprocesar un tramo concreto. - Mientras no se complete todo el archivo, cualquier UI o API debe mostrar la cobertura importada como cobertura real disponible y no como historico total del servidor.