1111from ....core .profile import ProfileSession
1212from ....messaging .models .base import BaseModel
1313from ....messaging .util import datetime_to_str , epoch_to_str
14- from ....storage .base import BaseStorage
14+ from ....storage .base import DEFAULT_PAGE_SIZE , BaseStorage
1515from ....storage .error import StorageNotFoundError
1616from ....storage .record import StorageRecord
1717from ....storage .type import (
3333
3434LOGGER = logging .getLogger (__name__ )
3535
36+ # Number of records fetched per storage page when sweeping event records so the
37+ # entire event category is never loaded into memory at once.
38+ EVENT_SCAN_BATCH_SIZE = DEFAULT_PAGE_SIZE
39+
3640all_event_types = [
3741 RECORD_TYPE_REV_REG_DEF_CREATE_EVENT ,
3842 RECORD_TYPE_REV_REG_DEF_STORE_EVENT ,
@@ -413,6 +417,34 @@ async def delete_event(
413417 correlation_id ,
414418 )
415419
420+ async def _iter_records_paginated (
421+ self ,
422+ type_filter : str ,
423+ tag_query : dict ,
424+ batch_size : Optional [int ] = None ,
425+ ):
426+ """Yield stored records for a type/tag query, paging through storage.
427+
428+ Reads a bounded page at a time so the entire record category is never
429+ held in memory at once.
430+ """
431+ batch = batch_size or EVENT_SCAN_BATCH_SIZE
432+ offset = 0
433+ while True :
434+ page = await self .storage .find_paginated_records (
435+ type_filter = type_filter ,
436+ tag_query = tag_query ,
437+ limit = batch ,
438+ offset = offset ,
439+ )
440+ if not page :
441+ return
442+ for record in page :
443+ yield record
444+ if len (page ) < batch :
445+ return
446+ offset += batch
447+
416448 async def get_in_progress_events (
417449 self ,
418450 event_type : Optional [str ] = None ,
@@ -433,13 +465,10 @@ async def get_in_progress_events(
433465
434466 for etype in event_types_to_search :
435467 try :
436- # Search for events that are not completed
437- records = await self .storage .find_all_records (
438- type_filter = etype ,
439- tag_query = {"state" : EVENT_STATE_REQUESTED },
440- )
441-
442- for record in records :
468+ # Search for events that are not completed, paging through storage
469+ async for record in self ._iter_records_paginated (
470+ etype , {"state" : EVENT_STATE_REQUESTED }
471+ ):
443472 record_data = json .loads (record .value )
444473
445474 # Apply expiry timestamp filtering if requested
@@ -504,13 +533,10 @@ async def get_failed_events(
504533
505534 for etype in event_types_to_search :
506535 try :
507- # Search for events that failed
508- records = await self .storage .find_all_records (
509- type_filter = etype ,
510- tag_query = {"state" : EVENT_STATE_RESPONSE_FAILURE },
511- )
512-
513- for record in records :
536+ # Search for events that failed, paging through storage
537+ async for record in self ._iter_records_paginated (
538+ etype , {"state" : EVENT_STATE_RESPONSE_FAILURE }
539+ ):
514540 record_data = json .loads (record .value )
515541 failed_events .append (
516542 {
@@ -539,6 +565,69 @@ async def get_failed_events(
539565
540566 return failed_events
541567
568+ async def _cleanup_record_if_expired (
569+ self , record : StorageRecord , max_age_hours : int
570+ ) -> bool :
571+ """Delete a completed event record if it is past its cleanup time.
572+
573+ Returns:
574+ True if the record was deleted, False otherwise.
575+
576+ """
577+ try :
578+ record_data = json .loads (record .value )
579+ created_at_str = record_data .get ("created_at" )
580+ expiry_timestamp = record_data .get ("expiry_timestamp" )
581+
582+ if not created_at_str :
583+ # If no created_at timestamp, skip this record
584+ LOGGER .warning (
585+ "Event record %s missing created_at timestamp, skipping cleanup." ,
586+ record .id ,
587+ )
588+ return False
589+
590+ created_at = datetime .fromisoformat (created_at_str .replace ("Z" , "+00:00" ))
591+ current_time = datetime .now (timezone .utc )
592+
593+ # Cleanup threshold: created_at + max_age_hours
594+ cleanup_threshold = created_at .timestamp () + (max_age_hours * 3600 )
595+ current_timestamp = current_time .timestamp ()
596+
597+ # Earliest cleanup time is the later of the age threshold and expiry
598+ earliest_cleanup_time = cleanup_threshold
599+ if expiry_timestamp :
600+ earliest_cleanup_time = max (cleanup_threshold , expiry_timestamp )
601+
602+ # Only clean up once current time is past the earliest cleanup time
603+ if current_timestamp >= earliest_cleanup_time :
604+ await self .storage .delete_record (record )
605+ LOGGER .debug (
606+ "Cleaned up event record %s (created: %s, cleanup_threshold: %s, "
607+ "expiry: %s, current: %s)" ,
608+ record .id ,
609+ created_at_str ,
610+ epoch_to_str (cleanup_threshold ),
611+ epoch_to_str (expiry_timestamp ) if expiry_timestamp else "None" ,
612+ epoch_to_str (current_timestamp ),
613+ )
614+ return True
615+
616+ LOGGER .debug (
617+ "Event record %s not ready for cleanup (earliest: %s, current: %s)" ,
618+ record .id ,
619+ epoch_to_str (earliest_cleanup_time ),
620+ epoch_to_str (current_timestamp ),
621+ )
622+ return False
623+ except (ValueError , KeyError ) as e :
624+ LOGGER .warning (
625+ "Error parsing event record %s for cleanup: %s" ,
626+ record .id ,
627+ str (e ),
628+ )
629+ return False
630+
542631 async def cleanup_completed_events (
543632 self ,
544633 event_type : Optional [str ] = None ,
@@ -559,82 +648,36 @@ async def cleanup_completed_events(
559648
560649 for etype in event_types_to_search :
561650 try :
562- # Search for completed events (SUCCESS and FAILURE states)
563- success_records = await self .storage .find_all_records (
564- type_filter = etype ,
565- tag_query = {"state" : EVENT_STATE_RESPONSE_SUCCESS },
566- )
567- failure_records = await self .storage .find_all_records (
568- type_filter = etype ,
569- tag_query = {"state" : EVENT_STATE_RESPONSE_FAILURE },
570- )
571-
572- for record in success_records + failure_records :
573- # Parse record data to get timestamps
574- try :
575- record_data = json .loads (record .value )
576- created_at_str = record_data .get ("created_at" )
577- expiry_timestamp = record_data .get ("expiry_timestamp" )
578-
579- if not created_at_str :
580- # If no created_at timestamp, skip this record
581- LOGGER .warning (
582- "Event record %s missing created_at timestamp, "
583- "skipping cleanup." ,
584- record .id ,
585- )
586- continue
587-
588- # Parse created_at timestamp
589- created_at = datetime .fromisoformat (
590- created_at_str .replace ("Z" , "+00:00" )
591- )
592- current_time = datetime .now (timezone .utc )
593-
594- # Calculate cleanup threshold: created_at + max_age_hours
595- cleanup_threshold = created_at .timestamp () + (
596- max_age_hours * 3600
597- )
598- current_timestamp = current_time .timestamp ()
599-
600- # Determine the earliest time we can clean up this record
601- # Use the maximum of cleanup_threshold and expiry_timestamp
602- earliest_cleanup_time = cleanup_threshold
603- if expiry_timestamp :
604- earliest_cleanup_time = max (
605- cleanup_threshold , expiry_timestamp
606- )
607-
608- # Only clean up if current time is past the earliest cleanup time
609- if current_timestamp >= earliest_cleanup_time :
610- await self .storage .delete_record (record )
611- cleaned_up += 1
612- LOGGER .debug (
613- "Cleaned up event record %s (created: %s, "
614- "cleanup_threshold: %s, expiry: %s, current: %s)" ,
615- record .id ,
616- created_at_str ,
617- epoch_to_str (cleanup_threshold ),
618- epoch_to_str (expiry_timestamp )
619- if expiry_timestamp
620- else "None" ,
621- epoch_to_str (current_timestamp ),
622- )
623- else :
624- LOGGER .debug (
625- "Event record %s not ready for cleanup "
626- "(earliest: %s, current: %s)" ,
627- record .id ,
628- epoch_to_str (earliest_cleanup_time ),
629- epoch_to_str (current_timestamp ),
630- )
631-
632- except (ValueError , KeyError ) as e :
633- LOGGER .warning (
634- "Error parsing event record %s for cleanup: %s" ,
635- record .id ,
636- str (e ),
651+ # Page through completed events (SUCCESS then FAILURE states) so the
652+ # entire event category is never loaded into memory at once.
653+ for state in (
654+ EVENT_STATE_RESPONSE_SUCCESS ,
655+ EVENT_STATE_RESPONSE_FAILURE ,
656+ ):
657+ offset = 0
658+ while True :
659+ page = await self .storage .find_paginated_records (
660+ type_filter = etype ,
661+ tag_query = {"state" : state },
662+ limit = EVENT_SCAN_BATCH_SIZE ,
663+ offset = offset ,
637664 )
665+ if not page :
666+ break
667+
668+ deleted_in_page = 0
669+ for record in page :
670+ if await self ._cleanup_record_if_expired (
671+ record , max_age_hours
672+ ):
673+ cleaned_up += 1
674+ deleted_in_page += 1
675+
676+ if len (page ) < EVENT_SCAN_BATCH_SIZE :
677+ break
678+ # Deleted records shift later rows back, so only advance the
679+ # offset past records that were kept.
680+ offset += len (page ) - deleted_in_page
638681
639682 except Exception as e :
640683 LOGGER .warning (
0 commit comments