feat(bus): 入库 JetStream 持久工作队列 —— 崩溃重投/背压/幂等

把入库从网关裸 goroutine 升级为和任务流同级的 JetStream 持久工作队列： - contract/ingest.go: IngestJob(claim-check 引用) + 流/消费者常量 - bus.go: EnsureIngestStream / PublishIngestJob / ConsumeIngestJobs (durable consumer + MaxAckPending 背压 + AckWait 可配(env) + MaxDeliver 4 毒消息兜底 + lastAttempt 终态收尾 + 优雅 drain) Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
2026-06-30 13:37:24 +08:00
parent ce70388e52
commit 6f16062dfc
2 changed files with 178 additions and 0 deletions
@@ -690,6 +690,144 @@ func (b *Bus) ConsumeTasks(ctx context.Context, h TaskHandler) (drain func(conte
 	}, nil
 }

+// ---- 入库工作队列（JetStream 持久，与任务流同级）----
+
+// IngestHandler 处理一条入库作业；lastAttempt=true 表示这是最后一次投递（重投已耗尽），
+// handler 应据此做终态收尾（不要再要求重试）。返回非 nil 错误 → 延迟重投（瞬时失败可恢复）。
+type IngestHandler func(ctx context.Context, job *contract.IngestJob, lastAttempt bool) error
+
+// ingestMaxDeliver 是单条入库作业的最大投递次数（含首发）。瞬时失败重投兜底，
+// 到第 N 次时 handler 收到 lastAttempt=true 做终态收尾，避免毒消息无限重投。
+const ingestMaxDeliver = 4
+
+// ingestConcurrency 返回单实例入库并发上限（env INGEST_CONCURRENCY，默认 4）。
+// 每篇入库内部还会并发 embedding/图谱，故此值不宜过大，避免打爆 provider 速率。
+func ingestConcurrency() int {
+	if v := os.Getenv("INGEST_CONCURRENCY"); v != "" {
+		if n, err := strconv.Atoi(v); err == nil && n > 0 {
+			return n
+		}
+	}
+	return 4
+}
+
+// ingestAckWait 是入库作业未 ack 的最长时长（env INGEST_ACKWAIT_SEC，默认 1800s=30min）。
+// 须覆盖单篇最长入库时长；崩溃在途作业在此时长后由 JetStream 重投兜底（值越小恢复越快、但越易把慢作业误判重投）。
+func ingestAckWait() time.Duration {
+	if v := os.Getenv("INGEST_ACKWAIT_SEC"); v != "" {
+		if n, err := strconv.Atoi(v); err == nil && n > 0 {
+			return time.Duration(n) * time.Second
+		}
+	}
+	return 30 * time.Minute
+}
+
+// EnsureIngestStream 幂等地创建/更新入库作业流，持久捕获 sundynix.ingest.>。
+func (b *Bus) EnsureIngestStream(ctx context.Context) error {
+	_, err := b.js.CreateOrUpdateStream(ctx, jetstream.StreamConfig{
+		Name:     contract.StreamIngest,
+		Subjects: []string{contract.SubjectIngestAll},
+		Storage:  jetstream.FileStorage,
+		MaxAge:   ingestStreamMaxAge,
+	})
+	return err
+}
+
+// ingestStreamMaxAge 限制入库流里消息的保留时长（已 ack 的会被清；未消费的超时丢弃），
+// 防止异常堆积无限增长。30 天足以覆盖任何正常重投窗口。
+const ingestStreamMaxAge = 30 * 24 * time.Hour
+
+// PublishIngestJob 把入库作业发布到 sundynix.ingest.<job>，持久入队（崩溃不丢）。
+func (b *Bus) PublishIngestJob(ctx context.Context, job *contract.IngestJob) error {
+	data, err := job.Marshal()
+	if err != nil {
+		return err
+	}
+	ctx, span := tracer().Start(ctx, "nats.publish ingest",
+		trace.WithSpanKind(trace.SpanKindProducer),
+		trace.WithAttributes(
+			attribute.String("messaging.system", "nats"),
+			attribute.String("messaging.destination.name", contract.IngestSubject(job.JobID)),
+		))
+	defer span.End()
+	msg := nats.NewMsg(contract.IngestSubject(job.JobID))
+	msg.Data = data
+	injectTrace(ctx, msg.Header)
+	if _, err := b.js.PublishMsg(ctx, msg); err != nil {
+		span.RecordError(err)
+		return fmt.Errorf("publish ingest job: %w", err)
+	}
+	return nil
+}
+
+// ConsumeIngestJobs 在持久消费者上消费入库作业，队列组内多副本负载均衡。
+// 每条作业派到独立 worker goroutine；并发由信号量 + MaxAckPending 双重约束（背压，削峰不 OOM）。
+// 失败延迟重投、网关崩溃由 AckWait 兜底重投——入库幂等（按 doc 先删后写），重跑安全。
+func (b *Bus) ConsumeIngestJobs(ctx context.Context, h IngestHandler) (drain func(context.Context), err error) {
+	concurrency := ingestConcurrency()
+	cons, err := b.js.CreateOrUpdateConsumer(ctx, contract.StreamIngest, jetstream.ConsumerConfig{
+		Durable:       contract.ConsumerIngest,
+		AckPolicy:     jetstream.AckExplicitPolicy,
+		FilterSubject: contract.SubjectIngestAll,
+		// 一篇几十万字含切块+上千次 embedding+至多 60 次图谱抽取，可达数分钟；
+		// AckWait 须覆盖单篇最长入库时长，否则在途未 ack 会被重投成重复入库。
+		AckWait:       ingestAckWait(),
+		MaxAckPending: concurrency, // 背压：服务端不下发超过本节点同时能处理的量
+		MaxDeliver:    ingestMaxDeliver, // 毒消息兜底：重投上限，避免坏作业无限循环
+	})
+	if err != nil {
+		return nil, fmt.Errorf("create ingest consumer: %w", err)
+	}
+	sem := make(chan struct{}, concurrency)
+	var wg sync.WaitGroup
+	cc, err := cons.Consume(func(msg jetstream.Msg) {
+		job, err := contract.UnmarshalIngestJob(msg.Data())
+		if err != nil {
+			_ = msg.Term() // 脏数据，丢弃不重投
+			return
+		}
+		select {
+		case sem <- struct{}{}:
+		case <-ctx.Done():
+			return
+		}
+		wg.Add(1)
+		go func() {
+			defer func() {
+				<-sem
+				wg.Done()
+				if r := recover(); r != nil {
+					log.Printf("[bus] ingest %s handler panic: %v", job.JobID, r)
+					_ = msg.Term() // panic 丢弃不重投，避免崩溃循环
+				}
+			}()
+			lastAttempt := true // 取不到投递元数据时按"最后一次"处理，宁可不重试也不死循环
+			if meta, merr := msg.Metadata(); merr == nil {
+				lastAttempt = meta.NumDelivered >= ingestMaxDeliver
+			}
+			// handler ctx 派生自 Background：停止消费不掐断在途入库，由 drain 等完 / AckWait 兜底。
+			mctx := extractTrace(context.Background(), nats.Header(msg.Headers()))
+			mctx, span := tracer().Start(mctx, "nats.consume ingest",
+				trace.WithSpanKind(trace.SpanKindConsumer),
+				trace.WithAttributes(attribute.String("sundynix.ingest_job", job.JobID)))
+			defer span.End()
+			if herr := h(mctx, job, lastAttempt); herr != nil {
+				span.RecordError(herr)
+				_ = msg.NakWithDelay(2 * time.Second) // 瞬时失败延迟重投（至多 ingestMaxDeliver 次）
+				return
+			}
+			_ = msg.Ack()
+		}()
+	})
+	if err != nil {
+		return nil, fmt.Errorf("consume ingest: %w", err)
+	}
+	return func(dctx context.Context) {
+		cc.Stop()
+		drainWait(&wg, dctx)
+	}, nil
+}
+
 // ---- JetStream KV：临时持久态（dispatcher compose checkpoint 等）----

 // KVHandle 是一个 JetStream KV 桶的薄封装，把 NATS 细节（ErrKeyNotFound 等）挡在 bus 内，