From a787f79f9299042faee173d051a902f9a355798d Mon Sep 17 00:00:00 2001 From: emi Date: Thu, 23 Jul 2026 22:47:26 +0000 Subject: [PATCH] Survive Contabo PUT throttling in the lake extractor (#22) --- backend/gen_era5_lake.py | 37 ++++++++++++++++++++++++------------- 1 file changed, 24 insertions(+), 13 deletions(-) diff --git a/backend/gen_era5_lake.py b/backend/gen_era5_lake.py index c7be634..891249b 100644 --- a/backend/gen_era5_lake.py +++ b/backend/gen_era5_lake.py @@ -65,11 +65,16 @@ class _Sink: self._s3 = None if not self.local: import boto3 # noqa: PLC0415 - extractor-only dep + from botocore.config import Config # noqa: PLC0415 + # Contabo throttles bursts with 429s (seen live at 16 concurrent + # PUTs); adaptive mode rate-limits client-side and retries with + # backoff instead of dying after botocore's default 4 attempts. self._s3 = boto3.client( "s3", endpoint_url=self.cfg["endpoint"], region_name=self.cfg["region"], aws_access_key_id=self.cfg["access_key"], - aws_secret_access_key=self.cfg["secret_key"]) + aws_secret_access_key=self.cfg["secret_key"], + config=Config(retries={"max_attempts": 10, "mode": "adaptive"})) def put(self, key: str, df: pl.DataFrame) -> None: if self.local: @@ -147,7 +152,7 @@ def _write_tile(sink: _Sink, ti: int, tj: int, tile_df: pl.DataFrame) -> pl.Data for (year, month), mdf in parts.group_by(["year", "month"]): puts.append((era5lake.daily_part_key(ti, tj, year, month), mdf.drop(["year", "month"]).sort(["lat_idx", "lon_idx", "date"]))) - with concurrent.futures.ThreadPoolExecutor(max_workers=16) as pool: + with concurrent.futures.ThreadPoolExecutor(max_workers=8) as pool: for f in [pool.submit(sink.put, k, df) for k, df in puts]: f.result() # surface the first failure instead of hiding it return pl.DataFrame(rows) @@ -197,29 +202,35 @@ def main() -> None: done = set() if manifest is None or args.overwrite else \ set(manifest["tile"].unique().to_list()) - written = skipped = empty = 0 + written = skipped = empty = failed = 0 for n, (ti, tj) in enumerate(todo, 1): if f"{ti}_{tj}" in done: skipped += 1 continue t0 = time.time() - tile_df = _tile_frame(ds, ti, tj, end) - if tile_df is None: - empty += 1 + try: + tile_df = _tile_frame(ds, ti, tj, end) + if tile_df is None: + empty += 1 + continue + if args.dry_run: + print(f"[dry-run] tile {ti}_{tj}: {tile_df.height} rows " + f"({time.time() - t0:.1f}s); nothing written") + print(tile_df.head()) + return + rows = _write_tile(sink, ti, tj, tile_df) + except Exception as e: # noqa: BLE001 - keep going; the tile isn't in the + failed += 1 # manifest, so the next run retries it + print(f"[{n}/{len(todo)}] FAILED tile {ti}_{tj}: {e}") continue - if args.dry_run: - print(f"[dry-run] tile {ti}_{tj}: {tile_df.height} rows " - f"({time.time() - t0:.1f}s); nothing written") - print(tile_df.head()) - return - rows = _write_tile(sink, ti, tj, tile_df) manifest = rows if manifest is None else pl.concat( [manifest.filter(pl.col("tile") != f"{ti}_{tj}"), rows]) sink.put(era5lake.MANIFEST_KEY, manifest) # after every tile: resumable written += 1 print(f"[{n}/{len(todo)}] tile {ti}_{tj}: {rows.height} points " f"in {time.time() - t0:.1f}s") - print(f"done: tiles written={written} skipped(done)={skipped} sea={empty}") + print(f"done: tiles written={written} skipped(done)={skipped} sea={empty} " + f"failed={failed}") if __name__ == "__main__":