From 42184411ec5cdd7365f155b8d0ff4a8446035428 Mon Sep 17 00:00:00 2001 From: Your Name Date: Mon, 30 Mar 2026 14:41:23 +0200 Subject: [PATCH 1/5] fix: harden object store metadata and reads --- src/ObjectStore/ObjectResult.php | 32 ++- src/ObjectStore/ObjectStore.php | 225 ++++++++++++------ .../ObjectStore/ObjectStoreTest.php | 51 ++++ tests/Unit/ObjectStore/ObjectResultTest.php | 45 +++- 4 files changed, 270 insertions(+), 83 deletions(-) diff --git a/src/ObjectStore/ObjectResult.php b/src/ObjectStore/ObjectResult.php index 3602a6f..4a0bb42 100644 --- a/src/ObjectStore/ObjectResult.php +++ b/src/ObjectStore/ObjectResult.php @@ -4,12 +4,17 @@ namespace Sofascore\Nats\ObjectStore; +use Sofascore\Nats\NatsException; + final class ObjectResult { private string $buffer = ''; private bool $closed = false; private bool $sourceStarted = false; private bool $sourceExhausted = false; + private int $bytesRead = 0; + private int $chunksRead = 0; + private readonly \HashContext $hashContext; /** @var null|\Generator */ private ?\Generator $source = null; @@ -17,7 +22,9 @@ public function __construct( private readonly ObjectInfo $info, private readonly \Closure $sourceFactory, private ?\Closure $onClose = null, - ) {} + ) { + $this->hashContext = hash_init('sha256'); + } public function read(int $length): string { @@ -130,6 +137,9 @@ private function nextChunk(): ?string try { $chunk = $this->source->current(); $this->source->next(); + $this->bytesRead += strlen($chunk); + $this->chunksRead++; + hash_update($this->hashContext, $chunk); return $chunk; } catch (\Throwable $error) { $this->close(); @@ -143,11 +153,31 @@ private function finishSource(): void return; } + $this->assertCompleteRead(); $this->sourceExhausted = true; $this->source = null; $this->runCleanup(); } + private function assertCompleteRead(): void + { + $expectedChunks = max(1, $this->info->chunks); + if ($this->bytesRead !== $this->info->size || $this->chunksRead !== $expectedChunks) { + throw new NatsException( + "Object read incomplete: {$this->info->name} ({$this->bytesRead}/{$this->info->size} bytes)" + ); + } + + if ($this->info->digest === '') { + return; + } + + $actualDigest = 'SHA-256=' . base64_encode(hash_final($this->hashContext, true)); + if ($actualDigest !== $this->info->digest) { + throw new NatsException("Object digest mismatch: {$this->info->name}"); + } + } + private function runCleanup(): void { if ($this->onClose === null) { diff --git a/src/ObjectStore/ObjectStore.php b/src/ObjectStore/ObjectStore.php index 49f3975..15e9e66 100644 --- a/src/ObjectStore/ObjectStore.php +++ b/src/ObjectStore/ObjectStore.php @@ -90,10 +90,11 @@ public function putStream(string $name, mixed $stream): ObjectInfo self::assertReadableStream($stream, $name); $chunkSize = self::requirePositiveChunkSize($this->chunkSize); $jsStream = $this->js->stream("OBJ_{$this->bucketName}"); + $existingMetaSubject = null; // Purge old chunks if overwriting try { - $existing = $this->getObjectInfo($name); + [$existing, $existingMetaSubject] = $this->getObjectInfoRecord($name, showDeleted: true); if (!$existing->deleted && $existing->nuid !== '') { $jsStream->purge(new \Sofascore\Nats\JetStream\Stream\StreamPurgeOptions( filter: "\$O.{$this->bucketName}.C.{$existing->nuid}", @@ -147,16 +148,17 @@ public function putStream(string $name, mixed $stream): ObjectInfo mtime: new \DateTimeImmutable(), ); - $metaSubject = "\$O.{$this->bucketName}.M.{$this->encodeName($name)}"; - $metaHeaders = new Headers(); - $metaHeaders->set('Nats-Rollup', 'sub'); + $this->publishObjectInfo($info); - $msg = new Message( - subject: $metaSubject, - data: json_encode($info->toArray(), JSON_THROW_ON_ERROR), - headers: $metaHeaders, - ); - $this->js->publishMessage($msg); + if ($existingMetaSubject !== null && $existingMetaSubject !== $this->metaSubject($name)) { + try { + $jsStream->purge(new \Sofascore\Nats\JetStream\Stream\StreamPurgeOptions( + filter: $existingMetaSubject, + )); + } catch (\Throwable) { + // Best effort cleanup for legacy metadata subjects. + } + } } catch (\Throwable $e) { try { $jsStream->purge(new \Sofascore\Nats\JetStream\Stream\StreamPurgeOptions( @@ -209,7 +211,7 @@ function () use ($info, $consumer): \Generator { } if ($fetched === 0) { - return; + throw new NatsException("Object read incomplete: {$info->name}"); } } }, @@ -260,11 +262,7 @@ public function getFile(string $name, string $filePath): void public function getInfo(string $name, bool $showDeleted = false): ObjectInfo { - $info = $this->getObjectInfo($name); - if (!$showDeleted && $info->deleted) { - throw new NatsException("Object not found: {$name}"); - } - return $info; + return $this->getObjectInfo($name, $showDeleted); } /** @@ -272,7 +270,7 @@ public function getInfo(string $name, bool $showDeleted = false): ObjectInfo */ public function delete(string $name): void { - $info = $this->getObjectInfo($name); + [$info, $metaSubject] = $this->getObjectInfoRecord($name); // Purge chunks $stream = $this->js->stream("OBJ_{$this->bucketName}"); @@ -291,21 +289,26 @@ public function delete(string $name): void deleted: true, ); - $metaSubject = "\$O.{$this->bucketName}.M.{$this->encodeName($name)}"; - $headers = new Headers(); - $headers->set('Nats-Rollup', 'sub'); + $this->publishObjectInfo($deletedInfo); - $msg = new Message( - subject: $metaSubject, - data: json_encode($deletedInfo->toArray(), JSON_THROW_ON_ERROR), - headers: $headers, - ); - $this->js->publishMessage($msg); + if ($metaSubject !== $this->metaSubject($name)) { + try { + $stream->purge(new \Sofascore\Nats\JetStream\Stream\StreamPurgeOptions( + filter: $metaSubject, + )); + } catch (\Throwable) { + // Best effort cleanup for legacy metadata subjects. + } + } } public function updateMeta(string $name, ObjectMeta $meta): ObjectInfo { - $existing = $this->getObjectInfo($name); + [$existing, $existingMetaSubject] = $this->getObjectInfoRecord($name); + + if ($name !== $meta->name && $this->activeObjectExists($meta->name)) { + throw new NatsException("Object already exists: {$meta->name}"); + } $updatedInfo = new ObjectInfo( name: $meta->name, @@ -322,16 +325,13 @@ public function updateMeta(string $name, ObjectMeta $meta): ObjectInfo metadata: $meta->metadata !== [] ? $meta->metadata : $existing->metadata, ); - $metaSubject = "\$O.{$this->bucketName}.M.{$this->encodeName($meta->name)}"; - $headers = new Headers(); - $headers->set('Nats-Rollup', 'sub'); + $this->publishObjectInfo($updatedInfo); - $msg = new Message( - subject: $metaSubject, - data: json_encode($updatedInfo->toArray(), JSON_THROW_ON_ERROR), - headers: $headers, - ); - $this->js->publishMessage($msg); + if ($existingMetaSubject !== $this->metaSubject($meta->name)) { + $this->js->stream($this->streamName())->purge(new \Sofascore\Nats\JetStream\Stream\StreamPurgeOptions( + filter: $existingMetaSubject, + )); + } return $updatedInfo; } @@ -360,17 +360,10 @@ public function addLink(string $name, ObjectInfo $target): ObjectInfo name: $target->name, ), ); + $previousMetaSubject = $this->existingMetaSubjectOrNull($name); - $metaSubject = "\$O.{$this->bucketName}.M.{$this->encodeName($name)}"; - $headers = new Headers(); - $headers->set('Nats-Rollup', 'sub'); - - $msg = new Message( - subject: $metaSubject, - data: json_encode($linkInfo->toArray(), JSON_THROW_ON_ERROR), - headers: $headers, - ); - $this->js->publishMessage($msg); + $this->publishObjectInfo($linkInfo); + $this->purgeLegacyMetaSubjectIfNeeded($name, $previousMetaSubject); return $linkInfo; } @@ -390,17 +383,10 @@ public function addBucketLink(string $name, ObjectStoreInterface $targetStore): bucket: $targetStore->bucket(), ), ); + $previousMetaSubject = $this->existingMetaSubjectOrNull($name); - $metaSubject = "\$O.{$this->bucketName}.M.{$this->encodeName($name)}"; - $headers = new Headers(); - $headers->set('Nats-Rollup', 'sub'); - - $msg = new Message( - subject: $metaSubject, - data: json_encode($linkInfo->toArray(), JSON_THROW_ON_ERROR), - headers: $headers, - ); - $this->js->publishMessage($msg); + $this->publishObjectInfo($linkInfo); + $this->purgeLegacyMetaSubjectIfNeeded($name, $previousMetaSubject); return $linkInfo; } @@ -537,10 +523,11 @@ private function putRaw( ?int $chunkSize = null, ): ObjectInfo { $chunkSize = self::requirePositiveChunkSize($chunkSize ?? $this->chunkSize); + $existingMetaSubject = null; // Purge old chunks if overwriting try { - $existing = $this->getObjectInfo($name); + [$existing, $existingMetaSubject] = $this->getObjectInfoRecord($name, showDeleted: true); if (!$existing->deleted && $existing->nuid !== '') { $stream = $this->js->stream("OBJ_{$this->bucketName}"); $stream->purge(new \Sofascore\Nats\JetStream\Stream\StreamPurgeOptions( @@ -590,38 +577,54 @@ private function putRaw( metadata: $metadata, ); - // Publish metadata with rollup - $metaSubject = "\$O.{$this->bucketName}.M.{$this->encodeName($name)}"; - $metaHeaders = new Headers(); - $metaHeaders->set('Nats-Rollup', 'sub'); + $this->publishObjectInfo($info); + $this->purgeLegacyMetaSubjectIfNeeded($name, $existingMetaSubject); - $msg = new Message( - subject: $metaSubject, - data: json_encode($info->toArray(), JSON_THROW_ON_ERROR), - headers: $metaHeaders, - ); - $this->js->publishMessage($msg); + return $info; + } + private function getObjectInfo(string $name, bool $showDeleted = false): ObjectInfo + { + [$info] = $this->getObjectInfoRecord($name, $showDeleted); return $info; } - private function getObjectInfo(string $name): ObjectInfo + /** + * @return array{ObjectInfo, string} + */ + private function getObjectInfoRecord(string $name, bool $showDeleted = false): array { - $metaSubject = $this->metaSubject($name); $stream = $this->js->stream($this->streamName()); - try { - $rawMsg = $stream->getLastMessageForSubject($metaSubject); - } catch (\Throwable) { - throw new NatsException("Object not found: {$name}"); + foreach ($this->metaSubjectCandidates($name) as $metaSubject) { + try { + $rawMsg = $stream->getLastMessageForSubject($metaSubject); + } catch (\Throwable) { + continue; + } + + $info = $this->decodeObjectInfo($rawMsg->data); + if ($info->name !== $name) { + continue; + } + + if (!$showDeleted && $info->deleted) { + throw new NatsException("Object not found: {$name}"); + } + + return [$info, $metaSubject]; } - return $this->decodeObjectInfo($rawMsg->data); + throw new NatsException("Object not found: {$name}"); } private function encodeName(string $name): string { - // Replace special chars that aren't valid in NATS subjects + return strtr(base64_encode($name), '+/', '-_'); + } + + private function encodeLegacyName(string $name): string + { return str_replace(['/', ' '], ['_', '_'], $name); } @@ -685,11 +688,83 @@ private function metaSubject(string $name): string return $this->metaSubjectPrefix() . $this->encodeName($name); } + /** + * @return list + */ + private function metaSubjectCandidates(string $name): array + { + $subjects = [$this->metaSubject($name)]; + $legacySubject = $this->metaSubjectPrefix() . $this->encodeLegacyName($name); + + if ($legacySubject !== $subjects[0]) { + $subjects[] = $legacySubject; + } + + return $subjects; + } + private function metaSubjectPrefix(): string { return "\$O.{$this->bucketName}.M."; } + private function publishObjectInfo(ObjectInfo $info): void + { + $headers = new Headers(); + $headers->set('Nats-Rollup', 'sub'); + + $this->js->publishMessage(new Message( + subject: $this->metaSubject($info->name), + data: json_encode($info->toArray(), JSON_THROW_ON_ERROR), + headers: $headers, + )); + } + + private function existingMetaSubjectOrNull(string $name): ?string + { + try { + [, $metaSubject] = $this->getObjectInfoRecord($name, showDeleted: true); + return $metaSubject; + } catch (\Throwable) { + return null; + } + } + + private function activeObjectExists(string $name): bool + { + $stream = $this->js->stream($this->streamName()); + + foreach ($this->metaSubjectCandidates($name) as $metaSubject) { + try { + $rawMsg = $stream->getLastMessageForSubject($metaSubject); + } catch (\Throwable) { + continue; + } + + $info = $this->decodeObjectInfo($rawMsg->data); + if ($info->name === $name && !$info->deleted) { + return true; + } + } + + return false; + } + + private function purgeLegacyMetaSubjectIfNeeded(string $name, ?string $existingMetaSubject): void + { + if ($existingMetaSubject === null || $existingMetaSubject === $this->metaSubject($name)) { + return; + } + + try { + $this->js->stream($this->streamName())->purge(new \Sofascore\Nats\JetStream\Stream\StreamPurgeOptions( + filter: $existingMetaSubject, + )); + } catch (\Throwable) { + // Best effort cleanup for legacy metadata subjects. + } + } + /** * @return int<1, max> */ diff --git a/tests/Integration/ObjectStore/ObjectStoreTest.php b/tests/Integration/ObjectStore/ObjectStoreTest.php index d9cc41c..394a31d 100644 --- a/tests/Integration/ObjectStore/ObjectStoreTest.php +++ b/tests/Integration/ObjectStore/ObjectStoreTest.php @@ -271,6 +271,39 @@ public function testOverwriteObject(): void self::assertSame('version 2', $content); } + public function testNamesWithSlashesAndUnderscoresDoNotCollide(): void + { + $store = $this->createStore(); + + $store->putBytes('a_b', 'underscore'); + $store->putBytes('a/b', 'slash'); + + self::assertSame('underscore', $store->getBytes('a_b')); + self::assertSame('slash', $store->getBytes('a/b')); + + $names = array_map(static fn($object) => $object->name, $store->list()); + sort($names); + + self::assertSame(['a/b', 'a_b'], $names); + } + + public function testUpdateMetaRenameRemovesOldName(): void + { + $store = $this->createStore(); + + $store->putBytes('old.txt', 'payload'); + $updated = $store->updateMeta('old.txt', new ObjectMeta( + name: 'new.txt', + description: 'renamed', + )); + + self::assertSame('new.txt', $updated->name); + self::assertSame('payload', $store->getBytes('new.txt')); + + $this->expectException(\Sofascore\Nats\NatsException::class); + $store->getBytes('old.txt'); + } + public function testDigestVerification(): void { $store = $this->createStore(); @@ -298,4 +331,22 @@ public function testGetFile(): void unlink($tmpFile); } } + + public function testGetFailsWhenChunksDisappearMidRead(): void + { + $store = $this->createStore(); + $store->putBytes('volatile.bin', str_repeat('ABCDEFGH', 32)); + + $result = $store->get('volatile.bin'); + + try { + $store->delete('volatile.bin'); + + $this->expectException(\Sofascore\Nats\NatsException::class); + $this->expectExceptionMessage('Object read incomplete'); + $result->readAll(); + } finally { + $result->close(); + } + } } diff --git a/tests/Unit/ObjectStore/ObjectResultTest.php b/tests/Unit/ObjectStore/ObjectResultTest.php index 6c49d8b..cdee8a9 100644 --- a/tests/Unit/ObjectStore/ObjectResultTest.php +++ b/tests/Unit/ObjectStore/ObjectResultTest.php @@ -5,6 +5,7 @@ namespace Sofascore\Nats\Tests\Unit\ObjectStore; use PHPUnit\Framework\TestCase; +use Sofascore\Nats\NatsException; use Sofascore\Nats\ObjectStore\ObjectInfo; use Sofascore\Nats\ObjectStore\ObjectResult; @@ -14,7 +15,7 @@ public function testReadConsumesChunksLazily(): void { $cleanupCalls = 0; $result = new ObjectResult( - $this->createInfo(size: 11), + $this->createInfo(data: 'hello world', chunks: 3), static function (): \Generator { yield 'hello'; yield ' '; @@ -35,7 +36,7 @@ function () use (&$cleanupCalls): void { public function testWriteToStreamsRemainingContent(): void { $result = new ObjectResult( - $this->createInfo(size: 10), + $this->createInfo(data: 'abcdefghij', chunks: 3), static function (): \Generator { yield 'abc'; yield 'defg'; @@ -63,7 +64,7 @@ public function testCloseRunsCleanupOnce(): void { $cleanupCalls = 0; $result = new ObjectResult( - $this->createInfo(size: 5), + $this->createInfo(data: 'hello', chunks: 1), static function (): \Generator { yield 'hello'; }, @@ -79,15 +80,45 @@ function () use (&$cleanupCalls): void { self::assertSame('', $result->readAll()); } - private function createInfo(int $size): ObjectInfo + public function testReadAllThrowsWhenStreamEndsBeforeExpectedBytes(): void + { + $result = new ObjectResult( + $this->createInfo(data: 'hello world', chunks: 3), + static function (): \Generator { + yield 'hello'; + }, + ); + + $this->expectException(NatsException::class); + $this->expectExceptionMessage('Object read incomplete'); + + $result->readAll(); + } + + public function testReadAllThrowsOnDigestMismatch(): void + { + $result = new ObjectResult( + $this->createInfo(data: 'hello', chunks: 1, digest: 'SHA-256=invalid'), + static function (): \Generator { + yield 'hello'; + }, + ); + + $this->expectException(NatsException::class); + $this->expectExceptionMessage('Object digest mismatch'); + + $result->readAll(); + } + + private function createInfo(string $data, int $chunks, ?string $digest = null): ObjectInfo { return new ObjectInfo( name: 'file.bin', bucket: 'files', nuid: 'nuid', - size: $size, - chunks: 1, - digest: 'SHA-256=test', + size: strlen($data), + chunks: $chunks, + digest: $digest ?? 'SHA-256=' . base64_encode(hash('sha256', $data, true)), deleted: false, ); } From 50b240465effc1e2964978679e880fa3fa2c4cc0 Mon Sep 17 00:00:00 2001 From: Your Name Date: Mon, 30 Mar 2026 14:49:19 +0200 Subject: [PATCH 2/5] refactor: reuse object lookup for rename conflict check --- src/ObjectStore/ObjectStore.php | 20 +++++--------------- 1 file changed, 5 insertions(+), 15 deletions(-) diff --git a/src/ObjectStore/ObjectStore.php b/src/ObjectStore/ObjectStore.php index 15e9e66..33937c4 100644 --- a/src/ObjectStore/ObjectStore.php +++ b/src/ObjectStore/ObjectStore.php @@ -732,22 +732,12 @@ private function existingMetaSubjectOrNull(string $name): ?string private function activeObjectExists(string $name): bool { - $stream = $this->js->stream($this->streamName()); - - foreach ($this->metaSubjectCandidates($name) as $metaSubject) { - try { - $rawMsg = $stream->getLastMessageForSubject($metaSubject); - } catch (\Throwable) { - continue; - } - - $info = $this->decodeObjectInfo($rawMsg->data); - if ($info->name === $name && !$info->deleted) { - return true; - } + try { + $this->getObjectInfo($name); + return true; + } catch (NatsException) { + return false; } - - return false; } private function purgeLegacyMetaSubjectIfNeeded(string $name, ?string $existingMetaSubject): void From 897dd092cd45972f776ccfc0246f247df0d824af Mon Sep 17 00:00:00 2001 From: Your Name Date: Mon, 30 Mar 2026 15:26:09 +0200 Subject: [PATCH 3/5] refactor: avoid exception control flow in object lookup --- src/ObjectStore/ObjectStore.php | 66 +++++++++++++++++---------------- 1 file changed, 35 insertions(+), 31 deletions(-) diff --git a/src/ObjectStore/ObjectStore.php b/src/ObjectStore/ObjectStore.php index 33937c4..3e945e7 100644 --- a/src/ObjectStore/ObjectStore.php +++ b/src/ObjectStore/ObjectStore.php @@ -594,28 +594,17 @@ private function getObjectInfo(string $name, bool $showDeleted = false): ObjectI */ private function getObjectInfoRecord(string $name, bool $showDeleted = false): array { - $stream = $this->js->stream($this->streamName()); - - foreach ($this->metaSubjectCandidates($name) as $metaSubject) { - try { - $rawMsg = $stream->getLastMessageForSubject($metaSubject); - } catch (\Throwable) { - continue; - } - - $info = $this->decodeObjectInfo($rawMsg->data); - if ($info->name !== $name) { - continue; - } - - if (!$showDeleted && $info->deleted) { - throw new NatsException("Object not found: {$name}"); - } + $record = $this->findObjectInfoRecord($name); + if ($record === null) { + throw new NatsException("Object not found: {$name}"); + } - return [$info, $metaSubject]; + [$info] = $record; + if (!$showDeleted && $info->deleted) { + throw new NatsException("Object not found: {$name}"); } - throw new NatsException("Object not found: {$name}"); + return $record; } private function encodeName(string $name): string @@ -722,22 +711,14 @@ private function publishObjectInfo(ObjectInfo $info): void private function existingMetaSubjectOrNull(string $name): ?string { - try { - [, $metaSubject] = $this->getObjectInfoRecord($name, showDeleted: true); - return $metaSubject; - } catch (\Throwable) { - return null; - } + $record = $this->findObjectInfoRecord($name); + return $record[1] ?? null; } private function activeObjectExists(string $name): bool { - try { - $this->getObjectInfo($name); - return true; - } catch (NatsException) { - return false; - } + $record = $this->findObjectInfoRecord($name); + return $record !== null && !$record[0]->deleted; } private function purgeLegacyMetaSubjectIfNeeded(string $name, ?string $existingMetaSubject): void @@ -755,6 +736,29 @@ private function purgeLegacyMetaSubjectIfNeeded(string $name, ?string $existingM } } + /** + * @return array{ObjectInfo, string}|null + */ + private function findObjectInfoRecord(string $name): ?array + { + $stream = $this->js->stream($this->streamName()); + + foreach ($this->metaSubjectCandidates($name) as $metaSubject) { + try { + $rawMsg = $stream->getLastMessageForSubject($metaSubject); + } catch (\Throwable) { + continue; + } + + $info = $this->decodeObjectInfo($rawMsg->data); + if ($info->name === $name) { + return [$info, $metaSubject]; + } + } + + return null; + } + /** * @return int<1, max> */ From a0d24780cd4fe7982e7553bb5598adee2e7b870d Mon Sep 17 00:00:00 2001 From: Your Name Date: Mon, 30 Mar 2026 15:34:13 +0200 Subject: [PATCH 4/5] refactor: add non-throwing stream message lookup --- src/JetStream/Stream/Stream.php | 14 ++++++++++++++ src/JetStream/Stream/StreamInterface.php | 2 ++ src/ObjectStore/ObjectStore.php | 5 ++--- tests/Integration/JetStream/StreamTest.php | 17 +++++++++++++++++ 4 files changed, 35 insertions(+), 3 deletions(-) diff --git a/src/JetStream/Stream/Stream.php b/src/JetStream/Stream/Stream.php index cf85390..c4109ab 100644 --- a/src/JetStream/Stream/Stream.php +++ b/src/JetStream/Stream/Stream.php @@ -11,6 +11,7 @@ use Sofascore\Nats\JetStream\Consumer\ConsumerPauseResponse; use Sofascore\Nats\JetStream\Consumer\OrderedConsumer; use Sofascore\Nats\JetStream\Consumer\OrderedConsumerConfig; +use Sofascore\Nats\JetStream\Error\JetStreamException; use Sofascore\Nats\JetStream\Consumer\PushConsumerInterface; use Sofascore\Nats\JetStream\JetStreamContext; @@ -170,6 +171,19 @@ public function getLastMessageForSubject(string $subject): RawStreamMessage return RawStreamMessage::fromArray(T::stringKeyArray($data['message'] ?? [])); } + public function tryGetLastMessageForSubject(string $subject): ?RawStreamMessage + { + try { + return $this->getLastMessageForSubject($subject); + } catch (JetStreamException $e) { + if ($e->apiError?->errCode === 10037) { + return null; + } + + throw $e; + } + } + public function deleteMessage(int $sequence): void { $this->js->apiRequest( diff --git a/src/JetStream/Stream/StreamInterface.php b/src/JetStream/Stream/StreamInterface.php index b9af97e..7b4cd54 100644 --- a/src/JetStream/Stream/StreamInterface.php +++ b/src/JetStream/Stream/StreamInterface.php @@ -46,6 +46,8 @@ public function getMessage(int $sequence): RawStreamMessage; public function getLastMessageForSubject(string $subject): RawStreamMessage; + public function tryGetLastMessageForSubject(string $subject): ?RawStreamMessage; + public function directGet(string $subject, ?int $sequence = null): RawStreamMessage; public function deleteMessage(int $sequence): void; diff --git a/src/ObjectStore/ObjectStore.php b/src/ObjectStore/ObjectStore.php index 3e945e7..e75f76d 100644 --- a/src/ObjectStore/ObjectStore.php +++ b/src/ObjectStore/ObjectStore.php @@ -744,9 +744,8 @@ private function findObjectInfoRecord(string $name): ?array $stream = $this->js->stream($this->streamName()); foreach ($this->metaSubjectCandidates($name) as $metaSubject) { - try { - $rawMsg = $stream->getLastMessageForSubject($metaSubject); - } catch (\Throwable) { + $rawMsg = $stream->tryGetLastMessageForSubject($metaSubject); + if ($rawMsg === null) { continue; } diff --git a/tests/Integration/JetStream/StreamTest.php b/tests/Integration/JetStream/StreamTest.php index 9d66716..c7887b2 100644 --- a/tests/Integration/JetStream/StreamTest.php +++ b/tests/Integration/JetStream/StreamTest.php @@ -107,6 +107,23 @@ public function testGetLastMessageForSubject(): void self::assertSame(2, $rawMsg->sequence); } + public function testTryGetLastMessageForSubject(): void + { + $stream = $this->js->createStream(new StreamConfig( + name: 'TEST_STREAM', + subjects: ['test.stream.>'], + storage: StorageType::Memory, + )); + + self::assertNull($stream->tryGetLastMessageForSubject('test.stream.missing')); + + $this->js->publish('test.stream.foo', 'present'); + + $rawMsg = $stream->tryGetLastMessageForSubject('test.stream.foo'); + self::assertNotNull($rawMsg); + self::assertSame('present', $rawMsg->data); + } + public function testDeleteMessage(): void { $stream = $this->js->createStream(new StreamConfig( From 83444bda1305ec28d7c87b6c9fbc4a388814f3d9 Mon Sep 17 00:00:00 2001 From: Your Name Date: Mon, 30 Mar 2026 15:44:00 +0200 Subject: [PATCH 5/5] refactor: tighten object result cleanup --- src/JetStream/Stream/Stream.php | 2 +- src/ObjectStore/ObjectResult.php | 26 +++++++++++++-- src/ObjectStore/ObjectStore.php | 11 +------ tests/Unit/ObjectStore/ObjectResultTest.php | 35 +++++++++++++++++++-- 4 files changed, 57 insertions(+), 17 deletions(-) diff --git a/src/JetStream/Stream/Stream.php b/src/JetStream/Stream/Stream.php index c4109ab..87df49a 100644 --- a/src/JetStream/Stream/Stream.php +++ b/src/JetStream/Stream/Stream.php @@ -11,8 +11,8 @@ use Sofascore\Nats\JetStream\Consumer\ConsumerPauseResponse; use Sofascore\Nats\JetStream\Consumer\OrderedConsumer; use Sofascore\Nats\JetStream\Consumer\OrderedConsumerConfig; -use Sofascore\Nats\JetStream\Error\JetStreamException; use Sofascore\Nats\JetStream\Consumer\PushConsumerInterface; +use Sofascore\Nats\JetStream\Error\JetStreamException; use Sofascore\Nats\JetStream\JetStreamContext; final class Stream implements StreamInterface diff --git a/src/ObjectStore/ObjectResult.php b/src/ObjectStore/ObjectResult.php index 4a0bb42..4bff568 100644 --- a/src/ObjectStore/ObjectResult.php +++ b/src/ObjectStore/ObjectResult.php @@ -14,7 +14,7 @@ final class ObjectResult private bool $sourceExhausted = false; private int $bytesRead = 0; private int $chunksRead = 0; - private readonly \HashContext $hashContext; + private ?\HashContext $hashContext; /** @var null|\Generator */ private ?\Generator $source = null; @@ -139,7 +139,12 @@ private function nextChunk(): ?string $this->source->next(); $this->bytesRead += strlen($chunk); $this->chunksRead++; - hash_update($this->hashContext, $chunk); + $hashContext = $this->hashContext; + if ($hashContext === null) { + throw new NatsException("Object digest state unavailable: {$this->info->name}"); + } + + hash_update($hashContext, $chunk); return $chunk; } catch (\Throwable $error) { $this->close(); @@ -153,10 +158,20 @@ private function finishSource(): void return; } - $this->assertCompleteRead(); + $error = null; + try { + $this->assertCompleteRead(); + } catch (\Throwable $e) { + $error = $e; + } + $this->sourceExhausted = true; $this->source = null; $this->runCleanup(); + + if ($error !== null) { + throw $error; + } } private function assertCompleteRead(): void @@ -172,7 +187,12 @@ private function assertCompleteRead(): void return; } + if ($this->hashContext === null) { + throw new NatsException("Object digest state unavailable: {$this->info->name}"); + } + $actualDigest = 'SHA-256=' . base64_encode(hash_final($this->hashContext, true)); + $this->hashContext = null; if ($actualDigest !== $this->info->digest) { throw new NatsException("Object digest mismatch: {$this->info->name}"); } diff --git a/src/ObjectStore/ObjectStore.php b/src/ObjectStore/ObjectStore.php index e75f76d..8b7e6e8 100644 --- a/src/ObjectStore/ObjectStore.php +++ b/src/ObjectStore/ObjectStore.php @@ -149,16 +149,7 @@ public function putStream(string $name, mixed $stream): ObjectInfo ); $this->publishObjectInfo($info); - - if ($existingMetaSubject !== null && $existingMetaSubject !== $this->metaSubject($name)) { - try { - $jsStream->purge(new \Sofascore\Nats\JetStream\Stream\StreamPurgeOptions( - filter: $existingMetaSubject, - )); - } catch (\Throwable) { - // Best effort cleanup for legacy metadata subjects. - } - } + $this->purgeLegacyMetaSubjectIfNeeded($name, $existingMetaSubject); } catch (\Throwable $e) { try { $jsStream->purge(new \Sofascore\Nats\JetStream\Stream\StreamPurgeOptions( diff --git a/tests/Unit/ObjectStore/ObjectResultTest.php b/tests/Unit/ObjectStore/ObjectResultTest.php index cdee8a9..3313eda 100644 --- a/tests/Unit/ObjectStore/ObjectResultTest.php +++ b/tests/Unit/ObjectStore/ObjectResultTest.php @@ -82,17 +82,46 @@ function () use (&$cleanupCalls): void { public function testReadAllThrowsWhenStreamEndsBeforeExpectedBytes(): void { + $cleanupCalls = 0; $result = new ObjectResult( $this->createInfo(data: 'hello world', chunks: 3), static function (): \Generator { yield 'hello'; }, + function () use (&$cleanupCalls): void { + $cleanupCalls++; + }, ); - $this->expectException(NatsException::class); - $this->expectExceptionMessage('Object read incomplete'); + try { + $this->expectException(NatsException::class); + $this->expectExceptionMessage('Object read incomplete'); + $result->readAll(); + } finally { + self::assertSame(1, $cleanupCalls); + } + } - $result->readAll(); + public function testDigestMismatchRunsCleanupOnce(): void + { + $cleanupCalls = 0; + $result = new ObjectResult( + $this->createInfo(data: 'hello', chunks: 1, digest: 'SHA-256=invalid'), + static function (): \Generator { + yield 'hello'; + }, + function () use (&$cleanupCalls): void { + $cleanupCalls++; + }, + ); + + try { + $this->expectException(NatsException::class); + $this->expectExceptionMessage('Object digest mismatch'); + $result->readAll(); + } finally { + self::assertSame(1, $cleanupCalls); + } } public function testReadAllThrowsOnDigestMismatch(): void