transaction types

src/clepsy/aggregator_worker.py

@@ -683,7 +683,12 @@ async def do_aggregation(
             )
         )
 
-    async with get_db_connection(start_transaction=True, commit_on_exit=True) as conn:
+    logger.info(
+        "[aggregator] Starting DEFERRED transaction for persisting aggregation results"
+    )
+    async with get_db_connection(
+        start_transaction=True, commit_on_exit=True, transaction_type="DEFERRED"
+    ) as conn:
         aggregation_id = await insert_aggregation(
             aggregation=E.Aggregation(
                 start_time=aggregation_time_span.start_time,
@@ -730,6 +735,8 @@ async def do_aggregation(
                 )
             )
 
+    logger.info("[aggregator] DEFERRED transaction committed successfully")
+
     formatted_function_logs = "\n".join(
         [utils.format_function_log(x) for x in collector.logs]
     )

src/clepsy/modules/goals/calculate_goals.py

@@ -567,7 +567,13 @@ async def update_current_progress_job(goal_id: int, ttl: timedelta) -> None:
     - Check goal_progress_current.updated_at for that definition; if present and fresh (now - updated_at <= ttl), log and return.
     - Otherwise, compute current progress and upsert.
     """
-    async with get_db_connection(commit_on_exit=True, start_transaction=True) as conn:
+    logger.info(
+        "[goals] Starting DEFERRED transaction for updating current progress (goal_id={})",
+        goal_id,
+    )
+    async with get_db_connection(
+        commit_on_exit=True, start_transaction=True, transaction_type="DEFERRED"
+    ) as conn:
         # Early-exit freshness check for the latest definition's current-progress row
         latest_updated_at = await select_latest_progress_updated_at_for_goal(
             conn, goal_id=goal_id
@@ -601,6 +607,8 @@ async def update_current_progress_job(goal_id: int, ttl: timedelta) -> None:
             now_utc=now_utc,
         )
 
+    logger.info("[goals] DEFERRED transaction committed for goal_id={}", goal_id)
+
 
 async def update_previous_full_period_goal_result_job(goal_id: int) -> None:
     """Compute and insert the last completed period's goal result.
@@ -612,7 +620,13 @@ async def update_previous_full_period_goal_result_job(goal_id: int) -> None:
     logger.info(
         f"[goals] Recomputing previous full-period result for goal_id={goal_id}"
     )
-    async with get_db_connection(commit_on_exit=True, start_transaction=True) as conn:
+    logger.info(
+        "[goals] Starting DEFERRED transaction for previous period update (goal_id={})",
+        goal_id,
+    )
+    async with get_db_connection(
+        commit_on_exit=True, start_transaction=True, transaction_type="DEFERRED"
+    ) as conn:
         # Find the currently latest definition to recover base goal shape (period/tz/op)
         latest_def_id = await select_latest_goal_definition_id_for_goal(
             conn, goal_id=goal_id
@@ -714,3 +728,8 @@ async def update_previous_full_period_goal_result_job(goal_id: int) -> None:
             eval_state=full_res.eval_state,
             eval_state_reason=full_res.eval_state_reason,
         )
+
+    logger.info(
+        "[goals] DEFERRED transaction committed for previous period (goal_id={})",
+        goal_id,
+    )

src/clepsy/modules/goals/router.py

@@ -166,7 +166,9 @@ async def get_goal_row(_: Request, goal_id: int) -> Response:
 
 @router.get("/row/{goal_id}/refresh")
 async def refresh_goal_row(_: Request, goal_id: int) -> Response:
-    async with get_db_connection(commit_on_exit=True, start_transaction=True) as conn:
+    async with get_db_connection(
+        commit_on_exit=True, start_transaction=True, transaction_type="DEFERRED"
+    ) as conn:
         gwrs = await select_goals_with_latest_definition(conn, last_successes_limit=8)
         gwr = next((g for g in gwrs if g.goal.id == goal_id), None)
         if not gwr:

src/clepsy/modules/sessions/tasks.py

@@ -828,7 +828,12 @@ async def finalize_carry_over_sessions_and_save(
     )
 
     # Save sessionization run
-    async with get_db_connection(start_transaction=True) as conn:
+    logger.info(
+        "[sessionization] Starting DEFERRED transaction for saving sessionization results"
+    )
+    async with get_db_connection(
+        start_transaction=True, transaction_type="DEFERRED"
+    ) as conn:
         sessionization_run = SessionizationRun(
             candidate_creation_start=candidate_creation_interval_start,
             candidate_creation_end=candidate_creation_interval_end,
@@ -1339,7 +1344,13 @@ async def run_sessionization():
 
             case _:
                 raise ValueError("Unexpected result from deal_with_island")
-    async with get_db_connection(start_transaction=True) as conn:
+
+    logger.info(
+        "[sessionization-isolated] Starting DEFERRED transaction for saving sessionization results"
+    )
+    async with get_db_connection(
+        start_transaction=True, transaction_type="DEFERRED"
+    ) as conn:
         sessionization_run = SessionizationRun(
             candidate_creation_start=candidate_creation_interval_start,
             candidate_creation_end=candidate_creation_interval_end,
@@ -1440,3 +1451,5 @@ async def run_sessionization():
 
         # Step 6: Cleanup delete_candidate_sessions_without_activities
         await delete_candidate_sessions_without_activities(conn)
+
+    logger.info("[sessionization-isolated] DEFERRED transaction committed successfully")