diff --git a/packages/common/types.ts b/packages/common/types.ts index bcbbc65cc..1132b71ab 100644 --- a/packages/common/types.ts +++ b/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; } diff --git a/packages/hydrojudge/src/hosts/builtin.ts b/packages/hydrojudge/src/hosts/builtin.ts index fffa6d3d9..b4468ed84 100644 --- a/packages/hydrojudge/src/hosts/builtin.ts +++ b/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, - SettingModel, StorageModel, SystemModel, TaskModel, + Context as HydroContext, db, JudgeHandler, JudgeResultCallbackContext, + 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) { - t._callbackAwait ||= Promise.resolve(); - return (data: Partial) => { - logger.debug('Next: %o', data); - if (data.case) data.case.message ||= ''; - 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) => { - 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 })); + getReporter(t: Context) { + const reporter = new JudgeResultCallbackContext(app, t.request); + return { + next: (a) => reporter.next(a), + end: (a) => reporter.end(a), }; }, getLang(lang: string, doThrow = true) { diff --git a/packages/hydrojudge/src/hosts/hydro.ts b/packages/hydrojudge/src/hosts/hydro.ts index 192a6a77f..6c75a5fed 100644 --- a/packages/hydrojudge/src/hosts/hydro.ts +++ b/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) { - return (data: Partial) => { + getReporter(t: JudgeTask) { + const next = (data: Partial) => { 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); } }; - } - - getEnd(t: JudgeTask) { - return (data: Partial) => { + const end = (data: Partial) => { 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) { diff --git a/packages/hydrojudge/src/hosts/vj4.ts b/packages/hydrojudge/src/hosts/vj4.ts index 6cd4120d3..64808c0bb 100644 --- a/packages/hydrojudge/src/hosts/vj4.ts +++ b/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) { - return (data: Partial) => { + getReporter(t: JudgeTask) { + const next = (data: Partial) => { log.debug('Next: %o', data); this.send((t.request as any).tag, 'next', data); }; - } - - getEnd(t: JudgeTask) { - return (data: Partial) => { + const end = (data: Partial) => { log.info('End: %o', data); this.send((t.request as any).tag, 'end', data); }; + return { next, end }; } get(url: string) { diff --git a/packages/hydrojudge/src/interface.ts b/packages/hydrojudge/src/interface.ts index ed4f3eefb..9dc5f684e 100644 --- a/packages/hydrojudge/src/interface.ts +++ b/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; - getEnd: (task: JudgeTask) => NextFunction; + getReporter: (task: JudgeTask) => { next: NextFunction, end: NextFunction }; fetchFile: (namespace: T, files: Record) => Promise; postFile: (target: string, filename: string, file: string) => Promise; config: { detail: boolean, host?: string, trusted?: boolean }; diff --git a/packages/hydrojudge/src/judge/interface.ts b/packages/hydrojudge/src/judge/interface.ts index ca9f6099d..0393edb3a 100644 --- a/packages/hydrojudge/src/judge/interface.ts +++ b/packages/hydrojudge/src/judge/interface.ts @@ -22,8 +22,6 @@ export interface RuntimeContext { executeInteractor?: Execute; executeManager?: Execute; executeUser?: Execute; - - _callbackAwait?: Promise; } export interface ContextSubTask { diff --git a/packages/hydrojudge/src/task.ts b/packages/hydrojudge/src/task.ts index 6b5e30257..5d2bebd75 100644 --- a/packages/hydrojudge/src/task.ts +++ b/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); - this.end = this.session.getEnd(this); + Object.assign(this, this.session.getReporter(this)); logger.info('Submission: %s/%s/%s', host, this.source, this.rid); await this.doSubmission(); } catch (e) { diff --git a/packages/hydrooj/src/handler/judge.ts b/packages/hydrooj/src/handler/judge.ts index f8d76c2b4..be8434b3c 100644 --- a/packages/hydrooj/src/handler/judge.ts +++ b/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,47 +72,124 @@ function processPayload(body: Partial) { }; } -export async function next(body: Partial) { - const { - $set, $push, $unset, $inc, - } = processPayload(body); - const rdoc = await record.update(body.domainId, new ObjectId(body.rid), $set, $push, $unset, $inc); - if (rdoc) bus.broadcast('record/change', rdoc, $set, $push, body); - return rdoc; -} +export class JudgeResultCallbackContext { + private resolve: (_: any) => void; + private finishPromise: Promise; + private operationPromise = Promise.resolve(null); + private relatedId = new ObjectId(); + private meta: JudgeMeta; -export async function 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); - if (rdoc.contest) await contest.updateStatus(rdoc.domainId, rdoc.contest, rdoc.uid, rdoc._id, rdoc.pid, rdoc); - else if (accept && updated) await domain.incUserInDomain(rdoc.domainId, rdoc.uid, 'nAccept', 1); - const isNormalSubmission = ![ - STATUS.STATUS_ETC, STATUS.STATUS_HACK_SUCCESSFUL, STATUS.STATUS_HACK_UNSUCCESSFUL, - STATUS.STATUS_FORMAT_ERROR, STATUS.STATUS_SYSTEM_ERROR, STATUS.STATUS_CANCELED, - ].includes(rdoc.status); - const pdoc = (accept && updated) - ? await problem.inc(rdoc.domainId, rdoc.pid, 'nAccept', 1) - : await problem.get(rdoc.domainId, rdoc.pid, undefined, true); - if (pdoc && isNormalSubmission) { - await Promise.all([ - problem.inc(pdoc.domainId, pdoc.docId, `stats.${builtin.STATUS_SHORT_TEXTS[rdoc.status]}`, 1), - problem.inc(pdoc.domainId, pdoc.docId, `stats.s${Math.floor(rdoc.score)}`, 1), - ]); + constructor(public ctx: Context, public readonly task: Omit & { 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) { + const { + $set, $push, $unset, $inc, + } = processPayload(body); + if (this.meta.rejudge === 'controlled') { + await record.collHistory.updateOne({ + _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); + } + } + + static async next(domainId: string, rid: ObjectId, body: Partial) { + 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) { + 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); + if (rdoc.contest) await contest.updateStatus(rdoc.domainId, rdoc.contest, rdoc.uid, rdoc._id, rdoc.pid, rdoc); + else if (accept && updated) await domain.incUserInDomain(rdoc.domainId, rdoc.uid, 'nAccept', 1); + const isNormalSubmission = ![ + STATUS.STATUS_ETC, STATUS.STATUS_HACK_SUCCESSFUL, STATUS.STATUS_HACK_UNSUCCESSFUL, + STATUS.STATUS_FORMAT_ERROR, STATUS.STATUS_SYSTEM_ERROR, STATUS.STATUS_CANCELED, + ].includes(rdoc.status); + const pdoc = (accept && updated) + ? await problem.inc(rdoc.domainId, rdoc.pid, 'nAccept', 1) + : await problem.get(rdoc.domainId, rdoc.pid, undefined, true); + if (pdoc && isNormalSubmission) { + await Promise.all([ + problem.inc(pdoc.domainId, pdoc.docId, `stats.${builtin.STATUS_SHORT_TEXTS[rdoc.status]}`, 1), + problem.inc(pdoc.domainId, pdoc.docId, `stats.s${Math.floor(rdoc.score)}`, 1), + ]); + } + await app.parallel('record/judge', rdoc, updated, pdoc, this); + } + + async _end(body: Partial) { + const { $set, $push } = processPayload(body); + const $unset: any = { progress: '' }; + $set.judgeAt = new Date(); + $set.judger = body.judger ?? 1; + + 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 JudgeResultCallbackContext.postJudge(rdoc); + this.resolve(rdoc); + } + + static async end(domainId: string, rid: ObjectId, body: Partial) { + 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) { + 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); } - await bus.parallel('record/judge', rdoc, updated, pdoc); } -export async function end(body: Partial) { - 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 (rdoc) bus.broadcast('record/change', rdoc, null, null, body); // trigger a full update - await postJudge(rdoc); - return rdoc; -} +/** @deprecated use JudgeResultCallbackContext.postJudge instead */ +export const postJudge = (rdoc: RecordDoc) => JudgeResultCallbackContext.postJudge(rdoc); export class JudgeFilesDownloadHandler extends Handler { noCheckPermView = true; @@ -185,12 +263,7 @@ export class JudgeConnectionHandler extends ConnectionHandler { query: any = { type: { $in: ['judge', 'generate'] } }; concurrency = 1; consumer: Consumer = null; - tasks: Record void, - queue: PQueue, - domainId: string, - t: Task, - }> = {}; + tasks: Record = {}; 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; - const p = new Promise((r) => { resolve = r; }); - this.tasks[rid] = { - queue: new PQueue({ concurrency: 1 }), - domainId: t.domainId, - resolve, - t, - }; + this.tasks[rid] = new JudgeResultCallbackContext(this.ctx, t); this.send({ task: t }); - this.tasks[rid].queue.add(() => next({ status: STATUS.STATUS_FETCHED, domainId: t.domainId, rid: t.rid })); - await p; + this.tasks[rid].next({ status: STATUS.STATUS_FETCHED }); + 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.queue.add(() => next(msg)); - if (msg.key === 'end') { - if (!msg.nop) t.queue.add(() => end({ judger: this.user._id, ...msg })); - t.resolve(null); - } + if (msg.key === 'next') t.next(msg); + if (msg.key === 'end') t.end(msg.nop ? undefined : { judger: this.user._id, ...msg }); } 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 }) => { - const rdoc = await record.reset(t.domainId, t.rid, false); - bus.broadcast('record/change', rdoc); - return task.add(t); - })); + await Promise.all(Object.values(this.tasks).map((cb) => cb.reset())); } } @@ -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; diff --git a/packages/hydrooj/src/model/record.ts b/packages/hydrooj/src/model/record.ts index f6b213fcb..a1114a8fb 100644 --- a/packages/hydrooj/src/model/record.ts +++ b/packages/hydrooj/src/model/record.ts @@ -248,18 +248,20 @@ 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) => ({ - ...pick(rdoc, [ - 'compilerTexts', 'judgeTexts', 'testCases', 'subtasks', - 'score', 'time', 'memory', 'status', 'judgeAt', 'judger', - ]), - rid: rdoc._id, - _id: new ObjectId(), - }))); + if (rdocs.length) { + await RecordModel.collHistory.insertMany(rdocs.map((rdoc) => ({ + ...pick(rdoc, [ + 'compilerTexts', 'judgeTexts', 'testCases', 'subtasks', + 'score', 'time', 'memory', 'status', 'judgeAt', 'judger', + ]), + rid: rdoc._id, + _id: new ObjectId(), + }))); + } return RecordModel.update(domainId, rid, upd); } diff --git a/packages/hydrooj/src/plugin-api.ts b/packages/hydrooj/src/plugin-api.ts index b75c70c3b..6c738d7d3 100644 --- a/packages/hydrooj/src/plugin-api.ts +++ b/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'; diff --git a/packages/hydrooj/src/service/bus.ts b/packages/hydrooj/src/service/bus.ts index 259f08817..cd4adee83 100644 --- a/packages/hydrooj/src/service/bus.ts +++ b/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) { diff --git a/packages/vjudge/src/index.ts b/packages/vjudge/src/index.ts index a852d302c..bec31c55c 100644 --- a/packages/vjudge/src/index.ts +++ b/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 end = (payload) => JudgeHandler.end({ ...payload, rid: task.rid, domainId: task.domainId }); + const context = new JudgeResultCallbackContext(this.ctx, task); + 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; }