fix(#370): ребейз на develop + правки ревью agent_vscode (раунд 3)
Ребейз ветки на текущий 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) <noreply@anthropic.com>
This commit is contained in:
@@ -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<string, number> = 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(
|
||||
|
||||
@@ -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');
|
||||
});
|
||||
});
|
||||
Reference in New Issue
Block a user