fix(docs): synchronize Worker session baselines

Author: DR-Univer

packages/docs-layout-worker/src/index.ts

@@ -62,6 +62,7 @@ export class DocsLayoutWorkerClientService extends Disposable implements IDocLay
     private _channelService: ChannelService | null = null;
     private _runtime: IDocsLayoutWorkerRuntime | null = null;
     private _initialization: Promise<void>;
+    private _initialized = false;
     private readonly _performanceTracker = new DocsLayoutWorkerPerformanceTracker();
 
     constructor(
@@ -85,9 +86,14 @@ export class DocsLayoutWorkerClientService extends Disposable implements IDocLay
         await this._initialization;
     }
 
-    async createSession(request: IDocLayoutCreateSessionRequest): Promise<void> {
-        await this._initialization;
-        return this._withTimeout(this._getRuntime().createSession(request), 'create session');
+    createSession(request: IDocLayoutCreateSessionRequest): Promise<void> {
+        if (this._initialized) {
+            return this._withTimeout(this._getRuntime().createSession(request), 'create session');
+        }
+
+        const capturedRequest = Tools.deepClone(request);
+        return this._initialization.then(() =>
+            this._withTimeout(this._getRuntime().createSession(capturedRequest), 'create session'));
     }
 
     async startLayout(request: IDocLayoutStartRequest): Promise<IDocLayoutStartResult> {
@@ -152,6 +158,7 @@ export class DocsLayoutWorkerClientService extends Disposable implements IDocLay
     }
 
     private async _replaceRuntime(): Promise<void> {
+        this._initialized = false;
         this._disposeRuntime();
         const worker = this._workerFactory();
         const channelService = new ChannelService(createDocsLayoutWorkerMessageProtocol(worker, this._performanceTracker));
@@ -161,9 +168,11 @@ export class DocsLayoutWorkerClientService extends Disposable implements IDocLay
             channelService.requestChannel(DOCS_LAYOUT_WORKER_CHANNEL)
         );
         await this._withTimeout(this._verifyCapabilities(), 'verify capabilities');
+        this._initialized = true;
     }
 
     private _disposeRuntime(): void {
+        this._initialized = false;
         this._channelService?.dispose();
         this._channelService = null;
         this._runtime = null;

packages/docs/src/services/__tests__/doc-layout-executor.service.spec.ts

@@ -242,6 +242,90 @@ describe('DocLayoutExecutorService', () => {
         });
     });
 
+    it('folds mutations committed before a deferred snapshot transfer into one consistent Worker baseline', async () => {
+        const injector = univer.__getInjector();
+        const service = injector.get(DocLayoutExecutorService);
+        const executor = createExecutor();
+        service.register(executor);
+        univer.createUnit<IDocumentData, DocumentDataModel>(
+            UniverInstanceType.UNIVER_DOC,
+            createDocumentData('traditional-doc', DocumentFlavor.TRADITIONAL)
+        );
+
+        const textX = new TextX();
+        textX.push({ t: TextXActionType.RETAIN, len: 5 });
+        textX.push({
+            t: TextXActionType.INSERT,
+            len: 1,
+            body: { dataStream: 'A' },
+        });
+        injector.get(ICommandService).syncExecuteCommand(RichTextEditingMutation.id, {
+            unitId: 'traditional-doc',
+            actions: JSONX.getInstance().editOp(textX.serialize(), ['body']),
+            textRanges: [],
+        });
+
+        await service.startLayout(createMountIdentity(), { reason: 'edit', anchor: 6 }, 32);
+
+        expect(executor.createSession).toHaveBeenCalledTimes(1);
+        expect(executor.createSession).toHaveBeenCalledWith(expect.objectContaining({
+            modelRevision: 1,
+            snapshot: expect.objectContaining({
+                body: expect.objectContaining({ dataStream: 'HelloA\r\n' }),
+            }),
+        }));
+        expect(executor.startLayout).toHaveBeenCalledWith(expect.objectContaining({
+            baseRevision: 1,
+            modelRevision: 1,
+            mutations: [],
+        }));
+    });
+
+    it('rebases the deferred Worker snapshot after executor initialization completes', async () => {
+        const injector = univer.__getInjector();
+        const service = injector.get(DocLayoutExecutorService);
+        const executor = createExecutor();
+        let resolveInitialization: () => void = () => {};
+        vi.mocked(executor.initialize).mockImplementation(() => new Promise<void>((resolve) => {
+            resolveInitialization = resolve;
+        }));
+        service.register(executor);
+        univer.createUnit<IDocumentData, DocumentDataModel>(
+            UniverInstanceType.UNIVER_DOC,
+            createDocumentData('traditional-doc', DocumentFlavor.TRADITIONAL)
+        );
+        await new Promise<void>((resolve) => setTimeout(resolve, 20));
+        expect(executor.createSession).not.toHaveBeenCalled();
+
+        const textX = new TextX();
+        textX.push({ t: TextXActionType.RETAIN, len: 5 });
+        textX.push({
+            t: TextXActionType.INSERT,
+            len: 1,
+            body: { dataStream: 'A' },
+        });
+        injector.get(ICommandService).syncExecuteCommand(RichTextEditingMutation.id, {
+            unitId: 'traditional-doc',
+            actions: JSONX.getInstance().editOp(textX.serialize(), ['body']),
+            textRanges: [],
+        });
+        resolveInitialization();
+
+        await service.startLayout(createMountIdentity(), { reason: 'edit', anchor: 6 }, 32);
+
+        expect(executor.createSession).toHaveBeenCalledWith(expect.objectContaining({
+            modelRevision: 1,
+            snapshot: expect.objectContaining({
+                body: expect.objectContaining({ dataStream: 'HelloA\r\n' }),
+            }),
+        }));
+        expect(executor.startLayout).toHaveBeenCalledWith(expect.objectContaining({
+            baseRevision: 1,
+            modelRevision: 1,
+            mutations: [],
+        }));
+    });
+
     it('reports capability initialization failures and unregisters the executor', async () => {
         const service = univer.__getInjector().get(DocLayoutExecutorService);
         const executor = createExecutor();
@@ -325,7 +409,7 @@ describe('DocLayoutExecutorService', () => {
             UniverInstanceType.UNIVER_DOC,
             createDocumentData('traditional-doc', DocumentFlavor.TRADITIONAL)
         );
-        await Promise.resolve();
+        await vi.waitFor(() => expect(executor.createSession).toHaveBeenCalledTimes(1));
 
         const commandService = injector.get(ICommandService);
         for (const [offset, text] of [[5, 'A'], [6, 'B']] as const) {
@@ -445,7 +529,7 @@ describe('DocLayoutExecutorService', () => {
             UniverInstanceType.UNIVER_DOC,
             createDocumentData('traditional-doc', DocumentFlavor.TRADITIONAL)
         );
-        await Promise.resolve();
+        await vi.waitFor(() => expect(executor.createSession).toHaveBeenCalledTimes(1));
 
         const commandService = injector.get(ICommandService);
         const pastedText = '大段粘贴内容'.repeat(1_024);

packages/docs/src/services/doc-layout-executor.service.ts

@@ -140,6 +140,10 @@ export interface IDocLayoutExecutor {
     readonly type: DocLayoutExecutorType;
     initialize(): Promise<void>;
     recover(): Promise<void>;
+    /**
+     * Capture the request synchronously before returning. The source document model remains mutable
+     * while the returned promise tracks remote session creation.
+     */
     createSession(request: IDocLayoutCreateSessionRequest): Promise<void>;
     startLayout(request: IDocLayoutStartRequest): Promise<IDocLayoutStartResult>;
     stepLayout(request: IDocLayoutStepRequest): Promise<IDocLayoutStepResult>;
@@ -210,6 +214,7 @@ export interface IDocLayoutStartOptions {
 export class DocLayoutExecutorService extends Disposable {
     private static readonly _performanceSampleLimit = 2048;
     private _executor: Nullable<IDocLayoutExecutor> = null;
+    private _executorReady: Promise<void> | null = null;
     private _recoveryFailure: { diagnostic: string; unitId: string } | null = null;
     private _sessionEpoch = 0;
     private readonly _sessions = new Map<string, IDocLayoutManagedSession>();
@@ -278,7 +283,9 @@ export class DocLayoutExecutorService extends Disposable {
             diagnostic: null,
             recoveryUnitId: null,
         });
-        executor.initialize().then(() => {
+        const executorReady = executor.initialize();
+        this._executorReady = executorReady;
+        executorReady.then(() => {
             if (this._executor === executor) {
                 this._executorStatus$.next({
                     state: DocLayoutExecutorState.ACTIVE,
@@ -292,6 +299,7 @@ export class DocLayoutExecutorService extends Disposable {
                 return;
             }
             this._executor = null;
+            this._executorReady = null;
             this._disposeManagedSessions();
             this._sessions.clear();
             const diagnostic = error instanceof Error ? error.message : String(error);
@@ -310,6 +318,7 @@ export class DocLayoutExecutorService extends Disposable {
         return toDisposable(() => {
             if (this._executor === executor) {
                 this._executor = null;
+                this._executorReady = null;
                 this._disposeManagedSessions();
                 this._sessions.clear();
                 this._executorStatus$.next({
@@ -364,10 +373,13 @@ export class DocLayoutExecutorService extends Disposable {
         this._hydrationSamples.clear();
 
         try {
-            await executor.recover();
+            const executorReady = executor.recover();
+            this._executorReady = executorReady;
+            await executorReady;
         } catch (error) {
             const recoveryDiagnostic = error instanceof Error ? error.message : String(error);
             this._executor = null;
+            this._executorReady = null;
             this._recoveryFailure = { diagnostic: recoveryDiagnostic, unitId };
             this._executorStatus$.next({
                 state: DocLayoutExecutorState.RECOVERING,
@@ -468,6 +480,7 @@ export class DocLayoutExecutorService extends Disposable {
         }
         this._sessions.clear();
         this._executor = null;
+        this._executorReady = null;
         this._recoveryFailure = null;
         this._executorStatus$.complete();
         super.dispose();
@@ -606,16 +619,7 @@ export class DocLayoutExecutorService extends Disposable {
             disposed: false,
             cancelPendingCreateTask: null,
         };
-        const request: IDocLayoutCreateSessionRequest = {
-            unitId,
-            sessionEpoch: session.sessionEpoch,
-            snapshot: dataModel.getSnapshot(),
-            modelRevision,
-            locale: this._localeService.getCurrentLocale(),
-            localeData: this._localeService.getLocales(),
-            direction: this._localeService.getDirection(),
-        };
-        const createTask = this._createSessionOnNextMacroTask(executor, session, request);
+        const createTask = this._createSessionOnNextMacroTask(executor, session, dataModel);
         session.queue = createTask;
         this._sessions.set(unitId, session);
         try {
@@ -685,20 +689,53 @@ export class DocLayoutExecutorService extends Disposable {
     private _createSessionOnNextMacroTask(
         executor: IDocLayoutExecutor,
         session: IDocLayoutManagedSession,
-        request: IDocLayoutCreateSessionRequest
+        dataModel: DocumentDataModel
     ): Promise<void> {
         return new Promise<void>((resolve, reject) => {
             const cancelTask = requestImmediateMacroTask(() => {
                 session.cancelPendingCreateTask = null;
+                const unitId = dataModel.getUnitId();
                 if (
                     session.disposed ||
                     this._executor !== executor ||
-                    this._sessions.get(request.unitId) !== session
+                    this._sessions.get(unitId) !== session
                 ) {
                     resolve();
                     return;
                 }
-                executor.createSession(request).then(resolve, reject);
+
+                const executorReady = this._executorReady;
+                if (executorReady == null) {
+                    resolve();
+                    return;
+                }
+                executorReady.then(() => {
+                    if (
+                        session.disposed ||
+                        this._executor !== executor ||
+                        this._sessions.get(unitId) !== session ||
+                        this._getEligibleModel(unitId) !== dataModel
+                    ) {
+                        resolve();
+                        return;
+                    }
+
+                    // Mutations committed before the deferred transfer are already present in the
+                    // live model snapshot. Fold them into the baseline instead of replaying them.
+                    session.sourceMutationRevision = dataModel.getMutationRevision();
+                    session.workerRevision = session.modelRevision;
+                    session.pendingMutations.length = 0;
+                    const request: IDocLayoutCreateSessionRequest = {
+                        unitId,
+                        sessionEpoch: session.sessionEpoch,
+                        snapshot: dataModel.getSnapshot(),
+                        modelRevision: session.modelRevision,
+                        locale: this._localeService.getCurrentLocale(),
+                        localeData: this._localeService.getLocales(),
+                        direction: this._localeService.getDirection(),
+                    };
+                    executor.createSession(request).then(resolve, reject);
+                }, reject);
             });
             session.cancelPendingCreateTask = () => {
                 cancelTask();