Skip to content

Commit

Permalink
♻ Throw error for pending ops on hard rollback (#626)
Browse files Browse the repository at this point in the history 
Our current flow for handling the rollback swallows some errors and discards pending ops.

1. Whenever we do hard rollback and fetch for hard rollback fails we should throw the error so that user is informed that something really wrong happen.
2. Whenever we preform the hard rollback and we have some pending ops and the `ERR_OP_PENDING_OP_SUBMIT_REJECTED` error is thrown we should soft reject, as this error is meant for the user to reject some of the ops from submition to DB, however we should throw for all the pending ops s the pending ops would be otherwise just discarded. This way we at least allow user to handle the error
3. For the soft rollback we just resolve for this error and allow pending ops to be submitted to the server.
  • Loading branch information
@dawidreedsy
dawidreedsy authored Oct 10, 2023
1 parent 8b0e5b8 commit dc033c2
Show file tree
Hide file tree
Showing 5 changed files with 190 additions and 9 deletions.
4 changes: 4 additions & 0 deletions docs/api/sharedb-error.md
View file
Original file line number Diff line number Diff line change
Expand Up @@ -35,6 +35,10 @@ Representation of an error, with a machine-parsable [code](#error-codes).
> This error might be used as part of standard control flow. For example, consumers may define a middleware that validates document structure, and rejects operations that do not conform to this schema using this error code to reset the client to a valid state.
### `ERR_PENDING_OP_REMOVED_BY_OP_SUBMIT_REJECTED`

> This may happen if server rejected op with ERR_OP_SUBMIT_REJECTED and the type is not invertible or there are some pending ops after the create op was rejected with ERR_OP_SUBMIT_REJECTED
### `ERR_OP_ALREADY_SUBMITTED`

> The same op has been received by the server twice.
Expand Down
55 changes: 46 additions & 9 deletions lib/client/doc.js
View file
Original file line number Diff line number Diff line change
Expand Up @@ -329,10 +329,6 @@ Doc.prototype._handleSubscribe = function(error, snapshot) {
Doc.prototype._handleOp = function(err, message) {
if (err) {
if (this.inflightOp) {
// The server has rejected submission of the current operation. If we get
// an "Op submit rejected" error, this was done intentionally
// and we should roll back but not return an error to the user.
if (err.code === ERROR_CODE.ERR_OP_SUBMIT_REJECTED) err = null;
return this._rollback(err);
}
return this.emit('error', err);
Expand Down Expand Up @@ -1014,6 +1010,13 @@ Doc.prototype._rollback = function(err) {
return this._hardRollback(error);
}

// The server has rejected submission of the current operation. If we get
// an "Op submit rejected" error, this was done intentionally
// and we should roll back but not return an error to the user.
if (err.code === ERROR_CODE.ERR_OP_SUBMIT_REJECTED) {
return this._clearInflightOp(null);
}

this._clearInflightOp(err);
};

Expand All @@ -1023,9 +1026,8 @@ Doc.prototype._hardRollback = function(err) {
// to hit a condition where we have no inflight op, but we do have pending
// ops. This can happen when an invalid op is submitted, which causes us
// to hard rollback before the pending op was flushed.
var pendingOps = [];
if (this.inflightOp) pendingOps.push(this.inflightOp);
pendingOps = pendingOps.concat(this.pendingOps);
var pendingOps = this.pendingOps;
var inflightOp = this.inflightOp;

// Cancel all pending ops and reset if we can't invert
this._setType(null);
Expand All @@ -1035,17 +1037,50 @@ Doc.prototype._hardRollback = function(err) {

// Fetch the latest version from the server to get us back into a working state
var doc = this;
this._fetch({force: true}, function() {
this._fetch({force: true}, function(fetchError) {
// We want to check that no errors are swallowed, so we check that:
// - there are callbacks to call, and
// - that every single pending op called a callback
// If there are no ops queued, or one of them didn't handle the error,
// then we emit the error.

if (fetchError) {
// This is critical error as it means that our doc is not in usable state
// anymore, we should throw doc error.
logger.error('Hard rollback doc fetch failed.', fetchError, inflightOp);

doc.emit('error', new ShareDBError(
ERROR_CODE.ERR_HARD_ROLLBACK_FETCH_FAILED,
'Hard rollback fetch failed: ' + fetchError.message
));
}

if (err.code === ERROR_CODE.ERR_OP_SUBMIT_REJECTED) {
/**
* Handle special case of ERR_OP_SUBMIT_REJECTED
* This ensures that we resolve the main op callback and reject
* all the pending ops. This is hard rollback so all the pending ops will be
* discarded. This will ensure that the user is at least informed about it.
* more info: https://github.com/share/sharedb/pull/626
*/
if (inflightOp) {
util.callEach(inflightOp.callbacks);
inflightOp = null;
}

if (!pendingOps.length) return;
err = new ShareDBError(
ERROR_CODE.ERR_PENDING_OP_REMOVED_BY_OP_SUBMIT_REJECTED,
'Discarding pending op because of hard rollback during ERR_OP_SUBMIT_REJECTED'
);
}

if (inflightOp) pendingOps.unshift(inflightOp);
var allOpsHadCallbacks = !!pendingOps.length;
for (var i = 0; i < pendingOps.length; i++) {
allOpsHadCallbacks = util.callEach(pendingOps[i].callbacks, err) && allOpsHadCallbacks;
}
if (err && !allOpsHadCallbacks) return doc.emit('error', err);
if (err && !allOpsHadCallbacks) doc.emit('error', err);
});
};

Expand All @@ -1061,3 +1096,5 @@ Doc.prototype._clearInflightOp = function(err) {

if (err && !called) return this.emit('error', err);
};


2 changes: 2 additions & 0 deletions lib/error.js
View file
Original file line number Diff line number Diff line change
Expand Up @@ -38,6 +38,8 @@ ShareDBError.CODES = {
ERR_OP_ALREADY_SUBMITTED: 'ERR_OP_ALREADY_SUBMITTED',
ERR_OP_NOT_ALLOWED_IN_PROJECTION: 'ERR_OP_NOT_ALLOWED_IN_PROJECTION',
ERR_OP_SUBMIT_REJECTED: 'ERR_OP_SUBMIT_REJECTED',
ERR_PENDING_OP_REMOVED_BY_OP_SUBMIT_REJECTED: 'ERR_PENDING_OP_REMOVED_BY_OP_SUBMIT_REJECTED',
ERR_HARD_ROLLBACK_FETCH_FAILED: 'ERR_HARD_ROLLBACK_FETCH_FAILED',
ERR_OP_VERSION_MISMATCH_AFTER_TRANSFORM: 'ERR_OP_VERSION_MISMATCH_AFTER_TRANSFORM',
ERR_OP_VERSION_MISMATCH_DURING_TRANSFORM: 'ERR_OP_VERSION_MISMATCH_DURING_TRANSFORM',
ERR_OP_VERSION_NEWER_THAN_CURRENT_SNAPSHOT: 'ERR_OP_VERSION_NEWER_THAN_CURRENT_SNAPSHOT',
Expand Down
33 changes: 33 additions & 0 deletions test/client/doc.js
View file
Original file line number Diff line number Diff line change
Expand Up @@ -6,6 +6,7 @@ var richText = require('rich-text').type;
var ShareDBError = require('../../lib/error');
var errorHandler = require('../util').errorHandler;
var sinon = require('sinon');
var types = require('../../lib/types');

describe('Doc', function() {
beforeEach(function() {
Expand Down Expand Up @@ -514,6 +515,38 @@ describe('Doc', function() {
], done);
});

it('throws an error when hard rollback fetch failed', function(done) {
var backend = this.backend;
doc = this.connection.get('dogs', 'scrappy');
types.register(richText);

async.series([
doc.create.bind(doc, {ops: [{insert: 'Scrappy'}]}, 'rich-text'),
function(next) {
backend.use('reply', function(replyContext, cb) {
if (replyContext.request.a !== 'f') return cb();
cb({code: 'FETCH_ERROR'});
});
backend.use('submit', function(_context, cb) {
cb(new ShareDBError('SUBMIT_ERROR'));
});
var nonInvertibleOp = [{insert: 'e'}];

var count = 0;
function expectError(code) {
count++;
return function(error) {
expect(error.code).to.equal(code);
count--;
if (!count) next();
};
}

doc.on('error', expectError('ERR_HARD_ROLLBACK_FETCH_FAILED'));
doc.submitOp(nonInvertibleOp, expectError('SUBMIT_ERROR'));
}
], done);
});

it('rescues an irreversible op collision', function(done) {
// This test case attempts to reconstruct the following corner case, with
Expand Down
105 changes: 105 additions & 0 deletions test/client/submit.js
View file
Original file line number Diff line number Diff line change
Expand Up @@ -5,9 +5,11 @@ var types = require('../../lib/types');
var deserializedType = require('./deserialized-type');
var numberType = require('./number-type');
var errorHandler = require('../util').errorHandler;
var richText = require('rich-text');
types.register(deserializedType.type);
types.register(deserializedType.type2);
types.register(numberType.type);
types.register(richText.type);

module.exports = function() {
describe('client submit', function() {
Expand Down Expand Up @@ -919,6 +921,27 @@ module.exports = function() {
], done);
});


it('request.rejectedError() soft rejects main op and throws for pending ops on hard rollback', function(done) {
this.backend.use('submit', function(request, next) {
if (request.op.create) {
next(request.rejectedError());
}
});

var connection = this.backend.connect();
var doc = connection.get('dogs', 'fido');
doc.preventCompose = true;

doc.create({age: 3}, function(error) {
if (error) done(error);
});
doc.submitOp({p: ['age'], na: 1}, function(err) {
expect(err.code).to.be.equal('ERR_PENDING_OP_REMOVED_BY_OP_SUBMIT_REJECTED');
done();
});
});

it('request.rejectedError() soft rejects a create', function(done) {
this.backend.use('submit', function(request, next) {
next(request.rejectedError());
Expand Down Expand Up @@ -949,6 +972,29 @@ module.exports = function() {
});
});

it(
'request.rejectedError() soft rejects main op and throws for pending ops on hard rollback without callback',
function(done) {
this.backend.use('submit', function(request, next) {
if (request.op.create) {
next(request.rejectedError());
}
});

var connection = this.backend.connect();
var doc = connection.get('dogs', 'fido');
doc.preventCompose = true;

doc.create({age: 3});
doc.submitOp({p: ['age'], na: 1});

doc.on('error', function(err) {
expect(err.code).to.be.equal('ERR_PENDING_OP_REMOVED_BY_OP_SUBMIT_REJECTED');
done();
});
}
);

it('passing an error in submit middleware rejects an op and calls back with the erorr', function(done) {
this.backend.use('submit', function(request, next) {
if ('op' in request.op) return next({message: 'Custom error'});
Expand Down Expand Up @@ -1043,6 +1089,65 @@ module.exports = function() {
});
});

it('request.rejectedError() soft rejects main op and pending ops for invertible type', function(done) {
var rejectedOnce = false;
this.backend.use('submit', function(request, next) {
if ('op' in request.op && !rejectedOnce) {
rejectedOnce = true;
return next(request.rejectedError());
}
next();
});
var doc = this.backend.connect().get('dogs', 'fido');
doc.preventCompose = true;

doc.create({age: 3}, function(err) {
if (err) return done(err);
doc.submitOp({p: ['age'], na: 1}, function(err) {
if (err) return done(err);
});
doc.submitOp({p: ['age'], na: 3}, function(err) {
if (err) return done(err);
expect(doc.version).equal(2);
expect(doc.data).eql({age: 6});
done();
});
expect(doc.version).equal(1);
expect(doc.data).eql({age: 7});
});
});

it(
'request.rejectedError() soft rejects main op and throws for pending ops for non invertible type',
function(done) {
var rejectedOnce = false;
this.backend.use('submit', function(request, next) {
if ('op' in request.op && !rejectedOnce) {
rejectedOnce = true;
return next(request.rejectedError());
}
next();
});
var doc = this.backend.connect().get('dogs', 'fido');
doc.preventCompose = true;

doc.create({ops: [{insert: 'Scrappy'}]}, 'rich-text', function(err) {
if (err) return done(err);

var nonInvertibleOp = [{insert: 'a'}];
doc.submitOp(nonInvertibleOp, function(err) {
if (err) return done(err);
});
doc.submitOp([{insert: 'b'}], function(err) {
expect(err.code).to.be.equal('ERR_PENDING_OP_REMOVED_BY_OP_SUBMIT_REJECTED');
done();
});
expect(doc.version).equal(1);
expect(doc.data.ops).eql([{insert: 'baScrappy'}]);
});
}
);

it('request.rejectedError() soft rejects an op without callback', function(done) {
this.backend.use('submit', function(request, next) {
if ('op' in request.op) return next(request.rejectedError());
Expand Down

0 comments on commit dc033c2

Please sign in to comment.