Conversation
PR SummaryMedium Risk Overview Exposes typed errors, read models (cron/every/RRULE, NATS delivery, message headers), queries ( Reviewed by Cursor Bugbot for commit 8123459. Bugbot is set up for automated code reviews on this repo. Configure here. |
|
Note Reviews pausedIt looks like this branch is under active development. To avoid overwhelming you with review comments due to an influx of new commits, CodeRabbit has automatically paused this review. You can configure this behavior by changing the Use the following commands to manage reviews:
Use the checkboxes below for quick actions:
WalkthroughAdds a new crate Changes
Sequence Diagram(s)sequenceDiagram
autonumber
participant Scheduler as Scheduler
participant KV as NATS KV (configs/leader)
participant Watch as Config Watcher
participant Leader as LeaderElection
participant Publisher as TickPublisher (JetStream)
participant Executor as Executor
Scheduler->>KV: load_jobs_and_watch()
KV-->>Scheduler: (snapshot, watch_stream)
loop main tick
Scheduler->>Leader: ensure_leader()
alt is leader
Scheduler->>Scheduler: compute due jobs(now)
Scheduler->>Executor: execute(job_state)
Executor->>Publisher: publish_tick(...) / spawn process
Publisher-->>Executor: publish result / ack
else not leader
Scheduler-->>Scheduler: skip execution
end
end
Watch-->>Scheduler: JobConfigChange (Put/Delete)
Scheduler->>Scheduler: apply config change (hot-reload)
sequenceDiagram
autonumber
participant Executor as Executor
participant Retry as RetryLoop
participant Jet as JetStream
participant DLQ as cron.errors
Executor->>Retry: prepare TickPayload & headers
Retry->>Jet: publish(payload)
alt success
Jet-->>Retry: ack
else transient failure
Retry->>Retry: backoff + jitter (respect max duration)
Retry->>Jet: publish(retry)
alt retries exhausted
Retry->>DLQ: publish dead-letter(details)
end
end
Estimated code review effort🎯 4 (Complex) | ⏱️ ~75 minutes Possibly related PRs
Poem
🚥 Pre-merge checks | ✅ 2 | ❌ 1❌ Failed checks (1 warning)
✅ Passed checks (2 passed)
✏️ Tip: You can configure your own custom pre-merge checks in the settings. ✨ 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 |
|
Code Coverage SummaryDetailsDiff against mainResults for commit: 5d02745 Minimum allowed coverage is ♻️ This comment has been updated with latest results |
yordis
changed the title
yordis
force-pushed
the
CRON
branch
4 times, most recently
from
1d1431c to
8652b66
Compare
![cursor[bot]](https://avatars.githubusercontent.com/in/1210556?s=60&v=4){kind=small}
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 10
🧹 Nitpick comments (5)
rsworkspace/crates/trogon-cron/Dockerfile (1)
16-16: Add--no-install-recommendsto reduce image size.The
apt-get installcommands should include--no-install-recommendsto avoid pulling unnecessary packages.📦 Proposed fix
-RUN apt-get update && apt-get install -y pkg-config libssl-dev && rm -rf /var/lib/apt/lists/* +RUN apt-get update && apt-get install -y --no-install-recommends pkg-config libssl-dev && rm -rf /var/lib/apt/lists/*🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@rsworkspace/crates/trogon-cron/Dockerfile` at line 16, Modify the Dockerfile RUN line that installs system packages so apt-get install uses --no-install-recommends to avoid pulling unnecessary packages; update the RUN command that currently calls "apt-get install -y pkg-config libssl-dev" to include "--no-install-recommends" and keep the apt-get update and rm -rf /var/lib/apt/lists/* steps intact.rsworkspace/crates/trogon-cron/src/error.rs (1)
3-4:KvandPublishvariants discard error context by storing strings.Per coding guidelines, errors should wrap the source error as a field or variant instead of converting to a string. The
Kv(String)andPublish(String)variants likely originate from NATS KV/publish errors and should preserve the original error type for proper error chaining.Consider wrapping the underlying
async_natserror types:pub enum CronError { - Kv(String), - Publish(String), + Kv(async_nats::jetstream::kv::CreateError), // or appropriate error type + Publish(async_nats::PublishError), // or appropriate error typeThen update
source()to returnSome(e)for these variants as well.As per coding guidelines: "Never discard error context by converting a typed error into a string; wrap the source error as a field or variant instead."
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@rsworkspace/crates/trogon-cron/src/error.rs` around lines 3 - 4, The Kv(String) and Publish(String) enum variants in the Error type discard original error context; change them to hold the underlying async_nats error (e.g., Kv(async_nats::Error) and Publish(async_nats::Error) or a boxed source like Kv(Box<dyn std::error::Error + Send + Sync>)) so the original error is preserved, update any constructors/From impls that currently convert to String to wrap the original error instead, and modify the Error::source(&self) implementation to return Some(inner_error) for the Kv and Publish variants so proper error chaining is maintained.rsworkspace/crates/trogon-cron/src/config.rs (1)
325-334: Consider movingTickPayloadbefore the tests module.
TickPayloadis defined after#[cfg(test)] mod tests, which is unconventional. While valid Rust, it may confuse readers who expect all public types to be defined before tests.🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@rsworkspace/crates/trogon-cron/src/config.rs` around lines 325 - 334, The TickPayload struct is defined after the #[cfg(test)] mod tests block which is unconventional and can confuse readers; move the TickPayload definition (the pub struct TickPayload with fields job_id, fired_at, execution_id, payload and its derives Serialize/Deserialize) so it appears before the tests module (i.e., place TickPayload above the tests mod) and ensure any test code referencing TickPayload still compiles after the reorder.rsworkspace/crates/trogon-cron/src/kv.rs (2)
96-127: The 500ms timeout for empty-bucket detection may be fragile.The hardcoded 500ms deadline assumes the initial watcher snapshot arrives within that window. On high-latency networks or under load, this could cause the scheduler to start with an incomplete job set, silently missing jobs that arrive after the timeout but before the first
delta == 0entry.Consider either:
- Making the timeout configurable
- Using the NATS
pending_countor similar metadata to detect completion- Documenting this limitation prominently
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@rsworkspace/crates/trogon-cron/src/kv.rs` around lines 96 - 127, The hardcoded 500ms deadline used via tokio::time::sleep (pinned as deadline) to decide the watcher initial snapshot is fragile and can drop jobs; replace it with a robust approach: either make the timeout configurable (expose a parameter/const used where deadline is created) or remove the timeout and use watcher metadata (e.g., NATS pending_count or another completion signal) to detect the end-of-snapshot instead of breaking on timeout; update the loop around watcher.next()/delta checks (and error path returning CronError::Kv) to rely on the chosen mechanism so JobConfig deserialization and the is_last (e.delta == 0) logic drive completion reliably.
55-62: Swallowing thecreate_streamerror loses diagnostic information.If
create_streamfails for a reason other than "stream already exists" (e.g., permission denied, invalid config), the code falls through toget_stream, which may fail with a misleading error. Consider matching on the specific "already exists" error variant rather than catching all errors.match js.create_stream(config).await { Ok(_) => Ok(()), Err(e) if e.kind() == CreateStreamErrorKind::StreamNameExists => { // Already exists, verify it's accessible js.get_stream(TICKS_STREAM).await.map(|_| ()).map_err(...) } Err(e) => Err(CronError::Stream(e)), // Propagate actual creation failures }🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@rsworkspace/crates/trogon-cron/src/kv.rs` around lines 55 - 62, The current match on js.create_stream(config).await swallows all errors and falls back to js.get_stream(TICKS_STREAM), losing the original failure details; update the match to pattern-match the specific "stream already exists" error (e.g., Err(e) if e.kind() == CreateStreamErrorKind::StreamNameExists) and only in that case call js.get_stream(TICKS_STREAM).await.map(|_| ()).map_err(|e| CronError::Kv(e.to_string())); for any other Err(e) from js.create_stream return Err(CronError::Stream(e.to_string())) (or the appropriate CronError variant) so creation failures are propagated instead of masked.
🤖 Prompt for all review comments with AI agents
Verify each finding against the current code and only fix it if needed.
Inline comments:
In `@rsworkspace/crates/trogon-cron/Dockerfile`:
- Around line 22-31: The runtime Dockerfile currently runs the container as
root; modify the runtime stage to create a dedicated non-root user and group
(e.g., trogon), ensure the copied binary /usr/local/bin/trogon-cron is owned by
that user (chown) and has executable permissions, set a sensible HOME if needed,
and add a USER trogon directive before ENTRYPOINT so the container runs as the
non-root user; reference the runtime stage, the COPY of
/usr/local/bin/trogon-cron, ENTRYPOINT ["trogon-cron"], and CMD ["serve"] when
making these changes.
In `@rsworkspace/crates/trogon-cron/src/client.rs`:
- Around line 73-80: The code in set_enabled uses CronError::Kv(format!(...))
for a missing job; add a typed error variant like CronError::JobNotFound { id:
String } in your error enum (e.g., in error.rs) and replace the ok_or_else
mapping in set_enabled to return CronError::JobNotFound { id: id.to_string() }
instead of constructing a formatted string; update any other sites that
currently create "job not found" Kv strings to use the new JobNotFound variant
so callers can match on the typed error.
In `@rsworkspace/crates/trogon-cron/src/executor.rs`:
- Around line 279-281: The current guard around publish_dead_letter (the if
max_retries > 0 check) prevents publishing a DLQ event when an execution
permanently fails with zero retries; update the failure path so
publish_dead_letter(&publisher, &tick, "publish", actual_attempts,
last_error).await is invoked whenever execution ultimately fails (i.e., when
last_error indicates a terminal failure), regardless of max_retries or retry:
None. Remove or replace the max_retries conditional in the block that handles
final failures (referencing publish_dead_letter, publisher, tick,
actual_attempts, last_error) and make the same change in the analogous block
around lines 442-444 so dead-letter events are always emitted for permanent
failures.
- Around line 224-280: The retry loop is converting errors to String (last_error
= e.to_string()) and losing typed context; instead introduce a small internal
error enum (e.g., PublishError with variants for NatsPublish, Spawn, etc.) and
change last_error from String to that enum (or Option<PublishError>), assign
Err(e) into the appropriate variant when publish_tick fails, and propagate that
typed error through the loop; only convert to String when calling
publish_dead_letter and in the final tracing::error (i.e., stringify inside
publish_dead_letter or right at the logging call). Update publish_dead_letter
(or its callsite) to accept the typed error or perform the to_string there, and
ensure the enum implements Display/Debug/From for the original error types so no
source context is lost.
- Around line 184-201: The concurrent check in RuntimeAction::Spawn uses
state.is_running.load(...) then later sets it, which is racy; replace the
load+separate store with an atomic compare_exchange on state.is_running to
change false -> true (using appropriate Orderings, e.g., SeqCst or
Acquire/Release) and only call spawn_process when compare_exchange succeeds;
apply the same compare_exchange replacement for the other occurrence around
lines 297-303 so the guard for non-concurrent jobs is claimed atomically before
invoking spawn_process.
In `@rsworkspace/crates/trogon-cron/src/nats_impls.rs`:
- Around line 195-218: The TickPublisher::publish_tick implementation is
discarding typed error context by converting PublishOutcome errors to strings;
update the error handling to preserve and wrap the original error types instead
of calling to_string(). Add variants to CronError (e.g.,
CronError::PublishError(PublishError) and CronError::AckError(AckError) or a
single CronError::PublishSource(Box<dyn Error + Send + Sync>)) and map
PublishOutcome::PublishFailed(e), ::AckFailed(e), and ::StoreFailed(e) to those
variants (or use .source() wrapping) when returning Err; ensure publish_tick
(and any call sites of publish_event / JetStreamContextPublisher) return the new
CronError variants so the original PublishError/AckError types and context are
preserved rather than being stringified.
- Around line 36-49: The NatsConfigStore::put_job implementation wraps SDK
errors into CronError::Kv by converting them to strings and uses map_err,
violating the zero-cost passthrough rule; change the implementation of
ConfigStore for NatsConfigStore to return the underlying SDK errors directly
(i.e., remove map_err calls around self.js.get_key_value(...) and kv.put(...)
and avoid converting errors to strings) or adjust CronError::Kv to carry the
original SDK error type so you can use the ? operator without losing context;
ensure serde_json::to_vec errors are propagated similarly (e.g., via ?), so the
body of put_job is a direct await/return passthrough to js.get_key_value and
kv.put with no map_err conversions.
- Around line 234-259: The code in NatsLeaderLock's try_acquire, renew, and
release is converting underlying store errors to strings (map_err(|e|
CronError::Kv(e.to_string()))) which discards original error types; change this
by making CronError::Kv carry the original error (or a boxed dyn Error) and/or
implement From for the store error types, then replace map_err calls with
map_err(Into::into) or map_err(CronError::from) so create/update/delete errors
are preserved; update the CronError enum definition to accept the concrete
source types (or Box<dyn std::error::Error + Send + Sync>) and adjust
try_acquire, renew, and release to forward the original
CreateError/UpdateError/DeleteError instead of to_string.
In `@rsworkspace/crates/trogon-cron/src/scheduler.rs`:
- Around line 180-201: The helper reestablish_config_watch currently blocks
retries and is not cancellation-aware, which can prevent run() from observing
shutdown signals; modify reestablish_config_watch to accept a cancellation
trigger (e.g., a ShutdownReceiver, a CancellationToken, or an async
shutdown_signal future) or return early on cancellation, then inside the loop
use tokio::select! to race ConfigStore::load_and_watch().await and the shutdown
trigger (and the sleep retry against the shutdown), returning an Err/early exit
when cancelled so the outer run() can handle graceful leader-lock release;
alternatively move the retry sleep/select logic up into run() and keep
reestablish_config_watch as a single-attempt helper that returns its
watcher/error to the caller.
- Around line 153-156: preserve_runtime_state currently only copies is_running
and publish_in_flight, causing last_fired to reset and interval jobs to fire
immediately after reload; update preserve_runtime_state (and the build_job_state
path that calls it) to also carry forward the prior firing state by copying
old_state.last_fired into new_state.last_fired when the schedule is unchanged,
or, if the schedule has changed, recompute new_state.last_fired from
old_state.last_fired (e.g., advance/snap it through the new schedule) so reloads
do not treat every reload as a first load; refer to preserve_runtime_state,
build_job_state, and JobState.last_fired to locate where to copy or recompute
the value.
---
Nitpick comments:
In `@rsworkspace/crates/trogon-cron/Dockerfile`:
- Line 16: Modify the Dockerfile RUN line that installs system packages so
apt-get install uses --no-install-recommends to avoid pulling unnecessary
packages; update the RUN command that currently calls "apt-get install -y
pkg-config libssl-dev" to include "--no-install-recommends" and keep the apt-get
update and rm -rf /var/lib/apt/lists/* steps intact.
In `@rsworkspace/crates/trogon-cron/src/config.rs`:
- Around line 325-334: The TickPayload struct is defined after the #[cfg(test)]
mod tests block which is unconventional and can confuse readers; move the
TickPayload definition (the pub struct TickPayload with fields job_id, fired_at,
execution_id, payload and its derives Serialize/Deserialize) so it appears
before the tests module (i.e., place TickPayload above the tests mod) and ensure
any test code referencing TickPayload still compiles after the reorder.
In `@rsworkspace/crates/trogon-cron/src/error.rs`:
- Around line 3-4: The Kv(String) and Publish(String) enum variants in the Error
type discard original error context; change them to hold the underlying
async_nats error (e.g., Kv(async_nats::Error) and Publish(async_nats::Error) or
a boxed source like Kv(Box<dyn std::error::Error + Send + Sync>)) so the
original error is preserved, update any constructors/From impls that currently
convert to String to wrap the original error instead, and modify the
Error::source(&self) implementation to return Some(inner_error) for the Kv and
Publish variants so proper error chaining is maintained.
In `@rsworkspace/crates/trogon-cron/src/kv.rs`:
- Around line 96-127: The hardcoded 500ms deadline used via tokio::time::sleep
(pinned as deadline) to decide the watcher initial snapshot is fragile and can
drop jobs; replace it with a robust approach: either make the timeout
configurable (expose a parameter/const used where deadline is created) or remove
the timeout and use watcher metadata (e.g., NATS pending_count or another
completion signal) to detect the end-of-snapshot instead of breaking on timeout;
update the loop around watcher.next()/delta checks (and error path returning
CronError::Kv) to rely on the chosen mechanism so JobConfig deserialization and
the is_last (e.delta == 0) logic drive completion reliably.
- Around line 55-62: The current match on js.create_stream(config).await
swallows all errors and falls back to js.get_stream(TICKS_STREAM), losing the
original failure details; update the match to pattern-match the specific "stream
already exists" error (e.g., Err(e) if e.kind() ==
CreateStreamErrorKind::StreamNameExists) and only in that case call
js.get_stream(TICKS_STREAM).await.map(|_| ()).map_err(|e|
CronError::Kv(e.to_string())); for any other Err(e) from js.create_stream return
Err(CronError::Stream(e.to_string())) (or the appropriate CronError variant) so
creation failures are propagated instead of masked.
🪄 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: Organization UI
Review profile: CHILL
Plan: Pro
Run ID: 824714aa-1b17-40de-8d8e-1d77a4a2f5c4
⛔ Files ignored due to path filters (1)
rsworkspace/Cargo.lockis excluded by!**/*.lock
📒 Files selected for processing (19)
rsworkspace/crates/AGENTS.mdrsworkspace/crates/trogon-cron/Cargo.tomlrsworkspace/crates/trogon-cron/Dockerfilersworkspace/crates/trogon-cron/docker-compose.ymlrsworkspace/crates/trogon-cron/src/client.rsrsworkspace/crates/trogon-cron/src/config.rsrsworkspace/crates/trogon-cron/src/domain.rsrsworkspace/crates/trogon-cron/src/error.rsrsworkspace/crates/trogon-cron/src/executor.rsrsworkspace/crates/trogon-cron/src/kv.rsrsworkspace/crates/trogon-cron/src/leader.rsrsworkspace/crates/trogon-cron/src/lib.rsrsworkspace/crates/trogon-cron/src/main.rsrsworkspace/crates/trogon-cron/src/mocks.rsrsworkspace/crates/trogon-cron/src/nats_impls.rsrsworkspace/crates/trogon-cron/src/scheduler.rsrsworkspace/crates/trogon-cron/src/traits.rsrsworkspace/crates/trogon-cron/tests/cron_unit.rsrsworkspace/crates/trogon-cron/tests/integration.rs
yordis
added
the
rust:coverage-baseline-reset
yordis
force-pushed
the
CRON
branch
2 times, most recently
from
f0d1276 to
6728775
Compare
There was a problem hiding this comment.
Actionable comments posted: 2
♻️ Duplicate comments (6)
rsworkspace/crates/trogon-cron/src/nats_impls.rs (3)
36-120:⚠️ Potential issue | 🟠 MajorStop stringifying JetStream errors in the
ConfigStoreimpl.These
map_err(|e| CronError::Kv(e.to_string()))calls erase the concreteasync-natserror types on every KV operation. Either letCronError::Kvwrap the SDK errors directly or giveConfigStorean associated error type so these methods can stay passthroughs and use?.As per coding guidelines: "Production implementations of infrastructure traits must be zero-cost passthroughs to the underlying SDK. No error wrapping (use the SDK's error types directly via associated
type Error), no return type conversion, nomap_err, nomap(|_| ())." and "Never discard error context by converting a typed error into a string; wrap the source error as a field or variant instead."🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@rsworkspace/crates/trogon-cron/src/nats_impls.rs` around lines 36 - 120, The current NatsConfigStore implementation for ConfigStore (methods put_job, get_job, delete_job, list_jobs, load_and_watch) is converting async-nats/jetstream SDK errors to strings via map_err(|e| CronError::Kv(e.to_string())), losing typed error context; fix by making these methods passthrough zero-cost: either change CronError::Kv to hold the original SDK error type (wrap the error instead of to_string) or update the ConfigStore signature to use an associated Error type and return the SDK errors directly (remove map_err and the e.to_string() conversions) so all KV calls (get_key_value, entry, delete, keys, get, etc.) propagate the concrete SDK errors via ?.
194-215:⚠️ Potential issue | 🟠 MajorPreserve the original
PublishOutcomefailure sources here.
PublishFailed,AckFailed,AckTimedOut, andStoreFailedall get collapsed intoCronError::Publish(String), so the caller loses the concrete publish/ack/store failure type and nested context. This should stay typed through theTickPublisherboundary.As per coding guidelines: "Never discard error context by converting a typed error into a string; wrap the source error as a field or variant instead."
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@rsworkspace/crates/trogon-cron/src/nats_impls.rs` around lines 194 - 215, The publish_tick implementation on jetstream::Context currently collapses PublishOutcome variants into CronError::Publish(String); change CronError to preserve source error types (e.g., add variants CronError::PublishFailed(PublishError), CronError::AckFailed(AckError), CronError::AckTimedOut(Duration) and CronError::StoreFailed(StoreError) or a single CronError::Publish{source: PublishOutcomeError} that wraps the original error), then update publish_tick to map each PublishOutcome variant to the corresponding typed CronError (use .map_err(|e| CronError::PublishFailed(e)) or construct the appropriate variant for AckTimedOut with the timeout), ensuring you propagate the original error values instead of converting them to strings; touch the publish_tick function, the PublishOutcome match arms, and the CronError enum to implement this typed error propagation.
233-258:⚠️ Potential issue | 🟠 MajorLeader-lock KV operations still erase SDK error types.
try_acquire,renew, andreleaseare still doingCronError::Kv(e.to_string()), which drops the original create/update/delete errors and breaks the zero-cost passthrough rule for infrastructure code.As per coding guidelines: "Production implementations of infrastructure traits must be zero-cost passthroughs to the underlying SDK. No error wrapping (use the SDK's error types directly via associated
type Error), no return type conversion, nomap_err, nomap(|_| ())." and "Errors must be typed—use structs or enums, neverStringorformat!()."🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@rsworkspace/crates/trogon-cron/src/nats_impls.rs` around lines 233 - 258, The impl for NatsLeaderLock currently converts SDK errors into CronError via map_err(e.to_string()) in try_acquire, renew and release; change it to be a zero-cost passthrough by making the impl's associated type Error the underlying KV/store error type (the concrete error returned by self.store.create/update/delete, e.g., crate::kv::Error or the store client's error type) and remove all map_err/string conversions so each method returns the store's Result directly (call self.store.create(LEADER_KEY, value).await, self.store.update(LEADER_KEY, value, revision).await, and self.store.delete(LEADER_KEY).await without mapping errors). Ensure references to NatsLeaderLock, try_acquire, renew, release, CronError and crate::kv::LEADER_KEY are used to locate and update the code.rsworkspace/crates/trogon-cron/src/executor.rs (3)
279-281:⚠️ Potential issue | 🟠 MajorDead-letter terminal failures even when retries are disabled.
Both branches only publish to
cron.errorswhenmax_retries > 0, so a permanent first-attempt failure withretry: Noneormax_retries: 0never reaches the DLQ path. Emit the dead-letter event whenever execution ends in failure.Suggested fix
- if max_retries > 0 { - publish_dead_letter(&publisher, &tick, "publish", actual_attempts, last_error).await; - } + publish_dead_letter(&publisher, &tick, "publish", actual_attempts, last_error).await;- if max_retries > 0 { - publish_dead_letter(&publisher, &tick, "spawn", actual_attempts, last_error).await; - } + publish_dead_letter(&publisher, &tick, "spawn", actual_attempts, last_error).await;Also applies to: 442-444
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@rsworkspace/crates/trogon-cron/src/executor.rs` around lines 279 - 281, The code only calls publish_dead_letter(&publisher, &tick, "publish", actual_attempts, last_error).await when max_retries > 0, so permanent failures with retry: None or max_retries == 0 never get sent to the DLQ; remove or change the max_retries guard so publish_dead_letter is invoked whenever execution finishes in a failure path regardless of max_retries (i.e., call publish_dead_letter in the failure handling branch that currently handles terminal errors), and mirror this same fix for the other failure branch that currently has the same max_retries check (the similar publish_dead_letter call later in the file).
184-201:⚠️ Potential issue | 🔴 CriticalClaim
is_runningatomically before spawning.
load()inexecute()and the laterstore(true)inspawn_process()are separate operations, so two callers can both observefalseand both spawn whenconcurrent == false. Usecompare_exchange(false, true, ...)inexecute()and remove the unconditional store inspawn_process().Suggested fix
- let is_run = state.is_running.load(Ordering::SeqCst); - if !command.concurrent() && is_run { + if !command.concurrent() + && state + .is_running + .compare_exchange(false, true, Ordering::SeqCst, Ordering::SeqCst) + .is_err() + { tracing::debug!(job_id = %state.job.id(), "Skipping tick — previous invocation still running"); return; }- is_running.store(true, Ordering::SeqCst);Also applies to: 297-303
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@rsworkspace/crates/trogon-cron/src/executor.rs` around lines 184 - 201, The check of state.is_running in RuntimeAction::Spawn is racy because load() followed by spawn_process()’s store(true) allows two threads to both spawn when concurrent() == false; change the logic in execute() (where RuntimeAction::Spawn is handled) to attempt an atomic compare_exchange(false, true, Ordering::SeqCst, Ordering::SeqCst) on state.is_running and only proceed to call spawn_process if the exchange succeeds, and remove the unconditional store(true) inside spawn_process(); ensure spawn_process (or its completion path) still clears is_running back to false when the job finishes or fails, and apply the same compare_exchange-based claim to the other similar invocation that currently mirrors this pattern.
223-278:⚠️ Potential issue | 🟠 MajorKeep retry failures typed until the DLQ/log boundary.
last_error = e.to_string(),Err(e.to_string()), and theformat!(...)-based retry failures flatten everything intoStringlong before the retry loop finishes. That drops source context and makes the DLQ/log path lossy. Please keep an internal typed failure enum here and stringify only when you finally serializeDeadLetterPayload.Based on learnings: "Errors must be typed—use structs or enums, never
Stringorformat!()." and "Never discard error context by converting a typed error into a string; wrap the source error as a field or variant instead."Also applies to: 305-440
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@rsworkspace/crates/trogon-cron/src/executor.rs` around lines 223 - 278, The retry loop currently collapses all errors to String early (last_error = e.to_string()), losing source/context; introduce a small internal typed error enum/struct (e.g., enum RetryError { Publish(trogon_nats::Error), Timeout, BackoffExceeded, ... }) and change last_error from String to Option<RetryError>, store Err variants as last_error = Some(RetryError::Publish(e)) inside the loop (refer to publisher.publish_tick and the for attempt loop), keep all context until the DLQ/log boundary, and only stringify or serialize the error when constructing DeadLetterPayload or when emitting the final tracing::error/tracing::warn (use %last_error.to_string() or implement Display on RetryError at that boundary). Ensure other retry-related logic (retry_backoff_with_jitter, max_retry_duration checks, and places around lines 305-440) uses the typed error similarly.
🧹 Nitpick comments (2)
devops/docker/compose/compose.yml (1)
85-87: Add optional NATS auth env pass-through for secured deployments.This service currently injects only
NATS_URL; when NATS auth is enabled,trogon-cronsupports env-based auth but won’t receive credentials from Compose.Suggested patch
trogon-cron: build: context: ../../../rsworkspace dockerfile: ../devops/docker/compose/services/trogon-cron/Dockerfile environment: NATS_URL: "nats://nats:4222" + NATS_CREDS: "${NATS_CREDS:-}" + NATS_NKEY: "${NATS_NKEY:-}" + NATS_USER: "${NATS_USER:-}" + NATS_PASSWORD: "${NATS_PASSWORD:-}" + NATS_TOKEN: "${NATS_TOKEN:-}" RUST_LOG: "${RUST_LOG:-info}"🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@devops/docker/compose/compose.yml` around lines 85 - 87, Add optional NATS auth env passthrough to the trogon-cron service by extending its environment block (the same block that currently contains NATS_URL and RUST_LOG) to include optional variables such as NATS_USER, NATS_PASS and NATS_TOKEN (or other auth vars your deployment uses) and map them to host environment values (e.g., ${NATS_USER:-} style) so credentials are forwarded only when provided; update the environment section alongside NATS_URL and RUST_LOG to include these optional keys.rsworkspace/crates/trogon-cron/src/executor.rs (1)
61-65: Use saturating multiplication to guard against u64 overflow in the backoff calculation.When
base_secis very large, the multiplicationbase_sec * multipliercan theoretically overflow (panic in debug builds, wraparound in release). While practical configurations use small values (1–40 seconds, all well below the u64::MAX / 32 threshold), usingsaturating_mulprovides defensive correctness without cost.Suggested fix
- Duration::from_secs(base_sec * multiplier) + Duration::from_secs(base_sec.saturating_mul(multiplier))🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@rsworkspace/crates/trogon-cron/src/executor.rs` around lines 61 - 65, The backoff multiplier computation in retry_backoff can overflow when multiplying a large base_sec by the multiplier; change the multiplication to a saturating multiplication (use base_sec.saturating_mul(multiplier)) before creating the Duration so it never overflows. Update the retry_backoff function to compute multiplier as currently done and pass base_sec.saturating_mul(multiplier) into Duration::from_secs to defensively guard against u64 overflow.
🤖 Prompt for all review comments with AI agents
Verify each finding against the current code and only fix it if needed.
Inline comments:
In `@rsworkspace/crates/trogon-cron/src/domain.rs`:
- Around line 293-300: PublishSubject::new currently only checks the "cron."
prefix and allows invalid NATS subjects (wildcards, whitespace, empty tokens);
update the constructor to validate the full subject using the existing
trogon-nats token validation (e.g., trogon_nats::token::validate or the SDK
subject wrapper) and return CronError::InvalidJobConfig with a clear reason on
failure; specifically, inside PublishSubject::new replace the simple starts_with
check with a call to the trogon-nats validation function (or construct the
trogon-nats Subject type) for the entire subject string and propagate validation
errors into the same InvalidJobConfig variant so invalid subjects cannot be
constructed.
In `@rsworkspace/crates/trogon-cron/src/error.rs`:
- Around line 2-9: Replace the String payloads in the CronError enum
(Kv(String), Publish(String), InvalidJobConfig { reason: String }) with typed
error wrappers so callers can preserve source chains: define concrete error
types (e.g., KvError, PublishError, JobConfigError) or use wrappers carrying the
original error (e.g., Kv(Box<dyn std::error::Error + Send + Sync>),
Publish(Box<dyn std::error::Error + Send + Sync>), InvalidJobConfig { source:
Box<...> }), update the CronError declaration to use those types, and adjust any
Display/impl std::error::Error for CronError to produce human-readable messages
while returning the original sources from source(). Ensure code that currently
calls to_string() on sources is updated to pass the original error into the new
Kv/Publish/InvalidJobConfig variants.
---
Duplicate comments:
In `@rsworkspace/crates/trogon-cron/src/executor.rs`:
- Around line 279-281: The code only calls publish_dead_letter(&publisher,
&tick, "publish", actual_attempts, last_error).await when max_retries > 0, so
permanent failures with retry: None or max_retries == 0 never get sent to the
DLQ; remove or change the max_retries guard so publish_dead_letter is invoked
whenever execution finishes in a failure path regardless of max_retries (i.e.,
call publish_dead_letter in the failure handling branch that currently handles
terminal errors), and mirror this same fix for the other failure branch that
currently has the same max_retries check (the similar publish_dead_letter call
later in the file).
- Around line 184-201: The check of state.is_running in RuntimeAction::Spawn is
racy because load() followed by spawn_process()’s store(true) allows two threads
to both spawn when concurrent() == false; change the logic in execute() (where
RuntimeAction::Spawn is handled) to attempt an atomic compare_exchange(false,
true, Ordering::SeqCst, Ordering::SeqCst) on state.is_running and only proceed
to call spawn_process if the exchange succeeds, and remove the unconditional
store(true) inside spawn_process(); ensure spawn_process (or its completion
path) still clears is_running back to false when the job finishes or fails, and
apply the same compare_exchange-based claim to the other similar invocation that
currently mirrors this pattern.
- Around line 223-278: The retry loop currently collapses all errors to String
early (last_error = e.to_string()), losing source/context; introduce a small
internal typed error enum/struct (e.g., enum RetryError {
Publish(trogon_nats::Error), Timeout, BackoffExceeded, ... }) and change
last_error from String to Option<RetryError>, store Err variants as last_error =
Some(RetryError::Publish(e)) inside the loop (refer to publisher.publish_tick
and the for attempt loop), keep all context until the DLQ/log boundary, and only
stringify or serialize the error when constructing DeadLetterPayload or when
emitting the final tracing::error/tracing::warn (use %last_error.to_string() or
implement Display on RetryError at that boundary). Ensure other retry-related
logic (retry_backoff_with_jitter, max_retry_duration checks, and places around
lines 305-440) uses the typed error similarly.
In `@rsworkspace/crates/trogon-cron/src/nats_impls.rs`:
- Around line 36-120: The current NatsConfigStore implementation for ConfigStore
(methods put_job, get_job, delete_job, list_jobs, load_and_watch) is converting
async-nats/jetstream SDK errors to strings via map_err(|e|
CronError::Kv(e.to_string())), losing typed error context; fix by making these
methods passthrough zero-cost: either change CronError::Kv to hold the original
SDK error type (wrap the error instead of to_string) or update the ConfigStore
signature to use an associated Error type and return the SDK errors directly
(remove map_err and the e.to_string() conversions) so all KV calls
(get_key_value, entry, delete, keys, get, etc.) propagate the concrete SDK
errors via ?.
- Around line 194-215: The publish_tick implementation on jetstream::Context
currently collapses PublishOutcome variants into CronError::Publish(String);
change CronError to preserve source error types (e.g., add variants
CronError::PublishFailed(PublishError), CronError::AckFailed(AckError),
CronError::AckTimedOut(Duration) and CronError::StoreFailed(StoreError) or a
single CronError::Publish{source: PublishOutcomeError} that wraps the original
error), then update publish_tick to map each PublishOutcome variant to the
corresponding typed CronError (use .map_err(|e| CronError::PublishFailed(e)) or
construct the appropriate variant for AckTimedOut with the timeout), ensuring
you propagate the original error values instead of converting them to strings;
touch the publish_tick function, the PublishOutcome match arms, and the
CronError enum to implement this typed error propagation.
- Around line 233-258: The impl for NatsLeaderLock currently converts SDK errors
into CronError via map_err(e.to_string()) in try_acquire, renew and release;
change it to be a zero-cost passthrough by making the impl's associated type
Error the underlying KV/store error type (the concrete error returned by
self.store.create/update/delete, e.g., crate::kv::Error or the store client's
error type) and remove all map_err/string conversions so each method returns the
store's Result directly (call self.store.create(LEADER_KEY, value).await,
self.store.update(LEADER_KEY, value, revision).await, and
self.store.delete(LEADER_KEY).await without mapping errors). Ensure references
to NatsLeaderLock, try_acquire, renew, release, CronError and
crate::kv::LEADER_KEY are used to locate and update the code.
---
Nitpick comments:
In `@devops/docker/compose/compose.yml`:
- Around line 85-87: Add optional NATS auth env passthrough to the trogon-cron
service by extending its environment block (the same block that currently
contains NATS_URL and RUST_LOG) to include optional variables such as NATS_USER,
NATS_PASS and NATS_TOKEN (or other auth vars your deployment uses) and map them
to host environment values (e.g., ${NATS_USER:-} style) so credentials are
forwarded only when provided; update the environment section alongside NATS_URL
and RUST_LOG to include these optional keys.
In `@rsworkspace/crates/trogon-cron/src/executor.rs`:
- Around line 61-65: The backoff multiplier computation in retry_backoff can
overflow when multiplying a large base_sec by the multiplier; change the
multiplication to a saturating multiplication (use
base_sec.saturating_mul(multiplier)) before creating the Duration so it never
overflows. Update the retry_backoff function to compute multiplier as currently
done and pass base_sec.saturating_mul(multiplier) into Duration::from_secs to
defensively guard against u64 overflow.
🪄 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: Organization UI
Review profile: CHILL
Plan: Pro
Run ID: 0b160d0f-18b6-4b16-99ce-1ce2a1741a29
⛔ Files ignored due to path filters (1)
rsworkspace/Cargo.lockis excluded by!**/*.lock
📒 Files selected for processing (18)
devops/docker/compose/compose.ymldevops/docker/compose/services/trogon-cron/Dockerfilersworkspace/crates/trogon-cron/Cargo.tomlrsworkspace/crates/trogon-cron/src/client.rsrsworkspace/crates/trogon-cron/src/config.rsrsworkspace/crates/trogon-cron/src/domain.rsrsworkspace/crates/trogon-cron/src/error.rsrsworkspace/crates/trogon-cron/src/executor.rsrsworkspace/crates/trogon-cron/src/kv.rsrsworkspace/crates/trogon-cron/src/leader.rsrsworkspace/crates/trogon-cron/src/lib.rsrsworkspace/crates/trogon-cron/src/main.rsrsworkspace/crates/trogon-cron/src/mocks.rsrsworkspace/crates/trogon-cron/src/nats_impls.rsrsworkspace/crates/trogon-cron/src/scheduler.rsrsworkspace/crates/trogon-cron/src/traits.rsrsworkspace/crates/trogon-cron/tests/cron_unit.rsrsworkspace/crates/trogon-cron/tests/integration.rs
✅ Files skipped from review due to trivial changes (3)
- rsworkspace/crates/trogon-cron/tests/cron_unit.rs
- rsworkspace/crates/trogon-cron/Cargo.toml
- rsworkspace/crates/trogon-cron/src/main.rs
🚧 Files skipped from review as they are similar to previous changes (7)
- rsworkspace/crates/trogon-cron/src/leader.rs
- rsworkspace/crates/trogon-cron/src/client.rs
- rsworkspace/crates/trogon-cron/src/mocks.rs
- rsworkspace/crates/trogon-cron/src/kv.rs
- rsworkspace/crates/trogon-cron/src/lib.rs
- rsworkspace/crates/trogon-cron/src/traits.rs
- rsworkspace/crates/trogon-cron/src/scheduler.rs
There was a problem hiding this comment.
Actionable comments posted: 2
♻️ Duplicate comments (7)
rsworkspace/crates/trogon-cron/src/error.rs (1)
2-33:⚠️ Potential issue | 🟠 MajorKeep
CronErrorvariants typed instead of storing strings.
Kv(String),Publish(String), and the stringreasonpayloads force callers to flatten failures before they reach this enum, which is why most variants cannot expose asource()chain. Please wrap concrete source errors or typed validation errors here and keep the human-readable rendering inDisplay.Based on learnings: "Errors must be typed—use structs or enums, never
Stringorformat!()." and "Never discard error context by converting a typed error into a string; wrap the source error as a field or variant instead."🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@rsworkspace/crates/trogon-cron/src/error.rs` around lines 2 - 33, CronError currently stores String payloads (Kv(String), Publish(String), InvalidJobConfig { reason: String }) which discards typed error context; change these variants to hold concrete error types or boxed error trait objects (e.g., Kv(Box<dyn std::error::Error + Send + Sync>), Publish(Box<dyn std::error::Error + Send + Sync>), and InvalidJobConfig(ValidationError) where ValidationError is a new struct/enum describing validation failure), update the Display impl to format human-readable messages from those types, and extend std::error::Error::source() to return the wrapped error for Kv, Publish and InvalidJobConfig so callers can access the original error chain (leave Serde and Io handling as-is).rsworkspace/crates/trogon-cron/src/nats_impls.rs (3)
233-258:⚠️ Potential issue | 🟠 MajorLeader-lock operations are also discarding the KV source errors.
create,update, anddeleteall end inCronError::Kv(e.to_string()), so CAS mismatches and backend failures lose their concrete types andsource()chain. Please preserve the underlying KV errors here too.Based on learnings: "Never discard error context by converting a typed error into a string; wrap the source error as a field or variant instead." and "Errors must be typed—use structs or enums, never
Stringorformat!()."🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@rsworkspace/crates/trogon-cron/src/nats_impls.rs` around lines 233 - 258, The leader-lock methods (NatsLeaderLock::try_acquire, ::renew, ::release) currently map KV errors to CronError::Kv(e.to_string()), discarding the original typed error and its source chain; change the CronError::Kv variant to hold the original error type (or a boxed dyn Error) and map errors directly (e.g., .map_err(|e| CronError::Kv(e)) or .map_err(|e| CronError::Kv { source: e }) instead of e.to_string()), so the errors returned from self.store.create/update/delete preserve their concrete type and source.
194-215:⚠️ Potential issue | 🟠 MajorPreserve
PublishOutcomesources inTickPublisher::Error.
PublishFailed,AckFailed,AckTimedOut, andStoreFailedare all collapsed intoCronError::Publish(String), so downstream retry and DLQ handling cannot inspect or chain the real JetStream failure. This needs the same typed passthrough/wrapping treatment as the KV paths.Based on learnings: "Never discard error context by converting a typed error into a string; wrap the source error as a field or variant instead." and "Errors must be typed—use structs or enums, never
Stringorformat!()."🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@rsworkspace/crates/trogon-cron/src/nats_impls.rs` around lines 194 - 215, The TickPublisher impl for jetstream::Context currently collapses PublishOutcome variants into CronError::Publish(String); instead preserve and wrap the original typed errors instead of converting to strings—update CronError to include variants/fields for PublishFailed, AckFailed, AckTimedOut (with duration), and StoreFailed (or a single Publish variant that contains the original PublishOutcome::... error type), then change the matching in publish_tick (which calls publish_event and uses JetStreamContextPublisher and PublishOutcome) to return Err(CronError::PublishFailed(e)) / Err(CronError::AckFailed(e)) / Err(CronError::AckTimedOut(timeout)) / Err(CronError::StoreFailed(e)) (or the equivalent wrapper) so the original error types are preserved and chainable instead of using to_string()/format!.
36-157:⚠️ Potential issue | 🟠 Major
ConfigStoreshould pass through SDK errors instead of stringifying them.Every
map_err(|e| CronError::Kv(e.to_string()))here loses the original JetStream/KV error type and breaks the zero-cost passthrough rule for production infrastructure traits. Either let the trait surface the SDK errors directly, or makeCronErrorwrap the concrete sources so these calls can use?without flattening.As per coding guidelines, "Production implementations of infrastructure traits must be zero-cost passthroughs to the underlying SDK. No error wrapping (use the SDK's error types directly via associated
type Error), no return type conversion (add associated types liketype Infoto match the SDK's return), nomap_err, nomap(|_| ())." and "Never discard error context by converting a typed error into a string; wrap the source error as a field or variant instead."🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@rsworkspace/crates/trogon-cron/src/nats_impls.rs` around lines 36 - 157, The code currently stringifies SDK errors via map_err(|e| CronError::Kv(e.to_string())) in NatsConfigStore (see put_job, get_job, delete_job, list_jobs, load_and_watch), which loses error types; change CronError::Kv to hold the concrete SDK error type (e.g., kv::Error or the appropriate JetStream error) or add a new variant wrapping Box<dyn std::error::Error + Send + Sync>, implement From<kv::Error> for CronError, then remove the map_err closures and use the ? operator directly (e.g., let kv = self.js.get_key_value(CONFIG_BUCKET).await?; and similarly for kv.put(...).await? etc.), so SDK errors are propagated without being stringified. Ensure any places that previously used CronError::Kv(String) are updated to handle the new variant.rsworkspace/crates/trogon-cron/src/executor.rs (3)
223-281:⚠️ Potential issue | 🟠 MajorDon't erase retry failures to
Stringbefore the terminal path.
last_error = e.to_string(),Err(format!(...)), and theString-based spawn/publish retry loops drop the original failure type long before logging or DLQ publication. Keep an internal typed error through the retry path and stringify only at the final log / dead-letter boundary.Based on learnings: "Errors must be typed—use structs or enums, never
Stringorformat!()." and "Never discard error context by converting a typed error into a string; wrap the source error as a field or variant instead."Also applies to: 306-444
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@rsworkspace/crates/trogon-cron/src/executor.rs` around lines 223 - 281, The retry loop currently converts errors to String early (last_error = e.to_string()), losing typed error context; change last_error to hold the original typed error (e.g., Option<anyhow::Error> or Option<Box<dyn std::error::Error + Send + Sync>>) and assign Some(e.into()) when publisher.publish_tick(subject.clone(), headers, payload.clone().into()).await returns Err(e), keep using that typed error for all logic, and only call .to_string() (or format!) when emitting the final tracing::error and when calling publish_dead_letter(&publisher, &tick, "publish", actual_attempts, last_error_string). Ensure references to publisher.publish_tick, last_error, actual_attempts, and publish_dead_letter are updated so the retry warn logs use the typed error (or its display) without dropping error type until the terminal path.
279-281:⚠️ Potential issue | 🟠 MajorEmit DLQ events for terminal failures even when retries are disabled.
Right now a first-attempt permanent failure disappears from
cron.errorswheneverretry: Noneormax_retries == 0. If dead-lettering is part of the scheduler contract, the DLQ publish should happen whenever execution ends in failure, not only when retries were configured.Also applies to: 442-444
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@rsworkspace/crates/trogon-cron/src/executor.rs` around lines 279 - 281, The DLQ publish is gated by max_retries > 0 so terminal failures when retries are disabled never get emitted; remove that condition and always call publish_dead_letter(...) when execution ends in failure. Locate the two guarded calls (the one using publish_dead_letter(&publisher, &tick, "publish", actual_attempts, last_error).await and the duplicate at the later block) and change the control flow so that whenever a terminal failure path is taken (use the existing failure context variables: publisher, tick, actual_attempts, last_error) you await publish_dead_letter(...) unconditionally instead of only when max_retries > 0.
184-201:⚠️ Potential issue | 🟠 MajorClaim
is_runningatomically forconcurrent: falsejobs.Line 185 does a
load(), and Line 300 later does a separatestore(true). Two ticks can both observefalsebefore either store happens, so the non-concurrent guard can still launch duplicate processes. This needs a single atomic claim, e.g.compare_exchange(false, true, ...), beforespawn_process()is called.🔧 Suggested change
- let is_run = state.is_running.load(Ordering::SeqCst); - if !command.concurrent() && is_run { + if !command.concurrent() + && state + .is_running + .compare_exchange(false, true, Ordering::SeqCst, Ordering::SeqCst) + .is_err() + { tracing::debug!(job_id = %state.job.id(), "Skipping tick — previous invocation still running"); return; } @@ - is_running.store(true, Ordering::SeqCst); - tokio::spawn(async move { let _reset = AtomicFlagReset::new(is_running);Also applies to: 297-300
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@rsworkspace/crates/trogon-cron/src/executor.rs` around lines 184 - 201, The non-concurrent guard currently does a separate load() then later store(true), which allows a race where two ticks both see false and both spawn; change the logic in the RuntimeAction::Spawn branch to atomically claim state.is_running using compare_exchange(false, true, Ordering::SeqCst, Ordering::SeqCst) when !command.concurrent(), and only call spawn_process() if the compare_exchange succeeds (otherwise log/debug skip). Also remove or adapt the separate state.is_running.store(true) in the spawn_process / completion path so that only the atomic claim sets the flag (and ensure the flag is cleared on process end as before).
🧹 Nitpick comments (2)
rsworkspace/crates/trogon-cron/src/config.rs (1)
7-31: Make invalid config states unrepresentable in the public API.These exported config types still model validated concepts as raw primitives (
expr,subject,bin,id,interval_sec,timeout_sec), so malformed jobs deserialize successfully and only fail later in aggregate conversion. Prefer per-field value objects/newtypes with serde support here, so invalid cron expressions, subjects, paths, and positive-second values are rejected at construction time instead of afterJobConfigalready exists.As per coding guidelines, "Prefer domain-specific value objects over primitives (e.g.,
AcpPrefixnotString). Each type's factory must guarantee correctness at construction—invalid instances should be unrepresentable." and "Validate per-type, not per-aggregate: avoid validating unrelated fields together in a single constructor."Also applies to: 79-90
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@rsworkspace/crates/trogon-cron/src/config.rs` around lines 7 - 31, Replace raw primitive fields (expr, subject, bin, interval_sec, timeout_sec) in the public config enums/structs (Schedule, Action, and JobConfig) with domain-specific value objects/newtypes that perform validation at construction and implement Serde (e.g., CronExpr, NatsSubject, ExecutablePath, PositiveSeconds); update Schedule::Interval to use PositiveSeconds.interval_sec and Schedule::Cron to use CronExpr.expr, and Action::Publish/Spawn to use NatsSubject.subject and ExecutablePath.bin, plus PositiveSeconds.timeout_sec; ensure each newtype exposes a fallible constructor that enforces invariants and Serde deserializes by invoking that constructor so malformed configs fail at deserialization rather than later during JobConfig conversion.rsworkspace/crates/trogon-cron/src/main.rs (1)
136-150: Consider using a typed error instead offormat!()-basedio::Error.The crate already defines
CronErrorvariants. Using a typed error (e.g.,CronError::NotFound { id }) instead of constructing anio::Errorwithformat!()would align better with the crate's error handling patterns.♻️ Suggested approach
async fn cmd_get(client: CronClient, id: &str) -> Result<(), Box<dyn std::error::Error>> { match client.get_job(id).await? { Some(job) => println!("{}", serde_json::to_string_pretty(&job).unwrap()), None => { - return Err(std::io::Error::new( - std::io::ErrorKind::NotFound, - format!("job '{id}' not found"), - ) - .into()); + return Err(trogon_cron::CronError::NotFound { id: id.to_string() }.into()); } } Ok(()) }This requires adding a
NotFoundvariant toCronErrorif it doesn't already exist.As per coding guidelines: "Errors must be typed—use structs or enums, never
Stringorformat!()."🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@rsworkspace/crates/trogon-cron/src/main.rs` around lines 136 - 150, Replace the ad-hoc io::Error in cmd_get with the crate's typed CronError: update/ensure CronError has a NotFound { id: String } (or similar) variant, then return Err(CronError::NotFound { id: id.to_string() }.into()) when client.get_job(id).await? yields None; modify the cmd_get function to construct and return that CronError instead of using format!() so error handling is consistent with the rest of the crate.
🤖 Prompt for all review comments with AI agents
Verify each finding against the current code and only fix it if needed.
Inline comments:
In `@rsworkspace/crates/trogon-cron/src/main.rs`:
- Around line 159-168: The current map_err closures for
std::fs::read_to_string(file) and serde_json::from_str(&json) throw away the
original error by formatting it to a string; instead, add a CLI-specific error
type (e.g., an enum like CliError with variants ReadFile { path: String, source:
std::io::Error } and InvalidJobConfig { source: serde_json::Error }) or switch
the function to return a boxed error (Box<dyn std::error::Error>) / use
anyhow/thiserror, and change the map_err calls to wrap the original source error
into those variants (preserving the source) rather than calling format!;
reference the read_to_string(file) map_err closure and the
serde_json::from_str(&json) -> JobConfig mapping to locate and update the
conversions.
---
Duplicate comments:
In `@rsworkspace/crates/trogon-cron/src/error.rs`:
- Around line 2-33: CronError currently stores String payloads (Kv(String),
Publish(String), InvalidJobConfig { reason: String }) which discards typed error
context; change these variants to hold concrete error types or boxed error trait
objects (e.g., Kv(Box<dyn std::error::Error + Send + Sync>), Publish(Box<dyn
std::error::Error + Send + Sync>), and InvalidJobConfig(ValidationError) where
ValidationError is a new struct/enum describing validation failure), update the
Display impl to format human-readable messages from those types, and extend
std::error::Error::source() to return the wrapped error for Kv, Publish and
InvalidJobConfig so callers can access the original error chain (leave Serde and
Io handling as-is).
In `@rsworkspace/crates/trogon-cron/src/executor.rs`:
- Around line 223-281: The retry loop currently converts errors to String early
(last_error = e.to_string()), losing typed error context; change last_error to
hold the original typed error (e.g., Option<anyhow::Error> or Option<Box<dyn
std::error::Error + Send + Sync>>) and assign Some(e.into()) when
publisher.publish_tick(subject.clone(), headers, payload.clone().into()).await
returns Err(e), keep using that typed error for all logic, and only call
.to_string() (or format!) when emitting the final tracing::error and when
calling publish_dead_letter(&publisher, &tick, "publish", actual_attempts,
last_error_string). Ensure references to publisher.publish_tick, last_error,
actual_attempts, and publish_dead_letter are updated so the retry warn logs use
the typed error (or its display) without dropping error type until the terminal
path.
- Around line 279-281: The DLQ publish is gated by max_retries > 0 so terminal
failures when retries are disabled never get emitted; remove that condition and
always call publish_dead_letter(...) when execution ends in failure. Locate the
two guarded calls (the one using publish_dead_letter(&publisher, &tick,
"publish", actual_attempts, last_error).await and the duplicate at the later
block) and change the control flow so that whenever a terminal failure path is
taken (use the existing failure context variables: publisher, tick,
actual_attempts, last_error) you await publish_dead_letter(...) unconditionally
instead of only when max_retries > 0.
- Around line 184-201: The non-concurrent guard currently does a separate load()
then later store(true), which allows a race where two ticks both see false and
both spawn; change the logic in the RuntimeAction::Spawn branch to atomically
claim state.is_running using compare_exchange(false, true, Ordering::SeqCst,
Ordering::SeqCst) when !command.concurrent(), and only call spawn_process() if
the compare_exchange succeeds (otherwise log/debug skip). Also remove or adapt
the separate state.is_running.store(true) in the spawn_process / completion path
so that only the atomic claim sets the flag (and ensure the flag is cleared on
process end as before).
In `@rsworkspace/crates/trogon-cron/src/nats_impls.rs`:
- Around line 233-258: The leader-lock methods (NatsLeaderLock::try_acquire,
::renew, ::release) currently map KV errors to CronError::Kv(e.to_string()),
discarding the original typed error and its source chain; change the
CronError::Kv variant to hold the original error type (or a boxed dyn Error) and
map errors directly (e.g., .map_err(|e| CronError::Kv(e)) or .map_err(|e|
CronError::Kv { source: e }) instead of e.to_string()), so the errors returned
from self.store.create/update/delete preserve their concrete type and source.
- Around line 194-215: The TickPublisher impl for jetstream::Context currently
collapses PublishOutcome variants into CronError::Publish(String); instead
preserve and wrap the original typed errors instead of converting to
strings—update CronError to include variants/fields for PublishFailed,
AckFailed, AckTimedOut (with duration), and StoreFailed (or a single Publish
variant that contains the original PublishOutcome::... error type), then change
the matching in publish_tick (which calls publish_event and uses
JetStreamContextPublisher and PublishOutcome) to return
Err(CronError::PublishFailed(e)) / Err(CronError::AckFailed(e)) /
Err(CronError::AckTimedOut(timeout)) / Err(CronError::StoreFailed(e)) (or the
equivalent wrapper) so the original error types are preserved and chainable
instead of using to_string()/format!.
- Around line 36-157: The code currently stringifies SDK errors via map_err(|e|
CronError::Kv(e.to_string())) in NatsConfigStore (see put_job, get_job,
delete_job, list_jobs, load_and_watch), which loses error types; change
CronError::Kv to hold the concrete SDK error type (e.g., kv::Error or the
appropriate JetStream error) or add a new variant wrapping Box<dyn
std::error::Error + Send + Sync>, implement From<kv::Error> for CronError, then
remove the map_err closures and use the ? operator directly (e.g., let kv =
self.js.get_key_value(CONFIG_BUCKET).await?; and similarly for
kv.put(...).await? etc.), so SDK errors are propagated without being
stringified. Ensure any places that previously used CronError::Kv(String) are
updated to handle the new variant.
---
Nitpick comments:
In `@rsworkspace/crates/trogon-cron/src/config.rs`:
- Around line 7-31: Replace raw primitive fields (expr, subject, bin,
interval_sec, timeout_sec) in the public config enums/structs (Schedule, Action,
and JobConfig) with domain-specific value objects/newtypes that perform
validation at construction and implement Serde (e.g., CronExpr, NatsSubject,
ExecutablePath, PositiveSeconds); update Schedule::Interval to use
PositiveSeconds.interval_sec and Schedule::Cron to use CronExpr.expr, and
Action::Publish/Spawn to use NatsSubject.subject and ExecutablePath.bin, plus
PositiveSeconds.timeout_sec; ensure each newtype exposes a fallible constructor
that enforces invariants and Serde deserializes by invoking that constructor so
malformed configs fail at deserialization rather than later during JobConfig
conversion.
In `@rsworkspace/crates/trogon-cron/src/main.rs`:
- Around line 136-150: Replace the ad-hoc io::Error in cmd_get with the crate's
typed CronError: update/ensure CronError has a NotFound { id: String } (or
similar) variant, then return Err(CronError::NotFound { id: id.to_string()
}.into()) when client.get_job(id).await? yields None; modify the cmd_get
function to construct and return that CronError instead of using format!() so
error handling is consistent with the rest of the crate.
🪄 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: Organization UI
Review profile: CHILL
Plan: Pro
Run ID: 855f74ed-89ed-4cbb-b212-99cf8e932726
⛔ Files ignored due to path filters (1)
rsworkspace/Cargo.lockis excluded by!**/*.lock
📒 Files selected for processing (20)
devops/docker/compose/compose.ymldevops/docker/compose/services/trogon-cron/Dockerfilersworkspace/crates/acp-telemetry/src/service_name.rsrsworkspace/crates/trogon-cron/Cargo.tomlrsworkspace/crates/trogon-cron/src/cli.rsrsworkspace/crates/trogon-cron/src/client.rsrsworkspace/crates/trogon-cron/src/config.rsrsworkspace/crates/trogon-cron/src/domain.rsrsworkspace/crates/trogon-cron/src/error.rsrsworkspace/crates/trogon-cron/src/executor.rsrsworkspace/crates/trogon-cron/src/kv.rsrsworkspace/crates/trogon-cron/src/leader.rsrsworkspace/crates/trogon-cron/src/lib.rsrsworkspace/crates/trogon-cron/src/main.rsrsworkspace/crates/trogon-cron/src/mocks.rsrsworkspace/crates/trogon-cron/src/nats_impls.rsrsworkspace/crates/trogon-cron/src/scheduler.rsrsworkspace/crates/trogon-cron/src/traits.rsrsworkspace/crates/trogon-cron/tests/cron_unit.rsrsworkspace/crates/trogon-cron/tests/integration.rs
✅ Files skipped from review due to trivial changes (3)
- devops/docker/compose/services/trogon-cron/Dockerfile
- rsworkspace/crates/trogon-cron/Cargo.toml
- rsworkspace/crates/trogon-cron/src/leader.rs
🚧 Files skipped from review as they are similar to previous changes (4)
- rsworkspace/crates/trogon-cron/tests/cron_unit.rs
- rsworkspace/crates/trogon-cron/src/client.rs
- rsworkspace/crates/trogon-cron/src/mocks.rs
- rsworkspace/crates/trogon-cron/src/traits.rs
There was a problem hiding this comment.
Actionable comments posted: 3
♻️ Duplicate comments (3)
rsworkspace/crates/trogon-cron/src/domain.rs (1)
296-304:⚠️ Potential issue | 🟠 MajorReject malformed NATS subjects in
PublishSubject.This factory still accepts invalid subjects as long as they start with
cron.. Inputs likecron.,cron..backup,cron.*, orcron.bad subjectbecome valid domain objects here and only fail later when the scheduler tries to publish. Tighten this value object so construction guarantees a real NATS subject, not just the prefix. As per coding guidelines: "Prefer domain-specific value objects over primitives (e.g.,AcpPrefixnotString). Each type's factory must guarantee correctness at construction—invalid instances should be unrepresentable."#!/bin/bash # Inspect the current PublishSubject factory and the existing NATS token helpers. sed -n '296,309p' rsworkspace/crates/trogon-cron/src/domain.rs fd -i 'token.rs' rsworkspace/crates/trogon-nats -x sed -n '1,220p' {} rg -n 'PublishSubject|validate|subject' rsworkspace/crates/trogon-cron/src rsworkspace/crates/trogon-nats/src🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@rsworkspace/crates/trogon-cron/src/domain.rs` around lines 296 - 304, The PublishSubject::new constructor currently only checks the "cron." prefix; tighten it to reject empty or malformed NATS subjects by validating the remainder after "cron.": ensure there's at least one non-empty segment after the prefix, disallow consecutive dots or trailing dots (reject "cron.", "cron..backup"), forbid wildcard tokens like '*' or '>' and any whitespace, and enforce the same token rules used by the NATS helpers in the trogon-nats crate (see token validation helpers in token.rs) so subject components are valid identifiers; if validation fails, return Err(CronError::invalid_job_config(JobConfigError::PublishSubjectPrefix { subject })) as before.rsworkspace/crates/trogon-cron/src/nats_impls.rs (1)
35-40: 🛠️ Refactor suggestion | 🟠 MajorKeep the production trait impls as SDK passthroughs.
NatsConfigStoreandNatsLeaderLockstill do error adaptation inside the trait impls viaCronError::kv_source(...)/map_err(...). That preserves context, but it still violates the repo rule that production infrastructure trait bodies should be thin passthroughs to the SDK. Consider moving this translation above the trait boundary or reshaping the traits so the impls can expose the SDK error/return types directly. As per coding guidelines: "Production implementations of infrastructure traits must be zero-cost passthroughs to the underlying SDK. No error wrapping (use the SDK's error types directly via associatedtype Error), no return type conversion (add associated types liketype Infoto match the SDK's return), nomap_err, nomap(|_| ())."#!/bin/bash # Inspect the trait signatures and the current production impl bodies. sed -n '1,240p' rsworkspace/crates/trogon-cron/src/traits.rs sed -n '1,260p' rsworkspace/crates/trogon-cron/src/nats_impls.rsAlso applies to: 44-99, 101-148, 228-253
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@rsworkspace/crates/trogon-cron/src/nats_impls.rs` around lines 35 - 40, The production impls (e.g., NatsConfigStore::config_bucket and NatsLeaderLock impls) are wrapping SDK errors with CronError via map_err/CronError::kv_source inside trait bodies; change the trait signatures to expose the SDK types/errors directly (add associated types like type Error and type Info to the trait in traits.rs) and make the impls zero-cost passthroughs that return the SDK's Result/values without map_err or wrapping; move the CronError translation to the call sites or adapter layer above the trait boundary so functions like config_bucket simply call self.js.get_key_value(CONFIG_BUCKET).await and return the SDK result as-is.rsworkspace/crates/trogon-cron/src/kv.rs (1)
36-48:⚠️ Potential issue | 🟠 MajorDon't treat any same-name JetStream resource as compatible.
After an
already existscreate failure, these helpers only prove the bucket/stream name exists. They never verify that the existing resource still has the requested history,max_age, or subjects. That can silently accept a leader bucket with the wrong TTL or tick/DLQ streams wired to the wrong subject set. Fetch the existing config and fail if the important fields differ.#!/bin/bash # Show the desired configs and the current already-exists fallback paths. sed -n '24,127p' rsworkspace/crates/trogon-cron/src/kv.rsAlso applies to: 56-106, 108-127
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@rsworkspace/crates/trogon-cron/src/kv.rs` around lines 36 - 48, The current get_or_create_leader_bucket helper (and the other helpers using get_or_create with kv::Config like the blocks at 56-106 and 108-127) only treats an "already exists" result as success without validating the existing resource; update the logic so that after a create failure due to already existing you fetch the existing bucket/stream config from JetStream (via js) and compare the important fields (for kv::Config: history and max_age; for streams also verify subjects and retention/replicas/etc. as appropriate) against the requested kv::Config/stream config used in get_or_create; if any critical field mismatches (e.g., history != 1, max_age != Duration::from_secs(10), wrong subjects) return a CronError indicating a config mismatch instead of silently accepting the existing resource. Ensure this validation is applied in get_or_create_leader_bucket and the similar helper blocks referenced so you fail fast when an incompatible same-name resource exists.
🤖 Prompt for all review comments with AI agents
Verify each finding against the current code and only fix it if needed.
Inline comments:
In `@rsworkspace/crates/trogon-cron/src/executor.rs`:
- Around line 63-68: The multiplication in retry_backoff(base_sec, attempt) can
overflow; replace base_sec * multiplier with a saturation or capped value to
avoid panics/wraps — e.g., compute let secs =
base_sec.saturating_mul(multiplier) (or use checked_mul and clamp to a defined
MAX_RETRY_BACKOFF_SECS constant) and then return Duration::from_secs(secs);
update retry_backoff to use base_sec.saturating_mul(multiplier) or checked_mul +
min(secs, MAX) to keep the retry delay well-defined.
In `@rsworkspace/crates/trogon-cron/src/kv.rs`:
- Around line 163-199: The current 500ms deadline can truncate the watcher
initial-snapshot; change the logic in the loop that drains the initial snapshot
(the tokio::select! using deadline, tokio::pin!(deadline), and the _ = &mut
deadline branch) to instead rely on the watcher entries' e.delta == 0 as the
end-of-snapshot marker: keep consuming watcher.next() until you see Some(Ok(e))
with e.delta == 0 (or None/Err handled as now), deserialize JobConfig on
kv::Operation::Put, and only treat an empty bucket specially (e.g., if watcher
yields None immediately) or load a point-in-time snapshot before attaching the
watch; if you must keep a timeout for defense, make it much larger and clearly
document it.
In `@rsworkspace/crates/trogon-cron/src/nats_impls.rs`:
- Around line 79-99: list_jobs currently swallows deserialization failures and
omits corrupt entries; change the logic in list_jobs (and where it calls
config_bucket(), kv.keys(), kv.get(), and serde_json::from_slice::<JobConfig>)
to fail fast or surface the offending key instead of silently skipping: after
obtaining the value from kv.get(&key).await, attempt serde_json::from_slice and
map any Err into a CronError that includes the key (e.g. via
CronError::kv_source or a new CronError variant) so the function returns an
error containing the key and underlying serde error rather than dropping the
entry; ensure any early returns use the existing CronError type so callers see
corrupted configs.
---
Duplicate comments:
In `@rsworkspace/crates/trogon-cron/src/domain.rs`:
- Around line 296-304: The PublishSubject::new constructor currently only checks
the "cron." prefix; tighten it to reject empty or malformed NATS subjects by
validating the remainder after "cron.": ensure there's at least one non-empty
segment after the prefix, disallow consecutive dots or trailing dots (reject
"cron.", "cron..backup"), forbid wildcard tokens like '*' or '>' and any
whitespace, and enforce the same token rules used by the NATS helpers in the
trogon-nats crate (see token validation helpers in token.rs) so subject
components are valid identifiers; if validation fails, return
Err(CronError::invalid_job_config(JobConfigError::PublishSubjectPrefix { subject
})) as before.
In `@rsworkspace/crates/trogon-cron/src/kv.rs`:
- Around line 36-48: The current get_or_create_leader_bucket helper (and the
other helpers using get_or_create with kv::Config like the blocks at 56-106 and
108-127) only treats an "already exists" result as success without validating
the existing resource; update the logic so that after a create failure due to
already existing you fetch the existing bucket/stream config from JetStream (via
js) and compare the important fields (for kv::Config: history and max_age; for
streams also verify subjects and retention/replicas/etc. as appropriate) against
the requested kv::Config/stream config used in get_or_create; if any critical
field mismatches (e.g., history != 1, max_age != Duration::from_secs(10), wrong
subjects) return a CronError indicating a config mismatch instead of silently
accepting the existing resource. Ensure this validation is applied in
get_or_create_leader_bucket and the similar helper blocks referenced so you fail
fast when an incompatible same-name resource exists.
In `@rsworkspace/crates/trogon-cron/src/nats_impls.rs`:
- Around line 35-40: The production impls (e.g., NatsConfigStore::config_bucket
and NatsLeaderLock impls) are wrapping SDK errors with CronError via
map_err/CronError::kv_source inside trait bodies; change the trait signatures to
expose the SDK types/errors directly (add associated types like type Error and
type Info to the trait in traits.rs) and make the impls zero-cost passthroughs
that return the SDK's Result/values without map_err or wrapping; move the
CronError translation to the call sites or adapter layer above the trait
boundary so functions like config_bucket simply call
self.js.get_key_value(CONFIG_BUCKET).await and return the SDK result as-is.
🪄 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: Organization UI
Review profile: CHILL
Plan: Pro
Run ID: c4dd3c74-8f61-46c5-aef2-1ac766fb5f6b
📒 Files selected for processing (11)
rsworkspace/crates/trogon-cron/src/client.rsrsworkspace/crates/trogon-cron/src/domain.rsrsworkspace/crates/trogon-cron/src/error.rsrsworkspace/crates/trogon-cron/src/executor.rsrsworkspace/crates/trogon-cron/src/kv.rsrsworkspace/crates/trogon-cron/src/lib.rsrsworkspace/crates/trogon-cron/src/main.rsrsworkspace/crates/trogon-cron/src/nats_impls.rsrsworkspace/crates/trogon-cron/src/scheduler.rsrsworkspace/crates/trogon-cron/tests/cron_unit.rsrsworkspace/crates/trogon-cron/tests/integration.rs
✅ Files skipped from review due to trivial changes (2)
- rsworkspace/crates/trogon-cron/src/lib.rs
- rsworkspace/crates/trogon-cron/src/client.rs
| fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { | ||
| match self { | ||
| Self::ScheduleNotFound { id } => write!(formatter, "schedule '{id}' does not exist"), | ||
| Self::ScheduleDeleted { id } => write!(formatter, "schedule '{id}' was deleted"), |
There was a problem hiding this comment.
Error Display impls regressed to Debug format
Medium Severity
RemoveScheduleError and ResumeScheduleError had their Display impls replaced with write!(formatter, "{self:?}"), producing raw debug output like ScheduleNotFound { id: ScheduleId("backup") } instead of the previous user-friendly messages (e.g., "schedule 'backup' does not exist"). This is inconsistent with peer error types CreateScheduleError and PauseScheduleError which still implement proper match-based Display formatting.
Additional Locations (1)
Reviewed by Cursor Bugbot for commit d881b74. Configure here.
|
|
||
| fn pause_job_command(id: &str) -> PauseSchedule { | ||
| PauseSchedule::new(ScheduleId::parse(id).unwrap()) | ||
| } |
There was a problem hiding this comment.
Duplicate function definition in pause_schedule tests
High Severity
The test module defines pause_job_command twice — once at the original location and again at a new location. This will cause a compilation error. The second definition appears to be a copy-paste artifact from adding create_job_command alongside helper functions from another test module.
Reviewed by Cursor Bugbot for commit dab428f. Configure here.
|
|
||
| fn pause_job_command(id: &str) -> PauseSchedule { | ||
| PauseSchedule::new(ScheduleId::parse(id).unwrap()) | ||
| } |
There was a problem hiding this comment.
Duplicate function and missing reference in test module
High Severity
The test module defines pause_job_command twice (lines 110 and 131), and create_job_command calls a nonexistent create_schedule function. This appears to be a copy-paste merge artifact — the old helper functions were replaced with job() but stale copies of create_job_command and a second pause_job_command were left behind, preventing the test module from compiling.
Reviewed by Cursor Bugbot for commit a557de5. Configure here.
| &self.schedule_publisher, | ||
| ).await?; | ||
| desired_schedules = rebuilt_jobs; | ||
| scheduler_watcher = watcher; |
There was a problem hiding this comment.
Initial leader establishment error crashes instead of retrying
Medium Severity
When the controller first becomes leader, establish_scheduler_processor is called with ?, propagating any transient error (e.g., a momentary NATS hiccup) out of run_until, which terminates the entire scheduler process. In contrast, when the watcher later breaks, reestablish_scheduler_processor is used — which retries with backoff and respects the shutdown signal. The initial establishment path lacks this same resilience, so a brief network blip during first leadership acquisition kills the controller instead of retrying.
Additional Locations (1)
Reviewed by Cursor Bugbot for commit c053082. Configure here.
| } | ||
| Some(ScheduleKind::Rrule(inner)) => { | ||
| let next = next_rrule_occurrence(inner, now)?; | ||
| Ok((format!("@at {}", next.to_rfc3339()), None)) |
There was a problem hiding this comment.
RRULE schedules fire once only
High Severity
RRULE schedules are turned into a single Nats-Schedule @at timestamp for the next occurrence only, while cron and @every expressions stay recurring on NATS. Nothing in the leader controller re-publishes after a fire, so multi-occurrence RRULEs (for example daily rules) stop after the first execution.
Reviewed by Cursor Bugbot for commit 624a450. Configure here.
| events, | ||
| state, | ||
| }); | ||
| } |
There was a problem hiding this comment.
No-stream path skips decide state
Medium Severity
For deciders with WritePrecondition::NoStream (including CreateSchedule), command execution now calls decide on initial_state() without reading the event stream. Existing or deleted schedule IDs still look “missing” at decide time, so callers get append/OCC failures instead of AlreadyExists or ScheduleDeleted, and domain rules in decide never run against real stream state.
Additional Locations (1)
Reviewed by Cursor Bugbot for commit 75d9733. Configure here.
| ) | ||
| .await?; | ||
|
|
||
| Ok(outcome) |
There was a problem hiding this comment.
Append succeeds before projection
Medium Severity
EventStore::append_stream persists events to JetStream first, then updates the schedules KV projection. If project_appended_events fails partway through a batch, events remain on the stream while the read model is stale or partially updated, and the command returns an error despite a successful append.
Additional Locations (1)
Reviewed by Cursor Bugbot for commit 75d9733. Configure here.
Signed-off-by: Yordis Prieto <yordis.prieto@gmail.com>
…-headers metadata Event payloads no longer carry timestamps or actor IDs; that data lives in the envelope headers, so the schedule event contracts and their consumers needed to follow. Signed-off-by: Yordis Prieto <yordis.prieto@gmail.com>
Align the command name with the ScheduleCreated event and schedule_created_from_job helper.
Signed-off-by: Yordis Prieto <yordis.prieto@gmail.com>
Align command-domain vocabulary with the read model so aggregates, errors, and helpers consistently use Schedule terminology.
Expose processors, projections, queries, and pause/remove/resume deciders; fix Schedule→ScheduleEventSchedule conversion and align Cargo deps for a buildable trogon-scheduler binary and tests.
Signed-off-by: Yordis Prieto <yordis.prieto@gmail.com>
Signed-off-by: Yordis Prieto <yordis.prieto@gmail.com>
Signed-off-by: Yordis Prieto <yordis.prieto@gmail.com>
Drop the scheduler controller, processors module, CLI, runtime config, and main binary, along with their dependencies on clap and confique. The crate is now a library only, with controller/leader-election logic removed pending a different orchestration approach. Signed-off-by: Yordis Prieto <yordis.prieto@gmail.com>
| pub const LEADER_KEY: &str = "lock"; | ||
| pub const EVENTS_STREAM: &str = "SCHEDULER_EVENTS"; | ||
| pub const EVENTS_SUBJECT_PREFIX: &str = "scheduler.schedules.events."; | ||
| pub const EVENTS_SUBJECT_PATTERN: &str = "scheduler.schedules.events.>"; |
There was a problem hiding this comment.
Event store worker stream mismatch
High Severity
Persisted schedule commands append to JetStream stream SCHEDULER_EVENTS on subjects like scheduler.schedules.events.{id}, while the execution processor’s durable consumer listens on SCHEDULER_SCHEDULE_EVENTS with filter scheduler.schedules.events.v1.>. Those names do not overlap, so reconciliation never sees command-appended events and enabled schedules are not published to execution.
Additional Locations (1)
Reviewed by Cursor Bugbot for commit ae4d6d6. Configure here.
Drop the `ResolvedSchedule`, `SchedulePublisher`, `LeaderLock` traits, and their associated KV constants and tests. These were carryovers from an earlier design where the scheduler directly managed NATS schedule streams and leader election; the current architecture handles delivery through the events stream alone. Signed-off-by: Yordis Prieto <yordis.prieto@gmail.com>
| commit_watched_projection_state(&mut state, stream_id, next_state); | ||
| ack_watch_message(&message).await; | ||
| if let Some(change) = change { | ||
| return Some((change_from_projection_change(change), (state, subscriber, kv))); |
There was a problem hiding this comment.
Watcher acks skipped projection errors
Medium Severity
In load_and_watch_schedules, when prepare_watched_projection_change cannot apply an event (decode/validation/state transition failure), it returns None, the loop still calls ack_watch_message, and the in-memory watcher state never advances for that JetStream message. The KV projection and live subscribers can miss updates while the consumer treats the message as processed.
Additional Locations (1)
Reviewed by Cursor Bugbot for commit 550f0e0. Configure here.
Signed-off-by: Yordis Prieto <yordis.prieto@gmail.com>
There was a problem hiding this comment.
Cursor Bugbot has reviewed your changes using default effort and found 1 potential issue.
There are 47 total unresolved issues (including 46 from previous reviews).
❌ Bugbot Autofix is OFF. To automatically fix reported issues with cloud agents, have a team admin enable autofix in the Cursor dashboard.
Reviewed by Cursor Bugbot for commit 8123459. Configure here.
| .iter() | ||
| .map(|header| (header.name.clone(), header.value.clone())), | ||
| ), | ||
| } |
There was a problem hiding this comment.
Projection skips header validation
Medium Severity
project_message builds read-model headers with MessageHeaders::from_pairs, which does not validate names or values, while MessageHeaders::deserialize uses MessageHeaders::new and rejects invalid pairs. Projected JSON can be written to KV but later get_schedule / list_schedules calls fail deserialization on the same record.
Additional Locations (1)
Reviewed by Cursor Bugbot for commit 8123459. Configure here.
yordis
changed the title
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
2 participants
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.
Uh oh!
There was an error while loading. Please reload this page.