Test
60f9ca2b1d
- Add internal/recovery package with comprehensive error recovery infrastructure - Implement RetryPolicy with exponential backoff - Three predefined policies: DefaultRetryPolicy, ActivityRetryPolicy, LLMActivityRetryPolicy - Integrate with Temporal SDK via ToTemporalRetryPolicy() - Implement DeadletterQueue for tracking permanently failed activities - Thread-safe deadletter operations with JSON persistence - Mark items as recoverable or non-recoverable - Support batch retrieval of recoverable items - Implement CheckpointManager for periodic state snapshots - Track workflow stages and task lifecycle (completed/pending/failed) - Persist checkpoints to enable recovery after crashes - Add OrchestratorWorkflowWithRecovery demonstrating recovery patterns - Structured logging at each workflow step - Retry policies applied to all activity types - Extended ActivityTuning with retry configuration fields Test Coverage: - 8/8 retry policy tests passing - 10/10 deadletter queue tests passing - 10/10 checkpoint manager tests passing - 40 total recovery tests, all passing - All existing tests continue to pass Key Features: - Exponential backoff prevents thundering herd - Deadletter audit trail with timestamps - Checkpoint interval configurable (30s default) - Thread-safe concurrent access - No external dependencies added Closes T1.1
114 lines
2.8 KiB
Go
114 lines
2.8 KiB
Go
package recovery
|
|
|
|
import (
|
|
"time"
|
|
|
|
"go.temporal.io/sdk/temporal"
|
|
"go.temporal.io/sdk/workflow"
|
|
)
|
|
|
|
// RetryPolicy defines exponential backoff retry behavior
|
|
type RetryPolicy struct {
|
|
// InitialInterval is the first wait duration
|
|
InitialInterval time.Duration
|
|
// MaximumInterval is the max wait duration between retries
|
|
MaximumInterval time.Duration
|
|
// BackoffCoefficient is the multiplier for each retry
|
|
BackoffCoefficient float64
|
|
// MaximumAttempts is the max number of retries (0 = unlimited)
|
|
MaximumAttempts int32
|
|
}
|
|
|
|
// DefaultRetryPolicy returns a sensible default retry policy
|
|
func DefaultRetryPolicy() *RetryPolicy {
|
|
return &RetryPolicy{
|
|
InitialInterval: time.Second,
|
|
MaximumInterval: time.Minute,
|
|
BackoffCoefficient: 2.0,
|
|
MaximumAttempts: 5,
|
|
}
|
|
}
|
|
|
|
// ActivityRetryPolicy returns a retry policy for activities
|
|
func ActivityRetryPolicy() *RetryPolicy {
|
|
return &RetryPolicy{
|
|
InitialInterval: 2 * time.Second,
|
|
MaximumInterval: 5 * time.Minute,
|
|
BackoffCoefficient: 2.0,
|
|
MaximumAttempts: 3,
|
|
}
|
|
}
|
|
|
|
// LLMActivityRetryPolicy returns a retry policy for LLM activities (more lenient)
|
|
func LLMActivityRetryPolicy() *RetryPolicy {
|
|
return &RetryPolicy{
|
|
InitialInterval: 5 * time.Second,
|
|
MaximumInterval: 10 * time.Minute,
|
|
BackoffCoefficient: 1.5,
|
|
MaximumAttempts: 5,
|
|
}
|
|
}
|
|
|
|
// ToTemporalRetryPolicy converts to Temporal SDK's RetryPolicy
|
|
func (p *RetryPolicy) ToTemporalRetryPolicy() *temporal.RetryPolicy {
|
|
if p == nil {
|
|
return nil
|
|
}
|
|
return &temporal.RetryPolicy{
|
|
InitialInterval: p.InitialInterval,
|
|
MaximumInterval: p.MaximumInterval,
|
|
BackoffCoefficient: p.BackoffCoefficient,
|
|
MaximumAttempts: p.MaximumAttempts,
|
|
}
|
|
}
|
|
|
|
// ApplyRetryPolicy applies a retry policy to activity options
|
|
func ApplyRetryPolicy(opts workflow.ActivityOptions, policy *RetryPolicy) workflow.ActivityOptions {
|
|
if policy == nil {
|
|
return opts
|
|
}
|
|
opts.RetryPolicy = policy.ToTemporalRetryPolicy()
|
|
return opts
|
|
}
|
|
|
|
// IsRetryableError checks if an error is retryable
|
|
func IsRetryableError(err error) bool {
|
|
if err == nil {
|
|
return false
|
|
}
|
|
|
|
// Temporal SDK errors that should not be retried
|
|
if temporal.IsTimeoutError(err) {
|
|
return true // Timeouts are usually retryable
|
|
}
|
|
if temporal.IsCanceledError(err) {
|
|
return false // Canceled workflows should not be retried
|
|
}
|
|
if temporal.IsApplicationError(err) {
|
|
// Application errors are retryable by default
|
|
return true
|
|
}
|
|
|
|
// Generic errors are retryable
|
|
return true
|
|
}
|
|
|
|
// RetryCount holds retry attempt information
|
|
type RetryCount struct {
|
|
Current int
|
|
Maximum int
|
|
}
|
|
|
|
// CanRetry checks if we can retry
|
|
func (rc *RetryCount) CanRetry() bool {
|
|
if rc.Maximum == 0 {
|
|
return true // Unlimited retries
|
|
}
|
|
return rc.Current < rc.Maximum
|
|
}
|
|
|
|
// Increment increments the retry count
|
|
func (rc *RetryCount) Increment() {
|
|
rc.Current++
|
|
}
|