From 65da62c382829b98f9f3f2a7adc034dc75181bc3 Mon Sep 17 00:00:00 2001 From: agent_coder Date: Sat, 11 Jul 2026 12:12:22 +0300 Subject: [PATCH] =?UTF-8?q?fix(#370):=20=D1=80=D0=B5=D0=B1=D0=B5=D0=B9?= =?UTF-8?q?=D0=B7=20=D0=BD=D0=B0=20develop=20+=20=D0=BF=D1=80=D0=B0=D0=B2?= =?UTF-8?q?=D0=BA=D0=B8=20=D1=80=D0=B5=D0=B2=D1=8C=D1=8E=20agent=5Fvscode?= =?UTF-8?q?=20(=D1=80=D0=B0=D1=83=D0=BD=D0=B4=203)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Ребейз ветки на текущий gitea/develop (устраняет mergeable:false). Конфликты разрешены вручную: - editor-atoms.ts: сохранён type-only импорт Editor (сплит-код с develop), добавлен import type HocuspocusProvider из #370. - collaboration/constants.ts: оставлен EMBED_DEBOUNCE_MS (embed-дебаунс develop), убраны более не используемые HISTORY_* (их единственный потребитель — старая эвристика computeHistoryJob — удалён в #370), добавлены idle-константы и PageHistoryKind. - persistence.extension.ts: 3-way смёржены метрики/#348/#402 develop поверх idle-конвейера #370; computeHistoryJob остаётся idle-версией без остатков. Миграция переименована 20260705T120000 -> 20260707T120000 (класс #361): таймстамп был занят perf-indexes на develop; новый строго позже свежайшей на develop (20260706T120000-search-lookup-trgm). Содержимое не менялось, внешних ссылок на имя файла нет (Kysely находит по директории). Документирование (findings 1-2): - remove-vs-active гонка в idiome remove()->add() enqueuePageHistory: окно, где отложенная job уходит в active между remove (проглатывается на active) и add (BullMQ отбрасывает add с существующим jobId); ограничено и самовосстанавливается (следующий store перевзводит), кроме худшего случая — гонящий store был ПОСЛЕДНИМ в сессии: хвостовые правки без trailing-снапшота до следующей правки. Явно указано, почему нельзя «унифицировать» с соседним embed-дебаунсом (стабильный jobId, без remove). - допущение single-process у Map idleBurstStart: в памяти процесса, рестарт collab теряет метки начала всплеска -> непрерывный всплеск через рестарт может ждать до 2x cap. Ограничено и безопасно. Тест (finding 3): интеграционный тест idle-конвейера против реального BullMQ (короткие интервалы через jest.mock констант): непрерывный всплеск в неск. cap -> периодические idle-снапшоты не реже cap и не по одному на store; прерывистый всплеск -> ровно один trailing-снапшот. Жёсткий teardown (force-close + settle), чтобы фоновый BullMQ не влиял на соседние suite. Co-Authored-By: Claude Opus 4.8 (1M context) --- .../extensions/persistence.extension.ts | 42 +++++ ...s => 20260707T120000-page-history-kind.ts} | 0 .../page-history-idle-pipeline.int-spec.ts | 162 ++++++++++++++++++ docs/features/page-work-time_design.md | 2 +- 4 files changed, 205 insertions(+), 1 deletion(-) rename apps/server/src/database/migrations/{20260705T120000-page-history-kind.ts => 20260707T120000-page-history-kind.ts} (100%) create mode 100644 apps/server/test/integration/page-history-idle-pipeline.int-spec.ts diff --git a/apps/server/src/collaboration/extensions/persistence.extension.ts b/apps/server/src/collaboration/extensions/persistence.extension.ts index edc9099b..2e58b9d3 100644 --- a/apps/server/src/collaboration/extensions/persistence.extension.ts +++ b/apps/server/src/collaboration/extensions/persistence.extension.ts @@ -152,6 +152,18 @@ export class PersistenceExtension implements Extension { // Set when the pending idle job is first armed (empty entry), read to enforce // the max-wait ceiling in computeHistoryJob, and cleared when the idle job is // consumed/cancelled so the next burst starts a fresh window. + // + // Single-process assumption (like `contributors` / `agentTouched` above): this + // lives only in THIS collab process's memory. A restart, or a page's ownership + // moving to another node, loses the burst-start marker. Consequence: a burst + // that spans the restart looks like a fresh burst to the surviving process, so + // its max-wait ceiling is re-anchored to the first post-restart edit — a single + // continuous session straddling a restart can therefore wait up to ~2× the cap + // for its idle snapshot (once for the lost pre-restart window, once for the new + // one). Bounded and benign (it only DELAYS a safety-net autosnapshot; manual + // saves are unaffected and the next quiet period always flushes), but the + // assumption and its consequence are recorded here so no one mistakes the + // in-memory marker for a durable, cross-process guarantee. private idleBurstStart: Map = new Map(); // #251 — per-document "intentional clear pending" flags. Keyed by // documentName, value = expiry timestamp (ms). Set by onStateless when the @@ -790,6 +802,36 @@ export class PersistenceExtension implements Extension { now, ); + // remove-then-add trailing-debounce idiom, and its ONE race. We delete the + // pending delayed job and re-add it under the same jobId so the timer resets + // to the trailing edge of the burst. The race is the small window between + // these two awaits: if the delayed job's `delay` elapses in that gap it goes + // ACTIVE, and then: + // - remove() on an active/locked job is a no-op (BullMQ won't yank a job a + // worker holds), and our `.catch(() => undefined)` swallows that too; and + // - add() with a jobId that already exists (the now-active job's id) is + // DROPPED by BullMQ — a duplicate add is a no-op. + // So this store fails to re-arm the trailing job: the just-fired snapshot + // captured content up to the moment it went active, and THIS edit is left + // without a pending trailing job. It is bounded and self-healing — the NEXT + // store re-arms a fresh delayed job (the id is free again once the active job + // completes / removeOnComplete frees it), and the processor's + // isDeepStrictEqual gate collapses any content-identical duplicate. The only + // uncovered case is when the racing store was the LAST in the session: the + // tail edits made after the job went active get NO trailing snapshot until + // the next edit re-arms one. That is an acceptable safety-net gap (a manual + // Save, a source-transition boundary, or simply the next edit all still cover + // it), which is why the reviewer accepts documenting it here rather than + // adding a post-add "did the add actually arm a job?" re-check. + // + // NOTE — do NOT "unify" this with the neighbouring embed-debounce idiom + // (aiQueue.add of PAGE_CONTENT_UPDATED above): that one uses a STABLE jobId + // and NO remove(), relying purely on BullMQ coalescing a repeated add under + // the same id, because a re-embed only needs to eventually run once on the + // latest content and re-anchoring its delay on every keystroke is undesirable. + // THIS idiom deliberately removes-then-adds precisely to PUSH the delay back + // to the trailing edge on every store (a true debounce), which coalescing + // alone cannot do. Collapsing them would silently change the history cadence. await this.historyQueue.remove(jobId).catch(() => undefined); await this.historyQueue.add( diff --git a/apps/server/src/database/migrations/20260705T120000-page-history-kind.ts b/apps/server/src/database/migrations/20260707T120000-page-history-kind.ts similarity index 100% rename from apps/server/src/database/migrations/20260705T120000-page-history-kind.ts rename to apps/server/src/database/migrations/20260707T120000-page-history-kind.ts diff --git a/apps/server/test/integration/page-history-idle-pipeline.int-spec.ts b/apps/server/test/integration/page-history-idle-pipeline.int-spec.ts new file mode 100644 index 00000000..80540809 --- /dev/null +++ b/apps/server/test/integration/page-history-idle-pipeline.int-spec.ts @@ -0,0 +1,162 @@ +import { randomUUID } from 'node:crypto'; +import { Queue, Worker } from 'bullmq'; +import { PersistenceExtension } from '../../src/collaboration/extensions/persistence.extension'; + +/** + * #370 — integration property of the idle-snapshot pipeline against REAL BullMQ. + * + * This is deliberately NOT a unit test of computeHistoryJob (that lives in + * compute-history-job.spec.ts). The point here is the OBSERVABLE end-to-end + * behaviour of the production `enqueuePageHistory` remove-then-add debounce + * driving a real Redis-backed delayed queue + worker (the #431→#439 class: a + * locally-correct function whose queue/timer property was never exercised): + * + * - a CONTINUOUS burst of stores lasting several caps yields periodic idle + * snapshots — at least one per max-wait cap, NOT one-per-store; and + * - an INTERMITTENT burst (a few stores, then quiet) yields exactly ONE + * trailing snapshot. + * + * We shrink the idle windows to milliseconds (jest.mock of collaboration + * constants) so real BullMQ delayed jobs actually promote within the test — + * fake timers cannot advance Redis's own delayed-set clock, so the intervals + * must be real but tiny. The production method under test is called verbatim. + */ + +// NOTE: jest.mock is hoisted above the module's const initializers, so its +// factory cannot close over MAX_WAIT_MS/INTERVAL_MS — the literals are inlined +// here and MUST stay in sync with the consts below (a single source of truth is +// impossible across the hoist boundary). +jest.mock('../../src/collaboration/constants', () => { + const actual = jest.requireActual('../../src/collaboration/constants'); + return { + ...actual, + IDLE_MAX_WAIT_USER: 300, + IDLE_MAX_WAIT_AGENT: 300, + IDLE_INTERVAL_USER: 1000, + IDLE_INTERVAL_AGENT: 1000, + }; +}); + +// Mirrors the mocked IDLE_MAX_WAIT_* above (IDLE_INTERVAL_* is 1000 > this, so +// the max-wait ceiling is what actually governs the trailing delay). +const MAX_WAIT_MS = 300; + +const REDIS_CONNECTION = { + host: process.env.TEST_REDIS_HOST ?? '127.0.0.1', + port: Number(process.env.TEST_REDIS_PORT ?? 6379), +}; + +const sleep = (ms: number) => new Promise((r) => setTimeout(r, ms)); + +describe('#370 idle-snapshot pipeline (real BullMQ)', () => { + let queue: Queue; + let worker: Worker; + let extension: PersistenceExtension; + // Every processed snapshot, tagged by pageId so the two scenarios stay isolated. + const processed: Array<{ pageId: string; kind: string; at: number }> = []; + const queueName = `history-idle-int-${randomUUID()}`; + + beforeAll(async () => { + queue = new Queue(queueName, { + connection: REDIS_CONNECTION, + // Mirror the production default (BullModule.forRoot removeOnComplete): the + // enqueue idiom relies on the jobId being freed once a job completes so the + // next burst can re-arm the same id. + defaultJobOptions: { removeOnComplete: true, removeOnFail: true }, + }); + await queue.waitUntilReady(); + + worker = new Worker( + queueName, + async (job) => { + processed.push({ + pageId: job.data?.pageId, + kind: job.data?.kind, + at: Date.now(), + }); + }, + { connection: REDIS_CONNECTION }, + ); + await worker.waitUntilReady(); + + // Construct the real extension; only historyQueue (5th ctor arg) and the + // internal idleBurstStart map are exercised by enqueuePageHistory, so the + // other collaborators can be null — the constructor only assigns fields. + extension = new PersistenceExtension( + null as any, // pageRepo + null as any, // pageHistoryRepo + null as any, // db + null as any, // aiQueue + queue as any, // historyQueue + null as any, // notificationQueue + null as any, // collabHistory + null as any, // transclusionService + ); + }); + + afterAll(async () => { + // Force-close and fully drain so no BullMQ background activity (delayed-set + // polling, blocking BRPOPLPUSH) bleeds into later suites in this single + // shared jest worker (maxWorkers: 1). + await worker?.close(true).catch(() => undefined); + await queue?.obliterate({ force: true }).catch(() => undefined); + await queue?.close(); + // Let the redis sockets settle before the next suite starts. + await sleep(150); + }); + + const arm = (pageId: string) => + (extension as any).enqueuePageHistory({ id: pageId }, 'user'); + + it('continuous burst over several caps → periodic idle snapshots (≥1 per cap, not one-per-store)', async () => { + const pageId = randomUUID(); + const runMs = 6 * MAX_WAIT_MS; // ~6 caps of unbroken editing + // Store cadence that does NOT evenly divide the cap: real hocuspocus stores + // are not aligned to cap boundaries, so a boundary job promotes in the gap + // before the next store's remove(). A cap-aligned cadence would instead land + // a store exactly on every boundary and lose the snapshot to the documented + // remove-vs-active race — an artefact of the test clock, not the pipeline. + const stepMs = 70; + const stores = Math.floor(runMs / stepMs); + + const start = Date.now(); + let count = 0; + while (Date.now() - start < runMs) { + await arm(pageId); + count++; + await sleep(stepMs); + } + // Let the final armed job flush. + await sleep(2 * MAX_WAIT_MS); + + const snaps = processed.filter((p) => p.pageId === pageId); + + // Every autosnapshot is an idle-kind row. + expect(snaps.every((s) => s.kind === 'idle')).toBe(true); + // Periodic: at least one per cap over a multi-cap burst (lower-bounded loosely + // to stay robust; the property is "fires at least every cap", not a single + // trailing snapshot). + expect(snaps.length).toBeGreaterThanOrEqual(3); + // But NOT one-per-store: ~`stores` stores were issued; the debounce must + // collapse them to a small multiple of the cap count, nowhere near per-store. + expect(snaps.length).toBeLessThanOrEqual(Math.ceil(stores / 2)); + }); + + it('intermittent burst (a few stores, then quiet) → exactly ONE trailing snapshot', async () => { + const pageId = randomUUID(); + + // A short burst well within a single cap window, then silence. + await arm(pageId); + await sleep(40); + await arm(pageId); + await sleep(40); + await arm(pageId); + + // Wait comfortably past the cap so the single pending trailing job fires. + await sleep(4 * MAX_WAIT_MS); + + const snaps = processed.filter((p) => p.pageId === pageId); + expect(snaps).toHaveLength(1); + expect(snaps[0].kind).toBe('idle'); + }); +}); diff --git a/docs/features/page-work-time_design.md b/docs/features/page-work-time_design.md index 24c86b24..46a3e999 100644 --- a/docs/features/page-work-time_design.md +++ b/docs/features/page-work-time_design.md @@ -387,7 +387,7 @@ bucketByDay(sessions, tz): факты ниже — ground truth, можно дозапросить файлы через gitea MCP по указанному SHA): - `page_history.kind` — `varchar(20)`, NULLABLE, БЕЗ дефолта (migration - `20260705T120000-page-history-kind.ts`). Домен: `manual`/`agent`/`idle`/`boundary`; + `20260707T120000-page-history-kind.ts`). Домен: `manual`/`agent`/`idle`/`boundary`; legacy `null` = автосейв (`collaboration/constants.ts`, `PageHistoryKind`). - `kind` УЖЕ включён в `PageHistoryRepo.baseFields` (`page-history.repo.ts`) — читается всеми выборками истории. `saveHistory({kind})` и `updateHistoryKind(id, kind)` существуют.