Merge pull request hyperledger#3032 from jamshale/feat/3017

Prevent getting stuck with no active registry

aries_cloudagent/protocols/issue_credential/v2_0/formats/indy/handler.py

@@ -1,19 +1,19 @@
 """V2.0 issue-credential indy credential format handler."""
 
+import asyncio
+import json
 import logging
+from typing import Mapping, Optional, Tuple
 
 from marshmallow import RAISE
-import json
-from typing import Mapping, Tuple
-import asyncio
 
 from ......cache.base import BaseCache
 from ......core.profile import Profile
-from ......indy.issuer import IndyIssuer, IndyIssuerRevocationRegistryFullError
 from ......indy.holder import IndyHolder, IndyHolderError
+from ......indy.issuer import IndyIssuer, IndyIssuerRevocationRegistryFullError
 from ......indy.models.cred import IndyCredentialSchema
-from ......indy.models.cred_request import IndyCredRequestSchema
 from ......indy.models.cred_abstract import IndyCredAbstractSchema
+from ......indy.models.cred_request import IndyCredRequestSchema
 from ......ledger.base import BaseLedger
 from ......ledger.multiple_ledger.ledger_requests_executor import (
     GET_CRED_DEF,
@@ -30,7 +30,6 @@
 from ......revocation.models.issuer_cred_rev_record import IssuerCredRevRecord
 from ......revocation.models.revocation_registry import RevocationRegistry
 from ......storage.base import BaseStorage
-
 from ...message_types import (
     ATTACHMENT_FORMAT,
     CRED_20_ISSUE,
@@ -39,16 +38,14 @@
     CRED_20_REQUEST,
 )
 from ...messages.cred_format import V20CredFormat
-from ...messages.cred_proposal import V20CredProposal
+from ...messages.cred_issue import V20CredIssue
 from ...messages.cred_offer import V20CredOffer
+from ...messages.cred_proposal import V20CredProposal
 from ...messages.cred_request import V20CredRequest
-from ...messages.cred_issue import V20CredIssue
 from ...models.cred_ex_record import V20CredExRecord
 from ...models.detail.indy import V20CredExRecordIndy
-
-from ..handler import CredFormatAttachment, V20CredFormatError, V20CredFormatHandler
 from ..anoncreds.handler import AnonCredsCredFormatHandler
-
+from ..handler import CredFormatAttachment, V20CredFormatError, V20CredFormatHandler
 
 LOGGER = logging.getLogger(__name__)
 
@@ -369,54 +366,18 @@ async def receive_request(
                 "Indy issue credential format cannot start from credential request"
             )
 
-    async def issue_credential(
-        self, cred_ex_record: V20CredExRecord, retries: int = 5
-    ) -> CredFormatAttachment:
-        """Issue indy credential."""
-        # Temporary shim while the new anoncreds library integration is in progress
-        if self.anoncreds_handler:
-            return await self.anoncreds_handler.issue_credential(
-                cred_ex_record, retries
-            )
-
-        await self._check_uniqueness(cred_ex_record.cred_ex_id)
-
-        cred_offer = cred_ex_record.cred_offer.attachment(IndyCredFormatHandler.format)
-        cred_request = cred_ex_record.cred_request.attachment(
-            IndyCredFormatHandler.format
-        )
-        cred_values = cred_ex_record.cred_offer.credential_preview.attr_dict(
-            decode=False
-        )
-        schema_id = cred_offer["schema_id"]
-        cred_def_id = cred_offer["cred_def_id"]
-
-        issuer = self.profile.inject(IndyIssuer)
-        multitenant_mgr = self.profile.inject_or(BaseMultitenantManager)
-        if multitenant_mgr:
-            ledger_exec_inst = IndyLedgerRequestsExecutor(self.profile)
-        else:
-            ledger_exec_inst = self.profile.inject(IndyLedgerRequestsExecutor)
-        ledger = (
-            await ledger_exec_inst.get_ledger_for_identifier(
-                schema_id,
-                txn_record_type=GET_SCHEMA,
-            )
-        )[1]
-        async with ledger:
-            schema = await ledger.get_schema(schema_id)
-            cred_def = await ledger.get_credential_definition(cred_def_id)
-        revocable = cred_def["value"].get("revocation")
-        result = None
-
-        for attempt in range(max(retries, 1)):
-            if attempt > 0:
-                LOGGER.info(
-                    "Waiting 2s before retrying credential issuance for cred def '%s'",
-                    cred_def_id,
-                )
-                await asyncio.sleep(2)
-
+    async def _issue_credential_retry(
+        self,
+        retries: int,
+        cred_def_id: str,
+        schema: dict,
+        cred_offer: dict,
+        cred_request: dict,
+        cred_values: dict[str, str],
+        revocable: bool,
+        issuer: IndyIssuer,
+    ) -> tuple[Optional[dict], Optional[str], Optional[str]]:
+        for _ in range(max(retries, 1)):
             if revocable:
                 revoc = IndyRevocation(self.profile)
                 registry_info = await revoc.get_or_create_active_registry(cred_def_id)
@@ -449,7 +410,71 @@ async def issue_credential(
                 del revoc
 
             result = self.get_format_data(CRED_20_ISSUE, json.loads(cred_json))
-            break
+            if result:
+                return result, rev_reg_id, cred_rev_id
+
+            LOGGER.info(
+                "Waiting 2s before retrying credential issuance for cred def '%s'",
+                cred_def_id,
+            )
+            await asyncio.sleep(2)
+
+        return None, None, None
+
+    async def _get_ledger_for_schema(self, schema_id: str):
+        multitenant_mgr = self.profile.inject_or(BaseMultitenantManager)
+        if multitenant_mgr:
+            ledger_exec_inst = IndyLedgerRequestsExecutor(self.profile)
+        else:
+            ledger_exec_inst = self.profile.inject(IndyLedgerRequestsExecutor)
+        return (
+            await ledger_exec_inst.get_ledger_for_identifier(
+                schema_id,
+                txn_record_type=GET_SCHEMA,
+            )
+        )[1]
+
+    async def issue_credential(
+        self, cred_ex_record: V20CredExRecord, retries: int = 5
+    ) -> CredFormatAttachment:
+        """Issue indy credential."""
+        # Temporary shim while the new anoncreds library integration is in progress
+        if self.anoncreds_handler:
+            return await self.anoncreds_handler.issue_credential(
+                cred_ex_record, retries
+            )
+
+        await self._check_uniqueness(cred_ex_record.cred_ex_id)
+
+        cred_offer = cred_ex_record.cred_offer.attachment(IndyCredFormatHandler.format)
+        cred_request = cred_ex_record.cred_request.attachment(
+            IndyCredFormatHandler.format
+        )
+        cred_values = cred_ex_record.cred_offer.credential_preview.attr_dict(
+            decode=False
+        )
+        schema_id = cred_offer["schema_id"]
+        cred_def_id = cred_offer["cred_def_id"]
+
+        issuer = self.profile.inject(IndyIssuer)
+        ledger = await self._get_ledger_for_schema(schema_id)
+
+        async with ledger:
+            schema = await ledger.get_schema(schema_id)
+            cred_def = await ledger.get_credential_definition(cred_def_id)
+
+        revocable = True if cred_def["value"].get("revocation") else False
+
+        result, rev_reg_id, cred_rev_id = await self._issue_credential_retry(
+            retries,
+            cred_def_id,
+            schema,
+            cred_offer,
+            cred_request,
+            cred_values,
+            revocable,
+            issuer,
+        )
 
         if not result:
             raise V20CredFormatError(

aries_cloudagent/revocation/indy.py

@@ -222,7 +222,7 @@ async def get_issuer_rev_reg_delta(
         return rev_reg_delta
 
     async def get_or_create_active_registry(
-        self, cred_def_id: str, max_cred_num: int = None
+        self, cred_def_id: str
     ) -> Optional[Tuple[IssuerRevRegRecord, RevocationRegistry]]:
         """Fetch the active revocation registry.
 
@@ -240,14 +240,35 @@ async def get_or_create_active_registry(
             pass
 
         async with self._profile.session() as session:
-            rev_reg_recs = await IssuerRevRegRecord.query_by_cred_def_id(
-                session, cred_def_id, {"$neq": IssuerRevRegRecord.STATE_FULL}
+            full_registries = await IssuerRevRegRecord.query_by_cred_def_id(
+                session, cred_def_id, None, IssuerRevRegRecord.STATE_FULL, 1
             )
-            if not rev_reg_recs:
+
+            # all registries are full, create a new one
+            if not full_registries:
+                # Use any registry to get max cred num
+                any_registry = (
+                    await IssuerRevRegRecord.query_by_cred_def_id(
+                        session, cred_def_id, limit=1
+                    )
+                )[0]
                 await self.init_issuer_registry(
                     cred_def_id,
-                    max_cred_num=max_cred_num,
+                    max_cred_num=any_registry.max_cred_num,
+                )
+            # if there is a posted registry, activate oldest
+            else:
+                posted_registries = await IssuerRevRegRecord.query_by_cred_def_id(
+                    session, cred_def_id, IssuerRevRegRecord.STATE_POSTED, None, None
                 )
+                if posted_registries:
+                    posted_registries = sorted(
+                        posted_registries, key=lambda r: r.created_at
+                    )
+                    await self._set_registry_status(
+                        revoc_reg_id=posted_registries[0].revoc_reg_id,
+                        state=IssuerRevRegRecord.STATE_ACTIVE,
+                    )
         return None
 
     async def get_ledger_registry(self, revoc_reg_id: str) -> RevocationRegistry:

aries_cloudagent/revocation/models/issuer_rev_reg_record.py

@@ -524,7 +524,12 @@ def get_registry(self) -> RevocationRegistry:
 
     @classmethod
     async def query_by_cred_def_id(
-        cls, session: ProfileSession, cred_def_id: str, state: str = None
+        cls,
+        session: ProfileSession,
+        cred_def_id: str,
+        state: str = None,
+        negative_state: str = None,
+        limit=None,
     ) -> Sequence["IssuerRevRegRecord"]:
         """Retrieve issuer revocation registry records by credential definition ID.
 
@@ -539,7 +544,13 @@ async def query_by_cred_def_id(
                 (("cred_def_id", cred_def_id), ("state", state)),
             )
         )
-        return await cls.query(session, tag_filter)
+        return await cls.query(
+            session,
+            tag_filter,
+            post_filter_positive={"state": state} if state else None,
+            post_filter_negative={"state": negative_state} if negative_state else None,
+            limit=limit,
+        )
 
     @classmethod
     async def query_by_pending(

aries_cloudagent/revocation/tests/test_indy.py

@@ -1,6 +1,7 @@
-from aries_cloudagent.tests import mock
 from unittest import IsolatedAsyncioTestCase
 
+from aries_cloudagent.tests import mock
+
 from ...core.in_memory import InMemoryProfile
 from ...ledger.base import BaseLedger
 from ...ledger.multiple_ledger.ledger_requests_executor import (
@@ -9,7 +10,6 @@
 from ...multitenant.base import BaseMultitenantManager
 from ...multitenant.manager import MultitenantManager
 from ...storage.error import StorageNotFoundError
-
 from ..error import (
     RevocationNotSupportedError,
     RevocationRegistryBadSizeError,
@@ -255,3 +255,72 @@ async def test_get_ledger_registry(self):
         mock_from_def.assert_called_once_with(
             self.ledger.get_revoc_reg_def.return_value, True
         )
+
+    @mock.patch(
+        "aries_cloudagent.revocation.indy.IndyRevocation.get_active_issuer_rev_reg_record",
+        mock.CoroutineMock(
+            return_value=mock.MagicMock(
+                get_registry=mock.MagicMock(
+                    return_value=mock.MagicMock(
+                        get_or_fetch_local_tails_path=mock.CoroutineMock(
+                            return_value="dummy"
+                        )
+                    )
+                )
+            )
+        ),
+    )
+    async def test_get_or_create_active_registry_has_active_registry(self, *_):
+        result = await self.revoc.get_or_create_active_registry("cred_def_id")
+        assert isinstance(result, tuple)
+
+    @mock.patch(
+        "aries_cloudagent.revocation.indy.IndyRevocation.get_active_issuer_rev_reg_record",
+        mock.CoroutineMock(side_effect=StorageNotFoundError("No such record")),
+    )
+    @mock.patch(
+        "aries_cloudagent.revocation.indy.IndyRevocation.init_issuer_registry",
+        mock.CoroutineMock(return_value=None),
+    )
+    @mock.patch.object(
+        IssuerRevRegRecord,
+        "query_by_cred_def_id",
+        side_effect=[[], [IssuerRevRegRecord(max_cred_num=3)]],
+    )
+    async def test_get_or_create_active_registry_has_no_active_and_only_full_registies(
+        self, *_
+    ):
+        result = await self.revoc.get_or_create_active_registry("cred_def_id")
+
+        assert not result
+        assert self.revoc.init_issuer_registry.call_args.kwargs["max_cred_num"] == 3
+
+    @mock.patch(
+        "aries_cloudagent.revocation.indy.IndyRevocation.get_active_issuer_rev_reg_record",
+        mock.CoroutineMock(side_effect=StorageNotFoundError("No such record")),
+    )
+    @mock.patch(
+        "aries_cloudagent.revocation.indy.IndyRevocation._set_registry_status",
+        mock.CoroutineMock(return_value=None),
+    )
+    @mock.patch.object(
+        IssuerRevRegRecord,
+        "query_by_cred_def_id",
+        side_effect=[
+            [IssuerRevRegRecord(max_cred_num=3)],
+            [
+                IssuerRevRegRecord(
+                    revoc_reg_id="test-rev-reg-id",
+                    state=IssuerRevRegRecord.STATE_POSTED,
+                )
+            ],
+        ],
+    )
+    async def test_get_or_create_active_registry_has_no_active_with_posted(self, *_):
+        result = await self.revoc.get_or_create_active_registry("cred_def_id")
+
+        assert not result
+        assert (
+            self.revoc._set_registry_status.call_args.kwargs["state"]
+            == IssuerRevRegRecord.STATE_ACTIVE
+        )