diff --git a/.env.example b/.env.example index c5326aa..b352954 100644 --- a/.env.example +++ b/.env.example @@ -27,6 +27,7 @@ VALKEY_ADDR=localhost:6379 # GH_MEMBERSHIP_CACHE_TTL=300 # seconds # JWT_TTL_SECONDS=900 # deploy-session JWT TTL # VALKEY_PASSWORD= # empty for unauthenticated dev +# VALKEY_CONNECT_RETRY_WINDOW=5s # boot-time retry window for the initial dial; 0 disables retry # REGISTRY_AUTHZ_TEAM=staff # GitHub team allowed to mutate the sites registry # ALIAS_PRODUCTION_KEY_FORMAT=/production # ALIAS_PREVIEW_KEY_FORMAT=/preview diff --git a/.github/workflows/test.yml b/.github/workflows/test.yml index f0c1fbf..850a028 100644 --- a/.github/workflows/test.yml +++ b/.github/workflows/test.yml @@ -28,6 +28,9 @@ jobs: go-version-file: go.mod cache-dependency-path: go.sum + - name: go mod tidy check + run: go mod tidy -diff + - name: go vet run: go vet ./... diff --git a/README.md b/README.md index aafe5a5..753fcb8 100644 --- a/README.md +++ b/README.md @@ -2,7 +2,7 @@ Static-apps deploy proxy for the freeCodeCamp Universe platform. Public hostname: `uploads.freecode.camp`. -Staff devs and CI run `universe deploy`; the artifact lands on R2 behind a Caddy `r2_alias` upstream. Zero R2 tokens reach staff hands or CI secrets — Artemis is the sole holder of the admin S3 token. Identity is GitHub team membership. +Staff developers and CI run `universe static deploy`. The CLI uploads the build artifact to artemis, artemis writes it to R2, and a Caddy `r2_alias` upstream serves it. Staff and CI hold no R2 tokens — artemis is the only holder of the admin S3 token. Caller identity comes from GitHub team membership. ## Quick start @@ -15,10 +15,12 @@ just # list every recipe ## Docs +- **[`docs/ORIENTATION.md`](docs/ORIENTATION.md)** — the read sequence for a new contributor. +- **[`docs/ARCHITECTURE.md`](docs/ARCHITECTURE.md)** — what the service does and how it is built, written from the source code. - **[`docs/README.md`](docs/README.md)** — API contract, configuration, observability, R2 layout, sites registry, integration testing, curl examples. - **[`docs/RELEASING.md`](docs/RELEASING.md)** — versioning rule, release-please flow, image build, downstream deploy pin. -The CLI ↔ artemis contract and per-site authorization model are specified in ADR-016 (Universe platform repo). +ADR-016 (Universe platform repo) specifies the CLI ↔ artemis contract and the per-site authorization model. ## License diff --git a/cmd/artemis/gcworkflows.go b/cmd/artemis/gcworkflows.go index fe4f89f..f1b6b86 100644 --- a/cmd/artemis/gcworkflows.go +++ b/cmd/artemis/gcworkflows.go @@ -8,6 +8,7 @@ import ( "errors" "fmt" "log/slog" + mrand "math/rand/v2" "time" "github.com/freeCodeCamp/artemis/internal/gc" @@ -71,13 +72,39 @@ const ( cronTombstonePurge = "0 3 * * *" cronReconcile = "0 4 * * *" relayInterval = 5 * time.Second - - reconcilePublishFloor = 4 * time.Second - reconcilePublishPerSite = 150 * time.Millisecond ) -func reconcilePublishDeadline(n int) time.Duration { - return reconcilePublishFloor + time.Duration(n)*reconcilePublishPerSite +func publishReconcileEvents(ctx context.Context, publisher worker.Publisher, sites []string, perPublish time.Duration) (int, error) { + shuffled := append([]string(nil), sites...) + mrand.Shuffle(len(shuffled), func(i, j int) { shuffled[i], shuffled[j] = shuffled[j], shuffled[i] }) + var firstErr error + published := 0 + for _, site := range shuffled { + if ctx.Err() != nil { + if firstErr == nil { + firstErr = ctx.Err() + } + break + } + payload, err := json.Marshal(map[string]string{"site": site}) + if err != nil { + if firstErr == nil { + firstErr = err + } + continue + } + pctx, cancel := context.WithTimeout(ctx, perPublish) + err = publisher.Publish(pctx, topicSiteReconcile, payload) + cancel() + if err != nil { + if firstErr == nil { + firstErr = err + } + continue + } + published++ + } + return published, firstErr } func runRelayLoop(ctx context.Context, relay *worker.Relay, interval time.Duration) { @@ -118,28 +145,12 @@ func gcWorkflowDefs(gcw *gcWiring, dryRun bool, publisher worker.Publisher, reco Cron: []string{cronReconcile}, Handler: withCheckIn(workflowReconcileScheduler, cronReconcile, observeWorkflow(workflowReconcileScheduler, func(ctx context.Context, _ map[string]any) error { sites := reconcileSites() - pctx, cancel := context.WithTimeout(ctx, reconcilePublishDeadline(len(sites))) - defer cancel() - var firstErr error - for _, site := range sites { - if pctx.Err() != nil { - if firstErr == nil { - firstErr = pctx.Err() - } - break - } - payload, err := json.Marshal(map[string]string{"site": site}) - if err != nil { - if firstErr == nil { - firstErr = err - } - continue - } - if err := publisher.Publish(pctx, topicSiteReconcile, payload); err != nil { - if firstErr == nil { - firstErr = err - } - } + published, firstErr := publishReconcileEvents(ctx, publisher, sites, worker.DefaultPublishTimeout) + if published < len(sites) { + slog.ErrorContext(ctx, "reconcile.schedule.incomplete", + "sites", len(sites), + "published", published, + "skipped", len(sites)-published) } if firstErr != nil { captureBackground("reconcile.schedule", firstErr) diff --git a/cmd/artemis/gcworkflows_test.go b/cmd/artemis/gcworkflows_test.go index 320385a..bbc54cc 100644 --- a/cmd/artemis/gcworkflows_test.go +++ b/cmd/artemis/gcworkflows_test.go @@ -4,6 +4,7 @@ import ( "context" "encoding/json" "errors" + "fmt" "log/slog" "sync" "testing" @@ -17,6 +18,8 @@ import ( "github.com/jackc/pgx/v5/pgconn" "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" + "sort" + "strings" ) type fakeReaper struct{} @@ -214,6 +217,67 @@ func TestReconcileScheduler_BoundsPublishDeadline(t *testing.T) { assert.LessOrEqual(t, d, 30*time.Second, "publish deadline is bounded, not open-ended") } +type slowPublisher struct { + stallOn string + mu sync.Mutex + sites []string +} + +func (p *slowPublisher) Publish(ctx context.Context, _ string, payload []byte) error { + var m map[string]string + if err := json.Unmarshal(payload, &m); err != nil { + return err + } + if p.stallOn != "" && m["site"] == p.stallOn { + <-ctx.Done() + return ctx.Err() + } + p.mu.Lock() + defer p.mu.Unlock() + p.sites = append(p.sites, m["site"]) + return nil +} + +func (p *slowPublisher) published() []string { + p.mu.Lock() + defer p.mu.Unlock() + return append([]string(nil), p.sites...) +} + +func reconcileSiteNames(n int) []string { + out := make([]string, n) + for i := range out { + out[i] = fmt.Sprintf("site-%03d", i) + } + return out +} + +func TestPublishReconcileEvents_PublishesEverySite(t *testing.T) { + sites := reconcileSiteNames(50) + pub := &slowPublisher{} + + published, err := publishReconcileEvents(context.Background(), pub, sites, 100*time.Millisecond) + + require.NoError(t, err) + assert.Equal(t, len(sites), published, + "every registered site must get a reconcile event; a per-run budget must never truncate the list") + assert.ElementsMatch(t, sites, pub.published()) +} + +func TestPublishReconcileEvents_StalledSiteDoesNotDropTheRest(t *testing.T) { + sites := reconcileSiteNames(20) + pub := &slowPublisher{stallOn: sites[5]} + + published, err := publishReconcileEvents(context.Background(), pub, sites, 20*time.Millisecond) + + require.Error(t, err, "a stalled publish is still reported to the caller") + assert.Equal(t, len(sites)-1, published, + "one stalled site must not stop the sites after it") + assert.NotContains(t, pub.published(), sites[5]) + assert.Contains(t, pub.published(), sites[19], + "the last site in the list must still be reached") +} + func TestGCWorkflowDefs(t *testing.T) { gcw := &gcWiring{SiteGC: &gc.SiteGC{}, Purge: &gc.TombstonePurge{}, Reconciler: &gc.Reconciler{}} defs := gcWorkflowDefs(gcw, true, &capturingPublisher{}, noSites) @@ -259,8 +323,10 @@ func TestReconcileScheduler_PublishesPerSite(t *testing.T) { require.Len(t, pub.topics, 2, "one site.reconcile event published per registered site") assert.Equal(t, []string{topicSiteReconcile, topicSiteReconcile}, pub.topics) - assert.Contains(t, string(pub.payloads[0]), `"site":"www"`) - assert.Contains(t, string(pub.payloads[1]), `"site":"learn"`) + payloads := []string{string(pub.payloads[0]), string(pub.payloads[1])} + sort.Strings(payloads) + assert.Equal(t, []string{`{"site":"learn"}`, `{"site":"www"}`}, payloads, + "every site gets one event; publish order is shuffled by design") } type exhaustingPublisher struct { @@ -461,3 +527,48 @@ func TestRegisterGCWorkflows(t *testing.T) { require.NoError(t, registerGCWorkflows(rt, gcw, false, &capturingPublisher{}, noSites)) assert.Len(t, rt.Registered(), 4) } + +type truncatingPublisher struct { + mu sync.Mutex + sites []string + after int + cancel context.CancelFunc +} + +func (p *truncatingPublisher) Publish(_ context.Context, _ string, payload []byte) error { + p.mu.Lock() + defer p.mu.Unlock() + var m map[string]string + if err := json.Unmarshal(payload, &m); err != nil { + return err + } + p.sites = append(p.sites, m["site"]) + if len(p.sites) >= p.after { + p.cancel() + } + return nil +} + +func TestReconcileScheduler_TruncatedRunsCoverDisjointSuffixes(t *testing.T) { + sites := reconcileSiteNames(40) + const keep = 8 + const runs = 6 + + publishedSets := make([]string, 0, runs) + for range runs { + ctx, cancel := context.WithCancel(context.Background()) + pub := &truncatingPublisher{after: keep, cancel: cancel} + _, _ = publishReconcileEvents(ctx, pub, sites, time.Second) + cancel() + got := append([]string(nil), pub.sites...) + sort.Strings(got) + publishedSets = append(publishedSets, strings.Join(got, ",")) + } + + distinct := map[string]bool{} + for _, s := range publishedSets { + distinct[s] = true + } + require.Greater(t, len(distinct), 1, + "a truncated run must not cover the identical site prefix every time; a fixed order starves the same tail every night") +} diff --git a/cmd/artemis/main_test.go b/cmd/artemis/main_test.go index 2ca41e1..b4132d7 100644 --- a/cmd/artemis/main_test.go +++ b/cmd/artemis/main_test.go @@ -21,7 +21,7 @@ func TestBootMigrations(t *testing.T) { testcontainers.SkipIfProviderIsNotHealthy(t) - container, err := postgres.Run(ctx, "postgres:16-alpine", + container, err := postgres.Run(ctx, testPostgresImage, postgres.WithDatabase("artemis_test"), postgres.WithUsername("artemis"), postgres.WithPassword("artemis"), @@ -46,3 +46,5 @@ func TestBootMigrations(t *testing.T) { require.Truef(t, exists, "table %q must exist after boot migrations", table) } } + +const testPostgresImage = "postgres:16-alpine" diff --git a/docker-compose.yml b/docker-compose.yml index 4509d52..071b4dd 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -2,7 +2,7 @@ name: artemis-local services: postgres: - image: postgres:17-alpine + image: postgres:16-alpine environment: POSTGRES_USER: artemis POSTGRES_PASSWORD: artemis diff --git a/docs/ARCHITECTURE.md b/docs/ARCHITECTURE.md new file mode 100644 index 0000000..31adcd3 --- /dev/null +++ b/docs/ARCHITECTURE.md @@ -0,0 +1,321 @@ +# Artemis — architecture overview + +This document tells you what the artemis service does and how it is built. It is written from the source code, not from the design records. Where this document and an ADR disagree, read the code. + +For the API reference, see [`README.md`](README.md). For the durable-execution rationale, see [`design/0001-durable-execution-model.md`](design/0001-durable-execution-model.md). + +## 1. What artemis is, and the problem it solves + +Artemis is one Go service. It is a deploy proxy for static sites. + +The service keeps the R2 credentials in its own configuration. A developer holds only a GitHub token. The developer never gets the R2 credentials. This is the one problem artemis solves: it gives staff a controlled write path into one object store. + +For each site, artemis writes two kinds of object into R2: + +- the site files, under one key prefix for each deploy; +- a small pointer object, called an **alias**. + +Artemis does not serve the site traffic. Its HTTP surface holds only the `/api/*` routes, plus `/healthz` and `/readyz`. + +The default R2 layout is: + +| Thing | Default R2 key | +| ------------------- | ---------------------------------------------- | +| Deploy files | `/deploys//...` | +| Production alias | `/production` | +| Preview alias | `/preview` | +| Completeness marker | `/deploys//_artemis_meta.json` | +| Trash | `_trash///` | + +## 2. The parts artemis depends on + +| Part | What it holds | Required | +| ------------ | ------------------------------------------------------------------------------------------------------------------ | -------- | +| **R2** | All deploy bytes, the alias objects, and the trash tree | Yes | +| **Valkey** | The site registry when Postgres is absent, the change events, and the GitHub team cache | Yes | +| **Postgres** | The deploy index, the alias rows, the tombstones, the outbox, the audit log, the site registry, and the repo queue | No | +| **GitHub** | The caller identity and the team membership | Yes | +| **Hatchet** | The background workflow runs | No | +| **Sentry** | The errors and the cron check-ins | No | + +Three switches control the optional parts: + +- `DATABASE_URL` starts Postgres and all the work that needs it. +- `HATCHET_ADDR`, together with Postgres, starts the worker and the event relay. +- `SENTRY_DSN` starts the error reporting. + +Valkey is not optional. The configuration rejects an empty `VALKEY_ADDR`, and the boot code always dials Valkey. + +Postgres is optional, but many functions need it. With no Postgres, artemis keeps no deploy index, writes no audit rows, runs no cleanup, and takes no site lock. + +## 3. The deploy lifecycle of a developer + +A deploy has three calls. Artemis makes the deploy id, and the client never chooses one. + +### Step 1 — init + +`POST /api/deploy/init` with a GitHub bearer token and a body of `{site, sha}`. + +1. Artemis reads the caller login from the GitHub token. +1. Artemis reads the teams of the site from the cached registry. An empty team list gives 403. +1. Artemis asks GitHub if the caller is a member of one of those teams. A negative answer gives 403. +1. Artemis makes the deploy id. The shape is `-`, in UTC. +1. Artemis signs a **deploy-session JWT**. The token holds the login, the site, and the deploy id. The default life is 15 minutes. + +Init writes nothing to R2, and the deploy prefix does not exist yet. Init does write one `audit_log` row in Postgres. That write is best effort, and a failure does not fail the request. + +### Step 2 — upload + +`PUT /api/deploy/{deployId}/upload?path=`, one call for each file, with the JWT. + +1. Artemis checks that the deploy id in the URL agrees with the deploy id in the JWT. +1. Artemis validates the `?path=` value as received: an absolute path, a non-canonical path, a `..` segment, a control byte, and a backslash are each rejected with 400. +1. Artemis streams the request body straight into R2 at ``. + +There is no staging area. Each object lands at its final key immediately. The default limit is 100 MiB for each file. + +> **Note.** Step 2 used to remove the leading `/` *before* it examined the path, so `?path=/index.html` was silently rewritten to `index.html`. The raw value is now validated first, and an absolute path gives 400. + +### Step 3 — finalize + +`POST /api/deploy/{deployId}/finalize` with `{mode, files}`. The mode is `preview` or `production`. + +Artemis runs these gates in order, and it stops at the first failure: + +| Order | Gate | Failure | +| ----- | ---------------------------------------------- | -------------------------- | +| 1 | The mode must be `preview` or `production` | 400 | +| 2 | The file manifest must not be empty | 400 | +| 3 | The manifest must hold a root `index.html` | 422 | +| 4 | R2 must hold every file in the manifest | 422, with the missing list | +| 5 | Artemis writes the `_artemis_meta.json` marker | 502 | +| 6 | Artemis measures the deploy size | not fatal | + +Artemis then takes a **per-site lock** in Postgres, and it does the last steps inside that lock: + +7. Artemis reads the site from the registry again. A deleted site gives 410. +1. Artemis writes the alias object. **This one PUT makes the deploy live.** +1. Artemis writes one Postgres transaction. The transaction adds or updates the deploy row, marks the previous alias target as released, adds or updates the alias row, and puts a `site.changed` event in the outbox. + +The response holds the public URL, the deploy id, and the mode. + +Two properties follow from this order: + +- Each failure before step 8 leaves the alias untouched. A partial upload never becomes live. +- The alias write and the Postgres write are not one atomic unit. If step 9 fails, the alias already points at the new deploy, and the client gets 502. + +## 4. What a deployment alias is + +An alias is one small R2 object. Its body is the deploy id string, and nothing more. Its content type is `text/plain`. + +Each site has two aliases: + +- `/production` +- `/preview` + +The alias is the pointer that tells the edge which deploy to serve. To make a deploy live, artemis replaces the body of one alias object. + +Artemis depends on R2 to make one PUT atomic for each key. No code in this repository can prove that property. It is an assumption of the design, and the safety of every alias write depends on it. + +### Promote and rollback + +Both verbs move the **production** alias. Both hold the per-site lock. Both refuse to point the alias at a deploy that has no root `index.html`. + +| | `POST .../promote` | `POST .../rollback` | +| ------------------------------ | --------------------------------------------- | ----------------------------------------- | +| Body | Optional | Required | +| Target | `deployId`, or else the current preview alias | `to` (required) | +| Examines the deploy prefix | No | Yes (422 `deploy_missing`) | +| Examines the root `index.html` | Yes | Yes | +| Order of the checks | Guard first, then the index check | Existence and index first, then the guard | + +Both verbs accept an optional `expectedCurrent` field. Artemis reads the current alias, and it gives 409 `alias_drift` on a mismatch. The guard is a read and then a write. Artemis never sets the `IfMatch` or `IfNoneMatch` fields of the PUT, although the S3 client supports them. The per-site lock, not a conditional write, is what makes the guard safe. + +An empty `expectedCurrent` stops the guard. It does not assert that no production alias exists yet. + +After the alias write, artemis writes the alias row and puts a `site.changed` event in the outbox, in one transaction. + +## 5. How a developer removes a deploy or a site + +Each removal in artemis is a move first, and a hard delete much later. All the destructive moves run on a context that ignores a client disconnect, with a limit of 10 minutes. + +### Remove one deploy + +`DELETE /api/site/{site}/deploys/{deployId}` + +1. Artemis reads both aliases. If one of them points at this deploy, artemis gives 409 `deploy_aliased`. A live deploy is not removable. +1. Artemis measures the deploy size. +1. Artemis moves each object from `/deploys//` to `_trash///`. +1. Artemis writes a tombstone row and removes the deploy row, in one transaction. + +The move is a copy and then a delete, for each object. It has no rollback. A failure in the middle leaves the deploy divided across the two prefixes. + +### Remove a site + +`DELETE /api/site/{slug}` has two behaviours: + +- **Without `?purge=true`** — artemis removes the registry row only. It touches no R2 bytes. It returns 204. +- **With `?purge=true`** — artemis moves the full `/` prefix into `_trash//`, writes a whole-site tombstone, and then removes the registry row. It returns 200. + +The purge moves the alias objects too, because they are under the same `/` prefix. + +### What you can recover + +| State | Recoverable | How | +| ------------------------------------------- | -------------------- | ------------------------------------------------------------------------- | +| Deploy in the trash, tombstone present | Yes | `POST .../deploys/{deployId}/restore` | +| Deploy in the trash, recovery window passed | No | The nightly purge deletes the bytes | +| Site removed without a purge | The bytes stay in R2 | Register the slug again | +| Site purged | Not through the API | The whole-site tombstone holds an empty deploy id, and restore rejects it | + +The recovery window is 7 days by default. `GET /api/site/{site}/trash` shows each tombstone with its `expiresAt` time. + +Restore moves the bytes back and makes the deploy row again. **Restore never writes an alias.** A restored deploy is in storage again, but it is not live. The developer must promote it. + +## 6. What reconciliation is + +Reconciliation repairs the drift between the two stores. R2 holds the bytes. Postgres holds the index. The two can disagree, because no write spans both stores atomically. + +For one site, the reconciler compares two sets of deploy ids: + +- **Side A** — the ids in the R2 key listing under `/deploys/`. +- **Side B** — the ids in the Postgres `deploys` table with the state `active`. + +It uses two signals to decide what to do: the set of alias targets, and the presence of the `_artemis_meta.json` marker. The marker means that a finalize completed at that prefix. + +| Drift | Condition | Repair | +| ----- | --------------------------------------------------------------------------- | ---------------------------------------- | +| 1 | In R2, not in Postgres, aliased, marker present | Write the index row again | +| 2 | In R2, not in Postgres, aliased, no marker | Report only. Never remove | +| 3 | In R2, not in Postgres, not aliased, marker present | Write the index row again | +| 4 | In R2, not in Postgres, not aliased, no marker, older than the grace window | Move to the trash, and write a tombstone | +| 5 | In Postgres, not in R2, aliased | Report only | +| 6 | In Postgres, not in R2, not aliased | Remove the index row | + +An unmarked orphan that is younger than the grace window (72 hours by default) agrees with no case. The reconciler does not touch it. This protects an upload that is still in progress. + +Before drift 4 removes anything, the reconciler reads the alias targets again. If an alias appeared in the interval, the reconciler skips that deploy and reports it. + +Two limits are important: + +- The reconciler finds **presence** drift only. If an id is on both sides, the reconciler compares no field. It never repairs a wrong size or a wrong time. +- The reconciler takes **no site lock**, unlike each other destructive path. + +Drift 2 and drift 5 are the dangerous cases. Artemis reports them at the error level and sends one Sentry event. + +## 7. Background work + +Artemis starts the background plane only when Postgres exists **and** `HATCHET_ADDR` is set. It registers four workflows. + +| Workflow | Trigger | What it does | +| --------------------- | -------------------------- | --------------------------------------------------------- | +| `gc-site` | The `site.changed` event | Runs the retention cleanup for one site | +| `reconcile` | The `site.reconcile` event | Runs the drift repair for one site | +| `reconcile-scheduler` | Cron `0 4 * * *` | Sends one `site.reconcile` event for each registered site | +| `tombstone-purge` | Cron `0 3 * * *` | Deletes the trash of each expired tombstone | + +Both event workflows run one at a time for each site. + +### The event path + +Artemis uses a transactional outbox, so a data write and its event commit together: + +1. A finalize, a promote, or a rollback adds a `site.changed` row to the `outbox` table, **inside the same transaction** as the data write. +1. A relay loop runs every 5 seconds. It claims a maximum of 100 unpublished rows, in id order. +1. The relay sends each row to Hatchet as an event, and then it sets `published_at`. + +The claim transaction commits before the relay sends anything. Delivery is therefore **at least once**, and never exactly once. A workflow must tolerate a repeat. + +The `reconcile-scheduler` workflow is the one exception. It sends the `site.reconcile` events straight to Hatchet, and it does not use the outbox. Those events are not durable. + +### The retention rules + +The `gc-site` workflow keeps a deploy if **one** of these conditions is true: + +1. An alias points at it. +1. It is one of the newest N deploys (3 by default). +1. An alias released it less than 15 seconds ago. +1. It is younger than the grace window (72 hours by default). +1. It holds the marker **and** it is younger than the retention window (7 days by default). + +Condition 5 is important. The retention window applies only to a deploy that holds the marker. An unmarked deploy gets the grace window only. + +Before it moves a deploy, `gc-site` takes the per-site lock and reads the **live R2 aliases** again. The plan uses the Postgres alias rows, but the execution uses the R2 objects. A deploy that became live in the interval is skipped. + +Neither `gc-site` nor `reconcile` deletes bytes. Both only move a prefix into the trash. The `tombstone-purge` workflow is the only hard delete in the service. + +## 8. How identity and authorization work + +There are two credentials, and they gate two separate groups of routes. + +| Credential | Routes | What it proves | +| ------------------- | ---------------------------------------------- | ------------------------------------ | +| GitHub bearer token | Each `/api/*` route except upload and finalize | The caller is a real GitHub user | +| Deploy-session JWT | `PUT .../upload` and `POST .../finalize` | The holder has a live deploy session | + +**Identity.** Artemis sends the bearer token to `GET /user` on the GitHub API and gets a login. It caches the result under a hash of the token, and never under the token itself. A positive result lives 5 minutes. A negative result lives a maximum of 30 seconds. + +**Authorization.** Each decision uses GitHub team membership. Artemis reads the team list of the site from the cached registry, and then it asks GitHub if the caller is a member of one of those teams. An empty team list always denies. + +Each GitHub team probe runs inside a handler body. The JWT middleware is not identity-only: it also refuses a site that has no teams in the registry, before any handler runs. + +There are four separate team gates: + +| Gate | Guards | Default team | +| ---------------- | --------------------------------------------------------------- | --------------------- | +| Per-site teams | promote, rollback, delete, restore, trash, alias, deploys, init | from the registry row | +| Registry team | register a site, update a site, remove a site | `staff` | +| Repo create team | `POST /api/repo` | `staff` | +| Audit read team | `GET /api/audit` | `staff` | + +**The deploy-session JWT.** Artemis signs it with HS256. The signing key must be 32 bytes or more. Artemis fixes the algorithm in two places, so an `alg=none` token and a key-confusion token both fail. Artemis also examines the issuer claim. + +On upload and finalize, artemis examines the teams of the site again. It does **not** examine the subject again. If an operator removes a user from the team, that user keeps the live deploy session until the token expires. + +Two more gaps are important: + +- `GET /api/sites` has no team gate. Any valid GitHub token can list each registered site. +- A failed GitHub probe becomes 503, even when the true cause is a bad token. + +## 9. Where the state lives + +| State | Store | Authoritative | +| ------------------- | ---------------------------- | --------------------------------------------- | +| Deploy files | R2 | **R2** | +| Alias pointer | R2 object | **R2** — the alias object decides what serves | +| Alias row | Postgres `aliases` | Follower. The cleanup plan reads it | +| Deploy index | Postgres `deploys` | Follower of R2. Reconciliation repairs it | +| Site registry | Postgres `sites`, or Valkey | **Postgres when it exists**, else Valkey | +| Registry read cache | In-process map | Follower | +| Tombstones | Postgres `tombstones` | **Postgres** | +| Outbox | Postgres `outbox` | **Postgres** | +| Audit log | Postgres `audit_log` | **Postgres** | +| Repo queue | Postgres `repo_requests` | **Postgres** | +| GitHub teams | Valkey and an in-process map | Follower of GitHub | + +### The registry, in detail + +The boot code selects the registry source of truth with one test: does a Postgres pool exist? + +- With Postgres, artemis copies the Valkey rows into Postgres one time, and then it uses Postgres for each read and each write. Valkey stays only as the change transport. +- Without Postgres, Valkey is the source of truth. + +An in-process cache sits in front of whichever store wins. It holds `slug -> teams`, and nothing more. **Each authorization check reads this cache, and not the store.** A write publishes the changed slug on a Valkey channel. Each replica then reads the full registry again. A timer of 60 seconds is the fallback. + +Team revocation is therefore eventually consistent, and not immediate. + +### Two invariants that hold the design together + +1. **The alias object is the only truth about what is live.** Postgres mirrors it. When the two disagree, reconciliation and the cleanup job both trust R2. +1. **One Postgres advisory lock, keyed by the site, serializes each mutation of that site.** Finalize, promote, rollback, delete, restore, purge, and the cleanup job all take the same key. The timeout is 30 seconds, and a contended request gets 409. + +The second invariant has one dangerous limit. With no Postgres, the lock becomes a silent no-op, and concurrent alias writes race with no error. + +## 10. Divergence between this code and the deployed release + +This document describes what the code at HEAD does. Two behaviours it describes are fixes that a deployed release older than this branch does not yet carry: + +- The scheduler used to bound the whole publish loop with one run deadline over an alphabetically sorted site list, so a slow run starved the same tail every night. Each publish now has its own 10-second bound, the order is shuffled, and a truncated run logs `reconcile.schedule.incomplete`. +- The relay claim used to release its row locks at commit, before the publish, so replicas could re-publish the same rows. A claim now stamps `claim_expires_at` and other replicas skip claimed rows until it expires; delivery stays at-least-once (an expired claim is re-published by design). + +Until the release carrying these fixes is deployed, the running service still shows the old behaviour in sections 6 and 7. diff --git a/docs/ORIENTATION.md b/docs/ORIENTATION.md new file mode 100644 index 0000000..6034844 --- /dev/null +++ b/docs/ORIENTATION.md @@ -0,0 +1,19 @@ +# Orientation — learn the architecture + +This page is for a new contributor. It gives the read sequence for the artemis architecture. Do the steps in the sequence shown. Each step names one document and tells you what the document gives you. + +## Steps + +1. Read the root [`README.md`](../README.md). It tells you what artemis is and how to start the service locally. +1. Read [`ARCHITECTURE.md`](ARCHITECTURE.md), sections 1 to 3. You learn the problem artemis solves, the parts it depends on, and the deploy lifecycle. +1. Read [`ARCHITECTURE.md`](ARCHITECTURE.md), sections 4 to 9. You learn the alias model, removal and recovery, reconciliation, background work, authorization, and where each piece of state lives. +1. Read section 10 of the same document. It lists the known divergence between the code and the deployed release. +1. Read ADR-016 in the Universe platform repo (`Architecture/decisions/016-deploy-proxy.md`). It is the authoritative specification for the API surface and the per-site authorization model. +1. Read [`design/0001-durable-execution-model.md`](design/0001-durable-execution-model.md). You learn why Postgres and Hatchet are part of the design, and the safety invariants of the retention GC. +1. Use [`README.md`](README.md) in this directory as a reference. Look up routes, configuration variables, observability, and the test suites there. Do not read it end to end. + +## Rules for all documents + +- The source code is the primary reference. When a document in this repository and the code disagree, the code is correct. +- Read [`RELEASING.md`](RELEASING.md) only when you prepare a release. It gives the version rules and the release flow. +- Read [`design/0002-scalability-capacity.md`](design/0002-scalability-capacity.md) and [`design/0003-postgres-durability.md`](design/0003-postgres-durability.md) only when you work on capacity or durability. diff --git a/docs/README.md b/docs/README.md index e423abc..6cf5076 100644 --- a/docs/README.md +++ b/docs/README.md @@ -1,6 +1,6 @@ # Artemis — reference -Audience: artemis contributors + maintainers. Architecture, API contract, configuration, observability, and the integration suite. Project overview lives in the [root README](../README.md); the release flow in [`RELEASING.md`](RELEASING.md). +Audience: artemis contributors and maintainers. This reference covers the API contract, configuration, observability, the R2 layout, the sites registry, and the test suites. The [root README](../README.md) gives the project overview. [`ORIENTATION.md`](ORIENTATION.md) gives the read sequence for a new contributor. [`ARCHITECTURE.md`](ARCHITECTURE.md) describes how the service is built. [`RELEASING.md`](RELEASING.md) gives the release flow. ## API @@ -8,7 +8,7 @@ Full route table, cross-checked against `internal/server/server.go` (`chi` wirin ``` GET /healthz → { ok: true } -GET /readyz → readiness (probes Valkey + R2) +GET /readyz → readiness (probes Valkey + R2 + Postgres) GET /api/whoami → { login, authorizedSites } POST /api/deploy/init { site, sha, files? } → { deployId, jwt, expiresAt } @@ -38,11 +38,13 @@ PUT /api/deploy/{deployId}/upload multipart stream → { received } POST /api/deploy/{deployId}/finalize { mode } → { url } · 422 missing_index ``` -`/api/repo*` is mounted only when `RepoEnabled()` is true (Apollo-11 App credentials configured — see Configuration). `DELETE /api/site/{slug}?purge=true` additionally moves the site's R2 prefix to `_trash/` and records a tombstone (gated the same as the plain delete); the bare `DELETE` only removes the registry row. `POST /api/site/{site}/deploys/{deployId}/restore` reverses a `DELETE .../deploys/{deployId}` tombstone, moving the bytes back from `_trash/` and re-marking the deploy active; `GET /api/site/{site}/trash` lists the site's tombstoned deploys with their purge-eligibility `expiresAt` (`CLEANUP_RECOVERY_DAYS` out from `trashedAt`). +`/readyz` probes the three upstreams concurrently (5 s each, `internal/handler/readyz.go`). The grades differ by upstream. An unreachable Valkey or R2 returns `503` with `valkey_unreachable` / `r2_unreachable`. An unreachable Postgres returns `200 {"ready":true,"degraded":true}` and only logs `readyz.postgres.degraded` — Postgres is optional, so it never fails the readiness gate. A failing Valkey or R2 probe reaches Sentry only after `readyzPageThreshold` (3) consecutive failures, and then once per outage until the probe recovers. -A deploy becomes live only if it is servable at `/`: `finalize`, `promote`, and `rollback` reject with `422 missing_index` (alias untouched, previous deploy keeps serving) when the target deploy has no root `index.html` — the one object the serve plane requires for `/`. On `finalize` the `422` body additionally carries an advisory `hint` when the upload looks like a framework build directory (e.g. a raw `.next` server build) rather than a static export. See ADR-016 §2026-07-26. +`/api/repo*` is mounted only when `RepoEnabled()` is true (Apollo-11 App credentials configured — see Configuration). `DELETE /api/site/{slug}?purge=true` also moves the site's R2 prefix to `_trash/` and records a tombstone; the same team gate applies as for the plain delete. The bare `DELETE` only removes the registry row. `POST /api/site/{site}/deploys/{deployId}/restore` reverses a `DELETE .../deploys/{deployId}` tombstone. It moves the bytes back from `_trash/` and marks the deploy active again. `GET /api/site/{site}/trash` lists the site's tombstoned deploys with their purge-eligibility `expiresAt` (`CLEANUP_RECOVERY_DAYS` after `trashedAt`). -`GET /api/audit` reads the durable, append-only `audit_log` — every privileged action attributed to an actor: staff/CI lifecycle (deploy, site, repo) plus system-driven GC rows (`gc.purge` under `actor=system:gc`, reconcile under `actor=system:reconcile`). Filter by `site` / `actor` / `action` / `since` (RFC3339), paginated (`limit` default 100, max 500 — `limit=0` clamps to the default 100, it does not return zero rows; `offset`), newest-first. It replaces the raw-`psql`-on-prod path for reading the trail. Because the trail is cross-tenant, the endpoint is team-gated: the caller must be on the Universe-org staff team (`AUDIT_READ_AUTHZ_TEAM`, default `staff`) — not merely any authenticated GitHub bearer. From the CLI: `universe audit ls [--actor --action --site --since --limit] [--json]` (universe-cli release follows artemis v1.5.0, since it depends on the deployed endpoint). +A deploy becomes live only if the serve plane can serve it at `/`. When the target deploy has no root `index.html` — the one object the serve plane requires for `/` — `finalize`, `promote`, and `rollback` reject with `422 missing_index`. The alias does not change, and the previous deploy continues to serve. On `finalize` the `422` body also carries an advisory `hint` when the upload looks like a framework build directory (for example a raw `.next` server build) and not a static export. See ADR-016 §2026-07-26. + +`GET /api/audit` reads the durable, append-only `audit_log`. The log holds every privileged action attributed to an actor: staff/CI lifecycle actions (deploy, site, repo) plus system-driven GC rows (`gc.purge` under `actor=system:gc`, reconcile under `actor=system:reconcile`). Filter by `site` / `actor` / `action` / `since` (RFC3339). Results are paginated newest-first (`limit` default 100, max 500; `offset`). `limit=0` clamps to the default 100 — it does not return zero rows. This endpoint replaces raw `psql` against production as the read path for the trail. The trail is cross-tenant, so the endpoint is team-gated: the caller must be on the Universe-org staff team (`AUDIT_READ_AUTHZ_TEAM`, default `staff`), not merely any authenticated GitHub bearer. From the CLI: `universe audit ls [--actor --action --site --since --limit] [--json]`. The endpoint shipped in artemis v1.5.0; universe-cli gained the verb in a later release, because the verb depends on the deployed endpoint. Auth headers (`/api/*` except `/healthz`, `/readyz`): @@ -51,7 +53,7 @@ Auth headers (`/api/*` except `/healthz`, `/readyz`): | `GET /api/*`, `POST /api/deploy/init`, `POST /api/site/*`, `POST`/`GET`/`DELETE /api/repo*` | GitHub token (PAT / OIDC) | | `PUT /api/deploy/{deployId}/upload`, `POST /api/deploy/{deployId}/finalize` | Deploy-session JWT (HS256, ≤15 min, scoped to one `(login, site, deployId)`) | -Team-gated beyond the base GitHub-bearer check: `POST /api/site/register`, `PATCH /api/site/{slug}`, `DELETE /api/site/{slug}` (`REGISTRY_AUTHZ_TEAM`); `POST /api/repo` (`REPO_CREATE_AUTHZ_TEAM`); `POST /api/repo/{id}/approve`, `POST /api/repo/{id}/reject`, `DELETE /api/repo/{id}` (`REPO_APPROVE_AUTHZ_TEAM`); `GET /api/audit` (`AUDIT_READ_AUTHZ_TEAM`, the sole team-gated read — cross-tenant trail). All other `/api/*` reads are open to any authenticated GitHub bearer. +These routes have a team gate beyond the base GitHub-bearer check: `POST /api/site/register`, `PATCH /api/site/{slug}`, `DELETE /api/site/{slug}` (`REGISTRY_AUTHZ_TEAM`); `POST /api/repo` (`REPO_CREATE_AUTHZ_TEAM`); `POST /api/repo/{id}/approve`, `POST /api/repo/{id}/reject`, `DELETE /api/repo/{id}` (`REPO_APPROVE_AUTHZ_TEAM`); `GET /api/audit` (`AUDIT_READ_AUTHZ_TEAM` — the only team-gated read, because the trail is cross-tenant). All other `/api/*` reads are open to any authenticated GitHub bearer. ## Configuration (env-driven) @@ -71,15 +73,16 @@ Loaded + validated in `internal/config/config.go` (`Load()` — fails fast on th **GitHub identity + site registry** -| Variable | Default | Description | -| ------------------------- | ------------------------ | ------------------------------------------------------- | -| `GH_CLIENT_ID` | _(required)_ | GitHub OAuth app client ID (CLI device flow) | -| `GH_ORG` | `freeCodeCamp` | GitHub org for site-registry team probes | -| `GH_API_BASE` | `https://api.github.com` | GitHub REST API base | -| `GH_MEMBERSHIP_CACHE_TTL` | `300` | GH `/user` + team membership cache TTL, seconds (5 min) | -| `VALKEY_ADDR` | _(required)_ | Valkey `host:port` for the sites registry | -| `VALKEY_PASSWORD` | _(empty)_ | Valkey AUTH password; empty for unauthenticated dev | -| `REGISTRY_AUTHZ_TEAM` | `staff` | GH team allowed to mutate the sites registry | +| Variable | Default | Description | +| ----------------------------- | ------------------------ | ------------------------------------------------------------------------------------ | +| `GH_CLIENT_ID` | _(required)_ | GitHub OAuth app client ID (CLI device flow) | +| `GH_ORG` | `freeCodeCamp` | GitHub org for site-registry team probes | +| `GH_API_BASE` | `https://api.github.com` | GitHub REST API base | +| `GH_MEMBERSHIP_CACHE_TTL` | `300` | GH `/user` + team membership cache TTL, seconds (5 min) | +| `VALKEY_ADDR` | _(required)_ | Valkey `host:port` for the sites registry | +| `VALKEY_PASSWORD` | _(empty)_ | Valkey AUTH password; empty for unauthenticated dev | +| `VALKEY_CONNECT_RETRY_WINDOW` | `5s` | Boot-time retry window for the initial Valkey dial (Go duration; `0` disables retry) | +| `REGISTRY_AUTHZ_TEAM` | `staff` | GH team allowed to mutate the sites registry | **Deploy-session JWT + R2 key layout** @@ -103,7 +106,7 @@ Loaded + validated in `internal/config/config.go` (`Load()` — fails fast on th | `GH_APP_INSTALLATION_ID` | _(empty)_ | App installation id (numeric string) | | `GH_APP_PRIVATE_KEY` | _(empty)_ | App private key PEM (PKCS#1 or PKCS#8) | -`GH_APP_ID` / `GH_APP_INSTALLATION_ID` / `GH_APP_PRIVATE_KEY` are all-or-none: set all three to enable the `/api/repo*` self-service repo-creation feature, or none. The two ids must be digit-only strings — `validate()` rejects a malformed value at boot (a YAML int sealed in sops renders as scientific notation through Helm `quote`; seal them as strings). +`GH_APP_ID` / `GH_APP_INSTALLATION_ID` / `GH_APP_PRIVATE_KEY` are all-or-none: set all three to enable the `/api/repo*` self-service repo-creation feature, or set none. The two ids must be digit-only strings — `validate()` rejects a malformed value at boot. A YAML int sealed in sops renders as scientific notation through Helm `quote`, so seal both ids as strings. **Sentry** @@ -133,14 +136,14 @@ Loaded + validated in `internal/config/config.go` (`Load()` — fails fast on th ## Observability -Observability is **Sentry-only** and independent. artemis is platform infra, so it does NOT route its own telemetry through the platform o11y stack (GlitchTip / VictoriaMetrics / ClickHouse) it deploys — that would be circular. `SENTRY_DSN` MUST point at an **external** Sentry project (`ingest.sentry.io`), never the self-hosted GlitchTip. Everything is off unless `SENTRY_DSN` is set, so dev/test runs send nothing. +Observability is **Sentry-only** and independent. artemis is platform infra, so it does NOT route its own telemetry through the platform observability stack (GlitchTip / VictoriaMetrics / ClickHouse) that it deploys — that path would be circular. `SENTRY_DSN` MUST point at an **external** Sentry project (`ingest.sentry.io`), never the self-hosted GlitchTip. All telemetry is off unless `SENTRY_DSN` is set, so dev and test runs send nothing. - **Issues** — errors, panics, and background-job failures via explicit `CaptureException` / `CaptureBackground` (op-tagged, fingerprinted). `slog.Error` does NOT create issues; the slog tee emits logs only. - **Performance (traces)** — inbound HTTP transactions ` ` + outbound spans (GitHub/R2). Probes sampled at 0; destructive routes at 100%; base `SENTRY_TRACES_SAMPLE_RATE` otherwise. - **Logs** — a `slog`→Sentry Logs tee (`EnableLogs`), scrubbed via `BeforeSendLog`, trace-correlated; numeric attributes preserved as typed values. - **Crons** — check-ins on `tombstone-purge` (`0 3 * * *`) and `reconcile-scheduler` (`0 4 * * *`). - **Stdout logs** — JSON via `log/slog` (`LOG_LEVEL`, default `info`) for `kubectl logs`; probe paths (`/healthz`, `/readyz`) silenced. Keep `LOG_LEVEL=info` in prod — several Sentry-Logs-covered signals are Info-level. -- **Durable audit trail** — a Postgres append-only `audit_log` (indefinite retention) records every privileged action attributed to an actor. This is the forensic system-of-record, distinct from Sentry Logs (a ~90-day glance stream): Sentry answers "what is happening / trending now"; `audit_log` answers "who did X, provably, months later". Read via `GET /api/audit` or `universe audit ls` (see API). `request_id` correlates a durable row back to its Sentry trace / stdout access-log line. +- **Durable audit trail** — a Postgres append-only `audit_log` (indefinite retention) records every privileged action attributed to an actor. This is the forensic system-of-record, distinct from Sentry Logs (a stream with ~90-day retention). Sentry answers "what happens now, and what is trending"; `audit_log` answers "who did X, provably, months later". Read it via `GET /api/audit` or `universe audit ls` (see API). `request_id` correlates a durable row back to its Sentry trace and its stdout access-log line. There is **no Prometheus `/metrics` endpoint** (removed in v1.4.0). Signals that were counters are covered by the mechanisms above; [ADR-016](../../fCC-U/Architecture/decisions/016-deploy-proxy.md) holds the design rationale + the full signal→Sentry map. @@ -161,7 +164,7 @@ Deferred: a DLQ-depth gauge — Hatchet v0.88.6 exposes queue depth only via a d ### Who-did-what dashboard (operator setup) -The at-a-glance "who did what" view is a **Sentry Logs** dashboard — the slog stream carries `actor` on every request-scoped line (auto-injected by the context log handler), so no code change is needed to query it. Add these widgets to the **Artemis** dashboard in the Sentry UI (the MCP has no dashboard-write API). This is the ~90-day glance; the durable/forensic answer is Postgres `audit_log` via `GET /api/audit` / `universe audit ls`. +The quick "who did what" view is a **Sentry Logs** dashboard. The slog stream carries `actor` on every request-scoped line (injected by the context log handler), so no code change is needed to query it. Add these widgets to the **Artemis** dashboard in the Sentry UI (the MCP has no dashboard-write API). This view has ~90-day retention; the durable, forensic answer is the Postgres `audit_log` via `GET /api/audit` / `universe audit ls`. Three widgets, dataset **Logs** for each: @@ -171,7 +174,7 @@ Three widgets, dataset **Logs** for each: | Actor leaderboard (7d) | Bar | `count(message)` grouped by `actor` | `message:[]` | | Unattributed actions | Table | `actor`, `message` | `message:[]` `!has:actor` | -The `` of terminal-success slog messages: `deploy.finalize`, `site.register`, `site.update`, `site.delete`, `site.purge`, `site.promote`, `site.rollback`, `repo.create.queued`, `repo.approve.created`, `repo.reject.recorded`, `repo.delete.removed`. The "Unattributed actions" widget is a regression tripwire — it should stay empty; anything in it means an action reached Sentry with no actor. GC tombstone/reconcile are system-driven and land in `audit_log` (`actor=system:gc` / `system:reconcile`) + Issues, not this human-activity view. +The `` of terminal-success slog messages: `deploy.finalize`, `site.register`, `site.update`, `site.delete`, `site.purge`, `site.promote`, `site.rollback`, `repo.create.queued`, `repo.approve.created`, `repo.reject.recorded`, `repo.delete.removed`. The "Unattributed actions" widget is a regression alarm. It stays empty in normal operation; a row in it means an action reached Sentry with no actor. GC tombstone and reconcile actions are system-driven. They land in `audit_log` (`actor=system:gc` / `system:reconcile`) and in Issues, not in this human-activity view. When enabled, Sentry captures: @@ -185,7 +188,7 @@ When enabled, Sentry captures: Each event carries `release = artemis@+`, the GitHub `login` as user, and the `request_id` tag — the same value returned in the `X-Request-ID` response header, so a Sentry issue joins directly to the stdout log line and the caller's request. -**Secrets never leave the process.** `SendDefaultPII` is off, and each of the three egress channels has its own scrubber (sharing one secret-aware core so they cannot diverge). Issues + transactions (`BeforeSend` / `BeforeSendTransaction`) strip the `Authorization`, `Cookie`, `Proxy-Authorization`, and `X-Forwarded-For` headers, the request body, the query string, and breadcrumbs, and redact secret-shaped substrings from exception values and messages. Logs (`BeforeSendLog` — the SDK does **not** run `BeforeSend` on log envelopes) redact the body and drop attributes keyed as secret or client IP. So GitHub bearer tokens, deploy-session JWTs, and upload bytes never ship on any channel. The R2 admin key, JWT signing key, and GitHub App private key are never attached (the SDK does not send the process env); the redaction pass is defense in depth over already-audited error wrapping. +**Secrets never leave the process.** `SendDefaultPII` is off. Each of the three egress channels has its own scrubber, and the three share one secret-aware core so they cannot diverge. Issues and transactions (`BeforeSend` / `BeforeSendTransaction`) strip the `Authorization`, `Cookie`, `Proxy-Authorization`, and `X-Forwarded-For` headers, the request body, the query string, and breadcrumbs. They also redact secret-shaped substrings from exception values and messages. Logs (`BeforeSendLog` — the SDK does **not** run `BeforeSend` on log envelopes) redact the body and drop attributes keyed as secret or client IP. GitHub bearer tokens, deploy-session JWTs, and upload bytes therefore never ship on any channel. The R2 admin key, JWT signing key, and GitHub App private key are never attached (the SDK does not send the process env). The redaction pass is defense in depth over already-audited error wrapping. ## R2 layout @@ -201,7 +204,7 @@ Each event carries `release = artemis@+`, the GitHub `login` as └── production # alias → "deploys/20260420-141522-abc1234" ``` -Atomic alias semantics: `PutObject` is atomic per-key in R2. Old deploy keeps serving until the alias `PUT` lands. Verify-then-PUT order means a partial deploy never becomes live. +Alias writes are atomic: R2 makes each `PutObject` atomic for one key. The old deploy serves until the alias `PUT` completes. Artemis verifies the deploy before the `PUT`, so a partial deploy never becomes live. ## Sites registry @@ -216,7 +219,7 @@ DELETE /api/site/{slug} → 204 Write endpoints are gated on `REGISTRY_AUTHZ_TEAM` (default `staff`). The read endpoint is open to any GitHub bearer. -Operator-facing CLI surface (universe-cli ≥ 0.5.0): +Operator-facing CLI surface (universe-cli; the verb table it ships is authoritative — see its `docs/reference.md`): ```sh universe sites register --team [,...] @@ -232,23 +235,39 @@ See `config/sites.yaml.example` for the on-disk schema shape. The live registry ## Local development ```sh -cp .env.example .env # then fill values -just run # boots HTTP server on $PORT -just test # go test ./... -cover (unit only) -just image # docker build -just # list all recipes +cp .env.example .env # then fill values +just run # boots HTTP server on $PORT +just test # go test -race -cover (unit only — integration excluded by build tag) +just cover # same, plus coverage.out + coverage.html +just lint # go vet +just tidy # go mod tidy +just image # docker build — multi-stage distroless +just clean # remove build artifacts +just # list all recipes +``` + +Heavier suites, each with its own stack (see the recipe body for what it boots): + +```sh +just e2e-local # artemis + pg + valkey + minio + hatchet, runs test/e2e +just hatchet-integration # real hatchet-lite via compose; R2/R3/R4/R5 workflow cases +just loadgen # scalability harness: ephemeral pg, registry/outbox/gc throughput (R14) +just smoke # repo create → approve → list against the local stack +just integration # live-deployment E2E (see Integration testing below) ``` ## Local stack (docker-compose) -A fully offline stack — no real GitHub, no real R2, no secrets — for exercising the repo command surface end to end. `docker-compose.yml` wires four services: +A fully offline stack — no real GitHub, no real R2, no secrets — that exercises the repo command surface end to end. `docker-compose.yml` wires six services: -| Service | Image / build | Role | -| ------------ | ------------------------ | ------------------------------------------------------------------ | -| `valkey` | `valkey/valkey:8-alpine` | Registry + name-claim store | -| `minio` | `minio/minio` | S3-compatible R2 stand-in (path-style; `minio-setup` seeds bucket) | -| `fakegithub` | `Dockerfile.fakegithub` | In-memory GitHub API double (`cmd/fakegithub`) | -| `artemis` | `Dockerfile` | The service under test, pointed at the three fakes via env | +| Service | Image / build | Role | +| ------------- | ------------------------- | ------------------------------------------------------- | +| `postgres` | `postgres:-alpine` | Deploy index, outbox, audit log, tombstones, repo queue | +| `valkey` | `valkey/valkey:8-alpine` | Registry + name-claim store | +| `minio` | `minio/minio:latest` | S3-compatible R2 stand-in (path-style) | +| `minio-setup` | `minio/mc:latest` | One-shot: seeds the bucket, then exits | +| `fakegithub` | `Dockerfile.fakegithub` | In-memory GitHub API double (`cmd/fakegithub`) | +| `artemis` | `Dockerfile` | The service under test, pointed at the fakes via env | `cmd/fakegithub` validates the App JWT (RS256 signature + `iss` + ≤600s `exp` cap, like real GitHub) and serves the identity (`/user`, `/user/teams`, team membership) and App (`access_tokens`, repo create/generate/get/list/contents) endpoints artemis calls. One staff user (`smoke-bot`) is a member of `staff` + `apollo-11-approvers`. @@ -263,7 +282,15 @@ just compose-down # tear down + drop volumes ## Integration testing -End-to-end suite under `internal/integration/`. Build-tagged behind `integration` so it stays out of `just test`. Hits a live, deployed artemis over HTTPS and exercises the full deploy lifecycle: +Three separate suites, none of them in `just test`: + +| Suite | Recipe | Runs against | +| --------------------------- | -------------------------- | -------------------------------------------------------- | +| `internal/integration/` | `just integration` | A live, deployed artemis over HTTPS | +| `test/e2e/` | `just e2e-local` | A locally composed full stack (pg + hatchet + R2 double) | +| `test/integration/hatchet/` | `just hatchet-integration` | A real `hatchet-lite` engine in compose | + +The rest of this section covers the first of the three. It is build-tagged behind `integration` so it stays out of `just test`, and exercises the full deploy lifecycle: ``` healthz → whoami → init → upload → finalize(preview) → curl preview @@ -279,7 +306,7 @@ ARTEMIS_URL=https://uploads.freecode.camp \ just integration ``` -`just integration-help` prints the full env-var reference. The suite is **safe to run against production** — it writes only under the `test` site (a staff-only smoke target registered in the artemis registry) and relies on the cleanup cron (7-day retention) for prefix GC. +`just integration-help` prints the full env-var reference. The suite is **safe to run against production**. It writes only under the `test` site (a staff-only smoke target registered in the artemis registry), and the cleanup cron (7-day retention) removes its prefixes. ### Setup / teardown diff --git a/docs/RELEASING.md b/docs/RELEASING.md index 6795fde..77dd717 100644 --- a/docs/RELEASING.md +++ b/docs/RELEASING.md @@ -6,9 +6,9 @@ Releases are driven by [release-please](https://github.com/googleapis/release-pl ## Versioning rule -artemis follows [Semantic Versioning 2.0.0](https://semver.org/spec/v2.0.0.html) with a single **pre-1.0 caveat**: until `v1.0.0` is cut, a `MINOR` bump may introduce a backwards-incompatible API change. This caveat is enforced mechanically by `bump-minor-pre-major: true` in `release-please-config.json` — a `BREAKING CHANGE` commit on a `0.x` line bumps `MINOR`, not `MAJOR`. +artemis follows [Semantic Versioning 2.0.0](https://semver.org/spec/v2.0.0.html). **`v1.0.0` is released** — read `.release-please-manifest.json` for the current version — so the standard contract applies in full: `MAJOR` for breaks, `MINOR` for additive features, `PATCH` for fixes. -Post-1.0, the standard semver contract applies: `MAJOR` for breaks, `MINOR` for additive features, `PATCH` for fixes. +> **Historical.** Before `v1.0.0`, a `MINOR` bump was allowed to break the API, enforced by `bump-minor-pre-major: true` in `release-please-config.json`. That key is still set but is now **inert** — it governs only a `0.x` line. It is not live policy. The "Pre-1.0 bump" column below is kept so that readers can understand old `CHANGELOG.md` entries. | Conventional Commit prefix | Pre-1.0 bump | Post-1.0 bump | | ---------------------------------------------------------------------------------- | ------------ | ------------- | @@ -18,11 +18,11 @@ Post-1.0, the standard semver contract applies: `MAJOR` for breaks, `MINOR` for | `chore(deps):` | `PATCH` | `PATCH` | | `test(*):`, `docs(*):`, `ci(*):`, `chore(*):` (non-deps), `style(*):`, `build(*):` | _no release_ | _no release_ | -`feat:` deliberately bumps `MINOR` even pre-1.0 — `bump-patch-for-minor-pre-major` is left **off**. A release PR is opened **only** when the unreleased commits contain at least one releasable change (`feat`, `fix`, `perf`, `refactor`, `chore(deps)`); pure test/docs/chore drift accumulates silently until a behaviour-bearing commit ships alongside it. +`feat:` deliberately bumps `MINOR` even pre-1.0 — `bump-patch-for-minor-pre-major` is left **off**. A release PR is opened **only** when the unreleased commits contain at least one releasable change (`feat`, `fix`, `perf`, `refactor`, `chore(deps)`). Pure test, docs, and chore commits accumulate without a release until a releasable change ships alongside them. -### `v1.0.0` trigger +### Forcing a version -Cut `v1.0.0` when the API surface is declared frozen — practically, after `GET /api/site/{site}/alias/{mode}`, the sites-registry CRUD, and the deploy/promote/rollback verbs have settled in production CLI use without breaking changes for two consecutive minor releases. Force it with a `Release-As: 1.0.0` footer (see step 2) or by editing the release PR. +`Release-As:` is the general override, not a one-off for `v1.0.0` (which has shipped). Any commit landing on `main` may carry the footer to pin the next version — see step 2. ## Release flow @@ -35,11 +35,11 @@ Every push to `main` runs `.github/workflows/release.yml`. The `release-please` - bumps `version.txt` and `.release-please-manifest.json` to the computed version, and - regenerates the `CHANGELOG.md` section from the Conventional Commits since the last release. -Read the PR. The diff **is** the proposed release: the version bump and the rendered changelog. Edit the changelog body directly in the PR if wording needs work — release-please preserves manual edits to its PR. +Read the PR. The diff **is** the proposed release: the version bump and the rendered changelog. If the text needs changes, edit the changelog body directly in the PR — release-please preserves manual edits to its PR. ### 2. (Optional) Override the version -release-please's computed bump is a strong default, not gospel — it cannot detect a quiet behaviour break hidden behind a `refactor:` prefix. To force a specific version, add a `Release-As:` footer to any commit that lands on `main` (e.g. an empty commit), then let release-please re-groom the PR: +The computed bump is a default, not a guarantee — release-please cannot detect a behaviour break hidden behind a `refactor:` prefix. To force a specific version, add a `Release-As:` footer to any commit that lands on `main` (for example an empty commit), then let release-please update the PR: ```bash git commit --allow-empty -m "chore: release 0.4.0" -m "Release-As: 0.4.0" @@ -109,6 +109,8 @@ Behaviour-bearing warnings emitted by the running service. Each entry lists the Telemetry consumers can grep `event=promote.legacy_bare` in the artemis access log to find remaining callers before the flip. +> **This deprecation is overdue.** The warn landed in `e050648`, before `v0.2.0`, and still emits from the `promote.legacy_bare` slog call in `internal/handler/site.go`. Against its own "one release after first appearance" trigger it is many releases late (every release since `v0.2.0`), and `docs/ARCHITECTURE.md` §4 still documents the promote body as optional. Either flip the empty-body branch to `400` in a dedicated behaviour-bearing commit, or retire the removal trigger and call the bare promote supported. It is an operator decision, not a docs edit. + ## Hotfix on an older release line If `v0.3.x` is current but `v0.2.x` is still pinned in some downstream deployment and needs a fix, run release-please against a maintenance branch: @@ -124,6 +126,6 @@ If `v0.3.x` is current but `v0.2.x` is still pinned in some downstream deploymen ## Why this shape -- Operators map deployed releases back to changelog entries via the semver portion of `image.tag`. Even though `@sha256:` is the load-bearing pin, the `X.Y.Z` semver prefix (no `v`, per OCI tag convention) gives a `grep`-able human anchor. Git tags carry the `v` (`v0.2.0`); registry tags do not (`0.2.0`) — intentional, consistent with release-please + docker/metadata-action defaults. -- The release decision is a **PR review**, not a local tag. The diff under review is exactly what ships (version + changelog), so a mistaken bump is caught before merge, and the version can be overridden with a `Release-As:` footer. -- `MINOR-may-break` pre-1.0 is enforced by `bump-minor-pre-major: true` and documented here so operators reading `CHANGELOG.md` see the caveat without diving into the tooling config. +- Operators map deployed releases back to changelog entries via the semver portion of `image.tag`. The `@sha256:` digest is the pin that the deployment resolves; the `X.Y.Z` semver prefix (no `v`, per OCI tag convention) gives a human anchor that `grep` can find. Git tags carry the `v` (`v0.2.0`); registry tags do not (`0.2.0`) — intentional, consistent with release-please and docker/metadata-action defaults. +- The release decision is a **PR review**, not a local tag. The diff under review is exactly what ships (version + changelog), so a review catches a mistaken bump before merge, and a `Release-As:` footer can override the version. +- `MINOR-may-break` pre-1.0 is enforced by `bump-minor-pre-major: true` and documented here, so operators who read `CHANGELOG.md` see the caveat without reading the tooling config. diff --git a/docs/design/0001-durable-execution-model.md b/docs/design/0001-durable-execution-model.md index 31e9e22..2bce317 100644 --- a/docs/design/0001-durable-execution-model.md +++ b/docs/design/0001-durable-execution-model.md @@ -6,7 +6,7 @@ ______________________________________________________________________ ## 1. Context -Moving deploy-retention GC off the Windmill cron into artemis surfaced a deeper truth: **every state-mutating artemis operation (deploy, promote, rollback, delete, GC, purge) is a durable, multi-step, crash-prone, concurrent saga**, not a request/response. Today they run synchronously in HTTP handlers with ad-hoc coordination. That mismatch is the root of the races, the orphan classes, and the "where does the sweep run" problem. +The move of deploy-retention GC off the Windmill cron into artemis showed a deeper property: **every state-mutating artemis operation (deploy, promote, rollback, delete, GC, purge) is a durable, multi-step, crash-prone, concurrent saga**, not a request/response. Today they run synchronously in HTTP handlers with ad-hoc coordination. That mismatch causes the races, the orphan classes, and the "where does the sweep run" problem. Target scale: **10s of thousands of sites → millions of deploys** (Vercel/Netlify-class). At that scale full-bucket-scan GC is impossible (event-driven incremental is mandatory) and in-RAM state is untenable (need disk-durable, queryable metadata). @@ -155,7 +155,7 @@ Audited this plan against the Universe Architecture ADRs. Building blocks + plac | ✓ | R2 = dumb swappable S3 (ADR-008 "swap R2→Ceph RGW = config change") | **aligned** | Our dumb-S3 port stance matches exactly. | | ✓ | Valkey hot-cache on gxy-management | **aligned** | Already a platform svc there; our cache role fits. (Caddy alias-cache would be cassiopeia-local Valkey — cross-galaxy, flag in M4.) | | ✓ | Observability — artemis is Sentry-only + independent (v1.4.0) | **diverged (intentional)** | Constellation apps use the platform stack; artemis manages that stack so it self-monitors via an EXTERNAL Sentry only (no GlitchTip / vmagent / `/metrics`). See ADR-016 + `docs/README.md`. | -| ✓ | Retiring Windmill **cleanup cron** (not Windmill itself) | **aligned** | Windmill stays for platform-ops; only its `cleanup_old_deploys` flow is boneyard'd. | +| ✓ | Retiring Windmill **cleanup cron** (not Windmill itself) | **aligned** | Windmill stays for platform-ops; only its `cleanup_old_deploys` flow is retired. | **Net:** the plan is *architecturally sound* but lands two platform decisions that need ADR-020 + amendments to ADR-017/008 before GA: **(D1) artemis becomes a stateful pillar**, and **(D2) Hatchet joins Windmill as a second, role-distinct engine.** Neither is a blocker — both have precedent (Veritas for stateful-on-cloud-via-CNPG; Windmill+PG for engine+DB on gxy-management) — but both must be ratified, not assumed. The dossier carries these as explicit tasks. diff --git a/docs/design/0002-scalability-capacity.md b/docs/design/0002-scalability-capacity.md index 76cfb11..bc0486d 100644 --- a/docs/design/0002-scalability-capacity.md +++ b/docs/design/0002-scalability-capacity.md @@ -81,7 +81,7 @@ Total PG connections = `pool_max_conns x replica_count` (R13: N >= 2 stateless r | Hatchet engine conns | ~20 | separate role/db on the same instance (ADR 0001 / T13) | | PG `max_connections` | 200 | 120 + 20 + ~60 headroom (admin, backup, autovac) | -The reference harness ran PG with `max_connections=200`, which comfortably holds this envelope. Below ~120 effective `max_connections` the fleet risks `too many clients` at full replica scale -- that is the first hard cliff (section 6). +The reference harness ran PG with `max_connections=200`, which holds this envelope with margin. Below ~120 effective `max_connections` the fleet risks `too many clients` at full replica scale -- that is the first hard cliff (section 6). ## 5. Hatchet per-site concurrency bound @@ -118,7 +118,7 @@ Measured against `valkey/valkey:8-alpine`, representative rows (two teams, RFC33 That is **~290 bytes/site** of resident memory (dataset portion ~263 B/site). The full 10k-site registry cache fits in **under 3 MB** on top of the ~1 MB Valkey baseline. Extrapolation is linear in site count (one hash + one set member per site); deploy count does NOT enter the registry cache. Formula: `valkey_bytes ~= 1_000_000 + 290 * sites`. -The auth `teamcache` (`internal/teamcache`) is a separate key space bounded by distinct GitHub logins seen within the membership TTL, not by site count; at fCC staff cardinality (hundreds of logins) it is negligible. A 256 MB Valkey `maxmemory` -- already over 80x the projected registry envelope -- leaves ample room; Valkey stays artemis-exclusive and NetworkPolicy-locked. +The auth `teamcache` (`internal/teamcache`) is a separate key space bounded by distinct GitHub logins seen within the membership TTL, not by site count; at fCC staff cardinality (hundreds of logins) it is negligible. A 256 MB Valkey `maxmemory` -- already over 80x the projected registry envelope -- leaves a large margin; Valkey stays artemis-exclusive and NetworkPolicy-locked. ## 7. PG storage envelope @@ -130,7 +130,7 @@ Measured table sizes after the 20,000-row run (heap + indexes, `pg_total_relatio | `sites` | 500 | 344,064 | 688 (inflated at low row count) | | `outbox` | 500 | 212,992 | 426 (transient -- drained + prunable) | -The deploy row is the dominant term at scale. At **326.5 bytes/row** including the `deploys_site_mtime_idx` index, **3,000,000 deploy rows ~= 980 MB**. Add the sites table (10k x ~500 B heap ~= 5 MB) and tombstones (bounded by the recovery window) and the artemis metadata DB lands **comfortably under 2 GB** at full target scale -- well inside a single bundled StatefulSet PVC with backup headroom (T13/T17). Formula: `deploys_bytes ~= 327 * deploy_rows`. +The deploy row is the dominant term at scale. At **326.5 bytes/row** including the `deploys_site_mtime_idx` index, **3,000,000 deploy rows ~= 980 MB**. Add the sites table (10k x ~500 B heap ~= 5 MB) and tombstones (bounded by the recovery window) and the artemis metadata DB stays **under 2 GB with margin** at full target scale -- well inside a single bundled StatefulSet PVC with backup headroom (T13/T17). Formula: `deploys_bytes ~= 327 * deploy_rows`. ## 8. Known cliffs + headroom diff --git a/docs/design/0003-postgres-durability.md b/docs/design/0003-postgres-durability.md new file mode 100644 index 0000000..04ad600 --- /dev/null +++ b/docs/design/0003-postgres-durability.md @@ -0,0 +1,42 @@ +# Local design 0003 — Postgres durability options (I3) + +> **Status:** Scoring note (2026-08-15) · **Decision:** none — the operator picks an option; the migration itself is out of scope for the `artemis-audit-fixes` wave and becomes its own dossier. **Probed live state (2026-08-15, read-only):** one `artemis-postgresql-0` pod, 70 d uptime, 10 Gi `local-path` PVC (RWO, node-pinned), on a 3-node k3s cluster (`gxy-vm-management-k3s-{1,2,3}`); nightly `artemis-backup` CronJob (`0 2 * * *`) runs `pg_dumpall` and rclones `artemis-.sql.gz` to R2 under `artemis/${GALAXY}/`. + +## 1. What the database carries, and what loss means + +One Postgres instance carries two tenant DBs: + +- **`artemis`** — deploy index, aliases, outbox, tombstones, `audit_log`, sites registry, repo queue. +- **`hatchet`** — durable-execution state, including the `v1_*_olap` run history. + +Loss is NOT a serving outage: the serve plane (Caddy `r2_alias` → R2) never touches Postgres (local ADR 0001 §3, §8). Loss means: no new deploys/GC until restore, a rebuildable deploy index (`BACKFILL_ON_BOOT` re-scans R2), an **unrebuildable `audit_log`** (the forensic system-of-record), and lost in-flight Hatchet runs. The `audit_log` sets the durability requirement. + +## 2. Failure modes the current setup does and does not cover + +| Failure | Covered today? | +| --------------------------------------------------- | ------------------------------------------------------------------------------------- | +| Pod restart, node reboot (same node) | yes — PVC remounts | +| Data corruption / bad migration noticed within 24 h | partial — restore to last 02:00 dump, losing up to 24 h of writes | +| **Node loss** (local-path PVC is node-pinned) | **no** — PVC is unschedulable elsewhere; restore from dump onto a new PVC, RPO ≤ 24 h | +| Disk loss on the node | no — same as node loss | +| R2 bucket loss (backup target) | out of scope here — R2 is the platform's own durability domain | + +RPO today: up to 24 h. RTO today: manual — new PVC + `psql < dump` + repoint; unrehearsed (unverified — no restore drill is recorded anywhere in this repo or the infra runbooks). + +## 3. Options scored + +Scale context from local design 0002: at the 10k-site target the control-plane DB stays small (3 M deploy rows ≈ single-digit GiB); durability, not capacity, is the constraint. + +| # | Option | RPO | RTO | Op burden | Fit | Score | +| --- | ------------------------------------------------------------------------- | --------------------------------------- | --------------------------- | ----------------------------------------------------------------------------- | ------------------------------------------------------------------------------------------------------------------- | ----------------------------- | +| A | Status quo (nightly `pg_dumpall` → R2) | ≤ 24 h | hours, manual | none new | already running | 2/5 | +| B | A + WAL archiving / PITR (e.g. `wal-g` or `pgBackRest` → R2) | minutes | ≤ 1 h, semi-manual | one sidecar + bucket lifecycle; restore drill needed | strong — R2 already the backup target; no new infra primitive | **4/5** | +| C | Streaming replica on a second node (operator-managed, e.g. CloudNativePG) | ~0 | minutes, automated failover | adopt an operator; chart rewrite; the current Bitnami StatefulSet is replaced | strong at steady state, largest migration step | 3/5 | +| D | Longhorn/replicated storage under the existing StatefulSet | node-loss survives; corruption does not | minutes | new storage layer on k3s; performance tax on etcd-colocated nodes | weak — replicates the block device, not the database; corruption replicates too | 1/5 | +| E | Managed Postgres (off-cluster) | provider SLO | provider SLO | least ops | conflicts with the locked operator constraint "self-hosted OSS only, artemis owns its own data" (local ADR 0001 §1) | 0/5 — ruled out by constraint | + +## 4. Recommendation to the operator + +**B first, C later if warranted.** PITR closes the real gap (24 h RPO on an unrebuildable `audit_log`) with the smallest operational delta, reusing R2 and the existing backup credentials. C (CloudNativePG) is the right end-state if the platform later needs minutes-level RTO or wants the `hatchet` tenant isolated, but it replaces the whole chart and deserves its own wave with a rehearsed cutover. Whichever option is picked, the first task of that wave is a **restore drill** — today's dump path has never been proven end-to-end (unverified, and that is itself the finding). + +Non-goals here: multi-region, connection pooling, capacity scaling (0002 covers capacity). diff --git a/internal/handler/client_disconnect_test.go b/internal/handler/client_disconnect_test.go new file mode 100644 index 0000000..c36583c --- /dev/null +++ b/internal/handler/client_disconnect_test.go @@ -0,0 +1,73 @@ +package handler + +import ( + "context" + "encoding/json" + "fmt" + "net/http" + "net/http/httptest" + "testing" + "time" + + "github.com/getsentry/sentry-go" + "github.com/stretchr/testify/require" +) + +func TestWriteUpstreamError_ClientCanceled_NoSentryNo502(t *testing.T) { + hub, ft := newHubWithTransport(t) + ctx := sentry.SetHubOnContext(t.Context(), hub) + req := httptest.NewRequestWithContext(ctx, http.MethodPost, "/api/site/www/promote", nil) + rec := httptest.NewRecorder() + + err := fmt.Errorf("site lock www: %w", context.Canceled) + writeUpstreamError(rec, req, http.StatusBadGateway, "site_lock_failed", "pg.lock.site", err) + hub.Flush(time.Second) + + require.Empty(t, ft.events, "client cancel must not reach Sentry") + require.NotEqual(t, http.StatusBadGateway, rec.Code) + require.NotContains(t, rec.Body.String(), "upstream call failed") +} + +func TestWriteUpstreamError_DeadlineExceeded_StillCaptures(t *testing.T) { + hub, ft := newHubWithTransport(t) + ctx := sentry.SetHubOnContext(t.Context(), hub) + req := httptest.NewRequestWithContext(ctx, http.MethodGet, "/api/site/www/deploys", nil) + rec := httptest.NewRecorder() + + err := fmt.Errorf("r2 list: %w", context.DeadlineExceeded) + writeUpstreamError(rec, req, http.StatusBadGateway, "r2_list_failed", "r2.list", err) + hub.Flush(time.Second) + + require.Len(t, ft.events, 1, "a genuine timeout is a real signal and must capture") + require.Equal(t, http.StatusBadGateway, rec.Code) +} + +func TestSitePromote_ClientAbort_NoUpstreamError(t *testing.T) { + h, _ := newTestHandlers(t, authedGH(), standardSites(), newFakeR2()) + h.DeployPrefix = mustDeployPrefixTemplate(prodShapedFormat) + h.Locker = &fakeErrLocker{err: fmt.Errorf("site lock %s: %w", "www", context.Canceled)} + + body, _ := json.Marshal(SitePromoteRequest{DeployID: "20260420-141522-abc1234"}) + w := withSiteRoute(http.MethodPost, "/api/site/{site}/promote", + "/api/site/www/promote", body, + contextWithLogin(context.Background(), "alice", "tok"), + h.SitePromote, + ) + require.NotEqual(t, http.StatusBadGateway, w.Code, w.Body.String()) + require.NotContains(t, w.Body.String(), "upstream call failed") +} + +func TestWriteUpstreamError_AbortedClient_LogsStatus499(t *testing.T) { + rec := httptest.NewRecorder() + sw := &statusWriter{ResponseWriter: rec, code: 200} + ctx, cancel := context.WithCancel(context.Background()) + cancel() + req := httptest.NewRequestWithContext(ctx, http.MethodPost, "/api/site/www/promote", nil) + + err := fmt.Errorf("site lock www: %w", context.Canceled) + writeUpstreamError(sw, req, http.StatusBadGateway, "site_lock_failed", "pg.lock.site", err) + + require.Equal(t, statusClientClosedRequest, sw.code, + "the access log must record 499 on a real disconnect, not a default 200") + require.Equal(t, "client_closed_request", sw.errCode) +} diff --git a/internal/handler/commit_detached_ctx_test.go b/internal/handler/commit_detached_ctx_test.go new file mode 100644 index 0000000..20af125 --- /dev/null +++ b/internal/handler/commit_detached_ctx_test.go @@ -0,0 +1,100 @@ +package handler + +import ( + "context" + "encoding/json" + "net/http" + "sync" + "testing" + "time" + + "github.com/stretchr/testify/require" +) + +type cancelOnPutAliasR2 struct { + *fakeR2 + cancel context.CancelFunc +} + +func (c *cancelOnPutAliasR2) PutAlias(ctx context.Context, key, deployID string) error { + c.cancel() + return c.fakeR2.PutAlias(ctx, key, deployID) +} + +type ctxAwareIndex struct { + mu sync.Mutex + finalized []string + aliased []string +} + +func (f *ctxAwareIndex) FinalizeAtomic(ctx context.Context, site, deployID, mode string, _ time.Time, _ int64) error { + if err := ctx.Err(); err != nil { + return err + } + f.mu.Lock() + defer f.mu.Unlock() + f.finalized = append(f.finalized, site+"/"+deployID+"/"+mode) + return nil +} + +func (f *ctxAwareIndex) AliasAtomic(ctx context.Context, site, name, deployID string, _ time.Time) error { + if err := ctx.Err(); err != nil { + return err + } + f.mu.Lock() + defer f.mu.Unlock() + f.aliased = append(f.aliased, site+"/"+name+"/"+deployID) + return nil +} + +func TestDeployFinalize_ClientAbort_StillWritesIndex(t *testing.T) { + store := newFakeR2() + reqCtx, cancel := context.WithCancel(context.Background()) + t.Cleanup(cancel) + h, jwt := newTestHandlers(t, &fakeGH{}, standardSites(), store) + h.R2 = &cancelOnPutAliasR2{fakeR2: store, cancel: cancel} + h.DeployPrefix = mustDeployPrefixTemplate(prodShapedFormat) + idx := &ctxAwareIndex{} + h.Index = idx + + deployID := "20260420-141522-abc1234" + store.objects["www.freecode.camp/deploys/"+deployID+"/index.html"] = []byte("hi") + + tok, _, err := jwt.Sign("alice", "www", deployID) + require.NoError(t, err) + body, _ := json.Marshal(DeployFinalizeRequest{Mode: "preview", Files: []string{"index.html"}}) + + withChiRoute(http.MethodPost, "/api/deploy/{deployId}/finalize", + "/api/deploy/"+deployID+"/finalize", body, + map[string]string{"Authorization": "Bearer " + tok}, + h.RequireDeployJWT(http.HandlerFunc(h.DeployFinalize)).ServeHTTP, + reqCtx, + ) + + require.Len(t, idx.finalized, 1, + "alias landed in R2, so the index write must survive the client abort — else R2 and Postgres diverge") +} + +func TestSitePromote_ClientAbort_StillWritesIndex(t *testing.T) { + store := newFakeR2() + reqCtx, cancel := context.WithCancel(context.Background()) + t.Cleanup(cancel) + h, _ := newTestHandlers(t, authedGH(), standardSites(), store) + h.R2 = &cancelOnPutAliasR2{fakeR2: store, cancel: cancel} + h.DeployPrefix = mustDeployPrefixTemplate(prodShapedFormat) + idx := &ctxAwareIndex{} + h.Index = idx + + deployID := "20260420-141522-abc1234" + store.objects["www.freecode.camp/deploys/"+deployID+"/index.html"] = []byte("hi") + + body, _ := json.Marshal(SitePromoteRequest{DeployID: deployID}) + withSiteRoute(http.MethodPost, "/api/site/{site}/promote", + "/api/site/www/promote", body, + contextWithLogin(reqCtx, "alice", "tok"), + h.SitePromote, + ) + + require.Len(t, idx.aliased, 1, + "alias landed in R2, so the alias row must survive the client abort") +} diff --git a/internal/handler/deploy.go b/internal/handler/deploy.go index 0756975..ad799c5 100644 --- a/internal/handler/deploy.go +++ b/internal/handler/deploy.go @@ -50,6 +50,10 @@ func (h *Handlers) DeployInit(w http.ResponseWriter, r *http.Request) { writeError(w, http.StatusBadRequest, "bad_request", "site and sha are required") return } + if !shaPattern.MatchString(req.SHA) { + writeError(w, http.StatusBadRequest, "bad_request", "sha must match [A-Za-z0-9-]{1,64}") + return + } telemetry.FromContext(r.Context()).SetResource(req.Site, "") h.logAction(r.Context(), "deploy.init", "start", slog.String("sha", req.SHA)) @@ -108,7 +112,7 @@ func (h *Handlers) DeployUpload(w http.ResponseWriter, r *http.Request) { return } - relPath := strings.TrimPrefix(r.URL.Query().Get("path"), "/") + relPath := r.URL.Query().Get("path") if relPath == "" { writeError(w, http.StatusBadRequest, "bad_request", "missing ?path=") return @@ -153,13 +157,6 @@ func (h *Handlers) DeployUpload(w http.ResponseWriter, r *http.Request) { "upload body exceeds configured limit") return } - if errors.Is(err, context.Canceled) { - slog.WarnContext(r.Context(), "deploy.upload.canceled", - "op", "r2.put.upload", - "path", r.URL.Path, - ) - return - } writeUpstreamError(w, r, http.StatusBadGateway, "r2_put_failed", "r2.put.upload", err) return } @@ -265,9 +262,11 @@ func (h *Handlers) DeployFinalize(w http.ResponseWriter, r *http.Request) { } aliasKey := h.aliasKey(claims.Site, mode) - lockErr := h.withSiteLock(r.Context(), h.DeployPrefix.SiteDirname(claims.Site), func() error { - telemetry.Breadcrumb(r.Context(), "lock", "site lock acquired") - if _, err := h.Registry.GetSite(r.Context(), claims.Site); err != nil { + commitCtx, cancelCommit := context.WithTimeout(context.WithoutCancel(r.Context()), aliasCommitTimeout) + defer cancelCommit() + lockErr := h.withSiteLock(commitCtx, h.DeployPrefix.SiteDirname(claims.Site), func() error { + telemetry.Breadcrumb(commitCtx, "lock", "site lock acquired") + if _, err := h.Registry.GetSite(commitCtx, claims.Site); err != nil { if errors.Is(err, registry.ErrNotFound) { writeError(w, http.StatusGone, "site_gone", "site was deleted; deploy cannot be finalized") return errAliasWriteHandled @@ -275,21 +274,21 @@ func (h *Handlers) DeployFinalize(w http.ResponseWriter, r *http.Request) { writeUpstreamError(w, r, http.StatusBadGateway, "registry_read_failed", "registry.get.finalize", err) return errAliasWriteHandled } - if err := telemetry.WithSpan(r.Context(), "r2.put.alias.finalize", func(ctx context.Context) error { + if err := telemetry.WithSpan(commitCtx, "r2.put.alias.finalize", func(ctx context.Context) error { return h.R2.PutAlias(ctx, aliasKey, deployID) }); err != nil { writeUpstreamError(w, r, http.StatusBadGateway, "r2_put_failed", "r2.put.alias.finalize", err) return errAliasWriteHandled } if h.Index != nil { - if err := telemetry.WithSpan(r.Context(), "pg.finalize.index", func(ctx context.Context) error { + if err := telemetry.WithSpan(commitCtx, "pg.finalize.index", func(ctx context.Context) error { return h.Index.FinalizeAtomic(ctx, h.DeployPrefix.SiteDirname(claims.Site), deployID, mode, time.Now().UTC(), deployBytes) }); err != nil { writeUpstreamError(w, r, http.StatusBadGateway, "pg_write_failed", "pg.finalize.index", err) return errAliasWriteHandled } } else { - h.emitSiteChanged(r.Context(), claims.Site) + h.emitSiteChanged(commitCtx, claims.Site) } return nil }) diff --git a/internal/handler/deploy_delete.go b/internal/handler/deploy_delete.go index 3336436..29e11b7 100644 --- a/internal/handler/deploy_delete.go +++ b/internal/handler/deploy_delete.go @@ -14,6 +14,8 @@ import ( const destructiveMoveTimeout = 10 * time.Minute +const aliasCommitTimeout = 60 * time.Second + func (h *Handlers) SiteDeployDelete(w http.ResponseWriter, r *http.Request) { site := chi.URLParam(r, "site") if err := h.requireSiteAuthz(w, r, site); err != nil { diff --git a/internal/handler/deploy_init_sha_test.go b/internal/handler/deploy_init_sha_test.go new file mode 100644 index 0000000..6673efb --- /dev/null +++ b/internal/handler/deploy_init_sha_test.go @@ -0,0 +1,51 @@ +package handler + +import ( + "context" + "encoding/json" + "net/http" + "strings" + "testing" + + "github.com/stretchr/testify/require" +) + +func TestDeployInit_RejectsShaOutsideCharset(t *testing.T) { + bad := []struct { + name string + sha string + }{ + {"semver dots", "v1.2.3"}, + {"underscore", "a_b"}, + {"space", "a b"}, + {"dotdot", ".."}, + {"slash", "a/b"}, + {"over 64", strings.Repeat("a", 65)}, + {"nul bytes", string(make([]byte, 8))}, + } + for _, tc := range bad { + t.Run(tc.name, func(t *testing.T) { + h, _ := newTestHandlers(t, authedGH(), standardSites(), newFakeR2()) + body, _ := json.Marshal(DeployInitRequest{Site: "www", SHA: tc.sha}) + w := withChiRoute(http.MethodPost, "/api/deploy/init", "/api/deploy/init", body, + nil, h.DeployInit, + contextWithLogin(context.Background(), "alice", "tok"), + ) + require.Equal(t, http.StatusBadRequest, w.Code, + "sha %q must be refused before an id is minted: %s", tc.sha, w.Body.String()) + }) + } + + t.Run("valid sha mints pattern-conformant id", func(t *testing.T) { + h, _ := newTestHandlers(t, authedGH(), standardSites(), newFakeR2()) + body, _ := json.Marshal(DeployInitRequest{Site: "www", SHA: strings.Repeat("a", 64)}) + w := withChiRoute(http.MethodPost, "/api/deploy/init", "/api/deploy/init", body, + nil, h.DeployInit, + contextWithLogin(context.Background(), "alice", "tok"), + ) + require.Equal(t, http.StatusOK, w.Code, w.Body.String()) + var resp DeployInitResponse + require.NoError(t, json.Unmarshal(w.Body.Bytes(), &resp)) + require.Regexp(t, deployIDPattern, resp.DeployID) + }) +} diff --git a/internal/handler/deploy_upload_path_test.go b/internal/handler/deploy_upload_path_test.go new file mode 100644 index 0000000..8ca13f0 --- /dev/null +++ b/internal/handler/deploy_upload_path_test.go @@ -0,0 +1,41 @@ +package handler + +import ( + "context" + "net/http" + "testing" + + "github.com/stretchr/testify/require" +) + +func TestDeployUpload_AbsolutePathRefused(t *testing.T) { + cases := []struct { + name string + path string + }{ + {"absolute", "/index.html"}, + {"double slash", "//index.html"}, + {"bare slash", "/"}, + {"absolute nested", "/assets/app.js"}, + } + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + store := newFakeR2() + h, jwt := newTestHandlers(t, &fakeGH{}, standardSites(), store) + deployID := "20260420-141522-abc1234" + tok, _, err := jwt.Sign("alice", "www", deployID) + require.NoError(t, err) + + w := withChiRoute(http.MethodPut, "/api/deploy/{deployId}/upload", + "/api/deploy/"+deployID+"/upload?path="+tc.path, + []byte("

hi

"), + map[string]string{"Authorization": "Bearer " + tok}, + h.RequireDeployJWT(http.HandlerFunc(h.DeployUpload)).ServeHTTP, + context.Background(), + ) + require.Equal(t, http.StatusBadRequest, w.Code, + "an absolute ?path= must be refused, not silently rewritten: %s", w.Body.String()) + require.Empty(t, store.objects, "nothing may reach R2 on a refused path") + }) + } +} diff --git a/internal/handler/handler.go b/internal/handler/handler.go index c38ee39..8c9670f 100644 --- a/internal/handler/handler.go +++ b/internal/handler/handler.go @@ -266,6 +266,23 @@ func writeErrorDetail(w http.ResponseWriter, status int, code, message string, e // a short filterable label for the failing operation (e.g., // "r2.put.alias", "valkey.register"). func writeUpstreamError(w http.ResponseWriter, r *http.Request, status int, code, op string, err error) { + if errors.Is(err, context.Canceled) { + slog.WarnContext(r.Context(), "client.disconnect", + "op", op, + "err", err, + "path", r.URL.Path, + ) + if r.Context().Err() == nil { + writeError(w, statusClientClosedRequest, "client_closed_request", "request canceled by client") + return + } + if sw, ok := w.(*statusWriter); ok { + sw.wrote = true + sw.code = statusClientClosedRequest + sw.errCode = "client_closed_request" + } + return + } slog.ErrorContext(r.Context(), "upstream.error", "op", op, "err", err, @@ -275,7 +292,12 @@ func writeUpstreamError(w http.ResponseWriter, r *http.Request, status int, code writeError(w, status, code, "upstream call failed") } +const statusClientClosedRequest = 499 + func reportUpstream(r *http.Request, code, op string, err error) { + if errors.Is(err, context.Canceled) { + return + } if hub := sentry.GetHubFromContext(r.Context()); hub != nil { sc := telemetry.FromContext(r.Context()) hub.WithScope(func(scope *sentry.Scope) { diff --git a/internal/handler/middleware.go b/internal/handler/middleware.go index 164a17a..31dca01 100644 --- a/internal/handler/middleware.go +++ b/internal/handler/middleware.go @@ -235,9 +235,19 @@ type statusWriter struct { http.ResponseWriter code int errCode string + wrote bool } func (s *statusWriter) WriteHeader(code int) { + if s.wrote { + return + } + s.wrote = true s.code = code s.ResponseWriter.WriteHeader(code) } + +func (s *statusWriter) Write(b []byte) (int, error) { + s.wrote = true + return s.ResponseWriter.Write(b) +} diff --git a/internal/handler/middleware_test.go b/internal/handler/middleware_test.go index 1f09109..1bc57ae 100644 --- a/internal/handler/middleware_test.go +++ b/internal/handler/middleware_test.go @@ -250,3 +250,28 @@ func TestWriteError_StashesErrCodeForAccessLog(t *testing.T) { assert.Equal(t, "user_unauthorized", sw.errCode) assert.Equal(t, http.StatusForbidden, sw.code) } + +func TestStatusWriter_IgnoresSecondWriteHeader(t *testing.T) { + rec := httptest.NewRecorder() + sw := &statusWriter{ResponseWriter: rec, code: 200} + + sw.WriteHeader(http.StatusOK) + sw.WriteHeader(http.StatusGatewayTimeout) + + assert.Equal(t, http.StatusOK, sw.code, + "a late 504 must not overwrite the status of a completed request") + assert.Equal(t, http.StatusOK, rec.Code) +} + +func TestStatusWriter_ImplicitWriteLatchesStatus(t *testing.T) { + rec := httptest.NewRecorder() + sw := &statusWriter{ResponseWriter: rec, code: 200} + + _, err := sw.Write([]byte("body")) + require.NoError(t, err) + sw.WriteHeader(http.StatusGatewayTimeout) + + assert.Equal(t, http.StatusOK, sw.code, + "an implicit-200 body write must latch the status against a late WriteHeader") + assert.Equal(t, http.StatusOK, rec.Code) +} diff --git a/internal/handler/readyz.go b/internal/handler/readyz.go index df7eb4e..40b2041 100644 --- a/internal/handler/readyz.go +++ b/internal/handler/readyz.go @@ -59,6 +59,10 @@ func (h *Handlers) ReadyZ(w http.ResponseWriter, r *http.Request) { wg.Wait() + if r.Context().Err() != nil { + return + } + switch { case valkeyErr != nil: page := h.readyzValkey.observe(true, true) diff --git a/internal/handler/readyz_test.go b/internal/handler/readyz_test.go index 06064a4..f47bdd5 100644 --- a/internal/handler/readyz_test.go +++ b/internal/handler/readyz_test.go @@ -281,3 +281,25 @@ func TestProbeState_ConcurrentObserveAtThreshold_PagesExactlyOnce(t *testing.T) assert.Equal(t, int64(1), pages.Load(), "merged observe-and-decide latches under one lock: N concurrent threshold-crossing failures emit exactly one page") } + +func TestReadyz_ClientAbort_DoesNotCountTowardPage(t *testing.T) { + hub, ft := newHubWithTransport(t) + r2 := newFakeR2() + r2.listErr = fmt.Errorf("r2 has_prefix: %w", context.Canceled) + h := &Handlers{Health: &fakeHealth{}, R2: r2} + + for i := 0; i < readyzPageThreshold+1; i++ { + ctx, cancel := context.WithCancel(sentry.SetHubOnContext(t.Context(), hub)) + cancel() + r := httptest.NewRequestWithContext(ctx, http.MethodGet, "/readyz", nil) + w := httptest.NewRecorder() + h.ReadyZ(w, r) + } + hub.Flush(time.Second) + + require.Empty(t, ft.events, "aborted probes must not page") + h.readyzR2.mu.Lock() + fails := h.readyzR2.fails + h.readyzR2.mu.Unlock() + require.Zero(t, fails, "aborted probes must not advance the strike counter") +} diff --git a/internal/handler/site.go b/internal/handler/site.go index 92ed551..74737a3 100644 --- a/internal/handler/site.go +++ b/internal/handler/site.go @@ -29,6 +29,8 @@ import ( // rollback paths. var deployIDPattern = regexp.MustCompile(`^\d{8}-\d{6}-[A-Za-z0-9-]{1,64}$`) +var shaPattern = regexp.MustCompile(`^[A-Za-z0-9-]{1,64}$`) + // SitePromoteRequest is the optional body for POST /api/site/{site}/promote. // Both fields are additive — an empty body keeps the legacy bare-promote // semantics (read preview alias, copy to production). @@ -82,13 +84,15 @@ func (h *Handlers) SitePromote(w http.ResponseWriter, r *http.Request) { prodKey := h.aliasKey(site, "production") var deployID string - lockErr := h.withSiteLock(r.Context(), h.DeployPrefix.SiteDirname(site), func() error { - telemetry.Breadcrumb(r.Context(), "lock", "site lock acquired") + commitCtx, cancelCommit := context.WithTimeout(context.WithoutCancel(r.Context()), aliasCommitTimeout) + defer cancelCommit() + lockErr := h.withSiteLock(commitCtx, h.DeployPrefix.SiteDirname(site), func() error { + telemetry.Breadcrumb(commitCtx, "lock", "site lock acquired") // CAS guard: read current production alias and bail on mismatch. // Treat missing-alias as the empty string so callers can use CAS // to assert "no prod yet" by passing ExpectedCurrent="". if req.ExpectedCurrent != "" { - current, err := h.R2.GetAlias(r.Context(), prodKey) + current, err := h.R2.GetAlias(commitCtx, prodKey) if err != nil && !r2.IsNotFound(err) { writeUpstreamError(w, r, http.StatusBadGateway, "r2_get_failed", "r2.get.alias.promote.cas", err) return errAliasWriteHandled @@ -112,7 +116,7 @@ func (h *Handlers) SitePromote(w http.ResponseWriter, r *http.Request) { deployID = req.DeployID if deployID == "" { previewKey := h.aliasKey(site, "preview") - v, err := h.R2.GetAlias(r.Context(), previewKey) + v, err := h.R2.GetAlias(commitCtx, previewKey) if err != nil { if r2.IsNotFound(err) { writeError(w, http.StatusUnprocessableEntity, "no_preview", "no preview alias to promote") @@ -128,7 +132,7 @@ func (h *Handlers) SitePromote(w http.ResponseWriter, r *http.Request) { } } - hasIndex, err := h.R2.HasObject(r.Context(), h.deployPrefix(site, deployID)+rootIndexKey) + hasIndex, err := h.R2.HasObject(commitCtx, h.deployPrefix(site, deployID)+rootIndexKey) if err != nil { writeUpstreamError(w, r, http.StatusBadGateway, "r2_head_failed", "r2.head.index.promote", err) return errAliasWriteHandled @@ -139,20 +143,20 @@ func (h *Handlers) SitePromote(w http.ResponseWriter, r *http.Request) { return errAliasWriteHandled } - telemetry.Breadcrumb(r.Context(), "promote", "production alias write") - if err := telemetry.WithSpan(r.Context(), "r2.put.alias.promote", func(ctx context.Context) error { + telemetry.Breadcrumb(commitCtx, "promote", "production alias write") + if err := telemetry.WithSpan(commitCtx, "r2.put.alias.promote", func(ctx context.Context) error { return h.R2.PutAlias(ctx, prodKey, deployID) }); err != nil { writeUpstreamError(w, r, http.StatusBadGateway, "r2_put_failed", "r2.put.alias.promote", err) return errAliasWriteHandled } if h.Index != nil { - if err := h.Index.AliasAtomic(r.Context(), h.DeployPrefix.SiteDirname(site), "production", deployID, time.Now().UTC()); err != nil { + if err := h.Index.AliasAtomic(commitCtx, h.DeployPrefix.SiteDirname(site), "production", deployID, time.Now().UTC()); err != nil { writeUpstreamError(w, r, http.StatusBadGateway, "pg_write_failed", "pg.alias.promote", err) return errAliasWriteHandled } } else { - h.emitSiteChanged(r.Context(), site) + h.emitSiteChanged(commitCtx, site) } return nil }) @@ -211,9 +215,11 @@ func (h *Handlers) SiteRollback(w http.ResponseWriter, r *http.Request) { prodKey := h.aliasKey(site, "production") - lockErr := h.withSiteLock(r.Context(), h.DeployPrefix.SiteDirname(site), func() error { + commitCtx, cancelCommit := context.WithTimeout(context.WithoutCancel(r.Context()), aliasCommitTimeout) + defer cancelCommit() + lockErr := h.withSiteLock(commitCtx, h.DeployPrefix.SiteDirname(site), func() error { prefix := h.deployPrefix(site, req.To) - exists, err := h.R2.HasPrefix(r.Context(), prefix) + exists, err := h.R2.HasPrefix(commitCtx, prefix) if err != nil { writeUpstreamError(w, r, http.StatusBadGateway, "r2_list_failed", "r2.has.prefix.rollback", err) return errAliasWriteHandled @@ -222,7 +228,7 @@ func (h *Handlers) SiteRollback(w http.ResponseWriter, r *http.Request) { writeError(w, http.StatusUnprocessableEntity, "deploy_missing", "target deploy no longer exists in r2") return errAliasWriteHandled } - hasIndex, err := h.R2.HasObject(r.Context(), prefix+rootIndexKey) + hasIndex, err := h.R2.HasObject(commitCtx, prefix+rootIndexKey) if err != nil { writeUpstreamError(w, r, http.StatusBadGateway, "r2_head_failed", "r2.head.index.rollback", err) return errAliasWriteHandled @@ -237,7 +243,7 @@ func (h *Handlers) SiteRollback(w http.ResponseWriter, r *http.Request) { // alias normalises to empty-string — symmetric with SitePromote so // callers can use a single response shape across both verbs. if req.ExpectedCurrent != "" { - current, err := h.R2.GetAlias(r.Context(), prodKey) + current, err := h.R2.GetAlias(commitCtx, prodKey) if err != nil && !r2.IsNotFound(err) { writeUpstreamError(w, r, http.StatusBadGateway, "r2_get_failed", "r2.get.alias.rollback.cas", err) return errAliasWriteHandled @@ -256,20 +262,20 @@ func (h *Handlers) SiteRollback(w http.ResponseWriter, r *http.Request) { } } - telemetry.Breadcrumb(r.Context(), "rollback", "production alias write") - if err := telemetry.WithSpan(r.Context(), "r2.put.alias.rollback", func(ctx context.Context) error { + telemetry.Breadcrumb(commitCtx, "rollback", "production alias write") + if err := telemetry.WithSpan(commitCtx, "r2.put.alias.rollback", func(ctx context.Context) error { return h.R2.PutAlias(ctx, prodKey, req.To) }); err != nil { writeUpstreamError(w, r, http.StatusBadGateway, "r2_put_failed", "r2.put.alias.rollback", err) return errAliasWriteHandled } if h.Index != nil { - if err := h.Index.AliasAtomic(r.Context(), h.DeployPrefix.SiteDirname(site), "production", req.To, time.Now().UTC()); err != nil { + if err := h.Index.AliasAtomic(commitCtx, h.DeployPrefix.SiteDirname(site), "production", req.To, time.Now().UTC()); err != nil { writeUpstreamError(w, r, http.StatusBadGateway, "pg_write_failed", "pg.alias.rollback", err) return errAliasWriteHandled } } else { - h.emitSiteChanged(r.Context(), site) + h.emitSiteChanged(commitCtx, site) } return nil }) diff --git a/internal/hatchet/adapter.go b/internal/hatchet/adapter.go index 4e9de3b..1647a96 100644 --- a/internal/hatchet/adapter.go +++ b/internal/hatchet/adapter.go @@ -1,4 +1,4 @@ -//lint:file-ignore SA1019 workaround: sdks/go NewClient's own public signature requires v0Client.ClientOpt (hatchet v0.88.1); no non-deprecated options surface exists yet — https://docs.hatchet.run/home/migration-guide-go +//lint:file-ignore SA1019 workaround: sdks/go NewClient's own public signature requires v0Client.ClientOpt (hatchet v0.88.6); no non-deprecated options surface exists yet — https://docs.hatchet.run/home/migration-guide-go package hatchet diff --git a/internal/hatchet/pin_test.go b/internal/hatchet/pin_test.go new file mode 100644 index 0000000..042e871 --- /dev/null +++ b/internal/hatchet/pin_test.go @@ -0,0 +1,39 @@ +package hatchet + +import ( + "os" + "path/filepath" + "regexp" + "testing" + + "github.com/stretchr/testify/require" +) + +func TestHatchetLitePinMatchesGoMod(t *testing.T) { + gomod, err := os.ReadFile("../../go.mod") + require.NoError(t, err) + modMatch := regexp.MustCompile(`github\.com/hatchet-dev/hatchet v([0-9.]+)`).FindSubmatch(gomod) + require.NotNil(t, modMatch, "go.mod must pin github.com/hatchet-dev/hatchet") + want := string(modMatch[1]) + + var globs []string + for _, pattern := range []string{"../../test/*/compose*.yaml", "../../test/*/*/compose*.yaml", "../../*compose*.yml", "../../*compose*.yaml"} { + m, gerr := filepath.Glob(pattern) + require.NoError(t, gerr) + globs = append(globs, m...) + } + require.NotEmpty(t, globs, "compose files must be discoverable from this package") + + liteRe := regexp.MustCompile(`hatchet-lite:v([0-9.]+)`) + found := 0 + for _, f := range globs { + body, rerr := os.ReadFile(f) + require.NoError(t, rerr) + for _, m := range liteRe.FindAllSubmatch(body, -1) { + found++ + require.Equal(t, want, string(m[1]), + "%s pins hatchet-lite v%s but go.mod pins v%s; every compose engine must stay in lockstep with the client", f, m[1], want) + } + } + require.NotZero(t, found, "at least one compose file must pin hatchet-lite, or this guard is vacuous") +} diff --git a/internal/observability/capture_test.go b/internal/observability/capture_test.go index d1908ed..7af7b59 100644 --- a/internal/observability/capture_test.go +++ b/internal/observability/capture_test.go @@ -103,11 +103,11 @@ func TestCaptureBackground_SuppressesTransient(t *testing.T) { CaptureBackground("gc.site.run", fmt.Errorf("tombstone-move: %w", context.Canceled)) CaptureBackground("relay.run", fmt.Errorf("outbox fetch: %w", &pgconn.PgError{Code: "57P03"})) - CaptureBackground("reconcile.schedule", fmt.Errorf("hatchet: publish site.reconcile: %w", status.Error(codes.DeadlineExceeded, "context deadline exceeded"))) + CaptureBackground("registry.refresh", fmt.Errorf("hatchet: publish site.reconcile: %w", status.Error(codes.DeadlineExceeded, "context deadline exceeded"))) CaptureBackground("gc.site.run", fmt.Errorf("site lock x: %w", &pgconn.PgError{Code: "55P03"})) sentry.CurrentHub().Flush(time.Second) - require.Empty(t, rt.events, "canceled, 57P03, gRPC DeadlineExceeded, and 55P03 must not create Sentry issues") + require.Empty(t, rt.events, "canceled, 57P03, gRPC DeadlineExceeded, and 55P03 blips on tracker-gated ops must not create Sentry issues; cron-shaped ops escalate by design") } func withTransientClock(t *testing.T, now func() time.Time) { @@ -170,7 +170,7 @@ func TestCaptureBackground_LowCadenceTransientStillEscalates(t *testing.T) { CaptureBackground("reconcile.schedule", transientErr) sentry.CurrentHub().Flush(time.Second) - require.Len(t, rt.events, 1, "3 daily-cadence failures 24h apart must still escalate") + require.Len(t, rt.events, 3, "a cron-shaped op escalates every occurrence; the cron cadence is the rate limit") require.Equal(t, []string{"reconcile.schedule", "sustained"}, rt.events[0].Fingerprint) } @@ -179,12 +179,12 @@ func TestCaptureBackground_GapBeyondResetWindowRearms(t *testing.T) { cur := time.Date(2026, 1, 1, 0, 0, 0, 0, time.UTC) withTransientClock(t, func() time.Time { return cur }) - transientErr := fmt.Errorf("tombstone-move: %w", context.Canceled) - CaptureBackground("tombstone.purge", transientErr) + transientErr := fmt.Errorf("outbox fetch: %w", context.Canceled) + CaptureBackground("relay.run", transientErr) cur = cur.Add(time.Hour) - CaptureBackground("tombstone.purge", transientErr) + CaptureBackground("relay.run", transientErr) cur = cur.Add(27 * time.Hour) - CaptureBackground("tombstone.purge", transientErr) + CaptureBackground("relay.run", transientErr) sentry.CurrentHub().Flush(time.Second) require.Empty(t, rt.events, "a gap beyond resetWindow must reset the streak, not escalate") @@ -241,3 +241,24 @@ func TestCaptureWorkflowPanic_CapturesFatalWithTag(t *testing.T) { require.Equal(t, "hatchet.task", rt.events[0].Tags["op"]) require.Equal(t, []string{"hatchet.panic"}, rt.events[0].Fingerprint) } + +func TestCaptureBackground_CronOpEscalatesEveryOccurrence(t *testing.T) { + rt := bindRecordingHub(t) + cur := time.Date(2026, 7, 15, 4, 0, 0, 0, time.UTC) + withTransientClock(t, func() time.Time { return cur }) + + transientErr := fmt.Errorf("hatchet: publish site.reconcile: %w", context.DeadlineExceeded) + const days = 30 + for range days { + CaptureBackground("reconcile.schedule", transientErr) + cur = cur.Add(24 * time.Hour) + } + sentry.CurrentHub().Flush(time.Second) + + require.Len(t, rt.events, days, + "a daily cron failing every day must escalate every day — the cron cadence IS the rate limit") + for _, ev := range rt.events { + require.Equal(t, []string{"reconcile.schedule", "sustained"}, ev.Fingerprint, + "all occurrences must group into one Sentry issue") + } +} diff --git a/internal/observability/sentry.go b/internal/observability/sentry.go index 56aa718..8656889 100644 --- a/internal/observability/sentry.go +++ b/internal/observability/sentry.go @@ -374,13 +374,18 @@ func NewSlogHandler(minLevel slog.Level) slog.Handler { }.NewSentryHandler(context.Background()) } +var cronShapedOps = map[string]bool{ + "reconcile.schedule": true, + "tombstone.purge": true, +} + // CaptureBackground reports an error raised outside any HTTP request // (e.g. the registry refresh goroutine). op becomes a tag and the // fingerprint so the failures group on their own. No-op when disabled. func CaptureBackground(op string, err error) { if IsTransient(err) { slog.Warn("background.transient", "op", op, "err", err) - if backgroundTransientRate.observe(op, backgroundTransientRate.clock()) { + if cronShapedOps[op] || backgroundTransientRate.observe(op, backgroundTransientRate.clock()) { sentry.WithScope(func(scope *sentry.Scope) { scope.SetTag("op", op) scope.SetTag("transient_sustained", "true") @@ -405,7 +410,7 @@ func IsTransient(err error) bool { case codes.Canceled, codes.DeadlineExceeded: return true } - return pg.IsInRecovery(err) || pg.IsLockTimeout(err) + return pg.IsInRecovery(err) || pg.IsLockTimeout(err) || pg.IsConnClosed(err) } func CaptureWorkflowPanic(recovered any) { diff --git a/internal/observability/transient_test.go b/internal/observability/transient_test.go index d2847db..d9227f1 100644 --- a/internal/observability/transient_test.go +++ b/internal/observability/transient_test.go @@ -28,6 +28,8 @@ func TestIsTransient(t *testing.T) { {"grpc unavailable is not transient", status.Error(codes.Unavailable, "backend down"), false}, {"lock timeout 55P03 is transient", &pgconn.PgError{Code: "55P03"}, true}, {"wrapped 55P03", fmt.Errorf("site lock x: %w", &pgconn.PgError{Code: "55P03"}), true}, + {"bare conn closed", pgconn.ErrConnClosed, true}, + {"wrapped conn closed", fmt.Errorf("relay: %w", pgconn.ErrConnClosed), true}, {"plain error", errors.New("boom"), false}, {"nil", nil, false}, } diff --git a/internal/observability/transientrate.go b/internal/observability/transientrate.go index 1241120..a39a277 100644 --- a/internal/observability/transientrate.go +++ b/internal/observability/transientrate.go @@ -8,18 +8,20 @@ import ( const ( defaultTransientRateThreshold = 3 defaultTransientResetWindow = 26 * time.Hour + defaultTransientReescalateGap = 24 * time.Hour ) type transientOpState struct { - count int - lastSeen time.Time - escalated bool + count int + lastSeen time.Time + escalatedAt time.Time } type transientRateTracker struct { mu sync.Mutex clock func() time.Time resetWindow time.Duration + reescalate time.Duration threshold int states map[string]*transientOpState } @@ -28,6 +30,7 @@ func newTransientRateTracker(clock func() time.Time, resetWindow time.Duration, return &transientRateTracker{ clock: clock, resetWindow: resetWindow, + reescalate: defaultTransientReescalateGap, threshold: threshold, states: make(map[string]*transientOpState), } @@ -44,12 +47,12 @@ func (t *transientRateTracker) observe(op string, now time.Time) bool { } if !st.lastSeen.IsZero() && now.Sub(st.lastSeen) > t.resetWindow { st.count = 0 - st.escalated = false + st.escalatedAt = time.Time{} } st.count++ st.lastSeen = now - if st.count >= t.threshold && !st.escalated { - st.escalated = true + if st.count >= t.threshold && (st.escalatedAt.IsZero() || now.Sub(st.escalatedAt) >= t.reescalate) { + st.escalatedAt = now return true } return false diff --git a/internal/observability/transientrate_test.go b/internal/observability/transientrate_test.go index 6b00842..72ecbba 100644 --- a/internal/observability/transientrate_test.go +++ b/internal/observability/transientrate_test.go @@ -70,3 +70,20 @@ func TestTransientRateTracker_DistinctOpsIndependent(t *testing.T) { t.Fatal("a fresh op must start its own count") } } + +func TestTransientRateTracker_ReescalatesAfterGap(t *testing.T) { + base := time.Date(2026, 1, 1, 0, 0, 0, 0, time.UTC) + tr := newTransientRateTracker(func() time.Time { return base }, 26*time.Hour, 3) + + tr.observe("op", base) + tr.observe("op", base.Add(time.Hour)) + if !tr.observe("op", base.Add(2*time.Hour)) { + t.Fatal("3rd observe must escalate") + } + if tr.observe("op", base.Add(3*time.Hour)) { + t.Fatal("re-escalation inside the 24h gap must stay latched") + } + if !tr.observe("op", base.Add(2*time.Hour+24*time.Hour)) { + t.Fatal("a sustained failure must re-escalate once the 24h gap has passed") + } +} diff --git a/internal/pg/errors.go b/internal/pg/errors.go index 7da03ef..ab188d4 100644 --- a/internal/pg/errors.go +++ b/internal/pg/errors.go @@ -23,3 +23,7 @@ func IsInRecovery(err error) bool { code, ok := pgCode(err) return ok && code == "57P03" } + +func IsConnClosed(err error) bool { + return errors.Is(err, pgconn.ErrConnClosed) +} diff --git a/internal/pg/errors_options_test.go b/internal/pg/errors_options_test.go new file mode 100644 index 0000000..3a84366 --- /dev/null +++ b/internal/pg/errors_options_test.go @@ -0,0 +1,73 @@ +package pg + +import ( + "errors" + "fmt" + "strings" + "testing" + "time" + + "github.com/jackc/pgx/v5/pgconn" + "github.com/stretchr/testify/require" +) + +func TestIsConnClosed(t *testing.T) { + t.Parallel() + + cases := []struct { + name string + err error + want bool + }{ + {"nil", nil, false}, + {"bare ErrConnClosed", pgconn.ErrConnClosed, true}, + {"wrapped ErrConnClosed", fmt.Errorf("relay publish: %w", pgconn.ErrConnClosed), true}, + {"doubly wrapped", fmt.Errorf("outer: %w", fmt.Errorf("inner: %w", pgconn.ErrConnClosed)), true}, + {"unrelated error", errors.New("conn closed"), false}, + {"pg error", &pgconn.PgError{Code: "57P03"}, false}, + } + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + t.Parallel() + require.Equal(t, tc.want, IsConnClosed(tc.err)) + }) + } +} + +func TestPgCodeClassifiers(t *testing.T) { + t.Parallel() + + require.True(t, IsLockTimeout(fmt.Errorf("take lock: %w", &pgconn.PgError{Code: "55P03"}))) + require.False(t, IsLockTimeout(&pgconn.PgError{Code: "57P03"})) + require.False(t, IsLockTimeout(pgconn.ErrConnClosed)) + + require.True(t, IsInRecovery(fmt.Errorf("query: %w", &pgconn.PgError{Code: "57P03"}))) + require.False(t, IsInRecovery(&pgconn.PgError{Code: "55P03"})) + require.False(t, IsInRecovery(errors.New("in recovery"))) +} + +func TestRegistryStore_WithClock(t *testing.T) { + t.Parallel() + + fixed := time.Date(2026, 8, 16, 4, 0, 0, 0, time.UTC) + s := (&RegistryStore{}).WithClock(func() time.Time { return fixed }) + require.Equal(t, fixed, s.now()) +} + +func TestRepoQueue_WithClockAndIDGen(t *testing.T) { + t.Parallel() + + fixed := time.Date(2026, 8, 16, 4, 0, 0, 0, time.UTC) + q := (&RepoQueue{}).WithClock(func() time.Time { return fixed }).WithIDGen(func() string { return "req_stub" }) + require.Equal(t, fixed, q.now()) + require.Equal(t, "req_stub", q.newID()) +} + +func TestDefaultRepoRequestID(t *testing.T) { + t.Parallel() + + first := defaultRepoRequestID() + require.True(t, strings.HasPrefix(first, "req_"), "got %q", first) + require.Len(t, first, len("req_")+20) + require.NotEqual(t, first, defaultRepoRequestID()) +} diff --git a/internal/pg/lock_deadlock_test.go b/internal/pg/lock_deadlock_test.go index 4dbf385..6558e4b 100644 --- a/internal/pg/lock_deadlock_test.go +++ b/internal/pg/lock_deadlock_test.go @@ -18,7 +18,7 @@ func newMaxConns2Repo(t *testing.T) *Repo { testcontainers.SkipIfProviderIsNotHealthy(t) ctx := context.Background() - container, err := postgres.Run(ctx, "postgres:16-alpine", + container, err := postgres.Run(ctx, testPostgresImage, postgres.WithDatabase("artemis_test"), postgres.WithUsername("artemis"), postgres.WithPassword("artemis"), diff --git a/internal/pg/migrate_test.go b/internal/pg/migrate_test.go index 5939a43..104df7f 100644 --- a/internal/pg/migrate_test.go +++ b/internal/pg/migrate_test.go @@ -14,7 +14,7 @@ func TestMigrations(t *testing.T) { testcontainers.SkipIfProviderIsNotHealthy(t) ctx := context.Background() - container, err := postgres.Run(ctx, "postgres:16-alpine", + container, err := postgres.Run(ctx, testPostgresImage, postgres.WithDatabase("artemis_test"), postgres.WithUsername("artemis"), postgres.WithPassword("artemis"), @@ -97,7 +97,7 @@ func TestReleaseAdvisoryLock_FreesLockOnCanceledCallerCtx(t *testing.T) { testcontainers.SkipIfProviderIsNotHealthy(t) ctx := context.Background() - container, err := postgres.Run(ctx, "postgres:16-alpine", + container, err := postgres.Run(ctx, testPostgresImage, postgres.WithDatabase("artemis_test"), postgres.WithUsername("artemis"), postgres.WithPassword("artemis"), @@ -175,7 +175,7 @@ func TestMigrateConcurrent_RecoversFromSameNameLeftoverIndex(t *testing.T) { testcontainers.SkipIfProviderIsNotHealthy(t) ctx := context.Background() - container, err := postgres.Run(ctx, "postgres:16-alpine", + container, err := postgres.Run(ctx, testPostgresImage, postgres.WithDatabase("artemis_test"), postgres.WithUsername("artemis"), postgres.WithPassword("artemis"), diff --git a/internal/pg/migrations/0009_outbox_claim.sql b/internal/pg/migrations/0009_outbox_claim.sql new file mode 100644 index 0000000..dec90e0 --- /dev/null +++ b/internal/pg/migrations/0009_outbox_claim.sql @@ -0,0 +1,2 @@ +ALTER TABLE outbox ADD COLUMN claimed_at timestamptz; +ALTER TABLE outbox ADD COLUMN claim_expires_at timestamptz; diff --git a/internal/pg/outbox.go b/internal/pg/outbox.go index 24256a9..e0eb794 100644 --- a/internal/pg/outbox.go +++ b/internal/pg/outbox.go @@ -3,7 +3,9 @@ package pg import ( "context" "encoding/json" + "errors" "fmt" + "slices" "time" "github.com/jackc/pgx/v5" @@ -60,15 +62,22 @@ func (r *Repo) FetchUnpublished(ctx context.Context, limit int) ([]OutboxEvent, return out, rows.Err() } +const claimTTL = 5 * time.Minute + func (r *Repo) claimBatch(ctx context.Context, limit int) ([]OutboxEvent, error) { var events []OutboxEvent err := r.WithTx(ctx, func(tx pgx.Tx) error { rows, err := tx.Query(ctx, - `SELECT id, topic, payload FROM outbox - WHERE published_at IS NULL - ORDER BY id - LIMIT $1 - FOR UPDATE SKIP LOCKED`, limit) + `UPDATE outbox + SET claimed_at = now(), claim_expires_at = now() + $2::interval + WHERE id IN ( + SELECT id FROM outbox + WHERE published_at IS NULL + AND (claim_expires_at IS NULL OR claim_expires_at < now()) + ORDER BY id + LIMIT $1 + FOR UPDATE SKIP LOCKED) + RETURNING id, topic, payload`, limit, claimTTL.String()) if err != nil { return fmt.Errorf("pg outbox claim: %w", err) } @@ -80,11 +89,15 @@ func (r *Repo) claimBatch(ctx context.Context, limit int) ([]OutboxEvent, error) } events = append(events, e) } - return rows.Err() + if err := rows.Err(); err != nil { + return fmt.Errorf("pg outbox claim rows: %w", err) + } + return nil }) if err != nil { return nil, err } + slices.SortFunc(events, func(a, b OutboxEvent) int { return int(a.ID - b.ID) }) return events, nil } @@ -106,8 +119,10 @@ func (r *Repo) RelayBatch(ctx context.Context, limit int, publish func(OutboxEve if len(doneIDs) == 0 { return 0, pubErr } - if err := r.MarkPublished(ctx, doneIDs, at); err != nil { - return 0, err + markCtx, cancel := context.WithTimeout(context.WithoutCancel(ctx), 10*time.Second) + defer cancel() + if err := r.MarkPublished(markCtx, doneIDs, at); err != nil { + return len(doneIDs), errors.Join(pubErr, err) } return len(doneIDs), pubErr } diff --git a/internal/pg/outbox_relaybatch_test.go b/internal/pg/outbox_relaybatch_test.go index a343920..366a375 100644 --- a/internal/pg/outbox_relaybatch_test.go +++ b/internal/pg/outbox_relaybatch_test.go @@ -12,7 +12,7 @@ import ( "github.com/stretchr/testify/require" ) -func TestRelayBatch_ConcurrentReplicasAtLeastOncePublish(t *testing.T) { +func TestRelayBatch_ExclusiveAcrossReplicas(t *testing.T) { repo := newTestRepo(t) ctx := context.Background() @@ -58,7 +58,7 @@ func TestRelayBatch_ConcurrentReplicasAtLeastOncePublish(t *testing.T) { require.Len(t, published, total, "every event claimed") for id, c := range published { - assert.GreaterOrEqual(t, c, 1, "event %d published at least once across %d replicas (at-least-once; consumer worker.WorkflowGCSite is idempotent, E1)", id, replicas) + assert.Equal(t, 1, c, "event %d must publish exactly once across %d replicas while its claim marker holds (B3); duplicates mean the claim is not exclusive", id, replicas) } remaining, err := repo.FetchUnpublished(ctx, total) @@ -98,3 +98,58 @@ func payloadSite(t *testing.T, e OutboxEvent) string { require.NoError(t, json.Unmarshal(e.Payload, &m)) return m["site"] } + +func TestRelayBatch_MarkSurvivesContextDeath(t *testing.T) { + repo := newTestRepo(t) + ctx, cancel := context.WithCancel(context.Background()) + t.Cleanup(cancel) + for _, s := range []string{"a", "b", "c"} { + require.NoError(t, repo.EnqueueSiteChanged(context.Background(), s)) + } + + calls := 0 + publish := func(OutboxEvent) error { + calls++ + if calls == 1 { + return nil + } + cancel() + return fmt.Errorf("publish: %w", context.Canceled) + } + + n, err := repo.RelayBatch(ctx, 10, publish, time.Now()) + require.Error(t, err, "the publish failure must surface, not be swallowed by the mark") + assert.Equal(t, 1, n, "the pre-failure publish must be marked even though the batch ctx died") + + remaining, ferr := repo.FetchUnpublished(context.Background(), 10) + require.NoError(t, ferr) + assert.Len(t, remaining, 2, "only the published event may be marked; the rest stay for retry") +} + +func TestClaimTTL_ExceedsBatchAndMarkBudget(t *testing.T) { + require.Greater(t, claimTTL, 2*time.Minute+10*time.Second, + "claimTTL must exceed worker.DefaultRelayBatchTimeout (2m) plus the 10s detached mark window, or an expiring claim re-publishes the rows it protects; worker cannot be imported here (cycle), so the bound is pinned literally") +} + +func TestClaimBatch_ExpiredClaimIsReclaimable(t *testing.T) { + repo := newTestRepo(t) + ctx := context.Background() + for _, s := range []string{"a", "b"} { + require.NoError(t, repo.EnqueueSiteChanged(ctx, s)) + } + + first, err := repo.claimBatch(ctx, 10) + require.NoError(t, err) + require.Len(t, first, 2) + + second, err := repo.claimBatch(ctx, 10) + require.NoError(t, err) + require.Empty(t, second, "a live claim must be exclusive") + + _, err = repo.pool.Exec(ctx, `UPDATE outbox SET claim_expires_at = now() - interval '1 second'`) + require.NoError(t, err) + + third, err := repo.claimBatch(ctx, 10) + require.NoError(t, err) + require.Len(t, third, 2, "an expired claim is the crash-recovery path; its rows must become claimable again") +} diff --git a/internal/pg/repo_test.go b/internal/pg/repo_test.go index 1517801..c7c5094 100644 --- a/internal/pg/repo_test.go +++ b/internal/pg/repo_test.go @@ -18,7 +18,7 @@ func newTestRepo(t *testing.T) *Repo { testcontainers.SkipIfProviderIsNotHealthy(t) ctx := context.Background() - container, err := postgres.Run(ctx, "postgres:16-alpine", + container, err := postgres.Run(ctx, testPostgresImage, postgres.WithDatabase("artemis_test"), postgres.WithUsername("artemis"), postgres.WithPassword("artemis"), @@ -139,3 +139,5 @@ func TestRepo_TombstoneLifecycle(t *testing.T) { require.NoError(t, err) assert.Empty(t, expired, "cleared tombstone gone") } + +const testPostgresImage = "postgres:16-alpine" diff --git a/internal/worker/relay.go b/internal/worker/relay.go index a3ab281..6359c0c 100644 --- a/internal/worker/relay.go +++ b/internal/worker/relay.go @@ -16,38 +16,42 @@ type Publisher interface { } const ( - relayTimeoutFloor = 4 * time.Second - relayTimeoutPerItem = 150 * time.Millisecond + DefaultPublishTimeout = 10 * time.Second + DefaultRelayBatchTimeout = 2 * time.Minute + defaultRelayBatch = 100 ) -func relayTimeout(batch int) time.Duration { - return relayTimeoutFloor + time.Duration(batch)*relayTimeoutPerItem -} - type Relay struct { - Source OutboxSource - Publisher Publisher - Batch int - Timeout time.Duration - Now func() time.Time + Source OutboxSource + Publisher Publisher + Batch int + PublishTimeout time.Duration + BatchTimeout time.Duration + Now func() time.Time } func (r *Relay) RunOnce(ctx context.Context) (int, error) { batch := r.Batch if batch <= 0 { - batch = 100 + batch = defaultRelayBatch + } + perPublish := r.PublishTimeout + if perPublish <= 0 { + perPublish = DefaultPublishTimeout } - timeout := r.Timeout - if timeout <= 0 { - timeout = relayTimeout(batch) + batchTimeout := r.BatchTimeout + if batchTimeout <= 0 { + batchTimeout = DefaultRelayBatchTimeout } now := time.Now if r.Now != nil { now = r.Now } - ctx, cancel := context.WithTimeout(ctx, timeout) + ctx, cancel := context.WithTimeout(ctx, batchTimeout) defer cancel() return r.Source.RelayBatch(ctx, batch, func(e pg.OutboxEvent) error { - return r.Publisher.Publish(ctx, e.Topic, e.Payload) + pctx, pcancel := context.WithTimeout(ctx, perPublish) + defer pcancel() + return r.Publisher.Publish(pctx, e.Topic, e.Payload) }, now()) } diff --git a/internal/worker/relay_test.go b/internal/worker/relay_test.go index 2c2e0f2..c9d9a0a 100644 --- a/internal/worker/relay_test.go +++ b/internal/worker/relay_test.go @@ -86,8 +86,32 @@ func TestOutboxRelay_BoundsPublishDeadline(t *testing.T) { require.True(t, pub.hadDeadline, "publish must run under a bounded deadline so a Hatchet stall can't pin the pool connection + FOR UPDATE row locks") - assert.InDelta(t, float64(19*time.Second), float64(time.Until(pub.deadline)), float64(2*time.Second), - "publish deadline is ~19s (4s floor + 100*150ms) so a full default batch has budget to drain") + assert.InDelta(t, float64(DefaultPublishTimeout), float64(time.Until(pub.deadline)), float64(2*time.Second), + "each publish carries its own bound, so one stalled publish cannot consume the whole batch's budget") +} + +type ctxRecordingSource struct { + *fakeSource + deadline time.Time + hadDeadline bool +} + +func (s *ctxRecordingSource) RelayBatch(ctx context.Context, limit int, publish func(pg.OutboxEvent) error, at time.Time) (int, error) { + s.deadline, s.hadDeadline = ctx.Deadline() + return s.fakeSource.RelayBatch(ctx, limit, publish, at) +} + +func TestOutboxRelay_BoundsTheWholeBatch(t *testing.T) { + src := &ctxRecordingSource{fakeSource: newFakeSource("a")} + relay := &Relay{Source: src, Publisher: &fakePublisher{}, Now: func() time.Time { return time.Unix(0, 0) }} + + _, err := relay.RunOnce(context.Background()) + require.NoError(t, err) + + require.True(t, src.hadDeadline, + "the batch runs under its own bound so a stalled drain cannot pin the pool connection and row locks") + assert.InDelta(t, float64(DefaultRelayBatchTimeout), float64(time.Until(src.deadline)), float64(2*time.Second), + "the batch bound is DefaultRelayBatchTimeout and does not follow the per-publish bound") } func TestOutboxRelay(t *testing.T) { diff --git a/justfile b/justfile index a90ad7b..79e9fa2 100644 --- a/justfile +++ b/justfile @@ -80,7 +80,7 @@ hatchet-integration: HATCHET_COMPOSE_FILE="$PWD/compose.hatchet.yaml" \ {{go}} test -tags=integration -count=1 -timeout=10m ../../../internal/hatchet/... -# go vet (CI also runs golangci-lint) +# go vet (the only linter CI runs) lint: {{go}} vet {{pkg}} diff --git a/test/e2e/compose.e2e.yaml b/test/e2e/compose.e2e.yaml index 5ad36cc..39c64b9 100644 --- a/test/e2e/compose.e2e.yaml +++ b/test/e2e/compose.e2e.yaml @@ -2,7 +2,7 @@ name: artemis-e2e services: postgres: - image: postgres:17-alpine + image: postgres:16-alpine environment: POSTGRES_USER: artemis POSTGRES_PASSWORD: artemis @@ -16,7 +16,7 @@ services: retries: 30 hatchet-postgres: - image: postgres:17-alpine + image: postgres:16-alpine environment: POSTGRES_USER: hatchet POSTGRES_PASSWORD: hatchet @@ -28,7 +28,7 @@ services: retries: 30 hatchet-lite: - image: ghcr.io/hatchet-dev/hatchet/hatchet-lite:v0.88.1 + image: ghcr.io/hatchet-dev/hatchet/hatchet-lite:v0.88.6 depends_on: hatchet-postgres: condition: service_healthy diff --git a/test/integration/hatchet/compose.hatchet.yaml b/test/integration/hatchet/compose.hatchet.yaml index 73f270a..b996237 100644 --- a/test/integration/hatchet/compose.hatchet.yaml +++ b/test/integration/hatchet/compose.hatchet.yaml @@ -2,7 +2,7 @@ name: artemis-hatchet-it services: postgres: - image: postgres:17-alpine + image: postgres:16-alpine environment: POSTGRES_USER: hatchet POSTGRES_PASSWORD: hatchet @@ -14,7 +14,7 @@ services: retries: 30 hatchet-lite: - image: ghcr.io/hatchet-dev/hatchet/hatchet-lite:v0.88.1 + image: ghcr.io/hatchet-dev/hatchet/hatchet-lite:v0.88.6 depends_on: postgres: condition: service_healthy