Avoid replaying parsed writes when resize flushes the write queue

src/common/input/WriteBuffer.test.ts

@@ -17,6 +17,13 @@ function fromBytes(bytes: Uint8Array): string {
   return bytes.toString();
 }
 
+class TestWriteBuffer extends WriteBuffer {
+  public processOneChunk(): void {
+    // An expired time budget makes _innerWrite yield after its first chunk.
+    this._innerWrite(-Infinity);
+  }
+}
+
 describe('WriteBuffer', () => {
   let wb: WriteBuffer;
   let stack: (string | Uint8Array)[] = [];
@@ -26,6 +33,7 @@ describe('WriteBuffer', () => {
     cbStack = [];
     wb = new WriteBuffer(data => { stack.push(data); });
   });
+  afterEach(() => wb.dispose());
   describe('write input', () => {
     it('string', done => {
       wb.write('a._');
@@ -120,6 +128,62 @@ describe('WriteBuffer', () => {
         done();
       });
     });
+    it('flushSync after a yield does not repeat parsed chunks or callbacks', async () => {
+      const buffer = new TestWriteBuffer(data => { stack.push(data); });
+      wb = buffer;
+      wb.write('a', () => { cbStack.push('a'); });
+      wb.write(toBytes('b'), () => { cbStack.push('b'); });
+      wb.write('c', () => { cbStack.push('c'); });
+      buffer.processOneChunk();
+      assert.deepEqual(stack, ['a']);
+      assert.deepEqual(cbStack, ['a']);
+
+      wb.flushSync();
+      assert.deepEqual(stack, ['a', toBytes('b'), 'c']);
+      assert.deepEqual(cbStack, ['a', 'b', 'c']);
+      // Let the previously scheduled continuation run before a new write can replace it.
+      await new Promise<void>(resolve => setTimeout(resolve, 0));
+      assert.deepEqual(stack, ['a', toBytes('b'), 'c']);
+      assert.deepEqual(cbStack, ['a', 'b', 'c']);
+      await new Promise<void>(resolve => wb.write('d', resolve));
+      assert.deepEqual(stack, ['a', toBytes('b'), 'c', 'd']);
+      assert.deepEqual(cbStack, ['a', 'b', 'c']);
+    });
+    it('flushSync from a write callback does not repeat parsed chunks or callbacks', async () => {
+      await new Promise<void>(resolve => {
+        wb.write('a', () => {
+          cbStack.push('a');
+          wb.flushSync();
+        });
+        wb.write('b', () => {
+          cbStack.push('b');
+          resolve();
+        });
+      });
+      assert.deepEqual(stack, ['a', 'b']);
+      assert.deepEqual(cbStack, ['a', 'b']);
+      await new Promise<void>(resolve => wb.write('c', resolve));
+      assert.deepEqual(stack, ['a', 'b', 'c']);
+    });
+    it('flushSync processes empty chunks and their callbacks', () => {
+      wb.write('a', () => { cbStack.push('a'); });
+      wb.write('', () => { cbStack.push('empty string'); });
+      wb.write(new Uint8Array(0), () => { cbStack.push('empty bytes'); });
+      wb.write('b', () => { cbStack.push('b'); });
+      wb.flushSync();
+      assert.deepEqual(stack, ['a', '', new Uint8Array(0), 'b']);
+      assert.deepEqual(cbStack, ['a', 'empty string', 'empty bytes', 'b']);
+    });
+    it('flushSync includes writes queued by a callback', () => {
+      wb.write('a', () => {
+        cbStack.push('a');
+        wb.write('c', () => { cbStack.push('c'); });
+      });
+      wb.write('b', () => { cbStack.push('b'); });
+      wb.flushSync();
+      assert.deepEqual(stack, ['a', 'b', 'c']);
+      assert.deepEqual(cbStack, ['a', 'b', 'c']);
+    });
     it('flushSync with no pending writes is a no-op', () => {
       wb.flushSync();
       assert.deepEqual(stack, []);

src/common/input/WriteBuffer.ts

@@ -78,13 +78,12 @@ export class WriteBuffer extends Disposable {
     }
     this._isSyncWriting = true;
 
-    // Process all pending chunks synchronously
-    let chunk: string | Uint8Array | undefined;
+    // _innerWrite leaves parsed chunks before _bufferOffset when it yields.
     let didProcess = false;
-    while (chunk = this._writeBuffer.shift()) {
+    while (this._bufferOffset < this._writeBuffer.length) {
       didProcess = true;
-      this._action(chunk);
-      const cb = this._callbacks.shift();
+      this._action(this._writeBuffer[this._bufferOffset]);
+      const cb = this._callbacks[this._bufferOffset++];
       if (cb) cb();
     }
 
@@ -287,10 +286,11 @@ export class WriteBuffer extends Disposable {
         return;
       }
 
+      // Retire the chunk before its callback can synchronously flush the remaining queue.
       const cb = this._callbacks[this._bufferOffset];
-      if (cb) cb();
       this._bufferOffset++;
       this._pendingData -= data.length;
+      if (cb) cb();
 
       if (performance.now() - startTime >= Constants.WRITE_TIMEOUT_MS) {
         break;

src/headless/public/Terminal.test.ts

@@ -67,6 +67,33 @@ describe('Headless API Tests', function (): void {
     strictEqual(result, 'ABC');
   });
 
+  it('resize after a write yield does not duplicate output or callbacks', async () => {
+    await writeSync('> ');
+    const callbacks: string[] = [];
+    const now = performance.now;
+    let time = 0;
+    // Exhaust the write time budget after the echo, before the command output.
+    performance.now = () => time += 20;
+    try {
+      await new Promise<void>(resolve => {
+        const listener = term.onWriteParsed(() => {
+          listener.dispose();
+          term.resize(40, 10);
+          resolve();
+        });
+        term.write('git status', () => { callbacks.push('echo'); });
+        term.write('\r\nOn branch main\r\n', () => { callbacks.push('output'); });
+      });
+    } finally {
+      performance.now = now;
+    }
+    lineEquals(0, '> git status');
+    lineEquals(1, 'On branch main');
+    deepStrictEqual(callbacks, ['echo', 'output']);
+    await writeSync('> ');
+    lineEquals(2, '> ');
+  });
+
   it('writeln', async () => {
     await writelnSync('foo');
     await writelnSync('bar');