Associate a timestamp with every subscription, indicating the earliest

acceptable timestamp of subscribed messages. For a new subscription,
the timestamp is initialised to the current time, so a new subscriber
to a group will not immediately receive any messages. (Subscribing to
a group is therefore more like joining a mailing list than joining a
Usenet group - you only receive messages written after you joined.)

Once the database fills up and starts expiring messages, the
timestamps of subscriptions are updated so that contacts need not send
messages that would expire immediately. This is done using the
*approximate* timestamp of the oldest message in the database, to
avoid revealing the presence or absence of any particular message.

components/net/sf/briar/db/Database.java

@@ -123,20 +123,30 @@ interface Database<T> {
 	boolean containsMessage(T txn, MessageId m) throws DbException;
 
 	/**
-	 * Returns true if the user is subscribed to the given group.
+	 * Returns true if the user subscribes to the given group.
 	 * <p>
 	 * Locking: subscriptions read.
 	 */
 	boolean containsSubscription(T txn, GroupId g) throws DbException;
 
 	/**
-	 * Returns true if the user is subscribed to the given group and the
-	 * group is visible to the given contact.
+	 * Returns true if the user has been subscribed to the given group since
+	 * the given time.
+	 * <p>
+	 * Locking: subscriptions read.
+	 */
+	boolean containsSubscription(T txn, GroupId g, long time)
+	throws DbException;
+
+	/**
+	 * Returns true if the user is subscribed to the given group, the group is
+	 * visible to the given contact, and the subscription has existed since the
+	 * given time.
 	 * <p>
 	 * Locking: contacts read, subscriptions read.
 	 */
-	boolean containsVisibleSubscription(T txn, GroupId g, ContactId c)
-	throws DbException;
+	boolean containsVisibleSubscription(T txn, GroupId g, ContactId c,
+			long time) throws DbException;
 
 	/**
 	 * Returns the IDs of any batches received from the given contact that need
@@ -301,7 +311,7 @@ interface Database<T> {
 	 * Returns the groups to which the user subscribes that are visible to the
 	 * given contact.
 	 */
-	Collection<Group> getVisibleSubscriptions(T txn, ContactId c)
+	Map<Group, Long> getVisibleSubscriptions(T txn, ContactId c)
 	throws DbException;
 
 	/**
@@ -402,7 +412,7 @@ interface Database<T> {
 	 * <p>
 	 * Locking: contacts write, subscriptions write.
 	 */
-	void setSubscriptions(T txn, ContactId c, Collection<Group> subs,
+	void setSubscriptions(T txn, ContactId c, Map<Group, Long> subs,
 			long timestamp) throws DbException;
 
 	/**

components/net/sf/briar/db/JdbcDatabase.java

@@ -10,6 +10,7 @@ import java.sql.SQLException;
 import java.sql.Statement;
 import java.util.ArrayList;
 import java.util.Collection;
+import java.util.HashMap;
 import java.util.LinkedList;
 import java.util.Map;
 import java.util.Map.Entry;
@@ -19,6 +20,7 @@ import java.util.logging.Logger;
 
 import net.sf.briar.api.ContactId;
 import net.sf.briar.api.Rating;
+import net.sf.briar.api.db.DatabaseComponent;
 import net.sf.briar.api.db.DbException;
 import net.sf.briar.api.db.Status;
 import net.sf.briar.api.protocol.AuthorId;
@@ -41,6 +43,7 @@ abstract class JdbcDatabase implements Database<Connection> {
 		+ " (groupId HASH NOT NULL,"
 		+ " groupName VARCHAR NOT NULL,"
 		+ " groupKey BINARY,"
+		+ " start TIMESTAMP NOT NULL,"
 		+ " PRIMARY KEY (groupId))";
 
 	private static final String CREATE_MESSAGES =
@@ -103,6 +106,7 @@ abstract class JdbcDatabase implements Database<Connection> {
 		+ " groupId HASH NOT NULL,"
 		+ " groupName VARCHAR NOT NULL,"
 		+ " groupKey BINARY,"
+		+ " start TIMESTAMP NOT NULL,"
 		+ " PRIMARY KEY (contactId, groupId),"
 		+ " FOREIGN KEY (contactId) REFERENCES contacts (contactId)"
 		+ " ON DELETE CASCADE)";
@@ -564,12 +568,13 @@ abstract class JdbcDatabase implements Database<Connection> {
 		PreparedStatement ps = null;
 		try {
 			String sql = "INSERT INTO subscriptions"
-				+ " (groupId, groupName, groupKey)"
-				+ " VALUES (?, ?, ?)";
+				+ " (groupId, groupName, groupKey, start)"
+				+ " VALUES (?, ?, ?, ?)";
 			ps = txn.prepareStatement(sql);
 			ps.setBytes(1, g.getId().getBytes());
 			ps.setString(2, g.getName());
 			ps.setBytes(3, g.getPublicKey());
+			ps.setLong(4, System.currentTimeMillis());
 			int affected = ps.executeUpdate();
 			if(affected != 1) throw new DbStateException();
 			ps.close();
@@ -651,26 +656,52 @@ abstract class JdbcDatabase implements Database<Connection> {
 		}
 	}
 
-	public boolean containsVisibleSubscription(Connection txn, GroupId g,
-			ContactId c) throws DbException {
+	public boolean containsSubscription(Connection txn, GroupId g, long time)
+	throws DbException {
+		boolean found = false;
 		PreparedStatement ps = null;
 		ResultSet rs = null;
 		try {
-			String sql = "SELECT COUNT(subscriptions.groupId)"
-				+ " FROM subscriptions JOIN visibilities"
+			String sql = "SELECT start FROM subscriptions WHERE groupId = ?";
+			ps = txn.prepareStatement(sql);
+			ps.setBytes(1, g.getBytes());
+			rs = ps.executeQuery();
+			if(rs.next()) {
+				long start = rs.getLong(1);
+				if(start <= time) found = true;
+				if(rs.next()) throw new DbStateException();
+			}
+			rs.close();
+			ps.close();
+			return found;
+		} catch(SQLException e) {
+			tryToClose(rs);
+			tryToClose(ps);
+			throw new DbException(e);
+		}
+	}
+
+	public boolean containsVisibleSubscription(Connection txn, GroupId g,
+			ContactId c, long time) throws DbException {
+		boolean found = false;
+		PreparedStatement ps = null;
+		ResultSet rs = null;
+		try {
+			String sql = "SELECT start FROM subscriptions JOIN visibilities"
 				+ " ON subscriptions.groupId = visibilities.groupId"
 				+ " WHERE subscriptions.groupId = ? AND contactId = ?";
 			ps = txn.prepareStatement(sql);
 			ps.setBytes(1, g.getBytes());
 			ps.setInt(2, c.getInt());
 			rs = ps.executeQuery();
-			if(!rs.next()) throw new DbStateException();
-			int count = rs.getInt(1);
-			if(count > 1) throw new DbStateException();
-			if(rs.next()) throw new DbStateException();
+			if(rs.next()) {
+				long start = rs.getLong(1);
+				if(start <= time) found = true;
+				if(rs.next()) throw new DbStateException();
+			}
 			rs.close();
 			ps.close();
-			return count > 0;
+			return found;
 		} catch(SQLException e) {
 			tryToClose(rs);
 			tryToClose(ps);
@@ -810,6 +841,7 @@ abstract class JdbcDatabase implements Database<Connection> {
 				+ " AND contactSubscriptions.contactId = ?"
 				+ " AND visibilities.contactId = ?"
 				+ " AND statuses.contactId = ?"
+				+ " AND timestamp >= start"
 				+ " AND status = ? AND sendability > ZERO()";
 			ps = txn.prepareStatement(sql);
 			ps.setBytes(1, m.getBytes());
@@ -1019,6 +1051,7 @@ abstract class JdbcDatabase implements Database<Connection> {
 				+ " WHERE contactSubscriptions.contactId = ?"
 				+ " AND visibilities.contactId = ?"
 				+ " AND statuses.contactId = ?"
+				+ " AND timestamp >= start"
 				+ " AND status = ? AND sendability > ZERO()";
 			// FIXME: Investigate the performance impact of "ORDER BY timestamp"
 			ps = txn.prepareStatement(sql);
@@ -1213,24 +1246,28 @@ abstract class JdbcDatabase implements Database<Connection> {
 		}
 	}
 
-	public Collection<Group> getVisibleSubscriptions(Connection txn,
+	public Map<Group, Long> getVisibleSubscriptions(Connection txn,
 			ContactId c) throws DbException {
+		long expiry = getApproximateExpiryTime(txn);
 		PreparedStatement ps = null;
 		ResultSet rs = null;
 		try {
-			String sql = "SELECT subscriptions.groupId, groupName, groupKey"
+			String sql =
+				"SELECT subscriptions.groupId, groupName, groupKey, start"
 				+ " FROM subscriptions JOIN visibilities"
 				+ " ON subscriptions.groupId = visibilities.groupId"
 				+ " WHERE contactId = ?";
 			ps = txn.prepareStatement(sql);
 			ps.setInt(1, c.getInt());
 			rs = ps.executeQuery();
-			Collection<Group> subs = new ArrayList<Group>();
+			Map<Group, Long> subs = new HashMap<Group, Long>();
 			while(rs.next()) {
 				GroupId id = new GroupId(rs.getBytes(1));
 				String name = rs.getString(2);
 				byte[] publicKey = rs.getBytes(3);
-				subs.add(groupFactory.createGroup(id, name, publicKey));
+				Group g = groupFactory.createGroup(id, name, publicKey);
+				long start = Math.max(rs.getLong(4), expiry);
+				subs.put(g, start);
 			}
 			rs.close();
 			ps.close();
@@ -1242,6 +1279,31 @@ abstract class JdbcDatabase implements Database<Connection> {
 		}
 	}
 
+	private long getApproximateExpiryTime(Connection txn) throws DbException {
+		PreparedStatement ps = null;
+		ResultSet rs = null;
+		try {
+			long timestamp = 0L;
+			String sql = "SELECT timestamp FROM messages"
+				+ " ORDER BY timestamp LIMIT ?";
+			ps = txn.prepareStatement(sql);
+			ps.setInt(1, 1);
+			rs = ps.executeQuery();
+			if(rs.next()) {
+				timestamp = rs.getLong(1);
+				timestamp -= timestamp % DatabaseComponent.EXPIRY_MODULUS;
+			}
+			if(rs.next()) throw new DbStateException();
+			rs.close();
+			ps.close();
+			return timestamp;
+		} catch(SQLException e) {
+			tryToClose(rs);
+			tryToClose(ps);
+			throw new DbException(e);
+		}
+	}
+
 	public boolean hasSendableMessages(Connection txn, ContactId c)
 	throws DbException {
 		PreparedStatement ps = null;
@@ -1256,6 +1318,7 @@ abstract class JdbcDatabase implements Database<Connection> {
 				+ " WHERE contactSubscriptions.contactId = ?"
 				+ " AND visibilities.contactId = ?"
 				+ " AND statuses.contactId = ?"
+				+ " AND timestamp >= start"
 				+ " AND status = ? AND sendability > ZERO()"
 				+ " LIMIT ?";
 			ps = txn.prepareStatement(sql);
@@ -1583,7 +1646,7 @@ abstract class JdbcDatabase implements Database<Connection> {
 	}
 
 	public void setSubscriptions(Connection txn, ContactId c,
-			Collection<Group> subs, long timestamp) throws DbException {
+			Map<Group, Long> subs, long timestamp) throws DbException {
 		PreparedStatement ps = null;
 		ResultSet rs = null;
 		try {
@@ -1607,14 +1670,16 @@ abstract class JdbcDatabase implements Database<Connection> {
 			ps.close();
 			// Store the new subscriptions
 			sql = "INSERT INTO contactSubscriptions"
-				+ " (contactId, groupId, groupName, groupKey)"
-				+ " VALUES (?, ?, ?, ?)";
+				+ " (contactId, groupId, groupName, groupKey, start)"
+				+ " VALUES (?, ?, ?, ?, ?)";
 			ps = txn.prepareStatement(sql);
 			ps.setInt(1, c.getInt());
-			for(Group g : subs) {
+			for(Entry<Group, Long> e : subs.entrySet()) {
+				Group g = e.getKey();
 				ps.setBytes(2, g.getId().getBytes());
 				ps.setString(3, g.getName());
 				ps.setBytes(4, g.getPublicKey());
+				ps.setLong(5, e.getValue());
 				ps.addBatch();
 			}
 			int[] batchAffected = ps.executeBatch();
@@ -1656,7 +1721,7 @@ abstract class JdbcDatabase implements Database<Connection> {
 			ps.close();
 			// Store the new details
 			sql = "INSERT INTO " + table + " (transportName, key, value)"
-				+ " VALUES (?, ?, ?)";
+			+ " VALUES (?, ?, ?)";
 			ps = txn.prepareStatement(sql);
 			ps.setString(1, name);
 			for(Entry<String, String> e : details.entrySet()) {

components/net/sf/briar/db/ReadWriteLockDatabaseComponent.java

@@ -170,16 +170,11 @@ class ReadWriteLockDatabaseComponent<Txn> extends DatabaseComponentImpl<Txn> {
 						Txn txn = db.startTransaction();
 						try {
 							// Don't store the message if the user has
-							// unsubscribed from the group
-							if(db.containsSubscription(txn, m.getGroup())) {
+							// unsubscribed from the group or the message
+							// predates the subscription
+							if(db.containsSubscription(txn, m.getGroup(),
+									m.getTimestamp())) {
 								added = storeMessage(txn, m, null);
-								if(!added) {
-									if(LOG.isLoggable(Level.FINE))
-										LOG.fine("Duplicate local message");
-								}
-							} else {
-								if(LOG.isLoggable(Level.FINE))
-									LOG.fine("Not subscribed");
 							}
 							db.commitTransaction(txn);
 						} catch(DbException e) {
@@ -473,7 +468,7 @@ class ReadWriteLockDatabaseComponent<Txn> extends DatabaseComponentImpl<Txn> {
 			try {
 				Txn txn = db.startTransaction();
 				try {
-					Collection<Group> subs = db.getVisibleSubscriptions(txn, c);
+					Map<Group, Long> subs = db.getVisibleSubscriptions(txn, c);
 					s.writeSubscriptions(subs);
 					if(LOG.isLoggable(Level.FINE))
 						LOG.fine("Added " + subs.size() + " subscriptions");
@@ -740,8 +735,8 @@ class ReadWriteLockDatabaseComponent<Txn> extends DatabaseComponentImpl<Txn> {
 							int received = 0, stored = 0;
 							for(Message m : b.getMessages()) {
 								received++;
-								GroupId g = m.getGroup();
-								if(db.containsVisibleSubscription(txn, g, c)) {
+								if(db.containsVisibleSubscription(txn,
+										m.getGroup(), c, m.getTimestamp())) {
 									if(storeMessage(txn, m, c)) {
 										anyAdded = true;
 										stored++;
@@ -826,7 +821,7 @@ class ReadWriteLockDatabaseComponent<Txn> extends DatabaseComponentImpl<Txn> {
 			try {
 				Txn txn = db.startTransaction();
 				try {
-					Collection<Group> subs = s.getSubscriptions();
+					Map<Group, Long> subs = s.getSubscriptions();
 					db.setSubscriptions(txn, c, subs, s.getTimestamp());
 					if(LOG.isLoggable(Level.FINE))
 						LOG.fine("Received " + subs.size() + " subscriptions");
@@ -1013,7 +1008,7 @@ class ReadWriteLockDatabaseComponent<Txn> extends DatabaseComponentImpl<Txn> {
 		try {
 			Txn txn = db.startTransaction();
 			try {
-				if(!db.containsSubscription(txn, g.getId())) {
+				if(db.containsSubscription(txn, g.getId())) {
 					db.addSubscription(txn, g);
 					added = true;
 				}

components/net/sf/briar/db/SynchronizedDatabaseComponent.java

@@ -126,8 +126,10 @@ class SynchronizedDatabaseComponent<Txn> extends DatabaseComponentImpl<Txn> {
 						Txn txn = db.startTransaction();
 						try {
 							// Don't store the message if the user has
-							// unsubscribed from the group
-							if(db.containsSubscription(txn, m.getGroup())) {
+							// unsubscribed from the group or the message
+							// predates the subscription
+							if(db.containsSubscription(txn, m.getGroup(),
+									m.getTimestamp())) {
 								added = storeMessage(txn, m, null);
 								if(!added) {
 									if(LOG.isLoggable(Level.FINE))
@@ -343,7 +345,7 @@ class SynchronizedDatabaseComponent<Txn> extends DatabaseComponentImpl<Txn> {
 			synchronized(subscriptionLock) {
 				Txn txn = db.startTransaction();
 				try {
-					Collection<Group> subs = db.getVisibleSubscriptions(txn, c);
+					Map<Group, Long> subs = db.getVisibleSubscriptions(txn, c);
 					s.writeSubscriptions(subs);
 					if(LOG.isLoggable(Level.FINE))
 						LOG.fine("Added " + subs.size() + " subscriptions");
@@ -549,7 +551,8 @@ class SynchronizedDatabaseComponent<Txn> extends DatabaseComponentImpl<Txn> {
 							for(Message m : b.getMessages()) {
 								received++;
 								GroupId g = m.getGroup();
-								if(db.containsVisibleSubscription(txn, g, c)) {
+								if(db.containsVisibleSubscription(txn, g, c,
+										m.getTimestamp())) {
 									if(storeMessage(txn, m, c)) {
 										anyAdded = true;
 										stored++;
@@ -612,7 +615,7 @@ class SynchronizedDatabaseComponent<Txn> extends DatabaseComponentImpl<Txn> {
 			synchronized(subscriptionLock) {
 				Txn txn = db.startTransaction();
 				try {
-					Collection<Group> subs = s.getSubscriptions();
+					Map<Group, Long> subs = s.getSubscriptions();
 					db.setSubscriptions(txn, c, subs, s.getTimestamp());
 					if(LOG.isLoggable(Level.FINE))
 						LOG.fine("Received " + subs.size() + " subscriptions");
@@ -758,7 +761,7 @@ class SynchronizedDatabaseComponent<Txn> extends DatabaseComponentImpl<Txn> {
 		synchronized(subscriptionLock) {
 			Txn txn = db.startTransaction();
 			try {
-				if(!db.containsSubscription(txn, g.getId())) {
+				if(db.containsSubscription(txn, g.getId())) {
 					db.addSubscription(txn, g);
 					added = true;
 				}

components/net/sf/briar/protocol/SubscriptionFactory.java

@@ -1,11 +1,12 @@
 package net.sf.briar.protocol;
 
-import java.util.Collection;
+import java.util.Map;
 
 import net.sf.briar.api.protocol.Group;
 import net.sf.briar.api.protocol.SubscriptionUpdate;
 
 interface SubscriptionFactory {
 
-	SubscriptionUpdate createSubscriptions(Collection<Group> subs, long timestamp);
+	SubscriptionUpdate createSubscriptions(Map<Group, Long> subs,
+			long timestamp);
 }

components/net/sf/briar/protocol/SubscriptionFactoryImpl.java

@@ -1,13 +1,13 @@
 package net.sf.briar.protocol;
 
-import java.util.Collection;
+import java.util.Map;
 
 import net.sf.briar.api.protocol.Group;
 import net.sf.briar.api.protocol.SubscriptionUpdate;
 
 class SubscriptionFactoryImpl implements SubscriptionFactory {
 
-	public SubscriptionUpdate createSubscriptions(Collection<Group> subs,
+	public SubscriptionUpdate createSubscriptions(Map<Group, Long> subs,
 			long timestamp) {
 		return new SubscriptionsImpl(subs, timestamp);
 	}

components/net/sf/briar/protocol/SubscriptionReader.java

@@ -1,7 +1,7 @@
 package net.sf.briar.protocol;
 
 import java.io.IOException;
-import java.util.Collection;
+import java.util.Map;
 
 import net.sf.briar.api.protocol.Group;
 import net.sf.briar.api.protocol.SubscriptionUpdate;
@@ -31,7 +31,7 @@ class SubscriptionReader implements ObjectReader<SubscriptionUpdate> {
 		r.addConsumer(counting);
 		r.readUserDefinedTag(Tags.SUBSCRIPTIONS);
 		r.addObjectReader(Tags.GROUP, groupReader);
-		Collection<Group> subs = r.readList(Group.class);
+		Map<Group, Long> subs = r.readMap(Group.class, Long.class);
 		r.removeObjectReader(Tags.GROUP);
 		long timestamp = r.readInt64();
 		r.removeConsumer(counting);

components/net/sf/briar/protocol/SubscriptionsImpl.java

@@ -1,21 +1,21 @@
 package net.sf.briar.protocol;
 
-import java.util.Collection;
+import java.util.Map;
 
 import net.sf.briar.api.protocol.Group;
 import net.sf.briar.api.protocol.SubscriptionUpdate;
 
 class SubscriptionsImpl implements SubscriptionUpdate {
 
-	private final Collection<Group> subs;
+	private final Map<Group, Long> subs;
 	private final long timestamp;
 
-	SubscriptionsImpl(Collection<Group> subs, long timestamp) {
+	SubscriptionsImpl(Map<Group, Long> subs, long timestamp) {
 		this.subs = subs;
 		this.timestamp = timestamp;
 	}
 
-	public Collection<Group> getSubscriptions() {
+	public Map<Group, Long> getSubscriptions() {
 		return subs;
 	}
 

components/net/sf/briar/protocol/writers/SubscriptionWriterImpl.java

@@ -2,7 +2,7 @@ package net.sf.briar.protocol.writers;
 
 import java.io.IOException;
 import java.io.OutputStream;
-import java.util.Collection;
+import java.util.Map;
 
 import net.sf.briar.api.protocol.Group;
 import net.sf.briar.api.protocol.Tags;
@@ -20,9 +20,9 @@ class SubscriptionWriterImpl implements SubscriptionWriter {
 		w = writerFactory.createWriter(out);
 	}
 
-	public void writeSubscriptions(Collection<Group> subs) throws IOException {
+	public void writeSubscriptions(Map<Group, Long> subs) throws IOException {
 		w.writeUserDefinedTag(Tags.SUBSCRIPTIONS);
-		w.writeList(subs);
+		w.writeMap(subs);
 		w.writeInt64(System.currentTimeMillis());
 		out.flush();
 	}