diff --git a/src/clepsy/aggregator_worker.py b/src/clepsy/aggregator_worker.py index c42fc11..8c9c5c8 100755 --- a/src/clepsy/aggregator_worker.py +++ b/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] ) diff --git a/src/clepsy/modules/goals/calculate_goals.py b/src/clepsy/modules/goals/calculate_goals.py index b143542..f1af621 100755 --- a/src/clepsy/modules/goals/calculate_goals.py +++ b/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, + ) diff --git a/src/clepsy/modules/goals/router.py b/src/clepsy/modules/goals/router.py index 95e2c57..955042e 100755 --- a/src/clepsy/modules/goals/router.py +++ b/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: diff --git a/src/clepsy/modules/sessions/tasks.py b/src/clepsy/modules/sessions/tasks.py index 1a44040..ed559e2 100755 --- a/src/clepsy/modules/sessions/tasks.py +++ b/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")