Claude Opus 4.8
acd1cc32b7
Some checks failed
CD / Promote to Int (push) Blocked by required conditions
CD / Build and push images (push) Successful in 2m56s
CI / Lint, typecheck, test (push) Failing after 1m17s
CI / Auth e2e pack (push) Successful in 2m24s
CI / Build container images (push) Has been skipped
CD / Deploy to Test (push) Successful in 9s
CD / Smoke tests against Test (push) Has been cancelled
Update logs grow with every edit; compaction bounds storage and load time. - prisma: `collab_open_sessions` table (page_id, heartbeat_at) — the live-session registry that lets the compaction job avoid pages being edited, decoupled from collab (no api↔collab network call) and self- healing (a crashed collab's rows age out of the freshness window). - collab: `PostgresSessionRegistry` marks a page open on document load and closed on unload, and refreshes an every-30s heartbeat for all open docs; wired into the server hooks and started/stopped in index.ts. - api: `CompactionService` runs hourly via the shared scheduler (#31). For pages with > 500 log rows and no fresh session it merges `page_updates` into `ydoc_state` and deletes the merged rows in one FOR UPDATE transaction — atomic, so a mid-run crash leaves the page untouched and the next run resumes. Content is unchanged (merged state = base + all updates), so the content cache is left as-is; updated_at is deliberately not bumped. Metric log line with pages compacted / rows / bytes removed. Tests: DB-backed compaction test (content hash unchanged, below-threshold skipped, active session skipped then picked up next run, stale heartbeat ignored, idempotent); session-registry DB test (open/close, heartbeat refresh). Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01PGdhRiwU1WRL4XxJfZYipY
84 lines
2.7 KiB
TypeScript
84 lines
2.7 KiB
TypeScript
import { randomUUID } from 'node:crypto';
|
|
|
|
import { Pool } from 'pg';
|
|
import { pino } from 'pino';
|
|
import { afterAll, beforeAll, describe, expect, it } from 'vitest';
|
|
|
|
import { PostgresSessionRegistry } from './session-registry.js';
|
|
import { collabTestDatabaseUrlOrUndefined } from './testing/test-db.js';
|
|
|
|
const url = collabTestDatabaseUrlOrUndefined;
|
|
const logger = pino({ enabled: false });
|
|
|
|
/** Poll `predicate` until it is true or the timeout elapses. */
|
|
async function waitFor(predicate: () => Promise<boolean>, timeoutMs = 2000): Promise<void> {
|
|
const deadline = Date.now() + timeoutMs;
|
|
while (Date.now() < deadline) {
|
|
if (await predicate()) return;
|
|
await new Promise((resolve) => setTimeout(resolve, 20));
|
|
}
|
|
throw new Error('timed out waiting for condition');
|
|
}
|
|
|
|
describe.skipIf(!url)('PostgresSessionRegistry (DB-backed, issue #40)', () => {
|
|
let pool: Pool;
|
|
const created: string[] = [];
|
|
|
|
beforeAll(() => {
|
|
pool = new Pool({ connectionString: url });
|
|
});
|
|
|
|
afterAll(async () => {
|
|
if (created.length > 0) {
|
|
await pool.query('DELETE FROM collab_open_sessions WHERE page_id = ANY($1::text[])', [
|
|
created,
|
|
]);
|
|
}
|
|
await pool.end();
|
|
});
|
|
|
|
async function heartbeatOf(pageId: string): Promise<Date | null> {
|
|
const res = await pool.query<{ heartbeat_at: Date }>(
|
|
'SELECT heartbeat_at FROM collab_open_sessions WHERE page_id = $1',
|
|
[pageId],
|
|
);
|
|
return res.rows[0]?.heartbeat_at ?? null;
|
|
}
|
|
|
|
it('records an open session and clears it on close', async () => {
|
|
const pageId = randomUUID();
|
|
created.push(pageId);
|
|
const registry = new PostgresSessionRegistry({ pool, logger });
|
|
|
|
registry.markOpen(pageId);
|
|
await waitFor(async () => (await heartbeatOf(pageId)) !== null);
|
|
expect(await heartbeatOf(pageId)).toBeInstanceOf(Date);
|
|
|
|
registry.markClosed(pageId);
|
|
await waitFor(async () => (await heartbeatOf(pageId)) === null);
|
|
expect(await heartbeatOf(pageId)).toBeNull();
|
|
});
|
|
|
|
it('refreshes the heartbeat of open documents on its interval', async () => {
|
|
const pageId = randomUUID();
|
|
created.push(pageId);
|
|
const registry = new PostgresSessionRegistry({ pool, logger, heartbeatMs: 40 });
|
|
|
|
registry.markOpen(pageId);
|
|
await waitFor(async () => (await heartbeatOf(pageId)) !== null);
|
|
const first = await heartbeatOf(pageId);
|
|
|
|
registry.start(() => [pageId]);
|
|
try {
|
|
await waitFor(async () => {
|
|
const current = await heartbeatOf(pageId);
|
|
return current !== null && first !== null && current.getTime() > first.getTime();
|
|
});
|
|
} finally {
|
|
registry.stop();
|
|
}
|
|
const refreshed = await heartbeatOf(pageId);
|
|
expect(refreshed!.getTime()).toBeGreaterThan(first!.getTime());
|
|
});
|
|
});
|