[ZEPPELIN-6563] Collaborative mode can silently desynchronize paragraph text, and a later commit overwrites the server copy

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
   }

zeppelin-server/src/main/java/org/apache/zeppelin/service/NotebookService.java

@@ -751,6 +751,23 @@ public void updateParagraph(String noteId,
                               Map<String, Object> config,
                               ServiceContext context,
                               ServiceCallback<Paragraph> 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<String, Object> params,
+                              Map<String, Object> config,
+                              Integer baseChecksum,
+                              ServiceContext context,
+                              ServiceCallback<Paragraph> 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<Paragraph> 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,

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<String, Object> params = (Map<String, Object>) fromMessage.get("params");
     Map<String, Object> config = (Map<String, Object>) 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<Paragraph>(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<String>(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<Paragraph>(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 {

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;
+    });
+  }
 }

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;
+}

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;

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]

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 {

zeppelin-web-angular/projects/zeppelin-sdk/src/message.ts

@@ -441,26 +441,43 @@ export class Message {
     paragraphData: string,
     paragraphConfig: ParagraphConfig,
     paragraphParams: ParagraphConfig,
-    noteId: string
+    noteId: string,
+    baseChecksum?: number
   ): void {
     return this.send<OP.COMMIT_PARAGRAPH>(OP.COMMIT_PARAGRAPH, {
       id: paragraphId,
       noteId,
       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>(OP.PATCH_PARAGRAPH, {
       id: paragraphId,
       noteId,
-      patch: normalPatch
+      patch: normalPatch,
+      baseChecksum,
+      afterChecksum
+    });
+  }
+
+  getParagraph(paragraphId: string, noteId: string): void {
+    return this.send<OP.GET_PARAGRAPH>(OP.GET_PARAGRAPH, {
+      id: paragraphId,
+      noteId
     });
   }
 

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';

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();
     }
   }