diff --git a/internal/governance/job_failures.go b/internal/governance/job_failures.go index cc9aaae..553bf7b 100644 --- a/internal/governance/job_failures.go +++ b/internal/governance/job_failures.go @@ -18,6 +18,8 @@ type JobFailure struct { TargetKBID string `json:"target_kb_id,omitempty"` State string `json:"state"` CreatedAt time.Time `json:"created_at"` + DocName string `json:"doc_name,omitempty"` + LastError string `json:"last_error,omitempty"` } type JobFailureFilter struct { diff --git a/internal/platform/postgres/filename_filters.go b/internal/platform/postgres/filename_filters.go index e78fdff..a1837ee 100644 --- a/internal/platform/postgres/filename_filters.go +++ b/internal/platform/postgres/filename_filters.go @@ -158,6 +158,14 @@ WHERE intent.source_connection_id=$1 AND filename_is_filtered(intent.source_conn AND job.payload->>'node_id'=intent.node_id AND job.state IN ('pending','retry','failed')`, connectionID, now); err != nil { return fmt.Errorf("cancel filename-filtered fetch jobs: %w", err) } + // A cancelled or later-succeeded job is no longer waiting on a human: + // retire its dead-letter projection so the DLQ lane shows only real work. + if _, err := tx.Exec(ctx, `UPDATE job_failures f SET state='requeued',resolved_at=$2,resolution_actor_ref='system:filename-filter',resolution_reason='任务已被格式白名单取消' +FROM jobs job +WHERE job.id=f.job_id AND f.state='open' AND job.type='artifact.fetch' + AND job.payload->>'source_connection_id'=$1 AND job.state='cancelled'`, connectionID, now); err != nil { + return fmt.Errorf("retire filename-filtered dead letters: %w", err) + } if _, err := tx.Exec(ctx, `UPDATE catalog_import_intents intent SET fetch_state='filtered',last_error='',updated_at=$2 WHERE intent.source_connection_id=$1 AND filename_is_filtered(intent.source_connection_id,intent.relative_path,intent.name) AND intent.fetch_state<>'ready'`, connectionID, now); err != nil { diff --git a/internal/platform/postgres/job_failures.go b/internal/platform/postgres/job_failures.go index 9fa2e20..f194162 100644 --- a/internal/platform/postgres/job_failures.go +++ b/internal/platform/postgres/job_failures.go @@ -15,10 +15,20 @@ const listJobFailuresSQL = ` SELECT jf.id, jf.job_id, jf.job_type, jf.publication_id, jf.error_code, jf.attempt, jf.max_attempts_at_failure, jf.business_change_set_id, jf.business_action, jf.business_target_kb_id, jf.state, jf.created_at, - COALESCE(siv.source_item_id,'') + COALESCE(siv.source_item_id,''), + COALESCE(NULLIF(item.title,''), NULLIF(ci.name,''), ''), + COALESCE(NULLIF(jf.last_error,''), j.last_error, '') FROM job_failures jf LEFT JOIN test_publications p ON p.id = jf.publication_id LEFT JOIN source_item_versions siv ON siv.id = p.source_item_version_id +LEFT JOIN source_items item ON item.id = siv.source_item_id +LEFT JOIN jobs j ON j.id = jf.job_id +LEFT JOIN LATERAL ( + SELECT intent.name FROM catalog_import_intents intent + WHERE intent.source_connection_id = j.payload->>'source_connection_id' + AND intent.node_id = j.payload->>'node_id' + LIMIT 1 +) ci ON TRUE WHERE ($1 = '' OR jf.state = $1) ORDER BY jf.created_at DESC, jf.id DESC LIMIT $2 OFFSET $3 @@ -60,7 +70,7 @@ func (s *Store) ListJobFailures(ctx context.Context, filter governance.JobFailur var item governance.JobFailure if err := rows.Scan(&item.ID, &item.JobID, &item.JobType, &item.PublicationID, &item.ErrorCode, &item.Attempt, &item.MaxAttempts, &item.ChangeSetID, &item.Action, &item.TargetKBID, - &item.State, &item.CreatedAt, &item.SourceItemID); err != nil { + &item.State, &item.CreatedAt, &item.SourceItemID, &item.DocName, &item.LastError); err != nil { return governance.JobFailurePage{}, fmt.Errorf("scan job failure: %w", err) } items = append(items, item) diff --git a/internal/platform/postgres/queue_sql.go b/internal/platform/postgres/queue_sql.go index 27a46bd..010140d 100644 --- a/internal/platform/postgres/queue_sql.go +++ b/internal/platform/postgres/queue_sql.go @@ -64,6 +64,16 @@ WHERE id = $1 AND state = 'running' ` +// retireOpenFailuresSQL keeps the dead-letter lane honest: once the job +// itself succeeded, its terminal-attempt projection is no longer waiting on +// a human (a later automatic retry recovered it). +const retireOpenFailuresSQL = ` +UPDATE job_failures +SET state = 'requeued', resolved_at = $2, resolution_actor_ref = 'system:queue', + resolution_reason = '任务后续重试成功' +WHERE job_id = $1 AND state = 'open' +` + const heartbeatJobSQL = ` UPDATE jobs SET lease_until = $4, diff --git a/internal/platform/postgres/store.go b/internal/platform/postgres/store.go index eff3694..19b99fc 100644 --- a/internal/platform/postgres/store.go +++ b/internal/platform/postgres/store.go @@ -87,13 +87,24 @@ func (s *Store) List(ctx context.Context, limit int) ([]jobs.Job, error) { } func (s *Store) Complete(ctx context.Context, id, owner string, now time.Time) error { - tag, err := s.db.Exec(ctx, completeJobSQL, id, owner, now) + tx, err := s.db.Begin(ctx) + if err != nil { + return fmt.Errorf("complete job: %w", err) + } + defer func() { _ = tx.Rollback(ctx) }() + tag, err := tx.Exec(ctx, completeJobSQL, id, owner, now) if err != nil { return fmt.Errorf("complete job: %w", err) } if tag.RowsAffected() != 1 { return ErrLeaseLost } + if _, err := tx.Exec(ctx, retireOpenFailuresSQL, id, now); err != nil { + return fmt.Errorf("complete job: retire failures: %w", err) + } + if err := tx.Commit(ctx); err != nil { + return fmt.Errorf("complete job: %w", err) + } return nil } diff --git a/internal/workflows/testweknora/upload.go b/internal/workflows/testweknora/upload.go index 960e2d5..f142319 100644 --- a/internal/workflows/testweknora/upload.go +++ b/internal/workflows/testweknora/upload.go @@ -134,15 +134,28 @@ func (handler *UploadHandler) Handle(ctx context.Context, job jobs.Job) error { return jobs.Permanent(jobs.ErrCodeArtifactInvalid, ErrUploadArtifactInvalid) } if uploadErr != nil { - if !errors.Is(uploadErr, weknora.ErrConflict) && !weknora.IsRetryable(uploadErr) { - return ErrUploadRemoteFailed - } - knowledge, err = handler.recoverAmbiguousUpload(ctx, candidate) - if err != nil { - if errors.Is(err, ErrUploadRemoteIdentity) { - return handler.recordTerminalUploadFailure(ctx, candidate.ID, "upload_identity_ambiguous") + if errors.Is(uploadErr, weknora.ErrConflict) && knowledge.ID != "" { + // WeKnora dedupes identical file bytes across publications: when the + // same document was already uploaded under a sibling publication (the + // same file living in two source folders), the conflict response names + // the existing item. Adopt it instead of dead-lettering forever. + if verified, verifyErr := handler.client.GetKnowledge(ctx, knowledge.ID); verifyErr == nil && + verified.ID == knowledge.ID && verified.KnowledgeBaseID == candidate.TestKBID && + shaMatches(verified, candidate.SHA256) { + knowledge = verified + } else { + return ErrUploadRemoteFailed + } + } else if !errors.Is(uploadErr, weknora.ErrConflict) && !weknora.IsRetryable(uploadErr) { + return ErrUploadRemoteFailed + } else { + knowledge, err = handler.recoverAmbiguousUpload(ctx, candidate) + if err != nil { + if errors.Is(err, ErrUploadRemoteIdentity) { + return handler.recordTerminalUploadFailure(ctx, candidate.ID, "upload_identity_ambiguous") + } + return err } - return err } } if knowledge.ID == "" || knowledge.KnowledgeBaseID != candidate.TestKBID { @@ -201,6 +214,15 @@ func metadataString(metadata map[string]any, key string) (string, bool) { return value, ok } +// shaMatches reports whether a remote item carries our content hash, either in +// its metadata (uploads through governance) or as its file hash. +func shaMatches(item weknora.Knowledge, sha string) bool { + if value, ok := metadataString(item.Metadata, "content_sha256"); ok && value == sha { + return true + } + return sha != "" && strings.EqualFold(item.FileHash, sha) +} + func (handler *UploadHandler) recordTerminalUploadFailure(ctx context.Context, publicationID, code string) error { if err := handler.store.RecordUploadFailure(ctx, publicationID, code, handler.now()); err != nil { return ErrUploadStateConflict diff --git a/internal/workflows/testweknora/upload_test.go b/internal/workflows/testweknora/upload_test.go index 422f1ec..3ace041 100644 --- a/internal/workflows/testweknora/upload_test.go +++ b/internal/workflows/testweknora/upload_test.go @@ -352,3 +352,53 @@ func (uploadClientStub) ListKnowledge(context.Context, string, int, int) ([]wekn func (uploadClientStub) GetKnowledge(context.Context, string) (weknora.Knowledge, error) { return weknora.Knowledge{}, errors.New("unexpected get") } + +func TestUploadHandlerAdoptsSameSHAItemOnDuplicateConflict(t *testing.T) { + root := t.TempDir() + if err := os.Mkdir(filepath.Join(root, "objects"), 0o700); err != nil { + t.Fatal(err) + } + body := []byte("duplicated across two folders\n") + if err := os.WriteFile(filepath.Join(root, "objects", "artifact.md"), body, 0o600); err != nil { + t.Fatal(err) + } + digest := sha256.Sum256(body) + sha := hex.EncodeToString(digest[:]) + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + switch { + case r.Method == http.MethodPost: + _, _ = io.Copy(io.Discard, r.Body) + w.Header().Set("Content-Type", "application/json") + w.WriteHeader(http.StatusConflict) + _, _ = w.Write([]byte(`{"success":false,"code":"duplicate_file","message":"file exists","data":{"id":"knowledge-sibling","knowledge_base_id":"kb-1","file_hash":"` + sha + `"}}`)) + case r.Method == http.MethodGet && r.URL.Path == "/api/v1/knowledge/knowledge-sibling": + w.Header().Set("Content-Type", "application/json") + _, _ = w.Write([]byte(`{"success":true,"data":{"id":"knowledge-sibling","knowledge_base_id":"kb-1","file_hash":"` + sha + `","metadata":{"publication_id":"publication-other","content_sha256":"` + sha + `"}}}`)) + default: + http.NotFound(w, r) + } + })) + defer server.Close() + client, err := weknora.NewClient(weknora.ClientConfig{BaseURL: server.URL, APIKey: "test-key", Timeout: time.Second}) + if err != nil { + t.Fatal(err) + } + store := &uploadStoreStub{candidate: postgresstore.PublicationForUpload{ + ID: "publication-1", TestKBID: "kb-1", IngestState: governance.IngestPending, + ProcessingState: governance.ProcessingPending, SHA256: sha, SizeBytes: int64(len(body)), + ValidatedFormat: "md", LocalSpoolRef: "objects/artifact.md", Title: "Safe title", + }} + handler, err := NewUploadHandler(UploadHandlerConfig{ + Store: store, Client: client, SpoolRoot: root, MaxUploadBytes: 50 * 1024 * 1024, + FirstPollDelay: time.Second, Now: func() time.Time { return time.Now() }, + }) + if err != nil { + t.Fatal(err) + } + if err := handler.Handle(context.Background(), jobs.Job{Payload: []byte(`{"publication_id":"publication-1"}`)}); err != nil { + t.Fatalf("Handle() error = %v, want duplicate adoption", err) + } + if store.recordedKnowledgeID != "knowledge-sibling" { + t.Fatalf("knowledge=%q", store.recordedKnowledgeID) + } +} diff --git a/web/src/WeKnoraMonitor.test.tsx b/web/src/WeKnoraMonitor.test.tsx index 63188ed..5383dcc 100644 --- a/web/src/WeKnoraMonitor.test.tsx +++ b/web/src/WeKnoraMonitor.test.tsx @@ -37,7 +37,10 @@ describe('WeKnoraMonitorView',()=>{ }) it('shows inbox lanes for the knowledge-base diff view',()=>{ const attention=renderToStaticMarkup() - for(const text of ['需要你出手','CR20A.pdf','remote_parse_failed','失败阶段 文档解析','批量重试所选（1）','死信（尝试耗尽，不可自动重试）'])expect(attention).toContain(text) + for(const text of ['需要你出手','CR20A.pdf','remote_parse_failed','失败阶段 文档解析','批量重试所选（1）','死信（重试次数用完，系统不再自动处理）','源头取件”＝从钉钉/AList'])expect(attention).toContain(text) + const withFailures=renderToStaticMarkup() + expect(withFailures).toContain('「ATOM整机出厂测试.adoc」') + expect(withFailures).toContain('来源服务当时不可达') const inflight=renderToStaticMarkup() for(const text of ['处理中','知识库解析进度','已导入 5750','待入库 120','对内 3300 / 4200','对外 1230 / 1550','源头取件 → Hash'])expect(inflight).toContain(text) const queueView=renderToStaticMarkup() diff --git a/web/src/WeKnoraMonitor.tsx b/web/src/WeKnoraMonitor.tsx index 24848e2..958afbe 100644 --- a/web/src/WeKnoraMonitor.tsx +++ b/web/src/WeKnoraMonitor.tsx @@ -29,6 +29,20 @@ const adviceFor = (stage: string, code: string) => { return '确认错误码后可直接重试；尝试次数耗尽将进入死信队列。' } +// failureHint turns a dead-letter row's machine code into a plain-Chinese +// diagnosis of what actually went wrong and what to do. +const failureHint = (failure: JobFailure) => { + const detail = `${failure.error_code} ${failure.last_error || ''}` + if (detail.includes('not ingestable')) return '钉钉侧该类型不是可下载文档（快捷方式、占位或内部对象），重试也不会成功；建议用“文件名筛选”把它排除，或在钉钉确认它是否为空目录残留。' + if (detail.includes('source_auth_required')) return '来源存储需要网页登录授权（服务凭据未配置），请联系管理员在“系统设置”重新扫码。' + if (detail.includes('invalid_format')) return '取回的内容不是有效的文档/文本格式，可能源头文件本身损坏；可下载原件确认后重试。' + if (detail.includes('source_unavailable')) return '来源服务当时不可达（网络/接口抖动）。任务若已自动重试成功，这里只是历史记录；仍在列表说明源头仍不可达。' + if (failure.error_code === 'test_upload_remote_failed' || detail.includes('test_upload_remote_failed')) return '向测试知识库上传被拒：确认 WeKnora 测试实例可达、目标知识库未停用后，点击重新入队。' + if (failure.error_code === 'remote_parse_failed') return '文档解析失败：建议先在 WeKnora 手动解析该文件确认格式，再重新入队。' + if (failure.error_code === 'handler_error') return '处理环节报错，展开原始错误定位；确认原因后可重新入队。' + return `错误码 ${failure.error_code}：确认原因后可重新入队。` +} + const pipelineStages = ['源头取件', 'Hash', 'OSS', '文档解析', 'WeKnora 上传', '索引', '完成'] const stageIndex = (label: string) => label === 'Hash / 文件校验' ? 1 : label === 'OSS' ? 2 : label === '文档解析' ? 3 : label === 'WeKnora 上传' ? 4 : label === '索引 / 增强' || label === '测试知识移除' || label === '治理策略' || label === '处理链路' ? 5 : -1 function chainSteps(item: TestPublication) { @@ -179,8 +193,8 @@ export function WeKnoraMonitorView({

需要你出手（{attentionItems.length}）

{batchMessage ?

{batchMessage}

: null} {attentionItems.length ? attentionItems.map(actionTask) : emptyLane('没有失败或停滞的任务。')} -

死信（尝试耗尽，不可自动重试）

原始任务与审计行保留不动；重新入队会创建新尝试
- {failures.length ?
    {failures.map((failure) =>
  • {failure.error_code}{failure.job_type === 'test.upload' ? '上传解析' : failure.job_type === 'artifact.fetch' ? '源头取件' : failure.job_type}{failure.publication_id ? {failure.publication_id} : null}{failure.attempt}/{failure.max_attempts_at_failure} 次 · {new Date(failure.created_at).toLocaleString('zh-CN', { hour12: false })}{failure.publication_id ? : null}
    • )}
: emptyLane('当前没有等待处理的死信记录。')} +

死信（重试次数用完，系统不再自动处理）

“源头取件”＝从钉钉/AList 把原始文件拉回本地；“上传解析”＝把文件送进知识库做解析。死信只是失败任务的登记表，不会改动原始记录；重新入队＝再给它一次机会
+ {failures.length ?
    {failures.map((failure) =>
  • {failure.error_code}{failure.job_type === 'test.upload' ? '上传解析' : failure.job_type === 'artifact.fetch' ? '源头取件' : failure.job_type}{failure.doc_name ? 「{failure.doc_name}」 : failure.publication_id ? {failure.publication_id} : null}{failure.attempt}/{failure.max_attempts_at_failure} 次 · {new Date(failure.created_at).toLocaleString('zh-CN', { hour12: false })}{failureHint(failure)}{failure.publication_id ? : null}
    • )}
: emptyLane('当前没有等待处理的死信记录。')} : null} {lane === 'removal' ?

需下架（{laneCounts.removal}）

来源目录里被改为“禁止导入”或撤销标注、但仍在知识库中可检索的文件
diff --git a/web/src/testPublications.ts b/web/src/testPublications.ts index e728fb4..87a2f99 100644 --- a/web/src/testPublications.ts +++ b/web/src/testPublications.ts @@ -170,7 +170,7 @@ export async function retryPublicationsBatch(publicationIds: string[], reason: s } } -export interface JobFailure { id: string; job_id: string; job_type: string; publication_id?: string; source_item_id?: string; error_code: string; attempt: number; max_attempts_at_failure: number; change_set_id?: string; action?: string; target_kb_id?: string; state: 'open' | 'requeued' | 'abandoned'; created_at: string } +export interface JobFailure { id: string; job_id: string; job_type: string; publication_id?: string; source_item_id?: string; error_code: string; attempt: number; max_attempts_at_failure: number; change_set_id?: string; action?: string; target_kb_id?: string; state: 'open' | 'requeued' | 'abandoned'; created_at: string; doc_name?: string; last_error?: string } export async function fetchJobFailures(signal?: AbortSignal): Promise { const response = await fetch('/api/v1/job-failures?state=open&limit=50', { headers: { Accept: 'application/json' }, signal }) diff --git a/web/src/weknoraMonitor.css b/web/src/weknoraMonitor.css index d19370e..92f21cd 100644 --- a/web/src/weknoraMonitor.css +++ b/web/src/weknoraMonitor.css @@ -14,7 +14,8 @@ .queue-batch-message{margin:6px 0;color:var(--c-done)} .dlq-panel{margin-top:14px} .dlq-list{list-style:none;margin:8px 0 0;padding:0;display:flex;flex-direction:column;gap:6px} -.dlq-list li{display:flex;align-items:center;gap:10px;padding:8px 10px;border:1px solid var(--border-strong);border-radius:8px;background:var(--bg-page)} +.dlq-list li{display:flex;align-items:center;gap:10px;flex-wrap:wrap;padding:8px 10px;border:1px solid var(--border-strong);border-radius:8px;background:var(--bg-page)} +.dlq-list li small{flex-basis:100%;margin-left:2px;color:var(--ink-mute)} .dlq-list small{margin-left:auto;color:var(--c-dead)} .run-metrics{margin-top:14px} .run-metrics-grid{display:grid;grid-template-columns:repeat(auto-fit,minmax(180px,1fr));gap:8px;margin-top:8px}