From 12fda00df9084b0736f8d4c79f0a757763e3eaa3 Mon Sep 17 00:00:00 2001 From: windyboy Date: Sun, 16 Nov 2025 17:24:16 +0800 Subject: [PATCH] =?UTF-8?q?=E2=9C=A8=20Introduce=20`seed-telegrams`=20tool?= =?UTF-8?q?=20for=20generating=20synthetic=20telegrams=20for=20testing=20a?= =?UTF-8?q?nd=20development.=20Enhance=20README=20with=20detailed=20usage?= =?UTF-8?q?=20instructions,=20command-line=20parameters,=20and=20examples.?= =?UTF-8?q?=20Add=20unit=20tests=20for=20telegram=20generation=20and=20sta?= =?UTF-8?q?tus=20selection=20logic=20to=20ensure=20robustness.=20Update=20?= =?UTF-8?q?documentation=20to=20reflect=20new=20features=20and=20usage=20s?= =?UTF-8?q?cenarios.?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- README.md | 81 ++++++++++ cmd/seed-telegrams/main.go | 262 ++++++++++++++++++++++++------- cmd/seed-telegrams/main_test.go | 211 +++++++++++++++++++++++++ docs/dev-guide.md | 19 +++ docs/reliability.md | 21 +++ internal/app/processor.go | 12 +- internal/infra/nats/consumer.go | 158 +++++++++++++++++-- internal/infra/nats/jetstream.go | 40 +++-- internal/infra/nats/publisher.go | 6 + 9 files changed, 730 insertions(+), 80 deletions(-) create mode 100644 cmd/seed-telegrams/main_test.go diff --git a/README.md b/README.md index dfbcf56..eb7b0e2 100644 --- a/README.md +++ b/README.md @@ -532,6 +532,87 @@ Flight plan messages contain detailed flight planning information. -) ``` +## Seed Tool: `cmd/seed-telegrams` + +`seed-telegrams` 是一个开发/测试用的报文发生器，用来向 NATS/JetStream 持续或突发地发送合成电报（ARR/DEP/CNL/DLA/FPL），用于驱动解析与下游流水线。 + +### 支持的类别与内容 + +生成的电报遵循与解析器相同的格式约定： + +- ARR / DEP：支持无 SSR、合法简单 SSR，以及刻意构造为“当前正则无法解析”的复杂 SSR。 +- CNL / DLA：与 `internal/parsers/aviation_parser_test.go` 中的测试样例同一类结构。 +- FPL：生成包含多行 route 与 `OtherInfo` 字段的完整 FPL，`OtherInfo` 中会随机组合 `PBN/`, `NAV/`, `REG/`, `EET/`, `SEL/`, `PER/`, `RIF/`, `RMK/` 等片段，以覆盖解析逻辑。 + +### 命令行参数 + +常用参数： + +- `--nats-url`：NATS 地址（默认 `nats://127.0.0.1:4222`，为空字符串则完全不连接 NATS）。 +- `--subject`：普通 NATS 发布 subject（默认 `telegram.raw`）。 +- `--jetstream`：是否使用 JetStream 发布。 +- `--stream` / `--js-subject`：JetStream 相关选项。 +- `--count`：要发送的电报数量（`mode=burst` 或有上限的 interval/mixed 时生效）。 +- `--category`：`ARR|DEP|CNL|DLA|FPL|mixed`，`mixed` 表示在五类中随机选择。 +- `--status`：`parsed|header_error|body_error|publish_error|repository_error|random`。 + - `random` 模式下，合法报文偏向标记为 `parsed`，刻意非法报文偏向标记为 `body_error`。 +- `--error-reason`：错误原因说明，将写入元数据 header（默认 `synthetic test payload`）。 +- `--dry-run`：只打印电报内容，不真正发布到 NATS。 +- `--header-format`：`json|none`，控制是否以 JSON 形式附加元数据 header。 +- `--mode`：seed 模式： + - `burst`：一次性快速发送完 `count` 条。 + - `interval`：按照给定时间间隔持续发送。 + - `mixed`：先按 interval 发送一部分，再以 burst 方式发送剩余。 +- `--interval-min` / `--interval-max`：`interval`/`mixed` 模式下两条电报之间的最小/最大间隔（默认 `1s` / `2s`）。 +- `--duration`：`interval`/`mixed` 模式下的总持续时间，`0` 表示仅按 `count` 控制停止条件。 + +### 使用示例 + +#### 1. 一次性快速打 100 条（突发流量） + +```bash +go run ./cmd/seed-telegrams \ + --count=100 \ + --mode=burst \ + --category=mixed \ + --status=random +``` + +#### 2. 模拟真实流量：每 1–2 秒发一条，持续 5 分钟 + +```bash +go run ./cmd/seed-telegrams \ + --mode=interval \ + --interval-min=1s \ + --interval-max=2s \ + --duration=5m \ + --status=random \ + --category=mixed +``` + +#### 3. 慢热 + 突发：前半段慢慢发，后半段瞬间打完 + +```bash +go run ./cmd/seed-telegrams \ + --count=200 \ + --mode=mixed \ + --interval-min=500ms \ + --interval-max=1500ms \ + --status=random +``` + +#### 4. 只打印合成电报，不发送（本地调试报文格式） + +```bash +go run ./cmd/seed-telegrams \ + --count=5 \ + --mode=burst \ + --dry-run \ + --category=FPL +``` + +该工具专门为开发与测试设计，不影响生产服务逻辑，推荐在本地或测试环境配合解析与存储流水线一起使用，用于回归测试、吞吐量观察和错误场景演练。 + **Parsed Fields:** - `category`: "FPL" - `flight_number`: Flight number diff --git a/cmd/seed-telegrams/main.go b/cmd/seed-telegrams/main.go index 253866c..bcc973f 100644 --- a/cmd/seed-telegrams/main.go +++ b/cmd/seed-telegrams/main.go @@ -23,6 +23,26 @@ var ( bodyCategories = []string{"ARR", "DEP", "CNL", "DLA", "FPL"} ) +// SeedConfig controls how telegrams are generated and dispatched. +type SeedConfig struct { + Count int + Mode string + IntervalMin time.Duration + IntervalMax time.Duration + Duration time.Duration + CategoryFlag string + StatusFlag string + ErrorReason string + HeaderFormat string + DryRun bool +} + +// publishFunc abstracts the publish side-effect so that it can be swapped in tests. +type publishFunc func(*telegram) error + +// sleepFunc is used instead of time.Sleep so tests can stub out real sleeping. +var sleepFunc = time.Sleep + func main() { natsURL := flag.String("nats-url", "nats://127.0.0.1:4222", "NATS server URL (empty skips publish)") subject := flag.String("subject", "telegram.raw", "Subject to publish telegrams to") @@ -36,6 +56,10 @@ func main() { jsStream := flag.String("stream", "", "JetStream stream (optional when --jetstream)") jsSubject := flag.String("js-subject", "", "Override subject for JetStream publish (defaults to --subject)") headerFormat := flag.String("header-format", "json", "Metadata header encoding: json|none") + mode := flag.String("mode", "burst", "Seed mode: burst|interval|mixed") + intervalMin := flag.Duration("interval-min", time.Second, "Minimum interval between messages in interval/mixed modes") + intervalMax := flag.Duration("interval-max", 2*time.Second, "Maximum interval between messages in interval/mixed modes") + duration := flag.Duration("duration", 0, "Total duration for interval/mixed modes (0 = rely on --count only)") flag.Parse() rand.Seed(time.Now().UnixNano()) @@ -64,62 +88,34 @@ func main() { } } - categories := bodyCategories - if strings.ToLower(*category) != "mixed" { - categories = []string{strings.ToUpper(*category)} + // Normalize and prepare configuration. + if *intervalMax < *intervalMin { + *intervalMax = *intervalMin } - statuses := statusValues - if strings.ToLower(*status) != "random" { - statuses = []string{strings.ToLower(*status)} + cfg := SeedConfig{ + Count: *count, + Mode: strings.ToLower(*mode), + IntervalMin: *intervalMin, + IntervalMax: *intervalMax, + Duration: *duration, + CategoryFlag: *category, + StatusFlag: *status, + ErrorReason: *errorReason, + HeaderFormat: *headerFormat, + DryRun: *dryRun, } - for i := 0; i < *count; i++ { - cat := categories[rand.Intn(len(categories))] - payload, intentionallyInvalid := buildTelegram(cat) + categories := buildCategories(cfg.CategoryFlag, bodyCategories) + statuses := buildStatuses(cfg.StatusFlag, statusValues) - // 当状态为 random 时，根据报文是否合法来倾向选择 parsed 或 body_error - if strings.ToLower(*status) == "random" { - if intentionallyInvalid { - // 故意非法的报文：大概率标记为 body_error - if rand.Intn(100) < 80 { - payload.Status = "body_error" - } else { - payload.Status = statusValues[rand.Intn(len(statusValues))] - } - } else { - // 合法报文：大概率标记为 parsed - if rand.Intn(100) < 70 { - payload.Status = "parsed" - } else { - payload.Status = statusValues[rand.Intn(len(statusValues))] - } - } - } else { - // 非 random 模式下沿用原有逻辑 - payload.Status = statuses[rand.Intn(len(statuses))] - } - - payload.ErrorReason = *errorReason - payload.Metadata = map[string]string{ - "message_id": payload.MessageID, - "category": payload.Category, - "comments": fmt.Sprintf("seeded iteration=%d", i), - "status": payload.Status, - } - - if *dryRun { - blob, _ := json.MarshalIndent(payload, "", " ") - fmt.Println(string(blob)) - fmt.Println("---") - continue - } - - if nc != nil && !*noNATS { + var publisher publishFunc + if nc != nil && !*noNATS { + publisher = func(payload *telegram) error { data := []byte(payload.Content) msg := &nats.Msg{Subject: *subject, Data: data, Header: nats.Header{}} msg.Header.Set("Nats-Msg-Id", payload.UUID) - if strings.ToLower(*headerFormat) == "json" { + if strings.ToLower(cfg.HeaderFormat) == "json" { headerJSON, _ := json.Marshal(payload.Metadata) msg.Header.Set("x-telegram-meta", string(headerJSON)) } @@ -134,16 +130,22 @@ func main() { } msg.Subject = pubSubject if _, err := js.PublishMsg(msg); err != nil { - log.Fatalf("jetstream publish: %v", err) - } - } else { - if err := nc.PublishMsg(msg); err != nil { - log.Fatalf("nats publish: %v", err) + return fmt.Errorf("jetstream publish: %w", err) } + return nil } + + if err := nc.PublishMsg(msg); err != nil { + return fmt.Errorf("nats publish: %w", err) + } + return nil } } + if err := RunSeed(cfg, categories, statuses, publisher); err != nil { + log.Fatalf("run seed: %v", err) + } + if !*dryRun { if nc != nil && !*noNATS { log.Printf("Published %d telegram(s) to %s", *count, *subject) @@ -151,6 +153,160 @@ func main() { } } +// RunSeed drives telegram generation according to the configured mode. +// It delegates message construction to buildTelegram and side-effects to the provided publisher. +func RunSeed(cfg SeedConfig, categories []string, statuses []string, publisher publishFunc) error { + if len(categories) == 0 { + return fmt.Errorf("no categories available") + } + if len(statuses) == 0 { + return fmt.Errorf("no statuses available") + } + + switch cfg.Mode { + case "interval": + return runIntervalMode(cfg, categories, statuses, publisher) + case "mixed": + return runMixedMode(cfg, categories, statuses, publisher) + default: + // Default to burst semantics. + return runBurstMode(cfg, categories, statuses, publisher) + } +} + +func runBurstMode(cfg SeedConfig, categories []string, statuses []string, publisher publishFunc) error { + count := cfg.Count + if count <= 0 { + return nil + } + for i := 0; i < count; i++ { + if err := sendTelegram(i, cfg, categories, statuses, publisher); err != nil { + return err + } + } + return nil +} + +func runIntervalMode(cfg SeedConfig, categories []string, statuses []string, publisher publishFunc) error { + start := time.Now() + sent := 0 + + for { + if cfg.Count > 0 && sent >= cfg.Count { + break + } + if cfg.Duration > 0 && time.Since(start) >= cfg.Duration { + break + } + + if err := sendTelegram(sent, cfg, categories, statuses, publisher); err != nil { + return err + } + sent++ + + // Compute next sleep duration within [IntervalMin, IntervalMax]. + sleepDur := cfg.IntervalMin + if cfg.IntervalMax > cfg.IntervalMin { + delta := cfg.IntervalMax - cfg.IntervalMin + sleepDur = cfg.IntervalMin + time.Duration(rand.Int63n(int64(delta)+1)) + } + if sleepDur > 0 && !cfg.DryRun { + sleepFunc(sleepDur) + } + } + + return nil +} + +// mixed 模式：前半段使用 interval 模式，后半段使用 burst。 +func runMixedMode(cfg SeedConfig, categories []string, statuses []string, publisher publishFunc) error { + // 简单策略：如果 Count>0，前半部分 interval，后半部分 burst；否则退化为 interval。 + if cfg.Count <= 0 { + return runIntervalMode(cfg, categories, statuses, publisher) + } + + half := cfg.Count / 2 + if half == 0 { + // Count==1 时直接按 burst 处理。 + return runBurstMode(cfg, categories, statuses, publisher) + } + + intervalCfg := cfg + intervalCfg.Count = half + if err := runIntervalMode(intervalCfg, categories, statuses, publisher); err != nil { + return err + } + + burstCfg := cfg + burstCfg.Count = cfg.Count - half + return runBurstMode(burstCfg, categories, statuses, publisher) +} + +// sendTelegram builds a single telegram, assigns status/metadata, and either prints or publishes it. +func sendTelegram(iteration int, cfg SeedConfig, categories []string, statuses []string, publisher publishFunc) error { + cat := categories[rand.Intn(len(categories))] + payload, intentionallyInvalid := buildTelegram(cat) + + payload.Status = chooseStatus(intentionallyInvalid, cfg.StatusFlag, statuses) + payload.ErrorReason = cfg.ErrorReason + payload.Metadata = map[string]string{ + "message_id": payload.MessageID, + "category": payload.Category, + "comments": fmt.Sprintf("seeded iteration=%d", iteration), + "status": payload.Status, + } + + if cfg.DryRun { + blob, _ := json.MarshalIndent(payload, "", " ") + fmt.Println(string(blob)) + fmt.Println("---") + return nil + } + + if publisher != nil { + return publisher(payload) + } + return nil +} + +// buildCategories returns the effective categories list based on the CLI flag. +func buildCategories(flagValue string, all []string) []string { + if strings.ToLower(flagValue) == "mixed" || flagValue == "" { + return all + } + return []string{strings.ToUpper(flagValue)} +} + +// buildStatuses returns the effective statuses list based on the CLI flag. +func buildStatuses(flagValue string, all []string) []string { + if strings.ToLower(flagValue) == "random" || flagValue == "" { + return all + } + return []string{strings.ToLower(flagValue)} +} + +// chooseStatus encapsulates the status selection logic, including the special \"random\" behaviour. +func chooseStatus(intentionallyInvalid bool, statusFlag string, statuses []string) string { + if strings.ToLower(statusFlag) != "random" { + return statuses[rand.Intn(len(statuses))] + } + + // random 模式：根据报文是否合法，对 parsed/body_error 做倾向性选择。 + if intentionallyInvalid { + // 非法报文：大概率 body_error。 + if rand.Intn(100) < 80 { + return "body_error" + } + } else { + // 合法报文：大概率 parsed。 + if rand.Intn(100) < 70 { + return "parsed" + } + } + + return statuses[rand.Intn(len(statuses))] +} + type telegram struct { UUID string `json:"uuid"` MessageID string `json:"message_id"` diff --git a/cmd/seed-telegrams/main_test.go b/cmd/seed-telegrams/main_test.go new file mode 100644 index 0000000..d9f960f --- /dev/null +++ b/cmd/seed-telegrams/main_test.go @@ -0,0 +1,211 @@ +package main + +import ( + "testing" + "time" +) + +// Test buildBody generates syntactically plausible bodies for each category. +func TestBuildBodyFormats(t *testing.T) { + tests := []struct { + category string + prefix string + suffix string + }{ + {"ARR", "(ARR-", ")"}, + {"DEP", "(DEP-", ")"}, + {"CNL", "(CNL-", ")"}, + {"DLA", "(DLA-", ")"}, + {"FPL", "(FPL-", ")"}, + } + + for _, tt := range tests { + body, invalid := buildBody(tt.category) + if !invalid && body == "" { + t.Fatalf("category %s: expected non-empty body", tt.category) + } + if body[0:len(tt.prefix)] != tt.prefix { + t.Fatalf("category %s: expected prefix %q, got %q", tt.category, tt.prefix, body[0:len(tt.prefix)]) + } + if body[len(body)-len(tt.suffix):] != tt.suffix { + t.Fatalf("category %s: expected suffix %q, got %q", tt.category, tt.suffix, body[len(body)-len(tt.suffix):]) + } + } +} + +// Test buildTelegram wires header and body into a full telegram. +func TestBuildTelegramBasicFields(t *testing.T) { + tg, _ := buildTelegram("ARR") + if tg.Category != "ARR" { + t.Fatalf("expected category ARR, got %s", tg.Category) + } + if tg.MessageID == "" { + t.Fatalf("expected non-empty MessageID") + } + if tg.UUID == "" { + t.Fatalf("expected non-empty UUID") + } + if tg.Content == "" { + t.Fatalf("expected non-empty Content") + } + if tg.ReceivedAt.IsZero() { + t.Fatalf("expected ReceivedAt to be set") + } + if tg.Metadata != nil { + t.Fatalf("expected Metadata to be nil from buildTelegram") + } +} + +func TestBuildCategories(t *testing.T) { + all := []string{"ARR", "DEP", "CNL", "DLA", "FPL"} + + got := buildCategories("mixed", all) + if len(got) != len(all) { + t.Fatalf("expected all categories for mixed, got %d", len(got)) + } + + got = buildCategories("", all) + if len(got) != len(all) { + t.Fatalf("expected all categories for empty flag, got %d", len(got)) + } + + got = buildCategories("arr", all) + if len(got) != 1 || got[0] != "ARR" { + t.Fatalf("expected single category ARR, got %#v", got) + } +} + +func TestBuildStatuses(t *testing.T) { + all := []string{"parsed", "header_error", "body_error"} + + got := buildStatuses("random", all) + if len(got) != len(all) { + t.Fatalf("expected all statuses for random, got %d", len(got)) + } + + got = buildStatuses("", all) + if len(got) != len(all) { + t.Fatalf("expected all statuses for empty flag, got %d", len(got)) + } + + got = buildStatuses("parsed", all) + if len(got) != 1 || got[0] != "parsed" { + t.Fatalf("expected single status parsed, got %#v", got) + } +} + +func TestChooseStatusNonRandomUsesProvidedList(t *testing.T) { + statuses := []string{"parsed"} + for i := 0; i < 10; i++ { + got := chooseStatus(false, "parsed", statuses) + if got != "parsed" { + t.Fatalf("expected parsed, got %s", got) + } + } +} + +func TestChooseStatusRandomOnlyReturnsKnownStatuses(t *testing.T) { + statuses := []string{"parsed", "header_error", "body_error"} + + for i := 0; i < 100; i++ { + got := chooseStatus(false, "random", statuses) + if !contains(statuses, got) && got != "body_error" && got != "parsed" { + t.Fatalf("unexpected status from chooseStatus: %s", got) + } + } +} + +func contains(list []string, v string) bool { + for _, s := range list { + if s == v { + return true + } + } + return false +} + +// Test RunSeed burst mode publishes Count messages. +func TestRunSeedBurstPublishesCount(t *testing.T) { + cfg := SeedConfig{ + Count: 5, + Mode: "burst", + DryRun: false, + } + categories := []string{"ARR"} + statuses := []string{"parsed"} + + var published int + pub := func(*telegram) error { + published++ + return nil + } + + if err := RunSeed(cfg, categories, statuses, pub); err != nil { + t.Fatalf("RunSeed burst returned error: %v", err) + } + if published != cfg.Count { + t.Fatalf("expected %d published messages, got %d", cfg.Count, published) + } +} + +// Test RunSeed interval mode respects Count and does not sleep in DryRun. +func TestRunSeedIntervalRespectsCount(t *testing.T) { + origSleep := sleepFunc + defer func() { sleepFunc = origSleep }() + sleepFunc = func(time.Duration) {} + + cfg := SeedConfig{ + Count: 3, + Mode: "interval", + IntervalMin: 10 * time.Millisecond, + IntervalMax: 20 * time.Millisecond, + DryRun: true, + } + categories := []string{"ARR"} + statuses := []string{"parsed"} + + var published int + pub := func(*telegram) error { + published++ + return nil + } + + if err := RunSeed(cfg, categories, statuses, pub); err != nil { + t.Fatalf("RunSeed interval returned error: %v", err) + } + if published != cfg.Count { + t.Fatalf("expected %d published messages, got %d", cfg.Count, published) + } +} + +// Test RunSeed mixed mode still results in exactly Count messages when Count>0. +func TestRunSeedMixedPublishesCount(t *testing.T) { + origSleep := sleepFunc + defer func() { sleepFunc = origSleep }() + sleepFunc = func(time.Duration) {} + + cfg := SeedConfig{ + Count: 6, + Mode: "mixed", + IntervalMin: 1 * time.Millisecond, + IntervalMax: 2 * time.Millisecond, + DryRun: true, + } + categories := []string{"ARR"} + statuses := []string{"parsed"} + + var published int + pub := func(*telegram) error { + published++ + return nil + } + + if err := RunSeed(cfg, categories, statuses, pub); err != nil { + t.Fatalf("RunSeed mixed returned error: %v", err) + } + if published != cfg.Count { + t.Fatalf("expected %d published messages, got %d", cfg.Count, published) + } +} + + diff --git a/docs/dev-guide.md b/docs/dev-guide.md index b40b28f..49a1b51 100644 --- a/docs/dev-guide.md +++ b/docs/dev-guide.md @@ -31,6 +31,25 @@ Stop and clean the stack when finished: docker compose -f docker-compose.dev.yml down -v ``` +In development (`GO_ENV=dev` or unset), if you run the application while +stopping and recreating the NATS/JetStream containers (for example: + +```bash +docker compose -f docker-compose.dev.yml down -v +docker compose -f docker-compose.dev.yml up -d postgres nats nats-box +``` + +), the JetStream state will be reset. The processor behaves as follows: + +- The NATS client keeps retrying the connection and automatically reconnects + when NATS is back. +- The JetStream consumer detects missing streams/consumers and, in dev/test + environments, uses shared `EnsureStream`/`ensureConsumer` logic to + auto-recreate them. +- In production environments, missing streams/consumers are treated as + configuration/operational errors and are not auto-recreated; operators + should investigate and fix the underlying issue. + ### Using Taskfile shortcuts The `Taskfile.yml` includes helper targets that wrap the commands above: diff --git a/docs/reliability.md b/docs/reliability.md index 05fab68..6d76b96 100644 --- a/docs/reliability.md +++ b/docs/reliability.md @@ -101,6 +101,27 @@ Recommended pattern: - Use a backoff array such as `[5s, 30s, 2m]`. - Treat messages that still fail after `max_deliver` as candidates for DLQ, via the permanent error/poison message path where applicable. +### JetStream Availability and Auto-Recovery (Dev vs Prod) + +- When the JetStream API is temporarily unavailable (for example, NATS has just + restarted and returns `ErrNoResponders`), the consumer uses an exponential + backoff when retrying `Fetch` calls (roughly `1s, 2s, 4s, ...` up to + around `30s`) to avoid log spam while allowing the system to recover. +- In dev/test environments, if the stream or consumer is detected as missing at + runtime (for example after `docker compose down -v`), the consumer calls the + shared `EnsureStream` and `ensureConsumer` logic to recreate them and + re-establish subscriptions. +- In production environments, missing streams/consumers are treated as + configuration or operational errors: + - They are **not** auto-recreated. + - Errors are logged prominently so operators can diagnose and fix the issue. +- On the publishing side, JetStream `ErrNoResponders` and similar errors are + treated as temporary by the processor: + - Such errors cause the consumer to NAK messages and rely on the configured + backoff for retries. + - Permanent configuration/permission errors remain mapped to permanent + failures and follow the DLQ + ACK flow. + ### Alerts and Dashboards Prometheus alert suggestions: diff --git a/internal/app/processor.go b/internal/app/processor.go index 6b25fe5..60da066 100644 --- a/internal/app/processor.go +++ b/internal/app/processor.go @@ -169,7 +169,17 @@ func (p *MessageProcessor) Handle(ctx context.Context, raw []byte, msgID string) p.telemetry.RecordFailure("publisher") p.telemetry.RecordProcessingResult(ctx, string(parsed.Status), parsed.Category, latency) p.persistRaw(ctx, parsed) - // Mark as permanent so the consumer will ack instead of retrying + + // Treat clearly temporary JetStream issues (e.g. no responders) as transient so + // the consumer will NAK and retry according to backoff settings. + lowerErr := strings.ToLower(err.Error()) + if strings.Contains(lowerErr, "no responders") { + pubSpan.End() + // Return a non-permanent error to trigger retry via nakWithStrategy in the consumer. + return fmt.Errorf("transient publish error: %w", err) + } + + // Other publish errors are treated as permanent and will go to DLQ + ACK. pubSpan.End() return Permanent(fmt.Errorf("failed to publish message: %w", err)) } diff --git a/internal/infra/nats/consumer.go b/internal/infra/nats/consumer.go index b9d16ad..396df22 100644 --- a/internal/infra/nats/consumer.go +++ b/internal/infra/nats/consumer.go @@ -10,6 +10,7 @@ import ( "encoding/json" "errors" "fmt" + "os" "strings" "time" @@ -49,6 +50,32 @@ type Consumer struct { consecutiveProcessErrors int } +func isDevLikeEnv() bool { + switch strings.ToLower(os.Getenv("GO_ENV")) { + case "", "dev", "development", "test", "testing": + return true + default: + return false + } +} + +func isJetStreamResourceNotFound(err error) bool { + if err == nil { + return false + } + if errors.Is(err, nats.ErrStreamNotFound) || errors.Is(err, nats.ErrConsumerNotFound) { + return true + } + + // Some JetStream API errors are only exposed via error strings. + msg := strings.ToLower(err.Error()) + if strings.Contains(msg, "stream not found") || strings.Contains(msg, "consumer not found") { + return true + } + + return false +} + // ProvideConsumer creates a NATS consumer func ProvideConsumer( conn *nats.Conn, @@ -201,6 +228,58 @@ func (c *Consumer) ensureConsumer() error { return nil } +// recoverJetStreamResources attempts to recreate the stream and consumer in +// dev/test environments if they are missing. It is safe to call multiple times. +func (c *Consumer) recoverJetStreamResources() error { + if c.js == nil { + return fmt.Errorf("jetstream context is nil") + } + if c.cfg == nil { + return fmt.Errorf("config is nil") + } + + // Ensure stream exists (dev/test may auto-create, prod will error). + if err := EnsureStream(c.js, c.cfg, c.logger); err != nil { + return fmt.Errorf("ensure stream %s: %w", c.streamName, err) + } + + // Ensure durable consumer exists and is properly bound. + if err := c.ensureConsumer(); err != nil { + return fmt.Errorf("ensure consumer %s: %w", c.consumerName, err) + } + + return nil +} + +// createPullSubscriptionWithRecovery creates a pull subscription and, in +// dev/test environments, attempts to self-heal missing stream/consumer +// by recreating them once. +func (c *Consumer) createPullSubscriptionWithRecovery() (*nats.Subscription, error) { + sub, err := c.js.PullSubscribe(c.subject, c.consumerName, nats.Bind(c.streamName, c.consumerName)) + if err == nil { + return sub, nil + } + + if isJetStreamResourceNotFound(err) && isDevLikeEnv() && shouldBootstrapStream() { + c.logger.Warn("PullSubscribe failed due to missing JetStream resources; attempting to recreate", + zap.Error(err), + zap.String("stream", c.streamName), + zap.String("consumer", c.consumerName), + ) + if recErr := c.recoverJetStreamResources(); recErr != nil { + return nil, fmt.Errorf("failed to recover JetStream resources: %w", recErr) + } + // Retry subscription after successful recovery. + sub, err = c.js.PullSubscribe(c.subject, c.consumerName, nats.Bind(c.streamName, c.consumerName)) + if err != nil { + return nil, fmt.Errorf("failed to create pull subscription after recovery: %w", err) + } + return sub, nil + } + + return nil, fmt.Errorf("failed to create pull subscription: %w", err) +} + // validateDLQ verifies whether DLQ routing should be enabled and, if so, whether // the configured DLQ subject is bound to a JetStream stream. If validation fails, // DLQ routing is disabled (by clearing c.dlqSubject) and a warning is logged, @@ -269,10 +348,10 @@ func (c *Consumer) Start(ctx context.Context) error { } func (c *Consumer) startJetStream(ctx context.Context) error { - // Create pull subscription - sub, err := c.js.PullSubscribe(c.subject, c.consumerName, nats.Bind(c.streamName, c.consumerName)) + // Create pull subscription (with simple self-healing in dev/test). + sub, err := c.createPullSubscriptionWithRecovery() if err != nil { - return fmt.Errorf("failed to create pull subscription: %w", err) + return err } defer sub.Unsubscribe() @@ -296,6 +375,8 @@ func (c *Consumer) startJetStream(ctx context.Context) error { defer statsCancel() go c.emitConsumerStats(statsCtx) + var fetchErrorStreak int + for { select { case <-ctx.Done(): @@ -308,25 +389,80 @@ func (c *Consumer) startJetStream(ctx context.Context) error { msgs, err := sub.Fetch(c.batchSize, nats.MaxWait(c.batchTimeout)) if err != nil { if errors.Is(err, nats.ErrTimeout) { - // Timeout is expected when no messages are available + // Timeout is expected when no messages are available. continue } + + // JetStream API is currently unavailable (e.g., NATS just restarted or JetStream not ready). if errors.Is(err, nats.ErrNoResponders) { - // JetStream API is currently unavailable (e.g., NATS just restarted or JetStream not ready). - // Back off a bit to avoid log spam while allowing the system to recover. - c.logger.Warn("JetStream not available, will retry", + fetchErrorStreak++ + backoff := time.Duration(fetchErrorStreak) * time.Second + if backoff > 30*time.Second { + backoff = 30 * time.Second + } + c.logger.Warn("JetStream not available, will retry with backoff", + zap.Error(err), + zap.String("stream", c.streamName), + zap.String("consumer", c.consumerName), + zap.Duration("backoff", backoff), + ) + time.Sleep(backoff) + continue + } + + // Underlying consumer/stream removed while app is running. + if isJetStreamResourceNotFound(err) { + if isDevLikeEnv() && shouldBootstrapStream() { + c.logger.Warn("JetStream consumer or stream missing; attempting to recreate", + zap.Error(err), + zap.String("stream", c.streamName), + zap.String("consumer", c.consumerName), + ) + if recErr := c.recoverJetStreamResources(); recErr != nil { + c.logger.Error("Failed to recover JetStream resources", zap.Error(recErr)) + return recErr + } + + // Recreate subscription after successful recovery. + sub.Unsubscribe() + sub, err = c.createPullSubscriptionWithRecovery() + if err != nil { + return err + } + + // Reset error streak after successful recovery. + fetchErrorStreak = 0 + continue + } + + // Production: treat as configuration/operational error. + c.logger.Error("JetStream consumer or stream missing; not auto-recreating in this environment", zap.Error(err), zap.String("stream", c.streamName), zap.String("consumer", c.consumerName), ) - time.Sleep(5 * time.Second) - continue + return err } - c.logger.Error("Failed to fetch messages", zap.Error(err)) - time.Sleep(time.Second) + + // Generic error path with modest backoff. + fetchErrorStreak++ + backoff := time.Duration(fetchErrorStreak) * time.Second + if backoff > 10*time.Second { + backoff = 10 * time.Second + } + c.logger.Error("Failed to fetch messages; backing off", + zap.Error(err), + zap.Duration("backoff", backoff), + ) + time.Sleep(backoff) continue } + // Successful fetch -> reset error streak. + if fetchErrorStreak > 0 { + fetchErrorStreak = 0 + } + // Process each message // TODO: consider buffering messages to take advantage of Repository.InsertBatch for higher throughput. for _, msg := range msgs { diff --git a/internal/infra/nats/jetstream.go b/internal/infra/nats/jetstream.go index 0e732a2..ad4acf5 100644 --- a/internal/infra/nats/jetstream.go +++ b/internal/infra/nats/jetstream.go @@ -45,15 +45,29 @@ func ProvideJetStream(nc *nats.Conn, cfg *config.Config, logger *zap.Logger) (na return nil, fmt.Errorf("failed to get JetStream context: %w", err) } - // Create stream if it doesn't exist + // Ensure the stream exists and is minimally aligned with configuration. + if err := EnsureStream(js, cfg, logger); err != nil { + nc.Close() + return nil, err + } + + return js, nil +} + +// EnsureStream ensures that the configured JetStream stream exists and has +// at least the expected subjects bound. It is safe to call multiple times. +// +// In dev/test environments (see shouldBootstrapStream), the stream will be +// auto-created if it does not exist. In production, a missing stream results +// in an error so that operators can intervene. +func EnsureStream(js nats.JetStreamContext, cfg *config.Config, logger *zap.Logger) error { streamName := cfg.NATS.Stream consumerSubject := cfg.EffectiveSubscriptionTopic() publisherSubject := strings.TrimSpace(cfg.Publisher.Topic) streamSubjects := dedupeSubjects([]string{consumerSubject, publisherSubject}) if len(streamSubjects) == 0 { - nc.Close() - return nil, fmt.Errorf("no subjects configured for JetStream stream %s", streamName) + return fmt.Errorf("no subjects configured for JetStream stream %s", streamName) } streamLimits := cfg.NATS.StreamLimits @@ -87,26 +101,22 @@ func ProvideJetStream(nc *nats.Conn, cfg *config.Config, logger *zap.Logger) (na if errors.Is(err, nats.ErrStreamNotFound) { if shouldBootstrapStream() { if _, err = js.AddStream(streamConfig); err != nil { - nc.Close() - return nil, fmt.Errorf("failed to create stream: %w", err) + return fmt.Errorf("failed to create stream %s: %w", streamName, err) } - logger.Info("Created JetStream", + logger.Info("Created JetStream stream", zap.String("stream", streamName), zap.Strings("subjects", streamSubjects), ) - } else { - nc.Close() - return nil, fmt.Errorf("stream %s not found and auto-creation disabled", streamName) + return nil } - } else { - nc.Close() - return nil, fmt.Errorf("failed to fetch stream info: %w", err) + return fmt.Errorf("stream %s not found and auto-creation disabled", streamName) } - } else { - validateStreamConfig(info, streamSubjects, logger) + return fmt.Errorf("failed to fetch stream info for %s: %w", streamName, err) } - return js, nil + // Stream exists: validate subjects but do not fail hard if they differ. + validateStreamConfig(info, streamSubjects, logger) + return nil } func shouldBootstrapStream() bool { diff --git a/internal/infra/nats/publisher.go b/internal/infra/nats/publisher.go index 89b4ee7..99e3955 100644 --- a/internal/infra/nats/publisher.go +++ b/internal/infra/nats/publisher.go @@ -5,7 +5,9 @@ import ( "caatsm/internal/infra/config" "caatsm/internal/model" "encoding/json" + "errors" "fmt" + "github.com/google/uuid" "github.com/nats-io/nats.go" "go.uber.org/zap" @@ -62,6 +64,10 @@ func (p *Publisher) Publish(message interface{}) error { // Publish to JetStream _, err = p.js.PublishMsg(jsMsg) if err != nil { + // Distinguish temporary JetStream unavailability from permanent config errors. + if errors.Is(err, nats.ErrNoResponders) { + return fmt.Errorf("transient publish error (no responders): %w", err) + } return fmt.Errorf("failed to publish message: %w", err) }