Merge pull request 'perf(comment): унести Yjs-обновление comment-mark с HTTP-критического пути (#399)' (#438) from perf/399-comment-resolve-async into develop

Reviewed-on: #438
2026-07-10 07:01:44 +03:00
parent 199fc9aa21 d84e5ddbad
commit 3a55c3097d
9 changed files with 631 additions and 80 deletions
@@ -5,6 +5,16 @@ import {
 } from '@nestjs/common';
 import { CommentService } from './comment.service';
 import { AuditEvent, AuditResource } from '../../common/events/audit-events';
+import { QueueJob } from '../../integrations/queue/constants';
+
+// #399: the resolve/unresolve flip and the ephemeral anchor removal are enqueued
+// as COMMENT_MARK_UPDATE jobs (off the HTTP path), NOT awaited against the collab
+// gateway. applyCommentSuggestion (the document TEXT edit) is untouched — it
+// still runs synchronously via the gateway.
+const markJob = (generalQueue: any, action: string) =>
+  generalQueue.add.mock.calls.find(
+    (c: any[]) => c[0] === QueueJob.COMMENT_MARK_UPDATE && c[1]?.action === action,
+  );

 /**
 * Focused coverage for CommentService.applySuggestion (comment.service.ts).
@@ -59,6 +69,7 @@ describe('CommentService — applySuggestion', () => {
      commentRepo,
      wsService,
      collaborationGateway,
+      generalQueue,
      auditService,
    };
  }
@@ -86,9 +97,15 @@ describe('CommentService — applySuggestion', () => {

  // --- no replies → ephemeral delete branch -------------------------------

-  it('applied=true, no replies → replaces text, hard-deletes, strips the anchor mark, audits APPLIED, outcome=deleted', async () => {
-    const { service, commentRepo, wsService, collaborationGateway, auditService } =
-      makeService({ applied: true, currentText: 'new text' });
+  it('applied=true, no replies → replaces text, hard-deletes, enqueues the anchor-mark removal, audits APPLIED, outcome=deleted', async () => {
+    const {
+      service,
+      commentRepo,
+      wsService,
+      collaborationGateway,
+      generalQueue,
+      auditService,
+    } = makeService({ applied: true, currentText: 'new text' });

    const result = await service.applySuggestion(suggestionComment(), user());

@@ -105,12 +122,20 @@ describe('CommentService — applySuggestion', () => {
    );

    // Ephemeral: the redundant comment is hard-deleted (atomic-conditional) and
-    // its inline anchor mark removed via the deleteCommentMark collab event.
+    // its inline anchor mark removal is ENQUEUED (#399), no longer a sync gateway
+    // call. The gateway was only touched for the applyCommentSuggestion text edit.
    expect(commentRepo.deleteCommentIfChildless).toHaveBeenCalledWith('c-1');
-    expect(collaborationGateway.handleYjsEvent).toHaveBeenCalledWith(
+    const del = markJob(generalQueue, 'delete');
+    expect(del).toBeDefined();
+    expect(del[1]).toMatchObject({
+      documentName: 'page.page-1',
+      commentId: 'c-1',
+      userId: 'user-1',
+    });
+    expect(collaborationGateway.handleYjsEvent).not.toHaveBeenCalledWith(
      'deleteCommentMark',
-      'page.page-1',
-      expect.objectContaining({ commentId: 'c-1', user: expect.any(Object) }),
+      expect.anything(),
+      expect.anything(),
    );
    // No applied stamps are written for a row about to be deleted.
    expect(appliedPatch(commentRepo)).toBeUndefined();
@@ -258,7 +283,7 @@ describe('CommentService — applySuggestion', () => {
    // The suggested text is already applied to the document, but between the
    // hasChildren read and the atomic delete a reply landed. The parent must NOT
    // be hard-deleted (cascade would destroy the reply); resolve the thread.
-    const { service, commentRepo, wsService, collaborationGateway } =
+    const { service, commentRepo, wsService, generalQueue } =
      makeService({ applied: true, currentText: 'new text' }, false, 0);

    const result = await service.applySuggestion(suggestionComment(), user());
@@ -275,11 +300,8 @@ describe('CommentService — applySuggestion', () => {
      .map((c: any[]) => c[0])
      .find((p: any) => 'resolvedAt' in p);
    expect(resolvePatch.resolvedAt).toBeInstanceOf(Date);
-    expect(collaborationGateway.handleYjsEvent).toHaveBeenCalledWith(
-      'resolveCommentMark',
-      'page.page-1',
-      expect.objectContaining({ commentId: 'c-1', resolved: true }),
-    );
+    // The resolve mark is enqueued (#399), not a sync gateway call.
+    expect(markJob(generalQueue, 'resolve')).toBeDefined();
    expect(result.outcome).toBe('resolved');
  });

@@ -313,11 +313,15 @@ describe('CommentService — behavior', () => {
      });

      const [patch] = commentRepo.updateComment.mock.calls[0];
-      expect(patch).toEqual({
+      // #399: resolve/unresolve now also stamps updatedAt (the async mark
+      // worker's race-guard reads it to order out-of-order events). The
+      // resolve-state fields are still cleared to null on unresolve.
+      expect(patch).toMatchObject({
        resolvedAt: null,
        resolvedById: null,
        resolvedSource: null,
      });
+      expect(patch.updatedAt).toBeInstanceOf(Date);
    });

    it("notifies the author when SOMEONE ELSE resolves their comment", async () => {
@@ -1,6 +1,15 @@
 import { BadRequestException } from '@nestjs/common';
 import { CommentService } from './comment.service';
 import { AuditEvent, AuditResource } from '../../common/events/audit-events';
+import { QueueJob } from '../../integrations/queue/constants';
+
+// #399: the inline comment-mark op (resolve flip / ephemeral-suggestion anchor
+// removal) is now enqueued as a COMMENT_MARK_UPDATE job instead of being awaited
+// against the collab gateway on the HTTP path. Find that job by action.
+const markJob = (generalQueue: any, action: string) =>
+  generalQueue.add.mock.calls.find(
+    (c: any[]) => c[0] === QueueJob.COMMENT_MARK_UPDATE && c[1]?.action === action,
+  );

 /**
 * Coverage for CommentService.dismissSuggestion (#329). Dismiss ("Не применять")
@@ -44,7 +53,14 @@ describe('CommentService — dismissSuggestion', () => {
      auditService,
    );

-    return { service, commentRepo, wsService, collaborationGateway, auditService };
+    return {
+      service,
+      commentRepo,
+      wsService,
+      collaborationGateway,
+      generalQueue,
+      auditService,
+    };
  }

  const suggestionComment = (over?: Partial<any>): any => ({
@@ -62,25 +78,30 @@ describe('CommentService — dismissSuggestion', () => {
  });
  const user = (over?: Partial<any>): any => ({ id: 'user-1', ...over });

-  it('no replies → hard-deletes, strips the anchor mark, does NOT touch page text, audits DISMISSED, outcome=deleted', async () => {
-    const { service, commentRepo, wsService, collaborationGateway, auditService } =
-      makeService(false);
+  it('no replies → hard-deletes, enqueues the anchor-mark removal, does NOT touch page text, audits DISMISSED, outcome=deleted', async () => {
+    const {
+      service,
+      commentRepo,
+      wsService,
+      collaborationGateway,
+      generalQueue,
+      auditService,
+    } = makeService(false);

    const result = await service.dismissSuggestion(suggestionComment(), user());

-    // Never applies the suggestion to the document.
-    expect(collaborationGateway.handleYjsEvent).not.toHaveBeenCalledWith(
-      'applyCommentSuggestion',
-      expect.anything(),
-      expect.anything(),
-    );
-    // Hard-delete (atomic-conditional) + strip mark.
+    // Never applies the suggestion to the document (no sync gateway call at all
+    // now — the mark op is off the HTTP path, #399).
+    expect(collaborationGateway.handleYjsEvent).not.toHaveBeenCalled();
+    // Hard-delete (atomic-conditional) + enqueue the anchor-mark strip.
    expect(commentRepo.deleteCommentIfChildless).toHaveBeenCalledWith('c-1');
-    expect(collaborationGateway.handleYjsEvent).toHaveBeenCalledWith(
-      'deleteCommentMark',
-      'page.page-1',
-      expect.objectContaining({ commentId: 'c-1', user: expect.any(Object) }),
-    );
+    const del = markJob(generalQueue, 'delete');
+    expect(del).toBeDefined();
+    expect(del[1]).toMatchObject({
+      documentName: 'page.page-1',
+      commentId: 'c-1',
+      userId: 'user-1',
+    });
    expect(wsService.emitCommentEvent).toHaveBeenCalledWith(
      'space-1',
      'page-1',
@@ -96,20 +117,20 @@ describe('CommentService — dismissSuggestion', () => {
    expect(result.outcome).toBe('deleted');
  });

-  it('no replies → if the anchor-mark removal FAILS, the row is NOT deleted and the error propagates (#329: no orphan anchor)', async () => {
-    const { service, commentRepo, wsService, collaborationGateway } =
-      makeService(false);
-    // Mark removal is FATAL and runs BEFORE the irreversible row delete: a collab
-    // failure (e.g. COLLAB_DISABLE_REDIS "no live instance") must abort the whole
-    // operation, leaving row + mark consistent — never a deleted row with an
-    // orphan anchor left in the document reporting success.
-    collaborationGateway.handleYjsEvent = jest.fn(async () => {
-      throw new Error('requires a live collaboration instance');
+  it('no replies → if the anchor-mark ENQUEUE FAILS, the row is NOT deleted and the error propagates (#329/#399: no orphan anchor)', async () => {
+    const { service, commentRepo, wsService, generalQueue } = makeService(false);
+    // #399: the mark removal now runs async in a worker, but the ENQUEUE is
+    // awaited BEFORE the irreversible row delete — so the anchor-removal job is
+    // durably scheduled before the row can vanish. If even the enqueue fails
+    // (e.g. Redis down), the whole operation aborts, leaving row + mark
+    // consistent — never a deleted row with an orphan anchor reporting success.
+    generalQueue.add = jest.fn(async () => {
+      throw new Error('queue add failed: no redis');
    });

    await expect(
      service.dismissSuggestion(suggestionComment(), user()),
-    ).rejects.toThrow(/live collaboration/);
+    ).rejects.toThrow(/queue add failed/);

    expect(commentRepo.deleteCommentIfChildless).not.toHaveBeenCalled();
    expect(wsService.emitCommentEvent).not.toHaveBeenCalledWith(
@@ -120,23 +141,29 @@ describe('CommentService — dismissSuggestion', () => {
  });

  it('WITH replies → resolves (not delete), does NOT apply, audits DISMISSED, outcome=resolved', async () => {
-    const { service, commentRepo, wsService, collaborationGateway, auditService } =
-      makeService(true);
+    const {
+      service,
+      commentRepo,
+      collaborationGateway,
+      generalQueue,
+      auditService,
+    } = makeService(true);

    const result = await service.dismissSuggestion(suggestionComment(), user());

-    // Resolved via resolveComment (resolve patch + resolve mark), NOT deleted.
+    // Resolved via resolveComment (resolve patch + enqueued resolve mark), NOT
+    // deleted.
    const resolvePatch = commentRepo.updateComment.mock.calls
      .map((c: any[]) => c[0])
      .find((p: any) => 'resolvedAt' in p);
    expect(resolvePatch.resolvedAt).toBeInstanceOf(Date);
    expect(resolvePatch.resolvedById).toBe('user-1');
    expect(commentRepo.deleteComment).not.toHaveBeenCalled();
-    expect(collaborationGateway.handleYjsEvent).toHaveBeenCalledWith(
-      'resolveCommentMark',
-      'page.page-1',
-      expect.objectContaining({ commentId: 'c-1', resolved: true }),
-    );
+    // No sync gateway call; the resolve mark is enqueued (#399).
+    expect(collaborationGateway.handleYjsEvent).not.toHaveBeenCalled();
+    const res = markJob(generalQueue, 'resolve');
+    expect(res).toBeDefined();
+    expect(res[1]).toMatchObject({ documentName: 'page.page-1', commentId: 'c-1' });
    // No applied stamp — dismiss does not apply the edit.
    const appliedPatch = commentRepo.updateComment.mock.calls
      .map((c: any[]) => c[0])
@@ -156,8 +183,7 @@ describe('CommentService — dismissSuggestion', () => {
    // but the atomic delete matches 0 rows because a reply landed in the window
    // between that read and the delete. The parent must NOT be hard-deleted
    // (a cascade would destroy the just-added reply); the thread is resolved.
-    const { service, commentRepo, wsService, collaborationGateway } =
-      makeService(false, 0);
+    const { service, commentRepo, wsService, generalQueue } = makeService(false, 0);

    const result = await service.dismissSuggestion(suggestionComment(), user());

@@ -175,11 +201,9 @@ describe('CommentService — dismissSuggestion', () => {
      .find((p: any) => 'resolvedAt' in p);
    expect(resolvePatch.resolvedAt).toBeInstanceOf(Date);
    expect(resolvePatch.resolvedById).toBe('user-1');
-    expect(collaborationGateway.handleYjsEvent).toHaveBeenCalledWith(
-      'resolveCommentMark',
-      'page.page-1',
-      expect.objectContaining({ commentId: 'c-1', resolved: true }),
-    );
+    // A resolve mark job is enqueued (the anchor was already delete-marked; the
+    // resolve mirror is idempotent — #399).
+    expect(markJob(generalQueue, 'resolve')).toBeDefined();
    expect(result.outcome).toBe('resolved');
  });

@@ -0,0 +1,179 @@
+import { Logger } from '@nestjs/common';
+import { CommentService } from './comment.service';
+import { QueueJob } from '../../integrations/queue/constants';
+
+// Flush pending microtasks so a fire-and-forget `.catch(...)` runs before we assert.
+const flushMicrotasks = () => new Promise((r) => setImmediate(r));
+
+/**
+ * #399: the comment inline-mark update is moved OFF the HTTP critical path.
+ * resolveComment / unresolve / the ephemeral-suggestion delete must NO LONGER
+ * await CollaborationGateway.handleYjsEvent (which loaded the whole Y.Doc and
+ * ran the store pipeline synchronously, ~4.5s p95). Instead they enqueue an
+ * idempotent COMMENT_MARK_UPDATE job onto the GENERAL_QUEUE with the payload the
+ * worker replays.
+ *
+ * The service is constructed directly with jest mocks (the @InjectQueue tokens
+ * cannot be resolved by Test.createTestingModule — see comment.service.spec.ts).
+ */
+describe('CommentService — async comment mark (#399)', () => {
+  function makeService() {
+    const commentRepo: any = {
+      findById: jest.fn(async (id: string) => ({
+        id,
+        content: {},
+        spaceId: 'space-1',
+        pageId: 'page-1',
+      })),
+      updateComment: jest.fn(async () => undefined),
+      hasChildren: jest.fn(async () => false),
+      deleteCommentIfChildless: jest.fn(async () => 1),
+    };
+    const pageRepo: any = {};
+    const wsService: any = { emitCommentEvent: jest.fn() };
+    // The gateway MUST NOT be touched on the HTTP path anymore.
+    const collaborationGateway: any = {
+      handleYjsEvent: jest.fn(async () => undefined),
+    };
+    const generalQueue: any = { add: jest.fn(() => Promise.resolve()) };
+    const notificationQueue: any = { add: jest.fn(async () => undefined) };
+    const auditService: any = { log: jest.fn() };
+
+    const service = new CommentService(
+      commentRepo,
+      pageRepo,
+      wsService,
+      collaborationGateway,
+      generalQueue,
+      notificationQueue,
+      auditService,
+    );
+    return {
+      service,
+      commentRepo,
+      collaborationGateway,
+      generalQueue,
+      auditService,
+    };
+  }
+
+  const comment = (over?: Partial<any>): any => ({
+    id: 'c-1',
+    creatorId: 'user-1',
+    pageId: 'page-1',
+    spaceId: 'space-1',
+    workspaceId: 'ws-1',
+    ...over,
+  });
+  const user = (over?: Partial<any>): any => ({ id: 'user-1', ...over });
+
+  const markJob = (generalQueue: any) =>
+    generalQueue.add.mock.calls.find(
+      (c: any[]) => c[0] === QueueJob.COMMENT_MARK_UPDATE,
+    );
+
+  it('resolveComment does NOT call the gateway synchronously, and enqueues a resolve mark job', async () => {
+    const { service, collaborationGateway, generalQueue } = makeService();
+
+    await service.resolveComment(comment(), true, user());
+
+    // The whole point of #399: the Y.Doc mark op is off the HTTP path.
+    expect(collaborationGateway.handleYjsEvent).not.toHaveBeenCalled();
+
+    const job = markJob(generalQueue);
+    expect(job).toBeDefined();
+    expect(job[0]).toBe(QueueJob.COMMENT_MARK_UPDATE);
+    expect(job[1]).toMatchObject({
+      documentName: 'page.page-1',
+      commentId: 'c-1',
+      action: 'resolve',
+      userId: 'user-1',
+    });
+    expect(typeof job[1].ts).toBe('number');
+    // ts equals the resolvedAt stamp written to the row (shared timestamp).
+    const [patch] = (service as any).commentRepo.updateComment.mock.calls[0];
+    expect(job[1].ts).toBe((patch.resolvedAt as Date).getTime());
+    expect(job[1].ts).toBe((patch.updatedAt as Date).getTime());
+  });
+
+  it('unresolve enqueues an unresolve mark job (action mapped from resolved=false)', async () => {
+    const { service, collaborationGateway, generalQueue } = makeService();
+
+    await service.resolveComment(comment(), false, user());
+
+    expect(collaborationGateway.handleYjsEvent).not.toHaveBeenCalled();
+    const job = markJob(generalQueue);
+    expect(job[1]).toMatchObject({
+      documentName: 'page.page-1',
+      commentId: 'c-1',
+      action: 'unresolve',
+      userId: 'user-1',
+    });
+  });
+
+  it('dismissing a childless ephemeral suggestion enqueues a delete mark job (not a sync gateway call)', async () => {
+    const { service, collaborationGateway, generalQueue } = makeService();
+
+    await service.dismissSuggestion(
+      comment({ suggestedText: 'new text', selection: 'old', resolvedAt: null }),
+      user(),
+    );
+
+    // The anchor removal is queued, not awaited against the gateway.
+    expect(collaborationGateway.handleYjsEvent).not.toHaveBeenCalled();
+    const job = markJob(generalQueue);
+    expect(job).toBeDefined();
+    expect(job[1]).toMatchObject({
+      documentName: 'page.page-1',
+      commentId: 'c-1',
+      action: 'delete',
+      userId: 'user-1',
+    });
+    expect(typeof job[1].ts).toBe('number');
+  });
+
+  it('awaits the delete ENQUEUE before the irreversible row hard-delete (ordering preserved)', async () => {
+    const { service, generalQueue, commentRepo } = makeService();
+    const order: string[] = [];
+    generalQueue.add.mockImplementation(async (name: string) => {
+      order.push(`enqueue:${name}`);
+    });
+    commentRepo.deleteCommentIfChildless.mockImplementation(async () => {
+      order.push('delete-row');
+      return 1;
+    });
+
+    await service.dismissSuggestion(
+      comment({ suggestedText: 'new text', selection: 'old', resolvedAt: null }),
+      user(),
+    );
+
+    // The mark-removal job must be durably queued BEFORE the row disappears.
+    expect(order).toEqual([
+      `enqueue:${QueueJob.COMMENT_MARK_UPDATE}`,
+      'delete-row',
+    ]);
+  });
+
+  it('resolve is fire-and-forget: a queue-add rejection does NOT fail the HTTP call (best-effort warn)', async () => {
+    const { service, generalQueue } = makeService();
+    // The queue is unavailable — the whole point of #399 is that this must NOT
+    // propagate out of resolveComment onto the HTTP request.
+    const queueErr = new Error('queue is down');
+    generalQueue.add.mockRejectedValue(queueErr);
+    const warnSpy = jest
+      .spyOn(Logger.prototype, 'warn')
+      .mockImplementation(() => undefined);
+
+    // Must resolve, never throw, even though the enqueue rejects.
+    await expect(service.resolveComment(comment(), true, user())).resolves.not.toThrow();
+
+    // The rejection is swallowed on a microtask AFTER the method returns; flush it.
+    await flushMicrotasks();
+    expect(warnSpy).toHaveBeenCalledWith(
+      expect.stringContaining('Failed to enqueue comment mark update for comment c-1'),
+      queueErr,
+    );
+    warnSpy.mockRestore();
+  });
+});
@@ -21,6 +21,7 @@ import { CursorPaginationResult } from '@docmost/db/pagination/cursor-pagination
 import { QueueJob, QueueName } from '../../integrations/queue/constants';
 import { extractUserMentionIdsFromJson } from '../../common/helpers/prosemirror/utils';
 import {
+  ICommentMarkUpdateJob,
  ICommentNotificationJob,
  ICommentResolvedNotificationJob,
 } from '../../integrations/queue/constants/queue.interface';
@@ -298,7 +299,11 @@ export class CommentService {
    // source is cleared alongside resolvedAt/resolvedById.
    provenance?: AuthProvenanceData,
  ): Promise<Comment> {
-    const resolvedAt = resolved ? new Date() : null;
+    // One shared timestamp: it stamps resolvedAt AND updatedAt on the row and is
+    // carried as the mark job's `ts`, so the worker's race-guard can order this
+    // event against the row's authoritative resolve-state mutation time (#399).
+    const now = new Date();
+    const resolvedAt = resolved ? now : null;
    const resolvedById = resolved ? authUser.id : null;
    const isAgent = provenance?.actor === 'agent';
    // Set the agent marker only when resolving; on unresolve clear it back to
@@ -307,25 +312,33 @@ export class CommentService {
    const resolvedSource = resolved && isAgent ? 'agent' : null;

    await this.commentRepo.updateComment(
-      { resolvedAt, resolvedById, resolvedSource },
+      // Bump updatedAt (not editedAt — that drives the "edited" badge) so the
+      // row records WHEN the resolve state last changed; the async mark worker
+      // compares its job ts against this to skip a superseded out-of-order event.
+      { resolvedAt, resolvedById, resolvedSource, updatedAt: now },
      comment.id,
    );

-    // Reflect the resolved state on the inline comment mark in the
-    // collaborative document so all connected clients stay in sync.
+    // #399: mirror the resolved state onto the inline comment mark OFF the HTTP
+    // critical path. The DB row above is the source of truth (updated in ms); the
+    // mark is an eventual mirror for connected clients, and its failure was
+    // ALREADY swallowed (best-effort warn) — so instead of awaiting the whole
+    // Y.Doc load + immediate store pipeline (~4.5s p95), enqueue an idempotent,
+    // retryable COMMENT_MARK_UPDATE job. (Store-pipeline cost itself is #348's
+    // scope, not duplicated here.)
    const documentName = `page.${comment.pageId}`;
-    try {
-      await this.collaborationGateway.handleYjsEvent(
-        'resolveCommentMark',
-        documentName,
-        { commentId: comment.id, resolved, user: authUser },
-      );
-    } catch (error) {
+    void this.enqueueCommentMarkUpdate(
+      documentName,
+      comment.id,
+      resolved ? 'resolve' : 'unresolve',
+      now.getTime(),
+      authUser.id,
+    ).catch((error) =>
      this.logger.warn(
-        `Failed to update comment mark for comment ${comment.id}`,
+        `Failed to enqueue comment mark update for comment ${comment.id}`,
        error,
-      );
-    }
+      ),
+    );

    // Notify the comment author when someone else resolves their comment.
    if (resolved && comment.creatorId !== authUser.id) {
@@ -671,23 +684,54 @@ export class CommentService {
  }

  /**
-   * Remove the inline `comment` mark for a comment from the collaborative
-   * document. FATAL, NOT best-effort: unlike resolveComment (which keeps the row,
-   * so a failed mark update is recoverable), this is used before an irreversible
-   * hard-delete, so the mark removal MUST succeed or throw. Under
-   * COLLAB_DISABLE_REDIS the gateway invokes the deleteCommentMark handler
-   * directly (never a silent no-op) and a missing live instance surfaces as a
-   * thrown error, which we let propagate so the caller aborts before deleting.
+   * Schedule removal of the inline `comment` anchor mark from the collaborative
+   * document (ephemeral suggestion #329), OFF the HTTP critical path (#399).
+   *
+   * ORDERING PRESERVED: we `await` the ENQUEUE (a fast Redis add), not the mark
+   * op, and the caller only proceeds to the irreversible row hard-delete after
+   * this resolves. So the anchor-removal job is DURABLY queued before the row
+   * vanishes — a queue-add failure throws here and aborts the delete (row + mark
+   * stay consistent), preserving the invariant the old FATAL sync call gave. The
+   * mark op itself now runs async in the worker: it is idempotent and retried
+   * (3 attempts), so a transient collab failure self-heals; only an exhausted-
+   * retries job leaves a DB↔mark divergence, now VISIBLE via BullMQ failed-job
+   * metrics (was a hard 5xx before). Delete carries no state guard — the row is
+   * being removed, and stripping an absent mark is a no-op.
   */
  private async deleteCommentMark(comment: Comment, user: User): Promise<void> {
    const documentName = `page.${comment.pageId}`;
-    await this.collaborationGateway.handleYjsEvent(
-      'deleteCommentMark',
+    await this.enqueueCommentMarkUpdate(
      documentName,
-      { commentId: comment.id, user },
+      comment.id,
+      'delete',
+      Date.now(),
+      user.id,
    );
  }

+  /**
+   * Enqueue an idempotent COMMENT_MARK_UPDATE job (#399) — the single path that
+   * mirrors a comment's inline-mark state into the collab Y.Doc off the HTTP
+   * response. The worker (GeneralQueueProcessor) runs the SAME handleYjsEvent
+   * the sync code used, so the mark op is byte-identical.
+   */
+  private enqueueCommentMarkUpdate(
+    documentName: string,
+    commentId: string,
+    action: 'resolve' | 'unresolve' | 'delete',
+    ts: number,
+    userId: string,
+  ): Promise<unknown> {
+    const jobData: ICommentMarkUpdateJob = {
+      documentName,
+      commentId,
+      action,
+      ts,
+      userId,
+    };
+    return this.generalQueue.add(QueueJob.COMMENT_MARK_UPDATE, jobData);
+  }
+
  private async queueCommentNotification(
    content: any,
    oldMentionIds: string[],
@@ -61,6 +61,9 @@ export enum QueueJob {

  COMMENT_NOTIFICATION = 'comment-notification',
  COMMENT_RESOLVED_NOTIFICATION = 'comment-resolved-notification',
+  // #399: off-critical-path mirror of a comment's inline mark into the collab
+  // Y.Doc (resolve/unresolve flip, or ephemeral-suggestion anchor removal).
+  COMMENT_MARK_UPDATE = 'comment-mark-update',
  PAGE_MENTION_NOTIFICATION = 'page-mention-notification',
  PAGE_PERMISSION_GRANTED = 'page-permission-granted',
  PAGE_UPDATE_DIGEST = 'page-update-digest',
@@ -63,6 +63,33 @@ export interface ICommentNotificationJob {
  notifyWatchers: boolean;
 }

+/**
+ * GENERAL_QUEUE payload for the off-critical-path comment inline-mark mirror
+ * (#399). The comment DB row is the source of truth and is already updated
+ * synchronously (ms); this job flips/removes the inline `comment` mark in the
+ * collaborative Y.Doc for connected clients, OFF the HTTP response path, so
+ * `POST /api/comments/resolve` no longer waits the whole Y.Doc load + store
+ * pipeline (was ~4.5s p95). The mark op is idempotent, so BullMQ retries are
+ * safe.
+ *
+ * `action`:
+ *   - 'resolve' / 'unresolve' → flip the mark's `resolved` attribute (exactly
+ *     what the synchronous resolveCommentMark path did);
+ *   - 'delete' → strip the anchor mark entirely (ephemeral suggestion #329).
+ * `ts` is the DB-mutation timestamp (ms). The worker's race-guard uses it (with
+ * the row's authoritative resolved state) to skip a resolve/unresolve event
+ * that a newer, opposite event has already superseded (out-of-order drain).
+ * `userId` supplies the connection-context user the store pipeline attributes
+ * the change to (persistence.extension reads context.user.id).
+ */
+export interface ICommentMarkUpdateJob {
+  documentName: string;
+  commentId: string;
+  action: 'resolve' | 'unresolve' | 'delete';
+  ts: number;
+  userId: string;
+}
+
 export interface ICommentResolvedNotificationJob {
  commentId: string;
  commentCreatorId: string;
@@ -0,0 +1,151 @@
+import { Job } from 'bullmq';
+import { GeneralQueueProcessor } from './general-queue.processor';
+import { QueueJob } from '../constants';
+import { ICommentMarkUpdateJob } from '../constants/queue.interface';
+
+/**
+ * #399: the GENERAL_QUEUE worker replays the comment inline-mark op that used to
+ * run synchronously on the HTTP path. It must call the SAME gateway handler with
+ * the SAME semantics (resolve/unresolve → flip the `resolved` attribute; delete
+ * → strip the anchor), and its timestamp race-guard must skip an event a newer,
+ * opposite event already superseded.
+ */
+describe('GeneralQueueProcessor — COMMENT_MARK_UPDATE (#399)', () => {
+  function makeProc() {
+    const collaborationGateway: any = {
+      handleYjsEvent: jest.fn(async () => undefined),
+    };
+    const commentRepo: any = { findById: jest.fn() };
+    // #399: the processor resolves CollaborationGateway lazily via ModuleRef
+    // (strict:false) to avoid a DI cycle; the fake returns our gateway spy.
+    const moduleRef: any = { get: jest.fn(() => collaborationGateway) };
+    const proc = new GeneralQueueProcessor(
+      {} as any, // db
+      {} as any, // backlinkRepo
+      {} as any, // watcherRepo
+      commentRepo,
+      moduleRef,
+    );
+    return { proc, collaborationGateway, commentRepo };
+  }
+
+  const job = (data: ICommentMarkUpdateJob): Job =>
+    ({ name: QueueJob.COMMENT_MARK_UPDATE, data }) as unknown as Job;
+
+  const base = {
+    documentName: 'page.page-1',
+    commentId: 'c-1',
+    userId: 'user-1',
+  };
+
+  it('resolve → resolveCommentMark with resolved:true and the same-shape args', async () => {
+    const { proc, collaborationGateway, commentRepo } = makeProc();
+    const ts = 1000;
+    // Row reflects the resolve (source of truth), stamped at the same ts.
+    commentRepo.findById.mockResolvedValue({
+      id: 'c-1',
+      resolvedAt: new Date(ts),
+      updatedAt: new Date(ts),
+    });
+
+    await proc.process(job({ ...base, action: 'resolve', ts }));
+
+    expect(collaborationGateway.handleYjsEvent).toHaveBeenCalledTimes(1);
+    expect(collaborationGateway.handleYjsEvent).toHaveBeenCalledWith(
+      'resolveCommentMark',
+      'page.page-1',
+      { commentId: 'c-1', resolved: true, user: { id: 'user-1' } },
+    );
+  });
+
+  it('unresolve → resolveCommentMark with resolved:false', async () => {
+    const { proc, collaborationGateway, commentRepo } = makeProc();
+    const ts = 2000;
+    commentRepo.findById.mockResolvedValue({
+      id: 'c-1',
+      resolvedAt: null,
+      updatedAt: new Date(ts),
+    });
+
+    await proc.process(job({ ...base, action: 'unresolve', ts }));
+
+    expect(collaborationGateway.handleYjsEvent).toHaveBeenCalledWith(
+      'resolveCommentMark',
+      'page.page-1',
+      { commentId: 'c-1', resolved: false, user: { id: 'user-1' } },
+    );
+  });
+
+  it('delete → deleteCommentMark (strip the anchor), no row lookup / no state guard', async () => {
+    const { proc, collaborationGateway, commentRepo } = makeProc();
+
+    await proc.process(job({ ...base, action: 'delete', ts: 123 }));
+
+    expect(collaborationGateway.handleYjsEvent).toHaveBeenCalledWith(
+      'deleteCommentMark',
+      'page.page-1',
+      { commentId: 'c-1', user: { id: 'user-1' } },
+    );
+    // Delete carries no state guard — the row is (being) removed.
+    expect(commentRepo.findById).not.toHaveBeenCalled();
+  });
+
+  it('SKIPS a stale resolve superseded by a newer unresolve (row unresolved, job ts older)', async () => {
+    const { proc, collaborationGateway, commentRepo } = makeProc();
+    // A later unresolve already set the row: resolvedAt null, updatedAt = 5000.
+    commentRepo.findById.mockResolvedValue({
+      id: 'c-1',
+      resolvedAt: null,
+      updatedAt: new Date(5000),
+    });
+
+    // Stale resolve job enqueued at ts=1000 (< 5000), intends resolved=true,
+    // but the row's authoritative state is unresolved → skip.
+    await proc.process(job({ ...base, action: 'resolve', ts: 1000 }));
+
+    expect(collaborationGateway.handleYjsEvent).not.toHaveBeenCalled();
+  });
+
+  it('SKIPS a stale unresolve superseded by a newer resolve', async () => {
+    const { proc, collaborationGateway, commentRepo } = makeProc();
+    commentRepo.findById.mockResolvedValue({
+      id: 'c-1',
+      resolvedAt: new Date(5000),
+      updatedAt: new Date(5000),
+    });
+
+    await proc.process(job({ ...base, action: 'unresolve', ts: 1000 }));
+
+    expect(collaborationGateway.handleYjsEvent).not.toHaveBeenCalled();
+  });
+
+  it('applies when the row state agrees even if ts is older (idempotent, not a stale flip)', async () => {
+    const { proc, collaborationGateway, commentRepo } = makeProc();
+    // Row is resolved and its updatedAt is newer than the job ts, but the state
+    // AGREES with the job → this is a harmless idempotent replay, not a stale
+    // opposite event, so it must still apply.
+    commentRepo.findById.mockResolvedValue({
+      id: 'c-1',
+      resolvedAt: new Date(9000),
+      updatedAt: new Date(9000),
+    });
+
+    await proc.process(job({ ...base, action: 'resolve', ts: 1000 }));
+
+    expect(collaborationGateway.handleYjsEvent).toHaveBeenCalledWith(
+      'resolveCommentMark',
+      'page.page-1',
+      { commentId: 'c-1', resolved: true, user: { id: 'user-1' } },
+    );
+  });
+
+  it('skips (no throw) when the comment row has vanished', async () => {
+    const { proc, collaborationGateway, commentRepo } = makeProc();
+    commentRepo.findById.mockResolvedValue(undefined);
+
+    await expect(
+      proc.process(job({ ...base, action: 'resolve', ts: 1000 })),
+    ).resolves.toBeUndefined();
+    expect(collaborationGateway.handleYjsEvent).not.toHaveBeenCalled();
+  });
+});
@@ -4,6 +4,7 @@ import { Job } from 'bullmq';
 import { QueueJob, QueueName } from '../constants';
 import {
  IAddPageWatchersJob,
+  ICommentMarkUpdateJob,
  IPageBacklinkJob,
 } from '../constants/queue.interface';
 import { InjectKysely } from 'nestjs-kysely';
@@ -13,8 +14,11 @@ import {
  WatcherRepo,
  WatcherType,
 } from '@docmost/db/repos/watcher/watcher.repo';
-import { InsertableWatcher } from '@docmost/db/types/entity.types';
+import { InsertableWatcher, User } from '@docmost/db/types/entity.types';
 import { processBacklinks } from '../tasks/backlinks.task';
+import { ModuleRef } from '@nestjs/core';
+import { CollaborationGateway } from '../../../collaboration/collaboration.gateway';
+import { CommentRepo } from '@docmost/db/repos/comment/comment.repo';

@Processor(QueueName.GENERAL_QUEUE)
 export class GeneralQueueProcessor
@@ -22,14 +26,32 @@ export class GeneralQueueProcessor
  implements OnModuleDestroy
 {
  private readonly logger = new Logger(GeneralQueueProcessor.name);
+  // #399: CollaborationGateway lives in CollaborationModule. We resolve it lazily
+  // via ModuleRef instead of importing that module into the @Global QueueModule —
+  // CollaborationModule's own HistoryProcessor injects this module's global
+  // GENERAL_QUEUE token, so a static import edge here would form a DI cycle. A
+  // lazy strict:false lookup (cached) sidesteps it; the gateway is a singleton in
+  // both the API-server and collab processes that run this worker.
+  private collaborationGateway?: CollaborationGateway;
  constructor(
    @InjectKysely() private readonly db: KyselyDB,
    private readonly backlinkRepo: BacklinkRepo,
    private readonly watcherRepo: WatcherRepo,
+    private readonly commentRepo: CommentRepo,
+    private readonly moduleRef: ModuleRef,
  ) {
    super();
  }

+  private getCollaborationGateway(): CollaborationGateway {
+    if (!this.collaborationGateway) {
+      this.collaborationGateway = this.moduleRef.get(CollaborationGateway, {
+        strict: false,
+      });
+    }
+    return this.collaborationGateway;
+  }
+
  async process(job: Job): Promise<void> {
    try {
      switch (job.name) {
@@ -56,12 +78,87 @@ export class GeneralQueueProcessor
          );
          break;
        }
+
+        case QueueJob.COMMENT_MARK_UPDATE: {
+          await this.processCommentMarkUpdate(
+            job.data as ICommentMarkUpdateJob,
+          );
+          break;
+        }
      }
    } catch (err) {
      throw err;
    }
  }

+  /**
+   * #399: apply a comment's inline-mark mirror in the collab Y.Doc, off the HTTP
+   * critical path. Runs the SAME gateway path the synchronous comment.service
+   * code used (byte-identical mark op):
+   *   - resolve / unresolve → resolveCommentMark (flip the `resolved` attribute);
+   *   - delete → deleteCommentMark (strip the ephemeral-suggestion anchor #329).
+   * The op is idempotent, so a BullMQ retry is safe. Throwing propagates to
+   * WorkerHost → the job is retried and, on exhaustion, surfaces in failed-job
+   * metrics (the divergence is now visible rather than a silently-swallowed warn).
+   */
+  private async processCommentMarkUpdate(
+    data: ICommentMarkUpdateJob,
+  ): Promise<void> {
+    const { documentName, commentId, action, ts, userId } = data;
+    // Minimal connection-context user: the store pipeline reads context.user.id
+    // to attribute the change (persistence.extension). The mark mutation itself
+    // does not depend on the user, so the op stays byte-identical. Deliberate
+    // trade-off: the store pipeline's transient `page.updated` broadcast carries
+    // only { id } here, so its live "who edited" badge loses name/avatarUrl for
+    // this async mark replay. lastUpdatedById is still set correctly; the diff is
+    // cosmetic and self-heals on the next real edit — worth it to stay off the
+    // HTTP path and avoid re-loading the users row.
+    const user = { id: userId } as User;
+
+    if (action === 'delete') {
+      await this.getCollaborationGateway().handleYjsEvent(
+        'deleteCommentMark',
+        documentName,
+        { commentId, user },
+      );
+      return;
+    }
+
+    // resolve / unresolve. The comment row is written SYNCHRONOUSLY before this
+    // job is enqueued, so it is the source of truth for the final resolved state
+    // and its updatedAt records when that state last changed. Race-guard: if a
+    // newer, OPPOSITE event has already superseded this one (its ts is older than
+    // the row's last resolve-state mutation AND the row's current resolved state
+    // disagrees with what this job intends — e.g. an unresolve that drained ahead
+    // of this resolve), skip it rather than flip the mark to a stale state.
+    const comment = await this.commentRepo.findById(commentId);
+    if (!comment) {
+      // The comment vanished (e.g. hard-deleted) → nothing left to mirror.
+      return;
+    }
+    const wantResolved = action === 'resolve';
+    const rowResolved = comment.resolvedAt != null;
+    const rowMutatedAt = new Date(comment.updatedAt).getTime();
+    // `<=`, not `<`: on a sub-millisecond tie (two opposite toggles stamped in
+    // the same ms) skip the disagreeing job rather than let queue order decide.
+    // The consistent job (whose intent matches the row) short-circuits on the
+    // first condition, so a real update is never dropped; only a mark that both
+    // disagrees with the row AND is no newer than it is discarded.
+    if (rowResolved !== wantResolved && ts <= rowMutatedAt) {
+      this.logger.debug(
+        `Skipping stale comment mark '${action}' for ${commentId} ` +
+          `(job ts ${ts} < row ${rowMutatedAt}, row resolved=${rowResolved})`,
+      );
+      return;
+    }
+
+    await this.getCollaborationGateway().handleYjsEvent(
+      'resolveCommentMark',
+      documentName,
+      { commentId, resolved: wantResolved, user },
+    );
+  }
+
  @OnWorkerEvent('active')
  onActive(job: Job) {
    this.logger.debug(`Processing ${job.name} job`);