Merge pull request openwallet-foundation#206 from nrempel/remove-old-exchange-records

Remove old exchange records

Author: andrewwhitehead

aries_cloudagent/messaging/credentials/manager.py

@@ -1,9 +1,13 @@
 """Classes to manage credentials."""
 
 import asyncio
+import datetime
 import json
 import logging
 import time
+import random
+
+from ..util import str_to_datetime, datetime_now
 
 from ...config.injection_context import InjectionContext
 from ...error import BaseError
@@ -93,16 +97,27 @@ async def prepare_send(
         source_credential_exchange = None
 
         if source_credential_exchange_id:
-
             # The cached credential exchange ID may not have an associated credential
             # request yet. Wait up to 30 seconds for that to be populated, then
             # move on and replace it as the cached credential exchange
-
             lookup_start = time.perf_counter()
             while True:
-                source_credential_exchange = await CredentialExchange.retrieve_by_id(
-                    self._context, source_credential_exchange_id
-                )
+                try:
+                    (
+                        source_credential_exchange
+                    ) = await CredentialExchange.retrieve_by_id(
+                        self._context, source_credential_exchange_id
+                    )
+                except StorageNotFoundError:
+                    # It's possible that the cached credential expired
+                    # and was deleted while we are waiting. In this case,
+                    # it is time to issue a new credential offer.
+                    self._logger.debug(
+                        "Credential exchange deleted while"
+                        + " waiting for credential request"
+                    )
+                    break
+
                 if source_credential_exchange.credential_request:
                     break
                 if lookup_start + 30 < time.perf_counter():
@@ -453,8 +468,12 @@ async def receive_credential(self, credential_message: CredentialIssue):
                 },
             )
 
+            # Copy values from parent but create new record on save (no id)
             credential_exchange_record._id = None
             credential_exchange_record.thread_id = credential_message._thread_id
+            credential_exchange_record.parent_thread_id = (
+                credential_message._thread.pthid
+            )
             credential_exchange_record.credential_id = None
             credential_exchange_record.credential = None
 
@@ -499,12 +518,6 @@ async def store_credential(
         credential_exchange_record.credential_id = credential_id
         credential_exchange_record.credential = credential
 
-        # clear unnecessary data
-        credential_exchange_record.credential_offer = None
-        credential_exchange_record.credential_request = None
-        credential_exchange_record.raw_credential = None
-        # credential_request_metadata may be reused
-
         await credential_exchange_record.save(self.context, reason="Store credential")
 
         credential_stored_message = CredentialStored()
@@ -513,6 +526,49 @@ async def store_credential(
             credential_exchange_record.parent_thread_id,
         )
 
+        # Always delete this record if it's a child
+
+        # Get parent exchange record if parent id exists
+        parent_thread_id = credential_exchange_record.parent_thread_id
+        if parent_thread_id:
+            # We delete the current record but only if it has a parent_id
+            # because we don't want to delete any new parents
+            try:
+                await credential_exchange_record.delete_record(self.context)
+            except StorageNotFoundError:
+                # It's possible for another thread to have already deleted
+                # this record
+                self._logger.debug("Failed to delete credential exchange record")
+
+        # Delete old records if they are no longer used
+
+        # Run approx every 100 runs
+        if random.randint(1, 100) == 1:
+            # Query undeleted stored exchange records for possible expired parents
+            old_credential_exchange_records = await CredentialExchange.query(
+                self.context,
+                tag_filter={
+                    "state": CredentialExchange.STATE_STORED,
+                    "initiator": CredentialExchange.INITIATOR_EXTERNAL,
+                },
+            )
+
+            for old_credential_exchange_record in old_credential_exchange_records:
+                last_updated_string = old_credential_exchange_record.updated_at
+                last_updated = str_to_datetime(last_updated_string)
+                one_hour_ago = datetime_now() - datetime.timedelta(hours=1)
+
+                # delete parent exchange records more than 1 hour old
+                if last_updated < one_hour_ago:
+                    try:
+                        await old_credential_exchange_record.delete_record(self.context)
+                    except StorageNotFoundError:
+                        # It's possible for another thread to have already deleted
+                        # this record
+                        self._logger.debug(
+                            "Failed to delete credential exchange record"
+                        )
+
         return credential_exchange_record, credential_stored_message
 
     async def credential_stored(self, credential_stored_message: CredentialStored):
@@ -524,6 +580,7 @@ async def credential_stored(self, credential_stored_message: CredentialStored):
 
         """
 
+        # Get current exchange record by thread id
         credential_exchange_record = await CredentialExchange.retrieve_by_tag_filter(
             self.context,
             tag_filter={
@@ -532,13 +589,76 @@ async def credential_stored(self, credential_stored_message: CredentialStored):
             },
         )
 
-        # clear unnecessary data
-        credential_exchange_record.credential_offer = None
-        credential_exchange_record.credential_request = None
-        credential_exchange_record.credential_request_metadata = None
-        credential_exchange_record.credential_values = None
-
         credential_exchange_record.state = CredentialExchange.STATE_STORED
         await credential_exchange_record.save(self.context, reason="Credential stored")
 
-        return credential_exchange_record
+        # Always delete this record if it's a child
+
+        # Get parent exchange record if parent id exists
+        parent_thread_id = credential_exchange_record.parent_thread_id
+        if parent_thread_id:
+            # We delete the current record but only if it has a parent_id
+            # because we don't want to delete any new parents
+            try:
+                await credential_exchange_record.delete_record(self.context)
+            except StorageNotFoundError:
+                # It's possible for another thread to have already deleted
+                # this record
+                self._logger.debug("Failed to delete credential exchange record")
+
+        # Delete old records if they are no longer used
+
+        # Run approx every 100 runs
+        if random.randint(1, 100) == 1:
+            # Query undeleted stored exchange records for possible expired parents
+            old_credential_exchange_records = await CredentialExchange.query(
+                self.context,
+                tag_filter={
+                    "state": CredentialExchange.STATE_STORED,
+                    "initiator": CredentialExchange.INITIATOR_SELF,
+                },
+            )
+
+            for old_credential_exchange_record in old_credential_exchange_records:
+                cache: BaseCache = await self._context.inject(BaseCache)
+
+                cached_credential_ex_id = await cache.get(
+                    "credential_exchange::"
+                    + f"{old_credential_exchange_record.credential_definition_id}::"
+                    + f"{old_credential_exchange_record.connection_id}"
+                )
+
+                # If this old credential is still in the cache, then it's definitely
+                # an active parent record
+                if (
+                    old_credential_exchange_record.credential_exchange_id
+                    != cached_credential_ex_id
+                ):
+                    # We check if any child threads are still relying on
+                    # information from this record. If not, we can delete.
+                    child_records = await CredentialExchange.query(
+                        self.context,
+                        tag_filter={
+                            "parent_thread_id": (
+                                old_credential_exchange_record.thread_id
+                            ),
+                            "initiator": CredentialExchange.INITIATOR_SELF,
+                        },
+                    )
+
+                    # If this credential isn't in the cache and there are no child
+                    # records which reference this as parent, we can delete
+                    if len(child_records) == 0:
+                        try:
+                            await old_credential_exchange_record.delete_record(
+                                self.context
+                            )
+                            self._logger.debug(
+                                "Parent credential exchange record successfully deleted"
+                            )
+                        except StorageNotFoundError:
+                            # It's possible for another thread to have already deleted
+                            # this record
+                            self._logger.debug(
+                                "Failed to delete parent credential exchange record"
+                            )