Survive Contabo PUT throttling in the lake extractor (#22)

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,12 +202,13 @@ 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()
+        try:
             tile_df = _tile_frame(ds, ti, tj, end)
             if tile_df is None:
                 empty += 1
@@ -213,13 +219,18 @@ def main() -> None:
                 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
         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__":