Incremental Push Protocol for Global State Updates & Performance Trade-offs

When synchronizing global state in distributed systems, we often face a dilemma: should we periodically send a full snapshot (Full State) or only send incremental changes (Incremental Delta)?

Sending full state is simple but extremely wasteful of bandwidth; sending only deltas requires maintaining complex change history and risks inconsistency if packets are dropped. In the design of the distributed_state module, we adopted a hybrid Incremental Push Protocol.

Protocol Strategy

The protocol balances real-time performance and bandwidth consumption using two key parameters:

1. GlobalDataUpdateInterval: The mandatory refresh cycle for global data (fallback mechanism).
2. LocalDeltaMaxAge: The maximum lifespan of local incremental changes.

The system prioritizes accumulating local Deltas. A push is triggered only when the quantity or age of the Deltas exceeds a threshold. This avoids establishing a network transmission for every tiny state change.

Code Demonstration (Zig)

Zig's explicit memory management and compact struct layout are perfectly suited for implementing such low-level protocols. The following code demonstrates how to build an efficient Delta batching mechanism for pushing to the network layer.

const std = @import("std");
const ArrayList = std.ArrayList;
const Allocator = std.mem.Allocator;

// Simulated Network Layer
const NetworkLayer = struct {
    pub fn send(data: []const u8) void {
        std.debug.print("Network: Sending {d} bytes payload.\n", .{data.len});
    }
};

// State Change Record
const Delta = struct {
    key: u64,
    value: f64,
    timestamp: i64,
};

const PushProtocol = struct {
    allocator: Allocator,
    deltas: ArrayList(Delta),
    max_age_ms: i64,
    max_batch_size: usize,
    last_flush: i64,

    pub fn init(allocator: Allocator, max_age: i64, batch_size: usize) PushProtocol {
        return PushProtocol{
            .allocator = allocator,
            .deltas = ArrayList(Delta).init(allocator),
            .max_age_ms = max_age,
            .max_batch_size = batch_size,
            .last_flush = std.time.milliTimestamp(),
        };
    }

    pub fn deinit(self: *PushProtocol) void {
        self.deltas.deinit();
    }

    // Try to add a change; trigger push if conditions are met
    pub fn push_delta(self: *PushProtocol, key: u64, value: f64) !void {
        const now = std.time.milliTimestamp();
        
        try self.deltas.append(Delta{
            .key = key,
            .value = value,
            .timestamp = now,
        });

        const age = now - self.last_flush;
        const size = self.deltas.items.len;

        // Core Decision Logic: Send only if batch is full or stale
        if (size >= self.max_batch_size or age >= self.max_age_ms) {
            try self.flush(now);
        }
    }

    fn flush(self: *PushProtocol, now: i64) !void {
        if (self.deltas.items.len == 0) return;

        std.debug.print("Flushing: Triggered by (Size: {d}, Age: {d}ms)\n", 
            .{self.deltas.items.len, now - self.last_flush});

        // Serialization logic (simplified: direct slice to bytes)
        const payload = std.mem.sliceAsBytes(self.deltas.items);
        NetworkLayer.send(payload);

        // Reset state but keep capacity
        self.deltas.clearRetainingCapacity();
        self.last_flush = now;
    }
};

pub fn main() !void {
    var gpa = std.heap.GeneralPurposeAllocator(.{}){};
    const allocator = gpa.allocator();
    defer _ = gpa.deinit();

    // Policy Config: Send every 10 changes OR every 500ms
    var proto = PushProtocol.init(allocator, 500, 10);
    defer proto.deinit();

    std.debug.print("--- Simulation Start ---\n", .{});

    // Scenario 1: Burst writes -> Trigger count threshold
    var i: u64 = 0;
    while (i < 12) : (i += 1) {
        try proto.push_delta(i, @as(f64, @floatFromInt(i)) * 1.5);
    }

    // Scenario 2: Slow writes -> Trigger time threshold
    std.time.sleep(600 * std.time.ns_per_ms);
    try proto.push_delta(99, 123.456);

    std.debug.print("--- Simulation End ---\n", .{});
}

Memory Layout Advantages

Using std.mem.sliceAsBytes to handle ArrayList(Delta) in Zig is incredibly efficient. Because the Delta struct is Plain Old Data (POD), its memory layout is contiguous and deterministic. We don't need heavy string serialization like JSON; instead, we directly hand the memory block (Raw Bytes) to the network layer.

This Zero-copy (or Near-zero-copy) philosophy is key to maintaining low CPU usage for high-performance incremental push protocols under high concurrency. By precisely controlling when to push (Batching) and how to serialize (Memory Mapping), we minimize network overhead.