fix: wire mark_as_cancelled into trigger cancellation flow & replace destructive deletes with TTL-based cleanup

state-manager/app/config/settings.py

@@ -6,7 +6,7 @@
 
 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/app/controller/upsert_graph_template.py

@@ -1,34 +1,46 @@
+from datetime import datetime, timedelta, timezone
+
 from app.singletons.logs_manager import LogsManager
 from app.models.graph_models import UpsertGraphTemplateRequest, UpsertGraphTemplateResponse
 from app.models.db.graph_template_model import GraphTemplate
 from app.models.graph_template_validation_status import GraphTemplateValidationStatus
 from app.tasks.verify_graph import verify_graph
 from app.models.db.trigger import DatabaseTriggers
 from app.models.trigger_models import TriggerStatusEnum, TriggerTypeEnum
-from beanie.operators import In
+from app.config.settings import get_settings
 
 from fastapi import BackgroundTasks, HTTPException
 
 logger = LogsManager().get_logger()
 
-async def upsert_graph_template(namespace_name: str, graph_name: str, body: UpsertGraphTemplateRequest, x_exosphere_request_id: str, background_tasks: BackgroundTasks) -> UpsertGraphTemplateResponse:
-    try:
 
+async def upsert_graph_template(
+    namespace_name: str,
+    graph_name: str,
+    body: UpsertGraphTemplateRequest,
+    x_exosphere_request_id: str,
+    background_tasks: BackgroundTasks,
+) -> UpsertGraphTemplateResponse:
+    try:
+        # Load settings at request time so runtime config changes are picked up
+        settings = get_settings()
         old_triggers = []
 
         graph_template = await GraphTemplate.find_one(
             GraphTemplate.name == graph_name,
-            GraphTemplate.namespace == namespace_name
+            GraphTemplate.namespace == namespace_name,
         )
-               
+
         try:
             if graph_template:
                 logger.info(
-                    "Graph template already exists in namespace", graph_template=graph_template,
+                    "Graph template already exists in namespace",
+                    graph_template=graph_template,
                     namespace_name=namespace_name,
-                    x_exosphere_request_id=x_exosphere_request_id)
+                    x_exosphere_request_id=x_exosphere_request_id,
+                )
                 old_triggers = graph_template.triggers
-                
+
                 graph_template.set_secrets(body.secrets)
                 graph_template.validation_status = GraphTemplateValidationStatus.PENDING
                 graph_template.validation_errors = []
@@ -37,14 +49,15 @@ async def upsert_graph_template(namespace_name: str, graph_name: str, body: Upse
                 graph_template.nodes = body.nodes
                 graph_template.triggers = body.triggers
                 await graph_template.save()
-                
+
             else:
                 logger.info(
                     "Graph template does not exist in namespace",
                     namespace_name=namespace_name,
                     graph_name=graph_name,
-                    x_exosphere_request_id=x_exosphere_request_id)
-
+                    x_exosphere_request_id=x_exosphere_request_id,
+                )
+
                 graph_template = await GraphTemplate.insert(
                     GraphTemplate(
                         name=graph_name,
@@ -54,35 +67,75 @@ async def upsert_graph_template(namespace_name: str, graph_name: str, body: Upse
                         validation_errors=[],
                         retry_policy=body.retry_policy,
                         store_config=body.store_config,
-                        triggers=body.triggers
+                        triggers=body.triggers,
                     ).set_secrets(body.secrets)
                 )
         except ValueError as e:
-            logger.error("Error validating graph template", error=e, x_exosphere_request_id=x_exosphere_request_id)
-            raise HTTPException(status_code=400, detail=f"Error validating graph template: {str(e)}")
-
+            logger.error(
+                "Error validating graph template",
+                error=e,
+                x_exosphere_request_id=x_exosphere_request_id,
+            )
+            raise HTTPException(
+                status_code=400,
+                detail=f"Error validating graph template: {str(e)}",
+            )
+
+        # Previously:
+        # await DatabaseTriggers.find(...).delete_many()
+        #
+        # Now: bulk update to mark matching CRON triggers as CANCELLED
+        # and set expires_at so TTL can clean them up later.
         if len(old_triggers) > 0:
-            await DatabaseTriggers.find(
-                DatabaseTriggers.graph_name == graph_name,
-                DatabaseTriggers.trigger_status == TriggerStatusEnum.PENDING,
-                DatabaseTriggers.type == TriggerTypeEnum.CRON,
-                In(DatabaseTriggers.expression, [trigger.value["expression"] for trigger in old_triggers if trigger.type == TriggerTypeEnum.CRON])
-            ).delete_many()
+            cron_expressions = [
+                trigger.value["expression"]
+                for trigger in old_triggers
+                if trigger.type == TriggerTypeEnum.CRON
+            ]
+
+            if cron_expressions:
+                expires_at = datetime.now(timezone.utc) + timedelta(
+                    hours=settings.trigger_retention_hours
+                )
+
+                await DatabaseTriggers.get_pymongo_collection().update_many(
+                    {
+                        "graph_name": graph_name,
+                        "namespace": namespace_name,  # ensure we only affect this namespace
+                        "trigger_status": TriggerStatusEnum.PENDING.value,
+                        "type": TriggerTypeEnum.CRON.value,
+                        "expression": {"$in": cron_expressions},
+                    },
+                    {
+                        "$set": {
+                            "trigger_status": TriggerStatusEnum.CANCELLED.value,
+                            "expires_at": expires_at,
+                        }
+                    },
+                )
 
         background_tasks.add_task(verify_graph, graph_template)
 
         return UpsertGraphTemplateResponse(
             nodes=graph_template.nodes,
             validation_status=graph_template.validation_status,
             validation_errors=graph_template.validation_errors,
-            secrets={secret_name: True for secret_name in graph_template.get_secrets().keys()},
+            secrets={
+                secret_name: True
+                for secret_name in graph_template.get_secrets().keys()
+            },
             retry_policy=graph_template.retry_policy,
             store_config=graph_template.store_config,
             triggers=graph_template.triggers,
             created_at=graph_template.created_at,
-            updated_at=graph_template.updated_at
+            updated_at=graph_template.updated_at,
         )
-    
+
     except Exception as e:
-        logger.error("Error upserting graph template", error=e, x_exosphere_request_id=x_exosphere_request_id)
-        raise e
+        logger.error(
+            "Error upserting graph template",
+            error=e,
+            x_exosphere_request_id=x_exosphere_request_id,
+        )
+        # re-raise with original traceback preserved
+        raise

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

@@ -44,7 +44,8 @@ class Settings:
                     "trigger_status": {
                         "$in": [
                             TriggerStatusEnum.TRIGGERED,
-                            TriggerStatusEnum.FAILED
+                            TriggerStatusEnum.FAILED,
+                            TriggerStatusEnum.CANCELLED
                         ]
                     }
                 }

state-manager/app/tasks/init_tasks.py

@@ -1,20 +1,54 @@
 # tasks to run when the server starts
+from datetime import datetime, timedelta, timezone
+import asyncio
+
+from app.config.settings import get_settings
 from app.models.db.trigger import DatabaseTriggers
 from app.models.trigger_models import TriggerStatusEnum
-import asyncio
+from app.singletons.logs_manager import LogsManager
+
+logger = LogsManager().get_logger()
+
+
+async def mark_old_triggers_cancelled() -> None:
+    """
+    Migrate legacy TRIGGERED/FAILED triggers that predate TTL.
 
-async def delete_old_triggers():
-    await DatabaseTriggers.get_pymongo_collection().delete_many(
+    These documents have expires_at = None, so we mark them as CANCELLED and
+    set expires_at so the TTL index can eventually clean them up.
+    """
+    settings = get_settings()
+    retention_hours = settings.trigger_retention_hours
+    expires_at = datetime.now(timezone.utc) + timedelta(hours=retention_hours)
+
+    # Use the same filter used before by delete_many()
+    filter_query = {
+        "trigger_status": {
+            "$in": [
+                TriggerStatusEnum.TRIGGERED.value,
+                TriggerStatusEnum.FAILED.value,
+            ]
+        },
+        "expires_at": None,
+    }
+
+    logger.info(
+        "Init task marking legacy TRIGGERED/FAILED triggers as CANCELLED "
+        f"for filter={filter_query}, expires_at={expires_at.isoformat()}"
+    )
+
+    await DatabaseTriggers.get_pymongo_collection().update_many(
+        filter_query,
         {
-            "trigger_status": {
-                "$in": [TriggerStatusEnum.TRIGGERED, TriggerStatusEnum.FAILED]
-            },
-            "expires_at": None
-        }
+            "$set": {
+                "trigger_status": TriggerStatusEnum.CANCELLED.value,
+                "expires_at": expires_at,
+            }
+        },
     )
 
-async def init_tasks():
+
+async def init_tasks() -> None:
     await asyncio.gather(
-        *[
-            delete_old_triggers()
-        ])
+        mark_old_triggers_cancelled(),
+    )

state-manager/app/tasks/trigger_cron.py

@@ -13,19 +13,21 @@
 
 logger = LogsManager().get_logger()
 
+
 async def get_due_triggers(cron_time: datetime) -> DatabaseTriggers | None:
     data = await DatabaseTriggers.get_pymongo_collection().find_one_and_update(
         {
             "trigger_time": {"$lte": cron_time},
-            "trigger_status": TriggerStatusEnum.PENDING
+            "trigger_status": TriggerStatusEnum.PENDING.value
         },
         {
-            "$set": {"trigger_status": TriggerStatusEnum.TRIGGERING}
+            "$set": {"trigger_status": TriggerStatusEnum.TRIGGERING.value}
         },
         return_document=ReturnDocument.AFTER
     )
     return DatabaseTriggers(**data) if data else None
 
+
 async def call_trigger_graph(trigger: DatabaseTriggers):
     await trigger_graph(
         namespace_name=trigger.namespace,
@@ -34,17 +36,45 @@ async def call_trigger_graph(trigger: DatabaseTriggers):
         x_exosphere_request_id=str(uuid4())
     )
 
+
 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,
+            "trigger_status": TriggerStatusEnum.FAILED.value,
+            "expires_at": expires_at
+        }}
+    )
+
+
+async def mark_as_cancelled(trigger: DatabaseTriggers, retention_hours: int):
+    """
+    Mark a trigger as CANCELLED and set expires_at so MongoDB TTL will remove it
+    after `retention_hours`.
+    """
+    expires_at = datetime.now(timezone.utc) + timedelta(hours=retention_hours)
+
+    await DatabaseTriggers.get_pymongo_collection().update_one(
+        {"_id": trigger.id},
+        {"$set": {
+            "trigger_status": TriggerStatusEnum.CANCELLED.value,  # keep .value ✅
             "expires_at": expires_at
         }}
     )
 
+
+async def cancel_trigger(trigger: DatabaseTriggers, retention_hours: int):
+    """
+    Cancel a trigger using the mark_as_cancelled helper.
+
+    This is intended to be used by other modules instead of duplicating
+    the cancellation logic inline.
+    """
+    await mark_as_cancelled(trigger, retention_hours)
+
+
 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)
@@ -60,7 +90,7 @@ async def create_next_triggers(trigger: DatabaseTriggers, cron_time: datetime, r
                 graph_name=trigger.graph_name,
                 namespace=trigger.namespace,
                 trigger_time=next_trigger_time,
-                trigger_status=TriggerStatusEnum.PENDING,
+                trigger_status=TriggerStatusEnum.PENDING,  # OK because insert() converts
                 expires_at=expires_at
             ).insert()
         except DuplicateKeyError:
@@ -72,19 +102,21 @@ async def create_next_triggers(trigger: DatabaseTriggers, cron_time: datetime, r
         if next_trigger_time > cron_time:
             break
 
+
 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,
+            "trigger_status": TriggerStatusEnum.TRIGGERED.value,
             "expires_at": expires_at
         }}
     )
 
+
 async def handle_trigger(cron_time: datetime, retention_hours: int):
-    while(trigger:= await get_due_triggers(cron_time)):
+    while (trigger := await get_due_triggers(cron_time)):
         try:
             await call_trigger_graph(trigger)
             await mark_as_triggered(trigger, retention_hours)
@@ -94,8 +126,12 @@ async def handle_trigger(cron_time: datetime, retention_hours: int):
         finally:
             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, settings.trigger_retention_hours) for _ in range(settings.trigger_workers)])
+    await asyncio.gather(*[
+        handle_trigger(cron_time, settings.trigger_retention_hours)
+        for _ in range(settings.trigger_workers)
+    ])