smg-v1 stand-up: worker reconcile, engine sweep/adopt, capped + self-healing daily fill

Stand-up of the SMG pyramid (`specs/avail-smg-pyramid.md`, done 2026-09-07 from `e`): 152 source days
backfilled, image `ctbk-engine:f5fefbe1` (job-def rev 12), Batch `-f` build → 224 shards registered in
D1, cover [2026-04-07, 2026-09-06).

Two engine-side surprises, both worked around here and documented in the spec:
- The engine's default `mem_budget` (70% of the cgroup limit) read the Fargate *host* (46 GB in a 32 GiB
  container) → OOM at 96/307 windows. Every `engine submit` now passes `-b` (pyrmts spec filed).
- The Batch base image (pyrmts `ed50cdb`) predates open-period classification, so an uncapped fill
  writes 0-row trailing shards over not-yet-existing days and then fails the strict source check.
  New `ctbk gbfs engine sweep CUTOFF` deletes such shards from R2 + D1 + manifest; new
  `ctbk gbfs engine adopt` records shards a dead job flushed but never wrote to the manifest;
  `engine manifest -H N` dumps raw records.

Daily cadence (`gbfs-compact.yml`): a `ctbk gbfs smg backfill -C -k -t <day>` self-heal step (no-op
when nothing is missing) precedes the fill, and the fill is capped at `<day+1>T00:00` (`-r`) with
`-b 20g -w 1h`. `RECONCILE_PYRAMIDS` in the api worker now includes `smg-v1`, so the daily fill's new
shards self-register.

Spec also records the heartbeat era (2026-05-03): before it `no_poll` ≡ "unpolled or stale".

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01GgLAzeRJeNeyizG4XHWRBK

Author: ryan-williams

.github/workflows/gbfs-compact.yml

@@ -46,6 +46,8 @@ jobs:
           echo "trigger_mo1=$([ "$NEXT_MONTH" = "01" ] && [ "$DOM" = "01" ] && echo true || echo false)" >> "$GITHUB_OUTPUT"
           echo "ym=$(echo $DATE | cut -c 1-7)" >> "$GITHUB_OUTPUT"
           echo "year=$(echo $DATE | cut -c 1-4)" >> "$GITHUB_OUTPUT"
+          # Pyramid fills are capped at the end of DATE: the last day whose source exists.
+          echo "range_to=$(date -u -d "$DATE +1 day" +%Y-%m-%dT00:00)" >> "$GITHUB_OUTPUT"
 
       - name: Compact WAL → parquet
         run: python3 gbfs/compact-r2.py all ${{ steps.date.outputs.date }}
@@ -60,12 +62,31 @@ jobs:
           CLOUDFLARE_ACCOUNT_ID: ${{ secrets.CLOUDFLARE_ACCOUNT_ID }}
         run: ctbk gbfs empty build -C -V ${{ steps.date.outputs.date }}
 
-      # SMG pyramid (`specs/avail-smg-pyramid.md`): `empty build` above also wrote the
-      # day's `gbfs/smg/<day>.parquet`; a declarative Batch gap-fill (`-f`) builds only
-      # the rungs that new day completes, into the REAL prefix (`-p smg-v1` — the
-      # default would be the `-engine-check` scratch prefix). Own AWS env: the
-      # job-level `AWS_*` above point at R2, and Batch is real AWS (us-east-1).
-      # `-W` tails the job so a failed fill fails this step.
+      # SMG source (`specs/avail-smg-pyramid.md`): `empty build` above also wrote the
+      # day's `gbfs/smg/<day>.parquet`. This self-heals any day that has a status
+      # parquet but no SMG parquet (a missed run, or days compacted before the SMG
+      # step existed) — a no-op when nothing is missing — so the fill below never
+      # sees a real source hole.
+      - name: Build any missing gbfs/smg day parquets
+        env:
+          R2_ACCESS_KEY_ID: ${{ secrets.R2_ACCESS_KEY_ID }}
+          R2_SECRET_ACCESS_KEY: ${{ secrets.R2_SECRET_ACCESS_KEY }}
+          CLOUDFLARE_ACCOUNT_ID: ${{ secrets.CLOUDFLARE_ACCOUNT_ID }}
+        run: ctbk gbfs smg backfill -C -k -t ${{ steps.date.outputs.date }}
+
+      # SMG pyramid: a declarative Batch gap-fill (`-f`) builds only the rungs the new
+      # day completes, into the REAL prefix (`-p smg-v1` — the default would be the
+      # `-engine-check` scratch prefix). Own AWS env: the job-level `AWS_*` above
+      # point at R2, and Batch is real AWS (us-east-1). `-W` tails the job so a
+      # failed fill fails this step.
+      # `-r /<range_to>` CAPS the range at the end of the day just built: the Batch
+      # base image (pyrmts ed50cdb) predates open-period classification, so an
+      # uncapped fill would (a) fail the strict missing-source check on today's
+      # not-yet-existing parquet and (b) first write 0-row shards over it, which
+      # every later fill then treats as built (`ctbk gbfs engine sweep` undoes that).
+      # `-b 20g`: the engine's default budget (70% of the cgroup limit) reads the
+      # Fargate HOST here (46 GB in a 32 GiB container → OOM on the first full
+      # build); an explicit budget under the container size keeps a catch-up safe.
       - name: Fill smg-v1 pyramid (Batch)
         env:
           AWS_ACCESS_KEY_ID: ${{ secrets.AWS_ACCESS_KEY_ID }}
@@ -76,7 +97,7 @@ jobs:
           R2_SECRET_ACCESS_KEY: ${{ secrets.R2_SECRET_ACCESS_KEY }}
           CLOUDFLARE_ACCOUNT_ID: ${{ secrets.CLOUDFLARE_ACCOUNT_ID }}
           CTBK_REGISTRY_SECRET: ${{ secrets.CTBK_REGISTRY_SECRET }}
-        run: ctbk gbfs engine submit -C smg-v1 -p smg-v1 -x ctbk_engine_src:smg_daily -f -W
+        run: ctbk gbfs engine submit -C smg-v1 -p smg-v1 -x ctbk_engine_src:smg_daily -f -W -b 20g -w 1h -r /${{ steps.date.outputs.range_to }}
 
       - name: Build /h1 avail-agg for the day
         run: |

ctbk/gbfs_cli.py

@@ -12,6 +12,7 @@
 from __future__ import annotations
 
 import json
+import re
 import os
 import subprocess
 import sys
@@ -1594,10 +1595,15 @@ def gbfs_engine_register(
 
 @gbfs_engine.command('manifest', help='Summarize a build manifest (local path or `s3://` URL): shard counts per (tier, rung), period span; optionally diff key sets vs another manifest.')
 @option('-d', '--diff', 'other', default=None, help='Second manifest; report key-set differences.')
+@option('-H', '--head', 'head_n', type=int, default=0, help='Print the first N raw records (JSONL) instead of the summary.')
 @argument('path', metavar='PATH')
-def gbfs_engine_manifest(other: str | None, path: str) -> None:
+def gbfs_engine_manifest(other: str | None, head_n: int, path: str) -> None:
 	load = _load_manifest
 	recs = load(path)
+	if head_n:
+		for r in recs[:head_n]:
+			print(json.dumps(r, separators=(',', ':')))
+		return
 	by_rung = Counter((r['tier'], r['shard_dur']) for r in recs)
 	starts = [r['period_start'] for r in recs]
 	ends = [r['period_end'] for r in recs]
@@ -1616,6 +1622,162 @@ def gbfs_engine_manifest(other: str | None, path: str) -> None:
 			sys.exit(1)
 
 
+def _period_from_key(prefix: str, key: str) -> tuple[str, str, datetime, datetime]:
+	"""`(tier, shard_dur, period_start, period_end)` for a `{prefix}/{tier}/{shard}/{label}.parquet` key.
+	Labels are the engine's period labels (`2026-09-07`, `2026-09-07T03-10`, …)."""
+	from pyrmts import shard_periods_covering
+	rel = key[len(prefix) + 1:]
+	if not rel.endswith('.parquet'):
+		raise click.ClickException(f'not a shard key: {key}')
+	tier, shard, label = rel[:-len('.parquet')].split('/')
+	iso = re.sub(r'T(\d\d)-(\d\d)', r'T\1:\2', label)
+	start = datetime.fromisoformat(iso).replace(tzinfo=timezone.utc)
+	p = shard_periods_covering(start, start + timedelta(milliseconds=1), shard)[0]
+	if p.start != start:
+		raise click.ClickException(f'{key}: label {label!r} is not {shard}-aligned (period starts {p.start.isoformat()})')
+	return tier, shard, p.start, p.end
+
+
+@gbfs_engine.command('sweep', help='Delete shards whose period extends past CUTOFF (UTC ISO) from all three stores — R2, the D1 `pyramid_shards` registry, and the prefix\'s manifest. Run after an UNCAPPED `engine submit -f` on an engine without open-period classification (base image < pyrmts 72f2552): the fill\'s expected cover reaches `now`, so shards spanning source days that don\'t exist yet build with 0 rows for those days (or partial rows, for a shard straddling the last real day) and read as "built" to every later fill. Then re-fill CAPPED: `engine submit -f -r /CUTOFF`. Candidates = manifest records ∪ R2 listing (keys missing from the manifest are reported).')
+@option('-C', '--config', 'config_name', default='smg-v1', show_default=True, help='Pyramid config basename (registry name and prefix default to it).')
+@option('-m', '--manifest', 'manifest_name', default='manifest.jsonl', show_default=True, help='Manifest object name under the prefix.')
+@option('-n', '--dry-run', is_flag=True, help='List affected shards; no deletes.')
+@option('-p', '--prefix', default=None, help='R2 key prefix [default: <config>].')
+@option('-P', '--pyramid', 'pyramid_name', default=None, help='Registry pyramid name [default: <config>].')
+@argument('cutoff', metavar='CUTOFF')
+def gbfs_engine_sweep(
+	config_name: str,
+	manifest_name: str,
+	dry_run: bool,
+	prefix: str | None,
+	pyramid_name: str | None,
+	cutoff: str,
+) -> None:
+	from ctbk.pyramid_cascade.d1_http import d1_query
+	prefix = prefix or config_name
+	pyramid_name = pyramid_name or config_name
+	cut = datetime.fromisoformat(cutoff)
+	cut = cut.replace(tzinfo=timezone.utc) if cut.tzinfo is None else cut.astimezone(timezone.utc)
+	cut_ms = int(cut.timestamp()) * 1000
+	client, bucket = _r2_client()
+	mkey = f'{prefix}/{manifest_name}'
+	recs = _load_manifest(f's3://{bucket}/{mkey}')
+	by_key = {r['key']: r for r in recs}
+	paginator = client.get_paginator('list_objects_v2')
+	r2_keys = [
+		o['Key']
+		for page in paginator.paginate(Bucket=bucket, Prefix=f'{prefix}/')
+		for o in page.get('Contents') or []
+		if o['Key'].endswith('.parquet')
+	]
+	unlisted = sorted(set(r2_keys) - set(by_key))
+	if unlisted:
+		err(f'sweep: {len(unlisted)} R2 shard(s) not in {mkey}: ' + ', '.join(unlisted[:5]) + (', …' if len(unlisted) > 5 else ''))
+	victims: list[tuple[str, str, str, int, int]] = []  # (key, tier, shard_dur, period_start_ms, period_end_ms)
+	for key in sorted(set(r2_keys) | set(by_key)):
+		r = by_key.get(key)
+		if r is not None:
+			tier, shard, p0, p1 = r['tier'], r['shard_dur'], r['period_start'], r['period_end']
+		else:
+			tier, shard, s, e = _period_from_key(prefix, key)
+			p0, p1 = int(s.timestamp()) * 1000, int(e.timestamp()) * 1000
+		if p1 > cut_ms:
+			victims.append((key, tier, shard, p0, p1))
+	if not victims:
+		err(f'sweep: no shards under {prefix}/ extend past {cut.isoformat()}')
+		return
+	err(f'sweep: {len(victims)} shard(s) under {prefix}/ extend past {cut.isoformat()}:')
+	fmt = lambda ms: datetime.fromtimestamp(ms / 1000, tz=timezone.utc).strftime('%Y-%m-%dT%H:%M')
+	for key, _t, _s, p0, p1 in victims:
+		err(f'  {key}  [{fmt(p0)}, {fmt(p1)})')
+	if dry_run:
+		return
+	keys = [v[0] for v in victims]
+	for i in range(0, len(keys), 1000):
+		client.delete_objects(Bucket=bucket, Delete={'Objects': [{'Key': k} for k in keys[i:i + 1000]], 'Quiet': True})  # type: ignore[attr-defined]
+	err(f'  deleted {len(keys)} R2 object(s)')
+	for i in range(0, len(keys), 50):
+		chunk = keys[i:i + 50]
+		d1_query(f"DELETE FROM pyramid_shards WHERE pyramid = ? AND key IN ({', '.join('?' * len(chunk))})", [pyramid_name, *chunk])
+	err(f'  dropped {len(keys)} registry row(s) (pyramid={pyramid_name}, where present)')
+	dropped = set(keys)
+	keep = [r for r in recs if r['key'] not in dropped]
+	if len(keep) != len(recs):
+		out = '\n'.join(json.dumps(r, separators=(',', ':')) for r in keep) + '\n'
+		client.put_object(Bucket=bucket, Key=mkey, Body=out.encode(), ContentType='application/jsonl')  # type: ignore[attr-defined]
+		err(f'  {mkey}: {len(recs)} → {len(keep)} records')
+
+
+@gbfs_engine.command('adopt', help='Adopt R2 shards that a build flushed but never recorded (the job died after the PUT — OOM, Spot reclaim): append them to the prefix\'s manifest (HEAD for bytes, streamed md5) and register them in D1. Default KEYS = every `.parquet` under the prefix missing from the manifest (what `sweep` reports as "not in manifest").')
+@option('-C', '--config', 'config_name', default='smg-v1', show_default=True, help='Pyramid config basename (registry name and prefix default to it).')
+@option('-m', '--manifest', 'manifest_name', default='manifest.jsonl', show_default=True, help='Manifest object name under the prefix.')
+@option('-n', '--dry-run', is_flag=True, help='List the shards that would be adopted; no writes.')
+@option('-p', '--prefix', default=None, help='R2 key prefix [default: <config>].')
+@option('-P', '--pyramid', 'pyramid_name', default=None, help='Registry pyramid name [default: <config>].')
+@argument('keys', metavar='KEYS', nargs=-1)
+def gbfs_engine_adopt(
+	config_name: str,
+	manifest_name: str,
+	dry_run: bool,
+	prefix: str | None,
+	pyramid_name: str | None,
+	keys: tuple[str, ...],
+) -> None:
+	from hashlib import md5
+	from ctbk.pyramid_cascade.d1_http import d1_query
+	prefix = prefix or config_name
+	pyramid_name = pyramid_name or config_name
+	client, bucket = _r2_client()
+	mkey = f'{prefix}/{manifest_name}'
+	recs = _load_manifest(f's3://{bucket}/{mkey}')
+	listed = {r['key'] for r in recs}
+	if not keys:
+		paginator = client.get_paginator('list_objects_v2')
+		keys = tuple(sorted(
+			o['Key']
+			for page in paginator.paginate(Bucket=bucket, Prefix=f'{prefix}/')
+			for o in page.get('Contents') or []
+			if o['Key'].endswith('.parquet') and o['Key'] not in listed
+		))
+	else:
+		dup = [k for k in keys if k in listed]
+		if dup:
+			raise click.ClickException(f'already in {mkey}: {dup}')
+	if not keys:
+		err(f'adopt: every shard under {prefix}/ is already in {mkey}')
+		return
+	err(f'adopt: {len(keys)} shard(s) → {mkey} + pyramid_shards({pyramid_name}):')
+	for k in keys:
+		err(f'  {k}')
+	if dry_run:
+		return
+	new: list[dict] = []
+	for k in keys:
+		tier, shard, s, e = _period_from_key(prefix, k)
+		obj = client.get_object(Bucket=bucket, Key=k)  # type: ignore[attr-defined]
+		h = md5()
+		n = 0
+		for chunk in obj['Body'].iter_chunks(8 << 20):
+			h.update(chunk)
+			n += len(chunk)
+		new.append({
+			'pyramid': pyramid_name, 'tier': tier, 'shard_dur': shard,
+			'period_start': int(s.timestamp()) * 1000, 'period_end': int(e.timestamp()) * 1000,
+			'key': k, 'written_at': int(obj['LastModified'].timestamp() * 1000),
+			'md5': h.hexdigest(), 'bytes': n,
+		})
+		err(f'  {k}: {n:,} B md5={h.hexdigest()}')
+	out = '\n'.join(json.dumps(r, separators=(',', ':')) for r in recs + new) + '\n'
+	client.put_object(Bucket=bucket, Key=mkey, Body=out.encode(), ContentType='application/jsonl')  # type: ignore[attr-defined]
+	err(f'  {mkey}: {len(recs)} → {len(recs) + len(new)} records')
+	cols = '(pyramid, tier, shard_dur, period_start, period_end, key, written_at)'
+	params: list = []
+	for r in new:
+		params += [r['pyramid'], r['tier'], r['shard_dur'], r['period_start'], r['period_end'], r['key'], r['written_at']]
+	d1_query(f"INSERT OR REPLACE INTO pyramid_shards {cols} VALUES {', '.join(['(?, ?, ?, ?, ?, ?, ?)'] * len(new))}", params)
+	err(f'  registered {len(new)}')
+
+
 @gbfs_r2.command('pq', help='Inspect a parquet object on R2: schema + row count; optionally head rows and row-group metadata (range reads, no full download).')
 @option('-m', '--metadata', 'show_meta', is_flag=True, help='Row-group count/sizes and file metadata.')
 @option('-n', '--head', 'head_n', type=int, default=0, help='Print the first N rows.')

gbfs/api/src/index.ts

@@ -1122,6 +1122,9 @@ const RECONCILE_PYRAMIDS: { name: string; prefix: string; rides?: boolean }[] =
 	{ name: 'avail', prefix: 'avail-v3/' },
 	{ name: 'avail-v5', prefix: 'avail-v5/' },
 	{ name: 'avail-v6', prefix: 'avail-v6/' },
+	// Same ladder + genesis as avail-v6 (`configs/pyramids/smg-v1.yaml`); the
+	// daily Batch `-f` fill writes shards but doesn't register them.
+	{ name: 'smg-v1', prefix: 'smg-v1/' },
 	{ name: 'rides-v5-start', prefix: 'rides-v5/start/', rides: true },
 	{ name: 'rides-v5-end', prefix: 'rides-v5/end/', rides: true },
 ];