Merge lp:~pedronis/u1db/push-in-generation-order into lp:u1db

[Samuele Pedroni]

send documents from source to target for sync in generation order, update incrementally the known generation from the source on the target

not sure how much sanity checking we want to do

[John A Meinel (jameinel)]

10 + def sync_exchange(self, docs, from_replica_generations,
11 + from_replica_uid,
12 last_known_generation, return_doc_cb):

It seems like the from_replica_generations should be attached to the docs, rather than a separate list. We can do it this way if you think it makes sense, but generally I'm not a big fan of separate lists that need to be iterated in sync.

One of my complaints about the current put and then update generation is that it is taking 2 database locks for a single text insert.

[Samuele Pedroni (pedronis)]

I think I would prefer optimizing put_doc_if_newer + set_sync_generation in a later branch once we got the code right also for the target -> source direction

I'm still open to suggestions whether to use tuples (gen, doc) or store generation on the docs, to address the first comment. The second lists seemed the path of list resistance

[lp:~pedronis/u1db/push-in-generation-order]

153. By Samuele Pedroni on 2011-12-14

pass list of tuples [(Doc, gen)] to sync_exchange()

154. By Samuele Pedroni on 2011-12-14

add a failure and retry test, fix tests to use string contents

[John A Meinel (jameinel)]

+ docs_by_generation = zip(docs_to_send, (gen for _, gen in changes)

I'm pretty sure that get_docs doesn't guarantee the return order. We should instead do:

doc_id_to_generation = dict(changes)
doc_generation = [(doc, doc_id_to_generation[doc.doc_id]) for doc in docs]
doc_generation.sort(key=lambda x: x[1])

I wonder if we are missing test coverage (or maybe all current implementations happen to return them in order). I did want us to be flexible about it so that implementations could return documents in whatever is the fastest for them.

And if we are going to use this, I think sync_exchange should probably check that cur_gen > last_gen, and have it be an error otherwise. (We could probably allow gen to be 0 to indicate to ignore a given entry.)

review: Approve

[lp:~pedronis/u1db/push-in-generation-order]

155. By Samuele Pedroni on 2011-12-14

document the expectation that get_docs returns documents in doc_ids order

=== modified file 'u1db/__init__.py'
--- u1db/__init__.py	2011-12-12 19:40:50 +0000
+++ u1db/__init__.py	2011-12-14 12:51:31 +0000
@@ -76,7 +76,7 @@
         :param doc_ids: A list of document identifiers.
         :param check_for_conflicts: If set to False, then the conflict check
             will be skipped, and 'None' will be returned instead of True/False.
-        :return: [Document] for each document id.
+        :return: [Document] for each document id and matching doc_ids order.
         """
         raise NotImplementedError(self.get_docs)
 
@@ -349,8 +349,8 @@
         """
         raise NotImplementedError(self.record_sync_info)
 
-    def sync_exchange(self, docs,
-                      from_replica_uid, from_replica_generation,
+    def sync_exchange(self, docs_by_generation,
+                      from_replica_uid,
                       last_known_generation, return_doc_cb):
         """Incorporate the documents sent from the other replica.
 
@@ -360,11 +360,16 @@
         This adds docs to the local store, and determines documents that need
         to be returned to the other replica.
 
-        :param docs: A list of [Document] objects indicating
-            documents which should be updated on this replica.
+        Documents must be supplied in docs_by_generation paired with
+        the generation of their latest change in order from the oldest
+        change to the newest, that means from the oldest generation 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 from_replica_uid: The other replica's identifier
-        :param from_replica_generation: The db generation for the other replica
-            indicating the tip of data being sent by docs_info.
         :param last_known_generation: The last generation that other replica
             knows about this
         :param: return_doc_cb(doc): is a callback

=== modified file 'u1db/remote/http_app.py'
--- u1db/remote/http_app.py	2011-12-12 20:59:22 +0000
+++ u1db/remote/http_app.py	2011-12-14 12:51:31 +0000
@@ -283,19 +283,17 @@
 
     @http_method(from_replica_generation=int, last_known_generation=int,
                  content_as_args=True)
-    def post_args(self, last_known_generation, from_replica_generation):
-        self.from_replica_generation = from_replica_generation
+    def post_args(self, last_known_generation):
         self.last_known_generation = last_known_generation
         self.sync_exch = self.target.get_sync_exchange()
 
     @http_method(content_as_args=True)
-    def post_stream_entry(self, id, rev, content):
+    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)
 
     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)

=== modified file 'u1db/remote/http_target.py'
--- u1db/remote/http_target.py	2011-12-06 20:26:29 +0000
+++ u1db/remote/http_target.py	2011-12-14 12:51:31 +0000
@@ -43,8 +43,7 @@
         self._request_json('PUT', ['sync-from', other_replica_uid], {},
                                   {'generation': other_replica_generation})
 
-    def sync_exchange(self, docs, from_replica_uid,
-                      from_replica_generation,
+    def sync_exchange(self, docs_by_generations, from_replica_uid,
                       last_known_generation, return_doc_cb):
         self._ensure_connection()
         self._conn.putrequest('POST',
@@ -57,10 +56,10 @@
             entry = simplejson.dumps(dic) + "\r\n"
             entries.append(entry)
             return len(entry)
-        size += prepare(last_known_generation=last_known_generation,
-                        from_replica_generation=from_replica_generation)
-        for doc in docs:
-            size += prepare(id=doc.doc_id, rev=doc.rev, content=doc.content)
+        size += prepare(last_known_generation=last_known_generation)
+        for doc, gen in docs_by_generations:
+            size += prepare(id=doc.doc_id, rev=doc.rev, content=doc.content,
+                            gen=gen)
         self._conn.putheader('content-length', str(size))
         self._conn.endheaders()
         for entry in entries:

=== modified file 'u1db/sync.py'
--- u1db/sync.py	2011-12-13 09:47:57 +0000
+++ u1db/sync.py	2011-12-14 12:51:31 +0000
@@ -93,18 +93,19 @@
          others_my_gen) = sync_target.get_sync_info(self.source._replica_uid)
         # what's changed since that generation and this current gen
         my_gen, changes = self.source.whats_changed(others_my_gen)
-        changed_doc_ids = set(doc_id for doc_id, _ in changes)
+        changed_doc_ids = [doc_id for doc_id, _ in changes]
         # prepare to send all the changed docs
         docs_to_send = self.source.get_docs(changed_doc_ids,
             check_for_conflicts=False)
+        docs_by_generation = zip(docs_to_send, (gen for _, gen in changes))
 
         # this source last-seen database generation for the target
         other_last_known_gen = self.source.get_sync_generation(
             other_replica_uid)
         # exchange documents and try to insert the returned ones with
         # the target, return target synced-up-to gen
-        new_gen = sync_target.sync_exchange(docs_to_send,
-                        self.source._replica_uid, my_gen, other_last_known_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
         self.source.set_sync_generation(other_replica_uid, new_gen)
@@ -232,16 +233,16 @@
     def get_sync_exchange(self):
         return SyncExchange(self._db)
 
-    def sync_exchange(self, docs,
-                      from_replica_uid, from_replica_generation,
+    def sync_exchange(self, docs_by_generations, from_replica_uid,
                       last_known_generation, return_doc_cb):
         sync_exch = self.get_sync_exchange()
         # 1st step: try to insert incoming docs
-        for doc in docs:
+        for doc, _ in docs_by_generations:
             sync_exch.insert_doc_from_source(doc)
         # record progress
-        sync_exch.record_sync_progress(from_replica_uid,
-                                       from_replica_generation)
+        if docs_by_generations:
+            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)
         # final step: return docs and record source replica sync point

=== modified file 'u1db/tests/test_backends.py'
--- u1db/tests/test_backends.py	2011-12-12 19:40:50 +0000
+++ u1db/tests/test_backends.py	2011-12-14 12:51:31 +0000
@@ -637,9 +637,8 @@
         st = self.db.get_sync_target()
         def ignore(doc_id, doc_rev, doc):
             pass
-        docs = [Document(doc.doc_id, other_rev, new_content)]
-        result = st.sync_exchange(docs, 'other-replica',
-                                  from_replica_generation=10,
+        docs_by_gen = [(Document(doc.doc_id, other_rev, new_content), 10)]
+        result = st.sync_exchange(docs_by_gen, 'other-replica',
                                   last_known_generation=0,
                                   return_doc_cb=ignore)
         self.assertGetDoc(self.db, doc.doc_id, other_rev, new_content, False)

=== modified file 'u1db/tests/test_http_app.py'
--- u1db/tests/test_http_app.py	2011-12-06 20:38:52 +0000
+++ u1db/tests/test_http_app.py	2011-12-14 12:51:31 +0000
@@ -554,11 +554,29 @@
         self.assertEqual(self.db0.get_sync_generation('other-id'), 2)
 
     def test_sync_exchange_send(self):
-        entry = {'id': 'doc-here', 'rev': 'replica:1', 'content':
-                 '{"value": "here"}'}
-        args = dict(from_replica_generation=10, last_known_generation=0)
+        entries = {
+            10: {'id': 'doc-here', 'rev': 'replica:1', 'content':
+                 '{"value": "here"}', 'gen':10 },
+            11: {'id': 'doc-here2', 'rev': 'replica:1', 'content':
+                 '{"value": "here2"}', 'gen':11 }
+            }
+
+        gens = []
+        _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)
+            self.assertGetDoc(self.db0, entries[other_gen]['id'],
+                              entries[other_gen]['rev'],
+                              entries[other_gen]['content'], False)
+
+        self.patch(self.db0, 'set_sync_generation',
+                   set_sync_generation_witness)
+
+        args = dict(last_known_generation=0)
         body = ("%s\r\n" % simplejson.dumps(args) +
-                "%s\r\n" % simplejson.dumps(entry))
+                "%s\r\n" % simplejson.dumps(entries[10]) +
+                "%s\r\n" % simplejson.dumps(entries[11]))
         resp = self.app.post('/db0/sync-from/replica',
                             params=body,
                             headers={'content-type':
@@ -566,13 +584,12 @@
         self.assertEqual(200, resp.status)
         self.assertEqual('application/x-u1db-multi-json',
                          resp.header('content-type'))
-        self.assertEqual({'new_generation': 1}, simplejson.loads(resp.body))
-        self.assertGetDoc(self.db0, 'doc-here', 'replica:1',
-                          '{"value": "here"}', False)
+        self.assertEqual({'new_generation': 2}, simplejson.loads(resp.body))
+        self.assertEqual([('replica', 10), ('replica', 11)], gens)
 
     def test_sync_exchange_receive(self):
         doc = self.db0.create_doc('{"value": "there"}')
-        args = dict(from_replica_generation=10, last_known_generation=0)
+        args = dict(last_known_generation=0)
         body = "%s\r\n" % simplejson.dumps(args)
         resp = self.app.post('/db0/sync-from/replica',
                             params=body,

=== modified file 'u1db/tests/test_remote_sync_target.py'
--- u1db/tests/test_remote_sync_target.py	2011-12-06 20:38:52 +0000
+++ u1db/tests/test_remote_sync_target.py	2011-12-14 12:51:31 +0000
@@ -15,10 +15,12 @@
 """Tests for the remote sync targets"""
 
 from wsgiref import simple_server
+import cStringIO
 #from paste import httpserver
 
 from u1db import (
     Document,
+    errors,
     tests,
     )
 from u1db.remote import (
@@ -91,26 +93,67 @@
         def receive_doc(doc):
             other_docs.append((doc.doc_id, doc.rev, doc.content))
         new_gen = remote_target.sync_exchange(
-                        [Document('doc-here', 'replica:1', {'value': 'here'})],
-                        'replica', from_replica_generation=10,
-                        last_known_generation=0, return_doc_cb=receive_doc)
+                [(Document('doc-here', 'replica:1', '{"value": "here"}'), 10)],
+                '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):
+            if doc.doc_id in trigger_ids:
+                raise Exception
+            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))
+        self.assertRaises(errors.HTTPError, remote_target.sync_exchange,
+                [(Document('doc-here', 'replica:1', '{"value": "here"}'), 10),
+                 (Document('doc-here2', 'replica:1', '{"value": "here2"}'), 11)
+                 ], 'replica', last_known_generation=0,
+                return_doc_cb=receive_doc)
+        self.assertGetDoc(db, 'doc-here', 'replica:1', '{"value": "here"}',
+                          False)
+        self.assertEqual(10, db.get_sync_generation('replica'))
+        self.assertEqual([], other_docs)
+        # retry
+        trigger_ids = []
+        new_gen = remote_target.sync_exchange(
+                [(Document('doc-here2', 'replica:1', '{"value": "here2"}'), 11)
+                 ], 'replica', last_known_generation=0,
+                return_doc_cb=receive_doc)
+        self.assertGetDoc(db, 'doc-here2', 'replica:1', '{"value": "here2"}',
+                          False)
+        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)
 
     def test_sync_exchange_receive(self):
         self.startServer()
         db = self.request_state._create_database('test')
-        doc = db.create_doc({'value': 'there'})
+        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))
         new_gen = remote_target.sync_exchange(
-                        [],
-                        'replica', from_replica_generation=10,
-                        last_known_generation=0, return_doc_cb=receive_doc)
+                        [], 'replica', last_known_generation=0,
+                        return_doc_cb=receive_doc)
         self.assertEqual(1, new_gen)
-        self.assertEqual([(doc.doc_id, doc.rev, {'value': 'there'})],
+        self.assertEqual([(doc.doc_id, doc.rev, '{"value": "there"}')],
                          other_docs)
 
 

=== modified file 'u1db/tests/test_sync.py'
--- u1db/tests/test_sync.py	2011-12-12 20:38:45 +0000
+++ u1db/tests/test_sync.py	2011-12-14 12:51:31 +0000
@@ -92,9 +92,8 @@
         self.assertEqual(('test', 0, 10), self.st.get_sync_info('replica'))
 
     def test_sync_exchange(self):
-        docs = [Document('doc-id', 'replica:1', simple_doc)]
-        new_gen = self.st.sync_exchange(docs,
-                                        'replica', from_replica_generation=10,
+        docs_by_gen = [(Document('doc-id', 'replica:1', simple_doc), 10)]
+        new_gen = self.st.sync_exchange(docs_by_gen, 'replica',
                                         last_known_generation=0,
                                         return_doc_cb=self.receive_doc)
         self.assertGetDoc(self.db, 'doc-id', 'replica:1', simple_doc, False)
@@ -102,13 +101,24 @@
         self.assertEqual(([], 1), (self.other_docs, new_gen))
         self.assertEqual(10, self.st.get_sync_info('replica')[-1])
 
+    def test_sync_exchange_push_many(self):
+        docs_by_gen = [(Document('doc-id', 'replica:1', simple_doc), 10),
+                       (Document('doc-id2', 'replica:1', nested_doc), 11)]
+        new_gen = self.st.sync_exchange(docs_by_gen, 'replica',
+                                        last_known_generation=0,
+                                        return_doc_cb=self.receive_doc)
+        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(11, self.st.get_sync_info('replica')[-1])
+
     def test_sync_exchange_refuses_conflicts(self):
         doc = self.db.create_doc(simple_doc)
         self.assertEqual([doc.doc_id], self.db._get_transaction_log())
         new_doc = '{"key": "altval"}'
-        docs = [Document(doc.doc_id, 'replica:1', new_doc)]
-        new_gen = self.st.sync_exchange(docs,
-                                        'replica', from_replica_generation=10,
+        docs_by_gen = [(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.assertEqual([doc.doc_id], self.db._get_transaction_log())
@@ -120,9 +130,8 @@
     def test_sync_exchange_ignores_convergence(self):
         doc = self.db.create_doc(simple_doc)
         self.assertEqual([doc.doc_id], self.db._get_transaction_log())
-        docs = [Document(doc.doc_id, doc.rev, simple_doc)]
-        new_gen = self.st.sync_exchange(docs,
-                                        'replica', from_replica_generation=10,
+        docs_by_gen = [(Document(doc.doc_id, doc.rev, simple_doc), 10)]
+        new_gen = self.st.sync_exchange(docs_by_gen, 'replica',
                                         last_known_generation=1,
                                         return_doc_cb=self.receive_doc)
         self.assertEqual([doc.doc_id], self.db._get_transaction_log())
@@ -132,7 +141,6 @@
         doc = self.db.create_doc(simple_doc)
         self.assertEqual([doc.doc_id], self.db._get_transaction_log())
         new_gen = self.st.sync_exchange([], 'other-replica',
-                                        from_replica_generation=10,
                                         last_known_generation=0,
                                         return_doc_cb=self.receive_doc)
         self.assertEqual([doc.doc_id], self.db._get_transaction_log())
@@ -145,10 +153,8 @@
         doc = self.db.create_doc(simple_doc)
         self.assertEqual([doc.doc_id], self.db._get_transaction_log())
         new_doc = '{"key": "altval"}'
-        docs = [Document(doc.doc_id, 'test:1|z:2', new_doc)]
-        new_gen = self.st.sync_exchange(docs,
-                                        'other-replica',
-                                        from_replica_generation=10,
+        docs_by_gen = [(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([doc.doc_id, doc.doc_id],
@@ -165,10 +171,8 @@
             return val
         self.db.whats_changed = after_whatschanged
         new_doc = '{"key": "altval"}'
-        docs = [Document(doc.doc_id, 'test:1|z:2', new_doc)]
-        new_gen = self.st.sync_exchange(docs,
-                                        'other-replica',
-                                        from_replica_generation=10,
+        docs_by_gen = [(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(([], 2), (self.other_docs, new_gen))
@@ -188,8 +192,7 @@
         self.assertEqual(0, self.sync(self.db1, self.db2))
         self.assertEqual(0, self.db1.get_sync_generation('test2'))
         self.assertEqual(0, self.db2.get_sync_generation('test1'))
-        self.assertEqual({'receive': {'docs': [], 'from_id': 'test1',
-                                      'from_gen': 0, 'last_known_gen': 0},
+        self.assertEqual({'receive': {'docs': [], 'last_known_gen': 0},
                           'return': {'docs': [], 'last_gen': 0}},
                          self.db2._last_exchange_log)
 
@@ -212,8 +215,7 @@
         self.assertGetDoc(self.db1, doc.doc_id, doc.rev, simple_doc, False)
         self.assertEqual(1, self.db1.get_sync_generation('test2'))
         self.assertEqual(1, self.db2.get_sync_generation('test1'))
-        self.assertEqual({'receive': {'docs': [], 'from_id': 'test1',
-                                      'from_gen': 0, 'last_known_gen': 0},
+        self.assertEqual({'receive': {'docs': [], 'last_known_gen': 0},
                           'return': {'docs': [(doc.doc_id, doc.rev)],
                                      'last_gen': 1}},
                          self.db2._last_exchange_log)
@@ -236,8 +238,7 @@
             return result
         self.db1.put_doc_if_newer = after_put_doc_if_newer
         self.assertEqual(0, self.sync(self.db1, self.db2))
-        self.assertEqual({'receive': {'docs': [], 'from_id': 'test1',
-                                      'from_gen': 0, 'last_known_gen': 0},
+        self.assertEqual({'receive': {'docs': [], 'last_known_gen': 0},
                           'return': {'docs': [(doc.doc_id, doc.rev)],
                                      'last_gen': 1}},
                          self.db2._last_exchange_log)