ERA5 lake: bucket-hosted history primary + SQL indexer service

Move climate history to ERA5 served from our own object storage, with a
prod-only query service in front:

- data/era5lake.py: lake layout (per-point whole-record 1940+ serving files,
  a tile/year/month hive table, an Iceberg-style manifest) plus grid math and
  the read client (local dir -> lake service -> bucket).
- gen_era5_lake.py: tile-aligned extractor from the Earthmover Icechunk ERA5
  archive (one 86-year pull covers all 144 points of a chunk tile; resumable
  from the manifest; --cities / --tiles / --land).
- lake_app.py + THERMOGRAPH_ROLE=lake: the lake service. /history serves a
  point's parquet off a disk cache (11ms cold / 3ms warm in rehearsal);
  /query runs SELECT-only SQL on DuckDB over the hive table (79ms pruned
  aggregate, 88ms full scan of a 4.5M-row tile). httpfs is baked at image
  build.
- climate.py: history chain is now era5-lake -> NASA POWER -> Open-Meteo; the
  lake slice starts at START_DATE so grading windows are unchanged, and an
  unconfigured lake costs nothing (beta/LAN unchanged).
- stack: lake service (1..2 replicas behind the VIP, own cache volume) plus a
  second autoscaler instance (autoscale.sh gains TARGET_SERVICE); web/worker
  get THERMOGRAPH_LAKE_URL.
- seed_era5.py: constants corrected against the live store (icechunkV2,
  single/temporal, valid_time, ECMWF short names, pcodec).

Bucket creds (THERMOGRAPH_LAKE_S3_ACCESS_KEY/_SECRET_KEY) go in the prod
vault; until they land the lake stays healthy and everything falls through to
NASA exactly as before.

backend/Dockerfile

@@ -13,6 +13,9 @@ RUN apt-get update \
 # Install deps first so this layer caches across code-only changes.
 COPY requirements.txt /tmp/requirements.txt
 RUN pip install --no-cache-dir -r /tmp/requirements.txt
+# Bake DuckDB's httpfs extension so the lake role's S3 reads need no runtime
+# extension download (tasks may roll while egress is degraded).
+RUN python -c "import duckdb; duckdb.connect().execute('INSTALL httpfs')"
 
 COPY . /app/
 RUN chmod +x /app/deploy/entrypoint.sh

backend/data/climate.py

@@ -21,6 +21,7 @@ from core import audit
 from core import metrics
 import paths
 from data import climate_store
+from data import era5lake
 from data import meteostat
 from data import store
 
@@ -789,21 +790,32 @@ def _load_history(cell: dict) -> tuple[pl.DataFrame, dict]:
         def _serve_stale():
             return _with_doy(stale), {"cached": True, "stale": True, "cache_age_days": None}
 
-        # Fetch. NASA POWER is the primary source (keyless, independent of Open-Meteo);
-        # the gusts it lacks are filled from Meteostat inside _fetch_history_nasa.
-        # Open-Meteo is the fallback when NASA is unavailable — still guarded by its
-        # rate-limit cooldown. Both sources must return a plausibly-full span before
-        # being accepted and cached as a complete record (a short/partial response is
-        # rejected rather than held indefinitely).
+        # Fetch. The ERA5 lake is first when configured (true ERA5 with measured
+        # gusts, served from our own bucket — no third-party API in the path);
+        # unconfigured or missing-point cells fall through at zero cost. NASA
+        # POWER is next (keyless, independent of Open-Meteo; its missing gusts
+        # are filled from Meteostat inside _fetch_history_nasa), then Open-Meteo
+        # — still guarded by its rate-limit cooldown. Every source must return a
+        # plausibly-full span before being accepted and cached as a complete
+        # record (a short/partial response is rejected rather than held
+        # indefinitely).
         df = None
         source = None
         try:
-            fetched = _fetch_history_nasa(cell)
+            fetched = era5lake.fetch_history_lake(cell, start=START_DATE)
             if fetched.height >= MIN_ARCHIVE_DAYS:
                 df = fetched
-                source = "nasa-power"
-        except Exception:  # noqa: BLE001 - primary unavailable; fall through to Open-Meteo
+                source = "era5-lake"
+        except Exception:  # noqa: BLE001 - lake unconfigured/miss; fall through to NASA
             pass
+        if df is None:
+            try:
+                fetched = _fetch_history_nasa(cell)
+                if fetched.height >= MIN_ARCHIVE_DAYS:
+                    df = fetched
+                    source = "nasa-power"
+            except Exception:  # noqa: BLE001 - primary unavailable; fall through to Open-Meteo
+                pass
         if df is None and time.time() >= _archive_cooldown_until:
             try:
                 fetched = _fetch_history(cell)

backend/data/era5lake.py

@@ -0,0 +1,173 @@
+"""The ERA5 lake: layout, grid math, and read clients for the bucket-hosted archive.
+
+The lake is a set of parquet files in an S3-compatible bucket (Contabo Object
+Storage, ``era5-thermograph``), extracted once from the public Earthmover ERA5
+Icechunk archive by ``gen_era5_lake.py`` and read three ways:
+
+- the **lake service** (``lake_app.py``, prod-only Swarm service) serves point
+  histories and SQL-style queries with a local disk cache in front;
+- the **web/worker backend** asks the lake service first
+  (``THERMOGRAPH_LAKE_URL``), or reads the bucket directly when only S3
+  credentials are configured (beta/LAN, no lake service);
+- ad-hoc analysis hits the hive table straight from the bucket.
+
+Two layouts, one manifest (all under ``era5/``):
+
+- ``daily/tile=<ti>_<tj>/year=<Y>/month=<M>/part.parquet`` — the analytical
+  table, hive-partitioned by location/year/month. The location grain is a
+  12x12-point tile (~3°x3°), aligned to the Earthmover store's chunk grid; a
+  per-point x per-month grain would be ~170M objects, which no object store
+  enjoys.
+- ``points/lat=<i>/lon=<j>.parquet`` — the serving projection: one file per
+  ERA5 grid point holding its full daily history, so the hot path (one cell's
+  history) is a single GET.
+- ``manifest.parquet`` — Iceberg-manifest-style index of every point file:
+  grid indices, coordinates, row count, date span. Readers prune with it and
+  the extractor resumes from it.
+
+Columns everywhere: the finalized history-store schema (date, tmax, tmin,
+precip, wind, gust, humid, fmax, fmin, feels, doy — °F/in/mph/%, as
+``climate._finalize_approximated`` emits it), so a lake frame drops straight
+into ``climate._write_history_backed`` exactly like a NASA fetch does.
+"""
+import io
+import os
+
+import polars as pl
+
+# --- ERA5 native grid (0.25°, 721 x 1440, lat 90..-90, lon 0..359.75) ---------
+LAT_N = 721
+LON_N = 1440
+STEP = 0.25
+TILE = 12                      # matches the Earthmover chunk grid (8736h x 12 x 12)
+
+PREFIX = "era5"
+MANIFEST_KEY = f"{PREFIX}/manifest.parquet"
+
+MIN_LAKE_DAYS = 10_000         # a plausibly-full 1980+ record, same spirit as
+                               # climate.MIN_ARCHIVE_DAYS (rejects partial files)
+
+
+def to_idx(lat: float, lon: float) -> tuple[int, int]:
+    """Nearest ERA5 grid indices for a lat/lon. lat_idx runs north->south."""
+    i = round((90.0 - lat) / STEP)
+    j = round((lon % 360.0) / STEP) % LON_N
+    return min(max(i, 0), LAT_N - 1), j
+
+
+def to_coords(i: int, j: int) -> tuple[float, float]:
+    """Center lat/lon ([-180, 180) longitude) of grid indices."""
+    lon = j * STEP
+    return 90.0 - i * STEP, lon - 360.0 if lon >= 180.0 else lon
+
+
+def tile_of(i: int, j: int) -> tuple[int, int]:
+    return i // TILE, j // TILE
+
+
+def point_key(i: int, j: int) -> str:
+    return f"{PREFIX}/points/lat={i}/lon={j}.parquet"
+
+
+def daily_part_key(ti: int, tj: int, year: int, month: int) -> str:
+    return f"{PREFIX}/daily/tile={ti}_{tj}/year={year}/month={month:02d}/part.parquet"
+
+
+# --- Config (all optional: unset means "no lake configured") ------------------
+
+def lake_url() -> str:
+    """Lake service base URL (prod: the Swarm VIP), '' when not deployed."""
+    return os.environ.get("THERMOGRAPH_LAKE_URL", "").rstrip("/")
+
+
+def s3_config() -> "dict | None":
+    """Bucket access for direct reads/writes, None unless fully configured.
+    Contabo is path-style; region is a signing formality there ('default')."""
+    key = os.environ.get("THERMOGRAPH_LAKE_S3_ACCESS_KEY", "")
+    secret = os.environ.get("THERMOGRAPH_LAKE_S3_SECRET_KEY", "")
+    if not (key and secret):
+        return None
+    return {
+        "endpoint": os.environ.get("THERMOGRAPH_LAKE_S3_ENDPOINT",
+                                   "https://eu2.contabostorage.com"),
+        "bucket": os.environ.get("THERMOGRAPH_LAKE_S3_BUCKET", "era5-thermograph"),
+        "region": os.environ.get("THERMOGRAPH_LAKE_S3_REGION", "default"),
+        "access_key": key,
+        "secret_key": secret,
+    }
+
+
+def local_dir() -> str:
+    """Filesystem mirror of the lake layout — tests and offline dev reads."""
+    return os.environ.get("THERMOGRAPH_LAKE_LOCAL_DIR", "")
+
+
+def storage_options(cfg: dict) -> dict:
+    """polars/object_store options for scan_parquet against the bucket."""
+    return {
+        "aws_access_key_id": cfg["access_key"],
+        "aws_secret_access_key": cfg["secret_key"],
+        "aws_region": cfg["region"],
+        "endpoint_url": cfg["endpoint"],
+        # Contabo serves buckets on the path, not as a subdomain.
+        "aws_virtual_hosted_style_request": "false",
+    }
+
+
+# --- Point-history reads (the app's hot path) ----------------------------------
+
+class LakeUnavailable(Exception):
+    """No lake configured, point not in the lake, or the read failed."""
+
+
+def _read_local(i: int, j: int) -> pl.DataFrame:
+    path = os.path.join(local_dir(), point_key(i, j))
+    if not os.path.exists(path):
+        raise LakeUnavailable(f"no local lake file for point ({i}, {j})")
+    return pl.read_parquet(path)
+
+
+def _read_service(i: int, j: int) -> pl.DataFrame:
+    import httpx  # noqa: PLC0415 - keep module import light for the extractor venv
+
+    lat, lon = to_coords(i, j)
+    r = httpx.get(f"{lake_url()}/history", params={"lat": lat, "lon": lon},
+                  timeout=httpx.Timeout(20.0, connect=3.0))
+    if r.status_code == 404:
+        raise LakeUnavailable(f"point ({i}, {j}) not in the lake")
+    r.raise_for_status()
+    return pl.read_parquet(io.BytesIO(r.content))
+
+
+def _read_bucket(i: int, j: int, cfg: dict) -> pl.DataFrame:
+    try:
+        return pl.read_parquet(f"s3://{cfg['bucket']}/{point_key(i, j)}",
+                               storage_options=storage_options(cfg))
+    except Exception as e:  # noqa: BLE001 - object_store errors are not one type
+        raise LakeUnavailable(f"bucket read failed for point ({i}, {j}): {e}") from e
+
+
+def fetch_history_lake(cell: dict, start: "str | None" = None) -> pl.DataFrame:
+    """Finalized daily history for the ERA5 point nearest a grid cell, from the
+    first configured source: local dir -> lake service -> bucket. The lake
+    stores each point's whole 1940+ record; pass ``start`` (the caller's
+    grading window, e.g. climate.START_DATE) to slice — grading semantics stay
+    the caller's, the deep record stays in the lake. Raises LakeUnavailable
+    when unconfigured, missing, or implausibly short."""
+    import datetime  # noqa: PLC0415
+
+    i, j = to_idx(cell["center_lat"], cell["center_lon"])
+    if local_dir():
+        df = _read_local(i, j)
+    elif lake_url():
+        df = _read_service(i, j)
+    else:
+        cfg = s3_config()
+        if cfg is None:
+            raise LakeUnavailable("no lake configured")
+        df = _read_bucket(i, j, cfg)
+    if start is not None:
+        df = df.filter(pl.col("date") >= datetime.date.fromisoformat(start))
+    if df.height < MIN_LAKE_DAYS:
+        raise LakeUnavailable(f"lake point ({i}, {j}) is short: {df.height} rows")
+    return df.drop([c for c in ("lat_idx", "lon_idx") if c in df.columns])

backend/deploy/entrypoint.sh

@@ -62,6 +62,13 @@ run_migrations() {
   done
 }
 
+# The lake role serves the ERA5 bucket only — no database, so no migrations: a
+# lake replica must come up (and stay up) even when Postgres is down.
+if [ "${THERMOGRAPH_ROLE:-}" = "lake" ]; then
+  echo "==> Starting lake service on 0.0.0.0:${PORT:-8141} with ${WORKERS:-1} worker(s)"
+  exec uvicorn lake_app:app --host 0.0.0.0 --port "${PORT:-8141}" --workers "${WORKERS:-1}"
+fi
+
 # One-shot migrate mode (`entrypoint.sh migrate`, or THERMOGRAPH_MIGRATE_ONLY=1): run
 # the migration and exit — a one-shot task, so it runs once against the DB rather
 # than racing per web replica.

backend/gen_era5_lake.py

@@ -0,0 +1,218 @@
+"""Extract the ERA5 lake: Earthmover Icechunk archive -> the Thermograph bucket.
+
+Builds the lake that ``data/era5lake.py`` reads (layout documented there):
+per-point serving files, the tile/year/month hive table, and the manifest.
+Pulls are tile-aligned to the source's chunk grid (8736h x 12 x 12) — one
+46-year pull (~19s measured) decodes all 144 points of a tile, so per-point
+pulls would waste 144x the wire. Roughly: ~600 tiles cover the curated
+cities (hours), ~3000 cover land (a day or two); both resume from the
+manifest, so ctrl-C and re-run is always safe.
+
+    pip install -r requirements-seed.txt 'numcodecs[pcodec]' boto3
+    python gen_era5_lake.py --cities --dry-run      # one tile, no writes
+    python gen_era5_lake.py --cities                # tiles covering cities.json
+    python gen_era5_lake.py --tiles 20:30,40:60     # explicit tile-index window
+    python gen_era5_lake.py --land                  # every land-containing tile
+
+Destination: the bucket from THERMOGRAPH_LAKE_S3_* (see era5lake.s3_config),
+or a local mirror when THERMOGRAPH_LAKE_LOCAL_DIR is set (tests/rehearsal).
+Seed-only deps (icechunk/xarray/zarr) stay out of the app image, same policy
+as seed_era5.py.
+"""
+import argparse
+import datetime
+import io
+import os
+import sys
+import time
+
+import polars as pl
+
+from data import cities
+from data import climate
+from data import era5lake
+from seed_era5 import ERA5_BUCKET, ERA5_GROUP, ERA5_PREFIX, ERA5_REGION, ERA5_VARS, \
+    hourly_to_daily
+
+# The lake keeps each location's WHOLE available record (the store reaches
+# back to 1940); the app slices to its own grading window (climate.START_DATE)
+# at read time, so the extra decades are analytical surface, not a grading
+# change.
+START_DATE = "1940-01-01"
+
+
+def _open_store():
+    import icechunk           # noqa: PLC0415 - seed-only dep
+    import xarray as xr       # noqa: PLC0415
+
+    storage = icechunk.s3_storage(
+        bucket=ERA5_BUCKET, prefix=ERA5_PREFIX, region=ERA5_REGION, anonymous=True)
+    repo = icechunk.Repository.open(storage)
+    return xr.open_zarr(repo.readonly_session("main").store,
+                        group=ERA5_GROUP, consolidated=False)
+
+
+class _Sink:
+    """Writes lake objects to the bucket (boto3) or a local mirror dir."""
+
+    def __init__(self):
+        self.local = era5lake.local_dir()
+        self.cfg = era5lake.s3_config()
+        if not self.local and self.cfg is None:
+            sys.exit("no destination: set THERMOGRAPH_LAKE_S3_* or "
+                     "THERMOGRAPH_LAKE_LOCAL_DIR")
+        self._s3 = None
+        if not self.local:
+            import boto3  # noqa: PLC0415 - extractor-only dep
+            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"])
+
+    def put(self, key: str, df: pl.DataFrame) -> None:
+        if self.local:
+            path = os.path.join(self.local, key)
+            os.makedirs(os.path.dirname(path), exist_ok=True)
+            df.write_parquet(path)
+            return
+        buf = io.BytesIO()
+        df.write_parquet(buf)
+        self._s3.put_object(Bucket=self.cfg["bucket"], Key=key,
+                            Body=buf.getvalue())
+
+    def read_manifest(self) -> "pl.DataFrame | None":
+        try:
+            if self.local:
+                return pl.read_parquet(
+                    os.path.join(self.local, era5lake.MANIFEST_KEY))
+            obj = self._s3.get_object(Bucket=self.cfg["bucket"],
+                                      Key=era5lake.MANIFEST_KEY)
+            return pl.read_parquet(io.BytesIO(obj["Body"].read()))
+        except Exception:  # noqa: BLE001 - a fresh lake has no manifest yet
+            return None
+
+
+def _tile_frame(ds, ti: int, tj: int, end: str) -> "pl.DataFrame | None":
+    """Pull one 12x12 tile's hourly series and return the finalized daily frame
+    for all its points (lat_idx, lon_idx, date, ...). None for all-sea tiles."""
+    t = era5lake.TILE
+    i0, j0 = ti * t, tj * t
+    lat_sl = slice(i0, min(i0 + t, era5lake.LAT_N))
+    lon_sl = slice(j0, j0 + t)
+    if "lsm" in ds:            # land-sea mask: skip all-sea tiles outright
+        lsm = ds["lsm"].isel(latitude=lat_sl, longitude=lon_sl)
+        if "valid_time" in lsm.dims:   # static field, stored with a time axis
+            lsm = lsm.isel(valid_time=-1)
+        if float(lsm.max()) < 0.05:
+            return None
+    sub = (ds[list(ERA5_VARS)]
+           .isel(latitude=lat_sl, longitude=lon_sl)
+           .sel(valid_time=slice(START_DATE, end))
+           .load())
+
+    frames = []
+    for a, lat in enumerate(sub.latitude.values):
+        for b, lon in enumerate(sub.longitude.values):
+            cols = {"time": sub.valid_time.values}
+            for src, dst in ERA5_VARS.items():
+                cols[dst] = sub[src].isel(latitude=a, longitude=b).values
+            daily = climate._finalize_approximated(hourly_to_daily(pl.DataFrame(cols)))
+            i, j = era5lake.to_idx(float(lat), float(lon))
+            frames.append(daily.with_columns(
+                pl.lit(i, dtype=pl.Int32).alias("lat_idx"),
+                pl.lit(j, dtype=pl.Int32).alias("lon_idx")))
+    return pl.concat(frames)
+
+
+def _write_tile(sink: _Sink, ti: int, tj: int, tile_df: pl.DataFrame) -> pl.DataFrame:
+    """Write one tile's point files + hive partitions; return its manifest rows."""
+    rows = []
+    for (i, j), pdf in tile_df.group_by(["lat_idx", "lon_idx"]):
+        pdf = pdf.sort("date")
+        sink.put(era5lake.point_key(i, j), pdf.drop(["lat_idx", "lon_idx"]))
+        lat, lon = era5lake.to_coords(i, j)
+        rows.append({"lat_idx": i, "lon_idx": j, "lat": lat, "lon": lon,
+                     "tile": f"{ti}_{tj}", "rows": pdf.height,
+                     "date_min": str(pdf["date"].min()),
+                     "date_max": str(pdf["date"].max())})
+    parts = tile_df.with_columns(
+        pl.col("date").dt.year().alias("year"),
+        pl.col("date").dt.month().alias("month"))
+    for (year, month), mdf in parts.group_by(["year", "month"]):
+        sink.put(era5lake.daily_part_key(ti, tj, year, month),
+                 mdf.drop(["year", "month"]).sort(["lat_idx", "lon_idx", "date"]))
+    return pl.DataFrame(rows)
+
+
+def _city_tiles() -> list:
+    from data import grid  # noqa: PLC0415
+
+    tiles = set()
+    for c in cities.all_cities():
+        cell = grid.snap(c["lat"], c["lon"])
+        tiles.add(era5lake.tile_of(
+            *era5lake.to_idx(cell["center_lat"], cell["center_lon"])))
+    return sorted(tiles)
+
+
+def main() -> None:
+    ap = argparse.ArgumentParser()
+    ap.add_argument("--cities", action="store_true")
+    ap.add_argument("--land", action="store_true")
+    ap.add_argument("--tiles", help="ti0:ti1,tj0:tj1 tile-index window")
+    ap.add_argument("--limit", type=int)
+    ap.add_argument("--overwrite", action="store_true")
+    ap.add_argument("--dry-run", action="store_true")
+    args = ap.parse_args()
+
+    if args.cities:
+        todo = _city_tiles()
+    elif args.tiles:
+        (a, b), (c, d) = (p.split(":") for p in args.tiles.split(","))
+        todo = [(ti, tj) for ti in range(int(a), int(b))
+                for tj in range(int(c), int(d))]
+    elif args.land:
+        todo = [(ti, tj) for ti in range(era5lake.LAT_N // era5lake.TILE + 1)
+                for tj in range(era5lake.LON_N // era5lake.TILE)]
+    else:
+        ap.error("pick one of --cities / --tiles / --land")
+    if args.limit:
+        todo = todo[:args.limit]
+
+    sink = _Sink()
+    ds = _open_store()
+    end = (datetime.date.today()
+           - datetime.timedelta(days=climate.ARCHIVE_LATENCY_DAYS)).isoformat()
+
+    manifest = sink.read_manifest()
+    done = set() if manifest is None or args.overwrite else \
+        set(manifest["tile"].unique().to_list())
+
+    written = skipped = empty = 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
+            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}")
+
+
+if __name__ == "__main__":
+    main()

backend/lake_app.py

@@ -0,0 +1,176 @@
+"""The lake service: low-latency SQL-ish reads over the ERA5 bucket (prod-only).
+
+A small FastAPI app run as the Swarm ``lake`` service (same backend image,
+``THERMOGRAPH_ROLE=lake`` — the entrypoint launches this module instead of the
+web app; no database, no migrations). Web replicas call it instead of touching
+the bucket themselves, so a cell's history costs one LAN round-trip against a
+warm disk cache instead of an S3 fetch, and analytical queries get
+partition-pruned scans without shipping S3 credentials to every web task.
+
+Endpoints:
+- ``GET /healthz``            liveness (also reports manifest row count)
+- ``GET /history?lat=&lon=``  the hot path: nearest ERA5 point's full daily
+                              history as parquet bytes, disk-cached per point
+                              (a plain single-file read — no engine needed)
+- ``POST /query {"sql": ..}`` SELECT-only SQL over two views: ``era5_daily``
+                              (the hive table, partition-pruned on
+                              tile/year/month predicates) and ``manifest``.
+                              Runs on DuckDB — a real lake engine (full SQL,
+                              parallel scans, hive + row-group pushdown); the
+                              httpfs extension is baked into the image so S3
+                              needs no runtime extension fetch.
+
+Scale-out safe: replicas share nothing — each keeps its own cache under
+THERMOGRAPH_LAKE_CACHE (a per-task local volume), so 1..2 replicas behind the
+Swarm VIP need no coordination.
+"""
+import os
+import re
+import threading
+import time
+
+import polars as pl
+from fastapi import FastAPI, HTTPException
+from fastapi.responses import FileResponse
+from pydantic import BaseModel
+
+from data import era5lake
+
+CACHE_DIR = os.environ.get("THERMOGRAPH_LAKE_CACHE", "/state/lake-cache")
+MANIFEST_TTL = 6 * 3600        # re-pull the manifest at most this often
+QUERY_ROW_CAP = 10_000         # /query responses are JSON; keep them bounded
+
+app = FastAPI(title="thermograph-lake")
+
+_manifest_lock = threading.Lock()
+_manifest: "pl.DataFrame | None" = None
+_manifest_at = 0.0
+
+
+def _source():
+    """(mode, cfg) for this process: a local lake dir (tests/dev) or the bucket."""
+    if era5lake.local_dir():
+        return "local", None
+    cfg = era5lake.s3_config()
+    return ("bucket", cfg) if cfg else ("none", None)
+
+
+def _manifest_frame() -> "pl.DataFrame | None":
+    global _manifest, _manifest_at
+    with _manifest_lock:
+        if _manifest is not None and time.time() - _manifest_at < MANIFEST_TTL:
+            return _manifest
+        mode, cfg = _source()
+        try:
+            if mode == "local":
+                df = pl.read_parquet(
+                    os.path.join(era5lake.local_dir(), era5lake.MANIFEST_KEY))
+            elif mode == "bucket":
+                df = pl.read_parquet(
+                    f"s3://{cfg['bucket']}/{era5lake.MANIFEST_KEY}",
+                    storage_options=era5lake.storage_options(cfg))
+            else:
+                return None
+        except Exception:  # noqa: BLE001 - empty/unreachable lake: serve health, 404 reads
+            return None
+        _manifest, _manifest_at = df, time.time()
+        return df
+
+
+@app.get("/healthz")
+def healthz():
+    mode, _ = _source()
+    m = _manifest_frame()
+    return {"status": "ok", "source": mode,
+            "points": None if m is None else m.height}
+
+
+@app.get("/history")
+def history(lat: float, lon: float):
+    """One ERA5 point's full daily history, as parquet bytes (cached on disk —
+    the cache file IS the response body, so a warm hit is a sendfile)."""
+    i, j = era5lake.to_idx(lat, lon)
+    path = os.path.join(CACHE_DIR, f"p{i}_{j}.parquet")
+    if not os.path.exists(path):
+        mode, cfg = _source()
+        if mode == "none":
+            raise HTTPException(status_code=503, detail="lake not configured")
+        m = _manifest_frame()
+        if m is not None and m.filter(
+                (pl.col("lat_idx") == i) & (pl.col("lon_idx") == j)).is_empty():
+            raise HTTPException(status_code=404, detail="point not in the lake")
+        try:
+            if mode == "local":
+                df = era5lake._read_local(i, j)
+            else:
+                df = era5lake._read_bucket(i, j, cfg)
+        except era5lake.LakeUnavailable as e:
+            raise HTTPException(status_code=404, detail=str(e)) from e
+        os.makedirs(CACHE_DIR, exist_ok=True)
+        tmp = f"{path}.{os.getpid()}.partial"
+        df.write_parquet(tmp)
+        os.replace(tmp, path)   # atomic: concurrent requests see whole files only
+    return FileResponse(path, media_type="application/vnd.apache.parquet",
+                        headers={"X-Lake-Point": f"{i},{j}"})
+
+
+class Query(BaseModel):
+    sql: str
+
+
+_FORBIDDEN = re.compile(
+    r"\b(insert|update|delete|drop|create|alter|attach|copy|truncate|install|load"
+    r"|export|import|pragma|set|call)\b", re.I)
+
+
+def _connect():
+    """A fresh in-memory DuckDB with the lake views registered. Per-request:
+    connections are cheap, and a throwaway one means a query can't leave state
+    behind for the next. Returns None when no lake is configured."""
+    import duckdb  # noqa: PLC0415 - only the lake role pays the import
+
+    mode, cfg = _source()
+    if mode == "local":
+        daily_glob = os.path.join(era5lake.local_dir(), era5lake.PREFIX,
+                                  "daily/*/*/*/*.parquet")
+        manifest_src = os.path.join(era5lake.local_dir(), era5lake.MANIFEST_KEY)
+    elif mode == "bucket":
+        daily_glob = f"s3://{cfg['bucket']}/{era5lake.PREFIX}/daily/*/*/*/*.parquet"
+        manifest_src = f"s3://{cfg['bucket']}/{era5lake.MANIFEST_KEY}"
+    else:
+        return None
+    con = duckdb.connect()
+    if mode == "bucket":
+        con.execute("LOAD httpfs")   # baked into the image at build time
+        host = cfg["endpoint"].split("://", 1)[-1]
+        con.execute(f"SET s3_endpoint='{host}'")
+        con.execute("SET s3_url_style='path'")
+        con.execute(f"SET s3_region='{cfg['region']}'")
+        con.execute("SET s3_access_key_id=?", [cfg["access_key"]])
+        con.execute("SET s3_secret_access_key=?", [cfg["secret_key"]])
+    con.execute(f"""
+        CREATE VIEW era5_daily AS
+        SELECT * FROM read_parquet('{daily_glob}', hive_partitioning=1)""")
+    con.execute(f"CREATE VIEW manifest AS SELECT * FROM read_parquet('{manifest_src}')")
+    return con
+
+
+@app.post("/query")
+def query(q: Query):
+    sql = q.sql.strip().rstrip(";").strip()
+    if not sql.lower().startswith(("select", "with")) or ";" in sql \
+            or _FORBIDDEN.search(sql):
+        raise HTTPException(status_code=400,
+                            detail="only a single SELECT statement is allowed")
+    con = _connect()
+    if con is None:
+        raise HTTPException(status_code=503, detail="lake not configured")
+    try:
+        out = con.execute(
+            f"SELECT * FROM ({sql}) LIMIT {QUERY_ROW_CAP}").fetchall()
+        cols = [d[0] for d in con.description]
+    except Exception as e:  # noqa: BLE001 - surface the engine's message to the caller
+        raise HTTPException(status_code=400, detail=f"query failed: {e}") from e
+    finally:
+        con.close()
+    return {"columns": cols, "rows": out, "row_cap": QUERY_ROW_CAP}

backend/requirements-seed.txt

@@ -7,3 +7,10 @@
 icechunk
 xarray
 zarr
+# The live store compresses with PCodec; without this extra every read fails
+# with "codec not available: 'pcodec'".
+numcodecs[pcodec]
+# gen_era5_lake.py's bucket writer.
+boto3
+numcodecs[pcodec]
+boto3

backend/requirements.txt

@@ -25,3 +25,6 @@ websockets==16.0
 # Recurring worker-tier jobs — city warming, IndexNow pings (see
 # notifications/scheduler.py). In-process; no broker/queue needed at this scale.
 apscheduler==3.11.3
+# The lake role's SQL engine (lake_app.py /query): embedded, parallel
+# partition-pruned scans over the ERA5 hive table. Only lake_app imports it.
+duckdb==1.4.2

backend/seed_era5.py

@@ -16,10 +16,10 @@ Idempotent by default: a cell that already has a cached history is skipped (pass
 --overwrite to reseed). Not run in CI or the app image — the icechunk/xarray/zarr
 stack lives only in requirements-seed.txt.
 
-IMPORTANT — the S3/Icechunk *access* layer (_open_store / fetch_point) targets a
-young, fast-moving API and the store's exact variable/coordinate names; it is NOT
-exercised by the test suite. VERIFY it with `--dry-run` on the seed box before a
-full run. The hourly→daily transform (hourly_to_daily) IS unit-tested and is the
+The access layer was VERIFIED against the live store 2026-07-23 (prefix
+`icechunkV2`, group `single/temporal`, coord `valid_time`, ECMWF short variable
+names, and the pcodec codec — install `numcodecs[pcodec]`). Re-verify with
+`--dry-run` if the store revs. The hourly→daily transform (hourly_to_daily) IS unit-tested and is the
 part that must be numerically correct.
 
 Source: https://registry.opendata.aws/earthmover-era5/  (group="temporal", anonymous)
@@ -38,19 +38,19 @@ START_DATE = climate.START_DATE            # "1980-01-01"
 
 # Anonymous AWS Open Data Icechunk store; overridable in case the registry moves.
 ERA5_BUCKET = os.environ.get("THERMOGRAPH_ERA5_BUCKET", "earthmover-icechunk-era5")
-ERA5_PREFIX = os.environ.get("THERMOGRAPH_ERA5_PREFIX", "")   # verify on the seed box
+ERA5_PREFIX = os.environ.get("THERMOGRAPH_ERA5_PREFIX", "icechunkV2")
 ERA5_REGION = os.environ.get("THERMOGRAPH_ERA5_REGION", "us-east-1")
-ERA5_GROUP = os.environ.get("THERMOGRAPH_ERA5_GROUP", "temporal")  # time-contiguous layout
+ERA5_GROUP = os.environ.get("THERMOGRAPH_ERA5_GROUP", "single/temporal")  # time-contiguous layout
 
 # ERA5 store variable name -> our short name. ERA5 gusts (i10fg) are real here, so the
 # seed keeps genuine measured gusts rather than the Meteostat/estimate fallback.
 ERA5_VARS = {
-    "2m_temperature": "t2m",                 # K
-    "2m_dewpoint_temperature": "d2m",        # K (→ RH with t2m)
-    "total_precipitation": "tp",             # m, hourly accumulation
-    "10m_u_component_of_wind": "u10",        # m/s
-    "10m_v_component_of_wind": "v10",        # m/s
-    "instantaneous_10m_wind_gust": "i10fg",  # m/s
+    "t2m": "t2m",      # 2m temperature, K
+    "d2m": "d2m",      # 2m dewpoint, K (→ RH with t2m)
+    "tp": "tp",        # total precipitation, m, hourly accumulation
+    "u10": "u10",      # 10m wind u, m/s
+    "v10": "v10",      # 10m wind v, m/s
+    "fg10": "i10fg",   # 10m wind gust, m/s (store short name; transform expects i10fg)
 }
 
 _M_TO_IN = 39.37007874
@@ -117,9 +117,9 @@ def fetch_point(ds, lat: float, lon: float, start: str, end: str) -> pl.DataFram
     may differ per store revision — verify with --dry-run."""
     sub = (ds[list(ERA5_VARS)]
            .sel(latitude=lat, longitude=lon % 360.0, method="nearest")
-           .sel(time=slice(start, end))
+           .sel(valid_time=slice(start, end))
            .load())
-    data = {"time": sub["time"].values}
+    data = {"time": sub["valid_time"].values}
     for src, dst in ERA5_VARS.items():
         data[dst] = sub[src].values
     return pl.DataFrame(data)

backend/tests/data/test_climate.py

@@ -311,6 +311,44 @@ def _full_history_frame(n=4000):
     }))
 
 
+def test_lake_is_first_history_source(monkeypatch, tmp_path):
+    """A configured ERA5 lake short-circuits the whole third-party chain: neither
+    NASA POWER nor Open-Meteo is consulted."""
+    monkeypatch.setattr(climate, "CACHE_DIR", str(tmp_path))
+    monkeypatch.setattr(climate, "_archive_cooldown_until", 0.0)
+    cell = {"id": "lake_primary", "center_lat": 47.6, "center_lon": -122.3}
+
+    full = _full_history_frame()
+    monkeypatch.setattr(climate.era5lake, "fetch_history_lake",
+                        lambda c, start=None: full)
+    def _nasa_should_not_run(c): raise AssertionError("NASA should not be called")
+    def _om_should_not_run(c): raise AssertionError("Open-Meteo should not be called")
+    monkeypatch.setattr(climate, "_fetch_history_nasa", _nasa_should_not_run)
+    monkeypatch.setattr(climate, "_fetch_history", _om_should_not_run)
+
+    df, meta = climate._load_history(cell)
+    assert meta["source"] == "era5-lake"
+    assert df.height == full.height
+
+
+def test_lake_miss_falls_through_to_nasa(monkeypatch, tmp_path):
+    """An unconfigured lake (or a point outside it) costs nothing: NASA serves
+    exactly as before the lake existed."""
+    monkeypatch.setattr(climate, "CACHE_DIR", str(tmp_path))
+    monkeypatch.setattr(climate, "_archive_cooldown_until", 0.0)
+    cell = {"id": "lake_miss", "center_lat": 47.6, "center_lon": -122.3}
+
+    def _no_lake(c, start=None):
+        raise climate.era5lake.LakeUnavailable("no lake configured")
+    monkeypatch.setattr(climate.era5lake, "fetch_history_lake", _no_lake)
+    full = _full_history_frame()
+    monkeypatch.setattr(climate, "_fetch_history_nasa", lambda c: full)
+
+    df, meta = climate._load_history(cell)
+    assert meta["source"] == "nasa-power"
+    assert df.height == full.height
+
+
 def test_nasa_is_primary_history_source(monkeypatch, tmp_path):
     """NASA POWER is the primary history source: a full NASA frame is accepted and
     Open-Meteo is never consulted."""

backend/tests/data/test_era5lake.py

@@ -0,0 +1,90 @@
+"""The ERA5 lake: grid math, layout keys, and the read/slice client (no network
+— the local-dir source stands in for the bucket)."""
+import datetime
+
+import polars as pl
+import pytest
+
+from data import era5lake
+
+
+def _daily_frame(start="1940-01-01", days=32000):
+    d0 = datetime.date.fromisoformat(start)
+    dates = [d0 + datetime.timedelta(days=k) for k in range(days)]
+    n = len(dates)
+    return pl.DataFrame({
+        "date": dates,
+        "tmax": [70.0] * n, "tmin": [50.0] * n, "precip": [0.1] * n,
+        "wind": [8.0] * n, "gust": [14.0] * n, "humid": [55.0] * n,
+        "fmax": [71.0] * n, "fmin": [48.0] * n,
+    })
+
+
+def test_grid_roundtrip_and_bounds():
+    # Nearest-index snap, poles clamped, longitude wrapped to [0, 360).
+    assert era5lake.to_idx(90.0, 0.0) == (0, 0)
+    assert era5lake.to_idx(-90.0, 0.0) == (720, 0)
+    assert era5lake.to_idx(51.5074, -0.1278) == (154, 1439)
+    i, j = era5lake.to_idx(35.6762, 139.6503)
+    lat, lon = era5lake.to_coords(i, j)
+    assert abs(lat - 35.6762) <= era5lake.STEP / 2 + 1e-9
+    assert abs(lon - 139.6503) <= era5lake.STEP / 2 + 1e-9
+
+
+def test_tile_and_keys_align_with_hive_layout():
+    assert era5lake.tile_of(154, 1439) == (12, 119)
+    assert era5lake.point_key(154, 1439) == "era5/points/lat=154/lon=1439.parquet"
+    assert era5lake.daily_part_key(12, 119, 1994, 3) == \
+        "era5/daily/tile=12_119/year=1994/month=03/part.parquet"
+
+
+def test_s3_config_requires_both_keys(monkeypatch):
+    monkeypatch.delenv("THERMOGRAPH_LAKE_S3_ACCESS_KEY", raising=False)
+    monkeypatch.delenv("THERMOGRAPH_LAKE_S3_SECRET_KEY", raising=False)
+    assert era5lake.s3_config() is None
+    monkeypatch.setenv("THERMOGRAPH_LAKE_S3_ACCESS_KEY", "k")
+    assert era5lake.s3_config() is None
+    monkeypatch.setenv("THERMOGRAPH_LAKE_S3_SECRET_KEY", "s")
+    cfg = era5lake.s3_config()
+    assert cfg["bucket"] == "era5-thermograph"
+    opts = era5lake.storage_options(cfg)
+    assert opts["aws_virtual_hosted_style_request"] == "false"
+
+
+def _write_point(tmp_path, cell, df):
+    i, j = era5lake.to_idx(cell["center_lat"], cell["center_lon"])
+    path = tmp_path / era5lake.point_key(i, j)
+    path.parent.mkdir(parents=True, exist_ok=True)
+    df.write_parquet(path)
+
+
+CELL = {"center_lat": 51.5, "center_lon": -0.125}
+
+
+def test_fetch_slices_to_start_and_keeps_schema(monkeypatch, tmp_path):
+    monkeypatch.setenv("THERMOGRAPH_LAKE_LOCAL_DIR", str(tmp_path))
+    _write_point(tmp_path, CELL, _daily_frame())
+    df = era5lake.fetch_history_lake(CELL, start="1980-01-01")
+    assert df["date"].min() == datetime.date(1980, 1, 1)
+    assert set(df.columns) == {"date", "tmax", "tmin", "precip", "wind",
+                               "gust", "humid", "fmax", "fmin"}
+    # Without a start, the whole 1940+ record comes back.
+    assert era5lake.fetch_history_lake(CELL).height == 32000
+
+
+def test_fetch_rejects_short_and_missing(monkeypatch, tmp_path):
+    monkeypatch.setenv("THERMOGRAPH_LAKE_LOCAL_DIR", str(tmp_path))
+    with pytest.raises(era5lake.LakeUnavailable):
+        era5lake.fetch_history_lake(CELL)          # nothing written
+    _write_point(tmp_path, CELL, _daily_frame(days=500))
+    with pytest.raises(era5lake.LakeUnavailable):  # implausibly short record
+        era5lake.fetch_history_lake(CELL)
+
+
+def test_fetch_unconfigured_raises(monkeypatch):
+    monkeypatch.delenv("THERMOGRAPH_LAKE_LOCAL_DIR", raising=False)
+    monkeypatch.delenv("THERMOGRAPH_LAKE_URL", raising=False)
+    monkeypatch.delenv("THERMOGRAPH_LAKE_S3_ACCESS_KEY", raising=False)
+    monkeypatch.delenv("THERMOGRAPH_LAKE_S3_SECRET_KEY", raising=False)
+    with pytest.raises(era5lake.LakeUnavailable):
+        era5lake.fetch_history_lake(CELL)

backend/tests/web/test_lake_app.py

@@ -0,0 +1,100 @@
+"""The lake service over a local-dir lake fixture: health, the point hot path
+(+ its disk cache), and the SELECT-only SQL surface."""
+import datetime
+import io
+
+import polars as pl
+import pytest
+from fastapi.testclient import TestClient
+
+
+@pytest.fixture
+def lake(tmp_path, monkeypatch):
+    """A tiny two-point lake (points, hive partitions, manifest) on disk."""
+    from data import era5lake
+
+    points = [(154, 1439), (154, 0)]           # around London
+    d0 = datetime.date(1940, 1, 1)
+    dates = [d0 + datetime.timedelta(days=k) for k in range(12000)]
+    n = len(dates)
+    rows = []
+    for i, j in points:
+        df = pl.DataFrame({
+            "date": dates,
+            "tmax": [70.0] * n, "tmin": [50.0] * n, "precip": [0.1] * n,
+            "wind": [8.0] * n, "gust": [14.0] * n, "humid": [55.0] * n,
+            "fmax": [71.0] * n, "fmin": [48.0] * n,
+        })
+        p = tmp_path / era5lake.point_key(i, j)
+        p.parent.mkdir(parents=True, exist_ok=True)
+        df.write_parquet(p)
+        ti, tj = era5lake.tile_of(i, j)
+        part = df.with_columns(pl.lit(i, dtype=pl.Int32).alias("lat_idx"),
+                               pl.lit(j, dtype=pl.Int32).alias("lon_idx"))
+        hp = tmp_path / era5lake.daily_part_key(ti, tj, 1940, 1)
+        hp.parent.mkdir(parents=True, exist_ok=True)
+        part.filter((pl.col("date").dt.year() == 1940)
+                    & (pl.col("date").dt.month() == 1)).write_parquet(hp)
+        lat, lon = era5lake.to_coords(i, j)
+        rows.append({"lat_idx": i, "lon_idx": j, "lat": lat, "lon": lon,
+                     "tile": f"{ti}_{tj}", "rows": n,
+                     "date_min": "1940-01-01", "date_max": str(dates[-1])})
+    (tmp_path / "era5").mkdir(exist_ok=True)
+    pl.DataFrame(rows).write_parquet(tmp_path / era5lake.MANIFEST_KEY)
+
+    monkeypatch.setenv("THERMOGRAPH_LAKE_LOCAL_DIR", str(tmp_path))
+    monkeypatch.setenv("THERMOGRAPH_LAKE_CACHE", str(tmp_path / "cache"))
+    import lake_app
+    monkeypatch.setattr(lake_app, "CACHE_DIR", str(tmp_path / "cache"))
+    monkeypatch.setattr(lake_app, "_manifest", None)
+    return TestClient(lake_app.app)
+
+
+def test_healthz_reports_points(lake):
+    body = lake.get("/healthz").json()
+    assert body["status"] == "ok"
+    assert body["source"] == "local"
+    assert body["points"] == 2
+
+
+def test_history_serves_parquet_and_caches(lake, tmp_path):
+    r = lake.get("/history", params={"lat": 51.5074, "lon": -0.1278})
+    assert r.status_code == 200
+    assert r.headers["x-lake-point"] == "154,1439"
+    df = pl.read_parquet(io.BytesIO(r.content))
+    assert df.height == 12000
+    assert (tmp_path / "cache" / "p154_1439.parquet").exists()
+    # Second hit comes off the cache file and is byte-identical.
+    assert lake.get("/history",
+                    params={"lat": 51.5074, "lon": -0.1278}).content == r.content
+
+
+def test_history_404_for_point_outside_lake(lake):
+    r = lake.get("/history", params={"lat": -45.0, "lon": 170.0})
+    assert r.status_code == 404
+
+
+def test_query_selects_with_partition_predicates(lake):
+    r = lake.post("/query", json={"sql":
+        "SELECT year, month, count(*) AS n FROM era5_daily "
+        "WHERE year = 1940 AND month = 1 GROUP BY year, month"})
+    assert r.status_code == 200
+    body = r.json()
+    assert body["columns"] == ["year", "month", "n"]
+    assert body["rows"][0][2] == 2 * 31          # two points x January 1940
+
+
+def test_query_manifest_view(lake):
+    r = lake.post("/query", json={"sql": "SELECT count(*) FROM manifest"})
+    assert r.status_code == 200
+    assert r.json()["rows"][0][0] == 2
+
+
+@pytest.mark.parametrize("sql", [
+    "DROP TABLE era5_daily",
+    "SELECT 1; SELECT 2",
+    "INSERT INTO era5_daily VALUES (1)",
+    "COPY era5_daily TO 'x'",
+])
+def test_query_rejects_non_select(lake, sql):
+    assert lake.post("/query", json={"sql": sql}).status_code == 400

infra/deploy/secrets/README.md

@@ -15,7 +15,7 @@ no per-host duplication.
 | File | Holds | Encrypted? |
 |------|-------|------------|
 | `common.yaml` | Secrets shared by every host — Discord tokens/secret, VAPID keypair, `THERMOGRAPH_AUTH_SECRET`, metrics token, indexnow key, SMTP creds | ✅ |
-| `prod.yaml` | Prod-only overrides — `POSTGRES_PASSWORD`, `REGISTRY_TOKEN` | ✅ |
+| `prod.yaml` | Prod-only overrides — `POSTGRES_PASSWORD`, `REGISTRY_TOKEN`, ERA5 lake bucket creds (`THERMOGRAPH_LAKE_S3_ACCESS_KEY` / `THERMOGRAPH_LAKE_S3_SECRET_KEY`, consumed by the `lake` Swarm service and `gen_era5_lake.py`) | ✅ |
 | `beta.yaml` | Beta-only overrides | ✅ |
 | `example.yaml` | Format reference / CI fixture (fake values) | ✅ |
 | `../../.sops.yaml` | Which age recipient files are encrypted to (plaintext config) | — |

infra/deploy/stack/autoscale.sh

@@ -21,7 +21,9 @@
 set -eu
 
 STACK_NAME="${STACK_NAME:-thermograph}"
-SERVICE="${STACK_NAME}_web"
+# Which stack service to scale (the stack runs one autoscaler instance per
+# scaled service — web and lake today; same loop, different bounds).
+SERVICE="${STACK_NAME}_${TARGET_SERVICE:-web}"
 MIN="${MIN_REPLICAS:-1}"
 MAX="${MAX_REPLICAS:-3}"
 UP_AT="${SCALE_UP_CPU:-220}"

infra/deploy/stack/thermograph-stack.yml

@@ -82,6 +82,10 @@ services:
       # not the compose bridge's 172.19.0.1 (an overlay has no host gateway).
       # provision-mail.sh's DOCKER_MAIL_GATEWAY/SUBNET cover this listener.
       THERMOGRAPH_SMTP_HOST: ${STACK_SMTP_HOST:-172.18.0.1}
+      # History reads ask the lake service (Swarm VIP over 1..2 replicas)
+      # before any third-party source; unset on beta/LAN, where compose has no
+      # lake service and climate.py falls straight through to NASA.
+      THERMOGRAPH_LAKE_URL: http://lake:8141
     volumes:
       - appdata:/state
       - applogs:/app/logs
@@ -121,6 +125,7 @@ services:
       THERMOGRAPH_SINGLETON_PG: "1"
       RUN_MIGRATIONS: "0"
       THERMOGRAPH_SMTP_HOST: ${STACK_SMTP_HOST:-172.18.0.1}
+      THERMOGRAPH_LAKE_URL: http://lake:8141
     volumes:
       - appdata:/state
       - applogs:/app/logs
@@ -138,6 +143,45 @@ services:
       restart_policy:
         condition: on-failure
 
+  # The ERA5 lake indexer: SQL-style, partition-pruned reads over the
+  # era5-thermograph bucket with a node-local parquet cache, so a cell's
+  # history is one LAN hop for web instead of an S3 fetch (or a NASA call).
+  # Same backend image; THERMOGRAPH_ROLE=lake makes the entrypoint start
+  # lake_app (no database, no migrations). Prod-only by construction: it only
+  # exists in this stack file, and its S3 credentials
+  # (THERMOGRAPH_LAKE_S3_ACCESS_KEY/_SECRET_KEY) arrive via the same rendered
+  # /host/thermograph.env as every other secret. Without them it stays
+  # healthy and answers 503/404, and web falls through to NASA — the lake is
+  # an accelerator, never a point of failure.
+  lake:
+    image: ${REGISTRY_HOST:-git.thermograph.org}/${BACKEND_IMAGE_PATH:-emi/thermograph/backend}:${BACKEND_IMAGE_TAG:?required}
+    entrypoint: ["/host/env-entrypoint.sh"]
+    environment:
+      THERMOGRAPH_ROLE: lake
+      PORT: 8141
+      THERMOGRAPH_SERVICE_ROLE: backend
+      WORKERS: "1"
+      THERMOGRAPH_LAKE_CACHE: /state/lake-cache
+    volumes:
+      # Replicas share the cache; lake_app's tmp+rename writes make that safe.
+      - lakecache:/state
+      - /opt/thermograph/infra/deploy/stack/env-entrypoint.sh:/host/env-entrypoint.sh:ro
+      - /etc/thermograph/stack.env:/host/thermograph.env:ro
+    networks:
+      - internal
+    deploy:
+      replicas: ${LAKE_MIN_REPLICAS:-1}
+      placement:
+        constraints: ["node.role == manager"]
+      resources:
+        limits:
+          cpus: "${LAKE_CPUS:-2}"
+      restart_policy:
+        condition: on-failure
+      update_config:
+        order: start-first
+        failure_action: rollback
+
   frontend:
     image: ${REGISTRY_HOST:-git.thermograph.org}/${FRONTEND_IMAGE_PATH:-emi/thermograph/frontend}:${FRONTEND_IMAGE_TAG:?required}
     entrypoint: ["/host/env-entrypoint.sh"]
@@ -200,6 +244,39 @@ services:
       restart_policy:
         condition: any
 
+  # Second autoscaler instance, watching `lake` (1..2 replicas). The lake is
+  # read-only and cache-heavy, so it saturates on decode CPU well below web's
+  # thresholds — scale up sooner, still down slowly.
+  autoscaler-lake:
+    image: docker:27-cli
+    entrypoint: ["/bin/sh", "/host/autoscale.sh"]
+    environment:
+      STACK_NAME: ${STACK_NAME:-thermograph}
+      TARGET_SERVICE: lake
+      MIN_REPLICAS: ${LAKE_MIN_REPLICAS:-1}
+      MAX_REPLICAS: ${LAKE_MAX_REPLICAS:-2}
+      SCALE_UP_CPU: ${LAKE_SCALE_UP_CPU:-120}
+      SCALE_DOWN_CPU: ${LAKE_SCALE_DOWN_CPU:-30}
+      POLL_SECONDS: ${POLL_SECONDS:-15}
+      UP_SAMPLES: ${UP_SAMPLES:-3}
+      DOWN_SAMPLES: ${DOWN_SAMPLES:-20}
+      COOLDOWN_SECONDS: ${COOLDOWN_SECONDS:-180}
+    volumes:
+      - /var/run/docker.sock:/var/run/docker.sock
+      - /opt/thermograph/infra/deploy/stack/autoscale.sh:/host/autoscale.sh:ro
+    networks:
+      - internal
+    deploy:
+      replicas: 1
+      placement:
+        constraints: ["node.role == manager"]
+      resources:
+        limits:
+          cpus: "0.2"
+          memory: 64m
+      restart_policy:
+        condition: any
+
 networks:
   internal:
     driver: overlay
@@ -221,3 +298,6 @@ volumes:
   applogs:
     external: true
     name: ${APPLOGS_VOLUME:-thermograph_applogs}
+  # Lake parquet cache: node-local and disposable (a cold cache just refills
+  # from the bucket), so NOT external — the stack owns its lifecycle.
+  lakecache: {}