Requirements

Functional Requirements:

• Support basic file operations: read, write, append, and delete files.
• Maintain metadata for each file including permissions, timestamps, file sizes, and storage locations.
• Replicate data across multiple servers for fault tolerance.
• Support concurrent access to files while maintaining data consistency.

Non-Functional Requirements:

• Durability: 0 data loss after write acknowledgement - survive simultaneous failure of N-1 replicas (typically N=3)
• High availabiity: Reads available even when minority of nodes are down; writes available as long as quorum is reachable
• scalability: petabytes of storage, billion of files, tohusands of concurrent clients
• throughput optimized: design for large sequential reads/writes (not small random IO) - target GB/s aggregate throughput
• COnsistency: Strong consistency for metadata; tunable consistency for data reads (reads your write guaranteed)
• Fault tolerance: Automatic detection and. recovery from node failures without manual intervention
• Low metadata ltency: file open/stat operations < 10ms p99

API Design

File Operations

Post /v1/files

Body: {path, replication_factor, block_size_mb}

Response: {file_id, created_at}

GET /v1/files/{file_id}/read?offset=0&length=blah

Response:

POST /v1/files/{file_id}/append

Body:

Response: {bytes_appended, new_size}

DELETE /v1/files/{file_id}

Response: 204 No Content

Metadata Operations

GET /v1/files/{file_id}/metadata

Response: {

path

size

block_size

replication factor

blocks:

permissions:

created_at

modified at

}

Locking (for conc access)

POST /v1/locks

Body: {file_id, type, ttl_seconds

Response: {lock_id, expires_at}

DELETE /v1/locks/{lock_id}

Response: 204 No Content

High-Level Design

The architecture follows the GFS/HDFS model: separate metadata management from data storage

Core Insight:

Files are split into fixed-size chunks (64-128mb). Metadata (namespace, chunk locations) is managed by a centralized master. Data flows directly between clients and chunk servers - the master is never in the data path.

Components:

1. Master Node (name node) - single logical node managing all metadata: file namespace (directory tree), file -> chunk mapping, chunk->server mapping. Replicated for HA.
2. Chunk Servers (Data Nodes) - Store actual chunk data on local disks. Report health and chunk inventory to master via heartbeats.
3. Client Library - Thin SDK that talks. to master for metadata, then directly to chunk servers for data. Caches metadata locally.
4. Replication Manager - Ensures each chunk maintains its target replica count. Triggers re-replication on node failure.
5. Lease Manager - Grants time-limited leases for write access to chunks (consistency without heavy locking)

Detailed Component Design

Master Node (Metadata Management)

What it store (all in RAM f or speed)

• Namespace (directory tree) -> prefix tree with read-write locks per node
• File -> chunk list: Hash map file id -> [chunk ids]
• Chunk -> locations: chunk id -> [datanode_ids]
• chunk leases: chunk id -> {leaseholder, expiry}

Scaling math: 1 bill files x 100 bytes = 100GB metadata. Fits in RAM on a large machine. Each file averages 2 chunks -> 2B chunk mappings x 48 bytes = ~96 GB. Total 200GB RAM - achievable on modern hardware

Persistence:

• Operation log: Every metadata mutation (create, delete, rename) is appended to a write-ahead log on disk before being applied in memory. Synchronously replicted to standby masters.
• Checkpoints: Periodic full snapshot of in-memory state to disk. Recovery = load latest checkpoint -> replay log

High availability:

• Active-standy with Raft consensus (or Zookeeper for leader election)
• Standby replays the operation log in real-time - failover in <10 seconds
• Chunk locations are NOT persisted - rebuilt from chunk server heartbeats on startup (they're the source of truth for what they store)

Why centralized master works: The master handles only metadata operations (open, stat, ls) - never data. At 10k metadata operations per sec with <1ms per op, a single master handles most workloads. For extreme scale (10B+files), shard the namespace across multiple masters by directory subtree (as HDFS federation does)

2. Chunk Storage and Replication Pipeline

Chunk structure on disk:

/data/chunks

blk1.dat (64mb data)

blk1.meta (checksum per 64kb block, version)

Write path (pipeline replication):

Client wants to write chunk to replicas (DN1, DN3, DNS)

1. Client asks mater "I want to write to file X, chunk 2"
2. Master grants lease to DN1 (primary), return sreplica list [DN1, DN3, DN5]
3. Client pushes data. to DN1
4. DN1 forwards to DN3 -> DN3 forwards to DN5 (pipeline, not fan-out)
5. Acks flow back: DN5 -> DN3 -> DN1 -> Client
6. Client receieves ack = data on all 3 replicas

Why pipeline, not fan-out?

• Fan out: client sends 64mb x 3 = 192mb of outbound traffic
• Pipeline: client sends 64mb once. Each hop sends 64mb. Total network usage the same, but client bandwidth is 3x lower. And dta flows as a stream - DN1 starts forwarding to DN3 before it has the full chunk.

Consistency via leases:

• The master grants a write lease on a chunk to one chunk server (the primary)
• All writes to that chunk go through the primary, which assigns a serial order.
• The primary applies writes in order and instructs secondaries to do the same.
• Lease expires after 60s - primary must renew. If primary dies, master waits for lease expiry, then grants to a new primary.

Read path:

1. Client asks master: Where is file X chunk 2
2. Master returns DN1, DN2, DN5 + chunk version
3. CLient reads fro mclosest replica (network topology-aware)
4. Client verifies checksum on received state

Fault tolerance and re-replication:

Failure detection:

• chunk servers send heartbeats to master evry 3 seconds
• If no heartbeat for 30 seconds -> master marks node as dead
• All chunks on that node as now under-replicated

Rereplication:

Master detects: chunk blk1 has 2 replicas (target: 3)

• Selects a new target DN (considering rack diversity, disk utilization, load)
• instructs an existing replica to copy the chunk to the new target
• priority queue: chunks. with fewer replicas get rereplicated. first
  • (1 replica = critical, 2 urgent, etc)

Rack-aware placement: Replicas are placed across failure domains:

• 2 replicas on different nodes in the same rack (fast intra-rack replication)
• 1 replica on a different rack (survives full rack failure)

Data integrity:

• each 64kb subblock has a CRC32 checksum stored in the .meta file.
• Chunk servers verify checksums on every read and during periodic background scans
• Corruption detected -> report to master -> master treats that replica as lost -> triggers re-replication from a healthy replica

Garbage collection

• Deleted files renamed to hidden path (trash). Actual chunk deleetion happens after a configurable grace period (e.g. 3 days)
• Orphaned chunks (no file references them) detected during periodic scans and reclaimed.

Key Tradeoffs

Large chunk size (64mb) -> wastes space for smll files vs fewer chunks to track, higher throughput for large sequntial IO

Centralized master -> single metadta bottleneck vs simple consistency, no distributed consensus for evey op

Pipeline replication vs fanout -> higher latency vs lower client bandwidth, streming starts immediately

Lease based consistency vs distributed locks: Lease expiry delay on failure vs simplier protocol, no deadlocks

In-memory metadata (RAM cost) vs microsecond metadata lookups

Rack-aware placement - constrains placement choices vs survives full rack failure

Lazy garbage collection - disk not reclaimed immediately vs accidental delete recovery -> simpler deletion path