Merge branch '1802-sync-via-removable-storage' into offline-testing

# Conflicts:
#	bramble-api/src/main/java/org/briarproject/bramble/api/FeatureFlags.java
#	bramble-core/build.gradle
#	bramble-core/src/test/java/org/briarproject/bramble/test/BrambleCoreIntegrationTestModule.java
#	bramble-core/witness.gradle
#	bramble-java/src/main/java/org/briarproject/bramble/plugin/tor/UnixTorPluginFactory.java
#	briar-android/src/main/java/org/briarproject/briar/android/AndroidComponent.java
#	briar-android/src/main/java/org/briarproject/briar/android/AppModule.java
#	briar-android/src/main/java/org/briarproject/briar/android/activity/ActivityComponent.java
#	briar-android/src/main/java/org/briarproject/briar/android/util/UiUtils.java
#	briar-android/src/main/res/values/strings.xml
#	briar-headless/src/main/java/org/briarproject/briar/headless/HeadlessModule.kt
#	briar-headless/src/test/java/org/briarproject/briar/headless/HeadlessTestModule.kt

bramble-core/src/main/java/org/briarproject/bramble/BrambleCoreEagerSingletons.java

@@ -11,6 +11,7 @@ import org.briarproject.bramble.properties.PropertiesModule;
 import org.briarproject.bramble.rendezvous.RendezvousModule;
 import org.briarproject.bramble.sync.validation.ValidationModule;
 import org.briarproject.bramble.transport.TransportModule;
+import org.briarproject.bramble.transport.agreement.TransportKeyAgreementModule;
 import org.briarproject.bramble.versioning.VersioningModule;
 
 public interface BrambleCoreEagerSingletons {
@@ -33,6 +34,8 @@ public interface BrambleCoreEagerSingletons {
 
 	void inject(RendezvousModule.EagerSingletons init);
 
+	void inject(TransportKeyAgreementModule.EagerSingletons init);
+
 	void inject(TransportModule.EagerSingletons init);
 
 	void inject(ValidationModule.EagerSingletons init);
@@ -51,6 +54,7 @@ public interface BrambleCoreEagerSingletons {
 			c.inject(new RendezvousModule.EagerSingletons());
 			c.inject(new PluginModule.EagerSingletons());
 			c.inject(new PropertiesModule.EagerSingletons());
+			c.inject(new TransportKeyAgreementModule.EagerSingletons());
 			c.inject(new TransportModule.EagerSingletons());
 			c.inject(new ValidationModule.EagerSingletons());
 			c.inject(new VersioningModule.EagerSingletons());

bramble-core/src/main/java/org/briarproject/bramble/BrambleCoreModule.java

@@ -23,6 +23,7 @@ import org.briarproject.bramble.settings.SettingsModule;
 import org.briarproject.bramble.sync.SyncModule;
 import org.briarproject.bramble.sync.validation.ValidationModule;
 import org.briarproject.bramble.transport.TransportModule;
+import org.briarproject.bramble.transport.agreement.TransportKeyAgreementModule;
 import org.briarproject.bramble.versioning.VersioningModule;
 
 import dagger.Module;
@@ -49,6 +50,7 @@ import dagger.Module;
 		RendezvousModule.class,
 		SettingsModule.class,
 		SyncModule.class,
+		TransportKeyAgreementModule.class,
 		TransportModule.class,
 		ValidationModule.class,
 		VersioningModule.class

bramble-core/src/main/java/org/briarproject/bramble/connection/Connection.java

@@ -45,7 +45,6 @@ abstract class Connection {
 	@Nullable
 	StreamContext recogniseTag(TransportConnectionReader reader,
 			TransportId transportId) {
-		StreamContext ctx;
 		try {
 			byte[] tag = readTag(reader.getInputStream());
 			return keyManager.getStreamContext(transportId, tag);

bramble-core/src/main/java/org/briarproject/bramble/connection/OutgoingSimplexSyncConnection.java

@@ -71,8 +71,10 @@ class OutgoingSimplexSyncConnection extends SyncConnection implements Runnable {
 		StreamWriter streamWriter = streamWriterFactory.createStreamWriter(
 				w.getOutputStream(), ctx);
 		ContactId c = requireNonNull(ctx.getContactId());
+		// Use eager retransmission if the transport is lossy and cheap
 		return syncSessionFactory.createSimplexOutgoingSession(c,
-				ctx.getTransportId(), w.getMaxLatency(), streamWriter);
+				ctx.getTransportId(), w.getMaxLatency(), w.isLossyAndCheap(),
+				streamWriter);
 	}
 }
 

bramble-core/src/main/java/org/briarproject/bramble/contact/ContactExchangeManagerImpl.java

@@ -47,6 +47,7 @@ import javax.inject.Inject;
 
 import static java.util.logging.Logger.getLogger;
 import static org.briarproject.bramble.api.identity.AuthorConstants.MAX_SIGNATURE_LENGTH;
+import static org.briarproject.bramble.api.system.Clock.MIN_REASONABLE_TIME_MS;
 import static org.briarproject.bramble.contact.ContactExchangeConstants.PROTOCOL_VERSION;
 import static org.briarproject.bramble.contact.ContactExchangeRecordTypes.CONTACT_INFO;
 import static org.briarproject.bramble.util.ValidationUtils.checkLength;
@@ -184,6 +185,10 @@ class ContactExchangeManagerImpl implements ContactExchangeManager {
 
 		// The agreed timestamp is the minimum of the peers' timestamps
 		long timestamp = Math.min(localTimestamp, remoteInfo.timestamp);
+		if (timestamp < MIN_REASONABLE_TIME_MS) {
+			LOG.warning("Timestamp is too old");
+			throw new FormatException();
+		}
 
 		// Add the contact
 		Contact contact = addContact(p, remoteInfo.author, localAuthor,

bramble-core/src/main/java/org/briarproject/bramble/db/Database.java

@@ -145,7 +145,7 @@ interface Database<T> {
 	/**
 	 * Stores a transport.
 	 */
-	void addTransport(T txn, TransportId t, int maxLatency)
+	void addTransport(T txn, TransportId t, long maxLatency)
 			throws DbException;
 
 	/**
@@ -162,6 +162,18 @@ interface Database<T> {
 	KeySetId addTransportKeys(T txn, PendingContactId p, TransportKeys k)
 			throws DbException;
 
+	/**
+	 * Returns true if there are any acks or messages to send to the given
+	 * contact over a transport with the given maximum latency.
+	 * <p/>
+	 * Read-only.
+	 *
+	 * @param eager True if messages that are not yet due for retransmission
+	 * should be included
+	 */
+	boolean containsAnythingToSend(T txn, ContactId c, long maxLatency,
+			boolean eager) throws DbException;
+
 	/**
 	 * Returns true if the database contains the given contact for the given
 	 * local pseudonym.
@@ -215,6 +227,16 @@ interface Database<T> {
 	 */
 	boolean containsTransport(T txn, TransportId t) throws DbException;
 
+	/**
+	 * Returns true if the database contains keys for communicating with the
+	 * given contact over the given transport. Handshake mode and rotation mode
+	 * keys are included, whether activated or not.
+	 * <p/>
+	 * Read-only.
+	 */
+	boolean containsTransportKeys(T txn, ContactId c, TransportId t)
+			throws DbException;
+
 	/**
 	 * Returns true if the database contains the given message, the message is
 	 * shared, and the visibility of the message's group to the given contact
@@ -461,7 +483,7 @@ interface Database<T> {
 	 * Read-only.
 	 */
 	Collection<MessageId> getMessagesToOffer(T txn, ContactId c,
-			int maxMessages, int maxLatency) throws DbException;
+			int maxMessages, long maxLatency) throws DbException;
 
 	/**
 	 * Returns the IDs of some messages that are eligible to be requested from
@@ -476,10 +498,36 @@ interface Database<T> {
 	 * Returns the IDs of some messages that are eligible to be sent to the
 	 * given contact, up to the given total length.
 	 * <p/>
+	 * Unlike {@link #getUnackedMessagesToSend(Object, ContactId)} this method
+	 * does not return messages that have already been sent unless they are
+	 * due for retransmission.
+	 * <p/>
 	 * Read-only.
 	 */
 	Collection<MessageId> getMessagesToSend(T txn, ContactId c, int maxLength,
-			int maxLatency) throws DbException;
+			long maxLatency) throws DbException;
+
+	/**
+	 * Returns the IDs of all messages that are eligible to be sent to the
+	 * given contact, together with their raw lengths.
+	 * <p/>
+	 * Unlike {@link #getMessagesToSend(Object, ContactId, int, long)} this
+	 * method may return messages that have already been sent and are not yet
+	 * due for retransmission.
+	 * <p/>
+	 * Read-only.
+	 */
+	Map<MessageId, Integer> getUnackedMessagesToSend(T txn, ContactId c)
+			throws DbException;
+
+	/**
+	 * Returns the total length, including headers, of all messages that are
+	 * eligible to be sent to the given contact. This may include messages
+	 * that have already been sent and are not yet due for retransmission.
+	 * <p/>
+	 * Read-only.
+	 */
+	long getUnackedMessageBytesToSend(T txn, ContactId c) throws DbException;
 
 	/**
 	 * Returns the IDs of any messages that need to be validated.
@@ -556,7 +604,7 @@ interface Database<T> {
 	 * Read-only.
 	 */
 	Collection<MessageId> getRequestedMessagesToSend(T txn, ContactId c,
-			int maxLength, int maxLatency) throws DbException;
+			int maxLength, long maxLatency) throws DbException;
 
 	/**
 	 * Returns all settings in the given namespace.
@@ -580,6 +628,16 @@ interface Database<T> {
 	Collection<TransportKeySet> getTransportKeys(T txn, TransportId t)
 			throws DbException;
 
+	/**
+	 * Returns the contact IDs and transport IDs for which the DB contains
+	 * at least one set of transport keys. Handshake mode and rotation mode
+	 * keys are included, whether activated or not.
+	 * <p/>
+	 * Read-only.
+	 */
+	Map<ContactId, Collection<TransportId>> getTransportsWithKeys(T txn)
+			throws DbException;
+
 	/**
 	 * Increments the outgoing stream counter for the given transport keys.
 	 */
@@ -787,7 +845,7 @@ interface Database<T> {
 	 * of the given message with respect to the given contact, using the latency
 	 * of the transport over which it was sent.
 	 */
-	void updateExpiryTimeAndEta(T txn, ContactId c, MessageId m, int maxLatency)
+	void updateExpiryTimeAndEta(T txn, ContactId c, MessageId m, long maxLatency)
 			throws DbException;
 
 	/**

bramble-core/src/main/java/org/briarproject/bramble/db/DatabaseComponentImpl.java

@@ -310,7 +310,7 @@ class DatabaseComponentImpl<T> implements DatabaseComponent {
 
 	@Override
 	public void addTransport(Transaction transaction, TransportId t,
-			int maxLatency) throws DbException {
+			long maxLatency) throws DbException {
 		if (transaction.isReadOnly()) throw new IllegalArgumentException();
 		T txn = unbox(transaction);
 		if (!db.containsTransport(txn, t))
@@ -341,6 +341,15 @@ class DatabaseComponentImpl<T> implements DatabaseComponent {
 		return db.addTransportKeys(txn, p, k);
 	}
 
+	@Override
+	public boolean containsAnythingToSend(Transaction transaction, ContactId c,
+			long maxLatency, boolean eager) throws DbException {
+		T txn = unbox(transaction);
+		if (!db.containsContact(txn, c))
+			throw new NoSuchContactException();
+		return db.containsAnythingToSend(txn, c, maxLatency, eager);
+	}
+
 	@Override
 	public boolean containsContact(Transaction transaction, AuthorId remote,
 			AuthorId local) throws DbException {
@@ -371,6 +380,13 @@ class DatabaseComponentImpl<T> implements DatabaseComponent {
 		return db.containsPendingContact(txn, p);
 	}
 
+	@Override
+	public boolean containsTransportKeys(Transaction transaction, ContactId c,
+			TransportId t) throws DbException {
+		T txn = unbox(transaction);
+		return db.containsTransportKeys(txn, c, t);
+	}
+
 	@Override
 	public void deleteMessage(Transaction transaction, MessageId m)
 			throws DbException {
@@ -408,28 +424,57 @@ class DatabaseComponentImpl<T> implements DatabaseComponent {
 	@Nullable
 	@Override
 	public Collection<Message> generateBatch(Transaction transaction,
-			ContactId c, int maxLength, int maxLatency) throws DbException {
+			ContactId c, int maxLength, long maxLatency) throws DbException {
 		if (transaction.isReadOnly()) throw new IllegalArgumentException();
 		T txn = unbox(transaction);
 		if (!db.containsContact(txn, c))
 			throw new NoSuchContactException();
 		Collection<MessageId> ids =
 				db.getMessagesToSend(txn, c, maxLength, maxLatency);
+		long totalLength = 0;
 		List<Message> messages = new ArrayList<>(ids.size());
 		for (MessageId m : ids) {
-			messages.add(db.getMessage(txn, m));
+			Message message = db.getMessage(txn, m);
+			totalLength += message.getRawLength();
+			messages.add(message);
 			db.updateExpiryTimeAndEta(txn, c, m, maxLatency);
 		}
 		if (ids.isEmpty()) return null;
 		db.lowerRequestedFlag(txn, c, ids);
-		transaction.attach(new MessagesSentEvent(c, ids));
+		transaction.attach(new MessagesSentEvent(c, ids, totalLength));
+		return messages;
+	}
+
+	@Override
+	public Collection<Message> generateBatch(Transaction transaction,
+			ContactId c, Collection<MessageId> ids, long maxLatency)
+			throws DbException {
+		if (transaction.isReadOnly()) throw new IllegalArgumentException();
+		T txn = unbox(transaction);
+		if (!db.containsContact(txn, c))
+			throw new NoSuchContactException();
+		long totalLength = 0;
+		List<Message> messages = new ArrayList<>(ids.size());
+		List<MessageId> sentIds = new ArrayList<>(ids.size());
+		for (MessageId m : ids) {
+			if (db.containsVisibleMessage(txn, c, m)) {
+				Message message = db.getMessage(txn, m);
+				totalLength += message.getRawLength();
+				messages.add(message);
+				sentIds.add(m);
+				db.updateExpiryTimeAndEta(txn, c, m, maxLatency);
+			}
+		}
+		if (messages.isEmpty()) return messages;
+		db.lowerRequestedFlag(txn, c, sentIds);
+		transaction.attach(new MessagesSentEvent(c, sentIds, totalLength));
 		return messages;
 	}
 
 	@Nullable
 	@Override
 	public Offer generateOffer(Transaction transaction, ContactId c,
-			int maxMessages, int maxLatency) throws DbException {
+			int maxMessages, long maxLatency) throws DbException {
 		if (transaction.isReadOnly()) throw new IllegalArgumentException();
 		T txn = unbox(transaction);
 		if (!db.containsContact(txn, c))
@@ -460,21 +505,24 @@ class DatabaseComponentImpl<T> implements DatabaseComponent {
 	@Nullable
 	@Override
 	public Collection<Message> generateRequestedBatch(Transaction transaction,
-			ContactId c, int maxLength, int maxLatency) throws DbException {
+			ContactId c, int maxLength, long maxLatency) throws DbException {
 		if (transaction.isReadOnly()) throw new IllegalArgumentException();
 		T txn = unbox(transaction);
 		if (!db.containsContact(txn, c))
 			throw new NoSuchContactException();
 		Collection<MessageId> ids =
 				db.getRequestedMessagesToSend(txn, c, maxLength, maxLatency);
+		long totalLength = 0;
 		List<Message> messages = new ArrayList<>(ids.size());
 		for (MessageId m : ids) {
-			messages.add(db.getMessage(txn, m));
+			Message message = db.getMessage(txn, m);
+			totalLength += message.getRawLength();
+			messages.add(message);
 			db.updateExpiryTimeAndEta(txn, c, m, maxLatency);
 		}
 		if (ids.isEmpty()) return null;
 		db.lowerRequestedFlag(txn, c, ids);
-		transaction.attach(new MessagesSentEvent(c, ids));
+		transaction.attach(new MessagesSentEvent(c, ids, totalLength));
 		return messages;
 	}
 
@@ -692,6 +740,25 @@ class DatabaseComponentImpl<T> implements DatabaseComponent {
 		return status;
 	}
 
+	@Override
+	public Map<MessageId, Integer> getUnackedMessagesToSend(
+			Transaction transaction,
+			ContactId c) throws DbException {
+		T txn = unbox(transaction);
+		if (!db.containsContact(txn, c))
+			throw new NoSuchContactException();
+		return db.getUnackedMessagesToSend(txn, c);
+	}
+
+	@Override
+	public long getUnackedMessageBytesToSend(Transaction transaction,
+			ContactId c) throws DbException {
+		T txn = unbox(transaction);
+		if (!db.containsContact(txn, c))
+			throw new NoSuchContactException();
+		return db.getUnackedMessageBytesToSend(txn, c);
+	}
+
 	@Override
 	public Map<MessageId, MessageState> getMessageDependencies(
 			Transaction transaction, MessageId m) throws DbException {
@@ -765,6 +832,13 @@ class DatabaseComponentImpl<T> implements DatabaseComponent {
 		return db.getTransportKeys(txn, t);
 	}
 
+	@Override
+	public Map<ContactId, Collection<TransportId>> getTransportsWithKeys(
+			Transaction transaction) throws DbException {
+		T txn = unbox(transaction);
+		return db.getTransportsWithKeys(txn);
+	}
+
 	@Override
 	public void incrementStreamCounter(Transaction transaction, TransportId t,
 			KeySetId k) throws DbException {

bramble-core/src/main/java/org/briarproject/bramble/db/ExponentialBackoff.java

@@ -1,5 +1,7 @@
 package org.briarproject.bramble.db;
 
+import static org.briarproject.bramble.api.sync.SyncConstants.MAX_TRANSPORT_LATENCY;
+
 class ExponentialBackoff {
 
 	/**
@@ -11,9 +13,11 @@ class ExponentialBackoff {
 	 * transmissions increases exponentially. If the expiry time would
 	 * be greater than Long.MAX_VALUE, Long.MAX_VALUE is returned.
 	 */
-	static long calculateExpiry(long now, int maxLatency, int txCount) {
+	static long calculateExpiry(long now, long maxLatency, int txCount) {
 		if (now < 0) throw new IllegalArgumentException();
-		if (maxLatency <= 0) throw new IllegalArgumentException();
+		if (maxLatency <= 0 || maxLatency > MAX_TRANSPORT_LATENCY) {
+			throw new IllegalArgumentException();
+		}
 		if (txCount < 0) throw new IllegalArgumentException();
 		// The maximum round-trip time is twice the maximum latency
 		long roundTrip = maxLatency * 2L;

bramble-core/src/main/java/org/briarproject/bramble/db/JdbcDatabase.java

@@ -51,6 +51,7 @@ import java.util.Collection;
 import java.util.Collections;
 import java.util.HashMap;
 import java.util.HashSet;
+import java.util.LinkedHashMap;
 import java.util.LinkedList;
 import java.util.List;
 import java.util.Map;
@@ -101,7 +102,7 @@ import static org.briarproject.bramble.util.LogUtils.now;
 abstract class JdbcDatabase implements Database<Connection> {
 
 	// Package access for testing
-	static final int CODE_SCHEMA_VERSION = 48;
+	static final int CODE_SCHEMA_VERSION = 49;
 
 	// Time period offsets for incoming transport keys
 	private static final int OFFSET_PREV = -1;
@@ -266,7 +267,7 @@ abstract class JdbcDatabase implements Database<Connection> {
 	private static final String CREATE_TRANSPORTS =
 			"CREATE TABLE transports"
 					+ " (transportId _STRING NOT NULL,"
-					+ " maxLatency INT NOT NULL,"
+					+ " maxLatency BIGINT NOT NULL,"
 					+ " PRIMARY KEY (transportId))";
 
 	private static final String CREATE_PENDING_CONTACTS =
@@ -344,6 +345,11 @@ abstract class JdbcDatabase implements Database<Connection> {
 			"CREATE INDEX IF NOT EXISTS statusesByContactIdTimestamp"
 					+ " ON statuses (contactId, timestamp)";
 
+	private static final String
+			INDEX_STATUSES_BY_CONTACT_ID_TX_COUNT_TIMESTAMP =
+			"CREATE INDEX IF NOT EXISTS statusesByContactIdTxCountTimestamp"
+					+ " ON statuses (contactId, txCount, timestamp)";
+
 	private static final String INDEX_MESSAGES_BY_CLEANUP_DEADLINE =
 			"CREATE INDEX IF NOT EXISTS messagesByCleanupDeadline"
 					+ " ON messages (cleanupDeadline)";
@@ -492,7 +498,8 @@ abstract class JdbcDatabase implements Database<Connection> {
 				new Migration44_45(),
 				new Migration45_46(),
 				new Migration46_47(dbTypes),
-				new Migration47_48()
+				new Migration47_48(),
+				new Migration48_49()
 		);
 	}
 
@@ -570,6 +577,7 @@ abstract class JdbcDatabase implements Database<Connection> {
 			s.executeUpdate(INDEX_MESSAGE_DEPENDENCIES_BY_DEPENDENCY_ID);
 			s.executeUpdate(INDEX_STATUSES_BY_CONTACT_ID_GROUP_ID);
 			s.executeUpdate(INDEX_STATUSES_BY_CONTACT_ID_TIMESTAMP);
+			s.executeUpdate(INDEX_STATUSES_BY_CONTACT_ID_TX_COUNT_TIMESTAMP);
 			s.executeUpdate(INDEX_MESSAGES_BY_CLEANUP_DEADLINE);
 			s.close();
 		} catch (SQLException e) {
@@ -999,7 +1007,7 @@ abstract class JdbcDatabase implements Database<Connection> {
 	}
 
 	@Override
-	public void addTransport(Connection txn, TransportId t, int maxLatency)
+	public void addTransport(Connection txn, TransportId t, long maxLatency)
 			throws DbException {
 		PreparedStatement ps = null;
 		try {
@@ -1120,6 +1128,55 @@ abstract class JdbcDatabase implements Database<Connection> {
 		}
 	}
 
+	@Override
+	public boolean containsAnythingToSend(Connection txn, ContactId c,
+			long maxLatency, boolean eager) throws DbException {
+		PreparedStatement ps = null;
+		ResultSet rs = null;
+		try {
+			String sql = "SELECT NULL FROM statuses"
+					+ " WHERE contactId = ? AND ack = TRUE";
+			ps = txn.prepareStatement(sql);
+			ps.setInt(1, c.getInt());
+			rs = ps.executeQuery();
+			boolean acksToSend = rs.next();
+			rs.close();
+			ps.close();
+			if (acksToSend) return true;
+			if (eager) {
+				sql = "SELECT NULL from statuses"
+						+ " WHERE contactId = ? AND state = ?"
+						+ " AND groupShared = TRUE AND messageShared = TRUE"
+						+ " AND deleted = FALSE AND seen = FALSE";
+				ps = txn.prepareStatement(sql);
+				ps.setInt(1, c.getInt());
+				ps.setInt(2, DELIVERED.getValue());
+			} else {
+				long now = clock.currentTimeMillis();
+				long eta = now + maxLatency;
+				sql = "SELECT NULL FROM statuses"
+						+ " WHERE contactId = ? AND state = ?"
+						+ " AND groupShared = TRUE AND messageShared = TRUE"
+						+ " AND deleted = FALSE AND seen = FALSE"
+						+ " AND (expiry <= ? OR eta > ?)";
+				ps = txn.prepareStatement(sql);
+				ps.setInt(1, c.getInt());
+				ps.setInt(2, DELIVERED.getValue());
+				ps.setLong(3, now);
+				ps.setLong(4, eta);
+			}
+			rs = ps.executeQuery();
+			boolean messagesToSend = rs.next();
+			rs.close();
+			ps.close();
+			return messagesToSend;
+		} catch (SQLException e) {
+			tryToClose(rs, LOG, WARNING);
+			tryToClose(ps, LOG, WARNING);
+			throw new DbException(e);
+		}
+	}
+
 	@Override
 	public boolean containsContact(Connection txn, AuthorId remote,
 			AuthorId local) throws DbException {
@@ -1277,6 +1334,29 @@ abstract class JdbcDatabase implements Database<Connection> {
 		}
 	}
 
+	@Override
+	public boolean containsTransportKeys(Connection txn, ContactId c,
+			TransportId t) throws DbException {
+		PreparedStatement ps = null;
+		ResultSet rs = null;
+		try {
+			String sql = "SELECT NULL FROM outgoingKeys"
+					+ " WHERE contactId = ? AND transportId = ?";
+			ps = txn.prepareStatement(sql);
+			ps.setInt(1, c.getInt());
+			ps.setString(2, t.getString());
+			rs = ps.executeQuery();
+			boolean found = rs.next();
+			rs.close();
+			ps.close();
+			return found;
+		} catch (SQLException e) {
+			tryToClose(rs, LOG, WARNING);
+			tryToClose(ps, LOG, WARNING);
+			throw new DbException(e);
+		}
+	}
+
 	@Override
 	public boolean containsVisibleMessage(Connection txn, ContactId c,
 			MessageId m) throws DbException {
@@ -2109,7 +2189,7 @@ abstract class JdbcDatabase implements Database<Connection> {
 
 	@Override
 	public Collection<MessageId> getMessagesToOffer(Connection txn,
-			ContactId c, int maxMessages, int maxLatency) throws DbException {
+			ContactId c, int maxMessages, long maxLatency) throws DbException {
 		long now = clock.currentTimeMillis();
 		long eta = now + maxLatency;
 		PreparedStatement ps = null;
@@ -2168,7 +2248,7 @@ abstract class JdbcDatabase implements Database<Connection> {
 
 	@Override
 	public Collection<MessageId> getMessagesToSend(Connection txn, ContactId c,
-			int maxLength, int maxLatency) throws DbException {
+			int maxLength, long maxLatency) throws DbException {
 		long now = clock.currentTimeMillis();
 		long eta = now + maxLatency;
 		PreparedStatement ps = null;
@@ -2205,6 +2285,63 @@ abstract class JdbcDatabase implements Database<Connection> {
 		}
 	}
 
+	@Override
+	public Map<MessageId, Integer> getUnackedMessagesToSend(Connection txn,
+			ContactId c) throws DbException {
+		PreparedStatement ps = null;
+		ResultSet rs = null;
+		try {
+			String sql = "SELECT length, messageId FROM statuses"
+					+ " WHERE contactId = ? AND state = ?"
+					+ " AND groupShared = TRUE AND messageShared = TRUE"
+					+ " AND deleted = FALSE AND seen = FALSE"
+					+ " ORDER BY txCount, timestamp";
+			ps = txn.prepareStatement(sql);
+			ps.setInt(1, c.getInt());
+			ps.setInt(2, DELIVERED.getValue());
+			rs = ps.executeQuery();
+			Map<MessageId, Integer> results = new LinkedHashMap<>();
+			while (rs.next()) {
+				int length = rs.getInt(1);
+				MessageId id = new MessageId(rs.getBytes(2));
+				results.put(id, length);
+			}
+			rs.close();
+			ps.close();
+			return results;
+		} catch (SQLException e) {
+			tryToClose(rs, LOG, WARNING);
+			tryToClose(ps, LOG, WARNING);
+			throw new DbException(e);
+		}
+	}
+
+	@Override
+	public long getUnackedMessageBytesToSend(Connection txn, ContactId c)
+			throws DbException {
+		PreparedStatement ps = null;
+		ResultSet rs = null;
+		try {
+			String sql = "SELECT SUM(length) FROM statuses"
+					+ " WHERE contactId = ? AND state = ?"
+					+ " AND groupShared = TRUE AND messageShared = TRUE"
+					+ " AND deleted = FALSE AND seen = FALSE";
+			ps = txn.prepareStatement(sql);
+			ps.setInt(1, c.getInt());
+			ps.setInt(2, DELIVERED.getValue());
+			rs = ps.executeQuery();
+			rs.next();
+			long total = rs.getInt(1);
+			rs.close();
+			ps.close();
+			return total;
+		} catch (SQLException e) {
+			tryToClose(rs, LOG, WARNING);
+			tryToClose(ps, LOG, WARNING);
+			throw new DbException(e);
+		}
+	}
+
 	@Override
 	public Collection<MessageId> getMessagesToValidate(Connection txn)
 			throws DbException {
@@ -2410,7 +2547,7 @@ abstract class JdbcDatabase implements Database<Connection> {
 
 	@Override
 	public Collection<MessageId> getRequestedMessagesToSend(Connection txn,
-			ContactId c, int maxLength, int maxLatency) throws DbException {
+			ContactId c, int maxLength, long maxLatency) throws DbException {
 		long now = clock.currentTimeMillis();
 		long eta = now + maxLatency;
 		PreparedStatement ps = null;
@@ -2574,6 +2711,38 @@ abstract class JdbcDatabase implements Database<Connection> {
 		}
 	}
 
+	@Override
+	public Map<ContactId, Collection<TransportId>> getTransportsWithKeys(
+			Connection txn) throws DbException {
+		Statement s = null;
+		ResultSet rs = null;
+		try {
+			String sql = "SELECT DISTINCT contactId, transportId"
+					+ " FROM outgoingKeys";
+			s = txn.createStatement();
+			rs = s.executeQuery(sql);
+			Map<ContactId, Collection<TransportId>> ids = new HashMap<>();
+			while (rs.next()) {
+				ContactId c = new ContactId(rs.getInt(1));
+				TransportId t = new TransportId(rs.getString(2));
+				Collection<TransportId> transportIds = ids.get(c);
+				if (transportIds == null) {
+					transportIds = new ArrayList<>();
+					ids.put(c, transportIds);
+				}
+				transportIds.add(t);
+			}
+			rs.close();
+			s.close();
+			return ids;
+		} catch (SQLException e) {
+			tryToClose(rs, LOG, WARNING);
+			tryToClose(s, LOG, WARNING);
+			tryToClose(s, LOG, WARNING);
+			throw new DbException(e);
+		}
+	}
+
 	@Override
 	public void incrementStreamCounter(Connection txn, TransportId t,
 			KeySetId k) throws DbException {
@@ -3449,7 +3618,7 @@ abstract class JdbcDatabase implements Database<Connection> {
 
 	@Override
 	public void updateExpiryTimeAndEta(Connection txn, ContactId c, MessageId m,
-			int maxLatency) throws DbException {
+			long maxLatency) throws DbException {
 		PreparedStatement ps = null;
 		ResultSet rs = null;
 		try {

bramble-core/src/main/java/org/briarproject/bramble/db/Migration48_49.java

@@ -0,0 +1,41 @@
+package org.briarproject.bramble.db;
+
+import org.briarproject.bramble.api.db.DbException;
+
+import java.sql.Connection;
+import java.sql.SQLException;
+import java.sql.Statement;
+import java.util.logging.Logger;
+
+import static java.util.logging.Level.WARNING;
+import static java.util.logging.Logger.getLogger;
+import static org.briarproject.bramble.db.JdbcUtils.tryToClose;
+
+class Migration48_49 implements Migration<Connection> {
+
+	private static final Logger LOG = getLogger(Migration48_49.class.getName());
+
+	@Override
+	public int getStartVersion() {
+		return 48;
+	}
+
+	@Override
+	public int getEndVersion() {
+		return 49;
+	}
+
+	@Override
+	public void migrate(Connection txn) throws DbException {
+		Statement s = null;
+		try {
+			s = txn.createStatement();
+			s.execute("ALTER TABLE transports"
+					+ " ALTER COLUMN maxLatency"
+					+ " SET DATA TYPE BIGINT");
+		} catch (SQLException e) {
+			tryToClose(s, LOG, WARNING);
+			throw new DbException(e);
+		}
+	}
+}

bramble-core/src/main/java/org/briarproject/bramble/lifecycle/LifecycleManagerImpl.java

@@ -12,6 +12,7 @@ import org.briarproject.bramble.api.lifecycle.Service;
 import org.briarproject.bramble.api.lifecycle.ServiceException;
 import org.briarproject.bramble.api.lifecycle.event.LifecycleEvent;
 import org.briarproject.bramble.api.nullsafety.NotNullByDefault;
+import org.briarproject.bramble.api.system.Clock;
 
 import java.util.List;
 import java.util.concurrent.CopyOnWriteArrayList;
@@ -34,11 +35,14 @@ import static org.briarproject.bramble.api.lifecycle.LifecycleManager.LifecycleS
 import static org.briarproject.bramble.api.lifecycle.LifecycleManager.LifecycleState.STARTING_SERVICES;
 import static org.briarproject.bramble.api.lifecycle.LifecycleManager.LifecycleState.STOPPING;
 import static org.briarproject.bramble.api.lifecycle.LifecycleManager.StartResult.ALREADY_RUNNING;
+import static org.briarproject.bramble.api.lifecycle.LifecycleManager.StartResult.CLOCK_ERROR;
 import static org.briarproject.bramble.api.lifecycle.LifecycleManager.StartResult.DATA_TOO_NEW_ERROR;
 import static org.briarproject.bramble.api.lifecycle.LifecycleManager.StartResult.DATA_TOO_OLD_ERROR;
 import static org.briarproject.bramble.api.lifecycle.LifecycleManager.StartResult.DB_ERROR;
 import static org.briarproject.bramble.api.lifecycle.LifecycleManager.StartResult.SERVICE_ERROR;
 import static org.briarproject.bramble.api.lifecycle.LifecycleManager.StartResult.SUCCESS;
+import static org.briarproject.bramble.api.system.Clock.MAX_REASONABLE_TIME_MS;
+import static org.briarproject.bramble.api.system.Clock.MIN_REASONABLE_TIME_MS;
 import static org.briarproject.bramble.util.LogUtils.logDuration;
 import static org.briarproject.bramble.util.LogUtils.logException;
 import static org.briarproject.bramble.util.LogUtils.now;
@@ -52,6 +56,7 @@ class LifecycleManagerImpl implements LifecycleManager, MigrationListener {
 
 	private final DatabaseComponent db;
 	private final EventBus eventBus;
+	private final Clock clock;
 	private final List<Service> services;
 	private final List<OpenDatabaseHook> openDatabaseHooks;
 	private final List<ExecutorService> executors;
@@ -63,9 +68,11 @@ class LifecycleManagerImpl implements LifecycleManager, MigrationListener {
 	private volatile LifecycleState state = STARTING;
 
 	@Inject
-	LifecycleManagerImpl(DatabaseComponent db, EventBus eventBus) {
+	LifecycleManagerImpl(DatabaseComponent db, EventBus eventBus,
+			Clock clock) {
 		this.db = db;
 		this.eventBus = eventBus;
+		this.clock = clock;
 		services = new CopyOnWriteArrayList<>();
 		openDatabaseHooks = new CopyOnWriteArrayList<>();
 		executors = new CopyOnWriteArrayList<>();
@@ -99,6 +106,13 @@ class LifecycleManagerImpl implements LifecycleManager, MigrationListener {
 			LOG.info("Already starting or stopping");
 			return ALREADY_RUNNING;
 		}
+		long now = clock.currentTimeMillis();
+		if (now < MIN_REASONABLE_TIME_MS || now > MAX_REASONABLE_TIME_MS) {
+			if (LOG.isLoggable(WARNING)) {
+				LOG.warning("System clock is unreasonable: " + now);
+			}
+			return CLOCK_ERROR;
+		}
 		try {
 			LOG.info("Opening database");
 			long start = now();

bramble-core/src/main/java/org/briarproject/bramble/plugin/bluetooth/AbstractBluetoothPlugin.java

@@ -79,7 +79,8 @@ abstract class AbstractBluetoothPlugin<S, SS> implements BluetoothPlugin,
 	private final SecureRandom secureRandom;
 	private final Backoff backoff;
 	private final PluginCallback callback;
-	private final int maxLatency, maxIdleTime;
+	private final long maxLatency;
+	private final int maxIdleTime;
 	private final AtomicBoolean used = new AtomicBoolean(false);
 	private final AtomicBoolean everConnected = new AtomicBoolean(false);
 
@@ -121,7 +122,7 @@ abstract class AbstractBluetoothPlugin<S, SS> implements BluetoothPlugin,
 			SecureRandom secureRandom,
 			Backoff backoff,
 			PluginCallback callback,
-			int maxLatency,
+			long maxLatency,
 			int maxIdleTime) {
 		this.connectionLimiter = connectionLimiter;
 		this.connectionFactory = connectionFactory;
@@ -158,7 +159,7 @@ abstract class AbstractBluetoothPlugin<S, SS> implements BluetoothPlugin,
 	}
 
 	@Override
-	public int getMaxLatency() {
+	public long getMaxLatency() {
 		return maxLatency;
 	}
 

bramble-core/src/main/java/org/briarproject/bramble/plugin/file/AbstractRemovableDrivePlugin.java

@@ -0,0 +1,126 @@
+package org.briarproject.bramble.plugin.file;
+
+import org.briarproject.bramble.api.Pair;
+import org.briarproject.bramble.api.nullsafety.NotNullByDefault;
+import org.briarproject.bramble.api.plugin.ConnectionHandler;
+import org.briarproject.bramble.api.plugin.PluginCallback;
+import org.briarproject.bramble.api.plugin.TransportConnectionReader;
+import org.briarproject.bramble.api.plugin.TransportConnectionWriter;
+import org.briarproject.bramble.api.plugin.TransportId;
+import org.briarproject.bramble.api.plugin.simplex.SimplexPlugin;
+import org.briarproject.bramble.api.properties.TransportProperties;
+
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.OutputStream;
+import java.util.Collection;
+import java.util.logging.Logger;
+
+import javax.annotation.concurrent.Immutable;
+
+import static java.util.Collections.singletonMap;
+import static java.util.logging.Level.WARNING;
+import static java.util.logging.Logger.getLogger;
+import static org.briarproject.bramble.api.plugin.Plugin.State.ACTIVE;
+import static org.briarproject.bramble.api.plugin.file.RemovableDriveConstants.ID;
+import static org.briarproject.bramble.api.plugin.file.RemovableDriveConstants.PROP_SUPPORTED;
+import static org.briarproject.bramble.util.LogUtils.logException;
+
+@Immutable
+@NotNullByDefault
+abstract class AbstractRemovableDrivePlugin implements SimplexPlugin {
+
+	private static final Logger LOG =
+			getLogger(AbstractRemovableDrivePlugin.class.getName());
+
+	private final long maxLatency;
+	private final PluginCallback callback;
+
+	abstract InputStream openInputStream(TransportProperties p)
+			throws IOException;
+
+	abstract OutputStream openOutputStream(TransportProperties p)
+			throws IOException;
+
+	AbstractRemovableDrivePlugin(PluginCallback callback, long maxLatency) {
+		this.callback = callback;
+		this.maxLatency = maxLatency;
+	}
+
+	@Override
+	public TransportId getId() {
+		return ID;
+	}
+
+	@Override
+	public long getMaxLatency() {
+		return maxLatency;
+	}
+
+	@Override
+	public int getMaxIdleTime() {
+		// Unused for simplex transports
+		throw new UnsupportedOperationException();
+	}
+
+	@Override
+	public void start() {
+		callback.mergeLocalProperties(
+				new TransportProperties(singletonMap(PROP_SUPPORTED, "true")));
+	}
+
+	@Override
+	public void stop() {
+	}
+
+	@Override
+	public State getState() {
+		return ACTIVE;
+	}
+
+	@Override
+	public int getReasonsDisabled() {
+		return 0;
+	}
+
+	@Override
+	public boolean shouldPoll() {
+		return false;
+	}
+
+	@Override
+	public int getPollingInterval() {
+		throw new UnsupportedOperationException();
+	}
+
+	@Override
+	public void poll(
+			Collection<Pair<TransportProperties, ConnectionHandler>> properties) {
+		throw new UnsupportedOperationException();
+	}
+
+	@Override
+	public boolean isLossyAndCheap() {
+		return true;
+	}
+
+	@Override
+	public TransportConnectionReader createReader(TransportProperties p) {
+		try {
+			return new TransportInputStreamReader(openInputStream(p));
+		} catch (IOException e) {
+			logException(LOG, WARNING, e);
+			return null;
+		}
+	}
+
+	@Override
+	public TransportConnectionWriter createWriter(TransportProperties p) {
+		try {
+			return new TransportOutputStreamWriter(this, openOutputStream(p));
+		} catch (IOException e) {
+			logException(LOG, WARNING, e);
+			return null;
+		}
+	}
+}

bramble-core/src/main/java/org/briarproject/bramble/plugin/file/FilePlugin.java

@@ -15,8 +15,8 @@ import java.util.logging.Logger;
 
 import static java.util.logging.Level.WARNING;
 import static java.util.logging.Logger.getLogger;
-import static org.briarproject.bramble.api.plugin.FileConstants.PROP_PATH;
 import static org.briarproject.bramble.api.plugin.Plugin.State.ACTIVE;
+import static org.briarproject.bramble.api.plugin.file.FileConstants.PROP_PATH;
 import static org.briarproject.bramble.util.LogUtils.logException;
 import static org.briarproject.bramble.util.StringUtils.isNullOrEmpty;
 
@@ -27,20 +27,20 @@ abstract class FilePlugin implements SimplexPlugin {
 			getLogger(FilePlugin.class.getName());
 
 	protected final PluginCallback callback;
-	protected final int maxLatency;
+	protected final long maxLatency;
 
 	protected abstract void writerFinished(File f, boolean exception);
 
 	protected abstract void readerFinished(File f, boolean exception,
 			boolean recognised);
 
-	FilePlugin(PluginCallback callback, int maxLatency) {
+	FilePlugin(PluginCallback callback, long maxLatency) {
 		this.callback = callback;
 		this.maxLatency = maxLatency;
 	}
 
 	@Override
-	public int getMaxLatency() {
+	public long getMaxLatency() {
 		return maxLatency;
 	}
 

bramble-core/src/main/java/org/briarproject/bramble/plugin/file/FileTransportWriter.java

@@ -27,7 +27,7 @@ class FileTransportWriter implements TransportConnectionWriter {
 	}
 
 	@Override
-	public int getMaxLatency() {
+	public long getMaxLatency() {
 		return plugin.getMaxLatency();
 	}
 
@@ -36,6 +36,11 @@ class FileTransportWriter implements TransportConnectionWriter {
 		return plugin.getMaxIdleTime();
 	}
 
+	@Override
+	public boolean isLossyAndCheap() {
+		return plugin.isLossyAndCheap();
+	}
+
 	@Override
 	public OutputStream getOutputStream() {
 		return out;

bramble-core/src/main/java/org/briarproject/bramble/plugin/file/RemovableDriveManagerImpl.java

@@ -0,0 +1,125 @@
+package org.briarproject.bramble.plugin.file;
+
+import org.briarproject.bramble.api.contact.ContactId;
+import org.briarproject.bramble.api.db.DatabaseComponent;
+import org.briarproject.bramble.api.db.DbException;
+import org.briarproject.bramble.api.lifecycle.IoExecutor;
+import org.briarproject.bramble.api.nullsafety.NotNullByDefault;
+import org.briarproject.bramble.api.plugin.file.RemovableDriveManager;
+import org.briarproject.bramble.api.plugin.file.RemovableDriveTask;
+import org.briarproject.bramble.api.properties.TransportProperties;
+import org.briarproject.bramble.api.properties.TransportPropertyManager;
+import org.briarproject.bramble.api.transport.KeyManager;
+
+import java.util.concurrent.Executor;
+
+import javax.annotation.Nullable;
+import javax.annotation.concurrent.GuardedBy;
+import javax.annotation.concurrent.ThreadSafe;
+import javax.inject.Inject;
+
+import static org.briarproject.bramble.api.plugin.file.RemovableDriveConstants.ID;
+import static org.briarproject.bramble.api.plugin.file.RemovableDriveConstants.PROP_SUPPORTED;
+import static org.briarproject.bramble.plugin.file.RemovableDrivePluginFactory.MAX_LATENCY;
+
+@ThreadSafe
+@NotNullByDefault
+class RemovableDriveManagerImpl
+		implements RemovableDriveManager, RemovableDriveTaskRegistry {
+
+	private final Executor ioExecutor;
+	private final DatabaseComponent db;
+	private final KeyManager keyManager;
+	private final TransportPropertyManager transportPropertyManager;
+	private final RemovableDriveTaskFactory taskFactory;
+	private final Object lock = new Object();
+
+	@GuardedBy("lock")
+	@Nullable
+	private RemovableDriveTask reader = null;
+	@GuardedBy("lock")
+	@Nullable
+	private RemovableDriveTask writer = null;
+
+	@Inject
+	RemovableDriveManagerImpl(
+			@IoExecutor Executor ioExecutor,
+			DatabaseComponent db,
+			KeyManager keyManager,
+			TransportPropertyManager transportPropertyManager,
+			RemovableDriveTaskFactory taskFactory) {
+		this.ioExecutor = ioExecutor;
+		this.db = db;
+		this.keyManager = keyManager;
+		this.transportPropertyManager = transportPropertyManager;
+		this.taskFactory = taskFactory;
+	}
+
+	@Nullable
+	@Override
+	public RemovableDriveTask getCurrentReaderTask() {
+		synchronized (lock) {
+			return reader;
+		}
+	}
+
+	@Nullable
+	@Override
+	public RemovableDriveTask getCurrentWriterTask() {
+		synchronized (lock) {
+			return writer;
+		}
+	}
+
+	@Override
+	public RemovableDriveTask startReaderTask(TransportProperties p) {
+		RemovableDriveTask created;
+		synchronized (lock) {
+			if (reader != null) return reader;
+			reader = created = taskFactory.createReader(this, p);
+		}
+		ioExecutor.execute(created);
+		return created;
+	}
+
+	@Override
+	public RemovableDriveTask startWriterTask(ContactId c,
+			TransportProperties p) {
+		RemovableDriveTask created;
+		synchronized (lock) {
+			if (writer != null) return writer;
+			writer = created = taskFactory.createWriter(this, c, p);
+		}
+		ioExecutor.execute(created);
+		return created;
+	}
+
+	@Override
+	public boolean isTransportSupportedByContact(ContactId c)
+			throws DbException {
+		if (!keyManager.canSendOutgoingStreams(c, ID)) return false;
+		TransportProperties p =
+				transportPropertyManager.getRemoteProperties(c, ID);
+		return "true".equals(p.get(PROP_SUPPORTED));
+	}
+
+	@Override
+	public boolean isWriterTaskNeeded(ContactId c) throws DbException {
+		return db.transactionWithResult(true, txn ->
+				db.containsAnythingToSend(txn, c, MAX_LATENCY, true));
+	}
+
+	@Override
+	public void removeReader(RemovableDriveTask task) {
+		synchronized (lock) {
+			if (reader == task) reader = null;
+		}
+	}
+
+	@Override
+	public void removeWriter(RemovableDriveTask task) {
+		synchronized (lock) {
+			if (writer == task) writer = null;
+		}
+	}
+}

bramble-core/src/main/java/org/briarproject/bramble/plugin/file/RemovableDriveModule.java

@@ -0,0 +1,25 @@
+package org.briarproject.bramble.plugin.file;
+
+import org.briarproject.bramble.api.plugin.file.RemovableDriveManager;
+
+import javax.inject.Singleton;
+
+import dagger.Module;
+import dagger.Provides;
+
+@Module
+public class RemovableDriveModule {
+
+	@Provides
+	@Singleton
+	RemovableDriveManager provideRemovableDriveManager(
+			RemovableDriveManagerImpl removableDriveManager) {
+		return removableDriveManager;
+	}
+
+	@Provides
+	RemovableDriveTaskFactory provideTaskFactory(
+			RemovableDriveTaskFactoryImpl taskFactory) {
+		return taskFactory;
+	}
+}

bramble-core/src/main/java/org/briarproject/bramble/plugin/file/RemovableDrivePlugin.java

@@ -0,0 +1,39 @@
+package org.briarproject.bramble.plugin.file;
+
+import org.briarproject.bramble.api.nullsafety.NotNullByDefault;
+import org.briarproject.bramble.api.plugin.PluginCallback;
+import org.briarproject.bramble.api.properties.TransportProperties;
+
+import java.io.FileInputStream;
+import java.io.FileOutputStream;
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.OutputStream;
+
+import javax.annotation.concurrent.Immutable;
+
+import static org.briarproject.bramble.api.plugin.file.RemovableDriveConstants.PROP_PATH;
+import static org.briarproject.bramble.util.StringUtils.isNullOrEmpty;
+
+@Immutable
+@NotNullByDefault
+class RemovableDrivePlugin extends AbstractRemovableDrivePlugin {
+
+	RemovableDrivePlugin(PluginCallback callback, long maxLatency) {
+		super(callback, maxLatency);
+	}
+
+	@Override
+	InputStream openInputStream(TransportProperties p) throws IOException {
+		String path = p.get(PROP_PATH);
+		if (isNullOrEmpty(path)) throw new IllegalArgumentException();
+		return new FileInputStream(path);
+	}
+
+	@Override
+	OutputStream openOutputStream(TransportProperties p) throws IOException {
+		String path = p.get(PROP_PATH);
+		if (isNullOrEmpty(path)) throw new IllegalArgumentException();
+		return new FileOutputStream(path);
+	}
+}

bramble-core/src/main/java/org/briarproject/bramble/plugin/file/RemovableDrivePluginFactory.java

@@ -0,0 +1,41 @@
+package org.briarproject.bramble.plugin.file;
+
+import org.briarproject.bramble.api.nullsafety.NotNullByDefault;
+import org.briarproject.bramble.api.plugin.PluginCallback;
+import org.briarproject.bramble.api.plugin.TransportId;
+import org.briarproject.bramble.api.plugin.simplex.SimplexPlugin;
+import org.briarproject.bramble.api.plugin.simplex.SimplexPluginFactory;
+
+import javax.annotation.Nullable;
+import javax.annotation.concurrent.Immutable;
+import javax.inject.Inject;
+
+import static java.util.concurrent.TimeUnit.DAYS;
+import static org.briarproject.bramble.api.plugin.file.RemovableDriveConstants.ID;
+
+@Immutable
+@NotNullByDefault
+public class RemovableDrivePluginFactory implements SimplexPluginFactory {
+
+	static final long MAX_LATENCY = DAYS.toMillis(28);
+
+	@Inject
+	RemovableDrivePluginFactory() {
+	}
+
+	@Override
+	public TransportId getId() {
+		return ID;
+	}
+
+	@Override
+	public long getMaxLatency() {
+		return MAX_LATENCY;
+	}
+
+	@Nullable
+	@Override
+	public SimplexPlugin createPlugin(PluginCallback callback) {
+		return new RemovableDrivePlugin(callback, MAX_LATENCY);
+	}
+}

bramble-core/src/main/java/org/briarproject/bramble/plugin/file/RemovableDriveReaderTask.java

@@ -0,0 +1,69 @@
+package org.briarproject.bramble.plugin.file;
+
+import org.briarproject.bramble.api.connection.ConnectionManager;
+import org.briarproject.bramble.api.event.EventBus;
+import org.briarproject.bramble.api.nullsafety.NotNullByDefault;
+import org.briarproject.bramble.api.plugin.PluginManager;
+import org.briarproject.bramble.api.plugin.TransportConnectionReader;
+import org.briarproject.bramble.api.properties.TransportProperties;
+
+import java.io.IOException;
+import java.io.InputStream;
+import java.util.concurrent.Executor;
+import java.util.logging.Logger;
+
+import static java.util.logging.Logger.getLogger;
+import static org.briarproject.bramble.api.plugin.file.RemovableDriveConstants.ID;
+
+@NotNullByDefault
+class RemovableDriveReaderTask extends RemovableDriveTaskImpl {
+
+	private final static Logger LOG =
+			getLogger(RemovableDriveReaderTask.class.getName());
+
+	RemovableDriveReaderTask(
+			Executor eventExecutor,
+			PluginManager pluginManager,
+			ConnectionManager connectionManager,
+			EventBus eventBus,
+			RemovableDriveTaskRegistry registry,
+			TransportProperties transportProperties) {
+		super(eventExecutor, pluginManager, connectionManager, eventBus,
+				registry, transportProperties);
+	}
+
+	@Override
+	public void run() {
+		TransportConnectionReader r =
+				getPlugin().createReader(transportProperties);
+		if (r == null) {
+			LOG.warning("Failed to create reader");
+			registry.removeReader(this);
+			setSuccess(false);
+			return;
+		}
+		connectionManager.manageIncomingConnection(ID, new DecoratedReader(r));
+	}
+
+	private class DecoratedReader implements TransportConnectionReader {
+
+		private final TransportConnectionReader delegate;
+
+		private DecoratedReader(TransportConnectionReader delegate) {
+			this.delegate = delegate;
+		}
+
+		@Override
+		public InputStream getInputStream() throws IOException {
+			return delegate.getInputStream();
+		}
+
+		@Override
+		public void dispose(boolean exception, boolean recognised)
+				throws IOException {
+			delegate.dispose(exception, recognised);
+			registry.removeReader(RemovableDriveReaderTask.this);
+			setSuccess(!exception && recognised);
+		}
+	}
+}

bramble-core/src/main/java/org/briarproject/bramble/plugin/file/RemovableDriveTaskFactory.java

@@ -0,0 +1,16 @@
+package org.briarproject.bramble.plugin.file;
+
+import org.briarproject.bramble.api.contact.ContactId;
+import org.briarproject.bramble.api.nullsafety.NotNullByDefault;
+import org.briarproject.bramble.api.plugin.file.RemovableDriveTask;
+import org.briarproject.bramble.api.properties.TransportProperties;
+
+@NotNullByDefault
+interface RemovableDriveTaskFactory {
+
+	RemovableDriveTask createReader(RemovableDriveTaskRegistry registry,
+			TransportProperties p);
+
+	RemovableDriveTask createWriter(RemovableDriveTaskRegistry registry,
+			ContactId c, TransportProperties p);
+}

bramble-core/src/main/java/org/briarproject/bramble/plugin/file/RemovableDriveTaskFactoryImpl.java

@@ -0,0 +1,55 @@
+package org.briarproject.bramble.plugin.file;
+
+import org.briarproject.bramble.api.connection.ConnectionManager;
+import org.briarproject.bramble.api.contact.ContactId;
+import org.briarproject.bramble.api.db.DatabaseComponent;
+import org.briarproject.bramble.api.event.EventBus;
+import org.briarproject.bramble.api.event.EventExecutor;
+import org.briarproject.bramble.api.nullsafety.NotNullByDefault;
+import org.briarproject.bramble.api.plugin.PluginManager;
+import org.briarproject.bramble.api.plugin.file.RemovableDriveTask;
+import org.briarproject.bramble.api.properties.TransportProperties;
+
+import java.util.concurrent.Executor;
+
+import javax.annotation.concurrent.Immutable;
+import javax.inject.Inject;
+
+@Immutable
+@NotNullByDefault
+class RemovableDriveTaskFactoryImpl implements RemovableDriveTaskFactory {
+
+	private final DatabaseComponent db;
+	private final Executor eventExecutor;
+	private final PluginManager pluginManager;
+	private final ConnectionManager connectionManager;
+	private final EventBus eventBus;
+
+	@Inject
+	RemovableDriveTaskFactoryImpl(
+			DatabaseComponent db,
+			@EventExecutor Executor eventExecutor,
+			PluginManager pluginManager,
+			ConnectionManager connectionManager,
+			EventBus eventBus) {
+		this.db = db;
+		this.eventExecutor = eventExecutor;
+		this.pluginManager = pluginManager;
+		this.connectionManager = connectionManager;
+		this.eventBus = eventBus;
+	}
+
+	@Override
+	public RemovableDriveTask createReader(RemovableDriveTaskRegistry registry,
+			TransportProperties p) {
+		return new RemovableDriveReaderTask(eventExecutor, pluginManager,
+				connectionManager, eventBus, registry, p);
+	}
+
+	@Override
+	public RemovableDriveTask createWriter(RemovableDriveTaskRegistry registry,
+			ContactId c, TransportProperties p) {
+		return new RemovableDriveWriterTask(db, eventExecutor, pluginManager,
+				connectionManager, eventBus, registry, c, p);
+	}
+}

bramble-core/src/main/java/org/briarproject/bramble/plugin/file/RemovableDriveTaskImpl.java

@@ -0,0 +1,114 @@
+package org.briarproject.bramble.plugin.file;
+
+import org.briarproject.bramble.api.Consumer;
+import org.briarproject.bramble.api.connection.ConnectionManager;
+import org.briarproject.bramble.api.event.EventBus;
+import org.briarproject.bramble.api.nullsafety.NotNullByDefault;
+import org.briarproject.bramble.api.plugin.PluginManager;
+import org.briarproject.bramble.api.plugin.file.RemovableDriveTask;
+import org.briarproject.bramble.api.plugin.simplex.SimplexPlugin;
+import org.briarproject.bramble.api.properties.TransportProperties;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.concurrent.Executor;
+
+import javax.annotation.concurrent.GuardedBy;
+import javax.annotation.concurrent.ThreadSafe;
+
+import static java.lang.Math.min;
+import static org.briarproject.bramble.api.nullsafety.NullSafety.requireNonNull;
+import static org.briarproject.bramble.api.plugin.file.RemovableDriveConstants.ID;
+
+@ThreadSafe
+@NotNullByDefault
+abstract class RemovableDriveTaskImpl implements RemovableDriveTask {
+
+	private final Executor eventExecutor;
+	private final PluginManager pluginManager;
+	final ConnectionManager connectionManager;
+	final EventBus eventBus;
+	final RemovableDriveTaskRegistry registry;
+	final TransportProperties transportProperties;
+
+	private final Object lock = new Object();
+	@GuardedBy("lock")
+	private final List<Consumer<State>> observers = new ArrayList<>();
+	@GuardedBy("lock")
+	private State state = new State(0, 0, false, false);
+
+	RemovableDriveTaskImpl(
+			Executor eventExecutor,
+			PluginManager pluginManager,
+			ConnectionManager connectionManager,
+			EventBus eventBus,
+			RemovableDriveTaskRegistry registry,
+			TransportProperties transportProperties) {
+		this.eventExecutor = eventExecutor;
+		this.pluginManager = pluginManager;
+		this.connectionManager = connectionManager;
+		this.eventBus = eventBus;
+		this.registry = registry;
+		this.transportProperties = transportProperties;
+	}
+
+	@Override
+	public TransportProperties getTransportProperties() {
+		return transportProperties;
+	}
+
+	@Override
+	public void addObserver(Consumer<State> o) {
+		State state;
+		synchronized (lock) {
+			observers.add(o);
+			state = this.state;
+			eventExecutor.execute(() -> o.accept(state));
+		}
+	}
+
+	@Override
+	public void removeObserver(Consumer<State> o) {
+		synchronized (lock) {
+			observers.remove(o);
+		}
+	}
+
+	SimplexPlugin getPlugin() {
+		return (SimplexPlugin) requireNonNull(pluginManager.getPlugin(ID));
+	}
+
+	void setTotal(long total) {
+		synchronized (lock) {
+			state = new State(state.getDone(), total, state.isFinished(),
+					state.isSuccess());
+			notifyObservers();
+		}
+	}
+
+	void addDone(long done) {
+		synchronized (lock) {
+			// Done and total come from different sources; make them consistent
+			done = min(state.getDone() + done, state.getTotal());
+			state = new State(done, state.getTotal(), state.isFinished(),
+					state.isSuccess());
+			notifyObservers();
+		}
+	}
+
+	void setSuccess(boolean success) {
+		synchronized (lock) {
+			state = new State(state.getDone(), state.getTotal(), true, success);
+			notifyObservers();
+		}
+	}
+
+	@GuardedBy("lock")
+	private void notifyObservers() {
+		List<Consumer<State>> observers = new ArrayList<>(this.observers);
+		State state = this.state;
+		eventExecutor.execute(() -> {
+			for (Consumer<State> o : observers) o.accept(state);
+		});
+	}
+}

bramble-core/src/main/java/org/briarproject/bramble/plugin/file/RemovableDriveTaskRegistry.java

@@ -0,0 +1,12 @@
+package org.briarproject.bramble.plugin.file;
+
+import org.briarproject.bramble.api.nullsafety.NotNullByDefault;
+import org.briarproject.bramble.api.plugin.file.RemovableDriveTask;
+
+@NotNullByDefault
+interface RemovableDriveTaskRegistry {
+
+	void removeReader(RemovableDriveTask task);
+
+	void removeWriter(RemovableDriveTask task);
+}

bramble-core/src/main/java/org/briarproject/bramble/plugin/file/RemovableDriveWriterTask.java

@@ -0,0 +1,126 @@
+package org.briarproject.bramble.plugin.file;
+
+import org.briarproject.bramble.api.connection.ConnectionManager;
+import org.briarproject.bramble.api.contact.ContactId;
+import org.briarproject.bramble.api.db.DatabaseComponent;
+import org.briarproject.bramble.api.db.DbException;
+import org.briarproject.bramble.api.event.Event;
+import org.briarproject.bramble.api.event.EventBus;
+import org.briarproject.bramble.api.event.EventListener;
+import org.briarproject.bramble.api.nullsafety.NotNullByDefault;
+import org.briarproject.bramble.api.plugin.PluginManager;
+import org.briarproject.bramble.api.plugin.TransportConnectionWriter;
+import org.briarproject.bramble.api.plugin.simplex.SimplexPlugin;
+import org.briarproject.bramble.api.properties.TransportProperties;
+import org.briarproject.bramble.api.sync.event.MessagesSentEvent;
+
+import java.io.IOException;
+import java.io.OutputStream;
+import java.util.concurrent.Executor;
+import java.util.logging.Logger;
+
+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.plugin.file.RemovableDriveConstants.ID;
+import static org.briarproject.bramble.util.LogUtils.logException;
+
+@NotNullByDefault
+class RemovableDriveWriterTask extends RemovableDriveTaskImpl
+		implements EventListener {
+
+	private static final Logger LOG =
+			getLogger(RemovableDriveWriterTask.class.getName());
+
+	private final DatabaseComponent db;
+	private final ContactId contactId;
+
+	RemovableDriveWriterTask(
+			DatabaseComponent db,
+			Executor eventExecutor,
+			PluginManager pluginManager,
+			ConnectionManager connectionManager,
+			EventBus eventBus,
+			RemovableDriveTaskRegistry registry,
+			ContactId contactId,
+			TransportProperties transportProperties) {
+		super(eventExecutor, pluginManager, connectionManager, eventBus,
+				registry, transportProperties);
+		this.db = db;
+		this.contactId = contactId;
+	}
+
+	@Override
+	public void run() {
+		SimplexPlugin plugin = getPlugin();
+		TransportConnectionWriter w = plugin.createWriter(transportProperties);
+		if (w == null) {
+			LOG.warning("Failed to create writer");
+			registry.removeWriter(this);
+			setSuccess(false);
+			return;
+		}
+		try {
+			setTotal(db.transactionWithResult(true, txn ->
+					db.getUnackedMessageBytesToSend(txn, contactId)));
+		} catch (DbException e) {
+			logException(LOG, WARNING, e);
+			registry.removeWriter(this);
+			setSuccess(false);
+			return;
+		}
+		eventBus.addListener(this);
+		connectionManager.manageOutgoingConnection(contactId, ID,
+				new DecoratedWriter(w));
+	}
+
+	@Override
+	public void eventOccurred(Event e) {
+		if (e instanceof MessagesSentEvent) {
+			MessagesSentEvent m = (MessagesSentEvent) e;
+			if (contactId.equals(m.getContactId())) {
+				if (LOG.isLoggable(INFO)) {
+					LOG.info(m.getMessageIds().size() + " messages sent");
+				}
+				addDone(m.getTotalLength());
+			}
+		}
+	}
+
+	private class DecoratedWriter implements TransportConnectionWriter {
+
+		private final TransportConnectionWriter delegate;
+
+		private DecoratedWriter(TransportConnectionWriter delegate) {
+			this.delegate = delegate;
+		}
+
+		@Override
+		public long getMaxLatency() {
+			return delegate.getMaxLatency();
+		}
+
+		@Override
+		public int getMaxIdleTime() {
+			return delegate.getMaxIdleTime();
+		}
+
+		@Override
+		public boolean isLossyAndCheap() {
+			return delegate.isLossyAndCheap();
+		}
+
+		@Override
+		public OutputStream getOutputStream() throws IOException {
+			return delegate.getOutputStream();
+		}
+
+		@Override
+		public void dispose(boolean exception) throws IOException {
+			delegate.dispose(exception);
+			registry.removeWriter(RemovableDriveWriterTask.this);
+			eventBus.removeListener(RemovableDriveWriterTask.this);
+			setSuccess(!exception);
+		}
+	}
+}

bramble-core/src/main/java/org/briarproject/bramble/plugin/file/TransportInputStreamReader.java

@@ -0,0 +1,34 @@
+package org.briarproject.bramble.plugin.file;
+
+import org.briarproject.bramble.api.nullsafety.NotNullByDefault;
+import org.briarproject.bramble.api.plugin.TransportConnectionReader;
+
+import java.io.InputStream;
+import java.util.logging.Logger;
+
+import static java.util.logging.Level.WARNING;
+import static java.util.logging.Logger.getLogger;
+import static org.briarproject.bramble.util.IoUtils.tryToClose;
+
+@NotNullByDefault
+class TransportInputStreamReader implements TransportConnectionReader {
+
+	private static final Logger LOG =
+			getLogger(TransportInputStreamReader.class.getName());
+
+	private final InputStream in;
+
+	TransportInputStreamReader(InputStream in) {
+		this.in = in;
+	}
+
+	@Override
+	public InputStream getInputStream() {
+		return in;
+	}
+
+	@Override
+	public void dispose(boolean exception, boolean recognised) {
+		tryToClose(in, LOG, WARNING);
+	}
+}

bramble-core/src/main/java/org/briarproject/bramble/plugin/file/TransportOutputStreamWriter.java

@@ -0,0 +1,52 @@
+package org.briarproject.bramble.plugin.file;
+
+import org.briarproject.bramble.api.nullsafety.NotNullByDefault;
+import org.briarproject.bramble.api.plugin.TransportConnectionWriter;
+import org.briarproject.bramble.api.plugin.simplex.SimplexPlugin;
+
+import java.io.OutputStream;
+import java.util.logging.Logger;
+
+import static java.util.logging.Level.WARNING;
+import static java.util.logging.Logger.getLogger;
+import static org.briarproject.bramble.util.IoUtils.tryToClose;
+
+@NotNullByDefault
+class TransportOutputStreamWriter implements TransportConnectionWriter {
+
+	private static final Logger LOG =
+			getLogger(TransportOutputStreamWriter.class.getName());
+
+	private final SimplexPlugin plugin;
+	private final OutputStream out;
+
+	TransportOutputStreamWriter(SimplexPlugin plugin, OutputStream out) {
+		this.plugin = plugin;
+		this.out = out;
+	}
+
+	@Override
+	public long getMaxLatency() {
+		return plugin.getMaxLatency();
+	}
+
+	@Override
+	public int getMaxIdleTime() {
+		return plugin.getMaxIdleTime();
+	}
+
+	@Override
+	public boolean isLossyAndCheap() {
+		return plugin.isLossyAndCheap();
+	}
+
+	@Override
+	public OutputStream getOutputStream() {
+		return out;
+	}
+
+	@Override
+	public void dispose(boolean exception) {
+		tryToClose(out, LOG, WARNING);
+	}
+}

bramble-core/src/main/java/org/briarproject/bramble/plugin/tcp/LanTcpPlugin.java

@@ -92,7 +92,7 @@ class LanTcpPlugin extends TcpPlugin {
 			Executor wakefulIoExecutor,
 			Backoff backoff,
 			PluginCallback callback,
-			int maxLatency,
+			long maxLatency,
 			int maxIdleTime,
 			int connectionTimeout) {
 		super(ioExecutor, wakefulIoExecutor, backoff, callback, maxLatency,

bramble-core/src/main/java/org/briarproject/bramble/plugin/tcp/LanTcpPluginFactory.java

@@ -50,7 +50,7 @@ public class LanTcpPluginFactory implements DuplexPluginFactory {
 	}
 
 	@Override
-	public int getMaxLatency() {
+	public long getMaxLatency() {
 		return MAX_LATENCY;
 	}
 

bramble-core/src/main/java/org/briarproject/bramble/plugin/tcp/TcpPlugin.java

@@ -69,8 +69,8 @@ abstract class TcpPlugin implements DuplexPlugin, EventListener {
 	protected final Executor ioExecutor, wakefulIoExecutor, bindExecutor;
 	protected final Backoff backoff;
 	protected final PluginCallback callback;
-	protected final int maxLatency, maxIdleTime;
-	protected final int connectionTimeout, socketTimeout;
+	protected final long maxLatency;
+	protected final int maxIdleTime, connectionTimeout, socketTimeout;
 	protected final AtomicBoolean used = new AtomicBoolean(false);
 	protected final PluginState state = new PluginState();
 
@@ -111,7 +111,7 @@ abstract class TcpPlugin implements DuplexPlugin, EventListener {
 			Executor wakefulIoExecutor,
 			Backoff backoff,
 			PluginCallback callback,
-			int maxLatency,
+			long maxLatency,
 			int maxIdleTime,
 			int connectionTimeout) {
 		this.ioExecutor = ioExecutor;
@@ -129,7 +129,7 @@ abstract class TcpPlugin implements DuplexPlugin, EventListener {
 	}
 
 	@Override
-	public int getMaxLatency() {
+	public long getMaxLatency() {
 		return maxLatency;
 	}
 

bramble-core/src/main/java/org/briarproject/bramble/plugin/tcp/WanTcpPlugin.java

@@ -35,7 +35,7 @@ class WanTcpPlugin extends TcpPlugin {
 			Backoff backoff,
 			PortMapper portMapper,
 			PluginCallback callback,
-			int maxLatency,
+			long maxLatency,
 			int maxIdleTime,
 			int connectionTimeout) {
 		super(ioExecutor, wakefulIoExecutor, backoff, callback, maxLatency,

bramble-core/src/main/java/org/briarproject/bramble/plugin/tcp/WanTcpPluginFactory.java

@@ -54,7 +54,7 @@ public class WanTcpPluginFactory implements DuplexPluginFactory {
 	}
 
 	@Override
-	public int getMaxLatency() {
+	public long getMaxLatency() {
 		return MAX_LATENCY;
 	}
 

bramble-core/src/main/java/org/briarproject/bramble/plugin/tor/TorPlugin.java

@@ -131,7 +131,8 @@ abstract class TorPlugin implements DuplexPlugin, EventHandler, EventListener {
 	private final String architecture;
 	private final CircumventionProvider circumventionProvider;
 	private final ResourceProvider resourceProvider;
-	private final int maxLatency, maxIdleTime, socketTimeout;
+	private final long maxLatency;
+	private final int maxIdleTime, socketTimeout;
 	private final File torDirectory, geoIpFile, configFile;
 	private final File doneFile, cookieFile;
 	private final AtomicBoolean used = new AtomicBoolean(false);
@@ -159,7 +160,7 @@ abstract class TorPlugin implements DuplexPlugin, EventHandler, EventListener {
 			TorRendezvousCrypto torRendezvousCrypto,
 			PluginCallback callback,
 			String architecture,
-			int maxLatency,
+			long maxLatency,
 			int maxIdleTime,
 			File torDirectory) {
 		this.ioExecutor = ioExecutor;
@@ -204,7 +205,7 @@ abstract class TorPlugin implements DuplexPlugin, EventHandler, EventListener {
 	}
 
 	@Override
-	public int getMaxLatency() {
+	public long getMaxLatency() {
 		return maxLatency;
 	}
 

bramble-core/src/main/java/org/briarproject/bramble/properties/TransportPropertyManagerImpl.java

@@ -44,6 +44,7 @@ import static org.briarproject.bramble.api.properties.TransportPropertyConstants
 import static org.briarproject.bramble.api.properties.TransportPropertyConstants.MSG_KEY_TRANSPORT_ID;
 import static org.briarproject.bramble.api.properties.TransportPropertyConstants.MSG_KEY_VERSION;
 import static org.briarproject.bramble.api.properties.TransportPropertyConstants.REFLECTED_PROPERTY_PREFIX;
+import static org.briarproject.bramble.api.sync.validation.IncomingMessageHook.DeliveryAction.ACCEPT_DO_NOT_SHARE;
 import static org.briarproject.bramble.util.StringUtils.isNullOrEmpty;
 
 @Immutable
@@ -115,8 +116,8 @@ class TransportPropertyManagerImpl implements TransportPropertyManager,
 	}
 
 	@Override
-	public boolean incomingMessage(Transaction txn, Message m, Metadata meta)
-			throws DbException, InvalidMessageException {
+	public DeliveryAction incomingMessage(Transaction txn, Message m,
+			Metadata meta) throws DbException, InvalidMessageException {
 		try {
 			// Find the latest update for this transport, if any
 			BdfDictionary d = metadataParser.parse(meta);
@@ -131,14 +132,14 @@ class TransportPropertyManagerImpl implements TransportPropertyManager,
 					// We've already received a newer update - delete this one
 					db.deleteMessage(txn, m.getId());
 					db.deleteMessageMetadata(txn, m.getId());
-					return false;
+					return ACCEPT_DO_NOT_SHARE;
 				}
 			}
 			txn.attach(new RemoteTransportPropertiesUpdatedEvent(t));
 		} catch (FormatException e) {
 			throw new InvalidMessageException(e);
 		}
-		return false;
+		return ACCEPT_DO_NOT_SHARE;
 	}
 
 	@Override

bramble-core/src/main/java/org/briarproject/bramble/sync/DuplexOutgoingSession.java

@@ -77,7 +77,7 @@ class DuplexOutgoingSession implements SyncSession, EventListener {
 	private final Clock clock;
 	private final ContactId contactId;
 	private final TransportId transportId;
-	private final int maxLatency, maxIdleTime;
+	private final long maxLatency, maxIdleTime;
 	private final StreamWriter streamWriter;
 	private final SyncRecordWriter recordWriter;
 	@Nullable
@@ -95,7 +95,7 @@ class DuplexOutgoingSession implements SyncSession, EventListener {
 
 	DuplexOutgoingSession(DatabaseComponent db, Executor dbExecutor,
 			EventBus eventBus, Clock clock, ContactId contactId,
-			TransportId transportId, int maxLatency, int maxIdleTime,
+			TransportId transportId, long maxLatency, int maxIdleTime,
 			StreamWriter streamWriter, SyncRecordWriter recordWriter,
 			@Nullable Priority priority) {
 		this.db = db;

bramble-core/src/main/java/org/briarproject/bramble/sync/SimplexOutgoingSession.java

@@ -15,6 +15,7 @@ import org.briarproject.bramble.api.plugin.TransportId;
 import org.briarproject.bramble.api.plugin.event.TransportInactiveEvent;
 import org.briarproject.bramble.api.sync.Ack;
 import org.briarproject.bramble.api.sync.Message;
+import org.briarproject.bramble.api.sync.MessageId;
 import org.briarproject.bramble.api.sync.SyncRecordWriter;
 import org.briarproject.bramble.api.sync.SyncSession;
 import org.briarproject.bramble.api.sync.Versions;
@@ -22,7 +23,11 @@ import org.briarproject.bramble.api.sync.event.CloseSyncConnectionsEvent;
 import org.briarproject.bramble.api.transport.StreamWriter;
 
 import java.io.IOException;
+import java.util.ArrayList;
 import java.util.Collection;
+import java.util.Iterator;
+import java.util.Map;
+import java.util.Map.Entry;
 import java.util.concurrent.BlockingQueue;
 import java.util.concurrent.Executor;
 import java.util.concurrent.LinkedBlockingQueue;
@@ -60,7 +65,8 @@ class SimplexOutgoingSession implements SyncSession, EventListener {
 	private final EventBus eventBus;
 	private final ContactId contactId;
 	private final TransportId transportId;
-	private final int maxLatency;
+	private final long maxLatency;
+	private final boolean eager;
 	private final StreamWriter streamWriter;
 	private final SyncRecordWriter recordWriter;
 	private final AtomicInteger outstandingQueries;
@@ -70,7 +76,7 @@ class SimplexOutgoingSession implements SyncSession, EventListener {
 
 	SimplexOutgoingSession(DatabaseComponent db, Executor dbExecutor,
 			EventBus eventBus, ContactId contactId, TransportId transportId,
-			int maxLatency, StreamWriter streamWriter,
+			long maxLatency, boolean eager, StreamWriter streamWriter,
 			SyncRecordWriter recordWriter) {
 		this.db = db;
 		this.dbExecutor = dbExecutor;
@@ -78,6 +84,7 @@ class SimplexOutgoingSession implements SyncSession, EventListener {
 		this.contactId = contactId;
 		this.transportId = transportId;
 		this.maxLatency = maxLatency;
+		this.eager = eager;
 		this.streamWriter = streamWriter;
 		this.recordWriter = recordWriter;
 		outstandingQueries = new AtomicInteger(2); // One per type of record
@@ -92,8 +99,9 @@ class SimplexOutgoingSession implements SyncSession, EventListener {
 			// Send our supported protocol versions
 			recordWriter.writeVersions(new Versions(SUPPORTED_VERSIONS));
 			// Start a query for each type of record
-			dbExecutor.execute(new GenerateAck());
-			dbExecutor.execute(new GenerateBatch());
+			dbExecutor.execute(this::generateAck);
+			if (eager) dbExecutor.execute(this::loadUnackedMessageIds);
+			else dbExecutor.execute(this::generateBatch);
 			// Write records until interrupted or no more records to write
 			try {
 				while (!interrupted) {
@@ -138,81 +146,110 @@ class SimplexOutgoingSession implements SyncSession, EventListener {
 		}
 	}
 
-	private class GenerateAck implements Runnable {
-
-		@DatabaseExecutor
-		@Override
-		public void run() {
-			if (interrupted) return;
-			try {
-				Ack a = db.transactionWithNullableResult(false, txn ->
-						db.generateAck(txn, contactId, MAX_MESSAGE_IDS));
-				if (LOG.isLoggable(INFO))
-					LOG.info("Generated ack: " + (a != null));
-				if (a == null) decrementOutstandingQueries();
-				else writerTasks.add(new WriteAck(a));
-			} catch (DbException e) {
-				logException(LOG, WARNING, e);
-				interrupt();
+	@DatabaseExecutor
+	private void loadUnackedMessageIds() {
+		if (interrupted) return;
+		try {
+			Map<MessageId, Integer> ids = db.transactionWithResult(true, txn ->
+					db.getUnackedMessagesToSend(txn, contactId));
+			if (LOG.isLoggable(INFO)) {
+				LOG.info(ids.size() + " unacked messages to send");
 			}
+			if (ids.isEmpty()) decrementOutstandingQueries();
+			else dbExecutor.execute(() -> generateEagerBatch(ids));
+		} catch (DbException e) {
+			logException(LOG, WARNING, e);
+			interrupt();
 		}
 	}
 
-	private class WriteAck implements ThrowingRunnable<IOException> {
-
-		private final Ack ack;
-
-		private WriteAck(Ack ack) {
-			this.ack = ack;
+	@DatabaseExecutor
+	private void generateEagerBatch(Map<MessageId, Integer> ids) {
+		if (interrupted) return;
+		// Take some message IDs from `ids` to form a batch
+		Collection<MessageId> batchIds = new ArrayList<>();
+		long totalLength = 0;
+		Iterator<Entry<MessageId, Integer>> it = ids.entrySet().iterator();
+		while (it.hasNext()) {
+			// Check whether the next message will fit in the batch
+			Entry<MessageId, Integer> e = it.next();
+			int length = e.getValue();
+			if (totalLength + length > MAX_RECORD_PAYLOAD_BYTES) break;
+			// Add the message to the batch
+			it.remove();
+			batchIds.add(e.getKey());
+			totalLength += length;
 		}
-
-		@IoExecutor
-		@Override
-		public void run() throws IOException {
-			if (interrupted) return;
-			recordWriter.writeAck(ack);
-			LOG.info("Sent ack");
-			dbExecutor.execute(new GenerateAck());
+		if (batchIds.isEmpty()) throw new AssertionError();
+		try {
+			Collection<Message> batch =
+					db.transactionWithResult(false, txn ->
+							db.generateBatch(txn, contactId, batchIds,
+									maxLatency));
+			writerTasks.add(() -> writeEagerBatch(batch, ids));
+		} catch (DbException e) {
+			logException(LOG, WARNING, e);
+			interrupt();
 		}
 	}
 
-	private class GenerateBatch implements Runnable {
+	@IoExecutor
+	private void writeEagerBatch(Collection<Message> batch,
+			Map<MessageId, Integer> ids) throws IOException {
+		if (interrupted) return;
+		for (Message m : batch) recordWriter.writeMessage(m);
+		LOG.info("Sent eager batch");
+		if (ids.isEmpty()) decrementOutstandingQueries();
+		else dbExecutor.execute(() -> generateEagerBatch(ids));
+	}
 
-		@DatabaseExecutor
-		@Override
-		public void run() {
-			if (interrupted) return;
-			try {
-				Collection<Message> b =
-						db.transactionWithNullableResult(false, txn ->
-								db.generateBatch(txn, contactId,
-										MAX_RECORD_PAYLOAD_BYTES, maxLatency));
-				if (LOG.isLoggable(INFO))
-					LOG.info("Generated batch: " + (b != null));
-				if (b == null) decrementOutstandingQueries();
-				else writerTasks.add(new WriteBatch(b));
-			} catch (DbException e) {
-				logException(LOG, WARNING, e);
-				interrupt();
-			}
+	@DatabaseExecutor
+	private void generateAck() {
+		if (interrupted) return;
+		try {
+			Ack a = db.transactionWithNullableResult(false, txn ->
+					db.generateAck(txn, contactId, MAX_MESSAGE_IDS));
+			if (LOG.isLoggable(INFO))
+				LOG.info("Generated ack: " + (a != null));
+			if (a == null) decrementOutstandingQueries();
+			else writerTasks.add(() -> writeAck(a));
+		} catch (DbException e) {
+			logException(LOG, WARNING, e);
+			interrupt();
 		}
 	}
 
-	private class WriteBatch implements ThrowingRunnable<IOException> {
+	@IoExecutor
+	private void writeAck(Ack ack) throws IOException {
+		if (interrupted) return;
+		recordWriter.writeAck(ack);
+		LOG.info("Sent ack");
+		dbExecutor.execute(this::generateAck);
+	}
 
-		private final Collection<Message> batch;
-
-		private WriteBatch(Collection<Message> batch) {
-			this.batch = batch;
+	@DatabaseExecutor
+	private void generateBatch() {
+		if (interrupted) return;
+		try {
+			Collection<Message> b =
+					db.transactionWithNullableResult(false, txn ->
+							db.generateBatch(txn, contactId,
+									MAX_RECORD_PAYLOAD_BYTES, maxLatency));
+			if (LOG.isLoggable(INFO))
+				LOG.info("Generated batch: " + (b != null));
+			if (b == null) decrementOutstandingQueries();
+			else writerTasks.add(() -> writeBatch(b));
+		} catch (DbException e) {
+			logException(LOG, WARNING, e);
+			interrupt();
 		}
+	}
 
-		@IoExecutor
-		@Override
-		public void run() throws IOException {
-			if (interrupted) return;
-			for (Message m : batch) recordWriter.writeMessage(m);
-			LOG.info("Sent batch");
-			dbExecutor.execute(new GenerateBatch());
-		}
+	@IoExecutor
+	private void writeBatch(Collection<Message> batch) throws IOException {
+		if (interrupted) return;
+		for (Message m : batch) recordWriter.writeMessage(m);
+		LOG.info("Sent batch");
+		dbExecutor.execute(this::generateBatch);
 	}
 }

bramble-core/src/main/java/org/briarproject/bramble/sync/SyncSessionFactoryImpl.java

@@ -60,17 +60,17 @@ class SyncSessionFactoryImpl implements SyncSessionFactory {
 
 	@Override
 	public SyncSession createSimplexOutgoingSession(ContactId c, TransportId t,
-			int maxLatency, StreamWriter streamWriter) {
+			long maxLatency, boolean eager, StreamWriter streamWriter) {
 		OutputStream out = streamWriter.getOutputStream();
 		SyncRecordWriter recordWriter =
 				recordWriterFactory.createRecordWriter(out);
 		return new SimplexOutgoingSession(db, dbExecutor, eventBus, c, t,
-				maxLatency, streamWriter, recordWriter);
+				maxLatency, eager, streamWriter, recordWriter);
 	}
 
 	@Override
 	public SyncSession createDuplexOutgoingSession(ContactId c, TransportId t,
-			int maxLatency, int maxIdleTime, StreamWriter streamWriter,
+			long maxLatency, int maxIdleTime, StreamWriter streamWriter,
 			@Nullable Priority priority) {
 		OutputStream out = streamWriter.getOutputStream();
 		SyncRecordWriter recordWriter =

bramble-core/src/main/java/org/briarproject/bramble/sync/validation/ValidationManagerImpl.java

@@ -20,6 +20,7 @@ import org.briarproject.bramble.api.sync.MessageContext;
 import org.briarproject.bramble.api.sync.MessageId;
 import org.briarproject.bramble.api.sync.event.MessageAddedEvent;
 import org.briarproject.bramble.api.sync.validation.IncomingMessageHook;
+import org.briarproject.bramble.api.sync.validation.IncomingMessageHook.DeliveryAction;
 import org.briarproject.bramble.api.sync.validation.MessageState;
 import org.briarproject.bramble.api.sync.validation.MessageValidator;
 import org.briarproject.bramble.api.sync.validation.ValidationManager;
@@ -40,6 +41,10 @@ import javax.inject.Inject;
 
 import static java.util.logging.Level.INFO;
 import static java.util.logging.Level.WARNING;
+import static org.briarproject.bramble.api.sync.validation.IncomingMessageHook.DeliveryAction.ACCEPT_DO_NOT_SHARE;
+import static org.briarproject.bramble.api.sync.validation.IncomingMessageHook.DeliveryAction.ACCEPT_SHARE;
+import static org.briarproject.bramble.api.sync.validation.IncomingMessageHook.DeliveryAction.DEFER;
+import static org.briarproject.bramble.api.sync.validation.IncomingMessageHook.DeliveryAction.REJECT;
 import static org.briarproject.bramble.api.sync.validation.MessageState.DELIVERED;
 import static org.briarproject.bramble.api.sync.validation.MessageState.INVALID;
 import static org.briarproject.bramble.api.sync.validation.MessageState.PENDING;
@@ -185,16 +190,19 @@ class ValidationManagerImpl implements ValidationManager, Service,
 						int majorVersion = g.getMajorVersion();
 						Metadata meta =
 								db.getMessageMetadataForValidator(txn, id);
-						DeliveryResult result =
+						DeliveryAction action =
 								deliverMessage(txn, m, c, majorVersion, meta);
-						if (result.valid) {
-							addPendingDependents(txn, id, pending);
-							if (result.share) {
-								db.setMessageShared(txn, id);
-								toShare.addAll(states.keySet());
-							}
-						} else {
+						if (action == REJECT) {
+							invalidateMessage(txn, id);
 							addDependentsToInvalidate(txn, id, invalidate);
+						} else if (action == ACCEPT_SHARE) {
+							db.setMessageState(txn, m.getId(), DELIVERED);
+							addPendingDependents(txn, id, pending);
+							db.setMessageShared(txn, id);
+							toShare.addAll(states.keySet());
+						} else if (action == ACCEPT_DO_NOT_SHARE) {
+							db.setMessageState(txn, m.getId(), DELIVERED);
+							addPendingDependents(txn, id, pending);
 						}
 					}
 				}
@@ -275,16 +283,21 @@ class ValidationManagerImpl implements ValidationManager, Service,
 					Metadata meta = context.getMetadata();
 					db.mergeMessageMetadata(txn, id, meta);
 					if (allDelivered) {
-						DeliveryResult result =
+						DeliveryAction action =
 								deliverMessage(txn, m, c, majorVersion, meta);
-						if (result.valid) {
-							addPendingDependents(txn, id, pending);
-							if (result.share) {
-								db.setMessageShared(txn, id);
-								toShare.addAll(dependencies);
-							}
-						} else {
+						if (action == REJECT) {
+							invalidateMessage(txn, id);
 							addDependentsToInvalidate(txn, id, invalidate);
+						} else if (action == DEFER) {
+							db.setMessageState(txn, id, PENDING);
+						} else if (action == ACCEPT_SHARE) {
+							db.setMessageState(txn, id, DELIVERED);
+							addPendingDependents(txn, id, pending);
+							db.setMessageShared(txn, id);
+							toShare.addAll(dependencies);
+						} else if (action == ACCEPT_DO_NOT_SHARE) {
+							db.setMessageState(txn, id, DELIVERED);
+							addPendingDependents(txn, id, pending);
 						}
 					} else {
 						db.setMessageState(txn, id, PENDING);
@@ -304,23 +317,21 @@ class ValidationManagerImpl implements ValidationManager, Service,
 	}
 
 	@DatabaseExecutor
-	private DeliveryResult deliverMessage(Transaction txn, Message m,
-			ClientId c, int majorVersion, Metadata meta) throws DbException {
-		// Deliver the message to the client if it's registered a hook
-		boolean shareMsg = false;
+	private DeliveryAction deliverMessage(Transaction txn, Message m,
+			ClientId c, int majorVersion, Metadata meta) {
+		// Deliver the message to the client if it has registered a hook
 		ClientMajorVersion cv = new ClientMajorVersion(c, majorVersion);
 		IncomingMessageHook hook = hooks.get(cv);
-		if (hook != null) {
-			try {
-				shareMsg = hook.incomingMessage(txn, m, meta);
-			} catch (InvalidMessageException e) {
-				logException(LOG, INFO, e);
-				invalidateMessage(txn, m.getId());
-				return new DeliveryResult(false, false);
-			}
+		if (hook == null) return ACCEPT_DO_NOT_SHARE;
+		try {
+			return hook.incomingMessage(txn, m, meta);
+		} catch (DbException e) {
+			logException(LOG, INFO, e);
+			return DEFER;
+		} catch (InvalidMessageException e) {
+			logException(LOG, INFO, e);
+			return REJECT;
 		}
-		db.setMessageState(txn, m.getId(), DELIVERED);
-		return new DeliveryResult(true, shareMsg);
 	}
 
 	@DatabaseExecutor
@@ -447,14 +458,4 @@ class ValidationManagerImpl implements ValidationManager, Service,
 			logException(LOG, WARNING, e);
 		}
 	}
-
-	private static class DeliveryResult {
-
-		private final boolean valid, share;
-
-		private DeliveryResult(boolean valid, boolean share) {
-			this.valid = valid;
-			this.share = share;
-		}
-	}
 }

bramble-core/src/main/java/org/briarproject/bramble/transport/KeyManagerImpl.java

@@ -19,9 +19,8 @@ import org.briarproject.bramble.api.lifecycle.Service;
 import org.briarproject.bramble.api.lifecycle.ServiceException;
 import org.briarproject.bramble.api.nullsafety.NotNullByDefault;
 import org.briarproject.bramble.api.plugin.PluginConfig;
+import org.briarproject.bramble.api.plugin.PluginFactory;
 import org.briarproject.bramble.api.plugin.TransportId;
-import org.briarproject.bramble.api.plugin.duplex.DuplexPluginFactory;
-import org.briarproject.bramble.api.plugin.simplex.SimplexPluginFactory;
 import org.briarproject.bramble.api.transport.KeyManager;
 import org.briarproject.bramble.api.transport.KeySetId;
 import org.briarproject.bramble.api.transport.StreamContext;
@@ -40,6 +39,7 @@ import javax.annotation.concurrent.ThreadSafe;
 import javax.inject.Inject;
 
 import static java.util.logging.Level.INFO;
+import static org.briarproject.bramble.api.sync.SyncConstants.MAX_TRANSPORT_LATENCY;
 
 @ThreadSafe
 @NotNullByDefault
@@ -51,7 +51,6 @@ class KeyManagerImpl implements KeyManager, Service, EventListener {
 	private final DatabaseComponent db;
 	private final Executor dbExecutor;
 	private final PluginConfig pluginConfig;
-	private final TransportKeyManagerFactory transportKeyManagerFactory;
 	private final TransportCrypto transportCrypto;
 
 	private final ConcurrentHashMap<TransportId, TransportKeyManager> managers;
@@ -61,34 +60,35 @@ class KeyManagerImpl implements KeyManager, Service, EventListener {
 	KeyManagerImpl(DatabaseComponent db,
 			@DatabaseExecutor Executor dbExecutor,
 			PluginConfig pluginConfig,
-			TransportKeyManagerFactory transportKeyManagerFactory,
-			TransportCrypto transportCrypto) {
+			TransportCrypto transportCrypto,
+			TransportKeyManagerFactory transportKeyManagerFactory) {
 		this.db = db;
 		this.dbExecutor = dbExecutor;
 		this.pluginConfig = pluginConfig;
-		this.transportKeyManagerFactory = transportKeyManagerFactory;
 		this.transportCrypto = transportCrypto;
 		managers = new ConcurrentHashMap<>();
+		for (PluginFactory<?> f : pluginConfig.getSimplexFactories()) {
+			TransportKeyManager m = transportKeyManagerFactory.
+					createTransportKeyManager(f.getId(), f.getMaxLatency());
+			managers.put(f.getId(), m);
+		}
+		for (PluginFactory<?> f : pluginConfig.getDuplexFactories()) {
+			TransportKeyManager m = transportKeyManagerFactory.
+					createTransportKeyManager(f.getId(), f.getMaxLatency());
+			managers.put(f.getId(), m);
+		}
 	}
 
 	@Override
 	public void startService() throws ServiceException {
 		if (used.getAndSet(true)) throw new IllegalStateException();
-		Map<TransportId, Integer> transports = new HashMap<>();
-		for (SimplexPluginFactory f : pluginConfig.getSimplexFactories())
-			transports.put(f.getId(), f.getMaxLatency());
-		for (DuplexPluginFactory f : pluginConfig.getDuplexFactories())
-			transports.put(f.getId(), f.getMaxLatency());
 		try {
 			db.transaction(false, txn -> {
-				for (Entry<TransportId, Integer> e : transports.entrySet())
-					db.addTransport(txn, e.getKey(), e.getValue());
-				for (Entry<TransportId, Integer> e : transports.entrySet()) {
-					TransportKeyManager m = transportKeyManagerFactory
-							.createTransportKeyManager(e.getKey(),
-									e.getValue());
-					managers.put(e.getKey(), m);
-					m.start(txn);
+				for (PluginFactory<?> f : pluginConfig.getSimplexFactories()) {
+					addTransport(txn, f);
+				}
+				for (PluginFactory<?> f : pluginConfig.getDuplexFactories()) {
+					addTransport(txn, f);
 				}
 			});
 		} catch (DbException e) {
@@ -96,14 +96,32 @@ class KeyManagerImpl implements KeyManager, Service, EventListener {
 		}
 	}
 
+	private void addTransport(Transaction txn, PluginFactory<?> f)
+			throws DbException {
+		long maxLatency = f.getMaxLatency();
+		if (maxLatency > MAX_TRANSPORT_LATENCY) {
+			throw new IllegalStateException();
+		}
+		db.addTransport(txn, f.getId(), maxLatency);
+		managers.get(f.getId()).start(txn);
+	}
+
 	@Override
 	public void stopService() {
 	}
 
 	@Override
-	public Map<TransportId, KeySetId> addRotationKeys(
-			Transaction txn, ContactId c, SecretKey rootKey, long timestamp,
-			boolean alice, boolean active) throws DbException {
+	public KeySetId addRotationKeys(Transaction txn, ContactId c,
+			TransportId t, SecretKey rootKey, long timestamp, boolean alice,
+			boolean active) throws DbException {
+		return withManager(t, m ->
+				m.addRotationKeys(txn, c, rootKey, timestamp, alice, active));
+	}
+
+	@Override
+	public Map<TransportId, KeySetId> addRotationKeys(Transaction txn,
+			ContactId c, SecretKey rootKey, long timestamp, boolean alice,
+			boolean active) throws DbException {
 		Map<TransportId, KeySetId> ids = new HashMap<>();
 		for (Entry<TransportId, TransportKeyManager> e : managers.entrySet()) {
 			TransportId t = e.getKey();
@@ -137,7 +155,7 @@ class KeyManagerImpl implements KeyManager, Service, EventListener {
 			PendingContactId p, PublicKey theirPublicKey, KeyPair ourKeyPair)
 			throws DbException, GeneralSecurityException {
 		SecretKey staticMasterKey = transportCrypto
-					.deriveStaticMasterKey(theirPublicKey, ourKeyPair);
+				.deriveStaticMasterKey(theirPublicKey, ourKeyPair);
 		SecretKey rootKey =
 				transportCrypto.deriveHandshakeRootKey(staticMasterKey, true);
 		boolean alice = transportCrypto.isAlice(theirPublicKey, ourKeyPair);

bramble-core/src/main/java/org/briarproject/bramble/transport/agreement/MessageEncoder.java

@@ -0,0 +1,22 @@
+package org.briarproject.bramble.transport.agreement;
+
+import org.briarproject.bramble.api.crypto.PublicKey;
+import org.briarproject.bramble.api.data.BdfDictionary;
+import org.briarproject.bramble.api.nullsafety.NotNullByDefault;
+import org.briarproject.bramble.api.plugin.TransportId;
+import org.briarproject.bramble.api.sync.GroupId;
+import org.briarproject.bramble.api.sync.Message;
+import org.briarproject.bramble.api.sync.MessageId;
+
+@NotNullByDefault
+interface MessageEncoder {
+
+	Message encodeKeyMessage(GroupId contactGroupId,
+			TransportId transportId, PublicKey publicKey);
+
+	Message encodeActivateMessage(GroupId contactGroupId,
+			TransportId transportId, MessageId previousMessageId);
+
+	BdfDictionary encodeMessageMetadata(TransportId transportId,
+			MessageType type, boolean local);
+}

bramble-core/src/main/java/org/briarproject/bramble/transport/agreement/MessageEncoderImpl.java

@@ -0,0 +1,77 @@
+package org.briarproject.bramble.transport.agreement;
+
+import org.briarproject.bramble.api.FormatException;
+import org.briarproject.bramble.api.client.ClientHelper;
+import org.briarproject.bramble.api.crypto.PublicKey;
+import org.briarproject.bramble.api.data.BdfDictionary;
+import org.briarproject.bramble.api.data.BdfEntry;
+import org.briarproject.bramble.api.data.BdfList;
+import org.briarproject.bramble.api.nullsafety.NotNullByDefault;
+import org.briarproject.bramble.api.plugin.TransportId;
+import org.briarproject.bramble.api.sync.GroupId;
+import org.briarproject.bramble.api.sync.Message;
+import org.briarproject.bramble.api.sync.MessageId;
+import org.briarproject.bramble.api.system.Clock;
+
+import javax.annotation.concurrent.Immutable;
+import javax.inject.Inject;
+
+import static org.briarproject.bramble.transport.agreement.MessageType.ACTIVATE;
+import static org.briarproject.bramble.transport.agreement.MessageType.KEY;
+import static org.briarproject.bramble.transport.agreement.TransportKeyAgreementConstants.MSG_KEY_IS_SESSION;
+import static org.briarproject.bramble.transport.agreement.TransportKeyAgreementConstants.MSG_KEY_LOCAL;
+import static org.briarproject.bramble.transport.agreement.TransportKeyAgreementConstants.MSG_KEY_MESSAGE_TYPE;
+import static org.briarproject.bramble.transport.agreement.TransportKeyAgreementConstants.MSG_KEY_TRANSPORT_ID;
+
+@Immutable
+@NotNullByDefault
+class MessageEncoderImpl implements MessageEncoder {
+
+	private final ClientHelper clientHelper;
+	private final Clock clock;
+
+	@Inject
+	MessageEncoderImpl(ClientHelper clientHelper, Clock clock) {
+		this.clientHelper = clientHelper;
+		this.clock = clock;
+	}
+
+	@Override
+	public Message encodeKeyMessage(GroupId contactGroupId,
+			TransportId transportId, PublicKey publicKey) {
+		BdfList body = BdfList.of(
+				KEY.getValue(),
+				transportId.getString(),
+				publicKey.getEncoded());
+		return encodeMessage(contactGroupId, body);
+	}
+
+	@Override
+	public Message encodeActivateMessage(GroupId contactGroupId,
+			TransportId transportId, MessageId previousMessageId) {
+		BdfList body = BdfList.of(
+				ACTIVATE.getValue(),
+				transportId.getString(),
+				previousMessageId);
+		return encodeMessage(contactGroupId, body);
+	}
+
+	@Override
+	public BdfDictionary encodeMessageMetadata(TransportId transportId,
+			MessageType type, boolean local) {
+		return BdfDictionary.of(
+				new BdfEntry(MSG_KEY_IS_SESSION, false),
+				new BdfEntry(MSG_KEY_TRANSPORT_ID, transportId.getString()),
+				new BdfEntry(MSG_KEY_MESSAGE_TYPE, type.getValue()),
+				new BdfEntry(MSG_KEY_LOCAL, local));
+	}
+
+	private Message encodeMessage(GroupId contactGroupId, BdfList body) {
+		try {
+			return clientHelper.createMessage(contactGroupId,
+					clock.currentTimeMillis(), clientHelper.toByteArray(body));
+		} catch (FormatException e) {
+			throw new AssertionError();
+		}
+	}
+}

bramble-core/src/main/java/org/briarproject/bramble/transport/agreement/MessageType.java

@@ -0,0 +1,29 @@
+package org.briarproject.bramble.transport.agreement;
+
+import org.briarproject.bramble.api.FormatException;
+import org.briarproject.bramble.api.nullsafety.NotNullByDefault;
+
+import javax.annotation.concurrent.Immutable;
+
+@Immutable
+@NotNullByDefault
+enum MessageType {
+
+	KEY(0),
+	ACTIVATE(1);
+
+	private final int value;
+
+	MessageType(int value) {
+		this.value = value;
+	}
+
+	int getValue() {
+		return value;
+	}
+
+	static MessageType fromValue(int value) throws FormatException {
+		for (MessageType t : values()) if (t.value == value) return t;
+		throw new FormatException();
+	}
+}

bramble-core/src/main/java/org/briarproject/bramble/transport/agreement/Session.java

@@ -0,0 +1,58 @@
+package org.briarproject.bramble.transport.agreement;
+
+import org.briarproject.bramble.api.crypto.KeyPair;
+import org.briarproject.bramble.api.nullsafety.NotNullByDefault;
+import org.briarproject.bramble.api.sync.MessageId;
+import org.briarproject.bramble.api.transport.KeySetId;
+
+import javax.annotation.Nullable;
+import javax.annotation.concurrent.Immutable;
+
+@Immutable
+@NotNullByDefault
+class Session {
+
+	private final State state;
+	@Nullable
+	private final MessageId lastLocalMessageId;
+	@Nullable
+	private final KeyPair localKeyPair;
+	@Nullable
+	private final Long localTimestamp;
+	@Nullable
+	private final KeySetId keySetId;
+
+	Session(State state, @Nullable MessageId lastLocalMessageId,
+			@Nullable KeyPair localKeyPair, @Nullable Long localTimestamp,
+			@Nullable KeySetId keySetId) {
+		this.state = state;
+		this.lastLocalMessageId = lastLocalMessageId;
+		this.localKeyPair = localKeyPair;
+		this.localTimestamp = localTimestamp;
+		this.keySetId = keySetId;
+	}
+
+	State getState() {
+		return state;
+	}
+
+	@Nullable
+	MessageId getLastLocalMessageId() {
+		return lastLocalMessageId;
+	}
+
+	@Nullable
+	KeyPair getLocalKeyPair() {
+		return localKeyPair;
+	}
+
+	@Nullable
+	Long getLocalTimestamp() {
+		return localTimestamp;
+	}
+
+	@Nullable
+	KeySetId getKeySetId() {
+		return keySetId;
+	}
+}

bramble-core/src/main/java/org/briarproject/bramble/transport/agreement/SessionEncoder.java

@@ -0,0 +1,13 @@
+package org.briarproject.bramble.transport.agreement;
+
+import org.briarproject.bramble.api.data.BdfDictionary;
+import org.briarproject.bramble.api.nullsafety.NotNullByDefault;
+import org.briarproject.bramble.api.plugin.TransportId;
+
+@NotNullByDefault
+interface SessionEncoder {
+
+	BdfDictionary encodeSession(Session s, TransportId transportId);
+
+	BdfDictionary getSessionQuery(TransportId transportId);
+}

bramble-core/src/main/java/org/briarproject/bramble/transport/agreement/SessionEncoderImpl.java

@@ -0,0 +1,68 @@
+package org.briarproject.bramble.transport.agreement;
+
+import org.briarproject.bramble.api.crypto.KeyPair;
+import org.briarproject.bramble.api.data.BdfDictionary;
+import org.briarproject.bramble.api.data.BdfEntry;
+import org.briarproject.bramble.api.nullsafety.NotNullByDefault;
+import org.briarproject.bramble.api.plugin.TransportId;
+import org.briarproject.bramble.api.transport.KeySetId;
+
+import javax.annotation.Nullable;
+import javax.annotation.concurrent.Immutable;
+import javax.inject.Inject;
+
+import static org.briarproject.bramble.api.data.BdfDictionary.NULL_VALUE;
+import static org.briarproject.bramble.transport.agreement.TransportKeyAgreementConstants.MSG_KEY_IS_SESSION;
+import static org.briarproject.bramble.transport.agreement.TransportKeyAgreementConstants.MSG_KEY_TRANSPORT_ID;
+import static org.briarproject.bramble.transport.agreement.TransportKeyAgreementConstants.SESSION_KEY_KEY_SET_ID;
+import static org.briarproject.bramble.transport.agreement.TransportKeyAgreementConstants.SESSION_KEY_LAST_LOCAL_MESSAGE_ID;
+import static org.briarproject.bramble.transport.agreement.TransportKeyAgreementConstants.SESSION_KEY_LOCAL_PRIVATE_KEY;
+import static org.briarproject.bramble.transport.agreement.TransportKeyAgreementConstants.SESSION_KEY_LOCAL_PUBLIC_KEY;
+import static org.briarproject.bramble.transport.agreement.TransportKeyAgreementConstants.SESSION_KEY_LOCAL_TIMESTAMP;
+import static org.briarproject.bramble.transport.agreement.TransportKeyAgreementConstants.SESSION_KEY_STATE;
+
+@Immutable
+@NotNullByDefault
+class SessionEncoderImpl implements SessionEncoder {
+
+	@Inject
+	SessionEncoderImpl() {
+	}
+
+	@Override
+	public BdfDictionary encodeSession(Session s, TransportId transportId) {
+		BdfDictionary meta = new BdfDictionary();
+		meta.put(MSG_KEY_IS_SESSION, true);
+		meta.put(MSG_KEY_TRANSPORT_ID, transportId.getString());
+		meta.put(SESSION_KEY_STATE, s.getState().getValue());
+		putNullable(meta, SESSION_KEY_LAST_LOCAL_MESSAGE_ID,
+				s.getLastLocalMessageId());
+		KeyPair localKeyPair = s.getLocalKeyPair();
+		if (localKeyPair == null) {
+			meta.put(SESSION_KEY_LOCAL_PUBLIC_KEY, NULL_VALUE);
+			meta.put(SESSION_KEY_LOCAL_PRIVATE_KEY, NULL_VALUE);
+		} else {
+			meta.put(SESSION_KEY_LOCAL_PUBLIC_KEY,
+					localKeyPair.getPublic().getEncoded());
+			meta.put(SESSION_KEY_LOCAL_PRIVATE_KEY,
+					localKeyPair.getPrivate().getEncoded());
+		}
+		putNullable(meta, SESSION_KEY_LOCAL_TIMESTAMP, s.getLocalTimestamp());
+		KeySetId keySetId = s.getKeySetId();
+		if (keySetId == null) meta.put(SESSION_KEY_KEY_SET_ID, NULL_VALUE);
+		else meta.put(SESSION_KEY_KEY_SET_ID, keySetId.getInt());
+		return meta;
+	}
+
+	@Override
+	public BdfDictionary getSessionQuery(TransportId transportId) {
+		return BdfDictionary.of(
+				new BdfEntry(MSG_KEY_IS_SESSION, true),
+				new BdfEntry(MSG_KEY_TRANSPORT_ID, transportId.getString()));
+	}
+
+	private void putNullable(BdfDictionary meta, String key,
+			@Nullable Object o) {
+		meta.put(key, o == null ? NULL_VALUE : o);
+	}
+}

bramble-core/src/main/java/org/briarproject/bramble/transport/agreement/SessionParser.java

@@ -0,0 +1,11 @@
+package org.briarproject.bramble.transport.agreement;
+
+import org.briarproject.bramble.api.FormatException;
+import org.briarproject.bramble.api.data.BdfDictionary;
+import org.briarproject.bramble.api.nullsafety.NotNullByDefault;
+
+@NotNullByDefault
+interface SessionParser {
+
+	Session parseSession(BdfDictionary meta) throws FormatException;
+}

bramble-core/src/main/java/org/briarproject/bramble/transport/agreement/SessionParserImpl.java

@@ -0,0 +1,67 @@
+package org.briarproject.bramble.transport.agreement;
+
+import org.briarproject.bramble.api.FormatException;
+import org.briarproject.bramble.api.crypto.KeyPair;
+import org.briarproject.bramble.api.crypto.PrivateKey;
+import org.briarproject.bramble.api.crypto.PublicKey;
+import org.briarproject.bramble.api.data.BdfDictionary;
+import org.briarproject.bramble.api.nullsafety.NotNullByDefault;
+import org.briarproject.bramble.api.sync.MessageId;
+import org.briarproject.bramble.api.transport.KeySetId;
+
+import javax.annotation.concurrent.Immutable;
+import javax.inject.Inject;
+
+import static org.briarproject.bramble.transport.agreement.TransportKeyAgreementConstants.SESSION_KEY_KEY_SET_ID;
+import static org.briarproject.bramble.transport.agreement.TransportKeyAgreementConstants.SESSION_KEY_LAST_LOCAL_MESSAGE_ID;
+import static org.briarproject.bramble.transport.agreement.TransportKeyAgreementConstants.SESSION_KEY_LOCAL_PRIVATE_KEY;
+import static org.briarproject.bramble.transport.agreement.TransportKeyAgreementConstants.SESSION_KEY_LOCAL_PUBLIC_KEY;
+import static org.briarproject.bramble.transport.agreement.TransportKeyAgreementConstants.SESSION_KEY_LOCAL_TIMESTAMP;
+import static org.briarproject.bramble.transport.agreement.TransportKeyAgreementConstants.SESSION_KEY_STATE;
+
+@Immutable
+@NotNullByDefault
+class SessionParserImpl implements SessionParser {
+
+	private final TransportKeyAgreementCrypto crypto;
+
+	@Inject
+	SessionParserImpl(TransportKeyAgreementCrypto crypto) {
+		this.crypto = crypto;
+	}
+
+	@Override
+	public Session parseSession(BdfDictionary meta) throws FormatException {
+		State state =
+				State.fromValue(meta.getLong(SESSION_KEY_STATE).intValue());
+
+		MessageId lastLocalMessageId = null;
+		byte[] lastLocalMessageIdBytes =
+				meta.getOptionalRaw(SESSION_KEY_LAST_LOCAL_MESSAGE_ID);
+		if (lastLocalMessageIdBytes != null) {
+			lastLocalMessageId = new MessageId(lastLocalMessageIdBytes);
+		}
+
+		KeyPair localKeyPair = null;
+		byte[] localPublicKeyBytes =
+				meta.getOptionalRaw(SESSION_KEY_LOCAL_PUBLIC_KEY);
+		byte[] localPrivateKeyBytes =
+				meta.getOptionalRaw(SESSION_KEY_LOCAL_PRIVATE_KEY);
+		if (localPublicKeyBytes != null && localPrivateKeyBytes != null) {
+			PublicKey pub = crypto.parsePublicKey(localPublicKeyBytes);
+			PrivateKey priv = crypto.parsePrivateKey(localPrivateKeyBytes);
+			localKeyPair = new KeyPair(pub, priv);
+		}
+
+		Long localTimestamp = meta.getOptionalLong(SESSION_KEY_LOCAL_TIMESTAMP);
+
+		KeySetId keySetId = null;
+		Long keySetIdLong = meta.getOptionalLong(SESSION_KEY_KEY_SET_ID);
+		if (keySetIdLong != null) {
+			keySetId = new KeySetId(keySetIdLong.intValue());
+		}
+
+		return new Session(state, lastLocalMessageId, localKeyPair,
+				localTimestamp, keySetId);
+	}
+}

bramble-core/src/main/java/org/briarproject/bramble/transport/agreement/State.java

@@ -0,0 +1,43 @@
+package org.briarproject.bramble.transport.agreement;
+
+import org.briarproject.bramble.api.FormatException;
+import org.briarproject.bramble.api.nullsafety.NotNullByDefault;
+
+import javax.annotation.concurrent.Immutable;
+
+@Immutable
+@NotNullByDefault
+enum State {
+
+	/**
+	 * We've sent a key message and are awaiting the contact's key message.
+	 */
+	AWAIT_KEY(0),
+
+	/**
+	 * We've exchanged key messages, derived the transport keys and sent an
+	 * activate message, and now we're awaiting the contact's activate message.
+	 */
+	AWAIT_ACTIVATE(1),
+
+	/**
+	 * We've exchanged key messages and activate messages, and have derived and
+	 * activated the transport keys. This is the end state.
+	 */
+	ACTIVATED(2);
+
+	private final int value;
+
+	State(int value) {
+		this.value = value;
+	}
+
+	int getValue() {
+		return value;
+	}
+
+	static State fromValue(int value) throws FormatException {
+		for (State s : values()) if (s.value == value) return s;
+		throw new FormatException();
+	}
+}

bramble-core/src/main/java/org/briarproject/bramble/transport/agreement/TransportKeyAgreementConstants.java

@@ -0,0 +1,27 @@
+package org.briarproject.bramble.transport.agreement;
+
+import org.briarproject.bramble.api.nullsafety.NotNullByDefault;
+
+@NotNullByDefault
+interface TransportKeyAgreementConstants {
+
+	String MSG_KEY_IS_SESSION = "isSession";
+	String MSG_KEY_MESSAGE_TYPE = "messageType";
+	String MSG_KEY_TRANSPORT_ID = "transportId";
+	String MSG_KEY_PUBLIC_KEY = "publicKey";
+	String MSG_KEY_LOCAL = "local";
+
+	String SESSION_KEY_STATE = "state";
+	String SESSION_KEY_LAST_LOCAL_MESSAGE_ID = "lastLocalMessageId";
+	String SESSION_KEY_LOCAL_PUBLIC_KEY = "localPublicKey";
+	String SESSION_KEY_LOCAL_PRIVATE_KEY = "localPrivateKey";
+	String SESSION_KEY_LOCAL_TIMESTAMP = "localTimestamp";
+	String SESSION_KEY_KEY_SET_ID = "keySetId";
+
+	/**
+	 * Label for deriving the root key from key pairs.
+	 */
+	String ROOT_KEY_LABEL =
+			"org.briarproject.bramble.transport.agreement/ROOT_KEY";
+
+}

bramble-core/src/main/java/org/briarproject/bramble/transport/agreement/TransportKeyAgreementCrypto.java

@@ -0,0 +1,23 @@
+package org.briarproject.bramble.transport.agreement;
+
+import org.briarproject.bramble.api.FormatException;
+import org.briarproject.bramble.api.crypto.KeyPair;
+import org.briarproject.bramble.api.crypto.PrivateKey;
+import org.briarproject.bramble.api.crypto.PublicKey;
+import org.briarproject.bramble.api.crypto.SecretKey;
+import org.briarproject.bramble.api.nullsafety.NotNullByDefault;
+
+import java.security.GeneralSecurityException;
+
+@NotNullByDefault
+interface TransportKeyAgreementCrypto {
+
+	KeyPair generateKeyPair();
+
+	SecretKey deriveRootKey(KeyPair localKeyPair, PublicKey remotePublicKey)
+			throws GeneralSecurityException;
+
+	PublicKey parsePublicKey(byte[] encoded) throws FormatException;
+
+	PrivateKey parsePrivateKey(byte[] encoded) throws FormatException;
+}

bramble-core/src/main/java/org/briarproject/bramble/transport/agreement/TransportKeyAgreementCryptoImpl.java

@@ -0,0 +1,66 @@
+package org.briarproject.bramble.transport.agreement;
+
+import org.briarproject.bramble.api.FormatException;
+import org.briarproject.bramble.api.crypto.CryptoComponent;
+import org.briarproject.bramble.api.crypto.KeyPair;
+import org.briarproject.bramble.api.crypto.PrivateKey;
+import org.briarproject.bramble.api.crypto.PublicKey;
+import org.briarproject.bramble.api.crypto.SecretKey;
+import org.briarproject.bramble.api.nullsafety.NotNullByDefault;
+
+import java.security.GeneralSecurityException;
+
+import javax.annotation.concurrent.Immutable;
+import javax.inject.Inject;
+
+import static org.briarproject.bramble.api.Bytes.compare;
+import static org.briarproject.bramble.transport.agreement.TransportKeyAgreementConstants.ROOT_KEY_LABEL;
+
+@Immutable
+@NotNullByDefault
+class TransportKeyAgreementCryptoImpl implements TransportKeyAgreementCrypto {
+
+	private final CryptoComponent crypto;
+
+	@Inject
+	TransportKeyAgreementCryptoImpl(CryptoComponent crypto) {
+		this.crypto = crypto;
+	}
+
+	@Override
+	public KeyPair generateKeyPair() {
+		return crypto.generateAgreementKeyPair();
+	}
+
+	@Override
+	public SecretKey deriveRootKey(KeyPair localKeyPair,
+			PublicKey remotePublicKey) throws GeneralSecurityException {
+		byte[] theirPublic = remotePublicKey.getEncoded();
+		byte[] ourPublic = localKeyPair.getPublic().getEncoded();
+		boolean alice = compare(ourPublic, theirPublic) < 0;
+		byte[][] inputs = {
+				alice ? ourPublic : theirPublic,
+				alice ? theirPublic : ourPublic
+		};
+		return crypto.deriveSharedSecret(ROOT_KEY_LABEL, remotePublicKey,
+				localKeyPair, inputs);
+	}
+
+	@Override
+	public PublicKey parsePublicKey(byte[] encoded) throws FormatException {
+		try {
+			return crypto.getAgreementKeyParser().parsePublicKey(encoded);
+		} catch (GeneralSecurityException e) {
+			throw new FormatException();
+		}
+	}
+
+	@Override
+	public PrivateKey parsePrivateKey(byte[] encoded) throws FormatException {
+		try {
+			return crypto.getAgreementKeyParser().parsePrivateKey(encoded);
+		} catch (GeneralSecurityException e) {
+			throw new FormatException();
+		}
+	}
+}

bramble-core/src/main/java/org/briarproject/bramble/transport/agreement/TransportKeyAgreementManagerImpl.java

@@ -0,0 +1,408 @@
+package org.briarproject.bramble.transport.agreement;
+
+import org.briarproject.bramble.api.FormatException;
+import org.briarproject.bramble.api.client.BdfIncomingMessageHook;
+import org.briarproject.bramble.api.client.ClientHelper;
+import org.briarproject.bramble.api.client.ContactGroupFactory;
+import org.briarproject.bramble.api.contact.Contact;
+import org.briarproject.bramble.api.contact.ContactId;
+import org.briarproject.bramble.api.contact.ContactManager.ContactHook;
+import org.briarproject.bramble.api.crypto.KeyPair;
+import org.briarproject.bramble.api.crypto.PublicKey;
+import org.briarproject.bramble.api.crypto.SecretKey;
+import org.briarproject.bramble.api.data.BdfDictionary;
+import org.briarproject.bramble.api.data.BdfList;
+import org.briarproject.bramble.api.data.MetadataParser;
+import org.briarproject.bramble.api.db.DatabaseComponent;
+import org.briarproject.bramble.api.db.DbException;
+import org.briarproject.bramble.api.db.Metadata;
+import org.briarproject.bramble.api.db.Transaction;
+import org.briarproject.bramble.api.identity.Author;
+import org.briarproject.bramble.api.identity.IdentityManager;
+import org.briarproject.bramble.api.lifecycle.LifecycleManager.OpenDatabaseHook;
+import org.briarproject.bramble.api.nullsafety.NotNullByDefault;
+import org.briarproject.bramble.api.plugin.PluginConfig;
+import org.briarproject.bramble.api.plugin.PluginFactory;
+import org.briarproject.bramble.api.plugin.TransportId;
+import org.briarproject.bramble.api.sync.Group;
+import org.briarproject.bramble.api.sync.Group.Visibility;
+import org.briarproject.bramble.api.sync.GroupId;
+import org.briarproject.bramble.api.sync.Message;
+import org.briarproject.bramble.api.sync.MessageId;
+import org.briarproject.bramble.api.transport.KeyManager;
+import org.briarproject.bramble.api.transport.KeySetId;
+import org.briarproject.bramble.api.transport.agreement.TransportKeyAgreementManager;
+import org.briarproject.bramble.api.versioning.ClientVersioningManager;
+import org.briarproject.bramble.api.versioning.ClientVersioningManager.ClientVersioningHook;
+
+import java.security.GeneralSecurityException;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.List;
+import java.util.Map;
+import java.util.logging.Logger;
+
+import javax.annotation.Nullable;
+import javax.annotation.concurrent.Immutable;
+import javax.inject.Inject;
+
+import static java.lang.Math.min;
+import static java.util.Collections.singletonMap;
+import static java.util.logging.Level.INFO;
+import static java.util.logging.Logger.getLogger;
+import static org.briarproject.bramble.api.Bytes.compare;
+import static org.briarproject.bramble.api.nullsafety.NullSafety.requireNonNull;
+import static org.briarproject.bramble.api.sync.validation.IncomingMessageHook.DeliveryAction.ACCEPT_DO_NOT_SHARE;
+import static org.briarproject.bramble.api.sync.validation.IncomingMessageHook.DeliveryAction.DEFER;
+import static org.briarproject.bramble.api.sync.validation.IncomingMessageHook.DeliveryAction.REJECT;
+import static org.briarproject.bramble.transport.agreement.MessageType.ACTIVATE;
+import static org.briarproject.bramble.transport.agreement.MessageType.KEY;
+import static org.briarproject.bramble.transport.agreement.State.ACTIVATED;
+import static org.briarproject.bramble.transport.agreement.State.AWAIT_ACTIVATE;
+import static org.briarproject.bramble.transport.agreement.State.AWAIT_KEY;
+import static org.briarproject.bramble.transport.agreement.TransportKeyAgreementConstants.MSG_KEY_MESSAGE_TYPE;
+import static org.briarproject.bramble.transport.agreement.TransportKeyAgreementConstants.MSG_KEY_PUBLIC_KEY;
+import static org.briarproject.bramble.transport.agreement.TransportKeyAgreementConstants.MSG_KEY_TRANSPORT_ID;
+
+@Immutable
+@NotNullByDefault
+class TransportKeyAgreementManagerImpl extends BdfIncomingMessageHook
+		implements TransportKeyAgreementManager, OpenDatabaseHook, ContactHook,
+		ClientVersioningHook {
+
+	private static final Logger LOG =
+			getLogger(TransportKeyAgreementManagerImpl.class.getName());
+
+	private final ContactGroupFactory contactGroupFactory;
+	private final ClientVersioningManager clientVersioningManager;
+	private final IdentityManager identityManager;
+	private final KeyManager keyManager;
+	private final MessageEncoder messageEncoder;
+	private final SessionEncoder sessionEncoder;
+	private final SessionParser sessionParser;
+	private final TransportKeyAgreementCrypto crypto;
+
+	private final List<TransportId> transports;
+	private final Group localGroup;
+
+	@Inject
+	TransportKeyAgreementManagerImpl(
+			DatabaseComponent db,
+			ClientHelper clientHelper,
+			MetadataParser metadataParser,
+			ContactGroupFactory contactGroupFactory,
+			ClientVersioningManager clientVersioningManager,
+			IdentityManager identityManager,
+			KeyManager keyManager,
+			MessageEncoder messageEncoder,
+			SessionEncoder sessionEncoder,
+			SessionParser sessionParser,
+			TransportKeyAgreementCrypto crypto,
+			PluginConfig config) {
+		super(db, clientHelper, metadataParser);
+		this.contactGroupFactory = contactGroupFactory;
+		this.clientVersioningManager = clientVersioningManager;
+		this.identityManager = identityManager;
+		this.keyManager = keyManager;
+		this.messageEncoder = messageEncoder;
+		this.sessionEncoder = sessionEncoder;
+		this.sessionParser = sessionParser;
+		this.crypto = crypto;
+		transports = new ArrayList<>();
+		for (PluginFactory<?> f : config.getDuplexFactories()) {
+			transports.add(f.getId());
+		}
+		for (PluginFactory<?> f : config.getSimplexFactories()) {
+			transports.add(f.getId());
+		}
+		localGroup = contactGroupFactory.createLocalGroup(CLIENT_ID,
+				MAJOR_VERSION);
+	}
+
+	@Override
+	public void onDatabaseOpened(Transaction txn) throws DbException {
+		Collection<Contact> contacts = db.getContacts(txn);
+		if (!db.containsGroup(txn, localGroup.getId())) {
+			db.addGroup(txn, localGroup);
+			// Set things up for any pre-existing contacts
+			for (Contact c : contacts) addingContact(txn, c);
+		}
+		// Find any contacts and transports that need keys
+		Map<ContactId, Collection<TransportId>> transportsWithKeys =
+				db.getTransportsWithKeys(txn);
+		for (Contact c : contacts) {
+			Collection<TransportId> withKeys =
+					transportsWithKeys.get(c.getId());
+			for (TransportId t : transports) {
+				if (withKeys == null || !withKeys.contains(t)) {
+					// We need keys for this contact and transport
+					GroupId contactGroupId = getContactGroup(c).getId();
+					SavedSession ss = loadSession(txn, contactGroupId, t);
+					if (ss == null) {
+						// Start a session by sending our key message
+						startSession(txn, contactGroupId, t);
+					}
+				}
+			}
+		}
+	}
+
+	@Override
+	public void addingContact(Transaction txn, Contact c) throws DbException {
+		// Create a group to share with the contact
+		Group g = getContactGroup(c);
+		db.addGroup(txn, g);
+		// Attach the contact ID to the group
+		clientHelper.setContactId(txn, g.getId(), c.getId());
+		// Apply the client's visibility to the contact group
+		Visibility client = clientVersioningManager.getClientVisibility(txn,
+				c.getId(), CLIENT_ID, MAJOR_VERSION);
+		db.setGroupVisibility(txn, c.getId(), g.getId(), client);
+	}
+
+	@Override
+	public void removingContact(Transaction txn, Contact c) throws DbException {
+		db.removeGroup(txn, getContactGroup(c));
+	}
+
+	@Override
+	public void onClientVisibilityChanging(Transaction txn, Contact c,
+			Visibility v) throws DbException {
+		// Apply the client's visibility to the contact group
+		Group g = getContactGroup(c);
+		db.setGroupVisibility(txn, c.getId(), g.getId(), v);
+	}
+
+	@Override
+	protected DeliveryAction incomingMessage(Transaction txn, Message m,
+			BdfList body, BdfDictionary meta)
+			throws DbException, FormatException {
+		MessageType type = MessageType.fromValue(
+				meta.getLong(MSG_KEY_MESSAGE_TYPE).intValue());
+		TransportId t = new TransportId(meta.getString(MSG_KEY_TRANSPORT_ID));
+		if (LOG.isLoggable(INFO)) {
+			LOG.info("Received " + type + " message for " + t);
+		}
+		if (!transports.contains(t)) {
+			// Defer handling the message until we support the transport
+			return DEFER;
+		}
+		SavedSession ss = loadSession(txn, m.getGroupId(), t);
+		if (type == KEY) return handleKeyMessage(txn, t, m, meta, ss);
+		else if (type == ACTIVATE) return handleActivateMessage(txn, t, ss);
+		else throw new AssertionError();
+	}
+
+	private DeliveryAction handleKeyMessage(Transaction txn, TransportId t,
+			Message m, BdfDictionary meta, @Nullable SavedSession ss)
+			throws DbException, FormatException {
+		ContactId c = clientHelper.getContactId(txn, m.getGroupId());
+		boolean haveKeys = db.containsTransportKeys(txn, c, t);
+		if (ss == null) {
+			if (haveKeys) {
+				// We have keys but no session, so we must have derived keys
+				// when adding the contact. If the contact didn't support
+				// the transport when they added us, they wouldn't have
+				// derived keys at that time. If they later added support for
+				// the transport then they would have started a session, so a
+				// key message is valid in this case
+				return handleKeyMessageForNewSession(txn, c, t, m, meta);
+			} else {
+				// We don't have keys, so we should have created a session at
+				// startup
+				throw new IllegalStateException();
+			}
+		} else if (ss.session.getState() == AWAIT_KEY) {
+			if (haveKeys) {
+				// We have keys, so we shouldn't be in the AWAIT_KEY state,
+				// even if the contact didn't derive keys when adding us and
+				// later started a session
+				throw new IllegalStateException();
+			} else {
+				// This is the key message we're waiting for
+				return handleKeyMessageForExistingSession(txn, c, t, m, meta,
+						ss);
+			}
+		} else {
+			return REJECT; // Not valid in this state
+		}
+	}
+
+	private DeliveryAction handleActivateMessage(Transaction txn,
+			TransportId t, @Nullable SavedSession ss) throws DbException {
+		if (ss != null && ss.session.getState() == AWAIT_ACTIVATE) {
+			// Activate the keys and finish the session
+			KeySetId keySetId = requireNonNull(ss.session.getKeySetId());
+			keyManager.activateKeys(txn, singletonMap(t, keySetId));
+			Session session = new Session(ACTIVATED,
+					ss.session.getLastLocalMessageId(), null, null, null);
+			saveSession(txn, t, ss.storageId, session);
+			return ACCEPT_DO_NOT_SHARE;
+		} else {
+			return REJECT; // Not valid in this state
+		}
+	}
+
+	private DeliveryAction handleKeyMessageForNewSession(Transaction txn,
+			ContactId c, TransportId t, Message m, BdfDictionary meta)
+			throws DbException, FormatException {
+		KeyPair localKeyPair = crypto.generateKeyPair();
+		PublicKey remotePublicKey =
+				crypto.parsePublicKey(meta.getRaw(MSG_KEY_PUBLIC_KEY));
+		Message keyMessage = sendKeyMessage(txn, m.getGroupId(), t,
+				localKeyPair.getPublic());
+		long minTimestamp = min(keyMessage.getTimestamp(), m.getTimestamp());
+		SecretKey rootKey;
+		try {
+			rootKey = crypto.deriveRootKey(localKeyPair, remotePublicKey);
+		} catch (GeneralSecurityException e) {
+			return REJECT; // Invalid public key
+		}
+		boolean alice = isLocalPartyAlice(txn, db.getContact(txn, c));
+		KeySetId keySetId = keyManager.addRotationKeys(txn, c, t, rootKey,
+				minTimestamp, alice, false);
+		Message activateMessage =
+				sendActivateMessage(txn, m.getGroupId(), t, keyMessage.getId());
+		Session session = new Session(AWAIT_ACTIVATE, activateMessage.getId(),
+				null, null, keySetId);
+		saveNewSession(txn, m.getGroupId(), t, session);
+		return ACCEPT_DO_NOT_SHARE;
+	}
+
+	private DeliveryAction handleKeyMessageForExistingSession(Transaction txn,
+			ContactId c, TransportId t, Message m, BdfDictionary meta,
+			SavedSession ss) throws DbException, FormatException {
+		KeyPair localKeyPair = requireNonNull(ss.session.getLocalKeyPair());
+		PublicKey remotePublicKey =
+				crypto.parsePublicKey(meta.getRaw(MSG_KEY_PUBLIC_KEY));
+		long localTimestamp = requireNonNull(ss.session.getLocalTimestamp());
+		long minTimestamp = min(localTimestamp, m.getTimestamp());
+		SecretKey rootKey;
+		try {
+			rootKey = crypto.deriveRootKey(localKeyPair, remotePublicKey);
+		} catch (GeneralSecurityException e) {
+			return REJECT; // Invalid public key
+		}
+		boolean alice = isLocalPartyAlice(txn, db.getContact(txn, c));
+		KeySetId keySetId = keyManager.addRotationKeys(txn, c, t, rootKey,
+				minTimestamp, alice, false);
+		MessageId previousMessageId =
+				requireNonNull(ss.session.getLastLocalMessageId());
+		Message activateMessage =
+				sendActivateMessage(txn, m.getGroupId(), t, previousMessageId);
+		Session session = new Session(AWAIT_ACTIVATE, activateMessage.getId(),
+				null, null, keySetId);
+		saveSession(txn, t, ss.storageId, session);
+		return ACCEPT_DO_NOT_SHARE;
+	}
+
+	private void startSession(Transaction txn, GroupId contactGroupId,
+			TransportId t) throws DbException {
+		KeyPair localKeyPair = crypto.generateKeyPair();
+		Message keyMessage = sendKeyMessage(txn, contactGroupId, t,
+				localKeyPair.getPublic());
+		Session session = new Session(AWAIT_KEY, keyMessage.getId(),
+				localKeyPair, keyMessage.getTimestamp(), null);
+		saveNewSession(txn, contactGroupId, t, session);
+	}
+
+	@Nullable
+	private SavedSession loadSession(Transaction txn, GroupId contactGroupId,
+			TransportId t) throws DbException {
+		try {
+			BdfDictionary query = sessionEncoder.getSessionQuery(t);
+			Collection<MessageId> ids =
+					clientHelper.getMessageIds(txn, contactGroupId, query);
+			if (ids.size() > 1) throw new DbException();
+			if (ids.isEmpty()) {
+				if (LOG.isLoggable(INFO)) LOG.info("No session for " + t);
+				return null;
+			}
+			MessageId storageId = ids.iterator().next();
+			BdfDictionary bdfSession =
+					clientHelper.getMessageMetadataAsDictionary(txn, storageId);
+			Session session = sessionParser.parseSession(bdfSession);
+			if (LOG.isLoggable(INFO)) {
+				LOG.info("Loaded session in state " + session.getState()
+						+ " for " + t);
+			}
+			return new SavedSession(session, storageId);
+		} catch (FormatException e) {
+			throw new DbException(e);
+		}
+	}
+
+	private void saveNewSession(Transaction txn, GroupId contactGroupId,
+			TransportId t, Session session) throws DbException {
+		Message m =
+				clientHelper.createMessageForStoringMetadata(contactGroupId);
+		db.addLocalMessage(txn, m, new Metadata(), false, false);
+		MessageId storageId = m.getId();
+		saveSession(txn, t, storageId, session);
+	}
+
+	private void saveSession(Transaction txn, TransportId t,
+			MessageId storageId, Session session) throws DbException {
+		if (LOG.isLoggable(INFO)) {
+			LOG.info("Saving session in state " + session.getState()
+					+ " for " + t);
+		}
+		BdfDictionary meta = sessionEncoder.encodeSession(session, t);
+		try {
+			clientHelper.mergeMessageMetadata(txn, storageId, meta);
+		} catch (FormatException e) {
+			throw new AssertionError();
+		}
+	}
+
+	private Message sendKeyMessage(Transaction txn, GroupId contactGroupId,
+			TransportId t, PublicKey publicKey) throws DbException {
+		Message m = messageEncoder.encodeKeyMessage(contactGroupId, t,
+				publicKey);
+		sendMessage(txn, t, m, KEY);
+		return m;
+	}
+
+	private Message sendActivateMessage(Transaction txn,
+			GroupId contactGroupId, TransportId t, MessageId previousMessageId)
+			throws DbException {
+		Message m = messageEncoder.encodeActivateMessage(contactGroupId, t,
+				previousMessageId);
+		sendMessage(txn, t, m, ACTIVATE);
+		return m;
+	}
+
+	private void sendMessage(Transaction txn, TransportId t, Message m,
+			MessageType type) throws DbException {
+		BdfDictionary meta =
+				messageEncoder.encodeMessageMetadata(t, type, true);
+		try {
+			clientHelper.addLocalMessage(txn, m, meta, true, false);
+		} catch (FormatException e) {
+			throw new AssertionError();
+		}
+	}
+
+	private Group getContactGroup(Contact c) {
+		return contactGroupFactory.createContactGroup(CLIENT_ID,
+				MAJOR_VERSION, c);
+	}
+
+	private boolean isLocalPartyAlice(Transaction txn, Contact c)
+			throws DbException {
+		Author local = identityManager.getLocalAuthor(txn);
+		Author remote = c.getAuthor();
+		return compare(local.getId().getBytes(), remote.getId().getBytes()) < 0;
+	}
+
+	private static class SavedSession {
+
+		private final Session session;
+		private final MessageId storageId;
+
+		private SavedSession(Session session, MessageId storageId) {
+			this.session = session;
+			this.storageId = storageId;
+		}
+	}
+}

bramble-core/src/main/java/org/briarproject/bramble/transport/agreement/TransportKeyAgreementModule.java

@@ -0,0 +1,83 @@
+package org.briarproject.bramble.transport.agreement;
+
+import org.briarproject.bramble.api.client.ClientHelper;
+import org.briarproject.bramble.api.contact.ContactManager;
+import org.briarproject.bramble.api.data.MetadataEncoder;
+import org.briarproject.bramble.api.lifecycle.LifecycleManager;
+import org.briarproject.bramble.api.sync.validation.ValidationManager;
+import org.briarproject.bramble.api.system.Clock;
+import org.briarproject.bramble.api.transport.agreement.TransportKeyAgreementManager;
+import org.briarproject.bramble.api.versioning.ClientVersioningManager;
+
+import javax.inject.Inject;
+import javax.inject.Singleton;
+
+import dagger.Module;
+import dagger.Provides;
+
+import static org.briarproject.bramble.api.transport.agreement.TransportKeyAgreementManager.CLIENT_ID;
+import static org.briarproject.bramble.api.transport.agreement.TransportKeyAgreementManager.MAJOR_VERSION;
+import static org.briarproject.bramble.api.transport.agreement.TransportKeyAgreementManager.MINOR_VERSION;
+
+@Module
+public class TransportKeyAgreementModule {
+
+	public static class EagerSingletons {
+		@Inject
+		TransportKeyAgreementManager transportKeyAgreementManager;
+		@Inject
+		TransportKeyAgreementValidator transportKeyAgreementValidator;
+	}
+
+	@Provides
+	@Singleton
+	TransportKeyAgreementManager provideTransportKeyAgreementManager(
+			LifecycleManager lifecycleManager,
+			ValidationManager validationManager,
+			ContactManager contactManager,
+			ClientVersioningManager clientVersioningManager,
+			TransportKeyAgreementManagerImpl transportKeyAgreementManager) {
+		lifecycleManager.registerOpenDatabaseHook(transportKeyAgreementManager);
+		validationManager.registerIncomingMessageHook(CLIENT_ID,
+				MAJOR_VERSION, transportKeyAgreementManager);
+		contactManager.registerContactHook(transportKeyAgreementManager);
+		clientVersioningManager.registerClient(CLIENT_ID, MAJOR_VERSION,
+				MINOR_VERSION, transportKeyAgreementManager);
+		return transportKeyAgreementManager;
+	}
+
+	@Provides
+	@Singleton
+	TransportKeyAgreementValidator provideTransportKeyAgreementValidator(
+			ClientHelper clientHelper, MetadataEncoder metadataEncoder,
+			Clock clock, MessageEncoder messageEncoder,
+			ValidationManager validationManager) {
+		TransportKeyAgreementValidator validator =
+				new TransportKeyAgreementValidator(clientHelper,
+						metadataEncoder, clock, messageEncoder);
+		validationManager.registerMessageValidator(CLIENT_ID, MAJOR_VERSION,
+				validator);
+		return validator;
+	}
+
+	@Provides
+	MessageEncoder provideMessageEncoder(MessageEncoderImpl messageEncoder) {
+		return messageEncoder;
+	}
+
+	@Provides
+	SessionEncoder provideSessionEncoder(SessionEncoderImpl sessionEncoder) {
+		return sessionEncoder;
+	}
+
+	@Provides
+	SessionParser provideSessionParser(SessionParserImpl sessionParser) {
+		return sessionParser;
+	}
+
+	@Provides
+	TransportKeyAgreementCrypto provideTransportKeyAgreementCrypto(
+			TransportKeyAgreementCryptoImpl transportKeyAgreementCrypto) {
+		return transportKeyAgreementCrypto;
+	}
+}

bramble-core/src/main/java/org/briarproject/bramble/transport/agreement/TransportKeyAgreementValidator.java

@@ -0,0 +1,79 @@
+package org.briarproject.bramble.transport.agreement;
+
+import org.briarproject.bramble.api.FormatException;
+import org.briarproject.bramble.api.client.BdfMessageContext;
+import org.briarproject.bramble.api.client.BdfMessageValidator;
+import org.briarproject.bramble.api.client.ClientHelper;
+import org.briarproject.bramble.api.data.BdfDictionary;
+import org.briarproject.bramble.api.data.BdfList;
+import org.briarproject.bramble.api.data.MetadataEncoder;
+import org.briarproject.bramble.api.nullsafety.NotNullByDefault;
+import org.briarproject.bramble.api.plugin.TransportId;
+import org.briarproject.bramble.api.sync.Group;
+import org.briarproject.bramble.api.sync.Message;
+import org.briarproject.bramble.api.sync.MessageId;
+import org.briarproject.bramble.api.system.Clock;
+
+import javax.annotation.concurrent.Immutable;
+
+import static java.util.Collections.singletonList;
+import static org.briarproject.bramble.api.crypto.CryptoConstants.MAX_AGREEMENT_PUBLIC_KEY_BYTES;
+import static org.briarproject.bramble.api.plugin.TransportId.MAX_TRANSPORT_ID_LENGTH;
+import static org.briarproject.bramble.api.system.Clock.MIN_REASONABLE_TIME_MS;
+import static org.briarproject.bramble.transport.agreement.MessageType.ACTIVATE;
+import static org.briarproject.bramble.transport.agreement.MessageType.KEY;
+import static org.briarproject.bramble.transport.agreement.TransportKeyAgreementConstants.MSG_KEY_PUBLIC_KEY;
+import static org.briarproject.bramble.util.ValidationUtils.checkLength;
+import static org.briarproject.bramble.util.ValidationUtils.checkSize;
+
+@Immutable
+@NotNullByDefault
+class TransportKeyAgreementValidator extends BdfMessageValidator {
+
+	private final MessageEncoder messageEncoder;
+
+	TransportKeyAgreementValidator(ClientHelper clientHelper,
+			MetadataEncoder metadataEncoder, Clock clock,
+			MessageEncoder messageEncoder) {
+		super(clientHelper, metadataEncoder, clock);
+		this.messageEncoder = messageEncoder;
+	}
+
+	@Override
+	protected BdfMessageContext validateMessage(Message m, Group g,
+			BdfList body) throws FormatException {
+		MessageType type = MessageType.fromValue(body.getLong(0).intValue());
+		if (type == KEY) return validateKeyMessage(m.getTimestamp(), body);
+		else if (type == ACTIVATE) return validateActivateMessage(body);
+		else throw new AssertionError();
+	}
+
+	private BdfMessageContext validateKeyMessage(long timestamp, BdfList body)
+			throws FormatException {
+		if (timestamp < MIN_REASONABLE_TIME_MS) throw new FormatException();
+		// Message type, transport ID, public key
+		checkSize(body, 3);
+		String transportId = body.getString(1);
+		checkLength(transportId, 1, MAX_TRANSPORT_ID_LENGTH);
+		byte[] publicKey = body.getRaw(2);
+		checkLength(publicKey, 1, MAX_AGREEMENT_PUBLIC_KEY_BYTES);
+		BdfDictionary meta = messageEncoder.encodeMessageMetadata(
+				new TransportId(transportId), KEY, false);
+		meta.put(MSG_KEY_PUBLIC_KEY, publicKey);
+		return new BdfMessageContext(meta);
+	}
+
+	private BdfMessageContext validateActivateMessage(BdfList body)
+			throws FormatException {
+		// Message type, transport ID, previous message ID
+		checkSize(body, 3);
+		String transportId = body.getString(1);
+		checkLength(transportId, 1, MAX_TRANSPORT_ID_LENGTH);
+		byte[] previousMessageId = body.getRaw(2);
+		checkLength(previousMessageId, MessageId.LENGTH);
+		BdfDictionary meta = messageEncoder.encodeMessageMetadata(
+				new TransportId(transportId), ACTIVATE, false);
+		MessageId dependency = new MessageId(previousMessageId);
+		return new BdfMessageContext(meta, singletonList(dependency));
+	}
+}

bramble-core/src/main/java/org/briarproject/bramble/versioning/ClientVersioningManagerImpl.java

@@ -50,6 +50,7 @@ import static java.util.Collections.emptyList;
 import static org.briarproject.bramble.api.sync.Group.Visibility.INVISIBLE;
 import static org.briarproject.bramble.api.sync.Group.Visibility.SHARED;
 import static org.briarproject.bramble.api.sync.Group.Visibility.VISIBLE;
+import static org.briarproject.bramble.api.sync.validation.IncomingMessageHook.DeliveryAction.ACCEPT_DO_NOT_SHARE;
 import static org.briarproject.bramble.versioning.ClientVersioningConstants.MSG_KEY_LOCAL;
 import static org.briarproject.bramble.versioning.ClientVersioningConstants.MSG_KEY_UPDATE_VERSION;
 
@@ -173,8 +174,8 @@ class ClientVersioningManagerImpl implements ClientVersioningManager,
 	}
 
 	@Override
-	public boolean incomingMessage(Transaction txn, Message m, Metadata meta)
-			throws DbException, InvalidMessageException {
+	public DeliveryAction incomingMessage(Transaction txn, Message m,
+			Metadata meta) throws DbException, InvalidMessageException {
 		try {
 			// Parse the new remote update
 			Update newRemoteUpdate = parseUpdate(clientHelper.toList(m));
@@ -187,7 +188,7 @@ class ClientVersioningManagerImpl implements ClientVersioningManager,
 					&& latest.remote.updateVersion > newRemoteUpdateVersion) {
 				db.deleteMessage(txn, m.getId());
 				db.deleteMessageMetadata(txn, m.getId());
-				return false;
+				return ACCEPT_DO_NOT_SHARE;
 			}
 			// Load and parse the latest local update
 			if (latest.local == null) throw new DbException();
@@ -241,7 +242,7 @@ class ClientVersioningManagerImpl implements ClientVersioningManager,
 		} catch (FormatException e) {
 			throw new InvalidMessageException(e);
 		}
-		return false;
+		return ACCEPT_DO_NOT_SHARE;
 	}
 
 	private void storeClientVersions(Transaction txn,