AI/firecrawl
1
0
Fork 0
mirror of https://git.mirrors.martin98.com/https://github.com/mendableai/firecrawl synced 2025-08-11 21:09:00 +08:00
Code Issues Packages Projects Releases Wiki Activity

fix(concurrency-limit): rework cc queue to work by time not priority (#1526)

Browse Source
This commit is contained in:
Gergő Móricz 2025-05-08 12:40:13 +02:00 committed by GitHub
parent 5d07cccd65
commit 7ad9a00ea8
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
3 changed files with 7 additions and 5 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

6
apps/api/src/__tests__/concurrency-limit.test.ts
View File

@ -126,7 +126,7 @@ describe("Concurrency Limit", () => {
describe("pushConcurrencyLimitedJob", () => { describe("pushConcurrencyLimitedJob", () => {
it("should add job to queue with priority", async () => { it("should add job to queue with priority", async () => {
await pushConcurrencyLimitedJob(mockTeamId, mockJob); await pushConcurrencyLimitedJob(mockTeamId, mockJob, 30000);
expect(redisConnection.zadd).toHaveBeenCalledWith( expect(redisConnection.zadd).toHaveBeenCalledWith(
"concurrency-limit-queue:test-team-id", "concurrency-limit-queue:test-team-id",
@ -139,7 +139,7 @@ describe("Concurrency Limit", () => {
const jobWithoutPriority = { ...mockJob }; const jobWithoutPriority = { ...mockJob };
delete jobWithoutPriority.priority; delete jobWithoutPriority.priority;
await pushConcurrencyLimitedJob(mockTeamId, jobWithoutPriority); await pushConcurrencyLimitedJob(mockTeamId, jobWithoutPriority, 30000);
expect(redisConnection.zadd).toHaveBeenCalledWith( expect(redisConnection.zadd).toHaveBeenCalledWith(
"concurrency-limit-queue:test-team-id", "concurrency-limit-queue:test-team-id",
@ -181,7 +181,7 @@ describe("Concurrency Limit", () => {
}; };
// Push job to queue // Push job to queue
await pushConcurrencyLimitedJob(mockTeamId, mockJob); await pushConcurrencyLimitedJob(mockTeamId, mockJob, 30000);
expect(redisConnection.zadd).toHaveBeenCalled(); expect(redisConnection.zadd).toHaveBeenCalled();
// Take job from queue // Take job from queue

4
apps/api/src/lib/concurrency-limit.ts
View File

@ -56,6 +56,7 @@ export type ConcurrencyLimitedJob = {
export async function takeConcurrencyLimitedJob( export async function takeConcurrencyLimitedJob(
team_id: string, team_id: string,
): Promise<ConcurrencyLimitedJob | null> { ): Promise<ConcurrencyLimitedJob | null> {
await redisConnection.zremrangebyscore(constructQueueKey(team_id), -Infinity, Date.now());
const res = await redisConnection.zmpop(1, constructQueueKey(team_id), "MIN"); const res = await redisConnection.zmpop(1, constructQueueKey(team_id), "MIN");
if (res === null || res === undefined) { if (res === null || res === undefined) {
return null; return null;
@ -67,10 +68,11 @@ export async function takeConcurrencyLimitedJob(
export async function pushConcurrencyLimitedJob( export async function pushConcurrencyLimitedJob(
team_id: string, team_id: string,
job: ConcurrencyLimitedJob, job: ConcurrencyLimitedJob,
timeout: number,
) { ) {
await redisConnection.zadd( await redisConnection.zadd(
constructQueueKey(team_id), constructQueueKey(team_id),
job.priority ?? 1, Date.now() + timeout,
JSON.stringify(job), JSON.stringify(job),
); );
} }

2
apps/api/src/services/queue-jobs.ts
View File

@ -46,7 +46,7 @@ async function _addScrapeJobToConcurrencyQueue(
jobId: jobId, jobId: jobId,
}, },
priority: jobPriority, priority: jobPriority,
}); }, webScraperOptions.scrapeOptions?.timeout ?? (60 * 1000));
} }
async function _addCrawlScrapeJobToConcurrencyQueue( async function _addCrawlScrapeJobToConcurrencyQueue(