Hands-On: Building a Distributed File Sharing System

March 18, 2024 Network P2P, File Sharing, Rust, Piece Download, System Architecture Network Development 1879 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
    subgraph Application
        CLI["CLI Interface"]
        API["REST API"]
    end

    subgraph Business Logic
        Index["File Index Manager"]
        Scheduler["Piece Download Scheduler"]
        Verify["Verify & Reassemble"]
    end

    subgraph P2P Network
        Discovery["Kad-DHT<br/>Peer Discovery + Metadata"]
        Broadcast["GossipSub<br/>Message Broadcast"]
        Transfer["Custom Protocol<br/>File Transfer"]
    end

    CLI --> Index
    API --> Index
    Index --> Scheduler
    Scheduler --> Verify
    Index --> Discovery
    Scheduler --> Transfer
    Broadcast -->|"New Peer Notification"| Index

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 LR
    S["Seeder Node"] -->|"1. Chunk and hash file"| FM["FileMetadata<br/>{filename, piece_size,<br/> piece_hashes[]}"]
    FM -->|"2. hash(fileID) as DHT key<br/>store on K closest nodes"| DHT["Kademlia DHT"]
    D["Downloader Node"] -->|"3. Query DHT with fileID"| DHT
    DHT -->|"4. Return FileMetadata"| D
    D -->|"5. Start per-Piece download"| S

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
    KW["Peer has Piece set"] --> B["Count replicas of<br/>each Piece globally"]
    B --> C{"Select rarest<br/>missing Piece"}
    C --> D["Request Piece from<br/>a peer that has it"]
    D --> E{"Download OK?"}
    E -->|"Yes"| F["Verify SHA-256"]
    E -->|"No"| G["Mark peer unavailable<br/>Pick another"]
    F -->|"Passed"| H["Mark Complete<br/>Notify peers"]
    F -->|"Failed"| I["Re-request Piece"]
    H --> J{"All Pieces<br/>Complete?"}
    J -->|"No"| A
    J -->|"Yes"| K["Reassemble file"]

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 Size	Block Size	Max Blocks
≤ 250 MB	128 KiB	~2000
1 GB	512 KiB	~2000
4 GB	2 MiB	~2000
> 16 GB	16 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:

Type	Local Changes	Remote Changes	Use Case
Send & Receive	Send to cluster	Receive from cluster	Default bidirectional sync
Send Only	Send to cluster	Ignored	NVR recording storage (write-only)
Receive Only	Ignored	Receive from cluster	Edge devices, archive replicas
Receive Encrypted	Ignored	Receive but can’t decrypt	Untrusted 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

Cohen, B. (2003). Incentives build robustness in BitTorrent. Workshop on Economics of Peer-to-Peer Systems.
IPFS. https://ipfs.tech/
anacrolix/torrent (Go BitTorrent). https://github.com/anacrolix/torrent
rust-libp2p examples. https://github.com/libp2p/rust-libp2p/tree/master/examples
Syncthing source. https://github.com/syncthing/syncthing
BEP Protocol Spec. https://docs.syncthing.net/specs/bep-v1.html
Syncthing Architecture Docs. https://docs.syncthing.net/dev/

Part of series: Network Development

← Previous Go P2P Development: From Basics to PubSub Next → P2P Production Best Practices