Adding triggers to graphs #428
Conversation
…templates - Added `croniter` as a dependency in `pyproject.toml` to support cron expressions. - Updated `upsert_graph_template` to handle triggers in graph templates. - Introduced `Trigger` and `CronTrigger` models to validate and manage triggers. - Enhanced `GraphTemplate` and request/response models to include triggers. This update improves the flexibility of graph templates by allowing scheduled executions based on cron expressions.
SafeDep Report Summary
Package Details
This report is generated by SafeDep Github App. |
|
Warning Rate limit exceeded@NiveditJain has exceeded the limit for the number of commits or files that can be reviewed per hour. Please wait 2 minutes and 19 seconds before requesting another review. ⌛ How to resolve this issue?After the wait time has elapsed, a review can be triggered using the We recommend that you space out your commits to avoid hitting the rate limit. 🚦 How do rate limits work?CodeRabbit enforces hourly rate limits for each developer per organization. Our paid plans have higher rate limits than the trial, open-source and free plans. In all cases, we re-allow further reviews after a brief timeout. Please see our FAQ for further information. 📒 Files selected for processing (1)
Note Other AI code review bot(s) detectedCodeRabbit has detected other AI code review bot(s) in this pull request and will avoid duplicating their findings in the review comments. This may lead to a less comprehensive review. 📝 WalkthroughSummary by CodeRabbit
WalkthroughAdds CRON trigger support: request/response and DB GraphTemplate include Changes
Sequence Diagram(s)sequenceDiagram
autonumber
actor Client
participant Controller as UpsertGraphTemplateController
participant ReqModel as Request (validates triggers)
participant DB as DB (GraphTemplate & DatabaseTriggers)
participant Verifier as Background verify_graph
note right of Verifier #e6f7ff: reconcile CRON triggers (cancel/create)
Client->>Controller: POST/PUT /graph-template (body includes triggers)
Controller->>ReqModel: validate body (cron expressions validated)
alt validation succeeds
Controller->>DB: insert/update GraphTemplate(..., triggers=body.triggers)
DB-->>Controller: saved GraphTemplate (provides old_triggers on update)
Controller->>Verifier: schedule verify_graph(graph_template, old_triggers)
Verifier->>DB: cancel_crons(old_triggers, graph_template.triggers)
Verifier->>DB: create_crons(graph_template.triggers, ahead_window)
DB-->>Verifier: persist DatabaseTriggers entries
Controller-->>Client: 200/201 with response (includes triggers)
else validation fails
Controller-->>Client: 4xx error
end
Estimated code review effort🎯 4 (Complex) | ⏱️ ~45 minutes Possibly related PRs
Suggested reviewers
Poem
Pre-merge checks and finishing touches❌ Failed checks (1 warning)
✅ Passed checks (2 passed)
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 |
Summary of ChangesHello @NiveditJain, I'm Gemini Code Assist1! I'm currently reviewing this pull request and will post my feedback shortly. In the meantime, here's a summary to help you and other reviewers quickly get up to speed! This pull request significantly enhances the system by introducing scheduled execution capabilities for graph templates. It integrates cron-based triggering mechanisms, allowing users to define specific schedules for their graph templates. This functionality is supported by new data models for triggers, robust validation logic for cron expressions, and updates to existing graph template models and API endpoints to accommodate these new scheduling capabilities, ultimately improving automation and flexibility. Highlights
Using Gemini Code AssistThe full guide for Gemini Code Assist can be found on our documentation page, here are some quick tips. Invoking Gemini You can request assistance from Gemini at any point by creating a comment using either
Customization To customize Gemini Code Assist for GitHub experience, repository maintainers can create a configuration file and/or provide a custom code review style guide (such as PEP-8 for Python) by creating and adding files to a Limitations & Feedback Gemini Code Assist may make mistakes. Please leave feedback on any instances where its feedback is incorrect or counter productive. You can react with 👍 and 👎 on @gemini-code-assist comments. If you're interested in giving your feedback about your experience with Gemini Code Assist for Github and other Google products, sign up here. You can also get AI-powered code generation, chat, as well as code reviews directly in the IDE at no cost with the Gemini Code Assist IDE Extension. Footnotes
|
NiveditJain
changed the title
Codecov Report❌ Patch coverage is 📢 Thoughts on this report? Let us know! |
![gemini-code-assist[bot]](https://avatars.githubusercontent.com/in/956858?s=60&v=4){kind=small}
There was a problem hiding this comment.
Code Review
This pull request introduces trigger functionality for graph templates, allowing for scheduled executions via cron expressions. A new dependency croniter is added for this purpose. The changes include new Pydantic models for triggers and updates to the graph template models and the upsert controller to handle triggers.
The implementation of the trigger models is functional, but I've provided a couple of suggestions in app/models/trigger_models.py to improve robustness and extensibility, particularly around type hinting and validation logic. These changes will make the system easier to maintain as new trigger types are added in the future.
Overall, the changes are well-structured and address the feature requirement effectively.
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 3
Caution
Some comments are outside the diff and can’t be posted inline due to platform limitations.
⚠️ Outside diff range comments (2)
state-manager/app/controller/upsert_graph_template.py (2)
25-33: Reset internal caches after updating nodes to avoid stale graph metadata.Assigning graph_template.nodes replaces the graph, but cached PrivateAttr structures (_node_by_identifier, _parents_by_identifier, _root_node, _path_by_identifier) are not invalidated. Subsequent validation or background verification can use stale caches and yield incorrect results.
Suggested fix here (minimal, local invalidation):
graph_template.retry_policy = body.retry_policy graph_template.store_config = body.store_config - graph_template.nodes = body.nodes + graph_template.nodes = body.nodes + # Invalidate internal caches so validators and lookups rebuild correctly + graph_template._node_by_identifier = None + graph_template._parents_by_identifier = None + graph_template._root_node = None + graph_template._path_by_identifier = None graph_template.triggers = body.triggers await graph_template.save()Alternatively, add a public method on GraphTemplate to reset caches and call it here (preferred API design). I can draft that method if you want.
59-67: Avoid decrypting secrets just to expose presence flags.get_secrets() decrypts values only to fetch keys. Use the stored encrypted keys instead to reduce work and avoid unnecessary crypto operations.
Apply:
- secrets={secret_name: True for secret_name in graph_template.get_secrets().keys()}, + secrets={secret_name: True for secret_name in graph_template.secrets.keys()},
📜 Review details
Configuration used: CodeRabbit UI
Review profile: ASSERTIVE
Plan: Pro
⛔ Files ignored due to path filters (1)
state-manager/uv.lockis excluded by!**/*.lock
📒 Files selected for processing (5)
state-manager/app/controller/upsert_graph_template.py(3 hunks)state-manager/app/models/db/graph_template_model.py(2 hunks)state-manager/app/models/graph_models.py(1 hunks)state-manager/app/models/trigger_models.py(1 hunks)state-manager/pyproject.toml(1 hunks)
🧰 Additional context used
🧬 Code graph analysis (3)
state-manager/app/models/db/graph_template_model.py (1)
state-manager/app/models/trigger_models.py (1)
Trigger(19-27)
state-manager/app/controller/upsert_graph_template.py (1)
state-manager/app/models/db/graph_template_model.py (1)
set_secrets(269-271)
state-manager/app/models/graph_models.py (1)
state-manager/app/models/trigger_models.py (1)
Trigger(19-27)
🔇 Additional comments (4)
state-manager/app/models/db/graph_template_model.py (1)
16-16: Triggers field addition looks correct and consistent.Import and Field(default_factory=list) are appropriate; Pydantic will validate Trigger instances. No DB index changes needed.
Also applies to: 25-25
state-manager/app/models/graph_models.py (2)
16-16: Request model addition for triggers looks good.Defaulting to an empty list with proper typing is correct and aligns with controller/DB changes.
24-24: Response model addition for triggers looks good.Matches the persisted field and controller response.
state-manager/pyproject.toml (1)
13-13: croniter>=6.0.0 is valid
croniter 6.0.0 is available on PyPI; no change needed.
There was a problem hiding this comment.
Actionable comments posted: 4
📜 Review details
Configuration used: CodeRabbit UI
Review profile: ASSERTIVE
Plan: Pro
📒 Files selected for processing (6)
state-manager/app/config/settings.py(2 hunks)state-manager/app/controller/upsert_graph_template.py(4 hunks)state-manager/app/models/db/trigger.py(1 hunks)state-manager/app/models/trigger_models.py(1 hunks)state-manager/app/routes.py(1 hunks)state-manager/app/tasks/verify_graph.py(3 hunks)
🧰 Additional context used
🧬 Code graph analysis (4)
state-manager/app/models/db/trigger.py (1)
state-manager/app/models/trigger_models.py (2)
TriggerTypeEnum(6-7)TriggerStatusEnum(9-13)
state-manager/app/controller/upsert_graph_template.py (1)
state-manager/app/tasks/verify_graph.py (1)
verify_graph(152-184)
state-manager/app/tasks/verify_graph.py (5)
state-manager/app/models/db/graph_template_model.py (1)
GraphTemplate(18-347)state-manager/app/models/db/registered_node.py (1)
RegisteredNode(8-44)state-manager/app/models/trigger_models.py (4)
Trigger(25-33)CronTrigger(15-23)TriggerStatusEnum(9-13)TriggerTypeEnum(6-7)state-manager/app/models/db/trigger.py (1)
Triggers(9-24)state-manager/app/config/settings.py (1)
get_settings(32-36)
state-manager/app/routes.py (1)
state-manager/app/models/trigger_graph_model.py (2)
TriggerGraphRequestModel(4-7)TriggerGraphResponseModel(9-11)
🪛 GitHub Actions: State Manager Unit Tests
state-manager/app/config/settings.py
[error] 1-1: ModuleNotFoundError: No module named 'config'
state-manager/app/tasks/verify_graph.py
[error] 1-1: ModuleNotFoundError: No module named 'config' while importing verify_graph (from config.settings import get_settings)
🔇 Additional comments (1)
state-manager/app/models/trigger_models.py (1)
26-27: Use default_factory for mutable trigger payloadEchoing the earlier feedback:
Field(default={})reuses the same dict across Trigger instances under pydantic v2, so one trigger mutatingvaluewill leak into others. Switch todefault_factory=dictso each trigger gets its own payload container.- value: dict[str, str] = Field(default={}, description="Value of the trigger") + value: dict[str, str] = Field(default_factory=dict, description="Value of the trigger")
- Updated the import path for `get_settings` to align with the new structure. - Modified the `create_crons` function to ensure at least one event is inserted for each cron trigger, improving the handling of trigger times and preventing unnecessary database inserts.
- Changed import path for TriggerGraphRequestModel and TriggerGraphResponseModel to reflect new model structure. - Introduced a new file for trigger_cron task to manage scheduled trigger executions.
There was a problem hiding this comment.
Actionable comments posted: 5
📜 Review details
Configuration used: CodeRabbit UI
Review profile: ASSERTIVE
Plan: Pro
📒 Files selected for processing (1)
state-manager/app/tasks/verify_graph.py(3 hunks)
🧰 Additional context used
🧬 Code graph analysis (1)
state-manager/app/tasks/verify_graph.py (4)
state-manager/app/models/db/registered_node.py (1)
RegisteredNode(8-44)state-manager/app/models/trigger_models.py (4)
Trigger(25-33)CronTrigger(15-23)TriggerStatusEnum(9-13)TriggerTypeEnum(6-7)state-manager/app/models/db/trigger.py (1)
Triggers(9-24)state-manager/app/config/settings.py (1)
get_settings(32-36)
🔇 Additional comments (5)
state-manager/app/tasks/verify_graph.py (5)
12-14: Import root fixed — good catch.Switching to
from app.config.settings import get_settingsresolves the prior module resolution error.
154-156: Empty bulk insert guard — good.Avoids
insert_many([])failures.
105-123: Sets of Pydantic models are unhashable; diff on validated strings + skip no‑ops.Current
set([CronTrigger(...)])raisesTypeError. Build sets on the validatedexpressionstring and return early when nothing to cancel.async def cancel_crons(graph_template: GraphTemplate, old_triggers: list[Trigger]): - old_crons = set([CronTrigger(**trigger.value) for trigger in old_triggers if trigger.type == TriggerTypeEnum.CRON]) - new_crons = set([CronTrigger(**trigger.value) for trigger in graph_template.triggers if trigger.type == TriggerTypeEnum.CRON]) + old_crons = { + CronTrigger(**trigger.value).expression + for trigger in (old_triggers or []) + if trigger.type == TriggerTypeEnum.CRON + } + new_crons = { + CronTrigger(**trigger.value).expression + for trigger in (graph_template.triggers or []) + if trigger.type == TriggerTypeEnum.CRON + } - removed = old_crons - new_crons + removed = old_crons - new_crons + if not removed: + return await DatabaseTriggers.find( DatabaseTriggers.graph_name == graph_template.name, DatabaseTriggers.trigger_status == TriggerStatusEnum.PENDING, DatabaseTriggers.type == TriggerTypeEnum.CRON, - In(DatabaseTriggers.expression, [cron.expression for cron in removed]) + In(DatabaseTriggers.expression, list(removed)) ).update( { "$set": { "trigger_status": TriggerStatusEnum.CANCELLED } } ) # type: ignore
124-129: Same unhashable-set bug in create path.Mirror the fix: diff on expression strings and guard
graph_template.triggersbeingNone.async def create_crons(graph_template: GraphTemplate, old_triggers: list[Trigger]): - old_crons = set([CronTrigger(**trigger.value) for trigger in old_triggers if trigger.type == TriggerTypeEnum.CRON]) - new_crons = set([CronTrigger(**trigger.value) for trigger in graph_template.triggers if trigger.type == TriggerTypeEnum.CRON]) + old_crons = { + CronTrigger(**trigger.value).expression + for trigger in (old_triggers or []) + if trigger.type == TriggerTypeEnum.CRON + } + new_crons = { + CronTrigger(**trigger.value).expression + for trigger in (graph_template.triggers or []) + if trigger.type == TriggerTypeEnum.CRON + } crons_to_create = new_crons - old_crons
105-156: No None guard needed forgraph_template.triggers
GraphTemplate.triggersis defined withField(default_factory=list)and will never beNone, so wrapping it in(graph_template.triggers or [])is unnecessary.Likely an incorrect or invalid review comment.
- Introduced the DatabaseTriggers model to manage trigger data in the database. - Updated the DOCUMENT_MODELS list in main.py to include DatabaseTriggers. - Adjusted import statements in verify_graph.py to reflect the new model name.
Co-authored-by: gemini-code-assist[bot] <176961590+gemini-code-assist[bot]@users.noreply.github.com>
Co-authored-by: gemini-code-assist[bot] <176961590+gemini-code-assist[bot]@users.noreply.github.com>
…tion logic - Added a new 'namespace' field to the DatabaseTriggers model to specify the graph's namespace. - Updated the 'create_crons' function in verify_graph.py to include the namespace when creating cron triggers, ensuring better organization and management of triggers.
There was a problem hiding this comment.
Actionable comments posted: 3
📜 Review details
Configuration used: CodeRabbit UI
Review profile: ASSERTIVE
Plan: Pro
📒 Files selected for processing (4)
state-manager/app/main.py(2 hunks)state-manager/app/models/db/trigger.py(1 hunks)state-manager/app/models/trigger_models.py(1 hunks)state-manager/app/tasks/verify_graph.py(3 hunks)
🧰 Additional context used
🧠 Learnings (2)
📓 Common learnings
Learnt from: NiveditJain
PR: exospherehost/exospherehost#428
File: state-manager/app/tasks/verify_graph.py:4-5
Timestamp: 2025-09-28T13:35:42.851Z
Learning: In the cron trigger scheduling logic for state-manager/app/tasks/verify_graph.py, the system intentionally schedules at least one trigger beyond the trigger_ahead_time window to ensure continuity of scheduled executions. The current logic of appending an event then breaking is by design to guarantee "at least one next" trigger.
📚 Learning: 2025-09-28T13:35:42.851Z
Learnt from: NiveditJain
PR: exospherehost/exospherehost#428
File: state-manager/app/tasks/verify_graph.py:4-5
Timestamp: 2025-09-28T13:35:42.851Z
Learning: In the cron trigger scheduling logic for state-manager/app/tasks/verify_graph.py, the system intentionally schedules at least one trigger beyond the trigger_ahead_time window to ensure continuity of scheduled executions. The current logic of appending an event then breaking is by design to guarantee "at least one next" trigger.
Applied to files:
state-manager/app/tasks/verify_graph.py
🧬 Code graph analysis (3)
state-manager/app/main.py (1)
state-manager/app/models/db/trigger.py (1)
DatabaseTriggers(9-36)
state-manager/app/models/db/trigger.py (1)
state-manager/app/models/trigger_models.py (2)
TriggerTypeEnum(6-7)TriggerStatusEnum(9-13)
state-manager/app/tasks/verify_graph.py (5)
state-manager/app/models/db/graph_template_model.py (1)
GraphTemplate(18-347)state-manager/app/models/db/registered_node.py (1)
RegisteredNode(8-44)state-manager/app/models/trigger_models.py (4)
Trigger(25-35)CronTrigger(15-23)TriggerStatusEnum(9-13)TriggerTypeEnum(6-7)state-manager/app/models/db/trigger.py (1)
DatabaseTriggers(9-36)state-manager/app/config/settings.py (1)
get_settings(32-36)
🔇 Additional comments (7)
state-manager/app/main.py (2)
25-25: Wiring DatabaseTriggers into startup looks correct.Import is consistent with other app-local models.
38-38: Beanie model registration added — good.This ensures indexes are created and health checks include triggers. Verify your PyMongo/async-pymongo version supports AsyncMongoClient initialization used above.
state-manager/app/tasks/verify_graph.py (3)
2-2: croniter dependency risk (tracking).croniter is unmaintained; keep the pin tight and plan migration (issue already opened). No action blocking this PR.
105-116: Unhashable CronTrigger in sets; missing namespace in cancellation filter.
- Pydantic models are unhashable → TypeError at runtime.
- Cancellation query lacks namespace and can cancel other tenants’ triggers.
Apply:
async def cancel_crons(graph_template: GraphTemplate, old_triggers: list[Trigger]): - old_crons = set([CronTrigger(**trigger.value) for trigger in old_triggers if trigger.type == TriggerTypeEnum.CRON]) - new_crons = set([CronTrigger(**trigger.value) for trigger in graph_template.triggers if trigger.type == TriggerTypeEnum.CRON]) + old_crons = { + CronTrigger(**t.value).expression + for t in old_triggers + if t.type == TriggerTypeEnum.CRON + } + new_crons = { + CronTrigger(**t.value).expression + for t in graph_template.triggers + if t.type == TriggerTypeEnum.CRON + } - removed = old_crons - new_crons + removed = old_crons - new_crons + if not removed: + return await DatabaseTriggers.find( DatabaseTriggers.graph_name == graph_template.name, + DatabaseTriggers.namespace == graph_template.namespace, DatabaseTriggers.trigger_status == TriggerStatusEnum.PENDING, DatabaseTriggers.type == TriggerTypeEnum.CRON, - In(DatabaseTriggers.expression, [cron.expression for cron in removed]) + In(DatabaseTriggers.expression, list(removed)) ).update( { "$set": { "trigger_status": TriggerStatusEnum.CANCELLED } } ) # type: ignore
179-191: Don’t flip a VALID template to INVALID due to scheduling errors. Save first; gather with return_exceptions.Keeps validation result independent of CRON persistence issues; log errors instead.
- graph_template.validation_status = GraphTemplateValidationStatus.VALID - graph_template.validation_errors = [] - - await asyncio.gather(*[cancel_crons(graph_template, old_triggers), create_crons(graph_template, old_triggers)]) - - await graph_template.save() + graph_template.validation_status = GraphTemplateValidationStatus.VALID + graph_template.validation_errors = [] + await graph_template.save() + + results = await asyncio.gather( + cancel_crons(graph_template, old_triggers), + create_crons(graph_template, old_triggers), + return_exceptions=True, + ) + for r in results: + if isinstance(r, Exception): + logger.error( + f"CRON scheduling error for graph template {graph_template.id}: {r}", + exc_info=True, + )state-manager/app/models/db/trigger.py (1)
9-36: Model and indexes look solid for idempotent scheduling.Unique compound index includes namespace to avoid cross-tenant collisions; time index supports scanning upcoming triggers.
state-manager/app/models/trigger_models.py (1)
18-23: Trim and clarify cron expression validation.Improves error messages and rejects whitespace-only inputs.
@field_validator("expression") @classmethod def validate_expression(cls, v: str) -> str: - if not croniter.is_valid(v): - raise ValueError("Invalid cron expression") - return v + v = v.strip() + if not v: + raise ValueError("Cron expression must not be empty") + if not croniter.is_valid(v): + raise ValueError(f"Invalid cron expression: {v!r}") + return v
- Introduced APScheduler as a dependency in `pyproject.toml` to facilitate scheduled tasks. - Implemented a cron job in `main.py` that triggers the `trigger_cron` function every minute. - Created a new `trigger_cron.py` file to define the task executed by the scheduler. This enhancement allows for periodic execution of tasks, improving the application's scheduling capabilities.
…ality - Updated import statements in multiple test files to replace `TriggerGraphRequestModel` with the new `trigger_graph_model`. - Enhanced the `test_lifespan_init_beanie_with_correct_models` to include `DatabaseTriggers` in the expected document models. - Improved mock setups in `test_verify_graph.py` to handle asynchronous operations and ensure proper validation checks. - Adjusted background task assertions in `test_upsert_graph_template.py` for clarity and correctness.
There was a problem hiding this comment.
Actionable comments posted: 3
Caution
Some comments are outside the diff and can’t be posted inline due to platform limitations.
⚠️ Outside diff range comments (1)
state-manager/app/main.py (1)
8-8: Wrong Mongo client and incorrect close semantics (will crash at runtime).
- PyMongo does not expose AsyncMongoClient here; Beanie expects Motor’s AsyncIOMotorClient.
- Motor’s close() is sync; remove await.
-from pymongo import AsyncMongoClient +from motor.motor_asyncio import AsyncIOMotorClient @@ - await client.close() + client.close()Also applies to: 82-82
📜 Review details
Configuration used: CodeRabbit UI
Review profile: ASSERTIVE
Plan: Pro
⛔ Files ignored due to path filters (1)
state-manager/uv.lockis excluded by!**/*.lock
📒 Files selected for processing (3)
state-manager/app/main.py(3 hunks)state-manager/app/tasks/trigger_cron.py(1 hunks)state-manager/pyproject.toml(1 hunks)
🧰 Additional context used
🧠 Learnings (1)
📓 Common learnings
Learnt from: NiveditJain
PR: exospherehost/exospherehost#428
File: state-manager/app/tasks/verify_graph.py:4-5
Timestamp: 2025-09-28T13:35:42.851Z
Learning: In the cron trigger scheduling logic for state-manager/app/tasks/verify_graph.py, the system intentionally schedules at least one trigger beyond the trigger_ahead_time window to ensure continuity of scheduled executions. The current logic of appending an event then breaking is by design to guarantee "at least one next" trigger.
🧬 Code graph analysis (1)
state-manager/app/main.py (3)
state-manager/app/models/db/trigger.py (1)
DatabaseTriggers(9-36)state-manager/app/models/trigger_models.py (1)
CronTrigger(15-23)state-manager/app/tasks/trigger_cron.py (1)
trigger_cron(3-4)
🔇 Additional comments (1)
state-manager/app/main.py (1)
70-75: Confirm that this every-minute job aligns with the trigger reconciliation flow.If verify_graph now owns CRON reconciliation (create/cancel within ahead window), ensure this heartbeat job either invokes that path or is removed to avoid duplicate scheduling logic. Based on learnings.
- Added functions to retrieve and update due triggers in the database, changing their status from PENDING to TRIGGERED. - Implemented the `call_trigger_graph` function to execute the trigger graph with a unique request ID. - Enhanced the `trigger_cron` function to log the start time and process due triggers in a loop, calling the necessary functions for each trigger. - Introduced a placeholder for `create_next_triggers` to facilitate future trigger creation logic.
There was a problem hiding this comment.
Actionable comments posted: 4
Caution
Some comments are outside the diff and can’t be posted inline due to platform limitations.
⚠️ Outside diff range comments (1)
state-manager/tests/unit/tasks/test_verify_graph.py (1)
542-567: Verify cron helpers run in the “valid graph” path.Same Ruff F841 issue here: the newly patched cron mocks aren’t touched. Assert they were awaited with the expected arguments to confirm the cron scheduling branch executes and to unblock the lint job.
graph_template.triggers = [] graph_template.name = "test_graph" # This should mark the graph as valid await verify_graph(graph_template, []) + + mock_cancel_crons.assert_awaited_once_with(graph_template, []) + mock_create_crons.assert_awaited_once_with(graph_template, []) # Verify that the graph was processed (status may vary based on actual validation) # The specific status depends on the actual validation logic assert graph_template.save.called
📜 Review details
Configuration used: CodeRabbit UI
Review profile: ASSERTIVE
Plan: Pro
📒 Files selected for processing (7)
state-manager/app/tasks/trigger_cron.py(1 hunks)state-manager/tests/unit/controller/test_trigger_graph.py(1 hunks)state-manager/tests/unit/controller/test_upsert_graph_template.py(2 hunks)state-manager/tests/unit/tasks/test_verify_graph.py(7 hunks)state-manager/tests/unit/test_main.py(2 hunks)state-manager/tests/unit/test_routes.py(1 hunks)state-manager/tests/unit/with_database/conftest.py(2 hunks)
🧰 Additional context used
🧠 Learnings (2)
📓 Common learnings
Learnt from: NiveditJain
PR: exospherehost/exospherehost#428
File: state-manager/app/tasks/verify_graph.py:4-5
Timestamp: 2025-09-28T13:35:42.851Z
Learning: In the cron trigger scheduling logic for state-manager/app/tasks/verify_graph.py, the system intentionally schedules at least one trigger beyond the trigger_ahead_time window to ensure continuity of scheduled executions. The current logic of appending an event then breaking is by design to guarantee "at least one next" trigger.
📚 Learning: 2025-09-28T13:35:42.851Z
Learnt from: NiveditJain
PR: exospherehost/exospherehost#428
File: state-manager/app/tasks/verify_graph.py:4-5
Timestamp: 2025-09-28T13:35:42.851Z
Learning: In the cron trigger scheduling logic for state-manager/app/tasks/verify_graph.py, the system intentionally schedules at least one trigger beyond the trigger_ahead_time window to ensure continuity of scheduled executions. The current logic of appending an event then breaking is by design to guarantee "at least one next" trigger.
Applied to files:
state-manager/app/tasks/trigger_cron.py
🧬 Code graph analysis (6)
state-manager/tests/unit/test_routes.py (1)
state-manager/app/models/trigger_graph_model.py (1)
TriggerGraphRequestModel(4-7)
state-manager/tests/unit/controller/test_upsert_graph_template.py (2)
state-manager/tests/unit/test_routes.py (1)
mock_background_tasks(361-363)state-manager/tests/unit/controller/test_executed_state.py (1)
mock_background_tasks(27-28)
state-manager/app/tasks/trigger_cron.py (5)
state-manager/app/models/db/trigger.py (1)
DatabaseTriggers(9-36)state-manager/app/models/trigger_models.py (1)
TriggerStatusEnum(9-13)state-manager/app/singletons/logs_manager.py (2)
LogsManager(9-66)get_logger(65-66)state-manager/app/controller/trigger_graph.py (1)
trigger_graph(31-115)state-manager/app/models/trigger_graph_model.py (1)
TriggerGraphRequestModel(4-7)
state-manager/tests/unit/tasks/test_verify_graph.py (3)
state-manager/app/tasks/verify_graph.py (1)
verify_graph(158-191)state-manager/app/models/db/registered_node.py (1)
list_nodes_by_templates(34-44)state-manager/app/models/graph_template_validation_status.py (1)
GraphTemplateValidationStatus(4-8)
state-manager/tests/unit/controller/test_trigger_graph.py (1)
state-manager/app/models/trigger_graph_model.py (1)
TriggerGraphRequestModel(4-7)
state-manager/tests/unit/test_main.py (1)
state-manager/app/models/db/trigger.py (1)
DatabaseTriggers(9-36)
🪛 GitHub Actions: Ruff check on changed files only
state-manager/tests/unit/tasks/test_verify_graph.py
[error] 388-388: F841 Local variable 'mock_cancel_crons' is assigned to but never used (remove assignment).
[error] 389-389: F841 Local variable 'mock_create_crons' is assigned to but never used (remove assignment).
[error] 502-502: F841 Local variable 'mock_cancel_crons' is assigned to but never used (remove assignment).
[error] 503-503: F841 Local variable 'mock_create_crons' is assigned to but never used (remove assignment).
[error] 542-542: F841 Local variable 'mock_cancel_crons' is assigned to but never used (remove assignment).
[error] 543-543: F841 Local variable 'mock_create_crons' is assigned to but never used (remove assignment).
- Updated the Settings model to replace `trigger_ahead_time` with `trigger_workers`, allowing for configurable worker count for trigger processing. - Added a new status `TRIGGERING` to the TriggerStatusEnum to better represent the trigger lifecycle. - Enhanced the `trigger_cron` function to utilize the new worker configuration, enabling concurrent processing of due triggers. - Refined the `create_next_triggers` function to calculate the next trigger time based on the cron expression, improving trigger scheduling logic.
|
/gemini review |
There was a problem hiding this comment.
Code Review
This pull request introduces cron-based triggers for graphs, a significant feature enhancement. The implementation correctly uses apscheduler for scheduling and croniter for cron expression handling. The database models and API endpoints are updated accordingly. My review focuses on improving the robustness of the new trigger handling logic, particularly around error handling and race conditions, to prevent triggers from getting into a stuck state. I've also included a suggestion to strengthen a unit test.
- Removed unused imports from verify_graph.py to streamline the code. - Updated mock setups in test_verify_graph.py to improve clarity by replacing mock variable names with underscores, enhancing readability without changing functionality.
|
@coderabbitai review this |
|
@NiveditJain I'll review the changes in this PR which introduces cron-based trigger functionality for graph templates. This is a significant enhancement that adds scheduled execution capabilities to your graph system. The implementation includes:
The changes look comprehensive and well-structured. I'll provide a detailed review of all the modifications. ✅ Actions performedReview triggered.
|
- Introduced a new function `mark_as_failed` to update the trigger status to FAILED in case of errors during processing. - Enhanced the `handle_trigger` function to include a try-except block, ensuring that exceptions are logged and the trigger status is updated appropriately when errors occur. - This change improves the robustness of the trigger processing logic by handling failures gracefully.
There was a problem hiding this comment.
Actionable comments posted: 1
📜 Review details
Configuration used: CodeRabbit UI
Review profile: ASSERTIVE
Plan: Pro
📒 Files selected for processing (1)
state-manager/app/tasks/trigger_cron.py(1 hunks)
🧰 Additional context used
🧠 Learnings (1)
📚 Learning: 2025-09-28T13:35:42.851Z
Learnt from: NiveditJain
PR: exospherehost/exospherehost#428
File: state-manager/app/tasks/verify_graph.py:4-5
Timestamp: 2025-09-28T13:35:42.851Z
Learning: In the cron trigger scheduling logic for state-manager/app/tasks/verify_graph.py, the system intentionally schedules at least one trigger beyond the trigger_ahead_time window to ensure continuity of scheduled executions. The current logic of appending an event then breaking is by design to guarantee "at least one next" trigger.
Applied to files:
state-manager/app/tasks/trigger_cron.py
🧬 Code graph analysis (1)
state-manager/app/tasks/trigger_cron.py (6)
state-manager/app/models/db/trigger.py (1)
DatabaseTriggers(9-36)state-manager/app/models/trigger_models.py (2)
TriggerStatusEnum(9-14)TriggerTypeEnum(6-7)state-manager/app/singletons/logs_manager.py (2)
LogsManager(9-66)get_logger(65-66)state-manager/app/controller/trigger_graph.py (1)
trigger_graph(31-115)state-manager/app/models/trigger_graph_model.py (1)
TriggerGraphRequestModel(4-7)state-manager/app/config/settings.py (1)
get_settings(32-36)
🔇 Additional comments (2)
state-manager/app/tasks/trigger_cron.py (2)
47-54: Handle duplicate next-trigger inserts without failing the run
DatabaseTriggershas a unique index on(type, expression, graph_name, namespace, trigger_time). Whenverify_graphor another worker has already enqueued the same next occurrence, this insert raisesDuplicateKeyError. The exception bubbles, we mark the already-fired trigger asFAILED, and the run looks broken even though the graph invocation succeeded. Catch the duplicate, log at info/debug, and continue so we still mark the trigger asTRIGGERED.-from pymongo import ReturnDocument +from pymongo import ReturnDocument +from pymongo.errors import DuplicateKeyError @@ - await DatabaseTriggers( - type=TriggerTypeEnum.CRON, - expression=trigger.expression, - graph_name=trigger.graph_name, - namespace=trigger.namespace, - trigger_time=next_trigger_time, - trigger_status=TriggerStatusEnum.PENDING - ).insert() + try: + await DatabaseTriggers( + type=TriggerTypeEnum.CRON, + expression=trigger.expression, + graph_name=trigger.graph_name, + namespace=trigger.namespace, + trigger_time=next_trigger_time, + trigger_status=TriggerStatusEnum.PENDING, + ).insert() + except DuplicateKeyError: + logger.info( + "next trigger already scheduled", + trigger_id=str(trigger.id), + next_trigger_time=next_trigger_time.isoformat(), + )
73-74: Make cron_time UTC-aware before querying MongoWe still compute
cron_time = datetime.now()(naive local time). Database trigger timestamps are stored in UTC, so on any host not set to UTC we start missing or delaying firings. Please switch todatetime.now(timezone.utc)(and add the import) and log with structured fields instead of string interpolation.-from datetime import datetime +from datetime import datetime, timezone @@ - cron_time = datetime.now() - logger.info(f"starting trigger_cron: {cron_time}") + cron_time = datetime.now(timezone.utc) + logger.info("starting trigger_cron", cron_time=cron_time.isoformat())
- Removed the CronTrigger class usage and replaced it with direct handling of cron expressions for improved clarity and performance. - Updated the logic in `cancel_crons` and `create_crons` functions to work with cron expressions instead of CronTrigger instances, streamlining the trigger management process. - Enhanced the readability of the code by simplifying the data structures used for managing cron triggers.
1 participant
croniteras a dependency inpyproject.tomlto support cron expressions.upsert_graph_templateto handle triggers in graph templates.TriggerandCronTriggermodels to validate and manage triggers.GraphTemplateand request/response models to include triggers.This update improves the flexibility of graph templates by allowing scheduled executions based on cron expressions.