Conversation
📝 WalkthroughWalkthroughAdds a serialized batch protocol, bounded asynchronous TCP transport, task queue recovery and lease requeue logic, CLI and benchmark binaries, and end-to-end, recovery, and transport-boundary tests. ChangesEngine transport and task lifecycle
Estimated code review effort: 5 (Critical) | ~120 minutes Sequence Diagram(s)sequenceDiagram
participant Client
participant EngineServer
participant ServerAPI
participant RocksDB
Client->>EngineServer: Submit, lease, and complete encoded requests
EngineServer->>ServerAPI: Authenticate and dispatch operations
ServerAPI->>RocksDB: Persist pending and finished task records
RocksDB-->>ServerAPI: Return operation results
ServerAPI-->>EngineServer: Build protocol response
EngineServer-->>Client: Return encoded response
Possibly related PRs
🚥 Pre-merge checks | ✅ 4 | ❌ 1❌ Failed checks (1 inconclusive)
✅ Passed checks (4 passed)
✨ Finishing Touches📝 Generate docstrings
🧪 Generate unit tests (beta)
Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out. Comment |
Greptile SummaryThis PR adds a TCP task transport and expands the distributed task lifecycle. The main changes are:
Confidence Score: 4/5The completion durability issue should be fixed before merging.
enginelib/src/nucleus/complete.rs Important Files Changed
Reviews (2): Last reviewed commit: "fix: missing pfp" | Re-trigger Greptile |
![greptile-apps[bot]](https://avatars.githubusercontent.com/in/867647?s=60&v=4){kind=small}
| if !Events::CheckAuth(&api, String::new(), auth, api.db.clone()) { | ||
| return unauthorized(); | ||
| } | ||
| match complete::complete_batch(api, task_type, results) { | ||
| Ok(ok) => Response::Completed { ok }, | ||
| Err(e) => Response::Err(e), | ||
| } | ||
| } | ||
| Request::Renew { task_type, task_id } => { | ||
| if !Events::CheckAuth(&api, String::new(), auth, api.db.clone()) { | ||
| return unauthorized(); | ||
| } | ||
| match renew::renew(api, task_type, &task_id) { | ||
| Ok(()) => Response::Renewed, | ||
| Err(e) => Response::Err(e), | ||
| } | ||
| } | ||
| Request::Cancel { task_type, task_id } => { | ||
| if !Events::CheckAuth(&api, String::new(), auth, api.db.clone()) { | ||
| return unauthorized(); | ||
| } | ||
| match cancel::cancel(api, task_type, task_id) { | ||
| Ok(()) => Response::Cancelled, | ||
| Err(e) => Response::Err(e), | ||
| } |
There was a problem hiding this comment.
These operations authenticate without the lease's user_id, while the nucleus finds leases only by task type and task ID. A different connected worker that learns an ID can complete it with forged bytes, cancel it, or keep renewing it even though another worker owns the lease.
Context Used: AGENTS.md (source)
| let first = match tokio::time::timeout( | ||
| std::time::Duration::from_secs(LEASE_LONG_POLL_SECS), | ||
| receiver.recv(), | ||
| ) | ||
| .await | ||
| { | ||
| Err(_elapsed) => return Ok(Vec::new()), | ||
| Ok(recv) => recv.map_err(|err| Error::new(format!("Task queue closed: {err}")))?, | ||
| }; | ||
| out.push(Arc::new(first)); | ||
| while (out.len() as u32) < max { | ||
| match receiver.try_recv() { | ||
| Ok(task) => out.push(Arc::new(task)), | ||
| Err(_) => break, | ||
| } | ||
| } | ||
|
|
||
| { |
There was a problem hiding this comment.
Tasks are removed from the channel before their lease records are inserted. If this future is cancelled after recv or during the batch drain, those tasks remain in the dedup set but are neither queued nor leased, so the loader skips them indefinitely.
Context Used: AGENTS.md (source)
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 6
🧹 Nitpick comments (6)
enginelib/src/protocol.rs (1)
78-80: 🔒 Security & Privacy | 🔵 Trivial | ⚡ Quick winConsider whether raw DB/IO error text should reach clients.
Response::Err(Error)forwardsError'smessageverbatim; call sites elsewhere (e.g.Error::io_error(format!("Failed to persist task batch: {err}"))) embed underlying RocksDB error text, which now flows straight to the wire client via this serializableError.🤖 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/protocol.rs` around lines 78 - 80, Review the Response::Err(Error) serialization path and prevent raw database or I/O error details embedded in Error::message from being exposed to wire clients. Sanitize or map internal failures to a stable client-safe message before constructing the response, while preserving useful server-side error context at the originating call sites such as Error::io_error.engine/src/client.rs (2)
21-27: 🚀 Performance & Scalability | 🔵 Trivial | ⚡ Quick winConsider sharing a
Contextacross connections.Each
Client::connectspins up its ownzmq::Context, which typically starts its own I/O thread(s). Inbench.rsthis is instantiated once per worker (up to dozens), multiplying OS threads unnecessarily. Exposing an optional shared-context constructor (e.g.connect_with_context(&Context, ...)) alongsideconnectwould let high-fanout callers like the benchmark share one context.🤖 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/client.rs` around lines 21 - 27, Update Client::connect to support context reuse by adding a public connect_with_context constructor that accepts a shared zmq::Context and performs the existing socket connection and error mapping. Keep connect as a compatibility wrapper that creates its own Context and delegates to the new constructor, allowing callers such as bench.rs to share one context without changing existing behavior.
30-52: 🩺 Stability & Availability | 🔵 Trivial | ⚡ Quick winNo timeout around
send/recv— a stalled peer hangs the call forever.
callawaitssendthennextwith no timeout. Any caller not wrapping this externally (e.g.bin/client.rs) will block indefinitely on a network partition or unresponsive server.🛠️ Sketch
- self.sock - .send(mp) - .await - .map_err(|e| Error::io_error(format!("send failed: {e}")))?; + tokio::time::timeout(SEND_TIMEOUT, self.sock.send(mp)) + .await + .map_err(|_| Error::io_error("send timed out".into()))? + .map_err(|e| Error::io_error(format!("send failed: {e}")))?;🤖 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/client.rs` around lines 30 - 52, Update Client::call to wrap both the socket send and reply receive awaits with the established request timeout, ensuring stalled peers return a timeout error instead of blocking indefinitely. Preserve the existing send/receive error mapping and response decoding behavior.engine/tests/e2e.rs (2)
68-75: 🩺 Stability & Availability | 🔵 Trivial | ⚡ Quick winFixed-delay sleep for server readiness is flaky.
Waiting a hardcoded 300ms for the spawned server to bind before connecting can flake under CI load. Prefer a retry-connect loop or an explicit readiness signal from
serve().🤖 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/tests/e2e.rs` around lines 68 - 75, Replace the fixed 300ms sleep before Client::connect with reliable server-readiness synchronization: either await an explicit readiness signal from serve() or retry Client::connect until it succeeds, using bounded retries or a timeout. Preserve the existing background server startup and only proceed once the endpoint is ready.
1-1: 🩺 Stability & Availability | 🔵 Trivial | ⚡ Quick winFixed-delay sleeps used as readiness checks in both the e2e test and the benchmark share the same flakiness risk. Both wait a hardcoded duration for a server (in-process task or child process) to finish binding before connecting, rather than confirming readiness.
engine/tests/e2e.rs#L68-75: replace the 300mstokio::time::sleepwith a retry-connect loop (or a readiness channel signaled onceserve()has bound the socket) before callingClient::connect.engine/src/bin/bench.rs#L42-49: replace the 800ms sleep waiting on the spawnedbench_serverprocess with a retry-connect loop for the submitter's first connection attempt.🤖 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/tests/e2e.rs` at line 1, Replace the fixed-delay sleeps in the e2e test and benchmark startup paths with readiness-based connection handling: in the e2e server task around serve() and Client::connect, retry connecting or signal readiness only after binding succeeds; in bench_server startup, retry the submitter’s initial connection instead of sleeping 800ms. Preserve the existing connection behavior once readiness is confirmed.engine/src/bin/bench.rs (1)
42-49: 🩺 Stability & Availability | 🔵 Trivial | ⚡ Quick winFixed-delay sleep to wait for
bench_serverto bind is flaky.800ms is a guess; a slow-starting child process (e.g. RocksDB open under contention) can still race the submitter's first connect attempt. Prefer a retry-connect loop instead of a blind sleep.
🤖 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/bench.rs` around lines 42 - 49, Replace the fixed 800ms sleep after spawning bench_server with a retry-connect loop that repeatedly attempts to connect to the server’s configured endpoint, waiting briefly between attempts and stopping once the connection succeeds or a bounded timeout is reached. Use the existing server/submitter connection symbols and preserve the subsequent benchmark flow only after readiness is confirmed.
🤖 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/Cargo.toml`:
- Line 36: Replace the machine-specific absolute path in the engine_core
dependency declaration with the correct repository-relative path to the local
engine_core crate, using the Rust workspace layout so cargo build --workspace
works for all contributors and CI.
In `@engine/src/bin/bench.rs`:
- Around line 96-119: Protect the submission phase in the `submit_task` flow
from hanging indefinitely: apply the same stall/timeout mechanism used by the
completion-wait loop before awaiting the spawned submitter, and preserve the
benchmark’s `stalled` outcome when the submission deadline is exceeded. Ensure
`submit_task.await` cannot block forever if `client.submit` or the server
connection stalls.
In `@engine/src/server.rs`:
- Around line 33-73: Update the serve loop’s reply path around reply_tx/reply_rx
and socket.send so reply production cannot grow memory without bound and a slow
peer cannot block request handling indefinitely. Use a bounded reply channel
with handler-side backpressure, and apply an appropriate timeout or equivalent
cancellation around sending replies while preserving identity-based routing and
existing error logging.
- Line 1: Add matching transport encryption to the ROUTER/DEALER socket setup in
the server initialization and corresponding client connection flow, using the
project’s supported CURVE or TLS mechanism. Configure both peers with compatible
security credentials before connecting or binding, while preserving the existing
request dispatch and connection behavior.
In `@enginelib/src/api.rs`:
- Around line 195-260: Update the send-failure branch in load_after to mirror
load()'s cleanup: remove dedup entries for the failed task and every remaining
unsent task in batch, then set the task state rescan flag with the existing
Ordering::Release semantics before returning the queue-closed error. Preserve
the existing queued/executing/solved synchronization and load_after success
behavior.
In `@enginelib/src/protocol.rs`:
- Around line 20-27: Replace the derived Debug implementation on Envelope with a
custom implementation that preserves useful request details while masking the
auth bearer token. Keep Clone, Serialize, and Deserialize derives unchanged, and
ensure formatting Envelope with {:?} never exposes the auth value.
---
Nitpick comments:
In `@engine/src/bin/bench.rs`:
- Around line 42-49: Replace the fixed 800ms sleep after spawning bench_server
with a retry-connect loop that repeatedly attempts to connect to the server’s
configured endpoint, waiting briefly between attempts and stopping once the
connection succeeds or a bounded timeout is reached. Use the existing
server/submitter connection symbols and preserve the subsequent benchmark flow
only after readiness is confirmed.
In `@engine/src/client.rs`:
- Around line 21-27: Update Client::connect to support context reuse by adding a
public connect_with_context constructor that accepts a shared zmq::Context and
performs the existing socket connection and error mapping. Keep connect as a
compatibility wrapper that creates its own Context and delegates to the new
constructor, allowing callers such as bench.rs to share one context without
changing existing behavior.
- Around line 30-52: Update Client::call to wrap both the socket send and reply
receive awaits with the established request timeout, ensuring stalled peers
return a timeout error instead of blocking indefinitely. Preserve the existing
send/receive error mapping and response decoding behavior.
In `@engine/tests/e2e.rs`:
- Around line 68-75: Replace the fixed 300ms sleep before Client::connect with
reliable server-readiness synchronization: either await an explicit readiness
signal from serve() or retry Client::connect until it succeeds, using bounded
retries or a timeout. Preserve the existing background server startup and only
proceed once the endpoint is ready.
- Line 1: Replace the fixed-delay sleeps in the e2e test and benchmark startup
paths with readiness-based connection handling: in the e2e server task around
serve() and Client::connect, retry connecting or signal readiness only after
binding succeeds; in bench_server startup, retry the submitter’s initial
connection instead of sleeping 800ms. Preserve the existing connection behavior
once readiness is confirmed.
In `@enginelib/src/protocol.rs`:
- Around line 78-80: Review the Response::Err(Error) serialization path and
prevent raw database or I/O error details embedded in Error::message from being
exposed to wire clients. Sanitize or map internal failures to a stable
client-safe message before constructing the response, while preserving useful
server-side error context at the originating call sites such as Error::io_error.
🪄 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: 62f2cc3d-d5f5-436e-8d3c-dcf04dcf1606
⛔ Files ignored due to path filters (1)
Cargo.lockis excluded by!**/*.lock
📒 Files selected for processing (21)
engine/Cargo.tomlengine/src/bin/bench.rsengine/src/bin/bench_server.rsengine/src/bin/client.rsengine/src/bin/server.rsengine/src/client.rsengine/src/lib.rsengine/src/server.rsengine/tests/e2e.rsengine/tests/recovery.rsenginelib/macros/src/lib.rsenginelib/src/api.rsenginelib/src/error.rsenginelib/src/lib.rsenginelib/src/nucleus/cancel.rsenginelib/src/nucleus/complete.rsenginelib/src/nucleus/lease.rsenginelib/src/nucleus/query.rsenginelib/src/nucleus/submit.rsenginelib/src/prelude.rsenginelib/src/protocol.rs
| futures = "0.3" | ||
| serde = { workspace = true } | ||
| tracing = "0.1" | ||
| engine_core = { path = "/tmp/claude-1000/-home-styly-projects-personal-engine/2031244b-7e64-4017-9758-353f174e1d37/scratchpad/voltaero-mod" } |
There was a problem hiding this comment.
🎯 Functional Correctness | 🔴 Critical | ⚡ Quick win
engine_core points at a machine-local scratchpad path, not a repo path.
path = "/tmp/claude-1000/-home-styly-projects-personal-engine/.../scratchpad/voltaero-mod" is an absolute, machine-specific temp directory. cargo build --workspace will fail for anyone else (CI, other contributors) since this path won't exist off this one machine.
As per coding guidelines, "For workspace-wide changes, use the Rust workspace defined in Cargo.toml and validate with cargo build --workspace or cargo test --workspace."
🛠️ Suggested fix
-engine_core = { path = "/tmp/claude-1000/-home-styly-projects-personal-engine/2031244b-7e64-4017-9758-353f174e1d37/scratchpad/voltaero-mod" }
+engine_core = { path = "../engine_core" }Adjust the relative path to wherever engine_core actually lives in this repo/workspace.
📝 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.
| engine_core = { path = "/tmp/claude-1000/-home-styly-projects-personal-engine/2031244b-7e64-4017-9758-353f174e1d37/scratchpad/voltaero-mod" } | |
| engine_core = { path = "../engine_core" } |
🤖 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/Cargo.toml` at line 36, Replace the machine-specific absolute path in
the engine_core dependency declaration with the correct repository-relative path
to the local engine_core crate, using the Rust workspace layout so cargo build
--workspace works for all contributors and CI.
Source: Coding guidelines
| let e2e_start = Instant::now(); | ||
| let submit_start = Instant::now(); | ||
| let submit_task = { | ||
| let endpoint = endpoint.clone(); | ||
| let task_type = task_type.clone(); | ||
| let payload = payload.clone(); | ||
| tokio::spawn(async move { | ||
| let mut client = Client::connect(&endpoint, String::new()).expect("submitter connect"); | ||
| let mut sent = 0u64; | ||
| while sent < n { | ||
| let this = std::cmp::min(submit_batch as u64, n - sent) as usize; | ||
| if client | ||
| .submit(task_type.clone(), vec![payload.clone(); this]) | ||
| .await | ||
| .is_err() | ||
| { | ||
| break; | ||
| } | ||
| sent += this as u64; | ||
| } | ||
| sent | ||
| }) | ||
| }; | ||
| let submitted = submit_task.await.unwrap_or(0); |
There was a problem hiding this comment.
🩺 Stability & Availability | 🟠 Major | ⚡ Quick win
Submission phase has no stall/timeout protection, unlike the completion wait.
The file's header states the benchmark reports stalled "rather than hanging or lying," but that guarantee only covers the completion-wait loop (lines 132-145). If the submitter's client.submit(...) call hangs — e.g., the freshly-spawned bench_server crashes or the connection stalls — submit_task.await at line 119 blocks indefinitely with no stall detection or timeout, silently defeating the documented no-hang guarantee for the whole run.
🤖 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/bench.rs` around lines 96 - 119, Protect the submission phase
in the `submit_task` flow from hanging indefinitely: apply the same
stall/timeout mechanism used by the completion-wait loop before awaiting the
spawned submitter, and preserve the benchmark’s `stalled` outcome when the
submission deadline is exceeded. Ensure `submit_task.await` cannot block forever
if `client.submit` or the server connection stalls.
| @@ -0,0 +1,156 @@ | |||
| //! ROUTER server loop: decodes wire requests and dispatches them to the nucleus. | |||
There was a problem hiding this comment.
🔒 Security & Privacy | 🟠 Major | ⚡ Quick win
🧩 Analysis chain
🌐 Web query:
Does the tmq Rust crate (0.5.0) support configuring ZeroMQ CURVE security on dealer/router sockets?
💡 Result:
Yes, the tmq crate version 0.5.0 supports configuring ZeroMQ CURVE security for sockets, including Dealer and Router sockets [1][2][3]. The 0.5.0 release explicitly added setters for CURVE key encryption [2][4][3]. Configuration is handled through the SocketBuilder, which provides methods to set the necessary ZMQ options before binding or connecting the socket [5]. Available methods for CURVE configuration on the SocketBuilder include [5]: - set_curve_server(value: bool) - set_curve_secretkey(value: &[u8]) - set_curve_serverkey(value: &[u8]) - set_curve_publickey(value: &[u8]) These methods allow you to configure CURVE authentication and encryption parameters on the underlying ZMQ sockets, which are then used by the Dealer/Router implementations [5][1].
Citations:
- 1: https://github.com/cetra3/tmq
- 2: https://docs.rs/crate/tmq/0.5.0
- 3: https://crates.io/crates/tmq/0.5.0
- 4: https://crates.io/crates/tmq
- 5: https://docs.rs/tmq/latest/tmq/struct.SocketBuilder.html
🏁 Script executed:
#!/bin/bash
set -euo pipefail
printf '\n== engine/src/server.rs ==\n'
cat -n engine/src/server.rs | sed -n '1,220p'
printf '\n== engine/src/client.rs ==\n'
cat -n engine/src/client.rs | sed -n '1,220p'
printf '\n== auth/config references ==\n'
rg -n "AuthEvent|cgrpc_token|curve|set_curve|plain|security|Envelope\.auth|auth token|token" engine -SRepository: voltaero/engine
Length of output: 13303
Encrypt the ROUTER/DEALER channel engine/src/server.rs#L23-L27 and engine/src/client.rs#L21-L25 create the socket pair with no transport security, so Envelope.auth and task payloads travel in plaintext over TCP. Add matching CURVE/TLS protection before exposing this beyond a trusted local network.
🤖 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/server.rs` at line 1, Add matching transport encryption to the
ROUTER/DEALER socket setup in the server initialization and corresponding client
connection flow, using the project’s supported CURVE or TLS mechanism. Configure
both peers with compatible security credentials before connecting or binding,
while preserving the existing request dispatch and connection behavior.
Source: Coding guidelines
| /// A request plus the caller's auth token, as sent on the wire. | ||
| #[derive(Debug, Clone, Serialize, Deserialize)] | ||
| pub struct Envelope { | ||
| /// Caller auth token, checked server-side before dispatch. | ||
| pub auth: String, | ||
| /// The requested operation. | ||
| pub req: Request, | ||
| } |
There was a problem hiding this comment.
🔒 Security & Privacy | 🟠 Major | ⚡ Quick win
🧩 Analysis chain
🌐 Web query:
Does ZeroMQ encrypt traffic by default over tcp:// sockets, or is CURVE/PLAIN required?
💡 Result:
No, ZeroMQ does not encrypt traffic by default over tcp:// sockets [1][2]. By default, ZeroMQ uses the NULL security mechanism, which provides no authentication or encryption [3][4][5]. Regarding the use of CURVE and PLAIN: * CURVE: This is the security mechanism required if you need both authentication and encryption [4][6][7]. It uses elliptic curve cryptography (Curve25519) to provide confidentiality and secure authentication [4][8]. * PLAIN: This mechanism provides simple username and password authentication but does not provide encryption or confidentiality [9][10]. It is intended for use on internal, trusted networks [9][10]. In summary, if you require encryption for your traffic, you must explicitly configure the CURVE mechanism on your sockets [6][7]. Using PLAIN or the default NULL mechanism will leave your traffic unencrypted [2][9].
Citations:
- 1: https://zeromq.github.io/zeromq.js/classes/Stream.html
- 2: https://libzmq.readthedocs.io/en/zeromq4-1/zmq_setsockopt.html
- 3: https://pyzmq.readthedocs.io/en/v25.0.2/api/zmq.auth.html
- 4: http://wiki.zeromq.org/docs:changes-4-0-0
- 5: http://api.zeromq.org/4-2:zmq
- 6: https://libzmq.readthedocs.io/en/zeromq4-x/zmq.html
- 7: https://libzmq.readthedocs.io/en/zeromq4-1/zmq%5Fcurve.html
- 8: https://rfc.zeromq.org/spec/25/
- 9: https://rfc.zeromq.org/spec/24/
- 10: https://zeromq.github.io/libzmq/zmq_plain.html
🏁 Script executed:
#!/bin/bash
set -euo pipefail
printf '\n== protocol.rs outline ==\n'
ast-grep outline enginelib/src/protocol.rs --view expanded || true
printf '\n== client/server references ==\n'
rg -n "CURVE|PLAIN|zmq|Envelope|auth token|auth:" enginelib/src -g '!**/target/**' || true
printf '\n== protocol.rs (small excerpt) ==\n'
cat -n enginelib/src/protocol.rs | sed -n '1,140p'
printf '\n== client/server files ==\n'
fd -a 'client.rs|server.rs' enginelib/src || trueRepository: voltaero/engine
Length of output: 5214
Redact Envelope's Debug output in enginelib/src/protocol.rs:20-27. auth is a bearer token, and the derived Debug impl will print it into logs; use a custom Debug impl that masks auth.
🤖 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/protocol.rs` around lines 20 - 27, Replace the derived Debug
implementation on Envelope with a custom implementation that preserves useful
request details while masking the auth bearer token. Keep Clone, Serialize, and
Deserialize derives unchanged, and ensure formatting Envelope with {:?} never
exposes the auth value.
| } | ||
| } | ||
| let mut wopts = WriteOptions::default(); | ||
| wopts.set_sync(false); |
There was a problem hiding this comment.
This write adds the finished record and deletes the pending record, but the server acknowledges completion without syncing the WAL. If the host loses power after returning Completed but before the write becomes durable, the pending record can reappear after restart. The loader then leases and executes a task that was already reported as completed.
Context Used: AGENTS.md (source)
There was a problem hiding this comment.
Actionable comments posted: 1
🤖 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/lease.rs`:
- Around line 101-128: Prevent oversized tasks detected in lease_batch_limited
from being requeued indefinitely: update both the first-task size check and the
subsequent drained-task check to route tasks exceeding max_bytes to an existing
dead-letter/quarantine mechanism, or reject them during submit_batch if the
leasing budget is available there. Ensure oversized tasks are removed from the
retryable queue and cannot be immediately redrawn.
🪄 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: 695c90c2-0a4b-4066-9f06-67f5b3b8f9c4
⛔ Files ignored due to path filters (1)
Cargo.lockis excluded by!**/*.lock
📒 Files selected for processing (20)
engine/Cargo.tomlengine/src/bench_task.rsengine/src/bin/bench.rsengine/src/bin/bench_server.rsengine/src/bin/server.rsengine/src/client.rsengine/src/lib.rsengine/src/server.rsengine/src/transport.rsengine/tests/e2e.rsengine/tests/recovery.rsengine/tests/transport_bounds.rsenginelib/src/api.rsenginelib/src/config.rsenginelib/src/error.rsenginelib/src/nucleus/complete.rsenginelib/src/nucleus/lease.rsenginelib/src/nucleus/query.rsenginelib/src/nucleus/submit.rsenginelib/src/protocol.rs
🚧 Files skipped from review as they are similar to previous changes (10)
- engine/src/lib.rs
- engine/src/bin/bench_server.rs
- enginelib/src/nucleus/query.rs
- enginelib/src/error.rs
- enginelib/src/protocol.rs
- engine/src/client.rs
- engine/src/bin/bench.rs
- engine/tests/e2e.rs
- enginelib/src/nucleus/complete.rs
- enginelib/src/api.rs
| let first = match tokio::time::timeout(poll_timeout, receiver.recv()).await { | ||
| Err(_elapsed) => return Ok(Vec::new()), | ||
| Ok(recv) => recv.map_err(|err| Error::new(format!("Task queue closed: {err}")))?, | ||
| }; | ||
|
|
||
| let mut used_bytes = estimated_wire_bytes(&first); | ||
| if used_bytes > max_bytes { | ||
| requeue_unleased(&api, &task_type, &sender, first); | ||
| return Err(Error::invalid_argument( | ||
| "Stored task exceeds the configured lease response limit", | ||
| )); | ||
| } | ||
|
|
||
| let mut out: Vec<Arc<StoredTask>> = vec![Arc::new(first)]; | ||
| while (out.len() as u32) < max { | ||
| match receiver.try_recv() { | ||
| Ok(task) => { | ||
| let task_bytes = estimated_wire_bytes(&task); | ||
| if used_bytes.saturating_add(task_bytes) > max_bytes { | ||
| requeue_unleased(&api, &task_type, &sender, task); | ||
| break; | ||
| } | ||
| used_bytes = used_bytes.saturating_add(task_bytes); | ||
| out.push(Arc::new(task)); | ||
| } | ||
| Err(_) => break, | ||
| } | ||
| } |
There was a problem hiding this comment.
🩺 Stability & Availability | 🟠 Major | 🏗️ Heavy lift
Oversized tasks are requeued but never quarantined — livelock risk.
When a task's estimated_wire_bytes exceeds max_bytes (first task at lines 106-112, or a subsequent drained task at lines 119-121), it's put back on the channel via requeue_unleased and the call fails/stops. Nothing marks the task as permanently unliftable: the same task can be redrawn on the very next lease_batch_limited call (this worker retrying, or another), fail the identical size check, and get requeued again — an unbounded retry loop for any payload whose encoded size structurally exceeds the configured max_bytes/max_reply_bytes (wired in engine/src/server.rs's dispatch). Since submit_batch in enginelib/src/nucleus/submit.rs doesn't validate payload size against this budget, a legitimately-submitted oversized task has no path to ever being leased or otherwise resolved.
Consider either rejecting oversized payloads at submit time (if the leasing budget is knowable there), or routing tasks that fail the size check to a dead-letter/quarantine state instead of unconditionally requeuing them for immediate re-draw.
🤖 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/lease.rs` around lines 101 - 128, Prevent oversized
tasks detected in lease_batch_limited from being requeued indefinitely: update
both the first-task size check and the subsequent drained-task check to route
tasks exceeding max_bytes to an existing dead-letter/quarantine mechanism, or
reject them during submit_batch if the leasing budget is available there. Ensure
oversized tasks are removed from the retryable queue and cannot be immediately
redrawn.
1 participant
Summary by CodeRabbit
submit,lease,complete,renew,cancel, andquery.