Give a rate-limit flush its own run, linked to the run that held the value

A flush ran its cascade under no id, so its executions and failures belonged
to no run. It now records a run with source "flush", and the holding run's id
travels on the queue item into a new `flow_run.parent` column, also on the
`/observability/runs` rows.

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