82 lines
3.3 KiB
Go
82 lines
3.3 KiB
Go
package orbit
|
|
|
|
import (
|
|
"context"
|
|
"encoding/json"
|
|
"strings"
|
|
"testing"
|
|
"time"
|
|
)
|
|
|
|
func TestIntegrationWorkerConnectionHistoryAndPayloadBounds(t *testing.T) {
|
|
client, queue, _ := integrationWorkerClient(t)
|
|
ctx, cancel := context.WithTimeout(context.Background(), 20*time.Second)
|
|
defer cancel()
|
|
handler := "history.bounds." + workerTestID(t)
|
|
// 原先默认 HTML 转义会把这份约 240 KiB 文档扩大到约 1.4 MiB。
|
|
payload := json.RawMessage(`{"text":"` + strings.Repeat("<>&", 80000) + `","n":9007199254740993}`)
|
|
in := SubmitTaskRequest{QueueID: queue, Handler: handler, HandlerVersion: "v1", PayloadVersion: "v1", Payload: payload, TaskTimeoutMS: 60000, Retry: RetryPolicy{MaxAttempts: 1, AttemptTimeoutMS: 30000}}
|
|
key := "history-bounds-" + workerTestID(t)
|
|
first, err := client.SubmitTask(ctx, key, in)
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
again, err := client.SubmitTask(ctx, key, in)
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if !again.Deduplicated || again.Task.ID != first.Task.ID || string(again.Task.Payload) != string(payload) {
|
|
t.Fatal("durable receipt inflated or altered payload")
|
|
}
|
|
r, err := NewWorkerRegistration(queue, 1, []WorkerCapability{{Handler: handler, HandlerVersion: "v1", PayloadVersion: "v1"}})
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
w, err := client.DialWorker(ctx, r)
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
defer w.Close()
|
|
if _, err := w.Capacity(ctx, workerTestID(t), WorkerCapacity{Sequence: 1, Capacity: 1}); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
assigned, err := w.NextAssignment(ctx)
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if assigned.Task.ID != first.Task.ID || string(assigned.Task.Payload) != string(payload) {
|
|
t.Fatal("assignment inflated or changed payload")
|
|
}
|
|
if _, err := w.Start(ctx, workerTestID(t), assigned.Attempt()); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
finish := WorkerFinish{Ref: assigned.Attempt(), Outcome: WorkerSucceeded, Result: payload}
|
|
if _, err := w.Finish(ctx, workerTestID(t), finish); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
page, err := client.ListTaskAttempts(ctx, first.Task.ID, HistoryPageRequest{Limit: 1})
|
|
if err != nil || len(page.Items) != 1 || page.Items[0].ID != assigned.AttemptID {
|
|
t.Fatalf("attempt list: %v", err)
|
|
}
|
|
detail, err := client.GetTaskAttempt(ctx, first.Task.ID, assigned.AttemptID)
|
|
if err != nil || detail.State != AttemptSucceeded || detail.WorkerInstanceID != r.InstanceID || string(detail.Result) != string(payload) {
|
|
t.Fatalf("attempt detail: %v", err)
|
|
}
|
|
finish.Result = json.RawMessage(strings.Replace(string(payload), "9007199254740993", "9007199254740995", 1))
|
|
if out, err := w.Finish(ctx, workerTestID(t), finish); err != nil || out.Receipt.Disposition != "EVIDENCE_ONLY" {
|
|
t.Fatalf("conflicting report: %v", err)
|
|
}
|
|
evidence, err := client.ListAttemptEvidence(ctx, first.Task.ID, assigned.AttemptID, HistoryPageRequest{Limit: 1})
|
|
if err != nil || len(evidence.Items) != 1 || evidence.Items[0].Reason != "RESULT_CONFLICT" {
|
|
t.Fatalf("evidence summary: %v", err)
|
|
}
|
|
e, err := client.GetAttemptEvidence(ctx, first.Task.ID, assigned.AttemptID, evidence.Items[0].ID)
|
|
if err != nil || string(e.Payload) != string(finish.Result) || e.Disposition != "EVIDENCE_ONLY" {
|
|
t.Fatalf("evidence detail: %v", err)
|
|
}
|
|
task, err := client.GetTask(ctx, first.Task.ID)
|
|
if err != nil || string(task.Result) != string(payload) || task.State != TaskSucceeded {
|
|
t.Fatalf("authoritative result changed: %v", err)
|
|
}
|
|
}
|