reliability_regression_tests

package
v1.1.64 Latest Latest
Warning

This package is not in the latest version of its module.

Go to latest
Published: Jul 22, 2026 License: Apache-2.0 Imports: 8 Imported by: 0

README

EventBus 可靠性回归测试

📋 概述

本目录包含 EventBus 的可靠性和故障恢复测试用例,专门测试 Hollywood Actor Pool 的自恢复能力。

🎯 测试目标

核心可靠性特性
  1. Actor 自动重启 - Supervisor 机制在 Actor panic 后自动重启
  2. 消息保证送达 - Actor 重启后,缓冲区中的消息不丢失
  3. 故障隔离 - 单个 Actor 故障不影响其他 Actor
  4. 优雅降级 - 达到最大重启次数后的行为
  5. DeadLetter 处理 - 无法送达的消息进入 DeadLetter 队列

🧪 测试场景

1. Actor Panic 恢复测试
  • 场景: Actor 处理消息时 panic
  • 预期: Supervisor 自动重启 Actor,继续处理后续消息
  • 验证: 消息不丢失,处理顺序正确
2. 多次 Panic 重启测试
  • 场景: Actor 连续多次 panic(不超过 MaxRestarts)
  • 预期: 每次 panic 后都能自动重启
  • 验证: 重启次数正确,消息最终都被处理
3. 达到最大重启次数测试
  • 场景: Actor panic 次数超过 MaxRestarts(3次)
  • 预期: Actor 停止重启,消息进入 DeadLetter
  • 验证: DeadLetterEvent 被触发,消息可追踪
4. 故障隔离测试
  • 场景: 多个 Actor 中只有一个 panic
  • 预期: 只有故障 Actor 重启,其他 Actor 正常工作
  • 验证: 其他 Actor 的消息处理不受影响
5. 消息缓冲区保证测试
  • 场景: Actor 重启期间,新消息持续到达
  • 预期: 重启后,缓冲区中的消息按顺序处理
  • 验证: 消息顺序正确,无丢失
6. 并发故障恢复测试
  • 场景: 多个 Actor 同时 panic
  • 预期: 所有 Actor 都能独立恢复
  • 验证: 所有消息最终都被处理
7. EventStream 监控测试
  • 场景: 订阅 ActorRestartedEvent 和 DeadLetterEvent
  • 预期: 所有故障事件都能被监控到
  • 验证: 事件数量和内容正确

📊 测试指标

可靠性指标
  • 消息送达率: 应该 100%(除非达到 MaxRestarts)
  • 重启成功率: 应该 100%(在 MaxRestarts 范围内)
  • 故障隔离率: 应该 100%(故障不扩散)
  • 恢复时间: 应该 < 100ms
性能指标
  • 重启延迟: Actor 重启到恢复处理的时间
  • 消息积压: 重启期间的消息堆积情况
  • 吞吐量影响: 故障恢复对整体吞吐量的影响

🚀 运行测试

运行所有可靠性测试
cd jxt-core/tests/eventbus/reliability_regression_tests
go test -v -timeout 180s
运行特定测试
# Actor Panic 恢复测试
go test -v -run TestActorPanicRecovery

# 多次重启测试
go test -v -run TestMultiplePanicRestarts

# 最大重启次数测试
go test -v -run TestMaxRestartsExceeded

# 故障隔离测试
go test -v -run TestFaultIsolation

# 消息缓冲区测试
go test -v -run TestMessageBufferGuarantee

# 并发故障恢复测试
go test -v -run TestConcurrentFaultRecovery

# EventStream 监控测试
go test -v -run TestEventStreamMonitoring

📝 测试文件

  • actor_recovery_test.go - Actor 恢复相关测试
  • fault_isolation_test.go - 故障隔离测试
  • message_guarantee_test.go - 消息保证送达测试
  • eventstream_monitoring_test.go - EventStream 监控测试
  • test_helper.go - 测试辅助函数

🔧 Hollywood Supervisor 机制

配置参数
actor.WithMaxRestarts(3)  // 最大重启次数(默认 3)
actor.WithInboxSize(1000) // Inbox 缓冲区大小(默认 1000)
重启策略
  • OneForOne: 只重启故障的 Actor(Hollywood 默认)
  • 自动重启: panic 后自动重启,无需手动干预
  • 消息保证: Inbox 中的消息在重启后继续处理
事件通知
  • actor.ActorRestartedEvent - Actor 重启事件
  • actor.DeadLetterEvent - 消息无法送达事件

✅ 验收标准

必须通过的测试
  • Actor Panic 恢复测试
  • 多次 Panic 重启测试
  • 达到最大重启次数测试
  • 故障隔离测试
  • 消息缓冲区保证测试
  • 并发故障恢复测试
  • EventStream 监控测试
性能要求
  • 重启延迟 < 100ms
  • 消息送达率 = 100%(在 MaxRestarts 范围内)
  • 故障隔离率 = 100%

📚 参考资料

Documentation

Index

Constants

This section is empty.

Variables

This section is empty.

Functions

This section is empty.

Types

type EventCollector

type EventCollector struct {
	// contains filtered or unexported fields
}

EventCollector 事件收集器

func NewEventCollector

func NewEventCollector() *EventCollector

NewEventCollector 创建事件收集器

func (*EventCollector) AddDeadLetterEvent

func (ec *EventCollector) AddDeadLetterEvent(event interface{})

AddDeadLetterEvent 添加 DeadLetter 事件

func (*EventCollector) AddInitializedEvent

func (ec *EventCollector) AddInitializedEvent(event interface{})

AddInitializedEvent 添加初始化事件

func (*EventCollector) AddRestartEvent

func (ec *EventCollector) AddRestartEvent(event interface{})

AddRestartEvent 添加重启事件

func (*EventCollector) AddStartedEvent

func (ec *EventCollector) AddStartedEvent(event interface{})

AddStartedEvent 添加启动事件

func (*EventCollector) AddStoppedEvent

func (ec *EventCollector) AddStoppedEvent(event interface{})

AddStoppedEvent 添加停止事件

func (*EventCollector) GetDeadLetterCount

func (ec *EventCollector) GetDeadLetterCount() int

GetDeadLetterCount 获取 DeadLetter 次数

func (*EventCollector) GetInitializedCount

func (ec *EventCollector) GetInitializedCount() int

GetInitializedCount 获取初始化次数

func (*EventCollector) GetRestartCount

func (ec *EventCollector) GetRestartCount() int

GetRestartCount 获取重启次数

func (*EventCollector) GetStartedCount

func (ec *EventCollector) GetStartedCount() int

GetStartedCount 获取启动次数

func (*EventCollector) GetStoppedCount

func (ec *EventCollector) GetStoppedCount() int

GetStoppedCount 获取停止次数

func (*EventCollector) Reset

func (ec *EventCollector) Reset()

Reset 重置所有计数器

type TestHelper

type TestHelper struct {
	// contains filtered or unexported fields
}

TestHelper 测试辅助工具

func NewTestHelper

func NewTestHelper(t *testing.T) *TestHelper

NewTestHelper 创建测试辅助工具

func (*TestHelper) AddCleanup

func (h *TestHelper) AddCleanup(fn func())

AddCleanup 添加清理函数

func (*TestHelper) AssertEqual

func (h *TestHelper) AssertEqual(expected, actual interface{}, message string)

AssertEqual 断言相等

func (*TestHelper) AssertError

func (h *TestHelper) AssertError(err error, message string)

AssertError 断言有错误

func (*TestHelper) AssertGreaterThan

func (h *TestHelper) AssertGreaterThan(actual, threshold int64, message string)

AssertGreaterThan 断言大于

func (*TestHelper) AssertGreaterThanOrEqual

func (h *TestHelper) AssertGreaterThanOrEqual(actual, threshold int64, message string)

AssertGreaterThanOrEqual 断言大于等于

func (*TestHelper) AssertLessThan

func (h *TestHelper) AssertLessThan(actual, threshold int64, message string)

AssertLessThan 断言小于

func (*TestHelper) AssertNoError

func (h *TestHelper) AssertNoError(err error, message string)

AssertNoError 断言无错误

func (*TestHelper) AssertTrue

func (h *TestHelper) AssertTrue(condition bool, message string)

AssertTrue 断言为真

func (*TestHelper) Cleanup

func (h *TestHelper) Cleanup()

Cleanup 执行所有清理函数

func (*TestHelper) CloseEventBus

func (h *TestHelper) CloseEventBus(bus eventbus.EventBus)

CloseEventBus 关闭 EventBus

func (*TestHelper) CreateKafkaEventBus

func (h *TestHelper) CreateKafkaEventBus(clientID string) eventbus.EventBus

CreateKafkaEventBus 创建 Kafka EventBus

func (*TestHelper) CreateMemoryEventBus

func (h *TestHelper) CreateMemoryEventBus() eventbus.EventBus

CreateMemoryEventBus 创建 Memory EventBus

func (*TestHelper) CreateNATSEventBus

func (h *TestHelper) CreateNATSEventBus(clientID, streamName string, subjects []string, durableName string) eventbus.EventBus

CreateNATSEventBus 创建 NATS JetStream EventBus,并在测试结束后清理 Stream

func (*TestHelper) GetTimestamp

func (h *TestHelper) GetTimestamp() int64

GetTimestamp 获取时间戳

func (*TestHelper) Logf

func (h *TestHelper) Logf(format string, args ...interface{})

Logf 日志输出

func (*TestHelper) Sleep

func (h *TestHelper) Sleep(duration time.Duration)

Sleep 睡眠

func (*TestHelper) WaitForCondition

func (h *TestHelper) WaitForCondition(condition func() bool, timeout time.Duration, message string) bool

WaitForCondition 等待条件满足

func (*TestHelper) WaitForMessages

func (h *TestHelper) WaitForMessages(received *int64, expected int64, timeout time.Duration) bool

WaitForMessages 等待消息数量达到预期

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL