Merge pull request #532 from digital-asset/python-acs-tx-advance

python: Fix a bug where the transaction stream didn't properly reset after reading the ACS.

Author: da-tanabe

VERSION

@@ -1 +1 @@
-8.4.1
+8.4.2

pyproject.toml

@@ -3,7 +3,7 @@
 
 [tool.poetry]
 name = "dazl"
-version = "8.4.1"
+version = "8.4.2"
 description = "high-level Ledger API client for Daml ledgers"
 license = "Apache-2.0"
 authors = ["Davin K. Tanabe <[email protected]>"]

python/dazl/__init__.py

@@ -57,4 +57,4 @@
     pass
 
 
-__version__ = "8.4.1"
+__version__ = "8.4.2"

python/dazl/ledger/aio/__init__.pyi

@@ -306,6 +306,15 @@ class Connection(PackageService, Protocol):
         token: Optional[TokenOrTokenProvider] = ...,
         timeout: Optional[TimeDeltaLike] = ...,
     ) -> MeteringReport: ...
+    async def prune(
+        self,
+        up_to: str,
+        submission_id: Optional[str] = ...,
+        prune_all_divulged_contracts=...,
+        *,
+        token: Optional[TokenOrTokenProvider] = ...,
+        timeout: Optional[TimeDeltaLike] = ...,
+    ) -> None: ...
     @property
     def is_closed(self) -> bool: ...
 

python/dazl/ledger/grpc/conn_aio.py

@@ -1119,6 +1119,29 @@ async def get_metering_report(
 
     # endregion
 
+    # region Miscellaneous admin calls
+
+    async def prune(
+        self,
+        up_to: str,
+        submission_id: Optional[str] = None,
+        prune_all_divulged_contracts=False,
+        *,
+        token: Optional[TokenOrTokenProvider] = None,
+        timeout: Optional[TimeDeltaLike] = None,
+    ) -> None:
+        with self._call(token=token, timeout=timeout) as call:
+            stub = call.grpc_stub(lapiadminpb.ParticipantPruningServiceStub)
+
+            request = lapiadminpb.PruneRequest(
+                prune_up_to=up_to,
+                submission_id=submission_id,
+                prune_all_divulged_contracts=prune_all_divulged_contracts,
+            )
+            await retry(lambda: stub.Prune(request, **call.grpc_kwargs), timeout=call.timeout)
+
+    # endregion
+
 
 class QueryStream(aio.QueryStreamBase):
     def __init__(
@@ -1193,6 +1216,7 @@ async def items(self):
                             await self._emit_create(event)
                         case Boundary():
                             await self._emit_boundary(event)
+                            offset = event.offset
                         case _:
                             warnings.warn(f"Received an unknown event: {event}", ProtocolWarning)
                     yield event
@@ -1208,7 +1232,7 @@ async def items(self):
             # now start returning events as they come off the transaction stream; note this
             # stream will never naturally close, so it's on the caller to call close() or to
             # otherwise exit our current context
-            log.debug("Reading a transaction stream: %s", self._offset_range)
+            log.debug("Reading a transaction stream: %s to %s", offset, self._offset_range.end)
             async for event in self._tx_events(tx_filter_pb, offset, self._offset_range.end):
                 log.debug("Received an event: %s", event)
                 match event: