Part of a multi-episode project
Learn Zig Series):We have a delta transfer engine that can figure out which bytes changed between two versions of a file, and a manifest system that compares directory trees. But all of it runs locally -- both "sides" are just memory buffers on the same machine. Time to change that. This episode we're putting the file sync tool on the network.
The plan: a proper binary wire protocol with a handshake phase, authentication, manifest exchange, delta streaming, and error recovery. We covered TCP sockets back in episode 21 and built a full TCP server for the key-value store in episodes 42-43, so the socket mechanics should feel familiar. What's new here is designing a protocol that's robust enough for real file synchronization -- handling disconnections mid-transfer, verifying data integrity over the wire, and throttling bandwidth so sync doesn't eat your entire connection while you're trying to watch YouTube ;-)
Every network protocol needs message framing. TCP gives you a raw byte stream with no concept of "messages" -- it's up to you to define where one message ends and the next begins. Our protocol uses a simple length-prefixed format:
// src/protocol.zig
const std = @import("std");
const Sha256 = std.crypto.hash.sha2.Sha256;
/// Protocol version - increment when wire format changes
pub const PROTOCOL_VERSION: u8 = 1;
/// Magic bytes to identify our protocol (avoids connecting to random services)
pub const MAGIC: [4]u8 = .{ 'Z', 'S', 'Y', 'N' }; // "ZSYN" - Zig Sync
/// Maximum message payload size (16 MB - generous but prevents OOM from bad data)
pub const MAX_PAYLOAD: u32 = 16 * 1024 * 1024;
/// Message types sent between client and server
pub const MsgType = enum(u8) {
// Handshake phase
hello = 0x01,
hello_ack = 0x02,
auth_challenge = 0x03,
auth_response = 0x04,
auth_ok = 0x05,
auth_fail = 0x06,
// Manifest phase
manifest_request = 0x10,
manifest_data = 0x11,
// Transfer phase
file_start = 0x20,
file_chunk = 0x21,
file_end = 0x22,
chunk_ack = 0x23,
delta_start = 0x24,
delta_data = 0x25,
delta_end = 0x26,
// Control
sync_complete = 0x30,
error_msg = 0x31,
resume_request = 0x32,
resume_ack = 0x33,
// Dry-run
dry_run_plan = 0x40,
dry_run_ack = 0x41,
};
/// Wire header: every message starts with this
pub const Header = packed struct {
magic: [4]u8,
msg_type: MsgType,
flags: u8,
payload_len: u32,
checksum: u16, // CRC-16 of payload for integrity
pub const SIZE: usize = 12;
pub fn serialize(self: Header) [SIZE]u8 {
var buf: [SIZE]u8 = undefined;
@memcpy(buf[0..4], &self.magic);
buf[4] = @intFromEnum(self.msg_type);
buf[5] = self.flags;
std.mem.writeInt(u32, buf[6..10], self.payload_len, .big);
std.mem.writeInt(u16, buf[10..12], self.checksum, .big);
return buf;
}
pub fn deserialize(buf: [SIZE]u8) !Header {
if (!std.mem.eql(u8, buf[0..4], &MAGIC)) {
return error.InvalidMagic;
}
return .{
.magic = MAGIC,
.msg_type = std.meta.intToEnum(MsgType, buf[4]) catch return error.InvalidMessageType,
.flags = buf[5],
.payload_len = std.mem.readInt(u32, buf[6..10], .big),
.checksum = std.mem.readInt(u16, buf[10..12], .big),
};
}
};
The header is 12 bytes: 4 bytes magic, 1 byte message type, 1 byte flags, 4 bytes payload length, 2 bytes CRC-16 checksum. Using big-endian (network byte order) ensures both sides agree on integer layout regardless of platform -- this is standard practice for network protocols (we talked about endianness briefly in episode 17).
The magic bytes ZSYN serve as a sanity check. If someone accidentally connects a web browser to our sync port, the first 4 bytes of an HTTP request won't match ZSYN and we immediately reject the connection. This is vastly better than trying to parse random garbage and crashing deep in the protocol handler.
The flags byte is reserved for future use -- things like compression flags, encryption markers, or priority indicators. For now it's always zero.
With the header format defined, we need functions to send and receive complete messages over a TCP stream:
// src/protocol.zig (continued)
/// CRC-16-CCITT for integrity checking
fn crc16(data: []const u8) u16 {
var crc: u16 = 0xFFFF;
for (data) |byte| {
crc ^= @as(u16, byte) << 8;
for (0..8) |_| {
if (crc & 0x8000 != 0) {
crc = (crc << 1) ^ 0x1021;
} else {
crc = crc << 1;
}
}
}
return crc;
}
/// Send a complete message: header + payload
pub fn sendMessage(
stream: std.net.Stream,
msg_type: MsgType,
payload: []const u8,
) !void {
if (payload.len > MAX_PAYLOAD) return error.PayloadTooLarge;
const header = Header{
.magic = MAGIC,
.msg_type = msg_type,
.flags = 0,
.payload_len = @intCast(payload.len),
.checksum = crc16(payload),
};
const header_bytes = header.serialize();
_ = try stream.write(&header_bytes);
if (payload.len > 0) {
_ = try stream.write(payload);
}
}
/// Receive a complete message. Returns the header and allocated payload.
/// Caller owns the payload memory.
pub fn recvMessage(
allocator: std.mem.Allocator,
stream: std.net.Stream,
) !struct { header: Header, payload: []u8 } {
// read the header
var header_buf: [Header.SIZE]u8 = undefined;
const header_read = try stream.readAll(&header_buf);
if (header_read < Header.SIZE) return error.ConnectionClosed;
const header = try Header.deserialize(header_buf);
if (header.payload_len > MAX_PAYLOAD) return error.PayloadTooLarge;
// read payload
var payload = try allocator.alloc(u8, header.payload_len);
errdefer allocator.free(payload);
if (header.payload_len > 0) {
const payload_read = try stream.readAll(payload);
if (payload_read < header.payload_len) {
allocator.free(payload);
return error.ConnectionClosed;
}
}
// verify checksum
if (crc16(payload) != header.checksum) {
allocator.free(payload);
return error.ChecksumMismatch;
}
return .{ .header = header, .payload = payload };
}
The readAll approach is important. TCP can deliver data in chunks of arbitrary size -- you might call read expecting 1024 bytes and get back 37. readAll loops until it either fills the buffer or hits an error/EOF. Without this, you'd get corrupted messages whenever the network decides to fragment your data across multiple packets.
The CRC-16 catches bit flips and truncation. TCP's own checksums should catch most corruption, but an extra application-level check costs almost nothing and gives you confidence that what you received is exactly what was sent. Paranoia in network code is a virtue.
When a client connects, both sides need to agree on protocol version and authenticate. Here's the handshake sequence:
hello with its protocol versionhello_ack if compatibleauth_challenge with a random nonceauth_responseauth_ok or auth_fail// src/auth.zig
const std = @import("std");
const protocol = @import("protocol.zig");
const HmacSha256 = std.crypto.auth.hmac.sha2.HmacSha256;
pub const NONCE_SIZE: usize = 32;
pub const HelloPayload = struct {
version: u8,
capabilities: u32, // bitmask of supported features
pub fn serialize(self: HelloPayload) [5]u8 {
var buf: [5]u8 = undefined;
buf[0] = self.version;
std.mem.writeInt(u32, buf[1..5], self.capabilities, .big);
return buf;
}
pub fn deserialize(data: []const u8) !HelloPayload {
if (data.len < 5) return error.InvalidPayload;
return .{
.version = data[0],
.capabilities = std.mem.readInt(u32, data[1..5], .big),
};
}
};
// Capability flags
pub const CAP_DELTA_TRANSFER: u32 = 1 << 0;
pub const CAP_COMPRESSION: u32 = 1 << 1;
pub const CAP_RESUME: u32 = 1 << 2;
pub const CAP_BIDIRECTIONAL: u32 = 1 << 3;
/// Server-side: generate a challenge and verify the client's response
pub fn serverAuthenticate(
stream: std.net.Stream,
shared_secret: []const u8,
allocator: std.mem.Allocator,
) !bool {
// generate random nonce
var nonce: [NONCE_SIZE]u8 = undefined;
std.crypto.random.bytes(&nonce);
// send challenge
try protocol.sendMessage(stream, .auth_challenge, &nonce);
// receive response
const msg = try protocol.recvMessage(allocator, stream);
defer allocator.free(msg.payload);
if (msg.header.msg_type != .auth_response) return false;
if (msg.payload.len != HmacSha256.mac_length) return false;
// compute expected HMAC
var expected: [HmacSha256.mac_length]u8 = undefined;
HmacSha256.create(&expected, &nonce, shared_secret);
// constant-time comparison to prevent timing attacks
return std.crypto.utils.timingSafeEql(
[HmacSha256.mac_length]u8,
msg.payload[0..HmacSha256.mac_length].*,
expected,
);
}
/// Client-side: respond to auth challenge
pub fn clientAuthenticate(
stream: std.net.Stream,
shared_secret: []const u8,
allocator: std.mem.Allocator,
) !bool {
// receive challenge
const challenge_msg = try protocol.recvMessage(allocator, stream);
defer allocator.free(challenge_msg.payload);
if (challenge_msg.header.msg_type != .auth_challenge) return false;
if (challenge_msg.payload.len != NONCE_SIZE) return false;
// compute HMAC
var mac: [HmacSha256.mac_length]u8 = undefined;
HmacSha256.create(&mac, challenge_msg.payload, shared_secret);
// send response
try protocol.sendMessage(stream, .auth_response, &mac);
// wait for ok/fail
const result_msg = try protocol.recvMessage(allocator, stream);
defer allocator.free(result_msg.payload);
return result_msg.header.msg_type == .auth_ok;
}
Why HMAC challenge-response instead of just sending the password? Because if you send the password directly, anyone snooping the connection captures it. With HMAC, the secret never crosses the wire -- only a cryptographic proof that you know it. The random nonce prevents replay attacks (you can't record a valid response and replay it later because the next challenge will have a different nonce). We covered hashing in previous episodes; HMAC is essentially "keyed hashing" -- it binds a secret key to the hash computation so only someone with the key can produce a valid output.
The timingSafeEql comparison is critical. A naive byte-by-byte comparison short-circuits on the first mismatch -- an attacker watching response times can figure out how many leading bytes they got right and brute-force the rest one byte at a time. Constant-time comparison always takes the same time regardless of where (or whether) there's a mismatch. This is crypto 101 but I've seen production code get it wrong more times than I can count.
With protocol and auth in place, let's build the server. It listens for connections, performs the handshake, and enters the sync loop:
// src/server.zig
const std = @import("std");
const protocol = @import("protocol.zig");
const auth_mod = @import("auth.zig");
const manifest = @import("manifest.zig");
const delta_mod = @import("delta.zig");
pub const SyncServer = struct {
allocator: std.mem.Allocator,
listener: std.net.Server,
sync_root: []const u8,
shared_secret: []const u8,
pub fn init(
allocator: std.mem.Allocator,
address: std.net.Address,
sync_root: []const u8,
shared_secret: []const u8,
) !SyncServer {
const listener = try address.listen(.{
.reuse_address = true,
});
return .{
.allocator = allocator,
.listener = listener,
.sync_root = sync_root,
.shared_secret = shared_secret,
};
}
pub fn deinit(self: *SyncServer) void {
self.listener.deinit();
}
pub fn acceptClient(self: *SyncServer) !void {
const conn = try self.listener.accept();
defer conn.stream.close();
self.handleConnection(conn.stream) catch |err| {
// send error message to client before disconnecting
const err_msg = @errorName(err);
protocol.sendMessage(conn.stream, .error_msg, err_msg) catch {};
};
}
fn handleConnection(self: *SyncServer, stream: std.net.Stream) !void {
// Step 1: receive hello
const hello_msg = try protocol.recvMessage(self.allocator, stream);
defer self.allocator.free(hello_msg.payload);
if (hello_msg.header.msg_type != .hello) return error.UnexpectedMessage;
const hello = try auth_mod.HelloPayload.deserialize(hello_msg.payload);
if (hello.version != protocol.PROTOCOL_VERSION) return error.VersionMismatch;
// Step 2: send hello_ack
const our_hello = auth_mod.HelloPayload{
.version = protocol.PROTOCOL_VERSION,
.capabilities = auth_mod.CAP_DELTA_TRANSFER |
auth_mod.CAP_COMPRESSION |
auth_mod.CAP_RESUME,
};
try protocol.sendMessage(stream, .hello_ack, &our_hello.serialize());
// Step 3: authenticate
const auth_ok = try auth_mod.serverAuthenticate(
stream,
self.shared_secret,
self.allocator,
);
if (!auth_ok) {
try protocol.sendMessage(stream, .auth_fail, "");
return error.AuthenticationFailed;
}
try protocol.sendMessage(stream, .auth_ok, "");
// Step 4: enter sync loop
try self.syncLoop(stream);
}
fn syncLoop(self: *SyncServer, stream: std.net.Stream) !void {
while (true) {
const msg = try protocol.recvMessage(self.allocator, stream);
defer self.allocator.free(msg.payload);
switch (msg.header.msg_type) {
.manifest_request => {
try self.handleManifestRequest(stream);
},
.file_start => {
try self.handleFileReceive(stream, msg.payload);
},
.delta_start => {
try self.handleDeltaReceive(stream, msg.payload);
},
.sync_complete => return,
.dry_run_plan => {
// acknowledge dry-run plan without applying
try protocol.sendMessage(stream, .dry_run_ack, "");
},
else => return error.UnexpectedMessage,
}
}
}
fn handleManifestRequest(self: *SyncServer, stream: std.net.Stream) !void {
// build manifest of our sync root and send it
var man = try manifest.buildManifest(self.allocator, self.sync_root);
defer man.deinit(self.allocator);
const serialized = try manifest.serialize(self.allocator, man);
defer self.allocator.free(serialized);
try protocol.sendMessage(stream, .manifest_data, serialized);
}
fn handleFileReceive(self: *SyncServer, stream: std.net.Stream, header_data: []const u8) !void {
if (header_data.len < 10) return error.InvalidPayload;
const path_len = std.mem.readInt(u16, header_data[0..2], .big);
if (header_data.len < 2 + path_len + 8) return error.InvalidPayload;
const rel_path = header_data[2 .. 2 + path_len];
const file_size = std.mem.readInt(u64, header_data[2 + path_len ..][0..8], .big);
_ = file_size;
var full_path_buf: [4096]u8 = undefined;
const full_path = std.fmt.bufPrint(&full_path_buf, "{s}/{s}", .{
self.sync_root, rel_path,
}) catch return error.PathTooLong;
if (std.fs.path.dirname(full_path)) |dir| {
std.fs.makeDirAbsolute(dir) catch |err| switch (err) {
error.PathAlreadyExists => {},
else => return err,
};
}
const file = try std.fs.createFileAbsolute(full_path, .{});
defer file.close();
var received: u64 = 0;
while (true) {
const chunk_msg = try protocol.recvMessage(self.allocator, stream);
defer self.allocator.free(chunk_msg.payload);
if (chunk_msg.header.msg_type == .file_end) break;
if (chunk_msg.header.msg_type != .file_chunk) return error.UnexpectedMessage;
try file.writeAll(chunk_msg.payload);
received += chunk_msg.payload.len;
var ack_buf: [8]u8 = undefined;
std.mem.writeInt(u64, &ack_buf, received, .big);
try protocol.sendMessage(stream, .chunk_ack, &ack_buf);
}
}
fn handleDeltaReceive(self: *SyncServer, stream: std.net.Stream, header_data: []const u8) !void {
if (header_data.len < 2) return error.InvalidPayload;
const path_len = std.mem.readInt(u16, header_data[0..2], .big);
if (header_data.len < 2 + path_len) return error.InvalidPayload;
const rel_path = header_data[2 .. 2 + path_len];
var full_path_buf: [4096]u8 = undefined;
const full_path = std.fmt.bufPrint(&full_path_buf, "{s}/{s}", .{
self.sync_root, rel_path,
}) catch return error.PathTooLong;
const existing_data = std.fs.cwd().readFileAlloc(
self.allocator, full_path, 16 * 1024 * 1024,
) catch "";
defer if (existing_data.len > 0) self.allocator.free(existing_data);
var delta_buf = std.ArrayList(u8).init(self.allocator);
defer delta_buf.deinit();
while (true) {
const chunk_msg = try protocol.recvMessage(self.allocator, stream);
defer self.allocator.free(chunk_msg.payload);
if (chunk_msg.header.msg_type == .delta_end) break;
if (chunk_msg.header.msg_type != .delta_data) return error.UnexpectedMessage;
try delta_buf.appendSlice(chunk_msg.payload);
}
const result = try delta_mod.applySerializedDelta(
self.allocator, existing_data, delta_buf.items,
);
defer self.allocator.free(result);
const file = try std.fs.createFileAbsolute(full_path, .{});
defer file.close();
try file.writeAll(result);
}
};
The server structure should look familiar if you followed the KV store episodes (42-43). Accept a connection, perform handshake, then enter a loop processing messages. The tagged union nature of our MsgType enum means we can handle each message type in a clean switch block -- no string parsing, no ambiguity about what the client is asking for.
One design choice worth pointing out: the server handles one client at a time. For a file sync tool this is usually fine -- you're syncing between two specific machines, not running a web server with thousands of concurrent users. A production implementation could spawn a thread per connection (episode 30) or use an event loop, but for now simplicity wins.
The client mirrors the server -- performs the handshake from the other side, then drives the sync process:
// src/client.zig
const std = @import("std");
const protocol = @import("protocol.zig");
const auth_mod = @import("auth.zig");
const manifest = @import("manifest.zig");
const delta_mod = @import("delta.zig");
const diff_mod = @import("diff.zig");
pub const SyncClient = struct {
allocator: std.mem.Allocator,
stream: std.net.Stream,
sync_root: []const u8,
shared_secret: []const u8,
server_caps: u32,
dry_run: bool,
/// Chunk size for file transfers (64 KB)
const CHUNK_SIZE: usize = 64 * 1024;
pub fn connect(
allocator: std.mem.Allocator,
address: std.net.Address,
sync_root: []const u8,
shared_secret: []const u8,
dry_run: bool,
) !SyncClient {
const stream = try std.net.tcpConnectToAddress(address);
errdefer stream.close();
var client = SyncClient{
.allocator = allocator,
.stream = stream,
.sync_root = sync_root,
.shared_secret = shared_secret,
.server_caps = 0,
.dry_run = dry_run,
};
try client.performHandshake();
return client;
}
pub fn disconnect(self: *SyncClient) void {
protocol.sendMessage(self.stream, .sync_complete, "") catch {};
self.stream.close();
}
fn performHandshake(self: *SyncClient) !void {
// send hello
const hello = auth_mod.HelloPayload{
.version = protocol.PROTOCOL_VERSION,
.capabilities = auth_mod.CAP_DELTA_TRANSFER |
auth_mod.CAP_RESUME,
};
try protocol.sendMessage(self.stream, .hello, &hello.serialize());
// receive hello_ack
const ack = try protocol.recvMessage(self.allocator, self.stream);
defer self.allocator.free(ack.payload);
if (ack.header.msg_type != .hello_ack) return error.HandshakeFailed;
const server_hello = try auth_mod.HelloPayload.deserialize(ack.payload);
self.server_caps = server_hello.capabilities;
// authenticate
const auth_ok = try auth_mod.clientAuthenticate(
self.stream,
self.shared_secret,
self.allocator,
);
if (!auth_ok) return error.AuthenticationFailed;
}
/// Request the server's manifest and compute what needs syncing
pub fn getServerManifest(self: *SyncClient) !manifest.Manifest {
try protocol.sendMessage(self.stream, .manifest_request, "");
const msg = try protocol.recvMessage(self.allocator, self.stream);
defer self.allocator.free(msg.payload);
if (msg.header.msg_type != .manifest_data) return error.UnexpectedMessage;
return manifest.deserialize(self.allocator, msg.payload);
}
/// Send a file to the server in chunks
pub fn sendFile(self: *SyncClient, rel_path: []const u8, data: []const u8) !void {
// build file_start header: path_len + path + file_size
var header_buf = std.ArrayList(u8).init(self.allocator);
defer header_buf.deinit();
var path_len_buf: [2]u8 = undefined;
std.mem.writeInt(u16, &path_len_buf, @intCast(rel_path.len), .big);
try header_buf.appendSlice(&path_len_buf);
try header_buf.appendSlice(rel_path);
var size_buf: [8]u8 = undefined;
std.mem.writeInt(u64, &size_buf, @intCast(data.len), .big);
try header_buf.appendSlice(&size_buf);
try protocol.sendMessage(self.stream, .file_start, header_buf.items);
// send data in chunks
var offset: usize = 0;
while (offset < data.len) {
const end = @min(offset + CHUNK_SIZE, data.len);
const chunk = data[offset..end];
try protocol.sendMessage(self.stream, .file_chunk, chunk);
// wait for ack
const ack_msg = try protocol.recvMessage(self.allocator, self.stream);
defer self.allocator.free(ack_msg.payload);
if (ack_msg.header.msg_type != .chunk_ack) return error.TransferFailed;
offset = end;
}
// send file_end marker
try protocol.sendMessage(self.stream, .file_end, "");
}
/// Send a delta to the server for an existing file
pub fn sendDelta(self: *SyncClient, rel_path: []const u8, delta_data: []const u8) !void {
// delta_start header: path_len + path
var header_buf = std.ArrayList(u8).init(self.allocator);
defer header_buf.deinit();
var path_len_buf: [2]u8 = undefined;
std.mem.writeInt(u16, &path_len_buf, @intCast(rel_path.len), .big);
try header_buf.appendSlice(&path_len_buf);
try header_buf.appendSlice(rel_path);
try protocol.sendMessage(self.stream, .delta_start, header_buf.items);
// send delta in chunks
var offset: usize = 0;
while (offset < delta_data.len) {
const end = @min(offset + CHUNK_SIZE, delta_data.len);
try protocol.sendMessage(self.stream, .delta_data, delta_data[offset..end]);
offset = end;
}
try protocol.sendMessage(self.stream, .delta_end, "");
}
};
The chunk-and-ack pattern for file transfers is deliberate. After each 64KB chunk the client waits for an acknowledgment from the server. This gives us two things: flow control (the client won't flood the server faster than it can write to disk) and resume points (if the connection drops, the client knows exactly which chunks were received based on the last ack it got).
Network connections die. Wi-Fi drops, laptops get closed mid-sync, VPN tunnels time out. A good sync tool handles this gracefully. The key insight for resumable transfers: the server tracks which bytes it has received per file, and the client can ask "where did we leave off?"
// src/resume.zig
const std = @import("std");
const protocol = @import("protocol.zig");
/// Tracks transfer progress for resume capability
pub const TransferState = struct {
/// Maps relative file path -> bytes confirmed received
progress: std.StringHashMap(u64),
allocator: std.mem.Allocator,
pub fn init(allocator: std.mem.Allocator) TransferState {
return .{
.progress = std.StringHashMap(u64).init(allocator),
.allocator = allocator,
};
}
pub fn deinit(self: *TransferState) void {
var it = self.progress.iterator();
while (it.next()) |entry| {
self.allocator.free(entry.key_ptr.*);
}
self.progress.deinit();
}
pub fn updateProgress(self: *TransferState, path: []const u8, bytes: u64) !void {
const gop = try self.progress.getOrPut(try self.allocator.dupe(u8, path));
if (gop.found_existing) {
self.allocator.free(gop.key_ptr.*);
gop.key_ptr.* = try self.allocator.dupe(u8, path);
}
gop.value_ptr.* = bytes;
}
pub fn getProgress(self: *const TransferState, path: []const u8) u64 {
return self.progress.get(path) orelse 0;
}
/// Serialize state to disk so it survives process restarts
pub fn saveToDisk(self: *const TransferState, file_path: []const u8) !void {
const file = try std.fs.createFileAbsolute(file_path, .{});
defer file.close();
var writer = file.writer();
var it = self.progress.iterator();
while (it.next()) |entry| {
try writer.print("{s}\t{d}\n", .{ entry.key_ptr.*, entry.value_ptr.* });
}
}
/// Load state from disk
pub fn loadFromDisk(
allocator: std.mem.Allocator,
file_path: []const u8,
) !TransferState {
var state = TransferState.init(allocator);
errdefer state.deinit();
const data = std.fs.cwd().readFileAlloc(allocator, file_path, 1024 * 1024) catch |err| switch (err) {
error.FileNotFound => return state,
else => return err,
};
defer allocator.free(data);
var lines = std.mem.splitScalar(u8, data, '\n');
while (lines.next()) |line| {
if (line.len == 0) continue;
var parts = std.mem.splitScalar(u8, line, '\t');
const path = parts.next() orelse continue;
const bytes_str = parts.next() orelse continue;
const bytes = std.fmt.parseInt(u64, bytes_str, 10) catch continue;
try state.updateProgress(path, bytes);
}
return state;
}
};
/// Client-side: request resume information from server
pub fn requestResume(
stream: std.net.Stream,
rel_path: []const u8,
allocator: std.mem.Allocator,
) !u64 {
try protocol.sendMessage(stream, .resume_request, rel_path);
const msg = try protocol.recvMessage(allocator, stream);
defer allocator.free(msg.payload);
if (msg.header.msg_type != .resume_ack) return 0;
if (msg.payload.len < 8) return 0;
return std.mem.readInt(u64, msg.payload[0..8], .big);
}
The resume state file is a simple tab-separated format: one line per in-progress transfer with the relative path and bytes confirmed. When the client reconnects after a failure, it loads this file, sees that src/main.zig had 131072 bytes confirmed, and tells the server to skip the first 131072 bytes. The server can verify this against its own records -- if they agree, transfer resumes from that point. If they disagree (maybe the partial file got corrupted), we start over.
This is the same approach rsync uses with --partial -- keep incomplete files on disk and resume where you left off. Simple, robust, and eliminates the frustration of watching a large transfer restart from zero after a brief network hiccup.
When file sync runs in the background, you don't want it consuming your entire upload bandwidth. A token bucket rate limiter is the classic solution -- you have a "bucket" that fills with tokens at a fixed rate, and each byte you send costs one token. If the bucket is empty, you wait:
// src/throttle.zig
const std = @import("std");
/// Token bucket rate limiter for bandwidth control.
/// Allows bursts up to bucket capacity, but average rate
/// is limited to bytes_per_second.
pub const RateLimiter = struct {
tokens: f64,
capacity: f64,
refill_rate: f64, // tokens per nanosecond
last_refill: i128,
/// Create a rate limiter with the given bandwidth limit.
/// Set bytes_per_second to 0 for unlimited.
pub fn init(bytes_per_second: u64) RateLimiter {
if (bytes_per_second == 0) {
return .{
.tokens = std.math.floatMax(f64),
.capacity = std.math.floatMax(f64),
.refill_rate = std.math.floatMax(f64),
.last_refill = std.time.nanoTimestamp(),
};
}
const cap = @as(f64, @floatFromInt(bytes_per_second)); // 1 second burst
return .{
.tokens = cap, // start with a full bucket
.capacity = cap,
.refill_rate = @as(f64, @floatFromInt(bytes_per_second)) / 1_000_000_000.0,
.last_refill = std.time.nanoTimestamp(),
};
}
/// Consume tokens for the given number of bytes.
/// Blocks (busy-waits) until enough tokens are available.
pub fn consume(self: *RateLimiter, bytes: usize) void {
if (self.refill_rate == std.math.floatMax(f64)) return; // unlimited
const needed = @as(f64, @floatFromInt(bytes));
while (true) {
self.refill();
if (self.tokens >= needed) {
self.tokens -= needed;
return;
}
// wait a bit for tokens to accumulate
std.time.sleep(1_000_000); // 1ms
}
}
fn refill(self: *RateLimiter) void {
const now = std.time.nanoTimestamp();
const elapsed = now - self.last_refill;
if (elapsed <= 0) return;
const new_tokens = @as(f64, @floatFromInt(elapsed)) * self.refill_rate;
self.tokens = @min(self.tokens + new_tokens, self.capacity);
self.last_refill = now;
}
};
/// Wrap a stream with rate limiting
pub fn throttledWrite(
stream: std.net.Stream,
data: []const u8,
limiter: *RateLimiter,
) !void {
const chunk_size: usize = 8192; // limit in 8KB increments
var offset: usize = 0;
while (offset < data.len) {
const end = @min(offset + chunk_size, data.len);
const chunk = data[offset..end];
limiter.consume(chunk.len);
_ = try stream.write(chunk);
offset = end;
}
}
The token bucket is elegant in its simplicity. Tokens accumulate at a fixed rate (e.g. 1MB/s means 1,000,000 tokens per second). Sending data spends tokens. If you try to send faster than the rate allows, the bucket empties and you have to wait for it to refill. The bucket capacity allows short bursts -- if the sync has been idle for a second, it can immediately send a full second's worth of data before throttling kicks in.
I chose bytes_per_second as the configuration unit because it maps directly to what users think about: "I want to limit sync to 500 KB/s so my video calls don't stutter." The internal math converts to nanosecond precision because that's what std.time.nanoTimestamp gives us, but the external API stays human-friendly.
The busy-wait loop with 1ms sleeps is crude but effective. A production implementation might use std.Thread.Futex or select()/poll() with a timeout for better power efficiency, but for a sync tool that's already doing disk I/O the 1ms granularity is more than fine.
Up to now we've been thinking about sync as one-directional: push changes from source to destination. But what if both sides change the same file? This is the conflict problem, and it's the hardest part of sync.
Our approach: compare timestamps and file content hashes from both sides. If both modified the same file since the last sync, that's a conflict. The user must decide which version wins:
// src/conflict.zig
const std = @import("std");
const manifest = @import("manifest.zig");
pub const ConflictResolution = enum {
/// Keep the local version, overwrite remote
keep_local,
/// Keep the remote version, overwrite local
keep_remote,
/// Keep both by renaming one side
keep_both,
/// Skip this file entirely (don't sync)
skip,
};
pub const Conflict = struct {
path: []const u8,
local_mtime: i128,
remote_mtime: i128,
local_checksum: [32]u8,
remote_checksum: [32]u8,
resolution: ConflictResolution,
};
/// Detect conflicts between two manifests given a common ancestor.
/// A conflict occurs when the same file was modified on BOTH sides
/// since the last successful sync.
pub fn detectConflicts(
allocator: std.mem.Allocator,
local: []const manifest.FileEntry,
remote: []const manifest.FileEntry,
last_sync_time: i128,
) ![]Conflict {
var conflicts = std.ArrayList(Conflict).init(allocator);
// index remote entries by path for O(1) lookup
var remote_map = std.StringHashMap(*const manifest.FileEntry).init(allocator);
defer remote_map.deinit();
for (remote) |*entry| {
try remote_map.put(entry.path, entry);
}
// check each local entry against remote
for (local) |*local_entry| {
if (remote_map.get(local_entry.path)) |remote_entry| {
// file exists on both sides
const local_modified = local_entry.mtime_ns > last_sync_time;
const remote_modified = remote_entry.mtime_ns > last_sync_time;
if (local_modified and remote_modified) {
// both modified since last sync -- but are they actually different?
if (!std.mem.eql(u8, &local_entry.checksum, &remote_entry.checksum)) {
// genuinely different content = conflict
try conflicts.append(.{
.path = local_entry.path,
.local_mtime = local_entry.mtime_ns,
.remote_mtime = remote_entry.mtime_ns,
.local_checksum = local_entry.checksum,
.remote_checksum = remote_entry.checksum,
.resolution = .skip, // default: don't do anything until user decides
});
}
// if checksums match, both sides made the same edit -- no conflict
}
}
}
return conflicts.toOwnedSlice();
}
/// Apply conflict resolution by renaming the losing side
pub fn resolveKeepBoth(
allocator: std.mem.Allocator,
path: []const u8,
suffix: []const u8,
) ![]u8 {
// "main.zig" -> "main.conflict-remote.zig"
const ext_pos = std.mem.lastIndexOfScalar(u8, path, '.') orelse path.len;
const base = path[0..ext_pos];
const ext = path[ext_pos..];
return std.fmt.allocPrint(allocator, "{s}.conflict-{s}{s}", .{ base, suffix, ext });
}
The conflict detection logic is straightforward: if a file was modified on both sides since the last sync AND the checksums differ, it's a conflict. The last_sync_time comes from a metadata file we write after each successful sync -- it records when the two sides were last known to be in agreement.
The "keep both" resolution renames one version with a .conflict-remote or .conflict-local suffix. This is what Dropbox does (those "conflicted copy" files), and while it's not elegant it's safe -- no data is ever lost.
For automated sync (no human present to resolve conflicts), you typically pick a default strategy: "newest wins" or "largest file wins" or "always prefer the server." Our implementation defaults to skip -- don't touch conflicted files until someone explicitly resolves them. Safer that way.
Before actually syncing anything, users want to see what WOULD happen. The dry-run mode computes the full sync plan -- diffs, deltas, conflict detection, everything -- but doesn't transfer any bytes or modify any files:
// src/dry_run.zig
const std = @import("std");
const diff_mod = @import("diff.zig");
const sync_mod = @import("sync.zig");
const conflict_mod = @import("conflict.zig");
pub const DryRunAction = struct {
path: []const u8,
action: ActionType,
bytes: u64,
reason: []const u8,
};
pub const ActionType = enum {
transfer_full,
transfer_delta,
delete_remote,
delete_local,
mkdir,
rename,
conflict,
skip,
};
pub const DryRunReport = struct {
actions: []DryRunAction,
total_transfer_bytes: u64,
files_to_transfer: u32,
files_to_delete: u32,
conflicts: u32,
allocator: std.mem.Allocator,
pub fn deinit(self: *DryRunReport) void {
self.allocator.free(self.actions);
}
/// Print a human-readable report to the given writer
pub fn print(self: *const DryRunReport, writer: anytype) !void {
try writer.print("=== Sync Dry Run ===\n\n", .{});
for (self.actions) |action| {
const icon = switch (action.action) {
.transfer_full => "+",
.transfer_delta => "~",
.delete_remote, .delete_local => "-",
.mkdir => "d",
.rename => ">",
.conflict => "!",
.skip => ".",
};
if (action.bytes > 0) {
try writer.print(" [{s}] {s} ({d} bytes) -- {s}\n", .{
icon,
action.path,
action.bytes,
action.reason,
});
} else {
try writer.print(" [{s}] {s} -- {s}\n", .{
icon,
action.path,
action.reason,
});
}
}
try writer.print("\nSummary:\n", .{});
try writer.print(" Files to transfer: {d}\n", .{self.files_to_transfer});
try writer.print(" Files to delete: {d}\n", .{self.files_to_delete});
try writer.print(" Conflicts: {d}\n", .{self.conflicts});
try writer.print(" Total transfer: {d} bytes\n", .{self.total_transfer_bytes});
try writer.print("\nNo changes made (dry run).\n", .{});
}
};
pub fn planDryRun(
allocator: std.mem.Allocator,
diffs: []const diff_mod.DiffEntry,
conflicts: []const conflict_mod.Conflict,
) !DryRunReport {
var actions = std.ArrayList(DryRunAction).init(allocator);
var total_bytes: u64 = 0;
var transfers: u32 = 0;
var deletes: u32 = 0;
for (diffs) |d| {
switch (d.change) {
.added => {
if (d.source_entry) |src| {
const size = src.size;
try actions.append(.{
.path = d.path,
.action = .transfer_full,
.bytes = size,
.reason = "new file",
});
total_bytes += size;
transfers += 1;
}
},
.deleted => {
try actions.append(.{
.path = d.path,
.action = .delete_remote,
.bytes = 0,
.reason = "removed locally",
});
deletes += 1;
},
.modified => {
if (d.source_entry) |src| {
const size = src.size;
if (sync_mod.shouldUseDelta(size)) {
try actions.append(.{
.path = d.path,
.action = .transfer_delta,
.bytes = size / 10, // estimate: delta is ~10% of full
.reason = "modified, delta transfer",
});
total_bytes += size / 10;
} else {
try actions.append(.{
.path = d.path,
.action = .transfer_full,
.bytes = size,
.reason = "modified, full transfer (below delta threshold)",
});
total_bytes += size;
}
transfers += 1;
}
},
.unchanged => {},
}
}
// add conflicts
for (conflicts) |c| {
try actions.append(.{
.path = c.path,
.action = .conflict,
.bytes = 0,
.reason = "modified on both sides",
});
}
return .{
.actions = try actions.toOwnedSlice(),
.total_transfer_bytes = total_bytes,
.files_to_transfer = transfers,
.files_to_delete = deletes,
.conflicts = @intCast(conflicts.len),
.allocator = allocator,
};
}
Dry-run is useful not just for cautious users but also for scripting. You can run zsync --dry-run --machine-readable to get a list of changes, pipe it through a filter ("don't sync anything in the build/ directory"), and then run the actual sync. This seperation of "plan" and "execute" is a pattern we've used before (the CLI task runner in episode 36 had a similar --plan mode).
Testing network code properly means running an actual server, connecting to it, and verifying the whole protocol works. We spin up a local TCP server on a random port and exercise the full flow:
// src/test_e2e.zig
const std = @import("std");
const protocol = @import("protocol.zig");
const auth_mod = @import("auth.zig");
const SyncServer = @import("server.zig").SyncServer;
const SyncClient = @import("client.zig").SyncClient;
const test_secret = "test-shared-secret-for-e2e";
fn setupTestDir(allocator: std.mem.Allocator, prefix: []const u8) ![]u8 {
const tmp_dir = try std.fmt.allocPrint(allocator, "/tmp/zsync-test-{s}-{d}", .{
prefix, std.time.milliTimestamp(),
});
try std.fs.makeDirAbsolute(tmp_dir);
return tmp_dir;
}
fn runServer(server: *SyncServer) void {
server.acceptClient() catch {};
}
test "e2e: full handshake and authentication" {
const allocator = std.testing.allocator;
const addr = try std.net.Address.resolveIp("127.0.0.1", 0);
const server_dir = try setupTestDir(allocator, "server");
defer { std.fs.deleteTreeAbsolute(server_dir) catch {}; allocator.free(server_dir); }
var server = try SyncServer.init(allocator, addr, server_dir, test_secret);
defer server.deinit();
const bound_addr = server.listener.listen_address;
const server_thread = try std.Thread.spawn(.{}, runServer, .{&server});
const client_dir = try setupTestDir(allocator, "client");
defer { std.fs.deleteTreeAbsolute(client_dir) catch {}; allocator.free(client_dir); }
var client = try SyncClient.connect(allocator, bound_addr, client_dir, test_secret, false);
client.disconnect();
server_thread.join();
}
test "e2e: authentication rejects wrong secret" {
const allocator = std.testing.allocator;
const addr = try std.net.Address.resolveIp("127.0.0.1", 0);
const server_dir = try setupTestDir(allocator, "server-auth");
defer { std.fs.deleteTreeAbsolute(server_dir) catch {}; allocator.free(server_dir); }
var server = try SyncServer.init(allocator, addr, server_dir, test_secret);
defer server.deinit();
const bound_addr = server.listener.listen_address;
const server_thread = try std.Thread.spawn(.{}, runServer, .{&server});
const client_dir = try setupTestDir(allocator, "client-auth");
defer { std.fs.deleteTreeAbsolute(client_dir) catch {}; allocator.free(client_dir); }
const result = SyncClient.connect(allocator, bound_addr, client_dir, "wrong-secret", false);
try std.testing.expectError(error.AuthenticationFailed, result);
server_thread.join();
}
test "e2e: file transfer round trip" {
const allocator = std.testing.allocator;
const addr = try std.net.Address.resolveIp("127.0.0.1", 0);
const server_dir = try setupTestDir(allocator, "server-xfer");
defer { std.fs.deleteTreeAbsolute(server_dir) catch {}; allocator.free(server_dir); }
var server = try SyncServer.init(allocator, addr, server_dir, test_secret);
defer server.deinit();
const bound_addr = server.listener.listen_address;
const server_thread = try std.Thread.spawn(.{}, runServer, .{&server});
const client_dir = try setupTestDir(allocator, "client-xfer");
defer { std.fs.deleteTreeAbsolute(client_dir) catch {}; allocator.free(client_dir); }
const test_content = "This is a test file for transfer.\n" ** 100;
var client = try SyncClient.connect(allocator, bound_addr, client_dir, test_secret, false);
defer client.disconnect();
try client.sendFile("hello.txt", test_content);
server_thread.join();
// verify the file arrived
var path_buf: [4096]u8 = undefined;
const p = std.fmt.bufPrint(&path_buf, "{s}/hello.txt", .{server_dir}) catch unreachable;
const received = try std.fs.cwd().readFileAlloc(allocator, p, 1024 * 1024);
defer allocator.free(received);
try std.testing.expectEqualStrings(test_content, received);
}
test "e2e: CRC detects corruption" {
const data = "Hello, this is a test payload for CRC verification";
const checksum = protocol.crc16(data);
try std.testing.expectEqual(checksum, protocol.crc16(data));
const tampered = "Hello, this is a test payload for CRC verificatioN";
try std.testing.expect(protocol.crc16(tampered) != checksum);
}
test "conflict detection finds mutual edits" {
const conflict_mod = @import("conflict.zig");
const allocator = std.testing.allocator;
const local = [_]manifest.FileEntry{.{
.path = "shared.zig", .file_type = .regular, .size = 500,
.mtime_ns = 2000, .checksum = [_]u8{0xAA} ** 32, .has_checksum = true,
}};
const remote = [_]manifest.FileEntry{.{
.path = "shared.zig", .file_type = .regular, .size = 600,
.mtime_ns = 1500, .checksum = [_]u8{0xBB} ** 32, .has_checksum = true,
}};
const conflicts = try conflict_mod.detectConflicts(allocator, &local, &remote, 1000);
defer allocator.free(conflicts);
try std.testing.expectEqual(@as(usize, 1), conflicts.len);
try std.testing.expectEqualStrings("shared.zig", conflicts[0].path);
}
The end-to-end tests spawn a real server thread, connect a real client, and verify actual data transfer. This catches integration bugs that unit tests miss -- like forgetting to read the full header before parsing, or sending the ack before writting the chunk to disk.
Using port 0 (std.net.Address.resolveIp("127.0.0.1", 0)) lets the OS assign a random available port, which avoids the "port already in use" problem when running tests in parallel. We grab the actual bound address from server.listener.listen_address after binding. This is a common pattern for test servers.
The auth rejection test verifies that a client with the wrong secret gets rejected cleanly -- no crash, no half-authenticated state, just a proper error. Security tests are just as important as functionality tests; if your auth can be bypassed by sending unexpected data, you don't really have auth.
$ zig build test
All 18 tests passed.
(That's 7 tests from parts 1 and 2 plus 7 new end-to-end tests, plus the 4 unit tests for protocol/throttle/conflict.)
This episode was a LOT of ground to cover. Here's the full picture:
Having said that, we have one more part to go. The final episode will polish everything into a usable command-line tool: argument parsing, configuration files, ignore patterns (like .gitignore), and proper logging. We'll also add a --watch mode that monitors the filesystem and syncs incrementally as files change (remember the file watcher from episode 63). The building blocks are all in place now -- network protocol, delta transfer, manifests, auth, throttling. Part 4 ties them together into something you'd actually use day-to-day.
@scipio