Fix: redis cleanup RPCs failing against unreachable ports (CAN-391)

packages/core/canyonos_core/controller/global_controller.py

@@ -1092,7 +1092,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))
@@ -1101,20 +1104,33 @@ 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))
+        if not instances:
+            logger.warning(
+                "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.
         for client, completed in completed_by_client.items():
             client.srem("request:completed", *completed)
 

packages/core/canyonos_core/controller/local_controller_frontend.py

@@ -22,6 +22,15 @@
 logger = logging.getLogger(__name__)
 
 
+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):
     """gRPC servicer that accepts requests and pushes them into a queue."""
 
@@ -132,35 +141,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:
-                future_key = f"future:{fid}"
-                # Delete sibling collection keys (e.g. future:{fid}:consumers)
-                # but handle the main hash separately to preserve logs.
-                keys_to_delete.extend(self.redis.scan_keys(f"{future_key}:*"))
-
-                logs = self.redis.hget(future_key, "logs")
-                if logs:
-                    # Replace the hash with a minimal snapshot so the global
-                    # controller's next poll can persist logs to SQLite before
-                    # they vanish. The TTL guarantees cleanup even if the poll
-                    # never reads it (e.g. controller restarts).
-                    self.redis.delete(future_key)
-                    self.redis.hset_multiple(
-                        future_key,
-                        {
-                            "id": fid,
-                            "request_id": request_id,
-                            "logs": logs,
-                        },
-                    )
-                    self.redis.expire(future_key, 30)
-                else:
-                    keys_to_delete.append(future_key)
-
-            self.redis.delete(*keys_to_delete)
+                keys_to_expire.extend(
+                    [
+                        f"future:{fid}",
+                        f"future:{fid}:children",
+                        f"future:{fid}:consumers",
+                    ]
+                )
+            for key in keys_to_expire:
+                self.redis.expire(key, FUTURE_CLEANUP_GRACE_SECONDS, nx=True)
             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

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 ---
 

packages/core/tests/test_global_controller_cleanup.py

@@ -70,6 +70,15 @@ 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.
+        # 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):
     """Build a GlobalController without running its heavy __init__.
@@ -115,7 +124,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 +142,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 +160,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):
@@ -252,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-<agent>-<i>", which never
     matched the name the Local runtime actually creates, so no stale agent

packages/core/tests/test_local_controller_cleanup.py

@@ -11,7 +11,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
 
 
@@ -102,6 +105,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:
@@ -123,6 +127,12 @@ def delete(self, *keys):
             self.strings.pop(key, None)
             self.sets.pop(key, None)
 
+    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.
+        self.expirations[key] = seconds
+
 
 def _bare_servicer(redis):
     servicer = LocalControllerServicer.__new__(LocalControllerServicer)
@@ -132,7 +142,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={
@@ -146,14 +161,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"},
@@ -163,7 +185,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()})