A rate limit now thins the work, not only the messages

The limit was applied in `apply_outputs`, which the executor reaches after the
item is off the queue — so a subscriber told to publish every 15s still cost a
queue entry, a `cascade_started`, a run record and a walk of everything
reachable from it per inbound message. Seven relay nodes behind one inverter
ran 192 times a minute to publish six.

Two halves, matching the two shapes it takes:

`trigger()` now keeps a value whose every port is inside its window and
journals nothing at all. The window split came out of `_throttled` as a
read-only `_window_split`, so the question is asked the same way in both
places and the exact split is still made once, at claim time.

A cascade carries the names it actually published, and the wave runs only the
nodes something in that set feeds. A node whose triggering inputs were all
held back is completed without running, which frees its own consumers to be
judged the same way — the case where a node re-published 619 messages a minute
off inputs that changed six times. Redeliveries and emissions carry no such
set and still walk everything, since one has a half-finished wave to finish
and the other is the value already being in state.

Skipping a node can make one ready that the scheduling pass has already walked
past, so `submit_ready` runs to a fixpoint. That also closes the same latent
hole on the replay path, where a done-marker skip could strand a join with no
future outstanding to come back for it.

Measured with the new `scripts/bench_engine.py`, 500 messages through the
house's shape: a limited source went from 500 cascades / 3500 node runs /
5009 events to 1 / 7 / 19, publishing the same 8 values; an unlimited source
into limited relays took the node reading them from 500 runs to 1.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01BpfSinyCBfjuieikyfMPbf

backend/fluksio/flow/executor.py

@@ -387,10 +387,17 @@ class ExecutionService:
                 }
             )
 
+        replay = item.deliveries > 1
         try:
-            pipeline.apply_outputs(node, item.outputs or None)
+            published = pipeline.apply_outputs(node, item.outputs or None)
             pipeline.run_downstream(
-                node, entry_id=item.entry_id, replay=item.deliveries > 1
+                node,
+                entry_id=item.entry_id,
+                replay=replay,
+                # A redelivery has to finish a walk that may be half done, and
+                # an item with no payload is the value already being in state.
+                # Neither can say what changed, so neither filters on it.
+                changed=None if replay or not item.outputs else published,
             )
         finally:
             # Paired, or a cascade that raised — state backend gone, say — is a

backend/fluksio/flow/pipeline.py

@@ -569,13 +569,14 @@ class Pipeline:
         """When a rate-limit window of this node is due to be let through."""
         return f"__flush__:{node_name}"
 
-    def _throttled(self, node: Node, result: dict[str, Any]) -> dict[str, Any]:
+    def _window_split(
-        """Hold back the outputs whose port is not due to publish yet.
+        self, node: Node, result: dict[str, Any], now: float
+    ) -> tuple[dict[str, Any], dict[str, Any], float, float | None]:
+        """What may publish now, what its window holds back, when it ends, and
+        whether a flush of this node is already booked.
 
-        The value is not lost: it is kept and published when the window ends,
+        Reads state and writes nothing, so the same question can be asked when
-        so a producer that goes quiet still delivers its last reading rather
+        a value arrives and again when the cascade carrying it is claimed.
-        than leaving the consumer on the one before it. Nothing declaring an
-        interval means nothing to look up.
         """
         limited = {
             name: spec.interval
@@ -583,9 +584,8 @@ class Pipeline:
             if spec.interval > 0 and name in result
         }
         if not limited:
-            return result
+            return result, {}, 0.0, None
 
-        now = time.time()
         flush_key = self._flush_key(node.id)
         stamps = self._state.get_multi(
             [self._timestamp_key(name) for name in limited] + [flush_key]
@@ -604,19 +604,32 @@ class Pipeline:
             else:
                 held[name] = value
                 due_at = min(due_at or window_ends, window_ends)
+        return passed, held, due_at, stamps.get(flush_key)
+
+    def _throttled(self, node: Node, result: dict[str, Any]) -> dict[str, Any]:
+        """Hold back the outputs whose port is not due to publish yet.
+
+        The value is not lost: it is kept and published when the window ends,
+        so a producer that goes quiet still delivers its last reading rather
+        than leaving the consumer on the one before it. Nothing declaring an
+        interval means nothing to look up.
+        """
+        now = time.time()
+        passed, held, due_at, pending = self._window_split(node, result, now)
 
         if self._queue is not None:
             # Without a queue there is no timer to let the value out later, so
             # holding it would only mean losing it more slowly.
-            for name in limited:
+            for name in passed:
-                if name in passed:
+                spec = node.provides.get(name)
+                if spec is not None and spec.interval > 0:
                     # A fresh publish makes anything held for that port stale.
                     self._state.delete(self._held_key(name))
             if held:
                 self._state.update(
                     {self._held_key(name): value for name, value in held.items()}
                 )
-                self._schedule_flush(node, due_at, now, stamps.get(flush_key))
+                self._schedule_flush(node, due_at, now, pending)
         return passed
 
     def _schedule_flush(
@@ -652,10 +665,14 @@ class Pipeline:
         }
         if held:
             # Publishing runs the limit again, which is what clears the hold —
-            # or puts the timer back, if this fired a hair early.
+            # or puts the timer back, if this fired a hair early. In that case
-            self.apply_outputs(node, held)
+            # nothing came out and there is nothing downstream to wake.
-            self.run_downstream(node)
+            published = self.apply_outputs(node, held)
+            if published:
+                self.run_downstream(node, changed=published)
         if any(spec.interval > 0 for spec in node.requires.values()):
+            # A wake-up that was held back, not a value that just changed —
+            # so this one deliberately walks everything reachable.
             self._execute_parallel(
                 {node, *self._get_downstream(node)}, self._state, check_ready=True
             )
@@ -1134,8 +1151,15 @@ class Pipeline:
         check_ready: bool = False,
         entry_id: str = "",
         replay: bool = False,
+        changed: set[str] | None = None,
     ) -> StateBackend:
-        """Execute nodes concurrently, scheduling each as its inputs arrive."""
+        """Execute nodes concurrently, scheduling each as its inputs arrive.
+
+        ``changed`` restricts the wave to nodes something published to: one
+        whose triggering inputs are all untouched is completed without being
+        run, so what is downstream of *it* is judged on the same footing. None
+        runs everything the subset holds, which is what a manual run means.
+        """
         # One view of the graph for the whole wave: a flow replaced halfway
         # through must not have this wave asking the new dependencies about a
         # node the old edges knew.
@@ -1154,42 +1178,64 @@ class Pipeline:
             n: sum(1 for dep in deps[n] if dep in target_nodes) for n in target_nodes
         }
 
+        # Copied: the caller's set is not this wave's to grow.
+        fresh = set(changed) if changed is not None else None
+
         submitted: set[Node] = set()
         skipped: set[Node] = set()
         node_futures: dict[Node, Future[dict[str, Any] | None]] = {}
 
-        def is_ready(n: Node) -> bool:
+        def untouched(n: Node) -> bool:
-            if in_degree[n] != 0:
+            """Did this wave publish nothing this node waits on?"""
-                return False
+            assert fresh is not None
-            if check_ready:
+            return not any(
-                return self._is_node_ready(n, state)
+                spec.trigger and name in fresh for name, spec in n.requires.items()
-            return True
+            )
+
+        def complete(n: Node) -> None:
+            """Count a node as done without running it, freeing its consumers."""
+            skipped.add(n)
+            for consumer in edges[n]:
+                if consumer in target_nodes:
+                    in_degree[consumer] -= 1
 
         def submit_ready(executor: ThreadPoolExecutor) -> None:
-            for n in target_nodes:
+            # To a fixpoint: completing a node without running it frees its
-                if n in submitted or n in skipped:
+            # consumers, and one this pass has already walked past would
-                    continue
+            # otherwise be stranded — there is no future outstanding to bring
-                # Gate before the readiness check, so a held-back node does not
+            # the drain loop back round for it.
-                # spend the synchronous claim it would need once it may run.
+            progressed = True
-                if self._gate_blocks(n):
+            while progressed:
-                    continue
+                progressed = False
-                if is_ready(n):
+                for n in target_nodes:
-                    submitted.add(n)
+                    if n in submitted or n in skipped:
+                        continue
+                    # Gate before the readiness check, so a held-back node does
+                    # not spend the synchronous claim it needs once it may run.
+                    if self._gate_blocks(n) or in_degree[n] != 0:
+                        continue
+                    if fresh is not None and untouched(n):
+                        # Ahead of the readiness check, so an unchanged node
+                        # spends neither a delivery stamp nor a synchronous
+                        # claim on a run it is not going to make.
+                        complete(n)
+                        progressed = True
+                        continue
+                    if check_ready and not self._is_node_ready(n, state):
+                        if n.synchronous:
+                            # Not ready now; a later trigger may make it ready.
+                            skipped.add(n)
+                        continue
                     if replay and entry_id and self._already_done(entry_id, n):
                         # Its side effect happened on an earlier delivery; its
                         # outputs are still in state, so downstream carries on.
-                        skipped.add(n)
+                        complete(n)
-                        submitted.discard(n)
+                        progressed = True
-                        for consumer in edges[n]:
-                            if consumer in target_nodes:
-                                in_degree[consumer] -= 1
                         continue
+                    submitted.add(n)
                     node_futures[n] = executor.submit(
                         self._execute_node, n, state, entry_id
                     )
-                elif n.synchronous and in_degree[n] == 0:
-                    # Not ready now; a later trigger may make it ready.
-                    skipped.add(n)
 
         def drain(executor: ThreadPoolExecutor) -> None:
             submit_ready(executor)
@@ -1203,6 +1249,10 @@ class Pipeline:
                     # A node returning nothing (rate limiting, an error) stops
                     # propagation along its branch.
                     if result is not None:
+                        # Before the decrement: a consumer freed by this node
+                        # is judged on what it just published.
+                        if fresh is not None:
+                            fresh.update(result)
                         for consumer in edges[n]:
                             if consumer in target_nodes:
                                 in_degree[consumer] -= 1
@@ -1229,11 +1279,14 @@ class Pipeline:
 
         return self._execute_parallel(nodes, self._state, check_ready=False)
 
-    def apply_outputs(self, node: Node, outputs: dict[str, Any] | None) -> None:
+    def apply_outputs(self, node: Node, outputs: dict[str, Any] | None) -> set[str]:
         """Record what a node emitted: state, history, versions and events.
 
         Shared by the direct path and by the execution service replaying a
         journaled item, so a value looks the same on the canvas either way.
+
+        Returns the message names that actually reached state — post rate
+        limiting — which is what the cascade behind it has to walk from.
         """
         if outputs:
             # This is where a chatty subscriber gets thinned out, so a port set
@@ -1241,7 +1294,7 @@ class Pipeline:
             outputs = self._throttled(node, outputs)
 
         if not outputs:
-            return
+            return set()
 
         state = self._state
         ts = time.time()
@@ -1274,9 +1327,14 @@ class Pipeline:
                 "ts": ts,
             }
         )
+        return set(outputs)
 
     def run_downstream(
-        self, node: Node, entry_id: str = "", replay: bool = False
+        self,
+        node: Node,
+        entry_id: str = "",
+        replay: bool = False,
+        changed: set[str] | None = None,
     ) -> StateBackend:
         """Run everything downstream of a node that has just published.
 
@@ -1284,7 +1342,17 @@ class Pipeline:
         node with outside side effects can record that it ran. On a ``replay``
         — the same item handed back after a crash — that record is checked
         first: at-least-once delivery must not mean two of the same request.
+
+        ``changed`` names the messages this wave actually published. A node
+        none of whose triggering inputs are in it would re-run on values it has
+        already read, which is how one node came to publish 619 messages a
+        minute off inputs that changed six times. ``None`` means walk
+        everything reachable, which is what a manual run wants.
         """
+        if changed is not None and not changed:
+            # Everything the cascade carried was held back by a rate limit, so
+            # there is nothing new for anything downstream to read.
+            return self._state
         downstream = set(self._get_downstream(node))
         if not downstream:
             return self._state
@@ -1294,6 +1362,7 @@ class Pipeline:
             check_ready=True,
             entry_id=entry_id,
             replay=replay,
+            changed=changed,
         )
 
     def trigger(
@@ -1309,9 +1378,13 @@ class Pipeline:
 
         A stopped flow drops the event: its subscriptions and schedules are torn
         down anyway, and anything still arriving from another thread would be
-        work the flow was explicitly told not to do. A *paused* flow still
+        work the flow was explicitly told not to do. A *paused* flow parks the
-        publishes, so the incoming value is visible on the canvas, and holds
+        item, so nothing of it publishes until the flow is resumed or stepped.
-        the nodes downstream of it.
+
+        A value whose every port is inside its rate-limit window is kept here
+        and never journaled at all: the limit is about how often the value
+        goes out, and enqueueing a cascade to discover that costs a run record
+        and a walk of everything downstream per message.
         """
         state = self._state
 
@@ -1321,11 +1394,37 @@ class Pipeline:
         if durable is None:
             durable = self._queue is not None
         if durable and self._queue is not None:
-            self._enqueue_cascade(node, outputs)
+            if not self._held_at_source(node, outputs):
+                self._enqueue_cascade(node, outputs)
             return state
 
         return self._run_here(node, outputs)
 
+    def _held_at_source(self, node: Node, outputs: dict[str, Any] | None) -> bool:
+        """Keep a value back before it costs a cascade, or say it has to run.
+
+        Only when *every* port is inside its window: anything that would
+        publish leaves the exact split to ``apply_outputs`` at claim time,
+        which is the one place it has always been made.
+        """
+        if not outputs:
+            # A cascade carrying nothing exists to walk, not to publish.
+            return False
+        try:
+            now = time.time()
+            passed, held, due_at, pending = self._window_split(node, outputs, now)
+            if passed or not held:
+                return False
+            self._state.update(
+                {self._held_key(name): value for name, value in held.items()}
+            )
+            self._schedule_flush(node, due_at, now, pending)
+            return True
+        except Exception:
+            # Journalling it is what this was avoiding, not what it depends on.
+            logger.exception("Could not hold '%s' back at the source", node.id)
+            return False
+
     def _run_here(
         self, node: Node, outputs: dict[str, Any] | None, cause: str = "manual"
     ) -> StateBackend:
@@ -1348,8 +1447,14 @@ class Pipeline:
             }
         )
         try:
-            self.apply_outputs(node, outputs)
+            published = self.apply_outputs(node, outputs)
-            state = self.run_downstream(node, entry_id=run_id)
+            state = self.run_downstream(
+                node,
+                entry_id=run_id,
+                # No payload means the value is already in state and this is a
+                # wake-up, which has nothing to name as changed.
+                changed=published if outputs else None,
+            )
         finally:
             # Paired, or a cascade that raised leaves the run open until the
             # abandoned sweep ten minutes later.

backend/scripts/bench_engine.py

@@ -14,6 +14,11 @@ Three scenarios, each answering a different question:
     Reports cascades enqueued, node executions and bus events per injection —
     the numbers, not the wall clock, are the point.
 
+``relay``
+    The same, with the *source* unlimited: the cascade really is owed, and
+    what the relays hold back is what the node reading them should not be
+    woken for.
+
 ``throughput``
     How many messages a second, and how late is the last node? A relay chain,
     driven until the queue drains, reporting end-to-end latency percentiles.
@@ -170,7 +175,27 @@ class Engine:
     def stop(self) -> None:
         self.service.stop()
 
-    def drain(self, timeout: float = 30.0) -> bool:
+    def deliver(self, source: Node, outputs: dict[str, Any]) -> None:
+        """Inject one value and run every cascade it owes, in this thread.
+
+        The counting scenarios drive the queue by hand rather than starting
+        the consumer: injecting flat out means five hundred messages land
+        before the first cascade writes the timestamp its window is measured
+        from, so every one of them looks due and the thing being counted never
+        happens. A house sends three readings a second into an engine that
+        finishes a cascade in four milliseconds, which is this.
+        """
+        source.inject(outputs)
+        while True:
+            self.queue.move_due(0.0)
+            items = self.queue.claim(16, 1)
+            if not items:
+                return
+            for item in items:
+                self.service._run_item(item)
+                self.queue.ack(item)
+
+    def drain(self, timeout: float = 60.0) -> bool:
         """Wait for the queue to go quiet. False means it never did.
 
         Delayed items are not waited for: a rate-limit flush is due a whole
@@ -224,13 +249,11 @@ def scenario_throttle(args: argparse.Namespace) -> dict[str, Any]:
         nodes.append(relay(f"relay{i}", "raw", f"out{i}", args.window, counters))
 
     engine = Engine(nodes, args.redis, counters)
-    engine.start()
     try:
         started = time.perf_counter()
         for i in range(args.messages):
-            source.inject({"raw": float(i)})
+            engine.deliver(source, {"raw": float(i)})
-        injected = time.perf_counter() - started
+        elapsed = time.perf_counter() - started
-        drained = engine.drain()
     finally:
         engine.stop()
 
@@ -239,8 +262,7 @@ def scenario_throttle(args: argparse.Namespace) -> dict[str, Any]:
         "messages": args.messages,
         "relays": args.relays,
         "window_s": args.window,
-        "drained": drained,
+        "elapsed_s": round(elapsed, 3),
-        "inject_s": round(injected, 3),
         # The three numbers the bug is about.
         "cascades_started": counters.by_event.get("cascade_started", 0),
         "node_executions": counters.executions,
@@ -256,6 +278,62 @@ def scenario_throttle(args: argparse.Namespace) -> dict[str, Any]:
     }
 
 
+def scenario_relay(args: argparse.Namespace) -> dict[str, Any]:
+    """An *unlimited* source into limited relays, and one node reading them all.
+
+    The other half of the house's shape, and the one holding it at the source
+    cannot help with: the readings really do arrive, so a cascade is owed —
+    but the relays hold their outputs back, and what reads them has nothing
+    new to read. That node published 619 messages a minute off inputs that
+    changed six times.
+    """
+    counters = Counters()
+    source = node("source", lambda params: None, provides=[spec("raw")])
+    nodes = [source]
+    for i in range(args.relays):
+        nodes.append(relay(f"relay{i}", "raw", f"out{i}", args.window, counters))
+
+    ports = [f"out{i}" for i in range(args.relays)]
+
+    def watch(**kwargs: Any) -> dict[str, Any]:
+        counters.executions += 1
+        return {"total": sum(kwargs[p] for p in ports)}
+
+    nodes.append(
+        node(
+            "watch",
+            watch,
+            requires=[spec(p) for p in ports],
+            provides=[spec("total")],
+        )
+    )
+
+    engine = Engine(nodes, args.redis, counters)
+    try:
+        started = time.perf_counter()
+        for i in range(args.messages):
+            engine.deliver(source, {"raw": float(i)})
+        elapsed = time.perf_counter() - started
+    finally:
+        engine.stop()
+
+    return {
+        "scenario": "relay",
+        "messages": args.messages,
+        "relays": args.relays,
+        "window_s": args.window,
+        "elapsed_s": round(elapsed, 3),
+        "cascades_started": counters.by_event.get("cascade_started", 0),
+        "node_executions": counters.executions,
+        "bus_events": counters.events,
+        "published": counters.by_event.get("message_value", 0),
+        "per_message": {
+            "executions": round(counters.executions / args.messages, 3),
+            "events": round(counters.events / args.messages, 3),
+        },
+    }
+
+
 def scenario_throughput(args: argparse.Namespace) -> dict[str, Any]:
     """A relay chain, driven flat out: messages a second, and how late the tail is."""
     counters = Counters()
@@ -289,7 +367,7 @@ def scenario_throughput(args: argparse.Namespace) -> dict[str, Any]:
             with lock:
                 starts[value] = time.perf_counter()
             source.inject({"v0": value})
-        drained = engine.drain()
+        drained = engine.drain(args.drain_timeout)
         elapsed = time.perf_counter() - started
     finally:
         engine.stop()
@@ -330,7 +408,7 @@ def scenario_fanout(args: argparse.Namespace) -> dict[str, Any]:
         started = time.perf_counter()
         for i in range(args.messages):
             source.inject({"raw": float(i)})
-        drained = engine.drain()
+        drained = engine.drain(args.drain_timeout)
         elapsed = time.perf_counter() - started
     finally:
         engine.stop()
@@ -349,6 +427,7 @@ def scenario_fanout(args: argparse.Namespace) -> dict[str, Any]:
 
 SCENARIOS: dict[str, Callable[[argparse.Namespace], dict[str, Any]]] = {
     "throttle": scenario_throttle,
+    "relay": scenario_relay,
     "throughput": scenario_throughput,
     "fanout": scenario_fanout,
 }
@@ -391,6 +470,7 @@ def main() -> int:
         metavar="HOST",
         help="Run against a real Redis (namespaces bench/benchq) rather than memory.",
     )
+    parser.add_argument("--drain-timeout", type=float, default=60.0)
     parser.add_argument("--profile", action="store_true", help="cProfile, top 25.")
     parser.add_argument("--json", action="store_true", help="Machine-readable output.")
     args = parser.parse_args()

backend/tests/flow/test_discretization.py

@@ -148,6 +148,106 @@ def test_a_limited_input_wakes_its_node_when_the_window_ends():
     assert seen == [20.0, 21.0]
 
 
+def test_an_inject_inside_the_window_costs_no_cascade():
+    """The limit used to thin the messages and not the work.
+
+    It was applied after the item came off the queue, so a subscriber told to
+    publish every 15s still cost a queue entry, a run record and a walk of
+    everything downstream for every message the broker sent.
+    """
+    source = make_node(
+        "source",
+        lambda params: None,
+        provides=[spec("temp", interval=WINDOW)],
+    )
+    pipeline, queue, service = _running([source])
+
+    source.inject({"temp": 20.0})
+    for item in queue.claim(10, 10):
+        service._run_item(item)
+    assert pipeline.state["demo.temp"] == 20.0
+
+    # Inside the window: held where it is, with nothing journaled for it.
+    source.inject({"temp": 21.0})
+    assert queue.claim(10, 10) == []
+
+    # And the timer still lets it out at the end of the window.
+    time.sleep(WINDOW * 2)
+    assert _run_due(queue, service) == 1
+    assert pipeline.state["demo.temp"] == 21.0
+
+
+def test_a_node_whose_inputs_did_not_change_is_not_run():
+    """A cascade used to walk everything reachable, changed or not."""
+    seen: list[float] = []
+    source = make_node("source", lambda params: None, provides=[spec("raw")])
+    relay = make_node(
+        "relay",
+        lambda raw, params: {"level": raw},
+        requires=[spec("raw")],
+        provides=[spec("level", interval=WINDOW)],
+    )
+    watcher = make_node(
+        "watcher",
+        lambda level, params: seen.append(level),
+        requires=[spec("level")],
+    )
+    _pipeline, queue, service = _running([source, relay, watcher])
+
+    def deliver(value: float) -> None:
+        source.inject({"raw": value})
+        for item in queue.claim(10, 10):
+            service._run_item(item)
+
+    deliver(1.0)
+    assert seen == [1.0]
+
+    # The relay runs — its own input did change — but publishes nothing, so
+    # the watcher is left on the value it already has.
+    deliver(2.0)
+    assert seen == [1.0]
+
+    # The held value reaches it when the window ends.
+    time.sleep(WINDOW * 2)
+    _run_due(queue, service)
+    assert seen == [1.0, 2.0]
+
+
+def test_a_join_behind_a_skipped_branch_still_runs():
+    """Skipping a node frees its consumers, which may already have been passed."""
+    seen: list[tuple[float, float]] = []
+    source = make_node(
+        "source",
+        lambda params: None,
+        provides=[spec("fast"), spec("slow", interval=WINDOW)],
+    )
+    middle = make_node(
+        "middle",
+        lambda slow, params: {"derived": slow},
+        requires=[spec("slow")],
+        provides=[spec("derived")],
+    )
+    join = make_node(
+        "join",
+        lambda fast, derived, params: seen.append((fast, derived)),
+        requires=[spec("fast"), spec("derived")],
+    )
+    _pipeline, queue, service = _running([source, middle, join])
+
+    def deliver(value: float) -> None:
+        source.inject({"fast": value, "slow": value})
+        for item in queue.claim(10, 10):
+            service._run_item(item)
+
+    deliver(1.0)
+    assert seen == [(1.0, 1.0)]
+
+    # `slow` is held this time, so `middle` is skipped — and `join` still has
+    # to run, because `fast` did change.
+    deliver(2.0)
+    assert seen == [(1.0, 1.0), (2.0, 1.0)]
+
+
 def test_a_manual_run_is_never_throttled_on_its_inputs():
     seen: list[float] = []
     source = make_node("source", lambda params: {"temp": 20.0}, provides=[spec("temp")])