final fix of worker

lightning-sagar · lightning-sagar · commit 5dbbe694cdf9 · 2025-09-16T17:16:26.000+05:30
diff --git a/example/src/index.js b/example/src/index.js
@@ -15,13 +15,13 @@ const main = async () => {
     language: "cpp",
     timeout: "2",
     sizeout: "64",
-  });   
-  console.log("this is ans ",result);
+  });
+  console.log("this is ans ", result);
 };
 
 main();
 
-function getPath(){
+function getPath() {
   const code = `
   #include<iostream>
   #include<vector>
@@ -49,11 +49,11 @@ function getPath(){
   return tmpPath;
 }
 
-// creating ~200 test case 
+// creating ~200 test case
 function createTestcase() {
   const inputs = [];
   const outputs = [];
-  for (let i = 1; i <= 15; i++) {
+  for (let i = 1; i <= 305; i++) {
     const arr = Array.from({ length: i }, (_, idx) => idx + 1).join(" ");
     inputs.push(`${i} ${arr}`);
     const sum = (i * (i + 1)) / 2;
@@ -63,5 +63,3 @@ function createTestcase() {
   const outputStr = outputs.join(" ### ");
   return { inputStr, outputStr };
 }
-
-
diff --git a/judge-image-for-worker/workers/index.js b/judge-image-for-worker/workers/index.js
@@ -33,7 +33,6 @@ app.use(express.json());
 
 const __filename = fileURLToPath(import.meta.url);
 const __dirname = path.dirname(__filename);
-const WORKER_FIELD = process.env.WORKER_FIELD;
 const port = process.env.PORT;
 const redis_server = await connectredis();
 
@@ -145,7 +144,7 @@ function runTestcase(
 async function processJob(ques_name, code, language, testcases) {
   const extension =
     language === "cpp" ? "cpp" : language === "java" ? "java" : "py";
-  const fileName = `${ques_name}_${WORKER_FIELD}.${extension}`;
+  const fileName = `${ques_name}.${extension}`;
   const filePath = path.join(__dirname, fileName);
   const execPath =
     language === "java"
@@ -170,24 +169,25 @@ async function processJob(ques_name, code, language, testcases) {
       )
     );
     console.log(results, "this is results");
+    // Store job result
     await redis_server.setEx(
-      `job:${ques_name}:worker:${WORKER_FIELD}`,
+      `job:${ques_name}:result`,
       30,
       JSON.stringify(results)
     );
     await redis_server.hSet(`job:${ques_name}:status`, {
-      [WORKER_FIELD]: "completed",
+      state: "completed",
     });
     await redis_server.expire(`job:${ques_name}:status`, 30);
   } catch (err) {
     console.error("Error during job processing:", err);
     await redis_server.setEx(
-      `job:${ques_name}:worker:${WORKER_FIELD}`,
+      `job:${ques_name}:result`,
       30,
       JSON.stringify([{ error: err.toString() }])
     );
     await redis_server.hSet(`job:${ques_name}:status`, {
-      [WORKER_FIELD]: "completed",
+      state: "failed",
     });
   } finally {
     try {
@@ -204,17 +204,44 @@ async function processJob(ques_name, code, language, testcases) {
 async function pollForJobs() {
   while (true) {
     try {
-      const { element: ques_name } = await redis_server.brPop("job_queue", 0);
-      console.log(`Got job: ${ques_name}`);
+      // Fetch a job name
+      const result = await redis_server.brPop("job_queue", 0);
 
+      if (!result) {
+        console.warn("[Worker] BRPOP returned empty");
+        continue;
+      }
+
+      const ques_name = result.element;
+      console.log(`[Worker] Got job: ${ques_name}`);
+
+      // Fetch code & language (from Redis Hash or DB)
       const code = await redis_server.hGet(ques_name, "code");
       const language = await redis_server.hGet(ques_name, "language");
-      const data_testcases = await redis_server.hGet(ques_name, WORKER_FIELD);
-      if (!data_testcases) continue;
-      const testcases = JSON.parse(data_testcases);
+
+      // Get next batch of test cases
+      const batchData = await redis_server.lPop(`testcase_queue:${ques_name}`);
+      if (!batchData) {
+        console.warn(`[Worker] No batch found for ${ques_name}`);
+        continue;
+      }
+
+      const testcases = JSON.parse(batchData);
+
+      // Process the job
       await processJob(ques_name, code, language, testcases);
+
+      // Update status
+      await redis_server.hIncrBy(
+        `job:${ques_name}:status`,
+        "completedBatches",
+        1
+      );
+      console.log(
+        `[Worker] Completed one batch of ${testcases.length} for ${ques_name}`
+      );
     } catch (err) {
-      console.error("Error while polling job:", err);
+      console.error("[Worker] Error while polling job:", err);
     }
   }
 }
@@ -226,5 +253,5 @@ app.get("/ping", (req, res) => {
 });
 
 app.listen(port, () => {
-  console.log(`${WORKER_FIELD} running at port ${port}`);
+  console.log(`running at port ${port}`);
 });
diff --git a/judge-image-for-worker/workers/run.sh b/judge-image-for-worker/workers/run.sh
@@ -1,6 +1,6 @@
-PORT=3001 WORKER_FIELD=worker_0 nodemon index.js &
-PORT=3002 WORKER_FIELD=worker_1 nodemon index.js &
-PORT=3003 WORKER_FIELD=worker_2 nodemon index.js &
+PORT=3001   nodemon index.js &
+PORT=3002   nodemon index.js &
+PORT=3003   nodemon index.js &
 
 
 
diff --git a/lib/src/run.js b/lib/src/run.js
@@ -2,22 +2,8 @@ import fs from "fs";
 import "dotenv/config";
 import { connectredis } from "./redis.js";
 
-const worker_running = [
-  "https://workers-judge.onrender.com/",
-  "https://workers-judge-1.onrender.com/",
-  "https://workers-judge-2.onrender.com/",
-];
-
 const redis = await connectredis();
 
-/**
- * @param {string} codePath - Absolute path to uploaded code file
- * @param {string} ques_name - Unique job/question ID
- * @param {string} input - Raw input test cases (separated by ###)
- * @param {string} output - Raw expected outputs (separated by ###)
- * @param {string} timeout - Max timeout in seconds
- * @param {string} sizeout - Max output size in KB
- */
 export async function judge({
   codePath,
   language,
@@ -27,79 +13,108 @@ export async function judge({
   timeout,
   sizeout,
 }) {
-  if (!codePath || !ques_name || !timeout || !sizeout || !language)
-    throw new Error("codePath and ques_name are required");
+  if (!codePath || !ques_name || !timeout || !sizeout || !language) {
+    throw new Error("codePath, ques_name, timeout, sizeout, and language are required");
+  }
 
   let code;
   try {
+    // 1. Read submitted code
     code = fs.readFileSync(codePath, "utf-8");
 
-    if (
-      code.includes("fopen") ||
-      code.includes("system") ||
-      code.includes("fork")
-    ) {
-      throw new Error("Potentially dangerous code detected.");
+    // 2. Validate code for dangerous functions
+    const dangerous = ["fopen", "system", "fork", "exec", "eval", "import os", "subprocess"];
+    for (const keyword of dangerous) {
+      if (code.includes(keyword)) {
+        throw new Error(`Potentially dangerous code detected: ${keyword}`);
+      }
     }
 
-    const inputParts = input
-      .split("###")
-      .map((s) => s.trim())
-      .filter(Boolean);
-    const outputParts = output
-      .split("###")
-      .map((s) => s.trim())
-      .filter(Boolean);
-
-    const testcases = inputParts.map((input, i) => ({
-      input,
+    // 3. Prepare test cases
+    const inputParts = input.split("###").map(s => s.trim()).filter(Boolean);
+    const outputParts = output.split("###").map(s => s.trim()).filter(Boolean);
+
+    if (inputParts.length !== outputParts.length) {
+      throw new Error(
+        `Mismatch between input (${inputParts.length}) and output (${outputParts.length})`
+      );
+    }
+
+    const testcases = inputParts.map((inp, i) => ({
+      input: inp,
       expected_output: outputParts[i],
       correct: null,
-      timeout:timeout || 2.5,
+      timeout: timeout || 2.5,
       sizeout,
       result: null,
     }));
 
-    const workerCount = worker_running.length;
-    const workerTaskMap = {};
-
-    if (testcases.length < 3) {
-      workerTaskMap['worker_0'] = testcases;
-    } else {
-      testcases.forEach((tc, i) => {
-        const workerId = `worker_${i % workerCount}`;
-        if (!workerTaskMap[workerId]) workerTaskMap[workerId] = [];
-        workerTaskMap[workerId].push(tc);
-      });
-    }
-
-    const redisPayload = {
+    // 4. Clean old data for this job
+    const pipeline = redis.multi();
+    pipeline.del(`testcase_queue:${ques_name}`);
+    pipeline.del(`results_queue:${ques_name}`);
+    pipeline.del(`job:${ques_name}:status`);
+    pipeline.del(ques_name);
+    pipeline.del(`job:${ques_name}:result`);
+    await pipeline.exec();
+
+    // 5. Save code + metadata
+    await redis.hSet(ques_name, {
       code,
       language,
-      ...Object.fromEntries(
-        Object.entries(workerTaskMap).map(([k, v]) => [k, JSON.stringify(v)])
-      ),
-    };
-
-    await redis.hSet(ques_name, redisPayload);
-    const assignedWorkers = Object.keys(workerTaskMap);
+      totalTestCases: testcases.length.toString(),
+      timeout: timeout.toString(),
+      sizeout: sizeout.toString(),
+      createdTime: Date.now().toString(),
+    });
+
+    // 6. Split testcases into batches of 100
+    const BATCH_SIZE = 100;
+    const batches = [];
+    for (let i = 0; i < testcases.length; i += BATCH_SIZE) {
+      batches.push(testcases.slice(i, i + BATCH_SIZE));
+    }
 
-    await Promise.all(
-      assignedWorkers.map(() => redis.lPush("job_queue", ques_name))
+    console.log(
+      `Splitting into ${batches.length} batches of up to ${BATCH_SIZE} test cases each`
     );
 
+    // 7. Push each batch to Redis queue
+    for (const batch of batches) {
+      await redis.rPush(`testcase_queue:${ques_name}`, JSON.stringify(batch));
+      await redis.lPush("job_queue", ques_name);  
+    }
+
+    // 8. Initialize job status
+    await redis.hSet(`job:${ques_name}:status`, {
+      status: "queued",
+      totalBatches: batches.length.toString(),
+      completedBatches: "0",
+      createdTime: Date.now().toString(),
+    });
+
+    console.log(`[DEBUG] Job ${ques_name} queued with ${batches.length} batches`);
+
+    // 9. Wait for job completion
     const waitUntilCompleted = async () => {
-      const POLL_INTERVAL = 500;
-      const MAX_ATTEMPTS = 60;
+      const POLL_INTERVAL = 500; // ms
+      const MAX_ATTEMPTS = 60; // 30 seconds
       let attempts = 0;
 
       while (attempts < MAX_ATTEMPTS) {
         const status = await redis.hGetAll(`job:${ques_name}:status`);
-        const completed = Object.keys(status).filter(
-          (k) => status[k] === "completed"
+
+        const totalBatches = parseInt(status.totalBatches || "0", 10);
+        const completedBatches = parseInt(status.completedBatches || "0", 10);
+
+        console.log(
+          `[DEBUG] Polling... Completed ${completedBatches}/${totalBatches} batches`
         );
-        if (assignedWorkers.every((worker) => completed.includes(worker)))
+
+        if (completedBatches === totalBatches && totalBatches > 0) {
           return true;
+        }
+
         await new Promise((res) => setTimeout(res, POLL_INTERVAL));
         attempts++;
       }
@@ -108,26 +123,36 @@ export async function judge({
     };
 
     const completed = await waitUntilCompleted();
-
     if (!completed) {
       throw new Error("Timeout waiting for workers to finish");
     }
 
-    const results = [];
-    for (const workerId of assignedWorkers) {
-      const data = await redis.get(`job:${ques_name}:worker:${workerId}`);
-      if (data) results.push(...JSON.parse(data));
+    const finalData = await redis.get(`job:${ques_name}:result`);
+    let results = [];
+    if (finalData) {
+      try {
+        results = JSON.parse(finalData);
+      } catch (err) {
+        console.error("Error parsing job result JSON:", err);
+      }
+    } else {
+      console.warn(`[WARN] No results found for job ${ques_name}`);
     }
-
     return {
+      status: "completed",
       jobId: ques_name,
+      totalTestCases: testcases.length,
+      totalBatches: batches.length,
       results,
     };
+
   } catch (err) {
-    console.log(err);
+    console.error("Judge error:", err.message);
+    throw err;
   } finally {
-    if (fs.existsSync(codePath)) {
+    if (codePath && fs.existsSync(codePath)) {
       fs.unlinkSync(codePath);
+      console.log(`Cleaned up uploaded code file: ${codePath}`);
     }
   }
-}
+}
