472 lines
12 KiB
Go
472 lines
12 KiB
Go
|
package backgroundtask
|
||
|
|
|
||
|
|
import (
|
||
|
|
"context"
|
||
|
|
"log/slog"
|
||
|
|
"sync"
|
||
|
|
"sync/atomic"
|
||
|
|
"testing"
|
||
|
|
"time"
|
||
|
|
)
|
||
|
|
|
||
|
|
// TestSignal_TriggersWork verifies that Signal() triggers work execution
|
||
|
|
func TestSignal_TriggersWork(t *testing.T) {
|
||
|
|
config := NewTestConfigMap()
|
||
|
|
testConfig, _ := GetTestConfig(config)
|
||
|
|
|
||
|
|
runner, err := Initialize(
|
||
|
|
config,
|
||
|
|
slog.Default(),
|
||
|
|
TestWorkFunc,
|
||
|
|
WithInterval(1*time.Hour), // Very long interval to ensure only signal triggers work
|
||
|
|
)
|
||
|
|
if err != nil {
|
||
|
|
t.Fatalf("Failed to initialize runner: %v", err)
|
||
|
|
}
|
||
|
|
|
||
|
|
if err := runner.Run(); err != nil {
|
||
|
|
t.Fatalf("Failed to start runner: %v", err)
|
||
|
|
}
|
||
|
|
defer func() {
|
||
|
|
ctx, cancel := context.WithTimeout(t.Context(), 2*time.Second)
|
||
|
|
defer cancel()
|
||
|
|
if err := runner.Shutdown(ctx); err != nil {
|
||
|
|
t.Errorf("Shutdown error: %v", err)
|
||
|
|
}
|
||
|
|
}()
|
||
|
|
|
||
|
|
// Signal work
|
||
|
|
if err := runner.Signal(); err != nil {
|
||
|
|
t.Fatalf("Signal failed: %v", err)
|
||
|
|
}
|
||
|
|
|
||
|
|
// Wait for work to complete
|
||
|
|
if !WaitForWorkCycles(runner, 1, 500*time.Millisecond) {
|
||
|
|
t.Fatal("Expected at least 1 work cycle after signal")
|
||
|
|
}
|
||
|
|
|
||
|
|
stats := runner.Stats()
|
||
|
|
if stats.WorkCycles() < 1 {
|
||
|
|
t.Errorf("Expected at least 1 work cycle, got %d", stats.WorkCycles())
|
||
|
|
}
|
||
|
|
|
||
|
|
if testConfig.Counter.Load() < 1 {
|
||
|
|
t.Errorf("Expected at least 1 counter increment, got %d", testConfig.Counter.Load())
|
||
|
|
}
|
||
|
|
}
|
||
|
|
|
||
|
|
// TestTicker_TriggersWork verifies that the ticker triggers work execution
|
||
|
|
func TestTicker_TriggersWork(t *testing.T) {
|
||
|
|
config := NewTestConfigMap()
|
||
|
|
testConfig, _ := GetTestConfig(config)
|
||
|
|
|
||
|
|
runner, err := Initialize(
|
||
|
|
config,
|
||
|
|
slog.Default(),
|
||
|
|
TestWorkFunc,
|
||
|
|
WithInterval(50*time.Millisecond), // Fast interval for testing
|
||
|
|
)
|
||
|
|
if err != nil {
|
||
|
|
t.Fatalf("Failed to initialize runner: %v", err)
|
||
|
|
}
|
||
|
|
|
||
|
|
if err := runner.Run(); err != nil {
|
||
|
|
t.Fatalf("Failed to start runner: %v", err)
|
||
|
|
}
|
||
|
|
defer func() {
|
||
|
|
ctx, cancel := context.WithTimeout(t.Context(), 2*time.Second)
|
||
|
|
defer cancel()
|
||
|
|
if err := runner.Shutdown(ctx); err != nil {
|
||
|
|
t.Errorf("Shutdown error: %v", err)
|
||
|
|
}
|
||
|
|
}()
|
||
|
|
|
||
|
|
// Wait for ticker-triggered work (should happen at least twice)
|
||
|
|
if !WaitForWorkCycles(runner, 2, 200*time.Millisecond) {
|
||
|
|
t.Fatal("Expected at least 2 work cycles from ticker")
|
||
|
|
}
|
||
|
|
|
||
|
|
stats := runner.Stats()
|
||
|
|
if stats.WorkCycles() < 2 {
|
||
|
|
t.Errorf("Expected at least 2 work cycles, got %d", stats.WorkCycles())
|
||
|
|
}
|
||
|
|
|
||
|
|
if testConfig.Counter.Load() < 2 {
|
||
|
|
t.Errorf("Expected at least 2 counter increments, got %d", testConfig.Counter.Load())
|
||
|
|
}
|
||
|
|
}
|
||
|
|
|
||
|
|
// TestSignal_AfterShutdown verifies that Signal() returns error after shutdown
|
||
|
|
func TestSignal_AfterShutdown(t *testing.T) {
|
||
|
|
config := NewTestConfigMap()
|
||
|
|
|
||
|
|
runner, err := Initialize(
|
||
|
|
config,
|
||
|
|
slog.Default(),
|
||
|
|
TestWorkFunc,
|
||
|
|
WithInterval(100*time.Millisecond),
|
||
|
|
)
|
||
|
|
if err != nil {
|
||
|
|
t.Fatalf("Failed to initialize runner: %v", err)
|
||
|
|
}
|
||
|
|
|
||
|
|
if err := runner.Run(); err != nil {
|
||
|
|
t.Fatalf("Failed to start runner: %v", err)
|
||
|
|
}
|
||
|
|
|
||
|
|
// Shutdown the runner
|
||
|
|
ctx, cancel := context.WithTimeout(t.Context(), 2*time.Second)
|
||
|
|
defer cancel()
|
||
|
|
if err := runner.Shutdown(ctx); err != nil {
|
||
|
|
t.Fatalf("Shutdown failed: %v", err)
|
||
|
|
}
|
||
|
|
|
||
|
|
// Signal should return ErrStopped
|
||
|
|
if err := runner.Signal(); err != ErrStopped {
|
||
|
|
t.Errorf("Expected ErrStopped after shutdown, got: %v", err)
|
||
|
|
}
|
||
|
|
}
|
||
|
|
|
||
|
|
// TestSignal_Coalescing verifies that multiple rapid signals are coalesced
|
||
|
|
func TestSignal_Coalescing(t *testing.T) {
|
||
|
|
config := NewTestConfigMap()
|
||
|
|
testConfig, _ := GetTestConfig(config)
|
||
|
|
testConfig.WorkDuration = 100 * time.Millisecond // Make work take some time
|
||
|
|
|
||
|
|
runner, err := Initialize(
|
||
|
|
config,
|
||
|
|
slog.Default(),
|
||
|
|
TestWorkFunc,
|
||
|
|
WithInterval(1*time.Hour), // Very long interval
|
||
|
|
)
|
||
|
|
if err != nil {
|
||
|
|
t.Fatalf("Failed to initialize runner: %v", err)
|
||
|
|
}
|
||
|
|
|
||
|
|
if err := runner.Run(); err != nil {
|
||
|
|
t.Fatalf("Failed to start runner: %v", err)
|
||
|
|
}
|
||
|
|
defer func() {
|
||
|
|
ctx, cancel := context.WithTimeout(t.Context(), 5*time.Second)
|
||
|
|
defer cancel()
|
||
|
|
if err := runner.Shutdown(ctx); err != nil {
|
||
|
|
t.Errorf("Shutdown error: %v", err)
|
||
|
|
}
|
||
|
|
}()
|
||
|
|
|
||
|
|
// Send many rapid signals
|
||
|
|
const signalCount = 20
|
||
|
|
for i := 0; i < signalCount; i++ {
|
||
|
|
if err := runner.Signal(); err != nil {
|
||
|
|
t.Fatalf("Signal %d failed: %v", i, err)
|
||
|
|
}
|
||
|
|
}
|
||
|
|
|
||
|
|
// Wait for work to complete
|
||
|
|
time.Sleep(500 * time.Millisecond)
|
||
|
|
|
||
|
|
// Should have fewer work cycles than signals due to coalescing
|
||
|
|
stats := runner.Stats()
|
||
|
|
workCycles := stats.WorkCycles()
|
||
|
|
|
||
|
|
if workCycles == 0 {
|
||
|
|
t.Fatal("Expected some work cycles")
|
||
|
|
}
|
||
|
|
|
||
|
|
// Should be significantly less than signalCount due to coalescing
|
||
|
|
if workCycles >= signalCount {
|
||
|
|
t.Errorf("Expected coalescing: work cycles (%d) should be less than signals (%d)", workCycles, signalCount)
|
||
|
|
}
|
||
|
|
|
||
|
|
// But should have at least processed some work
|
||
|
|
if testConfig.Counter.Load() == 0 {
|
||
|
|
t.Error("Expected some work to be done")
|
||
|
|
}
|
||
|
|
}
|
||
|
|
|
||
|
|
// TestWork_NoConcurrency verifies that work functions don't run concurrently
|
||
|
|
func TestWork_NoConcurrency(t *testing.T) {
|
||
|
|
var concurrentCount int64
|
||
|
|
var maxConcurrent int64
|
||
|
|
|
||
|
|
// Create a work function that tracks concurrent executions
|
||
|
|
workFunc := func(ctx context.Context, logger *slog.Logger, config map[string]any) error {
|
||
|
|
current := atomic.AddInt64(&concurrentCount, 1)
|
||
|
|
defer atomic.AddInt64(&concurrentCount, -1)
|
||
|
|
|
||
|
|
// Track max concurrent
|
||
|
|
for {
|
||
|
|
max := atomic.LoadInt64(&maxConcurrent)
|
||
|
|
if current <= max || atomic.CompareAndSwapInt64(&maxConcurrent, max, current) {
|
||
|
|
break
|
||
|
|
}
|
||
|
|
}
|
||
|
|
|
||
|
|
// Simulate some work
|
||
|
|
time.Sleep(50 * time.Millisecond)
|
||
|
|
return nil
|
||
|
|
}
|
||
|
|
|
||
|
|
config := map[string]any{"test": true}
|
||
|
|
|
||
|
|
runner, err := Initialize(
|
||
|
|
config,
|
||
|
|
slog.Default(),
|
||
|
|
workFunc,
|
||
|
|
WithInterval(25*time.Millisecond), // Faster than work duration
|
||
|
|
)
|
||
|
|
if err != nil {
|
||
|
|
t.Fatalf("Failed to initialize runner: %v", err)
|
||
|
|
}
|
||
|
|
|
||
|
|
if err := runner.Run(); err != nil {
|
||
|
|
t.Fatalf("Failed to start runner: %v", err)
|
||
|
|
}
|
||
|
|
|
||
|
|
// Let it run for a while to trigger multiple potential concurrent executions
|
||
|
|
time.Sleep(200 * time.Millisecond)
|
||
|
|
|
||
|
|
// Send some signals too
|
||
|
|
for i := 0; i < 5; i++ {
|
||
|
|
if err := runner.Signal(); err != nil {
|
||
|
|
t.Fatalf("Signal failed: %v", err)
|
||
|
|
}
|
||
|
|
}
|
||
|
|
|
||
|
|
time.Sleep(200 * time.Millisecond)
|
||
|
|
|
||
|
|
ctx, cancel := context.WithTimeout(t.Context(), 2*time.Second)
|
||
|
|
defer cancel()
|
||
|
|
if err := runner.Shutdown(ctx); err != nil {
|
||
|
|
t.Fatalf("Shutdown failed: %v", err)
|
||
|
|
}
|
||
|
|
|
||
|
|
// Should never have more than 1 concurrent execution
|
||
|
|
if maxConcurrent > 1 {
|
||
|
|
t.Errorf("Expected max concurrent executions to be 1, got %d", maxConcurrent)
|
||
|
|
}
|
||
|
|
|
||
|
|
// Should have done some work
|
||
|
|
stats := runner.Stats()
|
||
|
|
if stats.WorkCycles() == 0 {
|
||
|
|
t.Error("Expected some work cycles")
|
||
|
|
}
|
||
|
|
}
|
||
|
|
|
||
|
|
// TestWork_PendingWork verifies that pending work is executed after current work completes
|
||
|
|
func TestWork_PendingWork(t *testing.T) {
|
||
|
|
config := NewTestConfigMap()
|
||
|
|
testConfig, _ := GetTestConfig(config)
|
||
|
|
testConfig.WorkDuration = 100 * time.Millisecond // Make work take time
|
||
|
|
|
||
|
|
runner, err := Initialize(
|
||
|
|
config,
|
||
|
|
slog.Default(),
|
||
|
|
TestWorkFunc,
|
||
|
|
WithInterval(1*time.Hour), // Very long interval
|
||
|
|
)
|
||
|
|
if err != nil {
|
||
|
|
t.Fatalf("Failed to initialize runner: %v", err)
|
||
|
|
}
|
||
|
|
|
||
|
|
if err := runner.Run(); err != nil {
|
||
|
|
t.Fatalf("Failed to start runner: %v", err)
|
||
|
|
}
|
||
|
|
defer func() {
|
||
|
|
ctx, cancel := context.WithTimeout(t.Context(), 2*time.Second)
|
||
|
|
defer cancel()
|
||
|
|
if err := runner.Shutdown(ctx); err != nil {
|
||
|
|
t.Errorf("Shutdown error: %v", err)
|
||
|
|
}
|
||
|
|
}()
|
||
|
|
|
||
|
|
// Send first signal to start work
|
||
|
|
if err := runner.Signal(); err != nil {
|
||
|
|
t.Fatalf("First signal failed: %v", err)
|
||
|
|
}
|
||
|
|
|
||
|
|
// Wait a bit for work to start
|
||
|
|
time.Sleep(25 * time.Millisecond)
|
||
|
|
|
||
|
|
// Send more signals while work is in progress
|
||
|
|
for i := 0; i < 3; i++ {
|
||
|
|
if err := runner.Signal(); err != nil {
|
||
|
|
t.Fatalf("Signal %d failed: %v", i, err)
|
||
|
|
}
|
||
|
|
}
|
||
|
|
|
||
|
|
// Wait for all work to complete
|
||
|
|
time.Sleep(500 * time.Millisecond)
|
||
|
|
|
||
|
|
// Should have at least 2 work cycles (initial + pending)
|
||
|
|
stats := runner.Stats()
|
||
|
|
if stats.WorkCycles() < 2 {
|
||
|
|
t.Errorf("Expected at least 2 work cycles for pending work, got %d", stats.WorkCycles())
|
||
|
|
}
|
||
|
|
|
||
|
|
if testConfig.Counter.Load() < 2 {
|
||
|
|
t.Errorf("Expected at least 2 counter increments, got %d", testConfig.Counter.Load())
|
||
|
|
}
|
||
|
|
}
|
||
|
|
|
||
|
|
// TestWork_ErrorHandling verifies that work function errors are handled gracefully
|
||
|
|
func TestWork_ErrorHandling(t *testing.T) {
|
||
|
|
config := NewTestConfigMap()
|
||
|
|
|
||
|
|
runner, err := Initialize(
|
||
|
|
config,
|
||
|
|
slog.Default(),
|
||
|
|
ErrorWorkFunc, // Always returns error
|
||
|
|
WithInterval(1*time.Hour),
|
||
|
|
)
|
||
|
|
if err != nil {
|
||
|
|
t.Fatalf("Failed to initialize runner: %v", err)
|
||
|
|
}
|
||
|
|
|
||
|
|
if err := runner.Run(); err != nil {
|
||
|
|
t.Fatalf("Failed to start runner: %v", err)
|
||
|
|
}
|
||
|
|
defer func() {
|
||
|
|
ctx, cancel := context.WithTimeout(t.Context(), 2*time.Second)
|
||
|
|
defer cancel()
|
||
|
|
if err := runner.Shutdown(ctx); err != nil {
|
||
|
|
t.Errorf("Shutdown error: %v", err)
|
||
|
|
}
|
||
|
|
}()
|
||
|
|
|
||
|
|
// Signal work that will error
|
||
|
|
if err := runner.Signal(); err != nil {
|
||
|
|
t.Fatalf("Signal failed: %v", err)
|
||
|
|
}
|
||
|
|
|
||
|
|
// Wait for work to complete
|
||
|
|
if !WaitForWorkCycles(runner, 1, 500*time.Millisecond) {
|
||
|
|
t.Fatal("Expected at least 1 work cycle")
|
||
|
|
}
|
||
|
|
|
||
|
|
// Check that error is recorded in stats
|
||
|
|
stats := runner.Stats()
|
||
|
|
if stats.LastErr() == nil {
|
||
|
|
t.Error("Expected error to be recorded in stats")
|
||
|
|
}
|
||
|
|
|
||
|
|
// Runner should still be functional after error
|
||
|
|
if err := runner.Signal(); err != nil {
|
||
|
|
t.Fatalf("Signal after error failed: %v", err)
|
||
|
|
}
|
||
|
|
|
||
|
|
if !WaitForWorkCycles(runner, 2, 500*time.Millisecond) {
|
||
|
|
t.Fatal("Expected runner to continue working after error")
|
||
|
|
}
|
||
|
|
}
|
||
|
|
|
||
|
|
// TestWork_ContextCancellation verifies that work functions receive cancellation signals
|
||
|
|
func TestWork_ContextCancellation(t *testing.T) {
|
||
|
|
config := NewTestConfigMap()
|
||
|
|
testConfig, _ := GetTestConfig(config)
|
||
|
|
testConfig.WorkDuration = 1 * time.Second // Long duration
|
||
|
|
testConfig.RespectContext = true // Make it respect context
|
||
|
|
|
||
|
|
runner, err := Initialize(
|
||
|
|
config,
|
||
|
|
slog.Default(),
|
||
|
|
TestWorkFunc,
|
||
|
|
WithInterval(1*time.Hour),
|
||
|
|
)
|
||
|
|
if err != nil {
|
||
|
|
t.Fatalf("Failed to initialize runner: %v", err)
|
||
|
|
}
|
||
|
|
|
||
|
|
if err := runner.Run(); err != nil {
|
||
|
|
t.Fatalf("Failed to start runner: %v", err)
|
||
|
|
}
|
||
|
|
|
||
|
|
// Signal work to start
|
||
|
|
if err := runner.Signal(); err != nil {
|
||
|
|
t.Fatalf("Signal failed: %v", err)
|
||
|
|
}
|
||
|
|
|
||
|
|
// Let work start
|
||
|
|
time.Sleep(50 * time.Millisecond)
|
||
|
|
|
||
|
|
// Shutdown while work is in progress
|
||
|
|
ctx, cancel := context.WithTimeout(t.Context(), 2*time.Second)
|
||
|
|
defer cancel()
|
||
|
|
|
||
|
|
start := time.Now()
|
||
|
|
if err := runner.Shutdown(ctx); err != nil {
|
||
|
|
t.Fatalf("Shutdown failed: %v", err)
|
||
|
|
}
|
||
|
|
elapsed := time.Since(start)
|
||
|
|
|
||
|
|
// Should shutdown quickly due to context cancellation, not wait for full work duration
|
||
|
|
if elapsed > 500*time.Millisecond {
|
||
|
|
t.Errorf("Shutdown took too long (%v), context cancellation may not be working", elapsed)
|
||
|
|
}
|
||
|
|
|
||
|
|
// Work should have been attempted
|
||
|
|
stats := runner.Stats()
|
||
|
|
if stats.WorkCycles() == 0 {
|
||
|
|
t.Error("Expected at least one work cycle to start")
|
||
|
|
}
|
||
|
|
}
|
||
|
|
|
||
|
|
// TestMultipleSignalsRapidly verifies handling of many rapid signals
|
||
|
|
func TestMultipleSignalsRapidly(t *testing.T) {
|
||
|
|
config := NewTestConfigMap()
|
||
|
|
|
||
|
|
runner, err := Initialize(
|
||
|
|
config,
|
||
|
|
slog.Default(),
|
||
|
|
FastWorkFunc, // Very fast work function
|
||
|
|
WithInterval(1*time.Hour),
|
||
|
|
)
|
||
|
|
if err != nil {
|
||
|
|
t.Fatalf("Failed to initialize runner: %v", err)
|
||
|
|
}
|
||
|
|
|
||
|
|
if err := runner.Run(); err != nil {
|
||
|
|
t.Fatalf("Failed to start runner: %v", err)
|
||
|
|
}
|
||
|
|
defer func() {
|
||
|
|
ctx, cancel := context.WithTimeout(t.Context(), 2*time.Second)
|
||
|
|
defer cancel()
|
||
|
|
if err := runner.Shutdown(ctx); err != nil {
|
||
|
|
t.Errorf("Shutdown error: %v", err)
|
||
|
|
}
|
||
|
|
}()
|
||
|
|
|
||
|
|
// Send many signals from multiple goroutines
|
||
|
|
const numGoroutines = 10
|
||
|
|
const signalsPerGoroutine = 10
|
||
|
|
var wg sync.WaitGroup
|
||
|
|
|
||
|
|
for i := 0; i < numGoroutines; i++ {
|
||
|
|
wg.Add(1)
|
||
|
|
go func() {
|
||
|
|
defer wg.Done()
|
||
|
|
for j := 0; j < signalsPerGoroutine; j++ {
|
||
|
|
if err := runner.Signal(); err != nil && err != ErrStopped {
|
||
|
|
t.Errorf("Signal failed: %v", err)
|
||
|
|
}
|
||
|
|
}
|
||
|
|
}()
|
||
|
|
}
|
||
|
|
|
||
|
|
wg.Wait()
|
||
|
|
|
||
|
|
// Wait for work to settle
|
||
|
|
time.Sleep(100 * time.Millisecond)
|
||
|
|
|
||
|
|
// Should have processed some work without crashing
|
||
|
|
stats := runner.Stats()
|
||
|
|
if stats.WorkCycles() == 0 {
|
||
|
|
t.Error("Expected some work cycles")
|
||
|
|
}
|
||
|
|
|
||
|
|
// Max concurrent should still be 1
|
||
|
|
if stats.MaxConcurrent() > 1 {
|
||
|
|
t.Errorf("Expected max concurrent to be 1, got %d", stats.MaxConcurrent())
|
||
|
|
}
|
||
|
|
}
|