feat: validate extraction load workloads

2026-08-02 09:42:34 +08:00 · 2026-08-02 09:42:34 +08:00 · 97a515ddf0
commit 97a515ddf0
parent d04b7413eb
8 changed files with 358 additions and 74 deletions
--- a/README.md
+++ b/README.md
@ -198,7 +198,8 @@ go run ./cmd/proxy-loadgen `
 ```

 使用 `-duration 30s -rate 5000` 可运行限速场景；省略 `-rate` 时固定数量 worker
-会饱和发送。`-method`、重复的 `-header` 与 `-body` 可组合用于 Distribution 的提取接口：
+会饱和发送。普通 HTTP 场景中，`-method`、重复的 `-header` 与 `-body` 可组合用于
+Distribution 的提取接口：

 ```powershell
 go run ./cmd/proxy-loadgen `
@ -208,9 +209,22 @@ go run ./cmd/proxy-loadgen `
  -requests 1000 -concurrency 32 -timeout 10s
 ```

-命令输出 JSON 报告，包含成功/失败分类、`tunnelsEstablished`、固定内存的连接握手
-延迟分位上界、吞吐和 Go 运行时内存/GC 快照。它尚不包含 Extract 的专用数据准备/
-结果校验，也不构成 100,000 QPS 证明。
+`extract` 场景则自动构造 POST 请求、每请求独立的 `X-Request-ID` 与
+`Idempotency-Key`，并只校验响应的 `requestId`、数量和同响应内代理 ID 唯一性：
+
+```powershell
+go run ./cmd/proxy-loadgen `
+  -scenario extract `
+  -target http://CONTROLLER_HOST:8081/api/v1/proxies/extract `
+  -header "X-API-Key: DISTRIBUTION_API_KEY" `
+  -requests 1000 -concurrency 32 -timeout 10s `
+  -extract-count 1 -extract-fulfillment partial
+```
+
+命令输出 JSON 报告，包含成功/失败分类、`TunnelsEstablished`、
+`ExtractResponsesValidated`、`ExtractReturned`、`ExtractValidationFailures`、固定内存的
+连接握手延迟分位上界、吞吐和 Go 运行时内存/GC 快照。它不会输出提取响应中的地址或
+凭据，也不构成 100,000 QPS 证明。

 ## 关键配置与入口

--- a/cmd/proxy-loadgen/main.go
+++ b/cmd/proxy-loadgen/main.go
@ -30,8 +30,8 @@ func execute(ctx context.Context, args []string, run loadRun, stdout, stderr io.
 	flags.SetOutput(stderr)
 	targetURL := flags.String("target", "", "HTTP or HTTPS target URL")
 	proxyURL := flags.String("proxy", "", "optional HTTP or HTTPS forward proxy URL; connect requires HTTP")
-	scenario := flags.String("scenario", string(loadgen.ScenarioHTTP), "workload scenario: http or connect")
-	method := flags.String("method", http.MethodGet, "HTTP method")
+	scenario := flags.String("scenario", string(loadgen.ScenarioHTTP), "workload scenario: http, connect, or extract")
+	method := flags.String("method", "", "HTTP method; defaults to GET or POST for extract")
 	body := flags.String("body", "", "UTF-8 request body")
 	requests := flags.Int("requests", 0, "fixed request count; mutually exclusive with -duration")
 	duration := flags.Duration("duration", 0, "time-boxed workload duration")
@ -39,6 +39,8 @@ func execute(ctx context.Context, args []string, run loadRun, stdout, stderr io.
 	concurrency := flags.Int("concurrency", 64, "maximum concurrent requests")
 	timeout := flags.Duration("timeout", 10*time.Second, "HTTP request or CONNECT establishment timeout")
 	hold := flags.Duration("hold", 0, "CONNECT tunnel hold duration; required by -scenario connect")
+	extractCount := flags.Int("extract-count", 1, "proxies requested by each extract scenario request")
+	extractFulfillment := flags.String("extract-fulfillment", "partial", "extract fulfillment: partial or allOrNothing")
 	var headers headerValues
 	flags.Var(&headers, "header", "repeatable HTTP header in Name: Value form")
 	if err := flags.Parse(args); err != nil {
@ -59,7 +61,8 @@ func execute(ctx context.Context, args []string, run loadRun, stdout, stderr io.
 	report, err := run(ctx, loadgen.Options{
 		Scenario: loadgen.Scenario(*scenario), TargetURL: *targetURL, ProxyURL: *proxyURL, Method: *method, Headers: parsedHeaders, RequestBody: []byte(*body),
 		Requests: *requests, Duration: *duration, Rate: *rate, Concurrency: *concurrency, RequestTimeout: *timeout,
-		TunnelHold: *hold,
+		TunnelHold:   *hold,
+		ExtractCount: *extractCount, ExtractFulfillment: *extractFulfillment,
 	})
 	if err != nil {
 		if errors.Is(err, loadgen.ErrInvalidOptions) {
--- a/cmd/proxy-loadgen/main_test.go
+++ b/cmd/proxy-loadgen/main_test.go
@ -54,3 +54,19 @@ func TestExecutePassesCONNECTScenario(t *testing.T) {
 		t.Fatalf("execute() = %d; options=%+v stderr=%q", code, received, stderr.String())
 	}
 }
+
+func TestExecutePassesExtractScenario(t *testing.T) {
+	var received loadgen.Options
+	var stdout, stderr bytes.Buffer
+	code := execute(context.Background(), []string{
+		"-scenario", "extract", "-target", "http://controller.example:8081/api/v1/proxies/extract",
+		"-requests", "2", "-concurrency", "1", "-timeout", "2s", "-extract-count", "3", "-extract-fulfillment", "allOrNothing",
+	}, func(_ context.Context, options loadgen.Options) (loadgen.Report, error) {
+		received = options
+		return loadgen.Report{Requests: 2, Completed: 2, Succeeded: 2, Duration: time.Second}, nil
+	}, &stdout, &stderr)
+	if code != 0 || received.Scenario != loadgen.ScenarioExtract || received.Method != "" || received.ExtractCount != 3 ||
+		received.ExtractFulfillment != "allOrNothing" || stderr.Len() != 0 {
+		t.Fatalf("execute() = %d; options=%+v stderr=%q", code, received, stderr.String())
+	}
+}
--- a/docs/development/implementation-plan.md
+++ b/docs/development/implementation-plan.md
@ -299,10 +299,10 @@ Profile 在启用 Routing 与 Upstream 的组合上才进入调度。

 补充进度（2026-08-02）：已新增 `proxy-loadgen` HTTP 与 CONNECT 长连接场景。固定请求数
 和固定时长两种模式均通过固定 worker 数与有界派发通道执行，可选 QPS 限速；报告使用固定大小
-延迟直方图，输出状态分类、CONNECT 建立数、吞吐和 Go 内存/GC 快照。CONNECT 以原始 TCP
-握手连接 HTTP Gateway，建连成功后按 `hold` 保持，且不透明读取隧道内容。通用
-`method/header/body` 参数可覆盖 Distribution 提取 HTTP 请求；Extract 的专用数据准备与结果
-校验、故障注入以及代表性集群报告仍未实现。
+延迟直方图，输出状态分类、CONNECT 建立数、Extract 校验数、吞吐和 Go 内存/GC 快照。CONNECT
+以原始 TCP 握手连接 HTTP Gateway，建连成功后按 `hold` 保持，且不透明读取隧道内容。`extract`
+场景会自动生成独立 Request/Idempotency 标识，校验返回数量与单响应 ID 唯一性，且不记录地址或
+凭据。故障注入以及代表性集群报告仍未实现。

 ## Task 12: Machine-readable Contracts

--- a/docs/requirements/completion-audit.md
+++ b/docs/requirements/completion-audit.md
@ -100,7 +100,8 @@ Controller/Gateway 入口，完整 mTLS 运行时拓扑仍只有静态验证。

 以下已有设计、接口或部署位置，但尚无端到端生产实现：

-1. `proxy-loadgen` 已提供有界 HTTP 与 HTTP Gateway CONNECT 长连接进程；`proxy-checker` 的 BASIC 任务进程已经完成，
+1. `proxy-loadgen` 已提供有界 HTTP、HTTP Gateway CONNECT 长连接和 Distribution Extract
+   响应校验场景；`proxy-checker` 的 BASIC 任务进程已经完成，
   `proxy-controller` 已完成
   Admin/Distribution/Metrics 与 PostgreSQL/Redis 启动装配，`proxy-gateway` 已完成
   HTTP/Metrics 与控制面 Session 装配，但 Provider 和业务指标链未闭环。
@ -119,7 +120,7 @@ Controller/Gateway 入口，完整 mTLS 运行时拓扑仍只有静态验证。
   原子归并、Controller Reducer 和 Observation 上报 RPC 已完成；EGRESS、TARGET
   生产任务调度与 REMOVE 编排仍待实现。
 8. Admin/Distribution 细粒度授权和审计查询；Distribution 分布式限流已完成。
-9. Extract 专用场景、真实 Compose/Kubernetes 集成、故障演练和代表性集群负载测试。
+9. 真实 Compose/Kubernetes 集成、故障演练和代表性集群负载测试。
 10. 将 reject/wait/direct 接入 Distribution 运行链，补齐 Sequential 持久化恢复、跨实例 CAS
    和 disabled candidate 语义。
 11. 补齐 Proxy Capacity 动态降容契约、Reservation 全生命周期观测；短 TTL
--- a/docs/requirements/traceability.md
+++ b/docs/requirements/traceability.md
@ -7,7 +7,7 @@

 | ID | 最终需求 | 来源 | 验证证据 |
 |---|---|---|---|
-| ARCH-001 | 数据面 Worker 与控制面 Controller 分离 | 1-70 | 包、协议和部署拓扑已分离；Controller 已运行 Worker Register/Watch/ACK/Runtime/Outcome 与 Checker Observation gRPC，并发布 Proxy/Gateway Routing/按引用去重凭据完整快照；Checker 任务流已具备有界领取、租约栅栏和任务期凭据契约。Gateway 已将快照编译为同版本动态 View，并由独立进程维护控制面会话。`proxy-checker` 与 Redis BASIC/EGRESS/TARGET 共享任务运行态、`proxy-loadgen` 有界 HTTP/CONNECT 长连接场景均已实现 |
+| ARCH-001 | 数据面 Worker 与控制面 Controller 分离 | 1-70 | 包、协议和部署拓扑已分离；Controller 已运行 Worker Register/Watch/ACK/Runtime/Outcome 与 Checker Observation gRPC，并发布 Proxy/Gateway Routing/按引用去重凭据完整快照；Checker 任务流已具备有界领取、租约栅栏和任务期凭据契约。Gateway 已将快照编译为同版本动态 View，并由独立进程维护控制面会话。`proxy-checker` 与 Redis BASIC/EGRESS/TARGET 共享任务运行态、`proxy-loadgen` 有界 HTTP/CONNECT/Extract 校验场景均已实现 |
 | ARCH-002 | 热路径只做认证、本地路由和网络转发 | 1-70, 380-430 | Gateway bootstrap 集成测试验证启动期控制面会话与快照就绪，HTTP 请求只走本地 Snapshot/Dispatch；Outcome 仅写入有界非阻塞本地队列，代表性性能剖析待完成 |
 | ARCH-003 | Gateway、Distribution、Admin、Metrics 独立入口 | 8904-8958 | Controller 命令已装配 Distribution/Admin/Metrics 三个独立监听及联动停机；Gateway 命令已装配代理与 Metrics 监听，运行时 mTLS 部署 Overlay 待完成 |
 | ARCH-004 | Controller 集中 Provider 获取与切换 | 1403-1580 | Redis Leader、动态 Provider Supervisor 与 Bootstrap 生产装配已完成；Admin disable/reload 驱动取消替换，多副本按权威 HMAC 指纹和 revision 栅栏收敛并拒绝旧配置换主；Routing 切换到 Drain 的编排待完成 |
--- a/internal/loadgen/http.go
+++ b/internal/loadgen/http.go
@ -6,7 +6,10 @@ import (
 	"bufio"
 	"bytes"
 	"context"
+	"crypto/rand"
 	"encoding/base64"
+	"encoding/hex"
+	"encoding/json"
 	"errors"
 	"io"
 	"math/bits"
@ -14,6 +17,7 @@ import (
 	"net/http"
 	"net/url"
 	"runtime"
+	"strconv"
 	"strings"
 	"sync"
 	"sync/atomic"
@ -22,48 +26,57 @@ import (

 var ErrInvalidOptions = errors.New("invalid load generator options")

-const maximumResponseDrainBytes = 64 << 10
+const (
+	maximumResponseDrainBytes   = 64 << 10
+	maximumExtractResponseBytes = 256 << 10
+)

 type Scenario string

 const (
 	ScenarioHTTP    Scenario = "http"
 	ScenarioConnect Scenario = "connect"
+	ScenarioExtract Scenario = "extract"
 )

 // Options bounds one HTTP workload. Requests selects a fixed-size run;
 // otherwise Duration selects a time-boxed run and Rate caps its start rate.
 type Options struct {
-	Scenario       Scenario
-	TargetURL      string
-	ProxyURL       string
-	Method         string
-	Headers        http.Header
-	RequestBody    []byte
-	Requests       int
-	Duration       time.Duration
-	Rate           int
-	Concurrency    int
-	RequestTimeout time.Duration
-	TunnelHold     time.Duration
+	Scenario           Scenario
+	TargetURL          string
+	ProxyURL           string
+	Method             string
+	Headers            http.Header
+	RequestBody        []byte
+	Requests           int
+	Duration           time.Duration
+	Rate               int
+	Concurrency        int
+	RequestTimeout     time.Duration
+	TunnelHold         time.Duration
+	ExtractCount       int
+	ExtractFulfillment string
 }

 type Report struct {
-	Scenario           Scenario
-	StartedAt          time.Time
-	Duration           time.Duration
-	Requests           uint64
-	Completed          uint64
-	Succeeded          uint64
-	Failed             uint64
-	Status4xx          uint64
-	Status5xx          uint64
-	TimeoutErrors      uint64
-	RequestErrors      uint64
-	TunnelsEstablished uint64
-	Throughput         float64
-	Latency            LatencyReport
-	Runtime            RuntimeReport
+	Scenario                  Scenario
+	StartedAt                 time.Time
+	Duration                  time.Duration
+	Requests                  uint64
+	Completed                 uint64
+	Succeeded                 uint64
+	Failed                    uint64
+	Status4xx                 uint64
+	Status5xx                 uint64
+	TimeoutErrors             uint64
+	RequestErrors             uint64
+	TunnelsEstablished        uint64
+	ExtractResponsesValidated uint64
+	ExtractReturned           uint64
+	ExtractValidationFailures uint64
+	Throughput                float64
+	Latency                   LatencyReport
+	Runtime                   RuntimeReport
 }

 // LatencyReport uses logarithmic microsecond buckets. Percentiles are upper
@ -85,16 +98,19 @@ type RuntimeReport struct {
 }

 type counters struct {
-	requests           atomic.Uint64
-	completed          atomic.Uint64
-	succeeded          atomic.Uint64
-	failed             atomic.Uint64
-	status4xx          atomic.Uint64
-	status5xx          atomic.Uint64
-	timeoutErrors      atomic.Uint64
-	requestErrors      atomic.Uint64
-	tunnelsEstablished atomic.Uint64
-	latency            latencyHistogram
+	requests                  atomic.Uint64
+	completed                 atomic.Uint64
+	succeeded                 atomic.Uint64
+	failed                    atomic.Uint64
+	status4xx                 atomic.Uint64
+	status5xx                 atomic.Uint64
+	timeoutErrors             atomic.Uint64
+	requestErrors             atomic.Uint64
+	tunnelsEstablished        atomic.Uint64
+	extractResponsesValidated atomic.Uint64
+	extractReturned           atomic.Uint64
+	extractValidationFailures atomic.Uint64
+	latency                   latencyHistogram
 }

 type latencyHistogram struct {
@ -111,13 +127,22 @@ func Run(ctx context.Context, options Options) (Report, error) {
 	}
 	stats := &counters{}
 	var execute func(context.Context)
-	if normalized.Scenario == ScenarioHTTP {
+	if normalized.Scenario == ScenarioHTTP || normalized.Scenario == ScenarioExtract {
 		client, clientErr := newHTTPClient(normalized)
 		if clientErr != nil {
 			return Report{}, clientErr
 		}
 		defer client.CloseIdleConnections()
-		execute = func(requestCtx context.Context) { executeHTTP(requestCtx, client, normalized, stats) }
+		if normalized.Scenario == ScenarioHTTP {
+			execute = func(requestCtx context.Context) { executeHTTP(requestCtx, client, normalized, stats) }
+		} else {
+			workloadID, workloadErr := newExtractWorkloadID()
+			if workloadErr != nil {
+				return Report{}, workloadErr
+			}
+			workload := &extractWorkload{id: workloadID, options: normalized, stats: stats}
+			execute = func(requestCtx context.Context) { workload.execute(requestCtx, client) }
+		}
 	} else {
 		execute = func(requestCtx context.Context) { executeCONNECT(requestCtx, normalized, stats) }
 	}
@ -144,7 +169,11 @@ func normalizeOptions(options Options) (Options, error) {
 		normalized.Scenario = ScenarioHTTP
 	}
 	if normalized.Method == "" {
-		normalized.Method = http.MethodGet
+		if normalized.Scenario == ScenarioExtract {
+			normalized.Method = http.MethodPost
+		} else {
+			normalized.Method = http.MethodGet
+		}
 	}
 	if strings.TrimSpace(normalized.TargetURL) != normalized.TargetURL || normalized.TargetURL == "" ||
 		normalized.Concurrency <= 0 || normalized.RequestTimeout <= 0 || normalized.Requests < 0 || normalized.Duration < 0 ||
@ -156,7 +185,7 @@ func normalizeOptions(options Options) (Options, error) {
 	if err != nil || target.Scheme == "" || target.Host == "" || (target.Scheme != "http" && target.Scheme != "https") {
 		return Options{}, ErrInvalidOptions
 	}
-	if normalized.Scenario != ScenarioHTTP && normalized.Scenario != ScenarioConnect {
+	if normalized.Scenario != ScenarioHTTP && normalized.Scenario != ScenarioConnect && normalized.Scenario != ScenarioExtract {
 		return Options{}, ErrInvalidOptions
 	}
 	if normalized.Scenario == ScenarioHTTP {
@ -168,6 +197,20 @@ func normalizeOptions(options Options) (Options, error) {
 		}
 		return normalized, nil
 	}
+	if normalized.Scenario == ScenarioExtract {
+		if normalized.ProxyURL != "" || normalized.TunnelHold != 0 || len(normalized.RequestBody) != 0 ||
+			normalized.Method != http.MethodPost || normalized.ExtractCount <= 0 || normalized.ExtractCount > 64 ||
+			(normalized.ExtractFulfillment != "" && strings.TrimSpace(normalized.ExtractFulfillment) != normalized.ExtractFulfillment) {
+			return Options{}, ErrInvalidOptions
+		}
+		if normalized.ExtractFulfillment == "" {
+			normalized.ExtractFulfillment = "partial"
+		}
+		if normalized.ExtractFulfillment != "partial" && normalized.ExtractFulfillment != "allOrNothing" {
+			return Options{}, ErrInvalidOptions
+		}
+		return normalized, nil
+	}
 	if normalized.ProxyURL == "" || normalized.TunnelHold <= 0 || len(normalized.RequestBody) != 0 || normalized.Method != http.MethodGet {
 		return Options{}, ErrInvalidOptions
 	}
@ -307,31 +350,42 @@ func runForDuration(ctx context.Context, options Options, execute func(context.C
 	}
 }

+type requestBuilder func(context.Context) (*http.Request, responseValidator, error)
+
+type responseValidator func(*http.Response) error
+
 func executeHTTP(ctx context.Context, client *http.Client, options Options, stats *counters) {
+	executeHTTPRequest(ctx, client, options.RequestTimeout, stats, func(requestContext context.Context) (*http.Request, responseValidator, error) {
+		request, err := http.NewRequestWithContext(
+			requestContext,
+			options.Method,
+			options.TargetURL,
+			bytes.NewReader(options.RequestBody),
+		)
+		if err != nil {
+			return nil, nil, err
+		}
+		request.Header = options.Headers.Clone()
+		return request, discardResponse, nil
+	})
+}
+
+func executeHTTPRequest(ctx context.Context, client *http.Client, timeout time.Duration, stats *counters, build requestBuilder) {
 	stats.requests.Add(1)
 	started := time.Now()
-	requestContext, cancel := context.WithTimeout(ctx, options.RequestTimeout)
+	requestContext, cancel := context.WithTimeout(ctx, timeout)
 	defer cancel()
-	request, err := http.NewRequestWithContext(
-		requestContext,
-		options.Method,
-		options.TargetURL,
-		bytes.NewReader(options.RequestBody),
-	)
+	request, validate, err := build(requestContext)
 	if err != nil {
-		stats.failed.Add(1)
-		stats.requestErrors.Add(1)
-		stats.completed.Add(1)
-		stats.latency.Record(time.Since(started))
+		recordRequestError(started, stats, false)
 		return
 	}
-	request.Header = options.Headers.Clone()
 	response, err := client.Do(request)
 	latency := time.Since(started)
 	stats.latency.Record(latency)
-	stats.completed.Add(1)
 	if err != nil {
 		stats.failed.Add(1)
+		stats.completed.Add(1)
 		if errors.Is(err, context.DeadlineExceeded) || errors.Is(requestContext.Err(), context.DeadlineExceeded) {
 			stats.timeoutErrors.Add(1)
 		} else {
@ -339,10 +393,18 @@ func executeHTTP(ctx context.Context, client *http.Client, options Options, stat
 		}
 		return
 	}
-	_, _ = io.Copy(io.Discard, io.LimitReader(response.Body, maximumResponseDrainBytes))
-	_ = response.Body.Close()
+	defer response.Body.Close()
 	switch {
 	case response.StatusCode >= http.StatusOK && response.StatusCode < http.StatusMultipleChoices:
+		if validate == nil {
+			validate = discardResponse
+		}
+		if err := validate(response); err != nil {
+			stats.failed.Add(1)
+			stats.requestErrors.Add(1)
+			stats.completed.Add(1)
+			return
+		}
 		stats.succeeded.Add(1)
 	case response.StatusCode >= http.StatusInternalServerError:
 		stats.failed.Add(1)
@ -354,6 +416,114 @@ func executeHTTP(ctx context.Context, client *http.Client, options Options, stat
 		stats.failed.Add(1)
 		stats.requestErrors.Add(1)
 	}
+	stats.completed.Add(1)
+}
+
+func discardResponse(response *http.Response) error {
+	if response == nil || response.Body == nil {
+		return ErrInvalidOptions
+	}
+	_, err := io.Copy(io.Discard, io.LimitReader(response.Body, maximumResponseDrainBytes))
+	return err
+}
+
+type extractWorkload struct {
+	id      string
+	options Options
+	stats   *counters
+	next    atomic.Uint64
+}
+
+func newExtractWorkloadID() (string, error) {
+	var value [12]byte
+	if _, err := rand.Read(value[:]); err != nil {
+		return "", err
+	}
+	return "loadgen-" + hex.EncodeToString(value[:]), nil
+}
+
+func (workload *extractWorkload) execute(ctx context.Context, client *http.Client) {
+	sequence := workload.next.Add(1)
+	requestID := workload.id + "-" + strconv.FormatUint(sequence, 10)
+	executeHTTPRequest(ctx, client, workload.options.RequestTimeout, workload.stats, func(requestContext context.Context) (*http.Request, responseValidator, error) {
+		body, err := json.Marshal(struct {
+			Count       int    `json:"count"`
+			Fulfillment string `json:"fulfillment"`
+		}{Count: workload.options.ExtractCount, Fulfillment: workload.options.ExtractFulfillment})
+		if err != nil {
+			return nil, nil, err
+		}
+		request, err := http.NewRequestWithContext(requestContext, http.MethodPost, workload.options.TargetURL, bytes.NewReader(body))
+		if err != nil {
+			return nil, nil, err
+		}
+		request.Header = workload.options.Headers.Clone()
+		if request.Header == nil {
+			request.Header = make(http.Header)
+		}
+		request.Header.Set("Content-Type", "application/json")
+		request.Header.Set("X-Request-ID", requestID)
+		request.Header.Set("Idempotency-Key", requestID)
+		return request, func(response *http.Response) error {
+			return validateExtractResponse(response, requestID, workload.options, workload.stats)
+		}, nil
+	})
+}
+
+func validateExtractResponse(response *http.Response, requestID string, options Options, stats *counters) error {
+	if response == nil || response.Body == nil {
+		stats.extractValidationFailures.Add(1)
+		return ErrInvalidOptions
+	}
+	body, err := io.ReadAll(io.LimitReader(response.Body, maximumExtractResponseBytes+1))
+	if err != nil || len(body) > maximumExtractResponseBytes {
+		stats.extractValidationFailures.Add(1)
+		if err != nil {
+			return err
+		}
+		return ErrInvalidOptions
+	}
+	var payload struct {
+		RequestID string          `json:"requestId"`
+		Requested int             `json:"requested"`
+		Returned  int             `json:"returned"`
+		Proxies   json.RawMessage `json:"proxies"`
+	}
+	if err := json.Unmarshal(body, &payload); err != nil || payload.RequestID != requestID || payload.Requested != options.ExtractCount ||
+		len(payload.Proxies) == 0 || bytes.Equal(bytes.TrimSpace(payload.Proxies), []byte("null")) ||
+		(options.ExtractFulfillment == "allOrNothing" && payload.Returned != options.ExtractCount) {
+		stats.extractValidationFailures.Add(1)
+		if err != nil {
+			return err
+		}
+		return ErrInvalidOptions
+	}
+	var proxies []struct {
+		ID string `json:"id"`
+	}
+	if err := json.Unmarshal(payload.Proxies, &proxies); err != nil || payload.Returned < 0 || payload.Returned > options.ExtractCount ||
+		payload.Returned != len(proxies) {
+		stats.extractValidationFailures.Add(1)
+		if err != nil {
+			return err
+		}
+		return ErrInvalidOptions
+	}
+	ids := make(map[string]struct{}, len(proxies))
+	for _, proxy := range proxies {
+		if proxy.ID == "" {
+			stats.extractValidationFailures.Add(1)
+			return ErrInvalidOptions
+		}
+		if _, duplicate := ids[proxy.ID]; duplicate {
+			stats.extractValidationFailures.Add(1)
+			return ErrInvalidOptions
+		}
+		ids[proxy.ID] = struct{}{}
+	}
+	stats.extractResponsesValidated.Add(1)
+	stats.extractReturned.Add(uint64(payload.Returned))
+	return nil
 }

 func executeCONNECT(ctx context.Context, options Options, stats *counters) {
@ -483,9 +653,10 @@ func report(started time.Time, scenario Scenario, stats *counters) Report {
 		Scenario: scenario, StartedAt: started.UTC(), Duration: duration, Requests: stats.requests.Load(), Completed: completed,
 		Succeeded: stats.succeeded.Load(), Failed: stats.failed.Load(), Status4xx: stats.status4xx.Load(),
 		Status5xx: stats.status5xx.Load(), TimeoutErrors: stats.timeoutErrors.Load(), RequestErrors: stats.requestErrors.Load(),
-		TunnelsEstablished: stats.tunnelsEstablished.Load(),
-		Latency:            stats.latency.Report(completed),
-		Runtime:            RuntimeReport{NumCPU: runtime.NumCPU(), Goroutines: runtime.NumGoroutine(), AllocBytes: memory.Alloc, SysBytes: memory.Sys, GCCount: memory.NumGC},
+		TunnelsEstablished:        stats.tunnelsEstablished.Load(),
+		ExtractResponsesValidated: stats.extractResponsesValidated.Load(), ExtractReturned: stats.extractReturned.Load(),
+		ExtractValidationFailures: stats.extractValidationFailures.Load(), Latency: stats.latency.Report(completed),
+		Runtime: RuntimeReport{NumCPU: runtime.NumCPU(), Goroutines: runtime.NumGoroutine(), AllocBytes: memory.Alloc, SysBytes: memory.Sys, GCCount: memory.NumGC},
 	}
 	if duration > 0 {
 		result.Throughput = float64(completed) / duration.Seconds()
--- a/internal/loadgen/http_test.go
+++ b/internal/loadgen/http_test.go
@ -3,6 +3,7 @@ package loadgen
 import (
 	"bufio"
 	"context"
+	"encoding/json"
 	"errors"
 	"fmt"
 	"io"
@ -136,6 +137,74 @@ func TestRunClassifiesCONNECTProxyRejection(t *testing.T) {
 	}
 }

+func TestRunExecutesAndValidatesExtractWorkload(t *testing.T) {
+	var requests atomic.Int64
+	server := httptest.NewServer(http.HandlerFunc(func(response http.ResponseWriter, request *http.Request) {
+		if request.Method != http.MethodPost || request.Header.Get("Content-Type") != "application/json" ||
+			request.Header.Get("X-Request-ID") == "" || request.Header.Get("Idempotency-Key") == "" {
+			t.Errorf("extract request = method=%s content-type=%q request-id=%q idempotency=%q",
+				request.Method, request.Header.Get("Content-Type"), request.Header.Get("X-Request-ID"), request.Header.Get("Idempotency-Key"))
+		}
+		var body struct {
+			Count       int    `json:"count"`
+			Fulfillment string `json:"fulfillment"`
+		}
+		if err := json.NewDecoder(request.Body).Decode(&body); err != nil || body.Count != 2 || body.Fulfillment != "partial" {
+			t.Errorf("extract body = %+v, error = %v", body, err)
+		}
+		requests.Add(1)
+		response.Header().Set("Content-Type", "application/json")
+		_, _ = fmt.Fprintf(response, `{"requestId":%q,"requested":2,"returned":2,"proxies":[{"id":"proxy-a"},{"id":"proxy-b"}]}`,
+			request.Header.Get("X-Request-ID"))
+	}))
+	defer server.Close()
+
+	report, err := Run(context.Background(), Options{
+		Scenario: ScenarioExtract, TargetURL: server.URL, Requests: 2, Concurrency: 1, RequestTimeout: time.Second,
+		ExtractCount: 2, ExtractFulfillment: "partial",
+	})
+	if err != nil || report.Requests != 2 || report.Completed != 2 || report.Succeeded != 2 || report.Failed != 0 ||
+		report.ExtractResponsesValidated != 2 || report.ExtractReturned != 4 || report.ExtractValidationFailures != 0 || requests.Load() != 2 {
+		t.Fatalf("Run() = (%+v, %v); extract requests=%d", report, err, requests.Load())
+	}
+}
+
+func TestRunRejectsInvalidExtractResponse(t *testing.T) {
+	server := httptest.NewServer(http.HandlerFunc(func(response http.ResponseWriter, request *http.Request) {
+		response.Header().Set("Content-Type", "application/json")
+		_, _ = fmt.Fprintf(response, `{"requestId":%q,"requested":2,"returned":2,"proxies":[{"id":"proxy-a"}]}`,
+			request.Header.Get("X-Request-ID"))
+	}))
+	defer server.Close()
+
+	report, err := Run(context.Background(), Options{
+		Scenario: ScenarioExtract, TargetURL: server.URL, Requests: 1, Concurrency: 1, RequestTimeout: time.Second,
+		ExtractCount: 2, ExtractFulfillment: "partial",
+	})
+	if err != nil || report.Succeeded != 0 || report.Failed != 1 || report.ExtractResponsesValidated != 0 ||
+		report.ExtractValidationFailures != 1 || report.ExtractReturned != 0 {
+		t.Fatalf("Run() = (%+v, %v)", report, err)
+	}
+}
+
+func TestRunRejectsDuplicateProxyIDsInExtractResponse(t *testing.T) {
+	server := httptest.NewServer(http.HandlerFunc(func(response http.ResponseWriter, request *http.Request) {
+		response.Header().Set("Content-Type", "application/json")
+		_, _ = fmt.Fprintf(response, `{"requestId":%q,"requested":2,"returned":2,"proxies":[{"id":"proxy-a"},{"id":"proxy-a"}]}`,
+			request.Header.Get("X-Request-ID"))
+	}))
+	defer server.Close()
+
+	report, err := Run(context.Background(), Options{
+		Scenario: ScenarioExtract, TargetURL: server.URL, Requests: 1, Concurrency: 1, RequestTimeout: time.Second,
+		ExtractCount: 2, ExtractFulfillment: "partial",
+	})
+	if err != nil || report.Succeeded != 0 || report.Failed != 1 || report.ExtractResponsesValidated != 0 ||
+		report.ExtractValidationFailures != 1 {
+		t.Fatalf("Run() = (%+v, %v)", report, err)
+	}
+}
+
 func TestRunRejectsCONNECTWithoutProxyOrHold(t *testing.T) {
 	_, err := Run(context.Background(), Options{
 		Scenario: ScenarioConnect, TargetURL: "https://target.example/health", Requests: 1, Concurrency: 1, RequestTimeout: time.Second,
@ -145,6 +214,16 @@ func TestRunRejectsCONNECTWithoutProxyOrHold(t *testing.T) {
 	}
 }

+func TestRunRejectsInvalidExtractOptions(t *testing.T) {
+	_, err := Run(context.Background(), Options{
+		Scenario: ScenarioExtract, TargetURL: "http://127.0.0.1:8081/api/v1/proxies/extract", Requests: 1, Concurrency: 1,
+		RequestTimeout: time.Second, ExtractCount: 0,
+	})
+	if !errors.Is(err, ErrInvalidOptions) {
+		t.Fatalf("Run(invalid extract options) error = %v, want ErrInvalidOptions", err)
+	}
+}
+
 func TestRunRejectsUnboundedWorkload(t *testing.T) {
 	_, err := Run(context.Background(), Options{TargetURL: "http://127.0.0.1:8080", Concurrency: 1})
 	if !errors.Is(err, ErrInvalidOptions) {