diff --git a/backend/crates/gateway-admin/src/backup/task.rs b/backend/crates/gateway-admin/src/backup/task.rs index 34ae8eaf7..3fbbfdf85 100644 --- a/backend/crates/gateway-admin/src/backup/task.rs +++ b/backend/crates/gateway-admin/src/backup/task.rs @@ -487,12 +487,35 @@ impl BackupTask { ); } } - Ok(_) => { + Ok(Some(remote)) => { + let size_matches = remote.size_bytes == size_bytes; + let sha256_matches = remote.sha256 == sha256; + warn!( + backup_id = %record.id, + expected_size_bytes = size_bytes, + remote_size_bytes = remote.size_bytes, + sha256_matches, + "备份远端对象校验不一致" + ); let _ = store.delete_object(&config, &record.object_key).await; self.fail_task( record, code::REMOTE_VERIFICATION_FAILED, - "远端对象大小或校验值不一致", + match (size_matches, sha256_matches) { + (false, true) => "远端对象大小不一致", + (true, false) => "远端对象 sha256 校验值不一致", + (false, false) => "远端对象大小和 sha256 校验值均不一致", + (true, true) => "远端对象校验状态异常", + }, + ) + .await?; + } + Ok(None) => { + warn!(backup_id = %record.id, "上传完成后未找到远端对象"); + self.fail_task( + record, + code::REMOTE_VERIFICATION_FAILED, + "上传完成后未找到远端对象", ) .await?; } @@ -516,10 +539,17 @@ impl BackupTask { error_code: &'static str, message: impl Into, ) -> Result<(), WorkerTaskError> { + let message = message.into(); + warn!( + backup_id = %record.id, + error_code, + error_message = %message, + "备份任务失败" + ); let now = Utc::now(); let update = StatusTransitionUpdate { error_code: Some(error_code.to_string()), - error_message: Some(message.into()), + error_message: Some(message), completed_at: Some(now), ..Default::default() }; diff --git a/backend/crates/gateway-store/Cargo.toml b/backend/crates/gateway-store/Cargo.toml index b57de36e5..44fdf4fad 100644 --- a/backend/crates/gateway-store/Cargo.toml +++ b/backend/crates/gateway-store/Cargo.toml @@ -37,7 +37,7 @@ uuid = { version = "1.23.3", features = ["v4", "v7"] } [dev-dependencies] tempfile = "3.23" -tokio = { version = "1.52.3", default-features = false, features = ["macros", "rt-multi-thread", "fs", "io-util", "process", "time"] } +tokio = { version = "1.52.3", default-features = false, features = ["macros", "rt-multi-thread", "fs", "io-util", "net", "process", "time"] } [lints] workspace = true diff --git a/backend/crates/gateway-store/src/backup/s3.rs b/backend/crates/gateway-store/src/backup/s3.rs index e742ee8d0..26433cded 100644 --- a/backend/crates/gateway-store/src/backup/s3.rs +++ b/backend/crates/gateway-store/src/backup/s3.rs @@ -385,7 +385,13 @@ async fn upload_parts( BackupError::new(code::S3_UPLOAD_FAILED, "上传并发闸门关闭".to_owned()) })?; let offset = (part - 1) * (PART_SIZE as u64); - let bytes = read_part(&source, offset, PART_SIZE).await?; + let remaining = file_size.checked_sub(offset).ok_or_else(|| { + BackupError::new(code::S3_UPLOAD_FAILED, "分片偏移超出归档大小".to_owned()) + })?; + let part_len = usize::try_from(remaining.min(PART_SIZE as u64)).map_err(|_| { + BackupError::new(code::S3_UPLOAD_FAILED, "分片大小溢出".to_owned()) + })?; + let bytes = read_part(&source, offset, part_len).await?; let part_number = i32::try_from(part).map_err(|_| { BackupError::new(code::S3_UPLOAD_FAILED, "分片号溢出".to_owned()) })?; @@ -450,20 +456,19 @@ async fn upload_parts( .map_err(map_s3_error) } -/// 从文件中读取一个分片(最多 `PART_SIZE` 字节)。 -async fn read_part(source: &Path, offset: u64, max_len: usize) -> Result, BackupError> { +/// 从文件中读取指定长度的完整分片。 +async fn read_part(source: &Path, offset: u64, len: usize) -> Result, BackupError> { let mut file = tokio::fs::File::open(source) .await .map_err(|_| BackupError::new(code::S3_UPLOAD_FAILED, "打开暂存归档失败".to_owned()))?; file.seek(SeekFrom::Start(offset)) .await .map_err(|_| BackupError::new(code::S3_UPLOAD_FAILED, "定位暂存归档失败".to_owned()))?; - let mut buffer = vec![0_u8; max_len]; - let n = file - .read(&mut buffer) + let mut buffer = vec![0_u8; len]; + // Tokio 默认单次文件读取最多 2 MiB;分片必须读取完整,不能把短读当作末尾分片上传。 + file.read_exact(&mut buffer) .await .map_err(|_| BackupError::new(code::S3_UPLOAD_FAILED, "读取暂存归档失败".to_owned()))?; - buffer.truncate(n); Ok(buffer) } diff --git a/backend/crates/gateway-store/tests/main.rs b/backend/crates/gateway-store/tests/main.rs index 3ec752bac..6ff8065c4 100644 --- a/backend/crates/gateway-store/tests/main.rs +++ b/backend/crates/gateway-store/tests/main.rs @@ -1,3 +1,4 @@ mod postgres; mod redis; +mod s3; mod support; diff --git a/backend/crates/gateway-store/tests/s3.rs b/backend/crates/gateway-store/tests/s3.rs new file mode 100644 index 000000000..8805b2088 --- /dev/null +++ b/backend/crates/gateway-store/tests/s3.rs @@ -0,0 +1,141 @@ +use std::io; + +use chrono::Utc; +use gateway_admin::model::backup::{BackupObjectMetadata, BackupStorageConfig}; +use gateway_admin::ports::backup::{BackupObjectStorePort, UploadObjectRequest}; +use gateway_core::engine::CancellationToken; +use gateway_store::backup::s3::S3ObjectStoreAdapter; +use secrecy::SecretString; +use tokio::io::{AsyncReadExt as _, AsyncWriteExt as _}; +use tokio::net::{TcpListener, TcpStream}; + +const PART_SIZE_ABOVE_TOKIO_DEFAULT: usize = 2 * 1024 * 1024 + 1; + +#[tokio::test] +async fn upload_file_sends_full_part_beyond_tokio_default_buffer_limit() { + let directory = tempfile::tempdir().expect("create temporary directory"); + let source = directory.path().join("archive.dump"); + let expected = vec![b'x'; PART_SIZE_ABOVE_TOKIO_DEFAULT]; + tokio::fs::write(&source, &expected) + .await + .expect("write temporary archive"); + + let listener = TcpListener::bind("127.0.0.1:0") + .await + .expect("bind multipart probe server"); + let endpoint = format!( + "http://{}", + listener.local_addr().expect("read listener address") + ); + let server = tokio::spawn(serve_multipart_probe(listener)); + let storage = BackupStorageConfig { + storage_revision: 1, + endpoint, + region: "auto".to_owned(), + bucket: "backup-bucket".to_owned(), + access_key_id: "test-access-key".to_owned(), + secret_access_key: SecretString::from("test-secret-key"), + prefix: "backups".to_owned(), + force_path_style: true, + }; + let request = UploadObjectRequest { + object_key: "backups/archive.dump".to_owned(), + source, + metadata: BackupObjectMetadata::new( + "backup_test".to_owned(), + "a".repeat(64), + Utc::now(), + PART_SIZE_ABOVE_TOKIO_DEFAULT as u64, + ), + cancellation: CancellationToken::new(), + }; + + S3ObjectStoreAdapter::new() + .upload_file(&storage, request) + .await + .expect("upload multipart archive"); + let uploaded_len = server + .await + .expect("multipart probe task did not panic") + .expect("serve multipart probe"); + + assert_eq!(uploaded_len, PART_SIZE_ABOVE_TOKIO_DEFAULT); +} + +async fn serve_multipart_probe(listener: TcpListener) -> io::Result { + let mut uploaded_len = None; + for _ in 0..3 { + let (mut stream, _) = listener.accept().await?; + let (request, body) = read_http_request(&mut stream).await?; + let response = if request.starts_with("POST ") && request.contains("uploads") { + xml_response( + "probe-upload", + ) + } else if request.starts_with("PUT ") && request.contains("partNumber=") { + uploaded_len = Some(body.len()); + "HTTP/1.1 200 OK\r\ncontent-length: 0\r\nconnection: close\r\netag: \"probe-part\"\r\n\r\n".to_owned() + } else if request.starts_with("POST ") && request.contains("uploadId=") { + xml_response( + "\"probe-part\"", + ) + } else { + return Err(io::Error::other("unexpected multipart probe request")); + }; + stream.write_all(response.as_bytes()).await?; + stream.shutdown().await?; + } + uploaded_len.ok_or_else(|| io::Error::other("multipart probe did not receive an upload part")) +} + +async fn read_http_request(stream: &mut TcpStream) -> io::Result<(String, Vec)> { + let mut bytes = Vec::new(); + let header_end = loop { + if let Some(position) = bytes.windows(4).position(|window| window == b"\r\n\r\n") { + break position + 4; + } + let mut chunk = [0_u8; 8 * 1024]; + let read = stream.read(&mut chunk).await?; + if read == 0 { + return Err(io::Error::new( + io::ErrorKind::UnexpectedEof, + "multipart probe request ended before headers", + )); + } + bytes.extend_from_slice(&chunk[..read]); + }; + let headers = String::from_utf8_lossy(&bytes[..header_end]).into_owned(); + let content_len = headers + .lines() + .filter_map(|line| line.split_once(':')) + .find(|(name, _)| name.eq_ignore_ascii_case("content-length")) + .and_then(|(_, value)| value.trim().parse::().ok()) + .unwrap_or_default(); + if headers + .to_ascii_lowercase() + .contains("expect: 100-continue") + { + stream.write_all(b"HTTP/1.1 100 Continue\r\n\r\n").await?; + } + while bytes.len() < header_end + content_len { + let mut chunk = [0_u8; 8 * 1024]; + let read = stream.read(&mut chunk).await?; + if read == 0 { + return Err(io::Error::new( + io::ErrorKind::UnexpectedEof, + "multipart probe request ended before body", + )); + } + bytes.extend_from_slice(&chunk[..read]); + } + Ok(( + headers, + bytes[header_end..header_end + content_len].to_vec(), + )) +} + +fn xml_response(body: &str) -> String { + format!( + "HTTP/1.1 200 OK\r\ncontent-type: application/xml\r\ncontent-length: {}\r\nconnection: close\r\n\r\n{body}", + body.len() + ) +} diff --git a/frontend/src/components/base/BaseButton.vue b/frontend/src/components/base/BaseButton.vue index 10d983434..5dcb98dec 100644 --- a/frontend/src/components/base/BaseButton.vue +++ b/frontend/src/components/base/BaseButton.vue @@ -110,10 +110,13 @@ const labelClasses = computed(() => [ > diff --git a/frontend/src/views/accounts/components/AccountStatusBadge.vue b/frontend/src/views/accounts/components/AccountStatusBadge.vue index 36779b71e..30d39342a 100644 --- a/frontend/src/views/accounts/components/AccountStatusBadge.vue +++ b/frontend/src/views/accounts/components/AccountStatusBadge.vue @@ -26,31 +26,31 @@ const statusStyles = { success: { text: 'text-(--cp-success-text)', dot: 'bg-(--cp-success)', - badge: 'border-(--cp-success-border) bg-(--cp-success-bg) text-(--cp-success-text)', - icon: 'bg-(--cp-success-bg) text-(--cp-success-text) shadow-[inset_0_0_0_1px_var(--cp-success-border)]', + badge: 'bg-(--cp-success-bg) text-(--cp-success-text)', + icon: 'bg-(--cp-success-bg) text-(--cp-success-text)', }, danger: { text: 'text-(--cp-danger-text)', dot: 'bg-(--cp-danger)', - badge: 'border-(--cp-danger-border) bg-(--cp-danger-bg) text-(--cp-danger-text)', - icon: 'bg-(--cp-danger-bg) text-(--cp-danger-text) shadow-[inset_0_0_0_1px_var(--cp-danger-border)]', + badge: 'bg-(--cp-danger-bg) text-(--cp-danger-text)', + icon: 'bg-(--cp-danger-bg) text-(--cp-danger-text)', }, warning: { text: 'text-(--cp-warning-text)', dot: 'bg-(--cp-warning)', - badge: 'border-(--cp-warning-border) bg-(--cp-warning-bg) text-(--cp-warning-text)', - icon: 'bg-(--cp-warning-bg) text-(--cp-warning-text) shadow-[inset_0_0_0_1px_var(--cp-warning-border)]', + badge: 'bg-(--cp-warning-bg) text-(--cp-warning-text)', + icon: 'bg-(--cp-warning-bg) text-(--cp-warning-text)', }, info: { text: 'text-(--cp-info-text)', dot: 'bg-(--cp-info)', - badge: 'border-(--cp-info-border) bg-(--cp-info-bg) text-(--cp-info-text)', - icon: 'bg-(--cp-info-bg) text-(--cp-info-text) shadow-[inset_0_0_0_1px_var(--cp-info-border)]', + badge: 'bg-(--cp-info-bg) text-(--cp-info-text)', + icon: 'bg-(--cp-info-bg) text-(--cp-info-text)', }, normal: { text: 'text-(--cp-text-secondary)', dot: 'bg-(--cp-text-muted)', - badge: 'border-(--cp-divider-subtle) bg-(--cp-bg-subtle) text-(--cp-text-secondary)', + badge: 'bg-(--cp-bg-subtle) text-(--cp-text-secondary)', icon: 'bg-(--cp-bg-subtle) text-(--cp-text-secondary)', }, } as const @@ -165,7 +165,7 @@ const cardIcon = computed(() => {
-
+