ctbk gbfs manifest {status,backfill} + manifest_status/manifest_fill registry ops

Backfill runs server-side: the CLI drives the api worker's registry
proxy, which parses each shard's footer via its D1 binding + R2 binding
(truthful side of the 2026-07-28 REST split-brain) and fills
`rg_manifest` rows synchronously. `status` reports fill coverage per
pyramid incl. stale fills (shard re-registered since fill).

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

Author: ryan-williams

ctbk/gbfs_cli.py

@@ -15,6 +15,7 @@
 import os
 import subprocess
 import sys
+import time
 from collections import Counter
 from datetime import datetime, timedelta, timezone
 from pathlib import Path
@@ -155,6 +156,78 @@ def _fmt(ms: int | None) -> tuple[str, float | None]:
 		print(f'{r["tier"]:>4} {r["shard_dur"]:>10} {r["n"]:>7}  {earliest:>16}  {latest:>16}  {age_str}')
 
 
+# ─── RG manifest (specs/rg-manifest.md) ─────────────────────────────
+
+MANIFEST_PYRAMIDS = ('rides-v5-start', 'rides-v5-end')
+
+
+def _registry_post(env_name: str, body: dict) -> dict:
+	"""POST a registry-proxy op to the api worker (Bearer
+	`CTBK_REGISTRY_SECRET`). All D1 access happens worker-side via the
+	binding — the truthful side of the 2026-07-28 REST split-brain."""
+	import urllib.request
+	secret = os.environ.get('CTBK_REGISTRY_SECRET')
+	if not secret:
+		raise click.ClickException('CTBK_REGISTRY_SECRET not set. `source .envrc`.')
+	req = urllib.request.Request(
+		f'{API_URLS[env_name]}/api/registry',
+		data=json.dumps(body).encode(),
+		headers={
+			'Authorization': f'Bearer {secret}',
+			'Content-Type': 'application/json',
+			'User-Agent': 'ctbk-gbfs-cli/1.0',
+		},
+	)
+	with urllib.request.urlopen(req, timeout=120) as resp:
+		return json.loads(resp.read())
+
+
+@gbfs.group('manifest', help='RG manifest: D1 row-group index for parquet pyramid serving (`specs/rg-manifest.md`).')
+def gbfs_manifest() -> None:
+	pass
+
+
+@gbfs_manifest.command('status', help='Fill coverage per pyramid: registered keys vs completed fills (+ stale fills whose shard was re-registered).')
+@option('-e', '--env', 'env_name', type=click.Choice(['dev', 'prod']), default='prod', show_default=True, help='api worker to query (shared D1, but ops must be deployed there).')
+@option('-p', '--pyramid', 'pyramids', multiple=True, help=f'Pyramid name (repeatable) [default: {", ".join(MANIFEST_PYRAMIDS)}].')
+@option('-v', '--verbose', is_flag=True, help='List unfilled/stale keys.')
+def gbfs_manifest_status(env_name: str, pyramids: tuple[str, ...], verbose: bool) -> None:
+	for pyramid in pyramids or MANIFEST_PYRAMIDS:
+		s = _registry_post(env_name, {'op': 'manifest_status', 'pyramid': pyramid})
+		print(f'{pyramid}: {s["filled"]}/{s["registered"]} filled, {len(s["stale"])} stale, {len(s["unfilled"])} unfilled')
+		if verbose:
+			for k in s['stale']:
+				print(f'  stale: {k}')
+			for k in s['unfilled']:
+				print(f'  unfilled: {k}')
+
+
+@gbfs_manifest.command('backfill', help='Fill manifest rows for unfilled/stale shards (worker parses each footer server-side via `manifest_fill`; sequential, idempotent).')
+@option('-e', '--env', 'env_name', type=click.Choice(['dev', 'prod']), default='prod', show_default=True, help='api worker to run fills on.')
+@option('-m', '--max', 'max_keys', type=int, default=None, help='Stop after this many fills.')
+@option('-n', '--dry-run', is_flag=True, help='List keys that would fill; no writes.')
+@option('-p', '--pyramid', 'pyramids', multiple=True, help=f'Pyramid name (repeatable) [default: {", ".join(MANIFEST_PYRAMIDS)}].')
+def gbfs_manifest_backfill(env_name: str, max_keys: int | None, dry_run: bool, pyramids: tuple[str, ...]) -> None:
+	done = 0
+	for pyramid in pyramids or MANIFEST_PYRAMIDS:
+		s = _registry_post(env_name, {'op': 'manifest_status', 'pyramid': pyramid})
+		todo = s['unfilled'] + s['stale']
+		err(f'{pyramid}: {len(todo)} to fill ({len(s["unfilled"])} unfilled, {len(s["stale"])} stale)')
+		for key in todo:
+			if max_keys is not None and done >= max_keys:
+				err(f'stopping at --max {max_keys}')
+				return
+			if dry_run:
+				print(key)
+				continue
+			t0 = time.time()
+			res = _registry_post(env_name, {'op': 'manifest_fill', 'pyramid': pyramid, 'key': key})
+			done += 1
+			err(f'  filled {key}: {res["n_rgs"]} RGs in {time.time() - t0:.1f}s')
+	if not dry_run:
+		err(f'{done} fills')
+
+
 # ─── cascade tick ───────────────────────────────────────────────────
 
 # Env → worker URL. Dev is used constantly for smoke tests; prod

gbfs/api/src/index.ts

@@ -494,7 +494,9 @@ import { computeAndStoreHealthSnapshot, readCachedHealthSnapshot } from './healt
 import { runAlerts } from './alerts';
 import { DEFAULT_PYRAMID, repairGeneration, serveAvailV3, serveAvailV3Cells } from './avail_geo';
 import { serveRidesV1, serveRidesV1Cells, serveRidesV2, serveRidesV2Cells, serveRidesV3, serveRidesV3Cells, serveRidesV5 } from './rides_v1';
-import { withR2Retry } from './r2_retry';
+import { retryingStorage, withR2Retry } from './r2_retry';
+import { r2Storage } from 'pyrmts-cfw';
+import { backfillManifestKey, manifestStatus } from './rg_manifest';
 
 /**
  * Build an `AsyncBuffer` (hyparquet's slice-based file abstraction) backed by
@@ -1192,7 +1194,7 @@ export default {
 			const auth = request.headers.get('Authorization') ?? '';
 			if (auth !== `Bearer ${env.REGISTRY_SECRET}`) return errorResponse('unauthorized', 403, env);
 			try {
-				const body = await request.json<{ op: string; pyramid?: string; rows?: {
+				const body = await request.json<{ op: string; pyramid?: string; key?: string; rows?: {
 					pyramid: string; tier: string; shard_dur: string;
 					period_start: number; period_end: number; key: string; written_at: number;
 				}[] }>();
@@ -1226,6 +1228,18 @@ export default {
 					console.log(`registry: register n=${rows.length} keys=${rows.map((r) => r.key).join(',')} d1_meta=${JSON.stringify(meta)}`);
 					return jsonResponse({ registered: rows.length, entry, d1: meta }, env);
 				}
+				// RG-manifest ops (`specs/rg-manifest.md`; `ctbk gbfs manifest`).
+				if (body.op === 'manifest_status') {
+					if (!body.pyramid) return errorResponse('pyramid required', 400, env);
+					return jsonResponse(await manifestStatus(env.DB, body.pyramid), env);
+				}
+				if (body.op === 'manifest_fill') {
+					if (!body.pyramid || !body.key) return errorResponse('pyramid + key required', 400, env);
+					const t0 = performance.now();
+					const storage = retryingStorage(r2Storage(env.R2));
+					const res = await backfillManifestKey(env.DB, storage, body.pyramid, body.key);
+					return jsonResponse({ ...res, ms: Math.round(performance.now() - t0) }, env);
+				}
 				return errorResponse(`unknown op ${body.op}`, 400, env);
 			} catch (err: any) {
 				return errorResponse(err.message ?? 'registry proxy failed', 500, env);

gbfs/api/src/rg_manifest.ts

@@ -260,6 +260,58 @@ async function fallbackFetch(opts: SegmentFetchOpts): Promise<Row[]> {
 	}
 }
 
+/** Registry-proxy backfill (`ctbk gbfs manifest backfill`): parse one
+ *  shard's footer and fill its manifest rows, synchronously. `written_at`
+ *  is read from `pyramid_shards` via the same binding (truthful side of
+ *  the 2026-07-28 split-brain). */
+export async function backfillManifestKey(
+	db: D1Database,
+	storage: Storage,
+	pyramid: string,
+	key: string,
+): Promise<{ n_rgs: number; written_at: number }> {
+	const row = await db.prepare('SELECT written_at FROM pyramid_shards WHERE pyramid = ? AND key = ?')
+		.bind(pyramid, key).first<{ written_at: number }>();
+	const writtenAt = row?.written_at ?? 0;
+	const release = await acquireFooterSlot(30_000);
+	try {
+		const head = await storage.head(key);
+		if (head === null) throw new Error(`backfillManifestKey: object not found: ${key}`);
+		const file = storageBuffer(storage, key, head.size);
+		const metadata = await parquetMetadataAsync(file, { initialFetchSize: INITIAL_FETCH_SIZE });
+		await fillManifestInner(db, pyramid, key, writtenAt, metadata);
+		return { n_rgs: metadata.row_groups.length, written_at: writtenAt };
+	} finally {
+		release();
+	}
+}
+
+/** Fill coverage for a pyramid: registered keys vs completed fills, with
+ *  stale fills (shard re-registered since) called out separately. */
+export async function manifestStatus(
+	db: D1Database,
+	pyramid: string,
+): Promise<{ registered: number; filled: number; stale: string[]; unfilled: string[] }> {
+	const [shards, fills] = await db.batch([
+		db.prepare('SELECT key, written_at FROM pyramid_shards WHERE pyramid = ?').bind(pyramid),
+		db.prepare('SELECT key, shard_written_at FROM rg_manifest_fills WHERE pyramid = ?').bind(pyramid),
+	]);
+	const fillMap = new Map(
+		((fills.results ?? []) as { key: string; shard_written_at: number }[])
+			.map((r) => [r.key, r.shard_written_at]),
+	);
+	const stale: string[] = [];
+	const unfilled: string[] = [];
+	let filled = 0;
+	for (const s of (shards.results ?? []) as { key: string; written_at: number }[]) {
+		const fillAt = fillMap.get(s.key);
+		if (fillAt === undefined) unfilled.push(s.key);
+		else if (fillAt !== s.written_at) stale.push(s.key);
+		else filled++;
+	}
+	return { registered: (shards.results ?? []).length, filled, stale, unfilled };
+}
+
 /** In-isolate single-flight: skip duplicate fills for a key already being
  *  filled here. Cross-isolate races are harmless: fills are idempotent
  *  (OR REPLACE, identical content for identical (key, written_at)), and