From c61c2d3370d90469898b3fb850214759edba6661 Mon Sep 17 00:00:00 2001 From: Norbert Orzechowicz Date: Wed, 21 Oct 2026 11:35:59 -0500 Subject: [PATCH] fix(flow-php/etl): PSRCacheBuckets keys use PSR-16 reserved characters - separate keys with dots, default prefix flow.buckets - reject prefixes with reserved characters - test against filesystem cache instead of Redis - remove Redis from dev environment and CI --- .env.dist | 2 - .github/actions/setup-php-env/action.yml | 6 +- compose.yml.dist | 8 --- docker/otel-collector-config.yaml | 21 ------ documentation/contributing/environment.md | 2 +- .../ETL/Bucketing/Storage/PSRCacheBuckets.php | 15 +++-- .../Flow/ETL/Tests/Double/ArrayCache.php | 9 +-- .../Bucketing/Storage/PSRCacheBucketsTest.php | 31 +-------- .../Cache/PSRSimpleRedisCacheTest.php | 65 ------------------- .../Bucketing/Storage/PSRCacheBucketsTest.php | 37 ++++++++--- 10 files changed, 49 insertions(+), 147 deletions(-) delete mode 100644 src/core/etl/tests/Flow/ETL/Tests/Integration/Cache/PSRSimpleRedisCacheTest.php diff --git a/.env.dist b/.env.dist index da04f713c8..44bdb3748f 100644 --- a/.env.dist +++ b/.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 diff --git a/.github/actions/setup-php-env/action.yml b/.github/actions/setup-php-env/action.yml index 47c31baee5..084aa0023e 100644 --- a/.github/actions/setup-php-env/action.yml +++ b/.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('{}') }} diff --git a/compose.yml.dist b/compose.yml.dist index 6047b06095..f7049a12d6 100644 --- a/compose.yml.dist +++ b/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 diff --git a/docker/otel-collector-config.yaml b/docker/otel-collector-config.yaml index 31a3f8c919..928e3f099c 100644 --- a/docker/otel-collector-config.yaml +++ b/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] diff --git a/documentation/contributing/environment.md b/documentation/contributing/environment.md index 3397359dcd..77a125ffc4 100644 --- a/documentation/contributing/environment.md +++ b/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: diff --git a/src/core/etl/src/Flow/ETL/Bucketing/Storage/PSRCacheBuckets.php b/src/core/etl/src/Flow/ETL/Bucketing/Storage/PSRCacheBuckets.php index cc59d6f887..f14f4fccc5 100644 --- a/src/core/etl/src/Flow/ETL/Bucketing/Storage/PSRCacheBuckets.php +++ b/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'; } } diff --git a/src/core/etl/tests/Flow/ETL/Tests/Double/ArrayCache.php b/src/core/etl/tests/Flow/ETL/Tests/Double/ArrayCache.php index e71fc8d2f0..904dfc10ea 100644 --- a/src/core/etl/tests/Flow/ETL/Tests/Double/ArrayCache.php +++ b/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; } diff --git a/src/core/etl/tests/Flow/ETL/Tests/Integration/Bucketing/Storage/PSRCacheBucketsTest.php b/src/core/etl/tests/Flow/ETL/Tests/Integration/Bucketing/Storage/PSRCacheBucketsTest.php index 5c7f0db6f9..44afdaadf6 100644 --- a/src/core/etl/tests/Flow/ETL/Tests/Integration/Bucketing/Storage/PSRCacheBucketsTest.php +++ b/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')))); diff --git a/src/core/etl/tests/Flow/ETL/Tests/Integration/Cache/PSRSimpleRedisCacheTest.php b/src/core/etl/tests/Flow/ETL/Tests/Integration/Cache/PSRSimpleRedisCacheTest.php deleted file mode 100644 index b713d85a41..0000000000 --- a/src/core/etl/tests/Flow/ETL/Tests/Integration/Cache/PSRSimpleRedisCacheTest.php +++ /dev/null @@ -1,65 +0,0 @@ -dsn(), ['retry_interval' => 1, 'timeout' => 1]); - $this->redisAvailable = true; - } catch (Exception) { - self::markTestSkipped('Redis server is not available.'); - } - - parent::setUp(); - } - - #[Override] - protected function tearDown(): void - { - // Skipped in setUp before the parent state was initialized - there is - // nothing to clean up and the parent tearDown would error. - if (!$this->redisAvailable) { - return; - } - - parent::tearDown(); - } - - protected function cache(): Cache - { - return new PSRSimpleCache( - new Psr16Cache(new RedisAdapter(RedisAdapter::createConnection($this->dsn(), [ - 'retry_interval' => 2, - 'timeout' => 5, - ]))), - new FloeSerializer(new AdaptiveBackend()), - ); - } - - protected function dsn(): string - { - $host = getenv('REDIS_HOST'); - $port = getenv('REDIS_PORT'); - - return 'redis://' . ($host === false ? 'localhost' : $host) . ':' . ($port === false ? '6379' : $port) . '/0'; - } -} diff --git a/src/core/etl/tests/Flow/ETL/Tests/Unit/Bucketing/Storage/PSRCacheBucketsTest.php b/src/core/etl/tests/Flow/ETL/Tests/Unit/Bucketing/Storage/PSRCacheBucketsTest.php index a951db54fe..0b94ae7db9 100644 --- a/src/core/etl/tests/Flow/ETL/Tests/Unit/Bucketing/Storage/PSRCacheBucketsTest.php +++ b/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()));