qinqx
e7760ea80d
Add built-in storage support from v-hono-storage module: - StorageProvider interface with 4 backends - FileService with HTTP handlers - Multipart upload support - 86 property-based tests - Retry mechanism with exponential backoff Breaking: None (backward compatible)
649 lines
17 KiB
V
649 lines
17 KiB
V
module hono
|
|
|
|
import time
|
|
import crypto.md5
|
|
|
|
// 分片上传管理器
|
|
// 负责管理分片上传的生命周期，包括初始化、上传分片、完成和取消上传
|
|
@[heap]
|
|
pub struct ChunkManager {
|
|
mut:
|
|
db DatabaseManager
|
|
provider &StorageProvider = unsafe { nil }
|
|
config ChunkManagerConfig
|
|
}
|
|
|
|
// 分片管理器配置
|
|
pub struct ChunkManagerConfig {
|
|
pub:
|
|
default_chunk_size int = 5 * 1024 * 1024 // 5MB 默认分片大小
|
|
max_chunk_size int = 100 * 1024 * 1024 // 100MB 最大分片大小
|
|
min_chunk_size int = 1024 * 1024 // 1MB 最小分片大小
|
|
max_parts int = 10000 // 最大分片数量
|
|
retry_count int = 3 // 重试次数
|
|
retry_delay_ms int = 1000 // 重试延迟（毫秒）
|
|
}
|
|
|
|
// 上传进度信息
|
|
pub struct UploadProgress {
|
|
pub:
|
|
upload_id string
|
|
file_uuid string
|
|
file_name string
|
|
file_size i64
|
|
chunk_size int
|
|
total_chunks int
|
|
uploaded_chunks int
|
|
uploaded_bytes i64
|
|
progress_pct f64
|
|
status string
|
|
pending_parts []int
|
|
created_at i64
|
|
updated_at i64
|
|
}
|
|
|
|
// 分片上传初始化参数
|
|
pub struct InitMultipartParams {
|
|
pub:
|
|
bucket string
|
|
object_key string
|
|
file_name string
|
|
file_size i64
|
|
content_type string
|
|
chunk_size int // 可选，为0时使用默认值
|
|
}
|
|
|
|
// 分片上传结果
|
|
pub struct ChunkUploadResult {
|
|
pub:
|
|
upload_id string
|
|
part_number int
|
|
etag string
|
|
size i64
|
|
success bool
|
|
error_msg string
|
|
}
|
|
|
|
|
|
// 创建分片管理器
|
|
pub fn new_chunk_manager(mut db DatabaseManager, provider &StorageProvider, config ChunkManagerConfig) ChunkManager {
|
|
return ChunkManager{
|
|
db: db
|
|
provider: unsafe { provider }
|
|
config: config
|
|
}
|
|
}
|
|
|
|
// 创建使用默认配置的分片管理器
|
|
pub fn new_chunk_manager_default(mut db DatabaseManager, provider &StorageProvider) ChunkManager {
|
|
return ChunkManager{
|
|
db: db
|
|
provider: unsafe { provider }
|
|
config: ChunkManagerConfig{}
|
|
}
|
|
}
|
|
|
|
// ============================================================================
|
|
// 核心分片上传操作
|
|
// ============================================================================
|
|
|
|
// 初始化分片上传
|
|
// 创建一个新的分片上传会话，返回 upload_id
|
|
pub fn (mut cm ChunkManager) init_multipart(params InitMultipartParams) !MultipartUpload {
|
|
// 验证参数
|
|
if params.bucket == '' {
|
|
return error('Bucket name is required')
|
|
}
|
|
if params.object_key == '' {
|
|
return error('Object key is required')
|
|
}
|
|
if params.file_size <= 0 {
|
|
return error('File size must be positive')
|
|
}
|
|
|
|
// 确定分片大小
|
|
chunk_size := if params.chunk_size > 0 {
|
|
cm.validate_chunk_size(params.chunk_size)
|
|
} else {
|
|
cm.config.default_chunk_size
|
|
}
|
|
|
|
// 计算总分片数
|
|
total_chunks := cm.calculate_total_chunks(params.file_size, chunk_size)
|
|
if total_chunks > cm.config.max_parts {
|
|
return error('File too large: would require ${total_chunks} parts, max is ${cm.config.max_parts}')
|
|
}
|
|
|
|
// 调用存储提供者初始化分片上传
|
|
provider_upload_id := cm.provider.init_multipart(params.bucket, params.object_key,
|
|
params.content_type)!
|
|
|
|
// 在数据库中创建上传记录
|
|
upload := cm.db.create_multipart_upload(MultipartUpload{
|
|
upload_id: provider_upload_id
|
|
file_name: params.file_name
|
|
file_size: params.file_size
|
|
chunk_size: chunk_size
|
|
total_chunks: total_chunks
|
|
bucket: params.bucket
|
|
object_key: params.object_key
|
|
content_type: params.content_type
|
|
})!
|
|
|
|
return upload
|
|
}
|
|
|
|
// 上传单个分片
|
|
pub fn (mut cm ChunkManager) upload_part(upload_id string, part_number int, data []u8) !ChunkUploadResult {
|
|
// 获取上传状态
|
|
upload := cm.db.get_multipart_upload(upload_id) or {
|
|
return ChunkUploadResult{
|
|
upload_id: upload_id
|
|
part_number: part_number
|
|
success: false
|
|
error_msg: 'Upload not found: ${upload_id}'
|
|
}
|
|
}
|
|
|
|
// 验证上传状态
|
|
if upload.status != 'uploading' {
|
|
return ChunkUploadResult{
|
|
upload_id: upload_id
|
|
part_number: part_number
|
|
success: false
|
|
error_msg: 'Upload is not in uploading state: ${upload.status}'
|
|
}
|
|
}
|
|
|
|
// 验证分片号
|
|
if part_number < 1 || part_number > upload.total_chunks {
|
|
return ChunkUploadResult{
|
|
upload_id: upload_id
|
|
part_number: part_number
|
|
success: false
|
|
error_msg: 'Invalid part number: ${part_number}, expected 1-${upload.total_chunks}'
|
|
}
|
|
}
|
|
|
|
// 调用存储提供者上传分片
|
|
etag := cm.provider.upload_part(upload.bucket, upload.object_key, upload_id, part_number,
|
|
data) or {
|
|
return ChunkUploadResult{
|
|
upload_id: upload_id
|
|
part_number: part_number
|
|
success: false
|
|
error_msg: 'Failed to upload part: ${err}'
|
|
}
|
|
}
|
|
|
|
// 记录已上传的分片
|
|
cm.db.record_uploaded_part(upload_id, part_number, etag, i64(data.len)) or {
|
|
return ChunkUploadResult{
|
|
upload_id: upload_id
|
|
part_number: part_number
|
|
success: false
|
|
error_msg: 'Failed to record uploaded part: ${err}'
|
|
}
|
|
}
|
|
|
|
return ChunkUploadResult{
|
|
upload_id: upload_id
|
|
part_number: part_number
|
|
etag: etag
|
|
size: i64(data.len)
|
|
success: true
|
|
error_msg: ''
|
|
}
|
|
}
|
|
|
|
|
|
// 完成分片上传
|
|
// 合并所有分片并创建最终文件
|
|
pub fn (mut cm ChunkManager) complete_multipart(upload_id string) !StorageResult {
|
|
// 获取上传状态
|
|
upload := cm.db.get_multipart_upload(upload_id)!
|
|
|
|
// 验证上传状态
|
|
if upload.status != 'uploading' {
|
|
return error('Upload is not in uploading state: ${upload.status}')
|
|
}
|
|
|
|
// 获取所有已上传的分片
|
|
uploaded_parts := cm.db.get_uploaded_parts(upload_id)!
|
|
|
|
// 验证所有分片都已上传
|
|
if uploaded_parts.len != upload.total_chunks {
|
|
return error('Not all parts uploaded: ${uploaded_parts.len}/${upload.total_chunks}')
|
|
}
|
|
|
|
// 构建分片信息列表
|
|
mut parts := []PartInfo{}
|
|
for part in uploaded_parts {
|
|
parts << new_part_info(part.part_number, part.etag, part.size)
|
|
}
|
|
|
|
// 按分片号排序
|
|
parts.sort(a.part_number < b.part_number)
|
|
|
|
// 调用存储提供者完成上传
|
|
result := cm.provider.complete_multipart(upload.bucket, upload.object_key, upload_id,
|
|
parts)!
|
|
|
|
// 更新数据库状态
|
|
cm.db.update_multipart_status(upload_id, 'completed')!
|
|
|
|
// 在文件信息表中创建记录
|
|
file_hash := calculate_chunk_file_hash(upload_id, upload.file_size)
|
|
cm.db.insert_file(FileInfo{
|
|
file_uuid: upload.file_uuid
|
|
file_hash: file_hash
|
|
file_name: upload.file_name
|
|
file_size: upload.file_size
|
|
file_type: upload.content_type
|
|
storage_type: cm.provider.provider_name()
|
|
bucket: upload.bucket
|
|
object_key: upload.object_key
|
|
metadata: ''
|
|
}) or {
|
|
// 文件已存在，忽略错误
|
|
}
|
|
|
|
return result
|
|
}
|
|
|
|
// 取消分片上传
|
|
// 清理所有已上传的分片和临时数据
|
|
pub fn (mut cm ChunkManager) abort_multipart(upload_id string) ! {
|
|
// 获取上传状态
|
|
upload := cm.db.get_multipart_upload(upload_id) or {
|
|
return error('Upload not found: ${upload_id}')
|
|
}
|
|
|
|
// 调用存储提供者取消上传
|
|
cm.provider.abort_multipart(upload.bucket, upload.object_key, upload_id) or {
|
|
// 忽略存储提供者的错误，继续清理数据库
|
|
}
|
|
|
|
// 更新数据库状态
|
|
cm.db.update_multipart_status(upload_id, 'aborted')!
|
|
|
|
// 删除上传记录和分片记录
|
|
cm.db.delete_multipart_upload(upload_id)!
|
|
}
|
|
|
|
// ============================================================================
|
|
// 上传进度和状态查询
|
|
// ============================================================================
|
|
|
|
// 获取上传进度
|
|
pub fn (cm ChunkManager) get_upload_progress(upload_id string) !UploadProgress {
|
|
// 获取上传状态
|
|
upload := cm.db.get_multipart_upload(upload_id)!
|
|
|
|
// 获取已上传的分片
|
|
uploaded_parts := cm.db.get_uploaded_parts(upload_id)!
|
|
|
|
// 计算已上传字节数
|
|
mut uploaded_bytes := i64(0)
|
|
mut uploaded_part_numbers := map[int]bool{}
|
|
for part in uploaded_parts {
|
|
uploaded_bytes += part.size
|
|
uploaded_part_numbers[part.part_number] = true
|
|
}
|
|
|
|
// 计算待上传的分片
|
|
mut pending_parts := []int{}
|
|
for i in 1 .. upload.total_chunks + 1 {
|
|
if i !in uploaded_part_numbers {
|
|
pending_parts << i
|
|
}
|
|
}
|
|
|
|
// 计算进度百分比
|
|
progress_pct := if upload.total_chunks > 0 {
|
|
f64(uploaded_parts.len) / f64(upload.total_chunks) * 100.0
|
|
} else {
|
|
0.0
|
|
}
|
|
|
|
return UploadProgress{
|
|
upload_id: upload_id
|
|
file_uuid: upload.file_uuid
|
|
file_name: upload.file_name
|
|
file_size: upload.file_size
|
|
chunk_size: upload.chunk_size
|
|
total_chunks: upload.total_chunks
|
|
uploaded_chunks: uploaded_parts.len
|
|
uploaded_bytes: uploaded_bytes
|
|
progress_pct: progress_pct
|
|
status: upload.status
|
|
pending_parts: pending_parts
|
|
created_at: upload.created_at
|
|
updated_at: upload.updated_at
|
|
}
|
|
}
|
|
|
|
// 检查分片是否已上传
|
|
pub fn (cm ChunkManager) is_part_uploaded(upload_id string, part_number int) bool {
|
|
return cm.db.is_part_uploaded(upload_id, part_number)
|
|
}
|
|
|
|
// 获取已上传的分片列表
|
|
pub fn (cm ChunkManager) get_uploaded_parts(upload_id string) ![]UploadedPart {
|
|
return cm.db.get_uploaded_parts(upload_id)
|
|
}
|
|
|
|
// 列出所有进行中的上传
|
|
pub fn (cm ChunkManager) list_pending_uploads() ![]MultipartUpload {
|
|
return cm.db.list_pending_multipart_uploads()
|
|
}
|
|
|
|
|
|
// ============================================================================
|
|
// 分片重试和恢复
|
|
// ============================================================================
|
|
|
|
// 重试上传单个分片
|
|
// 带有指数退避重试逻辑
|
|
pub fn (mut cm ChunkManager) retry_upload_part(upload_id string, part_number int, data []u8) !ChunkUploadResult {
|
|
mut last_error := ''
|
|
mut delay := cm.config.retry_delay_ms
|
|
|
|
for attempt in 0 .. cm.config.retry_count + 1 {
|
|
result := cm.upload_part(upload_id, part_number, data)!
|
|
|
|
if result.success {
|
|
return result
|
|
}
|
|
|
|
last_error = result.error_msg
|
|
|
|
// 检查是否是可重试的错误
|
|
if !cm.is_retryable_error(result.error_msg) {
|
|
return result
|
|
}
|
|
|
|
// 如果不是最后一次尝试，等待后重试
|
|
if attempt < cm.config.retry_count {
|
|
time.sleep(delay * time.millisecond)
|
|
delay = delay * 2 // 指数退避
|
|
if delay > 30000 {
|
|
delay = 30000 // 最大延迟 30 秒
|
|
}
|
|
}
|
|
}
|
|
|
|
return ChunkUploadResult{
|
|
upload_id: upload_id
|
|
part_number: part_number
|
|
success: false
|
|
error_msg: 'All retry attempts failed: ${last_error}'
|
|
}
|
|
}
|
|
|
|
// 恢复上传（断点续传）
|
|
// 返回需要上传的分片列表
|
|
pub fn (cm ChunkManager) get_pending_parts(upload_id string) ![]int {
|
|
progress := cm.get_upload_progress(upload_id)!
|
|
return progress.pending_parts
|
|
}
|
|
|
|
// 批量上传待处理的分片
|
|
// data_provider 是一个函数，根据分片号返回分片数据
|
|
pub fn (mut cm ChunkManager) upload_pending_parts(upload_id string, data_provider fn (int) ![]u8) ![]ChunkUploadResult {
|
|
pending_parts := cm.get_pending_parts(upload_id)!
|
|
|
|
mut results := []ChunkUploadResult{}
|
|
for part_number in pending_parts {
|
|
data := data_provider(part_number) or {
|
|
results << ChunkUploadResult{
|
|
upload_id: upload_id
|
|
part_number: part_number
|
|
success: false
|
|
error_msg: 'Failed to get data for part ${part_number}: ${err}'
|
|
}
|
|
continue
|
|
}
|
|
|
|
result := cm.retry_upload_part(upload_id, part_number, data) or {
|
|
results << ChunkUploadResult{
|
|
upload_id: upload_id
|
|
part_number: part_number
|
|
success: false
|
|
error_msg: 'Failed to upload part ${part_number}: ${err}'
|
|
}
|
|
continue
|
|
}
|
|
|
|
results << result
|
|
}
|
|
|
|
return results
|
|
}
|
|
|
|
// ============================================================================
|
|
// 辅助方法
|
|
// ============================================================================
|
|
|
|
// 验证分片大小
|
|
fn (cm ChunkManager) validate_chunk_size(size int) int {
|
|
if size < cm.config.min_chunk_size {
|
|
return cm.config.min_chunk_size
|
|
}
|
|
if size > cm.config.max_chunk_size {
|
|
return cm.config.max_chunk_size
|
|
}
|
|
return size
|
|
}
|
|
|
|
// 计算总分片数
|
|
fn (cm ChunkManager) calculate_total_chunks(file_size i64, chunk_size int) int {
|
|
chunks := file_size / i64(chunk_size)
|
|
if file_size % i64(chunk_size) > 0 {
|
|
return int(chunks) + 1
|
|
}
|
|
return int(chunks)
|
|
}
|
|
|
|
// 判断错误是否可重试
|
|
fn (cm ChunkManager) is_retryable_error(error_msg string) bool {
|
|
retryable_keywords := ['timeout', 'connection', 'network', 'unavailable', 'rate limit',
|
|
'temporary']
|
|
lower_msg := error_msg.to_lower()
|
|
for keyword in retryable_keywords {
|
|
if lower_msg.contains(keyword) {
|
|
return true
|
|
}
|
|
}
|
|
return false
|
|
}
|
|
|
|
// 计算文件哈希（用于文件记录）
|
|
fn calculate_chunk_file_hash(upload_id string, file_size i64) string {
|
|
// 使用 upload_id 和 file_size 生成一个简单的哈希
|
|
// 实际应用中应该使用文件内容的 MD5/SHA256
|
|
data := '${upload_id}-${file_size}'.bytes()
|
|
hash := md5.sum(data)
|
|
mut result := ''
|
|
for b in hash {
|
|
result += '${b:02x}'
|
|
}
|
|
return result
|
|
}
|
|
|
|
// 获取分片的数据范围
|
|
pub fn (cm ChunkManager) get_part_range(upload_id string, part_number int) !(i64, i64) {
|
|
upload := cm.db.get_multipart_upload(upload_id)!
|
|
|
|
if part_number < 1 || part_number > upload.total_chunks {
|
|
return error('Invalid part number: ${part_number}')
|
|
}
|
|
|
|
start := i64(part_number - 1) * i64(upload.chunk_size)
|
|
mut end := start + i64(upload.chunk_size) - 1
|
|
|
|
// 最后一个分片可能小于 chunk_size
|
|
if end >= upload.file_size {
|
|
end = upload.file_size - 1
|
|
}
|
|
|
|
return start, end
|
|
}
|
|
|
|
// 获取分片的预期大小
|
|
pub fn (cm ChunkManager) get_expected_part_size(upload_id string, part_number int) !i64 {
|
|
start, end := cm.get_part_range(upload_id, part_number)!
|
|
return end - start + 1
|
|
}
|
|
|
|
|
|
// ============================================================================
|
|
// 断点续传支持
|
|
// ============================================================================
|
|
|
|
// 恢复上传会话信息
|
|
// 用于客户端重新连接后恢复上传状态
|
|
pub struct ResumeInfo {
|
|
pub:
|
|
upload_id string
|
|
file_uuid string
|
|
file_name string
|
|
file_size i64
|
|
chunk_size int
|
|
total_chunks int
|
|
uploaded_chunks int
|
|
pending_parts []int
|
|
can_resume bool
|
|
error_msg string
|
|
}
|
|
|
|
// 获取恢复上传所需的信息
|
|
pub fn (cm ChunkManager) get_resume_info(upload_id string) ResumeInfo {
|
|
upload := cm.db.get_multipart_upload(upload_id) or {
|
|
return ResumeInfo{
|
|
upload_id: upload_id
|
|
can_resume: false
|
|
error_msg: 'Upload not found: ${upload_id}'
|
|
}
|
|
}
|
|
|
|
// 只有 uploading 状态的上传可以恢复
|
|
if upload.status != 'uploading' {
|
|
return ResumeInfo{
|
|
upload_id: upload_id
|
|
file_uuid: upload.file_uuid
|
|
file_name: upload.file_name
|
|
file_size: upload.file_size
|
|
chunk_size: upload.chunk_size
|
|
total_chunks: upload.total_chunks
|
|
can_resume: false
|
|
error_msg: 'Upload is in ${upload.status} state, cannot resume'
|
|
}
|
|
}
|
|
|
|
// 获取进度信息
|
|
progress := cm.get_upload_progress(upload_id) or {
|
|
return ResumeInfo{
|
|
upload_id: upload_id
|
|
file_uuid: upload.file_uuid
|
|
file_name: upload.file_name
|
|
file_size: upload.file_size
|
|
chunk_size: upload.chunk_size
|
|
total_chunks: upload.total_chunks
|
|
can_resume: false
|
|
error_msg: 'Failed to get upload progress: ${err}'
|
|
}
|
|
}
|
|
|
|
return ResumeInfo{
|
|
upload_id: upload_id
|
|
file_uuid: upload.file_uuid
|
|
file_name: upload.file_name
|
|
file_size: upload.file_size
|
|
chunk_size: upload.chunk_size
|
|
total_chunks: upload.total_chunks
|
|
uploaded_chunks: progress.uploaded_chunks
|
|
pending_parts: progress.pending_parts
|
|
can_resume: true
|
|
error_msg: ''
|
|
}
|
|
}
|
|
|
|
// 根据文件名查找可恢复的上传
|
|
pub fn (cm ChunkManager) find_resumable_upload(file_name string, file_size i64) ?MultipartUpload {
|
|
pending_uploads := cm.db.list_pending_multipart_uploads() or { return none }
|
|
|
|
for upload in pending_uploads {
|
|
if upload.file_name == file_name && upload.file_size == file_size {
|
|
return upload
|
|
}
|
|
}
|
|
|
|
return none
|
|
}
|
|
|
|
// 检查上传是否可以完成（所有分片都已上传）
|
|
pub fn (cm ChunkManager) can_complete(upload_id string) bool {
|
|
progress := cm.get_upload_progress(upload_id) or { return false }
|
|
return progress.pending_parts.len == 0 && progress.status == 'uploading'
|
|
}
|
|
|
|
// 获取上传的详细状态报告
|
|
pub struct UploadStatusReport {
|
|
pub:
|
|
upload_id string
|
|
status string
|
|
file_name string
|
|
file_size i64
|
|
total_chunks int
|
|
uploaded_chunks int
|
|
pending_chunks int
|
|
uploaded_bytes i64
|
|
remaining_bytes i64
|
|
progress_pct f64
|
|
estimated_time_ms i64 // 基于平均上传速度估算的剩余时间
|
|
can_complete bool
|
|
can_resume bool
|
|
}
|
|
|
|
// 获取详细的上传状态报告
|
|
pub fn (cm ChunkManager) get_status_report(upload_id string) !UploadStatusReport {
|
|
progress := cm.get_upload_progress(upload_id)!
|
|
|
|
remaining_bytes := progress.file_size - progress.uploaded_bytes
|
|
can_complete := progress.pending_parts.len == 0 && progress.status == 'uploading'
|
|
can_resume := progress.status == 'uploading' && progress.pending_parts.len > 0
|
|
|
|
return UploadStatusReport{
|
|
upload_id: upload_id
|
|
status: progress.status
|
|
file_name: progress.file_name
|
|
file_size: progress.file_size
|
|
total_chunks: progress.total_chunks
|
|
uploaded_chunks: progress.uploaded_chunks
|
|
pending_chunks: progress.pending_parts.len
|
|
uploaded_bytes: progress.uploaded_bytes
|
|
remaining_bytes: remaining_bytes
|
|
progress_pct: progress.progress_pct
|
|
estimated_time_ms: 0 // 需要历史数据来估算
|
|
can_complete: can_complete
|
|
can_resume: can_resume
|
|
}
|
|
}
|
|
|
|
// 清理过期的上传（超过指定时间未更新的上传）
|
|
pub fn (mut cm ChunkManager) cleanup_stale_uploads(max_age_seconds i64) !int {
|
|
pending_uploads := cm.db.list_pending_multipart_uploads()!
|
|
now := time.now().unix()
|
|
mut cleaned := 0
|
|
|
|
for upload in pending_uploads {
|
|
if now - upload.updated_at > max_age_seconds {
|
|
cm.abort_multipart(upload.upload_id) or { continue }
|
|
cleaned++
|
|
}
|
|
}
|
|
|
|
return cleaned
|
|
}
|