appsec: assert app-extended-heartbeat re-emits full telemetry state

Updates libdatadog to a version that includes dependencies and integrations
in app-extended-heartbeat. The previous behaviour omitted those fields and
caused a duplicate app-integrations-change / app-dependencies-loaded on the
next regular flush.

Adds an integration test (TelemetryExtendedHeartbeatTests) that drives
phpredis and exec via custom_integrations.php and asserts:
  1. They appear in a regular app-started/app-integrations-change.
  2. The next app-extended-heartbeat re-emits configuration, dependencies,
     and integrations (with phpredis present).
  3. Loading a fresh integration afterwards produces an
     app-integrations-change containing only the new integration —
     phpredis is not re-emitted, confirming the leak is fixed.

The new test runs in its own container so DD_TELEMETRY_EXTENDED_HEARTBEAT_INTERVAL=15
applies before the worker spawns. The existing TelemetryTests no longer
sets that env var, so its 24h default no longer interferes with tests
that are not about the extended heartbeat (this addresses the intermittent
'telemetry reflects the loading of a new integration' failure caused by
the heartbeat re-queueing data into the next regular flush).

Folds two helpers into TelemetryHelpers:
  - filterMessages(...) gains a userAppOnly flag (default true) that
    skips the sidecar's own datadog-ipc-helper telemetry worker.
  - WithExtendedHeartbeat / waitForExtendedHeartbeat mirror the existing
    WithIntegrations / waitForIntegrations pattern.

custom_integrations.php gains an "fs" branch (fopen on /dev/null) so
tests can trigger the filesystem deferred-loading hook.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

Author: cataphract

Cargo.lock

@@ -2,6 +2,8 @@ package com.datadog.appsec.php
 
 import groovy.transform.Canonical
 import groovy.transform.ToString
+import groovy.transform.stc.ClosureParams
+import groovy.transform.stc.FromString
 import java.net.http.HttpRequest
 import java.net.http.HttpResponse
 import java.util.Base64
@@ -13,12 +15,31 @@ import static java.net.http.HttpResponse.BodyHandlers.ofString
  * @link https://github.com/DataDog/instrumentation-telemetry-api-docs/blob/main/GeneratedDocumentation/ApiDocs/v2/producing-telemetry.md
  */
 class TelemetryHelpers {
-   static <T> List<T> filterMessages(List<Map> telemetryData, Class<T> type) {
-       (telemetryData.findAll { it['request_type'] in type.names } +
-                telemetryData.findAll { it['request_type'] == 'message-batch'}
-                        *.payload*.findAll { it['request_type'] in type.names }.flatten())
-        .collect { type.newInstance([it['payload']] as Object[]) }
-   }
+    /**
+     * Filters drained telemetry to messages of the given wrapper type, unwrapping any
+     * {@code message-batch} envelopes.
+     *
+     * <p>The sidecar runs two telemetry workers per session: one for the host app and
+     * one for {@code service_name == "datadog-ipc-helper"} (its own self-telemetry).
+     * Tests almost always want only the former; pass {@code userAppOnly = false} to
+     * include the sidecar's own telemetry too.
+     */
+    static <T> List<T> filterMessages(List<Map> telemetryData, Class<T> type, boolean userAppOnly = true) {
+        List<Map> payloads = []
+        for (msg in telemetryData) {
+            if (userAppOnly && msg.application?.service_name == 'datadog-ipc-helper') continue
+            if (msg.request_type in type.names) {
+                payloads << (msg.payload as Map)
+            } else if (msg.request_type == 'message-batch') {
+                for (inner in (msg.payload as List)) {
+                    if (inner.request_type in type.names) {
+                        payloads << (inner.payload as Map)
+                    }
+                }
+            }
+        }
+        payloads.collect { type.newInstance([it] as Object[]) }
+    }
 
     static class GenerateMetrics {
         static names = ['generate-metrics']
@@ -361,6 +382,19 @@ class TelemetryHelpers {
         }
     }
 
+    static class WithExtendedHeartbeat {
+        static names = ['app-extended-heartbeat']
+        List<ConfigurationEntry> configuration
+        List<DependencyEntry> dependencies
+        List<IntegrationEntry> integrations
+
+        WithExtendedHeartbeat(Map m) {
+            configuration = m.configuration?.collect { new ConfigurationEntry(it) }
+            dependencies  = m.dependencies?.collect { new DependencyEntry(it) }
+            integrations  = m.integrations?.collect { new IntegrationEntry(it) }
+        }
+    }
+
     static class ConfigurationEntry {
         String name
         String value
@@ -399,7 +433,10 @@ class TelemetryHelpers {
         }
     }
 
-    public static <T> List<T> waitForTelemetryData(AppSecContainer container, int timeoutSec, Closure<Boolean> cl, Class<T> cls, String path = '/hello.php') {
+    static <T> List<T> waitForTelemetryData(AppSecContainer container, int timeoutSec,
+                                            @ClosureParams(value = FromString, options = 'java.util.List<T>')
+                                                    Closure<Boolean> cl,
+                                            Class<T> cls, String path = '/hello.php') {
         List<T> messages = []
         def deadline = System.currentTimeSeconds() + timeoutSec
         def lastHttpReq = System.currentTimeSeconds() - 6
@@ -422,23 +459,46 @@ class TelemetryHelpers {
         messages
     }
 
-    public static List<AppEndpoints> waitForAppEndpoints(AppSecContainer container, int timeoutSec, Closure<Boolean> cl, String path = '/') {
+    static List<AppEndpoints> waitForAppEndpoints(AppSecContainer container, int timeoutSec,
+                                                  @ClosureParams(value = FromString,
+                                                          options = 'java.util.List<com.datadog.appsec.php.TelemetryHelpers.AppEndpoints>')
+                                                          Closure<Boolean> cl,
+                                                  String path = '/') {
         waitForTelemetryData(container, timeoutSec, cl, AppEndpoints, path)
     }
 
-    public static List<GenerateDistributions> waitForDistributions(AppSecContainer container, int timeoutSec, Closure<Boolean> cl) {
+    static List<GenerateDistributions> waitForDistributions(AppSecContainer container, int timeoutSec,
+                                                            @ClosureParams(value = FromString,
+                                                                    options = 'java.util.List<com.datadog.appsec.php.TelemetryHelpers.GenerateDistributions>')
+                                                                    Closure<Boolean> cl) {
         waitForTelemetryData(container, timeoutSec, cl, GenerateDistributions)
     }
 
-    public static List<GenerateMetrics> waitForMetrics(AppSecContainer container, int timeoutSec, Closure<Boolean> cl) {
+    static List<GenerateMetrics> waitForMetrics(AppSecContainer container, int timeoutSec,
+                                                @ClosureParams(value = FromString,
+                                                        options = 'java.util.List<com.datadog.appsec.php.TelemetryHelpers.GenerateMetrics>')
+                                                        Closure<Boolean> cl) {
         waitForTelemetryData(container, timeoutSec, cl, GenerateMetrics)
     }
 
-    public static List<WithIntegrations> waitForIntegrations(AppSecContainer container, int timeoutSec, Closure<Boolean> cl) {
+    static List<WithIntegrations> waitForIntegrations(AppSecContainer container, int timeoutSec,
+                                                      @ClosureParams(value = FromString,
+                                                              options = 'java.util.List<com.datadog.appsec.php.TelemetryHelpers.WithIntegrations>')
+                                                              Closure<Boolean> cl) {
         waitForTelemetryData(container, timeoutSec, cl, WithIntegrations)
     }
 
-    public static List<Logs> waitForLogs(AppSecContainer container, int timeoutSec, Closure<Boolean> cl) {
+    static List<WithExtendedHeartbeat> waitForExtendedHeartbeat(AppSecContainer container, int timeoutSec,
+                                                                @ClosureParams(value = FromString,
+                                                                        options = 'java.util.List<com.datadog.appsec.php.TelemetryHelpers.WithExtendedHeartbeat>')
+                                                                        Closure<Boolean> cl) {
+        waitForTelemetryData(container, timeoutSec, cl, WithExtendedHeartbeat)
+    }
+
+    static List<Logs> waitForLogs(AppSecContainer container, int timeoutSec,
+                                  @ClosureParams(value = FromString,
+                                          options = 'java.util.List<com.datadog.appsec.php.TelemetryHelpers.Logs>')
+                                          Closure<Boolean> cl) {
         waitForTelemetryData(container, timeoutSec, cl, Logs)
     }
 }

appsec/tests/integration/src/main/groovy/com/datadog/appsec/php/TelemetryHelpers.groovy

@@ -96,7 +96,6 @@ class AppSecContainer<SELF extends AppSecContainer<SELF>> extends GenericContain
         withEnv '_DD_DEBUG_SIDECAR_LOG_METHOD', 'file:///tmp/logs/sidecar.log'
         withEnv 'DD_SPAWN_WORKER_USE_EXEC', '1' // gdb fails following child with fdexec
         withEnv 'DD_TELEMETRY_HEARTBEAT_INTERVAL', '10'
-        withEnv 'DD_TELEMETRY_EXTENDED_HEARTBEAT_INTERVAL', '10'
         // withEnv '_DD_SHARED_LIB_DEBUG', '1'
         if (System.getProperty('XDEBUG') == '1') {
             Testcontainers.exposeHostPorts(9003)

appsec/tests/integration/src/main/groovy/com/datadog/appsec/php/docker/AppSecContainer.groovy

@@ -0,0 +1,102 @@
+package com.datadog.appsec.php.integration
+
+import com.datadog.appsec.php.TelemetryHelpers
+import com.datadog.appsec.php.docker.AppSecContainer
+import com.datadog.appsec.php.docker.FailOnUnmatchedTraces
+import groovy.util.logging.Slf4j
+import org.junit.jupiter.api.Test
+import org.junit.jupiter.api.condition.DisabledIf
+import org.testcontainers.junit.jupiter.Container
+import org.testcontainers.junit.jupiter.Testcontainers
+
+import java.net.http.HttpResponse
+
+import static com.datadog.appsec.php.integration.TestParams.getPhpVersion
+import static com.datadog.appsec.php.integration.TestParams.getVariant
+
+/**
+ * Standalone test class for the extended-heartbeat re-emission behaviour.
+ *
+ * The default extended-heartbeat interval is 24h. To exercise it within the test
+ * timeout, this class starts its container with DD_TELEMETRY_EXTENDED_HEARTBEAT_INTERVAL=15
+ * (kept separate from TelemetryTests so the persistent telemetry worker for that
+ * class isn't already running with the long interval before this test runs).
+ *
+ * Per the spec (instrumentation-telemetry-api-docs/Source/ApiDocs/v2/openapi.yaml
+ * `AppExtendedHeartbeat`), the payload must carry configuration + dependencies +
+ * integrations so the agent can reconstruct application records on data loss.
+ */
+@Testcontainers
+@Slf4j
+@DisabledIf('isDisabled')
+class TelemetryExtendedHeartbeatTests {
+    static boolean disabled = variant.contains('zts') || phpVersion != '8.4'
+
+    @Container
+    @FailOnUnmatchedTraces
+    public static final AppSecContainer CONTAINER =
+            new AppSecContainer(
+                    workVolume: this.name,
+                    baseTag: 'apache2-fpm-php',
+                    phpVersion: phpVersion,
+                    phpVariant: variant,
+                    www: 'base',
+            ) {
+                @Override
+                void configure() {
+                    super.configure()
+                    // Short interval so the extended heartbeat fires within the test.
+                    withEnv 'DD_TELEMETRY_EXTENDED_HEARTBEAT_INTERVAL', '15'
+                }
+            }
+
+    @Test
+    void 'extended heartbeat re-emits configuration, dependencies and integrations'() {
+        // Trigger the integration loaders for phpredis (Redis class instantiation) and
+        // exec (system() call). These cause the tracer to push AddIntegration to the
+        // sidecar.
+        CONTAINER.traceFromRequest('/custom_integrations.php?integrations[]=redis&integrations[]=exec') {
+            HttpResponse<InputStream> resp -> assert resp.statusCode() == 200
+        }
+
+        // Stage 1: verify the regular flush path actually emits phpredis (and exec)
+        // in an app-started or app-integrations-change message. This proves we're
+        // about to test the heartbeat re-emitting *previously-flushed* data.
+        Set<String> flushed = [] as Set
+        TelemetryHelpers.waitForIntegrations(CONTAINER, 30) { msgs ->
+            msgs.each { (it.integrations ?: []).each { flushed << it.name } }
+            'phpredis' in flushed && 'exec' in flushed
+        }
+        assert 'phpredis' in flushed : "phpredis not emitted via app-started/app-integrations-change; saw: ${flushed}"
+        assert 'exec' in flushed : "exec not emitted via app-started/app-integrations-change; saw: ${flushed}"
+
+        // Stage 2: the extended heartbeat must re-emit the full triple
+        // (configuration + dependencies + integrations).
+        CONTAINER.drainTelemetry(0)
+        def hbs = TelemetryHelpers.waitForExtendedHeartbeat(CONTAINER, 35) { !it.empty }
+        assert !hbs.empty : 'No app-extended-heartbeat received within 35s'
+        def hb = hbs.first()
+        assert hb.configuration != null : 'heartbeat must include configuration field'
+        assert hb.dependencies  != null : 'heartbeat must include dependencies field'
+        assert hb.integrations  != null : 'heartbeat must include integrations field'
+        assert 'phpredis' in hb.integrations*.name :
+                "heartbeat should re-emit phpredis; got: ${hb.integrations*.name}"
+
+        // Stage 3: regression check for the leak that the libdd_telemetry fix addresses.
+        // After the heartbeat, the worker should pop the re-queued items out of unflushed.
+        // So when a NEW integration is loaded next, the following regular app-integrations-change
+        // must contain only the new integration, NOT the previously-flushed-and-re-queued phpredis.
+        CONTAINER.drainTelemetry(0)
+        CONTAINER.traceFromRequest('/custom_integrations.php?integrations[]=fs') {
+            HttpResponse<InputStream> resp -> assert resp.statusCode() == 200
+        }
+        Set<String> followup = [] as Set
+        TelemetryHelpers.waitForIntegrations(CONTAINER, 30) { msgs ->
+            msgs.each { (it.integrations ?: []).each { followup << it.name } }
+            'filesystem' in followup
+        }
+        assert 'filesystem' in followup : 'No app-integrations-change with filesystem within 30s after heartbeat'
+        assert !('phpredis' in followup) :
+                "phpredis must not be re-emitted after the heartbeat; got: ${followup}"
+    }
+}

appsec/tests/integration/src/test/groovy/com/datadog/appsec/php/integration/TelemetryExtendedHeartbeatTests.groovy

@@ -18,6 +18,10 @@ public function __construct()
         system('true');
     } elseif ($int === 'redis') {
         new Redis();
+    } elseif ($int === 'fs') {
+        // Triggers the filesystem integration via deferred-loaded fopen() hook.
+        $fp = fopen('/dev/null', 'r');
+        if ($fp) { fclose($fp); }
     } else {
         http_response_code(400);
     }