core: add controlled rejudge (#969)

Co-authored-by: panda <panda_dtdyy@outlook.com>

packages/common/types.ts

@@ -76,7 +76,7 @@ export interface FileInfo {
 export interface JudgeMeta {
     problemOwner: number;
     hackRejudge?: string;
-    rejudge?: boolean;
+    rejudge?: boolean | 'controlled';
     // FIXME stricter types
     type?: string;
 }

packages/hydrojudge/src/hosts/builtin.ts

@@ -4,8 +4,8 @@ import path from 'path';
 import { fs } from '@hydrooj/utils';
 import * as sysinfo from '@hydrooj/utils/lib/sysinfo';
 import {
-    Context as HydroContext, db, JudgeHandler, JudgeResultBody, ObjectId, RecordModel,
+    Context as HydroContext, db, JudgeHandler, JudgeResultCallbackContext,
-    SettingModel, StorageModel, SystemModel, TaskModel,
+    ObjectId, RecordModel, SettingModel, StorageModel, SystemModel, TaskModel,
 } from 'hydrooj';
 import { langs } from 'hydrooj/src/model/setting';
 import { getConfig } from '../config';
@@ -32,20 +32,11 @@ const session: Session = {
         }
         return null;
     },
-    getNext(t: Context) {
+    getReporter(t: Context) {
-        t._callbackAwait ||= Promise.resolve();
+        const reporter = new JudgeResultCallbackContext(app, t.request);
-        return (data: Partial<JudgeResultBody>) => {
+        return {
-            logger.debug('Next: %o', data);
+            next: (a) => reporter.next(a),
-            if (data.case) data.case.message ||= '';
+            end: (a) => reporter.end(a),
-            t._callbackAwait = t._callbackAwait.then(() => JudgeHandler.next({ ...data, rid: t.rid, domainId: t.request.domainId }));
-        };
-    },
-    getEnd(t: Context) {
-        t._callbackAwait ||= Promise.resolve();
-        return (data: Partial<JudgeResultBody>) => {
-            data.key = 'end';
-            logger.info('End: status=%d score=%d time=%dms memory=%dkb', data.status, data.score, data.time, data.memory);
-            t._callbackAwait = t._callbackAwait.then(() => JudgeHandler.end({ ...data, rid: t.rid, domainId: t.request.domainId }));
         };
     },
     getLang(lang: string, doThrow = true) {

packages/hydrojudge/src/hosts/hydro.ts

@@ -106,8 +106,8 @@ export default class Hydro implements Session {
         this.ws.send(JSON.stringify({ ...data, rid, key }));
     }
 
-    getNext(t: JudgeTask) {
+    getReporter(t: JudgeTask) {
-        return (data: Partial<JudgeResultBody>) => {
+        const next = (data: Partial<JudgeResultBody>) => {
             log.debug('Next: %o', data);
             const performanceMode = getConfig('performance') || t.meta.rejudge || t.meta.hackRejudge;
             if (performanceMode && data.case && !data.compilerText && !data.message) {
@@ -119,14 +119,12 @@ export default class Hydro implements Session {
                 this.send(t.request.rid, 'next', data);
             }
         };
-    }
+        const end = (data: Partial<JudgeResultBody>) => {
-
-    getEnd(t: JudgeTask) {
-        return (data: Partial<JudgeResultBody>) => {
             log.info('End: %o', data);
             if (t.callbackCache) data.cases = t.callbackCache;
             this.send(t.request.rid, 'end', data);
         };
+        return { next, end };
     }
 
     async consume(queue: PQueue) {

packages/hydrojudge/src/hosts/vj4.ts

@@ -79,18 +79,16 @@ export default class VJ4 implements Session {
         this.ws.send(JSON.stringify({ ...data, tag, key }));
     }
 
-    getNext(t: JudgeTask) {
+    getReporter(t: JudgeTask) {
-        return (data: Partial<JudgeResultBody>) => {
+        const next = (data: Partial<JudgeResultBody>) => {
             log.debug('Next: %o', data);
             this.send((t.request as any).tag, 'next', data);
         };
-    }
+        const end = (data: Partial<JudgeResultBody>) => {
-
-    getEnd(t: JudgeTask) {
-        return (data: Partial<JudgeResultBody>) => {
             log.info('End: %o', data);
             this.send((t.request as any).tag, 'end', data);
         };
+        return { next, end };
     }
 
     get(url: string) {

packages/hydrojudge/src/interface.ts

@@ -24,8 +24,7 @@ export { JudgeRequest } from '@hydrooj/common';
 
 export interface Session {
     getLang: (name: string, doThrow?: boolean) => LangConfig;
-    getNext: (task: JudgeTask) => NextFunction;
+    getReporter: (task: JudgeTask) => { next: NextFunction, end: NextFunction };
-    getEnd: (task: JudgeTask) => NextFunction;
     fetchFile: <T extends null | string>(namespace: T, files: Record<string, string>) => Promise<T extends null ? string : null>;
     postFile: (target: string, filename: string, file: string) => Promise<void>;
     config: { detail: boolean, host?: string, trusted?: boolean };

packages/hydrojudge/src/judge/interface.ts

@@ -22,8 +22,6 @@ export interface RuntimeContext {
     executeInteractor?: Execute;
     executeManager?: Execute;
     executeUser?: Execute;
-
-    _callbackAwait?: Promise<any>;
 }
 
 export interface ContextSubTask {

packages/hydrojudge/src/task.ts

@@ -70,8 +70,7 @@ export class JudgeTask {
                 HYDRO_USER: (this.request.uid || 0).toString(),
                 HYDRO_CONTEST: tid,
             };
-            this.next = this.session.getNext(this);
+            Object.assign(this, this.session.getReporter(this));
-            this.end = this.session.getEnd(this);
             logger.info('Submission: %s/%s/%s', host, this.source, this.rid);
             await this.doSubmission();
         } catch (e) {

packages/hydrooj/src/handler/judge.ts

@@ -3,9 +3,10 @@ import fs from 'fs-extra';
 import yaml from 'js-yaml';
 import { omit } from 'lodash';
 import { ObjectId } from 'mongodb';
-import PQueue from 'p-queue';
 import sanitize from 'sanitize-filename';
-import { JudgeResultBody, ProblemConfigFile, TestCase } from '@hydrooj/common';
+import {
+    JudgeMeta, JudgeResultBody, ProblemConfigFile, TestCase,
+} from '@hydrooj/common';
 import { Context } from '../context';
 import {
     BadRequestError, FileLimitExceededError, ForbiddenError, ProblemIsReferencedError, ValidationError,
@@ -71,16 +72,50 @@ function processPayload(body: Partial<JudgeResultBody>) {
     };
 }
 
-export async function next(body: Partial<JudgeResultBody>) {
+export class JudgeResultCallbackContext {
+    private resolve: (_: any) => void;
+    private finishPromise: Promise<any>;
+    private operationPromise = Promise.resolve(null);
+    private relatedId = new ObjectId();
+    private meta: JudgeMeta;
+
+    constructor(public ctx: Context, public readonly task: Omit<Task, '_id'> & { type: string }) { // eslint-disable-line @typescript-eslint/no-shadow
+        this.meta = task.meta as JudgeMeta;
+        this.finishPromise = new Promise((resolve) => {
+            this.resolve = resolve;
+        });
+    }
+
+    async _next(body: Partial<JudgeResultBody>) {
         const {
             $set, $push, $unset, $inc,
         } = processPayload(body);
-    const rdoc = await record.update(body.domainId, new ObjectId(body.rid), $set, $push, $unset, $inc);
+        if (this.meta.rejudge === 'controlled') {
-    if (rdoc) bus.broadcast('record/change', rdoc, $set, $push, body);
+            await record.collHistory.updateOne({
-    return rdoc;
+                _id: this.relatedId,
-}
+            }, {
+                $set, $push, $unset, $inc,
+            }, { upsert: true });
+        } else {
+            const rdoc = await record.update(this.task.domainId, new ObjectId(this.task.rid as string), $set, $push, $unset, $inc);
+            if (rdoc) this.ctx.broadcast('record/change', rdoc, $set, $push, body);
+        }
+    }
 
-export async function postJudge(rdoc: RecordDoc) {
+    static async next(domainId: string, rid: ObjectId, body: Partial<JudgeResultBody>) {
+        const {
+            $set, $push, $unset, $inc,
+        } = processPayload(body);
+        const rdoc = await record.update(domainId, rid, $set, $push, $unset, $inc);
+        if (rdoc) app.broadcast('record/change', rdoc, $set, $push, body);
+    }
+
+    next(body: Partial<JudgeResultBody>) {
+        this.operationPromise = this.operationPromise.then(() => this._next(body));
+        return this.operationPromise;
+    }
+
+    static async postJudge(rdoc: RecordDoc) {
         if (rdoc.contest?.toString().startsWith('0'.repeat(23))) return;
         const accept = rdoc.status === builtin.STATUS.STATUS_ACCEPTED;
         const updated = await problem.updateStatus(rdoc.domainId, rdoc.pid, rdoc.uid, rdoc._id, rdoc.status, rdoc.score);
@@ -99,20 +134,63 @@ export async function postJudge(rdoc: RecordDoc) {
                 problem.inc(pdoc.domainId, pdoc.docId, `stats.s${Math.floor(rdoc.score)}`, 1),
             ]);
         }
-    await bus.parallel('record/judge', rdoc, updated, pdoc);
+        await app.parallel('record/judge', rdoc, updated, pdoc, this);
-}
+    }
 
-export async function end(body: Partial<JudgeResultBody>) {
+    async _end(body: Partial<JudgeResultBody>) {
         const { $set, $push } = processPayload(body);
         const $unset: any = { progress: '' };
         $set.judgeAt = new Date();
         $set.judger = body.judger ?? 1;
-    const rdoc = await record.update(body.domainId, new ObjectId(body.rid), $set, $push, $unset);
+
+        if (this.meta.rejudge === 'controlled') {
+            await record.collHistory.updateOne({
+                _id: this.relatedId,
+            }, {
+                $set, $push, $unset,
+            }, { upsert: true });
+            this.resolve(null);
+            return;
+        }
+
+        const rdoc = await record.update(this.task.domainId, new ObjectId(this.task.rid as string), $set, $push, $unset);
         if (rdoc) bus.broadcast('record/change', rdoc, null, null, body); // trigger a full update
-    await postJudge(rdoc);
+        await JudgeResultCallbackContext.postJudge(rdoc);
-    return rdoc;
+        this.resolve(rdoc);
+    }
+
+    static async end(domainId: string, rid: ObjectId, body: Partial<JudgeResultBody>) {
+        const { $set, $push } = processPayload(body);
+        const $unset: any = { progress: '' };
+        $set.judgeAt = new Date();
+        $set.judger = body.judger ?? 1;
+        const rdoc = await record.update(domainId, rid, $set, $push, $unset);
+        if (rdoc) app.broadcast('record/change', rdoc, null, null, body); // trigger a full update
+        await JudgeResultCallbackContext.postJudge(rdoc);
+    }
+
+    end(body?: Partial<JudgeResultBody>) {
+        if (!body) this.resolve(null);
+        else this.operationPromise = this.operationPromise.then(() => this._end(body));
+        return this.operationPromise;
+    }
+
+    reset() {
+        return this.operationPromise.then(async () => {
+            const rdoc = await record.reset(this.task.domainId, this.task.rid, false);
+            this.ctx.broadcast('record/change', rdoc);
+            return task.add(this.task);
+        });
+    }
+
+    then(onfulfilled?: (value: any) => void, onrejected?: (reason: any) => void) {
+        return this.finishPromise.then(onfulfilled, onrejected);
+    }
 }
 
+/** @deprecated use JudgeResultCallbackContext.postJudge instead */
+export const postJudge = (rdoc: RecordDoc) => JudgeResultCallbackContext.postJudge(rdoc);
+
 export class JudgeFilesDownloadHandler extends Handler {
     noCheckPermView = true;
     notUsage = true;
@@ -185,12 +263,7 @@ export class JudgeConnectionHandler extends ConnectionHandler {
     query: any = { type: { $in: ['judge', 'generate'] } };
     concurrency = 1;
     consumer: Consumer = null;
-    tasks: Record<string, {
+    tasks: Record<string, JudgeResultCallbackContext> = {};
-        resolve: (_: any) => void,
-        queue: PQueue,
-        domainId: string,
-        t: Task,
-    }> = {};
 
     async prepare() {
         logger.info('Judge daemon connected from ', this.request.ip);
@@ -204,17 +277,10 @@ export class JudgeConnectionHandler extends ConnectionHandler {
 
     async newTask(t: Task) {
         const rid = t.rid.toHexString();
-        let resolve: (_: any) => void;
+        this.tasks[rid] = new JudgeResultCallbackContext(this.ctx, t);
-        const p = new Promise((r) => { resolve = r; });
-        this.tasks[rid] = {
-            queue: new PQueue({ concurrency: 1 }),
-            domainId: t.domainId,
-            resolve,
-            t,
-        };
         this.send({ task: t });
-        this.tasks[rid].queue.add(() => next({ status: STATUS.STATUS_FETCHED, domainId: t.domainId, rid: t.rid }));
+        this.tasks[rid].next({ status: STATUS.STATUS_FETCHED });
-        await p;
+        await this.tasks[rid];
         delete this.tasks[rid];
     }
 
@@ -227,12 +293,8 @@ export class JudgeConnectionHandler extends ConnectionHandler {
         if (['next', 'end'].includes(msg.key)) {
             const t = this.tasks[msg.rid];
             if (!t) return;
-            msg.domainId = t.domainId;
+            if (msg.key === 'next') t.next(msg);
-            if (msg.key === 'next') t.queue.add(() => next(msg));
+            if (msg.key === 'end') t.end(msg.nop ? undefined : { judger: this.user._id, ...msg });
-            if (msg.key === 'end') {
-                if (!msg.nop) t.queue.add(() => end({ judger: this.user._id, ...msg }));
-                t.resolve(null);
-            }
         } else if (msg.key === 'status') {
             await updateJudge(msg.info);
         } else if (msg.key === 'config') {
@@ -262,11 +324,7 @@ export class JudgeConnectionHandler extends ConnectionHandler {
     async cleanup() {
         this.consumer?.destroy();
         logger.info('Judge daemon disconnected from ', this.request.ip);
-        await Promise.all(Object.values(this.tasks).map(async ({ t }) => {
+        await Promise.all(Object.values(this.tasks).map((cb) => cb.reset()));
-            const rdoc = await record.reset(t.domainId, t.rid, false);
-            bus.broadcast('record/change', rdoc);
-            return task.add(t);
-        }));
     }
 }
 
@@ -274,7 +332,7 @@ export async function apply(ctx: Context) {
     ctx.Route('judge_files_download', '/judge/files', JudgeFilesDownloadHandler, builtin.PRIV.PRIV_JUDGE);
     ctx.Route('judge_files_upload', '/judge/upload', JudgeFileUpdateHandler, builtin.PRIV.PRIV_JUDGE);
     ctx.Connection('judge_conn', '/judge/conn', JudgeConnectionHandler, builtin.PRIV.PRIV_JUDGE);
-    ctx.on('record/judge', async (rdoc, updated, pdoc) => {
+    ctx.on('record/judge', async (rdoc, updated, pdoc, t) => {
         if (!pdoc || rdoc.status !== STATUS.STATUS_HACK_SUCCESSFUL) return;
         if (rdoc.contest) return;
         try {
@@ -299,7 +357,7 @@ export async function apply(ctx: Context) {
             const priority = await record.submissionPriority(rdoc.uid, -5000 - rdocs.length * 5 - 50);
             await record.judge(rdoc.domainId, rdocs.map((r) => r._id), priority, {}, { hackRejudge: input });
         } catch (e) {
-            next({
+            t.next({
                 rid: rdoc._id.toString(),
                 domainId: rdoc.domainId,
                 key: 'next',
@@ -309,5 +367,11 @@ export async function apply(ctx: Context) {
     });
 }
 
+/** @deprecated use JudgeResultCallbackContext.next instead */
+export const next = (payload: any) => JudgeResultCallbackContext.next(payload.domainId, payload.rid, payload);
+/** @deprecated use JudgeResultCallbackContext.end instead */
+export const end = (payload: any) => JudgeResultCallbackContext.end(payload.domainId, payload.rid, payload);
+/** @deprecated use JudgeResultCallbackContext.next instead */
 apply.next = next;
+/** @deprecated use JudgeResultCallbackContext.end instead */
 apply.end = end;

packages/hydrooj/src/model/record.ts

@@ -248,11 +248,12 @@ export default class RecordModel {
         };
         if (isRejudge) upd.rejudged = true;
         const [rdocs] = await Promise.all([
-            RecordModel.coll.find({ _id: { $in: rids } }).toArray(),
+            RecordModel.coll.find({ _id: { $in: rids }, judgeAt: { $exists: true, $ne: null } }).toArray(),
             RecordModel.collStat.deleteMany({ _id: { $in: rids } }),
             task.deleteMany({ rid: { $in: rids } }),
         ]);
-        await RecordModel.collHistory.insertMany(rdocs.filter((rdoc) => rdoc.judgeAt).map((rdoc) => ({
+        if (rdocs.length) {
+            await RecordModel.collHistory.insertMany(rdocs.map((rdoc) => ({
                 ...pick(rdoc, [
                     'compilerTexts', 'judgeTexts', 'testCases', 'subtasks',
                     'score', 'time', 'memory', 'status', 'judgeAt', 'judger',
@@ -260,6 +261,7 @@ export default class RecordModel {
                 rid: rdoc._id,
                 _id: new ObjectId(),
             })));
+        }
         return RecordModel.update(domainId, rid, upd);
     }
 

packages/hydrooj/src/plugin-api.ts

@@ -33,8 +33,9 @@ export { default as DomainModel } from './model/domain';
 export { default as StorageModel } from './model/storage';
 export { default as TaskModel } from './model/task';
 export * from './model/builtin';
+/** @deprecated */
 export * as JudgeHandler from './handler/judge';
-export { postJudge } from './handler/judge';
+export { postJudge, JudgeResultCallbackContext } from './handler/judge';
 export { Collections } from './service/db';
 export { Handler, ConnectionHandler, requireSudo } from './service/server';
 export { Service, Context } from './context';

packages/hydrooj/src/service/bus.ts

@@ -90,7 +90,7 @@ export interface EventMap {
     'training/get': (tdoc: TrainingDoc, handler: any) => VoidReturn
 
     'record/change': (rdoc: RecordDoc, $set?: any, $push?: any, body?: any) => void
-    'record/judge': (rdoc: RecordDoc, updated: boolean, pdoc?: ProblemDoc) => VoidReturn
+    'record/judge': (rdoc: RecordDoc, updated: boolean, pdoc: ProblemDoc, updater: any) => VoidReturn
 }
 
 export function apply(ctx: Context) {

packages/vjudge/src/index.ts

@@ -2,7 +2,7 @@
 import os from 'os';
 import { LangConfig, STATUS } from '@hydrooj/common';
 import {
-    Context, db, DomainModel, JudgeHandler, Logger,
+    Context, db, DomainModel, JudgeResultCallbackContext, Logger,
     ProblemModel, RecordModel, Service, SettingModel,
     sleep, SystemModel, TaskModel, Time, yaml,
 } from 'hydrooj';
@@ -23,7 +23,7 @@ class AccountService {
     working = false;
     error = '';
 
-    constructor(public Provider: BasicProvider, public account: RemoteAccount) {
+    constructor(public Provider: BasicProvider, public account: RemoteAccount, public ctx: Context) {
         this.api = new Provider(account, async (data) => {
             await coll.updateOne({ _id: account._id }, { $set: data });
         });
@@ -45,8 +45,9 @@ class AccountService {
     }
 
     async judge(task) {
-        const next = (payload) => JudgeHandler.next({ ...payload, rid: task.rid, domainId: task.domainId });
+        const context = new JudgeResultCallbackContext(this.ctx, task);
-        const end = (payload) => JudgeHandler.end({ ...payload, rid: task.rid, domainId: task.domainId });
+        const next = (payload) => context.next(payload);
+        const end = (payload) => context.end(payload);
         await next({ status: STATUS.STATUS_FETCHED });
         try {
             const langConfig = SettingModel.langs[task.lang];
@@ -203,7 +204,7 @@ class VJudgeService extends Service {
             const services = [];
             for (const account of this.accounts.filter((a) => a.type === type)) {
                 if (account.enableOn && !account.enableOn.includes(os.hostname())) continue;
-                const service = new AccountService(provider, account);
+                const service = new AccountService(provider, account, this.ctx);
                 services.push(service);
                 this.pool[`${account.type}/${account.handle}`] = service;
             }