diff --git a/CHANGELOG.md b/CHANGELOG.md index 441fba6d9..792a976e3 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -30,6 +30,13 @@ and this project adheres to wire encoding. Strict create (409 when the document already has content), initial content attributed to the optional `X-User-Id` header; guarded by standard document write access (admin JWT or user session) +- ✨(collaboration) soft-migrate legacy S3 documents into yhub on first access + (`SOFT_MIGRATION=true`): when yhub does not know a document yet, its legacy + snapshot (`{id}/file`, base64 Yjs update) is fetched from the Django S3 + media bucket and seeded server-side (attributed to `system`) before the + connection is admitted. A missing S3 object means a brand-new document and + yields an empty room; any real S3/compute failure fails closed (opaque 401, + the client retries with backoff). Enabled in the dev stack via compose.yml ### Changed diff --git a/compose.yml b/compose.yml index a4062bc7d..19e6ae65d 100644 --- a/compose.yml +++ b/compose.yml @@ -241,6 +241,9 @@ services: REDIS: redis://yhub-valkey:6379 POSTGRES: postgres://yhub:pass@yhub-postgres:5432/yhub REDIS_PREFIX: yhub + # seed rooms from the legacy Django/S3 document store on first access — + # S3 endpoint/credentials come from env.d/development/common + SOFT_MIGRATION: "true" env_file: - env.d/development/common - env.d/development/common.local @@ -252,6 +255,10 @@ services: condition: service_healthy yhub-postgres: condition: service_healthy + # soft migration reads the legacy document store at startup traffic — + # starting before minio would cache 401s for the first accessed docs + minio: + condition: service_healthy kc_postgresql: image: postgres:14.3 diff --git a/src/yhub-server/README.md b/src/yhub-server/README.md index cc907d9ea..d6e04670f 100644 --- a/src/yhub-server/README.md +++ b/src/yhub-server/README.md @@ -42,6 +42,88 @@ not be routed through the public ingress. The `Dockerfile` builds the container image used by the `yhub` service in `compose.yml`. +## Soft migration (`SOFT_MIGRATION=true`) + +Documents were historically stored by the Django backend in the S3 media +bucket, as UTF-8 text that is the base64 encoding of a raw Yjs update, at key +`{document-uuid}/file`. With `SOFT_MIGRATION=true`, this server migrates those +documents into yhub lazily, on first access: + +1. After a user's document authorization succeeds, the auth plugin checks + whether yhub already has content for the room — a bare postgres `SELECT` + (persisted rows), then the valkey stream (uncompacted `ydoc:update:v1` + messages), then the `SELECT` again to close the compaction race. Verdicts + are cached in-process (existing docs 10 min, empty docs 60 s, failures + 5 min). +2. If the room is unknown, the legacy object is fetched from S3 (10 s + timeout, 10 MiB decoded cap — the same limit as `create-ydoc`), decoded, + diffed through yhub's compute pool and appended to the room's stream — + attributed to the `system` identity with a `migration=s3` custom + attribution. This completes before the websocket upgrade resolves, so the + initial sync always includes the seeded content. First access to an + unmigrated document is therefore slower by one S3 round-trip plus one + compute pass. +3. Concurrent first-connections are collapsed: an in-process in-flight map, a + per-room valkey lock (`{prefix}:softmigrate:*`, 30 s TTL), and a cap of 4 + concurrent seeds per replica (excess connections fail fast and retry). + +Guarantees and failure behavior: + +- **Missing S3 object is not an error** — that is the brand-new-document case + (Django writes no object until the first content save); the room simply + starts empty. +- **Everything else fails closed**: network/auth errors, timeouts, oversized + objects and corrupt updates deny the connection (an opaque 401 the client + retries with backoff) and log a `soft-migration` error. A cached failure + verdict prevents retry storms from hammering S3 — permanent failures + (corrupt/oversized objects) for 5 minutes, transient ones (network errors, + timeouts) for 15 seconds, and per-replica seed backpressure (more than 4 + concurrent seeds) is denied without caching so the client's next retry goes + through. +- **Seeding is idempotent**: the legacy S3 snapshots are frozen (the + frontend no longer PATCHes content snapshots to Django) and share one Yjs + lineage with everything in yhub, so duplicate or concurrent seeds merge as + CRDT no-ops. Losing valkey before compaction merely makes the next access + re-seed from S3. Note that edits made *after* a document was migrated live + only in yhub — a re-seed after total yhub data loss restores the + pre-migration snapshot, nothing newer. +- **`SOFT_MIGRATION=false` does not undo anything** — migrated documents + stay correct in yhub — but since the frontend's client-side seeding was + removed along with the content GET/PATCH endpoints, an unmigrated legacy + document then opens as an *empty* room. Keep the flag on until a backfill + has migrated the full corpus. + +Configuration: `AWS_S3_ENDPOINT_URL`, `AWS_S3_ACCESS_KEY_ID`, +`AWS_S3_SECRET_ACCESS_KEY` (both with `*_FILE` indirection), optional +`AWS_S3_REGION_NAME`, and `AWS_STORAGE_BUCKET_NAME` (defaults to Django's dev +default `impress-media-storage`; production uses a different bucket name and +must set it explicitly). The server refuses to boot when the flag is set +without endpoint and credentials. In development the values arrive via +`env.d/development/common`. + +Operational notes: + +- Use **read-only, bucket-scoped S3 credentials** in production — never the + backend's read-write keys; this process terminates untrusted traffic. On + AWS the credentials must include `s3:ListBucket` on the bucket in addition + to `s3:GetObject`: without it, S3 reports a missing object as + `403 AccessDenied` instead of `404 NoSuchKey`, and every brand-new document + would fail closed instead of starting empty. +- `AWS_S3_ENDPOINT_URL` must not contain a path (the minio client cannot + address a base path); the server refuses to boot otherwise. +- After manually wiping a room's yhub state (postgres row + stream key), + **restart yhub** so the in-process verdict cache cannot serve a stale + "exists" and suppress the re-seed. +- Lazy migration never finishes on its own: documents that are never opened + stay in S3 forever, and they are only reachable through this flag now that + the frontend's client-side seeding is gone. A batch backfill (Django + posting each document's **exact** S3 bytes to `create-ydoc` with the admin + JWT, treating 409 as success) is the intended completion path and composes + safely with concurrent first-accesses — as long as content is never + re-converted: independently generated updates for the same document would + duplicate its content on merge, while re-posting the stored bytes is a + no-op. Only after the backfill may `SOFT_MIGRATION` be turned off. + ## ⚠️ License warning (AGPL) This directory depends on `@y/hub`, which is licensed under the diff --git a/src/yhub-server/package-lock.json b/src/yhub-server/package-lock.json index 9808e2ab1..814f8e8a0 100644 --- a/src/yhub-server/package-lock.json +++ b/src/yhub-server/package-lock.json @@ -7,7 +7,8 @@ "name": "yhub-server", "dependencies": { "@y/hub": "0.4.0", - "jose": "6.2.8" + "jose": "6.2.8", + "minio": "8.0.7" }, "engines": { "node": ">=22" diff --git a/src/yhub-server/package.json b/src/yhub-server/package.json index f4ec5cb51..e0129e490 100644 --- a/src/yhub-server/package.json +++ b/src/yhub-server/package.json @@ -7,7 +7,8 @@ }, "dependencies": { "@y/hub": "0.4.0", - "jose": "6.2.8" + "jose": "6.2.8", + "minio": "8.0.7" }, "engines": { "node": ">=22" diff --git a/src/yhub-server/server.js b/src/yhub-server/server.js index 4f7f303f2..79d2cca02 100644 --- a/src/yhub-server/server.js +++ b/src/yhub-server/server.js @@ -1,8 +1,9 @@ -import { createHash } from 'node:crypto'; +import { createHash, randomUUID } from 'node:crypto'; import { readFileSync } from 'node:fs'; -import { createApiEndpoint, createAuthPlugin, createYHub } from '@y/hub'; +import { createApiEndpoint, createAuthPlugin, createYHub, logger } from '@y/hub'; import { createRemoteJWKSet, jwtVerify } from 'jose'; +import { Client as S3Client } from 'minio'; // mirror y-provider's env.ts secret-file support const secret = (name, dflt) => @@ -21,8 +22,12 @@ const allowedOrigins = ( ).split(','); const Y_PROVIDER_API_KEY = secret('Y_PROVIDER_API_KEY', 'yprovider-api-key'); const ORG = process.env.YHUB_ORG || 'docs'; +// lowercase only (no /i): Django serializes UUIDs lowercase, while yhub rooms +// and S3 keys are case-sensitive strings — accepting case variants would let a +// client open a parallel room for the same document (and, with soft migration, +// miss its S3 object and fork the document's lineage) const UUID4 = - /^[0-9a-f]{8}-[0-9a-f]{4}-4[0-9a-f]{3}-[89ab][0-9a-f]{3}-[0-9a-f]{12}$/i; + /^[0-9a-f]{8}-[0-9a-f]{4}-4[0-9a-f]{3}-[89ab][0-9a-f]{3}-[0-9a-f]{12}$/; // an empty Yjs update (what `Y.encodeStateAsUpdate(new Y.Doc())` encodes to) — // hardcoded so we don't import @y/y for two bytes const EMPTY_YDOC = new Uint8Array([0, 0]); @@ -33,6 +38,61 @@ const EMPTY_YDOC = new Uint8Array([0, 0]); // websocket path. const MAX_CREATE_BYTES = 10 * 1024 * 1024; +// Soft migration (see README.md): legacy documents live in Django's S3 media +// bucket as UTF-8 base64 of a raw Yjs update at key `{docid}/file`. With +// SOFT_MIGRATION=true, the first access to a room yhub does not know yet +// fetches that object, seeds the room with it (attributed to 'system'), and +// only then admits the connection. +const SOFT_MIGRATION = process.env.SOFT_MIGRATION === 'true'; +const AWS_S3_ENDPOINT_URL = process.env.AWS_S3_ENDPOINT_URL; +const AWS_S3_ACCESS_KEY_ID = secret('AWS_S3_ACCESS_KEY_ID'); +const AWS_S3_SECRET_ACCESS_KEY = secret('AWS_S3_SECRET_ACCESS_KEY'); +const AWS_S3_REGION_NAME = process.env.AWS_S3_REGION_NAME; +// Django's default bucket name (impress settings.py) — prod overrides it +const AWS_STORAGE_BUCKET_NAME = + process.env.AWS_STORAGE_BUCKET_NAME || 'impress-media-storage'; +// base64 inflates 3 bytes to 4 — cap the streamed read at the encoded size of +// MAX_CREATE_BYTES (the same effective limit as create-ydoc) plus padding slack +const MAX_LEGACY_B64_BYTES = Math.ceil(MAX_CREATE_BYTES / 3) * 4 + 1024; +const S3_FETCH_TIMEOUT_MS = 10000; +const MIGRATE_LOCK_TTL_MS = 30000; +const MAX_CONCURRENT_SEEDS = 4; + +if ( + SOFT_MIGRATION && + (!AWS_S3_ENDPOINT_URL || !AWS_S3_ACCESS_KEY_ID || !AWS_S3_SECRET_ACCESS_KEY) +) { + // fail at boot instead of as an opaque 401 storm on first connect + throw new Error( + 'SOFT_MIGRATION=true requires AWS_S3_ENDPOINT_URL, AWS_S3_ACCESS_KEY_ID and AWS_S3_SECRET_ACCESS_KEY', + ); +} +const s3 = SOFT_MIGRATION + ? (() => { + const url = new URL(AWS_S3_ENDPOINT_URL); + if (url.pathname !== '/' && url.pathname !== '') { + // boto3 accepts path-prefixed endpoints but the minio client cannot + // address a base path — dropping it silently would probe the wrong + // keys and "migrate" every doc as empty + throw new Error('AWS_S3_ENDPOINT_URL must not contain a path'); + } + return new S3Client({ + endPoint: url.hostname, + port: + url.port !== '' + ? Number(url.port) + : url.protocol === 'https:' + ? 443 + : 80, + useSSL: url.protocol === 'https:', + accessKey: AWS_S3_ACCESS_KEY_ID, + secretKey: AWS_S3_SECRET_ACCESS_KEY, + ...(AWS_S3_REGION_NAME ? { region: AWS_S3_REGION_NAME } : {}), + }); + })() + : null; +const migrationLog = logger.child({ module: 'soft-migration' }); + // Public keys verifying the RS256 admin tokens Django issues (JWTService). // Lazily fetched on first use; jose caches the keys and refetches on unknown // "kid", so Django can rotate the signing key without a yhub restart. @@ -56,6 +116,273 @@ const backendFetch = async (path, { cookie, origin }) => { return res.json(); }; +// Legacy Django document store: object `{docid}/file`, body = UTF-8 text that +// is the base64 encoding of a raw Yjs update. Returns null when the object +// does not exist — a document that never had content saved, e.g. brand new. +// Throws on any other failure (network, auth, timeout, oversize); corrupt +// base64 decodes leniently to garbage that patchYdoc later rejects. +const fetchLegacyDoc = async (docid) => { + let stream = null; + let cancelTimeout = () => {}; + // minio 8 takes no AbortSignal — race a timer that also destroys the body + // stream once reading, so a stalled transfer cannot hold the ws upgrade + const timeout = new Promise((_, reject) => { + const timer = setTimeout(() => { + const err = new Error( + `s3 fetch timed out after ${S3_FETCH_TIMEOUT_MS}ms`, + ); + err.transient = true; // a slow S3 may recover — cache the failure briefly + stream?.destroy(err); + reject(err); + }, S3_FETCH_TIMEOUT_MS); + cancelTimeout = () => clearTimeout(timer); + }); + try { + let objPromise; + try { + objPromise = s3.getObject(AWS_STORAGE_BUCKET_NAME, `${docid}/file`); + stream = await Promise.race([objPromise, timeout]); + } catch (err) { + if (err?.code === 'NoSuchKey') return null; + // if the timeout won the race, getObject may still resolve later — + // destroy the late-arriving response stream, otherwise its never-read + // socket leaks (minio 8 sets no request timeout and cannot abort) + objPromise?.then((s) => s.destroy(err), () => {}); + throw err; + } + const body = await Promise.race([ + new Promise((resolve, reject) => { + const chunks = []; + let received = 0; + stream.on('data', (chunk) => { + received += chunk.byteLength; + if (received > MAX_LEGACY_B64_BYTES) { + stream.destroy( + new Error(`legacy object exceeds the ${MAX_LEGACY_B64_BYTES}B cap`), + ); + return; + } + chunks.push(chunk); + }); + stream.on('error', reject); + stream.on('end', () => resolve(Buffer.concat(chunks))); + }), + timeout, + ]); + const decoded = Buffer.from(body.toString('utf8'), 'base64'); + if (decoded.byteLength > MAX_CREATE_BYTES) { + throw new Error( + `decoded legacy update (${decoded.byteLength}B) exceeds the ${MAX_CREATE_BYTES}B cap`, + ); + } + // compute-task schema requires an exact Uint8Array (lib0 compares the + // constructor) — re-view the Buffer without copying + return new Uint8Array( + decoded.buffer, + decoded.byteOffset, + decoded.byteLength, + ); + } finally { + cancelTimeout(); + } +}; + +// Quick existence check: does yhub already have content for this room? +// Sequenced cheapest-first: a persisted postgres row (bare SELECT, no blob +// columns; rows are never deleted, so a hit is always safe) — then the valkey +// stream (only ydoc:update:v1 counts: awareness and auth-check messages share +// the stream but carry no content) — then the SELECT again, which closes the +// store-before-trim compaction race (and the worst case of a miss is only a +// redundant, idempotent re-seed). +const ydocExists = async (room) => { + if ((await yhub.persistence.retrieveDoc(room, {})).lastClock !== '0') { + return true; + } + const streams = await yhub.stream.getMessages([{ room, clock: '0' }]); + if ((streams[0]?.messages ?? []).some((m) => m.type === 'ydoc:update:v1')) { + return true; + } + return (await yhub.persistence.retrieveDoc(room, {})).lastClock !== '0'; +}; + +// Per-docid migration verdicts, in-memory (per replica). 'exists' is monotone +// in normal operation — its TTL only bounds staleness after an operator +// manually wipes a room's yhub state (restart yhub after a wipe to drop the +// cache immediately). 'empty' (no S3 object) keeps never-edited docs and +// rechecks off S3; 'failed' breaks the retry-refetch storm a permanently +// corrupt object would otherwise sustain (y-websocket retries denied upgrades +// forever). +const VERDICT_TTL_MS = { exists: 600000, empty: 60000, failed: 300000 }; +// transient failures (network blips, timeouts, S3 restarting) are cached just +// long enough to blunt a retry storm without turning a hiccup into a lockout +const TRANSIENT_TTL_MS = 15000; +const TRANSIENT_CODES = new Set([ + 'ECONNREFUSED', + 'ECONNRESET', + 'ETIMEDOUT', + 'EHOSTUNREACH', + 'ENETUNREACH', + 'ENOTFOUND', + 'EAI_AGAIN', + 'EPIPE', +]); +const isTransient = (err) => + err?.transient === true || + TRANSIENT_CODES.has(err?.code) || + TRANSIENT_CODES.has(err?.cause?.code); +const VERDICT_CACHE_MAX = 50000; +const verdicts = new Map(); // docid -> { verdict, error, expires } +const rememberVerdict = (docid, verdict, error = null, ttl = VERDICT_TTL_MS[verdict]) => { + // delete-then-set keeps Map insertion order ≈ recency, so the FIFO eviction + // drops the stalest entry — and re-setting an existing docid never evicts + // an unrelated one + if (!verdicts.delete(docid) && verdicts.size >= VERDICT_CACHE_MAX) { + verdicts.delete(verdicts.keys().next().value); + } + verdicts.set(docid, { + verdict, + error, + expires: Date.now() + ttl, + }); +}; +const inflightMigrations = new Map(); // docid -> Promise +let activeSeeds = 0; + +const migrate = async (room) => { + if (await ydocExists(room)) return 'exists'; + // collapse cross-replica herds: one seeder per room, the rest wait and + // re-probe. The key sits outside yhub's scanned `:room:*` patterns. + const lockKey = `${REDIS_PREFIX}:softmigrate:${room.org}:${room.docid}:${room.branch}`; + const lockToken = randomUUID(); + const redis = yhub.stream.redis; + const acquired = await redis.set(lockKey, lockToken, { + condition: 'NX', + expiration: { type: 'PX', value: MIGRATE_LOCK_TTL_MS }, + }); + try { + if (acquired == null) { + // another connection or replica is seeding — wait for its lock, then + // re-probe. If the doc is still absent (the holder crashed or its S3 + // fetch failed), fall through and seed ourselves: duplicate seeds use + // byte-identical updates from one lineage and merge as CRDT no-ops. + const deadline = Date.now() + MIGRATE_LOCK_TTL_MS + 5000; + while (Date.now() < deadline && (await redis.exists(lockKey)) === 1) { + await new Promise((resolve) => setTimeout(resolve, 300)); + } + if (await ydocExists(room)) return 'exists'; + } + if (activeSeeds >= MAX_CONCURRENT_SEEDS) { + // fail fast under a herd of distinct cold docs — the client's retry + // backoff spreads the load. Probes above stay uncapped. noCache: + // momentary per-replica backpressure must deny once, not be cached as + // a failure — a slot frees up within seconds + const err = new Error('too many concurrent soft migrations'); + err.noCache = true; + throw err; + } + activeSeeds++; + try { + const start = Date.now(); + const update = await fetchLegacyDoc(room.docid); + if (update == null) { + migrationLog.info( + { event: 'seed.empty', docid: room.docid }, + 'no legacy s3 object; room starts empty', + ); + return 'empty'; + } + // legacy content has no per-user history — attribute it to 'system' + // (the admin-JWT identity), marked so audits can tell migrated content + // apart from other system writes + const result = await yhub.computePool.patchYdoc( + { + update, + currentDoc: EMPTY_YDOC, + userid: 'system', + customAttributions: [{ k: 'migration', v: 's3' }], + }, + { room }, + ); + if (result == null) { + // structurally valid but no effective content — nothing to seed + migrationLog.info( + { event: 'seed.empty', docid: room.docid }, + 'legacy s3 object has no effective content; room starts empty', + ); + return 'empty'; + } + await yhub.stream.addMessage(room, { + type: 'ydoc:update:v1', + contentmap: result.contentmap, + update: result.update, + }); + migrationLog.info( + { + event: 'seed.ok', + docid: room.docid, + bytes: update.byteLength, + durationMs: Date.now() - start, + }, + 'seeded legacy doc from s3', + ); + return 'exists'; + } finally { + activeSeeds--; + } + } finally { + if (acquired != null) { + // compare-and-delete: if this seed outlived the lock TTL, another + // seeder holds a fresh lock — a bare DEL would release it under them + redis + .eval( + "if redis.call('get', KEYS[1]) == ARGV[1] then return redis.call('del', KEYS[1]) end", + { keys: [lockKey], arguments: [lockToken] }, + ) + .catch(() => {}); + } + } +}; + +// Resolves when the room is usable (already known, freshly seeded, or +// legitimately empty); rejects to deny access. Idempotent and safe to +// re-enter — it also runs on rechecks and default-purpose REST calls. +const maybeMigrate = async (room) => { + const cached = verdicts.get(room.docid); + if (cached != null && cached.expires > Date.now()) { + if (cached.verdict === 'failed') throw cached.error; + return; + } + let migration = inflightMigrations.get(room.docid); + if (migration == null) { + migration = migrate(room) + .then( + (verdict) => rememberVerdict(room.docid, verdict), + (err) => { + // logged here (once per attempt) rather than per denied connection: + // cached failures deny without new logs until the verdict expires + migrationLog.error( + { event: 'seed.failed', err, docid: room.docid }, + 'soft migration failed; denying access', + ); + if (err?.noCache !== true) { + // transient failures get a short TTL so a hiccup cannot lock a + // doc out for the full poison-object window + rememberVerdict( + room.docid, + 'failed', + err, + isTransient(err) ? TRANSIENT_TTL_MS : VERDICT_TTL_MS.failed, + ); + } + throw err; + }, + ) + .finally(() => inflightMigrations.delete(room.docid)); + inflightMigrations.set(room.docid, migration); + } + return migration; +}; + const auth = createAuthPlugin({ // uws req is only valid synchronously — read headers AND query before first await. async readAuthInfo(req) { @@ -128,18 +455,31 @@ const auth = createAuthPlugin({ ) { return null; } + let doc; try { - const doc = await backendFetch( - `/api/v1.0/documents/${docid}/`, - authInfo, - ); - if (!doc.abilities?.retrieve) { - return null; - } - return doc.abilities.update ? 'rw' : 'r'; + doc = await backendFetch(`/api/v1.0/documents/${docid}/`, authInfo); } catch { return null; } + if (!doc.abilities?.retrieve) { + return null; + } + if (SOFT_MIGRATION) { + // First access to a room yhub does not know: seed it from the legacy + // Django S3 store before admitting the connection. Awaited inside the + // upgrade handler, so the post-upgrade initial sync (which merges + // postgres and the stream from clock 0) is guaranteed to include the + // seed. Runs only for authorized readers. A missing S3 object is the + // brand-new-document case and allows an empty room; a real S3/compute + // failure denies access — an opaque 401 the client retries with + // backoff. + try { + await maybeMigrate({ org, docid, branch }); + } catch { + return null; // already logged (once per attempt) in maybeMigrate + } + } + return doc.abilities.update ? 'rw' : 'r'; }, }); @@ -281,7 +621,9 @@ const api = [ }), ]; -await createYHub({ +// the instance is referenced by the soft-migration helpers above — safe: auth +// callbacks only fire once the server is up, i.e. after this assignment +const yhub = await createYHub({ redis: { url: REDIS, prefix: REDIS_PREFIX,