[stream] rework sender termination condition

All data up to the FIN must have been ACK'd, not just the final
chunk.

src/aioquic/quic/stream.py

@@ -261,8 +261,8 @@ def on_data_delivery(
                     self._buffer_start += size
                     del self._buffer[:size]
 
-            if stop == self._buffer_fin:
-                # the FIN has been ACK'd, we're done sending
+            if self._buffer_start == self._buffer_fin:
+                # all date up to the FIN has been ACK'd, we're done sending
                 self.is_finished = True
         else:
             if stop > start:

tests/test_stream.py

@@ -416,6 +416,45 @@ def test_sender_data_and_fin(self):
         stream.sender.on_data_delivery(QuicDeliveryState.ACKED, 8, 16)
         self.assertTrue(stream.sender.is_finished)
 
+    def test_sender_data_and_fin_ack_out_of_order(self):
+        stream = QuicStream()
+
+        # nothing to send yet
+        frame = stream.sender.get_frame(8)
+        self.assertIsNone(frame)
+
+        # write data and EOF
+        stream.sender.write(b"0123456789012345", end_stream=True)
+        self.assertEqual(list(stream.sender._pending), [range(0, 16)])
+        self.assertEqual(stream.sender.next_offset, 0)
+
+        # send a chunk
+        frame = stream.sender.get_frame(8)
+        self.assertEqual(frame.data, b"01234567")
+        self.assertFalse(frame.fin)
+        self.assertEqual(frame.offset, 0)
+        self.assertEqual(stream.sender.next_offset, 8)
+
+        # send another chunk
+        frame = stream.sender.get_frame(8)
+        self.assertEqual(frame.data, b"89012345")
+        self.assertTrue(frame.fin)
+        self.assertEqual(frame.offset, 8)
+        self.assertEqual(stream.sender.next_offset, 16)
+
+        # nothing more to send
+        frame = stream.sender.get_frame(8)
+        self.assertIsNone(frame)
+        self.assertEqual(stream.sender.next_offset, 16)
+
+        # second chunk gets acknowledged
+        stream.sender.on_data_delivery(QuicDeliveryState.ACKED, 8, 16)
+        self.assertFalse(stream.sender.is_finished)
+
+        # first chunk gets acknowledged
+        stream.sender.on_data_delivery(QuicDeliveryState.ACKED, 0, 8)
+        self.assertTrue(stream.sender.is_finished)
+
     def test_sender_data_lost(self):
         stream = QuicStream()