feat: add atomic redis proxy extraction

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()
+}

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)
+	}
+}

internal/adapters/redisactivity/scripts.go

@@ -41,6 +41,7 @@ type healthScriptReply struct {
 
 type extractScriptReply struct {
 	Status        scriptStatus `json:"status"`
+	RequestDigest string       `json:"requestDigest"`
 	Record        string       `json:"record,omitempty"`
 }
 
@@ -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)
+	extractScript = redis.NewScript(extractSource)
 )
 
 func runScript(ctx context.Context, client redis.Scripter, script *redis.Script, keys []string, args ...any) (any, error) {

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)