Batch catalog metadata and publish reusable snapshots on wire 2.2

Author: rustyconover

docs/catalog_publication.md

@@ -0,0 +1,171 @@
+# Batched catalog contents and reusable publications
+
+VGI wire revision **2.2.0** retains the `vgi.v2` route and replaces
+`catalog_contents` with six catalog-wide Arrow IPC streams: schemas, tables,
+views, functions, macros and indexes. Each stream can contain multiple batches.
+Null means an empty category. Function and macro subtype fields remain intact.
+Existing schema-level RPCs retain their previous collection encoding.
+
+Function metadata stores `arguments`, `output_schema` and
+`parameter_default_values` as `dictionary<int32, binary>` columns. Repeated
+serialized schemas and typed defaults share dictionary entries by their byte
+content. Python callers continue to use bytes for schemas and an optional
+`RecordBatch` for defaults. Dictionary replacements across record batches are
+supported; values remain opaque binary data rather than text.
+Single-object and schema-level function RPCs use these same updated rows.
+
+`SchemaInfo` no longer contains attachment data on the wire. Existing Python
+authors may still pass `attach_opaque_data=` to its constructor during migration;
+the argument is discarded. Use `CatalogAttachResult.attach_opaque_data` for
+subsequent requests. Worker functions still receive their attachment state.
+
+Every minor wire revision is an explicit contract boundary. The server checks
+major and minor exactly, so clients and workers must both use 2.2. A patch bump
+does not permit schema changes. Serving 2.1 and 2.2 during migration requires
+separate deployment URLs or an explicit compatibility implementation.
+
+## Enable publication
+
+Publication is opt-in through `CatalogInterface.catalog_snapshot_policy`.
+Ordinary delivery and the existing frozen-catalog response cache remain the
+default. Published delivery uses the existing vgi-rpc `ExternalRef` mechanism;
+clients resolve the external object on all transports, including subprocess.
+
+Install `vgi-python[external]` on publishing workers and clients that resolve
+publications; add `http` when serving or connecting over HTTP, for example
+`pip install 'vgi-python[http,external]'`. The `external` extra supplies the RPC
+HTTP downloader, retry and compression dependencies. Supply your own configured
+S3 client and its SDK when using the S3 repository.
+
+The catalog implementation uses APIs available in published **vgi-rpc 0.50.0**.
+An isolated check against that release passed 157 functional catalog, client,
+serialization, binary dictionary and publication tests. The package retains its
+declared RPC minimum of 0.52.0 for Identity.v2 support; catalog publication does
+not introduce a dependency on unpublished RPC APIs.
+
+```python
+from vgi.catalog.snapshot import (
+    CatalogSnapshotPolicy,
+    S3CatalogPublicationRepository,
+    SnapshotIdentity,
+)
+from vgi_rpc.external import Compression
+
+# s3 is your configured boto3-compatible S3/R2 client. The client must support
+# conditional PutObject with IfNoneMatch and IfMatch.
+repository = S3CatalogPublicationRepository(
+    s3,
+    bucket="catalog-snapshots",
+    profile="production-r2-gzip",
+    url_ttl_seconds=3600,
+    retention_seconds=86400,
+)
+
+class MyCatalog(ExistingCatalog):
+    catalog_snapshot_policy = CatalogSnapshotPolicy(
+        delivery="published",
+        repository=repository,
+        identity=lambda request: SnapshotIdentity(
+            catalog_namespace="my-catalog",
+            source_revision=DEPLOYMENT_REVISION,
+            catalog_version=request.catalog_version,
+            # Omit this for the safe attachment-local default. Sharing requires
+            # an explicit guarantee that metadata is identical for this scope.
+            visibility_scope="public-default-options",
+        ),
+        compression=Compression(algorithm="gzip"),
+    )
+```
+
+The identity hook must be cheap and must change `source_revision` whenever
+metadata changes. A dynamic catalog can use a schema generation rather than a
+deployment revision. Include options, resolved implementation/data versions,
+tenant or principal, and any other metadata visibility differences in its scope.
+Do not put credentials into names or scope strings. Omitting `visibility_scope`
+isolates publications by a hash of the attachment envelope. Catalog version zero
+still requires a meaningful source revision.
+
+An optional `validator(request) -> str | None` can answer a matching ETag before
+repository access or inventory construction. Every request unwraps and authorizes
+its attachment first, even on a cache hit. The framework also checks that version
+and source identity remain unchanged while it builds a publication.
+
+## Publication lifecycle
+
+The framework writes a complete unary RPC envelope with one `result: binary`
+row. That binary contains the catalog response and its six nested streams.
+The stored object is compressed according to the policy and includes a mandatory
+SHA-256 over the raw envelope. Before committing its manifest, the framework
+fetches the actual HTTP URL and validates the digest, envelope, nested schemas,
+all rows/batches, schema membership, subtype fields and metadata equality.
+Raw and stored byte limits apply before upload and during readback, including
+when the caller supplies a custom fetch policy.
+
+The repository stores small trusted manifests separately from immutable objects.
+Keys include routing name, major/minor wire revision, generated canonical catalog
+contract hash, codec revision, namespace, source revision, visibility scope,
+catalog version, and storage/compression profile. A new worker process looks up
+the manifest before collecting metadata, then validates it with one HTTP fetch.
+A warm hit performs no inventory build, serialization, compression, upload or
+object GET. A client receiving the pointer still downloads the result as needed.
+
+`S3CatalogPublicationRepository` uses conditional manifest writes and tombstones
+for invalidation. It can re-sign a retained object without rebuilding it.
+Configure object lifecycle rules to retain objects for at least the declared
+retention interval; keep manifest retention separate. Signed URLs in manifests
+are credentials, so the manifest store must be access controlled. Newly changed
+retrieval URLs are validated before serving them.
+
+`MemoryCatalogPublicationRepository` supports a supplied RPC uploader for local
+use. It requires an explicit URL lifetime or stable-location declaration.
+It does not provide persistence across process restarts. A process coordinates
+one build per key; conditional repository commit makes separate publishers
+converge but can permit duplicate initial uploads across processes. Storage
+lifecycle rules should collect unused losing uploads.
+
+The manager bounds its in-memory manifest cache. URL or object-retention expiry
+causes renewal or rebuilding. Recovery of a corrupt or missing object
+conditionally invalidates that manifest and attempts a fresh publication.
+Publication errors fail the RPC by default. Set
+`on_publication_failure="ordinary"` to explicitly permit a newly validated
+ordinary full response when publication fails.
+
+Workers expose `warm_catalog_contents(attach_opaque_data)`,
+`import_catalog_publication(attach_opaque_data, manifest)` and
+`invalidate_catalog_publication(attach_opaque_data, manifest_id)` for authorized
+operational use. Import uses the same HTTP validation as recovery. A failed client
+download does not automatically notify the worker; use conditional invalidation
+to repair a known bad publication without deleting a concurrent replacement.
+
+## Validation
+
+Run the focused Python checks with the project environment on `PATH`:
+
+```sh
+uv run pytest tests/catalog/test_catalog_contents.py \
+  tests/catalog/test_catalog_publication.py \
+  tests/catalog/test_client_load_catalog.py tests/catalog/test_serialization.py
+```
+
+The publication tests use real HTTP object retrieval with uncompressed, gzip and
+zstd results. They exercise restart recovery, singleflight, default attachment
+isolation, corrupt publication, explicit fallback, authentication before cached
+delivery, caps, generation changes, manifest import and conditional invalidation.
+The S3 fixture verifies discovery and conditional writes across repository
+instances. Configure a real S3/R2 bucket and deployment CORS before production
+rollout; the in-memory conditional object fixture does not validate a vendor's
+credentials, CORS settings or lifecycle configuration.
+
+The installable native/browser fixture is:
+
+```sh
+python -m vgi._test_fixtures.catalog_publication
+```
+
+It accepts ordinary fixture worker options, including `--http`. Only
+`contents_probe` uses publication; other standard fixture catalogs retain their
+normal behavior. Set `VGI_CATALOG_PUBLICATION_PORT` to expose the local object
+server at a predictable port; `/stats` reports uploads and object GETs. Its CORS
+headers and preflight support permit browser conformance tests. This fixture is
+intended for local testing and its object storage lasts only for the worker's
+lifetime.

pyproject.toml

@@ -58,6 +58,9 @@ dependencies = [
 haybarn = ["haybarn==1.5.5rc1"]
 duckdb = ["duckdb==1.5.5"]
 http = ["vgi-rpc[http]"]
+# Publishing and resolving immutable external catalog results needs the RPC
+# download/compression dependencies on every transport, including subprocess.
+external = ["vgi-rpc[external]"]
 iroh = ["vgi-rpc[http,iroh]"]
 # transport="launch": vgi_rpc.launcher imports filelock unconditionally at
 # module level, which lives behind vgi-rpc's own `cli` extra (alongside

scripts/catalog_publication_fixture.py

@@ -0,0 +1,7 @@
+#!/usr/bin/env python3
+"""Run the installable catalog publication integration fixture."""
+
+from vgi._test_fixtures.catalog_publication import main
+
+if __name__ == "__main__":
+    main()

tests/catalog/test_catalog_contents.py

@@ -5,8 +5,10 @@
 from __future__ import annotations
 
 import re
+from dataclasses import asdict
 from typing import Any
 
+import pyarrow as pa
 import pytest
 
 from vgi._test_fixtures.catalog import InMemoryCatalog, SchemaData
@@ -32,8 +34,9 @@
     SchemaInfo,
     SchemaObjectType,
 )
+from vgi.catalog.contents_codec import CATEGORY_TYPES, decode_inventory
 from vgi.meta_worker import MetaWorker
-from vgi.protocol import CatalogAttachRequest, CatalogContentsResponse, SchemaContents
+from vgi.protocol import CatalogAttachRequest, CatalogContentsResponse, SchemaContents, _item_ipc_bytes
 from vgi.worker import Worker, catalog_contents_digest
 
 _KINDS = (
@@ -74,6 +77,17 @@ def _roundtrip(response: CatalogContentsResponse) -> CatalogContentsResponse:
     return CatalogContentsResponse.deserialize_from_bytes(response.serialize_to_bytes())
 
 
+def _entries(response: CatalogContentsResponse) -> list[SchemaContents]:
+    return [
+        SchemaContents(
+            path=list(info.schema.path),
+            schema=_item_ipc_bytes(info.schema),
+            **{kind: [_item_ipc_bytes(item) for item in getattr(info, kind)] for kind in _KINDS},
+        )
+        for info in decode_inventory(response)
+    ]
+
+
 # ---------------------------------------------------------------------------
 # Wire shape
 # ---------------------------------------------------------------------------
@@ -82,12 +96,18 @@ def _roundtrip(response: CatalogContentsResponse) -> CatalogContentsResponse:
 class TestShape:
     """The typed response: a struct per schema, items byte-identical to the per-schema RPCs."""
 
-    def test_schemas_is_a_list_of_structs(self) -> None:
-        """``schemas`` is ``list<struct>`` with ``path`` first, not a list of IPC blobs."""
-        field = CatalogContentsResponse.ARROW_SCHEMA.field("schemas")
-        assert field.type.value_type.num_fields == 10
-        assert [f.name for f in field.type.value_type] == ["path", "schema", *_KINDS]
-        assert CatalogContentsResponse.ARROW_SCHEMA.names == ["catalog_version", "etag", "not_modified", "schemas"]
+    def test_categories_are_nullable_binary_streams(self) -> None:
+        """Every category is one nullable typed collection stream."""
+        assert CatalogContentsResponse.ARROW_SCHEMA.names == [
+            "catalog_version",
+            "etag",
+            "not_modified",
+            *CATEGORY_TYPES,
+        ]
+        for category in CATEGORY_TYPES:
+            field = CatalogContentsResponse.ARROW_SCHEMA.field(category)
+            assert field.nullable and field.type == pa.binary()
+        assert "attach_opaque_data" not in SchemaInfo.ARROW_SCHEMA.names
 
     def test_items_match_the_per_schema_rpcs(self) -> None:
         """Every kind of every schema is byte-identical to its per-schema RPC answer."""
@@ -96,15 +116,15 @@ def test_items_match_the_per_schema_rpcs(self) -> None:
         response = _roundtrip(worker.catalog_contents(attach))
         assert response.etag is None
         assert not response.not_modified
-        assert [entry.path for entry in response.schemas] == [["main"], ["extra"]]
+        assert [entry.path for entry in _entries(response)] == [["main"], ["extra"]]
         schema_items = worker.catalog_schemas(attach).items
-        assert [entry.schema for entry in response.schemas] == schema_items
-        for entry in response.schemas:
+        assert [entry.schema for entry in _entries(response)] == schema_items
+        for entry in _entries(response):
             assert SchemaInfo.deserialize_from_bytes(entry.schema).path == entry.path
             expected = _per_schema_items(worker, attach, entry.path)
             for kind in _KINDS:
                 assert getattr(entry, kind) == expected[kind], (entry.path, kind)
-        main = response.schemas[0]
+        main = _entries(response)[0]
         # Every seeded kind is present in the fixture.
         for kind in _KINDS[:-1]:
             assert getattr(main, kind), kind
@@ -126,8 +146,8 @@ class NestedWorker(ContentsProbeWorker):
 
         worker = NestedWorker()
         response = _roundtrip(worker.catalog_contents(_attach(worker, CATALOG_PROBE)))
-        assert [entry.path for entry in response.schemas] == [["a"], ["a", "b"], ["a", "b", "c"]]
-        for entry in response.schemas:
+        assert [entry.path for entry in _entries(response)] == [["a"], ["a", "b"], ["a", "b", "c"]]
+        for entry in _entries(response):
             assert SchemaInfo.deserialize_from_bytes(entry.schema).path == entry.path
 
     @pytest.mark.parametrize(
@@ -186,14 +206,14 @@ def test_cheap_validator_short_circuits_without_building(self, monkeypatch: pyte
         full = worker.catalog_contents(attach)
         assert full.etag == "gen-1"
         assert not full.not_modified
-        assert [entry.path for entry in full.schemas] == [["main"]]
+        assert [entry.path for entry in _entries(full)] == [["main"]]
 
         with monkeypatch.context() as m:
             _forbid_build(m)
             hit = _roundtrip(worker.catalog_contents(attach, if_none_match="gen-1"))
         assert hit.not_modified
         assert hit.etag == "gen-1"
-        assert hit.schemas == []
+        assert hit.schemas is None
         assert hit.catalog_version == 1
 
         # A different etag gets the full contents.
@@ -215,7 +235,7 @@ def test_ddl_changes_the_validator(self) -> None:
         assert not response.not_modified
         assert response.etag == "gen-2"
         assert response.catalog_version == 2
-        assert [entry.path for entry in response.schemas] == [["main"], ["other"]]
+        assert [entry.path for entry in _entries(response)] == [["main"], ["other"]]
 
     def test_no_etag_ignores_if_none_match(self) -> None:
         """A catalog without an etag always answers in full, with etag null."""
@@ -224,7 +244,7 @@ def test_no_etag_ignores_if_none_match(self) -> None:
         response = worker.catalog_contents(attach, if_none_match="anything")
         assert response.etag is None
         assert not response.not_modified
-        assert [entry.path for entry in response.schemas] == [["main"]]
+        assert [entry.path for entry in _entries(response)] == [["main"]]
         # contents_memory reports version 0: exercises the client's version-0 rule.
         assert response.catalog_version == 0
 
@@ -245,7 +265,7 @@ class BuiltWorker(ContentsProbeWorker):
         attach = _attach(worker, CATALOG_PROBE)
         response = worker.catalog_contents(attach, if_none_match="v7")
         assert response.not_modified
-        assert response.schemas == []
+        assert response.schemas is None
         assert response.etag == "v7"
 
     @pytest.mark.parametrize(
@@ -317,11 +337,11 @@ def test_hash_is_deterministic_and_revalidates(self) -> None:
         second = worker.catalog_contents(attach)
         assert second.etag == first.etag
         assert second.schemas == first.schemas
-        assert catalog_contents_digest(first.schemas) == first.etag
+        assert catalog_contents_digest(first) == first.etag
 
         hit = worker.catalog_contents(attach, if_none_match=first.etag)
         assert hit.not_modified
-        assert hit.schemas == []
+        assert hit.schemas is None
         assert hit.etag == first.etag
 
         cat = worker._get_catalog()
@@ -330,7 +350,7 @@ def test_hash_is_deterministic_and_revalidates(self) -> None:
         changed = worker.catalog_contents(attach, if_none_match=first.etag)
         assert not changed.not_modified
         assert changed.etag != first.etag
-        assert [entry.path for entry in changed.schemas] == [["main"], ["other"]]
+        assert [entry.path for entry in _entries(changed)] == [["main"], ["other"]]
 
     def test_off_by_default(self) -> None:
         """Without the opt-in a catalog gets no etag."""
@@ -357,9 +377,9 @@ class OwnWorker(ContentsProbeWorker):
 
     def test_digest_separates_fields(self) -> None:
         """Moving bytes between fields changes the digest."""
-        a = [SchemaContents(path=["s"], schema=b"x", tables=[b"ab"])]
-        b = [SchemaContents(path=["s"], schema=b"x", tables=[b"a", b"b"])]
-        c = [SchemaContents(path=["s"], schema=b"x", views=[b"ab"])]
+        a = CatalogContentsResponse(catalog_version=1, schemas=b"x", tables=b"ab")
+        b = CatalogContentsResponse(catalog_version=1, schemas=b"xa", tables=b"b")
+        c = CatalogContentsResponse(catalog_version=1, schemas=b"x", views=b"ab")
         assert len({catalog_contents_digest(a), catalog_contents_digest(b), catalog_contents_digest(c)}) == 3
 
 
@@ -397,12 +417,7 @@ def test_cached_bytes_equal_a_fresh_encoding(self) -> None:
         """The pre-serialized response is byte-identical to encoding it normally."""
         worker = ContentsProbeWorker()
         cached = worker.catalog_contents(_attach(worker, CATALOG_PROBE))
-        fresh = CatalogContentsResponse(
-            catalog_version=cached.catalog_version,
-            etag=cached.etag,
-            not_modified=cached.not_modified,
-            schemas=list(cached.schemas),
-        )
+        fresh = CatalogContentsResponse(**asdict(cached))
         assert cached.serialize_to_bytes() == fresh.serialize_to_bytes()
         assert _roundtrip(cached) == fresh
 
@@ -436,7 +451,7 @@ class HashedWorker(ContentsProbeWorker):
         assert full.etag is not None
         hit = worker.catalog_contents(attach, if_none_match=full.etag)
         assert hit.not_modified
-        assert hit.schemas == []
+        assert hit.schemas is None
         assert worker.catalog_contents(attach, if_none_match="stale") is full
         assert len(calls) == 1