From 6ccb29ec760e27526b0ce543bb0be059fa73d62c Mon Sep 17 00:00:00 2001 From: Suhaib Mujahid Date: Tue, 15 Sep 2026 23:57:52 -0400 Subject: [PATCH 1/6] Add run dedupe key and idempotent create flow --- docs/hackbot/api.md | 71 ++++++++--- libs/hackbot-client/hackbot_client/client.py | 10 +- .../versions/a7d4e9c21b83_runs_dedupe_key.py | 40 ++++++ services/hackbot-api/app/database/models.py | 12 +- services/hackbot-api/app/routers/runs.py | 87 ++++++++++--- services/hackbot-api/app/schemas.py | 3 + services/hackbot-api/tests/conftest.py | 119 ++++++++++++++++++ .../hackbot-api/tests/test_create_run_api.py | 100 ++++++++++++++- .../hackbot-api/tests/test_finalize_run.py | 8 +- services/hackbot-ui/lib/types.ts | 1 + 10 files changed, 414 insertions(+), 37 deletions(-) create mode 100644 services/hackbot-api/alembic/versions/a7d4e9c21b83_runs_dedupe_key.py diff --git a/docs/hackbot/api.md b/docs/hackbot/api.md index 2a2244d1e7..fa84362a3a 100644 --- a/docs/hackbot/api.md +++ b/docs/hackbot/api.md @@ -7,20 +7,21 @@ component that knows the agent catalog, and the only writer of run state. ### Public — `X-API-Key` -| Method | Path | Does | -| ------ | --------------------------------- | ------------------------------------------------------ | -| GET | `/agents` | The catalog, each with its input JSON schema | -| POST | `/agents/{agent}/runs` | Validate inputs, create a run, start an execution | -| GET | `/runs` | List runs; filter by `agent`, `status`, `requested_by` | -| GET | `/runs/{run_id}` | One run: status, inputs, summary, artifacts | -| GET | `/runs/{run_id}/artifacts/{path}` | A short-lived signed GCS download URL | -| GET | `/runs/{run_id}/actions` | Recorded actions and their apply state | -| POST | `/runs/{run_id}/actions/apply` | Apply all pending actions (idempotent) | -| GET | `/health` | Health check | +| Method | Path | Does | +| ------ | --------------------------------- | -------------------------------------------------------------------- | +| GET | `/agents` | The catalog, each with its input JSON schema | +| POST | `/agents/{agent}/runs` | Validate inputs, create a run, start an execution | +| GET | `/runs` | List runs; filter by `agent`, `status`, `requested_by`, `dedupe_key` | +| GET | `/runs/{run_id}` | One run: status, inputs, summary, artifacts | +| GET | `/runs/{run_id}/artifacts/{path}` | A short-lived signed GCS download URL | +| GET | `/runs/{run_id}/actions` | Recorded actions and their apply state | +| POST | `/runs/{run_id}/actions/apply` | Apply all pending actions (idempotent) | +| GET | `/health` | Health check | `POST /agents/{agent}/runs` accepts an `X-On-Behalf-Of` header carrying the requesting user's email, stored as `requested_by` — the caller is a trusted service (the UI), so this -is attribution, not authentication. +is attribution, not authentication. It also accepts `X-Dedupe-Key`, which decides +whether the request starts a run at all (see [Deduplication](#deduplication)). Artifact downloads are restricted to artifacts already listed on the run, which both scopes the download to that run's prefix and prevents probing unrelated objects. @@ -68,15 +69,49 @@ To turn it on: **Interactivity & Shortcuts → Request URL** = ``` validate payload against the agent's input schema ── 422 on mismatch +insert Run(status=pending) and commit ── the key is now claimed + ...unless the unique index rejects it ── 200 + the run that won mint a V4 signed POST policy scoped to runs// -insert Run(status=pending) trigger a Cloud Run Job execution with env overrides on the `agent` container store execution_name ``` Env overrides are the run id, the results bucket/prefix/policy, and the inputs mapped from -the schema. If the trigger fails the run is marked `failed` with the reason and the caller -gets a 502 — a run row always exists, so a failed dispatch is visible rather than lost. +the schema. If the policy or the trigger fails the run is marked `failed` with the reason and +the caller gets a 502 — a run row always exists, so a failed start is visible rather than +lost. + +The insert comes **first**, so a duplicate is turned away before anything is prepared for +it, and it **commits** before the dispatch, so no transaction is held open across a call to +Cloud Run (the pool is 5 plus 5 overflow) and the row survives a crash mid-dispatch. Neither +is what makes deduplication correct, the unique index is. The cost is that a crash in between +leaves a pending run with no execution, which the stale-run sweep finalizes. + +## Deduplication + +External triggers fan out: 20 build tasks fail on one push, Phabricator retries a delivery, +a Slack button gets clicked twice. A caller **names the work** with `X-Dedupe-Key`, scoped to +the agent, and the rule is the whole design: + +> a key names one run, for good. + +A new run answers `201`. A request whose name another run already holds answers `200` with +that run, which is not an error and needs no handling; `GET /runs?dedupe_key=…` finds it +again later. The name is coalescing rather than a request fingerprint: requests sharing one +may carry different inputs, and the first to arrive is the one that runs. + +Recurrence lives in the key, since only the caller knows whether the work may happen again: + +| Intent | Key | +| ---------------------------------- | ---------------------------------------- | +| Investigate this push exactly once | `push::` | +| ...but let tomorrow try again | `push:::` | +| Handle this delivery exactly once | `phab-txn:`, `ni:` | +| Always run | (omit the header) | + +A run keeps its name whatever becomes of it, a failed dispatch included, so a repeated +trigger gets the failure rather than a silent retry. A transient error therefore spends the +key: name the work differently, or run unkeyed. ## Run states @@ -144,6 +179,14 @@ Two tables, defined in [app/database/models.py](../../services/hackbot-api/app/d **`runs`** is the system of record for a run — its inputs, execution name, summary, artifacts and terminal state. Listing orders by `created_at desc, run_id desc` rather than timestamp alone, so offset paging stays stable when two runs share a timestamp. +`dedupe_key` carries one index, `uq_runs_dedupe_key` over `(dedupe_key, agent)`, **unique**: + +- **Unique**, because a key names one run for good, so the index is the decision about a + duplicate trigger rather than a check on one. Runs without a key are unconstrained, since + Postgres treats each NULL as distinct. +- **`dedupe_key` first**, so the same index answers `GET /runs?dedupe_key=…`, which carries + no agent. The reverse order would constrain exactly the same thing but only serve lookups + that name the agent. **`run_actions`** holds one row per entry in a run's `summary.json` actions, unique on `(run_id, idx)`, carrying that action's apply state. That uniqueness is what makes replays diff --git a/libs/hackbot-client/hackbot_client/client.py b/libs/hackbot-client/hackbot_client/client.py index e9db069f88..9bb04cf4e6 100644 --- a/libs/hackbot-client/hackbot_client/client.py +++ b/libs/hackbot-client/hackbot_client/client.py @@ -26,11 +26,19 @@ async def trigger_run( inputs: Mapping[str, Any], *, on_behalf_of: str | None = None, + dedupe_key: str | None = None, ) -> RunRef: - """Create an agent run and return the API's typed run reference.""" + """Create an agent run and return the API's typed run reference. + + `dedupe_key` names the work the run does, and the name belongs to one + run for good: repeated triggers for the key will be no-ops, and the same + run reference will be returned. + """ headers = {"X-API-Key": self._api_key} if on_behalf_of is not None: headers["X-On-Behalf-Of"] = on_behalf_of + if dedupe_key is not None: + headers["X-Dedupe-Key"] = dedupe_key async with httpx.AsyncClient(timeout=self._timeout_seconds) as client: response = await client.post( diff --git a/services/hackbot-api/alembic/versions/a7d4e9c21b83_runs_dedupe_key.py b/services/hackbot-api/alembic/versions/a7d4e9c21b83_runs_dedupe_key.py new file mode 100644 index 0000000000..9a148a106a --- /dev/null +++ b/services/hackbot-api/alembic/versions/a7d4e9c21b83_runs_dedupe_key.py @@ -0,0 +1,40 @@ +"""Name the work a run does, so duplicate triggers can collapse onto it. + +Revision ID: a7d4e9c21b83 +Revises: f3c8a1d5b2e7 +Create Date: 2026-09-08 00:00:00.000000 + +""" + +from typing import Sequence, Union + +import sqlalchemy as sa +from alembic import op + +# revision identifiers, used by Alembic. +revision: str = "a7d4e9c21b83" +down_revision: Union[str, Sequence[str], None] = "f3c8a1d5b2e7" +branch_labels: Union[str, Sequence[str], None] = None +depends_on: Union[str, Sequence[str], None] = None + + +def upgrade() -> None: + """Upgrade schema.""" + op.add_column("runs", sa.Column("dedupe_key", sa.String(), nullable=True)) + # Unique, because a key names one run for good: the constraint is what + # decides a duplicate trigger, not just a check on one. `dedupe_key` leads + # so the index also serves the `?dedupe_key=` listing filter, which carries + # no agent. Existing rows all have a NULL key, and NULLs do not collide, so + # this is safe to add to a populated table. + op.create_index( + "uq_runs_dedupe_key", + "runs", + ["dedupe_key", "agent"], + unique=True, + ) + + +def downgrade() -> None: + """Downgrade schema.""" + op.drop_index("uq_runs_dedupe_key", table_name="runs") + op.drop_column("runs", "dedupe_key") diff --git a/services/hackbot-api/app/database/models.py b/services/hackbot-api/app/database/models.py index 6dd1b32dba..3c0252bbca 100644 --- a/services/hackbot-api/app/database/models.py +++ b/services/hackbot-api/app/database/models.py @@ -1,7 +1,15 @@ from datetime import datetime from uuid import UUID -from sqlalchemy import DateTime, ForeignKey, Integer, String, Text, UniqueConstraint +from sqlalchemy import ( + DateTime, + ForeignKey, + Index, + Integer, + String, + Text, + UniqueConstraint, +) from sqlalchemy.dialects.postgresql import JSONB from sqlalchemy.dialects.postgresql import UUID as PG_UUID from sqlalchemy.ext.asyncio import AsyncAttrs @@ -15,12 +23,14 @@ class Base(AsyncAttrs, DeclarativeBase): class Run(Base): __tablename__ = "runs" + __table_args__ = (Index("uq_runs_dedupe_key", "dedupe_key", "agent", unique=True),) run_id: Mapped[UUID] = mapped_column(PG_UUID(as_uuid=True), primary_key=True) agent: Mapped[str] = mapped_column(String, nullable=False, index=True) status: Mapped[str] = mapped_column(String, nullable=False, index=True) inputs: Mapped[dict] = mapped_column(JSONB, nullable=False) requested_by: Mapped[str | None] = mapped_column(String, nullable=True, index=True) + dedupe_key: Mapped[str | None] = mapped_column(String, nullable=True) created_at: Mapped[datetime] = mapped_column( DateTime(timezone=True), server_default=func.now(), diff --git a/services/hackbot-api/app/routers/runs.py b/services/hackbot-api/app/routers/runs.py index 7618f35ccf..fb4ec3b134 100644 --- a/services/hackbot-api/app/routers/runs.py +++ b/services/hackbot-api/app/routers/runs.py @@ -5,8 +5,10 @@ from typing import Annotated from fastapi import APIRouter, Depends, Header, HTTPException, Query, status -from pydantic import BeforeValidator +from fastapi.responses import Response +from pydantic import BeforeValidator, StringConstraints from sqlalchemy import select +from sqlalchemy.exc import IntegrityError from sqlalchemy.ext.asyncio import AsyncSession from app import gcs, jobs, pubsub @@ -44,6 +46,20 @@ def _normalize_identity(email: str | None) -> str | None: UserEmail = Annotated[str | None, BeforeValidator(_normalize_identity)] +DedupeKey = ( + Annotated[ + str, + StringConstraints( + strip_whitespace=True, + min_length=1, + max_length=200, + to_lower=True, + ascii_only=True, + ), + ] + | None +) + def _lookup_agent(name: str) -> AgentSpec: agent = AGENT_REGISTRY.get(name) @@ -78,8 +94,19 @@ async def create_run( description="Email of the user this run is requested for.", ), ] = None, + dedupe_key: Annotated[ + DedupeKey, + Header( + alias="X-Dedupe-Key", + description=( + "A key to deduplicate runs for the same work. If a run already " + "exists for this key, it will be returned instead of creating " + "a new one." + ), + ), + ] = None, db: AsyncSession = Depends(get_db), -) -> RunRef: +) -> RunRef | Response: agent = _lookup_agent(agent_name) try: inputs = agent.input_schema.model_validate(payload) @@ -89,33 +116,57 @@ async def create_run( run_id = uuid.uuid4() results_prefix = gcs.run_prefix(str(run_id)) - policy = await gcs.generate_results_policy(str(run_id)) - run = Run( run_id=run_id, agent=agent.name, status=RunStatus.pending.value, inputs=inputs.model_dump(mode="json"), requested_by=on_behalf_of, + dedupe_key=dedupe_key, results_prefix=results_prefix, artifacts=[], ) db.add(run) - await db.flush() - env_overrides: dict[str, str] = { - "RUN_ID": str(run_id), - "RESULTS_BUCKET": settings.results_bucket, - "RESULTS_PREFIX": results_prefix, - "RESULTS_POLICY_URL": policy["url"], - "RESULTS_POLICY_FIELDS": json.dumps(policy["fields"]), - **(agent.build_env or model_to_env)(inputs), - } + try: + await db.commit() + except IntegrityError: + if dedupe_key is None: + # No key means no name to lose, so this is a real constraint fault. + raise + + await db.rollback() + result = await db.execute( + select(Run).where(Run.agent == agent.name, Run.dedupe_key == dedupe_key) + ) + run = result.scalar_one() + log.info( + "Deduplicated %s request for key %r onto run %s (%s)", + run.agent, + run.dedupe_key, + run.run_id, + run.status, + ) + + return Response( + content=RunRef.model_validate(run).model_dump_json(), + media_type="application/json", + status_code=status.HTTP_200_OK, + ) try: + policy = await gcs.generate_results_policy(str(run_id)) + env_overrides: dict[str, str] = { + "RUN_ID": str(run_id), + "RESULTS_BUCKET": settings.results_bucket, + "RESULTS_PREFIX": results_prefix, + "RESULTS_POLICY_URL": policy["url"], + "RESULTS_POLICY_FIELDS": json.dumps(policy["fields"]), + **(agent.build_env or model_to_env)(inputs), + } execution_name = await jobs.trigger_execution(agent.job_name, env_overrides) except Exception as exc: - log.exception("Failed to trigger Cloud Run Job for run %s", run_id) + log.exception("Failed to start execution for run %s", run_id) run.status = RunStatus.failed.value run.error = f"Failed to start execution: {exc}" await db.commit() @@ -141,6 +192,10 @@ async def list_runs( UserEmail, Query(description="Only return runs requested by this user."), ] = None, + dedupe_key: Annotated[ + DedupeKey, + Query(description="Only return runs carrying this dedupe key."), + ] = None, db: AsyncSession = Depends(get_db), ) -> list[RunDoc]: stmt = select(Run) @@ -150,6 +205,10 @@ async def list_runs( stmt = stmt.where(Run.status == status_filter.value) if requested_by is not None: stmt = stmt.where(Run.requested_by == requested_by) + if dedupe_key is not None: + # The trigger's own handle on its run: a caller that named the work can + # find it again without having stored the run id. + stmt = stmt.where(Run.dedupe_key == dedupe_key) # created_at is the sort key; run_id is a deterministic tiebreaker so offset # paging is stable when timestamps collide. (agent/status/requested_by and # created_at are all indexed, so filtering + ordering stay index-backed.) diff --git a/services/hackbot-api/app/schemas.py b/services/hackbot-api/app/schemas.py index 93236fecb6..aa466dfe8e 100644 --- a/services/hackbot-api/app/schemas.py +++ b/services/hackbot-api/app/schemas.py @@ -49,6 +49,8 @@ class AgentDescriptor(BaseModel): class RunRef(BaseModel): + """The API's answer to "start this run": which run is doing the work.""" + model_config = ConfigDict(from_attributes=True) run_id: UUID @@ -64,6 +66,7 @@ class RunDoc(BaseModel): status: RunStatus inputs: dict[str, Any] requested_by: str | None = None + dedupe_key: str | None = None created_at: datetime updated_at: datetime execution_name: str | None = None diff --git a/services/hackbot-api/tests/conftest.py b/services/hackbot-api/tests/conftest.py index 500a1ff9af..efc3eb1497 100644 --- a/services/hackbot-api/tests/conftest.py +++ b/services/hackbot-api/tests/conftest.py @@ -12,3 +12,122 @@ os.environ.setdefault("BUGZILLA_WEBHOOK_BOT_LOGIN", "hackbot@mozilla.tld") os.environ.setdefault("BUGZILLA_API_KEY", "test-bugzilla-api-key") os.environ.setdefault("SLACK_SIGNING_SECRET", "test-signing-secret") + +import pytest # noqa: E402 +from app.auth import require_api_key # noqa: E402 +from app.database.connection import get_db # noqa: E402 +from app.database.models import Run # noqa: E402 +from app.main import app # noqa: E402 +from fastapi.testclient import TestClient # noqa: E402 +from sqlalchemy.exc import ( # noqa: E402 + MultipleResultsFound, + NoResultFound, + PendingRollbackError, +) + + +class _Scalars: + def __init__(self, rows): + self._rows = rows + + def first(self): + return self._rows[0] if self._rows else None + + def __iter__(self): + return iter(self._rows) + + +class _Result: + def __init__(self, rows): + self._rows = rows + + def scalars(self): + return _Scalars(self._rows) + + def scalar_one(self): + if not self._rows: + raise NoResultFound("No row was found when one was required") + if len(self._rows) > 1: + raise MultipleResultsFound("Multiple rows were found when one was required") + return self._rows[0] + + +class FakeSession: + """An AsyncSession stand-in that records what a handler did to it. + + Statements are kept so a test can assert on the SQL a handler built, and + `matches` is what any query against `runs` returns -- which is how a test + sets up "a run already holds this dedupe key" without a Postgres. Set + `on_commit` to a callable to interpose on the next commit, which is how a + test provokes the unique-index conflict the dedupe path is built around. + + A failed commit poisons the session until it is rolled back, as a real one + does, so a handler that reads after a rejected insert without rolling back + fails here rather than only against Postgres. + """ + + def __init__(self, matches: list | None = None): + self.matches = matches or [] + self.added = None + self.statements: list = [] + self.commits = 0 + self.rollbacks = 0 + self.on_commit = None + self.needs_rollback = False + + @property + def stmt(self): + """The last statement executed, for tests that assert on one query.""" + return self.statements[-1] if self.statements else None + + async def execute(self, stmt): + if self.needs_rollback: + raise PendingRollbackError( + "This Session's transaction has been rolled back due to a " + "previous exception during flush." + ) + + self.statements.append(stmt) + # Only queries over `runs` return rows; anything else a handler runs + # for effect gets an empty result. + if Run.__table__ in stmt.get_final_froms(): + return _Result(self.matches) + return _Result([]) + + def add(self, obj): + self.added = obj + + async def commit(self): + if self.on_commit is not None: + # One-shot, so a test can fail the claiming commit and still let the + # handler's later commits through. + hook, self.on_commit = self.on_commit, None + try: + hook() + except Exception: + self.needs_rollback = True + raise + self.commits += 1 + + async def rollback(self): + self.needs_rollback = False + self.rollbacks += 1 + + async def get(self, model, key): + return None + + +@pytest.fixture +def db(): + return FakeSession() + + +@pytest.fixture +def client(db): + """A TestClient wired to the fake session, with API-key auth stubbed out.""" + app.dependency_overrides[get_db] = lambda: db + app.dependency_overrides[require_api_key] = lambda: None + try: + yield TestClient(app) + finally: + app.dependency_overrides.clear() diff --git a/services/hackbot-api/tests/test_create_run_api.py b/services/hackbot-api/tests/test_create_run_api.py index d0084b73e8..a33cb3abbc 100644 --- a/services/hackbot-api/tests/test_create_run_api.py +++ b/services/hackbot-api/tests/test_create_run_api.py @@ -1,14 +1,20 @@ -"""Tests for POST /agents/{agent_name}/runs, focused on requester attribution. +"""Tests for POST /agents/{agent_name}/runs: requester attribution and dedupe. These go through a TestClient rather than calling the handler directly: the -`X-On-Behalf-Of` normalization lives in the parameter's annotation (see -`UserEmail` in app/routers/runs.py), so it only runs as part of FastAPI's request -handling. The GCS/Cloud Run collaborators are monkeypatched and the DB session is -the shared `FakeSession`, so no GCP or Postgres is needed. +`X-On-Behalf-Of` normalization and the dedupe key live in the parameters' +annotations (`UserEmail` and `DedupeKey` in app/routers/runs.py), so they only +run as part of FastAPI's request handling. The +GCS/Cloud Run collaborators are monkeypatched and the DB session is the shared +`FakeSession`, so no GCP or Postgres is needed. """ +import uuid +from types import SimpleNamespace + import pytest from app import gcs, jobs +from app.schemas import RunStatus +from sqlalchemy.exc import IntegrityError @pytest.fixture(autouse=True) @@ -49,3 +55,87 @@ def test_create_run_leaves_run_unattributed_without_header(client, db, headers): # blank one must not land as an empty-string requester either. _create(client, headers) assert db.added.requested_by is None + + +# --- deduplication --- + + +def _holding_run(): + """A run already holding the key, as the dedupe lookup would return it.""" + return SimpleNamespace( + run_id=uuid.uuid4(), + agent="bug-fix", + status=RunStatus.running.value, + dedupe_key="push:autoland:abc", + ) + + +def _create_keyed(client, key="push:autoland:abc"): + return client.post( + "/agents/bug-fix/runs", + json={"bug_id": 1889001}, + headers={"X-Dedupe-Key": key}, + ) + + +def _lose_the_key(db, winner): + """Make the next insert lose the name to `winner`, as the index would. + + `uq_runs_dedupe_key` rejects the insert and the run that won the name is + there to be read afterwards, which is the only way a request collapses. + """ + + def lose(): + db.matches = [winner] + raise IntegrityError("INSERT INTO runs", {}, Exception("uq_runs_dedupe_key")) + + db.on_commit = lose + + +def test_keyed_request_records_the_key_on_the_new_run(client, db): + # Stripped, so the same name spelled with stray whitespace is the same name. + _create(client, {"X-Dedupe-Key": " push:autoland:abc "}) + assert db.added.dedupe_key == "push:autoland:abc" + + +def test_keyed_request_is_answered_with_the_run_holding_the_key(client, db): + winner = _holding_run() + _lose_the_key(db, winner) + + resp = _create_keyed(client) + + # 200, not 201: this request created nothing. + assert resp.status_code == 200, resp.text + assert resp.json()["run_id"] == str(winner.run_id) + # Ours was given back rather than left in a broken transaction. + assert db.rollbacks == 1 + + +@pytest.mark.parametrize( + ("module", "attr"), + [(gcs, "generate_results_policy"), (jobs, "trigger_execution")], +) +def test_a_run_that_cannot_start_records_why_and_keeps_its_key( + client, db, monkeypatch, module, attr +): + # Both happen after the claim, so both are one failure to a caller: the run + # exists, says why, and holds its name still, so a repeated trigger gets the + # failure rather than quietly starting the work again. + async def fail(*_a, **_k): + raise RuntimeError("no quota") + + monkeypatch.setattr(module, attr, fail) + + assert _create_keyed(client).status_code == 502 + assert db.added.status == RunStatus.failed.value + assert "no quota" in db.added.error + assert db.added.dedupe_key == "push:autoland:abc" + + +@pytest.mark.parametrize("raw", ["", " ", "x" * 500]) +def test_an_unusable_dedupe_key_is_rejected(client, db, raw): + # Blank is not taken as "no key": a caller that built a name out of a value + # it did not have is told, rather than every such caller sharing the empty + # name. + assert _create_keyed(client, key=raw).status_code == 422 + assert db.added is None diff --git a/services/hackbot-api/tests/test_finalize_run.py b/services/hackbot-api/tests/test_finalize_run.py index 353981ee96..0ed627902e 100644 --- a/services/hackbot-api/tests/test_finalize_run.py +++ b/services/hackbot-api/tests/test_finalize_run.py @@ -230,10 +230,14 @@ async def test_run_without_execution_name_is_failed_not_asserted(monkeypatch): run = _FakeRun(execution_name=None) db = _FakeDB() - async def fail(*_a, **_k): + def fail(*_a, **_k): raise AssertionError("should not check status without an execution name") - monkeypatch.setattr(jobs, "get_execution_status", fail) + # The no-execution case is answered by `get_execution_status` itself, so it + # is the call to Cloud Run underneath that must not happen. + monkeypatch.setattr(jobs, "_execution_status_sync", fail) + monkeypatch.setattr(gcs, "read_summary", _async(None)) + monkeypatch.setattr(gcs, "list_artifacts", _async([])) await finalize_run(db, run) diff --git a/services/hackbot-ui/lib/types.ts b/services/hackbot-ui/lib/types.ts index 1f586f116e..5c9b74c66d 100644 --- a/services/hackbot-ui/lib/types.ts +++ b/services/hackbot-ui/lib/types.ts @@ -67,6 +67,7 @@ export interface RunDoc { status: RunStatus; inputs: Record; requested_by: string | null; + dedupe_key: string | null; created_at: string; updated_at: string; execution_name: string | null; From 748ddf66ad221e7a34cb0bbc9d77d2f19e757ad7 Mon Sep 17 00:00:00 2001 From: Suhaib Mujahid Date: Wed, 16 Sep 2026 10:36:34 -0400 Subject: [PATCH 2/6] Bump Hackbot API to Pydantic 2.13 --- services/hackbot-api/pyproject.toml | 4 ++-- uv.lock | 20 ++++++++++---------- 2 files changed, 12 insertions(+), 12 deletions(-) diff --git a/services/hackbot-api/pyproject.toml b/services/hackbot-api/pyproject.toml index e4bfb349bd..b74ae7713d 100644 --- a/services/hackbot-api/pyproject.toml +++ b/services/hackbot-api/pyproject.toml @@ -6,8 +6,8 @@ requires-python = ">=3.12" dependencies = [ "fastapi>=0.109.0", "uvicorn[standard]>=0.27.0", - "pydantic>=2.6.0", - "pydantic-settings>=2.1.0", + "pydantic>=2.13.0", + "pydantic-settings>=2.13.0", "sqlalchemy[asyncio]>=2.0.25", "asyncpg>=0.29.0", "cloud-sql-python-connector[asyncpg]>=1.5.0", diff --git a/uv.lock b/uv.lock index 6380bcd526..b8397e6cd4 100644 --- a/uv.lock +++ b/uv.lock @@ -2760,8 +2760,8 @@ requires-dist = [ { name = "lando-client", editable = "libs/lando-client" }, { name = "markdown2", specifier = ">=2.4.0" }, { name = "phabricator-client", editable = "libs/phabricator-client" }, - { name = "pydantic", specifier = ">=2.6.0" }, - { name = "pydantic-settings", specifier = ">=2.1.0" }, + { name = "pydantic", specifier = ">=2.13.0" }, + { name = "pydantic-settings", specifier = ">=2.13.0" }, { name = "pytest", marker = "extra == 'dev'", specifier = ">=8.0.0" }, { name = "pytest-asyncio", marker = "extra == 'dev'", specifier = ">=0.23.0" }, { name = "python-multipart", specifier = ">=0.0.9" }, @@ -4689,9 +4689,9 @@ resolution-markers = [ "python_full_version < '3.13' and platform_machine == 'x86_64' and sys_platform == 'darwin'", ] dependencies = [ - { name = "llvmlite", version = "0.36.0", source = { registry = "https://pypi.org/simple" } }, - { name = "numpy" }, - { name = "setuptools" }, + { name = "llvmlite", version = "0.36.0", source = { registry = "https://pypi.org/simple" }, marker = "platform_machine == 'x86_64' and sys_platform == 'darwin'" }, + { name = "numpy", marker = "platform_machine == 'x86_64' and sys_platform == 'darwin'" }, + { name = "setuptools", marker = "platform_machine == 'x86_64' and sys_platform == 'darwin'" }, ] sdist = { url = "https://files.pythonhosted.org/packages/e3/7d/3d61160836e49f40913741c464f119551c15ed371c1d91ea50308495b93b/numba-0.53.1.tar.gz", hash = "sha256:9cd4e5216acdc66c4e9dab2dfd22ddb5bef151185c070d4a3cd8e78638aff5b0", size = 2213956, upload-time = "2021-03-26T09:15:50.402Z" } @@ -4711,8 +4711,8 @@ resolution-markers = [ "(python_full_version < '3.13' and platform_machine != 'x86_64' and sys_platform == 'darwin') or (python_full_version < '3.13' and sys_platform != 'darwin' and sys_platform != 'emscripten' and sys_platform != 'win32')", ] dependencies = [ - { name = "llvmlite", version = "0.48.0", source = { registry = "https://pypi.org/simple" } }, - { name = "numpy" }, + { name = "llvmlite", version = "0.48.0", source = { registry = "https://pypi.org/simple" }, marker = "platform_machine != 'x86_64' or sys_platform != 'darwin'" }, + { name = "numpy", marker = "platform_machine != 'x86_64' or sys_platform != 'darwin'" }, ] sdist = { url = "https://files.pythonhosted.org/packages/ae/a0/570e3dc53e5602b49108f62a13e529f1eec8bfc7ef37d49c825924dcf546/numba-0.66.0.tar.gz", hash = "sha256:b900e63a0e26c05ea9a6d5a3a5a0a177cb64c5011887bf43edb8c3ed2c38d363", size = 2806181, upload-time = "2026-07-01T23:12:46.36Z" } wheels = [ @@ -5142,7 +5142,7 @@ name = "pexpect" version = "4.9.0" source = { registry = "https://pypi.org/simple" } dependencies = [ - { name = "ptyprocess" }, + { name = "ptyprocess", marker = "sys_platform != 'emscripten' and sys_platform != 'win32'" }, ] sdist = { url = "https://files.pythonhosted.org/packages/42/92/cc564bf6381ff43ce1f4d06852fc19a2f11d180f23dc32d9588bee2f149d/pexpect-4.9.0.tar.gz", hash = "sha256:ee7d41123f3c9911050ea2c2dac107568dc43b2d3b0c7557a33212c398ead30f", size = 166450, upload-time = "2023-11-25T09:07:26.339Z" } wheels = [ @@ -6776,8 +6776,8 @@ name = "secretstorage" version = "3.5.0" source = { registry = "https://pypi.org/simple" } dependencies = [ - { name = "cryptography" }, - { name = "jeepney" }, + { name = "cryptography", marker = "(platform_machine != 'x86_64' and sys_platform == 'darwin') or (sys_platform != 'darwin' and sys_platform != 'emscripten' and sys_platform != 'win32')" }, + { name = "jeepney", marker = "(platform_machine != 'x86_64' and sys_platform == 'darwin') or (sys_platform != 'darwin' and sys_platform != 'emscripten' and sys_platform != 'win32')" }, ] sdist = { url = "https://files.pythonhosted.org/packages/1c/03/e834bcd866f2f8a49a85eaff47340affa3bfa391ee9912a952a1faa68c7b/secretstorage-3.5.0.tar.gz", hash = "sha256:f04b8e4689cbce351744d5537bf6b1329c6fc68f91fa666f60a380edddcd11be", size = 19884, upload-time = "2025-11-23T19:02:53.191Z" } wheels = [ From 8bef4a48b0216aaf703f95a19795ac693835ae6d Mon Sep 17 00:00:00 2001 From: Suhaib Mujahid Date: Wed, 16 Sep 2026 10:47:15 -0400 Subject: [PATCH 3/6] Document deduped run response for create endpoint --- services/hackbot-api/app/routers/runs.py | 16 +++++++++++++++- 1 file changed, 15 insertions(+), 1 deletion(-) diff --git a/services/hackbot-api/app/routers/runs.py b/services/hackbot-api/app/routers/runs.py index fb4ec3b134..9896841044 100644 --- a/services/hackbot-api/app/routers/runs.py +++ b/services/hackbot-api/app/routers/runs.py @@ -83,7 +83,21 @@ async def list_agents() -> list[AgentDescriptor]: ] -@router.post("/agents/{agent_name}/runs", response_model=RunRef, status_code=201) +@router.post( + "/agents/{agent_name}/runs", + response_model=RunRef, + status_code=201, + response_description="A new run, started by this request.", + responses={ + status.HTTP_200_OK: { + "model": RunRef, + "description": ( + "A run already exists for this agent and dedupe key, so it is " + "returned instead of creating a new one." + ), + }, + }, +) async def create_run( agent_name: str, payload: dict, From 86a79f7db5b7623e8a006815b70ccf6d32b71cee Mon Sep 17 00:00:00 2001 From: Suhaib Mujahid Date: Wed, 16 Sep 2026 12:16:26 -0400 Subject: [PATCH 4/6] Use dedupe_key query param instead of X-Dedupe-Key header --- docs/hackbot/api.md | 6 +++--- libs/hackbot-client/hackbot_client/client.py | 5 +++-- libs/hackbot-client/tests/test_client.py | 6 ++++-- services/hackbot-api/app/routers/runs.py | 3 +-- services/hackbot-api/tests/test_create_run_api.py | 4 ++-- 5 files changed, 13 insertions(+), 11 deletions(-) diff --git a/docs/hackbot/api.md b/docs/hackbot/api.md index fa84362a3a..13ea594ce5 100644 --- a/docs/hackbot/api.md +++ b/docs/hackbot/api.md @@ -20,8 +20,8 @@ component that knows the agent catalog, and the only writer of run state. `POST /agents/{agent}/runs` accepts an `X-On-Behalf-Of` header carrying the requesting user's email, stored as `requested_by` — the caller is a trusted service (the UI), so this -is attribution, not authentication. It also accepts `X-Dedupe-Key`, which decides -whether the request starts a run at all (see [Deduplication](#deduplication)). +is attribution, not authentication. It also takes a `dedupe_key` query parameter, which +decides whether the request starts a run at all (see [Deduplication](#deduplication)). Artifact downloads are restricted to artifacts already listed on the run, which both scopes the download to that run's prefix and prevents probing unrelated objects. @@ -90,7 +90,7 @@ leaves a pending run with no execution, which the stale-run sweep finalizes. ## Deduplication External triggers fan out: 20 build tasks fail on one push, Phabricator retries a delivery, -a Slack button gets clicked twice. A caller **names the work** with `X-Dedupe-Key`, scoped to +a Slack button gets clicked twice. A caller **names the work** with `?dedupe_key=`, scoped to the agent, and the rule is the whole design: > a key names one run, for good. diff --git a/libs/hackbot-client/hackbot_client/client.py b/libs/hackbot-client/hackbot_client/client.py index 9bb04cf4e6..c0df79e9a5 100644 --- a/libs/hackbot-client/hackbot_client/client.py +++ b/libs/hackbot-client/hackbot_client/client.py @@ -37,14 +37,15 @@ async def trigger_run( headers = {"X-API-Key": self._api_key} if on_behalf_of is not None: headers["X-On-Behalf-Of"] = on_behalf_of - if dedupe_key is not None: - headers["X-Dedupe-Key"] = dedupe_key + + params = {} if dedupe_key is None else {"dedupe_key": dedupe_key} async with httpx.AsyncClient(timeout=self._timeout_seconds) as client: response = await client.post( f"{self._base_url}/agents/{agent_name}/runs", json=dict(inputs), headers=headers, + params=params, ) response.raise_for_status() diff --git a/libs/hackbot-client/tests/test_client.py b/libs/hackbot-client/tests/test_client.py index 44d022f525..58d1e132a2 100644 --- a/libs/hackbot-client/tests/test_client.py +++ b/libs/hackbot-client/tests/test_client.py @@ -33,8 +33,8 @@ async def __aenter__(self): async def __aexit__(self, *exc): return False - async def post(self, url, json=None, headers=None): - captured.update(url=url, json=json, headers=headers) + async def post(self, url, json=None, headers=None, params=None): + captured.update(url=url, json=json, headers=headers, params=params) response.request = httpx.Request("POST", url) return response @@ -66,6 +66,8 @@ async def test_trigger_run_posts_inputs_and_returns_typed_reference(monkeypatch) "X-API-Key": "secret", "X-On-Behalf-Of": "user@example.com", }, + # No key means no client-assigned id on the create call. + "params": {}, } diff --git a/services/hackbot-api/app/routers/runs.py b/services/hackbot-api/app/routers/runs.py index 9896841044..bfa8de6cb2 100644 --- a/services/hackbot-api/app/routers/runs.py +++ b/services/hackbot-api/app/routers/runs.py @@ -110,8 +110,7 @@ async def create_run( ] = None, dedupe_key: Annotated[ DedupeKey, - Header( - alias="X-Dedupe-Key", + Query( description=( "A key to deduplicate runs for the same work. If a run already " "exists for this key, it will be returned instead of creating " diff --git a/services/hackbot-api/tests/test_create_run_api.py b/services/hackbot-api/tests/test_create_run_api.py index a33cb3abbc..fe97e38296 100644 --- a/services/hackbot-api/tests/test_create_run_api.py +++ b/services/hackbot-api/tests/test_create_run_api.py @@ -74,7 +74,7 @@ def _create_keyed(client, key="push:autoland:abc"): return client.post( "/agents/bug-fix/runs", json={"bug_id": 1889001}, - headers={"X-Dedupe-Key": key}, + params={"dedupe_key": key}, ) @@ -94,7 +94,7 @@ def lose(): def test_keyed_request_records_the_key_on_the_new_run(client, db): # Stripped, so the same name spelled with stray whitespace is the same name. - _create(client, {"X-Dedupe-Key": " push:autoland:abc "}) + assert _create_keyed(client, key=" push:autoland:abc ").status_code == 201 assert db.added.dedupe_key == "push:autoland:abc" From 817ff63506bb651261211cb232c1fc58541dd4d1 Mon Sep 17 00:00:00 2001 From: Suhaib Mujahid Date: Wed, 16 Sep 2026 12:47:21 -0400 Subject: [PATCH 5/6] Clarify dedupe_key semantics --- docs/hackbot/api.md | 18 +++++++++--------- libs/hackbot-client/hackbot_client/client.py | 6 +++--- .../versions/a7d4e9c21b83_runs_dedupe_key.py | 2 +- services/hackbot-api/app/routers/runs.py | 5 +++-- 4 files changed, 16 insertions(+), 15 deletions(-) diff --git a/docs/hackbot/api.md b/docs/hackbot/api.md index 13ea594ce5..06e743df7d 100644 --- a/docs/hackbot/api.md +++ b/docs/hackbot/api.md @@ -90,14 +90,14 @@ leaves a pending run with no execution, which the stale-run sweep finalizes. ## Deduplication External triggers fan out: 20 build tasks fail on one push, Phabricator retries a delivery, -a Slack button gets clicked twice. A caller **names the work** with `?dedupe_key=`, scoped to -the agent, and the rule is the whole design: +a Slack button gets clicked twice. A caller **gives the work a key** with `?dedupe_key=`, +scoped to the agent, and the rule is the whole design: -> a key names one run, for good. +> a key belongs to one run, for good. -A new run answers `201`. A request whose name another run already holds answers `200` with +A new run answers `201`. A request whose key another run already holds answers `200` with that run, which is not an error and needs no handling; `GET /runs?dedupe_key=…` finds it -again later. The name is coalescing rather than a request fingerprint: requests sharing one +again later. The key is coalescing rather than a request fingerprint: requests sharing one may carry different inputs, and the first to arrive is the one that runs. Recurrence lives in the key, since only the caller knows whether the work may happen again: @@ -107,11 +107,11 @@ Recurrence lives in the key, since only the caller knows whether the work may ha | Investigate this push exactly once | `push::` | | ...but let tomorrow try again | `push:::` | | Handle this delivery exactly once | `phab-txn:`, `ni:` | -| Always run | (omit the header) | +| Always run | (omit the parameter) | -A run keeps its name whatever becomes of it, a failed dispatch included, so a repeated -trigger gets the failure rather than a silent retry. A transient error therefore spends the -key: name the work differently, or run unkeyed. +A run keeps its key whatever becomes of it, a failed dispatch included, so a repeated +trigger gets the failure rather than a silent retry. A transient error therefore spends that +key: key the work differently, or run unkeyed. ## Run states diff --git a/libs/hackbot-client/hackbot_client/client.py b/libs/hackbot-client/hackbot_client/client.py index c0df79e9a5..4668b8ec6b 100644 --- a/libs/hackbot-client/hackbot_client/client.py +++ b/libs/hackbot-client/hackbot_client/client.py @@ -30,9 +30,9 @@ async def trigger_run( ) -> RunRef: """Create an agent run and return the API's typed run reference. - `dedupe_key` names the work the run does, and the name belongs to one - run for good: repeated triggers for the key will be no-ops, and the same - run reference will be returned. + `dedupe_key` keys the work the run does, and a key belongs to one run + for good: repeated triggers carrying it are no-ops, answered with the + same run reference. """ headers = {"X-API-Key": self._api_key} if on_behalf_of is not None: diff --git a/services/hackbot-api/alembic/versions/a7d4e9c21b83_runs_dedupe_key.py b/services/hackbot-api/alembic/versions/a7d4e9c21b83_runs_dedupe_key.py index 9a148a106a..8cb0fab354 100644 --- a/services/hackbot-api/alembic/versions/a7d4e9c21b83_runs_dedupe_key.py +++ b/services/hackbot-api/alembic/versions/a7d4e9c21b83_runs_dedupe_key.py @@ -21,7 +21,7 @@ def upgrade() -> None: """Upgrade schema.""" op.add_column("runs", sa.Column("dedupe_key", sa.String(), nullable=True)) - # Unique, because a key names one run for good: the constraint is what + # Unique, because a key belongs to one run for good: the constraint is what # decides a duplicate trigger, not just a check on one. `dedupe_key` leads # so the index also serves the `?dedupe_key=` listing filter, which carries # no agent. Existing rows all have a NULL key, and NULLs do not collide, so diff --git a/services/hackbot-api/app/routers/runs.py b/services/hackbot-api/app/routers/runs.py index bfa8de6cb2..62c9d94759 100644 --- a/services/hackbot-api/app/routers/runs.py +++ b/services/hackbot-api/app/routers/runs.py @@ -145,7 +145,8 @@ async def create_run( await db.commit() except IntegrityError: if dedupe_key is None: - # No key means no name to lose, so this is a real constraint fault. + # Without a key there is no duplicate to read this as, so it is a + # real constraint fault. raise await db.rollback() @@ -219,7 +220,7 @@ async def list_runs( if requested_by is not None: stmt = stmt.where(Run.requested_by == requested_by) if dedupe_key is not None: - # The trigger's own handle on its run: a caller that named the work can + # The trigger's own handle on its run: a caller that keyed the work can # find it again without having stored the run id. stmt = stmt.where(Run.dedupe_key == dedupe_key) # created_at is the sort key; run_id is a deterministic tiebreaker so offset From 483d57ec570427051895c126f80fe718364ff47f Mon Sep 17 00:00:00 2001 From: Suhaib Mujahid Date: Wed, 16 Sep 2026 18:36:21 -0400 Subject: [PATCH 6/6] Use upsert for run dedupe claims --- services/hackbot-api/app/routers/runs.py | 43 ++++++----- services/hackbot-api/tests/conftest.py | 75 ++++++++++++------- .../hackbot-api/tests/test_create_run_api.py | 27 ++++--- 3 files changed, 85 insertions(+), 60 deletions(-) diff --git a/services/hackbot-api/app/routers/runs.py b/services/hackbot-api/app/routers/runs.py index 62c9d94759..cbdf2a726c 100644 --- a/services/hackbot-api/app/routers/runs.py +++ b/services/hackbot-api/app/routers/runs.py @@ -8,7 +8,7 @@ from fastapi.responses import Response from pydantic import BeforeValidator, StringConstraints from sqlalchemy import select -from sqlalchemy.exc import IntegrityError +from sqlalchemy.dialects.postgresql import insert as pg_insert from sqlalchemy.ext.asyncio import AsyncSession from app import gcs, jobs, pubsub @@ -129,27 +129,26 @@ async def create_run( run_id = uuid.uuid4() results_prefix = gcs.run_prefix(str(run_id)) - run = Run( - run_id=run_id, - agent=agent.name, - status=RunStatus.pending.value, - inputs=inputs.model_dump(mode="json"), - requested_by=on_behalf_of, - dedupe_key=dedupe_key, - results_prefix=results_prefix, - artifacts=[], + claim = ( + pg_insert(Run) + .values( + run_id=run_id, + agent=agent.name, + status=RunStatus.pending.value, + inputs=inputs.model_dump(mode="json"), + requested_by=on_behalf_of, + dedupe_key=dedupe_key, + results_prefix=results_prefix, + artifacts=[], + ) + .on_conflict_do_nothing(index_elements=["dedupe_key", "agent"]) + .returning(Run) ) - db.add(run) - - try: - await db.commit() - except IntegrityError: - if dedupe_key is None: - # Without a key there is no duplicate to read this as, so it is a - # real constraint fault. - raise + run = await db.scalar(claim) - await db.rollback() + if run is None: + # RETURNING only yields rows it inserted, so nothing came back: the + # index turned this insert away, which only a supplied key can cause. result = await db.execute( select(Run).where(Run.agent == agent.name, Run.dedupe_key == dedupe_key) ) @@ -168,6 +167,10 @@ async def create_run( status_code=status.HTTP_200_OK, ) + # Publishes the claim to the other API instances before this request does + # anything slow. + await db.commit() + try: policy = await gcs.generate_results_policy(str(run_id)) env_overrides: dict[str, str] = { diff --git a/services/hackbot-api/tests/conftest.py b/services/hackbot-api/tests/conftest.py index efc3eb1497..9f9a7f3ce8 100644 --- a/services/hackbot-api/tests/conftest.py +++ b/services/hackbot-api/tests/conftest.py @@ -19,13 +19,33 @@ from app.database.models import Run # noqa: E402 from app.main import app # noqa: E402 from fastapi.testclient import TestClient # noqa: E402 +from sqlalchemy import Insert, Update # noqa: E402 +from sqlalchemy.dialects import postgresql # noqa: E402 from sqlalchemy.exc import ( # noqa: E402 + IntegrityError, MultipleResultsFound, NoResultFound, - PendingRollbackError, ) +def _statement_values(stmt) -> dict: + """The column values a statement would write, as a plain dict. + + Compiled params also carry the WHERE clause's binds, which SQLAlchemy names + with a suffix (`run_id_1`), so matching column names exactly keeps only the + values being written. + """ + params = stmt.compile(dialect=postgresql.dialect()).params + columns = {column.name for column in Run.__table__.columns} + return {name: value for name, value in params.items() if name in columns} + + +def _skips_conflicts(stmt) -> bool: + """Whether the INSERT asked Postgres to skip a unique-index conflict.""" + sql = str(stmt.compile(dialect=postgresql.dialect())) + return "ON CONFLICT" in sql and "DO NOTHING" in sql + + class _Scalars: def __init__(self, rows): self._rows = rows @@ -57,60 +77,59 @@ class FakeSession: Statements are kept so a test can assert on the SQL a handler built, and `matches` is what any query against `runs` returns -- which is how a test - sets up "a run already holds this dedupe key" without a Postgres. Set - `on_commit` to a callable to interpose on the next commit, which is how a - test provokes the unique-index conflict the dedupe path is built around. + sets up "a run already holds this dedupe key" without a Postgres. - A failed commit poisons the session until it is rolled back, as a real one - does, so a handler that reads after a rejected insert without rolling back - fails here rather than only against Postgres. + An INSERT ... RETURNING answers with the row it would have written, kept as + `added` so a test can assert on what the handler built and on what it does + to the row afterwards. Set `conflict` to have the unique index turn that + insert away: it returns nothing if the statement asked for DO NOTHING, and + raises IntegrityError if it did not, which is what Postgres would do. """ def __init__(self, matches: list | None = None): self.matches = matches or [] self.added = None + self.conflict = False self.statements: list = [] self.commits = 0 self.rollbacks = 0 - self.on_commit = None - self.needs_rollback = False @property def stmt(self): """The last statement executed, for tests that assert on one query.""" return self.statements[-1] if self.statements else None - async def execute(self, stmt): - if self.needs_rollback: - raise PendingRollbackError( - "This Session's transaction has been rolled back due to a " - "previous exception during flush." - ) + async def scalar(self, stmt): + return (await self.execute(stmt)).scalars().first() + async def execute(self, stmt): self.statements.append(stmt) + + if isinstance(stmt, Update): + for column, value in _statement_values(stmt).items(): + setattr(self.added, column, value) + return _Result([]) + + if isinstance(stmt, Insert): + if self.conflict: + if not _skips_conflicts(stmt): + raise IntegrityError( + "INSERT INTO runs", {}, Exception("uq_runs_dedupe_key") + ) + return _Result([]) + self.added = Run(**_statement_values(stmt)) + return _Result([self.added]) + # Only queries over `runs` return rows; anything else a handler runs # for effect gets an empty result. if Run.__table__ in stmt.get_final_froms(): return _Result(self.matches) return _Result([]) - def add(self, obj): - self.added = obj - async def commit(self): - if self.on_commit is not None: - # One-shot, so a test can fail the claiming commit and still let the - # handler's later commits through. - hook, self.on_commit = self.on_commit, None - try: - hook() - except Exception: - self.needs_rollback = True - raise self.commits += 1 async def rollback(self): - self.needs_rollback = False self.rollbacks += 1 async def get(self, model, key): diff --git a/services/hackbot-api/tests/test_create_run_api.py b/services/hackbot-api/tests/test_create_run_api.py index fe97e38296..cf29f09bb4 100644 --- a/services/hackbot-api/tests/test_create_run_api.py +++ b/services/hackbot-api/tests/test_create_run_api.py @@ -14,7 +14,6 @@ import pytest from app import gcs, jobs from app.schemas import RunStatus -from sqlalchemy.exc import IntegrityError @pytest.fixture(autouse=True) @@ -79,17 +78,14 @@ def _create_keyed(client, key="push:autoland:abc"): def _lose_the_key(db, winner): - """Make the next insert lose the name to `winner`, as the index would. + """Make the next insert lose the key to `winner`, as the index would. - `uq_runs_dedupe_key` rejects the insert and the run that won the name is - there to be read afterwards, which is the only way a request collapses. + ON CONFLICT DO NOTHING writes nothing and returns nothing, and the run that + won the key is there to be read afterwards, which is the only way a request + collapses. """ - - def lose(): - db.matches = [winner] - raise IntegrityError("INSERT INTO runs", {}, Exception("uq_runs_dedupe_key")) - - db.on_commit = lose + db.conflict = True + db.matches = [winner] def test_keyed_request_records_the_key_on_the_new_run(client, db): @@ -107,8 +103,15 @@ def test_keyed_request_is_answered_with_the_run_holding_the_key(client, db): # 200, not 201: this request created nothing. assert resp.status_code == 200, resp.text assert resp.json()["run_id"] == str(winner.run_id) - # Ours was given back rather than left in a broken transaction. - assert db.rollbacks == 1 + # Nothing of ours was written, and skipping the conflict rather than raising + # it leaves the transaction usable, so there is nothing to roll back either. + assert db.commits == 0 + assert db.rollbacks == 0 + # The run handed back is looked up by agent as well as by key: two agents + # may hold the same key, and the other one's run is a different answer. + lookup = str(db.stmt.compile(compile_kwargs={"literal_binds": True})) + assert "runs.agent = 'bug-fix'" in lookup + assert "runs.dedupe_key = 'push:autoland:abc'" in lookup @pytest.mark.parametrize(