judge: allow unlimited access & fix file corruption when errored (#726)

Co-authored-by: undefined <i@undefined.moe>

packages/hydrojudge/src/hosts/hydro.ts

@@ -3,7 +3,7 @@ import path from 'path';
 import PQueue from 'p-queue';
 import superagent from 'superagent';
 import WebSocket from 'ws';
-import { fs } from '@hydrooj/utils';
+import { fs, pipeRequest } from '@hydrooj/utils';
 import { LangConfig } from '@hydrooj/utils/lib/lang';
 import * as sysinfo from '@hydrooj/utils/lib/sysinfo';
 import type { JudgeResultBody } from 'hydrooj';
@@ -91,30 +91,15 @@ export default class Hydro implements Session {
                 files: filenames,
             });
             if (!res.body.links) throw new FormatError('problem not exist');
-            const that = this;
-            // eslint-disable-next-line no-inner-declarations
-            async function download(name: string) {
-                if (name.includes('/')) await fs.ensureDir(path.join(filePath, name.split('/')[0]));
-                const w = fs.createWriteStream(path.join(filePath, name));
-                that.get(res.body.links[name]).pipe(w);
-                await new Promise((resolve, reject) => {
-                    const timeout = setTimeout(() => reject(new Error(`DownloadTimeout(${name}, 60s)`)), 60000);
-                    w.on('error', (e) => {
-                        clearTimeout(timeout);
-                        reject(new Error(`DownloadFail(${name}): ${e.message}`));
-                    });
-                    w.on('finish', () => {
-                        clearTimeout(timeout);
-                        resolve(null);
-                    });
-                });
-            }
             const tasks = [];
             const queue = new PQueue({ concurrency: 10 });
             for (const name in res.body.links) {
-                tasks.push(queue.add(() => download(name)));
+                tasks.push(queue.add(async () => {
+                    if (name.includes('/')) await fs.ensureDir(path.join(filePath, name.split('/')[0]));
+                    const w = fs.createWriteStream(path.join(filePath, name));
+                    await pipeRequest(this.get(res.body.links[name]), w, 60000, name);
+                }));
             }
-            queue.start();
             await Promise.all(tasks);
             await fs.writeFile(path.join(filePath, 'etags'), JSON.stringify(version));
         }
@@ -126,12 +111,7 @@ export default class Hydro implements Session {
         name = name.split('#')[0];
         const res = await this.post('judge/files', { id: name });
         const target = path.join(getConfig('tmp_dir'), name.replace(/\//g, '_'));
-        const w = fs.createWriteStream(target);
-        this.get(res.body.url).pipe(w);
-        await new Promise((resolve, reject) => {
-            w.on('finish', resolve);
-            w.on('error', (e) => reject(new Error(`DownloadFail(${name}): ${e.message}`)));
-        });
+        await pipeRequest(this.get(res.body.url), fs.createWriteStream(target), 60000, name);
         return target;
     }
 

packages/hydrojudge/src/sandbox/client.ts

@@ -1,5 +1,6 @@
 import fs from 'fs';
 import superagent from 'superagent';
+import { pipeRequest } from '@hydrooj/utils';
 import { getConfig } from '../config';
 import { SandboxRequest, SandboxResult, SandboxVersion } from './interface';
 
@@ -12,11 +13,7 @@ const client = new Proxy({
     getFile(fileId: string, dest?: string): Promise<Buffer> {
         if (dest) {
             const w = fs.createWriteStream(dest);
-            superagent.get(`${url}/file/${fileId}`).pipe(w);
-            return new Promise((resolve, reject) => {
-                w.on('finish', () => resolve(null));
-                w.on('error', reject);
-            });
+            return pipeRequest(superagent.get(`${url}/file/${fileId}`), w, 60000, fileId) as any;
         }
         return superagent.get(`${url}/file/${fileId}`).responseType('arraybuffer').then((res) => res.body);
     },