From aecef46436c066e7486e60e141774352582e7e7e Mon Sep 17 00:00:00 2001 From: roboomp Date: Fri, 17 Jul 2026 17:26:39 +0000 Subject: [PATCH 1/2] fix(robomp): deferred rate-limited submissions Stored a bounded per-login overflow backlog and promoted deferred events oldest-first as rolling-window capacity became available. Surfaced the deferred state through the dashboard contract and documented admission behavior. Fixes #5882 --- python/robomp/.env.example | 2 + python/robomp/README.md | 7 + python/robomp/src/db.py | 193 +++++++++++++++++- python/robomp/src/queue.py | 11 +- python/robomp/src/server.py | 13 +- python/robomp/tests/test_db.py | 42 ++++ python/robomp/tests/test_queue_dispatch.py | 22 ++ python/robomp/tests/test_server.py | 30 +-- .../web/src/components/shell/Vitals.tsx | 3 +- python/robomp/web/src/types.ts | 3 +- python/robomp/web/src/work-items.test.ts | 1 + .../web/test/fixtures/status-contract.json | 2 + 12 files changed, 297 insertions(+), 32 deletions(-) diff --git a/python/robomp/.env.example b/python/robomp/.env.example index a338cdf82..320bec941 100644 --- a/python/robomp/.env.example +++ b/python/robomp/.env.example @@ -144,6 +144,8 @@ ROBOMP_VOUCH_REVIEW_LABELER=github-actions[bot] # OWNER/MEMBER/COLLABORATOR bypass the limiter automatically. Use the # unlimited list (comma-separated logins, `@` optional) to whitelist # additional users — e.g. yourself when developing outside the repo. +# At capacity, up to the tier cap is retained as a deferred oldest-first +# backlog; overflow beyond that bound remains skipped. ROBOMP_RATE_LIMIT_WINDOW_SECONDS=3600 ROBOMP_RATE_LIMIT_DEFAULT=3 ROBOMP_RATE_LIMIT_CONTRIBUTOR=10 diff --git a/python/robomp/README.md b/python/robomp/README.md index 558c9bf4d..1da19607e 100644 --- a/python/robomp/README.md +++ b/python/robomp/README.md @@ -39,6 +39,13 @@ Flow: webhook → HMAC verify → `github_events.route` → sqlite `events` → `worker.run_task` spawns `omp --mode rpc` with `cwd=worktree`, persistent `session_dir`, model randomly drawn from `ROBOMP_MODEL` (CSV). +Queue-worthy submissions use a per-login rolling admission window. Accounts +reported as `OWNER`, `MEMBER`, or `COLLABORATOR`, plus configured unlimited +logins, bypass it. When a capped login fills its window, roboomp retains up to +that cap again as a deferred backlog and re-admits those events oldest-first as +slots expire; further overflow remains skipped. Configure the window and tier +caps with the `ROBOMP_RATE_LIMIT_*` variables in `.env.example`. + The agent uses omp's built-in tools (`read`/`edit`/`bash`/`lsp`, scoped to the worktree) plus the host tools in `src/host_tools.py` — the exclusive surface for GitHub writes. Every host-tool invocation is audited diff --git a/python/robomp/src/db.py b/python/robomp/src/db.py index 7fa13940d..46727533e 100644 --- a/python/robomp/src/db.py +++ b/python/robomp/src/db.py @@ -14,7 +14,7 @@ from typing import Any, Literal from robomp.github_client import IssueIndexEntry -EventState = Literal["queued", "running", "done", "failed", "skipped"] +EventState = Literal["queued", "deferred", "running", "done", "failed", "skipped"] INACTIVE_EVENT_STATES: tuple[EventState, ...] = ("done", "failed", "skipped") IssueState = Literal[ @@ -42,7 +42,7 @@ CREATE TABLE IF NOT EXISTS events ( payload_json TEXT NOT NULL, received_at TEXT NOT NULL, state TEXT NOT NULL - CHECK (state IN ('queued','running','done','failed','skipped')), + CHECK (state IN ('queued','deferred','running','done','failed','skipped')), attempts INTEGER NOT NULL DEFAULT 0, last_error TEXT, started_at TEXT, @@ -101,6 +101,16 @@ CREATE TABLE IF NOT EXISTS submissions ( ); CREATE INDEX IF NOT EXISTS submissions_login_ts ON submissions(login, ts); +CREATE TABLE IF NOT EXISTS deferred_submissions ( + delivery_id TEXT PRIMARY KEY, + login TEXT NOT NULL, + repo TEXT, + cap INTEGER NOT NULL CHECK (cap > 0), + created_at TEXT NOT NULL +); +CREATE INDEX IF NOT EXISTS deferred_submissions_login_created + ON deferred_submissions(login, created_at); + CREATE TABLE IF NOT EXISTS pending_closures ( issue_key TEXT PRIMARY KEY, repo TEXT NOT NULL, @@ -299,6 +309,47 @@ class Database: if "available_at" not in event_cols: self._conn.execute("ALTER TABLE events ADD COLUMN available_at TEXT") + event_table = self._conn.execute( + "SELECT sql FROM sqlite_master WHERE type='table' AND name='events'" + ).fetchone() + event_sql = str(event_table["sql"] or "") if event_table is not None else "" + if "'deferred'" not in event_sql: + with self._txn() as conn: + conn.execute( + """ + CREATE TABLE events_v2 ( + delivery_id TEXT PRIMARY KEY, + event_type TEXT NOT NULL, + repo TEXT, + issue_key TEXT, + payload_json TEXT NOT NULL, + received_at TEXT NOT NULL, + state TEXT NOT NULL + CHECK (state IN ('queued','deferred','running','done','failed','skipped')), + attempts INTEGER NOT NULL DEFAULT 0, + last_error TEXT, + started_at TEXT, + finished_at TEXT, + model TEXT, + available_at TEXT + ) + """ + ) + conn.execute( + """ + INSERT INTO events_v2 + (delivery_id, event_type, repo, issue_key, payload_json, received_at, + state, attempts, last_error, started_at, finished_at, model, available_at) + SELECT delivery_id, event_type, repo, issue_key, payload_json, received_at, + state, attempts, last_error, started_at, finished_at, model, available_at + FROM events + """ + ) + conn.execute("DROP TABLE events") + conn.execute("ALTER TABLE events_v2 RENAME TO events") + conn.execute("CREATE INDEX events_state_received ON events(state, received_at)") + conn.execute("CREATE INDEX events_issue_state ON events(issue_key, state)") + def close(self) -> None: with self._lock: self._conn.close() @@ -452,8 +503,9 @@ class Database: def remove_event(self, delivery_id: str) -> None: """Hard-delete an event row. Used to clear stale state before a manual re-trigger.""" - with self._lock: - self._conn.execute("DELETE FROM events WHERE delivery_id=?", (delivery_id,)) + with self._txn() as conn: + conn.execute("DELETE FROM deferred_submissions WHERE delivery_id=?", (delivery_id,)) + conn.execute("DELETE FROM events WHERE delivery_id=?", (delivery_id,)) def replace_event_if_state_in( self, @@ -476,6 +528,7 @@ class Database: if row is not None: if row["state"] not in allowed_existing_states: return False + conn.execute("DELETE FROM deferred_submissions WHERE delivery_id = ?", (delivery_id,)) conn.execute("DELETE FROM events WHERE delivery_id = ?", (delivery_id,)) conn.execute( """ @@ -557,7 +610,7 @@ class Database: """Return current row counts per event state, including states with zero rows.""" with self._lock: rows = self._conn.execute("SELECT state, COUNT(*) AS n FROM events GROUP BY state").fetchall() - counts: dict[str, int] = dict.fromkeys(("queued", "running", "done", "failed", "skipped"), 0) + counts: dict[str, int] = dict.fromkeys(("queued", "deferred", "running", "done", "failed", "skipped"), 0) for row in rows: counts[row["state"]] = int(row["n"]) return counts @@ -569,7 +622,7 @@ class Database: run clears an older failure for that issue, and ignored webhook noise does not make a failed issue look skipped. """ - counts: dict[str, int] = dict.fromkeys(("queued", "running", "done", "failed", "skipped"), 0) + counts: dict[str, int] = dict.fromkeys(("queued", "deferred", "running", "done", "failed", "skipped"), 0) seen: set[str] = set() with self._lock: rows = self._conn.execute( @@ -689,9 +742,9 @@ class Database: keep public retries from mutating queued/running rows while preserving internal recovery of a just-claimed running event. """ - with self._lock: + with self._txn() as conn: if from_states is None: - cur = self._conn.execute( + cur = conn.execute( "UPDATE events SET state='queued', available_at=NULL WHERE delivery_id=?", (delivery_id,), ) @@ -699,10 +752,12 @@ class Database: return False else: placeholders = ",".join("?" for _ in from_states) - cur = self._conn.execute( + cur = conn.execute( f"UPDATE events SET state='queued', available_at=NULL WHERE delivery_id=? AND state IN ({placeholders})", (delivery_id, *from_states), ) + if cur.rowcount > 0: + conn.execute("DELETE FROM deferred_submissions WHERE delivery_id=?", (delivery_id,)) return cur.rowcount > 0 def schedule_retry(self, delivery_id: str, *, delay_seconds: float, error: str | None = None) -> bool: @@ -1037,6 +1092,22 @@ class Database: used=int(row["n"]) if row is not None else 0, ) + if cap is not None: + deferred = conn.execute( + "SELECT 1 FROM deferred_submissions WHERE login=? LIMIT 1", + (normalized_login,), + ).fetchone() + if deferred is not None: + row = conn.execute( + "SELECT COUNT(*) AS n FROM submissions WHERE login=? AND ts>=?", + (normalized_login, since), + ).fetchone() + return SubmissionAdmission( + accepted=False, + duplicate=False, + used=int(row["n"]) if row is not None else 0, + ) + row = conn.execute( "SELECT COUNT(*) AS n FROM submissions WHERE login=? AND ts>=?", (normalized_login, since), @@ -1051,6 +1122,110 @@ class Database: ) return SubmissionAdmission(accepted=True, duplicate=False, used=used + 1) + def defer_submission_event( + self, + *, + delivery_id: str, + event_type: str, + login: str, + repo: str | None, + issue_key: str | None, + payload: Mapping[str, Any], + cap: int, + reason: str, + ) -> bool: + """Persist bounded rate-limit overflow for later admission. + + Returns True when the event is deferred, including idempotent redelivery + of an existing deferred event. Overflow beyond one cap-sized backlog is + retained as an ordinary skipped event for operator diagnosis. + """ + normalized_login = login.lower() + now = _utcnow() + with self._txn() as conn: + existing = conn.execute( + "SELECT state FROM events WHERE delivery_id=?", + (delivery_id,), + ).fetchone() + if existing is not None: + return existing["state"] == "deferred" + + row = conn.execute( + "SELECT COUNT(*) AS n FROM deferred_submissions WHERE login=?", + (normalized_login,), + ).fetchone() + backlog = int(row["n"]) if row is not None else 0 + deferred = cap > 0 and backlog < cap + state: EventState = "deferred" if deferred else "skipped" + last_error = reason if deferred else f"{reason}; deferred backlog full ({backlog}/{cap})" + conn.execute( + """ + INSERT INTO events + (delivery_id, event_type, repo, issue_key, payload_json, received_at, state, last_error) + VALUES (?, ?, ?, ?, ?, ?, ?, ?) + """, + ( + delivery_id, + event_type, + repo, + issue_key, + json.dumps(payload, separators=(",", ":")), + now, + state, + last_error, + ), + ) + if deferred: + conn.execute( + """ + INSERT INTO deferred_submissions (delivery_id, login, repo, cap, created_at) + VALUES (?, ?, ?, ?, ?) + """, + (delivery_id, normalized_login, repo, cap, now), + ) + return deferred + + def promote_deferred_submissions(self, *, since: str) -> int: + """Admit the oldest deferred events when their submitters have capacity.""" + promoted = 0 + now = _utcnow() + with self._txn() as conn: + rows = conn.execute( + """ + SELECT deferred.delivery_id, deferred.login, deferred.repo, deferred.cap + FROM deferred_submissions AS deferred + JOIN events ON events.delivery_id = deferred.delivery_id + WHERE events.state = 'deferred' + ORDER BY events.received_at, events.rowid + """ + ).fetchall() + for row in rows: + usage = conn.execute( + "SELECT COUNT(*) AS n FROM submissions WHERE login=? AND ts>=?", + (row["login"], since), + ).fetchone() + used = int(usage["n"]) if usage is not None else 0 + if used >= int(row["cap"]): + continue + conn.execute( + "INSERT OR IGNORE INTO submissions (delivery_id, login, repo, ts) VALUES (?, ?, ?, ?)", + (row["delivery_id"], row["login"], row["repo"], now), + ) + conn.execute( + """ + UPDATE events + SET state='queued', last_error=NULL, available_at=NULL, finished_at=NULL + WHERE delivery_id=? AND state='deferred' + """, + (row["delivery_id"],), + ) + conn.execute( + "DELETE FROM deferred_submissions WHERE delivery_id=?", + (row["delivery_id"],), + ) + promoted += 1 + return promoted + def record_submission( self, *, diff --git a/python/robomp/src/queue.py b/python/robomp/src/queue.py index 1174631c5..37e24dcbe 100644 --- a/python/robomp/src/queue.py +++ b/python/robomp/src/queue.py @@ -12,7 +12,7 @@ from contextlib import suppress from robomp import tasks from robomp.cancellation import clear_current_event, set_current_event from robomp.config import Settings -from robomp.db import Database, EventRow +from robomp.db import Database, EventRow, iso_seconds_ago from robomp.github_backend import GitHubBackend from robomp.sandbox import GitTransport, SandboxManager, _reap_slot from robomp.slot_pool import SlotPool @@ -214,7 +214,14 @@ class WorkerPool: # Naive but fine for v1 (small queue). row = await asyncio.to_thread(self.db.claim_next_event) if row is None: - return None + since = iso_seconds_ago(self.settings.rate_limit_window_seconds) + promoted = await asyncio.to_thread(self.db.promote_deferred_submissions, since=since) + if not promoted: + return None + log.info("deferred submissions promoted", extra={"count": promoted}) + row = await asyncio.to_thread(self.db.claim_next_event) + if row is None: + return None key = row.issue_key or row.delivery_id if key in self._inflight: # Put it back; another in-flight task is touching the same issue. diff --git a/python/robomp/src/server.py b/python/robomp/src/server.py index 36aa3d5a5..47ea10680 100644 --- a/python/robomp/src/server.py +++ b/python/robomp/src/server.py @@ -434,6 +434,7 @@ def create_app(settings: Settings | None = None) -> FastAPI: cap=cap, ) if not admission.accepted: + assert cap is not None window = int(cfg.rate_limit_window_seconds) reason = f"rate limit: @{submitter} has used {admission.used}/{cap} submissions in the last {window}s" log.info( @@ -447,17 +448,21 @@ def create_app(settings: Settings | None = None) -> FastAPI: "cap": cap, }, ) - db.record_event( + deferred = db.defer_submission_event( delivery_id=x_github_delivery, event_type=x_github_event, + login=submitter, repo=decision.repo, issue_key=decision.issue_key, payload=payload, - state="skipped", - last_error=reason, + cap=cap, + reason=reason, ) + event_state = "deferred" if deferred else "skipped" + if deferred: + bag["pool"].wake() return JSONResponse( - {"delivery": x_github_delivery, "state": "skipped", "reason": "rate_limited"}, + {"delivery": x_github_delivery, "state": event_state, "reason": "rate_limited"}, status_code=202, ) diff --git a/python/robomp/tests/test_db.py b/python/robomp/tests/test_db.py index 9ff0644e6..4d599a84b 100644 --- a/python/robomp/tests/test_db.py +++ b/python/robomp/tests/test_db.py @@ -368,6 +368,15 @@ def test_migration_adds_classification_to_existing_db(tmp_path: Path) -> None: assert row.classification is None # column exists, default NULL database.set_issue_classification("octo/widget#1", "bug") assert database.get_issue("octo/widget#1").classification == "bug" + assert database.record_event( + delivery_id="deferred", + event_type="issues", + repo="octo/widget", + issue_key="octo/widget#2", + payload={"action": "opened"}, + state="deferred", + ) + assert database.get_event("deferred").state == "deferred" database.close() @@ -466,6 +475,39 @@ def test_admit_submission_dedupes_by_delivery_before_rate_limit(db: Database) -> assert db.count_submissions_since("alice", since) == 1 +def test_deferred_submissions_are_promoted_oldest_first_when_capacity_frees(db: Database) -> None: + since = iso_seconds_ago(60) + for i in range(2): + assert db.record_submission(delivery_id=f"accepted-{i}", login="Alice", repo="octo/widget") + for i in range(2): + assert db.defer_submission_event( + delivery_id=f"deferred-{i}", + event_type="issues", + login="alice", + repo="octo/widget", + issue_key=f"octo/widget#{i + 10}", + payload={"action": "opened"}, + cap=2, + reason="rate limit", + ) + + newer = db.admit_submission( + delivery_id="newer", + login="alice", + repo="octo/widget", + since=iso_seconds_ago(-1), + cap=2, + ) + assert not newer.accepted + assert db.promote_deferred_submissions(since=iso_seconds_ago(-1)) == 2 + + first = db.claim_next_event() + second = db.claim_next_event() + assert first is not None and first.delivery_id == "deferred-0" + assert second is not None and second.delivery_id == "deferred-1" + assert db.count_submissions_since("alice", since) == 4 + + def test_admit_submission_enforces_cap_atomically_across_connections(tmp_path: Path) -> None: path = tmp_path / "admission.sqlite" # Pre-warm: open + migrate the schema once so the two racing threads below diff --git a/python/robomp/tests/test_queue_dispatch.py b/python/robomp/tests/test_queue_dispatch.py index 9ffa056be..12517b95d 100644 --- a/python/robomp/tests/test_queue_dispatch.py +++ b/python/robomp/tests/test_queue_dispatch.py @@ -92,3 +92,25 @@ async def test_dispatch_pr_synchronize_is_noop( await _make_pool(settings, db)._dispatch(_pr_row("synchronize")) # noqa: SLF001 assert called is False + + +@pytest.mark.asyncio +async def test_claim_promotes_deferred_submission_after_window_frees(settings: Settings, db: Database) -> None: + assert db.record_submission(delivery_id="accepted", login="alice", repo="octo/widget") + assert db.defer_submission_event( + delivery_id="deferred", + event_type="issues", + login="alice", + repo="octo/widget", + issue_key="octo/widget#8", + payload={"action": "opened", "issue": {"number": 8}}, + cap=1, + reason="rate limit", + ) + settings.rate_limit_window_seconds = -1 + + row = await _make_pool(settings, db)._claim_next_unique() # noqa: SLF001 + + assert row is not None + assert row.delivery_id == "deferred" + assert row.state == "running" diff --git a/python/robomp/tests/test_server.py b/python/robomp/tests/test_server.py index 7141fddc3..cd3f270f2 100644 --- a/python/robomp/tests/test_server.py +++ b/python/robomp/tests/test_server.py @@ -112,8 +112,8 @@ def test_api_status_reports_runtime_counts_and_inflight(settings: Settings) -> N assert runtime["uptime_seconds"] >= 0 counts = body["event_counts"] - # All five buckets must be present even when zero — the UI relies on it. - assert set(counts) == {"queued", "running", "done", "failed", "skipped"} + # All six buckets must be present even when zero — the UI relies on it. + assert set(counts) == {"queued", "deferred", "running", "done", "failed", "skipped"} assert counts["queued"] + counts["running"] == 2 # d-queued + d-running assert counts["skipped"] == 1 assert counts["running"] >= 1 @@ -905,9 +905,9 @@ def rate_limited_settings(monkeypatch: pytest.MonkeyPatch, env: dict[str, str]) def test_webhook_rate_limits_unknown_submitter_at_default_cap(rate_limited_settings: Settings) -> None: app = create_app(rate_limited_settings) with TestClient(app) as client: - # Default cap is 2 → first two queued, third throttled. + # Default cap is 2 → two queue, two enter the bounded backlog, then overflow skips. states = [] - for i in range(3): + for i in range(5): resp = _post_issue_opened( client, delivery=f"d-{i}", @@ -918,7 +918,7 @@ def test_webhook_rate_limits_unknown_submitter_at_default_cap(rate_limited_setti assert resp.status_code == 202 states.append(resp.json()["state"]) close_database() - assert states == ["queued", "queued", "skipped"] + assert states == ["queued", "queued", "deferred", "deferred", "skipped"] def test_webhook_incoming_pr_comment_without_directive_skips_without_counting_budget( @@ -955,7 +955,7 @@ def test_webhook_incoming_pr_comment_without_directive_skips_without_counting_bu assert unmapped is not None assert unmapped.issue_key == "octo/widget#900" assert "incoming PR comments ignored" in (unmapped.last_error or "") - assert states == ["queued", "queued", "skipped"] + assert states == ["queued", "queued", "deferred"] def test_webhook_delivery_populates_issue_index(settings: Settings) -> None: @@ -1015,7 +1015,7 @@ def test_webhook_contributor_gets_higher_cap(rate_limited_settings: Settings) -> number=299, association="CONTRIBUTOR", ) - assert resp.json()["state"] == "skipped" + assert resp.json()["state"] == "deferred" close_database() @@ -1066,7 +1066,7 @@ def test_webhook_rate_limit_per_user_is_independent(rate_limited_settings: Setti ).json()["state"] == "queued" ) - # alice's next attempt is skipped. + # alice's next attempt is deferred. assert ( _post_issue_opened( client, @@ -1075,7 +1075,7 @@ def test_webhook_rate_limit_per_user_is_independent(rate_limited_settings: Setti number=599, association="NONE", ).json()["state"] - == "skipped" + == "deferred" ) # bob is untouched. for i in range(2): @@ -1105,13 +1105,13 @@ def test_webhook_rate_limited_event_records_reason(rate_limited_settings: Settin association="NONE", ) db = get_database(rate_limited_settings.sqlite_path) - skipped = db.get_event("r-2") + deferred = db.get_event("r-2") close_database() - assert skipped is not None - assert skipped.state == "skipped" - assert skipped.last_error is not None - assert "rate limit" in skipped.last_error - assert "@charlie" in skipped.last_error + assert deferred is not None + assert deferred.state == "deferred" + assert deferred.last_error is not None + assert "rate limit" in deferred.last_error + assert "@charlie" in deferred.last_error # ---------- /api/github/issues ---------- diff --git a/python/robomp/web/src/components/shell/Vitals.tsx b/python/robomp/web/src/components/shell/Vitals.tsx index a47e71019..1052b9dc1 100644 --- a/python/robomp/web/src/components/shell/Vitals.tsx +++ b/python/robomp/web/src/components/shell/Vitals.tsx @@ -13,6 +13,7 @@ function relativeAgo(ms: number): string { const STATE_TONE: Record = { queued: "var(--color-info)", + deferred: "var(--color-ink-300)", running: "var(--color-warn)", done: "var(--color-ok)", failed: "var(--color-err)", @@ -25,7 +26,7 @@ export function Vitals(): JSX.Element { // issue_event_counts ?? event_counts — preserves the Stats.tsx accessor + title. const counts = (): Record => { const status = statusResource(); - if (!status) return { queued: 0, running: 0, done: 0, failed: 0, skipped: 0 }; + if (!status) return { queued: 0, deferred: 0, running: 0, done: 0, failed: 0, skipped: 0 }; return status.issue_event_counts ?? status.event_counts; }; diff --git a/python/robomp/web/src/types.ts b/python/robomp/web/src/types.ts index b0ffb4b97..35ff081b0 100644 --- a/python/robomp/web/src/types.ts +++ b/python/robomp/web/src/types.ts @@ -2,7 +2,7 @@ // purpose: anything `unknown` here is something the backend explicitly does // not promise to keep stable. -export type EventState = "queued" | "running" | "done" | "failed" | "skipped"; +export type EventState = "queued" | "deferred" | "running" | "done" | "failed" | "skipped"; export type IssueState = | "new" @@ -157,6 +157,7 @@ export const LEVEL_ORDER: Readonly> = { export const EVENT_STATE_ORDER: readonly EventState[] = [ "queued", + "deferred", "running", "done", "failed", diff --git a/python/robomp/web/src/work-items.test.ts b/python/robomp/web/src/work-items.test.ts index 4557449ca..8c74ce6ab 100644 --- a/python/robomp/web/src/work-items.test.ts +++ b/python/robomp/web/src/work-items.test.ts @@ -23,6 +23,7 @@ const BASE_RUNTIME: RuntimeInfo = { function eventCounts(): Record { return { queued: 0, + deferred: 0, running: 0, done: 0, failed: 0, diff --git a/python/robomp/web/test/fixtures/status-contract.json b/python/robomp/web/test/fixtures/status-contract.json index 797dfbaf4..7ef8cc38c 100644 --- a/python/robomp/web/test/fixtures/status-contract.json +++ b/python/robomp/web/test/fixtures/status-contract.json @@ -12,6 +12,7 @@ "inflight": [], "event_counts": { "queued": 1, + "deferred": 0, "running": 1, "done": 2, "failed": 3, @@ -19,6 +20,7 @@ }, "issue_event_counts": { "queued": 1, + "deferred": 0, "running": 1, "done": 2, "failed": 1, From 4d494cc5123d4faefda0a5170018d5805e190def Mon Sep 17 00:00:00 2001 From: roboomp Date: Fri, 17 Jul 2026 17:33:23 +0000 Subject: [PATCH 2/2] fix(robomp): promote deferred events on periodic sweep Ran deferred-submission promotion on an independent timer in addition to the empty-queue path, so a sustained ordinary queue can no longer starve a rate-limited submitter after their rolling window frees. Added ROBOMP_DEFERRED_PROMOTION_SCAN_SECONDS to tune or disable the sweep. Fixes #5882 --- python/robomp/.env.example | 3 ++ python/robomp/src/config.py | 5 +++ python/robomp/src/queue.py | 44 ++++++++++++++++++++-- python/robomp/tests/test_queue_dispatch.py | 36 ++++++++++++++++++ 4 files changed, 84 insertions(+), 4 deletions(-) diff --git a/python/robomp/.env.example b/python/robomp/.env.example index 320bec941..a2a85b723 100644 --- a/python/robomp/.env.example +++ b/python/robomp/.env.example @@ -150,6 +150,9 @@ ROBOMP_RATE_LIMIT_WINDOW_SECONDS=3600 ROBOMP_RATE_LIMIT_DEFAULT=3 ROBOMP_RATE_LIMIT_CONTRIBUTOR=10 ROBOMP_RATE_LIMIT_UNLIMITED= +# How often (seconds) deferred events are swept back into the queue as windows +# free up, independent of queue depth. Set to 0 to disable the periodic sweep. +ROBOMP_DEFERRED_PROMOTION_SCAN_SECONDS=60 # ============================================================================= diff --git a/python/robomp/src/config.py b/python/robomp/src/config.py index 046e7c656..4040c4b4f 100644 --- a/python/robomp/src/config.py +++ b/python/robomp/src/config.py @@ -129,6 +129,11 @@ class Settings(BaseSettings): rate_limit_default: int = Field(3, alias="ROBOMP_RATE_LIMIT_DEFAULT") rate_limit_contributor: int = Field(10, alias="ROBOMP_RATE_LIMIT_CONTRIBUTOR") rate_limit_unlimited_raw: str = Field("", alias="ROBOMP_RATE_LIMIT_UNLIMITED") + # How often the dispatcher sweeps deferred rate-limited events back into the + # queue as their submitters' rolling windows free up. The empty-queue path + # promotes immediately; this periodic sweep guarantees progress even while a + # sustained ordinary queue keeps `claim_next_event` returning work. + deferred_promotion_scan_seconds: float = Field(60.0, alias="ROBOMP_DEFERRED_PROMOTION_SCAN_SECONDS") # Logins (comma-separated, `@` prefix optional, case-insensitive) whose `@bot_login` # mentions are treated as authoritative directives. These accounts also # bypass rate limiting regardless of `author_association`. diff --git a/python/robomp/src/queue.py b/python/robomp/src/queue.py index 37e24dcbe..d9d69bfd5 100644 --- a/python/robomp/src/queue.py +++ b/python/robomp/src/queue.py @@ -100,6 +100,11 @@ class WorkerPool: # restarted orchestrator doesn't burn CPU on a cold cache. if self.sandbox.natives_cache is not None and self.settings.natives_cache_gc_interval_seconds > 0: self._workers.append(asyncio.create_task(self._natives_cache_gc_loop(), name="robomp-natives-gc")) + # Periodic promotion of deferred rate-limited events. Runs independently + # of the empty-queue path so a sustained ordinary queue can't starve a + # submitter whose rolling window has since freed. + if self.settings.deferred_promotion_scan_seconds > 0: + self._workers.append(asyncio.create_task(self._deferred_promotion_loop(), name="robomp-deferred-promotion")) async def stop(self, *, drain_timeout: float = 25.0, kill_timeout: float = 5.0) -> None: """Halt the dispatcher, then drain (or kill) in-flight `_run_event` tasks. @@ -186,6 +191,40 @@ class WorkerPool: except asyncio.CancelledError: raise + async def _deferred_promotion_loop(self) -> None: + """Sweep deferred rate-limited events back into the queue on a timer. + + Sleeps the configured interval first, then re-admits any deferred event + whose submitter now has rolling-window capacity and wakes the dispatcher + so promoted rows are claimed promptly. Cancellation is the only exit; + any per-sweep failure is logged and the loop continues. + """ + interval = self.settings.deferred_promotion_scan_seconds + log.info("deferred promotion loop online", extra={"interval": interval}) + try: + while not self._stop.is_set(): + try: + await asyncio.wait_for(self._stop.wait(), timeout=interval) + return # stop was set during the wait + except TimeoutError: + pass + try: + promoted = await self._promote_deferred() + if promoted: + self.wake() + except Exception: + log.exception("deferred promotion sweep raised") + except asyncio.CancelledError: + raise + + async def _promote_deferred(self) -> int: + """Re-admit deferred events whose submitters regained window capacity.""" + since = iso_seconds_ago(self.settings.rate_limit_window_seconds) + promoted = await asyncio.to_thread(self.db.promote_deferred_submissions, since=since) + if promoted: + log.info("deferred submissions promoted", extra={"count": promoted}) + return promoted + async def _dispatch_loop(self) -> None: log.info("dispatch loop online") try: @@ -214,11 +253,8 @@ class WorkerPool: # Naive but fine for v1 (small queue). row = await asyncio.to_thread(self.db.claim_next_event) if row is None: - since = iso_seconds_ago(self.settings.rate_limit_window_seconds) - promoted = await asyncio.to_thread(self.db.promote_deferred_submissions, since=since) - if not promoted: + if not await self._promote_deferred(): return None - log.info("deferred submissions promoted", extra={"count": promoted}) row = await asyncio.to_thread(self.db.claim_next_event) if row is None: return None diff --git a/python/robomp/tests/test_queue_dispatch.py b/python/robomp/tests/test_queue_dispatch.py index 12517b95d..1034d8e62 100644 --- a/python/robomp/tests/test_queue_dispatch.py +++ b/python/robomp/tests/test_queue_dispatch.py @@ -114,3 +114,39 @@ async def test_claim_promotes_deferred_submission_after_window_frees(settings: S assert row is not None assert row.delivery_id == "deferred" assert row.state == "running" + + +@pytest.mark.asyncio +async def test_deferred_promotion_sweep_runs_while_queue_is_busy(settings: Settings, db: Database) -> None: + """A sustained ordinary queue must not starve deferred events forever. + + The empty-queue path never fires when `claim_next_event` keeps returning + work, so the independent sweep is the only thing that re-admits a deferred + submission after its rolling window frees. + """ + # Ordinary queued work for another issue keeps `claim_next_event` busy. + assert db.record_event( + delivery_id="busy", + event_type="issues", + repo="octo/widget", + issue_key="octo/widget#1", + payload={"action": "opened"}, + ) + assert db.record_submission(delivery_id="accepted", login="alice", repo="octo/widget") + assert db.defer_submission_event( + delivery_id="deferred", + event_type="issues", + login="alice", + repo="octo/widget", + issue_key="octo/widget#8", + payload={"action": "opened", "issue": {"number": 8}}, + cap=1, + reason="rate limit", + ) + settings.rate_limit_window_seconds = -1 + + pool = _make_pool(settings, db) + promoted = await pool._promote_deferred() # noqa: SLF001 + + assert promoted == 1 + assert db.get_event("deferred").state == "queued"