The consumer close() that stranded its client, and a main that could not pass its own gate

CHANGELOG.md

@@ -4,6 +4,59 @@ All notable changes to LaraFly are documented here. This project uses CalVer (`Y
 
 ## [Unreleased]
 
+## [26.09.8] - 2026-09-25
+
+### Fixed
+
+- **`RdKafkaConsumerClient::close()` stranded its librdkafka client, and the release before this one
+  shipped a version number that disagreed with itself in three places.** Two unrelated repairs in one
+  release, because the second was found while verifying the first.
+
+  `close()` called `KafkaConsumer::close()` and then released the property. Releasing the property is
+  exactly right and was always there; it did nothing, because by the time it ran the handle was
+  already gone. ext-rdkafka 6.0.5, `kafka_consumer.c:531-542`, is the whole of that method:
+
+  ```
+  rd_kafka_consumer_close(intern->rk);
+  intern->rk = NULL;
+  ```
+
+  There is no `rd_kafka_destroy()`, and the free handler at `kafka_consumer.c:53-64` destroys the
+  handle only `if (intern->rk)` — which `close()` has just nulled. So the PHP object is freed and the
+  `rd_kafka_t` is not. Measured on PHP 8.5.8 / ext-rdkafka 6.0.5 / librdkafka 2.15.1, no broker:
+  **four OS threads stranded per closed consumer**, still there after five seconds of polling; zero
+  when the reference is simply dropped. Unconditional, not a race.
+
+  It now calls `unsubscribe()` — which is what leaves the consumer group, is safe on a consumer that
+  never subscribed, and lets the drop destroy the client (librdkafka logs `Destroying cgrp`). Pinned
+  by `RdKafkaConsumerClientLeavesNoThreadsTest`, which counts `rd_kafka_thread_cnt()` rather than
+  asserting a `WeakReference` goes null: **the natural test is green on this bug**, because the PHP
+  object really is freed and the leak is underneath it, in C.
+
+  A consumer of this framework found it the hard way. In dworkers the same call had been added
+  deliberately — to three integration tests and a long-running console command — to stop a segfault
+  at PHP shutdown after a green suite, under a comment explaining why it was necessary. It was the
+  cause. Anything that closes a consumer per restart accumulates dead clients and their threads for
+  as long as the process lives, which matters most where this class is used: a daemon.
+
+### Fixed — the release process itself
+
+- **The version was carried in four places and 26.09.6 moved one of them.** `Version::VERSION` said
+  `26.09.6`, the CHANGELOG's latest heading said `26.09.7`, the README badge said `26.09.5` and the
+  verbatim listing in `docs/versioning.md` said `26.09.5` — so the code inside the tag `v26.09.7`
+  reported itself as `26.09.6`. `VersionConsistencyTest` and `DocsCodeIsRealTest` had been red on
+  `main` since `26.09.6`, which is why CI was failing on all three PHP versions. All four now carry
+  `26.09.8`.
+
+- **`pint --test` was red on `main`** in three files, from the `26.09.7` commit: an import added out
+  of alphabetical order in `SecurityAutoConfiguration`, and the two security cache manifests
+  recompiled beside it.
+
+- **`phpstan` had three errors** in `InMemoryJwksProviderTest`, all from one missing
+  `@return array<string, mixed>` on its fixture helper.
+
+  Together these are why `composer check` could not pass on `main`. It passes now.
+
 ## [26.09.7] - 2026-09-25
 
 ### Fixed

README.md

@@ -14,7 +14,7 @@
   <a href="docs/installation.md#requirements"><img src="https://img.shields.io/badge/php-8.3%2B-blue?logo=php&logoColor=white" alt="PHP 8.3+"></a>
   <a href="docs/laravel-comparison.md"><img src="https://img.shields.io/badge/Laravel-13-FF2D20?logo=laravel&logoColor=white" alt="Laravel 13"></a>
   <a href="LICENSE"><img src="https://img.shields.io/badge/license-Apache%202.0-green" alt="License: Apache 2.0"></a>
-  <a href="CHANGELOG.md"><img src="https://img.shields.io/badge/version-26.09.5-brightgreen" alt="Version: 26.09.5"></a>
+  <a href="CHANGELOG.md"><img src="https://img.shields.io/badge/version-26.09.8-brightgreen" alt="Version: 26.09.5"></a>
   <a href="docs/contributing.md#conventions"><img src="https://img.shields.io/badge/PHPStan-max-8A2BE2" alt="PHPStan: max"></a>
   <a href="pint.json"><img src="https://img.shields.io/badge/code%20style-Pint-F55247" alt="Code Style: Pint"></a>
 </p>

docs/versioning.md

@@ -21,7 +21,7 @@ The single place the current version *is* asserted in code is:
 ```php
 final class Version
 {
-    public const string VERSION = '26.09.5';
+    public const string VERSION = '26.09.8';
 }
 ```
 

packages/eda-kafka/src/RdKafkaConsumerClient.php

@@ -184,9 +184,39 @@ public static function dltHeaders(ReceivedEnvelope $received, string $reason): a
         return $headers;
     }
 
+    /**
+     * Let the consumer go — and do NOT call `KafkaConsumer::close()` to do it.
+     *
+     * ext-rdkafka 6.0.5, kafka_consumer.c:531-542, is the whole body of that method:
+     *
+     *     rd_kafka_consumer_close(intern->rk);
+     *     intern->rk = NULL;
+     *
+     * There is no `rd_kafka_destroy()`. The free handler at kafka_consumer.c:53-64 destroys the
+     * handle only `if (intern->rk)` — which `close()` has just nulled — so `close()` hands the
+     * `rd_kafka_t` to nobody. The PHP object is then freed and the client outlives it, with its
+     * threads, until the process exits. Measured on PHP 8.5.8 / ext-rdkafka 6.0.5 / librdkafka
+     * 2.15.1: four OS threads stranded per closed consumer, still there after five seconds of
+     * polling; zero when the reference is simply dropped. It is unconditional, not a race.
+     *
+     * THAT IS WHY THE NULLING BELOW WAS NOT ENOUGH. Releasing the property is exactly right and was
+     * always here — but by the time it ran, `close()` had already detached the handle, so freeing
+     * the object freed nothing. The two lines looked like a careful teardown and were a leak.
+     *
+     * `unsubscribe()` is what leaves the consumer group, and it is safe on a consumer that never
+     * subscribed (verified: no throw). librdkafka logs "Destroying cgrp" when the reference is
+     * dropped afterwards, so group membership is still surrendered — nothing the old shape
+     * achieved is lost.
+     *
+     * A CONSUMER OF THIS FRAMEWORK FOUND IT THE HARD WAY. In dworkers the same call had been added
+     * to three integration tests and a long-running console command, deliberately, to stop a
+     * segfault at PHP shutdown after a green suite — and it was the cause of it. Anything that
+     * closes a consumer per restart accumulates dead clients and their threads for as long as the
+     * process lives, which matters most in exactly the place this class is used: a daemon.
+     */
     public function close(): void
     {
-        $this->consumer?->close();
+        $this->consumer?->unsubscribe();
         $this->consumer = null;
     }
 

packages/eda-kafka/tests/RdKafkaConsumerClientLeavesNoThreadsTest.php

@@ -0,0 +1,65 @@
+<?php
+
+declare(strict_types=1);
+
+use Firefly\Eda\JsonSerializer;
+use Firefly\Eda\Kafka\KafkaConsumerFactory;
+use Firefly\Eda\Kafka\KafkaProducerFactory;
+use Firefly\Eda\Kafka\RdKafkaConsumerClient;
+
+/**
+ * `RdKafkaConsumerClient::close()` MUST NOT STRAND ITS librdkafka CLIENT.
+ *
+ * It used to call `KafkaConsumer::close()` before releasing the property, which reads like a careful
+ * teardown and is a leak. ext-rdkafka 6.0.5, kafka_consumer.c:531-542 — the whole method:
+ *
+ *     rd_kafka_consumer_close(intern->rk);
+ *     intern->rk = NULL;
+ *
+ * No `rd_kafka_destroy()`, and the free handler at kafka_consumer.c:53-64 destroys the handle only
+ * `if (intern->rk)`. So after `close()` the PHP object is freed and the `rd_kafka_t` is not: it and
+ * its four OS threads live until the process exits. In a daemon — which is what this client is for —
+ * that accumulates for as long as the process runs.
+ *
+ * NO BROKER IS NEEDED and none is contacted. `new KafkaConsumer($conf)` starts librdkafka's threads
+ * and resolves the bootstrap address lazily, so an unroutable port costs nothing and keeps this test
+ * out of the docker-gated integration suite where KafkaRoundTripTest lives.
+ *
+ * `rd_kafka_thread_cnt()` IS THE INSTRUMENT, and the choice matters. A WeakReference assertion — the
+ * natural way to test that an object was released — is GREEN on this bug, because the PHP object
+ * really is freed; the leak is underneath it, in C. Only a process-global count of live handles can
+ * see it.
+ */
+it('destroys the librdkafka client when the consumer is closed, leaving no threads behind', function () {
+    $antes = rd_kafka_thread_cnt();
+
+    for ($i = 0; $i < 3; $i++) {
+        $client = new RdKafkaConsumerClient(
+            new KafkaConsumerFactory('127.0.0.1:59999', 'firefly-leak-probe-'.bin2hex(random_bytes(4))),
+            new KafkaProducerFactory('127.0.0.1:59999'),
+            new JsonSerializer,
+        );
+
+        $client->subscribe(['firefly-leak-probe-topic']);
+        $client->close();
+
+        unset($client);
+    }
+
+    // Half a second in case any part of the teardown were asynchronous. It is not — the count comes
+    // back at once — but a test that fails on a scheduler hiccup is worse than a slightly slow one.
+    usleep(500_000);
+
+    expect(rd_kafka_thread_cnt())->toBe(
+        $antes,
+        'the client was closed and its librdkafka handle outlived it. The usual cause is a '
+        .'`KafkaConsumer::close()` call: ext-rdkafka nulls the handle without destroying it, so '
+        .'closing strands the client and its threads. Call unsubscribe() and release the reference.',
+    );
+})->skip(
+    // Eagerly, at collection time, exactly as KafkaRoundTripTest gates itself: on a machine without
+    // the extension there is no handle to leak and nothing to count. No broker is required, so
+    // unlike that suite this one does NOT ask for FIREFLY_KAFKA_BROKERS.
+    ! extension_loaded('rdkafka') || ! function_exists('rd_kafka_thread_cnt'),
+    'ext-rdkafka is what leaks; without it there is nothing to count.',
+);

packages/kernel/src/Version.php

@@ -15,5 +15,5 @@
  */
 final class Version
 {
-    public const string VERSION = '26.09.6';
+    public const string VERSION = '26.09.8';
 }