From a30fd77a1c92c1bd41af9e575964a03fac07e251 Mon Sep 17 00:00:00 2001 From: Lasse Blaauwbroek Date: Mon, 16 Oct 2023 10:59:34 +0200 Subject: [PATCH] Fix retransmit bug for large messages causing message corruption Logic bug: We are looping over segments sent by the C++ library and sending them over a python transport. If the last message is larger than the transport pause threshold, this causes the transport to pause us. In that case, we forget to increment the current write_index, causing us to retransmit the same message in an infinite loop. This is a serious bug, because it causes messages to become corrupted. --- capnp/lib/capnp.pyx | 9 +++++---- 1 file changed, 5 insertions(+), 4 deletions(-) diff --git a/capnp/lib/capnp.pyx b/capnp/lib/capnp.pyx index dffb9ca..554144a 100644 --- a/capnp/lib/capnp.pyx +++ b/capnp/lib/capnp.pyx @@ -2371,6 +2371,7 @@ cdef class _AsyncIoStream: elif self.protocol.transport is not None and hasattr(self.protocol.transport, "close"): self.protocol.transport.close() # Call connection_lost immediately, instead of waiting for the transport to do it. + # TODO: This might be a questionable thing to do... self.protocol.connection_lost("Stream is closing") async def wait_closed(self): @@ -2558,15 +2559,15 @@ cdef class _PyAsyncIoStreamProtocol(DummyBaseClass, asyncio.BufferedProtocol): self.read_reset() cdef write_loop(self): - if not self.write_in_progress: return + if self.write_paused or not self.write_in_progress: return cdef const ArrayPtr[const uint8_t]* piece for i in range(self.write_index, self.write_pieces.size()): - if self.write_paused: - self.write_index = i - break piece = &self.write_pieces[i] view = memoryview.PyMemoryView_FromMemory(piece.begin(), piece.size(), buffer.PyBUF_READ) self.transport.write(view) + if self.write_paused: + self.write_index = i+1 + break if not self.write_paused: self.write_fulfiller.fulfill() self.write_reset()