From 739910f99b9257b1bbe30ed17da501a718542c87 Mon Sep 17 00:00:00 2001 From: youfak Date: Wed, 29 Jul 2026 17:46:20 +0800 Subject: [PATCH] test: add redis activity pool contract suite --- .../contract_integration_test.go | 16 + .../adapters/redisactivity/testredis_test.go | 24 +- .../activitypool/contract_external_test.go | 14 + .../activitypool/contracttest/contract.go | 508 ++++++++++++++++++ internal/domain/activitypool/pool.go | 55 +- scripts/test-redis.ps1 | 2 +- 6 files changed, 588 insertions(+), 31 deletions(-) create mode 100644 internal/adapters/redisactivity/contract_integration_test.go create mode 100644 internal/domain/activitypool/contract_external_test.go create mode 100644 internal/domain/activitypool/contracttest/contract.go diff --git a/internal/adapters/redisactivity/contract_integration_test.go b/internal/adapters/redisactivity/contract_integration_test.go new file mode 100644 index 0000000..03235a4 --- /dev/null +++ b/internal/adapters/redisactivity/contract_integration_test.go @@ -0,0 +1,16 @@ +//go:build integration + +package redisactivity + +import ( + "testing" + + "proxy-pool/internal/domain/activitypool/contracttest" +) + +func TestRedisActivityPoolContract(t *testing.T) { + contracttest.Run(t, func(t *testing.T) (contracttest.Store, func()) { + fixture := newRedisTestFixture(t) + return fixture.Adapter, fixture.Cleanup + }) +} diff --git a/internal/adapters/redisactivity/testredis_test.go b/internal/adapters/redisactivity/testredis_test.go index 1e8b7de..95b6a03 100644 --- a/internal/adapters/redisactivity/testredis_test.go +++ b/internal/adapters/redisactivity/testredis_test.go @@ -6,6 +6,7 @@ import ( "context" "fmt" "os" + "sync" "sync/atomic" "testing" "time" @@ -22,6 +23,7 @@ type redisTestFixture struct { Client *redis.Client Credentials *credentials.MemoryStore Namespace string + Cleanup func() } func TestRedisFixtureUsesIsolatedNamespace(t *testing.T) { @@ -66,16 +68,20 @@ func newRedisTestFixture(t *testing.T) redisTestFixture { _ = client.Close() t.Fatalf("New(): %v", err) } - t.Cleanup(func() { - cleanupCtx, cleanupCancel := context.WithTimeout(context.Background(), 5*time.Second) - defer cleanupCancel() - if err := deleteRedisNamespace(cleanupCtx, client, redisKeyPrefix+namespace+":*"); err != nil { - t.Errorf("clean Redis test namespace: %v", err) - } - _ = client.Close() - }) + var cleanupOnce sync.Once + cleanup := func() { + cleanupOnce.Do(func() { + cleanupCtx, cleanupCancel := context.WithTimeout(context.Background(), 5*time.Second) + defer cleanupCancel() + if err := deleteRedisNamespace(cleanupCtx, client, redisKeyPrefix+namespace+":*"); err != nil { + t.Errorf("clean Redis test namespace: %v", err) + } + _ = client.Close() + }) + } + t.Cleanup(cleanup) return redisTestFixture{ - Adapter: adapter, Client: client, Credentials: credentialStore, Namespace: namespace, + Adapter: adapter, Client: client, Credentials: credentialStore, Namespace: namespace, Cleanup: cleanup, } } diff --git a/internal/domain/activitypool/contract_external_test.go b/internal/domain/activitypool/contract_external_test.go new file mode 100644 index 0000000..b919d8c --- /dev/null +++ b/internal/domain/activitypool/contract_external_test.go @@ -0,0 +1,14 @@ +package activitypool_test + +import ( + "testing" + + "proxy-pool/internal/domain/activitypool" + "proxy-pool/internal/domain/activitypool/contracttest" +) + +func TestMemoryPoolContract(t *testing.T) { + contracttest.Run(t, func(*testing.T) (contracttest.Store, func()) { + return activitypool.NewMemoryPool(), func() {} + }) +} diff --git a/internal/domain/activitypool/contracttest/contract.go b/internal/domain/activitypool/contracttest/contract.go new file mode 100644 index 0000000..dd2b2ef --- /dev/null +++ b/internal/domain/activitypool/contracttest/contract.go @@ -0,0 +1,508 @@ +package contracttest + +import ( + "context" + "errors" + "fmt" + "sync" + "testing" + "time" + + "proxy-pool/internal/domain/activitypool" + extractionDomain "proxy-pool/internal/domain/extraction" + ownershipDomain "proxy-pool/internal/domain/ownership" + proxyDomain "proxy-pool/internal/domain/proxy" +) + +type Store interface { + activitypool.Upserter + activitypool.HealthStore + activitypool.InventoryReader + activitypool.Maintainer + extractionDomain.Store + ownershipDomain.Repository +} + +type Factory func(*testing.T) (Store, func()) + +func Run(t *testing.T, factory Factory) { + t.Helper() + t.Run("upsert capacity and incumbent lifecycle", func(t *testing.T) { + runUpsertContract(t, newStore(t, factory)) + }) + t.Run("health and filtered extraction", func(t *testing.T) { + runHealthAndExtractionContract(t, newStore(t, factory)) + }) + t.Run("fulfillment and gateway reserve", func(t *testing.T) { + runFulfillmentContract(t, factory) + }) + t.Run("business idempotency", func(t *testing.T) { + runIdempotencyContract(t, factory) + }) + t.Run("idempotency is bounded by proxy expiry", func(t *testing.T) { + runIdempotencyExpiryContract(t, newStore(t, factory)) + }) + t.Run("ownership lifecycle", func(t *testing.T) { + runOwnershipContract(t, factory) + }) + t.Run("inventory and bounded maintenance", func(t *testing.T) { + runMaintenanceContract(t, newStore(t, factory)) + }) + t.Run("concurrent exclusivity", func(t *testing.T) { + runConcurrencyContract(t, factory) + }) + t.Run("canceled contexts", func(t *testing.T) { + runCancellationContract(t, newStore(t, factory)) + }) +} + +func runUpsertContract(t *testing.T, store Store) { + t.Helper() + now := contractNow() + firstProxy := contractProxy("proxy-a", "192.0.2.10", proxyDomain.StateFetched) + secondProxy := contractProxy("proxy-b", "192.0.2.11", proxyDomain.StateFetched) + result, err := store.UpsertFetched(context.Background(), "provider-a", activitypool.FetchedBatch{ + ObservedAt: now, ConfiguredTTL: 30 * time.Second, AllocationSafetyMargin: 3 * time.Second, + MaxSize: 1, Proxies: []proxyDomain.Proxy{firstProxy, secondProxy}, + }) + if err != nil || result.Accepted != 2 || result.Inserted != 1 || result.Dropped != 1 { + t.Fatalf("UpsertFetched(capacity) = %+v, %v", result, err) + } + assertInventory(t, store, "provider-a", now, 1) + + refreshed, err := store.UpsertFetched(context.Background(), "provider-b", activitypool.FetchedBatch{ + ObservedAt: now.Add(time.Second), ConfiguredTTL: 5 * time.Minute, + AllocationSafetyMargin: 10 * time.Second, MaxSize: 10, Proxies: []proxyDomain.Proxy{firstProxy}, + }) + if err != nil || refreshed.Inserted != 0 || refreshed.Refreshed != 1 { + t.Fatalf("UpsertFetched(cross-provider refresh) = %+v, %v", refreshed, err) + } + assertInventory(t, store, "provider-a", now.Add(time.Second), 1) + assertInventory(t, store, "provider-b", now.Add(time.Second), 0) + + replaced, err := store.UpsertFetched(context.Background(), "provider-b", activitypool.FetchedBatch{ + ObservedAt: now.Add(31 * time.Second), ConfiguredTTL: time.Minute, + AllocationSafetyMargin: 5 * time.Second, MaxSize: 10, Proxies: []proxyDomain.Proxy{firstProxy}, + }) + if err != nil || replaced.Inserted != 1 || replaced.Refreshed != 0 { + t.Fatalf("UpsertFetched(after incumbent expiry) = %+v, %v", replaced, err) + } + assertInventory(t, store, "provider-a", now.Add(31*time.Second), 0) + assertInventory(t, store, "provider-b", now.Add(31*time.Second), 1) +} + +func runHealthAndExtractionContract(t *testing.T, store Store) { + t.Helper() + now := contractNow() + candidate := contractProxy("proxy-a", "192.0.2.10", proxyDomain.StateFetched) + candidate.Tags = map[string]string{"region": "cn", "carrier": "ct"} + upsertOne(t, store, "provider-a", now, time.Minute, candidate) + checking, err := store.ApplyHealth(context.Background(), activitypool.HealthUpdate{ + ProxyID: candidate.ID, CheckedAt: now.Add(time.Second), NextState: proxyDomain.StateChecking, + }) + if err != nil || checking.State != proxyDomain.StateChecking { + t.Fatalf("ApplyHealth(checking) = %+v, %v", checking, err) + } + available, err := store.ApplyHealth(context.Background(), activitypool.HealthUpdate{ + ProxyID: candidate.ID, CheckedAt: now.Add(2 * time.Second), + NextState: proxyDomain.StateAvailable, Latency: 25 * time.Millisecond, + }) + if err != nil || available.State != proxyDomain.StateAvailable || + available.Proxy.LastCheckedAt == nil || !available.Proxy.LastCheckedAt.Equal(now.Add(2*time.Second)) || + available.Proxy.LastSuccessAt == nil || !available.Proxy.LastSuccessAt.Equal(now.Add(2*time.Second)) { + t.Fatalf("ApplyHealth(available) = %+v, %v", available, err) + } + if _, err := store.ApplyHealth(context.Background(), activitypool.HealthUpdate{ + ProxyID: candidate.ID, CheckedAt: now.Add(time.Second), NextState: proxyDomain.StateSuspect, + }); !errors.Is(err, activitypool.ErrStaleHealthUpdate) { + t.Fatalf("ApplyHealth(stale) error = %v", err) + } + + result, err := store.Extract(context.Background(), extractionDomain.Command{ + RequestID: "req-filter", ClientID: "client-a", Requested: 1, + Fulfillment: extractionDomain.Partial, Now: now.Add(3 * time.Second), + MinRemainingTTL: 10 * time.Second, MaxHealthCheckAge: 5 * time.Second, + Protocols: []string{"http"}, Regions: []string{"cn"}, Carriers: []string{"ct"}, + Upstreams: []string{"provider-a"}, + }) + if err != nil || result.Returned != 1 || result.Items[0].ID != candidate.ID || + result.Items[0].State != extractionDomain.Extracted { + t.Fatalf("Extract(filtered) = %+v, %v", result, err) + } + assertInventory(t, store, "provider-a", now.Add(3*time.Second), 0) +} + +func runFulfillmentContract(t *testing.T, factory Factory) { + t.Helper() + now := contractNow() + store := newStore(t, factory) + seedAvailable(t, store, "provider-a", now, time.Minute, + contractProxy("proxy-a", "192.0.2.10", proxyDomain.StateAvailable)) + if result, err := store.Extract(context.Background(), extractionDomain.Command{ + RequestID: "req-all", ClientID: "client-a", Requested: 2, + Fulfillment: extractionDomain.AllOrNothing, Now: now.Add(time.Second), + }); !errors.Is(err, extractionDomain.ErrInsufficientProxies) || result.Returned != 0 { + t.Fatalf("Extract(allOrNothing) = %+v, %v", result, err) + } + if result, err := store.Extract(context.Background(), extractionDomain.Command{ + RequestID: "req-partial", ClientID: "client-a", Requested: 1, + Fulfillment: extractionDomain.Partial, Now: now.Add(time.Second), + }); err != nil || result.Returned != 1 { + t.Fatalf("Extract(partial after insufficient) = %+v, %v", result, err) + } + + reservedStore := newStore(t, factory) + seedAvailable(t, reservedStore, "provider-a", now, time.Minute, + contractProxy("proxy-reserved", "192.0.2.11", proxyDomain.StateAvailable)) + reserved, err := reservedStore.Extract(context.Background(), extractionDomain.Command{ + RequestID: "req-reserved", ClientID: "client-a", Requested: 1, + Fulfillment: extractionDomain.Partial, Now: now.Add(time.Second), ReserveForGateway: 1, + }) + if err != nil || reserved.Returned != 0 || len(reserved.Items) != 0 { + t.Fatalf("Extract(reserved) = %+v, %v", reserved, err) + } + + marginStore := newStore(t, factory) + marginProxy := contractProxy("proxy-margin", "192.0.2.12", proxyDomain.StateAvailable) + if result, err := marginStore.UpsertFetched(context.Background(), "provider-a", activitypool.FetchedBatch{ + ObservedAt: now, ConfiguredTTL: 30 * time.Second, AllocationSafetyMargin: 5 * time.Second, + MaxSize: 10, Proxies: []proxyDomain.Proxy{marginProxy}, + }); err != nil || result.Inserted != 1 { + t.Fatalf("UpsertFetched(safety margin) = %+v, %v", result, err) + } + if result, err := marginStore.Extract(context.Background(), extractionDomain.Command{ + RequestID: "req-margin", ClientID: "client-a", Requested: 1, + Fulfillment: extractionDomain.Partial, Now: now.Add(26 * time.Second), + }); err != nil || result.Returned != 0 { + t.Fatalf("Extract(after usableUntil) = %+v, %v", result, err) + } +} + +func runIdempotencyContract(t *testing.T, factory Factory) { + t.Helper() + now := contractNow() + store := newStore(t, factory) + candidate := contractProxy("proxy-a", "192.0.2.10", proxyDomain.StateAvailable) + candidate.Tags = map[string]string{"region": "cn", "carrier": "ct"} + checkedAt := now + candidate.LastCheckedAt = &checkedAt + seedAvailable(t, store, "provider-a", now, time.Minute, candidate) + command := extractionDomain.Command{ + RequestID: "req-idem-first", ClientID: "client-a", SourceIP: "192.0.2.100", + IdempotencyKey: "idem-contract", IdempotencyTTL: time.Minute, + Requested: 1, Fulfillment: extractionDomain.Partial, Now: now.Add(time.Second), + MinRemainingTTL: time.Second, MaxHealthCheckAge: time.Minute, + Protocols: []string{"http", "http"}, Regions: []string{"cn", "cn"}, + Carriers: []string{"ct", "ct"}, Upstreams: []string{"provider-a", "provider-a"}, + } + first, err := store.Extract(context.Background(), command) + if err != nil || first.Returned != 1 { + t.Fatalf("Extract(idempotent first) = %+v, %v", first, err) + } + command.RequestID = "req-idem-replay" + command.SourceIP = "198.51.100.200" + command.Now = now.Add(2 * time.Second) + command.IdempotencyTTL = 2 * time.Minute + command.MinRemainingTTL = 2 * time.Second + command.MaxHealthCheckAge = 2 * time.Minute + command.Protocols = []string{"http"} + command.Regions = []string{"cn"} + command.Carriers = []string{"ct"} + command.Upstreams = []string{"provider-a"} + replayed, err := store.Extract(context.Background(), command) + if err != nil || replayed.Returned != 1 || replayed.Items[0].ID != first.Items[0].ID || + !replayed.ExtractedAt.Equal(first.ExtractedAt) { + t.Fatalf("Extract(idempotent replay) = %+v, %v", replayed, err) + } + command.RequestID = "req-idem-conflict" + command.Requested = 2 + if _, err := store.Extract(context.Background(), command); !errors.Is(err, extractionDomain.ErrIdempotencyConflict) { + t.Fatalf("Extract(idempotent conflict) error = %v", err) + } + + zeroStore := newStore(t, factory) + zero := extractionDomain.Command{ + RequestID: "req-zero", ClientID: "client-a", IdempotencyKey: "idem-zero", + Requested: 0, Fulfillment: extractionDomain.Partial, Now: now, + } + if result, err := zeroStore.Extract(context.Background(), zero); err != nil || result.Returned != 0 { + t.Fatalf("Extract(zero first) = %+v, %v", result, err) + } + zero.RequestID = "req-zero-replay" + zero.Now = now.Add(time.Second) + if result, err := zeroStore.Extract(context.Background(), zero); err != nil || result.Returned != 0 { + t.Fatalf("Extract(zero replay) = %+v, %v", result, err) + } + zero.RequestID = "req-zero-conflict" + zero.Requested = 1 + if _, err := zeroStore.Extract(context.Background(), zero); !errors.Is(err, extractionDomain.ErrIdempotencyConflict) { + t.Fatalf("Extract(zero conflict) error = %v", err) + } +} + +func runIdempotencyExpiryContract(t *testing.T, store Store) { + t.Helper() + now := time.Now().UTC().Truncate(time.Millisecond) + candidate := contractProxy("short-lived", "192.0.2.20", proxyDomain.StateAvailable) + upsertOne(t, store, "provider-a", now, 600*time.Millisecond, candidate) + command := extractionDomain.Command{ + RequestID: "req-expiry-first", ClientID: "client-a", IdempotencyKey: "idem-expiry-contract", + IdempotencyTTL: time.Minute, Requested: 1, + Fulfillment: extractionDomain.Partial, Now: now.Add(time.Millisecond), + } + first, err := store.Extract(context.Background(), command) + if err != nil || first.Returned != 1 { + t.Fatalf("Extract(short-lived first) = %+v, %v", first, err) + } + time.Sleep(750 * time.Millisecond) + replacementAt := time.Now().UTC().Truncate(time.Millisecond) + upsertOne(t, store, "provider-a", replacementAt, time.Minute, candidate) + command.RequestID = "req-expiry-second" + command.Now = replacementAt.Add(time.Millisecond) + again, err := store.Extract(context.Background(), command) + if err != nil || again.Returned != 1 || again.ExtractedAt.Equal(first.ExtractedAt) { + t.Fatalf("Extract(after idempotency expiry) = %+v, %v", again, err) + } +} + +func runOwnershipContract(t *testing.T, factory Factory) { + t.Helper() + now := contractNow() + store := newStore(t, factory) + seedAvailable(t, store, "provider-a", now, 2*time.Minute, + contractProxy("proxy-a", "192.0.2.10", proxyDomain.StateAvailable)) + assigned, err := store.Assign(context.Background(), now.Add(time.Second), "proxy-a", "worker-a", 20*time.Second) + if err != nil || assigned.Epoch == 0 || assigned.Version != 1 { + t.Fatalf("Assign() = %+v, %v", assigned, err) + } + if _, err := store.Assign(context.Background(), now.Add(2*time.Second), "proxy-a", "worker-b", time.Minute); !errors.Is(err, ownershipDomain.ErrAlreadyOwned) { + t.Fatalf("Assign(already owned) error = %v", err) + } + renewed, err := store.Renew(context.Background(), now.Add(10*time.Second), "proxy-a", "worker-a", assigned.Epoch, 5*time.Minute) + if err != nil || renewed.Version != 2 || !renewed.ExpiresAt.Equal(now.Add(2*time.Minute)) { + t.Fatalf("Renew() = %+v, %v", renewed, err) + } + draining, err := store.BeginDrain(context.Background(), "proxy-a", "worker-a", assigned.Epoch) + if err != nil || !draining.Draining || draining.Version != 3 { + t.Fatalf("BeginDrain() = %+v, %v", draining, err) + } + if replayed, err := store.BeginDrain(context.Background(), "proxy-a", "worker-a", assigned.Epoch); err != nil || replayed != draining { + t.Fatalf("BeginDrain(replay) = %+v, %v", replayed, err) + } + if err := store.AcknowledgeDrain(context.Background(), "proxy-a", "worker-a", assigned.Epoch, 1, 0); !errors.Is(err, ownershipDomain.ErrDrainNotReady) { + t.Fatalf("AcknowledgeDrain(active) error = %v", err) + } + if err := store.AcknowledgeDrain(context.Background(), "proxy-a", "worker-a", assigned.Epoch, 0, 0); err != nil { + t.Fatalf("AcknowledgeDrain(): %v", err) + } + if current, ok, err := store.Get(context.Background(), "proxy-a"); err != nil || ok { + t.Fatalf("Get(after ACK) = %+v, %t, %v", current, ok, err) + } + if result, err := store.Extract(context.Background(), extractionDomain.Command{ + RequestID: "req-after-ack", ClientID: "client-a", Requested: 1, + Fulfillment: extractionDomain.Partial, Now: now.Add(20 * time.Second), + }); err != nil || result.Returned != 1 { + t.Fatalf("Extract(after ACK) = %+v, %v", result, err) + } + + takeoverStore := newStore(t, factory) + seedAvailable(t, takeoverStore, "provider-a", now, 2*time.Minute, + contractProxy("takeover", "192.0.2.11", proxyDomain.StateAvailable)) + old, err := takeoverStore.Assign(context.Background(), now.Add(time.Second), "takeover", "worker-a", time.Second) + if err != nil { + t.Fatalf("Assign(takeover old): %v", err) + } + newAssignment, err := takeoverStore.Assign(context.Background(), now.Add(3*time.Second), "takeover", "worker-b", time.Minute) + if err != nil || newAssignment.Epoch <= old.Epoch || newAssignment.WorkerID != "worker-b" { + t.Fatalf("Assign(takeover new) = %+v, %v; old=%+v", newAssignment, err, old) + } +} + +func runMaintenanceContract(t *testing.T, store Store) { + t.Helper() + now := contractNow() + for index := range 2 { + upsertOne(t, store, "provider-a", now, 5*time.Second, + contractProxy(fmt.Sprintf("proxy-%d", index), fmt.Sprintf("192.0.2.%d", index+10), proxyDomain.StateFetched)) + } + assertInventory(t, store, "provider-a", now.Add(time.Second), 2) + if removed, err := store.SweepExpired(context.Background(), now.Add(6*time.Second), 1); err != nil || removed != 1 { + t.Fatalf("SweepExpired(first) = %d, %v", removed, err) + } + if removed, err := store.SweepExpired(context.Background(), now.Add(6*time.Second), 1); err != nil || removed != 1 { + t.Fatalf("SweepExpired(second) = %d, %v", removed, err) + } + if removed, err := store.SweepExpired(context.Background(), now.Add(6*time.Second), 1); err != nil || removed != 0 { + t.Fatalf("SweepExpired(empty) = %d, %v", removed, err) + } + assertInventory(t, store, "provider-a", now.Add(6*time.Second), 0) +} + +func runConcurrencyContract(t *testing.T, factory Factory) { + t.Helper() + now := contractNow() + extractStore := newStore(t, factory) + for iteration := range 100 { + proxyID := fmt.Sprintf("extract-race-%d", iteration) + seedAvailable(t, extractStore, "provider-a", now, 10*time.Minute, + contractProxy(proxyID, fmt.Sprintf("198.51.100.%d", iteration+1), proxyDomain.StateAvailable)) + results := make(chan extractionDomain.Result, 2) + errorsCh := make(chan error, 2) + var workers sync.WaitGroup + for worker := range 2 { + workers.Add(1) + go func(worker int) { + defer workers.Done() + result, err := extractStore.Extract(context.Background(), extractionDomain.Command{ + RequestID: fmt.Sprintf("extract-race-%d-%d", iteration, worker), + ClientID: fmt.Sprintf("client-%d", worker), Requested: 1, + Fulfillment: extractionDomain.Partial, Now: now.Add(time.Second), + }) + results <- result + errorsCh <- err + }(worker) + } + workers.Wait() + for range 2 { + if err := <-errorsCh; err != nil { + t.Fatalf("iteration %d Extract(): %v", iteration, err) + } + } + returned := (<-results).Returned + (<-results).Returned + if returned != 1 { + t.Fatalf("iteration %d extracted = %d, want 1", iteration, returned) + } + } + + ownershipStore := newStore(t, factory) + for iteration := range 100 { + proxyID := fmt.Sprintf("ownership-race-%d", iteration) + seedAvailable(t, ownershipStore, "provider-a", now, 10*time.Minute, + contractProxy(proxyID, fmt.Sprintf("203.0.113.%d", iteration+1), proxyDomain.StateAvailable)) + var workers sync.WaitGroup + assigned := make(chan bool, 1) + extracted := make(chan bool, 1) + errorsCh := make(chan error, 2) + workers.Add(2) + go func() { + defer workers.Done() + _, err := ownershipStore.Assign(context.Background(), now.Add(time.Second), proxyID, "worker-a", time.Minute) + if err != nil && !errors.Is(err, ownershipDomain.ErrOwnershipUnavailable) { + errorsCh <- err + } + assigned <- err == nil + }() + go func() { + defer workers.Done() + result, err := ownershipStore.Extract(context.Background(), extractionDomain.Command{ + RequestID: fmt.Sprintf("ownership-race-%d", iteration), ClientID: "client-a", Requested: 1, + Fulfillment: extractionDomain.Partial, Now: now.Add(time.Second), + }) + if err != nil { + errorsCh <- err + } + extracted <- err == nil && result.Returned == 1 + }() + workers.Wait() + close(errorsCh) + for err := range errorsCh { + t.Fatalf("iteration %d ownership race: %v", iteration, err) + } + wins := 0 + if <-assigned { + wins++ + } + if <-extracted { + wins++ + } + if wins != 1 { + t.Fatalf("iteration %d winners = %d, want 1", iteration, wins) + } + } +} + +func runCancellationContract(t *testing.T, store Store) { + t.Helper() + now := contractNow() + ctx, cancel := context.WithCancel(context.Background()) + cancel() + checks := []struct { + name string + call func() error + }{ + {name: "upsert", call: func() error { + _, err := store.UpsertFetched(ctx, "provider-a", activitypool.FetchedBatch{ObservedAt: now, ConfiguredTTL: time.Minute, MaxSize: 1}) + return err + }}, + {name: "health", call: func() error { + _, err := store.ApplyHealth(ctx, activitypool.HealthUpdate{ProxyID: "proxy-a", CheckedAt: now, NextState: proxyDomain.StateChecking}) + return err + }}, + {name: "inventory", call: func() error { _, err := store.Inventory(ctx, "provider-a", now); return err }}, + {name: "sweep", call: func() error { _, err := store.SweepExpired(ctx, now, 1); return err }}, + {name: "extract", call: func() error { + _, err := store.Extract(ctx, extractionDomain.Command{Requested: 1, Fulfillment: extractionDomain.Partial, Now: now}) + return err + }}, + {name: "assign", call: func() error { _, err := store.Assign(ctx, now, "proxy-a", "worker-a", time.Minute); return err }}, + {name: "renew", call: func() error { _, err := store.Renew(ctx, now, "proxy-a", "worker-a", 1, time.Minute); return err }}, + {name: "begin drain", call: func() error { _, err := store.BeginDrain(ctx, "proxy-a", "worker-a", 1); return err }}, + {name: "acknowledge drain", call: func() error { return store.AcknowledgeDrain(ctx, "proxy-a", "worker-a", 1, 0, 0) }}, + {name: "get", call: func() error { _, _, err := store.Get(ctx, "proxy-a"); return err }}, + {name: "expire", call: func() error { _, err := store.Expire(ctx, now, 1); return err }}, + } + for _, check := range checks { + t.Run(check.name, func(t *testing.T) { + if err := check.call(); !errors.Is(err, context.Canceled) { + t.Fatalf("error = %v, want context.Canceled", err) + } + }) + } +} + +func newStore(t *testing.T, factory Factory) Store { + t.Helper() + store, cleanup := factory(t) + if store == nil { + t.Fatal("contract factory returned nil store") + } + if cleanup != nil { + t.Cleanup(cleanup) + } + return store +} + +func upsertOne(t *testing.T, store Store, upstreamID string, now time.Time, ttl time.Duration, candidate proxyDomain.Proxy) { + t.Helper() + result, err := store.UpsertFetched(context.Background(), upstreamID, activitypool.FetchedBatch{ + ObservedAt: now, ConfiguredTTL: ttl, MaxSize: 500, Proxies: []proxyDomain.Proxy{candidate}, + }) + if err != nil || result.Inserted != 1 { + t.Fatalf("UpsertFetched(%s) = %+v, %v", candidate.ID, result, err) + } +} + +func seedAvailable(t *testing.T, store Store, upstreamID string, now time.Time, ttl time.Duration, candidate proxyDomain.Proxy) { + t.Helper() + candidate.State = proxyDomain.StateAvailable + upsertOne(t, store, upstreamID, now, ttl, candidate) +} + +func assertInventory(t *testing.T, store Store, upstreamID string, now time.Time, want int) { + t.Helper() + inventory, err := store.Inventory(context.Background(), upstreamID, now) + if err != nil || inventory.UpstreamID != upstreamID || inventory.Managed != want { + t.Fatalf("Inventory(%s) = %+v, %v; want %d", upstreamID, inventory, err, want) + } +} + +func contractProxy(id, host string, state proxyDomain.State) proxyDomain.Proxy { + return proxyDomain.Proxy{ + ID: id, Scheme: proxyDomain.SchemeHTTP, Host: host, Port: 8080, State: state, + Tags: map[string]string{"region": "cn", "carrier": "ct"}, + } +} + +func contractNow() time.Time { + return time.Now().UTC().Add(time.Hour).Truncate(time.Millisecond) +} diff --git a/internal/domain/activitypool/pool.go b/internal/domain/activitypool/pool.go index 921f0d5..c8cd5ec 100644 --- a/internal/domain/activitypool/pool.go +++ b/internal/domain/activitypool/pool.go @@ -387,6 +387,7 @@ func (p *MemoryPool) Extract(ctx context.Context, command extractionDomain.Comma } } if command.Requested == 0 { + p.rememberExtractionLocked(idempotencyKey, command, result) return result, nil } @@ -421,20 +422,29 @@ func (p *MemoryPool) Extract(ctx context.Context, command extractionDomain.Comma if result.Returned > 0 { result.ExtractedAt = command.Now.UTC() } - if command.IdempotencyKey != "" { - expiresAt := command.Now.Add(idempotencyTTL(command.IdempotencyTTL)) - for _, item := range result.Items { - if !item.ExpiresAt.IsZero() && item.ExpiresAt.Before(expiresAt) { - expiresAt = item.ExpiresAt - } - } - if expiresAt.After(command.Now) { - p.idempotent[idempotencyKey] = idempotencyEntry{ - command: cloneCommand(command), result: cloneResult(result), expiresAt: expiresAt, - } + p.rememberExtractionLocked(idempotencyKey, command, result) + return result, nil +} + +func (p *MemoryPool) rememberExtractionLocked( + idempotencyKey string, + command extractionDomain.Command, + result extractionDomain.Result, +) { + if command.IdempotencyKey == "" { + return + } + expiresAt := command.Now.Add(idempotencyTTL(command.IdempotencyTTL)) + for _, item := range result.Items { + if !item.ExpiresAt.IsZero() && item.ExpiresAt.Before(expiresAt) { + expiresAt = item.ExpiresAt + } + } + if expiresAt.After(command.Now) { + p.idempotent[idempotencyKey] = idempotencyEntry{ + command: cloneCommand(command), result: cloneResult(result), expiresAt: expiresAt, } } - return result, nil } func (p *MemoryPool) Assign(ctx context.Context, now time.Time, proxyID, workerID string, ttl time.Duration) (ownershipDomain.Assignment, error) { @@ -702,7 +712,7 @@ func extractionCandidate(entry Entry) extractionDomain.Candidate { ID: entry.Proxy.ID, Protocol: string(entry.Proxy.Scheme), Host: entry.Proxy.Host, Port: entry.Proxy.Port, Username: entry.Proxy.Username, Region: entry.Proxy.Tags["region"], Carrier: entry.Proxy.Tags["carrier"], Upstream: entry.Proxy.SourceUpstream, - OwnerWorkerID: entry.OwnerWorkerID, State: extractionDomain.Available, + OwnerWorkerID: entry.OwnerWorkerID, State: extractionDomain.Extracted, ExpiresAt: expiresAt, LastCheckedAt: checkedAt, } } @@ -781,16 +791,19 @@ func sameIdempotentRequest(left, right extractionDomain.Command) bool { } func equalSet(left, right []string) bool { - if len(left) != len(right) { + leftSet := make(map[string]struct{}, len(left)) + for _, value := range left { + leftSet[value] = struct{}{} + } + rightSet := make(map[string]struct{}, len(right)) + for _, value := range right { + rightSet[value] = struct{}{} + } + if len(leftSet) != len(rightSet) { return false } - counts := make(map[string]int, len(left)) - for _, value := range left { - counts[value]++ - } - for _, value := range right { - counts[value]-- - if counts[value] < 0 { + for value := range leftSet { + if _, ok := rightSet[value]; !ok { return false } } diff --git a/scripts/test-redis.ps1 b/scripts/test-redis.ps1 index bc8d60f..c2c4220 100644 --- a/scripts/test-redis.ps1 +++ b/scripts/test-redis.ps1 @@ -13,7 +13,7 @@ try { $env:PROXY_POOL_TEST_REDIS_URL = "redis://127.0.0.1:16379/15" Push-Location $repositoryRoot try { - go test -count=1 -tags=integration -timeout 60s ./internal/adapters/redisactivity + go test -count=1 -tags=integration -timeout 60s ./internal/adapters/redisactivity/... if ($LASTEXITCODE -ne 0) { throw "Redis integration tests failed with exit code $LASTEXITCODE" }