Audit candidate catalogs before server-side promotion

Author: ryan-williams

ctbk/flow_routes_city.py

@@ -15,7 +15,7 @@
 from sqlite3 import connect
 from subprocess import Popen, run
 from time import monotonic
-from typing import Iterable
+from typing import Callable, Iterable, Iterator, TypeVar
 from urllib.error import HTTPError
 from urllib.parse import urlencode
 from urllib.request import urlopen
@@ -30,6 +30,8 @@
 from .ride_animation import decode_day
 
 PREFIX = 'routes-city/v1'
+T = TypeVar('T')
+U = TypeVar('U')
 
 
 def city_population(population: dict, limit: int = 0) -> dict[str, list[dict]]:
@@ -103,25 +105,30 @@ def packed(self, meta: dict) -> dict:
         return loads(decompress(blob))
 
 
-def publish_catalogs(
-    store: RouteStore,
-    catalogs: Iterable[tuple[str, dict]],
+def bounded_results(
+    function: Callable[[T], U],
+    items: Iterable[T],
     workers: int = 8,
-) -> dict[str, dict]:
-    """Bound independent CA uploads, consuming SQLite-backed input on the caller."""
+) -> Iterator[U]:
+    """Consume input and ordered results on the caller, retaining at most N tasks."""
     if workers < 1:
-        raise ValueError('publication workers must be positive')
-    published = {}
-    pending: deque[tuple[str, Future[dict]]] = deque()
+        raise ValueError('catalog workers must be positive')
+    pending: deque[Future[U]] = deque()
     with ThreadPoolExecutor(max_workers=workers) as pool:
-        for sid, data in catalogs:
-            pending.append((sid, pool.submit(store.publish, data)))
+        for item in items:
+            pending.append(pool.submit(function, item))
             if len(pending) >= workers:
-                oldest_sid, future = pending.popleft()
-                published[oldest_sid] = future.result()
-        for sid, future in pending:
-            published[sid] = future.result()
-    return published
+                yield pending.popleft().result()
+        for future in pending:
+            yield future.result()
+
+
+def publish_catalogs(
+    publish: Callable[[dict], dict],
+    catalogs: Iterable[tuple[str, dict]],
+    workers: int = 8,
+) -> dict[str, dict]:
+    return dict(bounded_results(lambda item: (item[0], publish(item[1])), catalogs, workers))
 
 
 @flow_routes.command('city')
@@ -250,15 +257,15 @@ def origin_catalogs() -> Iterable[tuple[str, dict]]:
             yield sid, {'version': 1, 'pairs': records}
 
     err(f'Publishing {len(origins)} origin catalogs with at most 8 uploads in flight')
-    encoded_origins = publish_catalogs(store, origin_catalogs())
+    encoded_origins = publish_catalogs(store.publish, origin_catalogs())
 
     def destination_catalogs() -> Iterable[tuple[str, dict]]:
         for sid in sorted(destination_ids):
             records = [loads(row[0]) for row in destination_db.execute('SELECT record FROM pairs WHERE destination = ? ORDER BY source', (sid,))]
             yield sid, {'version': 1, 'pairs': records}
 
     err(f'Publishing {len(destination_ids)} destination catalogs with at most 8 uploads in flight')
-    encoded_destinations = publish_catalogs(store, destination_catalogs())
+    encoded_destinations = publish_catalogs(store.publish, destination_catalogs())
     destination_db.close()
     system = store.publish({'version': 1, 'context': context, 'graph': graph_meta, 'volumes': sorted(census_volumes.items()), 'dateRange': population['dateRange'], 'selectedRides': sum(counts.values()), 'routedRides': routed_rides})
     summary = {
@@ -304,7 +311,8 @@ def validate_catalog_pair(pair: dict, segments: list[dict]) -> None:
 @option('-o', '--output', type=ClickPath(path_type=Path), default=Path('tmp/routes-city-audit.json'))
 def check_city(prefix: str, output: Path) -> None:
     store = RouteStore(prefix, rw=False)
-    index = loads(store.get('index.json'))
+    index_blob = store.get('index.json')
+    index = loads(index_blob)
     if index['version'] != 1:
         raise ValueError('unsupported city catalog')
     census = RouteStore(index['populationPrefix'], rw=False)
@@ -321,8 +329,11 @@ def check_city(prefix: str, output: Path) -> None:
     fallbacks, volumes = Counter(), Counter()
     if set(index['origins']) != set(population):
         raise ValueError('origin catalog disagrees with census population')
-    for sid, meta in index['origins'].items():
-        catalog = store.packed(meta)
+    def read_catalog(item: tuple[str, dict]) -> tuple[str, dict]:
+        sid, meta = item
+        return sid, store.packed(meta)
+
+    for sid, catalog in bounded_results(read_catalog, index['origins'].items()):
         keys = [(p['source'], p['destination']) for p in catalog['pairs']]
         wanted = [(sid, p['destination']['id']) for p in population[sid]]
         if catalog['version'] != 1 or keys != wanted:
@@ -339,8 +350,7 @@ def check_city(prefix: str, output: Path) -> None:
                 fallbacks[pair['fallback']] += 1
     if set(index['destinations']) != {b for a, b in expected}:
         raise ValueError('destination catalog disagrees with census population')
-    for sid, meta in index['destinations'].items():
-        catalog = store.packed(meta)
+    for sid, catalog in bounded_results(read_catalog, index['destinations'].items()):
         for pair in catalog['pairs']:
             key = pair['source'], pair['destination']
             if pair['destination'] != sid or fingerprints.pop(key, None) != md5(object_blob(pair)).hexdigest():
@@ -353,12 +363,56 @@ def check_city(prefix: str, output: Path) -> None:
     system = store.packed(index['system'])
     if system['context'] != index['context'] or system['graph'] != index['graph'] or system['volumes'] != [list(p) for p in sorted(volumes.items())]:
         raise ValueError('census flow volumes disagree with route assignment')
-    report = actual | {'context': index['context'], 'segments': len(segments), 'bytes': index['bytes'], 'timings': index['timings'], 'resources': index['resources']}
+    report = actual | {'indexMd5': md5(index_blob).hexdigest(), 'context': index['context'], 'segments': len(segments), 'bytes': index['bytes'], 'timings': index['timings'], 'resources': index['resources']}
     output.parent.mkdir(parents=True, exist_ok=True)
     output.write_text(dumps(report, indent=2, sort_keys=True) + '\n')
     print(dumps(report, indent=2, sort_keys=True))
 
 
+@flow_routes.command('promote-city')
+@option('-a', '--audit', type=ClickPath(path_type=Path, exists=True), default=Path('tmp/routes-city-audit.json'), show_default=True)
+@option('-i', '--input-prefix', default='routes-city-candidate/v1', show_default=True)
+@option('-k', '--prefix', default=PREFIX, show_default=True)
+def promote_city(
+    audit: Path,
+    input_prefix: str,
+    prefix: str,
+) -> None:
+    if input_prefix == prefix:
+        raise UsageError('promotion source and destination must differ')
+    source, target = RouteStore(input_prefix, rw=False), RouteStore(prefix)
+    blob = source.get('index.json')
+    index, accepted = loads(blob), loads(audit.read_text())
+    if index['version'] != 1 or index.get('limit') != 0:
+        raise UsageError('promotion requires an uncapped version-one city catalog')
+    if accepted.get('indexMd5') != md5(blob).hexdigest() or accepted.get('context') != index['context']:
+        raise ValueError('candidate index differs from the accepted audit')
+    objects = {}
+    for meta in [index['graph'], index['system'], *index['origins'].values(), *index['destinations'].values()]:
+        key, size = meta['key'], meta['bytes']
+        if not fullmatch(r'objects/[a-f0-9]{32}\.json\.gz', key) or not isinstance(size, int) or size <= 0:
+            raise ValueError('invalid indexed CA object')
+        if key in objects and objects[key] != meta:
+            raise ValueError('inconsistent indexed CA object metadata')
+        objects[key] = meta
+
+    def copy_object(meta: dict) -> dict:
+        key, size = meta['key'], meta['bytes']
+        expected_etag = '"' + Path(key).name.split('.')[0] + '"'
+        source_key = f'{source.prefix}/{key}'
+        head = source.r2.head_object(Bucket=source.bucket, Key=source_key)
+        if head['ContentLength'] != size or head['ETag'] != expected_etag:
+            raise ValueError('source CA object hash or size disagrees with index')
+        copied = target.r2.copy_object(Bucket=target.bucket, Key=f'{target.prefix}/{key}', CopySource={'Bucket': source.bucket, 'Key': source_key}, CopySourceIfMatch=expected_etag, MetadataDirective='COPY')
+        if copied['CopyObjectResult']['ETag'] != expected_etag:
+            raise ValueError('copied CA object hash disagrees with index')
+        return meta
+
+    publish_catalogs(copy_object, sorted(objects.items()))
+    target.put('index.json', blob, mutable=True)
+    print(dumps({'source': input_prefix, 'prefix': prefix, 'context': index['context'], 'indexMd5': accepted['indexMd5'], 'objects': len(objects), 'bytes': sum(meta['bytes'] for meta in objects.values()), 'pairs': index['pairs']}, indent=2, sort_keys=True))
+
+
 @flow_routes.command('system')
 @option('-d', '--date', 'ym', required=True, help='YYYYMM published animation month.')
 @option('-i', '--input-prefix', default='rides-anim-candidate/v1', show_default=True)

ctbk/tests/test_flow_routes_city.py

@@ -1,5 +1,6 @@
 from collections import Counter
 from gzip import decompress
+from hashlib import md5
 from io import BytesIO
 from json import loads
 from pathlib import Path
@@ -73,7 +74,7 @@ def generate():
                 assert parallel.state['generated'] - parallel.state['completed'] < 2
                 parallel.state['generated'] += 1
             yield sid, data
-    result = publish_catalogs(parallel, generate(), workers=2)
+    result = publish_catalogs(parallel.publish, generate(), workers=2)
     assert list(result) == ['A', 'B', 'C', 'D']
     assert result == expected
     assert parallel.objects == serial.objects
@@ -85,9 +86,92 @@ class Store:
         def publish(self, data):
             raise RuntimeError('upload failed')
     with pytest.raises(RuntimeError, match='^upload failed$'):
-        publish_catalogs(Store(), [('A', {'version': 1})])
-    with pytest.raises(ValueError, match='^publication workers must be positive$'):
-        publish_catalogs(Store(), [], workers=0)
+        publish_catalogs(Store().publish, [('A', {'version': 1})])
+    with pytest.raises(ValueError, match='^catalog workers must be positive$'):
+        publish_catalogs(Store().publish, [], workers=0)
+
+
+@pytest.fixture
+def promotion(monkeypatch, tmp_path):
+    import ctbk.flow_routes_city as module
+    source, target = 'routes-city-candidate/v1', 'routes-city/v1'
+    objects, metas = {}, []
+    for data in [{'version': 1, 'segments': []}, {'version': 1, 'pairs': []}, {'version': 1, 'volumes': []}]:
+        blob = object_blob(data)
+        key = f'objects/{md5(blob).hexdigest()}.json.gz'
+        objects[f'{source}/{key}'] = blob
+        metas.append({'key': key, 'bytes': len(blob)})
+    index = {'version': 1, 'limit': 0, 'context': 'accepted', 'pairs': 1, 'graph': metas[0], 'origins': {'A': metas[1]}, 'destinations': {'B': metas[1]}, 'system': metas[2]}
+    blob = module.dumps(index, sort_keys=True).encode()
+    objects[f'{source}/index.json'] = blob
+    objects[f'{source}/checkpoints/unindexed.json.gz'] = b'not copied'
+    objects[f'{target}/index.json'] = b'previous production index'
+    audit = tmp_path / 'audit.json'
+    audit.write_text(module.dumps({'indexMd5': md5(blob).hexdigest(), 'context': 'accepted'}))
+    state = SimpleNamespace(objects=objects, events=[], copy_error=False, lock=Lock(), index=index, source=source, target=target, audit=audit, module=module, metas=metas)
+    class Client:
+        def head_object(self, Bucket, Key):
+            data = objects[Key]
+            return {'ContentLength': len(data), 'ETag': f'"{md5(data).hexdigest()}"'}
+
+        def copy_object(self, **kwargs):
+            if state.copy_error:
+                raise RuntimeError('copy failed')
+            source_key = kwargs['CopySource']['Key']
+            data = objects[source_key]
+            assert kwargs == {'Bucket': 'test', 'Key': f'{target}/{source_key.removeprefix(source + "/")}', 'CopySource': {'Bucket': 'test', 'Key': source_key}, 'CopySourceIfMatch': f'"{md5(data).hexdigest()}"', 'MetadataDirective': 'COPY'}
+            with state.lock:
+                objects[kwargs['Key']] = data
+                state.events.append(('copy', kwargs['Key']))
+            return {'CopyObjectResult': {'ETag': f'"{md5(data).hexdigest()}"'}}
+    class Store:
+        def __init__(self, prefix, rw=True):
+            self.prefix, self.bucket, self.r2 = prefix, 'test', Client()
+
+        def get(self, key):
+            state.events.append(('get', f'{self.prefix}/{key}'))
+            return objects[f'{self.prefix}/{key}']
+
+        def put(self, key, data, mutable=False):
+            assert mutable is True
+            state.events.append(('put', f'{self.prefix}/{key}'))
+            objects[f'{self.prefix}/{key}'] = data
+    monkeypatch.setattr(module, 'RouteStore', Store)
+    return state
+
+
+def test_promotion_copies_only_indexed_ca_objects_and_publishes_index_last(promotion):
+    p = promotion
+    p.module.promote_city.callback(audit=p.audit, input_prefix=p.source, prefix=p.target)
+    assert p.events[0] == ('get', f'{p.source}/index.json')
+    assert sorted(p.events[1:-1]) == [('copy', f'{p.target}/{meta["key"]}') for meta in sorted(p.metas, key=lambda meta: meta['key'])]
+    assert p.events[-1] == ('put', f'{p.target}/index.json')
+    assert {key.removeprefix(p.target + '/'): value for key, value in p.objects.items() if key.startswith(p.target + '/')} == {
+        'index.json': p.objects[f'{p.source}/index.json'],
+        **{meta['key']: p.objects[f'{p.source}/{meta["key"]}'] for meta in p.metas},
+    }
+
+
+@pytest.mark.parametrize('failure', ['capped', 'index_changed', 'object_changed', 'copy_failed'])
+def test_promotion_keeps_previous_index_on_rejection_or_copy_failure(promotion, failure):
+    p = promotion
+    exception, message = ValueError, '^candidate index differs from the accepted audit$'
+    if failure == 'capped':
+        p.index['limit'] = 10000
+        p.objects[f'{p.source}/index.json'] = p.module.dumps(p.index).encode()
+        exception, message = p.module.UsageError, '^promotion requires an uncapped version-one city catalog$'
+    elif failure == 'index_changed':
+        p.objects[f'{p.source}/index.json'] += b' '
+    elif failure == 'object_changed':
+        p.objects[f'{p.source}/{p.metas[0]["key"]}'] = b'corrupt graph'
+        message = '^source CA object hash or size disagrees with index$'
+    else:
+        p.copy_error = True
+        exception, message = RuntimeError, '^copy failed$'
+    with pytest.raises(exception, match=message):
+        p.module.promote_city.callback(audit=p.audit, input_prefix=p.source, prefix=p.target)
+    assert p.objects[f'{p.target}/index.json'] == b'previous production index'
+    assert [event for event in p.events if event[0] == 'put'] == []
 
 
 def test_city_builder_publishes_sparse_both_way_catalogs_and_exact_trip_coverage(monkeypatch, tmp_path):
@@ -148,6 +232,12 @@ def response(url, timeout):
     assert {k: loads((tmp_path / 'audit.json').read_text())[k] for k in ['pairs', 'routedPairs', 'routedRides', 'alternatives', 'fallbacks']} == {
         'pairs': 2, 'routedPairs': 1, 'routedRides': 100, 'alternatives': 1, 'fallbacks': {'NoRoute': 1},
     }
+    parallel_report = loads((tmp_path / 'audit.json').read_text())
+    bounded = module.bounded_results
+    with monkeypatch.context() as serial:
+        serial.setattr(module, 'bounded_results', lambda function, items, workers=8: bounded(function, items, workers=1))
+        module.check_city.callback(prefix='test-city', output=tmp_path / 'serial-audit.json')
+    assert loads((tmp_path / 'serial-audit.json').read_text()) == parallel_report
     # A second run reads the saved origin checkpoints rather than contacting the router.
     monkeypatch.setattr(module, 'urlopen', lambda *args, **kwargs: pytest.fail('checkpoint route was recomputed'))
     city.callback(input_prefix='routes-census/v1', prefix='test-city', limit=0, output=tmp_path, snapshot='20261012')

specs/flow-routes.md

@@ -109,8 +109,12 @@ The city builder streams node annotations in two passes over per-origin compress
 
 CA catalog publication uses at most eight concurrent uploads, consuming origin encoding and SQLite destination reads on the caller thread. Only eight catalog payloads can remain queued; the ordered result map and deterministic compressed objects match serial publication byte-for-byte. Routing/checkpoint computation remains serial. Separate topology/publication timings and stage logs expose the post-routing work.
 
+Build the full rollout under `routes-city-candidate/v1`, then run `check-city -k routes-city-candidate/v1` and `promote-city` in the same remote task. The checker uses eight bounded catalog reads while preserving origin/destination iteration and floating-point aggregation order on the caller thread. Its small report binds acceptance to the exact candidate index MD5. Promotion refuses capped catalogs or a changed index, server-side copies only indexed CA objects with source size/hash checks and conditional copies, and publishes the production `index.json` last. It copies no checkpoints and downloads no catalog bodies to the laptop. Server-side copy and source ETag conditions are supported by R2's [S3 API compatibility table][r2-copy].
+
 The existing 2024 census is a sizing baseline, not a current-population guarantee. Refresh `routes-census.yml` for `202510-202609` before the full build; September remains provisional. Catalogs retain the immutable census object key, so a later census refresh does not invalidate an older build's audit. New or rare pairs outside the chosen population continue to use explicit fallback arcs/straight animation paths. The full JC/Hoboken catalog still overrides the city/sample namespace for its covered pairs.
 
 `ctbk flow-routes system -d YYYYMM` streams the month's immutable animation days, verifies hashes/counts, excludes carry-in and round trips, and assigns actual monthly directed-pair counts to route alternatives. It publishes `flows/v1` with exact encoded/routed/fallback/roundtrip accounting, source object hashes, provisional status, the route context and graph key, and expected signed street volumes. Missing routes are reported as uncovered volume rather than invented street traffic. `.github/workflows/routes-system.yml` runs this remotely for one month with a serialized writer; monthly automatic extension and time-mode interpolation remain follow-up work.
 
 `/stations?m=2506&network=1` shows the opt-in whole-network layer with direction chevrons and opposing lanes when no station is pinned. It loads one monthly volume object and one shared graph, without downloading every origin catalog. The legend reports the actual non-roundtrip routed share and explicitly names an unbuilt month. Selections continue to use lazy origin/destination catalogs; a missing city index keeps today's NJ/sample behavior. Unit tests cover exact directional expected volumes and floors; a complete hermetic two-origin build/audit tests both indexes and checkpoint resume; the monthly fixture proves carry/roundtrip exclusion and uncovered-pair accounting. Browser checks cover network opt-in/reload/view preservation and an unbuilt month. Chrome CIC could not connect because the installed browser runtime references a missing service module; automated Chromium rendering was checked instead, without claiming HCCSx verification.
+
+[r2-copy]: https://developers.cloudflare.com/r2/api/s3/api/