fs: make sure to write entire buffer

fs.write(v) is not guaranteed to write everything in a single
call. Make sure we don't assume so.
PR-URL: #49211
Co-authored-by: Chemi Atlow <chemi@atlow.co.il>
Reviewed-By: Benjamin Gruenbaum <benjamingr@gmail.com>
Reviewed-By: Robert Nagy <ronagy@icloud.com>

Author: 2 people

lib/internal/fs/streams.js

@@ -13,8 +13,10 @@ const {
 
 const{
 ERR_INVALID_ARG_TYPE,
-ERR_OUT_OF_RANGE,
 ERR_METHOD_NOT_IMPLEMENTED,
+ERR_OUT_OF_RANGE,
+ERR_STREAM_DESTROYED,
+ERR_SYSTEM_ERROR,
 }=require('internal/errors').codes;
 const{
  deprecate,
@@ -392,22 +394,75 @@ WriteStream.prototype.open = openWriteFs;
 
 WriteStream.prototype._construct=_construct;
 
+functionwriteAll(data,size,pos,cb,retries=0){
+this[kFs].write(this.fd,data,0,size,pos,(er,bytesWritten,buffer)=>{
+// No data currently available and operation should be retried later.
+if(er?.code==='EAGAIN'){
+er=null;
+bytesWritten=0;
+}
+
+if(this.destroyed||er){
+returncb(er||newERR_STREAM_DESTROYED('write'));
+}
+
+this.bytesWritten+=bytesWritten;
+
+retries=bytesWritten ? 0 : retries+1;
+size-=bytesWritten;
+pos+=bytesWritten;
+
+// Try writing non-zero number of bytes up to 5 times.
+if(retries>5){
+cb(newERR_SYSTEM_ERROR('write failed'));
+}elseif(size){
+writeAll.call(this,buffer.slice(bytesWritten),size,pos,cb,retries);
+}else{
+cb();
+}
+});
+}
+
+functionwritevAll(chunks,size,pos,cb,retries=0){
+this[kFs].writev(this.fd,chunks,this.pos,(er,bytesWritten,buffers)=>{
+// No data currently available and operation should be retried later.
+if(er?.code==='EAGAIN'){
+er=null;
+bytesWritten=0;
+}
+
+if(this.destroyed||er){
+returncb(er||newERR_STREAM_DESTROYED('writev'));
+}
+
+this.bytesWritten+=bytesWritten;
+
+retries=bytesWritten ? 0 : retries+1;
+size-=bytesWritten;
+pos+=bytesWritten;
+
+// Try writing non-zero number of bytes up to 5 times.
+if(retries>5){
+cb(newERR_SYSTEM_ERROR('writev failed'));
+}elseif(size){
+writevAll.call(this,[Buffer.concat(buffers).slice(bytesWritten)],size,pos,cb,retries);
+}else{
+cb();
+}
+});
+}
+
 WriteStream.prototype._write=function(data,encoding,cb){
 this[kIsPerformingIO]=true;
-this[kFs].write(this.fd,data,0,data.length,this.pos,(er,bytes)=>{
+writeAll.call(this,data,data.length,this.pos,(er)=>{
 this[kIsPerformingIO]=false;
 if(this.destroyed){
 // Tell ._destroy() that it's safe to close the fd now.
 cb(er);
 returnthis.emit(kIoDone,er);
 }
 
-if(er){
-returncb(er);
-}
-
-this.bytesWritten+=bytes;
-cb();
+cb(er);
 });
 
 if(this.pos!==undefined)
@@ -427,20 +482,15 @@ WriteStream.prototype._writev = function(data, cb) {
 }
 
 this[kIsPerformingIO]=true;
-this[kFs].writev(this.fd,chunks,this.pos,(er,bytes)=>{
+writevAll.call(this,chunks,size,this.pos,(er)=>{
 this[kIsPerformingIO]=false;
 if(this.destroyed){
 // Tell ._destroy() that it's safe to close the fd now.
 cb(er);
 returnthis.emit(kIoDone,er);
 }
 
-if(er){
-returncb(er);
-}
-
-this.bytesWritten+=bytes;
-cb();
+cb(er);
 });
 
 if(this.pos!==undefined)

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

@@ -0,0 +1,39 @@
+import*ascommonfrom'../common/index.mjs';
+importtmpdirfrom'../common/tmpdir.js';
+importassertfrom'node:assert';
+importfsfrom'node:fs';
+import{describe,it,mock}from'node:test';
+import{finished}from'node:stream/promises';
+
+tmpdir.refresh();
+constfile=tmpdir.resolve('writeStreamEAGAIN.txt');
+consterrorWithEAGAIN=(fd,buffer,offset,length,position,callback)=>{
+callback(Object.assign(newError(),{code: 'EAGAIN'}),0,buffer);
+};
+
+describe('WriteStream EAGAIN',{concurrency: true},()=>{
+it('_write',async()=>{
+constmockWrite=mock.fn(fs.write);
+mockWrite.mock.mockImplementationOnce(errorWithEAGAIN);
+conststream=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());
+awaitfinished(stream);
+assert.strictEqual(mockWrite.mock.callCount(),2);
+assert.strictEqual(fs.readFileSync(file,'utf8'),'foo');
+});
+
+it('_write',async()=>{
+conststream=fs.createWriteStream(file);
+mock.getter(stream,'destroyed',()=>true);
+stream.end('foo');
+awaitfinished(stream).catch(common.mustCall());
+});
+});