Merge pull request #294 from digital-asset/python-grpc-offsets

python: Expose starting offsets for query and streams (gRPC only)

Author: da-tanabe

VERSION

@@ -1 +1 @@
-7.5.12
+7.5.13

pyproject.toml

@@ -3,7 +3,7 @@
 
 [tool.poetry]
 name = "dazl"
-version = "7.5.12"
+version = "7.5.13"
 description = "high-level Ledger API client for Daml ledgers"
 license = "Apache-2.0"
 authors = ["Davin K. Tanabe <davin.tanabe@digitalasset.com>"]

python/dazl/ledger/_offsets.py

@@ -10,6 +10,8 @@
 
 __all__ = [
     "LedgerOffsetRange",
+    "End",
+    "END",
     "UNTIL_END",
     "FROM_BEGINNING_UNTIL_FOREVER",
     "from_offset_until_forever",
@@ -21,6 +23,8 @@ class End:
     Marker object that denotes the current end of the ledger.
     """
 
+    __slots__ = ()
+
     def __hash__(self):
         return 0
 
@@ -39,7 +43,7 @@ class LedgerOffsetRange:
     so this class actually represents the commonality between these two interfaces.
     """
 
-    def __init__(self, __begin: "Union[None, str]", __end: "Union[None, End]"):
+    def __init__(self, __begin: "Union[None, str]", __end: "Union[None, str, End]"):
         """
         Initialize a :class:`LedgerOffsetRange`.
 
@@ -48,9 +52,11 @@ def __init__(self, __begin: "Union[None, str]", __end: "Union[None, End]"):
             Otherwise, must be a legal ledger offset.
         :param __end:
             The end of the stream. If ``None``, then keep reading from the stream forever; if
-            ``END``, then terminate when reaching the _current_ end of stream. Note that offsets
-            are *not* allowed here, as the HTTP JSON API does not provide a mechanism for reading
-            *to* a specific transaction offset.
+            ``END``, then terminate when reaching the _current_ end of stream.
+
+            Note that offsets are only allowed here on the gRPC Ledger API; they are *not*
+            allowed here on the HTTP JSON API does not provide a mechanism for reading *to* a
+            specific transaction offset.
         """
         self.begin = __begin
         self.end = __end
@@ -65,6 +71,9 @@ def __eq__(self, other: Any) -> bool:
     def __hash__(self):
         return hash(self.begin) ^ hash(self.end)
 
+    def __repr__(self):
+        return f"({self.begin}, {self.end})"
+
 
 UNTIL_END = LedgerOffsetRange(None, END)
 FROM_BEGINNING_UNTIL_FOREVER = LedgerOffsetRange(None, None)

python/dazl/ledger/grpc/codec_aio.py

@@ -44,6 +44,7 @@
 from ...prim import ContractData, ContractId, Party
 from ...values import Context
 from ...values.protobuf import ProtobufDecoder, ProtobufEncoder, set_value
+from .._offsets import END, End
 from ..aio import PackageLoader
 from ..pkgcache import SHARED_PACKAGE_DATABASE
 
@@ -217,11 +218,22 @@ def encode_identifier(name: TypeConName) -> lapipb.Identifier:
 
     @staticmethod
     def encode_begin_offset(offset: Optional[str]) -> lapipb.LedgerOffset:
-        return (
-            lapipb.LedgerOffset(absolute=offset)
-            if offset is not None
-            else lapipb.LedgerOffset(boundary=0)
-        )
+        if offset is None:
+            return lapipb.LedgerOffset(boundary=0)
+        else:
+            return lapipb.LedgerOffset(absolute=offset)
+
+    @staticmethod
+    def encode_end_offset(offset: Union[str, None, End]) -> Optional[lapipb.LedgerOffset]:
+        if offset is None:
+            # there is no ending offset (the stream will never naturally terminate)
+            return None
+        elif isinstance(offset, End):
+            # the offset goes up until the current end of the ledger
+            return lapipb.LedgerOffset(boundary=1)
+        else:
+            # the offset is absolute
+            return lapipb.LedgerOffset(absolute=offset)
 
     async def decode_created_event(self, event: lapipb.CreatedEvent) -> CreateEvent:
         cid = self.decode_contract_id(event)

python/dazl/ledger/grpc/conn_aio.py

@@ -13,13 +13,14 @@
 from grpc.aio import Channel, UnaryStreamCall
 
 from .. import aio
+from ... import LOG
 from ..._gen.com.daml.ledger.api import v1 as lapipb
 from ..._gen.com.daml.ledger.api.v1 import admin as lapiadminpb
 from ...damlast.daml_lf_1 import PackageRef, TypeConName
 from ...damlast.util import is_match
 from ...prim import LEDGER_STRING_REGEX, ContractData, ContractId, Party
 from ...query import Filter, Queries, Query, parse_query
-from .._offsets import UNTIL_END, LedgerOffsetRange, from_offset_until_forever
+from .._offsets import END, UNTIL_END, End, LedgerOffsetRange, from_offset_until_forever
 from ..api_types import ArchiveEvent, Boundary, Command, CreateEvent, ExerciseResponse, PartyInfo
 from ..config import Config
 from ..config.access import PropertyBasedAccessConfig
@@ -455,7 +456,12 @@ def _submit_and_wait_request(
     # region Read API
 
     def query(
-        self, __template_id: Union[str, TypeConName] = "*", __query: Query = None
+        self,
+        __template_id: Union[str, TypeConName] = "*",
+        __query: Query = None,
+        *,
+        begin_offset: Optional[str] = None,
+        end_offset: Optional[str] = None,
     ) -> "QueryStream":
         """
         Return the create events from the active contract set service as a stream.
@@ -468,12 +474,25 @@ def query(
             The name of the template for which to fetch contracts.
         :param __query:
             A filter to apply to the set of returned contracts.
-        """
+        :param begin_offset:
+            The starting offset at which to read an active contract set. If ``None``, contracts
+            are read from the beginning, and using the Active Contract Set Service instead of the
+            Transaction Service.
+        :param end_offset:
+            The ending offset. If ``None``, contracts are read until the end of the stream.
+            In order to read indefinitely, use :meth:`stream` instead.
+        """
+        offset = LedgerOffsetRange(begin_offset, end_offset if end_offset is not None else END)
         return QueryStream(
-            self, parse_query({__template_id: __query}, server_side_filters=False), UNTIL_END
+            self, parse_query({__template_id: __query}, server_side_filters=False), offset
         )
 
-    def query_many(self, *queries: Queries) -> "QueryStream":
+    def query_many(
+        self,
+        *queries: Queries,
+        begin_offset: Optional[str] = None,
+        end_offset: Optional[str] = None,
+    ) -> "QueryStream":
         """
         Return the create events from the active contract set service as a stream.
 
@@ -483,8 +502,16 @@ def query_many(self, *queries: Queries) -> "QueryStream":
 
         :param queries:
             A map of template IDs to filter to apply to the set of returned contracts.
+        :param begin_offset:
+            The starting offset at which to read an active contract set. If ``None``, contracts
+            are read from the beginning, and using the Active Contract Set Service instead of the
+            Transaction Service.
+        :param end_offset:
+            The ending offset. If ``None``, contracts are read until the end of the stream.
+            In order to read indefinitely, use :meth:`stream_many` instead.
         """
-        return QueryStream(self, parse_query(*queries, server_side_filters=False), UNTIL_END)
+        offset = LedgerOffsetRange(begin_offset, end_offset if end_offset is not None else END)
+        return QueryStream(self, parse_query(*queries, server_side_filters=False), offset)
 
     def stream(
         self,
@@ -643,8 +670,10 @@ async def items(self):
             filters_by_party = {party: filters for party in self.conn.config.access.read_as}
             tx_filter_pb = lapipb.TransactionFilter(filters_by_party=filters_by_party)
 
-            offset = None
-            if self._offset_range.begin is None:
+            offset = self._offset_range.begin
+            if offset:
+                log.debug("Skipped reading from the ACS because begin offset is %r", offset)
+            else:
                 # when starting from the beginning of the ledger, the Active Contract Set service
                 # lets us catch up more quickly than having to parse every create/archive event
                 # ourselves
@@ -658,31 +687,30 @@ async def items(self):
                     else:
                         warnings.warn(f"Received an unknown event: {event}", ProtocolWarning)
                     yield event
-            else:
-                log.debug(
-                    "Skipped reading from the ACS because begin offset is %r",
-                    self._offset_range.begin,
-                )
 
-            if self._offset_range != UNTIL_END:
-                # 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)
-                async for event in self._tx_events(tx_filter_pb, offset):
-                    if isinstance(event, CreateEvent):
-                        await self._emit_create(event)
-                    elif isinstance(event, ArchiveEvent):
-                        await self._emit_archive(event)
-                    elif isinstance(event, Boundary):
-                        await self._emit_boundary(event)
-                    else:
-                        warnings.warn(f"Received an unknown event: {event}", ProtocolWarning)
-                    yield event
-            else:
-                log.debug(
-                    "Not reading from transaction stream because we were only asked for a snapshot."
-                )
+                # when reading from the Active Contract Set service, if we're supposed to stop
+                # at "the end", then the Active Contract Set data is all that we'll return
+                if self._offset_range.end == END:
+                    log.debug(
+                        "Not reading from transaction stream because we were only asked for a snapshot."
+                    )
+                    return
+
+            # 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)
+            async for event in self._tx_events(tx_filter_pb, offset, self._offset_range.end):
+                log.debug("Received an event: %s", event)
+                if isinstance(event, CreateEvent):
+                    await self._emit_create(event)
+                elif isinstance(event, ArchiveEvent):
+                    await self._emit_archive(event)
+                elif isinstance(event, Boundary):
+                    await self._emit_boundary(event)
+                else:
+                    warnings.warn(f"Received an unknown event: {event}", ProtocolWarning)
+                yield event
 
     async def _acs_events(
         self, filter_pb: lapipb.TransactionFilter
@@ -696,23 +724,33 @@ async def _acs_events(
 
         offset = None
         async for response in response_stream:
+            LOG.debug(
+                "ACS start (offset %r, %d event(s))",
+                response.offset,
+                len(response.active_contracts),
+            )
             for event in response.active_contracts:
                 c_evt = await self.conn.codec.decode_created_event(event)
                 if self._is_match(c_evt):
                     yield c_evt
             # for ActiveContractSetResponse messages, only the last offset is actually relevant
+            LOG.debug("ACS end (offset %r)", response.offset)
             offset = response.offset
         yield Boundary(offset)
 
     async def _tx_events(
-        self, filter_pb: lapipb.TransactionFilter, begin_offset: Optional[str]
+        self,
+        filter_pb: lapipb.TransactionFilter,
+        begin_offset: Optional[str],
+        end_offset: "Union[None, str, End]",
     ) -> AsyncIterable[Union[CreateEvent, ArchiveEvent, Boundary]]:
         stub = lapipb.TransactionServiceStub(self.conn.channel)
 
         request = lapipb.GetTransactionsRequest(
             ledger_id=self.conn.config.access.ledger_id,
             filter=filter_pb,
             begin=self.conn.codec.encode_begin_offset(begin_offset),
+            end=self.conn.codec.encode_end_offset(end_offset),
         )
 
         self._response_stream = response_stream = stub.GetTransactions(request)

python/tests/unit/test_ledger_query.py

@@ -1,51 +1,88 @@
 # Copyright (c) 2017-2021 Digital Asset (Switzerland) GmbH and/or its affiliates. All rights reserved.
 # SPDX-License-Identifier: Apache-2.0
 
-from asyncio import gather
+from asyncio import gather, wait_for
+import logging
 
 import dazl
+from dazl.ledger import Boundary, CreateEvent
+from dazl.ledger.aio import Connection
+from dazl.ledger.grpc import Connection as GrpcConnection
+from dazl.prim import ContractData, Party
 import pytest
 from tests.unit import dars
 
+TEMPLATE = "Simple:OperatorNotification"
+
+
+def payload(operator: Party, text: str) -> ContractData:
+    return {"operator": operator, "theObservers": [], "text": text}
+
 
 @pytest.mark.asyncio
-async def test_query_no_filter(sandbox):
+async def test_query_no_filter(sandbox) -> None:
     async with dazl.connect(url=sandbox, admin=True) as conn:
         party_info, _ = await gather(
             conn.allocate_party(), conn.upload_package(dars.Simple.read_bytes())
         )
 
     async with dazl.connect(url=sandbox, act_as=party_info.party) as conn:
-        for text in ["Red", "Red", "Green", "Blue", "Blue", "Blue"]:
-            await conn.create(
-                "Simple:OperatorNotification",
-                {"operator": party_info.party, "theObservers": [], "text": text},
-            )
+        texts = ["Red", "Red", "Green", "Blue", "Blue", "Blue"]
+        for text in texts:
+            await conn.create(TEMPLATE, payload(party_info.party, text))
 
-        async with conn.query("Simple:OperatorNotification") as stream:
-            events = []
-            async for event in stream.creates():
-                events.append(event)
+        # it could be possible for the creates to succeed, but not yet be reflected on the
+        # read side; we'll give it as much as 30 seconds before we fail this test
+        boundary = await wait_for(wait_for_n_notifications(conn, len(texts)), timeout=30)
+        logging.info("Created %s contracts, and our offset is now %s.", len(texts), boundary.offset)
 
-        assert len(events) == 6
+        await conn.create(TEMPLATE, payload(party_info.party, "Yellow"))
 
-        async with conn.query("Simple:OperatorNotification", {"text": "Red"}) as stream:
+        async with conn.query(TEMPLATE, {"text": "Red"}) as stream:
             events = []
             async for event in stream.creates():
                 events.append(event)
 
         assert len(events) == 2
 
-        async with conn.query("Simple:OperatorNotification", {"text": "Green"}) as stream:
+        async with conn.query(TEMPLATE, {"text": "Green"}) as stream:
             events = []
             async for event in stream.creates():
                 events.append(event)
 
         assert len(events) == 1
 
-        async with conn.query("Simple:OperatorNotification", {"text": "Blue"}) as stream:
+        async with conn.query(TEMPLATE, {"text": "Blue"}) as stream:
             events = []
             async for event in stream.creates():
                 events.append(event)
 
         assert len(events) == 3
+
+        logging.info("Making sure that our last event made it in...")
+        final_offset = await wait_for(wait_for_n_notifications(conn, len(texts) + 1), timeout=30)
+
+        logging.info("Yep it's there, and the final offset is %s", final_offset.offset)
+
+        # offset-based queries only work over the gRPC API
+        assert isinstance(conn, GrpcConnection)
+
+        async with conn.query(TEMPLATE, begin_offset=boundary.offset) as stream:
+            async for event in stream.creates():
+                # only one create is supposed to come after this part in the stream, and that is
+                # "Yellow"
+                assert event.payload["text"] == "Yellow"
+
+
+async def wait_for_n_notifications(conn: Connection, n: int) -> Boundary:
+    async with conn.stream("Simple:OperatorNotification") as stream:
+        event_count = 0
+        async for obj in stream:
+            if isinstance(obj, CreateEvent):
+                event_count += 1
+            elif isinstance(obj, Boundary):
+                # return the first Boundary object we see after the expected number of CreateEvents
+                if event_count >= n:
+                    return obj
+
+        raise Exception("unexpected end of stream")