TCP socket yield by http 'upgrade' event stops emitting data

[jcoglan]

I maintain the websocket-driver module. This module implements the WebSocket protocol but does not contain any of its own I/O logic. The user hooks up the driver to an I/O stream of their choosing.

The following script sets up two servers that should be functionally identical, one using the http.Server upgrade event, and one using a net server. Each server runs the following actions:

• Print all data coming in over the socket to stdout
• Create a WebSocket driver to handle the connection
• Pipe data back and forth between socket and driver.io
• When a message is received, echo it back
• When the protocol is closed, print the code and reason

Following the servers is a function testServer() to test each server using a client. This function makes a TCP connection to a port, attaches a driver to that connection. When the connection is established, a random message of a certain size is sent. When the message is received, the client closes the connection. And when the connection becomes closed, we print the code and reason.

var crypto    = require('crypto'),
    http      = require('http'),
    net       = require('net'),
    websocket = require('websocket-driver');


var httpServer = http.createServer();

httpServer.on('upgrade', function(request, socket, body) {
  socket.on('data', function(data) {
    console.log('[http received]    ', data);
  });

  var driver = websocket.http(request);

  driver.io.write(body);
  socket.pipe(driver.io).pipe(socket);

  driver.on('message', function(message) { driver.binary(message.data) });

  driver.on('close', function(close) {
    console.log('[http close]       ', close.code, close.reason);
  });

  driver.start();
});

httpServer.listen(4007);


var tcpServer = net.createServer(function(socket) {
  socket.on('data', function(data) {
    console.log('[tcp received]     ', data);
  });

  var driver = websocket.server();

  driver.on('connect', function() {
    if (websocket.isWebSocket(driver)) driver.start();
  });

  socket.pipe(driver.io).pipe(socket);

  driver.on('message', function(message) { driver.binary(message.data) });

  driver.on('close', function(close) {
    console.log('[tcp close]        ', close.code, close.reason);
  });

  driver.start();
});

tcpServer.listen(4008);


function testServer(port) {
  var client  = net.connect(port, 'localhost'),
      driver  = websocket.client('ws://localhost:' + port + '/'),
      payload = crypto.randomBytes(16379);

  client.pipe(driver.io).pipe(client);

  driver.on('open', function() {
    console.log('[client open]      ', port);
    driver.binary(payload);
  });

  driver.on('message', function(message) {
    console.log('[client received]  ', message.data.length);
    driver.close('test done', 1000);
  });

  driver.on('close', function(close) {
    console.log('[client close]     ', close.code, close.reason);
    client.end();
  });

  driver.start();
}

testServer(4007);

The script as given sents a message of 16,379 bytes to the HTTP server on port 4007. Here's what I get when I run this program:

[client open]       4007
[http received]     <Buffer 82 fe 3f fb d2 3c 92 42 2a 50 1e 02 03 fe 55 05 d7 77 e3 c1 75 58 b1 76 cb be 84 5f 82 3a 39 70 41 a4 b2 f8 1c ad 40 8a fa 44 ed d5 ff e7 63 93 28 0c ... >
[client received]   16379
[http received]     <Buffer 88 8b 41 1d dd 76 42 f5 a9 13 32 69 fd 12 2e 73 b8>
[http close]        1000 test done
[client close]      1000 test done

The final buffer beginning with 88 is a WebSocket closing frame, all previous buffers are part of the random message.

If I switch the port to 4008, I get:

[tcp received]      <Buffer 47 45 54 20 2f 20 48 54 54 50 2f 31 2e 31 0d 0a 48 6f 73 74 3a 20 6c 6f 63 61 6c 68 6f 73 74 3a 34 30 30 38 0d 0a 55 70 67 72 61 64 65 3a 20 77 65 62 ... >
[client open]       4008
[tcp received]      <Buffer 82 fe 3f fb 1a a4 d1 cb f5 ab bb d4 df 4e fd b9 c4 59 40 08 37 07 70 85 61 90 b7 aa 21 2f 2b d3 65 66 6b 68 26 fe 4a 48 60 d5 de d6 04 82 77 da cd dd ... >
[client received]   16379
[tcp received]      <Buffer 88 8b 0a df 58 93 09 37 2c f6 79 ab 78 f7 65 b1 3d>
[tcp close]         1000 test done
[client close]      1000 test done

So both servers work as expected. However, if I go back to port 4007 and increase the message size to 16,380:

[client open]       4007
[http received]     <Buffer 82 fe 3f fc 79 58 3a 8a 62 a0 39 9d 12 4e dc 57 70 d6 3d e6 c1 6e 3e 7a 4a df b6 6f ac 0b 66 fc be 20 14 38 52 0a 3f de a7 e9 1b 9b 8e 7c 96 6d 23 40 ... >
[client received]   16380

In this case, the HTTP receives the whole data message and echoes it back to the client, but it never receives the closing frame beginning 88 from the client. Given the client is the same in all experiments, and the TCP server works fine with larger messages, this seems to be an issue with the socket yielded by the HTTP module.

It doesn't seem to be a case that the stream simply stops emitting data after 16,379 bytes because you can send it much larger payloads, and it will receive them in multiple chunks and echo them:

[client open]       4007
[http received]     <Buffer 82 fe ff ff a2 37 a2 9b 91 06 1a 4a 91 fc ac 14 b4 c3 63 99 86 21 0f a8 b2 77 87 95 0c 48 d8 36 aa 16 fb 03 e1 b3 c7 92 8b 87 53 59 00 7f 96 47 d3 76 ... >
[http received]     <Buffer a7 eb df a8 c7 a7 78>
[client received]   65535

I'm sorry I can't make the test case any more simple, I'm not sure what to investigate next.

This problem exists in versions 4.0.0 and 4.1.2, but not in any earlier releases as far as I can tell.

[jcoglan]

On further investigation it turns out the pause() and resume() methods in my IO stream are being called one after another. Here's a transcript showing those calls:

[client open]       4007
IO pause
IO resume
[http received]     <Buffer 82 fe 3f fc 2b 2e d7 ff 7e b7 b6 bc 2f 98 e5 8a ff 93 f1 5c 74 40 cb 2a 6c 45 02 e4 2e ad 89 90 ec 86 37 a9 f9 9e ad 1f 59 9a b7 ae db 22 b8 73 e4 19 ... >
IO pause
IO resume
[client received]   16380
[http received]     <Buffer 88 8b 1d 7c bb 59 1e 94 cf 3c 6e 08 9b 3d 72 12 de>
[http close]        1000 test done
[client close]      1000 test done

This is from running the script on Node v0.12. Running on v4.0 gives the same result except it stops after the [client received] line. pause() and resume() are still being called in the same way.

I wonder if there's something about their implementation that means they work on all versions from 0.6 up to 0.12 and iojs-3.3.1 but not on 4.0.

[jcoglan]

(Although the difference between the HTTP and TCP servers on the same Node version with the same client and the same websocket-driver module suggests the issue is with HTTP, but I might be wrong.)

[jcoglan]

Some more info: applying this patch makes it work:

diff --git a/lib/websocket/streams.js b/lib/websocket/streams.js
index 96ab31f..298e858 100644
--- a/lib/websocket/streams.js
+++ b/lib/websocket/streams.js
@@ -76,9 +76,10 @@ IO.prototype.resume = function() {
 // When we receive input from a socket, send it to the parser and tell the
 // source whether to back off.
 IO.prototype.write = function(chunk) {
+  var result = !this._paused;
   if (!this.writable) return false;
   this._driver.parse(chunk);
-  return !this._paused;
+  return result;
 };

 // The IO end() method will be called when the socket piping into it emits
@@ -125,10 +126,11 @@ Messages.prototype.resume = function() {
 // When we receive messages from the user, send them to the formatter and tell
 // the source whether to back off.
 Messages.prototype.write = function(message) {
+  var result = !this._paused;
   if (!this.writable) return false;
   if (typeof message === 'string') this._driver.text(message);
   else this._driver.binary(message);
-  return !this._paused;
+  return result;
 };

 // The Messages end() method will be called when a stream piping into it emits

It turns out that when IO.write() calls driver.parse(), that can cause driver.io to emit data in response to what's come in, and that can make the downstream stream call IO.pause(). So sometimes the stream becomes paused during a write() call. Making write() return the pause status that existed at the beginning of the call makes this example work but I'm not sure if it's the correct thing to do.

This is surprising because every time write() returns false, IO.resume() gets called some time later and the stream thus emits drain, but it never receives any more data.

[calvinmetcalf]

Cc @nodejs/streams

[mscdex]

A couple of things:

1. Can you provide a reproducible example that does not involve third party modules?
2. Can you try against node master? I just landed a change in streams that fixed a similarly-sounding issue.

[jcoglan]

@mscdex How do I try it out with master? I usually install via nvm but I've not tried running from the git repo before.

[jcoglan]

@mscdex I've managed to reproduce the issue without using any external modules: https://gist.github.com/jcoglan/a1b1be5b82b2122a1df5

This script implements a simple protocol where messages consist of a 4-byte length header followed by the message content. The HTTP server accepts an Upgrade request and then echoes this protocol back to the client. The client opens a socket and writes an Upgrade request, then sends a sequence of random fixed-size messages to the server.

On my machine, setting messageSize = 2^13 makes the client loop indefinitely. At 2^14 it only sends and receives two messages, and at 2^16 and above it receives one before the server stops receiving data.

[jcoglan]

@mscdex The fix for this script is the same as in my websocket example: changing Encode.write() so that it returns the paused state from the start of the call, i.e.:

Encode.prototype.write = function(chunk) {
  if (!this.writable) return false;

  var paused = this._paused;

  var lengthHeader = new Buffer(4);
  lengthHeader.writeUInt32BE(chunk.length, 0);
  this.emit('data', lengthHeader);
  this.emit('data', chunk);

  return !paused;
};

makes the script loop at any message size.

[mscdex]

@jcoglan Here's generally what I usually do to test master (without git):

$ curl -L github.com/nodejs/node/tarball/master | tar zx
$ cd `ls | grep node`
$ ./configure && make -j8
$ ./node my-test-case.js

[jcoglan]

I've now added a TCP server that shares the protocol streams from the HTTP example for comparison. The TCP example loops at any message size while the HTTP one gets stuck.

[mscdex]

I'm also curious, why are you implementing streams from scratch instead of inheriting from stream.Writable, etc.?

[jcoglan]

@mscdex partly for historical reasons -- the Faye project that this is extracted from dates back to Node 0.1. I don't know what usage of pre-0.10 versions is like these days among my users but I've not had a strong reason to move to new streams so far.

Also, WebSocket can send both UTF-8 text and binary messages over the same connection. websocket-driver emits these as strings and buffers respectively and determines whether to send a text or binary message based on whether write() was passed a string or buffer. I'm not sure if such usage is supported in new streams; Readable.push() converts everything to a buffer before it reaches ondata and read() and Writable.write() does likewise before _write() and only the write behaviour can be disabled with decodeStrings=false.

Also, a WebSocket stream is a sequence of discrete messages. The Readable.read() method destroys the boundaries between chunks that you push() so it seems unsuitable for modelling such data.

It might be I've not sufficiently read or understood the docs so if I've got anything wrong I'd like to know :)

[mscdex]

@jcoglan FWIW streams2 streams can also be object streams (meaning you could write objects, Buffers, strings, whatever you want), not just binary data streams.

I would guess/hope that most people are at least on node v0.10 by now. Even if there are still users on v0.8 (or older), readable-stream is on npm which provides the same interfaces that currently exist in node core.

[jcoglan]

There are other issues with Readable, e.g. it does not emit a data event if you push(''), meaning empty WebSocket messages (which are entirely valid) would not be received.