Fix the failures shown on /status: DDNet imports, rollup deadlocks and stalls

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',

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<unzipper.Entry>) {
-    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<unzipper.Entry>) {
+      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();
   }
 }
 

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),
+});

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);

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<PrismaClient, 'gameServerSnapshot'>;
 }): AsyncGenerator<IteratedSnapshot, void, undefined> {
   let cursor = 0;
 

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<unknown>) =>
+      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);
+  });
+});

apps/worker/src/workers/ddnetOnlineImport.ts

@@ -94,7 +94,8 @@ async function importSource(source: typeof SOURCES[number]) {
   const decoder = new TextDecoder();
 
   let pendingRows: ReturnType<typeof finalizeDay> = [];
-  let current: DayAggregate | null = null;
+  const openDays = new Map<string, DayAggregate>();
+  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);
   }
 }