116 lines
4.0 KiB
Go
116 lines
4.0 KiB
Go
package orbit
|
||
|
||
import (
|
||
"context"
|
||
"encoding/json"
|
||
"errors"
|
||
"testing"
|
||
"time"
|
||
)
|
||
|
||
func TestIntegrationQueueResourcesSDK(t *testing.T) {
|
||
api, _, _ := integrationWorkerClient(t)
|
||
ctx, cancel := context.WithTimeout(context.Background(), 25*time.Second)
|
||
defer cancel()
|
||
name := "resources." + workerTestID(t)
|
||
created, err := api.CreateQueue(ctx, "create-"+name, CreateQueueRequest{Name: name, MaxConcurrency: 2, RatePerSecond: 1, MaxUnfinishedTasks: 2})
|
||
if err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
q := created.Queue
|
||
configure := func(paused bool) {
|
||
t.Helper()
|
||
out, err := api.ConfigureQueue(ctx, q.ID, workerTestID(t), ConfigureQueueRequest{ExpectedRevision: q.Revision, Reason: "SDK resource regression", Paused: paused, MaxConcurrency: 2, RatePerSecond: 1, MaxUnfinishedTasks: 2})
|
||
if err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
q = out.Queue
|
||
}
|
||
configure(true)
|
||
// 丢失 paused 不能被当成解除暂停；这是原始 HTTP 对象，SDK 本身总会发送 bool。
|
||
var ignored CreateQueueResult
|
||
err = api.call(ctx, "POST", "/v1/queues/"+q.ID+":configure", workerTestID(t), map[string]any{"expected_revision": "2", "reason": "missing pause", "max_concurrency": 2, "rate_per_second": 1, "max_unfinished_tasks": 2}, &ignored)
|
||
var apiErr *APIError
|
||
if !errors.As(err, &apiErr) || apiErr.StatusCode != 400 {
|
||
t.Fatalf("missing paused accepted: %v", err)
|
||
}
|
||
input := SubmitTaskRequest{QueueID: q.ID, Handler: name, HandlerVersion: "v1", PayloadVersion: "v1", Payload: json.RawMessage(`{"n":9007199254740993}`)}
|
||
first, err := api.SubmitTask(ctx, name+"-one", input)
|
||
if err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
second, err := api.SubmitTask(ctx, name+"-two", input)
|
||
if err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
_, err = api.SubmitTask(ctx, name+"-three", input)
|
||
if !errors.As(err, &apiErr) || apiErr.StatusCode != 429 || apiErr.Code != "QUEUE_FULL" || apiErr.RetryAfter != time.Second {
|
||
t.Fatalf("queue did not backpressure: %v", err)
|
||
}
|
||
if duplicate, err := api.SubmitTask(ctx, name+"-one", input); err != nil || !duplicate.Deduplicated || duplicate.Task.ID != first.Task.ID {
|
||
t.Fatal("full queue lost original identity")
|
||
}
|
||
if _, err := api.CancelTask(ctx, first.Task.ID, name+"-cancel", CancelTaskRequest{ExpectedRevision: first.Task.Revision, Reason: "free backlog slot"}); err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
third, err := api.SubmitTask(ctx, name+"-three", input)
|
||
if err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
w, err := NewWorker(api, WorkerOptions{QueueID: q.ID, Capacity: 2, Handlers: []WorkerHandler{runtimeHandler(name, func(ctx context.Context, e WorkerExecution) (json.RawMessage, error) { return e.Task.Payload, nil })}})
|
||
if err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
run, stop := context.WithCancel(ctx)
|
||
done := make(chan error, 1)
|
||
go func() { done <- w.Run(run) }()
|
||
waited := false
|
||
defer func() {
|
||
stop()
|
||
if !waited {
|
||
<-done
|
||
}
|
||
}()
|
||
select {
|
||
case <-time.After(350 * time.Millisecond):
|
||
case <-ctx.Done():
|
||
t.Fatal(ctx.Err())
|
||
}
|
||
current, err := api.GetTask(ctx, second.Task.ID)
|
||
if err != nil || current.AttemptCount != 0 {
|
||
t.Fatal("paused queue consumed attempt")
|
||
}
|
||
configure(false)
|
||
for _, id := range []string{second.Task.ID, third.Task.ID} {
|
||
finished := waitRuntimeState(t, ctx, api, id, TaskSucceeded)
|
||
if finished.AttemptCount != 1 || string(finished.Result) != `{"n":9007199254740993}` {
|
||
t.Fatal("resource controls altered execution")
|
||
}
|
||
}
|
||
var authorized []time.Time
|
||
for _, id := range []string{second.Task.ID, third.Task.ID} {
|
||
attempts, err := api.ListTaskAttempts(ctx, id, HistoryPageRequest{})
|
||
if err != nil || len(attempts.Items) != 1 || attempts.Items[0].AuthorizedAt == nil {
|
||
t.Fatal("missing start history")
|
||
}
|
||
authorized = append(authorized, *attempts.Items[0].AuthorizedAt)
|
||
}
|
||
gap := authorized[1].Sub(authorized[0])
|
||
if gap < 0 {
|
||
gap = -gap
|
||
}
|
||
if gap < time.Second {
|
||
t.Fatalf("start rate exceeded: %s", gap)
|
||
}
|
||
q, err = api.GetQueue(ctx, q.ID)
|
||
if err != nil || q.UnfinishedTasks != 0 || q.Revision != 3 {
|
||
t.Fatal("quota count or configuration revision drifted")
|
||
}
|
||
stop()
|
||
err = <-done
|
||
waited = true
|
||
if err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
}
|