Skip to content

Commit 4fe45a0

Browse files
authored
fix(billing): key the usage email claim on the exact period start and anchor shared sums to their start (#8360)
* fix(billing): key the usage email claim on the exact period start and anchor shared sums to their start * fix(billing): anchor shared sum expiry with a portable PX and a floor for slow sums * docs(billing): note the reconnect delay in the shared sum staleness bound
1 parent e1124ff commit 4fe45a0

5 files changed

Lines changed: 89 additions & 23 deletions

File tree

‎apps/sim/lib/billing/core/limit-notifications.ts‎

Lines changed: 8 additions & 9 deletions
Original file line numberDiff line numberDiff line change
@@ -79,8 +79,6 @@ async function claimThreshold(
7979
return writeLimitNotifications(scope, id, setExpr, onlyIfLower)
8080
}
8181

82-
const DAY_MS = 24 * 60 * 60 * 1000
83-
8482
/** One account's credits threshold, keyed on its billing period and limit. */
8583
export interface CreditsThresholdClaim {
8684
scope: 'user' | 'organization'
@@ -92,20 +90,21 @@ export interface CreditsThresholdClaim {
9290

9391
/**
9492
* The stored claims for a credits threshold, and the condition under which it is still unclaimed:
95-
* `credits` holds the highest threshold emailed while `creditsPeriod` (the period's start day) and
96-
* `creditsLimit` (the limit in cents) still match, so a new period or a changed limit — in either
97-
* direction — re-arms both thresholds with no reset write, while within one a claim of 100 also
98-
* retires 80, and never the reverse.
93+
* `credits` holds the highest threshold emailed while `creditsPeriod` (the period's exact start,
94+
* in epoch seconds) and `creditsLimit` (the limit in cents) still match, so a new period — even
95+
* one starting the same day as the last — or a changed limit, in either direction, re-arms both
96+
* thresholds with no reset write, while within one a claim of 100 also retires 80, and never the
97+
* reverse.
9998
*/
10099
function creditsThresholdSql(claim: CreditsThresholdClaim) {
101-
const periodDay = Math.floor(claim.periodStart.getTime() / DAY_MS)
100+
const periodStartSeconds = Math.floor(claim.periodStart.getTime() / 1000)
102101
const limitCents = Math.round(claim.limit * 100)
103102
const column =
104103
claim.scope === 'user' ? userStats.limitNotifications : organization.limitNotifications
105104
return {
106-
next: sql`coalesce(${column}, '{}'::jsonb) || jsonb_build_object('credits', ${claim.threshold}::int, 'creditsPeriod', ${periodDay}::bigint, 'creditsLimit', ${limitCents}::bigint)`,
105+
next: sql`coalesce(${column}, '{}'::jsonb) || jsonb_build_object('credits', ${claim.threshold}::int, 'creditsPeriod', ${periodStartSeconds}::bigint, 'creditsLimit', ${limitCents}::bigint)`,
107106
unclaimed: sql<boolean>`not (
108-
(${column} ->> 'creditsPeriod')::bigint is not distinct from ${periodDay}::bigint
107+
(${column} ->> 'creditsPeriod')::bigint is not distinct from ${periodStartSeconds}::bigint
109108
and (${column} ->> 'creditsLimit')::bigint is not distinct from ${limitCents}::bigint
110109
and coalesce((${column} ->> 'credits')::int, 0) >= ${claim.threshold}::int
111110
)`,

‎apps/sim/lib/billing/core/reporting-usage-cache.integration.ts‎

Lines changed: 38 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -22,7 +22,10 @@ const redisUrl = readTestRedisUrl()
2222
vi.mock('@sim/db', () => ({ db: { transaction }, dbReplica: {} }))
2323
vi.mock('@/lib/core/config/redis', () => redisConfigMock)
2424

25-
import { readSoftGateUsageCost } from '@/lib/billing/core/reporting-usage-cache'
25+
import {
26+
REPORTING_USAGE_CACHE_TTL_MS,
27+
readSoftGateUsageCost,
28+
} from '@/lib/billing/core/reporting-usage-cache'
2629
import type { BillingEntity, UsageQueryPeriod } from '@/lib/billing/core/usage-log'
2730

2831
const schemaName = `reporting_usage_${generateId().replaceAll('-', '')}`
@@ -71,6 +74,7 @@ describe.runIf(Boolean(redisUrl))('shared reporting usage read', () => {
7174
})
7275

7376
afterEach(async () => {
77+
vi.useRealTimers()
7478
await redis.del(sharedKey(payer))
7579
})
7680

@@ -101,6 +105,39 @@ describe.runIf(Boolean(redisUrl))('shared reporting usage read', () => {
101105
await expect(readSoftGateUsageCost(payer, REPORTING)).resolves.toBe(5.75)
102106
})
103107

108+
it('shares a sum that ran longer than the TTL for a short floor instead of dropping it', async () => {
109+
const now = Date.now()
110+
vi.useFakeTimers({ toFake: ['Date'] })
111+
vi.setSystemTime(now - REPORTING_USAGE_CACHE_TTL_MS - 1_000)
112+
transaction.mockImplementationOnce(async (callback) => {
113+
const result = await database.transaction(callback)
114+
vi.setSystemTime(now)
115+
return result
116+
})
117+
118+
await expect(readSoftGateUsageCost(payer, REPORTING)).resolves.toBe(5.75)
119+
expect(await redis.get(sharedKey(payer))).toBe('5.75')
120+
const ttl = await redis.pttl(sharedKey(payer))
121+
expect(ttl).toBeGreaterThan(4_000)
122+
expect(ttl).toBeLessThanOrEqual(5_000)
123+
})
124+
125+
it('anchors a stored sum expiry to when the sum began, not when it was written', async () => {
126+
const now = Date.now()
127+
vi.useFakeTimers({ toFake: ['Date'] })
128+
vi.setSystemTime(now - 20_000)
129+
transaction.mockImplementationOnce(async (callback) => {
130+
const result = await database.transaction(callback)
131+
vi.setSystemTime(now)
132+
return result
133+
})
134+
135+
await expect(readSoftGateUsageCost(payer, REPORTING)).resolves.toBe(5.75)
136+
const ttl = await redis.pttl(sharedKey(payer))
137+
expect(ttl).toBeGreaterThan(REPORTING_USAGE_CACHE_TTL_MS - 20_000 - 1_000)
138+
expect(ttl).toBeLessThanOrEqual(REPORTING_USAGE_CACHE_TTL_MS - 20_000 + 5_000)
139+
})
140+
104141
it('never lets a slower, older sum replace one stored while it ran', async () => {
105142
transaction.mockImplementationOnce(async (callback) => {
106143
const result = await database.transaction(callback)

‎apps/sim/lib/billing/core/reporting-usage-cache.ts‎

Lines changed: 30 additions & 10 deletions
Original file line numberDiff line numberDiff line change
@@ -28,15 +28,24 @@ const logger = createLogger('ReportingUsageCache')
2828
* small against a year-long allowance while turning a per-event scan into one per window.
2929
*
3030
* A sum is held both in Redis, shared by every process, and in each process that reads it. It
31-
* reflects the ledger as of the moment its sum began, so a served sum can omit usage written over
32-
* the sum's own duration, plus up to this TTL and its jitter in Redis, plus up to this TTL again
33-
* in the reading process.
31+
* reflects the ledger as of the moment its sum began, and its Redis expiry is anchored to that
32+
* moment, so it is served from Redis for at most max(this TTL + jitter, the sum's duration +
33+
* {@link MIN_SHARED_TTL_MS}) after it began (plus any reconnect delay for a resent write), plus
34+
* up to this TTL again in the reading process.
3435
*/
3536
export const REPORTING_USAGE_CACHE_TTL_MS = 30_000
3637

3738
/** Redis expiry is jittered by up to this much, so payers summed together do not expire together. */
3839
const SHARED_TTL_JITTER_MS = 5_000
3940

41+
/**
42+
* The shortest time a sum is kept in Redis. A sum that ran longer than the TTL would otherwise
43+
* expire on arrival, and under the database pressure that makes sums slow, every process would
44+
* then run the same slow sum again. Five seconds shares it with the processes waiting on it while
45+
* adding little staleness next to the time the sum itself took.
46+
*/
47+
const MIN_SHARED_TTL_MS = 5_000
48+
4049
/**
4150
* How long a read waits on a connected Redis before summing the ledger instead, so a socket that
4251
* has silently stopped answering costs one sum rather than the shared client's long timeouts.
@@ -95,17 +104,27 @@ function warnSharedWriteFailed(error: unknown): void {
95104
}
96105

97106
/**
98-
* Fire-and-forget: a read never waits on, or fails because of, the shared write. The write only
99-
* lands when no sum is stored (`NX`), so a slow, older sum can never replace a fresher one or
100-
* extend its expiry.
107+
* Fire-and-forget: a read never waits on, or fails because of, the shared write. The expiry is
108+
* anchored to when the sum began: the remaining lifetime is computed here and written with `PX`,
109+
* which every Redis version accepts, floored at {@link MIN_SHARED_TTL_MS}. A write the client
110+
* resends after a reconnect re-applies that same relative lifetime from the resend, so it can
111+
* extend the expiry by at most the reconnect delay. The write only lands when no sum is stored
112+
* (`NX`), so an older sum can never replace a fresher one or extend its expiry.
101113
*/
102-
function writeSharedReportingUsageCost(key: string, cost: number): void {
114+
function writeSharedReportingUsageCost(key: string, cost: number, sumStartedAt: number): void {
103115
try {
104116
const redis = readyRedisClient()
105117
if (!redis) return
106-
const ttlMs = REPORTING_USAGE_CACHE_TTL_MS + randomInt(0, SHARED_TTL_JITTER_MS)
118+
const remainingMs =
119+
sumStartedAt + REPORTING_USAGE_CACHE_TTL_MS + randomInt(0, SHARED_TTL_JITTER_MS) - Date.now()
107120
redis
108-
.set(sharedReportingUsageKey(key), String(cost), 'PX', ttlMs, 'NX')
121+
.set(
122+
sharedReportingUsageKey(key),
123+
String(cost),
124+
'PX',
125+
Math.max(remainingMs, MIN_SHARED_TTL_MS),
126+
'NX'
127+
)
109128
.catch(warnSharedWriteFailed)
110129
} catch (error) {
111130
warnSharedWriteFailed(error)
@@ -124,8 +143,9 @@ async function sumReportingUsageCost(
124143
): Promise<number> {
125144
const shared = await readSharedReportingUsageCost(key)
126145
if (shared !== undefined) return shared
146+
const sumStartedAt = Date.now()
127147
const cost = await getBillingPeriodUsageCost(entity, period)
128-
writeSharedReportingUsageCost(key, cost)
148+
writeSharedReportingUsageCost(key, cost, sumStartedAt)
129149
return cost
130150
}
131151

‎apps/sim/lib/billing/core/usage-threshold-email.integration.ts‎

Lines changed: 10 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -92,7 +92,7 @@ async function claims(): Promise<Record<string, number>> {
9292
function claimOf(threshold: 80 | 100, periodStart = SEPTEMBER, limitCents = 10_000) {
9393
return {
9494
credits: threshold,
95-
creditsPeriod: Math.floor(periodStart.getTime() / 86_400_000),
95+
creditsPeriod: periodStart.getTime() / 1000,
9696
creditsLimit: limitCents,
9797
}
9898
}
@@ -159,6 +159,15 @@ describe('usage threshold email', () => {
159159
expect(await claims()).toEqual(claimOf(80))
160160
})
161161

162+
it('re-arms for a new period that starts the same day as the one it replaces', async () => {
163+
const replacement = new Date('2026-09-01T12:00:00.000Z')
164+
await notify(85)
165+
await notify(85, { periodStart: replacement })
166+
167+
expect(delivered()).toEqual([warning(), warning()])
168+
expect(await claims()).toEqual(claimOf(80, replacement))
169+
})
170+
162171
it('warns again at a raised limit after the old one was reached', async () => {
163172
await notify(100)
164173
await notify(100, { limit: 125 })

‎packages/db/schema.ts‎

Lines changed: 3 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -1359,8 +1359,9 @@ export const userStats = pgTable('user_stats', {
13591359
* re-arms when usage drops back below the re-arm band. Keyed by limit
13601360
* category ('storage' | 'tables'); seats live on `organization`. `credits`
13611361
* instead holds the threshold emailed for the billing period and limit in
1362-
* `creditsPeriod` (start day) and `creditsLimit` (cents), so a new period or a
1363-
* changed limit re-arms it without a reset (see `claimCreditsThreshold`).
1362+
* `creditsPeriod` (start, epoch seconds) and `creditsLimit` (cents), so a new
1363+
* period or a changed limit re-arms it without a reset (see
1364+
* `claimCreditsThreshold`).
13641365
*
13651366
* Dedup granularity is per billing account per category — intentionally NOT
13661367
* per table, so a user hitting the row limit on several tables gets one

0 commit comments

Comments
 (0)