Files
poimen-workflows/internal/pause/snapshot_test.go
T
Test 37d7aea5a7 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
2026-08-23 17:11:56 -07:00

229 lines
6.0 KiB
Go

package pause
import (
"fmt"
"testing"
"time"
"github.com/stretchr/testify/assert"
)
func TestCreateSnapshot(t *testing.T) {
tmpDir := t.TempDir()
sm := NewSnapshotManager(tmpDir)
snapshot, err := sm.CreateSnapshot(
"wf-1",
"implement",
[]string{"T1.1", "T1.2"},
[]string{"T1.3", "T1.4"},
[]string{},
"T1.3",
"activity-1",
map[string]interface{}{"duration": 42.5},
map[string]interface{}{"total_time": 300},
map[string]interface{}{"timeout": 600},
)
assert.NoError(t, err)
assert.NotNil(t, snapshot)
assert.Equal(t, "wf-1", snapshot.WorkflowID)
assert.Equal(t, "implement", snapshot.Stage)
assert.Equal(t, 2, len(snapshot.CompletedTasks))
assert.Equal(t, 2, len(snapshot.PendingTasks))
assert.NotZero(t, snapshot.PausedAt)
}
func TestGetLatestSnapshot(t *testing.T) {
tmpDir := t.TempDir()
sm := NewSnapshotManager(tmpDir)
sm.CreateSnapshot("wf-1", "stage1", nil, nil, nil, "", "", nil, nil, nil)
retrieved := sm.GetLatestSnapshot("wf-1")
assert.NotNil(t, retrieved)
assert.Equal(t, "wf-1", retrieved.WorkflowID)
}
func TestHasSnapshot(t *testing.T) {
tmpDir := t.TempDir()
sm := NewSnapshotManager(tmpDir)
assert.False(t, sm.HasSnapshot("wf-1"))
sm.CreateSnapshot("wf-1", "stage1", nil, nil, nil, "", "", nil, nil, nil)
assert.True(t, sm.HasSnapshot("wf-1"))
}
func TestRestoreFromSnapshot(t *testing.T) {
tmpDir := t.TempDir()
sm := NewSnapshotManager(tmpDir)
_, _ = sm.CreateSnapshot("wf-1", "stage1", []string{"T1.1"}, []string{"T1.2"}, nil, "T1.2", "", nil, nil, nil)
restored, err := sm.RestoreFromSnapshot("wf-1")
assert.NoError(t, err)
assert.NotNil(t, restored)
assert.Equal(t, "wf-1", restored.WorkflowID)
}
func TestMarkResumed(t *testing.T) {
tmpDir := t.TempDir()
sm := NewSnapshotManager(tmpDir)
sm.CreateSnapshot("wf-1", "stage1", nil, nil, nil, "", "", nil, nil, nil)
err := sm.MarkResumed("wf-1")
assert.NoError(t, err)
snapshot := sm.GetLatestSnapshot("wf-1")
assert.NotNil(t, snapshot.ResumedAt)
}
func TestDeleteSnapshot(t *testing.T) {
tmpDir := t.TempDir()
sm := NewSnapshotManager(tmpDir)
sm.CreateSnapshot("wf-1", "stage1", nil, nil, nil, "", "", nil, nil, nil)
assert.True(t, sm.HasSnapshot("wf-1"))
err := sm.DeleteSnapshot("wf-1")
assert.NoError(t, err)
assert.False(t, sm.HasSnapshot("wf-1"))
}
func TestGetAllSnapshots(t *testing.T) {
tmpDir := t.TempDir()
sm := NewSnapshotManager(tmpDir)
sm.CreateSnapshot("wf-1", "stage1", nil, nil, nil, "", "", nil, nil, nil)
sm.CreateSnapshot("wf-2", "stage1", nil, nil, nil, "", "", nil, nil, nil)
sm.CreateSnapshot("wf-3", "stage1", nil, nil, nil, "", "", nil, nil, nil)
snapshots := sm.GetAllSnapshots()
assert.Equal(t, 3, len(snapshots))
}
func TestGetLastSnapshot(t *testing.T) {
tmpDir := t.TempDir()
sm := NewSnapshotManager(tmpDir)
sm.CreateSnapshot("wf-1", "stage1", nil, nil, nil, "", "", nil, nil, nil)
time.Sleep(10 * time.Millisecond)
sm.CreateSnapshot("wf-2", "stage2", nil, nil, nil, "", "", nil, nil, nil)
lastSnapshot := sm.GetLastSnapshot()
assert.Equal(t, "wf-2", lastSnapshot.WorkflowID)
assert.Equal(t, "stage2", lastSnapshot.Stage)
}
func TestGetSnapshotStats(t *testing.T) {
tmpDir := t.TempDir()
sm := NewSnapshotManager(tmpDir)
sm.CreateSnapshot("wf-1", "stage1", nil, nil, nil, "", "", nil, nil, nil)
sm.CreateSnapshot("wf-2", "stage1", nil, nil, nil, "", "", nil, nil, nil)
sm.MarkResumed("wf-1")
stats := sm.GetSnapshotStats()
assert.Equal(t, 2, stats["total"])
assert.Equal(t, 1, stats["paused"])
assert.Equal(t, 1, stats["resumed"])
}
func TestClearOldSnapshots(t *testing.T) {
tmpDir := t.TempDir()
sm := NewSnapshotManager(tmpDir)
sm.CreateSnapshot("wf-1", "stage1", nil, nil, nil, "", "", nil, nil, nil)
// Mark as old
snapshot := sm.GetLatestSnapshot("wf-1")
snapshot.PausedAt = time.Now().Add(-2 * time.Hour)
cleared, err := sm.ClearOldSnapshots(1 * time.Hour)
assert.NoError(t, err)
assert.Equal(t, 1, cleared)
assert.False(t, sm.HasSnapshot("wf-1"))
}
func TestSnapshotPersistence(t *testing.T) {
tmpDir := t.TempDir()
sm1 := NewSnapshotManager(tmpDir)
sm1.CreateSnapshot("wf-1", "stage1", []string{"T1.1"}, []string{"T1.2"}, nil, "T1.2", "", nil, nil, nil)
// Create new manager and load
sm2 := NewSnapshotManager(tmpDir)
err := sm2.Load()
assert.NoError(t, err)
snapshot := sm2.GetLatestSnapshot("wf-1")
assert.NotNil(t, snapshot)
assert.Equal(t, "wf-1", snapshot.WorkflowID)
assert.Equal(t, "stage1", snapshot.Stage)
}
func TestSnapshotMetrics(t *testing.T) {
tmpDir := t.TempDir()
sm := NewSnapshotManager(tmpDir)
metrics := map[string]interface{}{
"duration": 42.5,
"count": 10,
}
snapshot, err := sm.CreateSnapshot("wf-1", "stage1", nil, nil, nil, "", "", metrics, nil, nil)
assert.NoError(t, err)
assert.NotNil(t, snapshot.TaskMetrics["duration"])
assert.Equal(t, 42.5, snapshot.TaskMetrics["duration"])
}
func TestSnapshotConfiguration(t *testing.T) {
tmpDir := t.TempDir()
sm := NewSnapshotManager(tmpDir)
config := map[string]interface{}{
"timeout": 600,
"retries": 3,
}
snapshot, err := sm.CreateSnapshot("wf-1", "stage1", nil, nil, nil, "", "", nil, nil, config)
assert.NoError(t, err)
assert.Equal(t, 600, snapshot.Configuration["timeout"])
}
func TestLoadNoSnapshots(t *testing.T) {
tmpDir := t.TempDir()
sm := NewSnapshotManager(tmpDir)
err := sm.Load()
assert.NoError(t, err)
assert.Equal(t, 0, len(sm.GetAllSnapshots()))
}
func TestMultipleWorkflows(t *testing.T) {
tmpDir := t.TempDir()
sm := NewSnapshotManager(tmpDir)
for i := 1; i <= 5; i++ {
wfID := fmt.Sprintf("wf-%d", i)
sm.CreateSnapshot(wfID, "stage1", nil, nil, nil, "", "", nil, nil, nil)
}
snapshots := sm.GetAllSnapshots()
assert.Equal(t, 5, len(snapshots))
}
func TestSnapshotTimestamps(t *testing.T) {
tmpDir := t.TempDir()
sm := NewSnapshotManager(tmpDir)
before := time.Now()
sm.CreateSnapshot("wf-1", "stage1", nil, nil, nil, "", "", nil, nil, nil)
after := time.Now()
snapshot := sm.GetLatestSnapshot("wf-1")
assert.True(t, snapshot.Timestamp.After(before) || snapshot.Timestamp.Equal(before))
assert.True(t, snapshot.Timestamp.Before(after) || snapshot.Timestamp.Equal(after))
}