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.ts b/apps/worker/src/ddnet/stream.ts index e9f77b9..3f14637 100644 --- a/apps/worker/src/ddnet/stream.ts +++ b/apps/worker/src/ddnet/stream.ts @@ -79,6 +79,7 @@ async function parseCsvEntry( const parser = entry.pipe(parse({ relax_column_count: true, bom: true, + escape: '\\', })); let header: string[] | null = null; @@ -109,68 +110,74 @@ 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 { + zip.destroy(); + source.destroy(); } } diff --git a/apps/worker/src/prisma.ts b/apps/worker/src/prisma.ts index 773ed0e..9b30c65 100644 --- a/apps/worker/src/prisma.ts +++ b/apps/worker/src/prisma.ts @@ -5,3 +5,16 @@ 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(); +} + +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 ec5f81c..15a260c 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,8 @@ export async function writeDayRollup(day: Date, rollup: DayRollup) { await prisma.$transaction( async (tx) => { + await tx.$queryRawTyped(lockRollupWrite()); + const alreadyRolledUp = (await tx.globalDay.findUnique({ where: { day }, select: { day: true } })) !== null; @@ -220,7 +222,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/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/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/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 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}`, } }); }