`flush` does not guarantee that all inflight messages are sent

[yo1dog]

flush does not guarantee that all inflight messages are sent before calling the given callback. Instead, flush simply sends a batch of queued messages and waits for only that batch's response before callback.

This is contrary to common expectations that a "flush" function completely empties the buffer to the destination. In fact, there are several open issues caused by this assumption.

Further, this means there is no way to know when both all queued and all inflight messages are full sent (fully flushed). For example, if you were to call flush while another flush was already in progress, the callback could be called before the previous flush was complete.

This can be a very common occurrence in short-lived environments like cloud functions.

Example:

const analytics = new Analytics(writeKey, {flushAt: 2});
analytics.track(...); // This will trigger a flush because line 212: flushed === false
analytics.track(...);
analytics.track(...); // This will trigger a flush because line 218: queue.length >= flushAt
analytics.flush().then(() => {
  // This callback will be called immediately because line 249: queue.length === 0
  // The process would exit before any messages are sent.
  console.log('done');
  process.exit(0);
});

Similarly:

const analytics = new Analytics(writeKey, {flushAt: 2});
analytics.track(...); // This will trigger a flush because line 212: flushed === false
analytics.track(...);
// The following will trigger a flush because there is a message in the queue.
analytics.flush().then(() => {
  // This calllback may be called before the other flush completes.
  // The process may exit while the message from the other flush is still inflight.
  console.log('done');
  process.exit(0);
});
// There are now 2 concurrent flushes running in parallel.
// Waiting on the final flush creates a race condition as it may finish before the previous flush.

I suggest adding a new function that flush()es all queued messages and does not callback until all concurrent flushes are complete. Also updating the documentation to make it clear that flush may defy assumptions about awaiting all queued and inflight messages.

[felipe-najson-ntf]

Hello @yo1dog, thanks for your contribution to the library! I just tested the cases mentioned above and clearly flush creates a race condition in which 2 simultaneous downloads are executed in parallel.

I'll be working on a solution to solve this next week.

Thanks, Segment team.

[benjaminhoffman]

I'm also experiencing issues with queueing and flush. Would love to see this fixed to help eliminate issues in my debugging.

edit: we are using serverless functions.

[yo1dog]

@benjaminhoffman My quick workaround for Lambda functions was to manually track the callbacks for every Segment call. I used a small class to help. This works OK for cloud function executions as they have a predictably short and linear lifecycle. This solution would not work in a long-lived environment as the promises array would grow infinitely. Example:

class SegmentBatch {
  constructor() {
    this.promises = [];
  }
  
  createCB() {
    let resolve;
    this.promises.push(new Promise(_resolve => {resolve = _resolve;}));
    return err => resolve(err);
  }
  
  async finish() {
    const results = await Promise.all(this.promises);
    const err = results.find(err => err);
    if (err) throw err;
  }
}

All the class does is maintain a list of promises which are resolved when the callback is called. Then we can await all the promises to ensure they all callbacks have been called and thus all events have been fullsent. I use it like so:

const Analytics = require('analytics-node');

const client = new Analytics(writeKey);
client.flushed = true; // Prevent flushing after first event.

const batch = new SegmentBatch();
client.track({...}, batch.createCB());
client.track({...}, batch.createCB());
...
client.track({...}, batch.createCB());

client.flush();
await batch.finish();

You could modify this for use in long-lived environments by adding logic that removes promises from the array once they have been resolved.

[benjaminhoffman]

That worked great for me! Thank you 🙏 . I rewrote this in typescript and it's worth noting the flushed prop is not part of the type so you'll have to `@ts-ignore it

@yo1dog should I be concerned about latency or any delay from the Segment API? Even at 200ms, I hate to add unnecessary delay. Or what if the Segment API times out?

[yo1dog]

Yes. The only way to guarantee events are fully sent to Segment is to wait for the underlying HTTP request to complete. Network unreliability can cause this to take a significant amount of time: 100s to 10,000s of milliseconds depending on your setup and timeout settings.

Some cloud functions provide execution environments independent of individual executions (outside of the function's handler). You could move the management and flushing of events to this space so long as you can guarantee the flush can be awaited before the environment is stopped/shutdown. This is not possible with AWS Lambda, for example, as the execution environment is destroyed without warning. In this case, the only alternative to blocking invocation ("add unnecessary delay") would be to not emit events to Segment directly but instead to send the events to an external queue to be processed later. Presumably that queue could be "nearby" (peer process, or in same data center, etc.) resulting in much lower latency would and much higher reliability.

[yo1dog]

@benjaminhoffman I just realized you stated you are "not using serverless functions." The solution I posted above was geared specifically for short-lived environments such as a cloud function invocation.

In a traditional long-lived server process, rather than flushing events and blocking, you should manage the events outside of request handlers (as in, you should send events to Segment API "in the background"). This is what the Segment library aims to provide (I assume. I am not affiliated with Segment or this library in any way): You can simply call client.track(...) and know that events will be queued and eventually sent to Segment. The problem is when that process experiences an interruption (for example, a server restart) queued events can be lost because the Segment library does not provide any way to know when all queued messages have been fully sent to Segment. Again, this assumes you have the ability to know of and delay a process shutdown request.

My solution here was to use a wrapper around the Segment library which was similar to the batch class above. Something like this (untested):

class ManagedSegmentClient {
  private readonly client: Analytics;
  private readonly promises: Promise<Error | undefined>[];
  
  public constructor(client: Analytics) {
    this.client = client;
    this.promises = [];
  }
  
  public track(...args: Parameters<Analytics['track']>) {
    this.client.track(args[0], this.createCB(args[1]));
  }
  public identify(...args: Parameters<Analytics['identify']>) {
    this.client.identify(args[0], this.createCB(args[1]));
  }
  
  public async finish() {
    const results = await Promise.all(this.promises);
    const err = results.find(err => err);
    if (err) throw err;
  }
  
  private createCB(cb?: (err: Error) => void) {
    let resolve: (err?: Error) => void;
    const promise = new Promise<Error | undefined>(_resolve => {resolve = _resolve;});
    this.promises.push(promise);
    promise.finally(() => {
      const index = this.promises.indexOf(promise);
      if (index > -1) this.promises.splice(index, 1);
    });
    
    return (err?: Error) => {
      resolve(err);
      if (cb) cb(err);
    };
  }
}

And use it like so:

const managedSegmentClient = new ManagedSegmentClient(new Analytics(writeKey));

expressApp.get('/fubar', (req, res) => {
  // ...
  const utmSource = req.query.utm_source;
  managedSegmentClient.track({event: 'blah', properties: {utmSource}});
  return res.status(200).json({ok: true});
});

process.once('SIGINT', () => exitGracefully());
process.once('SIGTERM', () => exitGracefully());

function exitGracefully() {
  // ...
  
  // Flush Segment messages before exiting.
  managedSegmentClient.finish(err => {
    if (err) console.error('Error flushing Segment messages.');
    else console.log('Segment messages flushed.');
    
    // Now exit.
    process.exit(exitCode);
  });
}

Whenever a track or identify event is called, it adds a promise to an array which is resolved when the callback is called (which is after the message has been fully sent). Whenever the promise is resolved, it is removed from the array. Then the finish method awaits all promises in the array to ensure all messages are sent.

[benjaminhoffman]

super helpful @yo1dog and i think others reading this will also gain from your sample code... turns our we are using serverless functions (specifically, nextJS API routes deployed on Vercel) and your first solution worked great. I spoke with our CTO and he said the latency concerns at this point would be a pre-mature optimization and to keep an eye on things. If we see trouble from the SegmentBatch solution, then I'm going to build my own queue, as you've suggested above. Thanks so much for your help!