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

[Eric Casteleijn]

Simplified the checking of generations/transaction ids during sync. Invalid values now always result in errors, no autorecovery is attempted, and the validators do less work.

Simplified the checking of generations/transaction ids during sync. Invalid values now always result in errors, no autorecovery is attempted, and the validators do less work.

[Samuele Pedroni (pedronis)]

review: Approve

=== modified file 'include/u1db/u1db_internal.h'
--- include/u1db/u1db_internal.h	2012-07-03 15:48:59 +0000
+++ include/u1db/u1db_internal.h	2012-07-06 17:55:21 +0000
@@ -223,9 +223,7 @@
  *     superseded.
  */
 int u1db__validate_source(u1database *db, const char *replica_uid,
-                          int replica_gen, const char *replica_trans_id,
-                          u1db_vectorclock *cur_vcr,
-                          u1db_vectorclock *other_vcr, int *state);
+                          int replica_gen, const char *replica_trans_id);
 
 /**
  * Internal API, Get the global database rev.
@@ -239,6 +237,12 @@
                               char **trans_id);
 
 /**
+ * Internal API, Get the transaction id for the db generation.
+ */
+int u1db__get_trans_id_for_gen(u1database *db, int generation,
+                               char **trans_id);
+
+/**
  * Internal API, Validate generation and transaction id.
  */
 int u1db_validate_gen_and_trans_id(u1database *db, int generation,

=== modified file 'src/u1db.c'
--- src/u1db.c	2012-07-03 15:48:59 +0000
+++ src/u1db.c	2012-07-06 17:55:21 +0000
@@ -42,7 +42,6 @@
                              char **doc_rev);
 static int generate_transaction_id(char buf[35]);
 
-
 static int
 initialize(u1database *db)
 {
@@ -756,23 +755,17 @@
 
 int
 u1db__validate_source(u1database *db, const char *replica_uid, int replica_gen,
-                      const char *replica_trans_id, u1db_vectorclock *cur,
-                      u1db_vectorclock *other, int *state)
+                      const char *replica_trans_id)
 {
     int old_generation;
     char *old_trans_id = NULL;
     int status = U1DB_OK;
 
-    *state = U1DB_OK;
     status = u1db__get_replica_gen_and_trans_id(
         db, replica_uid, &old_generation, &old_trans_id);
     if (status != U1DB_OK)
         goto finish;
     if (replica_gen < old_generation) {
-        if (u1db__vectorclock_is_newer(cur, other)) {
-            *state = U1DB_SUPERSEDED;
-            goto finish;
-        }
         status = U1DB_INVALID_GENERATION;
         goto finish;
     }
@@ -782,7 +775,6 @@
         status = U1DB_INVALID_TRANSACTION_ID;
         goto finish;
     }
-    *state = U1DB_SUPERSEDED;
 finish:
     if (old_trans_id != NULL)
         free(old_trans_id);
@@ -830,15 +822,10 @@
     }
     if (replica_uid != NULL && replica_trans_id != NULL) {
         status = u1db__validate_source(
-            db, replica_uid, replica_gen, replica_trans_id, stored_vc, new_vc,
-            state);
+            db, replica_uid, replica_gen, replica_trans_id);
         if (status != U1DB_OK) {
             goto finish;
         }
-        if (*state != U1DB_OK) {
-            status = u1db__get_generation(db, at_gen);
-            goto finish;
-        }
     }
     if (stored_doc_rev == NULL) {
         status = U1DB_OK;
@@ -1469,9 +1456,51 @@
             *trans_id = strdup(tmp);
             if (*trans_id == NULL) {
                 status = U1DB_NOMEM;
-            }
-        }
-    }
+                goto finish;
+            }
+        }
+        status = U1DB_OK;
+    }
+finish:
+    sqlite3_finalize(statement);
+    return status;
+}
+
+int
+u1db__get_trans_id_for_gen(u1database *db, int generation,
+                           char **trans_id)
+{
+    int status = U1DB_OK;
+    sqlite3_stmt *statement;
+    const char *tmp;
+
+    if (db == NULL) {
+        return U1DB_INVALID_PARAMETER;
+    }
+    status = sqlite3_prepare_v2(db->sql_handle,
+        "SELECT transaction_id FROM transaction_log WHERE generation = ?", -1,
+        &statement, NULL);
+    if (status != SQLITE_OK) { goto finish; }
+    status = sqlite3_bind_int(statement, 1, generation);
+    if (status != SQLITE_OK) { goto finish; }
+    status = sqlite3_step(statement);
+    if (status == SQLITE_DONE) {
+        status = U1DB_INVALID_GENERATION;
+        goto finish;
+    } else if (status == SQLITE_ROW) {
+        tmp = (const char *)sqlite3_column_text(statement, 0);
+        if (tmp == NULL) {
+            *trans_id = NULL;
+        } else {
+            *trans_id = strdup(tmp);
+            if (*trans_id == NULL) {
+                status = U1DB_NOMEM;
+                goto finish;
+            }
+        }
+        status = U1DB_OK;
+    }
+finish:
     sqlite3_finalize(statement);
     return status;
 }
@@ -1481,34 +1510,21 @@
                                const char *trans_id)
 {
     int status = U1DB_OK;
-    sqlite3_stmt *statement;
+    char *known_trans_id = NULL;
 
     if (generation == 0)
         return status;
-    if (db == NULL) {
-        return U1DB_INVALID_PARAMETER;
-    }
-    status = sqlite3_prepare_v2(db->sql_handle,
-        "SELECT transaction_id FROM transaction_log WHERE generation = ?", -1,
-        &statement, NULL);
-    if (status != SQLITE_OK) { goto finish; }
-    status = sqlite3_bind_int(statement, 1, generation);
-    if (status != SQLITE_OK) { goto finish; }
-    status = sqlite3_step(statement);
-    if (status == SQLITE_DONE) {
-        status = U1DB_INVALID_GENERATION;
+    status = u1db__get_trans_id_for_gen(db, generation, &known_trans_id);
+    if (status != U1DB_OK)
         goto finish;
-    } else if (status == SQLITE_ROW) {
-        // Note: We may want to handle the column containing NULL
-        if (strcmp(trans_id,
-                   (const char *)sqlite3_column_text(statement, 0)) == 0) {
+    if (strcmp(trans_id, known_trans_id) == 0) {
             status = U1DB_OK;
             goto finish;
         }
-        status = U1DB_INVALID_TRANSACTION_ID;
-    }
+    status = U1DB_INVALID_TRANSACTION_ID;
 finish:
-    sqlite3_finalize(statement);
+    if (known_trans_id != NULL)
+        free(known_trans_id);
     return status;
 }
 

=== modified file 'src/u1db_sync_target.c'
--- src/u1db_sync_target.c	2012-07-03 15:48:59 +0000
+++ src/u1db_sync_target.c	2012-07-06 17:55:21 +0000
@@ -105,8 +105,8 @@
 
 static int
 st_get_sync_info(u1db_sync_target *st, const char *source_replica_uid,
-        const char **st_replica_uid, int *st_gen, int *source_gen,
-        char **source_trans_id)
+                 const char **st_replica_uid, int *st_gen, int *source_gen,
+                 char **source_trans_id)
 {
     int status = U1DB_OK;
     u1database *db;
@@ -614,8 +614,9 @@
     status = u1db_validate_gen_and_trans_id(
         db, local_gen_known_by_target, local_trans_id_known_by_target);
     if (status != U1DB_OK) { goto finish; }
-    status = u1db__get_replica_gen_and_trans_id(db, target_uid,
-        &target_gen_known_by_local, &target_trans_id_known_by_local);
+    status = u1db__get_replica_gen_and_trans_id(
+        db, target_uid, &target_gen_known_by_local,
+        &target_trans_id_known_by_local);
     if (status != U1DB_OK) { goto finish; }
     local_target_trans_id = target_trans_id_known_by_local;
     local_gen = local_gen_known_by_target;

=== modified file 'u1db/__init__.py'
--- u1db/__init__.py	2012-07-03 15:48:59 +0000
+++ u1db/__init__.py	2012-07-06 17:55:21 +0000
@@ -290,7 +290,7 @@
             encountered during synchronization. If we've never synchronized
             with the replica, this is (0, '').
         """
-        raise NotImplementedError(self._replica_gen_and_trans_id)
+        raise NotImplementedError(self._get_replica_gen_and_trans_id)
 
     def _set_replica_gen_and_trans_id(self, other_replica_uid,
                                       other_generation, other_transaction_id):

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

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

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

=== modified file 'u1db/sync.py'
--- u1db/sync.py	2012-07-03 15:48:59 +0000
+++ u1db/sync.py	2012-07-06 17:55:21 +0000
@@ -97,18 +97,17 @@
         (self.target_replica_uid, target_gen, target_my_gen,
          target_my_trans_id) = sync_target.get_sync_info(
              self.source._replica_uid)
-        # validate that the generation and transaction id the target knows
-        # about us are valid.
+        # validate the generation and transaction id the target knows about us
         self.source.validate_gen_and_trans_id(
             target_my_gen, target_my_trans_id)
         # what's changed since that generation and this current gen
         my_gen, _, changes = self.source.whats_changed(target_my_gen)
 
         # this source last-seen database generation for the target
-        (target_last_known_gen,
-         target_trans_id) = self.source._get_replica_gen_and_trans_id(
-             self.target_replica_uid)
+        target_last_known_gen, target_last_known_trans_id = \
+            self.source._get_replica_gen_and_trans_id(self.target_replica_uid)
         if not changes and target_last_known_gen == target_gen:
+            # TODO: we'll need to check the target's transaction id as well.
             return my_gen
         changed_doc_ids = [doc_id for doc_id, _, _ in changes]
         # prepare to send all the changed docs
@@ -123,7 +122,7 @@
         # the target, return target synced-up-to gen
         new_gen, new_trans_id = sync_target.sync_exchange(
             docs_by_generation, self.source._replica_uid,
-            target_last_known_gen, target_trans_id,
+            target_last_known_gen, target_last_known_trans_id,
             self._insert_doc_from_target)
         # record target synced-up-to generation including applying what we sent
         self.source._set_replica_gen_and_trans_id(

=== modified file 'u1db/tests/c_backend_wrapper.pyx'
--- u1db/tests/c_backend_wrapper.pyx	2012-07-03 15:48:59 +0000
+++ u1db/tests/c_backend_wrapper.pyx	2012-07-06 17:55:21 +0000
@@ -81,9 +81,7 @@
                           void *context, u1db_doc_callback cb)
     int u1db_put_doc(u1database *db, u1db_document *doc)
     int u1db__validate_source(u1database *db, const_char_ptr replica_uid,
-                              int replica_gen, const_char_ptr replica_trans_id,
-                              u1db_vectorclock *cur_vcr,
-                              u1db_vectorclock *other_vcr, int *state)
+                              int replica_gen, const_char_ptr replica_trans_id)
     int u1db__put_doc_if_newer(u1database *db, u1db_document *doc,
                                int save_conflict, char *replica_uid,
                                int replica_gen, char *replica_trans_id,
@@ -206,6 +204,7 @@
 
     int u1db__get_generation(u1database *, int *db_rev)
     int u1db__get_generation_info(u1database *, int *db_rev, char **trans_id)
+    int u1db__get_trans_id_for_gen(u1database *, int db_rev, char **trans_id)
     int u1db_validate_gen_and_trans_id(u1database *, int db_rev,
                                        const_char_ptr trans_id)
     char *u1db__allocate_doc_id(u1database *)
@@ -941,26 +940,16 @@
             u1db_put_doc(self._db, doc._doc))
         return doc.rev
 
-    def _validate_source(self, replica_uid, replica_gen, replica_trans_id,
-                         cur_vcr, other_vcr):
+    def _validate_source(self, replica_uid, replica_gen, replica_trans_id):
         cdef const_char_ptr c_uid, c_trans_id
-        cdef int c_gen, state = 0
-        cdef VectorClockRev cur
-        cdef VectorClockRev other
+        cdef int c_gen = 0
 
-        cur = VectorClockRev(cur_vcr.as_str())
-        other = VectorClockRev(other_vcr.as_str())
         c_uid = replica_uid
         c_trans_id = replica_trans_id
         c_gen = replica_gen
         handle_status(
             "invalid generation or transaction id",
-            u1db__validate_source(
-                self._db, c_uid, c_gen, c_trans_id, cur._clock, other._clock,
-                &state))
-        if state == U1DB_SUPERSEDED:
-            return 'superseded'
-        return 'ok'
+            u1db__validate_source(self._db, c_uid, c_gen, c_trans_id))
 
     def _put_doc_if_newer(self, CDocument doc, save_conflict, replica_uid=None,
                           replica_gen=None, replica_trans_id=None):
@@ -1100,6 +1089,18 @@
             "validate_gen_and_trans_id",
             u1db_validate_gen_and_trans_id(self._db, generation, trans_id))
 
+    def _get_trans_id_for_gen(self, generation):
+        cdef char *trans_id = NULL
+
+        handle_status(
+            "_get_trans_id_for_gen",
+            u1db__get_trans_id_for_gen(self._db, generation, &trans_id))
+        raw_trans_id = None
+        if trans_id != NULL:
+            raw_trans_id = trans_id
+            free(trans_id)
+        return raw_trans_id
+
     def _get_replica_gen_and_trans_id(self, replica_uid):
         cdef int generation, status
         cdef char *trans_id = NULL

=== modified file 'u1db/tests/test_backends.py'
--- u1db/tests/test_backends.py	2012-07-04 15:43:00 +0000
+++ u1db/tests/test_backends.py	2012-07-06 17:55:21 +0000
@@ -402,11 +402,12 @@
 
     def test_put_doc_if_newer_same_generation_same_txid(self):
         self.db._set_replica_gen_and_trans_id('other', 1, 'T-sid')
-        doc = self.make_document('doc_id', 'other:2', simple_doc)
+        doc = self.db.create_doc(simple_doc)
+        doc2 = self.make_document(doc.doc_id, 'other:1', simple_doc)
         state, _ = self.db._put_doc_if_newer(
             doc, save_conflict=False, replica_uid='other', replica_gen=1,
             replica_trans_id='T-sid')
-        self.assertEqual('superseded', state)
+        self.assertEqual('converged', state)
 
     def test_put_doc_if_newer_wrong_transaction_id(self):
         self.db._set_replica_gen_and_trans_id('other', 1, 'T-sid')
@@ -422,10 +423,10 @@
         doc_rev1 = doc.rev
         doc.set_json(simple_doc)
         self.db.put_doc(doc)
-        self.db._set_replica_gen_and_trans_id('other', 5, 'T-sid')
+        self.db._set_replica_gen_and_trans_id('other', 3, 'T-sid')
         older_doc = self.make_document(doc.doc_id, doc_rev1, simple_doc)
         state, _ = self.db._put_doc_if_newer(
-            older_doc, save_conflict=False, replica_uid='other', replica_gen=3,
+            older_doc, save_conflict=False, replica_uid='other', replica_gen=8,
             replica_trans_id='T-irrelevant')
         self.assertEqual('superseded', state)
 
@@ -557,45 +558,17 @@
 
     def test_validate_source_gen_and_trans_id_same(self):
         self.db._set_replica_gen_and_trans_id('other', 1, 'T-sid')
-        v1 = vectorclock.VectorClockRev('other:1|self:1')
-        v2 = vectorclock.VectorClockRev('other:1|self:1')
-        self.assertEqual(
-            'superseded',
-            self.db._validate_source('other', 1, 'T-sid', v1, v2))
+        self.db._validate_source('other', 1, 'T-sid')
 
     def test_validate_source_gen_newer(self):
         self.db._set_replica_gen_and_trans_id('other', 1, 'T-sid')
-        v1 = vectorclock.VectorClockRev('other:1|self:1')
-        v2 = vectorclock.VectorClockRev('other:2|self:2')
-        self.assertEqual(
-            'ok',
-            self.db._validate_source('other', 2, 'T-whatevs', v1, v2))
+        self.db._validate_source('other', 2, 'T-whatevs')
 
     def test_validate_source_wrong_txid(self):
         self.db._set_replica_gen_and_trans_id('other', 1, 'T-sid')
-        v1 = vectorclock.VectorClockRev('other:1|self:1')
-        v2 = vectorclock.VectorClockRev('other:2|self:2')
         self.assertRaises(
             errors.InvalidTransactionId,
-            self.db._validate_source, 'other', 1, 'T-sad', v1, v2)
-
-    def test_validate_source_gen_older_and_vcr_older(self):
-        self.db._set_replica_gen_and_trans_id('other', 1, 'T-sid')
-        self.db._set_replica_gen_and_trans_id('other', 2, 'T-sod')
-        v1 = vectorclock.VectorClockRev('other:1|self:1')
-        v2 = vectorclock.VectorClockRev('other:2|self:2')
-        self.assertEqual(
-            'superseded',
-            self.db._validate_source('other', 1, 'T-sid', v2, v1))
-
-    def test_validate_source_gen_older_vcr_newer(self):
-        self.db._set_replica_gen_and_trans_id('other', 1, 'T-sid')
-        self.db._set_replica_gen_and_trans_id('other', 2, 'T-sod')
-        v1 = vectorclock.VectorClockRev('other:1|self:1')
-        v2 = vectorclock.VectorClockRev('other:2|self:2')
-        self.assertRaises(
-            errors.InvalidGeneration,
-            self.db._validate_source, 'other', 1, 'T-sid', v1, v2)
+            self.db._validate_source, 'other', 1, 'T-sad')
 
 
 class LocalDatabaseWithConflictsTests(tests.DatabaseBaseTests):

=== modified file 'u1db/tests/test_sync.py'
--- u1db/tests/test_sync.py	2012-07-04 16:03:39 +0000
+++ u1db/tests/test_sync.py	2012-07-06 17:55:21 +0000
@@ -424,7 +424,8 @@
     return db
 
 
-def sync_via_synchronizer_and_http(test, db_source, db_target, trace_hook=None):
+def sync_via_synchronizer_and_http(test, db_source, db_target,
+                                   trace_hook=None):
     if trace_hook:
         test.skipTest("trace_hook unsupported over http")
     path = test._http_at[db_target]
@@ -988,6 +989,27 @@
         self.assertRaises(
             errors.InvalidTransactionId, self.sync, self.db1, db3)
 
+    def test_sync_detects_rollback_and_divergence_in_source(self):
+        self.db1.create_doc(tests.simple_doc, doc_id="divergent")
+        self.sync(self.db1, self.db2)
+        self.db1.create_doc(tests.simple_doc)
+        self.db2._set_replica_gen_and_trans_id(
+            self.db1._replica_uid, 2, 'T-madeup')
+        self.assertRaises(
+            errors.InvalidTransactionId, self.sync, self.db1, self.db2)
+
+    def test_sync_detects_rollback_and_divergence_in_target(self):
+        # TODO: reenable this once we check the trans_id in Synchronizer.sync
+        self.skip("TODO: check target_trans_id")
+        self.db1.create_doc(tests.simple_doc, doc_id="divergent")
+        self.sync(self.db1, self.db2)
+        self.db2.create_doc(tests.simple_doc)
+        self.db1._set_replica_gen_and_trans_id(
+            self.db2._replica_uid, 2, 'T-madeup')
+        self.sync(self.db1, self.db2)
+        self.assertRaises(
+            errors.InvalidTransactionId, self.sync, self.db1, self.db2)
+
 
 class TestDbSync(tests.TestCaseWithServer):
     """Test db.sync remote sync shortcut"""
@@ -1032,7 +1054,8 @@
         self.assertEqual(2, len(self.db2._get_transaction_log()))
         progress1 = []
         progress2 = []
-        _do_set_replica_gen_and_trans_id = self.db1._do_set_replica_gen_and_trans_id
+        _do_set_replica_gen_and_trans_id = \
+            self.db1._do_set_replica_gen_and_trans_id
 
         def set_sync_generation_witness1(other_uid, other_gen, trans_id):
             progress1.append((other_uid, other_gen,
@@ -1040,8 +1063,8 @@
             _do_set_replica_gen_and_trans_id(other_uid, other_gen, trans_id)
         self.patch(self.db1, '_do_set_replica_gen_and_trans_id',
                    set_sync_generation_witness1)
-
-        _do_set_replica_gen_and_trans_id2 = self.db2._do_set_replica_gen_and_trans_id
+        _do_set_replica_gen_and_trans_id2 = \
+            self.db2._do_set_replica_gen_and_trans_id
 
         def set_sync_generation_witness2(other_uid, other_gen, trans_id):
             progress2.append((other_uid, other_gen,