Transactions

lib/agent.js

@@ -70,6 +70,8 @@ function Agent(backend, stream) {
   this._firstReceivedMessage = null;
   this._handshakeReceived = false;
 
+  this._transactions = Object.create(null);
+
   // Send the legacy message to initialize old clients with the random agent Id
   this.send(this._initMessage(ACTIONS.initLegacy));
 }
@@ -470,7 +472,7 @@ Agent.prototype._handleMessage = function(request, callback) {
           ));
         }
         if (!op) return callback(new ShareDBError(ERROR_CODE.ERR_MESSAGE_BADLY_FORMED, 'Invalid op message'));
-        return this._submit(request.c, request.d, op, callback);
+        return this._submit(request.c, request.d, op, request.t, callback);
       case ACTIONS.snapshotFetch:
         return this._fetchSnapshot(request.c, request.d, request.v, callback);
       case ACTIONS.snapshotFetchByTimestamp:
@@ -494,6 +496,8 @@ Agent.prototype._handleMessage = function(request, callback) {
         return this._requestPresence(request.ch, callback);
       case ACTIONS.pingPong:
         return this._pingPong(callback);
+      case ACTIONS.transactionCommit:
+        return this._commitTransaction(request, callback);
       default:
         callback(new ShareDBError(ERROR_CODE.ERR_MESSAGE_BADLY_FORMED, 'Invalid or unknown message'));
     }
@@ -761,9 +765,12 @@ Agent.prototype._unsubscribeBulk = function(collection, ids, callback) {
   util.nextTick(callback);
 };
 
-Agent.prototype._submit = function(collection, id, op, callback) {
+Agent.prototype._submit = function(collection, id, op, transactionId, callback) {
   var agent = this;
-  this.backend.submit(this, collection, id, op, null, function(err, ops, request) {
+  var options = {
+    transaction: transactionId
+  };
+  this.backend.submit(this, collection, id, op, options, function(err, ops, request) {
     // Message to acknowledge the op was successfully submitted
     var ack = {src: op.src, seq: op.seq, v: op.v};
     if (request._fixupOps.length) ack[ACTIONS.fixup] = request._fixupOps;
@@ -993,6 +1000,16 @@ function createClientOp(request, clientId) {
         undefined;
 }
 
+Agent.prototype._commitTransaction = function(request, callback) {
+  var transaction = this._transactions[request.id];
+  transaction.commit(callback);
+};
+
+Agent.prototype._abortTransaction = function(request, callback) {
+  var transaction = this._transactions[request.id];
+  transaction.abort(callback);
+};
+
 function shallowCopy(object) {
   var out = {};
   for (var key in object) {

lib/client/connection.js

@@ -12,6 +12,7 @@ var util = require('../util');
 var logger = require('../logger');
 var DocPresenceEmitter = require('./presence/doc-presence-emitter');
 var protocol = require('../protocol');
+var Transaction = require('./transaction');
 
 var ERROR_CODE = ShareDBError.CODES;
 
@@ -64,6 +65,8 @@ function Connection(socket) {
   // A unique message number for presence
   this._presenceSeq = 1;
 
+  this._transactions = Object.create(null);
+
   // Equals agent.src on the server
   this.id = null;
 
@@ -258,6 +261,8 @@ Connection.prototype.handleMessage = function(message) {
       return this._handlePresenceRequest(err, message);
     case ACTIONS.pingPong:
       return this._handlePingPong(err);
+    case ACTIONS.transactionCommit:
+      return this._handleTransactionCommit(err, message);
 
     default:
       logger.warn('Ignoring unrecognized message', message);
@@ -467,6 +472,7 @@ Connection.prototype.sendOp = function(doc, op) {
   if (op.create) message.create = op.create;
   if (op.del) message.del = op.del;
   if (doc.submitSource) message.x.source = op.source;
+  if (op.transaction) message.t = op.transaction;
   this.send(message);
 };
 
@@ -782,6 +788,26 @@ Connection.prototype._initialize = function(message) {
   this._setState('connected');
 };
 
+Connection.prototype.startTransaction = function() {
+  var transaction = new Transaction(this);
+  return this._transactions[transaction.id] = transaction;
+};
+
+Connection.prototype._commitTransaction = function(transaction) {
+  this.send({a: ACTIONS.transactionCommit, id: transaction.id});
+};
+
+Connection.prototype._abortTransaction = function(transaction) {
+  this.send({a: ACTIONS.transactionAbort, id: transaction.id});
+};
+
+Connection.prototype._handleTransactionCommit = function(error, message) {
+  var transaction = this._transactions[message.id];
+  if (!transaction) return;
+  transaction._handleCommit(error, message);
+  delete this._transactions[message.id];
+};
+
 Connection.prototype.getPresence = function(channel) {
   var connection = this;
   var presence = util.digOrCreate(this._presences, channel, function() {

lib/client/doc.js

@@ -121,6 +121,8 @@ function Doc(connection, collection, id) {
   // Internal counter that gets incremented every time doc.data is updated.
   // Used as a cheap way to check if doc.data has changed.
   this._dataStateVersion = 0;
+
+  this._transaction = null;
 }
 emitter.mixin(Doc);
 
@@ -337,6 +339,9 @@ Doc.prototype._handleOp = function(err, message) {
       // the remote, so that we're in a nice consistent state
       return this.fetch(this._clearInflightOp.bind(this));
     }
+    if (this._transaction) {
+      return this._transaction._localAbort(err);
+    }
     if (this.inflightOp) {
       return this._rollback(err);
     }
@@ -763,6 +768,7 @@ Doc.prototype._submit = function(op, source, callback) {
     this._pushOp(op, source, callback);
     this._otApply(op, source);
   } catch (error) {
+    // TODO: Abort transaction
     return this._hardRollback(error);
   }
 
@@ -865,8 +871,10 @@ Doc.prototype.submitOp = function(component, options, callback) {
     callback = options;
     options = null;
   }
+  options = options || {};
   var op = {op: component};
-  var source = options && options.source;
+  this._setTransaction(op, options.transaction);
+  var source = options.source;
   this._submit(op, source, callback);
 };
 
@@ -888,6 +896,7 @@ Doc.prototype.create = function(data, type, options, callback) {
     callback = options;
     options = null;
   }
+  options = options || {};
   if (!type) {
     type = types.defaultType.uri;
   }
@@ -897,6 +906,7 @@ Doc.prototype.create = function(data, type, options, callback) {
     return this.emit('error', err);
   }
   var op = {create: {type: type, data: data}};
+  this._setTransaction(op, options.transaction);
   var source = options && options.source;
   this._submit(op, source, callback);
 };
@@ -913,12 +923,14 @@ Doc.prototype.del = function(options, callback) {
     callback = options;
     options = null;
   }
+  options = options || {};
   if (!this.type) {
     var err = new ShareDBError(ERROR_CODE.ERR_DOC_DOES_NOT_EXIST, 'Document does not exist');
     if (callback) return callback(err);
     return this.emit('error', err);
   }
   var op = {del: true};
+  this._setTransaction(op, options.transaction);
   var source = options && options.source;
   this._submit(op, source, callback);
 };
@@ -1106,4 +1118,25 @@ Doc.prototype._clearInflightOp = function(err) {
   if (err && !called) return this.emit('error', err);
 };
 
+Doc.prototype._setTransaction = function(op, transaction) {
+  if (transaction && typeof transaction === 'object') transaction = transaction.id;
+  if (transaction) {
+    op.transaction = transaction;
 
+    if (this._transaction) {
+      if (this._transaction.id === transaction) return;
+      throw new Error('Transaction already in progress. Commit transaction before starting a new one.');
+    }
+    this._transaction = this.connection._transactions[transaction];
+    if (!this._transaction) {
+      throw new Error('Transaction not started');
+    }
+    if (!this._transaction._writeable) {
+      throw new Error('Transaction is no longer writeable');
+    }
+    // TODO: Tidy up listener on commit/abort
+    this._transaction.on('abort', this._hardRollback.bind(this));
+  } else if (this._transaction) {
+    throw new Error('Transaction in progress. Commit transaction before submitting ops outside transaction.');
+  }
+};

lib/client/transaction.js

@@ -0,0 +1,45 @@
+var emitter = require('../emitter');
+
+var idCounter = 1;
+
+module.exports = Transaction;
+function Transaction(connection) {
+  emitter.EventEmitter.call(this);
+
+  this.connection = connection;
+  this.id = (idCounter++).toString();
+
+  this._callbacks = [];
+  this._writeable = true;
+}
+emitter.mixin(Transaction);
+
+Transaction.prototype.commit = function(callback) {
+  // TODO: Catch multiple calls
+  // TODO: Handle network changes
+  this._callbacks.push(callback);
+  this._writeable = false;
+  this.connection._commitTransaction(this);
+};
+
+Transaction.prototype.abort = function(callback) {
+  this._callbacks.push(callback);
+  this._writeable = false;
+};
+
+Transaction.prototype._handleCommit = function(error) {
+  this._writeable = false;
+  // TODO: Handle callbacks
+  if (error) this._localAbort(error);
+  else this.emit('commit');
+
+  var callbacks = this._callbacks;
+  this._callbacks = [];
+  if (!callbacks.length) this.emit('error', error);
+  for (var callback of callbacks) callback(error);
+};
+
+Transaction.prototype._localAbort = function(error) {
+  this._writeable = false;
+  this.emit('abort', error);
+};

lib/db/memory.js

@@ -25,6 +25,8 @@ function MemoryDB(options) {
   this.ops = Object.create(null);
 
   this.closed = false;
+
+  this._transactions = Object.create(null);
 };
 module.exports = MemoryDB;
 
@@ -39,20 +41,35 @@ MemoryDB.prototype.close = function(callback) {
 // callback(err, succeeded)
 MemoryDB.prototype.commit = function(collection, id, op, snapshot, options, callback) {
   var db = this;
+  options = options || {};
   if (typeof callback !== 'function') throw new Error('Callback required');
   util.nextTick(function() {
     var version = db._getVersionSync(collection, id);
+
+    var transaction;
+    if (options.transaction) {
+      transaction = db._transactions[options.transaction];
+      if (!transaction) return callback(null, false);
+      version = version + transaction.commits.length;
+    }
+
     if (snapshot.v !== version + 1) {
       var succeeded = false;
       return callback(null, succeeded);
     }
-    var err = db._writeOpSync(collection, id, op);
-    if (err) return callback(err);
-    err = db._writeSnapshotSync(collection, id, snapshot);
-    if (err) return callback(err);
 
-    var succeeded = true;
-    callback(null, succeeded);
+    var commit = function() {
+      var err = db._writeOpSync(collection, id, op);
+      if (err) return err;
+      err = db._writeSnapshotSync(collection, id, snapshot);
+      if (err) return err;
+    };
+
+    var error;
+    if (!transaction) error = commit();
+    callback(error, !error);
+
+    if (transaction) transaction.commits.push(commit);
   });
 };
 
@@ -133,6 +150,47 @@ MemoryDB.prototype.query = function(collection, query, fields, options, callback
   });
 };
 
+MemoryDB.prototype.startTransaction = function(id, callback) {
+  if (this._transactions[id]) callback(new Error('Transaction already started'));
+  this._transactions[id] = this._transactions[id] || {};
+  this._transactions[id].commits = [];
+  callback();
+};
+
+MemoryDB.prototype.restartTransaction = function(id, callback) {
+  delete this._transactions[id];
+  this.startTransaction(id, callback);
+};
+
+MemoryDB.prototype.commitTransaction = function(id, callback) {
+  var transaction = this._transactions[id];
+  if (!transaction) return callback();
+
+  delete this._transactions[id];
+
+  // Heavy-handed but effective approach to synchronous transaction writing:
+  // let's just save the DB state before and restore if the transaction fails
+  var docs = JSON.stringify(this.docs);
+  var ops = JSON.stringify(this.ops);
+
+  for (var commit of transaction.commits) {
+    var error = commit();
+
+    if (error) {
+      this.docs = JSON.parse(docs);
+      this.ops = JSON.parse(ops);
+      break;
+    }
+  }
+
+  callback(error, !error);
+};
+
+MemoryDB.prototype.abortTransaction = function(id, callback) {
+  delete this._transactions[id];
+  callback(null);
+};
+
 // For testing, it may be useful to implement the desired query
 // language by defining this function. Returns an object with
 // two properties:

lib/error.js

@@ -72,6 +72,7 @@ ShareDBError.CODES = {
    */
   ERR_SNAPSHOT_READS_REJECTED: 'ERR_SNAPSHOT_READS_REJECTED',
   ERR_SUBMIT_TRANSFORM_OPS_NOT_FOUND: 'ERR_SUBMIT_TRANSFORM_OPS_NOT_FOUND',
+  ERR_TRANSACTION_ABORTED: 'ERR_TRANSACTION_ABORTED',
   ERR_TYPE_CANNOT_BE_PROJECTED: 'ERR_TYPE_CANNOT_BE_PROJECTED',
   ERR_TYPE_DOES_NOT_SUPPORT_COMPOSE: 'ERR_TYPE_DOES_NOT_SUPPORT_COMPOSE',
   ERR_TYPE_DOES_NOT_SUPPORT_PRESENCE: 'ERR_TYPE_DOES_NOT_SUPPORT_PRESENCE',

lib/message-actions.js

@@ -15,6 +15,8 @@ exports.ACTIONS = {
   op: 'op',
   snapshotFetch: 'nf',
   snapshotFetchByTimestamp: 'nt',
+  transactionAbort: 'ta',
+  transactionCommit: 'tc',
   pingPong: 'pp',
   presence: 'p',
   presenceSubscribe: 'ps',