weichen/Agent-System
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix(server): gate parse enqueue to store-backed pipeline

Browse Source 
Co-authored-by: Cursor <cursoragent@cursor.com>
This commit is contained in:
weichen
2026-09-21 09:47:06 +00:00
co-authored by Cursor
parent 63f94ff23b
commit 5f9fbb9923
2 changed files with 10 additions and 4 deletions
Download Patch File Download Diff File Expand all files Collapse all files
internal/materialindex/pipeline.go
+7 -1
View File
@@ -125,6 +125,12 @@ func (p *Pipeline) Enabled() bool {
return p != nil && p.client != nil && p.client.Enabled()
}
// HasStore reports whether parse state persists (PostgreSQL available).
// Without a store, async parse would hit the live API with no state machine.
func (p *Pipeline) HasStore() bool {
return p != nil && p.pool != nil
}
// FulltextMaxBytes returns the hybrid injection threshold.
func (p *Pipeline) FulltextMaxBytes() int {
if p == nil || p.fullMax <= 0 {
@@ -267,7 +273,7 @@ func (p *Pipeline) EnsureParsed(ctx context.Context, fileName string, blobSHA st
// EnqueueAsync starts background parse without blocking the caller.
func (p *Pipeline) EnqueueAsync(fileName string, blobSHA string, data []byte, projectID int64) {
if p == nil || !p.Enabled() || !mineru.SupportsExt(fileName) || blobSHA == "" || len(data) == 0 {
if p == nil || !p.Enabled() || !p.HasStore() || !mineru.SupportsExt(fileName) || blobSHA == "" || len(data) == 0 {
return
}
go func() {
internal/server/asset_handlers.go
+3 -3
View File
@@ -84,7 +84,7 @@ func (s *Server) handleUploadProjectAsset(w http.ResponseWriter, r *http.Request
}
parseState := ""
if matIdx := bizdev.Snapshot(s.svc).MaterialIndex; matIdx != nil && matIdx.Enabled() {
if matIdx := bizdev.Snapshot(s.svc).MaterialIndex; matIdx != nil && matIdx.Enabled() && matIdx.HasStore() {
if mineru.SupportsExt(ref.FileName) {
matIdx.EnqueueAsync(ref.FileName, ref.BlobSHA256, data, projectID)
// Report the real state: same-content uploads hit an existing parse
@@ -141,7 +141,7 @@ func (s *Server) handleReparseProjectAsset(w http.ResponseWriter, r *http.Reques
return
}
matIdx := bizdev.Snapshot(s.svc).MaterialIndex
if matIdx == nil || !matIdx.Enabled() {
if matIdx == nil || !matIdx.Enabled() || !matIdx.HasStore() {
writeSafeError(w, r, http.StatusServiceUnavailable, "parse engine unavailable")
return
}
@@ -225,7 +225,7 @@ func (s *Server) handleListProjectAssets(w http.ResponseWriter, r *http.Request)
// 内容已入库但从未触发解析（如早于 MinerU 接入的历史资产）：
// 列表读取时自动补触发，前端轮询即可看到 pending→running→done。
seenSHA[ref.BlobSHA256] = true
if matIdx := bizdev.Snapshot(s.svc).MaterialIndex; matIdx != nil && matIdx.Enabled() && mineru.SupportsExt(ref.FileName) {
if matIdx := bizdev.Snapshot(s.svc).MaterialIndex; matIdx != nil && matIdx.Enabled() && matIdx.HasStore() && mineru.SupportsExt(ref.FileName) {
if data, _, err := cas.ReadRefBytes(r.Context(), ref.ID); err == nil {
matIdx.EnqueueAsync(ref.FileName, ref.BlobSHA256, data, projectID)
item.ParseState = "pending"