Conversation
- Rename task acquire/create/publish events to block-level variants - Update server auth checks to use immutable API borrows
- Change CreateTaskBlock/PublishTaskBlock to work on TaskBlock payloads - Add LEASED task state and move queued/solved access to sled helpers - Use direct DB helpers for queued/solved deletes, scans, and inserts
- Serialize queue refills per identifier to avoid TOCTOU duplicates - Remove queued entries after publish and prune expired leases periodically
- Switch client/server/packer paths to block-oriented task acquire, publish, and delete calls - Rename task lifecycle events to block variants and update API types - Add an in-process gRPC TPS benchmark for enqueue/process throughput
IGN-Styly
marked this pull request as ready for review
July 20, 2026 18:40
Contributor
Greptile SummaryThis PR establishes the task nucleus core —
Confidence Score: 5/5The core task lifecycle is now substantially correct and safe to merge. All previously-flagged blocking issues in the nucleus layer have been addressed. Remaining observations are limited to error-code naming inconsistencies and a return-semantics mismatch in the not-yet-wired event gate functions — neither causes current incorrect behavior. enginelib/src/events/mod.rs warrants a second look when the transport layer is wired: BeforeTaskBlock* gate functions return true=blocked while auth check functions return true=allowed. Important Files Changed
Reviews (5): Last reviewed commit: "grptile issue" | Re-trigger Greptile |
![greptile-apps[bot]](https://avatars.githubusercontent.com/in/867647?s=60&v=4){kind=small}
Co-authored-by: greptile-apps[bot] <165735046+greptile-apps[bot]@users.noreply.github.com>
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
Contributor
There was a problem hiding this comment.
Actionable comments posted: 4
Caution
Some comments are outside the diff and can’t be posted inline due to platform limitations.
⚠️ Outside diff range comments (1)
enginelib/src/event.rs (1)
126-132: 🩺 Stability & Availability | 🟠 Major | ⚡ Quick winClone the handler list before dispatching.
event_handler_registry.event_handlers.get(&id)keeps a DashMap read guard alive for the whole callback loop. If a handler re-enters the registry and callsregister_handler, that write will wait on the guard held here and can self-deadlock. Pull theArcs out first, then invoke handlers from the cloned list.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@enginelib/src/event.rs` around lines 126 - 132, Clone the handler Arc list before dispatching in the event handling loop around event_handler_registry.event_handlers.get(&id). Release the DashMap read guard before iterating, then apply the existing cancellation filtering and handler.handle(event) calls to the cloned list so callbacks can safely register handlers reentrantly.
🧹 Nitpick comments (3)
enginelib/src/api.rs (2)
47-51: 🩺 Stability & Availability | 🔵 Trivial | ⚡ Quick win
DB::open(...).unwrap()panics with no context on startup failure.A locked/corrupted DB, missing permissions, or a stale lock file will abort the process with an opaque panic. Prefer surfacing the path and error:
Proposed change
- let db = Arc::new(rust_rocksdb::DB::open(&opts, path).unwrap()); + let db = Arc::new( + rust_rocksdb::DB::open(&opts, path) + .unwrap_or_else(|err| panic!("Failed to open RocksDB at {path}: {err}")), + );🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@enginelib/src/api.rs` around lines 47 - 51, Update the database initialization around DB::open to handle startup failures without an opaque unwrap panic. Preserve the "./engine_db" path, but propagate or report the open error with both the path and underlying error details before terminating according to the existing startup error-handling pattern.
153-153: 📐 Maintainability & Code Quality | 🔵 Trivial | 💤 Low valueUnresolved TODO on queue capacity.
The
bounded(8192)capacity is hard-coded with a TODO to make it configurable/unbounded. A full channel will blockload()senders (handled) but also caps in-flight tasks per type. Want me to wire this toConfigor open a tracking issue?🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@enginelib/src/api.rs` at line 153, Resolve the queue-capacity TODO in the channel initialization within the API setup: replace the hard-coded 8192 value with the appropriate Config-backed capacity, or make the channel unbounded if that is the intended behavior. Remove the TODO comment and ensure the existing load sender and receiver flow continues using the selected channel.enginelib/src/prelude.rs (1)
13-13: 📐 Maintainability & Code Quality | 🔵 Trivial | 💤 Low valueDelete the stale
AdminAuthEventprelude export comment The symbol is still used internally, but it is no longer part of the public prelude, so the commented-out re-export can be removed.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@enginelib/src/prelude.rs` at line 13, Remove the stale commented-out AdminAuthEvent re-export from the prelude module; leave the internal AdminAuthEvent implementation and usages unchanged.
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Inline comments:
In `@engine/src/bin/server.rs`:
- Around line 4-6: Update the main server startup flow around ServerAPI::init so
the process remains running by restoring the enginelib::server::RPC construction
and RPC::run invocation. If this entrypoint is intentionally a stub, replace the
commented code with an explicit stub that preserves the intended non-exiting
behavior.
In `@enginelib/src/api.rs`:
- Around line 166-169: The LeasedTaskQueue background reaper currently sleeps
for the full lease TTL, delaying startup and allowing reclamation to lag
significantly. Update the loop surrounding LeasedTaskQueue::reap_expired to run
at a fraction of LEASE_TTL_SECS, such as one-sixth of the TTL, while preserving
the existing periodic reaping behavior.
In `@enginelib/src/nucleus/cancel.rs`:
- Around line 6-28: Update cancel() so the successful api.db.delete(...)
operation also contributes to removed, allowing cancellation of tasks that exist
only in RocksDB to return Ok(()). Preserve the existing leased_tasks and
task_queue removal checks and continue returning TaskNotFound only when none of
those removals or the database deletion indicates a record was removed.
In `@enginelib/src/nucleus/complete.rs`:
- Around line 32-45: Make the finished-record persistence and pending-record
deletion in the complete flow atomic by replacing the separate api.db.put and
api.db.delete calls with a single WriteBatch commit containing both operations.
Preserve the existing error propagation through Error::io_error, and only clear
the dedup entry after the atomic batch succeeds so partial failure leaves the
pending state recoverable.
---
Outside diff comments:
In `@enginelib/src/event.rs`:
- Around line 126-132: Clone the handler Arc list before dispatching in the
event handling loop around event_handler_registry.event_handlers.get(&id).
Release the DashMap read guard before iterating, then apply the existing
cancellation filtering and handler.handle(event) calls to the cloned list so
callbacks can safely register handlers reentrantly.
---
Nitpick comments:
In `@enginelib/src/api.rs`:
- Around line 47-51: Update the database initialization around DB::open to
handle startup failures without an opaque unwrap panic. Preserve the
"./engine_db" path, but propagate or report the open error with both the path
and underlying error details before terminating according to the existing
startup error-handling pattern.
- Line 153: Resolve the queue-capacity TODO in the channel initialization within
the API setup: replace the hard-coded 8192 value with the appropriate
Config-backed capacity, or make the channel unbounded if that is the intended
behavior. Remove the TODO comment and ensure the existing load sender and
receiver flow continues using the selected channel.
In `@enginelib/src/prelude.rs`:
- Line 13: Remove the stale commented-out AdminAuthEvent re-export from the
prelude module; leave the internal AdminAuthEvent implementation and usages
unchanged.
🪄 Autofix (Beta)
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: defaults
Review profile: CHILL
Plan: Pro
Run ID: b0e4239c-c0b9-440d-8fe7-ca69692e9b69
⛔ Files ignored due to path filters (1)
Cargo.lockis excluded by!**/*.lock
📒 Files selected for processing (48)
.github/workflows/rust.ymlCargo.tomlengine/Cargo.tomlengine/build.rsengine/proto/engine.protoengine/src/bin/client.rsengine/src/bin/packer.rsengine/src/bin/server.rsengine/src/lib.rsenginelib/Cargo.tomlenginelib/src/api.rsenginelib/src/config.rsenginelib/src/error.rsenginelib/src/event.rsenginelib/src/events/admin_auth_event.rsenginelib/src/events/auth_event.rsenginelib/src/events/before_task_acquire_event.rsenginelib/src/events/before_task_block_acquire_event.rsenginelib/src/events/before_task_block_execute_event.rsenginelib/src/events/before_task_block_publish_event.rsenginelib/src/events/before_task_execute_event.rsenginelib/src/events/before_task_publish_event.rsenginelib/src/events/cgrpc_event.rsenginelib/src/events/mod.rsenginelib/src/events/server_task_block_acquired_event.rsenginelib/src/events/server_task_block_created_event.rsenginelib/src/events/server_task_block_published_event.rsenginelib/src/events/server_task_created_event.rsenginelib/src/events/task_acquired_event.rsenginelib/src/events/task_block_acquired_event.rsenginelib/src/lib.rsenginelib/src/nucleus/cancel.rsenginelib/src/nucleus/complete.rsenginelib/src/nucleus/lease.rsenginelib/src/nucleus/mod.rsenginelib/src/nucleus/query.rsenginelib/src/nucleus/renew.rsenginelib/src/nucleus/submit.rsenginelib/src/plugin.rsenginelib/src/prelude.rsenginelib/src/task.rsenginelib/tests/event_tests.rsrfc/rfc1000.mdrfc/rfc1001.mdrfc/rfc1002.mdrfc/rfc1003.mdrfc/rfc1004.mdrfc/rfc1005.md
💤 Files with no reviewable changes (18)
- engine/build.rs
- rfc/rfc1005.md
- rfc/rfc1004.md
- enginelib/src/events/before_task_publish_event.rs
- rfc/rfc1000.md
- engine/src/lib.rs
- engine/src/bin/client.rs
- rfc/rfc1002.md
- rfc/rfc1003.md
- enginelib/tests/event_tests.rs
- enginelib/src/events/task_acquired_event.rs
- engine/src/bin/packer.rs
- engine/proto/engine.proto
- enginelib/src/events/before_task_acquire_event.rs
- enginelib/src/events/before_task_execute_event.rs
- enginelib/src/events/server_task_created_event.rs
- rfc/rfc1001.md
- .github/workflows/rust.yml
Comment on lines
+4
to
+6
| let mut api = ServerAPI::init(); | ||
| // let server = enginelib::server::RPC::new(&api); | ||
| // server.run(); |
Contributor
There was a problem hiding this comment.
📐 Maintainability & Code Quality | 🟡 Minor | ⚡ Quick win
🧩 Analysis chain
🏁 Script executed:
#!/bin/bash
# Check CI for a deny-warnings policy that would fail on unused_mut/unused_variables
fd -H -t f 'rust.yml' .github
rg -nP -C2 '(-D\s*warnings|RUSTFLAGS|deny\(warnings\)|clippy)' .github || echo "No explicit deny-warnings found"Repository: voltaero/engine
Length of output: 212
🏁 Script executed:
#!/bin/bash
set -euo pipefail
echo '--- engine/src/bin/server.rs ---'
cat -n engine/src/bin/server.rs
echo
echo '--- .github/workflows/rust.yml ---'
cat -n .github/workflows/rust.ymlRepository: voltaero/engine
Length of output: 1502
main exits immediately. With RPC::run() still commented out, the binary returns as soon as ServerAPI::init() completes, so the server never stays up. Restore the run call or leave an explicit stub if this entrypoint is intentional.
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@engine/src/bin/server.rs` around lines 4 - 6, Update the main server startup
flow around ServerAPI::init so the process remains running by restoring the
enginelib::server::RPC construction and RPC::run invocation. If this entrypoint
is intentionally a stub, replace the commented code with an explicit stub that
preserves the intended non-exiting behavior.
Comment on lines
+166
to
+169
| std::thread::spawn(move || { | ||
| loop { | ||
| std::thread::sleep(std::time::Duration::from_secs(3600)); | ||
| LeasedTaskQueue::reap_expired(&dapi); |
Contributor
There was a problem hiding this comment.
🎯 Functional Correctness | 🟡 Minor | ⚡ Quick win
Reap interval equals the lease TTL — expiry can lag up to ~2× TTL.
The background loop sleeps LEASE_TTL_SECS (3600s) and only then reaps, so an expired lease is reclaimed anywhere from 0 to ~3600s after it expires (and the first sweep is delayed a full hour after startup). If timely reclamation matters for task re-leasing, run the sweep at a fraction of the TTL (e.g. TTL/6).
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@enginelib/src/api.rs` around lines 166 - 169, The LeasedTaskQueue background
reaper currently sleeps for the full lease TTL, delaying startup and allowing
reclamation to lag significantly. Update the loop surrounding
LeasedTaskQueue::reap_expired to run at a fraction of LEASE_TTL_SECS, such as
one-sixth of the TTL, while preserving the existing periodic reaping behavior.
Comment on lines
+6
to
+28
| pub fn cancel(api: Arc<ServerAPI>, task_type: Identifier, task_id: String) -> Result<(), Error> { | ||
| let mut removed = false; | ||
| if let Some(mut leases) = api.leased_tasks.tasks.get_mut(&task_type) { | ||
| let before = leases.len(); | ||
| leases.retain(|lease| lease.stored_task.task_id != task_id); | ||
| removed = leases.len() != before; | ||
| } | ||
|
|
||
| // Delete the persisted record before clearing the dedup entry so a | ||
| // concurrent load() cannot re-enqueue the cancelled task. | ||
| api.db | ||
| .delete(task_key(&task_type, &task_id)) | ||
| .map_err(|err| Error::io_error(format!("Failed to delete task record: {err}")))?; | ||
| if let Some(queue) = api.task_queue.tasks.get(&task_type) { | ||
| removed |= queue.2.remove(&task_id).is_some(); | ||
| } | ||
|
|
||
| if removed { | ||
| Ok(()) | ||
| } else { | ||
| Err(Error::not_found("TaskNotFound")) | ||
| } | ||
| } |
Contributor
There was a problem hiding this comment.
🎯 Functional Correctness | 🟠 Major | ⚡ Quick win
cancel() can return TaskNotFound even though it just deleted a real record.
removed is derived only from leased_tasks membership and the queue dedup set. A task that was submit()-ted but not yet scanned into memory by load() exists only in RocksDB — api.db.delete(...) will remove it, but removed never reflects that, so the function still returns Err(NotFound) despite the successful cancellation.
🔧 Proposed fix: derive `removed` from the DB delete outcome too
pub fn cancel(api: Arc<ServerAPI>, task_type: Identifier, task_id: String) -> Result<(), Error> {
let mut removed = false;
if let Some(mut leases) = api.leased_tasks.tasks.get_mut(&task_type) {
let before = leases.len();
leases.retain(|lease| lease.stored_task.task_id != task_id);
removed = leases.len() != before;
}
// Delete the persisted record before clearing the dedup entry so a
// concurrent load() cannot re-enqueue the cancelled task.
- api.db
- .delete(task_key(&task_type, &task_id))
- .map_err(|err| Error::io_error(format!("Failed to delete task record: {err}")))?;
+ let key = task_key(&task_type, &task_id);
+ let existed = api
+ .db
+ .get(&key)
+ .map_err(|err| Error::io_error(format!("Failed to check task record: {err}")))?
+ .is_some();
+ api.db
+ .delete(&key)
+ .map_err(|err| Error::io_error(format!("Failed to delete task record: {err}")))?;
+ removed |= existed;
if let Some(queue) = api.task_queue.tasks.get(&task_type) {
removed |= queue.2.remove(&task_id).is_some();
}📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
Suggested change
| pub fn cancel(api: Arc<ServerAPI>, task_type: Identifier, task_id: String) -> Result<(), Error> { | |
| let mut removed = false; | |
| if let Some(mut leases) = api.leased_tasks.tasks.get_mut(&task_type) { | |
| let before = leases.len(); | |
| leases.retain(|lease| lease.stored_task.task_id != task_id); | |
| removed = leases.len() != before; | |
| } | |
| // Delete the persisted record before clearing the dedup entry so a | |
| // concurrent load() cannot re-enqueue the cancelled task. | |
| api.db | |
| .delete(task_key(&task_type, &task_id)) | |
| .map_err(|err| Error::io_error(format!("Failed to delete task record: {err}")))?; | |
| if let Some(queue) = api.task_queue.tasks.get(&task_type) { | |
| removed |= queue.2.remove(&task_id).is_some(); | |
| } | |
| if removed { | |
| Ok(()) | |
| } else { | |
| Err(Error::not_found("TaskNotFound")) | |
| } | |
| } | |
| pub fn cancel(api: Arc<ServerAPI>, task_type: Identifier, task_id: String) -> Result<(), Error> { | |
| let mut removed = false; | |
| if let Some(mut leases) = api.leased_tasks.tasks.get_mut(&task_type) { | |
| let before = leases.len(); | |
| leases.retain(|lease| lease.stored_task.task_id != task_id); | |
| removed = leases.len() != before; | |
| } | |
| // Delete the persisted record before clearing the dedup entry so a | |
| // concurrent load() cannot re-enqueue the cancelled task. | |
| let key = task_key(&task_type, &task_id); | |
| let existed = api | |
| .db | |
| .get(&key) | |
| .map_err(|err| Error::io_error(format!("Failed to check task record: {err}")))? | |
| .is_some(); | |
| api.db | |
| .delete(&key) | |
| .map_err(|err| Error::io_error(format!("Failed to delete task record: {err}")))?; | |
| removed |= existed; | |
| if let Some(queue) = api.task_queue.tasks.get(&task_type) { | |
| removed |= queue.2.remove(&task_id).is_some(); | |
| } | |
| if removed { | |
| Ok(()) | |
| } else { | |
| Err(Error::not_found("TaskNotFound")) | |
| } | |
| } |
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@enginelib/src/nucleus/cancel.rs` around lines 6 - 28, Update cancel() so the
successful api.db.delete(...) operation also contributes to removed, allowing
cancellation of tasks that exist only in RocksDB to return Ok(()). Preserve the
existing leased_tasks and task_queue removal checks and continue returning
TaskNotFound only when none of those removals or the database deletion indicates
a record was removed.
Comment on lines
+32
to
+45
| // Persist the finished record before deleting the pending one so a crash | ||
| // between the two writes never loses the result. | ||
| api.db | ||
| .put(finished_task_key(&task_type, &task_id), task_bytes) | ||
| .map_err(|err| Error::io_error(format!("Failed to persist finished task: {err}")))?; | ||
| api.db | ||
| .delete(task_key(&task_type, &task_id)) | ||
| .map_err(|err| Error::io_error(format!("Failed to delete task record: {err}")))?; | ||
|
|
||
| // Clear the dedup entry last so a concurrent load() cannot re-enqueue a task | ||
| // whose pending record is already gone. | ||
| if let Some(queue) = api.task_queue.tasks.get(&task_type) { | ||
| queue.2.remove(&task_id); | ||
| } |
Contributor
There was a problem hiding this comment.
🗄️ Data Integrity & Integration | 🟠 Major | ⚡ Quick win
Make the finish+cleanup write atomic to avoid a stuck duplicate record on partial failure.
put(finished_task_key) and delete(task_key) are separate calls. If the put succeeds but the delete fails, complete() returns Err, yet the lease was already dropped a few lines earlier — so a retry will fail with TaskNotLeased, and the stale pending record is left behind with no self-healing path (only recoverable via cancel() or an operator Query::PurgePrefix).
🔧 Proposed fix using an atomic WriteBatch
- api.db
- .put(finished_task_key(&task_type, &task_id), task_bytes)
- .map_err(|err| Error::io_error(format!("Failed to persist finished task: {err}")))?;
- api.db
- .delete(task_key(&task_type, &task_id))
- .map_err(|err| Error::io_error(format!("Failed to delete task record: {err}")))?;
+ let mut batch = rust_rocksdb::WriteBatch::default();
+ batch.put(finished_task_key(&task_type, &task_id), task_bytes);
+ batch.delete(task_key(&task_type, &task_id));
+ api.db
+ .write(&batch)
+ .map_err(|err| Error::io_error(format!("Failed to persist finished task: {err}")))?;📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
Suggested change
| // Persist the finished record before deleting the pending one so a crash | |
| // between the two writes never loses the result. | |
| api.db | |
| .put(finished_task_key(&task_type, &task_id), task_bytes) | |
| .map_err(|err| Error::io_error(format!("Failed to persist finished task: {err}")))?; | |
| api.db | |
| .delete(task_key(&task_type, &task_id)) | |
| .map_err(|err| Error::io_error(format!("Failed to delete task record: {err}")))?; | |
| // Clear the dedup entry last so a concurrent load() cannot re-enqueue a task | |
| // whose pending record is already gone. | |
| if let Some(queue) = api.task_queue.tasks.get(&task_type) { | |
| queue.2.remove(&task_id); | |
| } | |
| // Persist the finished record before deleting the pending one so a crash | |
| // between the two writes never loses the result. | |
| let mut batch = rust_rocksdb::WriteBatch::default(); | |
| batch.put(finished_task_key(&task_type, &task_id), task_bytes); | |
| batch.delete(task_key(&task_type, &task_id)); | |
| api.db | |
| .write(&batch) | |
| .map_err(|err| Error::io_error(format!("Failed to persist finished task: {err}")))?; | |
| // Clear the dedup entry last so a concurrent load() cannot re-enqueue a task | |
| // whose pending record is already gone. | |
| if let Some(queue) = api.task_queue.tasks.get(&task_type) { | |
| queue.2.remove(&task_id); | |
| } |
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@enginelib/src/nucleus/complete.rs` around lines 32 - 45, Make the
finished-record persistence and pending-record deletion in the complete flow
atomic by replacing the separate api.db.put and api.db.delete calls with a
single WriteBatch commit containing both operations. Preserve the existing error
propagation through Error::io_error, and only clear the dedup entry after the
atomic batch succeeds so partial failure leaves the pending state recoverable.
Contributor
There was a problem hiding this comment.
Actionable comments posted: 2
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Inline comments:
In `@enginelib/src/nucleus/cancel.rs`:
- Around line 16-34: Validate that the task type exists in api.task_queue.tasks
before entering the lease-removal block in the cancellation flow. Update the
code around the stored_task extraction and queue lookup so
task_queue.tasks.get(&task_type) returns TaskTypeNotFound without mutating
leased_tasks; preserve the existing lease removal and requeue behavior after
validation succeeds.
- Around line 5-46: Narrow the `cancel` documentation and contract to leased
tasks only, explicitly stating that queued-but-unleased tasks are not removed;
update the `Query::Delete` guidance to direct pending queued-task deletion
through the appropriate separate path rather than `cancel` or `complete`.
Preserve the existing lease removal and requeue behavior in `cancel`.
🪄 Autofix (Beta)
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: defaults
Review profile: CHILL
Plan: Pro
Run ID: 88dca659-412a-4567-be7b-1caf78129adf
📒 Files selected for processing (5)
enginelib/src/api.rsenginelib/src/lib.rsenginelib/src/nucleus/cancel.rsenginelib/src/nucleus/query.rsenginelib/src/nucleus/submit.rs
🚧 Files skipped from review as they are similar to previous changes (2)
- enginelib/src/lib.rs
- enginelib/src/api.rs
Comment on lines
+5
to
+46
| /// Cancels the *lease* on a task, not the task itself: removes it from the lease | ||
| /// list and resends it into the queue so another worker can pick it up. | ||
| /// | ||
| /// The persisted record is never touched, and the task_id stays in the dedup set | ||
| /// the whole time, so no concurrent `load()` can enqueue a duplicate. The requeue | ||
| /// is a non-blocking `try_send` so there is no await window in which the lease is | ||
| /// gone but the task isn't yet back in the channel: on success the task moves | ||
| /// straight from "leased" to "queued"; if the channel is full or closed we clear | ||
| /// the dedup entry instead, dropping the task back to "persisted only" so a later | ||
| /// `load()` recovers it. Either way the task is never stranded. | ||
| pub fn cancel(api: Arc<ServerAPI>, task_type: Identifier, task_id: String) -> Result<(), Error> { | ||
| // Remove the lease, capturing the stored task so we can requeue it. | ||
| let stored_task = { | ||
| let mut leases = api | ||
| .leased_tasks | ||
| .tasks | ||
| .get_mut(&task_type) | ||
| .ok_or(Error::not_found("TaskNotFound"))?; | ||
| let idx = leases | ||
| .iter() | ||
| .position(|lease| lease.stored_task.task_id == task_id) | ||
| .ok_or(Error::not_found("TaskNotFound"))?; | ||
| leases.swap_remove(idx).stored_task | ||
| }; | ||
|
|
||
| let queue = api | ||
| .task_queue | ||
| .tasks | ||
| .get(&task_type) | ||
| .ok_or(Error::not_found("TaskTypeNotFound"))?; | ||
|
|
||
| // Non-blocking requeue: no await means the lease removal and the requeue | ||
| // can't be split by an async cancellation. | ||
| match queue.0.try_send((*stored_task).clone()) { | ||
| // Queued again; task_id stays in the dedup set (still in-flight). | ||
| Ok(()) => Ok(()), | ||
| // Channel full/closed: clear the dedup entry so the still-persisted | ||
| // record is re-enqueued by a later load(). Never left stranded. | ||
| Err(_) => { | ||
| queue.2.remove(&task_id); | ||
| Ok(()) | ||
| } |
Contributor
There was a problem hiding this comment.
🎯 Functional Correctness | 🟡 Minor | ⚡ Quick win
🧩 Analysis chain
🏁 Script executed:
#!/bin/bash
rg -n -B2 -A15 '\bfn cancel\b|\bcancel\(' enginelib/src/nucleus/mod.rs enginelib/src -g '!nucleus/cancel.rs'Repository: voltaero/engine
Length of output: 2488
🏁 Script executed:
#!/bin/bash
set -euo pipefail
printf '\n== nucleus mod outline ==\n'
ast-grep outline enginelib/src/nucleus/mod.rs --view expanded || true
printf '\n== cancel.rs ==\n'
cat -n enginelib/src/nucleus/cancel.rs | sed -n '1,120p'
printf '\n== query.rs relevant lines ==\n'
rg -n -A12 -B8 'Delete|cancel|complete|pending task|queued or leased|dedup' enginelib/src/nucleus/query.rs
printf '\n== nucleus mod references to cancel ==\n'
rg -n -A8 -B8 '\bcancel\b' enginelib/src/nucleus/mod.rs enginelib/src -g '!enginelib/src/nucleus/cancel.rs' -g '!**/target/**'Repository: voltaero/engine
Length of output: 10104
🏁 Script executed:
#!/bin/bash
set -euo pipefail
printf '\n== file list in nucleus ==\n'
git ls-files enginelib/src/nucleus
printf '\n== transport/admin caller search ==\n'
rg -n -A4 -B4 'TaskNotFound|TaskTypeNotFound|cancel\(' enginelib/src -g '!enginelib/src/nucleus/cancel.rs' -g '!**/target/**'Repository: voltaero/engine
Length of output: 4067
🏁 Script executed:
#!/bin/bash
set -euo pipefail
printf '\n== inspect mod.rs and query.rs sizes ==\n'
wc -l enginelib/src/nucleus/mod.rs enginelib/src/nucleus/query.rs enginelib/src/nucleus/cancel.rs
printf '\n== mod.rs slice ==\n'
cat -n enginelib/src/nucleus/mod.rs | sed -n '1,220p'
printf '\n== query.rs slice ==\n'
cat -n enginelib/src/nucleus/query.rs | sed -n '1,220p'Repository: voltaero/engine
Length of output: 9952
🏁 Script executed:
#!/bin/bash
set -euo pipefail
printf '\n== search for API handlers exposing cancel/complete/delete ==\n'
rg -n -A6 -B6 '\b(cancel|complete|delete)\b' enginelib/src -g '!**/target/**' | sed -n '1,220p'Repository: voltaero/engine
Length of output: 8849
cancel() only releases leased tasks back into the queue; it does not remove queued-but-unleased tasks. The Query::Delete docs still point callers at cancel/complete for pending tasks, so the contract should be narrowed or the queued case handled separately.
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@enginelib/src/nucleus/cancel.rs` around lines 5 - 46, Narrow the `cancel`
documentation and contract to leased tasks only, explicitly stating that
queued-but-unleased tasks are not removed; update the `Query::Delete` guidance
to direct pending queued-task deletion through the appropriate separate path
rather than `cancel` or `complete`. Preserve the existing lease removal and
requeue behavior in `cancel`.
Comment on lines
+16
to
+34
| // Remove the lease, capturing the stored task so we can requeue it. | ||
| let stored_task = { | ||
| let mut leases = api | ||
| .leased_tasks | ||
| .tasks | ||
| .get_mut(&task_type) | ||
| .ok_or(Error::not_found("TaskNotFound"))?; | ||
| let idx = leases | ||
| .iter() | ||
| .position(|lease| lease.stored_task.task_id == task_id) | ||
| .ok_or(Error::not_found("TaskNotFound"))?; | ||
| leases.swap_remove(idx).stored_task | ||
| }; | ||
|
|
||
| let queue = api | ||
| .task_queue | ||
| .tasks | ||
| .get(&task_type) | ||
| .ok_or(Error::not_found("TaskTypeNotFound"))?; |
Contributor
There was a problem hiding this comment.
🗄️ Data Integrity & Integration | 🟠 Major | ⚡ Quick win
Lease is removed before the queue existence check — can strand a task.
leases.swap_remove(idx) (lines 17-28) mutates state before the fallible task_queue.tasks.get(&task_type) lookup (lines 30-34). If that lookup returns Err(TaskTypeNotFound), the lease is already gone, the task was never requeued, and the dedup entry is untouched — the task is now neither leased, queued, nor recoverable by a later load(). This directly contradicts the function's own doc comment ("Either way the task is never stranded", lines 12-14). Currently only reachable if task_queue and leased_tasks diverge for a task_type (they're populated together in ServerAPI::populate), but the fix is cheap and removes a latent landmine.
🔧 Proposed fix: validate queue existence before mutating lease state
pub fn cancel(api: Arc<ServerAPI>, task_type: Identifier, task_id: String) -> Result<(), Error> {
+ let queue = api
+ .task_queue
+ .tasks
+ .get(&task_type)
+ .ok_or(Error::not_found("TaskTypeNotFound"))?;
+
// Remove the lease, capturing the stored task so we can requeue it.
let stored_task = {
let mut leases = api
.leased_tasks
.tasks
.get_mut(&task_type)
.ok_or(Error::not_found("TaskNotFound"))?;
let idx = leases
.iter()
.position(|lease| lease.stored_task.task_id == task_id)
.ok_or(Error::not_found("TaskNotFound"))?;
leases.swap_remove(idx).stored_task
};
- let queue = api
- .task_queue
- .tasks
- .get(&task_type)
- .ok_or(Error::not_found("TaskTypeNotFound"))?;
-
// Non-blocking requeue: no await means the lease removal and the requeue
// can't be split by an async cancellation.
match queue.0.try_send((*stored_task).clone()) {📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
Suggested change
| // Remove the lease, capturing the stored task so we can requeue it. | |
| let stored_task = { | |
| let mut leases = api | |
| .leased_tasks | |
| .tasks | |
| .get_mut(&task_type) | |
| .ok_or(Error::not_found("TaskNotFound"))?; | |
| let idx = leases | |
| .iter() | |
| .position(|lease| lease.stored_task.task_id == task_id) | |
| .ok_or(Error::not_found("TaskNotFound"))?; | |
| leases.swap_remove(idx).stored_task | |
| }; | |
| let queue = api | |
| .task_queue | |
| .tasks | |
| .get(&task_type) | |
| .ok_or(Error::not_found("TaskTypeNotFound"))?; | |
| let queue = api | |
| .task_queue | |
| .tasks | |
| .get(&task_type) | |
| .ok_or(Error::not_found("TaskTypeNotFound"))?; | |
| // Remove the lease, capturing the stored task so we can requeue it. | |
| let stored_task = { | |
| let mut leases = api | |
| .leased_tasks | |
| .tasks | |
| .get_mut(&task_type) | |
| .ok_or(Error::not_found("TaskNotFound"))?; | |
| let idx = leases | |
| .iter() | |
| .position(|lease| lease.stored_task.task_id == task_id) | |
| .ok_or(Error::not_found("TaskNotFound"))?; | |
| leases.swap_remove(idx).stored_task | |
| }; | |
| // Non-blocking requeue: no await means the lease removal and the requeue | |
| // can't be split by an async cancellation. | |
| match queue.0.try_send((*stored_task).clone()) { |
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@enginelib/src/nucleus/cancel.rs` around lines 16 - 34, Validate that the task
type exists in api.task_queue.tasks before entering the lease-removal block in
the cancellation flow. Update the code around the stored_task extraction and
queue lookup so task_queue.tasks.get(&task_type) returns TaskTypeNotFound
without mutating leased_tasks; preserve the existing lease removal and requeue
behavior after validation succeeds.
Open
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
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
1 participant
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.
Goal: Establish a Core base for multiple transport methods like TCP, GRPC etc.
Summary by CodeRabbit
.rfpackages.auth_token.