From d2b0153549d94941943652e16907fe148953bafd Mon Sep 17 00:00:00 2001 From: jayanthchundru Date: Mon, 21 Sep 2026 17:49:28 -0700 Subject: [PATCH 1/4] fix: reach Local Controllers for cleanup and stop dropping data before telemetry reads it (CAN-391) Global Controller's cleanup broadcast dialed instance["endpoint"], the host-published port (e.g. localhost:8001), which is unreachable from inside the GC's own container. Every Cleanup RPC failed, but the completed-request batch was drained from Redis regardless, so futures and affinity keys behind those requests were never deleted and Redis grew unbounded. - global_controller.py: resolve the cleanup RPC endpoint through instance_manager._routing_endpoint_for() (container-reachable), matching every other in-container RPC caller in this file. Only drain "request:completed" once every instance has confirmed receipt; on any failure (or no instances at all), leave the batch queued for retry next cleanup cycle. The receiving RPC is already idempotent (setnx lock, no-op if futures are already gone), so retrying is safe. - local_controller_frontend.py: once cleanup RPCs actually succeed, a second issue surfaces under load -- _cleanup_request deleted future:{id} keys immediately, racing GlobalController's poll loop (every 5s), which reads those same keys to build OTel spans (telemetry_logging.pull_runtime_information). Whichever side lost the race silently dropped that request's span. Verified via a controlled before/after on the same image build: reverting just the reachability fix restored 7000/7000 spans landing, while keeping it landed only ~4900/7000. Switched future-key cleanup from redis.delete to redis.expire(30s), the same grace-period pattern deploy.py already uses for request:*:status/:result, so the poll loop has time to capture telemetry before the keys disappear. Affinity bindings are untouched since nothing else reads them. Verified with both benchmark.py runs and unit tests: cleanup succeeds with zero failures, future:*/affinity:*/request:*:futures keys no longer accumulate (confirmed expiring rather than piling up), and otel_counts_match is back to true with all 7000/7000 expected spans landing, on the same 1000-request/concurrency-5 workload that previously showed the leak and the span-loss regression. Co-Authored-By: Claude Fable 5.1 --- .../controller/global_controller.py | 23 ++++++++-- .../controller/local_controller_frontend.py | 24 ++++++++-- .../tests/test_global_controller_cleanup.py | 23 ++++++++-- .../tests/test_local_controller_cleanup.py | 46 +++++++++++++++---- 4 files changed, 95 insertions(+), 21 deletions(-) diff --git a/packages/core/canyonos_core/controller/global_controller.py b/packages/core/canyonos_core/controller/global_controller.py index 7e250636..1a88d529 100644 --- a/packages/core/canyonos_core/controller/global_controller.py +++ b/packages/core/canyonos_core/controller/global_controller.py @@ -815,7 +815,10 @@ def _trigger_cleanup(self): payload = json.dumps({"request_ids": list(all_completed)}) def _send(instance): - endpoint = instance["endpoint"] + # Container-reachable address (runtime_id:CONTAINER_PORT), not + # instance["endpoint"] -- that's the host-published port + # (e.g. localhost:8001), which the GC container can't reach. + endpoint = self.instance_manager._routing_endpoint_for(instance) try: stub = self._get_lc_stub(endpoint) stub.Cleanup(local_controler_pb2.JsonResponse(resonse=payload)) @@ -824,20 +827,34 @@ def _send(instance): len(all_completed), endpoint, ) + return True except Exception as e: logger.warning("Failed to trigger cleanup on %s: %s", endpoint, e) + return False instances = self.instance_manager.list_instances() if instances: with ThreadPoolExecutor(max_workers=len(instances)) as executor: - list(executor.map(_send, instances)) + all_sent = all(executor.map(_send, instances)) + else: + # Nothing to broadcast to -- don't drop the batch as if it were handled. + all_sent = False + + if not all_sent: + logger.warning( + "Cleanup broadcast failed for at least one instance; leaving %d " + "request(s) queued for retry on the next cycle.", + len(all_completed), + ) + return logger.info( "Triggered cleanup for %d completed request(s) across %d node(s)", len(all_completed), len(completed_by_client), ) - # Drain each node's own set from the same client it was read from. + # Drain each node's own set from the same client it was read from, + # now that every instance has confirmed receipt of the batch. for client, completed in completed_by_client.items(): client.srem("request:completed", *completed) diff --git a/packages/core/canyonos_core/controller/local_controller_frontend.py b/packages/core/canyonos_core/controller/local_controller_frontend.py index 3e78eb4e..a468fc21 100644 --- a/packages/core/canyonos_core/controller/local_controller_frontend.py +++ b/packages/core/canyonos_core/controller/local_controller_frontend.py @@ -21,6 +21,18 @@ logging.basicConfig(level=logging.INFO) logger = logging.getLogger(__name__) +# Grace period before a completed request's future keys actually disappear. +# GlobalController's poll loop (default every 5s) reads future:{id} to turn it +# into an OTel span -- see pull_runtime_information() in telemetry_logging.py. +# Deleting these keys the instant Cleanup runs races that read: whichever +# request's future gets deleted first loses its span forever, with no error +# anywhere. This used to never matter because the Cleanup RPC never actually +# reached here (CAN-391); now that it does, the race is real under load. +# Expiring instead of deleting keeps memory bounded (CAN-391's actual +# requirement) while giving the poll loop several chances to read the data +# first -- same pattern as deploy.py's COMPLETED_TTL_SECONDS. +FUTURE_CLEANUP_GRACE_SECONDS = 30 + class LocalControllerServicer(local_controler_pb2_grpc.LocalControllerServicer): """gRPC servicer that accepts requests and pushes them into a queue.""" @@ -127,18 +139,22 @@ def _cleanup_request(self, request_id): logger.info("No futures found for request %s on this node.", request_id) return - keys_to_delete = [futures_key] + keys_to_expire = [futures_key] for fid in future_ids: - keys_to_delete.extend( + keys_to_expire.extend( [ f"future:{fid}", f"future:{fid}:children", f"future:{fid}:consumers", ] ) - self.redis.delete(*keys_to_delete) + for key in keys_to_expire: + self.redis.expire(key, FUTURE_CLEANUP_GRACE_SECONDS) logger.info( - "Cleaned up %d future(s) for request %s", len(future_ids), request_id + "Scheduled %d future(s) for request %s to expire in %ds", + len(future_ids), + request_id, + FUTURE_CLEANUP_GRACE_SECONDS, ) # Clean up affinity bindings for this request diff --git a/packages/core/tests/test_global_controller_cleanup.py b/packages/core/tests/test_global_controller_cleanup.py index 597fec67..29340c9a 100644 --- a/packages/core/tests/test_global_controller_cleanup.py +++ b/packages/core/tests/test_global_controller_cleanup.py @@ -70,6 +70,13 @@ def __init__(self, instances): def list_instances(self): return self._instances + def _routing_endpoint_for(self, instance): + # Real InstanceManager resolves the container-reachable address + # (runtime_id:CONTAINER_PORT); these fixtures use "endpoint" as + # that already-resolved stand-in, since these tests are about + # batching/draining semantics, not address resolution itself. + return instance["endpoint"] + def _bare_controller(redis, instances, node_redis=None): """Build a GlobalController without running its heavy __init__. @@ -115,7 +122,12 @@ def test_sends_one_batched_call_per_instance_not_per_request(self): # Drained after broadcasting, same as before. self.assertEqual(redis.smembers("request:completed"), set()) - def test_one_instance_failing_does_not_block_others_or_stop_draining(self): + def test_one_instance_failing_leaves_the_batch_queued_for_retry(self): + # CAN-391: a batch is only ever removed from "request:completed" once + # every instance has confirmed receipt. If even one Cleanup RPC fails + # (e.g. an unreachable endpoint), the whole batch must stay queued -- + # dropping it here is how cleanup entries went missing and Redis grew + # unbounded. completed = {"reqA", "reqB"} expected = set(completed) # snapshot -- see note in the test above redis = _FakeRedis({"request:completed": completed}) @@ -128,9 +140,11 @@ def test_one_instance_failing_does_not_block_others_or_stop_draining(self): controller._trigger_cleanup() # must not raise + # The reachable instance still gets the batch... self.assertEqual(len(good_stub.calls), 1) self.assertEqual(set(good_stub.calls[0]["request_ids"]), expected) - self.assertEqual(redis.smembers("request:completed"), set()) + # ...but nothing is drained until every instance has confirmed. + self.assertEqual(redis.smembers("request:completed"), expected) def test_noop_when_nothing_completed(self): redis = _FakeRedis() @@ -144,12 +158,13 @@ def test_noop_when_nothing_completed(self): self.assertEqual(stub.calls, []) def test_noop_when_no_instances_registered(self): + # CAN-391: with nothing to broadcast to, nothing has confirmed the + # batch -- it must stay queued rather than being silently dropped. redis = _FakeRedis({"request:completed": {"req1"}}) controller = _bare_controller(redis, []) - # Should still drain the completed set even with nothing to broadcast to. controller._trigger_cleanup() - self.assertEqual(redis.smembers("request:completed"), set()) + self.assertEqual(redis.smembers("request:completed"), {"req1"}) class MultiNodeTriggerCleanupTests(unittest.TestCase): diff --git a/packages/core/tests/test_local_controller_cleanup.py b/packages/core/tests/test_local_controller_cleanup.py index 5d86b75f..e33046c1 100644 --- a/packages/core/tests/test_local_controller_cleanup.py +++ b/packages/core/tests/test_local_controller_cleanup.py @@ -10,7 +10,10 @@ 0, os.path.abspath(os.path.join(os.path.dirname(__file__), "..", "grpc_stubs")) ) -from canyonos_core.controller.local_controller_frontend import LocalControllerServicer +from canyonos_core.controller.local_controller_frontend import ( + FUTURE_CLEANUP_GRACE_SECONDS, + LocalControllerServicer, +) import local_controler_pb2 @@ -81,6 +84,7 @@ class _FakeRedisStore: def __init__(self, strings=None, sets=None): self.strings = strings or {} self.sets = sets or {} + self.expirations = {} # key -> seconds, as recorded by expire() def setnx(self, key, value): if key in self.strings: @@ -96,6 +100,12 @@ def delete(self, *keys): self.strings.pop(key, None) self.sets.pop(key, None) + def expire(self, key, seconds): + # Real Redis: schedules removal after `seconds`, doesn't touch the + # value now. This fake just records the call so tests can assert on + # it without needing to fake time passing. + self.expirations[key] = seconds + def _bare_servicer(redis): servicer = LocalControllerServicer.__new__(LocalControllerServicer) @@ -105,7 +115,12 @@ def _bare_servicer(redis): class CleanupRequestTests(unittest.TestCase): - def test_cleanup_deletes_consolidated_future_hashes_and_bookkeeping(self): + def test_cleanup_expires_consolidated_future_hashes_and_bookkeeping(self): + # CAN-391 follow-up: GlobalController's poll loop reads future:{id} to + # build an OTel span (see telemetry_logging.pull_runtime_information). + # Deleting it immediately here races that read and silently drops the + # span. Expiring with a grace period keeps memory bounded without + # deleting out from under the poll loop. redis = _FakeRedisStore( sets={"request:req1:futures": {"fut1", "fut2"}}, strings={ @@ -119,14 +134,21 @@ def test_cleanup_deletes_consolidated_future_hashes_and_bookkeeping(self): servicer._cleanup_request("req1") - # Future-resolution bookkeeping: gone. - self.assertNotIn("request:req1:futures", redis.sets) - self.assertNotIn("future:fut1", redis.strings) - self.assertNotIn("future:fut2", redis.strings) - self.assertNotIn("future:fut1:children", redis.strings) - self.assertNotIn("future:fut1:consumers", redis.strings) + # Not deleted outright -- still readable until the TTL elapses. + for key in ( + "request:req1:futures", + "future:fut1", + "future:fut2", + "future:fut1:children", + "future:fut1:consumers", + ): + self.assertEqual(redis.expirations.get(key), FUTURE_CLEANUP_GRACE_SECONDS) + self.assertIn("request:req1:futures", redis.sets) + self.assertIn("future:fut1", redis.strings) - def test_cleanup_still_deletes_affinity_bindings(self): + def test_cleanup_still_deletes_affinity_bindings_outright(self): + # Affinity bindings aren't read by the telemetry poll loop, so there's + # no race to protect against here -- immediate deletion is still fine. redis = _FakeRedisStore( sets={"request:req1:futures": {"fut1"}}, strings={"future:fut1": "x", "affinity:req1": "some-host"}, @@ -136,7 +158,11 @@ def test_cleanup_still_deletes_affinity_bindings(self): servicer._cleanup_request("req1") self.assertNotIn("affinity:req1", redis.strings) - self.assertNotIn("future:fut1", redis.strings) + # The future itself is expired (grace period), not deleted outright. + self.assertIn("future:fut1", redis.strings) + self.assertEqual( + redis.expirations.get("future:fut1"), FUTURE_CLEANUP_GRACE_SECONDS + ) def test_cleanup_releases_its_lock_even_with_no_futures(self): redis = _FakeRedisStore(sets={"request:req1:futures": set()}) From b6b5bded966d7c607399c3d8fbead9932f213e4b Mon Sep 17 00:00:00 2001 From: jayanthchundru Date: Mon, 21 Sep 2026 19:18:42 -0700 Subject: [PATCH 2/4] refactor: drop the all_sent flag in _trigger_cleanup for guard clauses No behavior change -- verified with the existing unit tests (all 257 still pass) and a repeat 1000-request/concurrency-5 benchmark run, which showed the same result as before the refactor: zero cleanup failures, no leftover future:*/affinity:*/request:*:futures keys, and otel_counts_match true with all 7000/7000 expected spans landing. Replaces computing all_sent in two branches and checking it two lines later with two early-return guard clauses (no instances -> return, not all(...) -> return), so there's no intermediate boolean carrying state between where it's computed and where it's checked. Co-Authored-By: Claude Fable 5.1 --- .../controller/global_controller.py | 23 +++++++++---------- 1 file changed, 11 insertions(+), 12 deletions(-) diff --git a/packages/core/canyonos_core/controller/global_controller.py b/packages/core/canyonos_core/controller/global_controller.py index 1a88d529..c5177f0b 100644 --- a/packages/core/canyonos_core/controller/global_controller.py +++ b/packages/core/canyonos_core/controller/global_controller.py @@ -833,28 +833,27 @@ def _send(instance): return False instances = self.instance_manager.list_instances() - if instances: - with ThreadPoolExecutor(max_workers=len(instances)) as executor: - all_sent = all(executor.map(_send, instances)) - else: - # Nothing to broadcast to -- don't drop the batch as if it were handled. - all_sent = False - - if not all_sent: + if not instances: logger.warning( - "Cleanup broadcast failed for at least one instance; leaving %d " - "request(s) queued for retry on the next cycle.", + "No instances to broadcast cleanup to; leaving %d request(s) queued.", len(all_completed), ) return + with ThreadPoolExecutor(max_workers=len(instances)) as executor: + if not all(executor.map(_send, instances)): + logger.warning( + "Cleanup broadcast failed for at least one instance; leaving %d " + "request(s) queued for retry on the next cycle.", + len(all_completed), + ) + return + logger.info( "Triggered cleanup for %d completed request(s) across %d node(s)", len(all_completed), len(completed_by_client), ) - # Drain each node's own set from the same client it was read from, - # now that every instance has confirmed receipt of the batch. for client, completed in completed_by_client.items(): client.srem("request:completed", *completed) From f50fa1153f281e3a53596407383852ea00b6e785 Mon Sep 17 00:00:00 2001 From: jayanthchundru Date: Tue, 22 Sep 2026 12:34:42 -0700 Subject: [PATCH 3/4] Derive future cleanup TTL from poll_interval and prevent TTL reset on retry FUTURE_CLEANUP_GRACE_SECONDS was hardcoded to 30s, so a configured poll_interval >= 30s (or a slow poll cycle) could let the TTL expire before GlobalController's poll loop ever read the future data, silently dropping telemetry spans. The grace period now scales with the actual CANYONOS_POLL_INTERVAL. RedisClient.expire() also gained an nx flag so retried Cleanup RPCs (e.g. against an unreachable instance) don't keep resetting the TTL on keys that already have one. Also adds RoutingEndpointTests to test_global_controller_cleanup.py, which give an instance's published and routing endpoints distinct values so a regression back to instance["endpoint"] (the CAN-391 bug) would actually fail the suite -- every existing fixture collapses the two to the same value and can't catch that. ' --- .../controller/local_controller_frontend.py | 21 +++--- .../controller/utils/redis_client.py | 4 +- .../tests/test_global_controller_cleanup.py | 68 ++++++++++++++++++- 3 files changed, 78 insertions(+), 15 deletions(-) diff --git a/packages/core/canyonos_core/controller/local_controller_frontend.py b/packages/core/canyonos_core/controller/local_controller_frontend.py index a468fc21..8089855e 100644 --- a/packages/core/canyonos_core/controller/local_controller_frontend.py +++ b/packages/core/canyonos_core/controller/local_controller_frontend.py @@ -21,17 +21,14 @@ logging.basicConfig(level=logging.INFO) logger = logging.getLogger(__name__) -# Grace period before a completed request's future keys actually disappear. -# GlobalController's poll loop (default every 5s) reads future:{id} to turn it -# into an OTel span -- see pull_runtime_information() in telemetry_logging.py. -# Deleting these keys the instant Cleanup runs races that read: whichever -# request's future gets deleted first loses its span forever, with no error -# anywhere. This used to never matter because the Cleanup RPC never actually -# reached here (CAN-391); now that it does, the race is real under load. -# Expiring instead of deleting keeps memory bounded (CAN-391's actual -# requirement) while giving the poll loop several chances to read the data -# first -- same pattern as deploy.py's COMPLETED_TTL_SECONDS. -FUTURE_CLEANUP_GRACE_SECONDS = 30 + +POLL_INTERVAL_SECONDS = float(os.environ.get("CANYONOS_POLL_INTERVAL", 5)) +FUTURE_CLEANUP_GRACE_MULTIPLIER = 3 +FUTURE_CLEANUP_GRACE_MIN_SECONDS = 30 +FUTURE_CLEANUP_GRACE_SECONDS = max( + FUTURE_CLEANUP_GRACE_MIN_SECONDS, + POLL_INTERVAL_SECONDS * FUTURE_CLEANUP_GRACE_MULTIPLIER, +) class LocalControllerServicer(local_controler_pb2_grpc.LocalControllerServicer): @@ -149,7 +146,7 @@ def _cleanup_request(self, request_id): ] ) for key in keys_to_expire: - self.redis.expire(key, FUTURE_CLEANUP_GRACE_SECONDS) + self.redis.expire(key, FUTURE_CLEANUP_GRACE_SECONDS, nx=True) logger.info( "Scheduled %d future(s) for request %s to expire in %ds", len(future_ids), diff --git a/packages/core/canyonos_core/controller/utils/redis_client.py b/packages/core/canyonos_core/controller/utils/redis_client.py index 9b52d8ee..3ba079e8 100644 --- a/packages/core/canyonos_core/controller/utils/redis_client.py +++ b/packages/core/canyonos_core/controller/utils/redis_client.py @@ -28,9 +28,9 @@ def setnx(self, key, value): """Set key to value only if it does not already exist. Returns True if set, False otherwise.""" return self.client.setnx(key, value) - def expire(self, key, seconds): + def expire(self, key, seconds, nx=False): """Set a TTL (in seconds) on a key. No-op if the key does not exist.""" - return self.client.expire(key, seconds) + return self.client.expire(key, seconds, nx=nx) # --- Hash operations --- diff --git a/packages/core/tests/test_global_controller_cleanup.py b/packages/core/tests/test_global_controller_cleanup.py index 29340c9a..377cabe4 100644 --- a/packages/core/tests/test_global_controller_cleanup.py +++ b/packages/core/tests/test_global_controller_cleanup.py @@ -75,7 +75,9 @@ def _routing_endpoint_for(self, instance): # (runtime_id:CONTAINER_PORT); these fixtures use "endpoint" as # that already-resolved stand-in, since these tests are about # batching/draining semantics, not address resolution itself. - return instance["endpoint"] + # A fixture that needs "endpoint" and the routing address to differ + # (see RoutingEndpointTests) sets "routing_endpoint" explicitly. + return instance.get("routing_endpoint", instance["endpoint"]) def _bare_controller(redis, instances, node_redis=None): @@ -267,6 +269,70 @@ def test_falls_back_to_self_redis_when_node_redis_is_empty_dict(self): self.assertEqual(redis.smembers("request:completed"), set()) +class RoutingEndpointTests(unittest.TestCase): + """CAN-391: the GC container must send Cleanup to each instance's + container-reachable routing endpoint, never its host-published endpoint + (e.g. localhost:8001) -- the GC can't reach that from inside Docker. + Every other test in this file gives an instance the same value for both, + so a regression back to instance["endpoint"] would pass them unnoticed.""" + + def test_cleanup_uses_routing_endpoint_not_published_endpoint(self): + completed = {"req1"} + redis = _FakeRedis({"request:completed": set(completed)}) + instances = [ + {"endpoint": "localhost:8001", "routing_endpoint": "runtime-abc:50051"} + ] + controller = _bare_controller(redis, instances) + + routing_stub = _FakeStub() + + def _get_lc_stub(endpoint): + if endpoint == "localhost:8001": + raise AssertionError( + "Cleanup must not be sent to the host-published endpoint; " + "the GC container cannot reach it." + ) + return routing_stub + + controller._get_lc_stub = _get_lc_stub + + controller._trigger_cleanup() + + self.assertEqual(len(routing_stub.calls), 1) + self.assertEqual(set(routing_stub.calls[0]["request_ids"]), completed) + self.assertEqual(redis.smembers("request:completed"), set()) + + def test_distinct_endpoints_across_multiple_instances(self): + completed = {"req1", "req2"} + redis = _FakeRedis({"request:completed": set(completed)}) + instances = [ + {"endpoint": f"localhost:800{i}", "routing_endpoint": f"runtime-{i}:50051"} + for i in range(3) + ] + controller = _bare_controller(redis, instances) + + stubs = {inst["routing_endpoint"]: _FakeStub() for inst in instances} + published = {inst["endpoint"] for inst in instances} + + def _get_lc_stub(endpoint): + self.assertNotIn( + endpoint, published, "must route by routing_endpoint, not endpoint" + ) + return stubs[endpoint] + + controller._get_lc_stub = _get_lc_stub + + controller._trigger_cleanup() + + for routing_endpoint, stub in stubs.items(): + self.assertEqual( + len(stub.calls), + 1, + f"expected exactly one Cleanup call to {routing_endpoint}", + ) + self.assertEqual(set(stub.calls[0]["request_ids"]), completed) + + class StaleContainerNameTests(unittest.TestCase): """Bug 14: the cleanup used to build "canyonos--", which never matched the name the Local runtime actually creates, so no stale agent From 577320427e02c311e39ca415e52a57d7091fef84 Mon Sep 17 00:00:00 2001 From: Jayanth Krishna Chundru Date: Tue, 22 Sep 2026 13:01:57 -0700 Subject: [PATCH 4/4] Add 'nx' parameter to expire method Updated expire method to include 'nx' parameter for conditional expiration. Co-authored-by: Copilot Autofix powered by AI <175728472+Copilot@users.noreply.github.com> --- packages/core/tests/test_local_controller_cleanup.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/packages/core/tests/test_local_controller_cleanup.py b/packages/core/tests/test_local_controller_cleanup.py index ae536451..dddc20a8 100644 --- a/packages/core/tests/test_local_controller_cleanup.py +++ b/packages/core/tests/test_local_controller_cleanup.py @@ -127,7 +127,7 @@ def delete(self, *keys): self.strings.pop(key, None) self.sets.pop(key, None) - def expire(self, key, seconds): + def expire(self, key, seconds, nx=False): # Real Redis: schedules removal after `seconds`, doesn't touch the # value now. This fake just records the call so tests can assert on # it without needing to fake time passing.