manifest v2 for non co-located seg and meshes

pychunkedgraph/app/common.py

@@ -82,7 +82,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

@@ -11,8 +11,11 @@
 from pychunkedgraph import __version__
 from pychunkedgraph.app import app_utils
 from pychunkedgraph.graph import chunkedgraph
+from pychunkedgraph.graph import exceptions as cg_exceptions
 from pychunkedgraph.app.meshing import tasks as meshing_tasks
 from pychunkedgraph.meshing import meshgen
+from pychunkedgraph.meshing.mesh_meta import MeshMeta
+from pychunkedgraph.meshing.manifest import v2
 from pychunkedgraph.meshing.manifest import get_highest_child_nodes_with_meshes
 from pychunkedgraph.meshing.manifest import get_children_before_start_layer
 from pychunkedgraph.meshing.manifest import ManifestCache
@@ -68,13 +71,21 @@ def handle_get_manifest(table_id, node_id):
         bounding_box = np.array([b.split("-") for b in bounds.split("_")], dtype=int).T
 
     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"])
@@ -91,8 +102,12 @@ 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):
@@ -107,25 +122,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)
 
 
@@ -180,22 +202,15 @@ def handle_remesh(table_id):
 
 def _remeshing(serialized_cg_info, lvl2_nodes):
     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/meshing/tasks.py

@@ -1,27 +1,19 @@
 from pychunkedgraph.app import app_utils
-from pychunkedgraph.meshing import meshgen, meshgen_utils
+from pychunkedgraph.meshing import meshgen
+from pychunkedgraph.meshing.mesh_meta import MeshMeta
 import numpy as np
-import os
 
 
 def remeshing(table_id, lvl2_nodes):
     lvl2_nodes = np.array(lvl2_nodes, dtype=np.uint64)
     cg = app_utils.get_cg(table_id, skip_cache=True)
-
-    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,
     )

pychunkedgraph/app/segmentation/common.py

@@ -14,6 +14,7 @@
 
 from pychunkedgraph import __version__
 from pychunkedgraph.app import app_utils
+from pychunkedgraph.meshing.mesh_meta import MeshMeta
 from pychunkedgraph.graph import (
     attributes,
     cutting,
@@ -111,6 +112,13 @@ def handle_info(table_id):
         combined_info["mesh_dir"] = mesh_dir
     elif combined_info.get("mesh_dir", None) is not None:
         combined_info["mesh_dir"] = "graphene_meshes"
+    # 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 = dict(combined_info.get("mesh_metadata", {}))
+    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/basetypes.py

@@ -0,0 +1,6 @@
+"""Compat shim: ``basetypes`` lives at ``graph/utils/basetypes.py`` on this line
+but at ``graph/basetypes.py`` on the pcgv3 line. Re-export so the shorter import
+path resolves identically on both, keeping meshing code branch-agnostic.
+"""
+
+from .utils.basetypes import *  # noqa: F401,F403

pychunkedgraph/graph/exceptions.py

@@ -64,6 +64,11 @@ 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."""
     status_code = http_client.CONFLICT

pychunkedgraph/meshing/manifest/cache.py

@@ -12,6 +12,8 @@
 DOES_NOT_EXIST = "X"
 INITIAL_PATH_PREFIX = "initial_path_prefix"
 
+MANIFEST_TTL_SECONDS = 3 * 24 * 3600
+
 REDIS_HOST = os.environ.get("MANIFEST_CACHE_REDIS_HOST", "localhost")
 REDIS_PORT = os.environ.get("MANIFEST_CACHE_REDIS_PORT", "6379")
 REDIS_PASSWORD = os.environ.get("MANIFEST_CACHE_REDIS_PASSWORD", "")
@@ -74,6 +76,27 @@ def clear_fragments(self, node_ids) -> None:
         keys = [f"{self.namespace}:{n}" for n in node_ids]
         REDIS.delete(*keys)
 
+    def clear_namespace(self, batch_size: int = 1000) -> int:
+        """Delete every key under this graph_id's namespace.
+
+        SCAN-based pattern delete (non-blocking on the redis side).
+        Returns the number of keys deleted.
+        """
+        if REDIS is None:
+            return 0
+
+        pattern = f"{self.namespace}:*"
+        deleted = 0
+        batch = []
+        for key in REDIS.scan_iter(match=pattern, count=batch_size):
+            batch.append(key)
+            if len(batch) >= batch_size:
+                deleted += REDIS.delete(*batch)
+                batch.clear()
+        if batch:
+            deleted += REDIS.delete(*batch)
+        return deleted
+
     def _get_cached_initial_fragments(self, node_ids: List[np.uint64]):
         if REDIS is None:
             return {}, node_ids, []
@@ -140,10 +163,16 @@ def _set_cached_initial_fragments(
         for node_id, fragment_info in fragments_d.items():
             path, offset, size = fragment_info
             key = f"{self.namespace}:{node_id}"
-            pipeline.set(key, f"{path[prefix_idx:]}:{offset}:{size}")
+            pipeline.set(
+                key, f"{path[prefix_idx:]}:{offset}:{size}", ex=MANIFEST_TTL_SECONDS
+            )
 
         for node_id in not_existing:
-            pipeline.set(f"{self.namespace}:{node_id}", DOES_NOT_EXIST)
+            pipeline.set(
+                f"{self.namespace}:{node_id}",
+                DOES_NOT_EXIST,
+                ex=MANIFEST_TTL_SECONDS,
+            )
 
         pipeline.execute()
 
@@ -155,9 +184,15 @@ def _set_cached_dynamic_fragments(
 
         pipeline = REDIS.pipeline()
         for node_id, fragment in fragments_d.items():
-            pipeline.set(f"{self.namespace}:{node_id}", fragment)
+            pipeline.set(
+                f"{self.namespace}:{node_id}", fragment, ex=MANIFEST_TTL_SECONDS
+            )
 
         for node_id in not_existing:
-            pipeline.set(f"{self.namespace}:{node_id}", DOES_NOT_EXIST)
+            pipeline.set(
+                f"{self.namespace}:{node_id}",
+                DOES_NOT_EXIST,
+                ex=MANIFEST_TTL_SECONDS,
+            )
 
         pipeline.execute()

pychunkedgraph/meshing/manifest/sharded.py

@@ -8,8 +8,9 @@
 from .utils import get_children_before_start_layer
 from ...graph import ChunkedGraph
 from ...graph.types import empty_1d
-from ...graph.utils.basetypes import NODE_ID
+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,9 +14,10 @@
 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.utils.basetypes import NODE_ID
+from ...graph.basetypes import NODE_ID
 from ...graph.utils import generic as misc_utils
 
 
@@ -40,7 +41,7 @@ def _get_children(cg, node_ids: Sequence[np.uint64], children_cache: Dict):
     if len(node_ids) == 0:
         return empty_1d.copy()
     node_ids = np.array(node_ids, dtype=NODE_ID)
-    mask = np.in1d(node_ids, np.fromiter(children_cache.keys(), dtype=NODE_ID))
+    mask = np.isin(node_ids, np.fromiter(children_cache.keys(), dtype=NODE_ID))
     children_d = cg.get_children(node_ids[~mask])
     children_cache.update(children_d)
 
@@ -105,8 +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_dir = cg.meta.custom_data.get("mesh", {}).get("dir", "graphene_meshes")
-    mesh_path = f"{cg.meta.data_source.WATERSHED}/{mesh_dir}/dynamic"
+    mesh_path = MeshMeta(cg).dynamic_path
     cf = CloudFiles(mesh_path)
     manifest_cache = ManifestCache(cg.graph_id, initial=False)
 
@@ -180,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
@@ -194,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 = {}