Merge lp:~jameinel/u1db/combined-put-and-gen into lp:u1db

[John A Meinel]

This updates Database.put_doc_if_newer to allow passing in the other replica_uid and replica_generation. It also updates SyncExchange.insert_doc_from_source to take those arguments and pass them on. And then it updates the HTTP sync code to just pass the arguments rather than calling set_sync_generation directly.

The main benefit is that we have 1 transaction per document, rather than 2.

It should be pretty easy to update insert_doc_from_target as well. It doesn't look like we need to update force_doc_conflict after all, because it is always called *after* put_doc_if_newer, so we'll have already handled the generation stuff.

Updated to merge the removal of force_doc_sync_conflict and clean up the api a bit.

Also updated to include target => source updates.

One thing I'm wondering, we currently have the thunk take_doc in order to pass the other_replica_uid to _insert_doc_from_target. We didn't need to do that for the Source => Target side, because we could use Source._replica_uid. However we don't have sync_target._replica_uid to do the same for the other direction.

I wonder if we want to change the callback api, instead of being (doc, gen) to be (doc, replica_uid, replica_gen).
Thoughts?

[Samuele Pedroni (pedronis)]

looks good

review: Approve

=== modified file 'u1db/__init__.py'
--- u1db/__init__.py	2011-12-16 11:55:07 +0000
+++ u1db/__init__.py	2011-12-16 12:23:24 +0000
@@ -102,7 +102,8 @@
         """
         raise NotImplementedError(self.put_doc)
 
-    def put_doc_if_newer(self, doc, save_conflict):
+    def put_doc_if_newer(self, doc, save_conflict, replica_uid=None,
+                         replica_gen=None):
         """Insert/update document into the database with a given revision.
 
         This api is used during synchronization operations.
@@ -123,6 +124,10 @@
         :param doc: A Document object
         :param save_conflict: If this document is a conflict, do you want to
             save it as a conflict, or just ignore it.
+        :param replica_uid: A unique replica identifier.
+        :param replica_gen: The generation of the replica corresponding to the
+            this document. The replica arguments are optional, but are used
+            during synchronization.
         :return: state -  If we don't have doc_id already, or if doc_rev
             supersedes the existing document revision, then the content will
             be inserted, and state is 'inserted'.

=== modified file 'u1db/backends/__init__.py'
--- u1db/backends/__init__.py	2011-12-16 11:55:07 +0000
+++ u1db/backends/__init__.py	2011-12-16 12:23:24 +0000
@@ -85,7 +85,8 @@
             result.append(doc)
         return result
 
-    def put_doc_if_newer(self, doc, save_conflict):
+    def put_doc_if_newer(self, doc, save_conflict, replica_uid=None,
+                         replica_gen=None):
         cur_doc = self._get_doc(doc.doc_id)
         doc_vcr = VectorClockRev(doc.rev)
         if cur_doc is None:
@@ -95,20 +96,22 @@
         if doc_vcr.is_newer(cur_vcr):
             self._put_and_update_indexes(cur_doc, doc)
             self._prune_conflicts(doc, doc_vcr)
-            return 'inserted'
+            state = 'inserted'
         elif doc.rev == cur_doc.rev:
             # magical convergence
-            return 'converged'
+            state = 'converged'
         elif cur_vcr.is_newer(doc_vcr):
             # Don't add this to seen_ids, because we have something newer,
             # so we should send it back, and we should not generate a
             # conflict
-            return 'superseded'
+            state = 'superseded'
         else:
             state = 'conflicted'
             if save_conflict:
                 self._force_doc_sync_conflict(doc)
-            return 'conflicted'
+        if replica_uid is not None and replica_gen is not None:
+            self._set_sync_generation(replica_uid, replica_gen)
+        return state
 
     def _ensure_maximal_rev(self, cur_rev, extra_revs):
         vcr = VectorClockRev(cur_rev)

=== modified file 'u1db/backends/inmemory.py'
--- u1db/backends/inmemory.py	2011-12-16 11:19:34 +0000
+++ u1db/backends/inmemory.py	2011-12-16 12:23:24 +0000
@@ -47,6 +47,9 @@
         return self._other_generations.get(other_replica_uid, 0)
 
     def set_sync_generation(self, other_replica_uid, other_generation):
+        self._set_sync_generation(other_replica_uid, other_generation)
+
+    def _set_sync_generation(self, other_replica_uid, other_generation):
         # TODO: to handle race conditions, we may want to check if the current
         #       value is greater than this new value.
         self._other_generations[other_replica_uid] = other_generation

=== modified file 'u1db/backends/sqlite_backend.py'
--- u1db/backends/sqlite_backend.py	2011-12-16 11:55:07 +0000
+++ u1db/backends/sqlite_backend.py	2011-12-16 12:23:24 +0000
@@ -420,15 +420,19 @@
 
     def set_sync_generation(self, other_replica_uid, other_generation):
         with self._db_handle:
+            self._set_sync_generation(other_replica_uid, other_generation)
+
+    def _set_sync_generation(self, other_replica_uid, other_generation):
             c = self._db_handle.cursor()
-            my_gen = self._get_generation()
             c.execute("INSERT OR REPLACE INTO sync_log VALUES (?, ?)",
                       (other_replica_uid, other_generation))
 
-    def put_doc_if_newer(self, doc, save_conflict):
+    def put_doc_if_newer(self, doc, save_conflict, replica_uid=None,
+                         replica_gen=None):
         with self._db_handle:
             return super(SQLiteDatabase, self).put_doc_if_newer(doc,
-                save_conflict=save_conflict)
+                save_conflict=save_conflict,
+                replica_uid=replica_uid, replica_gen=replica_gen)
 
     def _add_conflict(self, c, doc_id, my_doc_rev, my_content):
         c.execute("INSERT INTO conflicts VALUES (?, ?, ?)",

=== modified file 'u1db/remote/http_app.py'
--- u1db/remote/http_app.py	2011-12-14 15:37:05 +0000
+++ u1db/remote/http_app.py	2011-12-16 12:23:24 +0000
@@ -290,8 +290,7 @@
     @http_method(content_as_args=True)
     def post_stream_entry(self, id, rev, content, gen):
         doc = Document(id, rev, content)
-        self.sync_exch.insert_doc_from_source(doc)
-        self.sync_exch.record_sync_progress(self.from_replica_uid, gen)
+        self.sync_exch.insert_doc_from_source(doc, self.from_replica_uid, gen)
 
     def post_end(self):
         def send_doc(doc, gen):

=== modified file 'u1db/sync.py'
--- u1db/sync.py	2011-12-16 11:55:07 +0000
+++ u1db/sync.py	2011-12-16 12:23:24 +0000
@@ -36,7 +36,7 @@
         self.sync_target = sync_target
         self.num_inserted = 0
 
-    def _insert_doc_from_target(self, doc):
+    def _insert_doc_from_target(self, doc, replica_uid, replica_gen):
         """Try to insert synced document from target.
 
         Implements TAKE OTHER semantics: any document from the target
@@ -48,7 +48,8 @@
         """
         # Increases self.num_inserted depending whether the document
         # was effectively inserted.
-        state = self.source.put_doc_if_newer(doc, save_conflict=True)
+        state = self.source.put_doc_if_newer(doc, save_conflict=True,
+            replica_uid=replica_uid, replica_gen=replica_gen)
         if state == 'inserted':
             self.num_inserted += 1
         elif state == 'converged':
@@ -103,9 +104,7 @@
         # exchange documents and try to insert the returned ones with
         # the target, return target synced-up-to gen
         def take_doc(doc, gen):
-            self._insert_doc_from_target(doc)
-            # record target synced-up-to generation
-            self.source.set_sync_generation(other_replica_uid, gen)
+            return self._insert_doc_from_target(doc, other_replica_uid, gen)
 
         new_gen = sync_target.sync_exchange(docs_by_generation,
                         self.source._replica_uid, other_last_known_gen,
@@ -134,7 +133,7 @@
             'return': None
             }
 
-    def insert_doc_from_source(self, doc):
+    def insert_doc_from_source(self, doc, other_uid, other_gen):
         """Try to insert synced document from source.
 
         Conflicting documents are not inserted but will be sent over
@@ -146,7 +145,8 @@
         :param doc: A Document object.
         :return: None
         """
-        state = self._db.put_doc_if_newer(doc, save_conflict=False)
+        state = self._db.put_doc_if_newer(doc, save_conflict=False,
+            replica_uid=other_uid, replica_gen=other_gen)
         if state == 'inserted':
             self.seen_ids.add(doc.doc_id)
         elif state == 'converged':
@@ -241,8 +241,8 @@
                       last_known_generation, return_doc_cb):
         sync_exch = self.get_sync_exchange()
         # 1st step: try to insert incoming docs
-        for doc, _ in docs_by_generations:
-            sync_exch.insert_doc_from_source(doc)
+        for doc, doc_gen in docs_by_generations:
+            sync_exch.insert_doc_from_source(doc, from_replica_uid, doc_gen)
         # record progress
         if docs_by_generations:
             latest_gen = docs_by_generations[-1][1]

=== modified file 'u1db/tests/test_backends.py'
--- u1db/tests/test_backends.py	2011-12-16 11:55:07 +0000
+++ u1db/tests/test_backends.py	2011-12-16 12:23:24 +0000
@@ -315,6 +315,30 @@
         # The database wasn't altered
         self.assertGetDoc(self.db, doc1.doc_id, doc1.rev, simple_doc, False)
 
+    def test_put_doc_if_newer_replica_uid(self):
+        doc1 = self.db.create_doc(simple_doc)
+        self.db.set_sync_generation('other', 1)
+        doc2 = Document(doc1.doc_id, doc1.rev + '|other:1', nested_doc)
+        self.assertEqual('inserted',
+            self.db.put_doc_if_newer(doc2, save_conflict=True,
+                replica_uid='other', replica_gen=2))
+        self.assertEqual(2, self.db.get_sync_generation('other'))
+        doc2.rev = doc1.rev
+        self.assertEqual('superseded',
+            self.db.put_doc_if_newer(doc2, save_conflict=True,
+                replica_uid='other', replica_gen=3))
+        self.assertEqual(3, self.db.get_sync_generation('other'))
+        doc2.rev = doc1.rev + '|third:3'
+        self.assertEqual('conflicted',
+            self.db.put_doc_if_newer(doc2, save_conflict=True,
+                replica_uid='other', replica_gen=4))
+        self.assertEqual(4, self.db.get_sync_generation('other'))
+        doc2.rev = doc1.rev + '|fourth:1'
+        self.assertEqual('conflicted',
+            self.db.put_doc_if_newer(doc2, save_conflict=False,
+                replica_uid='other', replica_gen=5))
+        self.assertEqual(5, self.db.get_sync_generation('other'))
+
     def test_put_doc_if_newer_save_conflicted(self):
         doc1 = self.db.create_doc(simple_doc)
         # Document is inserted as a conflict

=== modified file 'u1db/tests/test_http_app.py'
--- u1db/tests/test_http_app.py	2011-12-15 19:43:40 +0000
+++ u1db/tests/test_http_app.py	2011-12-16 12:23:24 +0000
@@ -562,7 +562,7 @@
             }
 
         gens = []
-        _set_sync_generation = self.db0.set_sync_generation
+        _set_sync_generation = self.db0._set_sync_generation
         def set_sync_generation_witness(other_uid, other_gen):
             gens.append((other_uid, other_gen))
             _set_sync_generation(other_uid, other_gen)
@@ -570,7 +570,7 @@
                               entries[other_gen]['rev'],
                               entries[other_gen]['content'], False)
 
-        self.patch(self.db0, 'set_sync_generation',
+        self.patch(self.db0, '_set_sync_generation',
                    set_sync_generation_witness)
 
         args = dict(last_known_generation=0)

=== modified file 'u1db/tests/test_remote_sync_target.py'
--- u1db/tests/test_remote_sync_target.py	2011-12-16 11:55:07 +0000
+++ u1db/tests/test_remote_sync_target.py	2011-12-16 12:23:24 +0000
@@ -109,10 +109,12 @@
         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):
+        def bomb_put_doc_if_newer(doc, save_conflict,
+                                  replica_uid=None, replica_gen=None):
             if doc.doc_id in trigger_ids:
                 raise Exception
-            return _put_doc_if_newer(doc, save_conflict)
+            return _put_doc_if_newer(doc, save_conflict=save_conflict,
+                replica_uid=replica_uid, replica_gen=replica_gen)
         self.patch(db, 'put_doc_if_newer', bomb_put_doc_if_newer)
         remote_target = self.getSyncTarget('test')
         other_changes = []

=== modified file 'u1db/tests/test_sync.py'
--- u1db/tests/test_sync.py	2011-12-15 19:41:40 +0000
+++ u1db/tests/test_sync.py	2011-12-16 12:23:24 +0000
@@ -523,20 +523,20 @@
         self.assertEqual(2, len(self.db2._get_transaction_log()))
         progress1 = []
         progress2 = []
-        _set_sync_generation1 = self.db1.set_sync_generation
+        _set_sync_generation1 = self.db1._set_sync_generation
         def set_sync_generation_witness1(other_uid, other_gen):
             progress1.append((other_uid, other_gen,
                               self.db1._get_transaction_log()[2:]))
             _set_sync_generation1(other_uid, other_gen)
-        self.patch(self.db1, 'set_sync_generation',
+        self.patch(self.db1, '_set_sync_generation',
                    set_sync_generation_witness1)
 
-        _set_sync_generation2 = self.db2.set_sync_generation
+        _set_sync_generation2 = self.db2._set_sync_generation
         def set_sync_generation_witness2(other_uid, other_gen):
             progress2.append((other_uid, other_gen,
                               self.db2._get_transaction_log()[2:]))
             _set_sync_generation2(other_uid, other_gen)
-        self.patch(self.db2, 'set_sync_generation',
+        self.patch(self.db2, '_set_sync_generation',
                    set_sync_generation_witness2)
 
         db2_url = self.getURL('test2')