Feature: add support for jobs to run sequentially #2461
There are no files selected for viewing
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -96,7 +96,8 @@ void BedrockPlugin_Jobs::upgradeDatabase(SQLite& db) | |
| "data TEXT NOT NULL, " | ||
| "priority INTEGER NOT NULL DEFAULT " + SToStr(JOBS_DEFAULT_PRIORITY) + ", " | ||
| "parentJobID INTEGER NOT NULL DEFAULT 0, " | ||
| "retryAfter TEXT NOT NULL DEFAULT \"\", " | ||
| "sequentialKey TEXT NOT NULL DEFAULT \"\")", | ||
|
Comment on lines
96
to
+100
There was a problem hiding this comment.
This change only updates the Useful? React with 👍 / 👎.
Member
Author
There was a problem hiding this comment. we don't do this in code right? have to manually run |
||
| ignore)); | ||
| // verify and conditionally create indexes | ||
| SASSERT(db.verifyIndex("jobsName", "jobs", "( name )", false, !BedrockPlugin_Jobs::isLive)); | ||
|
|
@@ -608,13 +609,29 @@ void BedrockJobsCommand::process(SQLite& db) | |
| } | ||
| } | ||
|
|
||
| // If sequentialKey is set, check if there are other pending jobs with the same key. | ||
| // If so, this job should wait for them to complete before running by setting the initial state to WAITING. | ||
| const string& safeSequentialKey = SContains(job, "sequentialKey") ? SQ(job["sequentialKey"]) : SQ(""); | ||
| if (safeSequentialKey != SQ("")) { | ||
| SQResult result; | ||
| if (!db.read("SELECT 1 FROM jobs " | ||
| "WHERE sequentialKey=" + safeSequentialKey + " " | ||
| "AND state IN ('QUEUED', 'RUNQUEUED', 'RUNNING', 'WAITING') " | ||
| "LIMIT 1;", result)) { | ||
| STHROW("502 Select failed"); | ||
| } | ||
| if (!result.empty()) { | ||
| initialState = "WAITING"; | ||
| } | ||
| } | ||
|
Comment on lines
+612
to
+626
There was a problem hiding this comment. Missing validation to prevent unsupported combinations of sequentialKey with repeat or parentJobID. The PR description states these are unsupported because jobs with repeat or parentJobID don't get deleted on completion, preventing WAITING jobs from being promoted. However, there's no validation to prevent users from creating jobs with these combinations, which would lead to unexpected behavior where WAITING jobs never get promoted to QUEUED.
Comment on lines
+612
to
+626
There was a problem hiding this comment. The sequentialKey logic can override the parentJobID state logic, which could break parent-child job semantics. When both parentJobID and sequentialKey are set (an unsupported combination per the PR description), a child job that should be PAUSED (lines 605-610) can be set to WAITING (line 624), preventing it from running when the parent finishes. This is another reason why the combination should be explicitly validated and rejected. |
||
|
|
||
| // If no data was provided, use an empty object | ||
| const string& safeRetryAfter = SContains(job, "retryAfter") && !job["retryAfter"].empty() ? SQ(job["retryAfter"]) : SQ(""); | ||
|
|
||
| // Create this new job with a new generated ID | ||
| const int64_t jobIDToUse = SQLiteUtils::getRandomID(db, "jobs", "jobID"); | ||
| SINFO("Next jobID: " << jobIDToUse); | ||
| if (!db.writeIdempotent("INSERT INTO jobs ( jobID, created, state, name, nextRun, repeat, data, priority, parentJobID, retryAfter, sequentialKey ) " | ||
| "VALUES( " + | ||
| SQ(jobIDToUse) + ", " + | ||
| currentTime + ", " + | ||
|
|
@@ -625,7 +642,8 @@ void BedrockJobsCommand::process(SQLite& db) | |
| safeData + ", " + | ||
| SQ(priority) + ", " + | ||
| SQ(parentJobID) + ", " + | ||
| safeRetryAfter + ", " + | ||
| safeSequentialKey + " " + | ||
| " );")) { | ||
| STHROW("502 insert query failed"); | ||
| } | ||
|
|
@@ -1011,7 +1029,7 @@ void BedrockJobsCommand::process(SQLite& db) | |
|
|
||
| // Verify there is a job like this and it's running | ||
| SQResult result; | ||
| if (!db.read("SELECT state, nextRun, lastRun, repeat, parentJobID, json_extract(data, '$.mockRequest'), retryAfter, json_extract(data, '$.originalNextRun'), sequentialKey " | ||
| "FROM jobs " | ||
| "WHERE jobID=" + SQ(jobID) + ";", | ||
| result)) { | ||
|
|
@@ -1029,6 +1047,7 @@ void BedrockJobsCommand::process(SQLite& db) | |
| mockRequest = result[0][5] == "1"; | ||
| const string retryAfter = result[0][6]; | ||
| const string originalDataNextRun = result[0][7]; | ||
| const string& sequentialKey = result[0][8]; | ||
|
|
||
| // Make sure we're finishing a job that's actually running | ||
| if (state != "RUNNING" && state != "RUNQUEUED" && !mockRequest) { | ||
|
|
@@ -1199,6 +1218,17 @@ void BedrockJobsCommand::process(SQLite& db) | |
| if (!db.read("SELECT 1 FROM jobs WHERE parentJobID != 0 AND parentJobID=" + SQ(jobID) + " LIMIT 1;").empty()) { | ||
| STHROW("405 Failed to delete a job with outstanding children"); | ||
| } | ||
|
|
||
| // Promote the next job in sequence that is WAITING | ||
| if (!sequentialKey.empty()) { | ||
| db.writeIdempotent("UPDATE jobs SET state='QUEUED' " | ||
| "WHERE jobID = (" | ||
| " SELECT jobID FROM jobs " | ||
| " WHERE sequentialKey=" + SQ(sequentialKey) + " " | ||
| " AND state='WAITING' " | ||
| " ORDER BY created ASC " | ||
| " LIMIT 1);"); | ||
| } | ||
| } | ||
| } | ||
|
|
||
|
|
@@ -1258,23 +1288,25 @@ void BedrockJobsCommand::process(SQLite& db) | |
| // - data - Data to associate with this failed job | ||
| // | ||
| BedrockPlugin::verifyAttributeInt64(request, "jobID", 1); | ||
| const int64_t jobID = request.calc64("jobID"); | ||
|
|
||
| // Verify there is a job like this and it's running | ||
| SQResult result; | ||
| if (!db.read("SELECT state, nextRun, lastRun, repeat, sequentialKey " | ||
| "FROM jobs " | ||
| "WHERE jobID=" + SQ(jobID) + ";", | ||
| result)) { | ||
| STHROW("502 Select failed"); | ||
| } | ||
| if (result.empty()) { | ||
| STHROW("404 No job with this jobID"); | ||
| } | ||
| const string& state = result[0][0]; | ||
| const string& sequentialKey = result[0][4]; | ||
|
|
||
| // Make sure we're failing a job that's actually running or running with a retryAfter | ||
| if (state != "RUNNING" && state != "RUNQUEUED") { | ||
| SINFO("Trying to fail job#" << jobID << ", but isn't RUNNING or RUNQUEUED (" << state << ")"); | ||
| STHROW("405 Can only fail RUNNING or RUNQUEUED jobs"); | ||
| } | ||
|
|
||
|
|
@@ -1289,10 +1321,21 @@ void BedrockJobsCommand::process(SQLite& db) | |
| updateList.push_back("state='FAILED'"); | ||
|
|
||
| // Update this job | ||
| if (!db.writeIdempotent("UPDATE jobs SET " + SComposeList(updateList) + "WHERE jobID=" + SQ(jobID) + ";")) { | ||
| STHROW("502 Fail failed"); | ||
| } | ||
|
|
||
| // Promote the next WAITING job with the same sequentialKey | ||
| if (!sequentialKey.empty()) { | ||
| db.writeIdempotent("UPDATE jobs SET state='QUEUED' " | ||
| "WHERE jobID = (" | ||
| " SELECT jobID FROM jobs " | ||
| " WHERE sequentialKey=" + SQ(sequentialKey) + " " | ||
| " AND state='WAITING' " | ||
| " ORDER BY created ASC " | ||
| " LIMIT 1);"); | ||
| } | ||
|
|
||
| // Successfully processed | ||
| return; | ||
| } | ||
|
|
@@ -1306,32 +1349,46 @@ void BedrockJobsCommand::process(SQLite& db) | |
| // - jobID - ID of the job to delete | ||
| // | ||
| BedrockPlugin::verifyAttributeInt64(request, "jobID", 1); | ||
| int64_t jobID = request.calc64("jobID"); | ||
|
|
||
| // Verify there is a job like this and it's not running | ||
| SQResult result; | ||
| if (!db.read("SELECT state, sequentialKey " | ||
| "FROM jobs " | ||
| "WHERE jobID=" + SQ(jobID) + ";", | ||
| result)) { | ||
| STHROW("502 Select failed"); | ||
| } | ||
| if (result.empty()) { | ||
| STHROW("404 No job with this jobID"); | ||
| } | ||
| const string& state = result[0][0]; | ||
| const string& sequentialKey = result[0][1]; | ||
|
|
||
| if (state == "RUNNING") { | ||
| STHROW("405 Can't delete a RUNNING job"); | ||
| } | ||
| if (state == "PAUSED") { | ||
| STHROW("405 Can't delete a parent jobs with children running"); | ||
| } | ||
|
|
||
| // Delete the job | ||
| if (!db.writeIdempotent("DELETE FROM jobs " | ||
| "WHERE jobID=" + SQ(jobID) + ";")) { | ||
| STHROW("502 Delete failed"); | ||
| } | ||
|
|
||
| // Promote the next WAITING job with the same sequentialKey | ||
| if (!sequentialKey.empty()) { | ||
| db.writeIdempotent("UPDATE jobs SET state='QUEUED' " | ||
| "WHERE jobID = (" | ||
| " SELECT jobID FROM jobs " | ||
| " WHERE sequentialKey=" + SQ(sequentialKey) + " " | ||
| " AND state='WAITING' " | ||
| " ORDER BY created ASC " | ||
| " LIMIT 1);"); | ||
| } | ||
|
|
||
| // Successfully processed | ||
| return; | ||
| } | ||
|
|
||
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Oops, something went wrong.
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Missing database index for sequentialKey column. The queries at lines 617-620, 1224-1230, 1330-1336, and 1383-1389 filter by sequentialKey without an index, which will cause performance issues as the jobs table grows. An index should be added similar to other indexed columns.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Do we need an index? I thinkk let's add it when it becomes a problem