From bd095495dad2bbf9cd6649e8dbf3c2ca1183d96a Mon Sep 17 00:00:00 2001 From: DarkSky <25152247+darkskygit@users.noreply.github.com> Date: Wed, 26 Aug 2026 15:57:06 +0800 Subject: [PATCH] fix(server): index & gc queue & llm compatibility (#15528) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit fix #15523 fix #15526 #### PR Dependency Tree * **PR #15528** 👈 This tree was auto-generated by [Charcoal](https://github.com/danerwilliams/charcoal) ## Summary by CodeRabbit - **Bug Fixes** - Improved search generation cleanup, provider error reporting, and reconciliation reliability. - Retired search resources are cleaned up safely, including after credential changes. - Prompt size checks now ignore tool parameters and provide clearer errors. - Reserved documents are protected from accidental cleanup, and malformed identifiers are rejected. - **Configuration** - Managed Copilot profiles require explicit, non-duplicated model assignments. - Improved managed provider profile migration. - **Performance & Reliability** - Reduced unnecessary search-history cleanup and adjusted consistency-check intervals. - Failed reconciliation jobs stop after one attempt and are removed automatically. - **Data Updates** - Updated legacy AI session prompt names to current labels. - Improved cloud load-balancer health-check configuration. --- .../charts/front/templates/service-sync.yaml | 4 +- .github/helm/affine/charts/front/values.yaml | 6 +- .../helm/affine/templates/backend-config.yaml | 32 +++ .github/helm/affine/values.yaml | 6 +- .../helm/separate-config/backend-config.yaml | 10 - Cargo.lock | 10 +- packages/backend/native/index.d.ts | 2 +- packages/backend/native/src/lib.rs | 5 +- .../native/src/llm/core/prompt/session.rs | 28 +-- packages/backend/native/src/reserved_doc.rs | 54 +++++ .../backend_runtime/search/generation.rs | 217 +++++++++++++++++- .../src/runtime/backend_runtime/search/mod.rs | 2 +- .../search/provider/manticore.rs | 27 ++- .../backend_runtime/search/provider/mod.rs | 7 + .../backend_runtime/search/provider/remote.rs | 74 +++++- .../runtime/backend_runtime/search/runtime.rs | 13 +- .../backend_runtime/search/worker/document.rs | 63 ----- .../search/worker/workspace_state.rs | 2 +- packages/backend/native/src/runtime/config.rs | 81 +++---- .../storage_runtime/document_cleanup.rs | 75 +++++- .../backend/native/src/search_index/mod.rs | 4 + packages/backend/native/src/userdata_acl.rs | 45 ++-- .../migration.sql | 20 ++ .../server/src/__tests__/sync/gateway.spec.ts | 11 + .../server/src/core/backend-runtime/job.ts | 6 +- .../backend/server/src/core/sync/gateway.ts | 21 +- .../src/data/__tests__/migrations.spec.ts | 33 ++- ...0000-converge-managed-provider-profiles.ts | 31 ++- packages/backend/server/src/native.ts | 4 +- .../plugins/indexer/__tests__/service.spec.ts | 1 + 30 files changed, 691 insertions(+), 203 deletions(-) create mode 100644 .github/helm/affine/templates/backend-config.yaml delete mode 100644 .github/helm/separate-config/backend-config.yaml create mode 100644 packages/backend/native/src/reserved_doc.rs create mode 100644 packages/backend/server/migrations/20260826120000_migrate_legacy_copilot_prompt_names/migration.sql diff --git a/.github/helm/affine/charts/front/templates/service-sync.yaml b/.github/helm/affine/charts/front/templates/service-sync.yaml index dfdbb0d6be..6db650a498 100644 --- a/.github/helm/affine/charts/front/templates/service-sync.yaml +++ b/.github/helm/affine/charts/front/templates/service-sync.yaml @@ -4,8 +4,10 @@ metadata: name: {{ .Values.services.sync.name }} labels: {{- include "front.labels" . | nindent 4 }} - {{- with .Values.services.sync.annotations }} + {{- $annotations := omit .Values.services.sync.annotations "cloud.google.com/backend-config" }} annotations: + cloud.google.com/backend-config: {{ dict "default" .Values.services.sync.backendConfig.name | toJson | quote }} + {{- with $annotations }} {{- toYaml . | nindent 4 }} {{- end }} spec: diff --git a/.github/helm/affine/charts/front/values.yaml b/.github/helm/affine/charts/front/values.yaml index 4b4ba7cc83..c53d4d2206 100644 --- a/.github/helm/affine/charts/front/values.yaml +++ b/.github/helm/affine/charts/front/values.yaml @@ -52,8 +52,10 @@ services: name: affine-sync type: ClusterIP port: 3010 - annotations: - cloud.google.com/backend-config: '{"default": "affine-api-backendconfig"}' + backendConfig: + name: affine-sync-backendconfig + timeoutSec: 10800 + annotations: {} renderer: name: affine-renderer type: ClusterIP diff --git a/.github/helm/affine/templates/backend-config.yaml b/.github/helm/affine/templates/backend-config.yaml new file mode 100644 index 0000000000..652b9feb17 --- /dev/null +++ b/.github/helm/affine/templates/backend-config.yaml @@ -0,0 +1,32 @@ +{{- if eq .Values.global.deployment.platform "gcp" }} +apiVersion: cloud.google.com/v1 +kind: BackendConfig +metadata: + name: affine-api-backendconfig + annotations: + meta.helm.sh/release-name: {{ .Release.Name }} + meta.helm.sh/release-namespace: {{ .Release.Namespace }} + labels: + {{- include "affine.labels" . | nindent 4 }} +spec: + healthCheck: + timeoutSec: 1 + type: HTTP + requestPath: /info +--- +apiVersion: cloud.google.com/v1 +kind: BackendConfig +metadata: + name: {{ .Values.front.services.sync.backendConfig.name }} + annotations: + meta.helm.sh/release-name: {{ .Release.Name }} + meta.helm.sh/release-namespace: {{ .Release.Namespace }} + labels: + {{- include "affine.labels" . | nindent 4 }} +spec: + timeoutSec: {{ .Values.front.services.sync.backendConfig.timeoutSec }} + healthCheck: + timeoutSec: 1 + type: HTTP + requestPath: /info +{{- end }} diff --git a/.github/helm/affine/values.yaml b/.github/helm/affine/values.yaml index a04234e12e..5f82e7103b 100644 --- a/.github/helm/affine/values.yaml +++ b/.github/helm/affine/values.yaml @@ -52,8 +52,10 @@ front: name: affine-sync type: ClusterIP port: 3010 - annotations: - cloud.google.com/backend-config: '{"default": "affine-api-backendconfig"}' + backendConfig: + name: affine-sync-backendconfig + timeoutSec: 10800 + annotations: {} renderer: name: affine-renderer type: ClusterIP diff --git a/.github/helm/separate-config/backend-config.yaml b/.github/helm/separate-config/backend-config.yaml deleted file mode 100644 index cef9593179..0000000000 --- a/.github/helm/separate-config/backend-config.yaml +++ /dev/null @@ -1,10 +0,0 @@ -apiVersion: cloud.google.com/v1 -kind: BackendConfig -metadata: - name: "affine-api-backendconfig" -spec: - healthCheck: - timeoutSec: 1 - type: HTTP - requestPath: /info - diff --git a/Cargo.lock b/Cargo.lock index 03d420c6f1..3b521bc16b 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -2956,9 +2956,9 @@ dependencies = [ [[package]] name = "generator" -version = "0.8.8" +version = "0.8.9" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "52f04ae4152da20c76fe800fa48659201d5cf627c5149ca0b707b69d7eef6cf9" +checksum = "b3b854b0e584ead1a33f18b2fcad7cf7be18b3875c78816b753639aa501513ae" dependencies = [ "cc", "cfg-if", @@ -2966,7 +2966,7 @@ dependencies = [ "log", "rustversion", "windows-link 0.2.1", - "windows-result 0.4.1", + "windows-result 0.3.4", ] [[package]] @@ -4758,9 +4758,9 @@ dependencies = [ [[package]] name = "llm_adapter" -version = "0.2.21" +version = "0.2.23" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "02c6b4fa5178b8183331a7d51e8f7ece5421ecb966d3431f3d9ce3c724c6fcc0" +checksum = "bd18d17c60b800981d83224309ff9ce6b4093ae12929f2baad77dd5208383bdc" dependencies = [ "base64", "jsonschema", diff --git a/packages/backend/native/index.d.ts b/packages/backend/native/index.d.ts index 8445e9117d..c099112c17 100644 --- a/packages/backend/native/index.d.ts +++ b/packages/backend/native/index.d.ts @@ -180,7 +180,7 @@ export interface AssertSafeUrlRequest { url: string } -export declare function authorizeUserdataDocSubject(userId: string, workspaceId: string, docId: string): boolean +export declare function authorizeReservedDocSubject(userId: string, workspaceId: string, docId: string): boolean export declare function authSessionAccessTokenKeyId(token: string): string | null diff --git a/packages/backend/native/src/lib.rs b/packages/backend/native/src/lib.rs index 6faec57e0b..208f308151 100644 --- a/packages/backend/native/src/lib.rs +++ b/packages/backend/native/src/lib.rs @@ -12,6 +12,7 @@ pub mod image; pub mod license; pub mod llm; pub mod permission; +mod reserved_doc; pub mod runtime; pub mod safe_fetch; pub(crate) mod search_index; @@ -55,8 +56,8 @@ pub async fn validate_doc_update(update: Buffer) -> Result { } #[napi(catch_unwind)] -pub fn authorize_userdata_doc_subject(user_id: String, workspace_id: String, doc_id: String) -> bool { - userdata_acl::authorize(&user_id, &workspace_id, &doc_id) +pub fn authorize_reserved_doc_subject(user_id: String, workspace_id: String, doc_id: String) -> bool { + reserved_doc::authorize(&user_id, &workspace_id, &doc_id) } #[napi] diff --git a/packages/backend/native/src/llm/core/prompt/session.rs b/packages/backend/native/src/llm/core/prompt/session.rs index 9fe6c8b616..823382c2bf 100644 --- a/packages/backend/native/src/llm/core/prompt/session.rs +++ b/packages/backend/native/src/llm/core/prompt/session.rs @@ -144,20 +144,9 @@ fn estimated_message_bytes(message: &PromptMessageContract) -> usize { ) }); } - for value in [ - message.params.as_ref().map(|params| { - let mut metadata = params.clone(); - if let Some(Value::Array(attachments)) = metadata.get_mut("attachments") { - for attachment in attachments { - *attachment = attachment_metadata(attachment); - } - } - serde_json::to_vec(&metadata) - }), - message.response_format.as_ref().map(serde_json::to_vec), - ] - .into_iter() - .flatten() + for value in [message.response_format.as_ref().map(serde_json::to_vec)] + .into_iter() + .flatten() { size = size.saturating_add(value.map(|bytes| bytes.len()).unwrap_or(usize::MAX)); } @@ -199,7 +188,7 @@ fn select_history_turns( size.saturating_add(estimated_message_bytes(message)) }); if size > history_input_bytes { - return Err("session input exceeds history byte budget".to_string()); + return Err("session prompt and latest turn exceed input byte budget".to_string()); } let mut picked = Vec::new(); for message in history.iter().rev() { @@ -366,6 +355,13 @@ mod tests { estimated_message_bytes(¶ms_small), estimated_message_bytes(¶ms_large) ); + let without_params = message("system", "system"); + let mut with_large_params = without_params.clone(); + with_large_params.params = Some(json!({ "document": "x".repeat(200_000) })); + assert_eq!( + estimated_message_bytes(&with_large_params), + estimated_message_bytes(&without_params) + ); } #[test] @@ -394,7 +390,7 @@ mod tests { let prompt = vec![message("system", "{{content}}")]; let turns = vec![message("user", "large input")]; let error = render_session_prompt_with_budget(&prompt, None, &turns, &Map::new(), &Map::new(), 1).unwrap_err(); - assert_eq!(error, "session input exceeds history byte budget"); + assert_eq!(error, "session prompt and latest turn exceed input byte budget"); } #[test] diff --git a/packages/backend/native/src/reserved_doc.rs b/packages/backend/native/src/reserved_doc.rs new file mode 100644 index 0000000000..1dda55f685 --- /dev/null +++ b/packages/backend/native/src/reserved_doc.rs @@ -0,0 +1,54 @@ +const WORKSPACE_DATABASE_TABLES: [&str; 5] = [ + "folders", + "docProperties", + "docCustomPropertyInfo", + "pinnedCollections", + "explorerIcon", +]; +const USER_DATABASE_TABLES: [&str; 3] = ["favorite", "settings", "docIntegrationRef"]; + +pub(crate) enum ReservedDoc<'a> { + WorkspaceDatabase, + UserDatabase { owner_id: &'a str }, +} + +pub(crate) fn classify<'a>(workspace_id: &str, doc_id: &'a str) -> Option> { + let mut parts = doc_id.split('$'); + match parts.next()? { + "db" => { + let (Some(encoded_workspace_id), Some(table), None) = (parts.next(), parts.next(), parts.next()) else { + return None; + }; + (encoded_workspace_id == workspace_id && WORKSPACE_DATABASE_TABLES.contains(&table)) + .then_some(ReservedDoc::WorkspaceDatabase) + } + "userdata" => { + let (Some(owner_id), Some(encoded_workspace_id), Some(table), None) = + (parts.next(), parts.next(), parts.next(), parts.next()) + else { + return None; + }; + (!owner_id.is_empty() + && owner_id != "__local__" + && encoded_workspace_id == workspace_id + && USER_DATABASE_TABLES.contains(&table)) + .then_some(ReservedDoc::UserDatabase { owner_id }) + } + _ => None, + } +} + +pub(crate) fn authorize(user_id: &str, workspace_id: &str, doc_id: &str) -> bool { + if !doc_id.starts_with("db$") && !doc_id.starts_with("userdata$") { + return true; + } + match classify(workspace_id, doc_id) { + Some(ReservedDoc::WorkspaceDatabase) => true, + Some(ReservedDoc::UserDatabase { owner_id }) => owner_id == user_id, + None => false, + } +} + +pub(crate) fn is_userdata_table(table: &str) -> bool { + USER_DATABASE_TABLES.contains(&table) +} diff --git a/packages/backend/native/src/runtime/backend_runtime/search/generation.rs b/packages/backend/native/src/runtime/backend_runtime/search/generation.rs index 7528302a88..14e08570a9 100644 --- a/packages/backend/native/src/runtime/backend_runtime/search/generation.rs +++ b/packages/backend/native/src/runtime/backend_runtime/search/generation.rs @@ -4,7 +4,10 @@ use sqlx::{PgPool, Row}; use uuid::Uuid; use super::{SCHEMA_FINGERPRINT, SearchProvider, SearchTable}; -use crate::runtime::{RuntimeError, RuntimeResult, SearchRuntimeConfig}; +use crate::{ + runtime::{RuntimeError, RuntimeResult, SearchRuntimeConfig}, + search_index::EmbeddedSearchIndex, +}; #[derive(Clone)] pub(super) struct ActiveGeneration { @@ -29,6 +32,7 @@ pub(super) async fn ensure( rebuild_embedded: bool, ) -> RuntimeResult { let expected_config_hash = config_hash(config); + let expected_provider_identity = provider_identity(config); let mut transaction = pool .begin() .await @@ -37,6 +41,18 @@ pub(super) async fn ensure( .execute(&mut *transaction) .await .map_err(|error| RuntimeError::database("lock search generation", error))?; + sqlx::query( + r#"UPDATE search_projection.generations + SET manifest=jsonb_set(manifest,'{providerIdentity}',to_jsonb($1::text),true) + WHERE provider=$2 AND config_hash=$3 AND state IN ('building','active','draining') + AND NOT manifest ? 'providerIdentity'"#, + ) + .bind(&expected_provider_identity) + .bind(&config.provider) + .bind(&expected_config_hash) + .execute(&mut *transaction) + .await + .map_err(|error| RuntimeError::database("backfill search provider identity", error))?; let existing = sqlx::query( r#"SELECT id,provider,config_hash,schema_version,manifest,state FROM search_projection.generations @@ -126,10 +142,10 @@ pub(super) async fn ensure( { decode(row)? } else { - create_generation(&mut transaction, config, expected_config_hash.clone()).await? + create_or_reject_failed(&mut transaction, config, expected_config_hash.clone()).await? } } else { - create_generation(&mut transaction, config, expected_config_hash.clone()).await? + create_or_reject_failed(&mut transaction, config, expected_config_hash.clone()).await? } }; transaction @@ -140,7 +156,9 @@ pub(super) async fn ensure( if let Some(remote) = remote { for table in [SearchTable::Doc, SearchTable::Block] { if let Err(error) = remote.provision(generation.physical_table(table)?, table).await { - fail(pool, &generation).await?; + if !matches!(error, RuntimeError::SearchProviderUnavailable) { + fail(pool, &generation, &error.to_string()).await?; + } return Err(error); } } @@ -167,19 +185,109 @@ pub(super) async fn load_active( row.map(decode).transpose() } -async fn fail(pool: &PgPool, generation: &ActiveGeneration) -> RuntimeResult<()> { +pub(super) async fn cleanup_retired_generation( + pool: &PgPool, + embedded: &EmbeddedSearchIndex, + remote: Option<&SearchProvider>, + config: &SearchRuntimeConfig, +) -> RuntimeResult { + let row = sqlx::query( + r#"SELECT id,manifest, + generation.state='failed' + AND NOT EXISTS (SELECT 1 FROM search_projection.generations active + WHERE active.state='active' AND active.provider=generation.provider + AND active.config_hash=generation.config_hash + AND active.schema_version=generation.schema_version) + AND NOT EXISTS (SELECT 1 FROM search_projection.generations newer + WHERE newer.state='failed' AND newer.provider=generation.provider + AND newer.config_hash=generation.config_hash + AND newer.schema_version=generation.schema_version + AND newer.created_at > generation.created_at) + AS retain_failure_marker + FROM search_projection.generations generation + WHERE provider=$1 AND schema_version=$2 + AND (manifest->>'providerIdentity'=$3 + OR (NOT manifest ? 'providerIdentity' AND config_hash=$4)) + AND ((state='failed' AND created_at < now() - interval '1 hour' + AND manifest <> '{}'::jsonb) + OR (state='draining' AND drained_at < now() - interval '24 hours')) + ORDER BY COALESCE(drained_at,created_at),id LIMIT 1"#, + ) + .bind(&config.provider) + .bind(SCHEMA_FINGERPRINT) + .bind(provider_identity(config)) + .bind(config_hash(config)) + .fetch_optional(pool) + .await + .map_err(|error| RuntimeError::database("load retired search generation", error))?; + let Some(row) = row else { + return Ok(false); + }; + let retain_failure_marker: bool = row + .try_get("retain_failure_marker") + .map_err(|error| RuntimeError::database("decode retired search generation state", error))?; + let generation = decode(row)?; + if let Some(provider) = remote { + for table in [SearchTable::Doc, SearchTable::Block] { + provider + .drop_generation_asset(generation.physical_table(table)?) + .await?; + } + } else { + embedded.remove_generation(generation.id).await; + } + if retain_failure_marker { + sqlx::query("UPDATE search_projection.generations SET manifest='{}' WHERE id=$1 AND state='failed'") + .bind(generation.id) + .execute(pool) + .await + .map_err(|error| RuntimeError::database("retire failed search generation assets", error))?; + } else { + sqlx::query("DELETE FROM search_projection.generations WHERE id=$1 AND state IN ('failed','draining')") + .bind(generation.id) + .execute(pool) + .await + .map_err(|error| RuntimeError::database("delete retired search generation", error))?; + } + Ok(true) +} + +async fn fail(pool: &PgPool, generation: &ActiveGeneration, message: &str) -> RuntimeResult<()> { sqlx::query( "UPDATE search_projection.generations SET state='failed', last_error=$2 WHERE id=$1 AND state NOT IN \ ('active','draining')", ) .bind(generation.id) - .bind("search generation build failed") + .bind(message) .execute(pool) .await .map_err(|error| RuntimeError::database("fail search generation", error))?; Ok(()) } +async fn create_or_reject_failed( + transaction: &mut sqlx::Transaction<'_, sqlx::Postgres>, + config: &SearchRuntimeConfig, + config_hash: Vec, +) -> RuntimeResult { + let failure = sqlx::query_scalar::<_, String>( + r#"SELECT COALESCE(last_error,'search generation build failed') + FROM search_projection.generations + WHERE state='failed' AND provider=$1 AND config_hash=$2 AND schema_version=$3 + ORDER BY created_at DESC LIMIT 1"#, + ) + .bind(&config.provider) + .bind(&config_hash) + .bind(SCHEMA_FINGERPRINT) + .fetch_optional(&mut **transaction) + .await + .map_err(|error| RuntimeError::database("load failed search generation", error))?; + if let Some(failure) = failure { + return Err(RuntimeError::SearchIndexFailed(failure)); + } + create_generation(transaction, config, config_hash).await +} + pub(super) async fn activate( pool: &PgPool, generation: &ActiveGeneration, @@ -309,6 +417,16 @@ pub(super) fn config_hash(config: &SearchRuntimeConfig) -> Vec { hash.finalize().to_vec() } +pub(super) fn provider_identity(config: &SearchRuntimeConfig) -> String { + let mut hash = Sha256::new(); + hash.update(config.provider.as_bytes()); + hash.update([0]); + if config.provider != "embedded" { + hash.update(config.endpoint.trim_end_matches('/').as_bytes()); + } + hash.finalize().iter().map(|byte| format!("{byte:02x}")).collect() +} + fn decode(row: sqlx::postgres::PgRow) -> RuntimeResult { Ok(ActiveGeneration { id: row @@ -327,12 +445,14 @@ async fn create_generation( ) -> RuntimeResult { let generation_id = Uuid::new_v4(); let suffix = generation_id.simple().to_string(); + let provider_identity = provider_identity(config); let manifest = if config.provider == "embedded" { - json!({"doc":"doc","block":"block"}) + json!({"doc":"doc","block":"block","providerIdentity":provider_identity}) } else { json!({ "doc":format!("affine_search_doc_{suffix}"), - "block":format!("affine_search_block_{suffix}") + "block":format!("affine_search_block_{suffix}"), + "providerIdentity":provider_identity }) }; sqlx::query( @@ -356,12 +476,16 @@ async fn create_generation( #[cfg(test)] mod tests { + use serde_json::json; use sqlx::PgPool; use uuid::Uuid; - use super::ensure; - use crate::runtime::{ - SearchRuntimeConfig, backend_runtime::search::SEARCH_TEST_LOCK, migrations::migrate_search_tables, + use super::{cleanup_retired_generation, config_hash, ensure, provider_identity}; + use crate::{ + runtime::{ + RuntimeError, SearchRuntimeConfig, backend_runtime::search::SEARCH_TEST_LOCK, migrations::migrate_search_tables, + }, + search_index::EmbeddedSearchIndex, }; #[tokio::test] @@ -403,9 +527,78 @@ mod tests { Some("search generation superseded by configuration") ); + sqlx::query("DELETE FROM search_projection.generations WHERE id=$1") + .bind(generation.id) + .execute(&pool) + .await + .unwrap(); + let previous_config = SearchRuntimeConfig { + api_key: "previous credential".to_string(), + ..Default::default() + }; + let current_config = SearchRuntimeConfig { + api_key: "current credential".to_string(), + ..previous_config.clone() + }; + assert_ne!(config_hash(&previous_config), config_hash(¤t_config)); + assert_eq!(provider_identity(&previous_config), provider_identity(¤t_config)); + let retired_id = Uuid::new_v4(); + let failed_id = Uuid::new_v4(); + let manifest = json!({ + "doc":"doc", + "block":"block", + "providerIdentity":provider_identity(&previous_config) + }); + sqlx::query( + r#"INSERT INTO search_projection.generations + (id,provider,state,config_hash,schema_version,manifest,created_at) + VALUES($1,'embedded','failed',$3,1,$4,now() - interval '10 days'), + ($2,'embedded','failed',$3,1,$4,now() - interval '2 days')"#, + ) + .bind(retired_id) + .bind(failed_id) + .bind(config_hash(&previous_config)) + .bind(manifest) + .execute(&pool) + .await + .unwrap(); + let embedded = EmbeddedSearchIndex::new(); + embedded.prepare_generation(retired_id).await; + + assert!( + cleanup_retired_generation(&pool, &embedded, None, ¤t_config) + .await + .unwrap() + ); + assert!(!embedded.has_generation(retired_id).await); + assert!( + !sqlx::query_scalar::<_, bool>("SELECT EXISTS(SELECT 1 FROM search_projection.generations WHERE id=$1)") + .bind(retired_id) + .fetch_one(&pool) + .await + .unwrap() + ); + assert!( + cleanup_retired_generation(&pool, &embedded, None, ¤t_config) + .await + .unwrap() + ); + assert_eq!( + sqlx::query_scalar::<_, serde_json::Value>("SELECT manifest FROM search_projection.generations WHERE id=$1") + .bind(failed_id) + .fetch_one(&pool) + .await + .unwrap(), + serde_json::json!({}) + ); + assert!(matches!( + ensure(&pool, &previous_config, None, false).await, + Err(RuntimeError::SearchIndexFailed(_)) + )); + sqlx::query("DELETE FROM search_projection.generations WHERE id IN ($1,$2)") .bind(stale_id) - .bind(generation.id) + .bind(failed_id) .execute(&pool) .await .unwrap(); diff --git a/packages/backend/native/src/runtime/backend_runtime/search/mod.rs b/packages/backend/native/src/runtime/backend_runtime/search/mod.rs index 376a84fd72..a969a3a247 100644 --- a/packages/backend/native/src/runtime/backend_runtime/search/mod.rs +++ b/packages/backend/native/src/runtime/backend_runtime/search/mod.rs @@ -8,7 +8,7 @@ mod runtime; mod types; mod worker; -use generation::{ActiveGeneration, activate, config_hash, ensure, load_active}; +use generation::{ActiveGeneration, activate, cleanup_retired_generation, config_hash, ensure, load_active}; use projection::{ProjectionInput, project_document}; use provider::{SearchChange, SearchProvider, projection_external_id}; use query::{compile, compile_aggregate}; diff --git a/packages/backend/native/src/runtime/backend_runtime/search/provider/manticore.rs b/packages/backend/native/src/runtime/backend_runtime/search/provider/manticore.rs index 4e2336d787..4d36477d28 100644 --- a/packages/backend/native/src/runtime/backend_runtime/search/provider/manticore.rs +++ b/packages/backend/native/src/runtime/backend_runtime/search/provider/manticore.rs @@ -79,7 +79,32 @@ impl ManticoreSearchProvider { let status = response.status(); let body = read_response(response).await?; if !status.is_success() { - return Err(RuntimeError::SearchProviderUnavailable); + return Err(match status.as_u16() { + 400 => RuntimeError::invalid_state("provider_schema_failed"), + 401 | 403 => RuntimeError::config("search provider authentication failed"), + status => provider_write_error(status), + }); + } + let value: Value = serde_json::from_slice(&body) + .map_err(|error| RuntimeError::json("invalid search provider schema response", error))?; + if value.get("error").is_some() { + return Err(RuntimeError::invalid_state("provider_schema_failed")); + } + Ok(()) + } + + pub(super) async fn drop_generation_asset(&self, physical_table: &str) -> RuntimeResult<()> { + let response = self + .request(reqwest::Method::POST, "sql?mode=raw") + .header("content-type", "application/x-www-form-urlencoded") + .body(format!("DROP TABLE IF EXISTS {physical_table}")) + .send() + .await + .map_err(|_| RuntimeError::SearchProviderUnavailable)?; + let status = response.status(); + let body = read_response(response).await?; + if !status.is_success() { + return Err(provider_write_error(status.as_u16())); } let value: Value = serde_json::from_slice(&body) .map_err(|error| RuntimeError::json("invalid search provider schema response", error))?; diff --git a/packages/backend/native/src/runtime/backend_runtime/search/provider/mod.rs b/packages/backend/native/src/runtime/backend_runtime/search/provider/mod.rs index 43686fe89a..0b213f30ca 100644 --- a/packages/backend/native/src/runtime/backend_runtime/search/provider/mod.rs +++ b/packages/backend/native/src/runtime/backend_runtime/search/provider/mod.rs @@ -99,6 +99,13 @@ impl SearchProvider { } } + pub(super) async fn drop_generation_asset(&self, physical_table: &str) -> RuntimeResult<()> { + match self { + Self::Elasticsearch(provider) => provider.drop_generation_asset(physical_table).await, + Self::ManticoreSearch(provider) => provider.drop_generation_asset(physical_table).await, + } + } + pub(super) async fn apply(&self, physical_table: &str, changes: &[SearchChange]) -> RuntimeResult<()> { match self { Self::Elasticsearch(provider) => provider.apply(physical_table, changes).await, diff --git a/packages/backend/native/src/runtime/backend_runtime/search/provider/remote.rs b/packages/backend/native/src/runtime/backend_runtime/search/provider/remote.rs index e9efa13b99..44a755178c 100644 --- a/packages/backend/native/src/runtime/backend_runtime/search/provider/remote.rs +++ b/packages/backend/native/src/runtime/backend_runtime/search/provider/remote.rs @@ -114,22 +114,51 @@ impl RemoteProvider { } pub(super) async fn provision(&self, physical_table: &str, table: SearchTable) -> RuntimeResult<()> { - if self + let response = self .request(reqwest::Method::HEAD, physical_table) .send() .await - .is_ok_and(|response| response.status().is_success()) - { + .map_err(|_| RuntimeError::SearchProviderUnavailable)?; + if response.status().is_success() { return Ok(()); } + if response.status().as_u16() != 404 { + return Err(provision_error(response.status().as_u16(), &[])); + } let response = self .request(reqwest::Method::PUT, physical_table) .json(&super::mapping(table)) .send() - .await; - match response { - Ok(response) if response.status().is_success() => Ok(()), - _ => Err(RuntimeError::SearchProviderUnavailable), + .await + .map_err(|_| RuntimeError::SearchProviderUnavailable)?; + if response.status().is_success() { + return Ok(()); + } + let status = response.status(); + if status.as_u16() == 400 + && self + .request(reqwest::Method::HEAD, physical_table) + .send() + .await + .is_ok_and(|response| response.status().is_success()) + { + return Ok(()); + } + let body = read_response(response).await?; + Err(provision_error(status.as_u16(), &body)) + } + + pub(super) async fn drop_generation_asset(&self, physical_table: &str) -> RuntimeResult<()> { + let response = self + .request(reqwest::Method::DELETE, physical_table) + .send() + .await + .map_err(|_| RuntimeError::SearchProviderUnavailable)?; + let status = response.status(); + if status.is_success() || status.as_u16() == 404 { + Ok(()) + } else { + Err(provider_write_error(status.as_u16())) } } @@ -264,6 +293,26 @@ impl RemoteProvider { } } +fn provision_error(status: u16, body: &[u8]) -> RuntimeError { + if matches!(status, 401 | 403) { + return RuntimeError::config("search provider authentication failed"); + } + if status == 400 { + let error_type = serde_json::from_slice::(body) + .ok() + .and_then(|value| value.pointer("/error/type").and_then(Value::as_str).map(str::to_string)); + return if matches!( + error_type.as_deref(), + Some("mapper_parsing_exception" | "strict_dynamic_mapping_exception" | "illegal_argument_exception") + ) { + RuntimeError::invalid_state("provider_schema_failed") + } else { + RuntimeError::SearchProviderUnavailable + }; + } + provider_write_error(status) +} + fn validate_delete_response(value: &Value) -> RuntimeResult<()> { if value.get("timed_out").and_then(Value::as_bool) == Some(true) || value @@ -411,7 +460,7 @@ fn normalize_hit(hit: &Value) -> Value { mod tests { use serde_json::json; - use super::{normalize_aggregate, validate_bulk_response, validate_delete_response}; + use super::{normalize_aggregate, provision_error, validate_bulk_response, validate_delete_response}; use crate::runtime::RuntimeError; #[test] @@ -447,6 +496,15 @@ mod tests { validate_bulk_response(&json!({"errors":true,"items":[{"index":{"status":429,"error":{}}}]}), 1), Err(RuntimeError::SearchProviderUnavailable) )); + assert!(matches!( + provision_error(400, br#"{"error":{"type":"validation_exception"}}"#), + RuntimeError::SearchProviderUnavailable + )); + assert!(matches!( + provision_error(400, br#"{"error":{"type":"mapper_parsing_exception"}}"#), + RuntimeError::InvalidState(_) + )); + assert!(matches!(provision_error(401, &[]), RuntimeError::Config(_))); } #[test] diff --git a/packages/backend/native/src/runtime/backend_runtime/search/runtime.rs b/packages/backend/native/src/runtime/backend_runtime/search/runtime.rs index 448ad2e1c3..a3fcd89d39 100644 --- a/packages/backend/native/src/runtime/backend_runtime/search/runtime.rs +++ b/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, config_hash, ensure, load_active, - reconcile_workspace, sweep_generation_orphans, + ActiveGeneration, PermissionAuthorizer, SearchProvider, activate, cleanup_retired_generation, config_hash, ensure, + load_active, reconcile_workspace, sweep_generation_orphans, }; use crate::{ runtime::{RuntimeError, RuntimeResult, SearchRuntimeConfig}, @@ -130,6 +130,15 @@ impl SearchRuntime { let _lock = self.lifecycle_lock.lock().await; let started_at = Instant::now(); + if cleanup_retired_generation(&self.pool, &self.embedded, self.remote.as_ref(), &self.config) + .await + .is_err() + { + self + .observability + .generation_gc_failures + .fetch_add(1, Ordering::Relaxed); + } let generation = self.candidate_or_active().await?; if self.config.provider == "embedded" { self.embedded.prepare_generation(generation.id).await; diff --git a/packages/backend/native/src/runtime/backend_runtime/search/worker/document.rs b/packages/backend/native/src/runtime/backend_runtime/search/worker/document.rs index ef57cb489f..535bee8dda 100644 --- a/packages/backend/native/src/runtime/backend_runtime/search/worker/document.rs +++ b/packages/backend/native/src/runtime/backend_runtime/search/worker/document.rs @@ -23,8 +23,6 @@ struct ProjectionTuple { permission_version: i64, } -const HISTORY_GC_BATCH: usize = 100; - pub(super) async fn upsert_document( pool: &PgPool, embedded: &EmbeddedSearchIndex, @@ -108,7 +106,6 @@ pub(super) async fn upsert_document( ) .await?; apply_changes(embedded, remote, generation, changes).await?; - gc_document_history(pool, embedded, remote, generation, workspace_id, doc_id).await; } return Ok(()); }; @@ -127,69 +124,9 @@ pub(super) async fn upsert_document( renew_document_claim(pool, generation.id, workspace_id, doc_id, claim.fence).await?; apply_changes(embedded, remote, generation, changes).await?; complete_document(pool, generation.id, workspace_id, doc_id, &claim).await?; - gc_document_history(pool, embedded, remote, generation, workspace_id, doc_id).await; Ok(()) } -async fn gc_document_history( - pool: &PgPool, - embedded: &EmbeddedSearchIndex, - remote: Option<&SearchProvider>, - generation: &ActiveGeneration, - workspace_id: &str, - doc_id: &str, -) { - let Ok(Some((source_version, permission_version))) = sqlx::query_as::<_, (i64, i64)>( - r#"SELECT published_source_version,published_permission_version - FROM search_projection.document_states - WHERE generation_id=$1 AND workspace_id=$2 AND doc_id=$3 - AND target_source_version=published_source_version - AND target_source_exists=published_source_exists - AND target_permission_version=published_permission_version"#, - ) - .bind(generation.id) - .bind(workspace_id) - .bind(doc_id) - .fetch_optional(pool) - .await - else { - return; - }; - for table in [ProviderTable::Doc, ProviderTable::Block] { - let result = match remote { - Some(provider) => { - let Ok(physical_table) = generation.physical_table(table) else { - return; - }; - provider - .gc_document_history( - physical_table, - workspace_id, - doc_id, - source_version, - permission_version, - HISTORY_GC_BATCH, - ) - .await - } - None => embedded - .gc_document_history_for_generation( - generation.id, - table.as_str(), - workspace_id, - doc_id, - (source_version, permission_version), - HISTORY_GC_BATCH, - ) - .await - .map_err(|error| RuntimeError::invalid_state(format!("embedded search history GC failed: {error}"))), - }; - if result.is_err() { - return; - } - } -} - async fn projection_changes( pool: &PgPool, generation: &ActiveGeneration, diff --git a/packages/backend/native/src/runtime/backend_runtime/search/worker/workspace_state.rs b/packages/backend/native/src/runtime/backend_runtime/search/worker/workspace_state.rs index 3822fbea0f..6cc212d58f 100644 --- a/packages/backend/native/src/runtime/backend_runtime/search/worker/workspace_state.rs +++ b/packages/backend/native/src/runtime/backend_runtime/search/worker/workspace_state.rs @@ -6,7 +6,7 @@ use uuid::Uuid; use super::{ActiveGeneration, LEASE_SECONDS, SearchTable}; use crate::runtime::{RuntimeError, RuntimeResult}; -const ANTI_ENTROPY_INTERVAL_SECONDS: i64 = 300; +const ANTI_ENTROPY_INTERVAL_SECONDS: i64 = 3600; pub(super) struct WorkspaceClaim { pub(super) fence: i64, diff --git a/packages/backend/native/src/runtime/config.rs b/packages/backend/native/src/runtime/config.rs index b4779d8c3c..517e8682bc 100644 --- a/packages/backend/native/src/runtime/config.rs +++ b/packages/backend/native/src/runtime/config.rs @@ -182,7 +182,7 @@ pub(crate) struct CopilotManagedProfileConfigFile { priority: Option, #[serde(default = "enabled_by_default")] enabled: bool, - models: Option>, + models: Vec, middleware: Option, config: Map, } @@ -217,18 +217,6 @@ impl CopilotManagedProvider { Self::OpenAi => "openai", } } - - fn legacy_models(self) -> Vec { - let models: &[&str] = match self { - Self::OpenAi => &["gpt-5.6-luna", "gpt-5.6-terra", "gpt-image-1", "gpt-4o-mini"], - Self::CloudflareWorkersAi => &["@cf/baai/bge-reranker-base"], - Self::Fal => &["lora/image-to-image", "workflowutils/teed"], - Self::Gemini => &["gemini-3.7-flash", "gemini-embedding-001"], - Self::GeminiVertex => &["gemini-3.7-flash"], - Self::Anthropic | Self::AnthropicVertex => &["claude-sonnet-4-6"], - }; - models.iter().map(|model| (*model).to_string()).collect() - } } #[derive(Clone, Deserialize, serde::Serialize, schemars::JsonSchema)] @@ -306,12 +294,11 @@ impl TryFrom for CopilotManagedProfileConfig { "managed copilot profile id must contain only letters, numbers, hyphens, and underscores", )); } - let models = value.models.unwrap_or_else(|| value.provider.legacy_models()); Ok(Self { id: value.id, provider: value.provider.as_str().to_string(), enabled: value.enabled, - models, + models: value.models, config: serde_json::Value::Object(value.config), }) } @@ -441,6 +428,7 @@ pub(super) fn validate_copilot_config(config: &CopilotRuntimeConfig) -> RuntimeR } } let mut profile_ids = std::collections::HashSet::new(); + let mut managed_models = std::collections::HashMap::new(); for profile in &config.providers.profiles { if profile.id.trim().is_empty() || !profile_ids.insert(profile.id.as_str()) { return Err(RuntimeError::invalid_state( @@ -466,6 +454,14 @@ pub(super) fn validate_copilot_config(config: &CopilotRuntimeConfig) -> RuntimeR } provider_default_capability_upper_bound(&profile.provider, model) .ok_or_else(|| RuntimeError::invalid_state("managed copilot profile model is unsupported"))?; + if profile.enabled + && let Some(existing_profile) = managed_models.insert(model.as_str(), profile.id.as_str()) + { + return Err(RuntimeError::invalid_state(format!( + "managed copilot model {model} is assigned to both {existing_profile} and {}", + profile.id + ))); + } } } Ok(()) @@ -831,31 +827,15 @@ mod tests { assert_eq!(copilot.providers.profiles.len(), 1); assert_eq!(copilot.providers.profiles[0].id, "managed-openai"); - for (provider, expected_models) in [ - ( - "openai", - vec!["gpt-5.6-luna", "gpt-5.6-terra", "gpt-image-1", "gpt-4o-mini"], - ), - ("cloudflareWorkersAi", vec!["@cf/baai/bge-reranker-base"]), - ("fal", vec!["lora/image-to-image", "workflowutils/teed"]), - ("gemini", vec!["gemini-3.7-flash", "gemini-embedding-001"]), - ("geminiVertex", vec!["gemini-3.7-flash"]), - ("anthropic", vec!["claude-sonnet-4-6"]), - ("anthropicVertex", vec!["claude-sonnet-4-6"]), - ] { - let app_config = app_config_from_flat_overrides([( - "copilot.providers.profiles", - serde_json::json!([{ - "id": format!("{provider}-default"), - "type": provider, - "config": {} - }]), - )]) - .unwrap(); - let copilot: CopilotRuntimeConfig = app_config.copilot.unwrap().try_into().unwrap(); - validate_copilot_config(&copilot).unwrap(); - assert_eq!(copilot.providers.profiles[0].models, expected_models); - } + let missing_models = app_config_from_flat_overrides([( + "copilot.providers.profiles", + serde_json::json!([{ + "id": "managed-openai", + "type": "openai", + "config": {} + }]), + )]); + assert!(missing_models.is_err()); let app_config = app_config_from_flat_overrides([( "copilot.providers.profiles", @@ -870,6 +850,27 @@ mod tests { let copilot: CopilotRuntimeConfig = app_config.copilot.unwrap().try_into().unwrap(); assert!(validate_copilot_config(&copilot).is_err()); + let app_config = app_config_from_flat_overrides([( + "copilot.providers.profiles", + serde_json::json!([ + { + "id": "anthropic-direct", + "type": "anthropic", + "models": ["claude-sonnet-4-6"], + "config": {} + }, + { + "id": "anthropic-vertex", + "type": "anthropicVertex", + "models": ["claude-sonnet-4-6"], + "config": {} + } + ]), + )]) + .unwrap(); + let copilot: CopilotRuntimeConfig = app_config.copilot.unwrap().try_into().unwrap(); + assert!(validate_copilot_config(&copilot).is_err()); + let directory = tempfile::tempdir().unwrap(); let base_path = directory.path().join("base.json"); let override_path = directory.path().join("override.json"); diff --git a/packages/backend/native/src/runtime/storage_runtime/document_cleanup.rs b/packages/backend/native/src/runtime/storage_runtime/document_cleanup.rs index c52185aa11..601a46aa55 100644 --- a/packages/backend/native/src/runtime/storage_runtime/document_cleanup.rs +++ b/packages/backend/native/src/runtime/storage_runtime/document_cleanup.rs @@ -10,6 +10,7 @@ use super::{ RuntimeDocumentCleanupReconcileResult, RuntimeError, RuntimeResult, StorageRuntime, load_workspace_live_doc_ids, merge_current_doc, napi_error, }; +use crate::reserved_doc; #[derive(FromRow)] struct StoredDocActivity { @@ -114,7 +115,7 @@ async fn reconcile_workspace( workspace_id: &str, ) -> RuntimeResult { let pool = runtime.pool().await?; - let live_ids = match load_workspace_live_doc_ids(&pool, workspace_id).await { + let mut live_ids = match load_workspace_live_doc_ids(&pool, workspace_id).await { Ok(ids) => ids.into_iter().collect::>(), Err(err) => { record_reconcile_failure(&pool, workspace_id, "root", &err.to_string()).await?; @@ -142,6 +143,10 @@ async fn reconcile_workspace( }; for doc in &stored { + if reserved_doc::classify(workspace_id, &doc.doc_id).is_some() { + live_ids.insert(doc.doc_id.clone()); + continue; + } if live_ids.contains(&doc.doc_id) { continue; } @@ -518,6 +523,19 @@ async fn execute_one( return Ok(None); }; + if reserved_doc::classify(&candidate.workspace_id, &candidate.doc_id).is_some() { + sqlx::query("DELETE FROM document_cleanup_candidates WHERE workspace_id = $1 AND doc_id = $2") + .bind(&candidate.workspace_id) + .bind(&candidate.doc_id) + .execute(&mut *tx) + .await + .map_err(|err| RuntimeError::database("Document cleanup reserved candidate delete failed", err))?; + tx.commit() + .await + .map_err(|err| RuntimeError::database("Document cleanup reserved candidate commit failed", err))?; + return Ok(Some((candidate, -1))); + } + let root = match load_current_doc_for_update(&mut tx, &candidate.workspace_id, &candidate.workspace_id).await { Ok(Some(root)) => root, Ok(None) => { @@ -1222,6 +1240,26 @@ mod tests { .bind(missing_doc) .execute(&pool) .await?; + let reserved_db_doc_id = format!("db${workspace_id}$docProperties"); + let reserved_userdata_doc_id = format!("userdata${user_id}${workspace_id}$settings"); + let invalid_db_doc_id = "db$docProperties"; + let invalid_userdata_doc_id = format!("userdata$__local__${workspace_id}$favorite"); + for internal_doc_id in [ + reserved_db_doc_id.as_str(), + reserved_userdata_doc_id.as_str(), + invalid_db_doc_id, + invalid_userdata_doc_id.as_str(), + ] { + sqlx::query( + "INSERT INTO snapshots (workspace_id, guid, blob, updated_at) VALUES ($1, $2, $3, CURRENT_TIMESTAMP - \ + INTERVAL '90 days')", + ) + .bind(&workspace_id) + .bind(internal_doc_id) + .bind(affine_doc_loader::build_full_doc("Internal", "", internal_doc_id)?) + .execute(&pool) + .await?; + } sqlx::query( "INSERT INTO updates (workspace_id, guid, blob, created_at) VALUES ($1, $2, $3, CURRENT_TIMESTAMP - INTERVAL \ '89 days')", @@ -1252,7 +1290,40 @@ mod tests { .await?; let mark = reconcile_workspace(&runtime, &workspace_id).await?; - assert_eq!(mark.marked, 1); + assert_eq!(mark.marked, 3); + let marked_doc_ids = sqlx::query_scalar::<_, String>( + "SELECT doc_id FROM document_cleanup_candidates WHERE workspace_id = $1 ORDER BY doc_id", + ) + .bind(&workspace_id) + .fetch_all(&pool) + .await?; + assert_eq!(marked_doc_ids.len(), 3); + for doc_id in [doc_id, invalid_db_doc_id, invalid_userdata_doc_id.as_str()] { + assert!(marked_doc_ids.iter().any(|marked| marked == doc_id)); + } + for doc_id in [&reserved_db_doc_id, &reserved_userdata_doc_id] { + assert!(!marked_doc_ids.iter().any(|marked| marked == doc_id)); + } + sqlx::query( + "INSERT INTO document_cleanup_candidates (workspace_id, doc_id, status, missing_since, \ + last_observed_missing_at, last_doc_activity_at) VALUES ($1, $2, 'marked', CURRENT_TIMESTAMP - INTERVAL '31 \ + days', CURRENT_TIMESTAMP, CURRENT_TIMESTAMP - INTERVAL '90 days')", + ) + .bind(&workspace_id) + .bind(&reserved_db_doc_id) + .execute(&pool) + .await?; + let recovered_reserved = execute_one(&pool, Some(&workspace_id), 30).await?.unwrap(); + assert_eq!(recovered_reserved.0.doc_id, reserved_db_doc_id); + assert_eq!(recovered_reserved.1, -1); + assert_eq!( + sqlx::query_scalar::<_, i64>("SELECT COUNT(*) FROM snapshots WHERE workspace_id = $1 AND guid = $2") + .bind(&workspace_id) + .bind(&reserved_db_doc_id) + .fetch_one(&pool) + .await?, + 1 + ); sqlx::query( "UPDATE document_cleanup_candidates SET missing_since = CURRENT_TIMESTAMP - INTERVAL '29 days' WHERE \ workspace_id = $1 AND doc_id = $2", diff --git a/packages/backend/native/src/search_index/mod.rs b/packages/backend/native/src/search_index/mod.rs index 4d724a34e1..7874dd824b 100644 --- a/packages/backend/native/src/search_index/mod.rs +++ b/packages/backend/native/src/search_index/mod.rs @@ -128,6 +128,10 @@ impl EmbeddedSearchIndex { .retain(|id, _| *id == generation_id); } + pub(crate) async fn remove_generation(&self, generation_id: Uuid) { + self.manager.generations.write().await.remove(&generation_id); + } + #[cfg(test)] pub(crate) async fn write(&self, table: String, documents_json: String) -> napi::Result<()> { self.write_for_generation(Uuid::nil(), table, documents_json).await diff --git a/packages/backend/native/src/userdata_acl.rs b/packages/backend/native/src/userdata_acl.rs index 5f6518c9fa..5d3f6b3cd2 100644 --- a/packages/backend/native/src/userdata_acl.rs +++ b/packages/backend/native/src/userdata_acl.rs @@ -1,32 +1,21 @@ -const USERDATA_PREFIX: &str = "userdata$"; -const TABLES: [&str; 3] = ["favorite", "settings", "docIntegrationRef"]; +use crate::reserved_doc; pub(crate) fn authorize(user_id: &str, workspace_id: &str, doc_id: &str) -> bool { - if !doc_id.starts_with(USERDATA_PREFIX) { - return true; - } - let mut parts = doc_id.split('$'); - let (Some("userdata"), Some(owner_id), Some(encoded_workspace_id), Some(table), None) = - (parts.next(), parts.next(), parts.next(), parts.next(), parts.next()) - else { - return false; - }; - owner_id != "__local__" && owner_id == user_id && encoded_workspace_id == workspace_id && TABLES.contains(&table) + reserved_doc::authorize(user_id, workspace_id, doc_id) } pub(crate) fn doc_id(user_id: &str, workspace_id: &str, table: &str) -> Option { - TABLES - .contains(&table) - .then(|| format!("userdata${user_id}${workspace_id}${table}")) + reserved_doc::is_userdata_table(table).then(|| format!("userdata${user_id}${workspace_id}${table}")) } #[cfg(test)] mod tests { use super::*; + use crate::reserved_doc::ReservedDoc; #[test] fn userdata_subject_is_owner_only_and_closed() { - for table in TABLES { + for table in ["favorite", "settings", "docIntegrationRef"] { let id = doc_id("user-a", "workspace-a", table).unwrap(); assert!(authorize("user-a", "workspace-a", &id)); assert!(!authorize("user-b", "workspace-a", &id)); @@ -42,5 +31,29 @@ mod tests { assert!(!authorize("user-a", "workspace-a", id)); } assert!(authorize("user-a", "workspace-a", "ordinary-doc")); + + for table in [ + "folders", + "docProperties", + "docCustomPropertyInfo", + "pinnedCollections", + "explorerIcon", + ] { + assert!(matches!( + reserved_doc::classify("workspace-a", &format!("db$workspace-a${table}")), + Some(ReservedDoc::WorkspaceDatabase) + )); + assert!(authorize("user-a", "workspace-a", &format!("db$workspace-a${table}"))); + } + for id in [ + "db$docProperties", + "db$workspace-b$docProperties", + "db$workspace-a$unknown", + "userdata$__local__$workspace-a$favorite", + "userdata$user-a$workspace-b$favorite", + ] { + assert!(reserved_doc::classify("workspace-a", id).is_none()); + assert!(!authorize("user-a", "workspace-a", id)); + } } } diff --git a/packages/backend/server/migrations/20260826120000_migrate_legacy_copilot_prompt_names/migration.sql b/packages/backend/server/migrations/20260826120000_migrate_legacy_copilot_prompt_names/migration.sql new file mode 100644 index 0000000000..66e2403100 --- /dev/null +++ b/packages/backend/server/migrations/20260826120000_migrate_legacy_copilot_prompt_names/migration.sql @@ -0,0 +1,20 @@ +UPDATE ai_sessions_metadata +SET prompt_name = CASE prompt_name + WHEN 'Search With AFFiNE AI' THEN 'Chat With AFFiNE AI' + WHEN 'debug:action:dalle3' THEN 'Generate image' + WHEN 'debug:action:fal-sd15' THEN 'Generate image' + WHEN 'debug:action:gpt-image-1' THEN 'Generate image' + WHEN 'debug:action:fal-remove-bg' THEN 'Remove background' + WHEN 'debug:action:fal-upscaler' THEN 'Upscale image' + WHEN 'debug:action:fal-face-to-sticker' THEN 'Convert to sticker' + ELSE prompt_name +END +WHERE prompt_name IN ( + 'Search With AFFiNE AI', + 'debug:action:dalle3', + 'debug:action:fal-sd15', + 'debug:action:gpt-image-1', + 'debug:action:fal-remove-bg', + 'debug:action:fal-upscaler', + 'debug:action:fal-face-to-sticker' +); diff --git a/packages/backend/server/src/__tests__/sync/gateway.spec.ts b/packages/backend/server/src/__tests__/sync/gateway.spec.ts index e1ef95e6b8..052841af35 100644 --- a/packages/backend/server/src/__tests__/sync/gateway.spec.ts +++ b/packages/backend/server/src/__tests__/sync/gateway.spec.ts @@ -1444,6 +1444,17 @@ test('workspace sync push-doc-update should enforce doc update permissions', asy ); t.is(userdataError.name, 'SPACE_ACCESS_DENIED'); + const malformedDatabaseError = getErrorResponse( + t, + await emitWithAck(socket, 'space:push-doc-update', { + spaceType: 'workspace', + spaceId: workspace.id, + docId: 'db$docProperties', + update: createYjsUpdateBase64(), + }) + ); + t.is(malformedDatabaseError.name, 'SPACE_ACCESS_DENIED'); + const updates = await db.update.count({ where: { workspaceId: workspace.id, diff --git a/packages/backend/server/src/core/backend-runtime/job.ts b/packages/backend/server/src/core/backend-runtime/job.ts index c19684abae..bea5aea5a3 100644 --- a/packages/backend/server/src/core/backend-runtime/job.ts +++ b/packages/backend/server/src/core/backend-runtime/job.ts @@ -277,7 +277,11 @@ export class BackendRuntimeSearchJob { await this.queue.add( 'backendRuntime.reconcileSearchProjection', { limit: 100 }, - { jobId: 'backend-runtime-search-reconciliation', removeOnFail: true } + { + jobId: 'backend-runtime-search-reconciliation', + attempts: 1, + removeOnFail: true, + } ); } diff --git a/packages/backend/server/src/core/sync/gateway.ts b/packages/backend/server/src/core/sync/gateway.ts index 0c36179732..a306741543 100644 --- a/packages/backend/server/src/core/sync/gateway.ts +++ b/packages/backend/server/src/core/sync/gateway.ts @@ -33,7 +33,7 @@ import { SpaceAccessDenied, } from '../../base'; import { Models } from '../../models'; -import { authorizeUserdataDocSubject } from '../../native'; +import { authorizeReservedDocSubject } from '../../native'; import { CurrentUser } from '../auth'; import { DocReader, @@ -469,7 +469,7 @@ export class SpaceSyncGateway await this.ac.user(userId).doc(spaceId, docId).assert(action); } - private assertUserdataSubject( + private assertReservedDocSubject( spaceType: SpaceType, userId: string, workspaceId: string, @@ -477,7 +477,7 @@ export class SpaceSyncGateway ) { if ( spaceType === SpaceType.Workspace && - !authorizeUserdataDocSubject(userId, workspaceId, docId) + !authorizeReservedDocSubject(userId, workspaceId, docId) ) { throw new SpaceAccessDenied({ spaceId: workspaceId }); } @@ -928,7 +928,12 @@ export class SpaceSyncGateway } try { - this.assertUserdataSubject(event.spaceType, userId, spaceId, docId); + this.assertReservedDocSubject( + event.spaceType, + userId, + spaceId, + docId + ); await this.assertDocActionAllowed( event.spaceType, userId, @@ -1095,7 +1100,7 @@ export class SpaceSyncGateway if (space.docId === undefined) { continue; } - this.assertUserdataSubject( + this.assertReservedDocSubject( space.spaceType, user.id, space.spaceId, @@ -1212,7 +1217,7 @@ export class SpaceSyncGateway const id = new DocID(docId, spaceId); const adapter = this.selectAdapter(client, spaceType); adapter.assertIn(spaceId); - this.assertUserdataSubject(spaceType, user.id, spaceId, id.guid); + this.assertReservedDocSubject(spaceType, user.id, spaceId, id.guid); await this.assertDocActionAllowed( spaceType, user.id, @@ -1247,7 +1252,7 @@ export class SpaceSyncGateway @MessageBody() { spaceType, spaceId, docId }: DeleteDocMessage ): Promise> { const adapter = this.selectAdapter(client, spaceType); - this.assertUserdataSubject(spaceType, user.id, spaceId, docId); + this.assertReservedDocSubject(spaceType, user.id, spaceId, docId); await this.assertDocActionAllowed( spaceType, user.id, @@ -1273,7 +1278,7 @@ export class SpaceSyncGateway const adapter = this.selectAdapter(client, spaceType); // Quota recovery mode is intentionally not applied to sync. - this.assertUserdataSubject(spaceType, user.id, spaceId, docId); + this.assertReservedDocSubject(spaceType, user.id, spaceId, docId); await this.assertDocActionAllowed( spaceType, user.id, diff --git a/packages/backend/server/src/data/__tests__/migrations.spec.ts b/packages/backend/server/src/data/__tests__/migrations.spec.ts index 65115f320e..6995144968 100644 --- a/packages/backend/server/src/data/__tests__/migrations.spec.ts +++ b/packages/backend/server/src/data/__tests__/migrations.spec.ts @@ -137,9 +137,10 @@ test('managed provider migration preserves explicit profiles and converts legacy }); const profiles = [ { - id: 'openai-default', - type: 'openai', + id: 'cloudflare-existing', + type: 'cloudflareWorkersAi', priority: 7, + models: ['@cf/baai/bge-reranker-base'], config: { apiKey: 'profile-key' }, }, ]; @@ -148,12 +149,16 @@ test('managed provider migration preserves explicit profiles and converts legacy { id: 'copilot.providers.profiles', value: profiles }, { id: 'copilot.providers.openai', - value: { apiKey: 'shadowed-legacy-key' }, + value: { apiKey: 'openai-key' }, }, { id: 'copilot.providers.gemini', value: { apiKey: 'gemini-key' }, }, + { + id: 'copilot.providers.geminiVertex', + value: { projectId: 'gemini-vertex-project' }, + }, { id: 'copilot.providers.defaults', value: { fallback: 'openai-default' }, @@ -169,18 +174,38 @@ test('managed provider migration preserves explicit profiles and converts legacy }); t.deepEqual(migrated.value, [ ...profiles, + { + id: 'openai-default', + type: 'openai', + priority: 7, + models: ['gpt-5.6-luna', 'gpt-5.6-terra', 'gpt-image-1', 'gpt-4o-mini'], + config: { apiKey: 'openai-key' }, + }, { id: 'gemini-default', type: 'gemini', priority: 4, + models: ['gemini-3.7-flash', 'gemini-embedding-001'], config: { apiKey: 'gemini-key' }, }, + { + id: 'geminiVertex-default', + type: 'geminiVertex', + priority: 3, + models: ['gemini-3.7-flash'], + config: { projectId: 'gemini-vertex-project' }, + enabled: false, + }, ]); t.is( await t.context.db.appConfig.count({ where: { id: { - in: ['copilot.providers.openai', 'copilot.providers.gemini'], + in: [ + 'copilot.providers.openai', + 'copilot.providers.gemini', + 'copilot.providers.geminiVertex', + ], }, }, }), diff --git a/packages/backend/server/src/data/migrations/1786810000000-converge-managed-provider-profiles.ts b/packages/backend/server/src/data/migrations/1786810000000-converge-managed-provider-profiles.ts index 0729344307..e35700fe09 100644 --- a/packages/backend/server/src/data/migrations/1786810000000-converge-managed-provider-profiles.ts +++ b/packages/backend/server/src/data/migrations/1786810000000-converge-managed-provider-profiles.ts @@ -13,6 +13,15 @@ const PROVIDERS = [ ] as const; const PROVIDER_IDS = PROVIDERS.map(provider => `copilot.providers.${provider}`); +const PROVIDER_MODELS: Record<(typeof PROVIDERS)[number], string[]> = { + openai: ['gpt-5.6-luna', 'gpt-5.6-terra', 'gpt-image-1', 'gpt-4o-mini'], + cloudflareWorkersAi: ['@cf/baai/bge-reranker-base'], + fal: ['lora/image-to-image', 'workflowutils/teed'], + gemini: ['gemini-3.7-flash', 'gemini-embedding-001'], + geminiVertex: ['gemini-3.7-flash'], + anthropic: ['claude-sonnet-4-6'], + anthropicVertex: ['claude-sonnet-4-6'], +}; function isRecord(value: unknown): value is Record { return !!value && typeof value === 'object' && !Array.isArray(value); @@ -58,23 +67,39 @@ export class ConvergeManagedProviderProfiles1786810000000 { : [] ) ); + const assignedModels = new Set( + profiles.flatMap(profile => + isRecord(profile) && + profile.enabled !== false && + Array.isArray(profile.models) + ? profile.models.filter( + (model): model is string => typeof model === 'string' + ) + : [] + ) + ); for (const [index, provider] of PROVIDERS.entries()) { const legacy = byId.get(`copilot.providers.${provider}`); - if (!legacy) { - continue; - } + if (!legacy) continue; if (!isRecord(legacy.value)) { throw new Error(`copilot.providers.${provider} must be an object`); } const id = `${provider}-default`; if (!profileIds.has(id)) { + const models = PROVIDER_MODELS[provider].filter( + model => !assignedModels.has(model) + ); + const enabled = models.length > 0; profiles.push({ id, type: provider, priority: PROVIDERS.length - index, + models: enabled ? models : PROVIDER_MODELS[provider], config: legacy.value, + ...(enabled ? {} : { enabled: false }), }); + models.forEach(model => assignedModels.add(model)); profileIds.add(id); } } diff --git a/packages/backend/server/src/native.ts b/packages/backend/server/src/native.ts index 8e36e7ae59..d70c493d51 100644 --- a/packages/backend/server/src/native.ts +++ b/packages/backend/server/src/native.ts @@ -231,8 +231,8 @@ import type { } from './plugins/copilot/runtime/contracts/tool-contract'; export const mergeUpdatesInApplyWay = serverNativeModule.mergeUpdatesInApplyWay; -export const authorizeUserdataDocSubject = - serverNativeModule.authorizeUserdataDocSubject; +export const authorizeReservedDocSubject = + serverNativeModule.authorizeReservedDocSubject; export const authSessionAccessTokenKeyId = serverNativeModule.authSessionAccessTokenKeyId; export const createAuthSessionRefreshToken = diff --git a/packages/backend/server/src/plugins/indexer/__tests__/service.spec.ts b/packages/backend/server/src/plugins/indexer/__tests__/service.spec.ts index 8f48721c44..25eac19213 100644 --- a/packages/backend/server/src/plugins/indexer/__tests__/service.spec.ts +++ b/packages/backend/server/src/plugins/indexer/__tests__/service.spec.ts @@ -104,6 +104,7 @@ test('does not schedule or run native search reconciliation when disabled', asyn await job.scheduleReconciliation(); t.deepEqual(queue.add.firstCall.args[2], { jobId: 'backend-runtime-search-reconciliation', + attempts: 1, removeOnFail: true, }); });