fix(data): oom crash in stats sync

This commit is contained in:
Adam 2026-07-09 19:53:56 -05:00
commit 85ef35dd64
No known key found for this signature in database
GPG key ID: 9CB48779AF150E75
2 changed files with 26 additions and 19 deletions

View file

@ -185,7 +185,9 @@ export const statSync = new sst.aws.Service("StatsSyncService", {
cluster: lakeCluster, cluster: lakeCluster,
architecture: "arm64", architecture: "arm64",
cpu: "0.25 vCPU", cpu: "0.25 vCPU",
memory: "0.5 GB", // 0.5 GB caused an OOM crash loop: every restart immediately re-ran the 4 Athena
// stats queries (~$5/pass) every ~5 minutes instead of hourly.
memory: "2 GB",
image: { image: {
context: ".", context: ".",
dockerfile: "packages/stats/server/Dockerfile", dockerfile: "packages/stats/server/Dockerfile",

View file

@ -116,28 +116,33 @@ const poll: (
const results: ( const results: (
client: AwsAthenaClient, client: AwsAthenaClient,
queryExecutionId: string, queryExecutionId: string,
nextToken?: string,
) => Effect.Effect<AthenaData[], AthenaQueryError> = Effect.fn("Athena.results")(function* ( ) => Effect.Effect<AthenaData[], AthenaQueryError> = Effect.fn("Athena.results")(function* (
client: AwsAthenaClient, client: AwsAthenaClient,
queryExecutionId: string, queryExecutionId: string,
nextToken?: string,
) { ) {
const result = yield* Effect.tryPromise({ // Accumulate pages iteratively; recursive spreads copied every previously
try: () => // fetched row per page and blew up memory on large result sets.
client.send( const rows: AthenaData[] = []
new GetQueryResultsCommand({ let nextToken: string | undefined
QueryExecutionId: queryExecutionId, while (true) {
NextToken: nextToken, const result = yield* Effect.tryPromise({
MaxResults: ATHENA_PAGE_SIZE, try: () =>
}), client.send(
), new GetQueryResultsCommand({
catch: (cause) => new AthenaQueryError({ message: "Failed to read Athena stats results", queryExecutionId, cause }), QueryExecutionId: queryExecutionId,
}) NextToken: nextToken,
const columns = result.ResultSet?.ResultSetMetadata?.ColumnInfo?.map((item) => item.Name ?? "") ?? [] MaxResults: ATHENA_PAGE_SIZE,
const rows = (result.ResultSet?.Rows ?? []).slice(nextToken ? 0 : 1).map((row) => rowData(columns, row)) }),
),
if (!result.NextToken) return rows catch: (cause) =>
return [...rows, ...(yield* results(client, queryExecutionId, result.NextToken))] new AthenaQueryError({ message: "Failed to read Athena stats results", queryExecutionId, cause }),
})
const columns = result.ResultSet?.ResultSetMetadata?.ColumnInfo?.map((item) => item.Name ?? "") ?? []
// The first page starts with the header row.
for (const row of (result.ResultSet?.Rows ?? []).slice(nextToken ? 0 : 1)) rows.push(rowData(columns, row))
if (!result.NextToken) return rows
nextToken = result.NextToken
}
}) })
function rowData(columns: string[], row: Row): AthenaData { function rowData(columns: string[], row: Row): AthenaData {