From 6f16062dfc7685ff5bbe058e3276836a04be2516 Mon Sep 17 00:00:00 2001 From: Blizzard Date: Tue, 30 Jun 2026 13:37:24 +0800 Subject: [PATCH] =?UTF-8?q?feat(bus):=20=E5=85=A5=E5=BA=93=20JetStream=20?= =?UTF-8?q?=E6=8C=81=E4=B9=85=E5=B7=A5=E4=BD=9C=E9=98=9F=E5=88=97=20?= =?UTF-8?q?=E2=80=94=E2=80=94=20=E5=B4=A9=E6=BA=83=E9=87=8D=E6=8A=95/?= =?UTF-8?q?=E8=83=8C=E5=8E=8B/=E5=B9=82=E7=AD=89?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 把入库从网关裸 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 --- sundynix-shared/bus/bus.go | 138 +++++++++++++++++++++++++++++ sundynix-shared/contract/ingest.go | 40 +++++++++ 2 files changed, 178 insertions(+) create mode 100644 sundynix-shared/contract/ingest.go diff --git a/sundynix-shared/bus/bus.go b/sundynix-shared/bus/bus.go index 9c22aaf..732107b 100644 --- a/sundynix-shared/bus/bus.go +++ b/sundynix-shared/bus/bus.go @@ -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.，持久入队（崩溃不丢）。 +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 内， diff --git a/sundynix-shared/contract/ingest.go b/sundynix-shared/contract/ingest.go new file mode 100644 index 0000000..a585ed2 --- /dev/null +++ b/sundynix-shared/contract/ingest.go @@ -0,0 +1,40 @@ +package contract + +import "encoding/json" + +// 入库工作队列约定（准生产级）。把"解析→切块→向量化→图谱"从网关的裸 goroutine 升级为 +// JetStream 持久工作队列：崩溃可重投、有界并发背压、削峰不 OOM。与任务流(SUNDYNIX_TASKS)同级。 +const ( + StreamIngest = "SUNDYNIX_INGEST" // 入库作业 JetStream 流（持久，作业不因网关重启而丢） + SubjectIngest = "sundynix.ingest" // 入库作业发布前缀；实际 sundynix.ingest. + SubjectIngestAll = "sundynix.ingest.>" // 流捕获的通配 + ConsumerIngest = "ingest-workers" // 入库 worker 持久消费者（队列组：多副本负载均衡 + 背压） +) + +// IngestSubject 返回某入库作业的发布主题。 +func IngestSubject(jobID string) string { return SubjectIngest + "." + jobID } + +// IngestJob 是一条入库作业（claim-check 模式）：大文件原始字节/正文先暂存到对象存储， +// 作业消息只带「暂存键 + 元数据」这条小消息进队列，worker 消费时再按 StageKey 取回原料。 +// 这样作业消息恒小（不受 JetStream max_payload 限制），且崩溃重投只搬一个引用。 +type IngestJob struct { + JobID string `json:"job_id"` + Owner string `json:"owner"` // 雪花 user.id + KBName string `json:"kb_name"` // 知识库展示名（原文留存分区） + Scoped string `json:"scoped"` // owner/kb 作向量/全文/图谱分区键 + ForceDoc string `json:"force_doc,omitempty"` // 非空=强制文档名（笔记编辑保持身份稳定） + Filename string `json:"filename,omitempty"` // 非空=文件入库（StageKey 指向原始字节，需先解析） + StageKey string `json:"stage_key"` // 暂存对象键：原始文件字节 或 纯文本 + IsText bool `json:"is_text"` // true=StageKey 指向纯文本（跳过解析） +} + +// Marshal/Unmarshal 入库作业（JetStream 消息体）。 +func (j *IngestJob) Marshal() ([]byte, error) { return json.Marshal(j) } + +func UnmarshalIngestJob(data []byte) (*IngestJob, error) { + var j IngestJob + if err := json.Unmarshal(data, &j); err != nil { + return nil, err + } + return &j, nil +}