Merge branch '2045-flexible-sync' into '1802-sync-via-removable-storage'

Make retransmissions in the sync protocol more flexible

See merge request briar/briar!1482

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/db/Database.java

@@ -357,16 +357,6 @@ interface Database<T> {
 	 */
 	Message getMessage(T txn, MessageId m) throws DbException;
 
-	/**
-	 * Returns the total length, including headers, of any messages that are
-	 * eligible to be sent to the given contact via a transport with the given
-	 * max latency.
-	 * <p/>
-	 * Read-only.
-	 */
-	long getMessageBytesToSend(T txn, ContactId c, int maxLatency)
-			throws DbException;
-
 	/**
 	 * Returns the IDs and states of all dependencies of the given message.
 	 * For missing dependencies and dependencies in other groups, the state
@@ -496,11 +486,37 @@ 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;
 
+	/**
+	 * 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, int)} 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.
 	 * <p/>

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

@@ -436,6 +436,32 @@ class DatabaseComponentImpl<T> implements DatabaseComponent {
 		return messages;
 	}
 
+	@Override
+	public Collection<Message> generateBatch(Transaction transaction,
+			ContactId c, Collection<MessageId> ids, int 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,
@@ -582,15 +608,6 @@ class DatabaseComponentImpl<T> implements DatabaseComponent {
 		return db.getMessage(txn, m);
 	}
 
-	@Override
-	public long getMessageBytesToSend(Transaction transaction, ContactId c,
-			int maxLatency) throws DbException {
-		T txn = unbox(transaction);
-		if (!db.containsContact(txn, c))
-			throw new NoSuchContactException();
-		return db.getMessageBytesToSend(txn, c, maxLatency);
-	}
-
 	@Override
 	public Collection<MessageId> getMessageIds(Transaction transaction,
 			GroupId g) throws DbException {
@@ -714,6 +731,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 {

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;
@@ -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)";
@@ -570,6 +576,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) {
@@ -1781,37 +1788,6 @@ abstract class JdbcDatabase implements Database<Connection> {
 		}
 	}
 
-	@Override
-	public long getMessageBytesToSend(Connection txn, ContactId c,
-			int maxLatency) throws DbException {
-		long now = clock.currentTimeMillis();
-		long eta = now + maxLatency;
-		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"
-					+ " 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();
-			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> getMessageIds(Connection txn, GroupId g)
 			throws DbException {
@@ -2259,6 +2235,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 {

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

@@ -92,6 +92,11 @@ abstract class AbstractRemovableDrivePlugin implements SimplexPlugin {
 		throw new UnsupportedOperationException();
 	}
 
+	@Override
+	public boolean isLossyAndCheap() {
+		return true;
+	}
+
 	@Override
 	public TransportConnectionReader createReader(TransportProperties p) {
 		try {

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

@@ -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/RemovableDriveWriterTask.java

@@ -60,10 +60,9 @@ class RemovableDriveWriterTask extends RemovableDriveTaskImpl
 			setSuccess(false);
 			return;
 		}
-		int maxLatency = plugin.getMaxLatency();
 		try {
 			setTotal(db.transactionWithResult(true, txn ->
-					db.getMessageBytesToSend(txn, contactId, maxLatency)));
+					db.getUnackedMessageBytesToSend(txn, contactId)));
 		} catch (DbException e) {
 			logException(LOG, WARNING, e);
 			registry.removeWriter(this);
@@ -106,6 +105,11 @@ class RemovableDriveWriterTask extends RemovableDriveTaskImpl
 			return delegate.getMaxIdleTime();
 		}
 
+		@Override
+		public boolean isLossyAndCheap() {
+			return delegate.isLossyAndCheap();
+		}
+
 		@Override
 		public OutputStream getOutputStream() throws IOException {
 			return delegate.getOutputStream();

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

@@ -1,8 +1,8 @@
 package org.briarproject.bramble.plugin.file;
 
 import org.briarproject.bramble.api.nullsafety.NotNullByDefault;
-import org.briarproject.bramble.api.plugin.Plugin;
 import org.briarproject.bramble.api.plugin.TransportConnectionWriter;
+import org.briarproject.bramble.api.plugin.simplex.SimplexPlugin;
 
 import java.io.OutputStream;
 import java.util.logging.Logger;
@@ -17,10 +17,10 @@ class TransportOutputStreamWriter implements TransportConnectionWriter {
 	private static final Logger LOG =
 			getLogger(TransportOutputStreamWriter.class.getName());
 
-	private final Plugin plugin;
+	private final SimplexPlugin plugin;
 	private final OutputStream out;
 
-	TransportOutputStreamWriter(Plugin plugin, OutputStream out) {
+	TransportOutputStreamWriter(SimplexPlugin plugin, OutputStream out) {
 		this.plugin = plugin;
 		this.out = out;
 	}
@@ -35,6 +35,11 @@ class TransportOutputStreamWriter 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/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;
@@ -61,6 +66,7 @@ class SimplexOutgoingSession implements SyncSession, EventListener {
 	private final ContactId contactId;
 	private final TransportId transportId;
 	private final int 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,
+			int 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,12 +60,12 @@ class SyncSessionFactoryImpl implements SyncSessionFactory {
 
 	@Override
 	public SyncSession createSimplexOutgoingSession(ContactId c, TransportId t,
-			int maxLatency, StreamWriter streamWriter) {
+			int 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