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

[Eric Casteleijn]

Added the transaction ids to the sync_exchange

Added the transaction ids to the sync_exchange

[Eric Casteleijn (thisfred)]

Since we get a lot of uuid txids, I strip them in a lot of the tests. This means the tests will still fail if the txids are not there, but perhaps not in the most obvious way, and we make no assumptions about the contents of the txids, but I think that's probably a good thing.

Also I seem to have misplaced some 3.5K of memory, which I will hunt down and fix tomorrow.

[lp:~thisfred/u1db/txid_in_sync_exchange]

329. By Eric Casteleijn on 2012-06-14

merged trunk, fixed conflicts

330. By Eric Casteleijn on 2012-06-15

found and fixed memory leaks

[Eric Casteleijn (thisfred)]

Memory issue fixed.

[lp:~thisfred/u1db/txid_in_sync_exchange]

331. By Eric Casteleijn on 2012-06-19

fixed incorrect documentation

332. By Eric Casteleijn on 2012-06-19

merged trunk

333. By Eric Casteleijn on 2012-06-19

merged trunk

Unmerged revisions

=== modified file 'include/u1db/u1db.h'
--- include/u1db/u1db.h	2012-06-12 15:44:43 +0000
+++ include/u1db/u1db.h	2012-06-15 00:40:24 +0000
@@ -42,7 +42,8 @@
 typedef int (*u1db_doc_callback)(void *context, u1db_document *doc);
 typedef int (*u1db_key_callback)(void *context, int num_fields,
                                  const char **key);
-typedef int (*u1db_doc_gen_callback)(void *context, u1db_document *doc, int gen);
+typedef int (*u1db_doc_gen_callback)(void *context, u1db_document *doc,
+                                     int gen, const char *trans_id);
 typedef int (*u1db_doc_id_gen_callback)(void *context, const char *doc_id, int gen);
 typedef int (*u1db_trans_info_callback)(void *context, const char *doc_id,
                                         int gen, const char *trans_id);

=== modified file 'include/u1db/u1db_internal.h'
--- include/u1db/u1db_internal.h	2012-06-12 15:44:43 +0000
+++ include/u1db/u1db_internal.h	2012-06-15 00:40:24 +0000
@@ -98,6 +98,8 @@
      *                  want to send to the sync target
      * @param generations   An array of generations. Each generation
      *                      corresponds to a doc_id.
+     * @param trans_ids   An array of transaction ids. Each transaction id
+     *                      corresponds to a doc_id.
      * @param target_gen    (IN/OUT) Seed this with the generation of the
      *                      target that source_db has last seen, it will then
      *                      be filled with the final generation of the target
@@ -114,7 +116,7 @@
      */
     int (*sync_exchange_doc_ids)(u1db_sync_target *st, u1database *source_db,
             int n_doc_ids, const char **doc_ids, int *generations,
-            int *target_gen, char **target_trans_id,
+            const char **trans_ids, int *target_gen, char **target_trans_id,
             void *context, u1db_doc_gen_callback cb);
 
     /**
@@ -123,7 +125,8 @@
     int (*sync_exchange)(u1db_sync_target *st,
                          const char *source_replica_uid, int n_docs,
                          u1db_document **docs, int *generations,
-                         int *target_gen, char **target_trans_id,
+                         const char **trans_ids, int *target_gen,
+                         char **target_trans_id,
                          void *context, u1db_doc_gen_callback cb);
     /**
      * Create a sync_exchange state object.
@@ -170,6 +173,7 @@
     struct lh_table *seen_ids;
     int num_doc_ids;
     int *gen_for_doc_ids;
+    const char **trans_ids_for_doc_ids;
     char **doc_ids_to_return;
     void *trace_context;
     u1db__trace_callback trace_cb;
@@ -448,7 +452,7 @@
  * We have received a doc from source, record it.
  */
 int u1db__sync_exchange_insert_doc_from_source(u1db_sync_exchange *se,
-        u1db_document *doc, int source_gen);
+        u1db_document *doc, int source_gen, const char *trans_id);
 
 /**
  * We are done receiving docs, find what we want to return.
@@ -466,8 +470,9 @@
  *                  u1db_free_doc().
  */
 int u1db__sync_exchange_return_docs(u1db_sync_exchange *se, void *context,
-        int (*cb)(void *context, u1db_document *doc, int gen));
-
+                                    int (*cb)(void *context,
+                                    u1db_document *doc, int gen,
+                                    const char *trans_id));
 
 /**
  * Create a sync target pointing at a given URL.

=== modified file 'src/u1db_http_sync_target.c'
--- src/u1db_http_sync_target.c	2012-05-31 12:22:03 +0000
+++ src/u1db_http_sync_target.c	2012-06-15 00:40:24 +0000
@@ -45,14 +45,19 @@
                          int source_gen,
                          u1db_sync_exchange **exchange);
 static int st_http_sync_exchange(u1db_sync_target *st,
-                      const char *source_replica_uid,
-                      int n_docs, u1db_document **docs,
-                      int *generations, 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_http_sync_exchange_doc_ids(u1db_sync_target *st,
-        u1database *source_db, int n_doc_ids, const char **doc_ids,
-        int *generations, int *target_gen, char **target_trans_id,
-        void *context, u1db_doc_gen_callback cb);
+                                         u1database *source_db, int n_doc_ids,
+                                         const char **doc_ids,
+                                         int *generations,
+                                         const char **trans_ids,
+                                         int *target_gen,
+                                         char **target_trans_id, void *context,
+                                         u1db_doc_gen_callback cb);
 static void st_http_finalize_sync_exchange(u1db_sync_target *st,
                                u1db_sync_exchange **exchange);
 static int st_http_set_trace_hook(u1db_sync_target *st,
@@ -410,8 +415,8 @@
         goto finish;
     }
     if (req.body_buffer == NULL) {
-	status = U1DB_INVALID_HTTP_RESPONSE;
-	goto finish;
+        status = U1DB_INVALID_HTTP_RESPONSE;
+        goto finish;
     }
     json = json_tokener_parse(req.body_buffer);
     if (json == NULL) {
@@ -663,7 +668,7 @@
 
 
 static int
-doc_to_tempfile(u1db_document *doc, int gen, FILE *fd)
+doc_to_tempfile(u1db_document *doc, int gen, const char *trans_id, FILE *fd)
 {
     int status = U1DB_OK;
     json_object *json = NULL;
@@ -678,6 +683,7 @@
     json_object_object_add(
         json, "content", doc->json?json_object_new_string(doc->json):NULL);
     json_object_object_add(json, "gen", json_object_new_int(gen));
+    json_object_object_add(json, "trans_id", json_object_new_string(trans_id));
     fputs(json_object_to_json_string(json), fd);
 finish:
     if (json != NULL) {
@@ -791,6 +797,7 @@
     const char *doc_id, *content, *rev;
     const char *tmp = NULL;
     int gen;
+    const char *trans_id = NULL;
     u1db_document *doc;
 
     json = json_tokener_parse(response);
@@ -837,12 +844,14 @@
         content = json_object_get_string(attr);
         attr = json_object_object_get(obj, "gen");
         gen = json_object_get_int(attr);
+        attr = json_object_object_get(obj, "trans_id");
+        trans_id = json_object_get_string(attr);
         doc = u1db__allocate_document(doc_id, rev, content, 0);
         if (doc == NULL) {
             status = U1DB_NOMEM;
             goto finish;
         }
-        status = cb(context, doc, gen);
+        status = cb(context, doc, gen, trans_id);
         if (status != U1DB_OK) { goto finish; }
     }
 finish:
@@ -875,11 +884,10 @@
 }
 
 static int
-st_http_sync_exchange(u1db_sync_target *st,
-                      const char *source_replica_uid,
-                      int n_docs, u1db_document **docs,
-                      int *generations, int *target_gen, char **target_trans_id,
-                      void *context,
+st_http_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)
 {
     int status, i;
@@ -898,7 +906,8 @@
     status = init_temp_file(tmpname, &temp_fd, *target_gen);
     if (status != U1DB_OK) { goto finish; }
     for (i = 0; i < n_docs; ++i) {
-        status = doc_to_tempfile(docs[i], generations[i], temp_fd);
+        status = doc_to_tempfile(
+            docs[i], generations[i], trans_ids[i], temp_fd);
         if (status != U1DB_OK) { goto finish; }
     }
     status = finalize_and_send_temp_file(st, temp_fd, source_replica_uid, &req);
@@ -915,6 +924,7 @@
     int offset;
     int num;
     int *generations;
+    const char **trans_ids;
     FILE *temp_fd;
 };
 
@@ -930,6 +940,7 @@
         status = U1DB_INTERNAL_ERROR;
     } else {
         status = doc_to_tempfile(doc, state->generations[state->offset],
+                                 state->trans_ids[state->offset],
                                  state->temp_fd);
     }
     u1db_free_doc(&doc);
@@ -943,9 +954,10 @@
 
 static int
 st_http_sync_exchange_doc_ids(u1db_sync_target *st, u1database *source_db,
-        int n_doc_ids, const char **doc_ids, int *generations,
-        int *target_gen, char **target_trans_id,
-        void *context, u1db_doc_gen_callback cb)
+                              int n_doc_ids, const char **doc_ids,
+                              int *generations, const char **trans_ids,
+                              int *target_gen, char **target_trans_id,
+                              void *context, u1db_doc_gen_callback cb)
 {
     int status;
     FILE *temp_fd = NULL;
@@ -968,6 +980,7 @@
     if (status != U1DB_OK) { goto finish; }
     state.num = n_doc_ids;
     state.generations = generations;
+    state.trans_ids = trans_ids;
     state.temp_fd = temp_fd;
     status = u1db_get_docs(source_db, n_doc_ids, doc_ids, 0, 1,
             &state, get_docs_to_tempfile);

=== modified file 'src/u1db_sync_target.c'
--- src/u1db_sync_target.c	2012-06-08 10:06:05 +0000
+++ src/u1db_sync_target.c	2012-06-15 00:40:24 +0000
@@ -32,13 +32,15 @@
 static int st_sync_exchange(u1db_sync_target *st,
                           const char *source_replica_uid, int n_docs,
                           u1db_document **docs, int *generations,
-                          int *target_gen, char **target_trans_id,
-                          void *context, u1db_doc_gen_callback cb);
+                          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,
-        int *target_gen, char **target_trans_id,
-        void *context, u1db_doc_gen_callback cb);
+                                    u1database *source_db, int n_doc_ids,
+                                    const char **doc_ids, int *generations,
+                                    const char **trans_ids, int *target_gen,
+                                    char **target_trans_id, void *context,
+                                    u1db_doc_gen_callback cb);
 static int st_get_sync_exchange(u1db_sync_target *st,
                          const char *source_replica_uid,
                          int source_gen,
@@ -57,6 +59,7 @@
     void *user_context;
     u1db_doc_gen_callback user_cb;
     int *gen_for_doc_ids;
+    const char **trans_ids_for_doc_ids;
     int free_when_done;
 };
 
@@ -203,6 +206,10 @@
         free((*exchange)->gen_for_doc_ids);
         (*exchange)->gen_for_doc_ids = NULL;
     }
+    if ((*exchange)->trans_ids_for_doc_ids != NULL) {
+        free((*exchange)->trans_ids_for_doc_ids);
+        (*exchange)->trans_ids_for_doc_ids = NULL;
+    }
     if ((*exchange)->target_trans_id != NULL) {
         free((*exchange)->target_trans_id);
         (*exchange)->target_trans_id = NULL;
@@ -275,7 +282,7 @@
 
 int
 u1db__sync_exchange_insert_doc_from_source(u1db_sync_exchange *se,
-        u1db_document *doc, int source_gen)
+        u1db_document *doc, int source_gen, const char *trans_id)
 {
     int status = U1DB_OK;
     int insert_state;
@@ -285,7 +292,8 @@
     }
     // fprintf(stderr, "Inserting %s from source\n", doc->doc_id);
     status = u1db__put_doc_if_newer(se->db, doc, 0, se->source_replica_uid,
-                                    source_gen, NULL, &insert_state, &at_gen);
+                                    source_gen, trans_id, &insert_state,
+                                    &at_gen);
     if (insert_state == U1DB_INSERTED || insert_state == U1DB_CONVERGED) {
         lh_table_insert(se->seen_ids, strdup(doc->doc_id),
         (void *)(intptr_t)at_gen);
@@ -305,6 +313,7 @@
     struct lh_table *exclude_ids;
     char **doc_ids_to_return;
     int *gen_for_doc_ids;
+    const char **trans_ids_for_doc_ids;
 };
 
 // Callback for whats_changed to map the callback into the sync_exchange
@@ -327,25 +336,29 @@
     if (state->num_doc_ids >= state->max_doc_ids) {
         state->max_doc_ids = (state->max_doc_ids * 2) + 10;
         if (state->doc_ids_to_return == NULL) {
-            state->doc_ids_to_return = (char **)calloc(state->max_doc_ids,
-                                                       sizeof(char*));
-            state->gen_for_doc_ids = (int *)calloc(state->max_doc_ids,
-                                                   sizeof(int));
+            state->doc_ids_to_return = (char **)calloc(
+                state->max_doc_ids, sizeof(char*));
+            state->gen_for_doc_ids = (int *)calloc(
+                state->max_doc_ids, sizeof(int));
+            state->trans_ids_for_doc_ids = (const char **)calloc(
+                state->max_doc_ids, sizeof(char*));
         } else {
             state->doc_ids_to_return = (char **)realloc(
-                    state->doc_ids_to_return,
-                    state->max_doc_ids * sizeof(char*));
-            state->gen_for_doc_ids = (int *)realloc(state->gen_for_doc_ids,
-                    state->max_doc_ids * sizeof(int));
+                state->doc_ids_to_return, state->max_doc_ids * sizeof(char*));
+            state->gen_for_doc_ids = (int *)realloc(
+                state->gen_for_doc_ids, state->max_doc_ids * sizeof(int));
+            state->trans_ids_for_doc_ids = (const char **)realloc(
+                state->gen_for_doc_ids, state->max_doc_ids * sizeof(char*));
         }
-        if (state->doc_ids_to_return == NULL
-                || state->gen_for_doc_ids == NULL)
+        if (state->doc_ids_to_return == NULL || state->gen_for_doc_ids == NULL
+                || state->trans_ids_for_doc_ids == NULL)
         {
             return U1DB_NOMEM;
         }
     }
     state->doc_ids_to_return[state->num_doc_ids] = strdup(doc_id);
     state->gen_for_doc_ids[state->num_doc_ids] = gen;
+    state->trans_ids_for_doc_ids[state->num_doc_ids] = trans_id;
     state->num_doc_ids++;
     return 0;
 }
@@ -370,6 +383,7 @@
     if (status != U1DB_OK) {
         free(state.doc_ids_to_return);
         free(state.gen_for_doc_ids);
+        free(state.trans_ids_for_doc_ids);
         goto finish;
     }
     if (se->trace_cb) {
@@ -379,6 +393,7 @@
     se->num_doc_ids = state.num_doc_ids;
     se->doc_ids_to_return = state.doc_ids_to_return;
     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);
@@ -397,8 +412,9 @@
     // Note: using doc_offset in this way assumes that u1db_get_docs will
     //       always return them in exactly the order we requested. This is
     //       probably true, though.
-    status = ctx->user_cb(ctx->user_context, doc,
-            ctx->gen_for_doc_ids[ctx->doc_offset]);
+    status = ctx->user_cb(
+        ctx->user_context, doc, ctx->gen_for_doc_ids[ctx->doc_offset],
+        ctx->trans_ids_for_doc_ids[ctx->doc_offset]);
     ctx->doc_offset++;
     if (ctx->free_when_done) {
         u1db_free_doc(&doc);
@@ -409,7 +425,8 @@
 
 int
 u1db__sync_exchange_return_docs(u1db_sync_exchange *se, void *context,
-        int (*cb)(void *context, u1db_document *doc, int gen))
+                                int (*cb)(void *context, u1db_document *doc,
+                                          int gen, const char *trans_id))
 {
     int status = U1DB_OK;
     struct _get_docs_to_doc_gen_context state = {0};
@@ -420,6 +437,7 @@
     state.user_cb = cb;
     state.doc_offset = 0;
     state.gen_for_doc_ids = se->gen_for_doc_ids;
+    state.trans_ids_for_doc_ids = se->trans_ids_for_doc_ids;
     if (se->trace_cb) {
         status = se->trace_cb(se->trace_context, "before get_docs");
         if (status != U1DB_OK) { goto finish; }
@@ -440,14 +458,16 @@
 };
 
 static int
-return_doc_to_insert_from_target(void *context, u1db_document *doc, int gen)
+return_doc_to_insert_from_target(void *context, u1db_document *doc, int gen,
+                                 const char *trans_id)
 {
     int status, insert_state;
     struct _return_doc_state *state;
     state = (struct _return_doc_state *)context;
 
-    status = u1db__put_doc_if_newer(state->db, doc, 1, state->target_uid, gen,
-                                    NULL, &insert_state, NULL);
+    status = u1db__put_doc_if_newer(
+        state->db, doc, 1, state->target_uid, gen, trans_id, &insert_state,
+        NULL);
     u1db_free_doc(&doc);
     if (status == U1DB_OK) {
         if (insert_state == U1DB_INSERTED || insert_state == U1DB_CONFLICTED) {
@@ -462,7 +482,8 @@
 
 static int
 get_and_insert_docs(u1database *source_db, u1db_sync_exchange *se,
-                    int n_doc_ids, const char **doc_ids, int *generations)
+                    int n_doc_ids, const char **doc_ids, int *generations,
+                    const char **trans_ids)
 {
     struct _get_docs_to_doc_gen_context get_doc_state = {0};
 
@@ -473,6 +494,7 @@
     get_doc_state.user_cb =
         (u1db_doc_gen_callback)u1db__sync_exchange_insert_doc_from_source;
     get_doc_state.gen_for_doc_ids = generations;
+    get_doc_state.trans_ids_for_doc_ids = trans_ids;
     return u1db_get_docs(source_db, n_doc_ids, doc_ids,
             0, 1, &get_doc_state, get_docs_to_gen_docs);
 }
@@ -480,11 +502,13 @@
 
 static int
 st_sync_exchange(u1db_sync_target *st, const char *source_replica_uid,
-                 int n_docs, u1db_document **docs,
-                 int *generations, int *target_gen, char **target_trans_id,
-                 void *context, u1db_doc_gen_callback cb)
+                 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)
 {
     int status, i;
+
     u1db_sync_exchange *exchange = NULL;
     if (st == NULL || generations == NULL || target_gen == NULL
             || target_trans_id == NULL || cb == NULL)
@@ -498,8 +522,8 @@
                                    *target_gen, &exchange);
     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]);
+        status = u1db__sync_exchange_insert_doc_from_source(
+            exchange, docs[i], generations[i], trans_ids[i]);
         if (status != U1DB_OK) { goto finish; }
     }
     status = u1db__sync_exchange_find_doc_ids_to_return(exchange);
@@ -521,7 +545,7 @@
 static int
 st_sync_exchange_doc_ids(u1db_sync_target *st, u1database *source_db,
         int n_doc_ids, const char **doc_ids, int *generations,
-        int *target_gen, char **target_trans_id,
+        const char **trans_ids, int *target_gen, char **target_trans_id,
         void *context, u1db_doc_gen_callback cb)
 {
     int status;
@@ -542,7 +566,7 @@
     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);
+            n_doc_ids, doc_ids, generations, trans_ids);
         if (status != U1DB_OK) { goto finish; }
     }
     status = u1db__sync_exchange_find_doc_ids_to_return(exchange);
@@ -616,7 +640,7 @@
     status = target->sync_exchange_doc_ids(target, db,
         to_send_state.num_doc_ids,
         (const char**)to_send_state.doc_ids_to_return,
-        to_send_state.gen_for_doc_ids,
+        to_send_state.gen_for_doc_ids, to_send_state.trans_ids_for_doc_ids,
         &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; }
@@ -665,5 +689,8 @@
     if (to_send_state.gen_for_doc_ids != NULL) {
         free(to_send_state.gen_for_doc_ids);
     }
+    if (to_send_state.trans_ids_for_doc_ids != NULL) {
+        free(to_send_state.trans_ids_for_doc_ids);
+    }
     return status;
 }

=== modified file 'u1db/__init__.py'
--- u1db/__init__.py	2012-06-12 15:44:43 +0000
+++ u1db/__init__.py	2012-06-15 00:40:24 +0000
@@ -557,10 +557,9 @@
         their latest change in order from the oldest change to the
         newest.
 
-        :param docs_by_generation: A list of [(Document, generation)]
-              pairs indicating documents which should be updated on
-              this replica paired with the generation of their
-              latest change.
+        :param docs_by_generation: A list of [(Document, generation,
+            transaction_id)] pairs indicating documents which should be updated
+            on this replica paired with the generation 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

=== modified file 'u1db/remote/http_app.py'
--- u1db/remote/http_app.py	2012-05-31 12:22:03 +0000
+++ u1db/remote/http_app.py	2012-06-15 00:40:24 +0000
@@ -324,14 +324,14 @@
                                                   last_known_generation)
 
     @http_method(content_as_args=True)
-    def post_stream_entry(self, id, rev, content, gen):
+    def post_stream_entry(self, id, rev, content, gen, trans_id):
         doc = Document(id, rev, content)
-        self.sync_exch.insert_doc_from_source(doc, gen)
+        self.sync_exch.insert_doc_from_source(doc, gen, trans_id)
 
     def post_end(self):
-        def send_doc(doc, gen):
+        def send_doc(doc, gen, trans_id):
             entry = dict(id=doc.doc_id, rev=doc.rev, content=doc.get_json(),
-                         gen=gen)
+                         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'

=== modified file 'u1db/remote/http_target.py'
--- u1db/remote/http_target.py	2012-05-31 12:22:03 +0000
+++ u1db/remote/http_target.py	2012-06-15 00:40:24 +0000
@@ -65,7 +65,7 @@
                 line, comma = utils.check_and_strip_comma(entry)
                 entry = simplejson.loads(line)
                 doc = Document(entry['id'], entry['rev'], entry['content'])
-                return_doc_cb(doc, entry['gen'])
+                return_doc_cb(doc, entry['gen'], entry['trans_id'])
         if parts[-1] != ']':
             try:
                 partdic = simplejson.loads(parts[-1])
@@ -98,9 +98,9 @@
         comma = ''
         size += prepare(last_known_generation=last_known_generation)
         comma = ','
-        for doc, gen in docs_by_generations:
+        for doc, gen, trans_id in docs_by_generations:
             size += prepare(id=doc.doc_id, rev=doc.rev, content=doc.get_json(),
-                            gen=gen)
+                            gen=gen, trans_id=trans_id)
         entries.append('\r\n]')
         size += len(entries[-1])
         self._conn.putheader('content-length', str(size))

=== modified file 'u1db/sync.py'
--- u1db/sync.py	2012-05-31 12:22:03 +0000
+++ u1db/sync.py	2012-06-15 00:40:24 +0000
@@ -40,7 +40,7 @@
         self.target_replica_uid = None
         self.num_inserted = 0
 
-    def _insert_doc_from_target(self, doc, replica_gen):
+    def _insert_doc_from_target(self, doc, replica_gen, trans_id):
         """Try to insert synced document from target.
 
         Implements TAKE OTHER semantics: any document from the target
@@ -54,7 +54,7 @@
         # was effectively inserted.
         state, _ = self.source._put_doc_if_newer(doc, save_conflict=True,
             replica_uid=self.target_replica_uid, replica_gen=replica_gen,
-            replica_trans_id=None)
+            replica_trans_id=trans_id)
         if state == 'inserted':
             self.num_inserted += 1
         elif state == 'converged':
@@ -110,7 +110,10 @@
         # prepare to send all the changed docs
         docs_to_send = self.source.get_docs(changed_doc_ids,
             check_for_conflicts=False, include_deleted=True)
-        docs_by_generation = zip(docs_to_send, (gen for _, gen, _ in changes))
+        # TODO: there must be a way to not iterate twice
+        docs_by_generation = zip(
+            docs_to_send, (gen for _, gen, _ in changes),
+            (trans for _, _, trans in changes))
 
         # exchange documents and try to insert the returned ones with
         # the target, return target synced-up-to gen
@@ -153,7 +156,7 @@
             return
         self._trace_hook(state)
 
-    def insert_doc_from_source(self, doc, source_gen):
+    def insert_doc_from_source(self, doc, source_gen, trans_id):
         """Try to insert synced document from source.
 
         Conflicting documents are not inserted but will be sent over
@@ -171,7 +174,7 @@
         """
         state, at_gen = self._db._put_doc_if_newer(doc, save_conflict=False,
             replica_uid=self.source_replica_uid, replica_gen=source_gen,
-            replica_trans_id=None)
+            replica_trans_id=trans_id)
         if state == 'inserted':
             self.seen_ids[doc.doc_id] = at_gen
         elif state == 'converged':
@@ -212,10 +215,10 @@
         self.new_trans_id = trans_id
         seen_ids = self.seen_ids
         # changed docs that weren't superseded by or converged with
-        self.changes_to_return = [(doc_id, gen) for (doc_id, gen, _) in changes
-                                  # there was a subsequent update
-                                  if doc_id not in seen_ids or
-                                     seen_ids.get(doc_id) < gen]
+        self.changes_to_return = [
+            (doc_id, gen, trans_id) for (doc_id, gen, trans_id) in changes
+            # there was a subsequent update
+            if doc_id not in seen_ids or seen_ids.get(doc_id) < gen]
         return self.new_gen
 
     def return_docs(self, return_doc_cb):
@@ -224,21 +227,23 @@
 
         The final step of a sync exchange.
 
-        :param: return_doc_cb(doc, gen): is a callback
+        :param: return_doc_cb(doc, gen, trans_id): is a callback
                 used to return the documents with their last change generation
                 to the target replica.
         :return: None
         """
         changes_to_return = self.changes_to_return
         # return docs, including conflicts
-        changed_doc_ids = [doc_id for doc_id, _ in changes_to_return]
+        changed_doc_ids = [doc_id for doc_id, _, _ in changes_to_return]
         self._trace('before get_docs')
         docs = self._db.get_docs(
             changed_doc_ids, check_for_conflicts=False, include_deleted=True)
 
-        docs_by_gen = izip(docs, (gen for _, gen in changes_to_return))
-        for doc, gen in docs_by_gen:
-            return_doc_cb(doc, gen)
+        docs_by_gen = izip(
+            docs, (gen for _, gen, _ in changes_to_return),
+            (trans_id for _, _, trans_id in changes_to_return))
+        for doc, gen, trans_id in docs_by_gen:
+            return_doc_cb(doc, gen, trans_id)
         # for tests
         self._db._last_exchange_log['return'] = {
             'docs': [(d.doc_id, d.rev) for d in docs],
@@ -260,8 +265,8 @@
         if self._trace_hook:
             sync_exch._set_trace_hook(self._trace_hook)
         # 1st step: try to insert incoming docs and record progress
-        for doc, doc_gen in docs_by_generations:
-            sync_exch.insert_doc_from_source(doc, doc_gen)
+        for doc, doc_gen, trans_id in docs_by_generations:
+            sync_exch.insert_doc_from_source(doc, doc_gen, trans_id)
         # 2nd step: find changed documents (including conflicts) to return
         new_gen = sync_exch.find_changes_to_return()
         # final step: return docs and record source replica sync point

=== modified file 'u1db/tests/c_backend_wrapper.pyx'
--- u1db/tests/c_backend_wrapper.pyx	2012-06-13 11:15:57 +0000
+++ u1db/tests/c_backend_wrapper.pyx	2012-06-15 00:40:24 +0000
@@ -61,7 +61,7 @@
     ctypedef int (*u1db_key_callback)(void *context, int num_fields,
                                       const_char_ptr *key)
     ctypedef int (*u1db_doc_gen_callback)(void *context,
-        u1db_document *doc, int gen)
+        u1db_document *doc, int gen, const_char_ptr trans_id)
     ctypedef int (*u1db_trans_info_callback)(void *context,
         const_char_ptr doc_id, int gen, const_char_ptr trans_id)
 
@@ -164,6 +164,7 @@
         int num_doc_ids
         char **doc_ids_to_return
         int *gen_for_doc_ids
+        const_char_ptr *trans_ids_for_doc_ids
 
     ctypedef int (*u1db__trace_callback)(void *context, const_char_ptr state)
     ctypedef struct u1db_sync_target:
@@ -176,12 +177,16 @@
         int (*sync_exchange)(u1db_sync_target *st,
                              char *source_replica_uid, int n_docs,
                              u1db_document **docs, int *generations,
-                             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, int *target_gen, char **target_trans_id,
-                void *context, u1db_doc_gen_callback cb) nogil
+                                     u1database *source_db, int n_doc_ids,
+                                     const_char_ptr *doc_ids, int *generations,
+                                     const_char_ptr *trans_ids,
+                                     int *target_gen, char **target_trans_id,
+                                     void *context,
+                                     u1db_doc_gen_callback cb) nogil
         int (*get_sync_exchange)(u1db_sync_target *st,
                                  char *source_replica_uid,
                                  int last_known_source_gen,
@@ -228,10 +233,12 @@
                                 int *local_gen_before_sync) nogil
 
     int u1db__sync_exchange_insert_doc_from_source(u1db_sync_exchange *se,
-            u1db_document *doc, int source_gen)
+            u1db_document *doc, int source_gen, const_char_ptr trans_id)
     int u1db__sync_exchange_find_doc_ids_to_return(u1db_sync_exchange *se)
     int u1db__sync_exchange_return_docs(u1db_sync_exchange *se, void *context,
-            int (*cb)(void *context, u1db_document *doc, int gen))
+                                        int (*cb)(void *context,
+                                                  u1db_document *doc, int gen,
+                                                  const_char_ptr trans_id))
     int u1db__create_http_sync_target(char *url, u1db_sync_target **target)
     int u1db__create_oauth_http_sync_target(char *url,
         char *consumer_key, char *consumer_secret,
@@ -331,13 +338,13 @@
 
 
 cdef int return_doc_cb_wrapper(void *context, u1db_document *doc,
-        int gen) with gil:
+                               int gen, const_char_ptr trans_id) with gil:
     cdef CDocument pydoc
     user_cb = <object>context
     pydoc = CDocument()
     pydoc._doc = doc
     try:
-        user_cb(pydoc, gen)
+        user_cb(pydoc, gen, trans_id)
     except Exception, e:
         # We suppress the exception here, because intermediating through the C
         # layer gets a bit crazy
@@ -606,11 +613,12 @@
             self._check()
             return self._exchange.target_gen
 
-    def insert_doc_from_source(self, CDocument doc, source_gen):
+    def insert_doc_from_source(self, CDocument doc, source_gen,
+                               source_trans_id):
         self._check()
         handle_status("insert_doc_from_source",
             u1db__sync_exchange_insert_doc_from_source(self._exchange,
-                doc._doc, source_gen))
+                doc._doc, source_gen, source_trans_id))
 
     def find_doc_ids_to_return(self):
         self._check()
@@ -642,8 +650,10 @@
         if (self._exchange.num_doc_ids > 0
                 and self._exchange.doc_ids_to_return != NULL):
             for i from 0 <= i < self._exchange.num_doc_ids:
-                res.append((self._exchange.doc_ids_to_return[i],
-                            self._exchange.gen_for_doc_ids[i]))
+                res.append(
+                    (self._exchange.doc_ids_to_return[i],
+                     self._exchange.gen_for_doc_ids[i],
+                     self._exchange.trans_ids_for_doc_ids[i]))
         return res
 
 
@@ -714,25 +724,32 @@
         if generations == NULL:
             free(<void *>doc_ids)
             raise MemoryError
+        trans_ids = <const_char_ptr*>calloc(num_doc_ids, sizeof(char *))
+        if trans_ids == NULL:
+            raise MemoryError
         res_trans_id = ''
         try:
-            for i, (doc_id, gen) in enumerate(doc_id_generations):
+            for i, (doc_id, gen, trans_id) in enumerate(doc_id_generations):
                 doc_ids[i] = PyString_AsString(doc_id)
                 generations[i] = gen
+                trans_ids[i] = trans_id
             target_gen = last_known_generation
             with nogil:
                 status = self._st.sync_exchange_doc_ids(self._st, sdb._db,
-                    num_doc_ids, doc_ids, generations,
+                    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)
         finally:
-            free(<void *>doc_ids)
-            free(generations)
+            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,
@@ -740,6 +757,7 @@
         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 int i, count, status, target_gen
 
@@ -754,15 +772,19 @@
             generations = <int*>calloc(count, sizeof(int))
             if generations == NULL:
                 raise MemoryError
+            trans_ids = <const_char_ptr*>calloc(count, sizeof(char*))
+            if trans_ids == NULL:
+                raise MemoryError
             for i from 0 <= i < count:
                 cur_doc = docs_by_generations[i][0]
                 generations[i] = docs_by_generations[i][1]
+                trans_ids[i] = docs_by_generations[i][2]
                 docs[i] = cur_doc._doc
             target_gen = last_known_generation
             with nogil:
                 status = self._st.sync_exchange(self._st,
                         source_replica_uid, count,
-                        docs, generations, &target_gen, &trans_id,
+                        docs, generations, trans_ids, &target_gen, &trans_id,
                         <void *>return_doc_cb, return_doc_cb_wrapper)
             handle_status("sync_exchange", status)
         finally:
@@ -770,6 +792,8 @@
                 free(docs)
             if generations != NULL:
                 free(generations)
+            if trans_ids != NULL:
+                free(trans_ids)
             if trans_id != NULL:
                 res_trans_id = trans_id
                 free(trans_id)

=== modified file 'u1db/tests/test_backends.py'
--- u1db/tests/test_backends.py	2012-06-12 15:44:43 +0000
+++ u1db/tests/test_backends.py	2012-06-15 00:40:24 +0000
@@ -1376,7 +1376,7 @@
             pass
 
         doc_other = self.make_document(doc.doc_id, other_rev, new_content)
-        docs_by_gen = [(doc_other, 10)]
+        docs_by_gen = [(doc_other, 10, 'T-sid')]
         st.sync_exchange(
             docs_by_gen, 'other-replica', last_known_generation=0,
             return_doc_cb=ignore)

=== modified file 'u1db/tests/test_c_backend.py'
--- u1db/tests/test_c_backend.py	2012-06-12 14:57:44 +0000
+++ u1db/tests/test_c_backend.py	2012-06-15 00:40:24 +0000
@@ -287,10 +287,10 @@
         doc = c_backend_wrapper.make_document('doc-id', 'replica:1',
                 tests.simple_doc)
         self.assertEqual([], exc.get_seen_ids())
-        exc.insert_doc_from_source(doc, 10)
+        exc.insert_doc_from_source(doc, 10, 'T-sid')
         self.assertGetDoc(self.db, 'doc-id', 'replica:1', tests.simple_doc,
                           False)
-        self.assertEqual((10, None),
+        self.assertEqual((10, 'T-sid'),
                          self.db._get_sync_gen_info('source-uid'))
         self.assertEqual(['doc-id'], exc.get_seen_ids())
 
@@ -301,7 +301,7 @@
                 tests.nested_doc)
         self.assertEqual([], exc.get_seen_ids())
         # The insert should be rejected and the doc_id not considered 'seen'
-        exc.insert_doc_from_source(doc2, 10)
+        exc.insert_doc_from_source(doc2, 10, 'T-sid')
         self.assertGetDoc(
             self.db, doc.doc_id, doc.rev, tests.simple_doc, False)
         self.assertEqual([], exc.get_seen_ids())
@@ -311,7 +311,10 @@
         exc = self.st._get_sync_exchange("source-uid", 0)
         self.assertEqual(0, exc.target_gen)
         exc.find_doc_ids_to_return()
-        self.assertEqual([(doc.doc_id, 1)], exc.get_doc_ids_to_return())
+        doc_id = exc.get_doc_ids_to_return()[0]
+        self.assertEqual(
+            (doc.doc_id, 1), doc_id[:-1])
+        self.assertTrue(doc_id[-1].startswith('T-'))
         self.assertEqual(1, exc.target_gen)
 
     def test_sync_exchange_find_doc_ids_not_including_recently_inserted(self):
@@ -320,22 +323,23 @@
         exc = self.st._get_sync_exchange("source-uid", 0)
         doc3 = c_backend_wrapper.make_document(doc1.doc_id,
                 doc1.rev + "|zreplica:2", tests.simple_doc)
-        exc.insert_doc_from_source(doc3, 10)
+        exc.insert_doc_from_source(doc3, 10, 'T-sid')
         exc.find_doc_ids_to_return()
-        self.assertEqual([(doc2.doc_id, 2)], exc.get_doc_ids_to_return())
+        self.assertEqual(
+            (doc2.doc_id, 2), exc.get_doc_ids_to_return()[0][:-1])
         self.assertEqual(3, exc.target_gen)
 
     def test_sync_exchange_return_docs(self):
         returned = []
 
-        def return_doc_cb(doc, gen):
-            returned.append((doc, gen))
+        def return_doc_cb(doc, gen, trans_id):
+            returned.append((doc, gen, trans_id))
 
         doc1 = self.db.create_doc(tests.simple_doc)
         exc = self.st._get_sync_exchange("source-uid", 0)
         exc.find_doc_ids_to_return()
         exc.return_docs(return_doc_cb)
-        self.assertEqual([(doc1, 1)], returned)
+        self.assertEqual((doc1, 1), returned[0][:-1])
 
     def test_sync_exchange_doc_ids(self):
         doc1 = self.db.create_doc(tests.simple_doc, doc_id='doc-1')
@@ -343,16 +347,17 @@
         doc2 = db2.create_doc(tests.nested_doc, doc_id='doc-2')
         returned = []
 
-        def return_doc_cb(doc, gen):
-            returned.append((doc, gen))
-        val = self.st.sync_exchange_doc_ids(db2, [(doc2.doc_id, 1)], 0,
-                                            return_doc_cb)
+        def return_doc_cb(doc, gen, trans_id):
+            returned.append((doc, gen, trans_id))
+
+        val = self.st.sync_exchange_doc_ids(
+            db2, [(doc2.doc_id, 1, 'T-sid')], 0, 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)
         self.assertGetDoc(self.db, doc2.doc_id, doc2.rev, tests.nested_doc,
                           False)
-        self.assertEqual([(doc1, 1)], returned)
+        self.assertEqual((doc1, 1), returned[0][:-1])
 
 
 class TestCHTTPSyncTarget(BackendTests):

=== modified file 'u1db/tests/test_http_app.py'
--- u1db/tests/test_http_app.py	2012-05-31 12:22:03 +0000
+++ u1db/tests/test_http_app.py	2012-06-15 00:40:24 +0000
@@ -715,9 +715,9 @@
     def test_sync_exchange_send(self):
         entries = {
             10: {'id': 'doc-here', 'rev': 'replica:1', 'content':
-                 '{"value": "here"}', 'gen': 10},
+                 '{"value": "here"}', 'gen': 10, 'trans_id': 'T-sid'},
             11: {'id': 'doc-here2', 'rev': 'replica:1', 'content':
-                 '{"value": "here2"}', 'gen': 11}
+                 '{"value": "here2"}', 'gen': 11, 'trans_id': 'T-sed'}
             }
 
         gens = []
@@ -793,12 +793,18 @@
         self.assertEqual({'new_generation': 2,
                           'new_transaction_id': last_trans_id},
                          simplejson.loads(parts[1].rstrip(",")))
-        self.assertEqual({'content': '{"value": "there"}',
-                          'rev': doc.rev, 'id': doc.doc_id, 'gen': 1},
-                         simplejson.loads(parts[2].rstrip(",")))
-        self.assertEqual({'content': '{"value": "there2"}',
-                          'rev': doc2.rev, 'id': doc2.doc_id, 'gen': 2},
-                         simplejson.loads(parts[3].rstrip(",")))
+        part2 = simplejson.loads(parts[2].rstrip(","))
+        self.assertTrue(part2['trans_id'].startswith('T-'))
+        self.assertEqual('{"value": "there"}', part2['content'])
+        self.assertEqual(doc.rev, part2['rev'])
+        self.assertEqual(doc.doc_id, part2['id'])
+        self.assertEqual(1, part2['gen'])
+        part3 = simplejson.loads(parts[3].rstrip(","))
+        self.assertTrue(part3['trans_id'].startswith('T-'))
+        self.assertEqual('{"value": "there2"}', part3['content'])
+        self.assertEqual(doc2.rev, part3['rev'])
+        self.assertEqual(doc2.doc_id, part3['id'])
+        self.assertEqual(2, part3['gen'])
         self.assertEqual(']', parts[4])
 
     def test_sync_exchange_error_in_stream(self):

=== modified file 'u1db/tests/test_remote_sync_target.py'
--- u1db/tests/test_remote_sync_target.py	2012-05-31 12:22:03 +0000
+++ u1db/tests/test_remote_sync_target.py	2012-06-15 00:40:24 +0000
@@ -81,8 +81,9 @@
         self.assertRaises(errors.BrokenSyncStream,
                           tgt._parse_sync_stream,
                           '[\r\n{},\r\n{"id": "i", "rev": "r", '
-                          '"content": "c", "gen": 3},\r\n]',
-                          lambda doc, gen: None)
+                          '"content": "c", "gen": 3, "trans_id": "T-sid"}'
+                          ',\r\n]',
+                          lambda doc, gen, trans_id: None)
 
     def test_error_in_stream(self):
         tgt = http_target.HTTPSyncTarget("http://foo/foo")
@@ -102,6 +103,7 @@
 
 
 def http_server_def():
+
     def make_server(host_port, handler, state):
         application = http_app.HTTPApp(state)
         srv = simple_server.WSGIServer(host_port, handler)
@@ -111,9 +113,11 @@
         #                                handler
         #                                )
         return srv
+
     class req_handler(simple_server.WSGIRequestHandler):
         def log_request(*args):
             pass  # suppress
+
     #rh = httpserver.WSGIHandler
     return make_server, req_handler, "shutdown", "http"
 
@@ -123,6 +127,7 @@
 
 
 def oauth_http_server_def():
+
     def make_server(host_port, handler, state):
         app = http_app.HTTPApp(state)
         application = oauth_middleware.OAuthMiddleware(app, None)
@@ -132,9 +137,11 @@
         application.base_url = "http://%s:%s" % srv.server_address
         srv.set_app(application)
         return srv
+
     class req_handler(simple_server.WSGIRequestHandler):
         def log_request(*args):
             pass  # suppress
+
     return make_server, req_handler, "shutdown", "http"
 
 
@@ -181,26 +188,31 @@
         db = self.request_state._create_database('test')
         remote_target = self.getSyncTarget('test')
         other_docs = []
+
         def receive_doc(doc):
             other_docs.append((doc.doc_id, doc.rev, doc.get_json()))
+
         doc = self.make_document('doc-here', 'replica:1', '{"value": "here"}')
         new_gen, trans_id = remote_target.sync_exchange(
-                [(doc, 10)],
+                [(doc, 10, 'T-sid')],
                 'replica', last_known_generation=0,
                 return_doc_cb=receive_doc)
         self.assertEqual(1, new_gen)
-        self.assertGetDoc(db, 'doc-here', 'replica:1', '{"value": "here"}',
-                          False)
+        self.assertGetDoc(
+            db, 'doc-here', 'replica:1', '{"value": "here"}', False)
 
     def test_sync_exchange_send_failure_and_retry_scenario(self):
         self.startServer()
+
         def blackhole_getstderr(inst):
             return cStringIO.StringIO()
+
         self.patch(self.server.RequestHandlerClass, 'get_stderr',
                    blackhole_getstderr)
         db = self.request_state._create_database('test')
         _put_doc_if_newer = db._put_doc_if_newer
         trigger_ids = ['doc-here2']
+
         def bomb_put_doc_if_newer(doc, save_conflict,
                                   replica_uid=None, replica_gen=None,
                                   replica_trans_id=None):
@@ -212,33 +224,37 @@
         self.patch(db, '_put_doc_if_newer', bomb_put_doc_if_newer)
         remote_target = self.getSyncTarget('test')
         other_changes = []
-        def receive_doc(doc, gen):
-            other_changes.append((doc.doc_id, doc.rev, doc.get_json(), gen))
+
+        def receive_doc(doc, gen, trans_id):
+            other_changes.append(
+                (doc.doc_id, doc.rev, doc.get_json(), gen, trans_id))
+
         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),
-                 (doc2, 11)
+                [(doc1, 10, 'T-sid'),
+                 (doc2, 11, 'T-sud')
                  ], 'replica', last_known_generation=0,
                 return_doc_cb=receive_doc)
         self.assertGetDoc(db, 'doc-here', 'replica:1', '{"value": "here"}',
                           False)
-        self.assertEqual((10, None), db._get_sync_gen_info('replica'))
+        self.assertEqual((10, 'T-sid'), db._get_sync_gen_info('replica'))
         self.assertEqual([], other_changes)
         # retry
         trigger_ids = []
         new_gen, trans_id = remote_target.sync_exchange(
-                [(doc2, 11)
+                [(doc2, 11, 'T-sud')
                  ], 'replica', last_known_generation=0,
                 return_doc_cb=receive_doc)
         self.assertGetDoc(db, 'doc-here2', 'replica:1', '{"value": "here2"}',
                           False)
-        self.assertEqual((11, None), db._get_sync_gen_info('replica'))
+        self.assertEqual((11, 'T-sud'), db._get_sync_gen_info('replica'))
         self.assertEqual(2, new_gen)
         # bounced back to us
-        self.assertEqual([('doc-here', 'replica:1', '{"value": "here"}', 1)],
-                         other_changes)
+        self.assertEqual(
+            ('doc-here', 'replica:1', '{"value": "here"}', 1),
+            other_changes[0][:-1])
 
     def test_sync_exchange_in_stream_error(self):
         self.startServer()
@@ -261,14 +277,16 @@
         remote_target = self.getSyncTarget('test')
         other_changes = []
 
-        def receive_doc(doc, gen):
-            other_changes.append((doc.doc_id, doc.rev, doc.get_json(), gen))
+        def receive_doc(doc, gen, trans_id):
+            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.assertEqual([(doc.doc_id, doc.rev, '{"value": "there"}', 1)],
-                         other_changes)
+        self.assertEqual(
+            (doc.doc_id, doc.rev, '{"value": "there"}', 1),
+            other_changes[0][:-1])
 
     def test_sync_exchange_receive(self):
         self.startServer()
@@ -276,14 +294,18 @@
         doc = db.create_doc('{"value": "there"}')
         remote_target = self.getSyncTarget('test')
         other_changes = []
-        def receive_doc(doc, gen):
-            other_changes.append((doc.doc_id, doc.rev, doc.get_json(), gen))
+
+        def receive_doc(doc, gen, trans_id):
+            other_changes.append(
+                (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)
         self.assertEqual(1, new_gen)
-        self.assertEqual([(doc.doc_id, doc.rev, '{"value": "there"}', 1)],
-                         other_changes)
+        self.assertEqual(
+            (doc.doc_id, doc.rev, '{"value": "there"}', 1),
+            other_changes[0][:-1])
 
 
 load_tests = tests.load_with_scenarios

=== modified file 'u1db/tests/test_sync.py'
--- u1db/tests/test_sync.py	2012-06-08 07:39:35 +0000
+++ u1db/tests/test_sync.py	2012-06-15 00:40:24 +0000
@@ -128,8 +128,9 @@
         del self.db
         super(DatabaseSyncTargetTests, self).tearDown()
 
-    def receive_doc(self, doc, gen):
-        self.other_changes.append((doc.doc_id, doc.rev, doc.get_json(), gen))
+    def receive_doc(self, doc, gen, trans_id):
+        self.other_changes.append(
+            (doc.doc_id, doc.rev, doc.get_json(), gen, trans_id))
 
     def set_trace_hook(self, callback):
         try:
@@ -157,7 +158,8 @@
 
     def test_sync_exchange(self):
         docs_by_gen = [
-            (self.make_document('doc-id', 'replica:1', simple_doc), 10)]
+            (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,
                                         return_doc_cb=self.receive_doc)
@@ -172,7 +174,7 @@
         doc = self.db.create_doc('{}')
         edit_rev = 'replica:1|' + doc.rev
         docs_by_gen = [
-            (self.make_document(doc.doc_id, edit_rev, None), 10)]
+            (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)
@@ -186,8 +188,9 @@
 
     def test_sync_exchange_push_many(self):
         docs_by_gen = [
-            (self.make_document('doc-id', 'replica:1', simple_doc), 10),
-            (self.make_document('doc-id2', 'replica:1', nested_doc), 11)]
+            (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)
@@ -204,13 +207,15 @@
         self.assertTransactionLog([doc.doc_id], self.db)
         new_doc = '{"key": "altval"}'
         docs_by_gen = [
-            (self.make_document(doc.doc_id, 'replica:1', new_doc), 10)]
-        new_gen, _ = self.st.sync_exchange(docs_by_gen, 'replica',
-                                        last_known_generation=0,
-                                        return_doc_cb=self.receive_doc)
+            (self.make_document(doc.doc_id, 'replica:1', new_doc), 10,
+             'T-sid')]
+        new_gen, _ = self.st.sync_exchange(
+            docs_by_gen, 'replica', last_known_generation=0,
+            return_doc_cb=self.receive_doc)
         self.assertTransactionLog([doc.doc_id], self.db)
-        self.assertEqual(([(doc.doc_id, doc.rev, simple_doc, 1)], 1),
-                         (self.other_changes, new_gen))
+        self.assertEqual(
+            (doc.doc_id, doc.rev, simple_doc, 1), self.other_changes[0][:-1])
+        self.assertEqual(1, new_gen)
         if self.whitebox:
             self.assertEqual(self.db._last_exchange_log['return'],
                              {'last_gen': 1, 'docs': [(doc.doc_id, doc.rev)]})
@@ -219,7 +224,7 @@
         doc = self.db.create_doc(simple_doc)
         self.assertTransactionLog([doc.doc_id], self.db)
         docs_by_gen = [
-            (self.make_document(doc.doc_id, doc.rev, simple_doc), 10)]
+            (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)
@@ -233,8 +238,9 @@
                                         last_known_generation=0,
                                         return_doc_cb=self.receive_doc)
         self.assertTransactionLog([doc.doc_id], self.db)
-        self.assertEqual(([(doc.doc_id, doc.rev, simple_doc, 1)], 1),
-                         (self.other_changes, new_gen))
+        self.assertEqual(
+            (doc.doc_id, doc.rev, simple_doc, 1), self.other_changes[0][:-1])
+        self.assertEqual(1, new_gen)
         if self.whitebox:
             self.assertEqual(self.db._last_exchange_log['return'],
                              {'last_gen': 1, 'docs': [(doc.doc_id, doc.rev)]})
@@ -247,8 +253,9 @@
                                         last_known_generation=0,
                                         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)], 2),
-                         (self.other_changes, new_gen))
+        self.assertEqual(
+            (doc.doc_id, doc.rev, None, 2), self.other_changes[0][:-1])
+        self.assertEqual(2, new_gen)
         if self.whitebox:
             self.assertEqual(self.db._last_exchange_log['return'],
                              {'last_gen': 2, 'docs': [(doc.doc_id, doc.rev)]})
@@ -261,9 +268,11 @@
                                         last_known_generation=0,
                                         return_doc_cb=self.receive_doc)
         self.assertTransactionLog([doc.doc_id, doc2.doc_id], self.db)
-        self.assertEqual(([(doc.doc_id, doc.rev, simple_doc, 1),
-                           (doc2.doc_id, doc2.rev, nested_doc, 2)], 2),
-                         (self.other_changes, new_gen))
+        self.assertEqual(2, new_gen)
+        self.assertEqual(
+            [(doc.doc_id, doc.rev, simple_doc, 1),
+             (doc2.doc_id, doc2.rev, nested_doc, 2)],
+            [c[:-1] for c in self.other_changes])
         if self.whitebox:
             self.assertEqual(
                 self.db._last_exchange_log['return'],
@@ -275,7 +284,8 @@
         self.assertTransactionLog([doc.doc_id], self.db)
         new_doc = '{"key": "altval"}'
         docs_by_gen = [
-            (self.make_document(doc.doc_id, 'test:1|z:2', new_doc), 10)]
+            (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)
@@ -298,11 +308,13 @@
         self.assertTransactionLog([doc.doc_id], self.db)
         new_doc = '{"key": "altval"}'
         docs_by_gen = [
-            (self.make_document(doc.doc_id, 'test:1|z:2', new_doc), 10)]
-        new_gen, _ = self.st.sync_exchange(docs_by_gen, 'other-replica',
-                                        last_known_generation=0,
-                                        return_doc_cb=self.receive_doc)
-        self.assertEqual((expected, 3), (self.other_changes, new_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)
+        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):
@@ -314,7 +326,8 @@
         self.assertTransactionLog([doc.doc_id], self.db)
         new_doc = '{"key": "altval"}'
         docs_by_gen = [
-            (self.make_document(doc.doc_id, 'test:1|z:2', new_doc), 10)]
+            (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)
@@ -323,8 +336,9 @@
     def test_sync_exchange_converged_handling(self):
         doc = self.db.create_doc(simple_doc)
         docs_by_gen = [
-            (self.make_document('new', 'other:1', '{}'), 4),
-            (self.make_document(doc.doc_id, doc.rev, doc.get_json()), 5)]
+            (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)
@@ -352,8 +366,9 @@
             self.skipTest("sync_exchange_doc_ids not implemented")
         db2 = self.create_database('test2')
         doc = db2.create_doc(simple_doc)
-        new_gen, trans_id = sync_exchange_doc_ids(db2, [(doc.doc_id, 10)], 0,
-                return_doc_cb=self.receive_doc)
+        new_gen, trans_id = sync_exchange_doc_ids(
+            db2, [(doc.doc_id, 10, 'T-sid')], 0,
+            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)
         last_trans_id = self.getLastTransId(self.db)