Skip to content

Commit 29048ec

Browse files
evgeniy-scherbinaevgeniy-scherbina
committed
fix(coderd/database): correctly calculate pending and ranked jobs
1 parent 30f007c commit 29048ec

File tree

4 files changed

+131
-45
lines changed

4 files changed

+131
-45
lines changed

‎coderd/database/querier.go

Lines changed: 0 additions & 2 deletions
Some generated files are not rendered by default. Learn more about customizing how changed files appear on GitHub.

‎coderd/database/querier_test.go

Lines changed: 110 additions & 12 deletions
Original file line numberDiff line numberDiff line change
@@ -2180,6 +2180,11 @@ func TestGetProvisionerJobsByIDsWithQueuePosition(t *testing.T) {
21802180
daemonTags []database.StringMap
21812181
queueSizes []int64
21822182
queuePositions []int64
2183+
// GetProvisionerJobsByIDsWithQueuePosition takes jobIDs as a parameter.
2184+
// If skipJobIDs is empty, all jobs are passed to the function; otherwise, the specified jobs are skipped.
2185+
// NOTE: Skipping job IDs means they will be excluded from the result,
2186+
// but this should not affect the queue position or queue size of other jobs.
2187+
skipJobIDs map[int]struct{}
21832188
}{
21842189
{
21852190
name: "test-case-1",
@@ -2195,6 +2200,7 @@ func TestGetProvisionerJobsByIDsWithQueuePosition(t *testing.T) {
21952200
queueSizes: []int64{2, 2, 0},
21962201
queuePositions: []int64{1, 1, 0},
21972202
},
2203+
// Similar to the previous case, but includes an additional provisioner.
21982204
{
21992205
name: "test-case-2",
22002206
jobTags: []database.StringMap{
@@ -2210,6 +2216,83 @@ func TestGetProvisionerJobsByIDsWithQueuePosition(t *testing.T) {
22102216
queueSizes: []int64{3, 3, 3},
22112217
queuePositions: []int64{1, 1, 3},
22122218
},
2219+
// Similar to the previous case, but skips job at index 0
2220+
{
2221+
name: "test-case-3",
2222+
jobTags: []database.StringMap{
2223+
{"a": "1", "b": "2"},
2224+
{"a": "1"},
2225+
{"a": "1", "c": "3"},
2226+
},
2227+
daemonTags: []database.StringMap{
2228+
{"a": "1", "b": "2"},
2229+
{"a": "1"},
2230+
{"a": "1", "b": "2", "c": "3"},
2231+
},
2232+
queueSizes: []int64{3, 3},
2233+
queuePositions: []int64{1, 3},
2234+
skipJobIDs: map[int]struct{}{
2235+
0: {},
2236+
},
2237+
},
2238+
// Skips job at index 1
2239+
{
2240+
name: "test-case-4",
2241+
jobTags: []database.StringMap{
2242+
{"a": "1", "b": "2"},
2243+
{"a": "1"},
2244+
{"a": "1", "c": "3"},
2245+
},
2246+
daemonTags: []database.StringMap{
2247+
{"a": "1", "b": "2"},
2248+
{"a": "1"},
2249+
{"a": "1", "b": "2", "c": "3"},
2250+
},
2251+
queueSizes: []int64{3, 3},
2252+
queuePositions: []int64{1, 3},
2253+
skipJobIDs: map[int]struct{}{
2254+
1: {},
2255+
},
2256+
},
2257+
// Skips job at index 2
2258+
{
2259+
name: "test-case-5",
2260+
jobTags: []database.StringMap{
2261+
{"a": "1", "b": "2"},
2262+
{"a": "1"},
2263+
{"a": "1", "c": "3"},
2264+
},
2265+
daemonTags: []database.StringMap{
2266+
{"a": "1", "b": "2"},
2267+
{"a": "1"},
2268+
{"a": "1", "b": "2", "c": "3"},
2269+
},
2270+
queueSizes: []int64{3, 3},
2271+
queuePositions: []int64{1, 1},
2272+
skipJobIDs: map[int]struct{}{
2273+
2: {},
2274+
},
2275+
},
2276+
// Skips jobs at indexes 0 and 2
2277+
{
2278+
name: "test-case-6",
2279+
jobTags: []database.StringMap{
2280+
{"a": "1", "b": "2"},
2281+
{"a": "1"},
2282+
{"a": "1", "c": "3"},
2283+
},
2284+
daemonTags: []database.StringMap{
2285+
{"a": "1", "b": "2"},
2286+
{"a": "1"},
2287+
{"a": "1", "b": "2", "c": "3"},
2288+
},
2289+
queueSizes: []int64{3},
2290+
queuePositions: []int64{1},
2291+
skipJobIDs: map[int]struct{}{
2292+
0: {},
2293+
2: {},
2294+
},
2295+
},
22132296
}
22142297

22152298
for _, tc := range testCases {
@@ -2247,19 +2330,28 @@ func TestGetProvisionerJobsByIDsWithQueuePosition(t *testing.T) {
22472330
require.Equal(t, database.ProvisionerJobStatusPending, job.JobStatus, "expected job %d to have status %s", idx, database.ProvisionerJobStatusPending)
22482331
}
22492332

2250-
var jobIDs []uuid.UUID
2251-
for _, job := range allJobs {
2252-
jobIDs = append(jobIDs, job.ID)
2333+
filteredJobs := make([]database.ProvisionerJob, 0)
2334+
filteredJobIDs := make([]uuid.UUID, 0)
2335+
for idx, job := range allJobs {
2336+
if _, skip := tc.skipJobIDs[idx]; skip {
2337+
continue
2338+
}
2339+
2340+
filteredJobs = append(filteredJobs, job)
2341+
filteredJobIDs = append(filteredJobIDs, job.ID)
22532342
}
22542343

22552344
// When: we fetch the jobs by their IDs
2256-
actualJobs, err := db.GetProvisionerJobsByIDsWithQueuePosition(ctx, jobIDs)
2345+
actualJobs, err := db.GetProvisionerJobsByIDsWithQueuePosition(ctx, filteredJobIDs)
22572346
require.NoError(t, err)
2258-
require.Len(t, actualJobs, len(allJobs), "should return all jobs")
2347+
require.Len(t, actualJobs, len(filteredJobIDs), "should return all unskipped jobs")
22592348

2260-
// Then: the jobs should be returned in the correct order (by IDs in the input slice)
2349+
// Then: the jobs should be returned in the correct order (sorted by createdAt)
2350+
sort.Slice(filteredJobs, func(i, j int) bool {
2351+
return filteredJobs[i].CreatedAt.Before(filteredJobs[j].CreatedAt)
2352+
})
22612353
for idx, job := range actualJobs {
2262-
assert.EqualValues(t, allJobs[idx], job.ProvisionerJob)
2354+
assert.EqualValues(t, filteredJobs[idx], job.ProvisionerJob)
22632355
}
22642356

22652357
// Then: the queue size should be set correctly
@@ -2395,7 +2487,10 @@ func TestGetProvisionerJobsByIDsWithQueuePosition_MixedStatuses(t *testing.T) {
23952487
require.NoError(t, err)
23962488
require.Len(t, actualJobs, len(allJobs), "should return all jobs")
23972489

2398-
// Then: the jobs should be returned in the correct order (by IDs in the input slice)
2490+
// Then: the jobs should be returned in the correct order (sorted by createdAt)
2491+
sort.Slice(allJobs, func(i, j int) bool {
2492+
return allJobs[i].CreatedAt.Before(allJobs[j].CreatedAt)
2493+
})
23992494
for idx, job := range actualJobs {
24002495
assert.EqualValues(t, allJobs[idx], job.ProvisionerJob)
24012496
}
@@ -2405,14 +2500,14 @@ func TestGetProvisionerJobsByIDsWithQueuePosition_MixedStatuses(t *testing.T) {
24052500
for _, job := range actualJobs {
24062501
queueSizes = append(queueSizes, job.QueueSize)
24072502
}
2408-
assert.EqualValues(t, []int64{2, 2, 0, 0, 0, 0, 0}, queueSizes, "expected queue positions to be set correctly")
2503+
assert.EqualValues(t, []int64{0, 0, 0, 0, 0, 2, 2}, queueSizes, "expected queue positions to be set correctly")
24092504

24102505
// Then: the queue position should be set correctly:
24112506
var queuePositions []int64
24122507
for _, job := range actualJobs {
24132508
queuePositions = append(queuePositions, job.QueuePosition)
24142509
}
2415-
assert.EqualValues(t, []int64{2, 1, 0, 0, 0, 0, 0}, queuePositions, "expected queue positions to be set correctly")
2510+
assert.EqualValues(t, []int64{0, 0, 0, 0, 0, 1, 2}, queuePositions, "expected queue positions to be set correctly")
24162511
}
24172512

24182513
func TestGetProvisionerJobsByIDsWithQueuePosition_OrderValidation(t *testing.T) {
@@ -2489,7 +2584,10 @@ func TestGetProvisionerJobsByIDsWithQueuePosition_OrderValidation(t *testing.T)
24892584
require.NoError(t, err)
24902585
require.Len(t, actualJobs, len(allJobs), "should return all jobs")
24912586

2492-
// Then: the jobs should be returned in the correct order (by IDs in the input slice)
2587+
// Then: the jobs should be returned in the correct order (sorted by createdAt)
2588+
sort.Slice(allJobs, func(i, j int) bool {
2589+
return allJobs[i].CreatedAt.Before(allJobs[j].CreatedAt)
2590+
})
24932591
for idx, job := range actualJobs {
24942592
assert.EqualValues(t, allJobs[idx], job.ProvisionerJob)
24952593
assert.EqualValues(t, allJobs[idx].CreatedAt, job.ProvisionerJob.CreatedAt)
@@ -2507,7 +2605,7 @@ func TestGetProvisionerJobsByIDsWithQueuePosition_OrderValidation(t *testing.T)
25072605
for _, job := range actualJobs {
25082606
queuePositions = append(queuePositions, job.QueuePosition)
25092607
}
2510-
assert.EqualValues(t, []int64{3, 2, 1, 4, 5, 6}, queuePositions, "expected queue positions to be set correctly")
2608+
assert.EqualValues(t, []int64{1, 2, 3, 4, 5, 6}, queuePositions, "expected queue positions to be set correctly")
25112609
}
25122610

25132611
func TestGroupRemovalTrigger(t *testing.T) {

‎coderd/database/queries.sql.go

Lines changed: 11 additions & 18 deletions
Some generated files are not rendered by default. Learn more about customizing how changed files appear on GitHub.

‎coderd/database/queries/provisionerjobs.sql

Lines changed: 10 additions & 13 deletions
Original file line numberDiff line numberDiff line change
@@ -51,19 +51,18 @@ WHERE
5151

5252
-- name: GetProvisionerJobsByIDsWithQueuePosition :many
5353
WITH filtered_provisioner_jobs AS (
54-
-- Step 1: Filter provisioner_jobs and assign an order from upstream system
54+
-- Step 1: Filter provisioner_jobs
5555
SELECT
56-
*,
57-
ROW_NUMBER() OVER () AS ordinality -- Track original order
56+
*
5857
FROM
5958
provisioner_jobs
6059
WHERE
6160
id = ANY(@ids :: uuid [ ]) -- Apply filter early to reduce dataset size before expensive JOINs
6261
),
6362
pending_jobs AS (
64-
-- Step 2: Extract only pending jobs from the already filtered dataset
63+
-- Step 2: Extract only pending jobs
6564
SELECT *
66-
FROM filtered_provisioner_jobs
65+
FROM provisioner_jobs
6766
WHERE started_at IS NULL
6867
AND canceled_at IS NULL
6968
AND completed_at IS NULL
@@ -79,38 +78,36 @@ ranked_jobs AS (
7978
FROM
8079
pending_jobs pj
8180
INNER JOIN provisioner_daemons pd
82-
ON provisioner_tagset_contains(pd.tags, pj.tags) -- Join only on the small filtered pending set
81+
ON provisioner_tagset_contains(pd.tags, pj.tags) -- Join only on the small pending set
8382
),
8483
final_jobs AS (
8584
-- Step 4: Compute best queue position and max queue size per job
8685
SELECT
8786
fpj.id,
8887
fpj.created_at,
8988
COALESCE(MIN(rj.queue_position), 0) :: BIGINT AS queue_position, -- Best queue position across provisioners
90-
COALESCE(MAX(rj.queue_size), 0) :: BIGINT AS queue_size, -- Max queue size across provisioners
91-
fpj.ordinality -- Preserve original order
89+
COALESCE(MAX(rj.queue_size), 0) :: BIGINT AS queue_size -- Max queue size across provisioners
9290
FROM
9391
filtered_provisioner_jobs fpj -- Use the pre-filtered dataset instead of full provisioner_jobs
9492
LEFT JOIN ranked_jobs rj
9593
ON fpj.id = rj.id -- Ensure we only keep jobs that have a ranking
9694
GROUP BY
97-
fpj.id, fpj.created_at, fpj.ordinality -- Include `ordinality` in GROUP BY
95+
fpj.id, fpj.created_at
9896
)
99-
-- Step 5: Final SELECT with INNER JOIN provisioner_jobs
10097
SELECT
98+
-- Step 5: Final SELECT with INNER JOIN provisioner_jobs
10199
fj.id,
102100
fj.created_at,
103101
sqlc.embed(pj),
104102
fj.queue_position,
105-
fj.queue_size,
106-
fj.ordinality
103+
fj.queue_size
107104
FROM
108105
final_jobs fj
109106
INNER JOIN provisioner_jobs pj
110107
ON fj.id = pj.id -- Ensure we retrieve full details from `provisioner_jobs`.
111108
-- JOIN with pj is required for sqlc.embed(pj) to compile successfully.
112109
ORDER BY
113-
fj.ordinality; -- Preserve original ID order from upstream
110+
fj.created_at;
114111

115112
-- name: GetProvisionerJobsByOrganizationAndStatusWithQueuePositionAndProvisioner :many
116113
WITH pending_jobs AS (

0 commit comments

Comments
 (0)