mirror of
https://github.com/zyycn/codex-proxy-rs.git
synced 2026-11-06 08:06:19 +00:00
fix: harden backup verification and detail UI
- read every multipart backup payload to its requested length - make remote verification failures diagnosable - refine status details, loading icons, and attempt-chain presentation Co-authored-by: Codex <noreply@openai.com>
This commit is contained in:
@@ -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<String>,
|
||||
) -> 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()
|
||||
};
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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<Vec<u8>, BackupError> {
|
||||
/// 从文件中读取指定长度的完整分片。
|
||||
async fn read_part(source: &Path, offset: u64, len: usize) -> Result<Vec<u8>, 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)
|
||||
}
|
||||
|
||||
|
||||
@@ -1,3 +1,4 @@
|
||||
mod postgres;
|
||||
mod redis;
|
||||
mod s3;
|
||||
mod support;
|
||||
|
||||
@@ -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<usize> {
|
||||
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(
|
||||
"<InitiateMultipartUploadResult xmlns=\"http://s3.amazonaws.com/doc/2006-03-01/\"><UploadId>probe-upload</UploadId></InitiateMultipartUploadResult>",
|
||||
)
|
||||
} 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(
|
||||
"<CompleteMultipartUploadResult xmlns=\"http://s3.amazonaws.com/doc/2006-03-01/\"><ETag>\"probe-part\"</ETag></CompleteMultipartUploadResult>",
|
||||
)
|
||||
} 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<u8>)> {
|
||||
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::<usize>().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()
|
||||
)
|
||||
}
|
||||
@@ -110,10 +110,13 @@ const labelClasses = computed(() => [
|
||||
>
|
||||
<span
|
||||
v-if="loading"
|
||||
class="inline-flex shrink-0 animate-spin origin-center will-change-transform"
|
||||
class="inline-flex shrink-0 items-center justify-center leading-none"
|
||||
aria-hidden="true"
|
||||
>
|
||||
<LoaderCircle :size="loadingIconSize[size]" />
|
||||
<LoaderCircle
|
||||
:size="loadingIconSize[size]"
|
||||
class="block animate-spin origin-center will-change-transform transform-fill"
|
||||
/>
|
||||
</span>
|
||||
<span v-if="$slots.icon && !loading" class="inline-flex shrink-0">
|
||||
<slot name="icon" />
|
||||
|
||||
@@ -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(() => {
|
||||
</template>
|
||||
|
||||
<section class="overflow-hidden rounded-(--cp-popover-radius)">
|
||||
<header class="flex items-start gap-3 border-b border-(--cp-divider-subtle) bg-(--cp-bg-subtle) px-4 py-3">
|
||||
<header class="flex items-start gap-3 bg-(--cp-bg-subtle) px-4 py-3">
|
||||
<span
|
||||
aria-hidden="true"
|
||||
class="inline-flex size-9 shrink-0 items-center justify-center rounded-(--cp-icon-button-radius)"
|
||||
@@ -184,7 +184,7 @@ const cardIcon = computed(() => {
|
||||
</div>
|
||||
|
||||
<span
|
||||
class="inline-flex h-5 shrink-0 items-center rounded-full border px-2 text-[10px] leading-none font-heavy"
|
||||
class="inline-flex h-5 shrink-0 items-center rounded-full px-2 text-[10px] leading-none font-heavy"
|
||||
:class="statusStyle.badge"
|
||||
>
|
||||
{{ label }}
|
||||
@@ -206,7 +206,7 @@ const cardIcon = computed(() => {
|
||||
</span>
|
||||
</div>
|
||||
|
||||
<div class="rounded-(--cp-input-radius-base) border border-(--cp-divider-subtle) bg-(--cp-bg-surface) px-3 py-2.5">
|
||||
<div class="rounded-(--cp-input-radius-base) bg-(--cp-bg-subtle) px-3 py-2.5">
|
||||
<p class="m-0 text-[11px] leading-none font-heavy text-(--cp-text-tertiary)">
|
||||
建议操作
|
||||
</p>
|
||||
@@ -217,13 +217,13 @@ const cardIcon = computed(() => {
|
||||
|
||||
<div
|
||||
v-if="errorText"
|
||||
class="overflow-hidden rounded-(--cp-input-radius-base) border border-(--cp-divider-subtle) bg-(--cp-bg-subtle)"
|
||||
class="overflow-hidden rounded-(--cp-input-radius-base) bg-(--cp-bg-muted)"
|
||||
>
|
||||
<div class="flex items-center justify-between gap-3 border-b border-(--cp-divider-subtle) px-3 py-2">
|
||||
<div class="flex items-center justify-between gap-3 px-3 pt-2.5 pb-1.5">
|
||||
<span class="text-[11px] leading-none font-heavy text-(--cp-text-tertiary)">上游原始反馈</span>
|
||||
<span class="shrink-0 text-[10px] leading-none font-emphasis text-(--cp-text-muted)">仅供排查</span>
|
||||
</div>
|
||||
<BaseScrollbar max-height="124px" view-class="px-3 py-2 pr-2">
|
||||
<BaseScrollbar class="bg-(--cp-bg-subtle)" max-height="124px" view-class="px-3 py-2 pr-2">
|
||||
<pre class="m-0 whitespace-pre-wrap wrap-break-word font-mono text-[11px] leading-[1.55] font-emphasis text-(--cp-text-secondary)">{{ errorText }}</pre>
|
||||
</BaseScrollbar>
|
||||
</div>
|
||||
|
||||
@@ -409,16 +409,17 @@ const tokenDonutOption = computed<EChartsOption>(() => {
|
||||
尝试链路
|
||||
</h3>
|
||||
<span
|
||||
class="rounded-full bg-(--cp-bg-muted) px-2 py-0.5 text-[11px] leading-none font-emphasis text-(--cp-text-secondary)"
|
||||
class="text-[11px] leading-none font-emphasis text-(--cp-text-muted)"
|
||||
title="中间尝试可能有缺口,仅展示可观测到的记录"
|
||||
>
|
||||
尽力观测
|
||||
</span>
|
||||
</div>
|
||||
<BaseTable
|
||||
class="mt-3 min-w-0 font-sans"
|
||||
class="attempt-table mt-2.5 min-w-0 font-mono tabular-nums"
|
||||
:columns="attemptColumns"
|
||||
:rows="attemptRows"
|
||||
:stripe="false"
|
||||
compact
|
||||
row-key="id"
|
||||
min-width="640px"
|
||||
@@ -478,3 +479,18 @@ const tokenDonutOption = computed<EChartsOption>(() => {
|
||||
</template>
|
||||
</BaseModal>
|
||||
</template>
|
||||
|
||||
<style scoped>
|
||||
.attempt-table :deep(thead th) {
|
||||
background-color: transparent;
|
||||
box-shadow: none;
|
||||
}
|
||||
|
||||
.attempt-table :deep(tbody tr) {
|
||||
background-color: transparent;
|
||||
}
|
||||
|
||||
.attempt-table :deep(tbody td) {
|
||||
background-color: var(--cp-bg-muted);
|
||||
}
|
||||
</style>
|
||||
|
||||
Reference in New Issue
Block a user