stream: fix UTF-8 character corruption in fast-utf8-stream

Fix releaseWritingBuf() to correctly handle partial writes that split
multi-byte UTF-8 characters. The previous implementation incorrectly
converted byte counts to character counts, causing:
- 3-byte characters (CJK) to be silently dropped
- 4-byte characters (emoji) to leave lone surrogates in the buffer
The fix backs up from the byte position to find a valid UTF-8 character
boundary by checking for continuation bytes (pattern 10xxxxxx), then
decodes the properly-aligned bytes to get the correct character count.
Also fixes a typo where this._asyncDrainScheduled was used instead of
the private field this.#asyncDrainScheduled.
Fixes: #61744
PR-URL: #61745
Reviewed-By: Yagiz Nizipli <yagiz@nizipli.com>
Reviewed-By: James M Snell <jasnell@gmail.com>
Reviewed-By: Colin Ihrig <cjihrig@gmail.com>
Reviewed-By: Paolo Insogna <paolo@cowtech.it>

Author: mcollina

lib/internal/streams/fast-utf8-stream.js

@@ -237,7 +237,7 @@ class Utf8Stream extends EventEmitter {
 
 this.on('newListener',(name)=>{
 if(name==='drain'){
-this._asyncDrainScheduled=false;
+this.#asyncDrainScheduled=false;
 }
 });
 
@@ -894,11 +894,23 @@ class Utf8Stream extends EventEmitter {
  * @returns {{writingBuf: string | Buffer, len: number}} released writingBuf and length
  */
 functionreleaseWritingBuf(writingBuf,len,n){
-// if Buffer.byteLength is equal to n, that means writingBuf contains no multi-byte character
-if(typeofwritingBuf==='string'&&Buffer.byteLength(writingBuf)!==n){
-// Since the fs.write callback parameter `n` means how many bytes the passed of string
-// We calculate the original string length for avoiding the multi-byte character issue
-n=Buffer.from(writingBuf).subarray(0,n).toString().length;
+if(typeofwritingBuf==='string'){
+constbyteLength=Buffer.byteLength(writingBuf);
+if(byteLength!==n){
+// Since fs.write returns the number of bytes written, we need to find
+// how many complete characters fit within those n bytes.
+// If a partial write splits a multi-byte UTF-8 character, we must back up
+// to the start of that character to avoid data corruption.
+constbuf=Buffer.from(writingBuf);
+// Back up from position n to find a valid UTF-8 character boundary.
+// UTF-8 continuation bytes have the pattern 10xxxxxx (0x80-0xBF).
+// We need to find the start of the character that was split.
+while(n>0&&(buf[n]&0xC0)===0x80){
+n--;
+}
+// Decode the properly-aligned bytes to get the character count.
+n=buf.subarray(0,n).toString().length;
+}
 }
 len=MathMax(len-n,0);
 writingBuf=writingBuf.slice(n);

test/parallel/test-fastutf8stream-partial-write-utf8.js

@@ -0,0 +1,322 @@
+'use strict';
+
+// Tests for UTF-8 character preservation when partial writes split multi-byte characters.
+// See: https://github.com/nodejs/node/issues/61744
+
+constcommon=require('../common');
+consttmpdir=require('../common/tmpdir');
+constassert=require('node:assert');
+const{
+ openSync,
+ write,
+ writeSync,
+}=require('node:fs');
+const{ Utf8Stream }=require('node:fs');
+const{ join }=require('node:path');
+const{ isMainThread }=require('node:worker_threads');
+
+tmpdir.refresh();
+if(isMainThread){
+process.umask(0o000);
+}
+
+letfileCounter=0;
+
+functiongetTempFile(){
+returnjoin(tmpdir.path,`fastutf8stream-partial-${process.pid}-${Date.now()}-${fileCounter++}.log`);
+}
+
+runTests(false);
+runTests(true);
+
+functionrunTests(sync){
+// Test 1: Partial write splitting a 3-byte UTF-8 character (CJK)
+// "abc中def" where "中" is 3 bytes (E4 B8 AD)
+// Simulate partial write of 4 bytes: "abc" (3 bytes) + first byte of "中"
+// The remaining buffer should be "中def" (not "def")
+{
+constdest=getTempFile();
+constfd=openSync(dest,'w');
+
+letfirstWrite=true;
+constwrittenChunks=[];
+constfsOverride={};
+
+if(sync){
+fsOverride.writeSync=common.mustCall((...args)=>{
+constdata=args[1];
+writtenChunks.push(typeofdata==='string' ? data : data.toString());
+if(firstWrite){
+firstWrite=false;
+// Simulate partial write: only 4 bytes written out of 9
+// This splits the 3-byte "中" character
+return4;
+}
+returnwriteSync(...args);
+},2);
+}else{
+fsOverride.write=common.mustCall((...args)=>{
+constdata=args[1];
+writtenChunks.push(typeofdata==='string' ? data : data.toString());
+constcallback=args[args.length-1];
+if(firstWrite){
+firstWrite=false;
+// Simulate partial write: only 4 bytes written out of 9
+process.nextTick(callback,null,4);
+return;
+}
+returnwrite(...args);
+},2);
+}
+
+conststream=newUtf8Stream({
+ fd,
+ sync,
+minLength: 0,
+fs: fsOverride,
+});
+
+stream.on('ready',common.mustCall(()=>{
+stream.write('abc中def');
+stream.end();
+
+stream.on('finish',common.mustCall(()=>{
+// Verify the second chunk contains the preserved CJK character
+assert.strictEqual(writtenChunks.length,2);
+assert.strictEqual(writtenChunks[0],'abc中def');// First attempt
+assert.strictEqual(writtenChunks[1],'中def');// Retry with preserved char
+}));
+}));
+}
+
+// Test 2: Partial write splitting a 4-byte UTF-8 character (emoji)
+// "hello🌍world" where "🌍" is 4 bytes (F0 9F 8C 8D)
+// Simulate partial write of 7 bytes: "hello" (5 bytes) + first 2 bytes of "🌍"
+// The remaining buffer should be "🌍world" (not a lone surrogate + "world")
+{
+constdest=getTempFile();
+constfd=openSync(dest,'w');
+
+letfirstWrite=true;
+constwrittenChunks=[];
+constfsOverride={};
+
+if(sync){
+fsOverride.writeSync=common.mustCall((...args)=>{
+constdata=args[1];
+writtenChunks.push(typeofdata==='string' ? data : data.toString());
+if(firstWrite){
+firstWrite=false;
+// Simulate partial write: only 7 bytes written
+return7;
+}
+returnwriteSync(...args);
+},2);
+}else{
+fsOverride.write=common.mustCall((...args)=>{
+constdata=args[1];
+writtenChunks.push(typeofdata==='string' ? data : data.toString());
+constcallback=args[args.length-1];
+if(firstWrite){
+firstWrite=false;
+process.nextTick(callback,null,7);
+return;
+}
+returnwrite(...args);
+},2);
+}
+
+conststream=newUtf8Stream({
+ fd,
+ sync,
+minLength: 0,
+fs: fsOverride,
+});
+
+stream.on('ready',common.mustCall(()=>{
+stream.write('hello🌍world');
+stream.end();
+
+stream.on('finish',common.mustCall(()=>{
+assert.strictEqual(writtenChunks.length,2);
+assert.strictEqual(writtenChunks[0],'hello🌍world');// First attempt
+assert.strictEqual(writtenChunks[1],'🌍world');// Retry with preserved emoji
+
+// Verify no lone surrogates in the retry chunk
+constretryChunk=writtenChunks[1];
+for(leti=0;i<retryChunk.length;i++){
+constcode=retryChunk.charCodeAt(i);
+if(code>=0xD800&&code<=0xDBFF){
+// High surrogate - next must be low surrogate
+constnext=retryChunk.charCodeAt(i+1);
+assert.ok(next>=0xDC00&&next<=0xDFFF,
+`Found lone high surrogate at position ${i}`);
+i++;// Skip the low surrogate we just verified
+}elseif(code>=0xDC00&&code<=0xDFFF){
+// Low surrogate without preceding high surrogate
+assert.fail(`Found lone low surrogate at position ${i}: 0x${code.toString(16)}`);
+}
+}
+}));
+}));
+}
+
+// Test 3: Partial write at exactly 0 bytes (edge case)
+{
+constdest=getTempFile();
+constfd=openSync(dest,'w');
+
+letfirstWrite=true;
+constwrittenChunks=[];
+constfsOverride={};
+
+if(sync){
+fsOverride.writeSync=common.mustCall((...args)=>{
+constdata=args[1];
+writtenChunks.push(typeofdata==='string' ? data : data.toString());
+if(firstWrite){
+firstWrite=false;
+return0;// No bytes written
+}
+returnwriteSync(...args);
+},2);
+}else{
+fsOverride.write=common.mustCall((...args)=>{
+constdata=args[1];
+writtenChunks.push(typeofdata==='string' ? data : data.toString());
+constcallback=args[args.length-1];
+if(firstWrite){
+firstWrite=false;
+process.nextTick(callback,null,0);
+return;
+}
+returnwrite(...args);
+},2);
+}
+
+conststream=newUtf8Stream({
+ fd,
+ sync,
+minLength: 0,
+fs: fsOverride,
+});
+
+stream.on('ready',common.mustCall(()=>{
+stream.write('中文');
+stream.end();
+
+stream.on('finish',common.mustCall(()=>{
+assert.strictEqual(writtenChunks.length,2);
+assert.strictEqual(writtenChunks[0],'中文');
+assert.strictEqual(writtenChunks[1],'中文');// Entire string retried
+}));
+}));
+}
+
+// Test 4: Partial write splitting between characters (not mid-character)
+// This should work the same as before - no character preservation needed
+{
+constdest=getTempFile();
+constfd=openSync(dest,'w');
+
+letfirstWrite=true;
+constwrittenChunks=[];
+constfsOverride={};
+
+if(sync){
+fsOverride.writeSync=common.mustCall((...args)=>{
+constdata=args[1];
+writtenChunks.push(typeofdata==='string' ? data : data.toString());
+if(firstWrite){
+firstWrite=false;
+// Write exactly 3 bytes ("abc"), which is a clean character boundary
+return3;
+}
+returnwriteSync(...args);
+},2);
+}else{
+fsOverride.write=common.mustCall((...args)=>{
+constdata=args[1];
+writtenChunks.push(typeofdata==='string' ? data : data.toString());
+constcallback=args[args.length-1];
+if(firstWrite){
+firstWrite=false;
+process.nextTick(callback,null,3);
+return;
+}
+returnwrite(...args);
+},2);
+}
+
+conststream=newUtf8Stream({
+ fd,
+ sync,
+minLength: 0,
+fs: fsOverride,
+});
+
+stream.on('ready',common.mustCall(()=>{
+stream.write('abc中def');
+stream.end();
+
+stream.on('finish',common.mustCall(()=>{
+assert.strictEqual(writtenChunks.length,2);
+assert.strictEqual(writtenChunks[0],'abc中def');
+assert.strictEqual(writtenChunks[1],'中def');// Remaining after 3 bytes
+}));
+}));
+}
+
+// Test 5: Single multi-byte character with partial write of 1 byte
+{
+constdest=getTempFile();
+constfd=openSync(dest,'w');
+
+letfirstWrite=true;
+constwrittenChunks=[];
+constfsOverride={};
+
+if(sync){
+fsOverride.writeSync=common.mustCall((...args)=>{
+constdata=args[1];
+writtenChunks.push(typeofdata==='string' ? data : data.toString());
+if(firstWrite){
+firstWrite=false;
+// Write only 1 byte of a 3-byte character
+return1;
+}
+returnwriteSync(...args);
+},2);
+}else{
+fsOverride.write=common.mustCall((...args)=>{
+constdata=args[1];
+writtenChunks.push(typeofdata==='string' ? data : data.toString());
+constcallback=args[args.length-1];
+if(firstWrite){
+firstWrite=false;
+process.nextTick(callback,null,1);
+return;
+}
+returnwrite(...args);
+},2);
+}
+
+conststream=newUtf8Stream({
+ fd,
+ sync,
+minLength: 0,
+fs: fsOverride,
+});
+
+stream.on('ready',common.mustCall(()=>{
+stream.write('中');
+stream.end();
+
+stream.on('finish',common.mustCall(()=>{
+assert.strictEqual(writtenChunks.length,2);
+assert.strictEqual(writtenChunks[0],'中');
+assert.strictEqual(writtenChunks[1],'中');// Full character retried
+}));
+}));
+}
+}