fix(temporal): fix missing reference storage error #214
Conversation
We launch the schedule before inserting it in the database, so we can have instances inserted before schedule is inserted leading to errors.
WalkthroughThe pull request introduces changes to the scheduling process across multiple files in the internal connectors engine. The primary modification involves reordering the schedule creation workflow by first storing the schedule in persistent storage using Changes
Assessment against linked issues
Poem
Tip CodeRabbit's docstrings feature is now available as part of our Early Access Program! Simply use the command Thank you for using CodeRabbit. We offer it for free to the OSS community and would appreciate your support in helping us grow. If you find it useful, would you consider giving us a shout-out on your favorite social media? 🪧 TipsChatThere are 3 ways to chat with CodeRabbit:
Note: Be mindful of the bot's finite context window. It's strongly recommended to break down tasks such as reading entire modules into smaller chunks. For a focused discussion, use review comments to chat about specific files and their changes, instead of using the PR comments. CodeRabbit Commands (Invoked using PR comments)
Other keywords and placeholders
CodeRabbit Configuration File (
|
|
@coderabbitai review |
✅ Actions performedReview triggered.
|
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 0
🧹 Nitpick comments (3)
internal/connectors/engine/workflow/create_payout.go (1)
Line range hint
125-159: Consider wrapping both operations in a transaction-like patternWhile the sequential operations fix the immediate race condition, consider implementing a more robust pattern to handle failures between storage and temporal creation.
Consider implementing a cleanup mechanism in case temporal schedule creation fails:
err = activities.StorageSchedulesStore(...) if err != nil { return err } err = activities.TemporalScheduleCreate(...) if err != nil { + // Cleanup the stored schedule if temporal creation fails + cleanupErr := activities.StorageSchedulesDelete( + infiniteRetryContext(ctx), + scheduleID, + ) + if cleanupErr != nil { + // Log cleanup error but return original error + workflow.GetLogger(ctx).Error("Failed to cleanup schedule after temporal creation failed", + "scheduleID", scheduleID, + "error", cleanupErr) + } return err }internal/connectors/engine/workflow/plugin_workflow.go (1)
Line range hint
118-148: Consider consistent error handling pattern across workflowsSimilar to the create_payout workflow, consider implementing the same robust cleanup pattern here.
Consider implementing the same cleanup mechanism in case temporal schedule creation fails:
err = activities.StorageSchedulesStore(...) if err != nil { return err } err = activities.TemporalScheduleCreate(...) if err != nil { + // Cleanup the stored schedule if temporal creation fails + cleanupErr := activities.StorageSchedulesDelete( + infiniteRetryContext(ctx), + scheduleID, + ) + if cleanupErr != nil { + // Log cleanup error but return original error + workflow.GetLogger(ctx).Error("Failed to cleanup schedule after temporal creation failed", + "scheduleID", scheduleID, + "error", cleanupErr) + } return err }internal/connectors/engine/workflow/create_transfer.go (1)
116-127: Consider improving error handling and timestamp consistency.A few suggestions for improvement:
- Consider adding cleanup logic for the stored schedule if temporal schedule creation fails
- Ensure consistent usage of UTC timestamps across both storage and schedule creation
- Consider validating the schedule ID for special characters that might cause issues
Here's a suggested improvement:
err = activities.StorageSchedulesStore( infiniteRetryContext(ctx), models.Schedule{ ID: scheduleID, ConnectorID: createTransfer.ConnectorID, CreatedAt: workflow.Now(ctx).UTC(), }) if err != nil { return err } +// Ensure cleanup if temporal schedule creation fails err = activities.TemporalScheduleCreate( infiniteRetryContext(ctx), activities.ScheduleCreateOptions{ ScheduleID: scheduleID, // ... other options ... }, ) if err != nil { + // Attempt to clean up the stored schedule + if cleanupErr := activities.StorageSchedulesDelete( + infiniteRetryContext(ctx), + scheduleID, + ); cleanupErr != nil { + return fmt.Errorf("schedule creation failed: %v, cleanup failed: %v", err, cleanupErr) + } return err }
📜 Review details
Configuration used: CodeRabbit UI
Review profile: CHILL
Plan: Pro
📒 Files selected for processing (4)
internal/connectors/engine/activities/temporal_schedule_create.go(3 hunks)internal/connectors/engine/workflow/create_payout.go(1 hunks)internal/connectors/engine/workflow/create_transfer.go(1 hunks)internal/connectors/engine/workflow/plugin_workflow.go(1 hunks)
🔇 Additional comments (6)
internal/connectors/engine/activities/temporal_schedule_create.go (2)
Line range hint 23-52: LGTM! Function signature simplified to focus on error handling
The removal of the schedule ID return value is a good change since the ID is now managed externally, making the function more focused.
57-61: LGTM! Workflow helper updated consistently
The workflow helper function has been updated to match the new signature, maintaining consistency.
internal/connectors/engine/workflow/create_payout.go (1)
113-124: LGTM! Schedule storage added before temporal creation
Good addition of schedule storage before temporal creation to prevent race conditions.
internal/connectors/engine/workflow/plugin_workflow.go (2)
107-116: LGTM! Schedule storage added before temporal creation
Good addition of schedule storage before temporal creation to prevent race conditions.
Line range hint 107-148: Verify the impact of jitter on schedule creation
The addition of jitter might affect the timing of schedule execution.
Let's verify the jitter configuration across the codebase:
✅ Verification successful
Jitter configuration is consistently implemented and safe
The verification shows that:
- Jitter is consistently set to half of the polling period (
config.PollingPeriod / 2) - The polling period has a minimum safety threshold of 30 seconds as enforced in
internal/models/config.go - The jitter configuration is properly passed through to the temporal schedule creation activity
🏁 Scripts executed
The following scripts were executed for the analysis:
Script:
#!/bin/bash
# Description: Check for jitter configurations in schedule creation
# to ensure consistent usage across the codebase
# Search for jitter configurations in schedule creation
rg -A 2 "Jitter:\s+"
# Search for PollingPeriod configurations to verify jitter calculation
rg "PollingPeriod"
Length of output: 4400
internal/connectors/engine/workflow/create_transfer.go (1)
115-125: LGTM! The schedule storage implementation fixes the reference issue.
The implementation correctly addresses the core issue by ensuring the schedule is stored in persistent storage before creating the temporal schedule. The error handling ensures atomic operations.
Let's verify the schedule storage implementation:
✅ Verification successful
Schedule storage implementation is correctly synchronized across the codebase
Based on the verification results, the implementation follows a consistent pattern across all workflow files (create_transfer.go, create_payout.go, and plugin_workflow.go). In each case:
- The schedule is stored in persistent storage using
StorageSchedulesStorebefore creating the temporal schedule - The same error handling pattern and UTC timestamp usage is maintained
- Schedule IDs follow a consistent format with proper namespacing
Key findings:
- All three workflow files that use
TemporalScheduleCreateproperly store schedules first - The error handling pattern is consistent across all implementations
- The activity registration in
activity.goconfirms proper integration
🏁 Scripts executed
The following scripts were executed for the analysis:
Script:
#!/bin/bash
# Description: Verify the schedule storage implementation across the codebase
# Check for consistent schedule storage pattern and potential race conditions
# Check for other places where TemporalScheduleCreate is used without StorageSchedulesStore
ast-grep --pattern 'activities.TemporalScheduleCreate($$$)' | grep -B 5 'TemporalScheduleCreate' | grep -v 'StorageSchedulesStore'
# Check for proper error handling pattern
ast-grep --pattern 'err := activities.StorageSchedulesStore($$$)
if err != nil {
return err
}'
Length of output: 227
Script:
#!/bin/bash
# Let's try a different approach to verify the schedule storage implementation
# 1. Find all files that use TemporalScheduleCreate
rg "TemporalScheduleCreate" -l
# 2. Search for the pattern with more context
rg "TemporalScheduleCreate" -A 5 -B 5
# 3. Look for StorageSchedulesStore usage
rg "StorageSchedulesStore" -A 5 -B 5
# 4. Find all workflow files that might handle schedules
fd "workflow.*\.go$"
Length of output: 12760
We launch the schedule before inserting it in the
database, so we can have instances inserted before schedule is inserted leading to errors.
Fixes ENG-1569