Fix unbounded queries crashing review queues (#160)

+1

client/src/graphql/generated.ts

··· 2119 2119 2120 2120 export type GQLManualReviewQueueJobsArgs = { 2121 2121 ids?: InputMaybe<ReadonlyArray<Scalars['ID']['input']>>; 2122 + limit?: InputMaybe<Scalars['Int']['input']>; 2122 2123 }; 2123 2124 2124 2125 export type GQLManualReviewQueueNameExistsError = GQLError & {

-19

package-lock.json

··· 830 830 } 831 831 } 832 832 }, 833 - "node_modules/@graphql-codegen/cli/node_modules/@types/node": { 834 - "version": "25.5.2", 835 - "resolved": "https://registry.npmjs.org/@types/node/-/node-25.5.2.tgz", 836 - "integrity": "sha512-tO4ZIRKNC+MDWV4qKVZe3Ql/woTnmHDr5JD8UI5hn2pwBrHEwOEMZK7WlNb5RKB6EoJ02gwmQS9OrjuFnZYdpg==", 837 - "license": "MIT", 838 - "optional": true, 839 - "peer": true, 840 - "dependencies": { 841 - "undici-types": "~7.18.0" 842 - } 843 - }, 844 833 "node_modules/@graphql-codegen/cli/node_modules/cosmiconfig": { 845 834 "version": "9.0.0", 846 835 "resolved": "https://registry.npmjs.org/cosmiconfig/-/cosmiconfig-9.0.0.tgz", ··· 5270 5259 "engines": { 5271 5260 "node": ">=0.10.0" 5272 5261 } 5273 - }, 5274 - "node_modules/undici-types": { 5275 - "version": "7.18.2", 5276 - "resolved": "https://registry.npmjs.org/undici-types/-/undici-types-7.18.2.tgz", 5277 - "integrity": "sha512-AsuCzffGHJybSaRrmr5eHr81mwJU3kjw6M+uprWvCXiNeN9SOGwQ3Jn8jb8m3Z6izVgknn1R0FTCEAP2QrLY/w==", 5278 - "license": "MIT", 5279 - "optional": true, 5280 - "peer": true 5281 5262 }, 5282 5263 "node_modules/universalify": { 5283 5264 "version": "2.0.0",

+30

scripts/load_test_reports.sh

··· 1 + #!/bin/bash 2 + set -e 3 + 4 + API_KEY="$1" 5 + COUNT="${2:-5000}" 6 + BASE_URL="${3:-http://localhost:3000}" 7 + 8 + if [ -z "$API_KEY" ]; then 9 + echo "Usage: bash load_test_reports.sh API_KEY [COUNT] [BASE_URL]" 10 + echo " COUNT defaults to 5000, BASE_URL defaults to http://localhost:3000" 11 + exit 1 12 + fi 13 + 14 + echo "=== Submitting $COUNT reports to $BASE_URL ===" 15 + 16 + for i in $(seq 1 "$COUNT"); do 17 + GUID=$(uuidgen | tr '[:upper:]' '[:lower:]') 18 + REPORTER_ID="usr$(openssl rand -hex 4)" 19 + OWNER_ID="own$(openssl rand -hex 4)" 20 + 21 + curl -s -o /dev/null -w "Report $i: %{http_code}\n" \ 22 + -X POST "$BASE_URL/api/v1/report" \ 23 + -H "Content-Type: application/json" \ 24 + -H "x-api-key: $API_KEY" \ 25 + -d "{\"reporter\":{\"kind\":\"user\",\"id\":\"$REPORTER_ID\",\"typeId\":\"502ec98c7e\"},\"reportedAt\":\"2026-03-24T12:00:00Z\",\"reportedItem\":{\"id\":\"$GUID\",\"typeId\":\"a8481310e8c\",\"data\":{\"text\":\"Load test post $i\",\"images\":[],\"owner_id\":{\"id\":\"$OWNER_ID\",\"typeId\":\"502ec98c7e\"},\"num_likes\":$((RANDOM % 100)),\"num_comments\":$((RANDOM % 50)),\"num_user_reports\":$((RANDOM % 20))}}}" & 26 + 27 + if (( i % 10 == 0 )); then wait; echo " ... $i/$COUNT sent"; fi 28 + done 29 + wait 30 + echo "=== Done — $COUNT reports submitted ==="

+1

server/graphql/generated.ts

··· 2188 2188 2189 2189 export type GQLManualReviewQueueJobsArgs = { 2190 2190 ids?: InputMaybe<ReadonlyArray<Scalars['ID']['input']>>; 2191 + limit?: InputMaybe<Scalars['Int']['input']>; 2191 2192 }; 2192 2193 2193 2194 export type GQLManualReviewQueueNameExistsError = GQLError & {

+7 -11

server/graphql/modules/manualReviewTool.ts

··· 42 42 import { gqlErrorResult, gqlSuccessResult } from '../utils/gqlResult.js'; 43 43 import { oneOfInputToTaggedUnion } from '../utils/inputHelpers.js'; 44 44 45 - const { omit, sum, sumBy } = _; 45 + const { omit, sumBy } = _; 46 46 47 47 const typeDefs = /* GraphQL */ ` 48 48 type ManualReviewQueue { ··· 51 51 description: String 52 52 orgId: ID! 53 53 isDefaultQueue: Boolean! 54 - jobs(ids: [ID!]): [ManualReviewJob!]! 54 + jobs(ids: [ID!], limit: Int): [ManualReviewJob!]! 55 55 pendingJobCount: Int! 56 56 oldestJobCreatedAt: DateTime 57 57 explicitlyAssignedReviewers: [User!]! ··· 1632 1632 }; 1633 1633 1634 1634 const ManualReviewQueue: GQLManualReviewQueueResolvers = { 1635 - async jobs(queue, { ids: jobIds }, context) { 1635 + async jobs(queue, { ids: jobIds, limit }, context) { 1636 1636 const { orgId, id: queueId } = queue; 1637 1637 1638 1638 if (!jobIds) { 1639 1639 return context.services.ManualReviewToolService.getAllJobsForQueue({ 1640 1640 orgId, 1641 1641 queueId, 1642 + limit: limit ?? undefined, 1642 1643 }); 1643 1644 } else { 1644 1645 return context.services.ManualReviewToolService.getJobsForQueue({ ··· 1923 1924 { orgId: user.orgId }, 1924 1925 ); 1925 1926 1926 - const jobsPerQueue = await Promise.all( 1927 - allQueues.map(async (queue) => 1928 - context.services.ManualReviewToolService.getPendingJobCount({ 1929 - orgId: user.orgId, 1930 - queueId: queue.id, 1931 - }), 1932 - ), 1927 + return context.services.ManualReviewToolService.getTotalPendingJobCountForQueues( 1928 + user.orgId, 1929 + allQueues.map((q) => q.id), 1933 1930 ); 1934 - return sum(jobsPerQueue); 1935 1931 }, 1936 1932 1937 1933 async getRecentDecisions(_: unknown, { input }, context) {

+12 -1

server/services/manualReviewToolService/manualReviewToolService.ts

··· 951 951 }); 952 952 } 953 953 954 - async getAllJobsForQueue(opts: { orgId: string; queueId: string }) { 954 + async getAllJobsForQueue(opts: { 955 + orgId: string; 956 + queueId: string; 957 + limit?: number; 958 + }) { 955 959 return this.queueOps.getAllJobsForQueue(opts); 956 960 } 957 961 958 962 async getPendingJobCount(opts: { orgId: string; queueId: string }) { 959 963 return this.queueOps.getPendingJobCount(opts); 964 + } 965 + 966 + async getTotalPendingJobCountForQueues( 967 + orgId: string, 968 + queueIds: string[], 969 + ) { 970 + return this.queueOps.getTotalPendingJobCountForQueues(orgId, queueIds); 960 971 } 961 972 962 973 async getOldestJobCreatedAt(opts: {

+32 -9

server/services/manualReviewToolService/modules/QueueOperations.ts

··· 794 794 return filterNullOrUndefined(jobs).map((job) => job.data); 795 795 } 796 796 797 - /** 798 - * This is currently only being used for testing, and should be 799 - * deleted after we rework the list view. 800 - */ 801 - async getAllJobsForQueue(opts: { orgId: string; queueId: string }) { 802 - const limit = pLimit(10); 797 + async getAllJobsForQueue(opts: { 798 + orgId: string; 799 + queueId: string; 800 + limit?: number; 801 + }) { 802 + const concurrencyLimit = pLimit(10); 803 803 const { orgId, queueId } = opts; 804 804 const queue = await this.#getBullQueue(orgId, queueId); 805 - // Get up to 50 jobs per queue. 806 - const legacyJobs = await queue.getJobs(undefined, 0, 50); 805 + const maxJobs = Math.max(0, Math.min(opts.limit ?? 50, 50)); 806 + const legacyJobs = await queue.getJobs(undefined, 0, maxJobs); 807 807 const jobs = await Promise.all( 808 808 // eslint-disable-next-line @typescript-eslint/promise-function-async 809 - legacyJobs.map((job) => limit(() => this.legacyJobToJob(job, orgId))), 809 + legacyJobs.map((job) => 810 + concurrencyLimit(async () => this.legacyJobToJob(job, orgId)), 811 + ), 810 812 ); 811 813 return filterNullOrUndefined(jobs).map((job) => job.data); 812 814 } ··· 1242 1244 // Returns the number of waiting or delayed jobs 1243 1245 // https://api.docs.bullmq.io/classes/Queue.html#count 1244 1246 return queue.count(); 1247 + } 1248 + 1249 + /** 1250 + * Batched variant that skips per-queue existence checks. The caller 1251 + * must have already verified the queues exist (e.g. via 1252 + * getAllQueuesForOrgAndDangerouslyBypassPermissioning). 1253 + */ 1254 + async getTotalPendingJobCountForQueues( 1255 + orgId: string, 1256 + queueIds: string[], 1257 + ): Promise<number> { 1258 + const concurrencyLimit = pLimit(10); 1259 + const counts = await Promise.all( 1260 + queueIds.map(async (queueId) => 1261 + concurrencyLimit(async () => { 1262 + const queue = await this.getOrCreateBullQueue({ orgId, queueId }); 1263 + return queue.count(); 1264 + }), 1265 + ), 1266 + ); 1267 + return counts.reduce((sum, count) => sum + count, 0); 1245 1268 } 1246 1269 1247 1270 async getOldestJobCreatedAt(opts: {

+2 -1

server/services/manualReviewToolService/modules/SkipOperations.ts

··· 84 84 // Skips aren't a decision or are associated with policies, so just skip 85 85 // the query 86 86 if ( 87 - (policyIds && policyIds.length > 0) ?? 87 + (policyIds && policyIds.length > 0) || 88 88 (decisions && decisions.length > 0) 89 89 ) { 90 90 return []; ··· 128 128 ]), 129 129 ), 130 130 ) 131 + .limit(1000) 131 132 .execute(); 132 133 } 133 134 }

Configure Feed

Configure Feed