More fixes and boltkit tests for routing

Author: lutovich

package.json

@@ -52,10 +52,12 @@
     "lolex": "^1.5.2",
     "merge-stream": "^1.0.0",
     "minimist": "^1.2.0",
+    "mustache": "^2.3.0",
     "phantomjs-prebuilt": "^2.1.7 ",
     "run-sequence": "^1.1.4",
     "semver": "^5.3.0",
     "through2": "~2.0.0",
+    "tmp": "0.0.31",
     "vinyl-buffer": "^1.0.0",
     "vinyl-source-stream": "^1.1.0"
   },

src/v1/internal/get-servers-util.js

@@ -31,7 +31,7 @@ export default class GetServersUtil {
       session.close();
       return result.records;
     }).catch(error => {
-      if (error.code === PROCEDURE_NOT_FOUND_CODE) {
+      if (this._isProcedureNotFoundError(error)) {
         // throw when getServers procedure not found because this is clearly a configuration issue
         throw newError('Server ' + routerAddress + ' could not perform routing. ' +
           'Make sure you are connecting to a causal cluster', SERVICE_UNAVAILABLE);
@@ -92,4 +92,16 @@ export default class GetServersUtil {
         PROTOCOL_ERROR);
     }
   }
+
+  _isProcedureNotFoundError(error) {
+    let errorCode = error.code;
+    if (!errorCode) {
+      try {
+        errorCode = error.fields[0].code;
+      } catch (e) {
+        errorCode = 'UNKNOWN';
+      }
+    }
+    return errorCode === PROCEDURE_NOT_FOUND_CODE;
+  }
 }

src/v1/internal/rediscovery.js

@@ -47,11 +47,8 @@ export default class Rediscovery {
 
       Rediscovery._assertNonEmpty(routers, 'routers', routerAddress);
       Rediscovery._assertNonEmpty(readers, 'readers', routerAddress);
-
-      if (writers.isEmpty()) {
-        // retrieved routing table has no writers, next router should be queried
-        return null;
-      }
+      // case with no writers is processed higher in the promise chain because only RoutingDriver knows
+      // how to deal with such table and how to treat router that returned such table
 
       return new RoutingTable(routers, readers, writers, expirationTime);
     });

src/v1/internal/routing-table.js

@@ -38,6 +38,14 @@ export default class RoutingTable {
     this.writers.remove(address);
   }
 
+  forgetRouter(address) {
+    this.routers.remove(address);
+  }
+
+  forgetWriter(address) {
+    this.writers.remove(address);
+  }
+
   serversDiff(otherRoutingTable) {
     const oldServers = new Set(this._allServers());
     const newServers = otherRoutingTable._allServers();

src/v1/routing-driver.js

@@ -70,10 +70,10 @@ class RoutingDriver extends Driver {
         let url = 'UNKNOWN';
         if (conn) {
           url = conn.url;
-          this._routingTable.writers.remove(conn.url);
+          this._routingTable.forgetWriter(conn.url);
         } else {
           connectionPromise.then((conn) => {
-            this._routingTable.writers.remove(conn.url);
+            this._routingTable.forgetWriter(conn.url);
           }).catch(() => {/*ignore*/});
         }
         return newError("No longer possible to write to server at " + url, SESSION_EXPIRED);
@@ -118,35 +118,43 @@ class RoutingDriver extends Driver {
     const refreshedTablePromise = knownRouters.reduce((refreshedTablePromise, currentRouter, currentIndex) => {
       return refreshedTablePromise.then(newRoutingTable => {
         if (newRoutingTable) {
-          // correct routing table was fetched, just return it
-          return newRoutingTable
-        }
-
-        // returned routing table was undefined, this means a connection error happened and we need to forget the
-        // previous router and try the next one
-        const previousRouter = knownRouters[currentIndex - 1];
-        if (previousRouter) {
-          this._forget(previousRouter);
+          if (!newRoutingTable.writers.isEmpty()) {
+            // valid routing table was fetched - just return it, try next router otherwise
+            return newRoutingTable;
+          }
+        } else {
+          // returned routing table was undefined, this means a connection error happened and we need to forget the
+          // previous router and try the next one
+          const previousRouter = knownRouters[currentIndex - 1];
+          if (previousRouter) {
+            currentRoutingTable.forgetRouter(previousRouter);
+          }
         }
 
-        const connection = this._pool.acquire(currentRouter);
-        const connectionPromise = Promise.resolve(connection);
-        const session = this._createSession(connectionPromise, this._releaseConnection(connectionPromise));
-
         // try next router
+        const session = this._createSessionForRediscovery(currentRouter);
         return this._rediscovery.lookupRoutingTableOnRouter(session, currentRouter);
       })
     }, Promise.resolve(null));
 
     return refreshedTablePromise.then(newRoutingTable => {
-      if (newRoutingTable) {
+      if (newRoutingTable && !newRoutingTable.writers.isEmpty()) {
         this._updateRoutingTable(newRoutingTable);
         return newRoutingTable
       }
       throw newError('Could not perform discovery. No routing servers available.', SERVICE_UNAVAILABLE);
     });
   }
 
+  _createSessionForRediscovery(routerAddress) {
+    const connection = this._pool.acquire(routerAddress);
+    const connectionPromise = Promise.resolve(connection);
+    // error transformer here is a no-op unlike the one in a regular session, this is so because errors are
+    // handled in the rediscovery promise chain and we do not need to do this in the error transformer
+    const errorTransformer = error => error;
+    return new RoutingSession(connectionPromise, this._releaseConnection(connectionPromise), errorTransformer);
+  }
+
   _forget(url) {
     this._routingTable.forget(url);
     this._pool.purge(url);

test/internal/rediscovery.test.js

@@ -129,7 +129,7 @@ describe('rediscovery', () => {
     });
   });
 
-  it('should return null when no writers', done => {
+  it('should return routing table when no writers', done => {
     const util = new FakeGetServersUtil({
       callGetServers: () => [new Record(['a'], ['aaa'])],
       parseTtl: () => int(42),
@@ -143,7 +143,8 @@ describe('rediscovery', () => {
     });
 
     lookupRoutingTableOnRouter(util).then(routingTable => {
-      expect(routingTable).toBeNull();
+      expect(routingTable).toBeDefined();
+      expect(routingTable).not.toBeNull();
       done();
     });
   });

test/internal/routing-table.test.js

@@ -87,6 +87,26 @@ describe('routing-table', () => {
     expect(table.writers.toArray()).toEqual([]);
   });
 
+  it('should forget router', () => {
+    const table = createTable([1, 2], [1, 3], [4, 1], notExpired());
+
+    table.forgetRouter(1);
+
+    expect(table.routers.toArray()).toEqual([2]);
+    expect(table.readers.toArray()).toEqual([1, 3]);
+    expect(table.writers.toArray()).toEqual([4, 1]);
+  });
+
+  it('should forget writer', () => {
+    const table = createTable([1, 2, 3], [2, 1, 5], [5, 1], notExpired());
+
+    table.forgetWriter(1);
+
+    expect(table.routers.toArray()).toEqual([1, 2, 3]);
+    expect(table.readers.toArray()).toEqual([2, 1, 5]);
+    expect(table.writers.toArray()).toEqual([5]);
+  });
+
   it('should return all servers in diff when other table is empty', () => {
     const oldTable = createTable([1, 2], [3, 4], [5, 6], notExpired());
     const newTable = createTable([], [], [], notExpired());

test/resources/boltkit/handle_empty_get_servers_response.script renamed to test/resources/boltkit/empty_get_servers_response.script

@@ -0,0 +1,10 @@
+!: AUTO INIT
+!: AUTO RESET
+!: AUTO PULL_ALL
+
+C: RUN "CALL dbms.cluster.routing.getServers" {}
+   PULL_ALL
+S: SUCCESS {"fields": ["ttl", "servers"]}
+   RECORD [9223372036854775807, [{"addresses": ["127.0.0.1:9001"],"role": "WRITE"}, {"addresses": ["127.0.0.1:9002","127.0.0.1:9003"], "role": "READ"},{"addresses": ["127.0.0.1:9001","127.0.0.1:9002"], "role": "ROUTE"}]]
+   RECORD [9223372036854775807, [{"addresses": ["127.0.0.1:9005"],"role": "WRITE"}, {"addresses": ["127.0.0.1:9005","127.0.0.1:9003"], "role": "READ"},{"addresses": ["127.0.0.1:9001"], "role": "ROUTE"}]]
+   SUCCESS {}

test/resources/boltkit/multiple_records_get_servers.script

@@ -0,0 +1,9 @@
+!: AUTO INIT
+!: AUTO RESET
+!: AUTO PULL_ALL
+
+C: RUN "CALL dbms.cluster.routing.getServers" {}
+   PULL_ALL
+S: SUCCESS {"fields": ["ttl", "servers"]}
+   RECORD [9223372036854775807, [{"addresses": ["127.0.0.1:9001"],"role": "WRITE"},{"addresses": ["127.0.0.1:9001","127.0.0.1:9002","127.0.0.1:9003"], "role": "ROUTE"}]]
+   SUCCESS {}