369 lines
11 KiB
Go
369 lines
11 KiB
Go
package orbit
|
||
|
||
import (
|
||
"context"
|
||
"errors"
|
||
"fmt"
|
||
"math/rand/v2"
|
||
"sync/atomic"
|
||
"time"
|
||
)
|
||
|
||
var ErrEventConsumerAlreadyRun = errors.New("event consumer may only Run once")
|
||
var ErrEventShutdownTimeout = errors.New("event consumer shutdown grace period expired")
|
||
|
||
// EventConsumerOptions 定义自动消费器。Handle必须以事件ID做持久化去重，
|
||
// 在去重记录和业务效果同一事务提交后才返回nil；提交结果未知时返回error。
|
||
// error或panic不ACK，平台在租约到期后按原预算补发；SDK不在内存中立即重跑。
|
||
type EventConsumerOptions struct {
|
||
SubscriptionID string
|
||
Capacity uint32 // Capacity 最大本地在途事件/处理器，默认4，范围1至32。
|
||
Handle func(context.Context, ResultEvent) error // Handle不接收租约秘密；必须响应context取消。
|
||
ShutdownTimeout time.Duration // ShutdownTimeout停止时取消处理器，等待返回及ACK，默认30秒，最多5分钟。
|
||
ReconnectMinDelay time.Duration // ReconnectMinDelay默认200毫秒，至少10毫秒。
|
||
ReconnectMaxDelay time.Duration // ReconnectMaxDelay默认5秒，最多30秒；指数退避并加入抖动。
|
||
}
|
||
|
||
// EventConsumerEvent 只用于本地观察，缓冲满可丢弃；不是可靠结果或审计。
|
||
type EventConsumerEvent struct {
|
||
Type string // Type为connected、reconnecting、handler_started、handler_failed、acknowledged、proof_lost、draining。
|
||
EventID string
|
||
Code string // Code不含业务异常、数据或秘密。
|
||
}
|
||
type EventConsumerStats struct {
|
||
Running int64
|
||
Started uint64
|
||
HandlerFailures uint64
|
||
Acknowledged uint64
|
||
Reconnects uint64
|
||
DroppedEvents uint64
|
||
}
|
||
type EventConsumerRunError struct {
|
||
Cause error
|
||
Unconfirmed int // Unconfirmed本进程已知且未确认的事件数，不是订阅全局积压。
|
||
Running int64 // Running仍未退出的处理器；Go不能强杀忽略context的函数。
|
||
}
|
||
|
||
func (e *EventConsumerRunError) Error() string {
|
||
return fmt.Sprintf("event consumer stopped with %d unconfirmed events and %d running handlers: %v", e.Unconfirmed, e.Running, e.Cause)
|
||
}
|
||
func (e *EventConsumerRunError) Unwrap() error { return e.Cause }
|
||
|
||
// EventConsumer 自动维护心跳、重连、处理器并发和ACK核对。只能Run一次。
|
||
// 服务端交付租约不是业务锁；跨进程重复及并行消费由业务Inbox唯一键和事务处理。
|
||
type EventConsumer struct {
|
||
client *Client
|
||
options EventConsumerOptions
|
||
used atomic.Bool
|
||
events chan EventConsumerEvent
|
||
running atomic.Int64
|
||
started, failures, acked, reconnects, dropped atomic.Uint64
|
||
}
|
||
|
||
func NewEventConsumer(client *Client, in EventConsumerOptions) (*EventConsumer, error) {
|
||
if client == nil || client.token == "" || !validID(in.SubscriptionID) || in.Handle == nil {
|
||
return nil, errors.New("authenticated client, subscription and durable event handler are required")
|
||
}
|
||
if in.Capacity == 0 {
|
||
in.Capacity = 4
|
||
}
|
||
if in.ShutdownTimeout == 0 {
|
||
in.ShutdownTimeout = 30 * time.Second
|
||
}
|
||
if in.ReconnectMinDelay == 0 {
|
||
in.ReconnectMinDelay = 200 * time.Millisecond
|
||
}
|
||
if in.ReconnectMaxDelay == 0 {
|
||
in.ReconnectMaxDelay = 5 * time.Second
|
||
}
|
||
if in.Capacity > 32 || in.ShutdownTimeout < time.Millisecond || in.ShutdownTimeout > 5*time.Minute || in.ReconnectMinDelay < 10*time.Millisecond || in.ReconnectMaxDelay < in.ReconnectMinDelay || in.ReconnectMaxDelay > 30*time.Second {
|
||
return nil, errors.New("invalid event consumer capacity or durations")
|
||
}
|
||
return &EventConsumer{client: client, options: in, events: make(chan EventConsumerEvent, 128)}, nil
|
||
}
|
||
func (c *EventConsumer) Events() <-chan EventConsumerEvent { return c.events }
|
||
func (c *EventConsumer) Stats() EventConsumerStats {
|
||
return EventConsumerStats{Running: c.running.Load(), Started: c.started.Load(), HandlerFailures: c.failures.Load(), Acknowledged: c.acked.Load(), Reconnects: c.reconnects.Load(), DroppedEvents: c.dropped.Load()}
|
||
}
|
||
func (c *EventConsumer) emit(typ, id, code string) {
|
||
select {
|
||
case c.events <- EventConsumerEvent{Type: typ, EventID: id, Code: code}:
|
||
default:
|
||
c.dropped.Add(1)
|
||
}
|
||
}
|
||
func (c *EventConsumer) String() string {
|
||
return fmt.Sprintf("EventConsumer{SubscriptionID:%s Capacity:%d secrets:redacted}", c.options.SubscriptionID, c.options.Capacity)
|
||
}
|
||
func (c *EventConsumer) GoString() string { return c.String() }
|
||
|
||
type consumerJob struct {
|
||
proof EventProof
|
||
complete bool
|
||
}
|
||
type consumerResult struct {
|
||
id string
|
||
failed bool
|
||
}
|
||
|
||
// Run在ctx取消后停止启动新处理器，取消在途处理器，在收尾时限内确认已提交结果。
|
||
// 收到认证/协议等永久错误时也取消并等待在途处理器；未确认结果仍保留在服务端。
|
||
// 重连先核对已完成处理器的原证明；证明被替代时等待新投递，由业务去重命中后确认。
|
||
func (c *EventConsumer) Run(parent context.Context) error {
|
||
if !c.used.CompareAndSwap(false, true) {
|
||
return ErrEventConsumerAlreadyRun
|
||
}
|
||
defer close(c.events)
|
||
if parent.Err() != nil {
|
||
return nil
|
||
}
|
||
ctx, cancel := context.WithCancel(context.WithoutCancel(parent))
|
||
defer cancel()
|
||
handlerCtx, stopHandlers := context.WithCancel(ctx)
|
||
defer stopHandlers()
|
||
abort := make(chan struct{})
|
||
watchDone := make(chan struct{})
|
||
go func() {
|
||
defer close(watchDone)
|
||
select {
|
||
case <-ctx.Done():
|
||
return
|
||
case <-parent.Done():
|
||
case <-abort:
|
||
}
|
||
stopHandlers()
|
||
timer := time.NewTimer(c.options.ShutdownTimeout)
|
||
defer timer.Stop()
|
||
select {
|
||
case <-ctx.Done():
|
||
case <-timer.C:
|
||
cancel()
|
||
}
|
||
}()
|
||
defer func() { cancel(); <-watchDone }()
|
||
var conn *EventConnection
|
||
var heartbeatDone <-chan struct{}
|
||
closeConnection := func() {
|
||
if conn != nil {
|
||
_ = conn.Close()
|
||
<-heartbeatDone
|
||
conn = nil
|
||
}
|
||
}
|
||
defer closeConnection()
|
||
jobs := make(map[string]*consumerJob)
|
||
results := make(chan consumerResult, c.options.Capacity)
|
||
draining := false
|
||
var cause error
|
||
beginDrain := func(err error) {
|
||
if !draining {
|
||
draining = true
|
||
stopHandlers()
|
||
close(abort)
|
||
c.emit("draining", "", "")
|
||
}
|
||
if cause == nil {
|
||
cause = err
|
||
}
|
||
}
|
||
finish := func(err error) error {
|
||
if err == nil {
|
||
return nil
|
||
}
|
||
return &EventConsumerRunError{Cause: err, Unconfirmed: len(jobs), Running: c.running.Load()}
|
||
}
|
||
backoff := c.options.ReconnectMinDelay
|
||
var dialAt time.Time
|
||
everConnected := false
|
||
disconnect := func(err error) {
|
||
closeConnection()
|
||
if !eventRetryable(err) {
|
||
beginDrain(err)
|
||
return
|
||
}
|
||
floor := backoff / 2
|
||
dialAt = time.Now().Add(floor + time.Duration(rand.Int64N(int64(backoff-floor)+1)))
|
||
backoff = min(backoff*2, c.options.ReconnectMaxDelay)
|
||
c.emit("reconnecting", "", eventErrorCode(err))
|
||
}
|
||
for {
|
||
if parent.Err() != nil {
|
||
beginDrain(nil)
|
||
}
|
||
if ctx.Err() != nil {
|
||
return finish(ErrEventShutdownTimeout)
|
||
}
|
||
if draining && (len(jobs) == 0 || cause != nil && c.running.Load() == 0) {
|
||
return finish(cause)
|
||
}
|
||
if conn == nil && cause == nil && !time.Now().Before(dialAt) {
|
||
if everConnected {
|
||
c.reconnects.Add(1)
|
||
}
|
||
var err error
|
||
conn, err = c.client.DialEvents(ctx, EventRegistration{SubscriptionID: c.options.SubscriptionID, Capacity: c.options.Capacity})
|
||
if err != nil {
|
||
disconnect(err)
|
||
continue
|
||
}
|
||
everConnected = true
|
||
heartbeatDone = startEventHeartbeat(ctx, conn)
|
||
c.emit("connected", "", "")
|
||
continue
|
||
}
|
||
if conn != nil {
|
||
if err := conn.Err(); err != nil {
|
||
disconnect(err)
|
||
continue
|
||
}
|
||
// 一个ACK在途；读取与自动心跳独立，不需要接收方不断调用Next维持连接。
|
||
acknowledged := false
|
||
for id, j := range jobs {
|
||
if !j.complete {
|
||
continue
|
||
}
|
||
messageID, err := NewEventMessageID()
|
||
if err == nil {
|
||
_, err = conn.Ack(ctx, messageID, j.proof)
|
||
}
|
||
if err == nil {
|
||
delete(jobs, id)
|
||
c.acked.Add(1)
|
||
c.emit("acknowledged", id, "")
|
||
backoff = c.options.ReconnectMinDelay
|
||
} else {
|
||
var remote *EventError
|
||
if errors.As(err, &remote) && remote.Code == "DELIVERY_LEASE_LOST" {
|
||
delete(jobs, id)
|
||
c.emit("proof_lost", id, remote.Code)
|
||
} else {
|
||
disconnect(err)
|
||
}
|
||
}
|
||
acknowledged = true
|
||
break
|
||
}
|
||
if acknowledged {
|
||
continue
|
||
}
|
||
}
|
||
var deliveries <-chan EventGrant
|
||
var disconnected <-chan struct{}
|
||
if conn != nil {
|
||
disconnected = conn.Done()
|
||
if !draining && len(jobs) < int(c.options.Capacity) {
|
||
deliveries = conn.deliveries
|
||
}
|
||
}
|
||
var parentDone <-chan struct{}
|
||
if !draining {
|
||
parentDone = parent.Done()
|
||
}
|
||
// 定时器只负责重连；活跃连接没有每轮轮询。
|
||
var wake <-chan time.Time
|
||
var timer *time.Timer
|
||
if conn == nil && cause == nil {
|
||
timer = time.NewTimer(max(time.Millisecond, time.Until(dialAt)))
|
||
wake = timer.C
|
||
}
|
||
select {
|
||
case <-ctx.Done():
|
||
case <-parentDone:
|
||
beginDrain(nil)
|
||
case <-wake:
|
||
case <-disconnected:
|
||
disconnect(conn.Err())
|
||
case result := <-results:
|
||
j := jobs[result.id]
|
||
if result.failed {
|
||
delete(jobs, result.id)
|
||
c.failures.Add(1)
|
||
c.emit("handler_failed", result.id, "HANDLER_FAILED")
|
||
} else {
|
||
j.complete = true
|
||
}
|
||
case grant := <-deliveries:
|
||
if parent.Err() != nil || conn.Err() != nil {
|
||
break
|
||
}
|
||
id := grant.Event.ID
|
||
if j := jobs[id]; j != nil {
|
||
if j.proof.DeliveryID != grant.Delivery.ID {
|
||
disconnect(ErrEventProtocol)
|
||
break
|
||
}
|
||
if grant.Delivery.LeaseEpoch > j.proof.Epoch {
|
||
j.proof = grant.Proof()
|
||
}
|
||
break
|
||
}
|
||
jobs[id] = &consumerJob{proof: grant.Proof()}
|
||
c.running.Add(1)
|
||
c.started.Add(1)
|
||
c.emit("handler_started", id, "")
|
||
go func(event ResultEvent) {
|
||
failed := true
|
||
defer func() {
|
||
_ = recover()
|
||
c.running.Add(-1)
|
||
select {
|
||
case results <- consumerResult{id: event.ID, failed: failed}:
|
||
case <-ctx.Done():
|
||
}
|
||
}()
|
||
failed = c.options.Handle(handlerCtx, event) != nil
|
||
}(grant.Event)
|
||
}
|
||
if timer != nil {
|
||
timer.Stop()
|
||
}
|
||
}
|
||
}
|
||
func startEventHeartbeat(ctx context.Context, conn *EventConnection) <-chan struct{} {
|
||
done := make(chan struct{})
|
||
go func() {
|
||
defer close(done)
|
||
ticker := time.NewTicker(time.Duration(conn.Welcome().HeartbeatIntervalMS) * time.Millisecond)
|
||
defer ticker.Stop()
|
||
for {
|
||
select {
|
||
case <-ctx.Done():
|
||
return
|
||
case <-conn.Done():
|
||
return
|
||
case <-ticker.C:
|
||
id, err := NewEventMessageID()
|
||
if err == nil {
|
||
_, err = conn.Ping(ctx, id)
|
||
}
|
||
if err != nil {
|
||
conn.fail(err)
|
||
return
|
||
}
|
||
}
|
||
}
|
||
}()
|
||
return done
|
||
}
|
||
func eventRetryable(err error) bool {
|
||
if errors.Is(err, ErrEventDisconnected) || errors.Is(err, ErrEventBackpressure) || errors.Is(err, context.DeadlineExceeded) {
|
||
return true
|
||
}
|
||
var remote *EventError
|
||
if errors.As(err, &remote) {
|
||
return remote.Code == "UNAVAILABLE"
|
||
}
|
||
var httpErr *HTTPError
|
||
if errors.As(err, &httpErr) {
|
||
return httpErr.StatusCode == 408 || httpErr.StatusCode == 429 || httpErr.StatusCode >= 500 && httpErr.StatusCode <= 599
|
||
}
|
||
return false
|
||
}
|
||
func eventErrorCode(err error) string {
|
||
var remote *EventError
|
||
if errors.As(err, &remote) {
|
||
return remote.Code
|
||
}
|
||
return "TRANSPORT_INTERRUPTED"
|
||
}
|