[Bug] Future result not properly distrubuted to all consumers.

ventis/controller/local_controller.py

@@ -61,6 +61,9 @@ def __init__(self, port=50051):
 
         self.server, self.servicer = start_server(port, my_endpoint=self._my_endpoint)
         self.request_queue = self.servicer.request_queue
+        # Let the gRPC servicer fan a result out to this node's consumers as
+        # soon as it arrives via WriteResult (see _fan_out_to_consumers).
+        self.servicer.on_result = self._fan_out_to_consumers
 
         # Connect to Redis and report healthy status
         redis_host = os.environ.get("VENTIS_REDIS_HOST", "localhost")
@@ -314,6 +317,10 @@ def _mark_future_failed(self, future_id, error, origin=None):
             {"failed": 1, "error_message": error_message},
         )
 
+        # Unblock any consumers waiting on this future so a failed dependency
+        # surfaces as an error instead of a 300s timeout.
+        self._fan_out_to_consumers(future_id, failed=1, error_message=error_message)
+
         if origin and origin != self._my_endpoint:
             self._send_result_callback(
                 origin,
@@ -427,7 +434,9 @@ def _process_request(self, data):
                         )
 
                         # If the result is already available, push it immediately.
-                        # This handles the race where _notify_consumers already ran.
+                        # This handles the race where the producer resolved the
+                        # future before this consumer registered (and thus before
+                        # _fan_out_to_consumers could see it).
                         existing_result = self.redis.hget(future_key, "result")
                         if existing_result is not None and existing_result != "":
                             logger.info(
@@ -586,6 +595,9 @@ def _execute_locally(
             # Write result to local Redis
             self.redis.hset(f"future:{future_id}", "result", serialized)
 
+            # Push the result to any consumers registered on this node.
+            self._fan_out_to_consumers(future_id, result=serialized)
+
             # If the request came from another node, send result back to origin
             if origin and origin != self._my_endpoint:
                 self._send_result_callback(
@@ -700,6 +712,33 @@ def _send_result_callback(
             logger.error("Failed to send result callback to %s: %s", origin, e)
             self._mark_future_failed(future_id, f"Result callback failed: {e}")
 
+    def _fan_out_to_consumers(self, future_id, result=None, failed=0, error_message=""):
+        """Push a completed future (result or failure) to every endpoint registered
+        as a consumer on THIS node's Redis.
+
+        Called at every site that writes a terminal value for a future -- local
+        production (_execute_locally), failure (_mark_future_failed), and remote
+        arrival (WriteResult) -- so propagation is event-driven and never depends
+        on anyone calling Future.value(). Whichever node holds the consumer set is
+        always either the producer or the origin, so one of those sites fires; the
+        value then re-fans-out at each node it lands on, walking the graph.
+
+        Delivery is intentionally not deduped: a consumer's WriteResult just
+        re-writes the same value, so a redundant push (e.g. racing the immediate
+        push in _process_request) is idempotent and harmless.
+        """
+        if not future_id:
+            return
+        for endpoint in self.redis.smembers(f"future:{future_id}:consumers"):
+            if endpoint and endpoint != self._my_endpoint:
+                self._send_result_callback(
+                    endpoint,
+                    future_id,
+                    result=result,
+                    failed=failed,
+                    error_message=error_message,
+                )
+
     # ------------------------------------------------------------------ #
     #  Shutdown                                                            #
     # ------------------------------------------------------------------ #

ventis/controller/local_controller_frontend.py

@@ -36,6 +36,9 @@ def __init__(self, my_endpoint="unknown"):
         except ImportError:
             from redis_client import RedisClient
         self.redis = RedisClient(host=redis_host, port=redis_port)
+        # Set by LocalController: fans a just-arrived result out to this node's
+        # consumers. Signature: on_result(future_id, result, failed, error_message).
+        self.on_result = None
 
     def Execute(self, request, context):
         """Accept an Execute request and push it into the queue."""
@@ -78,6 +81,16 @@ def WriteResult(self, request, context):
                         future_id,
                         result,
                     )
+                # Relay the just-arrived value to any consumers registered on
+                # this node (the origin is where consumer sets live). This is
+                # what walks the value hop-by-hop through the graph.
+                if self.on_result:
+                    self.on_result(
+                        future_id,
+                        result=result,
+                        failed=failed,
+                        error_message=error_message,
+                    )
             else:
                 logger.error("WriteResult: missing future_id in %s", data)
         except Exception as e:

ventis/future.py

@@ -176,9 +176,9 @@ def value(self, timeout=None):
                 )
         self.calculated = True
 
-        # Push result to all consumers
-        self._notify_consumers()
-
+        # Note: consumer fan-out is handled entirely by the local controllers at
+        # result-write time (see LocalController._fan_out_to_consumers). value()
+        # is now purely a top-level pull for whoever holds the root future.
         return self.result
 
     # def __call__(self, timeout=None):
@@ -191,39 +191,6 @@ def is_available(self):
         """
         return self.result is not None
 
-    def _get_consumers(self):
-        """Return the list of consumers from Redis."""
-        return self.redis.smembers(self._consumers_key())
-
-    def _notify_consumers(self):
-        """Push this future's result to all registered consumer endpoints via gRPC WriteResult."""
-        consumers = self._get_consumers()
-        if not consumers:
-            return
-        for endpoint in consumers:
-            try:
-                if not self.result:
-                    logger.warning(
-                        "Future %s is notifying consumer %s with an empty/None result",
-                        self.id,
-                        endpoint,
-                    )
-                channel = grpc.insecure_channel(endpoint)
-                stub = local_controler_pb2_grpc.LocalControllerStub(channel)
-                payload = json.dumps({"future_id": self.id, "result": self.result})
-                request = local_controler_pb2.JsonResponse(resonse=payload)
-                stub.WriteResult(request)
-                logger.info(
-                    "Notified consumer %s with result for future %s", endpoint, self.id
-                )
-            except Exception as e:
-                logger.error(
-                    "Failed to notify consumer %s for future %s: %s",
-                    endpoint,
-                    self.id,
-                    e,
-                )
-
     def _add_consumer(self, consumer):
         """Add a consumer."""
         self.consumers.append(consumer)