diff --git a/README.md b/README.md index 2b7cf4e..f8b3eac 100644 --- a/README.md +++ b/README.md @@ -108,6 +108,7 @@ api_key = "" | `GET` | `/api/v1/tasks` | 查询任务列表。按创建时间倒序返回。 | | `GET` | `/api/v1/tasks/{id}` | 查询单个任务详情。 | | `GET` | `/api/v1/tasks/{id}/logs` | 分页读取任务日志。 | +| `GET` | `/api/v1/tasks/{id}/results` | 分页读取逐次提交结果。 | | `POST` | `/api/v1/surveys/parse` | 解析问卷，不提交答案。 | | `POST` | `/api/v1/configs` | 生成 V1 默认运行配置。 | | `POST` | `/api/v1/submissions` | 同步提交一次，返回本次结果和耗时。 | @@ -126,6 +127,8 @@ api_key = "" 单次提交可调用 Go SDK 的 `client.Submit(ctx, &surveycore.SubmissionRequest{Config: cfg})`，或把配置放在 `config` 字段中发送到 `POST /api/v1/submissions`。该接口同步返回 `attempt`、`provider`、`success`、`elapsed_seconds` 和 `error`，支持 Go 与 Python v1 配置。单次提交固定执行一次，平台失败时通过 `success: false` 和 `error` 返回结果。 +批量运行可用 `for result, err := range client.RunResults(ctx, cfg)` 按完成顺序读取结果，`attempt` 从 1 开始。目标数表示尝试次数；每次失败保留结果并继续后续尝试，不自动重试。除第一次外，每次尝试开始前都会等待配置中的随机提交间隔。原生 Go 配置显式启用 `failStop` 时仍会在连续失败达到阈值后停止。HTTP 任务用 `GET /api/v1/tasks/{id}/results?after=0&limit=200` 读取同样的逐次结果。 + ## 错误响应 请求级错误统一返回稳定错误码、用户消息和调试详情： diff --git a/api/openapi.yaml b/api/openapi.yaml index acceb7c..149b2bf 100644 --- a/api/openapi.yaml +++ b/api/openapi.yaml @@ -77,6 +77,17 @@ paths: '400': { $ref: '#/components/responses/InvalidQuery' } '404': { $ref: '#/components/responses/NotFound' } '500': { $ref: '#/components/responses/InternalError' } + /api/v1/tasks/{id}/results: + parameters: [{ $ref: '#/components/parameters/TaskID' }] + get: + operationId: listTaskResults + description: Completed attempts in completion order. Use next_cursor as the next after value; has_more only describes stored results, not task completion. Results survive service restarts. + parameters: [{ $ref: '#/components/parameters/After' }, { $ref: '#/components/parameters/Limit' }] + responses: + '200': { description: Cursor-paginated per-attempt results, content: { application/json: { schema: { $ref: '#/components/schemas/TaskResultPage' } } } } + '400': { $ref: '#/components/responses/InvalidQuery' } + '404': { $ref: '#/components/responses/NotFound' } + '500': { $ref: '#/components/responses/InternalError' } /api/v1/qrcode/decode: post: operationId: decodeQRCode @@ -182,10 +193,10 @@ components: additionalProperties: false required: [attempt, provider, success, elapsed_seconds, error] properties: - attempt: { type: integer, minimum: 0, description: Standalone Submit uses zero } + attempt: { type: integer, minimum: 0, description: Standalone Submit uses zero; batch attempts start at one and are returned in completion order } provider: { $ref: '#/components/schemas/Provider' } success: { type: boolean } - elapsed_seconds: { type: number, minimum: 0 } + elapsed_seconds: { type: number, minimum: 0, description: Standalone Submit includes preparation; batch results exclude shared preparation and submit intervals } error: { type: [string, 'null'] } RunRequest: type: object @@ -206,7 +217,7 @@ components: enabled: { type: boolean } ratios: { type: object, additionalProperties: { type: integer } } PythonConfig: - description: Imports surveycore-py schema_version 1. Task responses use RunRequest; scheduling and results retain Go semantics. + description: Imports surveycore-py schema_version 1. Task responses use RunRequest. Target counts attempts, failures continue without automatic retries, and results use completion order. type: object additionalProperties: false required: [schema_version, url] @@ -275,9 +286,11 @@ components: additionalProperties: false required: [target, threads, submitInterval, answerDuration, failStop, pauseOnAliyunCaptcha] properties: - target: { type: integer, minimum: 1 } + target: { type: integer, minimum: 1, description: Number of attempts; each failed attempt counts once without automatic retries } threads: { type: integer, minimum: 1 } - submitInterval: { $ref: '#/components/schemas/IntegerPair' } + submitInterval: + $ref: '#/components/schemas/IntegerPair' + description: Random seconds waited inside a concurrency slot before each attempt after the first, including new workers answerDuration: { $ref: '#/components/schemas/IntegerPair' } answerDatetimeWindow: { $ref: '#/components/schemas/StringPair' } failStop: { type: boolean } @@ -494,6 +507,7 @@ components: current: { type: integer } total: { type: integer } time: { type: string, format: date-time } + result: { $ref: '#/components/schemas/SubmissionResult' } Task: type: object additionalProperties: false @@ -538,6 +552,14 @@ components: logs: { type: array, items: { $ref: '#/components/schemas/TaskLog' } } next_cursor: { type: integer } has_more: { type: boolean } + TaskResultPage: + type: object + additionalProperties: false + required: [results, next_cursor, has_more] + properties: + results: { type: array, items: { $ref: '#/components/schemas/SubmissionResult' } } + next_cursor: { type: integer, minimum: 0, description: Opaque log ID cursor; an empty page preserves the input cursor } + has_more: { type: boolean, description: Whether more completed results are currently stored } ProxyExtractRequest: type: object additionalProperties: false diff --git a/docs/sdk/api.md b/docs/sdk/api.md index 2114821..8cea3ef 100644 --- a/docs/sdk/api.md +++ b/docs/sdk/api.md @@ -203,7 +203,9 @@ Content-Type: application/json } ``` -`202` 只代表任务已经创建。 +`202` 代表任务已经创建。任务独立于创建请求运行，响应结束或客户端断开不会取消已创建的任务。使用停止接口取消任务，服务关闭时也会停止运行中的任务。 + +目标数表示尝试次数，失败保留结果并继续后续尝试，不自动重试。第 2 次起，每次尝试开始前在并发槽内等待配置中的随机提交间隔。显式启用 `execution.failStop` 时，连续失败达到阈值可提前结束。存在单次失败时最终状态为 `failed`，已完成次数见汇总和逐次结果。 ## 查询任务 @@ -230,6 +232,8 @@ POST /api/v1/tasks/{id}/stop 返回停止后的 `Task`。已结束的任务保持原状态。 +停止操作会取消运行中的平台请求。工作线程退出后更新 `finished_at` 和汇总；取消中的请求不计为失败，也不产生逐次结果。 + ## 读取任务日志 ```http @@ -256,6 +260,29 @@ GET /api/v1/tasks/{id}/logs?after=0&limit=200 } ``` +## 读取逐次提交结果 + +```http +GET /api/v1/tasks/{id}/results?after=0&limit=200 +``` + +`after` 为非负游标，初次传 0，后续使用上一页的 `next_cursor`。游标使用底层日志 ID，不能用尝试编号推算。`limit` 默认 200，0 使用默认值，超过 1000 按 1000 处理；负数或非法参数返回 `400 invalid_query`，任务不存在返回 `404 not_found`。 + +```json +{ + "results": [ + {"attempt": 2, "provider": "wjx", "success": true, "elapsed_seconds": 0.8, "error": null}, + {"attempt": 1, "provider": "wjx", "success": false, "elapsed_seconds": 1.2, "error": "提交失败"} + ], + "next_cursor": 16, + "has_more": false +} +``` + +结果按完成记录顺序返回，`attempt` 从 1 开始，因此并发运行时编号可能乱序。每次失败单独保留 `error`。批量耗时不含公共准备和提交间隔。结果保存在 SQLite 任务日志中，服务重启后仍可读取；升级前未记录的逐次结果不会补生成。 + +空页返回 `results: []` 并保留请求游标。`has_more` 只表示当前还有已存储结果；任务运行期间为 false 时，可沿用游标继续轮询，并通过任务状态判断是否结束。任务日志接口中的完成事件也包含 `event.result`。 + ## 解析二维码 ```http diff --git a/docs/sdk/errors.md b/docs/sdk/errors.md index bc25874..e34f949 100644 --- a/docs/sdk/errors.md +++ b/docs/sdk/errors.md @@ -27,7 +27,7 @@ outline: deep | `400` | `invalid_query` | 查询参数不是非负整数。 | | `404` | `not_found` | 任务不存在。 | | `422` | `validation_error` | 单次提交前置参数无效、任务保存失败或二维码内容不是有效问卷链接。 | -| `500` | `internal_error` | 停止任务或读取日志时发生内部错误。 | +| `500` | `internal_error` | 停止任务、读取日志或逐次结果时发生内部错误。 | | `502` | `upstream_error` | 问卷或官方代理上游请求失败。 | ## 接口错误速查 @@ -41,6 +41,7 @@ outline: deep | `GET /api/v1/tasks/{id}` | `404` | | `POST /api/v1/tasks/{id}/stop` | `404`、`500` | | `GET /api/v1/tasks/{id}/logs` | `400`、`404`、`500` | +| `GET /api/v1/tasks/{id}/results` | `400`、`404`、`500` | | `POST /api/v1/qrcode/decode` | `400`、`422` | | `/api/v1/proxy/*` | `400`、`502` | diff --git a/docs/sdk/examples.md b/docs/sdk/examples.md index 8d7a060..35a82cf 100644 --- a/docs/sdk/examples.md +++ b/docs/sdk/examples.md @@ -57,6 +57,26 @@ if result != nil { 该调用固定执行一次。已有问卷快照可通过 `Survey` 传入。原始 `cfg` 中的目标数和并发数保持原值。 +### Go 逐次提交结果 + +```go +cfg.ExecutionPlan.Target = 3 +cfg.ExecutionPlan.Threads = 2 +cfg.ExecutionPlan.FailStop = false +for result, err := range client.RunResults(ctx, cfg) { + if err != nil { + log.Printf("运行结束: %v", err) + break + } + log.Printf("尝试=%d 成功=%t 耗时=%.3f秒", result.Attempt, result.Success, result.ElapsedSeconds) + if result.Error != nil { + log.Print(*result.Error) + } +} +``` + +结果按完成顺序返回。单次失败后继续尝试；准备或上下文错误终止迭代。消费方提前 `break` 时，SDK 会取消未完成请求并等待工作线程退出。 + ## REST API ### JavaScript @@ -97,8 +117,14 @@ async function createTaskFromURL(url) { async function getTask(taskID) { return requestJSON(`/tasks/${taskID}`); } + +async function readTaskResults(taskID, after = 0) { + return requestJSON(`/tasks/${taskID}/results?after=${after}&limit=200`); +} ``` +将返回的 `next_cursor` 用作下次 `after`。`has_more` 仅表示当前还有已存储的结果；任务仍在运行时，即使它为 false，也应保留游标继续轮询，并结合任务状态判断是否结束。 + ### Python ```python diff --git a/docs/sdk/index.md b/docs/sdk/index.md index 63e8de1..c439ea2 100644 --- a/docs/sdk/index.md +++ b/docs/sdk/index.md @@ -14,10 +14,14 @@ go get github.com/SurveyController/SurveyCore@ `pkg/surveycore` 只负责解析、默认配置和运行。公共结构在 `pkg/surveycore/model`。低层执行、代理和本地配置能力分别在实验性的 `pkg/surveycore/runtime`、`pkg/surveycore/proxy`、`pkg/surveycore/config`。 -`Parse`、`DefaultConfig`、`Submit`、`Run` 和 `RunWithEvents` 接受 `context.Context`。取消后不再安排新提交，并等待已开始的平台请求返回。事件回调可能来自工作 goroutine，回调实现必须自行保证并发安全。 +`Parse`、`DefaultConfig`、`Submit`、`Run`、`RunWithEvents` 和 `RunResults` 接受 `context.Context`。取消后不再安排新提交，并等待已开始的平台请求返回。事件回调可能来自工作 goroutine，回调实现必须自行保证并发安全。 `Submit` 使用 `surveycore.SubmissionRequest`，其中 `Config` 是 `*model.RunRequest`，`Survey` 是可选的 `*model.SurveyDefinition`。它固定提交一次并返回 `SubmissionResult`；失败时同时返回可用 `errors.Is` 或 `ClassifySubmissionError` 判断的错误。参数中的目标数、并发数和提交间隔不会触发额外提交。 +`RunResults` 返回 `iter.Seq2[SubmissionResult, error]`，按完成顺序产生结果，尝试编号从 1 开始。单次失败保存在结果的 `error` 中，迭代继续；准备失败或调用方取消通过迭代错误返回。提前 `break` 会取消未完成的请求并等待工作线程退出。`Run` 和 `RunWithEvents` 返回汇总，存在失败时在运行结束后返回首个错误；`RunWithEvents` 的 `event.Result` 包含逐次结果。 + +批量运行以 `target` 为尝试次数，受 `threads` 并发限制，不自动重试。第 2 次起，每次尝试开始前均在并发槽内等待 `submitInterval` 范围内的随机秒数。显式 `failStop`、停止操作或取消可以使运行提前结束。Python v1 配置导入时 `failStop` 为 false。 + 错误可用 `errors.Is` 判断 `ErrInvalidConfig`、`ErrParseFailed`、`ErrPrepareConfigFailed`、`ErrRunFailed` 和 `ErrUnsupportedOperation`，也可用 `ClassifyRunError` 获取稳定分类。 ## 文档结构 @@ -60,6 +64,7 @@ Accept: application/json 4. 调 `POST /api/v1/tasks` 创建任务。 5. 调 `GET /api/v1/tasks/{id}` 查询状态。 6. 调 `GET /api/v1/tasks/{id}/logs` 读取日志。 +7. 调 `GET /api/v1/tasks/{id}/results` 分页读取逐次提交结果。 ## 接口清单 @@ -75,6 +80,7 @@ Accept: application/json | `GET` | `/api/v1/tasks/{id}` | 查询任务。 | | `POST` | `/api/v1/tasks/{id}/stop` | 停止任务。 | | `GET` | `/api/v1/tasks/{id}/logs` | 读取任务日志。 | +| `GET` | `/api/v1/tasks/{id}/results` | 读取逐次提交结果。 | | `POST` | `/api/v1/qrcode/decode` | 解析二维码。 | | `POST` | `/api/v1/proxy/session` | 激活官方代理会话。 | | `GET` | `/api/v1/proxy/usage` | 查询官方代理余量。 | diff --git a/docs/sdk/schemas.md b/docs/sdk/schemas.md index 6cc9bb9..764ba36 100644 --- a/docs/sdk/schemas.md +++ b/docs/sdk/schemas.md @@ -50,9 +50,9 @@ outline: deep | `source.url` | `string` | 问卷链接。创建任务时不能为空。 | | `source.provider` | `string` | `wjx`、`qq` 或 `credamo`。 | | `definition` | `SurveyDefinition` | 解析得到的问卷快照。 | -| `execution.target` | `integer` | 目标提交份数。 | +| `execution.target` | `integer` | 目标尝试次数，失败占一次，不自动重试。 | | `execution.threads` | `integer` | 并发数。 | -| `execution.submitInterval` | `[integer, integer]` | 每个工作线程两次提交之间的随机等待秒数。 | +| `execution.submitInterval` | `[integer, integer]` | 第 2 次起，每次尝试开始前在并发槽内随机等待的秒数范围，包括新工作线程的首次提交。 | | `execution.answerDuration` | `[integer, integer]` | 作答时长范围，单位秒。 | | `execution.answerDatetimeWindow` | `[string, string]` | 可选作答时间窗口。 | | `execution.failStop` | `boolean` | 连续失败达到阈值时停止。 | @@ -140,14 +140,28 @@ outline: deep | 字段 | 类型 | 说明 | |---|---|---| -| `attempt` | `integer` | 独立 `Submit` 调用固定为 0。 | +| `attempt` | `integer` | 独立 `Submit` 调用固定为 0；批量运行从 1 开始编号，结果按完成顺序返回。 | | `provider` | `string` | `wjx`、`qq` 或 `credamo`。 | | `success` | `boolean` | 平台是否确认本次提交成功。 | -| `elapsed_seconds` | `number` | 本次调用耗时，单位秒，包含解析与准备。 | +| `elapsed_seconds` | `number` | 单位秒。独立 `Submit` 包含解析与准备；批量结果仅计本次执行耗时，不含公共准备及提交间隔。 | | `error` | `string` 或 null | 成功时显式返回 null，失败时返回详情。 | Go 的 `Submit` 返回 `(result, error)`：缺少配置、链接或平台不兼容时 `result` 为 nil；执行失败时同时返回失败结果和带分类的错误。取消保留 `context.Canceled` 或 `context.DeadlineExceeded`。原始配置及问卷快照不会被修改。 +`RunResults` 在单次失败时返回失败结果和 nil 迭代错误，继续后续尝试；准备失败或调用方取消返回迭代错误。取消中的请求及尚未开始的尝试不计入成功数、失败数或逐次结果。 + +## Event 与 TaskResultPage + +`Event` 保留 `worker`、`message`、`success`、`fail`、`current`、`total` 和 `time`。每次尝试完成时，事件额外包含 `result: SubmissionResult`。普通进度事件省略 `result`；获取逐次结果时应筛选该字段。 + +`GET /api/v1/tasks/{id}/results` 返回： + +| 字段 | 类型 | 说明 | +|---|---|---| +| `results` | `SubmissionResult[]` | 按完成记录顺序返回，空页为 `[]`。 | +| `next_cursor` | `integer` | 下次请求的 `after`，空页保留原游标。游标使用底层日志 ID，可能不连续。 | +| `has_more` | `boolean` | 当前是否还有未返回的已存储结果。任务结束状态由 `Task.status` 判断。 | + ## Task | 字段 | 类型 | 说明 | @@ -183,4 +197,4 @@ Go 的 `Submit` 返回 `(result, error)`：缺少配置、链接或平台不兼 默认目标数和并发数都是 1，作答时长为 `[60, 120]`，间隔为 `[0, 0]`，信效度开关为 true、目标 alpha 为 0.85，随机 User-Agent 默认关闭。Python 序列化格式不包含 `answer_generator` 或题目 `ai_enabled`，导入会拒绝这些字段。 -这一兼容入口用于配置导入。任务仍使用 Go 的重试、间隔和结果返回规则；Python 的逐次提交结果接口另行对齐。 +批量运行已对齐 Python 的尝试次数、并发限制、失败后继续、无自动重试、尝试前间隔和按完成顺序返回结果的规则。Go SDK 使用 `RunResults` 迭代，HTTP 使用任务结果分页。 diff --git a/internal/service/manager.go b/internal/service/manager.go index 3cf5d6a..4f2da59 100644 --- a/internal/service/manager.go +++ b/internal/service/manager.go @@ -54,6 +54,9 @@ func (m *Manager) Create(ctx context.Context, cfg *model.RunRequest) (*Task, err if cfg == nil { return nil, errors.New("配置不能为空") } + if ctx.Err() != nil { + return nil, ctx.Err() + } id, err := taskID() if err != nil { return nil, err @@ -64,7 +67,8 @@ func (m *Manager) Create(ctx context.Context, cfg *model.RunRequest) (*Task, err return nil, err } m.appendLog(id, TaskLog{Timestamp: now, Level: "INFO", Message: "任务已创建"}) - runCtx, cancel := context.WithCancel(ctx) + // Async tasks outlive the HTTP request that created them. Stop owns cancellation. + runCtx, cancel := context.WithCancel(context.WithoutCancel(ctx)) m.mu.Lock() m.tasks[id] = task m.runtimes[id] = &taskRuntime{cancel: cancel} @@ -145,7 +149,9 @@ func (m *Manager) Stop(id string) (*Task, error) { t.StopMessage = "用户请求停止" rt.cancel() snap := cloneTask(t) - go m.store.SaveTask(snap) + if err := m.store.SaveTask(snap); err != nil { + return nil, err + } return snap, nil } func (m *Manager) Logs(id string, after int64, limit int) (*TaskLogPage, error) { @@ -154,6 +160,13 @@ func (m *Manager) Logs(id string, after int64, limit int) (*TaskLogPage, error) } return m.store.LoadLogs(id, after, limit) } + +func (m *Manager) Results(id string, after int64, limit int) (*TaskResultPage, error) { + if _, ok := m.Get(id); !ok { + return nil, ErrTaskNotFound + } + return m.store.LoadResults(id, after, limit) +} func (m *Manager) Parse(ctx context.Context, url string) (*model.SurveyDefinition, error) { return m.client.Parse(ctx, url) } diff --git a/internal/service/results_test.go b/internal/service/results_test.go new file mode 100644 index 0000000..0a1646e --- /dev/null +++ b/internal/service/results_test.go @@ -0,0 +1,49 @@ +package service + +import ( + "context" + "errors" + "testing" + "time" + + "github.com/SurveyController/SurveyCore/pkg/surveycore/model" +) + +func TestResultsExcludeOtherTasksAndProgressLogs(t *testing.T) { + store := NewStore(t.TempDir() + "/results.db") + if err := store.Init(); err != nil { + t.Fatal(err) + } + defer store.Close() + for _, id := range []string{"one", "two"} { + if err := store.SaveTask(&Task{ID: id, CreatedAt: time.Now()}); err != nil { + t.Fatal(err) + } + } + for _, entry := range []struct { + id string + event *model.Event + }{ + {"one", nil}, {"one", &model.Event{Message: "progress"}}, + {"two", &model.Event{Result: &model.SubmissionResult{Attempt: 2, Provider: "qq", Success: true}}}, + {"one", &model.Event{Result: &model.SubmissionResult{Attempt: 1, Provider: "wjx", Success: true}}}, + } { + if err := store.AppendLog(entry.id, TaskLog{Timestamp: time.Now(), Event: entry.event}); err != nil { + t.Fatal(err) + } + } + page, err := store.LoadResults("one", 0, 0) + if err != nil || len(page.Results) != 1 || page.Results[0].Provider != "wjx" || page.NextCursor != 4 || page.HasMore { + t.Fatalf("page=%#v error=%v", page, err) + } + ctx, cancel := context.WithCancel(context.Background()) + cancel() + manager := NewManager(store, nil) + if task, err := manager.Create(ctx, &model.RunRequest{}); task != nil || !errors.Is(err, context.Canceled) { + t.Fatalf("canceled creation task=%#v error=%v", task, err) + } + tasks, errs := store.LoadTasks() + if len(tasks) != 2 || len(errs) != 0 { + t.Fatalf("canceled creation persisted a task: count=%d errors=%v", len(tasks), errs) + } +} diff --git a/internal/service/store.go b/internal/service/store.go index 0a6c0f4..840e358 100644 --- a/internal/service/store.go +++ b/internal/service/store.go @@ -13,6 +13,7 @@ import ( "sync" "time" + "github.com/SurveyController/SurveyCore/pkg/surveycore/model" "github.com/SurveyController/SurveyCore/pkg/surveycore/proxy" _ "modernc.org/sqlite" ) @@ -277,6 +278,43 @@ func normalizeLogLimit(limit int) int { return limit } +// LoadResults reads completion records from the existing task event log. +func (s *Store) LoadResults(taskID string, afterID int64, limit int) (*TaskResultPage, error) { + if afterID < 0 { + return nil, errors.New("结果游标不能小于 0") + } + limit = normalizeLogLimit(limit) + rows, err := s.database().Query(` + SELECT id, json_extract(event_json, '$.result') + FROM v1_task_logs + WHERE task_id = ? AND id > ? AND json_extract(event_json, '$.result') IS NOT NULL + ORDER BY id ASC LIMIT ? + `, taskID, afterID, limit+1) + if err != nil { + return nil, err + } + defer rows.Close() + page := &TaskResultPage{Results: make([]model.SubmissionResult, 0, limit), NextCursor: afterID} + for rows.Next() { + var id int64 + var raw string + if err := rows.Scan(&id, &raw); err != nil { + return nil, err + } + if len(page.Results) == limit { + page.HasMore = true + break + } + var result model.SubmissionResult + if err := json.Unmarshal([]byte(raw), &result); err != nil { + return nil, fmt.Errorf("解析提交结果失败: %w", err) + } + page.Results = append(page.Results, result) + page.NextCursor = id + } + return page, rows.Err() +} + func (s *Store) database() *sql.DB { s.mu.Lock() defer s.mu.Unlock() diff --git a/internal/service/types.go b/internal/service/types.go index 0aedf03..0094482 100644 --- a/internal/service/types.go +++ b/internal/service/types.go @@ -46,6 +46,12 @@ type TaskLogPage struct { HasMore bool `json:"has_more"` } +type TaskResultPage struct { + Results []model.SubmissionResult `json:"results"` + NextCursor int64 `json:"next_cursor"` + HasMore bool `json:"has_more"` +} + type taskRuntime struct { cancel context.CancelFunc result *model.RunResult diff --git a/pkg/restapi/server.go b/pkg/restapi/server.go index 8647ea7..4c7cfa3 100644 --- a/pkg/restapi/server.go +++ b/pkg/restapi/server.go @@ -71,6 +71,7 @@ func (s *Server) routes() http.Handler { mux.HandleFunc("GET /api/v1/tasks/{id}", s.get) mux.HandleFunc("POST /api/v1/tasks/{id}/stop", s.stop) mux.HandleFunc("GET /api/v1/tasks/{id}/logs", s.logs) + mux.HandleFunc("GET /api/v1/tasks/{id}/results", s.results) mux.HandleFunc("POST /api/v1/qrcode/decode", s.qr) mux.HandleFunc("POST /api/v1/proxy/session", s.proxySession) mux.HandleFunc("GET /api/v1/proxy/usage", s.proxyUsage) @@ -233,6 +234,29 @@ func (s *Server) logs(w http.ResponseWriter, r *http.Request) { } write(w, 200, p) } +func (s *Server) results(w http.ResponseWriter, r *http.Request) { + after, err := queryInt(r, "after", 0) + if err != nil { + fail(w, 400, "invalid_query", err) + return + } + limit, err := queryInt(r, "limit", 200) + if err != nil { + fail(w, 400, "invalid_query", err) + return + } + page, err := s.manager.Results(r.PathValue("id"), int64(after), limit) + if errors.Is(err, service.ErrTaskNotFound) { + fail(w, 404, "not_found", err) + return + } + if err != nil { + fail(w, 500, "internal_error", err) + return + } + write(w, 200, page) +} + func (s *Server) qr(w http.ResponseWriter, r *http.Request) { if err := r.ParseMultipartForm(8 << 20); err != nil { fail(w, 400, "invalid_request", err) diff --git a/pkg/restapi/server_test.go b/pkg/restapi/server_test.go index 1bc32a9..9cda870 100644 --- a/pkg/restapi/server_test.go +++ b/pkg/restapi/server_test.go @@ -1,7 +1,6 @@ package restapi import ( - "context" "encoding/json" "io" "net/http" @@ -40,16 +39,9 @@ func TestV1VersionUsesConfiguredBuildVersion(t *testing.T) { } func TestV1CreateTaskAcceptsPythonConfig(t *testing.T) { - s, err := New(Config{DBPath: t.TempDir() + "/tasks.db"}) - if err != nil { - t.Fatal(err) - } - defer s.Close() - // A cancelled context exercises task creation and persistence without upstream requests. - ctx, cancel := context.WithCancel(context.Background()) - cancel() + s, _ := submissionTestServer(t, true) body := `{"schema_version":1,"url":"https://www.wjx.cn/vm/demo.aspx","target":3,"concurrency":2,"random_user_agent":true,"user_agent_ratios":{"pc":100},"questions":[{"question_type":"single","probabilities":[0,1]}]}` - req := httptest.NewRequest(http.MethodPost, "/api/v1/tasks", strings.NewReader(body)).WithContext(ctx) + req := httptest.NewRequest(http.MethodPost, "/api/v1/tasks", strings.NewReader(body)) rec := httptest.NewRecorder() s.Handler().ServeHTTP(rec, req) if rec.Code != http.StatusAccepted { diff --git a/pkg/restapi/task_results_test.go b/pkg/restapi/task_results_test.go new file mode 100644 index 0000000..cdb09f3 --- /dev/null +++ b/pkg/restapi/task_results_test.go @@ -0,0 +1,191 @@ +package restapi + +import ( + "encoding/json" + "fmt" + "io" + "net/http" + "net/http/httptest" + "strings" + "sync/atomic" + "testing" + "time" + + "github.com/SurveyController/SurveyCore/internal/service" + "github.com/SurveyController/SurveyCore/pkg/surveycore" +) + +const taskResultsHTML = `Task results
Choice
A
B
` + +func taskResultsServer(t *testing.T, path string, transport submissionTestTransport) *Server { + t.Helper() + s, err := New(Config{DBPath: path}) + if err != nil { + t.Fatal(err) + } + s.manager = service.NewManager(s.store, surveycore.New(surveycore.WithHTTPClient(&http.Client{Transport: transport}))) + t.Cleanup(func() { _ = s.Close() }) + return s +} + +func createResultTask(t *testing.T, serverURL string, target int) service.Task { + t.Helper() + body := fmt.Sprintf(`{"schema_version":1,"url":"https://www.wjx.cn/vm/demo.aspx","target":%d,"concurrency":1,"answer_duration":[0,0]}`, target) + response, err := http.Post(serverURL+"/api/v1/tasks", "application/json", strings.NewReader(body)) + if err != nil { + t.Fatal(err) + } + defer response.Body.Close() + var task service.Task + if response.StatusCode != http.StatusAccepted { + data, _ := io.ReadAll(response.Body) + t.Fatalf("create status=%d body=%s", response.StatusCode, data) + } + if err := json.NewDecoder(response.Body).Decode(&task); err != nil { + t.Fatal(err) + } + return task +} + +func waitForTask(t *testing.T, s *Server, id string) *service.Task { + t.Helper() + deadline := time.NewTimer(3 * time.Second) + defer deadline.Stop() + ticker := time.NewTicker(time.Millisecond) + defer ticker.Stop() + for { + if task, ok := s.manager.Get(id); ok && task.FinishedAt != nil { + return task + } + select { + case <-deadline.C: + t.Fatal("task did not finish") + case <-ticker.C: + } + } +} + +func readResultPage(t *testing.T, s *Server, id string, after int64, limit int) service.TaskResultPage { + t.Helper() + rec := httptest.NewRecorder() + s.Handler().ServeHTTP(rec, httptest.NewRequest(http.MethodGet, fmt.Sprintf("/api/v1/tasks/%s/results?after=%d&limit=%d", id, after, limit), nil)) + if rec.Code != http.StatusOK { + t.Fatalf("page status=%d body=%s", rec.Code, rec.Body.String()) + } + var page service.TaskResultPage + if err := json.Unmarshal(rec.Body.Bytes(), &page); err != nil { + t.Fatal(err) + } + return page +} + +func TestTaskResultsSurviveRequestCompletionAndServiceRestart(t *testing.T) { + path := t.TempDir() + "/results.db" + release := make(chan struct{}) + var posts atomic.Int32 + s := taskResultsServer(t, path, func(request *http.Request) (*http.Response, error) { + body := taskResultsHTML + if request.Method == http.MethodPost { + if posts.Add(1) == 1 { + select { + case <-release: + body = "1〒1〒失败" + case <-request.Context().Done(): + return nil, request.Context().Err() + } + } else { + body = "10" + } + } + return &http.Response{StatusCode: 200, Header: make(http.Header), Body: io.NopCloser(strings.NewReader(body)), Request: request}, nil + }) + httpServer := httptest.NewServer(s.Handler()) + defer httpServer.Close() + created := createResultTask(t, httpServer.URL, 3) + // The HTTP creation request has ended before any attempt can complete. + close(release) + finished := waitForTask(t, s, created.ID) + if finished.Status != service.TaskFailed || finished.Result == nil || finished.Result.Success != 2 || finished.Result.Fail != 1 || posts.Load() != 3 { + t.Fatalf("finished=%#v aggregate=%#v posts=%d", finished, finished.Result, posts.Load()) + } + first := readResultPage(t, s, created.ID, 0, 2) + if len(first.Results) != 2 || !first.HasMore || first.NextCursor <= 0 || first.Results[0].Attempt != 1 || first.Results[0].Success || first.Results[0].Error == nil || first.Results[1].Attempt != 2 || !first.Results[1].Success { + t.Fatalf("first page=%#v", first) + } + second := readResultPage(t, s, created.ID, first.NextCursor, 2) + if len(second.Results) != 1 || second.HasMore || second.NextCursor <= first.NextCursor || second.Results[0].Attempt != 3 || !second.Results[0].Success { + t.Fatalf("second page=%#v", second) + } + empty := readResultPage(t, s, created.ID, second.NextCursor, 2) + if empty.Results == nil || len(empty.Results) != 0 || empty.HasMore || empty.NextCursor != second.NextCursor { + t.Fatalf("empty page=%#v", empty) + } + if err := s.Close(); err != nil { + t.Fatal(err) + } + reopened, err := New(Config{DBPath: path}) + if err != nil { + t.Fatal(err) + } + defer reopened.Close() + page := readResultPage(t, reopened, created.ID, 0, 200) + if len(page.Results) != 3 || page.NextCursor != second.NextCursor || page.HasMore { + t.Fatalf("persisted page=%#v", page) + } + if task, ok := reopened.manager.Get(created.ID); !ok || task.Status != service.TaskFailed || task.FinishedAt == nil { + t.Fatalf("persisted task=%#v", task) + } +} + +func TestStoppingTaskCancelsInflightSubmissionWithoutFailureResult(t *testing.T) { + started := make(chan struct{}) + var posts atomic.Int32 + s := taskResultsServer(t, t.TempDir()+"/stop.db", func(request *http.Request) (*http.Response, error) { + if request.Method == http.MethodPost { + posts.Add(1) + close(started) + <-request.Context().Done() + return nil, request.Context().Err() + } + return &http.Response{StatusCode: 200, Header: make(http.Header), Body: io.NopCloser(strings.NewReader(taskResultsHTML)), Request: request}, nil + }) + httpServer := httptest.NewServer(s.Handler()) + defer httpServer.Close() + created := createResultTask(t, httpServer.URL, 100) + select { + case <-started: + case <-time.After(3 * time.Second): + t.Fatal("submission did not start") + } + rec := httptest.NewRecorder() + s.Handler().ServeHTTP(rec, httptest.NewRequest(http.MethodPost, "/api/v1/tasks/"+created.ID+"/stop", nil)) + if rec.Code != http.StatusOK { + t.Fatalf("stop status=%d body=%s", rec.Code, rec.Body.String()) + } + finished := waitForTask(t, s, created.ID) + page := readResultPage(t, s, created.ID, 0, 200) + if finished.Status != service.TaskStopped || finished.Result == nil || !finished.Result.Stopped || finished.Result.Fail != 0 || finished.Result.Success != 0 || len(page.Results) != 0 || posts.Load() != 1 { + t.Fatalf("finished=%#v aggregate=%#v page=%#v posts=%d", finished, finished.Result, page, posts.Load()) + } + s.manager.StopAll() + saved, errs := s.store.LoadTasks() + if len(errs) != 0 || len(saved) != 1 || saved[0].Status != service.TaskStopped || saved[0].FinishedAt == nil || saved[0].Result == nil || !saved[0].Result.Stopped { + t.Fatalf("persisted stopped task=%#v errors=%v", saved, errs) + } +} + +func TestTaskResultPageRejectsInvalidQueries(t *testing.T) { + s, _ := submissionTestServer(t, true) + for _, test := range []struct { + query string + status int + }{ + {"?after=-1", 400}, {"?after=x", 400}, {"?limit=-1", 400}, {"?limit=x", 400}, {"", 404}, + } { + rec := httptest.NewRecorder() + s.Handler().ServeHTTP(rec, httptest.NewRequest(http.MethodGet, "/api/v1/tasks/missing/results"+test.query, nil)) + if rec.Code != test.status { + t.Fatalf("query=%s status=%d body=%s", test.query, rec.Code, rec.Body.String()) + } + } +} diff --git a/pkg/surveycore/client.go b/pkg/surveycore/client.go index 4f284b0..45b24ca 100644 --- a/pkg/surveycore/client.go +++ b/pkg/surveycore/client.go @@ -95,7 +95,9 @@ func (client *Client) DefaultConfig(ctx context.Context, surveyURL string) (*mod return client.inner.DefaultConfig(ctx, surveyURL) } -// Run submits responses according to cfg. +// Run submits cfg.Target attempts without automatic retries and returns totals. +// Individual failures do not stop later attempts unless cfg.FailStop is enabled. +// If any attempt fails, Run returns the first error after the run finishes. func (client *Client) Run(ctx context.Context, cfg *model.RunRequest) (*RunResult, error) { return client.inner.Run(ctx, cfg) } @@ -104,6 +106,7 @@ func (client *Client) Run(ctx context.Context, cfg *model.RunRequest) (*RunResul type EventHandler func(Event) // RunWithEvents submits responses and reports progress through handler. +// Completed attempts include Event.Result; ordinary progress events omit it. func (client *Client) RunWithEvents(ctx context.Context, cfg *model.RunRequest, handler EventHandler) (*RunResult, error) { if handler == nil { return client.inner.RunWithEvents(ctx, cfg, nil) diff --git a/pkg/surveycore/example_test.go b/pkg/surveycore/example_test.go index ddea3d7..2cac4b6 100644 --- a/pkg/surveycore/example_test.go +++ b/pkg/surveycore/example_test.go @@ -43,3 +43,18 @@ func ExampleClient_Submit() { } fmt.Println(result.Success, result.ElapsedSeconds) } + +func ExampleClient_RunResults() { + client := surveycore.New() + config := &model.RunRequest{ + SurveySource: model.SurveySource{URL: "https://www.wjx.cn/vm/example.aspx"}, + ExecutionPlan: model.ExecutionPlan{Target: 3, Threads: 2}, + } + for result, err := range client.RunResults(context.Background(), config) { + if err != nil { + fmt.Println(err) + return + } + fmt.Println(result.Attempt, result.Success, result.Error) + } +} diff --git a/pkg/surveycore/model/run_types.go b/pkg/surveycore/model/run_types.go index 185e35a..f2295ec 100644 --- a/pkg/surveycore/model/run_types.go +++ b/pkg/surveycore/model/run_types.go @@ -111,11 +111,12 @@ type ThreadProgress struct { // Event reports one execution progress update. type Event struct { - Worker string `json:"worker"` - Message string `json:"message"` - Success bool `json:"success"` - Fail bool `json:"fail"` - Current int `json:"current"` - Total int `json:"total"` - Time time.Time `json:"time"` + Worker string `json:"worker"` + Message string `json:"message"` + Success bool `json:"success"` + Fail bool `json:"fail"` + Current int `json:"current"` + Total int `json:"total"` + Time time.Time `json:"time"` + Result *SubmissionResult `json:"result,omitempty"` } diff --git a/pkg/surveycore/run_results.go b/pkg/surveycore/run_results.go new file mode 100644 index 0000000..911ae52 --- /dev/null +++ b/pkg/surveycore/run_results.go @@ -0,0 +1,55 @@ +package surveycore + +import ( + "context" + "iter" + + "github.com/SurveyController/SurveyCore/pkg/surveycore/model" +) + +// RunResults yields attempts in completion order with one-based attempt numbers. +// Platform failures are unsuccessful results with a nil iterator error. Setup and +// context errors terminate iteration with an error. Breaking the loop cancels +// outstanding requests and waits for workers before returning. +func (client *Client) RunResults(ctx context.Context, config *model.RunRequest) iter.Seq2[SubmissionResult, error] { + return func(yield func(SubmissionResult, error) bool) { + runCtx, cancel := context.WithCancel(ctx) + results := make(chan SubmissionResult) + done := make(chan struct{}) + var aggregate *RunResult + var runErr error + go func() { + defer close(done) + defer close(results) + aggregate, runErr = client.RunWithEvents(runCtx, config, func(event Event) { + if event.Result == nil { + return + } + select { + case results <- *event.Result: + case <-runCtx.Done(): + } + }) + }() + defer func() { + cancel() + <-done + }() + for result := range results { + if !yield(result, nil) { + return + } + } + <-done + if ctx.Err() != nil { + yield(SubmissionResult{}, ctx.Err()) + } else if aggregate == nil && runErr != nil { + yield(SubmissionResult{}, runErr) + } + } +} + +// RunResults uses a default Client to yield results from a run. +func RunResults(ctx context.Context, config *model.RunRequest) iter.Seq2[SubmissionResult, error] { + return New().RunResults(ctx, config) +} diff --git a/pkg/surveycore/run_results_test.go b/pkg/surveycore/run_results_test.go new file mode 100644 index 0000000..61e61d1 --- /dev/null +++ b/pkg/surveycore/run_results_test.go @@ -0,0 +1,130 @@ +package surveycore_test + +import ( + "context" + "encoding/json" + "errors" + "io" + "net/http" + "strings" + "sync/atomic" + "testing" + "time" + + "github.com/SurveyController/SurveyCore/pkg/surveycore" + "github.com/SurveyController/SurveyCore/pkg/surveycore/model" +) + +const runResultsHTML = `Run results
Choice
A
B
` + +func runResultsResponse(request *http.Request, body string) (*http.Response, error) { + return &http.Response{StatusCode: 200, Header: make(http.Header), Body: io.NopCloser(strings.NewReader(body)), Request: request}, nil +} + +func TestRunResultsContinuesAfterFailureWithoutModifyingConfig(t *testing.T) { + var gets, posts atomic.Int32 + client := surveycore.New(surveycore.WithHTTPClient(&http.Client{Transport: submissionTransport(func(request *http.Request) (*http.Response, error) { + if request.Method != http.MethodPost { + gets.Add(1) + return runResultsResponse(request, runResultsHTML) + } + if posts.Add(1) == 1 { + return runResultsResponse(request, "1〒1〒失败") + } + return runResultsResponse(request, "10") + })})) + config := &model.RunRequest{SurveySource: model.SurveySource{URL: "https://www.wjx.cn/vm/demo.aspx"}, ExecutionPlan: model.ExecutionPlan{Target: 3, Threads: 1}} + original, _ := json.Marshal(config) + var results []surveycore.SubmissionResult + for result, err := range client.RunResults(context.Background(), config) { + if err != nil { + t.Fatal(err) + } + results = append(results, result) + } + if len(results) != 3 || gets.Load() != 1 || posts.Load() != 3 { + t.Fatalf("results=%#v gets=%d posts=%d", results, gets.Load(), posts.Load()) + } + for index, result := range results { + if result.Attempt != index+1 || result.Provider != "wjx" || result.Success != (index > 0) || (result.Error == nil) != result.Success { + t.Fatalf("result=%#v", result) + } + } + current, _ := json.Marshal(config) + if string(original) != string(current) { + t.Fatal("run modified caller config") + } +} + +func TestBreakingRunResultsCancelsAndDrainsWorkers(t *testing.T) { + ctx, cancel := context.WithTimeout(context.Background(), 3*time.Second) + defer cancel() + secondStarted := make(chan struct{}) + var posts, active atomic.Int32 + client := surveycore.New(surveycore.WithHTTPClient(&http.Client{Transport: submissionTransport(func(request *http.Request) (*http.Response, error) { + if request.Method != http.MethodPost { + return runResultsResponse(request, runResultsHTML) + } + active.Add(1) + defer active.Add(-1) + switch posts.Add(1) { + case 1: + select { + case <-secondStarted: + return runResultsResponse(request, "10") + case <-request.Context().Done(): + return nil, request.Context().Err() + } + case 2: + close(secondStarted) + } + <-request.Context().Done() + return nil, request.Context().Err() + })})) + config := &model.RunRequest{SurveySource: model.SurveySource{URL: "https://www.wjx.cn/vm/demo.aspx"}, ExecutionPlan: model.ExecutionPlan{Target: 100, Threads: 2}} + count := 0 + for result, err := range client.RunResults(ctx, config) { + if err != nil || !result.Success { + t.Fatalf("result=%#v error=%v", result, err) + } + count++ + break + } + if count != 1 || active.Load() != 0 || posts.Load() < 2 || posts.Load() >= 100 || ctx.Err() != nil { + t.Fatalf("count=%d active=%d posts=%d context=%v", count, active.Load(), posts.Load(), ctx.Err()) + } +} + +func TestRunResultsReportsSetupAndContextErrors(t *testing.T) { + count := 0 + for _, err := range surveycore.RunResults(context.Background(), nil) { + count++ + if !errors.Is(err, surveycore.ErrInvalidConfig) { + t.Fatalf("setup error=%v", err) + } + } + if count != 1 { + t.Fatalf("setup error count=%d", count) + } + ctx, cancel := context.WithCancel(context.Background()) + var posts atomic.Int32 + client := surveycore.New(surveycore.WithHTTPClient(&http.Client{Transport: submissionTransport(func(request *http.Request) (*http.Response, error) { + if request.Method != http.MethodPost { + return runResultsResponse(request, runResultsHTML) + } + posts.Add(1) + cancel() + return nil, request.Context().Err() + })})) + config := &model.RunRequest{SurveySource: model.SurveySource{URL: "https://www.wjx.cn/vm/demo.aspx"}, ExecutionPlan: model.ExecutionPlan{Target: 3, Threads: 1}} + count = 0 + for _, err := range client.RunResults(ctx, config) { + count++ + if !errors.Is(err, context.Canceled) { + t.Fatalf("context error=%v", err) + } + } + if count != 1 || posts.Load() != 1 { + t.Fatalf("context error count=%d posts=%d", count, posts.Load()) + } +} diff --git a/pkg/surveycore/runtime/engine.go b/pkg/surveycore/runtime/engine.go index a54a49d..bc739cc 100644 --- a/pkg/surveycore/runtime/engine.go +++ b/pkg/surveycore/runtime/engine.go @@ -76,7 +76,7 @@ feedLoop: workers.Wait() result := state.result() - if ctx.Err() != nil && firstErr == nil { + if ctx.Err() != nil { firstErr = ctx.Err() } result.Stopped = errors.Is(firstErr, context.Canceled) || errors.Is(firstErr, context.DeadlineExceeded) diff --git a/pkg/surveycore/runtime/engine_control.go b/pkg/surveycore/runtime/engine_control.go index da8479d..634b8b8 100644 --- a/pkg/surveycore/runtime/engine_control.go +++ b/pkg/surveycore/runtime/engine_control.go @@ -2,11 +2,34 @@ package runtime import ( "context" + "math/rand" "time" "github.com/SurveyController/SurveyCore/pkg/surveycore/model" ) +func waitAttemptInterval(ctx context.Context, interval [2]int, state *executionState, workerIndex int, workerName string, wait func(context.Context, time.Duration) error) error { + low := maxInt(0, interval[0]) + high := maxInt(low, interval[1]) + if high == 0 { + return ctx.Err() + } + seconds := float64(low) + rand.Float64()*float64(high-low) + duration := time.Duration(seconds * float64(time.Second)) + state.setProgress(workerIndex, workerName, "等待提交间隔", true) + if wait != nil { + return wait(ctx, duration) + } + timer := time.NewTimer(duration) + defer timer.Stop() + select { + case <-ctx.Done(): + return ctx.Err() + case <-timer.C: + return nil + } +} + func waitSubmitInterval(ctx context.Context, interval [2]int, state *executionState, workerIndex int, workerName string) { seconds := model.SampleSubmitIntervalSeconds(interval) if seconds <= 0 { diff --git a/pkg/surveycore/runtime/engine_progress.go b/pkg/surveycore/runtime/engine_progress.go index 6435a5f..0e10cbb 100644 --- a/pkg/surveycore/runtime/engine_progress.go +++ b/pkg/surveycore/runtime/engine_progress.go @@ -4,10 +4,13 @@ import ( "fmt" "sync" "time" + + "github.com/SurveyController/SurveyCore/pkg/surveycore/model" ) type executionState struct { mu sync.Mutex + resultMu sync.Mutex target int success int fail int @@ -116,6 +119,16 @@ func (s *executionState) callHandler(event Event) { } } +func (s *executionState) emitResult(worker string, result model.SubmissionResult) { + // Keep completion records ordered even when progress callbacks run concurrently. + s.resultMu.Lock() + defer s.resultMu.Unlock() + s.mu.Lock() + current := s.success + s.fail + s.mu.Unlock() + s.callHandler(Event{Worker: worker, Message: "本次提交完成", Success: result.Success, Fail: !result.Success, Current: current, Total: s.target, Time: s.now(), Result: &result}) +} + func (s *executionState) result() *RunResult { s.mu.Lock() defer s.mu.Unlock() diff --git a/pkg/surveycore/runtime/engine_types.go b/pkg/surveycore/runtime/engine_types.go index 6c0e513..74d710c 100644 --- a/pkg/surveycore/runtime/engine_types.go +++ b/pkg/surveycore/runtime/engine_types.go @@ -37,21 +37,24 @@ type PauseController interface { } type ExecutionOptions struct { - Target int - Threads int - MaxRetries int - FailStop bool - FailStopThreshold int - UseRandomIP bool - RetryDelay time.Duration - CooldownOnError time.Duration - LeaseManager LeaseManager - PauseController PauseController - Now func() time.Time - UserAgent model.UserAgentSettings - AIProfile model.AIProfile - AnswerRuntime model.AnswerRuntime - ConfigureJob func(ctx context.Context, jobIndex int, attempt int, job *JobRequest) error + Target int + Threads int + MaxRetries int + ContinueOnFailure bool + IntervalBeforeAttempt bool + WaitInterval func(context.Context, time.Duration) error + FailStop bool + FailStopThreshold int + UseRandomIP bool + RetryDelay time.Duration + CooldownOnError time.Duration + LeaseManager LeaseManager + PauseController PauseController + Now func() time.Time + UserAgent model.UserAgentSettings + AIProfile model.AIProfile + AnswerRuntime model.AnswerRuntime + ConfigureJob func(ctx context.Context, jobIndex int, attempt int, job *JobRequest) error } type JobRequest struct { diff --git a/pkg/surveycore/runtime/engine_worker.go b/pkg/surveycore/runtime/engine_worker.go index b73915a..37f1b3f 100644 --- a/pkg/surveycore/runtime/engine_worker.go +++ b/pkg/surveycore/runtime/engine_worker.go @@ -2,7 +2,9 @@ package runtime import ( "context" + "errors" "fmt" + "time" "github.com/SurveyController/SurveyCore/pkg/surveycore/internal/answerplan" "github.com/SurveyController/SurveyCore/pkg/surveycore/model" @@ -27,16 +29,33 @@ func runExecutionWorker( if waitIfExecutionPaused(ctx, options, state, workerIndex, workerName) { return } - if hasSubmitted { + if options.IntervalBeforeAttempt && jobIndex > 0 { + if err := waitAttemptInterval(ctx, config.ExecutionPlan.SubmitInterval, state, workerIndex, workerName, options.WaitInterval); err != nil { + recordError(err) + cancel() + return + } + } else if !options.IntervalBeforeAttempt && hasSubmitted { waitSubmitInterval(ctx, config.ExecutionPlan.SubmitInterval, state, workerIndex, workerName) } if waitIfExecutionPaused(ctx, options, state, workerIndex, workerName) || ctx.Err() != nil { return } + started := time.Now() consecutiveFailures, err := runOneJob(ctx, config, submit, options, state, workerIndex, workerName, jobIndex) + // Retain completed results if cancellation arrives just after the attempt. + contextErr := ctx.Err() + if contextErr == nil || !errors.Is(err, contextErr) { + result := model.SubmissionResult{Attempt: jobIndex + 1, Provider: config.SurveySource.Provider, Success: err == nil, ElapsedSeconds: time.Since(started).Seconds()} + if err != nil { + message := err.Error() + result.Error = &message + } + state.emitResult(workerName, result) + } if err != nil { recordError(err) - if !isRetryableRunError(err) || (options.FailStop && consecutiveFailures >= failStopThreshold(options)) { + if (!options.ContinueOnFailure && !isRetryableRunError(err)) || (options.FailStop && consecutiveFailures >= failStopThreshold(options)) { cancel() } } else { @@ -61,6 +80,9 @@ func runOneJob(ctx context.Context, cfg *RunRequest, submit SubmitFunc, options owner := fmt.Sprintf("%s-%d-%d", workerName, jobIndex+1, attempt) lease, leased, leaseErr := acquireExecutionLease(ctx, options, state, workerIndex, workerName, owner) if leaseErr != nil { + if ctx.Err() != nil { + return 0, ctx.Err() + } if shouldRetry(leaseErr, attempt, attempts) { state.emit(workerName, "代理不可用，准备重试", false, true) sleepRetry(ctx, options.RetryDelay) @@ -92,6 +114,9 @@ func runOneJob(ctx context.Context, cfg *RunRequest, submit SubmitFunc, options if err := options.ConfigureJob(ctx, jobIndex, attempt, &job); err != nil { releaseExecutionLease(options, owner, lease, leased, err) resetPendingAnswerRuntime(job.Submission.Context.Runtime, owner) + if ctx.Err() != nil { + return 0, ctx.Err() + } return state.addFail(workerIndex, workerName, "配置失败"), err } } @@ -112,6 +137,10 @@ func runOneJob(ctx context.Context, cfg *RunRequest, submit SubmitFunc, options }) releaseExecutionLease(options, owner, lease, leased, err) finalizeAnswerRuntime(job.Submission.Context.Runtime, owner, err == nil) + if err != nil && ctx.Err() != nil { + state.setProgress(workerIndex, workerName, "已停止", false) + return 0, ctx.Err() + } if err == nil { state.addSuccess(workerIndex, workerName, "提交成功") diff --git a/pkg/surveycore/runtime/run.go b/pkg/surveycore/runtime/run.go index 296c71d..f55e30b 100644 --- a/pkg/surveycore/runtime/run.go +++ b/pkg/surveycore/runtime/run.go @@ -19,7 +19,11 @@ func (c *Client) Run(ctx context.Context, cfg *RunRequest) (*RunResult, error) { } func (c *Client) RunWithEvents(ctx context.Context, cfg *RunRequest, handler EventHandler) (*RunResult, error) { - return c.RunWithExecutionOptions(ctx, cfg, handler, ExecutionOptionsFromConfig(cfg)) + result, err := c.RunWithExecutionOptions(ctx, cfg, handler, ExecutionOptionsFromConfig(cfg)) + if err != nil && ctx.Err() != nil { + err = ctx.Err() + } + return result, err } func (c *Client) RunWithExecutionOptions(ctx context.Context, cfg *RunRequest, handler EventHandler, options ExecutionOptions) (*RunResult, error) { @@ -29,14 +33,18 @@ func (c *Client) RunWithExecutionOptions(ctx context.Context, cfg *RunRequest, h if strings.TrimSpace(cfg.SurveySource.URL) == "" { return nil, fmt.Errorf("%w: 必须提供问卷链接", ErrInvalidConfig) } - provider := detectProvider(cfg.SurveySource.URL) - if cfg.SurveySource.Provider != "" { - provider = cfg.SurveySource.Provider + provider, err := submissionProvider(cfg) + if err != nil { + return nil, err + } + if ctx.Err() != nil { + return nil, ctx.Err() } runCfg, runOptions, err := c.prepareReverseFillExecution(ctx, cfg, provider, options) if err != nil { return nil, err } + runCfg.SurveySource.Provider = provider if err := prepareAnswerDatetimeWindowExecution(runCfg, provider); err != nil { return nil, err } @@ -69,6 +77,7 @@ func (c *Client) RunWithExecutionOptions(ctx context.Context, cfg *RunRequest, h if len(runCfg.SurveyDefinition.Questions) == 0 { runCfg.SurveyDefinition = prepared.Definition } + ensureQuestionStrategies(runCfg) emitPrepared(handler) result, err := RunExecution(ctx, runCfg, func(runCtx context.Context, local *model.SubmissionRequest, localHandler EventHandler) (*RunResult, error) { localRunner := runner @@ -103,6 +112,7 @@ func (c *Client) RunWithExecutionOptions(ctx context.Context, cfg *RunRequest, h if len(runCfg.SurveyDefinition.Questions) == 0 { runCfg.SurveyDefinition = prepared.Definition } + ensureQuestionStrategies(runCfg) emitPrepared(handler) result, err := RunExecution(ctx, runCfg, func(runCtx context.Context, local *model.SubmissionRequest, localHandler EventHandler) (*RunResult, error) { localRunner := runner @@ -139,6 +149,7 @@ func (c *Client) RunWithExecutionOptions(ctx context.Context, cfg *RunRequest, h if len(runCfg.SurveyDefinition.Questions) == 0 { runCfg.SurveyDefinition = prepared.Definition } + ensureQuestionStrategies(runCfg) emitPrepared(handler) result, err := RunExecution(ctx, runCfg, func(runCtx context.Context, local *model.SubmissionRequest, localHandler EventHandler) (*RunResult, error) { localRunner := runner diff --git a/pkg/surveycore/runtime/run_options.go b/pkg/surveycore/runtime/run_options.go index 72621c1..277f66e 100644 --- a/pkg/surveycore/runtime/run_options.go +++ b/pkg/surveycore/runtime/run_options.go @@ -19,17 +19,14 @@ func ExecutionOptionsFromConfig(cfg *RunRequest) ExecutionOptions { if threads <= 0 { threads = 1 } - maxRetries := 0 - if cfg.PsychometricPolicy.Enabled { - maxRetries = 1 - } options := ExecutionOptions{ - Target: target, - Threads: threads, - MaxRetries: maxRetries, - FailStop: cfg.FailStop, - FailStopThreshold: defaultFailStopThreshold, - CooldownOnError: 30 * time.Second, + Target: target, + Threads: threads, + ContinueOnFailure: true, + IntervalBeforeAttempt: true, + FailStop: cfg.FailStop, + FailStopThreshold: defaultFailStopThreshold, + CooldownOnError: 30 * time.Second, } if cfg.UserAgent != nil { options.UserAgent = *cfg.UserAgent diff --git a/pkg/surveycore/runtime/run_results_test.go b/pkg/surveycore/runtime/run_results_test.go new file mode 100644 index 0000000..d41fa87 --- /dev/null +++ b/pkg/surveycore/runtime/run_results_test.go @@ -0,0 +1,198 @@ +package runtime + +import ( + "context" + "errors" + "fmt" + "strconv" + "strings" + "sync/atomic" + "testing" + "time" + + "github.com/SurveyController/SurveyCore/pkg/surveycore/internal/runerror" + "github.com/SurveyController/SurveyCore/pkg/surveycore/model" +) + +func tagAttempt(_ context.Context, index, _ int, job *JobRequest) error { + job.Submission.Source.URL = fmt.Sprintf("https://example.test/%d", index+1) + return nil +} + +func attemptNumber(request *model.SubmissionRequest) int { + value, _ := strconv.Atoi(strings.TrimPrefix(request.Source.URL, "https://example.test/")) + return value +} + +func TestExecutionResultsUseCompletionOrderAndBoundedConcurrency(t *testing.T) { + ctx, cancel := context.WithTimeout(context.Background(), 3*time.Second) + defer cancel() + firstStarted := make(chan struct{}) + releaseFirst := make(chan struct{}) + var active, maximum atomic.Int32 + var results []model.SubmissionResult + config := executionTestConfig(3, 2) + config.SurveySource.Provider = ProviderWJX + options := ExecutionOptionsFromConfig(config) + options.ConfigureJob = tagAttempt + result, err := RunExecution(ctx, config, func(ctx context.Context, request *model.SubmissionRequest, _ EventHandler) (*RunResult, error) { + current := active.Add(1) + defer active.Add(-1) + for old := maximum.Load(); current > old && !maximum.CompareAndSwap(old, current); old = maximum.Load() { + } + if attemptNumber(request) == 1 { + close(firstStarted) + select { + case <-releaseFirst: + case <-ctx.Done(): + return nil, ctx.Err() + } + } else { + select { + case <-firstStarted: + case <-ctx.Done(): + return nil, ctx.Err() + } + } + return &RunResult{Success: 1}, nil + }, func(event Event) { + if event.Result == nil { + return + } + results = append(results, *event.Result) + if event.Result.Attempt == 3 { + close(releaseFirst) + } + }, options) + if err != nil || result.Success != 3 || maximum.Load() != 2 || active.Load() != 0 { + t.Fatalf("result=%#v error=%v maximum=%d active=%d", result, err, maximum.Load(), active.Load()) + } + if len(results) != 3 || results[0].Attempt != 2 || results[1].Attempt != 3 || results[2].Attempt != 1 { + t.Fatalf("completion order = %#v", results) + } + for _, item := range results { + if !item.Success || item.Error != nil || item.Provider != ProviderWJX || item.ElapsedSeconds < 0 { + t.Fatalf("result metadata = %#v", item) + } + } +} + +func TestDefaultRunContinuesAfterErrorsAndWaitsBeforeEachLaterAttempt(t *testing.T) { + config := executionTestConfig(4, 1) + config.PsychometricPolicy.Enabled = true + config.SubmitInterval = [2]int{2, 4} + options := ExecutionOptionsFromConfig(config) + options.ConfigureJob = tagAttempt + var waits []time.Duration + options.WaitInterval = func(_ context.Context, duration time.Duration) error { + waits = append(waits, duration) + return nil + } + var calls int + var results []model.SubmissionResult + result, err := RunExecution(context.Background(), config, func(_ context.Context, request *model.SubmissionRequest, _ EventHandler) (*RunResult, error) { + calls++ + if attemptNumber(request)%2 == 1 { + return nil, runerror.Wrap(runerror.KindConfig, errors.New("attempt configuration failed")) + } + return &RunResult{Success: 1}, nil + }, func(event Event) { + if event.Result != nil { + results = append(results, *event.Result) + } + }, options) + if err == nil || calls != 4 || len(waits) != 3 || len(results) != 4 || result.Success != 2 || result.Fail != 2 || options.MaxRetries != 0 { + t.Fatalf("result=%#v error=%v calls=%d waits=%v results=%#v", result, err, calls, waits, results) + } + for _, duration := range waits { + if duration < 2*time.Second || duration > 4*time.Second { + t.Fatalf("interval outside Python range: %v", duration) + } + } + if results[0].Success || results[0].Error == nil || !results[1].Success || results[1].Error != nil { + t.Fatalf("failure did not remain in its own result: %#v", results) + } +} + +func TestLaterAttemptWaitsEvenOnANewWorker(t *testing.T) { + config := executionTestConfig(2, 2) + config.SubmitInterval = [2]int{1, 1} + options := ExecutionOptionsFromConfig(config) + options.ConfigureJob = tagAttempt + waiting := make(chan struct{}) + release := make(chan struct{}) + options.WaitInterval = func(ctx context.Context, duration time.Duration) error { + if duration != time.Second { + t.Errorf("interval=%v", duration) + } + close(waiting) + select { + case <-release: + return nil + case <-ctx.Done(): + return ctx.Err() + } + } + ctx, cancel := context.WithTimeout(context.Background(), 3*time.Second) + defer cancel() + result, err := RunExecution(ctx, config, func(ctx context.Context, request *model.SubmissionRequest, _ EventHandler) (*RunResult, error) { + if attemptNumber(request) == 1 { + select { + case <-waiting: + close(release) + case <-ctx.Done(): + return nil, ctx.Err() + } + } + return &RunResult{Success: 1}, nil + }, nil, options) + if err != nil || result.Success != 2 { + t.Fatalf("result=%#v error=%v", result, err) + } +} + +func TestCancellationDuringIntervalDoesNotCreateAResult(t *testing.T) { + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + config := executionTestConfig(3, 1) + config.SubmitInterval = [2]int{100, 100} + options := ExecutionOptionsFromConfig(config) + options.WaitInterval = func(ctx context.Context, _ time.Duration) error { + cancel() + return ctx.Err() + } + calls, records := 0, 0 + result, err := RunExecution(ctx, config, func(context.Context, *model.SubmissionRequest, EventHandler) (*RunResult, error) { + calls++ + return &RunResult{Success: 1}, nil + }, func(event Event) { + if event.Result != nil { + records++ + } + }, options) + if !errors.Is(err, context.Canceled) || !result.Stopped || result.Success != 1 || result.Fail != 0 || calls != 1 || records != 1 { + t.Fatalf("result=%#v error=%v calls=%d records=%d", result, err, calls, records) + } +} + +func TestCancellationTakesPriorityOverAnEarlierAttemptError(t *testing.T) { + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + config := executionTestConfig(3, 1) + calls, records := 0, 0 + result, err := RunExecution(ctx, config, func(context.Context, *model.SubmissionRequest, EventHandler) (*RunResult, error) { + calls++ + if calls == 1 { + return nil, errors.New("first attempt failed") + } + cancel() + return nil, ctx.Err() + }, func(event Event) { + if event.Result != nil { + records++ + } + }, ExecutionOptionsFromConfig(config)) + if !errors.Is(err, context.Canceled) || !result.Stopped || result.Fail != 1 || calls != 2 || records != 1 { + t.Fatalf("result=%#v error=%v calls=%d records=%d", result, err, calls, records) + } +} diff --git a/tests/consumer/sdk_test.go b/tests/consumer/sdk_test.go index 4131145..f20e0fa 100644 --- a/tests/consumer/sdk_test.go +++ b/tests/consumer/sdk_test.go @@ -52,6 +52,16 @@ func TestPublicPackagesCompileForExternalConsumer(t *testing.T) { if submission != nil || !errors.Is(err, surveycore.ErrInvalidConfig) || surveycore.ClassifySubmissionError(err) != surveycore.ErrorKindConfig { t.Fatalf("single submission contract: result=%#v error=%v", submission, err) } + count := 0 + for _, err := range client.RunResults(context.Background(), nil) { + count++ + if !errors.Is(err, surveycore.ErrInvalidConfig) { + t.Fatalf("run iterator error=%v", err) + } + } + if count != 1 { + t.Fatalf("run iterator errors=%d", count) + } pool := proxy.NewPool(proxy.PoolOptions{}) if pool == nil {