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 {