170 lines
8.2 KiB
Go
170 lines
8.2 KiB
Go
//go:build integration
|
|
|
|
package redisactivity
|
|
|
|
import (
|
|
"context"
|
|
"errors"
|
|
"testing"
|
|
"time"
|
|
|
|
"github.com/redis/go-redis/v9"
|
|
|
|
"proxy-pool/internal/domain/activitypool"
|
|
healthDomain "proxy-pool/internal/domain/health"
|
|
proxyDomain "proxy-pool/internal/domain/proxy"
|
|
)
|
|
|
|
func TestRedisAppliesGlobalHealthObservationAtomically(t *testing.T) {
|
|
fixture := newRedisTestFixture(t)
|
|
now := redisTestNow()
|
|
if _, err := fixture.Adapter.UpsertFetched(context.Background(), "provider-a", activitypool.FetchedBatch{
|
|
ObservedAt: now, ConfiguredTTL: time.Minute, MaxSize: 1,
|
|
Proxies: []proxyDomain.Proxy{testProxy("proxy-a", "192.0.2.10")},
|
|
}); err != nil {
|
|
t.Fatalf("UpsertFetched(): %v", err)
|
|
}
|
|
if _, err := fixture.Adapter.ApplyHealth(context.Background(), activitypool.HealthUpdate{
|
|
ProxyID: "proxy-a", CheckedAt: now.Add(time.Second), NextState: proxyDomain.StateChecking,
|
|
}); err != nil {
|
|
t.Fatalf("ApplyHealth(checking): %v", err)
|
|
}
|
|
first := redisGlobalObservation("task-1", false, now.Add(2*time.Second))
|
|
failed, err := fixture.Adapter.ApplyGlobalObservation(context.Background(), activitypool.GlobalHealthCommand{
|
|
Observation: first, MaxConsecutiveFailures: 3,
|
|
})
|
|
if err != nil || failed.State != proxyDomain.StateUnhealthy || failed.GlobalHealth.ConsecutiveFailures != 1 ||
|
|
!failed.GlobalHealth.UnhealthySince.Equal(now.Add(2*time.Second)) {
|
|
t.Fatalf("ApplyGlobalObservation(initial failure) = %+v, %v", failed, err)
|
|
}
|
|
replayed, err := fixture.Adapter.ApplyGlobalObservation(context.Background(), activitypool.GlobalHealthCommand{
|
|
Observation: first, MaxConsecutiveFailures: 3,
|
|
})
|
|
if err != nil || replayed.GlobalHealth != failed.GlobalHealth {
|
|
t.Fatalf("ApplyGlobalObservation(replay) = %+v, %v", replayed, err)
|
|
}
|
|
conflicting := first
|
|
conflicting.Success = true
|
|
conflicting.FailureClass = ""
|
|
if _, err := fixture.Adapter.ApplyGlobalObservation(context.Background(), activitypool.GlobalHealthCommand{
|
|
Observation: conflicting, MaxConsecutiveFailures: 3,
|
|
}); !errors.Is(err, healthDomain.ErrConflictingObservation) {
|
|
t.Fatalf("ApplyGlobalObservation(conflict) error = %v, want ErrConflictingObservation", err)
|
|
}
|
|
if _, err := fixture.Adapter.ApplyHealth(context.Background(), activitypool.HealthUpdate{
|
|
ProxyID: "proxy-a", CheckedAt: now.Add(3 * time.Second), NextState: proxyDomain.StateChecking,
|
|
}); err != nil {
|
|
t.Fatalf("ApplyHealth(recheck): %v", err)
|
|
}
|
|
recovered, err := fixture.Adapter.ApplyGlobalObservation(context.Background(), activitypool.GlobalHealthCommand{
|
|
Observation: redisGlobalObservation("task-2", true, now.Add(4*time.Second)), MaxConsecutiveFailures: 3,
|
|
})
|
|
if err != nil || recovered.State != proxyDomain.StateAvailable || recovered.GlobalHealth.ConsecutiveFailures != 0 ||
|
|
!recovered.GlobalHealth.UnhealthySince.IsZero() ||
|
|
recovered.GlobalHealth.LastObservedAt != now.Add(4*time.Second) {
|
|
t.Fatalf("ApplyGlobalObservation(recovery) = %+v, %v", recovered, err)
|
|
}
|
|
}
|
|
|
|
func TestRedisSweepsUnownedProxiesAfterUnhealthyGrace(t *testing.T) {
|
|
fixture := newRedisTestFixture(t)
|
|
now := redisTestNow()
|
|
if _, err := fixture.Adapter.UpsertFetched(context.Background(), "provider-a", activitypool.FetchedBatch{
|
|
ObservedAt: now, ConfiguredTTL: time.Minute, MaxSize: 1,
|
|
Proxies: []proxyDomain.Proxy{testProxy("proxy-a", "192.0.2.10")},
|
|
}); err != nil {
|
|
t.Fatalf("UpsertFetched() = %v", err)
|
|
}
|
|
if _, err := fixture.Adapter.ApplyHealth(context.Background(), activitypool.HealthUpdate{
|
|
ProxyID: "proxy-a", CheckedAt: now.Add(time.Second), NextState: proxyDomain.StateChecking,
|
|
}); err != nil {
|
|
t.Fatalf("ApplyHealth(checking) = %v", err)
|
|
}
|
|
if _, err := fixture.Adapter.ApplyGlobalObservation(context.Background(), activitypool.GlobalHealthCommand{
|
|
Observation: redisGlobalObservation("task-unhealthy", false, now.Add(2*time.Second)), MaxConsecutiveFailures: 2,
|
|
}); err != nil {
|
|
t.Fatalf("ApplyGlobalObservation(unhealthy) = %v", err)
|
|
}
|
|
if _, err := fixture.Client.ZScore(context.Background(), fixture.Adapter.keys.healthUnhealthy, "proxy-a").Result(); err != nil {
|
|
t.Fatalf("health unhealthy index = %v", err)
|
|
}
|
|
command := activitypool.UnhealthySweepCommand{
|
|
Now: now.Add(3 * time.Second), Limit: 4, RemoveAfterByUpstream: map[string]time.Duration{"provider-a": 3 * time.Second},
|
|
}
|
|
if result, err := fixture.Adapter.SweepUnhealthy(context.Background(), command); err != nil || result != (activitypool.UnhealthySweepResult{}) {
|
|
t.Fatalf("SweepUnhealthy(before grace) = %+v, %v", result, err)
|
|
}
|
|
command.Now = now.Add(6 * time.Second)
|
|
if result, err := fixture.Adapter.SweepUnhealthy(context.Background(), command); err != nil || result != (activitypool.UnhealthySweepResult{Removed: 1}) {
|
|
t.Fatalf("SweepUnhealthy(after grace) = %+v, %v", result, err)
|
|
}
|
|
if _, err := fixture.Adapter.UpstreamForProxy(context.Background(), "proxy-a", command.Now); !errors.Is(err, activitypool.ErrActivityNotFound) {
|
|
t.Fatalf("UpstreamForProxy(removed) error = %v, want ErrActivityNotFound", err)
|
|
}
|
|
if _, err := fixture.Client.ZScore(context.Background(), fixture.Adapter.keys.healthUnhealthy, "proxy-a").Result(); !errors.Is(err, redis.Nil) {
|
|
t.Fatalf("health unhealthy index after removal = %v, want redis.Nil", err)
|
|
}
|
|
}
|
|
|
|
func TestRedisUnhealthySweepDefersOwnedProxyUntilDrainAcknowledgement(t *testing.T) {
|
|
fixture := newRedisTestFixture(t)
|
|
now := redisTestNow()
|
|
if _, err := fixture.Adapter.UpsertFetched(context.Background(), "provider-a", activitypool.FetchedBatch{
|
|
ObservedAt: now, ConfiguredTTL: time.Minute, MaxSize: 1,
|
|
Proxies: []proxyDomain.Proxy{testProxy("proxy-a", "192.0.2.10")},
|
|
}); err != nil {
|
|
t.Fatalf("UpsertFetched() = %v", err)
|
|
}
|
|
if _, err := fixture.Adapter.ApplyHealth(context.Background(), activitypool.HealthUpdate{
|
|
ProxyID: "proxy-a", CheckedAt: now.Add(time.Second), NextState: proxyDomain.StateChecking,
|
|
}); err != nil {
|
|
t.Fatalf("ApplyHealth(checking) = %v", err)
|
|
}
|
|
if _, err := fixture.Adapter.ApplyGlobalObservation(context.Background(), activitypool.GlobalHealthCommand{
|
|
Observation: redisGlobalObservation("task-available", true, now.Add(2*time.Second)), MaxConsecutiveFailures: 2,
|
|
}); err != nil {
|
|
t.Fatalf("ApplyGlobalObservation(available) = %v", err)
|
|
}
|
|
assignment, err := fixture.Adapter.Assign(context.Background(), now.Add(3*time.Second), "proxy-a", "worker-a", time.Minute)
|
|
if err != nil {
|
|
t.Fatalf("Assign() = %v", err)
|
|
}
|
|
if _, err := fixture.Adapter.ApplyGlobalObservation(context.Background(), activitypool.GlobalHealthCommand{
|
|
Observation: redisGlobalObservation("task-suspect", false, now.Add(4*time.Second)), MaxConsecutiveFailures: 2,
|
|
}); err != nil {
|
|
t.Fatalf("ApplyGlobalObservation(suspect) = %v", err)
|
|
}
|
|
if _, err := fixture.Adapter.ApplyGlobalObservation(context.Background(), activitypool.GlobalHealthCommand{
|
|
Observation: redisGlobalObservation("task-unhealthy", false, now.Add(5*time.Second)), MaxConsecutiveFailures: 2,
|
|
}); err != nil {
|
|
t.Fatalf("ApplyGlobalObservation(unhealthy) = %v", err)
|
|
}
|
|
command := activitypool.UnhealthySweepCommand{
|
|
Now: now.Add(7 * time.Second), Limit: 4, RemoveAfterByUpstream: map[string]time.Duration{"provider-a": time.Second},
|
|
}
|
|
if result, err := fixture.Adapter.SweepUnhealthy(context.Background(), command); err != nil || result != (activitypool.UnhealthySweepResult{DeferredOwned: 1}) {
|
|
t.Fatalf("SweepUnhealthy(owned) = %+v, %v", result, err)
|
|
}
|
|
if _, err := fixture.Adapter.BeginDrain(context.Background(), "proxy-a", "worker-a", assignment.Epoch); err != nil {
|
|
t.Fatalf("BeginDrain() = %v", err)
|
|
}
|
|
if err := fixture.Adapter.AcknowledgeDrain(context.Background(), "proxy-a", "worker-a", assignment.Epoch, 0, 0); err != nil {
|
|
t.Fatalf("AcknowledgeDrain() = %v", err)
|
|
}
|
|
command.Now = now.Add(8 * time.Second)
|
|
if result, err := fixture.Adapter.SweepUnhealthy(context.Background(), command); err != nil || result != (activitypool.UnhealthySweepResult{Removed: 1}) {
|
|
t.Fatalf("SweepUnhealthy(unowned) = %+v, %v", result, err)
|
|
}
|
|
}
|
|
|
|
func redisGlobalObservation(taskID string, success bool, observedAt time.Time) healthDomain.Observation {
|
|
observation := healthDomain.Observation{
|
|
TaskID: taskID, ProxyID: "proxy-a", Level: healthDomain.LevelBasic,
|
|
Success: success, Latency: 10 * time.Millisecond, ObservedAt: observedAt,
|
|
}
|
|
if !success {
|
|
observation.FailureClass = "timeout"
|
|
}
|
|
return observation
|
|
}
|