Skip to content

Commit 7fc9e0f

Browse files
GiovaniGuizzoGiovaniGuizzo
authored
fix: added Math.min to the dispatcher and removed redundant checks (#152)
* fix: added Math.min to the dispatcher and removed redundant checks * fix: update test to create 4 jobs for claiming logic * fix: refactor CI workflow to use setup action and streamline job steps * fix: add checkout step to pull request workflow and update action description * fix: add missing cache paths for Yarn and Node.js modules
1 parent a5df939 commit 7fc9e0f

7 files changed

Lines changed: 181 additions & 100 deletions

File tree

‎.github/actions/setup/action.yml‎

Lines changed: 31 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,31 @@
1+
name: Setup Project
2+
description: Setup Node.js, restore build cache, and setup Yarn
3+
4+
inputs:
5+
node-version:
6+
description: 'Node.js version to use'
7+
required: false
8+
default: '22'
9+
10+
runs:
11+
using: composite
12+
steps:
13+
- name: Set up Node.js
14+
uses: actions/setup-node@v4
15+
with:
16+
node-version: ${{ inputs.node-version }}
17+
18+
- name: Restore build outputs
19+
uses: actions/cache@v4
20+
with:
21+
path: |
22+
.yarn
23+
node_modules
24+
**/node_modules
25+
**/dist
26+
key: ${{ runner.os }}-build-${{ hashFiles('**/yarn.lock') }}-${{ github.sha }}
27+
28+
- name: Setup Latest Yarn
29+
uses: threeal/setup-yarn-action@v2.0.0
30+
with:
31+
version: berry

‎.github/workflows/pull-request.yml‎

Lines changed: 65 additions & 20 deletions
Original file line numberDiff line numberDiff line change
@@ -15,9 +15,47 @@ concurrency:
1515
cancel-in-progress: true
1616

1717
jobs:
18-
build-and-test:
18+
build:
1919
runs-on: ubuntu-latest
20+
steps:
21+
- name: Checkout repository
22+
uses: actions/checkout@v4
23+
24+
- name: Setup project
25+
uses: ./.github/actions/setup
26+
27+
- name: Build project
28+
run: yarn build
29+
30+
format-check:
31+
runs-on: ubuntu-latest
32+
needs: build
33+
steps:
34+
- name: Checkout repository
35+
uses: actions/checkout@v4
36+
37+
- name: Setup project
38+
uses: ./.github/actions/setup
39+
40+
- name: Run format checker
41+
run: yarn format:check
42+
43+
lint:
44+
runs-on: ubuntu-latest
45+
needs: build
46+
steps:
47+
- name: Checkout repository
48+
uses: actions/checkout@v4
2049

50+
- name: Setup project
51+
uses: ./.github/actions/setup
52+
53+
- name: Run linter
54+
run: yarn lint
55+
56+
test:
57+
runs-on: ubuntu-latest
58+
needs: build
2159
services:
2260
postgres:
2361
image: postgres:latest
@@ -59,25 +97,8 @@ jobs:
5997
- name: Checkout repository
6098
uses: actions/checkout@v4
6199

62-
- name: Set up Node.js
63-
uses: actions/setup-node@v4
64-
with:
65-
node-version: "22"
66-
67-
- name: Setup Latest Yarn
68-
uses: threeal/setup-yarn-action@v2.0.0
69-
with:
70-
version: berry
71-
cache: false
72-
73-
- name: Build project
74-
run: yarn build
75-
76-
- name: Run format checker
77-
run: yarn format:check
78-
79-
- name: Run linter
80-
run: yarn lint
100+
- name: Setup project
101+
uses: ./.github/actions/setup
81102

82103
- name: Run tests
83104
env:
@@ -86,6 +107,30 @@ jobs:
86107
MONGODB_URL: mongodb://127.0.0.1:27017/test
87108
run: yarn test:all:ci
88109

110+
integration-test:
111+
runs-on: ubuntu-latest
112+
needs: build
113+
services:
114+
postgres:
115+
image: postgres:latest
116+
ports:
117+
- 5432:5432
118+
env:
119+
POSTGRES_USER: postgres
120+
POSTGRES_PASSWORD: postgres
121+
POSTGRES_DB: postgres
122+
options: >-
123+
--health-cmd="pg_isready -U postgres"
124+
--health-interval=10s
125+
--health-timeout=5s
126+
--health-retries=5
127+
steps:
128+
- name: Checkout repository
129+
uses: actions/checkout@v4
130+
131+
- name: Setup project
132+
uses: ./.github/actions/setup
133+
89134
- name: Run integration tests
90135
env:
91136
POSTGRES_URL: postgresql://postgres:postgres@localhost:5432/postgres

‎packages/core/src/transitions/snooze-transition.ts‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -10,7 +10,7 @@ import { JobTransition } from "./transition";
1010
* If the job is currently running, it will decrement the attempt count.
1111
* This allows the job to be retried after the delay.
1212
*
13-
* Only jobs in "waiting" or "claimed" or "running" state can be snoozed.
13+
* Only jobs in "waiting" or "running" state can be snoozed.
1414
*/
1515
export class SnoozeTransition extends JobTransition {
1616
/** The delay in milliseconds. */
@@ -40,6 +40,6 @@ export class SnoozeTransition extends JobTransition {
4040
}
4141

4242
shouldRun(job: JobData): boolean {
43-
return ["waiting", "claimed", "running"].includes(job.state);
43+
return ["waiting", "running"].includes(job.state);
4444
}
4545
}

‎packages/engine/src/execution/dispatcher.test.ts‎

Lines changed: 44 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -120,5 +120,49 @@ describe("Dispatcher", () => {
120120

121121
await dispatcher.stop();
122122
});
123+
124+
sidequestTest(
125+
"claims min(availableSlots, globalSlots) jobs when queue has more slots than global",
126+
async ({ backend }) => {
127+
// Setup: Queue has concurrency of 10, but global has only 3 slots
128+
const configWithHighQueueConcurrency: EngineConfig = {
129+
backend: { driver: "@sidequest/sqlite-backend" },
130+
queues: [{ name: "default", concurrency: 10 }],
131+
maxConcurrentJobs: 3,
132+
};
133+
134+
// Create 4 jobs to ensure there are enough jobs to claim
135+
await createJob(backend, "default");
136+
await createJob(backend, "default");
137+
await createJob(backend, "default");
138+
await createJob(backend, "default");
139+
140+
expect(await backend.listJobs({ state: "waiting" })).toHaveLength(5);
141+
142+
const mockClaim = vi.spyOn(backend, "claimPendingJob");
143+
144+
const dispatcher = new Dispatcher(
145+
backend,
146+
new QueueManager(backend, configWithHighQueueConcurrency.queues!),
147+
new ExecutorManager(backend, configWithHighQueueConcurrency as NonNullableEngineConfig),
148+
100,
149+
);
150+
dispatcher.start();
151+
152+
runMock.mockImplementation(() => {
153+
return { type: "completed", result: "foo", __is_job_transition__: true } as CompletedResult;
154+
});
155+
156+
// Wait for the first claim to happen
157+
await vi.waitUntil(() => mockClaim.mock.calls.length > 0);
158+
159+
// Verify that claimPendingJob was called with Math.min(availableSlots, globalSlots)
160+
// Queue has 10 slots available, global has 3 slots available
161+
// So it should claim min(10, 3) = 3 jobs
162+
expect(mockClaim).toHaveBeenCalledWith("default", 3);
163+
164+
await dispatcher.stop();
165+
},
166+
);
123167
});
124168
});

‎packages/engine/src/execution/dispatcher.ts‎

Lines changed: 5 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -48,14 +48,18 @@ export class Dispatcher {
4848
continue;
4949
}
5050

51-
const jobs: JobData[] = await this.backend.claimPendingJob(queue.name, availableSlots);
51+
const jobs: JobData[] = await this.backend.claimPendingJob(queue.name, Math.min(availableSlots, globalSlots));
5252

5353
if (jobs.length > 0) {
5454
// if a job was found on any queue do not sleep
5555
shouldSleep = false;
5656
}
5757

5858
for (const job of jobs) {
59+
// adds jobs to active sets before execution to avoid race conditions
60+
// because the execution is not awaited. This way we ensure that available slots
61+
// are correctly calculated.
62+
this.executorManager.queueJob(queue, job);
5963
// does not await for job execution.
6064
void this.executorManager.execute(queue, job);
6165
}

‎packages/engine/src/execution/executor-manager.test.ts‎

Lines changed: 0 additions & 40 deletions
Original file line numberDiff line numberDiff line change
@@ -171,46 +171,6 @@ describe("ExecutorManager", () => {
171171

172172
await executorManager.destroy();
173173
});
174-
175-
sidequestTest("snoozes job when queue is full", async ({ backend, config }) => {
176-
const queryConfig = await grantQueueConfig(backend, { name: "default", concurrency: 1 });
177-
const executorManager = new ExecutorManager(backend, config);
178-
179-
vi.spyOn(executorManager, "availableSlotsByQueue").mockReturnValue(0);
180-
181-
// Set up job in claimed state (as it would be when passed to execute)
182-
jobData = await backend.updateJob({ ...jobData, state: "claimed", claimed_at: new Date() });
183-
184-
await executorManager.execute(queryConfig, jobData);
185-
186-
// Verify the job runner was NOT called since the job was snoozed
187-
expect(runMock).not.toHaveBeenCalled();
188-
189-
// Verify slots remain unchanged (no job was actually executed)
190-
expect(executorManager.availableSlotsByQueue(queryConfig)).toEqual(0);
191-
expect(executorManager.totalActiveWorkers()).toEqual(0);
192-
await executorManager.destroy();
193-
});
194-
195-
sidequestTest("snoozes job when global slots are full", async ({ backend, config }) => {
196-
const queryConfig = await grantQueueConfig(backend, { name: "default", concurrency: 5 });
197-
const executorManager = new ExecutorManager(backend, { ...config, maxConcurrentJobs: 1 });
198-
199-
vi.spyOn(executorManager, "availableSlotsGlobal").mockReturnValue(0);
200-
201-
// Set up job in claimed state
202-
jobData = await backend.updateJob({ ...jobData, state: "claimed", claimed_at: new Date() });
203-
204-
await executorManager.execute(queryConfig, jobData);
205-
206-
// Verify the job runner was NOT called
207-
expect(runMock).not.toHaveBeenCalled();
208-
209-
// Verify global slots show as full
210-
expect(executorManager.availableSlotsGlobal()).toEqual(0);
211-
expect(executorManager.totalActiveWorkers()).toEqual(0);
212-
await executorManager.destroy();
213-
});
214174
});
215175

216176
describe("availableSlotsByQueue", () => {

‎packages/engine/src/execution/executor-manager.ts‎

Lines changed: 34 additions & 37 deletions
Original file line numberDiff line numberDiff line change
@@ -1,13 +1,5 @@
11
import { Backend } from "@sidequest/backend";
2-
import {
3-
JobData,
4-
JobTransitionFactory,
5-
logger,
6-
QueueConfig,
7-
RetryTransition,
8-
RunTransition,
9-
SnoozeTransition,
10-
} from "@sidequest/core";
2+
import { JobData, JobTransitionFactory, logger, QueueConfig, RetryTransition, RunTransition } from "@sidequest/core";
113
import EventEmitter from "events";
124
import { inspect } from "util";
135
import { NonNullableEngineConfig } from "../engine";
@@ -77,44 +69,49 @@ export class ExecutorManager {
7769
}
7870

7971
/**
80-
* Executes a job in the given queue.
72+
* Prepares a job for execution by marking it as active and adding it to a queue slot.
8173
* @param queueConfig The queue configuration.
82-
* @param job The job data to execute.
74+
* @param job The job data.
8375
*/
84-
async execute(queueConfig: QueueConfig, job: JobData): Promise<void> {
85-
logger("Executor Manager").debug(`Submitting job ${job.id} for execution in queue ${queueConfig.name}`);
76+
queueJob(queueConfig: QueueConfig, job: JobData) {
8677
if (!this.activeByQueue[queueConfig.name]) {
8778
this.activeByQueue[queueConfig.name] = new Set();
8879
}
89-
90-
if (this.availableSlotsByQueue(queueConfig) <= 0 || this.availableSlotsGlobal() <= 0) {
91-
logger("Executor Manager").debug(`No available slots for job ${job.id} in queue ${queueConfig.name}`);
92-
await JobTransitioner.apply(this.backend, job, new SnoozeTransition(0));
93-
return;
94-
}
95-
9680
this.activeByQueue[queueConfig.name].add(job.id);
9781
this.activeJobs.add(job.id);
82+
}
9883

99-
job = await JobTransitioner.apply(this.backend, job, new RunTransition());
100-
101-
const signal = new EventEmitter();
102-
let isRunning = true;
103-
const cancellationCheck = async () => {
104-
while (isRunning) {
105-
const watchedJob = await this.backend.getJob(job.id);
106-
if (watchedJob!.state === "canceled") {
107-
logger("Executor Manager").debug(`Emitting abort signal for job ${job.id}`);
108-
signal.emit("abort");
109-
isRunning = false;
110-
return;
84+
/**
85+
* Executes a job in the given queue.
86+
* @param queueConfig The queue configuration.
87+
* @param job The job data to execute.
88+
*/
89+
async execute(queueConfig: QueueConfig, job: JobData): Promise<void> {
90+
let isRunning = false;
91+
try {
92+
logger("Executor Manager").debug(`Submitting job ${job.id} for execution in queue ${queueConfig.name}`);
93+
// We call prepareJob here again to make sure the jobs are in the queues.
94+
// This might not be necessary, but for the sake of consistency we do it.
95+
this.queueJob(queueConfig, job);
96+
97+
job = await JobTransitioner.apply(this.backend, job, new RunTransition());
98+
99+
isRunning = true;
100+
const signal = new EventEmitter();
101+
const cancellationCheck = async () => {
102+
while (isRunning) {
103+
const watchedJob = await this.backend.getJob(job.id);
104+
if (watchedJob!.state === "canceled") {
105+
logger("Executor Manager").debug(`Emitting abort signal for job ${job.id}`);
106+
signal.emit("abort");
107+
isRunning = false;
108+
return;
109+
}
110+
await new Promise((r) => setTimeout(r, 1000));
111111
}
112-
await new Promise((r) => setTimeout(r, 1000));
113-
}
114-
};
115-
void cancellationCheck();
112+
};
113+
void cancellationCheck();
116114

117-
try {
118115
logger("Executor Manager").debug(`Running job ${job.id} in queue ${queueConfig.name}`);
119116

120117
const runPromise = this.runnerPool.run(job, signal);

0 commit comments

Comments
 (0)