fix: added tests for double sink error

Author: maschad

src/constants.ts

@@ -27,6 +27,8 @@ export const ERR_STREAM_RESET = 'ERR_STREAM_RESET'
 export const ERR_STREAM_ABORT = 'ERR_STREAM_ABORT'
 export const ERR_MAX_OUTBOUND_STREAMS_EXCEEDED = 'ERROR_MAX_OUTBOUND_STREAMS_EXCEEDED'
 export const ERR_DECODE_IN_PROGRESS = 'ERR_DECODE_IN_PROGRESS'
+export const ERR_DOUBLE_SINK = 'ERR_DOUBLE_SINK'
+export const ERR_SINK_ENDED = 'ERR_SINK_ENDED'
 
 /**
  * INITIAL_STREAM_WINDOW is the initial stream window size.

src/stream.ts

@@ -5,7 +5,7 @@ import { CodeError } from '@libp2p/interfaces/errors'
 import { abortableSource } from 'abortable-iterator'
 import type { Uint8ArrayList } from 'uint8arraylist'
 import { Flag, FrameHeader, FrameType, HEADER_LENGTH } from './frame.js'
-import { ERR_RECV_WINDOW_EXCEEDED, ERR_STREAM_ABORT, ERR_STREAM_RESET, INITIAL_STREAM_WINDOW } from './constants.js'
+import { ERR_DOUBLE_SINK, ERR_RECV_WINDOW_EXCEEDED, ERR_SINK_ENDED, ERR_STREAM_ABORT, ERR_STREAM_RESET, INITIAL_STREAM_WINDOW } from './constants.js'
 import type { Logger } from '@libp2p/logger'
 import type { Config } from './config.js'
 import { anySignal } from 'any-signal'
@@ -85,6 +85,7 @@ export class YamuxStream implements Stream {
 
   private sourceEnd: boolean
   private sinkEnd: boolean
+  private sinkSunk: boolean
 
   private readonly sendFrame: (header: FrameHeader, body?: Uint8Array) => void
   private readonly onStreamEnd: () => void
@@ -122,6 +123,7 @@ export class YamuxStream implements Stream {
 
     this.sourceEnd = false
     this.sinkEnd = false
+    this.sinkSunk = false
 
     this.sourceInput = pushable<Uint8ArrayList>({
       onEnd: (err?: Error) => {
@@ -140,8 +142,14 @@ export class YamuxStream implements Stream {
     this.source = this.createSource()
 
     this.sink = async (source: Source<Uint8Array | Uint8ArrayList>): Promise<void> => {
+      if (this.sinkSunk) {
+        throw new CodeError('sink already called on stream', ERR_DOUBLE_SINK)
+      }
+
+      this.sinkSunk = true
+
       if (this.writeState !== HalfStreamState.Open) {
-        throw new Error('stream closed for writing')
+        throw new CodeError('stream closed for writing', ERR_SINK_ENDED)
       }
 
       const signal = anySignal([this.abortController.signal, this.resetController.signal, this.closeController.signal])
@@ -292,7 +300,7 @@ export class YamuxStream implements Stream {
     }
   }
 
-  abort (err?: Error): void {
+  abort (err: Error): void {
     switch (this.state) {
       case StreamState.Finished:
         return

test/stream.spec.ts

@@ -2,6 +2,7 @@
 
 import { pipe } from 'it-pipe'
 import { expect } from 'aegir/chai'
+import randomInt from 'random-int'
 import { sleep, testClientServer } from './util.js'
 import { HalfStreamState, StreamState } from '../src/stream.js'
 import { Pushable, pushable } from 'it-pushable'
@@ -219,7 +220,7 @@ describe('stream', () => {
     expect(server['localGoAway']).to.equal(GoAwayCode.ProtocolError)
   })
 
-  it.skip('test stream sink error', async () => {
+  it('test stream sink error', async () => {
     const { client, server } = testClientServer()
 
     // don't let the server respond
@@ -242,4 +243,17 @@ describe('stream', () => {
 
     await sendPipe
   })
+
+  it('test stream double sink error', async () => {
+    const { client } = testClientServer()
+
+    const id = randomInt(1000)
+    const c1 = client.newStream(id.toString())
+
+    await c1.sink([])
+
+    // cannot sink twice
+    await expect(c1.sink([]))
+      .to.eventually.be.rejected.with.property('code', 'ERR_DOUBLE_SINK')
+  })
 })