Reads in absence of writer

Author: Zhen

neo4j/v1/routing.py

@@ -132,11 +132,12 @@ def __init__(self, routers=(), readers=(), writers=(), ttl=0):
         self.last_updated_time = self.timer()
         self.ttl = ttl
 
-    def is_fresh(self):
+    def is_fresh(self, access_mode):
         """ Indicator for whether routing information is still usable.
         """
         expired = self.last_updated_time + self.ttl <= self.timer()
-        return not expired and len(self.routers) > 1 and self.readers and self.writers
+        has_server_for_mode = (access_mode == READ_ACCESS and self.readers) or (access_mode == WRITE_ACCESS and self.writers)
+        return not expired and len(self.routers) >= 1 and has_server_for_mode
 
     def update(self, new_routing_table):
         """ Update the current routing table with new routing information
@@ -162,6 +163,7 @@ def __init__(self, connector, initial_address, routing_context, *routers):
         self.initial_address = initial_address
         self.routing_context = routing_context
         self.routing_table = RoutingTable(routers)
+        self.missing_writer = False
         self.refresh_lock = Lock()
 
     def routing_info_procedure(self, connection):
@@ -211,6 +213,11 @@ def fetch_routing_table(self, address):
         num_readers = len(new_routing_table.readers)
         num_writers = len(new_routing_table.writers)
 
+        # No writers are available. This likely indicates a temporary state,
+        # such as leader switching, so we should not signal an error.
+        # When no writers available, then we flag we are reading in absence of writer
+        self.missing_writer = (num_writers == 0)
+
         # No routers
         if num_routers == 0:
             raise ProtocolError("No routing servers returned from server %r" % (address,))
@@ -219,12 +226,6 @@ def fetch_routing_table(self, address):
         if num_readers == 0:
             raise ProtocolError("No read servers returned from server %r" % (address,))
 
-        # No writers
-        if num_writers == 0:
-            # No writers are available. This likely indicates a temporary state,
-            # such as leader switching, so we should not signal an error.
-            return None
-
         # At least one of each is fine, so return this table
         return new_routing_table
 
@@ -245,21 +246,30 @@ def update_routing_table(self):
         """
         # copied because it can be modified
         copy_of_routers = list(self.routing_table.routers)
+
+        has_tried_initial_routers = False
+        if self.missing_writer:
+            has_tried_initial_routers = True
+            if self.update_routing_table_with_routers(resolve(self.initial_address)):
+                return
+
         if self.update_routing_table_with_routers(copy_of_routers):
             return
 
-        initial_routers = resolve(self.initial_address)
-        for router in copy_of_routers:
-            if router in initial_routers:
-                initial_routers.remove(router)
-        if initial_routers:
-            if self.update_routing_table_with_routers(initial_routers):
-                return
+        if not has_tried_initial_routers:
+            initial_routers = resolve(self.initial_address)
+            for router in copy_of_routers:
+                if router in initial_routers:
+                    initial_routers.remove(router)
+            if initial_routers:
+                if self.update_routing_table_with_routers(initial_routers):
+                    return
+
 
         # None of the routers have been successful, so just fail
         raise ServiceUnavailable("Unable to retrieve routing information")
 
-    def refresh_routing_table(self):
+    def ensure_routing_table(self, access_mode):
         """ Update the routing table if stale.
 
         This method performs two freshness checks, before and after acquiring
@@ -272,10 +282,13 @@ def refresh_routing_table(self):
 
         :return: `True` if an update was required, `False` otherwise.
         """
-        if self.routing_table.is_fresh():
+        if self.routing_table.is_fresh(access_mode):
             return False
         with self.refresh_lock:
-            if self.routing_table.is_fresh():
+            if self.routing_table.is_fresh(access_mode):
+                if access_mode == READ_ACCESS:
+                    # if reader is fresh but writers is not fresh, then we are reading in absence of writer
+                    self.missing_writer = not self.routing_table.is_fresh(WRITE_ACCESS)
                 return False
             self.update_routing_table()
             return True
@@ -289,18 +302,21 @@ def acquire(self, access_mode=None):
             server_list = self.routing_table.writers
         else:
             raise ValueError("Unsupported access mode {}".format(access_mode))
+
+        self.ensure_routing_table(access_mode)
         while True:
-            address = None
-            while address is None:
-                self.refresh_routing_table()
-                address = next(server_list)
+            address = next(server_list)
+            if address is None:
+                break
             try:
                 connection = self.acquire_direct(address)  # should always be a resolved address
                 connection.Error = SessionExpired
             except ServiceUnavailable:
                 self.remove(address)
             else:
                 return connection
+        raise SessionExpired("Failed to obtain connection towards '" + access_mode +
+                             "' server. Known routing table is: '" + self.routing_table + "'.")
 
     def remove(self, address):
         """ Remove an address from the connection pool, if present, closing

test/stub/test_routing.py

@@ -156,6 +156,7 @@ def test_should_get_table_from_router(self):
                 assert table.readers == {("127.0.0.1", 9004), ("127.0.0.1", 9005)}
                 assert table.writers == {("127.0.0.1", 9006)}
                 assert table.ttl == 300
+                assert not pool.missing_writer
 
     def test_null_info_should_return_null_table(self):
         address = ("127.0.0.1", 9001)
@@ -177,12 +178,17 @@ def test_no_readers_should_raise_protocol_error(self):
                 with self.assertRaises(ProtocolError):
                     _ = pool.fetch_routing_table(address)
 
-    def test_no_writers_should_return_null_table(self):
+    def test_no_writers_should_return_table_with_no_writer(self):
         with StubCluster({9001: "router_no_writers.script"}):
             address = ("127.0.0.1", 9001)
             with RoutingPool() as pool:
                 table = pool.fetch_routing_table(address)
-                assert table is None
+                assert table.routers == {("127.0.0.1", 9001), ("127.0.0.1", 9002),
+                                         ("127.0.0.1", 9003)}
+                assert table.readers == {("127.0.0.1", 9004), ("127.0.0.1", 9005)}
+                assert not table.writers
+                assert table.ttl == 300
+                assert pool.missing_writer
 
 
 class RoutingConnectionPoolUpdateRoutingTableTestCase(StubTestCase):
@@ -211,6 +217,20 @@ def test_roll_back_to_initial_server_if_failed_update_with_existing_routers(self
                 assert table.writers == {("127.0.0.1", 9006)}
                 assert table.ttl == 300
 
+    def test_try_initial_server_first_if_missing_writer(self):
+        with StubCluster({9001: "router.script"}):
+            initial_address = ("127.0.0.1", 9001)
+            with RoutingConnectionPool(connector, initial_address, {}) as pool:
+                pool.missing_writer = True
+                pool.update_routing_table()
+                table = pool.routing_table
+                assert table.routers == {("127.0.0.1", 9001), ("127.0.0.1", 9002),
+                                         ("127.0.0.1", 9003)}
+                assert table.readers == {("127.0.0.1", 9004), ("127.0.0.1", 9005)}
+                assert table.writers == {("127.0.0.1", 9006)}
+                assert table.ttl == 300
+                assert not pool.missing_writer
+
     def test_update_with_no_routers_should_signal_service_unavailable(self):
         with RoutingPool() as pool:
             with self.assertRaises(ServiceUnavailable):
@@ -226,7 +246,7 @@ def _test_server_outcome(self, server_outcomes, overall_outcome):
         routers = []
         for port, outcome in enumerate(server_outcomes, 9001):
             if outcome is None:
-                servers[port] = "router_no_writers.script"
+                servers[port] = "rude_router.script"
             elif outcome is RoutingTable:
                 servers[port] = "router.script"
             elif outcome is ServiceUnavailable:
@@ -251,26 +271,37 @@ def _test_server_outcome(self, server_outcomes, overall_outcome):
                     assert False, "Unexpected overall outcome %r" % overall_outcome
 
 
-class RoutingConnectionPoolRefreshRoutingTableTestCase(StubTestCase):
+class RoutingConnectionPoolEnsureRoutingTableTestCase(StubTestCase):
     def test_should_update_if_stale(self):
         with StubCluster({9001: "router.script"}):
             address = ("127.0.0.1", 9001)
             with RoutingPool(address) as pool:
                 first_updated_time = pool.routing_table.last_updated_time
                 pool.routing_table.ttl = 0
-                pool.refresh_routing_table()
+                pool.ensure_routing_table(WRITE_ACCESS)
                 second_updated_time = pool.routing_table.last_updated_time
                 assert second_updated_time != first_updated_time
+                assert not pool.missing_writer
 
     def test_should_not_update_if_fresh(self):
         with StubCluster({9001: "router.script"}):
             address = ("127.0.0.1", 9001)
             with RoutingPool(address) as pool:
-                pool.refresh_routing_table()
+                pool.ensure_routing_table(WRITE_ACCESS)
                 first_updated_time = pool.routing_table.last_updated_time
-                pool.refresh_routing_table()
+                pool.ensure_routing_table(WRITE_ACCESS)
                 second_updated_time = pool.routing_table.last_updated_time
                 assert second_updated_time == first_updated_time
+                assert not pool.missing_writer
+
+    def test_should_flag_reading_without_writer(self):
+        with StubCluster({9001: "router_no_writers.script"}):
+            address = ("127.0.0.1", 9001)
+            with RoutingPool(address) as pool:
+                assert not pool.routing_table.is_fresh(READ_ACCESS)
+                assert not pool.routing_table.is_fresh(WRITE_ACCESS)
+                pool.ensure_routing_table(READ_ACCESS)
+                assert pool.missing_writer
 
     # TODO: fix flaky test
     # def test_concurrent_refreshes_should_not_block_if_fresh(self):
@@ -372,65 +403,90 @@ def test_should_refresh(self):
         with StubCluster({9001: "router.script", 9004: "empty.script"}):
             address = ("127.0.0.1", 9001)
             with RoutingPool(address) as pool:
-                assert not pool.routing_table.is_fresh()
+                assert not pool.routing_table.is_fresh(READ_ACCESS)
                 _ = pool.acquire(access_mode=READ_ACCESS)
-                assert pool.routing_table.is_fresh()
+                assert pool.routing_table.is_fresh(READ_ACCESS)
+                assert not pool.missing_writer
 
     def test_connected_to_reader(self):
         with StubCluster({9001: "router.script", 9004: "empty.script"}):
             address = ("127.0.0.1", 9001)
             with RoutingPool(address) as pool:
-                assert not pool.routing_table.is_fresh()
+                assert not pool.routing_table.is_fresh(READ_ACCESS)
                 connection = pool.acquire(access_mode=READ_ACCESS)
                 assert connection.server.address in pool.routing_table.readers
+                assert not pool.missing_writer
 
     def test_should_retry_if_first_reader_fails(self):
         with StubCluster({9001: "router.script",
                           9004: "fail_on_init.script",
                           9005: "empty.script"}):
             address = ("127.0.0.1", 9001)
             with RoutingPool(address) as pool:
-                assert not pool.routing_table.is_fresh()
+                assert not pool.routing_table.is_fresh(READ_ACCESS)
                 _ = pool.acquire(access_mode=READ_ACCESS)
                 assert ("127.0.0.1", 9004) not in pool.routing_table.readers
                 assert ("127.0.0.1", 9005) in pool.routing_table.readers
 
+    def test_should_connect_to_read_in_absent_of_writer(self):
+        with StubCluster({9001: "router_no_writers.script", 9004: "empty.script"}):
+            address = ("127.0.0.1", 9001)
+            with RoutingPool(address) as pool:
+                assert not pool.routing_table.is_fresh(READ_ACCESS)
+                connection = pool.acquire(access_mode=READ_ACCESS)
+                assert connection.server.address in pool.routing_table.readers
+                assert not pool.routing_table.is_fresh(WRITE_ACCESS)
+                assert pool.missing_writer
+
 
 class RoutingConnectionPoolAcquireForWriteTestCase(StubTestCase):
     def test_should_refresh(self):
         with StubCluster({9001: "router.script", 9006: "empty.script"}):
             address = ("127.0.0.1", 9001)
             with RoutingPool(address) as pool:
-                assert not pool.routing_table.is_fresh()
+                assert not pool.routing_table.is_fresh(WRITE_ACCESS)
                 _ = pool.acquire(access_mode=WRITE_ACCESS)
-                assert pool.routing_table.is_fresh()
+                assert pool.routing_table.is_fresh(WRITE_ACCESS)
+                assert not pool.missing_writer
 
     def test_connected_to_writer(self):
         with StubCluster({9001: "router.script", 9006: "empty.script"}):
             address = ("127.0.0.1", 9001)
             with RoutingPool(address) as pool:
-                assert not pool.routing_table.is_fresh()
+                assert not pool.routing_table.is_fresh(WRITE_ACCESS)
                 connection = pool.acquire(access_mode=WRITE_ACCESS)
                 assert connection.server.address in pool.routing_table.writers
+                assert not pool.missing_writer
 
     def test_should_retry_if_first_writer_fails(self):
         with StubCluster({9001: "router_with_multiple_writers.script",
                           9006: "fail_on_init.script",
                           9007: "empty.script"}):
             address = ("127.0.0.1", 9001)
             with RoutingPool(address) as pool:
-                assert not pool.routing_table.is_fresh()
+                assert not pool.routing_table.is_fresh(WRITE_ACCESS)
                 _ = pool.acquire(access_mode=WRITE_ACCESS)
                 assert ("127.0.0.1", 9006) not in pool.routing_table.writers
                 assert ("127.0.0.1", 9007) in pool.routing_table.writers
 
+    def test_should_error_to_writer_in_absent_of_reader(self):
+        with StubCluster({9001: "router_no_readers.script"}):
+            address = ("127.0.0.1", 9001)
+            with RoutingPool(address) as pool:
+                assert not pool.routing_table.is_fresh(WRITE_ACCESS)
+                with self.assertRaises(ProtocolError):
+                    _ = pool.acquire(access_mode=WRITE_ACCESS)
+                assert not pool.routing_table.is_fresh(READ_ACCESS)
+                assert not pool.routing_table.is_fresh(WRITE_ACCESS)
+                assert not pool.missing_writer
+
 
 class RoutingConnectionPoolRemoveTestCase(StubTestCase):
     def test_should_remove_router_from_routing_table_if_present(self):
         with StubCluster({9001: "router.script"}):
             address = ("127.0.0.1", 9001)
             with RoutingPool(address) as pool:
-                pool.refresh_routing_table()
+                pool.ensure_routing_table(WRITE_ACCESS)
                 target = ("127.0.0.1", 9001)
                 assert target in pool.routing_table.routers
                 pool.remove(target)
@@ -440,7 +496,7 @@ def test_should_remove_reader_from_routing_table_if_present(self):
         with StubCluster({9001: "router.script"}):
             address = ("127.0.0.1", 9001)
             with RoutingPool(address) as pool:
-                pool.refresh_routing_table()
+                pool.ensure_routing_table(WRITE_ACCESS)
                 target = ("127.0.0.1", 9004)
                 assert target in pool.routing_table.readers
                 pool.remove(target)
@@ -450,7 +506,7 @@ def test_should_remove_writer_from_routing_table_if_present(self):
         with StubCluster({9001: "router.script"}):
             address = ("127.0.0.1", 9001)
             with RoutingPool(address) as pool:
-                pool.refresh_routing_table()
+                pool.ensure_routing_table(WRITE_ACCESS)
                 target = ("127.0.0.1", 9006)
                 assert target in pool.routing_table.writers
                 pool.remove(target)
@@ -460,6 +516,6 @@ def test_should_not_fail_if_absent(self):
         with StubCluster({9001: "router.script"}):
             address = ("127.0.0.1", 9001)
             with RoutingPool(address) as pool:
-                pool.refresh_routing_table()
+                pool.ensure_routing_table(WRITE_ACCESS)
                 target = ("127.0.0.1", 9007)
                 pool.remove(target)

test/stub/test_routingdriver.py

@@ -198,4 +198,20 @@ def test_should_call_get_routing_table_with_context(self):
                     result = session.run("RETURN $x", {"x": 1})
                     for record in result:
                         assert record["x"] == 1
-                    assert result.summary().server.address == ('127.0.0.1', 9002)
+                    assert result.summary().server.address == ('127.0.0.1', 9002)
+
+    def test_should_serve_read_when_missing_writer(self):
+        with StubCluster({9001: "router_no_writers.script", 9005: "return_1.script"}):
+            uri = "bolt+routing://127.0.0.1:9001"
+            with GraphDatabase.driver(uri, auth=self.auth_token, encrypted=False) as driver:
+                with driver.session(READ_ACCESS) as session:
+                    result = session.run("RETURN $x", {"x": 1})
+                    for record in result:
+                        assert record["x"] == 1
+                    assert result.summary().server.address == ('127.0.0.1', 9005)
+
+    def test_should_error_when_missing_reader(self):
+        with StubCluster({9001: "router_no_readers.script"}):
+            uri = "bolt+routing://127.0.0.1:9001"
+            with self.assertRaises(ProtocolError):
+                GraphDatabase.driver(uri, auth=self.auth_token, encrypted=False)