231 lines
9.1 KiB
Go
231 lines
9.1 KiB
Go
package mysql
|
|
|
|
import (
|
|
"context"
|
|
"encoding/json"
|
|
"fmt"
|
|
"os"
|
|
"sync"
|
|
"testing"
|
|
"time"
|
|
|
|
core "git.yuntuo.hk/k2safe/orbit/model"
|
|
"git.yuntuo.hk/k2safe/orbit/platform/auth"
|
|
"git.yuntuo.hk/k2safe/orbit/pulse/service"
|
|
)
|
|
|
|
func taskBatch(f *executionFixture, n int) service.BatchSubmitInput {
|
|
in := service.BatchSubmitInput{}
|
|
for i := 0; i < n; i++ {
|
|
in.Items = append(in.Items, service.BatchSubmitItem{IdempotencyKey: fmt.Sprintf("batch-%d", i), Task: service.SubmitInput{QueueID: f.resources.QueueID, Handler: "object.prepare", HandlerVersion: "v1", PayloadVersion: "v1", Payload: json.RawMessage(`{"n":9007199254740993}`)}})
|
|
}
|
|
return in
|
|
}
|
|
func requireBatchStatuses(t *testing.T, out service.BatchSubmitResult, err error, wants ...service.BatchItemStatus) {
|
|
t.Helper()
|
|
if err != nil || len(out.Items) != len(wants) {
|
|
t.Fatal(out, err)
|
|
}
|
|
for i, want := range wants {
|
|
if out.Items[i].Index != i || out.Items[i].Status != want {
|
|
t.Fatalf("item %d=%+v want=%s", i, out.Items[i], want)
|
|
}
|
|
}
|
|
}
|
|
|
|
type batchAfterCommitStore struct {
|
|
service.Store
|
|
after func()
|
|
}
|
|
|
|
func (s *batchAfterCommitStore) Submit(ctx context.Context, token auth.Token, key string, in service.SubmitInput) (service.SubmitResult, error) {
|
|
out, err := s.Store.Submit(ctx, token, key, in)
|
|
if err == nil && s.after != nil {
|
|
fn := s.after
|
|
s.after = nil
|
|
fn()
|
|
}
|
|
return out, err
|
|
}
|
|
|
|
func TestIntegrationTaskBatch(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) *executionFixture { return newExecutionFixture(t, ctx, db) }
|
|
t.Run("partial_rejections_replay_and_quota", func(t *testing.T) {
|
|
f := fixture(t)
|
|
configureResources(t, f, 10, 4, 2, true)
|
|
in := taskBatch(f, 5)
|
|
in.Items[1].Task.Handler = ""
|
|
in.Items[2].Task.QueueID = executionID(t)
|
|
out, err := f.tasks.SubmitBatch(ctx, f.token, in)
|
|
requireBatchStatuses(t, out, err, service.BatchAccepted, service.BatchRejected, service.BatchRejected, service.BatchAccepted, service.BatchRejected)
|
|
if out.Items[1].Error.Code != "INVALID_ARGUMENT" || out.Items[2].Error.Code != "NOT_FOUND" || out.Items[4].Error.Code != "QUEUE_FULL" || out.Items[4].Error.RetryAfterMS != 1000 {
|
|
t.Fatal(out)
|
|
}
|
|
assertQueueCount(t, f, 2)
|
|
assertApplicationCounts(t, f, 2, 0, 0)
|
|
for table, filter := range map[string]string{"orbit_idempotency_keys": "operation='task.submit'", "orbit_audit_events": "action='task.submit'", "orbit_events": "type='task.accepted'"} {
|
|
if f.count(table, filter) != 2 {
|
|
t.Fatal("partial batch leaked writes", table)
|
|
}
|
|
}
|
|
if string(f.get(out.Items[0].Receipt.TaskID).Payload) != `{"n":9007199254740993}` {
|
|
t.Fatal("payload precision lost")
|
|
}
|
|
one, err := f.tasks.Submit(ctx, f.token, in.Items[0].IdempotencyKey, in.Items[0].Task)
|
|
if err != nil || !one.Deduplicated || one.Task.ID != out.Items[0].Receipt.TaskID {
|
|
t.Fatal("single batch identity diverged", err)
|
|
}
|
|
first := out.Items[0].Receipt
|
|
if _, err := f.tasks.CancelTask(ctx, f.token, first.TaskID, "cancel", service.CancelInput{ExpectedRevision: first.Revision, Reason: "free batch slot"}); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
in.Items = []service.BatchSubmitItem{in.Items[0], in.Items[4]}
|
|
again, err := f.tasks.SubmitBatch(ctx, f.token, in)
|
|
requireBatchStatuses(t, again, err, service.BatchAccepted, service.BatchAccepted)
|
|
if !again.Items[0].Receipt.Deduplicated || again.Items[0].Receipt.Revision != 1 || again.Items[1].Receipt.Deduplicated {
|
|
t.Fatal("receipt/rejected identity not reusable")
|
|
}
|
|
in.Items[0].Task.Payload = json.RawMessage(`{"changed":true}`)
|
|
conflict, err := f.tasks.SubmitBatch(ctx, f.token, in)
|
|
requireBatchStatuses(t, conflict, err, service.BatchRejected, service.BatchAccepted)
|
|
if conflict.Items[0].Error.Code != "IDEMPOTENCY_CONFLICT" {
|
|
t.Fatal(conflict)
|
|
}
|
|
assertQueueCount(t, f, 2)
|
|
assertApplicationCounts(t, f, 2, 0, 0)
|
|
})
|
|
t.Run("cross_namespace_and_queue_scope", func(t *testing.T) {
|
|
f, g := fixture(t), fixture(t)
|
|
in := taskBatch(f, 2)
|
|
in.Items[0].Task.QueueID = g.resources.QueueID
|
|
out, err := f.tasks.SubmitBatch(ctx, f.token, in)
|
|
requireBatchStatuses(t, out, err, service.BatchRejected, service.BatchAccepted)
|
|
if out.Items[0].Error.Code != "NOT_FOUND" {
|
|
t.Fatal(out)
|
|
}
|
|
f.exec(`UPDATE orbit_credentials SET permissions=? WHERE namespace_id=? AND id=?`, `{"actions":["tasks:submit"],"queues":["`+string(g.resources.QueueID)+`"]}`, f.resources.NamespaceID, f.resources.CredentialID)
|
|
out, err = f.tasks.SubmitBatch(ctx, f.token, service.BatchSubmitInput{Items: in.Items[1:]})
|
|
requireBatchStatuses(t, out, err, service.BatchRejected)
|
|
if out.Items[0].Error.Code != "PERMISSION_DENIED" || out.Items[0].Receipt != nil {
|
|
t.Fatal("cached receipt bypassed current queue scope")
|
|
}
|
|
if g.count("pulse_tasks", "1=1") != 0 {
|
|
t.Fatal("cross namespace write")
|
|
}
|
|
})
|
|
t.Run("revoke_between_items_is_rechecked", func(t *testing.T) {
|
|
f := fixture(t)
|
|
s := &batchAfterCommitStore{Store: NewTaskStore(db), after: func() {
|
|
f.exec(`UPDATE orbit_credentials SET revoked_at=UTC_TIMESTAMP(6) WHERE namespace_id=? AND id=?`, f.resources.NamespaceID, f.resources.CredentialID)
|
|
}}
|
|
out, err := service.New(s).SubmitBatch(ctx, f.token, taskBatch(f, 3))
|
|
requireBatchStatuses(t, out, err, service.BatchAccepted, service.BatchRejected, service.BatchNotAttempted)
|
|
if out.Items[1].Error.Code != "UNAUTHENTICATED" || f.count("pulse_tasks", "1=1") != 1 {
|
|
t.Fatal(out)
|
|
}
|
|
})
|
|
t.Run("cancel_after_first_commit_leaves_recoverable_prefix", func(t *testing.T) {
|
|
f := fixture(t)
|
|
run, stop := context.WithCancel(ctx)
|
|
defer stop()
|
|
s := &batchAfterCommitStore{Store: NewTaskStore(db), after: stop}
|
|
in := taskBatch(f, 3)
|
|
out, err := service.New(s).SubmitBatch(run, f.token, in)
|
|
requireBatchStatuses(t, out, err, service.BatchAccepted, service.BatchNotAttempted, service.BatchNotAttempted)
|
|
if f.count("pulse_tasks", "1=1") != 1 {
|
|
t.Fatal("cancelled suffix wrote tasks")
|
|
}
|
|
again, err := f.tasks.SubmitBatch(ctx, f.token, in)
|
|
requireBatchStatuses(t, again, err, service.BatchAccepted, service.BatchAccepted, service.BatchAccepted)
|
|
if !again.Items[0].Receipt.Deduplicated || again.Items[0].Receipt.TaskID != out.Items[0].Receipt.TaskID {
|
|
t.Fatal("lost committed prefix")
|
|
}
|
|
assertApplicationCounts(t, f, 3, 0, 0)
|
|
})
|
|
t.Run("audit_failure_rolls_back_item_and_preserves_previous_commit", func(t *testing.T) {
|
|
f := fixture(t)
|
|
in := taskBatch(f, 3)
|
|
if _, err := f.tasks.Submit(ctx, f.token, in.Items[0].IdempotencyKey, in.Items[0].Task); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
trigger := "batch_audit_" + string(executionID(t))
|
|
f.exec("CREATE TRIGGER " + trigger + " BEFORE INSERT ON orbit_audit_events FOR EACH ROW SIGNAL SQLSTATE '45000' SET MESSAGE_TEXT='private injected audit fault'")
|
|
out, err := f.tasks.SubmitBatch(ctx, f.token, in)
|
|
f.exec("DROP TRIGGER " + trigger)
|
|
requireBatchStatuses(t, out, err, service.BatchAccepted, service.BatchUnknown, service.BatchNotAttempted)
|
|
if out.Items[1].Error.Code != "UNAVAILABLE" {
|
|
t.Fatal(out)
|
|
}
|
|
assertQueueCount(t, f, 1)
|
|
assertApplicationCounts(t, f, 1, 0, 0)
|
|
for table, filter := range map[string]string{"orbit_idempotency_keys": "operation='task.submit'", "orbit_audit_events": "action='task.submit'", "orbit_events": "type='task.accepted'"} {
|
|
if f.count(table, filter) != 1 {
|
|
t.Fatal("atomic item rollback failed", table)
|
|
}
|
|
}
|
|
again, err := f.tasks.SubmitBatch(ctx, f.token, in)
|
|
requireBatchStatuses(t, again, err, service.BatchAccepted, service.BatchAccepted, service.BatchAccepted)
|
|
assertQueueCount(t, f, 3)
|
|
})
|
|
t.Run("concurrent_replay_and_shared_application_backpressure", func(t *testing.T) {
|
|
f := fixture(t)
|
|
grantApplicationResources(f)
|
|
configureApplication(t, f, 100, 10, 3)
|
|
in := taskBatch(f, 4)
|
|
type result struct {
|
|
out service.BatchSubmitResult
|
|
err error
|
|
}
|
|
ch := make(chan result, 8)
|
|
var wg sync.WaitGroup
|
|
for i := 0; i < 8; i++ {
|
|
wg.Add(1)
|
|
go func() { defer wg.Done(); out, err := f.tasks.SubmitBatch(ctx, f.token, in); ch <- result{out, err} }()
|
|
}
|
|
wg.Wait()
|
|
close(ch)
|
|
ids := map[int]core.ID{}
|
|
fresh := 0
|
|
for r := range ch {
|
|
requireBatchStatuses(t, r.out, r.err, service.BatchAccepted, service.BatchAccepted, service.BatchAccepted, service.BatchRejected)
|
|
for i := 0; i < 3; i++ {
|
|
receipt := r.out.Items[i].Receipt
|
|
if previous := ids[i]; previous != "" && previous != receipt.TaskID {
|
|
t.Fatal("duplicate identity")
|
|
}
|
|
ids[i] = receipt.TaskID
|
|
if !receipt.Deduplicated {
|
|
fresh++
|
|
}
|
|
}
|
|
if r.out.Items[3].Error.Code != "APPLICATION_FULL" {
|
|
t.Fatal(r.out)
|
|
}
|
|
}
|
|
if fresh != 3 || f.count("pulse_tasks", "1=1") != 3 {
|
|
t.Fatal("duplicate creation", fresh)
|
|
}
|
|
assertApplicationCounts(t, f, 3, 0, 0)
|
|
g := newExecutionFixture(t, ctx, db, f.resources.ApplicationID)
|
|
out, err := g.tasks.SubmitBatch(ctx, g.token, taskBatch(g, 1))
|
|
requireBatchStatuses(t, out, err, service.BatchRejected)
|
|
if out.Items[0].Error.Code != "APPLICATION_FULL" {
|
|
t.Fatal(out)
|
|
}
|
|
})
|
|
}
|