feat: add reliable event consumer SDK and process recovery tests

2026-09-17 20:22:41 +08:00
parent 8485a662b2
commit b17d9b1977
25 changed files with 2170 additions and 28 deletions
--- a/sdk/go/event_consumer.go
+++ b/sdk/go/event_consumer.go
@@ -0,0 +1,368 @@
+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"
+}