feat: fix threaded connection

Signed-off-by: Alexandre Rulleau <[email protected]>

Author: Leiyks

ext/ddtrace.c

@@ -433,10 +433,6 @@ static void dd_activate_once(void) {
 
     // must run before the first zai_hook_activate as ddtrace_telemetry_setup installs a global hook
     if (!ddtrace_disable) {
-        bool appsec_activation = false;
-        bool appsec_config = false;
-
-#ifndef _WIN32
         // Only disable sidecar sender when explicitly disabled
         bool bgs_fallback = DD_SIDECAR_TRACE_SENDER_DEFAULT && get_global_DD_TRACE_SIDECAR_TRACE_SENDER() && zai_config_memoized_entries[DDTRACE_CONFIG_DD_TRACE_SIDECAR_TRACE_SENDER].name_index == ZAI_CONFIG_ORIGIN_DEFAULT && !get_global_DD_INSTRUMENTATION_TELEMETRY_ENABLED();
         zend_string *bgs_service = NULL;
@@ -452,15 +448,15 @@ static void dd_activate_once(void) {
             }
         }
 
-        // if we're to enable appsec, we need to enable sidecar
+        // If we're to enable appsec, we need to enable sidecar
+        bool appsec_activation = false;
+        bool appsec_config = false;
         bool enable_sidecar = ddtrace_sidecar_maybe_enable_appsec(&appsec_activation, &appsec_config);
         if (!enable_sidecar) {
             enable_sidecar = get_global_DD_INSTRUMENTATION_TELEMETRY_ENABLED() || get_global_DD_TRACE_SIDECAR_TRACE_SENDER();
         }
 
-        if (enable_sidecar)
-#endif
-        {
+        if (enable_sidecar) {
             bool request_startup = PG(during_request_startup);
             PG(during_request_startup) = false;
             ddtrace_sidecar_setup(appsec_activation, appsec_config);
@@ -1550,6 +1546,10 @@ static PHP_MINIT_FUNCTION(ddtrace) {
     ddtrace_signals_minit();
 #endif
 
+    // Store master PID for threaded sidecar connectivity
+    // The actual sidecar setup happens in first RINIT via pthread_once
+    ddtrace_sidecar_minit(false, false);
+
     return SUCCESS;
 }
 
@@ -2639,19 +2639,33 @@ void dd_internal_handle_fork(void) {
     ddtrace_generate_runtime_id();
 
 #ifndef _WIN32
-    // If we're a child process and using sidecar, reuse the master's sidecar connection
-    // Don't call ddtrace_reset_sidecar() which would create a new sidecar instance
-    if (is_child_process && get_global_DD_TRACE_SIDECAR_TRACE_SENDER() && ddtrace_sidecar) {
-        // Child process reuses master's sidecar - just update instance ID
-        ddtrace_force_new_instance_id();
-        // Reconnect with fork flag
-        if (ddtrace_sidecar) {
-            ddtrace_sidecar_submit_root_span_data();
+    // Handle sidecar after fork with threaded connectivity
+    if (get_global_DD_TRACE_SIDECAR_TRACE_SENDER() && ddtrace_sidecar) {
+        if (is_child_process) {
+            // Child process connects to master's sidecar as a worker
+            ddtrace_force_new_instance_id();
+
+            // Connect to master as worker using threaded connectivity
+            if (ddtrace_sidecar_connect_worker_after_fork()) {
+                LOG(DEBUG, "Child process connected to master sidecar as worker (child_pid=%d, master_pid=%d)",
+                    current_pid, ddtrace_master_pid);
+
+                // Submit root span data with the new connection
+                if (ddtrace_sidecar) {
+                    ddtrace_sidecar_submit_root_span_data();
+                }
+            } else {
+                LOG(WARN, "Failed to connect child process as worker to master sidecar (child_pid=%d, master_pid=%d)",
+                    current_pid, ddtrace_master_pid);
+                // Fallback: reset sidecar on failure
+                ddtrace_reset_sidecar();
+            }
         }
+        // Parent process: keep existing connection, no action needed
     } else
 #endif
     {
-        // Not a child process, or not using sidecar - normal reset
+        // Not using sidecar - normal reset for non-sidecar trace sender
         ddtrace_reset_sidecar();
     }
     if (!get_DD_TRACE_FORKED_PROCESS()) {

ext/sidecar.c

@@ -152,6 +152,9 @@ static void dd_sidecar_on_reconnect(ddog_SidecarTransport *transport) {
 
 }
 
+// Forward declaration for reconnection handler
+static ddog_SidecarTransport *dd_sidecar_connection_factory(void);
+
 static ddog_SidecarTransport *dd_sidecar_connection_factory_ex(bool is_fork) {
     // Should not happen, unless the agent url is malformed
     if (!ddtrace_endpoint) {
@@ -172,13 +175,18 @@ static ddog_SidecarTransport *dd_sidecar_connection_factory_ex(bool is_fork) {
     }
 
     ddog_SidecarTransport *sidecar_transport;
-    if (!ddtrace_ffi_try("Failed connecting to the sidecar", ddog_sidecar_connect_php(&sidecar_transport, logpath, dd_zend_string_to_CharSlice(get_global_DD_TRACE_LOG_LEVEL()), get_global_DD_INSTRUMENTATION_TELEMETRY_ENABLED(), dd_sidecar_on_reconnect))) {
+    // With threaded connectivity, use connect_worker to connect to master's listener
+    if (!ddtrace_ffi_try("Failed connecting to sidecar as worker",
+                        ddog_sidecar_connect_worker((int32_t)ddtrace_master_pid, &sidecar_transport))) {
         dd_free_endpoints();
         return NULL;
     }
 
     dd_sidecar_post_connect(&sidecar_transport, is_fork, logpath);
 
+    // Set up automatic reconnection handler
+    ddtrace_sidecar_reconnect(&sidecar_transport, dd_sidecar_connection_factory);
+
     return sidecar_transport;
 }
 
@@ -210,27 +218,153 @@ bool ddtrace_sidecar_maybe_enable_appsec(bool *appsec_activation, bool *appsec_c
 #endif
 }
 
-void ddtrace_sidecar_setup(bool appsec_activation, bool appsec_config) {
-    ddtrace_set_non_resettable_sidecar_globals();
-    ddtrace_set_resettable_sidecar_globals();
+// Called in MINIT - stores master PID for threaded connectivity
+void ddtrace_sidecar_minit(bool appsec_activation, bool appsec_config) {
+    UNUSED(appsec_activation, appsec_config);
 
-    // Store master PID for later fork detection
+    // Store master PID for threaded connectivity only if not already set
+    // This ensures that after fork, child processes keep the parent's PID
+    if (ddtrace_master_pid == 0) {
+#ifdef _WIN32
+        ddtrace_master_pid = _getpid();
+#else
+        ddtrace_master_pid = getpid();
+#endif
+        LOG(DEBUG, "Sidecar MINIT: Stored master PID=%d", ddtrace_master_pid);
+    } else {
+        LOG(DEBUG, "Sidecar MINIT: Master PID already set to %d, current PID=%d",
+            ddtrace_master_pid,
+#ifdef _WIN32
+            _getpid()
+#else
+            getpid()
+#endif
+        );
+    }
+}
+
+// Setup master sidecar connection - called once on first RINIT
+static void ddtrace_sidecar_setup_master(bool appsec_activation, bool appsec_config) {
+    // Check if we're in a forked child process
+    // pthread_once state is not inherited across fork, so child processes
+    // might try to call this function. They should NOT start a new master listener.
 #ifndef _WIN32
-    ddtrace_master_pid = getpid();
+    ddtrace_pid_t current_pid = getpid();
+
+    LOG(DEBUG, "ddtrace_sidecar_setup_master called: current_pid=%d, ddtrace_master_pid=%d",
+        current_pid, ddtrace_master_pid);
+
+    bool is_child_process = (ddtrace_master_pid != 0 && current_pid != ddtrace_master_pid);
+
+    if (is_child_process) {
+        LOG(DEBUG, "Skipping master sidecar setup in child process (child_pid=%d, master_pid=%d)",
+            current_pid, ddtrace_master_pid);
+        // Child processes should only connect as workers, which is handled in dd_internal_handle_fork()
+        return;
+    }
+
+    LOG(DEBUG, "Proceeding with master sidecar setup for PID=%d", current_pid);
 #endif
 
+    ddtrace_set_non_resettable_sidecar_globals();
+    ddtrace_set_resettable_sidecar_globals();
+
     ddog_init_remote_config(get_global_DD_INSTRUMENTATION_TELEMETRY_ENABLED(), appsec_activation, appsec_config);
 
-    ddtrace_sidecar = dd_sidecar_connection_factory();
-    if (!ddtrace_sidecar) { // Something went wrong
+    // Start master listener for worker processes/threads to connect to
+    if (!ddtrace_ffi_try("Failed starting sidecar master listener", ddog_sidecar_connect_master((int32_t)ddtrace_master_pid))) {
+        LOG(WARN, "Failed to start sidecar master listener");
         if (ddtrace_endpoint) {
             dd_free_endpoints();
         }
+        return;
     }
 
+    // Master process also needs a transport - connect as first worker
+    ddog_SidecarTransport *sidecar_transport = NULL;
+    if (!ddtrace_ffi_try("Failed connecting master to sidecar", ddog_sidecar_connect_worker((int32_t)ddtrace_master_pid, &sidecar_transport))) {
+        LOG(WARN, "Failed to connect master process to sidecar");
+        if (ddtrace_endpoint) {
+            dd_free_endpoints();
+        }
+        return;
+    }
+
+    char logpath[MAXPATHLEN];
+    int error_fd = atomic_load(&ddtrace_error_log_fd);
+    if (error_fd == -1 || ddtrace_get_fd_path(error_fd, logpath) < 0) {
+        *logpath = 0;
+    }
+
+    dd_sidecar_post_connect(&sidecar_transport, false, logpath);
+    ddtrace_sidecar = sidecar_transport;
+
     if (get_global_DD_INSTRUMENTATION_TELEMETRY_ENABLED()) {
         ddtrace_telemetry_first_init();
     }
+
+    LOG(DEBUG, "Sidecar master setup complete, PID=%d", ddtrace_master_pid);
+}
+
+// Called in RINIT - handles git metadata and root span data
+void ddtrace_sidecar_rinit(void) {
+    if (get_DD_TRACE_GIT_METADATA_ENABLED()) {
+        zval git_object;
+        ZVAL_UNDEF(&git_object);
+        ddtrace_inject_git_metadata(&git_object);
+        if (Z_TYPE(git_object) == IS_OBJECT) {
+            ddtrace_git_metadata *git_metadata = (ddtrace_git_metadata *) Z_OBJ(git_object);
+            if (Z_TYPE(git_metadata->property_commit) == IS_STRING) {
+                UNUSED(ddog_Vec_Tag_push(&DDTRACE_G(active_global_tags), DDOG_CHARSLICE_C("git.commit.sha"),
+                                         dd_zend_string_to_CharSlice(Z_STR(git_metadata->property_commit))));
+            }
+            if (Z_TYPE(git_metadata->property_repository) == IS_STRING) {
+                UNUSED(ddog_Vec_Tag_push(&DDTRACE_G(active_global_tags), DDOG_CHARSLICE_C("git.repository_url"),
+                                         dd_zend_string_to_CharSlice(Z_STR(git_metadata->property_repository))));
+            }
+            OBJ_RELEASE(&git_metadata->std);
+        }
+    }
+
+    ddtrace_sidecar_submit_root_span_data_direct_defaults(&ddtrace_sidecar, NULL);
+}
+
+// Setup function called via pthread_once on first RINIT
+void ddtrace_sidecar_setup(bool appsec_activation, bool appsec_config) {
+    // This is called on first RINIT when config is available
+    // Set up the master sidecar connection with threaded connectivity
+    ddtrace_sidecar_setup_master(appsec_activation, appsec_config);
+}
+
+// Connect child process to master sidecar as worker after fork
+bool ddtrace_sidecar_connect_worker_after_fork(void) {
+    if (!ddtrace_endpoint || ddtrace_master_pid == 0) {
+        return false;
+    }
+
+    // Connect to master as worker using threaded connectivity
+    ddog_SidecarTransport *sidecar_transport = NULL;
+    if (!ddtrace_ffi_try("Failed connecting worker to sidecar after fork",
+                        ddog_sidecar_connect_worker((int32_t)ddtrace_master_pid, &sidecar_transport))) {
+        return false;
+    }
+
+    char logpath[MAXPATHLEN];
+    int error_fd = atomic_load(&ddtrace_error_log_fd);
+    if (error_fd == -1 || ddtrace_get_fd_path(error_fd, logpath) < 0) {
+        *logpath = 0;
+    }
+
+    // Post-connect with fork flag set to true
+    dd_sidecar_post_connect(&sidecar_transport, true, logpath);
+
+    // Drop old transport and use the new worker connection
+    if (ddtrace_sidecar) {
+        ddog_sidecar_transport_drop(ddtrace_sidecar);
+    }
+    ddtrace_sidecar = sidecar_transport;
+
+    return true;
 }
 
 void ddtrace_sidecar_ensure_active(void) {
@@ -562,28 +696,6 @@ void ddtrace_sidecar_activate(void) {
     } ZEND_HASH_FOREACH_END();
 }
 
-void ddtrace_sidecar_rinit(void) {
-    if (get_DD_TRACE_GIT_METADATA_ENABLED()) {
-        zval git_object;
-        ZVAL_UNDEF(&git_object);
-        ddtrace_inject_git_metadata(&git_object);
-        if (Z_TYPE(git_object) == IS_OBJECT) {
-            ddtrace_git_metadata *git_metadata = (ddtrace_git_metadata *) Z_OBJ(git_object);
-            if (Z_TYPE(git_metadata->property_commit) == IS_STRING) {
-                UNUSED(ddog_Vec_Tag_push(&DDTRACE_G(active_global_tags), DDOG_CHARSLICE_C("git.commit.sha"),
-                                         dd_zend_string_to_CharSlice(Z_STR(git_metadata->property_commit))));
-            }
-            if (Z_TYPE(git_metadata->property_repository) == IS_STRING) {
-                UNUSED(ddog_Vec_Tag_push(&DDTRACE_G(active_global_tags), DDOG_CHARSLICE_C("git.repository_url"),
-                                         dd_zend_string_to_CharSlice(Z_STR(git_metadata->property_repository))));
-            }
-            OBJ_RELEASE(&git_metadata->std);
-        }
-    }
-
-    ddtrace_sidecar_submit_root_span_data_direct_defaults(&ddtrace_sidecar, NULL);
-}
-
 void ddtrace_sidecar_rshutdown(void) {
     ddog_Vec_Tag_drop(DDTRACE_G(active_global_tags));
 }

ext/sidecar.h

@@ -30,7 +30,9 @@ struct telemetry_rc_info {
 };
 DDTRACE_PUBLIC struct telemetry_rc_info ddtrace_get_telemetry_rc_info(void);
 
+void ddtrace_sidecar_minit(bool appsec_activation, bool appsec_config);
 void ddtrace_sidecar_setup(bool appsec_activation, bool appsec_config);
+bool ddtrace_sidecar_connect_worker_after_fork(void);
 bool ddtrace_sidecar_maybe_enable_appsec(bool *appsec_activation, bool *appsec_config);
 void ddtrace_sidecar_ensure_active(void);
 void ddtrace_sidecar_finalize(bool clear_id);