From 74912627c5a2a57fdda3d7e898e6a7fa291739d2 Mon Sep 17 00:00:00 2001 From: YONGJAE LEE Date: Thu, 17 Sep 2026 04:18:05 +0900 Subject: [PATCH 1/4] [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 | 8 +- 11 files changed, 659 insertions(+), 41 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 0875d18c967..6d026e679b8 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 00000000000..9ae1e0c27b7 --- /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 7da8c388176..03ea679971b 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 e90d35f5b8e..60478f20618 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 00000000000..22394d72915 --- /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 e4058a3f12a..c5c7ec61ae0 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 @@ -150,6 +150,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 f9dff2e804a..8c51a5ef066 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]> { @@ -453,7 +454,7 @@ export class Message { paragraphConfig: ParagraphConfig, paragraphParams: ParagraphConfig, noteId: string - ): void { + ): string { return this.send(OP.COMMIT_PARAGRAPH, { id: paragraphId, noteId, @@ -468,7 +469,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 9794f25b07b..3e42c17ff02 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'; @@ -246,3 +246,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 b9e514abd6e..f2e49ea4207 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 c7365dd3c9b..a1d0e1e4c33 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 937754d400a..2d642f65a8d 100644 --- a/zeppelin-web-angular/src/app/services/message.service.ts +++ b/zeppelin-web-angular/src/app/services/message.service.ts @@ -74,8 +74,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]> { @@ -300,8 +300,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 785c7090d4678079bc059c5ff86eb7b42dbcc018 Mon Sep 17 00:00:00 2001 From: YONGJAE LEE Date: Sun, 20 Sep 2026 10:28:39 +0900 Subject: [PATCH 2/4] [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 3e42c17ff02..e0e8b9aa1c4 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 @@ -373,18 +373,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', () => { @@ -395,7 +393,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 f2e49ea4207..927e85bf42b 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; } From 79020d91736b04e7216ca7eea2b354305598653a Mon Sep 17 00:00:00 2001 From: YONGJAE LEE Date: Fri, 25 Sep 2026 23:01:17 +0900 Subject: [PATCH 3/4] Read the save acknowledgement msgId from the message envelope --- .../e2e/scenarios/notebook-parity.json | 4 ++ .../e2e/scenarios/notebook-parity.md | 2 +- .../interfaces/message-notebook.interface.ts | 1 - .../paragraph-base/paragraph-base.spec.ts | 67 +++++++++++++----- .../app/core/paragraph-base/paragraph-base.ts | 70 ++++++++----------- .../paragraph/paragraph.component.spec.ts | 2 +- 6 files changed, 83 insertions(+), 63 deletions(-) diff --git a/zeppelin-web-angular/e2e/scenarios/notebook-parity.json b/zeppelin-web-angular/e2e/scenarios/notebook-parity.json index 03ea679971b..c9774f2ac1f 100644 --- a/zeppelin-web-angular/e2e/scenarios/notebook-parity.json +++ b/zeppelin-web-angular/e2e/scenarios/notebook-parity.json @@ -592,6 +592,10 @@ { "path": "zeppelin-web-angular/src/app/pages/workspace/notebook/notebook.component.ts", "symbol": "NotebookComponent" + }, + { + "path": "zeppelin-web-angular/src/app/core/paragraph-base/paragraph-base.ts", + "symbol": "ParagraphBase" } ], "verificationEvidence": [ diff --git a/zeppelin-web-angular/e2e/scenarios/notebook-parity.md b/zeppelin-web-angular/e2e/scenarios/notebook-parity.md index 60478f20618..fecb5a03d49 100644 --- a/zeppelin-web-angular/e2e/scenarios/notebook-parity.md +++ b/zeppelin-web-angular/e2e/scenarios/notebook-parity.md @@ -169,7 +169,7 @@ Coverage note: `covered` mechanically means this registry points to a matching e - 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) +- 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); zeppelin-web-angular/src/app/core/paragraph-base/paragraph-base.ts (ParagraphBase) - 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/projects/zeppelin-sdk/src/interfaces/message-notebook.interface.ts b/zeppelin-web-angular/projects/zeppelin-sdk/src/interfaces/message-notebook.interface.ts index c5c7ec61ae0..e4058a3f12a 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 @@ -150,7 +150,6 @@ export interface ParagraphMoved { } export interface UpdateParagraph { - msgId?: string; paragraph: ParagraphItem; } 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 e0e8b9aa1c4..5f894dcf409 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 @@ -11,7 +11,7 @@ */ import { ChangeDetectorRef } from '@angular/core'; -import { DatasetType, Message, ParagraphItem } from '@zeppelin/sdk'; +import { DatasetType, Message, OP, ParagraphItem } from '@zeppelin/sdk'; import { EMPTY } from 'rxjs'; import { afterEach, describe, expect, it, vi } from 'vitest'; @@ -29,7 +29,7 @@ class TestParagraph extends ParagraphBase { constructor(paragraph: ParagraphItem) { super( - { receive: () => EMPTY } as unknown as Message, + { receive: () => EMPTY, receiveEnvelope: () => EMPTY } as unknown as Message, { isParagraphRunning: item => item.status === 'RUNNING', isEntireNoteRunning: () => false }, {} as AngularContextManager, { markForCheck: vi.fn() } as unknown as ChangeDetectorRef @@ -94,7 +94,10 @@ describe('ParagraphBase streaming state isolation', () => { const component = new TestParagraph(paragraph('A')); beginOutput(component); - component.paragraphData({ paragraph: paragraph('B', status, '2026-01-01T00:00:01Z') }); + component.paragraphData({ + op: OP.PARAGRAPH, + data: { paragraph: paragraph('B', status, '2026-01-01T00:00:01Z') } + }); appendOutput(component); expect(component.results).toEqual([{ type: DatasetType.TEXT, data: 'first\nsecond\n' }]); @@ -108,7 +111,7 @@ describe('ParagraphBase streaming state isolation', () => { beginOutput(component); const finished = { ...paragraph('A', 'FINISHED'), results: { msg: [{ type: DatasetType.TEXT, data: 'final\n' }] } }; - component.paragraphData({ paragraph: finished }); + component.paragraphData({ op: OP.PARAGRAPH, data: { paragraph: finished } }); appendOutput(component); expect(component.results).toEqual([{ type: DatasetType.TEXT, data: 'final\n' }]); @@ -118,8 +121,11 @@ describe('ParagraphBase streaming state isolation', () => { it('accepts fresh output when its own paragraph starts a new run', () => { const component = new TestParagraph(paragraph('A')); beginOutput(component); - component.paragraphData({ paragraph: paragraph('A', 'FINISHED') }); - component.paragraphData({ paragraph: paragraph('A', 'RUNNING', '2026-01-01T00:00:01Z') }); + component.paragraphData({ op: OP.PARAGRAPH, data: { paragraph: paragraph('A', 'FINISHED') } }); + component.paragraphData({ + op: OP.PARAGRAPH, + data: { paragraph: paragraph('A', 'RUNNING', '2026-01-01T00:00:01Z') } + }); beginOutput(component); appendOutput(component); @@ -137,10 +143,13 @@ describe('ParagraphBase streaming boundaries', () => { component.results = saved; component.revisionView = true; - component.paragraphData({ paragraph: paragraph('A', 'RUNNING', '2026-01-01T00:00:01Z') }); + component.paragraphData({ + op: OP.PARAGRAPH, + data: { paragraph: paragraph('A', 'RUNNING', '2026-01-01T00:00:01Z') } + }); beginOutput(component); appendOutput(component); - component.paragraphData({ paragraph: paragraph('A') }); + component.paragraphData({ op: OP.PARAGRAPH, data: { paragraph: paragraph('A') } }); expect(component.results).toEqual(saved); expect(component.paragraph?.status).toBe('FINISHED'); @@ -156,12 +165,15 @@ describe('ParagraphBase streaming boundaries', () => { const statusOnly = paragraph('A'); delete statusOnly.dateStarted; - component.paragraphData({ paragraph: statusOnly }); - component.paragraphData({ paragraph: statusOnly }); + component.paragraphData({ op: OP.PARAGRAPH, data: { paragraph: statusOnly } }); + component.paragraphData({ op: OP.PARAGRAPH, data: { paragraph: statusOnly } }); expect(update).not.toHaveBeenCalled(); expect(component.paragraph?.dateStarted).toBe('2026-01-01T00:00:00Z'); - component.paragraphData({ paragraph: paragraph('A', 'RUNNING', '2026-01-01T00:00:01Z') }); + component.paragraphData({ + op: OP.PARAGRAPH, + data: { paragraph: paragraph('A', 'RUNNING', '2026-01-01T00:00:01Z') } + }); expect(update).toHaveBeenCalledOnce(); expect(component.paragraph?.dateStarted).toBe('2026-01-01T00:00:01Z'); component.ngOnDestroy(); @@ -175,7 +187,7 @@ describe('ParagraphBase streaming boundaries', () => { delete finished.dateStarted; finished.results = { msg: [{ type: DatasetType.TEXT, data: 'first\nfinal\n' }] }; - component.paragraphData({ paragraph: finished }); + component.paragraphData({ op: OP.PARAGRAPH, data: { paragraph: finished } }); expect(component.results).toEqual([{ type: DatasetType.TEXT, data: 'first\nfinal\n' }]); expect(component.updateParagraphResult).toHaveBeenCalledWith(0, expect.anything(), { @@ -190,8 +202,8 @@ describe('ParagraphBase streaming boundaries', () => { beginOutput(component); const running = paragraph('A'); delete running.dateStarted; - component.paragraphData({ paragraph: running }); - component.paragraphData({ paragraph: paragraph('A') }); + component.paragraphData({ op: OP.PARAGRAPH, data: { paragraph: running } }); + component.paragraphData({ op: OP.PARAGRAPH, data: { paragraph: paragraph('A') } }); appendOutput(component); expect(component.results).toEqual([{ type: DatasetType.TEXT, data: 'first\nsecond\n' }]); @@ -202,8 +214,8 @@ describe('ParagraphBase streaming boundaries', () => { it('preserves streaming output across ordinary RUNNING and unrelated paragraph snapshots', () => { const component = new TestParagraph(paragraph('A')); beginOutput(component); - component.paragraphData({ paragraph: paragraph('A') }); - component.paragraphData({ paragraph: paragraph('B') }); + component.paragraphData({ op: OP.PARAGRAPH, data: { paragraph: paragraph('A') } }); + component.paragraphData({ op: OP.PARAGRAPH, data: { paragraph: paragraph('B') } }); appendOutput(component); expect(component.results).toEqual([{ type: DatasetType.TEXT, data: 'first\nsecond\n' }]); component.ngOnDestroy(); @@ -268,7 +280,7 @@ afterEach(() => { const createSaveTestParagraph = (text: string, dirtyText?: string): SaveTestParagraph => { const paragraph = new SaveTestParagraph( - { receive: () => EMPTY } as unknown as Message, + { receive: () => EMPTY, receiveEnvelope: () => EMPTY } as unknown as Message, { isParagraphRunning: () => false, isEntireNoteRunning: () => false }, { setContextValue: vi.fn(), @@ -403,14 +415,31 @@ describe('ParagraphBase save responses', () => { component.paragraph = { ...paragraph('A', 'FINISHED'), text: 'saved edit' }; component.trackSave('save-a'); - component.paragraphData({ paragraph: { ...component.paragraph }, msgId: 'save-a' }); + component.paragraphData({ op: OP.PARAGRAPH, data: { paragraph: { ...component.paragraph } }, msgId: 'save-a' }); expect(component.originalText).toBe('saved edit'); component.originalText = 'previous save'; - component.paragraphData({ paragraph: { ...component.paragraph }, msgId: 'save-a' }); + component.paragraphData({ op: OP.PARAGRAPH, data: { paragraph: { ...component.paragraph } }, msgId: 'save-a' }); expect(component.originalText).toBe('previous save'); }); + it.each([ + ['its own save acknowledgement', 'save-a', 'latest edit'], + ['a broadcast without msgId', undefined, 'committed edit'] + ])('reads the msgId from the envelope when paragraphData delivers %s', (_case, msgId, expectedText) => { + const component = createSaveTestParagraph('latest edit', 'latest edit'); + component.paragraph = { ...paragraph('A', 'FINISHED'), text: 'latest edit' }; + component.trackSave('save-a'); + + component.paragraphData({ + op: OP.PARAGRAPH, + data: { paragraph: { ...component.paragraph, text: 'committed edit' } }, + msgId + }); + + expect(component.paragraph.text).toBe(expectedText); + }); + it('accepts a remote edit that is not an acknowledgement of the last local save', () => { const paragraph = createSaveTestParagraph('local edit', 'local edit'); 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 927e85bf42b..efdab3dc24b 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 @@ -23,13 +23,18 @@ import { ParagraphConfigResults, ParagraphEditorSetting, ParagraphItem, - ParagraphIResultsMsgItem + ParagraphIResultsMsgItem, + WebSocketMessage } from '@zeppelin/sdk'; import { diff_match_patch as DiffMatchPatch } from 'diff-match-patch'; import { isEmpty, isEqual } from 'lodash'; -import { MessageListener, MessageListenersManager } from '../message-listener/message-listener'; +import { + MessageEnvelopeListener, + MessageListener, + MessageListenersManager +} from '../message-listener/message-listener'; import { AngularContextManager } from './angular-context-manager'; import { NoteStatus } from './note-status'; import { ParagraphOutputState } from './paragraph-output-state'; @@ -147,13 +152,13 @@ export abstract class ParagraphBase extends MessageListenersManager { } } - @MessageListener(OP.PARAGRAPH) - paragraphData(data: MessageReceiveDataTypeMap[OP.PARAGRAPH]) { + @MessageEnvelopeListener(OP.PARAGRAPH) + paragraphData(message: WebSocketMessage) { const oldPara = this.paragraph; - if (!oldPara) { + if (!oldPara || message.data === undefined) { return; } - const newPara = data.paragraph; + const newPara = message.data.paragraph; if (this.revisionView || newPara.id !== oldPara.id) { return; } @@ -189,11 +194,11 @@ export abstract class ParagraphBase extends MessageListenersManager { } this.cdr.markForCheck(); }, - data.msgId + message.msgId ); this.cdr.markForCheck(); } else { - this.consumeParagraphSave(data.msgId, newPara.text); + this.consumeParagraphSave(message.msgId, newPara.text); } } @@ -319,61 +324,44 @@ export abstract class ParagraphBase extends MessageListenersManager { 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); + private consumeParagraphSave( + msgId: string | undefined, + savedText: string + ): { acknowledged: boolean; hasNewerSave: boolean } { + const pendingSave = msgId === undefined ? undefined : this.pendingParagraphSaves.get(msgId); if (pendingSave === undefined) { - return false; + return { acknowledged: false, hasNewerSave: false }; } - const hasNewerPendingSave = Array.from(this.pendingParagraphSaves.values()).some( + const hasNewerSave = Array.from(this.pendingParagraphSaves.values()).some( ({ sequence }) => sequence > pendingSave.sequence ); + // The server answers COMMIT_PARAGRAPH in send order on one socket, + // so an older save is never acknowledged after a newer one. 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) { + if (!hasNewerSave && this.originalText === pendingSave.originalText) { this.originalText = savedText; } - return true; + return { acknowledged: true, hasNewerSave }; } updateAllScopeTexts(oldPara: ParagraphItem, newPara: ParagraphItem, msgId?: string) { if (!this.paragraph) { throw new Error('paragraph is not defined'); } - const acknowledged = this.consumeParagraphSave(msgId, newPara.text); + const { acknowledged, hasNewerSave } = this.consumeParagraphSave(msgId, newPara.text); if (oldPara.text !== newPara.text) { - // 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; - } - 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 { - // 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 { + const staleSaveAcknowledgement = acknowledged && (hasLocalEdit || hasNewerSave); + const staleBroadcastOfSavedText = hasLocalEdit && this.originalText === newPara.text; + if (!staleSaveAcknowledgement && !staleBroadcastOfSavedText) { this.paragraph.text = newPara.text; + this.dirtyText = undefined; this.originalText = newPara.text; } } diff --git a/zeppelin-web-angular/src/app/pages/workspace/published/paragraph/paragraph.component.spec.ts b/zeppelin-web-angular/src/app/pages/workspace/published/paragraph/paragraph.component.spec.ts index a9c9b75c407..731519047ea 100644 --- a/zeppelin-web-angular/src/app/pages/workspace/published/paragraph/paragraph.component.spec.ts +++ b/zeppelin-web-angular/src/app/pages/workspace/published/paragraph/paragraph.component.spec.ts @@ -55,7 +55,7 @@ afterEach(() => components.splice(0).forEach(component => component.ngOnDestroy( const setup = (useReact = false) => { const params = new BehaviorSubject({ noteId: 'note-A', paragraphId: 'same-paragraph' }); const component = new PublishedParagraphComponent( - { receive: () => EMPTY, getNote: vi.fn() } as never, + { receive: () => EMPTY, receiveEnvelope: () => EMPTY, getNote: vi.fn() } as never, { params, queryParamMap: of(convertToParamMap({})) } as never, {} as never, {} as never, From 16c388d11810c80d9899caf7b475e59aabda0888 Mon Sep 17 00:00:00 2001 From: YONGJAE LEE Date: Fri, 25 Sep 2026 23:01:18 +0900 Subject: [PATCH 4/4] Track local paragraph add msgIds from the send return value --- .../projects/zeppelin-sdk/src/message.spec.ts | 20 +++++++++ .../projects/zeppelin-sdk/src/message.ts | 8 ++-- .../src/app/services/message.service.spec.ts | 45 +++++++++++++++++++ .../src/app/services/message.service.ts | 32 ++++--------- 4 files changed, 77 insertions(+), 28 deletions(-) create mode 100644 zeppelin-web-angular/src/app/services/message.service.spec.ts diff --git a/zeppelin-web-angular/projects/zeppelin-sdk/src/message.spec.ts b/zeppelin-web-angular/projects/zeppelin-sdk/src/message.spec.ts index 37f8d44c2bf..78db844c4dc 100644 --- a/zeppelin-web-angular/projects/zeppelin-sdk/src/message.spec.ts +++ b/zeppelin-web-angular/projects/zeppelin-sdk/src/message.spec.ts @@ -162,3 +162,23 @@ describe('Message.receive', () => { expect(listener).toHaveBeenCalledWith(envelope); }); }); + +describe('Message.send', () => { + const connectedMessage = () => { + const message = new Message(); + const next = vi.fn(); + (message as unknown as { ws: { next: typeof next } }).ws = { next }; + return { message, next }; + }; + + it('returns the msgId of each sent message', () => { + const { message, next } = connectedMessage(); + + const commitMsgId = message.commitParagraph('p1', undefined, 'text', {}, {}, 'n1'); + const insertMsgId = message.insertParagraph(0); + const copyMsgId = message.copyParagraph(1, undefined, 'text', {}, {}); + + expect(next.mock.calls.map(([sent]) => sent.msgId)).toEqual([commitMsgId, insertMsgId, copyMsgId]); + expect(new Set([commitMsgId, insertMsgId, copyMsgId]).size).toBe(3); + }); +}); diff --git a/zeppelin-web-angular/projects/zeppelin-sdk/src/message.ts b/zeppelin-web-angular/projects/zeppelin-sdk/src/message.ts index 8c51a5ef066..6950877b807 100644 --- a/zeppelin-web-angular/projects/zeppelin-sdk/src/message.ts +++ b/zeppelin-web-angular/projects/zeppelin-sdk/src/message.ts @@ -300,8 +300,8 @@ export class Message { this.send(OP.MOVE_PARAGRAPH, { id: paragraphId, index: newIndex }); } - insertParagraph(newIndex: number): void { - this.send(OP.INSERT_PARAGRAPH, { index: newIndex }); + insertParagraph(newIndex: number): string { + return this.send(OP.INSERT_PARAGRAPH, { index: newIndex }); } copyParagraph( @@ -310,8 +310,8 @@ export class Message { paragraphData: string, paragraphConfig: ParagraphConfig, paragraphParams: ParagraphParams - ): void { - this.send(OP.COPY_PARAGRAPH, { + ): string { + return this.send(OP.COPY_PARAGRAPH, { index: newIndex, title: paragraphTitle, paragraph: paragraphData, diff --git a/zeppelin-web-angular/src/app/services/message.service.spec.ts b/zeppelin-web-angular/src/app/services/message.service.spec.ts new file mode 100644 index 00000000000..e1269c32e5f --- /dev/null +++ b/zeppelin-web-angular/src/app/services/message.service.spec.ts @@ -0,0 +1,45 @@ +/* + * 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 { describe, expect, it, vi } from 'vitest'; + +import { MessageInterceptor } from '@zeppelin/interfaces'; +import { BaseUrlService } from './base-url.service'; +import { MessageService } from './message.service'; +import { TicketService } from './ticket.service'; + +const connectedMessageService = (): MessageService => { + const service = new MessageService({} as BaseUrlService, {} as TicketService, null as unknown as MessageInterceptor); + (service as unknown as { ws: { next: () => void } }).ws = { next: vi.fn() }; + return service; +}; + +describe('MessageService local add focus', () => { + it.each([ + ['insertParagraph', (service: MessageService) => service.insertParagraph(0)], + ['copyParagraph', (service: MessageService) => service.copyParagraph(1, undefined, 'text', {}, {})] + ])('records the msgId returned by %s once', (_method, add) => { + const service = connectedMessageService(); + + const msgId = add(service); + + expect(service.consumeLocalAddFocusMsgId(msgId)).toBe(true); + expect(service.consumeLocalAddFocusMsgId(msgId)).toBe(false); + }); + + it('does not record a msgId for a paragraph added by another client', () => { + const service = connectedMessageService(); + service.insertParagraph(0); + + expect(service.consumeLocalAddFocusMsgId('other-client-1')).toBe(false); + }); +}); diff --git a/zeppelin-web-angular/src/app/services/message.service.ts b/zeppelin-web-angular/src/app/services/message.service.ts index 2d642f65a8d..1b5746482da 100644 --- a/zeppelin-web-angular/src/app/services/message.service.ts +++ b/zeppelin-web-angular/src/app/services/message.service.ts @@ -12,7 +12,6 @@ import { Inject, Injectable, OnDestroy, Optional } from '@angular/core'; import { Observable } from 'rxjs'; -import { take } from 'rxjs/operators'; import { MessageInterceptor, MESSAGE_INTERCEPTOR } from '@zeppelin/interfaces'; import { @@ -89,23 +88,6 @@ export class MessageService extends Message implements OnDestroy { return this.localAddFocusMsgIds.delete(msgId); } - private captureLocalAddFocusMsgId(sendMessage: () => void): void { - const subscription = super - .sent() - .pipe(take(1)) - .subscribe(message => { - if (message.msgId) { - this.localAddFocusMsgIds.add(message.msgId); - } - }); - try { - sendMessage(); - } catch (error) { - subscription.unsubscribe(); - throw error; - } - } - opened(): Observable { return super.opened(); } @@ -194,8 +176,10 @@ export class MessageService extends Message implements OnDestroy { super.moveParagraph(paragraphId, newIndex); } - insertParagraph(newIndex: number): void { - this.captureLocalAddFocusMsgId(() => super.insertParagraph(newIndex)); + insertParagraph(newIndex: number): string { + const msgId = super.insertParagraph(newIndex); + this.localAddFocusMsgIds.add(msgId); + return msgId; } copyParagraph( @@ -204,10 +188,10 @@ export class MessageService extends Message implements OnDestroy { paragraphData: string, paragraphConfig: ParagraphConfig, paragraphParams: ParagraphParams - ): void { - this.captureLocalAddFocusMsgId(() => - super.copyParagraph(newIndex, paragraphTitle, paragraphData, paragraphConfig, paragraphParams) - ); + ): string { + const msgId = super.copyParagraph(newIndex, paragraphTitle, paragraphData, paragraphConfig, paragraphParams); + this.localAddFocusMsgIds.add(msgId); + return msgId; } angularObjectUpdate(