105 lines
3.6 KiB
Go
105 lines
3.6 KiB
Go
package mysql
|
|
|
|
import (
|
|
"context"
|
|
"encoding/json"
|
|
"net/http/httptest"
|
|
"os"
|
|
"testing"
|
|
"time"
|
|
|
|
"git.yuntuo.hk/k2safe/orbit/internal/transport/httpserver"
|
|
"git.yuntuo.hk/k2safe/orbit/pulse/worker"
|
|
)
|
|
|
|
func TestIntegrationFairDispatchApplicationIsolation(t *testing.T) {
|
|
dsn := os.Getenv("ORBIT_TEST_MYSQL_DSN")
|
|
if dsn == "" {
|
|
t.Skip("requires disposable MySQL")
|
|
}
|
|
ctx, cancel := context.WithTimeout(context.Background(), 25*time.Second)
|
|
defer cancel()
|
|
db, err := Open(ctx, dsn, 4, 2)
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
defer db.Close()
|
|
if err = Migrate(ctx, db); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
noisy := newExecutionFixture(t, ctx, db)
|
|
quiet := newExecutionFixture(t, ctx, db)
|
|
if noisy.binding.ApplicationID != noisy.resources.ApplicationID || quiet.binding.ApplicationID != quiet.resources.ApplicationID {
|
|
t.Fatal("dispatch group was not derived from authenticated namespace")
|
|
}
|
|
hub := httpserver.NewWorkerHub(ctx, noisy.a)
|
|
server := httpserver.New("", func(context.Context) error { return nil })
|
|
httpserver.RegisterWorkers(server, hub)
|
|
tls := httptest.NewTLSServer(server.Handler)
|
|
defer tls.Close()
|
|
defer hub.Shutdown(ctx)
|
|
var connections []*testWorkerSocket
|
|
for range 12 {
|
|
f := *noisy
|
|
f.hello.InstanceID = executionID(t)
|
|
c := connectTestWorker(t, ctx, tls, &f)
|
|
id := c.send("worker.capacity", worker.CapacityInput{Sequence: 1, Capacity: 0}, "")
|
|
if reply := c.read(); reply.ReplyTo != id || reply.Error != nil {
|
|
t.Fatal("could not disable initial claims")
|
|
}
|
|
connections = append(connections, c)
|
|
}
|
|
quiet.hello.InstanceID = executionID(t)
|
|
q := connectTestWorker(t, ctx, tls, quiet)
|
|
id := q.send("worker.capacity", worker.CapacityInput{Sequence: 1, Capacity: 0}, "")
|
|
if reply := q.read(); reply.ReplyTo != id || reply.Error != nil {
|
|
t.Fatal("quiet worker setup failed")
|
|
}
|
|
noisy.submit(1)
|
|
want := quiet.submit(1)
|
|
lock, err := db.BeginTx(ctx, nil)
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
defer lock.Rollback()
|
|
var ignored string
|
|
if err = lock.QueryRowContext(ctx, `SELECT HEX(application_id) FROM orbit_application_resources WHERE application_id=? FOR UPDATE`, noisy.resources.ApplicationID).Scan(&ignored); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
for _, c := range connections {
|
|
id := c.send("worker.capacity", worker.CapacityInput{Sequence: 2, Capacity: 1}, "")
|
|
if reply := c.read(); reply.ReplyTo != id || reply.Error != nil {
|
|
t.Fatal("noisy worker control blocked by claims")
|
|
}
|
|
}
|
|
// 等到真的有领取事务在额度行上等待，再验证另一应用；不靠固定 sleep 猜测故障已发生。
|
|
for {
|
|
var waiting int
|
|
if err = db.QueryRowContext(ctx, `SELECT COUNT(*) FROM performance_schema.data_lock_waits`).Scan(&waiting); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if waiting > 0 {
|
|
break
|
|
}
|
|
select {
|
|
case <-ctx.Done():
|
|
t.Fatal("held application did not receive a claim")
|
|
case <-time.After(20 * time.Millisecond):
|
|
}
|
|
}
|
|
id = q.send("worker.capacity", worker.CapacityInput{Sequence: 2, Capacity: 1}, "")
|
|
if reply := q.read(); reply.ReplyTo != id || reply.Error != nil {
|
|
t.Fatal("other application heartbeat starved")
|
|
}
|
|
reply := q.read()
|
|
var assignment worker.Assignment
|
|
if reply.Type != "task.assign" || json.Unmarshal(reply.Body, &assignment) != nil || assignment.Task.ID != want.ID {
|
|
t.Fatal("quiet application could not claim while noisy application was blocked")
|
|
}
|
|
// 此时锁仍未释放；静默应用的真实领取已提交，证明没有耗尽四连接数据库池。
|
|
if err := lock.Rollback(); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
t.Log("12 noisy connections blocked on one application; another application claimed with a four-connection database pool before lock release")
|
|
}
|