fix(flow-php/etl): PSRCacheBuckets keys use PSR-16 reserved characters

.env.dist

@@ -11,8 +11,6 @@ SFTP_HOST=localhost
 SFTP_PORT=2222
 SFTP_USER=flow
 SFTP_PASSWORD=flow
-REDIS_HOST=localhost
-REDIS_PORT=6379
 PGSQL_DATABASE_URL=pgsql://postgres:postgres@127.0.0.1:5432/postgres?serverVersion=18&charset=utf8
 MYSQL_DATABASE_URL=mysql://mysql:mysql@127.0.0.1:3306/mysql
 SQLITE_DATABASE_PATH=${FLOW_MONOREPO_PROJECT_ROOT}/var/db/sqlite/test.db

.github/actions/setup-php-env/action.yml

@@ -80,7 +80,7 @@ runs:
       uses: shivammathur/cache-extensions@22e24453af1bf3851994136f122ce065ab828a74 # 1.15.2
       with:
         php-version: ${{ inputs.php-version }}
-        extensions: "${{ inputs.extensions }}, redis-6.3.0"
+        extensions: "${{ inputs.extensions }}, :redis"
         key: "php-extensions-${{ runner.os }}-${{ runner.arch }}-v1"
       env:
         share-cache-across-jobs: true
@@ -101,9 +101,7 @@ runs:
         ini-values: ${{ inputs.ini-values }}
         # The runner image ships ext-redis 5.3.7 pre-installed, which conflicts with
         # symfony/cache (requires ext-redis >= 6.1) and breaks `composer install`.
-        # Pin a compatible phpredis (>= 6.1, and new enough to build on PHP 8.5) so the
-        # extension stays available for the redis cache tests instead of being dropped.
-        extensions: "${{ inputs.extensions }}, redis-6.3.0"
+        extensions: "${{ inputs.extensions }}, :redis"
         coverage: ${{ inputs.coverage != '' && inputs.coverage || null }}
       env: ${{ inputs.php-env != '' && fromJSON(inputs.php-env) || fromJSON('{}') }}
 

compose.yml.dist

@@ -53,13 +53,6 @@ services:
             - ./var/azurite:/workspace
         networks:
             - flow-php
-    redis:
-        image: redis:latest
-        container_name: flow-php-redis
-        ports:
-            - "6379:6379"
-        networks:
-            - flow-php
     rustfs:
         image: rustfs/rustfs:1.0.0
         container_name: flow-php-rustfs
@@ -117,6 +110,5 @@ services:
             - aspire
             - postgres
             - mysql
-            - redis
         networks:
             - flow-php

docker/otel-collector-config.yaml

@@ -37,14 +37,6 @@ receivers:
     tls:
       insecure: true
 
-  # Redis metrics receiver
-  # Docs: https://github.com/open-telemetry/opentelemetry-collector-contrib/tree/main/receiver/redisreceiver
-  redis:
-    endpoint: redis:6379
-    collection_interval: 10s
-    tls:
-      insecure: true
-
 exporters:
   debug:
     verbosity: basic
@@ -84,15 +76,6 @@ processors:
         value: flow-php
         action: upsert
 
-  # Add service.name to Redis metrics
-  resource/redis:
-    attributes:
-      - key: service.name
-        value: redis
-        action: upsert
-      - key: service.namespace
-        value: flow-php
-        action: upsert
 
 extensions:
   health_check:
@@ -154,10 +137,6 @@ service:
       receivers: [mysql]
       processors: [resource/mysql]
       exporters: [debug, otlp_grpc/aspire]
-    metrics/redis:
-      receivers: [redis]
-      processors: [resource/redis]
-      exporters: [debug, otlp_grpc/aspire]
     logs:
       receivers: [otlp]
       exporters: [debug, otlp_grpc/aspire]

documentation/contributing/environment.md

@@ -14,7 +14,7 @@ just install
 docker compose up -d
 ```
 
-Service and test env config (database URLs, S3/Redis/Azurite/OTel endpoints, the filesystem cache dir) lives in the
+Service and test env config (database URLs, S3/Azurite/OTel endpoints, the filesystem cache dir) lives in the
 committed `.env.dist`, loaded for both PHPUnit and phpbench by the root `bootstrap.php`. The `.dist` defaults match the
 `compose.yml.dist` ports, so a standard setup needs no extra step. If you shift service ports in your own `compose.yml`,
 copy `.env.dist` to `.env` (gitignored) and set the shifted values there - `.env` overrides `.env.dist` per-variable:

src/core/etl/src/Flow/ETL/Bucketing/Storage/PSRCacheBuckets.php

@@ -6,6 +6,7 @@
 
 use DateInterval;
 use Flow\ETL\Bucketing\BucketsStorage;
+use Flow\ETL\Exception\InvalidArgumentException;
 use Flow\ETL\Rows;
 use Flow\Serializer\Serializer;
 use Generator;
@@ -15,15 +16,21 @@
 use function Flow\Serializer\DSL\unserialize_from_string;
 use function is_int;
 use function is_string;
+use function strpbrk;
 
 final readonly class PSRCacheBuckets implements BucketsStorage
 {
     public function __construct(
         private CacheInterface $cache,
         private Serializer $serializer,
-        private string $prefix = 'flow:buckets',
+        private string $prefix = 'flow.buckets',
         private ?DateInterval $ttl = null,
-    ) {}
+    ) {
+        if (strpbrk($this->prefix, '{}()/\\@:') !== false) {
+            throw new InvalidArgumentException('Prefix must not contain PSR-16 reserved characters {}()/\\@:, given: '
+            . $this->prefix);
+        }
+    }
 
     public function append(string $bucketId, Rows $rows): void
     {
@@ -81,11 +88,11 @@ private function chunkCount(string $bucketId): int
 
     private function chunkKey(string $bucketId, int $n): string
     {
-        return $this->prefix . ':' . $bucketId . ':chunk:' . $n;
+        return $this->prefix . '.' . $bucketId . '.chunk.' . $n;
     }
 
     private function chunksKey(string $bucketId): string
     {
-        return $this->prefix . ':' . $bucketId . ':chunks';
+        return $this->prefix . '.' . $bucketId . '.chunks';
     }
 }

src/core/etl/tests/Flow/ETL/Tests/Double/ArrayCache.php

@@ -6,6 +6,7 @@
 
 use DateInterval;
 use Psr\SimpleCache\CacheInterface;
+use Symfony\Component\Cache\CacheItem;
 
 use function array_key_exists;
 
@@ -34,7 +35,7 @@ public function clear(): bool
      */
     public function delete($key): bool
     {
-        unset($this->values[$key]);
+        unset($this->values[CacheItem::validateKey($key)]);
 
         return true;
     }
@@ -56,7 +57,7 @@ public function deleteMultiple($keys): bool
      */
     public function get($key, mixed $default = null): mixed
     {
-        return array_key_exists($key, $this->values) ? $this->values[$key] : $default;
+        return array_key_exists(CacheItem::validateKey($key), $this->values) ? $this->values[$key] : $default;
     }
 
     /**
@@ -80,7 +81,7 @@ public function getMultiple($keys, mixed $default = null): iterable
      */
     public function has($key): bool
     {
-        return array_key_exists($key, $this->values);
+        return array_key_exists(CacheItem::validateKey($key), $this->values);
     }
 
     /**
@@ -89,7 +90,7 @@ public function has($key): bool
      */
     public function set($key, mixed $value, $ttl = null): bool
     {
-        $this->values[$key] = $value;
+        $this->values[CacheItem::validateKey($key)] = $value;
 
         return true;
     }

src/core/etl/tests/Flow/ETL/Tests/Integration/Bucketing/Storage/PSRCacheBucketsTest.php

@@ -4,52 +4,25 @@
 
 namespace Flow\ETL\Tests\Integration\Bucketing\Storage;
 
-use Exception;
 use Flow\ETL\Bucketing\Storage\PSRCacheBuckets;
 use Flow\ETL\Column\AdaptiveBackend;
 use Flow\ETL\Tests\Context\BucketsStorageContext;
 use Flow\ETL\Tests\FlowIntegrationTestCase;
 use Flow\Floe\FloeSerializer;
-use Override;
-use Symfony\Component\Cache\Adapter\RedisAdapter;
+use Symfony\Component\Cache\Adapter\FilesystemAdapter;
 use Symfony\Component\Cache\Psr16Cache;
 
 use function Flow\ETL\DSL\array_to_rows;
 use function Flow\ETL\DSL\int_schema;
 use function Flow\ETL\DSL\schema;
-use function getenv;
 
 final class PSRCacheBucketsTest extends FlowIntegrationTestCase
 {
-    #[Override]
-    protected function setUp(): void
-    {
-        parent::setUp();
-
-        $host = getenv('REDIS_HOST') === false ? 'localhost' : getenv('REDIS_HOST');
-        $port = getenv('REDIS_PORT') === false ? '6379' : getenv('REDIS_PORT');
-
-        try {
-            RedisAdapter::createConnection("redis://{$host}:{$port}/0", ['retry_interval' => 1, 'timeout' => 1]);
-        } catch (Exception) {
-            self::markTestSkipped('Redis server is not available.');
-        }
-    }
-
     public function test_round_trips_rows_across_chunks_and_removes_the_bucket(): void
     {
-        $host = getenv('REDIS_HOST') === false ? 'localhost' : getenv('REDIS_HOST');
-        $port = getenv('REDIS_PORT') === false ? '6379' : getenv('REDIS_PORT');
-
-        $cache = new Psr16Cache(new RedisAdapter(RedisAdapter::createConnection("redis://{$host}:{$port}/0", [
-            'retry_interval' => 2,
-            'timeout' => 5,
-        ])));
-
         $storage = new PSRCacheBuckets(
-            $cache,
+            new Psr16Cache(new FilesystemAdapter(directory: __DIR__ . '/var/psr-cache-buckets')),
             new FloeSerializer(new AdaptiveBackend()),
-            prefix: 'flow:buckets:test:round_trip',
         );
         $storage->append('bucket', array_to_rows([['id' => 1], ['id' => 2]], schema(int_schema('id'))));
         $storage->append('bucket', array_to_rows([['id' => 3]], schema(int_schema('id'))));

src/core/etl/tests/Flow/ETL/Tests/Unit/Bucketing/Storage/PSRCacheBucketsTest.php

@@ -6,10 +6,12 @@
 
 use Flow\ETL\Bucketing\Storage\PSRCacheBuckets;
 use Flow\ETL\Column\AdaptiveBackend;
+use Flow\ETL\Exception\InvalidArgumentException;
 use Flow\ETL\Tests\Context\BucketsStorageContext;
 use Flow\ETL\Tests\Double\ArrayCache;
 use Flow\ETL\Tests\FlowTestCase;
 use Flow\Floe\FloeSerializer;
+use PHPUnit\Framework\Attributes\TestWith;
 
 use function Flow\ETL\DSL\array_to_rows;
 use function Flow\ETL\DSL\int_schema;
@@ -33,20 +35,20 @@ public function test_custom_prefix_is_used_in_keys(): void
         (new PSRCacheBuckets(
             $cache,
             new FloeSerializer(new AdaptiveBackend()),
-            prefix: 'custom:pfx',
+            prefix: 'custom.pfx',
         ))->append('bucket', array_to_rows([[
             'id' => 1,
         ]], schema(int_schema('id'))));
 
-        static::assertTrue($cache->has('custom:pfx:bucket:chunks'));
-        static::assertTrue($cache->has('custom:pfx:bucket:chunk:0'));
+        static::assertTrue($cache->has('custom.pfx.bucket.chunks'));
+        static::assertTrue($cache->has('custom.pfx.bucket.chunk.0'));
     }
 
     public function test_get_skips_non_string_chunk_payloads(): void
     {
         $cache = new ArrayCache();
-        $cache->set('flow:buckets:bucket:chunks', 1);
-        $cache->set('flow:buckets:bucket:chunk:0', 123);
+        $cache->set('flow.buckets.bucket.chunks', 1);
+        $cache->set('flow.buckets.bucket.chunk.0', 123);
 
         static::assertSame(
             [],
@@ -59,7 +61,7 @@ public function test_get_skips_non_string_chunk_payloads(): void
     public function test_get_treats_non_integer_chunk_counter_as_empty(): void
     {
         $cache = new ArrayCache();
-        $cache->set('flow:buckets:bucket:chunks', 'garbage');
+        $cache->set('flow.buckets.bucket.chunks', 'garbage');
 
         static::assertSame(
             [],
@@ -89,9 +91,9 @@ public function test_remove_deletes_all_chunk_keys_and_the_counter(): void
 
         $storage->remove('bucket');
 
-        static::assertFalse($cache->has('flow:buckets:bucket:chunks'));
-        static::assertFalse($cache->has('flow:buckets:bucket:chunk:0'));
-        static::assertFalse($cache->has('flow:buckets:bucket:chunk:1'));
+        static::assertFalse($cache->has('flow.buckets.bucket.chunks'));
+        static::assertFalse($cache->has('flow.buckets.bucket.chunk.0'));
+        static::assertFalse($cache->has('flow.buckets.bucket.chunk.1'));
         static::assertSame([], BucketsStorageContext::rows($storage->get('bucket')));
     }
 
@@ -107,6 +109,23 @@ public function test_round_trips_schema_changing_rows(): void
         static::assertSame($input->toArray(), BucketsStorageContext::rows($storage->get('bucket')));
     }
 
+    #[TestWith(['flow{buckets'])]
+    #[TestWith(['flow}buckets'])]
+    #[TestWith(['flow(buckets'])]
+    #[TestWith(['flow)buckets'])]
+    #[TestWith(['flow/buckets'])]
+    #[TestWith(['flow\\buckets'])]
+    #[TestWith(['flow@buckets'])]
+    #[TestWith(['flow:buckets'])]
+    public function test_rejects_prefix_with_psr16_reserved_character(string $prefix): void
+    {
+        $this->expectException(InvalidArgumentException::class);
+        $this->expectExceptionMessage('Prefix must not contain PSR-16 reserved characters {}()/\\@:, given: '
+        . $prefix);
+
+        new PSRCacheBuckets(new ArrayCache(), new FloeSerializer(new AdaptiveBackend()), prefix: $prefix);
+    }
+
     public function test_set_replaces_previous_content(): void
     {
         $storage = new PSRCacheBuckets(new ArrayCache(), new FloeSerializer(new AdaptiveBackend()));