Merge lp:~thisfred/u1db/validate_transaction_id_of_target into lp:u1db

[Eric Casteleijn]

Added a validation function that checks that the generation + txid the sync target knows about us is correct.

Added a validation function that checks that the generation + txid the sync target knows about us is correct.

[lp:~thisfred/u1db/validate_transaction_id_of_target]

338. By Eric Casteleijn on 2012-06-19

merged trunk resolved conflicts

339. By Eric Casteleijn on 2012-06-19

merged txid branch

Unmerged revisions

=== modified file 'include/u1db/u1db.h'
--- include/u1db/u1db.h	2012-06-18 18:24:19 +0000
+++ include/u1db/u1db.h	2012-06-18 18:24:19 +0000
@@ -78,6 +78,10 @@
 #define U1DB_CONVERGED 3
 #define U1DB_CONFLICTED 4
 
+// Used by validate_source_gen_and_trans_id
+#define U1DB_INVALID_TRANSACTION_ID -20
+#define U1DB_INVALID_GENERATION -21
+
 /**
  * The basic constructor for a new connection.
  */

=== modified file 'include/u1db/u1db_internal.h'
--- include/u1db/u1db_internal.h	2012-06-18 18:24:19 +0000
+++ include/u1db/u1db_internal.h	2012-06-18 18:24:19 +0000
@@ -22,6 +22,7 @@
 #include <stdarg.h>
 #include "u1db/u1db.h"
 #include "u1db/compat.h"
+#include "u1db/u1db_vectorclock.h"
 
 typedef struct sqlite3 sqlite3;
 typedef struct sqlite3_stmt sqlite3_stmt;
@@ -208,11 +209,41 @@
                            int *state, int *at_gen);
 
 /**
+ * Validate source generation and transaction id.
+ *
+ * @param replica_uid uid of source replica at the time of the
+ *     change we are syncing.
+ * @param replica_gen Generation of the replica at the time of the
+ *     change we are syncing.
+ * @param replica_trans_id Transaction id of the replica at the time of the
+ *     change we are syncing.
+ * @param cur_vcr Vector clock for the document in the database.
+ * @param other_vcr Vector clock of the document being put.
+ * @param state (OUT) 0 for success, U1DB_SUPERSEDED if the document is
+ *     superseded.
+ */
+int u1db__validate_source(u1database *db, const char *replica_uid,
+                          int replica_gen, const char *replica_trans_id,
+                          u1db_vectorclock *cur_vcr,
+                          u1db_vectorclock *other_vcr, int *state);
+
+/**
  * Internal API, Get the global database rev.
  */
 int u1db__get_generation(u1database *db, int *generation);
 
 /**
+ * Internal API, Get the global database rev and transaction id.
+ */
+int u1db__get_generation_info(u1database *db, int *generation,
+                              char **trans_id);
+
+/**
+ * Internal API, Validate generation and transaction id.
+ */
+int u1db_validate_gen_and_trans_id(u1database *db, int generation,
+                                   const char *trans_id);
+/**
  * Internal API, Allocate a new document id, for cases when callers do not
  * supply their own. Callers of this API are expected to free the result.
  */

=== modified file 'src/u1db.c'
--- src/u1db.c	2012-06-12 12:35:05 +0000
+++ src/u1db.c	2012-06-18 18:24:19 +0000
@@ -722,6 +722,41 @@
     return status;
 }
 
+int
+u1db__validate_source(u1database *db, const char *replica_uid, int replica_gen,
+                      const char *replica_trans_id, u1db_vectorclock *cur,
+                      u1db_vectorclock *other, int *state)
+{
+    int old_generation;
+    char *old_trans_id = NULL;
+    int status = U1DB_OK;
+
+    *state = U1DB_OK;
+    status = u1db__get_sync_gen_info(
+        db, replica_uid, &old_generation, &old_trans_id);
+    if (status != U1DB_OK)
+        goto finish;
+    if (replica_gen < old_generation) {
+        if (u1db__vectorclock_is_newer(cur, other)) {
+            *state = U1DB_SUPERSEDED;
+            goto finish;
+        }
+        status = U1DB_INVALID_GENERATION;
+        goto finish;
+    }
+    if (replica_gen > old_generation)
+        goto finish;
+    if (strcmp(replica_trans_id, old_trans_id) != 0) {
+        status = U1DB_INVALID_TRANSACTION_ID;
+        goto finish;
+    }
+    *state = U1DB_SUPERSEDED;
+finish:
+    if (old_trans_id != NULL)
+        free(old_trans_id);
+    return status;
+}
+
 
 int
 u1db__put_doc_if_newer(u1database *db, u1db_document *doc, int save_conflict,
@@ -748,6 +783,31 @@
     status = lookup_doc(db, doc->doc_id, &stored_doc_rev, &stored_content,
                         &stored_content_len, &statement);
     if (status != SQLITE_OK) { goto finish; }
+    // TODO: u1db__vectorclock_from_str returns NULL if there is an error
+    //       in the vector clock, or if we run out of memory... Probably
+    //       shouldn't be U1DB_NOMEM
+    stored_vc = u1db__vectorclock_from_str(stored_doc_rev);
+    if (stored_vc == NULL) {
+        status = U1DB_NOMEM;
+        goto finish;
+    }
+    new_vc = u1db__vectorclock_from_str(doc->doc_rev);
+    if (new_vc == NULL) {
+        status = U1DB_NOMEM;
+        goto finish;
+    }
+    if (replica_uid != NULL && replica_trans_id != NULL) {
+        status = u1db__validate_source(
+            db, replica_uid, replica_gen, replica_trans_id, stored_vc, new_vc,
+            state);
+        if (status != U1DB_OK) {
+            goto finish;
+        }
+        if (*state != U1DB_OK) {
+            status = u1db__get_generation(db, at_gen);
+            goto finish;
+        }
+    }
     if (stored_doc_rev == NULL) {
         status = U1DB_OK;
         *state = U1DB_INSERTED;
@@ -757,19 +817,6 @@
         *state = U1DB_CONVERGED;
         store = 0;
     } else {
-        // TODO: u1db__vectorclock_from_str returns NULL if there is an error
-        //       in the vector clock, or if we run out of memory... Probably
-        //       shouldn't be U1DB_NOMEM
-        stored_vc = u1db__vectorclock_from_str(stored_doc_rev);
-        if (stored_vc == NULL) {
-            status = U1DB_NOMEM;
-            goto finish;
-        }
-        new_vc = u1db__vectorclock_from_str(doc->doc_rev);
-        if (new_vc == NULL) {
-            status = U1DB_NOMEM;
-            goto finish;
-        }
         if (u1db__vectorclock_is_newer(new_vc, stored_vc)) {
             // Just take the newer version
             store = 1;
@@ -818,8 +865,8 @@
                            (stored_doc_rev != NULL));
     }
     if (status == U1DB_OK && replica_uid != NULL) {
-        status = u1db__set_sync_info(db, replica_uid, replica_gen,
-                                     replica_trans_id);
+        status = u1db__set_sync_info(
+            db, replica_uid, replica_gen, replica_trans_id);
     }
     if (status == U1DB_OK && at_gen != NULL) {
         status = u1db__get_generation(db, at_gen);
@@ -1182,14 +1229,26 @@
         status = U1DB_OK;
         *gen = 0;
         *trans_id = strdup("");
+        if (*trans_id == NULL) {
+            status = U1DB_NOMEM;
+            goto finish;
+        }
     } else if (status == SQLITE_ROW) {
         status = U1DB_OK;
         *gen = sqlite3_column_int(statement, 0);
         tmp = (const char *)sqlite3_column_text(statement, 1);
         if (tmp == NULL) {
             *trans_id = strdup("");
+            if (*trans_id == NULL) {
+                status = U1DB_NOMEM;
+                goto finish;
+            }
         } else {
             *trans_id = strdup(tmp);
+            if (*trans_id == NULL) {
+                status = U1DB_NOMEM;
+                goto finish;
+            }
         }
     }
 finish:
@@ -1329,6 +1388,80 @@
     return status;
 }
 
+int
+u1db__get_generation_info(u1database *db, int *generation, char **trans_id)
+{
+    int status;
+    const char *tmp;
+
+    sqlite3_stmt *statement;
+    if (db == NULL || generation == NULL) {
+        return U1DB_INVALID_PARAMETER;
+    }
+    status = sqlite3_prepare_v2(db->sql_handle,
+        "SELECT max(generation), transaction_id FROM transaction_log", -1,
+        &statement, NULL);
+    if (status != SQLITE_OK) {
+        return status;
+    }
+    status = sqlite3_step(statement);
+    if (status == SQLITE_DONE) {
+        // No records, we are at rev 0
+        status = SQLITE_OK;
+        *generation = 0;
+    } else if (status == SQLITE_ROW) {
+        status = SQLITE_OK;
+        *generation = sqlite3_column_int(statement, 0);
+        tmp = (const char *)sqlite3_column_text(statement, 1);
+        if (tmp == NULL) {
+            *trans_id = NULL;
+        } else {
+            *trans_id = strdup(tmp);
+            if (*trans_id == NULL) {
+                status = U1DB_NOMEM;
+            }
+        }
+    }
+    sqlite3_finalize(statement);
+    return status;
+}
+
+int
+u1db_validate_gen_and_trans_id(u1database *db, int generation,
+                               const char *trans_id)
+{
+    int status = U1DB_OK;
+    sqlite3_stmt *statement;
+
+    if (generation == 0)
+        return status;
+    if (db == NULL) {
+        return U1DB_INVALID_PARAMETER;
+    }
+    status = sqlite3_prepare_v2(db->sql_handle,
+        "SELECT transaction_id FROM transaction_log WHERE generation = ?", -1,
+        &statement, NULL);
+    if (status != SQLITE_OK) { goto finish; }
+    status = sqlite3_bind_int(statement, 1, generation);
+    if (status != SQLITE_OK) { goto finish; }
+    status = sqlite3_step(statement);
+    if (status == SQLITE_DONE) {
+        status = U1DB_INVALID_GENERATION;
+        goto finish;
+    } else if (status == SQLITE_ROW) {
+        // Note: We may want to handle the column containing NULL
+        if (strcmp(trans_id,
+                   (const char *)sqlite3_column_text(statement, 0)) == 0) {
+            status = U1DB_OK;
+            goto finish;
+        }
+        status = U1DB_INVALID_TRANSACTION_ID;
+    }
+finish:
+    sqlite3_finalize(statement);
+    return status;
+}
+
 char *
 u1db__allocate_doc_id(u1database *db)
 {
@@ -1462,7 +1595,10 @@
         // Note: We may want to handle the column containing NULL
         tmp = (const char *)sqlite3_column_text(statement, 1);
         if (tmp == NULL) {
-            *trans_id = NULL;
+            *trans_id = strdup("");
+            if (*trans_id == NULL) {
+                status = U1DB_NOMEM;
+            }
         } else {
             *trans_id = strdup(tmp);
             if (*trans_id == NULL) {

=== modified file 'src/u1db_sync_target.c'
--- src/u1db_sync_target.c	2012-06-18 18:24:19 +0000
+++ src/u1db_sync_target.c	2012-06-18 18:24:19 +0000
@@ -24,7 +24,7 @@
 static int st_get_sync_info(u1db_sync_target *st,
         const char *source_replica_uid,
         const char **st_replica_uid, int *st_gen, int *source_gen,
-        char **trans_id);
+        char **source_trans_id);
 
 static int st_record_sync_info(u1db_sync_target *st,
         const char *source_replica_uid, int source_gen, const char *trans_id);
@@ -106,7 +106,7 @@
 static int
 st_get_sync_info(u1db_sync_target *st, const char *source_replica_uid,
         const char **st_replica_uid, int *st_gen, int *source_gen,
-        char **trans_id)
+        char **source_trans_id)
 {
     int status = U1DB_OK;
     u1database *db;
@@ -125,8 +125,8 @@
     db = (u1database *)st->implementation;
     status = u1db_get_replica_uid(db, st_replica_uid);
     if (status != U1DB_OK) { goto finish; }
-    status = u1db__get_sync_gen_info(db, source_replica_uid, source_gen,
-                                     trans_id);
+    status = u1db__get_sync_gen_info(
+        db, source_replica_uid, source_gen, source_trans_id);
     if (status != U1DB_OK) { goto finish; }
     status = u1db__get_generation(db, st_gen);
 finish:
@@ -614,6 +614,9 @@
     status = target->get_sync_info(target, local_uid, &target_uid, &target_gen,
                    &local_gen_known_by_target, &local_trans_id_known_by_target);
     if (status != U1DB_OK) { goto finish; }
+    status = u1db_validate_gen_and_trans_id(
+            db, local_gen_known_by_target, local_trans_id_known_by_target);
+    if (status != U1DB_OK) { goto finish; }
     status = u1db__get_sync_gen_info(db, target_uid,
         &target_gen_known_by_local, &target_trans_id_known_by_local);
     if (status != U1DB_OK) { goto finish; }
@@ -623,8 +626,9 @@
     // Before we start the sync exchange, get the list of doc_ids that we want
     // to send. We have to do this first, so that local_gen_before_sync will
     // match exactly the list of doc_ids we send
-    status = u1db_whats_changed(db, &local_gen, &local_trans_id,
-                            (void*)&to_send_state, whats_changed_to_doc_ids);
+    status = u1db_whats_changed(
+        db, &local_gen, &local_trans_id, (void*)&to_send_state,
+        whats_changed_to_doc_ids);
     if (status != U1DB_OK) { goto finish; }
     if (local_gen == local_gen_known_by_target
         && target_gen == target_gen_known_by_local)
@@ -644,19 +648,23 @@
         &target_gen_known_by_local, &target_trans_id_known_by_local,
         &return_doc_state, return_doc_to_insert_from_target);
     if (status != U1DB_OK) { goto finish; }
-    status = u1db__get_generation(db, &local_gen);
+    if (local_trans_id != NULL) {
+        free(local_trans_id);
+    }
+    status = u1db__get_generation_info(db, &local_gen, &local_trans_id);
     if (status != U1DB_OK) { goto finish; }
     // Now we successfully sent and received docs, make sure we record the
     // current remote generation
-    status = u1db__set_sync_info(db, target_uid, target_gen_known_by_local,
-                                 "T-sid");
+    status = u1db__set_sync_info(
+        db, target_uid, target_gen_known_by_local,
+        target_trans_id_known_by_local);
     if (status != U1DB_OK) { goto finish; }
     if (return_doc_state.num_inserted > 0 &&
-            ((*local_gen_before_sync + return_doc_state.num_inserted)
-              == local_gen))
+        ((*local_gen_before_sync + return_doc_state.num_inserted)
+         == local_gen))
     {
-        status = target->record_sync_info(target, local_uid, local_gen,
-                                          "T-sid");
+        status = target->record_sync_info(
+            target, local_uid, local_gen, local_trans_id);
         if (status != U1DB_OK) { goto finish; }
     }
 finish:

=== modified file 'u1db/backends/__init__.py'
--- u1db/backends/__init__.py	2012-05-30 14:39:36 +0000
+++ u1db/backends/__init__.py	2012-06-18 18:24:19 +0000
@@ -53,8 +53,17 @@
             raise errors.InvalidDocId()
 
     def _get_generation(self):
+        """Return the current generation.
+
+        """
         raise NotImplementedError(self._get_generation)
 
+    def _get_generation_info(self):
+        """Return the current generation and transaction id.
+
+        """
+        raise NotImplementedError(self._get_generation_info)
+
     def _get_doc(self, doc_id):
         """Extract the document from storage.
 
@@ -93,6 +102,35 @@
             result.append(doc)
         return result
 
+    def validate_gen_and_trans_id(self, generation, trans_id):
+        """Validate the generation and transaction id.
+
+        Raises an InvalidGeneration when the generation does not exist, and an
+        InvalidTransactionId when it does but with a different transaction id.
+
+        """
+        raise NotImplementedError(self.validate_gen_and_trans_id)
+
+    def _validate_source(self, other_replica_uid, other_generation,
+                         other_transaction_id, cur_vcr, other_vcr):
+        """Validate the new generation and transaction id.
+
+        other_generation must be greater than what we have stored for this
+        replica, *or* it must be the same and the transaction_id must be the
+        same as well.
+        """
+        old_generation, old_transaction_id = self._get_sync_gen_info(
+            other_replica_uid)
+        if other_generation < old_generation:
+            if cur_vcr.is_newer(other_vcr):
+                return 'superseded'
+            raise errors.InvalidGeneration
+        if other_generation > old_generation:
+            return 'ok'
+        if other_transaction_id == old_transaction_id:
+            return 'superseded'
+        raise errors.InvalidTransactionId
+
     def _put_doc_if_newer(self, doc, save_conflict, replica_uid=None,
                           replica_gen=None, replica_trans_id=None):
         cur_doc = self._get_doc(doc.doc_id)
@@ -101,6 +139,12 @@
             cur_vcr = VectorClockRev(None)
         else:
             cur_vcr = VectorClockRev(cur_doc.rev)
+        if replica_uid is not None and replica_gen is not None:
+            state = self._validate_source(
+                replica_uid, replica_gen, replica_trans_id, cur_vcr,
+                doc_vcr)
+            if state != 'ok':
+                return state, self._get_generation()
         if doc_vcr.is_newer(cur_vcr):
             self._put_and_update_indexes(cur_doc, doc)
             self._prune_conflicts(doc, doc_vcr)

=== modified file 'u1db/backends/inmemory.py'
--- u1db/backends/inmemory.py	2012-06-12 15:44:43 +0000
+++ u1db/backends/inmemory.py	2012-06-18 18:24:19 +0000
@@ -79,6 +79,18 @@
     def _get_generation(self):
         return len(self._transaction_log)
 
+    def _get_generation_info(self):
+        return len(self._transaction_log), self._transaction_log[-1][1]
+
+    def validate_gen_and_trans_id(self, generation, trans_id):
+        if generation == 0:
+            return
+        if generation > len(self._transaction_log):
+            raise errors.InvalidGeneration
+        if self._transaction_log[generation - 1][1] == trans_id:
+            return
+        raise errors.InvalidTransactionId
+
     def put_doc(self, doc):
         if doc.doc_id is None:
             raise errors.InvalidDocId()

=== modified file 'u1db/backends/sqlite_backend.py'
--- u1db/backends/sqlite_backend.py	2012-06-12 15:44:43 +0000
+++ u1db/backends/sqlite_backend.py	2012-06-18 18:24:19 +0000
@@ -265,6 +265,28 @@
             return 0
         return val
 
+    def _get_generation_info(self):
+        c = self._db_handle.cursor()
+        c.execute(
+            'SELECT max(generation), transaction_id FROM transaction_log ')
+        val = c.fetchone()
+        if val[0] is None:
+            return(0, val[1])
+        return val
+
+    def validate_gen_and_trans_id(self, generation, trans_id):
+        if generation == 0:
+            return
+        c = self._db_handle.cursor()
+        c.execute(
+            'SELECT transaction_id FROM transaction_log WHERE generation = ?',
+            (generation,))
+        val = c.fetchone()
+        if val is None:
+            raise errors.InvalidGeneration
+        if val[0] != trans_id:
+            raise errors.InvalidTransactionId
+
     def _get_transaction_log(self):
         c = self._db_handle.cursor()
         c.execute("SELECT doc_id, transaction_id FROM transaction_log"

=== modified file 'u1db/errors.py'
--- u1db/errors.py	2012-05-25 18:14:06 +0000
+++ u1db/errors.py	2012-06-18 18:24:19 +0000
@@ -39,6 +39,14 @@
     wire_description = "invalid document id"
 
 
+class InvalidTransactionId(U1DBError):
+    """Invalid transaction for generation."""
+
+
+class InvalidGeneration(U1DBError):
+    """Generation was previously synced with a different transaction id."""
+
+
 class ConflictedDoc(U1DBError):
     """The document is conflicted, you must call resolve before put()"""
 

=== modified file 'u1db/sync.py'
--- u1db/sync.py	2012-06-18 18:24:19 +0000
+++ u1db/sync.py	2012-06-18 18:24:19 +0000
@@ -83,11 +83,11 @@
         record with the target that they are fully up to date with our
         new generation.
         """
-        cur_gen = self.source._get_generation()
+        cur_gen, trans_id = self.source._get_generation_info()
         if (cur_gen == start_generation + self.num_inserted
-            and self.num_inserted > 0):
-            self.sync_target.record_sync_info(self.source._replica_uid,
-                                              cur_gen, 'T-sid')
+                and self.num_inserted > 0):
+            self.sync_target.record_sync_info(
+                self.source._replica_uid, cur_gen, trans_id)
 
     def sync(self, callback=None):
         """Synchronize documents between source and target."""
@@ -98,6 +98,8 @@
          target_my_trans_id) = sync_target.get_sync_info(
              self.source._replica_uid)
         # what's changed since that generation and this current gen
+        self.source.validate_gen_and_trans_id(
+            target_my_gen, target_my_trans_id)
         my_gen, _, changes = self.source.whats_changed(target_my_gen)
 
         # this source last-seen database generation for the target
@@ -121,7 +123,8 @@
                         self.source._replica_uid, target_last_known_gen,
                         return_doc_cb=self._insert_doc_from_target)
         # record target synced-up-to generation including applying what we sent
-        self.source._set_sync_info(self.target_replica_uid, new_gen, 'T-sid')
+        self.source._set_sync_info(
+            self.target_replica_uid, new_gen, new_trans_id)
 
         # if gapless record current reached generation with target
         self._record_sync_info_with_the_target(my_gen)

=== modified file 'u1db/tests/c_backend_wrapper.pyx'
--- u1db/tests/c_backend_wrapper.pyx	2012-06-18 18:24:19 +0000
+++ u1db/tests/c_backend_wrapper.pyx	2012-06-18 18:24:19 +0000
@@ -80,6 +80,10 @@
     int u1db_get_all_docs(u1database *db, int include_deleted, int *generation,
                           void *context, u1db_doc_callback cb)
     int u1db_put_doc(u1database *db, u1db_document *doc)
+    int u1db__validate_source(u1database *db, const_char_ptr replica_uid,
+                              int replica_gen, const_char_ptr replica_trans_id,
+                              u1db_vectorclock *cur_vcr,
+                              u1db_vectorclock *other_vcr, int *state)
     int u1db__put_doc_if_newer(u1database *db, u1db_document *doc,
                                int save_conflict, char *replica_uid,
                                int replica_gen, char *replica_trans_id,
@@ -131,6 +135,8 @@
     int U1DB_BROKEN_SYNC_STREAM
     int U1DB_DUPLICATE_INDEX_NAME
     int U1DB_INDEX_DOES_NOT_EXIST
+    int U1DB_INVALID_GENERATION
+    int U1DB_INVALID_TRANSACTION_ID
     int U1DB_INTERNAL_ERROR
 
     int U1DB_INSERTED
@@ -198,6 +204,9 @@
 
 
     int u1db__get_generation(u1database *, int *db_rev)
+    int u1db__get_generation_info(u1database *, int *db_rev, char **trans_id)
+    int u1db_validate_gen_and_trans_id(u1database *, int db_rev,
+                                       const_char_ptr trans_id)
     char *u1db__allocate_doc_id(u1database *)
     int u1db__sql_close(u1database *)
     int u1db__sql_is_open(u1database *)
@@ -580,6 +589,10 @@
         raise errors.IndexNameTakenError()
     if status == U1DB_INDEX_DOES_NOT_EXIST:
         raise errors.IndexDoesNotExist
+    if status == U1DB_INVALID_GENERATION:
+        raise errors.InvalidGeneration
+    if status == U1DB_INVALID_TRANSACTION_ID:
+        raise errors.InvalidTransactionId
     raise RuntimeError('%s (status: %s)' % (context, status))
 
 
@@ -912,6 +925,27 @@
             u1db_put_doc(self._db, doc._doc))
         return doc.rev
 
+    def _validate_source(self, replica_uid, replica_gen, replica_trans_id,
+                         cur_vcr, other_vcr):
+        cdef const_char_ptr c_uid, c_trans_id
+        cdef int c_gen, state = 0
+        cdef VectorClockRev cur
+        cdef VectorClockRev other
+
+        cur = VectorClockRev(cur_vcr.as_str())
+        other = VectorClockRev(other_vcr.as_str())
+        c_uid = replica_uid
+        c_trans_id = replica_trans_id
+        c_gen = replica_gen
+        handle_status(
+            "invalid generation or transaction id",
+            u1db__validate_source(
+                self._db, c_uid, c_gen, c_trans_id, cur._clock, other._clock,
+                &state))
+        if state == U1DB_SUPERSEDED:
+            return 'superseded'
+        return 'ok'
+
     def _put_doc_if_newer(self, CDocument doc, save_conflict, replica_uid=None,
                           replica_gen=None, replica_trans_id=None):
         cdef char *c_uid, *c_trans_id
@@ -1033,6 +1067,22 @@
             u1db__get_generation(self._db, &generation))
         return generation
 
+    def _get_generation_info(self):
+        cdef int generation
+        cdef char *trans_id
+        handle_status("get_generation_info",
+            u1db__get_generation_info(self._db, &generation, &trans_id))
+        raw_trans_id = None
+        if trans_id != NULL:
+            raw_trans_id = trans_id
+            free(trans_id)
+        return generation, raw_trans_id
+
+    def validate_gen_and_trans_id(self, generation, trans_id):
+        handle_status(
+            "validate_gen_and_trans_id",
+            u1db_validate_gen_and_trans_id(self._db, generation, trans_id))
+
     def _get_sync_gen_info(self, replica_uid):
         cdef int generation, status
         cdef char *trans_id = NULL

=== modified file 'u1db/tests/test_backends.py'
--- u1db/tests/test_backends.py	2012-06-18 18:24:19 +0000
+++ u1db/tests/test_backends.py	2012-06-18 18:24:19 +0000
@@ -392,6 +392,112 @@
         # The database wasn't altered
         self.assertGetDoc(self.db, doc1.doc_id, doc1.rev, simple_doc, False)
 
+    def test_put_doc_if_newer_newer_generation(self):
+        self.db._set_sync_info('other', 1, 'T-sid')
+        doc = self.make_document('doc_id', 'other:2', simple_doc)
+        state, _ = self.db._put_doc_if_newer(
+            doc, save_conflict=False, replica_uid='other', replica_gen=2,
+            replica_trans_id='T-irrelevant')
+        self.assertEqual('inserted', state)
+
+    def test_put_doc_if_newer_same_generation_same_txid(self):
+        self.db._set_sync_info('other', 1, 'T-sid')
+        doc = self.make_document('doc_id', 'other:2', simple_doc)
+        state, _ = self.db._put_doc_if_newer(
+            doc, save_conflict=False, replica_uid='other', replica_gen=1,
+            replica_trans_id='T-sid')
+        self.assertEqual('superseded', state)
+
+    def test_put_doc_if_newer_wrong_transaction_id(self):
+        self.db._set_sync_info('other', 1, 'T-sid')
+        doc = self.make_document('doc_id', 'other:1', simple_doc)
+        self.assertRaises(
+            errors.InvalidTransactionId,
+            self.db._put_doc_if_newer, doc, save_conflict=False,
+            replica_uid='other', replica_gen=1, replica_trans_id='T-sad')
+
+    def test_put_doc_if_newer_old_generation_older_doc(self):
+        orig_doc = '{"new": "doc"}'
+        doc = self.db.create_doc(orig_doc)
+        doc_rev1 = doc.rev
+        doc.set_json(simple_doc)
+        self.db.put_doc(doc)
+        self.db._set_sync_info('other', 5, 'T-sid')
+        older_doc = self.make_document(doc.doc_id, doc_rev1, simple_doc)
+        state, _ = self.db._put_doc_if_newer(
+            older_doc, save_conflict=False, replica_uid='other', replica_gen=3,
+            replica_trans_id='T-irrelevant')
+        self.assertEqual('superseded', state)
+
+    def test_put_doc_if_newer_old_generation_newer_doc(self):
+        self.db._set_sync_info('other', 5, 'T-sid')
+        doc = self.make_document('doc_id', 'other:1', simple_doc)
+        self.assertRaises(
+            errors.InvalidGeneration,
+            self.db._put_doc_if_newer, doc, save_conflict=False,
+            replica_uid='other', replica_gen=1, replica_trans_id='T-sad')
+
+    def test_validate_gen_and_trans_id(self):
+        self.db.create_doc(simple_doc)
+        gen, trans_id = self.db._get_generation_info()
+        self.db.validate_gen_and_trans_id(gen, trans_id)
+
+    def test_validate_gen_and_trans_id_invalid_txid(self):
+        self.db.create_doc(simple_doc)
+        gen, _ = self.db._get_generation_info()
+        self.assertRaises(
+            errors.InvalidTransactionId,
+            self.db.validate_gen_and_trans_id, gen, 'wrong')
+
+    def test_validate_gen_and_trans_id_invalid_txid(self):
+        self.db.create_doc(simple_doc)
+        gen, trans_id = self.db._get_generation_info()
+        self.assertRaises(
+            errors.InvalidGeneration,
+            self.db.validate_gen_and_trans_id, gen + 1, trans_id)
+
+    def test_validate_source_gen_and_trans_id_same(self):
+        self.db._set_sync_info('other', 1, 'T-sid')
+        v1 = vectorclock.VectorClockRev('other:1|self:1')
+        v2 = vectorclock.VectorClockRev('other:2|self:2')
+        self.assertEqual(
+            'superseded',
+            self.db._validate_source('other', 1, 'T-sid', v1, v2))
+
+    def test_validate_source_gen_newer(self):
+        self.db._set_sync_info('other', 1, 'T-sid')
+        v1 = vectorclock.VectorClockRev('other:1|self:1')
+        v2 = vectorclock.VectorClockRev('other:2|self:2')
+        self.assertEqual(
+            'ok',
+            self.db._validate_source('other', 2, 'T-whatevs', v1, v2))
+
+    def test_validate_source_wrong_txid(self):
+        self.db._set_sync_info('other', 1, 'T-sid')
+        v1 = vectorclock.VectorClockRev('other:1|self:1')
+        v2 = vectorclock.VectorClockRev('other:2|self:2')
+        self.assertRaises(
+            errors.InvalidTransactionId,
+            self.db._validate_source, 'other', 1, 'T-sad', v1, v2)
+
+    def test_validate_source_gen_older_and_vcr_older(self):
+        self.db._set_sync_info('other', 1, 'T-sid')
+        self.db._set_sync_info('other', 2, 'T-sod')
+        v1 = vectorclock.VectorClockRev('other:1|self:1')
+        v2 = vectorclock.VectorClockRev('other:2|self:2')
+        self.assertEqual(
+            'superseded',
+            self.db._validate_source('other', 1, 'T-sid', v2, v1))
+
+    def test_validate_source_gen_older_vcr_newer(self):
+        self.db._set_sync_info('other', 1, 'T-sid')
+        self.db._set_sync_info('other', 2, 'T-sod')
+        v1 = vectorclock.VectorClockRev('other:1|self:1')
+        v2 = vectorclock.VectorClockRev('other:2|self:2')
+        self.assertRaises(
+            errors.InvalidGeneration,
+            self.db._validate_source, 'other', 1, 'T-sid', v1, v2)
+
     def test_put_doc_if_newer_replica_uid(self):
         doc1 = self.db.create_doc(simple_doc)
         self.db._set_sync_info('other', 1, 'T-sid')
@@ -636,7 +742,8 @@
         doc2 = self.make_document(doc1.doc_id, 'alternate:1', nested_doc)
         self.db._put_doc_if_newer(doc2, save_conflict=True)
         self.assertTrue(doc2.has_conflicts)
-        self.assertGetDoc(self.db, doc1.doc_id, 'alternate:1', nested_doc, True)
+        self.assertGetDoc(
+            self.db, doc1.doc_id, 'alternate:1', nested_doc, True)
         self.assertGetDocConflicts(self.db, doc1.doc_id,
             [('alternate:1', nested_doc), (doc1.rev, None)])
 

=== modified file 'u1db/tests/test_c_backend.py'
--- u1db/tests/test_c_backend.py	2012-06-18 18:24:19 +0000
+++ u1db/tests/test_c_backend.py	2012-06-18 18:24:19 +0000
@@ -73,6 +73,14 @@
         db.create_doc(tests.simple_doc)
         self.assertEqual(1, db._get_generation())
 
+    def test__get_generation_info(self):
+        db = c_backend_wrapper.CDatabase(':memory:')
+        self.assertEqual((0, None), db._get_generation_info())
+        db.create_doc(tests.simple_doc)
+        info = db._get_generation_info()
+        self.assertEqual(1, info[0])
+        self.assertTrue(info[1].startswith('T-'))
+
     def test__set_replica_uid(self):
         db = c_backend_wrapper.CDatabase(':memory:')
         self.assertIsNot(None, db._replica_uid)

=== modified file 'u1db/tests/test_sqlite_backend.py'
--- u1db/tests/test_sqlite_backend.py	2012-05-31 16:31:31 +0000
+++ u1db/tests/test_sqlite_backend.py	2012-06-18 18:24:19 +0000
@@ -152,6 +152,9 @@
     def test__get_generation(self):
         self.assertEqual(0, self.db._get_generation())
 
+    def test__get_generation_info(self):
+        self.assertEqual((0, None), self.db._get_generation_info())
+
     def test_create_index(self):
         self.db.create_index('test-idx', "key")
         self.assertEqual([('test-idx', ["key"])], self.db.list_indexes())
@@ -261,7 +264,7 @@
     def test__open_database_with_factory(self):
         temp_dir = self.createTempDir(prefix='u1db-test-')
         path = temp_dir + '/test.sqlite'
-        db = sqlite_backend.SQLitePartialExpandDatabase(path)
+        sqlite_backend.SQLitePartialExpandDatabase(path)
         db2 = sqlite_backend.SQLiteDatabase._open_database(
             path, document_factory=TestAlternativeDocument)
         self.assertEqual(TestAlternativeDocument, db2._factory)
@@ -322,7 +325,7 @@
     def test_open_database_with_factory(self):
         temp_dir = self.createTempDir(prefix='u1db-test-')
         path = temp_dir + '/existing.sqlite'
-        db = sqlite_backend.SQLitePartialExpandDatabase(path)
+        sqlite_backend.SQLitePartialExpandDatabase(path)
         db2 = sqlite_backend.SQLiteDatabase.open_database(
             path, create=False, document_factory=TestAlternativeDocument)
         self.assertEqual(TestAlternativeDocument, db2._factory)

=== modified file 'u1db/tests/test_sync.py'
--- u1db/tests/test_sync.py	2012-06-18 18:24:19 +0000
+++ u1db/tests/test_sync.py	2012-06-18 18:24:19 +0000
@@ -662,6 +662,12 @@
         doc1 = self.db1.create_doc('{"a": 1}', doc_id='the-doc')
         db3 = self.create_database('test3')
         self.sync(self.db2, self.db1)
+        self.assertEqual(
+            self.db1._get_generation_info(),
+            self.db2._get_sync_gen_info(self.db1._replica_uid))
+        self.assertEqual(
+            self.db2._get_generation_info(),
+            self.db1._get_sync_gen_info(self.db2._replica_uid))
         self.sync(db3, self.db1)
         # update on 2
         doc2 = self.make_document('the-doc', doc1.rev, '{"a": 2}')
@@ -695,7 +701,19 @@
         self.db2.create_doc('{"b": 1}', doc_id='the-doc')
         db3.create_doc('{"c": 1}', doc_id='the-doc')
         self.sync(db3, self.db1)
+        self.assertEqual(
+            self.db1._get_generation_info(),
+            db3._get_sync_gen_info(self.db1._replica_uid))
+        self.assertEqual(
+            db3._get_generation_info(),
+            self.db1._get_sync_gen_info(db3._replica_uid))
         self.sync(db3, self.db2)
+        self.assertEqual(
+            self.db2._get_generation_info(),
+            db3._get_sync_gen_info(self.db2._replica_uid))
+        self.assertEqual(
+            db3._get_generation_info(),
+            self.db2._get_sync_gen_info(db3._replica_uid))
         self.assertEqual(3, len(db3.get_doc_conflicts('the-doc')))
         doc1.set_json('{"a": 2}')
         self.db1.put_doc(doc1)