Stream ballots into a compact per-race projection for tabulation

packages/backend/src/Controllers/Ballot/castVoteController.ts

@@ -16,8 +16,7 @@ import { io } from "../../socketHandler";
 import { Server } from "socket.io";
 import { expectPermission } from "../controllerUtils";
 import { permissions } from "@equal-vote/star-vote-shared/domain_model/permissions";
-import { OrderedVote } from "@equal-vote/star-vote-shared/domain_model/Vote";
-import { Score } from "@equal-vote/star-vote-shared/domain_model/Score";
+import { OrderedVoteFormatError, orderedVotesToVotes } from "@equal-vote/star-vote-shared/domain_model/OrderedVoteCodec";
 import { makeUniqueID, ID_LENGTHS, ID_PREFIXES } from "@equal-vote/star-vote-shared/utils/makeID";
 
 const ElectionsModel = ServiceLocator.electionsDb();
@@ -124,28 +123,15 @@ async function makeBallotEvent(req: IElectionRequest, targetElection: Election,
 }
 
 const mapOrderedNewBallot = (ballot: OrderedNewBallot, raceOrder: RaceCandidateOrder[]): NewBallot => {
-    let subBallot: any = {...ballot};
-    delete subBallot.orderedVotes;
-    if(ballot.orderedVotes.length != raceOrder.length){
-        throw new BadRequest(`Ballot contains different number of races than race_order: ${ballot.orderedVotes.length} != ${raceOrder.length}`)
-    }
-    return {
-        ...subBallot,
-        votes: ballot.orderedVotes.map((vote: OrderedVote, i) => {
-            // +2 accounts for overvote_rank and has_duplicate_rank
-            if(vote.length != raceOrder[i].candidate_id_order.length+2){
-                throw new BadRequest(`Race ${i} contains different number of candidates than race_order: ${vote.length} != ${raceOrder[i].candidate_id_order.length+2}`)
-            }
-            return {
-                race_id: raceOrder[i].race_id,
-                scores: vote.slice(0, -2).map((s, j) => ({
-                    candidate_id: raceOrder[i].candidate_id_order[j],
-                    score: s
-                } as Score)),
-                overvote_rank: vote.at(-2),
-                has_duplicate_rank: vote.at(-1) == 1,
-            }
-        })
+    const {orderedVotes, ...subBallot} = ballot;
+    try {
+        return {
+            ...subBallot,
+            votes: orderedVotesToVotes(orderedVotes, raceOrder)
+        }
+    } catch (err: any) {
+        if (err instanceof OrderedVoteFormatError) throw new BadRequest(err.message);
+        throw err;
     }
 }
 async function uploadBallotsController(req: IElectionRequest, res: Response, next: NextFunction) {

packages/backend/src/Controllers/Election/getElectionResultsController.ts

@@ -1,16 +1,13 @@
 import ServiceLocator from "../../ServiceLocator";
 import Logger from "../../Services/Logging/Logger";
-import { BadRequest, Forbidden } from "@curveball/http-errors";
-import { Ballot } from '@equal-vote/star-vote-shared/domain_model/Ballot';
+import { Forbidden } from "@curveball/http-errors";
 import { expectPermission } from "../controllerUtils";
 import { permissions } from '@equal-vote/star-vote-shared/domain_model/permissions';
 import { VotingMethods } from '../../Tabulators/VotingMethodSelecter';
 import { IElectionRequest } from "../../IRequest";
 import { Response, NextFunction } from 'express';
-import { ElectionResults, candidate, rawVote } from "@equal-vote/star-vote-shared/domain_model/ITabulators";
-import { makeWriteInCandidateId } from "@equal-vote/star-vote-shared/utils/makeID";
-import { Candidate } from "@equal-vote/star-vote-shared/domain_model/Candidate";
-import { trimLower } from "@equal-vote/star-vote-shared/domain_model/Util";
+import { ElectionResults, rawVote } from "@equal-vote/star-vote-shared/domain_model/ITabulators";
+import { projectBallots } from "../../Tabulators/BallotProjection";
 import shuffleCandidatesForRandomTiebreak from "../../Tabulators/shuffleCandidatesForRandomTiebreak";
 
 const BallotModel = ServiceLocator.ballotsDb();
@@ -30,87 +27,34 @@ const getElectionResults = async (req: IElectionRequest, res: Response, next: Ne
         expectPermission(req.user_auth.roles, permissions.canViewPreliminaryResults)
     }
 
-    const ballots = await BallotModel.getBallotsByElectionID(String(electionId), req);
+    // Stream ballots from a db cursor and project each one into the compact
+    // per-race representation, so the verbose rows are never all in memory at
+    // once (see Tabulators/BallotProjection).
+    const projections = await projectBallots(
+        election.races,
+        BallotModel.streamVotesByElectionID(String(electionId), req),
+        {
+            debug: (msg) => Logger.debug(req, msg),
+            warn: (msg) => Logger.warn(req, msg),
+        }
+    )
 
     let results: ElectionResults[] = []
     for (let race_index = 0; race_index < election.races.length; race_index++) {
         const race = election.races[race_index]
-        const useWriteIns = race.enable_write_in && race.write_in_candidates && race.write_in_candidates.length > 0
-        const writeInCandidates = useWriteIns && race.write_in_candidates ? race.write_in_candidates : []
-
-        // Build candidate list including approved write-in candidates
-        const candidates: candidate[] = race.candidates.map((c: Candidate, i) => ({
-            id: c.candidate_id,
-            name: c.candidate_name,
-            tieBreakOrder: -1,
-            votesPreferredOver: {},
-            winsAgainst: {}
-        }))
-
-        Logger.debug(req, `[WriteIn Debug] race=${race.race_id} useWriteIns=${useWriteIns} writeInCandidates=${JSON.stringify(writeInCandidates.map(wc => ({name: wc.candidate_name, approved: wc.approved, aliases: wc.aliases})))}`);
+        const projection = projections[race_index]
+        const candidates = projection.candidates
+        const numUnprocessedWriteIns = projection.numUnprocessedWriteIns
+        const numExcludedWriteIns = projection.numExcludedWriteIns
 
-        if (useWriteIns) {
-            writeInCandidates.forEach((wc, i) => {
-                if (wc.approved) {
-                    candidates.push({
-                        id: makeWriteInCandidateId(wc.candidate_name),
-                        name: wc.candidate_name,
-                        tieBreakOrder: -1,
-                        votesPreferredOver: {},
-                        winsAgainst: {}
-                    })
-                }
-            })
-        }
+        Logger.debug(req, `[WriteIn Debug] race=${race.race_id} useWriteIns=${projection.useWriteIns} writeInCandidates=${JSON.stringify(projection.writeInCandidates.map(wc => ({name: wc.candidate_name, approved: wc.approved, aliases: wc.aliases})))}`);
         Logger.debug(req, `[WriteIn Debug] candidates for tabulation: ${JSON.stringify(candidates.map(c => ({id: c.id, name: c.name})))}`);
 
-        const race_id = race.race_id
-        const cvr: rawVote[] = []
         const num_winners = race.num_winners
         const voting_method = race.voting_method
-        let numUnprocessedWriteIns = 0
-        let numExcludedWriteIns = 0
-
-        ballots.forEach((ballot: Ballot) => {
-            const vote = ballot.votes.find((vote) => vote.race_id === race_id)
-            if (vote) {
-                const marks: {[key: string]: number | null} = {}
-                vote.scores.forEach(score => {
-                    const isRegularCandidate = race.candidates.some((c: Candidate) => c.candidate_id === score.candidate_id)
-                    if (isRegularCandidate) {
-                        if (score.candidate_id in marks) {
-                            Logger.warn(req, `[Tabulation] Duplicate score for candidate "${score.candidate_id}" on same ballot, keeping first score`);
-                        } else {
-                            marks[score.candidate_id] = score.score
-                        }
-                    } else if (race.enable_write_in && score.write_in_name) {
-                        const write_in_name = score.write_in_name
-                        const writeInCandidate = writeInCandidates.find(wc => wc.aliases.includes(trimLower(write_in_name)))
-                        Logger.debug(req, `[WriteIn Debug] ballot write_in_name="${write_in_name}" matched=${!!writeInCandidate} approved=${writeInCandidate?.approved} matchedAliases=${JSON.stringify(writeInCandidate?.aliases)}`);
-                        if (!writeInCandidate) {
-                            numUnprocessedWriteIns += 1
-                            numExcludedWriteIns += 1
-                        } else if (writeInCandidate.approved) {
-                            const wcId = makeWriteInCandidateId(writeInCandidate.candidate_name)
-                            if (!(wcId in marks)) {
-                                marks[wcId] = score.score
-                            } else {
-                                Logger.warn(req, `[WriteIn] Duplicate write-in score for "${writeInCandidate.candidate_name}" on same ballot, keeping first score`);
-                            }
-                        } else {
-                            numExcludedWriteIns += 1
-                        }
-                    }
-                })
-                cvr.push({
-                    marks,
-                    overvote_rank: vote?.overvote_rank,
-                    has_duplicate_rank: vote?.has_duplicate_rank,
-                })
-            }
-        })
 
         if (candidates.length < 1) {
+            projection.release()
             results[race_index] = {
                 votingMethod: voting_method,
                 elected: [],
@@ -138,6 +82,10 @@ const getElectionResults = async (req: IElectionRequest, res: Response, next: Ne
             throw new Error(`Invalid Voting Method: ${voting_method}`)
         }
 
+        // Expand the compact store into the tabulator's input only now, and only
+        // for this race, so at most one race's verbose cvr is alive at a time.
+        const cvr: rawVote[] = projection.takeRawVotes()
+
         shuffleCandidatesForRandomTiebreak(election.create_date, candidates, cvr.length, race.race_id);
         const perm = candidates.map(candidate => candidate.id);
 

packages/backend/src/Models/Ballots.ts

@@ -3,7 +3,7 @@ import { Uid } from '@equal-vote/star-vote-shared/domain_model/Uid';
 import { ILoggingContext } from '../Services/Logging/ILogger';
 import Logger from '../Services/Logging/Logger';
 import { logSafeHash } from '../Services/Logging/logSafeHash';
-import { IBallotStore } from './IBallotStore';
+import { BallotVotes, IBallotStore } from './IBallotStore';
 import { Kysely, sql, Transaction } from 'kysely';
 import { Database } from './Database';
 import { InternalServerError } from '@curveball/http-errors';
@@ -135,6 +135,26 @@ export default class BallotsDB implements IBallotStore {
             .stream(500) as AsyncIterableIterator<Ballot>;
     }
 
+    // Tabulation only reads the marks, so select just the `votes` column and
+    // stream it from a cursor. Skipping ballot_id/history/timestamps and never
+    // holding the full result set is what keeps a large election's tabulation
+    // from inflating the heap (see issue #1425).
+    //
+    // The filter deliberately matches getBallotsByElectionID (head only, any
+    // status) rather than streamSubmittedBallotsByElectionID — narrowing it to
+    // submitted ballots here would silently change election results.
+    streamVotesByElectionID(election_id: string, ctx: ILoggingContext, db?: Kysely<Database> | Transaction<Database>): AsyncIterableIterator<BallotVotes> {
+        Logger.debug(ctx, `${tableName}.streamVotesByElectionID ${election_id}`);
+        const client = db || this._postgresClient;
+
+        return client
+            .selectFrom(tableName)
+            .select('votes')
+            .where('election_id', '=', election_id)
+            .where('head', '=', true)
+            .stream(500) as AsyncIterableIterator<BallotVotes>;
+    }
+
     getBallotByVoterID(voter_id: string, election_id: string, ctx: ILoggingContext, db?: Kysely<Database> | Transaction<Database>): Promise<Ballot | undefined> {
         Logger.debug(ctx, `${tableName}.getBallotByVoterID ${logSafeHash(voter_id)} ${election_id}`);
         const client = db || this._postgresClient;

packages/backend/src/Models/IBallotStore.ts

@@ -1,9 +1,15 @@
 import { Ballot } from "@equal-vote/star-vote-shared/domain_model/Ballot";
 import { Uid } from "@equal-vote/star-vote-shared/domain_model/Uid";
+import { Vote } from "@equal-vote/star-vote-shared/domain_model/Vote";
 import { ILoggingContext } from "../Services/Logging/ILogger";
 import { Kysely, Transaction } from 'kysely';
 import { Database } from './Database';
 
+/** The only column tabulation reads off a ballot row. */
+export interface BallotVotes {
+    votes: Vote[];
+}
+
 export interface IBallotStore {
     submitBallot: (ballot: Ballot, ctx: ILoggingContext, reason: string, db?: Kysely<Database> | Transaction<Database>) => Promise<Ballot>;
     updateBallot: (ballot: Ballot, ctx: ILoggingContext, reason: string, db?: Kysely<Database> | Transaction<Database>) => Promise<Ballot>;
@@ -12,6 +18,8 @@ export interface IBallotStore {
     getBallotsByElectionID: (election_id: string, ctx: ILoggingContext, db?: Kysely<Database> | Transaction<Database>) => Promise<Ballot[]>;
     // Streams head, submitted ballots in random order (see Ballots.ts for the anonymity rationale)
     streamSubmittedBallotsByElectionID: (election_id: string, ctx: ILoggingContext, db?: Kysely<Database> | Transaction<Database>) => AsyncIterableIterator<Ballot>;
+    // Streams just the votes column of every head ballot, for tabulation
+    streamVotesByElectionID: (election_id: string, ctx: ILoggingContext, db?: Kysely<Database> | Transaction<Database>) => AsyncIterableIterator<BallotVotes>;
     getBallotByVoterID: (voter_id: string, election_id: string, ctx: ILoggingContext, db?: Kysely<Database> | Transaction<Database>) => Promise<Ballot | undefined>;
     delete(ballot_id: Uid, ctx: ILoggingContext, reason: string, db?: Kysely<Database> | Transaction<Database>): Promise<boolean>;
     deleteAllBallotsForElectionID: (election_id: string, ctx: ILoggingContext, db?: Kysely<Database> | Transaction<Database>) => Promise<boolean>;

packages/backend/src/Models/__mocks__/Ballots.ts

@@ -1,7 +1,7 @@
 import { Ballot } from "@equal-vote/star-vote-shared/domain_model/Ballot";
 import { Uid } from "@equal-vote/star-vote-shared/domain_model/Uid";
 import { ILoggingContext } from "../../Services/Logging/ILogger";
-import { IBallotStore } from "../IBallotStore";
+import { BallotVotes, IBallotStore } from "../IBallotStore";
 
 export default class BallotsDB implements IBallotStore {
     ballots: Ballot[] = [];
@@ -47,6 +47,17 @@ export default class BallotsDB implements IBallotStore {
         }
     }
 
+    // mirrors getBallotsByElectionID's filter (see Ballots.ts) so tabulation
+    // sees exactly the same ballots as the old non-streaming path
+    async *streamVotesByElectionID(election_id: string, ctx:ILoggingContext): AsyncIterableIterator<BallotVotes> {
+        const ballots = this.ballots.filter(
+            (ballot) => ballot.election_id === election_id
+        );
+        for (const ballot of ballots) {
+            yield { votes: JSON.parse(JSON.stringify(ballot.votes)) };
+        }
+    }
+
     getBallotByVoterID(voter_id: string, election_id: string, ctx:ILoggingContext): Promise<Ballot | undefined> {
         const ballots = this.ballots.filter(
             (ballot) => ballot.user_id === voter_id