From 75ce125b4268db0ea07e18cba89446cfc88bf62a Mon Sep 17 00:00:00 2001 From: needs <624097+needs@users.noreply.github.com> Date: Tue, 8 Sep 2026 13:50:28 +0200 Subject: [PATCH 1/4] Fix the failing DDNet imports and rollup deadlocks shown on /status - Parse the DDNet stats dump with MySQL-style backslash escapes; every backfill tick died on a player name containing an escaped quote. - Tolerate out-of-order days in the DDNet online CSVs, which finalized the same day twice and hit the (day, kind, key) unique constraint. - Close the dump download and the online CSV reader when an import aborts. - Serialize rollup writers with an advisory lock: the hourly day rollup and the midnight backfill deadlocked on Player and PlayerPartner rows. - Give the map count status check an hourly window instead of 10 minutes. - Prefix the game type count dedup id so an empty game type name schedules. Co-Authored-By: Claude Fable 5.1 --- apps/frontend/app/status/page.tsx | 2 +- apps/worker/src/ddnet/stream.test.ts | 24 ++++ apps/worker/src/ddnet/stream.ts | 123 ++++++++++-------- apps/worker/src/rollup/writeDayRollup.ts | 8 +- .../src/workers/ddnetOnlineImport.test.ts | 40 ++++++ apps/worker/src/workers/ddnetOnlineImport.ts | 76 +++++++---- libs/prisma/prisma/sql/lockRollupWrite.sql | 1 + .../src/lib/bullmq/queueGameTypeCount.ts | 4 +- 8 files changed, 189 insertions(+), 89 deletions(-) create mode 100644 apps/worker/src/ddnet/stream.test.ts create mode 100644 apps/worker/src/workers/ddnetOnlineImport.test.ts create mode 100644 libs/prisma/prisma/sql/lockRollupWrite.sql diff --git a/apps/frontend/app/status/page.tsx b/apps/frontend/app/status/page.tsx index b3f2165..66d1e57 100644 --- a/apps/frontend/app/status/page.tsx +++ b/apps/frontend/app/status/page.tsx @@ -145,7 +145,7 @@ export default async function Index() { { title: 'Map count', date: lastMapCountDate, - staleAfterMinutes: 10, + staleAfterMinutes: 75, }, { title: 'Archiving snapshots', diff --git a/apps/worker/src/ddnet/stream.test.ts b/apps/worker/src/ddnet/stream.test.ts new file mode 100644 index 0000000..b865d69 --- /dev/null +++ b/apps/worker/src/ddnet/stream.test.ts @@ -0,0 +1,24 @@ +import { Readable } from "stream"; +import { parseCsvEntry } from "./stream"; + +describe('parseCsvEntry', () => { + it('decodes the MySQL-style backslash escapes used by the dump', async () => { + const csv = [ + '"Map","Name","Time"', + '"For Idiots 1","Tobias\\"",514.2', + '"Bootcamp #2","I\\\\I4I\\\\/I3",808.46', + '"Lowcore","-/Noob\\\\-Fra",167.14', + ].join('\n') + '\n'; + + const records: string[][] = []; + await parseCsvEntry(Readable.from([csv]), (record) => { + records.push(record); + }); + + expect(records).toEqual([ + ['For Idiots 1', 'Tobias"', '514.2'], + ['Bootcamp #2', 'I\\I4I\\/I3', '808.46'], + ['Lowcore', '-/Noob\\-Fra', '167.14'], + ]); + }); +}); diff --git a/apps/worker/src/ddnet/stream.ts b/apps/worker/src/ddnet/stream.ts index e9f77b9..1adf08e 100644 --- a/apps/worker/src/ddnet/stream.ts +++ b/apps/worker/src/ddnet/stream.ts @@ -72,13 +72,15 @@ function parseSplits(record: string[]): number[] { return hasData ? splits : []; } -async function parseCsvEntry( +export async function parseCsvEntry( entry: NodeJS.ReadableStream, onRecord: (record: string[], header: string[]) => void | Promise ) { + // The dump is a MySQL export: quotes and backslashes are backslash-escaped. const parser = entry.pipe(parse({ relax_column_count: true, bom: true, + escape: '\\', })); let header: string[] | null = null; @@ -109,68 +111,75 @@ export async function streamStatsDump(handlers: DumpHandlers) { throw new Error(`GET ${DDNET_STATS_URL} failed: ${response.status}`); } - const zip = Readable.fromWeb(response.body as never).pipe(unzipper.Parse({ forceStream: true })); - - for await (const entry of zip as AsyncIterable) { - const fileName = entry.path.split('/').pop() ?? ''; - - if (fileName === 'maps.csv' && handlers.onMap !== undefined) { - await parseCsvEntry(entry, async (record) => { - await handlers.onMap!({ - name: record[0], - category: record[1], - points: Number(record[2]), - stars: Number(record[3]), - mapper: record[4], - releasedAt: parseDumpTimestamp(record[5]), + const source = Readable.fromWeb(response.body as never); + const zip = source.pipe(unzipper.Parse({ forceStream: true })); + + try { + for await (const entry of zip as AsyncIterable) { + const fileName = entry.path.split('/').pop() ?? ''; + + if (fileName === 'maps.csv' && handlers.onMap !== undefined) { + await parseCsvEntry(entry, async (record) => { + await handlers.onMap!({ + name: record[0], + category: record[1], + points: Number(record[2]), + stars: Number(record[3]), + mapper: record[4], + releasedAt: parseDumpTimestamp(record[5]), + }); }); - }); - } else if (fileName === 'mapinfo.csv' && handlers.onMapInfo !== undefined) { - await parseCsvEntry(entry, async (record, header) => { - const tiles: string[] = []; - for (let index = 3; index < header.length; index++) { - if (Number(record[index]) > 0) { - tiles.push(header[index]); + } else if (fileName === 'mapinfo.csv' && handlers.onMapInfo !== undefined) { + await parseCsvEntry(entry, async (record, header) => { + const tiles: string[] = []; + for (let index = 3; index < header.length; index++) { + if (Number(record[index]) > 0) { + tiles.push(header[index]); + } } - } - await handlers.onMapInfo!({ - name: record[0], - width: Number(record[1]), - height: Number(record[2]), - tiles, + await handlers.onMapInfo!({ + name: record[0], + width: Number(record[1]), + height: Number(record[2]), + tiles, + }); }); - }); - } else if (fileName === 'race.csv' && handlers.onRace !== undefined) { - await parseCsvEntry(entry, async (record) => { - const timestamp = parseDumpTimestamp(record[3]); - if (timestamp === null) { - return; - } - await handlers.onRace!({ - mapName: record[0], - playerName: record[1], - time: Number(record[2]), - timestamp, - splits: parseSplits(record), + } else if (fileName === 'race.csv' && handlers.onRace !== undefined) { + await parseCsvEntry(entry, async (record) => { + const timestamp = parseDumpTimestamp(record[3]); + if (timestamp === null) { + return; + } + await handlers.onRace!({ + mapName: record[0], + playerName: record[1], + time: Number(record[2]), + timestamp, + splits: parseSplits(record), + }); }); - }); - } else if (fileName === 'teamrace.csv' && handlers.onTeamRace !== undefined) { - await parseCsvEntry(entry, async (record) => { - const timestamp = parseDumpTimestamp(record[4]); - if (timestamp === null) { - return; - } - await handlers.onTeamRace!({ - mapName: record[0], - playerName: record[1], - time: Number(record[2]), - teamId: record[3].toLowerCase(), - timestamp, + } else if (fileName === 'teamrace.csv' && handlers.onTeamRace !== undefined) { + await parseCsvEntry(entry, async (record) => { + const timestamp = parseDumpTimestamp(record[4]); + if (timestamp === null) { + return; + } + await handlers.onTeamRace!({ + mapName: record[0], + playerName: record[1], + time: Number(record[2]), + teamId: record[3].toLowerCase(), + timestamp, + }); }); - }); - } else { - entry.autodrain(); + } else { + entry.autodrain(); + } } + } finally { + // Bailing out mid-archive must not leave the download open. + zip.destroy(); + source.destroy(); } } diff --git a/apps/worker/src/rollup/writeDayRollup.ts b/apps/worker/src/rollup/writeDayRollup.ts index ec5f81c..ee9ffa1 100644 --- a/apps/worker/src/rollup/writeDayRollup.ts +++ b/apps/worker/src/rollup/writeDayRollup.ts @@ -1,7 +1,7 @@ import { chunk } from "lodash"; import { minutesToMilliseconds } from "date-fns"; import { formatUtcDay, isStubName } from "@teerank/teerank"; -import { incrementPlayerPollCounts, upsertPlayerPartners } from "@prisma/client/sql"; +import { incrementPlayerPollCounts, lockRollupWrite, upsertPlayerPartners } from "@prisma/client/sql"; import { prisma } from "../prisma"; import { DayRollup } from "./aggregateDay"; import { ensureRollupPartitions } from "./partitions"; @@ -166,6 +166,10 @@ export async function writeDayRollup(day: Date, rollup: DayRollup) { await prisma.$transaction( async (tx) => { + // Day and backfill rollups touch the same Player and PlayerPartner + // rows; serializing writers avoids deadlocking on them. + await tx.$queryRawTyped(lockRollupWrite()); + const alreadyRolledUp = (await tx.globalDay.findUnique({ where: { day }, select: { day: true } })) !== null; @@ -220,7 +224,7 @@ export async function writeDayRollup(day: Date, rollup: DayRollup) { } } }, - { timeout: minutesToMilliseconds(5), maxWait: minutesToMilliseconds(1) } + { timeout: minutesToMilliseconds(15), maxWait: minutesToMilliseconds(1) } ); const dayLabel = formatUtcDay(day); diff --git a/apps/worker/src/workers/ddnetOnlineImport.test.ts b/apps/worker/src/workers/ddnetOnlineImport.test.ts new file mode 100644 index 0000000..78945c5 --- /dev/null +++ b/apps/worker/src/workers/ddnetOnlineImport.test.ts @@ -0,0 +1,40 @@ +import { prismaMock } from "../../test/mockPrisma"; +import { ddnetOnlineImport } from "./ddnetOnlineImport"; + +function csvResponse(text: string) { + return new Response(text, { status: 200 }); +} + +describe('ddnetOnlineImport', () => { + beforeEach(() => { + prismaMock.ddnetState.findUnique.mockResolvedValue(null); + prismaMock.$transaction.mockImplementation(((callback: (tx: typeof prismaMock) => Promise) => + callback(prismaMock)) as never); + }); + + it('merges a sample logged after the next day began into its own day', async () => { + const bycountry = [ + '2021-10-21 23:56,GER:10,USA:2', + '2021-10-22 00:00,GER:4,USA:0', + '2021-10-21 23:58,GER:20,USA:2', + '2021-10-22 00:02,GER:6,USA:0', + '2021-10-23 00:00,GER:1,USA:1', + ].join('\n') + '\n'; + + global.fetch = jest.fn(async (url: string) => + csvResponse(url.endsWith('/bycountry') ? bycountry : '') + ) as never; + + await ddnetOnlineImport(); + + const rows = prismaMock.ddnetOnlineDay.createMany.mock.calls + .flatMap(([args]) => args!.data as { day: Date; key: string; avgPlayers: number; maxPlayers: number }[]); + + const keys = rows.map((row) => `${row.day.toISOString().slice(0, 10)}/${row.key}`); + expect(new Set(keys).size).toBe(keys.length); + + const ger21 = rows.find((row) => row.key === 'GER' && row.day.toISOString().startsWith('2021-10-21')); + expect(ger21).toMatchObject({ avgPlayers: 15, maxPlayers: 20 }); + expect(rows.some((row) => row.day.toISOString().startsWith('2021-10-23'))).toBe(false); + }); +}); diff --git a/apps/worker/src/workers/ddnetOnlineImport.ts b/apps/worker/src/workers/ddnetOnlineImport.ts index 0ce98b4..7f299b7 100644 --- a/apps/worker/src/workers/ddnetOnlineImport.ts +++ b/apps/worker/src/workers/ddnetOnlineImport.ts @@ -94,7 +94,8 @@ async function importSource(source: typeof SOURCES[number]) { const decoder = new TextDecoder(); let pendingRows: ReturnType = []; - let current: DayAggregate | null = null; + const openDays = new Map(); + let currentDay: string | null = null; let currentDayStart = offset; let lineStart = offset; let buffered = ''; @@ -106,11 +107,16 @@ async function importSource(source: typeof SOURCES[number]) { return; } - if (current === null || current.day !== day) { - if (current !== null) { - pendingRows.push(...finalizeDay(current, source.kind, source.withTotal)); + let current = openDays.get(day); + if (current === undefined) { + // A sample logged after the next day began belongs to a day that may + // already be flushed; a single stray sample is not worth reopening it. + if (currentDay !== null && day < currentDay) { + return; } current = newDayAggregate(day); + openDays.set(day, current); + currentDay = day; currentDayStart = lineStart; } @@ -143,34 +149,50 @@ async function importSource(source: typeof SOURCES[number]) { current.totalSum += total; }; - for (;;) { - const { done, value } = await reader.read(); - const text = done ? decoder.decode() : decoder.decode(value, { stream: true }); - buffered += text; - - let newlineIndex: number; - while ((newlineIndex = buffered.indexOf('\n')) !== -1) { - const line = buffered.slice(0, newlineIndex); - handleLine(line.trimEnd()); - lineStart += Buffer.byteLength(line, 'utf8') + 1; - buffered = buffered.slice(newlineIndex + 1); + // Every day but the current one is closed: later samples for it are dropped. + const finalizeClosedDays = () => { + for (const [day, aggregate] of openDays) { + if (day !== currentDay) { + pendingRows.push(...finalizeDay(aggregate, source.kind, source.withTotal)); + openDays.delete(day); + } } + }; - if (pendingRows.length >= FLUSH_DAY_COUNT * 10) { - // The offset stays at the start of the still-open day so the next run - // re-reads and re-finalizes it. - await flushDays(pendingRows, source.kind, source.stateKey, currentDayStart); - pendingRows = []; - } + try { + for (;;) { + const { done, value } = await reader.read(); + const text = done ? decoder.decode() : decoder.decode(value, { stream: true }); + buffered += text; + + let newlineIndex: number; + while ((newlineIndex = buffered.indexOf('\n')) !== -1) { + const line = buffered.slice(0, newlineIndex); + handleLine(line.trimEnd()); + lineStart += Buffer.byteLength(line, 'utf8') + 1; + buffered = buffered.slice(newlineIndex + 1); + } - if (done) { - break; + if (openDays.size > FLUSH_DAY_COUNT) { + // The offset stays at the start of the still-open day so the next run + // re-reads and re-finalizes it. + finalizeClosedDays(); + await flushDays(pendingRows, source.kind, source.stateKey, currentDayStart); + pendingRows = []; + } + + if (done) { + break; + } } - } - // The last (incomplete) day is intentionally not finalized. - if (pendingRows.length > 0) { - await flushDays(pendingRows, source.kind, source.stateKey, currentDayStart); + // The last (incomplete) day is intentionally not finalized. + finalizeClosedDays(); + if (pendingRows.length > 0) { + await flushDays(pendingRows, source.kind, source.stateKey, currentDayStart); + } + } finally { + await reader.cancel().catch(() => undefined); } } diff --git a/libs/prisma/prisma/sql/lockRollupWrite.sql b/libs/prisma/prisma/sql/lockRollupWrite.sql new file mode 100644 index 0000000..1d7fd17 --- /dev/null +++ b/libs/prisma/prisma/sql/lockRollupWrite.sql @@ -0,0 +1 @@ +SELECT pg_advisory_xact_lock(hashtext('rollup-write')) IS NULL AS locked; diff --git a/libs/teerank/src/lib/bullmq/queueGameTypeCount.ts b/libs/teerank/src/lib/bullmq/queueGameTypeCount.ts index a42466c..b560d4d 100644 --- a/libs/teerank/src/lib/bullmq/queueGameTypeCount.ts +++ b/libs/teerank/src/lib/bullmq/queueGameTypeCount.ts @@ -22,9 +22,9 @@ export type GameTypeCountJobData = z.infer; export async function scheduleGameTypeCount(data: GameTypeCountJobData) { const queue = getQueueGameTypeCount(); - await queue.add(data.gameTypeName, data, { + await queue.add('game-type-count', data, { deduplication: { - id: data.gameTypeName, + id: `game-type-count:${data.gameTypeName}`, } }); } From 8be553ae384e8058ec53290684dc4aa6e2e80b6e Mon Sep 17 00:00:00 2001 From: needs <624097+needs@users.noreply.github.com> Date: Wed, 9 Sep 2026 00:55:34 +0200 Subject: [PATCH 2/4] Read the day rollup through a dedicated Prisma pool Since late August every hourly rollup-day job ran out its 10 minute budget with the database and the worker both idle: each 2000-snapshot batch waited seconds for a pooled connection behind the hundred concurrent poll transactions. Give the rollup reads their own two-connection pool and log progress so the next stall is measurable. Co-Authored-By: Claude Fable 5.1 --- apps/worker/src/prisma.ts | 16 ++++++++++++++++ apps/worker/src/snapshots.ts | 6 ++++-- apps/worker/src/workers/rollupDay.ts | 17 ++++++++++++++--- apps/worker/test/mockPrisma.ts | 8 ++++---- 4 files changed, 38 insertions(+), 9 deletions(-) diff --git a/apps/worker/src/prisma.ts b/apps/worker/src/prisma.ts index 773ed0e..86c416b 100644 --- a/apps/worker/src/prisma.ts +++ b/apps/worker/src/prisma.ts @@ -5,3 +5,19 @@ export const prismaDatabaseUrl = process.env.DATABASE_URL; export const prisma = new PrismaClient({ datasourceUrl: prismaDatabaseUrl, }); + +function withConnectionLimit(url: string | undefined, limit: number) { + if (url === undefined) { + return undefined; + } + const parsed = new URL(url); + parsed.searchParams.set('connection_limit', String(limit)); + return parsed.toString(); +} + +// The day rollup streams a whole day in small batches; on the shared pool each +// batch queues behind the hundred concurrent poll transactions and the job +// runs out its time budget while both the database and the worker sit idle. +export const rollupPrisma = new PrismaClient({ + datasourceUrl: withConnectionLimit(prismaDatabaseUrl, 2), +}); diff --git a/apps/worker/src/snapshots.ts b/apps/worker/src/snapshots.ts index d3d1f5c..1fb9735 100644 --- a/apps/worker/src/snapshots.ts +++ b/apps/worker/src/snapshots.ts @@ -1,5 +1,5 @@ -import { Prisma } from "@prisma/client"; -import { prisma } from "./prisma"; +import { Prisma, PrismaClient } from "@prisma/client"; +import { prisma as defaultPrisma } from "./prisma"; const snapshotSelect = { id: true, @@ -35,10 +35,12 @@ export async function* iterateSnapshots({ from, to, batchSize, + prisma = defaultPrisma, }: { from: Date; to: Date; batchSize: number; + prisma?: Pick; }): AsyncGenerator { let cursor = 0; diff --git a/apps/worker/src/workers/rollupDay.ts b/apps/worker/src/workers/rollupDay.ts index c1799c6..562149c 100644 --- a/apps/worker/src/workers/rollupDay.ts +++ b/apps/worker/src/workers/rollupDay.ts @@ -5,7 +5,7 @@ import { parseUtcDay, processRollupDayJobs, } from "@teerank/teerank"; -import { prisma } from "../prisma"; +import { rollupPrisma } from "../prisma"; import { iterateSnapshots } from "../snapshots"; import { DayAggregator } from "../rollup/aggregateDay"; import { writeDayRollup } from "../rollup/writeDayRollup"; @@ -14,7 +14,7 @@ const ROLLUP_BATCH_SIZE = getEnvInt('ROLLUP_BATCH_SIZE', 2000); const ROLLUP_TIME_BUDGET_MS = getEnvInt('ROLLUP_TIME_BUDGET_MS', 10 * 60 * 1000); export async function isDayRolledUp(day: Date) { - const existing = await prisma.globalDay.findUnique({ + const existing = await rollupPrisma.globalDay.findUnique({ where: { day }, select: { day: true }, }); @@ -38,14 +38,25 @@ export async function rollupDay(data: RollupDayJobData) { } const aggregator = new DayAggregator(); + let snapshotCount = 0; for await (const snapshot of iterateSnapshots({ from: day, to: dayEnd, batchSize: ROLLUP_BATCH_SIZE, + prisma: rollupPrisma, })) { if (Date.now() - startedAt > ROLLUP_TIME_BUDGET_MS) { - throw new Error(`Rollup for ${data.day} exceeded time budget, nothing written`); + throw new Error( + `Rollup for ${data.day} exceeded time budget after ${snapshotCount} snapshots, nothing written` + ); + } + + snapshotCount += 1; + if (snapshotCount % 50_000 === 0) { + console.log( + `Rollup for ${data.day}: ${snapshotCount} snapshots read in ${Math.round((Date.now() - startedAt) / 1000)}s` + ); } aggregator.addSnapshot({ diff --git a/apps/worker/test/mockPrisma.ts b/apps/worker/test/mockPrisma.ts index 6ea7607..b2b221f 100644 --- a/apps/worker/test/mockPrisma.ts +++ b/apps/worker/test/mockPrisma.ts @@ -3,9 +3,9 @@ import { mockDeep, DeepMockProxy } from 'jest-mock-extended' import { prisma } from '../src/prisma' -jest.mock('../src/prisma', () => ({ - __esModule: true, - prisma: mockDeep(), -})) +jest.mock('../src/prisma', () => { + const prisma = mockDeep() + return { __esModule: true, prisma, rollupPrisma: prisma } +}) export const prismaMock = prisma as unknown as DeepMockProxy From aeb9b8897f5e09195905671447ad50d469b4632f Mon Sep 17 00:00:00 2001 From: needs <624097+needs@users.noreply.github.com> Date: Thu, 10 Sep 2026 14:39:33 +0200 Subject: [PATCH 3/4] Drop rationale comments and the CSV escape unit test Co-Authored-By: Claude Fable 5.1 --- apps/worker/src/ddnet/stream.test.ts | 24 ------------------------ apps/worker/src/ddnet/stream.ts | 2 -- apps/worker/src/prisma.ts | 3 --- apps/worker/src/rollup/writeDayRollup.ts | 2 -- 4 files changed, 31 deletions(-) delete mode 100644 apps/worker/src/ddnet/stream.test.ts diff --git a/apps/worker/src/ddnet/stream.test.ts b/apps/worker/src/ddnet/stream.test.ts deleted file mode 100644 index b865d69..0000000 --- a/apps/worker/src/ddnet/stream.test.ts +++ /dev/null @@ -1,24 +0,0 @@ -import { Readable } from "stream"; -import { parseCsvEntry } from "./stream"; - -describe('parseCsvEntry', () => { - it('decodes the MySQL-style backslash escapes used by the dump', async () => { - const csv = [ - '"Map","Name","Time"', - '"For Idiots 1","Tobias\\"",514.2', - '"Bootcamp #2","I\\\\I4I\\\\/I3",808.46', - '"Lowcore","-/Noob\\\\-Fra",167.14', - ].join('\n') + '\n'; - - const records: string[][] = []; - await parseCsvEntry(Readable.from([csv]), (record) => { - records.push(record); - }); - - expect(records).toEqual([ - ['For Idiots 1', 'Tobias"', '514.2'], - ['Bootcamp #2', 'I\\I4I\\/I3', '808.46'], - ['Lowcore', '-/Noob\\-Fra', '167.14'], - ]); - }); -}); diff --git a/apps/worker/src/ddnet/stream.ts b/apps/worker/src/ddnet/stream.ts index 1adf08e..8601977 100644 --- a/apps/worker/src/ddnet/stream.ts +++ b/apps/worker/src/ddnet/stream.ts @@ -76,7 +76,6 @@ export async function parseCsvEntry( entry: NodeJS.ReadableStream, onRecord: (record: string[], header: string[]) => void | Promise ) { - // The dump is a MySQL export: quotes and backslashes are backslash-escaped. const parser = entry.pipe(parse({ relax_column_count: true, bom: true, @@ -177,7 +176,6 @@ export async function streamStatsDump(handlers: DumpHandlers) { } } } finally { - // Bailing out mid-archive must not leave the download open. zip.destroy(); source.destroy(); } diff --git a/apps/worker/src/prisma.ts b/apps/worker/src/prisma.ts index 86c416b..9b30c65 100644 --- a/apps/worker/src/prisma.ts +++ b/apps/worker/src/prisma.ts @@ -15,9 +15,6 @@ function withConnectionLimit(url: string | undefined, limit: number) { return parsed.toString(); } -// The day rollup streams a whole day in small batches; on the shared pool each -// batch queues behind the hundred concurrent poll transactions and the job -// runs out its time budget while both the database and the worker sit idle. export const rollupPrisma = new PrismaClient({ datasourceUrl: withConnectionLimit(prismaDatabaseUrl, 2), }); diff --git a/apps/worker/src/rollup/writeDayRollup.ts b/apps/worker/src/rollup/writeDayRollup.ts index ee9ffa1..15a260c 100644 --- a/apps/worker/src/rollup/writeDayRollup.ts +++ b/apps/worker/src/rollup/writeDayRollup.ts @@ -166,8 +166,6 @@ export async function writeDayRollup(day: Date, rollup: DayRollup) { await prisma.$transaction( async (tx) => { - // Day and backfill rollups touch the same Player and PlayerPartner - // rows; serializing writers avoids deadlocking on them. await tx.$queryRawTyped(lockRollupWrite()); const alreadyRolledUp = From d4a283da90c7969d0b7f051f3ac893f9d721fa7d Mon Sep 17 00:00:00 2001 From: needs <624097+needs@users.noreply.github.com> Date: Thu, 10 Sep 2026 16:04:29 +0200 Subject: [PATCH 4/4] Keep parseCsvEntry private to the stream module Co-Authored-By: Claude Fable 5.1 --- apps/worker/src/ddnet/stream.ts | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/apps/worker/src/ddnet/stream.ts b/apps/worker/src/ddnet/stream.ts index 8601977..3f14637 100644 --- a/apps/worker/src/ddnet/stream.ts +++ b/apps/worker/src/ddnet/stream.ts @@ -72,7 +72,7 @@ function parseSplits(record: string[]): number[] { return hasData ? splits : []; } -export async function parseCsvEntry( +async function parseCsvEntry( entry: NodeJS.ReadableStream, onRecord: (record: string[], header: string[]) => void | Promise ) {