Merge lp:~pedronis/u1db/sync-simple-reorg-n-hide-details into lp:u1db

[Samuele Pedroni]

some things I think are necessary for incremental sync:

- don't send/treat conflicts separately if the plan is to send back documents sorted somehow by generation of changes

- make recording the source last known generation its own helper if we want to call it repeatedly for the generation of incoming docs

[John A Meinel (jameinel)]

I like this in general. I'm a little concerned about not tracking conflicts
separately. It might just be paranoia, though. So I'm trying to think
through it.

A-1 doc-x created
B-1 doc-x sync
A-2 x updates rev-A:2
B-2 x updated rev-A:1¦B:1 (maybe B:2 for gen)
A starts sync, sends doc-x A:2 gen 2
B skips the doc, does(n't?) record gen of A
B tries to return x@A:1¦B:1 but loses connection
A-3 doc-y
A starts sync, B wants everything since A-2
A sends y, and last B gen of 1
B will naturally return x@A:1¦B:1

So all of that seems fine. Can we poke holes in it to make it fail? It does
seem that anything which should be in conflict must have been updated since
the last successful synchronization.

So it seems ok to me, but we'll want to think carefully about edge cases.

John
=:->
On Dec 13, 2011 11:02 AM, "Samuele Pedroni" <email address hidden>
wrote:

> Samuele Pedroni has proposed merging
> lp:~pedronis/u1db/sync-simple-reorg-n-hide-details into lp:u1db.
>
> Requested reviews:
> Ubuntu One hackers (ubuntuone-hackers)
>
> For more details, see:
>
> https://code.launchpad.net/~pedronis/u1db/sync-simple-reorg-n-hide-details/+merge/85453
>
> some things I think are necessary for incremental sync:
>
> - don't send/treat conflicts separately if the plan is to send back
> documents sorted somehow by generation of changes
>
> - make recording the source last known generation its own helper if we
> want to call it repeatedly for the generation of incoming docs
> --
>
> https://code.launchpad.net/~pedronis/u1db/sync-simple-reorg-n-hide-details/+merge/85453
> Your team Ubuntu One hackers is requested to review the proposed merge of
> lp:~pedronis/u1db/sync-simple-reorg-n-hide-details into lp:u1db.
>
> === modified file 'u1db/remote/http_app.py'
> --- u1db/remote/http_app.py 2011-12-06 20:32:48 +0000
> +++ u1db/remote/http_app.py 2011-12-13 09:59:32 +0000
> @@ -294,16 +294,15 @@
> self.sync_exch.insert_doc_from_source(doc)
>
> def post_end(self):
> + self.sync_exch.record_sync_progress(self.from_replica_uid,
> + self.from_replica_generation)
> def send_doc(doc):
> entry = dict(id=doc.doc_id, rev=doc.rev, content=doc.content)
> self.responder.stream_entry(entry)
> new_gen =
> self.sync_exch.find_docs_to_return(self.last_known_generation)
> self.responder.content_type = 'application/x-u1db-multi-json'
> self.responder.start_response(200, {"new_generation": new_gen})
> - new_gen = self.sync_exch.return_docs_and_record_sync(
> - self.from_replica_uid,
> -
> self.from_replica_generation,
> - send_doc)
> + new_gen = self.sync_exch.return_docs(send_doc)
> self.responder.finish_response()
>
>
>
> === modified file 'u1db/sync.py'
> --- u1db/sync.py 2011-12-12 19:40:50 +0000
> +++ u1db/sync.py 2011-12-13 09:59:32 +0000
> @@ -120,19 +120,21 @@
>
> def __init__(self, db):
> self._db = db
> - self.seen_ids = set() # incoming ids not superseded (or
> conflicted)
> + self.seen_id...

=== modified file 'u1db/remote/http_app.py'
--- u1db/remote/http_app.py	2011-12-06 20:32:48 +0000
+++ u1db/remote/http_app.py	2011-12-13 09:59:32 +0000
@@ -294,16 +294,15 @@
         self.sync_exch.insert_doc_from_source(doc)
 
     def post_end(self):
+        self.sync_exch.record_sync_progress(self.from_replica_uid,
+                                            self.from_replica_generation)
         def send_doc(doc):
             entry = dict(id=doc.doc_id, rev=doc.rev, content=doc.content)
             self.responder.stream_entry(entry)
         new_gen = self.sync_exch.find_docs_to_return(self.last_known_generation)
         self.responder.content_type = 'application/x-u1db-multi-json'
         self.responder.start_response(200, {"new_generation": new_gen})
-        new_gen = self.sync_exch.return_docs_and_record_sync(
-                                                  self.from_replica_uid,
-                                                  self.from_replica_generation,
-                                                  send_doc)
+        new_gen = self.sync_exch.return_docs(send_doc)
         self.responder.finish_response()
 
 

=== modified file 'u1db/sync.py'
--- u1db/sync.py	2011-12-12 19:40:50 +0000
+++ u1db/sync.py	2011-12-13 09:59:32 +0000
@@ -120,19 +120,21 @@
 
     def __init__(self, db):
         self._db = db
-        self.seen_ids = set()  # incoming ids not superseded (or conflicted)
+        self.seen_ids = set()  # incoming ids not superseded
         self.doc_ids_to_return = None
-        self.conflict_ids = set()
         self.new_gen = None
         # for tests
         self._incoming_trace = []
-        self._last_known_generation = None
+        self._db._last_exchange_log = {
+            'receive': {'docs': self._incoming_trace},
+            'return': None
+            }
 
     def insert_doc_from_source(self, doc):
         """Try to insert synced document from source.
 
-        Conflicting documents are not inserted but the current revision
-        marked to be sent over to the sync source.
+        Conflicting documents are not inserted but will be sent over
+        to the sync source.
 
         The 1st step of a sync exchange is to call this repeatedly to
         try insert all incoming documents from the source.
@@ -150,76 +152,75 @@
             # we have something newer that we will return
             pass
         else:
-            # conflict, returned independently
+            # conflict that we will returne
             assert state == 'conflicted'
-            self.seen_ids.add(doc.doc_id)
-            self.conflict_ids.add(doc.doc_id)
         # for tests
         self._incoming_trace.append((doc.doc_id, doc.rev))
 
+    def record_sync_progress(self, from_replica_uid, from_replica_generation):
+        """Record the sync information of from_replica_uid
+        the sync source identifier and the generation until which it
+        sent its documents from_replica_generation.
+
+        :param from_replica_uid: The source replica's identifier
+        :param from_replica_generation: The db generation for the
+            source replica indicating the tip of data that was sent.
+        :return: None
+        """
+        # record sync point
+        self._db.set_sync_generation(from_replica_uid,
+                                     from_replica_generation)
+        # for tests
+        self._db._last_exchange_log['receive'].update({
+            'from_id': from_replica_uid,
+            'from_gen': from_replica_generation
+            })
+
     def find_docs_to_return(self, last_known_generation):
         """Find and further mark documents to return to the sync source.
 
         This finds the document identifiers for any documents that
         have been updated since last_known_generation. It excludes
-        documents ids that have already been considered (marked conflicts,
-        or superseded by the sender, etc).
+        documents ids that have already been considered
+        (superseded by the sender, etc).
 
         :return: new_generation - the generation of this database
             which the caller can consider themselves to be synchronized after
             processing the returned documents.
         """
-        self._last_known_generation = last_known_generation  # for tests
+        self._db._last_exchange_log['receive'].update({  # for tests
+            'last_known_gen': last_known_generation
+            })
         gen, changes = self._db.whats_changed(last_known_generation)
         changed_doc_ids = set(doc_id for doc_id, _ in changes)
         self.new_gen = gen
         seen_ids = self.seen_ids
         # changed docs that weren't superseded by or converged with
-        # nor conflicted, conflicts are returned independently
         self.doc_ids_to_return = set(doc_id for doc_id in changed_doc_ids
                                      if doc_id not in seen_ids)
         return gen
 
-    def return_docs_and_record_sync(self,
-                                    from_replica_uid, from_replica_generation,
-                                    return_doc_cb):
+    def return_docs(self, return_doc_cb):
         """Return the marked documents repeatedly invoking the callback
-        return_doc_cb, record the sync information of from_replica_uid
-        the sync source identifier and the generation until which it
-        sent its documents from_replica_generation.
+        return_doc_cb.
 
         The final step of a sync exchange.
 
-        :param from_replica_uid: The source replica's identifier
-        :param from_replica_generation: The db generation for the
-            source replica indicating the tip of data that was sent.
         :param: return_doc_cb(doc): is a callback
                 used to return the marked documents to the target replica,
         :return: None
         """
         doc_ids_to_return = self.doc_ids_to_return
-        conflict_ids = self.conflict_ids
-        # return docs
-        new_docs = self._db.get_docs(doc_ids_to_return,
+        # return docs, including conflicts
+        docs = self._db.get_docs(doc_ids_to_return,
                                      check_for_conflicts=False)
-        for doc in new_docs:
-            return_doc_cb(doc)
-        conflicts = self._db.get_docs(conflict_ids, check_for_conflicts=False)
-        for doc in conflicts:
-            return_doc_cb(doc)
-        # record sync point
-        self._db.set_sync_generation(from_replica_uid,
-                                     from_replica_generation)
+        for doc in docs:
+            return_doc_cb(doc)
         # for tests
-        self._db._last_exchange_log = {
-            'receive': {'docs': self._incoming_trace,
-                        'from_id': from_replica_uid,
-                        'from_gen': from_replica_generation,
-                        'last_known_gen': self._last_known_generation},
-            'return': {'new_docs': [(d.doc_id, d.rev) for d in new_docs],
-                       'conf_docs': [(d.doc_id, d.rev) for d in conflicts],
-                       'last_gen': self.new_gen}
-        }
+        self._db._last_exchange_log['return'] = {
+            'docs': [(d.doc_id, d.rev) for d in docs],
+            'last_gen': self.new_gen
+            }
 
 
 class LocalSyncTarget(u1db.SyncTarget):
@@ -238,10 +239,11 @@
         # 1st step: try to insert incoming docs
         for doc in docs:
             sync_exch.insert_doc_from_source(doc)
+        # record progress
+        sync_exch.record_sync_progress(from_replica_uid,
+                                       from_replica_generation)
         # 2nd step: find changed documents (including conflicts) to return
         new_gen = sync_exch.find_docs_to_return(last_known_generation)
         # final step: return docs and record source replica sync point
-        sync_exch.return_docs_and_record_sync(from_replica_uid,
-                                              from_replica_generation,
-                                              return_doc_cb)
+        sync_exch.return_docs(return_doc_cb)
         return new_gen

=== modified file 'u1db/tests/test_sync.py'
--- u1db/tests/test_sync.py	2011-12-06 20:44:53 +0000
+++ u1db/tests/test_sync.py	2011-12-13 09:59:32 +0000
@@ -115,8 +115,7 @@
         self.assertEqual(([(doc.doc_id, doc.rev, simple_doc)], 1),
                          (self.other_docs, new_gen))
         self.assertEqual(self.db._last_exchange_log['return'],
-                         {'last_gen': 1, 'conf_docs': [(doc.doc_id, doc.rev)],
-                          'new_docs': []})
+                         {'last_gen': 1, 'docs': [(doc.doc_id, doc.rev)]})
 
     def test_sync_exchange_ignores_convergence(self):
         doc = self.db.create_doc(simple_doc)
@@ -140,8 +139,7 @@
         self.assertEqual(([(doc.doc_id, doc.rev, simple_doc)], 1),
                          (self.other_docs, new_gen))
         self.assertEqual(self.db._last_exchange_log['return'],
-                         {'last_gen': 1, 'new_docs': [(doc.doc_id, doc.rev)],
-                          'conf_docs': []})
+                         {'last_gen': 1, 'docs': [(doc.doc_id, doc.rev)]})
 
     def test_sync_exchange_getting_newer_docs(self):
         doc = self.db.create_doc(simple_doc)
@@ -192,8 +190,7 @@
         self.assertEqual(0, self.db2.get_sync_generation('test1'))
         self.assertEqual({'receive': {'docs': [], 'from_id': 'test1',
                                       'from_gen': 0, 'last_known_gen': 0},
-                          'return': {'new_docs': [], 'conf_docs': [],
-                                     'last_gen': 0}},
+                          'return': {'docs': [], 'last_gen': 0}},
                          self.db2._last_exchange_log)
 
     def test_sync_puts_changes(self):
@@ -205,8 +202,7 @@
         self.assertEqual({'receive': {'docs': [(doc.doc_id, doc.rev)],
                                       'from_id': 'test1',
                                       'from_gen': 1, 'last_known_gen': 0},
-                          'return': {'new_docs': [], 'conf_docs': [],
-                                     'last_gen': 1}},
+                          'return': {'docs': [], 'last_gen': 1}},
                          self.db2._last_exchange_log)
 
     def test_sync_pulls_changes(self):
@@ -218,8 +214,8 @@
         self.assertEqual(1, self.db2.get_sync_generation('test1'))
         self.assertEqual({'receive': {'docs': [], 'from_id': 'test1',
                                       'from_gen': 0, 'last_known_gen': 0},
-                          'return': {'new_docs': [(doc.doc_id, doc.rev)],
-                                     'conf_docs': [], 'last_gen': 1}},
+                          'return': {'docs': [(doc.doc_id, doc.rev)],
+                                     'last_gen': 1}},
                          self.db2._last_exchange_log)
         self.assertEqual([doc],
                          self.db1.get_from_index('test-idx', [('value',)]))
@@ -242,8 +238,8 @@
         self.assertEqual(0, self.sync(self.db1, self.db2))
         self.assertEqual({'receive': {'docs': [], 'from_id': 'test1',
                                       'from_gen': 0, 'last_known_gen': 0},
-                          'return': {'new_docs': [(doc.doc_id, doc.rev)],
-                                     'conf_docs': [], 'last_gen': 1}},
+                          'return': {'docs': [(doc.doc_id, doc.rev)],
+                                     'last_gen': 1}},
                          self.db2._last_exchange_log)
         self.assertEqual(1, self.db1.get_sync_generation('test2'))
         # c2 should not have gotten a '_record_sync_info' call, because the
@@ -260,8 +256,7 @@
         self.assertEqual({'receive': {'docs': [(doc.doc_id, doc.rev)],
                                       'from_id': 'test1',
                                       'from_gen': 1, 'last_known_gen': 0},
-                          'return': {'new_docs': [],
-                                     'conf_docs': [], 'last_gen': 1}},
+                          'return': {'docs': [], 'last_gen': 1}},
                          self.db2._last_exchange_log)
 
     def test_sync_ignores_superseded(self):
@@ -278,8 +273,8 @@
         self.assertEqual({'receive': {'docs': [(doc.doc_id, doc_rev1)],
                                       'from_id': 'test2',
                                       'from_gen': 1, 'last_known_gen': 0},
-                          'return': {'new_docs': [(doc.doc_id, doc_rev2)],
-                                     'conf_docs': [], 'last_gen': 2}},
+                          'return': {'docs': [(doc.doc_id, doc_rev2)],
+                                     'last_gen': 2}},
                          self.db1._last_exchange_log)
         self.assertGetDoc(self.db1, doc.doc_id, doc_rev2, new_content, False)
 
@@ -296,8 +291,7 @@
         self.assertEqual({'receive': {'docs': [(doc_id, doc1_rev)],
                                       'from_id': 'test1',
                                       'from_gen': 1, 'last_known_gen': 0},
-                          'return': {'new_docs': [],
-                                     'conf_docs': [(doc_id, doc2_rev)],
+                          'return': {'docs': [(doc_id, doc2_rev)],
                                      'last_gen': 1}},
                          self.db2._last_exchange_log)
         self.assertEqual([doc_id, doc_id], self.db1._get_transaction_log())
@@ -322,8 +316,7 @@
         self.assertEqual({'receive': {'docs': [(doc_id, doc1.rev)],
                                       'from_id': 'test1',
                                       'from_gen': 2, 'last_known_gen': 1},
-                          'return': {'new_docs': [],
-                                     'conf_docs': [(doc_id, doc2.rev)],
+                          'return': {'docs': [(doc_id, doc2.rev)],
                                      'last_gen': 2}},
                          self.db2._last_exchange_log)
         self.assertEqual([doc_id, doc_id, doc_id],
@@ -375,8 +368,7 @@
         self.assertEqual({'receive': {'docs': [(doc_id, deleted_rev)],
                                       'from_id': 'test1',
                                       'from_gen': 2, 'last_known_gen': 1},
-                          'return': {'new_docs': [], 'conf_docs': [],
-                                     'last_gen': 2}},
+                          'return': {'docs': [], 'last_gen': 2}},
                          self.db2._last_exchange_log)
         self.assertGetDoc(self.db1, doc_id, deleted_rev, None, False)
         self.assertGetDoc(self.db2, doc_id, deleted_rev, None, False)
@@ -386,8 +378,7 @@
         self.assertEqual({'receive': {'docs': [(doc_id, deleted_rev)],
                                       'from_id': 'test2',
                                       'from_gen': 2, 'last_known_gen': 0},
-                          'return': {'new_docs': [], 'conf_docs': [],
-                                     'last_gen': 2}},
+                          'return': {'docs': [], 'last_gen': 2}},
                          self.db3._last_exchange_log)
         self.assertGetDoc(self.db3, doc_id, deleted_rev, None, False)