Test
37d7aea5a7
- 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
258 lines
6.7 KiB
Go
258 lines
6.7 KiB
Go
package pause
|
|
|
|
import (
|
|
"fmt"
|
|
"testing"
|
|
"time"
|
|
|
|
"github.com/stretchr/testify/assert"
|
|
)
|
|
|
|
func TestRequestPause(t *testing.T) {
|
|
tmpDir := t.TempDir()
|
|
sm := NewSnapshotManager(tmpDir)
|
|
ph := NewPauseHandler(sm)
|
|
|
|
signal := &PauseSignal{
|
|
WorkflowID: "wf-1",
|
|
Reason: "manual pause",
|
|
RequestedAt: time.Now(),
|
|
}
|
|
|
|
err := ph.RequestPause(signal)
|
|
assert.NoError(t, err)
|
|
assert.True(t, ph.IsPaused("wf-1"))
|
|
}
|
|
|
|
func TestRequestResume(t *testing.T) {
|
|
tmpDir := t.TempDir()
|
|
sm := NewSnapshotManager(tmpDir)
|
|
ph := NewPauseHandler(sm)
|
|
|
|
// First pause
|
|
ph.RequestPause(&PauseSignal{WorkflowID: "wf-1", Reason: "pause", RequestedAt: time.Now()})
|
|
assert.True(t, ph.IsPaused("wf-1"))
|
|
|
|
// Then resume
|
|
err := ph.RequestResume(&ResumeSignal{WorkflowID: "wf-1", Reason: "resume", RequestedAt: time.Now()})
|
|
assert.NoError(t, err)
|
|
assert.False(t, ph.IsPaused("wf-1"))
|
|
}
|
|
|
|
func TestIsPaused(t *testing.T) {
|
|
tmpDir := t.TempDir()
|
|
sm := NewSnapshotManager(tmpDir)
|
|
ph := NewPauseHandler(sm)
|
|
|
|
assert.False(t, ph.IsPaused("wf-1"))
|
|
|
|
ph.RequestPause(&PauseSignal{WorkflowID: "wf-1", Reason: "pause", RequestedAt: time.Now()})
|
|
assert.True(t, ph.IsPaused("wf-1"))
|
|
}
|
|
|
|
func TestGetPauseState(t *testing.T) {
|
|
tmpDir := t.TempDir()
|
|
sm := NewSnapshotManager(tmpDir)
|
|
ph := NewPauseHandler(sm)
|
|
|
|
ph.RequestPause(&PauseSignal{WorkflowID: "wf-1", Reason: "pause", RequestedAt: time.Now()})
|
|
state := ph.GetPauseState("wf-1")
|
|
|
|
assert.NotNil(t, state)
|
|
assert.Equal(t, "wf-1", state.WorkflowID)
|
|
assert.True(t, state.IsPaused)
|
|
assert.Equal(t, "pause", state.PauseReason)
|
|
}
|
|
|
|
func TestSaveSnapshot(t *testing.T) {
|
|
tmpDir := t.TempDir()
|
|
sm := NewSnapshotManager(tmpDir)
|
|
ph := NewPauseHandler(sm)
|
|
|
|
snapshot, err := ph.SaveSnapshot(
|
|
"wf-1",
|
|
"stage1",
|
|
[]string{"T1.1"},
|
|
[]string{"T1.2"},
|
|
nil,
|
|
"T1.2",
|
|
"activity-1",
|
|
nil,
|
|
nil,
|
|
nil,
|
|
)
|
|
|
|
assert.NoError(t, err)
|
|
assert.NotNil(t, snapshot)
|
|
assert.Equal(t, "wf-1", snapshot.WorkflowID)
|
|
}
|
|
|
|
func TestRestoreSnapshot(t *testing.T) {
|
|
tmpDir := t.TempDir()
|
|
sm := NewSnapshotManager(tmpDir)
|
|
ph := NewPauseHandler(sm)
|
|
|
|
// Save snapshot
|
|
ph.SaveSnapshot("wf-1", "stage1", []string{"T1.1"}, []string{"T1.2"}, nil, "T1.2", "", nil, nil, nil)
|
|
|
|
// Restore it
|
|
snapshot, err := ph.RestoreSnapshot("wf-1")
|
|
assert.NoError(t, err)
|
|
assert.NotNil(t, snapshot)
|
|
assert.Equal(t, "wf-1", snapshot.WorkflowID)
|
|
}
|
|
|
|
func TestResetPauseState(t *testing.T) {
|
|
tmpDir := t.TempDir()
|
|
sm := NewSnapshotManager(tmpDir)
|
|
ph := NewPauseHandler(sm)
|
|
|
|
ph.RequestPause(&PauseSignal{WorkflowID: "wf-1", Reason: "pause", RequestedAt: time.Now()})
|
|
assert.True(t, ph.IsPaused("wf-1"))
|
|
|
|
err := ph.ResetPauseState("wf-1")
|
|
assert.NoError(t, err)
|
|
assert.Nil(t, ph.GetPauseState("wf-1"))
|
|
}
|
|
|
|
func TestGetAllPauseStates(t *testing.T) {
|
|
tmpDir := t.TempDir()
|
|
sm := NewSnapshotManager(tmpDir)
|
|
ph := NewPauseHandler(sm)
|
|
|
|
ph.RequestPause(&PauseSignal{WorkflowID: "wf-1", Reason: "pause", RequestedAt: time.Now()})
|
|
ph.RequestPause(&PauseSignal{WorkflowID: "wf-2", Reason: "pause", RequestedAt: time.Now()})
|
|
ph.RequestPause(&PauseSignal{WorkflowID: "wf-3", Reason: "pause", RequestedAt: time.Now()})
|
|
|
|
states := ph.GetAllPauseStates()
|
|
assert.Equal(t, 3, len(states))
|
|
}
|
|
|
|
func TestGetPauseStats(t *testing.T) {
|
|
tmpDir := t.TempDir()
|
|
sm := NewSnapshotManager(tmpDir)
|
|
ph := NewPauseHandler(sm)
|
|
|
|
ph.RequestPause(&PauseSignal{WorkflowID: "wf-1", Reason: "pause", RequestedAt: time.Now()})
|
|
ph.RequestPause(&PauseSignal{WorkflowID: "wf-2", Reason: "pause", RequestedAt: time.Now()})
|
|
ph.RequestResume(&ResumeSignal{WorkflowID: "wf-1", Reason: "resume", RequestedAt: time.Now()})
|
|
|
|
stats := ph.GetPauseStats()
|
|
assert.Equal(t, 2, stats["total"])
|
|
assert.Equal(t, 1, stats["paused"])
|
|
assert.Equal(t, 1, stats["resumed"])
|
|
}
|
|
|
|
func TestPauseStateFields(t *testing.T) {
|
|
tmpDir := t.TempDir()
|
|
sm := NewSnapshotManager(tmpDir)
|
|
ph := NewPauseHandler(sm)
|
|
|
|
pausedTime := time.Now()
|
|
ph.RequestPause(&PauseSignal{WorkflowID: "wf-1", Reason: "manual pause", RequestedAt: pausedTime})
|
|
|
|
state := ph.GetPauseState("wf-1")
|
|
assert.Equal(t, "wf-1", state.WorkflowID)
|
|
assert.True(t, state.IsPaused)
|
|
assert.Equal(t, "manual pause", state.PauseReason)
|
|
assert.NotZero(t, state.PausedAt)
|
|
}
|
|
|
|
func TestResumedAt(t *testing.T) {
|
|
tmpDir := t.TempDir()
|
|
sm := NewSnapshotManager(tmpDir)
|
|
ph := NewPauseHandler(sm)
|
|
|
|
ph.RequestPause(&PauseSignal{WorkflowID: "wf-1", Reason: "pause", RequestedAt: time.Now()})
|
|
ph.RequestResume(&ResumeSignal{WorkflowID: "wf-1", Reason: "resume", RequestedAt: time.Now()})
|
|
|
|
state := ph.GetPauseState("wf-1")
|
|
assert.NotNil(t, state.ResumedAt)
|
|
}
|
|
|
|
func TestResumeNotPausedError(t *testing.T) {
|
|
tmpDir := t.TempDir()
|
|
sm := NewSnapshotManager(tmpDir)
|
|
ph := NewPauseHandler(sm)
|
|
|
|
// Try to resume without pausing first
|
|
err := ph.RequestResume(&ResumeSignal{WorkflowID: "wf-1", Reason: "resume", RequestedAt: time.Now()})
|
|
assert.Error(t, err)
|
|
}
|
|
|
|
func TestNilSignals(t *testing.T) {
|
|
tmpDir := t.TempDir()
|
|
sm := NewSnapshotManager(tmpDir)
|
|
ph := NewPauseHandler(sm)
|
|
|
|
err := ph.RequestPause(nil)
|
|
assert.Error(t, err)
|
|
|
|
err = ph.RequestResume(nil)
|
|
assert.Error(t, err)
|
|
}
|
|
|
|
func TestMultipleWorkflowsPause(t *testing.T) {
|
|
tmpDir := t.TempDir()
|
|
sm := NewSnapshotManager(tmpDir)
|
|
ph := NewPauseHandler(sm)
|
|
|
|
for i := 1; i <= 5; i++ {
|
|
wfID := fmt.Sprintf("wf-%d", i)
|
|
ph.RequestPause(&PauseSignal{WorkflowID: wfID, Reason: "pause", RequestedAt: time.Now()})
|
|
}
|
|
|
|
states := ph.GetAllPauseStates()
|
|
assert.Equal(t, 5, len(states))
|
|
|
|
for _, state := range states {
|
|
assert.True(t, state.IsPaused)
|
|
}
|
|
}
|
|
|
|
func TestWaitForPauseOrResume(t *testing.T) {
|
|
tmpDir := t.TempDir()
|
|
sm := NewSnapshotManager(tmpDir)
|
|
ph := NewPauseHandler(sm)
|
|
|
|
// Send pause signal in goroutine
|
|
go func() {
|
|
time.Sleep(100 * time.Millisecond)
|
|
ph.RequestPause(&PauseSignal{WorkflowID: "wf-1", Reason: "pause", RequestedAt: time.Now()})
|
|
}()
|
|
|
|
// Wait for pause
|
|
isPaused, err := ph.WaitForPauseOrResume("wf-1", 1*time.Second)
|
|
assert.NoError(t, err)
|
|
assert.True(t, isPaused)
|
|
}
|
|
|
|
func TestWaitTimeout(t *testing.T) {
|
|
tmpDir := t.TempDir()
|
|
sm := NewSnapshotManager(tmpDir)
|
|
ph := NewPauseHandler(sm)
|
|
|
|
// Wait with timeout should fail
|
|
_, err := ph.WaitForPauseOrResume("wf-1", 100*time.Millisecond)
|
|
assert.Error(t, err)
|
|
}
|
|
|
|
func TestSnapshotWithPause(t *testing.T) {
|
|
tmpDir := t.TempDir()
|
|
sm := NewSnapshotManager(tmpDir)
|
|
ph := NewPauseHandler(sm)
|
|
|
|
// Save snapshot before pausing
|
|
snapshot, err := ph.SaveSnapshot("wf-1", "stage1", []string{"T1.1"}, []string{"T1.2"}, nil, "T1.2", "", nil, nil, nil)
|
|
assert.NoError(t, err)
|
|
|
|
// Pause
|
|
ph.RequestPause(&PauseSignal{WorkflowID: "wf-1", Reason: "pause", RequestedAt: time.Now()})
|
|
|
|
// State should have snapshot
|
|
state := ph.GetPauseState("wf-1")
|
|
assert.NotNil(t, state)
|
|
assert.NotNil(t, state.CurrentSnapshot)
|
|
assert.Equal(t, snapshot.WorkflowID, state.CurrentSnapshot.WorkflowID)
|
|
}
|