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

[Eric Casteleijn]

Added the target transaction id to the return parameters of get_sync_info, since we need it in sync() to be able to tell if the target diverged if it has the generation we expect.

Unmerged revisions

=== modified file 'include/u1db/u1db_internal.h'
--- include/u1db/u1db_internal.h	2012-07-03 15:48:59 +0000
+++ include/u1db/u1db_internal.h	2012-07-06 20:51:26 +0000
@@ -62,23 +62,25 @@
      *                          Note that this is const char and memory will be
      *                          managed by the sync_target, so it should *not*
      *                          be freed.
-     * @param st_get            (OUT) The database generation for this sync
+     * @param st_gen            (OUT) The database generation for this sync
+     *                          target, matches st_replica_uid
+     * @param st_trans_id       (OUT) The database transaction id for this sync
      *                          target, matches st_replica_uid
      * @param source_gen        (OUT) The last generation of source_replica_uid
      *                          that st has synchronized with.
-     * @param trans_id          (OUT) The transaction id associated with the
+     * @param source_trans_id   (OUT) The transaction id associated with the
      *                          source generation, the memory must be freed by
      *                          the caller.
      */
     int (*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);
+        const char **st_replica_uid, int *st_gen, char **st_trans_id,
+        int *source_gen, char **source_trans_id);
     /**
      * Set the synchronization information about another replica.
      *
      * @param st    Pass this sync_target to the function,
-     *              eg st->get_sync_info(st, ...)
+     *              eg st->record_sync_info(st, ...)
      * @param source_replica_uid    The unique identifier for the source we
      *                              want to synchronize from.
      * @param source_gen        The last generation of source_replica_uid
@@ -223,9 +225,7 @@
  *     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);
+                          int replica_gen, const char *replica_trans_id);
 
 /**
  * Internal API, Get the global database rev.
@@ -239,6 +239,12 @@
                               char **trans_id);
 
 /**
+ * Internal API, Get the transaction id for the db generation.
+ */
+int u1db__get_trans_id_for_gen(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,

=== modified file 'src/u1db.c'
--- src/u1db.c	2012-07-03 15:48:59 +0000
+++ src/u1db.c	2012-07-06 20:51:26 +0000
@@ -42,7 +42,6 @@
                              char **doc_rev);
 static int generate_transaction_id(char buf[35]);
 
-
 static int
 initialize(u1database *db)
 {
@@ -756,23 +755,17 @@
 
 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)
+                      const char *replica_trans_id)
 {
     int old_generation;
     char *old_trans_id = NULL;
     int status = U1DB_OK;
 
-    *state = U1DB_OK;
     status = u1db__get_replica_gen_and_trans_id(
         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;
     }
@@ -782,7 +775,6 @@
         status = U1DB_INVALID_TRANSACTION_ID;
         goto finish;
     }
-    *state = U1DB_SUPERSEDED;
 finish:
     if (old_trans_id != NULL)
         free(old_trans_id);
@@ -830,15 +822,10 @@
     }
     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);
+            db, replica_uid, replica_gen, replica_trans_id);
         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;
@@ -1464,14 +1451,60 @@
         *generation = 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;
+            }
+        }
+        status = U1DB_OK;
+    }
+finish:
+    sqlite3_finalize(statement);
+    return status;
+}
+
+int
+u1db__get_trans_id_for_gen(u1database *db, int generation,
+                           char **trans_id)
+{
+    int status = U1DB_OK;
+    sqlite3_stmt *statement;
+    const char *tmp;
+
+    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) {
+        tmp = (const char *)sqlite3_column_text(statement, 0);
+        if (tmp == NULL) {
             *trans_id = NULL;
         } else {
             *trans_id = strdup(tmp);
             if (*trans_id == NULL) {
                 status = U1DB_NOMEM;
+                goto finish;
             }
         }
+        status = U1DB_OK;
     }
+finish:
     sqlite3_finalize(statement);
     return status;
 }
@@ -1481,34 +1514,21 @@
                                const char *trans_id)
 {
     int status = U1DB_OK;
-    sqlite3_stmt *statement;
+    char *known_trans_id = NULL;
 
     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;
+    status = u1db__get_trans_id_for_gen(db, generation, &known_trans_id);
+    if (status != U1DB_OK)
         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) {
+    if (strcmp(trans_id, known_trans_id) == 0) {
             status = U1DB_OK;
             goto finish;
         }
-        status = U1DB_INVALID_TRANSACTION_ID;
-    }
+    status = U1DB_INVALID_TRANSACTION_ID;
 finish:
-    sqlite3_finalize(statement);
+    if (known_trans_id != NULL)
+        free(known_trans_id);
     return status;
 }
 

=== modified file 'src/u1db_http_sync_target.c'
--- src/u1db_http_sync_target.c	2012-07-03 15:48:59 +0000
+++ src/u1db_http_sync_target.c	2012-07-06 20:51:26 +0000
@@ -41,9 +41,10 @@
 struct _http_request;
 
 static int st_http_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);
+                                 const char *source_replica_uid,
+                                 const char **st_replica_uid, int *st_gen,
+                                 char **st_trans_id, int *source_gen,
+                                 char **trans_id);
 
 static int st_http_record_sync_info(u1db_sync_target *st,
         const char *source_replica_uid, int source_gen, const char *trans_id);
@@ -366,15 +367,15 @@
 }
 
 static int
-st_http_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)
+st_http_get_sync_info(u1db_sync_target *st, const char *source_replica_uid,
+                      const char **st_replica_uid, int *st_gen,
+                      char **st_trans_id, int *source_gen, char **trans_id)
 {
     struct _http_state *state;
     struct _http_request req = {0};
     char *url = NULL;
     const char *tmp = NULL;
+    const char *tmp2 = NULL;
     int status = U1DB_OK;
     long http_code;
     struct curl_slist *headers = NULL;
@@ -481,6 +482,21 @@
         goto finish;
     }
     *st_gen = json_object_get_int(obj);
+    obj = json_object_object_get(json, "target_replica_transaction_id");
+    if (obj == NULL) {
+        status = U1DB_INVALID_HTTP_RESPONSE;
+        goto finish;
+    }
+    tmp2 = json_object_get_string(obj);
+    if (tmp2 == NULL) {
+        *st_trans_id = NULL;
+    } else {
+        *st_trans_id = strdup(tmp2);
+        if (*st_trans_id == NULL) {
+            status = U1DB_NOMEM;
+            goto finish;
+        }
+    }
     obj = json_object_object_get(json, "source_replica_generation");
     if (obj == NULL) {
         status = U1DB_INVALID_HTTP_RESPONSE;

=== modified file 'src/u1db_sync_target.c'
--- src/u1db_sync_target.c	2012-07-03 15:48:59 +0000
+++ src/u1db_sync_target.c	2012-07-06 20:51:26 +0000
@@ -22,9 +22,10 @@
 
 
 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 **source_trans_id);
+                            const char *source_replica_uid,
+                            const char **st_replica_uid, int *st_gen,
+                            char **st_trans_id, int *source_gen,
+                            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);
@@ -105,8 +106,8 @@
 
 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 **source_trans_id)
+                 const char **st_replica_uid, int *st_gen, char **st_trans_id,
+                 int *source_gen, char **source_trans_id)
 {
     int status = U1DB_OK;
     u1database *db;
@@ -128,7 +129,7 @@
     status = u1db__get_replica_gen_and_trans_id(
         db, source_replica_uid, source_gen, source_trans_id);
     if (status != U1DB_OK) { goto finish; }
-    status = u1db__get_generation(db, st_gen);
+    status = u1db__get_generation_info(db, st_gen, st_trans_id);
 finish:
     return status;
 }
@@ -594,6 +595,7 @@
     char *local_target_trans_id = NULL;
     char *target_trans_id_known_by_local = NULL;
     char *local_trans_id_known_by_target = NULL;
+    char *target_trans_id = NULL;
     int target_gen, local_gen;
     int local_gen_known_by_target, target_gen_known_by_local;
 
@@ -608,14 +610,15 @@
     if (status != U1DB_OK) { goto finish; }
     // fprintf(stderr, "Local uid: %s\n", local_uid);
     status = target->get_sync_info(
-        target, local_uid, &target_uid, &target_gen,
+        target, local_uid, &target_uid, &target_gen, &target_trans_id,
         &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_replica_gen_and_trans_id(db, target_uid,
-        &target_gen_known_by_local, &target_trans_id_known_by_local);
+    status = u1db__get_replica_gen_and_trans_id(
+        db, target_uid, &target_gen_known_by_local,
+        &target_trans_id_known_by_local);
     if (status != U1DB_OK) { goto finish; }
     local_target_trans_id = target_trans_id_known_by_local;
     local_gen = local_gen_known_by_target;
@@ -630,6 +633,9 @@
     if (local_gen == local_gen_known_by_target
         && target_gen == target_gen_known_by_local)
     {
+        if (strcmp(target_trans_id, target_trans_id_known_by_local) != 0) {
+            status = U1DB_INVALID_TRANSACTION_ID;
+        }
         // We know status == U1DB_OK, and we can shortcut the rest of the
         // logic, no need to look for more information.
         goto finish;
@@ -671,6 +677,9 @@
     if (local_trans_id_known_by_target != NULL) {
         free(local_trans_id_known_by_target);
     }
+    if (target_trans_id != NULL) {
+        free(target_trans_id);
+    }
     if (local_target_trans_id != NULL) {
         if (target_trans_id_known_by_local == local_target_trans_id) {
             // Don't double free

=== modified file 'u1db/__init__.py'
--- u1db/__init__.py	2012-07-03 15:48:59 +0000
+++ u1db/__init__.py	2012-07-06 20:51:26 +0000
@@ -290,7 +290,7 @@
             encountered during synchronization. If we've never synchronized
             with the replica, this is (0, '').
         """
-        raise NotImplementedError(self._replica_gen_and_trans_id)
+        raise NotImplementedError(self._get_replica_gen_and_trans_id)
 
     def _set_replica_gen_and_trans_id(self, other_replica_uid,
                                       other_generation, other_transaction_id):
@@ -536,7 +536,7 @@
         :param source_replica_uid: Another replica which we might have
             synchronized with in the past.
         :return: (target_replica_uid, target_replica_generation,
-                  source_replica_last_known_generation,
+                  target_trans_id, source_replica_last_known_generation,
                   source_replica_last_known_transaction_id)
         """
         raise NotImplementedError(self.get_sync_info)

=== modified file 'u1db/backends/__init__.py'
--- u1db/backends/__init__.py	2012-07-03 13:50:27 +0000
+++ u1db/backends/__init__.py	2012-07-06 20:51:26 +0000
@@ -102,6 +102,14 @@
             result.append(doc)
         return result
 
+    def _get_trans_id_for_gen(self, generation):
+        """Get the transaction id corresponding to a particular generation.
+
+        Raises an InvalidGeneration when the generation does not exist.
+
+        """
+        raise NotImplementedError(self._get_trans_id_for_gen)
+
     def validate_gen_and_trans_id(self, generation, trans_id):
         """Validate the generation and transaction id.
 
@@ -109,10 +117,14 @@
         InvalidTransactionId when it does but with a different transaction id.
 
         """
-        raise NotImplementedError(self.validate_gen_and_trans_id)
+        if generation == 0:
+            return
+        known_trans_id = self._get_trans_id_for_gen(generation)
+        if known_trans_id != trans_id:
+            raise errors.InvalidTransactionId
 
     def _validate_source(self, other_replica_uid, other_generation,
-                         other_transaction_id, cur_vcr, other_vcr):
+                         other_transaction_id):
         """Validate the new generation and transaction id.
 
         other_generation must be greater than what we have stored for this
@@ -123,13 +135,11 @@
          old_transaction_id) = self._get_replica_gen_and_trans_id(
              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'
+            return
         if other_transaction_id == old_transaction_id:
-            return 'superseded'
+            return
         raise errors.InvalidTransactionId
 
     def _put_doc_if_newer(self, doc, save_conflict, replica_uid=None,
@@ -141,11 +151,8 @@
         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()
+            self._validate_source(
+                replica_uid, replica_gen, replica_trans_id)
         if doc_vcr.is_newer(cur_vcr):
             rev = doc.rev
             self._prune_conflicts(doc, doc_vcr)

=== modified file 'u1db/backends/inmemory.py'
--- u1db/backends/inmemory.py	2012-07-03 15:48:59 +0000
+++ u1db/backends/inmemory.py	2012-07-06 20:51:26 +0000
@@ -85,16 +85,16 @@
         return len(self._transaction_log)
 
     def _get_generation_info(self):
+        if not self._transaction_log:
+            return 0, ''
         return len(self._transaction_log), self._transaction_log[-1][1]
 
-    def validate_gen_and_trans_id(self, generation, trans_id):
+    def _get_trans_id_for_gen(self, generation):
         if generation == 0:
-            return
+            return ''
         if generation > len(self._transaction_log):
             raise errors.InvalidGeneration
-        if self._transaction_log[generation - 1][1] == trans_id:
-            return
-        raise errors.InvalidTransactionId
+        return self._transaction_log[generation - 1][1]
 
     def put_doc(self, doc):
         if doc.doc_id is None:
@@ -448,10 +448,12 @@
 class InMemorySyncTarget(CommonSyncTarget):
 
     def get_sync_info(self, source_replica_uid):
-        source_gen, trans_id = self._db._get_replica_gen_and_trans_id(
+        source_gen, source_trans_id = self._db._get_replica_gen_and_trans_id(
             source_replica_uid)
-        return (self._db._replica_uid, len(self._db._transaction_log),
-                source_gen, trans_id)
+        my_gen, my_trans_id = self._db._get_generation_info()
+        return (
+            self._db._replica_uid, my_gen, my_trans_id, source_gen,
+            source_trans_id)
 
     def record_sync_info(self, source_replica_uid, source_replica_generation,
                          source_transaction_id):

=== modified file 'u1db/backends/sqlite_backend.py'
--- u1db/backends/sqlite_backend.py	2012-07-03 13:50:27 +0000
+++ u1db/backends/sqlite_backend.py	2012-07-06 20:51:26 +0000
@@ -271,12 +271,12 @@
             'SELECT max(generation), transaction_id FROM transaction_log ')
         val = c.fetchone()
         if val[0] is None:
-            return(0, val[1])
+            return(0, '')
         return val
 
-    def validate_gen_and_trans_id(self, generation, trans_id):
+    def _get_trans_id_for_gen(self, generation):
         if generation == 0:
-            return
+            return ''
         c = self._db_handle.cursor()
         c.execute(
             'SELECT transaction_id FROM transaction_log WHERE generation = ?',
@@ -284,8 +284,7 @@
         val = c.fetchone()
         if val is None:
             raise errors.InvalidGeneration
-        if val[0] != trans_id:
-            raise errors.InvalidTransactionId
+        return val[0]
 
     def _get_transaction_log(self):
         c = self._db_handle.cursor()
@@ -788,10 +787,12 @@
 class SQLiteSyncTarget(CommonSyncTarget):
 
     def get_sync_info(self, source_replica_uid):
-        source_gen, trans_id = self._db._get_replica_gen_and_trans_id(
+        source_gen, source_trans_id = self._db._get_replica_gen_and_trans_id(
             source_replica_uid)
-        my_gen = self._db._get_generation()
-        return self._db._replica_uid, my_gen, source_gen, trans_id
+        my_gen, my_trans_id = self._db._get_generation_info()
+        return (
+            self._db._replica_uid, my_gen, my_trans_id, source_gen,
+            source_trans_id)
 
     def record_sync_info(self, source_replica_uid, source_replica_generation,
                          source_replica_transaction_id):

=== modified file 'u1db/remote/http_app.py'
--- u1db/remote/http_app.py	2012-07-03 15:48:59 +0000
+++ u1db/remote/http_app.py	2012-07-06 20:51:26 +0000
@@ -309,9 +309,10 @@
         result = self.target.get_sync_info(self.source_replica_uid)
         self.responder.send_response_json(
             target_replica_uid=result[0], target_replica_generation=result[1],
+            target_replica_transaction_id=result[2],
             source_replica_uid=self.source_replica_uid,
-            source_replica_generation=result[2],
-            source_transaction_id=result[3])
+            source_replica_generation=result[3],
+            source_transaction_id=result[4])
 
     @http_method(generation=int,
                  content_as_args=True, no_query=True)

=== modified file 'u1db/remote/http_target.py'
--- u1db/remote/http_target.py	2012-06-19 02:35:49 +0000
+++ u1db/remote/http_target.py	2012-07-06 20:51:26 +0000
@@ -42,6 +42,7 @@
         self._ensure_connection()
         res, _ = self._request_json('GET', ['sync-from', source_replica_uid])
         return (res['target_replica_uid'], res['target_replica_generation'],
+                res['target_replica_transaction_id'],
                 res['source_replica_generation'], res['source_transaction_id'])
 
     def record_sync_info(self, source_replica_uid, source_replica_generation,

=== modified file 'u1db/sync.py'
--- u1db/sync.py	2012-07-03 15:48:59 +0000
+++ u1db/sync.py	2012-07-06 20:51:26 +0000
@@ -18,6 +18,7 @@
 from itertools import izip
 
 import u1db
+from u1db import errors
 
 
 class Synchronizer(object):
@@ -94,21 +95,21 @@
         sync_target = self.sync_target
         # get target identifier, its current generation,
         # and its last-seen database generation for this source
-        (self.target_replica_uid, target_gen, target_my_gen,
+        (self.target_replica_uid, target_gen, target_trans_id, target_my_gen,
          target_my_trans_id) = sync_target.get_sync_info(
              self.source._replica_uid)
-        # validate that the generation and transaction id the target knows
-        # about us are valid.
+        # validate the generation and transaction id the target knows about us
         self.source.validate_gen_and_trans_id(
             target_my_gen, target_my_trans_id)
         # what's changed since that generation and this current gen
         my_gen, _, changes = self.source.whats_changed(target_my_gen)
 
         # this source last-seen database generation for the target
-        (target_last_known_gen,
-         target_trans_id) = self.source._get_replica_gen_and_trans_id(
-             self.target_replica_uid)
+        target_last_known_gen, target_last_known_trans_id = \
+            self.source._get_replica_gen_and_trans_id(self.target_replica_uid)
         if not changes and target_last_known_gen == target_gen:
+            if target_trans_id != target_last_known_trans_id:
+                raise errors.InvalidTransactionId
             return my_gen
         changed_doc_ids = [doc_id for doc_id, _, _ in changes]
         # prepare to send all the changed docs
@@ -123,7 +124,7 @@
         # the target, return target synced-up-to gen
         new_gen, new_trans_id = sync_target.sync_exchange(
             docs_by_generation, self.source._replica_uid,
-            target_last_known_gen, target_trans_id,
+            target_last_known_gen, target_last_known_trans_id,
             self._insert_doc_from_target)
         # record target synced-up-to generation including applying what we sent
         self.source._set_replica_gen_and_trans_id(

=== modified file 'u1db/tests/c_backend_wrapper.pyx'
--- u1db/tests/c_backend_wrapper.pyx	2012-07-03 15:48:59 +0000
+++ u1db/tests/c_backend_wrapper.pyx	2012-07-06 20:51:26 +0000
@@ -81,9 +81,7 @@
                           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 replica_gen, const_char_ptr replica_trans_id)
     int u1db__put_doc_if_newer(u1database *db, u1db_document *doc,
                                int save_conflict, char *replica_uid,
                                int replica_gen, char *replica_trans_id,
@@ -175,10 +173,10 @@
 
     ctypedef int (*u1db__trace_callback)(void *context, const_char_ptr state)
     ctypedef struct u1db_sync_target:
-        int (*get_sync_info)(u1db_sync_target *st,
-            char *source_replica_uid,
-            const_char_ptr *st_replica_uid, int *st_gen, int *source_gen,
-            char **source_trans_id) nogil
+        int (*get_sync_info)(u1db_sync_target *st, char *source_replica_uid,
+                             const_char_ptr *st_replica_uid, int *st_gen,
+                             char **st_trans_id, int *source_gen,
+                             char **source_trans_id) nogil
         int (*record_sync_info)(u1db_sync_target *st,
             char *source_replica_uid, int source_gen, char *trans_id) nogil
         int (*sync_exchange)(u1db_sync_target *st,
@@ -206,6 +204,7 @@
 
     int u1db__get_generation(u1database *, int *db_rev)
     int u1db__get_generation_info(u1database *, int *db_rev, char **trans_id)
+    int u1db__get_trans_id_for_gen(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 *)
@@ -698,18 +697,25 @@
         cdef const_char_ptr st_replica_uid = NULL
         cdef int st_gen = 0, source_gen = 0, status
         cdef char *trans_id = NULL
+        cdef char *st_trans_id = NULL
 
         self._check()
         assert self._st.get_sync_info != NULL, "get_sync_info is NULL?"
         with nogil:
             status = self._st.get_sync_info(self._st, source_replica_uid,
-                &st_replica_uid, &st_gen, &source_gen, &trans_id)
+                &st_replica_uid, &st_gen, &st_trans_id, &source_gen, &trans_id)
         handle_status("get_sync_info", status)
         res_trans_id = None
+        res_st_trans_id = None
         if trans_id != NULL:
             res_trans_id = trans_id
             free(trans_id)
-        return (safe_str(st_replica_uid), st_gen, source_gen, res_trans_id)
+        if st_trans_id != NULL:
+            res_st_trans_id = st_trans_id
+            free(st_trans_id)
+        return (
+            safe_str(st_replica_uid), st_gen, res_st_trans_id, source_gen,
+            res_trans_id)
 
     def record_sync_info(self, source_replica_uid, source_gen, source_trans_id):
         cdef int status
@@ -941,26 +947,16 @@
             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):
+    def _validate_source(self, replica_uid, replica_gen, replica_trans_id):
         cdef const_char_ptr c_uid, c_trans_id
-        cdef int c_gen, state = 0
-        cdef VectorClockRev cur
-        cdef VectorClockRev other
+        cdef int c_gen = 0
 
-        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'
+            u1db__validate_source(self._db, c_uid, c_gen, c_trans_id))
 
     def _put_doc_if_newer(self, CDocument doc, save_conflict, replica_uid=None,
                           replica_gen=None, replica_trans_id=None):
@@ -1100,6 +1096,18 @@
             "validate_gen_and_trans_id",
             u1db_validate_gen_and_trans_id(self._db, generation, trans_id))
 
+    def _get_trans_id_for_gen(self, generation):
+        cdef char *trans_id = NULL
+
+        handle_status(
+            "_get_trans_id_for_gen",
+            u1db__get_trans_id_for_gen(self._db, generation, &trans_id))
+        raw_trans_id = None
+        if trans_id != NULL:
+            raw_trans_id = trans_id
+            free(trans_id)
+        return raw_trans_id
+
     def _get_replica_gen_and_trans_id(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-07-04 15:43:00 +0000
+++ u1db/tests/test_backends.py	2012-07-06 20:51:26 +0000
@@ -402,11 +402,12 @@
 
     def test_put_doc_if_newer_same_generation_same_txid(self):
         self.db._set_replica_gen_and_trans_id('other', 1, 'T-sid')
-        doc = self.make_document('doc_id', 'other:2', simple_doc)
+        doc = self.db.create_doc(simple_doc)
+        doc2 = self.make_document(doc.doc_id, 'other:1', 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)
+        self.assertEqual('converged', state)
 
     def test_put_doc_if_newer_wrong_transaction_id(self):
         self.db._set_replica_gen_and_trans_id('other', 1, 'T-sid')
@@ -422,10 +423,10 @@
         doc_rev1 = doc.rev
         doc.set_json(simple_doc)
         self.db.put_doc(doc)
-        self.db._set_replica_gen_and_trans_id('other', 5, 'T-sid')
+        self.db._set_replica_gen_and_trans_id('other', 3, '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,
+            older_doc, save_conflict=False, replica_uid='other', replica_gen=8,
             replica_trans_id='T-irrelevant')
         self.assertEqual('superseded', state)
 
@@ -557,45 +558,17 @@
 
     def test_validate_source_gen_and_trans_id_same(self):
         self.db._set_replica_gen_and_trans_id('other', 1, 'T-sid')
-        v1 = vectorclock.VectorClockRev('other:1|self:1')
-        v2 = vectorclock.VectorClockRev('other:1|self:1')
-        self.assertEqual(
-            'superseded',
-            self.db._validate_source('other', 1, 'T-sid', v1, v2))
+        self.db._validate_source('other', 1, 'T-sid')
 
     def test_validate_source_gen_newer(self):
         self.db._set_replica_gen_and_trans_id('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))
+        self.db._validate_source('other', 2, 'T-whatevs')
 
     def test_validate_source_wrong_txid(self):
         self.db._set_replica_gen_and_trans_id('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_replica_gen_and_trans_id('other', 1, 'T-sid')
-        self.db._set_replica_gen_and_trans_id('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_replica_gen_and_trans_id('other', 1, 'T-sid')
-        self.db._set_replica_gen_and_trans_id('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)
+            self.db._validate_source, 'other', 1, 'T-sad')
 
 
 class LocalDatabaseWithConflictsTests(tests.DatabaseBaseTests):

=== modified file 'u1db/tests/test_c_backend.py'
--- u1db/tests/test_c_backend.py	2012-07-03 15:48:59 +0000
+++ u1db/tests/test_c_backend.py	2012-07-06 20:51:26 +0000
@@ -75,7 +75,7 @@
 
     def test__get_generation_info(self):
         db = c_backend_wrapper.CDatabase(':memory:')
-        self.assertEqual((0, None), db._get_generation_info())
+        self.assertEqual((0, ''), db._get_generation_info())
         db.create_doc(tests.simple_doc)
         info = db._get_generation_info()
         self.assertEqual(1, info[0])

=== modified file 'u1db/tests/test_http_app.py'
--- u1db/tests/test_http_app.py	2012-07-03 15:48:59 +0000
+++ u1db/tests/test_http_app.py	2012-07-06 20:51:26 +0000
@@ -697,6 +697,7 @@
         self.assertEqual('application/json', resp.header('content-type'))
         self.assertEqual(dict(target_replica_uid='db0',
                               target_replica_generation=0,
+                              target_replica_transaction_id='',
                               source_replica_uid='other-id',
                               source_replica_generation=1,
                               source_transaction_id='T-transid'),
@@ -722,7 +723,8 @@
             }
 
         gens = []
-        _do_set_replica_gen_and_trans_id = self.db0._do_set_replica_gen_and_trans_id
+        _do_set_replica_gen_and_trans_id = \
+            self.db0._do_set_replica_gen_and_trans_id
 
         def set_sync_generation_witness(other_uid, other_gen, other_trans_id):
             gens.append((other_uid, other_gen))

=== modified file 'u1db/tests/test_remote_sync_target.py'
--- u1db/tests/test_remote_sync_target.py	2012-07-03 15:48:59 +0000
+++ u1db/tests/test_remote_sync_target.py	2012-07-06 20:51:26 +0000
@@ -173,7 +173,7 @@
         db = self.request_state._create_database('test')
         db._set_replica_gen_and_trans_id('other-id', 1, 'T-transid')
         remote_target = self.getSyncTarget('test')
-        self.assertEqual(('test', 0, 1, 'T-transid'),
+        self.assertEqual(('test', 0, '', 1, 'T-transid'),
                          remote_target.get_sync_info('other-id'))
 
     def test_record_sync_info(self):

=== modified file 'u1db/tests/test_sqlite_backend.py'
--- u1db/tests/test_sqlite_backend.py	2012-06-15 19:41:07 +0000
+++ u1db/tests/test_sqlite_backend.py	2012-07-06 20:51:26 +0000
@@ -153,7 +153,7 @@
         self.assertEqual(0, self.db._get_generation())
 
     def test__get_generation_info(self):
-        self.assertEqual((0, None), self.db._get_generation_info())
+        self.assertEqual((0, ''), self.db._get_generation_info())
 
     def test_create_index(self):
         self.db.create_index('test-idx', "key")

=== modified file 'u1db/tests/test_sync.py'
--- u1db/tests/test_sync.py	2012-07-04 16:03:39 +0000
+++ u1db/tests/test_sync.py	2012-07-06 20:51:26 +0000
@@ -143,18 +143,19 @@
         self.assertIsNot(None, self.st)
 
     def test_get_sync_info(self):
-        self.assertEqual(('test', 0, 0, ''), self.st.get_sync_info('other'))
+        self.assertEqual(
+            ('test', 0, '', 0, ''), self.st.get_sync_info('other'))
 
     def test_create_doc_updates_sync_info(self):
-        self.assertEqual(('test', 0, 0, ''), self.st.get_sync_info('other'))
+        self.assertEqual(
+            ('test', 0, '', 0, ''), self.st.get_sync_info('other'))
         self.db.create_doc(simple_doc)
-        self.assertEqual(('test', 1, 0, ''), self.st.get_sync_info('other'))
+        self.assertEqual(1, self.st.get_sync_info('other')[1])
 
     def test_record_sync_info(self):
-        self.assertEqual(('test', 0, 0, ''), self.st.get_sync_info('replica'))
         self.st.record_sync_info('replica', 10, 'T-transid')
-        self.assertEqual(('test', 0, 10, 'T-transid'),
-                         self.st.get_sync_info('replica'))
+        self.assertEqual(
+            ('test', 0, '', 10, 'T-transid'), self.st.get_sync_info('replica'))
 
     def test_sync_exchange(self):
         docs_by_gen = [
@@ -168,7 +169,7 @@
         last_trans_id = self.getLastTransId(self.db)
         self.assertEqual(([], 1, last_trans_id),
                          (self.other_changes, new_gen, last_trans_id))
-        self.assertEqual(10, self.st.get_sync_info('replica')[2])
+        self.assertEqual(10, self.st.get_sync_info('replica')[3])
 
     def test_sync_exchange_deleted(self):
         doc = self.db.create_doc('{}')
@@ -184,7 +185,7 @@
         last_trans_id = self.getLastTransId(self.db)
         self.assertEqual(([], 2, last_trans_id),
                          (self.other_changes, new_gen, trans_id))
-        self.assertEqual(10, self.st.get_sync_info('replica')[2])
+        self.assertEqual(10, self.st.get_sync_info('replica')[3])
 
     def test_sync_exchange_push_many(self):
         docs_by_gen = [
@@ -200,7 +201,7 @@
         last_trans_id = self.getLastTransId(self.db)
         self.assertEqual(([], 2, last_trans_id),
                          (self.other_changes, new_gen, trans_id))
-        self.assertEqual(11, self.st.get_sync_info('replica')[2])
+        self.assertEqual(11, self.st.get_sync_info('replica')[3])
 
     def test_sync_exchange_refuses_conflicts(self):
         doc = self.db.create_doc(simple_doc)
@@ -378,7 +379,7 @@
         last_trans_id = self.getLastTransId(self.db)
         self.assertEqual(([], 1, last_trans_id),
                          (self.other_changes, new_gen, trans_id))
-        self.assertEqual(10, self.st.get_sync_info(db2._replica_uid)[2])
+        self.assertEqual(10, self.st.get_sync_info(db2._replica_uid)[3])
 
     def test__set_trace_hook(self):
         called = []
@@ -424,7 +425,8 @@
     return db
 
 
-def sync_via_synchronizer_and_http(test, db_source, db_target, trace_hook=None):
+def sync_via_synchronizer_and_http(test, db_source, db_target,
+                                   trace_hook=None):
     if trace_hook:
         test.skipTest("trace_hook unsupported over http")
     path = test._http_at[db_target]
@@ -988,6 +990,24 @@
         self.assertRaises(
             errors.InvalidTransactionId, self.sync, self.db1, db3)
 
+    def test_sync_detects_rollback_and_divergence_in_source(self):
+        self.db1.create_doc(tests.simple_doc, doc_id="divergent")
+        self.sync(self.db1, self.db2)
+        self.db1.create_doc(tests.simple_doc)
+        self.db2._set_replica_gen_and_trans_id(
+            self.db1._replica_uid, 2, 'T-madeup')
+        self.assertRaises(
+            errors.InvalidTransactionId, self.sync, self.db1, self.db2)
+
+    def test_sync_detects_rollback_and_divergence_in_target(self):
+        self.db1.create_doc(tests.simple_doc, doc_id="divergent")
+        self.sync(self.db1, self.db2)
+        self.db2.create_doc(tests.simple_doc)
+        self.db1._set_replica_gen_and_trans_id(
+            self.db2._replica_uid, 2, 'T-madeup')
+        self.assertRaises(
+            errors.InvalidTransactionId, self.sync, self.db1, self.db2)
+
 
 class TestDbSync(tests.TestCaseWithServer):
     """Test db.sync remote sync shortcut"""
@@ -1032,7 +1052,8 @@
         self.assertEqual(2, len(self.db2._get_transaction_log()))
         progress1 = []
         progress2 = []
-        _do_set_replica_gen_and_trans_id = self.db1._do_set_replica_gen_and_trans_id
+        _do_set_replica_gen_and_trans_id = \
+            self.db1._do_set_replica_gen_and_trans_id
 
         def set_sync_generation_witness1(other_uid, other_gen, trans_id):
             progress1.append((other_uid, other_gen,
@@ -1040,8 +1061,8 @@
             _do_set_replica_gen_and_trans_id(other_uid, other_gen, trans_id)
         self.patch(self.db1, '_do_set_replica_gen_and_trans_id',
                    set_sync_generation_witness1)
-
-        _do_set_replica_gen_and_trans_id2 = self.db2._do_set_replica_gen_and_trans_id
+        _do_set_replica_gen_and_trans_id2 = \
+            self.db2._do_set_replica_gen_and_trans_id
 
         def set_sync_generation_witness2(other_uid, other_gen, trans_id):
             progress2.append((other_uid, other_gen,