From b6602fad1ff7f4808459d0c5b29ab682025f2ebe Mon Sep 17 00:00:00 2001 From: Jasper Mayone Date: Sat, 10 Oct 2026 16:17:42 -0400 Subject: [PATCH 1/7] feat: add the quilt_outbox table --- .../versions/83af68b415ca_add_quilt_outbox.py | 70 +++++++++++++++++++ krater/models/__init__.py | 4 ++ krater/models/enums.py | 9 +++ krater/models/quilt_outbox.py | 62 ++++++++++++++++ 4 files changed, 145 insertions(+) create mode 100644 alembic/versions/83af68b415ca_add_quilt_outbox.py create mode 100644 krater/models/quilt_outbox.py diff --git a/alembic/versions/83af68b415ca_add_quilt_outbox.py b/alembic/versions/83af68b415ca_add_quilt_outbox.py new file mode 100644 index 0000000..867378f --- /dev/null +++ b/alembic/versions/83af68b415ca_add_quilt_outbox.py @@ -0,0 +1,70 @@ +"""add quilt outbox + +Revision ID: 83af68b415ca +Revises: 5f3b46edeb95 +Create Date: 2026-10-10 16:06:56.374626 + +`quilt_outbox` holds the events Krater sends to Quilt's patch API. The services write a row in the same +transaction as the change it reports, and the worker sends the rows. See docs/quilt-integration.md. + +The `state` column's `sa.Enum(..., create_constraint=True)` attaches its own CHECK constraint, so there is +no separate `sa.CheckConstraint` here (see the initial schema migration). +""" + +from collections.abc import Sequence + +import sqlalchemy as sa +from sqlalchemy.dialects import postgresql + +from alembic import op + +# revision identifiers, used by Alembic. +revision: str = "83af68b415ca" +down_revision: str | Sequence[str] | None = "5f3b46edeb95" +branch_labels: str | Sequence[str] | None = None +depends_on: str | Sequence[str] | None = None + + +def upgrade() -> None: + """Upgrade schema.""" + op.create_table( + "quilt_outbox", + sa.Column("id", sa.UUID(), nullable=False), + sa.Column("seq", sa.BigInteger(), sa.Identity(always=True), nullable=False), + sa.Column("type", sa.String(length=64), nullable=False), + sa.Column("external_id", sa.String(length=200), nullable=False), + sa.Column("payload", postgresql.JSONB(astext_type=sa.Text()), nullable=False), + sa.Column("occurred_at", sa.DateTime(timezone=True), nullable=False), + sa.Column( + "state", + sa.Enum( + "pending", + "sent", + "failed", + "skipped", + name="quilt_outbox_state", + native_enum=False, + create_constraint=True, + length=64, + ), + server_default="pending", + nullable=False, + ), + sa.Column("attempts", sa.Integer(), server_default="0", nullable=False), + sa.Column("next_attempt_at", sa.DateTime(timezone=True), nullable=True), + sa.Column("sent_at", sa.DateTime(timezone=True), nullable=True), + sa.Column("last_status", sa.Integer(), nullable=True), + sa.Column("last_error", sa.Text(), nullable=True), + sa.Column("created_at", sa.DateTime(timezone=True), server_default=sa.text("now()"), nullable=False), + sa.PrimaryKeyConstraint("id", name=op.f("pk_quilt_outbox")), + sa.UniqueConstraint("seq", name=op.f("uq_quilt_outbox_seq")), + ) + op.create_index("ix_quilt_outbox_external_id_seq", "quilt_outbox", ["external_id", "seq"], unique=False) + op.create_index("ix_quilt_outbox_state_seq", "quilt_outbox", ["state", "seq"], unique=False) + + +def downgrade() -> None: + """Downgrade schema.""" + op.drop_index("ix_quilt_outbox_state_seq", table_name="quilt_outbox") + op.drop_index("ix_quilt_outbox_external_id_seq", table_name="quilt_outbox") + op.drop_table("quilt_outbox") diff --git a/krater/models/__init__.py b/krater/models/__init__.py index 1f4d47b..decfe45 100644 --- a/krater/models/__init__.py +++ b/krater/models/__init__.py @@ -7,6 +7,7 @@ ApprovalStage, BudgetEntryKind, ProjectStatus, + QuiltOutboxState, ReviewDecision, ReviewSource, RevisionKind, @@ -16,6 +17,7 @@ from krater.models.gpu_price import GpuPrice from krater.models.project import Project from krater.models.project_revision import ProjectRevision +from krater.models.quilt_outbox import QuiltOutbox from krater.models.review import Review from krater.models.slack_notification import SlackNotification from krater.models.spend_snapshot import SpendSnapshot @@ -31,6 +33,8 @@ "Project", "ProjectRevision", "ProjectStatus", + "QuiltOutbox", + "QuiltOutboxState", "Review", "ReviewDecision", "ReviewSource", diff --git a/krater/models/enums.py b/krater/models/enums.py index 96397b7..66c9339 100644 --- a/krater/models/enums.py +++ b/krater/models/enums.py @@ -78,6 +78,15 @@ class SpendSource(enum.StrEnum): SKYPILOT_COST_REPORT = "skypilot_cost_report" +class QuiltOutboxState(enum.StrEnum): + """Where a `QuiltOutbox` row is in its delivery to Quilt. See `docs/quilt-integration.md`.""" + + PENDING = "pending" # not sent yet, or waiting for a retry + SENT = "sent" # Quilt answered 201 applied or 200 duplicate + FAILED = "failed" # Quilt refused it for good; an admin must look (blocks later events of the subject) + SKIPPED = "skipped" # never sent: nothing to tell Quilt, or an admin dismissed it + + class ApprovalStage(enum.StrEnum): PROPOSAL = "proposal" COMPLETION = "completion" diff --git a/krater/models/quilt_outbox.py b/krater/models/quilt_outbox.py new file mode 100644 index 0000000..3fc607f --- /dev/null +++ b/krater/models/quilt_outbox.py @@ -0,0 +1,62 @@ +"""QuiltOutbox: one event for Quilt's patch API, written in the same transaction as the change it reports. + +The row's `id` is the event id Quilt sees, so a retry always sends the same id and payload. `seq` gives +the order: the sender sends the rows of one subject (`external_id`) strictly in `seq` order. See +`docs/quilt-integration.md`. +""" + +from __future__ import annotations + +import uuid +from datetime import UTC, datetime + +import sqlalchemy as sa +from sqlalchemy.dialects.postgresql import JSONB, UUID +from sqlalchemy.orm import Mapped, mapped_column + +from krater.db import Base +from krater.models.enums import QuiltOutboxState, pg_enum +from krater.models.mixins import CreatedAtMixin + + +class QuiltOutbox(CreatedAtMixin, Base): + __tablename__ = "quilt_outbox" + __table_args__ = ( + sa.Index("ix_quilt_outbox_external_id_seq", "external_id", "seq"), + sa.Index("ix_quilt_outbox_state_seq", "state", "seq"), + ) + + # The event id sent to Quilt. Derived with uuid5 from the source rows, never random: see + # `krater.services.quilt_events`. + id: Mapped[uuid.UUID] = mapped_column(UUID(as_uuid=True), primary_key=True) + seq: Mapped[int] = mapped_column(sa.BigInteger, sa.Identity(always=True), nullable=False, unique=True) + type: Mapped[str] = mapped_column(sa.String(64), nullable=False) + external_id: Mapped[str] = mapped_column(sa.String(200), nullable=False) + # The event's `data` object, exactly as it is sent. + payload: Mapped[dict] = mapped_column(JSONB, nullable=False) + occurred_at: Mapped[datetime] = mapped_column(sa.DateTime(timezone=True), nullable=False) + state: Mapped[QuiltOutboxState] = mapped_column( + pg_enum(QuiltOutboxState, name="quilt_outbox_state"), + nullable=False, + default=QuiltOutboxState.PENDING, + server_default=QuiltOutboxState.PENDING.value, + ) + attempts: Mapped[int] = mapped_column(sa.Integer, nullable=False, default=0, server_default="0") + # `None` means "send it on the next run". Set after a failed attempt, for the backoff. + next_attempt_at: Mapped[datetime | None] = mapped_column(sa.DateTime(timezone=True), nullable=True) + sent_at: Mapped[datetime | None] = mapped_column(sa.DateTime(timezone=True), nullable=True) + # The HTTP status of the last attempt (`None`: no attempt yet, or no answer). + last_status: Mapped[int | None] = mapped_column(sa.Integer, nullable=True) + last_error: Mapped[str | None] = mapped_column(sa.Text, nullable=True) + + def event_body(self) -> dict: + """The JSON body for `POST /api/v1/events`.""" + return { + "id": str(self.id), + "type": self.type, + "occurred_at": self.occurred_at.astimezone(UTC).isoformat(), + "data": self.payload, + } + + def __repr__(self) -> str: + return f"" From 4c28dd9155e7d076db1a04b82516554e02070c6f Mon Sep 17 00:00:00 2001 From: Jasper Mayone Date: Sat, 10 Oct 2026 16:17:57 -0400 Subject: [PATCH 2/7] feat: get a weave token with the quilt scope --- docs/dev/weave-e2e.md | 2 +- krater/weave/client.py | 10 ++++++ krater/weave/live.py | 51 +++++++++++++++++++++-------- krater/weave/stub.py | 15 ++++++++- scripts/dev/weave_e2e_provision.rb | 6 ++-- tests/weave/test_live_client.py | 52 ++++++++++++++++++++++++++++++ tests/weave/test_stub_client.py | 10 +++++- 7 files changed, 126 insertions(+), 20 deletions(-) diff --git a/docs/dev/weave-e2e.md b/docs/dev/weave-e2e.md index c8383d8..cc587cd 100644 --- a/docs/dev/weave-e2e.md +++ b/docs/dev/weave-e2e.md @@ -21,7 +21,7 @@ idempotently create, in Weave's own database: confirmed emails and have accepted the Code of Conduct. Weave refuses every app to a user who has not (patchworklabsorg/weave#171). - a confidential OAuth application ("Krater (e2e)") with redirect URI - `http://localhost:8201/auth/callback`, scopes `openid profile email groups roles slack directory`, and + `http://localhost:8201/auth/callback`, scopes `openid profile email groups roles slack directory quilt`, and `access_policy` `everyone`. The open policy lets the non-member reach Krater, so the check proves that Krater itself refuses them. The script **recreates** the app every run, because Weave hashes its secret at rest and shows it only right after creation. diff --git a/krater/weave/client.py b/krater/weave/client.py index 40211bd..2162b04 100644 --- a/krater/weave/client.py +++ b/krater/weave/client.py @@ -33,3 +33,13 @@ def get_user(self, sub: str) -> WeaveUser | None: def list_users_with_role(self, role: str) -> list[WeaveUser]: """Every user who holds a Krater role (a `ganymede:*` name, e.g. `ganymede:reviewer`).""" ... + + def quilt_token(self) -> str: + """Krater's own access token for Quilt's patch API: a client_credentials token with the `quilt` + scope, cached until shortly before it expires. Raises `WeaveUnavailableError` if Weave can't be + reached or refuses the token (for example, the scope isn't allowed on the Krater app).""" + ... + + def invalidate_quilt_token(self) -> None: + """Forget the cached Quilt token (Quilt answered 401), so the next call fetches a new one.""" + ... diff --git a/krater/weave/live.py b/krater/weave/live.py index d6e7d1b..6b08057 100644 --- a/krater/weave/live.py +++ b/krater/weave/live.py @@ -5,8 +5,10 @@ The directory API is called with an access token from Krater's own OAuth app (the client_credentials grant, scope `directory`). The token is cached until shortly before it expires, and refetched once if -Weave answers 401. `get_user` answers are cached for a short TTL to absorb bursts of lookups (a page -of actions, Slack button clicks). See `docs/weave-integration.md` for the full contract. +Weave answers 401. A second client_credentials token, scope `quilt`, is what Krater shows Quilt (see +`quilt_token`); each scope has its own cache entry. `get_user` answers are cached for a short TTL to +absorb bursts of lookups (a page of actions, Slack button clicks). See `docs/weave-integration.md` for +the full contract. """ from __future__ import annotations @@ -36,6 +38,9 @@ #: The scope Krater asks for on its client_credentials token for the directory API. DIRECTORY_SCOPE = "directory" +#: The scope Krater asks for on its client_credentials token for Quilt's patch API. +QUILT_SCOPE = "quilt" + #: How long a `get_user` answer is trusted before re-fetching. DIRECTORY_CACHE_TTL_SECONDS = 60.0 @@ -61,8 +66,8 @@ def __init__(self, settings: Settings, *, http_client: httpx.Client | None = Non self._roles = RoleMapping.from_settings(settings) self._discovery_doc: dict | None = None self._jwk_set: KeySet | None = None - self._access_token: str | None = None - self._access_token_expires_at = 0.0 + # client_credentials tokens by scope: (token, monotonic time to refetch at). + self._service_tokens: dict[str, tuple[str, float]] = {} self._directory_cache: TTLCache[str, WeaveUser | None] = TTLCache(DIRECTORY_CACHE_TTL_SECONDS) # -- OIDC ------------------------------------------------------------------------------------ @@ -192,7 +197,7 @@ def _directory_get(self, path: str, *, params: dict[str, str] | None = None) -> response = self._authorized_get(url, params) if response.status_code == 401: # The token may have been revoked or rotated early: fetch a new one and try once more. - self._access_token = None + self._service_tokens.pop(DIRECTORY_SCOPE, None) response = self._authorized_get(url, params) if response.status_code == 403: # Not a token problem, so no refetch: the token lacks the `directory` scope, or Weave doesn't @@ -207,28 +212,47 @@ def _directory_get(self, path: str, *, params: dict[str, str] | None = None) -> return response def _authorized_get(self, url: str, params: dict[str, str] | None) -> httpx.Response: - headers = {"Authorization": f"Bearer {self._directory_token()}"} + headers = {"Authorization": f"Bearer {self._service_token(DIRECTORY_SCOPE)}"} try: return self._http.get(url, params=params, headers=headers) except httpx.HTTPError as exc: raise WeaveUnavailableError(f"could not reach Weave's directory at {url}") from exc - def _directory_token(self) -> str: - """Krater's client_credentials access token, cached until shortly before it expires.""" - if self._access_token is not None and time.monotonic() < self._access_token_expires_at: - return self._access_token + # -- client_credentials tokens --------------------------------------------------------------- + + def quilt_token(self) -> str: + """Krater's client_credentials access token for Quilt's patch API (scope `quilt`).""" + return self._service_token(QUILT_SCOPE) + + def invalidate_quilt_token(self) -> None: + """Forget the cached Quilt token, so the next `quilt_token` call fetches a new one.""" + self._service_tokens.pop(QUILT_SCOPE, None) + + def _service_token(self, scope: str) -> str: + """Krater's client_credentials access token for `scope`, cached until shortly before it expires.""" + cached = self._service_tokens.get(scope) + if cached is not None and time.monotonic() < cached[1]: + return cached[0] token_url = f"{self._settings.weave_issuer.rstrip('/')}/oauth/token" try: response = self._http.post( token_url, - data={"grant_type": "client_credentials", "scope": DIRECTORY_SCOPE}, + data={"grant_type": "client_credentials", "scope": scope}, auth=(self._settings.weave_client_id, self._settings.weave_client_secret), ) except httpx.HTTPError as exc: raise WeaveUnavailableError("could not reach Weave's token endpoint") from exc if response.status_code != 200: - raise WeaveUnavailableError(f"Weave refused Krater's directory token: {response.status_code}") + if response.status_code in (400, 401, 403): + # Usually `invalid_scope`: an admin hasn't allowed this scope on the Krater app in Weave. + logger.error( + "Weave refused Krater a client_credentials token for scope %r (%s): allow the scope on " + "the Krater app in Weave and check KRATER_WEAVE_CLIENT_ID/SECRET", + scope, + response.status_code, + ) + raise WeaveUnavailableError(f"Weave refused Krater's {scope} token: {response.status_code}") body = response.json() token = body.get("access_token") if isinstance(body, dict) else None @@ -237,8 +261,7 @@ def _directory_token(self) -> str: expires_in = body.get("expires_in") lifetime = float(expires_in) if isinstance(expires_in, int | float) else DEFAULT_TOKEN_LIFETIME_SECONDS - self._access_token = token - self._access_token_expires_at = time.monotonic() + max(lifetime - TOKEN_EXPIRY_MARGIN_SECONDS, 0.0) + self._service_tokens[scope] = (token, time.monotonic() + max(lifetime - TOKEN_EXPIRY_MARGIN_SECONDS, 0.0)) return token @staticmethod diff --git a/krater/weave/stub.py b/krater/weave/stub.py index de27cca..89b718b 100644 --- a/krater/weave/stub.py +++ b/krater/weave/stub.py @@ -24,6 +24,9 @@ from krater.weave.roles import RoleMapping from krater.weave.types import WeaveIdentity, WeaveUser +#: The fake client_credentials token `quilt_token` returns. A real Quilt refuses it (Weave never issued it). +STUB_QUILT_TOKEN = "stub-quilt-token" + #: The bundled fixture, used whenever `settings.weave_stub_users_file` is blank. DEFAULT_STUB_USERS_FILE = Path(__file__).parent / "stub_users.json" @@ -53,6 +56,9 @@ def __init__(self, stub_users_file: str | Path | None = None, *, role_mapping: R self._mapping = role_mapping if role_mapping is not None else RoleMapping.default() self._users_by_sub: dict[str, StubUser] = {} + # How often `quilt_token` / `invalidate_quilt_token` were called, for tests. + self.quilt_token_requests = 0 + self.quilt_token_invalidations = 0 for entry in raw_users: raw_roles = entry.get("roles") user = StubUser( @@ -99,6 +105,13 @@ def list_users_with_role(self, role: str) -> list[WeaveUser]: records = (self._record(user) for user in self._users_by_sub.values()) return [record for record in records if role in record.roles] + def quilt_token(self) -> str: + self.quilt_token_requests += 1 + return STUB_QUILT_TOKEN + + def invalidate_quilt_token(self) -> None: + self.quilt_token_invalidations += 1 + # -- Stub-only helpers --------------------------------------------------------------------------- def list_all_users(self) -> list[WeaveUser]: @@ -162,4 +175,4 @@ def _record(self, user: StubUser) -> WeaveUser: ) -__all__ = ["DEFAULT_STUB_USERS_FILE", "StubUser", "StubWeaveClient"] +__all__ = ["DEFAULT_STUB_USERS_FILE", "STUB_QUILT_TOKEN", "StubUser", "StubWeaveClient"] diff --git a/scripts/dev/weave_e2e_provision.rb b/scripts/dev/weave_e2e_provision.rb index 57d8a4c..993bd1f 100644 --- a/scripts/dev/weave_e2e_provision.rb +++ b/scripts/dev/weave_e2e_provision.rb @@ -3,8 +3,8 @@ # Provisions a local Weave with everything the live Krater<->Weave e2e check # needs: three users (a member, an admin and a non-member), a confidential # OAuth application for Krater with the scopes Krater asks for plus -# `directory` for its client_credentials token, and Krater's app roles on that -# application. Idempotent (upsert-by-email), so it can be re-run against the +# `directory` and `quilt` for its client_credentials tokens, and Krater's app +# roles on that application. The `quilt` scope needs patchworklabsorg/weave#176. Idempotent (upsert-by-email), so it can be re-run against the # same dev database. # # Weave owns Krater's roles (patchworklabsorg/weave#165, #166). This script @@ -65,7 +65,7 @@ def upsert_user(email:, first_name:, last_name:) name: app_name, redirect_uri: redirect_uri, confidential: true, - scopes: "openid profile email groups roles slack directory", + scopes: "openid profile email groups roles slack directory quilt", access_policy: "everyone", requires_code_of_conduct: true ) diff --git a/tests/weave/test_live_client.py b/tests/weave/test_live_client.py index ba57545..258555c 100644 --- a/tests/weave/test_live_client.py +++ b/tests/weave/test_live_client.py @@ -471,3 +471,55 @@ def handler(request: httpx.Request) -> httpx.Response: with pytest.raises(WeaveUnavailableError): client.list_users_with_role(GROUP_REVIEWER) + + +# -------------------------------------------------------------------------------------------------- +# The Quilt token (scope `quilt`) +# -------------------------------------------------------------------------------------------------- + + +def test_the_quilt_token_asks_for_the_quilt_scope_with_basic_auth_and_is_cached( + fake_weave: FakeWeave, live_client: LiveWeaveClient +) -> None: + first = live_client.quilt_token() + second = live_client.quilt_token() + + assert first == second == "cc-1" + [request] = _token_requests(fake_weave) + assert parse_qs(request.content.decode()) == {"grant_type": ["client_credentials"], "scope": ["quilt"]} + expected_basic = base64.b64encode(f"{CLIENT_ID}:{CLIENT_SECRET}".encode()).decode() + assert request.headers["authorization"] == f"Basic {expected_basic}" + + +def test_the_quilt_and_directory_tokens_are_cached_separately( + fake_weave: FakeWeave, live_client: LiveWeaveClient +) -> None: + quilt = live_client.quilt_token() + live_client.get_user("PWLNOBODY") + + scopes = [parse_qs(r.content.decode())["scope"] for r in _token_requests(fake_weave)] + assert scopes == [["quilt"], ["directory"]] + directory_requests = [r for r in fake_weave.requests if r.url.path.startswith("/api/v1/directory/")] + assert {r.headers["authorization"] for r in directory_requests} == {"Bearer cc-2"} + assert live_client.quilt_token() == quilt + + +def test_invalidating_the_quilt_token_fetches_a_new_one(fake_weave: FakeWeave, live_client: LiveWeaveClient) -> None: + live_client.quilt_token() + live_client.invalidate_quilt_token() + + assert live_client.quilt_token() == "cc-2" + + +def test_a_refused_quilt_token_raises_weave_unavailable_and_logs_the_config_error(caplog, monkeypatch) -> None: + # Alembic's `fileConfig` (run by the `engine` fixture) disables loggers that already exist. + monkeypatch.setattr(live_module.logger, "disabled", False) + + def handler(request: httpx.Request) -> httpx.Response: + return httpx.Response(400, json={"error": "invalid_scope"}) + + client = LiveWeaveClient(_settings(), http_client=httpx.Client(transport=httpx.MockTransport(handler))) + + with pytest.raises(WeaveUnavailableError): + client.quilt_token() + assert "scope 'quilt'" in caplog.text diff --git a/tests/weave/test_stub_client.py b/tests/weave/test_stub_client.py index c726bb7..e8c40c3 100644 --- a/tests/weave/test_stub_client.py +++ b/tests/weave/test_stub_client.py @@ -8,7 +8,7 @@ from krater.services.actor import GROUP_ADMIN, GROUP_MEMBER, GROUP_REVIEWER from krater.weave.errors import WeaveAuthError from krater.weave.roles import RoleMapping -from krater.weave.stub import StubWeaveClient +from krater.weave.stub import STUB_QUILT_TOKEN, StubWeaveClient @pytest.fixture @@ -98,3 +98,11 @@ def test_a_minimal_fixture_entry_defaults_to_verified_and_active_with_no_roles(t assert record.slack_id is None assert record.slack_member is None assert record.roles == frozenset() + + +def test_the_stub_quilt_token_is_a_fixed_fake_and_counts_calls() -> None: + stub = StubWeaveClient() + + assert stub.quilt_token() == STUB_QUILT_TOKEN + stub.invalidate_quilt_token() + assert (stub.quilt_token_requests, stub.quilt_token_invalidations) == (1, 1) From 25cde39b497965ba773b4e5f61f1d6632042739f Mon Sep 17 00:00:00 2001 From: Jasper Mayone Date: Sat, 10 Oct 2026 16:18:10 -0400 Subject: [PATCH 3/7] feat: write quilt events in the same transaction as the change --- krater/services/projects.py | 9 +- krater/services/quilt_events.py | 447 +++++++++++++++++++++++ krater/services/skypilot_sync.py | 4 +- scripts/quilt_backfill.py | 32 ++ tests/services/test_concurrency.py | 2 + tests/services/test_quilt_events.py | 531 ++++++++++++++++++++++++++++ 6 files changed, 1023 insertions(+), 2 deletions(-) create mode 100644 krater/services/quilt_events.py create mode 100644 scripts/quilt_backfill.py create mode 100644 tests/services/test_quilt_events.py diff --git a/krater/services/projects.py b/krater/services/projects.py index 4c2ffe3..19116e5 100644 --- a/krater/services/projects.py +++ b/krater/services/projects.py @@ -40,7 +40,7 @@ RevisionKind, RevisionOutcome, ) -from krater.services import approval_policy, audit, budget +from krater.services import approval_policy, audit, budget, quilt_events from krater.services.actor import Actor from krater.services.errors import InvalidState, NotAllowed, NotFound, ValidationFailed @@ -267,6 +267,7 @@ def submit(session: Session, actor: Actor, *, project: Project) -> Project: # `amendment`: the project stays `approved` while the amendment is reviewed. session.flush() + quilt_events.sync_project(session, project) return project @@ -417,6 +418,7 @@ def submit_completion(session: Session, actor: Actor, *, project: Project) -> Pr project.status = ProjectStatus.PENDING_COMPLETION_REVIEW session.flush() + quilt_events.sync_project(session, project) return project @@ -539,6 +541,7 @@ def record_review( elif approval_policy.is_satisfied(session, revision): _apply_approve(session, actor, revision) + quilt_events.sync_project(session, project) return review @@ -680,6 +683,7 @@ def admin_decide( payload={"revision_id": str(revision.id), "decision": decision.value}, reason=reason, ) + quilt_events.sync_project(session, project) return revision @@ -719,6 +723,7 @@ def admin_adjust_budget( audit.record( session, actor, "admin_adjust_budget", project=project, payload={"amount_cents": amount_cents}, reason=reason ) + quilt_events.sync_project(session, project) return entry @@ -753,6 +758,7 @@ def reclaim_budget( audit.record( session, actor, "admin_reclaim_budget", project=project, payload={"amount_cents": amount_cents}, reason=reason ) + quilt_events.sync_project(session, project) return entry @@ -809,6 +815,7 @@ def withdraw(session: Session, actor: Actor, *, project: Project, reason: str | ) session.flush() + quilt_events.sync_project(session, project) return project diff --git a/krater/services/quilt_events.py b/krater/services/quilt_events.py new file mode 100644 index 0000000..20c54f3 --- /dev/null +++ b/krater/services/quilt_events.py @@ -0,0 +1,447 @@ +"""What Krater tells Quilt: the mapping from Krater's projects, ledger and spend to Quilt's patch API events. + +Quilt keeps a thin record of each submission and a ledger of approved budget that is not spent yet (see +`docs/quilt-integration.md` and Quilt's `docs/patch-api.md`). This module writes those events to the +`quilt_outbox` table, in the caller's transaction, so an event exists exactly when the change it reports +commits. `krater.quilt.sender` sends them later. + +`sync_project` is the one entry point. It compares the project with what the outbox already says about it +and adds only the difference: + +- the first submission of a proposal -> `submission.created`; +- a change of title, status, requested amount or link -> `submission.updated`, with the changed fields; +- each `BudgetEntry` not yet reported -> `budget.committed` (a positive amount) or `budget.released` (a + negative one), in ledger order; +- a change of the latest `SpendSnapshot` -> `spend.recorded`, with the cumulative total. + +Because it is a diff, calling it again changes nothing, and the backfill (`backfill_all`) is the same +function run over every project. Event ids are uuid5 values from the source rows, so the backfill and the +live path agree on them. + +The external id of a submission is the Krater `Project.id`: it stays the same through resubmissions, +amendments and the completion review. +""" + +from __future__ import annotations + +import logging +import uuid +from dataclasses import dataclass +from datetime import UTC, datetime + +import sqlalchemy as sa +from sqlalchemy.dialects.postgresql import insert +from sqlalchemy.orm import Session + +from krater.config import get_settings +from krater.models import ( + AuditEvent, + BudgetEntry, + Project, + ProjectRevision, + QuiltOutbox, + QuiltOutboxState, + RevisionKind, + RevisionOutcome, + SpendSnapshot, + User, +) +from krater.services import audit +from krater.services.actor import Actor +from krater.services.errors import InvalidState, NotAllowed, NotFound, ValidationFailed + +logger = logging.getLogger(__name__) + +SUBMISSION_CREATED = "submission.created" +SUBMISSION_UPDATED = "submission.updated" +BUDGET_COMMITTED = "budget.committed" +BUDGET_RELEASED = "budget.released" +SPEND_RECORDED = "spend.recorded" + +#: The namespace of every Krater event id. Never change it: Quilt matches retries and the backfill by id. +EVENT_ID_NAMESPACE = uuid.UUID("6f1d2c1e-4b7a-5f0e-9c3d-2a8b7e6f5d40") + +#: The fields of a submission that `submission.updated` can change. +_SUBMISSION_FIELDS = ("title", "status", "requested_cents", "url") + + +def event_id(*parts: object) -> uuid.UUID: + """A stable event id from the source of the event (for example `"budget", entry.id`).""" + return uuid.uuid5(EVENT_ID_NAMESPACE, ":".join(str(part) for part in parts)) + + +def project_url(project: Project) -> str: + """The link Quilt shows for the submission: the project's page in Krater.""" + return f"{get_settings().base_url.rstrip('/')}/projects/{project.id}" + + +@dataclass +class _Replay: + """What Quilt knows about one submission, if it applied every outbox row so far (in `seq` order).""" + + created: bool = False + fields: dict | None = None + updated_count: int = 0 + remaining_cents: int = 0 # Quilt's remaining commitment: committed - released - spend drawdown + spend_total_cents: int | None = None # the last total sent + spend_high_cents: int = 0 # Quilt keeps the highest total it has seen + reported_ids: frozenset[uuid.UUID] = frozenset() + + +def _replay(session: Session, external_id: str) -> _Replay: + """Fold the outbox rows of `external_id` into what Quilt will know once it has them all. + + Rows that will never reach Quilt (`failed`, `skipped`) don't count toward Quilt's numbers, but they + still mark their source as reported, so a skipped release is never sent later out of order. + """ + replay = _Replay(fields={}) + reported: set[uuid.UUID] = set() + rows = session.scalars( + sa.select(QuiltOutbox).where(QuiltOutbox.external_id == external_id).order_by(QuiltOutbox.seq) + ) + for row in rows: + reported.add(row.id) + if row.type == SUBMISSION_UPDATED: + replay.updated_count += 1 + if row.state in (QuiltOutboxState.FAILED, QuiltOutboxState.SKIPPED): + if row.type == SUBMISSION_CREATED: + replay.created = True + continue + data = row.payload + if row.type == SUBMISSION_CREATED: + replay.created = True + replay.fields = {field: data.get(field) for field in _SUBMISSION_FIELDS} + elif row.type == SUBMISSION_UPDATED: + replay.fields.update({field: data[field] for field in _SUBMISSION_FIELDS if field in data}) + elif row.type == BUDGET_COMMITTED: + replay.remaining_cents += data["amount_cents"] + elif row.type == BUDGET_RELEASED: + replay.remaining_cents -= data["amount_cents"] + elif row.type == SPEND_RECORDED: + total = data["spent_cents_total"] + increase = total - replay.spend_high_cents + if increase > 0: + replay.remaining_cents -= min(increase, replay.remaining_cents) + replay.spend_high_cents = total + replay.spend_total_cents = total + replay.reported_ids = frozenset(reported) + return replay + + +def _enqueue( + session: Session, + *, + event_uuid: uuid.UUID, + type: str, + external_id: str, + data: dict, + occurred_at: datetime, + state: QuiltOutboxState = QuiltOutboxState.PENDING, + note: str | None = None, +) -> None: + """Insert one outbox row. A row with the same id already there is left as it is.""" + session.flush() + session.execute( + insert(QuiltOutbox) + .values( + id=event_uuid, + type=type, + external_id=external_id, + payload=data, + occurred_at=occurred_at, + state=state, + last_error=note, + ) + .on_conflict_do_nothing(index_elements=["id"]) + ) + + +def _submitted_revisions(session: Session, project: Project) -> list[ProjectRevision]: + return list( + session.scalars( + sa.select(ProjectRevision) + .where(ProjectRevision.project_id == project.id, ProjectRevision.submitted_at.is_not(None)) + .order_by(ProjectRevision.number) + ) + ) + + +def _requested_cents(project: Project, submitted: list[ProjectRevision]) -> int: + """The budget the submitter asks for now. + + A proposal or amendment still under review wins. Otherwise the approved revision's amount (so a + rejected amendment shows the budget that still stands), else the newest proposal's amount. + """ + requests = [rev for rev in submitted if rev.kind is not RevisionKind.COMPLETION] + pending = [rev for rev in requests if rev.outcome is RevisionOutcome.PENDING] + if pending: + return pending[-1].budget_requested_cents + approved = next((rev for rev in requests if rev.id == project.approved_revision_id), None) + if approved is not None: + return approved.budget_requested_cents + return requests[-1].budget_requested_cents + + +def _submission_fields(project: Project, submitted: list[ProjectRevision]) -> dict: + return { + "title": project.title, + "status": project.status.value, + "requested_cents": _requested_cents(project, submitted), + "url": project_url(project), + } + + +def sync_project(session: Session, project: Project) -> None: + """Add to the outbox every event Quilt hasn't been given yet about `project`. Flushes, doesn't commit. + + Call it after any change to a project's status, title, budget ledger or spend, in the same + transaction. It takes the project's row lock (callers in `krater.services.projects` hold it already), + so two transactions never diff against the same outbox state. A project that was never submitted + gives nothing: Quilt only hears about submissions. + """ + session.flush() + session.execute(sa.select(Project.id).where(Project.id == project.id).with_for_update()) + + submitted = _submitted_revisions(session, project) + proposals = [rev for rev in submitted if rev.kind is RevisionKind.PROPOSAL] + if not proposals: + return + + external_id = str(project.id) + replay = _replay(session, external_id) + now = datetime.now(UTC) + fields = _submission_fields(project, submitted) + + if not replay.created: + submitted_at = proposals[0].submitted_at + applicant_sub = session.scalar(sa.select(User.weave_sub).where(User.id == project.submitter_id)) + _enqueue( + session, + event_uuid=event_id(SUBMISSION_CREATED, project.id), + type=SUBMISSION_CREATED, + external_id=external_id, + data={ + "external_id": external_id, + "applicant_sub": applicant_sub, + **fields, + "submitted_at": submitted_at.astimezone(UTC).isoformat(), + }, + occurred_at=submitted_at, + ) + else: + changed = {field: value for field, value in fields.items() if replay.fields.get(field) != value} + if changed: + _enqueue( + session, + event_uuid=event_id(SUBMISSION_UPDATED, project.id, replay.updated_count), + type=SUBMISSION_UPDATED, + external_id=external_id, + data={"external_id": external_id, **changed}, + occurred_at=now, + ) + + remaining = replay.remaining_cents + entries = session.execute( + sa.select(BudgetEntry, User.weave_sub) + .join(User, User.id == BudgetEntry.actor_id) + .where(BudgetEntry.project_id == project.id) + # Ledger order. Rows of one transaction share `created_at` (Postgres `now()`); put additions first, + # so a release never comes before the commitment it releases. + .order_by(BudgetEntry.created_at, BudgetEntry.amount_cents.desc(), BudgetEntry.id) + ).all() + for entry, actor_sub in entries: + entry_event_id = event_id("budget", entry.id) + if entry_event_id in replay.reported_ids: + continue + remaining = _enqueue_budget_entry( + session, entry, event_uuid=entry_event_id, actor_sub=actor_sub, external_id=external_id, remaining=remaining + ) + + spend = session.scalars( + sa.select(SpendSnapshot) + .where(SpendSnapshot.project_id == project.id) + .order_by(SpendSnapshot.taken_at.desc()) + .limit(1) + ).first() + last_sent = replay.spend_total_cents if replay.spend_total_cents is not None else 0 + if spend is not None and spend.estimated_spend_cents != last_sent: + _enqueue( + session, + event_uuid=event_id("spend", spend.id), + type=SPEND_RECORDED, + external_id=external_id, + data={"external_id": external_id, "spent_cents_total": spend.estimated_spend_cents}, + occurred_at=spend.taken_at, + ) + session.flush() + + +def _enqueue_budget_entry( + session: Session, entry: BudgetEntry, *, event_uuid: uuid.UUID, actor_sub: str, external_id: str, remaining: int +) -> int: + """Add the event for one ledger entry and return Quilt's remaining commitment after it. + + A positive amount is `budget.committed`. A negative amount is `budget.released`, capped at Quilt's + remaining commitment: Quilt refuses a release above it (422 `release_exceeds_commitment`), and it can + be lower than Krater's `ceiling - spend` after spend went past the ceiling or an admin cut the + budget below the spend. A zero amount, or a release with nothing left to release, is stored as + `skipped` so it is never sent later, out of order. + """ + occurred_at = entry.created_at + if entry.amount_cents > 0: + _enqueue( + session, + event_uuid=event_uuid, + type=BUDGET_COMMITTED, + external_id=external_id, + data={"external_id": external_id, "amount_cents": entry.amount_cents, "actor_sub": actor_sub}, + occurred_at=occurred_at, + ) + return remaining + entry.amount_cents + + release = min(-entry.amount_cents, remaining) + if release <= 0: + note = "Zero ledger entry." if entry.amount_cents == 0 else "Nothing left to release in Quilt." + _enqueue( + session, + event_uuid=event_uuid, + type=BUDGET_RELEASED if entry.amount_cents < 0 else BUDGET_COMMITTED, + external_id=external_id, + data={"external_id": external_id, "amount_cents": 0, "actor_sub": actor_sub}, + occurred_at=occurred_at, + state=QuiltOutboxState.SKIPPED, + note=note, + ) + return remaining + if release < -entry.amount_cents: + logger.info( + "quilt: release of ledger entry %s capped at %d cents, Quilt's remaining commitment", entry.id, release + ) + _enqueue( + session, + event_uuid=event_uuid, + type=BUDGET_RELEASED, + external_id=external_id, + data={"external_id": external_id, "amount_cents": release, "actor_sub": actor_sub}, + occurred_at=occurred_at, + ) + return remaining - release + + +def backfill_all(session: Session) -> int: + """Run `sync_project` over every project, oldest first, and return how many outbox rows it added. + + Safe to run again: a second run adds nothing (see the module docstring). Flushes, doesn't commit. + """ + before = session.scalar(sa.select(sa.func.count()).select_from(QuiltOutbox)) or 0 + for project in session.scalars(sa.select(Project).order_by(Project.created_at, Project.id)).all(): + sync_project(session, project) + after = session.scalar(sa.select(sa.func.count()).select_from(QuiltOutbox)) or 0 + return after - before + + +# -------------------------------------------------------------------------------------------------- +# Admin: see and unblock delivery +# -------------------------------------------------------------------------------------------------- + + +@dataclass(frozen=True) +class DeliveryOverview: + """What `/admin` shows about delivery to Quilt.""" + + pending_count: int + #: Pending rows that Quilt answered with an error at least once (they wait for a retry). + retrying: list[QuiltOutbox] + #: Rows Quilt refused for good. Each one blocks the later events of its submission. + failed: list[QuiltOutbox] + + +def delivery_overview(session: Session, actor: Actor) -> DeliveryOverview: + """The outbox rows an admin may need to act on. Admin only.""" + if not actor.is_admin: + raise NotAllowed("Only an admin may see delivery to Quilt.") + pending_count = session.scalar( + sa.select(sa.func.count()).select_from(QuiltOutbox).where(QuiltOutbox.state == QuiltOutboxState.PENDING) + ) + retrying = session.scalars( + sa.select(QuiltOutbox) + .where(QuiltOutbox.state == QuiltOutboxState.PENDING, QuiltOutbox.attempts > 0) + .order_by(QuiltOutbox.seq) + .limit(100) + ).all() + failed = session.scalars( + sa.select(QuiltOutbox).where(QuiltOutbox.state == QuiltOutboxState.FAILED).order_by(QuiltOutbox.seq) + ).all() + return DeliveryOverview(pending_count=int(pending_count or 0), retrying=list(retrying), failed=list(failed)) + + +def _admin_row(session: Session, actor: Actor, row_id: uuid.UUID, reason: str) -> QuiltOutbox: + if not actor.is_admin: + raise NotAllowed("Only an admin may change delivery to Quilt.") + if not (reason and reason.strip()): + raise ValidationFailed({"reason": "A reason is required."}) + row = session.scalars(sa.select(QuiltOutbox).where(QuiltOutbox.id == row_id).with_for_update()).first() + if row is None: + raise NotFound(f"No Quilt event with id {row_id}.") + return row + + +def _audit(session: Session, actor: Actor, action: str, row: QuiltOutbox, reason: str) -> AuditEvent: + project = session.get(Project, uuid.UUID(row.external_id)) + return audit.record( + session, + actor, + action, + project=project, + payload={"event_id": str(row.id), "type": row.type, "last_status": row.last_status}, + reason=reason, + ) + + +def admin_retry(session: Session, actor: Actor, *, row_id: uuid.UUID, reason: str) -> QuiltOutbox: + """Send a failed event again, with the same id and payload, on the next run. Admin only. + + For when the cause was fixed on Quilt's side. Writes an `AuditEvent`. + """ + row = _admin_row(session, actor, row_id, reason) + if row.state is not QuiltOutboxState.FAILED: + raise InvalidState("Only a failed event can be retried.") + row.state = QuiltOutboxState.PENDING + row.next_attempt_at = None + session.flush() + _audit(session, actor, "quilt_event_retry", row, reason) + return row + + +def admin_dismiss(session: Session, actor: Actor, *, row_id: uuid.UUID, reason: str) -> QuiltOutbox: + """Give up on a failed or pending event, so the later events of its submission can go. Admin only. + + The event is never sent. Later releases are capped to what Quilt then has (see `sync_project`). + Writes an `AuditEvent`. + """ + row = _admin_row(session, actor, row_id, reason) + if row.state not in (QuiltOutboxState.FAILED, QuiltOutboxState.PENDING): + raise InvalidState("Only a failed or pending event can be dismissed.") + row.state = QuiltOutboxState.SKIPPED + row.next_attempt_at = None + session.flush() + _audit(session, actor, "quilt_event_dismiss", row, reason) + return row + + +__all__ = [ + "DeliveryOverview", + "admin_dismiss", + "admin_retry", + "delivery_overview", + "BUDGET_COMMITTED", + "BUDGET_RELEASED", + "EVENT_ID_NAMESPACE", + "SPEND_RECORDED", + "SUBMISSION_CREATED", + "SUBMISSION_UPDATED", + "backfill_all", + "event_id", + "project_url", + "sync_project", +] diff --git a/krater/services/skypilot_sync.py b/krater/services/skypilot_sync.py index 93e19f0..21a4f00 100644 --- a/krater/services/skypilot_sync.py +++ b/krater/services/skypilot_sync.py @@ -26,7 +26,7 @@ from sqlalchemy.orm import Session from krater.models import AuditEvent, Project, ProjectStatus, SpendSnapshot, SpendSource, User -from krater.services import audit, budget +from krater.services import audit, budget, quilt_events from krater.services.actor import GROUP_MEMBER from krater.skypilot.client import SkyPilotClient from krater.skypilot.errors import SkyPilotError, SkyPilotWorkspaceNotFoundError @@ -223,6 +223,7 @@ def _tear_down_workspace(session: Session, client: SkyPilotClient, project: Proj ) ) session.flush() + quilt_events.sync_project(session, project) if not already_gone: client.delete_workspace(name) @@ -266,6 +267,7 @@ def sync_spend(session: Session, client: SkyPilotClient) -> None: ) ) session.flush() + quilt_events.sync_project(session, project) def _latest_audit_event(session: Session, project: Project, action: str) -> AuditEvent | None: diff --git a/scripts/quilt_backfill.py b/scripts/quilt_backfill.py new file mode 100644 index 0000000..8f029e6 --- /dev/null +++ b/scripts/quilt_backfill.py @@ -0,0 +1,32 @@ +"""Add Quilt events for every existing project to the `quilt_outbox` table. + +Run it once after the first deploy of the Quilt integration (DATABASE_URL settings as for the app): + + uv run python scripts/quilt_backfill.py + +It runs `krater.services.quilt_events.backfill_all`: for each project, oldest first, it adds the events +Quilt doesn't have yet. Event ids come from the source rows (uuid5), so a second run adds nothing. The +worker sends the rows; see docs/quilt-integration.md. +""" + +from __future__ import annotations + +import sys + +from krater.db import get_sessionmaker +from krater.services.quilt_events import backfill_all + + +def main() -> int: + session = get_sessionmaker()() + try: + added = backfill_all(session) + session.commit() + finally: + session.close() + print(f"quilt backfill: added {added} outbox rows") + return 0 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/tests/services/test_concurrency.py b/tests/services/test_concurrency.py index 83fa906..474b4b9 100644 --- a/tests/services/test_concurrency.py +++ b/tests/services/test_concurrency.py @@ -21,6 +21,7 @@ Project, ProjectRevision, ProjectStatus, + QuiltOutbox, Review, ReviewDecision, ReviewSource, @@ -55,6 +56,7 @@ def _cleanup(engine: Engine, *, project_id: uuid.UUID, user_ids: list[uuid.UUID] ) ) session.execute(sa.delete(BudgetEntry).where(BudgetEntry.project_id == project_id)) + session.execute(sa.delete(QuiltOutbox).where(QuiltOutbox.external_id == str(project_id))) session.execute(sa.delete(ProjectRevision).where(ProjectRevision.project_id == project_id)) session.execute(sa.delete(Project).where(Project.id == project_id)) session.execute(sa.delete(User).where(User.id.in_(user_ids))) diff --git a/tests/services/test_quilt_events.py b/tests/services/test_quilt_events.py new file mode 100644 index 0000000..145b7cb --- /dev/null +++ b/tests/services/test_quilt_events.py @@ -0,0 +1,531 @@ +"""`krater.services.quilt_events`: what Krater tells Quilt, and when. + +Every mapping from `docs/quilt-integration.md`, the same-transaction guarantee, the release cap, the +backfill, and the admin Retry/Dismiss actions. +""" + +from __future__ import annotations + +import uuid +from datetime import UTC, datetime + +import pytest +import sqlalchemy as sa +from sqlalchemy.orm import Session + +from krater.config import get_settings +from krater.models import ( + ApprovalPolicy, + ApprovalStage, + AuditEvent, + BudgetEntry, + Project, + ProjectStatus, + QuiltOutbox, + QuiltOutboxState, + ReviewDecision, + ReviewSource, + SpendSnapshot, + SpendSource, +) +from krater.services import projects, quilt_events +from krater.services.actor import Actor +from krater.services.errors import InvalidState, NotAllowed, ValidationFailed +from krater.services.quilt_events import ( + BUDGET_COMMITTED, + BUDGET_RELEASED, + SPEND_RECORDED, + SUBMISSION_CREATED, + SUBMISSION_UPDATED, + event_id, +) +from krater.services.skypilot_sync import sync_spend, sync_workspaces +from krater.skypilot.fake import FakeSkyPilotClient + + +def _rows(session: Session, project: Project) -> list[QuiltOutbox]: + session.flush() + return list( + session.scalars( + sa.select(QuiltOutbox).where(QuiltOutbox.external_id == str(project.id)).order_by(QuiltOutbox.seq) + ) + ) + + +def _types(session: Session, project: Project) -> list[str]: + return [row.type for row in _rows(session, project)] + + +def _submit(session: Session, member: Actor, *, budget_cents: int = 100_000, title: str = "Rover") -> Project: + project = projects.create_project( + session, member, title=title, write_up="A rover.", budget_requested_cents=budget_cents + ) + return projects.submit(session, member, project=project) + + +def _approve(session: Session, reviewer: Actor, project: Project) -> Project: + projects.record_review( + session, reviewer, revision=project.current_revision, decision=ReviewDecision.APPROVE, source=ReviewSource.WEB + ) + session.refresh(project) + return project + + +def _approved(session: Session, member: Actor, reviewer: Actor, *, budget_cents: int = 100_000) -> Project: + return _approve(session, reviewer, _submit(session, member, budget_cents=budget_cents)) + + +def _spend(session: Session, project: Project, cents: int) -> None: + session.add( + SpendSnapshot(project_id=project.id, estimated_spend_cents=cents, source=SpendSource.SKYPILOT_COST_REPORT) + ) + session.flush() + quilt_events.sync_project(session, project) + + +def _entry(session: Session, project: Project) -> BudgetEntry: + return session.scalars( + sa.select(BudgetEntry).where(BudgetEntry.project_id == project.id).order_by(BudgetEntry.created_at.desc()) + ).first() + + +# -------------------------------------------------------------------------------------------------- +# submission.created / submission.updated +# -------------------------------------------------------------------------------------------------- + + +def test_a_draft_sends_nothing(db_session: Session, member: Actor) -> None: + project = projects.create_project(db_session, member, title="Rover", write_up="x", budget_requested_cents=5) + quilt_events.sync_project(db_session, project) + + assert _rows(db_session, project) == [] + + +def test_withdrawing_a_draft_sends_nothing(db_session: Session, member: Actor) -> None: + project = projects.create_project(db_session, member, title="Rover", write_up="x", budget_requested_cents=5) + projects.withdraw(db_session, member, project=project) + + assert _rows(db_session, project) == [] + + +def test_the_first_submission_sends_submission_created(db_session: Session, member: Actor) -> None: + project = _submit(db_session, member, budget_cents=123_45) + + [row] = _rows(db_session, project) + revision = project.current_revision + assert row.type == SUBMISSION_CREATED + assert row.id == event_id(SUBMISSION_CREATED, project.id) + assert row.state is QuiltOutboxState.PENDING + assert row.payload == { + "external_id": str(project.id), + "applicant_sub": member.user.weave_sub, + "title": "Rover", + "status": "pending_review", + "requested_cents": 123_45, + "url": f"{get_settings().base_url.rstrip('/')}/projects/{project.id}", + "submitted_at": revision.submitted_at.astimezone(UTC).isoformat(), + } + assert row.occurred_at == revision.submitted_at + body = row.event_body() + assert body["id"] == str(row.id) + assert body["type"] == SUBMISSION_CREATED + assert datetime.fromisoformat(body["occurred_at"]) == revision.submitted_at + + +def test_a_rejection_and_resubmission_send_updates_not_a_second_created( + db_session: Session, member: Actor, reviewer: Actor +) -> None: + project = _submit(db_session, member, budget_cents=10_000) + projects.record_review( + db_session, + reviewer, + revision=project.current_revision, + decision=ReviewDecision.REJECT, + reason="Too vague.", + source=ReviewSource.WEB, + ) + projects.update_draft(db_session, member, project=project, title="Rover 2", budget_requested_cents=20_000) + projects.submit(db_session, member, project=project) + + rows = _rows(db_session, project) + assert [row.type for row in rows] == [SUBMISSION_CREATED, SUBMISSION_UPDATED, SUBMISSION_UPDATED] + assert rows[1].payload == {"external_id": str(project.id), "status": "changes_requested"} + assert rows[2].payload == { + "external_id": str(project.id), + "title": "Rover 2", + "status": "pending_review", + "requested_cents": 20_000, + } + assert rows[1].id == event_id(SUBMISSION_UPDATED, project.id, 0) + assert rows[2].id == event_id(SUBMISSION_UPDATED, project.id, 1) + + +def test_approval_sends_the_status_and_budget_committed(db_session: Session, member: Actor, reviewer: Actor) -> None: + project = _approved(db_session, member, reviewer, budget_cents=50_000) + + rows = _rows(db_session, project) + assert [row.type for row in rows] == [SUBMISSION_CREATED, SUBMISSION_UPDATED, BUDGET_COMMITTED] + assert rows[1].payload == {"external_id": str(project.id), "status": "approved"} + entry = _entry(db_session, project) + assert rows[2].id == event_id("budget", entry.id) + assert rows[2].payload == { + "external_id": str(project.id), + "amount_cents": 50_000, + "actor_sub": reviewer.user.weave_sub, + } + assert rows[2].occurred_at == entry.created_at + + +def test_an_approval_that_does_not_meet_the_policy_yet_sends_nothing( + db_session: Session, member: Actor, reviewer: Actor +) -> None: + db_session.add(ApprovalPolicy(stage=ApprovalStage.PROPOSAL, min_approvals=2)) + project = _submit(db_session, member) + before = _types(db_session, project) + _approve(db_session, reviewer, project) + + assert _types(db_session, project) == before + + +def test_an_amendment_increase_sends_the_new_request_then_the_added_budget( + db_session: Session, member: Actor, reviewer: Actor +) -> None: + project = _approved(db_session, member, reviewer, budget_cents=100_000) + projects.start_amendment(db_session, member, project=project) + projects.update_draft(db_session, member, project=project, budget_requested_cents=150_000) + projects.submit(db_session, member, project=project) + _approve(db_session, reviewer, project) + + rows = _rows(db_session, project)[3:] + assert [row.type for row in rows] == [SUBMISSION_UPDATED, BUDGET_COMMITTED] + assert rows[0].payload == {"external_id": str(project.id), "requested_cents": 150_000} + assert rows[1].payload["amount_cents"] == 50_000 + + +def test_an_amendment_decrease_sends_budget_released(db_session: Session, member: Actor, reviewer: Actor) -> None: + project = _approved(db_session, member, reviewer, budget_cents=100_000) + projects.start_amendment(db_session, member, project=project) + projects.update_draft(db_session, member, project=project, budget_requested_cents=60_000) + projects.submit(db_session, member, project=project) + _approve(db_session, reviewer, project) + + last = _rows(db_session, project)[-1] + assert last.type == BUDGET_RELEASED + assert last.payload == { + "external_id": str(project.id), + "amount_cents": 40_000, + "actor_sub": reviewer.user.weave_sub, + } + + +def test_an_unchanged_amendment_budget_is_skipped(db_session: Session, member: Actor, reviewer: Actor) -> None: + project = _approved(db_session, member, reviewer, budget_cents=100_000) + projects.start_amendment(db_session, member, project=project) + projects.update_draft(db_session, member, project=project, write_up="A bigger rover.") + projects.submit(db_session, member, project=project) + _approve(db_session, reviewer, project) + + zero = db_session.scalars( + sa.select(BudgetEntry).where(BudgetEntry.project_id == project.id, BudgetEntry.amount_cents == 0) + ).one() + last = _rows(db_session, project)[-1] + assert last.id == event_id("budget", zero.id) + assert last.state is QuiltOutboxState.SKIPPED + + +def test_a_rejected_amendment_puts_the_approved_request_back( + db_session: Session, member: Actor, reviewer: Actor +) -> None: + project = _approved(db_session, member, reviewer, budget_cents=100_000) + projects.start_amendment(db_session, member, project=project) + projects.update_draft(db_session, member, project=project, budget_requested_cents=150_000) + projects.submit(db_session, member, project=project) + projects.record_review( + db_session, + reviewer, + revision=project.current_revision, + decision=ReviewDecision.REJECT, + reason="No.", + source=ReviewSource.WEB, + ) + + rows = _rows(db_session, project)[3:] + assert [row.payload for row in rows] == [ + {"external_id": str(project.id), "requested_cents": 150_000}, + {"external_id": str(project.id), "requested_cents": 100_000}, + ] + + +def test_admin_adjustments_and_reclaims_follow_the_ledger( + db_session: Session, member: Actor, reviewer: Actor, admin: Actor +) -> None: + project = _approved(db_session, member, reviewer, budget_cents=100_000) + projects.admin_adjust_budget(db_session, admin, project=project, amount_cents=25_000, reason="More GPUs.") + projects.admin_adjust_budget(db_session, admin, project=project, amount_cents=-5_000, reason="Less.") + projects.reclaim_budget(db_session, admin, project=project, amount_cents=20_000, reason="Stalled.") + + rows = _rows(db_session, project)[3:] + assert [(row.type, row.payload["amount_cents"]) for row in rows] == [ + (BUDGET_COMMITTED, 25_000), + (BUDGET_RELEASED, 5_000), + (BUDGET_RELEASED, 20_000), + ] + assert {row.payload["actor_sub"] for row in rows} == {admin.user.weave_sub} + + +def test_completion_sends_the_statuses_and_releases_the_unspent_budget( + db_session: Session, member: Actor, reviewer: Actor +) -> None: + project = _approved(db_session, member, reviewer, budget_cents=100_000) + _spend(db_session, project, 30_000) + projects.start_completion(db_session, member, project=project) + projects.submit_completion(db_session, member, project=project) + _approve(db_session, reviewer, project) + + rows = _rows(db_session, project)[3:] + assert [(row.type, row.payload) for row in rows] == [ + (SPEND_RECORDED, {"external_id": str(project.id), "spent_cents_total": 30_000}), + (SUBMISSION_UPDATED, {"external_id": str(project.id), "status": "pending_completion_review"}), + (SUBMISSION_UPDATED, {"external_id": str(project.id), "status": "completed"}), + ( + BUDGET_RELEASED, + {"external_id": str(project.id), "amount_cents": 70_000, "actor_sub": reviewer.user.weave_sub}, + ), + ] + + +def test_withdrawal_sends_the_status_and_releases_the_rest(db_session: Session, member: Actor, reviewer: Actor) -> None: + project = _approved(db_session, member, reviewer, budget_cents=100_000) + _spend(db_session, project, 10_000) + projects.withdraw(db_session, member, project=project) + + rows = _rows(db_session, project)[-2:] + assert [(row.type, row.payload.get("status"), row.payload.get("amount_cents")) for row in rows] == [ + (SUBMISSION_UPDATED, "withdrawn", None), + (BUDGET_RELEASED, None, 90_000), + ] + + +def test_withdrawing_a_pending_submission_sends_only_the_status(db_session: Session, member: Actor) -> None: + project = _submit(db_session, member) + projects.withdraw(db_session, member, project=project) + + assert _types(db_session, project) == [SUBMISSION_CREATED, SUBMISSION_UPDATED] + + +# -------------------------------------------------------------------------------------------------- +# spend.recorded +# -------------------------------------------------------------------------------------------------- + + +def test_spend_is_sent_as_the_cumulative_total_only_when_it_changes( + db_session: Session, member: Actor, reviewer: Actor +) -> None: + project = _approved(db_session, member, reviewer) + _spend(db_session, project, 1_000) + _spend(db_session, project, 1_000) + _spend(db_session, project, 2_500) + + spend_rows = [row for row in _rows(db_session, project) if row.type == SPEND_RECORDED] + assert [row.payload["spent_cents_total"] for row in spend_rows] == [1_000, 2_500] + latest = db_session.scalars( + sa.select(SpendSnapshot).where(SpendSnapshot.project_id == project.id).order_by(SpendSnapshot.taken_at.desc()) + ).first() + assert spend_rows[-1].id == event_id("spend", latest.id) + assert spend_rows[-1].occurred_at == latest.taken_at + + +def test_the_spend_reconciler_writes_spend_recorded(db_session: Session, member: Actor, reviewer: Actor, weave) -> None: + client = FakeSkyPilotClient() + project = _approved(db_session, member, reviewer) + sync_workspaces(db_session, client, weave) + client.add_cluster(project.skypilot_workspace, cost_cents=4_200) + + sync_spend(db_session, client) + sync_spend(db_session, client) + + spend_rows = [row for row in _rows(db_session, project) if row.type == SPEND_RECORDED] + assert [row.payload["spent_cents_total"] for row in spend_rows] == [4_200] + + +# -------------------------------------------------------------------------------------------------- +# The release cap +# -------------------------------------------------------------------------------------------------- + + +def test_a_release_is_capped_at_what_quilt_still_has_committed( + db_session: Session, member: Actor, reviewer: Actor, admin: Actor +) -> None: + project = _approved(db_session, member, reviewer, budget_cents=100_000) + _spend(db_session, project, 80_000) # Quilt draws down 80_000; 20_000 stays committed + projects.admin_adjust_budget(db_session, admin, project=project, amount_cents=-50_000, reason="Cut.") + + last = _rows(db_session, project)[-1] + assert (last.type, last.payload["amount_cents"], last.state) == (BUDGET_RELEASED, 20_000, QuiltOutboxState.PENDING) + + # Nothing is left in Quilt, so a further release is stored as skipped and never sent. + projects.reclaim_budget(db_session, admin, project=project, amount_cents=10_000, reason="More cut.") + last = _rows(db_session, project)[-1] + assert last.state is QuiltOutboxState.SKIPPED + assert last.last_error == "Nothing left to release in Quilt." + + +def test_a_skipped_release_is_never_sent_later( + db_session: Session, member: Actor, reviewer: Actor, admin: Actor +) -> None: + project = _approved(db_session, member, reviewer, budget_cents=10_000) + _spend(db_session, project, 10_000) + projects.reclaim_budget(db_session, admin, project=project, amount_cents=5_000, reason="Cut.") + projects.admin_adjust_budget(db_session, admin, project=project, amount_cents=50_000, reason="More.") + + rows = _rows(db_session, project) + assert [(row.type, row.state) for row in rows[-2:]] == [ + (BUDGET_RELEASED, QuiltOutboxState.SKIPPED), + (BUDGET_COMMITTED, QuiltOutboxState.PENDING), + ] + count = len(rows) + quilt_events.sync_project(db_session, project) + assert len(_rows(db_session, project)) == count + + +# -------------------------------------------------------------------------------------------------- +# Transactions, idempotence and the backfill +# -------------------------------------------------------------------------------------------------- + + +def test_the_outbox_row_is_part_of_the_callers_transaction(db_session: Session, member: Actor) -> None: + project = projects.create_project(db_session, member, title="Rover", write_up="x", budget_requested_cents=5) + nested = db_session.begin_nested() + projects.submit(db_session, member, project=project) + assert len(_rows(db_session, project)) == 1 + + nested.rollback() + + assert _rows(db_session, project) == [] + db_session.refresh(project) + assert project.status is ProjectStatus.DRAFT + + +def test_a_failed_submit_writes_nothing(db_session: Session, member: Actor) -> None: + project = projects.create_project(db_session, member, title="", write_up="x", budget_requested_cents=5) + with pytest.raises(ValidationFailed): + projects.submit(db_session, member, project=project) + + assert _rows(db_session, project) == [] + + +def test_sync_is_idempotent(db_session: Session, member: Actor, reviewer: Actor) -> None: + project = _approved(db_session, member, reviewer) + _spend(db_session, project, 500) + count = len(_rows(db_session, project)) + + quilt_events.sync_project(db_session, project) + quilt_events.sync_project(db_session, project) + + assert len(_rows(db_session, project)) == count + + +def test_backfill_enqueues_existing_projects_once( + db_session: Session, member: Actor, reviewer: Actor, admin: Actor +) -> None: + draft = projects.create_project(db_session, member, title="Draft", write_up="x", budget_requested_cents=5) + completed = _approved(db_session, member, reviewer, budget_cents=100_000) + _spend(db_session, completed, 40_000) + projects.start_completion(db_session, member, project=completed) + projects.submit_completion(db_session, member, project=completed) + _approve(db_session, reviewer, completed) + pending = _submit(db_session, member, title="Pending") + live_budget_ids = {row.id for row in _rows(db_session, completed) if row.type.startswith("budget.")} + + # As if these projects existed before the Quilt integration. + db_session.execute(sa.delete(QuiltOutbox).where(QuiltOutbox.external_id.in_([str(completed.id), str(pending.id)]))) + + assert quilt_events.backfill_all(db_session) == 5 + assert _rows(db_session, draft) == [] + assert [(row.type, row.payload.get("status")) for row in _rows(db_session, pending)] == [ + (SUBMISSION_CREATED, "pending_review") + ] + rows = _rows(db_session, completed) + assert [row.type for row in rows] == [SUBMISSION_CREATED, BUDGET_COMMITTED, BUDGET_RELEASED, SPEND_RECORDED] + assert rows[0].payload["status"] == "completed" + assert rows[0].payload["requested_cents"] == 100_000 + assert [row.payload["amount_cents"] for row in rows[1:3]] == [100_000, 60_000] + assert rows[3].payload["spent_cents_total"] == 40_000 + # The same ids the live path uses, so Quilt sees a duplicate, not a new event. + assert {row.id for row in rows[1:3]} == live_budget_ids + + assert quilt_events.backfill_all(db_session) == 0 + + +# -------------------------------------------------------------------------------------------------- +# Admin: overview, Retry, Dismiss +# -------------------------------------------------------------------------------------------------- + + +def _failed_row(session: Session, member: Actor) -> QuiltOutbox: + project = _submit(session, member) + [row] = _rows(session, project) + row.state = QuiltOutboxState.FAILED + row.last_status = 422 + row.last_error = "invalid_event" + session.flush() + return row + + +def test_only_an_admin_sees_the_delivery_overview(db_session: Session, member: Actor, reviewer: Actor) -> None: + with pytest.raises(NotAllowed): + quilt_events.delivery_overview(db_session, reviewer) + + +def test_the_overview_lists_failed_and_retrying_rows(db_session: Session, member: Actor, admin: Actor) -> None: + failed = _failed_row(db_session, member) + retrying = _rows(db_session, _submit(db_session, member, title="Other"))[0] + retrying.attempts = 2 + db_session.flush() + + overview = quilt_events.delivery_overview(db_session, admin) + + assert failed in overview.failed + assert retrying in overview.retrying + assert overview.pending_count >= 1 + + +def test_admin_retry_puts_a_failed_row_back_and_audits_it(db_session: Session, member: Actor, admin: Actor) -> None: + row = _failed_row(db_session, member) + + quilt_events.admin_retry(db_session, admin, row_id=row.id, reason="Fixed in Quilt.") + + assert row.state is QuiltOutboxState.PENDING + assert row.next_attempt_at is None + event = db_session.scalars(sa.select(AuditEvent).where(AuditEvent.action == "quilt_event_retry")).one() + assert event.payload["event_id"] == str(row.id) + assert event.project_id == uuid.UUID(row.external_id) + assert event.reason == "Fixed in Quilt." + + +def test_admin_dismiss_skips_a_row_and_audits_it(db_session: Session, member: Actor, admin: Actor) -> None: + row = _failed_row(db_session, member) + + quilt_events.admin_dismiss(db_session, admin, row_id=row.id, reason="Not needed.") + + assert row.state is QuiltOutboxState.SKIPPED + assert db_session.scalars(sa.select(AuditEvent).where(AuditEvent.action == "quilt_event_dismiss")).one() + + +def test_admin_actions_need_an_admin_a_reason_and_the_right_state( + db_session: Session, member: Actor, reviewer: Actor, admin: Actor +) -> None: + row = _failed_row(db_session, member) + + with pytest.raises(NotAllowed): + quilt_events.admin_retry(db_session, reviewer, row_id=row.id, reason="x") + with pytest.raises(NotAllowed): + quilt_events.admin_dismiss(db_session, reviewer, row_id=row.id, reason="x") + with pytest.raises(ValidationFailed): + quilt_events.admin_retry(db_session, admin, row_id=row.id, reason=" ") + + quilt_events.admin_dismiss(db_session, admin, row_id=row.id, reason="x") + with pytest.raises(InvalidState): + quilt_events.admin_retry(db_session, admin, row_id=row.id, reason="x") + with pytest.raises(InvalidState): + quilt_events.admin_dismiss(db_session, admin, row_id=row.id, reason="x") From c356ffcf54a2b98763036ff76c84216714b993a7 Mon Sep 17 00:00:00 2001 From: Jasper Mayone Date: Sat, 10 Oct 2026 16:18:28 -0400 Subject: [PATCH 4/7] feat: send the quilt outbox from the worker --- krater/config.py | 10 + krater/quilt/__init__.py | 12 + krater/quilt/client.py | 50 ++++ krater/quilt/sender.py | 198 +++++++++++++ krater/web/routers/projects.py | 2 + krater/worker/app.py | 58 ++++ tests/quilt/__init__.py | 0 tests/quilt/test_sender.py | 468 +++++++++++++++++++++++++++++++ tests/test_config.py | 10 + tests/worker/test_quilt_tasks.py | 79 ++++++ 10 files changed, 887 insertions(+) create mode 100644 krater/quilt/__init__.py create mode 100644 krater/quilt/client.py create mode 100644 krater/quilt/sender.py create mode 100644 tests/quilt/__init__.py create mode 100644 tests/quilt/test_sender.py create mode 100644 tests/worker/test_quilt_tasks.py diff --git a/krater/config.py b/krater/config.py index bd3fd75..195040a 100644 --- a/krater/config.py +++ b/krater/config.py @@ -98,6 +98,14 @@ class Settings(BaseSettings): s3_access_key_id: str = "" s3_secret_access_key: str = "" + # Quilt (Patchwork Labs finance): Krater sends submission and budget events to Quilt's patch API. See + # docs/quilt-integration.md. Blank means "don't send": the services still write every event to the + # `quilt_outbox` table, so nothing is lost, and the sender sends the backlog once this is set. + quilt_url: str = "" + quilt_timeout_seconds: float = 10.0 + # The most outbox rows one sender run sends. The periodic task runs every minute. + quilt_batch_size: int = 200 + @model_validator(mode="after") def _validate_production_safety(self) -> Settings: if self.env == "production": @@ -121,6 +129,8 @@ def _validate_production_safety(self) -> Settings: f"KRATER_SECRET_KEY must be set to a non-default value of at least " f"{MIN_SECRET_KEY_LENGTH} characters when KRATER_ENV=production" ) + if self.quilt_url and not self.quilt_url.startswith("https://"): + raise ValueError("KRATER_QUILT_URL must be an https:// URL when KRATER_ENV=production") if not self.base_url.startswith("https://"): raise ValueError("KRATER_BASE_URL must be an https:// URL when KRATER_ENV=production") # The live modes above are mandatory in production, so their secrets must actually be set -- diff --git a/krater/quilt/__init__.py b/krater/quilt/__init__.py new file mode 100644 index 0000000..22cdc33 --- /dev/null +++ b/krater/quilt/__init__.py @@ -0,0 +1,12 @@ +"""`krater.quilt`: the only code that talks to Quilt (Patchwork Labs finance) or knows its URLs. + +`QuiltClient` posts one event to Quilt's patch API. `sender.deliver` sends the `quilt_outbox` rows that +`krater.services.quilt_events` writes. The token comes from `krater.weave` (`WeaveClient.quilt_token`). +See `docs/quilt-integration.md`. +""" + +from __future__ import annotations + +from krater.quilt.client import QuiltClient, QuiltResponse + +__all__ = ["QuiltClient", "QuiltResponse"] diff --git a/krater/quilt/client.py b/krater/quilt/client.py new file mode 100644 index 0000000..f852f48 --- /dev/null +++ b/krater/quilt/client.py @@ -0,0 +1,50 @@ +"""`QuiltClient`: `POST {KRATER_QUILT_URL}/api/v1/events`, one event per request. + +The client only reports what Quilt answered. What an answer means (sent, retry, failed) is decided in +`krater.quilt.sender`. See Quilt's `docs/patch-api.md` for the contract. +""" + +from __future__ import annotations + +from dataclasses import dataclass + +import httpx + +EVENTS_PATH = "/api/v1/events" + + +@dataclass(frozen=True) +class QuiltResponse: + """What came back for one event. `status` is `None` when Quilt didn't answer (a network error).""" + + status: int | None + error: str | None = None # Quilt's `error` key, or a description of the network error + body: str = "" + + +class QuiltClient: + """Posts events to Quilt. `http_client` is injectable for tests (`httpx.MockTransport`).""" + + def __init__(self, base_url: str, *, timeout_seconds: float = 10.0, http_client: httpx.Client | None = None): + self._url = f"{base_url.rstrip('/')}{EVENTS_PATH}" + self._http = http_client if http_client is not None else httpx.Client(timeout=timeout_seconds) + + def send_event(self, event: dict, *, token: str) -> QuiltResponse: + try: + response = self._http.post(self._url, json=event, headers={"Authorization": f"Bearer {token}"}) + except httpx.HTTPError as exc: + return QuiltResponse(status=None, error=f"network error: {type(exc).__name__}: {exc}") + error = None + try: + body = response.json() + if isinstance(body, dict) and isinstance(body.get("error"), str): + error = body["error"] + except ValueError: + pass + return QuiltResponse(status=response.status_code, error=error, body=response.text[:2000]) + + def close(self) -> None: + self._http.close() + + +__all__ = ["EVENTS_PATH", "QuiltClient", "QuiltResponse"] diff --git a/krater/quilt/sender.py b/krater/quilt/sender.py new file mode 100644 index 0000000..90f1e35 --- /dev/null +++ b/krater/quilt/sender.py @@ -0,0 +1,198 @@ +"""The sender: delivers `quilt_outbox` rows to Quilt, oldest first, one row per transaction. + +Rules (see `docs/quilt-integration.md`, "Delivery"): + +- Order per subject. A row is sent only when no earlier row of the same `external_id` is still `pending` + or `failed`. So when a row fails, the later rows of that subject wait, in this run and after it. +- Two workers never send the same row: each row is taken with `SELECT ... FOR UPDATE SKIP LOCKED`, and a + row that another worker holds still blocks the later rows of its subject. +- 201 and 200 mean sent. 409 `unknown_submission`, 5xx and network errors are retried with backoff. + 401 and 403 are retried too, but they are a configuration error: the run stops and logs it. + 409 `id_conflict` / `submission_exists`, 413 and 422 fail for good: an admin sees them on `/admin`. +""" + +from __future__ import annotations + +import logging +from collections.abc import Callable +from dataclasses import dataclass +from datetime import UTC, datetime, timedelta +from enum import Enum + +import sqlalchemy as sa +from sqlalchemy.orm import Session, aliased + +from krater.config import Settings +from krater.models import QuiltOutbox, QuiltOutboxState +from krater.quilt.client import QuiltClient, QuiltResponse +from krater.weave import WeaveClient, WeaveUnavailableError + +logger = logging.getLogger(__name__) + +#: The first retry waits this long; each later one waits twice as long, up to `MAX_BACKOFF`. +BASE_BACKOFF = timedelta(seconds=30) +MAX_BACKOFF = timedelta(hours=1) + +#: Quilt's 409 errors that mean "send it again later" (the submission isn't there yet). +_RETRYABLE_CONFLICTS = frozenset({"unknown_submission"}) + + +class Outcome(Enum): + SENT = "sent" + RETRY = "retry" + RETRY_CONFIG = "retry_config" # 401/403: retry later, but stop this run, and tell an operator + FAILED = "failed" + + +def classify(response: QuiltResponse) -> Outcome: + """What one answer from Quilt means for the row.""" + status = response.status + if status in (200, 201): + return Outcome.SENT + if status is None or status >= 500: + return Outcome.RETRY + if status in (401, 403): + return Outcome.RETRY_CONFIG + if status == 409: + return Outcome.RETRY if response.error in _RETRYABLE_CONFLICTS else Outcome.FAILED + if status in (413, 422): + return Outcome.FAILED + # Anything else (a 404 from a wrong KRATER_QUILT_URL, a 400 from a proxy) is not the event's fault. + return Outcome.RETRY_CONFIG + + +def backoff(attempts: int) -> timedelta: + """How long to wait after the `attempts`-th failed attempt.""" + return min(BASE_BACKOFF * (2 ** max(attempts - 1, 0)), MAX_BACKOFF) + + +@dataclass +class DeliveryResult: + sent: int = 0 + retried: int = 0 + failed: int = 0 + stopped: str | None = None # why the run stopped early, if it did + + +def _utcnow() -> datetime: + return datetime.now(UTC) + + +def next_row(session: Session, now: datetime) -> QuiltOutbox | None: + """Lock and return the oldest row that may be sent now, or `None`.""" + earlier = aliased(QuiltOutbox) + blocked = ( + sa.exists() + .where( + earlier.external_id == QuiltOutbox.external_id, + earlier.seq < QuiltOutbox.seq, + earlier.state.in_((QuiltOutboxState.PENDING, QuiltOutboxState.FAILED)), + ) + .correlate(QuiltOutbox) + ) + stmt = ( + sa.select(QuiltOutbox) + .where( + QuiltOutbox.state == QuiltOutboxState.PENDING, + sa.or_(QuiltOutbox.next_attempt_at.is_(None), QuiltOutbox.next_attempt_at <= now), + ~blocked, + ) + .order_by(QuiltOutbox.seq) + .limit(1) + .with_for_update(skip_locked=True, of=QuiltOutbox) + ) + return session.scalars(stmt).first() + + +def _record(row: QuiltOutbox, response: QuiltResponse, outcome: Outcome, now: datetime) -> None: + row.attempts += 1 + row.last_status = response.status + if outcome is Outcome.SENT: + row.state = QuiltOutboxState.SENT + row.sent_at = now + row.next_attempt_at = None + row.last_error = None + return + row.last_error = response.error or (response.body[:500] if response.body else None) + if outcome is Outcome.FAILED: + row.state = QuiltOutboxState.FAILED + row.next_attempt_at = None + logger.error( + "quilt refused event %s (%s for %s): %s %s; an admin must look at /admin", + row.id, + row.type, + row.external_id, + response.status, + row.last_error, + ) + else: + row.next_attempt_at = now + backoff(row.attempts) + + +def deliver( + session: Session, + client: QuiltClient, + weave_client: WeaveClient, + *, + limit: int, + now: Callable[[], datetime] = _utcnow, +) -> DeliveryResult: + """Send up to `limit` due rows, committing after each one. Returns what happened.""" + result = DeliveryResult() + try: + token = weave_client.quilt_token() + except WeaveUnavailableError: + logger.warning("quilt: no `quilt` token from Weave; the outbox waits for the next run", exc_info=True) + result.stopped = "no_token" + return result + + for _ in range(limit): + current = now() + row = next_row(session, current) + if row is None: + session.commit() + break + response = client.send_event(row.event_body(), token=token) + outcome = classify(response) + _record(row, response, outcome, current) + session.commit() + + if outcome is Outcome.SENT: + result.sent += 1 + elif outcome is Outcome.FAILED: + result.failed += 1 + else: + result.retried += 1 + if outcome is Outcome.RETRY_CONFIG: + if response.status == 401: + weave_client.invalidate_quilt_token() + logger.error( + "quilt answered %s (%s): check KRATER_QUILT_URL, the `quilt` scope on the Krater app in Weave, " + "and the Ganymede patch's Weave client id in Quilt; the outbox waits", + response.status, + response.error, + ) + result.stopped = f"http_{response.status}" + break + return result + + +_disabled_logged = False + + +def run(session: Session, weave_client: WeaveClient, settings: Settings) -> DeliveryResult | None: + """The worker's entry point. Does nothing (and says so once per process) when `KRATER_QUILT_URL` is blank.""" + global _disabled_logged + if not settings.quilt_url: + if not _disabled_logged: + logger.info("quilt: KRATER_QUILT_URL is blank; events stay in quilt_outbox until it is set") + _disabled_logged = True + return None + client = QuiltClient(settings.quilt_url, timeout_seconds=settings.quilt_timeout_seconds) + try: + return deliver(session, client, weave_client, limit=settings.quilt_batch_size) + finally: + client.close() + + +__all__ = ["DeliveryResult", "Outcome", "backoff", "classify", "deliver", "next_row", "run"] diff --git a/krater/web/routers/projects.py b/krater/web/routers/projects.py index f35ca9f..7b574c8 100644 --- a/krater/web/routers/projects.py +++ b/krater/web/routers/projects.py @@ -45,6 +45,7 @@ from krater.web.money import InvalidDollarAmount, cents_to_input, parse_dollars from krater.web.templates import templates from krater.worker.app import ( + kick_quilt_delivery, slack_archive_channel, slack_notify_decision, slack_notify_revision_submitted, @@ -120,6 +121,7 @@ def _success_redirect( db_session.commit() if after_commit is not None: after_commit() + kick_quilt_delivery() flash(request, message, "success") return _redirect_to_project(project_id) diff --git a/krater/worker/app.py b/krater/worker/app.py index 2818d97..ec2f508 100644 --- a/krater/worker/app.py +++ b/krater/worker/app.py @@ -13,6 +13,7 @@ from krater.config import get_settings from krater.db import get_sessionmaker from krater.models import Project, ProjectRevision +from krater.quilt import sender as quilt_sender from krater.services import slack_notify, slack_reviews from krater.services.pricing import refresh_prices from krater.services.skypilot_sync import reconcile @@ -213,6 +214,7 @@ def slack_process_approve(revision_id: str, slack_user_id: str, response_url: st slack_user_id=slack_user_id, response_url=response_url, ) + kick_quilt_delivery() except SlackError: logger.exception("krater.slack_process_approve failed") session.rollback() @@ -234,6 +236,7 @@ def slack_process_reject(revision_id: str, slack_user_id: str, reason: str, resp reason=reason, response_url=response_url, ) + kick_quilt_delivery() except SlackError: logger.exception("krater.slack_process_reject failed") session.rollback() @@ -262,3 +265,58 @@ def slack_reconcile(timestamp: int) -> None: logger.exception("krater.slack_reconcile task failed") finally: session.close() + + +# -------------------------------------------------------------------------------------------------- +# Quilt: send the `quilt_outbox` rows the services wrote. A periodic run every minute, plus a kick +# deferred right after a web action commits, so events usually reach Quilt within seconds. See +# `krater.quilt.sender` and docs/quilt-integration.md. +# -------------------------------------------------------------------------------------------------- + +#: Only one kick waits in the queue at a time; more kicks while one waits add nothing. +QUILT_KICK_LOCK = "quilt_deliver_now" + + +def _quilt_deliver() -> None: + session = get_sessionmaker()() + try: + result = quilt_sender.run(session, get_weave_client(), get_settings()) + if result is not None and (result.sent or result.retried or result.failed): + logger.info( + "krater quilt_deliver: sent %d, retry %d, failed %d", + result.sent, + result.retried, + result.failed, + ) + except Exception: + logger.exception("krater.quilt_deliver failed") + session.rollback() + finally: + session.close() + + +@app.periodic(cron="* * * * *") +@app.task(name="quilt_deliver") +def quilt_deliver(timestamp: int) -> None: + """Send due outbox rows to Quilt. A no-op while `KRATER_QUILT_URL` is blank.""" + del timestamp + _quilt_deliver() + + +@app.task(name="quilt_deliver_now", queueing_lock=QUILT_KICK_LOCK) +def quilt_deliver_now() -> None: + """The same as `quilt_deliver`, deferred after a web action commits.""" + _quilt_deliver() + + +def kick_quilt_delivery() -> None: + """Defer `quilt_deliver_now`, if Quilt is configured. Never fails the caller: the periodic run is + the safety net.""" + if not get_settings().quilt_url: + return + try: + quilt_deliver_now.defer() + except procrastinate.exceptions.AlreadyEnqueued: + pass + except Exception: + logger.warning("krater: could not defer quilt_deliver_now; the periodic run sends the events", exc_info=True) diff --git a/tests/quilt/__init__.py b/tests/quilt/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/tests/quilt/test_sender.py b/tests/quilt/test_sender.py new file mode 100644 index 0000000..0bfb9b4 --- /dev/null +++ b/tests/quilt/test_sender.py @@ -0,0 +1,468 @@ +"""`krater.quilt`: the HTTP client and the sender, against a fake Quilt (`httpx.MockTransport`). + +Each response class from Quilt's contract, the order per subject, the backoff, the token, and +`SELECT ... FOR UPDATE SKIP LOCKED` with two real connections. +""" + +from __future__ import annotations + +import json +import logging +import uuid +from collections.abc import Callable +from datetime import UTC, datetime, timedelta + +import httpx +import pytest +import sqlalchemy as sa +from sqlalchemy.engine import Engine +from sqlalchemy.orm import Session + +from krater.config import Settings +from krater.models import QuiltOutbox, QuiltOutboxState +from krater.quilt import QuiltClient, QuiltResponse +from krater.quilt import sender as sender_module +from krater.quilt.sender import Outcome, backoff, classify, deliver, next_row +from krater.weave import StubWeaveClient, WeaveUnavailableError +from krater.weave.stub import STUB_QUILT_TOKEN + +QUILT_URL = "https://quilt.test" +NOW = datetime(2026, 10, 10, 12, 0, tzinfo=UTC) + + +class FakeQuilt: + """Answers each event with `answers[event id]` (a status and an optional error), else 201.""" + + def __init__(self) -> None: + self.requests: list[httpx.Request] = [] + self.answers: dict[str, tuple[int, str | None]] = {} + self.network_error_for: set[str] = set() + + def handler(self, request: httpx.Request) -> httpx.Response: + self.requests.append(request) + event = json.loads(request.content) + if event["id"] in self.network_error_for: + raise httpx.ConnectError("no route") + status, error = self.answers.get(event["id"], (201, None)) + if error is None: + return httpx.Response(status, json={"status": "applied" if status == 201 else "duplicate"}) + return httpx.Response(status, json={"error": error}) + + def sent_ids(self) -> list[str]: + return [json.loads(r.content)["id"] for r in self.requests] + + +@pytest.fixture(autouse=True) +def _empty_outbox(request: pytest.FixtureRequest) -> None: + """Start each `db_session` test from an empty outbox (rolled back with the rest of the test).""" + if "db_session" in request.fixturenames: + request.getfixturevalue("db_session").execute(sa.delete(QuiltOutbox)) + + +@pytest.fixture(autouse=True) +def _sender_logs(monkeypatch: pytest.MonkeyPatch) -> None: + """Alembic's `fileConfig` (run by the `engine` fixture) disables loggers that already exist.""" + monkeypatch.setattr(sender_module.logger, "disabled", False) + + +@pytest.fixture +def quilt() -> FakeQuilt: + return FakeQuilt() + + +@pytest.fixture +def client(quilt: FakeQuilt) -> QuiltClient: + return QuiltClient(QUILT_URL, http_client=httpx.Client(transport=httpx.MockTransport(quilt.handler))) + + +@pytest.fixture +def add_row(db_session: Session) -> Callable[..., QuiltOutbox]: + def _add(external_id: str, type: str = "submission.updated", **fields) -> QuiltOutbox: + row = QuiltOutbox( + id=uuid.uuid4(), + type=type, + external_id=external_id, + payload={"external_id": external_id, "status": "approved"}, + occurred_at=NOW - timedelta(minutes=5), + **fields, + ) + db_session.add(row) + db_session.flush() + return row + + return _add + + +def _deliver(db_session: Session, client: QuiltClient, weave: StubWeaveClient, *, at: datetime = NOW, limit=50): + return deliver(db_session, client, weave, limit=limit, now=lambda: at) + + +# -------------------------------------------------------------------------------------------------- +# classify / backoff +# -------------------------------------------------------------------------------------------------- + + +@pytest.mark.parametrize( + ("status", "error", "outcome"), + [ + (201, None, Outcome.SENT), + (200, None, Outcome.SENT), + (409, "unknown_submission", Outcome.RETRY), + (409, "id_conflict", Outcome.FAILED), + (409, "submission_exists", Outcome.FAILED), + (422, "invalid_event", Outcome.FAILED), + (422, "release_exceeds_commitment", Outcome.FAILED), + (413, "payload_too_large", Outcome.FAILED), + (401, "invalid_token", Outcome.RETRY_CONFIG), + (403, "insufficient_scope", Outcome.RETRY_CONFIG), + (403, "unknown_client", Outcome.RETRY_CONFIG), + (404, None, Outcome.RETRY_CONFIG), + (500, None, Outcome.RETRY), + (503, "weave_unavailable", Outcome.RETRY), + (None, "network error", Outcome.RETRY), + ], +) +def test_classify(status: int | None, error: str | None, outcome: Outcome) -> None: + assert classify(QuiltResponse(status=status, error=error)) is outcome + + +def test_backoff_doubles_up_to_an_hour() -> None: + assert [backoff(n) for n in (1, 2, 3)] == [timedelta(seconds=30), timedelta(seconds=60), timedelta(seconds=120)] + assert backoff(20) == timedelta(hours=1) + + +# -------------------------------------------------------------------------------------------------- +# The client +# -------------------------------------------------------------------------------------------------- + + +def test_the_client_posts_the_event_as_json_with_the_bearer_token(quilt: FakeQuilt, client: QuiltClient) -> None: + event = {"id": "e1", "type": "spend.recorded", "occurred_at": NOW.isoformat(), "data": {"spent_cents_total": 5}} + + response = client.send_event(event, token="tok") + + assert response.status == 201 + [request] = quilt.requests + assert str(request.url) == f"{QUILT_URL}/api/v1/events" + assert request.headers["authorization"] == "Bearer tok" + assert request.headers["content-type"] == "application/json" + assert json.loads(request.content) == event + + +def test_the_client_reads_quilts_error_key(quilt: FakeQuilt, client: QuiltClient) -> None: + quilt.answers["e1"] = (409, "id_conflict") + + response = client.send_event({"id": "e1"}, token="tok") + + assert (response.status, response.error) == (409, "id_conflict") + + +def test_a_network_error_has_no_status(quilt: FakeQuilt, client: QuiltClient) -> None: + quilt.network_error_for.add("e1") + + response = client.send_event({"id": "e1"}, token="tok") + + assert response.status is None + assert "ConnectError" in response.error + + +# -------------------------------------------------------------------------------------------------- +# deliver: one row per response class +# -------------------------------------------------------------------------------------------------- + + +@pytest.mark.parametrize("status", [201, 200]) +def test_applied_and_duplicate_mark_the_row_sent( + db_session: Session, quilt: FakeQuilt, client: QuiltClient, weave_stub: StubWeaveClient, add_row, status +) -> None: + row = add_row("sub-a") + quilt.answers[str(row.id)] = (status, None) + + result = _deliver(db_session, client, weave_stub) + + assert result.sent == 1 + assert (row.state, row.sent_at, row.attempts, row.last_status) == (QuiltOutboxState.SENT, NOW, 1, status) + [request] = quilt.requests + assert request.headers["authorization"] == f"Bearer {STUB_QUILT_TOKEN}" + assert json.loads(request.content) == row.event_body() + + +@pytest.mark.parametrize( + ("status", "error"), [(409, "unknown_submission"), (503, "weave_unavailable"), (500, None), (None, None)] +) +def test_retryable_answers_keep_the_row_pending_with_backoff( + db_session: Session, quilt: FakeQuilt, client: QuiltClient, weave_stub: StubWeaveClient, add_row, status, error +) -> None: + row = add_row("sub-a") + if status is None: + quilt.network_error_for.add(str(row.id)) + else: + quilt.answers[str(row.id)] = (status, error) + + result = _deliver(db_session, client, weave_stub) + + assert result.retried == 1 + assert row.state is QuiltOutboxState.PENDING + assert row.attempts == 1 + assert row.last_status == status + assert row.next_attempt_at == NOW + timedelta(seconds=30) + + # Not due yet: nothing is sent. + _deliver(db_session, client, weave_stub, at=NOW + timedelta(seconds=10)) + assert len(quilt.requests) == 1 + + # Due again: the same event goes again, and the next wait is longer. + _deliver(db_session, client, weave_stub, at=NOW + timedelta(seconds=31)) + assert quilt.sent_ids() == [str(row.id), str(row.id)] + assert row.attempts == 2 + assert row.next_attempt_at == NOW + timedelta(seconds=31) + timedelta(seconds=60) + + +@pytest.mark.parametrize( + ("status", "error"), + [ + (409, "id_conflict"), + (409, "submission_exists"), + (422, "invalid_event"), + (422, "release_exceeds_commitment"), + (413, "payload_too_large"), + ], +) +def test_permanent_refusals_mark_the_row_failed( + db_session: Session, + quilt: FakeQuilt, + client: QuiltClient, + weave_stub: StubWeaveClient, + add_row, + status, + error, + caplog, +) -> None: + row = add_row("sub-a") + quilt.answers[str(row.id)] = (status, error) + + result = _deliver(db_session, client, weave_stub) + + assert result.failed == 1 + assert (row.state, row.last_status, row.last_error) == (QuiltOutboxState.FAILED, status, error) + assert row.next_attempt_at is None + assert "an admin must look" in caplog.text + + _deliver(db_session, client, weave_stub, at=NOW + timedelta(days=1)) + assert len(quilt.requests) == 1 + + +def test_a_401_retries_later_drops_the_token_and_stops_the_run( + db_session: Session, quilt: FakeQuilt, client: QuiltClient, weave_stub: StubWeaveClient, add_row, caplog +) -> None: + first = add_row("sub-a") + add_row("sub-b") + quilt.answers[str(first.id)] = (401, "invalid_token") + + result = _deliver(db_session, client, weave_stub) + + assert result.stopped == "http_401" + assert quilt.sent_ids() == [str(first.id)] + assert first.state is QuiltOutboxState.PENDING + assert first.next_attempt_at == NOW + timedelta(seconds=30) + assert weave_stub.quilt_token_invalidations == 1 + assert "check KRATER_QUILT_URL" in caplog.text + + +def test_a_403_retries_later_and_keeps_the_token( + db_session: Session, quilt: FakeQuilt, client: QuiltClient, weave_stub: StubWeaveClient, add_row +) -> None: + row = add_row("sub-a") + quilt.answers[str(row.id)] = (403, "unknown_client") + + result = _deliver(db_session, client, weave_stub) + + assert result.stopped == "http_403" + assert row.state is QuiltOutboxState.PENDING + assert weave_stub.quilt_token_invalidations == 0 + + +def test_no_token_touches_no_row( + db_session: Session, quilt: FakeQuilt, client: QuiltClient, weave_stub: StubWeaveClient, add_row, monkeypatch +) -> None: + row = add_row("sub-a") + + def refuse() -> str: + raise WeaveUnavailableError("down") + + monkeypatch.setattr(weave_stub, "quilt_token", refuse) + + result = _deliver(db_session, client, weave_stub) + + assert result.stopped == "no_token" + assert quilt.requests == [] + assert (row.state, row.attempts) == (QuiltOutboxState.PENDING, 0) + + +def test_the_token_is_fetched_once_per_run( + db_session: Session, client: QuiltClient, weave_stub: StubWeaveClient, add_row +) -> None: + add_row("sub-a") + add_row("sub-b") + + _deliver(db_session, client, weave_stub) + + assert weave_stub.quilt_token_requests == 1 + + +def test_a_run_sends_at_most_limit_rows( + db_session: Session, quilt: FakeQuilt, client: QuiltClient, weave_stub: StubWeaveClient, add_row +) -> None: + for n in range(3): + add_row(f"sub-{n}") + + assert _deliver(db_session, client, weave_stub, limit=2).sent == 2 + assert len(quilt.requests) == 2 + + +# -------------------------------------------------------------------------------------------------- +# Order per subject +# -------------------------------------------------------------------------------------------------- + + +def test_rows_go_oldest_first( + db_session: Session, quilt: FakeQuilt, client: QuiltClient, weave_stub: StubWeaveClient, add_row +) -> None: + rows = [add_row("sub-a"), add_row("sub-b"), add_row("sub-a")] + + _deliver(db_session, client, weave_stub) + + assert quilt.sent_ids() == [str(row.id) for row in rows] + + +def test_a_failing_row_holds_back_the_later_rows_of_its_subject_only( + db_session: Session, quilt: FakeQuilt, client: QuiltClient, weave_stub: StubWeaveClient, add_row +) -> None: + a1 = add_row("sub-a", type="submission.created") + a2 = add_row("sub-a", type="budget.committed") + b1 = add_row("sub-b") + quilt.answers[str(a1.id)] = (503, "weave_unavailable") + + _deliver(db_session, client, weave_stub) + + assert quilt.sent_ids() == [str(a1.id), str(b1.id)] + assert a2.state is QuiltOutboxState.PENDING + assert a2.attempts == 0 + + # Before a1 is due again, a2 still waits behind it. + _deliver(db_session, client, weave_stub, at=NOW + timedelta(seconds=5)) + assert quilt.sent_ids() == [str(a1.id), str(b1.id)] + + del quilt.answers[str(a1.id)] + _deliver(db_session, client, weave_stub, at=NOW + timedelta(minutes=1)) + assert quilt.sent_ids()[-2:] == [str(a1.id), str(a2.id)] + assert {a1.state, a2.state} == {QuiltOutboxState.SENT} + + +def test_a_failed_row_blocks_its_subject_until_an_admin_acts( + db_session: Session, quilt: FakeQuilt, client: QuiltClient, weave_stub: StubWeaveClient, add_row +) -> None: + add_row("sub-a", state=QuiltOutboxState.FAILED) + blocked = add_row("sub-a") + + _deliver(db_session, client, weave_stub) + assert quilt.requests == [] + + blocked_before = add_row("sub-b", state=QuiltOutboxState.SKIPPED) + free = add_row("sub-b") + _deliver(db_session, client, weave_stub) + assert quilt.sent_ids() == [str(free.id)] + assert blocked.state is QuiltOutboxState.PENDING + assert blocked_before.state is QuiltOutboxState.SKIPPED + + +def test_sent_and_skipped_rows_are_never_sent_again( + db_session: Session, quilt: FakeQuilt, client: QuiltClient, weave_stub: StubWeaveClient, add_row +) -> None: + add_row("sub-a", state=QuiltOutboxState.SENT) + add_row("sub-a", state=QuiltOutboxState.SKIPPED) + + _deliver(db_session, client, weave_stub) + + assert quilt.requests == [] + + +# -------------------------------------------------------------------------------------------------- +# SKIP LOCKED: two workers, two real connections +# -------------------------------------------------------------------------------------------------- + + +def test_two_workers_never_take_the_same_row_or_skip_ahead_in_a_subject(engine: Engine) -> None: + tag = uuid.uuid4().hex + subject_a, subject_b = f"lock-a-{tag}", f"lock-b-{tag}" + ids = [] + with Session(engine) as setup: + for subject in (subject_a, subject_a, subject_b): + row = QuiltOutbox( + id=uuid.uuid4(), type="submission.updated", external_id=subject, payload={}, occurred_at=NOW + ) + setup.add(row) + setup.flush() + ids.append(row.id) + setup.commit() + a1, a2, b1 = ids + + worker_one = Session(engine) + worker_two = Session(engine) + try: + taken_by_one = next_row(worker_one, NOW) + assert taken_by_one is not None + first_id = taken_by_one.id + + # a1 is locked by worker one: worker two skips it, and a2 must wait behind it, so b1 is next. + taken_by_two = next_row(worker_two, NOW) + assert taken_by_two is not None + second_id = taken_by_two.id + finally: + worker_one.rollback() + worker_two.rollback() + worker_one.close() + worker_two.close() + with Session(engine) as cleanup: + cleanup.execute(sa.delete(QuiltOutbox).where(QuiltOutbox.external_id.in_([subject_a, subject_b]))) + cleanup.commit() + assert (first_id, second_id) == (a1, b1) + assert a2 not in (first_id, second_id) + + +# -------------------------------------------------------------------------------------------------- +# run: the worker's entry point +# -------------------------------------------------------------------------------------------------- + + +def test_run_does_nothing_without_a_quilt_url( + db_session: Session, weave_stub: StubWeaveClient, add_row, monkeypatch, caplog +) -> None: + caplog.set_level(logging.INFO, logger="krater.quilt.sender") + monkeypatch.setattr(sender_module, "_disabled_logged", False) + row = add_row("sub-a") + + assert sender_module.run(db_session, weave_stub, Settings(quilt_url="")) is None + assert sender_module.run(db_session, weave_stub, Settings(quilt_url="")) is None + + assert row.state is QuiltOutboxState.PENDING + assert weave_stub.quilt_token_requests == 0 + assert caplog.text.count("KRATER_QUILT_URL is blank") == 1 + + +def test_run_sends_to_the_configured_url( + db_session: Session, quilt: FakeQuilt, weave_stub: StubWeaveClient, add_row, monkeypatch +) -> None: + row = add_row("sub-a") + real_client = QuiltClient + + def fake_client(base_url: str, **kwargs) -> QuiltClient: + return real_client(base_url, http_client=httpx.Client(transport=httpx.MockTransport(quilt.handler))) + + monkeypatch.setattr(sender_module, "QuiltClient", fake_client) + + result = sender_module.run(db_session, weave_stub, Settings(quilt_url=QUILT_URL + "/")) + + assert result.sent >= 1 + assert row.state is QuiltOutboxState.SENT + assert str(quilt.requests[0].url) == f"{QUILT_URL}/api/v1/events" diff --git a/tests/test_config.py b/tests/test_config.py index 85473b9..06aeff4 100644 --- a/tests/test_config.py +++ b/tests/test_config.py @@ -130,3 +130,13 @@ def test_weave_role_keys_read_from_the_environment(monkeypatch) -> None: def test_bootstrap_admins_setting_is_gone() -> None: assert "bootstrap_admins" not in Settings.model_fields + + +def test_production_accepts_a_blank_or_https_quilt_url() -> None: + Settings(**_PRODUCTION_READY, quilt_url="") + Settings(**_PRODUCTION_READY, quilt_url="https://quilt.patchworklabs.example") + + +def test_production_rejects_a_plain_http_quilt_url() -> None: + with pytest.raises(ValueError, match="KRATER_QUILT_URL"): + Settings(**_PRODUCTION_READY, quilt_url="http://quilt.patchworklabs.example") diff --git a/tests/worker/test_quilt_tasks.py b/tests/worker/test_quilt_tasks.py new file mode 100644 index 0000000..eba5c21 --- /dev/null +++ b/tests/worker/test_quilt_tasks.py @@ -0,0 +1,79 @@ +"""Thin wiring tests for the Quilt procrastinate tasks (`krater/worker/app.py`): registration, the +schedule, the call through to `krater.quilt.sender.run`, and the post-commit kick. The sender itself is +tested in `tests/quilt/`.""" + +from __future__ import annotations + +from unittest.mock import patch + +import procrastinate + +from krater.config import Settings +from krater.worker import app as worker_module +from krater.worker.app import app, kick_quilt_delivery, quilt_deliver, quilt_deliver_now + + +def test_quilt_deliver_runs_every_minute() -> None: + assert "quilt_deliver" in app.tasks + assert app.periodic_registry.periodic_tasks[("quilt_deliver", "")].cron == "* * * * *" + + +def test_the_kick_task_has_a_queueing_lock() -> None: + assert quilt_deliver_now.queueing_lock == worker_module.QUILT_KICK_LOCK + + +def test_both_tasks_call_through_to_the_sender() -> None: + with ( + patch("krater.worker.app.quilt_sender.run") as mock_run, + patch("krater.worker.app.get_weave_client") as mock_weave, + patch("krater.worker.app.get_sessionmaker") as mock_sessionmaker, + ): + session = mock_sessionmaker.return_value.return_value + mock_run.return_value = None + + quilt_deliver.func(timestamp=1) + quilt_deliver_now.func() + + assert mock_run.call_count == 2 + assert mock_run.call_args.args[:2] == (session, mock_weave.return_value) + assert session.close.call_count == 2 + + +def test_a_sender_crash_is_logged_and_rolled_back() -> None: + with ( + patch("krater.worker.app.quilt_sender.run", side_effect=RuntimeError("boom")), + patch("krater.worker.app.get_weave_client"), + patch("krater.worker.app.get_sessionmaker") as mock_sessionmaker, + ): + session = mock_sessionmaker.return_value.return_value + + quilt_deliver.func(timestamp=1) + + session.rollback.assert_called_once() + session.close.assert_called_once() + + +def test_the_kick_does_nothing_without_a_quilt_url() -> None: + with ( + patch("krater.worker.app.get_settings", return_value=Settings(quilt_url="")), + patch.object(quilt_deliver_now, "defer") as mock_defer, + ): + kick_quilt_delivery() + + mock_defer.assert_not_called() + + +def test_the_kick_defers_when_quilt_is_configured_and_never_raises() -> None: + settings = Settings(quilt_url="https://quilt.test") + with ( + patch("krater.worker.app.get_settings", return_value=settings), + patch.object(quilt_deliver_now, "defer") as mock_defer, + ): + kick_quilt_delivery() + mock_defer.assert_called_once_with() + + mock_defer.side_effect = procrastinate.exceptions.AlreadyEnqueued("waiting") + kick_quilt_delivery() + + mock_defer.side_effect = RuntimeError("db down") + kick_quilt_delivery() From 0509e6a536dbe0ef33e5c77b0ae5122db68f245f Mon Sep 17 00:00:00 2001 From: Jasper Mayone Date: Sat, 10 Oct 2026 16:18:41 -0400 Subject: [PATCH 5/7] feat: show quilt delivery on the admin page --- krater/web/routers/admin.py | 59 +++++++++-- krater/web/templates/admin/index.html | 138 +++++++++++++++++++++++++- tests/web/test_admin_page.py | 103 ++++++++++++++++++- 3 files changed, 290 insertions(+), 10 deletions(-) diff --git a/krater/web/routers/admin.py b/krater/web/routers/admin.py index 417763e..6421e33 100644 --- a/krater/web/routers/admin.py +++ b/krater/web/routers/admin.py @@ -1,20 +1,27 @@ -"""The admin overview: `/admin`, admins only -- every project grouped by status, plus a read-only -table of `ApprovalPolicy` rows.""" +"""The admin overview: `/admin`, admins only -- every project grouped by status, a read-only table of +`ApprovalPolicy` rows, and delivery to Quilt (events Quilt refused, with Retry and Dismiss).""" from __future__ import annotations +import uuid from typing import Annotated import sqlalchemy as sa -from fastapi import APIRouter, Depends, Request +from fastapi import APIRouter, Depends, Form, Request +from fastapi.responses import RedirectResponse from sqlalchemy.orm import Session from krater.db import get_session from krater.models import ApprovalPolicy, Project, ProjectStatus from krater.services import projects as project_service +from krater.services import quilt_events from krater.services.actor import Actor -from krater.web.deps import require_admin +from krater.services.errors import InvalidState, ValidationFailed +from krater.web.csrf import verify_csrf_token +from krater.web.deps import fresh_actor, require_admin +from krater.web.flash import flash from krater.web.templates import templates +from krater.worker.app import kick_quilt_delivery router = APIRouter() @@ -25,8 +32,6 @@ def admin_overview( db_session: Annotated[Session, Depends(get_session)], actor: Annotated[Actor, Depends(require_admin)], ): - del actor # `require_admin` is the gate; nothing here is scoped to who's asking - all_projects = list(db_session.scalars(sa.select(Project).order_by(Project.created_at.desc()))) summaries = [project_service.project_summary(db_session, project=project) for project in all_projects] @@ -41,8 +46,48 @@ def admin_overview( return templates.TemplateResponse( request, "admin/index.html", - {"by_status": by_status, "policies": policies, "statuses": list(ProjectStatus)}, + { + "by_status": by_status, + "policies": policies, + "statuses": list(ProjectStatus), + "quilt": quilt_events.delivery_overview(db_session, actor), + }, ) +def _quilt_action(db_session: Session, request: Request, action, actor: Actor, event_id: uuid.UUID, reason: str): + try: + action(db_session, actor, row_id=event_id, reason=reason) + except (InvalidState, ValidationFailed) as exc: + db_session.rollback() + flash(request, str(exc), "error") + return RedirectResponse("/admin#quilt", status_code=303) + db_session.commit() + kick_quilt_delivery() + flash(request, "Quilt event updated.", "success") + return RedirectResponse("/admin#quilt", status_code=303) + + +@router.post("/admin/quilt/{event_id}/retry", dependencies=[Depends(verify_csrf_token)]) +def quilt_retry( + request: Request, + event_id: uuid.UUID, + db_session: Annotated[Session, Depends(get_session)], + actor: Annotated[Actor, Depends(fresh_actor)], + reason: Annotated[str, Form()] = "", +): + return _quilt_action(db_session, request, quilt_events.admin_retry, actor, event_id, reason) + + +@router.post("/admin/quilt/{event_id}/dismiss", dependencies=[Depends(verify_csrf_token)]) +def quilt_dismiss( + request: Request, + event_id: uuid.UUID, + db_session: Annotated[Session, Depends(get_session)], + actor: Annotated[Actor, Depends(fresh_actor)], + reason: Annotated[str, Form()] = "", +): + return _quilt_action(db_session, request, quilt_events.admin_dismiss, actor, event_id, reason) + + __all__ = ["router"] diff --git a/krater/web/templates/admin/index.html b/krater/web/templates/admin/index.html index b8ed96d..5b898cf 100644 --- a/krater/web/templates/admin/index.html +++ b/krater/web/templates/admin/index.html @@ -1,7 +1,75 @@ {% extends "base.html" %} {% import "_macros.html" as macros %} -{% block title %}Admin{% endblock %} +{% block title %}Admin
+

Delivery to Quilt

+

+ Krater sends submission and budget events to Quilt. {{ quilt.pending_count }} waiting to send. +

+ {% if quilt.failed %} +

Refused by Quilt

+

+ Quilt refused these events. Later events of the same project wait until you retry or dismiss the event. +

+ + + + + + + + + + + {% for row in quilt.failed %} + + + + + + + {% endfor %} + +
EventProjectAnswerAction
{{ row.type }}
{{ row.id }}		{{ row.external_id }}{{ row.last_status }} {{ row.last_error or "" }} +
+ + + + +
+
+ + + + +
+
+ {% endif %} + {% if quilt.retrying %} +

Waiting for a retry

+ + + + + + + + + + + {% for row in quilt.retrying %} + + + + + + + {% endfor %} + +
EventProjectAttemptsLast answer
{{ row.type }}
{{ row.id }}		{{ row.external_id }}{{ row.attempts }}{{ row.last_status or "no answer" }} {{ row.last_error or "" }}
+ {% endif %} +
+{% endblock %} {% block content %}
@@ -63,4 +131,72 @@

Approval policy

{% endif %}
+
+

Delivery to Quilt

+

+ Krater sends submission and budget events to Quilt. {{ quilt.pending_count }} waiting to send. +

+ {% if quilt.failed %} +

Refused by Quilt

+

+ Quilt refused these events. Later events of the same project wait until you retry or dismiss the event. +

+ + + + + + + + + + + {% for row in quilt.failed %} + + + + + + + {% endfor %} + +
EventProjectAnswerAction
{{ row.type }}
{{ row.id }}		{{ row.external_id }}{{ row.last_status }} {{ row.last_error or "" }} +
+ + + + +
+
+ + + + +
+
+ {% endif %} + {% if quilt.retrying %} +

Waiting for a retry

+ + + + + + + + + + + {% for row in quilt.retrying %} + + + + + + + {% endfor %} + +
EventProjectAttemptsLast answer
{{ row.type }}
{{ row.id }}		{{ row.external_id }}{{ row.attempts }}{{ row.last_status or "no answer" }} {{ row.last_error or "" }}
+ {% endif %} +
{% endblock %} diff --git a/tests/web/test_admin_page.py b/tests/web/test_admin_page.py index 4264cf2..2ffed4c 100644 --- a/tests/web/test_admin_page.py +++ b/tests/web/test_admin_page.py @@ -2,10 +2,11 @@ from __future__ import annotations +import sqlalchemy as sa from fastapi.testclient import TestClient -from krater.models import ApprovalPolicy, ApprovalStage -from tests.conftest import ADMIN_SUB, MEMBER_SUB, REVIEWER_SUB +from krater.models import ApprovalPolicy, ApprovalStage, QuiltOutbox, QuiltOutboxState +from tests.conftest import ADMIN_SUB, MEMBER_SUB, REVIEWER_SUB, get_csrf_token def test_admin_page_requires_sign_in(client: TestClient) -> None: @@ -47,3 +48,101 @@ def test_admin_page_lists_configured_policies(client: TestClient, login_as, db_s assert response.status_code == 200 assert "No custom policies configured" not in response.text assert "Proposal" in response.text + + +# -------------------------------------------------------------------------------------------------- +# Delivery to Quilt +# -------------------------------------------------------------------------------------------------- + + +def _failed_quilt_row(db_session, project) -> QuiltOutbox: + row = db_session.scalars(sa.select(QuiltOutbox).where(QuiltOutbox.external_id == str(project.id))).one() + row.state = QuiltOutboxState.FAILED + row.last_status = 409 + row.last_error = "submission_exists" + db_session.flush() + return row + + +def test_admin_page_lists_events_quilt_refused(client: TestClient, login_as, submitted_project, db_session) -> None: + member = login_as(MEMBER_SUB) + row = _failed_quilt_row(db_session, submitted_project(member, title="Refused One")) + + login_as(ADMIN_SUB) + response = client.get("/admin") + + assert response.status_code == 200 + assert "Delivery to Quilt" in response.text + assert "Refused by Quilt" in response.text + assert str(row.id) in response.text + assert "submission_exists" in response.text + + +def test_admin_can_retry_a_refused_event(client: TestClient, login_as, submitted_project, db_session) -> None: + member = login_as(MEMBER_SUB) + row = _failed_quilt_row(db_session, submitted_project(member)) + login_as(ADMIN_SUB) + csrf = get_csrf_token(client.get("/admin").text) + + response = client.post( + f"/admin/quilt/{row.id}/retry", data={"csrf_token": csrf, "reason": "Fixed."}, follow_redirects=False + ) + + assert response.status_code == 303 + assert response.headers["location"] == "/admin#quilt" + db_session.refresh(row) + assert row.state is QuiltOutboxState.PENDING + + +def test_admin_can_dismiss_a_refused_event(client: TestClient, login_as, submitted_project, db_session) -> None: + member = login_as(MEMBER_SUB) + row = _failed_quilt_row(db_session, submitted_project(member)) + login_as(ADMIN_SUB) + csrf = get_csrf_token(client.get("/admin").text) + + response = client.post( + f"/admin/quilt/{row.id}/dismiss", data={"csrf_token": csrf, "reason": "Not needed."}, follow_redirects=False + ) + + assert response.status_code == 303 + db_session.refresh(row) + assert row.state is QuiltOutboxState.SKIPPED + + +def test_a_quilt_action_without_a_reason_changes_nothing( + client: TestClient, login_as, submitted_project, db_session +) -> None: + member = login_as(MEMBER_SUB) + row = _failed_quilt_row(db_session, submitted_project(member)) + login_as(ADMIN_SUB) + csrf = get_csrf_token(client.get("/admin").text) + + response = client.post(f"/admin/quilt/{row.id}/retry", data={"csrf_token": csrf, "reason": ""}) + + assert response.status_code == 200 # back on /admin, with the error flashed + assert "A reason is required" in response.text + db_session.refresh(row) + assert row.state is QuiltOutboxState.FAILED + + +def test_a_non_admin_cannot_retry_or_dismiss(client: TestClient, login_as, submitted_project, db_session) -> None: + member = login_as(MEMBER_SUB) + row = _failed_quilt_row(db_session, submitted_project(member)) + login_as(REVIEWER_SUB) + csrf = get_csrf_token(client.get("/projects/new").text) + + for action in ("retry", "dismiss"): + response = client.post(f"/admin/quilt/{row.id}/{action}", data={"csrf_token": csrf, "reason": "x"}) + assert response.status_code == 403 + db_session.refresh(row) + assert row.state is QuiltOutboxState.FAILED + + +def test_quilt_actions_need_a_csrf_token(client: TestClient, login_as, submitted_project, db_session) -> None: + member = login_as(MEMBER_SUB) + row = _failed_quilt_row(db_session, submitted_project(member)) + login_as(ADMIN_SUB) + + response = client.post(f"/admin/quilt/{row.id}/retry", data={"reason": "x"}) + + assert response.status_code == 403 From f366fa28ebfc1e7d627ac6ed9247a5078ae702e1 Mon Sep 17 00:00:00 2001 From: Jasper Mayone Date: Sat, 10 Oct 2026 16:18:44 -0400 Subject: [PATCH 6/7] docs: describe the quilt integration --- CLAUDE.md | 8 +- docs/SPEC.md | 8 +- docs/quilt-integration.md | 180 ++++++++++++++++++++++++++++++++++++++ 3 files changed, 192 insertions(+), 4 deletions(-) create mode 100644 docs/quilt-integration.md diff --git a/CLAUDE.md b/CLAUDE.md index 28cf635..9f2a959 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -6,7 +6,8 @@ Guidance for Claude Code (and humans) working in this repo. Krater is the Project Ganymede portal (Patchwork Labs): proposal review, compute budget allocation/enforcement, and a public gallery. **Read `docs/SPEC.md` before changing behavior.** Integration contracts live in -`docs/weave-integration.md` (sign-in and roles) and `docs/skypilot-integration.md` (budget enforcement). +`docs/weave-integration.md` (sign-in and roles), `docs/skypilot-integration.md` (budget enforcement) and +`docs/quilt-integration.md` (submission and budget events to Quilt). ## Stack @@ -40,6 +41,7 @@ krater/ models/ SQLAlchemy models, one module per aggregate services/ domain logic (framework-free): approval_policy, projects, budget, audit, users weave/ WeaveClient protocol: OIDC sign-in, directory API, role mapping, dev stub + quilt/ Quilt patch API client and the outbox sender (the only code that knows Quilt's URLs) web/ FastAPI app factory, routers, deps (current user, authz), templates/, static/ worker/ procrastinate app + tasks alembic/ migrations @@ -61,7 +63,9 @@ tests/ mirrors krater/ layout Krater has no role tables, no admin users page and no disable switch: those are done in Weave. - **All Weave access goes through `krater.weave`.** Nothing else imports httpx for Weave or knows Weave's URLs. Krater asks for `openid profile email groups roles slack`, and calls the directory with a client_credentials token - of its own app (scope `directory`). + of its own app (scope `directory`). Its token for Quilt (scope `quilt`) comes from `WeaveClient.quilt_token`. +- **Events for Quilt go through the outbox.** A service that changes a project's status, title, ledger or spend + calls `krater.services.quilt_events.sync_project` in the same transaction. Never post to Quilt from a request. - **Every admin override writes an `AuditEvent`** (who, what, reason). Budget changes write a `BudgetEntry` (append-only; never update or delete ledger rows). - **Stub mode** (`KRATER_WEAVE_MODE=stub`) replaces Weave with a local fixture of fake users for dev and tests. The diff --git a/docs/SPEC.md b/docs/SPEC.md index ed4fd72..7ba5545 100644 --- a/docs/SPEC.md +++ b/docs/SPEC.md @@ -4,7 +4,8 @@ Revised Sep 26, 2026. Supersedes the Sep 24 draft (drafted with @Adam). Roles li Oct 7, 2026; see "Roles & authentication"). The previous draft was checked against the Weave codebase and the SkyPilot docs and source. This revision fixes the places where it assumed capabilities that don't exist. The integration details are in -[weave-integration.md](weave-integration.md) and [skypilot-integration.md](skypilot-integration.md). +[weave-integration.md](weave-integration.md), [skypilot-integration.md](skypilot-integration.md) and +[quilt-integration.md](quilt-integration.md). ## What changed from the Sep 24 draft @@ -238,6 +239,9 @@ Uploads use presigned URLs, and only objects from approved completion revisions - The reconciler warns the project channel at 80% of the ceiling, and tears down the project's clusters and managed jobs at 100%. - **No automatic expiry.** Stalled projects are reclaimed manually by an admin (`BudgetEntry(reclaim)`). +- **Quilt.** Krater tells Quilt (Patchwork Labs finance) about each submission, each ledger entry and each change of + spend, through Quilt's patch API. The events go through an outbox table in the same transaction as the change. + See [quilt-integration.md](quilt-integration.md). ## Admin overrides @@ -257,7 +261,7 @@ arm64; SkyPilot publishes arm64 images, so that works. | Service | What it is | | --- | --- | | `portal` | FastAPI web app: UI, OIDC, Slack endpoints, public gallery, SkyPilot admin-policy endpoint | -| `worker` | Same image, runs background jobs: Slack work, the spend reconciler, reviewer-channel sync | +| `worker` | Same image, runs background jobs: Slack work, the spend reconciler, reviewer-channel sync, events to Quilt | | `db` | Postgres | | `skypilot` | SkyPilot API server with the Vast credentials; out of scope except for its configuration | | `storage` | Temporary self-hosted S3-compatible storage (SeaweedFS) for gallery screenshots, until a provider is chosen | diff --git a/docs/quilt-integration.md b/docs/quilt-integration.md new file mode 100644 index 0000000..6d31351 --- /dev/null +++ b/docs/quilt-integration.md @@ -0,0 +1,180 @@ +# Quilt integration + +Quilt ([patchworklabsorg/quilt](https://github.com/patchworklabsorg/quilt)) is the Patchwork Labs finance app. Krater +is the app of the Ganymede patch. Krater tells Quilt about each submission and about the budget that Krater approves, +releases and spends. Quilt keeps a thin record of each submission and a ledger of the approved budget that is not +spent yet. + +The contract is Quilt's `docs/patch-api.md` (patchworklabsorg/quilt#8). The `quilt` scope comes from +patchworklabsorg/weave#176. + +## How it works + +```mermaid +sequenceDiagram + participant S as Krater service + participant DB as Postgres (quilt_outbox) + participant Wk as Krater worker + participant W as Weave + participant Q as Quilt + S->>DB: domain change + outbox rows (one transaction) + Note over S,DB: rollback removes both + Wk->>DB: SELECT next due row FOR UPDATE SKIP LOCKED + Wk->>W: POST /oauth/token (client_credentials, scope=quilt), cached + W-->>Wk: access_token + Wk->>Q: POST /api/v1/events (Bearer access_token) + Q-->>Wk: 201 / 200 / 409 / 422 / ... + Wk->>DB: sent, retry later, or failed; commit +``` + +1. A service changes a project. In the same transaction it calls `krater.services.quilt_events.sync_project`. +2. `sync_project` compares the project with what the outbox already holds for it. It adds a row for each difference. + If the transaction rolls back, the rows go too. +3. The worker task `quilt_deliver` runs every minute. A web action that commits also defers `quilt_deliver_now`, so + events usually reach Quilt in seconds. +4. The sender (`krater.quilt.sender`) sends due rows, oldest first, one row for each transaction. + +Code map: + +| Code | Job | +| --- | --- | +| `krater.models.QuiltOutbox` | The `quilt_outbox` table. The row id is the event id | +| `krater.services.quilt_events` | The mapping, the release cap, the backfill, and the admin Retry and Dismiss | +| `krater.quilt.client` | `POST {KRATER_QUILT_URL}/api/v1/events`. The only code that knows Quilt's URLs | +| `krater.quilt.sender` | Order, locks, retries and backoff | +| `krater.weave` (`WeaveClient.quilt_token`) | The client_credentials token with the `quilt` scope | +| `krater.worker.app` | `quilt_deliver` (every minute) and `quilt_deliver_now` (the kick after a commit) | + +## What Krater sends + +The `external_id` of a submission is the Krater `Project.id` (a UUID). It stays the same through resubmissions, +amendments and the completion review. The `status` is Krater's project status, for example `pending_review`, +`approved`, `completed` or `withdrawn`. + +| Krater change | Where | Event | Data | +| --- | --- | --- | --- | +| First submission of a proposal | `projects.submit` | `submission.created` | `external_id`, `applicant_sub` (the submitter's Weave sub), `title`, `status`, `requested_cents`, `url`, `submitted_at` (the first submission) | +| Change of status, title, requested amount or link | `submit`, `submit_completion`, `record_review`, `admin_decide`, `withdraw` | `submission.updated` | `external_id` and the changed fields only | +| `BudgetEntry` with a positive amount: `initial_approval`, an amendment that adds budget, an admin adjustment up | `record_review`, `admin_decide`, `admin_adjust_budget` | `budget.committed` | `amount_cents` (the entry's amount), `actor_sub` (the Weave sub of the entry's actor) | +| `BudgetEntry` with a negative amount: an amendment that cuts budget, an admin adjustment down, a `reclaim` (admin, withdrawal, completion) | `record_review`, `admin_decide`, `admin_adjust_budget`, `reclaim_budget`, `withdraw` | `budget.released` | `amount_cents` (the entry's amount, capped, see below), `actor_sub` | +| `BudgetEntry` with amount 0 (an amendment that keeps the budget) | | nothing | The row is stored as `skipped` | +| A new `SpendSnapshot` with a different total | `skypilot_sync.sync_spend` and the final spend at teardown | `spend.recorded` | `spent_cents_total` (the cumulative total) | + +Rules of the mapping: + +- A project that was never submitted sends nothing. A draft that is withdrawn sends nothing. +- `requested_cents` is the amount of the proposal or amendment that is under review. When none is under review, it + is the amount of the approved revision. When there is no approved revision, it is the amount of the newest + proposal. So a rejected amendment puts the approved amount back. +- `url` is `{KRATER_BASE_URL}/projects/{id}`. +- The budget events follow the ledger: one event for each `BudgetEntry`, in ledger order. Entries of one transaction + have the same `created_at`, so additions go before releases. +- `occurred_at` is the time of the source: the first `submitted_at`, the entry's `created_at`, the snapshot's + `taken_at`. For `submission.updated` it is the time of the sync. + +### The release cap + +Quilt refuses a release above its remaining commitment (committed, minus released, minus the part that spend used). +That can be less than Krater's "ceiling minus spend", for example when spend went past the ceiling, or when an admin +cut the budget below the spend. So Krater replays the outbox rows of the submission to find Quilt's remaining +commitment, and sends `min(release, remaining)`. When nothing remains, the row is stored as `skipped` with +`last_error` "Nothing left to release in Quilt.", and it is never sent. A skipped row is never sent later, so the +order stays correct. + +### Event ids + +Event ids are uuid5 values in a fixed namespace (`quilt_events.EVENT_ID_NAMESPACE`). They come from the source rows, +so a retry, a second sync or the backfill always gives the same id: + +| Event | Id from | +| --- | --- | +| `submission.created` | `submission.created:{project id}` | +| `submission.updated` | `submission.updated:{project id}:{n}`, where n counts the earlier updates of the project | +| `budget.committed`, `budget.released` | `budget:{BudgetEntry id}` | +| `spend.recorded` | `spend:{SpendSnapshot id}` | + +The outbox stores the full payload and `occurred_at`. A retry sends the same id and the same payload, so Quilt +answers `200 duplicate` for an event that it applied before. + +## Delivery + +- **Order for each submission.** The sender sends a row only when no earlier row of the same `external_id` is + `pending` or `failed`. When a row fails, the later rows of that submission wait. Other submissions go on. +- **Two workers.** The sender takes each row with `SELECT ... FOR UPDATE SKIP LOCKED`. A row that another worker holds + still blocks the later rows of its submission, so two workers never send the same row, and never send out of order. +- **Token.** The sender gets one token for each run from `WeaveClient.quilt_token`. Weave's answer is cached until + 30 seconds before it expires. If Weave can't give a token, the run stops and no row changes. + +What each answer does: + +| Quilt answers | Row state | What happens | +| --- | --- | --- | +| `201 applied`, `200 duplicate` | `sent` | `sent_at` is set | +| `409 unknown_submission` | `pending` | Retry with backoff. `submission.created` comes first, so this is usually short | +| `5xx`, `503 weave_unavailable`, no answer | `pending` | Retry with backoff | +| `401` | `pending` | Retry with backoff. Krater drops its cached token. The run stops and logs a configuration error | +| `403`, or another unexpected status such as `404` | `pending` | Retry with backoff. The run stops and logs a configuration error | +| `409 id_conflict`, `409 submission_exists`, `413`, `422` | `failed` | No retry. Krater logs an error. The event shows on `/admin` | + +The backoff is 30 seconds after the first failed attempt, then twice as long each time, up to 1 hour +(`next_attempt_at`). Each attempt sets `attempts`, `last_status` and `last_error`. + +### Failed events + +The admin page `/admin` has a "Delivery to Quilt" section. It shows how many rows wait, the rows that wait for a +retry, and the rows that Quilt refused. For a refused row, an admin can: + +- **Retry**: send it again with the same id and payload. Use this when the cause was fixed in Quilt. +- **Dismiss**: never send it. The later events of the submission can then go. Later releases are capped to what Quilt + has (see the release cap). + +Both need a reason and write an `AuditEvent` (`quilt_event_retry`, `quilt_event_dismiss`). + +## When `KRATER_QUILT_URL` is blank + +The services still write every outbox row. The sender does nothing and logs once per worker process that the events +wait. When an operator sets `KRATER_QUILT_URL`, the worker sends the backlog in order. Nothing is lost, and the +backfill is only needed for data from before this feature. + +In production `KRATER_QUILT_URL` must be blank or an `https://` URL. Production still refuses `KRATER_WEAVE_MODE=stub`. +In stub mode the Quilt token is a fixed fake (`stub-quilt-token`), and a real Quilt refuses it. + +## Setup + +1. Deploy patchworklabsorg/weave#176 and patchworklabsorg/quilt#8. +2. In Weave, an admin allows the `quilt` scope on the Krater app. Without it, Weave refuses the token and Krater logs + "Weave refused Krater a client_credentials token for scope 'quilt'". +3. In Weave, a superadmin allows the `introspect` scope on the Quilt app. +4. In Quilt, an admin opens the Ganymede patch, clicks **Edit**, and sets **Weave client id of the patch app** to + Krater's `KRATER_WEAVE_CLIENT_ID`. Without it, Quilt answers `403 unknown_client`. +5. Set `KRATER_QUILT_URL` (for example `https://quilt.patchworklabs.org`) on the portal and the worker, and restart them. +6. Run the backfill once (see below). + +Settings: + +| Setting | Default | Meaning | +| --- | --- | --- | +| `KRATER_QUILT_URL` | blank | Quilt's base URL. Blank means "don't send yet" | +| `KRATER_QUILT_TIMEOUT_SECONDS` | `10` | The HTTP timeout for each event | +| `KRATER_QUILT_BATCH_SIZE` | `200` | The most rows one run sends | + +The Weave token uses the same `KRATER_WEAVE_CLIENT_ID` and `KRATER_WEAVE_CLIENT_SECRET` as sign-in and the directory. + +## Backfill + +```bash +uv run python scripts/quilt_backfill.py +``` + +The script runs `quilt_events.backfill_all`: `sync_project` for each project, oldest first. For a project from before +this feature, it adds `submission.created` with the current state, then one budget event for each ledger entry, then +the latest spend total. Event ids come from the source rows, so a second run adds nothing, and an event that the live +path already wrote is not added again. + +## Limits + +- Quilt sees the current status at the time of the backfill, not the history of statuses. +- When spend goes down (a SkyPilot state reset), Krater sends the lower total. Quilt keeps the higher total and marks + the event "needs review". +- After overspend, Quilt's remaining commitment can differ from Krater's "ceiling minus spend". The release cap keeps + every release valid. Krater's ledger stays the source of truth for the ceiling. From 2756abca5e998b7267ec8e841a9f45f09687b9dd Mon Sep 17 00:00:00 2001 From: Jasper Mayone Date: Sat, 10 Oct 2026 16:23:51 -0400 Subject: [PATCH 7/7] fix: follow quilt's overspend rule in the release cap --- docs/quilt-integration.md | 4 +++- krater/services/quilt_events.py | 9 +++++++-- tests/services/test_quilt_events.py | 13 +++++++++++++ 3 files changed, 23 insertions(+), 3 deletions(-) diff --git a/docs/quilt-integration.md b/docs/quilt-integration.md index 6d31351..2adf660 100644 --- a/docs/quilt-integration.md +++ b/docs/quilt-integration.md @@ -77,7 +77,9 @@ Rules of the mapping: Quilt refuses a release above its remaining commitment (committed, minus released, minus the part that spend used). That can be less than Krater's "ceiling minus spend", for example when spend went past the ceiling, or when an admin cut the budget below the spend. So Krater replays the outbox rows of the submission to find Quilt's remaining -commitment, and sends `min(release, remaining)`. When nothing remains, the row is stored as `skipped` with +commitment, and sends `min(release, remaining)`. The replay follows Quilt's overspend rule: spend above the +commitment is overspend, and a later commitment covers that overspend first, so only the rest adds to the remaining +commitment. When nothing remains, the row is stored as `skipped` with `last_error` "Nothing left to release in Quilt.", and it is never sent. A skipped row is never sent later, so the order stays correct. diff --git a/krater/services/quilt_events.py b/krater/services/quilt_events.py index 20c54f3..9c98208 100644 --- a/krater/services/quilt_events.py +++ b/krater/services/quilt_events.py @@ -85,6 +85,7 @@ class _Replay: remaining_cents: int = 0 # Quilt's remaining commitment: committed - released - spend drawdown spend_total_cents: int | None = None # the last total sent spend_high_cents: int = 0 # Quilt keeps the highest total it has seen + overspend_cents: int = 0 # spend above the commitment; Quilt's next commitment covers it first reported_ids: frozenset[uuid.UUID] = frozenset() @@ -114,14 +115,18 @@ def _replay(session: Session, external_id: str) -> _Replay: elif row.type == SUBMISSION_UPDATED: replay.fields.update({field: data[field] for field in _SUBMISSION_FIELDS if field in data}) elif row.type == BUDGET_COMMITTED: - replay.remaining_cents += data["amount_cents"] + covered = min(data["amount_cents"], replay.overspend_cents) + replay.remaining_cents += data["amount_cents"] - covered + replay.overspend_cents -= covered elif row.type == BUDGET_RELEASED: replay.remaining_cents -= data["amount_cents"] elif row.type == SPEND_RECORDED: total = data["spent_cents_total"] increase = total - replay.spend_high_cents if increase > 0: - replay.remaining_cents -= min(increase, replay.remaining_cents) + drawdown = min(increase, replay.remaining_cents) + replay.remaining_cents -= drawdown + replay.overspend_cents += increase - drawdown replay.spend_high_cents = total replay.spend_total_cents = total replay.reported_ids = frozenset(reported) diff --git a/tests/services/test_quilt_events.py b/tests/services/test_quilt_events.py index 145b7cb..13cd51d 100644 --- a/tests/services/test_quilt_events.py +++ b/tests/services/test_quilt_events.py @@ -388,6 +388,19 @@ def test_a_skipped_release_is_never_sent_later( assert len(_rows(db_session, project)) == count +def test_a_new_commitment_covers_earlier_overspend_before_the_release_cap( + db_session: Session, member: Actor, reviewer: Actor, admin: Actor +) -> None: + project = _approved(db_session, member, reviewer, budget_cents=10_000) + _spend(db_session, project, 12_000) # 2_000 over the commitment + projects.admin_adjust_budget(db_session, admin, project=project, amount_cents=5_000, reason="More.") + # Quilt uses 2_000 of the new 5_000 to cover the overspend, so 3_000 stays committed. + projects.admin_adjust_budget(db_session, admin, project=project, amount_cents=-4_000, reason="Cut.") + + last = _rows(db_session, project)[-1] + assert (last.type, last.payload["amount_cents"], last.state) == (BUDGET_RELEASED, 3_000, QuiltOutboxState.PENDING) + + # -------------------------------------------------------------------------------------------------- # Transactions, idempotence and the backfill # --------------------------------------------------------------------------------------------------