Mi&Bee Blog
中文

Hands-On: Building a Distributed File Sharing System

Network P2P, File Sharing, Rust, Piece Download, System Architecture Network Development Practice 1881 words 9 min read
🔊

Combining theory with practice, we’ll build a real distributed file sharing system. This system will leverage technologies introduced in previous articles — Kademlia DHT for node discovery and metadata distribution, Gossipsub for broadcasting, and a custom file transfer protocol.

System Architecture

mermaid
flowchart TD
    APP["Application<br/>CLI / REST API"] --> LOGIC["Business Logic<br/>File index / download scheduler / verify"]
    LOGIC --> NET["P2P Network<br/>Kad-DHT / Gossipsub / file transfer"]
    NET -.->|"new peer notification"| LOGIC
    style APP fill:#2196F3,color:#fff
    style LOGIC fill:#FF9800,color:#fff
    style NET fill:#4CAF50,color:#fff

Three strictly separated layers: application faces outward, business logic handles scheduling and verification, and the P2P network layer handles peer discovery, message broadcast, and file transfer.

Core design principles:

  • Layered abstraction: Business logic strictly separated from P2P network layer
  • Modularity: Each component independently testable
  • Fault tolerance: Node failures don’t affect overall system availability

Metadata Distribution

How does a downloader know which Pieces a file has and their hashes? This is done through DHT-based FileMetadata distribution:

mermaid
flowchart TD
    S["Seeder<br/>chunk + hash"] --> FM["FileMetadata<br/>store in DHT"]
    FM --> DHT["Kademlia DHT<br/>key = hash(fileID)"]
    D["Downloader"] -->|"query fileID"| DHT
    DHT -->|"return metadata"| D
    D -->|"download per Piece"| S
    style S fill:#2196F3,color:#fff
    style FM fill:#FF9800,color:#fff
    style DHT fill:#9C27B0,color:#fff
    style D fill:#4CAF50,color:#fff

The Seeder stores file metadata (including each Piece’s SHA-256 hash) in the DHT keyed by fileID. The Downloader only needs to know fileID to retrieve complete metadata and start downloading. fileID is typically the hash of file content, so identical files always map to the same ID.

Rust Core Module Implementation

Piece State Management

First, define the Piece status enum and thread-safe shared state:

rust
 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19

use std::collections::{HashMap, HashSet};
use std::sync::Arc;
use tokio::sync::RwLock;

#[derive(Debug, Clone, PartialEq)]
enum PieceStatus {
    Missing,     // Not yet downloaded
    Downloading, // Currently downloading from a peer
    Complete,    // Downloaded and verified
}

/// Thread-safe download scheduler
/// Arc<RwLock<>> allows multiple async tasks to safely read/write concurrently
struct DownloadScheduler {
    metadata: FileMetadata,
    piece_status: Arc<RwLock<Vec<PieceStatus>>>,
    peer_pieces: Arc<RwLock<HashMap<PeerId, HashSet<usize>>>>,
    output_path: String,
}

Why Arc<RwLock<>>? In async P2P programs, multiple Pieces may download simultaneously from different peers, each as an independent async task. These tasks need to concurrently update piece_status (marking Pieces as Downloading/Complete). RwLock allows multiple tasks to read state simultaneously (non-blocking) while exclusive access during writes prevents data races. Arc provides shared ownership across tasks.

File Chunking and Verification

rust
 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37

use sha2::{Sha256, Digest};
use std::fs::File;
use std::io::{Read, Write, Seek, SeekFrom};

const PIECE_SIZE: usize = 256 * 1024;

#[derive(Debug, Clone, Serialize, Deserialize)]
struct FileMetadata {
    filename: String,
    total_size: u64,
    piece_size: usize,
    piece_hashes: Vec<String>,
}

impl FileMetadata {
    fn from_file(path: &str) -> anyhow::Result<Self> {
        let mut file = File::open(path)?;
        let total_size = file.metadata()?.len();
        let mut piece_hashes = Vec::new();
        let mut buffer = vec![0u8; PIECE_SIZE];
        loop {
            let n = file.read(&mut buffer)?;
            if n == 0 { break; }
            let mut hasher = Sha256::new();
            hasher.update(&buffer[..n]);
            piece_hashes.push(hex::encode(hasher.finalize()));
        }
        Ok(Self { filename: path.to_string(), total_size, piece_size: PIECE_SIZE, piece_hashes })
    }

    fn verify_piece(&self, index: usize, data: &[u8]) -> bool {
        if index >= self.piece_hashes.len() { return false; }
        let mut hasher = Sha256::new();
        hasher.update(data);
        hex::encode(hasher.finalize()) == self.piece_hashes[index]
    }
}

Rarest First Download Scheduling

mermaid
flowchart TD
    B["Count replicas of<br/>each Piece globally"] --> C{"Select rarest<br/>missing Piece"}
    C --> D["Request from a peer<br/>that has it"]
    D --> E{"SHA-256 verify<br/>passed?"}
    E -->|"passed"| H["Mark Complete"]
    E -->|"failed / timeout"| B
    H -->|"still missing"| B
    H -->|"all Complete"| K["Reassemble file"]
    style B fill:#2196F3,color:#fff
    style C fill:#9C27B0,color:#fff
    style D fill:#FF9800,color:#fff
    style E fill:#9C27B0,color:#fff
    style H fill:#4CAF50,color:#fff
    style K fill:#4CAF50,color:#fff
rust
 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49

impl DownloadScheduler {
    fn new(metadata: FileMetadata, output_path: String) -> Self {
        let piece_count = metadata.piece_hashes.len();
        Self {
            metadata,
            piece_status: Arc::new(RwLock::new(vec![PieceStatus::Missing; piece_count])),
            peer_pieces: Arc::new(RwLock::new(HashMap::new())),
            output_path,
        }
    }

    async fn select_next_piece(&self, peer: PeerId) -> Option<usize> {
        let status = self.piece_status.read().await;
        let peer_map = self.peer_pieces.read().await;
        let peer_has = peer_map.get(&peer)?;

        let mut rarest: Vec<(usize, usize)> = status
            .iter().enumerate()
            .filter(|(i, s)| *s == PieceStatus::Missing && peer_has.contains(i))
            .map(|(i, _)| {
                let count = peer_map.values()
                    .filter(|pieces| pieces.contains(&i)).count();
                (i, count)
            }).collect();
        rarest.sort_by_key(|&(_, count)| count);
        rarest.first().map(|&(i, _)| i)
    }

    async fn write_piece(&self, index: usize, data: &[u8]) -> anyhow::Result<()> {
        if !self.metadata.verify_piece(index, data) {
            anyhow::bail!("Piece {} hash verification failed", index);
        }

        let mut file = OpenOptions::new()
            .write(true).create(true).open(&self.output_path)?;
        let offset = index as u64 * self.metadata.piece_size as u64;
        file.seek(SeekFrom::Start(offset))?;
        file.write_all(data)?;

        let mut status = self.piece_status.write().await;
        status[index] = PieceStatus::Complete;
        Ok(())
    }

    async fn is_complete(&self) -> bool {
        let status = self.piece_status.read().await;
        status.iter().all(|s| *s == PieceStatus::Complete)
    }
}

Seeder Side: Serving Piece Requests

The seeder node must listen for download requests and return data by Piece index:

rust
 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23

async fn handle_piece_request(
    metadata: &FileMetadata,
    file_path: &str,
    piece_index: usize,
) -> anyhow::Result<Vec<u8>> {
    if piece_index >= metadata.piece_hashes.len() {
        anyhow::bail!("Invalid piece index");
    }

    let mut file = File::open(file_path)?;
    let offset = piece_index as u64 * metadata.piece_size as u64;
    file.seek(SeekFrom::Start(offset))?;

    let remaining = metadata.total_size - offset;
    let read_size = std::cmp::min(metadata.piece_size as u64, remaining) as usize;
    let mut data = vec![0u8; read_size];
    file.read_exact(&mut data)?;

    if !metadata.verify_piece(piece_index, &data) {
        anyhow::bail!("Local file corrupted at piece {}", piece_index);
    }
    Ok(data)
}

Running and Testing

Start a seeder node:

bash
1
2
3
4

cargo run -- serve --file ./ubuntu-22.04.iso
# Output: File ID: QmYyQSo1c1Ym7orWxLYvCrM2EmxFTANf8wXmmE7DWjhx5N
# Output: Metadata published to DHT
# Output: Listening for piece requests...

Start a downloader node:

bash
1
2
3

cargo run -- download \
    --peer /ip4/127.0.0.1/tcp/4001/p2p/Qm... \
    --file-id QmYyQSo1c1Ym7orWxLYvCrM2EmxFTANf8wXmmE7DWjhx5N

Download Flow

mermaid
sequenceDiagram
    participant D as Downloader
    participant T as DHT
    participant S as Seeder

    D->>T: Query fileID metadata
    T-->>D: Return FileMetadata (piece_hashes[])
    D->>S: Request Piece 0 (rarest first)
    S-->>D: Send Piece 0 data
    Note over D: Verify SHA-256 | Mark Complete
    D->>S: Request Piece 1
    S-->>D: Send Piece 1 data
    Note over D: All Pieces Complete
    Note over D: Reassemble in order

Production Reference: Syncthing Sync Engine Design

Syncthing is a production-grade P2P file synchronization engine (85K+ stars), employing fundamentally different design strategies from the BitTorrent-style approach that this system is based on. The following analysis covers five key dimensions, offering design insights for custom implementations.

Variable Block Sizing

This system uses a fixed 256KB piece size. Syncthing dynamically selects the optimal block size based on file size:

File SizeBlock SizeMax Blocks
≤ 250 MB128 KiB~2000
1 GB512 KiB~2000
4 GB2 MiB~2000
> 16 GB16 MiB~2000

The design rule is: select the smallest block size that keeps the block count under 2000. This ensures fine granularity for small files (128 KiB blocks) while avoiding excessive metadata for large files (metadata grows with block count, not file size). The 2000-block limit stems from BEP’s Index message Protobuf encoding efficiency, ensuring a single Index transmission completes within a typical MTU.

Version Vector Conflict Resolution

When two devices modify the same file simultaneously, a conflict occurs — one of the hardest problems in distributed file synchronization. Syncthing uses Version Vectors to detect and resolve conflicts:

go
1
2
3
4
5
6
7
8
9

// lib/protocol/vector.go
type Vector struct {
    Counters []Counter // Sorted array of {ID, Value} pairs
}

type Counter struct {
    ID    uint64 // Device short ID
    Value uint64 // Monotonically increasing counter
}

Version vector comparison yields one of five relationships:

go
1
2
3
4
5
6
7

const (
    Equal             // Identical
    Greater           // Strictly newer
    Lesser            // Strictly older
    ConcurrentLesser  // Concurrent but sorts as older
    ConcurrentGreater // Concurrent but sorts as newer
)
mermaid
flowchart TD
    A["Device A<br/>{A:3} → {A:4}"]
    B["Device B<br/>{B:2} → {B:3}"]
    C{"Comparison?"}
    A --> C
    B --> C
    C -->|"Concurrent"| Conflict["Keep both<br/>Create .conflict"]
    C -->|"Greater/Lesser"| Sync["Direct overwrite"]

    style Conflict fill:#FF9800,color:#fff
    style Sync fill:#4CAF50,color:#fff

Unlike the DHT single-key metadata distribution in this system, version vectors precisely describe the causal relationships of file changes: when two version vectors are neither Greater than the other, they are Concurrent — triggering conflict handling that preserves both versions (generating .conflict files) to ensure no data loss. This is the most critical data safety guarantee in P2P file synchronization.

Delta Index Exchange

This system uses DHT for full metadata storage with complete pull by the downloader. Syncthing employs delta index exchange instead:

When two devices connect, they first exchange IndexID (a UUID identifying the local index) and Sequence (index sequence number). If the remote device’s IndexID matches the locally recorded value, only changes after the Sequence need to be transmitted — this is the essence of incremental sync:

go
1
2
3
4
5
6
7
8

// Pseudo-code: delta index exchange logic
if local.indexID == remote.indexID {
    // Only send changes after Sequence
    sendDelta(local.sequence, remote.sequence)
} else {
    // Full index
    sendFullIndex()
}

This design is highly efficient for frequent reconnection scenarios: after device sleep/wake cycles, only recent changes need to be exchanged rather than a full sync. The same principle can be applied to custom systems — maintaining a sequence-driven delta channel alongside DHT.

Folder Type Semantics

Syncthing defines four folder types, each with different sync semantics:

TypeLocal ChangesRemote ChangesUse Case
Send & ReceiveSend to clusterReceive from clusterDefault bidirectional sync
Send OnlySend to clusterIgnoredNVR recording storage (write-only)
Receive OnlyIgnoredReceive from clusterEdge devices, archive replicas
Receive EncryptedIgnoredReceive but can’t decryptUntrusted device hosting

The Receive Encrypted type is particularly noteworthy — a device can store encrypted data without being able to decrypt it, as the private key stays on authorized devices. This offers an “edge storage” design pattern for custom file sharing systems: trusted nodes handle reads and writes, while untrusted nodes only provide storage space.

Parallel Hash Computation

Syncthing’s file scanner uses a parallel hasher pool to accelerate large file processing:

go
 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13

type parallelHasher struct {
    workChan chan *hasherWork
    outChan  chan *hashResult
    workers  int // Default 2 parallel workers
}

func (ph *parallelHasher) HashFile(ctx context.Context, path string) {
    // 1. Initial Stat
    // 2. Read chunks and send to workChan
    // 3. Parallel workers compute SHA-256
    // 4. Final Stat to verify file unchanged
    // 5. Aggregate results from all workers
}

A key design detail: Stat() is called both before and after hash computation. If the file’s mtime or size changes during computation, the result is discarded — ensuring stale hashes are never used. This check can be applied equally well in Rust implementations.

In contrast, this system uses serial hash computation (single-threaded read → hash → next block). On multi-core systems, Syncthing’s parallel hasher pool approach can be adopted to accelerate large file scanning.

References

Part of series: Network Development Practice
← Previous Go P2P Development: From Basics to PubSub Next → P2P Production Best Practices