diff --git a/backend/fluksio/alembic/versions/930ffa094aad_flow_run_parent.py b/backend/fluksio/alembic/versions/930ffa094aad_flow_run_parent.py new file mode 100644 index 0000000..0018c81 --- /dev/null +++ b/backend/fluksio/alembic/versions/930ffa094aad_flow_run_parent.py @@ -0,0 +1,37 @@ +"""flow_run.parent + +A rate-limit flush is a run of its own now, and this names the run whose value +it released. Rows written before this carry an empty string, the same as every +run that continues no other. + +Revision ID: 930ffa094aad +Revises: b7d2e9c4a1f3 +Create Date: 2026-10-15 + +""" + +import sqlalchemy as sa +import sqlmodel.sql.sqltypes +from alembic import op + +# revision identifiers, used by Alembic. +revision = "930ffa094aad" +down_revision = "b7d2e9c4a1f3" +branch_labels = None +depends_on = None + + +def upgrade(): + op.add_column( + "flow_run", + sa.Column( + "parent", + sqlmodel.sql.sqltypes.AutoString(length=64), + nullable=False, + server_default="", + ), + ) + + +def downgrade(): + op.drop_column("flow_run", "parent") diff --git a/backend/fluksio/api/routes/observability.py b/backend/fluksio/api/routes/observability.py index 154a989..f72733c 100644 --- a/backend/fluksio/api/routes/observability.py +++ b/backend/fluksio/api/routes/observability.py @@ -85,6 +85,8 @@ class RunRow(BaseModel): id: str flow: str source: str + #: The run this one continues: for a flush, the run whose value it released. + parent: str = "" status: str started_at: datetime finished_at: datetime | None = None diff --git a/backend/fluksio/flow/executor.py b/backend/fluksio/flow/executor.py index 0778dfb..c183a9c 100644 --- a/backend/fluksio/flow/executor.py +++ b/backend/fluksio/flow/executor.py @@ -440,13 +440,6 @@ class ExecutionService: self.queue.park(node.flow, item) return True - if item.kind == "flush": - # A rate-limit window ended; nothing to replay, only to let out. - # ponytail: no run record for a flush — it is the tail of the run - # that scheduled it, not a run of its own. - pipeline.flush(node) - return True - if item.guard_key and str(node.recall(item.guard_key, "")) != item.guard_value: # The node moved on while this waited — a restarted timer, say. logger.debug("Guard no longer holds for '%s', dropped", item.node) @@ -461,6 +454,7 @@ class ExecutionService: "flow": item.flow, "node": item.node, "cause": item.cause, + "parent": item.parent, "deliveries": item.deliveries, "ts": now, } @@ -483,6 +477,11 @@ class ExecutionService: replay = item.deliveries > 1 emission = item.kind == "emission" try: + if item.kind == "flush": + # A rate-limit window ended; nothing to replay, only to let out. + # A run of its own, so what it sets off is recorded somewhere. + pipeline.flush(node, item.entry_id) + return True # An emission's values went into state when the node produced them; # this item carries them so its readers get the chunk that caused # the wave rather than whichever is newest by the time they run. @@ -491,7 +490,7 @@ class ExecutionService: published = ( set(item.outputs) if emission - else pipeline.apply_outputs(node, item.outputs or None) + else pipeline.apply_outputs(node, item.outputs or None, item.entry_id) ) pipeline.run_downstream( node, diff --git a/backend/fluksio/flow/metrics.py b/backend/fluksio/flow/metrics.py index 7aaa441..6eacba9 100644 --- a/backend/fluksio/flow/metrics.py +++ b/backend/fluksio/flow/metrics.py @@ -291,6 +291,7 @@ class MetricsCollector: "id": run_id, "flow": str(event.get("flow") or "")[:NAME_MAX], "source": str(event.get("cause") or ""), + "parent": str(event.get("parent") or "")[:64], "started_ts": ts, "started_at": datetime.fromtimestamp(ts, UTC), "finished_at": None, diff --git a/backend/fluksio/flow/pipeline.py b/backend/fluksio/flow/pipeline.py index 94e27c9..8076def 100644 --- a/backend/fluksio/flow/pipeline.py +++ b/backend/fluksio/flow/pipeline.py @@ -670,7 +670,7 @@ class Pipeline: return passed, held, due_at, stamps.get(flush_key) def _throttled( - self, node: Node, result: dict[str, Any] + self, node: Node, result: dict[str, Any], entry_id: str = "" ) -> tuple[dict[str, Any], list[str]]: """Hold back the outputs whose port is not due to publish yet. @@ -679,6 +679,9 @@ class Pipeline: than leaving the consumer on the one before it. Nothing declaring an interval means nothing to look up. + ``entry_id`` is the run doing the holding, which the flush names as + its parent. + Returns what passed and the held keys the publish makes stale — the caller drops them in the same round trip it writes the values with, rather than one DEL per port here. @@ -699,28 +702,35 @@ class Pipeline: self._state.update( {self._held_key(name): value for name, value in held.items()} ) - self._schedule_flush(node, due_at, now, pending) + self._schedule_flush(node, due_at, now, pending, entry_id) return passed, stale def _schedule_flush( - self, node: Node, at: float, now: float, pending: float | None + self, + node: Node, + at: float, + now: float, + pending: float | None, + entry_id: str = "", ) -> None: """Come back to this node when its rate-limit window ends. One timer in flight per node: a pending one that is early enough is left alone, and one that turns out to be too late simply finds nothing - to do when it fires. + to do when it fires. So the parent a flush names is the run that booked + it, which held the first value of the window, not necessarily the last. """ if self._queue is None or (pending and now < pending <= at): return self._state[self._flush_key(node.id)] = at - self.defer(node, {}, at - now, kind="flush") + self.defer(node, {}, at - now, kind="flush", parent=entry_id) - def flush(self, node: Node) -> None: + def flush(self, node: Node, entry_id: str = "") -> None: """Let through what this node's rate limits held back. Runs when a window ends: the last value a limited output kept back is published, and a node whose inputs were all held back gets its run. + ``entry_id`` is the flush's own run. """ limited_out = [name for name, s in node.provides.items() if s.interval > 0] stored = ( @@ -737,14 +747,17 @@ class Pipeline: # Publishing runs the limit again, which is what clears the hold — # or puts the timer back, if this fired a hair early. In that case # nothing came out and there is nothing downstream to wake. - published = self.apply_outputs(node, held) + published = self.apply_outputs(node, held, entry_id) if published: - self.run_downstream(node, changed=published) + self.run_downstream(node, entry_id=entry_id, 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 + {node, *self._get_downstream(node)}, + self._state, + check_ready=True, + entry_id=entry_id, ) def _increment_message_versions(self, outputs: dict[str, Any]) -> None: @@ -791,7 +804,7 @@ class Pipeline: return self._state.compare_and_swap_multi(expected, updates) - def _input_is_due(self, node: Node) -> bool: + def _input_is_due(self, node: Node, entry_id: str = "") -> bool: """Has any rate-limited input waited out its interval? A node runs when *any* of its inputs is due, so a slow port next to a @@ -827,6 +840,7 @@ class Pipeline: ), now, stamps.get(flush_key), + entry_id, ) return False @@ -834,7 +848,7 @@ class Pipeline: return True def _is_node_ready( - self, node: Node, state: StateBackend + self, node: Node, state: StateBackend, entry_id: str = "" ) -> tuple[bool, dict[str, Any]]: """Whether the node may run, and the inputs the answer was read from. @@ -851,7 +865,7 @@ class Pipeline: if any(msg not in values for msg in waited_on): return False, values - if not self._input_is_due(node): + if not self._input_is_due(node, entry_id): return False, values if not node.synchronous: @@ -1062,7 +1076,7 @@ class Pipeline: stale: list[str] = [] if result: - result, stale = self._throttled(node, result) + result, stale = self._throttled(node, result, entry_id) if result: self._record_outputs(node, result, state, stale) @@ -1358,7 +1372,7 @@ class Pipeline: # The values are kept whether or not the answer is yes: # a node that may run reads them below instead of # asking state for the same keys again. - ready, ready_inputs = self._is_node_ready(n, state) + ready, ready_inputs = self._is_node_ready(n, state, entry_id) if not ready: # Completed, not merely passed over: a later trigger # may still make it ready, but not running is not @@ -1440,11 +1454,14 @@ class Pipeline: return self._execute_parallel(nodes, self._state, check_ready=False) - def apply_outputs(self, node: Node, outputs: dict[str, Any] | None) -> set[str]: + def apply_outputs( + self, node: Node, outputs: dict[str, Any] | None, entry_id: str = "" + ) -> 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. + ``entry_id`` is the run this happens in. Returns the message names that actually reached state — post rate limiting — which is what the cascade behind it has to walk from. @@ -1453,7 +1470,7 @@ class Pipeline: if outputs: # This is where a chatty subscriber gets thinned out, so a port set # to publish every 60s does so whatever the broker sends. - outputs, stale = self._throttled(node, outputs) + outputs, stale = self._throttled(node, outputs, entry_id) if not outputs: return set() @@ -1569,6 +1586,7 @@ class Pipeline: self._state.update( {self._held_key(name): value for name, value in held.items()} ) + # Held before any run began, so its flush names no parent. self._schedule_flush(node, due_at, now, pending) return True except Exception: @@ -1607,7 +1625,9 @@ class Pipeline: ) try: published = ( - set(outputs or {}) if delivered else self.apply_outputs(node, outputs) + set(outputs or {}) + if delivered + else self.apply_outputs(node, outputs, run_id) ) state = self.run_downstream( node, @@ -1684,6 +1704,7 @@ class Pipeline: seconds: float, guard: tuple[str, Any] | None = None, kind: str = "cascade", + parent: str = "", ) -> bool: """Publish a node's outputs later, without holding a worker thread. @@ -1691,6 +1712,8 @@ class Pipeline: over; if it has moved on, the item is dropped. That is how a wait which gets restarted cancels the one it replaced. + ``parent`` is the run the item continues, recorded beside its own. + Returns False when there is no queue to hold the item, in which case the caller has to wait however it waited before. """ @@ -1703,9 +1726,10 @@ class Pipeline: node=node.id, flow=node.flow, outputs=outputs, - cause="delay", + cause="flush" if kind == "flush" else "delay", guard_key=guard[0] if guard else "", guard_value=str(guard[1]) if guard else "", + parent=parent, ) try: self._queue.add_delayed(item, time.time() + seconds) diff --git a/backend/fluksio/flow/queue.py b/backend/fluksio/flow/queue.py index 6837616..3d46bad 100644 --- a/backend/fluksio/flow/queue.py +++ b/backend/fluksio/flow/queue.py @@ -94,6 +94,9 @@ class WorkItem: #: The run this item is, on a ``run`` item. Everything else about a run is #: in its database row, so the journal only has to name it. run_id: str = "" + #: The run that scheduled this item, on a ``flush``: the one whose value + #: the rate limit held back. + parent: str = "" def to_fields(self) -> dict[str, str]: return { @@ -107,6 +110,7 @@ class WorkItem: "guard_value": self.guard_value, "enqueued_at": str(self.enqueued_at), "run_id": self.run_id, + "parent": self.parent, } @classmethod @@ -126,6 +130,7 @@ class WorkItem: deliveries=deliveries, enqueued_at=float(fields.get("enqueued_at") or time.time()), run_id=fields.get("run_id", ""), + parent=fields.get("parent", ""), ) diff --git a/backend/fluksio/models.py b/backend/fluksio/models.py index 3235b06..16152e6 100644 --- a/backend/fluksio/models.py +++ b/backend/fluksio/models.py @@ -298,6 +298,9 @@ class FlowRun(SQLModel, table=True): flow: str = Field(index=True, max_length=255) #: What caused it: the work item's cause, "manual", or "run" for a batch. source: str = "" + #: The run this one continues: on a rate-limit flush, the run whose value + #: the window held back. Empty for everything else. + parent: str = Field(default="", max_length=64) started_at: datetime = Field( index=True, sa_type=UTCDateTime, diff --git a/backend/tests/test_metrics.py b/backend/tests/test_metrics.py index 688573e..7d8d03a 100644 --- a/backend/tests/test_metrics.py +++ b/backend/tests/test_metrics.py @@ -11,7 +11,12 @@ from datetime import UTC, datetime from sqlmodel import Session, select from fluksio.flow.events import EventBus +from fluksio.flow.executor import ExecutionService +from fluksio.flow.messages import DType, MessageSpec from fluksio.flow.metrics import DETAIL_CAP, RUN_STALE_S, MetricsCollector +from fluksio.flow.nodes import Node +from fluksio.flow.pipeline import Pipeline +from fluksio.flow.queue import MemoryWorkQueue from fluksio.models import EngineEvent, FlowRun, MetricBucket FLOW = "metrics-test" @@ -292,3 +297,52 @@ def test_a_failure_keeps_the_run_it_happened_in(db: Session) -> None: failure = db.exec(select(EngineEvent).where(EngineEvent.run == "run-abc123")).one() assert failure.detail.startswith("ValueError: in a run") + + +def test_a_rate_limit_flush_is_a_run_that_names_its_parent(db: Session) -> None: + """A flush used to run under no id, so what it set off was in no run.""" + collector = MetricsCollector(EventBus()) + + class Bus: + """Hands every event straight to the collector.""" + + def publish(self, event: dict) -> None: + collector.handle(event) + + window = 0.05 + source = Node( + f=lambda params: None, + provides=[MessageSpec(name="raw", dtype=DType.FLOAT)], + name="in", + ) + relay = Node( + f=lambda raw, params: {"level": raw}, + requires=[MessageSpec(name="raw", dtype=DType.FLOAT)], + provides=[MessageSpec(name="level", dtype=DType.FLOAT, interval=window)], + name="relay", + ) + source.assign_flow("metrics-flush", "in") + relay.assign_flow("metrics-flush", "relay") + queue = MemoryWorkQueue() + pipeline = Pipeline(nodes=[source, relay], work_queue=queue, events=Bus()) + service = ExecutionService(queue, events=Bus()) + service.bind(pipeline) + + # The second value lands inside the relay's window and is held. + held_in = "" + for value in (1.0, 2.0): + source.inject({"raw": value}) + for item in queue.claim(10, 10): + service._run_item(item) + held_in = item.entry_id + time.sleep(window * 2) + queue.move_due(time.time()) + for item in queue.claim(10, 10): + service._run_item(item) + asyncio.run(collector.flush()) + + flush = db.exec(select(FlowRun).where(FlowRun.source == "flush")).one() + assert (flush.status, flush.parent) == ("ok", held_in) + parent = db.get(FlowRun, held_in) + assert parent is not None + assert (parent.flow, parent.source) == ("metrics-flush", "external") diff --git a/frontend/src/client/schemas.gen.ts b/frontend/src/client/schemas.gen.ts index 3349cee..69c579f 100644 --- a/frontend/src/client/schemas.gen.ts +++ b/frontend/src/client/schemas.gen.ts @@ -4777,6 +4777,11 @@ export const fluksio__api__routes__observability__RunRowSchema = { type: 'string', title: 'Source' }, + parent: { + type: 'string', + title: 'Parent', + default: '' + }, status: { type: 'string', title: 'Status' diff --git a/frontend/src/client/types.gen.ts b/frontend/src/client/types.gen.ts index 0f8c144..6520c41 100644 --- a/frontend/src/client/types.gen.ts +++ b/frontend/src/client/types.gen.ts @@ -506,6 +506,7 @@ export type fluksio__api__routes__observability__RunRow = { id: string; flow: string; source: string; + parent?: string; status: string; started_at: string; finished_at?: (string | null);