MiFaZhan
127 lines
3.6 KiB
Go
127 lines
3.6 KiB
Go
package data
|
|
|
|
import (
|
|
"context"
|
|
"errors"
|
|
"fmt"
|
|
"sync/atomic"
|
|
"testing"
|
|
"time"
|
|
|
|
"git.5iot.com/iotlcm-public/iotlcm-worker-svr/internal/model"
|
|
)
|
|
|
|
type fairnessRepository struct{ *durableFake }
|
|
|
|
func (f *fairnessRepository) Load(context.Context, string) (*Task, error) {
|
|
return nil, errors.New("公平性测试不应准备父批")
|
|
}
|
|
func (f *fairnessRepository) SaveValidationFailure(context.Context, *Task) error { return nil }
|
|
func (f *fairnessRepository) SaveResult(context.Context, *Task, *Detail, Outcome) (Outcome, error) {
|
|
return Outcome{}, nil
|
|
}
|
|
func (f *fairnessRepository) Finalize(context.Context, *Task) error { return nil }
|
|
|
|
type fairnessUpstream struct {
|
|
asyncFake
|
|
rejected atomic.Int32
|
|
}
|
|
|
|
func (f *fairnessUpstream) ChangeSimApnFunction(context.Context, *model.SimCardApnChangeInput) ([]model.SimBaseRes[model.SimCardApnChangeOutputItem], error) {
|
|
return nil, errors.New("调度不得走同步办理")
|
|
}
|
|
|
|
func (f *fairnessUpstream) CheckSimApnAsyncCapability(_ context.Context, in *model.SimCardApnChangeInput) error {
|
|
if in.SimKeys[0].ChanKey == "removed" {
|
|
f.rejected.Add(1)
|
|
return errors.New("通道已移除，无protocolVersion")
|
|
}
|
|
return nil
|
|
}
|
|
|
|
func TestDurableAPNPrecheckBackoffAllowsHealthy129thTask(t *testing.T) {
|
|
repo := newDurableFake()
|
|
ctx, cancel := context.WithCancel(context.Background())
|
|
defer cancel()
|
|
for i := 0; i < 129; i++ {
|
|
sub := fixtureSub()
|
|
sub.OperationID = fmt.Sprintf("fair-%03d", i)
|
|
sub.Request.ReqData.OperationID = sub.OperationID
|
|
sub.ChanKey = "removed"
|
|
if i == 128 {
|
|
sub.ChanKey = "healthy"
|
|
}
|
|
for _, key := range sub.Request.SimKeys {
|
|
key.ChanKey = sub.ChanKey
|
|
}
|
|
if err := repo.PrepareSubBatch(ctx, sub); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
}
|
|
completed := make(chan struct{}, 1)
|
|
upstream := &fairnessUpstream{}
|
|
upstream.submit = func(in *model.SimCardApnChangeInput) ([]model.SimBaseRes[model.SimCardApnChangeOutputItem], error) {
|
|
if in.SimKeys[0].ChanKey != "healthy" {
|
|
t.Error("拒绝通道发出办理")
|
|
}
|
|
out := pendingResponse(in, 0)
|
|
for i := range out {
|
|
out[i].ResData.Pending = false
|
|
}
|
|
completed <- struct{}{}
|
|
return out, nil
|
|
}
|
|
e := &executor{repository: &fairnessRepository{repo}, upstream: upstream}
|
|
done := make(chan error, 1)
|
|
go func() { done <- e.RunReconciler(ctx) }()
|
|
select {
|
|
case <-completed:
|
|
case <-time.After(15 * time.Second):
|
|
cancel()
|
|
<-done
|
|
t.Fatal("前128个预检拒绝使第129健康任务饥饿")
|
|
}
|
|
// 健康提交可能早于已入队的拒绝任务完成；等待退避持久化后再取消调度器。
|
|
ticker := time.NewTicker(10 * time.Millisecond)
|
|
defer ticker.Stop()
|
|
timeout := time.NewTimer(15 * time.Second)
|
|
defer timeout.Stop()
|
|
for {
|
|
repo.mu.Lock()
|
|
delayed := 0
|
|
for id, sub := range repo.batches {
|
|
if sub.ChanKey == "removed" && !repo.next[id].IsZero() {
|
|
delayed++
|
|
}
|
|
}
|
|
repo.mu.Unlock()
|
|
if delayed == 128 {
|
|
break
|
|
}
|
|
select {
|
|
case <-ticker.C:
|
|
case <-timeout.C:
|
|
cancel()
|
|
<-done
|
|
t.Fatalf("拒绝任务退避未全部持久化: delayed=%d rejected=%d", delayed, upstream.rejected.Load())
|
|
}
|
|
}
|
|
cancel()
|
|
if err := <-done; err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if upstream.submitCalls != 1 || upstream.rejected.Load() != 128 {
|
|
t.Fatalf("submit=%d rejected=%d", upstream.submitCalls, upstream.rejected.Load())
|
|
}
|
|
repo.mu.Lock()
|
|
defer repo.mu.Unlock()
|
|
for id, sub := range repo.batches {
|
|
if sub.ChanKey == "healthy" {
|
|
continue
|
|
}
|
|
if sub.Status != "prepared" || sub.PollCount != 0 || !sub.SubmittedAt.IsZero() || sub.LeaseToken != "" || repo.next[id].IsZero() {
|
|
t.Fatalf("只读退避改变办理意图: %+v", sub)
|
|
}
|
|
}
|
|
}
|