perf(builder): default --extract-workers 8→64; size boto3 pool to match

Catalog extraction is latency-bound: each file is 1-2 sequential S3 range
GETs (~190ms each over WAN), so throughput = concurrency / per-file latency.
At the old default of 8 workers a full cold scan runs ~20 files/s (measured
on a ~24.5k-file partition: ~20 min).

Raise the default to 64. The companion fix is essential: boto3's default
connection pool is 10, so >10 workers would contend and retry (pool-full →
slower). Build the S3 client with Config(max_pool_connections=max(workers,10))
so the pool tracks the worker count. botocore is imported defensively so the
s3 path stays exercisable with a bare fake boto3 (no botocore) in tests.

Docs + test updated: README flag table, --help text, and test_parser_defaults
(asserts 64; fake boto3.client stub now accepts the config= kwarg).

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

Author: facontidavide

mcap_catalog/mcap_catalog_builder/README.md

@@ -24,7 +24,7 @@ python3 -m mcap_catalog_builder --source s3 --s3-bucket B --sqs-url U  # S3
 | `--tag-socket` | off | path for the tag-edit IPC unix socket (daemon mode only; see [Tag-edit IPC](#tag-edit-ipc)) |
 | `--rescan-interval` | `300.0` | seconds between safety re-scans |
 | `--no-watch` | off | daemon mode: start **no** live event producer (no local watchdog/inotify observer, no S3 SQS long-poll thread) — discovery is then rescan-only, driven purely by `--rescan-interval`. With `--source s3`, also drops the `--sqs-url` requirement. No-op for `--source gcs` (already rescan-only) and for `--once` |
-| `--extract-workers` | `8` | threads that fetch MCAP summaries during a full reconcile (the network-bound, out-of-transaction read). DB writes stay single-threaded; `1` = sequential. Higher values hide per-file round-trip latency on a remote bucket |
+| `--extract-workers` | `64` | threads that fetch MCAP summaries during a full reconcile (the network-bound, out-of-transaction read). DB writes stay single-threaded; `1` = sequential. Extraction is latency-bound (each file is 1–2 sequential range GETs), so throughput scales ~linearly with this until S3's per-prefix limits — the boto3 client's HTTP connection pool is sized to match (`max_pool_connections = max(workers, 10)`). Lower it for a tiny or local bucket |
 | `--debounce` | `2.0` | [local] seconds to debounce file events |
 | `--stability-checks` | `3` | [local] size-stability polls before cataloging |
 | `--stability-interval` | `0.5` | [local] seconds between stability polls |

mcap_catalog/mcap_catalog_builder/__main__.py

@@ -79,12 +79,15 @@ def build_parser() -> argparse.ArgumentParser:
                    help="[local] size-stability poll count before cataloging (default: 3)")
     p.add_argument("--stability-interval", type=float, default=0.5,
                    help="[local] seconds between size-stability polls (default: 0.5)")
-    p.add_argument("--extract-workers", type=int, default=8,
+    p.add_argument("--extract-workers", type=int, default=64,
                    help="number of threads that fetch MCAP summaries during a full "
                         "reconcile (the network-bound, out-of-transaction read). DB "
                         "writes stay single-threaded; this only parallelizes reads. "
-                        "1 = sequential. Higher values hide per-file round-trip "
-                        "latency on a remote bucket (default: 8).")
+                        "1 = sequential. Extraction is latency-bound (each file is "
+                        "1-2 sequential range GETs), so throughput scales ~linearly "
+                        "with this until S3's per-prefix limits — the S3 client's HTTP "
+                        "connection pool is sized to match. Lower it for a tiny or "
+                        "local bucket (default: 64).")
     p.add_argument("--log-level", default="INFO",
                    choices=["DEBUG", "INFO", "WARNING", "ERROR"])
     return p
@@ -165,7 +168,24 @@ def main(argv: list[str] | None = None) -> int:
         from .s3_storage import S3Source
         from .s3_producer import s3_event_producer
 
-        source = S3Source(boto3.client("s3"), args.s3_bucket, args.s3_prefix)
+        # Size the HTTP connection pool to the extract concurrency: extraction fans
+        # out --extract-workers threads of range GETs, and boto3's default pool of 10
+        # would throttle anything above ~10 (pool-full warnings + retries → SLOWER,
+        # not faster). Never drop below the default 10 for a low worker count.
+        # botocore is imported defensively so the s3 path stays exercisable with a
+        # bare fake boto3 (no botocore) in tests / no-AWS environments.
+        client_kwargs: dict = {}
+        try:
+            from botocore.config import Config
+
+            client_kwargs["config"] = Config(
+                max_pool_connections=max(args.extract_workers, 10)
+            )
+        except ImportError:
+            pass  # no botocore -> fall back to boto3's default connection pool
+        source = S3Source(
+            boto3.client("s3", **client_kwargs), args.s3_bucket, args.s3_prefix
+        )
 
         def start_producer() -> None:
             threading.Thread(

mcap_catalog/mcap_catalog_builder/tests/test_cli.py

@@ -34,6 +34,7 @@ def test_parser_defaults():
     assert args.stability_interval == 0.5
     assert args.log_level == "INFO"
     assert args.source == "local"  # default backend
+    assert args.extract_workers == 64  # wide default: extraction is latency-bound
 
 
 def test_parser_s3_options():
@@ -129,7 +130,9 @@ def test_main_s3_daemon_no_watch_skips_sqs_requirement(tmp_path, monkeypatch):
     import mcap_catalog_builder.s3_producer as s3_producer_mod
 
     fake_boto3 = types.ModuleType("boto3")
-    fake_boto3.client = lambda name: _EmptyS3Client() if name == "s3" else object()
+    # Real boto3.client accepts a `config=` kwarg (we pass a botocore Config to size
+    # the connection pool); the fake must tolerate it.
+    fake_boto3.client = lambda name, **kw: _EmptyS3Client() if name == "s3" else object()
     monkeypatch.setitem(sys.modules, "boto3", fake_boto3)
 
     producer_calls = []