Fix: Stop rebasing sent changes when receiving remote updates (#78)

See prior version in #75 - this is a more formal change

When we have sent changes for an entity and are waiting for server's
response we might first receive updates from the server that represent
changes which another client made to the same entity.

In those cases we have been forgetting about the changes we sent and
then transforming our local changes to the new updates. To our client it
appears like we have made changes locally on top of the remote changes
when in reality we only made changes to the previous base version of the
entity.

The result of this forgetfulness is that we replay changes that should
only happen once. This might appear in the form of duplicated content
but could expose itself in other forms.

In this patch we're guarding the transformation based on the presence of
those waiting changes. If we get updates while waiting, simply apply
those updates to our internal ghost and let the client handle our
original submission once it comes back. If it comes back confirmed by
the server we'll be able to apply the updated patch which the server
created. If it gets rejected we will end up sending the full copy of our
changes and mess up the note but the original data should remian in
the history.

This patch includes a test which demonstrates the sequence that produces
the issue and verifies that it doesn't return.

Props to @beaucollins for his work figuring this stuff out.

Author: dmsnell

RELEASE-NOTES.txt

@@ -0,0 +1,5 @@
+# Changelog
+
+## Future Release
+
+- Fixed data corruption when remote updates arrive while waiting on confirmation of local changes [#78](https://github.com/Simperium/node-simperium/pull/78)

src/simperium/channel.js

@@ -116,7 +116,7 @@ internal.updateObjectVersion = function( id, version, data, original, patch, ack
 	// If it's not an ack, it's a change initiated on a different client
 	// we need to provide a way for the current client to respond to
 	// a potential conflict if it has modifications that have not been synced
-	if ( !acknowledged ) {
+	if ( !acknowledged && !this.localQueue.sent[id] ) {
 		changes = this.localQueue.dequeueChangesFor( id );
 		localModifications = change_util.compressChanges( changes, original );
 		remoteModifications = patch;

test/simperium/channel_test.js

@@ -357,6 +357,53 @@ describe( 'Channel', function() {
 			bucket.update( key, {title: 'Goodbye world'} );
 		} ) );
 
+		it( 'should not rebase local changes waiting for confirmation from the server', async () => {
+			await channel.store.put( 'thing', 1, { content: 'AC' } );
+
+			channel.localQueue.queue( {
+				id: 'thing',
+				o: 'M',
+				sv: 1,
+				ev: 2,
+				ccid: 'local',
+				v: diff( { content: 'AC' }, { content: 'ACD' } )
+			} );
+
+			await new Promise( resolve => channel.once( 'send', resolve ) );
+			channel.once( 'send', () => done( 'Should not re-send changes which are already outbound' ) );
+
+			const ghost = await channel.store.get( 'thing' );
+			deepEqual( ghost.data, { content: 'AC' } );
+
+			channel.handleMessage( 'c:' + JSON.stringify( [ {
+				id: 'thing',
+				o: 'M',
+				sv: 1,
+				ev: 2,
+				ccids: [ 'remote' ],
+				v: diff( { content: 'AC' }, { content: 'ABC' } )
+			} ] ) );
+
+			await channel.store.get( 'thing' );
+
+			return new Promise( resolve => {
+				channel.once( 'acknowledge', async () => {
+					const ghost = await channel.store.get( 'thing' );
+					deepEqual( ghost.data, { content: 'ABCD' } );
+					resolve();
+				} );
+
+				channel.handleMessage( 'c:' + JSON.stringify( [ {
+					id: 'thing',
+					o: 'M',
+					sv: 2,
+					ev: 3,
+					ccids: [ 'local' ],
+					v: diff( { content: 'ABC' }, { content: 'ABCD' } )
+				} ] ) )
+			} );
+		} );
+
 		it( 'should emit errors on the bucket instance', ( done ) => {
 			const error = {error: 404, id: 'thing', ccids: ['abc']}
 			bucket.on( 'error', ( e ) => {