python: Fix a bug that would cause long-running streams to be aborted due to idle timeout, even if the connection had active traffic. (#549)

Author: da-tanabe

python/dazl/ledger/grpc/_args_aio.py

@@ -177,6 +177,16 @@ def grpc_kwargs(self) -> GrpcKwargs:
             kwargs["metadata"] = aio.Metadata(("authorization", f"Bearer {self.token}"))
         return kwargs
 
+    @property
+    def grpc_kwargs_infinite_timeout(self) -> GrpcKwargs:
+        """
+        Generate keyword arguments that should accompany a gRPC Ledger API call.
+        """
+        kwargs = GrpcKwargs()
+        if self.token:
+            kwargs["metadata"] = aio.Metadata(("authorization", f"Bearer {self.token}"))
+        return kwargs
+
 
 class Cache:
     pass

python/dazl/ledger/grpc/conn_aio.py

@@ -1293,19 +1293,20 @@ async def _acs_events(
     ) -> AsyncIterable[CreateEvent | Boundary]:
         stub = self._call.grpc_stub(lapipb.ActiveContractsServiceStub)
         request = lapipb.GetActiveContractsRequest(ledger_id=self._call.ledger_id, filter=filter_pb)
-        self._response_stream = response_stream = stub.GetActiveContracts(
-            request,
-            **self._call.grpc_kwargs,
-        )
-
-        offset = None
 
         # Unidirectional gRPC streams cannot sensibly have a deadline because the stream may be
         # open indefinitely. However, if fetching an individual message from the stream takes a
         # long time here, we can reasonably assume that the stream is dead, because Active
         # Contract Set messages are really supposed to be sent as quickly as the server can send
         # them. In other words, a long timeout pause in the middle of pulling down ACS messages
         # would be highly unusual, so we treat them as fatal
+        self._response_stream = response_stream = stub.GetActiveContracts(
+            request,
+            **self._call.grpc_kwargs_infinite_timeout,
+        )
+
+        offset = None
+
         i = response_stream.__aiter__()
         while True:
             try:
@@ -1347,7 +1348,7 @@ async def _tx_events(
 
             self._response_stream = response_stream = stub.GetTransactions(
                 request,
-                **self._call.grpc_kwargs,
+                **self._call.grpc_kwargs_infinite_timeout,
             )
             i = response_stream.__aiter__()