From 72b34280bc7f68febd422cc06b464abae2f0dcd1 Mon Sep 17 00:00:00 2001 From: Alisson Fantin Rodrigues Date: Thu, 16 Jul 2026 14:27:25 -0300 Subject: [PATCH 1/4] fix simple lock and add retry in blockchain --- aries_cloudagent/anoncreds/revocation.py | 303 ++++++++++++++---- .../anoncreds/tests/test_revocation.py | 91 ++++-- .../v2_0/handlers/cred_request_handler.py | 19 ++ 3 files changed, 326 insertions(+), 87 deletions(-) diff --git a/aries_cloudagent/anoncreds/revocation.py b/aries_cloudagent/anoncreds/revocation.py index 42ad5107e6..9d98994cbd 100644 --- a/aries_cloudagent/anoncreds/revocation.py +++ b/aries_cloudagent/anoncreds/revocation.py @@ -80,6 +80,16 @@ class RevokeResult(NamedTuple): failed: Optional[Sequence[str]] = None +# Module-level (not instance-level) coordination for background backup-registry +# creation. AnonCredsRevocation is constructed fresh on every call, so anything +# needed to survive across calls -- strong references to keep asyncio from +# garbage-collecting an in-flight task, and dedup so overlapping rotations +# don't launch redundant concurrent ledger writes -- has to live at module +# scope. See plano-implementacao-for-update-e-resiliencia.md item 1.3. +_BACKGROUND_TASKS: set = set() +_PENDING_BACKUP_CREATIONS: set = set() # cred_def_ids with creation in flight + + class AnonCredsRevocation: """Revocation registry operations manager.""" @@ -745,6 +755,130 @@ async def get_or_fetch_local_tails_path(self, rev_reg_def: RevRegDef) -> str: # Registry Management + def _ensure_backup_creation_started( + self, issuer_id: str, cred_def_id: str, registry_type: str, max_cred_num: int + ) -> None: + """Start creating a backup registry for cred_def_id, if not already in flight. + + Safe to call from multiple concurrent rotation attempts for the same + cred_def_id: only the first call launches a task; the rest are no-ops + until that task finishes (successfully or not), so overlapping + rotations never launch redundant concurrent ledger writes. + """ + if cred_def_id in _PENDING_BACKUP_CREATIONS: + return + _PENDING_BACKUP_CREATIONS.add(cred_def_id) + + task = asyncio.create_task( + self._create_backup_with_retry( + issuer_id, cred_def_id, registry_type, max_cred_num + ) + ) + _BACKGROUND_TASKS.add(task) + + def _cleanup(completed_task: asyncio.Task) -> None: + _BACKGROUND_TASKS.discard(completed_task) + _PENDING_BACKUP_CREATIONS.discard(cred_def_id) + + task.add_done_callback(_cleanup) + + async def _activate_if_current_is_full( + self, cred_def_id: str, new_rev_reg_def_id: str + ) -> None: + """Promote a freshly created backup registry if the cred def is stuck. + + handle_full_registry marks the exhausted registry's state FULL (but + leaves its `active` tag untouched -- see there for why) when it finds + no backup to rotate into. Nothing else revisits that registry + afterwards: every later issuance attempt fails fast on the same + exhausted index before ever reaching rotation logic again, so the new + backup created here would otherwise sit at active=false forever, + needing a human to call the manual /rotate endpoint. This closes that + loop: once the backup finishes creating, check whether the cred def + is in that stuck state and, if so, self-promote it. + """ + async with self.profile.session() as session: + active_entries = await session.handle.fetch_all( + CATEGORY_REV_REG_DEF, + {"cred_def_id": cred_def_id, "active": json.dumps(True)}, + limit=1, + ) + + is_stuck = not active_entries or ( + active_entries[0].tags.get("state") == RevRegDefState.STATE_FULL + ) + if is_stuck: + LOGGER.info( + "Cred def %s had no usable active registry; promoting new " + "backup %s to active.", + cred_def_id, + new_rev_reg_def_id, + ) + await self.set_active_registry(new_rev_reg_def_id) + + async def _create_backup_with_retry( + self, issuer_id: str, cred_def_id: str, registry_type: str, max_cred_num: int + ) -> None: + """Create a new backup revocation registry, retrying with backoff. + + Runs as a background task so a slow or failing ledger write never + blocks the request that triggered it. Retries with a growing delay + (capped) for up to a day, since the only alternative to eventually + succeeding here is a cred def stuck until a human runs the manual + /rotate endpoint. + """ + base_delay = 10 # first retry after 10s + max_delay = 300 # never wait more than 5 minutes between attempts + max_total_duration = 24 * 3600 # give up only after 24h of trying + started_at = time.monotonic() + attempt = 0 + + while True: + attempt += 1 + try: + result = await self.create_and_register_revocation_registry_definition( + issuer_id=issuer_id, + cred_def_id=cred_def_id, + registry_type=registry_type, + tag=str(uuid4()), + max_cred_num=max_cred_num, + ) + LOGGER.info( + "Successfully created new backup registry for cred def " + "%s after %d attempt(s).", + cred_def_id, + attempt, + ) + await self._activate_if_current_is_full( + cred_def_id, result.rev_reg_def_id + ) + return + except Exception as e: + elapsed = time.monotonic() - started_at + if elapsed >= max_total_duration: + LOGGER.error( + "Giving up creating backup registry for cred def %s " + "after %d attempts over %.0fs. Manual rotation via " + "/anoncreds/revocation/active-registry/%s/rotate will " + "be required once the active registry fills up: %s", + cred_def_id, + attempt, + elapsed, + cred_def_id, + e, + ) + return + delay = min(base_delay * (2 ** (attempt - 1)), max_delay) + LOGGER.warning( + "Failed to create backup registry for cred def %s " + "(attempt %d, retrying in %ds): %s", + cred_def_id, + attempt, + delay, + e, + ) + await asyncio.sleep(delay) + async def handle_full_registry(self, rev_reg_def_id: str): """Update the registry status and start the next registry generation.""" async with self.profile.session() as session: @@ -766,11 +900,34 @@ async def handle_full_registry(self, rev_reg_def_id: str): if len(rev_reg_defs): backup_rev_reg_def_id = rev_reg_defs[0].name else: - # attempted to create and register here but fails in practical usage. - # the indexes and list do not get set properly (timing issue?) - # if max cred num = 4 for instance, will get - # Revocation status list does not have the index 4 - # in _create_credential calling Credential.create + # Nothing to rotate into right now. Mark the exhausted + # registry FULL -- its `active` tag is intentionally left + # untouched, since _create_credential only filters on + # `active`, not `state`, and clearing it here with no + # replacement ready would make every in-flight retry fail + # immediately with "No active registry" instead of the + # expected AnonCredsRevocationRegistryFullError. Setting + # `state` here is what lets _activate_if_current_is_full + # (below) later recognize this cred def as stuck once a + # new backup is ready, instead of leaving it stuck + # forever until a human runs the manual /rotate endpoint. + full_tags = active_rev_reg_def.tags + full_tags["state"] = RevRegDefState.STATE_FULL + await session.handle.replace( + CATEGORY_REV_REG_DEF, + active_rev_reg_def.name, + active_rev_reg_def.value, + full_tags, + ) + + self._ensure_backup_creation_started( + issuer_id=active_rev_reg_def.value_json["issuerId"], + cred_def_id=active_rev_reg_def.value_json["credDefId"], + registry_type=active_rev_reg_def.value_json["revocDefType"], + max_cred_num=active_rev_reg_def.value_json["value"][ + "maxCredNum" + ], + ) raise AnonCredsRevocationError( "Error handling full registry. No backup registry available." ) @@ -794,17 +951,18 @@ async def handle_full_registry(self, rev_reg_def_id: str): ) await txn.commit() - # create our next fallover/backup - backup_reg = await self.create_and_register_revocation_registry_definition( + # Launch background task to create the *next* backup. We do NOT + # await it here to avoid blocking the caller (the HTTP admin + # issue endpoint, or DIDComm return-route delivery) while the + # blockchain confirms the new registry. + self._ensure_backup_creation_started( issuer_id=active_rev_reg_def.value_json["issuerId"], cred_def_id=active_rev_reg_def.value_json["credDefId"], registry_type=active_rev_reg_def.value_json["revocDefType"], - tag=str(uuid4()), max_cred_num=active_rev_reg_def.value_json["value"]["maxCredNum"], ) LOGGER.info(f"previous rev_reg_def_id = {rev_reg_def_id}") LOGGER.info(f"current rev_reg_def_id = {backup_rev_reg_def_id}") - LOGGER.info(f"backup reg = {backup_reg}") async def decommission_registry(self, cred_def_id: str): """Decommission post-init registries and start the next registry generation.""" @@ -903,9 +1061,22 @@ async def _create_credential( credential_offer: dict, credential_request: dict, credential_values: dict, - rev_reg_def_id: Optional[str] = None, - tails_file_path: Optional[str] = None, - ) -> Tuple[str, str]: + revocable: bool = False, + ) -> Tuple[str, Optional[str], Optional[str], Optional[int]]: + """Create a credential, reserving a revocation index if revocable. + + Finding the active registry and reserving an index in it used to be + two separate connection-pool checkouts (a call to + get_or_create_active_registry, then this method's own transaction). + They're combined into one transaction here, since both need the same + CATEGORY_REV_REG_DEF row -- this halves the number of pool checkouts + per issuance attempt under concurrent load. + + Returns: + (credential_json, credential_revocation_id, rev_reg_def_id, + max_cred_num) -- the last three are None when not revocable. + + """ try: async with self.profile.session() as session: cred_def = await session.handle.fetch( @@ -937,23 +1108,35 @@ async def _create_credential( raw_values[attribute] = str(credential_value) - if rev_reg_def_id and tails_file_path: + rev_reg_def_id = None + max_cred_num = None + if revocable: try: async with self.profile.transaction() as txn: - rev_list = await txn.handle.fetch(CATEGORY_REV_LIST, rev_reg_def_id) - rev_reg_def = await txn.handle.fetch( - CATEGORY_REV_REG_DEF, rev_reg_def_id + rev_reg_defs = await txn.handle.fetch_all( + CATEGORY_REV_REG_DEF, + { + "cred_def_id": credential_definition_id, + "active": json.dumps(True), + }, + limit=1, + ) + if not rev_reg_defs: + raise AnonCredsRevocationError("No active registry") + rev_reg_def_entry = rev_reg_defs[0] + rev_reg_def_id = rev_reg_def_entry.name + + # for_update: serializes concurrent index reservations for + # this registry, across replicas (not just this process). + rev_list_entry = await txn.handle.fetch( + CATEGORY_REV_LIST, rev_reg_def_id, for_update=True ) - rev_key = await txn.handle.fetch( + rev_key_entry = await txn.handle.fetch( CATEGORY_REV_REG_DEF_PRIVATE, rev_reg_def_id ) - if not rev_list: + if not rev_list_entry: raise AnonCredsRevocationError("Revocation registry not found") - if not rev_reg_def: - raise AnonCredsRevocationError( - "Revocation registry definition not found" - ) - if not rev_key: + if not rev_key_entry: raise AnonCredsRevocationError( "Revocation registry definition private data not found" ) @@ -962,19 +1145,19 @@ async def _create_credential( # be updated because we always use ISSUANCE_BY_DEFAULT. # If something goes wrong later, the index will be skipped. # FIXME - double check issuance type in case of upgraded wallet? - rev_info = rev_list.value_json - rev_info_tags = rev_list.tags + rev_info = rev_list_entry.value_json + rev_info_tags = rev_list_entry.tags rev_reg_index = rev_info["next_index"] try: rev_reg_def = RevocationRegistryDefinition.load( - rev_reg_def.raw_value + rev_reg_def_entry.raw_value ) rev_list = RevocationStatusList.load(rev_info["rev_list"]) except AnoncredsError as err: raise AnonCredsRevocationError( "Error loading revocation registry definition" ) from err - if rev_reg_index > rev_reg_def.max_cred_num: + if rev_reg_index >= rev_reg_def.max_cred_num: raise AnonCredsRevocationRegistryFullError( "Revocation registry is full" ) @@ -991,11 +1174,12 @@ async def _create_credential( "Error updating revocation registry index" ) from err + max_cred_num = rev_reg_def.max_cred_num # rev_info["next_index"] is 1 based but getting from # rev_list is zero based... revoc = CredentialRevocationConfig( rev_reg_def, - rev_key.raw_value, + rev_key_entry.raw_value, rev_list, rev_reg_index, ) @@ -1003,7 +1187,6 @@ async def _create_credential( else: revoc = None credential_revocation_id = None - rev_list = None try: credential = await asyncio.get_event_loop().run_in_executor( @@ -1021,7 +1204,12 @@ async def _create_credential( except AnoncredsError as err: raise AnonCredsRevocationError("Error creating credential") from err - return credential.to_json(), credential_revocation_id + return ( + credential.to_json(), + credential_revocation_id, + rev_reg_def_id, + max_cred_num, + ) async def create_credential( self, @@ -1054,39 +1242,30 @@ async def create_credential( for attempt in range(max(retries, 1)): if attempt > 0: + delay = min(5 * (3 ** (attempt - 1)), 60) LOGGER.info( - "Waiting 2s before retrying credential issuance for cred def '%s'", + "Waiting %ds before retrying credential issuance for " + "cred def '%s' (attempt %d/%d)", + delay, cred_def_id, + attempt + 1, + max(retries, 1), ) - await asyncio.sleep(2) - - rev_reg_def_result = None - if revocable: - rev_reg_def_result = await self.get_or_create_active_registry( - cred_def_id - ) - if ( - rev_reg_def_result.revocation_registry_definition_state.state - != STATE_FINISHED - ): - continue - rev_reg_def_id = rev_reg_def_result.rev_reg_def_id - tails_file_path = self.get_local_tails_path( - rev_reg_def_result.rev_reg_def - ) - else: - rev_reg_def_id = None - tails_file_path = None + await asyncio.sleep(delay) try: - cred_json, cred_rev_id = await self._create_credential( + ( + cred_json, + cred_rev_id, + rev_reg_def_id, + max_cred_num, + ) = await self._create_credential( cred_def_id, schema_result.schema_value.attr_names, credential_offer, credential_request, credential_values, - rev_reg_def_id, - tails_file_path, + revocable, ) except AnonCredsRevocationRegistryFullError: # unlucky, another instance filled the registry first @@ -1095,12 +1274,18 @@ async def create_credential( # cred rev id is zero based # max cred num is one based # however, if we wait until max cred num is reached, we are too late. - if rev_reg_def_result: - if ( - rev_reg_def_result.rev_reg_def.value.max_cred_num - <= int(cred_rev_id) + 1 - ): - await self.handle_full_registry(rev_reg_def_id) + if rev_reg_def_id and max_cred_num is not None: + if max_cred_num <= int(cred_rev_id) + 1: + try: + await self.handle_full_registry(rev_reg_def_id) + except AnonCredsRevocationError as rot_err: + LOGGER.warning( + "Credential issued successfully but rotation of " + "registry %s failed (will retry via backup-creation " + "retry): %s", + rev_reg_def_id, + rot_err, + ) return cred_json, cred_rev_id, rev_reg_def_id diff --git a/aries_cloudagent/anoncreds/tests/test_revocation.py b/aries_cloudagent/anoncreds/tests/test_revocation.py index 26a75d57c2..50284f7a78 100644 --- a/aries_cloudagent/anoncreds/tests/test_revocation.py +++ b/aries_cloudagent/anoncreds/tests/test_revocation.py @@ -1,3 +1,4 @@ +import asyncio import http import json import os @@ -887,11 +888,20 @@ async def test_upload_tails_file(self): @mock.patch.object( test_module.AnonCredsRevocation, "create_and_register_revocation_registry_definition", - return_value="backup", ) async def test_handle_full_registry( self, mock_create_and_register, mock_set_active_registry, mock_handle ): + mock_create_and_register.return_value = mock.MagicMock( + rev_reg_def_id="backup", + job_id="backup-job", + ) + # backup creation is dispatched via module-level background-task + # sets; start each scenario from a clean state so this test doesn't + # depend on ordering relative to other tests. + test_module._BACKGROUND_TASKS.clear() + test_module._PENDING_BACKUP_CREATIONS.clear() + mock_handle.fetch = mock.CoroutineMock(return_value=MockRevRegDefEntry()) mock_handle.fetch_all = mock.CoroutineMock( return_value=[ @@ -902,17 +912,34 @@ async def test_handle_full_registry( mock_handle.replace = mock.CoroutineMock(return_value=None) await self.revocation.handle_full_registry("test-rev-reg-def-id") - assert mock_create_and_register.called assert mock_set_active_registry.called assert mock_handle.fetch.call_count == 2 assert mock_handle.fetch_all.called assert mock_handle.replace.called - # no backup registry available + # backup creation for the *next* rotation runs in the background; + # wait for it before asserting it happened, and confirm cleanup. + assert test_module._BACKGROUND_TASKS + await asyncio.gather(*test_module._BACKGROUND_TASKS) + assert mock_create_and_register.called + assert not test_module._BACKGROUND_TASKS + assert not test_module._PENDING_BACKUP_CREATIONS + + mock_create_and_register.reset_mock() + + # no backup registry available: still raises immediately, but also + # schedules a background attempt so a *future* call has something + # to rotate into instead of being stuck forever. mock_handle.fetch_all = mock.CoroutineMock(return_value=[]) with self.assertRaises(test_module.AnonCredsRevocationError): await self.revocation.handle_full_registry("test-rev-reg-def-id") + assert test_module._BACKGROUND_TASKS + await asyncio.gather(*test_module._BACKGROUND_TASKS) + assert mock_create_and_register.called + assert not test_module._BACKGROUND_TASKS + assert not test_module._PENDING_BACKUP_CREATIONS + @mock.patch.object(InMemoryProfileSession, "handle") async def test_decommission_registry(self, mock_handle): mock_handle.fetch_all = mock.CoroutineMock( @@ -1084,31 +1111,50 @@ async def call_test_func(): "attr1": "value1", "attr2": "value2", }, - rev_reg_def_id="test-rev-reg-def-id", - tails_file_path="tails-file-path", + revocable=True, ) - # missing rev list - mock_handle.fetch = mock.CoroutineMock( - side_effect=[MockEntry(), MockEntry(), None, MockEntry(), MockEntry()] + active_registry_entry = MockEntry( + name="test-rev-reg-def-id", raw_value=rev_reg_def.serialize() ) + + # no active registry + mock_handle.fetch_all = mock.CoroutineMock(return_value=[]) + mock_handle.fetch = mock.CoroutineMock(side_effect=[MockEntry(), MockEntry()]) with self.assertRaises(test_module.AnonCredsRevocationError): await call_test_func() - # missing rev def + + # missing rev list + mock_handle.fetch_all = mock.CoroutineMock( + return_value=[active_registry_entry] + ) mock_handle.fetch = mock.CoroutineMock( - side_effect=[MockEntry(), MockEntry(), MockEntry(), None, MockEntry()] + side_effect=[MockEntry(), MockEntry(), None, MockEntry()] ) with self.assertRaises(test_module.AnonCredsRevocationError): await call_test_func() # missing rev key + mock_handle.fetch_all = mock.CoroutineMock( + return_value=[active_registry_entry] + ) mock_handle.fetch = mock.CoroutineMock( - side_effect=[MockEntry(), MockEntry(), MockEntry(), MockEntry(), None] + side_effect=[ + MockEntry(), + MockEntry(), + MockEntry( + value_json={"rev_list": rev_list.serialize(), "next_index": 0} + ), + None, + ] ) with self.assertRaises(test_module.AnonCredsRevocationError): await call_test_func() # valid mock_handle.replace = mock.CoroutineMock(return_value=None) + mock_handle.fetch_all = mock.CoroutineMock( + return_value=[active_registry_entry] + ) mock_handle.fetch = mock.CoroutineMock( side_effect=[ MockEntry(), @@ -1119,16 +1165,19 @@ async def call_test_func(): "next_index": 0, } ), - MockEntry(raw_value=rev_reg_def.serialize()), MockEntry(), ] ) await call_test_func() assert mock_create.called assert mock_handle.replace.called - assert mock_handle.fetch.call_count == 5 + assert mock_handle.fetch.call_count == 4 + assert mock_handle.fetch_all.called # revocation registry is full + mock_handle.fetch_all = mock.CoroutineMock( + return_value=[active_registry_entry] + ) mock_handle.fetch = mock.CoroutineMock( side_effect=[ MockEntry(), @@ -1139,7 +1188,6 @@ async def call_test_func(): "next_index": 101, } ), - MockEntry(raw_value=rev_reg_def.serialize()), MockEntry(), ] ) @@ -1167,22 +1215,9 @@ async def test_create_credential(self, mock_supports_revocation): ) ) ) - self.revocation.get_or_create_active_registry = mock.CoroutineMock( - return_value=RevRegDefResult( - job_id="test-job-id", - revocation_registry_definition_state=RevRegDefState( - state=RevRegDefState.STATE_FINISHED, - revocation_registry_definition_id="active-reg-reg", - revocation_registry_definition=rev_reg_def, - ), - registration_metadata={}, - revocation_registry_definition_metadata={}, - ) - ) - # Test private funtion seperately - very large self.revocation._create_credential = mock.CoroutineMock( - return_value=({"cred": "cred"}, 98) + return_value=({"cred": "cred"}, "98", "active-reg-reg", 100) ) result = await self.revocation.create_credential( diff --git a/aries_cloudagent/protocols/issue_credential/v2_0/handlers/cred_request_handler.py b/aries_cloudagent/protocols/issue_credential/v2_0/handlers/cred_request_handler.py index 06a45fb8c6..e101cad064 100644 --- a/aries_cloudagent/protocols/issue_credential/v2_0/handlers/cred_request_handler.py +++ b/aries_cloudagent/protocols/issue_credential/v2_0/handlers/cred_request_handler.py @@ -2,6 +2,7 @@ from .....core.oob_processor import OobMessageProcessor from .....anoncreds.issuer import AnonCredsIssuerError +from .....anoncreds.revocation import AnonCredsRevocationError from .....indy.issuer import IndyIssuerError from .....ledger.error import LedgerError from .....messaging.base_handler import BaseHandler, HandlerException @@ -80,6 +81,9 @@ async def handle(self, context: RequestContext, responder: BaseResponder): # If auto_issue is enabled, respond immediately if cred_ex_record and cred_ex_record.auto_issue: + self._logger.info( + "Iniciando auto-issue para %s", cred_ex_record.cred_ex_id + ) cred_issue_message = None try: ( @@ -93,6 +97,7 @@ async def handle(self, context: RequestContext, responder: BaseResponder): except ( BaseModelError, AnonCredsIssuerError, + AnonCredsRevocationError, IndyIssuerError, LedgerError, StorageError, @@ -110,6 +115,20 @@ async def handle(self, context: RequestContext, responder: BaseResponder): ProblemReportReason.ISSUANCE_ABANDONED.value, # them: vague ) ) + except BaseException as diag_err: # DIAGNOSTICO TEMPORARIO + # BaseException (nao Exception) de proposito: e o unico jeito + # de enxergar asyncio.CancelledError/TimeoutError, que nao + # herdam de Exception a partir do Python 3.8. So loga e + # relanca -- nao muda o comportamento atual, so da + # visibilidade a uma excecao que hoje morre em silencio. + self._logger.error( + "Excecao INESPERADA (tipo %s) emitindo credencial para %s: %s", + type(diag_err).__name__, + cred_ex_record.cred_ex_id if cred_ex_record else "?", + diag_err, + exc_info=True, + ) + raise trace_event( context.settings, From 3b5c7e11bf26ac8f43c4b4cc1d4f22fbacd4857c Mon Sep 17 00:00:00 2001 From: Alisson Fantin Rodrigues Date: Thu, 16 Jul 2026 14:31:29 -0300 Subject: [PATCH 2/4] remov --ruff (lint) and fix unit test --- pyproject.toml | 1 - 1 file changed, 1 deletion(-) diff --git a/pyproject.toml b/pyproject.toml index 58133cff2d..8a5340a9e8 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -127,7 +127,6 @@ testpaths = "aries_cloudagent" addopts = """ --quiet --junitxml=./test-reports/junit.xml --cov-config .coveragerc --cov=aries_cloudagent --cov-report term --cov-report xml - --ruff """ markers = [ "anoncreds: Tests specifically relating to AnonCreds support", From c7a2af55e868f0d730eff72174763074fa59018c Mon Sep 17 00:00:00 2001 From: Alisson Fantin Rodrigues Date: Fri, 17 Jul 2026 15:29:20 -0300 Subject: [PATCH 3/4] async blockchain and lock --- .../anoncreds/default/did_besu/registry.py | 113 +++++++++++------- aries_cloudagent/anoncreds/revocation.py | 112 +++++++++++++---- .../anoncreds/tests/test_revocation.py | 16 ++- 3 files changed, 172 insertions(+), 69 deletions(-) diff --git a/aries_cloudagent/anoncreds/default/did_besu/registry.py b/aries_cloudagent/anoncreds/default/did_besu/registry.py index 9ed891c7cd..8e1854d00a 100644 --- a/aries_cloudagent/anoncreds/default/did_besu/registry.py +++ b/aries_cloudagent/anoncreds/default/did_besu/registry.py @@ -3,7 +3,7 @@ import json import logging import re -from asyncio import shield +from asyncio import Lock, shield, to_thread from typing import List, Optional, Pattern, Sequence, Tuple from base58 import alphabet @@ -128,6 +128,16 @@ def __init__(self): self.ROLE_CONTROL_ADDRESS = None self.REVOCATION_ADDRESS = None self.REVOCATION_LIST_GAS_LIMIT = 0x1FFFFFFFFFFFFF + # Serializes nonce-fetch-through-submit for this replica. get_transaction_count + # reflects only *confirmed* transactions, so two concurrent writers on this + # same account could otherwise fetch the same nonce before either transaction + # is mined. Held from the nonce fetch through send_transaction_tx returning + # (i.e. through confirmation) in register_revocation, + # register_revocation_registry_definition, and _revoc_reg_entry_with_fix -- + # not around anything else, so unrelated credential issuance/verification + # (which never touch the ledger) and reads (the resolve*().call() checks) + # are unaffected. + self._tx_lock = Lock() @property def supported_identifiers_regex(self) -> Pattern: @@ -227,16 +237,23 @@ async def get_schema(self, profile: Profile, schema_id: str) -> GetSchemaResult: return result - async def send_transaction_tx(self, call_function) -> TxReceipt: - """DEPRECATED.""" - # FIXME: remove me + def _send_transaction_tx_sync(self, call_function) -> TxReceipt: + """Synchronous body of send_transaction_tx. + + Runs off the event loop (see send_transaction_tx below) since + web3.eth.wait_for_transaction_receipt polls synchronously for up to + 120s (web3.py default, no timeout configured here) waiting for the + besu transaction to be mined -- without offloading this, that wait + blocks the entire ACA-Py process on this replica, not just the + caller, for the whole duration. + """ # Sign transaction signed_tx = self.web3.eth.account.sign_transaction( call_function, private_key=self.PKEY ) # Send transaction - # LOGGER.debug("Transaction: %s", signed_tx.rawTransaction) + # LOGGER.debug("Transaction: %s", signed_tx.rawTransaction) send_tx = self.web3.eth.send_raw_transaction(signed_tx.rawTransaction) # Wait for transaction receipt @@ -250,6 +267,11 @@ async def send_transaction_tx(self, call_function) -> TxReceipt: return tx_receipt + async def send_transaction_tx(self, call_function) -> TxReceipt: + """DEPRECATED.""" + # FIXME: remove me + return await to_thread(self._send_transaction_tx_sync, call_function) + async def register_revocation( self, revocation_id: str, issuer_id: str, credDef_id: str ) -> TxReceipt: @@ -260,21 +282,22 @@ async def register_revocation( address = self.web3.to_checksum_address(self.REVOCATION_ADDRESS) contract = self.web3.eth.contract(address=address, abi=abi) Chain_id = self.web3.eth.chain_id - nonce = self.web3.eth.get_transaction_count(self.ACCOUNT) call_function = contract.functions.createRevocation(rev_json) - - tx = call_function.build_transaction( - { - "chainId": Chain_id, - "from": self.ACCOUNT, - "nonce": nonce, - "gas": 3000000, - "gasPrice": self.web3.eth.gas_price, - } - ) - LOGGER.debug("Sending contract function %s: tuple %s", call_function.fn_name, call_function.arguments) - tx_receipt = await self.send_transaction_tx(tx) + async with self._tx_lock: + nonce = self.web3.eth.get_transaction_count(self.ACCOUNT) + tx = call_function.build_transaction( + { + "chainId": Chain_id, + "from": self.ACCOUNT, + "nonce": nonce, + "gas": 3000000, + "gasPrice": self.web3.eth.gas_price, + } + ) + + LOGGER.debug("Sending contract function %s: tuple %s", call_function.fn_name, call_function.arguments) + tx_receipt = await self.send_transaction_tx(tx) # receipt = contract.functions.createSchema(indy_schema).transact({"from": self.ACCOUNT}) LOGGER.debug("Receipt: %s", tx_receipt) @@ -546,24 +569,25 @@ async def register_revocation_registry_definition( address = self.web3.to_checksum_address(self.REVOCATION_ADDRESS) contract = self.web3.eth.contract(address=address, abi=abi) Chain_id = self.web3.eth.chain_id - nonce = self.web3.eth.get_transaction_count(self.ACCOUNT) LOGGER.debug(f"Creating rev reg: {indy_rev_reg_def}") call_function = contract.functions.createRevocationRegistry( indy_rev_reg_def ) - - tx = call_function.build_transaction( - { - "chainId": Chain_id, - "from": self.ACCOUNT, - "nonce": nonce, - "gas": 3000000, - "gasPrice": self.web3.eth.gas_price, - } - ) - LOGGER.debug("Sending contract function %s: tuple %s", call_function.fn_name, call_function.arguments) - tx_receipt = await self.send_transaction_tx(tx) + async with self._tx_lock: + nonce = self.web3.eth.get_transaction_count(self.ACCOUNT) + tx = call_function.build_transaction( + { + "chainId": Chain_id, + "from": self.ACCOUNT, + "nonce": nonce, + "gas": 3000000, + "gasPrice": self.web3.eth.gas_price, + } + ) + + LOGGER.debug("Sending contract function %s: tuple %s", call_function.fn_name, call_function.arguments) + tx_receipt = await self.send_transaction_tx(tx) # receipt = contract.functions.createSchema(indy_schema).transact({"from": self.ACCOUNT}) LOGGER.debug("Receipt: %s", tx_receipt) @@ -712,7 +736,6 @@ async def _revoc_reg_entry_with_fix( address = self.web3.to_checksum_address(self.REVOCATION_ADDRESS) contract = self.web3.eth.contract(address=address, abi=abi) Chain_id = self.web3.eth.chain_id - nonce = self.web3.eth.get_transaction_count(self.ACCOUNT) rev_entry = { "revDefId": rev_list.rev_reg_def_id, "regDefType": rev_reg_def_type, @@ -722,18 +745,22 @@ async def _revoc_reg_entry_with_fix( call_function = contract.functions.createOrUpdateEntry( rev_entry ) - tx = call_function.build_transaction( - { - "chainId": Chain_id, - "from": self.ACCOUNT, - "nonce": nonce, - "gas": int(self.REVOCATION_LIST_GAS_LIMIT), - "gasPrice": self.web3.eth.gas_price, - } - ) - LOGGER.debug("Sending contract function %s: tuple %s", call_function.fn_name, call_function.arguments) - rev_entry_res = await self.send_transaction_tx(tx) + async with self._tx_lock: + nonce = self.web3.eth.get_transaction_count(self.ACCOUNT) + tx = call_function.build_transaction( + { + "chainId": Chain_id, + "from": self.ACCOUNT, + "nonce": nonce, + "gas": int(self.REVOCATION_LIST_GAS_LIMIT), + "gasPrice": self.web3.eth.gas_price, + } + ) + + LOGGER.debug("Sending contract function %s: tuple %s", call_function.fn_name, call_function.arguments) + rev_entry_res = await self.send_transaction_tx(tx) + rev_entry_res = contract.functions.resolveEntry( rev_list.rev_reg_def_id ).call() diff --git a/aries_cloudagent/anoncreds/revocation.py b/aries_cloudagent/anoncreds/revocation.py index 9d98994cbd..4ca2d6b32a 100644 --- a/aries_cloudagent/anoncreds/revocation.py +++ b/aries_cloudagent/anoncreds/revocation.py @@ -807,14 +807,44 @@ async def _activate_if_current_is_full( is_stuck = not active_entries or ( active_entries[0].tags.get("state") == RevRegDefState.STATE_FULL ) - if is_stuck: - LOGGER.info( - "Cred def %s had no usable active registry; promoting new " - "backup %s to active.", - cred_def_id, + if not is_stuck: + return + + # Guard against promoting a registry whose revocation list was never + # created. Normally the RevRegDefFinishedEvent listener + # (DefaultRevocationSetup.on_rev_reg_def) creates it, but + # EventBus.notify() only logs subscriber exceptions -- it never + # re-raises them -- so a failure there (e.g. tails upload flaking + # under load) leaves the rev reg def looking fine here with no + # CATEGORY_REV_LIST behind it. Promoting it anyway would make every + # future issuance for this cred def fail with "Revocation registry + # not found", so create the list ourselves first if it's missing. + if not await self.get_created_revocation_list(new_rev_reg_def_id): + LOGGER.warning( + "Backup registry %s for cred def %s has no revocation list " + "yet; creating it now before promoting.", new_rev_reg_def_id, + cred_def_id, ) - await self.set_active_registry(new_rev_reg_def_id) + try: + await self.create_and_register_revocation_list(new_rev_reg_def_id) + except AnonCredsRevocationError as err: + LOGGER.error( + "Could not create revocation list for backup registry " + "%s (cred def %s); not promoting it: %s", + new_rev_reg_def_id, + cred_def_id, + err, + ) + return + + LOGGER.info( + "Cred def %s had no usable active registry; promoting new " + "backup %s to active.", + cred_def_id, + new_rev_reg_def_id, + ) + await self.set_active_registry(new_rev_reg_def_id) async def _create_backup_with_retry( self, issuer_id: str, cred_def_id: str, registry_type: str, max_cred_num: int @@ -895,22 +925,48 @@ async def handle_full_registry(self, rev_reg_def_id: str): "cred_def_id": active_rev_reg_def.value_json["credDefId"], "state": RevRegDefState.STATE_FINISHED, }, - limit=1, + limit=5, ) - if len(rev_reg_defs): - backup_rev_reg_def_id = rev_reg_defs[0].name - else: - # Nothing to rotate into right now. Mark the exhausted - # registry FULL -- its `active` tag is intentionally left - # untouched, since _create_credential only filters on - # `active`, not `state`, and clearing it here with no - # replacement ready would make every in-flight retry fail - # immediately with "No active registry" instead of the - # expected AnonCredsRevocationRegistryFullError. Setting - # `state` here is what lets _activate_if_current_is_full - # (below) later recognize this cred def as stuck once a - # new backup is ready, instead of leaving it stuck - # forever until a human runs the manual /rotate endpoint. + # Guard against promoting a backup whose revocation list was + # never created. Normally the RevRegDefFinishedEvent listener + # (DefaultRevocationSetup.on_rev_reg_def) creates it, but + # EventBus.notify() only logs subscriber exceptions -- it + # never re-raises them -- so a failure there (e.g. tails + # upload flaking under load) leaves a rev reg def looking + # like a perfectly good backup here with no CATEGORY_REV_LIST + # behind it. This is the primary rotation path (most + # rotations go through here, not through the self-heal path + # below), so an unguarded promotion here is what made every + # future issuance fail with "Revocation registry not found" + # until a human ran the manual /rotate endpoint. Checking a + # few candidates, not just the first, means one broken + # leftover backup doesn't block rotation when a good one is + # also available. + backup_rev_reg_def_id = None + for candidate in rev_reg_defs: + if await self.get_created_revocation_list(candidate.name): + backup_rev_reg_def_id = candidate.name + break + LOGGER.warning( + "Skipping backup registry %s for cred def %s: no " + "revocation list found for it.", + candidate.name, + active_rev_reg_def.value_json["credDefId"], + ) + + if not backup_rev_reg_def_id: + # Nothing usable to rotate into right now. Mark the + # exhausted registry FULL -- its `active` tag is + # intentionally left untouched, since _create_credential + # only filters on `active`, not `state`, and clearing it + # here with no replacement ready would make every + # in-flight retry fail immediately with "No active + # registry" instead of the expected + # AnonCredsRevocationRegistryFullError. Setting `state` + # here is what lets _activate_if_current_is_full (below) + # later recognize this cred def as stuck once a new + # backup is ready, instead of leaving it stuck forever + # until a human runs the manual /rotate endpoint. full_tags = active_rev_reg_def.tags full_tags["state"] = RevRegDefState.STATE_FULL await session.handle.replace( @@ -1217,7 +1273,7 @@ async def create_credential( credential_request: dict, credential_values: dict, *, - retries: int = 5, + retries: int = 8, ) -> Tuple[str, str, str]: """Create a credential. @@ -1226,7 +1282,15 @@ async def create_credential( credential_request: Credential request to create credential for credential_values: Values to go in credential revoc_reg_id: ID of the revocation registry - retries: number of times to retry credential creation + retries: number of times to retry credential creation. Default + and backoff cap are sized to comfortably outlast a full + registry rotation (up to ~240s in the worst case: two + sequential on-chain writes -- registry definition, then + revocation list -- each up to 120s, the web3.py default + for waiting on a besu transaction receipt). With the old + default (5 retries, 60s cap: ~125s total), a rotation that + was genuinely succeeding could still lose the race and fail + the caller right before finishing. Returns: A tuple of created credential and revocation id @@ -1242,7 +1306,7 @@ async def create_credential( for attempt in range(max(retries, 1)): if attempt > 0: - delay = min(5 * (3 ** (attempt - 1)), 60) + delay = min(5 * (3 ** (attempt - 1)), 90) LOGGER.info( "Waiting %ds before retrying credential issuance for " "cred def '%s' (attempt %d/%d)", diff --git a/aries_cloudagent/anoncreds/tests/test_revocation.py b/aries_cloudagent/anoncreds/tests/test_revocation.py index 50284f7a78..e5c8b3affc 100644 --- a/aries_cloudagent/anoncreds/tests/test_revocation.py +++ b/aries_cloudagent/anoncreds/tests/test_revocation.py @@ -902,7 +902,16 @@ async def test_handle_full_registry( test_module._BACKGROUND_TASKS.clear() test_module._PENDING_BACKUP_CREATIONS.clear() - mock_handle.fetch = mock.CoroutineMock(return_value=MockRevRegDefEntry()) + def fetch_side_effect(category, *args, **kwargs): + # handle_full_registry now also checks CATEGORY_REV_LIST (via + # get_created_revocation_list) before trusting a candidate + # backup -- return a real-shaped list entry for that category, + # and the rev reg def entry for everything else. + if category == test_module.CATEGORY_REV_LIST: + return MockRevListEntry() + return MockRevRegDefEntry() + + mock_handle.fetch = mock.CoroutineMock(side_effect=fetch_side_effect) mock_handle.fetch_all = mock.CoroutineMock( return_value=[ MockRevRegDefEntry(), @@ -913,7 +922,10 @@ async def test_handle_full_registry( await self.revocation.handle_full_registry("test-rev-reg-def-id") assert mock_set_active_registry.called - assert mock_handle.fetch.call_count == 2 + # 1 fetch for the active rev reg def, 1 for the CATEGORY_REV_LIST + # check on the first backup candidate (which now has a list, so the + # loop stops there), 1 to re-fetch the old active for marking FULL. + assert mock_handle.fetch.call_count == 3 assert mock_handle.fetch_all.called assert mock_handle.replace.called From 3931a0cb74c4a6b57923ff8df622513ef5938ca3 Mon Sep 17 00:00:00 2001 From: Alisson Fantin Rodrigues Date: Wed, 22 Jul 2026 13:31:53 -0300 Subject: [PATCH 4/4] add gas stimated in blockchain transaction --- .../anoncreds/default/did_besu/registry.py | 29 ++++++++++++++++--- aries_cloudagent/ledger/besu_vdr.py | 18 ++++++++---- 2 files changed, 37 insertions(+), 10 deletions(-) diff --git a/aries_cloudagent/anoncreds/default/did_besu/registry.py b/aries_cloudagent/anoncreds/default/did_besu/registry.py index 8e1854d00a..5beae93505 100644 --- a/aries_cloudagent/anoncreds/default/did_besu/registry.py +++ b/aries_cloudagent/anoncreds/default/did_besu/registry.py @@ -127,7 +127,7 @@ def __init__(self): self.VALIDATOR_CONTROL_ADDRESS = None self.ROLE_CONTROL_ADDRESS = None self.REVOCATION_ADDRESS = None - self.REVOCATION_LIST_GAS_LIMIT = 0x1FFFFFFFFFFFFF + self.REVOCATION_LIST_GAS_LIMIT = 3000000 # Serializes nonce-fetch-through-submit for this replica. get_transaction_count # reflects only *confirmed* transactions, so two concurrent writers on this # same account could otherwise fetch the same nonce before either transaction @@ -284,6 +284,13 @@ async def register_revocation( Chain_id = self.web3.eth.chain_id call_function = contract.functions.createRevocation(rev_json) + try: + estimated_gas = call_function.estimate_gas({"from": self.ACCOUNT}) + gas_limit = int(estimated_gas * 1.2) # 20% safety margin + except Exception as e: + LOGGER.warning("Failed to estimate gas: %s. Using fallback from config.", e) + gas_limit = int(self.REVOCATION_LIST_GAS_LIMIT) + async with self._tx_lock: nonce = self.web3.eth.get_transaction_count(self.ACCOUNT) tx = call_function.build_transaction( @@ -291,7 +298,7 @@ async def register_revocation( "chainId": Chain_id, "from": self.ACCOUNT, "nonce": nonce, - "gas": 3000000, + "gas": gas_limit, "gasPrice": self.web3.eth.gas_price, } ) @@ -574,6 +581,13 @@ async def register_revocation_registry_definition( indy_rev_reg_def ) + try: + estimated_gas = call_function.estimate_gas({"from": self.ACCOUNT}) + gas_limit = int(estimated_gas * 1.2) # 20% safety margin + except Exception as e: + LOGGER.warning("Failed to estimate gas: %s. Using fallback from config.", e) + gas_limit = int(self.REVOCATION_LIST_GAS_LIMIT) + async with self._tx_lock: nonce = self.web3.eth.get_transaction_count(self.ACCOUNT) tx = call_function.build_transaction( @@ -581,7 +595,7 @@ async def register_revocation_registry_definition( "chainId": Chain_id, "from": self.ACCOUNT, "nonce": nonce, - "gas": 3000000, + "gas": gas_limit, "gasPrice": self.web3.eth.gas_price, } ) @@ -746,6 +760,13 @@ async def _revoc_reg_entry_with_fix( rev_entry ) + try: + estimated_gas = call_function.estimate_gas({"from": self.ACCOUNT}) + gas_limit = int(estimated_gas * 1.2) # 20% safety margin + except Exception as e: + LOGGER.warning("Failed to estimate gas: %s. Using fallback from config.", e) + gas_limit = int(self.REVOCATION_LIST_GAS_LIMIT) + async with self._tx_lock: nonce = self.web3.eth.get_transaction_count(self.ACCOUNT) tx = call_function.build_transaction( @@ -753,7 +774,7 @@ async def _revoc_reg_entry_with_fix( "chainId": Chain_id, "from": self.ACCOUNT, "nonce": nonce, - "gas": int(self.REVOCATION_LIST_GAS_LIMIT), + "gas": gas_limit, "gasPrice": self.web3.eth.gas_price, } ) diff --git a/aries_cloudagent/ledger/besu_vdr.py b/aries_cloudagent/ledger/besu_vdr.py index 0f58cf2f68..711ed11362 100644 --- a/aries_cloudagent/ledger/besu_vdr.py +++ b/aries_cloudagent/ledger/besu_vdr.py @@ -364,7 +364,7 @@ async def send_credential_definition_anoncreds( abi=self.ledgerConfig.contractAbis[CREDENTIAL_DEFINITION_REGISTRY], ) call_function = contract.functions.createCredentialDefinition(cred_def) - tx_receipt = self._send_signed_transaction(call_function, False) + tx_receipt = self._send_signed_transaction(call_function) LOGGER.debug("Receipt: %s", tx_receipt) result = await self.fetch_credential_definition(cred_def_id) @@ -374,8 +374,8 @@ async def send_credential_definition_anoncreds( return "besu" def _send_signed_transaction( - self, contractFunction: ContractFunction, includeGasInTx: bool = True - ) -> TxReceipt: + self, contractFunction: ContractFunction + ) -> TxReceipt: nonce = self.web3.eth.get_transaction_count(self.ledgerConfig.trusteeAccount) chain_id = self.web3.eth.chain_id txParams = { @@ -384,8 +384,14 @@ def _send_signed_transaction( "nonce": nonce, "gasPrice": self.web3.eth.gas_price, } - if includeGasInTx: - txParams["gas"] = 3000000 + try: + estimated_gas = contractFunction.estimate_gas(txParams) + txParams["gas"] = int(estimated_gas * 1.2) # 20% safety margin + except Exception as e: + LOGGER.warning( + "Failed to estimate gas: %s. Using fallback gas limit.", e + ) + txParams["gas"] = 3000000 tx = contractFunction.build_transaction(txParams) # Sign transaction signed_tx = self.web3.eth.account.sign_transaction( @@ -461,7 +467,7 @@ async def update_endpoint_for_did( print(f"didDoc: {didDocObj}") try: call_function = contract.functions.updateDid(didDocObj) - tx_receipt = self._send_signed_transaction(call_function, False) + tx_receipt = self._send_signed_transaction(call_function) LOGGER.debug("Receipt: %s", tx_receipt) except Exception as e: