feat: add TTL-based cleanup strategy for DatabaseTriggers collection (#464)

* feat: add TTL-based cleanup strategy for DatabaseTriggers collection

Implements automatic cleanup of completed/failed triggers using MongoDB TTL index
to prevent unbounded growth of the DatabaseTriggers collection.

Changes:
- Added expires_at field to DatabaseTriggers model for TTL tracking
- Created MongoDB TTL index on expires_at with expireAfterSeconds=0
- Added TRIGGER_RETENTION_DAYS setting (default: 30 days, configurable via env)
- Updated mark_as_triggered() to set expiration time on completed triggers
- Updated mark_as_failed() to set expiration time on failed triggers
- PENDING/TRIGGERING triggers remain without expiration (never cleaned up)

Implementation:
- MongoDB automatically deletes documents when expires_at timestamp is reached
- TTL runs in background every 60 seconds (MongoDB default)
- Retention period configurable via TRIGGER_RETENTION_DAYS environment variable
- Only terminal states (TRIGGERED, FAILED) are marked for cleanup

Tests:
- Added comprehensive unit tests for TTL expiration logic
- Tests verify expires_at is set correctly for both TRIGGERED and FAILED states
- Tests verify custom retention periods are respected
- All 4 new tests passing

Also bumped python-sdk version to 0.0.3b2

Resolves #433

* perf: optimize settings access and fix timezone handling for TTL

Performance optimization:
- Fetch settings once at trigger_cron() level instead of per-trigger
- Pass retention_days as parameter to mark functions
- Prevents repeated environment variable reads in high-volume scenarios
- Eliminates performance degradation when processing triggers in loops

Timezone fix (critical):
- Use timezone-aware datetime.now(timezone.utc) instead of naive datetime.now()
- Prevents shifted expirations on non-UTC hosts
- PyMongo treats naive datetimes as UTC, causing incorrect TTL windows
- All expires_at timestamps now explicitly UTC for consistency

CANCELLED trigger cleanup:
- Added mark_as_cancelled() function for future cancellation logic
- CANCELLED triggers now expire under same retention policy as TRIGGERED/FAILED
- Ensures all terminal states (TRIGGERED, FAILED, CANCELLED) are cleaned up

Test improvements:
- Refactored using pytest.mark.parametrize to reduce duplication
- Added timezone-awareness assertions to verify UTC timestamps
- Combined similar tests for all 3 terminal states
- Reduced test code from ~120 lines to ~85 lines while increasing coverage
- 6 parameterized tests passing (3 states × 2 test scenarios)

Signed-off-by: Sparsh <sparsh.raj30@gmail.com>

* docs: add docstring to mark_as_cancelled and improve test coverage

- Add comprehensive docstring to mark_as_cancelled() explaining it's reserved
  for future cancellation feature implementation
- Rename test_trigger_ttl.py to test_trigger_cron.py to match module name
- Expand parametrized tests: test all 3 mark functions with 3 retention periods
  (9 combinations) for consistent coverage
- Add 9 new tests covering all trigger_cron.py functions:
  - get_due_triggers (with/without triggers)
  - call_trigger_graph
  - create_next_triggers (success, DuplicateKeyError, other exceptions)
  - handle_trigger (success and failure paths)
  - trigger_cron orchestration
- Total: 18 comprehensive tests with timezone-aware datetime assertions

Signed-off-by: Sparsh <sparsh.raj30@gmail.com>

* Removed Cancelled state from partial index

Signed-off-by: Sparsh <sparsh.raj30@gmail.com>

* refactor: update trigger status filtering and add expiration logic

- Changed the partial filter expression in DatabaseTriggers to exclude PENDING and TRIGGERING statuses.
- Introduced an expires_at field in create_crons to set expiration time for triggers based on retention settings, ensuring proper cleanup of triggers after their designated retention period.

This enhances the management of trigger states and ensures that only relevant triggers are retained in the database.

* chore: update codespell ignore words list

- Added "nin" to the list of ignored words in .codespellignorewords.
- Ensured consistency by maintaining the existing format of the file.

* nin->in

* fix: update trigger retention hours to 720

- Changed the default value of trigger_retention_hours from 24 to 720 in settings.py to extend the retention period for triggers, allowing for better management of trigger states.

* docs: add TRIGGER_WORKERS and TRIGGER_RETENTION_HOURS to state manager setup

- Updated the state manager setup documentation to include new environment variables: TRIGGER_WORKERS (default: 1) and TRIGGER_RETENTION_HOURS (default: 720) for better configuration of trigger management.

* feat: add initialization tasks for server startup

- Introduced a new module for initialization tasks that run when the server starts.
- Implemented a task to delete old triggers from the DatabaseTriggers collection based on their status.
- Updated the main application file to call the new initialization tasks during the lifespan of the FastAPI app, ensuring proper cleanup of outdated triggers at startup.

* feat: enhance create_next_triggers with expiration logic

- Updated create_next_triggers function to accept retention_hours as a parameter.
- Introduced expires_at calculation to set expiration time for triggers based on retention settings.
- Ensured proper handling of trigger creation with expiration in the database, improving trigger management.

* docs: update MongoDB database name in state manager setup

- Changed the default value of MONGO_DATABASE_NAME from 'exosphere' to 'exosphere-state-manager' in the state manager setup documentation for clarity and accuracy in configuration.

* fix: update trigger retention hours in tests and settings

- Changed the default value of trigger_retention_hours in settings.py from 24 to 720 to extend the retention period for triggers.
- Updated test cases in test_main.py and test_trigger_cron.py to reflect the new retention_hours parameter in create_next_triggers function, ensuring consistency in trigger management across the application.

---------

Signed-off-by: Sparsh <sparsh.raj30@gmail.com>
Co-authored-by: NiveditJain <nivedit@aikin.club>

Author: spa-raj

.github/.codespellignorewords

@@ -14,4 +14,5 @@ YML
 SDK
 S3
 Kusto
-NotIn
+NotIn
+nin

docs/docs/exosphere/state-manager-setup.md

@@ -29,6 +29,8 @@ The Exosphere state manager is the core backend service that handles workflow ex
          -e MONGO_DATABASE_NAME="your-database-name" \
          -e STATE_MANAGER_SECRET="your-secret-key" \
          -e SECRETS_ENCRYPTION_KEY="your-base64-encoded-encryption-key" \
+         -e TRIGGER_WORKERS="1" \
+         -e TRIGGER_RETENTION_HOURS="720" \
          ghcr.io/exospherehost/exosphere-state-manager:latest
        ```
 
@@ -84,6 +86,8 @@ The Exosphere state manager is the core backend service that handles workflow ex
        export MONGO_DATABASE_NAME="your-database-name"
        export STATE_MANAGER_SECRET="your-secret-key"
        export SECRETS_ENCRYPTION_KEY="your-base64-encoded-encryption-key"
+       export TRIGGER_WORKERS="1"
+       export TRIGGER_RETENTION_HOURS="720"
        ```
 
     4. **Run the state manager**:
@@ -151,6 +155,8 @@ The state manager uri and key would be configured accordingly while setting up n
              - MONGO_DATABASE_NAME=${MONGO_DATABASE_NAME}
              - STATE_MANAGER_SECRET=${STATE_MANAGER_SECRET}
              - SECRETS_ENCRYPTION_KEY=${SECRETS_ENCRYPTION_KEY}
+             - TRIGGER_WORKERS=${TRIGGER_WORKERS:-1}
+             - TRIGGER_RETENTION_HOURS=${TRIGGER_RETENTION_HOURS:-720}
            deploy:
              replicas: 3
              update_config:
@@ -182,6 +188,8 @@ The state manager uri and key would be configured accordingly while setting up n
        data:
          MONGO_URI: "your-mongodb-connection-string"
          MONGO_DATABASE_NAME: "your-database-name"
+         TRIGGER_WORKERS: "1"
+         TRIGGER_RETENTION_HOURS: "720"
          LOG_LEVEL: "INFO"
        ```
 
@@ -212,9 +220,11 @@ The state manager uri and key would be configured accordingly while setting up n
 | Variable | Description | Required | Default |
 |----------|-------------|----------|---------|
 | `MONGO_URI` | MongoDB connection string | Yes | - |
-| `MONGO_DATABASE_NAME` | Database name | Yes | `exosphere` |
+| `MONGO_DATABASE_NAME` | Database name | Yes | `exosphere-state-manager` |
 | `STATE_MANAGER_SECRET` | Secret API key for authentication | Yes | - |
 | `SECRETS_ENCRYPTION_KEY` | Base64-encoded key for data encryption | Yes | - |
+| `TRIGGER_WORKERS` | Number of workers to run the trigger cron | No | `1` |
+| `TRIGGER_RETENTION_HOURS` | Number of hours to retain completed/failed triggers before cleanup | No | `720` (30 days) |
 | `LOG_LEVEL` | Logging level (DEBUG, INFO, WARNING, ERROR) | No | `INFO` |
 
 ## Monitoring and Health Checks

state-manager/app/config/settings.py

@@ -6,13 +6,14 @@
 
 class Settings(BaseModel):
     """Application settings loaded from environment variables."""
-    
+
     # MongoDB Configuration
     mongo_uri: str = Field(..., description="MongoDB connection URI" )
     mongo_database_name: str = Field(default="exosphere-state-manager", description="MongoDB database name")
     state_manager_secret: str = Field(..., description="Secret key for API authentication")
     secrets_encryption_key: str = Field(..., description="Key for encrypting secrets")
     trigger_workers: int = Field(default=1, description="Number of workers to run the trigger cron")
+    trigger_retention_hours: int = Field(default=720, description="Number of hours to retain completed/failed triggers before cleanup")
 
     @classmethod
     def from_env(cls) -> "Settings":
@@ -21,7 +22,8 @@ def from_env(cls) -> "Settings":
             mongo_database_name=os.getenv("MONGO_DATABASE_NAME", "exosphere-state-manager"), # type: ignore
             state_manager_secret=os.getenv("STATE_MANAGER_SECRET"), # type: ignore
             secrets_encryption_key=os.getenv("SECRETS_ENCRYPTION_KEY"), # type: ignore
-            trigger_workers=int(os.getenv("TRIGGER_WORKERS", 1)) # type: ignore
+            trigger_workers=int(os.getenv("TRIGGER_WORKERS", 1)), # type: ignore
+            trigger_retention_hours=int(os.getenv("TRIGGER_RETENTION_HOURS", 720)) # type: ignore
         )
 
 

state-manager/app/main.py

@@ -38,6 +38,9 @@
 from apscheduler.schedulers.asyncio import AsyncIOScheduler
 from apscheduler.triggers.cron import CronTrigger
 from .tasks.trigger_cron import trigger_cron
+
+# init tasks
+from .tasks.init_tasks import init_tasks
 
 # Define models list
 DOCUMENT_MODELS = [State, GraphTemplate, RegisteredNode, Store, Run, DatabaseTriggers]
@@ -59,6 +62,10 @@ async def lifespan(app: FastAPI):
     await init_beanie(db, document_models=DOCUMENT_MODELS)
     logger.info("beanie dbs initialized")
 
+    # performing init tasks
+    await init_tasks()
+    logger.info("init tasks completed")
+
     # initialize secret
     if not settings.state_manager_secret:
         raise ValueError("STATE_MANAGER_SECRET is not set")

state-manager/app/models/db/trigger.py

@@ -13,8 +13,9 @@ class DatabaseTriggers(Document):
     namespace: str = Field(..., description="Namespace of the graph")
     trigger_time: datetime = Field(..., description="Trigger time of the trigger")
     trigger_status: TriggerStatusEnum = Field(..., description="Status of the trigger")
+    expires_at: Optional[datetime] = Field(default=None, description="Expiration time for automatic cleanup of completed triggers")
 
-    class Settings: 
+    class Settings:
         indexes = [
             IndexModel(
                 [
@@ -32,5 +33,20 @@ class Settings:
                 ],
                 name="uniq_graph_type_expr_time",
                 unique=True
+            ),
+            IndexModel(
+                [
+                    ("expires_at", 1),
+                ],
+                name="ttl_expires_at",
+                expireAfterSeconds=0,  # Delete immediately when expires_at is reached
+                partialFilterExpression={
+                    "trigger_status": {
+                        "$in": [
+                            TriggerStatusEnum.TRIGGERED,
+                            TriggerStatusEnum.FAILED
+                        ]
+                    }
+                }
             )
         ]

state-manager/app/tasks/init_tasks.py

@@ -0,0 +1,20 @@
+# tasks to run when the server starts
+from app.models.db.trigger import DatabaseTriggers
+from app.models.trigger_models import TriggerStatusEnum
+import asyncio
+
+async def delete_old_triggers():
+    await DatabaseTriggers.get_pymongo_collection().delete_many(
+        {
+            "trigger_status": {
+                "$in": [TriggerStatusEnum.TRIGGERED, TriggerStatusEnum.FAILED]
+            },
+            "expires_at": None
+        }
+    )
+
+async def init_tasks():
+    await asyncio.gather(
+        *[
+            delete_old_triggers()
+        ])

state-manager/app/tasks/trigger_cron.py

@@ -1,4 +1,4 @@
-from datetime import datetime
+from datetime import datetime, timedelta, timezone
 from uuid import uuid4
 from app.models.db.trigger import DatabaseTriggers
 from app.models.trigger_models import TriggerStatusEnum, TriggerTypeEnum
@@ -34,18 +34,24 @@ async def call_trigger_graph(trigger: DatabaseTriggers):
         x_exosphere_request_id=str(uuid4())
     )
 
-async def mark_as_failed(trigger: DatabaseTriggers):
+async def mark_as_failed(trigger: DatabaseTriggers, retention_hours: int):
+    expires_at = datetime.now(timezone.utc) + timedelta(hours=retention_hours)
+
     await DatabaseTriggers.get_pymongo_collection().update_one(
         {"_id": trigger.id},
-        {"$set": {"trigger_status": TriggerStatusEnum.FAILED}}
+        {"$set": {
+            "trigger_status": TriggerStatusEnum.FAILED,
+            "expires_at": expires_at
+        }}
     )
 
-async def create_next_triggers(trigger: DatabaseTriggers, cron_time: datetime):
+async def create_next_triggers(trigger: DatabaseTriggers, cron_time: datetime, retention_hours: int):
     assert trigger.expression is not None
     iter = croniter.croniter(trigger.expression, trigger.trigger_time)
 
     while True:
         next_trigger_time = iter.get_next(datetime)
+        expires_at = next_trigger_time + timedelta(hours=retention_hours)
 
         try:
             await DatabaseTriggers(
@@ -54,7 +60,8 @@ async def create_next_triggers(trigger: DatabaseTriggers, cron_time: datetime):
                 graph_name=trigger.graph_name,
                 namespace=trigger.namespace,
                 trigger_time=next_trigger_time,
-                trigger_status=TriggerStatusEnum.PENDING
+                trigger_status=TriggerStatusEnum.PENDING,
+                expires_at=expires_at
             ).insert()
         except DuplicateKeyError:
             logger.error(f"Duplicate trigger found for expression {trigger.expression}")
@@ -65,24 +72,30 @@ async def create_next_triggers(trigger: DatabaseTriggers, cron_time: datetime):
         if next_trigger_time > cron_time:
             break
 
-async def mark_as_triggered(trigger: DatabaseTriggers):
+async def mark_as_triggered(trigger: DatabaseTriggers, retention_hours: int):
+    expires_at = datetime.now(timezone.utc) + timedelta(hours=retention_hours)
+
     await DatabaseTriggers.get_pymongo_collection().update_one(
         {"_id": trigger.id},
-        {"$set": {"trigger_status": TriggerStatusEnum.TRIGGERED}}
+        {"$set": {
+            "trigger_status": TriggerStatusEnum.TRIGGERED,
+            "expires_at": expires_at
+        }}
     )
 
-async def handle_trigger(cron_time: datetime):
+async def handle_trigger(cron_time: datetime, retention_hours: int):
     while(trigger:= await get_due_triggers(cron_time)):
         try:
             await call_trigger_graph(trigger)
-            await mark_as_triggered(trigger)
+            await mark_as_triggered(trigger, retention_hours)
         except Exception as e:
-            await mark_as_failed(trigger)
+            await mark_as_failed(trigger, retention_hours)
             logger.error(f"Error calling trigger graph: {e}")
         finally:
-            await create_next_triggers(trigger, cron_time)
+            await create_next_triggers(trigger, cron_time, retention_hours)
 
 async def trigger_cron():
     cron_time = datetime.now()
+    settings = get_settings()
     logger.info(f"starting trigger_cron: {cron_time}")
-    await asyncio.gather(*[handle_trigger(cron_time) for _ in range(get_settings().trigger_workers)])
+    await asyncio.gather(*[handle_trigger(cron_time, settings.trigger_retention_hours) for _ in range(settings.trigger_workers)])

state-manager/app/tasks/verify_graph.py

@@ -10,9 +10,13 @@
 from app.singletons.logs_manager import LogsManager
 from app.models.trigger_models import TriggerStatusEnum, TriggerTypeEnum
 from app.models.db.trigger import DatabaseTriggers
+from app.config.settings import get_settings
+from datetime import timedelta
 
 logger = LogsManager().get_logger()
 
+settings = get_settings()
+
 async def verify_node_exists(graph_template: GraphTemplate, registered_nodes: list[RegisteredNode]) -> list[str]:
     errors = []
     template_nodes_set = set([(node.node_name, node.namespace) for node in graph_template.nodes])
@@ -110,6 +114,7 @@ async def create_crons(graph_template: GraphTemplate):
         iter = croniter.croniter(expression, current_time)
 
         next_trigger_time = iter.get_next(datetime)
+        expires_at = next_trigger_time + timedelta(hours=settings.trigger_retention_hours)
 
         new_db_triggers.append(
             DatabaseTriggers(
@@ -118,7 +123,8 @@ async def create_crons(graph_template: GraphTemplate):
                 graph_name=graph_template.name,
                 namespace=graph_template.namespace,
                 trigger_status=TriggerStatusEnum.PENDING,
-                trigger_time=next_trigger_time
+                trigger_time=next_trigger_time,
+                expires_at=expires_at
             )
         )