From cffd30a1a15f7bb555d3a724d481d1c07cc7d4d9 Mon Sep 17 00:00:00 2001 From: Jake Barnby Date: Thu, 27 Aug 2026 13:46:08 +1200 Subject: [PATCH 1/3] (feat): Add canonical dedicated database usage samples --- CHANGELOG.md | 11 + README.md | 66 +++- src/Usage/Adapter.php | 23 ++ src/Usage/Adapter/ClickHouse.php | 316 ++++++++++++++++++- src/Usage/Sample.php | 120 +++++++ src/Usage/SampleGap.php | 17 + src/Usage/SampleRange.php | 61 ++++ src/Usage/SampleResult.php | 72 +++++ src/Usage/Usage.php | 18 ++ tests/Usage/Adapter/ClickHouseSampleTest.php | 246 +++++++++++++++ tests/Usage/SampleTest.php | 71 +++++ 11 files changed, 1013 insertions(+), 8 deletions(-) create mode 100644 src/Usage/Sample.php create mode 100644 src/Usage/SampleGap.php create mode 100644 src/Usage/SampleRange.php create mode 100644 src/Usage/SampleResult.php create mode 100644 tests/Usage/Adapter/ClickHouseSampleTest.php create mode 100644 tests/Usage/SampleTest.php diff --git a/CHANGELOG.md b/CHANGELOG.md index 776ea8b..7ce61eb 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -2,6 +2,17 @@ ## Unreleased — query 0.6.x builder +### Added + +- Added a separate immutable ClickHouse sample ledger for billable usage. A + canonical identity covers environment, region, project/database internal + IDs, member, generation, sequence and metric. Identical retries are + deduplicated at read time; conflicting payloads, sequence gaps, bounded-read + truncation and stable-watermark exclusions are explicit in `SampleResult`. +- Added `Usage::addSamples()`, `Usage::getSampleWatermark()` and + `Usage::findSamples()`. Existing events, gauges and daily rollups are + unchanged and are not used as a canonical sample source. + ### Breaking - Bumped `utopia-php/query` from `0.1.*` to `0.6.*`. diff --git a/README.md b/README.md index 9272193..64111a2 100644 --- a/README.md +++ b/README.md @@ -185,6 +185,69 @@ $usage->addBatch([ ], Usage::TYPE_GAUGE); ``` +### Canonical Samples + +Billable inputs that must survive request retries use the separate immutable +sample ledger. A sample's identity is derived from its environment, region, +project and database internal IDs, member, generation, sequence and metric. +Retrying the same identity and payload is safe. Reusing an identity with a +different interval, value or event version is returned as a conflict. +The event ID is SHA-256 over those identity fields in the documented order, +each encoded as its decimal byte length, `:`, then its UTF-8 value. The payload +hash uses the same encoding over event ID, UTC millisecond interval bounds, +value and event version. + +```php +use Utopia\Usage\Sample; +use Utopia\Usage\SampleRange; + +$sample = new Sample( + environment: 'production', + region: 'fra1', + projectInternalId: '101', + databaseInternalId: '202', + member: 'mysql-0', + generation: '01J...', + sequence: 42, + metric: 'bandwidth.inbound', + intervalStart: new DateTimeImmutable('2026-08-01T00:42:00Z'), + intervalEnd: new DateTimeImmutable('2026-08-01T00:43:00Z'), + value: 4096, + eventVersion: 1, +); + +$usage->addSamples([$sample]); +$watermark = $usage->getSampleWatermark(); +$result = $usage->findSamples( + new SampleRange( + environment: 'production', + region: 'fra1', + projectInternalId: '101', + databaseInternalId: '202', + member: 'mysql-0', + generation: '01J...', + metric: 'bandwidth.inbound', + firstSequence: 42, + lastSequence: 42, + intervalStart: new DateTimeImmutable('2026-08-01T00:42:00Z'), + intervalEnd: new DateTimeImmutable('2026-08-01T00:43:00Z'), + ), + $watermark, + limit: 100, +); + +if (!$result->isComplete()) { + // Refuse billing. Inspect conflicts, compact gap ranges and truncation. +} +``` + +`findSamples()` is bounded by an explicit row limit and reads no rows ingested +after the supplied ClickHouse watermark. A result is complete only when it is +not truncated and contains no conflicts, sequence gaps or interval-boundary +discontinuities. The sample ledger does not make HTTP delivery or a producer's +local spool durable; callers must retain a sample until the write is +acknowledged and retry the identical payload. + ## Querying Metrics ### Find with Query Objects @@ -296,6 +359,7 @@ $usage->purge('project_123', [], Usage::TYPE_GAUGE); | `{ns}_usage_gauges` | MergeTree | Resource snapshot gauges | | `{ns}_usage_events_daily` | SummingMergeTree | Pre-aggregated daily event totals | | `{ns}_usage_events_daily_mv` | Materialized View | Auto-populates daily table on insert | +| `{ns}_usage_samples` | MergeTree | Immutable canonical samples with retry/conflict evidence | ### Events Table Schema @@ -378,7 +442,7 @@ coroutines. ## System Requirements -Utopia Framework requires PHP 8.0 or later. We recommend using the latest PHP version whenever possible. +Utopia Framework requires PHP 8.4 or later. We recommend using the latest PHP version whenever possible. ## Copyright and license diff --git a/src/Usage/Adapter.php b/src/Usage/Adapter.php index 9fd6fea..55a6f28 100644 --- a/src/Usage/Adapter.php +++ b/src/Usage/Adapter.php @@ -37,6 +37,29 @@ abstract public function setup(): void; */ abstract public function addBatch(array $metrics, string $type, int $batchSize = 1000): bool; + /** + * Add immutable, canonically identified usage samples. + * + * Adapters that do not provide canonical sample storage leave this + * unsupported. Existing telemetry APIs are unaffected. + * + * @param list $samples + */ + public function addSamples(array $samples, int $batchSize = 1000): bool + { + throw new \Exception($this->getName() . ' does not support canonical samples'); + } + + public function getSampleWatermark(): \DateTimeImmutable + { + throw new \Exception($this->getName() . ' does not support canonical samples'); + } + + public function findSamples(SampleRange $range, \DateTimeImmutable $watermark, int $limit): SampleResult + { + throw new \Exception($this->getName() . ' does not support canonical samples'); + } + /** * Get time series data for metrics with query-time aggregation. * diff --git a/src/Usage/Adapter/ClickHouse.php b/src/Usage/Adapter/ClickHouse.php index c905006..2596dda 100644 --- a/src/Usage/Adapter/ClickHouse.php +++ b/src/Usage/Adapter/ClickHouse.php @@ -4,6 +4,7 @@ use ArrayObject; use DateTime; +use DateTimeImmutable; use DateTimeZone; use Exception; use Psr\Http\Client\ClientInterface; @@ -19,6 +20,10 @@ use Utopia\Query\Schema\ClickHouse\Engine; use Utopia\Query\Schema\Table\ClickHouse as ClickHouseTable; use Utopia\Usage\Metric; +use Utopia\Usage\Sample; +use Utopia\Usage\SampleGap; +use Utopia\Usage\SampleRange; +use Utopia\Usage\SampleResult; use Utopia\Usage\Usage; use Utopia\Usage\UsageQuery; use Utopia\Validator\Hostname; @@ -55,6 +60,8 @@ class ClickHouse extends SQL private const INSERT_BATCH_SIZE = 1_000; + private const int SAMPLE_BATCH_SIZE = 1_000; + private const ROUTE_LOG_MAX = 1_000; /** @var array Maps interval strings to ClickHouse time functions */ @@ -433,6 +440,11 @@ private function getEventsDailyTableName(): string return $this->getTableName() . '_events_daily'; } + private function getSamplesTableName(): string + { + return $this->getTableName() . '_samples'; + } + /** * Get the appropriate table name for a given type. * @@ -616,7 +628,7 @@ private function buildHeaders(): array * @param array $data Array of JSON strings (one per row) * @throws Exception */ - private function insert(string $table, string $sql, array $data): void + private function insert(string $table, string $sql, array $data, bool $durable = false): void { if (empty($data)) { return; @@ -632,7 +644,7 @@ private function insert(string $table, string $sql, array $data): void $queryParams = ['query' => $sql]; if ($this->asyncInserts) { $queryParams['async_insert'] = '1'; - $queryParams['wait_for_async_insert'] = $this->asyncInsertWait ? '1' : '0'; + $queryParams['wait_for_async_insert'] = ($durable || $this->asyncInsertWait) ? '1' : '0'; } $url = "{$scheme}://{$this->host}:{$this->port}/?" . http_build_query($queryParams); @@ -920,7 +932,6 @@ public function setup(): void $createDbSql = "CREATE DATABASE IF NOT EXISTS {$escapedDatabase}"; $this->query($createDbSql); - // --- Events table --- $this->createTable( $this->getEventsTableName(), 'event', @@ -931,15 +942,12 @@ public function setup(): void $this->applyRetention($this->getEventsTableName()); - // --- Events daily table (SummingMergeTree) --- $this->createDailyTable(); $this->applyRetention($this->getEventsDailyTableName()); - // --- Events daily materialized view --- $this->createDailyMaterializedView(); - // --- Gauges table --- $this->createTable( $this->getGaugesTableName(), 'gauge', @@ -948,7 +956,8 @@ public function setup(): void $this->ensureGaugeDimColumns(); - // --- Per-dim projections on the events / gauges base tables --- + $this->createSamplesTable(); + $this->setLightweightMutationProjectionMode($this->getEventsTableName()); foreach (self::EVENT_PROJECTIONS as $projection) { $this->addProjection( @@ -969,6 +978,53 @@ public function setup(): void } } + /** + * Create the immutable canonical-sample ledger. Retries remain as raw + * physical rows; findSamples() groups by the canonical identity and + * exposes conflicting payloads instead of allowing them to affect totals. + */ + private function createSamplesTable(): void + { + $tableName = $this->getSamplesTableName(); + $table = $this->newSchema()->table($tableName); + + $table->rawColumn('`id` String CODEC(ZSTD(3))'); + $table->rawColumn('`payloadHash` String CODEC(ZSTD(3))'); + $table->rawColumn('`environment` LowCardinality(String)'); + $table->rawColumn('`region` LowCardinality(String)'); + $table->rawColumn('`projectInternalId` String CODEC(ZSTD(3))'); + $table->rawColumn('`databaseInternalId` String CODEC(ZSTD(3))'); + $table->rawColumn('`member` String CODEC(ZSTD(3))'); + $table->rawColumn('`generation` String CODEC(ZSTD(3))'); + $table->rawColumn('`sequence` UInt64'); + $table->rawColumn('`metric` LowCardinality(String)'); + $table->rawColumn("`intervalStart` DateTime64(3, 'UTC') CODEC(Delta(4), LZ4)"); + $table->rawColumn("`intervalEnd` DateTime64(3, 'UTC') CODEC(Delta(4), LZ4)"); + $table->rawColumn('`value` Int64'); + $table->rawColumn('`eventVersion` UInt32'); + $table->rawColumn("`ingestedAt` DateTime64(3, 'UTC') DEFAULT now64(3) CODEC(Delta(4), LZ4)"); + + $table->engine(Engine::MergeTree) + ->orderBy([ + 'environment', + 'region', + 'projectInternalId', + 'databaseInternalId', + 'member', + 'generation', + 'metric', + 'sequence', + 'id', + 'payloadHash', + ]) + ->partitionBy('toYYYYMM(intervalStart)') + ->settings(['index_granularity' => 8192]); + + $statement = $table->createIfNotExists(); + + $this->query($this->qualifyDdl($statement->query, $tableName)); + } + /** * Apply (or strip) the retention TTL on a table as a separate idempotent * ALTER. CREATE TABLE IF NOT EXISTS won't add a TTL to an existing table, @@ -1679,6 +1735,252 @@ public function addBatch(array $metrics, string $type, int $batchSize = self::IN return true; } + /** + * @param list $samples + */ + #[\Override] + public function addSamples(array $samples, int $batchSize = self::SAMPLE_BATCH_SIZE): bool + { + if ($samples === []) { + return true; + } + + $this->setOperationContext('addSamples()'); + + $batchSize = min(self::SAMPLE_BATCH_SIZE, max(1, $batchSize)); + $tableName = $this->getSamplesTableName(); + $columns = [ + 'id', + 'payloadHash', + 'environment', + 'region', + 'projectInternalId', + 'databaseInternalId', + 'member', + 'generation', + 'sequence', + 'metric', + 'intervalStart', + 'intervalEnd', + 'value', + 'eventVersion', + ]; + $escapedColumns = implode(', ', array_map($this->escapeIdentifier(...), $columns)); + $insertSql = 'INSERT INTO ' . $this->buildTableReference($tableName) + . " ({$escapedColumns}) FORMAT JSONEachRow"; + + foreach (array_chunk($samples, $batchSize) as $batch) { + $rows = []; + + foreach ($batch as $sample) { + $rows[] = json_encode([ + 'id' => $sample->getId(), + 'payloadHash' => $sample->getPayloadHash(), + 'environment' => $sample->environment, + 'region' => $sample->region, + 'projectInternalId' => $sample->projectInternalId, + 'databaseInternalId' => $sample->databaseInternalId, + 'member' => $sample->member, + 'generation' => $sample->generation, + 'sequence' => $sample->sequence, + 'metric' => $sample->metric, + 'intervalStart' => $sample->getFormattedIntervalStart(), + 'intervalEnd' => $sample->getFormattedIntervalEnd(), + 'value' => $sample->value, + 'eventVersion' => $sample->eventVersion, + ], JSON_THROW_ON_ERROR); + } + + $this->insert($tableName, $insertSql, $rows, durable: true); + } + + return true; + } + + #[\Override] + public function getSampleWatermark(): DateTimeImmutable + { + $this->setOperationContext('getSampleWatermark()'); + + $rows = $this->decodeRows($this->query("SELECT toString(now64(3, 'UTC')) AS watermark FORMAT JSON")); + $watermark = self::toStr($rows[0]['watermark'] ?? null); + + if ($watermark === '') { + throw new Exception('ClickHouse did not return a sample watermark'); + } + + return new DateTimeImmutable($watermark, new DateTimeZone('UTC')); + } + + #[\Override] + public function findSamples(SampleRange $range, DateTimeImmutable $watermark, int $limit): SampleResult + { + if ($limit < 1 || $limit === PHP_INT_MAX) { + throw new \InvalidArgumentException('Sample limit must be positive and leave room for truncation detection'); + } + + $this->setOperationContext('findSamples()'); + + $table = $this->buildTableReference($this->getSamplesTableName()); + $sql = <<= {firstSequence:UInt64} + AND sequence <= {lastSequence:UInt64} + AND ingestedAt <= {watermark:DateTime64(3)} + GROUP BY + environment, + region, + projectInternalId, + databaseInternalId, + member, + generation, + sequence, + metric + ORDER BY sequence ASC + LIMIT {queryLimit:UInt64} + FORMAT JSON + SQL; + + $rows = $this->decodeRows($this->query($sql, [ + 'environment' => $range->environment, + 'region' => $range->region, + 'projectInternalId' => $range->projectInternalId, + 'databaseInternalId' => $range->databaseInternalId, + 'member' => $range->member, + 'generation' => $range->generation, + 'metric' => $range->metric, + 'firstSequence' => $range->firstSequence, + 'lastSequence' => $range->lastSequence, + 'watermark' => $watermark->setTimezone(new DateTimeZone('UTC'))->format('Y-m-d H:i:s.v'), + 'queryLimit' => $limit + 1, + ])); + + $truncated = count($rows) > $limit; + if ($truncated) { + $rows = array_slice($rows, 0, $limit); + } + + $samples = []; + $conflicts = []; + $duplicates = 0; + + foreach ($rows as $row) { + $sequence = self::toInt($row['sequence'] ?? null); + $copies = self::toInt($row['copies'] ?? null); + $variants = self::toInt($row['variants'] ?? null); + $intervalStart = new DateTimeImmutable(self::toStr($row['intervalStart'] ?? null), new DateTimeZone('UTC')); + $intervalEnd = new DateTimeImmutable(self::toStr($row['intervalEnd'] ?? null), new DateTimeZone('UTC')); + + if ( + $variants !== 1 + || $intervalStart < $range->intervalStart + || $intervalEnd > $range->intervalEnd + ) { + $conflicts[] = $sequence; + } + + $duplicates += max(0, $copies - max(1, $variants)); + $samples[] = new Sample( + environment: self::toStr($row['environment'] ?? null), + region: self::toStr($row['region'] ?? null), + projectInternalId: self::toStr($row['projectInternalId'] ?? null), + databaseInternalId: self::toStr($row['databaseInternalId'] ?? null), + member: self::toStr($row['member'] ?? null), + generation: self::toStr($row['generation'] ?? null), + sequence: $sequence, + metric: self::toStr($row['metric'] ?? null), + intervalStart: $intervalStart, + intervalEnd: $intervalEnd, + value: self::toInt($row['value'] ?? null), + eventVersion: self::toInt($row['eventVersion'] ?? null), + ); + } + + $gaps = $this->findSampleGaps($samples, $range->firstSequence, $range->lastSequence); + $discontinuities = $this->findSampleDiscontinuities($samples, $range); + + return new SampleResult( + samples: $samples, + conflicts: array_values(array_unique($conflicts)), + gaps: $gaps, + discontinuities: $discontinuities, + duplicateCount: $duplicates, + truncated: $truncated, + watermark: $watermark, + ); + } + + /** + * @param list $samples + * @return list + */ + private function findSampleGaps(array $samples, int $first, int $last): array + { + $gaps = []; + $expected = $first; + + foreach ($samples as $sample) { + if ($sample->sequence > $expected) { + $gaps[] = new SampleGap($expected, $sample->sequence - 1); + } + + $expected = $sample->sequence + 1; + } + + if ($expected <= $last) { + $gaps[] = new SampleGap($expected, $last); + } + + return $gaps; + } + + /** + * @param list $samples + * @return list + */ + private function findSampleDiscontinuities(array $samples, SampleRange $range): array + { + $discontinuities = []; + $expectedStart = $range->intervalStart; + + foreach ($samples as $sample) { + if ($sample->intervalStart != $expectedStart) { + $discontinuities[] = $sample->sequence; + } + + $expectedStart = $sample->intervalEnd; + } + + if ($samples !== [] && $expectedStart != $range->intervalEnd) { + $last = $samples[array_key_last($samples)]; + $discontinuities[] = $last->sequence; + } + + return array_values(array_unique($discontinuities)); + } + /** * Columns declared in the INSERT envelope for the given type. Matches * the row shape produced by addBatch(): base columns, the type's diff --git a/src/Usage/Sample.php b/src/Usage/Sample.php new file mode 100644 index 0000000..c44a8ac --- /dev/null +++ b/src/Usage/Sample.php @@ -0,0 +1,120 @@ + $environment, + 'region' => $region, + 'projectInternalId' => $projectInternalId, + 'databaseInternalId' => $databaseInternalId, + 'member' => $member, + 'generation' => $generation, + 'metric' => $metric, + ] as $field => $value) { + if ($value === '') { + throw new InvalidArgumentException("{$field} cannot be empty"); + } + } + + if ($sequence < 0) { + throw new InvalidArgumentException('sequence cannot be negative'); + } + + if ($eventVersion < 1 || $eventVersion > 4_294_967_295) { + throw new InvalidArgumentException('eventVersion must fit an unsigned 32-bit integer'); + } + + if ($intervalStart >= $intervalEnd) { + throw new InvalidArgumentException('intervalStart must be before intervalEnd'); + } + } + + /** + * Canonical stream identity. A retry of one observation must retain this + * ID even if a faulty producer changes its payload, so readers can expose + * the conflict rather than counting both values. + * + */ + public function getId(): string + { + return $this->hashParts([ + $this->environment, + $this->region, + $this->projectInternalId, + $this->databaseInternalId, + $this->member, + $this->generation, + $this->sequence, + $this->metric, + ]); + } + + /** + * Hash every money-bearing field. Equal IDs with different payload hashes + * are conflicting observations and make the stream incomplete. + * + */ + public function getPayloadHash(): string + { + return $this->hashParts([ + $this->getId(), + $this->formatDateTime($this->intervalStart), + $this->formatDateTime($this->intervalEnd), + $this->value, + $this->eventVersion, + ]); + } + + public function getFormattedIntervalStart(): string + { + return $this->formatDateTime($this->intervalStart); + } + + public function getFormattedIntervalEnd(): string + { + return $this->formatDateTime($this->intervalEnd); + } + + private function formatDateTime(DateTimeImmutable $time): string + { + return $time->setTimezone(new DateTimeZone('UTC'))->format('Y-m-d H:i:s.v'); + } + + /** + * Length-prefixing makes the digest unambiguous and reproducible by + * producers in other languages without depending on JSON encoding rules. + * + * @param list $parts + */ + private function hashParts(array $parts): string + { + $encoded = ''; + + foreach ($parts as $part) { + $value = (string) $part; + $encoded .= strlen($value) . ':' . $value; + } + + return hash('sha256', $encoded); + } +} diff --git a/src/Usage/SampleGap.php b/src/Usage/SampleGap.php new file mode 100644 index 0000000..95f5cc4 --- /dev/null +++ b/src/Usage/SampleGap.php @@ -0,0 +1,17 @@ + $environment, + 'region' => $region, + 'projectInternalId' => $projectInternalId, + 'databaseInternalId' => $databaseInternalId, + 'member' => $member, + 'generation' => $generation, + 'metric' => $metric, + ] as $field => $value) { + if ($value === '') { + throw new InvalidArgumentException("{$field} cannot be empty"); + } + } + + if ($firstSequence < 0 || $lastSequence < $firstSequence) { + throw new InvalidArgumentException('Invalid sample sequence range'); + } + + if ($intervalStart >= $intervalEnd) { + throw new InvalidArgumentException('intervalStart must be before intervalEnd'); + } + } + + public function getFormattedIntervalStart(): string + { + return $this->formatDateTime($this->intervalStart); + } + + public function getFormattedIntervalEnd(): string + { + return $this->formatDateTime($this->intervalEnd); + } + + private function formatDateTime(DateTimeImmutable $time): string + { + return $time->setTimezone(new DateTimeZone('UTC'))->format('Y-m-d H:i:s.v'); + } +} diff --git a/src/Usage/SampleResult.php b/src/Usage/SampleResult.php new file mode 100644 index 0000000..ff5c4cd --- /dev/null +++ b/src/Usage/SampleResult.php @@ -0,0 +1,72 @@ + $samples + * @param list $conflicts + * @param list $gaps + * @param list $discontinuities + */ + public function __construct( + private array $samples, + private array $conflicts, + private array $gaps, + private array $discontinuities, + private int $duplicateCount, + private bool $truncated, + private DateTimeImmutable $watermark, + ) { + } + + /** @return list */ + public function getSamples(): array + { + return $this->samples; + } + + /** @return list */ + public function getConflicts(): array + { + return $this->conflicts; + } + + /** @return list */ + public function getGaps(): array + { + return $this->gaps; + } + + /** @return list */ + public function getDiscontinuities(): array + { + return $this->discontinuities; + } + + public function getDuplicateCount(): int + { + return $this->duplicateCount; + } + + public function isTruncated(): bool + { + return $this->truncated; + } + + public function getWatermark(): DateTimeImmutable + { + return $this->watermark; + } + + public function isComplete(): bool + { + return !$this->truncated + && $this->conflicts === [] + && $this->gaps === [] + && $this->discontinuities === []; + } +} diff --git a/src/Usage/Usage.php b/src/Usage/Usage.php index c70a96e..fc6d86c 100644 --- a/src/Usage/Usage.php +++ b/src/Usage/Usage.php @@ -77,6 +77,24 @@ public function addBatch(array $metrics, string $type, int $batchSize = 1000): b return $this->adapter->addBatch($metrics, $type, $batchSize); } + /** + * @param list $samples + */ + public function addSamples(array $samples, int $batchSize = 1000): bool + { + return $this->adapter->addSamples($samples, $batchSize); + } + + public function getSampleWatermark(): \DateTimeImmutable + { + return $this->adapter->getSampleWatermark(); + } + + public function findSamples(SampleRange $range, \DateTimeImmutable $watermark, int $limit): SampleResult + { + return $this->adapter->findSamples($range, $watermark, $limit); + } + /** * Get time series data for metrics. * diff --git a/tests/Usage/Adapter/ClickHouseSampleTest.php b/tests/Usage/Adapter/ClickHouseSampleTest.php new file mode 100644 index 0000000..e95033d --- /dev/null +++ b/tests/Usage/Adapter/ClickHouseSampleTest.php @@ -0,0 +1,246 @@ +usage = new Usage($adapter); + $this->usage->setup(); + } + + public function testSampleTableHasCanonicalIdentityAndWatermarkColumns(): void + { + $adapter = $this->usage->getAdapter(); + $this->assertInstanceOf(ClickHouseAdapter::class, $adapter); + + $table = $this->resolveTableName($adapter, 'getSamplesTableName'); + $database = $this->databaseName($adapter); + $ddl = $this->queryRaw($adapter, "SHOW CREATE TABLE `{$database}`.`{$table}` FORMAT TabSeparatedRaw"); + + foreach ([ + '`environment` LowCardinality(String)', + '`region` LowCardinality(String)', + '`projectInternalId` String', + '`databaseInternalId` String', + '`member` String', + '`generation` String', + '`sequence` UInt64', + '`metric` LowCardinality(String)', + "`intervalStart` DateTime64(3, 'UTC')", + "`intervalEnd` DateTime64(3, 'UTC')", + '`value` Int64', + '`eventVersion` UInt32', + "`ingestedAt` DateTime64(3, 'UTC') DEFAULT now64(3)", + ] as $column) { + $this->assertStringContainsString($column, $ddl); + } + } + + public function testCanonicalizesConcurrentDuplicatesAndCrashRetry(): void + { + $key = bin2hex(random_bytes(8)); + $sample = $this->sample($key, sequence: 0); + + $this->assertTrue($this->usage->addSamples([$sample, $sample])); + $this->assertTrue($this->usage->addSamples([$sample])); + + $result = $this->usage->findSamples( + $this->range($key, firstSequence: 0, lastSequence: 0), + $this->usage->getSampleWatermark(), + 10, + ); + + $this->assertTrue($result->isComplete()); + $this->assertCount(1, $result->getSamples()); + $this->assertSame(2, $result->getDuplicateCount()); + $this->assertSame([], $result->getConflicts()); + $this->assertSame([], $result->getGaps()); + $this->assertSame([], $result->getDiscontinuities()); + } + + public function testCanonicalSamplesWaitForAsyncInsertDurability(): void + { + $key = bin2hex(random_bytes(8)); + $adapter = new ClickHouseAdapter( + getenv('CLICKHOUSE_HOST') ?: 'clickhouse', + getenv('CLICKHOUSE_USER') ?: 'default', + getenv('CLICKHOUSE_PASSWORD') ?: 'clickhouse', + (int) (getenv('CLICKHOUSE_PORT') ?: 8123), + (bool) (getenv('CLICKHOUSE_SECURE') ?: false), + namespace: 'utopia_usage_samples_async', + database: getenv('CLICKHOUSE_DATABASE') ?: 'default', + sharedTables: true, + asyncInserts: true, + asyncInsertWait: false, + ); + $usage = new Usage($adapter); + $usage->setup(); + + $this->assertTrue($usage->addSamples([$this->sample($key, sequence: 0)])); + + $result = $usage->findSamples( + $this->range($key, firstSequence: 0, lastSequence: 0), + $usage->getSampleWatermark(), + 10, + ); + + $this->assertTrue($result->isComplete()); + $this->assertCount(1, $result->getSamples()); + } + + public function testConflictingDuplicateFailsCompleteness(): void + { + $key = bin2hex(random_bytes(8)); + + $this->assertTrue($this->usage->addSamples([ + $this->sample($key, sequence: 0, value: 10), + $this->sample($key, sequence: 0, value: 11), + ])); + + $result = $this->usage->findSamples( + $this->range($key, firstSequence: 0, lastSequence: 0), + $this->usage->getSampleWatermark(), + 10, + ); + + $this->assertFalse($result->isComplete()); + $this->assertSame([0], $result->getConflicts()); + } + + public function testDetectsGapsWithoutExpandingEveryMissingSequence(): void + { + $key = bin2hex(random_bytes(8)); + + $this->assertTrue($this->usage->addSamples([ + $this->sample($key, sequence: 0), + $this->sample($key, sequence: 4), + ])); + + $result = $this->usage->findSamples( + $this->range($key, firstSequence: 0, lastSequence: 4), + $this->usage->getSampleWatermark(), + 10, + ); + + $this->assertFalse($result->isComplete()); + $this->assertCount(1, $result->getGaps()); + $this->assertSame(1, $result->getGaps()[0]->first); + $this->assertSame(3, $result->getGaps()[0]->last); + } + + public function testDetectsIntervalDiscontinuityWithContiguousSequences(): void + { + $key = bin2hex(random_bytes(8)); + + $this->assertTrue($this->usage->addSamples([ + $this->sample($key, sequence: 0), + $this->sample($key, sequence: 1, startMinute: 2), + ])); + + $result = $this->usage->findSamples( + $this->range($key, firstSequence: 0, lastSequence: 1, endMinute: 3), + $this->usage->getSampleWatermark(), + 10, + ); + + $this->assertFalse($result->isComplete()); + $this->assertSame([], $result->getGaps()); + $this->assertSame([1], $result->getDiscontinuities()); + } + + public function testReportsTruncationAndHonorsAStableWatermark(): void + { + $key = bin2hex(random_bytes(8)); + + $this->assertTrue($this->usage->addSamples([ + $this->sample($key, sequence: 0), + $this->sample($key, sequence: 1), + $this->sample($key, sequence: 2), + ])); + + $watermark = $this->usage->getSampleWatermark(); + usleep(10_000); + $this->assertTrue($this->usage->addSamples([$this->sample($key, sequence: 3)])); + + $bounded = $this->usage->findSamples( + $this->range($key, firstSequence: 0, lastSequence: 3), + $watermark, + 2, + ); + $watermarked = $this->usage->findSamples( + $this->range($key, firstSequence: 0, lastSequence: 3), + $watermark, + 10, + ); + + $this->assertTrue($bounded->isTruncated()); + $this->assertCount(2, $bounded->getSamples()); + $this->assertFalse($bounded->isComplete()); + + $this->assertFalse($watermarked->isTruncated()); + $this->assertCount(3, $watermarked->getSamples()); + $this->assertSame(3, $watermarked->getGaps()[0]->first); + $this->assertSame(3, $watermarked->getGaps()[0]->last); + } + + private function sample(string $key, int $sequence, int $value = 10, ?int $startMinute = null): Sample + { + $start = new DateTimeImmutable('2026-08-01T00:00:00Z'); + $startMinute ??= $sequence; + + return new Sample( + environment: 'test-' . $key, + region: 'fra1', + projectInternalId: '101', + databaseInternalId: '202', + member: 'mysql-0', + generation: 'generation-1', + sequence: $sequence, + metric: 'bandwidth.inbound', + intervalStart: $start->modify("+{$startMinute} minutes"), + intervalEnd: $start->modify('+' . ($startMinute + 1) . ' minutes'), + value: $value, + eventVersion: 1, + ); + } + + private function range(string $key, int $firstSequence, int $lastSequence, ?int $endMinute = null): SampleRange + { + $endMinute ??= $lastSequence + 1; + + return new SampleRange( + environment: 'test-' . $key, + region: 'fra1', + projectInternalId: '101', + databaseInternalId: '202', + member: 'mysql-0', + generation: 'generation-1', + metric: 'bandwidth.inbound', + firstSequence: $firstSequence, + lastSequence: $lastSequence, + intervalStart: new DateTimeImmutable('2026-08-01T00:00:00Z'), + intervalEnd: new DateTimeImmutable("2026-08-01T00:{$endMinute}:00Z"), + ); + } +} diff --git a/tests/Usage/SampleTest.php b/tests/Usage/SampleTest.php new file mode 100644 index 0000000..73e81ee --- /dev/null +++ b/tests/Usage/SampleTest.php @@ -0,0 +1,71 @@ +sample(value: 42); + $retry = $this->sample(value: 42); + $conflict = $this->sample(value: 43); + + $this->assertSame($sample->getId(), $retry->getId()); + $this->assertSame($sample->getPayloadHash(), $retry->getPayloadHash()); + $this->assertSame($sample->getId(), $conflict->getId()); + $this->assertNotSame($sample->getPayloadHash(), $conflict->getPayloadHash()); + $this->assertSame('a8e0eebf28f6eb0e2f632fd59b40734624d5c46a83edd2ba0530b0d83fbf3249', $sample->getId()); + $this->assertSame('dadbc87dbbe5fecba1e96f6c9c608c4ca09447566745c8e3e35bf06f76766568', $sample->getPayloadHash()); + } + + public function testEventVersionChangesPayloadButNotIdentity(): void + { + $first = $this->sample(eventVersion: 1); + $second = $this->sample(eventVersion: 2); + + $this->assertSame($first->getId(), $second->getId()); + $this->assertNotSame($first->getPayloadHash(), $second->getPayloadHash()); + } + + public function testRejectsAnInvalidInterval(): void + { + $this->expectException(\InvalidArgumentException::class); + + new Sample( + environment: 'production', + region: 'fra1', + projectInternalId: '101', + databaseInternalId: '202', + member: 'mysql-0', + generation: 'generation-1', + sequence: 7, + metric: 'bandwidth.inbound', + intervalStart: new DateTimeImmutable('2026-08-01T00:01:00Z'), + intervalEnd: new DateTimeImmutable('2026-08-01T00:00:00Z'), + value: 42, + eventVersion: 1, + ); + } + + private function sample(int $value = 42, int $eventVersion = 1): Sample + { + return new Sample( + environment: 'production', + region: 'fra1', + projectInternalId: '101', + databaseInternalId: '202', + member: 'mysql-0', + generation: 'generation-1', + sequence: 7, + metric: 'bandwidth.inbound', + intervalStart: new DateTimeImmutable('2026-08-01T00:00:00Z'), + intervalEnd: new DateTimeImmutable('2026-08-01T00:01:00Z'), + value: $value, + eventVersion: $eventVersion, + ); + } +} From cae499073fe94f36fb67fc21dd5ae3c4017aa205 Mon Sep 17 00:00:00 2001 From: Jake Barnby Date: Thu, 27 Aug 2026 14:12:46 +1200 Subject: [PATCH 2/3] (fix): Stabilize canonical sample snapshots --- CHANGELOG.md | 4 +- README.md | 51 +++--- src/Usage/Adapter.php | 8 +- src/Usage/Adapter/ClickHouse.php | 153 +++++++++++++---- src/Usage/SampleResult.php | 7 +- src/Usage/SampleWatermark.php | 59 +++++++ src/Usage/Usage.php | 6 +- tests/Usage/Adapter/ClickHouseSampleTest.php | 170 +++++++++++++++++-- tests/Usage/SampleWatermarkTest.php | 58 +++++++ 9 files changed, 434 insertions(+), 82 deletions(-) create mode 100644 src/Usage/SampleWatermark.php create mode 100644 tests/Usage/SampleWatermarkTest.php diff --git a/CHANGELOG.md b/CHANGELOG.md index 7ce61eb..6fd82e6 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -8,7 +8,9 @@ canonical identity covers environment, region, project/database internal IDs, member, generation, sequence and metric. Identical retries are deduplicated at read time; conflicting payloads, sequence gaps, bounded-read - truncation and stable-watermark exclusions are explicit in `SampleResult`. + truncation and exact ingestion-ID watermark exclusions are explicit in + `SampleResult`. Conflict representatives are selected as one physical tuple, + never assembled from independent column aggregates. - Added `Usage::addSamples()`, `Usage::getSampleWatermark()` and `Usage::findSamples()`. Existing events, gauges and daily rollups are unchanged and are not used as a canonical sample source. diff --git a/README.md b/README.md index 64111a2..3de625d 100644 --- a/README.md +++ b/README.md @@ -216,22 +216,24 @@ $sample = new Sample( eventVersion: 1, ); +$range = new SampleRange( + environment: 'production', + region: 'fra1', + projectInternalId: '101', + databaseInternalId: '202', + member: 'mysql-0', + generation: '01J...', + metric: 'bandwidth.inbound', + firstSequence: 42, + lastSequence: 42, + intervalStart: new DateTimeImmutable('2026-08-01T00:42:00Z'), + intervalEnd: new DateTimeImmutable('2026-08-01T00:43:00Z'), +); + $usage->addSamples([$sample]); -$watermark = $usage->getSampleWatermark(); +$watermark = $usage->getSampleWatermark($range, limit: 100); $result = $usage->findSamples( - new SampleRange( - environment: 'production', - region: 'fra1', - projectInternalId: '101', - databaseInternalId: '202', - member: 'mysql-0', - generation: '01J...', - metric: 'bandwidth.inbound', - firstSequence: 42, - lastSequence: 42, - intervalStart: new DateTimeImmutable('2026-08-01T00:42:00Z'), - intervalEnd: new DateTimeImmutable('2026-08-01T00:43:00Z'), - ), + $range, $watermark, limit: 100, ); @@ -241,12 +243,21 @@ if (!$result->isComplete()) { } ``` -`findSamples()` is bounded by an explicit row limit and reads no rows ingested -after the supplied ClickHouse watermark. A result is complete only when it is -not truncated and contains no conflicts, sequence gaps or interval-boundary -discontinuities. The sample ledger does not make HTTP delivery or a producer's -local spool durable; callers must retain a sample until the write is -acknowledged and retry the identical payload. +Each supplied sample row receives an adapter-owned random ingestion ID before +its request is sent. `getSampleWatermark()` performs one bounded ClickHouse +snapshot read and captures the exact IDs visible for that stream and range. +`findSamples()` admits only those IDs, so later inserts cannot cross the +boundary even when their server timestamps would be identical. A transport +retry of the same request retains its IDs and is counted once; a new logical +retry gets a new ID and is included only when visible to the watermark query. + +Both the watermark evidence and `findSamples()` result are explicitly bounded. +A result is complete only when neither bound is truncated and there are no +conflicts, sequence gaps or interval-boundary discontinuities. Conflicting +physical rows are never combined into a synthetic sample. The sample ledger +does not make HTTP delivery or a producer's local spool durable; callers must +retain a sample until the write is acknowledged and retry the identical +payload. ## Querying Metrics diff --git a/src/Usage/Adapter.php b/src/Usage/Adapter.php index 55a6f28..b3473e1 100644 --- a/src/Usage/Adapter.php +++ b/src/Usage/Adapter.php @@ -50,12 +50,16 @@ public function addSamples(array $samples, int $batchSize = 1000): bool throw new \Exception($this->getName() . ' does not support canonical samples'); } - public function getSampleWatermark(): \DateTimeImmutable + /** + * Capture at most $limit logical ingestion IDs from one exact range. + * Truncation is carried by the returned watermark and fails completeness. + */ + public function getSampleWatermark(SampleRange $range, int $limit): SampleWatermark { throw new \Exception($this->getName() . ' does not support canonical samples'); } - public function findSamples(SampleRange $range, \DateTimeImmutable $watermark, int $limit): SampleResult + public function findSamples(SampleRange $range, SampleWatermark $watermark, int $limit): SampleResult { throw new \Exception($this->getName() . ' does not support canonical samples'); } diff --git a/src/Usage/Adapter/ClickHouse.php b/src/Usage/Adapter/ClickHouse.php index 2596dda..2d191ef 100644 --- a/src/Usage/Adapter/ClickHouse.php +++ b/src/Usage/Adapter/ClickHouse.php @@ -24,6 +24,7 @@ use Utopia\Usage\SampleGap; use Utopia\Usage\SampleRange; use Utopia\Usage\SampleResult; +use Utopia\Usage\SampleWatermark; use Utopia\Usage\Usage; use Utopia\Usage\UsageQuery; use Utopia\Validator\Hostname; @@ -957,6 +958,7 @@ public function setup(): void $this->ensureGaugeDimColumns(); $this->createSamplesTable(); + $this->ensureSampleColumns(); $this->setLightweightMutationProjectionMode($this->getEventsTableName()); foreach (self::EVENT_PROJECTIONS as $projection) { @@ -990,6 +992,7 @@ private function createSamplesTable(): void $table->rawColumn('`id` String CODEC(ZSTD(3))'); $table->rawColumn('`payloadHash` String CODEC(ZSTD(3))'); + $table->rawColumn('`ingestId` String CODEC(ZSTD(3))'); $table->rawColumn('`environment` LowCardinality(String)'); $table->rawColumn('`region` LowCardinality(String)'); $table->rawColumn('`projectInternalId` String CODEC(ZSTD(3))'); @@ -1002,7 +1005,6 @@ private function createSamplesTable(): void $table->rawColumn("`intervalEnd` DateTime64(3, 'UTC') CODEC(Delta(4), LZ4)"); $table->rawColumn('`value` Int64'); $table->rawColumn('`eventVersion` UInt32'); - $table->rawColumn("`ingestedAt` DateTime64(3, 'UTC') DEFAULT now64(3) CODEC(Delta(4), LZ4)"); $table->engine(Engine::MergeTree) ->orderBy([ @@ -1016,6 +1018,7 @@ private function createSamplesTable(): void 'sequence', 'id', 'payloadHash', + 'ingestId', ]) ->partitionBy('toYYYYMM(intervalStart)') ->settings(['index_granularity' => 8192]); @@ -1025,6 +1028,16 @@ private function createSamplesTable(): void $this->query($this->qualifyDdl($statement->query, $tableName)); } + /** + * Older pre-release tables have no exact snapshot identifier. Keep their + * default empty so watermark reads omit them and fail closed with gaps. + */ + private function ensureSampleColumns(): void + { + $table = $this->buildTableReference($this->getSamplesTableName()); + $this->query("ALTER TABLE {$table} ADD COLUMN IF NOT EXISTS `ingestId` String DEFAULT '' CODEC(ZSTD(3))"); + } + /** * Apply (or strip) the retention TTL on a table as a separate idempotent * ALTER. CREATE TABLE IF NOT EXISTS won't add a TTL to an existing table, @@ -1752,6 +1765,7 @@ public function addSamples(array $samples, int $batchSize = self::SAMPLE_BATCH_S $columns = [ 'id', 'payloadHash', + 'ingestId', 'environment', 'region', 'projectInternalId', @@ -1773,9 +1787,14 @@ public function addSamples(array $samples, int $batchSize = self::SAMPLE_BATCH_S $rows = []; foreach ($batch as $sample) { + // The ID belongs to this logical row, not its HTTP attempt. + // A transport retry repeats the encoded body and therefore + // keeps the same ID; a later addSamples() call receives a new + // one and cannot cross an already captured watermark. $rows[] = json_encode([ 'id' => $sample->getId(), 'payloadHash' => $sample->getPayloadHash(), + 'ingestId' => bin2hex(random_bytes(16)), 'environment' => $sample->environment, 'region' => $sample->region, 'projectInternalId' => $sample->projectInternalId, @@ -1798,27 +1817,70 @@ public function addSamples(array $samples, int $batchSize = self::SAMPLE_BATCH_S } #[\Override] - public function getSampleWatermark(): DateTimeImmutable + public function getSampleWatermark(SampleRange $range, int $limit): SampleWatermark { + if ($limit < 1 || $limit === PHP_INT_MAX) { + throw new \InvalidArgumentException('Sample watermark limit must be positive and leave room for truncation detection'); + } + $this->setOperationContext('getSampleWatermark()'); - $rows = $this->decodeRows($this->query("SELECT toString(now64(3, 'UTC')) AS watermark FORMAT JSON")); - $watermark = self::toStr($rows[0]['watermark'] ?? null); + $table = $this->buildTableReference($this->getSamplesTableName()); + $sql = <<= {firstSequence:UInt64} + AND sequence <= {lastSequence:UInt64} + AND ingestId != '' + LIMIT 1 BY ingestId + LIMIT {queryLimit:UInt64} + FORMAT JSON + SQL; - if ($watermark === '') { - throw new Exception('ClickHouse did not return a sample watermark'); + $rows = $this->decodeRows($this->query($sql, [ + 'environment' => $range->environment, + 'region' => $range->region, + 'projectInternalId' => $range->projectInternalId, + 'databaseInternalId' => $range->databaseInternalId, + 'member' => $range->member, + 'generation' => $range->generation, + 'metric' => $range->metric, + 'firstSequence' => $range->firstSequence, + 'lastSequence' => $range->lastSequence, + 'queryLimit' => $limit + 1, + ])); + + $truncated = count($rows) > $limit; + if ($truncated) { + $rows = array_slice($rows, 0, $limit); } - return new DateTimeImmutable($watermark, new DateTimeZone('UTC')); + $ingestIds = []; + foreach ($rows as $row) { + $ingestIds[] = self::toStr($row['ingestId'] ?? null); + } + + return new SampleWatermark($range, $ingestIds, $truncated); } #[\Override] - public function findSamples(SampleRange $range, DateTimeImmutable $watermark, int $limit): SampleResult + public function findSamples(SampleRange $range, SampleWatermark $watermark, int $limit): SampleResult { if ($limit < 1 || $limit === PHP_INT_MAX) { throw new \InvalidArgumentException('Sample limit must be positive and leave room for truncation detection'); } + if (!$watermark->matches($range)) { + throw new \InvalidArgumentException('Sample watermark does not match the requested range'); + } + $this->setOperationContext('findSamples()'); $table = $this->buildTableReference($this->getSamplesTableName()); @@ -1832,12 +1894,12 @@ public function findSamples(SampleRange $range, DateTimeImmutable $watermark, in generation, sequence, metric, - any(intervalStart) AS intervalStart, - any(intervalEnd) AS intervalEnd, - any(value) AS value, - any(eventVersion) AS eventVersion, - count() AS copies, - uniqExact(payloadHash) AS variants + argMin( + tuple(intervalStart, intervalEnd, value, eventVersion, payloadHash), + tuple(payloadHash, intervalStart, intervalEnd, value, eventVersion, ingestId) + ) AS observation, + uniqExact(ingestId) AS copies, + uniqExact(tuple(payloadHash, intervalStart, intervalEnd, value, eventVersion)) AS variants FROM {$table} WHERE environment = {environment:String} AND region = {region:String} @@ -1848,7 +1910,7 @@ public function findSamples(SampleRange $range, DateTimeImmutable $watermark, in AND metric = {metric:String} AND sequence >= {firstSequence:UInt64} AND sequence <= {lastSequence:UInt64} - AND ingestedAt <= {watermark:DateTime64(3)} + AND has({ingestIds:Array(String)}, ingestId) GROUP BY environment, region, @@ -1873,7 +1935,9 @@ public function findSamples(SampleRange $range, DateTimeImmutable $watermark, in 'metric' => $range->metric, 'firstSequence' => $range->firstSequence, 'lastSequence' => $range->lastSequence, - 'watermark' => $watermark->setTimezone(new DateTimeZone('UTC'))->format('Y-m-d H:i:s.v'), + 'ingestIds' => $watermark->getIngestIds() === [] + ? '[]' + : "['" . implode("','", $watermark->getIngestIds()) . "']", 'queryLimit' => $limit + 1, ])); @@ -1890,32 +1954,49 @@ public function findSamples(SampleRange $range, DateTimeImmutable $watermark, in $sequence = self::toInt($row['sequence'] ?? null); $copies = self::toInt($row['copies'] ?? null); $variants = self::toInt($row['variants'] ?? null); - $intervalStart = new DateTimeImmutable(self::toStr($row['intervalStart'] ?? null), new DateTimeZone('UTC')); - $intervalEnd = new DateTimeImmutable(self::toStr($row['intervalEnd'] ?? null), new DateTimeZone('UTC')); + $duplicates += max(0, $copies - max(1, $variants)); + + if ($variants !== 1) { + $conflicts[] = $sequence; + continue; + } + + $observation = $row['observation'] ?? null; + if (!is_array($observation) || count($observation) !== 5) { + $conflicts[] = $sequence; + continue; + } + + try { + $sample = new Sample( + environment: self::toStr($row['environment'] ?? null), + region: self::toStr($row['region'] ?? null), + projectInternalId: self::toStr($row['projectInternalId'] ?? null), + databaseInternalId: self::toStr($row['databaseInternalId'] ?? null), + member: self::toStr($row['member'] ?? null), + generation: self::toStr($row['generation'] ?? null), + sequence: $sequence, + metric: self::toStr($row['metric'] ?? null), + intervalStart: new DateTimeImmutable(self::toStr($observation[0] ?? null), new DateTimeZone('UTC')), + intervalEnd: new DateTimeImmutable(self::toStr($observation[1] ?? null), new DateTimeZone('UTC')), + value: self::toInt($observation[2] ?? null), + eventVersion: self::toInt($observation[3] ?? null), + ); + } catch (\InvalidArgumentException) { + $conflicts[] = $sequence; + continue; + } if ( - $variants !== 1 - || $intervalStart < $range->intervalStart - || $intervalEnd > $range->intervalEnd + $sample->getPayloadHash() !== self::toStr($observation[4] ?? null) + || $sample->intervalStart < $range->intervalStart + || $sample->intervalEnd > $range->intervalEnd ) { $conflicts[] = $sequence; + continue; } - $duplicates += max(0, $copies - max(1, $variants)); - $samples[] = new Sample( - environment: self::toStr($row['environment'] ?? null), - region: self::toStr($row['region'] ?? null), - projectInternalId: self::toStr($row['projectInternalId'] ?? null), - databaseInternalId: self::toStr($row['databaseInternalId'] ?? null), - member: self::toStr($row['member'] ?? null), - generation: self::toStr($row['generation'] ?? null), - sequence: $sequence, - metric: self::toStr($row['metric'] ?? null), - intervalStart: $intervalStart, - intervalEnd: $intervalEnd, - value: self::toInt($row['value'] ?? null), - eventVersion: self::toInt($row['eventVersion'] ?? null), - ); + $samples[] = $sample; } $gaps = $this->findSampleGaps($samples, $range->firstSequence, $range->lastSequence); diff --git a/src/Usage/SampleResult.php b/src/Usage/SampleResult.php index ff5c4cd..2a4934f 100644 --- a/src/Usage/SampleResult.php +++ b/src/Usage/SampleResult.php @@ -2,8 +2,6 @@ namespace Utopia\Usage; -use DateTimeImmutable; - final readonly class SampleResult { /** @@ -19,7 +17,7 @@ public function __construct( private array $discontinuities, private int $duplicateCount, private bool $truncated, - private DateTimeImmutable $watermark, + private SampleWatermark $watermark, ) { } @@ -57,7 +55,7 @@ public function isTruncated(): bool return $this->truncated; } - public function getWatermark(): DateTimeImmutable + public function getWatermark(): SampleWatermark { return $this->watermark; } @@ -65,6 +63,7 @@ public function getWatermark(): DateTimeImmutable public function isComplete(): bool { return !$this->truncated + && !$this->watermark->isTruncated() && $this->conflicts === [] && $this->gaps === [] && $this->discontinuities === []; diff --git a/src/Usage/SampleWatermark.php b/src/Usage/SampleWatermark.php new file mode 100644 index 0000000..e85ca38 --- /dev/null +++ b/src/Usage/SampleWatermark.php @@ -0,0 +1,59 @@ + $ingestIds + */ + public function __construct( + private SampleRange $range, + private array $ingestIds, + private bool $truncated, + ) { + if ($ingestIds !== array_values(array_unique($ingestIds))) { + throw new InvalidArgumentException('ingestIds must be a unique list'); + } + + foreach ($ingestIds as $ingestId) { + if (preg_match('/^[a-f0-9]{32}$/', $ingestId) !== 1) { + throw new InvalidArgumentException('ingestIds must contain lowercase 128-bit hexadecimal IDs'); + } + } + } + + public function matches(SampleRange $range): bool + { + return $this->range->environment === $range->environment + && $this->range->region === $range->region + && $this->range->projectInternalId === $range->projectInternalId + && $this->range->databaseInternalId === $range->databaseInternalId + && $this->range->member === $range->member + && $this->range->generation === $range->generation + && $this->range->metric === $range->metric + && $this->range->firstSequence === $range->firstSequence + && $this->range->lastSequence === $range->lastSequence + && $this->range->getFormattedIntervalStart() === $range->getFormattedIntervalStart() + && $this->range->getFormattedIntervalEnd() === $range->getFormattedIntervalEnd(); + } + + /** @return list */ + public function getIngestIds(): array + { + return $this->ingestIds; + } + + public function isTruncated(): bool + { + return $this->truncated; + } +} diff --git a/src/Usage/Usage.php b/src/Usage/Usage.php index fc6d86c..9572925 100644 --- a/src/Usage/Usage.php +++ b/src/Usage/Usage.php @@ -85,12 +85,12 @@ public function addSamples(array $samples, int $batchSize = 1000): bool return $this->adapter->addSamples($samples, $batchSize); } - public function getSampleWatermark(): \DateTimeImmutable + public function getSampleWatermark(SampleRange $range, int $limit): SampleWatermark { - return $this->adapter->getSampleWatermark(); + return $this->adapter->getSampleWatermark($range, $limit); } - public function findSamples(SampleRange $range, \DateTimeImmutable $watermark, int $limit): SampleResult + public function findSamples(SampleRange $range, SampleWatermark $watermark, int $limit): SampleResult { return $this->adapter->findSamples($range, $watermark, $limit); } diff --git a/tests/Usage/Adapter/ClickHouseSampleTest.php b/tests/Usage/Adapter/ClickHouseSampleTest.php index e95033d..54571ba 100644 --- a/tests/Usage/Adapter/ClickHouseSampleTest.php +++ b/tests/Usage/Adapter/ClickHouseSampleTest.php @@ -39,6 +39,7 @@ public function testSampleTableHasCanonicalIdentityAndWatermarkColumns(): void $ddl = $this->queryRaw($adapter, "SHOW CREATE TABLE `{$database}`.`{$table}` FORMAT TabSeparatedRaw"); foreach ([ + '`ingestId` String', '`environment` LowCardinality(String)', '`region` LowCardinality(String)', '`projectInternalId` String', @@ -51,7 +52,6 @@ public function testSampleTableHasCanonicalIdentityAndWatermarkColumns(): void "`intervalEnd` DateTime64(3, 'UTC')", '`value` Int64', '`eventVersion` UInt32', - "`ingestedAt` DateTime64(3, 'UTC') DEFAULT now64(3)", ] as $column) { $this->assertStringContainsString($column, $ddl); } @@ -65,9 +65,10 @@ public function testCanonicalizesConcurrentDuplicatesAndCrashRetry(): void $this->assertTrue($this->usage->addSamples([$sample, $sample])); $this->assertTrue($this->usage->addSamples([$sample])); + $range = $this->range($key, firstSequence: 0, lastSequence: 0); $result = $this->usage->findSamples( - $this->range($key, firstSequence: 0, lastSequence: 0), - $this->usage->getSampleWatermark(), + $range, + $this->usage->getSampleWatermark($range, 10), 10, ); @@ -99,9 +100,10 @@ public function testCanonicalSamplesWaitForAsyncInsertDurability(): void $this->assertTrue($usage->addSamples([$this->sample($key, sequence: 0)])); + $range = $this->range($key, firstSequence: 0, lastSequence: 0); $result = $usage->findSamples( - $this->range($key, firstSequence: 0, lastSequence: 0), - $usage->getSampleWatermark(), + $range, + $usage->getSampleWatermark($range, 10), 10, ); @@ -118,14 +120,37 @@ public function testConflictingDuplicateFailsCompleteness(): void $this->sample($key, sequence: 0, value: 11), ])); + $range = $this->range($key, firstSequence: 0, lastSequence: 0); $result = $this->usage->findSamples( - $this->range($key, firstSequence: 0, lastSequence: 0), - $this->usage->getSampleWatermark(), + $range, + $this->usage->getSampleWatermark($range, 10), 10, ); $this->assertFalse($result->isComplete()); $this->assertSame([0], $result->getConflicts()); + $this->assertSame([], $result->getSamples()); + } + + public function testConflictingIntervalsReturnEvidenceWithoutSyntheticSample(): void + { + $key = bin2hex(random_bytes(8)); + + $this->assertTrue($this->usage->addSamples([ + $this->sample($key, sequence: 0, value: 10, startMinute: 0), + $this->sample($key, sequence: 0, value: 11, startMinute: 2), + ])); + + $range = $this->range($key, firstSequence: 0, lastSequence: 0, endMinute: 3); + $result = $this->usage->findSamples( + $range, + $this->usage->getSampleWatermark($range, 10), + 10, + ); + + $this->assertFalse($result->isComplete()); + $this->assertSame([0], $result->getConflicts()); + $this->assertSame([], $result->getSamples()); } public function testDetectsGapsWithoutExpandingEveryMissingSequence(): void @@ -137,9 +162,10 @@ public function testDetectsGapsWithoutExpandingEveryMissingSequence(): void $this->sample($key, sequence: 4), ])); + $range = $this->range($key, firstSequence: 0, lastSequence: 4); $result = $this->usage->findSamples( - $this->range($key, firstSequence: 0, lastSequence: 4), - $this->usage->getSampleWatermark(), + $range, + $this->usage->getSampleWatermark($range, 10), 10, ); @@ -158,9 +184,10 @@ public function testDetectsIntervalDiscontinuityWithContiguousSequences(): void $this->sample($key, sequence: 1, startMinute: 2), ])); + $range = $this->range($key, firstSequence: 0, lastSequence: 1, endMinute: 3); $result = $this->usage->findSamples( - $this->range($key, firstSequence: 0, lastSequence: 1, endMinute: 3), - $this->usage->getSampleWatermark(), + $range, + $this->usage->getSampleWatermark($range, 10), 10, ); @@ -169,7 +196,7 @@ public function testDetectsIntervalDiscontinuityWithContiguousSequences(): void $this->assertSame([1], $result->getDiscontinuities()); } - public function testReportsTruncationAndHonorsAStableWatermark(): void + public function testReportsTruncationAndHonorsAnExactWatermark(): void { $key = bin2hex(random_bytes(8)); @@ -179,17 +206,17 @@ public function testReportsTruncationAndHonorsAStableWatermark(): void $this->sample($key, sequence: 2), ])); - $watermark = $this->usage->getSampleWatermark(); - usleep(10_000); + $range = $this->range($key, firstSequence: 0, lastSequence: 3); + $watermark = $this->usage->getSampleWatermark($range, 10); $this->assertTrue($this->usage->addSamples([$this->sample($key, sequence: 3)])); $bounded = $this->usage->findSamples( - $this->range($key, firstSequence: 0, lastSequence: 3), + $range, $watermark, 2, ); $watermarked = $this->usage->findSamples( - $this->range($key, firstSequence: 0, lastSequence: 3), + $range, $watermark, 10, ); @@ -204,6 +231,117 @@ public function testReportsTruncationAndHonorsAStableWatermark(): void $this->assertSame(3, $watermarked->getGaps()[0]->last); } + public function testWatermarkEvidenceLimitFailsClosed(): void + { + $key = bin2hex(random_bytes(8)); + $sample = $this->sample($key, sequence: 0); + + $this->assertTrue($this->usage->addSamples([$sample, $sample, $sample])); + + $range = $this->range($key, firstSequence: 0, lastSequence: 0); + $watermark = $this->usage->getSampleWatermark($range, 2); + $result = $this->usage->findSamples($range, $watermark, 10); + + $this->assertTrue($watermark->isTruncated()); + $this->assertFalse($result->isComplete()); + $this->assertTrue($result->getWatermark()->isTruncated()); + } + + public function testTransportRetryAfterWatermarkCannotChangeSnapshot(): void + { + $key = bin2hex(random_bytes(8)); + $sample = $this->sample($key, sequence: 0); + $range = $this->range($key, firstSequence: 0, lastSequence: 0); + + $this->assertTrue($this->usage->addSamples([$sample])); + $watermark = $this->usage->getSampleWatermark($range, 10); + $this->assertCount(1, $watermark->getIngestIds()); + + $this->insertRawSample($sample, $watermark->getIngestIds()[0]); + + $result = $this->usage->findSamples($range, $watermark, 10); + + $this->assertTrue($result->isComplete()); + $this->assertCount(1, $result->getSamples()); + $this->assertSame(0, $result->getDuplicateCount()); + } + + public function testReusedIngestIdWithDifferentPayloadFailsClosed(): void + { + $key = bin2hex(random_bytes(8)); + $sample = $this->sample($key, sequence: 0, value: 10); + $range = $this->range($key, firstSequence: 0, lastSequence: 0); + + $this->assertTrue($this->usage->addSamples([$sample])); + $watermark = $this->usage->getSampleWatermark($range, 10); + $this->assertCount(1, $watermark->getIngestIds()); + + $this->insertRawSample( + $this->sample($key, sequence: 0, value: 11), + $watermark->getIngestIds()[0], + ); + + $result = $this->usage->findSamples($range, $watermark, 10); + + $this->assertFalse($result->isComplete()); + $this->assertSame([0], $result->getConflicts()); + $this->assertSame([], $result->getSamples()); + } + + public function testRejectsWatermarkFromAnotherRange(): void + { + $key = bin2hex(random_bytes(8)); + $range = $this->range($key, firstSequence: 0, lastSequence: 0); + $watermark = $this->usage->getSampleWatermark($range, 10); + + $this->expectException(\InvalidArgumentException::class); + $this->expectExceptionMessage('Sample watermark does not match the requested range'); + + $this->usage->findSamples( + $this->range($key, firstSequence: 0, lastSequence: 1), + $watermark, + 10, + ); + } + + private function insertRawSample(Sample $sample, string $ingestId): void + { + $adapter = $this->usage->getAdapter(); + $this->assertInstanceOf(ClickHouseAdapter::class, $adapter); + $database = $this->databaseName($adapter); + $table = $this->resolveTableName($adapter, 'getSamplesTableName'); + $sql = <<queryRaw($adapter, $sql, [ + 'id' => $sample->getId(), + 'payloadHash' => $sample->getPayloadHash(), + 'ingestId' => $ingestId, + 'environment' => $sample->environment, + 'region' => $sample->region, + 'projectInternalId' => $sample->projectInternalId, + 'databaseInternalId' => $sample->databaseInternalId, + 'member' => $sample->member, + 'generation' => $sample->generation, + 'sequence' => $sample->sequence, + 'metric' => $sample->metric, + 'intervalStart' => $sample->getFormattedIntervalStart(), + 'intervalEnd' => $sample->getFormattedIntervalEnd(), + 'value' => $sample->value, + 'eventVersion' => $sample->eventVersion, + ]); + } + private function sample(string $key, int $sequence, int $value = 10, ?int $startMinute = null): Sample { $start = new DateTimeImmutable('2026-08-01T00:00:00Z'); diff --git a/tests/Usage/SampleWatermarkTest.php b/tests/Usage/SampleWatermarkTest.php new file mode 100644 index 0000000..3336f26 --- /dev/null +++ b/tests/Usage/SampleWatermarkTest.php @@ -0,0 +1,58 @@ +range(lastSequence: 1); + $watermark = new SampleWatermark( + $range, + ['0123456789abcdef0123456789abcdef'], + false, + ); + + $this->assertTrue($watermark->matches($range)); + $this->assertFalse($watermark->matches($this->range(lastSequence: 2))); + $this->assertSame(['0123456789abcdef0123456789abcdef'], $watermark->getIngestIds()); + $this->assertFalse($watermark->isTruncated()); + } + + public function testRejectsDuplicateIngestIds(): void + { + $this->expectException(\InvalidArgumentException::class); + $this->expectExceptionMessage('ingestIds must be a unique list'); + + new SampleWatermark( + $this->range(lastSequence: 1), + [ + '0123456789abcdef0123456789abcdef', + '0123456789abcdef0123456789abcdef', + ], + false, + ); + } + + private function range(int $lastSequence): SampleRange + { + return new SampleRange( + environment: 'production', + region: 'fra1', + projectInternalId: '101', + databaseInternalId: '202', + member: 'mysql-0', + generation: 'generation-1', + metric: 'bandwidth.inbound', + firstSequence: 0, + lastSequence: $lastSequence, + intervalStart: new DateTimeImmutable('2026-08-01T00:00:00Z'), + intervalEnd: new DateTimeImmutable('2026-08-01T00:03:00Z'), + ); + } +} From edb11155408b229b5a63092f0fc2f1fd61030657 Mon Sep 17 00:00:00 2001 From: Jake Barnby Date: Thu, 27 Aug 2026 14:18:13 +1200 Subject: [PATCH 3/3] (fix): Bind watermarks to exact sample entries --- README.md | 10 +++++--- src/Usage/Adapter/ClickHouse.php | 27 ++++++++++++-------- src/Usage/SampleWatermark.php | 23 +++++++++-------- tests/Usage/Adapter/ClickHouseSampleTest.php | 22 ++++++++++------ tests/Usage/SampleWatermarkTest.php | 17 ++++++++---- 5 files changed, 61 insertions(+), 38 deletions(-) diff --git a/README.md b/README.md index 3de625d..cd23928 100644 --- a/README.md +++ b/README.md @@ -245,11 +245,13 @@ if (!$result->isComplete()) { Each supplied sample row receives an adapter-owned random ingestion ID before its request is sent. `getSampleWatermark()` performs one bounded ClickHouse -snapshot read and captures the exact IDs visible for that stream and range. -`findSamples()` admits only those IDs, so later inserts cannot cross the +snapshot read and captures each visible ingestion ID bound to its canonical ID +and payload hash for that stream and range. `findSamples()` admits only those +exact entry fingerprints, so later inserts or changed rows cannot cross the boundary even when their server timestamps would be identical. A transport -retry of the same request retains its IDs and is counted once; a new logical -retry gets a new ID and is included only when visible to the watermark query. +retry of the same request retains its fingerprint and is counted once; a new +logical retry gets a new ingestion ID and is included only when visible to the +watermark query. Both the watermark evidence and `findSamples()` result are explicitly bounded. A result is complete only when neither bound is truncated and there are no diff --git a/src/Usage/Adapter/ClickHouse.php b/src/Usage/Adapter/ClickHouse.php index 2d191ef..59316c6 100644 --- a/src/Usage/Adapter/ClickHouse.php +++ b/src/Usage/Adapter/ClickHouse.php @@ -1827,7 +1827,7 @@ public function getSampleWatermark(SampleRange $range, int $limit): SampleWaterm $table = $this->buildTableReference($this->getSamplesTableName()); $sql = <<= {firstSequence:UInt64} AND sequence <= {lastSequence:UInt64} AND ingestId != '' - LIMIT 1 BY ingestId + LIMIT 1 BY entryId LIMIT {queryLimit:UInt64} FORMAT JSON SQL; @@ -1862,12 +1862,12 @@ public function getSampleWatermark(SampleRange $range, int $limit): SampleWaterm $rows = array_slice($rows, 0, $limit); } - $ingestIds = []; + $entries = []; foreach ($rows as $row) { - $ingestIds[] = self::toStr($row['ingestId'] ?? null); + $entries[] = self::toStr($row['entryId'] ?? null); } - return new SampleWatermark($range, $ingestIds, $truncated); + return new SampleWatermark($range, $entries, $truncated); } #[\Override] @@ -1896,9 +1896,16 @@ public function findSamples(SampleRange $range, SampleWatermark $watermark, int metric, argMin( tuple(intervalStart, intervalEnd, value, eventVersion, payloadHash), - tuple(payloadHash, intervalStart, intervalEnd, value, eventVersion, ingestId) + tuple( + payloadHash, + intervalStart, + intervalEnd, + value, + eventVersion, + concat(ingestId, ':', id, ':', payloadHash) + ) ) AS observation, - uniqExact(ingestId) AS copies, + uniqExact(concat(ingestId, ':', id, ':', payloadHash)) AS copies, uniqExact(tuple(payloadHash, intervalStart, intervalEnd, value, eventVersion)) AS variants FROM {$table} WHERE environment = {environment:String} @@ -1910,7 +1917,7 @@ public function findSamples(SampleRange $range, SampleWatermark $watermark, int AND metric = {metric:String} AND sequence >= {firstSequence:UInt64} AND sequence <= {lastSequence:UInt64} - AND has({ingestIds:Array(String)}, ingestId) + AND has({entries:Array(String)}, concat(ingestId, ':', id, ':', payloadHash)) GROUP BY environment, region, @@ -1935,9 +1942,9 @@ public function findSamples(SampleRange $range, SampleWatermark $watermark, int 'metric' => $range->metric, 'firstSequence' => $range->firstSequence, 'lastSequence' => $range->lastSequence, - 'ingestIds' => $watermark->getIngestIds() === [] + 'entries' => $watermark->getEntries() === [] ? '[]' - : "['" . implode("','", $watermark->getIngestIds()) . "']", + : "['" . implode("','", $watermark->getEntries()) . "']", 'queryLimit' => $limit + 1, ])); diff --git a/src/Usage/SampleWatermark.php b/src/Usage/SampleWatermark.php index e85ca38..33f824f 100644 --- a/src/Usage/SampleWatermark.php +++ b/src/Usage/SampleWatermark.php @@ -7,26 +7,27 @@ /** * Exact bounded membership captured by one adapter snapshot query. * - * IDs are generated inside addSamples() before transport, so a repeated HTTP - * request retains them while any later logical insert receives different IDs. + * Ingestion IDs are generated inside addSamples() before transport. Each + * captured entry also binds the canonical and payload hashes, so a repeated + * HTTP request retains the same entry while any changed row is excluded. */ final readonly class SampleWatermark { /** - * @param list $ingestIds + * @param list $entries */ public function __construct( private SampleRange $range, - private array $ingestIds, + private array $entries, private bool $truncated, ) { - if ($ingestIds !== array_values(array_unique($ingestIds))) { - throw new InvalidArgumentException('ingestIds must be a unique list'); + if ($entries !== array_values(array_unique($entries))) { + throw new InvalidArgumentException('entries must be a unique list'); } - foreach ($ingestIds as $ingestId) { - if (preg_match('/^[a-f0-9]{32}$/', $ingestId) !== 1) { - throw new InvalidArgumentException('ingestIds must contain lowercase 128-bit hexadecimal IDs'); + foreach ($entries as $entry) { + if (preg_match('/^[a-f0-9]{32}:[a-f0-9]{64}:[a-f0-9]{64}$/', $entry) !== 1) { + throw new InvalidArgumentException('entries must bind an ingestion ID, canonical ID and payload hash'); } } } @@ -47,9 +48,9 @@ public function matches(SampleRange $range): bool } /** @return list */ - public function getIngestIds(): array + public function getEntries(): array { - return $this->ingestIds; + return $this->entries; } public function isTruncated(): bool diff --git a/tests/Usage/Adapter/ClickHouseSampleTest.php b/tests/Usage/Adapter/ClickHouseSampleTest.php index 54571ba..667b213 100644 --- a/tests/Usage/Adapter/ClickHouseSampleTest.php +++ b/tests/Usage/Adapter/ClickHouseSampleTest.php @@ -255,9 +255,9 @@ public function testTransportRetryAfterWatermarkCannotChangeSnapshot(): void $this->assertTrue($this->usage->addSamples([$sample])); $watermark = $this->usage->getSampleWatermark($range, 10); - $this->assertCount(1, $watermark->getIngestIds()); + $this->assertCount(1, $watermark->getEntries()); - $this->insertRawSample($sample, $watermark->getIngestIds()[0]); + $this->insertRawSample($sample, $this->ingestId($watermark->getEntries()[0])); $result = $this->usage->findSamples($range, $watermark, 10); @@ -266,7 +266,7 @@ public function testTransportRetryAfterWatermarkCannotChangeSnapshot(): void $this->assertSame(0, $result->getDuplicateCount()); } - public function testReusedIngestIdWithDifferentPayloadFailsClosed(): void + public function testReusedIngestIdWithDifferentPayloadCannotChangeSnapshot(): void { $key = bin2hex(random_bytes(8)); $sample = $this->sample($key, sequence: 0, value: 10); @@ -274,18 +274,19 @@ public function testReusedIngestIdWithDifferentPayloadFailsClosed(): void $this->assertTrue($this->usage->addSamples([$sample])); $watermark = $this->usage->getSampleWatermark($range, 10); - $this->assertCount(1, $watermark->getIngestIds()); + $this->assertCount(1, $watermark->getEntries()); $this->insertRawSample( $this->sample($key, sequence: 0, value: 11), - $watermark->getIngestIds()[0], + $this->ingestId($watermark->getEntries()[0]), ); $result = $this->usage->findSamples($range, $watermark, 10); - $this->assertFalse($result->isComplete()); - $this->assertSame([0], $result->getConflicts()); - $this->assertSame([], $result->getSamples()); + $this->assertTrue($result->isComplete()); + $this->assertSame([], $result->getConflicts()); + $this->assertCount(1, $result->getSamples()); + $this->assertSame(10, $result->getSamples()[0]->value); } public function testRejectsWatermarkFromAnotherRange(): void @@ -342,6 +343,11 @@ private function insertRawSample(Sample $sample, string $ingestId): void ]); } + private function ingestId(string $entry): string + { + return explode(':', $entry, 2)[0]; + } + private function sample(string $key, int $sequence, int $value = 10, ?int $startMinute = null): Sample { $start = new DateTimeImmutable('2026-08-01T00:00:00Z'); diff --git a/tests/Usage/SampleWatermarkTest.php b/tests/Usage/SampleWatermarkTest.php index 3336f26..5bb1ae5 100644 --- a/tests/Usage/SampleWatermarkTest.php +++ b/tests/Usage/SampleWatermarkTest.php @@ -14,31 +14,38 @@ public function testBindsEvidenceToOneExactRange(): void $range = $this->range(lastSequence: 1); $watermark = new SampleWatermark( $range, - ['0123456789abcdef0123456789abcdef'], + [$this->entry()], false, ); $this->assertTrue($watermark->matches($range)); $this->assertFalse($watermark->matches($this->range(lastSequence: 2))); - $this->assertSame(['0123456789abcdef0123456789abcdef'], $watermark->getIngestIds()); + $this->assertSame([$this->entry()], $watermark->getEntries()); $this->assertFalse($watermark->isTruncated()); } public function testRejectsDuplicateIngestIds(): void { $this->expectException(\InvalidArgumentException::class); - $this->expectExceptionMessage('ingestIds must be a unique list'); + $this->expectExceptionMessage('entries must be a unique list'); new SampleWatermark( $this->range(lastSequence: 1), [ - '0123456789abcdef0123456789abcdef', - '0123456789abcdef0123456789abcdef', + $this->entry(), + $this->entry(), ], false, ); } + private function entry(): string + { + return '0123456789abcdef0123456789abcdef' + . ':' . str_repeat('a', 64) + . ':' . str_repeat('b', 64); + } + private function range(int $lastSequence): SampleRange { return new SampleRange(