From cddc4c90d18188d0bcabc01e5ddf2211e8514645 Mon Sep 17 00:00:00 2001 From: hermes Date: Thu, 8 Oct 2026 12:09:34 +0800 Subject: [PATCH] =?UTF-8?q?fix(monitor):=20=E4=BF=AE=E5=A4=8D=E5=8F=96?= =?UTF-8?q?=E4=BB=B6=E6=AD=BB=E4=BF=A1=E6=97=A0=E6=B3=95=E9=87=8D=E6=8E=92?= =?UTF-8?q?=E4=B8=8E=E5=88=97=E8=A1=A8=20NULL=20=E6=A0=87=E5=BF=97?= =?UTF-8?q?=E5=AF=BC=E8=87=B4=E7=9A=84=20500?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - platform/postgres: listTestPublicationsSQL 的 can_start/can_retry 在 LEFT JOIN 内容对象为空时算出 NULL，Go 侧 bool 扫描失败使整个列表接口 500，前端三个 lane 全空（表现为“待入库 0 / 全部任务 0”）。改为 COALESCE(..., false)。 - 新增 POST /api/v1/job-failures/{id}/requeue：取件任务的幂等键绑在版本上， 失败那次已占住它，重新标注同一文件不会再入队；现在按新轮次 (…:r) 重排、 复位意图取件状态并把死信标记为 requeued（幂等信封 + 审计字段）。 - web: 死信区对 artifact.fetch 显示「重新入队」；系统设置新增「知识库绑定」分区， 挂上一直没被渲染的 RoutingBindingsSection，导入目标库可在页面上切换。 - 测试：真实 PostgreSQL 集成测试（NULL 折叠 + 重排轮次/幂等/拒绝）、前端死信按钮 与设置分区渲染测试。 --- api/openapi.yaml | 54 +++++++ internal/governance/job_failures.go | 22 +++ internal/httpapi/server.go | 50 +++++++ internal/httpapi/server_test.go | 11 ++ ...artifact_fetch_requeue_integration_test.go | 135 ++++++++++++++++++ internal/platform/postgres/job_failures.go | 129 +++++++++++++++++ .../platform/postgres/test_publications.go | 12 +- web/src/SystemSettings.test.tsx | 11 ++ web/src/SystemSettings.tsx | 5 +- web/src/WeKnoraMonitor.test.tsx | 19 +++ web/src/WeKnoraMonitor.tsx | 28 +++- web/src/testPublications.ts | 27 ++++ 12 files changed, 495 insertions(+), 8 deletions(-) create mode 100644 internal/platform/postgres/artifact_fetch_requeue_integration_test.go diff --git a/api/openapi.yaml b/api/openapi.yaml index 97837c5..e18c1f0 100644 --- a/api/openapi.yaml +++ b/api/openapi.yaml @@ -1675,6 +1675,60 @@ } } }, + "/api/v1/job-failures/{id}/requeue": { + "post": { + "operationId": "requeueJobFailure", + "summary": "重排一条终端失败的“源头取件”死信", + "description": "取件任务的幂等键绑在来源版本上，失败那次已占住它，所以重新标注同一文件不会再入队。本接口按新轮次重新入队并复位意图取件状态，同时把死信标记为 requeued。仅限 artifact.fetch 类型且仍为 open 的记录。", + "parameters": [ + { + "name": "id", + "in": "path", + "required": true, + "schema": { + "type": "string" + } + }, + { + "$ref": "#/components/parameters/IdempotencyKey" + } + ], + "requestBody": { + "required": true, + "content": { + "application/json": { + "schema": { + "type": "object", + "required": [ + "reason" + ], + "properties": { + "reason": { + "type": "string", + "minLength": 1, + "maxLength": 500 + } + } + } + } + } + }, + "responses": { + "200": { + "description": "已重新入队；replayed=true 表示幂等重放" + }, + "400": { + "description": "请求无效或缺少 Idempotency-Key" + }, + "409": { + "description": "该失败记录不可重排，或幂等键与不同请求冲突" + }, + "500": { + "description": "服务端错误" + } + } + } + }, "/api/v1/job-failures": { "get": { "summary": "分页列出解析死信（DLQ）投影", diff --git a/internal/governance/job_failures.go b/internal/governance/job_failures.go index 18d2fd3..7664a1b 100644 --- a/internal/governance/job_failures.go +++ b/internal/governance/job_failures.go @@ -81,3 +81,25 @@ type CancelPublicationsBatchRequest struct { Reason string `json:"reason"` Now time.Time `json:"-"` } + +// RequeueArtifactFetchRequest re-enqueues ONE terminal 源头取件 (artifact.fetch) +// dead letter. The fetch job's idempotency key is derived from the version +// ("artifact-fetch:"), and the failed attempt permanently occupies +// it — so a plain re-mark of the same file resolves to the same version and its +// ON CONFLICT DO NOTHING insert is silently swallowed. A requeue therefore opens +// a new ROUND ("...:r") instead of replaying the original key. +type RequeueArtifactFetchRequest struct { + FailureID string + OperatorRef string + Reason string + Now time.Time +} + +type RequeueArtifactFetchResult struct { + FailureID string `json:"failure_id"` + JobID string `json:"job_id"` + VersionID string `json:"version_id"` + Round int `json:"round"` + DocName string `json:"doc_name,omitempty"` + AlreadyQueued bool `json:"already_queued"` +} diff --git a/internal/httpapi/server.go b/internal/httpapi/server.go index 992b02a..ea9b4c5 100644 --- a/internal/httpapi/server.go +++ b/internal/httpapi/server.go @@ -96,6 +96,7 @@ type Store interface { ListJobFailures(ctx context.Context, filter governance.JobFailureFilter) (governance.JobFailurePage, error) RetryPublicationsBatchWithOperation(ctx context.Context, request governance.RetryPublicationsBatchRequest, operationID string) (governance.RetryPublicationsBatchResult, bool, error) CancelPublicationsBatchWithOperation(ctx context.Context, request governance.CancelPublicationsBatchRequest, operationID string) (governance.RetryPublicationsBatchResult, bool, error) + RequeueArtifactFetchWithOperation(ctx context.Context, request governance.RequeueArtifactFetchRequest, operationID string) (governance.RequeueArtifactFetchResult, bool, error) } type Config struct { @@ -394,6 +395,9 @@ func NewHandler(config Config) (http.Handler, error) { mux.HandleFunc("GET /api/v1/job-failures", func(w http.ResponseWriter, r *http.Request) { listJobFailures(w, r, config.Store) }) + mux.HandleFunc("POST /api/v1/job-failures/{id}/requeue", func(w http.ResponseWriter, r *http.Request) { + requeueJobFailure(w, r, config) + }) if config.Auth != nil { if config.Auth.Store == nil { return nil, errors.New("auth configuration requires a store") @@ -2016,6 +2020,52 @@ func listJobFailures(w http.ResponseWriter, r *http.Request, store Store) { writeJSON(w, http.StatusOK, map[string]any{"data": page}) } +// requeueJobFailure re-enqueues one terminal 源头取件 (artifact.fetch) dead +// letter. 上传解析 dead letters keep their publication-keyed retry path; a +// fetch failure carries no publication id, so without this route the file is +// stuck forever after a single terminal fetch failure (expired source session). +func requeueJobFailure(w http.ResponseWriter, r *http.Request, config Config) { + operationID, ok := requireOperationID(w, r) + if !ok { + return + } + failureID := strings.TrimSpace(r.PathValue("id")) + if failureID == "" || len(failureID) > 64 || !validCatalogID(failureID) { + writeJSON(w, http.StatusBadRequest, map[string]string{"error": "invalid failure id"}) + return + } + var input struct { + Reason string `json:"reason"` + } + if err := decodeStrictJSON(r, &input); err != nil { + writeJSON(w, http.StatusBadRequest, map[string]string{"error": "invalid requeue request"}) + return + } + reason := strings.TrimSpace(input.Reason) + if reason == "" || len(reason) > 500 { + writeJSON(w, http.StatusBadRequest, map[string]string{"error": "reason is required"}) + return + } + result, replayed, err := config.Store.RequeueArtifactFetchWithOperation(r.Context(), governance.RequeueArtifactFetchRequest{ + FailureID: failureID, + OperatorRef: actorFor(r), + Reason: reason, + Now: config.Now().UTC(), + }, operationID) + switch { + case errors.Is(err, postgresstore.ErrJobFailureNotRequeueable): + writeJSON(w, http.StatusConflict, map[string]string{"error": "该失败记录不是可重排的源头取件任务，或已经处理过"}) + return + case errors.Is(err, postgresstore.ErrOperationRequestConflict): + writeJSON(w, http.StatusConflict, map[string]string{"error": "Idempotency-Key conflicts with a different request"}) + return + case err != nil: + writeJSON(w, http.StatusInternalServerError, map[string]string{"error": "could not requeue artifact fetch"}) + return + } + writeJSON(w, http.StatusOK, map[string]any{"data": result, "replayed": replayed}) +} + func retryPublicationsBatch(w http.ResponseWriter, r *http.Request, config Config) { operationID, ok := requireOperationID(w, r) if !ok { diff --git a/internal/httpapi/server_test.go b/internal/httpapi/server_test.go index 6492fb1..c757d99 100644 --- a/internal/httpapi/server_test.go +++ b/internal/httpapi/server_test.go @@ -1515,6 +1515,11 @@ type fakeJobStore struct { cancelBatchErr error retryBatchReplayed bool retryBatchErr error + requeueFailureRequest *governance.RequeueArtifactFetchRequest + requeueFailureOperation string + requeueFailureResult governance.RequeueArtifactFetchResult + requeueFailureReplayed bool + requeueFailureErr error enqueued jobs.EnqueueRequest enqueueResult *jobs.Job listResult []jobs.Job @@ -1925,6 +1930,12 @@ func (s *fakeJobStore) CancelPublicationsBatchWithOperation(_ context.Context, r return s.cancelBatchResult, s.cancelBatchReplayed, s.cancelBatchErr } +func (s *fakeJobStore) RequeueArtifactFetchWithOperation(_ context.Context, request governance.RequeueArtifactFetchRequest, operationID string) (governance.RequeueArtifactFetchResult, bool, error) { + s.requeueFailureRequest = &request + s.requeueFailureOperation = operationID + return s.requeueFailureResult, s.requeueFailureReplayed, s.requeueFailureErr +} + func (s *fakeJobStore) ListScopeRules(_ context.Context, connectionID string) ([]governance.ScopeRule, error) { s.scopeRuleConnectionID = connectionID return s.scopeRules, s.scopeRuleErr diff --git a/internal/platform/postgres/artifact_fetch_requeue_integration_test.go b/internal/platform/postgres/artifact_fetch_requeue_integration_test.go new file mode 100644 index 0000000..deea2fb --- /dev/null +++ b/internal/platform/postgres/artifact_fetch_requeue_integration_test.go @@ -0,0 +1,135 @@ +package postgres + +import ( + "context" + "errors" + "os" + "testing" + "time" + + "github.com/jackc/pgx/v5/pgxpool" + + "knowledge-governance/internal/governance" +) + +// Two defects this pins down (both hit on 2026-10-08 while running the first +// end-to-end document): +// +// 1. listTestPublicationsSQL computed can_start/can_retry against a +// LEFT JOINed content object, so a planned row whose artifact was never +// fetched produced NULL — and scanning NULL into a Go bool made the whole +// /api/v1/test-publications call 500, which the monitor renders as three +// empty lanes ("待入库 0 / 全部任务 0"). +// 2. An artifact.fetch dead letter carried no publication id, so the UI had no +// retry affordance and the fetch idempotency key (artifact-fetch:) +// was permanently occupied by the failed attempt — re-marking the file +// could never re-enqueue it. +func TestPlannedPublicationFlagsFoldNullAndArtifactFetchRequeue(t *testing.T) { + databaseURL := isolatedIntegrationDatabaseURL(t, os.Getenv("TEST_DATABASE_URL")) + if databaseURL == "" { + t.Skip("TEST_DATABASE_URL is not set") + } + ctx := context.Background() + pool, err := pgxpool.New(ctx, databaseURL) + if err != nil { + t.Fatal(err) + } + defer pool.Close() + if err = ApplyMigrations(ctx, pool); err != nil { + t.Fatal(err) + } + if _, err = pool.Exec(ctx, ` +INSERT INTO source_connections(id,source_type,name,enabled) VALUES('conn-fetch','dingtalk','Fetch DT',true); +INSERT INTO source_items(id,source_connection_id,stable_external_id,canonical_uri_hash,normalized_path,title,item_type,revision,visibility,visibility_origin) +VALUES('item-fetch','conn-fetch','node-1',repeat('b',64),'技术文档/环境部署/关闭PC2远程桌面弹窗.docx','关闭PC2远程桌面弹窗.docx','document','locator:h1','internal','configured_scope'); +INSERT INTO source_item_versions(id,source_item_id,revision,content_object_id,content_state) +VALUES('version-fetch','item-fetch','locator:h1',NULL,'unknown'); +INSERT INTO test_publications(id,source_item_version_id,test_kb_id,processing_state,target_scope) +VALUES('pub-fetch','version-fetch','kb-internal','planned','internal'); +INSERT INTO jobs(id,type,payload,state,priority,max_attempts,idempotency_key,last_error,last_error_code) +VALUES('job-fetch-1','artifact.fetch', + '{"source_connection_id":"conn-fetch","node_id":"node-1","version_id":"version-fetch"}'::jsonb, + 'failed',20,8,'artifact-fetch:version-fetch','artifact_fetch_rejected: http_401','artifact_invalid'); +INSERT INTO job_failures(id,job_id,job_type,publication_id,error_code,last_error,attempt,max_attempts_at_failure,state) +VALUES('jf-fetch-1','job-fetch-1','artifact.fetch','','artifact_invalid','artifact_fetch_rejected: http_401',1,8,'open'); +INSERT INTO catalog_import_intents(source_connection_id,node_id,visibility,decided_by,relative_path,name,locator_hash,fetch_state,last_error) +VALUES('conn-fetch','node-1','internal','tester','技术文档/环境部署/关闭PC2远程桌面弹窗.docx','关闭PC2远程桌面弹窗.docx',repeat('c',64),'failed','artifact_fetch_rejected: http_401'); +`); err != nil { + t.Fatal(err) + } + store := NewStore(pool) + now := time.Date(2026, time.October, 8, 3, 40, 0, 0, time.UTC) + + // 1. The unfetched planned row must list cleanly with false flags, not error. + page, err := store.ListTestPublications(ctx, governance.TestPublicationFilter{Limit: 50}) + if err != nil { + t.Fatalf("ListTestPublications() error = %v", err) + } + if page.Total != 1 || len(page.Items) != 1 || page.Items[0].ID != "pub-fetch" { + t.Fatalf("page = %#v", page) + } + if page.Items[0].CanStart || page.Items[0].CanRetry { + t.Fatalf("NULL flags must fold to false, got %#v", page.Items[0]) + } + if page.Items[0].Title != "关闭PC2远程桌面弹窗.docx" { + t.Fatalf("title = %q", page.Items[0].Title) + } + + // 2. Requeue opens a new round instead of replaying the occupied key. + request := governance.RequeueArtifactFetchRequest{ + FailureID: "jf-fetch-1", OperatorRef: "tester", Reason: "钉钉会话已刷新", Now: now, + } + result, replayed, err := store.RequeueArtifactFetchWithOperation(ctx, request, "op-requeue-1") + if err != nil { + t.Fatalf("RequeueArtifactFetchWithOperation() error = %v", err) + } + if replayed || result.AlreadyQueued { + t.Fatalf("first requeue must enqueue, got %#v replayed=%v", result, replayed) + } + if result.Round != 2 || result.VersionID != "version-fetch" || result.JobID == "" { + t.Fatalf("result = %#v", result) + } + var state, key string + if err = pool.QueryRow(ctx, `SELECT state, idempotency_key FROM jobs WHERE id = $1`, result.JobID).Scan(&state, &key); err != nil { + t.Fatal(err) + } + if state != "pending" || key != "artifact-fetch:version-fetch:r2" { + t.Fatalf("requeued job = %s / %s", state, key) + } + var failureState, fetchState string + if err = pool.QueryRow(ctx, `SELECT state FROM job_failures WHERE id = 'jf-fetch-1'`).Scan(&failureState); err != nil { + t.Fatal(err) + } + if failureState != "requeued" { + t.Fatalf("dead letter state = %s", failureState) + } + if err = pool.QueryRow(ctx, `SELECT fetch_state FROM catalog_import_intents WHERE node_id = 'node-1'`).Scan(&fetchState); err != nil { + t.Fatal(err) + } + if fetchState != "pending" { + t.Fatalf("intent fetch state = %s", fetchState) + } + + // Replaying the SAME operation must return the stored result, not enqueue again. + again, replayedAgain, err := store.RequeueArtifactFetchWithOperation(ctx, request, "op-requeue-1") + if err != nil { + t.Fatalf("replay error = %v", err) + } + if !replayedAgain || again.JobID != result.JobID || again.Round != result.Round { + t.Fatalf("replay = %#v replayed=%v", again, replayedAgain) + } + var fetchJobs int + if err = pool.QueryRow(ctx, `SELECT COUNT(*) FROM jobs WHERE type = 'artifact.fetch'`).Scan(&fetchJobs); err != nil { + t.Fatal(err) + } + if fetchJobs != 2 { + t.Fatalf("artifact.fetch job count = %d, want 2 (original + one requeue)", fetchJobs) + } + + // A resolved dead letter is no longer requeueable. + if _, _, err = store.RequeueArtifactFetchWithOperation(ctx, request, "op-requeue-2"); err == nil { + t.Fatal("second requeue of the same dead letter must fail") + } else if !errors.Is(err, ErrJobFailureNotRequeueable) { + t.Fatalf("unexpected error: %v", err) + } +} \ No newline at end of file diff --git a/internal/platform/postgres/job_failures.go b/internal/platform/postgres/job_failures.go index e41ef72..bc80aff 100644 --- a/internal/platform/postgres/job_failures.go +++ b/internal/platform/postgres/job_failures.go @@ -11,6 +11,135 @@ import ( "knowledge-governance/internal/governance" ) +// ErrJobFailureNotRequeueable marks a dead letter that cannot be re-enqueued: +// not an artifact fetch, already resolved, or missing its source coordinates. +var ErrJobFailureNotRequeueable = errors.New("job failure is not requeueable") + +const loadRequeueableArtifactFailureSQL = ` +SELECT jf.state, jf.job_type, + COALESCE(j.payload->>'source_connection_id',''), + COALESCE(j.payload->>'node_id',''), + COALESCE(j.payload->>'version_id',''), + COALESCE(NULLIF(intent.name,''),'') +FROM job_failures jf +LEFT JOIN jobs j ON j.id = jf.job_id +LEFT JOIN catalog_import_intents intent + ON intent.source_connection_id = j.payload->>'source_connection_id' + AND intent.node_id = j.payload->>'node_id' +WHERE jf.id = $1 +FOR UPDATE OF jf +` + +const countArtifactFetchRoundsSQL = ` +SELECT COUNT(*) FROM jobs WHERE type='artifact.fetch' AND payload->>'version_id' = $1 +` + +const requeueArtifactFetchJobSQL = ` +INSERT INTO jobs(id,type,payload,state,priority,available_at,max_attempts,idempotency_key,created_at,updated_at) +VALUES($1,'artifact.fetch',$2::jsonb,'pending',20,$3,8,$4,$3,$3) +ON CONFLICT (idempotency_key) DO NOTHING +` + +const resetIntentFetchStateSQL = ` +UPDATE catalog_import_intents SET fetch_state='pending', last_error='', updated_at=$3 +WHERE source_connection_id=$1 AND node_id=$2 AND fetch_state IN ('failed','missing_local','fetching') +` + +const resolveRequeuedFailureSQL = ` +UPDATE job_failures +SET state = 'requeued', resolved_at = $2, resolution_actor_ref = $3, resolution_reason = $4 +WHERE id = $1 AND state = 'open' +` + +// RequeueArtifactFetchWithOperation re-enqueues one terminal 源头取件 dead +// letter under a new round-suffixed idempotency key, resets the intent's fetch +// state and closes the dead letter — all in one operation envelope. +func (s *Store) RequeueArtifactFetchWithOperation(ctx context.Context, request governance.RequeueArtifactFetchRequest, operationID string) (governance.RequeueArtifactFetchResult, bool, error) { + if operationID == "" || len(operationID) > 128 { + return governance.RequeueArtifactFetchResult{}, false, errors.New("invalid operation id") + } + if request.FailureID == "" || len(request.FailureID) > 64 { + return governance.RequeueArtifactFetchResult{}, false, errors.New("invalid failure id") + } + if request.OperatorRef == "" || len(request.OperatorRef) > 255 || request.Reason == "" || len(request.Reason) > 500 || request.Now.IsZero() { + return governance.RequeueArtifactFetchResult{}, false, errors.New("actor_ref and a non-empty reason are required") + } + fingerprint, err := operationFingerprint(struct { + FailureID string `json:"failure_id"` + OperatorRef string `json:"operator_ref"` + Reason string `json:"reason"` + }{request.FailureID, request.OperatorRef, request.Reason}) + if err != nil { + return governance.RequeueArtifactFetchResult{}, false, err + } + raw, replayed, err := s.executeOperation(ctx, "artifact.fetch.requeue", operationID, fingerprint, request.Now, func(tx pgx.Tx) (json.RawMessage, error) { + result, err := requeueArtifactFetchTx(ctx, tx, request) + if err != nil { + return nil, err + } + return json.Marshal(result) + }) + if err != nil { + return governance.RequeueArtifactFetchResult{}, false, err + } + var result governance.RequeueArtifactFetchResult + if err := json.Unmarshal(raw, &result); err != nil { + return governance.RequeueArtifactFetchResult{}, false, errors.New("invalid stored requeue result") + } + return result, replayed, nil +} + +func requeueArtifactFetchTx(ctx context.Context, tx pgx.Tx, request governance.RequeueArtifactFetchRequest) (governance.RequeueArtifactFetchResult, error) { + var state, jobType, connectionID, nodeID, versionID, docName string + err := tx.QueryRow(ctx, loadRequeueableArtifactFailureSQL, request.FailureID). + Scan(&state, &jobType, &connectionID, &nodeID, &versionID, &docName) + if errors.Is(err, pgx.ErrNoRows) { + return governance.RequeueArtifactFetchResult{}, ErrJobFailureNotRequeueable + } + if err != nil { + return governance.RequeueArtifactFetchResult{}, fmt.Errorf("load requeueable failure: %w", err) + } + if jobType != "artifact.fetch" || state != "open" || connectionID == "" || nodeID == "" || versionID == "" { + return governance.RequeueArtifactFetchResult{}, ErrJobFailureNotRequeueable + } + var rounds int + if err := tx.QueryRow(ctx, countArtifactFetchRoundsSQL, versionID).Scan(&rounds); err != nil { + return governance.RequeueArtifactFetchResult{}, fmt.Errorf("count artifact fetch rounds: %w", err) + } + round := rounds + 1 + jobID, err := newOpaqueID("job-") + if err != nil { + return governance.RequeueArtifactFetchResult{}, err + } + payload, err := json.Marshal(map[string]string{ + "source_connection_id": connectionID, "node_id": nodeID, "version_id": versionID, + }) + if err != nil { + return governance.RequeueArtifactFetchResult{}, errors.New("encode artifact fetch job") + } + tag, err := tx.Exec(ctx, requeueArtifactFetchJobSQL, jobID, string(payload), request.Now, + fmt.Sprintf("artifact-fetch:%s:r%d", versionID, round)) + if err != nil { + return governance.RequeueArtifactFetchResult{}, fmt.Errorf("requeue artifact fetch: %w", err) + } + result := governance.RequeueArtifactFetchResult{ + FailureID: request.FailureID, JobID: jobID, VersionID: versionID, Round: round, DocName: docName, + } + if tag.RowsAffected() == 0 { + // The round key already exists (concurrent requeue): report it instead of + // resolving the dead letter twice. + result.AlreadyQueued = true + return result, nil + } + if _, err := tx.Exec(ctx, resetIntentFetchStateSQL, connectionID, nodeID, request.Now); err != nil { + return governance.RequeueArtifactFetchResult{}, fmt.Errorf("reset intent fetch state: %w", err) + } + if _, err := tx.Exec(ctx, resolveRequeuedFailureSQL, request.FailureID, request.Now, request.OperatorRef, request.Reason); err != nil { + return governance.RequeueArtifactFetchResult{}, fmt.Errorf("resolve job failure: %w", err) + } + return result, nil +} + 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, diff --git a/internal/platform/postgres/test_publications.go b/internal/platform/postgres/test_publications.go index 42a0777..46da3a3 100644 --- a/internal/platform/postgres/test_publications.go +++ b/internal/platform/postgres/test_publications.go @@ -128,16 +128,20 @@ SELECT p.id, si.title, p.ingest_state, p.processing_state, p.retrieval_state, JOIN source_item_versions other_version ON other_version.id=other.source_item_version_id WHERE other.id<>p.id AND other_version.source_item_id=siv.source_item_id ) THEN 'updated' ELSE 'added' END AS change_type, - (p.ingest_state='pending' AND p.processing_state IN ('pending','planned') + -- Both flags compare against a LEFT-JOINed content object, so the raw + -- expression is NULL (not false) for a row whose artifact was never + -- fetched; a NULL here makes the whole list fail to scan into a Go bool + -- and 500s the endpoint. Fold NULL to false. + COALESCE((p.ingest_state='pending' AND p.processing_state IN ('pending','planned') AND p.retrieval_state='unpublished' AND p.knowledge_id='' AND co.content_state='verified' AND co.local_spool_ref<>'' AND co.size_bytes>0 AND NOT EXISTS (SELECT 1 FROM processing_confirmations confirmation WHERE confirmation.publication_id=p.id) - AND NOT EXISTS (SELECT 1 FROM jobs job WHERE job.type='test.upload' AND job.payload->>'publication_id'=p.id)) AS can_start, + AND NOT EXISTS (SELECT 1 FROM jobs job WHERE job.type='test.upload' AND job.payload->>'publication_id'=p.id)), false) AS can_start, p.retry_count, p.last_error, - ((p.processing_state IN ('failed', 'stalled') + COALESCE(((p.processing_state IN ('failed', 'stalled') OR (p.ingest_state = 'failed' AND p.processing_state = 'pending')) AND co.content_state = 'verified' - AND co.local_spool_ref <> '') AS can_retry, + AND co.local_spool_ref <> ''), false) AS can_retry, p.updated_at FROM test_publications p JOIN source_item_versions siv ON siv.id = p.source_item_version_id diff --git a/web/src/SystemSettings.test.tsx b/web/src/SystemSettings.test.tsx index 211fc8f..24e1a8e 100644 --- a/web/src/SystemSettings.test.tsx +++ b/web/src/SystemSettings.test.tsx @@ -55,6 +55,17 @@ describe('SystemSettingsView',()=>{ expect(html).not.toContain('添加内容源') expect(html).not.toContain('管理员编辑') }) + it('lets an administrator bind the internal/external import knowledge bases from 系统设置',()=>{ + // 组件一直在（RoutingBindingsSection），但从未挂到任何页面 —— 导入目标库只能靠 + // 部署 env 钉死，换库必须重建容器。 + const view=(section:'sources'|'bindings')=>renderToStaticMarkup() + expect(view('sources')).toContain('知识库绑定') + const bound=view('bindings') + expect(bound).toContain('id="settings-routing-bindings"') + expect(bound).toContain('按这里的绑定分发到对应 WeKnora 知识库') + // 未读到绑定数据时（SSR 不跑 effect）面板也要如实存在，而不是空白。 + expect(bound).toContain('读取中…') + }) it('does not show healthy simulation when the settings data failed to load',()=>{ const html=renderToStaticMarkup() expect(html).toContain('加载失败');expect(html).not.toContain('模拟运行中') diff --git a/web/src/SystemSettings.tsx b/web/src/SystemSettings.tsx index d2b164f..f53aedf 100644 --- a/web/src/SystemSettings.tsx +++ b/web/src/SystemSettings.tsx @@ -2,6 +2,7 @@ import { useCallback, useEffect, useState } from 'react' import { MembersSection } from './MembersSection' import { DingtalkSessionControl } from './DingtalkSessionControl' +import { RoutingBindingsSection } from './RoutingBindingsSection' import { DingtalkSpaceRegistrar } from './DingtalkSpaceRegistrar' import { FilenameFilterEditor } from './FilenameFilterEditor' import { useOptionalAuth } from './auth' @@ -34,11 +35,12 @@ interface Props { initialSection?: SectionKey } -type SectionKey = 'sources' | 'services' | 'storage' | 'refresh' | 'members' +type SectionKey = 'sources' | 'services' | 'bindings' | 'storage' | 'refresh' | 'members' const sections: { key: SectionKey; label: string; adminOnly?: boolean }[] = [ { key: 'sources', label: '数据源配置' }, { key: 'services', label: '服务配置' }, + { key: 'bindings', label: '知识库绑定' }, { key: 'storage', label: '存储与备份' }, { key: 'refresh', label: '刷新策略' }, { key: 'members', label: '成员与角色', adminOnly: true }, @@ -258,6 +260,7 @@ export function SystemSettingsView({ sources, connections, monitor, settings, pr {serviceSettingsValid ? null :

格式不合法：地址需 http:// 或 https:// 开头的纯地址，不能带用户名密码、? 查询串或 # 片段；生产 ID 需为十进制数字；知识库 ID 不能含空格、/、?、# 或 \。

} : 服务配置仅管理员可修改；当前值：测试地址 {settings.test_weknora_base_url || '未配置'} · 生产地址 {settings.production_weknora_base_url || '未配置'} · 生产 ID {settings.production_tenant_id || '未配置'} · 知识库 ID {settings.production_knowledge_base_id || '未配置'}。} : null} + {section === 'bindings' ? : null} {section === 'storage' ?

存储与备份

文件资源与生产发布前快照使用独立受限配置。

Endpoint / Bucket
尚未配置
AccessKey / Secret
仅保存凭据引用，不可回读
CDN / 公开访问域名
尚未配置
临时签名有效期
待配置
测试备份前缀
/backup/weknora/test/
生产备份前缀
/backup/weknora/prod/
快照保留
最近 30 个生产快照
恢复验证
由生产门禁读回
  • PostgreSQL 未配置
  • Elasticsearch 未配置
  • 对象存储 未配置
  • Neo4j 未配置
  • 生产发布请求 {productionEnabled ? '等待管理员核对' : '关闭'}
  • Release Worker 网页不可启用
diff --git a/web/src/WeKnoraMonitor.test.tsx b/web/src/WeKnoraMonitor.test.tsx index 4f55ca3..ad527ca 100644 --- a/web/src/WeKnoraMonitor.test.tsx +++ b/web/src/WeKnoraMonitor.test.tsx @@ -70,4 +70,23 @@ describe('WeKnoraMonitorView',()=>{ expect(html).toContain('已显示 3 / 共 115') expect(html).toContain('加载更多') }) + it('offers 重新入队 for a 源头取件 dead letter that carries no publication id',()=>{ + // 取件死信没有 publication_id，所以不能走发布重试链路；没有这个按钮，一个终态 + // 取件失败（例如钉钉会话过期）在界面上就永远救不回来。 + const failures=[{id:'jf-fetch',job_id:'job-fetch',job_type:'artifact.fetch',error_code:'artifact_invalid',attempt:1,max_attempts_at_failure:8,state:'open' as const,created_at:'2026-10-08T03:20:35Z',doc_name:'关闭PC2远程桌面弹窗.docx',last_error:'artifact_fetch_rejected: http_401'}] + const html=renderToStaticMarkup() + expect(html).toContain('源头取件') + expect(html).toContain('关闭PC2远程桌面弹窗.docx') + expect(html).toContain('重新入队') + const busy=renderToStaticMarkup() + expect(busy).toContain('重排中…') + expect(busy).not.toContain('>重新入队<') + }) + it('does not offer 重新入队 for a 上传解析 dead letter without a publication id',()=>{ + const failures=[{id:'jf-upload',job_id:'job-upload',job_type:'test.upload',error_code:'handler_error',attempt:8,max_attempts_at_failure:8,state:'open' as const,created_at:'2026-10-08T03:20:35Z',last_error:'remote_parse_failed'}] + const html=renderToStaticMarkup() + expect(html).toContain('上传解析') + // 段落提示语里始终有“重新入队＝再给它一次机会”，所以断言按钮本身不存在。 + expect(html).not.toContain('>重新入队') + }) }) diff --git a/web/src/WeKnoraMonitor.tsx b/web/src/WeKnoraMonitor.tsx index 688de4d..2801813 100644 --- a/web/src/WeKnoraMonitor.tsx +++ b/web/src/WeKnoraMonitor.tsx @@ -3,7 +3,7 @@ import { useContext, useEffect, useMemo, useState } from 'react' import { fetchMonitorFiles, fetchMonitorOverview, fetchRunMetrics } from './monitor' import type { RunMetrics } from './monitor' import type { KnowledgeProgress, MonitorFilePage, MonitorOverview, MonitorStatus } from './monitor' -import { fetchJobFailures, fetchTestPublications, retryPublication, retryPublicationsBatch, cancelPublicationsBatch, startTestPublications } from './testPublications' +import { fetchJobFailures, fetchTestPublications, retryPublication, retryPublicationsBatch, cancelPublicationsBatch, requeueJobFailure, startTestPublications } from './testPublications' import type { JobFailure, ProcessingState, TestPublication, TestPublicationPage } from './testPublications' import { fetchActiveUISettings, settingsRefreshDelays } from './uiSettings' import { WorkspaceContextTabsSetter } from './workspaceContext' @@ -87,6 +87,7 @@ interface ViewProps { loading: boolean error: string | null retryingIDs?: string[] + requeueingIDs?: string[] cancellingIDs?: string[] starting?: boolean startMessage?: string @@ -102,6 +103,7 @@ interface ViewProps { onRetry?: (id: string) => void onBatchRetry?: (ids: string[]) => void onBatchCancel?: (ids: string[]) => void + onRequeueFailure?: (id: string) => void onStart?: (ids: string[]) => void onLoadMore?: () => void onRefresh: () => void @@ -122,6 +124,7 @@ export function WeKnoraMonitorView({ loading, error, retryingIDs = [], + requeueingIDs = [], cancellingIDs = [], starting = false, loadingMore = false, @@ -136,6 +139,7 @@ export function WeKnoraMonitorView({ onRetry = () => undefined, onBatchRetry = () => undefined, onBatchCancel = () => undefined, + onRequeueFailure = () => undefined, onStart = () => undefined, onLoadMore = () => undefined, onRefresh, @@ -209,7 +213,7 @@ export function WeKnoraMonitorView({ {batchMessage ?

{batchMessage}

: null} {attentionItems.length ? attentionItems.map(actionTask) : 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('当前没有等待处理的死信记录。')} + {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.job_type === 'artifact.fetch' ? : failure.publication_id ? : null}
    • )}
: emptyLane('当前没有等待处理的死信记录。')}
: null} {lane === 'removal' ?

需下架（{laneCounts.removal}）

来源目录里被改为“禁止导入”或撤销标注、但仍在知识库中可检索的文件
@@ -265,6 +269,7 @@ export function WeKnoraMonitor() { const [documentRefreshMs, setDocumentRefreshMs] = useState(15000) const [resourceRefreshMs, setResourceRefreshMs] = useState(30000) const [retryingIDs, setRetryingIDs] = useState([]) + const [requeueingIDs, setRequeueingIDs] = useState([]) const [cancellingIDs, setCancellingIDs] = useState([]) const [starting, setStarting] = useState(false) const [startMessage, setStartMessage] = useState('') @@ -344,6 +349,23 @@ export function WeKnoraMonitor() { return () => controller.abort() }, [documentRevision, resourceRevision]) + const requeueFailure = async (failureID: string) => { + setRequeueingIDs((current) => [...current, failureID]) + setBatchMessage('') + try { + const result = await requeueJobFailure(failureID, '源头取件死信重新入队') + setError(null) + setBatchMessage(result.already_queued + ? `该取件任务已在队列中（第 ${result.round} 轮）` + : `已重新入队：${result.doc_name || result.version_id.slice(0, 12)}（第 ${result.round} 轮取件），取回后自动进入待入库`) + setDocumentRevision((value) => value + 1) + } catch (cause) { + setError(cause instanceof Error ? cause.message : '重新取件失败') + } finally { + setRequeueingIDs((current) => current.filter((value) => value !== failureID)) + } + } + const retryOne = async (id: string) => { setRetryingIDs((current) => [...current, id]) try { @@ -419,5 +441,5 @@ export function WeKnoraMonitor() { setStarting(false) } } - return { setSelected(kb); setStatus(''); setFiles(null) }} onStatus={(value) => { setStatus(value); setFiles(null) }} onQueueStatus={setQueueStatus} onQueueScope={setQueueScope} lane={lane} onLaneChange={(value) => setLane(value as typeof lane)} onRetry={(id) => void retryOne(id)} onBatchRetry={(ids) => void retryBatch(ids)} onBatchCancel={(ids) => void cancelBatch(ids)} onStart={(ids) => void startPlans(ids)} attention={attention ?? undefined} loadingMore={loadingMore} onLoadMore={() => void loadMoreQueue()} onRefresh={() => { setDocumentRevision((value) => value + 1); setResourceRevision((value) => value + 1) }} /> + return { setSelected(kb); setStatus(''); setFiles(null) }} onStatus={(value) => { setStatus(value); setFiles(null) }} onQueueStatus={setQueueStatus} onQueueScope={setQueueScope} lane={lane} onLaneChange={(value) => setLane(value as typeof lane)} onRetry={(id) => void retryOne(id)} onBatchRetry={(ids) => void retryBatch(ids)} onBatchCancel={(ids) => void cancelBatch(ids)} onRequeueFailure={(id) => void requeueFailure(id)} onStart={(ids) => void startPlans(ids)} attention={attention ?? undefined} loadingMore={loadingMore} onLoadMore={() => void loadMoreQueue()} onRefresh={() => { setDocumentRevision((value) => value + 1); setResourceRevision((value) => value + 1) }} /> } diff --git a/web/src/testPublications.ts b/web/src/testPublications.ts index d7c9a42..542c150 100644 --- a/web/src/testPublications.ts +++ b/web/src/testPublications.ts @@ -208,3 +208,30 @@ export async function fetchJobFailures(signal?: AbortSignal): Promise { + const response = await fetch(`/api/v1/job-failures/${encodeURIComponent(failureId)}/requeue`, { + method: 'POST', + headers: { Accept: 'application/json', 'Content-Type': 'application/json', 'Idempotency-Key': newOperationID() }, + body: JSON.stringify({ reason }), + }) + if (!response.ok) { + const body = (await response.json().catch(() => null)) as { error?: string } | null + throw new Error(body?.error ?? `重新取件失败（HTTP ${response.status}）`) + } + const value = await response.json() as unknown + if (!isRecord(value) || !isRecord(value.data)) throw new Error('重新取件响应无效') + const data = value.data as Record + return { + failure_id: typeof data.failure_id === 'string' ? data.failure_id : '', + job_id: typeof data.job_id === 'string' ? data.job_id : '', + version_id: typeof data.version_id === 'string' ? data.version_id : '', + round: isCount(data.round) ? data.round : 0, + doc_name: typeof data.doc_name === 'string' ? data.doc_name : undefined, + already_queued: data.already_queued === true, + } +}