fix(server): index generation error handle (#15532)

<!--
Thank you for contributing to AFFiNE!

The PR title must follow Conventional Commits (enforced by CI):
type(scope): description e.g. fix(editor): keep selection after paste
Types: feat fix docs style refactor perf test build ci chore revert
-->

## Description

<!-- What does this PR do? Link related issues, e.g. "Closes #1234".
Screenshots or recordings are welcome for UI changes. -->

## Checklist

- [ ] I have signed the [AFFiNE Contributor License
Agreement](https://cla-assistant.io/toeverything/AFFiNE) — required
before merge; the `license/cla` check must be green ([how it
works](https://github.com/toeverything/AFFiNE/blob/canary/docs/BUILDING.md#sign-the-cla-first))
- [ ] The PR targets the `canary` branch and its title follows
[Conventional Commits](https://www.conventionalcommits.org/)
- [ ] Tests are added or updated where it makes sense
- [ ] `yarn lint` and `yarn typecheck` pass locally



#### PR Dependency Tree


* **PR #15532** 👈

This tree was auto-generated by
[Charcoal](https://github.com/danerwilliams/charcoal)

<!-- This is an auto-generated comment: release notes by coderabbit.ai
-->
## Summary by CodeRabbit

* **Bug Fixes**
* Improved search reconciliation and publication status handling when
workspace reconciliation fails.
* Prevented failed workspace reconciliation from incorrectly blocking
generation completion.
* Preserved active generation state so reconciliation can retry and
complete pending publications.
* Improved managed provider profile migration, including legacy
configurations, unavailable models, conflicting assignments, and missing
defaults.

* **Tests**
* Expanded coverage for workspace recovery and managed provider profile
migration scenarios.
<!-- end of auto-generated comment: release notes by coderabbit.ai -->

packages/backend/native/src/runtime/backend_runtime/search/generation.rs

@@ -3,7 +3,7 @@ use sha2::{Digest, Sha256};
 use sqlx::{PgPool, Row};
 use uuid::Uuid;
 
-use super::{SCHEMA_FINGERPRINT, SearchProvider, SearchTable};
+use super::{SCHEMA_FINGERPRINT, SearchProvider, SearchTable, WORKSPACE_RECONCILE_FAILED};
 use crate::{
   runtime::{RuntimeError, RuntimeResult, SearchRuntimeConfig},
   search_index::EmbeddedSearchIndex,
@@ -311,6 +311,7 @@ pub(super) async fn activate(
                 SELECT 1
                 FROM search_projection.workspace_states state
                 WHERE state.generation_id=generation.id
+                  AND state.last_error IS DISTINCT FROM $2
                   AND (
                     NOT state.covered
                     OR state.required_permission_version > state.applied_permission_version
@@ -331,6 +332,12 @@ pub(super) async fn activate(
               NOT EXISTS (
                 SELECT 1 FROM search_projection.document_states state
                 WHERE state.generation_id=generation.id
+                  AND NOT EXISTS (
+                    SELECT 1 FROM search_projection.workspace_states workspace
+                    WHERE workspace.generation_id=state.generation_id
+                      AND workspace.workspace_id=state.workspace_id
+                      AND workspace.last_error=$2
+                  )
                   AND (state.target_source_version <> state.published_source_version
                     OR state.target_source_exists <> state.published_source_exists
                     OR state.target_permission_version <> state.published_permission_version)
@@ -340,6 +347,7 @@ pub(super) async fn activate(
        FOR UPDATE"#,
   )
   .bind(generation.id)
+  .bind(WORKSPACE_RECONCILE_FAILED)
   .fetch_optional(&mut *transaction)
   .await
   .map_err(|error| RuntimeError::database("load search generation activation state", error))?;

packages/backend/native/src/runtime/backend_runtime/search/mod.rs

@@ -24,6 +24,7 @@ use super::{
 };
 
 const SCHEMA_FINGERPRINT: i32 = 1;
+const WORKSPACE_RECONCILE_FAILED: &str = "search_workspace_reconcile_failed";
 
 #[cfg(test)]
 pub(crate) static SEARCH_TEST_LOCK: tokio::sync::Mutex<()> = tokio::sync::Mutex::const_new(());

packages/backend/native/src/runtime/backend_runtime/search/runtime.rs

@@ -8,8 +8,8 @@ use tokio::{
 use uuid::Uuid;
 
 use super::{
-  ActiveGeneration, PermissionAuthorizer, SearchProvider, activate, cleanup_retired_generation, config_hash, ensure,
-  load_active, reconcile_workspace, sweep_generation_orphans,
+  ActiveGeneration, PermissionAuthorizer, SearchProvider, WORKSPACE_RECONCILE_FAILED, activate,
+  cleanup_retired_generation, config_hash, ensure, load_active, reconcile_workspace, sweep_generation_orphans,
 };
 use crate::{
   runtime::{RuntimeError, RuntimeResult, SearchRuntimeConfig},
@@ -212,12 +212,16 @@ impl SearchRuntime {
       .await
       .map_err(|error| RuntimeError::database("count search workspaces", error))?;
     let pending: bool = sqlx::query_scalar(
-      "SELECT EXISTS (SELECT 1 FROM search_projection.workspace_states WHERE generation_id=$1 AND (NOT covered OR \
-       pending_scope <> 'none')) OR EXISTS (SELECT 1 FROM search_projection.document_states WHERE generation_id=$1 \
-       AND (target_source_version <> published_source_version OR target_source_exists <> published_source_exists OR \
-       target_permission_version <> published_permission_version))",
+      "SELECT EXISTS (SELECT 1 FROM search_projection.workspace_states WHERE generation_id=$1 AND last_error IS \
+       DISTINCT FROM $2 AND (NOT covered OR pending_scope <> 'none')) OR EXISTS (SELECT 1 FROM \
+       search_projection.document_states document WHERE generation_id=$1 AND NOT EXISTS (SELECT 1 FROM \
+       search_projection.workspace_states workspace WHERE workspace.generation_id=document.generation_id AND \
+       workspace.workspace_id=document.workspace_id AND workspace.last_error=$2) AND (target_source_version <> \
+       published_source_version OR target_source_exists <> published_source_exists OR target_permission_version <> \
+       published_permission_version))",
     )
     .bind(generation.id)
+    .bind(WORKSPACE_RECONCILE_FAILED)
     .fetch_one(&self.pool)
     .await
     .map_err(|error| RuntimeError::database("check pending search projection", error))?;
@@ -318,6 +322,12 @@ impl SearchRuntime {
       r#"SELECT id,provider,state,scan_cursor_sid,scan_high_water_sid,
                 (SELECT count(*) FROM search_projection.document_states document
                  WHERE document.generation_id=generations.id
+                   AND NOT EXISTS (
+                     SELECT 1 FROM search_projection.workspace_states workspace
+                     WHERE workspace.generation_id=document.generation_id
+                       AND workspace.workspace_id=document.workspace_id
+                       AND workspace.last_error=$4
+                   )
                    AND (document.target_source_version <> document.published_source_version
                      OR document.target_source_exists <> document.published_source_exists
                      OR document.target_permission_version <> document.published_permission_version)) AS pending_publications,
@@ -332,6 +342,7 @@ impl SearchRuntime {
     .bind(&self.config.provider)
     .bind(config_hash(&self.config))
     .bind(super::SCHEMA_FINGERPRINT)
+    .bind(WORKSPACE_RECONCILE_FAILED)
     .fetch_optional(&self.pool)
     .await
     .map_err(|error| RuntimeError::database("load search projection status", error))?;

packages/backend/native/src/runtime/backend_runtime/search/worker/mod.rs

@@ -36,6 +36,6 @@ use workspace_state::{
 };
 
 use super::{
-  ActiveGeneration, ProjectionInput, SearchChange, SearchProvider, SearchTable, project_document,
-  projection_external_id, provider_payload,
+  ActiveGeneration, ProjectionInput, SearchChange, SearchProvider, SearchTable, WORKSPACE_RECONCILE_FAILED,
+  project_document, projection_external_id, provider_payload,
 };

packages/backend/native/src/runtime/backend_runtime/search/worker/workspace_state.rs

@@ -3,7 +3,7 @@ use serde_json::Value;
 use sqlx::{PgPool, Row};
 use uuid::Uuid;
 
-use super::{ActiveGeneration, LEASE_SECONDS, SearchTable};
+use super::{ActiveGeneration, LEASE_SECONDS, SearchTable, WORKSPACE_RECONCILE_FAILED};
 use crate::runtime::{RuntimeError, RuntimeResult};
 
 const ANTI_ENTROPY_INTERVAL_SECONDS: i64 = 3600;
@@ -350,29 +350,9 @@ pub(super) async fn mark_workspace_failed(
     .begin()
     .await
     .map_err(|error| RuntimeError::database("begin failed search workspace update", error))?;
-  let generation_state: Option<String> = sqlx::query_scalar(
-    r#"SELECT generation.state
-       FROM search_projection.workspace_states state
-       JOIN search_projection.generations generation ON generation.id=state.generation_id
-       WHERE state.generation_id=$1 AND state.workspace_id=$2 AND state.claim_fence=$3
-       FOR UPDATE"#,
-  )
-  .bind(generation_id)
-  .bind(workspace_id)
-  .bind(fence)
-  .fetch_optional(&mut *transaction)
-  .await
-  .map_err(|error| RuntimeError::database("load failed search workspace claim", error))?;
-  let Some(generation_state) = generation_state else {
-    transaction
-      .commit()
-      .await
-      .map_err(|error| RuntimeError::database("commit empty search workspace update", error))?;
-    return Ok(());
-  };
-  sqlx::query(
+  let updated = sqlx::query(
     r#"UPDATE search_projection.workspace_states
-       SET covered=false, pending_scope='workspace', last_error='search_workspace_reconcile_failed',
+       SET covered=false, pending_scope='workspace', last_error=$4,
            progress=NULL, available_at='infinity'::timestamptz,
            lease_owner=NULL, lease_expires_at=NULL, updated_at=now()
        WHERE generation_id=$1 AND workspace_id=$2 AND claim_fence=$3"#,
@@ -380,9 +360,17 @@ pub(super) async fn mark_workspace_failed(
   .bind(generation_id)
   .bind(workspace_id)
   .bind(fence)
+  .bind(WORKSPACE_RECONCILE_FAILED)
   .execute(&mut *transaction)
   .await
   .map_err(|error| RuntimeError::database("mark search workspace failed", error))?;
+  if updated.rows_affected() == 0 {
+    transaction
+      .commit()
+      .await
+      .map_err(|error| RuntimeError::database("commit empty search workspace update", error))?;
+    return Ok(());
+  }
   sqlx::query(
     r#"UPDATE search_projection.document_states
        SET available_at='infinity'::timestamptz, lease_owner=NULL, lease_expires_at=NULL, updated_at=now()
@@ -393,17 +381,6 @@ pub(super) async fn mark_workspace_failed(
   .execute(&mut *transaction)
   .await
   .map_err(|error| RuntimeError::database("pause failed search document publications", error))?;
-  if generation_state != "active" {
-    sqlx::query(
-      r#"UPDATE search_projection.generations
-         SET state='failed', last_error='search_workspace_reconcile_failed'
-         WHERE id=$1 AND state='building'"#,
-    )
-    .bind(generation_id)
-    .execute(&mut *transaction)
-    .await
-    .map_err(|error| RuntimeError::database("mark search generation failed", error))?;
-  }
   transaction
     .commit()
     .await
@@ -416,7 +393,11 @@ mod tests {
   use serde_json::json;
 
   use super::*;
-  use crate::runtime::{backend_runtime::search::SEARCH_TEST_LOCK, migrations::migrate_search_tables};
+  use crate::runtime::{
+    SearchRuntimeConfig,
+    backend_runtime::search::{SEARCH_TEST_LOCK, SearchRuntime, config_hash},
+    migrations::migrate_search_tables,
+  };
 
   #[test]
   fn progress_round_trips_versioned_context_for_each_phase() {
@@ -465,6 +446,7 @@ mod tests {
     migrate_search_tables(&pool).await.unwrap();
     let generation_id = Uuid::new_v4();
     let workspace_id = format!("claim-workspace-{}", Uuid::new_v4().simple());
+    let config = SearchRuntimeConfig::default();
     sqlx::query("INSERT INTO workspaces(id) VALUES($1)")
       .bind(&workspace_id)
       .execute(&pool)
@@ -485,10 +467,12 @@ mod tests {
     .await
     .unwrap();
     sqlx::query(
-      r#"INSERT INTO search_projection.generations(id,provider,state,config_hash,schema_version)
-         VALUES($1,'embedded','failed',decode(repeat('00',32),'hex'),1)"#,
+      r#"INSERT INTO search_projection.generations(
+           id,provider,state,config_hash,schema_version,scan_high_water_sid,scan_cursor_sid
+         ) VALUES($1,'embedded','building',$2,1,0,0)"#,
     )
     .bind(generation_id)
+    .bind(config_hash(&config))
     .execute(&pool)
     .await
     .unwrap();
@@ -536,14 +520,9 @@ mod tests {
     checkpoint_workspace(&pool, generation_id, &workspace_id, first.fence, checkpoint)
       .await
       .unwrap();
-    sqlx::query("UPDATE snapshots SET updated_at='2026-01-02 UTC' WHERE workspace_id=$1 AND guid=$1")
-      .bind(&workspace_id)
-      .execute(&pool)
-      .await
-      .unwrap();
     sqlx::query(
-      "UPDATE search_projection.workspace_states SET required_permission_version=4 WHERE generation_id=$1 AND \
-       workspace_id=$2",
+      "UPDATE search_projection.workspace_states SET target_root_revision=target_root_revision+1, \
+       required_permission_version=4 WHERE generation_id=$1 AND workspace_id=$2",
     )
     .bind(generation_id)
     .bind(&workspace_id)
@@ -557,12 +536,25 @@ mod tests {
       .unwrap();
     assert_eq!(second.progress.captured_root_revision, initial_root);
     assert_eq!(second.progress.captured_permission_version, 3);
+    sqlx::query(
+      r#"INSERT INTO search_projection.document_states(
+           generation_id,workspace_id,doc_id,target_source_version,target_source_exists
+         ) VALUES($1,$2,'pending-doc',1,true)"#,
+    )
+    .bind(generation_id)
+    .bind(&workspace_id)
+    .execute(&pool)
+    .await
+    .unwrap();
     mark_workspace_failed(&pool, generation_id, &workspace_id, second.fence)
       .await
       .unwrap();
-    let failed: (bool, Option<String>, Option<Value>, bool) = sqlx::query_as(
-      r#"SELECT covered,last_error,progress,available_at='infinity'::timestamptz
-         FROM search_projection.workspace_states WHERE generation_id=$1 AND workspace_id=$2"#,
+    let failed: (bool, Option<String>, Option<Value>, bool, String) = sqlx::query_as(
+      r#"SELECT state.covered,state.last_error,state.progress,state.available_at='infinity'::timestamptz,
+                generation.state
+         FROM search_projection.workspace_states state
+         JOIN search_projection.generations generation ON generation.id=state.generation_id
+         WHERE state.generation_id=$1 AND state.workspace_id=$2"#,
     )
     .bind(generation_id)
     .bind(&workspace_id)
@@ -571,9 +563,28 @@ mod tests {
     .unwrap();
     assert_eq!(
       failed,
-      (false, Some("search_workspace_reconcile_failed".to_string()), None, true)
+      (
+        false,
+        Some(WORKSPACE_RECONCILE_FAILED.to_string()),
+        None,
+        true,
+        "building".to_string()
+      )
     );
 
+    let runtime = SearchRuntime::new(pool.clone(), config).unwrap();
+    runtime.embedded.prepare_generation(generation_id).await;
+    assert_eq!(runtime.reconcile_pending(1).await.unwrap(), 0);
+    assert_eq!(
+      sqlx::query_scalar::<_, String>("SELECT state FROM search_projection.generations WHERE id=$1")
+        .bind(generation_id)
+        .fetch_one(&pool)
+        .await
+        .unwrap(),
+      "active"
+    );
+    assert_eq!(runtime.status().await.unwrap()["metrics"]["pendingPublications"], 0);
+
     sqlx::query("DELETE FROM search_projection.generations WHERE id=$1")
       .bind(generation_id)
       .execute(&pool)

packages/backend/native/src/runtime/migrations.rs

@@ -293,6 +293,7 @@ mod tests {
     assert!(!SEARCH_PROJECTION_MIGRATION.contains("clock_timestamp()"));
     assert!(!SEARCH_PROJECTION_MIGRATION.contains("CREATE TABLE search_runtime_projections"));
     assert!(!SEARCH_PROJECTION_MIGRATION.contains("payload JSONB NOT NULL"));
+    assert!(SEARCH_PROJECTION_MIGRATION.contains("search_workspace_reconcile_failed"));
   }
 
   #[tokio::test]

packages/backend/native/src/runtime/sql/search_projection.sql

@@ -266,16 +266,23 @@ RETURNS boolean LANGUAGE SQL STABLE AS $$
   AND NOT EXISTS (
     SELECT 1 FROM search_projection.workspace_states
     WHERE generation_id = target_generation
+      AND last_error IS DISTINCT FROM 'search_workspace_reconcile_failed'
       AND (NOT covered OR pending_scope <> 'none'
         OR required_permission_version > applied_permission_version
         OR last_error IS NOT NULL)
   )
   AND NOT EXISTS (
-    SELECT 1 FROM search_projection.document_states
-    WHERE generation_id = target_generation
-      AND (target_source_version <> published_source_version
-        OR target_source_exists <> published_source_exists
-        OR target_permission_version <> published_permission_version)
+    SELECT 1 FROM search_projection.document_states document
+    WHERE document.generation_id = target_generation
+      AND NOT EXISTS (
+        SELECT 1 FROM search_projection.workspace_states workspace
+        WHERE workspace.generation_id = document.generation_id
+          AND workspace.workspace_id = document.workspace_id
+          AND workspace.last_error = 'search_workspace_reconcile_failed'
+      )
+      AND (document.target_source_version <> document.published_source_version
+        OR document.target_source_exists <> document.published_source_exists
+        OR document.target_permission_version <> document.published_permission_version)
   )
 $$;