Send one heartbeat on scheduling service startup anyway

Author: artem-shelkovnikov

connectors/protocol/connectors.py

@@ -734,9 +734,10 @@ def sync_cursor(self):
     def api_key_secret_id(self):
         return self.get("api_key_secret_id")
 
-    async def heartbeat(self, interval):
+    async def heartbeat(self, interval, force=False):
         if (
-            self.last_seen is None
+            force
+            or self.last_seen is None
             or (datetime.now(timezone.utc) - self.last_seen).total_seconds() > interval
         ):
             self.log_debug("Sending heartbeat")

connectors/services/job_scheduling.py

@@ -37,9 +37,14 @@ def __init__(self, config):
         self.idling = self.service_config["idling"]
         self.heartbeat_interval = self.service_config["heartbeat"]
         self.source_list = config["sources"]
+        self.first_run = True
         self.last_wake_up_time = datetime.now(timezone.utc)
 
     async def _schedule(self, connector):
+        # To do some first-time stuff
+        just_started = self.first_run
+        self.first_run = False
+
         if self.running is False:
             connector.log_debug("Skipping run because service is terminating")
             return
@@ -66,7 +71,7 @@ async def _schedule(self, connector):
             raise
 
         # the heartbeat is always triggered
-        await connector.heartbeat(self.heartbeat_interval)
+        await connector.heartbeat(self.heartbeat_interval, force=just_started)
 
         connector.log_debug(f"Status is {connector.status}")
 

tests/services/test_job_scheduling.py

@@ -438,3 +438,25 @@ def _source_klass(config):
     data_source_mock.close.assert_awaited_once()
 
     connector.error.assert_awaited_with(error)
+
+
+@pytest.mark.asyncio
+async def test_initial_loop_run_heartbeat_only_once(
+    connector_index_mock, sync_job_index_mock, set_env
+):
+    connector = mock_connector(next_sync=None)
+    connector_index_mock.supported_connectors.return_value = AsyncIterator(
+        [connector, connector, connector, connector]
+    )  # to emulate 4 runs
+    await create_and_run_service(JobSchedulingService)
+
+    connector.prepare.assert_awaited()
+    connector.heartbeat.assert_awaited()
+    assert connector.heartbeat.call_count == 4
+    # Only the first call to heartbeat should pass "True"
+    assert connector.heartbeat.call_args_list[0].kwargs["force"] is True
+    assert connector.heartbeat.call_args_list[1].kwargs["force"] is False
+    assert connector.heartbeat.call_args_list[2].kwargs["force"] is False
+    assert connector.heartbeat.call_args_list[3].kwargs["force"] is False
+    connector.update_last_sync_scheduled_at_by_job_type.assert_not_awaited()
+    sync_job_index_mock.create.assert_not_awaited()