diff --git a/docs/superpowers/specs/2026-07-01-go-ingest-redesign-design.md b/docs/superpowers/specs/2026-07-01-go-ingest-redesign-design.md index 49bf5cd7..dc46d2a2 100644 --- a/docs/superpowers/specs/2026-07-01-go-ingest-redesign-design.md +++ b/docs/superpowers/specs/2026-07-01-go-ingest-redesign-design.md @@ -182,6 +182,13 @@ go/vehicle-gateway/ 接入层必须先写 RAW topic，再写 unified event。后续消费者只依赖 Kafka，不直接依赖接入进程内存。 +发布可靠性： + +- Kafka 生产端必须开启同步写入，RAW 写成功后才允许写 unified event。 +- Kafka 写入必须支持可配置重试、单次写超时和短退避，默认值为 `KAFKA_PUBLISH_ATTEMPTS=3`、`KAFKA_PUBLISH_TIMEOUT_MS=3000`、`KAFKA_PUBLISH_BACKOFF_MS=100`。 +- 当前阶段的重试只解决短暂网络抖动；后续阶段需要增加本地磁盘 spool/WAL，使 Kafka 长时间不可用时 RAW 不丢失，并支持恢复后补发。 +- Kafka topic 不允许自动创建，topic 和分区数由部署脚本或运维初始化，避免生产拼写错误造成隐性分流。 + ## TDengine 数据库设计 如果现有 TDengine 库不符合目标，可以新建库并迁移。建议库名： @@ -399,7 +406,7 @@ flowchart LR - 协议坏帧仍写 RAW topic，`parse_status=BAD_FRAME`。 - 可部分解析的帧写 `parse_status=PARTIAL`，保留 parse error。 -- Kafka 写失败时接入层不确认业务成功；协议是否应答按协议要求处理，但必须打错误日志和 metrics。 +- Kafka 写失败时接入层先按配置重试；重试耗尽后不写 unified event，必须打错误日志和 metrics。 - TDengine 写失败不提交 Kafka offset。 - MySQL 统计写失败不提交 Kafka offset。 - Redis 写失败不影响历史和统计，但要通过 metrics 暴露。 @@ -465,6 +472,7 @@ TDengine 连接： - Go gateway 能在 ECS 上接收真实 32960、808、宇通 MQTT 数据。 - 三种协议 RAW 都进入 Kafka。 +- Kafka 短暂写失败时，gateway 按配置重试；单元测试覆盖首次失败后成功和重试耗尽返回错误。 - 三种协议 RAW 和 parsed JSON 都进入 TDengine。 - 32960 和 808 的位置进入 `vehicle_locations`。 - 32960 和 808 的总里程采样进入 `vehicle_mileage_points`。 diff --git a/go/vehicle-gateway/cmd/gateway/main.go b/go/vehicle-gateway/cmd/gateway/main.go index abfdc625..26c9991d 100644 --- a/go/vehicle-gateway/cmd/gateway/main.go +++ b/go/vehicle-gateway/cmd/gateway/main.go @@ -138,7 +138,7 @@ func buildSink(logger *slog.Logger) (eventbus.Sink, error) { logger.Warn("KAFKA_BROKERS is empty; using log sink") return eventbus.NewLogSink(logger), nil } - return eventbus.NewKafkaSink(eventbus.KafkaConfig{ + sink, err := eventbus.NewKafkaSink(eventbus.KafkaConfig{ Brokers: brokers, RawTopics: map[envelope.Protocol]string{ envelope.ProtocolGB32960: env("KAFKA_TOPIC_GB32960_RAW", "vehicle.raw.gb32960.v1"), @@ -147,6 +147,14 @@ func buildSink(logger *slog.Logger) (eventbus.Sink, error) { }, UnifiedTopic: env("KAFKA_TOPIC_UNIFIED", "vehicle.event.unified.v1"), }) + if err != nil { + return nil, err + } + return eventbus.NewRetryingSink(sink, eventbus.RetryConfig{ + Attempts: envInt("KAFKA_PUBLISH_ATTEMPTS", 3), + Backoff: time.Duration(envInt("KAFKA_PUBLISH_BACKOFF_MS", 100)) * time.Millisecond, + AttemptTimeout: time.Duration(envInt("KAFKA_PUBLISH_TIMEOUT_MS", 3000)) * time.Millisecond, + }), nil } func env(key string, fallback string) string { diff --git a/go/vehicle-gateway/internal/eventbus/retry_sink.go b/go/vehicle-gateway/internal/eventbus/retry_sink.go new file mode 100644 index 00000000..02315b7b --- /dev/null +++ b/go/vehicle-gateway/internal/eventbus/retry_sink.go @@ -0,0 +1,80 @@ +package eventbus + +import ( + "context" + "time" + + "lingniu-vehicle-ingest/go/vehicle-gateway/internal/envelope" +) + +type RetryConfig struct { + Attempts int + Backoff time.Duration + AttemptTimeout time.Duration +} + +type RetryingSink struct { + delegate Sink + cfg RetryConfig +} + +func NewRetryingSink(delegate Sink, cfg RetryConfig) *RetryingSink { + if delegate == nil { + panic("retry delegate sink must not be nil") + } + if cfg.Attempts <= 0 { + cfg.Attempts = 1 + } + return &RetryingSink{delegate: delegate, cfg: cfg} +} + +func (s *RetryingSink) PublishRaw(ctx context.Context, env envelope.FrameEnvelope) error { + return s.withRetry(ctx, func(attemptCtx context.Context) error { + return s.delegate.PublishRaw(attemptCtx, env) + }) +} + +func (s *RetryingSink) PublishUnified(ctx context.Context, env envelope.FrameEnvelope) error { + return s.withRetry(ctx, func(attemptCtx context.Context) error { + return s.delegate.PublishUnified(attemptCtx, env) + }) +} + +func (s *RetryingSink) Close() error { + return s.delegate.Close() +} + +func (s *RetryingSink) withRetry(ctx context.Context, publish func(context.Context) error) error { + var lastErr error + for attempt := 1; attempt <= s.cfg.Attempts; attempt++ { + attemptCtx := ctx + cancel := func() {} + if s.cfg.AttemptTimeout > 0 { + attemptCtx, cancel = context.WithTimeout(ctx, s.cfg.AttemptTimeout) + } + err := publish(attemptCtx) + cancel() + if err == nil { + return nil + } + lastErr = err + if ctx.Err() != nil { + return ctx.Err() + } + if attempt == s.cfg.Attempts { + break + } + if s.cfg.Backoff > 0 { + timer := time.NewTimer(s.cfg.Backoff) + select { + case <-ctx.Done(): + if !timer.Stop() { + <-timer.C + } + return ctx.Err() + case <-timer.C: + } + } + } + return lastErr +} diff --git a/go/vehicle-gateway/internal/eventbus/retry_sink_test.go b/go/vehicle-gateway/internal/eventbus/retry_sink_test.go new file mode 100644 index 00000000..ca5641ec --- /dev/null +++ b/go/vehicle-gateway/internal/eventbus/retry_sink_test.go @@ -0,0 +1,80 @@ +package eventbus + +import ( + "context" + "errors" + "testing" + "time" + + "lingniu-vehicle-ingest/go/vehicle-gateway/internal/envelope" +) + +func TestRetryingSinkRetriesRawUntilSuccess(t *testing.T) { + delegate := &flakySink{rawFailures: 1} + sink := NewRetryingSink(delegate, RetryConfig{Attempts: 3}) + + err := sink.PublishRaw(context.Background(), envelope.FrameEnvelope{Protocol: envelope.ProtocolJT808}) + if err != nil { + t.Fatalf("PublishRaw() error = %v", err) + } + if delegate.rawCalls != 2 { + t.Fatalf("raw calls = %d, want 2", delegate.rawCalls) + } +} + +func TestRetryingSinkReturnsLastErrorAfterAttempts(t *testing.T) { + delegate := &flakySink{rawFailures: 3} + sink := NewRetryingSink(delegate, RetryConfig{Attempts: 2}) + + err := sink.PublishRaw(context.Background(), envelope.FrameEnvelope{Protocol: envelope.ProtocolJT808}) + if !errors.Is(err, errFlakyPublish) { + t.Fatalf("PublishRaw() error = %v, want %v", err, errFlakyPublish) + } + if delegate.rawCalls != 2 { + t.Fatalf("raw calls = %d, want 2", delegate.rawCalls) + } +} + +func TestRetryingSinkStopsWhenContextCanceledDuringBackoff(t *testing.T) { + delegate := &flakySink{rawFailures: 10} + sink := NewRetryingSink(delegate, RetryConfig{Attempts: 5, Backoff: time.Minute}) + ctx, cancel := context.WithCancel(context.Background()) + cancel() + + err := sink.PublishRaw(ctx, envelope.FrameEnvelope{Protocol: envelope.ProtocolJT808}) + if !errors.Is(err, context.Canceled) { + t.Fatalf("PublishRaw() error = %v, want context canceled", err) + } + if delegate.rawCalls != 1 { + t.Fatalf("raw calls = %d, want 1", delegate.rawCalls) + } +} + +var errFlakyPublish = errors.New("publish failed") + +type flakySink struct { + rawFailures int + unifiedFailures int + rawCalls int + unifiedCalls int +} + +func (s *flakySink) PublishRaw(context.Context, envelope.FrameEnvelope) error { + s.rawCalls++ + if s.rawCalls <= s.rawFailures { + return errFlakyPublish + } + return nil +} + +func (s *flakySink) PublishUnified(context.Context, envelope.FrameEnvelope) error { + s.unifiedCalls++ + if s.unifiedCalls <= s.unifiedFailures { + return errFlakyPublish + } + return nil +} + +func (s *flakySink) Close() error { + return nil +}