diff --git a/lib/zlib.js b/lib/zlib.js
index c86fe467a23..c506bb16c3d 100644
--- a/lib/zlib.js
+++ b/lib/zlib.js
@@ -87,6 +87,7 @@ const {
 
 const kFlushFlag = Symbol('kFlushFlag');
 const kError = Symbol('kError');
+const kBufferResult = Symbol('kBufferResult');
 
 const constants = internalBinding('constants').zlib;
 const {
@@ -124,7 +125,7 @@ for (const ckey of ObjectKeys(codes)) {
   codes[codes[ckey]] = ckey;
 }
 
-function zlibBuffer(engine, buffer, callback) {
+function zlibBuffer(engine, buffer, callback, finishInOneWrite) {
   validateFunction(callback, 'callback');
   // Streams do not support non-Uint8Array ArrayBufferViews yet. Convert it to a
   // Buffer without copying.
@@ -136,44 +137,81 @@ function zlibBuffer(engine, buffer, callback) {
   engine.buffers = null;
   engine.nread = 0;
   engine.cb = callback;
-  engine.on('data', zlibBufferOnData);
   engine.on('error', zlibBufferOnError);
+
+  // The whole input is known up front. Unless the engine is handed to the
+  // caller (`info`) or a custom flush is requested, process it with the finish
+  // flush directly, like the synchronous methods do. This takes one threadpool
+  // round trip instead of a write followed by a separate final flush, and skips
+  // the stream machinery.
+  if (finishInOneWrite && !engine._info &&
+      engine._defaultFlushFlag === FLUSH_BOUND[engine._flushBoundIdx][0]) {
+    if (typeof buffer === 'string') {
+      // Encode strings like Writable#write() does.
+      buffer = Buffer.from(buffer, engine._writableState.defaultEncoding);
+    }
+    if (isUint8Array(buffer)) {
+      engine[kBufferResult] = true;
+      processChunk(engine, buffer, engine._finishFlushFlag,
+                   zlibBufferOnWriteDone);
+      return;
+    }
+  }
+
+  engine.on('data', zlibBufferOnData);
   engine.on('end', zlibBufferOnEnd);
   engine.end(buffer);
 }
 
-function zlibBufferOnData(chunk) {
-  if (!this.buffers) {
-    this.buffers = [chunk];
+function zlibBufferAddChunk(engine, chunk) {
+  if (!engine.buffers) {
+    engine.buffers = [chunk];
   } else {
-    this.buffers.push(chunk);
+    engine.buffers.push(chunk);
   }
-  this.nread += chunk.length;
-  if (this.nread > this._maxOutputLength) {
-    this.close();
-    this.removeAllListeners('end');
-    this.cb(new ERR_BUFFER_TOO_LARGE(this._maxOutputLength));
+  engine.nread += chunk.length;
+  if (engine.nread > engine._maxOutputLength) {
+    engine.close();
+    engine.removeAllListeners('end');
+    engine.cb(new ERR_BUFFER_TOO_LARGE(engine._maxOutputLength));
   }
 }
 
+function zlibBufferOnData(chunk) {
+  zlibBufferAddChunk(this, chunk);
+}
+
 function zlibBufferOnError(err) {
   this.removeAllListeners('end');
   this.cb(err);
 }
 
-function zlibBufferOnEnd() {
+function zlibBufferFinish(engine) {
   let buf;
-  if (this.nread === 0) {
+  if (engine.nread === 0) {
     buf = new FastBuffer();
   } else {
-    const bufs = this.buffers;
-    buf = (bufs.length === 1 ? bufs[0] : Buffer.concat(bufs, this.nread));
+    const bufs = engine.buffers;
+    buf = (bufs.length === 1 ? bufs[0] : Buffer.concat(bufs, engine.nread));
   }
-  this.close();
-  if (this._info)
-    this.cb(null, { buffer: buf, engine: this });
+  engine.close();
+  if (engine._info)
+    engine.cb(null, { buffer: buf, engine });
   else
-    this.cb(null, buf);
+    engine.cb(null, buf);
+}
+
+function zlibBufferOnEnd() {
+  zlibBufferFinish(this);
+}
+
+// Called by processCallback(), with the handle as `this`, once the whole input
+// has been processed. Errors have already been reported to the callback, either
+// through the 'error' event or by zlibBufferAddChunk().
+function zlibBufferOnWriteDone(err) {
+  const engine = this[owner_symbol];
+  if (err === undefined && !engine.destroyed)
+    zlibBufferFinish(engine);
 }
 
 function zlibBufferSync(engine, buffer) {
@@ -552,7 +590,10 @@ function processCallback() {
   if (have > 0) {
     const out = self._outBuffer.slice(self._outOffset, self._outOffset + have);
     self._outOffset += have;
-    streamBufferIsFull = !self.push(out);
+    if (self[kBufferResult] === true)
+      zlibBufferAddChunk(self, out);
+    else
+      streamBufferIsFull = !self.push(out);
   } else {
     assert(have === 0, 'have should not go down');
   }
@@ -618,7 +659,8 @@ function processCallback() {
       return;
     }
 
-    self.push(null);
+    if (self[kBufferResult] !== true)
+      self.push(null);
   }
 
   // Finished with the chunk.
@@ -791,7 +833,8 @@ class Unzip extends Zlib {
   }
 }
 
-function createConvenienceMethod(ctor, sync, prepareOpts) {
+function createConvenienceMethod(ctor, sync, prepareOpts,
+                                 finishInOneWrite = true) {
   if (sync) {
     return function syncBufferWrapper(buffer, opts) {
       return zlibBufferSync(new ctor(opts), buffer);
@@ -805,7 +848,7 @@ function createConvenienceMethod(ctor, sync, prepareOpts) {
     if (prepareOpts !== undefined) {
       opts = prepareOpts(buffer, opts);
     }
-    return zlibBuffer(new ctor(opts), buffer, callback);
+    return zlibBuffer(new ctor(opts), buffer, callback, finishInOneWrite);
   };
 }
 
@@ -1118,7 +1161,11 @@ module.exports = {
   brotliCompressSync: createConvenienceMethod(BrotliCompress, true),
   brotliDecompress: createConvenienceMethod(BrotliDecompress, false),
   brotliDecompressSync: createConvenienceMethod(BrotliDecompress, true),
-  zstdCompress: createConvenienceMethod(ZstdCompress, false, withPledgedSrcSize),
+  // A zstd frame written by a single ZSTD_e_end call can be a few bytes larger
+  // than one written by ZSTD_e_continue followed by ZSTD_e_end, so
+  // zstdCompress() keeps writing the input and ending the frame separately.
+  zstdCompress: createConvenienceMethod(ZstdCompress, false, withPledgedSrcSize,
+                                        false),
   zstdCompressSync: createConvenienceMethod(ZstdCompress, true),
   zstdDecompress: createConvenienceMethod(ZstdDecompress, false),
   zstdDecompressSync: createConvenienceMethod(ZstdDecompress, true),
diff --git a/test/parallel/test-zlib-convenience-async-round-trip.js b/test/parallel/test-zlib-convenience-async-round-trip.js
new file mode 100644
index 00000000000..ab48f885272
--- /dev/null
+++ b/test/parallel/test-zlib-convenience-async-round-trip.js
@@ -0,0 +1,74 @@
+'use strict';
+// The asynchronous convenience methods know the whole input up front, so they
+// process it with the finish flush in a single write. Check that a call whose
+// output fits in one output buffer takes exactly one threadpool round trip.
+
+const common = require('../common');
+const assert = require('assert');
+const async_hooks = require('async_hooks');
+const zlib = require('zlib');
+
+const input = Buffer.from('hello world '.repeat(64));
+// [method, argument, function that turns the result back into `input`]
+const cases = [
+  ['deflate', input, zlib.inflateSync],
+  ['inflate', zlib.deflateSync(input)],
+  ['gzip', input, zlib.gunzipSync],
+  ['gunzip', zlib.gzipSync(input)],
+  ['unzip', zlib.gzipSync(input)],
+  ['deflateRaw', input, zlib.inflateRawSync],
+  ['inflateRaw', zlib.deflateRawSync(input)],
+  ['brotliCompress', input, zlib.brotliDecompressSync],
+  ['brotliDecompress', zlib.brotliCompressSync(input)],
+  ['zstdDecompress', zlib.zstdCompressSync(input)],
+];
+
+// Counts the callbacks into JavaScript (`before` hooks) of each ZLIB resource
+// created while `fn` runs.
+async function countRoundTrips(fn) {
+  const callbacks = new Map();
+  const hook = async_hooks.createHook({
+    init(asyncId, type) {
+      if (type === 'ZLIB') callbacks.set(asyncId, 0);
+    },
+    before(asyncId) {
+      if (callbacks.has(asyncId))
+        callbacks.set(asyncId, callbacks.get(asyncId) + 1);
+    },
+  }).enable();
+  try {
+    await fn();
+  } finally {
+    hook.disable();
+  }
+  return [...callbacks.values()];
+}
+
+function callAsync(method, buffer, options = {}) {
+  return new Promise((resolve, reject) => {
+    zlib[method](buffer, options, common.mustCall((err, result) => {
+      if (err) reject(err);
+      else resolve(result);
+    }));
+  });
+}
+
+(async () => {
+  for (const [method, buffer, undo = (result) => result] of cases) {
+    let result;
+    const roundTrips = await countRoundTrips(async () => {
+      result = await callAsync(method, buffer);
+    });
+    assert.deepStrictEqual(roundTrips, [1], method);
+    assert.deepStrictEqual(undo(result), input, method);
+  }
+
+  // Errors are also reported after a single round trip.
+  let error;
+  const roundTrips = await countRoundTrips(async () => {
+    error = await callAsync('inflate', Buffer.from('not zlib data'))
+      .catch((err) => err);
+  });
+  assert.deepStrictEqual(roundTrips, [1]);
+  assert.strictEqual(error.code, 'Z_DATA_ERROR');
+})().then(common.mustCall());
diff --git a/test/parallel/test-zlib-convenience-async.js b/test/parallel/test-zlib-convenience-async.js
new file mode 100644
index 00000000000..c66f259fea0
--- /dev/null
+++ b/test/parallel/test-zlib-convenience-async.js
@@ -0,0 +1,309 @@
+'use strict';
+// Tests the contract of the asynchronous convenience methods
+// (zlib.deflate(), zlib.gunzip(), zlib.brotliCompress(), ...).
+
+const common = require('../common');
+const assert = require('assert');
+const { AsyncLocalStorage } = require('async_hooks');
+const fs = require('fs');
+const { promisify } = require('util');
+const zlib = require('zlib');
+
+const {
+  Z_SYNC_FLUSH,
+  Z_FULL_FLUSH,
+  Z_BUF_ERROR,
+  Z_DATA_ERROR,
+  Z_NEED_DICT,
+  Z_MIN_CHUNK,
+  Z_DEFAULT_CHUNK,
+  BROTLI_OPERATION_FLUSH,
+  ZSTD_e_flush,
+} = zlib.constants;
+
+const pairs = [
+  { compress: 'deflate', decompress: 'inflate', engine: 'Inflate' },
+  { compress: 'gzip', decompress: 'gunzip', engine: 'Gunzip' },
+  { compress: 'gzip', decompress: 'unzip', engine: 'Unzip' },
+  { compress: 'deflateRaw', decompress: 'inflateRaw', engine: 'InflateRaw' },
+  {
+    compress: 'brotliCompress',
+    decompress: 'brotliDecompress',
+    engine: 'BrotliDecompress',
+    partialFlush: BROTLI_OPERATION_FLUSH,
+  },
+  {
+    compress: 'zstdCompress',
+    decompress: 'zstdDecompress',
+    engine: 'ZstdDecompress',
+    partialFlush: ZSTD_e_flush,
+  },
+];
+
+// Calls an asynchronous convenience method and resolves with its result.
+// The callback must be called exactly once, and never synchronously.
+function callAsync(method, input, options) {
+  return new Promise((resolve, reject) => {
+    let returned = false;
+    const callback = common.mustCall((err, result) => {
+      assert.strictEqual(returned, true,
+                         `${method}() called its callback synchronously`);
+      if (err) reject(err);
+      else resolve(result);
+    });
+    if (options === undefined) {
+      zlib[method](input, callback);
+    } else {
+      zlib[method](input, options, callback);
+    }
+    returned = true;
+  });
+}
+
+// Text plus pseudo-random bytes, large enough that the output of every
+// algorithm spans several Z_DEFAULT_CHUNK-sized output buffers.
+const text = fs.readFileSync(__filename);
+const random = Buffer.alloc(3 * Z_DEFAULT_CHUNK);
+{
+  let seed = 0x9e3779b9;
+  for (let i = 0; i < random.length; i++) {
+    seed ^= seed << 13;
+    seed ^= seed >>> 17;
+    seed ^= seed << 5;
+    random[i] = seed & 0xff;
+  }
+}
+const data = Buffer.concat([text, random, Buffer.alloc(8 * Z_DEFAULT_CHUNK, 'a')]);
+const shortString = 'blah'.repeat(8);
+
+async function testRoundTrip({ compress, decompress, partialFlush }) {
+  const syncCompress = `${compress}Sync`;
+  const syncDecompress = `${decompress}Sync`;
+
+  // Every documented input type.
+  const inputs = [
+    ['string', shortString, Buffer.from(shortString)],
+    ['long string', text.toString(), text],
+    ['Buffer', data, data],
+    ['large ArrayBuffer', new Uint8Array(data).buffer, data],
+    ...common.getBufferSources(Buffer.from(shortString)).map((view) => [
+      view[Symbol.toStringTag], view, Buffer.from(shortString),
+    ]),
+  ];
+
+  for (const [type, input, expected] of inputs) {
+    const label = `${compress}/${decompress} with ${type}`;
+    const compressed = await callAsync(compress, input);
+    assert.ok(Buffer.isBuffer(compressed), label);
+    // zstdCompress() pledges the input size and ends the frame separately,
+    // which yields different (not larger) frames than zstdCompressSync().
+    if (compress !== 'zstdCompress') {
+      assert.deepStrictEqual(compressed, zlib[syncCompress](input), label);
+    }
+    // Interoperable with the synchronous API in both directions.
+    assert.deepStrictEqual(zlib[syncDecompress](compressed), expected, label);
+    const decompressed =
+      await callAsync(decompress, zlib[syncCompress](input));
+    assert.ok(Buffer.isBuffer(decompressed), label);
+    assert.deepStrictEqual(decompressed, expected, label);
+    assert.deepStrictEqual(await callAsync(decompress, compressed), expected,
+                           label);
+  }
+
+  // A minimal chunkSize makes the work for one call span many output
+  // buffers.
+  const options = { chunkSize: Z_MIN_CHUNK };
+  const compressed = await callAsync(compress, data, options);
+  if (compress !== 'zstdCompress') {
+    assert.deepStrictEqual(compressed, zlib[syncCompress](data, options));
+  }
+  assert.deepStrictEqual(await callAsync(decompress, compressed, options),
+                         data);
+
+  // Empty input.
+  const empty = await callAsync(compress, Buffer.alloc(0));
+  assert.deepStrictEqual(await callAsync(decompress, empty), Buffer.alloc(0));
+  assert.deepStrictEqual(
+    await callAsync(decompress, Buffer.alloc(0),
+                    { finishFlush: partialFlush ?? Z_SYNC_FLUSH }),
+    Buffer.alloc(0));
+}
+
+async function testErrors({ compress, decompress, partialFlush }) {
+  const compressed = zlib[`${compress}Sync`](data);
+
+  // Truncated input.
+  const truncated = compressed.subarray(0, compressed.length >> 1);
+  await assert.rejects(callAsync(decompress, truncated), {
+    name: 'Error',
+    code: 'Z_BUF_ERROR',
+    errno: Z_BUF_ERROR,
+    message: 'unexpected end of file',
+  });
+
+  // A non-finalizing finishFlush returns as much as can be decompressed.
+  const partial = await callAsync(decompress, truncated, {
+    finishFlush: partialFlush ?? Z_SYNC_FLUSH,
+  });
+  // Zstd only emits whole blocks, and this input is truncated inside the
+  // first one.
+  if (decompress !== 'zstdDecompress') assert.ok(partial.length > 0);
+  assert.deepStrictEqual(partial, data.subarray(0, partial.length));
+
+  // maxOutputLength applies to the output of the whole call.
+  await assert.rejects(callAsync(decompress, compressed,
+                                 { maxOutputLength: data.length - 1 }), {
+    name: 'RangeError',
+    code: 'ERR_BUFFER_TOO_LARGE',
+    message: `Cannot create a Buffer larger than ${data.length - 1} bytes`,
+  });
+  assert.deepStrictEqual(
+    await callAsync(decompress, compressed, { maxOutputLength: data.length }),
+    data);
+  await assert.rejects(callAsync(compress, data, { maxOutputLength: 64 }), {
+    name: 'RangeError',
+    code: 'ERR_BUFFER_TOO_LARGE',
+    message: 'Cannot create a Buffer larger than 64 bytes',
+  });
+}
+
+async function testInfo({ compress, decompress, engine }) {
+  const compressed = zlib[`${compress}Sync`](data);
+  const result = await callAsync(decompress, compressed, { info: true });
+  assert.deepStrictEqual(result.buffer, data);
+  assert.ok(result.engine instanceof zlib[engine]);
+  assert.strictEqual(result.engine.bytesWritten, compressed.length);
+}
+
+async function testTrailingGarbage({ compress, decompress }) {
+  const compressed = zlib[`${compress}Sync`](data);
+  const withJunk = Buffer.concat([compressed, Buffer.from('junk')]);
+  await assert.rejects(
+    callAsync(decompress, withJunk, { rejectGarbageAfterEnd: true }), {
+      name: 'TypeError',
+      code: 'ERR_TRAILING_JUNK_AFTER_STREAM_END',
+    });
+  // Gzip treats bytes after a member as the start of another member.
+  if (decompress !== 'gunzip' && decompress !== 'unzip') {
+    assert.deepStrictEqual(await callAsync(decompress, withJunk), data);
+  }
+}
+
+(async () => {
+  for (const pair of pairs) {
+    await testRoundTrip(pair);
+    await testErrors(pair);
+    await testInfo(pair);
+    await testTrailingGarbage(pair);
+  }
+
+  // Invalid compressed data.
+  for (const [method, expected] of [
+    ['inflate', { code: 'Z_DATA_ERROR', errno: Z_DATA_ERROR,
+                  message: 'incorrect header check' }],
+    ['gunzip', { code: 'Z_DATA_ERROR', errno: Z_DATA_ERROR,
+                 message: 'incorrect header check' }],
+    ['unzip', { code: 'Z_DATA_ERROR', errno: Z_DATA_ERROR,
+                message: 'incorrect header check' }],
+    ['inflateRaw', { code: 'Z_DATA_ERROR', errno: Z_DATA_ERROR }],
+    ['brotliDecompress', { code: /^ERR__ERROR_FORMAT_/,
+                           message: 'Decompression failed' }],
+    ['zstdDecompress', { code: 'ZSTD_error_prefix_unknown',
+                         message: 'Unknown frame descriptor' }],
+  ]) {
+    await assert.rejects(
+      callAsync(method, Buffer.from('this is not compressed data')),
+      { name: 'Error', ...expected });
+  }
+
+  // Concatenated gzip members are all decompressed.
+  {
+    const members = Buffer.concat([zlib.gzipSync('abc'), zlib.gzipSync('def')]);
+    assert.strictEqual((await callAsync('gunzip', members)).toString(),
+                       'abcdef');
+  }
+
+  // Dictionaries.
+  {
+    const dictionary = Buffer.from('a dictionary for the zlib tests');
+    const input = 'zlib tests, a dictionary for the zlib tests';
+    const compressed = await callAsync('deflate', input, { dictionary });
+    assert.strictEqual(
+      (await callAsync('inflate', compressed, { dictionary })).toString(),
+      input);
+    await assert.rejects(callAsync('inflate', compressed), {
+      code: 'Z_NEED_DICT',
+      errno: Z_NEED_DICT,
+      message: 'Missing dictionary',
+    });
+    await assert.rejects(
+      callAsync('inflate', compressed, { dictionary: Buffer.from('nope') }), {
+        code: 'Z_NEED_DICT',
+        message: 'Bad dictionary',
+      });
+  }
+
+  // Non-default flush and compression options still round-trip.
+  for (const options of [
+    { flush: Z_SYNC_FLUSH },
+    { flush: Z_FULL_FLUSH, level: 1 },
+    { level: 9, memLevel: 9, windowBits: 9, strategy: zlib.constants.Z_RLE },
+  ]) {
+    const compressed = await callAsync('deflate', data, options);
+    assert.deepStrictEqual(zlib.inflateSync(compressed, options), data);
+  }
+
+  // Strings are encoded with the stream defaultEncoding option.
+  {
+    const compressed = await callAsync('deflate', 'ff00',
+                                       { defaultEncoding: 'hex' });
+    assert.deepStrictEqual(zlib.inflateSync(compressed),
+                           Buffer.from([0xff, 0x00]));
+  }
+
+  // Many concurrent calls.
+  {
+    const inputs = Array.from({ length: 64 },
+                              (_, i) => data.subarray(i * 512, i * 1024 + 7));
+    const results = await Promise.all(inputs.map(async (input) => {
+      const compressed = await callAsync('gzip', input);
+      return callAsync('gunzip', compressed);
+    }));
+    for (let i = 0; i < inputs.length; i++) {
+      assert.deepStrictEqual(results[i], inputs[i]);
+    }
+  }
+
+  // util.promisify() works.
+  assert.deepStrictEqual(
+    zlib.inflateSync(await promisify(zlib.deflate)(data)), data);
+
+  // The callback runs in the caller's AsyncLocalStorage context.
+  {
+    const als = new AsyncLocalStorage();
+    const compressed = zlib.deflateSync(data);
+    await new Promise((resolve) => {
+      als.run('store', () => {
+        zlib.inflate(compressed, common.mustSucceed((result) => {
+          assert.strictEqual(als.getStore(), 'store');
+          assert.deepStrictEqual(result, data);
+          resolve();
+        }));
+      });
+    });
+  }
+
+  // Invalid arguments throw synchronously without calling the callback.
+  assert.throws(() => zlib.deflate(1, common.mustNotCall()), {
+    name: 'TypeError',
+    code: 'ERR_INVALID_ARG_TYPE',
+  });
+  assert.throws(() => zlib.deflate(data), {
+    name: 'TypeError',
+    code: 'ERR_INVALID_ARG_TYPE',
+  });
+  assert.throws(() => zlib.deflate(data, { level: 42 }, common.mustNotCall()), {
+    name: 'RangeError',
+    code: 'ERR_OUT_OF_RANGE',
+  });
+})().then(common.mustCall());