Blizzard
6f16062dfc
把入库从网关裸 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>
41 lines
2.2 KiB
Go
41 lines
2.2 KiB
Go
package contract
|
|
|
|
import "encoding/json"
|
|
|
|
// 入库工作队列约定（准生产级）。把"解析→切块→向量化→图谱"从网关的裸 goroutine 升级为
|
|
// JetStream 持久工作队列：崩溃可重投、有界并发背压、削峰不 OOM。与任务流(SUNDYNIX_TASKS)同级。
|
|
const (
|
|
StreamIngest = "SUNDYNIX_INGEST" // 入库作业 JetStream 流（持久，作业不因网关重启而丢）
|
|
SubjectIngest = "sundynix.ingest" // 入库作业发布前缀；实际 sundynix.ingest.<job_id>
|
|
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
|
|
}
|