improving skeleton service by upgrading python, messaging client, and tests

.github/workflows/python-package.yml

@@ -16,12 +16,14 @@ jobs:
     strategy:
       fail-fast: false
       matrix:
-        python-version: ["3.9"]
+        # Matches the Dockerfile base image (tiangolo/uwsgi-nginx-flask:python3.11) and release.yml.
+        # messagingclient >= 0.4.0 declares requires-python >= 3.11, so 3.9 cannot install requirements.txt.
+        python-version: ["3.11"]
 
     steps:
-    - uses: actions/checkout@v4
+    - uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7.0.1
     - name: Set up Python ${{ matrix.python-version }}
-      uses: actions/setup-python@v3
+      uses: actions/setup-python@5fda3b95a4ea91299a34e894583c3862153e4b97 # v7.0.0
       with:
         python-version: ${{ matrix.python-version }}
     - name: Install dependencies

.github/workflows/release.yml

@@ -35,15 +35,15 @@ jobs:
       MINOR_VERSION: ${{ steps.get-version.outputs.MINOR_VERSION }}
     steps:
       - name: Checkout
-        uses: actions/checkout@v4
+        uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7.0.1
       - name: Get tags
         run: git fetch --tags origin
       - name: Configure git for github-actions[bot]
         run: |
           git config --global user.name "github-actions[bot]"
           git config --global user.email "github-actions[bot]@users.noreply.github.com"
       - name: Install Python
-        uses: actions/setup-python@v5
+        uses: actions/setup-python@5fda3b95a4ea91299a34e894583c3862153e4b97 # v7.0.0
         with:
           python-version: "3.11"
       - name: Install bumpversion
@@ -73,7 +73,7 @@ jobs:
 
       - name: Create GitHub Release
         if: ${{ github.event.inputs.dry-run == 'false' }}
-        uses: softprops/action-gh-release@v2
+        uses: softprops/action-gh-release@3d0d9888cb7fd7b750713d6e236d1fcb99157228 # v3.0.2
         with:
           tag_name: v${{ steps.get-version.outputs.VERSION }}
           name: Release v${{ steps.get-version.outputs.VERSION }}
@@ -99,10 +99,10 @@ jobs:
       id-token: write
     steps:
       - name: Checkout SkeletonService repo
-        uses: actions/checkout@v4
+        uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7.0.1
 
       - name: Checkout cave-helm-charts repo
-        uses: actions/checkout@v4
+        uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7.0.1
         with:
           repository: CAVEconnectome/cave-helm-charts
           token: ${{ secrets.HELM_CHART_UPDATE_TOKEN }}

Dockerfile

@@ -1,4 +1,4 @@
-FROM tiangolo/uwsgi-nginx-flask:python3.9
+FROM tiangolo/uwsgi-nginx-flask:python3.11
 
 COPY requirements.txt /app/.
 RUN python -m pip install --upgrade pip

requirements.in

@@ -9,7 +9,12 @@ middle-auth-client>=3.16.0
 caveclient>=8.2.1
 meshparty==1.18.3
 pcg_skel>=1.3.1
-messagingclient
+# >=0.6.1 for RetryableError, which skeletonservice/datasets/messaging.py raises to return a
+# transient failure to the subscription instead of acking and dropping it, and for the fix that
+# makes that path survive: with max_messages=1 a nacked message left ack_ids empty and
+# acknowledge() then raised "400 You have not specified an ack ID", ending the worker's consume
+# loop on every retryable failure.
+messagingclient>=0.6.1
 cloud-files>=5.0.3
 cloud-volume>=11.2.0
 Flask-Limiter[redis]

requirements.txt

@@ -1,11 +1,9 @@
 # This file was autogenerated by uv via the following command:
-#    uv pip compile requirements.in -o requirements.txt
+#    uv pip compile requirements.in -o /app/requirements.txt
 aniso8601==9.0.1
     # via flask-restx
 asttokens==2.4.1
     # via stack-data
-async-timeout==5.0.1
-    # via redis
 attrs==23.2.0
     # via
     #   caveclient
@@ -17,11 +15,11 @@ blosc==1.11.1
     # via meshparty
 blosc2==2.5.1
     # via tables
-boto3==1.7.84
+boto3==1.43.72
     # via
     #   cloud-files
     #   cloud-volume
-botocore==1.10.84
+botocore==1.43.72
     # via
     #   boto3
     #   s3transfer
@@ -83,14 +81,8 @@ dill==0.3.8
     # via
     #   multiprocess
     #   pathos
-docutils==0.21.2
-    # via botocore
 dracopy==1.3.0
     # via cloud-volume
-exceptiongroup==1.2.0
-    # via
-    #   ipython
-    #   pytest
 executing==2.0.1
     # via stack-data
 fasteners==0.19
@@ -198,14 +190,9 @@ idna==3.6
 imageio==2.34.1
     # via meshparty
 importlib-metadata==7.0.1
-    # via
-    #   flask
-    #   markdown
-    #   opentelemetry-api
+    # via opentelemetry-api
 importlib-resources==6.1.1
-    # via
-    #   flask-restx
-    #   matplotlib
+    # via flask-restx
 inflection==0.5.1
     # via python-jsonschema-objects
 iniconfig==2.0.0
@@ -260,7 +247,7 @@ meshparty==1.18.3
     # via
     #   -r requirements.in
     #   pcg-skel
-messagingclient==0.3.0
+messagingclient==0.6.1
     # via -r requirements.in
 microviewer==1.11.0
     # via cloud-volume
@@ -462,7 +449,7 @@ rsa==4.9
     #   google-auth
 rtree==1.2.0
     # via meshparty
-s3transfer==0.1.13
+s3transfer==0.19.2
     # via boto3
 scikit-learn==1.4.2
     # via meshparty
@@ -502,8 +489,6 @@ tenacity==8.2.3
     #   cloud-volume
 threadpoolctl==3.5.0
     # via scikit-learn
-tomli==2.0.1
-    # via pytest
 tqdm==4.66.2
     # via
     #   cloud-files
@@ -519,13 +504,12 @@ trimesh==4.3.1
 typing-extensions==4.9.0
     # via
     #   flask-limiter
-    #   ipython
     #   limits
-    #   rich
 tzdata==2024.1
     # via pandas
 urllib3==2.2.1
     # via
+    #   botocore
     #   caveclient
     #   cloud-files
     #   cloud-volume
@@ -545,9 +529,7 @@ werkzeug==3.1.3
 wrapt==1.17.2
     # via deprecated
 zipp==3.17.0
-    # via
-    #   importlib-metadata
-    #   importlib-resources
+    # via importlib-metadata
 zope-event==5.0
     # via gevent
 zope-interface==6.3

skeletonservice/datasets/messaging.py

@@ -5,7 +5,7 @@
 from timeit import default_timer
 from messagingclient import MessagingClientConsumer
 from messagingclient import RetryableError
-from .service import SkeletonService
+from .service import SkeletonService, _PhaseTimer
 
 # messagingclient logs one line per received message with the bare `logging` module, i.e. on the
 # ROOT logger, not on a 'messagingclient' logger (see messagingclient/client.py, _consume_round_robin).
@@ -36,6 +36,27 @@
 }
 
 
+try:
+    from pcg_skel.chunk_tools import CompleteDataException
+except ImportError:  # pcg_skel restructured; fall back to matching by name below.
+    CompleteDataException = None
+
+
+def _is_incomplete_l2_cache(exc):
+    """True when the skeleton failed only because the l2 cache is not yet populated.
+
+    pcg_skel raises CompleteDataException from dense_spatial_lookup when any level 2 id in the
+    root lacks rep_coord_nm (require_complete=True). Requesting those ids is itself what makes
+    pcgl2cache enqueue them for computation, so the work becomes possible a short time later --
+    this is a wait, not a failure, and the message should come back rather than be dropped.
+
+    Observed on minniev7: one missing id out of 13,065 fails the whole skeleton.
+    """
+    if CompleteDataException is not None and isinstance(exc, CompleteDataException):
+        return True
+    return type(exc).__name__ == "CompleteDataException"
+
+
 def _retryable_status(exc):
     """Return the HTTP status this exception should be retried on, else None.
 
@@ -124,6 +145,16 @@ def callback(payload):
                         f"SkeletonService.get_skeleton_by_datastack_and_rid(); returning the message "
                         f"for redelivery.", session_timestamp_=session_timestamp)
                     raise RetryableError(f"HTTP {status}") from e
+                if _is_incomplete_l2_cache(e):
+                    # The read that just failed is also what queues the missing ids for
+                    # computation, so redelivery is the retry. No traceback: this is an expected
+                    # wait, and it was ~30% of messages while the trigger was broken.
+                    SkeletonService.print_with_session_timestamp(
+                        "Skeleton Cache message-processor hit an incomplete l2 cache for rid "
+                        f"{payload.attributes['skeleton_params_rid']}; returning the message for "
+                        "redelivery so pcgl2cache has time to compute the missing ids.",
+                        session_timestamp_=session_timestamp)
+                    raise RetryableError("incomplete l2 cache") from e
                 SkeletonService.print_with_session_timestamp("Skeleton Cache message-processor received error from SkeletonService.get_skeleton_by_datastack_and_rid(): ", repr(e), session_timestamp_=session_timestamp)
                 SkeletonService.print_with_session_timestamp(tb.format_exc(), session_timestamp_=session_timestamp)
                 raise e
@@ -138,6 +169,9 @@ def callback(payload):
                     payload.attributes["skeleton_params_datastack_name"],
                     int(payload.attributes["skeleton_params_rid"]),
                     int(payload.attributes["verbose_level"]),
+                    # Carried on the message by retry_refusal_list. Absent for ordinary requests,
+                    # so a first-time refusal records 0 and stays eligible for one retry.
+                    retry_count=int(payload.attributes.get("refusal_retry_count", 0)),
                 )
                 if verbose_level >= 1:
                     SkeletonService.print_with_session_timestamp("Skeleton Cache message-processor returned from SkeletonService.add_rid_to_refusal_list() with result: ", result, session_timestamp_=session_timestamp)
@@ -158,6 +192,13 @@ def callback(payload):
         print("Skeleton Cache messaging message-processor suffered a failure that was not caught at lower granularity: ", repr(e))
         tb.print_exc()
     finally:
+        # Only the early exits (refused / cache_hit / invalid) emitted their own PHASE_TIMINGS, so
+        # generations -- the 150-800s messages we actually need to explain -- logged nothing. This
+        # emits whatever the timer accumulated, on every path including failures.
+        try:
+            _PhaseTimer.emit_current(message_outcome)
+        except Exception:
+            pass  # instrumentation must never affect message handling
         if log_phase_timings:
             try:
                 print("MESSAGE_TIMING " + json.dumps({
@@ -167,17 +208,66 @@ def callback(payload):
             except Exception:
                 pass  # instrumentation must never affect message handling
 
+# Which of the three subscriptions this process consumes: a comma separated selection of the keys
+# low, high and dead. Mandatory -- there is no default, because silently falling back to all three
+# would turn a chart that forgot to set it into a fleet quietly competing with the others.
+#
+# These are selector KEYS, not subscription names. The names come from the SKELETON_CACHE_*_
+# RETRIEVE_QUEUE variables and are used verbatim (they are conventionally upper case, e.g.
+# minniev7_SKELETON_CACHE_WORKER_LOW_PRIORITY); only the keys here are matched case-insensitively.
+#
+# One key per deployment gives a fleet per queue. That matters because the round-robin loop polls
+# every configured subscription in turn and a blocking pull against an empty one costs the whole
+# wait: measured on minniev7 2026-08-17, 50 workers sat at 23.5% occupancy with ~40s idle per
+# message while thousands of messages waited. Separate fleets also scale on their own backlog, so
+# the high-priority and dead-letter fleets can sit at zero while their queues are empty.
+#
+# All three queue-name variables must stay set whatever is consumed here: callback() distinguishes a
+# dead-letter message from a skeleton request by matching the dead-letter queue name against the
+# subscription the message arrived on.
+CONSUME_QUEUES_ENV = "SKELETON_CACHE_CONSUME_QUEUES"
+
 try:
     c = MessagingClientConsumer()
     skeletoncache_low_priority_queue = os.getenv("SKELETON_CACHE_LOW_PRIORITY_RETRIEVE_QUEUE", None)
     skeletoncache_high_priority_queue = os.getenv("SKELETON_CACHE_HIGH_PRIORITY_RETRIEVE_QUEUE", None)
     skeletoncache_dead_letter_queue = os.getenv("SKELETON_CACHE_DEAD_LETTER_RETRIEVE_QUEUE", None)
     if not skeletoncache_low_priority_queue or not skeletoncache_high_priority_queue or not skeletoncache_dead_letter_queue:
         raise ValueError(f"Skeleton Cache messaging client: one or more of the messaging queues are not set: LOW:{skeletoncache_low_priority_queue}, HIGH:{skeletoncache_high_priority_queue}, DEAD:{skeletoncache_dead_letter_queue}")
-    c.consume_multiple([skeletoncache_low_priority_queue,
-                        skeletoncache_high_priority_queue,
-                        skeletoncache_dead_letter_queue],
-                        callback)
+
+    available = {
+        "low": skeletoncache_low_priority_queue,
+        "high": skeletoncache_high_priority_queue,
+        "dead": skeletoncache_dead_letter_queue,
+    }
+    raw = os.environ.get(CONSUME_QUEUES_ENV)
+    if raw is None or not raw.strip():
+        raise ValueError(
+            f"{CONSUME_QUEUES_ENV} is required and must be a comma separated selection of "
+            f"{sorted(available)} (it is not defaulted, so that a fleet cannot silently consume "
+            f"queues it was not meant to)")
+    keys = [k.strip().lower() for k in raw.split(",") if k.strip()]
+    unknown = [k for k in keys if k not in available]
+    if unknown:
+        raise ValueError(
+            f"{CONSUME_QUEUES_ENV}={raw!r} has unknown entries {unknown}; "
+            f"valid keys are {sorted(available)}")
+    if not keys:
+        raise ValueError(f"{CONSUME_QUEUES_ENV}={raw!r} resolved to no queues")
+    # dict.fromkeys: drop duplicates but keep the configured order, which is the round-robin order.
+    keys = list(dict.fromkeys(keys))
+    queues = [available[k] for k in keys]
+
+    print(f"Skeleton Cache messaging client consuming {keys} -> {queues}")
+
+    # consume_bounded rather than consume_multiple: the latter routes a single-queue list to
+    # consume(), a streaming-pull path whose callback wrapper acks unconditionally and knows nothing
+    # about RetryableError. A one-queue fleet would silently lose nack-based redelivery, the
+    # 60s..600s backoff and the retryable/fatal distinction, waiting out the 600s ack deadline
+    # instead. consume_bounded shares the round-robin engine and handles a list of length one, so
+    # the semantics are identical however many queues are configured. Unbounded here (no
+    # message_limit, no idle_timeout) matches the previous always-on behaviour.
+    c.consume_bounded(queues, callback)
     print("Skeleton Cache messaging client registered callback successfully (barring any exceptions that are trapped inside MessagingClientConsumer).")
 except Exception as e:
     print("Skeleton Cache messaging client failed to register callback: ", repr(e))