Transactional Batch Saving

CHANGELOG.md

@@ -23,6 +23,7 @@ For upgrade instructions, please check the [migration guide](MIGRATIONS.md).
 - Brush size is independent of zoom value, now. This change simplifies volume annotations, as brush sizes can be adapted to certain structures (e.g., vesicles) and don't need to be changed when zooming. [#3868](https://github.com/scalableminds/webknossos/pull/3889)
 
 ### Fixed
+- Fixed a bug where failed large save requests lead to inconsistent tracings on the server. [#3829](https://github.com/scalableminds/webknossos/pull/3829)
 - Fixed the setting which enables to hide the planes within the 3D viewport. [#3857](https://github.com/scalableminds/webknossos/pull/3857)
 - Fixed a bug which allowed the brush size to become negative when using shortcuts. [#3861](https://github.com/scalableminds/webknossos/pull/3861)
 - Fixed interpolation along z-axis. [#3888](https://github.com/scalableminds/webknossos/pull/3888)

MIGRATIONS.md

@@ -5,6 +5,7 @@ This project adheres to [Calendar Versioning](http://calver.org/) `0Y.0M.MICRO`.
 User-facing changes are documented in the [changelog](CHANGELOG.md).
 
 ## Unreleased
+- Redis is now needed for the tracingstore module. Make sure to install redis in your setup and adapt the config keys `tracingstore.redis.address` and `tracingstore.redis.port`.
 - To ensure that the existing behavior for loading data is preserved ("best quality first" as opposed to the new "progressive quality" default) execute: `update webknossos.user_datasetconfigurations set configuration = configuration || jsonb '{"loadingStrategy":"BEST_QUALITY_FIRST"}'`. See [#3801](https://github.com/scalableminds/webknossos/pull/3801) for additional context.
 
 ### Postgres Evolutions:

README.md

@@ -51,6 +51,7 @@ See the [wiki](https://github.com/scalableminds/webknossos/wiki/Try-setup) for i
 * [Oracle JDK 8+](http://www.oracle.com/technetwork/java/javase/downloads/index.html) or [Open JDK 8+](http://openjdk.java.net/) (full JDK, JRE is not enough)
 * [sbt](http://www.scala-sbt.org/)
 * [PostgreSQL 10](https://www.postgresql.org/)
+* [Redis 5+](https://redis.io/)
 * [node.js 9+](http://nodejs.org/download/)
 * [yarn package manager](https://yarnpkg.com/)
 * [git](http://git-scm.com/downloads)
@@ -66,7 +67,7 @@ Or install Java manually and run:
 /usr/bin/ruby -e "$(curl -fsSL https://raw.githubusercontent.com/Homebrew/install/master/install)"
 
 # Install git, node.js, postgres, sbt, gfind, gsed
-brew install git node postgresql sbt findutils coreutils gnu-sed
+brew install git node postgresql sbt findutils coreutils gnu-sed redis
 npm install -g yarn
 
 # Start postgres
@@ -98,7 +99,7 @@ echo "deb https://dl.yarnpkg.com/debian/ stable main" | sudo tee /etc/apt/source
 
 # Installing everything
 sudo apt-get update
-sudo apt-get install -y git postgresql-10 postgresql-client-10 nodejs scala sbt openjdk-8-jdk yarn
+sudo apt-get install -y git postgresql-10 postgresql-client-10 nodejs scala sbt openjdk-8-jdk yarn redis-server
 
 # Assign a password to PostgreSQL user
 sudo -u postgres psql -c "ALTER USER postgres WITH ENCRYPTED PASSWORD 'postgres';"
@@ -119,6 +120,9 @@ See: http://www.scala-sbt.org/release/docs/Getting-Started/Setup.html
 * Install PostgreSQL from https://www.postgresql.org/download/
 * PostgreSQL version **10+ is required**
 
+##### Redis
+* Install Redis from https://redis.io/download
+
 ##### node.js & yarn
 * Install node from http://nodejs.org/download/
 * node version **9+ is required**
@@ -129,7 +133,7 @@ See: http://www.scala-sbt.org/release/docs/Getting-Started/Setup.html
 yarn start
 ```
 Will fetch all Scala, Java and node dependencies and run the application on Port 9000.
-Make sure that the PostgreSQL service is running before you start the application.
+Make sure that the PostgreSQL and Redis services are running before you start the application.
 
 ## Production setup
 [See wiki](https://github.com/scalableminds/webknossos/wiki/Production-setup) for recommended production setup.

conf/application.conf

@@ -38,6 +38,10 @@ tracingstore {
     address = "localhost"
     port = 7155
   }
+  redis {
+    address = "localhost"
+    port = 6379
+  }
 }
 
 http {

docker-compose.yml

@@ -11,6 +11,7 @@ services:
     links:
       - "fossildb-persisted:fossildb"
       - "postgres-persisted:postgres"
+      - redis
     depends_on:
       postgres-persisted:
         condition: service_healthy
@@ -21,6 +22,7 @@ services:
       - -Djava.net.preferIPv4Stack=true
       - -Dhttp.address=0.0.0.0
       - -Dtracingstore.fossildb.address=fossildb
+      - -Dtracingstore.redis.address=redis
       - -Dslick.db.url=jdbc:postgresql://postgres/webknossos
       - -Dapplication.insertInitialData=false
       - -Dapplication.authentication.enableDevAutoLogin=false
@@ -65,9 +67,11 @@ services:
       - -Dhttp.address=0.0.0.0
       - -Dhttp.uri=http://webknossos-datastore:9050
       - -Dtracingstore.fossildb.address=fossildb
+      - -Dtracingstore.redis.address=redis
       - -Ddatastore.oxalis.uri=webknossos:9000
     links:
       - fossildb-persisted:fossildb
+      - redis
     depends_on:
       fossildb-persisted:
         condition: service_healthy
@@ -117,6 +121,7 @@ services:
     links:
       - "fossildb-dev:fossildb"
       - "postgres-dev:postgres"
+      - redis
     depends_on:
       postgres-dev:
         condition: service_healthy
@@ -133,7 +138,8 @@ services:
         "run
           -Djava.net.preferIPv4Stack=true
           -Dhttp.address=0.0.0.0
-          -Dtracingstore.fossildb.address=fossildb"
+          -Dtracingstore.fossildb.address=fossildb
+          -Dtracingstore.redis.address=redis"
     stdin_open: true
 
   # Tests
@@ -159,6 +165,7 @@ services:
     links:
       - postgres
       - fossildb
+      - redis
     depends_on:
       postgres:
         condition: service_healthy
@@ -173,7 +180,8 @@ services:
         sbt
         -v
         "testOnly e2e.* --
-          -Dtracingstore.fossildb.address=fossildb"
+          -Dtracingstore.fossildb.address=fossildb
+          -Dtracingstore.redis.address=redis"
     volumes:
       - ./binaryData/Connectomics department:/home/${USER_NAME:-sbt-user}/webknossos/binaryData/Organization_X
 
@@ -261,3 +269,9 @@ services:
     volumes:
       - "./fossildb-dev/data:/fossildb/data"
       - "./fossildb-dev/backup:/fossildb/backup"
+
+  # Redis
+  redis:
+    image: redis:5.0-alpine
+    command:
+      - redis-server

frontend/javascripts/oxalis/model/actions/save_actions.js

@@ -1,13 +1,15 @@
 // @flow
 import type { UpdateAction } from "oxalis/model/sagas/update_actions";
+import { getUid } from "libs/uid_generator";
 import Date from "libs/date";
 
 type Tracing = "skeleton" | "volume";
 
-type PushSaveQueueAction = {
-  type: "PUSH_SAVE_QUEUE",
+type PushSaveQueueTransaction = {
+  type: "PUSH_SAVE_QUEUE_TRANSACTION",
   items: Array<UpdateAction>,
   tracingType: Tracing,
+  transactionId: string,
 };
 type SaveNowAction = { type: "SAVE_NOW" };
 type ShiftSaveQueueAction = {
@@ -30,7 +32,7 @@ type SetVersionNumberAction = {
 type UndoAction = { type: "UNDO" };
 type RedoAction = { type: "REDO" };
 export type SaveAction =
-  | PushSaveQueueAction
+  | PushSaveQueueTransaction
   | SaveNowAction
   | ShiftSaveQueueAction
   | DiscardSaveQueuesAction
@@ -40,13 +42,15 @@ export type SaveAction =
   | UndoAction
   | RedoAction;
 
-export const pushSaveQueueAction = (
+export const pushSaveQueueTransaction = (
   items: Array<UpdateAction>,
   tracingType: Tracing,
-): PushSaveQueueAction => ({
-  type: "PUSH_SAVE_QUEUE",
+  transactionId: string = getUid(),
+): PushSaveQueueTransaction => ({
+  type: "PUSH_SAVE_QUEUE_TRANSACTION",
   items,
   tracingType,
+  transactionId,
 });
 
 export const saveNowAction = (): SaveNowAction => ({

frontend/javascripts/oxalis/model/bucket_data_handling/wkstore_adapter.js

@@ -11,7 +11,7 @@ import {
   getByteCountFromLayer,
 } from "oxalis/model/accessors/dataset_accessor";
 import { parseAsMaybe } from "libs/utils";
-import { pushSaveQueueAction } from "oxalis/model/actions/save_actions";
+import { pushSaveQueueTransaction } from "oxalis/model/actions/save_actions";
 import { updateBucket } from "oxalis/model/sagas/update_actions";
 import ByteArrayToBase64Worker from "oxalis/workers/byte_array_to_base64.worker";
 import DecodeFourBitWorker from "oxalis/workers/decode_four_bit.worker";
@@ -178,5 +178,5 @@ export async function sendToStore(batch: Array<DataBucket>): Promise<void> {
     const base64 = await byteArrayToBase64(bucketData);
     items.push(updateBucket(bucketInfo, base64));
   }
-  Store.dispatch(pushSaveQueueAction(items, "volume"));
+  Store.dispatch(pushSaveQueueTransaction(items, "volume"));
 }

frontend/javascripts/oxalis/model/helpers/action_logger_middleware.js

@@ -11,7 +11,7 @@ const actionBlacklist = [
   "MOVE_FLYCAM",
   "MOVE_FLYCAM_ORTHO",
   "MOVE_PLANE_FLYCAM_ORTHO",
-  "PUSH_SAVE_QUEUE",
+  "PUSH_SAVE_QUEUE_TRANSACTION",
   "SET_DIRECTION",
   "SET_INPUT_CATCHER_RECT",
   "SET_MOUSE_POSITION",

frontend/javascripts/oxalis/model/reducers/save_reducer.js

@@ -8,36 +8,45 @@ import update from "immutability-helper";
 
 import type { Action } from "oxalis/model/actions/actions";
 import type { OxalisState } from "oxalis/store";
+import { getActionLog } from "oxalis/model/helpers/action_logger_middleware";
 import { getStats } from "oxalis/model/accessors/skeletontracing_accessor";
+import { maximumActionCountPerBatch } from "oxalis/model/sagas/save_saga_constants";
 import Date from "libs/date";
 import * as Utils from "libs/utils";
-import { getActionLog } from "oxalis/model/helpers/action_logger_middleware";
 
 function SaveReducer(state: OxalisState, action: Action): OxalisState {
   switch (action.type) {
-    case "PUSH_SAVE_QUEUE": {
+    case "PUSH_SAVE_QUEUE_TRANSACTION": {
       // Only report tracing statistics, if a "real" update to the tracing happened
       const stats = _.some(action.items, ua => ua.name !== "updateTracing")
         ? Utils.toNullable(getStats(state.tracing))
         : null;
-      const { items } = action;
+      const { items, transactionId } = action;
+
       if (items.length > 0) {
+        const updateActionChunks = _.chunk(items, maximumActionCountPerBatch);
+        const transactionGroupCount = updateActionChunks.length;
+
+        const oldQueue = state.save.queue[action.tracingType];
+        const newQueue = oldQueue.concat(
+          updateActionChunks.map((actions, transactionGroupIndex) => ({
+            // Placeholder, the version number will be updated before sending to the server
+            version: -1,
+            transactionId,
+            transactionGroupCount,
+            transactionGroupIndex,
+            timestamp: Date.now(),
+            actions,
+            stats,
+            // TODO: Redux Action Log context for debugging purposes. Remove this again if it is no longer needed.
+            info: JSON.stringify(getActionLog().slice(-50)),
+          })),
+        );
         return update(state, {
           save: {
             queue: {
               [action.tracingType]: {
-                $push: [
-                  {
-                    // Placeholder, the version number and requestId will be updated before sending to the server
-                    version: -1,
-                    requestId: "",
-                    timestamp: Date.now(),
-                    actions: items,
-                    stats,
-                    // TODO: Redux Action Log context for debugging purposes. Remove this again if it is no longer needed.
-                    info: JSON.stringify(getActionLog().slice(-50)),
-                  },
-                ],
+                $set: newQueue,
               },
             },
             progressInfo: {

frontend/javascripts/oxalis/model/sagas/save_saga.js

@@ -8,6 +8,13 @@ import Maybe from "data.maybe";
 import _ from "lodash";
 
 import { FlycamActions } from "oxalis/model/actions/flycam_actions";
+import {
+  PUSH_THROTTLE_TIME,
+  SAVE_RETRY_WAITING_TIME,
+  MAX_SAVE_RETRY_WAITING_TIME,
+  UNDO_HISTORY_SIZE,
+  maximumActionCountPerSave,
+} from "oxalis/model/sagas/save_saga_constants";
 import type { Tracing, Flycam, SaveQueueEntry } from "oxalis/store";
 import { type UpdateAction, moveTreeComponent } from "oxalis/model/sagas/update_actions";
 import { VolumeTracingSaveRelevantActions } from "oxalis/model/actions/volumetracing_actions";
@@ -33,26 +40,17 @@ import {
   shiftSaveQueueAction,
   setSaveBusyAction,
   setLastSaveTimestampAction,
-  pushSaveQueueAction,
+  pushSaveQueueTransaction,
   setVersionNumberAction,
 } from "oxalis/model/actions/save_actions";
 import Date from "libs/date";
 import Request, { type RequestOptionsWithData } from "libs/request";
 import Toast from "libs/toast";
 import messages from "messages";
 import window, { alert, document, location } from "libs/window";
-import { getUid } from "libs/uid_generator";
 
 import { enforceSkeletonTracing } from "../accessors/skeletontracing_accessor";
 
-const PUSH_THROTTLE_TIME = 30000; // 30s
-const SAVE_RETRY_WAITING_TIME = 2000;
-const MAX_SAVE_RETRY_WAITING_TIME = 300000; // 5m
-const UNDO_HISTORY_SIZE = 20;
-
-export const maximumActionCountPerBatch = 5000;
-const maximumActionCountPerSave = 15000;
-
 export function* collectUndoStates(): Saga<void> {
   const undoStack = [];
   const redoStack = [];
@@ -106,13 +104,13 @@ export function* pushTracingTypeAsync(tracingType: "skeleton" | "volume"): Saga<
   yield* put(setLastSaveTimestampAction(tracingType));
   while (true) {
     let saveQueue;
-    // Check whether the save queue is actually empty, the PUSH_SAVE_QUEUE action
+    // Check whether the save queue is actually empty, the PUSH_SAVE_QUEUE_TRANSACTION action
     // could have been triggered during the call to sendRequestToServer
 
     saveQueue = yield* select(state => state.save.queue[tracingType]);
     if (saveQueue.length === 0) {
       // Save queue is empty, wait for push event
-      yield* take("PUSH_SAVE_QUEUE");
+      yield* take("PUSH_SAVE_QUEUE_TRANSACTION");
     }
     yield* race({
       timeout: _call(delay, PUSH_THROTTLE_TIME),
@@ -174,8 +172,6 @@ export function* sendRequestToServer(tracingType: "skeleton" | "volume"): Saga<v
   const tracingStoreUrl = yield* select(state => state.tracing.tracingStore.url);
   compactedSaveQueue = addVersionNumbers(compactedSaveQueue, version);
 
-  compactedSaveQueue = addRequestIds(compactedSaveQueue, getUid());
-
   let retryCount = 0;
   while (true) {
     try {
@@ -230,13 +226,6 @@ export function addVersionNumbers(
   return updateActionsBatches.map(batch => Object.assign({}, batch, { version: ++lastVersion }));
 }
 
-export function addRequestIds(
-  updateActionsBatches: Array<SaveQueueEntry>,
-  requestId: string,
-): Array<SaveQueueEntry> {
-  return updateActionsBatches.map(batch => Object.assign({}, batch, { requestId }));
-}
-
 function removeUnrelevantUpdateActions(updateActions: Array<UpdateAction>) {
   // This functions removes update actions that should not be sent to the server.
   return updateActions.filter(ua => ua.name !== "toggleTree");
@@ -484,11 +473,7 @@ export function* saveTracingTypeAsync(tracingType: "skeleton" | "volume"): Saga<
       ),
     );
     if (items.length > 0) {
-      const updateActionChunks = _.chunk(items, maximumActionCountPerBatch);
-
-      for (const updateActionChunk of updateActionChunks) {
-        yield* put(pushSaveQueueAction(updateActionChunk, tracingType));
-      }
+      yield* put(pushSaveQueueTransaction(items, tracingType));
     }
     prevTracing = tracing;
     prevFlycam = flycam;

frontend/javascripts/oxalis/model/sagas/save_saga_constants.js

@@ -0,0 +1,9 @@
+// @flow
+
+export const PUSH_THROTTLE_TIME = 30000; // 30s
+export const SAVE_RETRY_WAITING_TIME = 2000;
+export const MAX_SAVE_RETRY_WAITING_TIME = 300000; // 5m
+export const UNDO_HISTORY_SIZE = 20;
+
+export const maximumActionCountPerBatch = 5000;
+export const maximumActionCountPerSave = 15000;

frontend/javascripts/oxalis/model/sagas/skeletontracing_saga.js

@@ -186,7 +186,7 @@ export function* watchSkeletonTracingAsync(): Saga<void> {
     ],
     centerActiveNode,
   );
-  yield _throttle(5000, "PUSH_SAVE_QUEUE", watchTracingConsistency);
+  yield _throttle(5000, "PUSH_SAVE_QUEUE_TRANSACTION", watchTracingConsistency);
   yield* fork(watchFailedNodeCreations);
   yield* fork(watchBranchPointDeletion);
   yield* fork(watchVersionRestoreParam);