memory improvements to changelog and uwsgi reload-on-rss fix

pychunkedgraph/app/common.py

@@ -17,6 +17,37 @@
 ENABLE_LOGS = os.environ.get("PCG_SERVER_ENABLE_LOGS", "") != ""
 LOG_LEAVES_MANY = os.environ.get("PCG_SERVER_LOGS_LEAVES_MANY", "") != ""
 
+# Health-check paths to skip, matched EXACTLY (not as prefixes) so the start-log signal
+# isn't flooded by probes. These are the literal probe paths from the pychunkedgraph chart:
+# the read/write deployments' readiness+liveness probes hit "/segmentation" and the GCP load
+# balancer health check hits "/". Real API traffic lives under "/segmentation/api/..." and
+# "/meshing/api/...", which are NOT equal to these entries and so are still logged.
+_REQUEST_START_SKIP_PATHS = frozenset(("/", "/segmentation"))
+
+
+def _log_request_start():
+    # Emit a line to stdout at the *start* of a request, before any work runs, so it is
+    # captured by Cloud Logging even if the request goes on to OOM-kill or otherwise crash
+    # its worker before completing (such requests never reach after_request and so are
+    # invisible in the Datastore server_logs completion logs). The `content_length` field is
+    # the on-the-wire request body size (e.g. for a roots_binary POST, ~8 bytes per node id)
+    # and `pid` is the uwsgi worker, so a spike/OOM can be traced to the specific in-flight
+    # request and worker. Gated by the LOG_REQUEST_START Flask config value (default False);
+    # verbose, intended for temporary diagnosis.
+    try:
+        user_id = g.auth_user["id"]
+    except (AttributeError, KeyError):
+        user_id = USER_NOT_FOUND
+    current_app.logger.info(
+        "REQUEST_START pid=%s method=%s path=%s content_length=%s user=%s remote=%s",
+        os.getpid(),
+        request.method,
+        request.path,
+        request.content_length,
+        user_id,
+        request.remote_addr,
+    )
+
 
 def _log_request(response_time):
     try:
@@ -58,6 +89,10 @@ def before_request():
     current_app.table_id = None
     current_app.operation_id = None
     current_app.request_type = None
+    if current_app.config.get("LOG_REQUEST_START", False) and (
+        request.path not in _REQUEST_START_SKIP_PATHS
+    ):
+        _log_request_start()
     content_encoding = request.headers.get("Content-Encoding", "")
     if "gzip" in content_encoding.lower():
         request.data = compression.decompress(request.data, "gzip")

pychunkedgraph/app/config.py

@@ -14,6 +14,26 @@ class BaseConfig(object):
     LOGGING_DATEFORMAT = "%Y-%m-%dT%H:%M:%S.0Z"
     LOGGING_LEVEL = logging.DEBUG
 
+    # Opt-in start-of-request logging (see pychunkedgraph.app.common._log_request_start).
+    # When True, every non-probe request emits a REQUEST_START line to stdout before any work
+    # runs, so requests that OOM-kill their worker mid-flight (and thus never reach
+    # after_request) are still visible in Cloud Logging. Verbose; enable only for temporary
+    # diagnosis, e.g. by setting LOG_REQUEST_START = True in the instance config.cfg.
+    LOG_REQUEST_START = False
+
+    # Reject /lvl2_graph requests whose node resolves to more than this many level 2 nodes
+    # (see pychunkedgraph.graph.analysis.pathing.get_lvl2_edge_list). Such objects — typically
+    # erroneous mega-merges — produce a multi-GB induced edge list that can OOM the worker.
+    # None disables the guard; set a concrete integer in the instance config.cfg to enable.
+    LVL2_GRAPH_MAX_NODES = None
+
+    # Guard for /subgraph (see pychunkedgraph.graph.subgraph.get_subgraph_edges_and_leaves).
+    # Counts chunks rather than level 2 nodes: the endpoint reads every edge in every chunk the
+    # object touches (all objects in the chunk, not just the requested one), so cost tracks the
+    # volume queried, not the object. A large 'bounds' is expensive even for a small object.
+    # None disables the guard; set a concrete integer in the instance config.cfg to enable.
+    SUBGRAPH_MAX_CHUNKS = None
+
     CHUNKGRAPH_INSTANCE_ID = "pychunkedgraph"
     PROJECT_ID = os.environ.get("PROJECT_ID", None)
     CG_READ_ONLY = os.environ.get("CG_READ_ONLY", None) is not None

pychunkedgraph/app/meshing/common.py

@@ -1,18 +1,16 @@
 # pylint: disable=invalid-name, missing-docstring
 import json
 import os
-import threading
 
 import numpy as np
 import redis
 from rq import Queue, Connection, Retry
-from flask import Response, current_app, jsonify, make_response, request
+from flask import Response, current_app, g, jsonify, make_response, request
 
 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.manifest import get_highest_child_nodes_with_meshes
 from pychunkedgraph.meshing.manifest import get_children_before_start_layer
 from pychunkedgraph.meshing.manifest import ManifestCache
@@ -75,9 +73,20 @@ def handle_get_manifest(table_id, node_id):
     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 = int(request.args.get("start_layer", start_layer))
-    if "start_layer" in data:
-        start_layer = int(data["start_layer"])
+    raw_start_layer = data.get("start_layer", request.args.get("start_layer", start_layer))
+    try:
+        start_layer = int(raw_start_layer)
+    except (TypeError, ValueError):
+        raise cg_exceptions.BadRequest(
+            f"start_layer must be an integer, got {raw_start_layer!r}."
+        )
+
+    # Meshes only exist from layer 2 upwards. Below that this endpoint cannot return anything:
+    if start_layer < 2:
+        raise cg_exceptions.BadRequest(
+            f"start_layer must be at least 2, got {start_layer}. Meshes exist from layer 2 "
+            "upwards, so a lower value can only ever produce an empty manifest."
+        )
 
     flexible_start_layer = None
     if "flexible_start_layer" in data:
@@ -142,9 +151,64 @@ def _check_post_options(cg, resp, data, seg_ids):
 
 
 ## REMESHING -----------------------------------------------------
+def publish_remesh(table_id: str, user_id: str, lvl2_ids, is_priority: bool = True):
+    """Enqueue a remesh onto the same Pub/Sub topic the edit path publishes to.
+
+    Mirrors segmentation.common.publish_edit deliberately: one topic, and the
+    `remesh_priority` attribute is what routes a message to a subscription. The
+    infrastructure defines those subscriptions with attribute filters
+    (terraform-google-cave/modules/local_cluster/pubsub.tf):
+
+        <prefix>_PCG_HIGH_PRIORITY_REMESH   remesh_priority="true"   -> meshworker
+        <prefix>_PCG_LOW_PRIORITY_REMESH    remesh_priority="false"  -> remeshworker
+
+    so priority here is not a hint, it selects the consumer fleet.
+
+    Note the same topic also feeds <prefix>_<ws>_L2CACHE_{HIGH,LOW}_PRIORITY_TRIGGER, so a
+    manual remesh now also refreshes the l2 cache for these ids. That is intended -- a manual
+    remesh usually follows a data problem, and the l2 cache derives from the same chunks -- but
+    it is a real fan-out, not a no-op.
+    """
+    import pickle
+
+    from messagingclient import MessagingClient
+
+    attributes = {
+        "table_id": table_id,
+        "user_id": user_id,
+        "remesh_priority": "true" if is_priority else "false",
+        "remesh": "true",
+    }
+    payload = {
+        # 0 means "no operation". A manual remesh has no GraphEditOperation behind it, and the
+        # graph does not record which operation created a given level 2 node -- OperationID is
+        # written only onto root-id rows (edits.py::_update_root_id_lineage), so the best available
+        # answer is the latest operation on the whole object, which is not this node's provenance.
+        # A wrong id in the worker's log line is worse than an honest unknown.
+        #
+        # The key must still exist and be int-convertible: mesh_worker.callback does
+        # int(data["operation_id"]) unconditionally.
+        "operation_id": 0,
+        "new_lvl2_ids": np.asarray(lvl2_ids, dtype=np.uint64).tolist(),
+        # Neither consumer reads these (mesh_worker uses new_lvl2_ids, the l2cache trigger uses
+        # new_lvl2_ids); present so the payload shape stays identical to publish_edit's.
+        "new_root_ids": [],
+        "old_root_ids": [],
+    }
+
+    exchange = os.getenv("PYCHUNKEDGRAPH_EDITS_EXCHANGE", "pychunkedgraph")
+    c = MessagingClient()
+    c.publish(exchange, pickle.dumps(payload), attributes)
+
+
 def handle_remesh(table_id):
     current_app.request_type = "remesh_enque"
     current_app.table_id = table_id
+    # Same `priority` parameter, default, and semantics as every edit endpoint in
+    # segmentation.common, so the two paths cannot drift. Unset means high priority, which is
+    # the right default for an interactive request; a programmatic caller (caveclient, a
+    # backfill script) should pass priority=false so bulk work lands on the low-priority
+    # subscription and cannot starve human-triggered remeshes.
     is_priority = request.args.get("priority", True, type=str2bool)
     is_redisjob = request.args.get("use_redis", False, type=str2bool)
 
@@ -166,38 +230,23 @@ def handle_remesh(table_id):
 
         return jsonify(response_object), 202
     else:
+        # Publish, don't mesh here. This used to run meshgen.remeshing in a threading.Thread
+        # inside the api pod, which put an unbounded, unretryable, invisible workload in a
+        # request-serving process: the 202 was already returned, so a failure left no trace, and
+        # a worker recycle, rollout or HPA scale-down silently discarded the work. Measured on
+        # api6 2026-08-23, one remesh took the meshing pod from 198Mi to a 491Mi peak and left
+        # it at 420Mi -- rss does not fall back -- so pods ratcheted up until they died: one had
+        # reached 847Mi after 14h and was OOMKilled (whole cgroup, supervisord included) when a
+        # remesh pushed it past its 1536Mi limit. The mesh workers exist for exactly this work,
+        # request 3000Mi, and get retries and dead-lettering from Pub/Sub.
         new_lvl2_ids = np.array(new_lvl2_ids, dtype=np.uint64)
-        cg = app_utils.get_cg(table_id)
 
         if len(new_lvl2_ids) > 0:
-            t = threading.Thread(
-                target=_remeshing, args=(cg.get_serialized_info(), new_lvl2_ids)
-            )
-            t.start()
+            user_id = str(g.auth_user.get("id", current_app.user_id))
+            publish_remesh(table_id, user_id, new_lvl2_ids, is_priority=is_priority)
 
         return Response(status=202)
 
-
-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
-    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,
-    )
-
     return Response(status=200)
 
 

pychunkedgraph/app/segmentation/common.py

@@ -4,7 +4,6 @@
 import os
 import time
 from datetime import datetime
-from functools import reduce
 from collections import deque, defaultdict
 
 import numpy as np
@@ -785,8 +784,9 @@ def handle_subgraph(table_id, root_id, only_internal_edges=True):
         int(root_id),
         bbox=bounding_box,
         bbox_is_coordinate=True,
+        max_num_chunks=current_app.config.get("SUBGRAPH_MAX_CHUNKS"),
     )
-    edges = reduce(lambda x, y: x + y, edges, cg_edges.Edges([], []))
+    edges = cg_edges.Edges.concatenate(edges)
 
     if only_internal_edges:
         supervoxels = np.concatenate(
@@ -830,31 +830,43 @@ def tabular_change_log_recent(table_id):
     # Call ChunkedGraph
     cg = app_utils.get_cg(table_id)
 
-    log_rows = cg.client.read_log_entries(start_time=start_time, end_time=end_time)
-
+    # Stream the operation-log rows instead of materializing them all at once. Only the
+    # timestamp, user, and merge/split flag are needed, so the read is restricted to those
+    # columns (the default pulls the large variable-length added/removed-edge, coordinate,
+    # and affinity arrays for every operation, which dominate row size). AddedEdge is only
+    # existence-checked (merge vs split). RootID is not used directly but is kept so the
+    # streaming reader's timestamp fallback still works on older rows that predate the
+    # OperationTimeStamp column.
+    #
+    # read_log_entries_streaming yields one (operation_id, record) at a time and frees each
+    # decoded row before reading the next, so peak memory is bounded by the compact output
+    # columns below rather than by the full set of Bigtable cell objects for the window.
+    # Rows arrive already ordered by operation id (fixed-width keys), so no sort is needed.
+    operation_ids = []
     timestamp_list = []
     user_list = []
     is_merge_list = []
-
-    operation_ids = np.sort(list(log_rows.keys()))
-    for operation_id in operation_ids:
-        operation = log_rows[operation_id]
-
-        timestamp = operation["timestamp"]
-        timestamp_list.append(timestamp)
-
-        user_id = operation[attributes.OperationLogs.UserID]
-        user_list.append(user_id)
-
-        is_merge = attributes.OperationLogs.AddedEdge in operation
-        is_merge_list.append(is_merge)
+    for operation_id, operation in cg.client.read_log_entries_streaming(
+        start_time=start_time,
+        end_time=end_time,
+        properties=[
+            attributes.OperationLogs.OperationTimeStamp,
+            attributes.OperationLogs.UserID,
+            attributes.OperationLogs.AddedEdge,
+            attributes.OperationLogs.RootID,
+        ],
+    ):
+        operation_ids.append(operation_id)
+        timestamp_list.append(operation["timestamp"])
+        user_list.append(operation[attributes.OperationLogs.UserID])
+        is_merge_list.append(attributes.OperationLogs.AddedEdge in operation)
 
     return pd.DataFrame.from_dict(
         {
-            "operation_id": operation_ids,
+            "operation_id": np.array(operation_ids, dtype=np.uint64),
             "timestamp": timestamp_list,
             "user_id": user_list,
-            "is_merge": is_merge_list,
+            "is_merge": np.array(is_merge_list, dtype=bool),
         }
     )
 
@@ -1142,7 +1154,12 @@ def handle_get_layer2_graph(table_id, node_id):
 
     cg = app_utils.get_cg(table_id)
     print("Finding edge graph...")
-    edge_graph = pathing.get_lvl2_edge_list(cg, int(node_id), bbox=bounding_box)
+    edge_graph = pathing.get_lvl2_edge_list(
+        cg,
+        int(node_id),
+        bbox=bounding_box,
+        max_num_lvl2_ids=current_app.config.get("LVL2_GRAPH_MAX_NODES"),
+    )
     print("Edge graph found len: {}".format(len(edge_graph)))
     return {"edge_graph": edge_graph}