563 lines
18 KiB
Markdown
563 lines
18 KiB
Markdown
# DedupFs + S3Vfs Combination Design
|
||
|
||
**Date**: 2026-06-25
|
||
**Status**: Design proposal
|
||
**Goal**: Distributed deduplication storage via MinIO/S3 backend
|
||
|
||
---
|
||
|
||
## Executive Summary
|
||
|
||
### Current State
|
||
|
||
**DedupStore**（`dedup.rs`, 224 行）：
|
||
- 基于**本地文件系统**的 dedup 存储
|
||
- SHA-256 块哈希 + 引用计数
|
||
- 块存储到本地目录（`store_path/.dedup/`）
|
||
|
||
**问题**：
|
||
- ❌ 无法跨节点共享 dedup 块
|
||
- ❌ 无分布式容错能力
|
||
- ❌ 单节点存储限制
|
||
|
||
### Proposed Solution
|
||
|
||
**DedupS3Store**：
|
||
- 块存储到 **MinIO/S3** 对象（跨节点共享）
|
||
- 引用计数存储到 S3 object metadata
|
||
- Manifest 存储到 S3 对象（JSON 格式）
|
||
|
||
**优势**：
|
||
- ✅ 跨节点 dedup 共享（MinIO 分布式）
|
||
- ✅ 自动容错（MinIO erasure coding）
|
||
- ✅ 无单节点限制（MinIO 可扩展）
|
||
- ✅ 与现有 S3Vfs 集成（无需新 HTTP API）
|
||
|
||
---
|
||
|
||
## Architecture
|
||
|
||
```
|
||
┌─────────────────────────────────────────────────────────────────────────┐
|
||
│ MarkBase Node A │
|
||
│ ├── DedupS3Store │
|
||
│ │ ├── store_block() → S3 PUT <hash> │
|
||
│ │ ├── get_block() → S3 GET <hash> │
|
||
│ │ └── dedup_file() → 分块 + S3 PUT + manifest │
|
||
│ └───────────────────────────────────────────────────────────────────────┘
|
||
│ ↓ │
|
||
┌─────────────────────────────────────────────────────────────────────────┐
|
||
│ MinIO Cluster (S3-compatible) │
|
||
│ ├── Bucket: markbase-dedup │
|
||
│ │ ├── Objects: <sha256-hash> (dedup 块) │
|
||
│ │ ├── Metadata: x-amz-meta-ref-count (引用计数) │
|
||
│ │ └── Manifests: manifests/<file-id>.json │
|
||
│ │ │
|
||
│ ├── Erasure Coding: EC:2 (自动容错) │
|
||
│ ├── Replication: Node A → Node B (DR) │
|
||
│ └─────────────────────────────────────────────────────────────────────┘
|
||
│ ↓ │
|
||
┌─────────────────────────────────────────────────────────────────────────┐
|
||
│ MarkBase Node B │
|
||
│ ├── DedupS3Store │
|
||
│ │ ├── get_block() → S3 GET <hash> (共享 Node A 的块) │
|
||
│ │ └── restore_file() → S3 GET manifest + S3 GET blocks │
|
||
│ └─────────────────────────────────────────────────────────────────────┘
|
||
```
|
||
|
||
---
|
||
|
||
## Implementation Design
|
||
|
||
### DedupS3Store Struct
|
||
|
||
```rust
|
||
pub struct DedupS3Store {
|
||
s3vfs: S3Vfs, // S3 backend
|
||
bucket: String, // Bucket name (markbase-dedup)
|
||
block_prefix: String, // Object key prefix (blocks/)
|
||
manifest_prefix: String, // Manifest prefix (manifests/)
|
||
config: VfsDedupConfig, // block_size, min_file_size
|
||
}
|
||
|
||
pub struct DedupManifest {
|
||
original_size: usize,
|
||
block_hashes: Vec<String>,
|
||
dedup_ratio: f64,
|
||
file_id: String, // UUID for manifest storage
|
||
}
|
||
```
|
||
|
||
### Core Methods
|
||
|
||
| Method | Current (LocalFs) | Proposed (S3Vfs) |
|
||
|--------|------------------|------------------|
|
||
| `store_block(data)` | `std::fs::write(store_path/hash, data)` | `S3Vfs.put_object(blocks/hash, data)` |
|
||
| `get_block(hash)` | `std::fs::read(store_path/hash)` | `S3Vfs.get_object(blocks/hash)` |
|
||
| `increment_ref(hash)` | `std::fs::write(hash.ref, count)` | `S3Vfs.put_object(blocks/hash, data) + metadata update` |
|
||
| `decrement_ref(hash)` | `std::fs::write/remove` | `S3Vfs.delete_object + metadata check` |
|
||
| `dedup_file(source)` | Local file read + block store | Local file read + S3 PUT blocks |
|
||
| `restore_file(manifest)` | Local file write + block read | Local file write + S3 GET blocks |
|
||
| `get_ref_count(hash)` | `std::fs::read(hash.ref)` | `S3Vfs.head_object(blocks/hash) → metadata` |
|
||
|
||
---
|
||
|
||
## S3 Object Layout
|
||
|
||
```
|
||
Bucket: markbase-dedup
|
||
├── blocks/
|
||
│ ├── <sha256-hash-1> # Dedup 块（4KB）
|
||
│ │ └── Metadata: x-amz-meta-ref-count: 5
|
||
│ ├── <sha256-hash-2>
|
||
│ │ └── Metadata: x-amz-meta-ref-count: 2
|
||
│ └── ...
|
||
│
|
||
├── manifests/
|
||
│ ├── <file-id-1>.json # Manifest JSON
|
||
│ │ └── Content: {"original_size": 1024, "block_hashes": [...], ...}
|
||
│ ├── <file-id-2>.json
|
||
│ └── ...
|
||
│
|
||
└── stats.json # DedupStats（可选）
|
||
```
|
||
|
||
---
|
||
|
||
## Reference Count Management
|
||
|
||
### Challenge
|
||
|
||
S3 对象不支持 atomic increment/decrement 操作。
|
||
|
||
### Solution 1: Metadata Update (推荐 ⭐⭐⭐⭐⭐)
|
||
|
||
**流程**：
|
||
```rust
|
||
fn increment_ref(&self, hash: &str) -> Result<(), VfsError> {
|
||
// 1. GET current metadata
|
||
let head = self.s3vfs.head_object(&format!("blocks/{}", hash))?;
|
||
let current_ref = head.metadata.get("x-amz-meta-ref-count")
|
||
.and_then(|v| v.parse::<u64>().ok())
|
||
.unwrap_or(0);
|
||
|
||
// 2. PUT with updated metadata
|
||
let block_data = self.s3vfs.get_object(&format!("blocks/{}", hash))?;
|
||
self.s3vfs.put_object_with_metadata(
|
||
&format!("blocks/{}", hash),
|
||
&block_data,
|
||
[("x-amz-meta-ref-count", (current_ref + 1).to_string())]
|
||
)?;
|
||
|
||
Ok(())
|
||
}
|
||
```
|
||
|
||
**优势**：
|
||
- ✅ 简单实现
|
||
- ✅ 与 S3 标准兼容
|
||
- ⚠️ 需要两次请求（GET + PUT）
|
||
|
||
**劣势**：
|
||
- ⚠️ 非原子操作（并发问题）
|
||
- ⚠️ 需要读取块数据（PUT 需要 body）
|
||
|
||
---
|
||
|
||
### Solution 2: Separate Ref Count Object
|
||
|
||
**流程**：
|
||
```rust
|
||
fn increment_ref(&self, hash: &str) -> Result<(), VfsError> {
|
||
// 1. GET ref count object
|
||
let ref_key = format!("refs/{}/count", hash);
|
||
let current = self.s3vfs.get_object(&ref_key)
|
||
.and_then(|data| data.parse::<u64>())
|
||
.unwrap_or(0);
|
||
|
||
// 2. PUT updated ref count
|
||
self.s3vfs.put_object(&ref_key, (current + 1).to_string())?;
|
||
Ok(())
|
||
}
|
||
```
|
||
|
||
**优势**：
|
||
- ✅ 无需读取块数据
|
||
- ✅ 更小的对象（仅数字）
|
||
|
||
**劣势**：
|
||
- ⚠️ 需要额外对象存储
|
||
- ⚠️ 非原子操作（并发问题）
|
||
|
||
---
|
||
|
||
### Solution 3: MinIO Extended API (企业版)
|
||
|
||
MinIO 企业版提供 `mc admin bucket policy` 和 object locking API。
|
||
|
||
**优势**：
|
||
- ✅ 可能提供 atomic operation
|
||
|
||
**劣势**：
|
||
- ⚠️ 仅 MinIO 企业版
|
||
- ⚠️ 需要研究具体 API
|
||
|
||
---
|
||
|
||
## Concurrency Problem
|
||
|
||
### Scenario
|
||
|
||
Node A 和 Node B 同时 dedup 相同文件：
|
||
1. Node A: `increment_ref(hash-abc)` → GET count=2 → PUT count=3
|
||
2. Node B: `increment_ref(hash-abc)` → GET count=2 → PUT count=3
|
||
3. 结果：count=3（错误，应为 count=4）
|
||
|
||
### Solution 1: Optimistic Locking
|
||
|
||
使用 S3 versioning 检测冲突：
|
||
```rust
|
||
fn increment_ref(&self, hash: &str) -> Result<(), VfsError> {
|
||
loop {
|
||
// 1. GET current version + metadata
|
||
let (version_id, current_ref) = self.get_ref_with_version(hash)?;
|
||
|
||
// 2. PUT with version check
|
||
let result = self.s3vfs.put_object_if_version(
|
||
&format!("blocks/{}", hash),
|
||
block_data,
|
||
(current_ref + 1),
|
||
version_id // Only succeed if version unchanged
|
||
);
|
||
|
||
if result.is_ok() {
|
||
break;
|
||
}
|
||
// Retry if version mismatch
|
||
}
|
||
Ok(())
|
||
}
|
||
```
|
||
|
||
**要求**：MinIO versioning enabled。
|
||
|
||
---
|
||
|
||
### Solution 2: Distributed Lock Service
|
||
|
||
使用外部分布式锁（如 Redis/Zookeeper）：
|
||
```rust
|
||
fn increment_ref(&self, hash: &str) -> Result<(), VfsError> {
|
||
// 1. Acquire distributed lock
|
||
let lock = self.lock_service.acquire(&format!("lock:{}", hash))?;
|
||
|
||
// 2. Increment ref count
|
||
self.update_ref_count(hash)?;
|
||
|
||
// 3. Release lock
|
||
lock.release();
|
||
Ok(())
|
||
}
|
||
```
|
||
|
||
**劣势**：需要额外服务（Redis）。
|
||
|
||
---
|
||
|
||
### Solution 3: Accept Non-Atomic (简化方案)
|
||
|
||
对于 MarkBase Lightweight 定位：
|
||
- ⚠️ 接受非原子操作风险
|
||
- ⚠️ 偶尔 ref count 不准确（不影响数据完整性）
|
||
- ⚠️ 定期修复（scrub job）
|
||
|
||
**推荐**：Phase 1 使用 Solution 1（Metadata Update），Phase 2 研究 MinIO versioning。
|
||
|
||
---
|
||
|
||
## Implementation Phases
|
||
|
||
| Phase | Task | Code Lines | Priority | Risk |
|
||
|-------|------|------------|----------|------|
|
||
| **Phase 1** | DedupS3Store struct + basic I/O | ~300 | P0 | Medium |
|
||
| **Phase 2** | Reference count metadata | ~100 | P0 | Medium |
|
||
| **Phase 3** | Manifest storage to S3 | ~50 | P1 | Low |
|
||
| **Phase 4** | CLI integration | ~100 | P1 | Low |
|
||
| **Phase 5** | Async version (DedupAsyncS3Store) | ~200 | P2 | High |
|
||
| **Phase 6** | Concurrency fix (versioning) | ~150 | P2 | High |
|
||
| **Phase 7** | Performance benchmark | ~100 | P2 | Low |
|
||
| **Total** | | **~1000** | | |
|
||
|
||
---
|
||
|
||
## DedupS3Store Implementation (Phase 1 Draft)
|
||
|
||
```rust
|
||
use super::s3_fs::S3Vfs;
|
||
use super::{VfsDedupConfig, VfsError};
|
||
use sha2::{Sha256, Digest};
|
||
use std::path::Path;
|
||
|
||
pub struct DedupS3Store {
|
||
s3vfs: S3Vfs,
|
||
bucket: String,
|
||
block_prefix: String,
|
||
manifest_prefix: String,
|
||
config: VfsDedupConfig,
|
||
}
|
||
|
||
impl DedupS3Store {
|
||
pub fn new(
|
||
endpoint: &str,
|
||
region: &str,
|
||
bucket: &str,
|
||
access_key: &str,
|
||
secret_key: &str,
|
||
config: VfsDedupConfig,
|
||
) -> Result<Self, VfsError> {
|
||
let s3vfs = S3Vfs::new(endpoint, region, bucket, access_key, secret_key)?;
|
||
Ok(Self {
|
||
s3vfs,
|
||
bucket: bucket.to_string(),
|
||
block_prefix: "blocks/".to_string(),
|
||
manifest_prefix: "manifests/".to_string(),
|
||
config,
|
||
})
|
||
}
|
||
|
||
pub fn store_block(&self, data: &[u8]) -> Result<String, VfsError> {
|
||
if data.len() > self.config.block_size {
|
||
return Err(VfsError::Io(format!("Block size exceeds limit")));
|
||
}
|
||
|
||
let hash = Self::hash_block(data);
|
||
let key = format!("{}{}", self.block_prefix, hash);
|
||
|
||
// Check if block exists
|
||
if !self.s3vfs.object_exists(&key)? {
|
||
// PUT with initial ref count = 1
|
||
self.s3vfs.put_object_with_metadata(
|
||
&key,
|
||
data,
|
||
[("x-amz-meta-ref-count", "1")]
|
||
)?;
|
||
} else {
|
||
// Increment ref count
|
||
self.increment_ref(&hash)?;
|
||
}
|
||
|
||
Ok(hash)
|
||
}
|
||
|
||
pub fn get_block(&self, hash: &str) -> Result<Vec<u8>, VfsError> {
|
||
let key = format!("{}{}", self.block_prefix, hash);
|
||
self.s3vfs.get_object(&key)
|
||
}
|
||
|
||
pub fn increment_ref(&self, hash: &str) -> Result<(), VfsError> {
|
||
let key = format!("{}{}", self.block_prefix, hash);
|
||
let head = self.s3vfs.head_object(&key)?;
|
||
|
||
let current_ref = head.metadata
|
||
.get("x-amz-meta-ref-count")
|
||
.and_then(|v| v.parse::<u64>().ok())
|
||
.unwrap_or(1);
|
||
|
||
// Need to GET block data + PUT with new metadata
|
||
let block_data = self.get_block(hash)?;
|
||
self.s3vfs.put_object_with_metadata(
|
||
&key,
|
||
&block_data,
|
||
[("x-amz-meta-ref-count", (current_ref + 1).to_string())]
|
||
)?;
|
||
|
||
Ok(())
|
||
}
|
||
|
||
pub fn dedup_file(&self, source: &Path) -> Result<DedupManifest, VfsError> {
|
||
let mut file = std::fs::File::open(source)?;
|
||
let mut manifest = DedupManifest::new();
|
||
let mut buffer = vec![0u8; self.config.block_size];
|
||
|
||
loop {
|
||
let n = file.read(&mut buffer)?;
|
||
if n == 0 { break; }
|
||
|
||
manifest.original_size += n;
|
||
let hash = self.store_block(&buffer[..n])?;
|
||
manifest.block_hashes.push(hash);
|
||
}
|
||
|
||
// Store manifest to S3
|
||
let file_id = uuid::Uuid::new_v4().to_string();
|
||
manifest.file_id = file_id;
|
||
let manifest_key = format!("{}{}.json", self.manifest_prefix, file_id);
|
||
let manifest_json = serde_json::to_string(&manifest)?;
|
||
self.s3vfs.put_object(&manifest_key, manifest_json.as_bytes())?;
|
||
|
||
Ok(manifest)
|
||
}
|
||
|
||
pub fn restore_file(&self, manifest_id: &str, target: &Path) -> Result<(), VfsError> {
|
||
let manifest_key = format!("{}{}.json", self.manifest_prefix, manifest_id);
|
||
let manifest_json = self.s3vfs.get_object(&manifest_key)?;
|
||
let manifest: DedupManifest = serde_json::from_slice(&manifest_json)?;
|
||
|
||
let mut file = std::fs::File::create(target)?;
|
||
for hash in &manifest.block_hashes {
|
||
let block = self.get_block(hash)?;
|
||
file.write_all(&block)?;
|
||
}
|
||
|
||
Ok(())
|
||
}
|
||
|
||
fn hash_block(data: &[u8]) -> String {
|
||
let mut hasher = Sha256::new();
|
||
hasher.update(data);
|
||
hex::encode(hasher.finalize())
|
||
}
|
||
}
|
||
```
|
||
|
||
---
|
||
|
||
## Integration with MarkBase VFS
|
||
|
||
### Option 1: Standalone DedupS3Store
|
||
|
||
用户手动创建 DedupS3Store：
|
||
```bash
|
||
# CLI tool
|
||
markbase dedup-upload --s3 --s3-endpoint http://localhost:9000 --file /data/large.iso
|
||
markbase dedup-download --s3 --manifest-id <uuid> --output /data/restored.iso
|
||
```
|
||
|
||
---
|
||
|
||
### Option 2: DedupVfsBackend (VfsBackend trait)
|
||
|
||
创建 VfsBackend wrapper，自动 dedup：
|
||
```rust
|
||
pub struct DedupS3Backend {
|
||
dedup_store: DedupS3Store,
|
||
manifest_dir: PathBuf, // Local cache for manifests
|
||
}
|
||
|
||
impl VfsBackend for DedupS3Backend {
|
||
fn open_file(&self, path: &Path, flags: &OpenFlags) -> Result<Box<dyn VfsFile>, VfsError> {
|
||
// 1. Read manifest from S3
|
||
let manifest = self.load_manifest(path)?;
|
||
|
||
// 2. DedupS3File (read blocks from S3)
|
||
Ok(Box::new(DedupS3File::new(self.dedup_store.clone(), manifest)))
|
||
}
|
||
|
||
fn stat(&self, path: &Path) -> Result<VfsStat, VfsError> {
|
||
// Read from manifest metadata
|
||
let manifest = self.load_manifest(path)?;
|
||
Ok(VfsStat {
|
||
size: manifest.original_size,
|
||
mtime: manifest.mtime,
|
||
...
|
||
})
|
||
}
|
||
|
||
fn read_dir(&self, path: &Path) -> Result<Vec<VfsDirEntry>, VfsError> {
|
||
// List manifests from S3
|
||
self.dedup_store.s3vfs.list_objects(&self.manifest_prefix)
|
||
}
|
||
}
|
||
```
|
||
|
||
**优势**：
|
||
- ✅ 透明 dedup（用户无需关心）
|
||
- ✅ 与 SMB/WebDAV/SFTP 无缝集成
|
||
|
||
---
|
||
|
||
### Option 3: Hybrid (LocalFs + DedupS3Store)
|
||
|
||
```rust
|
||
pub struct HybridDedupBackend {
|
||
local: LocalFs, // Small files (<1MB) 存本地
|
||
dedup_s3: DedupS3Store, // Large files (>1MB) dedup to S3
|
||
}
|
||
|
||
impl VfsBackend for HybridDedupBackend {
|
||
fn open_file(&self, path: &Path, flags: &OpenFlags) -> Result<Box<dyn VfsFile>, VfsError> {
|
||
// Check file size
|
||
let stat = self.local.stat(path)?;
|
||
|
||
if stat.size < self.dedup_s3.config.min_file_size {
|
||
// Small file: direct LocalFs
|
||
self.local.open_file(path, flags)
|
||
} else {
|
||
// Large file: dedup to S3
|
||
self.dedup_s3.dedup_file(path)?;
|
||
self.dedup_s3.open_file_from_manifest(path)
|
||
}
|
||
}
|
||
}
|
||
```
|
||
|
||
**推荐**：Option 1（Phase 1），Option 3（Phase 2）。
|
||
|
||
---
|
||
|
||
## Performance Considerations
|
||
|
||
### Network Latency
|
||
|
||
| Operation | LocalFs | S3Vfs | Overhead |
|
||
|-----------|---------|-------|----------|
|
||
| store_block (4KB) | ~0.1ms | ~5-10ms (HTTP) | ~50-100x |
|
||
| get_block (4KB) | ~0.1ms | ~5-10ms (HTTP) | ~50-100x |
|
||
| dedup_file (100MB) | ~2s (25MB/s) | ~10s (10MB/s) | ~5x |
|
||
|
||
**缓解方案**：
|
||
- ✅ Async concurrent upload（4-8 并发）
|
||
- ✅ ReadCache（64MB cache）
|
||
- ✅ Local cache for hot blocks
|
||
|
||
---
|
||
|
||
### Dedup Ratio Impact
|
||
|
||
| File Type | Dedup Ratio | Network Traffic Saved |
|
||
|-----------|-------------|----------------------|
|
||
| VM images (similar OS) | ~80% | -80% upload bandwidth |
|
||
| Log files (daily) | ~60% | -60% upload bandwidth |
|
||
| Unique files (photos) | ~5% | -5% upload bandwidth |
|
||
|
||
---
|
||
|
||
## Next Steps
|
||
|
||
1. **Phase 1 Implementation** (~300 lines)
|
||
- `DedupS3Store` struct
|
||
- `store_block()` / `get_block()` via S3Vfs
|
||
- `increment_ref()` with metadata update
|
||
|
||
2. **Phase 2 CLI Integration** (~100 lines)
|
||
- `markbase dedup-upload --s3`
|
||
- `markbase dedup-download --manifest-id`
|
||
|
||
3. **Phase 3 Performance Test**
|
||
- Benchmark dedup_file (100MB)
|
||
- Compare LocalFs vs S3Vfs
|
||
|
||
---
|
||
|
||
## Open Questions
|
||
|
||
1. **Concurrency**: Accept non-atomic ref count vs implement versioning?
|
||
2. **Backend choice**: Standalone CLI vs VfsBackend integration?
|
||
3. **Min versioning**: Should we require MinIO versioning enabled?
|
||
4. **Ref count object**: Metadata vs separate object?
|
||
5. **Block cache**: Should we cache blocks locally?
|
||
|
||
---
|
||
|
||
**文档创建**: 2026-06-25
|
||
**最后更新**: 2026-06-25 |