426 lines
16 KiB
Go
426 lines
16 KiB
Go
package mysql
|
|
|
|
import (
|
|
"context"
|
|
"database/sql"
|
|
"encoding/json"
|
|
"os"
|
|
"sync"
|
|
"testing"
|
|
"time"
|
|
|
|
core "git.yuntuo.hk/k2safe/orbit/model"
|
|
"git.yuntuo.hk/k2safe/orbit/platform/resources"
|
|
pulse "git.yuntuo.hk/k2safe/orbit/pulse/model"
|
|
"git.yuntuo.hk/k2safe/orbit/pulse/service"
|
|
"git.yuntuo.hk/k2safe/orbit/pulse/worker"
|
|
)
|
|
|
|
func grantApplicationResources(f *executionFixture) {
|
|
f.exec(`UPDATE orbit_credentials SET permissions=JSON_SET(permissions,'$.actions',JSON_ARRAY_APPEND(JSON_EXTRACT(permissions,'$.actions'),'$','application:resources:read','$','application:resources:manage')) WHERE namespace_id=? AND id=?`, f.resources.NamespaceID, f.resources.CredentialID)
|
|
}
|
|
func applicationAPI(f *executionFixture) *resources.Service { return resources.New(NewTaskStore(f.db)) }
|
|
func configureApplication(t *testing.T, f *executionFixture, rate, concurrency, unfinished uint32) resources.View {
|
|
t.Helper()
|
|
api := applicationAPI(f)
|
|
v, err := api.Get(f.ctx, f.token)
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
out, err := api.Configure(f.ctx, f.token, string(executionID(t)), resources.ConfigureInput{ExpectedRevision: v.Revision, Reason: "application quota regression", RatePerSecond: rate, MaxConcurrency: concurrency, MaxUnfinishedTasks: unfinished})
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
return out.Resources
|
|
}
|
|
func assertApplicationCounts(t *testing.T, f *executionFixture, unfinished, active, pending uint64) {
|
|
t.Helper()
|
|
// 数据库核对不依赖当前应用是否停用，恢复仍应释放额度。
|
|
var got [3]uint64
|
|
if err := f.db.QueryRowContext(f.ctx, `SELECT unfinished_tasks,active_attempts,pending_starts FROM orbit_application_resources WHERE application_id=?`, f.resources.ApplicationID).Scan(&got[0], &got[1], &got[2]); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if got != [3]uint64{unfinished, active, pending} {
|
|
t.Fatalf("application counters=%v want=%v", got, [3]uint64{unfinished, active, pending})
|
|
}
|
|
var actual [3]uint64
|
|
if err := f.db.QueryRowContext(f.ctx, `SELECT COUNT(*),COALESCE(SUM(t.state IN ('LEASED','RUNNING')),0),COALESCE(SUM(t.state='LEASED'),0) FROM pulse_tasks t JOIN orbit_namespaces n ON n.id=t.namespace_id WHERE n.application_id=? AND t.state IN ('WAITING','LEASED','RUNNING','NEEDS_REVIEW')`, f.resources.ApplicationID).Scan(&actual[0], &actual[1], &actual[2]); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if actual != got {
|
|
t.Fatalf("application counters=%v actual states=%v", got, actual)
|
|
}
|
|
}
|
|
func TestIntegrationApplicationResources(t *testing.T) {
|
|
dsn := os.Getenv("ORBIT_TEST_MYSQL_DSN")
|
|
if dsn == "" {
|
|
t.Skip("requires disposable MySQL")
|
|
}
|
|
ctx, cancel := context.WithTimeout(context.Background(), 90*time.Second)
|
|
defer cancel()
|
|
db, err := Open(ctx, dsn, 32, 8)
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
defer db.Close()
|
|
if err := Migrate(ctx, db); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
fixture := func(t *testing.T, app ...core.ID) *executionFixture {
|
|
f := newExecutionFixture(t, ctx, db, app...)
|
|
grantApplicationResources(f)
|
|
return f
|
|
}
|
|
t.Run("explicit_authority_CAS_idempotency_and_atomic_audit", func(t *testing.T) {
|
|
f := newExecutionFixture(t, ctx, db)
|
|
api := applicationAPI(f)
|
|
_, err := api.Get(ctx, f.token)
|
|
expectWorkerCode(t, err, "PERMISSION_DENIED")
|
|
in := resources.ConfigureInput{ExpectedRevision: 1, Reason: "configure", MaxConcurrency: 2, RatePerSecond: 10, MaxUnfinishedTasks: 10}
|
|
_, err = api.Configure(ctx, f.token, "manage", in)
|
|
expectWorkerCode(t, err, "PERMISSION_DENIED")
|
|
grantApplicationResources(f)
|
|
v, err := api.Configure(ctx, f.token, "manage", in)
|
|
if err != nil || v.Resources.Revision != 2 {
|
|
t.Fatal(v, err)
|
|
}
|
|
f.submit(1)
|
|
replay, err := api.Configure(ctx, f.token, "manage", in)
|
|
if err != nil || !replay.Deduplicated || replay.Resources.UnfinishedTasks != 0 {
|
|
t.Fatal("receipt changed with usage", err)
|
|
}
|
|
_, err = api.Configure(ctx, f.token, "stale", in)
|
|
expectWorkerCode(t, err, "STATE_CONFLICT")
|
|
in.RatePerSecond = 20
|
|
_, err = api.Configure(ctx, f.token, "manage", in)
|
|
expectWorkerCode(t, err, "IDEMPOTENCY_CONFLICT")
|
|
in.ExpectedRevision = 2
|
|
trigger := "test_app_audit_" + string(executionID(t))
|
|
f.exec("CREATE TRIGGER " + trigger + " BEFORE INSERT ON orbit_audit_events FOR EACH ROW SIGNAL SQLSTATE '45000' SET MESSAGE_TEXT='injected audit failure'")
|
|
_, err = api.Configure(ctx, f.token, "rollback", in)
|
|
f.exec("DROP TRIGGER " + trigger)
|
|
if err == nil {
|
|
t.Fatal("audit fault ignored")
|
|
}
|
|
v2, err := api.Get(ctx, f.token)
|
|
if err != nil || v2.Revision != 2 || v2.RatePerSecond != 10 {
|
|
t.Fatal("configuration did not roll back", v2, err)
|
|
}
|
|
if _, err := api.Configure(ctx, f.token, "rollback", in); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if f.count("orbit_audit_events", "action='application.resources.configure'") != 2 || f.count("orbit_events", "type='application.resources.configured'") != 2 {
|
|
t.Fatal("duplicate or missing audit/event")
|
|
}
|
|
g := fixture(t, f.resources.ApplicationID)
|
|
shared, err := applicationAPI(g).Get(ctx, g.token)
|
|
if err != nil || shared.Revision != 3 || shared.UnfinishedTasks != 1 {
|
|
t.Fatal(shared, err)
|
|
}
|
|
other := fixture(t)
|
|
if v, err := applicationAPI(other).Get(ctx, other.token); err != nil || v.Revision != 1 || v.UnfinishedTasks != 0 {
|
|
t.Fatal("other application leaked", v, err)
|
|
}
|
|
})
|
|
t.Run("concurrent_admission_across_namespaces_and_full_replay", func(t *testing.T) {
|
|
f := fixture(t)
|
|
g := fixture(t, f.resources.ApplicationID)
|
|
configureApplication(t, f, 100, 100, 3)
|
|
type answer struct {
|
|
f *executionFixture
|
|
key string
|
|
in service.SubmitInput
|
|
out service.SubmitResult
|
|
err error
|
|
}
|
|
ch := make(chan answer, 16)
|
|
var wg sync.WaitGroup
|
|
for i := 0; i < 16; i++ {
|
|
target := f
|
|
if i%2 == 1 {
|
|
target = g
|
|
}
|
|
key := string(executionID(t))
|
|
wg.Add(1)
|
|
go func() {
|
|
defer wg.Done()
|
|
in := service.SubmitInput{QueueID: target.resources.QueueID, Handler: "object.prepare", HandlerVersion: "v1", PayloadVersion: "v1", Payload: json.RawMessage(`{}`)}
|
|
out, err := target.tasks.Submit(ctx, target.token, key, in)
|
|
ch <- answer{target, key, in, out, err}
|
|
}()
|
|
}
|
|
wg.Wait()
|
|
close(ch)
|
|
var accepted []answer
|
|
for a := range ch {
|
|
if a.err == nil {
|
|
accepted = append(accepted, a)
|
|
} else {
|
|
expectWorkerCode(t, a.err, "APPLICATION_FULL")
|
|
}
|
|
}
|
|
if len(accepted) != 3 {
|
|
t.Fatalf("accepted=%d", len(accepted))
|
|
}
|
|
assertApplicationCounts(t, f, 3, 0, 0)
|
|
a := accepted[0]
|
|
replay, err := a.f.tasks.Submit(ctx, a.f.token, a.key, a.in)
|
|
if err != nil || !replay.Deduplicated || replay.Task.ID != a.out.Task.ID {
|
|
t.Fatal("full app lost receipt", err)
|
|
}
|
|
configureApplication(t, g, 100, 100, 1)
|
|
assertApplicationCounts(t, f, 3, 0, 0)
|
|
for _, a := range accepted {
|
|
_, err := a.f.tasks.CancelTask(ctx, a.f.token, a.out.Task.ID, "cancel-"+a.key, service.CancelInput{ExpectedRevision: 1, Reason: "release"})
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
}
|
|
assertApplicationCounts(t, g, 0, 0, 0)
|
|
g.submit(1)
|
|
assertApplicationCounts(t, f, 1, 0, 0)
|
|
})
|
|
t.Run("queue_rejection_and_failed_admission_do_not_leak_application_budget", func(t *testing.T) {
|
|
f := fixture(t)
|
|
configureApplication(t, f, 100, 10, 10)
|
|
configureResources(t, f, 100, 10, 1, false)
|
|
first := f.submit(1)
|
|
in := service.SubmitInput{QueueID: f.resources.QueueID, Handler: "object.prepare", HandlerVersion: "v1", PayloadVersion: "v1", Payload: json.RawMessage(`{}`)}
|
|
_, err := f.tasks.Submit(ctx, f.token, "second", in)
|
|
expectWorkerCode(t, err, "QUEUE_FULL")
|
|
assertApplicationCounts(t, f, 1, 0, 0)
|
|
if _, err := f.tasks.CancelTask(ctx, f.token, first.ID, "cancel-first", service.CancelInput{ExpectedRevision: first.Revision, Reason: "release"}); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
trigger := "test_app_admission_" + string(executionID(t))
|
|
f.exec("CREATE TRIGGER " + trigger + " BEFORE INSERT ON orbit_events FOR EACH ROW SIGNAL SQLSTATE '45000' SET MESSAGE_TEXT='injected admission failure'")
|
|
_, err = f.tasks.Submit(ctx, f.token, "second", in)
|
|
f.exec("DROP TRIGGER " + trigger)
|
|
if err == nil {
|
|
t.Fatal("admission failure ignored")
|
|
}
|
|
assertApplicationCounts(t, f, 0, 0, 0)
|
|
assertQueueCount(t, f, 0)
|
|
if _, err := f.tasks.Submit(ctx, f.token, "second", in); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
assertApplicationCounts(t, f, 1, 0, 0)
|
|
assertQueueCount(t, f, 1)
|
|
})
|
|
t.Run("shared_concurrency_and_exact_release", func(t *testing.T) {
|
|
f := fixture(t)
|
|
g := fixture(t, f.resources.ApplicationID)
|
|
configureApplication(t, f, 100, 1, 10)
|
|
f.submit(1)
|
|
g.submit(1)
|
|
type claim struct {
|
|
f *executionFixture
|
|
a *worker.Assignment
|
|
err error
|
|
}
|
|
ch := make(chan claim, 2)
|
|
for _, target := range []*executionFixture{f, g} {
|
|
go func() { a, err := target.a.Claim(ctx, target.token, target.binding); ch <- claim{target, a, err} }()
|
|
}
|
|
var winner claim
|
|
for i := 0; i < 2; i++ {
|
|
a := <-ch
|
|
if a.err != nil {
|
|
t.Fatal(a.err)
|
|
}
|
|
if a.a != nil {
|
|
if winner.a != nil {
|
|
t.Fatal("cross namespace concurrency exceeded")
|
|
}
|
|
winner = a
|
|
}
|
|
}
|
|
if winner.a == nil {
|
|
t.Fatal("no claim")
|
|
}
|
|
assertApplicationCounts(t, f, 2, 1, 1)
|
|
task := winner.a.Task
|
|
_, err := winner.f.tasks.CancelTask(ctx, winner.f.token, task.ID, "cancel-leased", service.CancelInput{ExpectedRevision: task.Revision, Reason: "release before start"})
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
assertApplicationCounts(t, f, 1, 0, 0)
|
|
other := f
|
|
if winner.f == f {
|
|
other = g
|
|
}
|
|
ref := retryStart(t, other)
|
|
assertApplicationCounts(t, f, 1, 1, 0)
|
|
message := executionID(t)
|
|
input := worker.FinishInput{Ref: ref, Outcome: "SUCCEEDED", Result: json.RawMessage(`{}`)}
|
|
if _, err := other.a.Finish(ctx, other.token, other.binding, message, input); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if out, err := other.a.Finish(ctx, other.token, other.binding, message, input); err != nil || !out.Deduplicated {
|
|
t.Fatal("finish replay", err)
|
|
}
|
|
assertApplicationCounts(t, f, 0, 0, 0)
|
|
})
|
|
t.Run("shared_start_window_downgrade_and_rollback", func(t *testing.T) {
|
|
f := fixture(t)
|
|
g := fixture(t, f.resources.ApplicationID)
|
|
configureApplication(t, f, 2, 10, 10)
|
|
f.submit(1)
|
|
g.submit(1)
|
|
_, a := f.claim()
|
|
_, b := g.claim()
|
|
assertApplicationCounts(t, f, 2, 2, 2)
|
|
configureApplication(t, g, 1, 1, 10)
|
|
trigger := "test_app_start_" + string(executionID(t))
|
|
f.exec("CREATE TRIGGER " + trigger + " BEFORE INSERT ON orbit_events FOR EACH ROW SIGNAL SQLSTATE '45000' SET MESSAGE_TEXT='injected start failure'")
|
|
message := executionID(t)
|
|
_, err := f.a.Start(ctx, f.token, f.binding, message, a)
|
|
f.exec("DROP TRIGGER " + trigger)
|
|
if err == nil {
|
|
t.Fatal("start failure ignored")
|
|
}
|
|
assertApplicationCounts(t, f, 2, 2, 2)
|
|
if _, err := f.a.Start(ctx, f.token, f.binding, message, a); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
// 固定尚未过期窗口，避免慢测试主机把竞争断言跨到下一秒。
|
|
f.exec(`UPDATE orbit_application_resources SET start_window_until=DATE_ADD(UTC_TIMESTAMP(6),INTERVAL 1 HOUR) WHERE application_id=?`, f.resources.ApplicationID)
|
|
if out, err := f.a.Start(ctx, f.token, f.binding, message, a); err != nil || !out.Deduplicated {
|
|
t.Fatal("start replay charged again", err)
|
|
}
|
|
_, err = g.a.Start(ctx, g.token, g.binding, executionID(t), b)
|
|
expectWorkerCode(t, err, "RATE_LIMITED")
|
|
assertApplicationCounts(t, f, 2, 2, 1)
|
|
retryFinish(t, f, a, "SUCCEEDED", false)
|
|
configureApplication(t, f, 1, 10, 10)
|
|
g.submit(1)
|
|
if a, err := g.a.Claim(ctx, g.token, g.binding); err != nil || a != nil {
|
|
t.Fatal("pending start reservation bypassed", err)
|
|
}
|
|
g.exec(`UPDATE orbit_application_resources SET start_window_until=DATE_SUB(UTC_TIMESTAMP(6),INTERVAL 1 SECOND) WHERE application_id=?`, g.resources.ApplicationID)
|
|
if a, err := g.a.Claim(ctx, g.token, g.binding); err != nil || a != nil {
|
|
t.Fatal("unstarted reservation lost across window", err)
|
|
}
|
|
if _, err := g.a.Start(ctx, g.token, g.binding, executionID(t), b); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
assertApplicationCounts(t, f, 2, 1, 0)
|
|
})
|
|
t.Run("retry_recovery_unknown_and_disabled_application", func(t *testing.T) {
|
|
f := fixture(t)
|
|
configureApplication(t, f, 100, 10, 10)
|
|
task := retryTask(t, f, 2, pulse.RecoveryReconcileOnly)
|
|
ref := retryStart(t, f)
|
|
retryFinish(t, f, ref, "FAILED", true)
|
|
assertApplicationCounts(t, f, 1, 0, 0)
|
|
makeRetryDue(f, task.ID)
|
|
ref = retryStart(t, f)
|
|
f.expire(ref)
|
|
f.exec(`UPDATE orbit_applications SET enabled=FALSE WHERE id=?`, f.resources.ApplicationID)
|
|
if _, err := f.storeB.Recover(ctx, 100); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
assertApplicationCounts(t, f, 1, 0, 0)
|
|
if _, err := f.storeA.Recover(ctx, 100); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
assertApplicationCounts(t, f, 1, 0, 0)
|
|
f.exec(`UPDATE orbit_applications SET enabled=TRUE WHERE id=?`, f.resources.ApplicationID)
|
|
if f.get(task.ID).State != pulse.TaskNeedsReview {
|
|
t.Fatal("unknown effect lost")
|
|
}
|
|
f.submit(1)
|
|
_, leased := f.claim()
|
|
f.expire(leased)
|
|
if _, err := f.storeB.Recover(ctx, 100); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
assertApplicationCounts(t, f, 1, 0, 0)
|
|
waiting := f.submit(1)
|
|
f.exec(`UPDATE pulse_tasks SET available_at=DATE_SUB(UTC_TIMESTAMP(6),INTERVAL 2 SECOND),deadline_at=DATE_SUB(UTC_TIMESTAMP(6),INTERVAL 1 SECOND) WHERE namespace_id=? AND id=?`, f.resources.NamespaceID, waiting.ID)
|
|
if _, err := f.storeB.Recover(ctx, 100); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
assertApplicationCounts(t, f, 1, 0, 0)
|
|
})
|
|
t.Run("schedule_partial_batch_and_global_backpressure", func(t *testing.T) {
|
|
f := fixture(t)
|
|
g := fixture(t, f.resources.ApplicationID)
|
|
configureApplication(t, f, 100, 10, 2)
|
|
g.submit(1)
|
|
sf := newScheduleFixture(t, f)
|
|
v := sf.create(sf.definition())
|
|
sf.due(v, 2*time.Second)
|
|
sf.tick()
|
|
items := sf.occurrences(v.ID)
|
|
if len(items) != 1 || items[0].TaskID == nil {
|
|
t.Fatal("application schedule limit exceeded", len(items))
|
|
}
|
|
next := sf.current(v.ID).NextFireAt
|
|
if next == nil || !next.Equal(items[0].ScheduledAt.Add(time.Second)) {
|
|
t.Fatal("backpressure advanced unaccepted cursor")
|
|
}
|
|
assertApplicationCounts(t, f, 2, 0, 0)
|
|
assertQueueCount(t, f, 1)
|
|
assertQueueCount(t, g, 1)
|
|
_, err := f.tasks.CancelTask(ctx, f.token, *items[0].TaskID, "free-schedule", service.CancelInput{ExpectedRevision: 1, Reason: "free one"})
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
sf.due(v, 0)
|
|
sf.tick()
|
|
items = sf.occurrences(v.ID)
|
|
if len(items) != 2 {
|
|
t.Fatal("original point not resumed")
|
|
}
|
|
assertApplicationCounts(t, f, 2, 0, 0)
|
|
assertQueueCount(t, f, 1)
|
|
})
|
|
t.Run("busy_application_is_skipped_during_recovery", func(t *testing.T) {
|
|
f := fixture(t)
|
|
g := fixture(t)
|
|
for _, v := range []*executionFixture{f, g} {
|
|
v.submit(1)
|
|
_, ref := v.claim()
|
|
v.expire(ref)
|
|
}
|
|
tx, err := db.BeginTx(ctx, &sql.TxOptions{Isolation: sql.LevelReadCommitted})
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
defer tx.Rollback()
|
|
if _, err := lockApplicationResources(ctx, tx, f.resources.NamespaceID, false); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
short, stop := context.WithTimeout(ctx, time.Second)
|
|
defer stop()
|
|
if _, err := g.storeB.Recover(short, 100); err != nil {
|
|
t.Fatal("hot application blocked unrelated recovery", err)
|
|
}
|
|
assertApplicationCounts(t, f, 1, 1, 1)
|
|
assertApplicationCounts(t, g, 0, 0, 0)
|
|
if err := tx.Rollback(); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if _, err := f.storeB.Recover(ctx, 100); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
assertApplicationCounts(t, f, 0, 0, 0)
|
|
})
|
|
t.Run("bootstrap_existing_application_rejects_invalid_parent", func(t *testing.T) {
|
|
f := fixture(t)
|
|
f.exec(`UPDATE orbit_applications SET enabled=FALSE WHERE id=?`, f.resources.ApplicationID)
|
|
for _, in := range []BootstrapInput{
|
|
{ExistingApplicationID: executionID(t), Namespace: "missing", Environment: "test", Queue: "q"},
|
|
{ExistingApplicationID: f.resources.ApplicationID, Namespace: "disabled", Environment: "test", Queue: "q"},
|
|
{Application: "new", ExistingApplicationID: f.resources.ApplicationID, Namespace: "ambiguous", Environment: "test", Queue: "q"},
|
|
} {
|
|
delivered := false
|
|
err := Bootstrap(ctx, db, in, func(BootstrapResult) error { delivered = true; return nil })
|
|
if err == nil || delivered {
|
|
t.Fatal("invalid application accepted")
|
|
}
|
|
}
|
|
})
|
|
}
|