[question] Question on cleaning up write callbacks in case of error

[vishnureddy17]

For write(), we could potentially get stuck at line 11 if the stream is destroyed before the drain is emitted:

aedes/lib/write.js

Lines 5 to 22 in e3e50de

	function write (client, packet, done) {
	let error = null
	if (client.connecting || client.connected) {
	try {
	const result = mqtt.writeToStream(packet, client.conn)
	if (!result && !client.errored) {
	client.conn.once('drain', done)
	return
	}
	} catch (e) {
	error = new Error('packet received not valid')
	}
	} else {
	error = new Error('connection closed')
	}
	setImmediate(done, error, client)
	}

To solve this problem there is this hack in the code that uses the _writableState property on the stream to force drain to be emitted if there is an error emitted from the stream:

aedes/lib/client.js

Lines 193 to 196 in 39ccdb5

	// hack to clean up the write callbacks in case of error
	const state = this.conn._writableState
	const list = typeof state.getBuffer === 'function' ? state.getBuffer() : state.buffer
	list.forEach(drainRequest)

Why do we use this private property on the stream to solve this problem? Why can't we just listen for the error event on the client within write() instead? There's likely a good reason, but I'm a bit ignorant on this.

Thanks!

I think @robertsLando or @mcollina might be able to answer this.

[robertsLando]

Hi @vishnureddy17 , I sincerly dunno the answer to your qustion, what I remember is that we had to re-introduce that hacky hack to clean up write callbacks to fix a related issue. Could you show how you would fix that without that hack?

That is one of the fix we added for: #483

[vishnureddy17]

I feel like this should prevent write() from getting stuck if the client errors without relying on the hack:

 function write (client, packet, done) { 
   let error = null 
   if (client.connecting || client.connected) { 
     try { 
       const result = mqtt.writeToStream(packet, client.conn) 
       if (!result && !client.errored) {
         // BEGIN CHANGED CODE
         const onClientError = () => {
           client.conn.removeListener('drain', onDrain);
           done()
         }
         const onDrain = () => {
           client.removeListener('error', onClientError)
           done()
         }
         client.conn.once('drain', onDrain)
         client.once('error', onError) 
         // END CHANGED CODE
         return 
       } 
     } catch (e) { 
       error = new Error('packet received not valid') 
     } 
   } else { 
     error = new Error('connection closed') 
   } 
  
   setImmediate(done, error, client) 
 } 

However, I know the stream API has a lot of weird history so I may be missing something from the picture.

This issue in the node repo might be relevant?

[robertsLando]

I would like to know @mcollina thoughts on this

[robertsLando]

@mcollina Ping