feat(usage): signed deltas, cross-tenant reads and max aggregation (CLO-4542) - #22
Merged
Conversation
Realtime connections are stored as +1/-1 event deltas aggregated with SUM, so a plain MAX(value) is meaningless. Add an `aggregate` query hint (sum | peak) mirroring the groupBy pattern. `peak` computes the peak concurrent value as max(running_sum(value)) ordered by time, cross-pod correct with no producer change. - UsageQuery: TYPE_AGGREGATE, VALID_AGGREGATES, aggregate() plus isAggregate/extractAggregate/removeAggregate helpers. - ClickHouse: parseQueries splits time vs non-time filters and records the window-start param for peak; findFromTable routes peak to a new findPeakFromTable that builds a windowed running-sum with a pre-window baseline subquery (connections still open at start). Honours interval bucketing, dimensions, limit/offset/orderBy. The sum/default path is unchanged. - Tests: flat + interval peak, pre-window baseline, interleaved-producer sum-before-max, and UsageQuery unit coverage.
Make the peak path correct end-to-end for delta metrics like realtime connections, where the write side previously dropped -1 disconnects and folded sub-flush bursts away. - Reject negative values by default for every metric (events included) so a buggy negative count/bandwidth is still caught. Callers emitting a genuine signed delta opt in per row via allowNegative: Accumulator::collect(..., bool $allowNegative = false) carries the flag onto the buffered entry and hands it to addBatch; ClickHouse::validateMetricData(..., bool $allowNegative = false) gates the guard, read from each row's `allowNegative` in validateMetricsBatch. The flag is validation-only — never written as a column. The library stays generic; the caller decides which metrics may be negative. - Add optional foldSeconds to Accumulator::collect(). When set, the second bucket (floor(ts / foldSeconds) * foldSeconds) joins the fold key and becomes the entry time, so events fold only within the same bucket and intra-flush peaks survive. When null, behaviour is unchanged. - Tests: negatives rejected by default (collect + addBatch), persisted and netted when opted in, gauges still reject; per-second fold groups/splits by second; peak over per-second net rows captures a burst a per-flush net hides.
findAcrossTenants() applies no tenant filter so an aggregation job can roll
every tenant up in one pass instead of issuing N per-tenant queries. Shared
tables only; groupBy('tenant') keeps the rows attributable.
parseQueries() now takes a nullable tenant - null is the explicit cross-tenant
read, '' still fails fast so an empty scope can never be read silently.
The peak concurrency figure is now pre-computed into a gauge level series by
an operator-side job, so nothing reads a running-sum max at request time. Drop
findPeakFromTable and its baseline plumbing, and keep the aggregate hint for
what does need it: rolling a gauge series up to a coarser interval.
Gauges default to argMax(value, time) - the latest reading in the bucket, right
for a snapshot but wrong for a sampled level series, where the bucket's highest
sample is the answer. aggregate('max') selects that.
Removing the peak path also removes its baseline defect: the pre-window
baseline was a single non-correlated scalar added to a running sum partitioned
by metric[, tenant][, dims], so any dimension break-down had every series
inflated by the combined baseline of all of them.
lohanidamodar
changed the title
findFromTable() only checked for groupByInterval or groupBy, so an
aggregate('max') with neither fell through and returned raw rows instead of one
aggregated row. That is exactly the flat 'highest value over this window' shape
billing reads, which would have taken an arbitrary row's value for the peak.
The removed peak branch used to catch this case before the check.
Also declare aggregate on the parsed-array shape so PHPStan sees the offset,
fix a test that asserted a rejection while passing a valid function, and assert
the tenant is present before using it as an array key.
lohanidamodar
marked this pull request as ready for review
August 5, 2026 03:36
Greptile SummaryThe PR adds explicit support for signed delta rows, operator-only cross-tenant ClickHouse reads, and maximum aggregation for sampled gauge levels.
Confidence Score: 5/5The PR appears safe to merge. No blocking failure remains; the previously reported folded-row, Database signed-delta, and gauge-sum issues are resolved in the current code. Important Files Changed
Reviews (4): Last reviewed commit: "Honour the negative opt-in in the Databa..." | Re-trigger Greptile |
![greptile-apps[bot]](https://avatars.githubusercontent.com/in/867647?s=60&v=4){kind=small}
foldSeconds was added for a request-time peak that no longer exists: with the concurrency level pre-computed into a gauge and sampled every five minutes, per-second rows cost roughly twenty times the storage to sharpen a resolution the bucketing discards. Nothing passed it. Leaving it in would have meant an unused parameter on the hottest write path of a shared library, on a branch nothing exercises. It is one commit back in history if a caller ever wants it, and then it arrives with the caller that justifies it. allowNegative stays - the concurrency level is the cumulative sum of the +1/-1s, so without it there is nothing to fold.
Two findings from review.
The opt-in was stored only when a buffer entry was created, so folding a signed
delta into an entry opened by a plain positive left the net row looking
unauthorised. It would be rejected at write time, and because a failed batch
keeps its entries buffered, every later flush would retry the same rejection -
a stuck buffer rather than one lost row. The flag now ORs across the fold, with
tests in both call orders.
Unreachable from cloud today, since the flag is derived from the metric name
and the fold key includes it. But the fold key uses the name as written while
the flag is keyed on the source, so a mirrored metric that was also a delta
would mix them. Cheap to make the invariant hold on its own rather than resting
on a caller-side coincidence.
aggregate('sum') was accepted but left gauges on argMax, returning the latest
sample rather than the documented total. Rather than make it sum, it is gone:
it is already the default for events, and on gauges it would total
point-in-time snapshots, which this library elsewhere refuses to do. max is now
the only selectable aggregate - it exists to override the per-type default.
Accumulator marks a signed-delta row allowNegative and the ClickHouse adapter honours it, but Database::addBatch() rejected the value regardless. The opt-in is part of the row contract, so an adapter that ignores it turns a valid metric into a permanently stuck buffer: the batch throws, flush() keeps unwritten entries, and every later flush retries the same rejection. Not reachable from cloud, which runs ClickHouse, but the two adapters have to agree about what a row means.
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
1 participant
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
CLO-4542 — support for peak concurrent realtime connections
realtime.connectionsis a delta metric:+1on connect,-1on disconnect, emitted by every realtime pod. Concurrency is therefore the cumulative sum of those deltas — summing them over a window gives the net change (roughly zero for a balanced window), andMAX(value)gives1.Cloud pre-computes that level in
StatsResourcesand stores it as a gauge (appwrite-labs/cloud#4662). This PR is the library support that job needs.Changes
Signed deltas, strict by default —
Accumulator::collect()andClickHouse::validateMetricData()still reject negative values for every metric, so a buggy negative count or bandwidth figure is still caught. A caller emitting a genuine signed delta opts in per row withallowNegative: true. The flag is validation-only and never stored as a column; the library stays generic and holds no metric names.Cross-tenant reads —
findAcrossTenants(array $queries, ?string $type)applies no tenant filter, so an operator-side aggregation job can roll every tenant up in one pass instead of issuing N per-tenant queries. Shared tables only.tenantis now accepted as agroupBydimension so the returned rows stay attributable.This is deliberately a separate method rather than a nullable
$tenantonfind().find()keeps its non-nullable signature, and only the internals (findScoped,findFromTable,parseQueries) take?string, so the unscoped path can't be reached without naming it. A nullable public parameter would mean anullarriving from an uninitialised project id silently returns every tenant's rows; as a named method it can't happen by accident and every cross-tenant read is greppable. NoteparseQueriesstill rejects''— an empty tenant remains a fail-fast error, not a wildcard.aggregate('max')— a query hint that overrides the per-type default value expression. Gauges default toargMax(value, time), the latest reading in the bucket. That's right for a snapshot such as storage, but wrong for a sampled level series, where the bucket's highest reading is the answer.maxcomposes, so the max of 5-minute samples is the hourly, daily or billing-period peak.Superseded
findPeakFromTable,aggregate('peak')and the per-second fold are gone.The fold existed to keep intra-flush bursts visible to a request-time running-sum max. With the level pre-computed into a gauge and sampled every five minutes, per-second rows cost roughly 20× the storage to sharpen a resolution the bucketing discards — so nothing passed it, and an unused parameter on the hottest write path of a shared library is not worth carrying. One commit back in history if a caller ever wants it.
On the query side, nothing reads a running-sum max at request time once the level is pre-computed, and the query was expensive on the path that would have used it — it needed every delta since the project began.
Removing it also removed a defect. The pre-window baseline was a single non-correlated scalar:
added to a running sum partitioned by
metric[, tenant][, dims]. Those only agree when the window covers exactly one series. Any dimension break-down —aggregate('peak')withgroupBy('country')— had every series inflated by the combined baseline of all of them. The peak tests only coveredgroupByInterval, never a dimension, so nothing caught it.Tests
ClickHouseGaugeMaxTestreplacesClickHousePeakTest:lastvsmaxon the same data, per-bucket maxima, the flat single-row shape billing uses, tenant-scopedmaxunder shared tables,findAcrossTenantsreturning a row per tenant, and the non-shared-tables rejection.AccumulatorTestcovers negatives rejected by default through bothcollect()andaddBatch, persisted and netted when opted in, and gauges still rejecting.UsageQueryTestcovers theaggregatehint and helpers.Not yet run locally — no composer on this machine, and the ClickHouse suite needs a live instance. Relying on CI.
Merge order
Cloud depends on a tagged release of this. Draft until then.