From aff3e73650fcbd0701e43eff56001fbff2da8ec8 Mon Sep 17 00:00:00 2001 From: Bao ZhiFei Date: Fri, 7 Aug 2026 22:14:50 -0700 Subject: [PATCH] fix: run dependent pipeline stages sequentially --- internal/core/pipeline.go | 64 +++++++++---------------- internal/core/pipeline_test.go | 86 ++++++++++++++++++++++++++++++++++ 2 files changed, 109 insertions(+), 41 deletions(-) create mode 100644 internal/core/pipeline_test.go diff --git a/internal/core/pipeline.go b/internal/core/pipeline.go index 86aa441..986d93b 100644 --- a/internal/core/pipeline.go +++ b/internal/core/pipeline.go @@ -4,7 +4,6 @@ import ( "context" "fmt" "sort" - "sync" "time" "RealityChecker/internal/detectors" @@ -69,8 +68,8 @@ func (p *Pipeline) Execute(ctx context.Context, domain string) (*types.Detection EarlyExit: false, } - // 并发执行检测阶段，提高检测效率 - p.executeStagesConcurrently(ctx, pipelineCtx) + // 按依赖顺序执行单个域名的检测阶段；域名之间由批量管理器并发。 + p.executeStages(ctx, pipelineCtx) // 计算总耗时 pipelineCtx.Result.Duration = time.Since(startTime) @@ -81,8 +80,8 @@ func (p *Pipeline) Execute(ctx context.Context, domain string) (*types.Detection return pipelineCtx.Result, nil } -// executeStagesConcurrently 并发执行检测阶段 -func (p *Pipeline) executeStagesConcurrently(ctx context.Context, pipelineCtx *types.PipelineContext) { +// executeStages 按依赖关系执行检测阶段 +func (p *Pipeline) executeStages(ctx context.Context, pipelineCtx *types.PipelineContext) { // 将检测阶段分为两组：阻塞检测和网络检测 var blockingStages []types.DetectionStage var networkStages []types.DetectionStage @@ -123,50 +122,33 @@ func (p *Pipeline) executeStagesConcurrently(ctx context.Context, pipelineCtx *t return } - // 并发执行网络检测阶段 + // TLS/CDN 与热门网站检测共享结果，必须按优先级顺序执行。 if len(networkStages) > 0 { - p.executeNetworkStagesConcurrently(ctx, pipelineCtx, networkStages) + p.executeNetworkStagesSequentially(ctx, pipelineCtx, networkStages) } } -// executeNetworkStagesConcurrently 并发执行网络检测阶段 -func (p *Pipeline) executeNetworkStagesConcurrently(ctx context.Context, pipelineCtx *types.PipelineContext, stages []types.DetectionStage) { - // 使用信号量控制网络检测的并发数 - networkConcurrency := 4 // 网络检测使用4个并发 - semaphore := make(chan struct{}, networkConcurrency) - - var wg sync.WaitGroup - for i, stage := range stages { - wg.Add(1) - go func(index int, s types.DetectionStage) { - defer wg.Done() - - // 获取信号量 - select { - case semaphore <- struct{}{}: - defer func() { - <-semaphore - }() - case <-ctx.Done(): - return - } - - // 执行检测阶段 - func() { - defer func() { - if r := recover(); r != nil { - pipelineCtx.Result.Error = fmt.Errorf("检测阶段 %s panic: %v", s.Name(), r) - } - }() +// executeNetworkStagesSequentially 按优先级执行有数据依赖的网络检测阶段。 +func (p *Pipeline) executeNetworkStagesSequentially(ctx context.Context, pipelineCtx *types.PipelineContext, stages []types.DetectionStage) { + for _, stage := range stages { + select { + case <-ctx.Done(): + return + default: + } - if err := s.Execute(pipelineCtx); err != nil { - pipelineCtx.Result.Error = err + func() { + defer func() { + if recovered := recover(); recovered != nil { + pipelineCtx.Result.Error = fmt.Errorf("检测阶段 %s panic: %v", stage.Name(), recovered) } }() - }(i, stage) - } - wg.Wait() + if err := stage.Execute(pipelineCtx); err != nil { + pipelineCtx.Result.Error = err + } + }() + } } // evaluateSuitability 评估适合性 diff --git a/internal/core/pipeline_test.go b/internal/core/pipeline_test.go new file mode 100644 index 0000000..0a09e54 --- /dev/null +++ b/internal/core/pipeline_test.go @@ -0,0 +1,86 @@ +package core + +import ( + "context" + "errors" + "reflect" + "strings" + "testing" + + "RealityChecker/internal/types" +) + +type testStage struct { + name string + execute func(*types.PipelineContext) error +} + +func (s testStage) Execute(ctx *types.PipelineContext) error { return s.execute(ctx) } +func (s testStage) CanEarlyExit() bool { return false } +func (s testStage) Priority() int { return 0 } +func (s testStage) Name() string { return s.name } + +func TestExecuteNetworkStagesSequentiallyPreservesDependencies(t *testing.T) { + var order []string + pipelineCtx := &types.PipelineContext{ + Result: &types.DetectionResult{}, + } + + stages := []types.DetectionStage{ + testStage{name: "redirect", execute: func(ctx *types.PipelineContext) error { + order = append(order, "redirect") + ctx.Result.Network = &types.NetworkResult{FinalDomain: "example.com"} + return nil + }}, + testStage{name: "tls", execute: func(ctx *types.PipelineContext) error { + order = append(order, "tls") + if ctx.Result.Network == nil || ctx.Result.Network.FinalDomain != "example.com" { + return errors.New("redirect result is unavailable") + } + ctx.Result.TLS = &types.TLSResult{SupportsTLS13: true} + return nil + }}, + testStage{name: "hot-website", execute: func(ctx *types.PipelineContext) error { + order = append(order, "hot-website") + if ctx.Result.TLS == nil || !ctx.Result.TLS.SupportsTLS13 { + return errors.New("TLS result is unavailable") + } + return nil + }}, + } + + (&Pipeline{}).executeNetworkStagesSequentially(context.Background(), pipelineCtx, stages) + + if pipelineCtx.Result.Error != nil { + t.Fatalf("unexpected stage error: %v", pipelineCtx.Result.Error) + } + wantOrder := []string{"redirect", "tls", "hot-website"} + if !reflect.DeepEqual(order, wantOrder) { + t.Fatalf("execution order = %v, want %v", order, wantOrder) + } +} + +func TestExecuteNetworkStagesSequentiallyRecoversPanic(t *testing.T) { + continued := false + pipelineCtx := &types.PipelineContext{ + Result: &types.DetectionResult{}, + } + stages := []types.DetectionStage{ + testStage{name: "panic-stage", execute: func(*types.PipelineContext) error { + panic("boom") + }}, + testStage{name: "next-stage", execute: func(*types.PipelineContext) error { + continued = true + return nil + }}, + } + + (&Pipeline{}).executeNetworkStagesSequentially(context.Background(), pipelineCtx, stages) + + if !continued { + t.Fatal("pipeline did not continue after recovering a stage panic") + } + if pipelineCtx.Result.Error == nil || !strings.Contains(pipelineCtx.Result.Error.Error(), "panic-stage") { + t.Fatalf("panic error = %v, want stage name", pipelineCtx.Result.Error) + } +}