From 08de6007c14cb12e2d22a13c3f09d6dab83cf88e Mon Sep 17 00:00:00 2001 From: syn Date: Tue, 28 Jul 2026 12:20:23 -0500 Subject: [PATCH 1/3] fix(container-usage): recover missing meter intervals --- .../container-usage-meter/src/postgres.ts | 64 ++++++- .../test/postgres.test.ts | 174 +++++++++++++++--- 2 files changed, 207 insertions(+), 31 deletions(-) diff --git a/services/container-usage-meter/src/postgres.ts b/services/container-usage-meter/src/postgres.ts index 48d4b7a09e..81cd40e0fd 100644 --- a/services/container-usage-meter/src/postgres.ts +++ b/services/container-usage-meter/src/postgres.ts @@ -126,6 +126,33 @@ function appliedUsageSeconds( ); } +async function recoverMissingInterval( + tx: Parameters[0]>[0], + intervalId: string, + startEpochMs: number, + context: UsageContext, + contextFingerprint: string, + receivedAtMs: number +): Promise { + const receivedAt = timestamp(receivedAtMs); + const [inserted] = await tx + .insert(container_usage_interval) + .values(intervalValues(intervalId, startEpochMs, context, contextFingerprint, receivedAt)) + .onConflictDoNothing({ target: container_usage_interval.id }) + .returning(); + if (inserted) return inserted; + + const [existing] = await tx + .select() + .from(container_usage_interval) + .where(eq(container_usage_interval.id, intervalId)) + .for('update') + .limit(1); + if (!existing) throw new Error('Container usage interval recovery lost without a winner'); + assertMatchingContext(existing, context, contextFingerprint); + return existing; +} + export async function applyStart( env: Cloudflare.Env, input: RecordStartInput, @@ -256,13 +283,22 @@ export async function applyHeartbeatWithDb( receivedAtMs: number ): Promise { const operation: Promise = db.transaction(async tx => { - const [interval] = await tx + const [existingInterval] = await tx .select() .from(container_usage_interval) .where(eq(container_usage_interval.id, intervalId)) .for('update') .limit(1); - if (!interval) throw new UsageIntervalNotFoundError(intervalId); + const interval = + existingInterval ?? + (await recoverMissingInterval( + tx, + intervalId, + input.startEpochMs, + input.context, + contextFingerprint, + receivedAtMs + )); assertMatchingContext(interval, input.context, contextFingerprint); const [existingSegment] = await tx @@ -365,13 +401,22 @@ export async function applyStopWithDb( input.seq ); return db.transaction(async tx => { - const [interval] = await tx + const [existingInterval] = await tx .select() .from(container_usage_interval) .where(eq(container_usage_interval.id, intervalId)) .for('update') .limit(1); - if (!interval) throw new UsageIntervalNotFoundError(intervalId); + const interval = + existingInterval ?? + (await recoverMissingInterval( + tx, + intervalId, + input.startEpochMs, + input.context, + contextFingerprint, + receivedAtMs + )); assertMatchingContext(interval, input.context, contextFingerprint); const [existingSegment] = await tx .select() @@ -416,7 +461,10 @@ export async function applyStopWithDb( }); } - const stopAt = timestamp(Math.max(new Date(interval.last_seen_at).getTime(), receivedAtMs)); + const wasReconciled = interval.status === 'closed' && interval.close_reason === 'unconfirmed'; + const stopAt = wasReconciled + ? (interval.stopped_at ?? interval.last_seen_at) + : timestamp(Math.max(new Date(interval.last_seen_at).getTime(), receivedAtMs)); await tx .update(container_usage_interval) @@ -425,10 +473,12 @@ export async function applyStopWithDb( close_reason: input.reason, exit_code: input.exitCode, final_stop_seq: input.seq, - last_seen_at: stopAt, + last_seen_at: wasReconciled ? interval.last_seen_at : stopAt, stopped_at: stopAt, last_heartbeat_seq: sql`GREATEST(${container_usage_interval.last_heartbeat_seq}, ${input.seq})`, - confirmed_seconds: interval.confirmed_seconds + finalSeconds, + confirmed_seconds: wasReconciled + ? interval.confirmed_seconds + : interval.confirmed_seconds + finalSeconds, }) .where(eq(container_usage_interval.id, intervalId)); return { kind: 'applied', dedup: false }; diff --git a/services/container-usage-meter/test/postgres.test.ts b/services/container-usage-meter/test/postgres.test.ts index 1b7285698d..fde6ac3b1b 100644 --- a/services/container-usage-meter/test/postgres.test.ts +++ b/services/container-usage-meter/test/postgres.test.ts @@ -171,27 +171,6 @@ describe('container usage PostgreSQL application', () => { stopped_at: stale.last_seen_at, confirmed_seconds: 0, }); - const lateHeartbeat = { - service: staleContext.service, - instanceId: staleContext.instanceId, - startEpochMs: 456, - idempotencyKey: heartbeatIdempotencyKey( - staleContext.service, - staleContext.instanceId, - 456, - 1 - ), - seq: 1, - usageSinceLast: 5, - context: staleContext, - }; - await applyHeartbeatWithDb( - client.db, - lateHeartbeat, - staleId, - staleFingerprint, - 20 * 60_000 + 5_000 - ); await applyStopWithDb( client.db, { @@ -199,8 +178,8 @@ describe('container usage PostgreSQL application', () => { instanceId: staleContext.instanceId, startEpochMs: 456, idempotencyKey: stopIdempotencyKey(staleContext.service, staleContext.instanceId, 456), - seq: 2, - usageSinceLast: 0, + seq: 1, + usageSinceLast: 5, reason: 'runtime_signal', context: staleContext, }, @@ -215,13 +194,160 @@ describe('container usage PostgreSQL application', () => { expect(corrected).toMatchObject({ status: 'closed', close_reason: 'runtime_signal', - confirmed_seconds: 5, + stopped_at: stale.stopped_at, + last_seen_at: stale.last_seen_at, + confirmed_seconds: stale.confirmed_seconds, }); await client.db .delete(container_usage_interval) .where(eq(container_usage_interval.id, staleId)); }); + it('recovers a missing interval from a heartbeat without reapplying SKU admission', async () => { + const recoveryContext = { ...context, instanceId: `heartbeat-recovery-${suffix}` }; + const recoveryFingerprint = await usageContextFingerprint(recoveryContext); + const recoveryId = `cloud-agent-next:${recoveryContext.instanceId}:789`; + await client.db + .update(cloud_billing_sku) + .set({ accepts_new_usage: false }) + .where(eq(cloud_billing_sku.id, skuId)); + + const heartbeat = { + service: recoveryContext.service, + instanceId: recoveryContext.instanceId, + startEpochMs: 789, + idempotencyKey: heartbeatIdempotencyKey( + recoveryContext.service, + recoveryContext.instanceId, + 789, + 1 + ), + seq: 1, + usageSinceLast: 300, + context: recoveryContext, + }; + await expect( + applyHeartbeatWithDb(client.db, heartbeat, recoveryId, recoveryFingerprint, 10_000) + ).resolves.toEqual({ kind: 'applied', dedup: false }); + await expect( + applyHeartbeatWithDb(client.db, heartbeat, recoveryId, recoveryFingerprint, 11_000) + ).resolves.toEqual({ kind: 'applied', dedup: true }); + + const [recovered] = await client.db + .select() + .from(container_usage_interval) + .where(eq(container_usage_interval.id, recoveryId)); + expect(recovered).toMatchObject({ + status: 'open', + confirmed_seconds: 0, + }); + expect(new Date(recovered.started_at).getTime()).toBe(10_000); + expect(new Date(recovered.last_seen_at).getTime()).toBe(10_000); + await expect( + applyHeartbeatWithDb( + client.db, + { ...heartbeat, context: { ...recoveryContext, sku: `${skuId}-other` } }, + recoveryId, + await usageContextFingerprint({ ...recoveryContext, sku: `${skuId}-other` }), + 12_000 + ) + ).rejects.toBeInstanceOf(UsageMutationConflictError); + await client.db + .delete(container_usage_interval) + .where(eq(container_usage_interval.id, recoveryId)); + }); + + it('recovers a missing interval from a stop at a zero-duration boundary', async () => { + const recoveryContext = { ...context, instanceId: `stop-recovery-${suffix}` }; + const recoveryFingerprint = await usageContextFingerprint(recoveryContext); + const recoveryId = `cloud-agent-next:${recoveryContext.instanceId}:790`; + const stop = { + service: recoveryContext.service, + instanceId: recoveryContext.instanceId, + startEpochMs: 790, + idempotencyKey: stopIdempotencyKey(recoveryContext.service, recoveryContext.instanceId, 790), + seq: 1, + usageSinceLast: 300, + reason: 'exit' as const, + exitCode: 0, + context: recoveryContext, + }; + await expect( + applyStopWithDb(client.db, stop, recoveryId, recoveryFingerprint, 20_000) + ).resolves.toEqual({ kind: 'applied', dedup: false }); + await expect( + applyStopWithDb(client.db, stop, recoveryId, recoveryFingerprint, 21_000) + ).resolves.toEqual({ kind: 'applied', dedup: true }); + + const [recovered] = await client.db + .select() + .from(container_usage_interval) + .where(eq(container_usage_interval.id, recoveryId)); + expect(recovered).toMatchObject({ + status: 'closed', + confirmed_seconds: 0, + close_reason: 'exit', + }); + expect(new Date(recovered.started_at).getTime()).toBe(20_000); + expect(new Date(recovered.last_seen_at).getTime()).toBe(20_000); + expect(new Date(recovered.stopped_at ?? 0).getTime()).toBe(20_000); + await client.db + .delete(container_usage_interval) + .where(eq(container_usage_interval.id, recoveryId)); + }); + + it('rejects missing-interval recovery when a newer generation owns the open slot', async () => { + await client.db + .update(cloud_billing_sku) + .set({ accepts_new_usage: true }) + .where(eq(cloud_billing_sku.id, skuId)); + const recoveryContext = { ...context, instanceId: `recovery-generation-${suffix}` }; + const recoveryFingerprint = await usageContextFingerprint(recoveryContext); + const newerId = `cloud-agent-next:${recoveryContext.instanceId}:900`; + await applyStartWithDb( + client.db, + { + ...recoveryContext, + startEpochMs: 900, + idempotencyKey: startIdempotencyKey( + recoveryContext.service, + recoveryContext.instanceId, + 900 + ), + }, + newerId, + recoveryFingerprint, + 30_000 + ); + + const olderId = `cloud-agent-next:${recoveryContext.instanceId}:800`; + await expect( + applyHeartbeatWithDb( + client.db, + { + service: recoveryContext.service, + instanceId: recoveryContext.instanceId, + startEpochMs: 800, + idempotencyKey: heartbeatIdempotencyKey( + recoveryContext.service, + recoveryContext.instanceId, + 800, + 1 + ), + seq: 1, + usageSinceLast: 1, + context: recoveryContext, + }, + olderId, + recoveryFingerprint, + 31_000 + ) + ).rejects.toThrow('Another usage interval is already open'); + await client.db + .delete(container_usage_interval) + .where(eq(container_usage_interval.id, newerId)); + }); + it('does not let an older start supersede a newer generation', async () => { const generationInstance = `generation-${suffix}`; const generationContext = { ...context, instanceId: generationInstance }; From 58df57426c90c6dc82b68d0116a6ed72ee59654f Mon Sep 17 00:00:00 2001 From: syn Date: Tue, 28 Jul 2026 12:28:55 -0500 Subject: [PATCH 2/3] fix(container-usage): cap reconciled stop segments --- .../container-usage-meter/src/postgres.ts | 8 +++-- .../test/postgres.test.ts | 29 +++++++++++++++++++ 2 files changed, 35 insertions(+), 2 deletions(-) diff --git a/services/container-usage-meter/src/postgres.ts b/services/container-usage-meter/src/postgres.ts index 81cd40e0fd..941ae8dd2f 100644 --- a/services/container-usage-meter/src/postgres.ts +++ b/services/container-usage-meter/src/postgres.ts @@ -400,7 +400,7 @@ export async function applyStopWithDb( input.startEpochMs, input.seq ); - return db.transaction(async tx => { + const operation: Promise = db.transaction(async tx => { const [existingInterval] = await tx .select() .from(container_usage_interval) @@ -450,7 +450,10 @@ export async function applyStopWithDb( throw new UsageMutationConflictError('Final usage segment has conflicting payload'); } } else { - finalSeconds = appliedUsageSeconds(interval, input.usageSinceLast, receivedAtMs); + finalSeconds = + interval.status === 'closed' && interval.close_reason === 'unconfirmed' + ? 0 + : appliedUsageSeconds(interval, input.usageSinceLast, receivedAtMs); await tx.insert(container_usage_segment).values({ interval_id: intervalId, seq: input.seq, @@ -483,6 +486,7 @@ export async function applyStopWithDb( .where(eq(container_usage_interval.id, intervalId)); return { kind: 'applied', dedup: false }; }); + return operation.catch(mapSingleOpenIntervalConflict); } export async function reconcileStaleIntervals( diff --git a/services/container-usage-meter/test/postgres.test.ts b/services/container-usage-meter/test/postgres.test.ts index fde6ac3b1b..753f548263 100644 --- a/services/container-usage-meter/test/postgres.test.ts +++ b/services/container-usage-meter/test/postgres.test.ts @@ -198,6 +198,12 @@ describe('container usage PostgreSQL application', () => { last_seen_at: stale.last_seen_at, confirmed_seconds: stale.confirmed_seconds, }); + const lateStopSegments = await client.db + .select() + .from(container_usage_segment) + .where(eq(container_usage_segment.interval_id, staleId)); + expect(lateStopSegments).toHaveLength(1); + expect(lateStopSegments[0]).toMatchObject({ reported_seconds: 5, usage_seconds: 0 }); await client.db .delete(container_usage_interval) .where(eq(container_usage_interval.id, staleId)); @@ -343,6 +349,29 @@ describe('container usage PostgreSQL application', () => { 31_000 ) ).rejects.toThrow('Another usage interval is already open'); + + await expect( + applyStopWithDb( + client.db, + { + service: recoveryContext.service, + instanceId: recoveryContext.instanceId, + startEpochMs: 800, + idempotencyKey: stopIdempotencyKey( + recoveryContext.service, + recoveryContext.instanceId, + 800 + ), + seq: 1, + usageSinceLast: 1, + reason: 'exit', + context: recoveryContext, + }, + olderId, + recoveryFingerprint, + 32_000 + ) + ).rejects.toBeInstanceOf(UsageMutationConflictError); await client.db .delete(container_usage_interval) .where(eq(container_usage_interval.id, newerId)); From 75a2ccdb16348bfa3d1a65e4c388ff4ac9688134 Mon Sep 17 00:00:00 2001 From: syn Date: Tue, 28 Jul 2026 13:32:15 -0500 Subject: [PATCH 3/3] fix(container-usage): validate recovered SKUs --- .../container-usage-meter/src/postgres.ts | 10 +++ .../test/postgres.test.ts | 67 +++++++++++++++++++ 2 files changed, 77 insertions(+) diff --git a/services/container-usage-meter/src/postgres.ts b/services/container-usage-meter/src/postgres.ts index 941ae8dd2f..0c57e15c70 100644 --- a/services/container-usage-meter/src/postgres.ts +++ b/services/container-usage-meter/src/postgres.ts @@ -134,6 +134,16 @@ async function recoverMissingInterval( contextFingerprint: string, receivedAtMs: number ): Promise { + const [sku] = await tx + .select({ unit: cloud_billing_sku.unit }) + .from(cloud_billing_sku) + .where(eq(cloud_billing_sku.id, context.sku)) + .limit(1); + if (!sku) throw new UsageMutationConflictError('Billing SKU not found during interval recovery'); + if (sku.unit !== 'second') { + throw new UsageMutationConflictError('Billing SKU is not measured in seconds'); + } + const receivedAt = timestamp(receivedAtMs); const [inserted] = await tx .insert(container_usage_interval) diff --git a/services/container-usage-meter/test/postgres.test.ts b/services/container-usage-meter/test/postgres.test.ts index 753f548263..f9beceed70 100644 --- a/services/container-usage-meter/test/postgres.test.ts +++ b/services/container-usage-meter/test/postgres.test.ts @@ -263,6 +263,73 @@ describe('container usage PostgreSQL application', () => { .where(eq(container_usage_interval.id, recoveryId)); }); + it('rejects missing-interval recovery for an unknown or non-second SKU', async () => { + const recoveryContext = { ...context, instanceId: `invalid-sku-recovery-${suffix}` }; + const unknownSkuContext = { ...recoveryContext, sku: `${skuId}-unknown` }; + const unknownId = `cloud-agent-next:${recoveryContext.instanceId}:791`; + await expect( + applyHeartbeatWithDb( + client.db, + { + service: recoveryContext.service, + instanceId: recoveryContext.instanceId, + startEpochMs: 791, + idempotencyKey: heartbeatIdempotencyKey( + recoveryContext.service, + recoveryContext.instanceId, + 791, + 1 + ), + seq: 1, + usageSinceLast: 1, + context: unknownSkuContext, + }, + unknownId, + await usageContextFingerprint(unknownSkuContext), + 13_000 + ) + ).rejects.toThrow('Billing SKU not found during interval recovery'); + + const nonSecondSkuId = `meter-test-request-${suffix}`; + await client.db.insert(cloud_billing_sku).values({ + id: nonSecondSkuId, + name: 'Non-container meter integration test', + unit: 'request', + rate_cents_per_unit: '0.000001', + }); + const nonSecondContext = { ...recoveryContext, sku: nonSecondSkuId }; + const nonSecondId = `cloud-agent-next:${recoveryContext.instanceId}:792`; + await expect( + applyStopWithDb( + client.db, + { + service: recoveryContext.service, + instanceId: recoveryContext.instanceId, + startEpochMs: 792, + idempotencyKey: stopIdempotencyKey( + recoveryContext.service, + recoveryContext.instanceId, + 792 + ), + seq: 1, + usageSinceLast: 1, + reason: 'exit', + context: nonSecondContext, + }, + nonSecondId, + await usageContextFingerprint(nonSecondContext), + 14_000 + ) + ).rejects.toThrow('Billing SKU is not measured in seconds'); + + const invalidIntervals = await client.db + .select() + .from(container_usage_interval) + .where(eq(container_usage_interval.instance_id, recoveryContext.instanceId)); + expect(invalidIntervals).toHaveLength(0); + await client.db.delete(cloud_billing_sku).where(eq(cloud_billing_sku.id, nonSecondSkuId)); + }); + it('recovers a missing interval from a stop at a zero-duration boundary', async () => { const recoveryContext = { ...context, instanceId: `stop-recovery-${suffix}` }; const recoveryFingerprint = await usageContextFingerprint(recoveryContext);