core: judge: ensure callback order (#719)

packages/hydrooj/src/handler/judge.ts

@@ -3,6 +3,7 @@ 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 {
     FileLimitExceededError, ForbiddenError, ProblemIsReferencedError, ValidationError,
@@ -139,7 +140,6 @@ export async function end(body: Partial<JudgeResultBody>) {
     const $unset: any = { progress: '' };
     $set.judgeAt = new Date();
     $set.judger = body.judger ?? 1;
-    await sleep(100); // Make sure that all 'next' event already triggered
     let rdoc = await record.update(body.domainId, body.rid, $set, $push, $unset);
     await postJudge(rdoc);
     rdoc = await record.get(body.rid);
@@ -218,6 +218,7 @@ class JudgeConnectionHandler extends ConnectionHandler {
     query: any = { type: { $in: ['judge', 'generate'] } };
     rdocs: Record<string, RecordDoc> = {};
     concurrency = 1;
+    queue: PQueue = new PQueue({ concurrency: 1 });
     callback: (res?: any) => void = null;
 
     async prepare() {
@@ -272,12 +273,15 @@ class JudgeConnectionHandler extends ConnectionHandler {
         if (['next', 'end'].includes(msg.key)) {
             const rdoc = this.rdocs[msg.rid];
             if (!rdoc) return;
-            if (msg.key === 'next') await next({ ...msg, domainId: rdoc.domainId });
+            msg.domainId = rdoc.domainId;
+            if (msg.key === 'next') this.queue.add(() => next(msg));
             if (msg.key === 'end') {
                 this.processing = this.processing.filter((t) => t.rid.toHexString() !== msg.rid);
                 delete this.rdocs[msg.rid];
                 this.callback?.();
-                if (!msg.nop) await end({ judger: this.user._id, ...msg, domainId: rdoc.domainId }).catch((e) => logger.error(e));
+                if (!msg.nop) {
+                    this.queue.add(() => end({ judger: this.user._id, ...msg }).catch((e) => logger.error(e)));
+                }
             }
         } else if (msg.key === 'status') {
             await updateJudge(msg.info);