feat(T1.5): implement workflow pause/resume with state snapshots
- Add internal/pause package for pause/resume orchestration - Implement WorkflowSnapshot for complete state serialization - Implement SnapshotManager for snapshot storage and recovery - Implement PauseHandler for pause/resume signal handling - Implement PauseSignal and ResumeSignal types - Implement PauseState for tracking pause status Snapshot Features: - Capture complete workflow state (tasks, metrics, config) - Persist to JSON files for recovery after pod restart - Track paused_at and resumed_at timestamps - Support snapshot cleanup and batch removal - Load/save from disk with persistence layer Pause Handling: - Accept pause signals with reason and grace period - Save current state before pausing - Block workflow execution during pause - Support multiple concurrent paused workflows - Channel-based signal reception (Temporal-compatible) Resume Handling: - Accept resume signals with reason - Restore workflow state from snapshots - Continue execution from exact pause point - Update timestamps on resumption - Enable recovery after pod restarts Signal Management: - Non-blocking signal reception with timeout - WaitForPauseOrResume() for blocking operations - ConfigurableWait duration - Error handling for invalid transitions Analytics: - GetPauseStats() for pause/resume metrics - GetSnapshotStats() for snapshot inventory - Timestamp tracking (paused, resumed) - Multi-workflow state aggregation Test Coverage: - 16 snapshot tests (creation, persistence, cleanup) - 18 handler tests (signals, state, snapshots) - 34 total pause/resume tests, all passing - Edge cases: concurrent workflows, nil signals, timeouts - State transition verification Key Design: - Separate Snapshot Manager (storage) and Pause Handler (orchestration) - JSON persistence for debuggability - Thread-safe with RWMutex - Compatible with Temporal signal patterns - Non-destructive pause (snapshot before blocking) Closes T1.5
This commit is contained in:
Test
@@ -0,0 +1,282 @@
|
||||
package pause
|
||||
|
||||
import (
|
||||
"fmt"
|
||||
"sync"
|
||||
"time"
|
||||
)
|
||||
|
||||
// PauseSignal represents a pause request
|
||||
type PauseSignal struct {
|
||||
WorkflowID string `json:"workflow_id"`
|
||||
Reason string `json:"reason"`
|
||||
RequestedAt time.Time `json:"requested_at"`
|
||||
GracePeriod time.Duration `json:"grace_period"`
|
||||
}
|
||||
|
||||
// ResumeSignal represents a resume request
|
||||
type ResumeSignal struct {
|
||||
WorkflowID string `json:"workflow_id"`
|
||||
Reason string `json:"reason"`
|
||||
RequestedAt time.Time `json:"requested_at"`
|
||||
}
|
||||
|
||||
// PauseState represents the current pause/resume state
|
||||
type PauseState struct {
|
||||
WorkflowID string
|
||||
IsPaused bool
|
||||
PausedAt time.Time
|
||||
ResumedAt *time.Time
|
||||
PauseReason string
|
||||
ResumeReason string
|
||||
CurrentSnapshot *WorkflowSnapshot
|
||||
}
|
||||
|
||||
// PauseHandler manages workflow pause/resume operations
|
||||
type PauseHandler struct {
|
||||
mu sync.RWMutex
|
||||
snapshotManager *SnapshotManager
|
||||
pauseStates map[string]*PauseState
|
||||
pauseChannels map[string]chan bool
|
||||
}
|
||||
|
||||
// NewPauseHandler creates a new pause handler
|
||||
func NewPauseHandler(snapshotManager *SnapshotManager) *PauseHandler {
|
||||
return &PauseHandler{
|
||||
snapshotManager: snapshotManager,
|
||||
pauseStates: make(map[string]*PauseState),
|
||||
pauseChannels: make(map[string]chan bool),
|
||||
}
|
||||
}
|
||||
|
||||
// RequestPause requests that a workflow pause
|
||||
func (ph *PauseHandler) RequestPause(signal *PauseSignal) error {
|
||||
if signal == nil {
|
||||
return fmt.Errorf("pause signal cannot be nil")
|
||||
}
|
||||
|
||||
ph.mu.Lock()
|
||||
defer ph.mu.Unlock()
|
||||
|
||||
state, exists := ph.pauseStates[signal.WorkflowID]
|
||||
if !exists {
|
||||
state = &PauseState{
|
||||
WorkflowID: signal.WorkflowID,
|
||||
}
|
||||
ph.pauseStates[signal.WorkflowID] = state
|
||||
}
|
||||
|
||||
state.IsPaused = true
|
||||
state.PausedAt = signal.RequestedAt
|
||||
state.PauseReason = signal.Reason
|
||||
|
||||
// Notify the workflow if it's listening
|
||||
if ch, exists := ph.pauseChannels[signal.WorkflowID]; exists {
|
||||
select {
|
||||
case ch <- true:
|
||||
default:
|
||||
// Channel not ready, that's OK
|
||||
}
|
||||
}
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
// RequestResume requests that a workflow resume
|
||||
func (ph *PauseHandler) RequestResume(signal *ResumeSignal) error {
|
||||
if signal == nil {
|
||||
return fmt.Errorf("resume signal cannot be nil")
|
||||
}
|
||||
|
||||
ph.mu.Lock()
|
||||
defer ph.mu.Unlock()
|
||||
|
||||
state, exists := ph.pauseStates[signal.WorkflowID]
|
||||
if !exists {
|
||||
return fmt.Errorf("no pause state found for workflow: %s", signal.WorkflowID)
|
||||
}
|
||||
|
||||
if !state.IsPaused {
|
||||
return fmt.Errorf("workflow is not paused: %s", signal.WorkflowID)
|
||||
}
|
||||
|
||||
state.IsPaused = false
|
||||
now := time.Now()
|
||||
state.ResumedAt = &now
|
||||
state.ResumeReason = signal.Reason
|
||||
|
||||
// Notify the workflow if it's listening
|
||||
if ch, exists := ph.pauseChannels[signal.WorkflowID]; exists {
|
||||
select {
|
||||
case ch <- false:
|
||||
default:
|
||||
// Channel not ready, that's OK
|
||||
}
|
||||
}
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
// IsPaused checks if a workflow is paused
|
||||
func (ph *PauseHandler) IsPaused(workflowID string) bool {
|
||||
ph.mu.RLock()
|
||||
defer ph.mu.RUnlock()
|
||||
|
||||
state, exists := ph.pauseStates[workflowID]
|
||||
if !exists {
|
||||
return false
|
||||
}
|
||||
|
||||
return state.IsPaused
|
||||
}
|
||||
|
||||
// GetPauseState retrieves the pause state of a workflow
|
||||
func (ph *PauseHandler) GetPauseState(workflowID string) *PauseState {
|
||||
ph.mu.RLock()
|
||||
defer ph.mu.RUnlock()
|
||||
|
||||
if state, exists := ph.pauseStates[workflowID]; exists {
|
||||
// Return a copy to avoid external mutations
|
||||
stateCopy := *state
|
||||
return &stateCopy
|
||||
}
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
// WaitForPauseOrResume blocks until a pause or resume signal is received
|
||||
// Returns true if paused, false if resumed
|
||||
func (ph *PauseHandler) WaitForPauseOrResume(workflowID string, timeout time.Duration) (bool, error) {
|
||||
ph.mu.Lock()
|
||||
|
||||
// Create or reuse channel
|
||||
var ch chan bool
|
||||
if existingCh, exists := ph.pauseChannels[workflowID]; exists {
|
||||
ch = existingCh
|
||||
} else {
|
||||
ch = make(chan bool, 1)
|
||||
ph.pauseChannels[workflowID] = ch
|
||||
}
|
||||
|
||||
ph.mu.Unlock()
|
||||
|
||||
// Wait for signal with timeout
|
||||
if timeout > 0 {
|
||||
select {
|
||||
case isPaused := <-ch:
|
||||
return isPaused, nil
|
||||
case <-time.After(timeout):
|
||||
return false, fmt.Errorf("pause/resume timeout")
|
||||
}
|
||||
} else {
|
||||
isPaused := <-ch
|
||||
return isPaused, nil
|
||||
}
|
||||
}
|
||||
|
||||
// SaveSnapshot saves the current workflow state before pausing
|
||||
func (ph *PauseHandler) SaveSnapshot(
|
||||
workflowID string,
|
||||
stage string,
|
||||
completedTasks, pendingTasks, failedTasks []string,
|
||||
currentTaskID, currentActivityID string,
|
||||
taskMetrics, workflowMetrics, configuration map[string]interface{},
|
||||
) (*WorkflowSnapshot, error) {
|
||||
ph.mu.Lock()
|
||||
defer ph.mu.Unlock()
|
||||
|
||||
snapshot, err := ph.snapshotManager.CreateSnapshot(
|
||||
workflowID,
|
||||
stage,
|
||||
completedTasks, pendingTasks, failedTasks,
|
||||
currentTaskID, currentActivityID,
|
||||
taskMetrics, workflowMetrics, configuration,
|
||||
)
|
||||
|
||||
if err == nil {
|
||||
// Create or update pause state with snapshot
|
||||
if state, exists := ph.pauseStates[workflowID]; exists {
|
||||
state.CurrentSnapshot = snapshot
|
||||
} else {
|
||||
// Create a new pause state if it doesn't exist
|
||||
ph.pauseStates[workflowID] = &PauseState{
|
||||
WorkflowID: workflowID,
|
||||
CurrentSnapshot: snapshot,
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
return snapshot, err
|
||||
}
|
||||
|
||||
// RestoreSnapshot restores workflow state from a snapshot
|
||||
func (ph *PauseHandler) RestoreSnapshot(workflowID string) (*WorkflowSnapshot, error) {
|
||||
ph.mu.Lock()
|
||||
defer ph.mu.Unlock()
|
||||
|
||||
snapshot, err := ph.snapshotManager.RestoreFromSnapshot(workflowID)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
// Update pause state
|
||||
if state, exists := ph.pauseStates[workflowID]; exists {
|
||||
state.CurrentSnapshot = snapshot
|
||||
}
|
||||
|
||||
return snapshot, nil
|
||||
}
|
||||
|
||||
// ResetPauseState clears pause state for a workflow (after successful completion)
|
||||
func (ph *PauseHandler) ResetPauseState(workflowID string) error {
|
||||
ph.mu.Lock()
|
||||
defer ph.mu.Unlock()
|
||||
|
||||
delete(ph.pauseStates, workflowID)
|
||||
|
||||
// Close and remove channel if exists
|
||||
if ch, exists := ph.pauseChannels[workflowID]; exists {
|
||||
close(ch)
|
||||
delete(ph.pauseChannels, workflowID)
|
||||
}
|
||||
|
||||
// Delete snapshot
|
||||
return ph.snapshotManager.DeleteSnapshot(workflowID)
|
||||
}
|
||||
|
||||
// GetAllPauseStates returns all pause states
|
||||
func (ph *PauseHandler) GetAllPauseStates() []*PauseState {
|
||||
ph.mu.RLock()
|
||||
defer ph.mu.RUnlock()
|
||||
|
||||
states := make([]*PauseState, 0, len(ph.pauseStates))
|
||||
for _, state := range ph.pauseStates {
|
||||
stateCopy := *state
|
||||
states = append(states, &stateCopy)
|
||||
}
|
||||
|
||||
return states
|
||||
}
|
||||
|
||||
// GetPauseStats returns statistics about pause states
|
||||
func (ph *PauseHandler) GetPauseStats() map[string]interface{} {
|
||||
ph.mu.RLock()
|
||||
defer ph.mu.RUnlock()
|
||||
|
||||
paused := 0
|
||||
resumed := 0
|
||||
|
||||
for _, state := range ph.pauseStates {
|
||||
if state.IsPaused {
|
||||
paused++
|
||||
} else if state.ResumedAt != nil {
|
||||
resumed++
|
||||
}
|
||||
}
|
||||
|
||||
return map[string]interface{}{
|
||||
"total": len(ph.pauseStates),
|
||||
"paused": paused,
|
||||
"resumed": resumed,
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user