UT

Author: atlowChemi

lib/internal/fs/streams.js

@@ -415,7 +415,7 @@ function writeAll(data, size, pos, cb, retries = 0) {
     if (retries > 5) {
       cb(new ERR_SYSTEM_ERROR('write failed'));
     } else if (size) {
-      writeAll(buffer.slice(bytesWritten), size, pos, cb, retries);
+      writeAll.call(this, buffer.slice(bytesWritten), size, pos, cb, retries);
     } else {
       cb();
     }
@@ -444,7 +444,7 @@ function writevAll(chunks, size, pos, cb, retries = 0) {
     if (retries > 5) {
       cb(new ERR_SYSTEM_ERROR('writev failed'));
     } else if (size) {
-      writevAll([Buffer.concat(buffers).slice(bytesWritten)], size, pos, cb, retries);
+      writevAll.call(this, [Buffer.concat(buffers).slice(bytesWritten)], size, pos, cb, retries);
     } else {
       cb();
     }

test/parallel/test-fs-write-stream-eagain.mjs

@@ -0,0 +1,31 @@
+import * as common from '../common/index.mjs';
+import tmpdir from '../common/tmpdir.js';
+import assert from 'node:assert';
+import fs from 'node:fs';
+import { describe, it, mock } from 'node:test';
+import { finished } from 'node:stream/promises';
+
+const file = tmpdir.resolve('writeStreamEAGAIN.txt');
+const errorWithEAGAIN = (fd, buffer, offset, length, position, callback) => {
+  callback(Object.assign(new Error(), { code: 'EAGAIN' }), 0, buffer);
+}
+
+describe('WriteStream EAGAIN', { concurrency: true }, () => {
+  it('_write', async () => {
+    const mockWrite = mock.fn(fs.write);
+    mockWrite.mock.mockImplementationOnce(errorWithEAGAIN);
+    const stream = fs.createWriteStream(file, {
+      fs: {
+        open: common.mustCall(fs.open),
+        write: mockWrite,
+        close: common.mustCall(fs.close),
+      }
+    });
+    stream.end('foo');
+    stream.on('close', common.mustCall());
+    stream.on('error', common.mustNotCall());
+    await finished(stream);
+    assert.strictEqual(mockWrite.mock.callCount(), 2);
+    assert.strictEqual(fs.readFileSync(file, 'utf8'), 'foo');
+  });
+});