列表上「有几条在等」按任务各取各的上限（复核 L4）：等着的事件按任务、还没派发的事件按（事件种类, 对象）用 ROW_NUMBER 各取前 N 条（同改前每个任务各读的那么多），不再所有任务合起来一个上限——积压多的任务会把别的挤掉，它们的 due_at 为空、未派发数少算。补测试

backend/app/scheduler/outbox.py

@@ -559,11 +559,17 @@ async def pending_for_many(session: Any, schedules: list[Schedule]) -> dict[str,
     counts = dict((await session.execute(select(TriggerDelivery.schedule_id, func.count()).where(
         TriggerDelivery.schedule_id.in_(ids), TriggerDelivery.status == "pending")
         .group_by(TriggerDelivery.schedule_id))).all())
+    # 每个任务各读最早的 DRAIN_BATCH 条（同 _drain 一拍读的）：合起来一个上限的话，积压多的任务会把别的挤掉，它们的 due_at 为空
+    ranked = (select(TriggerDelivery.id.label("did"), func.row_number().over(
+        partition_by=TriggerDelivery.schedule_id,
+        order_by=(TriggerDelivery.occurred_at, TriggerDelivery.created_at, TriggerDelivery.id)).label("rn"))
+        .where(TriggerDelivery.schedule_id.in_([i for i in ids if counts.get(i)]), TriggerDelivery.status == "pending")
+        .subquery())
     waiting = (await session.execute(
         select(TriggerDelivery.schedule_id, TriggerDelivery.id, TriggerDelivery.created_at, TriggerEvent)
         .join(TriggerEvent, TriggerEvent.id == TriggerDelivery.event_id)
-        .where(TriggerDelivery.schedule_id.in_([i for i in ids if counts.get(i)]), TriggerDelivery.status == "pending")
-        .order_by(TriggerDelivery.occurred_at, TriggerEvent.created_at).limit(DRAIN_BATCH))).all() if counts else []
+        .join(ranked, ranked.c.did == TriggerDelivery.id).where(ranked.c.rn <= DRAIN_BATCH)
+        .order_by(TriggerDelivery.occurred_at, TriggerEvent.created_at))).all() if counts else []
     by: dict[str, list[tuple[str, TriggerEvent, datetime]]] = {}
     stale: dict[str, int] = {}
     for sid, did, routed, ev in waiting:
@@ -585,8 +591,14 @@ async def pending_for_many(session: Any, schedules: list[Schedule]) -> dict[str,
 
         cond = or_(*[and_(TriggerEvent.event == kind, TriggerEvent.subject_id.in_(sorted(subs)))
                      for kind, subs in pairs.items()])
+        # 每个（事件种类, 对象）各读前 ROUTE_BATCH 条（同改前每个任务各读的那么多）：合起来一个上限会把别的对象挤掉、少算
+        ranked = (select(TriggerEvent.id.label("eid"), func.row_number().over(
+            partition_by=(TriggerEvent.event, TriggerEvent.subject_id),
+            order_by=(TriggerEvent.created_at, TriggerEvent.id)).label("rn"))
+            .where(TriggerEvent.status == "pending", cond).subquery())
         unrouted = (await session.execute(select(TriggerEvent.event, TriggerEvent.subject_id, TriggerEvent.payload)
-                                          .where(TriggerEvent.status == "pending", cond).limit(ROUTE_BATCH))).all()
+                                          .join(ranked, ranked.c.eid == TriggerEvent.id)
+                                          .where(ranked.c.rn <= ROUTE_BATCH))).all()
         for kind, subject, payload in unrouted:
             for sid in ids:
                 if triggers.matches(trig[sid], kind, subject, payload):

backend/tests/test_schedule_events_scale.py

@@ -185,3 +185,30 @@ async def test_trigger_option_search_ignores_case_and_takes_wildcards_literally(
 
     assert await names("weekly_report") == ["Weekly_Report 样例"], "大小写不分；_ 不匹配任意一个字"
     assert await names("100%") == ["满 100% 样例"], "% 不匹配任意几个字"
+
+
+async def test_a_backlog_on_one_schedule_does_not_crowd_out_the_others(monkeypatch):
+    """积压多的任务不把别的任务挤掉：等着的事件、还没派发的事件按任务（按对象）各读各的上限，不是合起来一个上限。"""
+    monkeypatch.setattr(outbox, "DRAIN_BATCH", 3)
+    monkeypatch.setattr(outbox, "ROUTE_BATCH", 3)
+    busy_up, quiet_up = await workflow("积压的上游"), await workflow("安静的上游")
+    downstream = await workflow("区域汇总")
+    busy = await event_schedule(downstream, {"event": "workflow.finished", "workflow_id": busy_up, "debounce_s": 600},
+                                name="积压的")
+    quiet = await event_schedule(downstream, {"event": "workflow.finished", "workflow_id": quiet_up, "debounce_s": 600},
+                                 name="安静的")
+    t0 = utcnow().replace(microsecond=0)
+    for i in range(5):
+        await put_event("workflow.finished", busy_up, {"workflow_id": busy_up, "run_id": f"b{i}"}, t0 + timedelta(seconds=i))
+    await put_event("workflow.finished", quiet_up, {"workflow_id": quiet_up, "run_id": "q0"}, t0 + timedelta(seconds=10))
+    for _ in range(4):                       # 一拍只派 ROUTE_BATCH 条：多派几拍，六条都派给任务
+        await outbox.route(t0 + timedelta(seconds=20))
+    for i in range(5):
+        await put_event("workflow.finished", busy_up, {"workflow_id": busy_up, "run_id": f"c{i}"}, t0 + timedelta(seconds=30 + i))
+    await put_event("workflow.finished", quiet_up, {"workflow_id": quiet_up, "run_id": "q1"}, t0 + timedelta(seconds=40))
+    async with SessionLocal() as session:
+        rows = list((await session.execute(select(Schedule).where(Schedule.id.in_([busy, quiet])))).scalars())
+        state = await outbox.pending_for_many(session, rows)
+    assert state[quiet]["due_at"] == (t0 + timedelta(seconds=610)).isoformat(), state
+    assert state[quiet]["events"] == 1 and state[busy]["events"] == 3, state
+    assert state[busy]["due_at"] == (t0 + timedelta(seconds=600)).isoformat() and state[busy]["waiting"] == 5