core: judge: bug fix

packages/hydrojudge/package.json

@@ -1,14 +1,13 @@
 {
     "name": "@hydrooj/hydrojudge",
     "bin": "bin/hydrojudge.js",
-    "version": "3.3.0",
+    "version": "3.3.1",
     "main": "package.json",
     "author": "undefined <i@undefined.moe>",
     "repository": "https://github.com/hydro-dev/Hydro.git",
     "dependencies": {
         "@hydrooj/utils": "workspace:*",
         "cac": "^6.7.14",
-        "mongodb": "^5.9.2",
         "p-queue": "^8.0.1",
         "schemastery": "^3.14.3",
         "semver": "^7.5.4",

packages/hydrojudge/src/hosts/builtin.ts

@@ -25,8 +25,8 @@ const session = {
     getNext(t: Context) {
         return async (data: Partial<JudgeResultBody>) => {
             logger.debug('Next: %o', data);
-            data.rid = t.rid as any;
+            data.rid = new ObjectId(t.rid);
-            const rdoc = await RecordModel.get(new ObjectId(t.rid));
+            const rdoc = await RecordModel.get(data.rid);
             if (data.case) data.case.message ||= '';
             next({ ...data, domainId: rdoc.domainId });
         };
@@ -34,8 +34,8 @@ const session = {
     getEnd(t: Context) {
         return async (data: Partial<JudgeResultBody>) => {
             data.key = 'end';
-            data.rid = t.rid as any;
+            data.rid = new ObjectId(t.rid);
-            const rdoc = await RecordModel.get(new ObjectId(t.rid));
+            const rdoc = await RecordModel.get(data.rid);
             logger.info('End: status=%d score=%d time=%dms memory=%dkb', data.status, data.score, data.time, data.memory);
             end({ ...data, domainId: rdoc.domainId });
         };

packages/hydrojudge/src/hosts/hydro.ts

@@ -1,6 +1,5 @@
 /* eslint-disable no-await-in-loop */
 import path from 'path';
-import { ObjectId } from 'mongodb';
 import PQueue from 'p-queue';
 import superagent from 'superagent';
 import WebSocket from 'ws';
@@ -150,8 +149,7 @@ export default class Hydro implements Session {
         return null;
     }
 
-    send(rid: string | ObjectId, key: 'next' | 'end', data: Partial<JudgeResultBody>) {
+    send(rid: string, key: 'next' | 'end', data: Partial<JudgeResultBody>) {
-        data.rid = new ObjectId(rid);
         data.key = key;
         if (data.case && typeof data.case.message === 'string') data.case.message = removeNixPath(data.case.message);
         if (typeof data.message === 'string') data.message = removeNixPath(data.message);

packages/hydrooj/package.json

@@ -1,6 +1,6 @@
 {
     "name": "hydrooj",
-    "version": "4.10.5",
+    "version": "4.10.6",
     "bin": "bin/hydrooj.js",
     "main": "src/plugin-api",
     "module": "src/plugin-api",

packages/hydrooj/src/handler/judge.ts

@@ -72,7 +72,7 @@ function processPayload(body: Partial<JudgeResultBody>) {
 }
 
 export async function next(body: Partial<JudgeResultBody>) {
-    body.rid = new ObjectId(body.rid);
+    body.rid = new ObjectId(body.rid); // TODO remove this
     const {
         $set, $push, $unset, $inc,
     } = processPayload(body);
@@ -141,7 +141,7 @@ export async function postJudge(rdoc: RecordDoc) {
 }
 
 export async function end(body: Partial<JudgeResultBody>) {
-    body.rid = new ObjectId(body.rid);
+    body.rid = new ObjectId(body.rid); // TODO remove this
     const { $set, $push } = processPayload(body);
     const $unset: any = { progress: '' };
     $set.judgeAt = new Date();
@@ -269,7 +269,7 @@ export class JudgeConnectionHandler extends ConnectionHandler {
             logger[method]('%o', omit(msg, keys));
         }
         if (['next', 'end'].includes(msg.key)) {
-            const t = this.tasks[msg.key];
+            const t = this.tasks[msg.rid];
             if (!t) return;
             msg.domainId = t.domainId;
             if (msg.key === 'next') t.queue.add(() => next(msg));

packages/hydrooj/src/model/task.ts

@@ -1,6 +1,6 @@
 import { hostname } from 'os';
 import cac from 'cac';
-import { Filter, ObjectId } from 'mongodb';
+import { BSON, Filter, ObjectId } from 'mongodb';
 import { nanoid } from 'nanoid';
 import { sleep } from '@hydrooj/utils/lib/utils';
 import { Context } from '../context';
@@ -141,7 +141,7 @@ export async function apply(ctx: Context) {
         collEvent.insertOne({
             ack: [id],
             event,
-            payload: JSON.stringify(payload),
+            payload: BSON.EJSON.stringify(payload),
             expire: new Date(Date.now() + 10000),
         });
     });
@@ -149,8 +149,8 @@ export async function apply(ctx: Context) {
     if (process.env.NODE_APP_INSTANCE !== '0') return;
     const stream = collEvent.watch();
     const handleEvent = async (doc: EventDoc) => {
-        const payload = JSON.parse(doc.payload);
+        process.send?.({ type: 'hydro:broadcast', data: doc });
-        process.send?.({ type: 'hydro:broadcast', data: { event: doc.event, payload } });
+        const payload = BSON.EJSON.parse(doc.payload);
         await (bus.parallel as any)(doc.event, ...payload);
     };
     stream.on('change', async (change) => {

packages/hydrooj/src/service/bus.ts

@@ -1,6 +1,6 @@
 /* eslint-disable no-await-in-loop */
-import type {
+import {
-    Db, Filter, ObjectId, OnlyFieldsOfType,
+    BSON, Db, Filter, ObjectId, OnlyFieldsOfType,
 } from 'mongodb';
 import pm2 from '@hydrooj/utils/lib/locate-pm2';
 import type { ProblemSolutionHandler } from '../handler/problem';
@@ -111,10 +111,10 @@ try {
     pm2.launchBus((err, bus) => {
         if (err) throw new Error();
         bus.on('hydro:broadcast', (packet) => {
-            (app.parallel as any)(packet.data.event, ...packet.data.payload);
+            (app.parallel as any)(packet.data.event, ...BSON.EJSON.parse(packet.data.payload));
         });
         app.on('bus/broadcast', (event, payload) => {
-            process.send({ type: 'hydro:broadcast', data: { event, payload } });
+            process.send({ type: 'hydro:broadcast', data: { event, payload: BSON.EJSON.stringify(payload) } });
         });
         console.debug('Using pm2 event bus');
     });

packages/prom-client/index.ts

@@ -50,7 +50,7 @@ export function apply(ctx: Context) {
         const [gateway, name, pass] = SystemModel.getMany(['prom-client.gateway', 'prom-client.name', 'prom-client.password']);
         if (gateway) {
             const prefix = gateway.endsWith('/') ? gateway : `${gateway}/`;
-            const endpoint = `${prefix}metrics/job/hydro-web/instance/${encodeURIComponent(hostname())}:${process.env.HYDRO_INSTANCE}`;
+            const endpoint = `${prefix}metrics/job/hydro-web/instance/${encodeURIComponent(hostname())}:${process.env.NODE_APP_INSTANCE}`;
             let req = superagent.post(endpoint);
             if (name) req = req.set('Authorization', `Basic ${Buffer.from(`${name}:${pass}`).toString('base64')}`);
             await req.send(await registry.metrics());

packages/prom-client/package.json

@@ -1,6 +1,6 @@
 {
     "name": "@hydrooj/prom-client",
-    "version": "0.1.6",
+    "version": "0.1.7",
     "main": "index.ts",
     "dependencies": {
         "prom-client": "^15.1.0"