feat: Introduce Job Manager to track long running tasks#1100
Conversation
|
Important Review skippedAuto incremental reviews are disabled on this repository. Please check the settings in the CodeRabbit UI or the ⚙️ Run configurationConfiguration used: Organization UI Review profile: CHILL Plan: Pro Run ID: You can disable this status message by setting the Use the checkbox below for a quick retry:
WalkthroughThis change introduces a persisted background-job system backed by Kronos. Reduce and priority recomputation now submit asynchronous jobs, job callbacks execute workflows and track status, and workspace-scoped APIs list, inspect, and cancel jobs. ChangesBackground job system
Estimated code review effort: 5 (Critical) | ~120 minutes Sequence Diagram(s)sequenceDiagram
participant Client
participant API
participant Kronos
participant JobDispatch
participant Database
Client->>API: Submit reduce or recompute request
API->>Database: Create background job
API->>Kronos: Schedule job
Kronos->>JobDispatch: Invoke job callback
JobDispatch->>Database: Update status and logs
JobDispatch-->>Kronos: Complete execution
Possibly related PRs
Suggested labels: Suggested reviewers: Poem
🚥 Pre-merge checks | ✅ 4 | ❌ 1❌ Failed checks (1 warning)
✅ Passed checks (4 passed)
✨ Finishing Touches🧪 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 |
types Signed-off-by: datron <Datron@users.noreply.github.com>
Signed-off-by: datron <Datron@users.noreply.github.com>
Signed-off-by: datron <Datron@users.noreply.github.com>
Signed-off-by: datron <Datron@users.noreply.github.com>
Signed-off-by: datron <Datron@users.noreply.github.com>
Signed-off-by: datron <Datron@users.noreply.github.com>
Signed-off-by: datron <Datron@users.noreply.github.com>
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 8
🧹 Nitpick comments (9)
crates/superposition/src/jobs/handlers.rs (2)
110-149: 🎯 Functional Correctness | 🔵 Trivial | 🏗️ Heavy liftCancellation is recorded as
Failed, losing the distinction from an actual execution failure.
BackgroundJobStatushas noCancelledvariant (Created, Scheduled, Inprogress, Failed, Completed), socancel_handlermarks cancelled jobs asFailed. Clients/UIs can no longer tell a deliberate cancellation apart from a genuine failure in job history. Fixing this cleanly requires a schema/enum change incrates/superposition_types/src/database/models.rs(out of this file's scope), so flagging for awareness rather than blocking.🤖 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 `@crates/superposition/src/jobs/handlers.rs` around lines 110 - 149, The cancel_handler flow records successful cancellations as BackgroundJobStatus::Failed, conflating them with execution failures. This requires adding a distinct Cancelled status to BackgroundJobStatus and updating the related database schema/model in superposition_types, then use that status in cancel_handler after Kronos cancellation succeeds; preserve Failed for actual cancellation errors.
28-41: 🚀 Performance & Scalability | 🔵 Trivial | ⚡ Quick winNo pagination on job listing.
list_jobs(vialist_handler) returns every matching job for the workspace with no limit/offset. Asjob_manageraccumulates history this list will grow unbounded per request.🤖 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 `@crates/superposition/src/jobs/handlers.rs` around lines 28 - 41, Update list_handler and the underlying list_jobs query to accept pagination parameters, applying a bounded limit and offset when fetching jobs. Extend JobListFilters or the existing request model with validated pagination fields and preserve the current workspace, job_type, and status filters while returning only the requested page.crates/superposition_types/src/database/models.rs (2)
311-338: 📐 Maintainability & Code Quality | 🔵 Trivial | 💤 Low valueUse idiomatic CamelCase for the
InProgressvariant.The
Inprogressvariant can be renamed toInProgressto follow standard Rust conventions. Because#[serde(rename_all = "UPPERCASE")]andDbValueStyle = "UPPERCASE"are used, both names evaluate to the identical"INPROGRESS"representation over the wire and in the database, making this a completely safe stylistic improvement.♻️ Proposed refactor
pub enum BackgroundJobStatus { Created, Scheduled, - Inprogress, + InProgress, Failed, Completed, }🤖 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 `@crates/superposition_types/src/database/models.rs` around lines 311 - 338, Rename the BackgroundJobStatus::Inprogress enum variant to BackgroundJobStatus::InProgress, preserving the existing serde and database UPPERCASE representations so the serialized value remains "INPROGRESS".
395-418: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick winImplement
Frominstead ofTryFromfor infallible conversions.Since the conversion from
StringtoJobWorkspacenever fails, it is more idiomatic to implementFrom<String>rather thanTryFrom<String>. This prevents downstream callers from unnecessarily handling impossible error states via.unwrap().♻️ Proposed refactor
-impl TryFrom<String> for JobWorkspace { - type Error = String; - fn try_from(value: String) -> Result<Self, Self::Error> { - if value == Self::GLOBAL { - Ok(Self::Global) - } else { - Ok(Self::Workspace(value)) - } - } -} +impl From<String> for JobWorkspace { + fn from(value: String) -> Self { + if value == Self::GLOBAL { + Self::Global + } else { + Self::Workspace(value) + } + } +} impl From<&JobWorkspace> for String { fn from(value: &JobWorkspace) -> Self { value.as_db_string() } } #[cfg(feature = "disable_db_data_validation")] impl DisableDBValidation for JobWorkspace { type Source = String; fn from_db_unvalidated(data: Self::Source) -> Self { - Self::try_from(data).unwrap_or(Self::Global) + Self::from(data) } }🤖 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 `@crates/superposition_types/src/database/models.rs` around lines 395 - 418, Replace the infallible TryFrom<String> implementation for JobWorkspace with From<String>, preserving the existing GLOBAL-to-Global mapping and Workspace fallback. Update from_db_unvalidated to use the new infallible conversion directly instead of unwrap_or, while keeping its current behavior.crates/superposition_types/migrations/2026-07-16-000001_job_manager/down.sql (1)
3-8: 📐 Maintainability & Code Quality | 🔵 Trivial | 💤 Low valueRemove redundant
DROP INDEXstatements.In PostgreSQL, dropping a table automatically cascades and drops all of its associated indexes. Since
superposition.job_manageris dropped on line 1, these statements are effectively no-ops.♻️ Proposed refactor
-DROP INDEX IF EXISTS superposition.idx_job_manager_workspace_schema; -DROP INDEX IF EXISTS superposition.idx_job_manager_status; -DROP INDEX IF EXISTS superposition.idx_job_manager_type; -DROP INDEX IF EXISTS superposition.idx_job_manager_kronos_job_id; -DROP INDEX IF EXISTS superposition.idx_job_manager_status_job_type; -DROP INDEX IF EXISTS superposition.idx_job_manager_created_at;🤖 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 `@crates/superposition_types/migrations/2026-07-16-000001_job_manager/down.sql` around lines 3 - 8, Remove the redundant DROP INDEX statements from the down migration, leaving only the existing superposition.job_manager table drop because PostgreSQL removes its associated indexes automatically.crates/superposition_types/src/database/superposition_schema.rs (1)
81-98: 🚀 Performance & Scalability | 🔵 TrivialConsider indexing
workspace_schemaon the physicaljob_managertable.Every per-workspace
job_managerview filters onWHERE workspace_schema = '{workspace}', but the migration only creates indexes onjob_typeand(status, job_type)— none coveringworkspace_schema. As this table accumulates jobs across all workspaces, every workspace-scoped listing/detail query will require a sequential scan filtered byworkspace_schema. Consider addingCREATE INDEX ... ON superposition.job_manager (workspace_schema)(or a composite index withcreated_at/statusto match query patterns) in the migration.🤖 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 `@crates/superposition_types/src/database/superposition_schema.rs` around lines 81 - 98, Add a migration index for the physical superposition.job_manager table on workspace_schema, optionally incorporating created_at or status if needed to match existing workspace-scoped query patterns. Keep the schema declaration in superposition_schema.rs aligned with the migrated table and ensure per-workspace job queries can use the new index.smithy/models/context.smithy (1)
262-278: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick winRemove now-unused
WeightRecomputeResponse/WeightRecomputeResponses.Since
WeightRecompute(lines 280-291) now returnsJobCreateResponseinstead ofWeightRecomputeResponses, these structures are orphaned in the model and no longer referenced anywhere in this file. Leaving them in generates unused types across every SDK target.♻️ Proposed cleanup
-structure WeightRecomputeResponse for Context { - `@required` - $id - - `@required` - condition: Condition - - `@required` - old_weight: Weight - - `@required` - new_weight: Weight -} - -list WeightRecomputeResponses { - member: WeightRecomputeResponse -} -🤖 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 `@smithy/models/context.smithy` around lines 262 - 278, Remove the unused WeightRecomputeResponse structure and WeightRecomputeResponses list from the Context model, leaving the WeightRecompute operation and its JobCreateResponse return type unchanged.crates/service_utils/src/kronos_dispatch.rs (2)
446-457: 🩺 Stability & Availability | 🔵 Trivial | 💤 Low valueMinor: eager
Utc::now()and non-atomic log merge.key.unwrap_or(Utc::now().to_rfc3339())computes the timestamp even whenkeyisSome; preferunwrap_or_else(|| Utc::now().to_rfc3339()). Separately, the read-then-write oflogsis not atomic — concurrent appends (or a progress write interleaving) can drop entries under last-writer-wins. A singlejsonb ||update would avoid the RMW race.♻️ unwrap_or_else
- let timed_key = key.unwrap_or(Utc::now().to_rfc3339()); + let timed_key = key.unwrap_or_else(|| Utc::now().to_rfc3339());🤖 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 `@crates/service_utils/src/kronos_dispatch.rs` around lines 446 - 457, Update the log-key fallback in the surrounding dispatch function to use lazy evaluation via unwrap_or_else, then replace the current logs read/merge/write sequence with a single atomic database update using PostgreSQL JSONB concatenation (jsonb ||). Preserve the existing keyed log entry and job_id filtering while ensuring concurrent appends cannot overwrite one another.
295-334: 🗄️ Data Integrity & Integration | 🔵 Trivial | ⚖️ Poor tradeoffNon-transactional insert + remote create + update leaves orphan rows. The
CREATEDrow is inserted, thencreate_jobawaits a network call, then the row is updated toSCHEDULED. If the process dies (or theSCHEDULEDupdate?-fails) aftercreate_jobsucceeds, the BJM row is stuck inCREATEDwhile a Kronos job actually runs. Consider a reconciliation/sweep for staleCREATEDrows, or persistkronos_job_id/status via a more robust ordering.🤖 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 `@crates/service_utils/src/kronos_dispatch.rs` around lines 295 - 334, The job dispatch flow around the BJM insert and `kronos_client.create_job` can leave rows stuck in CREATED after the remote job succeeds. Add reconciliation for stale CREATED rows that can recover or reschedule jobs using the persisted request data, or otherwise reorder/persist the Kronos job ID and status so successful remote creation cannot leave an unrecoverable orphan; preserve the existing SCHEDULED update for successful dispatches.
🤖 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 `@crates/context_aware_config/src/api/config/handlers.rs`:
- Around line 490-495: Update reduce_handler to accept the request’s approve
value and construct ReduceRequest with that value instead of using
ReduceRequest::default(). Preserve forwarding the populated ReduceRequest
through JobRequest::Reduce so the worker receives the caller’s approval flag.
In `@crates/service_utils/src/kronos_dispatch.rs`:
- Around line 419-431: The update logic in update_job_progress must always
persist terminal progress of 100, even when the difference from
previous_progress is within threshold. Add a forced-write condition for progress
== 100 while preserving the existing threshold-based behavior for non-terminal
progress.
In `@crates/superposition_types/src/api/jobs.rs`:
- Around line 51-57: Update JobDispatchRequest to remove #[serde(flatten)] from
job_request and represent JobRequest as a nested payload compatible with its
adjacent tagging. Ensure dispatch_job_handler can deserialize the request body
successfully while preserving the existing job_id field and JobRequest data.
In `@crates/superposition/src/dispatch/handlers.rs`:
- Around line 104-123: Update the completion and failure branches in the result
handling around append_job_logs so they append new timestamped entries instead
of reusing the original Job started key. Pass no reused timed_key to both calls,
while preserving the existing log messages and status updates.
- Around line 66-102: Update dispatch_job_handler to resolve and validate the
job using both workspace_context and job_id before calling update_job_status,
ensuring the mutation is workspace-scoped. Preserve the initial "Job started"
log key, and pass None or a distinct key for the terminal append_job_logs call
so it appends rather than replaces the existing entry.
In `@crates/superposition/src/jobs/handlers.rs`:
- Around line 28-41: Update list_handler to map each WorkspaceJobView returned
by list_jobs through JobResponse::from_view(...) before wrapping the collection
in Json, and change the handler’s response type to Vec<JobResponse>. Preserve
the existing filtering and error handling.
In `@smithy/models/jobs.smithy`:
- Around line 11-20: Update the job response model field corresponding to
BackgroundJobType so JobResponse serializes and deserializes it as “type”,
matching JobDetailResponse and JobSummary; use the existing field name with the
appropriate serde rename attribute or rename the field directly.
- Around line 141-160: Update the ListJobs response handling in the jobs handler
so it returns the Smithy-declared output shape with a required data wrapper
containing the job list, rather than returning a bare Vec<WorkspaceJobView>.
Preserve the existing job retrieval and filtering behavior while aligning the
serialized response with the ListJobs output definition.
---
Nitpick comments:
In `@crates/service_utils/src/kronos_dispatch.rs`:
- Around line 446-457: Update the log-key fallback in the surrounding dispatch
function to use lazy evaluation via unwrap_or_else, then replace the current
logs read/merge/write sequence with a single atomic database update using
PostgreSQL JSONB concatenation (jsonb ||). Preserve the existing keyed log entry
and job_id filtering while ensuring concurrent appends cannot overwrite one
another.
- Around line 295-334: The job dispatch flow around the BJM insert and
`kronos_client.create_job` can leave rows stuck in CREATED after the remote job
succeeds. Add reconciliation for stale CREATED rows that can recover or
reschedule jobs using the persisted request data, or otherwise reorder/persist
the Kronos job ID and status so successful remote creation cannot leave an
unrecoverable orphan; preserve the existing SCHEDULED update for successful
dispatches.
In
`@crates/superposition_types/migrations/2026-07-16-000001_job_manager/down.sql`:
- Around line 3-8: Remove the redundant DROP INDEX statements from the down
migration, leaving only the existing superposition.job_manager table drop
because PostgreSQL removes its associated indexes automatically.
In `@crates/superposition_types/src/database/models.rs`:
- Around line 311-338: Rename the BackgroundJobStatus::Inprogress enum variant
to BackgroundJobStatus::InProgress, preserving the existing serde and database
UPPERCASE representations so the serialized value remains "INPROGRESS".
- Around line 395-418: Replace the infallible TryFrom<String> implementation for
JobWorkspace with From<String>, preserving the existing GLOBAL-to-Global mapping
and Workspace fallback. Update from_db_unvalidated to use the new infallible
conversion directly instead of unwrap_or, while keeping its current behavior.
In `@crates/superposition_types/src/database/superposition_schema.rs`:
- Around line 81-98: Add a migration index for the physical
superposition.job_manager table on workspace_schema, optionally incorporating
created_at or status if needed to match existing workspace-scoped query
patterns. Keep the schema declaration in superposition_schema.rs aligned with
the migrated table and ensure per-workspace job queries can use the new index.
In `@crates/superposition/src/jobs/handlers.rs`:
- Around line 110-149: The cancel_handler flow records successful cancellations
as BackgroundJobStatus::Failed, conflating them with execution failures. This
requires adding a distinct Cancelled status to BackgroundJobStatus and updating
the related database schema/model in superposition_types, then use that status
in cancel_handler after Kronos cancellation succeeds; preserve Failed for actual
cancellation errors.
- Around line 28-41: Update list_handler and the underlying list_jobs query to
accept pagination parameters, applying a bounded limit and offset when fetching
jobs. Extend JobListFilters or the existing request model with validated
pagination fields and preserve the current workspace, job_type, and status
filters while returning only the requested page.
In `@smithy/models/context.smithy`:
- Around line 262-278: Remove the unused WeightRecomputeResponse structure and
WeightRecomputeResponses list from the Context model, leaving the
WeightRecompute operation and its JobCreateResponse return type unchanged.
🪄 Autofix (Beta)
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: Organization UI
Review profile: CHILL
Plan: Pro
Run ID: bac57646-3e64-4b45-8aa7-a0aa649ea4c6
📒 Files selected for processing (30)
crates/context_aware_config/src/api/config.rscrates/context_aware_config/src/api/config/handlers.rscrates/context_aware_config/src/api/context.rscrates/context_aware_config/src/api/context/handlers.rscrates/service_utils/src/kronos_dispatch.rscrates/service_utils/src/middlewares/auth_n.rscrates/superposition/src/app_state.rscrates/superposition/src/dispatch/handlers.rscrates/superposition/src/jobs.rscrates/superposition/src/jobs/handlers.rscrates/superposition/src/main.rscrates/superposition/src/workspace/handlers.rscrates/superposition_types/migrations/2026-07-16-000001_job_manager/down.sqlcrates/superposition_types/migrations/2026-07-16-000001_job_manager/up.sqlcrates/superposition_types/src/api.rscrates/superposition_types/src/api/jobs.rscrates/superposition_types/src/database/models.rscrates/superposition_types/src/database/models/others.rscrates/superposition_types/src/database/schema.rscrates/superposition_types/src/database/superposition_schema.rscrates/superposition_types/src/lib.rsdocker-compose/postgres/db_init.sqlkronos_setup.sqlsmithy/models/config.smithysmithy/models/context.smithysmithy/models/jobs.smithysmithy/models/main.smithysmithy/smithy-build.jsonsuperposition.sqlworkspace_template.sql
| #[derive(Debug, Clone, Serialize, Deserialize)] | ||
| pub struct JobDispatchRequest { | ||
| #[serde(with = "crate::database::models::i64_formatter")] | ||
| pub job_id: i64, | ||
| #[serde(flatten)] | ||
| pub job_request: JobRequest, | ||
| } |
There was a problem hiding this comment.
🎯 Functional Correctness | 🔴 Critical | ⚡ Quick win
🧩 Analysis chain
🏁 Script executed:
#!/bin/bash
set -euo pipefail
echo "== relevant files =="
git ls-files | rg 'crates/superposition_types/src/api/jobs\.rs|crates/superposition_types/src/api|dispatch_job_handler|JobDispatchRequest|JobRequest|job_type|job_data'
echo
echo "== outline: jobs.rs =="
ast-grep outline crates/superposition_types/src/api/jobs.rs --view expanded || true
echo
echo "== show jobs.rs around the relevant types =="
cat -n crates/superposition_types/src/api/jobs.rs | sed -n '1,220p'
echo
echo "== search for JobDispatchRequest / JobRequest usage =="
rg -n "JobDispatchRequest|JobRequest|dispatch_job_handler|job_type|job_data" crates -g '!target'Repository: juspay/superposition
Length of output: 12373
🌐 Web query:
serde flatten adjacently tagged enum unsupported deserialize
💡 Result:
While Serde does not explicitly forbid the use of #[serde(flatten)] with adjacently tagged enums in its documentation [1][2], attempts to use them together often encounter significant deserialization issues [3][4][5]. The primary challenges arise because of how Serde handles both flattening and tagged enum deserialization: 1. Buffering and Field Consumption: When a struct field is marked with #[serde(flatten)], Serde often buffers the incoming data to parse it into the flattened target [3]. If the flattened target is an adjacently tagged enum, the deserialization process may fail to correctly consume or identify the fields, leading to errors such as "unknown field" or "invalid type" [3][4]. 2. Field-Naming Restrictions: Historically, adjacently tagged enums used internal visitors that expected field names to be &str, which caused failures when used with formats that provide non-string identifiers (like some self-describing or compact formats) or when combined with flattening logic [4][6]. While some of these specific bugs have been addressed in recent versions (e.g., support for bytes was improved) [4], the interaction between flattening and tagged enums remains a complex area [5]. 3. Structural Limitations: The combination of flattening and tagged enums is frequently cited in issues involving non-self-describing formats (like Bincode or MessagePack) [7][2]. Because these formats are not self-describing, they may struggle to roundtrip the metadata required for both the enum tag/content structure and the flattening operation, often resulting in runtime errors [2][8]. In practice, if you encounter errors when using an adjacently tagged enum with #[serde(flatten)], it is often because the deserializer cannot reconcile the expected enum structure (tag and content fields) with the requirements of the flattening logic [3][5]. Developers typically work around these issues by: - Avoiding #[serde(flatten)] if the structure allows for nested fields instead [9]. - Using a custom deserializer if the data structure is non-standard or requires specific handling of the fields [9]. - Ensuring that no conflicting field names are used (e.g., if a struct and its flattened enum both expect a field with the same name) [9]. For robust applications, especially those using non-JSON formats, it is recommended to test the roundtrip behavior thoroughly or consider alternative data modeling if the flattened approach fails to deserialize correctly [2].
Citations:
- 1: https://serde.rs/enum-representations.html
- 2: https://docs.rs/ron/latest/ron/
- 3: can serialize but can't deserilize when combined
tag = ""withflattenserde-rs/serde#2984 - 4: Error Deserializing Struct With
flatten-ed Tagged Enum BurntSushi/rust-csv#278 - 5: Flattening an Enum doesn't "consume" the fields serde-rs/serde#2200
- 6: Add
Deserializer::deserialize_identifiersupport for adjacently tagged enums serde-rs/serde#2475 - 7: Adjacently and internally tagged enums for non-self-describing formats serde-rs/serde#1762
- 8: Flatten serde-rs/serde#119
- 9: parsing enums inside structs with adjacent tags serde-rs/json#361
🌐 Web query:
site:docs.rs serde flatten adjacently tagged enum deserialize unsupported
💡 Result:
In Serde, using #[serde(flatten)] on a field that is an adjacently tagged enum is generally not supported and often results in errors or unexpected behavior because of how Serde buffers data during deserialization [1]. When Serde encounters #[serde(flatten)] or complex enum representations (such as adjacently tagged, internally tagged, or untagged enums), it often needs to buffer the incoming data to perform look-ahead or to determine which variant matches [1][2]. Non-self-describing formats—or even certain complex combinations in self-describing formats—struggle to resolve the state of the deserializer when these features are combined, as the required buffering often interferes with the structure of the data expected by the enum representation [1][2]. Key limitations include: - Buffering conflicts: Combining #[serde(flatten)] with tagged enums often causes deserialization failure because the data is not structured in a way that allows the flattened field to be extracted while simultaneously identifying the enum variant [1]. - Format dependence: While some specific crates or formats (e.g., serde_dynamo) may provide specialized support for these patterns, they are not standard features of the core Serde crate and are highly sensitive to the underlying data format [3][4]. - Alternative approaches: If you encounter this limitation, you may need to manually implement Deserialize for the struct or enum, or restructure your data model to avoid nesting flattened fields within tagged enums [1][5]. If you are using a non-self-describing format, these issues are often even more pronounced, as the deserializer cannot easily "peek" ahead or re-parse buffered content [1][5]. If your use case requires this pattern, consider whether the enum can be converted to an externally tagged enum, which is the standard, most compatible representation in Serde [6].
Citations:
- 1: https://docs.rs/ron/latest/ron/
- 2: https://docs.rs/serde-saphyr/latest/serde_saphyr/struct.Spanned.html
- 3: https://docs.rs/serde_dynamo/latest/serde_dynamo/index.html
- 4: https://docs.rs/serde_dynamo/latest/serde_dynamo/
- 5: https://docs.rs/serde_args/latest/serde_args/
- 6: https://docs.rs/serde_with/latest/serde_with/guide/serde_as_transformations/index.html
🌐 Web query:
serde.rs flatten not supported enum representations tag content
💡 Result:
In Serde, using the #[serde(flatten)] attribute in combination with tagged enum representations (internally tagged #[serde(tag = "...")] or adjacently tagged #[serde(tag = "...", content = "...")]) is not supported [1][2]. The Serde documentation explicitly notes that the flatten attribute is not supported in combination with these enum tags [1][2]. Furthermore, the general use of #[serde(flatten)] with enums has significant limitations [3][4]. While Serde allows flattening struct and map types, flattening enums is constrained; for example, attempting to flatten an enum may result in runtime errors such as "can only flatten structs and maps" depending on the specific enum variant and serialization context [3]. These limitations exist because the flattening process is designed primarily to inline keys from a field into the parent struct, a mechanism that does not easily reconcile with the specific key-value structures required by internal or adjacent tagging [5].
Citations:
- 1: https://serde.rs/container-attrs.html
- 2: https://serde.rs/container-attrs
- 3: Fix roundtrip of flatten externally tagged enum unit variant serde-rs/serde#2786
- 4: Flattening an Enum doesn't "consume" the fields serde-rs/serde#2200
- 5: https://serde.rs/attr-flatten.html
Remove flatten from JobDispatchRequest
crates/superposition_types/src/api/jobs.rs:51-57 — JobDispatchRequest flattens the adjacently tagged JobRequest, and serde doesn’t support that combination. dispatch_job_handler deserializes this type from the request body, so the callback path will fail at runtime. Use a nested payload or a custom Deserialize instead.
🤖 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 `@crates/superposition_types/src/api/jobs.rs` around lines 51 - 57, Update
JobDispatchRequest to remove #[serde(flatten)] from job_request and represent
JobRequest as a nested payload compatible with its adjacent tagging. Ensure
dispatch_job_handler can deserialize the request body successfully while
preserving the existing job_id field and JobRequest data.
| async fn dispatch_job_handler( | ||
| workspace_context: WorkspaceContext, | ||
| state: Data<AppState>, | ||
| db_conn: DbConnection, | ||
| user: User, | ||
| body: Json<JobDispatchRequest>, | ||
| ) -> superposition::Result<HttpResponse> { | ||
| let JobDispatchRequest { | ||
| job_id, | ||
| job_request, | ||
| } = body.into_inner(); | ||
| let DbConnection(mut conn) = db_conn; | ||
| let webhook = | ||
| fetch_webhook(&webhook_name, &workspace_context.schema_name, &mut conn)?; | ||
|
|
||
| update_job_status(&mut conn, job_id, BackgroundJobStatus::Inprogress) | ||
| .map_err(|e| unexpected_error!("Failed to update job status: {}", e))?; | ||
| let key = append_job_logs(&mut conn, job_id, "Job started", None) | ||
| .map_err(|e| unexpected_error!("Failed to append logs: {}", e))?; | ||
|
|
||
| let result = match &job_request { | ||
| JobRequest::Webhook(req) => { | ||
| execute_webhook_dispatch( | ||
| &workspace_context, | ||
| &state, | ||
| &mut conn, | ||
| &req.webhook_name, | ||
| &req.data, | ||
| ) | ||
| .await | ||
| } | ||
| JobRequest::PriorityRecompute(_) => { | ||
| execute_priority_recompute(&workspace_context, &state, &mut conn, &user).await | ||
| } | ||
| JobRequest::Reduce(req) => { | ||
| execute_reduce(&workspace_context, &state, &mut conn, &user, req.approve) | ||
| .await | ||
| } | ||
| }; |
There was a problem hiding this comment.
🔒 Security & Privacy | 🟡 Minor | ⚡ Quick win
🧩 Analysis chain
🏁 Script executed:
#!/bin/bash
set -euo pipefail
printf '\n== handlers.rs outline ==\n'
ast-grep outline crates/superposition/src/dispatch/handlers.rs --view expanded || true
printf '\n== jobs/handlers.rs outline ==\n'
ast-grep outline crates/superposition/src/jobs/handlers.rs --view expanded || true
printf '\n== kronos_dispatch.rs outline ==\n'
ast-grep outline crates/service_utils/src/kronos_dispatch.rs --view expanded || trueRepository: juspay/superposition
Length of output: 1909
🏁 Script executed:
#!/bin/bash
set -euo pipefail
printf '\n== relevant symbol search ==\n'
rg -n "dispatch_job_handler|update_job_status|append_job_logs|get_job_by_id|JobWorkspace|schema_name|authorized|workspace_context" \
crates/superposition/src/dispatch/handlers.rs \
crates/superposition/src/jobs/handlers.rs \
crates/service_utils/src/kronos_dispatch.rsRepository: juspay/superposition
Length of output: 6148
🏁 Script executed:
#!/bin/bash
set -euo pipefail
echo '--- handlers.rs (selected lines) ---'
sed -n '1,220p' crates/superposition/src/dispatch/handlers.rs | cat -n
echo '--- jobs/handlers.rs (selected lines) ---'
sed -n '1,260p' crates/superposition/src/jobs/handlers.rs | cat -n
echo '--- kronos_dispatch.rs (selected lines) ---'
sed -n '1,260p' crates/service_utils/src/kronos_dispatch.rs | cat -nRepository: juspay/superposition
Length of output: 25992
🏁 Script executed:
#!/bin/bash
set -euo pipefail
# Show route definitions and authorization around dispatch/job and jobs endpoints.
rg -n "dispatch/job|jobs/|authorized|kronos|workspace_context|JobDispatchRequest|JobWorkspace" \
crates/superposition/src \
-g '*.rs'Repository: juspay/superposition
Length of output: 11426
🏁 Script executed:
#!/bin/bash
set -euo pipefail
echo "ping"Repository: juspay/superposition
Length of output: 163
🌐 Web query:
"dispatch_job_handler" "WorkspaceContext" "JobDispatchRequest" "append_job_logs" "update_job_status"
💡 Result:
The terms provided appear to refer to specific methods and types within the context of job processing frameworks or systems, rather than a single unified API. Based on the terminology, these relate to the following domains: 1. Job Status and Management: The methods update_job_status and append_job_logs are common patterns for tracking the progress of background jobs. For instance, in Oracle APEX, APEX_PLSQL_JOB.UPDATE_JOB_STATUS is a documented procedure used to autonomously update the status of a running background PL/SQL job [1][2][3]. Similar custom implementations exist in various agent-based or task-processing systems, such as search-agent projects that use update functions to track job history [4]. 2. Job Dispatch and Context: - dispatch_job_handler and JobDispatchRequest: These terms are indicative of internal interfaces for routing work in distributed systems or agent frameworks. For example, systems like Pipecat use decorators (e.g., @job) to dispatch requests to handlers [5]. Other systems, such as Kestra, utilize internal dispatcher services (e.g., WorkerJobDispatcher) to manage job distribution to workers [6]. - WorkspaceContext: This is a frequently used pattern in multi-tenant or CLI-based applications to manage state, routing, or directory scopes. It appears in various contexts, including Rust-based middleware and TypeScript core utils for path validation [7][8]. 3. Dispatch (Platform): There is a specific platform known as Dispatch (dispatch.computer) which utilizes a WebSocket-based protocol for job lifecycle management (e.g., job_assign, job_complete) [9][10]. This system manages jobs through a coordinator-worker architecture, which is distinct from the PL/SQL procedural approach [10]. In summary, there is no single authoritative library that defines all these terms together; they are likely components of a specific codebase or architecture you are interacting with. If you are working within a particular framework (such as Dispatch, Kestra, or a custom internal system), these methods likely follow the standard design patterns of request handling, asynchronous state updates, and contextual routing described above [5][6][11][10].
Citations:
- 1: https://docs.oracle.com/database/121/AEAPI/apex_plsql_job.htm
- 2: https://docs.oracle.com/cd/E37097_01/doc.42/e35127/GUID-754B42EF-AD1F-4A1C-9FD0-C633BCDF0BEB.htm
- 3: https://docs.oracle.com/database/121/HTMDB/advnc_plsql.htm
- 4: https://github.com/sreekar2858/JobSearch-Agent/blob/main/main_api.py
- 5: https://daily.mintlify.app/pipecat/learn/job-coordination
- 6: https://github.com/kestra-io/kestra/blob/f313ee22/worker-controller/src/main/java/io/kestra/controller/grpc/services/WorkerJobDispatcher.java
- 7: https://docs.rs/mockforge-core/latest/mockforge_core/multi_tenant/middleware/struct.WorkspaceContext.html
- 8: https://github.com/QwenLM/qwen-code/blob/407a66c9/packages/core/src/utils/workspaceContext.ts
- 9: https://docs.dispatch.computer/docs/api/websocket
- 10: https://docs.dispatch.computer/docs/concepts/job-lifecycle
- 11: https://dispatchagents.ai/docs/building-agents.md
🏁 Script executed:
#!/bin/bash
set -euo pipefail
printf '\n== main route registration ==\n'
rg -n "dispatch::handlers|dispatch_job_handler|endpoints\(\)|Scope::new|service\(.*dispatch|workspace_context::OrgWorkspaceMiddlewareFactory|OrgWorkspaceMiddlewareFactory|authorized" \
crates/superposition/src/main.rs crates/superposition/src -g '*.rs'
printf '\n== app setup slices ==\n'
sed -n '120,260p' crates/superposition/src/main.rs | cat -n
printf '\n== dispatch endpoints slice ==\n'
sed -n '36,90p' crates/superposition/src/dispatch/handlers.rs | cat -nRepository: juspay/superposition
Length of output: 20251
🏁 Script executed:
#!/bin/bash
set -euo pipefail
printf '\n== authorized macro definition references ==\n'
rg -n "macro_rules! authorized|proc_macro.*authorized|authorized\]" crates -g '*.rs'
printf '\n== workspace middleware / auth context references ==\n'
rg -n "x-workspace|x-org-id|dispatch/job|DISPATCHER_USERNAME|DISPATCHER_SECRET_NAME|Basic" crates/superposition/src crates/service_utils/src -g '*.rs'Repository: juspay/superposition
Length of output: 20192
🏁 Script executed:
#!/bin/bash
set -euo pipefail
printf '\n== append_job_logs implementation ==\n'
sed -n '403,470p' crates/service_utils/src/kronos_dispatch.rs | cat -n
printf '\n== update_job_status implementation ==\n'
sed -n '403,440p' crates/service_utils/src/kronos_dispatch.rs | cat -nRepository: juspay/superposition
Length of output: 3759
🏁 Script executed:
#!/bin/bash
set -euo pipefail
python3 - <<'PY'
from pathlib import Path
p = Path('crates/service_utils/src/kronos_dispatch.rs')
text = p.read_text()
for name in ['get_job_by_id', 'update_job_status', 'append_job_logs']:
idx = text.find(f'pub fn {name}')
print(f'\n--- {name} ---')
print(text[idx: idx+1200])
PYRepository: juspay/superposition
Length of output: 3433
🏁 Script executed:
#!/bin/bash
set -euo pipefail
sed -n '140,240p' crates/superposition/src/main.rs | cat -nRepository: juspay/superposition
Length of output: 4758
🏁 Script executed:
#!/bin/bash
set -euo pipefail
printf '\n== append_job_logs and related helpers ==\n'
sed -n '360,470p' crates/service_utils/src/kronos_dispatch.rs | cat -n
printf '\n== job log schema / model references ==\n'
rg -n "job_logs|logs_key|append_job_logs|BackgroundJobStatus|WorkspaceJobView|job_manager" crates/superposition crates/service_utils crates/superposition_types -g '*.rs'Repository: juspay/superposition
Length of output: 12570
🏁 Script executed:
#!/bin/bash
set -euo pipefail
printf '\n== DbConnection / workspace schema wiring ==\n'
rg -n "type DbConnection|struct DbConnection|schema_name\(|set_config|search_path|WorkspaceContext|schema" \
crates/service_utils crates/superposition/src crates/superposition_types -g '*.rs'
printf '\n== service types slices ==\n'
sed -n '1,220p' crates/service_utils/src/service/types.rs | cat -n
printf '\n== WorkspaceContext definition slice ==\n'
rg -n "pub struct WorkspaceContext|schema_name" crates/service_utils/src/service -g '*.rs'Repository: juspay/superposition
Length of output: 44560
🏁 Script executed:
#!/bin/bash
set -euo pipefail
sed -n '360,465p' crates/service_utils/src/kronos_dispatch.rs | cat -nRepository: juspay/superposition
Length of output: 3929
Scope the job mutation and keep both log entries. dispatch_job_handler updates superposition.job_manager by job_id alone, so add a workspace-scoped lookup first. Also, append_job_logs uses the key as a JSON map key, so reusing Some(key) for the terminal log replaces "Job started" instead of appending it.
🤖 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 `@crates/superposition/src/dispatch/handlers.rs` around lines 66 - 102, Update
dispatch_job_handler to resolve and validate the job using both
workspace_context and job_id before calling update_job_status, ensuring the
mutation is workspace-scoped. Preserve the initial "Job started" log key, and
pass None or a distinct key for the terminal append_job_logs call so it appends
rather than replaces the existing entry.
| match result { | ||
| Ok(()) => { | ||
| update_job_status(&mut conn, job_id, BackgroundJobStatus::Completed) | ||
| .map_err(|e| unexpected_error!("Failed to update job status: {}", e))?; | ||
| update_job_progress(&mut conn, job_id, 100) | ||
| .map_err(|e| unexpected_error!("Failed to update job progress: {}", e))?; | ||
| append_job_logs(&mut conn, job_id, "Job completed", Some(key)) | ||
| .map_err(|e| unexpected_error!("Failed to append logs: {}", e))?; | ||
| Ok(HttpResponse::Ok().finish()) | ||
| } | ||
| Err(e) => { | ||
| let error_msg = format!("{e}"); | ||
| log::error!("Job {job_id} failed: {error_msg}"); | ||
| update_job_status(&mut conn, job_id, BackgroundJobStatus::Failed) | ||
| .map_err(|e| unexpected_error!("Failed to update job status: {}", e))?; | ||
| append_job_logs(&mut conn, job_id, &format!("Job failed: {error_msg}"), Some(key)) | ||
| .map_err(|e| unexpected_error!("Failed to append logs: {}", e))?; | ||
| Err(unexpected_error!("Job {job_id} failed: {error_msg}")) | ||
| } | ||
| } |
There was a problem hiding this comment.
🗄️ Data Integrity & Integration | 🟠 Major | ⚡ Quick win
Reused log key overwrites the "Job started" entry instead of appending a new one.
append_job_logs inserts the log line under the exact timed_key provided (see its implementation in the sibling review of crates/superposition/src/jobs/handlers.rs), so passing Some(key) — the same key returned for "Job started" — into the completion/failure call replaces that entry rather than adding a new timestamped line. The job's logs JSONB ends up missing the start event. Contrast with cancel_handler, where the same key is reused across two mutually exclusive branches (safe); here both calls execute sequentially on the same job.
🐛 Suggested fix
- append_job_logs(&mut conn, job_id, "Job completed", Some(key))
+ append_job_logs(&mut conn, job_id, "Job completed", None)
.map_err(|e| unexpected_error!("Failed to append logs: {}", e))?;- append_job_logs(&mut conn, job_id, &format!("Job failed: {error_msg}"), Some(key))
+ append_job_logs(&mut conn, job_id, &format!("Job failed: {error_msg}"), None)
.map_err(|e| unexpected_error!("Failed to append logs: {}", e))?;📝 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.
| match result { | |
| Ok(()) => { | |
| update_job_status(&mut conn, job_id, BackgroundJobStatus::Completed) | |
| .map_err(|e| unexpected_error!("Failed to update job status: {}", e))?; | |
| update_job_progress(&mut conn, job_id, 100) | |
| .map_err(|e| unexpected_error!("Failed to update job progress: {}", e))?; | |
| append_job_logs(&mut conn, job_id, "Job completed", Some(key)) | |
| .map_err(|e| unexpected_error!("Failed to append logs: {}", e))?; | |
| Ok(HttpResponse::Ok().finish()) | |
| } | |
| Err(e) => { | |
| let error_msg = format!("{e}"); | |
| log::error!("Job {job_id} failed: {error_msg}"); | |
| update_job_status(&mut conn, job_id, BackgroundJobStatus::Failed) | |
| .map_err(|e| unexpected_error!("Failed to update job status: {}", e))?; | |
| append_job_logs(&mut conn, job_id, &format!("Job failed: {error_msg}"), Some(key)) | |
| .map_err(|e| unexpected_error!("Failed to append logs: {}", e))?; | |
| Err(unexpected_error!("Job {job_id} failed: {error_msg}")) | |
| } | |
| } | |
| match result { | |
| Ok(()) => { | |
| update_job_status(&mut conn, job_id, BackgroundJobStatus::Completed) | |
| .map_err(|e| unexpected_error!("Failed to update job status: {}", e))?; | |
| update_job_progress(&mut conn, job_id, 100) | |
| .map_err(|e| unexpected_error!("Failed to update job progress: {}", e))?; | |
| append_job_logs(&mut conn, job_id, "Job completed", None) | |
| .map_err(|e| unexpected_error!("Failed to append logs: {}", e))?; | |
| Ok(HttpResponse::Ok().finish()) | |
| } | |
| Err(e) => { | |
| let error_msg = format!("{e}"); | |
| log::error!("Job {job_id} failed: {error_msg}"); | |
| update_job_status(&mut conn, job_id, BackgroundJobStatus::Failed) | |
| .map_err(|e| unexpected_error!("Failed to update job status: {}", e))?; | |
| append_job_logs(&mut conn, job_id, &format!("Job failed: {error_msg}"), None) | |
| .map_err(|e| unexpected_error!("Failed to append logs: {}", e))?; | |
| Err(unexpected_error!("Job {job_id} failed: {error_msg}")) | |
| } | |
| } |
🤖 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 `@crates/superposition/src/dispatch/handlers.rs` around lines 104 - 123, Update
the completion and failure branches in the result handling around
append_job_logs so they append new timestamped entries instead of reusing the
original Job started key. Pass no reused timed_key to both calls, while
preserving the existing log messages and status updates.
Signed-off-by: datron <Datron@users.noreply.github.com>
1 participant
Problem
#1093
Solution
Introduce Background Job Manager
Summary by CodeRabbit