diff --git a/docs/hackbot/api.md b/docs/hackbot/api.md index 2a2244d1e7..06e743df7d 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 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. @@ -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 **gives the work a key** with `?dedupe_key=`, +scoped to the agent, and the rule is the whole design: + +> a key belongs to one run, for good. + +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 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: + +| 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 parameter) | + +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 @@ -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..4668b8ec6b 100644 --- a/libs/hackbot-client/hackbot_client/client.py +++ b/libs/hackbot-client/hackbot_client/client.py @@ -26,17 +26,26 @@ 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` 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: headers["X-On-Behalf-Of"] = on_behalf_of + 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/alembic/versions/a7d4e9c21b83_runs_dedupe_key.py b/services/hackbot-api/alembic/versions/a7d4e9c21b83_runs_dedupe_key.py new file mode 100644 index 0000000000..8cb0fab354 --- /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 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 + # 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..cbdf2a726c 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.dialects.postgresql import insert as pg_insert 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) @@ -67,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, @@ -78,8 +108,18 @@ async def create_run( description="Email of the user this run is requested for.", ), ] = None, + dedupe_key: Annotated[ + DedupeKey, + 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 " + "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 +129,61 @@ 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, - 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) - 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), - } + run = await db.scalar(claim) + + 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) + ) + 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, + ) + + # 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] = { + "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 +209,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 +222,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 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 # 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/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/services/hackbot-api/tests/conftest.py b/services/hackbot-api/tests/conftest.py index 500a1ff9af..9f9a7f3ce8 100644 --- a/services/hackbot-api/tests/conftest.py +++ b/services/hackbot-api/tests/conftest.py @@ -12,3 +12,141 @@ 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 import Insert, Update # noqa: E402 +from sqlalchemy.dialects import postgresql # noqa: E402 +from sqlalchemy.exc import ( # noqa: E402 + IntegrityError, + MultipleResultsFound, + NoResultFound, +) + + +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 + + 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. + + 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 + + @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 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([]) + + async def commit(self): + self.commits += 1 + + async def rollback(self): + 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..cf29f09bb4 100644 --- a/services/hackbot-api/tests/test_create_run_api.py +++ b/services/hackbot-api/tests/test_create_run_api.py @@ -1,14 +1,19 @@ -"""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 @pytest.fixture(autouse=True) @@ -49,3 +54,91 @@ 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}, + params={"dedupe_key": key}, + ) + + +def _lose_the_key(db, winner): + """Make the next insert lose the key to `winner`, as the index would. + + 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. + """ + db.conflict = True + db.matches = [winner] + + +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. + assert _create_keyed(client, key=" push:autoland:abc ").status_code == 201 + 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) + # 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( + ("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; 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 = [