v2 cancel bit

Author: blast-hardcheese

src/replit_river/v2/client_session.py

@@ -37,6 +37,7 @@
 from replit_river.session import Session
 from replit_river.transport_options import MAX_MESSAGE_BUFFER_SIZE, TransportOptions
 
+STREAM_CANCEL_BIT = 0b00100  # Synonymous with the cancel bit in v2
 STREAM_CLOSED_BIT = 0b01000  # Synonymous with the cancel bit in v2
 
 
@@ -207,11 +208,10 @@ async def send_rpc[R, A](
                 async with asyncio.timeout(timeout.total_seconds()):
                     response = await output.get()
             except asyncio.TimeoutError as e:
-                # TODO(dstewart) After protocol v2, change this to STREAM_CANCEL_BIT
                 await self.send_message(
                     stream_id=stream_id,
-                    control_flags=STREAM_CLOSED_BIT,
-                    payload={"type": "CLOSE"},
+                    control_flags=STREAM_CANCEL_BIT,
+                    payload={"type": "CANCEL"},
                     service_name=service_name,
                     procedure_name=procedure_name,
                     span=span,