diff --git a/Simperium/src/androidTestSupport/java/com/simperium/BucketTest.java b/Simperium/src/androidTestSupport/java/com/simperium/BucketTest.java index 8ad68917..3fa81904 100644 --- a/Simperium/src/androidTestSupport/java/com/simperium/BucketTest.java +++ b/Simperium/src/androidTestSupport/java/com/simperium/BucketTest.java @@ -191,4 +191,59 @@ public void testMergeLocalChangesWithUpdatedGhost() assertEquals("Line 1\nLine 2\nLine 3\n", note.getContent()); } + /** + * If two different notes are both saved and then both deleted, they should both be missing from persistent store + * but present in the backup store. + * + * This also checks that the backup store isn't being cleared before the Channel has a chance to retrieve a backup + * object and send it. + */ + public void testConsecutiveSaveDeleteObjects() { + Note note1 = mBucket.newObject(); + note1.setTitle("Hello World"); + + Note note2 = mBucket.newObject(); + note2.setTitle("Hello Again World"); + + note1.save(); + note2.save(); + + note1.delete(); + note2.delete(); + + // Test retrieving notes from persistent store + BucketObjectMissingException note1MissingException = null; + BucketObjectMissingException note2MissingException = null; + try { + mBucket.getObject(note1.getSimperiumKey()); + } catch (BucketObjectMissingException e) { + note1MissingException = e; + } + try { + mBucket.getObject(note2.getSimperiumKey()); + } catch (BucketObjectMissingException e) { + note2MissingException = e; + } + // Retrieval from persistent store should fail + assertNotNull(note1MissingException); + assertNotNull(note2MissingException); + + // Test retrieving notes from backup store + note1MissingException = null; + note2MissingException = null; + try { + mBucket.getObjectOrBackup(note1.getSimperiumKey()); + } catch (BucketObjectMissingException e) { + note1MissingException = e; + } + try { + mBucket.getObjectOrBackup(note2.getSimperiumKey()); + } catch (BucketObjectMissingException e) { + note2MissingException = e; + } + // Retrieval from backup store should succeed + assertNull(note1MissingException); + assertNull(note2MissingException); + } + } diff --git a/Simperium/src/androidTestSupport/java/com/simperium/ChannelTest.java b/Simperium/src/androidTestSupport/java/com/simperium/ChannelTest.java index 2d36cc71..8565da7f 100644 --- a/Simperium/src/androidTestSupport/java/com/simperium/ChannelTest.java +++ b/Simperium/src/androidTestSupport/java/com/simperium/ChannelTest.java @@ -99,7 +99,7 @@ protected void tearDown() throws Exception { * Some objects require a certain state before they can be deleted from a bucket so this ensures * the object is at the desired state before the delete operation is sent. */ - public void testSendFinalModificationBeforeDeleteOperation() throws Exception { + public void testQueueFinalModificationBeforeDeleteOperation() throws Exception { mListener.autoAcknowledge = true; @@ -135,6 +135,40 @@ public void testSendFinalModificationBeforeDeleteOperation() throws Exception { } + /** + * See https://github.com/Simperium/simperium-android/issues/159 + * Ensures that the channel is able to send out final modification operations for an object even if + * the object has since been removed from the local persistent store. + */ + public void testSendFinalModificationBeforeDeleteOperation() throws Exception { + + startWithEmptyIndex(); + + String key = "test-modify-before-delete-object"; + Note note = mBucket.newObject(key); + note.setTitle("Bonjour le monde!"); + + note.save(); + + // Queue a deletion and remove the object from the local persistent store before the modification has been sent + note.delete(); + + clearMessages(); + waitForMessage(); + + // Message should be a change message "c:{}" + assertMatchesRegex("^c:\\{.*\\}$", mListener.lastMessage.toString()); + + // First change has been sent and it's a modification + assertEquals(1, mChannelSerializer.queue.pending.size()); + assertTrue(mChannelSerializer.queue.pending.get(key).isModifyOperation()); + + // Second change is queued and it's a deletion + assertEquals(1, mChannelSerializer.queue.queued.size()); + assertTrue(mChannelSerializer.queue.queued.get(0).isRemoveOperation()); + + } + /** * Testing the receipt of cv:? from the server when using api 1.1 */ diff --git a/Simperium/src/androidTestSupport/java/com/simperium/ConcurrencyTest.java b/Simperium/src/androidTestSupport/java/com/simperium/ConcurrencyTest.java new file mode 100644 index 00000000..2342370e --- /dev/null +++ b/Simperium/src/androidTestSupport/java/com/simperium/ConcurrencyTest.java @@ -0,0 +1,308 @@ +package com.simperium; + +import com.simperium.client.Bucket; +import com.simperium.client.BucketObjectMissingException; +import com.simperium.client.BucketSchema; +import com.simperium.client.Channel; +import com.simperium.client.GhostStorageProvider; +import com.simperium.client.User; +import com.simperium.models.Note; +import com.simperium.storage.MemoryStore; +import com.simperium.test.MockChannelListener; +import com.simperium.test.MockChannelSerializer; +import com.simperium.test.MockGhostStore; +import com.simperium.util.Logger; + +import org.json.JSONArray; +import org.json.JSONObject; + +import java.util.Map; +import java.util.concurrent.Executors; +import java.util.concurrent.ThreadPoolExecutor; + +import static android.test.MoreAsserts.assertMatchesRegex; +import static com.simperium.TestHelpers.makeUser; +import static com.simperium.TestHelpers.waitUntil; + +public class ConcurrencyTest extends BaseSimperiumTest { + public static final String TAG = "SimperiumTest"; + + private static String BUCKET_NAME = "concurrency-test"; + public static String SESSION_ID = "SESSION-ID"; + public static String APP_ID = "APP_ID"; + + private Bucket mBucket; + private Channel mChannel; + private MockChannelSerializer mChannelSerializer = new MockChannelSerializer(); + final private MockChannelListener mListener = new MockChannelListener(); + + private BucketSchema mSchema; + private User mUser; + private MemoryStore mStorage; + private GhostStorageProvider mGhostStore; + + protected User.Status mAuthStatus; + + private ThreadPoolExecutor mExecutor; + + /** + * Build Bucket and Channel instances using a multi-threaded Executor + */ + protected void setUp() throws Exception { + super.setUp(); + + // Mimic AndroidClient's Executor setup + int threads = Runtime.getRuntime().availableProcessors(); + if (threads > 1) { + threads -= 1; + } + + mExecutor = (ThreadPoolExecutor) Executors.newFixedThreadPool(threads); + + mUser = makeUser(); + mSchema = new Note.Schema(); + mStorage = new MemoryStore(); + mGhostStore = new MockGhostStore(); + mBucket = new Bucket<>(mExecutor, BUCKET_NAME, mSchema, mUser, mStorage.createStore(BUCKET_NAME, mSchema), mGhostStore); + + mChannel = new Channel(mExecutor, APP_ID, SESSION_ID, mBucket, mChannelSerializer, mListener); + mBucket.setChannel(mChannel); + + mBucket.getUser().setStatusChangeListener(new User.StatusChangeListener(){ + @Override + public void onUserStatusChange(User.Status status){ + mAuthStatus = status; + } + }); + + } + + protected void tearDown() throws Exception { + mChannel.stop(); + mChannel.reset(); + super.tearDown(); + } + + /** + * Same as ChannelTest.testSendFinalModificationBeforeDeleteOperation() but with concurrency enabled. + * + * See https://github.com/Simperium/simperium-android/issues/159 + * Ensures that the channel is able to send out final modification operations for an object even if + * the object has since been removed from the local persistent store. + */ + public void testSendFinalModificationBeforeDeleteOperationConcurrent() throws Exception { + + startWithEmptyIndex(); + + String key = "test-modify-before-delete-object"; + Note note = mBucket.newObject(key); + note.setTitle("Bonjour le monde!"); + + note.save(); + + // Queue a deletion and remove the object from the local persistent store before the modification has been sent + note.delete(); + + clearMessages(); + waitForMessage(); + + // Message should be a change message "c:{}" + assertMatchesRegex("^c:\\{.*\\}$", mListener.lastMessage.toString()); + + // First change has been sent and it's a modification + assertEquals(1, mChannelSerializer.queue.pending.size()); + assertTrue(mChannelSerializer.queue.pending.get(key).isModifyOperation()); + + // Second change is queued and it's a deletion + assertEquals(1, mChannelSerializer.queue.queued.size()); + assertTrue(mChannelSerializer.queue.queued.get(0).isRemoveOperation()); + + } + + /** + * Same as BucketTest.testConsecutiveSaveDeleteObjects() but with concurrency enabled. + * + * If two different notes are both saved and then both deleted, they should both be missing from persistent store + * but present in the backup store. + * + * This also checks that the backup store isn't being cleared before the Channel has a chance to retrieve a backup + * object and send it. + */ + public void testConsecutiveSaveDeleteObjectsConcurrent() throws InterruptedException { + Note note1 = mBucket.newObject(); + note1.setTitle("Hello World"); + + Note note2 = mBucket.newObject(); + note2.setTitle("Hello Again World"); + + note1.save(); + note2.save(); + + note1.delete(); + note2.delete(); + + // Allow the save and delete tasks to finish before querying storage + waitForExecutorCompletedTasks(4); + + // Test retrieving notes from persistent store + BucketObjectMissingException note1MissingException = null; + BucketObjectMissingException note2MissingException = null; + try { + mBucket.getObject(note1.getSimperiumKey()); + } catch (BucketObjectMissingException e) { + note1MissingException = e; + } + try { + mBucket.getObject(note2.getSimperiumKey()); + } catch (BucketObjectMissingException e) { + note2MissingException = e; + } + // Retrieval from persistent store should fail + assertNotNull(note1MissingException); + assertNotNull(note2MissingException); + + // Test retrieving notes from backup store + note1MissingException = null; + note2MissingException = null; + try { + mBucket.getObjectOrBackup(note1.getSimperiumKey()); + } catch (BucketObjectMissingException e) { + note1MissingException = e; + } + try { + mBucket.getObjectOrBackup(note2.getSimperiumKey()); + } catch (BucketObjectMissingException e) { + note2MissingException = e; + } + // Retrieval from backup store should succeed + assertNull(note1MissingException); + assertNull(note2MissingException); + } + + + /** + * Gets the channel into a started state + */ + protected void start(){ + mChannel.onConnect(); + mChannel.start(); + // send auth success message + mChannel.receiveMessage("auth:user@example.com"); + } + + protected void startWithEmptyIndex() + throws Exception { + start(); + sendEmptyIndex(); + } + + protected void startWithIndex(Map objects) + throws Exception { + startWithIndex("mock-cv", objects); + } + + protected void startWithIndex(String cv, Map objects) + throws Exception { + start(); + sendIndex(cv, objects); + } + + /** + * Simulates a brand new bucket with and empty index + */ + protected void sendEmptyIndex() + throws Exception { + sendMessage("i:{\"index\":[]}"); + waitForIndex(); + } + + protected void sendIndex(String cv, Map objects) + throws Exception { + JSONObject index = new JSONObject(); + String period = "."; + index.put("current", cv); + + JSONArray versions = new JSONArray(); + index.put("index", versions); + + for(Map.Entry entry : objects.entrySet()) { + String key = entry.getKey(); + int dot = key.indexOf(period); + String id = key.substring(0, dot); + int version = Integer.parseInt(key.substring(dot+1)); + JSONObject versionData = new JSONObject(); + versionData.put("v", version); + versionData.put("id", id); + + versions.put(versionData); + } + + mListener.indexVersions = versions; + mListener.indexData = objects; + sendMessage(String.format("i:%s", index)); + waitForIndex(); + } + + protected void sendMessage(String message){ + mChannel.receiveMessage(message); + } + + protected Channel.MessageEvent waitForMessage() throws InterruptedException { + return waitForMessage(2000); + } + + /** + * Wait until a message received. More than likely clearMessages() should + * be called before waitForMessage() + */ + protected Channel.MessageEvent waitForMessage(int waitFor) throws InterruptedException { + + NewMessageFlagger flagger = new NewMessageFlagger(); + + waitUntil(flagger, "No message received", waitFor); + + return flagger.message; + + } + + /** + * Empties the list of received messages and sets last message to null + */ + protected void clearMessages(){ + mListener.clearMessages(); + } + + protected void waitForIndex() + throws InterruptedException { + waitUntil(new TestHelpers.Flag(){ + @Override + public boolean isComplete(){ + return mChannel.haveCompleteIndex(); + } + }, "Index never received", 5000); + } + + protected void waitForExecutorCompletedTasks(final int completedTasks) throws InterruptedException { + waitUntil(new TestHelpers.Flag(){ + @Override + public boolean isComplete(){ + return (mExecutor.getCompletedTaskCount() == completedTasks); + } + }, "Completed task amount never reached", 5000); + } + + private class NewMessageFlagger implements TestHelpers.Flag { + + Channel.MessageEvent message; + + @Override + public boolean isComplete() { + + message = mListener.lastMessage; + + return mListener.lastMessage != null; + + } + + } +} diff --git a/Simperium/src/main/java/com/simperium/client/Bucket.java b/Simperium/src/main/java/com/simperium/client/Bucket.java index 0be5f06d..d9a19c5d 100644 --- a/Simperium/src/main/java/com/simperium/client/Bucket.java +++ b/Simperium/src/main/java/com/simperium/client/Bucket.java @@ -33,8 +33,10 @@ import org.json.JSONObject; import java.util.Collections; +import java.util.HashMap; import java.util.HashSet; import java.util.Iterator; +import java.util.Map; import java.util.Set; import java.util.concurrent.Executor; @@ -47,6 +49,7 @@ public interface Channel { public void start(); public void stop(); public void reset(); + public boolean isIdle(); } public interface OnBeforeUpdateObjectListener { @@ -71,11 +74,61 @@ public interface Listener extends // implements all listener methods } + /** + * A one-way exclusion lock that stores multiple keys in a Set. Designed for use with two types of threads, + * primary and secondary, which share a key. + * + * Primary threads control locking by creating a key using lock(E) and deleting it using + * unlock(E). + * + * Secondary threads check whether their key is safe to use using checkLocked(E). If their key is + * present it's not safe to use it, and the threads are blocked until the key is removed from the set. + * + * @param the type of keys maintained + */ + public class LockSet { + private final Set mKeys = new HashSet<>(); + + /** + * Create a new key, blocking secondary threads using that key until unlock(E) is called + */ + public synchronized void lock(E key) { + mKeys.add(key); + } + + /** + * Block calling thread until lock is released + */ + public synchronized void checkLocked(E key) throws InterruptedException { + while (mKeys.contains(key)) { + wait(); + } + } + + /** + * Remove a key from the map, releasing the lock + */ + public synchronized void unlock(E key){ + mKeys.remove(key); + notify(); + } + + /** + * Return lock status for given key + */ + public synchronized boolean isLocked(E key){ + return mKeys.contains(key); + } + } + public enum ChangeType { REMOVE, MODIFY, INDEX, RESET, INSERT } public static final String TAG="Simperium.Bucket"; + + private static final int BACKUP_STORE_RESET_DELAY = 5000; + // The name used for the Simperium namespace private String mName; // User provides the access token for authentication @@ -96,6 +149,9 @@ public enum ChangeType { private BucketSchema mSchema; private GhostStorageProvider mGhostStore; final private Executor mExecutor; + private final Map mBackupStore = new TimestampHashMap<>(BACKUP_STORE_RESET_DELAY); + + private final LockSet mSaveDeleteLock = new LockSet<>(); /** * Represents a Simperium bucket which is a namespace where an app syncs a user's data @@ -184,22 +240,66 @@ public T getObject() { } + /** + * A HashMap which maintains a timestamp of its last put call and only performs remove(Object) + * and clear() operations if a long enough interval has elapsed. + */ + public class TimestampHashMap extends HashMap { + private long mTimestamp; + private long mClearDelay; + + public TimestampHashMap(long clearDelay) { + mClearDelay = clearDelay; + } + + @Override + public V put(K key, V value) { + V result = super.put(key, value); + mTimestamp = System.currentTimeMillis(); + return result; + } + + @Override + public V remove(Object key) { + V object = null; + if ((System.currentTimeMillis() - mTimestamp) > mClearDelay) { + object = super.remove(key); + } + return object; + } + + @Override + public void clear() { + if ((System.currentTimeMillis() - mTimestamp) > mClearDelay) { + super.clear(); + } + } + } + /** * Tell the bucket to sync changes. */ public void sync(final T object) { + mSaveDeleteLock.lock(object.getSimperiumKey()); mExecutor.execute(new Runnable() { @Override public void run() { - Boolean modified = object.isModified(); - mStorage.save(object, mSchema.indexesFor(object)); + try { + Boolean modified = object.isModified(); + mStorage.save(object, mSchema.indexesFor(object)); + + // Save a copy in case the object is removed from storage before this modification has been processed + storeBackupCopy(object); - mChannel.queueLocalChange(object); + mChannel.queueLocalChange(object); - if (modified) { - // Notify listeners that an object has been saved, this was - // triggered locally - notifyOnSaveListeners(object); + if (modified) { + // Notify listeners that an object has been saved, this was + // triggered locally + notifyOnSaveListeners(object); + } + } finally { + mSaveDeleteLock.unlock(object.getSimperiumKey()); } } }); @@ -226,6 +326,17 @@ private void remove(final T object, final boolean isLocal) { @Override public void run() { if (isLocal) { + try { + mSaveDeleteLock.checkLocked(object.getSimperiumKey()); + } catch (InterruptedException e) { + Logger.log(TAG, String.format("Delete lock interrupted for %s, did not remove from storage", + object.getSimperiumKey())); + // Proceed, but don't remove from storage + mChannel.queueLocalDeletion(object); + notifyOnDeleteListeners(object); + return; + } + mChannel.queueLocalDeletion(object); } @@ -250,6 +361,19 @@ private void removeObjectWithKey(String key) } } + /** + * Store a copy of the object in the backup store + */ + private void storeBackupCopy(T object) { + synchronized(mBackupStore) { + if (mChannel.isIdle()) { + // If there is no activity in the Channel, we can try to clean up obsolete backups + mBackupStore.clear(); + } + mBackupStore.put(object.getSimperiumKey(), object); + } + } + /** * Get the bucket's namespace * @return (String) bucket's namespace @@ -365,6 +489,27 @@ public T get(String key) throws BucketObjectMissingException { Logger.log(TAG, String.format("Fetched ghost for %s %s", key, ghost)); object.setBucket(this); object.setGhost(ghost); + updateBackupStoreGhost(ghost); + return object; + } + + /** + * Get an object by its key, checking the backup store if the object has been removed from the persistent store + */ + public T getObjectOrBackup(String key) throws BucketObjectMissingException { + T object; + try { + object = get(key); + } catch (BucketObjectMissingException e) { + // If the object has been removed from the persistent store, check the backup store + object = mBackupStore.get(key); + if (object == null) { + throw(new BucketObjectMissingException(String.format( + "Storage provider for bucket:%s did not have object %s and there was no stored backup", + getName(), key))); + } + Logger.log(TAG, String.format("Fetched backup copy for %s", key)); + } return object; } @@ -520,6 +665,16 @@ public void run() { }); } + /** + * Update the ghost of an object in the backup store + */ + private void updateBackupStoreGhost(Ghost ghost) { + T object = mBackupStore.get(ghost.getSimperiumKey()); + if (object != null) { + object.setGhost(ghost); + } + } + public Ghost getGhost(String key) throws GhostMissingException { return mGhostStore.getGhost(this, key); } @@ -743,12 +898,13 @@ public Ghost acknowledgeChange(RemoteChange remoteChange, Change change) Ghost ghost = null; if (!remoteChange.isRemoveOperation()) { try { - T object = get(remoteChange.getKey()); + T object = getObjectOrBackup(remoteChange.getKey()); // apply the diff to the underyling object ghost = remoteChange.apply(object.getGhost()); mGhostStore.saveGhost(this, ghost); // update the object's ghost object.setGhost(ghost); + updateBackupStoreGhost(ghost); } catch (BucketObjectMissingException e) { throw(new RemoteChangeInvalidException(remoteChange, e)); } @@ -780,7 +936,7 @@ public Ghost applyRemoteChange(RemoteChange change) object = newObject(change.getKey()); isNew = true; } else { - object = getObject(change.getKey()); + object = getObjectOrBackup(change.getKey()); isNew = false; notifyOnBeforeUpdateObjectListeners(object); @@ -803,6 +959,7 @@ public Ghost applyRemoteChange(RemoteChange change) // persist the ghost to storage mGhostStore.saveGhost(this, updatedGhost); object.setGhost(updatedGhost); + updateBackupStoreGhost(updatedGhost); // allow the schema to update the object instance with the new if (isNew) { diff --git a/Simperium/src/main/java/com/simperium/client/Channel.java b/Simperium/src/main/java/com/simperium/client/Channel.java index b8dc81b7..41711674 100644 --- a/Simperium/src/main/java/com/simperium/client/Channel.java +++ b/Simperium/src/main/java/com/simperium/client/Channel.java @@ -1484,7 +1484,7 @@ private void sendChange(Change change) try { log(LOG_DEBUG, String.format("Sending change for id: %s op: %s ccid: %s", change.getKey(), change.getOperation(), change.getChangeId())); - Syncable target = mBucket.getObject(change.getKey()); + Syncable target = mBucket.getObjectOrBackup(change.getKey()); Ghost ghost = mBucket.getGhost(change.getKey()); sendMessage(String.format("c:%s", change.toJSONObject(target.getDiffableValue(), ghost))); mSerializer.onSendChange(change); diff --git a/Simperium/src/support/java/com/simperium/test/MockChannel.java b/Simperium/src/support/java/com/simperium/test/MockChannel.java index c23b8e1a..21854582 100644 --- a/Simperium/src/support/java/com/simperium/test/MockChannel.java +++ b/Simperium/src/support/java/com/simperium/test/MockChannel.java @@ -23,6 +23,7 @@ public class MockChannel implements Bucket.Channel { private Bucket mBucket; private boolean started = false; + private boolean idle = false; public boolean autoAcknowledge = true; @@ -41,6 +42,11 @@ public Change queueLocalDeletion(Syncable object){ return change; } + @Override + public boolean isIdle() { + return idle; + } + @Override public Change queueLocalChange(Syncable object) { Change change = new Change(Change.OPERATION_MODIFY, object);