Designing a Distributed File System: Metadata Servers, Chunk Replication, and Consistency Guarantees at Scale

Most engineers have used a distributed file system without thinking much about what breaks when you push it past its comfort zone. You hit a network partition during a write, or your metadata server becomes a bottleneck at 100 million files, and suddenly the design decisions that seemed academic become the ones that determine your uptime.

This article covers the architectural core: the split between metadata and data, how chunk replication actually works, how lease-based ordering keeps mutations consistent, and where the real tradeoffs live between strong consistency and availability.

The Fundamental Split: Metadata vs. Data Plane

Every production distributed file system separates metadata from data. This is not an aesthetic choice. It is a scaling constraint.

Metadata operations (open, stat, rename, delete, list) need strong consistency and low latency. Data operations (read, write large files) need high throughput and need to scale horizontally without touching the metadata tier.

In GFS, the master holds all file namespace metadata, chunk location maps, and lease assignments in memory. Chunk servers hold the actual data. In HDFS, the NameNode is the metadata server, and DataNodes hold blocks.

The memory constraint is real. GFS reported roughly 64 bytes of metadata per chunk. A petabyte of data stored in 64 MB chunks produces about 16 million chunks, which fits in memory. If you shrink the chunk size or grow the dataset, the metadata tier is your ceiling.

interface ChunkHandle {
  id: bigint;
  version: number;
}

interface ChunkLocationMap {
  handle: ChunkHandle;
  primary: string;       // primary replica host
  secondaries: string[]; // secondary replica hosts
  leaseExpiry: number;   // unix timestamp
}

interface FileMetadata {
  path: string;
  chunkCount: number;
  chunks: ChunkHandle[];
  createdAt: number;
  modifiedAt: number;
}

class MetadataServer {
  private namespace: Map<string, FileMetadata> = new Map();
  private chunkMap: Map<bigint, ChunkLocationMap> = new Map();
  private nextChunkId: bigint = 0n;

  allocateChunk(filePath: string): ChunkHandle {
    const handle: ChunkHandle = {
      id: this.nextChunkId++,
      version: 1,
    };
    const file = this.namespace.get(filePath);
    if (!file) throw new Error(`File not found: ${filePath}`);
    file.chunks.push(handle);
    file.modifiedAt = Date.now();
    return handle;
  }

  getChunkLocations(handle: ChunkHandle): ChunkLocationMap | undefined {
    return this.chunkMap.get(handle.id);
  }
}

The metadata server writes every mutation to an operation log before acknowledging it to clients. The log is the source of truth. On restart, the server replays the log to rebuild in-memory state, checkpointing periodically so replay stays bounded.

This is where HDFS added federated namespaces: multiple NameNodes each owning a subtree of the filesystem. This breaks the single-server ceiling but introduces cross-namespace operation complexity.

Chunk Replication: Placement and Propagation

The replication factor (typically 3) is not just about durability. It is about read distribution and fault tolerance under correlated failures.

Placement policy matters more than most engineers initially think. Naive placement puts all three replicas on three different servers. Better placement puts one replica on a local rack and two on different remote racks. This way, a rack failure (power, top-of-rack switch) does not take out a majority of replicas, but intra-rack bandwidth is available for the first copy.

interface ReplicaPlacement {
  rack: string;
  host: string;
}

function selectReplicaLocations(
  availableNodes: { host: string; rack: string; freeBytes: number }[],
  replicationFactor: number,
  preferredRack?: string
): ReplicaPlacement[] {
  const byRack = new Map<string, typeof availableNodes>();
  for (const node of availableNodes) {
    if (!byRack.has(node.rack)) byRack.set(node.rack, []);
    byRack.get(node.rack)!.push(node);
  }

  const placements: ReplicaPlacement[] = [];

  // First replica: local rack (or any if no preference)
  const localRack = preferredRack ?? availableNodes[0].rack;
  const localCandidates = byRack.get(localRack) ?? [];
  if (localCandidates.length > 0) {
    const selected = localCandidates.sort((a, b) => b.freeBytes - a.freeBytes)[0];
    placements.push({ rack: selected.rack, host: selected.host });
  }

  // Remaining replicas: different racks, most free space first
  const otherRacks = [...byRack.keys()].filter(r => r !== localRack);
  for (const rack of otherRacks) {
    if (placements.length >= replicationFactor) break;
    const candidates = byRack.get(rack)!.sort((a, b) => b.freeBytes - a.freeBytes);
    if (candidates.length > 0) {
      placements.push({ rack: candidates[0].rack, host: candidates[0].host });
    }
  }

  return placements;
}

Write propagation uses a pipeline. The client writes to the primary; the primary forwards to the first secondary; that secondary forwards to the next. This saturates network bandwidth more efficiently than fan-out from the client. Each hop in the chain acknowledges back to the previous node once the data is received and written.

Re-replication is triggered when the metadata server detects a chunk is under-replicated, either because a chunk server went down or a disk failed. The server schedules re-replication from an existing replica to a new host, prioritizing chunks with the fewest remaining copies.

Lease-Based Mutation Ordering

Without a mechanism to designate a single authoritative writer, concurrent mutations to the same chunk produce undefined results. GFS solves this with leases.

The metadata server grants a lease to one chunk server, making it the primary for that chunk. The lease has a timeout (typically 60 seconds). While the lease is held, all mutations to that chunk must go through the primary. The primary assigns serial mutation numbers and forwards them to secondaries in order. All replicas apply mutations in the same sequence.

interface Lease {
  chunkId: bigint;
  primary: string;
  grantedAt: number;
  durationMs: number;
}

class LeaseManager {
  private leases: Map<bigint, Lease> = new Map();

  grantLease(chunkId: bigint, primary: string, durationMs = 60_000): Lease {
    const existing = this.leases.get(chunkId);
    if (existing && this.isValid(existing)) {
      throw new Error(`Lease already held by ${existing.primary}`);
    }
    const lease: Lease = {
      chunkId,
      primary,
      grantedAt: Date.now(),
      durationMs,
    };
    this.leases.set(chunkId, lease);
    return lease;
  }

  renewLease(chunkId: bigint, requestor: string): Lease {
    const lease = this.leases.get(chunkId);
    if (!lease || lease.primary !== requestor) {
      throw new Error("Lease not held or requestor mismatch");
    }
    lease.grantedAt = Date.now(); // extend from now
    return lease;
  }

  isValid(lease: Lease): boolean {
    return Date.now() < lease.grantedAt + lease.durationMs;
  }

  revoke(chunkId: bigint): void {
    this.leases.delete(chunkId);
  }
}

The primary also enforces write ordering by assigning monotonically increasing serial numbers per mutation:

class PrimaryChunkServer {
  private mutationCounter = 0;
  private pendingMutations: Map<number, Buffer> = new Map();

  async applyMutation(data: Buffer, secondaries: string[]): Promise<void> {
    const serial = ++this.mutationCounter;
    this.pendingMutations.set(serial, data);

    // Forward to all secondaries in parallel, each applies in serial order
    await Promise.all(
      secondaries.map(addr => this.forwardToSecondary(addr, serial, data))
    );

    // Commit locally after secondaries acknowledge
    await this.commitToStorage(serial, data);
    this.pendingMutations.delete(serial);
  }

  private async forwardToSecondary(
    addr: string,
    serial: number,
    data: Buffer
  ): Promise<void> {
    // network call; secondary buffers and applies in serial order
    await fetch(`http://${addr}/apply`, {
      method: "POST",
      body: JSON.stringify({ serial, data: data.toString("base64") }),
    });
  }

  private async commitToStorage(serial: number, data: Buffer): Promise<void> {
    // write to local disk
  }
}

If the primary crashes during a write, the metadata server waits for the lease to expire before granting a new lease to another replica. This timeout-based approach ensures the old primary cannot apply any more mutations while the new primary is being established. The 60-second wait is the cost of this safety guarantee.

Consistency Model: What GFS Actually Gives You

GFS does not give strong consistency. It gives a defined consistency model that is weaker than most engineers assume. Understanding the actual guarantees prevents debugging sessions that last days.

A file region is defined if after a mutation, all clients see the mutation and agree on the data written. A region is consistent if all clients see the same data, but it may or may not reflect a specific mutation. A region is inconsistent if different clients see different data.

Mutation Type	Concurrent Mutations	Result
Serial write to defined offset	No	Defined
Concurrent writes to same region	Yes	Consistent, undefined
Serial record append	No	Defined (with padding)
Concurrent record append	Yes	Defined (each append atomic, may have gaps)

Record append is the primitive GFS optimizes for. The append is atomic: the primary picks the offset, all replicas write at that offset, and the offset is returned to the client. Under concurrent appends, the file may have duplicates if a replica acknowledged but the client retried on timeout, and it may have padding gaps where an append was abandoned and re-tried at a later offset.

Applications that use GFS are expected to handle this: checksums to detect corruption, unique record IDs to detect duplicates, and offset-based reads to skip padding.

HDFS took a stricter path: one writer at a time, pipeline of DataNodes, and a lease revocation mechanism that allows a single client to hold the write lease. This gives defined behavior for writes but constrains concurrent access patterns.

Metadata Replication and Shadow Masters

The metadata server is a single point of failure in the original GFS design. Two mechanisms reduce this risk.

First, the operation log is replicated to multiple log servers. A mutation is only acknowledged to the client after the log entry is written to a quorum of log servers. This is functionally similar to how Raft or Paxos handle log replication, but GFS did it without a full consensus protocol: the master drives replication and requires a majority ack.

Second, shadow masters maintain a nearly-current read-only copy of the master state by replaying the operation log in near-real-time. Clients can read from shadow masters when the primary master is unavailable. Stale reads are possible: shadow masters may lag by seconds. For metadata operations that tolerate staleness (directory listings, size queries), this is acceptable.

class ShadowMaster {
  private logOffset: number = 0;
  private namespace: Map<string, FileMetadata> = new Map();

  async replayLog(logEntries: LogEntry[]): Promise<void> {
    for (const entry of logEntries) {
      if (entry.offset <= this.logOffset) continue;
      this.applyEntry(entry);
      this.logOffset = entry.offset;
    }
  }

  private applyEntry(entry: LogEntry): void {
    switch (entry.type) {
      case "CREATE_FILE":
        this.namespace.set(entry.path, {
          path: entry.path,
          chunkCount: 0,
          chunks: [],
          createdAt: entry.timestamp,
          modifiedAt: entry.timestamp,
        });
        break;
      case "ADD_CHUNK":
        const file = this.namespace.get(entry.path);
        if (file) {
          file.chunks.push(entry.chunk);
          file.chunkCount++;
          file.modifiedAt = entry.timestamp;
        }
        break;
      case "DELETE_FILE":
        this.namespace.delete(entry.path);
        break;
    }
  }

  getFileMetadata(path: string): FileMetadata | undefined {
    return this.namespace.get(path);
  }
}

interface LogEntry {
  offset: number;
  type: "CREATE_FILE" | "ADD_CHUNK" | "DELETE_FILE";
  path: string;
  timestamp: number;
  chunk?: ChunkHandle;
}

Modern systems (HDFS with QJM, or Ceph with RADOS) use actual quorum-based protocols for metadata replication. The Journal Manager in HDFS writes each edit log entry to a majority of journal nodes before the NameNode considers it committed. Failover to a standby NameNode can be automatic and sub-minute.

Consistency vs. Availability: The Real Tradeoffs

Property	GFS / original HDFS	HDFS HA with QJM	Ceph RADOS
Metadata consistency	Strong (single master)	Strong (quorum)	Strong (quorum per PG)
Write consistency	Defined per append, undefined for concurrent offset writes	One writer, defined	Strong with journaling
Failover time	Minutes (manual)	Sub-minute (automatic)	Sub-minute
Metadata scalability	Single server memory ceiling	Federated NameNodes	Distributed MDS
Read staleness	Possible from shadow	Minimal from standby	Per-placement group
Complexity	Lower	Medium	High

The choice lives in your workload. Append-only workloads (log aggregation, ML training data pipelines) fit GFS-style systems well. Random-write workloads need stronger primitives. Analytics jobs that can tolerate stale metadata reads get better throughput from shadow masters than from round-tripping to the primary for every stat call.

One pattern that trips people up: chunk server heartbeats carry chunk inventory reports, and the metadata server uses these to build the chunk location map at startup. The chunk location map is never durably stored on disk at the master. If the master restarts, it rebuilds the map from heartbeats. This means startup time is a function of cluster size, and it means the in-memory map can diverge from reality if a chunk server loses data without properly reporting it. Chunk servers periodically send full inventories to catch this drift.

Production Considerations

Chunk size selection. Larger chunks (64-128 MB) mean fewer metadata entries but worse space efficiency for small files (many files smaller than one chunk each occupy a full chunk). GFS accepted this for batch workloads but it is a poor fit for mixed workloads with many small files.

Hot chunk bottleneck. If many clients read the same chunk simultaneously, the small set of replicas becomes a hotspot. GFS handled this by creating additional on-demand replicas. Systems with consistent hashing distribute load across more replicas.

Staggered checkpoints. The metadata server checkpoints its state periodically, writing a snapshot so log replay stays bounded. Running a checkpoint competes with serving live mutations. GFS checkpointed on a separate thread with a copy-on-write fork; HDFS uses a secondary NameNode or standby to produce checkpoints off the primary path.

Lease revocation on network partition. If a primary chunk server is partitioned from the metadata server but can still reach clients, it may continue serving writes after its lease expires. Clients caching lease locations can talk to a now-invalid primary. This is why GFS chunk servers check lease validity on every mutation and why the metadata server invalidates chunk location caches on lease expiry. Defense in depth here: the 60-second lease window bounds the damage from stale lease state.

Re-replication priority. When a chunk server goes down, chunks with only one remaining replica need re-replication before chunks with two remaining replicas. Priority queues ordered by replica count keep the most at-risk data covered first. Background re-replication is rate-limited to avoid flooding cluster network capacity during normal operations.

The architecture of a distributed file system is a series of decisions about where to trade consistency for availability, where to trade scalability for simplicity, and where to push complexity into the application layer rather than absorbing it into the storage system. GFS made these tradeoffs explicitly and documented them. Understanding those decisions tells you more about distributed systems design than most textbooks.

Systems built on these foundations (Colossus, HDFS with HA, Ceph) have all moved toward stronger guarantees as the operational cost of reasoning about weakly-consistent behavior proved higher than the engineering cost of implementing quorum protocols correctly. That trajectory is worth keeping in mind when designing storage systems today: weak consistency buys you simplicity in the happy path and costs you complexity in every edge case.