Merge lp:~pedronis/u1db/sync-returns-doc-in-gen-order into lp:u1db

[Samuele Pedroni]

return synced documents from the target to the source in generation order, integration test to check that this is used to set known last generation incrementally as we get the docs

=== modified file 'u1db/__init__.py'
--- u1db/__init__.py	2011-12-14 12:43:53 +0000
+++ u1db/__init__.py	2011-12-14 16:05:29 +0000
@@ -365,6 +365,10 @@
         change to the newest, that means from the oldest generation to
         the newest.
 
+        Documents are also returned paired with the generation of
+        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
@@ -372,11 +376,12 @@
         :param from_replica_uid: The other replica's identifier
         :param last_known_generation: The last generation that other replica
             knows about this
-        :param: return_doc_cb(doc): is a callback
+        :param: return_doc_cb(doc, gen): is a callback
                 used to return documents to the other replica, it will
                 be invoked in turn with Documents that have changed since
-                last_known_generation.
-        :return: new_generation - After applying docs_info, this is
+                last_known_generation together with the generation of
+                their last change.
+        :return: new_generation - After applying docs_by_generation, this is
             the current generation for this replica
         """
         raise NotImplementedError(self.sync_exchange)

=== modified file 'u1db/backends/inmemory.py'
--- u1db/backends/inmemory.py	2011-12-13 09:27:56 +0000
+++ u1db/backends/inmemory.py	2011-12-14 16:05:29 +0000
@@ -54,7 +54,8 @@
         return InMemorySyncTarget(self)
 
     def _get_transaction_log(self):
-        return self._transaction_log
+        # snapshot!
+        return self._transaction_log[:]
 
     def _get_generation(self):
         return len(self._transaction_log)

=== modified file 'u1db/remote/http_app.py'
--- u1db/remote/http_app.py	2011-12-13 15:27:18 +0000
+++ u1db/remote/http_app.py	2011-12-14 16:05:29 +0000
@@ -294,10 +294,12 @@
         self.sync_exch.record_sync_progress(self.from_replica_uid, gen)
 
     def post_end(self):
-        def send_doc(doc):
-            entry = dict(id=doc.doc_id, rev=doc.rev, content=doc.content)
+        def send_doc(doc, gen):
+            entry = dict(id=doc.doc_id, rev=doc.rev, content=doc.content,
+                         gen=gen)
             self.responder.stream_entry(entry)
-        new_gen = self.sync_exch.find_docs_to_return(self.last_known_generation)
+        new_gen = self.sync_exch.find_changes_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(send_doc)

=== modified file 'u1db/remote/http_target.py'
--- u1db/remote/http_target.py	2011-12-14 10:37:17 +0000
+++ u1db/remote/http_target.py	2011-12-14 16:05:29 +0000
@@ -71,7 +71,7 @@
         for entry in data[1:]:
             entry = simplejson.loads(entry)
             doc = Document(entry['id'], entry['rev'], entry['content'])
-            return_doc_cb(doc)
+            return_doc_cb(doc, entry['gen'])
         data = None
         return res['new_generation']
 

=== modified file 'u1db/sync.py'
--- u1db/sync.py	2011-12-14 10:37:17 +0000
+++ u1db/sync.py	2011-12-14 16:05:29 +0000
@@ -104,10 +104,15 @@
             other_replica_uid)
         # 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)
+
         new_gen = sync_target.sync_exchange(docs_by_generation,
                         self.source._replica_uid, other_last_known_gen,
-                        return_doc_cb=self._insert_doc_from_target)
-        # record target synced-up-to generation
+                        return_doc_cb=take_doc)
+        # record target synced-up-to generation including applying what we sent
         self.source.set_sync_generation(other_replica_uid, new_gen)
 
         # if gapless record current reached generation with target
@@ -122,7 +127,7 @@
     def __init__(self, db):
         self._db = db
         self.seen_ids = set()  # incoming ids not superseded
-        self.doc_ids_to_return = None
+        self.changes_to_return = None
         self.new_gen = None
         # for tests
         self._incoming_trace = []
@@ -177,13 +182,12 @@
             '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.
+    def find_changes_to_return(self, last_known_generation):
+        """Find changes to return.
 
-        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
-        (superseded by the sender, etc).
+        Find changes since last_known_generation in db generation
+        order using whats_changed. It excludes 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
@@ -193,30 +197,32 @@
             '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
-        self.doc_ids_to_return = set(doc_id for doc_id in changed_doc_ids
-                                     if doc_id not in seen_ids)
+        self.changes_to_return = [(doc_id, gen) for (doc_id, gen)  in changes
+                                         if doc_id not in seen_ids]
         return gen
 
     def return_docs(self, return_doc_cb):
-        """Return the marked documents repeatedly invoking the callback
-        return_doc_cb.
+        """Return the changed documents and their last change generation
+        repeatedly invoking the callback return_doc_cb.
 
         The final step of a sync exchange.
 
-        :param: return_doc_cb(doc): is a callback
-                used to return the marked documents to the target replica,
+        :param: return_doc_cb(doc, gen): is a callback
+                used to return the documents with their last change generation
+                to the target replica.
         :return: None
         """
-        doc_ids_to_return = self.doc_ids_to_return
+        changes_to_return = self.changes_to_return
         # return docs, including conflicts
-        docs = self._db.get_docs(doc_ids_to_return,
-                                     check_for_conflicts=False)
-        for doc in docs:
-            return_doc_cb(doc)
+        changed_doc_ids = [doc_id for doc_id, _ in changes_to_return]
+        docs = self._db.get_docs(changed_doc_ids, check_for_conflicts=False)
+
+        docs_by_gen = zip(docs, (gen for _, gen in changes_to_return))
+        for doc, gen in docs_by_gen:
+            return_doc_cb(doc, gen)
         # for tests
         self._db._last_exchange_log['return'] = {
             'docs': [(d.doc_id, d.rev) for d in docs],
@@ -244,7 +250,7 @@
             latest_gen = docs_by_generations[-1][1]
             sync_exch.record_sync_progress(from_replica_uid, latest_gen)
         # 2nd step: find changed documents (including conflicts) to return
-        new_gen = sync_exch.find_docs_to_return(last_known_generation)
+        new_gen = sync_exch.find_changes_to_return(last_known_generation)
         # final step: return docs and record source replica sync point
         sync_exch.return_docs(return_doc_cb)
         return new_gen

=== modified file 'u1db/tests/test_http_app.py'
--- u1db/tests/test_http_app.py	2011-12-13 15:27:18 +0000
+++ u1db/tests/test_http_app.py	2011-12-14 16:05:29 +0000
@@ -556,9 +556,9 @@
     def test_sync_exchange_send(self):
         entries = {
             10: {'id': 'doc-here', 'rev': 'replica:1', 'content':
-                 '{"value": "here"}', 'gen':10 },
+                 '{"value": "here"}', 'gen':10},
             11: {'id': 'doc-here2', 'rev': 'replica:1', 'content':
-                 '{"value": "here2"}', 'gen':11 }
+                 '{"value": "here2"}', 'gen': 11}
             }
 
         gens = []
@@ -589,6 +589,7 @@
 
     def test_sync_exchange_receive(self):
         doc = self.db0.create_doc('{"value": "there"}')
+        doc2 = self.db0.create_doc('{"value": "there2"}')
         args = dict(last_known_generation=0)
         body = "%s\r\n" % simplejson.dumps(args)
         resp = self.app.post('/db0/sync-from/replica',
@@ -599,11 +600,14 @@
         self.assertEqual('application/x-u1db-multi-json',
                          resp.header('content-type'))
         parts = resp.body.splitlines()
-        self.assertEqual(2, len(parts))
-        self.assertEqual({'new_generation': 1}, simplejson.loads(parts[0]))
+        self.assertEqual(3, len(parts))
+        self.assertEqual({'new_generation': 2}, simplejson.loads(parts[0]))
         self.assertEqual({'content': '{"value": "there"}',
-                          'rev': doc.rev, 'id': doc.doc_id},
+                          'rev': doc.rev, 'id': doc.doc_id, 'gen': 1},
                          simplejson.loads(parts[1]))
+        self.assertEqual({'content': '{"value": "there2"}',
+                          'rev': doc2.rev, 'id': doc2.doc_id, 'gen': 2},
+                         simplejson.loads(parts[2]))
 
 
 class TestHTTPErrors(tests.TestCase):

=== modified file 'u1db/tests/test_remote_sync_target.py'
--- u1db/tests/test_remote_sync_target.py	2011-12-14 11:05:51 +0000
+++ u1db/tests/test_remote_sync_target.py	2011-12-14 16:05:29 +0000
@@ -115,9 +115,9 @@
             return _put_doc_if_newer(doc)
         self.patch(db, 'put_doc_if_newer', bomb_put_doc_if_newer)
         remote_target = self.getSyncTarget('test')
-        other_docs = []
-        def receive_doc(doc):
-            other_docs.append((doc.doc_id, doc.rev, doc.content))
+        other_changes = []
+        def receive_doc(doc, gen):
+            other_changes.append((doc.doc_id, doc.rev, doc.content, gen))
         self.assertRaises(errors.HTTPError, remote_target.sync_exchange,
                 [(Document('doc-here', 'replica:1', '{"value": "here"}'), 10),
                  (Document('doc-here2', 'replica:1', '{"value": "here2"}'), 11)
@@ -126,7 +126,7 @@
         self.assertGetDoc(db, 'doc-here', 'replica:1', '{"value": "here"}',
                           False)
         self.assertEqual(10, db.get_sync_generation('replica'))
-        self.assertEqual([], other_docs)
+        self.assertEqual([], other_changes)
         # retry
         trigger_ids = []
         new_gen = remote_target.sync_exchange(
@@ -138,23 +138,23 @@
         self.assertEqual(11, db.get_sync_generation('replica'))
         self.assertEqual(2, new_gen)
         # bounced back to us
-        self.assertEqual([('doc-here', 'replica:1', '{"value": "here"}')],
-                         other_docs)
+        self.assertEqual([('doc-here', 'replica:1', '{"value": "here"}', 1)],
+                         other_changes)
 
     def test_sync_exchange_receive(self):
         self.startServer()
         db = self.request_state._create_database('test')
         doc = db.create_doc('{"value": "there"}')
         remote_target = self.getSyncTarget('test')
-        other_docs = []
-        def receive_doc(doc):
-            other_docs.append((doc.doc_id, doc.rev, doc.content))
+        other_changes = []
+        def receive_doc(doc, gen):
+            other_changes.append((doc.doc_id, doc.rev, doc.content, gen))
         new_gen = 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"}')],
-                         other_docs)
+        self.assertEqual([(doc.doc_id, doc.rev, '{"value": "there"}', 1)],
+                         other_changes)
 
 
 load_tests = tests.load_with_scenarios

=== modified file 'u1db/tests/test_sync.py'
--- u1db/tests/test_sync.py	2011-12-14 10:37:17 +0000
+++ u1db/tests/test_sync.py	2011-12-14 16:05:29 +0000
@@ -64,7 +64,7 @@
     def setUp(self):
         super(DatabaseSyncTargetTests, self).setUp()
         self.db, self.st = self.create_db_and_target(self)
-        self.other_docs = []
+        self.other_changes = []
 
     def tearDown(self):
         # We delete them explicitly, so that connections are cleanly closed
@@ -72,8 +72,8 @@
         del self.st
         super(DatabaseSyncTargetTests, self).tearDown()
 
-    def receive_doc(self, doc):
-        self.other_docs.append((doc.doc_id, doc.rev, doc.content))
+    def receive_doc(self, doc, gen):
+        self.other_changes.append((doc.doc_id, doc.rev, doc.content, gen))
 
     def test_get_sync_target(self):
         self.assertIsNot(None, self.st)
@@ -98,7 +98,7 @@
                                         return_doc_cb=self.receive_doc)
         self.assertGetDoc(self.db, 'doc-id', 'replica:1', simple_doc, False)
         self.assertEqual(['doc-id'], self.db._get_transaction_log())
-        self.assertEqual(([], 1), (self.other_docs, new_gen))
+        self.assertEqual(([], 1), (self.other_changes, new_gen))
         self.assertEqual(10, self.st.get_sync_info('replica')[-1])
 
     def test_sync_exchange_push_many(self):
@@ -110,7 +110,7 @@
         self.assertGetDoc(self.db, 'doc-id', 'replica:1', simple_doc, False)
         self.assertGetDoc(self.db, 'doc-id2', 'replica:1', nested_doc, False)
         self.assertEqual(['doc-id', 'doc-id2'], self.db._get_transaction_log())
-        self.assertEqual(([], 2), (self.other_docs, new_gen))
+        self.assertEqual(([], 2), (self.other_changes, new_gen))
         self.assertEqual(11, self.st.get_sync_info('replica')[-1])
 
     def test_sync_exchange_refuses_conflicts(self):
@@ -122,8 +122,8 @@
                                         last_known_generation=0,
                                         return_doc_cb=self.receive_doc)
         self.assertEqual([doc.doc_id], self.db._get_transaction_log())
-        self.assertEqual(([(doc.doc_id, doc.rev, simple_doc)], 1),
-                         (self.other_docs, new_gen))
+        self.assertEqual(([(doc.doc_id, doc.rev, simple_doc, 1)], 1),
+                         (self.other_changes, new_gen))
         self.assertEqual(self.db._last_exchange_log['return'],
                          {'last_gen': 1, 'docs': [(doc.doc_id, doc.rev)]})
 
@@ -135,7 +135,7 @@
                                         last_known_generation=1,
                                         return_doc_cb=self.receive_doc)
         self.assertEqual([doc.doc_id], self.db._get_transaction_log())
-        self.assertEqual(([], 1), (self.other_docs, new_gen))
+        self.assertEqual(([], 1), (self.other_changes, new_gen))
 
     def test_sync_exchange_returns_new_docs(self):
         doc = self.db.create_doc(simple_doc)
@@ -144,11 +144,28 @@
                                         last_known_generation=0,
                                         return_doc_cb=self.receive_doc)
         self.assertEqual([doc.doc_id], self.db._get_transaction_log())
-        self.assertEqual(([(doc.doc_id, doc.rev, simple_doc)], 1),
-                         (self.other_docs, new_gen))
+        self.assertEqual(([(doc.doc_id, doc.rev, simple_doc, 1)], 1),
+                         (self.other_changes, new_gen))
         self.assertEqual(self.db._last_exchange_log['return'],
                          {'last_gen': 1, 'docs': [(doc.doc_id, doc.rev)]})
 
+    def test_sync_exchange_returns_many_new_docs(self):
+        doc = self.db.create_doc(simple_doc)
+        doc2 = self.db.create_doc(nested_doc)
+        self.assertEqual([doc.doc_id, doc2.doc_id],
+                         self.db._get_transaction_log())
+        new_gen = self.st.sync_exchange([], 'other-replica',
+                                        last_known_generation=0,
+                                        return_doc_cb=self.receive_doc)
+        self.assertEqual([doc.doc_id, doc2.doc_id],
+                         self.db._get_transaction_log())
+        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(self.db._last_exchange_log['return'],
+                         {'last_gen': 2, 'docs': [(doc.doc_id, doc.rev),
+                                                  (doc2.doc_id, doc2.rev)]})
+
     def test_sync_exchange_getting_newer_docs(self):
         doc = self.db.create_doc(simple_doc)
         self.assertEqual([doc.doc_id], self.db._get_transaction_log())
@@ -159,7 +176,7 @@
                                         return_doc_cb=self.receive_doc)
         self.assertEqual([doc.doc_id, doc.doc_id],
                          self.db._get_transaction_log())
-        self.assertEqual(([], 2), (self.other_docs, new_gen))
+        self.assertEqual(([], 2), (self.other_changes, new_gen))
 
     def test_sync_exchange_with_concurrent_updates(self):
         doc = self.db.create_doc(simple_doc)
@@ -175,7 +192,7 @@
         new_gen = self.st.sync_exchange(docs_by_gen, 'other-replica',
                                         last_known_generation=0,
                                         return_doc_cb=self.receive_doc)
-        self.assertEqual(([], 2), (self.other_docs, new_gen))
+        self.assertEqual(([], 2), (self.other_changes, new_gen))
 
 
 class DatabaseSyncTests(tests.DatabaseBaseTests):
@@ -440,4 +457,54 @@
                           False)
 
 
+class TestRemoteSyncIntegration(tests.TestCaseWithServer):
+    """Integration tests for the most common sync scenario local -> remote"""
+
+    server_def = staticmethod(http_server_def)
+
+    def setUp(self):
+        super(TestRemoteSyncIntegration, self).setUp()
+        self.startServer()
+        self.db1 = inmemory.InMemoryDatabase('test1')
+        self.db2 = self.request_state._create_database('test2')
+
+    def test_sync_tracks_generations_incrementally(self):
+        doc11 = self.db1.create_doc('{"a": 1}')
+        doc12 = self.db1.create_doc('{"a": 2}')
+        doc21 = self.db2.create_doc('{"b": 1}')
+        doc22 = self.db2.create_doc('{"b": 2}')
+        #sanity
+        self.assertEqual(2, len(self.db1._get_transaction_log()))
+        self.assertEqual(2, len(self.db2._get_transaction_log()))
+        progress1 = []
+        progress2 = []
+        _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',
+                   set_sync_generation_witness1)
+
+        _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',
+                   set_sync_generation_witness2)
+
+        db2_url = self.getURL('test2')
+        self.db1.sync(db2_url)
+
+        self.assertEqual([('test2', 1, [doc21.doc_id]),
+                          ('test2', 2, [doc21.doc_id, doc22.doc_id]),
+                          ('test2', 4, [doc21.doc_id, doc22.doc_id])],
+                         progress1)
+        self.assertEqual([('test1', 1, [doc11.doc_id]),
+                          ('test1', 2, [doc11.doc_id, doc12.doc_id]),
+                          ('test1', 4, [doc11.doc_id, doc12.doc_id])],
+                         progress2)
+
+
 load_tests = tests.load_with_scenarios