fix(monitor): 修复取件死信无法重排与列表 NULL 标志导致的 500
- 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<n>) 重排、
复位意图取件状态并把死信标记为 requeued(幂等信封 + 审计字段)。
- web: 死信区对 artifact.fetch 显示「重新入队」;系统设置新增「知识库绑定」分区,
挂上一直没被渲染的 RoutingBindingsSection,导入目标库可在页面上切换。
- 测试:真实 PostgreSQL 集成测试(NULL 折叠 + 重排轮次/幂等/拒绝)、前端死信按钮
与设置分区渲染测试。
This commit is contained in:
@@ -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)投影",
|
||||
|
||||
@@ -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:<version_id>"), 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<n>") 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"`
|
||||
}
|
||||
|
||||
@@ -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 {
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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:<version>)
|
||||
// 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)
|
||||
}
|
||||
}
|
||||
@@ -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,
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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(<AuthContext.Provider value={authValue('admin')}><SystemSettingsView sources={sources} connections={connections} monitor={monitor} settings={settings} productionEnabled={false} routingCapabilities={{internal_enabled:true,external_enabled:true}} loading={false} saving={false} error={null} saveMessage={null} onSave={vi.fn()} onSourcesChanged={vi.fn()} onRefresh={vi.fn()} initialSection={section} /></AuthContext.Provider>)
|
||||
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(<SystemSettingsView sources={[]} connections={[]} monitor={null} settings={null} productionEnabled={false} routingCapabilities={{internal_enabled:false,external_enabled:false}} loading={false} saving={false} error="加载失败" saveMessage={null} onSave={vi.fn()} onSourcesChanged={vi.fn()} onRefresh={vi.fn()} />)
|
||||
expect(html).toContain('加载失败');expect(html).not.toContain('模拟运行中')
|
||||
|
||||
@@ -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 : <p className="error-message" role="alert">格式不合法:地址需 http:// 或 https:// 开头的纯地址,不能带用户名密码、? 查询串或 # 片段;生产 ID 需为十进制数字;知识库 ID 不能含空格、/、?、# 或 \。</p>}</>
|
||||
: <small>服务配置仅管理员可修改;当前值:测试地址 {settings.test_weknora_base_url || '未配置'} · 生产地址 {settings.production_weknora_base_url || '未配置'} · 生产 ID {settings.production_tenant_id || '未配置'} · 知识库 ID {settings.production_knowledge_base_id || '未配置'}。</small>}
|
||||
</section> : null}
|
||||
{section === 'bindings' ? <RoutingBindingsSection /> : null}
|
||||
{section === 'storage' ? <section className="settings-panel wide storage-settings-panel" id="settings-storage">
|
||||
<div className="settings-section-heading"><div><h2>存储与备份</h2><p>文件资源与生产发布前快照使用独立受限配置。</p></div></div>
|
||||
<div className="storage-settings-grid"><dl><dt>Endpoint / Bucket</dt><dd>尚未配置</dd><dt>AccessKey / Secret</dt><dd>仅保存凭据引用,不可回读</dd><dt>CDN / 公开访问域名</dt><dd>尚未配置</dd><dt>临时签名有效期</dt><dd>待配置</dd></dl><dl><dt>测试备份前缀</dt><dd>/backup/weknora/test/</dd><dt>生产备份前缀</dt><dd>/backup/weknora/prod/</dd><dt>快照保留</dt><dd>最近 30 个生产快照</dd><dt>恢复验证</dt><dd>由生产门禁读回</dd></dl><ul className="target-list"><li>PostgreSQL <b>未配置</b></li><li>Elasticsearch <b>未配置</b></li><li>对象存储 <b>未配置</b></li><li>Neo4j <b>未配置</b></li><li>生产发布请求 <b>{productionEnabled ? '等待管理员核对' : '关闭'}</b></li><li>Release Worker <b>网页不可启用</b></li></ul></div>
|
||||
|
||||
@@ -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(<WeKnoraMonitorView {...baseProps} lane='attention' failures={failures} onRequeueFailure={vi.fn()} />)
|
||||
expect(html).toContain('源头取件')
|
||||
expect(html).toContain('关闭PC2远程桌面弹窗.docx')
|
||||
expect(html).toContain('重新入队')
|
||||
const busy=renderToStaticMarkup(<WeKnoraMonitorView {...baseProps} lane='attention' failures={failures} requeueingIDs={['jf-fetch']} />)
|
||||
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(<WeKnoraMonitorView {...baseProps} lane='attention' failures={failures} />)
|
||||
expect(html).toContain('上传解析')
|
||||
// 段落提示语里始终有“重新入队=再给它一次机会”,所以断言按钮本身不存在。
|
||||
expect(html).not.toContain('>重新入队</button>')
|
||||
})
|
||||
})
|
||||
|
||||
@@ -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 ? <p className="queue-batch-message" role="status">{batchMessage}</p> : null}
|
||||
{attentionItems.length ? attentionItems.map(actionTask) : emptyLane('没有失败或停滞的任务。')}
|
||||
<div className="section-head sub"><h3>死信(重试次数用完,系统不再自动处理)</h3><span className="tiny">“源头取件”=从钉钉/AList 把原始文件拉回本地;“上传解析”=把文件送进知识库做解析。死信只是失败任务的登记表,不会改动原始记录;重新入队=再给它一次机会</span></div>
|
||||
{failures.length ? <ul className="dlq-list">{failures.map((failure) => <li key={failure.id}><span className="monitor-status failed">{failure.error_code}</span><b>{failure.job_type === 'test.upload' ? '上传解析' : failure.job_type === 'artifact.fetch' ? '源头取件' : failure.job_type}</b>{failure.doc_name ? <b>「{failure.doc_name}」</b> : failure.publication_id ? <code>{failure.publication_id}</code> : null}<small>{failure.attempt}/{failure.max_attempts_at_failure} 次 · {new Date(failure.created_at).toLocaleString('zh-CN', { hour12: false })}</small><small>{failureHint(failure)}</small>{failure.publication_id ? <button disabled={retryingIDs.includes(failure.publication_id)} onClick={() => onBatchRetry([failure.publication_id as string])}>重新入队</button> : null}</li>)}</ul> : emptyLane('当前没有等待处理的死信记录。')}
|
||||
{failures.length ? <ul className="dlq-list">{failures.map((failure) => <li key={failure.id}><span className="monitor-status failed">{failure.error_code}</span><b>{failure.job_type === 'test.upload' ? '上传解析' : failure.job_type === 'artifact.fetch' ? '源头取件' : failure.job_type}</b>{failure.doc_name ? <b>「{failure.doc_name}」</b> : failure.publication_id ? <code>{failure.publication_id}</code> : null}<small>{failure.attempt}/{failure.max_attempts_at_failure} 次 · {new Date(failure.created_at).toLocaleString('zh-CN', { hour12: false })}</small><small>{failureHint(failure)}</small>{failure.job_type === 'artifact.fetch' ? <button disabled={requeueingIDs.includes(failure.id)} onClick={() => onRequeueFailure(failure.id)}>{requeueingIDs.includes(failure.id) ? '重排中…' : '重新入队'}</button> : failure.publication_id ? <button disabled={retryingIDs.includes(failure.publication_id)} onClick={() => onBatchRetry([failure.publication_id as string])}>重新入队</button> : null}</li>)}</ul> : emptyLane('当前没有等待处理的死信记录。')}
|
||||
</section> : null}
|
||||
{lane === 'removal' ? <section className="inbox-section" aria-label="需下架">
|
||||
<div className="section-head"><h2>需下架({laneCounts.removal})</h2><span className="tiny">来源目录里被改为“禁止导入”或撤销标注、但仍在知识库中可检索的文件</span></div>
|
||||
@@ -265,6 +269,7 @@ export function WeKnoraMonitor() {
|
||||
const [documentRefreshMs, setDocumentRefreshMs] = useState(15000)
|
||||
const [resourceRefreshMs, setResourceRefreshMs] = useState(30000)
|
||||
const [retryingIDs, setRetryingIDs] = useState<string[]>([])
|
||||
const [requeueingIDs, setRequeueingIDs] = useState<string[]>([])
|
||||
const [cancellingIDs, setCancellingIDs] = useState<string[]>([])
|
||||
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 <WeKnoraMonitorView overview={overview} files={files} queue={queue} plans={plans} selectedKB={selected} status={status} queueStatus={queueStatus} queueScope={queueScope} loading={loading} error={error} retryingIDs={retryingIDs} cancellingIDs={cancellingIDs} starting={starting} startMessage={startMessage} batchMessage={batchMessage} failures={failures} metrics={metrics} onSelectKB={(kb) => { 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 <WeKnoraMonitorView overview={overview} files={files} queue={queue} plans={plans} selectedKB={selected} status={status} queueStatus={queueStatus} queueScope={queueScope} loading={loading} error={error} retryingIDs={retryingIDs} requeueingIDs={requeueingIDs} cancellingIDs={cancellingIDs} starting={starting} startMessage={startMessage} batchMessage={batchMessage} failures={failures} metrics={metrics} onSelectKB={(kb) => { 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) }} />
|
||||
}
|
||||
|
||||
@@ -208,3 +208,30 @@ export async function fetchJobFailures(signal?: AbortSignal): Promise<JobFailure
|
||||
if (!isRecord(value) || !isRecord(value.data) || !Array.isArray(value.data.items)) throw new Error('失败队列响应无效')
|
||||
return value.data.items as JobFailure[]
|
||||
}
|
||||
|
||||
export interface RequeuedArtifactFetch { failure_id: string; job_id: string; version_id: string; round: number; doc_name?: string; already_queued: boolean }
|
||||
|
||||
// 源头取件死信重排:取件任务的幂等键绑在版本上,失败那次已占住它,所以重新标注
|
||||
// 同一文件不会再入队;这里让后端按新轮次重新排队(新键),并复位意图状态。
|
||||
export async function requeueJobFailure(failureId: string, reason: string): Promise<RequeuedArtifactFetch> {
|
||||
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<string, unknown>
|
||||
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,
|
||||
}
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user