diff --git a/.github/workflows/ci.yaml b/.github/workflows/ci.yaml index 3202fc907bb..4a52781a2bf 100644 --- a/.github/workflows/ci.yaml +++ b/.github/workflows/ci.yaml @@ -1139,7 +1139,7 @@ jobs: runs-on: ubuntu-latest timeout-minutes: 30 env: - DECKD_REF: 95bfe7a3aaa355745d0260b5eb581d95fc8cc004 + DECKD_REF: eb116367aa9b29d3be8c781b20fd05aabb12afcb steps: # This repository too, for its upload action. - uses: actions/checkout@de0fac2e4500dabe0009e67214ff5f5447ce83dd # v6.0.2 diff --git a/docs/realm-history-recovery-runbook.md b/docs/realm-history-recovery-runbook.md new file mode 100644 index 00000000000..97d3adc057d --- /dev/null +++ b/docs/realm-history-recovery-runbook.md @@ -0,0 +1,120 @@ +# Realm history recovery runbook + +A realm that records history treats its history (the deckd repository at +`/.realm-history/`) as the truth, and its directory on disk as a copy of +HEAD. Recovery keeps the two in agreement. This runbook covers what to do when +it can't, and when it reports a realm changed outside its history. + +## What recovery does + +Recovery runs when a realm is mounted, before its boot index, and from an hourly +cron in worker-manager (`REALM_HISTORY_RECOVERY_CRON_SCHEDULE`), which asks +every realm that records to check itself. It also runs when a write finds disk +behind HEAD at one of its paths. Each run: + +1. **Settles writes whose answer was lost.** Each write is journaled `pending` + in `realm_history_revisions` before deckd is asked to record it. A row still + pending after five minutes has its outcome asked of deckd, as does one a + later write to the same files finds, right away. Asking fences the + operation ID: once deckd reports it absent, it can never commit later, and a + retry under that ID gets a 409 `operation-fenced`. +2. **Completes revisions a crash cut short.** For a revision recorded but not + all on disk, or never indexed, recovery writes what HEAD holds at its paths + under those paths' write locks, then announces and indexes them. +3. **Compares disk with HEAD.** A file that differs from HEAD is recorded as a + system revision: + - _Unrecorded changes (recording was disabled)_ (`kind = gap`, + trailer `Boxel-Gap: recording-disabled`), when the realm last ran without + recording; + - _Unattributed external changes_ (`kind = drift`) otherwise. An edit that a + write overwrites before recovery sees it is lost. + +Each run logs one `boxel:operations` line with `kind: "realm-recovery"`. That +line carries the outcome (`clean`, `recovered`, `read-only` or `leased`), what +the run did, and how long the disk scan took. + +## Alerts + +Both alerts are in `packages/observability/provisioning/alerting/realm-history-group.json`. + +### Realm Changed Outside Its History + +Something edited the realm directory directly: a shell on the host, an rsync, or +a tool writing into EFS. Nothing is lost, because the change is now a revision. +Find the line: + +``` +{service="realm-server"} |= "realm-recovery" | json | line_format "{{ if .log }}{{ .log }}{{ else }}{{ __line__ }}{{ end }}" | json | driftPaths > 0 +``` + +It names the realm and the revision (`driftRevisions`). Find out who made the +edit, and have them make it through the realm (or `boxel realm push`) from now +on. + +Until the "every write path" work lands, `/_atomic`, source `DELETE`, create-realm +and publish also write without recording, so on a recording realm their changes +show up here too. + +### Realm Read-Only After Failed Recovery + +Recovery couldn't finish, so the realm now refuses writes. Reads still work. A +write gets: + +``` +HTTP/1.1 423 Locked +X-Boxel-Realm-Read-Only: recovery-failed +{"errors":[{"status":"423","code":"realm-read-only", ...}]} +``` + +`_readiness-check` answers 200 with the same `X-Boxel-Realm-Read-Only` header. +The realm stays read-only until an operator clears it. Neither a restart nor the +schedule clears it. + +## Clearing a read-only realm + +1. **Read the reason.** The `realm-recovery` line's `error` field holds it, and + so does the database: + + ```sql + SELECT read_only_at, read_only_reason, read_only_detail + FROM realm_history_state WHERE realm_url = ''; + ``` + +2. **Fix the cause.** + - _deckd unreachable or timing out_ (`fetch failed`, `ECONNREFUSED`, + `lock-timeout`): check the deckd sidecar's health (`GET /health`) and its + restarts. Recovery needs deckd for every step. + - _`history-unsettled`_: deckd found jj's HEAD and its recorded HEAD + disagreeing for more than a few seconds. It usually settles; retry. + - _A corrupt repository_ (`internal` errors from `/head`, `/tree` or + `/read`): restore `/.realm-history/` from backup. Don't delete it. + Deleting it discards the realm's history, and the next recovery would then + take in the whole realm as one gap revision. + - _Disk full or a filesystem error_: free space, then retry. + - _A gap or drift too large for one revision_ (`payload-too-large`): one + revision carries at most 40 MiB. Until recovery splits by size, record the + change in smaller pieces through the realm, then retry. +3. **Clear the seal**, with the Grafana operator secret. The clear re-runs + recovery first: + + ```sh + curl -X POST -H "Authorization: Bearer $GRAFANA_SECRET" \ + "https:///_grafana-clear-realm-read-only?realm=" + ``` + + - It answers 200 with `"readOnly": false` once recovery succeeds and writes + are back. + - It answers 409 with `"readOnly": true` and the new reason if recovery fails + again. + +4. **Only if you've repaired the realm by hand** and recovery still can't run, + add `&force=true`. That lifts the seal without recovering. Whatever disk then + holds that history doesn't is recorded as drift the next time recovery runs. + +## Configuration + +| Variable | Where | Purpose | +| -------------------------------------- | ---------------------------- | ---------------------------------------------------------- | +| `REALM_HISTORY_DECKD_URL` | realm-server, worker-manager | deckd's address. The cron only runs where it is set | +| `REALM_HISTORY_REALMS` | realm-server | Realms that record: a comma-separated list of URLs, or `*` | +| `REALM_HISTORY_RECOVERY_CRON_SCHEDULE` | worker-manager | When the scheduled sweep runs (default `20 * * * *`) | diff --git a/packages/host/config/schema/1793070304376_schema.sql b/packages/host/config/schema/1793129747681_schema.sql similarity index 100% rename from packages/host/config/schema/1793070304376_schema.sql rename to packages/host/config/schema/1793129747681_schema.sql diff --git a/packages/observability/provisioning/alerting/realm-history-group.json b/packages/observability/provisioning/alerting/realm-history-group.json new file mode 100644 index 00000000000..c5849414174 --- /dev/null +++ b/packages/observability/provisioning/alerting/realm-history-group.json @@ -0,0 +1,181 @@ +{ + "apiVersion": 1, + "groups": [ + { + "orgId": 1, + "name": "realm-history-group", + "folder": "defd2d156sav4d", + "interval": "60s", + "rules": [ + { + "uid": "realm-history-read-only", + "title": "Realm Read-Only After Failed Recovery", + "condition": "C", + "data": [ + { + "refId": "A", + "queryType": "instant", + "relativeTimeRange": { + "from": 300, + "to": 0 + }, + "datasourceUid": "loki", + "model": { + "datasource": { + "type": "loki", + "uid": "loki" + }, + "editorMode": "code", + "expr": "sum(count_over_time({service=\"realm-server\"} |= \"realm-recovery\" | json | line_format \"{{ if .log }}{{ .log }}{{ else }}{{ __line__ }}{{ end }}\" | json | channel=\"boxel:operations\" | kind=\"realm-recovery\" | outcome=\"read-only\" [5m])) or vector(0)", + "intervalMs": 1000, + "maxDataPoints": 43200, + "queryType": "instant", + "refId": "A" + } + }, + { + "refId": "B", + "queryType": "", + "relativeTimeRange": { + "from": 0, + "to": 0 + }, + "datasourceUid": "__expr__", + "model": { + "datasource": { + "type": "__expr__", + "uid": "__expr__" + }, + "expression": "A", + "intervalMs": 1000, + "maxDataPoints": 43200, + "reducer": "last", + "refId": "B", + "type": "reduce" + } + }, + { + "refId": "C", + "queryType": "", + "relativeTimeRange": { + "from": 0, + "to": 0 + }, + "datasourceUid": "__expr__", + "model": { + "conditions": [ + { + "evaluator": { + "params": [0], + "type": "gt" + } + } + ], + "datasource": { + "type": "__expr__", + "uid": "__expr__" + }, + "expression": "B", + "intervalMs": 1000, + "maxDataPoints": 43200, + "refId": "C", + "type": "threshold" + } + } + ], + "noDataState": "OK", + "execErrState": "Error", + "for": "0s", + "annotations": { + "summary": "A realm that records history could not be brought back into agreement with it, and now refuses writes (423 realm-read-only) while still serving reads. The boxel:operations line with kind=\"realm-recovery\" and outcome=\"read-only\" names the realm and carries the error. Follow docs/realm-history-recovery-runbook.md, then clear it with POST /_grafana-clear-realm-read-only?realm=." + }, + "isPaused": false + }, + { + "uid": "realm-history-drift", + "title": "Realm Changed Outside Its History", + "condition": "C", + "data": [ + { + "refId": "A", + "queryType": "instant", + "relativeTimeRange": { + "from": 300, + "to": 0 + }, + "datasourceUid": "loki", + "model": { + "datasource": { + "type": "loki", + "uid": "loki" + }, + "editorMode": "code", + "expr": "sum(count_over_time({service=\"realm-server\"} |= \"realm-recovery\" | json | line_format \"{{ if .log }}{{ .log }}{{ else }}{{ __line__ }}{{ end }}\" | json | channel=\"boxel:operations\" | kind=\"realm-recovery\" | driftPaths > 0 [5m])) or vector(0)", + "intervalMs": 1000, + "maxDataPoints": 43200, + "queryType": "instant", + "refId": "A" + } + }, + { + "refId": "B", + "queryType": "", + "relativeTimeRange": { + "from": 0, + "to": 0 + }, + "datasourceUid": "__expr__", + "model": { + "datasource": { + "type": "__expr__", + "uid": "__expr__" + }, + "expression": "A", + "intervalMs": 1000, + "maxDataPoints": 43200, + "reducer": "last", + "refId": "B", + "type": "reduce" + } + }, + { + "refId": "C", + "queryType": "", + "relativeTimeRange": { + "from": 0, + "to": 0 + }, + "datasourceUid": "__expr__", + "model": { + "conditions": [ + { + "evaluator": { + "params": [0], + "type": "gt" + } + } + ], + "datasource": { + "type": "__expr__", + "uid": "__expr__" + }, + "expression": "B", + "intervalMs": 1000, + "maxDataPoints": 43200, + "refId": "C", + "type": "threshold" + } + } + ], + "noDataState": "OK", + "execErrState": "Error", + "for": "0s", + "annotations": { + "summary": "Recovery found files in a realm that records history which no revision explained (an edit made to the realm directory directly), and recorded them as a system \"Unattributed external changes\" revision. The boxel:operations line with kind=\"realm-recovery\" names the realm, the revision and how many files. Nothing is lost, but whoever edited the directory should go through the realm instead; see docs/realm-history-recovery-runbook.md." + }, + "isPaused": false + } + ] + } + ] +} diff --git a/packages/postgres/migrations/1793129747681_realm-history-recovery.js b/packages/postgres/migrations/1793129747681_realm-history-recovery.js new file mode 100644 index 00000000000..93e16da0fb3 --- /dev/null +++ b/packages/postgres/migrations/1793129747681_realm-history-recovery.js @@ -0,0 +1,90 @@ +// What a realm's recovery needs from the history journal, and the realm's own +// recovery state. +// +// `realm_history_revisions` gains: +// - `state`: a row is inserted `pending`, and committed to the database, +// before its write asks history to record anything, so a write whose +// outcome is lost still leaves a trace that recovery resolves. It becomes +// `committed` beside the revision, or `aborted` once history reports the +// operation absent (which fences it). `revision` and `parent` are unknown +// until then. +// - `paths`: the paths the revision wrote or deleted, so recovery knows what +// to bring up to HEAD for a revision whose bytes never all reached disk. +// - `kind`: `write` for a revision a request made, `drift` for changes found +// on disk that no revision explains, `gap` for changes made while the realm +// was not recording. +// - `post_committed_at`: when the revision's index pass was queued, so a +// revision whose bytes reached disk but whose indexing never started is +// told apart from one that is complete. +// +// `realm_history_state` is one row per realm that a history-aware process has +// mounted: whether it recorded when last mounted (a realm that did not, or has +// no row, has unrecorded changes to take in as a gap), the read-only seal a +// failed recovery leaves, and the lease that keeps two replicas from +// recovering the realm at once. It stays off `realm_metadata` so the archived +// filters there never have to tell a read-only realm from an archived one: +// read-only realms keep serving reads. + +exports.shorthands = undefined; + +exports.up = (pgm) => { + pgm.sql(` + ALTER TABLE realm_history_revisions + ALTER COLUMN revision DROP NOT NULL, + ALTER COLUMN parent DROP NOT NULL, + ADD COLUMN IF NOT EXISTS state varchar NOT NULL DEFAULT 'committed', + ADD COLUMN IF NOT EXISTS paths text[] NOT NULL DEFAULT '{}', + ADD COLUMN IF NOT EXISTS kind varchar NOT NULL DEFAULT 'write', + ADD COLUMN IF NOT EXISTS post_committed_at timestamptz + `); + // Rows journaled before this column existed were indexed by the write that + // materialized them. + pgm.sql(` + UPDATE realm_history_revisions SET post_committed_at = materialized_at + WHERE post_committed_at IS NULL AND materialized_at IS NOT NULL + `); + pgm.sql(` + CREATE INDEX IF NOT EXISTS realm_history_revisions_pending + ON realm_history_revisions (realm_url, seq) + WHERE state = 'pending' + `); + pgm.sql(` + CREATE INDEX IF NOT EXISTS realm_history_revisions_not_post_committed + ON realm_history_revisions (realm_url, seq) + WHERE post_committed_at IS NULL + `); + // History order is read from each revision's parent, and the newest + // revision is the one no other names as its parent. + pgm.sql(` + CREATE INDEX IF NOT EXISTS realm_history_revisions_parent + ON realm_history_revisions (realm_url, parent) + `); + pgm.sql(` + CREATE TABLE IF NOT EXISTS realm_history_state ( + realm_url varchar PRIMARY KEY, + recording boolean NOT NULL, + recording_changed_at timestamptz NOT NULL DEFAULT now(), + read_only_at timestamptz, + read_only_reason varchar, + read_only_detail text, + recovery_lease_owner varchar, + recovery_lease_until timestamptz, + last_recovered_at timestamptz, + last_recovery_outcome varchar + ) + `); +}; + +exports.down = (pgm) => { + pgm.sql(`DROP TABLE IF EXISTS realm_history_state`); + pgm.sql(`DROP INDEX IF EXISTS realm_history_revisions_parent`); + pgm.sql(`DROP INDEX IF EXISTS realm_history_revisions_not_post_committed`); + pgm.sql(`DROP INDEX IF EXISTS realm_history_revisions_pending`); + pgm.sql(` + ALTER TABLE realm_history_revisions + DROP COLUMN IF EXISTS post_committed_at, + DROP COLUMN IF EXISTS kind, + DROP COLUMN IF EXISTS paths, + DROP COLUMN IF EXISTS state + `); +}; diff --git a/packages/postgres/scripts/schema-dump.sh b/packages/postgres/scripts/schema-dump.sh index 8c57ca8427a..d349caaf81b 100755 --- a/packages/postgres/scripts/schema-dump.sh +++ b/packages/postgres/scripts/schema-dump.sh @@ -46,6 +46,7 @@ docker exec boxel-pg pg_dump \ --exclude-table-and-children=retired_realm_event_rooms \ --exclude-table-and-children=host_shell_generation \ --exclude-table-and-children=realm_history_revisions \ + --exclude-table-and-children=realm_history_state \ --no-tablespaces \ --no-table-access-method \ --no-owner \ diff --git a/packages/realm-server/handlers/handle-clear-realm-read-only.ts b/packages/realm-server/handlers/handle-clear-realm-read-only.ts new file mode 100644 index 00000000000..b7cdd084838 --- /dev/null +++ b/packages/realm-server/handlers/handle-clear-realm-read-only.ts @@ -0,0 +1,96 @@ +import type Koa from 'koa'; +import { + type Realm, + RealmPaths, + SupportedMimeType, + clearRealmReadOnly, + realmReadOnly, +} from '@cardstack/runtime-common'; +import { + sendResponseForBadRequest, + sendResponseForSystemError, + setContextResponse, +} from '../middleware/index.ts'; +import type { CreateRoutesArgs } from '../routes.ts'; + +// The operator's way out of read-only for a realm whose recovery failed (see +// docs/realm-history-recovery-runbook.md). Runs the realm's recovery again and +// lifts the seal only if it succeeds, so a realm is never put back to taking +// writes while it still disagrees with its history. `force=true` lifts the +// seal without recovering, for a realm an operator has already repaired by +// hand; whatever disk then holds that history does not is taken in as drift by +// the next recovery. +// +// `realm=` resolves as for `/_grafana-reindex`, and only through the realm +// registry. +export default function handleClearRealmReadOnly({ + serverURL, + dbAdapter, + reconciler, +}: CreateRoutesArgs): (ctxt: Koa.Context, next: Koa.Next) => Promise { + return async function (ctxt: Koa.Context, _next: Koa.Next) { + let realmPath = ctxt.URL.searchParams.get('realm')?.replace(/\/$/, ''); + if (!realmPath) { + await sendResponseForBadRequest( + ctxt, + 'Request missing "realm" query param', + ); + return; + } + let force = ctxt.URL.searchParams.get('force') === 'true'; + let realmURL: string; + try { + realmURL = new RealmPaths(new URL(serverURL)).directoryURL( + realmPath, + ).href; + } catch (e: any) { + await sendResponseForBadRequest( + ctxt, + `invalid "realm" value: ${e.message}`, + ); + return; + } + let realm: Realm | undefined; + try { + realm = await reconciler.lookupOrMount(realmURL); + } catch (e: any) { + await sendResponseForSystemError(ctxt, e.message); + return; + } + if (!realm) { + await sendResponseForBadRequest( + ctxt, + `realm ${realmURL} does not exist on this server`, + ); + return; + } + if (!realm.history) { + await sendResponseForBadRequest( + ctxt, + `realm ${realmURL} does not record history, so it is never read-only`, + ); + return; + } + let body: Record; + if (force) { + await clearRealmReadOnly(dbAdapter, realmURL); + body = { realm: realmURL, readOnly: false, forced: true }; + } else { + let result = await realm.recoverHistory({ trigger: 'operator' }); + let readOnly = await realmReadOnly(dbAdapter, realmURL); + body = { + realm: realmURL, + readOnly: Boolean(readOnly), + ...(readOnly ? { reason: readOnly.detail ?? readOnly.reason } : {}), + ...(result ? { recovery: result.event } : {}), + }; + } + await setContextResponse( + ctxt, + new Response(JSON.stringify(body, null, 2), { + status: body.readOnly ? 409 : 200, + headers: { 'content-type': SupportedMimeType.JSON }, + }), + ); + }; +} diff --git a/packages/realm-server/lib/cron-scheduler.ts b/packages/realm-server/lib/cron-scheduler.ts index 76844e229a1..e91c7e6465b 100644 --- a/packages/realm-server/lib/cron-scheduler.ts +++ b/packages/realm-server/lib/cron-scheduler.ts @@ -25,6 +25,12 @@ import { MEDIA_CACHE_GC_CRON_TZ, createMediaCacheGcCronJob, } from './media-cache-gc-config.ts'; +import { enqueueRealmHistoryRecovery } from '../scripts/realm-history-recovery.ts'; +import { + REALM_HISTORY_RECOVERY_CRON_SCHEDULE, + REALM_HISTORY_RECOVERY_CRON_TZ, + createRealmHistoryRecoveryCronJob, +} from './realm-history-recovery-config.ts'; import { enqueueScopedCssGc } from '../scripts/scoped-css-gc.ts'; import { SCOPED_CSS_GC_CRON_SCHEDULE, @@ -61,6 +67,11 @@ export function startCronJobs(): void { if (scopedCssGcJob) { jobs.push(scopedCssGcJob); } + + let realmHistoryRecoveryJob = startRealmHistoryRecoveryCron(); + if (realmHistoryRecoveryJob) { + jobs.push(realmHistoryRecoveryJob); + } } export function stopCronJobs(): void { @@ -183,3 +194,31 @@ function startScopedCssGcCron(): CronJob | undefined { ); return job; } + +// Only where some realm records history: the sweep visits the realms that do, +// and with none there is nothing for it to check. +function startRealmHistoryRecoveryCron(): CronJob | undefined { + if (!process.env.REALM_HISTORY_DECKD_URL) { + log.info( + 'REALM_HISTORY_DECKD_URL not set, skipping realm-history-recovery cron setup', + ); + return undefined; + } + let job = createRealmHistoryRecoveryCronJob( + async () => { + try { + await enqueueRealmHistoryRecovery(); + } catch (error) { + Sentry.captureException(error); + log.error('realm-history-recovery cron failed to enqueue job', error); + } + }, + { runOnInit: false }, + ); + + job.start(); + log.info( + `realm-history-recovery cron scheduled for ${REALM_HISTORY_RECOVERY_CRON_SCHEDULE} ${REALM_HISTORY_RECOVERY_CRON_TZ}`, + ); + return job; +} diff --git a/packages/realm-server/lib/deckd-client.ts b/packages/realm-server/lib/deckd-client.ts index f01b4bc3ca7..15975eb0cbf 100644 --- a/packages/realm-server/lib/deckd-client.ts +++ b/packages/realm-server/lib/deckd-client.ts @@ -7,6 +7,8 @@ import { type HistoryCommitResult, type HistoryOutcome, type PreparedCommit, + type HistoryReadResult, + type HistoryTreeEntry, } from '@cardstack/runtime-common/realm-history'; // How long a call to deckd may take before its outcome is treated as unknown. @@ -71,6 +73,7 @@ export class DeckdHistoryBackend implements HistoryBackend { ); return { revision: String(body.revision), + parent: parentOf(body, '/commit'), created: Boolean(body.created), replayed: Boolean(body.replayed), }; @@ -81,7 +84,12 @@ export class DeckdHistoryBackend implements HistoryBackend { async outcome(operationId: string): Promise { let body = await this.#post('/outcome', { repo: this.repo, operationId }); if (body.state === 'committed') { - return { state: 'committed', revision: String(body.revision) }; + return { + state: 'committed', + revision: String(body.revision), + parent: parentOf(body, '/outcome'), + created: Boolean(body.created), + }; } if (body.state === 'absent') { return { state: 'absent' }; @@ -94,6 +102,40 @@ export class DeckdHistoryBackend implements HistoryBackend { return String(body.revision); } + async tree(revision: string): Promise { + let body = await this.#post('/tree', { repo: this.repo, revision }); + return (body.entries as { path: string; blob: string }[]).map( + ({ path, blob }) => ({ path: String(path), blob: String(blob) }), + ); + } + + async read(revision: string, paths: string[]): Promise { + let body = await this.#post('/read', { + repo: this.repo, + revision, + paths, + }); + return { + files: ( + body.files as { + path: string; + contentBase64?: string; + absent?: boolean; + }[] + ).map((file) => + file.absent || file.contentBase64 === undefined + ? { path: String(file.path) } + : { + path: String(file.path), + content: new Uint8Array( + Buffer.from(file.contentBase64, 'base64'), + ), + }, + ), + remaining: (body.remaining as string[] | undefined) ?? [], + }; + } + async #post( endpoint: string, payload: Record, @@ -136,3 +178,18 @@ export class DeckdHistoryBackend implements HistoryBackend { ); } } + +// A revision's parent as deckd answered it: null only for the root commit. A +// deckd that leaves it out predates reporting it, and the journal can't be +// kept in history order without it. +function parentOf( + body: Record, + endpoint: string, +): string | undefined { + if (!('parent' in body)) { + throw new Error( + `deckd answered ${endpoint} without the revision's parent; it is older than this realm server needs`, + ); + } + return body.parent == null ? undefined : String(body.parent); +} diff --git a/packages/realm-server/lib/realm-history-recovery-config.ts b/packages/realm-server/lib/realm-history-recovery-config.ts new file mode 100644 index 00000000000..85173a78cf7 --- /dev/null +++ b/packages/realm-server/lib/realm-history-recovery-config.ts @@ -0,0 +1,28 @@ +import { CronJob } from 'cron'; + +// Hourly by default. Recovery also runs whenever a realm is mounted and +// whenever a write finds disk behind HEAD, so the schedule is the backstop +// for what those miss: an edit made to disk directly while the realm was up, +// and a write whose answer was lost and never retried. Each run scans every +// file of every realm that records, so the cadence is a tuning knob via the +// env override. Off the top of the hour to stay clear of the hourly +// prerender-html reconcile. +export const REALM_HISTORY_RECOVERY_CRON_SCHEDULE = + process.env.REALM_HISTORY_RECOVERY_CRON_SCHEDULE ?? '20 * * * *'; +export const REALM_HISTORY_RECOVERY_CRON_TZ = + process.env.REALM_HISTORY_RECOVERY_CRON_TZ ?? 'America/New_York'; + +export function createRealmHistoryRecoveryCronJob( + onTick: () => void, + options: { runOnInit?: boolean } = {}, +) { + return new CronJob( + REALM_HISTORY_RECOVERY_CRON_SCHEDULE, + onTick, + null, + false, + REALM_HISTORY_RECOVERY_CRON_TZ, + null, + options.runOnInit ?? false, + ); +} diff --git a/packages/realm-server/main.ts b/packages/realm-server/main.ts index 494e7992b0b..5056e088b88 100644 --- a/packages/realm-server/main.ts +++ b/packages/realm-server/main.ts @@ -852,6 +852,8 @@ const reportHostShellToManager = async (dbAdapter: PgAdapter) => { kind: row.kind, diskPath, }), + historyConfigured: + REALM_HISTORY !== undefined && row.kind !== 'published', }, { ...(fullIndexOnStartup ? { fullIndexOnStartup: true as const } : {}), diff --git a/packages/realm-server/routes.ts b/packages/realm-server/routes.ts index 6a09f78b838..0c554dfb78a 100644 --- a/packages/realm-server/routes.ts +++ b/packages/realm-server/routes.ts @@ -50,6 +50,7 @@ import handleClientTelemetry from './handlers/handle-client-telemetry.ts'; import handleQueueStatusRequest from './handlers/handle-queue-status.ts'; import handleSkillValidation from './handlers/handle-skill-validation.ts'; import handleReindex from './handlers/handle-reindex.ts'; +import handleClearRealmReadOnly from './handlers/handle-clear-realm-read-only.ts'; import handleFullReindex from './handlers/handle-full-reindex.ts'; import handleRemoveJob from './handlers/handle-remove-job.ts'; import handleAddCredit from './handlers/handle-add-credit.ts'; @@ -619,6 +620,10 @@ export function createRoutes(args: CreateRoutesArgs) { '/_grafana-set-realm-catalog', handleSetRealmCatalog(args), ); + registerGrafanaEndpoint( + '/_grafana-clear-realm-read-only', + handleClearRealmReadOnly(args), + ); router.post('/_post-deployment', handlePostDeployment(args)); router.post( '/_realm-auth', diff --git a/packages/realm-server/scripts/realm-history-recovery.ts b/packages/realm-server/scripts/realm-history-recovery.ts new file mode 100644 index 00000000000..18bf00e5ad0 --- /dev/null +++ b/packages/realm-server/scripts/realm-history-recovery.ts @@ -0,0 +1,44 @@ +import '../instrument.ts'; +import '../setup-logger.ts'; // This should be first +import { + logger, + systemInitiatedPrerenderHtmlPriority, +} from '@cardstack/runtime-common'; +import { REALM_HISTORY_RECOVERY_JOB_TIMEOUT_SEC } from '@cardstack/runtime-common/tasks/realm-history-recovery'; +import { PgAdapter, PgQueuePublisher } from '@cardstack/postgres'; +import * as Sentry from '@sentry/node'; + +const log = logger('realm-history-recovery'); + +// Enqueue the scheduled recovery sweep rather than running it in the +// worker-manager process: a worker visits each realm that records history and +// asks the realm-server serving it to check the realm against its history. At +// the background tier, so it never competes with indexing or user work. +export async function enqueueRealmHistoryRecovery({ + priority = systemInitiatedPrerenderHtmlPriority, + migrateDB, +}: { + priority?: number; + migrateDB?: boolean; +} = {}) { + let dbAdapter = new PgAdapter({ autoMigrate: migrateDB || undefined }); + let queue = new PgQueuePublisher(dbAdapter); + + try { + await queue.publish({ + jobType: 'realm-history-recovery', + concurrencyGroup: 'realm-history-recovery', + timeout: REALM_HISTORY_RECOVERY_JOB_TIMEOUT_SEC, + priority, + args: {}, + }); + log.info('enqueued realm-history-recovery job'); + } catch (error) { + Sentry.captureException(error); + log.error('failed to enqueue realm-history-recovery job', error); + throw error; + } finally { + await queue.destroy(); + await dbAdapter.close(); + } +} diff --git a/packages/realm-server/tests/helpers/index.ts b/packages/realm-server/tests/helpers/index.ts index 0bbecdab714..969616bf2c3 100644 --- a/packages/realm-server/tests/helpers/index.ts +++ b/packages/realm-server/tests/helpers/index.ts @@ -1500,6 +1500,7 @@ export async function createRealm({ cardDocumentCache = new CardDocumentCache(), realmFor, history, + historyConfigured, published, }: { dir: string; @@ -1564,6 +1565,9 @@ export async function createRealm({ // The realm's revision history, built for the realm's directory. Absent, // the realm records none. history?: (realmDir: string) => HistoryBackend; + // Forwarded to the Realm constructor's `historyConfigured`: a realm mounted + // without `history` notes that its writes go unrecorded. + historyConfigured?: true; // The other realms the realm can reach, for an explain on its policy card // that asks about a target in one of them. realmFor?: (url: URL) => Promise; @@ -1648,6 +1652,7 @@ export async function createRealm({ cardDocumentCache, ...(realmFor ? { realmFor } : {}), ...(history ? { history: history(dir) } : {}), + ...(historyConfigured ? { historyConfigured: true } : {}), callApi: callApiSender, }, { diff --git a/packages/realm-server/tests/helpers/realm-history.ts b/packages/realm-server/tests/helpers/realm-history.ts new file mode 100644 index 00000000000..e83342d0f1c --- /dev/null +++ b/packages/realm-server/tests/helpers/realm-history.ts @@ -0,0 +1,134 @@ +import { execFileSync, spawn, type ChildProcess } from 'child_process'; +import { createServer } from 'net'; +import { join } from 'path'; +import { HISTORY_DIR } from '@cardstack/runtime-common/realm-history'; + +// Running a real deckd for the realm-server tests that record history, and +// reading what it recorded. deckd's API reads a revision's files but not its +// metadata, so a revision's author, message and trailers are read straight +// out of the git store inside the repository deckd keeps (`jj/store/git`), +// with git. + +// The root commit: HEAD of a history nothing has been recorded in. +export const ROOT_REVISION = '0'.repeat(40); + +// --- deckd ------------------------------------------------------------------ + +// The deckd binary: `DECKD_BIN`, or `deckd` on PATH. CI provides one, so there +// its absence is a failure; a developer without one skips this module. +export function findDeckd(): string | undefined { + if (process.env.DECKD_BIN) { + return process.env.DECKD_BIN; + } + try { + return ( + execFileSync('which', ['deckd'], { stdio: ['ignore', 'pipe', 'ignore'] }) + .toString() + .trim() || undefined + ); + } catch { + return undefined; + } +} + +async function freePort(): Promise { + return await new Promise((resolve, reject) => { + let server = createServer(); + server.once('error', reject); + server.listen(0, '127.0.0.1', () => { + let address = server.address(); + server.close(() => + typeof address === 'object' && address + ? resolve(address.port) + : reject(new Error('no port')), + ); + }); + }); +} + +export async function startDeckd( + binary: string, +): Promise<{ url: string; process: ChildProcess }> { + let port = await freePort(); + let child = spawn(binary, [], { + env: { ...process.env, DECKD_BIND: `127.0.0.1:${port}` }, + stdio: ['ignore', 'ignore', 'inherit'], + }); + let url = `http://127.0.0.1:${port}`; + let deadline = Date.now() + 20_000; + while (Date.now() < deadline) { + try { + let response = await fetch(`${url}/health`); + if (response.ok) { + return { url, process: child }; + } + } catch { + // not listening yet + } + await new Promise((resolve) => setTimeout(resolve, 100)); + } + child.kill(); + throw new Error(`deckd at ${binary} did not answer /health within 20s`); +} + +// --- reading a revision ------------------------------------------------------- + +export function git(realmDir: string, ...args: string[]): string { + return execFileSync('git', [ + '--git-dir', + join(realmDir, HISTORY_DIR, 'jj', 'store', 'git'), + ...args, + ]).toString(); +} + +export interface Revision { + author: string; + email: string; + parents: string[]; + subject: string; + trailers: Record; + // `A path`, `M path` or `D path`, sorted. + changes: string[]; +} + +export function readRevision(realmDir: string, revision: string): Revision { + let raw = git( + realmDir, + 'show', + '--no-renames', + '--name-status', + '--format=%an%x00%ae%x00%P%x00%B%x00', + revision, + ); + let [author, email, parents, body, rest] = raw.split('\0'); + let trailers: Record = {}; + for (let line of body.split('\n')) { + let match = /^([A-Za-z0-9-]+): (.*)$/.exec(line); + if (match) { + trailers[match[1]] = match[2]; + } + } + return { + author, + email, + parents: parents.trim() ? parents.trim().split(' ') : [], + subject: body.split('\n')[0], + trailers, + changes: rest + .split('\n') + .map((line) => line.trim()) + .filter(Boolean) + .map((line) => line.replace(/\t/, ' ')) + .sort(), + }; +} + +// Every revision from HEAD back, newest first. +export function historyFrom(realmDir: string, head: string): string[] { + if (head === ROOT_REVISION) { + return []; + } + return git(realmDir, 'rev-list', '--first-parent', head) + .split('\n') + .filter(Boolean); +} diff --git a/packages/realm-server/tests/realm-history-recovery-test.ts b/packages/realm-server/tests/realm-history-recovery-test.ts new file mode 100644 index 00000000000..2bb8936a3c2 --- /dev/null +++ b/packages/realm-server/tests/realm-history-recovery-test.ts @@ -0,0 +1,1052 @@ +import QUnit from 'qunit'; +const { module, test } = QUnit; +import { basename, join } from 'path'; +import type { ChildProcess } from 'child_process'; +import fsExtra from 'fs-extra'; +import { dirSync, setGracefulCleanup, type DirResult } from 'tmp'; +import { + realmHistoryState, + type DefinitionLookup, + type QueuePublisher, + type Realm, + type RealmAdapter, +} from '@cardstack/runtime-common'; +import { + commitBatch, + isOperationFailure, + setRealmRecoverySink, + type RealmRecoveryEvent, +} from '@cardstack/runtime-common/card-operations'; +import { + HISTORY_DIR, + type HistoryBackend, + type HistoryCommitRequest, + type HistoryCommitResult, + type HistoryOutcome, + type HistoryReadResult, + type HistoryTreeEntry, + type PreparedCommit, +} from '@cardstack/runtime-common/realm-history'; +import { + DRIFT_MESSAGE, + GAP_MESSAGE, +} from '@cardstack/runtime-common/realm-history-recovery'; +import { realmHistoryRecovery } from '@cardstack/runtime-common/tasks/realm-history-recovery'; +import type { PgAdapter } from '@cardstack/postgres'; +import { DeckdHistoryBackend } from '../lib/deckd-client.ts'; +import { + createJWT, + createRealm, + createVirtualNetwork, + setupDB, +} from './helpers/index.ts'; +import { + findDeckd, + historyFrom, + readRevision, + startDeckd, +} from './helpers/realm-history.ts'; + +const { ensureDirSync, existsSync, readFileSync, removeSync, writeFileSync } = + fsExtra; + +// Recovery brings a realm that records history back into agreement with it: +// after a crash at any step of a write, a write whose answer was lost, or an +// edit made to the realm directory directly. Each module below mounts a realm, +// does something to it, and mounts it again — as a realm-server restarting +// would — over one directory, one database and one real deckd. + +const realmURL = 'http://127.0.0.1:4464/recovering/'; + +// deckd, with the faults a test asks for injected around it. +class FaultyHistory implements HistoryBackend { + inner: DeckdHistoryBackend; + // Commit, then lose the response. + loseCommitResponses = false; + // Lose the request before it reaches deckd, so nothing commits. + dropCommits = false; + // Every outcome query fails, as with deckd unreachable. + failOutcomes = false; + // Every read of a revision's files fails. + failReads = false; + + constructor(inner: DeckdHistoryBackend) { + this.inner = inner; + } + + prepareCommit(request: Omit): PreparedCommit { + let prepared = this.inner.prepareCommit(request); + return { + commit: async (parent: string): Promise => { + if (this.dropCommits) { + throw new Error('socket hang up'); + } + let result = await prepared.commit(parent); + if (this.loseCommitResponses) { + throw new Error('socket hang up'); + } + return result; + }, + }; + } + + async outcome(operationId: string): Promise { + if (this.failOutcomes) { + throw new Error('connect ECONNREFUSED'); + } + return await this.inner.outcome(operationId); + } + + head(): Promise { + return this.inner.head(); + } + + tree(revision: string): Promise { + return this.inner.tree(revision); + } + + async read(revision: string, paths: string[]): Promise { + if (this.failReads) { + throw new Error('connect ECONNREFUSED'); + } + return await this.inner.read(revision, paths); + } +} + +class Crash extends Error {} + +const deckdBinary = findDeckd(); +if (!deckdBinary && process.env.CI) { + throw new Error( + 'realm-history-recovery-test needs deckd: set DECKD_BIN or put deckd on PATH', + ); +} +const recordingTest = deckdBinary ? test : QUnit.skip; + +module(basename(import.meta.filename), function (hooks) { + let deckd: { url: string; process: ChildProcess } | undefined; + let dbAdapter: PgAdapter; + let publisher: QueuePublisher; + let tmpDir: DirResult; + let realmDir: string; + let recoveries: RealmRecoveryEvent[]; + let broadcasts: { realm: number; event: Record }[]; + let mounted: number; + + setGracefulCleanup(); + + hooks.before(async function () { + if (deckdBinary) { + deckd = await startDeckd(deckdBinary); + } + }); + + hooks.after(function () { + deckd?.process.kill(); + }); + + setupDB(hooks, { + beforeEach: async (_dbAdapter, _publisher) => { + dbAdapter = _dbAdapter; + publisher = _publisher; + tmpDir = dirSync({ unsafeCleanup: true }); + realmDir = join(tmpDir.name, 'realm'); + ensureDirSync(realmDir); + writeFileSync( + join(realmDir, 'realm.json'), + JSON.stringify({ + data: { + type: 'card', + attributes: { cardInfo: { name: 'Recovering Realm' } }, + meta: { + adoptsFrom: { + module: '@cardstack/base/realm-config', + name: 'RealmConfig', + }, + }, + }, + }), + ); + writeFileSync(join(realmDir, 'a.txt'), 'alpha'); + writeFileSync(join(realmDir, 'b.txt'), 'bravo'); + recoveries = []; + broadcasts = []; + mounted = 0; + setRealmRecoverySink((event) => recoveries.push(event)); + }, + }); + + hooks.afterEach(function () { + setRealmRecoverySink(undefined); + }); + + // Raw source writes resolve no definition, so a lookup that only answers + // the write path's invalidation is enough. + let definitionLookup = { + forRealm() { + return this; + }, + async invalidate() { + return []; + }, + } as unknown as DefinitionLookup; + + // A realm over the test's directory and database, as one realm-server + // replica mounts it. Started unless asked not to, which runs recovery. + async function mount(opts?: { + recording?: false; + historyConfigured?: true; + start?: false; + faulty?: (history: FaultyHistory) => void; + }): Promise<{ + realm: Realm; + adapter: RealmAdapter; + history: FaultyHistory | undefined; + }> { + let history: FaultyHistory | undefined; + let { realm, adapter } = await createRealm({ + dir: realmDir, + definitionLookup, + realmURL, + permissions: { + '*': ['read', 'write'], + owner: ['read', 'write', 'realm-owner'], + }, + virtualNetwork: createVirtualNetwork(), + publisher, + dbAdapter, + skipBootIndex: true, + ...(opts?.historyConfigured ? { historyConfigured: true } : {}), + ...(opts?.recording === false + ? {} + : { + history: (dir: string) => { + history = new FaultyHistory( + new DeckdHistoryBackend({ + deckdURL: deckd!.url, + realmDir: dir, + }), + ); + opts?.faulty?.(history); + return history; + }, + }), + }); + let index = mounted++; + let broadcast = adapter.broadcastRealmEvent.bind(adapter); + adapter.broadcastRealmEvent = async (event, ...rest) => { + broadcasts.push({ realm: index, event: event as Record }); + return await broadcast(event, ...rest); + }; + if (opts?.start !== false) { + await realm.start(); + } + return { realm, adapter, history }; + } + + function write( + realm: Realm, + files: Record, + operationId: string, + ) { + return commitBatch( + realm.batchCore, + Object.entries(files).map(([path, content]) => ({ + op: 'update' as const, + href: `${realmURL}${path}`, + content, + rawSource: true, + })), + { + waitForIndex: false, + // A retry is recognized by asking for what its first attempt asked + // for, as a request's digest would show. + recording: { + operationId, + channel: 'operations', + fingerprint: JSON.stringify(files), + }, + }, + ); + } + + // What a write that failed was refused with: the operation error it + // answered, or the error itself for one that is not an operation's. + async function failure(promise: Promise): Promise<{ + status?: number; + code?: string; + meta?: Record; + message: string; + }> { + try { + await promise; + } catch (err: unknown) { + if (isOperationFailure(err)) { + return { + status: err.error.status, + code: err.error.code, + meta: err.error.meta as Record | undefined, + message: err.message, + }; + } + return { + code: (err as { code?: string }).code, + message: err instanceof Error ? err.message : String(err), + }; + } + throw new Error('expected the write to fail'); + } + + function onDisk(path: string): string | undefined { + let file = join(realmDir, path); + return existsSync(file) ? readFileSync(file, 'utf8') : undefined; + } + + function backend() { + return new DeckdHistoryBackend({ deckdURL: deckd!.url, realmDir }); + } + + async function atHead(path: string): Promise { + let history = backend(); + let { files } = await history.read(await history.head(), [path]); + let content = files[0]?.content; + return content ? new TextDecoder().decode(content) : undefined; + } + + async function journal() { + return (await dbAdapter.execute( + `SELECT operation_id, revision, parent, state, kind, paths, + materialized_at IS NOT NULL AS materialized, + post_committed_at IS NOT NULL AS post_committed + FROM realm_history_revisions WHERE realm_url = $1 ORDER BY seq`, + { bind: [realmURL] }, + )) as { + operation_id: string; + revision: string | null; + parent: string | null; + state: string; + kind: string; + paths: string[]; + materialized: boolean; + post_committed: boolean; + }[]; + } + + // No worker runs here, so the index passes earlier steps queued are still + // pending, and a later pass in the same lane would merge into one of them. + // Settled as a worker would have, so each step's passes are its own. Answers + // the latest job id, for `indexedSince`. + async function settleJobs(): Promise { + await dbAdapter.execute( + `UPDATE jobs SET status = 'resolved' WHERE status = 'unfulfilled'`, + ); + let [row] = (await dbAdapter.execute( + 'SELECT coalesce(max(id), 0) AS max FROM jobs', + )) as { max: number }[]; + return Number(row.max); + } + + // The paths every incremental index pass queued since `since` covers. + async function indexedSince(since: number): Promise { + let rows = (await dbAdapter.execute( + `SELECT args FROM jobs WHERE id > $1 AND job_type = 'incremental-index' + ORDER BY id`, + { bind: [since] }, + )) as { args: { changes: { url: string }[] } }[]; + return [ + ...new Set( + rows.flatMap(({ args }) => + args.changes.map(({ url }) => url.replace(realmURL, '')), + ), + ), + ].sort(); + } + + function updatedBy(realm: number): string[] { + return [ + ...new Set( + broadcasts + .filter( + ({ realm: from, event }) => + from === realm && event.eventName === 'update', + ) + .flatMap(({ event }) => [ + ...(event.added ?? []), + ...(event.updated ?? []), + ...(event.removed ?? []), + ]), + ), + ].sort(); + } + + async function handle(realm: Realm, request: Request) { + request.headers.set( + 'Authorization', + `Bearer ${createJWT(realm, 'owner', ['read', 'write', 'realm-owner'])}`, + ); + return (await realm.handle(request))!; + } + + module('a realm mounted recording for the first time', function () { + recordingTest( + 'takes in what the realm already held as one gap revision', + async function (assert) { + await mount(); + let head = await backend().head(); + assert.strictEqual(historyFrom(realmDir, head).length, 1); + let gap = readRevision(realmDir, head); + assert.strictEqual(gap.subject, GAP_MESSAGE); + assert.strictEqual(gap.trailers['Boxel-Gap'], 'recording-disabled'); + assert.strictEqual(gap.trailers['Boxel-Channel'], 'system'); + assert.deepEqual(gap.changes, ['A a.txt', 'A b.txt', 'A realm.json']); + let [row] = await journal(); + assert.strictEqual(row.kind, 'gap', 'the journal knows it is a gap'); + assert.strictEqual( + recoveries.at(-1)?.gapRevisions[0], + head, + 'reported as a gap, not drift', + ); + assert.deepEqual(recoveries.at(-1)?.driftRevisions, []); + assert.true( + (await realmHistoryState(dbAdapter, realmURL))?.recording, + 'the realm is noted as recording', + ); + + await mount(); + assert.strictEqual( + await backend().head(), + head, + 'a second mount records nothing', + ); + assert.strictEqual(recoveries.at(-1)?.outcome, 'clean'); + }, + ); + + recordingTest( + 'records one gap for what changed while recording was off', + async function (assert) { + let { realm: first } = await mount(); + await write(first, { 'a.txt': 'alpha 2' }, 'recorded-1'); + let before = await backend().head(); + + // A deploy that turns recording off for the realm. + let { realm: unrecorded } = await mount({ + recording: false, + historyConfigured: true, + }); + assert.false( + (await realmHistoryState(dbAdapter, realmURL))?.recording, + 'noted as not recording', + ); + await commitBatch( + unrecorded.batchCore, + [ + { + op: 'update', + href: `${realmURL}b.txt`, + content: 'bravo, unrecorded', + rawSource: true, + }, + ], + { waitForIndex: false }, + ); + assert.strictEqual( + await backend().head(), + before, + 'the unrecorded write made no revision', + ); + + // And one that turns it back on. + await mount(); + let head = await backend().head(); + assert.deepEqual( + historyFrom(realmDir, head).slice(0, 2), + [head, before], + 'exactly one revision was added', + ); + let gap = readRevision(realmDir, head); + assert.strictEqual(gap.subject, GAP_MESSAGE); + assert.deepEqual(gap.changes, ['M b.txt']); + assert.strictEqual( + await first.history!.revisionKind(head), + 'gap', + 'the revision is known to be a gap', + ); + assert.deepEqual(recoveries.at(-1)?.driftRevisions, [], 'not drift'); + }, + ); + }); + + module('a crash part-way through a write', function () { + for (let step of [ + 'afterRecord', + 'midMaterialize', + 'afterMaterialize', + ] as const) { + recordingTest( + `at ${step} is completed by the next mount`, + async function (assert) { + let { realm: crashed } = await mount(); + crashed.__testOnlySetHistoryFaults({ + [step]: async () => { + throw new Crash(step); + }, + }); + let err = await failure( + write( + crashed, + { 'c.txt': `charlie ${step}`, 'd.txt': `delta ${step}` }, + `crash-${step}`, + ), + ); + assert.strictEqual( + err.status, + 500, + `the write reports it was recorded and not completed: ${err.message}`, + ); + let head = await backend().head(); + assert.notStrictEqual( + await crashed.history!.materializedRevision(), + head, + 'disk is behind HEAD', + ); + + let jobsBefore = await settleJobs(); + let { realm } = await mount(); + assert.strictEqual(onDisk('c.txt'), `charlie ${step}`); + assert.strictEqual(onDisk('d.txt'), `delta ${step}`); + assert.strictEqual( + await realm.history!.materializedRevision(), + head, + 'disk is at HEAD', + ); + assert.deepEqual( + await indexedSince(jobsBefore), + ['c.txt', 'd.txt'], + 'both paths were queued for indexing', + ); + assert.deepEqual( + updatedBy(1), + ['c.txt', 'd.txt'], + 'and both were announced', + ); + assert.strictEqual(recoveries.at(-1)?.outcome, 'recovered'); + assert.strictEqual(recoveries.at(-1)?.caughtUpRevisions, 1); + let row = (await journal()).at(-1)!; + assert.true(row.materialized && row.post_committed, 'complete'); + assert.strictEqual( + await backend().head(), + head, + 'completing the write recorded nothing new', + ); + }, + ); + } + + recordingTest( + 'after post-commit and before the answer leaves nothing to do', + async function (assert) { + let { realm: crashed } = await mount(); + crashed.__testOnlySetHistoryFaults({ + afterPostCommit: async () => { + throw new Crash('afterPostCommit'); + }, + }); + await failure( + write(crashed, { 'c.txt': 'charlie late' }, 'crash-late'), + ); + let head = await backend().head(); + assert.strictEqual(onDisk('c.txt'), 'charlie late'); + + let jobsBefore = await settleJobs(); + let { realm } = await mount(); + assert.strictEqual(recoveries.at(-1)?.outcome, 'clean'); + assert.strictEqual(await realm.history!.materializedRevision(), head); + assert.deepEqual(await indexedSince(jobsBefore), [], 'no index pass'); + assert.strictEqual(await backend().head(), head); + }, + ); + + recordingTest( + 'two replicas mounting at once complete it once', + async function (assert) { + let { realm: crashed } = await mount(); + crashed.__testOnlySetHistoryFaults({ + afterRecord: async () => { + throw new Crash('afterRecord'); + }, + }); + await failure(write(crashed, { 'c.txt': 'charlie' }, 'crash-twice')); + let head = await backend().head(); + + recoveries = []; + await Promise.all([mount(), mount()]); + let outcomes = recoveries.map(({ outcome }) => outcome).sort(); + assert.deepEqual( + outcomes, + ['leased', 'recovered'], + 'one replica recovered, the other waited for it', + ); + assert.strictEqual(onDisk('c.txt'), 'charlie'); + assert.deepEqual( + [...updatedBy(1), ...updatedBy(2)], + ['c.txt'], + 'the path was written and announced once', + ); + assert.strictEqual(await backend().head(), head); + }, + ); + + recordingTest( + 'holds back a later write to its paths until recovery completes it', + async function (assert) { + let { realm: crashed } = await mount(); + crashed.__testOnlySetHistoryFaults({ + afterRecord: async () => { + throw new Crash('afterRecord'); + }, + }); + await failure(write(crashed, { 'c.txt': 'charlie' }, 'crash-guard')); + + // A peer replica that was already serving the realm, so no mount of + // its own brings disk up to HEAD first. + let { realm: peer } = await mount({ start: false }); + let err = await failure( + write(peer, { 'c.txt': 'charlie, again' }, 'after-crash'), + ); + assert.deepEqual( + { status: err.status, code: err.code }, + { status: 503, code: 'realm-recovering' }, + `refused while disk is behind HEAD there: ${err.message}`, + ); + // The refusal set recovery going; waiting on it joins that run. + await peer.recoverHistory({ trigger: 'write' }); + assert.strictEqual(onDisk('c.txt'), 'charlie', 'brought up to HEAD'); + await write(peer, { 'c.txt': 'charlie, again' }, 'after-crash-2'); + assert.strictEqual(onDisk('c.txt'), 'charlie, again'); + assert.strictEqual(await atHead('c.txt'), 'charlie, again'); + }, + ); + }); + + module('a write whose answer was lost', function () { + async function backdatePending() { + // Recovery leaves a pending write five minutes to finish before asking + // about it, since asking fences it. + await dbAdapter.execute( + `UPDATE realm_history_revisions + SET recorded_at = now() - interval '1 hour' + WHERE state = 'pending' AND realm_url = $1`, + { bind: [realmURL] }, + ); + } + + recordingTest( + 'that committed is completed, and a retry is pointed at its revision', + async function (assert) { + let { realm: first, history } = await mount(); + history!.loseCommitResponses = true; + history!.failOutcomes = true; + let err = await failure( + write(first, { 'c.txt': 'charlie' }, 'lost-commit'), + ); + assert.strictEqual(err.code, 'revision-outcome-unknown', err.message); + assert.strictEqual((await journal()).at(-1)?.state, 'pending'); + assert.strictEqual(onDisk('c.txt'), undefined, 'nothing on disk yet'); + let head = await backend().head(); + + await backdatePending(); + let { realm } = await mount(); + assert.strictEqual(recoveries.at(-1)?.resolvedCommitted, 1); + assert.strictEqual(onDisk('c.txt'), 'charlie', 'completed'); + let row = (await journal()).at(-1)!; + assert.deepEqual( + { state: row.state, revision: row.revision }, + { state: 'committed', revision: head }, + ); + + let retry = await failure( + write(realm, { 'c.txt': 'charlie' }, 'lost-commit'), + ); + assert.deepEqual( + { code: retry.code, revision: retry.meta?.revision }, + { code: 'revision-already-applied', revision: head }, + `the retry is told its write landed: ${retry.message}`, + ); + assert.strictEqual(await backend().head(), head, 'nothing rewritten'); + }, + ); + + recordingTest( + 'that never committed is fenced, and stays aborted', + async function (assert) { + let { realm: first, history } = await mount(); + let before = await backend().head(); + history!.dropCommits = true; + history!.failOutcomes = true; + await failure(write(first, { 'c.txt': 'charlie' }, 'lost-request')); + assert.strictEqual((await journal()).at(-1)?.state, 'pending'); + + await backdatePending(); + let { realm } = await mount(); + assert.strictEqual(recoveries.at(-1)?.resolvedAborted, 1); + assert.strictEqual((await journal()).at(-1)?.state, 'aborted'); + assert.strictEqual(onDisk('c.txt'), undefined); + assert.strictEqual(await backend().head(), before); + + let retry = await failure( + write(realm, { 'c.txt': 'charlie' }, 'lost-request'), + ); + assert.deepEqual( + { status: retry.status, code: retry.code }, + { status: 409, code: 'operation-fenced' }, + `a retry under the id is refused: ${retry.message}`, + ); + let late = await failure( + backend() + .prepareCommit({ + operationId: 'lost-request', + author: { name: 'late', email: 'late@example.com' }, + message: 'Update c.txt', + trailers: [], + changes: [ + { path: 'c.txt', content: new TextEncoder().encode('late') }, + ], + }) + .commit(before), + ); + assert.strictEqual( + late.code, + 'operation-fenced', + 'and so is a commit that arrives late', + ); + }, + ); + + recordingTest( + 'that a retry completes is journaled on its own parent', + async function (assert) { + let { realm, history } = await mount(); + let root = await backend().head(); + history!.loseCommitResponses = true; + history!.failOutcomes = true; + await failure(write(realm, { 'c.txt': 'charlie' }, 'lost-retried')); + let lost = await backend().head(); + history!.loseCommitResponses = false; + history!.failOutcomes = false; + + // A write to another file lands on top of the lost one before the + // retry comes. + await write(realm, { 'd.txt': 'delta' }, 'lands-between'); + let newest = await backend().head(); + await write(realm, { 'c.txt': 'charlie' }, 'lost-retried'); + + assert.strictEqual(await backend().head(), newest, 'nothing new'); + assert.strictEqual(onDisk('c.txt'), 'charlie'); + let rows = await journal(); + let row = (operationId: string) => + rows.find((r) => r.operation_id === operationId); + assert.deepEqual( + { + state: row('lost-retried')?.state, + revision: row('lost-retried')?.revision, + parent: row('lost-retried')?.parent, + }, + { state: 'committed', revision: lost, parent: root }, + 'the retried revision has the parent history gave it', + ); + assert.strictEqual(row('lands-between')?.parent, lost); + assert.strictEqual( + await realm.history!.materializedRevision(), + newest, + 'the newest revision is the one on top, though it was journaled first', + ); + }, + ); + + recordingTest( + 'that committed holds back a later write to its paths until recovery completes it', + async function (assert) { + let { realm, history } = await mount(); + history!.loseCommitResponses = true; + history!.failOutcomes = true; + await failure(write(realm, { 'c.txt': 'charlie' }, 'lost-then-edit')); + let head = await backend().head(); + assert.strictEqual(await atHead('c.txt'), 'charlie'); + assert.strictEqual(onDisk('c.txt'), undefined); + history!.loseCommitResponses = false; + history!.failOutcomes = false; + + let err = await failure( + write(realm, { 'c.txt': 'charlie, again' }, 'edit-after-lost'), + ); + assert.deepEqual( + { status: err.status, code: err.code }, + { status: 503, code: 'realm-recovering' }, + `refused while disk lacks the lost write: ${err.message}`, + ); + assert.strictEqual(await backend().head(), head, 'nothing recorded'); + assert.strictEqual( + (await journal()).find((row) => row.operation_id === 'lost-then-edit') + ?.state, + 'committed', + 'the lost write was settled without waiting out its grace period', + ); + + await realm.recoverHistory({ trigger: 'write' }); + assert.strictEqual(onDisk('c.txt'), 'charlie', 'brought up to HEAD'); + await write(realm, { 'c.txt': 'charlie, again' }, 'edit-after-lost'); + assert.strictEqual(await atHead('c.txt'), 'charlie, again'); + let retry = await failure( + write(realm, { 'c.txt': 'charlie' }, 'lost-then-edit'), + ); + assert.deepEqual( + { code: retry.code, revision: retry.meta?.revision }, + { code: 'revision-already-applied', revision: head }, + `a retry of the lost write is told it landed: ${retry.message}`, + ); + }, + ); + + recordingTest( + 'that never committed is fenced by a later write to its paths', + async function (assert) { + let { realm, history } = await mount(); + history!.dropCommits = true; + history!.failOutcomes = true; + await failure( + write(realm, { 'c.txt': 'charlie' }, 'dropped-then-edit'), + ); + history!.dropCommits = false; + history!.failOutcomes = false; + + await write(realm, { 'c.txt': 'charlie, again' }, 'edit-after-dropped'); + assert.strictEqual(onDisk('c.txt'), 'charlie, again'); + assert.strictEqual(await atHead('c.txt'), 'charlie, again'); + assert.strictEqual( + (await journal()).find( + (row) => row.operation_id === 'dropped-then-edit', + )?.state, + 'aborted', + ); + let retry = await failure( + write(realm, { 'c.txt': 'charlie' }, 'dropped-then-edit'), + ); + assert.deepEqual( + { status: retry.status, code: retry.code }, + { status: 409, code: 'operation-fenced' }, + `a retry of the dropped write is refused: ${retry.message}`, + ); + }, + ); + + recordingTest( + 'whose outcome still cannot be had holds back a later write to its paths', + async function (assert) { + let { realm, history } = await mount(); + history!.loseCommitResponses = true; + history!.failOutcomes = true; + await failure(write(realm, { 'c.txt': 'charlie' }, 'lost-unreachable')); + let head = await backend().head(); + history!.loseCommitResponses = false; + + let err = await failure( + write(realm, { 'c.txt': 'charlie, again' }, 'edit-unreachable'), + ); + assert.deepEqual( + { status: err.status, code: err.code }, + { status: 503, code: 'revision-aborted' }, + err.message, + ); + assert.strictEqual(await backend().head(), head, 'nothing recorded'); + assert.strictEqual( + (await journal()).find( + (row) => row.operation_id === 'lost-unreachable', + )?.state, + 'pending', + ); + }, + ); + }); + + module('an edit made to the realm directory directly', function () { + recordingTest( + 'is recorded at the next mount as a system revision', + async function (assert) { + let { realm: first } = await mount(); + let before = await backend().head(); + writeFileSync(join(realmDir, 'a.txt'), 'alpha, by hand'); + writeFileSync(join(realmDir, 'e.txt'), 'echo'); + removeSync(join(realmDir, 'b.txt')); + + let jobsBefore = await settleJobs(); + await mount(); + let head = await backend().head(); + assert.deepEqual(historyFrom(realmDir, head).slice(0, 2), [ + head, + before, + ]); + let drift = readRevision(realmDir, head); + assert.strictEqual(drift.subject, DRIFT_MESSAGE); + assert.strictEqual(drift.trailers['Boxel-Channel'], 'system'); + assert.notStrictEqual(drift.author, 'owner', 'authored by no user'); + assert.deepEqual(drift.changes, ['A e.txt', 'D b.txt', 'M a.txt']); + assert.strictEqual(await first.history!.revisionKind(head), 'drift'); + let event = recoveries.at(-1)!; + assert.deepEqual(event.driftRevisions, [head]); + assert.strictEqual(event.driftPaths, 3, 'the alert has its count'); + assert.deepEqual(await indexedSince(jobsBefore), [ + 'a.txt', + 'b.txt', + 'e.txt', + ]); + }, + ); + + recordingTest( + 'is recorded by the scheduled check, which leaves a clean realm alone', + async function (assert) { + let { realm } = await mount(); + let head = await backend().head(); + let jobsBefore = await settleJobs(); + let task = realmHistoryRecovery({ + dbAdapter, + reportStatus: () => undefined, + log: { info: () => undefined, warn: () => undefined }, + getAuthedFetch: async () => + (async (input: RequestInfo | URL) => + await handle( + realm, + new Request(String(input), { + method: 'POST', + headers: { Accept: 'application/json' }, + }), + )) as unknown as typeof globalThis.fetch, + } as any); + let { realms } = await task({}); + assert.deepEqual(realms, { [realmURL]: 'clean' }); + assert.strictEqual(await backend().head(), head, 'no revision'); + assert.deepEqual(await indexedSince(jobsBefore), [], 'no index pass'); + + writeFileSync(join(realmDir, 'a.txt'), 'alpha, by hand'); + ({ realms } = await task({})); + assert.deepEqual(realms, { [realmURL]: 'recovered' }); + let drift = readRevision(realmDir, await backend().head()); + assert.strictEqual(drift.subject, DRIFT_MESSAGE); + assert.deepEqual(drift.changes, ['M a.txt']); + }, + ); + }); + + module('a realm whose recovery fails', function () { + async function crashAfterRecord() { + let { realm: crashed } = await mount(); + crashed.__testOnlySetHistoryFaults({ + afterRecord: async () => { + throw new Crash('afterRecord'); + }, + }); + await failure(write(crashed, { 'c.txt': 'charlie' }, 'crash-sealed')); + await settleJobs(); + } + + async function assertReadOnly(assert: Assert, realm: Realm) { + let write = await handle( + realm, + new Request(`${realmURL}f.txt`, { + method: 'POST', + headers: { Accept: 'application/vnd.card+source' }, + body: 'foxtrot', + }), + ); + assert.strictEqual(write.status, 423, 'a write is refused'); + assert.strictEqual( + write.headers.get('X-Boxel-Realm-Read-Only'), + 'recovery-failed', + ); + let body = JSON.parse(await write.text()); + assert.strictEqual(body.errors[0].code, 'realm-read-only'); + assert.false(existsSync(join(realmDir, 'f.txt')), 'nothing written'); + + let read = await handle( + realm, + new Request(`${realmURL}a.txt`, { + headers: { Accept: 'application/vnd.card+source' }, + }), + ); + assert.strictEqual(read.status, 200, 'a read is served'); + let ready = await handle( + realm, + new Request(`${realmURL}_readiness-check`, { + headers: { Accept: 'application/vnd.api+json' }, + }), + ); + assert.strictEqual(ready.status, 200, 'the realm is ready to read'); + assert.strictEqual( + ready.headers.get('X-Boxel-Realm-Read-Only'), + 'recovery-failed', + ); + } + + recordingTest( + 'with history unreachable goes read-only until an operator clears it', + async function (assert) { + await crashAfterRecord(); + let { realm, history } = await mount({ + faulty: (history) => (history.failReads = true), + }); + assert.strictEqual(recoveries.at(-1)?.outcome, 'read-only'); + assert.ok(recoveries.at(-1)?.error, 'with its reason'); + await assertReadOnly(assert, realm); + + // Remounting, or the schedule, leaves it read-only. + history!.failReads = false; + assert.strictEqual( + await realm.recoverHistory({ trigger: 'schedule' }), + undefined, + 'the schedule does not clear it', + ); + await assertReadOnly(assert, realm); + + let cleared = await realm.recoverHistory({ trigger: 'operator' }); + assert.strictEqual(cleared?.outcome, 'recovered'); + assert.strictEqual(onDisk('c.txt'), 'charlie', 'completed'); + let write = await handle( + realm, + new Request(`${realmURL}f.txt`, { + method: 'POST', + headers: { Accept: 'application/vnd.card+source' }, + body: 'foxtrot', + }), + ); + assert.strictEqual(write.status, 204, 'writes are back'); + }, + ); + + recordingTest( + 'with a corrupt history goes read-only', + async function (assert) { + await crashAfterRecord(); + removeSync( + join(realmDir, HISTORY_DIR, 'jj', 'store', 'git', 'objects'), + ); + let { realm } = await mount(); + assert.strictEqual(recoveries.at(-1)?.outcome, 'read-only'); + await assertReadOnly(assert, realm); + }, + ); + }); + + module('a realm that records no history', function () { + recordingTest('runs no recovery', async function (assert) { + await mount({ recording: false }); + assert.deepEqual(recoveries, []); + assert.strictEqual( + await realmHistoryState(dbAdapter, realmURL), + undefined, + 'and keeps no recovery state', + ); + assert.false(existsSync(join(realmDir, HISTORY_DIR))); + }); + }); +}); diff --git a/packages/realm-server/tests/realm-history-write-test.ts b/packages/realm-server/tests/realm-history-write-test.ts index 11f329b12d3..1466acbb59a 100644 --- a/packages/realm-server/tests/realm-history-write-test.ts +++ b/packages/realm-server/tests/realm-history-write-test.ts @@ -1,8 +1,7 @@ import QUnit from 'qunit'; const { module, test } = QUnit; import { basename, join } from 'path'; -import { execFileSync, spawn, type ChildProcess } from 'child_process'; -import { createServer } from 'net'; +import type { ChildProcess } from 'child_process'; import fsExtra from 'fs-extra'; import { dirSync, setGracefulCleanup, type DirResult } from 'tmp'; import type { SuperTest, Test } from 'supertest'; @@ -30,9 +29,19 @@ import { type HistoryCommitResult, type HistoryOutcome, type PreparedCommit, + type HistoryReadResult, + type HistoryTreeEntry, } from '@cardstack/runtime-common/realm-history'; import type { PgAdapter } from '@cardstack/postgres'; import { DeckdHistoryBackend } from '../lib/deckd-client.ts'; +import { + ROOT_REVISION, + findDeckd, + git, + historyFrom, + readRevision, + startDeckd, +} from './helpers/realm-history.ts'; import { createJWT, createRealm, @@ -52,11 +61,6 @@ const { ensureDirSync, existsSync, readFileSync, statSync, writeJSONSync } = // visible anywhere — on disk, in Postgres, to a realm-event subscriber — until // that revision exists. These run against a real deckd, the history backend // realm-server talks to in every environment. -// -// deckd's API reads no revision's contents, so they are read straight out of -// the git store inside the repository deckd keeps (`jj/store/git`), with git. - -const ROOT_REVISION = '0'.repeat(40); const personGts = ` import { contains, field, CardDef } from "@cardstack/base/card-api"; @@ -78,65 +82,6 @@ function person(firstName: string, lastName: string): LooseSingleCardDocument { }; } -// --- deckd ------------------------------------------------------------------ - -// The deckd binary: `DECKD_BIN`, or `deckd` on PATH. CI provides one, so there -// its absence is a failure; a developer without one skips this module. -function findDeckd(): string | undefined { - if (process.env.DECKD_BIN) { - return process.env.DECKD_BIN; - } - try { - return ( - execFileSync('which', ['deckd'], { stdio: ['ignore', 'pipe', 'ignore'] }) - .toString() - .trim() || undefined - ); - } catch { - return undefined; - } -} - -async function freePort(): Promise { - return await new Promise((resolve, reject) => { - let server = createServer(); - server.once('error', reject); - server.listen(0, '127.0.0.1', () => { - let address = server.address(); - server.close(() => - typeof address === 'object' && address - ? resolve(address.port) - : reject(new Error('no port')), - ); - }); - }); -} - -async function startDeckd( - binary: string, -): Promise<{ url: string; process: ChildProcess }> { - let port = await freePort(); - let child = spawn(binary, [], { - env: { ...process.env, DECKD_BIND: `127.0.0.1:${port}` }, - stdio: ['ignore', 'ignore', 'inherit'], - }); - let url = `http://127.0.0.1:${port}`; - let deadline = Date.now() + 20_000; - while (Date.now() < deadline) { - try { - let response = await fetch(`${url}/health`); - if (response.ok) { - return { url, process: child }; - } - } catch { - // not listening yet - } - await new Promise((resolve) => setTimeout(resolve, 100)); - } - child.kill(); - throw new Error(`deckd at ${binary} did not answer /health within 20s`); -} - // deckd, with the faults a test asks for injected around it. class FaultyHistory implements HistoryBackend { inner: DeckdHistoryBackend; @@ -184,68 +129,14 @@ class FaultyHistory implements HistoryBackend { head(): Promise { return this.inner.head(); } -} - -// --- reading a revision ------------------------------------------------------- - -function git(realmDir: string, ...args: string[]): string { - return execFileSync('git', [ - '--git-dir', - join(realmDir, HISTORY_DIR, 'jj', 'store', 'git'), - ...args, - ]).toString(); -} -interface Revision { - author: string; - email: string; - parents: string[]; - subject: string; - trailers: Record; - // `A path`, `M path` or `D path`, sorted. - changes: string[]; -} - -function readRevision(realmDir: string, revision: string): Revision { - let raw = git( - realmDir, - 'show', - '--no-renames', - '--name-status', - '--format=%an%x00%ae%x00%P%x00%B%x00', - revision, - ); - let [author, email, parents, body, rest] = raw.split('\0'); - let trailers: Record = {}; - for (let line of body.split('\n')) { - let match = /^([A-Za-z0-9-]+): (.*)$/.exec(line); - if (match) { - trailers[match[1]] = match[2]; - } + tree(revision: string): Promise { + return this.inner.tree(revision); } - return { - author, - email, - parents: parents.trim() ? parents.trim().split(' ') : [], - subject: body.split('\n')[0], - trailers, - changes: rest - .split('\n') - .map((line) => line.trim()) - .filter(Boolean) - .map((line) => line.replace(/\t/, ' ')) - .sort(), - }; -} -// Every revision from HEAD back, newest first. -function historyFrom(realmDir: string, head: string): string[] { - if (head === ROOT_REVISION) { - return []; + read(revision: string, paths: string[]): Promise { + return this.inner.read(revision, paths); } - return git(realmDir, 'rev-list', '--first-parent', head) - .split('\n') - .filter(Boolean); } const deckdBinary = findDeckd(); @@ -316,7 +207,10 @@ module(basename(import.meta.filename), function (hooks) { }, }); - hooks.beforeEach(function () { + hooks.beforeEach(async function () { + // Mounting took in the realm's files as its first revision, a gap, and + // queued their index pass. Each test starts once that has landed. + await realm.indexing(); events = []; setRevisionWriteSink((event) => events.push(event)); }); @@ -401,10 +295,9 @@ module(basename(import.meta.filename), function (hooks) { ); let recorded = readRevision(realmDir, revision); assert.strictEqual(recorded.author, 'john', 'authored by the session'); - // The realm's history begins empty, so a file the realm held before - // recording began enters history as added, with its first recorded - // write. The message describes the write, which updated the file. - assert.deepEqual(recorded.changes, ['A person-1.json']); + // Mounting took in what the realm already held as its first revision, + // so the write modifies a file history already has. + assert.deepEqual(recorded.changes, ['M person-1.json']); assert.strictEqual(recorded.subject, 'Update person-1.json'); assert.deepEqual( { @@ -497,18 +390,18 @@ module(basename(import.meta.filename), function (hooks) { ); assert.strictEqual(batch.status, 200, JSON.stringify(batch.body)); let afterBatch = readRevision(realmDir, batch.get(REVISION_HEADER)!); - // Added to history rather than modified, as in the first test: none - // of the three had been recorded before. + // Modified rather than added, as in the first test: mounting took in + // all three. assert.deepEqual(afterBatch.changes, [ - 'A person-1.json', - 'A person-2.json', - 'A person-3.json', + 'M person-1.json', + 'M person-2.json', + 'M person-3.json', ]); assert.strictEqual(afterBatch.trailers['Boxel-Channel'], 'operations'); assert.strictEqual( historyFrom(realmDir, batch.get(REVISION_HEADER)!).length, - 3, - 'three requests, three revisions', + 4, + 'three requests, three revisions, after the one mounting recorded', ); }, ); @@ -587,6 +480,7 @@ module(basename(import.meta.filename), function (hooks) { 'a write that fails before it is recorded writes nothing', async function (assert) { let before = stored('person-1.json'); + let head = await history.head(); let broadcasts = 0; let broadcast = adapter.broadcastRealmEvent.bind(adapter); adapter.broadcastRealmEvent = async (...args) => { @@ -609,11 +503,7 @@ module(basename(import.meta.filename), function (hooks) { adapter.broadcastRealmEvent = broadcast; } assert.strictEqual(stored('person-1.json'), before, 'disk unchanged'); - assert.strictEqual( - await history.head(), - ROOT_REVISION, - 'nothing recorded', - ); + assert.strictEqual(await history.head(), head, 'nothing recorded'); assert.strictEqual(broadcasts, 0, 'nothing broadcast'); assert.strictEqual(events.at(-1)?.outcome, 'aborted'); assert.strictEqual(events.at(-1)?.revision, null); @@ -677,7 +567,6 @@ module(basename(import.meta.filename), function (hooks) { .at(-1) ?.historyCalls.map(({ endpoint, status }) => [endpoint, status]), [ - ['head', 'ok'], ['commit', 'transport'], ['outcome', 'ok'], ], diff --git a/packages/realm-server/tests/worker-job-registration-test.ts b/packages/realm-server/tests/worker-job-registration-test.ts index 6e7318c824c..cac9f0663cd 100644 --- a/packages/realm-server/tests/worker-job-registration-test.ts +++ b/packages/realm-server/tests/worker-job-registration-test.ts @@ -68,6 +68,7 @@ module(basename(import.meta.filename), function () { 'media-cache-gc', 'prerender-html-reconcile', 'prerender_html', + 'realm-history-recovery', 'run-command', 'scoped-css-gc', 'screenshot-card', diff --git a/packages/runtime-common/card-operations/index.ts b/packages/runtime-common/card-operations/index.ts index f30c5e81e29..c2687e9ac58 100644 --- a/packages/runtime-common/card-operations/index.ts +++ b/packages/runtime-common/card-operations/index.ts @@ -158,8 +158,11 @@ export { setPolicySnapshotReadSink, setRevisionWriteSink, emitRevisionWrite, + setRealmRecoverySink, + emitRealmRecovery, } from './telemetry.ts'; export type { + RealmRecoveryEvent, CapabilityCheckEvent, KeyUseEvent, KeyUseOutcome, diff --git a/packages/runtime-common/card-operations/telemetry.ts b/packages/runtime-common/card-operations/telemetry.ts index c17de43e5c7..ea12ab2e0a3 100644 --- a/packages/runtime-common/card-operations/telemetry.ts +++ b/packages/runtime-common/card-operations/telemetry.ts @@ -800,3 +800,72 @@ let recordFailureLog: ReturnType | undefined; export function elapsedMs(started: number): number { return Math.round((performance.now() - started) * 100) / 100; } + +// ============================================================================ +// Per-recovery telemetry for a realm that records history. +// +// One line each time a realm checks itself against its history — at mount, on +// the schedule, when a write finds disk behind HEAD, or when an operator asks — +// whatever it found. Counted by `outcome` and by the revisions it recorded, +// these are the recoveries, drift revisions, gap revisions and realms put +// read-only that alerting watches. The scan timings are what drift detection +// costs on a realm of `fileCount` files. +// ============================================================================ + +export interface RealmRecoveryEvent { + kind: 'realm-recovery'; + realmURL: string; + trigger: 'mount' | 'schedule' | 'write' | 'operator'; + // `clean`: nothing to do. `recovered`: something was completed or recorded. + // `read-only`: recovery failed and the realm refuses writes. `leased`: + // another replica was recovering the realm. + outcome: 'clean' | 'recovered' | 'read-only' | 'leased'; + // Pending writes whose outcome recovery asked for, by what it was. + resolvedCommitted: number; + resolvedAborted: number; + // Revisions whose bytes, or whose index pass, recovery completed. + caughtUpRevisions: number; + caughtUpPaths: number; + // What the disk scan found that no revision explained, and the revisions + // that took it in. A realm that last ran without recording takes it in as + // a gap; any other realm has drifted. Counts beside the lists, because + // alert queries filter on numbers. + driftPaths: number; + driftRevisions: string[]; + gapPaths: number; + gapRevisions: string[]; + // Null where recovery failed before scanning. + fileCount: number | null; + bytesHashed: number | null; + scanMs: number | null; + treeMs: number | null; + totalMs: number; + // Why the realm went read-only. + error: string | null; +} + +let realmRecoverySink: ((event: RealmRecoveryEvent) => void) | undefined; + +export function setRealmRecoverySink( + sink: ((event: RealmRecoveryEvent) => void) | undefined, +): void { + realmRecoverySink = sink; +} + +let realmRecoveryLog: ReturnType | undefined; + +export function emitRealmRecovery(event: RealmRecoveryEvent): void { + if (realmRecoverySink) { + realmRecoverySink(event); + return; + } + let log = (realmRecoveryLog ??= logger(OPERATIONS_CHANNEL)); + let line = JSON.stringify({ channel: OPERATIONS_CHANNEL, ...event }); + if (event.outcome === 'read-only') { + log.error(line); + } else if (event.driftRevisions.length > 0) { + log.warn(line); + } else { + log.info(line); + } +} diff --git a/packages/runtime-common/card-operations/types.ts b/packages/runtime-common/card-operations/types.ts index 84d9e0ba6d4..4c72999e9e6 100644 --- a/packages/runtime-common/card-operations/types.ts +++ b/packages/runtime-common/card-operations/types.ts @@ -1720,6 +1720,14 @@ export type OperationErrorCode = // The write's operation id already committed a different write. Carries a // 409 and the revision it committed. | 'operation-id-conflict' + // The realm's history reported the write's operation id absent, which fences + // it: the write never happened and can no longer. Carries a 409; the write + // is sent again under a new operation id. + | 'operation-fenced' + // A revision touching one of the write's paths was recorded and has not all + // reached disk, so disk is behind HEAD there and the realm is completing it. + // Carries a 503 and that revision; the same write sent again is the remedy. + | 'realm-recovering' // The operation could not be carried out for a reason that is not the // caller's — an unreadable definition, an errored index row, a failure // inside the executor. diff --git a/packages/runtime-common/db-queries/realm-history-state-queries.ts b/packages/runtime-common/db-queries/realm-history-state-queries.ts new file mode 100644 index 00000000000..019d11524b8 --- /dev/null +++ b/packages/runtime-common/db-queries/realm-history-state-queries.ts @@ -0,0 +1,199 @@ +// Read/write helpers for `realm_history_state`: one row per realm a +// history-aware realm-server has mounted, saying whether it recorded then, and +// carrying what recovery leaves behind — the read-only seal a failed recovery +// sets, and the lease that keeps two replicas from recovering a realm at once. +// +// Postgres only: only realm-server realms record history. +import type { DBAdapter } from '../db.ts'; +import { param, query } from '../expression.ts'; + +export interface RealmHistoryState { + recording: boolean; + readOnly?: { at: string; reason: string; detail?: string }; +} + +// Why a realm that records history refuses writes. +export type ReadOnlyReason = 'recovery-failed'; + +export async function realmHistoryState( + dbAdapter: DBAdapter, + realmURL: string, +): Promise { + let [row] = (await query(dbAdapter, [ + `SELECT recording, read_only_at, read_only_reason, read_only_detail + FROM realm_history_state WHERE realm_url =`, + param(realmURL), + ])) as { + recording: boolean; + read_only_at: string | null; + read_only_reason: string | null; + read_only_detail: string | null; + }[]; + if (!row) { + return undefined; + } + return { + recording: Boolean(row.recording), + ...(row.read_only_at != null + ? { + readOnly: { + at: String(row.read_only_at), + reason: String(row.read_only_reason ?? 'recovery-failed'), + ...(row.read_only_detail != null + ? { detail: String(row.read_only_detail) } + : {}), + }, + } + : {}), + }; +} + +// Note whether the realm records its writes as of this mount. A realm whose +// row says it did not, or that has none, is taken in as a gap the next time it +// is mounted recording. +export async function setRealmRecording( + dbAdapter: DBAdapter, + realmURL: string, + recording: boolean, +): Promise { + await query(dbAdapter, [ + `INSERT INTO realm_history_state (realm_url, recording) VALUES (`, + param(realmURL), + `,`, + param(recording), + `) ON CONFLICT (realm_url) DO UPDATE SET recording = EXCLUDED.recording, + recording_changed_at = now() + WHERE realm_history_state.recording IS DISTINCT FROM EXCLUDED.recording`, + ]); +} + +// The realm's read-only seal, or nothing for a realm that takes writes. Read +// fresh on every write, so a peer's seal or clear takes effect at once. +export async function realmReadOnly( + dbAdapter: DBAdapter, + realmURL: string, +): Promise<{ reason: string; detail?: string } | undefined> { + let [row] = (await query(dbAdapter, [ + `SELECT read_only_reason, read_only_detail FROM realm_history_state + WHERE read_only_at IS NOT NULL AND realm_url =`, + param(realmURL), + ])) as { + read_only_reason: string | null; + read_only_detail: string | null; + }[]; + if (!row) { + return undefined; + } + return { + reason: String(row.read_only_reason ?? 'recovery-failed'), + ...(row.read_only_detail != null + ? { detail: String(row.read_only_detail) } + : {}), + }; +} + +export async function setRealmReadOnly( + dbAdapter: DBAdapter, + realmURL: string, + reason: ReadOnlyReason, + detail: string, +): Promise { + await query(dbAdapter, [ + `INSERT INTO realm_history_state + (realm_url, recording, read_only_at, read_only_reason, read_only_detail) + VALUES (`, + param(realmURL), + `, true, now(),`, + param(reason), + `,`, + param(detail), + `) ON CONFLICT (realm_url) DO UPDATE SET read_only_at = now(), + read_only_reason = EXCLUDED.read_only_reason, + read_only_detail = EXCLUDED.read_only_detail`, + ]); +} + +export async function clearRealmReadOnly( + dbAdapter: DBAdapter, + realmURL: string, +): Promise { + await query(dbAdapter, [ + `UPDATE realm_history_state SET read_only_at = NULL, + read_only_reason = NULL, read_only_detail = NULL + WHERE realm_url =`, + param(realmURL), + ]); +} + +// Take the realm's recovery lease for `owner`, for `ms`. Resolves false while +// another owner holds an unexpired one. A lease rather than an advisory lock: +// recovery takes other pool connections while it runs, and a lock held across +// those, with other replicas' waiters each pinning a connection of their own, +// is how a pool runs dry. +export async function claimRecoveryLease( + dbAdapter: DBAdapter, + realmURL: string, + owner: string, + ms: number, +): Promise { + let rows = await query(dbAdapter, [ + `INSERT INTO realm_history_state + (realm_url, recording, recovery_lease_owner, recovery_lease_until) + VALUES (`, + param(realmURL), + `, false,`, + param(owner), + `, now() + CAST(`, + param(ms), + `AS float8) * interval '1 millisecond') ON CONFLICT (realm_url) DO UPDATE SET + recovery_lease_owner = EXCLUDED.recovery_lease_owner, + recovery_lease_until = EXCLUDED.recovery_lease_until + WHERE realm_history_state.recovery_lease_until IS NULL + OR realm_history_state.recovery_lease_until < now() + OR realm_history_state.recovery_lease_owner = EXCLUDED.recovery_lease_owner + RETURNING realm_url`, + ]); + return rows.length > 0; +} + +export async function releaseRecoveryLease( + dbAdapter: DBAdapter, + realmURL: string, + owner: string, + outcome: string, +): Promise { + await query(dbAdapter, [ + `UPDATE realm_history_state SET recovery_lease_owner = NULL, + recovery_lease_until = NULL, last_recovered_at = now(), + last_recovery_outcome =`, + param(outcome), + `WHERE realm_url =`, + param(realmURL), + `AND recovery_lease_owner =`, + param(owner), + ]); +} + +// Whether a replica holds an unexpired recovery lease on the realm. +export async function recoveryLeased( + dbAdapter: DBAdapter, + realmURL: string, +): Promise { + let rows = await query(dbAdapter, [ + `SELECT 1 FROM realm_history_state WHERE recovery_lease_until > now() + AND realm_url =`, + param(realmURL), + ]); + return rows.length > 0; +} + +// The realms recovery's schedule visits: those that recorded when last +// mounted. +export async function recordingRealmURLs( + dbAdapter: DBAdapter, +): Promise { + let rows = await query(dbAdapter, [ + `SELECT realm_url FROM realm_history_state WHERE recording ORDER BY realm_url`, + ]); + return rows.map((row) => String(row.realm_url)); +} diff --git a/packages/runtime-common/index.ts b/packages/runtime-common/index.ts index e3bb1cbf1a1..aa83d3db096 100644 --- a/packages/runtime-common/index.ts +++ b/packages/runtime-common/index.ts @@ -1996,6 +1996,7 @@ export { createResponse } from './create-response.ts'; export * from './db-queries/db-types.ts'; export * from './db-queries/realm-metadata-queries.ts'; +export * from './db-queries/realm-history-state-queries.ts'; export * from './db-queries/realm-permission-queries.ts'; export * from './db-queries/session-room-queries.ts'; export * from './db-queries/realm-event-room-queries.ts'; diff --git a/packages/runtime-common/realm-history-recovery.ts b/packages/runtime-common/realm-history-recovery.ts new file mode 100644 index 00000000000..664ec68ea47 --- /dev/null +++ b/packages/runtime-common/realm-history-recovery.ts @@ -0,0 +1,412 @@ +import { logger } from './log.ts'; +import type { DBAdapter } from './db.ts'; +import type { LocalPath } from './paths.ts'; +import { isHistoryPath, isPartialWritePath } from './paths.ts'; +import { + gitBlobId, + newOperationId, + type HistoryChange, + type RealmHistoryRecorder, + type RevisionKind, + type WriteRecording, +} from './realm-history.ts'; +import { + claimRecoveryLease, + realmHistoryState, + recoveryLeased, + releaseRecoveryLease, + setRealmReadOnly, + setRealmRecording, +} from './db-queries/realm-history-state-queries.ts'; +import { + emitRealmRecovery, + type RealmRecoveryEvent, +} from './card-operations/telemetry.ts'; + +// ============================================================================ +// Realm recovery: bringing a realm that records history back to agreement +// with it. +// +// History is the realm's truth and disk is a materialization of HEAD, so +// after a crash, a lost answer from history, or an edit made to disk directly, +// recovery leaves the realm in one of three states: +// +// - disk matches HEAD, and every revision has been indexed and announced; +// - what disk held that no revision explained has been recorded as a system +// revision — drift, or a gap left while the realm did not record; or +// - the realm is read-only: it serves reads and refuses writes until an +// operator clears it (see docs/realm-history-recovery-runbook.md). +// +// It runs in order: +// +// 1. Pending writes — those journaled before they asked history to record, +// whose answer never came back — are settled by asking history for their +// outcome. Asking fences each one, so an absent write can never commit +// late. +// 2. Revisions recorded but not all on disk, or never indexed, are +// completed: their paths are brought to what HEAD holds, under those +// paths' write locks, through the realm's own commit, which announces +// and indexes them. +// 3. Disk is compared with HEAD. What still differs is recorded as one +// revision — a gap when the realm last ran without recording, drift +// otherwise — after a second look under the paths' write locks, so a file +// a peer has just written is never mistaken for one written out of band. +// +// Any failure leaves the realm read-only, and the mount it runs in goes on. +// ============================================================================ + +// How long a pending write is left to finish before recovery asks history for +// its outcome. Asking fences the operation, so asking about a write still in +// flight would abort it. A write's calls to history give up well inside this. +export const PENDING_GRACE_MS = 5 * 60 * 1000; + +// How long a recovery lease lasts. A replica that dies holding one stops +// excluding the others when it lapses. +const LEASE_MS = 10 * 60 * 1000; + +// How long a replica mounting a realm another replica is recovering waits for +// it, before mounting anyway. Writes to anything recovery has yet to bring up +// to HEAD wait for it either way. +const LEASE_WAIT_MS = 60 * 1000; + +// The most paths one drift or gap revision takes, which bounds the file locks +// held while it is recorded. A realm with more changes is taken in over +// several revisions that share a `Boxel-Recovery-Id` trailer. +const MAX_PATHS_PER_REVISION = 2000; + +export const DRIFT_MESSAGE = 'Unattributed external changes'; +export const GAP_MESSAGE = 'Unrecorded changes (recording was disabled)'; + +export type RecoveryTrigger = RealmRecoveryEvent['trigger']; + +// What recovery needs of the realm it recovers. +export interface RecoveryHost { + url: string; + history: RealmHistoryRecorder; + dbAdapter: DBAdapter; + // Every file in the realm directory a revision could hold, with its bytes. + listFiles(): AsyncIterable; + readFile(path: LocalPath): Promise; + withFileWriteLocks( + paths: LocalPath[], + fn: (releaseLocks: () => void) => Promise, + ): Promise; + // Bring `paths` to `contents` (absent meaning removed) through the realm's + // commit, announcing and indexing every one of them, and stamp `revisions` + // materialized and post-committed. Called holding the paths' locks. + applyRevisions( + contents: Map, + revisions: string[], + releaseLocks: () => void, + ): Promise; + // A system revision's author: the realm's own matrix user. + systemRecording(): WriteRecording; + reportError(message: string, err: unknown): void; +} + +export interface RecoveryResult { + outcome: RealmRecoveryEvent['outcome']; + event: RealmRecoveryEvent; +} + +export async function recoverRealmHistory( + host: RecoveryHost, + trigger: RecoveryTrigger, + opts: { leaseOwner: string }, +): Promise { + let log = logger('realm:history-recovery'); + let started = Date.now(); + let owner = opts.leaseOwner; + let event: RealmRecoveryEvent = { + kind: 'realm-recovery', + realmURL: host.url, + trigger, + outcome: 'clean', + resolvedCommitted: 0, + resolvedAborted: 0, + caughtUpRevisions: 0, + caughtUpPaths: 0, + driftPaths: 0, + driftRevisions: [], + gapPaths: 0, + gapRevisions: [], + fileCount: null, + bytesHashed: null, + scanMs: null, + treeMs: null, + totalMs: 0, + error: null, + }; + let finish = (outcome: RealmRecoveryEvent['outcome']) => { + event.outcome = outcome; + event.totalMs = Date.now() - started; + emitRealmRecovery(event); + return { outcome, event }; + }; + + if (!(await claimRecoveryLease(host.dbAdapter, host.url, owner, LEASE_MS))) { + if (trigger === 'mount') { + // Another replica is recovering the realm. Waiting for it keeps this + // replica from serving disk it is about to change. + let deadline = Date.now() + LEASE_WAIT_MS; + while ( + Date.now() < deadline && + (await recoveryLeased(host.dbAdapter, host.url)) + ) { + await new Promise((resolve) => setTimeout(resolve, 250)); + } + } + return finish('leased'); + } + + let outcome: RealmRecoveryEvent['outcome'] = 'clean'; + try { + let state = await realmHistoryState(host.dbAdapter, host.url); + + // 1. Pending writes. + for (let operationId of await host.history.pendingOperations( + new Date(Date.now() - PENDING_GRACE_MS), + )) { + if (await host.history.resolvePending(operationId)) { + event.resolvedCommitted++; + } else { + event.resolvedAborted++; + } + } + + // 2. Revisions not complete. + let incomplete = await host.history.incompleteRevisions(); + if (incomplete.length > 0) { + let paths = [ + ...new Set(incomplete.flatMap(({ paths }) => paths)), + ] as LocalPath[]; + let revisions = incomplete.map(({ revision }) => revision); + await host.withFileWriteLocks(paths, async (releaseLocks) => { + // Read under the locks, so no write to these paths can land between + // reading HEAD and bringing disk to it. HEAD rather than each revision + // in turn: it is the newest state of every one of these paths, so no + // write that landed after the one being completed is rolled back. + let head = await host.history.head(); + let contents = await host.history.read(head, paths); + await host.applyRevisions( + contents as Map, + revisions, + releaseLocks, + ); + }); + event.caughtUpRevisions = revisions.length; + event.caughtUpPaths = paths.length; + outcome = 'recovered'; + log.info( + `${host.url}: completed ${revisions.length} revision(s) across ${paths.length} path(s)`, + ); + } + + // 3. Disk against HEAD. + let kind: RevisionKind = state?.recording ? 'drift' : 'gap'; + let differing = await scan(host, event); + if (differing.length > 0) { + let recorded = await recordDifferences(host, differing, kind); + if (recorded.revisions.length > 0) { + outcome = 'recovered'; + if (kind === 'gap') { + event.gapPaths = recorded.paths; + event.gapRevisions = recorded.revisions; + } else { + event.driftPaths = recorded.paths; + event.driftRevisions = recorded.revisions; + // Unsupported: an edit made to disk directly. Raised so whoever made + // it hears that it was taken in, under the realm's own name. + host.reportError( + `${host.url} had ${recorded.paths} file(s) changed outside the realm, recorded as ${recorded.revisions.join(', ')}`, + undefined, + ); + } + } + } + if (!state?.recording) { + await setRealmRecording(host.dbAdapter, host.url, true); + } + } catch (err: unknown) { + outcome = 'read-only'; + event.error = describe(err); + try { + await setRealmReadOnly( + host.dbAdapter, + host.url, + 'recovery-failed', + event.error, + ); + } catch (sealErr: unknown) { + log.error( + `${host.url}: could not seal the realm read-only after recovery failed: ${describe(sealErr)}`, + ); + } + host.reportError( + `${host.url} could not recover from its history, and refuses writes until an operator clears it: ${event.error}`, + err, + ); + } finally { + try { + await releaseRecoveryLease(host.dbAdapter, host.url, owner, outcome); + } catch (err: unknown) { + // It lapses on its own. + log.warn( + `${host.url}: could not release the recovery lease: ${describe(err)}`, + ); + } + } + return finish(outcome); +} + +// Every path where disk and HEAD disagree, found by comparing each file's git +// blob id with HEAD's tree, so nothing is read out of history. +async function scan( + host: RecoveryHost, + event: RealmRecoveryEvent, +): Promise { + let head = await host.history.head(); + let treeStarted = Date.now(); + let tree = new Map( + (await host.history.tree(head)).map(({ path, blob }) => [path, blob]), + ); + event.treeMs = Date.now() - treeStarted; + let scanStarted = Date.now(); + let differing: LocalPath[] = []; + let seen = new Set(); + let fileCount = 0; + let bytesHashed = 0; + for await (let path of host.listFiles()) { + let bytes = await host.readFile(path); + if (!bytes) { + continue; + } + fileCount++; + bytesHashed += bytes.byteLength; + seen.add(path); + let blob = tree.get(path); + if (blob === undefined || blob !== (await gitBlobId(bytes))) { + differing.push(path); + } + } + for (let path of tree.keys()) { + if (!seen.has(path)) { + differing.push(path as LocalPath); + } + } + event.fileCount = fileCount; + event.bytesHashed = bytesHashed; + event.scanMs = Date.now() - scanStarted; + return differing; +} + +// Record what disk holds at `differing` as system revisions of `kind`, each +// path looked at again under its write lock first. A path that agrees with +// HEAD by then — a peer's write just landed there, and the first look saw a +// stale listing — is left out. +async function recordDifferences( + host: RecoveryHost, + differing: LocalPath[], + kind: RevisionKind, +): Promise<{ revisions: string[]; paths: number }> { + let recoveryId = newOperationId(); + let revisions: string[] = []; + let recordedPaths = 0; + // Bounded by path count; a chunk whose bytes are over what one revision + // takes is refused by `record`, which fails recovery. + let chunks: LocalPath[][] = []; + for (let i = 0; i < differing.length; i += MAX_PATHS_PER_REVISION) { + chunks.push(differing.slice(i, i + MAX_PATHS_PER_REVISION)); + } + for (let chunk of chunks) { + let paths = chunk; + await host.withFileWriteLocks(paths, async (releaseLocks) => { + let head = await host.history.head(); + let atHead = await host.history.read(head, paths); + let changes: { + change: HistoryChange; + verb: 'Create' | 'Update' | 'Delete'; + }[] = []; + let contents = new Map(); + for (let path of paths) { + let onDisk = await host.readFile(path); + let inHead = atHead.get(path); + if (sameBytes(onDisk, inHead)) { + continue; + } + contents.set(path, onDisk); + changes.push( + onDisk + ? { + change: { path, content: onDisk }, + verb: inHead ? 'Update' : 'Create', + } + : { change: { path, delete: true }, verb: 'Delete' }, + ); + } + if (changes.length === 0) { + releaseLocks(); + return; + } + let recording = host.systemRecording(); + let revision = await host.history.record(recording, changes, { + kind, + message: recoveryMessage(kind, changes), + ...(chunks.length > 1 + ? { trailers: [{ key: 'Boxel-Recovery-Id', value: recoveryId }] } + : {}), + }); + revisions.push(revision); + recordedPaths += changes.length; + // Disk already holds these bytes. What is left is what a write does + // after them: announcing them, their metadata rows, and indexing. + await host.applyRevisions(contents, [revision], releaseLocks); + }); + } + return { revisions, paths: recordedPaths }; +} + +function recoveryMessage( + kind: RevisionKind, + changes: { change: HistoryChange; verb: 'Create' | 'Update' | 'Delete' }[], +): string { + return [ + kind === 'gap' ? GAP_MESSAGE : DRIFT_MESSAGE, + '', + ...changes.map(({ change, verb }) => `${verb} ${change.path}`), + ].join('\n'); +} + +function sameBytes( + a: Uint8Array | undefined, + b: Uint8Array | undefined, +): boolean { + if (!a || !b) { + return a === b; + } + if (a.byteLength !== b.byteLength) { + return false; + } + for (let i = 0; i < a.byteLength; i++) { + if (a[i] !== b[i]) { + return false; + } + } + return true; +} + +// Whether a path the disk scan finds is one a revision could hold: not the +// history itself, not a git checkout's own store, and not a file the realm is +// part-way through writing. +export function isRecordablePath(path: LocalPath): boolean { + return ( + !isHistoryPath(path) && + path !== '.git' && + !path.startsWith('.git/') && + !isPartialWritePath(path) + ); +} + +function describe(err: unknown): string { + return err instanceof Error ? err.message : String(err); +} diff --git a/packages/runtime-common/realm-history.ts b/packages/runtime-common/realm-history.ts index 95529d7dca9..19b7b6977d7 100644 --- a/packages/runtime-common/realm-history.ts +++ b/packages/runtime-common/realm-history.ts @@ -1,7 +1,12 @@ import { v4 as uuidv4 } from '@lukeed/uuid'; import { Sha256 } from '@aws-crypto/sha256-js'; import { logger } from './log.ts'; -import { param, query } from './expression.ts'; +import { + param, + query, + separatedByCommas, + type Expression, +} from './expression.ts'; import type { DBAdapter } from './db.ts'; import { OperationFailure } from './card-operations/types.ts'; @@ -122,6 +127,10 @@ export interface WriteRecording { // Whether a commit took this write through recording at all. False for a // realm that does not record history, which leaves every other field unset. engaged?: boolean; + // Revisions already recorded whose bytes this commit brings to disk, as + // recovery completes them: the commit records nothing of its own, and + // stamps these materialized and post-committed instead. + recovered?: string[]; // Per-step wall clock, for the write's telemetry line. startedAt?: number; lockWaitMs?: number; @@ -135,7 +144,7 @@ export interface WriteRecording { } export interface HistoryCallTiming { - endpoint: 'commit' | 'outcome' | 'head'; + endpoint: 'commit' | 'outcome' | 'head' | 'tree' | 'read'; ms: number; // `ok`, or the backend's error code, or `transport` for a call whose // response never arrived. @@ -225,7 +234,11 @@ export interface HistoryCommitRequest { export interface HistoryCommitResult { revision: string; + // `revision`'s own parent, absent only for the root revision. On a replay + // it can be older than the parent the retry asked against. + parent?: string; // False when the changes left HEAD's tree as it was, and `revision` is HEAD. + // On a replay, whether the operation's first commit wrote `revision`. created: boolean; // True when the operation id had already committed, and `revision` is the // revision it committed then. @@ -239,10 +252,28 @@ export interface PreparedCommit { } export type HistoryOutcome = - | { state: 'committed'; revision: string } + // `created` and `parent` as in `HistoryCommitResult`. + | { state: 'committed'; revision: string; parent?: string; created: boolean } // Reporting `absent` fences the id: it can never commit afterwards. | { state: 'absent' }; +// One file in a revision's tree. `blob` is the git blob id of its bytes (see +// `gitBlobId`), so a file on disk is compared with it without reading the +// revision's copy. +export interface HistoryTreeEntry { + path: string; + blob: string; +} + +export interface HistoryReadResult { + // Each path asked for, in order: its bytes at the revision, or no content + // where the revision has no file there. + files: { path: string; content?: Uint8Array }[]; + // The paths asked for and not read, because the answer reached the + // backend's ceiling. They are asked for again. + remaining: string[]; +} + export interface HistoryBackend { // Encode a commit for sending before its parent is known. The recorder // prepares a commit before it takes the history commit lock, so the writer @@ -250,6 +281,8 @@ export interface HistoryBackend { prepareCommit(request: Omit): PreparedCommit; outcome(operationId: string): Promise; head(): Promise; + tree(revision: string): Promise; + read(revision: string, paths: string[]): Promise; } // A refusal the backend made with a typed answer, so its outcome is known. @@ -290,8 +323,20 @@ export function isHistoryBackendError( // can reach disk in the opposite order to the one they recorded in. // --------------------------------------------------------------------------- +// `pending` from before the write asks history to record it until its +// outcome is known; `committed` with its revision; `aborted` once history has +// reported the operation absent, which fences it. +export type JournalState = 'pending' | 'committed' | 'aborted'; + +// `write` for a revision a write made, `drift` for changes recovery found on +// disk that no revision explained, `gap` for changes made while the realm did +// not record. +export type RevisionKind = 'write' | 'drift' | 'gap'; + export interface JournalEntry { - revision: string; + state: JournalState; + // Absent until the row is committed. + revision?: string; materialized: boolean; // See `WriteRecording.fingerprint`. fingerprint?: string; @@ -348,8 +393,8 @@ export class RealmHistoryRecorder { // The journal row for an operation id, when this realm recorded it. async journalEntry(operationId: string): Promise { let [row] = await query(this.#dbAdapter, [ - `SELECT revision, materialized_at IS NOT NULL AS materialized, results, - fingerprint + `SELECT state, revision, materialized_at IS NOT NULL AS materialized, + results, fingerprint FROM realm_history_revisions WHERE realm_url =`, param(this.#realmURL), `AND operation_id =`, @@ -359,7 +404,8 @@ export class RealmHistoryRecorder { return undefined; } return { - revision: String(row.revision), + state: String(row.state) as JournalState, + ...(row.revision != null ? { revision: String(row.revision) } : {}), materialized: Boolean(row.materialized), ...(row.fingerprint != null ? { fingerprint: String(row.fingerprint) } @@ -371,17 +417,53 @@ export class RealmHistoryRecorder { // The newest revision whose bytes, and every earlier revision's, have // reached disk. Behind HEAD while a write is between Record and the end of // Materialize, or after one died there; recovery brings disk up to HEAD. + // + // History order is read from each row's parent, not from the order rows + // were journaled: a retry that finds its revision already committed, or + // recovery settling a write whose answer was lost, journals a revision + // after others that history put on top of it. The walk starts at the + // newest revision, the one no other revision has as its parent, and goes + // back only until it has passed every revision not yet on disk. The + // answer is the parent of the oldest of those, or the newest revision when + // there are none. async materializedRevision(): Promise { + let realm = () => param(this.#realmURL); let [row] = await query(this.#dbAdapter, [ - `SELECT revision FROM realm_history_revisions WHERE realm_url =`, - param(this.#realmURL), - `AND seq < COALESCE((SELECT min(seq) FROM realm_history_revisions - WHERE realm_url =`, - param(this.#realmURL), - `AND materialized_at IS NULL), 9223372036854775807) - ORDER BY seq DESC LIMIT 1`, + `WITH RECURSIVE behind AS ( + SELECT count(*) AS n FROM realm_history_revisions + WHERE state = 'committed' AND materialized_at IS NULL + AND realm_url =`, + realm(), + `), tip AS ( + SELECT revision, parent, materialized_at IS NOT NULL AS materialized + FROM realm_history_revisions r + WHERE state = 'committed' AND realm_url =`, + realm(), + `AND NOT EXISTS ( + SELECT 1 FROM realm_history_revisions child + WHERE child.realm_url = r.realm_url AND child.state = 'committed' + AND child.parent = r.revision) + ORDER BY seq DESC LIMIT 1 + ), chain AS ( + SELECT revision, parent, materialized, + CASE WHEN materialized THEN 0 ELSE 1 END AS passed + FROM tip + UNION ALL + SELECT r.revision, r.parent, r.materialized_at IS NOT NULL, + chain.passed + + CASE WHEN r.materialized_at IS NULL THEN 1 ELSE 0 END + FROM chain JOIN realm_history_revisions r + ON r.revision = chain.parent AND r.state = 'committed' + AND r.realm_url =`, + realm(), + `WHERE chain.passed < (SELECT n FROM behind) + ) + SELECT (SELECT revision FROM tip) AS tip, + (SELECT parent FROM chain WHERE NOT materialized + ORDER BY passed DESC LIMIT 1) AS before_behind`, ]); - return row ? String(row.revision) : undefined; + let answer = row?.before_behind ?? row?.tip; + return answer == null ? undefined : String(answer); } async head(recording?: WriteRecording): Promise { @@ -395,9 +477,21 @@ export class RealmHistoryRecorder { // was. Rejects with an `OperationFailure` whose outcome is set on // `recording`: `aborted` when nothing was recorded, `unknown` when that // could not be established. + // + // The write is journaled `pending` before history is asked anything, in a + // statement of its own, so a write whose answer is lost — and whose outcome + // could not be asked for either — leaves a row that recovery resolves. A + // write that definitely recorded nothing takes its row away again, except + // one history reported absent: that id is fenced, and its row says so to a + // retry. async record( recording: WriteRecording, changes: { change: HistoryChange; verb: 'Create' | 'Update' | 'Delete' }[], + opts?: { + kind?: RevisionKind; + message?: string; + trailers?: { key: string; value: string }[]; + }, ): Promise { if (!this.#dbAdapter.withHistoryCommitLock) { throw new Error( @@ -405,16 +499,38 @@ export class RealmHistoryRecorder { ); } let author = this.#author(recording.author); + let kind = opts?.kind ?? 'write'; let request: Omit = { operationId: recording.operationId, author, - message: commitMessage(changes), + message: opts?.message ?? commitMessage(changes), trailers: [ { key: 'Boxel-Operation-Id', value: recording.operationId }, { key: 'Boxel-Channel', value: recording.channel }, + ...(kind === 'gap' + ? [{ key: 'Boxel-Gap', value: 'recording-disabled' }] + : []), + ...(opts?.trailers ?? []), ], changes: changes.map(({ change }) => change), }; + let paths = changes.map(({ change }) => change.path); + await query(this.#dbAdapter, [ + `INSERT INTO realm_history_revisions + (realm_url, operation_id, fingerprint, paths, kind, state, + recorded_at) + VALUES (`, + param(this.#realmURL), + `,`, + param(recording.operationId), + `,`, + param(recording.fingerprint ?? null), + `,`, + ...textArray(paths), + `,`, + param(kind), + `, 'pending', now()) ON CONFLICT (realm_url, operation_id) DO NOTHING`, + ]); let started = Date.now(); try { // Encoded before the lock is taken: every other writer in the realm @@ -428,11 +544,11 @@ export class RealmHistoryRecorder { let result = await this.#commitWithReparent(recording, prepared); // In the same transaction as the lock, so rows are journaled in the // order revisions were committed. A retry of an operation that had - // already committed finds its row already there. + // already committed finds its row already committed. await txQuerier([ `INSERT INTO realm_history_revisions - (realm_url, revision, parent, operation_id, fingerprint, - recorded_at) + (realm_url, revision, parent, operation_id, fingerprint, paths, + kind, state, recorded_at) VALUES (`, param(this.#realmURL), `,`, @@ -443,25 +559,225 @@ export class RealmHistoryRecorder { param(recording.operationId), `,`, param(recording.fingerprint ?? null), - `, now()) ON CONFLICT (realm_url, operation_id) DO NOTHING`, + `,`, + ...textArray(paths), + `,`, + param(kind), + `, 'committed', now()) + ON CONFLICT (realm_url, operation_id) DO UPDATE SET + revision = EXCLUDED.revision, parent = EXCLUDED.parent, + state = 'committed', seq = nextval(pg_get_serial_sequence( + 'realm_history_revisions', 'seq')) + WHERE realm_history_revisions.state = 'pending'`, ]); return result.revision; }, ); + } catch (err: unknown) { + // After the lock is released, so no other pool checkout waits inside it. + if (err instanceof FencedSignal) { + await this.#markAborted(recording.operationId); + throw err.failure; + } + if ( + err instanceof UnchangedTreeSignal || + recording.outcome === 'aborted' + ) { + await this.#forgetPending(recording.operationId); + } + throw err; } finally { recording.recordMs = Date.now() - started; } } - // Stamp the revision materialized: its bytes are on disk. - async markMaterialized(revision: string): Promise { + // Stamp the revisions materialized: their bytes are on disk. + async markMaterialized(revisions: string | string[]): Promise { + let list = typeof revisions === 'string' ? [revisions] : revisions; + if (list.length === 0) { + return; + } await query(this.#dbAdapter, [ `UPDATE realm_history_revisions SET materialized_at = now() - WHERE realm_url =`, + WHERE materialized_at IS NULL AND realm_url =`, + param(this.#realmURL), + `AND revision = ANY(`, + ...textArray(list), + `)`, + ]); + } + + // Stamp the revisions post-committed: the index pass over their paths is + // queued, and their change has been announced. + async markPostCommitted(revisions: string[]): Promise { + if (revisions.length === 0) { + return; + } + await query(this.#dbAdapter, [ + `UPDATE realm_history_revisions SET post_committed_at = now() + WHERE post_committed_at IS NULL AND realm_url =`, + param(this.#realmURL), + `AND revision = ANY(`, + ...textArray(revisions), + `)`, + ]); + } + + // A committed revision other than `operationId`'s that touches one of + // `paths` and has not reached disk. Asked by a write holding those paths' + // locks: a writer keeps its locks until its bytes are on disk, so such a + // revision's writer is gone, and disk holds older bytes than HEAD there. + async unmaterializedRevisionTouching( + paths: string[], + operationId: string, + ): Promise { + if (paths.length === 0) { + return undefined; + } + let [row] = await query(this.#dbAdapter, [ + `SELECT revision FROM realm_history_revisions WHERE realm_url =`, + param(this.#realmURL), + `AND state = 'committed' AND materialized_at IS NULL + AND operation_id <>`, + param(operationId), + `AND paths &&`, + ...textArray(paths), + `ORDER BY seq LIMIT 1`, + ]); + return row ? String(row.revision) : undefined; + } + + // Writes other than `operationId` journaled `pending` that touch one of + // `paths`. Asked by a write holding those paths' locks: a writer keeps its + // locks until it has its answer or has given up on it, so each one's + // answer was lost. + async pendingOperationsTouching( + paths: string[], + operationId: string, + ): Promise { + if (paths.length === 0) { + return []; + } + let rows = await query(this.#dbAdapter, [ + `SELECT operation_id FROM realm_history_revisions WHERE realm_url =`, + param(this.#realmURL), + `AND state = 'pending' AND operation_id <>`, + param(operationId), + `AND paths &&`, + ...textArray(paths), + `ORDER BY seq`, + ]); + return rows.map((row) => String(row.operation_id)); + } + + // Writes journaled `pending` since before `olderThan`: each one's answer was + // lost, along with any chance of asking for it then. + async pendingOperations(olderThan: Date): Promise { + let rows = await query(this.#dbAdapter, [ + `SELECT operation_id FROM realm_history_revisions WHERE realm_url =`, + param(this.#realmURL), + `AND state = 'pending' AND recorded_at <`, + param(olderThan.toISOString()), + `ORDER BY seq`, + ]); + return rows.map((row) => String(row.operation_id)); + } + + // Settle a pending write by asking history for its outcome, which fences + // the id when it never committed. Resolves with the revision it committed, + // or with nothing when it did not. + async resolvePending(operationId: string): Promise { + let outcome = await this.#timed(undefined, 'outcome', () => + this.#backend.outcome(operationId), + ); + if (outcome.state === 'absent') { + await this.#markAborted(operationId); + return undefined; + } + if (!outcome.created) { + // It found the tree as staged and wrote nothing, so there is no + // revision to journal. A retry is answered by history as unchanged. + await this.#forgetPending(operationId); + return undefined; + } + // Its commit stood, on its own parent, which is where it sits in history + // however many revisions were journaled after it. + await query(this.#dbAdapter, [ + `UPDATE realm_history_revisions SET state = 'committed', revision =`, + param(outcome.revision), + `, parent =`, + param(outcome.parent ?? null), + `, seq = nextval(pg_get_serial_sequence('realm_history_revisions', + 'seq')) + WHERE state = 'pending' AND realm_url =`, + param(this.#realmURL), + `AND operation_id =`, + param(operationId), + ]); + return outcome.revision; + } + + // Committed revisions whose bytes have not all reached disk, or whose index + // pass was never queued, in the order they were recorded. + async incompleteRevisions(): Promise< + { revision: string; paths: string[]; materialized: boolean }[] + > { + let rows = await query(this.#dbAdapter, [ + `SELECT revision, paths, materialized_at IS NOT NULL AS materialized + FROM realm_history_revisions WHERE realm_url =`, + param(this.#realmURL), + `AND state = 'committed' + AND (materialized_at IS NULL OR post_committed_at IS NULL) + ORDER BY seq`, + ]); + return rows.map((row) => ({ + revision: String(row.revision), + paths: parseTextArray(row.paths), + materialized: Boolean(row.materialized), + })); + } + + // The kind of a revision this realm journaled: whether it is a write, drift + // recovery found, or a gap left while recording was off. + async revisionKind(revision: string): Promise { + let [row] = await query(this.#dbAdapter, [ + `SELECT kind FROM realm_history_revisions WHERE realm_url =`, param(this.#realmURL), `AND revision =`, param(revision), ]); + return row ? (String(row.kind) as RevisionKind) : undefined; + } + + async tree(revision: string): Promise { + return await this.#timed(undefined, 'tree', () => + this.#backend.tree(revision), + ); + } + + // Every path's bytes at `revision`, over as many calls as the backend's + // ceiling takes. + async read( + revision: string, + paths: string[], + ): Promise> { + let found = new Map(); + let remaining = paths; + while (remaining.length > 0) { + let result = await this.#timed(undefined, 'read', () => + this.#backend.read(revision, remaining), + ); + if (result.files.length === 0) { + throw new Error( + `history read no file of ${remaining.length} asked for at ${revision}`, + ); + } + for (let { path, content } of result.files) { + found.set(path, content); + } + remaining = result.remaining; + } + return found; } async recordResults(operationId: string, results: unknown): Promise { @@ -475,6 +791,33 @@ export class RealmHistoryRecorder { ]); } + async #markAborted(operationId: string): Promise { + await query(this.#dbAdapter, [ + `UPDATE realm_history_revisions SET state = 'aborted' + WHERE state = 'pending' AND realm_url =`, + param(this.#realmURL), + `AND operation_id =`, + param(operationId), + ]); + } + + async #forgetPending(operationId: string): Promise { + try { + await query(this.#dbAdapter, [ + `DELETE FROM realm_history_revisions + WHERE state = 'pending' AND realm_url =`, + param(this.#realmURL), + `AND operation_id =`, + param(operationId), + ]); + } catch (err: unknown) { + // Left pending, recovery asks for its outcome and settles it. + this.#log.warn( + `could not remove the pending row of ${operationId} in ${this.#realmURL}: ${describe(err)}`, + ); + } + } + async #commitWithReparent( recording: WriteRecording, prepared: PreparedCommit, @@ -503,14 +846,19 @@ export class RealmHistoryRecorder { return await this.#resolveFailedCommit(recording, err, parent); } if (result.replayed) { - // The revision this id committed earlier. HEAD may have moved on - // since, so it is asked for again next time. + // The revision this id committed earlier, journaled with its own + // parent: HEAD may have moved on since, so it is asked for again + // next time. this.#head = undefined; - return { revision: result.revision, parent }; + if (!result.created) { + // Its first attempt found the tree as staged and wrote nothing. + throw new UnchangedTreeSignal(result.revision); + } + return { revision: result.revision, parent: result.parent ?? parent }; } this.#head = result.revision; if (result.created) { - return { revision: result.revision, parent }; + return { revision: result.revision, parent: result.parent ?? parent }; } // The changes left HEAD's tree as it was, which the realm's own // unchanged check rules out unless disk had drifted from HEAD. Nothing @@ -529,6 +877,10 @@ export class RealmHistoryRecorder { parent: string, ): Promise<{ revision: string; parent: string }> { if (isHistoryBackendError(err) && err.code !== 'internal') { + if (err.code === 'operation-fenced') { + recording.outcome = 'aborted'; + throw new FencedSignal(operationFenced(recording.operationId)); + } if (err.code === 'operation-mismatch') { recording.outcome = 'aborted'; throw new OperationFailure({ @@ -584,10 +936,15 @@ export class RealmHistoryRecorder { } this.#head = undefined; if (outcome.state === 'committed') { - return { revision: outcome.revision, parent }; + if (!outcome.created) { + throw new UnchangedTreeSignal(outcome.revision); + } + return { revision: outcome.revision, parent: outcome.parent ?? parent }; } + // Asking fenced the id, so it can never commit late, and a retry under it + // is told to send the write under a new one. recording.outcome = 'aborted'; - throw this.#aborted(recording, err); + throw new FencedSignal(this.#aborted(recording, err)); } #aborted(recording: WriteRecording, cause: unknown): OperationFailure { @@ -632,6 +989,29 @@ export class RealmHistoryRecorder { } } +// Thrown inside the history lock by a write whose operation id is fenced, so +// its journal row is marked aborted once the lock is released. +class FencedSignal extends Error { + readonly failure: OperationFailure; + constructor(failure: OperationFailure) { + super(failure.message); + this.name = 'FencedSignal'; + this.failure = failure; + } +} + +// The answer to a write under an operation id history has fenced: it reported +// the id absent, so the write it named never happened and never can. +export function operationFenced(operationId: string): OperationFailure { + return new OperationFailure({ + status: 409, + code: 'operation-fenced', + title: 'Operation id already aborted', + detail: `operation ${operationId} was not recorded, and can no longer be; send this write under a new ${OPERATION_ID_HEADER}`, + meta: { operationId }, + }); +} + // Thrown by a commit that recorded nothing because the tree was already as // staged: the realm answers with `head`, as it does for any unchanged write. export class UnchangedTreeSignal extends Error { @@ -669,6 +1049,46 @@ function describe(err: unknown): string { return err instanceof Error ? err.message : String(err); } +// `values` as a postgres `text[]`. +function textArray(values: string[]): Expression { + if (values.length === 0) { + return [`'{}'::text[]`]; + } + return [ + 'CAST(ARRAY[', + ...separatedByCommas(values.map((value) => [param(value)])), + '] AS text[])', + ] as Expression; +} + +// A postgres `text[]`, which a SQLite-backed or older driver may hand back as +// its literal. +function parseTextArray(value: unknown): string[] { + if (Array.isArray(value)) { + return value.map(String); + } + if (typeof value === 'string') { + let inner = value.replace(/^\{|\}$/g, ''); + return inner === '' + ? [] + : inner.split(',').map((item) => item.replace(/^"|"$/g, '')); + } + return []; +} + +// The git blob id of `bytes`: what history's tree names a file by, so a file +// on disk is compared with a revision without reading the revision's copy. +export async function gitBlobId(bytes: Uint8Array): Promise { + let header = new TextEncoder().encode(`blob ${bytes.byteLength}\0`); + let object = new Uint8Array(header.byteLength + bytes.byteLength); + object.set(header, 0); + object.set(bytes, header.byteLength); + let digest = await globalThis.crypto.subtle.digest('SHA-1', object); + return Array.from(new Uint8Array(digest), (byte) => + byte.toString(16).padStart(2, '0'), + ).join(''); +} + function parseJSON(value: unknown): unknown { return typeof value === 'string' ? JSON.parse(value) : value; } diff --git a/packages/runtime-common/realm.ts b/packages/runtime-common/realm.ts index b434ada0071..1f28bcc8459 100644 --- a/packages/runtime-common/realm.ts +++ b/packages/runtime-common/realm.ts @@ -447,6 +447,7 @@ import type { } from './router.ts'; import { ArchivedRealmError, + ReadOnlyRealmError, AuthenticationError, AuthenticationErrorMessages, AuthorizationError, @@ -563,7 +564,20 @@ import { type HistoryBackend, type HistoryChange, type WriteRecording, + newOperationId, + operationFenced, } from './realm-history.ts'; +import { + clearRealmReadOnly, + realmReadOnly, + setRealmRecording, +} from './db-queries/realm-history-state-queries.ts'; +import { + isRecordablePath, + recoverRealmHistory, + type RecoveryResult, + type RecoveryTrigger, +} from './realm-history-recovery.ts'; import type { JsonValue } from './json-validation.ts'; import { computeContentHash, @@ -1106,6 +1120,9 @@ const OPERATIONAL_ENDPOINT = { operationalEndpoint: true } as const; // probe's routes read it, and so does the probe that reaches no route because // it sends no `Accept` they answer (see `#isBareReadinessProbe`). const READINESS_CHECK_PATH = '/_readiness-check'; +// Where the realm's recovery is asked for, by the scheduled job that checks +// every realm that records history against it. +const HISTORY_RECOVERY_PATH = '_history-recovery'; // Marks the routes whose handlers hand the realm ACL's recorded outcome to the // policy gate: the `/_operations` envelope, which resolves every entry through // the gate, the card+json read, whose operation runs through it, the search, @@ -2105,6 +2122,14 @@ function historyCommitTooLarge( interface HistoryFaults { afterStage?: () => Promise; afterRecord?: () => Promise; + // After the first of the commit's files is on disk and before the next. + midMaterialize?: () => Promise; + // After every byte is on disk and before the change is announced, its + // metadata written or the revision stamped materialized. + afterMaterialize?: () => Promise; + // After the revision's index pass is queued and it is stamped + // post-committed, and before the write answers. + afterPostCommit?: () => Promise; } async function collectBytes( @@ -2415,6 +2440,13 @@ export interface WriteOptions { // the write. A commit without one — every writer but the operations // coordinator — is not recorded. recording?: WriteRecording; + // Treat a file already holding the staged bytes as written, and a removal + // of a file already gone as made: each is announced, its metadata row + // written, and its index pass queued as for any change. Only recovery sets + // it, completing a revision whose bytes may have partly reached disk before + // its writer stopped, and taking in changes found on disk that were never + // indexed. + rewriteUnchanged?: true; } // Who a realm event is delivered to. By default it goes to every session room @@ -2846,6 +2878,15 @@ export class Realm { #testOnlyCoarseAdmission: CoarseAdmission | undefined; #testOnlyBeforeBatchLock: (() => Promise) | undefined; #testOnlyHistoryFaults: HistoryFaults | undefined; + // The recovery in progress, which concurrent callers share. + #recovering: Promise | undefined; + // Who this realm claims the recovery lease as. Per realm rather than per + // process, so two realms over one directory exclude each other as two + // replicas do. + #recoveryLeaseOwner = newOperationId(); + // Set in a process that records history for some realms but not this one, + // which notes at mount that this realm's writes go unrecorded. + #historyConfigured = false; // Present when the realm records history: every write the operations // coordinator commits becomes a revision before it reaches disk. #history: RealmHistoryRecorder | undefined; @@ -3153,6 +3194,7 @@ export class Realm { anonymousRateLimiter, callApi, history, + historyConfigured, }: { url: string; adapter: RealmAdapter; @@ -3200,6 +3242,10 @@ export class Realm { // the operations coordinator commits as a revision, before any of it // reaches disk. Without one, the realm keeps no history. history?: HistoryBackend; + // Whether this process records history for other realms. A realm it + // mounts without `history` then notes that its writes go unrecorded, so + // the changes made meanwhile are taken in as a gap if it records later. + historyConfigured?: boolean; }, opts?: Options, ) { @@ -3220,6 +3266,7 @@ export class Realm { this.#matrixClient.username, this.#matrixClient.matrixURL.href, ); + this.#historyConfigured = Boolean(historyConfigured); if (history) { this.#history = new RealmHistoryRecorder({ backend: history, @@ -3498,6 +3545,12 @@ export class Realm { this.invalidateURLs.bind(this), ACL_ONLY, ) + .post( + `/${HISTORY_RECOVERY_PATH}`, + SupportedMimeType.JSON, + this.handleHistoryRecovery.bind(this), + ACL_ONLY, + ) .post( '(/|/.+/)', SupportedMimeType.CardJson, @@ -3702,7 +3755,12 @@ export class Realm { // different remedy, and the poll loops that consume this discard the // body, so the header is the only place an operator can read it from. let notReady = ( - stage: 'startup' | 'index' | 'index-failed' | 'prerender-html', + stage: + | 'startup' + | 'recovering' + | 'index' + | 'index-failed' + | 'prerender-html', detail?: string, ) => createResponse({ @@ -3746,6 +3804,14 @@ export class Realm { // happened. let waitStartedAt = Date.now(); let requestDeadline = waitStartedAt + READINESS_REQUEST_BUDGET_MS; + // Bringing the realm back into agreement with its history, at mount or + // since: what it serves may be about to change. + if ( + this.#recovering && + !(await settledBy(this.#recovering, requestDeadline)) + ) { + return notReady('recovering'); + } if (!(await settledBy(this.#startedUp.promise, requestDeadline))) { this.#log.warn( `readiness check for ${this.url} is still waiting on realm startup after ${Date.now() - waitStartedAt}ms`, @@ -3886,10 +3952,19 @@ export class Realm { } } + // A read-only realm is ready: it serves every read. Writes are what it + // refuses, which a caller polling for readiness learns here rather than + // from its first write. + let readOnly = this.#history + ? await realmReadOnly(this.#dbAdapter, this.url) + : undefined; return createResponse({ body: null, init: { - headers: { 'content-type': 'text/html' }, + headers: { + 'content-type': 'text/html', + ...(readOnly ? { 'X-Boxel-Realm-Read-Only': readOnly.reason } : {}), + }, status: 200, }, requestContext, @@ -4067,6 +4142,30 @@ export class Realm { }); } + // Check the realm against its history and recover it, for the schedule + // that visits every realm that records. Answers with what recovery found; a + // realm that does not record, or that is read-only, is left alone. + private async handleHistoryRecovery( + _request: Request, + requestContext: RequestContext, + ) { + let result = await this.recoverHistory({ trigger: 'schedule' }); + return createResponse({ + body: JSON.stringify( + result + ? { outcome: result.outcome, recovery: result.event } + : // Left alone: a realm that records gets here only when it is + // already read-only. + { outcome: this.#history ? 'read-only' : 'not-recording' }, + ), + init: { + status: 200, + headers: { 'content-type': SupportedMimeType.JSON }, + }, + requestContext, + }); + } + private async queueReindex( _request: Request, requestContext: RequestContext, @@ -5274,6 +5373,55 @@ export class Realm { `a commit recorded in ${this.url}'s history stages its bytes before writing them, so it cannot serialize them as it writes`, ); } + // Staging reads what disk holds, which is behind HEAD at a path whose last + // revision never all reached disk: an append would be recorded on top of + // the older bytes and drop the revision's. A writer holds its files' locks + // until its bytes are on disk, and this commit holds these paths' locks + // now, so such a revision's writer is gone. Recovery brings disk up to + // HEAD; until then this write waits for it. + let paths = [ + ...(batch.writes?.keys() ?? []), + ...(batch.appends?.keys() ?? []), + ...(batch.deletes ?? []), + ]; + // A write whose answer was lost stays pending until something asks + // history for its outcome, and it may have committed: staging over the + // older bytes on disk would record its paths back as they were before it. + // Asking now settles it. One that never committed is fenced, so it cannot + // commit later either; one that did is a revision disk lacks, which the + // check below holds this write back for. + for (let pending of await history.pendingOperationsTouching( + paths, + recording.operationId, + )) { + try { + await history.resolvePending(pending); + } catch (err: unknown) { + recording.outcome = 'aborted'; + throw new OperationFailure({ + status: 503, + code: 'revision-aborted', + title: 'Write not recorded', + detail: `could not establish whether earlier operation ${pending} to the same files of ${this.url} was recorded (${err instanceof Error ? err.message : String(err)}), so operation ${recording.operationId} was not recorded, and nothing was written; retry this write`, + meta: { operationId: recording.operationId, pending }, + }); + } + } + let behind = await history.unmaterializedRevisionTouching( + paths, + recording.operationId, + ); + if (behind) { + recording.outcome = 'aborted'; + void this.recoverHistory({ trigger: 'write' }); + throw new OperationFailure({ + status: 503, + code: 'realm-recovering', + title: 'Realm recovering', + detail: `revision ${behind} of ${this.url} has not finished reaching disk, and the realm is completing it; retry this write`, + meta: { operationId: recording.operationId, revision: behind }, + }); + } let staged = await this.#stageForHistory(batch, history); await this.#testOnlyHistoryFaults?.afterStage?.(); if (staged.length === 0) { @@ -5474,9 +5622,19 @@ export class Realm { // realm event, file-meta row and index job this commit produces follows the // revision it describes. let recording = this.#history ? options?.recording : undefined; - let recordedRevision = recording - ? await this.#recordBatch(batch, options!, recording) - : undefined; + let recordedRevision = + recording && !recording.recovered + ? await this.#recordBatch(batch, options!, recording) + : undefined; + // The revisions whose bytes this commit brings to disk: the one it just + // recorded, or those recovery is completing. + let committedRevisions = recording?.recovered + ? recording.recovered + : recordedRevision + ? [recordedRevision] + : []; + let faults = recording ? this.#testOnlyHistoryFaults : undefined; + let wroteFirstFile = false; let urls: URL[] = []; // Collect write results for all files we wrote let results: { @@ -5741,7 +5899,7 @@ export class Realm { // caller sent to replace it would be dropped. unchanged = stored; } - if (unchanged) { + if (unchanged && !options?.rewriteUnchanged) { // Identical bytes: the file is left alone, so its modification time // stands and nothing is queued for indexing. The content hash is // still the file's own — the bytes in hand are the bytes on disk — @@ -5787,9 +5945,13 @@ export class Realm { ) { modulesWithoutSchemaChange.add(url.href); } + if (faults?.midMaterialize && wroteFirstFile) { + await faults.midMaterialize(); + } announceIndexing(url); await this.trackOwnWrite(path); let { lastModified } = await this.#adapter.write(path, content); + wroteFirstFile = true; (isNewFile ? addedFiles : updatedFiles).push(path); this.invalidateCache(path); await this.#notifyFileChange(path); @@ -5966,13 +6128,16 @@ export class Realm { let url = this.paths.fileURL(path); announceIndexing(url); await this.trackOwnWrite(path, { isDelete: true }); - await this.#adapter.remove(path); + if (!options?.rewriteUnchanged || (await this.#adapter.exists(path))) { + await this.#adapter.remove(path); + } this.invalidateCache(path); await this.#notifyFileChange(path); removedFiles.push(path); deleteURLs.push(url); } + await faults?.afterMaterialize?.(); if ( addedFiles.length > 0 || updatedFiles.length > 0 || @@ -6012,12 +6177,29 @@ export class Realm { if (recording) { // Every byte the revision describes is on disk, so the realm has caught // up with its history to here. + if (committedRevisions.length > 0) { + await this.#history!.markMaterialized(committedRevisions); + } if (recordedRevision) { - await this.#history!.markMaterialized(recordedRevision); recording.outcome = 'committed'; } recording.durableAt = Date.now(); } + // The last of a revision's post-commit: once its index pass is queued, + // nothing about it is left for recovery to redo. A stamp that fails leaves + // recovery to queue the pass again, which costs a pass and loses nothing. + let stampPostCommitted = async () => { + if (committedRevisions.length === 0) { + return; + } + try { + await this.#history!.markPostCommitted(committedRevisions); + } catch (err: unknown) { + this.#log.warn( + `could not stamp ${committedRevisions.join(', ')} post-committed in ${this.url}; recovery queues their index pass again: ${stringifyErrorForLog(err)}`, + ); + } + }; // And the boundary the write locks end at. Indexing reads the files back // off disk rather than from anything this commit is holding, so the pass // below indexes whatever the last write left — which is what the realm @@ -6242,6 +6424,15 @@ export class Realm { ); } } + if (landingAfterAnswer) { + // The write answers ahead of its pass, so it is stamped when the pass + // lands. A pass that fails leaves it unstamped, for recovery to queue + // again. + void landingAfterAnswer.then(stampPostCommitted, () => undefined); + } else { + await stampPostCommitted(); + } + await faults?.afterPostCommit?.(); // A mixed batch whose instances all turned out to be byte-identical // leaves the flush's deferred set with no later pass to fold it into. // Those URLs are real dependents of a module that did change, so the @@ -8451,7 +8642,20 @@ export class Realm { }, }); } + if (entry.state === 'pending') { + // An attempt under this id lost its answer. The retry goes ahead as + // the write: history answers a commit under an id that already + // committed with that revision, so the retry either completes the + // first attempt or makes the write itself. + return undefined; + } recording.engaged = true; + if (entry.state === 'aborted') { + // History reported the id absent, which fenced it: the write it + // named never happened and never can. + recording.outcome = 'aborted'; + throw operationFenced(recording.operationId); + } recording.revision = entry.revision; if (entry.materialized && entry.results === undefined) { // On disk, with the answer it gave not kept: the attempt is still @@ -8540,6 +8744,14 @@ export class Realm { async #startup(opts?: { fromScratchIndexPriority?: number }) { await Promise.resolve(); let startTime = Date.now(); + // Before the index decision, so whatever the index pass reads off disk is + // what the realm's history holds. Recovery never fails the mount: a realm + // it cannot recover serves its reads read-only. + if (this.#history) { + await this.recoverHistory({ trigger: 'mount' }); + } else if (this.#historyConfigured) { + await setRealmRecording(this.#dbAdapter, this.url, false); + } let isNewIndex = await this.#realmIndexUpdater.isNewIndex(); if (this.#skipBootIndex) { // Mount-and-serve only: no from-scratch index, even on a new index. @@ -8775,7 +8987,7 @@ export class Realm { let localPath = this.paths.local(new URL(request.url)); let requiredPermission: RealmAction = 'read'; - if (localPath === '_permissions') { + if (localPath === '_permissions' || localPath === HISTORY_RECOVERY_PATH) { requiredPermission = 'realm-owner'; } else if (localPath === CAPABILITIES_PATH) { // A `POST` because it carries a body, and the body is a list of @@ -8913,6 +9125,24 @@ export class Realm { requestContext.archivedSeal = seal; } } + // A realm whose recovery failed keeps serving reads and refuses every + // write until an operator clears it. Read fresh, like the archived seal, + // so a peer's seal or clear takes effect here at once. Recovery's own + // route passes: it is how a schedule finds out the realm is sealed. + if ( + this.#history && + requiredPermission !== 'read' && + !dispatch.operationalEndpoint && + localPath !== HISTORY_RECOVERY_PATH + ) { + let readOnly = await realmReadOnly(this.#dbAdapter, this.url); + if (readOnly) { + throw new ReadOnlyRealmError( + `Realm ${this.url} is read-only: ${readOnly.detail ?? readOnly.reason}`, + readOnly.reason, + ); + } + } if (!isLocal && request.method === 'HEAD' && dispatch.coarseReadOnly) { let answer = await this.#headForNonReader(request, requestContext); if (answer) { @@ -8978,6 +9208,32 @@ export class Realm { }); } + if (e instanceof ReadOnlyRealmError) { + // 423, not the archived 403: the realm is there and readable, and + // refuses only writes, until an operator clears it. + return createResponse({ + body: JSON.stringify({ + errors: [ + { + status: '423', + code: 'realm-read-only', + title: 'Realm Read-Only', + detail: e.message, + }, + ], + }), + init: { + status: 423, + headers: { + 'content-type': SupportedMimeType.JSONAPI, + 'X-Boxel-Realm-Read-Only': e.reason, + 'X-Boxel-Realm-Url': requestContext.realm.url, + }, + }, + requestContext, + }); + } + if (e instanceof AuthorizationError) { return new Response(`${e.message}`, { status: 403, @@ -9541,6 +9797,121 @@ export class Realm { return this.#history; } + // Check the realm against its history and bring it back into agreement: + // complete revisions whose bytes or index pass a crash cut short, settle + // writes whose answer was lost, and record what disk holds that no revision + // explains (see `realm-history-recovery.ts`). A failure leaves the realm + // read-only. Resolves with nothing for a realm that does not record, or — + // unless an operator asked — one already read-only, which stays that way + // until an operator clears it. Concurrent calls share one run. + async recoverHistory({ + trigger, + }: { + trigger: RecoveryTrigger; + }): Promise { + if (!this.#history) { + return undefined; + } + if (this.#recovering) { + return await this.#recovering; + } + let run = (async () => { + if ( + trigger !== 'operator' && + (await realmReadOnly(this.#dbAdapter, this.url)) + ) { + return undefined; + } + let result = await recoverRealmHistory(this.#recoveryHost(), trigger, { + leaseOwner: this.#recoveryLeaseOwner, + }); + if (trigger === 'operator' && result.outcome !== 'read-only') { + await clearRealmReadOnly(this.#dbAdapter, this.url); + } + return result; + })(); + this.#recovering = run; + try { + return await run; + } finally { + this.#recovering = undefined; + } + } + + #recoveryHost() { + let history = this.#history!; + let adapter = this.#adapter; + async function* walk(dir: LocalPath): AsyncGenerator { + for await (let entry of adapter.readdir(dir)) { + let path = entry.path as LocalPath; + if (!isRecordablePath(path)) { + continue; + } + if (entry.kind === 'directory') { + yield* walk(path); + } else { + yield path; + } + } + } + return { + url: this.url, + history, + dbAdapter: this.#dbAdapter, + listFiles: () => walk('' as LocalPath), + readFile: async (path: LocalPath) => { + let stored = await adapter.openFile(path); + if (!stored) { + return undefined; + } + return stored.size !== undefined + ? await readRangeBytes(adapter, path, 0, stored.size) + : await fileContentToBytes(stored); + }, + withFileWriteLocks: ( + paths: LocalPath[], + fn: (releaseLocks: () => void) => Promise, + ) => this.#dbAdapter.withFileWriteLocks(this.url, paths, fn), + applyRevisions: async ( + contents: Map, + revisions: string[], + releaseLocks: () => void, + ) => { + let writes = new Map(); + let deletes: LocalPath[] = []; + for (let [path, content] of contents) { + if (content) { + writes.set(path, content); + } else { + deletes.push(path); + } + } + await this._commitBatchUnlocked( + { writes, deletes }, + { + recording: { + ...systemWriteRecording(), + engaged: true, + recovered: revisions, + }, + rewriteUnchanged: true, + // Queued rather than waited for: recovery runs at mount, which + // does not wait on the realm's indexing either. + waitForIndex: false, + onDurable: releaseLocks, + }, + ); + }, + systemRecording: () => systemWriteRecording(), + reportError: (message: string, err: unknown) => { + this.#log.error( + err ? `${message}: ${stringifyErrorForLog(err)}` : message, + ); + reportError(new Error(message)); + }, + }; + } + // Whether the archived seal covers the request `dispatch` answers: the // realm is archived and the dispatch is not an operational endpoint. That // is read off the dispatch rather than the request's path, so a request diff --git a/packages/runtime-common/router.ts b/packages/runtime-common/router.ts index 68e3cda5446..aec0718e508 100644 --- a/packages/runtime-common/router.ts +++ b/packages/runtime-common/router.ts @@ -27,6 +27,17 @@ export function isCoarseRefusal(error: unknown): error is CoarseRefusal { // (sealed) realm. Surfaced as a 403 carrying an "archived" marker so the // client can render the sealed state rather than a generic forbidden error. export class ArchivedRealmError extends Error {} +// Thrown at the realm request boundary when a write targets a realm that has +// gone read-only because its recovery failed. Surfaced as a 423, distinct from +// the archived 403: a read-only realm still serves reads, and its writes come +// back once an operator clears it. +export class ReadOnlyRealmError extends Error { + readonly reason: string; + constructor(message: string, reason: string) { + super(message); + this.reason = reason; + } +} // A `const` object (rather than a TS `enum`) so the declaration is // erasable and runs under Node's native `--experimental-strip-types`. export const AuthenticationErrorMessages = { diff --git a/packages/runtime-common/tasks/index.ts b/packages/runtime-common/tasks/index.ts index 3515efc2d3f..8b7883beff3 100644 --- a/packages/runtime-common/tasks/index.ts +++ b/packages/runtime-common/tasks/index.ts @@ -18,6 +18,7 @@ export * from './full-reindex.ts'; export * from './daily-credit-grant.ts'; export * from './indexer.ts'; export * from './media-cache-gc.ts'; +export * from './realm-history-recovery.ts'; export * from './scoped-css-gc.ts'; export * from './prerender-html.ts'; export * from './prerender-html-reconcile.ts'; diff --git a/packages/runtime-common/tasks/realm-history-recovery.ts b/packages/runtime-common/tasks/realm-history-recovery.ts new file mode 100644 index 00000000000..8a9196c511d --- /dev/null +++ b/packages/runtime-common/tasks/realm-history-recovery.ts @@ -0,0 +1,104 @@ +import type * as JSONTypes from 'json-typescript'; +import type { Task } from './index.ts'; +import { jobIdentity } from '../index.ts'; +import { + registerQueueJobDefinition, + type QueueCoalesceContext, + type QueueCoalesceDecision, +} from '../queue.ts'; +import { recordingRealmURLs } from '../db-queries/realm-history-state-queries.ts'; +import { fetchRealmOwnerUsername } from '../db-queries/realm-permission-queries.ts'; + +// The cron enqueues this job with no arguments; it visits every realm that +// records history. +type RealmHistoryRecoveryArgs = JSONTypes.Object; + +export interface RealmHistoryRecoveryResult extends JSONTypes.Object { + // Each realm visited, by what its recovery found: `clean`, `recovered`, + // `read-only`, `leased`, `not-recording`, or `failed` for a realm that could + // not be asked. + realms: Record; +} + +// Each realm's recovery scans every file in it, so a sweep over many realms +// is long; it is the only job in its group, so a long lease delays nothing +// but the next sweep, which coalesces into this one anyway. +export const REALM_HISTORY_RECOVERY_JOB_TIMEOUT_SEC = 30 * 60; + +export { realmHistoryRecovery }; + +// The fixed concurrency group already runs one sweep at a time; this also +// collapses a queued tick into a pending or in-flight sweep, so overlapping +// cron ticks never pile up. +function chooseRealmHistoryRecoveryCoalesceDecision( + context: QueueCoalesceContext, +): QueueCoalesceDecision { + let { incoming, candidates, inFlightCandidates } = context; + let twin = + candidates.find((candidate) => candidate.jobType === incoming.jobType) ?? + inFlightCandidates.find( + (candidate) => candidate.jobType === incoming.jobType, + ); + if (!twin) { + return { type: 'insert' }; + } + return { type: 'join', jobId: twin.id }; +} + +registerQueueJobDefinition({ + jobType: 'realm-history-recovery', + coalesce: chooseRealmHistoryRecoveryCoalesceDecision, +}); + +// The scheduled half of realm recovery: for each realm that recorded history +// when last mounted, ask the realm-server serving it to check the realm +// against its history (`POST /_history-recovery`). Recovery runs there +// rather than here, because only that process holds the realm's files, its +// history and its broadcaster. The request is made as the realm's owner, as +// the realm's other system jobs are. A realm that is already consistent comes +// back `clean`, having written nothing. +const realmHistoryRecovery: Task< + RealmHistoryRecoveryArgs, + RealmHistoryRecoveryResult +> = ({ dbAdapter, getAuthedFetch, reportStatus, log }) => + async function (args) { + let { jobInfo } = args; + reportStatus(jobInfo, 'start'); + let realms: Record = {}; + for (let realmURL of await recordingRealmURLs(dbAdapter)) { + try { + let realmUsername = await fetchRealmOwnerUsername(dbAdapter, realmURL); + if (!realmUsername) { + realms[realmURL] = 'failed'; + log.warn( + `${jobIdentity(jobInfo)} realm-history-recovery: ${realmURL} has no owner to act as; skipped`, + ); + continue; + } + let fetch = await getAuthedFetch({ realmURL, realmUsername }); + let response = await fetch(`${realmURL}_history-recovery`, { + method: 'POST', + headers: { Accept: 'application/json' }, + }); + if (!response.ok) { + realms[realmURL] = 'failed'; + log.warn( + `${jobIdentity(jobInfo)} realm-history-recovery: ${realmURL} answered ${response.status}: ${(await response.text()).slice(0, 200)}`, + ); + continue; + } + let body = (await response.json()) as { outcome?: string }; + realms[realmURL] = body.outcome ?? 'failed'; + } catch (err: unknown) { + realms[realmURL] = 'failed'; + log.warn( + `${jobIdentity(jobInfo)} realm-history-recovery: could not ask ${realmURL} to recover: ${err instanceof Error ? err.message : String(err)}`, + ); + } + } + log.info( + `${jobIdentity(jobInfo)} realm-history-recovery visited ${Object.keys(realms).length} realm(s): ${JSON.stringify(realms)}`, + ); + reportStatus(jobInfo, 'finish'); + return { realms }; + }; diff --git a/packages/runtime-common/worker.ts b/packages/runtime-common/worker.ts index b957380d463..67cde7cc1e1 100644 --- a/packages/runtime-common/worker.ts +++ b/packages/runtime-common/worker.ts @@ -326,6 +326,11 @@ export class Worker { this.#queue.register(`media-cache-gc`, Tasks['mediaCacheGc'](taskArgs)), 'scoped-css-gc': () => this.#queue.register(`scoped-css-gc`, Tasks['scopedCssGc'](taskArgs)), + 'realm-history-recovery': () => + this.#queue.register( + `realm-history-recovery`, + Tasks['realmHistoryRecovery'](taskArgs), + ), 'lint-source': () => this.#queue.register(`lint-source`, Tasks['lintSource'](taskArgs)), 'full-reindex': () =>