Make MockEventQueue run jobs inline; drop test-suite leak workarounds

packages/backend/package.json

@@ -5,7 +5,7 @@
   "main": "index.js",
   "scripts": {
     "pretest": "npm run generate:openapi",
-    "test": "jest --forceExit --detectOpenHandles",
+    "test": "jest",
     "start": "npm run build && node ./build/src/index.js",
     "predev": "npm run generate:openapi",
     "dev": "node verifyShared.js && tsx watch ./src",

packages/backend/src/Services/EventQueue/MockEventQueue.ts

@@ -1,6 +1,5 @@
 import { randomUUID } from "crypto";
-import { ILoggingContext } from "../Logging/ILogger";
-import { EventHandler, IEventQueue, JobInsert } from "./IEventQueue";
+import { EventHandler, IEventQueue } from "./IEventQueue";
 import { QueueName } from "./QueueName";
 
 type Job = {
@@ -9,19 +8,24 @@ type Job = {
     data: object
 }
 
+/**
+ * In-memory stand-in for PGBossEventQueue. pg-boss keeps its jobs in Postgres,
+ * so the backend unit tests — which run entirely against in-memory fakes — need
+ * a substitute rather than the real thing.
+ *
+ * Jobs run inline: publish() does not resolve until its handlers have finished.
+ * That keeps tests deterministic and leaves nothing scheduled once a test file
+ * ends. When a test needs to observe the state *between* "job enqueued" and
+ * "job handled" — e.g. to check that a response is sent without waiting on the
+ * queue — use pause() / resume() rather than a sleep.
+ */
 export class MockEventQueue implements IEventQueue {
 
-    private _delay = 1000;
-
     private _handlers:Map<QueueName,EventHandler> = new Map();
     private _pendingJobs:Array<Job> = [];
-    private _working:boolean = false;
+    private _draining:boolean = false;
     private _paused:boolean = false;
 
-
-    constructor(){
-    }
-
     public subscribe(queue:QueueName, handler:EventHandler):void {
         if (this._handlers.has(queue)){
             console.error("ERROR: already have handler for queue "+queue);
@@ -30,54 +34,46 @@ export class MockEventQueue implements IEventQueue {
     }
 
     public async publish(queue:QueueName, data:object):Promise<string> {
-        var j = {
+        const job = {
             queue: queue,
             data: data,
             id: randomUUID()
         }
-        this._pendingJobs.push(j);
-        this.triggerJobs();
-        return j.id;
+        this._pendingJobs.push(job);
+        await this.drain();
+        return job.id;
     }
 
     public async publishBatch(queue:QueueName, data:object[]):Promise<object> {
-        var j = data.map(d => ({
+        const jobs = data.map(d => ({
             queue: queue,
             data: d,
             id: randomUUID()
         }))
-        this._pendingJobs.push(...j);
-        this.triggerJobs();
-        return j;
-    }
-
-    private async triggerJobs(){
-        if (this._working){
-            return;
-        }
-        if (this._paused){
-            return
-        }
-        this._working = true;
-        await new Promise(r => setTimeout(r, this._delay));
-        await this.processNextJob();
+        this._pendingJobs.push(...jobs);
+        await this.drain();
+        return jobs;
     }
 
-    private async processNextJob(){
-        var j = this._pendingJobs.shift();
-        if (!j){
-            console.info("Event queue empty");
-            this._working = false;
+    /**
+     * Runs every pending job to completion. Does nothing while paused, and is
+     * re-entrant safe: a handler that publishes another job leaves it for the
+     * loop already running rather than starting a nested one.
+     */
+    public async drain():Promise<void> {
+        if (this._paused || this._draining){
             return;
         }
+        this._draining = true;
         try {
-            console.info("MEQ: Processing job: " + JSON.stringify(j));
-            await this.doJob(j);
-        } catch (e:any) {
-            console.info("MEQ: Exception handling job: " + JSON.stringify(j));
+            var job = this._pendingJobs.shift();
+            while (job){
+                await this.doJob(job);
+                job = this._pendingJobs.shift();
+            }
+        } finally {
+            this._draining = false;
         }
-        this._working = false;
-        this.triggerJobs();
     }
 
     private async doJob(job:Job){
@@ -86,16 +82,26 @@ export class MockEventQueue implements IEventQueue {
             console.info("ERROR: no handler for queue "+job.queue);
             return;
         }
-        await h(job);
+        try {
+            await h(job);
+        } catch (e:any) {
+            // pg-boss retries a failed job; the mock just reports it, so that a
+            // broken handler shows up in the test output instead of vanishing.
+            console.info(`MEQ: Exception handling job ${job.id} on ${job.queue}: ${e?.stack ?? e}`);
+        }
     }
 
     public pause(){
         this._paused = true;
     }
 
-    public resume(){
+    public async resume(){
         this._paused = false;
-        this.triggerJobs();
+        await this.drain();
+    }
+
+    public pendingJobCount():number {
+        return this._pendingJobs.length;
     }
 
     public async clearStorage():Promise<void> {
@@ -105,12 +111,4 @@ export class MockEventQueue implements IEventQueue {
     public async debugInfo():Promise<string> {
         return "MEQ Debug:  " + JSON.stringify(this._pendingJobs);
     }
-
-    public async waitUntilJobsFinished():Promise<void> {
-        while(this._pendingJobs.length > 0){
-            await new Promise(r => setTimeout(r, this._delay));
-        }
-    }
-
-
-}
+}

packages/backend/src/test/anonymizedBallots.test.ts

@@ -9,10 +9,6 @@ import { TestHelper } from "./TestHelper";
 
 const th = new TestHelper();
 
-// The mock event queue processes ballots asynchronously with a 1s delay.
-// We must wait for it to flush before reading ballots back.
-const waitForQueue = async () => (await th.eventQueue).waitUntilJobsFinished();
-
 afterEach(() => {
     jest.clearAllMocks();
     th.afterEach();
@@ -70,8 +66,6 @@ describe("Anonymized ballots endpoint", () => {
     });
 
     test("Returns all submitted ballots, anonymized", async () => {
-        await waitForQueue();
-
         const res = await th.getRequest(
             `/API/Election/${election.election_id}/anonymizedBallots`,
             testInputs.user1token,

packages/backend/src/test/ballotReceiptEmail.test.ts

@@ -0,0 +1,103 @@
+require("dotenv").config();
+
+import { Election } from "@equal-vote/star-vote-shared/domain_model/Election";
+import { NewBallot } from "@equal-vote/star-vote-shared/domain_model/Ballot";
+import { Race } from "@equal-vote/star-vote-shared/domain_model/Race";
+import { ElectionSettings } from "@equal-vote/star-vote-shared/domain_model/ElectionSettings";
+import { MockEventQueue } from "../Services/EventQueue/MockEventQueue";
+import testInputs from "./testInputs";
+import { TestHelper } from "./TestHelper";
+
+const th = new TestHelper();
+
+afterEach(() => {
+    jest.clearAllMocks();
+    th.afterEach();
+});
+
+const ReceiptElection: Election = {
+    election_id: "0",
+    title: 'Receipt Election',
+    state: 'open',
+    frontend_url: '',
+    owner_id: 'Alice1234',
+    races: [
+        {
+            race_id: 'race0',
+            title: 'Best Leader',
+            num_winners: 1,
+            voting_method: 'STAR',
+            candidates: [
+                { candidate_id: '0', candidate_name: 'Alice' },
+                { candidate_id: '1', candidate_name: 'Bob' },
+            ],
+        },
+    ] as Race[],
+    settings: {
+        voter_access: 'open',
+        voter_authentication: {},
+        public_results: true,
+    } as ElectionSettings,
+} as Election;
+
+const makeBallot = (election_id: string): NewBallot => ({
+    election_id,
+    votes: [{
+        race_id: 'race0',
+        scores: [
+            { candidate_id: '0', score: 5 },
+            { candidate_id: '1', score: 0 },
+        ],
+    }],
+} as NewBallot);
+
+// Covers handleCastVoteEvent, the one queue consumer the suite exercises.
+// The receipt matters because castVoteController deliberately scrubs ballot_id
+// from the HTTP response, so the email is the only channel that carries it back
+// to the voter.
+describe("Ballot receipt email", () => {
+    var election: Election;
+    var eventQueue: MockEventQueue;
+
+    beforeAll(async () => {
+        eventQueue = await th.eventQueue;
+        const response = await th.createElection(ReceiptElection, testInputs.user1token);
+        expect(response.statusCode).toBe(200);
+        election = response.election;
+    });
+
+    beforeEach(() => {
+        th.emailService.clear();
+    });
+
+    test("Submitting a ballot mails the voter a receipt carrying the ballot_id", async () => {
+        const res = await th.submitBallot(election.election_id, makeBallot(election.election_id), testInputs.user1token);
+        expect(res.statusCode).toBe(200);
+        // the response withholds the ballot_id on purpose
+        expect(res.body.ballot.ballot_id).toBeUndefined();
+
+        const sent = th.emailService.sentEmails;
+        expect(sent).toHaveLength(1);
+        expect(sent[0].to).toBe('Alice@email.com');
+        expect(sent[0].subject).toBe(`Ballot Receipt For ${election.title}`);
+        expect(sent[0].text).toMatch(new RegExp(`/${election.election_id}/ballot/b-`));
+        th.testComplete();
+    });
+
+    test("The receipt is sent off the queue, not inline with the response", async () => {
+        eventQueue.pause();
+        try {
+            const res = await th.submitBallot(election.election_id, makeBallot(election.election_id), testInputs.user1token);
+            expect(res.statusCode).toBe(200);
+            // the voter is told their ballot was accepted before the email goes out
+            expect(th.emailService.sentEmails).toHaveLength(0);
+            expect(eventQueue.pendingJobCount()).toBe(1);
+        } finally {
+            await eventQueue.resume();
+        }
+
+        expect(th.emailService.sentEmails).toHaveLength(1);
+        expect(eventQueue.pendingJobCount()).toBe(0);
+        th.testComplete();
+    });
+});

packages/backend/src/test/emailRoll.test.ts

@@ -1,12 +1,10 @@
 require("dotenv").config();
 const request = require("supertest");
-import makeApp from "../app";
-import { MockEventQueue } from "../Services/EventQueue/MockEventQueue";
 import { TestHelper } from "./TestHelper";
 import testInputs from "./testInputs";
 
-const app = makeApp();
 const th = new TestHelper();
+const app = th.expressApp;
 
 afterEach(() => {
     jest.clearAllMocks();
@@ -85,8 +83,6 @@ describe("Email List Voter Auth", () => {
             aliceVoterId
         );
         expect(response.statusCode).toBe(200);
-        const eventQueue: MockEventQueue = await th.eventQueue;
-        await eventQueue.waitUntilJobsFinished();
         th.testComplete();
     });
 

packages/backend/src/test/finalizeElection.test.ts

@@ -41,9 +41,6 @@ describe("Finalize Election", () => {
             electionId,
             testInputs.user1token
         );
-        //Wait a few seconds for job queue to process emails, I imagine there's a better way to do this.
-        await new Promise(resolve => setTimeout(resolve, 4000));
-
         expect(response.statusCode).toBe(200);
         th.testComplete();
     });

packages/backend/src/test/idRoll.test.ts

@@ -1,7 +1,6 @@
 require('dotenv').config();
 const request = require('supertest');
 import { ElectionRoll, ElectionRollState } from '@equal-vote/star-vote-shared/domain_model/ElectionRoll';
-import { MockEventQueue } from '../Services/EventQueue/MockEventQueue';
 import { TestHelper } from './TestHelper';
 import testInputs from './testInputs';
 
@@ -46,9 +45,6 @@ describe("ID Roll", () => {
     test("Authorized voter submits ballot", async () => {
         const response = await th.submitBallotWithId(ID, testInputs.Ballot2, testInputs.user1token, testInputs.IDRoll[0].voter_id);
         expect(response.statusCode).toBe(200)
-
-        const eventQueue:MockEventQueue = await th.eventQueue;
-        await eventQueue.waitUntilJobsFinished();
         th.testComplete();
     })
     test("Get voter auth, is authorized and has voted", async () => {

packages/backend/src/test/writeIns.test.ts

@@ -11,9 +11,6 @@ import { TestHelper } from "./TestHelper";
 const th = new TestHelper();
 type TiebreakCandidate = candidate & { tieBreakOrder: number };
 
-// The mock event queue processes ballots asynchronously with a 1s delay.
-// We must wait for it to flush before reading ballots back.
-const waitForQueue = async () => (await th.eventQueue).waitUntilJobsFinished();
 
 afterEach(() => {
     jest.clearAllMocks();
@@ -207,8 +204,6 @@ describe("Write-In Candidates", () => {
     });
 
     test("Get write-in names", async () => {
-        await waitForQueue();
-
         const res = await th.getRequest(
             `/API/Election/${election.election_id}/getWriteIns`,
             testInputs.user1token,
@@ -250,8 +245,6 @@ describe("Write-In Candidates", () => {
     });
 
     test("Get results includes approved write-in, excludes unapproved", async () => {
-        await waitForQueue();
-
         const res = await th.getRequest(
             `/API/ElectionResult/${election.election_id}`,
             testInputs.user1token,