feat(people): protoCache 全量重建改为分批可暂停执行

将原单次 buildClustProtoCache 全量重建重构为分批 full rebuild，降低 NAS（22万+ person）单次峰值负载，并支持 foreground active 时暂停/恢复：

- 新增 protoCacheRebuildJob 状态机（running/paused/completed/failed）：runBatchedFullRebuild 分页拉取全部 person ID（collectAllPersonIDsForRebuild + FaceRepository.ListAssignedPersonIDsPaged），按批构建原型（buildClustProtoCacheBatch，每批 200）累积到 staging，全部完成后原子切换 protoCache，避免中途半成品可见
- 批次间根据 BackgroundTaskCoordinator.LoadSnapshot 动态让行（空闲 250ms / CPU≥60·IO≥10·内存≥80 时 2s）；foreground active 时暂停 job 保留 cursor，worker loop 在前台空闲后继续推进剩余批次
- 重建期间立即对活动缓存 applyTombstonesToCache，避免已删除/合并 person 在重建过程中仍作为匹配目标
- 新增 ProtoCacheRebuildStatusResponse 状态字段（generation/state/cursor/total/batches/pause_reason/cold_building），peopleService.ProtoCacheRebuildStatus 经 rebuildSnapshot 加锁读取，后台状态 API（/background/status）暴露 rebuild 进度与冷启动标识
- SQLite busy/locked 上报 ReportDBBusy 进入冷却；失败丢弃 staging 保留旧缓存；stop 时取消进行中 job
- 补充 people_clustering_coordinator / face_repo / background_handler 单元测试覆盖分批、暂停恢复、tombstone、分页查询等路径

Author: davidhoo

backend/internal/api/v1/handler/background_handler.go

@@ -11,17 +11,22 @@ import (
 
 // BackgroundHandler 暴露后台任务治理的只读状态。
 type BackgroundHandler struct {
-	coordinator *service.BackgroundTaskCoordinator
-	loadSampler *service.BackgroundLoadSampler
+	coordinator   *service.BackgroundTaskCoordinator
+	loadSampler   *service.BackgroundLoadSampler
+	rebuildStatus func() *model.ProtoCacheRebuildStatusResponse
 }
 
 // NewBackgroundHandler 构造后台状态处理器。coordinator 可能为 nil（未注入时返回空快照），
-// 不应阻塞 API。
-func NewBackgroundHandler(coordinator *service.BackgroundTaskCoordinator, loadSampler *service.BackgroundLoadSampler) *BackgroundHandler {
-	return &BackgroundHandler{
+// 不应阻塞 API。rebuildStatus 可选，用于附加 protoCache rebuild 进度快照（nil 时省略）。
+func NewBackgroundHandler(coordinator *service.BackgroundTaskCoordinator, loadSampler *service.BackgroundLoadSampler, rebuildStatus ...func() *model.ProtoCacheRebuildStatusResponse) *BackgroundHandler {
+	h := &BackgroundHandler{
 		coordinator: coordinator,
 		loadSampler: loadSampler,
 	}
+	if len(rebuildStatus) > 0 {
+		h.rebuildStatus = rebuildStatus[0]
+	}
+	return h
 }
 
 // GetStatus 返回后台任务治理的只读快照。
@@ -33,6 +38,9 @@ func NewBackgroundHandler(coordinator *service.BackgroundTaskCoordinator, loadSa
 // @Router /api/v1/background/status [get]
 func (h *BackgroundHandler) GetStatus(c *gin.Context) {
 	resp := buildBackgroundStatusResponse(h.coordinator, h.loadSampler)
+	if h.rebuildStatus != nil {
+		resp.ProtoCacheRebuild = h.rebuildStatus()
+	}
 	c.JSON(http.StatusOK, model.Response{
 		Success: true,
 		Message: "后台任务状态",
@@ -76,7 +84,7 @@ func buildBackgroundStatusResponse(coord *service.BackgroundTaskCoordinator, sam
 		CPUPauseThreshold:    status.CPUPauseThreshold,
 		IOWaitPauseThreshold: status.IOWaitPauseThreshold,
 		MemoryPauseThreshold: status.MemoryPauseThreshold,
-		DBLockedCooldownMs:  status.DBLockedCooldownMs,
+		DBLockedCooldownMs:   status.DBLockedCooldownMs,
 	}
 	for _, r := range status.Running {
 		resp.Running = append(resp.Running, model.BackgroundTaskRuntimeResponse{

backend/internal/api/v1/handler/background_handler_test.go

@@ -64,4 +64,38 @@ func TestBackgroundHandler_GetStatus_NilCoordinatorNoPanic(t *testing.T) {
 	assert.True(t, status.AutoTasksEnabled)
 	// 负载字段为 unknown（-1）。
 	assert.Equal(t, -1.0, status.Load.CPUUserPct)
+	// 无 rebuild 时省略 proto_cache_rebuild 字段（向后兼容）。
+	assert.Nil(t, status.ProtoCacheRebuild, "proto_cache_rebuild should be omitted when no rebuild")
+}
+
+// TestBackgroundHandler_GetStatus_ProtoCacheRebuildSnapshot 验证注入 rebuildStatus 回调时，
+// 响应携带 protoCache rebuild 进度快照（含 cold_building 区分冷启动）。
+func TestBackgroundHandler_GetStatus_ProtoCacheRebuildSnapshot(t *testing.T) {
+	coord := service.NewBackgroundTaskCoordinator()
+	coord.SetBackgroundConfig(true, 70, 15, 85, nil, 120*time.Second)
+	rebuildStatus := func() *model.ProtoCacheRebuildStatusResponse {
+		return &model.ProtoCacheRebuildStatusResponse{
+			Generation:   42,
+			State:        "running",
+			ColdBuilding: true,
+			Cursor:       600,
+			Total:        2200,
+			Batches:      3,
+		}
+	}
+	h := NewBackgroundHandler(coord, nil, rebuildStatus)
+	rec := performJSONRequest(t, http.MethodGet, "/api/v1/background/status", nil, nil, h.GetStatus)
+	assert.Equal(t, http.StatusOK, rec.Code)
+
+	resp := decodeAPIResponse(t, rec)
+	assert.True(t, resp.Success)
+
+	var status model.BackgroundTaskStatusResponse
+	require.NoError(t, json.Unmarshal(toJSON(resp.Data), &status))
+	require.NotNil(t, status.ProtoCacheRebuild, "proto_cache_rebuild should be present when rebuildStatus set")
+	assert.Equal(t, uint64(42), status.ProtoCacheRebuild.Generation)
+	assert.Equal(t, "running", status.ProtoCacheRebuild.State)
+	assert.True(t, status.ProtoCacheRebuild.ColdBuilding, "cold_building should reflect cold-start build")
+	assert.Equal(t, 600, status.ProtoCacheRebuild.Cursor)
+	assert.Equal(t, 2200, status.ProtoCacheRebuild.Total)
 }

backend/internal/api/v1/handler/handler.go

@@ -42,7 +42,7 @@ func NewHandlers(db *gorm.DB, services *service.Services, repos *repository.Repo
 		Auth:       NewAuthHandler(services.Auth),
 		Analyzer:   NewAnalyzerHandler(services.Photo, services.Analysis, services.AnalysisRuntime),
 		Event:      NewEventHandler(services.EventClustering, repos.Event, db),
-		Background: NewBackgroundHandler(services.BackgroundCoordinator, services.BackgroundLoadSampler),
+		Background: NewBackgroundHandler(services.BackgroundCoordinator, services.BackgroundLoadSampler, services.ProtoCacheRebuildStatus),
 	}
 
 	// AI Handler - 即使 AI 服务未配置也创建，以便配置变更后动态更新

backend/internal/model/background_task.go

@@ -22,10 +22,26 @@ type BackgroundTaskStatusResponse struct {
 	Load BackgroundLoadSnapshotResponse `json:"load"`
 	// Thresholds 是 advisory 资源背压阈值（0 表示禁用）。
 	Thresholds BackgroundThresholdsResponse `json:"thresholds"`
+	// ProtoCacheRebuild 是 protoCache 分批 full rebuild 的进度快照。nil/省略表示当前没有
+	// rebuild 在进行（向后兼容：旧客户端忽略该字段即可）。cold_building 标识冷启动构建中。
+	ProtoCacheRebuild *ProtoCacheRebuildStatusResponse `json:"proto_cache_rebuild,omitempty"`
 	// CapturedAt 是快照采集时间（RFC3339）。
 	CapturedAt string `json:"captured_at"`
 }
 
+// ProtoCacheRebuildStatusResponse 描述一次 protoCache 分批 full rebuild 的只读进度快照。
+// 所有字段向后兼容。state 取值：idle/running/paused/completed/failed；cold_building 为
+// 独立布尔，当无可用旧缓存且 rebuild 进行中时为 true，用于前端区分冷启动与普通后台 refresh。
+type ProtoCacheRebuildStatusResponse struct {
+	Generation   uint64 `json:"generation"`
+	State        string `json:"state"`
+	ColdBuilding bool   `json:"cold_building"`
+	Cursor       int    `json:"cursor"`
+	Total        int    `json:"total"`
+	Batches      int    `json:"batches"`
+	PauseReason  string `json:"pause_reason,omitempty"`
+}
+
 // BackgroundTaskRuntimeResponse 描述一个正在运行的后台任务（脱敏，仅 class/dedupe/priority/started_at）。
 type BackgroundTaskRuntimeResponse struct {
 	Class     string `json:"class"`
@@ -54,7 +70,7 @@ type BackgroundThresholdsResponse struct {
 	CPUPauseThreshold    float64 `json:"cpu_pause_threshold"`
 	IOWaitPauseThreshold float64 `json:"iowait_pause_threshold"`
 	MemoryPauseThreshold float64 `json:"memory_pause_threshold"`
-	DBLockedCooldownMs  int64   `json:"db_locked_cooldown_ms"`
+	DBLockedCooldownMs   int64   `json:"db_locked_cooldown_ms"`
 }
 
 // FormatBackgroundTimeRFC3339 格式化时间为 RFC3339 字符串；零值返回空串。

backend/internal/repository/face_repo.go

@@ -42,6 +42,7 @@ type FaceRepository interface {
 	ListByIDs(ids []uint) ([]*model.Face, error)
 	ListAssigned() ([]*model.Face, error)
 	ListAssignedPersonIDs() ([]uint, error)
+	ListAssignedPersonIDsPaged(offset, limit int) ([]uint, error)
 	ListPending(limit int) ([]*model.Face, error)
 	GetPendingStats() (*PendingFaceStats, error)
 	ListPrototypeEmbeddings(personIDs []uint, perPerson int) ([]*model.Face, error)
@@ -186,6 +187,25 @@ func (r *faceRepository) ListAssignedPersonIDs() ([]uint, error) {
 	return ids, err
 }
 
+// ListAssignedPersonIDsPaged returns a page of distinct assigned person IDs
+// ordered by person_id ascending. offset is 0-based; limit is the page size.
+// Designed for batched protoCache rebuild to avoid loading all person IDs in
+// a single query on large datasets (NAS: 220K+ rows).
+func (r *faceRepository) ListAssignedPersonIDsPaged(offset, limit int) ([]uint, error) {
+	if limit <= 0 {
+		return nil, nil
+	}
+	var ids []uint
+	err := r.db.Model(&model.Face{}).
+		Where("person_id IS NOT NULL").
+		Distinct("person_id").
+		Order("person_id ASC").
+		Offset(offset).
+		Limit(limit).
+		Pluck("person_id", &ids).Error
+	return ids, err
+}
+
 func (r *faceRepository) ListPending(limit int) ([]*model.Face, error) {
 	var faces []*model.Face
 	// 退避策略：根据 retry_count 计算最小重试间隔

backend/internal/repository/face_repo_test.go

@@ -1,6 +1,7 @@
 package repository
 
 import (
+	"encoding/json"
 	"sync/atomic"
 	"testing"
 	"time"
@@ -579,3 +580,100 @@ func TestFaceRepository_ListPersonIDsSharingPhotos_NoNPlusOne(t *testing.T) {
 	assert.Less(t, atomic.LoadInt32(qcount), int32(600), "no per-candidate N+1; chunked and bounded")
 	assert.Greater(t, atomic.LoadInt32(qcount), int32(0))
 }
+
+func TestFaceRepository_ListAssignedPersonIDsPaged(t *testing.T) {
+	db := setupTestDB(t)
+	defer teardownTestDB(db)
+	faceRepo := NewFaceRepository(db)
+	personRepo := NewPersonRepository(db)
+
+	// Create 5 persons with faces assigned.
+	personIDs := make([]uint, 5)
+	for i := 0; i < 5; i++ {
+		p := &model.Person{Category: model.PersonCategoryFriend}
+		require.NoError(t, personRepo.Create(p))
+		personIDs[i] = p.ID
+		require.NoError(t, faceRepo.Create(&model.Face{
+			PhotoID:      uint(i + 1),
+			PersonID:     &p.ID,
+			BBoxX:        0.1, BBoxY: 0.1, BBoxWidth: 0.2, BBoxHeight: 0.2,
+			Confidence:   0.9, QualityScore: 0.8,
+		}))
+	}
+	// Also create an unassigned face — should not appear.
+	require.NoError(t, faceRepo.Create(&model.Face{PhotoID: 99, BBoxX: 0.1, BBoxY: 0.1, BBoxWidth: 0.2, BBoxHeight: 0.2}))
+
+	// Page 1: offset=0, limit=3 → first 3 person IDs (ascending).
+	page1, err := faceRepo.ListAssignedPersonIDsPaged(0, 3)
+	require.NoError(t, err)
+	assert.Len(t, page1, 3)
+	assert.Equal(t, personIDs[0], page1[0])
+	assert.Equal(t, personIDs[1], page1[1])
+	assert.Equal(t, personIDs[2], page1[2])
+
+	// Page 2: offset=3, limit=3 → last 2 person IDs.
+	page2, err := faceRepo.ListAssignedPersonIDsPaged(3, 3)
+	require.NoError(t, err)
+	assert.Len(t, page2, 2)
+	assert.Equal(t, personIDs[3], page2[0])
+	assert.Equal(t, personIDs[4], page2[1])
+
+	// Page 3: offset=5, limit=3 → empty.
+	page3, err := faceRepo.ListAssignedPersonIDsPaged(5, 3)
+	require.NoError(t, err)
+	assert.Empty(t, page3)
+
+	// limit=0 → empty (no error).
+	empty, err := faceRepo.ListAssignedPersonIDsPaged(0, 0)
+	require.NoError(t, err)
+	assert.Empty(t, empty)
+}
+
+func TestFaceRepository_ListPrototypeEmbeddings_Batched(t *testing.T) {
+	db := setupTestDB(t)
+	defer teardownTestDB(db)
+	faceRepo := NewFaceRepository(db)
+	personRepo := NewPersonRepository(db)
+
+	// Create 2 persons, each with 3 faces (different quality).
+	emb := encodeFloat32(t, []float32{1.0, 0.0, 0.0})
+	for pid := 1; pid <= 2; pid++ {
+		p := &model.Person{Category: model.PersonCategoryFriend}
+		require.NoError(t, personRepo.Create(p))
+		for f := 0; f < 3; f++ {
+			require.NoError(t, faceRepo.Create(&model.Face{
+				PhotoID:      uint(pid*10 + f),
+				PersonID:     &p.ID,
+				BBoxX:        0.1, BBoxY: 0.1, BBoxWidth: 0.2, BBoxHeight: 0.2,
+				Confidence:   0.9,
+				QualityScore: float64(3 - f), // 3, 2, 1
+				Embedding:    emb,
+			}))
+		}
+	}
+
+	personIDs, err := faceRepo.ListAssignedPersonIDsPaged(0, 10)
+	require.NoError(t, err)
+	assert.Len(t, personIDs, 2)
+
+	// Request top 2 per person → 4 faces total.
+	faces, err := faceRepo.ListPrototypeEmbeddings(personIDs, 2)
+	require.NoError(t, err)
+	assert.Len(t, faces, 4) // 2 persons × 2 per person
+
+	// Verify each person got 2 faces (highest quality first).
+	byPerson := make(map[uint]int)
+	for _, f := range faces {
+		byPerson[*f.PersonID]++
+	}
+	for _, count := range byPerson {
+		assert.Equal(t, 2, count)
+	}
+}
+
+func encodeFloat32(t *testing.T, vals []float32) []byte {
+	t.Helper()
+	payload, err := json.Marshal(vals)
+	require.NoError(t, err)
+	return payload
+}

backend/internal/service/background_task_coordinator.go

@@ -44,24 +44,24 @@ const (
 //   - MaxRuntime：预留字段，第一波仅记录到 decision，不强制 kill（执行预算由各 worker
 //     自身的 checkpoint 保证）。
 type BackgroundTaskRequest struct {
-	Class     BackgroundTaskClass
-	Priority  BackgroundTaskPriority
-	DedupeKey string
+	Class      BackgroundTaskClass
+	Priority   BackgroundTaskPriority
+	DedupeKey  string
 	MaxRuntime time.Duration
 }
 
 // BackgroundDecisionReason 是准入决策的原因码，用于可观测性与状态 API。
 type BackgroundDecisionReason string
 
 const (
-	BackgroundDecisionAllowed          BackgroundDecisionReason = "allowed"
-	BackgroundDecisionCoalesced        BackgroundDecisionReason = "coalesced"
-	BackgroundDecisionForeground       BackgroundDecisionReason = "foreground_active"
-	BackgroundDecisionCooldown         BackgroundDecisionReason = "cooldown"
-	BackgroundDecisionAlreadyRunning   BackgroundDecisionReason = "already_running"
-	BackgroundDecisionCPUHigh          BackgroundDecisionReason = "cpu_high"
-	BackgroundDecisionIOWaitHigh       BackgroundDecisionReason = "iowait_high"
-	BackgroundDecisionMemoryHigh       BackgroundDecisionReason = "memory_high"
+	BackgroundDecisionAllowed           BackgroundDecisionReason = "allowed"
+	BackgroundDecisionCoalesced         BackgroundDecisionReason = "coalesced"
+	BackgroundDecisionForeground        BackgroundDecisionReason = "foreground_active"
+	BackgroundDecisionCooldown          BackgroundDecisionReason = "cooldown"
+	BackgroundDecisionAlreadyRunning    BackgroundDecisionReason = "already_running"
+	BackgroundDecisionCPUHigh           BackgroundDecisionReason = "cpu_high"
+	BackgroundDecisionIOWaitHigh        BackgroundDecisionReason = "iowait_high"
+	BackgroundDecisionMemoryHigh        BackgroundDecisionReason = "memory_high"
 	BackgroundDecisionAutomaticDisabled BackgroundDecisionReason = "automatic_disabled"
 )
 
@@ -103,9 +103,9 @@ type BackgroundTaskDedupeEntry struct {
 // dedupeSlot 记录一个 (class, dedupeKey) 的 running/pending 状态。同一 slot 同时最多
 // 一个 running + 一个 pending；超过的请求被 coalesce 拒绝。
 type dedupeSlot struct {
-	running  bool
-	pending  bool
-	priority BackgroundTaskPriority
+	running   bool
+	pending   bool
+	priority  BackgroundTaskPriority
 	startedAt time.Time
 }
 
@@ -205,6 +205,18 @@ func (c *BackgroundTaskCoordinator) ForegroundActive() bool {
 	return c.foregroundActive > 0
 }
 
+// LoadSnapshot 返回当前系统负载快照。loadFn 未注入时返回零值（所有字段 -1=unknown）。
+// 供 protoCache rebuild 等后台任务做动态速度控制。
+func (c *BackgroundTaskCoordinator) LoadSnapshot() BackgroundLoadSnapshot {
+	c.mu.Lock()
+	fn := c.loadFn
+	c.mu.Unlock()
+	if fn == nil {
+		return BackgroundLoadSnapshot{CPUUserPct: -1, CPUIOWaitPct: -1, MemUsedPct: -1}
+	}
+	return fn()
+}
+
 // CanRun 评估一次后台任务请求是否可以运行，不占用 slot。用于调用方在启动重工作前快速
 // 检查。返回的 decision 不改变 coordinator 状态。
 //
@@ -227,6 +239,7 @@ func (c *BackgroundTaskCoordinator) CanRun(req BackgroundTaskRequest) (Backgroun
 // 对于 DedupeKey 非空的 automatic 请求，若当前已有 running：
 //   - 若尚无 pending → 记录一个 pending slot 并返回 coalesced（调用方可选择稍后重试）；
 //   - 若已有 pending → 返回 coalesced，不增加 pending（至多一 pending）。
+//
 // running 释放时不会自动触发 pending（pending 仅作为“有积压”的可观测标记；实际重试由
 // worker 调度循环驱动，避免引入任务队列语义）。
 func (c *BackgroundTaskCoordinator) Begin(req BackgroundTaskRequest) (func(), BackgroundTaskDecision, bool) {
@@ -414,7 +427,7 @@ type BackgroundTaskStatus struct {
 	CPUPauseThreshold    float64
 	IOWaitPauseThreshold float64
 	MemoryPauseThreshold float64
-	DBLockedCooldownMs  int64
+	DBLockedCooldownMs   int64
 }
 
 // Status 返回 Snapshot + 配置/阈值的完整状态视图（供状态 API，Task 16）。
@@ -431,7 +444,7 @@ func (c *BackgroundTaskCoordinator) Status() BackgroundTaskStatus {
 		CPUPauseThreshold:    c.cpuPauseThreshold,
 		IOWaitPauseThreshold: c.iowaitPauseThreshold,
 		MemoryPauseThreshold: c.memoryPauseThreshold,
-		DBLockedCooldownMs:  c.dbLockedCooldown.Milliseconds(),
+		DBLockedCooldownMs:   c.dbLockedCooldown.Milliseconds(),
 	}
 	now := time.Now()
 	for class, until := range c.cooldowns {