Merge lp:~cmiller/desktopcouch/views-with-reconnector-proxy into lp:desktopcouch

[Chad Miller]

[Eric Casteleijn (thisfred)]

Looks good tests pass

review: Approve

[lp:~cmiller/desktopcouch/views-with-reconnector-proxy]

172. By Chad Miller on 2010-08-25

First pass at adding tests to find and fix problems when the couchdb crashes.
We crash the couchdb in tests. Whee!

[Stuart Colville (muffinresearch)]

Running the tests I get a lot of failures. Here's the output of running the tests in case it helps https://pastebin.canonical.com/36781/

Looking at the commit message for r172 suggests this may be a work in progress, if that's the case could you change the status of this merge request?

[Chad Miller (cmiller)]

Yes, Stuart, I intended to merge this, proposed it, then starting adding tests to crash couchdb to make sure we recover well. This is not appropriate to merge now.

VDS is taking over this branch.

Unmerged revisions

=== modified file 'desktopcouch/records/server.py'
--- desktopcouch/records/server.py	2010-02-26 22:23:42 +0000
+++ desktopcouch/records/server.py	2010-08-25 21:55:59 +0000
@@ -49,9 +49,17 @@
     def __init__(self, database, uri=None, record_factory=None, create=False,
                  server_class=OAuthCapableServer, oauth_tokens=None,
                  ctx=desktopcouch.local_files.DEFAULT_CONTEXT):
-        if not uri:
-            port = desktopcouch.find_port(ctx=ctx)
-            uri = "http://localhost:%s" % port
+        self.ctx = ctx
+        self.server_uri = uri
         super(CouchDatabase, self).__init__(
                 database, uri, record_factory=record_factory, create=create,
                 server_class=server_class, oauth_tokens=oauth_tokens, ctx=ctx)
+
+    def _reconnect(self):
+        if not self.server_uri:
+            port = desktopcouch.find_port(ctx=self.ctx)
+            uri = "http://localhost:%s" % port
+        else:
+            uri = self.server_uri
+        super(CouchDatabase, self)._reconnect(uri=uri)
+

=== modified file 'desktopcouch/records/server_base.py'
--- desktopcouch/records/server_base.py	2010-07-02 15:32:25 +0000
+++ desktopcouch/records/server_base.py	2010-08-25 21:55:59 +0000
@@ -22,7 +22,7 @@
 """The Desktop Couch Records API."""
 
 import httplib2, urlparse, cgi, copy
-from time import time
+from time import time, sleep
 import socket
 
 # please keep desktopcouch python 2.5 compatible for now
@@ -150,6 +150,8 @@
         self._server_class_extras = server_class_extras
         self.record_factory = record_factory or Record
         self.server_uri = uri
+        self._server = None
+        self.db = None
         self._reconnect()
         self._changes_since = self.db.info()["update_seq"]
         self._changes_last_used = 0  # Immediate run works.
@@ -160,32 +162,40 @@
                 ex.args == ("'NoneType' object has no attribute 'makefile'",)
 
     def with_reconnects(self, func, *args, **kwargs):
-        for retry in (2, 1, None):
+        for retry in (3, 2, 1, None):
             try:
                 return func(*args, **kwargs)
             except Exception, e:
                 if self._is_bug_lp539674(e) and retry:
-                    logging.warn("DB connection timed out.  Reconnecting.")
+                    logging.warn("DB connection failed.  Reconnecting.")
                     self._reconnect()
                     continue
                 elif isinstance(e, socket.error):
                     logging.warn("Other socket error %s.  Reconnecting.", e)
-                    time.sleep(0.3)  
+                    sleep(0.3)  
                     self._reconnect()
                     continue
                 else:
                     raise
-
-    def _reconnect(self):
-        logging.info("Connecting to %s." % (self.server_uri or "discovered local port",))
-        self._server = self._server_class(self.server_uri,
+        raise ValueError("failed to (re-)connect to couchdb server")
+
+    def _reconnect(self, uri=None):
+        logging.info("Connecting to %s.", self.server_uri or "discovered local port")
+
+        self._server = self._server_class(uri or self.server_uri,
                 **self._server_class_extras)
         if self._database_name not in self._server:
             if self._create:
                 self._server.create(self._database_name)
             else:
                 raise NoSuchDatabase(self._database_name)
-        self.db = self._server[self._database_name]
+        if self.db is None:
+            self.db = self._server[self._database_name]
+        else:
+            # Monkey-patch the object the user already uses.  Oook!
+            new_db = self._server[self._database_name]
+            self.db.resource = new_db.resource
+
 
     def _temporary_query(self, map_fun, reduce_fun=None, language='javascript',
             wrapper=None, **options):
@@ -380,9 +390,10 @@
 
     def record_exists(self, record_id):
         """Check if record with given id exists."""
-        if record_id not in self.db:
+        try:
+            record = self.with_reconnects(self.db.__getitem__, record_id)
+        except ResourceNotFound:
             return False
-        record = self.with_reconnects(self.db.__getitem__, record_id)
         return not row_is_deleted(record)
 
     def delete_view(self, view_name, design_doc=DEFAULT_DESIGN_DOCUMENT):
@@ -432,12 +443,36 @@
     def execute_view(self, view_name, design_doc=DEFAULT_DESIGN_DOCUMENT,
             **params):
         """Execute view and return results."""
+
+        class ReconnectingViewWrapper(object):
+            """A view from python-couchdb is an object with attributes that
+            cause HTTP requests to be fired off when accessed.  If we wish
+            to be able to reconnect on disappearance of the server, then
+            we must intercept calls from user to python-couchdb."""
+
+            def __init__(wrapper, obj, *args, **kwargs):
+                wrapper.obj = self.with_reconnects(obj, *args, **kwargs)
+
+            def ___call__(wrapper, **options):
+                return self.with_reconnects(wrapper.obj.__call__, **options)
+
+            def __iter__(wrapper):
+                return self.with_reconnects(wrapper.obj.__iter__)
+
+            def __len__(wrapper):
+                return self.with_reconnects(wrapper.obj.__len__)
+
+            def __getitem__(wrapper, key):
+                return ReconnectingViewWrapper(wrapper.obj.__getitem__, key)
+
         if design_doc is None:
             design_doc = view_name
 
         view_id_fmt = "_design/%(design_doc)s/_view/%(view_name)s"
-        return self.with_reconnects(self.db.view, view_id_fmt % locals(),
-                **params)
+        wrapper = ReconnectingViewWrapper(self.db.view, 
+                                          view_id_fmt % locals(),
+                                          **params)
+        return wrapper
 
     def add_view(self, view_name, map_js, reduce_js,
             design_doc=DEFAULT_DESIGN_DOCUMENT):
@@ -447,7 +482,7 @@
             design_doc = view_name
 
         view = ViewDefinition(design_doc, view_name, map_js, reduce_js)
-        view.sync(self.db)
+        self.with_reconnects(view.sync, self.db)
         assert self.view_exists(view_name, design_doc)
 
     def view_exists(self, view_name, design_doc=DEFAULT_DESIGN_DOCUMENT):
@@ -573,6 +608,7 @@
             uri = couchdburi(
                 self._server.resource.uri, self.db.name, "_changes",
                 since=self._changes_since)
+            ## Assume server has not crashed and URI is the same.  FIXME
             resp, data = self._server.resource.http.request(uri, "GET", "", {})
             if resp["status"] != '200':
                 raise IOError(

=== modified file 'desktopcouch/records/tests/test_server.py'
--- desktopcouch/records/tests/test_server.py	2010-07-02 15:42:05 +0000
+++ desktopcouch/records/tests/test_server.py	2010-08-25 21:55:59 +0000
@@ -19,12 +19,17 @@
 
 """testing database/contact.py module"""
 import testtools
+import os
+import signal
+import time
 
 import desktopcouch.tests as test_environment
 from desktopcouch.records.server import CouchDatabase
 from desktopcouch.records.server_base import (
     row_is_deleted, NoSuchDatabase, FieldsConflict, ResourceConflict)
 from desktopcouch.records.record import Record
+from desktopcouch.stop_local_couchdb import stop_couchdb
+from desktopcouch import find_pid
 
 # pylint can't deal with failing imports even when they're handled
 # pylint: disable-msg=F0401
@@ -53,7 +58,7 @@
         # Connect to CouchDB server
         self.dbname = self._testMethodName
         self.database = CouchDatabase(self.dbname, create=True,
-                ctx=test_environment.test_context)
+                ctx=self.get_test_context())
         #create some records to pull out and test
         self.database.put_record(Record({
             "key1_1": "val1_1", "key1_2": "val1_2", "key1_3": "val1_3",
@@ -67,8 +72,29 @@
 
     def tearDown(self):
         """tear down each test"""
+        #del self.database._server[self.dbname]
+        this_context = self.get_test_context()
+        if this_context != test_environment.test_context:
+            stop_couchdb(ctx=this_context)
         super(TestCouchDatabase, self).tearDown()
-        del self.database._server[self.dbname]
+
+    def get_test_context(self):
+        return test_environment.test_context
+
+    def maybe_die(self):
+        pass
+
+    def wait_until_server_dead(self, pid=None):
+        if pid is not None:
+            pid = find_pid(start_if_not_running=False, ctx=self.get_test_context())
+        if pid is None:
+            return
+        while True:
+            try:
+                os.kill(pid, 0)  # Wait until exited
+                time.sleep(0.1)
+            except OSError:
+                break
 
     def test_database_not_exists(self):
         self.assertRaises(
@@ -78,12 +104,14 @@
         """Test getting mutliple records by type"""
         records = self.database.get_records(
             record_type="test.com",create_view=True)
-        self.assertEqual(3,len(records))
+        self.maybe_die()  # should be able to survive couchdb death
+        self.assertEqual(3, len(records))
 
     def test_get_record(self):
         """Test getting a record."""
         record = Record({'record_number': 0}, record_type="http://example.com/")
         record_id = self.database.put_record(record)
+        self.maybe_die()  # should be able to survive couchdb death
         retrieved_record = self.database.get_record(record_id)
         self.assertEqual(0, retrieved_record['record_number'])
 
@@ -91,6 +119,7 @@
         """Test putting a record."""
         record = Record({'record_number': 0}, record_type="http://example.com/")
         record_id = self.database.put_record(record)
+        ###self.maybe_die()  # should be able to survive couchdb death
         retrieved_record = self.database._server[self.dbname][record_id]
         self.assertEqual(
             record['record_number'], retrieved_record['record_number'])
@@ -121,6 +150,7 @@
         record = Record({'record_number': 0}, record_type="http://example.com/")
         record_id = self.database.put_record(record)
         self.database.delete_record(record_id)
+        ###self.maybe_die()  # should be able to survive couchdb death
         deleted_record = self.database._server[self.dbname][record_id]
         self.assert_(deleted_record['application_annotations']['Ubuntu One'][
             'private_application_annotations']['deleted'])
@@ -154,6 +184,7 @@
         record = Record({'record_number': 0}, record_type="http://example.com/")
         record_id = self.database.put_record(record)
         self.database.delete_record(record_id)
+        self.maybe_die()  # should be able to survive couchdb death
         retrieved_record = self.database.get_record(record_id)
         self.assertEqual(None, retrieved_record)
 
@@ -162,6 +193,7 @@
         record = Record({'record_number': 0}, record_type="http://example.com/")
         self.assert_(not self.database.record_exists("ThisMustNotExist"))
         record_id = self.database.put_record(record)
+        self.maybe_die()  # should be able to survive couchdb death
         self.assert_(self.database.record_exists(record_id))
         self.database.delete_record(record_id)
         self.assert_(not self.database.record_exists(record_id))
@@ -171,12 +203,14 @@
         dictionary = {'record_number': 0, 'field1': 1, 'field2': 2}
         record = Record(dictionary, record_type="http://example.com/")
         record_id = self.database.put_record(record)
+        self.maybe_die()  # should be able to survive couchdb death
         # manipulate the database 'out of view'
         non_working_copy = self.database.get_record(record_id)
         non_working_copy['field2'] = 22
         non_working_copy['field3'] = 3
         self.database.put_record(non_working_copy)
         self.database.update_fields(record_id, {'field1': 11})
+        self.maybe_die()  # should be able to survive couchdb death
         working_copy = self.database.get_record(record_id)
         self.assertEqual(0, working_copy['record_number'])
         self.assertEqual(11, working_copy['field1'])
@@ -200,10 +234,12 @@
         self.assertRaises(
             KeyError, self.database.delete_view, view2_name, design_doc)
         self.database.add_view(view1_name, map_js, reduce_js, design_doc)
+        self.maybe_die()  # should be able to survive couchdb death
         self.database.add_view(view2_name, map_js, reduce_js, design_doc)
         self.database.delete_view(view1_name, design_doc)
         self.assertRaises(
             KeyError, self.database.delete_view, view1_name, design_doc)
+        self.maybe_die()  # should be able to survive couchdb death
         self.database.delete_view(view2_name, design_doc)
         self.assertRaises(
             KeyError, self.database.delete_view, view2_name, design_doc)
@@ -237,6 +273,7 @@
             record_ids_we_care_about.remove(row.id)
             self.assertFalse(row_is_deleted(row))
 
+        ###self.maybe_die()  # should be able to survive couchdb death
         self.assertTrue(len(record_ids_we_care_about) == 0, "expected zero")
 
         self.assertRaises(KeyError, self.database.get_records,
@@ -250,6 +287,7 @@
         map_js = """function(doc) { emit(doc._id, null) }"""
         self.database.add_view(view_name, map_js, None, design_doc)
 
+        ###self.maybe_die()  # should be able to survive couchdb death
         self.assertEqual(self.database.list_views(design_doc), [view_name])
         self.database.delete_view(view_name, design_doc)
 
@@ -257,11 +295,13 @@
 
     def test_get_view_by_type_new_but_already(self):
         self.database.get_records(create_view=True)
+        self.maybe_die()  # should be able to survive couchdb death
         self.database.get_records(create_view=True)
         # No exceptions on second run?  Yay.
 
     def test_get_view_by_type_createxcl_fail(self):
         self.database.get_records(create_view=True)
+        self.maybe_die()  # should be able to survive couchdb death
         self.assertRaises(KeyError, self.database.get_records, create_view=None)
 
     def test_get_changes(self):
@@ -321,6 +361,8 @@
         # Ensure time is same.
         self.assertEqual(saved_time, self.database._changes_last_used)
 
+        ###self.maybe_die()  # should be able to survive couchdb death
+
         # Next time we run, we get the same event again.
         # Consume queued changes.
         count = self.database.report_changes(rep)
@@ -370,6 +412,7 @@
         constructed_record.attach(content, "nu/mbe/rs", "text/plain")
         constructed_record.attach("string", "another document", "text/plain")
 
+        self.maybe_die()  # should be able to survive couchdb death
         constructed_record.attach("XXXXXXXXX", "never used", "text/plain")
         constructed_record.detach("never used")  # detach works before commit.
 
@@ -382,6 +425,7 @@
         self.assertRaises(KeyError, constructed_record.attach, content,
                 "another document", "text/x-rst")
 
+        self.maybe_die()  # should be able to survive couchdb death
         record_id = self.database.put_record(constructed_record)
         retrieved_record = self.database.get_record(record_id)
 
@@ -402,6 +446,7 @@
         # get new
         retrieved_record = self.database.get_record(record_id)
 
+        self.maybe_die()  # should be able to survive couchdb death
         # We can get a list of attachments.
         self.assertEqual(set(retrieved_record.list_attachments()),
                 set(["nu/mbe/rs", "Document"]))
@@ -424,6 +469,7 @@
         self.assertEqual(out_data, content.getvalue())
         self.assertEqual(out_content_type, "text/plain")
 
+        self.maybe_die()  # should be able to survive couchdb death
         # Asking for a named document that does not exist causes KeyError.
         self.assertRaises(KeyError, retrieved_record.attachment_data,
                 "NoExist")
@@ -454,6 +500,7 @@
         # ordinary requests are in key order
         self.assertEqual(data, sorted(data))
 
+        ###self.maybe_die()  # should be able to survive couchdb death
         # now request descending order and confirm that it *is* descending
         descdata = [i.key for i in
             list(self.database.execute_view(view1_name, design_doc,
@@ -513,3 +560,36 @@
             self.fail()
         except FieldsConflict, e:
             self.assertEqual({('field1',): (22, 11)}, e.conflicts)
+
+
+class TestServerDiesSegv(TestCouchDatabase):
+    def get_test_context(self):
+        try:
+            return self.ctx
+        except AttributeError:
+            self.ctx = test_environment.create_new_test_environment()
+            return self.ctx
+
+    def maybe_die(self):
+        time.sleep(2)
+        pid = find_pid(start_if_not_running=False, ctx=self.get_test_context())
+        if pid is None:
+            print "couchdb has already quit!  That's unexpected."
+            return
+        print "DIE, process", pid, "!"
+        os.kill(pid, signal.SIGSEGV)
+        self.wait_until_server_dead(pid=pid)
+
+
+class TestServerDiesNormal(TestCouchDatabase):
+    def get_test_context(self):
+        try:
+            return self.ctx
+        except AttributeError:
+            self.ctx = test_environment.create_new_test_environment()
+            return self.ctx
+
+    def maybe_die(self):
+        time.sleep(2)
+        stop_couchdb(ctx=self.get_test_context())
+        self.wait_until_server_dead()

=== modified file 'desktopcouch/stop_local_couchdb.py'
--- desktopcouch/stop_local_couchdb.py	2010-02-04 22:31:41 +0000
+++ desktopcouch/stop_local_couchdb.py	2010-08-25 21:55:59 +0000
@@ -25,7 +25,7 @@
 from desktopcouch import local_files
 
 def stop_couchdb(ctx=local_files.DEFAULT_CONTEXT):
-    local_exec = ctx.couch_exec_command + ["-k"]
+    local_exec = ctx.couch_exec_command + ["-d"]
     try:
         retcode = subprocess.call(local_exec, shell=False)
         if retcode < 0: