Merge branch '2294-contact-list-worker' into 'master'

Mailbox worker for updating our own mailbox's contact list

Closes #2294

See merge request briar/briar!1677

bramble-api/src/main/java/org/briarproject/bramble/api/mailbox/MailboxUpdateManager.java

@@ -63,9 +63,26 @@ public interface MailboxUpdateManager {
 	 */
 	String GROUP_KEY_SENT_CLIENT_SUPPORTS = "sentClientSupports";
 
+	/**
+	 * Returns the latest {@link MailboxUpdate} sent to the given contact.
+	 * <p>
+	 * If we have our own mailbox then the update will be a
+	 * {@link MailboxUpdateWithMailbox} containing the
+	 * {@link MailboxProperties} the contact should use for communicating with
+	 * our mailbox.
+	 */
 	MailboxUpdate getLocalUpdate(Transaction txn, ContactId c)
 			throws DbException;
 
+	/**
+	 * Returns the latest {@link MailboxUpdate} received from the given
+	 * contact, or null if no update has been received.
+	 * <p>
+	 * If the contact has a mailbox then the update will be a
+	 * {@link MailboxUpdateWithMailbox} containing the
+	 * {@link MailboxProperties} we should use for communicating with the
+	 * contact's mailbox.
+	 */
 	@Nullable
 	MailboxUpdate getRemoteUpdate(Transaction txn, ContactId c)
 			throws DbException;

bramble-core/src/main/java/org/briarproject/bramble/mailbox/OwnMailboxConnectivityChecker.java

@@ -4,11 +4,13 @@ import org.briarproject.bramble.api.db.DbException;
 import org.briarproject.bramble.api.db.TransactionManager;
 import org.briarproject.bramble.api.mailbox.MailboxProperties;
 import org.briarproject.bramble.api.mailbox.MailboxSettingsManager;
+import org.briarproject.bramble.api.mailbox.MailboxVersion;
 import org.briarproject.bramble.api.nullsafety.NotNullByDefault;
 import org.briarproject.bramble.api.system.Clock;
 import org.briarproject.bramble.mailbox.MailboxApi.ApiException;
 
 import java.io.IOException;
+import java.util.List;
 import java.util.logging.Logger;
 
 import javax.annotation.concurrent.ThreadSafe;
@@ -55,11 +57,12 @@ class OwnMailboxConnectivityChecker extends ConnectivityCheckerImpl {
 	private boolean checkConnectivityAndStoreResult(
 			MailboxProperties properties) throws DbException {
 		try {
-			if (!mailboxApi.checkStatus(properties)) throw new ApiException();
+			List<MailboxVersion> serverSupports =
+					mailboxApi.getServerSupports(properties);
 			LOG.info("Own mailbox is reachable");
 			long now = clock.currentTimeMillis();
 			db.transaction(false, txn -> mailboxSettingsManager
-					.recordSuccessfulConnection(txn, now));
+					.recordSuccessfulConnection(txn, now, serverSupports));
 			// Call the observers and cache the result
 			onConnectivityCheckSucceeded(now);
 			return false; // Don't retry

bramble-core/src/main/java/org/briarproject/bramble/mailbox/OwnMailboxContactListWorker.java

@@ -0,0 +1,369 @@
+package org.briarproject.bramble.mailbox;
+
+import org.briarproject.bramble.api.Cancellable;
+import org.briarproject.bramble.api.contact.Contact;
+import org.briarproject.bramble.api.contact.ContactId;
+import org.briarproject.bramble.api.contact.event.ContactAddedEvent;
+import org.briarproject.bramble.api.contact.event.ContactRemovedEvent;
+import org.briarproject.bramble.api.db.DatabaseComponent;
+import org.briarproject.bramble.api.db.DbException;
+import org.briarproject.bramble.api.db.NoSuchContactException;
+import org.briarproject.bramble.api.event.Event;
+import org.briarproject.bramble.api.event.EventBus;
+import org.briarproject.bramble.api.event.EventExecutor;
+import org.briarproject.bramble.api.event.EventListener;
+import org.briarproject.bramble.api.lifecycle.IoExecutor;
+import org.briarproject.bramble.api.mailbox.MailboxProperties;
+import org.briarproject.bramble.api.mailbox.MailboxUpdate;
+import org.briarproject.bramble.api.mailbox.MailboxUpdateManager;
+import org.briarproject.bramble.api.mailbox.MailboxUpdateWithMailbox;
+import org.briarproject.bramble.api.nullsafety.NotNullByDefault;
+import org.briarproject.bramble.mailbox.ConnectivityChecker.ConnectivityObserver;
+import org.briarproject.bramble.mailbox.MailboxApi.ApiException;
+import org.briarproject.bramble.mailbox.MailboxApi.MailboxContact;
+import org.briarproject.bramble.mailbox.MailboxApi.TolerableFailureException;
+
+import java.io.IOException;
+import java.util.Collection;
+import java.util.HashSet;
+import java.util.LinkedList;
+import java.util.Queue;
+import java.util.Set;
+import java.util.concurrent.Executor;
+import java.util.logging.Logger;
+
+import javax.annotation.Nullable;
+import javax.annotation.concurrent.GuardedBy;
+import javax.annotation.concurrent.ThreadSafe;
+
+import static java.util.logging.Level.INFO;
+import static java.util.logging.Level.WARNING;
+import static java.util.logging.Logger.getLogger;
+import static org.briarproject.bramble.api.nullsafety.NullSafety.requireNonNull;
+import static org.briarproject.bramble.util.LogUtils.logException;
+
+@ThreadSafe
+@NotNullByDefault
+class OwnMailboxContactListWorker
+		implements MailboxWorker, ConnectivityObserver, EventListener {
+
+	/**
+	 * When the worker is started it waits for a connectivity check, then
+	 * fetches the remote contact list and compares it to the local contact
+	 * list.
+	 * <p>
+	 * Any contacts that are missing from the remote list are added to the
+	 * mailbox's contact list, while any contacts that are missing from the
+	 * local list are removed from the mailbox's contact list.
+	 * <p>
+	 * Once the remote contact list has been brought up to date, the worker
+	 * waits for events indicating that contacts have been added or removed.
+	 * Each time an event is received, the worker updates the mailbox's
+	 * contact list and then goes back to waiting.
+	 */
+	private enum State {
+		CREATED,
+		CONNECTIVITY_CHECK,
+		FETCHING_CONTACT_LIST,
+		UPDATING_CONTACT_LIST,
+		WAITING_FOR_CHANGES,
+		DESTROYED
+	}
+
+	private static final Logger LOG =
+			getLogger(OwnMailboxContactListWorker.class.getName());
+
+	private final Executor ioExecutor;
+	private final DatabaseComponent db;
+	private final EventBus eventBus;
+	private final ConnectivityChecker connectivityChecker;
+	private final MailboxApiCaller mailboxApiCaller;
+	private final MailboxApi mailboxApi;
+	private final MailboxUpdateManager mailboxUpdateManager;
+	private final MailboxProperties mailboxProperties;
+	private final Object lock = new Object();
+
+	@GuardedBy("lock")
+	private State state = State.CREATED;
+
+	@GuardedBy("lock")
+	@Nullable
+	private Cancellable apiCall = null;
+
+	/**
+	 * A queue of updates waiting to be applied to the remote contact list.
+	 */
+	@GuardedBy("lock")
+	private final Queue<Update> updates = new LinkedList<>();
+
+	OwnMailboxContactListWorker(@IoExecutor Executor ioExecutor,
+			DatabaseComponent db,
+			EventBus eventBus,
+			ConnectivityChecker connectivityChecker,
+			MailboxApiCaller mailboxApiCaller,
+			MailboxApi mailboxApi,
+			MailboxUpdateManager mailboxUpdateManager,
+			MailboxProperties mailboxProperties) {
+		if (!mailboxProperties.isOwner()) throw new IllegalArgumentException();
+		this.ioExecutor = ioExecutor;
+		this.db = db;
+		this.connectivityChecker = connectivityChecker;
+		this.mailboxApiCaller = mailboxApiCaller;
+		this.mailboxApi = mailboxApi;
+		this.mailboxUpdateManager = mailboxUpdateManager;
+		this.mailboxProperties = mailboxProperties;
+		this.eventBus = eventBus;
+	}
+
+	@Override
+	public void start() {
+		LOG.info("Started");
+		synchronized (lock) {
+			if (state != State.CREATED) return;
+			state = State.CONNECTIVITY_CHECK;
+		}
+		// Avoid leaking observer in case destroy() is called concurrently
+		// before observer is added
+		connectivityChecker.checkConnectivity(mailboxProperties, this);
+		boolean destroyed;
+		synchronized (lock) {
+			destroyed = state == State.DESTROYED;
+		}
+		if (destroyed) connectivityChecker.removeObserver(this);
+	}
+
+	@Override
+	public void destroy() {
+		LOG.info("Destroyed");
+		Cancellable apiCall;
+		synchronized (lock) {
+			state = State.DESTROYED;
+			apiCall = this.apiCall;
+			this.apiCall = null;
+		}
+		if (apiCall != null) apiCall.cancel();
+		connectivityChecker.removeObserver(this);
+		eventBus.removeListener(this);
+	}
+
+	@Override
+	public void onConnectivityCheckSucceeded() {
+		LOG.info("Connectivity check succeeded");
+		synchronized (lock) {
+			if (state != State.CONNECTIVITY_CHECK) return;
+			state = State.FETCHING_CONTACT_LIST;
+			apiCall = mailboxApiCaller.retryWithBackoff(
+					new SimpleApiCall(this::apiCallFetchContactList));
+		}
+	}
+
+	@IoExecutor
+	private void apiCallFetchContactList() throws IOException, ApiException {
+		synchronized (lock) {
+			if (state != State.FETCHING_CONTACT_LIST) return;
+		}
+		LOG.info("Fetching remote contact list");
+		Collection<ContactId> remote =
+				mailboxApi.getContacts(mailboxProperties);
+		ioExecutor.execute(() -> loadLocalContactList(remote));
+	}
+
+	@IoExecutor
+	private void loadLocalContactList(Collection<ContactId> remote) {
+		synchronized (lock) {
+			if (state != State.FETCHING_CONTACT_LIST) return;
+			apiCall = null;
+		}
+		LOG.info("Loading local contact list");
+		try {
+			db.transaction(true, txn -> {
+				Collection<Contact> local = db.getContacts(txn);
+				// Handle the result on the event executor to avoid races with
+				// incoming events
+				txn.attach(() -> reconcileContactLists(local, remote));
+			});
+		} catch (DbException e) {
+			logException(LOG, WARNING, e);
+		}
+	}
+
+	@EventExecutor
+	private void reconcileContactLists(Collection<Contact> local,
+			Collection<ContactId> remote) {
+		Set<ContactId> localIds = new HashSet<>();
+		for (Contact c : local) localIds.add(c.getId());
+		remote = new HashSet<>(remote);
+		synchronized (lock) {
+			if (state != State.FETCHING_CONTACT_LIST) return;
+			for (ContactId c : localIds) {
+				if (!remote.contains(c)) updates.add(new Update(true, c));
+			}
+			for (ContactId c : remote) {
+				if (!localIds.contains(c)) updates.add(new Update(false, c));
+			}
+			if (updates.isEmpty()) {
+				LOG.info("Contact list is up to date");
+				state = State.WAITING_FOR_CHANGES;
+			} else {
+				if (LOG.isLoggable(INFO)) {
+					LOG.info(updates.size() + " updates to apply");
+				}
+				state = State.UPDATING_CONTACT_LIST;
+				ioExecutor.execute(this::updateContactList);
+			}
+		}
+	}
+
+	@IoExecutor
+	private void updateContactList() {
+		Update update;
+		synchronized (lock) {
+			if (state != State.UPDATING_CONTACT_LIST) return;
+			update = updates.poll();
+			if (update == null) {
+				LOG.info("No more updates to process");
+				state = State.WAITING_FOR_CHANGES;
+				apiCall = null;
+				return;
+			}
+		}
+		if (update.add) loadMailboxProperties(update.contactId);
+		else removeContact(update.contactId);
+	}
+
+	@IoExecutor
+	private void loadMailboxProperties(ContactId c) {
+		synchronized (lock) {
+			if (state != State.UPDATING_CONTACT_LIST) return;
+		}
+		LOG.info("Loading mailbox properties for contact");
+		try {
+			MailboxUpdate mailboxUpdate = db.transactionWithResult(true, txn ->
+					mailboxUpdateManager.getLocalUpdate(txn, c));
+			if (mailboxUpdate instanceof MailboxUpdateWithMailbox) {
+				addContact(c, (MailboxUpdateWithMailbox) mailboxUpdate);
+			} else {
+				// Our own mailbox was concurrently unpaired. This worker will
+				// be destroyed soon, so we can stop here
+				LOG.info("Own mailbox was unpaired");
+			}
+		} catch (NoSuchContactException e) {
+			// Contact was removed concurrently. Move on to the next update.
+			// Later we may process a removal update for this contact, which
+			// was never added to the mailbox's contact list. The removal API
+			// call should fail safely with a TolerableFailureException
+			LOG.info("No such contact");
+			updateContactList();
+		} catch (DbException e) {
+			logException(LOG, WARNING, e);
+		}
+	}
+
+	@IoExecutor
+	private void addContact(ContactId c, MailboxUpdateWithMailbox withMailbox) {
+		MailboxProperties props = withMailbox.getMailboxProperties();
+		MailboxContact contact = new MailboxContact(c, props.getAuthToken(),
+				requireNonNull(props.getInboxId()),
+				requireNonNull(props.getOutboxId()));
+		synchronized (lock) {
+			if (state != State.UPDATING_CONTACT_LIST) return;
+			apiCall = mailboxApiCaller.retryWithBackoff(new SimpleApiCall(() ->
+					apiCallAddContact(contact)));
+		}
+	}
+
+	@IoExecutor
+	private void apiCallAddContact(MailboxContact contact)
+			throws IOException, ApiException, TolerableFailureException {
+		synchronized (lock) {
+			if (state != State.UPDATING_CONTACT_LIST) return;
+		}
+		LOG.info("Adding contact to remote contact list");
+		mailboxApi.addContact(mailboxProperties, contact);
+		updateContactList();
+	}
+
+	@IoExecutor
+	private void removeContact(ContactId c) {
+		synchronized (lock) {
+			if (state != State.UPDATING_CONTACT_LIST) return;
+			apiCall = mailboxApiCaller.retryWithBackoff(new SimpleApiCall(() ->
+					apiCallRemoveContact(c)));
+		}
+	}
+
+	@IoExecutor
+	private void apiCallRemoveContact(ContactId c)
+			throws IOException, ApiException {
+		synchronized (lock) {
+			if (state != State.UPDATING_CONTACT_LIST) return;
+		}
+		LOG.info("Removing contact from remote contact list");
+		try {
+			mailboxApi.deleteContact(mailboxProperties, c);
+		} catch (TolerableFailureException e) {
+			// Catch this so we can continue to the next update
+			logException(LOG, INFO, e);
+		}
+		updateContactList();
+	}
+
+	@Override
+	public void eventOccurred(Event e) {
+		if (e instanceof ContactAddedEvent) {
+			LOG.info("Contact added");
+			onContactAdded(((ContactAddedEvent) e).getContactId());
+		} else if (e instanceof ContactRemovedEvent) {
+			LOG.info("Contact removed");
+			onContactRemoved(((ContactRemovedEvent) e).getContactId());
+		}
+	}
+
+	@EventExecutor
+	private void onContactAdded(ContactId c) {
+		synchronized (lock) {
+			if (state != State.UPDATING_CONTACT_LIST &&
+					state != State.WAITING_FOR_CHANGES) {
+				return;
+			}
+			updates.add(new Update(true, c));
+			if (state == State.WAITING_FOR_CHANGES) {
+				state = State.UPDATING_CONTACT_LIST;
+				ioExecutor.execute(this::updateContactList);
+			}
+		}
+	}
+
+	@EventExecutor
+	private void onContactRemoved(ContactId c) {
+		synchronized (lock) {
+			if (state != State.UPDATING_CONTACT_LIST &&
+					state != State.WAITING_FOR_CHANGES) {
+				return;
+			}
+			updates.add(new Update(false, c));
+			if (state == State.WAITING_FOR_CHANGES) {
+				state = State.UPDATING_CONTACT_LIST;
+				ioExecutor.execute(this::updateContactList);
+			}
+		}
+	}
+
+	/**
+	 * An update that should be applied to the remote contact list.
+	 */
+	private static class Update {
+
+		/**
+		 * True if the contact should be added, false if the contact should be
+		 * removed.
+		 */
+		private final boolean add;
+		private final ContactId contactId;
+
+		private Update(boolean add, ContactId contactId) {
+			this.add = add;
+			this.contactId = contactId;
+		}
+	}
+}

bramble-core/src/test/java/org/briarproject/bramble/mailbox/OwnMailboxConnectivityCheckerTest.java

@@ -5,6 +5,7 @@ import org.briarproject.bramble.api.db.Transaction;
 import org.briarproject.bramble.api.db.TransactionManager;
 import org.briarproject.bramble.api.mailbox.MailboxProperties;
 import org.briarproject.bramble.api.mailbox.MailboxSettingsManager;
+import org.briarproject.bramble.api.mailbox.MailboxVersion;
 import org.briarproject.bramble.api.system.Clock;
 import org.briarproject.bramble.mailbox.ConnectivityChecker.ConnectivityObserver;
 import org.briarproject.bramble.test.BrambleMockTestCase;
@@ -15,8 +16,10 @@ import org.jmock.lib.action.DoAllAction;
 import org.junit.Test;
 
 import java.io.IOException;
+import java.util.List;
 import java.util.concurrent.atomic.AtomicReference;
 
+import static java.util.Collections.singletonList;
 import static org.briarproject.bramble.api.mailbox.MailboxConstants.CLIENT_SUPPORTS;
 import static org.briarproject.bramble.test.TestUtils.getMailboxProperties;
 import static org.junit.Assert.assertFalse;
@@ -39,6 +42,8 @@ public class OwnMailboxConnectivityCheckerTest extends BrambleMockTestCase {
 	private final MailboxProperties properties =
 			getMailboxProperties(true, CLIENT_SUPPORTS);
 	private final long now = System.currentTimeMillis();
+	private final List<MailboxVersion> serverSupports =
+			singletonList(new MailboxVersion(123, 456));
 
 	@Test
 	public void testObserverIsCalledWhenCheckSucceeds() throws Exception {
@@ -62,12 +67,13 @@ public class OwnMailboxConnectivityCheckerTest extends BrambleMockTestCase {
 		// When the check succeeds, the success should be recorded in the DB
 		// and the observer should be called
 		context.checking(new DbExpectations() {{
-			oneOf(mailboxApi).checkStatus(properties);
+			oneOf(mailboxApi).getServerSupports(properties);
-			will(returnValue(true));
+			will(returnValue(serverSupports));
 			oneOf(clock).currentTimeMillis();
 			will(returnValue(now));
 			oneOf(db).transaction(with(false), withDbRunnable(txn));
-			oneOf(mailboxSettingsManager).recordSuccessfulConnection(txn, now);
+			oneOf(mailboxSettingsManager).recordSuccessfulConnection(txn, now,
+					serverSupports);
 			oneOf(observer).onConnectivityCheckSucceeded();
 		}});
 
@@ -97,7 +103,7 @@ public class OwnMailboxConnectivityCheckerTest extends BrambleMockTestCase {
 		// When the check fails, the failure should be recorded in the DB and
 		// the observer should not be called
 		context.checking(new DbExpectations() {{
-			oneOf(mailboxApi).checkStatus(properties);
+			oneOf(mailboxApi).getServerSupports(properties);
 			will(throwException(new IOException()));
 			oneOf(clock).currentTimeMillis();
 			will(returnValue(now));

bramble-core/src/test/java/org/briarproject/bramble/mailbox/OwnMailboxContactListWorkerTest.java

@@ -0,0 +1,406 @@
+package org.briarproject.bramble.mailbox;
+
+import org.briarproject.bramble.api.Cancellable;
+import org.briarproject.bramble.api.contact.Contact;
+import org.briarproject.bramble.api.contact.ContactId;
+import org.briarproject.bramble.api.contact.event.ContactAddedEvent;
+import org.briarproject.bramble.api.contact.event.ContactRemovedEvent;
+import org.briarproject.bramble.api.db.DatabaseComponent;
+import org.briarproject.bramble.api.db.NoSuchContactException;
+import org.briarproject.bramble.api.db.Transaction;
+import org.briarproject.bramble.api.event.EventBus;
+import org.briarproject.bramble.api.mailbox.MailboxProperties;
+import org.briarproject.bramble.api.mailbox.MailboxUpdate;
+import org.briarproject.bramble.api.mailbox.MailboxUpdateManager;
+import org.briarproject.bramble.api.mailbox.MailboxUpdateWithMailbox;
+import org.briarproject.bramble.mailbox.MailboxApi.MailboxContact;
+import org.briarproject.bramble.mailbox.MailboxApi.TolerableFailureException;
+import org.briarproject.bramble.test.BrambleMockTestCase;
+import org.briarproject.bramble.test.CaptureArgumentAction;
+import org.briarproject.bramble.test.DbExpectations;
+import org.briarproject.bramble.test.RunAction;
+import org.jmock.Expectations;
+import org.jmock.lib.action.DoAllAction;
+import org.junit.Test;
+
+import java.util.List;
+import java.util.concurrent.Executor;
+import java.util.concurrent.atomic.AtomicReference;
+
+import static java.util.Collections.emptyList;
+import static java.util.Collections.singletonList;
+import static org.briarproject.bramble.api.mailbox.MailboxConstants.CLIENT_SUPPORTS;
+import static org.briarproject.bramble.api.nullsafety.NullSafety.requireNonNull;
+import static org.briarproject.bramble.test.TestUtils.getContact;
+import static org.briarproject.bramble.test.TestUtils.getMailboxProperties;
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
+
+public class OwnMailboxContactListWorkerTest extends BrambleMockTestCase {
+
+	private final Executor ioExecutor = context.mock(Executor.class);
+	private final DatabaseComponent db = context.mock(DatabaseComponent.class);
+	private final EventBus eventBus = context.mock(EventBus.class);
+	private final ConnectivityChecker connectivityChecker =
+			context.mock(ConnectivityChecker.class);
+	private final MailboxApiCaller mailboxApiCaller =
+			context.mock(MailboxApiCaller.class);
+	private final MailboxApi mailboxApi = context.mock(MailboxApi.class);
+	private final MailboxUpdateManager mailboxUpdateManager =
+			context.mock(MailboxUpdateManager.class);
+	private final Cancellable apiCall = context.mock(Cancellable.class);
+
+	private final MailboxProperties mailboxProperties =
+			getMailboxProperties(true, CLIENT_SUPPORTS);
+	private final Contact contact1 = getContact(), contact2 = getContact();
+	private final MailboxProperties contactProperties1 =
+			getMailboxProperties(false, CLIENT_SUPPORTS);
+	private final MailboxProperties contactProperties2 =
+			getMailboxProperties(false, CLIENT_SUPPORTS);
+	private final MailboxUpdateWithMailbox update1 =
+			new MailboxUpdateWithMailbox(CLIENT_SUPPORTS, contactProperties1);
+	private final MailboxUpdateWithMailbox update2 =
+			new MailboxUpdateWithMailbox(CLIENT_SUPPORTS, contactProperties2);
+
+	private final OwnMailboxContactListWorker worker =
+			new OwnMailboxContactListWorker(ioExecutor, db, eventBus,
+					connectivityChecker, mailboxApiCaller, mailboxApi,
+					mailboxUpdateManager, mailboxProperties);
+
+	@Test
+	public void testChecksConnectivityWhenStartedAndRemovesObserverWhenDestroyed() {
+		// When the worker is started it should start a connectivity check
+		expectStartConnectivityCheck();
+
+		worker.start();
+
+		// When the worker is destroyed it should remove the connectivity
+		// observer and event listener
+		expectRemoveConnectivityObserverAndEventListener();
+
+		worker.destroy();
+	}
+
+	@Test
+	public void testUpdatesContactListWhenConnectivityCheckSucceeds()
+			throws Exception {
+		// When the worker is started it should start a connectivity check
+		expectStartConnectivityCheck();
+
+		worker.start();
+
+		// When the connectivity check succeeds, the worker should start a
+		// task to fetch the remote contact list
+		AtomicReference<ApiCall> fetchList = new AtomicReference<>();
+		expectStartTaskToFetchRemoteContactList(fetchList);
+
+		worker.onConnectivityCheckSucceeded();
+
+		// When the fetch task runs it should fetch the remote contact list,
+		// load the local contact list, and find the differences. Contact 2
+		// needs to be added and contact 1 needs to be removed. The worker
+		// should load the mailbox update for contact 2 and start a task to
+		// add contact 2 to the mailbox
+		expectFetchRemoteContactList(singletonList(contact1.getId()));
+		expectRunTaskOnIoExecutor();
+		expectLoadLocalContactList(singletonList(contact2));
+		expectRunTaskOnIoExecutor();
+		expectLoadMailboxUpdate(contact2, update2);
+		AtomicReference<ApiCall> addContact = new AtomicReference<>();
+		expectStartTaskToAddContact(addContact);
+
+		assertFalse(fetchList.get().callApi());
+
+		// When the add-contact task runs it should add contact 2 to the
+		// mailbox, then continue with the next update
+		AtomicReference<MailboxContact> added = new AtomicReference<>();
+		expectAddContactToMailbox(added);
+		AtomicReference<ApiCall> removeContact = new AtomicReference<>();
+		expectStartTaskToRemoveContact(removeContact);
+
+		assertFalse(addContact.get().callApi());
+
+		// Check that the added contact has the expected properties
+		MailboxContact expected = new MailboxContact(contact2.getId(),
+				contactProperties2.getAuthToken(),
+				requireNonNull(contactProperties2.getInboxId()),
+				requireNonNull(contactProperties2.getOutboxId()));
+		assertMailboxContactEquals(expected, added.get());
+
+		// When the remove-contact task runs it should remove contact 1 from
+		// the mailbox
+		expectRemoveContactFromMailbox(contact1);
+
+		assertFalse(removeContact.get().callApi());
+
+		// When the worker is destroyed it should remove the connectivity
+		// observer and event listener
+		expectRemoveConnectivityObserverAndEventListener();
+
+		worker.destroy();
+	}
+
+	@Test
+	public void testHandlesEventsAfterMakingInitialUpdates() throws Exception {
+		// When the worker is started it should start a connectivity check
+		expectStartConnectivityCheck();
+
+		worker.start();
+
+		// When the connectivity check succeeds, the worker should start a
+		// task to fetch the remote contact list
+		AtomicReference<ApiCall> fetchList = new AtomicReference<>();
+		expectStartTaskToFetchRemoteContactList(fetchList);
+
+		worker.onConnectivityCheckSucceeded();
+
+		// When the fetch task runs it should fetch the remote contact list,
+		// load the local contact list, and find the differences. The lists
+		// are the same, so the worker should wait for changes
+		expectFetchRemoteContactList(emptyList());
+		expectRunTaskOnIoExecutor();
+		expectLoadLocalContactList(emptyList());
+
+		assertFalse(fetchList.get().callApi());
+
+		// When a contact is added, the worker should load the contact's
+		// mailbox update and start a task to add the contact to the mailbox
+		expectRunTaskOnIoExecutor();
+		expectLoadMailboxUpdate(contact1, update1);
+		AtomicReference<ApiCall> addContact = new AtomicReference<>();
+		expectStartTaskToAddContact(addContact);
+
+		worker.eventOccurred(new ContactAddedEvent(contact1.getId(), true));
+
+		// When the add-contact task runs it should add contact 1 to the
+		// mailbox
+		AtomicReference<MailboxContact> added = new AtomicReference<>();
+		expectAddContactToMailbox(added);
+
+		assertFalse(addContact.get().callApi());
+
+		// Check that the added contact has the expected properties
+		MailboxContact expected = new MailboxContact(contact1.getId(),
+				contactProperties1.getAuthToken(),
+				requireNonNull(contactProperties1.getInboxId()),
+				requireNonNull(contactProperties1.getOutboxId()));
+		assertMailboxContactEquals(expected, added.get());
+
+		// When the contact is removed again, the worker should start a task
+		// to remove the contact from the mailbox
+		expectRunTaskOnIoExecutor();
+		AtomicReference<ApiCall> removeContact = new AtomicReference<>();
+		expectStartTaskToRemoveContact(removeContact);
+
+		worker.eventOccurred(new ContactRemovedEvent(contact1.getId()));
+
+		// When the remove-contact task runs it should remove the contact from
+		// the mailbox
+		expectRemoveContactFromMailbox(contact1);
+
+		assertFalse(removeContact.get().callApi());
+
+		// When the worker is destroyed it should remove the connectivity
+		// observer and event listener
+		expectRemoveConnectivityObserverAndEventListener();
+
+		worker.destroy();
+	}
+
+
+	@Test
+	public void testHandlesNoSuchContactException() throws Exception {
+		// When the worker is started it should start a connectivity check
+		expectStartConnectivityCheck();
+
+		worker.start();
+
+		// When the connectivity check succeeds, the worker should start a
+		// task to fetch the remote contact list
+		AtomicReference<ApiCall> fetchList = new AtomicReference<>();
+		expectStartTaskToFetchRemoteContactList(fetchList);
+
+		worker.onConnectivityCheckSucceeded();
+
+		// When the fetch task runs it should fetch the remote contact list,
+		// load the local contact list, and find the differences. Contact 1
+		// needs to be added, so the worker should submit a task to the
+		// IO executor to load the contact's mailbox update
+		expectFetchRemoteContactList(emptyList());
+		expectRunTaskOnIoExecutor();
+		expectLoadLocalContactList(singletonList(contact1));
+		AtomicReference<Runnable> loadUpdate = new AtomicReference<>();
+		context.checking(new Expectations() {{
+			oneOf(ioExecutor).execute(with(any(Runnable.class)));
+			will(new CaptureArgumentAction<>(loadUpdate, Runnable.class, 0));
+		}});
+
+		assertFalse(fetchList.get().callApi());
+
+		// Before the contact's mailbox update can be loaded, the contact
+		// is removed
+		worker.eventOccurred(new ContactRemovedEvent(contact1.getId()));
+
+		// When the load-update task runs, a NoSuchContactException is thrown.
+		// The worker should abandon adding the contact and move on to the
+		// next update, which is the removal of the same contact. The worker
+		// should start a task to remove the contact from the mailbox
+		Transaction txn = new Transaction(null, false);
+		context.checking(new DbExpectations() {{
+			oneOf(db).transactionWithResult(with(true), withDbCallable(txn));
+			oneOf(mailboxUpdateManager).getLocalUpdate(txn, contact1.getId());
+			will(throwException(new NoSuchContactException()));
+		}});
+		AtomicReference<ApiCall> removeContact = new AtomicReference<>();
+		expectStartTaskToRemoveContact(removeContact);
+
+		loadUpdate.get().run();
+
+		// When the remove-contact task runs it should remove the contact from
+		// the mailbox. The contact was never added, so the call throws a
+		// TolerableFailureException
+		context.checking(new Expectations() {{
+			oneOf(mailboxApi).deleteContact(mailboxProperties,
+					contact1.getId());
+			will(throwException(new TolerableFailureException()));
+		}});
+
+		assertFalse(removeContact.get().callApi());
+
+		// When the worker is destroyed it should remove the connectivity
+		// observer and event listener
+		expectRemoveConnectivityObserverAndEventListener();
+
+		worker.destroy();
+	}
+
+	@Test
+	public void testCancelsApiCallWhenDestroyed() {
+		// When the worker is started it should start a connectivity check
+		expectStartConnectivityCheck();
+
+		worker.start();
+
+		// When the connectivity check succeeds, the worker should start a
+		// task to fetch the remote contact list
+		AtomicReference<ApiCall> fetchList = new AtomicReference<>();
+		expectStartTaskToFetchRemoteContactList(fetchList);
+
+		worker.onConnectivityCheckSucceeded();
+
+		// The worker is destroyed before the task runs. The worker should
+		// cancel the task remove the connectivity observer and event listener
+		context.checking(new Expectations() {{
+			oneOf(apiCall).cancel();
+		}});
+		expectRemoveConnectivityObserverAndEventListener();
+
+		worker.destroy();
+	}
+
+	private void expectStartConnectivityCheck() {
+		context.checking(new Expectations() {{
+			oneOf(connectivityChecker).checkConnectivity(
+					with(mailboxProperties), with(worker));
+		}});
+	}
+
+	private void expectRemoveConnectivityObserverAndEventListener() {
+		context.checking(new Expectations() {{
+			oneOf(connectivityChecker).removeObserver(worker);
+			oneOf(eventBus).removeListener(worker);
+		}});
+	}
+
+	private void expectStartTaskToFetchRemoteContactList(
+			AtomicReference<ApiCall> task) {
+		context.checking(new Expectations() {{
+			oneOf(mailboxApiCaller).retryWithBackoff(with(any(ApiCall.class)));
+			will(new DoAllAction(
+					new CaptureArgumentAction<>(task, ApiCall.class, 0),
+					returnValue(apiCall)
+			));
+		}});
+	}
+
+	private void expectFetchRemoteContactList(List<ContactId> remote)
+			throws Exception {
+		context.checking(new Expectations() {{
+			oneOf(mailboxApi).getContacts(mailboxProperties);
+			will(returnValue(remote));
+		}});
+	}
+
+	private void expectLoadLocalContactList(List<Contact> local)
+			throws Exception {
+		Transaction txn = new Transaction(null, true);
+
+		context.checking(new DbExpectations() {{
+			oneOf(db).transaction(with(true), withDbRunnable(txn));
+			oneOf(db).getContacts(txn);
+			will(returnValue(local));
+		}});
+	}
+
+	private void expectLoadMailboxUpdate(Contact c, MailboxUpdate update)
+			throws Exception {
+		Transaction txn = new Transaction(null, true);
+
+		context.checking(new DbExpectations() {{
+			oneOf(db).transactionWithResult(with(true),
+					withDbCallable(txn));
+			oneOf(mailboxUpdateManager).getLocalUpdate(txn, c.getId());
+			will(returnValue(update));
+		}});
+	}
+
+	private void expectStartTaskToAddContact(AtomicReference<ApiCall> task) {
+		context.checking(new Expectations() {{
+			oneOf(mailboxApiCaller).retryWithBackoff(with(any(ApiCall.class)));
+			will(new DoAllAction(
+					new CaptureArgumentAction<>(task, ApiCall.class, 0),
+					returnValue(apiCall)
+			));
+		}});
+	}
+
+	private void expectAddContactToMailbox(
+			AtomicReference<MailboxContact> added) throws Exception {
+		context.checking(new DbExpectations() {{
+			oneOf(mailboxApi).addContact(with(mailboxProperties),
+					with(any(MailboxContact.class)));
+			will(new CaptureArgumentAction<>(added, MailboxContact.class, 1));
+		}});
+	}
+
+	private void expectStartTaskToRemoveContact(AtomicReference<ApiCall> task) {
+		context.checking(new DbExpectations() {{
+			oneOf(mailboxApiCaller).retryWithBackoff(with(any(ApiCall.class)));
+			will(new DoAllAction(
+					new CaptureArgumentAction<>(task, ApiCall.class, 0),
+					returnValue(apiCall)
+			));
+		}});
+	}
+
+	private void expectRemoveContactFromMailbox(Contact c) throws Exception {
+		context.checking(new Expectations() {{
+			oneOf(mailboxApi).deleteContact(mailboxProperties, c.getId());
+		}});
+	}
+
+	private void expectRunTaskOnIoExecutor() {
+		context.checking(new Expectations() {{
+			oneOf(ioExecutor).execute(with(any(Runnable.class)));
+			will(new RunAction());
+		}});
+	}
+
+	private void assertMailboxContactEquals(MailboxContact expected,
+			MailboxContact actual) {
+		assertEquals(expected.contactId, actual.contactId);
+		assertEquals(expected.token, actual.token);
+		assertEquals(expected.inboxId, actual.inboxId);
+		assertEquals(expected.outboxId, actual.outboxId);
+	}
+}