Skip to content

Commit 77ab975

Browse files
committed
fix: reset cron
1 parent 5b8096d commit 77ab975

10 files changed

Lines changed: 231 additions & 68 deletions

File tree

ai

Submodule ai updated from bce748e to ae59fe5

package.json

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -130,6 +130,7 @@
130130
"d:test": "ENV_FILE=.env infisical run --env=test --recursive -- bun scripts/dev.ts",
131131
"p": "ENV_FILE=.env.prod infisical run --env=prod --recursive -- bun scripts/dev.ts",
132132
"s": "ENV_FILE=.env.staging infisical run --env=staging --recursive -- bun scripts/dev.ts",
133+
"reset-v2": "bun server/src/cron/resetV2.ts",
133134
"l": "bash ./scripts/dev-local.sh",
134135
"setup": "node scripts/setup/setup.js",
135136
"contracts": "infisical run --env=dev --recursive -- bun scripts/s3/contracts.ts",

server/experiments/explainResetContext.ts

Lines changed: 3 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -38,7 +38,7 @@ const main = async () => {
3838
}
3939
console.log(`Scanned ${page.length} eligible customer entitlement IDs\n`);
4040

41-
// Full repo call (query + zod parse), cold then warm.
41+
// Full repo call, cold then warm.
4242
for (const run of ["cold", "warm"]) {
4343
const start = performance.now();
4444
const result = await getResetContextByIds({
@@ -47,14 +47,11 @@ const main = async () => {
4747
});
4848
const elapsed = performance.now() - start;
4949
console.log(
50-
`[hydrate][${run}] rows=${result.customerEntitlements.length}, missing=${result.missingIds.length}, invalid=${result.invalidIds.length}, wall-clock=${elapsed.toFixed(0)}ms`,
50+
`[hydrate][${run}] rows=${result.customerEntitlements.length}, missing=${result.missingIds.length}, wall-clock=${elapsed.toFixed(0)}ms`,
5151
);
52-
for (const invalid of result.invalidIds.slice(0, 3)) {
53-
console.log(` invalid ${invalid.id}: ${invalid.error.slice(0, 200)}`);
54-
}
5552
}
5653

57-
// Query-only timing (no zod parse), to separate DB cost from parse cost.
54+
// Query-only timing, to separate row-mapping cost from DB cost.
5855
{
5956
const start = performance.now();
6057
const rows = await db.execute(

server/src/cron/resetV2.ts

Lines changed: 141 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,141 @@
1+
import { spawn } from "node:child_process";
2+
import { fileURLToPath } from "node:url";
3+
4+
type ResetEnvironment = "staging" | "prod";
5+
6+
const WRAPPED_ENV_VAR = "AUTUMN_RESET_V2_WRAPPED";
7+
const ENV_FILES: Record<ResetEnvironment, string> = {
8+
staging: ".env.staging",
9+
prod: ".env.prod",
10+
};
11+
12+
const parseEnvironment = (): ResetEnvironment => {
13+
const environment = process.argv[2];
14+
if (environment === "staging" || environment === "prod") return environment;
15+
16+
console.error("Usage: bun reset-v2 <staging|prod>");
17+
process.exit(2);
18+
};
19+
20+
const runWithInfisical = async ({
21+
environment,
22+
}: {
23+
environment: ResetEnvironment;
24+
}) => {
25+
const child = spawn(
26+
"infisical",
27+
[
28+
"run",
29+
`--env=${environment}`,
30+
"--recursive",
31+
"--",
32+
"bun",
33+
fileURLToPath(import.meta.url),
34+
environment,
35+
],
36+
{
37+
stdio: "inherit",
38+
env: {
39+
...process.env,
40+
[WRAPPED_ENV_VAR]: "1",
41+
ENV_FILE: ENV_FILES[environment],
42+
NODE_ENV: "development",
43+
},
44+
},
45+
);
46+
47+
const exitCode = await new Promise<number>((resolve) => {
48+
child.on("close", (code) => resolve(code ?? 1));
49+
child.on("error", (error) => {
50+
console.error(`Failed to start Infisical: ${error.message}`);
51+
resolve(1);
52+
});
53+
});
54+
process.exit(exitCode);
55+
};
56+
57+
const runResetV2 = async ({
58+
environment,
59+
}: {
60+
environment: ResetEnvironment;
61+
}) => {
62+
await import("../sentry.js");
63+
await import("../internal/misc/resetJobV2/resetJobV2Store.js");
64+
65+
const { initDrizzle } = await import("../db/initDrizzle.js");
66+
const { startPgPoolMonitor, stopPgPoolMonitor } = await import(
67+
"../db/pgPoolMonitor.js"
68+
);
69+
const { logger } = await import("../external/logtail/logtailUtils.js");
70+
const { runResetLoopV2 } = await import(
71+
"../internal/balances/batchReset/runResetLoopV2.js"
72+
);
73+
const { getResetJobV2Config, getResetJobV2ConfigStatus } = await import(
74+
"../internal/misc/resetJobV2/resetJobV2Store.js"
75+
);
76+
const { startAllEdgeConfigPolling, stopAllEdgeConfigPolling } = await import(
77+
"../internal/misc/edgeConfig/edgeConfigRegistry.js"
78+
);
79+
80+
await startAllEdgeConfigPolling({ logger });
81+
82+
const { db, client } = initDrizzle({
83+
name: "reset-cron-v2",
84+
maxConnections: 10,
85+
});
86+
startPgPoolMonitor();
87+
88+
const controller = new AbortController();
89+
const loopPromise = runResetLoopV2({
90+
ctx: { db, logger },
91+
signal: controller.signal,
92+
});
93+
let shuttingDown = false;
94+
95+
// Everything under `data`: the express dataset is at its Axiom column
96+
// limit, so new top-level (flattened) field names get the whole ingest
97+
// batch rejected.
98+
logger.info("[reset-cus-ents-v2] dedicated scanner started", {
99+
jobName: "reset-cus-ents-v2",
100+
data: {
101+
environment,
102+
config: getResetJobV2Config(),
103+
configStatus: getResetJobV2ConfigStatus(),
104+
},
105+
});
106+
107+
// The scanner must never die because telemetry hiccuped (e.g. an Axiom
108+
// ingest rejection inside the pino transport becomes an unhandled
109+
// rejection, which kills the process by default).
110+
process.on("unhandledRejection", (reason) => {
111+
console.error("[reset-cus-ents-v2] unhandled rejection (ignored):", reason);
112+
});
113+
114+
const shutdown = async (signal: string) => {
115+
if (shuttingDown) return;
116+
shuttingDown = true;
117+
logger.info(`[reset-cus-ents-v2] received ${signal}, shutting down`);
118+
controller.abort();
119+
stopPgPoolMonitor();
120+
stopAllEdgeConfigPolling();
121+
await loopPromise;
122+
await client.end();
123+
};
124+
125+
process.once("SIGINT", () => void shutdown("SIGINT"));
126+
process.once("SIGTERM", () => void shutdown("SIGTERM"));
127+
128+
await loopPromise;
129+
};
130+
131+
const main = async () => {
132+
const environment = parseEnvironment();
133+
if (process.env[WRAPPED_ENV_VAR] !== "1") {
134+
await runWithInfisical({ environment });
135+
return;
136+
}
137+
138+
await runResetV2({ environment });
139+
};
140+
141+
await main();

server/src/internal/balances/batchReset/logs/logBatchCustomerEntitlementsV2.ts

Lines changed: 0 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -31,7 +31,6 @@ export const logBatchCustomerEntitlementsV2 = ({
3131
data: {
3232
requested: payload.customerEntitlementIds.length,
3333
missing: batchResetContext.missingIds.length,
34-
invalid: batchResetContext.invalidCount,
3534
orgs: batchResetContext.groups.length,
3635
resetOrgs: classifiedBatchResetContext.resetGroups.length,
3736
resettable: resettableCount,

server/src/internal/balances/batchReset/setup/setupBatchResetContext.ts

Lines changed: 1 addition & 10 deletions
Original file line numberDiff line numberDiff line change
@@ -66,20 +66,12 @@ export const setupBatchResetContext = async ({
6666
logger: Logger;
6767
payload: BatchResetCustomerEntitlementsV2Payload;
6868
}): Promise<BatchResetContext> => {
69-
const { customerEntitlements, invalidIds, missingIds } =
69+
const { customerEntitlements, missingIds } =
7070
await customerEntitlementsRepo.getResetContextByIds({
7171
db,
7272
customerEntitlementIds: payload.customerEntitlementIds,
7373
});
7474

75-
// Rows that failed schema validation stay unreset (and will be re-picked
76-
// by the scan) — surface them loudly so the data can be fixed.
77-
for (const invalid of invalidIds) {
78-
logger.error(
79-
`[batchReset] cusEnt ${invalid.id} failed validation: ${invalid.error}`,
80-
);
81-
}
82-
8375
const orgContexts = await fetchUniqueOrgContexts({
8476
db,
8577
logger,
@@ -105,6 +97,5 @@ export const setupBatchResetContext = async ({
10597
return {
10698
groups: [...groupsByOrgEnv.values()],
10799
missingIds,
108-
invalidCount: invalidIds.length,
109100
};
110101
};

server/src/internal/balances/batchReset/types.ts

Lines changed: 0 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -17,7 +17,6 @@ export type BatchResetContext = {
1717
* worker context (org + features loaded once per group). */
1818
groups: BatchResetGroup[];
1919
missingIds: string[];
20-
invalidCount: number;
2120
};
2221

2322
/** Every non-resettable candidate receives exactly one verdict. */

server/src/internal/customers/cusProducts/cusEnts/repos/getResetContextByIds.ts

Lines changed: 22 additions & 45 deletions
Original file line numberDiff line numberDiff line change
@@ -1,28 +1,24 @@
1-
import {
2-
CustomerSchema,
3-
FullCusEntWithFullCusProductSchema,
4-
RELEVANT_STATUSES,
5-
} from "@autumn/shared";
1+
import type { Customer, FullCusEntWithFullCusProduct } from "@autumn/shared";
2+
import { RELEVANT_STATUSES } from "@autumn/shared";
63
import { sql } from "drizzle-orm";
7-
import { z } from "zod/v4";
84
import type { DrizzleCli } from "@/db/initDrizzle.js";
95
import { resetCronQueryTag } from "@/internal/balances/batchReset/resetCronQueryTag.js";
106

117
/**
128
* A customer entitlement hydrated with everything batch reset needs:
139
* the FullCusEntWithFullCusProduct shape processReset consumes, plus the
1410
* owning customer for org/env checks, cache invalidation and redis routing.
11+
*
12+
* Hydration is intentionally UNVALIDATED (same trust in the DB shape as the
13+
* V1 cron and lazy-reset paths): legacy quirks like an entity balance keyed
14+
* "null" with a null id must still reset — a validation reject would leave
15+
* such rows permanently unreset at the head of the scan.
1516
*/
16-
export const ResetContextCustomerEntitlementSchema =
17-
FullCusEntWithFullCusProductSchema.extend({
18-
customer: CustomerSchema,
19-
// Reset-scan denormalization flag; not yet part of the shared API model.
20-
expired: z.boolean().nullable().optional(),
21-
});
22-
23-
export type ResetContextCustomerEntitlement = z.infer<
24-
typeof ResetContextCustomerEntitlementSchema
25-
>;
17+
export type ResetContextCustomerEntitlement = FullCusEntWithFullCusProduct & {
18+
customer: Customer;
19+
// Reset-scan denormalization flag; not yet part of the shared API model.
20+
expired?: boolean | null;
21+
};
2622

2723
/**
2824
* Single-statement hydration query for an explicit set of customer entitlement
@@ -98,8 +94,9 @@ export const buildResetContextByIdsQuery = ({
9894

9995
/**
10096
* Hydrates the requested customer entitlement IDs. IDs can legitimately be
101-
* missing if their rows were deleted between scan and execution. Every
102-
* returned row is zod-parsed; validation failures are surfaced to the caller.
97+
* missing if their rows were deleted between scan and execution. Rows are
98+
* returned as-is — see ResetContextCustomerEntitlement for why hydration is
99+
* unvalidated.
103100
*/
104101
export const getResetContextByIds = async ({
105102
db,
@@ -109,42 +106,22 @@ export const getResetContextByIds = async ({
109106
customerEntitlementIds: string[];
110107
}): Promise<{
111108
customerEntitlements: ResetContextCustomerEntitlement[];
112-
invalidIds: { id: string; error: string }[];
113109
missingIds: string[];
114110
}> => {
115111
const uniqueIds = [...new Set(customerEntitlementIds)];
116112
if (uniqueIds.length === 0) {
117-
return {
118-
customerEntitlements: [],
119-
invalidIds: [],
120-
missingIds: [],
121-
};
113+
return { customerEntitlements: [], missingIds: [] };
122114
}
123115

124-
const rows = await db.execute<{ id: string; reset_context: unknown }>(
125-
buildResetContextByIdsQuery({ customerEntitlementIds: uniqueIds }),
126-
);
116+
const rows = await db.execute<{
117+
id: string;
118+
reset_context: ResetContextCustomerEntitlement;
119+
}>(buildResetContextByIdsQuery({ customerEntitlementIds: uniqueIds }));
127120

128-
const resultCustomerEntitlements: ResetContextCustomerEntitlement[] = [];
129-
const invalidIds: { id: string; error: string }[] = [];
130-
const returnedIds = new Set<string>();
131-
132-
for (const row of rows) {
133-
returnedIds.add(row.id);
134-
const parsed = ResetContextCustomerEntitlementSchema.safeParse(
135-
row.reset_context,
136-
);
137-
if (!parsed.success) {
138-
invalidIds.push({ id: row.id, error: parsed.error.message });
139-
continue;
140-
}
141-
142-
resultCustomerEntitlements.push(parsed.data);
143-
}
121+
const returnedIds = new Set(rows.map((row) => row.id));
144122

145123
return {
146-
customerEntitlements: resultCustomerEntitlements,
147-
invalidIds,
124+
customerEntitlements: rows.map((row) => row.reset_context),
148125
missingIds: uniqueIds.filter((id) => !returnedIds.has(id)),
149126
};
150127
};

server/src/queue/initWorkers.ts

Lines changed: 20 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -103,6 +103,7 @@ export const startPollingLoop = async ({
103103
getSqsClientFn,
104104
recreateSqsClientFn,
105105
shouldPoll = () => true,
106+
visibilityTimeoutSeconds = 30,
106107
}: {
107108
db: DrizzleCli;
108109
queueId: string;
@@ -111,6 +112,10 @@ export const startPollingLoop = async ({
111112
getSqsClientFn: () => SQSClient;
112113
recreateSqsClientFn: () => SQSClient;
113114
shouldPoll?: () => boolean;
115+
/** Raise for queues whose jobs legitimately run long (e.g. batch resets) —
116+
* a message redelivered mid-processing means two workers mutating the same
117+
* rows concurrently. */
118+
visibilityTimeoutSeconds?: number;
114119
}) => {
115120
// Per-loop state
116121
let messagesProcessed = 0;
@@ -214,7 +219,7 @@ export const startPollingLoop = async ({
214219
QueueUrl: queueUrl,
215220
MaxNumberOfMessages: maxNumberOfMessages,
216221
WaitTimeSeconds: 20,
217-
VisibilityTimeout: 30,
222+
VisibilityTimeout: visibilityTimeoutSeconds,
218223
MessageSystemAttributeNames: ["SentTimestamp", "ApproximateReceiveCount"],
219224
...(isFifo && { ReceiveRequestAttemptId: generateId("receive") }),
220225
});
@@ -515,7 +520,12 @@ export const initWorkers = async ({
515520
);
516521
const pollingLoops = [];
517522

518-
for (const { queueId, queueUrl, defaultEnabled } of [
523+
for (const {
524+
queueId,
525+
queueUrl,
526+
defaultEnabled,
527+
visibilityTimeoutSeconds,
528+
} of [
519529
{
520530
queueId: JOB_QUEUE_IDS.primary,
521531
queueUrl: QUEUE_URL,
@@ -539,7 +549,13 @@ export const initWorkers = async ({
539549
{
540550
queueId: JOB_QUEUE_IDS.batchReset,
541551
queueUrl: process.env.BATCH_RESET_SQS_QUEUE_URL,
542-
defaultEnabled: false,
552+
defaultEnabled: true,
553+
// Reset batches can legitimately run long (Stripe anchor checks on
554+
// month-edge dates); a short window would redeliver mid-processing and
555+
// have two workers resetting the same rows concurrently. SQS maximum
556+
// (12h) — failed resets are re-found by the next scan, so redelivery
557+
// latency doesn't matter.
558+
visibilityTimeoutSeconds: 43_200,
543559
},
544560
]) {
545561
if (!queueUrl) continue;
@@ -555,6 +571,7 @@ export const initWorkers = async ({
555571
shouldPoll: () =>
556572
isJobQueueEnabled({ queue: queueId, defaultEnabled }) &&
557573
isActiveSlot({ serviceName: "workers" }),
574+
visibilityTimeoutSeconds,
558575
}),
559576
);
560577
}

0 commit comments

Comments
 (0)