fix(clustering): handle race cond for new DP and new cfg (#9616)

* fix(clustering): handle race cond for new DP and new cfg
* docs(changelog): add changelog entry for #9616

Co-authored-by: Michael Martin <[email protected]>

CHANGELOG.md

@@ -109,6 +109,13 @@
 - Fixed an issue with error-handling and process cleanup in `kong start`.
   [#9337](https://github.com/Kong/kong/pull/9337)
 
+#### Hybrid Mode
+
+- Fixed a race condition that can cause configuration push events to be dropped
+  when the first data-plane connection is established with a control-plane
+  worker.
+  [#9616](https://github.com/Kong/kong/pull/9616)
+
 #### CLI
 
 - Fix slow CLI performance due to pending timer jobs

kong/clustering/control_plane.lua

@@ -35,6 +35,7 @@ local sub = string.sub
 local gsub = string.gsub
 local deflate_gzip = utils.deflate_gzip
 local isempty = require("table.isempty")
+local sleep = ngx.sleep
 
 local calculate_config_hash = require("kong.clustering.config_helper").calculate_config_hash
 
@@ -551,6 +552,11 @@ local function push_config_loop(premature, self, push_config_semaphore, delay)
     if ok then
       if isempty(self.clients) then
         ngx_log(ngx_DEBUG, _log_prefix, "skipping config push (no connected clients)")
+        sleep(1)
+        -- re-queue the task. wait until we have clients connected
+        if push_config_semaphore:count() <= 0 then
+          push_config_semaphore:post()
+        end
 
       else
         ok, err = pcall(self.push_config, self)

kong/clustering/wrpc_control_plane.lua

@@ -27,6 +27,7 @@ local ngx_time = ngx.time
 local ngx_var = ngx.var
 local timer_at = ngx.timer.at
 local isempty = require("table.isempty")
+local sleep = ngx.sleep
 
 local plugins_list_to_map = clustering_utils.plugins_list_to_map
 local deflate_gzip = utils.deflate_gzip
@@ -316,6 +317,11 @@ local function push_config_loop(premature, self, push_config_semaphore, delay)
     if ok then
       if isempty(self.clients) then
         ngx_log(ngx_DEBUG, _log_prefix, "skipping config push (no connected clients)")
+        sleep(1)
+        -- re-queue the task. wait until we have clients connected
+        if push_config_semaphore:count() <= 0 then
+          push_config_semaphore:post()
+        end
 
       else
         ok, err = pcall(self.push_config, self)