markbase/docs/DEDUP_S3_COMBINATION.md

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

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 (推荐 ⭐⭐⭐⭐⭐)

流程：

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

流程：

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 检测冲突：

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）：

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)

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：

# 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：

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)

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