189 lines
7.6 KiB
Go
189 lines
7.6 KiB
Go
package orbit
|
|
|
|
import (
|
|
"context"
|
|
"encoding/json"
|
|
"errors"
|
|
"os"
|
|
"strings"
|
|
"testing"
|
|
"time"
|
|
)
|
|
|
|
func applicationIntegrationClient(t *testing.T, key string) (*Client, string) {
|
|
t.Helper()
|
|
base := os.Getenv("ORBIT_TEST_WORKER_URL")
|
|
path := os.Getenv(key)
|
|
if base == "" || path == "" {
|
|
t.Skip("requires disposable application resource fixtures")
|
|
}
|
|
if !strings.HasPrefix(base, "http://127.0.0.1:") {
|
|
t.Fatal("requires isolated loopback server")
|
|
}
|
|
raw, err := os.ReadFile(path)
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
var v struct {
|
|
Token string `json:"token"`
|
|
QueueID string `json:"queue_id"`
|
|
}
|
|
if json.Unmarshal(raw, &v) != nil || v.Token == "" || !validID(v.QueueID) {
|
|
t.Fatal("invalid private credentials")
|
|
}
|
|
api, err := NewClient(base, Options{Token: v.Token})
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
return api, v.QueueID
|
|
}
|
|
func TestIntegrationApplicationResourcesSDK(t *testing.T) {
|
|
a, qa := applicationIntegrationClient(t, "ORBIT_TEST_APP_CREDENTIALS_FILE")
|
|
b, qb := applicationIntegrationClient(t, "ORBIT_TEST_APP_SECOND_CREDENTIALS_FILE")
|
|
ctx, cancel := context.WithTimeout(context.Background(), 20*time.Second)
|
|
defer cancel()
|
|
before, err := a.GetApplicationResources(ctx)
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
peer, err := b.GetApplicationResources(ctx)
|
|
if err != nil || peer.ApplicationID != before.ApplicationID || qa == qb {
|
|
t.Fatal("invalid cross namespace fixture", err)
|
|
}
|
|
ordinary, _, _ := integrationWorkerClient(t)
|
|
if _, err := ordinary.GetApplicationResources(ctx); err == nil {
|
|
t.Fatal("queue admin inherited application permission")
|
|
} else {
|
|
var ae *APIError
|
|
if !errors.As(err, &ae) || ae.StatusCode != 403 {
|
|
t.Fatal(err)
|
|
}
|
|
}
|
|
key := workerTestID(t)
|
|
in := ConfigureApplicationResourcesRequest{ExpectedRevision: before.Revision, Reason: "SDK application quota regression", MaxConcurrency: 2, RatePerSecond: 10, MaxUnfinishedTasks: 1}
|
|
configured, err := a.ConfigureApplicationResources(ctx, key, in)
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
replay, err := a.ConfigureApplicationResources(ctx, key, in)
|
|
if err != nil || !replay.Deduplicated || replay.Resources.Revision != configured.Resources.Revision {
|
|
t.Fatal("configure replay failed", err)
|
|
}
|
|
if _, err := b.ConfigureApplicationResources(ctx, key, in); err == nil {
|
|
t.Fatal("same key in different namespace replayed authority")
|
|
} else {
|
|
var ae *APIError
|
|
if !errors.As(err, &ae) || ae.Code != "STATE_CONFLICT" {
|
|
t.Fatal(err)
|
|
}
|
|
}
|
|
var ignored ConfigureApplicationResourcesResult
|
|
for _, body := range []any{
|
|
map[string]any{"expected_revision": "2", "reason": "missing limits"},
|
|
map[string]any{"expected_revision": 2, "reason": "bad number", "max_unfinished_tasks": 1, "max_concurrency": 1, "rate_per_second": 1},
|
|
map[string]any{"expected_revision": "2", "reason": "override application", "max_unfinished_tasks": 1, "max_concurrency": 1, "rate_per_second": 1, "application_id": before.ApplicationID},
|
|
} {
|
|
if err := a.call(ctx, "POST", "/v1/application/resources:configure", workerTestID(t), body, &ignored); err == nil {
|
|
t.Fatal("invalid configuration accepted")
|
|
} else {
|
|
var ae *APIError
|
|
if !errors.As(err, &ae) || ae.StatusCode != 400 {
|
|
t.Fatal(err)
|
|
}
|
|
}
|
|
}
|
|
input := SubmitTaskRequest{QueueID: qa, Handler: "object.prepare", HandlerVersion: "v1", PayloadVersion: "v1", Payload: json.RawMessage(`{"n":9007199254740993}`)}
|
|
first, err := a.SubmitTask(ctx, key+"-one", input)
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
input.QueueID = qb
|
|
_, err = b.SubmitTask(ctx, key+"-two", input)
|
|
var ae *APIError
|
|
if !errors.As(err, &ae) || ae.Code != "APPLICATION_FULL" || ae.StatusCode != 429 || ae.RetryAfter != time.Second {
|
|
t.Fatal("missing application backpressure", err)
|
|
}
|
|
peer, err = b.GetApplicationResources(ctx)
|
|
if err != nil || peer.UnfinishedTasks != 1 || peer.Revision != configured.Resources.Revision {
|
|
t.Fatal(peer, err)
|
|
}
|
|
if _, err := a.CancelTask(ctx, first.Task.ID, key+"-cancel", CancelTaskRequest{ExpectedRevision: 1, Reason: "free shared slot"}); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
second, err := b.SubmitTask(ctx, key+"-two", input)
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if _, err := b.CancelTask(ctx, second.Task.ID, key+"-cancel-two", CancelTaskRequest{ExpectedRevision: 1, Reason: "finish regression"}); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
after, err := a.GetApplicationResources(ctx)
|
|
if err != nil || after.UnfinishedTasks != 0 || after.ActiveAttempts != 0 || after.PendingStarts != 0 {
|
|
t.Fatal(after, err)
|
|
}
|
|
if _, err := a.ConfigureApplicationResources(ctx, key+"-restore", ConfigureApplicationResourcesRequest{ExpectedRevision: configured.Resources.Revision, Reason: "restore sample defaults", MaxUnfinishedTasks: before.MaxUnfinishedTasks, MaxConcurrency: before.MaxConcurrency, RatePerSecond: before.RatePerSecond}); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
}
|
|
|
|
func TestIntegrationWorkerProcessApplicationBackpressure(t *testing.T) {
|
|
p := startOrbitProcess(t)
|
|
ca, qa := applicationIntegrationClient(t, "ORBIT_TEST_APP_CREDENTIALS_FILE")
|
|
cb, qb := applicationIntegrationClient(t, "ORBIT_TEST_APP_SECOND_CREDENTIALS_FILE")
|
|
a := processClient(t, p, ca.token)
|
|
ctx, cancel := context.WithTimeout(context.Background(), 20*time.Second)
|
|
defer cancel()
|
|
before, err := a.GetApplicationResources(ctx)
|
|
if err != nil || before.UnfinishedTasks != 0 {
|
|
t.Fatal("dedicated app must be idle", err)
|
|
}
|
|
key := workerTestID(t)
|
|
changed, err := a.ConfigureApplicationResources(ctx, key, ConfigureApplicationResourcesRequest{ExpectedRevision: before.Revision, Reason: "application crash regression", MaxUnfinishedTasks: 1, MaxConcurrency: 1, RatePerSecond: 1})
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
input := SubmitTaskRequest{QueueID: qa, Handler: "object.prepare", HandlerVersion: "v1", PayloadVersion: "v1", Payload: json.RawMessage(`{}`)}
|
|
first, err := a.SubmitTask(ctx, key+"-first", input)
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
p.kill(t)
|
|
next := startOrbitProcess(t)
|
|
if next.nodeID == p.nodeID {
|
|
t.Fatal("node identity reused")
|
|
}
|
|
a = processClient(t, next, ca.token)
|
|
b := processClient(t, next, cb.token)
|
|
v, err := b.GetApplicationResources(ctx)
|
|
if err != nil || v.Revision != changed.Resources.Revision || v.UnfinishedTasks != 1 || v.MaxUnfinishedTasks != 1 {
|
|
t.Fatal("restart lost shared configuration or usage", v, err)
|
|
}
|
|
if out, err := a.SubmitTask(ctx, key+"-first", input); err != nil || !out.Deduplicated || out.Task.ID != first.Task.ID {
|
|
t.Fatal("restart lost admission receipt", err)
|
|
}
|
|
input.QueueID = qb
|
|
_, err = b.SubmitTask(ctx, key+"-second", input)
|
|
var rejected *APIError
|
|
if !errors.As(err, &rejected) || rejected.Code != "APPLICATION_FULL" {
|
|
t.Fatal("restart allowed another namespace to exceed quota", err)
|
|
}
|
|
if _, err := a.CancelTask(ctx, first.Task.ID, key+"-cancel-first", CancelTaskRequest{ExpectedRevision: 1, Reason: "release after restart"}); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
second, err := b.SubmitTask(ctx, key+"-second", input)
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if _, err := b.CancelTask(ctx, second.Task.ID, key+"-cancel-second", CancelTaskRequest{ExpectedRevision: 1, Reason: "finish process regression"}); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
v, err = a.GetApplicationResources(ctx)
|
|
if err != nil || v.UnfinishedTasks != 0 || v.ActiveAttempts != 0 || v.PendingStarts != 0 {
|
|
t.Fatal("release after restart leaked usage", err)
|
|
}
|
|
if _, err := a.ConfigureApplicationResources(ctx, key+"-restore", ConfigureApplicationResourcesRequest{ExpectedRevision: changed.Resources.Revision, Reason: "restore defaults", MaxUnfinishedTasks: before.MaxUnfinishedTasks, MaxConcurrency: before.MaxConcurrency, RatePerSecond: before.RatePerSecond}); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
t.Log("server killed and exited; new node preserved shared quota and receipt; second namespace blocked until original task cancelled")
|
|
}
|