Reconnecting a stream on retry error

[tikidave]

Hello, looking for best practices on implementing a reconnect after a stream is disconnected, and running into a snag w/ the provided documentation.

Referenced documentation:

https://github.com/sindresorhus/got/blob/main/documentation/3-streams.md#createretrystream

Referenced code:

import fs from 'node:fs';
import got from 'got';

let writeStream;

const fn = retryStream => {
	const options = {
		headers: {
			foo: 'bar'
		},
	};

	const stream = retryStream ?? got.stream('https://example.com', options);

	if (writeStream) {
		writeStream.destroy();
	}

	writeStream = fs.createWriteStream('example-com.html');

	stream.pipe(writeStream);

	// If you don't attach the listener, it will NOT make a retry.
	// It automatically checks the listener count so it knows whether to retry or not :)
	stream.once('retry', (retryCount, error, createRetryStream) => {
		fn(createRetryStream()); // or: fn(createRetryStream(optionsToMerge))
	});
};

fn();

I am implementing this for Twitter's search stream, which has a tendency to disconnect at times.

The problem comes when the retry is triggered - createRetryStream is undefined and causes a type error when it attempts to execute it.

My code:

function openStream(retryStream) {
  const stream =
    retryStream ??
    got.stream(
      `${apiUrl}tweets/search/stream?expansions=author_id,referenced_tweets.id.author_id&user.fields=id,name,username,profile_image_url`,
      {
        headers,
        responseType: 'json',
      }
    )
  stream.on('response', () => {
    log.info('twitter stream open')
  })
  stream.on('data', async (buffer) => {
    try {
      const data = JSON.parse(buffer)
      notify.twitter(data)
    } catch (error) {
      // Heartbeats
    }
  })
  stream.once('retry', (retryCount, error, createRetryStream) => { 
  // createRetryStream is undefined
    log.error({ error, retryCount }, 'twitter stream retry')
    openStream(createRetryStream()) // this throws type error 
  })
  stream.on('error', (error) => {
    log.error({ error }, 'twitter stream error')
  })
}

openStream()

For further context, I tested this by disconnecting my wifi to simulate an outage and it throws an ECONNRESET error, which does trigger the retry hook.

I am not sure if this is working as designed or a bug (in my code or in got). Appreciate your attention and any help here, thanks in advance.

[sindresorhus]

createRetryStream is only availble inside the retry event handler - if its undefined, the retry event isnt firing. theres a few common reasons:

1. you need to attach the retry listener BEFORE the stream starts

function openStream() {
  const stream = got.stream('https://stream.example.com', {
    retry: {
      limit: 5,
      methods: ['GET'],
      statusCodes: [408, 413, 429, 500, 502, 503, 504],
      errorCodes: ['ECONNRESET', 'ETIMEDOUT', 'ENOTFOUND']
    }
  });

  // Attach retry handler immediately
  stream.once('retry', (retryCount, error, createRetryStream) => {
    console.log(`Retrying (attempt ${retryCount}):`, error.code);
    
    // Clean up old stream first
    stream.destroy();
    
    // Create and handle new stream
    openStream();
    
    // OR use createRetryStream() if you want to pass it somewhere:
    // handleStream(createRetryStream());
  });

  stream.on('data', (chunk) => {
    // Handle data
  });

  stream.on('error', (error) => {
    console.error('Stream error:', error);
  });

  return stream;
}

openStream();

2. only ONE retry listener allowed** - if you try to attach multiple, got throws an error

3. update to latest got version** - older versions had bugs where retry events didnt fire properly

for long-lived streams like Twitter's streaming API, you might want to recreate the entire stream instead of relying on createRetryStream:

function connectToTwitterStream() {
  const stream = got.stream('https://api.twitter.com/2/tweets/search/stream', {
    headers: {Authorization: `Bearer ${token}`},
    retry: {limit: 3}
  });

  stream.once('retry', (retryCount, error) => {
    console.log(`Connection lost, reconnecting (${retryCount})...`);
    setTimeout(() => connectToTwitterStream(), 5000);
  });

  stream.on('data', processData);
  stream.on('error', (error) => {
    console.error('Fatal error:', error);
  });
}

the key is attaching the retry listener synchronously when creating the stream.