fix(#370): ES2021-safe spec, source-specific idle ceiling, processor lock, tested index mapping (review round 1)

F1 [BLOCKER] persistence-store.spec used Array.prototype.at(-1) (ES2022) but the
server targets ES2021, so server tsc failed (TS2550) and ts-jest could not
compile the suite — 22 core manual-save/idle/boundary tests silently did not run
in CI. Replaced with [length - 1] index access.

F2 [WARNING] The idle burst-reset used a hardcoded IDLE_MAX_WAIT_USER for both
tiers, but computeHistoryJob's ceiling is source-specific. On a continuously
agent-edited page the burst marker stayed stale for 5..10m, forcing delay=0 on
every store and writing one idle row per store — the exact per-store bloat the
debounce prevents. The reset now uses the same source-specific max-wait.

F3 [WARNING] The processor did an unlocked findPageLastHistory -> saveHistory,
which TOCTOU-races a concurrent manual-save (that runs under a page-row lock),
producing two page_history rows with identical content (one idle, one manual) and
defeating promote-not-dup. The snapshot decision is now wrapped in executeTx with
the same page-row lock, so the second writer observes the first's committed row
and the isDeepStrictEqual gate collapses the duplicate.

F4 [WARNING] The risky client filtered-index -> full-list mapping had no tests.
Extracted it to a pure resolvePrevSnapshotId(fullItems, id) helper (diff/restore
baseline against the true previous snapshot in the FULL list, never the previous
visible version) and unit-tested it; removed the now-vestigial index threading.

F5/F6 [low] Renamed the misleading ceiling test + fixed its comment; added a
CHANGELOG entry for the user-facing versioning feature.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

apps/server/src/collaboration/extensions/compute-history-job.spec.ts

@@ -43,10 +43,10 @@ describe('computeHistoryJob (#370 — shared trailing idle pipeline)', () => {
   describe('max-wait ceiling', () => {
     const T0 = 1_000_000; // arbitrary fixed epoch for deterministic tests
 
-    it('early in the burst, the full trailing interval is used', () => {
-      // 1 minute into the burst: remaining budget (10m - 1m = 9m) still exceeds
-      // nothing that clamps it below the interval only if interval > remaining.
-      // For USER, interval 60m > remaining 9m, so delay clamps to the remaining.
+    it('once a burst is armed, delay clamps to the remaining max-wait budget', () => {
+      // 1 minute into the burst the USER interval (60m) far exceeds the remaining
+      // max-wait budget (10m - 1m = 9m), so the delay is clamped DOWN to that
+      // remaining budget — the full interval is NOT used once a ceiling applies.
       const { delay } = computeHistoryJob(page, 'user', T0, T0 + 60_000);
       expect(delay).toBe(IDLE_MAX_WAIT_USER - 60_000);
     });

apps/server/src/collaboration/extensions/persistence-store.spec.ts

@@ -560,7 +560,7 @@ describe('PersistenceExtension.onStoreDocument — Approach-A boundary snapshot'
       // The pending idle autosnapshot is cancelled by the explicit version.
       expect(historyQueue.remove).toHaveBeenCalledWith(PAGE_ID);
       const msg = JSON.parse(
-        (document as any).broadcastStateless.mock.calls.at(-1)[0],
+        (document as any).broadcastStateless.mock.calls[(document as any).broadcastStateless.mock.calls.length - 1][0],
       );
       expect(msg).toMatchObject({
         type: 'version.saved',
@@ -576,7 +576,7 @@ describe('PersistenceExtension.onStoreDocument — Approach-A boundary snapshot'
 
       await emitSave(document, 'agent');
 
-      expect(pageHistoryRepo.saveHistory.mock.calls.at(-1)[1]).toEqual(
+      expect(pageHistoryRepo.saveHistory.mock.calls[pageHistoryRepo.saveHistory.mock.calls.length - 1][1]).toEqual(
         expect.objectContaining({ kind: 'agent' }),
       );
     });
@@ -601,7 +601,7 @@ describe('PersistenceExtension.onStoreDocument — Approach-A boundary snapshot'
       );
       expect(pageHistoryRepo.saveHistory).not.toHaveBeenCalled();
       const msg = JSON.parse(
-        (document as any).broadcastStateless.mock.calls.at(-1)[0],
+        (document as any).broadcastStateless.mock.calls[(document as any).broadcastStateless.mock.calls.length - 1][0],
       );
       expect(msg).toMatchObject({ historyId: 'auto-1', alreadySaved: false });
     });
@@ -621,7 +621,7 @@ describe('PersistenceExtension.onStoreDocument — Approach-A boundary snapshot'
       expect(pageHistoryRepo.updateHistoryKind).not.toHaveBeenCalled();
       expect(pageHistoryRepo.saveHistory).not.toHaveBeenCalled();
       const msg = JSON.parse(
-        (document as any).broadcastStateless.mock.calls.at(-1)[0],
+        (document as any).broadcastStateless.mock.calls[(document as any).broadcastStateless.mock.calls.length - 1][0],
       );
       expect(msg).toMatchObject({ alreadySaved: true, kind: 'manual' });
     });

apps/server/src/collaboration/extensions/persistence.extension.ts

@@ -693,12 +693,18 @@ export class PersistenceExtension implements Extension {
     // The FIRST arm of a burst records `burstStart`; computeHistoryJob shrinks
     // the delay to the remaining max-wait budget from that point, so a continuous
     // session cannot re-arm the trailing timer forever and starve the snapshot.
-    // A burst marker older than the (larger, USER) max-wait means the previous
-    // idle job has already fired — start a fresh window instead of firing
-    // immediately on the next edit.
+    // A burst marker older than THIS TIER's max-wait means the previous idle job
+    // has already fired — start a fresh window instead of firing immediately on
+    // the next edit. Must use the SAME source-specific max-wait computeHistoryJob
+    // uses (agent 5m / user 10m): a hardcoded USER ceiling would leave an agent
+    // burst's marker stale for 5..10m, forcing delay=0 on every store in that
+    // window and writing one idle row per store — exactly the per-store bloat the
+    // debounce exists to prevent, on the continuous-agent path.
+    const maxWait =
+      lastUpdatedSource === 'agent' ? IDLE_MAX_WAIT_AGENT : IDLE_MAX_WAIT_USER;
     const now = Date.now();
     let burstStart = this.idleBurstStart.get(page.id);
-    if (burstStart === undefined || now - burstStart >= IDLE_MAX_WAIT_USER) {
+    if (burstStart === undefined || now - burstStart >= maxWait) {
       burstStart = now;
       this.idleBurstStart.set(page.id, burstStart);
     }

apps/server/src/collaboration/processors/history.processor.spec.ts

@@ -66,6 +66,15 @@ describe('HistoryProcessor.process', () => {
     notificationQueue = { add: jest.fn().mockResolvedValue(undefined) };
     generalQueue = { add: jest.fn().mockResolvedValue(undefined) };
 
+    // #370 F3 — the processor now serializes its find+save under a page-row lock
+    // via executeTx. A db whose transaction().execute(fn) runs fn with a trx stub
+    // drives the real executeTx() helper without a database.
+    const db = {
+      transaction: () => ({
+        execute: (fn: (trx: any) => Promise<any>) => fn({ __trx: true }),
+      }),
+    };
+
     // WorkerHost's constructor reads `this.worker`; passing repos positionally
     // matches the constructor and avoids the Nest DI container.
     proc = new HistoryProcessor(
@@ -73,6 +82,7 @@ describe('HistoryProcessor.process', () => {
       pageRepo as any,
       collabHistory as any,
       watcherService as any,
+      db as any,
       notificationQueue as any,
       generalQueue as any,
     );
@@ -134,7 +144,8 @@ describe('HistoryProcessor.process', () => {
     );
     expect(pageHistoryRepo.saveHistory).toHaveBeenCalledWith(
       expect.objectContaining({ id: PAGE_ID }),
-      { contributorIds: ['u1', 'u2'], kind: 'idle' },
+      // #370 F3 — saveHistory now runs inside the locked tx, so it carries trx.
+      expect.objectContaining({ contributorIds: ['u1', 'u2'], kind: 'idle' }),
     );
     expect(generalQueue.add).toHaveBeenCalledWith(
       QueueJob.PAGE_BACKLINKS,

apps/server/src/collaboration/processors/history.processor.ts

@@ -19,6 +19,9 @@ import { isDeepStrictEqual } from 'node:util';
 import { CollabHistoryService } from '../services/collab-history.service';
 import { WatcherService } from '../../core/watcher/watcher.service';
 import { isEmptyParagraphDoc } from '../collaboration.util';
+import { InjectKysely } from 'nestjs-kysely';
+import { KyselyDB } from '@docmost/db/types/kysely.types';
+import { executeTx } from '@docmost/db/utils';
 
 @Processor(QueueName.HISTORY_QUEUE)
 export class HistoryProcessor extends WorkerHost implements OnModuleDestroy {
@@ -29,6 +32,7 @@ export class HistoryProcessor extends WorkerHost implements OnModuleDestroy {
     private readonly pageRepo: PageRepo,
     private readonly collabHistory: CollabHistoryService,
     private readonly watcherService: WatcherService,
+    @InjectKysely() private readonly db: KyselyDB,
     @InjectQueue(QueueName.NOTIFICATION_QUEUE) private notificationQueue: Queue,
     @InjectQueue(QueueName.GENERAL_QUEUE) private generalQueue: Queue,
   ) {
@@ -41,6 +45,9 @@ export class HistoryProcessor extends WorkerHost implements OnModuleDestroy {
     try {
       const { pageId } = job.data;
 
+      // Read the page WITHOUT a lock first, only to bail early on the two cheap
+      // no-write cases (page gone / empty first snapshot) without opening a
+      // transaction. The authoritative check-then-write happens locked below.
       const page = await this.pageRepo.findById(pageId, {
         includeContent: true,
       });
@@ -51,44 +58,81 @@ export class HistoryProcessor extends WorkerHost implements OnModuleDestroy {
         return;
       }
 
-      const lastHistory = await this.pageHistoryRepo.findPageLastHistory(
-        pageId,
-        { includeContent: true },
-      );
+      // #370 F3 — the snapshot decision (findPageLastHistory → saveHistory) must
+      // be serialized against manual-save/boundary writers, which run under a
+      // page-row lock in onStoreDocument. Without it, this processor and a
+      // concurrent manual-save each read the same lastHistory (MVCC), both see
+      // content != lastHistory, and both insert — producing two page_history rows
+      // with IDENTICAL content (one 'idle', one 'manual'), defeating
+      // promote-not-dup and the version-vs-autosave split. Taking the same
+      // page-row lock makes the second writer observe the first's committed row so
+      // the isDeepStrictEqual gate collapses the duplicate. Only the read+write
+      // is transacted; the post-snapshot queue work stays outside.
+      let contributorIds: string[] = [];
+      let snapshotWritten = false;
+      let lastHistoryContent: unknown;
 
-      if (!lastHistory && isEmptyParagraphDoc(page.content as any)) {
-        this.logger.debug(
-          `Skipping first history for page ${pageId}: empty content`,
+      await executeTx(this.db, async (trx) => {
+        const lockedPage = await this.pageRepo.findById(pageId, {
+          includeContent: true,
+          withLock: true,
+          trx,
+        });
+        if (!lockedPage) return;
+
+        const lastHistory = await this.pageHistoryRepo.findPageLastHistory(
+          pageId,
+          { includeContent: true, trx },
         );
-        await this.collabHistory.clearContributors(pageId);
-        return;
-      }
+        lastHistoryContent = lastHistory?.content;
 
-      if (
-        !lastHistory ||
-        !isDeepStrictEqual(lastHistory.content, page.content)
-      ) {
-        const contributorIds = await this.collabHistory.popContributors(pageId);
+        if (!lastHistory && isEmptyParagraphDoc(lockedPage.content as any)) {
+          this.logger.debug(
+            `Skipping first history for page ${pageId}: empty content`,
+          );
+          return;
+        }
 
+        if (
+          lastHistory &&
+          isDeepStrictEqual(lastHistory.content, lockedPage.content)
+        ) {
+          return; // already snapshotted at this content — nothing to write
+        }
+
+        contributorIds = await this.collabHistory.popContributors(pageId);
         try {
           await this.watcherService.addPageWatchers(
             contributorIds,
             pageId,
-            page.spaceId,
-            page.workspaceId,
+            lockedPage.spaceId,
+            lockedPage.workspaceId,
           );
 
           // #370 — every job on this queue is a trailing idle-flush autosnapshot.
-          await this.pageHistoryRepo.saveHistory(page, {
+          await this.pageHistoryRepo.saveHistory(lockedPage, {
             contributorIds,
             kind: job.data.kind ?? 'idle',
+            trx,
           });
+          snapshotWritten = true;
           this.logger.debug(`History created for page: ${pageId}`);
         } catch (err) {
           await this.collabHistory.addContributors(pageId, contributorIds);
           throw err;
         }
+      });
 
+      // No snapshot written (page vanished / empty-first / unchanged content) →
+      // clear the contributor set for the skip cases and stop.
+      if (!snapshotWritten) {
+        if (!lastHistoryContent && isEmptyParagraphDoc(page.content as any)) {
+          await this.collabHistory.clearContributors(pageId);
+        }
+        return;
+      }
+
+      {
         const mentions = extractMentions(page.content);
         const pageMentions = extractPageMentions(mentions);
         const internalLinkSlugIds = extractInternalLinkSlugIds(page.content);
@@ -106,7 +150,7 @@ export class HistoryProcessor extends WorkerHost implements OnModuleDestroy {
             );
           });
 
-        if (contributorIds.length > 0 && lastHistory?.content) {
+        if (contributorIds.length > 0 && lastHistoryContent) {
           await this.notificationQueue
             .add(QueueJob.PAGE_UPDATED, {
               pageId,