trying to fix concurrency issue

Author: matusdrobuliak66

services/resource-usage-tracker/src/simcore_service_resource_usage_tracker/services/background_task_periodic_heartbeat_check.py

@@ -48,16 +48,16 @@ async def _check_service_heartbeat(
         modified_at
         < base_start_timestamp - timedelta(minutes=2)
     ):
-        missed_heartbeat_counter += 1
+        new_missed_heartbeat_counter = missed_heartbeat_counter + 1
         if (
-            missed_heartbeat_counter
+            new_missed_heartbeat_counter
             > resource_usage_tracker_missed_heartbeat_counter_fail
         ):
             # Handle unhealthy service
             _logger.error(
                 "Service run id: %s is considered unhealthy and not billed. Counter %s",
                 service_run_id,
-                missed_heartbeat_counter,
+                new_missed_heartbeat_counter,
             )
             await _close_unhealthy_service(
                 db_engine, service_run_id, base_start_timestamp
@@ -66,14 +66,24 @@ async def _check_service_heartbeat(
             _logger.warning(
                 "Service run id: %s missed heartbeat. Counter %s",
                 service_run_id,
-                missed_heartbeat_counter,
+                new_missed_heartbeat_counter,
             )
-            await service_runs_db.update_service_missed_heartbeat_counter(
-                db_engine,
-                service_run_id=service_run_id,
-                last_heartbeat_at=last_heartbeat_at,
-                missed_heartbeat_counter=missed_heartbeat_counter,
+            # Use the original last_heartbeat_at and modified_at as keys for the update
+            # to ensure we're updating the correct record and prevent race conditions
+            updated_service = (
+                await service_runs_db.update_service_missed_heartbeat_counter(
+                    db_engine,
+                    service_run_id=service_run_id,
+                    last_heartbeat_at=last_heartbeat_at,
+                    missed_heartbeat_counter=new_missed_heartbeat_counter,
+                )
             )
+            # If the update returned None, it means another process already updated this record
+            if updated_service is None:
+                _logger.warning(
+                    "Service run id: %s was already updated by another process",
+                    service_run_id,
+                )
 
 
 async def _close_unhealthy_service(
@@ -164,8 +174,10 @@ async def check_running_services(app: FastAPI) -> None:
     base_start_timestamp = datetime.now(tz=UTC)
 
     # Get all current running services (across all products)
-    total_count: PositiveInt = await service_runs_db.total_service_runs_with_running_status_across_all_products(
-        _db_engine
+    total_count: PositiveInt = (
+        await service_runs_db.total_service_runs_with_running_status_across_all_products(
+            _db_engine
+        )
     )
 
     for offset in range(0, total_count, _BATCH_SIZE):