From 21339ea5951c9f8f2b3782e82c8d89fa458470b9 Mon Sep 17 00:00:00 2001 From: Patrik Oldsberg Date: Thu, 22 Sep 2022 14:51:07 +0200 Subject: [PATCH] backend-common: work around premature close errors Signed-off-by: Patrik Oldsberg --- .changeset/swift-suits-reply.md | 5 ++ .../src/reading/GerritUrlReader.ts | 8 +-- .../src/reading/tree/ReadableArrayResponse.ts | 10 ++- .../src/reading/tree/TarArchiveResponse.ts | 15 ++--- .../src/reading/tree/util.test.ts | 63 +++++++++++++++++++ .../backend-common/src/reading/tree/util.ts | 35 ++++++++--- 6 files changed, 109 insertions(+), 27 deletions(-) create mode 100644 .changeset/swift-suits-reply.md create mode 100644 packages/backend-common/src/reading/tree/util.test.ts diff --git a/.changeset/swift-suits-reply.md b/.changeset/swift-suits-reply.md new file mode 100644 index 0000000000..f81f39bee2 --- /dev/null +++ b/.changeset/swift-suits-reply.md @@ -0,0 +1,5 @@ +--- +'@backstage/backend-common': patch +--- + +Refactored internal usage of the build-in `pipeline` from `'stream'` to work around `tar` bug in Node 18. diff --git a/packages/backend-common/src/reading/GerritUrlReader.ts b/packages/backend-common/src/reading/GerritUrlReader.ts index 59b33bb2cb..9ec5e35f65 100644 --- a/packages/backend-common/src/reading/GerritUrlReader.ts +++ b/packages/backend-common/src/reading/GerritUrlReader.ts @@ -32,8 +32,7 @@ import fetch, { Response } from 'node-fetch'; import os from 'os'; import { join as joinPath } from 'path'; import tar from 'tar'; -import { pipeline as pipelineCb, Readable } from 'stream'; -import { promisify } from 'util'; +import { Readable } from 'stream'; import { ReaderFactory, ReadTreeOptions, @@ -45,8 +44,7 @@ import { UrlReader, } from './types'; import { ScmIntegrations } from '@backstage/integration'; - -const pipeline = promisify(pipelineCb); +import { pipeStream } from './tree/util'; const createTemporaryDirectory = async (workDir: string): Promise => await fs.mkdtemp(joinPath(workDir, '/gerrit-clone-')); @@ -197,7 +195,7 @@ export class GerritUrlReader implements UrlReader { }); const data = await new Promise(async resolve => { - await pipeline( + await pipeStream( tar.create({ cwd: tempDir }, ['']), concatStream(resolve), ); diff --git a/packages/backend-common/src/reading/tree/ReadableArrayResponse.ts b/packages/backend-common/src/reading/tree/ReadableArrayResponse.ts index eabaa3bc56..da145b0f4b 100644 --- a/packages/backend-common/src/reading/tree/ReadableArrayResponse.ts +++ b/packages/backend-common/src/reading/tree/ReadableArrayResponse.ts @@ -19,17 +19,15 @@ import platformPath, { basename } from 'path'; import getRawBody from 'raw-body'; import fs from 'fs-extra'; -import { promisify } from 'util'; import tar from 'tar'; -import { pipeline as pipelineCb, Readable } from 'stream'; +import { Readable } from 'stream'; import { ReadTreeResponse, ReadTreeResponseFile, ReadTreeResponseDirOptions, FromReadableArrayOptions, } from '../types'; - -const pipeline = promisify(pipelineCb); +import { pipeStream } from './util'; /** * Wraps a array of Readable objects into a tree response reader. @@ -75,7 +73,7 @@ export class ReadableArrayResponse implements ReadTreeResponse { try { const data = await new Promise(async resolve => { - await pipeline( + await pipeStream( tar.create({ cwd: tmpDir }, ['']), concatStream(resolve), ); @@ -95,7 +93,7 @@ export class ReadableArrayResponse implements ReadTreeResponse { for (let i = 0; i < this.stream.length; i++) { if (!this.stream[i].path.endsWith('/')) { - await pipeline( + await pipeStream( this.stream[i].data, fs.createWriteStream( platformPath.join(dir, basename(this.stream[i].path)), diff --git a/packages/backend-common/src/reading/tree/TarArchiveResponse.ts b/packages/backend-common/src/reading/tree/TarArchiveResponse.ts index 493e283c5f..7d0d8f42a2 100644 --- a/packages/backend-common/src/reading/tree/TarArchiveResponse.ts +++ b/packages/backend-common/src/reading/tree/TarArchiveResponse.ts @@ -17,21 +17,18 @@ import concatStream from 'concat-stream'; import fs from 'fs-extra'; import platformPath from 'path'; -import { pipeline as pipelineCb, Readable } from 'stream'; +import { Readable } from 'stream'; import tar, { Parse, ParseStream, ReadEntry } from 'tar'; -import { promisify } from 'util'; import { ReadTreeResponse, ReadTreeResponseDirOptions, ReadTreeResponseFile, } from '../types'; -import { stripFirstDirectoryFromPath } from './util'; +import { pipeStream, stripFirstDirectoryFromPath } from './util'; // Tar types for `Parse` is not a proper constructor, but it should be const TarParseStream = Parse as unknown as { new (): ParseStream }; -const pipeline = promisify(pipelineCb); - /** * Wraps a tar archive stream into a tree response reader. */ @@ -99,7 +96,7 @@ export class TarArchiveResponse implements ReadTreeResponse { } const content = new Promise(async resolve => { - await pipeline(entry, concatStream(resolve)); + await pipeStream(entry, concatStream(resolve)); }); files.push({ @@ -110,7 +107,7 @@ export class TarArchiveResponse implements ReadTreeResponse { entry.resume(); }); - await pipeline(this.stream, parser); + await pipeStream(this.stream, parser); return files; } @@ -128,7 +125,7 @@ export class TarArchiveResponse implements ReadTreeResponse { try { const data = await new Promise(async resolve => { - await pipeline( + await pipeStream( tar.create({ cwd: tmpDir }, ['']), concatStream(resolve), ); @@ -152,7 +149,7 @@ export class TarArchiveResponse implements ReadTreeResponse { let filterError: Error | undefined = undefined; - await pipeline( + await pipeStream( this.stream, tar.extract({ strip, diff --git a/packages/backend-common/src/reading/tree/util.test.ts b/packages/backend-common/src/reading/tree/util.test.ts new file mode 100644 index 0000000000..662ff3f194 --- /dev/null +++ b/packages/backend-common/src/reading/tree/util.test.ts @@ -0,0 +1,63 @@ +/* + * Copyright 2021 The Backstage Authors + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +import { Readable, Writable } from 'stream'; +import { pipeStream, streamToBuffer } from './util'; + +describe('pipeStream', () => { + it('should pipe a stream', async () => { + const from = Readable.from(['hello']); + + let written = ''; + const to = new Writable({ + write(chunk, encoding, callback) { + written = `${encoding}:${chunk}`; + callback(); + }, + }); + + await pipeStream(from, to); + expect(written).toBe('buffer:hello'); + }); + + it('should forward errors', async () => { + const from = new Readable({ + read() { + throw new Error('oh no'); + }, + }); + const to = new Writable(); + + await expect(pipeStream(from, to)).rejects.toThrow('oh no'); + }); +}); + +describe('streamToBuffer', () => { + it('should read a stream', async () => { + await expect(streamToBuffer(Readable.from(['hello']))).resolves.toBe( + 'hello', + ); + }); + + it('should fail on errors', async () => { + const stream = new Readable({ + read() { + throw new Error('oh no'); + }, + }); + await expect(streamToBuffer(stream)).rejects.toThrow('oh no'); + }); +}); diff --git a/packages/backend-common/src/reading/tree/util.ts b/packages/backend-common/src/reading/tree/util.ts index 63192102f4..a4e9a6686c 100644 --- a/packages/backend-common/src/reading/tree/util.ts +++ b/packages/backend-common/src/reading/tree/util.ts @@ -14,12 +14,9 @@ * limitations under the License. */ -import { Readable, pipeline as pipelineCb } from 'stream'; -import { promisify } from 'util'; +import { Readable, finished } from 'stream'; import concatStream from 'concat-stream'; -const pipeline = promisify(pipelineCb); - // Matches a directory name + one `/` at the start of any string, // containing any character except `/` one or more times, and ending with a `/` // e.g. Will match `dirA/` in `dirA/dirB/file.ext` @@ -29,13 +26,37 @@ export function stripFirstDirectoryFromPath(path: string): string { return path.replace(directoryNameRegex, ''); } +// Custom pipeline implementation, since pipeline doesn't work well with tar on node 18 +// See https://github.com/npm/node-tar/issues/321 +export function pipeStream( + from: NodeJS.ReadableStream, + to: NodeJS.WritableStream, +): Promise { + return new Promise((resolve, reject) => { + from.pipe(to); + finished(from, fromErr => { + if (fromErr) { + reject(fromErr); + } else { + finished(to, toErr => { + if (toErr) { + reject(toErr); + } else { + resolve(); + } + }); + } + }); + }); +} + // Collect the stream into a buffer and return -export const streamToBuffer = (stream: Readable): Promise => { +export function streamToBuffer(stream: Readable): Promise { return new Promise(async (resolve, reject) => { try { - await pipeline(stream, concatStream(resolve)); + await pipeStream(stream, concatStream(resolve)); } catch (ex) { reject(ex); } }); -}; +}