108 lines
4.2 KiB
Go
108 lines
4.2 KiB
Go
package orbit
|
||
|
||
import (
|
||
"context"
|
||
"encoding/json"
|
||
"io"
|
||
"net/http"
|
||
"net/http/httptest"
|
||
"net/http/httputil"
|
||
"net/url"
|
||
"strings"
|
||
"sync/atomic"
|
||
"testing"
|
||
"time"
|
||
)
|
||
|
||
func TestIntegrationTaskBatchSDK(t *testing.T) {
|
||
api, _, _ := integrationWorkerClient(t)
|
||
ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
|
||
defer cancel()
|
||
created, err := api.CreateQueue(ctx, workerTestID(t), CreateQueueRequest{Name: "batch." + workerTestID(t), MaxConcurrency: 2, RatePerSecond: 10, MaxUnfinishedTasks: 2})
|
||
if err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
q := created.Queue
|
||
if _, err := api.ConfigureQueue(ctx, q.ID, workerTestID(t), ConfigureQueueRequest{ExpectedRevision: q.Revision, Reason: "batch acceptance regression", Paused: true, MaxConcurrency: 2, RatePerSecond: 10, MaxUnfinishedTasks: 2}); err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
input := SubmitTaskRequest{QueueID: q.ID, Handler: "file.convert", HandlerVersion: "v1", PayloadVersion: "v1", Payload: json.RawMessage(`{"n":9007199254740993}`)}
|
||
in := SubmitTasksRequest{Items: []SubmitTasksItem{{IdempotencyKey: workerTestID(t), Task: input}, {IdempotencyKey: workerTestID(t), Task: input}, {IdempotencyKey: workerTestID(t), Task: input}}}
|
||
// 真实HTTP代理先完整读到数据库受理回执，再强制断开客户端连接，丢弃整批响应。
|
||
target, _ := url.Parse(api.baseURL)
|
||
proxy := httputil.NewSingleHostReverseProxy(target)
|
||
var dropped atomic.Bool
|
||
var calls atomic.Int32
|
||
proxy.ModifyResponse = func(r *http.Response) error {
|
||
if r.Request.URL.Path != "/v1/tasks:batch" {
|
||
return nil
|
||
}
|
||
calls.Add(1)
|
||
if dropped.CompareAndSwap(false, true) {
|
||
raw, err := io.ReadAll(r.Body)
|
||
r.Body.Close()
|
||
if err != nil {
|
||
return err
|
||
}
|
||
if r.StatusCode != 200 || !strings.Contains(string(raw), `"ACCEPTED"`) || !strings.Contains(string(raw), `"QUEUE_FULL"`) {
|
||
t.Error("unexpected pre-drop server result", string(raw))
|
||
}
|
||
return io.ErrUnexpectedEOF
|
||
}
|
||
return nil
|
||
}
|
||
proxy.ErrorHandler = func(w http.ResponseWriter, r *http.Request, err error) {
|
||
conn, _, hijackErr := w.(http.Hijacker).Hijack()
|
||
if hijackErr == nil {
|
||
conn.Close()
|
||
}
|
||
}
|
||
server := httptest.NewServer(proxy)
|
||
defer server.Close()
|
||
client, err := NewClient(server.URL, Options{Token: api.token})
|
||
if err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
if _, err := client.SubmitTasks(ctx, in); err == nil || !dropped.Load() || calls.Load() != 1 {
|
||
t.Fatal("response loss not surfaced without silent retry", err)
|
||
}
|
||
// 原连接已关闭；用原键和原内容核对，前两项去重，满额项保持明确拒绝。
|
||
out, err := client.SubmitTasks(ctx, in)
|
||
if err != nil || len(out.Items) != 3 {
|
||
t.Fatal(out, err)
|
||
}
|
||
for i := 0; i < 2; i++ {
|
||
if out.Items[i].Status != BatchAccepted || !out.Items[i].Receipt.Deduplicated {
|
||
t.Fatal("lost response duplicated task", out)
|
||
}
|
||
}
|
||
if out.Items[2].Status != BatchRejected || out.Items[2].Error.Code != "QUEUE_FULL" || out.Items[2].Error.RetryAfterMS != 1000 {
|
||
t.Fatal(out)
|
||
}
|
||
page, err := api.ListTasks(ctx, ListTasksRequest{QueueID: q.ID})
|
||
if err != nil || len(page.Items) != 2 {
|
||
t.Fatal("duplicate durable rows", page, err)
|
||
}
|
||
single, err := api.SubmitTask(ctx, in.Items[0].IdempotencyKey, input)
|
||
if err != nil || !single.Deduplicated || single.Task.ID != out.Items[0].Receipt.TaskID || string(single.Task.Payload) != `{"n":9007199254740993}` {
|
||
t.Fatal("batch/single mismatch", err)
|
||
}
|
||
if _, err := api.CancelTask(ctx, single.Task.ID, workerTestID(t), CancelTaskRequest{ExpectedRevision: single.Task.Revision, Reason: "free slot for partial retry"}); err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
suffix, err := api.SubmitTasks(ctx, SubmitTasksRequest{Items: in.Items[2:]})
|
||
if err != nil || suffix.Items[0].Status != BatchAccepted || suffix.Items[0].Receipt.Deduplicated || suffix.Items[0].Index != 0 {
|
||
t.Fatal(suffix, err)
|
||
}
|
||
for _, id := range []string{out.Items[1].Receipt.TaskID, suffix.Items[0].Receipt.TaskID} {
|
||
current, err := api.GetTask(ctx, id)
|
||
if err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
if _, err := api.CancelTask(ctx, id, workerTestID(t), CancelTaskRequest{ExpectedRevision: current.Revision, Reason: "finish batch regression"}); err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
}
|
||
t.Log("lost real HTTP response: two committed tasks recovered by identity, no duplicates; backpressured suffix independently accepted after capacity released")
|
||
}
|