fix: rollover reset

.opencode/opencode.json

@@ -9,6 +9,11 @@
 		"mintlify": {
 			"type": "remote",
 			"url": "https://mintlify.com/docs/mcp"
+		},
+
+		"planetscale": {
+			"type": "remote",
+			"url": "https://mcp.pscale.dev/mcp/planetscale"
 		}
 	}
 }

server/src/cron/resetCron/runResetCron.ts

@@ -29,12 +29,12 @@ export const runResetCron = async ({ ctx }: { ctx: CronContext }) => {
 				limit: 5_000,
 			});
 
-			if (cusEnts.length < 5_000) {
-				console.log(
-					`Reset cron: only ${cusEnts.length} entitlements to reset, skipping (lazy reset will handle)`,
-				);
-				break;
-			}
+			// if (cusEnts.length < 5_000) {
+			// 	console.log(
+			// 		`Reset cron: only ${cusEnts.length} entitlements to reset, skipping (lazy reset will handle)`,
+			// 	);
+			// 	break;
+			// }
 
 			console.log(
 				`Reset cron iteration ${iteration}: processing ${cusEnts.length} entitlements`,

server/src/internal/customers/actions/resetCustomerEntitlements/applyResetResults.ts

@@ -1,8 +1,4 @@
-import type {
-	FullCustomer,
-	FullCustomerEntitlement,
-	Rollover,
-} from "@autumn/shared";
+import type { FullCustomer, FullCustomerEntitlement } from "@autumn/shared";
 import type { AutumnContext } from "@/honoUtils/HonoEnv.js";
 import { RolloverService } from "@/internal/customers/cusProducts/cusEnts/cusRollovers/RolloverService.js";
 import type { ProcessResetResult } from "./processReset.js";
@@ -29,6 +25,7 @@ const findCusEnt = ({
 /**
  * Applies computed reset values to in-memory FullCustomer for all cusEnts,
  * and runs rollover max-clearing only for DB-applied (non-skipped) ones.
+ * For skipped entries (another request won the race), re-reads rollovers from DB.
  */
 export const applyResetResults = async ({
 	ctx,
@@ -43,7 +40,6 @@ export const applyResetResults = async ({
 }): Promise<void> => {
 	const { db } = ctx;
 	const skippedSet = new Set(skipped);
-	const clearingPromises: Promise<Rollover[]>[] = [];
 
 	for (const { cusEntId, result } of computed) {
 		const original = findCusEnt({ fullCus, cusEntId });
@@ -57,20 +53,24 @@ export const applyResetResults = async ({
 		if (updates.entities !== null) original.entities = updates.entities;
 		original.next_reset_at = updates.next_reset_at;
 
-		// Only run rollover clearing for DB-applied entries.
-		// Skipped entries were already cleared by the winning request.
-		if (!skippedSet.has(cusEntId) && result.rolloverInsert) {
-			clearingPromises.push(
-				RolloverService.clearExcessRollovers({
-					db,
-					newRows: result.rolloverInsert.rows,
-					fullCusEnt: original,
-				}),
-			);
+		if (!result.rolloverInsert) continue;
+
+		if (!skippedSet.has(cusEntId)) {
+			// Winner: we inserted the rollover into DB. Clear excess and
+			// update the in-memory array to include the new rollovers.
+			const clearedRollovers = await RolloverService.clearExcessRollovers({
+				db,
+				newRows: result.rolloverInsert.rows,
+				fullCusEnt: original,
+			});
+			original.rollovers = clearedRollovers;
+		} else {
+			// Loser: the winning request already inserted the rollover and
+			// cleared excess. Re-read from DB to get the authoritative state.
+			original.rollovers = await RolloverService.getCurrentRollovers({
+				db,
+				cusEntID: cusEntId,
+			});
 		}
 	}
-
-	if (clearingPromises.length > 0) {
-		await Promise.all(clearingPromises);
-	}
 };

server/tests/integration/crud/customers/reset-customer-entitlements/get-customer-reset-rollovers-concurrent.test.ts

@@ -0,0 +1,202 @@
+import { expect, test } from "bun:test";
+import {
+	type ApiCustomer,
+	type CheckResponseV2,
+	RolloverExpiryDurationType,
+} from "@autumn/shared";
+import { findCustomerEntitlement } from "@tests/balances/utils/findCustomerEntitlement.js";
+import { TestFeature } from "@tests/setup/v2Features.js";
+import { expireCusEntForReset } from "@tests/utils/cusProductUtils/resetTestUtils.js";
+import { items } from "@tests/utils/fixtures/items.js";
+import { products } from "@tests/utils/fixtures/products.js";
+import { initScenario, s } from "@tests/utils/testInitUtils/initScenario.js";
+import chalk from "chalk";
+
+const rolloverConfig = {
+	max: 500,
+	length: 1,
+	duration: RolloverExpiryDurationType.Month,
+};
+
+// ─────────────────────────────────────────────────────────────────
+// Concurrent GET /customers with rollover — reset exactly once
+// ─────────────────────────────────────────────────────────────────
+
+test.concurrent(`${chalk.yellowBright("concurrent rollover reset: multiple GETs all return reset balance with rollover")}`, async () => {
+	const messagesItem = items.monthlyMessagesWithRollover({
+		includedUsage: 400,
+		rolloverConfig,
+	});
+	const pro = products.pro({ items: [messagesItem] });
+
+	const { customerId, autumnV2, ctx } = await initScenario({
+		customerId: "reset-rollover-conc-get",
+		setup: [
+			s.customer({ paymentMethod: "success", testClock: false }),
+			s.products({ list: [pro] }),
+		],
+		actions: [
+			s.attach({ productId: pro.id }),
+			s.track({ featureId: TestFeature.Messages, value: 250, timeout: 2000 }),
+		],
+	});
+
+	// Before reset: 400 - 250 = 150 remaining
+	const before = await autumnV2.customers.get<ApiCustomer>(customerId, {
+		skip_cache: "true",
+	});
+	expect(before.balances[TestFeature.Messages].current_balance).toBe(150);
+
+	await expireCusEntForReset({
+		ctx,
+		customerId,
+		featureId: TestFeature.Messages,
+	});
+
+	// Fire 5 concurrent GET requests — all should see reset balance with rollover
+	// Expected: rollover = min(150, 500) = 150, fresh grant = 400, total = 550
+	const results = await Promise.all(
+		Array.from({ length: 5 }, () =>
+			autumnV2.customers.get<ApiCustomer>(customerId),
+		),
+	);
+
+	for (const customer of results) {
+		expect(customer.balances[TestFeature.Messages].current_balance).toBe(550);
+		expect(customer.balances[TestFeature.Messages].usage).toBe(0);
+		expect(customer.balances[TestFeature.Messages].rollovers).toBeDefined();
+		expect(customer.balances[TestFeature.Messages].rollovers!.length).toBe(1);
+		expect(customer.balances[TestFeature.Messages].rollovers![0].balance).toBe(
+			150,
+		);
+	}
+
+	// DB should also reflect the reset (only applied once)
+	const cusEntAfter = await findCustomerEntitlement({
+		ctx,
+		customerId,
+		featureId: TestFeature.Messages,
+	});
+	expect(cusEntAfter).toBeDefined();
+	expect(cusEntAfter!.next_reset_at).toBeGreaterThan(Date.now());
+	expect(cusEntAfter!.rollovers.length).toBe(1);
+});
+
+// ─────────────────────────────────────────────────────────────────
+// Concurrent checks with rollover — all return reset balance
+// ─────────────────────────────────────────────────────────────────
+
+test.concurrent(`${chalk.yellowBright("concurrent rollover reset: multiple checks all return reset balance with rollover")}`, async () => {
+	const messagesItem = items.monthlyMessagesWithRollover({
+		includedUsage: 400,
+		rolloverConfig,
+	});
+	const pro = products.pro({ items: [messagesItem] });
+
+	const { customerId, autumnV2, ctx } = await initScenario({
+		customerId: "reset-rollover-conc-check",
+		setup: [
+			s.customer({ paymentMethod: "success", testClock: false }),
+			s.products({ list: [pro] }),
+		],
+		actions: [
+			s.attach({ productId: pro.id }),
+			s.track({ featureId: TestFeature.Messages, value: 100, timeout: 2000 }),
+		],
+	});
+
+	// Before reset: 400 - 100 = 300 remaining
+	await expireCusEntForReset({
+		ctx,
+		customerId,
+		featureId: TestFeature.Messages,
+	});
+
+	// Fire 5 concurrent check requests
+	// Expected: rollover = min(300, 500) = 300, fresh grant = 400, total = 700
+	const results = await Promise.all(
+		Array.from({ length: 5 }, () =>
+			autumnV2.check({
+				customer_id: customerId,
+				feature_id: TestFeature.Messages,
+			}),
+		),
+	);
+
+	for (const res of results) {
+		const check = res as unknown as CheckResponseV2;
+		expect(check.allowed).toBe(true);
+		expect(check.balance?.current_balance).toBe(700);
+		expect(check.balance?.usage).toBe(0);
+	}
+});
+
+// ─────────────────────────────────────────────────────────────────
+// Concurrent tracks with rollover — reset once, deductions atomic
+// ─────────────────────────────────────────────────────────────────
+
+test.concurrent(`${chalk.yellowBright("concurrent rollover reset: multiple tracks reset once then deduct atomically")}`, async () => {
+	const messagesItem = items.monthlyMessagesWithRollover({
+		includedUsage: 400,
+		rolloverConfig,
+	});
+	const pro = products.pro({ items: [messagesItem] });
+
+	const { customerId, autumnV2, ctx } = await initScenario({
+		customerId: "reset-rollover-conc-track",
+		setup: [
+			s.customer({ paymentMethod: "success", testClock: false }),
+			s.products({ list: [pro] }),
+		],
+		actions: [
+			s.attach({ productId: pro.id }),
+			s.track({ featureId: TestFeature.Messages, value: 200, timeout: 2000 }),
+		],
+	});
+
+	// Before reset: 400 - 200 = 200 remaining
+	await expireCusEntForReset({
+		ctx,
+		customerId,
+		featureId: TestFeature.Messages,
+	});
+
+	// Fire 5 concurrent tracks of 10 each
+	// Expected: rollover = min(200, 500) = 200, fresh grant = 400, total = 600
+	// Then deduct 50 total → 550
+	await Promise.all(
+		Array.from({ length: 5 }, () =>
+			autumnV2.track({
+				customer_id: customerId,
+				feature_id: TestFeature.Messages,
+				value: 10,
+			}),
+		),
+	);
+
+	// Verify final balance: 600 (reset) - 50 (5 * 10) = 550
+	const customer = await autumnV2.customers.get<ApiCustomer>(customerId);
+	expect(customer.balances[TestFeature.Messages].current_balance).toBe(550);
+	expect(customer.balances[TestFeature.Messages].usage).toBe(50);
+	expect(customer.balances[TestFeature.Messages].rollovers).toBeDefined();
+	expect(customer.balances[TestFeature.Messages].rollovers!.length).toBe(1);
+
+	// Wait for DB sync and verify DB agrees
+	await new Promise((resolve) => setTimeout(resolve, 2000));
+
+	const customerDb = await autumnV2.customers.get<ApiCustomer>(customerId, {
+		skip_cache: "true",
+	});
+	expect(customerDb.balances[TestFeature.Messages].current_balance).toBe(550);
+	expect(customerDb.balances[TestFeature.Messages].usage).toBe(50);
+
+	const cusEntAfter = await findCustomerEntitlement({
+		ctx,
+		customerId,
+		featureId: TestFeature.Messages,
+	});
+	expect(cusEntAfter).toBeDefined();
+	expect(cusEntAfter!.next_reset_at).toBeGreaterThan(Date.now());
+	// Only 1 rollover should exist (not duplicated by concurrent requests)
+	expect(cusEntAfter!.rollovers.length).toBe(1);
+});

server/tests/integration/crud/customers/reset-customer-entitlements/get-customer-reset-rollovers.test.ts

@@ -0,0 +1,149 @@
+import { expect, test } from "bun:test";
+import { type ApiCustomer, RolloverExpiryDurationType } from "@autumn/shared";
+import { findCustomerEntitlement } from "@tests/balances/utils/findCustomerEntitlement.js";
+import { TestFeature } from "@tests/setup/v2Features.js";
+import { expireCusEntForReset } from "@tests/utils/cusProductUtils/resetTestUtils.js";
+import { items } from "@tests/utils/fixtures/items.js";
+import { products } from "@tests/utils/fixtures/products.js";
+import { timeout } from "@tests/utils/genUtils";
+import { initScenario, s } from "@tests/utils/testInitUtils/initScenario.js";
+import chalk from "chalk";
+
+// ─────────────────────────────────────────────────────────────────
+// Lazy reset with rollovers (DB path) — GET /customers skip_cache
+//
+// Attach product with rollover config → track some usage → expire
+// cusEnt → GET customer (skip_cache) → verify lazy reset created
+// a rollover from the unused balance and refreshed the grant.
+// ─────────────────────────────────────────────────────────────────
+
+test.concurrent(`${chalk.yellowBright("lazy reset rollover (DB): creates rollover from unused balance on reset")}`, async () => {
+	const messagesItem = items.monthlyMessagesWithRollover({
+		includedUsage: 400,
+		rolloverConfig: {
+			max: 500,
+			length: 1,
+			duration: RolloverExpiryDurationType.Month,
+		},
+	});
+	const pro = products.pro({ items: [messagesItem] });
+
+	const { customerId, autumnV2, ctx } = await initScenario({
+		customerId: "reset-rollover-db",
+		setup: [
+			s.customer({ paymentMethod: "success", testClock: false }),
+			s.products({ list: [pro] }),
+		],
+		actions: [
+			s.attach({ productId: pro.id }),
+			s.track({ featureId: TestFeature.Messages, value: 250 }),
+		],
+	});
+
+	await timeout(2000);
+
+	// Before reset: 400 - 250 = 150 remaining
+	const before = await autumnV2.customers.get<ApiCustomer>(customerId, {
+		skip_cache: "true",
+	});
+	expect(before.balances[TestFeature.Messages].current_balance).toBe(150);
+	expect(before.balances[TestFeature.Messages].usage).toBe(250);
+
+	// Expire cusEnt so the next read triggers a lazy reset
+	await expireCusEntForReset({
+		ctx,
+		customerId,
+		featureId: TestFeature.Messages,
+	});
+
+	// GET customer (DB path) should trigger lazy reset:
+	// - Unused balance = 150 → rollover = min(150, cap 500) = 150
+	// - Fresh grant = 400
+	// - Total = 400 + 150 = 550
+	const after = await autumnV2.customers.get<ApiCustomer>(customerId, {
+		skip_cache: "true",
+	});
+
+	expect(after.balances[TestFeature.Messages].usage).toBe(0);
+	expect(after.balances[TestFeature.Messages].current_balance).toBe(550);
+	expect(after.balances[TestFeature.Messages].rollovers).toBeDefined();
+	expect(after.balances[TestFeature.Messages].rollovers!.length).toBe(1);
+	expect(after.balances[TestFeature.Messages].rollovers![0].balance).toBe(150);
+
+	// Verify next_reset_at advanced into the future
+	const cusEntAfter = await findCustomerEntitlement({
+		ctx,
+		customerId,
+		featureId: TestFeature.Messages,
+	});
+	expect(cusEntAfter).toBeDefined();
+	expect(cusEntAfter!.next_reset_at).toBeGreaterThan(Date.now());
+});
+
+// ─────────────────────────────────────────────────────────────────
+// Lazy reset with rollovers (cache path) — GET /customers (cached)
+//
+// Same idea but through the cache path, and with a lower rollover
+// cap to verify the cap is respected.
+// ─────────────────────────────────────────────────────────────────
+
+test.concurrent(`${chalk.yellowBright("lazy reset rollover (cache): caps rollover at max and resets via cache")}`, async () => {
+	const messagesItem = items.monthlyMessagesWithRollover({
+		includedUsage: 300,
+		rolloverConfig: {
+			max: 100,
+			length: 1,
+			duration: RolloverExpiryDurationType.Month,
+		},
+	});
+	const free = products.base({
+		id: "free",
+		items: [messagesItem],
+	});
+
+	const { customerId, autumnV2, ctx } = await initScenario({
+		customerId: "reset-rollover-cache",
+		setup: [
+			s.customer({ paymentMethod: "success", testClock: false }),
+			s.products({ list: [free] }),
+		],
+		actions: [
+			s.billing.attach({ productId: free.id }),
+			s.track({ featureId: TestFeature.Messages, value: 50, timeout: 2000 }),
+		],
+	});
+
+	// Before reset: 300 - 50 = 250 remaining
+	// Warm the cache
+	const before = await autumnV2.customers.get<ApiCustomer>(customerId);
+	expect(before.balances[TestFeature.Messages].current_balance).toBe(250);
+	expect(before.balances[TestFeature.Messages].usage).toBe(50);
+
+	// Expire cusEnt so the next read triggers a lazy reset
+	await expireCusEntForReset({
+		ctx,
+		customerId,
+		featureId: TestFeature.Messages,
+	});
+
+	// GET customer (cache path) should trigger lazy reset:
+	// - Unused balance = 250, but rollover cap = 100 → rollover = 100
+	// - Fresh grant = 300
+	// - Total = 300 + 100 = 400
+	const after = await autumnV2.customers.get<ApiCustomer>(customerId);
+
+	expect(after.balances[TestFeature.Messages].usage).toBe(0);
+	expect(after.balances[TestFeature.Messages].current_balance).toBe(400);
+	expect(after.balances[TestFeature.Messages].rollovers).toBeDefined();
+	expect(after.balances[TestFeature.Messages].rollovers!.length).toBe(1);
+	expect(after.balances[TestFeature.Messages].rollovers![0].balance).toBe(100);
+
+	// Verify next_reset_at advanced into the future
+	const cusEntAfter = await findCustomerEntitlement({
+		ctx,
+		customerId,
+		featureId: TestFeature.Messages,
+	});
+	expect(cusEntAfter).toBeDefined();
+	expect(cusEntAfter!.next_reset_at).toBeGreaterThan(Date.now());
+});