Add Message Dependencies to Database

This adds a new table to the database to hold message dependencies. It introduces two more message states: pending and delivered The valid column in the database was renamed to state to better reflect its new extended meaning. The DatabaseComponent was extended with three methods for: * adding dependencies * getting dependencies of a message * getting messages that depend on a message (dependents) * getting messages to be delivered (by startup hook) * getting pending messages to be possibly delivered (by startup hook) In order to reflect the new states, things that were previously true for VALID messages have been changed to now be true for DELIVERED messages. Since pending messages should not be available to clients, many database queries have been modified to only return results for delivered messages. All added methods and changes should come with updated unit tests. Please note that the database version was bumped in this commit.
2026-02-14 19:59:05 +01:00 · 2016-05-18 18:24:03 -03:00
parent 512940e82b
commit b03d0a206b
18 changed files with 924 additions and 185 deletions
--- a/briar-core/src/org/briarproject/db/Database.java
+++ b/briar-core/src/org/briarproject/db/Database.java
@@ -16,7 +16,7 @@ import org.briarproject.api.sync.GroupId;
 import org.briarproject.api.sync.Message;
 import org.briarproject.api.sync.MessageId;
 import org.briarproject.api.sync.MessageStatus;
-import org.briarproject.api.sync.ValidationManager.Validity;
+import org.briarproject.api.sync.ValidationManager.State;
 import org.briarproject.api.transport.TransportKeys;

 import java.util.Collection;
@@ -79,7 +79,7 @@ interface Database<T> {
 	/**
 	 * Stores a message.
 	 */
-	void addMessage(T txn, Message m, Validity validity, boolean shared)
+	void addMessage(T txn, Message m, State state, boolean shared)
 			throws DbException;

 	/**
@@ -326,6 +326,24 @@ interface Database<T> {
 	MessageStatus getMessageStatus(T txn, ContactId c, MessageId m)
 			throws DbException;

+	/**
+	 * Returns the dependencies of the given message.
+	 * This method makes sure that dependencies in different groups
+	 * are returned as {@link ValidationManager.State.INVALID}.
+	 * <p/>
+	 * Read-only.
+	 */
+	Map<MessageId, State> getMessageDependencies(T txn, MessageId m)
+			throws DbException;
+
+	/**
+	 * Returns all IDs of messages that depend on the given message.
+	 * <p/>
+	 * Read-only.
+	 */
+	Map<MessageId, State> getMessageDependents(T txn, MessageId m)
+			throws DbException;
+
 	/**
 	 * Returns the IDs of some messages received from the given contact that
 	 * need to be acknowledged, up to the given number of messages.
@@ -371,6 +389,24 @@ interface Database<T> {
 	Collection<MessageId> getMessagesToValidate(T txn, ClientId c)
 			throws DbException;

+	/**
+	 * Returns the IDs of any messages that need to be delivered to the given
+	 * client.
+	 * <p/>
+	 * Read-only.
+	 */
+	Collection<MessageId> getMessagesToDeliver(T txn, ClientId c)
+			throws DbException;
+
+	/**
+	 * Returns the IDs of any messages that are still pending due to
+	 * dependencies to other messages for the given client.
+	 * <p/>
+	 * Read-only.
+	 */
+	Collection<MessageId> getPendingMessages(T txn, ClientId c)
+			throws DbException;
+
 	/**
 	 * Returns the message with the given ID, in serialised form, or null if
 	 * the message has been deleted.
@@ -538,7 +574,14 @@ interface Database<T> {
 	/**
 	 * Marks the given message as valid or invalid.
 	 */
-	void setMessageValid(T txn, MessageId m, boolean valid) throws DbException;
+	void setMessageState(T txn, MessageId m, State state)
+			throws DbException;
+
+	/*
+	 * Adds a dependency between two MessageIds
+	 */
+	void addMessageDependency(T txn, MessageId dependentId,
+			MessageId dependencyId) throws DbException;

 	/**
 	 * Sets the reordering window for the given contact and transport in the
--- a/briar-core/src/org/briarproject/db/DatabaseComponentImpl.java
+++ b/briar-core/src/org/briarproject/db/DatabaseComponentImpl.java
@@ -29,8 +29,8 @@ import org.briarproject.api.event.MessageRequestedEvent;
 import org.briarproject.api.event.MessageSharedEvent;
 import org.briarproject.api.event.MessageToAckEvent;
 import org.briarproject.api.event.MessageToRequestEvent;
-import org.briarproject.api.event.MessageValidatedEvent;
 import org.briarproject.api.event.MessagesAckedEvent;
+import org.briarproject.api.event.MessageStateChangedEvent;
 import org.briarproject.api.event.MessagesSentEvent;
 import org.briarproject.api.event.SettingsUpdatedEvent;
 import org.briarproject.api.identity.Author;
@@ -47,7 +47,7 @@ import org.briarproject.api.sync.MessageId;
 import org.briarproject.api.sync.MessageStatus;
 import org.briarproject.api.sync.Offer;
 import org.briarproject.api.sync.Request;
-import org.briarproject.api.sync.ValidationManager.Validity;
+import org.briarproject.api.sync.ValidationManager.State;
 import org.briarproject.api.transport.TransportKeys;

 import java.util.ArrayList;
@@ -64,8 +64,8 @@ import java.util.logging.Logger;
 import javax.inject.Inject;

 import static java.util.logging.Level.WARNING;
-import static org.briarproject.api.sync.ValidationManager.Validity.UNKNOWN;
-import static org.briarproject.api.sync.ValidationManager.Validity.VALID;
+import static org.briarproject.api.sync.ValidationManager.State.DELIVERED;
+import static org.briarproject.api.sync.ValidationManager.State.UNKNOWN;
 import static org.briarproject.db.DatabaseConstants.MAX_OFFERED_MESSAGES;

 class DatabaseComponentImpl<T> implements DatabaseComponent {
@@ -193,17 +193,18 @@ class DatabaseComponentImpl<T> implements DatabaseComponent {
 		if (!db.containsGroup(txn, m.getGroupId()))
 			throw new NoSuchGroupException();
 		if (!db.containsMessage(txn, m.getId())) {
-			addMessage(txn, m, VALID, shared);
+			addMessage(txn, m, DELIVERED, shared);
 			transaction.attach(new MessageAddedEvent(m, null));
-			transaction.attach(new MessageValidatedEvent(m, c, true, true));
+			transaction.attach(new MessageStateChangedEvent(m, c, true,
+					DELIVERED));
 			if (shared) transaction.attach(new MessageSharedEvent(m));
 		}
 		db.mergeMessageMetadata(txn, m.getId(), meta);
 	}

-	private void addMessage(T txn, Message m, Validity validity, boolean shared)
+	private void addMessage(T txn, Message m, State state, boolean shared)
 			throws DbException {
-		db.addMessage(txn, m, validity, shared);
+		db.addMessage(txn, m, state, shared);
 		for (ContactId c : db.getVisibility(txn, m.getGroupId())) {
 			boolean offered = db.removeOfferedMessage(txn, c, m.getId());
 			db.addStatus(txn, c, m.getId(), offered, offered);
@@ -411,6 +412,18 @@ class DatabaseComponentImpl<T> implements DatabaseComponent {
 		return db.getMessagesToValidate(txn, c);
 	}

+	public Collection<MessageId> getMessagesToDeliver(Transaction transaction,
+			ClientId c) throws DbException {
+		T txn = unbox(transaction);
+		return db.getMessagesToDeliver(txn, c);
+	}
+
+	public Collection<MessageId> getPendingMessages(Transaction transaction,
+			ClientId c) throws DbException {
+		T txn = unbox(transaction);
+		return db.getPendingMessages(txn, c);
+	}
+
 	public byte[] getRawMessage(Transaction transaction, MessageId m)
 			throws DbException {
 		T txn = unbox(transaction);
@@ -463,6 +476,22 @@ class DatabaseComponentImpl<T> implements DatabaseComponent {
 		return db.getMessageStatus(txn, c, m);
 	}

+	public Map<MessageId, State> getMessageDependencies(Transaction transaction,
+			MessageId m) throws DbException {
+		T txn = unbox(transaction);
+		if (!db.containsMessage(txn, m))
+			throw new NoSuchMessageException();
+		return db.getMessageDependencies(txn, m);
+	}
+
+	public Map<MessageId, State> getMessageDependents(Transaction transaction,
+			MessageId m) throws DbException {
+		T txn = unbox(transaction);
+		if (!db.containsMessage(txn, m))
+			throw new NoSuchMessageException();
+		return db.getMessageDependents(txn, m);
+	}
+
 	public Settings getSettings(Transaction transaction, String namespace)
 			throws DbException {
 		T txn = unbox(transaction);
@@ -664,14 +693,26 @@ class DatabaseComponentImpl<T> implements DatabaseComponent {
 		if (shared) transaction.attach(new MessageSharedEvent(m));
 	}

-	public void setMessageValid(Transaction transaction, Message m, ClientId c,
-			boolean valid) throws DbException {
+	public void setMessageState(Transaction transaction, Message m, ClientId c,
+			State state) throws DbException {
 		if (transaction.isReadOnly()) throw new IllegalArgumentException();
 		T txn = unbox(transaction);
 		if (!db.containsMessage(txn, m.getId()))
 			throw new NoSuchMessageException();
-		db.setMessageValid(txn, m.getId(), valid);
-		transaction.attach(new MessageValidatedEvent(m, c, false, valid));
+		db.setMessageState(txn, m.getId(), state);
+		transaction.attach(new MessageStateChangedEvent(m, c, false, state));
+	}
+
+	public void addMessageDependencies(Transaction transaction,
+			Message dependent, Collection<MessageId> dependencies)
+			throws DbException {
+		if (transaction.isReadOnly()) throw new IllegalArgumentException();
+		T txn = unbox(transaction);
+		if (!db.containsMessage(txn, dependent.getId()))
+			throw new NoSuchMessageException();
+		for (MessageId dependencyId : dependencies) {
+			db.addMessageDependency(txn, dependent.getId(), dependencyId);
+		}
 	}

 	public void setReorderingWindow(Transaction transaction, ContactId c,
--- a/briar-core/src/org/briarproject/db/JdbcDatabase.java
+++ b/briar-core/src/org/briarproject/db/JdbcDatabase.java
@@ -19,7 +19,7 @@ import org.briarproject.api.sync.GroupId;
 import org.briarproject.api.sync.Message;
 import org.briarproject.api.sync.MessageId;
 import org.briarproject.api.sync.MessageStatus;
-import org.briarproject.api.sync.ValidationManager.Validity;
+import org.briarproject.api.sync.ValidationManager.State;
 import org.briarproject.api.system.Clock;
 import org.briarproject.api.transport.IncomingKeys;
 import org.briarproject.api.transport.OutgoingKeys;
@@ -49,9 +49,11 @@ import java.util.logging.Logger;

 import static java.util.logging.Level.WARNING;
 import static org.briarproject.api.db.Metadata.REMOVE;
-import static org.briarproject.api.sync.ValidationManager.Validity.INVALID;
-import static org.briarproject.api.sync.ValidationManager.Validity.UNKNOWN;
-import static org.briarproject.api.sync.ValidationManager.Validity.VALID;
+import static org.briarproject.api.sync.ValidationManager.State.DELIVERED;
+import static org.briarproject.api.sync.ValidationManager.State.INVALID;
+import static org.briarproject.api.sync.ValidationManager.State.PENDING;
+import static org.briarproject.api.sync.ValidationManager.State.UNKNOWN;
+import static org.briarproject.api.sync.ValidationManager.State.VALID;
 import static org.briarproject.db.DatabaseConstants.DB_SETTINGS_NAMESPACE;
 import static org.briarproject.db.DatabaseConstants.DEVICE_ID_KEY;
 import static org.briarproject.db.DatabaseConstants.DEVICE_SETTINGS_NAMESPACE;
@@ -65,8 +67,8 @@ import static org.briarproject.db.ExponentialBackoff.calculateExpiry;
 */
 abstract class JdbcDatabase implements Database<Connection> {

-	private static final int SCHEMA_VERSION = 24;
-	private static final int MIN_SCHEMA_VERSION = 24;
+	private static final int SCHEMA_VERSION = 25;
+	private static final int MIN_SCHEMA_VERSION = 25;

 	private static final String CREATE_SETTINGS =
 			"CREATE TABLE settings"
@@ -131,7 +133,7 @@ abstract class JdbcDatabase implements Database<Connection> {
 					+ " (messageId HASH NOT NULL,"
 					+ " groupId HASH NOT NULL,"
 					+ " timestamp BIGINT NOT NULL,"
-					+ " valid INT NOT NULL,"
+					+ " state INT NOT NULL,"
 					+ " shared BOOLEAN NOT NULL,"
 					+ " length INT NOT NULL,"
 					+ " raw BLOB," // Null if message has been deleted
@@ -150,6 +152,14 @@ abstract class JdbcDatabase implements Database<Connection> {
 					+ " REFERENCES messages (messageId)"
 					+ " ON DELETE CASCADE)";

+	private static final String CREATE_MESSAGE_DEPENDENCIES =
+			"CREATE TABLE messageDependencies"
+					+ " (messageId HASH NOT NULL,"
+					+ " dependencyId HASH NOT NULL," // Not a foreign key
+					+ " FOREIGN KEY (messageId)"
+					+ " REFERENCES messages (messageId)"
+					+ " ON DELETE CASCADE)";
+
 	private static final String CREATE_OFFERS =
 			"CREATE TABLE offers"
 					+ " (messageId HASH NOT NULL," // Not a foreign key
@@ -320,6 +330,7 @@ abstract class JdbcDatabase implements Database<Connection> {
 			s.executeUpdate(insertTypeNames(CREATE_GROUP_VISIBILITIES));
 			s.executeUpdate(insertTypeNames(CREATE_MESSAGES));
 			s.executeUpdate(insertTypeNames(CREATE_MESSAGE_METADATA));
+			s.executeUpdate(insertTypeNames(CREATE_MESSAGE_DEPENDENCIES));
 			s.executeUpdate(insertTypeNames(CREATE_OFFERS));
 			s.executeUpdate(insertTypeNames(CREATE_STATUSES));
 			s.executeUpdate(insertTypeNames(CREATE_TRANSPORTS));
@@ -520,18 +531,18 @@ abstract class JdbcDatabase implements Database<Connection> {
 		}
 	}

-	public void addMessage(Connection txn, Message m, Validity validity,
+	public void addMessage(Connection txn, Message m, State state,
 			boolean shared) throws DbException {
 		PreparedStatement ps = null;
 		try {
 			String sql = "INSERT INTO messages (messageId, groupId, timestamp,"
-					+ " valid, shared, length, raw)"
+					+ " state, shared, length, raw)"
 					+ " VALUES (?, ?, ?, ?, ?, ?, ?)";
 			ps = txn.prepareStatement(sql);
 			ps.setBytes(1, m.getId().getBytes());
 			ps.setBytes(2, m.getGroupId().getBytes());
 			ps.setLong(3, m.getTimestamp());
-			ps.setInt(4, validity.getValue());
+			ps.setInt(4, state.getValue());
 			ps.setBoolean(5, shared);
 			byte[] raw = m.getRaw();
 			ps.setInt(6, raw.length);
@@ -596,6 +607,25 @@ abstract class JdbcDatabase implements Database<Connection> {
 		}
 	}

+	public void addMessageDependency(Connection txn, MessageId dependentId,
+			MessageId dependencyId) throws DbException {
+		PreparedStatement ps = null;
+		try {
+			String sql =
+					"INSERT INTO messageDependencies (messageId, dependencyId)"
+							+ " VALUES (?, ?)";
+			ps = txn.prepareStatement(sql);
+			ps.setBytes(1, dependentId.getBytes());
+			ps.setBytes(2, dependencyId.getBytes());
+			int affected = ps.executeUpdate();
+			if (affected != 1) throw new DbStateException();
+			ps.close();
+		} catch (SQLException e) {
+			tryToClose(ps);
+			throw new DbException(e);
+		}
+	}
+
 	public void addTransport(Connection txn, TransportId t, int maxLatency)
 			throws DbException {
 		PreparedStatement ps = null;
@@ -1135,10 +1165,33 @@ abstract class JdbcDatabase implements Database<Connection> {
 		}
 	}

+	private Collection<MessageId> getMessageIds(Connection txn, GroupId g,
+			State state) throws DbException {
+		PreparedStatement ps = null;
+		ResultSet rs = null;
+		try {
+			String sql = "SELECT messageId FROM messages"
+					+ " WHERE state = ? AND groupId = ?";
+			ps = txn.prepareStatement(sql);
+			ps.setInt(1, state.getValue());
+			ps.setBytes(2, g.getBytes());
+			rs = ps.executeQuery();
+			List<MessageId> ids = new ArrayList<MessageId>();
+			while (rs.next()) ids.add(new MessageId(rs.getBytes(1)));
+			rs.close();
+			ps.close();
+			return Collections.unmodifiableList(ids);
+		} catch (SQLException e) {
+			tryToClose(rs);
+			tryToClose(ps);
+			throw new DbException(e);
+		}
+	}
+
 	public Collection<MessageId> getMessageIds(Connection txn, GroupId g,
 			Metadata query) throws DbException {
-		// If there are no query terms, return all messages
-		if (query.isEmpty()) return getMessageIds(txn, g);
+		// If there are no query terms, return all delivered messages
+		if (query.isEmpty()) return getMessageIds(txn, g, DELIVERED);
 		PreparedStatement ps = null;
 		ResultSet rs = null;
 		try {
@@ -1148,12 +1201,14 @@ abstract class JdbcDatabase implements Database<Connection> {
 					+ " FROM messages AS m"
 					+ " JOIN messageMetadata AS md"
 					+ " ON m.messageId = md.messageId"
-					+ " WHERE groupId = ? AND key = ? AND value = ?";
+					+ " WHERE state = ? AND groupId = ?"
+					+ " AND key = ? AND value = ?";
 			for (Entry<String, byte[]> e : query.entrySet()) {
 				ps = txn.prepareStatement(sql);
-				ps.setBytes(1, g.getBytes());
-				ps.setString(2, e.getKey());
-				ps.setBytes(3, e.getValue());
+				ps.setInt(1, DELIVERED.getValue());
+				ps.setBytes(2, g.getBytes());
+				ps.setString(3, e.getKey());
+				ps.setBytes(4, e.getValue());
 				rs = ps.executeQuery();
 				Set<MessageId> ids = new HashSet<MessageId>();
 				while (rs.next()) ids.add(new MessageId(rs.getBytes(1)));
@@ -1182,10 +1237,11 @@ abstract class JdbcDatabase implements Database<Connection> {
 					+ " FROM messages AS m"
 					+ " JOIN messageMetadata AS md"
 					+ " ON m.messageId = md.messageId"
-					+ " WHERE groupId = ?"
+					+ " WHERE state = ? AND groupId = ?"
 					+ " ORDER BY m.messageId";
 			ps = txn.prepareStatement(sql);
-			ps.setBytes(1, g.getBytes());
+			ps.setInt(1, DELIVERED.getValue());
+			ps.setBytes(2, g.getBytes());
 			rs = ps.executeQuery();
 			Map<MessageId, Metadata> all = new HashMap<MessageId, Metadata>();
 			Metadata metadata = null;
@@ -1223,23 +1279,38 @@ abstract class JdbcDatabase implements Database<Connection> {

 	public Metadata getGroupMetadata(Connection txn, GroupId g)
 			throws DbException {
-		return getMetadata(txn, g.getBytes(), "groupMetadata", "groupId");
+		PreparedStatement ps = null;
+		ResultSet rs = null;
+		try {
+			String sql = "SELECT key, value FROM groupMetadata"
+					+ " WHERE groupId = ?";
+			ps = txn.prepareStatement(sql);
+			ps.setBytes(1, g.getBytes());
+			rs = ps.executeQuery();
+			Metadata metadata = new Metadata();
+			while (rs.next()) metadata.put(rs.getString(1), rs.getBytes(2));
+			rs.close();
+			ps.close();
+			return metadata;
+		} catch (SQLException e) {
+			tryToClose(rs);
+			tryToClose(ps);
+			throw new DbException(e);
+		}
 	}

 	public Metadata getMessageMetadata(Connection txn, MessageId m)
 			throws DbException {
-		return getMetadata(txn, m.getBytes(), "messageMetadata", "messageId");
-	}
-
-	private Metadata getMetadata(Connection txn, byte[] id, String tableName,
-			String columnName) throws DbException {
 		PreparedStatement ps = null;
 		ResultSet rs = null;
 		try {
-			String sql = "SELECT key, value FROM " + tableName
-					+ " WHERE " + columnName + " = ?";
+			String sql = "SELECT key, value FROM messageMetadata AS md"
+					+ " JOIN messages AS m"
+					+ " ON m.messageId = md.messageId"
+					+ " WHERE m.state = ? AND md.messageId = ?";
 			ps = txn.prepareStatement(sql);
-			ps.setBytes(1, id);
+			ps.setInt(1, DELIVERED.getValue());
+			ps.setBytes(2, m.getBytes());
 			rs = ps.executeQuery();
 			Metadata metadata = new Metadata();
 			while (rs.next()) metadata.put(rs.getString(1), rs.getBytes(2));
@@ -1312,6 +1383,91 @@ abstract class JdbcDatabase implements Database<Connection> {
 		}
 	}

+	public Map<MessageId, State> getMessageDependencies(Connection txn,
+			MessageId m) throws DbException {
+		PreparedStatement ps = null;
+		ResultSet rs = null;
+		try {
+			String sql = "SELECT d.dependencyId, m.state, m.groupId"
+					+ " FROM messageDependencies AS d"
+					+ " LEFT OUTER JOIN messages AS m"
+					+ " ON d.dependencyId = m.messageId"
+					+ " WHERE d.messageId = ?";
+			ps = txn.prepareStatement(sql);
+			ps.setBytes(1, m.getBytes());
+			rs = ps.executeQuery();
+			Map<MessageId, State> dependencies = new HashMap<MessageId, State>();
+			while (rs.next()) {
+				MessageId messageId = new MessageId(rs.getBytes(1));
+				State state = State.fromValue(rs.getInt(2));
+				if (state != UNKNOWN) {
+					// set dependency invalid if it is in a different group
+					if (!hasGroupId(txn, m, rs.getBytes(3))) state = INVALID;
+				}
+				dependencies.put(messageId, state);
+			}
+			rs.close();
+			ps.close();
+			return Collections.unmodifiableMap(dependencies);
+		} catch (SQLException e) {
+			tryToClose(rs);
+			tryToClose(ps);
+			throw new DbException(e);
+		}
+	}
+
+	private boolean hasGroupId(Connection txn, MessageId m, byte[] g)
+			throws DbException {
+		PreparedStatement ps = null;
+		ResultSet rs = null;
+		try {
+			String sql = "SELECT NULL FROM messages"
+					+ " WHERE messageId = ? AND groupId = ?";
+			ps = txn.prepareStatement(sql);
+			ps.setBytes(1, m.getBytes());
+			ps.setBytes(2, g);
+			rs = ps.executeQuery();
+			boolean same = rs.next();
+			if (rs.next()) throw new DbStateException();
+			rs.close();
+			ps.close();
+			return same;
+		} catch (SQLException e) {
+			tryToClose(rs);
+			tryToClose(ps);
+			throw new DbException(e);
+		}
+	}
+
+	public Map<MessageId, State> getMessageDependents(Connection txn,
+			MessageId m) throws DbException {
+		PreparedStatement ps = null;
+		ResultSet rs = null;
+		try {
+			String sql = "SELECT d.messageId, m.state"
+					+ " FROM messageDependencies AS d"
+					+ " LEFT OUTER JOIN messages AS m"
+					+ " ON d.messageId = m.messageId"
+					+ " WHERE dependencyId = ?";
+			ps = txn.prepareStatement(sql);
+			ps.setBytes(1, m.getBytes());
+			rs = ps.executeQuery();
+			Map<MessageId, State> dependents = new HashMap<MessageId, State>();
+			while (rs.next()) {
+				MessageId messageId = new MessageId(rs.getBytes(1));
+				State state = State.fromValue(rs.getInt(2));
+				dependents.put(messageId, state);
+			}
+			rs.close();
+			ps.close();
+			return Collections.unmodifiableMap(dependents);
+		} catch (SQLException e) {
+			tryToClose(rs);
+			tryToClose(ps);
+			throw new DbException(e);
+		}
+	}
+
 	public Collection<MessageId> getMessagesToAck(Connection txn, ContactId c,
 			int maxMessages) throws DbException {
 		PreparedStatement ps = null;
@@ -1346,13 +1502,13 @@ abstract class JdbcDatabase implements Database<Connection> {
 					+ " JOIN statuses AS s"
 					+ " ON m.messageId = s.messageId"
 					+ " WHERE contactId = ?"
-					+ " AND valid = ? AND shared = TRUE AND raw IS NOT NULL"
+					+ " AND state = ? AND shared = TRUE AND raw IS NOT NULL"
 					+ " AND seen = FALSE AND requested = FALSE"
 					+ " AND expiry < ?"
 					+ " ORDER BY timestamp DESC LIMIT ?";
 			ps = txn.prepareStatement(sql);
 			ps.setInt(1, c.getInt());
-			ps.setInt(2, VALID.getValue());
+			ps.setInt(2, DELIVERED.getValue());
 			ps.setLong(3, now);
 			ps.setInt(4, maxMessages);
 			rs = ps.executeQuery();
@@ -1402,13 +1558,13 @@ abstract class JdbcDatabase implements Database<Connection> {
 					+ " JOIN statuses AS s"
 					+ " ON m.messageId = s.messageId"
 					+ " WHERE contactId = ?"
-					+ " AND valid = ? AND shared = TRUE AND raw IS NOT NULL"
+					+ " AND state = ? AND shared = TRUE AND raw IS NOT NULL"
 					+ " AND seen = FALSE"
 					+ " AND expiry < ?"
 					+ " ORDER BY timestamp DESC";
 			ps = txn.prepareStatement(sql);
 			ps.setInt(1, c.getInt());
-			ps.setInt(2, VALID.getValue());
+			ps.setInt(2, DELIVERED.getValue());
 			ps.setLong(3, now);
 			rs = ps.executeQuery();
 			List<MessageId> ids = new ArrayList<MessageId>();
@@ -1431,14 +1587,29 @@ abstract class JdbcDatabase implements Database<Connection> {

 	public Collection<MessageId> getMessagesToValidate(Connection txn,
 			ClientId c) throws DbException {
+		return getMessagesInState(txn, c, UNKNOWN);
+	}
+
+	public Collection<MessageId> getMessagesToDeliver(Connection txn,
+			ClientId c) throws DbException {
+		return getMessagesInState(txn, c, VALID);
+	}
+
+	public Collection<MessageId> getPendingMessages(Connection txn,
+			ClientId c) throws DbException {
+		return getMessagesInState(txn, c, PENDING);
+	}
+
+	private Collection<MessageId> getMessagesInState(Connection txn, ClientId c,
+			State state) throws DbException {
 		PreparedStatement ps = null;
 		ResultSet rs = null;
 		try {
 			String sql = "SELECT messageId FROM messages AS m"
 					+ " JOIN groups AS g ON m.groupId = g.groupId"
-					+ " WHERE valid = ? AND clientId = ? AND raw IS NOT NULL";
+					+ " WHERE state = ? AND clientId = ? AND raw IS NOT NULL";
 			ps = txn.prepareStatement(sql);
-			ps.setInt(1, UNKNOWN.getValue());
+			ps.setInt(1, state.getValue());
 			ps.setBytes(2, c.getBytes());
 			rs = ps.executeQuery();
 			List<MessageId> ids = new ArrayList<MessageId>();
@@ -1485,13 +1656,13 @@ abstract class JdbcDatabase implements Database<Connection> {
 					+ " JOIN statuses AS s"
 					+ " ON m.messageId = s.messageId"
 					+ " WHERE contactId = ?"
-					+ " AND valid = ? AND shared = TRUE AND raw IS NOT NULL"
+					+ " AND state = ? AND shared = TRUE AND raw IS NOT NULL"
 					+ " AND seen = FALSE AND requested = TRUE"
 					+ " AND expiry < ?"
 					+ " ORDER BY timestamp DESC";
 			ps = txn.prepareStatement(sql);
 			ps.setInt(1, c.getInt());
-			ps.setInt(2, VALID.getValue());
+			ps.setInt(2, DELIVERED.getValue());
 			ps.setLong(3, now);
 			rs = ps.executeQuery();
 			List<MessageId> ids = new ArrayList<MessageId>();
@@ -2081,13 +2252,13 @@ abstract class JdbcDatabase implements Database<Connection> {
 		}
 	}

-	public void setMessageValid(Connection txn, MessageId m, boolean valid)
+	public void setMessageState(Connection txn, MessageId m, State state)
 			throws DbException {
 		PreparedStatement ps = null;
 		try {
-			String sql = "UPDATE messages SET valid = ? WHERE messageId = ?";
+			String sql = "UPDATE messages SET state = ? WHERE messageId = ?";
 			ps = txn.prepareStatement(sql);
-			ps.setInt(1, valid ? VALID.getValue() : INVALID.getValue());
+			ps.setInt(1, state.getValue());
 			ps.setBytes(2, m.getBytes());
 			int affected = ps.executeUpdate();
 			if (affected < 0 || affected > 1) throw new DbStateException();
--- a/briar-core/src/org/briarproject/sync/ValidationManagerImpl.java
+++ b/briar-core/src/org/briarproject/sync/ValidationManagerImpl.java
@@ -36,6 +36,8 @@ import javax.inject.Inject;
 import static java.util.logging.Level.INFO;
 import static java.util.logging.Level.WARNING;
 import static org.briarproject.api.sync.SyncConstants.MESSAGE_HEADER_LENGTH;
+import static org.briarproject.api.sync.ValidationManager.State.INVALID;
+import static org.briarproject.api.sync.ValidationManager.State.VALID;

 class ValidationManagerImpl implements ValidationManager, Service,
 		EventListener {
@@ -178,7 +180,7 @@ class ValidationManagerImpl implements ValidationManager, Service,
 					try {
 						Metadata meta = result.getMetadata();
 						db.mergeMessageMetadata(txn, m.getId(), meta);
-						db.setMessageValid(txn, m, c, true);
+						db.setMessageState(txn, m.getId(), c, VALID);
 						db.setMessageShared(txn, m, true);
 						IncomingMessageHook hook = hooks.get(c);
 						if (hook != null)
@@ -202,7 +204,7 @@ class ValidationManagerImpl implements ValidationManager, Service,
 				try {
 					Transaction txn = db.startTransaction(false);
 					try {
-						db.setMessageValid(txn, m, c, false);
+						db.setMessageState(txn, m.getId(), c, INVALID);
 						txn.setComplete();
 					} finally {
 						db.endTransaction(txn);