From 09d6aa866564250468f59a0b39d193d4a08d9ff1 Mon Sep 17 00:00:00 2001 From: YONGJAE LEE Date: Thu, 17 Sep 2026 04:18:05 +0900 Subject: [PATCH 1/2] [ZEPPELIN-6661] Cover notebook editor save timing --- .../e2e/models/notebook-keyboard-page.ts | 2 + .../e2e/models/notebook-save-timing.util.ts | 189 ++++++++++++++++ .../e2e/scenarios/notebook-parity.json | 33 ++- .../e2e/scenarios/notebook-parity.md | 12 +- .../persistence/notebook-save-timing.spec.ts | 153 +++++++++++++ .../interfaces/message-notebook.interface.ts | 1 + .../projects/zeppelin-sdk/src/message.ts | 7 +- .../paragraph-base/paragraph-base.spec.ts | 203 +++++++++++++++++- .../app/core/paragraph-base/paragraph-base.ts | 89 ++++++-- .../notebook/paragraph/paragraph.component.ts | 3 +- .../src/app/services/message.service.ts | 16 +- 11 files changed, 665 insertions(+), 43 deletions(-) create mode 100644 zeppelin-web-angular/e2e/models/notebook-save-timing.util.ts create mode 100644 zeppelin-web-angular/e2e/tests/notebook/persistence/notebook-save-timing.spec.ts diff --git a/zeppelin-web-angular/e2e/models/notebook-keyboard-page.ts b/zeppelin-web-angular/e2e/models/notebook-keyboard-page.ts index 0875d18c967d..6d026e679b80 100644 --- a/zeppelin-web-angular/e2e/models/notebook-keyboard-page.ts +++ b/zeppelin-web-angular/e2e/models/notebook-keyboard-page.ts @@ -22,6 +22,7 @@ export class NotebookKeyboardPage extends BasePage { readonly codeEditor: Locator; readonly paragraphContainer: Locator; readonly firstParagraph: Locator; + readonly firstEditorInput: Locator; readonly runButton: Locator; readonly paragraphResult: Locator; readonly newParagraphButton: Locator; @@ -45,6 +46,7 @@ export class NotebookKeyboardPage extends BasePage { this.codeEditor = page.locator('.monaco-editor .monaco-mouse-cursor-text'); this.paragraphContainer = page.locator('zeppelin-notebook-paragraph'); this.firstParagraph = this.paragraphContainer.first(); + this.firstEditorInput = this.firstParagraph.locator('.monaco-editor textarea.inputarea'); this.runButton = page.locator('button[title="Run this paragraph"], button:has-text("Run")'); this.paragraphResult = page.locator(PARAGRAPH_RESULT_SELECTOR); this.newParagraphButton = page.locator('button:has-text("Add Paragraph"), .new-paragraph-button'); diff --git a/zeppelin-web-angular/e2e/models/notebook-save-timing.util.ts b/zeppelin-web-angular/e2e/models/notebook-save-timing.util.ts new file mode 100644 index 000000000000..9ae1e0c27b7b --- /dev/null +++ b/zeppelin-web-angular/e2e/models/notebook-save-timing.util.ts @@ -0,0 +1,189 @@ +/* + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * http://www.apache.org/licenses/LICENSE-2.0 + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +import { expect, Page, WebSocketRoute } from '@playwright/test'; + +const IDLE_SAVE_TIMEOUT_MS = 30000; +const PROXY_TIMEOUT_MS = 15000; +const COMMIT_STABILITY_MS = 2000; +const ZEPPELIN_WS_URL_PATTERN = /\/ws(\?|$)/; + +interface NotebookSocketMessage { + op?: string; + msgId?: string; + data?: { + id?: string; + noteId?: string; + paragraph?: string | Record; + status?: boolean; + }; +} + +interface CommitParagraphMessage extends NotebookSocketMessage { + op: 'COMMIT_PARAGRAPH'; + msgId: string; + data: { + id: string; + noteId: string; + paragraph: string; + }; +} + +export class CommitParagraphSocketProbe { + private shouldHoldFirstCommitResponse = false; + private shouldQueueServerMessages = false; + private heldResponseMsgId: string | null = null; + private collaborativeModeSignal: string | null = null; + private readonly commits: CommitParagraphMessage[] = []; + private readonly forwardedResponseMsgIds: string[] = []; + private readonly heldResponses = new Map(); + private readonly queuedResponses: Array<{ socket: WebSocketRoute; message: string | Buffer }> = []; + + handleClientMessage(server: WebSocketRoute, message: string | Buffer): void { + const parsed = parseSocketMessage(message); + if (parsed?.op === 'PATCH_PARAGRAPH') { + this.collaborativeModeSignal ??= 'client sent PATCH_PARAGRAPH'; + } + if (isCommitParagraphMessage(parsed)) { + this.commits.push(parsed); + if (this.shouldHoldFirstCommitResponse && this.heldResponseMsgId === null) { + this.heldResponseMsgId = parsed.msgId; + } + } + server.send(message); + } + + handleServerMessage(socket: WebSocketRoute, message: string | Buffer): void { + const parsed = parseSocketMessage(message); + if (parsed?.op === 'COLLABORATIVE_MODE_STATUS' && parsed.data?.status === true) { + this.collaborativeModeSignal ??= 'server sent COLLABORATIVE_MODE_STATUS with status true'; + } + if (parsed?.op === 'PARAGRAPH' && parsed.msgId) { + if (parsed.msgId === this.heldResponseMsgId && !this.heldResponses.has(parsed.msgId)) { + this.heldResponses.set(parsed.msgId, { socket, message }); + this.shouldQueueServerMessages = true; + return; + } + } + // Preserve server order while the delayed response is delivered and observed in isolation. + if (this.shouldQueueServerMessages) { + this.queuedResponses.push({ socket, message }); + return; + } + if (parsed?.op === 'PARAGRAPH' && parsed.msgId) { + this.forwardedResponseMsgIds.push(parsed.msgId); + } + socket.send(message); + } + + holdFirstCommitParagraphResponse(): void { + this.shouldHoldFirstCommitResponse = true; + } + + async waitForCommitCount(expectedCount: number): Promise { + await expect + .poll(() => this.collaborativeModeSignal !== null || this.commits.length >= expectedCount, { + timeout: IDLE_SAVE_TIMEOUT_MS + }) + .toBe(true); + if (this.collaborativeModeSignal !== null) { + throw new Error( + `Notebook entered collaborative mode (${this.collaborativeModeSignal}); edits are sent as PATCH_PARAGRAPH, so no COMMIT_PARAGRAPH will arrive` + ); + } + return this.commits.slice(); + } + + async waitForHeldResponse(msgId: string): Promise { + await expect.poll(() => this.heldResponses.has(msgId), { timeout: PROXY_TIMEOUT_MS }).toBe(true); + } + + async waitForForwardedResponse(msgId: string): Promise { + await expect + .poll(() => this.forwardedResponseCount(msgId), { timeout: PROXY_TIMEOUT_MS }) + .toBeGreaterThanOrEqual(1); + } + + async expectCommitCountToStay(expectedCount: number): Promise { + const stableUntil = Date.now() + COMMIT_STABILITY_MS; + await expect + .poll(() => (this.commits.length !== expectedCount || Date.now() >= stableUntil ? this.commits.length : null), { + timeout: PROXY_TIMEOUT_MS + }) + .toBe(expectedCount); + } + + commitCount(): number { + return this.commits.length; + } + + forwardedResponseCount(msgId: string): number { + return this.forwardedResponseMsgIds.filter(forwardedMsgId => forwardedMsgId === msgId).length; + } + + releaseHeldResponseWithParagraphTitle(msgId: string, title: string): void { + const held = this.heldResponses.get(msgId); + if (!held) { + throw new Error(`No held PARAGRAPH response for msgId ${msgId}`); + } + const parsed = parseSocketMessage(held.message); + if (parsed?.op !== 'PARAGRAPH' || typeof parsed.data?.paragraph !== 'object') { + throw new Error(`Held response for msgId ${msgId} is not a PARAGRAPH snapshot`); + } + // The held response only echoes the committed text, so a marker title lets the test observe when it is applied. + const paragraph = parsed.data.paragraph; + const config = typeof paragraph.config === 'object' && paragraph.config !== null ? paragraph.config : {}; + paragraph.title = title; + paragraph.config = { ...config, title: true }; + + this.heldResponses.delete(msgId); + this.heldResponseMsgId = null; + this.shouldHoldFirstCommitResponse = false; + this.forwardedResponseMsgIds.push(msgId); + held.socket.send(JSON.stringify(parsed)); + } + + releaseQueuedResponses(): void { + this.shouldQueueServerMessages = false; + for (const queued of this.queuedResponses.splice(0)) { + this.handleServerMessage(queued.socket, queued.message); + } + } +} + +export const installCommitParagraphProbe = async (page: Page): Promise => { + const probe = new CommitParagraphSocketProbe(); + await page.routeWebSocket(ZEPPELIN_WS_URL_PATTERN, socket => { + const server = socket.connectToServer(); + socket.onMessage(message => probe.handleClientMessage(server, message)); + server.onMessage(message => probe.handleServerMessage(socket, message)); + }); + return probe; +}; + +const parseSocketMessage = (message: string | Buffer): NotebookSocketMessage | null => { + try { + return JSON.parse(message.toString()) as NotebookSocketMessage; + } catch { + return null; + } +}; + +const isCommitParagraphMessage = (message: NotebookSocketMessage | null): message is CommitParagraphMessage => { + return ( + message?.op === 'COMMIT_PARAGRAPH' && + typeof message.msgId === 'string' && + typeof message.data?.id === 'string' && + typeof message.data.noteId === 'string' && + typeof message.data.paragraph === 'string' + ); +}; diff --git a/zeppelin-web-angular/e2e/scenarios/notebook-parity.json b/zeppelin-web-angular/e2e/scenarios/notebook-parity.json index 7da8c388176d..03ea679971b0 100644 --- a/zeppelin-web-angular/e2e/scenarios/notebook-parity.json +++ b/zeppelin-web-angular/e2e/scenarios/notebook-parity.json @@ -514,9 +514,14 @@ "runner": "not-applicable" }, "coverage": { - "status": "gap", - "tests": [], - "issues": ["ZEPPELIN-6661"], + "status": "covered", + "tests": [ + { + "path": "zeppelin-web-angular/e2e/tests/notebook/persistence/notebook-save-timing.spec.ts", + "tag": "@NB-PARITY-050" + } + ], + "issues": [], "uncoveredOutcomes": [] }, "implementationEvidence": [ @@ -529,6 +534,10 @@ { "path": "zeppelin-web-angular/e2e/models/notebook-keyboard-page.ts", "symbol": "NotebookKeyboardPage" + }, + { + "path": "zeppelin-web-angular/e2e/models/notebook-save-timing.util.ts", + "symbol": "CommitParagraphSocketProbe" } ] }, @@ -565,9 +574,14 @@ "runner": "not-applicable" }, "coverage": { - "status": "gap", - "tests": [], - "issues": ["ZEPPELIN-6661"], + "status": "covered", + "tests": [ + { + "path": "zeppelin-web-angular/e2e/tests/notebook/persistence/notebook-save-timing.spec.ts", + "tag": "@NB-PARITY-051" + } + ], + "issues": [], "uncoveredOutcomes": [] }, "implementationEvidence": [ @@ -580,7 +594,12 @@ "symbol": "NotebookComponent" } ], - "verificationEvidence": [] + "verificationEvidence": [ + { + "path": "zeppelin-web-angular/e2e/models/notebook-save-timing.util.ts", + "symbol": "CommitParagraphSocketProbe" + } + ] }, { "id": "NB-PARITY-060", diff --git a/zeppelin-web-angular/e2e/scenarios/notebook-parity.md b/zeppelin-web-angular/e2e/scenarios/notebook-parity.md index e90d35f5b8ef..60478f206186 100644 --- a/zeppelin-web-angular/e2e/scenarios/notebook-parity.md +++ b/zeppelin-web-angular/e2e/scenarios/notebook-parity.md @@ -33,8 +33,8 @@ Coverage note: `covered` mechanically means this registry points to a matching e | NB-PARITY-011 | editor | The second Escape after inline completion dismissal blurs the editor | covered | not-applicable | zeppelin-web-angular/e2e/tests/notebook/inline-completion.spec.ts
@NB-PARITY-011 | | | NB-PARITY-021 | result | Text and table result displays preserve output semantics after paragraph execution | partial | owner: allow
writer: allow
reader: deny
runner: allow | zeppelin-web-angular/e2e/tests/notebook/paragraph/paragraph-functionality.spec.ts
@NB-PARITY-021 | ZEPPELIN-6514, ZEPPELIN-6516 | | NB-PARITY-022 | result | Streaming interpreter output accumulates while a paragraph is running | covered | owner: allow
writer: allow
reader: deny
runner: allow | zeppelin-web-angular/e2e/tests/notebook/paragraph/paragraph-functionality.spec.ts
@NB-PARITY-022 | | -| NB-PARITY-050 | persistence | Notebook editor persists the latest text after typing stops | gap | owner: allow
writer: allow
reader: deny
runner: not-applicable | | ZEPPELIN-6661 | -| NB-PARITY-051 | persistence | Notebook editor does not lose an edit made while a prior save is in flight | gap | owner: allow
writer: allow
reader: deny
runner: not-applicable | | ZEPPELIN-6661 | +| NB-PARITY-050 | persistence | Notebook editor persists the latest text after typing stops | covered | owner: allow
writer: allow
reader: deny
runner: not-applicable | zeppelin-web-angular/e2e/tests/notebook/persistence/notebook-save-timing.spec.ts
@NB-PARITY-050 | | +| NB-PARITY-051 | persistence | Notebook editor does not lose an edit made while a prior save is in flight | covered | owner: allow
writer: allow
reader: deny
runner: not-applicable | zeppelin-web-angular/e2e/tests/notebook/persistence/notebook-save-timing.spec.ts
@NB-PARITY-051 | | | NB-PARITY-060 | theme | Notebook honors host theme selection | gap | not-applicable | | ZEPPELIN-6640 | ## Scenario Details @@ -151,26 +151,26 @@ Coverage note: `covered` mechanically means this registry points to a matching e ### NB-PARITY-050 Notebook editor persists the latest text after typing stops - Area: persistence -- Coverage: gap +- Coverage: covered - Interpreter: not-applicable - Role verification: owner: unverified; writer: unverified; reader: unverified; runner: not-applicable - Preconditions: A disposable notebook with one editable paragraph is open. The user can edit the paragraph. - Action: Replace the paragraph text and stop typing long enough for the notebook save path to acknowledge the edit. - Observable outcomes: NB-PARITY-050-OUTCOME-001: The persisted paragraph text equals the latest typed text. NB-PARITY-050-OUTCOME-002: The save assertion is based on observable persistence or wire evidence, not an internal timer. - Implementation evidence: zeppelin-web-angular/src/app/pages/workspace/notebook/paragraph/code-editor/code-editor.component.ts (NotebookParagraphCodeEditorComponent) -- Verification evidence: zeppelin-web-angular/e2e/models/notebook-keyboard-page.ts (NotebookKeyboardPage) +- Verification evidence: zeppelin-web-angular/e2e/models/notebook-keyboard-page.ts (NotebookKeyboardPage); zeppelin-web-angular/e2e/models/notebook-save-timing.util.ts (CommitParagraphSocketProbe) ### NB-PARITY-051 Notebook editor does not lose an edit made while a prior save is in flight - Area: persistence -- Coverage: gap +- Coverage: covered - Interpreter: not-applicable - Role verification: owner: unverified; writer: unverified; reader: unverified; runner: not-applicable - Preconditions: A disposable notebook with one editable paragraph is open. The first paragraph save request can be observed before it completes. - Action: Edit the paragraph, keep the first save in flight, then make a second edit. - Observable outcomes: NB-PARITY-051-OUTCOME-001: The first in-flight save does not overwrite or drop the second edit. NB-PARITY-051-OUTCOME-002: A later observable save or reconciliation persists the second edit. - Implementation evidence: zeppelin-web-angular/src/app/pages/workspace/notebook/paragraph/code-editor/code-editor.component.ts (NotebookParagraphCodeEditorComponent); zeppelin-web-angular/src/app/pages/workspace/notebook/notebook.component.ts (NotebookComponent) -- Verification evidence: not-applicable +- Verification evidence: zeppelin-web-angular/e2e/models/notebook-save-timing.util.ts (CommitParagraphSocketProbe) ### NB-PARITY-060 Notebook honors host theme selection diff --git a/zeppelin-web-angular/e2e/tests/notebook/persistence/notebook-save-timing.spec.ts b/zeppelin-web-angular/e2e/tests/notebook/persistence/notebook-save-timing.spec.ts new file mode 100644 index 000000000000..22394d72915c --- /dev/null +++ b/zeppelin-web-angular/e2e/tests/notebook/persistence/notebook-save-timing.spec.ts @@ -0,0 +1,153 @@ +/* + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * http://www.apache.org/licenses/LICENSE-2.0 + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +import { expect, test } from '@playwright/test'; +import { NotebookKeyboardPage } from '../../../models/notebook-keyboard-page'; +import { CommitParagraphSocketProbe, installCommitParagraphProbe } from '../../../models/notebook-save-timing.util'; +import { addPageAnnotationBeforeEach, createTestNotebook, PAGES, waitForZeppelinReady } from '../../../utils'; + +const PERSISTENCE_TIMEOUT_MS = 15000; +const INITIAL_PARAGRAPH_TEXT = 'E2E save baseline'; + +test.describe('Notebook editor save timing', () => { + addPageAnnotationBeforeEach(PAGES.WORKSPACE.NOTEBOOK); + addPageAnnotationBeforeEach(PAGES.WORKSPACE.NOTEBOOK_PARAGRAPH_CODE_EDITOR); + + let notebookPage: NotebookKeyboardPage; + let commitProbe: CommitParagraphSocketProbe; + let noteId: string | undefined; + + test.beforeEach(async ({ page }) => { + noteId = undefined; + notebookPage = new NotebookKeyboardPage(page); + commitProbe = await installCommitParagraphProbe(page); + + await test.step('Given a disposable notebook with an editable paragraph', async () => { + await page.goto('/#/'); + await waitForZeppelinReady(page); + const notebook = await createTestNotebook(page); + noteId = notebook.noteId; + await page.goto(`/#/notebook/${notebook.noteId}`); + await expect(notebookPage.firstParagraph).toBeVisible({ timeout: 30000 }); + await notebookPage.waitForEditorRendered(0); + await notebookPage.setCodeEditorContent(INITIAL_PARAGRAPH_TEXT); + await expect.poll(() => notebookPage.getParagraphTextByIndex(0)).toBe(INITIAL_PARAGRAPH_TEXT); + await expect + .poll(() => notebookPage.getCodeEditorContentByIndex(0), { timeout: PERSISTENCE_TIMEOUT_MS }) + .toBe(INITIAL_PARAGRAPH_TEXT); + await expect(notebookPage.firstEditorInput).toBeFocused(); + expect(commitProbe.commitCount()).toBe(0); + }); + }); + + test.afterEach(async ({ page }) => { + if (noteId) { + const response = await page.request.delete(`/api/notebook/${noteId}`); + expect(response.ok(), 'Delete the disposable notebook').toBe(true); + } + }); + + test('persists the latest paragraph text after typing stops', { tag: '@NB-PARITY-050' }, async ({ page }) => { + const text = '%md Idle save keeps the latest text'; + + await test.step('When typing stops with the editor still focused', async () => { + await notebookPage.pressSelectAll(); + await page.keyboard.insertText(text); + await expect(notebookPage.firstEditorInput).toBeFocused(); + await expect + .poll(() => notebookPage.getCodeEditorContentByIndex(0), { timeout: PERSISTENCE_TIMEOUT_MS }) + .toBe(text); + }); + + await test.step('Then an automatic save persists the text without a blur or run action', async () => { + const [commit] = await commitProbe.waitForCommitCount(1); + expect(commit.data.paragraph).toBe(text); + await commitProbe.waitForForwardedResponse(commit.msgId); + await expect(notebookPage.firstEditorInput).toBeFocused(); + await expect.poll(() => notebookPage.getParagraphTextByIndex(0), { timeout: PERSISTENCE_TIMEOUT_MS }).toBe(text); + }); + + await test.step('Then reopening the notebook renders the saved text', async () => { + await page.reload(); + await waitForZeppelinReady(page); + await expect + .poll(() => notebookPage.getCodeEditorContentByIndex(0), { timeout: PERSISTENCE_TIMEOUT_MS }) + .toBe(text); + await expect.poll(() => notebookPage.getParagraphTextByIndex(0), { timeout: PERSISTENCE_TIMEOUT_MS }).toBe(text); + }); + }); + + test('keeps an edit made while an earlier save is still pending', { tag: '@NB-PARITY-051' }, async ({ page }) => { + const firstText = '%md First pending save'; + const latestEdit = '; latest edit wins'; + const latestText = `${firstText}${latestEdit}`; + const delayedResponseTitle = 'Delayed save response delivered'; + let firstMsgId: string; + let secondMsgId: string; + + await test.step('When the real server response to the first save is delayed', async () => { + commitProbe.holdFirstCommitParagraphResponse(); + await notebookPage.pressSelectAll(); + await page.keyboard.insertText(firstText); + await expect + .poll(() => notebookPage.getCodeEditorContentByIndex(0), { timeout: PERSISTENCE_TIMEOUT_MS }) + .toBe(firstText); + const [firstCommit] = await commitProbe.waitForCommitCount(1); + expect(firstCommit.data.paragraph).toBe(firstText); + firstMsgId = firstCommit.msgId; + await commitProbe.waitForHeldResponse(firstMsgId); + expect(commitProbe.forwardedResponseCount(firstMsgId)).toBe(0); + }); + + await test.step('When another edit is made before the earlier response arrives', async () => { + await page.keyboard.insertText(latestEdit); + await expect + .poll(() => notebookPage.getCodeEditorContentByIndex(0), { timeout: PERSISTENCE_TIMEOUT_MS }) + .toBe(latestText); + expect(commitProbe.forwardedResponseCount(firstMsgId)).toBe(0); + }); + + await test.step('Then receiving the first save response preserves the unsaved newer edit', async () => { + commitProbe.releaseHeldResponseWithParagraphTitle(firstMsgId, delayedResponseTitle); + await expect(notebookPage.firstParagraph.locator('zeppelin-elastic-input')).toHaveText(delayedResponseTitle); + await expect + .poll(() => notebookPage.getCodeEditorContentByIndex(0), { timeout: PERSISTENCE_TIMEOUT_MS }) + .toBe(latestText); + }); + + await test.step('Then the newer edit is saved exactly once with the latest text', async () => { + const [, secondCommit] = await commitProbe.waitForCommitCount(2); + expect(secondCommit.data.paragraph).toBe(latestText); + secondMsgId = secondCommit.msgId; + await commitProbe.expectCommitCountToStay(2); + commitProbe.releaseQueuedResponses(); + }); + + await test.step('Then the newer automatic save response persists the latest edit', async () => { + await commitProbe.waitForForwardedResponse(secondMsgId); + await expect + .poll(() => notebookPage.getParagraphTextByIndex(0), { timeout: PERSISTENCE_TIMEOUT_MS }) + .toBe(latestText); + }); + + await test.step('Then reopening the notebook preserves the latest edit', async () => { + await page.reload(); + await waitForZeppelinReady(page); + await expect + .poll(() => notebookPage.getCodeEditorContentByIndex(0), { timeout: PERSISTENCE_TIMEOUT_MS }) + .toBe(latestText); + await expect + .poll(() => notebookPage.getParagraphTextByIndex(0), { timeout: PERSISTENCE_TIMEOUT_MS }) + .toBe(latestText); + }); + }); +}); diff --git a/zeppelin-web-angular/projects/zeppelin-sdk/src/interfaces/message-notebook.interface.ts b/zeppelin-web-angular/projects/zeppelin-sdk/src/interfaces/message-notebook.interface.ts index 665e8dfd71f6..ff9711fce6d1 100644 --- a/zeppelin-web-angular/projects/zeppelin-sdk/src/interfaces/message-notebook.interface.ts +++ b/zeppelin-web-angular/projects/zeppelin-sdk/src/interfaces/message-notebook.interface.ts @@ -143,6 +143,7 @@ export interface ParagraphMoved { } export interface UpdateParagraph { + msgId?: string; paragraph: ParagraphItem; } diff --git a/zeppelin-web-angular/projects/zeppelin-sdk/src/message.ts b/zeppelin-web-angular/projects/zeppelin-sdk/src/message.ts index 4d559a86aa1e..3f6408a1156c 100644 --- a/zeppelin-web-angular/projects/zeppelin-sdk/src/message.ts +++ b/zeppelin-web-angular/projects/zeppelin-sdk/src/message.ts @@ -157,7 +157,7 @@ export class Message { return this.received$.asObservable(); } - send(...args: SendArgumentsType): void { + send(...args: SendArgumentsType): string { if (!this.ws) { throw new Error('WebSocket is not connected. Bootstrap first.'); } @@ -172,6 +172,7 @@ export class Message { this.ws.next(message); this.sent$.next(message); + return message.msgId; } receive(op: K): Observable[K]> { @@ -459,7 +460,7 @@ export class Message { paragraphConfig: ParagraphConfig, paragraphParams: ParagraphConfig, noteId: string - ): void { + ): string { return this.send(OP.COMMIT_PARAGRAPH, { id: paragraphId, noteId, @@ -474,7 +475,7 @@ export class Message { // javascript add "," if change contains several patches // but java library requires patch list without "," const normalPatch = patch.replace(/,@@/g, '@@'); - return this.send(OP.PATCH_PARAGRAPH, { + this.send(OP.PATCH_PARAGRAPH, { id: paragraphId, noteId, patch: normalPatch diff --git a/zeppelin-web-angular/src/app/core/paragraph-base/paragraph-base.spec.ts b/zeppelin-web-angular/src/app/core/paragraph-base/paragraph-base.spec.ts index 34d5cc441834..9574649a75fe 100644 --- a/zeppelin-web-angular/src/app/core/paragraph-base/paragraph-base.spec.ts +++ b/zeppelin-web-angular/src/app/core/paragraph-base/paragraph-base.spec.ts @@ -13,7 +13,7 @@ import { ChangeDetectorRef } from '@angular/core'; import { DatasetType, Message, ParagraphItem } from '@zeppelin/sdk'; import { EMPTY } from 'rxjs'; -import { describe, expect, it, vi } from 'vitest'; +import { afterEach, describe, expect, it, vi } from 'vitest'; import { AngularContextManager } from './angular-context-manager'; import { ParagraphBase } from './paragraph-base'; @@ -245,3 +245,204 @@ describe('ParagraphBase streaming boundaries', () => { component.ngOnDestroy(); }); }); + +class SaveTestParagraph extends ParagraphBase { + protected currentNoteId = 'note'; + + changeColWidth(): void {} + updateParagraphResult(): void {} + + trackSave(msgId: string): void { + this.trackParagraphSave(msgId); + } +} + +const saveTestParagraphs: SaveTestParagraph[] = []; + +afterEach(() => { + for (const paragraph of saveTestParagraphs.splice(0)) { + paragraph.ngOnDestroy(); + } +}); + +const createSaveTestParagraph = (text: string, dirtyText?: string): SaveTestParagraph => { + const paragraph = new SaveTestParagraph( + { receive: () => EMPTY } as unknown as Message, + { isParagraphRunning: () => false, isEntireNoteRunning: () => false }, + { + setContextValue: vi.fn(), + unsetContextValue: vi.fn(), + contextChanged: () => EMPTY, + runParagraphAction: () => EMPTY + } as unknown as AngularContextManager, + { markForCheck: vi.fn() } as unknown as ChangeDetectorRef + ); + paragraph.paragraph = { text } as ParagraphItem; + paragraph.originalText = 'previous save'; + paragraph.dirtyText = dirtyText; + saveTestParagraphs.push(paragraph); + return paragraph; +}; + +describe('ParagraphBase save responses', () => { + it.each(['latest edit', ''])('preserves an unsaved edit %j when a broadcast carries the saved text', dirtyText => { + const paragraph = createSaveTestParagraph(dirtyText, dirtyText); + + paragraph.updateAllScopeTexts(paragraph.paragraph!, { text: 'previous save' } as ParagraphItem); + + expect(paragraph.paragraph?.text).toBe(dirtyText); + expect(paragraph.dirtyText).toBe(dirtyText); + expect(paragraph.originalText).toBe('previous save'); + }); + + it('accepts a remote edit when there is no unsaved local edit', () => { + const paragraph = createSaveTestParagraph('previous save'); + + paragraph.updateAllScopeTexts(paragraph.paragraph!, { text: 'remote edit' } as ParagraphItem); + + expect(paragraph.paragraph?.text).toBe('remote edit'); + expect(paragraph.originalText).toBe('remote edit'); + expect(paragraph.dirtyText).toBeUndefined(); + }); + + it('keeps the latest edit when a delayed save acknowledgement arrives after the baseline advanced', () => { + const paragraph = createSaveTestParagraph('committed edit', 'committed edit'); + paragraph.trackSave('save-a'); + paragraph.originalText = 'committed edit'; + paragraph.dirtyText = undefined; + paragraph.paragraph!.text = 'latest edit'; + paragraph.dirtyText = 'latest edit'; + + paragraph.updateAllScopeTexts(paragraph.paragraph!, { text: 'committed edit' } as ParagraphItem, 'save-a'); + + expect(paragraph.paragraph?.text).toBe('latest edit'); + expect(paragraph.dirtyText).toBe('latest edit'); + expect(paragraph.originalText).toBe('committed edit'); + }); + + it('keeps an unsaved edit when a commit acknowledgement arrives without an advanced baseline', () => { + const paragraph = createSaveTestParagraph('latest edit', 'latest edit'); + paragraph.trackSave('save-a'); + + paragraph.updateAllScopeTexts(paragraph.paragraph!, { text: 'committed edit' } as ParagraphItem, 'save-a'); + + expect(paragraph.paragraph?.text).toBe('latest edit'); + expect(paragraph.dirtyText).toBe('latest edit'); + expect(paragraph.originalText).toBe('committed edit'); + }); + + it.each(['latest edit', 'committed edit'])( + 'keeps a later edit when a broadcast repeats a save acknowledged while the editor showed %j', + acknowledgedEditorText => { + const paragraph = createSaveTestParagraph(acknowledgedEditorText, acknowledgedEditorText); + paragraph.trackSave('save-a'); + + paragraph.updateAllScopeTexts(paragraph.paragraph!, { text: 'committed edit' } as ParagraphItem, 'save-a'); + paragraph.paragraph!.text = 'latest edit'; + paragraph.dirtyText = 'latest edit'; + paragraph.updateAllScopeTexts(paragraph.paragraph!, { text: 'committed edit' } as ParagraphItem); + + expect(paragraph.paragraph?.text).toBe('latest edit'); + expect(paragraph.dirtyText).toBe('latest edit'); + expect(paragraph.originalText).toBe('committed edit'); + } + ); + + it('does not rewind a baseline that moved after the acknowledged save was sent', () => { + const paragraph = createSaveTestParagraph('patched edit', 'patched edit'); + paragraph.trackSave('save-a'); + paragraph.originalText = 'patched edit'; + + paragraph.updateAllScopeTexts(paragraph.paragraph!, { text: 'committed edit' } as ParagraphItem, 'save-a'); + + expect(paragraph.paragraph?.text).toBe('patched edit'); + expect(paragraph.originalText).toBe('patched edit'); + }); + + it.each([ + ['an acknowledgement', 'save-a'], + ['a broadcast without msgId', undefined] + ])('keeps collaborative patches when %s arrives with a stale local edit', (_case, msgId) => { + const paragraph = createSaveTestParagraph('patched edit', 'stale edit'); + paragraph.trackSave('save-a'); + paragraph.originalText = 'committed edit'; + + paragraph.updateAllScopeTexts(paragraph.paragraph!, { text: 'committed edit' } as ParagraphItem, msgId); + + expect(paragraph.paragraph?.text).toBe('patched edit'); + }); + + it('keeps a collaborative patch applied after a settings commit when its acknowledgement arrives', () => { + const paragraph = createSaveTestParagraph('committed edit'); + paragraph.originalText = 'committed edit'; + paragraph.trackSave('save-a'); + paragraph.paragraph!.text = 'patched edit'; + paragraph.originalText = 'patched edit'; + + paragraph.updateAllScopeTexts(paragraph.paragraph!, { text: 'committed edit' } as ParagraphItem, 'save-a'); + + expect(paragraph.paragraph?.text).toBe('patched edit'); + expect(paragraph.originalText).toBe('patched edit'); + expect(paragraph.dirtyText).toBeUndefined(); + }); + + it('clears the local edit once its acknowledgement arrives', () => { + const paragraph = createSaveTestParagraph('saved edit'); + paragraph.originalText = 'saved edit'; + paragraph.dirtyText = 'dirty edit'; + paragraph.trackSave('save-a'); + + paragraph.updateAllScopeTexts(paragraph.paragraph!, { text: 'dirty edit' } as ParagraphItem, 'save-a'); + + expect(paragraph.paragraph?.text).toBe('saved edit'); + expect(paragraph.dirtyText).toBeUndefined(); + }); + + it('consumes a tracked save when its acknowledgement does not change the paragraph', () => { + const component = createSaveTestParagraph('saved edit'); + component.paragraph = { ...paragraph('A', 'FINISHED'), text: 'saved edit' }; + component.trackSave('save-a'); + + component.paragraphData({ paragraph: { ...component.paragraph }, msgId: 'save-a' }); + expect(component.originalText).toBe('saved edit'); + + component.originalText = 'previous save'; + component.paragraphData({ paragraph: { ...component.paragraph }, msgId: 'save-a' }); + expect(component.originalText).toBe('previous save'); + }); + + it('accepts a remote edit that is not an acknowledgement of the last local save', () => { + const paragraph = createSaveTestParagraph('local edit', 'local edit'); + + paragraph.updateAllScopeTexts(paragraph.paragraph!, { text: 'remote edit' } as ParagraphItem); + + expect(paragraph.paragraph?.text).toBe('remote edit'); + expect(paragraph.originalText).toBe('remote edit'); + expect(paragraph.dirtyText).toBeUndefined(); + }); + + it('ignores an older acknowledgement after a newer save has been sent', () => { + const paragraph = createSaveTestParagraph('latest save'); + paragraph.originalText = 'latest save'; + paragraph.trackSave('save-a'); + paragraph.trackSave('save-b'); + + paragraph.updateAllScopeTexts(paragraph.paragraph!, { text: 'earlier save' } as ParagraphItem, 'save-a'); + + expect(paragraph.paragraph?.text).toBe('latest save'); + expect(paragraph.originalText).toBe('latest save'); + expect(paragraph.dirtyText).toBeUndefined(); + }); + + it('does not advance the saved baseline for an older acknowledgement while editing', () => { + const paragraph = createSaveTestParagraph('latest edit', 'latest edit'); + paragraph.trackSave('save-a'); + paragraph.trackSave('save-b'); + + paragraph.updateAllScopeTexts(paragraph.paragraph!, { text: 'earlier save' } as ParagraphItem, 'save-a'); + + expect(paragraph.paragraph?.text).toBe('latest edit'); + expect(paragraph.dirtyText).toBe('latest edit'); + expect(paragraph.originalText).toBe('previous save'); + }); +}); diff --git a/zeppelin-web-angular/src/app/core/paragraph-base/paragraph-base.ts b/zeppelin-web-angular/src/app/core/paragraph-base/paragraph-base.ts index b9e514abd6e7..f2e49ea4207b 100644 --- a/zeppelin-web-angular/src/app/core/paragraph-base/paragraph-base.ts +++ b/zeppelin-web-angular/src/app/core/paragraph-base/paragraph-base.ts @@ -63,6 +63,8 @@ export abstract class ParagraphBase extends MessageListenersManager { forms: {} }; private readonly outputState = new ParagraphOutputState(); + private readonly pendingParagraphSaves = new Map(); + private paragraphSaveSequence = 0; constructor( public messageService: Message, @@ -170,21 +172,28 @@ export abstract class ParagraphBase extends MessageListenersManager { this.outputState.finish(newPara.results?.msg); } if (this.isUpdateRequired(oldPara, newPara)) { - this.updateParagraph(oldPara, newPara, () => { - if (newPara.results && newPara.results.msg) { - newPara.results.msg.forEach((newResult, idx) => { - const oldResult = - oldPara.results && oldPara.results.msg ? oldPara.results.msg[idx] : new ParagraphIResultsMsgItem(); - const newConfig = newPara.config.results ? newPara.config.results[idx] : { graph: new GraphConfig() }; - const oldConfig = oldPara.config.results ? oldPara.config.results[idx] : { graph: new GraphConfig() }; - if (!isEqual(newResult, oldResult) || !isEqual(newConfig, oldConfig)) { - this.updateParagraphResult(idx, newConfig, newResult); - } - }); - } - this.cdr.markForCheck(); - }); + this.updateParagraph( + oldPara, + newPara, + () => { + if (newPara.results && newPara.results.msg) { + newPara.results.msg.forEach((newResult, idx) => { + const oldResult = + oldPara.results && oldPara.results.msg ? oldPara.results.msg[idx] : new ParagraphIResultsMsgItem(); + const newConfig = newPara.config.results ? newPara.config.results[idx] : { graph: new GraphConfig() }; + const oldConfig = oldPara.config.results ? oldPara.config.results[idx] : { graph: new GraphConfig() }; + if (!isEqual(newResult, oldResult) || !isEqual(newConfig, oldConfig)) { + this.updateParagraphResult(idx, newConfig, newResult); + } + }); + } + this.cdr.markForCheck(); + }, + data.msgId + ); this.cdr.markForCheck(); + } else { + this.consumeParagraphSave(data.msgId, newPara.text); } } @@ -257,7 +266,7 @@ export abstract class ParagraphBase extends MessageListenersManager { this.cdr.markForCheck(); } - updateParagraph(oldPara: ParagraphItem, newPara: ParagraphItem, updateCallback: () => void) { + updateParagraph(oldPara: ParagraphItem, newPara: ParagraphItem, updateCallback: () => void, msgId?: string) { // 1. can't update on revision view if (!this.revisionView) { // 2. get status, refreshed @@ -269,7 +278,7 @@ export abstract class ParagraphBase extends MessageListenersManager { (newPara.status === ParagraphStatus.FINISHED && statusChanged); // 3. update texts managed by paragraph - this.updateAllScopeTexts(oldPara, newPara); + this.updateAllScopeTexts(oldPara, newPara, msgId); // 4. execute callback to update result updateCallback(); @@ -306,21 +315,63 @@ export abstract class ParagraphBase extends MessageListenersManager { ); } - updateAllScopeTexts(oldPara: ParagraphItem, newPara: ParagraphItem) { + protected trackParagraphSave(msgId: string): void { + this.pendingParagraphSaves.set(msgId, { sequence: ++this.paragraphSaveSequence, originalText: this.originalText }); + } + + private consumeParagraphSave(msgId: string | undefined, savedText: string): boolean { + if (!msgId) { + return false; + } + const pendingSave = this.pendingParagraphSaves.get(msgId); + if (pendingSave === undefined) { + return false; + } + const hasNewerPendingSave = Array.from(this.pendingParagraphSaves.values()).some( + ({ sequence }) => sequence > pendingSave.sequence + ); + for (const [pendingMsgId, { sequence }] of this.pendingParagraphSaves) { + if (sequence <= pendingSave.sequence) { + this.pendingParagraphSaves.delete(pendingMsgId); + } + } + // Skip when a newer save or a patch has already moved the saved baseline past this acknowledgement. + if (!hasNewerPendingSave && this.originalText === pendingSave.originalText) { + this.originalText = savedText; + } + return true; + } + + updateAllScopeTexts(oldPara: ParagraphItem, newPara: ParagraphItem, msgId?: string) { if (!this.paragraph) { throw new Error('paragraph is not defined'); } + const acknowledged = this.consumeParagraphSave(msgId, newPara.text); + if (oldPara.text !== newPara.text) { - if (this.dirtyText) { + // Keep the editor text as is: local edits or collaborative patches can follow the acknowledged save. + if (acknowledged) { + if (this.dirtyText === newPara.text) { + this.dirtyText = undefined; + } + this.cdr.markForCheck(); + return; + } + if (this.dirtyText !== undefined) { // check if editor has local update if (this.dirtyText === newPara.text) { // when local update is the same from remote, clear local update this.paragraph.text = newPara.text; this.dirtyText = undefined; this.originalText = newPara.text; + } else if (this.originalText === newPara.text) { + // An earlier save response must not replace an edit made while it was in flight. } else { - // if there're local update, keep it. + // A different server value is a remote edit, not an acknowledgement of the + // last local save. Accept it and discard the superseded local edit. this.paragraph.text = newPara.text; + this.dirtyText = undefined; + this.originalText = newPara.text; } } else { this.paragraph.text = newPara.text; diff --git a/zeppelin-web-angular/src/app/pages/workspace/notebook/paragraph/paragraph.component.ts b/zeppelin-web-angular/src/app/pages/workspace/notebook/paragraph/paragraph.component.ts index c7365dd3c9b3..a1d0e1e4c334 100644 --- a/zeppelin-web-angular/src/app/pages/workspace/notebook/paragraph/paragraph.component.ts +++ b/zeppelin-web-angular/src/app/pages/workspace/notebook/paragraph/paragraph.component.ts @@ -522,7 +522,8 @@ export class NotebookParagraphComponent config, settings: { params } } = this.paragraph; - this.messageService.commitParagraph(id, title, text, config, params, this.note.id); + const msgId = this.messageService.commitParagraph(id, title, text, config, params, this.note.id); + this.trackParagraphSave(msgId); this.cdr.markForCheck(); } diff --git a/zeppelin-web-angular/src/app/services/message.service.ts b/zeppelin-web-angular/src/app/services/message.service.ts index 050d78c44ccf..c6da4edf011e 100644 --- a/zeppelin-web-angular/src/app/services/message.service.ts +++ b/zeppelin-web-angular/src/app/services/message.service.ts @@ -52,8 +52,12 @@ export class MessageService extends Message implements OnDestroy { interceptReceived(data: WebSocketMessage): WebSocketMessage { const received = this.messageInterceptor ? this.messageInterceptor.received(data) : super.interceptReceived(data); - if (received.op === OP.PARAGRAPH_ADDED && received.data && received.msgId) { - (received.data as MessageReceiveDataTypeMap[OP.PARAGRAPH_ADDED]).msgId = received.msgId; + if (received.data && received.msgId) { + if (received.op === OP.PARAGRAPH_ADDED) { + (received.data as MessageReceiveDataTypeMap[OP.PARAGRAPH_ADDED]).msgId = received.msgId; + } else if (received.op === OP.PARAGRAPH) { + (received.data as MessageReceiveDataTypeMap[OP.PARAGRAPH]).msgId = received.msgId; + } } return received; } @@ -78,8 +82,8 @@ export class MessageService extends Message implements OnDestroy { return super.received(); } - send(...args: SendArgumentsType): void { - super.send(...args); + send(...args: SendArgumentsType): string { + return super.send(...args); } receive(op: K): Observable[K]> { @@ -304,8 +308,8 @@ export class MessageService extends Message implements OnDestroy { paragraphConfig: ParagraphConfig, paragraphParams: ParagraphConfig, noteId: string - ): void { - super.commitParagraph(paragraphId, paragraphTitle, paragraphData, paragraphConfig, paragraphParams, noteId); + ): string { + return super.commitParagraph(paragraphId, paragraphTitle, paragraphData, paragraphConfig, paragraphParams, noteId); } patchParagraph(paragraphId: string, noteId: string, patch: string): void { From f350173e00066afe3bb772c5a3c6f35d9bfb3f90 Mon Sep 17 00:00:00 2001 From: YONGJAE LEE Date: Sun, 20 Sep 2026 10:28:39 +0900 Subject: [PATCH 2/2] [ZEPPELIN-6661] Apply an acknowledged save when no newer edit follows it --- .../paragraph-base/paragraph-base.spec.ts | 19 +++++++++---------- .../app/core/paragraph-base/paragraph-base.ts | 9 ++++----- 2 files changed, 13 insertions(+), 15 deletions(-) diff --git a/zeppelin-web-angular/src/app/core/paragraph-base/paragraph-base.spec.ts b/zeppelin-web-angular/src/app/core/paragraph-base/paragraph-base.spec.ts index 9574649a75fe..15144eb821e4 100644 --- a/zeppelin-web-angular/src/app/core/paragraph-base/paragraph-base.spec.ts +++ b/zeppelin-web-angular/src/app/core/paragraph-base/paragraph-base.spec.ts @@ -372,18 +372,16 @@ describe('ParagraphBase save responses', () => { expect(paragraph.paragraph?.text).toBe('patched edit'); }); - it('keeps a collaborative patch applied after a settings commit when its acknowledgement arrives', () => { - const paragraph = createSaveTestParagraph('committed edit'); - paragraph.originalText = 'committed edit'; + it('applies the server text when a remote edit lands between a save and its acknowledgement', () => { + const paragraph = createSaveTestParagraph('local save'); paragraph.trackSave('save-a'); - paragraph.paragraph!.text = 'patched edit'; - paragraph.originalText = 'patched edit'; + paragraph.originalText = 'local save'; - paragraph.updateAllScopeTexts(paragraph.paragraph!, { text: 'committed edit' } as ParagraphItem, 'save-a'); + paragraph.updateAllScopeTexts(paragraph.paragraph!, { text: 'remote edit' } as ParagraphItem); + paragraph.updateAllScopeTexts(paragraph.paragraph!, { text: 'local save' } as ParagraphItem, 'save-a'); - expect(paragraph.paragraph?.text).toBe('patched edit'); - expect(paragraph.originalText).toBe('patched edit'); - expect(paragraph.dirtyText).toBeUndefined(); + expect(paragraph.paragraph?.text).toBe('local save'); + expect(paragraph.originalText).toBe('local save'); }); it('clears the local edit once its acknowledgement arrives', () => { @@ -394,7 +392,8 @@ describe('ParagraphBase save responses', () => { paragraph.updateAllScopeTexts(paragraph.paragraph!, { text: 'dirty edit' } as ParagraphItem, 'save-a'); - expect(paragraph.paragraph?.text).toBe('saved edit'); + expect(paragraph.paragraph?.text).toBe('dirty edit'); + expect(paragraph.originalText).toBe('dirty edit'); expect(paragraph.dirtyText).toBeUndefined(); }); diff --git a/zeppelin-web-angular/src/app/core/paragraph-base/paragraph-base.ts b/zeppelin-web-angular/src/app/core/paragraph-base/paragraph-base.ts index f2e49ea4207b..927e85bf42bc 100644 --- a/zeppelin-web-angular/src/app/core/paragraph-base/paragraph-base.ts +++ b/zeppelin-web-angular/src/app/core/paragraph-base/paragraph-base.ts @@ -349,11 +349,10 @@ export abstract class ParagraphBase extends MessageListenersManager { const acknowledged = this.consumeParagraphSave(msgId, newPara.text); if (oldPara.text !== newPara.text) { - // Keep the editor text as is: local edits or collaborative patches can follow the acknowledged save. - if (acknowledged) { - if (this.dirtyText === newPara.text) { - this.dirtyText = undefined; - } + // Keep an edit typed or saved after this save; otherwise the response is the server copy. + // Saves still pending after consumeParagraphSave are newer than the acknowledged one. + const hasLocalEdit = this.dirtyText !== undefined && this.dirtyText !== newPara.text; + if (acknowledged && (hasLocalEdit || this.pendingParagraphSaves.size > 0)) { this.cdr.markForCheck(); return; }