From a104d999fa1d4faacc665b39fccd1aca005f6377 Mon Sep 17 00:00:00 2001 From: xhaktm00 <153787023+xhaktm00@users.noreply.github.com> Date: Fri, 7 Aug 2026 11:24:52 +0900 Subject: [PATCH] [ZEPPELIN-6563] Detect diverged paragraph text in collaborative mode patch_apply never throws: it drops a hunk whose context is gone, and fuzzy matching can place one at the wrong offset while still reporting success, so neither shape is visible to the receiver. A diverged client then commits its whole paragraph and the server stores it unconditionally, making that text authoritative for everyone. Senders now attach checksums of their text before and after the patch. A receiver that started from the same text but did not reach the same result knows it diverged and asks for that single paragraph again via GET_PARAGRAPH, instead of refetching the whole note. Commits carry the checksum of the text the client believed the server held, and the server keeps its own copy when they disagree. Both fields are optional, so clients that omit them behave as before. PatchParagraphSend and PatchParagraphReceived were mapped to the opposite directions, which left the receiving side without noteId or the new fields; they are swapped to match what is actually sent and received. --- .../org/apache/zeppelin/common/Message.java | 1 + .../zeppelin/service/NotebookService.java | 54 +++++++++++++++++++ .../zeppelin/socket/NotebookServer.java | 42 ++++++++++++++- .../zeppelin/service/NotebookServiceTest.java | 35 ++++++++++++ .../projects/zeppelin-sdk/src/checksum.ts | 23 ++++++++ .../message-data-type-map.interface.ts | 6 ++- .../interfaces/message-operator.interface.ts | 1 + .../interfaces/message-paragraph.interface.ts | 18 ++++++- .../projects/zeppelin-sdk/src/message.ts | 25 +++++++-- .../projects/zeppelin-sdk/src/public-api.ts | 1 + .../app/core/paragraph-base/paragraph-base.ts | 11 +++- .../notebook/paragraph/paragraph.component.ts | 12 +++-- .../src/app/services/message.service.ts | 27 ++++++++-- 13 files changed, 238 insertions(+), 18 deletions(-) create mode 100644 zeppelin-web-angular/projects/zeppelin-sdk/src/checksum.ts diff --git a/zeppelin-common/src/main/java/org/apache/zeppelin/common/Message.java b/zeppelin-common/src/main/java/org/apache/zeppelin/common/Message.java index 6ec66e63dbd..ca0222563d0 100644 --- a/zeppelin-common/src/main/java/org/apache/zeppelin/common/Message.java +++ b/zeppelin-common/src/main/java/org/apache/zeppelin/common/Message.java @@ -201,6 +201,7 @@ public enum OP { INTERPRETER_INSTALL_RESULT, // [s-c] Status of an interpreter installation COLLABORATIVE_MODE_STATUS, // [s-c] collaborative mode status PATCH_PARAGRAPH, // [c-s][s-c] patch editor text + GET_PARAGRAPH, // [c-s] resend a single paragraph after a failed patch NOTE_RUNNING_STATUS, // [s-c] sequential run status will be change NOTICE // [s-c] Notice } diff --git a/zeppelin-server/src/main/java/org/apache/zeppelin/service/NotebookService.java b/zeppelin-server/src/main/java/org/apache/zeppelin/service/NotebookService.java index 38ce280ad6e..26edc67d205 100644 --- a/zeppelin-server/src/main/java/org/apache/zeppelin/service/NotebookService.java +++ b/zeppelin-server/src/main/java/org/apache/zeppelin/service/NotebookService.java @@ -751,6 +751,23 @@ public void updateParagraph(String noteId, Map config, ServiceContext context, ServiceCallback callback) throws IOException { + updateParagraph(noteId, paragraphId, title, text, params, config, null, context, callback); + } + + /** + * @param baseChecksum checksum of the text the client believed the server held, or null to skip + * the check. When it does not match, the client text is stale or diverged, + * so it is not stored and the server copy is sent back instead. + */ + public void updateParagraph(String noteId, + String paragraphId, + String title, + String text, + Map params, + Map config, + Integer baseChecksum, + ServiceContext context, + ServiceCallback callback) throws IOException { if (!checkPermission(noteId, Permission.WRITER, Message.OP.COMMIT_PARAGRAPH, context, callback)) { return; @@ -767,6 +784,11 @@ public void updateParagraph(String noteId, callback.onFailure(new ParagraphNotFoundException(paragraphId), context); return null; } + if (baseChecksum != null && baseChecksum != checksum(p.getText())) { + LOGGER.info("Rejecting stale commit of paragraph {} in note {}", paragraphId, noteId); + callback.onSuccess(p, context); + return null; + } // In personalized mode only the note owner may update the master paragraph, so that // new users inherit the owner's changes while a non-owner's changes stay in their copy. if (!note.isPersonalizedMode() @@ -1529,6 +1551,38 @@ public void patchParagraph(final String noteId, final String paragraphId, String } } + /** + * Resend a single paragraph to a client whose patched text diverged, so that the note as a + * whole does not have to be reloaded. + */ + public void getParagraph(String noteId, String paragraphId, ServiceContext context, + ServiceCallback callback) throws IOException { + if (!checkPermission(noteId, Permission.READER, Message.OP.GET_PARAGRAPH, context, callback)) { + return; + } + notebook.processNote(noteId, + note -> { + if (note == null) { + callback.onFailure(new NoteNotFoundException(noteId), context); + return null; + } + Paragraph p = note.getParagraph(paragraphId); + if (p == null) { + callback.onFailure(new ParagraphNotFoundException(paragraphId), context); + return null; + } + callback.onSuccess(p, context); + return null; + }); + } + + /** + * Same algorithm as {@link String#hashCode()} so that the client can compute it identically. + */ + static int checksum(String text) { + return text == null ? "".hashCode() : text.hashCode(); + } + enum Permission { READER, diff --git a/zeppelin-server/src/main/java/org/apache/zeppelin/socket/NotebookServer.java b/zeppelin-server/src/main/java/org/apache/zeppelin/socket/NotebookServer.java index 2d78ae7fb57..bab433270ce 100644 --- a/zeppelin-server/src/main/java/org/apache/zeppelin/socket/NotebookServer.java +++ b/zeppelin-server/src/main/java/org/apache/zeppelin/socket/NotebookServer.java @@ -479,6 +479,9 @@ public void onMessage(NotebookSocket conn, String msg) { case PATCH_PARAGRAPH: patchParagraph(conn, context, receivedMessage); break; + case GET_PARAGRAPH: + getParagraph(conn, context, receivedMessage); + break; default: break; } @@ -1087,8 +1090,11 @@ private void updateParagraph(NotebookSocket conn, ServiceContext context, Messag String text = (String) fromMessage.get("paragraph"); Map params = (Map) fromMessage.get("params"); Map config = (Map) fromMessage.get("config"); + Integer baseChecksum = fromMessage.get("baseChecksum") == null + ? null : ((Number) fromMessage.get("baseChecksum")).intValue(); - getNotebookService().updateParagraph(noteId, paragraphId, title, text, params, config, context, + getNotebookService().updateParagraph(noteId, paragraphId, title, text, params, config, + baseChecksum, context, new WebSocketServiceCallback(conn) { @Override public void onSuccess(Paragraph p, ServiceContext context) throws IOException { @@ -1126,6 +1132,10 @@ private void patchParagraph(NotebookSocket conn, if (patchText == null) { return; } + // checksums of the sender's text before and after the patch, so receivers can tell whether + // their own patchApply produced the same text. Absent for clients that do not send them. + Object baseChecksum = fromMessage.get("baseChecksum"); + Object afterChecksum = fromMessage.get("afterChecksum"); getNotebookService().patchParagraph(noteId, paragraphId, patchText, context, new WebSocketServiceCallback(conn) { @@ -1134,12 +1144,40 @@ public void onSuccess(String result, ServiceContext context) throws IOException super.onSuccess(result, context); Message message = new Message(OP.PATCH_PARAGRAPH) .put("patch", result) - .put("paragraphId", paragraphId); + .put("paragraphId", paragraphId) + .put("noteId", noteId2) + .put("baseChecksum", baseChecksum) + .put("afterChecksum", afterChecksum); connectionManager.broadcastExcept(noteId2, message, conn); } }); } + private void getParagraph(NotebookSocket conn, + ServiceContext context, + Message fromMessage) throws IOException { + String paragraphId = fromMessage.getType("id", LOGGER); + if (paragraphId == null) { + return; + } + String noteId = connectionManager.getAssociatedNoteId(conn); + if (noteId == null) { + noteId = fromMessage.getType("noteId", LOGGER); + if (noteId == null) { + return; + } + } + + getNotebookService().getParagraph(noteId, paragraphId, context, + new WebSocketServiceCallback(conn) { + @Override + public void onSuccess(Paragraph p, ServiceContext context) throws IOException { + super.onSuccess(p, context); + conn.send(serializeMessage(new Message(OP.PARAGRAPH).put("paragraph", p))); + } + }); + } + private void cloneNote(NotebookSocket conn, ServiceContext context, Message fromMessage) throws IOException { diff --git a/zeppelin-server/src/test/java/org/apache/zeppelin/service/NotebookServiceTest.java b/zeppelin-server/src/test/java/org/apache/zeppelin/service/NotebookServiceTest.java index 2d53f34dec8..4da2ea48247 100644 --- a/zeppelin-server/src/test/java/org/apache/zeppelin/service/NotebookServiceTest.java +++ b/zeppelin-server/src/test/java/org/apache/zeppelin/service/NotebookServiceTest.java @@ -718,4 +718,39 @@ void testNormalizeNotePath() throws IOException { assertEquals("Note name shouldn't end with '/'", e.getMessage()); } } + + @Test + void testUpdateParagraphChecksIsBasedOnCurrentServerText() throws IOException { + String noteId = notebookService.createNote("/note_checksum", "test", true, context, callback); + String paragraphId = notebook.processNote(noteId, note -> { + Paragraph p = note.getParagraph(0); + p.setText("server text"); + return p.getId(); + }); + + // a commit based on the text the server holds is stored + notebookService.updateParagraph(noteId, paragraphId, "title", "agreed text", + new HashMap<>(), new HashMap<>(), "server text".hashCode(), context, callback); + notebook.processNote(noteId, note -> { + assertEquals("agreed text", note.getParagraph(paragraphId).getText()); + return null; + }); + + // a commit based on text the server no longer holds is rejected, keeping the server copy + notebookService.updateParagraph(noteId, paragraphId, "stale title", "diverged text", + new HashMap<>(), new HashMap<>(), "text nobody has".hashCode(), context, callback); + notebook.processNote(noteId, note -> { + assertEquals("agreed text", note.getParagraph(paragraphId).getText()); + assertEquals("title", note.getParagraph(paragraphId).getTitle()); + return null; + }); + + // clients that send no checksum keep the previous behaviour + notebookService.updateParagraph(noteId, paragraphId, "no checksum", "text without checksum", + new HashMap<>(), new HashMap<>(), context, callback); + notebook.processNote(noteId, note -> { + assertEquals("text without checksum", note.getParagraph(paragraphId).getText()); + return null; + }); + } } diff --git a/zeppelin-web-angular/projects/zeppelin-sdk/src/checksum.ts b/zeppelin-web-angular/projects/zeppelin-sdk/src/checksum.ts new file mode 100644 index 00000000000..fa28394647c --- /dev/null +++ b/zeppelin-web-angular/projects/zeppelin-sdk/src/checksum.ts @@ -0,0 +1,23 @@ +/* + * 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. + */ + +/** + * Same algorithm as java.lang.String#hashCode(), so a checksum computed here matches the one + * NotebookService computes for the same text. + */ +export function textChecksum(text: string): number { + let hash = 0; + for (let i = 0; i < text.length; i++) { + hash = (Math.imul(31, hash) + text.charCodeAt(i)) | 0; + } + return hash; +} diff --git a/zeppelin-web-angular/projects/zeppelin-sdk/src/interfaces/message-data-type-map.interface.ts b/zeppelin-web-angular/projects/zeppelin-sdk/src/interfaces/message-data-type-map.interface.ts index 6c6088c73ae..8ceeaf7741e 100644 --- a/zeppelin-web-angular/projects/zeppelin-sdk/src/interfaces/message-data-type-map.interface.ts +++ b/zeppelin-web-angular/projects/zeppelin-sdk/src/interfaces/message-data-type-map.interface.ts @@ -72,6 +72,7 @@ import { ParagraphRemove, ParagraphRemoved, ParagraphStatus, + GetParagraph, ParasInfo, PatchParagraphReceived, PatchParagraphSend, @@ -108,7 +109,7 @@ export interface MessageReceiveDataTypeMap { [OP.IMPORT_NOTE]: ImportNoteReceived; [OP.SAVE_NOTE_FORMS]: SaveNoteFormsSend; [OP.PARAGRAPH]: UpdateParagraph; - [OP.PATCH_PARAGRAPH]: PatchParagraphSend; + [OP.PATCH_PARAGRAPH]: PatchParagraphReceived; [OP.PARAGRAPH_REMOVED]: ParagraphRemoved; [OP.EDITOR_SETTING]: EditorSettingReceived; [OP.PROGRESS]: Progress; @@ -159,7 +160,8 @@ export interface MessageSendDataTypeMap { [OP.PARAGRAPH_CLEAR_ALL_OUTPUT]: ParagraphClearAllOutput; [OP.COMPLETION]: Completion; [OP.COMMIT_PARAGRAPH]: CommitParagraph; - [OP.PATCH_PARAGRAPH]: PatchParagraphReceived; + [OP.PATCH_PARAGRAPH]: PatchParagraphSend; + [OP.GET_PARAGRAPH]: GetParagraph; [OP.IMPORT_NOTE]: ImportNote; [OP.CHECKPOINT_NOTE]: CheckpointNote; [OP.SET_NOTE_REVISION]: SetNoteRevision; diff --git a/zeppelin-web-angular/projects/zeppelin-sdk/src/interfaces/message-operator.interface.ts b/zeppelin-web-angular/projects/zeppelin-sdk/src/interfaces/message-operator.interface.ts index 1f8036b3931..054d44f644b 100644 --- a/zeppelin-web-angular/projects/zeppelin-sdk/src/interfaces/message-operator.interface.ts +++ b/zeppelin-web-angular/projects/zeppelin-sdk/src/interfaces/message-operator.interface.ts @@ -494,6 +494,7 @@ export enum OP { * patch editor text */ PATCH_PARAGRAPH = 'PATCH_PARAGRAPH', + GET_PARAGRAPH = 'GET_PARAGRAPH', /** * [s-c] diff --git a/zeppelin-web-angular/projects/zeppelin-sdk/src/interfaces/message-paragraph.interface.ts b/zeppelin-web-angular/projects/zeppelin-sdk/src/interfaces/message-paragraph.interface.ts index 2ea3916ea1e..5ecf8d0f74e 100644 --- a/zeppelin-web-angular/projects/zeppelin-sdk/src/interfaces/message-paragraph.interface.ts +++ b/zeppelin-web-angular/projects/zeppelin-sdk/src/interfaces/message-paragraph.interface.ts @@ -171,6 +171,8 @@ export interface RunParagraph extends SendParagraph { export interface CommitParagraph extends SendParagraph { noteId: string; + // checksum of the text this client believed the server held; absent for older clients + baseChecksum?: number; } export interface RunAllParagraphs { @@ -260,14 +262,26 @@ export interface CompletionReceived { } export interface PatchParagraphReceived { - id: string; + paragraphId: string; noteId: string; patch: string; + // checksums of the sender's text before and after the patch; absent for older clients + baseChecksum?: number; + afterChecksum?: number; +} + +export interface GetParagraph { + id: string; + noteId: string; } export interface PatchParagraphSend { - paragraphId: string; + id: string; + noteId: string; patch: string; + // checksums of this client's text before and after the patch; let receivers verify the result + baseChecksum?: number; + afterChecksum?: number; } export interface ParagraphRemoved { diff --git a/zeppelin-web-angular/projects/zeppelin-sdk/src/message.ts b/zeppelin-web-angular/projects/zeppelin-sdk/src/message.ts index 6262bff26a4..a742be56c77 100644 --- a/zeppelin-web-angular/projects/zeppelin-sdk/src/message.ts +++ b/zeppelin-web-angular/projects/zeppelin-sdk/src/message.ts @@ -441,7 +441,8 @@ export class Message { paragraphData: string, paragraphConfig: ParagraphConfig, paragraphParams: ParagraphConfig, - noteId: string + noteId: string, + baseChecksum?: number ): void { return this.send(OP.COMMIT_PARAGRAPH, { id: paragraphId, @@ -449,18 +450,34 @@ export class Message { title: paragraphTitle, paragraph: paragraphData, config: paragraphConfig, - params: paragraphParams + params: paragraphParams, + baseChecksum }); } - patchParagraph(paragraphId: string, noteId: string, patch: string): void { + patchParagraph( + paragraphId: string, + noteId: string, + patch: string, + baseChecksum?: number, + afterChecksum?: number + ): void { // 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, { id: paragraphId, noteId, - patch: normalPatch + patch: normalPatch, + baseChecksum, + afterChecksum + }); + } + + getParagraph(paragraphId: string, noteId: string): void { + return this.send(OP.GET_PARAGRAPH, { + id: paragraphId, + noteId }); } diff --git a/zeppelin-web-angular/projects/zeppelin-sdk/src/public-api.ts b/zeppelin-web-angular/projects/zeppelin-sdk/src/public-api.ts index 5e6b79271ae..6488b1953f9 100644 --- a/zeppelin-web-angular/projects/zeppelin-sdk/src/public-api.ts +++ b/zeppelin-web-angular/projects/zeppelin-sdk/src/public-api.ts @@ -10,5 +10,6 @@ * limitations under the License. */ +export * from './checksum'; export * from './interfaces/public-api'; export * from './message'; 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 4e7c0e0fde8..44011927760 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,7 +23,8 @@ import { ParagraphConfigResults, ParagraphEditorSetting, ParagraphItem, - ParagraphIResultsMsgItem + ParagraphIResultsMsgItem, + textChecksum } from '@zeppelin/sdk'; import * as DiffMatchPatch from 'diff-match-patch'; @@ -159,8 +160,16 @@ export abstract class ParagraphBase extends MessageListenersManager { if (!this.paragraph.text) { this.paragraph.text = ''; } + // patch_apply never throws: it drops a hunk it cannot place, and fuzzy matching can apply + // one at the wrong offset while still reporting success. Comparing against the sender's + // checksums is the only way to tell both apart. + const startedFromSameText = data.baseChecksum === textChecksum(this.paragraph.text); this.paragraph.text = this.diffMatchPatch.patch_apply(patch, this.paragraph.text)[0]; this.originalText = this.paragraph.text; + if (startedFromSameText && data.afterChecksum !== textChecksum(this.paragraph.text)) { + // we no longer hold the text the sender produced, so ask for this paragraph again + this.messageService.getParagraph(this.paragraph.id, data.noteId); + } this.cdr.markForCheck(); } } 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 186b4c595c8..dbd439b56ed 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 @@ -41,7 +41,8 @@ import { Note, ParagraphConfigResult, ParagraphItem, - ParagraphIResultsMsgItem + ParagraphIResultsMsgItem, + textChecksum } from '@zeppelin/sdk'; import { HeliumService, @@ -202,8 +203,10 @@ export class NotebookParagraphComponent } this.originalText = this.originalText ? this.originalText : ''; const patch = this.diffMatchPatch.patch_make(this.originalText, this.dirtyText).toString(); + const baseChecksum = textChecksum(this.originalText); + const afterChecksum = textChecksum(this.dirtyText); this.originalText = this.dirtyText; - this.messageService.patchParagraph(this.paragraph.id, this.note.id, patch); + this.messageService.patchParagraph(this.paragraph.id, this.note.id, patch, baseChecksum, afterChecksum); } startSaveTimer() { @@ -521,7 +524,10 @@ export class NotebookParagraphComponent config, settings: { params } } = this.paragraph; - this.messageService.commitParagraph(id, title, text, config, params, this.note.id); + // in collaborative mode let the server reject this commit when our text is based on a + // paragraph state the server no longer holds + const baseChecksum = this.collaborativeMode ? textChecksum(this.originalText || '') : undefined; + this.messageService.commitParagraph(id, title, text, config, params, this.note.id, baseChecksum); 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 9b86a7d42d6..af6a55c80fc 100644 --- a/zeppelin-web-angular/src/app/services/message.service.ts +++ b/zeppelin-web-angular/src/app/services/message.service.ts @@ -299,13 +299,32 @@ export class MessageService extends Message implements OnDestroy { paragraphData: string, paragraphConfig: ParagraphConfig, paragraphParams: ParagraphConfig, - noteId: string + noteId: string, + baseChecksum?: number + ): void { + super.commitParagraph( + paragraphId, + paragraphTitle, + paragraphData, + paragraphConfig, + paragraphParams, + noteId, + baseChecksum + ); + } + + patchParagraph( + paragraphId: string, + noteId: string, + patch: string, + baseChecksum?: number, + afterChecksum?: number ): void { - super.commitParagraph(paragraphId, paragraphTitle, paragraphData, paragraphConfig, paragraphParams, noteId); + super.patchParagraph(paragraphId, noteId, patch, baseChecksum, afterChecksum); } - patchParagraph(paragraphId: string, noteId: string, patch: string): void { - super.patchParagraph(paragraphId, noteId, patch); + getParagraph(paragraphId: string, noteId: string): void { + super.getParagraph(paragraphId, noteId); } importNote(note: ImportNote['note']): void {