WEBDIENSTE/lasuite-docs
1
0
Fork 0
mirror of https://github.com/suitenumerique/docs.git synced 2026-09-13 05:07:52 +02:00
Code Issues Packages Projects Releases Wiki Activity

✨(collaboration) replay legacy s3 version history into yhub

Browse Source 
Add POST /collaboration/migrate/v1/docs/{id}, which replays every S3 version
of a document's legacy `{id}/file` object into one gc:false Yjs document and
stores it as a single row at clock 0, crediting each version with its own S3
timestamp. Nothing existing is deleted and nothing goes on the stream, so the
next compaction merges that row like any other. The clock-0 insert is ON
CONFLICT DO NOTHING and migrated ids are kept in a valkey set, so the endpoint
is idempotent without a lock. The activity api then reports the same timeline
as the backend's /documents/{id}/versions/, instead of the single
migration-time change the lazy soft migration leaves behind.

That lazy seed now writes no insertAt/deleteAt. Persisted contentmaps are
merged rather than de-duplicated, so a seed timestamp would survive next to
the real per-version one on the same ids and the activity api would report
whichever the unordered row scan put last. A seed is not an editing event and
has no honest time to report.

Upgrade yhub to 0.5.0, where error codes encode retry semantics (4xx
permanent, 5xx and 429 retryable) and auth plugins may throw apiError(503). A
temporarily unreachable Django backend, JWKS endpoint or legacy S3 store is
now reported as 503 rather than denied like a permission failure, so clients
retry instead of giving up.

The legacy-store code moves out of server.js into migration.js, with the
shared *_FILE secret helper in env.js.

Signed-off-by: Kevin Jahns <kevin.jahns@protonmail.com>
This commit is contained in:
Kevin Jahns
2026-08-13 12:08:28 +02:00
committed by Manuel Raynaud
parent 00ac755283
commit 4c60fc5c65
8 changed files with 898 additions and 393 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/yhub-server/server.js
+126 -352
View File
@@ -1,15 +1,21 @@
import { createHash, randomUUID } from 'node:crypto';
import { readFileSync } from 'node:fs';
import { createHash } from 'node:crypto';
import { createApiEndpoint, createAuthPlugin, createYHub, logger } from '@y/hub';
import {
apiError,
createApiEndpoint,
createAuthPlugin,
createYHub,
} 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) =>
process.env[`${name}_FILE`]
? readFileSync(process.env[`${name}_FILE`], 'utf8').trim()
: process.env[name] || dflt;
import { secret } from './env.js';
// legacy Django/S3 document store — see migration.js and README.md
import {
SOFT_MIGRATION,
fullMigrate,
isTransientFailure,
maybeMigrate,
} from './migration.js';
const PORT = Number(process.env.PORT || 3002);
const REDIS = process.env.REDIS;
@@ -44,61 +50,6 @@ 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.
@@ -122,273 +73,6 @@ 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<void>
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) {
@@ -420,9 +104,20 @@ const auth = createAuthPlugin({
return payload.admin === true
? { userid: 'system', admin: true }
: null;
} catch {
// bad signature / expired / wrong (or missing) audience / JWKS
// unreachable — fail closed
} catch (err) {
// jose tags every token-validation failure with an `ERR_J…` code (bad
// signature, expired, wrong or missing audience) — those are permanent,
// fail closed. A JWKS fetch that times out or never connects has no
// such code (or ERR_JWKS_TIMEOUT): the token may be perfectly valid and
// we simply cannot check it, so report it as retryable instead of
// accusing the caller of forging it.
if (
err?.code === 'ERR_JWKS_TIMEOUT' ||
typeof err?.code !== 'string' ||
!err.code.startsWith('ERR_J')
) {
throw apiError(503, 'Token verification keys are unavailable');
}
return null;
}
}
@@ -437,11 +132,14 @@ const auth = createAuthPlugin({
return { userid: String(user.id), cookie, origin }; // MUST be string (yhub server.js:667)
} catch (err) {
// Only a genuine "not signed in" falls back to the anonymous identity.
// On backend failure (5xx/network) fail closed: a signed-in editor
// authorized under an anon userid would be invisible to the targeted
// reset-connections recheck (users: [<uuid>]) for the connection's
// whole lifetime.
if (err?.status !== 401 && err?.status !== 403) return null;
// On backend failure (5xx/network) still refuse to admit the connection —
// a signed-in editor authorized under an anon userid would be invisible
// to the targeted reset-connections recheck (users: [<uuid>]) for the
// connection's whole lifetime — but report it as retryable rather than as
// an authentication failure the client should give up on.
if (err?.status !== 401 && err?.status !== 403) {
throw apiError(503, 'Authentication backend is unavailable');
}
// anonymous (public docs): stable per-session id — random ids would mint a new
// permanent attribution identity per reconnect
const anon = createHash('sha256')
@@ -467,8 +165,14 @@ const auth = createAuthPlugin({
let doc;
try {
doc = await backendFetch(`/api/v1.0/documents/${docid}/`, authInfo);
} catch {
return null;
} catch (err) {
// the backend answered "no": a real, permanent denial (403 Forbidden)
if (err?.status === 401 || err?.status === 403 || err?.status === 404) {
return null;
}
// it did not answer at all — say so, so the caller retries instead of
// reading a 5xx or a network blip as a permission decision
throw apiError(503, 'Document authorization backend is unavailable');
}
if (!doc.abilities?.retrieve) {
return null;
@@ -480,12 +184,21 @@ const auth = createAuthPlugin({
// 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.
// failure denies access, either way already logged (once per attempt)
// inside maybeMigrate.
try {
await maybeMigrate({ org, docid, branch });
} catch {
return null; // already logged (once per attempt) in maybeMigrate
// `yhub` is declared at the bottom of this file — safe: auth callbacks
// only fire once the server is up, i.e. after that assignment
await maybeMigrate(yhub, { org, docid, branch });
} catch (err) {
// A corrupt or oversized legacy object will fail the same way forever,
// so that denial is permanent (403). An S3 timeout, a network blip or
// momentary seed backpressure will not — 503 tells the caller to come
// back rather than to treat the document as unreadable.
if (isTransientFailure(err)) {
throw apiError(503, 'Legacy document store is unavailable');
}
return null;
}
}
return doc.abilities.update ? 'rw' : 'r';
@@ -527,6 +240,66 @@ const api = [
},
},
}),
// POST /collaboration/migrate/v1/{org}/{docid} — replay a document's full
// legacy version history from the S3 media bucket into yhub (see README.md).
// Backend-internal, like reset-connections: gated to the admin token via the
// 'migrate' access purpose, since it writes history and reads the legacy
// store.
createApiEndpoint('migrate', {
accessPurpose: 'migrate',
post: {
handler: async (req) => {
if (req.org !== ORG) {
return jsonResponse(400, { error: 'Unknown org' });
}
if (!UUID4.test(req.docid)) {
return jsonResponse(400, { error: 'Room name is invalid' });
}
if (req.branch !== 'main') {
// the legacy store is branchless: `{docid}/file` is the main branch
return jsonResponse(400, { error: 'Unknown branch' });
}
if (!SOFT_MIGRATION) {
// the flag is what configures the S3 client (migration.js)
return jsonResponse(503, { error: 'Legacy store is not configured' });
}
// ?force=true replays a document that is already in the migrated set.
// Only safe while its clock-0 row is still there: once compaction has
// folded that row away, a second replay attributes the same content a
// second time and the activity timestamps become ambiguous.
const { status, ...stats } = await fullMigrate(req.yhub, req.room, {
force: req.query.force === 'true',
});
if (status === 'already') {
return jsonResponse(200, {
message: 'Already migrated',
migrated: false,
});
}
if (status === 'empty') {
// brand-new documents never had a legacy object; nothing to replay
// and nothing wrong — a backfill driver treats this as done
return jsonResponse(200, {
message: 'No legacy document in s3',
migrated: false,
...stats,
});
}
if (status === 'nothing') {
return jsonResponse(200, {
message: 'No usable content in the legacy versions',
migrated: false,
...stats,
});
}
return jsonResponse(200, {
message: 'Migration completed',
migrated: true,
...stats,
});
},
},
}),
// POST /collaboration/create-ydoc/v1/{org}/{docid} — create a document's
// initial Yjs state from a RAW binary update (`Y.encodeStateAsUpdate` /
// pycrdt `get_update()` output) posted as application/octet-stream. Unlike
@@ -561,8 +334,9 @@ const api = [
body.byteLength,
);
if (update.byteLength > MAX_CREATE_BYTES) {
// 413 is missing from yhub's status-line map, so the reason phrase
// is empty ("HTTP/1.1 413 ") — legal, and callers switch on the code
// 413 is missing from yhub's status-line map (503 was added in
// 0.5.0, 413 was not), so the reason phrase is empty
// ("HTTP/1.1 413 ") — legal, and callers switch on the code
return jsonResponse(413, { error: 'Update too large' });
}
// <= 3 bytes is yhub's "no effective content" convention (an empty
@@ -630,8 +404,8 @@ const api = [
}),
];
// 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
// referenced by getAccessType above — safe: auth callbacks only fire once the
// server is up, i.e. after this assignment
const yhub = await createYHub({
redis: {
url: REDIS,
@@ -647,6 +421,6 @@ const yhub = await createYHub({
server: { port: PORT, auth, api, apiPrefix: 'collaboration' },
worker: { taskConcurrency: 5 },
// TODO(yhub): worker.events.docUpdate could push snapshots to Django and replace the
// client useSaveDoc PATCH flow — blocked upstream: the payload is a DocTable without
// room/org/docid (yhub src/index.js:90); needs an upstream change first.
// client useSaveDoc PATCH flow. No longer blocked upstream — yhub 0.5.0 adds `room`
// to the event payload, which was the missing piece.
});