diff --git a/internal/adapters/redisactivity/extract.go b/internal/adapters/redisactivity/extract.go new file mode 100644 index 0000000..17cc155 --- /dev/null +++ b/internal/adapters/redisactivity/extract.go @@ -0,0 +1,208 @@ +package redisactivity + +import ( + "context" + "crypto/sha256" + "encoding/hex" + "encoding/json" + "errors" + "fmt" + "net" + "net/url" + "sort" + "strconv" + "time" + + extractionDomain "proxy-pool/internal/domain/extraction" +) + +const defaultRedisIdempotencyTTL = 5 * time.Minute + +type extractionDigestInput struct { + Requested int `json:"requested"` + Fulfillment extractionDomain.Fulfillment `json:"fulfillment"` + Protocols []string `json:"protocols"` + Regions []string `json:"regions"` + Carriers []string `json:"carriers"` + Upstreams []string `json:"upstreams"` +} + +type extractionFilterWire struct { + Protocols []string `json:"protocols"` + Regions []string `json:"regions"` + Carriers []string `json:"carriers"` + Upstreams []string `json:"upstreams"` +} + +var _ extractionDomain.Store = (*Adapter)(nil) + +func (a *Adapter) Extract(ctx context.Context, command extractionDomain.Command) (extractionDomain.Result, error) { + result := extractionDomain.Result{Requested: command.Requested} + if ctx == nil { + return result, extractionDomain.ErrInvalidCommand + } + if err := ctx.Err(); err != nil { + return result, err + } + if a == nil || command.Now.IsZero() || command.Requested < 0 || command.ReserveForGateway < 0 || + command.MinRemainingTTL < 0 || command.MaxHealthCheckAge < 0 || command.IdempotencyTTL < 0 || + (command.IdempotencyKey != "" && command.ClientID == "") || + (command.Fulfillment != extractionDomain.Partial && command.Fulfillment != extractionDomain.AllOrNothing) { + return result, extractionDomain.ErrInvalidCommand + } + + digestInput := extractionDigestInput{ + Requested: command.Requested, Fulfillment: command.Fulfillment, + Protocols: canonicalFilter(command.Protocols), Regions: canonicalFilter(command.Regions), + Carriers: canonicalFilter(command.Carriers), Upstreams: canonicalFilter(command.Upstreams), + } + requestDigest, err := extractionRequestDigest(digestInput) + if err != nil { + return result, err + } + if command.Requested == 0 && command.IdempotencyKey == "" { + return result, nil + } + operationID := command.RequestID + if operationID == "" { + operationID, err = newOperationID() + if err != nil { + return result, err + } + } + operationKey := a.keys.operation(digestParts(command.ClientID, operationID)) + idempotencyKey := operationKey + hasIdempotency := 0 + if command.IdempotencyKey != "" { + hasIdempotency = 1 + idempotencyKey = a.keys.idempotency(command.ClientID, command.IdempotencyKey) + } + filterPayload, err := json.Marshal(extractionFilterWire{ + Protocols: digestInput.Protocols, Regions: digestInput.Regions, + Carriers: digestInput.Carriers, Upstreams: digestInput.Upstreams, + }) + if err != nil { + return result, fmt.Errorf("encode Redis extraction filters: %w", err) + } + keys := []string{ + a.keys.records, a.keys.unique, a.keys.idkeys, a.keys.expiry, a.keys.available, + a.keys.inventory, a.keys.owners, a.keys.ownerExpiry, operationKey, idempotencyKey, + } + keys = append(keys, a.extractionDriverKeys(digestInput)...) + idempotencyTTL := command.IdempotencyTTL + if idempotencyTTL == 0 { + idempotencyTTL = defaultRedisIdempotencyTTL + } + scriptResult, err := runScript(ctx, a.client, extractScript, keys, + command.Now.UnixMilli(), command.Requested, string(command.Fulfillment), command.ReserveForGateway, + durationMillis(command.MinRemainingTTL), durationMillis(command.MaxHealthCheckAge), + a.options.MaxCandidateScan, a.options.CleanupLimit, durationMillis(idempotencyTTL), + operationTTLMillis(a.options.OperationTTL), requestDigest, hasIdempotency, string(filterPayload)) + if err != nil { + return result, err + } + var reply extractScriptReply + if err := decodeScriptResult(scriptResult, &reply); err != nil { + return result, err + } + if reply.RequestDigest != requestDigest { + return result, invalidScriptReply("extraction reply digest mismatch") + } + switch reply.Status { + case scriptConflict: + return result, extractionDomain.ErrIdempotencyConflict + case scriptInsufficient: + return result, extractionDomain.ErrInsufficientProxies + case scriptUnavailable: + return result, extractionDomain.ErrStoreUnavailable + case scriptInvalid: + return result, extractionDomain.ErrInvalidCommand + case scriptOK: + if reply.Record == "" { + return result, invalidScriptReply("extraction reply omitted record") + } + committed, err := decodeIdempotencyRecord(reply.Record) + if err != nil { + return result, errors.Join( + invalidScriptReply("extraction reply contained an invalid record"), + fmt.Errorf("decode extraction record: %w", err), + ) + } + if committed.RequestDigest != requestDigest { + return result, invalidScriptReply("extraction reply contained an invalid record") + } + return buildExtractionResult(committed.Result), nil + default: + return result, invalidScriptReply("unexpected extraction status") + } +} + +func (a *Adapter) extractionDriverKeys(input extractionDigestInput) []string { + keys := make([]string, 0, 4) + if len(input.Protocols) == 1 { + keys = append(keys, a.keys.protocol(input.Protocols[0])) + } + if len(input.Regions) == 1 { + keys = append(keys, a.keys.region(input.Regions[0])) + } + if len(input.Carriers) == 1 { + keys = append(keys, a.keys.carrier(input.Carriers[0])) + } + if len(input.Upstreams) == 1 { + keys = append(keys, a.keys.upstream(input.Upstreams[0])) + } + return keys +} + +func extractionRequestDigest(input extractionDigestInput) (string, error) { + payload, err := json.Marshal(input) + if err != nil { + return "", fmt.Errorf("encode extraction request digest: %w", err) + } + digest := sha256.Sum256(payload) + return hex.EncodeToString(digest[:]), nil +} + +func canonicalFilter(values []string) []string { + if len(values) == 0 { + return []string{} + } + unique := make(map[string]struct{}, len(values)) + for _, value := range values { + unique[value] = struct{}{} + } + result := make([]string, 0, len(unique)) + for value := range unique { + result = append(result, value) + } + sort.Strings(result) + return result +} + +func durationMillis(duration time.Duration) int64 { + if duration <= 0 { + return 0 + } + return operationTTLMillis(duration) +} + +func buildExtractionResult(result extractionDomain.Result) extractionDomain.Result { + result.Items = append([]extractionDomain.Candidate(nil), result.Items...) + for index := range result.Items { + result.Items[index].URL = proxyURL(result.Items[index]) + } + return result +} + +func proxyURL(candidate extractionDomain.Candidate) string { + parsed := url.URL{ + Scheme: candidate.Protocol, + Host: net.JoinHostPort(candidate.Host, strconv.FormatUint(uint64(candidate.Port), 10)), + } + if candidate.Password != "" { + parsed.User = url.UserPassword(candidate.Username, candidate.Password) + } else if candidate.Username != "" { + parsed.User = url.User(candidate.Username) + } + return parsed.String() +} diff --git a/internal/adapters/redisactivity/extract_integration_test.go b/internal/adapters/redisactivity/extract_integration_test.go new file mode 100644 index 0000000..3eb9f7a --- /dev/null +++ b/internal/adapters/redisactivity/extract_integration_test.go @@ -0,0 +1,455 @@ +//go:build integration + +package redisactivity + +import ( + "context" + "encoding/json" + "errors" + "fmt" + "net/url" + "sync" + "testing" + "time" + + "proxy-pool/internal/domain/activitypool" + extractionDomain "proxy-pool/internal/domain/extraction" + proxyDomain "proxy-pool/internal/domain/proxy" + "proxy-pool/internal/platform/credentials" +) + +func TestRedisExtractAppliesEveryFilterAndBuildsCredentialURL(t *testing.T) { + fixture := newRedisTestFixture(t) + now := redisTestNow() + reference, err := fixture.Credentials.Put(context.Background(), "target", credentials.Value{ + Username: "user@tenant", Password: "p:/@ss", + }) + if err != nil { + t.Fatalf("Credentials.Put(): %v", err) + } + target := testProxy("target", "192.0.2.10") + target.Username = "user@tenant" + target.SecretRef = reference.SecretRef + target.CredentialVersion = reference.CredentialVersion + target.Tags = map[string]string{"region": "cn", "carrier": "ct"} + seedRedisAvailable(t, fixture.Adapter, "provider-a", now, now.Add(2*time.Second), time.Minute, target) + + other := testProxy("other", "192.0.2.11") + other.Scheme = proxyDomain.SchemeSOCKS5 + other.Tags = map[string]string{"region": "us", "carrier": "cu"} + seedRedisAvailable(t, fixture.Adapter, "provider-b", now, now.Add(2*time.Second), time.Minute, other) + + result, err := fixture.Adapter.Extract(context.Background(), extractionDomain.Command{ + RequestID: "req-filter", ClientID: "client-a", Requested: 2, + Fulfillment: extractionDomain.Partial, Now: now.Add(3 * 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 != "target" { + t.Fatalf("Extract(filtered) = %+v, %v", result, err) + } + parsed, err := url.Parse(result.Items[0].URL) + if err != nil { + t.Fatalf("parse extracted URL: %v", err) + } + password, hasPassword := parsed.User.Password() + if parsed.User.Username() != "user@tenant" || !hasPassword || password != "p:/@ss" { + t.Fatalf("URL credentials = (%q, %q, %t)", parsed.User.Username(), password, hasPassword) + } +} + +func TestRedisExtractSupportsPartialAndAllOrNothing(t *testing.T) { + fixture := newRedisTestFixture(t) + now := redisTestNow() + candidate := testProxy("proxy-a", "192.0.2.10") + seedRedisAvailable(t, fixture.Adapter, "provider-a", now, now.Add(2*time.Second), time.Minute, candidate) + + _, err := fixture.Adapter.Extract(context.Background(), extractionDomain.Command{ + RequestID: "req-all", ClientID: "client-a", Requested: 2, + Fulfillment: extractionDomain.AllOrNothing, Now: now.Add(3 * time.Second), + }) + if !errors.Is(err, extractionDomain.ErrInsufficientProxies) { + t.Fatalf("Extract(allOrNothing) error = %v", err) + } + result, err := fixture.Adapter.Extract(context.Background(), extractionDomain.Command{ + RequestID: "req-partial", ClientID: "client-a", Requested: 2, + Fulfillment: extractionDomain.Partial, Now: now.Add(3 * time.Second), + }) + if err != nil || result.Returned != 1 || result.Items[0].ID != "proxy-a" { + t.Fatalf("Extract(partial) = %+v, %v", result, err) + } +} + +func TestRedisExtractReturnsEmptyItemsForValidZeroResults(t *testing.T) { + t.Run("gateway reserve", func(t *testing.T) { + fixture := newRedisTestFixture(t) + now := redisTestNow() + seedRedisAvailable(t, fixture.Adapter, "provider-a", now, now.Add(time.Second), time.Minute, + testProxy("proxy-a", "192.0.2.10")) + + result, err := fixture.Adapter.Extract(context.Background(), extractionDomain.Command{ + RequestID: "req-empty-reserve", ClientID: "client-a", Requested: 1, + Fulfillment: extractionDomain.Partial, Now: now.Add(2 * time.Second), ReserveForGateway: 1, + }) + if err != nil || result.Returned != 0 || len(result.Items) != 0 { + t.Fatalf("Extract(reserved empty) = %+v, %v", result, err) + } + }) + + t.Run("zero requested with idempotency", func(t *testing.T) { + fixture := newRedisTestFixture(t) + now := redisTestNow() + command := extractionDomain.Command{ + RequestID: "req-empty-zero", ClientID: "client-a", IdempotencyKey: "idem-empty-zero", + Requested: 0, Fulfillment: extractionDomain.Partial, Now: now, + } + result, err := fixture.Adapter.Extract(context.Background(), command) + if err != nil || result.Requested != 0 || result.Returned != 0 || len(result.Items) != 0 { + t.Fatalf("Extract(zero requested) = %+v, %v", result, err) + } + command.RequestID = "req-empty-zero-replay" + command.Now = now.Add(time.Second) + replayed, err := fixture.Adapter.Extract(context.Background(), command) + if err != nil || replayed.Requested != 0 || replayed.Returned != 0 || len(replayed.Items) != 0 { + t.Fatalf("Extract(zero requested replay) = %+v, %v", replayed, err) + } + command.RequestID = "req-empty-zero-conflict" + command.Requested = 1 + if _, err := fixture.Adapter.Extract(context.Background(), command); !errors.Is(err, extractionDomain.ErrIdempotencyConflict) { + t.Fatalf("Extract(zero requested conflict) error = %v", err) + } + }) +} + +func TestRedisExtractAppliesTTLHealthAgeAndGatewayReserve(t *testing.T) { + fixture := newRedisTestFixture(t) + now := redisTestNow() + observedAt := now.Add(-30 * time.Second) + + short := testProxy("short", "192.0.2.10") + seedRedisAvailable(t, fixture.Adapter, "provider-a", observedAt, now.Add(-time.Second), 35*time.Second, short) + stale := testProxy("stale", "192.0.2.11") + seedRedisAvailable(t, fixture.Adapter, "provider-a", observedAt, now.Add(-10*time.Second), 2*time.Minute, stale) + fresh := testProxy("fresh", "192.0.2.12") + seedRedisAvailable(t, fixture.Adapter, "provider-a", observedAt, now.Add(-time.Second), 2*time.Minute, fresh) + + result, err := fixture.Adapter.Extract(context.Background(), extractionDomain.Command{ + RequestID: "req-policy", ClientID: "client-a", Requested: 2, + Fulfillment: extractionDomain.Partial, Now: now, + MinRemainingTTL: 30 * time.Second, MaxHealthCheckAge: 5 * time.Second, + }) + if err != nil || result.Returned != 1 || result.Items[0].ID != "fresh" { + t.Fatalf("Extract(policy) = %+v, %v", result, err) + } + + reserveFixture := newRedisTestFixture(t) + for index := range 3 { + candidate := testProxy(fmt.Sprintf("reserve-%d", index), fmt.Sprintf("192.0.2.%d", index+20)) + seedRedisAvailable(t, reserveFixture.Adapter, "provider-a", now, now.Add(time.Second), time.Minute, candidate) + } + reserved, err := reserveFixture.Adapter.Extract(context.Background(), extractionDomain.Command{ + RequestID: "req-reserve", ClientID: "client-a", Requested: 2, + Fulfillment: extractionDomain.Partial, Now: now.Add(2 * time.Second), ReserveForGateway: 2, + }) + if err != nil || reserved.Returned != 1 { + t.Fatalf("Extract(reserve) = %+v, %v", reserved, err) + } +} + +func TestRedisExtractIdempotencyReplayConflictAndExpiryBound(t *testing.T) { + fixture := newRedisTestFixture(t) + now := redisTestNow() + seedRedisAvailable(t, fixture.Adapter, "provider-a", now, now.Add(time.Second), time.Minute, + testProxy("proxy-a", "192.0.2.10")) + command := extractionDomain.Command{ + RequestID: "req-first", ClientID: "client-a", IdempotencyKey: "idem-12345678", + IdempotencyTTL: 5 * time.Minute, Requested: 1, + Fulfillment: extractionDomain.Partial, Now: now.Add(2 * time.Second), + } + first, err := fixture.Adapter.Extract(context.Background(), command) + if err != nil || first.Returned != 1 { + t.Fatalf("first Extract() = %+v, %v", first, err) + } + command.RequestID = "req-replay" + command.Now = now.Add(10 * time.Second) + replayed, err := fixture.Adapter.Extract(context.Background(), command) + if err != nil || replayed.Returned != 1 || !replayed.ExtractedAt.Equal(first.ExtractedAt) { + t.Fatalf("replayed Extract() = %+v, %v", replayed, err) + } + command.RequestID = "req-conflict" + command.Requested = 2 + if _, err := fixture.Adapter.Extract(context.Background(), command); !errors.Is(err, extractionDomain.ErrIdempotencyConflict) { + t.Fatalf("conflicting Extract() error = %v", err) + } + + expiryFixture := newRedisTestFixture(t) + realNow := time.Now().UTC().Truncate(time.Millisecond) + shortLived := testProxy("short-lived", "192.0.2.30") + shortLived.State = proxyDomain.StateAvailable + if _, err := expiryFixture.Adapter.UpsertFetched(context.Background(), "provider-a", activitypool.FetchedBatch{ + ObservedAt: realNow, ConfiguredTTL: 2 * time.Second, MaxSize: 10, + Proxies: []proxyDomain.Proxy{shortLived}, + }); err != nil { + t.Fatalf("UpsertFetched(short-lived): %v", err) + } + expiryCommand := extractionDomain.Command{ + RequestID: "req-expiry-first", ClientID: "client-a", IdempotencyKey: "idem-expiry", + IdempotencyTTL: time.Minute, Requested: 1, + Fulfillment: extractionDomain.Partial, Now: realNow.Add(time.Millisecond), + } + original, err := expiryFixture.Adapter.Extract(context.Background(), expiryCommand) + if err != nil || original.Returned != 1 { + t.Fatalf("Extract(short-lived) = %+v, %v", original, err) + } + time.Sleep(2200 * time.Millisecond) + replacementAt := time.Now().UTC().Truncate(time.Millisecond) + if _, err := expiryFixture.Adapter.UpsertFetched(context.Background(), "provider-a", activitypool.FetchedBatch{ + ObservedAt: replacementAt, ConfiguredTTL: time.Minute, MaxSize: 10, + Proxies: []proxyDomain.Proxy{shortLived}, + }); err != nil { + t.Fatalf("UpsertFetched(replacement): %v", err) + } + expiryCommand.RequestID = "req-expiry-second" + expiryCommand.Now = replacementAt.Add(time.Millisecond) + again, err := expiryFixture.Adapter.Extract(context.Background(), expiryCommand) + if err != nil || again.Returned != 1 || again.ExtractedAt.Equal(original.ExtractedAt) { + t.Fatalf("Extract(after idempotency expiry) = %+v, %v", again, err) + } +} + +func TestRedisExtractIdempotencyDigestCanonicalizesBusinessFilters(t *testing.T) { + fixture := newRedisTestFixture(t) + now := redisTestNow() + candidate := testProxy("proxy-a", "192.0.2.10") + candidate.Tags = map[string]string{"region": "cn", "carrier": "ct"} + seedRedisAvailable(t, fixture.Adapter, "provider-a", now, now.Add(time.Second), time.Minute, candidate) + + command := extractionDomain.Command{ + RequestID: "req-canonical-first", ClientID: "client-a", SourceIP: "192.0.2.100", + IdempotencyKey: "idem-canonical", IdempotencyTTL: time.Minute, + Requested: 1, Fulfillment: extractionDomain.Partial, Now: now.Add(2 * 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 := fixture.Adapter.Extract(context.Background(), command) + if err != nil || first.Returned != 1 { + t.Fatalf("first Extract() = %+v, %v", first, err) + } + + command.RequestID = "req-canonical-replay" + command.SourceIP = "198.51.100.200" + command.Now = now.Add(10 * 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 := fixture.Adapter.Extract(context.Background(), command) + if err != nil || replayed.Returned != 1 || !replayed.ExtractedAt.Equal(first.ExtractedAt) || + replayed.Items[0].ID != first.Items[0].ID { + t.Fatalf("replayed Extract() = %+v, %v", replayed, err) + } +} + +func TestRedisExtractReplaysTheCommittedRequestOperation(t *testing.T) { + fixture := newRedisTestFixture(t) + now := redisTestNow() + seedRedisAvailable(t, fixture.Adapter, "provider-a", now, now.Add(time.Second), time.Minute, + testProxy("proxy-a", "192.0.2.10")) + command := extractionDomain.Command{ + RequestID: "req-operation-replay", ClientID: "client-a", Requested: 1, + Fulfillment: extractionDomain.Partial, Now: now.Add(2 * time.Second), + } + + first, err := fixture.Adapter.Extract(context.Background(), command) + if err != nil || first.Returned != 1 { + t.Fatalf("first Extract() = %+v, %v", first, err) + } + command.Now = now.Add(3 * time.Second) + replayed, err := fixture.Adapter.Extract(context.Background(), command) + if err != nil || replayed.Returned != 1 || !replayed.ExtractedAt.Equal(first.ExtractedAt) || + replayed.Items[0].ID != first.Items[0].ID { + t.Fatalf("replayed Extract() = %+v, %v", replayed, err) + } +} + +func TestRedisExtractIsExclusiveUnderConcurrentRaces(t *testing.T) { + fixture := newRedisTestFixture(t) + now := redisTestNow() + for iteration := range 100 { + candidate := testProxy(fmt.Sprintf("race-%d", iteration), fmt.Sprintf("198.51.100.%d", iteration+1)) + candidate.State = proxyDomain.StateAvailable + if _, err := fixture.Adapter.UpsertFetched(context.Background(), "provider-a", activitypool.FetchedBatch{ + ObservedAt: now, ConfiguredTTL: 10 * time.Minute, MaxSize: 10, + Proxies: []proxyDomain.Proxy{candidate}, + }); err != nil { + t.Fatalf("iteration %d UpsertFetched(): %v", iteration, err) + } + + 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 := fixture.Adapter.Extract(context.Background(), extractionDomain.Command{ + RequestID: fmt.Sprintf("req-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() + close(results) + close(errorsCh) + for err := range errorsCh { + if err != nil { + t.Fatalf("iteration %d Extract(): %v", iteration, err) + } + } + returned := 0 + for result := range results { + returned += result.Returned + } + if returned != 1 { + t.Fatalf("iteration %d total returned = %d, want 1", iteration, returned) + } + } +} + +func TestRedisExtractScanBudgetExhaustionDoesNotMutateCandidates(t *testing.T) { + fixture := newRedisTestFixture(t) + now := redisTestNow() + for index, ttl := range []time.Duration{3 * time.Minute, 2 * time.Minute, time.Minute} { + candidate := testProxy(fmt.Sprintf("scan-%d", index), fmt.Sprintf("203.0.113.%d", index+1)) + candidate.State = proxyDomain.StateAvailable + candidate.Tags["region"] = "none" + if index == 2 { + candidate.Tags["region"] = "target" + } + if _, err := fixture.Adapter.UpsertFetched(context.Background(), "provider-a", activitypool.FetchedBatch{ + ObservedAt: now, ConfiguredTTL: ttl, MaxSize: 10, Proxies: []proxyDomain.Proxy{candidate}, + }); err != nil { + t.Fatalf("UpsertFetched(scan-%d): %v", index, err) + } + } + bounded, err := New(fixture.Client, Options{ + Namespace: fixture.Namespace, Credentials: fixture.Credentials, + OperationTTL: time.Minute, MaxCandidateScan: 2, CleanupLimit: 16, + }) + if err != nil { + t.Fatalf("New(bounded): %v", err) + } + _, err = bounded.Extract(context.Background(), extractionDomain.Command{ + RequestID: "req-scan", ClientID: "client-a", Requested: 1, + Fulfillment: extractionDomain.AllOrNothing, Now: now.Add(time.Second), + Regions: []string{"target", "other"}, + }) + if !errors.Is(err, extractionDomain.ErrStoreUnavailable) { + t.Fatalf("Extract(scan exhausted) error = %v", err) + } + + remaining, err := fixture.Adapter.Extract(context.Background(), extractionDomain.Command{ + RequestID: "req-after-scan", ClientID: "client-a", Requested: 3, + Fulfillment: extractionDomain.AllOrNothing, Now: now.Add(time.Second), + }) + if err != nil || remaining.Returned != 3 { + t.Fatalf("Extract(after scan exhaustion) = %+v, %v", remaining, err) + } +} + +func TestRedisExtractDoesNotConsumeIncompleteRecords(t *testing.T) { + tests := []struct { + name string + mutate func(map[string]any) + }{ + {name: "empty host", mutate: func(record map[string]any) { record["host"] = "" }}, + {name: "missing creation time", mutate: func(record map[string]any) { record["createdAtMs"] = float64(0) }}, + {name: "negative latency", mutate: func(record map[string]any) { record["latencyNs"] = float64(-1) }}, + {name: "negative concurrency", mutate: func(record map[string]any) { record["maxConcurrency"] = float64(-1) }}, + {name: "invalid state", mutate: func(record map[string]any) { record["state"] = "BROKEN" }}, + {name: "invalid credential version", mutate: func(record map[string]any) { record["credentialVersion"] = float64(1) }}, + {name: "invalid last success", mutate: func(record map[string]any) { record["lastSuccessAtMs"] = float64(-1) }}, + {name: "foreign index key", mutate: func(record map[string]any) { record["indexKeys"] = []any{"pp:{other}:available"} }}, + } + for _, test := range tests { + t.Run(test.name, func(t *testing.T) { + fixture := newRedisTestFixture(t) + now := redisTestNow() + seedRedisAvailable(t, fixture.Adapter, "provider-a", now, now.Add(time.Second), time.Minute, + testProxy("corrupt", "192.0.2.10")) + + raw, err := fixture.Client.HGet(context.Background(), fixture.Adapter.keys.records, "corrupt").Result() + if err != nil { + t.Fatalf("HGet(corrupt): %v", err) + } + var stored map[string]any + if err := json.Unmarshal([]byte(raw), &stored); err != nil { + t.Fatalf("decode stored record: %v", err) + } + test.mutate(stored) + expectedState := stored["state"] + corrupted, err := json.Marshal(stored) + if err != nil { + t.Fatalf("encode corrupt record: %v", err) + } + if err := fixture.Client.HSet(context.Background(), fixture.Adapter.keys.records, "corrupt", corrupted).Err(); err != nil { + t.Fatalf("HSet(corrupt): %v", err) + } + + result, err := fixture.Adapter.Extract(context.Background(), extractionDomain.Command{ + RequestID: "req-corrupt", ClientID: "client-a", Requested: 1, + Fulfillment: extractionDomain.Partial, Now: now.Add(2 * time.Second), + }) + if err != nil || result.Returned != 0 || len(result.Items) != 0 { + t.Fatalf("Extract(corrupt) = %+v, %v", result, err) + } + + after, err := fixture.Client.HGet(context.Background(), fixture.Adapter.keys.records, "corrupt").Result() + if err != nil { + t.Fatalf("HGet(corrupt after extraction): %v", err) + } + if err := json.Unmarshal([]byte(after), &stored); err != nil { + t.Fatalf("decode record after extraction: %v", err) + } + if stored["state"] != expectedState { + t.Fatalf("corrupt record state = %q, want unchanged %q", stored["state"], expectedState) + } + }) + } +} + +func seedRedisAvailable( + t *testing.T, + adapter *Adapter, + upstreamID string, + observedAt time.Time, + availableAt time.Time, + ttl time.Duration, + candidate proxyDomain.Proxy, +) { + t.Helper() + candidate.State = proxyDomain.StateFetched + if _, err := adapter.UpsertFetched(context.Background(), upstreamID, activitypool.FetchedBatch{ + ObservedAt: observedAt, ConfiguredTTL: ttl, MaxSize: 100, + Proxies: []proxyDomain.Proxy{candidate}, + }); err != nil { + t.Fatalf("UpsertFetched(%s): %v", candidate.ID, err) + } + if _, err := adapter.ApplyHealth(context.Background(), activitypool.HealthUpdate{ + ProxyID: candidate.ID, CheckedAt: availableAt.Add(-time.Millisecond), NextState: proxyDomain.StateChecking, + }); err != nil { + t.Fatalf("ApplyHealth(%s, checking): %v", candidate.ID, err) + } + if _, err := adapter.ApplyHealth(context.Background(), activitypool.HealthUpdate{ + ProxyID: candidate.ID, CheckedAt: availableAt, NextState: proxyDomain.StateAvailable, + }); err != nil { + t.Fatalf("ApplyHealth(%s, available): %v", candidate.ID, err) + } +} diff --git a/internal/adapters/redisactivity/scripts.go b/internal/adapters/redisactivity/scripts.go index 5042017..1bf04ac 100644 --- a/internal/adapters/redisactivity/scripts.go +++ b/internal/adapters/redisactivity/scripts.go @@ -40,8 +40,9 @@ type healthScriptReply struct { } type extractScriptReply struct { - Status scriptStatus `json:"status"` - Record string `json:"record,omitempty"` + Status scriptStatus `json:"status"` + RequestDigest string `json:"requestDigest"` + Record string `json:"record,omitempty"` } type ownershipScriptReply struct { @@ -60,9 +61,13 @@ var upsertSource string //go:embed scripts/health.lua var healthSource string +//go:embed scripts/extract.lua +var extractSource string + var ( - upsertScript = redis.NewScript(upsertSource) - healthScript = redis.NewScript(healthSource) + upsertScript = redis.NewScript(upsertSource) + healthScript = redis.NewScript(healthSource) + extractScript = redis.NewScript(extractSource) ) func runScript(ctx context.Context, client redis.Scripter, script *redis.Script, keys []string, args ...any) (any, error) { diff --git a/internal/adapters/redisactivity/scripts/extract.lua b/internal/adapters/redisactivity/scripts/extract.lua new file mode 100644 index 0000000..c3e4b6f --- /dev/null +++ b/internal/adapters/redisactivity/scripts/extract.lua @@ -0,0 +1,339 @@ +local records_key = KEYS[1] +local unique_key = KEYS[2] +local idkeys_key = KEYS[3] +local expiry_key = KEYS[4] +local available_key = KEYS[5] +local inventory_key = KEYS[6] +local owners_key = KEYS[7] +local owner_expiry_key = KEYS[8] +local operation_key = KEYS[9] +local idempotency_key = KEYS[10] + +local now_ms = tonumber(ARGV[1]) +local requested = tonumber(ARGV[2]) +local fulfillment = ARGV[3] +local reserve = tonumber(ARGV[4]) +local min_remaining_ttl_ms = tonumber(ARGV[5]) +local max_health_age_ms = tonumber(ARGV[6]) +local max_candidate_scan = tonumber(ARGV[7]) +local cleanup_limit = tonumber(ARGV[8]) +local idempotency_ttl_ms = tonumber(ARGV[9]) +local operation_ttl_ms = tonumber(ARGV[10]) +local request_digest = ARGV[11] +local has_idempotency = tonumber(ARGV[12]) == 1 +local filters = cjson.decode(ARGV[13]) + +local function finish(reply, hard_expiry_ms) + local encoded = cjson.encode(reply) + redis.call('SET', operation_key, encoded, 'PX', operation_ttl_ms) + if hard_expiry_ms then + local operation_expiry_ms = redis.call('PEXPIRETIME', operation_key) + if operation_expiry_ms > hard_expiry_ms then + redis.call('PEXPIREAT', operation_key, hard_expiry_ms) + end + end + return encoded +end + +local committed = redis.call('GET', operation_key) +if committed then + local reply = cjson.decode(committed) + if reply.requestDigest == request_digest then + return committed + end + return cjson.encode({status = 'conflict', requestDigest = request_digest}) +end + +if has_idempotency then + local replay = redis.call('GET', idempotency_key) + if replay then + local replay_record = cjson.decode(replay) + if tonumber(replay_record.expiresAtMs) <= now_ms then + redis.call('DEL', idempotency_key) + elseif replay_record.requestDigest ~= request_digest then + return finish({status = 'conflict', requestDigest = request_digest}) + else + return finish({status = 'ok', requestDigest = request_digest, record = replay}, tonumber(replay_record.expiresAtMs)) + end + end +end + +local function is_managed(state) + return state == 'FETCHED' or state == 'CHECKING' or state == 'AVAILABLE' or + state == 'SUSPECT' or state == 'DRAINING' +end + +local function decrement_inventory(upstream) + if type(upstream) ~= 'string' or upstream == '' then + return + end + local value = redis.call('HINCRBY', inventory_key, upstream, -1) + if value < 0 then + redis.call('HSET', inventory_key, upstream, 0) + end +end + +local function remove_available(proxy_id, record) + redis.call('ZREM', available_key, proxy_id) + local index_keys = record and record.indexKeys + if type(index_keys) == 'table' then + for _, index_key in ipairs(index_keys) do + if type(index_key) == 'string' and index_key ~= '' then + redis.call('ZREM', index_key, proxy_id) + end + end + end +end + +local function remove_proxy(proxy_id) + local raw = redis.call('HGET', records_key, proxy_id) + local record = nil + if raw then + local decoded + decoded, record = pcall(cjson.decode, raw) + remove_available(proxy_id, decoded and record or nil) + if decoded and type(record) == 'table' and is_managed(record.state) then + decrement_inventory(record.sourceUpstream) + end + else + redis.call('ZREM', available_key, proxy_id) + end + local digest = redis.call('HGET', idkeys_key, proxy_id) + if digest and redis.call('HGET', unique_key, digest) == proxy_id then + redis.call('HDEL', unique_key, digest) + end + redis.call('HDEL', idkeys_key, proxy_id) + redis.call('HDEL', records_key, proxy_id) + redis.call('ZREM', expiry_key, proxy_id) + redis.call('HDEL', owners_key, proxy_id) + redis.call('ZREM', owner_expiry_key, proxy_id) +end + +local expired = redis.call('ZRANGEBYSCORE', expiry_key, '-inf', now_ms, 'LIMIT', 0, cleanup_limit) +for _, proxy_id in ipairs(expired) do + remove_proxy(proxy_id) +end + +local function to_set(values) + local result = {} + for _, value in ipairs(values or {}) do + result[value] = true + end + return result +end + +local protocol_filter = to_set(filters.protocols) +local region_filter = to_set(filters.regions) +local carrier_filter = to_set(filters.carriers) +local upstream_filter = to_set(filters.upstreams) + +local function matches(filter_values, filter_set, value) + return #filter_values == 0 or filter_set[value] == true +end + +local function valid_scheme(value) + return value == 'http' or value == 'https' or value == 'socks5' +end + +local function valid_optional_string(value) + return value == nil or type(value) == 'string' +end + +local function valid_integer(value) + return type(value) == 'number' and value == math.floor(value) +end + +local function valid_proxy_state(value) + return value == 'FETCHED' or value == 'CHECKING' or value == 'AVAILABLE' or + value == 'SUSPECT' or value == 'DRAINING' or value == 'UNHEALTHY' or + value == 'EXTRACTED' or value == 'EXPIRED' or value == 'REMOVED' +end + +local function valid_proxy_record(proxy_id, record) + if type(record) ~= 'table' or record.version ~= 1 or record.id ~= proxy_id or + not valid_scheme(record.scheme) or type(record.host) ~= 'string' or record.host == '' or + type(record.sourceUpstream) ~= 'string' or record.sourceUpstream == '' or + not valid_optional_string(record.username) or not valid_optional_string(record.password) or + not valid_optional_string(record.credentialVersion) or + not valid_optional_string(record.ownerWorkerId) or not valid_proxy_state(record.state) or + (record.tags ~= nil and type(record.tags) ~= 'table') or + (record.indexKeys ~= nil and type(record.indexKeys) ~= 'table') then + return false + end + if not valid_integer(record.port) or record.port <= 0 or record.port > 65535 or + not valid_integer(record.createdAtMs) or record.createdAtMs <= 0 or + not valid_integer(record.expiresAtMs) or record.expiresAtMs <= 0 or + not valid_integer(record.usableUntilMs) or record.usableUntilMs <= 0 or + record.usableUntilMs > record.expiresAtMs or + not valid_integer(record.latencyNs) or record.latencyNs < 0 or + not valid_integer(record.maxConcurrency) or record.maxConcurrency < 0 or + (record.lastCheckedAtMs ~= nil and + (not valid_integer(record.lastCheckedAtMs) or record.lastCheckedAtMs < 0)) or + (record.lastSuccessAtMs ~= nil and + (not valid_integer(record.lastSuccessAtMs) or record.lastSuccessAtMs < 0)) then + return false + end + for _, index_key in ipairs(record.indexKeys or {}) do + if type(index_key) ~= 'string' or index_key == '' or + not string.find(index_key, '{activity}', 1, true) then + return false + end + end + for tag_key, tag_value in pairs(record.tags or {}) do + if type(tag_key) ~= 'string' or type(tag_value) ~= 'string' then + return false + end + end + return true +end + +local function empty_result_record(expires_at_ms) + return '{"version":1,"requestDigest":' .. cjson.encode(request_digest) .. + ',"expiresAtMs":' .. tostring(expires_at_ms) .. + ',"result":{"requested":' .. tostring(requested) .. ',"returned":0,"items":[]}}' +end + +if requested == 0 then + local expires_at_ms = now_ms + idempotency_ttl_ms + local record = empty_result_record(expires_at_ms) + if has_idempotency then + redis.call('SET', idempotency_key, record) + redis.call('PEXPIREAT', idempotency_key, expires_at_ms) + end + return finish({status = 'ok', requestDigest = request_digest, record = record}, expires_at_ms) +end + +local driver_key = available_key +local driver_size = redis.call('ZCARD', available_key) +for index = 11, #KEYS do + local size = redis.call('ZCARD', KEYS[index]) + if size < driver_size then + driver_key = KEYS[index] + driver_size = size + end +end + +local candidate_ids = redis.call('ZREVRANGE', driver_key, 0, max_candidate_scan - 1) +local matches_found = {} +local required_matches = requested + reserve +local scanned = 0 + +for _, proxy_id in ipairs(candidate_ids) do + scanned = scanned + 1 + local raw = redis.call('HGET', records_key, proxy_id) + if not raw then + redis.call('ZREM', driver_key, proxy_id) + redis.call('ZREM', available_key, proxy_id) + else + local decoded, record = pcall(cjson.decode, raw) + if not decoded or not valid_proxy_record(proxy_id, record) then + remove_available(proxy_id, decoded and record or nil) + elseif tonumber(record.expiresAtMs) <= now_ms then + remove_proxy(proxy_id) + else + local owned = (record.ownerWorkerId and record.ownerWorkerId ~= '') or redis.call('HEXISTS', owners_key, proxy_id) == 1 + local usable = record.state == 'AVAILABLE' and not owned and tonumber(record.usableUntilMs) > now_ms + if not usable then + remove_available(proxy_id, record) + else + local tags = record.tags or {} + local health_fresh = max_health_age_ms == 0 or + (record.lastCheckedAtMs and now_ms - tonumber(record.lastCheckedAtMs) <= max_health_age_ms) + local ttl_eligible = tonumber(record.expiresAtMs) - now_ms >= min_remaining_ttl_ms + if health_fresh and ttl_eligible and + matches(filters.protocols, protocol_filter, record.scheme) and + matches(filters.regions, region_filter, tags.region or '') and + matches(filters.carriers, carrier_filter, tags.carrier or '') and + matches(filters.upstreams, upstream_filter, record.sourceUpstream) then + matches_found[#matches_found + 1] = {id = proxy_id, record = record} + if #matches_found >= required_matches then + break + end + end + end + end + end +end + +if #matches_found < required_matches and driver_size > scanned then + return finish({status = 'unavailable', requestDigest = request_digest}) +end + +local available_count = #matches_found - reserve +if available_count < 0 then + available_count = 0 +end +if fulfillment == 'allOrNothing' and available_count < requested then + return finish({status = 'insufficient', requestDigest = request_digest}) +end + +local selected_count = requested +if selected_count > available_count then + selected_count = available_count +end +local result_items = cjson.decode('[]') +local earliest_expiry_ms = nil + +for index = 1, selected_count do + local selected = matches_found[index] + local record = selected.record + remove_available(selected.id, record) + if is_managed(record.state) then + decrement_inventory(record.sourceUpstream) + end + record.state = 'EXTRACTED' + local encoded = cjson.encode(record) + redis.call('HSET', records_key, selected.id, encoded) + + local tags = record.tags or {} + local item = { + id = record.id, + protocol = record.scheme, + host = record.host, + port = record.port, + username = record.username, + password = record.password, + region = tags.region, + carrier = tags.carrier, + upstream = record.sourceUpstream, + ownerWorkerId = record.ownerWorkerId, + state = 'EXTRACTED', + expiresAtMs = record.expiresAtMs, + lastCheckedAtMs = record.lastCheckedAtMs, + } + result_items[#result_items + 1] = item + local hard_expiry_ms = tonumber(record.expiresAtMs) + if not earliest_expiry_ms or hard_expiry_ms < earliest_expiry_ms then + earliest_expiry_ms = hard_expiry_ms + end +end + +local result_expiry_ms = now_ms + idempotency_ttl_ms +if earliest_expiry_ms and earliest_expiry_ms < result_expiry_ms then + result_expiry_ms = earliest_expiry_ms +end +local result_value = { + requested = requested, + returned = selected_count, + items = result_items, +} +if selected_count > 0 then + result_value.extractedAtMs = now_ms +end +local result_record +if selected_count == 0 then + result_record = empty_result_record(result_expiry_ms) +else + result_record = cjson.encode({ + version = 1, + requestDigest = request_digest, + expiresAtMs = result_expiry_ms, + result = result_value, + }) +end + +if has_idempotency then + redis.call('SET', idempotency_key, result_record) + redis.call('PEXPIREAT', idempotency_key, result_expiry_ms) +end +return finish({status = 'ok', requestDigest = request_digest, record = result_record}, result_expiry_ms)