Remove 10min timeout from topic stream

Author: vgvoleg

ydb/_grpc/grpcwrapper/common_utils.py

@@ -34,6 +34,10 @@
     from ..common.protos import ydb_topic_pb2, ydb_issue_message_pb2
 
 from ... import issues, connection
+from ...settings import BaseRequestSettings
+
+
+DEFAULT_LONG_TIMEOUT = 31536000  # year
 
 
 class IFromProto(abc.ABC):
@@ -131,7 +135,7 @@ async def __anext__(self):
 
 class IGrpcWrapperAsyncIO(abc.ABC):
     @abc.abstractmethod
-    async def receive(self) -> Any:
+    async def receive(self, timeout: Optional[int] = None) -> Any:
         ...
 
     @abc.abstractmethod
@@ -161,6 +165,13 @@ def __init__(self, convert_server_grpc_to_wrapper):
         self._stream_call = None
         self._wait_executor = None
 
+        self._stream_settings: BaseRequestSettings = (
+            BaseRequestSettings()
+            .with_operation_timeout(DEFAULT_LONG_TIMEOUT)
+            .with_cancel_after(DEFAULT_LONG_TIMEOUT)
+            .with_timeout(DEFAULT_LONG_TIMEOUT)
+        )
+
     def __del__(self):
         self._clean_executor(wait=False)
 
@@ -188,6 +199,7 @@ async def _start_asyncio_driver(self, driver: DriverIO, stub, method):
             requests_iterator,
             stub,
             method,
+            settings=self._stream_settings,
         )
         self._stream_call = stream_call
         self.from_server_grpc = stream_call.__aiter__()
@@ -196,14 +208,29 @@ async def _start_sync_driver(self, driver: Driver, stub, method):
         requests_iterator = AsyncQueueToSyncIteratorAsyncIO(self.from_client_grpc)
         self._wait_executor = concurrent.futures.ThreadPoolExecutor(max_workers=1)
 
-        stream_call = await to_thread(driver, requests_iterator, stub, method, executor=self._wait_executor)
+        stream_call = await to_thread(
+            driver,
+            requests_iterator,
+            stub,
+            method,
+            executor=self._wait_executor,
+            settings=self._stream_settings,
+        )
         self._stream_call = stream_call
         self.from_server_grpc = SyncToAsyncIterator(stream_call.__iter__(), self._wait_executor)
 
-    async def receive(self) -> Any:
+    async def receive(self, timeout: Optional[int] = None) -> Any:
         # todo handle grpc exceptions and convert it to internal exceptions
         try:
-            grpc_message = await self.from_server_grpc.__anext__()
+            if timeout is None:
+                grpc_message = await self.from_server_grpc.__anext__()
+            else:
+
+                async def get_response():
+                    return await self.from_server_grpc.__anext__()
+
+                grpc_message = await asyncio.wait_for(get_response(), timeout)
+
         except (grpc.RpcError, grpc.aio.AioRpcError) as e:
             raise connection._rpc_error_handler(self._connection_state, e)
 

ydb/_topic_common/test_helpers.py

@@ -15,7 +15,7 @@ def __init__(self):
         self.from_client = asyncio.Queue()
         self._closed = False
 
-    async def receive(self) -> typing.Any:
+    async def receive(self, timeout: typing.Optional[int] = None) -> typing.Any:
         if self._closed:
             raise Exception("read from closed StreamMock")
 

ydb/_topic_reader/topic_reader_asyncio.py

@@ -490,7 +490,7 @@ async def _start(self, stream: IGrpcWrapperAsyncIO, init_message: StreamReadMess
         logger.debug("reader stream %s send init request", self._id)
 
         stream.write(StreamReadMessage.FromClient(client_message=init_message))
-        init_response = await stream.receive()  # type: StreamReadMessage.FromServer
+        init_response = await stream.receive(timeout=10)  # type: StreamReadMessage.FromServer
         if isinstance(init_response.server_message, StreamReadMessage.InitResponse):
             self._session_id = init_response.server_message.session_id
             logger.debug("reader stream %s initialized session=%s", self._id, self._session_id)

ydb/_topic_writer/topic_writer_asyncio.py

@@ -799,7 +799,7 @@ async def _start(self, stream: IGrpcWrapperAsyncIO, init_message: StreamWriteMes
         logger.debug("writer stream %s send init request", self._id)
         stream.write(StreamWriteMessage.FromClient(init_message))
 
-        resp = await stream.receive()
+        resp = await stream.receive(timeout=10)
         self._ensure_ok(resp)
         if not isinstance(resp, StreamWriteMessage.InitResponse):
             raise TopicWriterError("Unexpected answer for init request: %s" % resp)