Stall detection is entirely non-functional

[tremby]

I've had a few cases now where the "keepalive" messages (which I log when I find a data keep-alive event) stop coming, and tweets I expect to receive are not received. There aren't other tweets or events coming in during these periods (I log something for every event I'm aware of). The process is definitely still running.

I've never seen any of the connection error * events happen -- I wonder if there's a bug somewhere.

A snippet from my code:

ts.on('connection aborted', function tsConnectionAborted() {
        console.error("Twitter stream: connection aborted");
});
ts.on('connection error network', function tsConnectionErrorNetwork(error) {
        console.error("Twitter stream: connection network error:", error);
});
ts.on('connection error stall', function tsConnectionErrorStall() {
        console.error("Twitter stream: connection stalled");
});
ts.on('connection error http', function tsConnectionErrorHttp(httpStatusCode) {
        console.error("Twitter stream: connection HTTP error:", httpStatusCode);
});
ts.on('connection error unknown', function tsConnectionErrorUnknown(error) {
        console.error("Twitter stream: connection error:", error);
});
ts.on('data keep-alive', function tsKeepAlive() {
        console.log("Twitter stream: keepalive");
});
ts.on('data error', function tsDataError(error) {
        console.error("Twitter stream: data error:", error);
});

Out of these, the only one I've ever seen is the "keepalive", and they come pretty often in the logs. The most recent time it was missing was for a period of about an hour, after which at least two matching tweets were missed, and then I restarted the process.

What can I do to help debug?

[tremby]

I started digging in the codebase since this happened again today.

I notice a chunk of code which has been commented out. Should it be...?

twitter-stream-api/lib/connection.js

Lines 93 to 99 in ff86689

	// Connection has stalled if data has not received after 90 seconds
	/*
	self.request.setTimeout(90000, function () {
	self.emit('connection error stall');
	self.destroy();
	});
	*/

[tremby]

It was commented out in this commit 064741b

[tremby]

Uncommenting it gives an error that self.request.setTimeout is not a function.

[tremby]

So the whole 90-second stall detection described at https://github.com/trygve-lie/twitter-stream-api#connection-error-stall is in fact non-functional.

[koenrad]

I was planning to use this in my project, however after looking at the code I believe you are correct. I have not tried anything yet, but I think maybe if you add timeout: 90000 to the conf object in connection.js and pass a callback function to to the request() call and catch timeout errors and emit the stall event it may work. I will post if I can get it working.

[AyazAI]

@koenrad Have you make it working...

[tremby]

For what it's worth, I checked the commit log for the project I was working on back then and see that on July 16 of last year I switched the library being used from twitter-stream-api to https://github.com/mileszim/twat and I don't remember having any problems since.

It looks like that package is no longer on NPM. I haven't dug into it.

Here's most of that commit, in case it's helpful.

commit eb1f1d366882a7169ee6504e19dde4adc39a3e23
Author: tremby
Date:   Tue Jul 16 15:58:07 2019 -0700

    Switch stream library from 'twitter-stream-api' to 'twat'

diff --git a/listen.js b/listen.js
index a524a0c..3cb625a 100644
--- a/listen.js
+++ b/listen.js
@@ -7,7 +7,7 @@ const {
 } = require('./handle-tweet');
 const redis = require('./redis');
 
-const TwitterStream = require('twitter-stream-api');
+const Twat = require('twat');
 
 const retweetHandles = process.env.RETWEET_HANDLES ? process.env.RETWEET_HANDLES.split(',') : [];
 const retweetHashtag = process.env.RETWEET_HASHTAG || null;
@@ -17,11 +17,11 @@ const QUEUE_LOCK = 'replyqueue:processing';
 const QUEUE_LOCK_TIMEOUT = parseInt(process.env.QUEUE_LOCK_TIMEOUT, 10) * 1e3;
 
 // Set up the streaming API
-const ts = new TwitterStream({
+const twat = new Twat({
 	consumer_key: process.env.TWITTER_CONSUMER_KEY,
 	consumer_secret: process.env.TWITTER_CONSUMER_SECRET,
-	token: process.env.TWITTER_ACCESS_TOKEN_KEY,
-	token_secret: process.env.TWITTER_ACCESS_TOKEN_SECRET,
+	access_token: process.env.TWITTER_ACCESS_TOKEN_KEY,
+	access_token_secret: process.env.TWITTER_ACCESS_TOKEN_SECRET,
 });
 
 // Get a promise for our bot's user ID
@@ -29,7 +29,7 @@ const botUserIdPromise = getUserIdFromHandle(process.env.TWITTER_HANDLE);
 
 // Prepare streaming configuration
 const filterConfig = {
-	track: [`@${process.env.TWITTER_HANDLE}`],
+	track: `@${process.env.TWITTER_HANDLE}`,
 	stall_warnings: true,
 };
 
@@ -44,47 +44,40 @@ retweetHandleIdsPromise.then((retweetHandleIds) => {
 	if (retweetHandles.length) {
 		filterConfig.follow = retweetHandleIds.join(',');
 	}
-	ts.stream('statuses/filter', filterConfig);
+	twat.stream('statuses/filter', filterConfig, handleStream);
 });
-ts.on('connection success', function tsConnectionSuccess() {
+
+function handleStream(stream) {
 	const activities = [
-		`tracking ${filterConfig.track.map(term => `'${term}'`).join(', ')}`,
+		`tracking ${filterConfig.track}`,
 	];
 	if (filterConfig.follow) {
 		activities.push(`following users with IDs ${filterConfig.follow}`);
 	}
 	console.log(`Twitter stream: connected; ${activities.join('; ')}`);
-});
-ts.on('connection aborted', function tsConnectionAborted() {
-	console.error("Twitter stream: connection aborted");
-});
-ts.on('connection error network', function tsConnectionErrorNetwork(error) {
-	console.error("Twitter stream: connection network error:", error);
-});
-ts.on('connection error stall', function tsConnectionErrorStall() {
-	console.error("Twitter stream: connection stalled");
-});
-ts.on('connection error http', function tsConnectionErrorHttp(httpStatusCode) {
-	console.error("Twitter stream: connection HTTP error:", httpStatusCode);
-});
-ts.on('connection error unknown', function tsConnectionErrorUnknown(error) {
-	console.error("Twitter stream: connection error:", error);
-});
-ts.on('data keep-alive', function tsKeepAlive() {
-	console.log("Twitter stream: keepalive");
-});
-ts.on('data error', function tsDataError(error) {
-	console.error("Twitter stream: data error:", error);
-});
 
-// Handle incoming tweets
-ts.on('data', function (tweet) {
-	handleIncomingTweet(tweet, {
-		retweetHandles,
-		retweetHashtag,
-		botUserIdPromise,
-	}, queueResponse);
-});
+	stream.on('error', function streamError(type, info) {
+		console.error("Twitter stream: error:", type, info);
+	});
+	stream.on('reconnect', function streamReconnect(info) {
+		console.error("Twitter stream: reconnect:", info.error, info.attempts);
+	});
+	stream.on('end', function streamEnd(response) {
+		console.error("Twitter stream: end", response);
+	});
+	stream.on('destroy', function streamDestroy(message) {
+		console.error("Twitter stream: destroy", message);
+	});
+
+	// Handle incoming tweets
+	stream.on('tweet', function (tweet) {
+		handleIncomingTweet(tweet, {
+			retweetHandles,
+			retweetHashtag,
+			botUserIdPromise,
+		}, queueResponse);
+	});
+}
 
 /**
  * Process all items in the queue in sequence
diff --git a/package-lock.json b/package-lock.json
index 32fb8e5..2c199d9 100644
--- a/package-lock.json
+++ b/package-lock.json
[snip]
diff --git a/package.json b/package.json
index ac7f13a..4e5be27 100644
--- a/package.json
+++ b/package.json
@@ -20,7 +20,7 @@
     "redis-lock": "^0.1.4",
     "seedrandom": "^3.0.1",
     "twitter": "^1.7.1",
-    "twitter-stream-api": "^0.5.2"
+    "twat": "github:mileszim/twat#v0.8.5"
   },
   "devDependencies": {
     "argparse": "^1.0.10"