pcgv3: manifest v2 for non co-located seg and meshes

pychunkedgraph/app/common.py

@@ -85,7 +85,7 @@ def after_request(response):
 
     response.data = compression.gzip_compress(response.data)
     response.headers["Content-Encoding"] = "gzip"
-    response.headers["Vary"] = "Accept-Encoding"
+    response.vary.add("Accept-Encoding")
     response.headers["Content-Length"] = len(response.data)
     return response
 

pychunkedgraph/app/meshing/common.py

@@ -9,6 +9,7 @@
 from pychunkedgraph import __version__
 from pychunkedgraph.app import app_utils
 from pychunkedgraph.graph import chunkedgraph
+from pychunkedgraph.graph import exceptions as cg_exceptions
 
 __meshing_url_prefix__ = os.environ.get("MESHING_URL_PREFIX", "meshing")
 
@@ -62,14 +63,25 @@ def handle_get_manifest(table_id, node_id):
         bounds = request.args["bounds"]
         bounding_box = np.array([b.split("-") for b in bounds.split("_")], dtype=int).T
 
+    from pychunkedgraph.meshing.mesh_meta import MeshMeta
+    from pychunkedgraph.meshing.manifest import v2
+
     cg = app_utils.get_cg(table_id)
+    mm = MeshMeta(cg)
+    manifest_version = v2.requested_manifest_version(request.headers.get("Accept"))
+    if manifest_version < 2 and mm.needs_v2:
+        raise cg_exceptions.NotAcceptable(
+            "This dataset serves meshes from an absolute path; upgrade your "
+            "client to one that requests "
+            "'Accept: application/x.cave;manifest_version=2'."
+        )
     verify = request.args.get("verify", False)
     verify = verify in ["True", "true", "1", True]
     return_seg_ids = request.args.get("return_seg_ids", False)
     prepend_seg_ids = request.args.get("prepend_seg_ids", False)
     return_seg_ids = return_seg_ids in ["True", "true", "1", True]
     prepend_seg_ids = prepend_seg_ids in ["True", "true", "1", True]
-    start_layer = cg.meta.custom_data.get("mesh", {}).get("max_layer", 2)
+    start_layer = mm.max_layer
     start_layer = int(request.args.get("start_layer", start_layer))
     if "start_layer" in data:
         start_layer = int(data["start_layer"])
@@ -86,13 +98,19 @@ def handle_get_manifest(table_id, node_id):
         flexible_start_layer,
         bounding_box,
         data,
+        manifest_version,
     )
-    return manifest_response(cg, args)
+    response = make_response(jsonify(manifest_response(cg, args)))
+    response.headers["X-Manifest-Version"] = str(manifest_version)
+    response.headers["Vary"] = "Accept"
+    return response
 
 
 def manifest_response(cg, args):
     from pychunkedgraph.meshing.manifest import speculative_manifest_sharded
     from pychunkedgraph.meshing.manifest import get_highest_child_nodes_with_meshes
+    from pychunkedgraph.meshing.manifest import v2
+    from pychunkedgraph.meshing.mesh_meta import MeshMeta
 
     (
         node_id,
@@ -103,25 +121,32 @@ def manifest_response(cg, args):
         flexible_start_layer,
         bounding_box,
         data,
+        manifest_version,
     ) = args
-    resp = {}
-    seg_ids = []
     if not verify:
-        seg_ids, resp["fragments"] = speculative_manifest_sharded(
+        seg_ids, fragments = speculative_manifest_sharded(
             cg, node_id, start_layer=start_layer, bounding_box=bounding_box
         )
-
     else:
-        seg_ids, resp["fragments"] = get_highest_child_nodes_with_meshes(
+        seg_ids, fragments = get_highest_child_nodes_with_meshes(
             cg,
             np.uint64(node_id),
             start_layer=start_layer,
             bounding_box=bounding_box,
         )
-    if prepend_seg_ids:
-        resp["fragments"] = [f"~{i}:{f}" for i, f in zip(seg_ids, resp["fragments"])]
-    if return_seg_ids:
-        resp["seg_ids"] = seg_ids
+
+    if manifest_version >= 2:
+        mm = MeshMeta(cg)
+        initial, dynamic = v2.to_v2_groups(seg_ids, fragments, prepend_seg_ids)
+        resp = v2.assemble(mm.initial_path, mm.dynamic_path, initial, dynamic)
+        if return_seg_ids:
+            resp["seg_ids"] = seg_ids
+    else:
+        resp = {"fragments": fragments}
+        if prepend_seg_ids:
+            resp["fragments"] = [f"~{i}:{f}" for i, f in zip(seg_ids, fragments)]
+        if return_seg_ids:
+            resp["seg_ids"] = seg_ids
     return _check_post_options(cg, resp, data, seg_ids)
 
 
@@ -155,24 +180,18 @@ def handle_remesh(table_id):
 def _remeshing(serialized_cg_info, lvl2_nodes):
     # nested: pulls meshing/cloudvolume, only needed at call time
     from pychunkedgraph.meshing import meshgen
+    from pychunkedgraph.meshing.mesh_meta import MeshMeta
 
     cg = chunkedgraph.ChunkedGraph(**serialized_cg_info)
-    cv_mesh_dir = cg.meta.dataset_info["mesh"]
-    cv_unsharded_mesh_dir = cg.meta.dataset_info["mesh_metadata"]["unsharded_mesh_dir"]
-    cv_unsharded_mesh_path = os.path.join(
-        cg.meta.data_source.WATERSHED, cv_mesh_dir, cv_unsharded_mesh_dir
-    )
-    mesh_data = cg.meta.custom_data["mesh"]
-
-    # TODO: stop_layer and mip should be configurable by dataset
+    mm = MeshMeta(cg)
     meshgen.remeshing(
         cg,
         lvl2_nodes,
-        stop_layer=mesh_data["max_layer"],
-        mip=mesh_data["mip"],
-        max_err=mesh_data["max_error"],
-        cv_sharded_mesh_dir=cv_mesh_dir,
-        cv_unsharded_mesh_path=cv_unsharded_mesh_path,
+        stop_layer=mm.max_layer,
+        mip=mm.mip,
+        max_err=mm.max_error,
+        cv_sharded_mesh_dir=mm.dir,
+        cv_unsharded_mesh_path=mm.dynamic_path,
     )
 
     return Response(status=200)

pychunkedgraph/app/segmentation/common.py

@@ -18,6 +18,7 @@
 
 logger = get_logger(__name__)
 from pychunkedgraph.app import app_utils
+from pychunkedgraph.meshing.mesh_meta import MeshMeta
 from pychunkedgraph.graph import attributes, cutting, segmenthistory, ChunkedGraph
 from pychunkedgraph.graph import (
     edges as cg_edges,
@@ -122,6 +123,11 @@ def handle_info(table_id):
     # the right dir. Copy the dict so cg.meta.dataset_info is untouched.
     mesh_metadata = dict(combined_info.get("mesh_metadata", {}))
     mesh_metadata["unsharded_mesh_dir"] = dynamic_dir
+    # Absolute mesh bucket locations (== the v2 manifest keys) so clients that
+    # bypass the manifest resolve initial/dynamic meshes from the real buckets.
+    mm = MeshMeta(cg)
+    mesh_metadata["initial_mesh_path"] = mm.initial_path
+    mesh_metadata["dynamic_mesh_path"] = mm.dynamic_path
     combined_info["mesh_metadata"] = mesh_metadata
     return jsonify(combined_info)
 

pychunkedgraph/graph/exceptions.py

@@ -63,6 +63,12 @@ class Forbidden(ClientError):
     status_code = http_client.FORBIDDEN
 
 
+class NotAcceptable(ClientError):
+    """Exception mapping a ``406 Not Acceptable`` response."""
+
+    status_code = http_client.NOT_ACCEPTABLE
+
+
 class Conflict(ClientError):
     """Exception mapping a ``409 Conflict`` response."""
 

pychunkedgraph/meshing/manifest/sharded.py

@@ -10,6 +10,7 @@
 from ...graph.types import empty_1d
 from ...graph.basetypes import NODE_ID
 from ...graph.chunks import utils as chunk_utils
+from ..mesh_meta import MeshMeta
 
 
 def verified_manifest(
@@ -61,7 +62,7 @@ def speculative_manifest(
     from ..meshgen_utils import get_json_info
 
     if start_layer is None:
-        start_layer = cg.meta.custom_data.get("mesh", {}).get("max_layer", 2)
+        start_layer = MeshMeta(cg).max_layer
 
     start = time()
     bounding_box = chunk_utils.normalize_bounding_box(
@@ -89,7 +90,7 @@ def speculative_manifest(
 
     readers = CloudVolume(  # pylint: disable=no-member
         "graphene://https://localhost/segmentation/table/dummy",
-        mesh_dir=cg.meta.custom_data.get("mesh", {}).get("dir", "graphene_meshes"),
+        mesh_dir=MeshMeta(cg).dir,
         info=get_json_info(cg),
     ).mesh.readers
 

pychunkedgraph/meshing/manifest/utils.py

@@ -14,6 +14,7 @@
 from .cache import ManifestCache
 from ..meshgen_utils import get_mesh_name
 from ..meshgen_utils import get_json_info
+from ..mesh_meta import MeshMeta
 from ...graph import ChunkedGraph
 from ...graph.types import empty_1d
 from ...graph.basetypes import NODE_ID
@@ -105,10 +106,7 @@ def _get_dynamic_meshes(cg, node_ids: Sequence[np.uint64]) -> Tuple[Dict, List]:
     if len(node_ids) == 0:
         return result, not_existing
 
-    mesh_meta = cg.meta.custom_data.get("mesh", {})
-    mesh_dir = mesh_meta.get("dir", "graphene_meshes")
-    dynamic_dir = mesh_meta.get("dynamic_mesh_dir", "dynamic")
-    mesh_path = f"{cg.meta.data_source.WATERSHED}/{mesh_dir}/{dynamic_dir}"
+    mesh_path = MeshMeta(cg).dynamic_path
     cf = CloudFiles(mesh_path)
     manifest_cache = ManifestCache(cg.graph_id, initial=False)
 
@@ -182,7 +180,7 @@ def segregate_node_ids(cg, node_ids):
     new = created by proofreading edit operations
     """
 
-    initial_ts = cg.meta.custom_data["mesh"]["initial_ts"]
+    initial_ts = MeshMeta(cg).initial_ts
     initial_mesh_dt = np.datetime64(datetime.fromtimestamp(initial_ts))
     node_ids_ts = cg.get_node_timestamps(node_ids)
     initial_mesh_mask = node_ids_ts < initial_mesh_dt
@@ -196,10 +194,17 @@ def get_mesh_paths(
     node_ids: Sequence[np.uint64],
     stop_layer: int = 2,
 ) -> Dict:
+    # Point the shard reader at initial_path: cloud-volume resolves shards at
+    # join(info["data_dir"], info["mesh"], "initial"), so reader_anchor splits
+    # initial_path into that (data_dir, mesh) pair — repoints a migrated bucket.
+    info = get_json_info(cg)
+    data_dir, mesh_dir = MeshMeta(cg).reader_anchor
+    info["data_dir"] = data_dir
+    info["mesh"] = mesh_dir
     shard_readers = CloudVolume(  # pylint: disable=no-member
         "graphene://https://localhost/segmentation/table/dummy",
-        mesh_dir=cg.meta.custom_data.get("mesh", {}).get("dir", "graphene_meshes"),
-        info=get_json_info(cg),
+        mesh_dir=mesh_dir,
+        info=info,
     ).mesh
 
     result = {}

pychunkedgraph/meshing/manifest/v2.py

@@ -0,0 +1,59 @@
+"""v2 graphene manifest: Accept-header negotiation + format assembly.
+
+Branch-agnostic, pure-stdlib. The v2 format keys ``fragments`` by absolute mesh
+bucket path so mesh location travels in the manifest itself; a client advertises
+support via ``Accept: application/x.cave;manifest_version=2``.
+"""
+
+ACCEPT_MEDIA_TYPE = "application/x.cave"
+MANIFEST_VERSION = 2  # highest manifest version this server produces
+
+
+def requested_manifest_version(accept_header, default: int = 1) -> int:
+    """Manifest version the client advertised via
+    ``Accept: application/x.cave;manifest_version=N``.
+
+    Parsed with stdlib — werkzeug's ``MIMEAccept`` mangles media-type params.
+    """
+    if not accept_header:
+        return default
+    for part in accept_header.split(","):
+        params = [p.strip() for p in part.split(";")]
+        if params[0].lower() != ACCEPT_MEDIA_TYPE:
+            continue
+        for param in params[1:]:
+            key, _, value = param.partition("=")
+            if key.strip().lower() == "manifest_version":
+                try:
+                    return int(value.strip())
+                except ValueError:
+                    return default
+    return default
+
+
+def to_v2_groups(node_ids, fragments, prepend_seg_ids):
+    """Group v1 fragments into ``(initial, dynamic)`` by the leading ``~`` marker.
+
+    Each fragment is emitted exactly as the v1 manifest would (seg id prepended
+    when requested); only the initial-vs-dynamic grouping is added. The raw ``~``
+    selects the group, matching v1's ``~<segid>:<fragment>``.
+    """
+    initial, dynamic = [], []
+    for node_id, frag in zip(node_ids, fragments):
+        out = f"~{node_id}:{frag}" if prepend_seg_ids else frag
+        (initial if frag.startswith("~") else dynamic).append(out)
+    return initial, dynamic
+
+
+def assemble(initial_path: str, dynamic_path: str, initial_frags, dynamic_frags) -> dict:
+    """v2 manifest dict: fragment lists grouped by absolute bucket path.
+
+    Empty groups are omitted; each bucket value is a sub-object so per-bucket
+    metadata can be added later without breaking the shape.
+    """
+    fragments = {}
+    if len(initial_frags):
+        fragments[initial_path] = {"fragments": list(initial_frags)}
+    if len(dynamic_frags):
+        fragments[dynamic_path] = {"fragments": list(dynamic_frags)}
+    return {"manifest_version": MANIFEST_VERSION, "fragments": fragments}

pychunkedgraph/meshing/mesh_meta/README.md

@@ -0,0 +1,69 @@
+# mesh_meta
+
+Single, branch-agnostic home for **graphene mesh metadata**. Every value comes
+from **`custom_data["mesh"]`, the single source of truth**. `MeshMeta(cg)` wraps a
+ChunkedGraph and exposes, as properties, the mesh dir/bucket locations and the
+per-dataset mesh params used by the v2 manifest and by remeshing; multi-resolution
+(LOD) mesh meta will live here too as it lands, so this is the module to grow
+rather than scattering mesh-meta reads across the code.
+
+## Why it exists
+
+Supervoxels (watershed) can move to a cheap, egress-waived bucket while meshes
+stay put. The v2 graphene manifest therefore carries **absolute mesh bucket
+paths** rather than paths composed relative to the watershed `data_dir`, so a
+client fetches meshes straight from wherever they live. This module is the one
+place that decides where meshes live, and what the mesh params are, for a given
+ChunkedGraph.
+
+## Two mesh locations
+
+- **initial** — the sharded meshes produced at ingest. **Shared / graph-independent**:
+  several ChunkedGraphs (e.g. copies) point at the *same* initial meshes, so this
+  location is never namespaced by graph id.
+- **dynamic** — the unsharded meshes produced by proofreading edits.
+  **Per-graph**: each graph keeps its own, so on the pcgv3 line it defaults to a
+  graph-id-derived dir (`dynamic_<graph_id>`) and stays isolated even while
+  sharing initial meshes; on the pcgv2 line it is a plain subdir.
+
+## Configuration
+
+All read from `custom_data["mesh"]`:
+
+- `dir` — the mesh root under the watershed (default `graphene_meshes`).
+- `initial_mesh_dir` — the initial location (default `initial`).
+- `dynamic_mesh_dir` — the dynamic location (default `dynamic`; pcgv3 setup fills
+  `dynamic_<graph_id>`).
+- `max_layer`, `mip`, `max_error` — mesh params (start layer for manifests, and
+  the mip / max-error used by remeshing).
+- `initial_ts` — the ingest timestamp boundary that splits initial (sharded) node
+  ids from proofread (dynamic) ones.
+
+Resolution rule for the two location values: a value containing a cloud scheme
+(`gs://`, `s3://`, …) is treated as an **absolute** bucket and used as-is;
+otherwise it is joined under `<watershed>/<dir>`. This lets a dataset move
+initial and/or dynamic meshes to different buckets without touching anything else.
+
+Branch-agnostic on purpose: this module reads only `custom_data["mesh"]` and the
+watershed path — accessors identical on the pcgv2 and pcgv3 lines — so it
+cherry-picks clean between them. How that config gets *populated* (e.g. pcgv3's
+setup-time `MeshConfig` from the dataset yaml) is branch-specific and lives
+upstream; this module only reads the result.
+
+## API
+
+`MeshMeta(cg)` exposes these properties:
+
+- `initial_path` → absolute dir of the initial (sharded) meshes.
+- `dynamic_path` → absolute dir of the dynamic (unsharded) meshes.
+- `dir` → the sharded mesh dir name under the watershed.
+- `max_layer`, `mip`, `max_error`, `initial_ts` → the mesh params.
+- `needs_v2` → true when either location falls **outside** the watershed dir,
+  i.e. meshes are not co-located with the watershed. Old clients compose paths
+  relative to the watershed, so they cannot reach such meshes: the v2 manifest
+  serves them correctly, and the v1 path must fail loudly rather than return
+  unreachable paths.
+- `reader_anchor` → the `(data_dir, mesh)` pair to hand a CloudVolume mesh reader.
+  The reader appends `initial/` internally, so it must be anchored at the
+  **parent** of the initial dir; this yields shard byte-ranges from the right
+  bucket even for a migrated graph.

pychunkedgraph/meshing/mesh_meta/__init__.py

@@ -0,0 +1,5 @@
+"""Branch-agnostic graphene mesh metadata. See README.md."""
+
+from .core import MeshMeta
+
+__all__ = ["MeshMeta"]