454 lines
10 KiB
Go
454 lines
10 KiB
Go
|
package tests
|
||
|
|
|
||
|
|
import (
|
||
|
|
"fmt"
|
||
|
|
"sync"
|
||
|
|
"testing"
|
||
|
|
"time"
|
||
|
|
|
||
|
|
"github.com/stretchr/testify/assert"
|
||
|
|
"github.com/rockliang/poimen/workflows/internal/board"
|
||
|
|
"github.com/rockliang/poimen/workflows/internal/pause"
|
||
|
|
"github.com/rockliang/poimen/workflows/internal/recovery"
|
||
|
|
)
|
||
|
|
|
||
|
|
// TestConcurrentWorkflows tests multiple workflows executing concurrently
|
||
|
|
func TestConcurrentWorkflows(t *testing.T) {
|
||
|
|
tmpDir := t.TempDir()
|
||
|
|
numWorkflows := 5
|
||
|
|
|
||
|
|
// Initialize shared managers
|
||
|
|
snapshotMgr := pause.NewSnapshotManager(tmpDir)
|
||
|
|
pauseHandler := pause.NewPauseHandler(snapshotMgr)
|
||
|
|
stateTracker := board.NewStateTracker(tmpDir)
|
||
|
|
|
||
|
|
var wg sync.WaitGroup
|
||
|
|
errors := make(chan error, numWorkflows)
|
||
|
|
|
||
|
|
// Launch concurrent workflows
|
||
|
|
for i := 1; i <= numWorkflows; i++ {
|
||
|
|
wg.Add(1)
|
||
|
|
go func(id int) {
|
||
|
|
defer wg.Done()
|
||
|
|
|
||
|
|
workflowID := fmt.Sprintf("wf-%d", id)
|
||
|
|
|
||
|
|
// Save snapshot
|
||
|
|
_, err := pauseHandler.SaveSnapshot(
|
||
|
|
workflowID,
|
||
|
|
"implement",
|
||
|
|
[]string{fmt.Sprintf("T%d.1", id)},
|
||
|
|
[]string{fmt.Sprintf("T%d.2", id)},
|
||
|
|
nil,
|
||
|
|
fmt.Sprintf("T%d.2", id),
|
||
|
|
"activity-1",
|
||
|
|
nil,
|
||
|
|
nil,
|
||
|
|
nil,
|
||
|
|
)
|
||
|
|
if err != nil {
|
||
|
|
errors <- fmt.Errorf("wf-%d: snapshot failed: %v", id, err)
|
||
|
|
return
|
||
|
|
}
|
||
|
|
|
||
|
|
// Update state
|
||
|
|
err = stateTracker.UpdateTaskState(fmt.Sprintf("T%d.1", id), "completed", "branch", nil)
|
||
|
|
if err != nil {
|
||
|
|
errors <- fmt.Errorf("wf-%d: state update failed: %v", id, err)
|
||
|
|
return
|
||
|
|
}
|
||
|
|
|
||
|
|
// Pause and resume
|
||
|
|
err = pauseHandler.RequestPause(&pause.PauseSignal{
|
||
|
|
WorkflowID: workflowID,
|
||
|
|
Reason: "test pause",
|
||
|
|
RequestedAt: time.Now(),
|
||
|
|
})
|
||
|
|
if err != nil {
|
||
|
|
errors <- fmt.Errorf("wf-%d: pause failed: %v", id, err)
|
||
|
|
return
|
||
|
|
}
|
||
|
|
|
||
|
|
err = pauseHandler.RequestResume(&pause.ResumeSignal{
|
||
|
|
WorkflowID: workflowID,
|
||
|
|
Reason: "test resume",
|
||
|
|
RequestedAt: time.Now(),
|
||
|
|
})
|
||
|
|
if err != nil {
|
||
|
|
errors <- fmt.Errorf("wf-%d: resume failed: %v", id, err)
|
||
|
|
return
|
||
|
|
}
|
||
|
|
}(i)
|
||
|
|
}
|
||
|
|
|
||
|
|
wg.Wait()
|
||
|
|
close(errors)
|
||
|
|
|
||
|
|
// Check for errors
|
||
|
|
for err := range errors {
|
||
|
|
assert.NoError(t, err)
|
||
|
|
}
|
||
|
|
|
||
|
|
// Verify all workflows were tracked
|
||
|
|
states := pauseHandler.GetAllPauseStates()
|
||
|
|
assert.Equal(t, numWorkflows, len(states))
|
||
|
|
}
|
||
|
|
|
||
|
|
// TestConcurrentBoardOperations tests concurrent board validation and healing
|
||
|
|
func TestConcurrentBoardOperations(t *testing.T) {
|
||
|
|
boardContent := `# Task Board — Milestone T1: Production Hardening
|
||
|
|
|
||
|
|
**Submilestone:** T1 (Error recovery, observability, metrics, reliability)
|
||
|
|
|
||
|
|
| ID | Scope | Status | Branch | Verification |
|
||
|
|
|----|-------|--------|--------|--------------|
|
||
|
|
| T1.1 | Task 1 | [x] | task/T1.1 | Verify recovery works |
|
||
|
|
| T1.2 | Task 2 | [x] | task/T1.2 | Verify metrics visible |
|
||
|
|
| T1.3 | Task 3 | [ ] | task/T1.3 | Verify recommendations |
|
||
|
|
| T1.4 | Task 4 | [ ] | task/T1.4 | Verify healing works |
|
||
|
|
`
|
||
|
|
|
||
|
|
validator := board.NewBoardValidator("")
|
||
|
|
numValidations := 10
|
||
|
|
|
||
|
|
var wg sync.WaitGroup
|
||
|
|
errors := make(chan error, numValidations)
|
||
|
|
|
||
|
|
for i := 0; i < numValidations; i++ {
|
||
|
|
wg.Add(1)
|
||
|
|
go func(id int) {
|
||
|
|
defer wg.Done()
|
||
|
|
|
||
|
|
// Validate
|
||
|
|
if !validator.ValidateBoard(boardContent) {
|
||
|
|
errors <- fmt.Errorf("validation %d failed", id)
|
||
|
|
return
|
||
|
|
}
|
||
|
|
|
||
|
|
// Parse
|
||
|
|
tasks, err := validator.ParseTasks(boardContent)
|
||
|
|
if err != nil {
|
||
|
|
errors <- fmt.Errorf("parse %d failed: %v", id, err)
|
||
|
|
return
|
||
|
|
}
|
||
|
|
|
||
|
|
if len(tasks) != 4 {
|
||
|
|
errors <- fmt.Errorf("validation %d: expected 4 tasks, got %d", id, len(tasks))
|
||
|
|
return
|
||
|
|
}
|
||
|
|
}(i)
|
||
|
|
}
|
||
|
|
|
||
|
|
wg.Wait()
|
||
|
|
close(errors)
|
||
|
|
|
||
|
|
for err := range errors {
|
||
|
|
assert.NoError(t, err)
|
||
|
|
}
|
||
|
|
}
|
||
|
|
|
||
|
|
// TestConcurrentStateTracking tests concurrent state updates
|
||
|
|
func TestConcurrentStateTracking(t *testing.T) {
|
||
|
|
tmpDir := t.TempDir()
|
||
|
|
tracker := board.NewStateTracker(tmpDir)
|
||
|
|
numTasks := 20
|
||
|
|
var wg sync.WaitGroup
|
||
|
|
|
||
|
|
// Concurrent state updates
|
||
|
|
for i := 1; i <= numTasks; i++ {
|
||
|
|
wg.Add(1)
|
||
|
|
go func(id int) {
|
||
|
|
defer wg.Done()
|
||
|
|
|
||
|
|
taskID := fmt.Sprintf("T%d", id)
|
||
|
|
_ = tracker.UpdateTaskState(taskID, "in_progress", "branch", nil)
|
||
|
|
|
||
|
|
time.Sleep(time.Duration(id%5) * time.Millisecond)
|
||
|
|
|
||
|
|
_ = tracker.UpdateTaskState(taskID, "completed", "branch", nil)
|
||
|
|
}(i)
|
||
|
|
}
|
||
|
|
|
||
|
|
wg.Wait()
|
||
|
|
|
||
|
|
completed := tracker.GetCompletedTasks()
|
||
|
|
assert.Equal(t, numTasks, len(completed))
|
||
|
|
}
|
||
|
|
|
||
|
|
// TestConcurrentSnapshotCreation tests concurrent snapshot creation and restoration
|
||
|
|
func TestConcurrentSnapshotCreation(t *testing.T) {
|
||
|
|
tmpDir := t.TempDir()
|
||
|
|
snapMgr := pause.NewSnapshotManager(tmpDir)
|
||
|
|
numSnapshots := 10
|
||
|
|
var wg sync.WaitGroup
|
||
|
|
|
||
|
|
// Create snapshots concurrently
|
||
|
|
for i := 1; i <= numSnapshots; i++ {
|
||
|
|
wg.Add(1)
|
||
|
|
go func(id int) {
|
||
|
|
defer wg.Done()
|
||
|
|
|
||
|
|
workflowID := fmt.Sprintf("wf-%d", id)
|
||
|
|
_, _ = snapMgr.CreateSnapshot(
|
||
|
|
workflowID,
|
||
|
|
"stage",
|
||
|
|
[]string{},
|
||
|
|
[]string{},
|
||
|
|
nil,
|
||
|
|
"",
|
||
|
|
"",
|
||
|
|
nil,
|
||
|
|
nil,
|
||
|
|
nil,
|
||
|
|
)
|
||
|
|
}(i)
|
||
|
|
}
|
||
|
|
|
||
|
|
wg.Wait()
|
||
|
|
|
||
|
|
// Restore snapshots
|
||
|
|
for i := 1; i <= numSnapshots; i++ {
|
||
|
|
wg.Add(1)
|
||
|
|
go func(id int) {
|
||
|
|
defer wg.Done()
|
||
|
|
|
||
|
|
workflowID := fmt.Sprintf("wf-%d", id)
|
||
|
|
snapshot, err := snapMgr.RestoreFromSnapshot(workflowID)
|
||
|
|
assert.NoError(t, err)
|
||
|
|
assert.NotNil(t, snapshot)
|
||
|
|
}(i)
|
||
|
|
}
|
||
|
|
|
||
|
|
wg.Wait()
|
||
|
|
}
|
||
|
|
|
||
|
|
// TestRecoveryWithConcurrency tests retry policies under concurrent load
|
||
|
|
func TestRecoveryWithConcurrency(t *testing.T) {
|
||
|
|
retryPolicy := recovery.ActivityRetryPolicy()
|
||
|
|
assert.NotNil(t, retryPolicy)
|
||
|
|
|
||
|
|
numAttempts := 20
|
||
|
|
var wg sync.WaitGroup
|
||
|
|
|
||
|
|
for i := 0; i < numAttempts; i++ {
|
||
|
|
wg.Add(1)
|
||
|
|
go func(id int) {
|
||
|
|
defer wg.Done()
|
||
|
|
|
||
|
|
rc := recovery.RetryCount{Current: 0, Maximum: 3}
|
||
|
|
for rc.CanRetry() {
|
||
|
|
rc.Increment()
|
||
|
|
time.Sleep(time.Millisecond)
|
||
|
|
}
|
||
|
|
assert.Equal(t, 3, rc.Current)
|
||
|
|
}(i)
|
||
|
|
}
|
||
|
|
|
||
|
|
wg.Wait()
|
||
|
|
}
|
||
|
|
|
||
|
|
// TestIntegrationHealthCheck tests health checks under concurrent operations
|
||
|
|
func TestIntegrationHealthCheck(t *testing.T) {
|
||
|
|
tmpDir := t.TempDir()
|
||
|
|
|
||
|
|
// Simulate concurrent operations with health checks
|
||
|
|
var wg sync.WaitGroup
|
||
|
|
numConcurrent := 5
|
||
|
|
|
||
|
|
for i := 0; i < numConcurrent; i++ {
|
||
|
|
wg.Add(1)
|
||
|
|
go func(id int) {
|
||
|
|
defer wg.Done()
|
||
|
|
|
||
|
|
// Simulate workflow with state changes
|
||
|
|
stateTracker := board.NewStateTracker(tmpDir)
|
||
|
|
_ = stateTracker.UpdateTaskState("T1", "in_progress", "branch", nil)
|
||
|
|
|
||
|
|
stats := stateTracker.GetStats()
|
||
|
|
assert.Equal(t, 1, stats["total"])
|
||
|
|
|
||
|
|
_ = stateTracker.UpdateTaskState("T1", "completed", "branch", nil)
|
||
|
|
}(i)
|
||
|
|
}
|
||
|
|
|
||
|
|
wg.Wait()
|
||
|
|
}
|
||
|
|
|
||
|
|
// TestPauseResumeUnderLoad tests pause/resume with concurrent state changes
|
||
|
|
func TestPauseResumeUnderLoad(t *testing.T) {
|
||
|
|
tmpDir := t.TempDir()
|
||
|
|
pauseMgr := pause.NewSnapshotManager(tmpDir)
|
||
|
|
pauseHandler := pause.NewPauseHandler(pauseMgr)
|
||
|
|
|
||
|
|
numWorkflows := 10
|
||
|
|
var wg sync.WaitGroup
|
||
|
|
|
||
|
|
// Start workflows and pause them concurrently
|
||
|
|
for i := 1; i <= numWorkflows; i++ {
|
||
|
|
wg.Add(1)
|
||
|
|
go func(id int) {
|
||
|
|
defer wg.Done()
|
||
|
|
|
||
|
|
workflowID := fmt.Sprintf("wf-%d", id)
|
||
|
|
|
||
|
|
// Save snapshot
|
||
|
|
_, _ = pauseHandler.SaveSnapshot(
|
||
|
|
workflowID,
|
||
|
|
"stage",
|
||
|
|
[]string{},
|
||
|
|
[]string{},
|
||
|
|
nil,
|
||
|
|
"",
|
||
|
|
"",
|
||
|
|
nil,
|
||
|
|
nil,
|
||
|
|
nil,
|
||
|
|
)
|
||
|
|
|
||
|
|
// Pause
|
||
|
|
_ = pauseHandler.RequestPause(&pause.PauseSignal{
|
||
|
|
WorkflowID: workflowID,
|
||
|
|
Reason: "load test",
|
||
|
|
RequestedAt: time.Now(),
|
||
|
|
})
|
||
|
|
|
||
|
|
// Small delay to simulate work
|
||
|
|
time.Sleep(time.Duration(id%3) * time.Millisecond)
|
||
|
|
|
||
|
|
// Resume
|
||
|
|
_ = pauseHandler.RequestResume(&pause.ResumeSignal{
|
||
|
|
WorkflowID: workflowID,
|
||
|
|
Reason: "load test resume",
|
||
|
|
RequestedAt: time.Now(),
|
||
|
|
})
|
||
|
|
}(i)
|
||
|
|
}
|
||
|
|
|
||
|
|
wg.Wait()
|
||
|
|
|
||
|
|
// Verify all workflows
|
||
|
|
stats := pauseHandler.GetPauseStats()
|
||
|
|
assert.Equal(t, numWorkflows, stats["total"])
|
||
|
|
}
|
||
|
|
|
||
|
|
// TestDataConsistencyUnderConcurrency ensures data consistency with concurrent access
|
||
|
|
func TestDataConsistencyUnderConcurrency(t *testing.T) {
|
||
|
|
tmpDir := t.TempDir()
|
||
|
|
tracker := board.NewStateTracker(tmpDir)
|
||
|
|
|
||
|
|
const numGoroutines = 20
|
||
|
|
const operationsPerGoroutine = 10
|
||
|
|
|
||
|
|
var wg sync.WaitGroup
|
||
|
|
|
||
|
|
// Concurrent reads and writes
|
||
|
|
for g := 0; g < numGoroutines; g++ {
|
||
|
|
wg.Add(1)
|
||
|
|
go func() {
|
||
|
|
defer wg.Done()
|
||
|
|
|
||
|
|
for op := 0; op < operationsPerGoroutine; op++ {
|
||
|
|
taskID := fmt.Sprintf("T%d", op%5)
|
||
|
|
|
||
|
|
if op%2 == 0 {
|
||
|
|
// Write
|
||
|
|
_ = tracker.UpdateTaskState(taskID, "in_progress", "branch", nil)
|
||
|
|
} else {
|
||
|
|
// Read
|
||
|
|
_ = tracker.GetTaskState(taskID)
|
||
|
|
}
|
||
|
|
}
|
||
|
|
}()
|
||
|
|
}
|
||
|
|
|
||
|
|
wg.Wait()
|
||
|
|
|
||
|
|
// Verify final state is consistent
|
||
|
|
allStates := tracker.GetAllStates()
|
||
|
|
assert.Greater(t, len(allStates), 0)
|
||
|
|
}
|
||
|
|
|
||
|
|
// TestNetworkFlakinessSim simulates network issues with retries
|
||
|
|
func TestNetworkFlakinessSim(t *testing.T) {
|
||
|
|
retryPolicy := recovery.ActivityRetryPolicy()
|
||
|
|
numAttempts := 0
|
||
|
|
maxAttempts := retryPolicy.MaximumAttempts
|
||
|
|
|
||
|
|
// Simulate retryable errors
|
||
|
|
for numAttempts < int(maxAttempts) {
|
||
|
|
numAttempts++
|
||
|
|
time.Sleep(1 * time.Millisecond)
|
||
|
|
}
|
||
|
|
|
||
|
|
assert.Equal(t, 3, numAttempts)
|
||
|
|
}
|
||
|
|
|
||
|
|
// TestCrossWorkflowIsolation ensures workflows don't interfere with each other
|
||
|
|
func TestCrossWorkflowIsolation(t *testing.T) {
|
||
|
|
tmpDir := t.TempDir()
|
||
|
|
wf1Handler := pause.NewPauseHandler(pause.NewSnapshotManager(tmpDir))
|
||
|
|
wf2Handler := pause.NewPauseHandler(pause.NewSnapshotManager(tmpDir))
|
||
|
|
|
||
|
|
// Workflow 1
|
||
|
|
_ = wf1Handler.RequestPause(&pause.PauseSignal{
|
||
|
|
WorkflowID: "wf-1",
|
||
|
|
Reason: "test",
|
||
|
|
RequestedAt: time.Now(),
|
||
|
|
})
|
||
|
|
|
||
|
|
// Workflow 2 should not be affected
|
||
|
|
assert.False(t, wf2Handler.IsPaused("wf-1"))
|
||
|
|
assert.False(t, wf2Handler.IsPaused("wf-2"))
|
||
|
|
|
||
|
|
_ = wf2Handler.RequestPause(&pause.PauseSignal{
|
||
|
|
WorkflowID: "wf-2",
|
||
|
|
Reason: "test",
|
||
|
|
RequestedAt: time.Now(),
|
||
|
|
})
|
||
|
|
|
||
|
|
// Both should be paused independently
|
||
|
|
assert.True(t, wf1Handler.IsPaused("wf-1"))
|
||
|
|
assert.True(t, wf2Handler.IsPaused("wf-2"))
|
||
|
|
}
|
||
|
|
|
||
|
|
// BenchmarkConcurrentSnapshot benchmarks concurrent snapshot creation
|
||
|
|
func BenchmarkConcurrentSnapshot(b *testing.B) {
|
||
|
|
tmpDir := b.TempDir()
|
||
|
|
snapMgr := pause.NewSnapshotManager(tmpDir)
|
||
|
|
|
||
|
|
b.RunParallel(func(pb *testing.PB) {
|
||
|
|
i := 0
|
||
|
|
for pb.Next() {
|
||
|
|
workflowID := fmt.Sprintf("wf-bench-%d", i%100)
|
||
|
|
_, _ = snapMgr.CreateSnapshot(
|
||
|
|
workflowID,
|
||
|
|
"stage",
|
||
|
|
nil,
|
||
|
|
nil,
|
||
|
|
nil,
|
||
|
|
"",
|
||
|
|
"",
|
||
|
|
nil,
|
||
|
|
nil,
|
||
|
|
nil,
|
||
|
|
)
|
||
|
|
i++
|
||
|
|
}
|
||
|
|
})
|
||
|
|
}
|
||
|
|
|
||
|
|
// BenchmarkConcurrentStateUpdate benchmarks concurrent state updates
|
||
|
|
func BenchmarkConcurrentStateUpdate(b *testing.B) {
|
||
|
|
tmpDir := b.TempDir()
|
||
|
|
tracker := board.NewStateTracker(tmpDir)
|
||
|
|
|
||
|
|
b.RunParallel(func(pb *testing.PB) {
|
||
|
|
i := 0
|
||
|
|
for pb.Next() {
|
||
|
|
taskID := fmt.Sprintf("T%d", i%50)
|
||
|
|
_ = tracker.UpdateTaskState(taskID, "completed", "branch", nil)
|
||
|
|
i++
|
||
|
|
}
|
||
|
|
})
|
||
|
|
}
|