diff --git a/apps/web/src/app/admin/cloud-billing-skus/UsageRecordsContent.tsx b/apps/web/src/app/admin/cloud-billing-skus/UsageRecordsContent.tsx index 10291b8351..f6fafc7000 100644 --- a/apps/web/src/app/admin/cloud-billing-skus/UsageRecordsContent.tsx +++ b/apps/web/src/app/admin/cloud-billing-skus/UsageRecordsContent.tsx @@ -44,18 +44,38 @@ type SearchRequest = skuId?: string; } | { - kind: SearchKind; + kind: 'interval'; value: string; status?: 'open' | 'closed'; closeReason?: CloseReason; skuId?: string; + } + | { + kind: 'user' | 'org'; + value: string; + status?: 'open' | 'closed'; + closeReason?: CloseReason; + skuId?: string; + summaryStart?: string; + summaryEnd?: string; }; type Cursor = { startedAt: string; id: string }; +type UsageSummaryRequest = { + subjectType: 'user' | 'org'; + subjectId: string; + start: string; + end: string; +}; function formatTimestamp(value: string | null): string { return value ? new Date(value).toLocaleString() : '—'; } +function toDateTimeLocalValue(value: Date): string { + const offset = value.getTimezoneOffset() * 60_000; + return new Date(value.getTime() - offset).toISOString().slice(0, 16); +} + function adminSubjectHref(type: 'user' | 'org', id: string): string { return type === 'user' ? `/admin/users/${encodeURIComponent(id)}` @@ -200,6 +220,12 @@ export default function UsageRecordsContent() { const [cursor, setCursor] = useState(); const [previousCursors, setPreviousCursors] = useState>([]); const [expandedId, setExpandedId] = useState(null); + const [summaryStart, setSummaryStart] = useState(() => + toDateTimeLocalValue(new Date(Date.now() - 24 * 60 * 60 * 1_000)) + ); + const [summaryEnd, setSummaryEnd] = useState(() => toDateTimeLocalValue(new Date())); + const [summaryRequest, setSummaryRequest] = useState(null); + const [summaryInputError, setSummaryInputError] = useState(null); const input = { search: @@ -211,6 +237,8 @@ export default function UsageRecordsContent() { kind: 'subject', subjectType: submitted.kind, subjectId: submitted.value, + start: submitted.summaryStart ?? '', + end: submitted.summaryEnd ?? '', } as const), status: submitted.status, closeReason: submitted.closeReason, @@ -219,6 +247,17 @@ export default function UsageRecordsContent() { limit: submitted.kind === 'recent' ? 10 : 25, }; const results = useQuery(trpc.admin.cloudBillingSkus.searchUsageIntervals.queryOptions(input)); + const summary = useQuery({ + ...trpc.admin.cloudBillingSkus.getUsageSummary.queryOptions( + summaryRequest ?? { + subjectType: 'user', + subjectId: 'not-submitted', + start: new Date(0).toISOString(), + end: new Date(1).toISOString(), + } + ), + enabled: summaryRequest !== null, + }); const rows = results.data?.items ?? []; const resetResultNavigation = () => { @@ -258,160 +297,362 @@ export default function UsageRecordsContent() {
{ event.preventDefault(); const trimmed = value.trim(); if (!trimmed) return; + let summaryWindow: { summaryStart?: string; summaryEnd?: string } = {}; + if (kind === 'user' || kind === 'org') { + const start = new Date(summaryStart); + const end = new Date(summaryEnd); + const windowMs = end.getTime() - start.getTime(); + if ( + Number.isNaN(start.getTime()) || + Number.isNaN(end.getTime()) || + windowMs <= 0 || + windowMs > 31 * 24 * 60 * 60 * 1_000 + ) { + setSummaryInputError('Choose a valid window of no more than 31 days.'); + return; + } + summaryWindow = { + summaryStart: start.toISOString(), + summaryEnd: end.toISOString(), + }; + } const next: SearchRequest = { kind, value: trimmed, status: status === 'all' ? undefined : status, closeReason: closeReason === 'all' ? undefined : closeReason, skuId: skuId === 'all' ? undefined : skuId, + ...summaryWindow, }; + const submittedValue = submitted.kind === 'recent' ? undefined : submitted.value; + const nextValue = next.kind === 'recent' ? undefined : next.value; + const nextSummaryStart = + next.kind === 'user' || next.kind === 'org' ? next.summaryStart : undefined; + const nextSummaryEnd = + next.kind === 'user' || next.kind === 'org' ? next.summaryEnd : undefined; + const submittedSummaryStart = + submitted.kind === 'user' || submitted.kind === 'org' + ? submitted.summaryStart + : undefined; + const submittedSummaryEnd = + submitted.kind === 'user' || submitted.kind === 'org' + ? submitted.summaryEnd + : undefined; const unchanged = cursor === undefined && submitted.kind === next.kind && - submitted.value === next.value && + submittedValue === nextValue && submitted.status === next.status && submitted.closeReason === next.closeReason && - submitted.skuId === next.skuId; + submitted.skuId === next.skuId && + submittedSummaryStart === nextSummaryStart && + submittedSummaryEnd === nextSummaryEnd; + setSummaryInputError(null); + setSummaryRequest( + next.kind === 'user' || next.kind === 'org' + ? { + subjectType: next.kind, + subjectId: next.value, + start: next.summaryStart ?? '', + end: next.summaryEnd ?? '', + } + : null + ); + // Sync the URL to the applied filter (not the draft dropdown value) so a + // submitted search can be bookmarked/deep-linked without re-triggering a + // search merely from changing the Close reason dropdown before submitting. + replaceCloseReasonParam(next.closeReason); setSubmitted(next); resetResultNavigation(); if (unchanged) void results.refetch(); }} > -
- - -
-
- - setValue(event.target.value)} - /> -
-
- - -
-
- - -
-
- - + {/* Primary bar: what to search for, and the actions that trigger it. + items-start (not items-end): every column below shares the same + label + gap + control rhythm, so their controls line up on the + same top edge regardless of any incidental height differences + inside a column (e.g. Radix Select's hidden native setKind(next as SearchKind)}> + + + + + Interval ID + User ID + Organization ID + + +
+
+ + setValue(event.target.value)} + /> +
+
+ +
+ + +
+
-
- - + + {/* Filters: narrow the search above. Wraps freely; never forces the actions off-card. */} +
+
+ + +
+
+ + +
+
+ + +
+ {(kind === 'user' || kind === 'org') && ( +
+ +
+ setSummaryStart(event.target.value)} + /> + setSummaryEnd(event.target.value)} + /> +
+ {summaryInputError && ( + + )} +
+ )}
+ {(submitted.kind === 'user' || submitted.kind === 'org') && ( + + + Usage summary + + Accepted seconds and shadow estimated cents for this exact subject. This does not + debit credits. + + + + {summaryRequest && ( +

+ {summaryRequest.subjectType} {summaryRequest.subjectId} ·{' '} + {formatTimestamp(summaryRequest.start)} to {formatTimestamp(summaryRequest.end)} · + window is [start, end) +

+ )} + + {summary.isFetching && ( +

+ Calculating usage summary... +

+ )} + + {summary.isError && ( + + Usage summary could not be calculated + +

{summary.error.message}

+ {summary.error.data?.code !== 'BAD_REQUEST' && ( + + )} + + + )} + + {summary.isSuccess && ( +
+

+ {summary.data.acceptedSeconds.toLocaleString()} accepted seconds across{' '} + {summary.data.items + .reduce((total, item) => total + item.intervals, 0) + .toLocaleString()}{' '} + interval + {summary.data.items.reduce((total, item) => total + item.intervals, 0) === 1 + ? '' + : 's'} +

+ {summary.data.items.length === 0 ? ( +

+ No accepted usage was recorded in this window. +

+ ) : ( +
+ + + + + SKU + Accepted seconds + Rate + Estimated cents + + + + {summary.data.items.map(item => ( + + + {item.skuId} +

{item.skuName}

+ + + {item.acceptedSeconds.toLocaleString()}s + + + {item.rateCentsPerSecond} cents/s + + + {item.estimatedCents} cents + + + ))} + + Total + + {summary.data.acceptedSeconds.toLocaleString()}s + + + + {summary.data.estimatedCents} cents + + + +
+ Usage summary for {summary.data.subjectType} {summary.data.subjectId} from{' '} + {formatTimestamp(summary.data.start)} to {formatTimestamp(summary.data.end)} +
+
+ )} +
+ )} + + + )} + {results.isError && ( Usage records could not be loaded diff --git a/apps/web/src/routers/admin/cloud-billing-skus-router.test.ts b/apps/web/src/routers/admin/cloud-billing-skus-router.test.ts index d15fff5f65..8689f249be 100644 --- a/apps/web/src/routers/admin/cloud-billing-skus-router.test.ts +++ b/apps/web/src/routers/admin/cloud-billing-skus-router.test.ts @@ -209,22 +209,180 @@ describe('admin.cloudBillingSkus usage records', () => { await insertUsageInterval({ id: 'interval-b', subjectId: 'subject-1' }); await insertUsageInterval({ id: 'interval-a', subjectId: 'subject-1' }); await insertUsageInterval({ id: 'interval-other', subjectId: 'subject-2' }); + await db.insert(container_usage_segment).values([ + { + interval_id: 'interval-b', + seq: 1, + idempotency_key: 'interval-b-segment', + reported_seconds: 1, + usage_seconds: 1, + received_at: '2026-07-22T10:01:00.000Z', + }, + { + interval_id: 'interval-a', + seq: 1, + idempotency_key: 'interval-a-segment', + reported_seconds: 1, + usage_seconds: 1, + received_at: '2026-07-22T10:01:00.000Z', + }, + { + interval_id: 'interval-other', + seq: 1, + idempotency_key: 'interval-other-segment', + reported_seconds: 1, + usage_seconds: 1, + received_at: '2026-07-22T10:01:00.000Z', + }, + ]); const caller = await createCallerForUser(admin.id); const first = await caller.admin.cloudBillingSkus.searchUsageIntervals({ - search: { kind: 'subject', subjectType: 'user', subjectId: 'subject-1' }, + search: { + kind: 'subject', + subjectType: 'user', + subjectId: 'subject-1', + start: '2026-07-22T09:00:00.000Z', + end: '2026-07-22T11:00:00.000Z', + }, limit: 1, }); expect(first.items.map(item => item.id)).toEqual(['interval-b']); expect(first.nextCursor).not.toBeNull(); if (!first.nextCursor) throw new Error('Expected an interval cursor'); const second = await caller.admin.cloudBillingSkus.searchUsageIntervals({ - search: { kind: 'subject', subjectType: 'user', subjectId: 'subject-1' }, + search: { + kind: 'subject', + subjectType: 'user', + subjectId: 'subject-1', + start: '2026-07-22T09:00:00.000Z', + end: '2026-07-22T11:00:00.000Z', + }, limit: 1, cursor: first.nextCursor, }); expect(second.items.map(item => item.id)).toEqual(['interval-a']); }); + it('summarizes only an exact subject and bounded activity window by SKU', async () => { + await db.insert(cloud_billing_sku).values({ + ...validInput('usage-summary-sku'), + rate_cents_per_unit: '0.125', + created_by_user_id: admin.id, + }); + await insertUsageInterval({ + id: 'summary-in-window', + subjectId: 'summary-subject', + startedAt: '2026-07-22T10:00:00.000Z', + }); + await db + .update(container_usage_interval) + .set({ cloud_billing_sku_id: 'usage-summary-sku', confirmed_seconds: 999 }) + .where(eq(container_usage_interval.id, 'summary-in-window')); + await db.insert(container_usage_segment).values([ + { + interval_id: 'summary-in-window', + seq: 1, + idempotency_key: 'summary-start-boundary', + reported_seconds: 12, + usage_seconds: 12, + received_at: '2026-07-22T09:00:00.000Z', + }, + { + interval_id: 'summary-in-window', + seq: 2, + idempotency_key: 'summary-end-boundary', + reported_seconds: 100, + usage_seconds: 100, + received_at: '2026-07-22T11:00:00.000Z', + }, + ]); + await insertUsageInterval({ + id: 'summary-other-subject', + subjectId: 'other-subject', + startedAt: '2026-07-22T10:00:00.000Z', + }); + await db + .update(container_usage_interval) + .set({ confirmed_seconds: 999 }) + .where(eq(container_usage_interval.id, 'summary-other-subject')); + await db.insert(container_usage_segment).values({ + interval_id: 'summary-other-subject', + seq: 1, + idempotency_key: 'summary-other-subject', + reported_seconds: 999, + usage_seconds: 999, + received_at: '2026-07-22T10:00:00.000Z', + }); + await insertUsageInterval({ + id: 'summary-outside-window', + subjectId: 'summary-subject', + startedAt: '2026-07-23T10:00:00.000Z', + }); + await db + .update(container_usage_interval) + .set({ confirmed_seconds: 999 }) + .where(eq(container_usage_interval.id, 'summary-outside-window')); + await db.insert(container_usage_segment).values({ + interval_id: 'summary-outside-window', + seq: 1, + idempotency_key: 'summary-outside-window', + reported_seconds: 999, + usage_seconds: 999, + received_at: '2026-07-23T10:00:00.000Z', + }); + + const caller = await createCallerForUser(admin.id); + const summary = await caller.admin.cloudBillingSkus.getUsageSummary({ + subjectType: 'user', + subjectId: 'summary-subject', + start: '2026-07-22T09:00:00.000Z', + end: '2026-07-22T11:00:00.000Z', + }); + + expect(summary).toMatchObject({ + acceptedSeconds: 12, + estimatedCents: '1.5', + items: [ + { + skuId: 'usage-summary-sku', + acceptedSeconds: 12, + estimatedCents: '1.5', + intervals: 1, + }, + ], + }); + + const nonAdminCaller = await createCallerForUser(nonAdmin.id); + await expect( + nonAdminCaller.admin.cloudBillingSkus.getUsageSummary({ + subjectType: 'user', + subjectId: 'summary-subject', + start: '2026-07-22T09:00:00.000Z', + end: '2026-07-22T11:00:00.000Z', + }) + ).rejects.toMatchObject({ code: 'FORBIDDEN' }); + }); + + it('rejects invalid and oversized usage summary windows', async () => { + const caller = await createCallerForUser(admin.id); + await expect( + caller.admin.cloudBillingSkus.getUsageSummary({ + subjectType: 'user', + subjectId: 'summary-subject', + start: '2026-07-22T11:00:00.000Z', + end: '2026-07-22T10:00:00.000Z', + }) + ).rejects.toMatchObject({ code: 'BAD_REQUEST' }); + await expect( + caller.admin.cloudBillingSkus.getUsageSummary({ + subjectType: 'user', + subjectId: 'summary-subject', + start: '2026-06-01T00:00:00.000Z', + end: '2026-07-03T00:00:00.000Z', + }) + ).rejects.toMatchObject({ code: 'BAD_REQUEST' }); + }); + it('returns ordered, safe segment details and rejects unknown intervals', async () => { await insertUsageInterval({ id: 'interval-segments' }); await db diff --git a/apps/web/src/routers/admin/cloud-billing-skus-router.ts b/apps/web/src/routers/admin/cloud-billing-skus-router.ts index 776e3147d0..7fbb444a17 100644 --- a/apps/web/src/routers/admin/cloud-billing-skus-router.ts +++ b/apps/web/src/routers/admin/cloud-billing-skus-router.ts @@ -38,6 +38,8 @@ const usageSearchSchema = z kind: z.literal('subject'), subjectType: z.enum(['user', 'org']), subjectId: z.string().trim().min(1).max(256), + start: z.iso.datetime(), + end: z.iso.datetime(), }), ]), status: z.enum(['open', 'closed']).optional(), @@ -58,7 +60,11 @@ const usageSearchSchema = z .optional(), limit: z.number().int().min(1).max(100).default(25), }) - .strict(); + .strict() + .superRefine((input, context) => { + if (input.search.kind !== 'subject') return; + validateUsageWindow(input.search.start, input.search.end, context); + }); const segmentSearchSchema = z .object({ @@ -68,6 +74,18 @@ const segmentSearchSchema = z }) .strict(); +const usageSummarySchema = z + .object({ + subjectType: z.enum(['user', 'org']), + subjectId: z.string().trim().min(1).max(256), + start: z.iso.datetime(), + end: z.iso.datetime(), + }) + .strict() + .superRefine((input, context) => { + validateUsageWindow(input.start, input.end, context); + }); + const BILLING_HEALTH_WINDOW_MS = 24 * 60 * 60 * 1_000; const STALE_OPEN_INTERVAL_MS = 15 * 60 * 1_000; const usageMetadataSchema = z @@ -77,6 +95,22 @@ const usageMetadataSchema = z 'Metadata may contain at most 16 entries' ); +function validateUsageWindow(startValue: string, endValue: string, context: z.RefinementCtx): void { + const start = new Date(startValue).getTime(); + const end = new Date(endValue).getTime(); + if (end <= start) { + context.addIssue({ code: 'custom', path: ['end'], message: 'End must be after start' }); + return; + } + if (end - start > 31 * 24 * 60 * 60 * 1_000) { + context.addIssue({ + code: 'custom', + path: ['end'], + message: 'Usage windows may not exceed 31 days', + }); + } +} + export type SerializedUsageInterval = Pick< ContainerUsageInterval, | 'id' @@ -212,7 +246,13 @@ export const cloudBillingSkusRouter = createTRPCRouter({ } else { predicates.push( eq(container_usage_interval.subject_type, input.search.subjectType), - eq(container_usage_interval.subject_id, input.search.subjectId) + eq(container_usage_interval.subject_id, input.search.subjectId), + sql`exists ( + select 1 from ${container_usage_segment} + where ${container_usage_segment.interval_id} = ${container_usage_interval.id} + and ${container_usage_segment.received_at} >= ${input.search.start} + and ${container_usage_segment.received_at} < ${input.search.end} + )` ); } if (input.status) predicates.push(eq(container_usage_interval.status, input.status)); @@ -278,6 +318,69 @@ export const cloudBillingSkusRouter = createTRPCRouter({ }; }), + getUsageSummary: adminProcedure.input(usageSummarySchema).query(async ({ input }) => { + const rows = await db + .select({ + skuId: container_usage_interval.cloud_billing_sku_id, + skuName: cloud_billing_sku.name, + rateCentsPerSecond: cloud_billing_sku.rate_cents_per_unit, + acceptedSeconds: + sql`coalesce(sum(${container_usage_segment.usage_seconds}), 0)`.mapWith(Number), + intervals: sql`count(distinct ${container_usage_interval.id})`.mapWith(Number), + estimatedCents: + sql`coalesce(sum(${container_usage_segment.usage_seconds}::numeric * ${cloud_billing_sku.rate_cents_per_unit}), 0)::text`.mapWith( + String + ), + totalAcceptedSeconds: + sql`sum(sum(${container_usage_segment.usage_seconds})) over ()`.mapWith(Number), + totalEstimatedCents: + sql`sum(sum(${container_usage_segment.usage_seconds}::numeric * ${cloud_billing_sku.rate_cents_per_unit})) over ()::text`.mapWith( + String + ), + }) + .from(container_usage_segment) + .innerJoin( + container_usage_interval, + eq(container_usage_segment.interval_id, container_usage_interval.id) + ) + .innerJoin( + cloud_billing_sku, + eq(container_usage_interval.cloud_billing_sku_id, cloud_billing_sku.id) + ) + .where( + and( + eq(container_usage_interval.subject_type, input.subjectType), + eq(container_usage_interval.subject_id, input.subjectId), + sql`${container_usage_segment.received_at} >= ${input.start}`, + lt(container_usage_segment.received_at, input.end) + ) + ) + .groupBy( + container_usage_interval.cloud_billing_sku_id, + cloud_billing_sku.name, + cloud_billing_sku.rate_cents_per_unit + ) + .orderBy(container_usage_interval.cloud_billing_sku_id); + const items = rows.map(row => ({ + skuId: row.skuId, + skuName: row.skuName, + rateCentsPerSecond: normalizeCloudBillingSkuRate(row.rateCentsPerSecond), + acceptedSeconds: row.acceptedSeconds, + estimatedCents: normalizeCloudBillingSkuRate(row.estimatedCents), + intervals: row.intervals, + })); + const totals = rows[0]; + return { + subjectType: input.subjectType, + subjectId: input.subjectId, + start: input.start, + end: input.end, + items, + acceptedSeconds: totals?.totalAcceptedSeconds ?? 0, + estimatedCents: normalizeCloudBillingSkuRate(totals?.totalEstimatedCents ?? '0'), + }; + }), + usageHealth: adminProcedure.query(async () => { const end = new Date(); const start = new Date(end.getTime() - BILLING_HEALTH_WINDOW_MS); diff --git a/services/container-usage-meter/src/meter.test.ts b/services/container-usage-meter/src/meter.test.ts index bf065d46cb..c33245ebd6 100644 --- a/services/container-usage-meter/src/meter.test.ts +++ b/services/container-usage-meter/src/meter.test.ts @@ -84,6 +84,7 @@ beforeEach(() => { describe('ContainerUsageMeter', () => { it('writes an admitted start directly to PostgreSQL', async () => { + const log = vi.spyOn(console, 'log').mockImplementation(() => undefined); await expect(createMeter().recordStart(validStart())).resolves.toEqual({ success: true, ack: { intervalId: 'cloud-agent-next:instance-1:123', durable: 'pg', dedup: false }, @@ -95,9 +96,20 @@ describe('ContainerUsageMeter', () => { expect.stringMatching(/^[a-f0-9]{64}$/), expect.any(Number) ); + expect(log).toHaveBeenCalledWith( + JSON.stringify({ + message: 'Container usage meter RPC completed', + event: 'container_usage_rpc', + operation: 'start', + service: 'cloud-agent-next', + outcome: 'accepted', + dedup: false, + }) + ); }); it('returns permanent SKU admission failures', async () => { + const log = vi.spyOn(console, 'log').mockImplementation(() => undefined); vi.mocked(applyStart).mockResolvedValue({ kind: 'rejected', code: 'sku_not_accepting_new_usage', @@ -110,6 +122,9 @@ describe('ContainerUsageMeter', () => { message: 'Billing SKU is not accepting new usage', }, }); + expect(log).toHaveBeenCalledWith( + expect.stringContaining('"rejectionCode":"sku_not_accepting_new_usage"') + ); }); it('writes heartbeats directly and returns the shadow budget verdict', async () => { @@ -132,9 +147,20 @@ describe('ContainerUsageMeter', () => { }); it('propagates transient PostgreSQL failures for bounded client retry', async () => { + const log = vi.spyOn(console, 'log').mockImplementation(() => undefined); vi.mocked(applyHeartbeat).mockRejectedValue(new Error('postgres unavailable')); await expect(createMeter().recordHeartbeat(validHeartbeat())).rejects.toThrow( 'postgres unavailable' ); + expect(log).toHaveBeenCalledWith( + JSON.stringify({ + message: 'Container usage meter RPC completed', + event: 'container_usage_rpc', + operation: 'heartbeat', + service: 'cloud-agent-next', + outcome: 'failed', + errorName: 'Error', + }) + ); }); }); diff --git a/services/container-usage-meter/src/meter.ts b/services/container-usage-meter/src/meter.ts index 7b8bf614ee..68ccf1092d 100644 --- a/services/container-usage-meter/src/meter.ts +++ b/services/container-usage-meter/src/meter.ts @@ -45,6 +45,26 @@ function copyUsageContext(context: UsageContext): UsageContext { }; } +type MeterOperation = 'start' | 'heartbeat' | 'stop'; + +function logRpcOutcome( + operation: MeterOperation, + service: string, + outcome: 'accepted' | 'rejected' | 'failed', + details: { dedup?: boolean; rejectionCode?: string; errorName?: string } = {} +): void { + console.log( + JSON.stringify({ + message: 'Container usage meter RPC completed', + event: 'container_usage_rpc', + operation, + service, + outcome, + ...details, + }) + ); +} + export class ContainerUsageMeter extends WorkerEntrypoint implements ContainerUsageRpcMethods @@ -57,17 +77,27 @@ export class ContainerUsageMeter ); const context = copyUsageContext(parsed); const id = intervalId(parsed.service, parsed.instanceId, parsed.startEpochMs); - const result = await applyStart( - this.env, - parsed, - id, - await usageContextFingerprint(context), - Date.now() - ); + let result: Awaited>; + try { + result = await applyStart( + this.env, + parsed, + id, + await usageContextFingerprint(context), + Date.now() + ); + } catch (error) { + logRpcOutcome('start', parsed.service, 'failed', { + errorName: error instanceof Error ? error.name : 'UnknownError', + }); + throw error; + } switch (result.kind) { case 'rejected': + logRpcOutcome('start', parsed.service, 'rejected', { rejectionCode: result.code }); return { success: false, error: { code: result.code, message: result.message } }; case 'applied': + logRpcOutcome('start', parsed.service, 'accepted', { dedup: result.dedup }); return { success: true, ack: { intervalId: id, durable: 'pg', dedup: result.dedup }, @@ -83,13 +113,22 @@ export class ContainerUsageMeter heartbeatIdempotencyKey(parsed.service, parsed.instanceId, parsed.startEpochMs, parsed.seq) ); const id = intervalId(parsed.service, parsed.instanceId, parsed.startEpochMs); - const result = await applyHeartbeat( - this.env, - parsed, - id, - await usageContextFingerprint(parsed.context), - Date.now() - ); + let result: Awaited>; + try { + result = await applyHeartbeat( + this.env, + parsed, + id, + await usageContextFingerprint(parsed.context), + Date.now() + ); + } catch (error) { + logRpcOutcome('heartbeat', parsed.service, 'failed', { + errorName: error instanceof Error ? error.name : 'UnknownError', + }); + throw error; + } + logRpcOutcome('heartbeat', parsed.service, 'accepted', { dedup: result.dedup }); return { intervalId: id, durable: 'pg', @@ -106,13 +145,22 @@ export class ContainerUsageMeter stopIdempotencyKey(parsed.service, parsed.instanceId, parsed.startEpochMs) ); const id = intervalId(parsed.service, parsed.instanceId, parsed.startEpochMs); - const result = await applyStop( - this.env, - parsed, - id, - await usageContextFingerprint(parsed.context), - Date.now() - ); + let result: Awaited>; + try { + result = await applyStop( + this.env, + parsed, + id, + await usageContextFingerprint(parsed.context), + Date.now() + ); + } catch (error) { + logRpcOutcome('stop', parsed.service, 'failed', { + errorName: error instanceof Error ? error.name : 'UnknownError', + }); + throw error; + } + logRpcOutcome('stop', parsed.service, 'accepted', { dedup: result.dedup }); return { intervalId: id, durable: 'pg', dedup: result.dedup }; } } diff --git a/services/container-usage-meter/src/reconciliation.test.ts b/services/container-usage-meter/src/reconciliation.test.ts new file mode 100644 index 0000000000..a552c90722 --- /dev/null +++ b/services/container-usage-meter/src/reconciliation.test.ts @@ -0,0 +1,34 @@ +import { beforeEach, describe, expect, it, vi } from 'vitest'; + +vi.mock('./postgres', () => ({ + reconcileStaleIntervals: vi.fn(), +})); + +import { reconcileStaleIntervals } from './postgres'; +import { runReconciliation } from './reconciliation'; + +beforeEach(() => { + vi.clearAllMocks(); +}); + +describe('runReconciliation', () => { + it('logs completed runs and unconfirmed closes', async () => { + const log = vi.spyOn(console, 'log').mockImplementation(() => undefined); + vi.mocked(reconcileStaleIntervals).mockResolvedValue(3); + + await runReconciliation({} as Cloudflare.Env); + + expect(log).toHaveBeenCalledWith(expect.stringContaining('"outcome":"completed"')); + expect(log).toHaveBeenCalledWith(expect.stringContaining('"reconciledIntervals":3')); + }); + + it('logs and propagates failed runs', async () => { + const error = vi.spyOn(console, 'error').mockImplementation(() => undefined); + vi.mocked(reconcileStaleIntervals).mockRejectedValue(new Error('postgres unavailable')); + + await expect(runReconciliation({} as Cloudflare.Env)).rejects.toThrow('postgres unavailable'); + + expect(error).toHaveBeenCalledWith(expect.stringContaining('"outcome":"failed"')); + expect(error).toHaveBeenCalledWith(expect.stringContaining('"errorName":"Error"')); + }); +}); diff --git a/services/container-usage-meter/src/reconciliation.ts b/services/container-usage-meter/src/reconciliation.ts index 95b1f92a26..52ecfea6e7 100644 --- a/services/container-usage-meter/src/reconciliation.ts +++ b/services/container-usage-meter/src/reconciliation.ts @@ -3,12 +3,28 @@ import { reconcileStaleIntervals } from './postgres'; export const CONTAINER_USAGE_RECONCILIATION_CRON = '*/5 * * * *'; export async function runReconciliation(env: Cloudflare.Env): Promise { - const reconciledIntervals = await reconcileStaleIntervals(env); - console.log( - JSON.stringify({ - message: 'Container usage reconciliation completed', - event: 'container_usage_reconciliation', - reconciledIntervals, - }) - ); + const startedAt = Date.now(); + try { + const reconciledIntervals = await reconcileStaleIntervals(env); + console.log( + JSON.stringify({ + message: 'Container usage reconciliation completed', + event: 'container_usage_reconciliation', + outcome: 'completed', + reconciledIntervals, + durationMs: Date.now() - startedAt, + }) + ); + } catch (error) { + console.error( + JSON.stringify({ + message: 'Container usage reconciliation failed', + event: 'container_usage_reconciliation', + outcome: 'failed', + durationMs: Date.now() - startedAt, + errorName: error instanceof Error ? error.name : 'UnknownError', + }) + ); + throw error; + } }