Error out waiting clients on close, handle too many connections

lotofconnstest.js

@@ -0,0 +1,56 @@
+#! node
+const MemJS = require(".");
+
+async function nextTick() {
+  return new Promise((resolve) => setImmediate(resolve));
+}
+
+const clients = [];
+
+async function body() {
+  // local memcached is default 1024 max connections
+  for (let i = 0; i < 1025; i++) {
+    const client = MemJS.Client.create(undefined, {timeout: 1, connTimeout: 2, retries: 1});
+    console.log("created", i);
+
+    client.servers.forEach((server) => {
+      server.onConnect((sock) => {
+        console.log("connected", i);
+
+        sock.once("close", (err) => {
+          console.log("closed", i, err);
+        })
+        sock.once("error", (err) => {
+          console.log("error handler", i, err);
+        })
+      })
+    })
+
+    try {
+      await client.get("foo");
+      console.log(i)
+    } catch (error) {
+      console.log("error", error, i);
+    }
+
+    clients.push(client);
+  }
+
+}
+
+async function main() {
+  try {
+    await body();
+    console.log("smoketest ok");
+  } catch (error) {
+    console.error("fatal", error);
+    process.exit(1);
+  } finally {
+    clients.forEach((client) => {
+      client.servers.forEach((server) => server._socket?.removeAllListeners("close"));
+      client.quit()
+    });
+  }
+}
+
+main();

src/memjs/server.ts

@@ -151,7 +151,14 @@ export class Server extends events.EventEmitter {
     return this.responseBuffer;
   }
   responseHandler(dataBuf: Buffer) {
-    let response = parseMessage(this.appendToBuffer(dataBuf));
+    let response: Message | false;
+    try {
+      response = parseMessage(this.appendToBuffer(dataBuf));
+    } catch (e) {
+      this.error(e as Error);
+      return;
+    }
+
     let respLength: number;
     while (response) {
       if (response.header.opcode === 0x20) {
@@ -218,6 +225,9 @@ export class Server extends events.EventEmitter {
       });
 
       self._socket.on("close", function () {
+        if (Object.keys(self.errorCallbacks).length > 0) {
+          self.error(new Error("socket closed unexpectedly."));
+        }
         self.connected = false;
         if (self.timeoutSet) {
           self._socket?.setTimeout(0);

src/memjs/utils.ts

@@ -144,10 +144,22 @@ export interface Message {
   extras: Buffer;
 }
 
+// Error message from memcached when it rejects a request for having too many
+// open connections
+// https://github.com/memcached/memcached/blob/efee763c93249358ea5b3b42c7fd4e57e2599c30/memcached.c#L3044
+const ERROR_TOO_MANY_OPEN_CONNECTIONS = "ERROR Too many open connections\r\n";
+
 export const parseMessage = function (dataBuf: Buffer): Message | false {
   if (dataBuf.length < 24) {
     return false;
   }
+
+  if (dataBuf.length === ERROR_TOO_MANY_OPEN_CONNECTIONS.length) {
+    if (dataBuf.toString() === ERROR_TOO_MANY_OPEN_CONNECTIONS) {
+      throw new Error("ERROR Too many open connections");
+    }
+  }
+
   const responseHeader = header.fromBuffer(dataBuf);
 
   if (

src/test/server_test.js

@@ -35,6 +35,20 @@ test('ResponseHandler with authentication error', function(t) {
   t.end();
 });
 
+test('ResponseHandler with too many connections error', function(t) {
+  const server = new MemJS.Server('localhost', 11211);
+
+  server.onError('test', function(err) {
+    t.equal('ERROR Too many open connections', err.message);
+  });
+
+  const responseBuf = Buffer.from('ERROR Too many open connections\r\n');
+
+  server.responseHandler(responseBuf);
+
+  t.end();
+});
+
 test('Authenticate', function(t) {
   const expectedBuf = makeRequestBuffer(0x21, 'PLAIN', '', '\0user1\0password');
   const dummySocket = {