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

[Eric Casteleijn]

Added integration tests for the detection of a diverged source or target db during sync and made them pass.

Added integration tests for the detection of a diverged source or target db during sync and made them pass.

[Eric Casteleijn (thisfred)]

shoot, something's still not right with sync_exchange_doc_ids

[Eric Casteleijn (thisfred)]

fixed

[John O'Brien (jdobrien)]

I'm completely unqualified to mark this approved. but the changes look sound to me and I really like the formatting cleanup.

[John A Meinel (jameinel)]

-----BEGIN PGP SIGNED MESSAGE-----
Hash: SHA1

On 6/20/2012 9:27 PM, Eric Casteleijn wrote:
> Eric Casteleijn has proposed merging
> lp:~thisfred/u1db/sync_integration_tests into lp:u1db.
>
> Requested reviews: Ubuntu One hackers (ubuntuone-hackers) Related
> bugs: Bug #1006872 in U1DB: "sync_exchange should transmit the txid
> it thinks the target was at"
> https://bugs.launchpad.net/u1db/+bug/1006872
>
> For more details, see:
> https://code.launchpad.net/~thisfred/u1db/sync_integration_tests/+merge/111278
>
> Added integration tests for the detection of a diverged source or
> target db during sync and made them pass.
>

+ def test_sync_detects_rollback_in_source(self):
+ self.db1.create_doc(tests.simple_doc, doc_id="divergent")
+ self.sync(self.db1, self.db2)
+ # make db2 think it's synced with a much later version of db1
+ self.db2._set_sync_info(self.db1._replica_uid, 28, 'T-madeup')
+ self.assertRaises(
+ errors.InvalidGeneration, self.sync, self.db1, self.db2)

^- I think this is something that we eventually wanted to support, as
long as there wasn't a local change on source.

Otherwise, the diverged tests are certainly valid.

John
=:->
-----BEGIN PGP SIGNATURE-----
Version: GnuPG v1.4.9 (Cygwin)
Comment: Using GnuPG with Mozilla - http://enigmail.mozdev.org/

iEYEARECAAYFAk/kIZsACgkQJdeBCYSNAAPCqQCgy7rzs2uAYMKIZMGKzVyiHrcZ
O7kAn1apHU4pnrBBhC0vNqWysVe+ny7g
=6nPi
-----END PGP SIGNATURE-----

[Eric Casteleijn (thisfred)]

I think we should as well, but we need more information to do it: all this says is that the target knows a newer generation than the source knows of itself. That still doesn't mean it hasn't diverged. I believe we probably can just start the sync though, and put_doc_if_newer would complain at the point of divergence if any. What I need to think about though is how to determine which generation to start the sync from source to target at. Can we do better than 0 there?

[Samuele Pedroni (pedronis)]

as discussed _validate_source isn't triggered:

https://pastebin.canonical.com/69190/

[Eric Casteleijn (thisfred)]

Ah, no: The code is triggered (just put a print in there) but all it does is shortcut the put_doc_if_newer, because what the target has is already strictly newer (this is the case where _validate_source returns 'superseded' rather than ok.) In the case where something is really wrong we raise an InvalidTransactionId, and that code is also exercised by the tests.

I'll try and change the relevant tests so that they detect that the put_doc_if_newer is broken off early, although I'm not sure that it's important, since it should not have any effect when it isn't (and indeed doesn't seem to, as your commenting out the check shows.)

[lp:~thisfred/u1db/sync_integration_tests]

342. By Eric Casteleijn on 2012-07-03

merged trunk, and fixed conflicts

343. By Eric Casteleijn on 2012-07-04

ignore

344. By Eric Casteleijn on 2012-07-04

fixed errors

345. By Eric Casteleijn on 2012-07-04

fixed errors

346. By Eric Casteleijn on 2012-07-04

merged lp:~pedronis/u1db/sync_integration_tests

[Samuele Pedroni (pedronis)]

review: Approve

=== modified file '.bzrignore'
--- .bzrignore	2012-06-29 17:00:49 +0000
+++ .bzrignore	2012-07-04 16:16:18 +0000
@@ -15,3 +15,9 @@
 src/CMakeFiles
 src/Makefile
 src/cmake_install.cmake
+CMakeCache.txt
+CMakeFiles
+CPackConfig.cmake
+CPackSourceConfig.cmake
+Makefile
+cmake_install.cmake

=== modified file 'include/u1db/u1db_internal.h'
--- include/u1db/u1db_internal.h	2012-07-03 13:50:27 +0000
+++ include/u1db/u1db_internal.h	2012-07-04 16:16:18 +0000
@@ -127,8 +127,8 @@
                          const char *source_replica_uid, int n_docs,
                          u1db_document **docs, int *generations,
                          const char **trans_ids, int *target_gen,
-                         char **target_trans_id,
-                         void *context, u1db_doc_gen_callback cb);
+                         char **target_trans_id, void *context,
+                         u1db_doc_gen_callback cb);
     /**
      * Create a sync_exchange state object.
      *

=== modified file 'src/u1db.c'
--- src/u1db.c	2012-07-03 13:50:27 +0000
+++ src/u1db.c	2012-07-04 16:16:18 +0000
@@ -1618,7 +1618,7 @@
 {
     int status;
     sqlite3_stmt *statement;
-    const char *tmp;
+    const char *tmp = NULL;
 
     if (db == NULL || replica_uid == NULL || generation == NULL
         || trans_id == NULL)

=== modified file 'src/u1db_http_sync_target.c'
--- src/u1db_http_sync_target.c	2012-06-27 18:56:17 +0000
+++ src/u1db_http_sync_target.c	2012-07-04 16:16:18 +0000
@@ -773,7 +773,8 @@
 
 
 static int
-init_temp_file(char tmpname[], FILE **temp_fd, int target_gen)
+init_temp_file(char tmpname[], FILE **temp_fd, int target_gen,
+               char *target_trans_id)
 {
     int status = U1DB_OK;
     *temp_fd = make_tempfile(tmpname);
@@ -786,12 +787,14 @@
     }
     // Spool all of the documents to a temporary file, so that it we can
     // determine Content-Length before we start uploading the data.
-    fprintf(*temp_fd, "[\r\n{\"last_known_generation\": %d}", target_gen);
+    fprintf(
+        *temp_fd,
+        "[\r\n{\"last_known_generation\": %d, \"last_known_trans_id\": \"%s\"}",
+        target_gen, target_trans_id);
 finish:
     return status;
 }
 
-
 static int
 finalize_and_send_temp_file(u1db_sync_target *st, FILE *temp_fd,
                             const char *source_replica_uid,
@@ -981,7 +984,7 @@
     if (n_docs > 0 && (docs == NULL || generations == NULL)) {
         return U1DB_INVALID_PARAMETER;
     }
-    status = init_temp_file(tmpname, &temp_fd, *target_gen);
+    status = init_temp_file(tmpname, &temp_fd, *target_gen, *target_trans_id);
     if (status != U1DB_OK) { goto finish; }
     for (i = 0; i < n_docs; ++i) {
         status = doc_to_tempfile(
@@ -1054,7 +1057,7 @@
     }
     status = u1db_get_replica_uid(source_db, &source_replica_uid);
     if (status != U1DB_OK) { goto finish; }
-    status = init_temp_file(tmpname, &temp_fd, *target_gen);
+    status = init_temp_file(tmpname, &temp_fd, *target_gen, *target_trans_id);
     if (status != U1DB_OK) { goto finish; }
     state.num = n_doc_ids;
     state.generations = generations;

=== modified file 'src/u1db_sync_target.c'
--- src/u1db_sync_target.c	2012-07-03 13:50:27 +0000
+++ src/u1db_sync_target.c	2012-07-04 16:16:18 +0000
@@ -30,11 +30,11 @@
         const char *source_replica_uid, int source_gen, const char *trans_id);
 
 static int st_sync_exchange(u1db_sync_target *st,
-                          const char *source_replica_uid, int n_docs,
-                          u1db_document **docs, int *generations,
-                          const char **trans_ids, int *target_gen,
-                          char **target_trans_id, void *context,
-                          u1db_doc_gen_callback cb);
+                            const char *source_replica_uid, int n_docs,
+                            u1db_document **docs, int *generations,
+                            const char **trans_ids, int *target_gen,
+                            char **target_trans_id, void *context,
+                            u1db_doc_gen_callback cb);
 static int st_sync_exchange_doc_ids(u1db_sync_target *st,
                                     u1database *source_db, int n_doc_ids,
                                     const char **doc_ids, int *generations,
@@ -370,7 +370,6 @@
 {
     int status;
     struct _whats_changed_doc_ids_state state = {0};
-    char *target_trans_id = NULL;
     if (se == NULL) {
         return U1DB_INVALID_PARAMETER;
     }
@@ -396,9 +395,6 @@
     se->gen_for_doc_ids = state.gen_for_doc_ids;
     se->trans_ids_for_doc_ids = state.trans_ids_for_doc_ids;
 finish:
-    if (target_trans_id != NULL) {
-        free(target_trans_id);
-    }
     return status;
 }
 
@@ -522,6 +518,9 @@
     status = st->get_sync_exchange(st, source_replica_uid,
                                    *target_gen, &exchange);
     if (status != U1DB_OK) { goto finish; }
+    status = u1db_validate_gen_and_trans_id(
+        exchange->db, *target_gen, *target_trans_id);
+    if (status != U1DB_OK) { goto finish; }
     for (i = 0; i < n_docs; ++i) {
         status = u1db__sync_exchange_insert_doc_from_source(
             exchange, docs[i], generations[i], trans_ids[i]);
@@ -530,14 +529,11 @@
     status = u1db__sync_exchange_find_doc_ids_to_return(exchange);
     if (status != U1DB_OK) { goto finish; }
     status = u1db__sync_exchange_return_docs(exchange, context, cb);
-    if (status != U1DB_OK) { goto finish; }
-    if (status == U1DB_OK) {
-        *target_gen = exchange->target_gen;
-        *target_trans_id = exchange->target_trans_id;
-        // We set this to NULL, because the caller is now responsible for it
-        exchange->target_trans_id = NULL;
-    }
 finish:
+    *target_gen = exchange->target_gen;
+    *target_trans_id = exchange->target_trans_id;
+    // We set this to NULL, because the caller is now responsible for it
+    exchange->target_trans_id = NULL;
     st->finalize_sync_exchange(st, &exchange);
     return status;
 }
@@ -565,6 +561,9 @@
     status = st->get_sync_exchange(st, source_replica_uid,
                                    *target_gen, &exchange);
     if (status != U1DB_OK) { goto finish; }
+    status = u1db_validate_gen_and_trans_id(
+        exchange->db, *target_gen, *target_trans_id);
+    if (status != U1DB_OK) { goto finish; }
     if (n_doc_ids > 0) {
         status = get_and_insert_docs(source_db, exchange,
             n_doc_ids, doc_ids, generations, trans_ids);
@@ -573,15 +572,11 @@
     status = u1db__sync_exchange_find_doc_ids_to_return(exchange);
     if (status != U1DB_OK) { goto finish; }
     status = u1db__sync_exchange_return_docs(exchange, context, cb);
-    if (status != U1DB_OK) { goto finish; }
+finish:
     *target_gen = exchange->target_gen;
-    if (status == U1DB_OK) {
-        *target_gen = exchange->target_gen;
-        *target_trans_id = exchange->target_trans_id;
-        // We set this to NULL, because the caller is now responsible for it
-        exchange->target_trans_id = NULL;
-    }
-finish:
+    *target_trans_id = exchange->target_trans_id;
+    // We set this to NULL, because the caller is now responsible for it
+    exchange->target_trans_id = NULL;
     st->finalize_sync_exchange(st, &exchange);
     return status;
 }
@@ -612,11 +607,12 @@
     status = u1db_get_replica_uid(db, &local_uid);
     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,
-                   &local_gen_known_by_target, &local_trans_id_known_by_target);
+    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);
+        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);

=== modified file 'u1db/__init__.py'
--- u1db/__init__.py	2012-07-03 14:35:42 +0000
+++ u1db/__init__.py	2012-07-04 16:16:18 +0000
@@ -565,7 +565,8 @@
         raise NotImplementedError(self.record_sync_info)
 
     def sync_exchange(self, docs_by_generation, source_replica_uid,
-                      last_known_generation, return_doc_cb):
+                      last_known_generation, last_known_trans_id,
+                      return_doc_cb):
         """Incorporate the documents sent from the source replica.
 
         This is not meant to be called by client code directly, but is used as
@@ -589,7 +590,9 @@
             id of their latest change.
         :param source_replica_uid: The source replica's identifier
         :param last_known_generation: The last generation that the source
-            replica knows about this
+            replica knows about this target replica
+        :param last_known_trans_id: The last transaction id that the source
+            replica knows about this target replica
         :param: return_doc_cb(doc, gen): is a callback
                 used to return documents to the source replica, it will
                 be invoked in turn with Documents that have changed since

=== modified file 'u1db/backends/inmemory.py'
--- u1db/backends/inmemory.py	2012-07-03 13:50:27 +0000
+++ u1db/backends/inmemory.py	2012-07-04 16:16:18 +0000
@@ -46,6 +46,10 @@
         self._last_exchange_log = None
         self._factory = document_factory or Document
 
+    def _set_replica_uid(self, replica_uid):
+        """Force the replica_uid to be set."""
+        self._replica_uid = replica_uid
+
     def set_document_factory(self, factory):
         self._factory = factory
 

=== modified file 'u1db/errors.py'
--- u1db/errors.py	2012-06-27 18:56:17 +0000
+++ u1db/errors.py	2012-07-04 16:16:18 +0000
@@ -46,10 +46,14 @@
 class InvalidTransactionId(U1DBError):
     """Invalid transaction for generation."""
 
+    wire_description = "invalid transaction id"
+
 
 class InvalidGeneration(U1DBError):
     """Generation was previously synced with a different transaction id."""
 
+    wire_description = "invalid generation"
+
 
 class ConflictedDoc(U1DBError):
     """The document is conflicted, you must call resolve before put()"""

=== modified file 'u1db/remote/http_app.py'
--- u1db/remote/http_app.py	2012-06-27 18:35:52 +0000
+++ u1db/remote/http_app.py	2012-07-04 16:16:18 +0000
@@ -45,6 +45,12 @@
     return False
 
 
+def none_or_str(expression):
+    if expression is None:
+        return None
+    return str(expression)
+
+
 class BadRequest(Exception):
     """Bad request."""
 
@@ -316,12 +322,13 @@
 
     # Implements the same logic as LocalSyncTarget.sync_exchange
 
-    @http_method(last_known_generation=int,
+    @http_method(last_known_generation=int, last_known_trans_id=none_or_str,
                  content_as_args=True)
-    def post_args(self, last_known_generation):
-        self.sync_exch = self.sync_exchange_class(self.db,
-                                                  self.source_replica_uid,
-                                                  last_known_generation)
+    def post_args(self, last_known_generation, last_known_trans_id=None):
+        self.db.validate_gen_and_trans_id(
+            last_known_generation, last_known_trans_id)
+        self.sync_exch = self.sync_exchange_class(
+            self.db, self.source_replica_uid, last_known_generation)
 
     @http_method(content_as_args=True)
     def post_stream_entry(self, id, rev, content, gen, trans_id):
@@ -329,17 +336,19 @@
         self.sync_exch.insert_doc_from_source(doc, gen, trans_id)
 
     def post_end(self):
+
         def send_doc(doc, gen, trans_id):
             entry = dict(id=doc.doc_id, rev=doc.rev, content=doc.get_json(),
                          gen=gen, trans_id=trans_id)
             self.responder.stream_entry(entry)
+
         new_gen = self.sync_exch.find_changes_to_return()
         self.responder.content_type = 'application/x-u1db-sync-stream'
         self.responder.start_response(200)
         self.responder.start_stream(),
         self.responder.stream_entry({"new_generation": new_gen,
                      "new_transaction_id": self.sync_exch.new_trans_id})
-        new_gen = self.sync_exch.return_docs(send_doc)
+        self.sync_exch.return_docs(send_doc)
         self.responder.end_stream()
         self.responder.finish_response()
 

=== modified file 'u1db/remote/http_errors.py'
--- u1db/remote/http_errors.py	2012-04-24 13:40:36 +0000
+++ u1db/remote/http_errors.py	2012-07-04 16:16:18 +0000
@@ -29,6 +29,8 @@
     (errors.DocumentDoesNotExist.wire_description, 404),
     (errors.DocumentAlreadyDeleted.wire_description, 404),
     (errors.RevisionConflict.wire_description, 409),
+    (errors.InvalidGeneration.wire_description, 409),
+    (errors.InvalidTransactionId.wire_description, 409),
     (errors.Unavailable.wire_description, 503),
 # without matching exception
     (errors.DOCUMENT_DELETED, 404)

=== modified file 'u1db/remote/http_target.py'
--- u1db/remote/http_target.py	2012-06-14 23:28:58 +0000
+++ u1db/remote/http_target.py	2012-07-04 16:16:18 +0000
@@ -80,7 +80,8 @@
         return res
 
     def sync_exchange(self, docs_by_generations, source_replica_uid,
-                      last_known_generation, return_doc_cb):
+                      last_known_generation, last_known_trans_id,
+                      return_doc_cb):
         self._ensure_connection()
         url = '%s/sync-from/%s' % (self._url.path, source_replica_uid)
         self._conn.putrequest('POST', url)
@@ -96,7 +97,9 @@
             return len(entry)
 
         comma = ''
-        size += prepare(last_known_generation=last_known_generation)
+        size += prepare(
+            last_known_generation=last_known_generation,
+            last_known_trans_id=last_known_trans_id)
         comma = ','
         for doc, gen, trans_id in docs_by_generations:
             size += prepare(id=doc.doc_id, rev=doc.rev, content=doc.get_json(),

=== modified file 'u1db/sync.py'
--- u1db/sync.py	2012-07-03 13:50:27 +0000
+++ u1db/sync.py	2012-07-04 16:16:18 +0000
@@ -97,9 +97,11 @@
         (self.target_replica_uid, target_gen, target_my_gen,
          target_my_trans_id) = sync_target.get_sync_info(
              self.source._replica_uid)
-        # what's changed since that generation and this current gen
+        # validate that the generation and transaction id the target knows
+        # about us are valid.
         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
@@ -119,9 +121,10 @@
 
         # exchange documents and try to insert the returned ones with
         # 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,
-                        return_doc_cb=self._insert_doc_from_target)
+        new_gen, new_trans_id = sync_target.sync_exchange(
+            docs_by_generation, self.source._replica_uid,
+            target_last_known_gen, target_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(
             self.target_replica_uid, new_gen, new_trans_id)
@@ -262,9 +265,12 @@
         self._trace_hook = None
 
     def sync_exchange(self, docs_by_generations, source_replica_uid,
-                      last_known_generation, return_doc_cb):
-        sync_exch = SyncExchange(self._db, source_replica_uid,
-                                 last_known_generation)
+                      last_known_generation, last_known_trans_id,
+                      return_doc_cb):
+        self._db.validate_gen_and_trans_id(
+            last_known_generation, last_known_trans_id)
+        sync_exch = SyncExchange(
+            self._db, source_replica_uid, last_known_generation)
         if self._trace_hook:
             sync_exch._set_trace_hook(self._trace_hook)
         # 1st step: try to insert incoming docs and record progress

=== modified file 'u1db/tests/c_backend_wrapper.pyx'
--- u1db/tests/c_backend_wrapper.pyx	2012-07-03 13:50:27 +0000
+++ u1db/tests/c_backend_wrapper.pyx	2012-07-04 16:16:18 +0000
@@ -184,9 +184,9 @@
         int (*sync_exchange)(u1db_sync_target *st,
                              char *source_replica_uid, int n_docs,
                              u1db_document **docs, int *generations,
-                             const_char_ptr *trans_ids, int *target_gen,
-                             char **target_trans_id, void *context,
-                             u1db_doc_gen_callback cb) nogil
+                             const_char_ptr *trans_ids,
+                             int *target_gen, char **target_trans_id,
+                             void *context, u1db_doc_gen_callback cb) nogil
         int (*sync_exchange_doc_ids)(u1db_sync_target *st,
                                      u1database *source_db, int n_doc_ids,
                                      const_char_ptr *doc_ids, int *generations,
@@ -713,6 +713,7 @@
 
     def record_sync_info(self, source_replica_uid, source_gen, source_trans_id):
         cdef int status
+
         self._check()
         assert self._st.record_sync_info != NULL, "record_sync_info is NULL?"
         with nogil:
@@ -725,12 +726,13 @@
         return CSyncExchange(self, source_replica_uid, source_gen)
 
     def sync_exchange_doc_ids(self, source_db, doc_id_generations,
-                              last_known_generation, return_doc_cb):
+                              last_known_generation, last_known_trans_id,
+                              return_doc_cb):
         cdef const_char_ptr *doc_ids
         cdef int *generations
         cdef int num_doc_ids
         cdef int target_gen
-        cdef char *target_trans_id
+        cdef char *target_trans_id = NULL
         cdef int status
         cdef CDatabase sdb
 
@@ -755,31 +757,35 @@
                 generations[i] = gen
                 trans_ids[i] = trans_id
             target_gen = last_known_generation
+            if last_known_trans_id is not None:
+                target_trans_id = last_known_trans_id
             with nogil:
                 status = self._st.sync_exchange_doc_ids(self._st, sdb._db,
                     num_doc_ids, doc_ids, generations, trans_ids,
                     &target_gen, &target_trans_id,
                     <void*>return_doc_cb, return_doc_cb_wrapper)
             handle_status("sync_exchange_doc_ids", status)
+            if target_trans_id != NULL:
+                res_trans_id = target_trans_id
         finally:
+            if target_trans_id != NULL:
+                free(target_trans_id)
             if doc_ids != NULL:
                 free(<void *>doc_ids)
             if generations != NULL:
                 free(generations)
             if trans_ids != NULL:
                 free(trans_ids)
-            if target_trans_id != NULL:
-                res_trans_id = target_trans_id
-                free(target_trans_id)
         return target_gen, res_trans_id
 
     def sync_exchange(self, docs_by_generations, source_replica_uid,
-                      last_known_generation, return_doc_cb):
+                      last_known_generation, last_known_trans_id,
+                      return_doc_cb):
         cdef CDocument cur_doc
         cdef u1db_document **docs = NULL
         cdef int *generations = NULL
         cdef const_char_ptr *trans_ids = NULL
-        cdef char *trans_id = NULL
+        cdef char *target_trans_id = NULL
         cdef int i, count, status, target_gen
 
         self._check()
@@ -802,11 +808,13 @@
                 trans_ids[i] = docs_by_generations[i][2]
                 docs[i] = cur_doc._doc
             target_gen = last_known_generation
+            if last_known_trans_id is not None:
+                target_trans_id = last_known_trans_id
             with nogil:
-                status = self._st.sync_exchange(self._st,
-                        source_replica_uid, count,
-                        docs, generations, trans_ids, &target_gen, &trans_id,
-                        <void *>return_doc_cb, return_doc_cb_wrapper)
+                status = self._st.sync_exchange(
+                    self._st, source_replica_uid, count, docs, generations,
+                    trans_ids, &target_gen, &target_trans_id,
+                    <void *>return_doc_cb, return_doc_cb_wrapper)
             handle_status("sync_exchange", status)
         finally:
             if docs != NULL:
@@ -815,9 +823,9 @@
                 free(generations)
             if trans_ids != NULL:
                 free(trans_ids)
-            if trans_id != NULL:
-                res_trans_id = trans_id
-                free(trans_id)
+            if target_trans_id != NULL:
+                res_trans_id = target_trans_id
+                free(target_trans_id)
         return target_gen, res_trans_id
 
     def _set_trace_hook(self, cb):

=== modified file 'u1db/tests/test_backends.py'
--- u1db/tests/test_backends.py	2012-07-03 22:24:43 +0000
+++ u1db/tests/test_backends.py	2012-07-04 16:16:18 +0000
@@ -1556,7 +1556,7 @@
         docs_by_gen = [(doc_other, 10, 'T-sid')]
         st.sync_exchange(
             docs_by_gen, 'other-replica', last_known_generation=0,
-            return_doc_cb=ignore)
+            last_known_trans_id=None, return_doc_cb=ignore)
         self.assertGetDoc(self.db, doc.doc_id, other_rev, new_content, False)
         self.assertEqual(
             [doc_other], self.db.get_from_index('test-idx', 'altval'))

=== modified file 'u1db/tests/test_c_backend.py'
--- u1db/tests/test_c_backend.py	2012-07-03 13:50:27 +0000
+++ u1db/tests/test_c_backend.py	2012-07-04 16:16:18 +0000
@@ -359,7 +359,7 @@
             returned.append((doc, gen, trans_id))
 
         val = self.st.sync_exchange_doc_ids(
-            db2, [(doc2.doc_id, 1, 'T-sid')], 0, return_doc_cb)
+            db2, [(doc2.doc_id, 1, 'T-sid')], 0, None, return_doc_cb)
         last_trans_id = self.db._get_transaction_log()[-1][1]
         self.assertEqual(2, self.db._get_generation())
         self.assertEqual((2, last_trans_id), val)

=== modified file 'u1db/tests/test_http_app.py'
--- u1db/tests/test_http_app.py	2012-07-03 13:50:27 +0000
+++ u1db/tests/test_http_app.py	2012-07-04 16:16:18 +0000
@@ -1016,5 +1016,6 @@
             sync_exchange_class = MySyncExchange
 
         sync_res = MySyncResource('foo', 'src', self.state, None)
-        sync_res.post_args({'last_known_generation': 0}, '{}')
+        sync_res.post_args(
+            {'last_known_generation': 0, 'last_known_trans_id': None}, '{}')
         self.assertIsInstance(sync_res.sync_exch, MySyncExchange)

=== modified file 'u1db/tests/test_remote_sync_target.py'
--- u1db/tests/test_remote_sync_target.py	2012-07-03 13:50:27 +0000
+++ u1db/tests/test_remote_sync_target.py	2012-07-04 16:16:18 +0000
@@ -195,9 +195,8 @@
 
         doc = self.make_document('doc-here', 'replica:1', '{"value": "here"}')
         new_gen, trans_id = remote_target.sync_exchange(
-                [(doc, 10, 'T-sid')],
-                'replica', last_known_generation=0,
-                return_doc_cb=receive_doc)
+            [(doc, 10, 'T-sid')], 'replica', last_known_generation=0,
+            last_known_trans_id=None, return_doc_cb=receive_doc)
         self.assertEqual(1, new_gen)
         self.assertGetDoc(
             db, 'doc-here', 'replica:1', '{"value": "here"}', False)
@@ -233,11 +232,12 @@
         doc1 = self.make_document('doc-here', 'replica:1', '{"value": "here"}')
         doc2 = self.make_document('doc-here2', 'replica:1',
                                   '{"value": "here2"}')
-        self.assertRaises(errors.HTTPError, remote_target.sync_exchange,
-                [(doc1, 10, 'T-sid'),
-                 (doc2, 11, 'T-sud')
-                 ], 'replica', last_known_generation=0,
-                return_doc_cb=receive_doc)
+        self.assertRaises(
+            errors.HTTPError,
+            remote_target.sync_exchange,
+            [(doc1, 10, 'T-sid'), (doc2, 11, 'T-sud')],
+            'replica', last_known_generation=0, last_known_trans_id=None,
+            return_doc_cb=receive_doc)
         self.assertGetDoc(db, 'doc-here', 'replica:1', '{"value": "here"}',
                           False)
         self.assertEqual(
@@ -246,9 +246,8 @@
         # retry
         trigger_ids = []
         new_gen, trans_id = remote_target.sync_exchange(
-                [(doc2, 11, 'T-sud')
-                 ], 'replica', last_known_generation=0,
-                return_doc_cb=receive_doc)
+            [(doc2, 11, 'T-sud')], 'replica', last_known_generation=0,
+            last_known_trans_id=None, return_doc_cb=receive_doc)
         self.assertGetDoc(db, 'doc-here2', 'replica:1', '{"value": "here2"}',
                           False)
         self.assertEqual(
@@ -284,9 +283,10 @@
             other_changes.append(
                 (doc.doc_id, doc.rev, doc.get_json(), gen, trans_id))
 
-        self.assertRaises(errors.Unavailable, remote_target.sync_exchange,
-                          [], 'replica', last_known_generation=0,
-                          return_doc_cb=receive_doc)
+        self.assertRaises(
+            errors.Unavailable, remote_target.sync_exchange, [], 'replica',
+            last_known_generation=0, last_known_trans_id=None,
+            return_doc_cb=receive_doc)
         self.assertEqual(
             (doc.doc_id, doc.rev, '{"value": "there"}', 1),
             other_changes[0][:-1])
@@ -303,8 +303,8 @@
                 (doc.doc_id, doc.rev, doc.get_json(), gen, trans_id))
 
         new_gen, trans_id = remote_target.sync_exchange(
-                        [], 'replica', last_known_generation=0,
-                        return_doc_cb=receive_doc)
+            [], 'replica', last_known_generation=0, last_known_trans_id=None,
+            return_doc_cb=receive_doc)
         self.assertEqual(1, new_gen)
         self.assertEqual(
             (doc.doc_id, doc.rev, '{"value": "there"}', 1),

=== modified file 'u1db/tests/test_sync.py'
--- u1db/tests/test_sync.py	2012-07-03 18:01:54 +0000
+++ u1db/tests/test_sync.py	2012-07-04 16:16:18 +0000
@@ -157,11 +157,12 @@
                          self.st.get_sync_info('replica'))
 
     def test_sync_exchange(self):
-        doc = self.make_document('doc-id', 'replica:1', simple_doc)
-        docs_by_gen = [(doc, 10, 'T-sid')]
-        new_gen, trans_id = self.st.sync_exchange(docs_by_gen, 'replica',
-                                        last_known_generation=0,
-                                        return_doc_cb=self.receive_doc)
+        docs_by_gen = [
+            (self.make_document('doc-id', 'replica:1', simple_doc), 10,
+             'T-sid')]
+        new_gen, trans_id = self.st.sync_exchange(
+            docs_by_gen, 'replica', last_known_generation=0,
+            last_known_trans_id=None, return_doc_cb=self.receive_doc)
         self.assertGetDoc(self.db, 'doc-id', 'replica:1', simple_doc, False)
         self.assertTransactionLog(['doc-id'], self.db)
         last_trans_id = self.getLastTransId(self.db)
@@ -174,9 +175,9 @@
         edit_rev = 'replica:1|' + doc.rev
         docs_by_gen = [
             (self.make_document(doc.doc_id, edit_rev, None), 10, 'T-sid')]
-        new_gen, trans_id = self.st.sync_exchange(docs_by_gen, 'replica',
-                                        last_known_generation=0,
-                                        return_doc_cb=self.receive_doc)
+        new_gen, trans_id = self.st.sync_exchange(
+            docs_by_gen, 'replica', last_known_generation=0,
+            last_known_trans_id=None, return_doc_cb=self.receive_doc)
         self.assertGetDocIncludeDeleted(
             self.db, doc.doc_id, edit_rev, None, False)
         self.assertTransactionLog([doc.doc_id, doc.doc_id], self.db)
@@ -190,9 +191,9 @@
             (self.make_document('doc-id', 'replica:1', simple_doc), 10, 'T-1'),
             (self.make_document('doc-id2', 'replica:1', nested_doc), 11,
              'T-2')]
-        new_gen, trans_id = self.st.sync_exchange(docs_by_gen, 'replica',
-                                        last_known_generation=0,
-                                        return_doc_cb=self.receive_doc)
+        new_gen, trans_id = self.st.sync_exchange(
+            docs_by_gen, 'replica', last_known_generation=0,
+            last_known_trans_id=None, return_doc_cb=self.receive_doc)
         self.assertGetDoc(self.db, 'doc-id', 'replica:1', simple_doc, False)
         self.assertGetDoc(self.db, 'doc-id2', 'replica:1', nested_doc, False)
         self.assertTransactionLog(['doc-id', 'doc-id2'], self.db)
@@ -210,7 +211,7 @@
              'T-sid')]
         new_gen, _ = self.st.sync_exchange(
             docs_by_gen, 'replica', last_known_generation=0,
-            return_doc_cb=self.receive_doc)
+            last_known_trans_id=None, return_doc_cb=self.receive_doc)
         self.assertTransactionLog([doc.doc_id], self.db)
         self.assertEqual(
             (doc.doc_id, doc.rev, simple_doc, 1), self.other_changes[0][:-1])
@@ -222,20 +223,21 @@
     def test_sync_exchange_ignores_convergence(self):
         doc = self.db.create_doc(simple_doc)
         self.assertTransactionLog([doc.doc_id], self.db)
+        gen, txid = self.db._get_generation_info()
         docs_by_gen = [
             (self.make_document(doc.doc_id, doc.rev, simple_doc), 10, 'T-sid')]
-        new_gen, _ = self.st.sync_exchange(docs_by_gen, 'replica',
-                                        last_known_generation=1,
-                                        return_doc_cb=self.receive_doc)
+        new_gen, _ = self.st.sync_exchange(
+            docs_by_gen, 'replica', last_known_generation=gen,
+            last_known_trans_id=txid, return_doc_cb=self.receive_doc)
         self.assertTransactionLog([doc.doc_id], self.db)
         self.assertEqual(([], 1), (self.other_changes, new_gen))
 
     def test_sync_exchange_returns_new_docs(self):
         doc = self.db.create_doc(simple_doc)
         self.assertTransactionLog([doc.doc_id], self.db)
-        new_gen, _ = self.st.sync_exchange([], 'other-replica',
-                                        last_known_generation=0,
-                                        return_doc_cb=self.receive_doc)
+        new_gen, _ = self.st.sync_exchange(
+            [], 'other-replica', last_known_generation=0,
+            last_known_trans_id=None, return_doc_cb=self.receive_doc)
         self.assertTransactionLog([doc.doc_id], self.db)
         self.assertEqual(
             (doc.doc_id, doc.rev, simple_doc, 1), self.other_changes[0][:-1])
@@ -248,9 +250,9 @@
         doc = self.db.create_doc(simple_doc)
         self.db.delete_doc(doc)
         self.assertTransactionLog([doc.doc_id, doc.doc_id], self.db)
-        new_gen, _ = self.st.sync_exchange([], 'other-replica',
-                                        last_known_generation=0,
-                                        return_doc_cb=self.receive_doc)
+        new_gen, _ = self.st.sync_exchange(
+            [], 'other-replica', last_known_generation=0,
+            last_known_trans_id=None, return_doc_cb=self.receive_doc)
         self.assertTransactionLog([doc.doc_id, doc.doc_id], self.db)
         self.assertEqual(
             (doc.doc_id, doc.rev, None, 2), self.other_changes[0][:-1])
@@ -263,9 +265,9 @@
         doc = self.db.create_doc(simple_doc)
         doc2 = self.db.create_doc(nested_doc)
         self.assertTransactionLog([doc.doc_id, doc2.doc_id], self.db)
-        new_gen, _ = self.st.sync_exchange([], 'other-replica',
-                                        last_known_generation=0,
-                                        return_doc_cb=self.receive_doc)
+        new_gen, _ = self.st.sync_exchange(
+            [], 'other-replica', last_known_generation=0,
+            last_known_trans_id=None, return_doc_cb=self.receive_doc)
         self.assertTransactionLog([doc.doc_id, doc2.doc_id], self.db)
         self.assertEqual(2, new_gen)
         self.assertEqual(
@@ -285,9 +287,9 @@
         docs_by_gen = [
             (self.make_document(doc.doc_id, 'test:1|z:2', new_doc), 10,
              'T-sid')]
-        new_gen, _ = self.st.sync_exchange(docs_by_gen, 'other-replica',
-                                        last_known_generation=0,
-                                        return_doc_cb=self.receive_doc)
+        new_gen, _ = self.st.sync_exchange(
+            docs_by_gen, 'other-replica', last_known_generation=0,
+            last_known_trans_id=None, return_doc_cb=self.receive_doc)
         self.assertTransactionLog([doc.doc_id, doc.doc_id], self.db)
         self.assertEqual(([], 2), (self.other_changes, new_gen))
 
@@ -311,15 +313,17 @@
              'T-sid')]
         new_gen, _ = self.st.sync_exchange(
             docs_by_gen, 'other-replica', last_known_generation=0,
-            return_doc_cb=self.receive_doc)
+            last_known_trans_id=None, return_doc_cb=self.receive_doc)
         self.assertEqual(expected, [c[:-1] for c in self.other_changes])
         self.assertEqual(3, new_gen)
 
     def test_sync_exchange_with_concurrent_updates(self):
+
         def after_whatschanged_cb(state):
             if state != 'after whats_changed':
                 return
             self.db.create_doc('{"new": "doc"}')
+
         self.set_trace_hook(after_whatschanged_cb)
         doc = self.db.create_doc(simple_doc)
         self.assertTransactionLog([doc.doc_id], self.db)
@@ -327,9 +331,9 @@
         docs_by_gen = [
             (self.make_document(doc.doc_id, 'test:1|z:2', new_doc), 10,
              'T-sid')]
-        new_gen, _ = self.st.sync_exchange(docs_by_gen, 'other-replica',
-                                        last_known_generation=0,
-                                        return_doc_cb=self.receive_doc)
+        new_gen, _ = self.st.sync_exchange(
+            docs_by_gen, 'other-replica', last_known_generation=0,
+            last_known_trans_id=None, return_doc_cb=self.receive_doc)
         self.assertEqual(([], 2), (self.other_changes, new_gen))
 
     def test_sync_exchange_converged_handling(self):
@@ -338,9 +342,9 @@
             (self.make_document('new', 'other:1', '{}'), 4, 'T-foo'),
             (self.make_document(doc.doc_id, doc.rev, doc.get_json()), 5,
              'T-bar')]
-        new_gen, _ = self.st.sync_exchange(docs_by_gen, 'other-replica',
-                                        last_known_generation=0,
-                                        return_doc_cb=self.receive_doc)
+        new_gen, _ = self.st.sync_exchange(
+            docs_by_gen, 'other-replica', last_known_generation=0,
+            last_known_trans_id=None, return_doc_cb=self.receive_doc)
         self.assertEqual(([], 2), (self.other_changes, new_gen))
 
     def test_sync_exchange_detect_incomplete_exchange(self):
@@ -354,10 +358,11 @@
                    'log_exception', lambda h, exc_info: None)
         doc = self.db.create_doc(simple_doc)
         self.assertTransactionLog([doc.doc_id], self.db)
-        self.assertRaises((errors.U1DBError, errors.BrokenSyncStream),
-                          self.st.sync_exchange, [], 'other-replica',
-                          last_known_generation=0,
-                          return_doc_cb=self.receive_doc)
+        self.assertRaises(
+            (errors.U1DBError, errors.BrokenSyncStream),
+            self.st.sync_exchange, [], 'other-replica',
+            last_known_generation=0, last_known_trans_id=None,
+            return_doc_cb=self.receive_doc)
 
     def test_sync_exchange_doc_ids(self):
         sync_exchange_doc_ids = getattr(self.st, 'sync_exchange_doc_ids', None)
@@ -366,7 +371,7 @@
         db2 = self.create_database('test2')
         doc = db2.create_doc(simple_doc)
         new_gen, trans_id = sync_exchange_doc_ids(
-            db2, [(doc.doc_id, 10, 'T-sid')], 0,
+            db2, [(doc.doc_id, 10, 'T-sid')], 0, None,
             return_doc_cb=self.receive_doc)
         self.assertGetDoc(self.db, doc.doc_id, doc.rev, simple_doc, False)
         self.assertTransactionLog([doc.doc_id], self.db)
@@ -382,7 +387,7 @@
             called.append(state)
 
         self.set_trace_hook(cb)
-        self.st.sync_exchange([], 'replica', 0, self.receive_doc)
+        self.st.sync_exchange([], 'replica', 0, None, self.receive_doc)
         self.st.record_sync_info('replica', 0, 'T-sid')
         self.assertEqual(['before whats_changed',
                           'after whats_changed',
@@ -410,13 +415,19 @@
 def make_database_for_http_test(test, replica_uid):
     if test.server is None:
         test.startServer()
-    return test.request_state._create_database(replica_uid)
+    db = test.request_state._create_database(replica_uid)
+    try:
+        http_at = test._http_at
+    except AttributeError:
+        http_at = test._http_at = {}
+    http_at[db] = replica_uid
+    return db
 
 
 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 = db_target._replica_uid
+    path = test._http_at[db_target]
     target = http_target.HTTPSyncTarget.connect(test.getURL(path))
     return sync.Synchronizer(db_source, target).sync()
 
@@ -937,6 +948,46 @@
 
         self.sync(self.db1, self.db2, trace_hook=put_hook)
 
+    def test_sync_detects_rollback_in_source(self):
+        self.db1.create_doc(tests.simple_doc, doc_id="divergent")
+        self.sync(self.db1, self.db2)
+        # make db2 think it's synced with a much later version of db1
+        self.db2._set_replica_gen_and_trans_id(
+            self.db1._replica_uid, 28, 'T-madeup')
+        self.assertRaises(
+            errors.InvalidGeneration, self.sync, self.db1, self.db2)
+
+    def test_sync_detects_rollback_in_target(self):
+        self.db1.create_doc(tests.simple_doc, doc_id="divergent")
+        self.sync(self.db1, self.db2)
+        # make db1 think it's synced with a much later version of db2
+        self.db1._set_replica_gen_and_trans_id(
+            self.db2._replica_uid, 28, 'T-madeup')
+        self.assertRaises(
+            errors.InvalidGeneration, self.sync, self.db1, self.db2)
+
+    def test_sync_detects_diverged_source(self):
+        self.db1.create_doc(tests.simple_doc, doc_id="divergent")
+        self.sync(self.db1, self.db2)
+        # create a new db with the same replica as the source, and add
+        # a document, so it will have a different txid for the same generation.
+        db3 = self.create_database('test3')
+        db3.create_doc(tests.nested_doc, doc_id="divergent")
+        db3._set_replica_uid(self.db1._replica_uid)
+        self.assertRaises(
+            errors.InvalidTransactionId, self.sync, db3, self.db2)
+
+    def test_sync_detects_diverged_target(self):
+        self.db1.create_doc(tests.simple_doc, doc_id="divergent")
+        self.sync(self.db1, self.db2)
+        # create a new db with the same replica as the target, and add
+        # a document, so it will have a different txid for the same generation.
+        db3 = self.create_database('test3')
+        db3.create_doc(tests.nested_doc, doc_id="divergent")
+        db3._set_replica_uid(self.db2._replica_uid)
+        self.assertRaises(
+            errors.InvalidTransactionId, self.sync, self.db1, db3)
+
 
 class TestDbSync(tests.TestCaseWithServer):
     """Test db.sync remote sync shortcut"""