mirror of
https://github.com/toeverything/AFFiNE.git
synced 2026-08-24 13:17:10 +08:00
chore(server): fix connection
This commit is contained in:
@@ -41,7 +41,7 @@ pub(crate) use super::types;
|
||||
pub(super) use super::{
|
||||
BackendRuntimeConfig, ConfigSource, InviteQuotaConfig, RuntimeError, RuntimeResult,
|
||||
migrations::{embedding_schema_health, migrate_all_tables},
|
||||
napi_error, to_napi_error,
|
||||
napi_error, to_napi_error, webpki_tls_config,
|
||||
};
|
||||
use crate::llm::{
|
||||
ByokLocalLeaseOutput, ByokPolicyOutput, ByokProbeResultOutput, ByokProfileOutput, CreateByokLocalLeaseInput,
|
||||
|
||||
@@ -11,6 +11,8 @@ mod worker;
|
||||
pub(super) use runtime::SearchRuntime;
|
||||
pub(super) use types::{RuntimeAggregateRequest, RuntimeSearchRequest};
|
||||
|
||||
pub(super) use super::webpki_tls_config;
|
||||
|
||||
const SCHEMA_FINGERPRINT: &str = "search-runtime-v5";
|
||||
|
||||
fn exact_token(value: &str) -> String {
|
||||
|
||||
@@ -5,7 +5,7 @@ use serde_json::{Value, json};
|
||||
use sqlx::PgPool;
|
||||
|
||||
use super::{
|
||||
super::{store::SearchChange, types::SearchTable},
|
||||
super::{store::SearchChange, types::SearchTable, webpki_tls_config},
|
||||
manticore::{manticore_exact_tokens, manticore_fields, prepare_manticore_payload, prepare_manticore_search},
|
||||
};
|
||||
use crate::runtime::{RuntimeError, RuntimeResult, SearchRuntimeConfig};
|
||||
@@ -33,6 +33,10 @@ impl RemoteProvider {
|
||||
return Err(RuntimeError::config("invalid search provider endpoint"));
|
||||
}
|
||||
let mut client = Client::builder()
|
||||
.tls_backend_preconfigured(
|
||||
webpki_tls_config()
|
||||
.map_err(|error| RuntimeError::invalid_state(format!("search TLS config failed: {error}")))?,
|
||||
)
|
||||
.redirect(Policy::none())
|
||||
.timeout(Duration::from_secs(30));
|
||||
if config.provider == "manticoresearch" {
|
||||
|
||||
@@ -0,0 +1,13 @@
|
||||
use rustls::{ClientConfig, RootCertStore};
|
||||
|
||||
pub(in crate::runtime) fn webpki_tls_config() -> Result<ClientConfig, rustls::Error> {
|
||||
let roots = RootCertStore {
|
||||
roots: webpki_roots::TLS_SERVER_ROOTS.to_vec(),
|
||||
};
|
||||
Ok(
|
||||
ClientConfig::builder_with_provider(rustls::crypto::aws_lc_rs::default_provider().into())
|
||||
.with_safe_default_protocol_versions()?
|
||||
.with_root_certificates(roots)
|
||||
.with_no_client_auth(),
|
||||
)
|
||||
}
|
||||
@@ -4,6 +4,7 @@ pub mod storage_runtime;
|
||||
pub(crate) mod config;
|
||||
mod config_descriptor;
|
||||
pub(crate) mod error;
|
||||
mod http;
|
||||
pub(crate) mod migrations;
|
||||
pub(crate) mod object_storage;
|
||||
pub(crate) mod types;
|
||||
@@ -15,3 +16,4 @@ pub(crate) use config::{
|
||||
use config::{SUPPORTED_BYOK_PROVIDERS, validate_copilot_config};
|
||||
pub use config_descriptor::{AppConfigDescriptor, app_config_descriptors, validate_app_config_value};
|
||||
pub(crate) use error::{RuntimeError, RuntimeResult, napi_error, to_napi_error};
|
||||
pub(in crate::runtime) use http::webpki_tls_config;
|
||||
|
||||
@@ -8,7 +8,6 @@ use reqwest::{
|
||||
Client as ReqwestClient, Method, StatusCode,
|
||||
header::{CONTENT_LENGTH, CONTENT_TYPE, ETAG, HeaderMap, HeaderName, HeaderValue, LAST_MODIFIED},
|
||||
};
|
||||
use rustls::RootCertStore;
|
||||
use rusty_s3::{
|
||||
Bucket, Credentials,
|
||||
actions::{
|
||||
@@ -27,6 +26,7 @@ use super::{
|
||||
ObjectListPage, ObjectMetadata, ObjectPrefix, ObjectPutMetadata, PresignedObjectRequest, completed_multipart_parts,
|
||||
trim_etag,
|
||||
},
|
||||
webpki_tls_config,
|
||||
};
|
||||
|
||||
const DEFAULT_REQUEST_TIMEOUT_MS: u64 = 30_000;
|
||||
@@ -60,7 +60,10 @@ struct ReqwestStorageHttpClient {
|
||||
impl ReqwestStorageHttpClient {
|
||||
fn new(request_timeout_ms: Option<u64>) -> ObjectStorageResult<Self> {
|
||||
let builder = ReqwestClient::builder()
|
||||
.tls_backend_preconfigured(Self::webpki_tls_config()?)
|
||||
.tls_backend_preconfigured(
|
||||
webpki_tls_config()
|
||||
.map_err(|err| ObjectStorageError::Config(format!("ObjectStorage TLS config failed: {err}")))?,
|
||||
)
|
||||
.timeout(Duration::from_millis(
|
||||
request_timeout_ms.unwrap_or(DEFAULT_REQUEST_TIMEOUT_MS),
|
||||
));
|
||||
@@ -69,19 +72,6 @@ impl ReqwestStorageHttpClient {
|
||||
})
|
||||
}
|
||||
|
||||
fn webpki_tls_config() -> ObjectStorageResult<rustls::ClientConfig> {
|
||||
let roots = RootCertStore {
|
||||
roots: webpki_roots::TLS_SERVER_ROOTS.to_vec(),
|
||||
};
|
||||
Ok(
|
||||
rustls::ClientConfig::builder_with_provider(rustls::crypto::aws_lc_rs::default_provider().into())
|
||||
.with_safe_default_protocol_versions()
|
||||
.map_err(|err| ObjectStorageError::Config(format!("ObjectStorage TLS config failed: {err}")))?
|
||||
.with_root_certificates(roots)
|
||||
.with_no_client_auth(),
|
||||
)
|
||||
}
|
||||
|
||||
async fn execute(&self, request: StorageHttpRequest) -> ObjectStorageResult<StorageHttpResponse> {
|
||||
let mut builder = self.client.request(request.method, request.url);
|
||||
for (key, value) in request.headers {
|
||||
|
||||
@@ -14,4 +14,6 @@ pub(in crate::runtime) use backend::{FsStorageConfig, StorageBackendConfig};
|
||||
pub(in crate::runtime) use config::ObjectStorageConfig;
|
||||
pub(crate) use service::ObjectStorageService;
|
||||
|
||||
pub(super) use super::webpki_tls_config;
|
||||
|
||||
pub(in crate::runtime) const MAX_BLOB_SIZE: i64 = i32::MAX as i64;
|
||||
|
||||
+22
-16
@@ -4,26 +4,32 @@
|
||||
DO $$
|
||||
BEGIN
|
||||
IF to_regclass('public.ai_contexts') IS NOT NULL AND EXISTS (
|
||||
WITH referenced_blobs AS (
|
||||
SELECT DISTINCT
|
||||
session.workspace_id,
|
||||
value #>> '{}' AS blob_key
|
||||
FROM ai_contexts context
|
||||
JOIN ai_sessions_metadata session ON session.id = context.session_id
|
||||
CROSS JOIN LATERAL jsonb_path_query(
|
||||
context.config::jsonb,
|
||||
'$.** ? (@.type() == "string")'
|
||||
) AS referenced_value(value)
|
||||
)
|
||||
SELECT 1
|
||||
FROM ai_contexts context
|
||||
JOIN ai_sessions_metadata session ON session.id = context.session_id
|
||||
FROM referenced_blobs referenced
|
||||
JOIN blobs blob
|
||||
ON blob.workspace_id = session.workspace_id
|
||||
ON blob.workspace_id = referenced.workspace_id
|
||||
AND blob.key = referenced.blob_key
|
||||
AND blob.deleted_at IS NULL
|
||||
AND blob.status = 'completed'
|
||||
WHERE jsonb_path_exists(
|
||||
context.config::jsonb,
|
||||
'$.** ? (@ == $blobKey)',
|
||||
jsonb_build_object('blobKey', to_jsonb(blob.key::text))
|
||||
)
|
||||
AND NOT EXISTS (
|
||||
SELECT 1
|
||||
FROM workspace_artifacts artifact
|
||||
WHERE artifact.workspace_id = session.workspace_id
|
||||
AND artifact.status = 'ready'
|
||||
AND artifact.storage_scope = 'blob'
|
||||
AND artifact.storage_key = concat(session.workspace_id, '/', blob.key)
|
||||
)
|
||||
WHERE NOT EXISTS (
|
||||
SELECT 1
|
||||
FROM workspace_artifacts artifact
|
||||
WHERE artifact.workspace_id = referenced.workspace_id
|
||||
AND artifact.status = 'ready'
|
||||
AND artifact.storage_scope = 'blob'
|
||||
AND artifact.storage_key = concat(referenced.workspace_id, '/', blob.key)
|
||||
)
|
||||
) THEN
|
||||
RAISE EXCEPTION
|
||||
'legacy context blob artifact admission is incomplete; run the data migration before cleanup';
|
||||
|
||||
+24
-18
@@ -39,30 +39,36 @@ export class MigrateLegacyContextBlobArtifacts1786820000000 {
|
||||
|
||||
const runtime = injector.get(BackendRuntimeProvider, { strict: false });
|
||||
const blobs = await db.$queryRaw<LegacyContextBlob[]>`
|
||||
SELECT DISTINCT
|
||||
session.workspace_id AS "workspaceId",
|
||||
WITH referenced_blobs AS (
|
||||
SELECT DISTINCT
|
||||
session.workspace_id,
|
||||
value #>> '{}' AS blob_key
|
||||
FROM ai_contexts context
|
||||
JOIN ai_sessions_metadata session ON session.id = context.session_id
|
||||
CROSS JOIN LATERAL jsonb_path_query(
|
||||
context.config::jsonb,
|
||||
'$.** ? (@.type() == "string")'
|
||||
) AS referenced_value(value)
|
||||
)
|
||||
SELECT
|
||||
referenced.workspace_id AS "workspaceId",
|
||||
blob.key AS "blobId",
|
||||
blob.mime AS "mimeType"
|
||||
FROM ai_contexts context
|
||||
JOIN ai_sessions_metadata session ON session.id = context.session_id
|
||||
FROM referenced_blobs referenced
|
||||
JOIN blobs blob
|
||||
ON blob.workspace_id = session.workspace_id
|
||||
ON blob.workspace_id = referenced.workspace_id
|
||||
AND blob.key = referenced.blob_key
|
||||
AND blob.deleted_at IS NULL
|
||||
AND blob.status = 'completed'
|
||||
WHERE jsonb_path_exists(
|
||||
context.config::jsonb,
|
||||
'$.** ? (@ == $blobKey)',
|
||||
jsonb_build_object('blobKey', to_jsonb(blob.key::text))
|
||||
WHERE NOT EXISTS (
|
||||
SELECT 1
|
||||
FROM workspace_artifacts artifact
|
||||
WHERE artifact.workspace_id = referenced.workspace_id
|
||||
AND artifact.storage_scope = 'blob'
|
||||
AND artifact.storage_key = concat(referenced.workspace_id, '/', blob.key)
|
||||
AND artifact.status = 'ready'
|
||||
)
|
||||
AND NOT EXISTS (
|
||||
SELECT 1
|
||||
FROM workspace_artifacts artifact
|
||||
WHERE artifact.workspace_id = session.workspace_id
|
||||
AND artifact.storage_scope = 'blob'
|
||||
AND artifact.storage_key = concat(session.workspace_id, '/', blob.key)
|
||||
AND artifact.status = 'ready'
|
||||
)
|
||||
ORDER BY session.workspace_id, blob.key
|
||||
ORDER BY referenced.workspace_id, blob.key
|
||||
`;
|
||||
|
||||
for (const blob of blobs) {
|
||||
|
||||
@@ -42,7 +42,9 @@ export async function run() {
|
||||
const url = app.get(URLHelper);
|
||||
let telemetry: TelemetryService | null = null;
|
||||
try {
|
||||
telemetry = app.get(TelemetryService, { strict: false });
|
||||
if (env.role !== ServerRole.Worker) {
|
||||
telemetry = app.get(TelemetryService, { strict: false });
|
||||
}
|
||||
} catch {
|
||||
telemetry = null;
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user