From e851ed6cebfb1a1d01958fdb5f65da73a78a3158 Mon Sep 17 00:00:00 2001 From: Vladyslav Vildanov <117659936+vladvildanov@users.noreply.github.com> Date: Mon, 9 Feb 2026 10:25:51 +0200 Subject: [PATCH] Added support for idempotent stream API (#1632) * Added augment commands support * Added support for Redis 8.6 * Codestyle changes and test fixes * Fixed hybdrid tests * Updated CHANGELOG.md and composer.json * Fixed XINFO tests * Added support for idempontent stream API * Update CHANGELOG.md * Updated expected exported files * Removed agentic files * Removed .augment entry --- CHANGELOG.md | 1 + src/ClientContextInterface.php | 1 + src/ClientInterface.php | 1 + src/Command/Redis/XADD.php | 18 ++- src/Command/Redis/XCFGSET.php | 58 +++++++ tests/Predis/Command/Redis/XADD_Test.php | 84 +++++++++- tests/Predis/Command/Redis/XCFGSET_Test.php | 166 ++++++++++++++++++++ 7 files changed, 324 insertions(+), 5 deletions(-) create mode 100644 src/Command/Redis/XCFGSET.php create mode 100644 tests/Predis/Command/Redis/XCFGSET_Test.php diff --git a/CHANGELOG.md b/CHANGELOG.md index 3dce179e..afe6b75d 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -10,6 +10,7 @@ ### Added - Added retry support (#1616) - Added support for VRANGE command (#1623) +- Added support for idempotent stream API (#1632) - Added support for HOTKEYS container command (#1630) ### Maintenance diff --git a/src/ClientContextInterface.php b/src/ClientContextInterface.php index e156f9e3..1177b85b 100644 --- a/src/ClientContextInterface.php +++ b/src/ClientContextInterface.php @@ -295,6 +295,7 @@ use Predis\Command\Redis\VADD; * @method $this xadd(string $key, array $dictionary, string $id = '*', array $options = null) * @method $this xautoclaim(string $key, string $group, string $consumer, int $minIdleTime, string $start, ?int $count = null, bool $justId = false) * @method $this xclaim(string $key, string $group, string $consumer, int $minIdleTime, string|array $ids, ?int $idle = null, ?int $time = null, ?int $retryCount = null, bool $force = false, bool $justId = false, ?string $lastId = null) + * @method $this xcfgset(string $key, ?int $duration = null, ?int $maxsize = null) * @method $this xdel(string $key, string ...$id) * @method $this xdelex(string $key, string $mode, array $ids) * @method $this xlen(string $key) diff --git a/src/ClientInterface.php b/src/ClientInterface.php index 7320c9ff..ad932b64 100644 --- a/src/ClientInterface.php +++ b/src/ClientInterface.php @@ -306,6 +306,7 @@ use Predis\Response\Status; * @method string xadd(string $key, array $dictionary, string $id = '*', array $options = null) * @method array xautoclaim(string $key, string $group, string $consumer, int $minIdleTime, string $start, ?int $count = null, bool $justId = false) * @method array xclaim(string $key, string $group, string $consumer, int $minIdleTime, string|array $ids, ?int $idle = null, ?int $time = null, ?int $retryCount = null, bool $force = false, bool $justId = false, ?string $lastId = null) + * @method Status xcfgset(string $key, ?int $duration = null, ?int $maxsize = null) * @method int xdel(string $key, string ...$id) * @method array xdelex(string $key, string $mode, array $ids) * @method int xlen(string $key) diff --git a/src/Command/Redis/XADD.php b/src/Command/Redis/XADD.php index 49d4f5ea..1c3db56d 100644 --- a/src/Command/Redis/XADD.php +++ b/src/Command/Redis/XADD.php @@ -41,6 +41,20 @@ class XADD extends RedisCommand $args[] = 'NOMKSTREAM'; } + if (isset($options['trimming'])) { + $args[] = strtoupper($options['trimming']); + } + + // IDMPAUTO or IDMP options (mutually exclusive) + if (isset($options['idmpauto'])) { + $args[] = 'IDMPAUTO'; + $args[] = $options['idmpauto']; + } elseif (isset($options['idmp']) && is_array($options['idmp'])) { + $args[] = 'IDMP'; + $args[] = $options['idmp'][0]; // pid + $args[] = $options['idmp'][1]; // iid + } + if (isset($options['trim']) && is_array($options['trim'])) { array_push($args, ...$options['trim']); @@ -50,10 +64,6 @@ class XADD extends RedisCommand } } - if (isset($options['trimming'])) { - $args[] = strtoupper($options['trimming']); - } - // ID, default to * to let Redis set it $args[] = $arguments[2] ?? '*'; diff --git a/src/Command/Redis/XCFGSET.php b/src/Command/Redis/XCFGSET.php new file mode 100644 index 00000000..c6cf65d5 --- /dev/null +++ b/src/Command/Redis/XCFGSET.php @@ -0,0 +1,58 @@ +applyPrefixForFirstArgument($prefix); + } +} diff --git a/tests/Predis/Command/Redis/XADD_Test.php b/tests/Predis/Command/Redis/XADD_Test.php index bb12e38f..f5537b5a 100644 --- a/tests/Predis/Command/Redis/XADD_Test.php +++ b/tests/Predis/Command/Redis/XADD_Test.php @@ -129,7 +129,36 @@ class XADD_Test extends PredisCommandTestCase '*', ['trim' => ['MINID', '~', '0-1'], 'limit' => 5, 'nomkstream' => true, 'trimming' => 'KEEPREF'], ], - ['stream', 'NOMKSTREAM', 'MINID', '~', '0-1', 'LIMIT', 5, 'KEEPREF', '*', 'key', 'val'], + ['stream', 'NOMKSTREAM', 'KEEPREF', 'MINID', '~', '0-1', 'LIMIT', 5, '*', 'key', 'val'], + ]; + + // Test IDMPAUTO option + $data[] = [ + ['stream', ['key' => 'val'], '*', ['idmpauto' => 'producer1']], + ['stream', 'IDMPAUTO', 'producer1', '*', 'key', 'val'], + ]; + + // Test IDMP option + $data[] = [ + ['stream', ['key' => 'val'], '*', ['idmp' => ['producer1', '42']]], + ['stream', 'IDMP', 'producer1', '42', '*', 'key', 'val'], + ]; + + // Test IDMPAUTO with trimming + $data[] = [ + ['stream', ['key' => 'val'], '*', ['idmpauto' => 'producer1', 'trimming' => 'KEEPREF']], + ['stream', 'KEEPREF', 'IDMPAUTO', 'producer1', '*', 'key', 'val'], + ]; + + // Test IDMP with trim and nomkstream + $data[] = [ + [ + 'stream', + ['key' => 'val'], + '*', + ['idmp' => ['producer1', '42'], 'trim' => ['MAXLEN', '~', '100'], 'nomkstream' => true], + ], + ['stream', 'NOMKSTREAM', 'IDMP', 'producer1', '42', 'MAXLEN', '~', '100', '*', 'key', 'val'], ]; return $data; @@ -340,4 +369,57 @@ class XADD_Test extends PredisCommandTestCase $redis->set('foo', 'bar'); $redis->xadd('foo', ['key' => 'val']); } + + /** + * @group connected + * @requiresRedisVersion >= 8.5.0 + */ + public function testIdmpautoGeneratesSameIdForSameContent(): void + { + $redis = $this->getClient(); + + // Add the same message twice with IDMPAUTO + $id1 = $redis->xadd('stream', ['field1' => 'value1', 'field2' => 'value2'], '*', ['idmpauto' => 'producer1']); + $id2 = $redis->xadd('stream', ['field1' => 'value1', 'field2' => 'value2'], '*', ['idmpauto' => 'producer1']); + + // Both should return the same ID (idempotent) + $this->assertSame($id1, $id2); + // Only one entry should exist in the stream + $this->assertSame(1, $redis->xlen('stream')); + } + + /** + * @group connected + * @requiresRedisVersion >= 8.5.0 + */ + public function testIdmpUsesSameIdForSameIid(): void + { + $redis = $this->getClient(); + + // Add the same message twice with IDMP using the same iid + $id1 = $redis->xadd('stream', ['field1' => 'value1', 'field2' => 'value2'], '*', ['idmp' => ['producer1', '42']]); + $id2 = $redis->xadd('stream', ['field1' => 'value1', 'field2' => 'value2'], '*', ['idmp' => ['producer1', '42']]); + + // Both should return the same ID (idempotent) + $this->assertSame($id1, $id2); + // Only one entry should exist in the stream + $this->assertSame(1, $redis->xlen('stream')); + } + + /** + * @group connected + * @requiresRedisVersion >= 8.5.0 + */ + public function testIdmpWithDifferentIidCreatesNewEntry(): void + { + $redis = $this->getClient(); + + // Add messages with different iids + $id1 = $redis->xadd('stream', ['field1' => 'value1'], '*', ['idmp' => ['producer1', '42']]); + $id2 = $redis->xadd('stream', ['field1' => 'value1'], '*', ['idmp' => ['producer1', '43']]); + + // Should create two different entries + $this->assertNotSame($id1, $id2); + $this->assertSame(2, $redis->xlen('stream')); + } } diff --git a/tests/Predis/Command/Redis/XCFGSET_Test.php b/tests/Predis/Command/Redis/XCFGSET_Test.php new file mode 100644 index 00000000..8016d0bc --- /dev/null +++ b/tests/Predis/Command/Redis/XCFGSET_Test.php @@ -0,0 +1,166 @@ +getCommand(); + $command->setArguments($actualArguments); + + $this->assertSame($expectedArguments, $command->getArguments()); + } + + /** + * @group disconnected + */ + public function testParseResponse(): void + { + $this->assertSame('OK', $this->getCommand()->parseResponse('OK')); + } + + /** + * @group disconnected + */ + public function testPrefixKeys(): void + { + $arguments = ['stream', 100, 1000]; + $expected = ['prefix:stream', 'IDMP-DURATION', 100, 'IDMP-MAXSIZE', 1000]; + + $command = $this->getCommandWithArgumentsArray($arguments); + $command->prefixKeys('prefix:'); + + $this->assertSame($expected, $command->getArguments()); + } + + /** + * @group connected + * @requiresRedisVersion >= 8.5.0 + */ + public function testConfigureStreamIdempotencyParameters(): void + { + $redis = $this->getClient(); + + // Create a stream first + $redis->xadd('stream', ['field' => 'value']); + + // Configure IDMP parameters + $response = $redis->xcfgset('stream', 100, 1000); + + $this->assertEquals('OK', $response); + } + + /** + * @group connected + * @requiresRedisVersion >= 8.5.0 + */ + public function testIIDsAreFlushedAfterDuration(): void + { + $redis = $this->getClient(); + + // Configure stream with very short IDMP-DURATION (2 seconds) + $redis->xadd('stream', ['field' => 'value']); + $redis->xcfgset('stream', 1, 1000); + + // Add message with IID using IDMP + $id1 = $redis->xadd('stream', ['field' => 'value1'], '*', ['idmp' => ['producer1', 'iid-1']]); + + // Try to add duplicate immediately - should return same ID (idempotent) + $id2 = $redis->xadd('stream', ['field' => 'value2'], '*', ['idmp' => ['producer1', 'iid-1']]); + $this->assertSame($id1, $id2); + + // Wait for duration to expire (2 seconds + buffer) + sleep(2); + + // Now the same IID should create a new entry (IID was flushed) + $id3 = $redis->xadd('stream', ['field' => 'value3'], '*', ['idmp' => ['producer1', 'iid-1']]); + $this->assertNotSame($id1, $id3); + } + + /** + * @group connected + * @requiresRedisVersion >= 8.5.0 + */ + public function testIIDsAreFlushedWhenExceedingMaxsize(): void + { + $redis = $this->getClient(); + + // Configure stream with small IDMP-MAXSIZE (3 IIDs) + $redis->xadd('stream', ['field' => 'value']); + $redis->xcfgset('stream', 100, 3); + + // Add 3 messages with different IIDs (fills the IDMP map) + $id1 = $redis->xadd('stream', ['field' => 'value1'], '*', ['idmp' => ['producer1', 'iid-1']]); + $id2 = $redis->xadd('stream', ['field' => 'value2'], '*', ['idmp' => ['producer1', 'iid-2']]); + $id3 = $redis->xadd('stream', ['field' => 'value3'], '*', ['idmp' => ['producer1', 'iid-3']]); + + // Try to add duplicate of iid-1 - should return same ID (idempotent) + $id1Dup = $redis->xadd('stream', ['field' => 'duplicate'], '*', ['idmp' => ['producer1', 'iid-1']]); + $this->assertSame($id1, $id1Dup); + + // Add a 4th unique IID - this should cause oldest IID (iid-1) to be flushed + $id4 = $redis->xadd('stream', ['field' => 'value4'], '*', ['idmp' => ['producer1', 'iid-4']]); + $this->assertNotNull($id4); + + // Now iid-1 should create a new entry (it was flushed due to maxsize) + $id1New = $redis->xadd('stream', ['field' => 'value5'], '*', ['idmp' => ['producer1', 'iid-1']]); + $this->assertNotSame($id1, $id1New); + } + + public function argumentsProvider(): array + { + return [ + 'with key only' => [ + ['stream'], + ['stream'], + ], + 'with IDMP-DURATION only' => [ + ['stream', 100], + ['stream', 'IDMP-DURATION', 100], + ], + 'with IDMP-MAXSIZE only' => [ + ['stream', null, 1000], + ['stream', 'IDMP-MAXSIZE', 1000], + ], + 'with both IDMP-DURATION and IDMP-MAXSIZE' => [ + ['stream', 100, 1000], + ['stream', 'IDMP-DURATION', 100, 'IDMP-MAXSIZE', 1000], + ], + ]; + } +}