Skip to content

fix(container-usage): recover missing meter intervals #4840

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
pandemicsyn merged 3 commits into from
Jul 28, 2026
Merged

fix(container-usage): recover missing meter intervals #4840

Show file tree
Hide file tree
Changes from 2 commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
08de600
fix(container-usage): recover missing meter intervals
pandemicsyn Jul 28, 2026
58df574
fix(container-usage): cap reconciled stop segments
pandemicsyn Jul 28, 2026
75a2ccd
fix(container-usage): validate recovered SKUs
pandemicsyn Jul 28, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
72 changes: 63 additions & 9 deletions services/container-usage-meter/src/postgres.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -126,6 +126,33 @@ function appliedUsageSeconds(
);
}

async function recoverMissingInterval(
Comment thread
pandemicsyn marked this conversation as resolved.
tx: Parameters<Parameters<WorkerDb['transaction']>[0]>[0],
intervalId: string,
startEpochMs: number,
context: UsageContext,
contextFingerprint: string,
receivedAtMs: number
): Promise<typeof container_usage_interval.$inferSelect> {
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,
Expand Down Expand Up @@ -256,13 +283,22 @@ export async function applyHeartbeatWithDb(
receivedAtMs: number
): Promise<ApplyResult> {
const operation: Promise<ApplyResult> = 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
Expand Down Expand Up @@ -364,14 +400,23 @@ export async function applyStopWithDb(
input.startEpochMs,
input.seq
);
return db.transaction(async tx => {
const [interval] = await tx
const operation: Promise<ApplyResult> = db.transaction(async 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()
Expand Down Expand Up @@ -405,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,
Expand All @@ -416,7 +464,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)
Expand All @@ -425,14 +476,17 @@ 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 };
});
return operation.catch(mapSingleOpenIntervalConflict);
}

export async function reconcileStaleIntervals(
Expand Down
203 changes: 179 additions & 24 deletions services/container-usage-meter/test/postgres.test.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -171,36 +171,15 @@ 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,
{
service: staleContext.service,
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,
},
Expand All @@ -215,13 +194,189 @@ 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,
});
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));
});

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 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));
});

it('does not let an older start supersede a newer generation', async () => {
const generationInstance = `generation-${suffix}`;
const generationContext = { ...context, instanceId: generationInstance };
Expand Down