I recently encountered an issue in a Node.js v12 service where output streamed from a child process is intermittently missing from downstream processing. The pipeline streams the data into a ZIP archive and uploads it to S3. The service is deployed on AWS Lambda.
In this post, I describe the problem, walk through the debugging process, and evaluate several solutions.
Data pipeline Link to heading
At a high level, the data pipeline inside the Lambda function looks like:
.-------------------. .---------------. .----.
----------->| custom executable |----------->| node-archiver |----------->| S3 |
event '-------------------' stdout '---------------' pipe '----'
The custom executable accepts parameters from the Lambda event and outputs data on its stdout stream. The data is compressed into a ZIP file using node-archiver and piped to a stream that uploads to S3.
The project has integration tests that run the pipeline with different parameters and verify the contents of the generated ZIP files. When the problem occurs, the ZIP files are valid archives, but the single compressed file inside is either empty or is missing the first chunk of 65536 bytes.
I confirmed this data loss in a Python REPL by comparing the data from successful and failed runs:
>>> with open('good.data', 'rb') as gf, open('bad.data', 'rb') as bf:
... good = gf.read()
... bad = bf.read()
...
>>> len(bad)
63532
>>> len(good)
129068
>>> bad in good
True
>>> good.find(bad)
65536
In this example, the bad data is missing the first 65536
bytes, but is otherwise correct.
Source code Link to heading
This section includes distilled code samples to show how the data pipeline is constructed. The actual project code is structured very differently and includes configuration and error handling.
Create a stream to upload to S3 Link to heading
The first step initializes an upload to an S3 bucket. The output is a writeable stream and a Promise that resolves when the upload completes:
import AWS from ‘aws-sdk’;
import stream from ‘stream’;
const s3 = new AWS.S3();
const bucketName = 'my-bucket';
const key = 'my-key';
const uploadStream = new stream.PassThrough();
const params = {
Bucket: bucketName,
Key: key,
Body: uploadStream,
};
const uploadPromise = s3.upload(params).promise();
Run the custom executable Link to heading
Next, we run the custom executable and get its standard output stream:
import { spawn } from 'child_process';
const args = [...];
const myProcess = spawn('./myExecutable', args);
const myProcessStdout = myProcess.stdout;
Create the ZIP file Link to heading
The last step adds the custom executable’s output to a ZIP file and streams the file to S3.
const archive = archiver('zip');
// Streams from previous steps
const inputStream = myProcessStdout;
const outputStream = uploadStream;
const filename = 'my.data';
// Promise that resolves when archiver is finished writing to the output
// stream
const archivePromise = new Promise((resolve, reject) => {
outputStream.on('finish', () => {
resolve();
});
outputStream.on('error', (err) => {
reject(err);
});
archive.on('error', (err) => {
reject(err);
});
});
archive.pipe(outputStream);
archive.append(inputStream, { name: filename });
archive.finalize();
Wait for asynchronous operations Link to heading
Once the pipeline is established, we wait for the promises to resolve:
await Promise.all([archivePromise, uploadPromise]);
Debugging Link to heading
When the pipeline fails, a valid ZIP file is uploaded to S3, but the data inside
the archive is incomplete. No failures were observed when skipping the ZIP step,
i.e. when directly piping the child_process stdout stream to the S3 upload
stream. This suggests that the problem is related to node-archiver.
node-archiver Link to heading
To figure out what’s happening, I read node-archiver’s source code.
Asynchronous queue Link to heading
When adding an entry to a ZIP file, node-archiver’s append() function doesn’t
immediately perform the operation. Instead, it adds a task to an async
queue. On a subsequent tick
of the event loop, the task pipes the input stream (the child_process stdout
stream) into the output stream (the S3 upload stream).
Stream flowing mode Link to heading
Running in the debugger revealed that the input stream may already be in
“flowing mode” by the time node-archiver calls pipe(); the stream’s
readableFlowing
property is true. When in flowing mode, a stream’s data is available through the
EventEmitter interface.
The Node.js stream documentation describes flowing mode in detail:
All Readable streams begin in paused mode but can be switched to flowing mode in one of the following ways:
- Adding a ‘data’ event handler.
- Calling the stream.resume() method.
- Calling the stream.pipe() method to send the data to a Writable.
In this project, there are no ‘data’ event handlers, and the stream is already
flowing by the time stream.pipe() is called. Therefore, something must have
called stream.resume().
Race condition Link to heading
In the debugger, I found that if the child process exits before its output
stream is piped, Node.js flushes the stream by calling
stream.resume().
This explains how the stream switches to flowing mode.
Therefore, there’s a race condition where the child process might exit and flush
its output before node-archiver calls pipe(). In that case, some or all of
the data from the child process could be discarded before being added to the ZIP
file.
Solutions Link to heading
To avoid the race condition that may result in lost data, pipe() must be
called on the child_process stdout stream in the same tick of the event loop
in which the stream was created. This section describes several ways to
accomplish this.
Solution 1: zip-stream Link to heading
Node-archiver builds on several low-level libraries, including
zip-stream. Using zip-stream
avoids the async queue. pipe() is called immediately when adding a stream to
the archive, and no data from the child process is lost.
Pros:
- Drop-in solution with minor code changes
Cons:
- The documentation suggests that the module “is meant to be wrapped internally by other modules”
- Less flexible than node-archiver in case requirements change
Solution 2: Wait for a patched node-archiver release Link to heading
Node-archiver uses another low-level library, called archiver-utils. A patch to this library fixes the underlying problem. From https://github.com/archiverjs/archiver-utils/pull/17:
normalizeInputSource: always pipe stream through a PassThrough stream
This is needed to guarantee pausing the stream if it’s already flowing, since it will only be processed in a (distant) future iteration of the event loop, and will lose data if already flowing when added to the queue.
The latest version of node-archive as of this writing–4.0.1–doesn’t include the patched archiver-utils; see https://github.com/archiverjs/node-archiver/issues/364.
Pros:
- Drop-in solution with no code changes
Cons:
- A patched node-archiver has yet to be released
Solution 3: Local workaround Link to heading
Until a fixed version of node-archiver is released, we can fix the race condition by adding a workaround directly in our project.
The archiver-utils patch described above fixes the data loss by piping the input
stream through a PassThrough stream. The PassThrough stream effectively
pauses the data flow until it’s piped to another stream.
The updated ZIP file creation code looks like:
const passThrough = new PassThrough();
inputStream = inputStream.pipe(passThrough);
archive.append(inputStream, { name: filename });
archive.finalize();
Because we spawn the child process and pipe its stdout stream in the same tick of the event loop, no data is lost, even if the child process exits quickly.
Conclusion Link to heading
This issue highlights the importance of having multiple layers of automated testing and the value of open source software and the open source community.
Intermittent integration test failures uncovered this race condition and provided artifacts that aided debugging. Having access to the source code of Node.js and node-archiver helped explain the root cause of the issue. Discussions on GitHub led to a solution.