DEV Community: Yashaswi Kumar Mishra

I Implemented Amazon's Dynamo Paper in Go

Yashaswi Kumar Mishra — Mon, 23 Feb 2026 06:41:05 +0000

I started rereading the papers I skimmed last year, and this time the Dynamo paper hit different. Not because the ideas are new, but because I finally understood why every design decision exists. So I did what any reasonable person would do: I spent a week building the whole thing from scratch.

The result is plethora - about 1500 lines of Go, 60 tests, all 9 key techniques from the paper. Before you get excited: this is not production software. All nodes run in a single process, storage is in-memory, there's no disk persistence. It's a learning project. But the algorithms are real, the nodes talk over real gRPC, and the tests spin up actual servers on random ports.

Let me walk you through how I built it, bottom-up, one problem at a time. Each layer exists because the previous one left a gap.

First, What Even Is Dynamo?

Dynamo is Amazon's internal key-value store. The paper came out in 2007 and it changed how people think about distributed databases. The core idea is deceptively simple:

Availability over consistency.

Amazon's shopping cart is the motivating example. Imagine a customer adds items to their cart. That write must succeed. If a data center is on fire, if nodes are crashing, if the network is partitioned, the customer should still be able to add to their cart. Losing a write means losing revenue. Amazon would rather show you a slightly stale cart than show you an error page.

This is the opposite of how traditional databases work. A MySQL primary will reject your write if it can't confirm it's been safely replicated. That's strong consistency, you always read what you last wrote, but sometimes you can't write at all. Dynamo flips this: you can always write, but when you read, you might get slightly stale data, or even multiple conflicting versions. The system will sort itself out eventually.

This tradeoff has a name: eventual consistency. It means:

Writes are never rejected (as long as at least one node is reachable)
Reads might return old data temporarily
But given enough time and network connectivity, all replicas will converge to the same state

The "eventually" part is not hand-waving. Dynamo has concrete mechanisms to make convergence happen: vector clocks track causality, anti-entropy repairs divergence, gossip detects failures. Every technique in the paper exists to make "eventually" happen faster and more reliably.

There's a deeper idea here too. In traditional databases, the database resolves conflicts (last-write-wins, or reject the write). In Dynamo, the application resolves conflicts. When you read a key and get two conflicting versions, your application decides what to do. For a shopping cart, you merge them (union of items). For a user profile, you might pick the latest. This is called semantic reconciliation and it's why Dynamo stores a context (vector clock) with every value -so the application has enough information to make that decision.

Storage That Never Says No

Most key-value stores have a simple contract: one key, one value. Put("name", "alice") replaces whatever was there before. Dynamo can't do this because it allows concurrent writes from multiple nodes. Two nodes might accept writes to the same key at the same time with no coordination between them.

So our storage maps each key to a list of values:

type Storage struct {
    lock sync.RWMutex
    data map[Key][]Value  // key -> list of concurrent versions
}

When you Get("name"), you might get back ["alice", "bob"]. Both values are valid. They were written concurrently and neither one is "more correct" than the other. The application gets to decide which one wins (or if it should merge them).

But we can't just blindly accumulate values forever. If someone reads a value, modifies it, and writes it back, the old version is no longer relevant, the new one causally supersedes it. We need a way to detect this. That's where vector clocks come in.

Vector Clocks - Tracking Who Knows What

Here's the problem with timestamps in distributed systems: clocks lie. Node A's clock might say 3:00:01pm, Node B's might say 3:00:03pm. If both write to the same key, comparing timestamps tells you nothing about which write knew about the other. Maybe B's write came first in real time but A's clock is fast. Maybe they wrote simultaneously and neither knew about the other.

What we actually care about is causality. Did write B happen after seeing write A? Or did they happen independently?

A vector clock solves this. It's a map from node IDs to counters:

{node-1: 3, node-2: 1}

This means: "this value incorporates 3 writes from node-1 and 1 write from node-2." Every time a node writes, it bumps its own counter. When you read before writing (the read-modify-write pattern), the new clock inherits all the entries from the value you read.

Two clocks can be compared:

// Descends returns true if vc happened after (or at the same time as) other
func (vc VClock) Descends(other VClock) bool {
    for nodeID, version := range other {
        if vc[nodeID] < version {
            return false
        }
    }
    return true
}

If clock A has every entry >= clock B, then A happened after B. This means A saw B (or something that saw B). B is an ancestor, it's safe to discard.

If neither clock descends from the other, the writes are concurrent. Neither writer saw the other's value. Both are equally valid.

Let's trace through an example:

1. client writes "alice" through node-1
   clock: {node-1: 1}

2. client reads "alice", gets clock {node-1: 1}
   client writes "alice-updated" through node-1
   clock: {node-1: 2}

   {node-1: 2} descends from {node-1: 1}, so "alice" is dropped.
   storage: ["alice-updated"]

3. two clients read "alice-updated" simultaneously
   client A writes "version-A" through node-1 → clock: {node-1: 3}
   client B writes "version-B" through node-2 → clock: {node-1: 2, node-2: 1}

   neither descends from the other! concurrent siblings.
   storage: ["version-A", "version-B"]

This gives us smart conflict resolution in storage:

func (s *Storage) Put(key Key, val Value) {
    existing := s.data[key]

    var kept []Value
    for _, ev := range existing {
        if ev.Clock.Descends(val.Clock) {
            return // existing is newer, ignore this write
        }
        if val.Clock.Descends(ev.Clock) {
            continue // new value supersedes, drop the old one
        }
        kept = append(kept, ev) // concurrent, keep both
    }
    s.data[key] = append(kept, val)
}

Three outcomes, every time:

Existing descends from new: new write is stale, throw it away
New descends from existing: existing is an ancestor, replace it
Neither descends: concurrent conflict, keep both as siblings

This is the core of Dynamo's consistency model. We never reject a write. We never silently overwrite. We keep everything that's causally distinct, and let the reader deal with it later. Consistency is lazy here -it happens when someone reads, not when someone writes.

The Node

A node wraps storage with an identity and a hint store (we'll need this for hinted handoff later):

type Node struct {
    NodeID    string
    Addr      string
    Storage   *storage.Storage
    HintStore map[string][]HintedItem
}

There are three write paths, each for a different situation:

Put(key, val) -builds a clock, stores. Used in unit tests.
Store(key, val) -takes a pre-built value, stores as-is. Used by the ring to replicate an already-prepared value. The coordinator builds the clock once, and all replicas store the exact same value.
StoreHint(targetNodeID, key, val) -stores in a separate hint store, tagged with the intended owner. Never shows up in Get. It's a temporary holding pen.

Why three? Because in a replicated system, the coordinator node is the one that increments the clock. The other N-1 replicas just store the value the coordinator gave them. And sometimes a stand-in node holds data that isn't even meant for it. Each scenario needs different behavior.

Consistent Hashing : Distributing Keys

Now we need to spread keys across nodes. The naive approach -hash(key) % numNodes -breaks catastrophically when nodes join or leave. If you have 10 nodes and add an 11th, hash % 10 and hash % 11 give different results for almost every key. You'd need to move nearly all your data. That's not going to work.

Consistent hashing solves this. The Dynamo paper describes three strategies:

Strategy 1 -each node picks T random positions on the hash ring. Keys are assigned to the nearest node clockwise. The problem: when a node joins, it picks random positions that split existing ranges unevenly. Data transfer during membership changes is unpredictable and requires scanning.

Strategy 2 -same random positions, but decouple key ownership from partitioning. Keys hash to positions on the ring, but the partition structure adapts. Better than Strategy 1 but still has the "random positions" problem where load balance depends on randomness.

Strategy 3 -what we use. Forget random positions entirely. Divide the hash space into Q equal-sized partitions upfront and assign them round-robin:

partitions := make([]Partition, q)
for i := 0; i < q; i++ {
    partitions[i] = Partition{
        ID:    i,
        Token: Token{NodeID: nodes[i%s].NodeID},
    }
}

With Q=20 and 10 nodes, each node owns exactly 2 partitions. The boundaries never move. Adding a node just means reassigning some partitions. No data rehashing. Lookup is O(1):

func (r *Ring) Lookup(key string) *node.Node {
    hash := md5Hash(key)
    partitionID := int(hash % uint64(r.Q))
    ownerID := r.Partitions[partitionID].Token.NodeID
    return r.NodeMap[ownerID]
}

Hash the key, mod by Q, look up who owns that partition. Done. The paper recommends Strategy 3 because it gives even load distribution, deterministic partition assignment, and simple data transfer during membership changes.

Replication and Sloppy Quorum : The Availability Trick

A single copy of data on one node is a disaster waiting to happen. If that node goes down, the data is gone (well, in our case it's in-memory so it's really gone, but even with persistence, it's unreachable).

Dynamo replicates each key to N nodes. The preference list for a key walks the ring clockwise from the key's partition and collects N distinct nodes. With N=3, every key lives on three nodes.

Now, here's where it gets interesting. Traditional quorum says: to write, you need W nodes to acknowledge. To read, you need R nodes to respond. As long as R + W > N, at least one node in the read set has the latest write. With N=3, R=2, W=2: 2+2 > 3, so you're guaranteed to read the latest write.

But strict quorum means if 2 out of 3 nodes are down, you can't write. That violates Dynamo's "always writable" guarantee. Enter sloppy quorum: when a preferred node is down, don't fail. Find the next healthy node on the ring and use it as a stand-in. The stand-in stores the data temporarily and hands it back when the real owner recovers.

This is the key insight of Dynamo. You don't need the specific W nodes to acknowledge. You need any W healthy nodes. The preference list just shifts to accommodate failures:

type Target struct {
    Node    *node.Node
    HintFor string  // empty = preferred node, non-empty = stand-in
}

Our preference list builds this in two passes:

First pass: walk the ring, find the N ideal preferred nodes, ignoring health
Second pass: walk again. Preferred + alive = regular target. Preferred + dead = queue for stand-in. Non-preferred + alive + dead queue non-empty = stand-in covering the first dead preferred node.

if preferred[nid] {
    if r.isAlive(nid) {
        targets = append(targets, Target{Node: r.NodeMap[nid]})
    } else {
        deadPreferred = append(deadPreferred, nid)
    }
} else if r.isAlive(nid) && len(deadPreferred) > 0 {
    targetID := deadPreferred[0]
    deadPreferred = deadPreferred[1:]
    targets = append(targets, Target{Node: r.NodeMap[nid], HintFor: targetID})
}

This makes Put beautifully simple. It just iterates targets ; if HintFor is empty, it's a regular put. If not, it's a hinted put:

for _, t := range targets {
    if t.HintFor == "" {
        if client.RemotePut(t.Node.Addr, key, value) == nil { acks++ }
    } else {
        if client.RemoteHintedPut(t.Node.Addr, key, value, t.HintFor) == nil { acks++ }
    }
}

The write succeeds with 3 acks (>= W=2). The customer's cart item is saved. Node-2 is on fire. Nobody cares. Availability is preserved.

gRPC -Making It Real

Everything so far was in-process function calls. A distributed system needs actual network communication. We use gRPC with protobuf:

service KV {
    rpc Put(PutRequest) returns (PutResponse);
    rpc Get(GetRequest) returns (GetResponse);
    rpc HintedPut(HintedPutRequest) returns (PutResponse);
    rpc GetKeyHashes(GetKeyHashesRequest) returns (GetKeyHashesResponse);
    rpc SyncKeys(SyncKeysRequest) returns (SyncKeysResponse);
    rpc Gossip(GossipRequest) returns (GossipResponse);
}

Six RPCs. First three: reads and writes. Next two: anti-entropy (merkle trees). Last one: gossip protocol. Each node runs a gRPC server. The ring talks to nodes through thin client wrappers.

The tests are the fun part - they spin up real gRPC servers on random OS-assigned ports:

lis, _ := net.Listen("tcp", "localhost:0")  // OS picks the port
srv := server.NewServer(n, seeds, replicaPeers, tFail)
go srv.Start(lis)

No mocks. Real servers. Real network calls. When a test kills a node, it actually stops the gRPC server and the connection fails for real.

Hinted Handoff - Getting Data Back Home

The stand-in saved the data. Great. But it's on the wrong node. The data for key user:alice is sitting on node-4, but node-2 (the rightful owner) doesn't know it exists. When node-2 recovers, we need to move that data back.

Each server runs a background goroutine on a 5-second ticker:

func (s *Server) runHandoff() {
    ticker := time.NewTicker(5 * time.Second)
    defer ticker.Stop()

    for range ticker.C {
        for _, m := range s.members.Alive() {
            items := s.node.DrainHints(m.NodeID)
            for _, item := range items {
                client.RemotePut(m.Addr, item.Key, item.Value)
            }
        }
    }
}

Every 5 seconds: check who's alive (via gossip), drain any hints destined for them, forward via regular RemotePut. DrainHints is atomic -it returns all hints and deletes them in one operation. Each hint is forwarded exactly once.

Hints live in a separate store. They never show up in Get responses. They're invisible to the outside world -just temporary holding pens waiting for the real owner to come back. Once forwarded, they're gone. The stand-in goes back to its normal life. (It never asked for this responsibility.)

Merkle Trees - Catching What Hinted Handoff Misses

Hinted handoff handles known failures, node was down, stand-in picked up the slack. But what about silent divergence? Maybe a put partially succeeded (2 out of 3 replicas got it, the third's network packet got dropped). Maybe a bug corrupted one replica's storage. Maybe a cosmic ray flipped a bit. (Okay probably not that last one. But you get the idea.)

The point is: replicas can drift apart without anyone noticing. We need a background process that periodically checks: "hey, do we still agree on everything?"

This is anti-entropy, and the efficient way to do it is with merkle trees.

A merkle tree is a binary hash tree. The leaves are hashes of individual key-value pairs. Each parent node's hash is computed from its children's hashes. The root hash is a fingerprint of the entire dataset.

        [root: a1b2c3]
       /              \
  [left: d4e5]    [right: f6a7]
   /       \        /       \
[key:a]  [key:b]  [key:c]  [key:d]

The magic: if two trees have the same root hash, all data is identical. You're done. One comparison. If roots differ, walk down: compare left children. If they match, the divergence is on the right side. Keep going until you find exactly which leaves differ. That's O(log n) comparisons instead of comparing every single key.

func Diff(a, b *MerkleNode) []string {
    if a == nil && b == nil { return nil }
    if a == nil { return collectKeys(b) }
    if b == nil { return collectKeys(a) }
    if a.Hash == b.Hash { return nil }  // subtree in sync!

    if a.Left == nil && b.Left == nil {
        // both are leaves -this specific key diverged
        return []string{a.Key}
    }
    // recurse into children
    return append(Diff(a.Left, b.Left), Diff(a.Right, b.Right)...)
}

Every 30 seconds, each server picks a replica peer (a node that shares some of its key ranges), fetches their key hashes, builds both merkle trees locally, diffs them, and syncs only the divergent keys:

Important detail: the merkle tree never goes over the wire. Only flat key-hash pairs and actual values cross the network. Each node builds the tree locally. The tree is a local optimization for efficient comparison, not a data structure that needs serialization.

The node also caches its merkle tree with a dirty flag -every write marks it dirty, and MerkleTree() rebuilds lazily only when someone asks. No point rebuilding a tree nobody's looking at.

This is the "eventual" in eventual consistency made concrete. Even if everything else fails -hinted handoff didn't fire, a write was partially lost, anti-entropy will catch it within 30 seconds and repair the divergence. Given enough time, replicas always converge.

Gossip Protocol - Who's Alive, Who's Dead

All of the above depends on one thing: knowing which nodes are alive. The preference list needs to know who's dead to pick stand-ins. Hinted handoff needs to know who recovered. Anti-entropy needs to skip dead peers.

A centralized health checker would work, but that's a single point of failure, exactly what Dynamo avoids. Instead, Dynamo uses a gossip protocol. It's called gossip because it works like actual gossip: you tell a friend, they tell a friend, eventually everyone knows.

Each node maintains a membership list. Each entry has:

NodeID and address: who and where
Heartbeat counter: a monotonic number, only incremented by the owning node
Timestamp: local wall time when we last saw this node's heartbeat increase

Every second, each node:

Increments its own heartbeat counter
Picks a random known peer
Sends its full membership list
Receives the peer's list back
Merges: for each entry, keep the higher heartbeat

func (m *MemberList) Merge(remote []MemberEntry) {
    now := time.Now()
    for _, r := range remote {
        if r.NodeID == m.selfID { continue }

        local, exists := m.members[r.NodeID]
        if !exists {
            // new node! discovered through gossip
            m.members[r.NodeID] = &MemberEntry{
                NodeID: r.NodeID, Addr: r.Addr,
                Heartbeat: r.Heartbeat, Timestamp: now,
            }
        } else if r.Heartbeat > local.Heartbeat {
            local.Heartbeat = r.Heartbeat
            local.Timestamp = now
        }
    }
}

Failure detection is timestamp-based. If we haven't seen a node's heartbeat increase within tFail (say, 10 seconds), we consider it dead:

func (m *MemberList) IsAlive(nodeID string) bool {
    e, exists := m.members[nodeID]
    if !exists { return false }
    return time.Since(e.Timestamp) < m.tFail
}

The beautiful thing is transitive discovery. Nodes don't need to know the full cluster at startup. Each node starts with just 2 seed nodes. When node-1 gossips with node-2, it learns about everyone node-2 knows. Information spreads exponentially. In our demo with 10 nodes, each starting with only 2 seeds:

[GOSSIP] after 1s: each node sees 3/10 members
[GOSSIP] after 3s: most nodes see 10/10
[GOSSIP] fully converged after 4s!

4 seconds for 10 nodes to fully discover each other, starting from knowing only 2 neighbors. Gossip is absurdly effective. And because RandomPeer() includes dead peers, we can also detect recoveries ; if a dead node starts responding again, its heartbeat counter will jump and we'll mark it alive.

The ring's IsAlive callback reads directly from gossip state. When a node dies, gossip detects it within tFail seconds. The preference list automatically routes around it. When it recovers, gossip picks it up, handoff fires, data flows home. No manual intervention. No central coordinator. Just peers talking to peers.

The Three Background Loops

Every server runs three goroutines. Together, they make "eventual" happen:

Goroutine	Interval	What It Does
`runGossip`	1s	Tick heartbeat, pick random peer, exchange membership lists
`runHandoff`	5s	Forward hints to recovered nodes
`runAntiEntropy`	30s	Compare merkle trees with replica peers, sync divergent keys

Think of them as layers of a safety net:

Gossip keeps the membership view fresh (who's alive, who's dead)
Handoff recovers from known failures (data on wrong node)
Anti-entropy catches everything else (silent divergence, missed writes)

If gossip fails, nodes can't detect recovery, so handoff won't fire. But anti-entropy will still eventually sync the data. If anti-entropy is slow, handoff gets the urgent stuff done fast. They're complementary. Belt and suspenders. Belt and suspenders and also a rope tied to the ceiling just in case.

The Full Picture

Here's a complete write flow with all layers working together:

Every layer has a job. Storage accepts concurrent versions. Vector clocks track causality. Consistent hashing distributes keys. Sloppy quorum ensures writes always succeed. gRPC makes it networked. Hinted handoff gets data home. Merkle trees catch drift. Gossip keeps the membership view alive.

What This Doesn't Cover

Real talk: this is a learning implementation. Here's what production Dynamo has that we don't:

Persistence - our storage is in-memory. A restart loses everything. Real Dynamo uses a write-ahead log and persistent storage (originally BDB, later custom).
Multi-process - all our nodes run in one Go binary. Real Dynamo runs each node as a separate process on separate machines in separate racks.
Dynamic ring - our ring topology is fixed at startup. Real Dynamo gossips partition ownership so nodes can join and leave without downtime.
Read repair - we deduplicate on reads but don't write the resolved value back to stale replicas. Real Dynamo does this for faster convergence.
Per-node ring view - we have one shared ring. In reality, each node has its own ring instance driven by its own gossip state.

These are engineering concerns, not algorithmic ones. The hard part of Dynamo is the algorithms, the consistent hashing, the vector clock causality, the sloppy quorum trick, the merkle tree diffing, the gossip convergence. Those are all here, working, tested.

Learnings

Building Dynamo changes how one thinks about correctness.

In a traditional database, correctness means "every read returns the last write." That's a clean, satisfying guarantee. Dynamo says: forget that. Correctness here means "no write is ever lost, and the system never stops accepting writes." Those are fundamentally different promises, and everything in the paper - vector clocks, sloppy quorum, merkle trees, gossip - exists to make the second promise work without the first one falling apart too badly.

The thing that surprised me most is what the paper doesn't do. There's no optimistic locking. No compare-and-swap. No way for a client to say "write this only if nobody else has written since I last read." If two clients read the same key, compute something, and write back - both writes succeed. You get siblings. The paper's answer is "the application merges them later." For a shopping cart, that's fine - take the union. For an inventory counter - say, "3 left in stock" - two concurrent decrements both succeed, both think they got the last valid item, and now you've oversold.

This is why DynamoDB (the actual AWS product) bolted on conditional writes and transactions years later. The paper's design is beautiful and internally consistent, but it only works for a specific class of applications - ones where conflicts can be resolved by simple merge logic, not ones where you need "this or nothing."

The other thing: availability isn't free. Dynamo trades consistency for availability, but the engineering cost of that trade is enormous. You need vector clocks to track causality. You need sloppy quorum so writes survive failures. You need hinted handoff to get data back home. You need merkle trees to catch silent drift. You need gossip so nodes can find each other without a coordinator. Nine techniques, all layered on top of each other, all because we refused to say "sorry, try again later" to a single write.

That's the real lesson. Choosing availability over consistency isn't the easy path - it's the harder one. You just pay the complexity cost in different places.

lowkey: Building a Distributed Lock Service That Actually Works

Yashaswi Kumar Mishra — Sun, 28 Dec 2025 16:25:32 +0000

Yes, sometimes the payments processing cron job can run twice. In the same batch and for the same customer. Unfortunately, the customer would be charged twice.

How did this happen? Two instances of the same server ran the cron job twice. “But we did have locks, right?” You did. It was probably Redis Redlock. It worked perfectly in development. But production isn't kind to assumptions. Say a network partition occurred leading two instances think they're in charge and the poor customer gets charged twice and raises angry support tickets. Poor you are awake at 3 AM.

This is why I built lowkey - a distributed lock service that understands Murphy's Law: If it can go wrong, it will.

What is lowkey?

lowkey makes three promises:

Only ONE - Strong consistency via Raft consensus
Fencing tokens - Simple math prevents stale writes
Fast - 3.24ms latency (faster than etcd's 5-10ms) [p50 benchmark]

Let's talk about why distributed locks are genuinely hard, and how to stop charging the same customer twice.

The Fundamental Problem of Time

You cannot really trust time in distributed systems.

Your process thinks it's been 1 second. But in reality it's been 30 seconds or anything longer than 1 second. You paused for a GC cycle, or the OS scheduled someone else, or the network was slow. Doesn't matter. You missed the lease expiration, and when your process wakes up, it thinks that only a second has passed.

This is called a process pause. It is not theoretical and almost causes production chaos daily.

So how do we solve it?

The CAP Theorem Choice

CAP theorem: Consistency, Availability, Partition Tolerance. We have to pick one out of C and A.

CAP theorem states that in a distributed system, when a network partition happens, you can only guarantee one of the following two properties out of C and A:

Consistency (C)

Every node sees the same data at the same time. Once a write succeeds, all future reads see it.
Availability (A)

Every request receives a response, even if that response might not be the latest value.
Partition tolerance (P)

The system continues operating even when the network splits and messages are dropped or delayed.

For distributed locks, this isn't a choice. It's a requirement: You MUST pick CP (Consistency + Partition Tolerance).

Here's why Availability is the wrong choice:

In an AP system like Redis Redlock, if a network partition occurs :

node A can’t see node B
node B can’t see node A

Both nodes are still alive. Both think they’re healthy. Both say:

“I am available”

This is already the critical moment. This is called Split-Brain situation.

Availability is being preserved by lying about the global state.

Availability means "always respond, even if wrong". For locks, wrong equals disaster.

Whereas, lowkey follows a CP system.

In a CP system, nodes don’t pretend they’re alone in the universe. They are explicitly aware that:

other nodes exist
agreement is required
proceeding without quorum is dangerous

So when a network partition happens, the system splits into two very unequal worlds.

The Three Pillars: Leases, Fencing Tokens, and Raft

Leases

A lease is like a parking meter. You pay for 2 hours. After 2 hours, your time expires. You lose the spot. Automatic.

Lease is the main entity in lowkey. Every owner(node) has to start a lease to acquire locks. Every lock is associated with a lease.

// Client creates a lease
lease := client.CreateLease(TTL: 10 * time.Second)
// lease_id = 100

// Client uses this lease for all locks
lock := client.Acquire("job-name", lease_id: 100)

// Client must send keepalive every ~3 seconds (TTL/3)
// If keepalive stops -> lease expires -> locks released

Without leases, a crashed client holds locks forever. System gets stuck. With leases, system recovers automatically.

The flow looks like this.

Time: 0s → Client creates lease (expires at 10s) Time: 3s → Client: keepalive → expires at 13s Time: 6s → Client: keepalive → expires at 16s Time: 9s → Client: keepalive → expires at 19s --- CLIENT CRASHES --- Time: 12s → (no keepalive) Time: 15s → (no keepalive) Time: 19s → Lease expires → Server auto-releases all locks → Other clients can proceed

Fencing Tokens

Imagine a deli counter with number tickets. Person #43 cannot be served after person #44. The number proves freshness.

Fencing tokens are the same idea: monotonically increasing numbers that prove "I'm not a zombie client from the past".

Why are fencing tokens important?

This is how it happens in lowkey :

The token must be validated by the protected resource (database, API, file, etc.). The lock service provides the token, but the resource enforces it.

In lowkey:

// Client gets token when acquiring lock
lock := client.Acquire("migration-job")
token := lock.Token()  // e.g., 42

// Client includes token in EVERY protected operation
database.Execute(query, token: token)

// Database validates
func (db *Database) Execute(query string, token uint64) error {
    if token < db.lastSeenToken {
        return errors.New("stale token - rejected")
    }

    // Execute query
    db.lastSeenToken = token
    return nil
}

Raft Consensus

Leases and Fencing tokens solve client-side failures. What about server-side failures? If you have a single lock server and it crashes, your entire system is down. No locks translates to no progress.

The solution: Multiple servers with consensus

But here's the catch: multiple servers create a new problem, who's in charge?

Without consensus, the problem of split brain occurs that we discussed before.

This is why we need consensus - a protocol that ensures all servers agree on:

Who is the leader (only one can hand out locks)
What the current state is (which locks are held)
What operations have been committed (which token numbers are valid)

Why Raft specifically?

There are other consensus algorithms (Paxos, ZAB, Viewstamped Replication). We chose Raft because:

Understandable - Designed explicitly to be easier to understand than Paxos
Proven - Used in production by etcd, Consul, CockroachDB
Battle-tested - HashiCorp's implementation has years of production use
Leader-based - Simpler than leaderless protocols, easier to reason about failures

This is how a leader based consensus helps us :

Network partition: [Server 1] X [Server 2, Server 3] Server 1 (minority): → Tries to become leader → Can't get majority votes (1 out of 3) → Stays as follower → Rejects all write requests → READ-ONLY mode Server 2, 3 (majority): → Server 2 becomes leader (2 out of 3 votes) → Can commit writes (has quorum) → Hands out locks with monotonic tokens → System continues When partition heals: → Server 1 realizes it's behind → Syncs from Server 2 (the true leader) → Consistency restored

Raft is like a jury reaching a verdict. Even if some jurors leave, the remaining majority must agree.

This is CP in action: sacrifice availability (minority can't write) for consistency (no split-brain).

lowkey Architecture

lowkey is built in 4 layers.

Layer 1: FSM (Finite State Machine)

The FSM is the "brain" that manages state:

type FSM struct {
    locks  map[string]*Lock   // "migration-job" → {owner, token, lease}
    leases map[uint64]*Lease  // 100 → {owner, expires_at}

    fencingCounter uint64     // Monotonically increasing
    clock *monotime.Clock     // NOT wall clock!
}

Design decision #1: Monotonic time

We don't use time.Now(). Why? System clocks can jump backwards (NTP sync, manual changes and so many reasons). Instead:

// Monotonic clock - only moves forward
type Clock struct {
    start time.Time  // Fixed reference point
}

func (c *Clock) Elapsed() Duration {
    return Duration(time.Since(c.start))  // Monotonic!
}

func (c *Clock) ExpiresAt(ttl Duration) Duration {
    return c.Elapsed() + ttl
}

This gives us a clock immune to system time changes. It only knows "time since server started", which only goes forward.

Design decision #2: Fencing counter is sacred

Every lock acquisition increments the fencing counter. This is strictly monotonic - never decreases, never repeats:

func (f *FSM) AcquireLock(lockName, ownerID string, leaseID uint64) (uint64, error) {
    // Validate lease exists and isn't expired
    lease := f.leases[leaseID]
    if lease == nil || lease.IsExpired(f.clock.Elapsed()) {
        return 0, ErrLeaseExpired
    }

    // Check if lock is available
    if lock, held := f.locks[lockName]; held && lock.LeaseID != leaseID {
        return 0, ErrLockAlreadyHeld
    }

    // THIS IS CRITICAL: increment before assigning
    f.fencingCounter++

    f.locks[lockName] = &Lock{
        Name:         lockName,
        OwnerID:      ownerID,
        FencingToken: f.fencingCounter,  // Unique, monotonic
        LeaseID:      leaseID,
    }

    return f.fencingCounter, nil
}

Layer 2: Raft Consensus

We use HashiCorp's Raft implementation - battle-tested, production-proven.

The critical optimization: Not everything needs consensus.

Wait, lease renewal doesn't go through Raft?

Operations that need Raft:

CreateLease (affects state)
AcquireLock (affects state)
ReleaseLock (affects state)

Operations that DON'T need Raft:

RenewLease (just extends time)

Here's the trick:

// RenewLeaseLocal - leader-only operation
func (n *Node) RenewLeaseLocal(leaseID uint64) (time.Duration, error) {
    // Safety check: must be leader
    if !n.IsLeader() {
        return 0, fmt.Errorf("not leader")
    }

    // Update expiration time in FSM (no Raft consensus)
    return n.fsm.RenewLeaseLocal(leaseID)
}

This is safe because:

Only the leader processes renewals
If leader crashes, then clients reconnect to new leader
If client can't reach new leader within TTL, then lease expires

Benchmark impact: This made heartbeats 10-100x faster. Raft consensus is expensive - use it only when necessary.

Layer 3: Server (gRPC + HTTP)

The server validates leadership and routes requests:

func (s *Server) AcquireLock(ctx context.Context, req *pb.AcquireLockRequest) (*pb.AcquireLockResponse, error) {
    // Apply command through Raft
    result, err := s.node.Apply(types.AcquireLockCmd{
        LockName: req.LockName,
        OwnerID:  req.OwnerId,
        LeaseID:  req.LeaseId,
    })

    if err != nil {
        return nil, toGRPCError(err)
    }

    token := result.(fsm.AcquireLockResponse).FencingToken

    return &pb.AcquireLockResponse{
        FencingToken: token,  // Critical: return token to client
    }, nil
}

Why Protocol Buffers (Protobuf)?

We chose Protobuf over JSON for the API layer. Here's why:

1. Performance - Binary encoding beats JSON

Message size for AcquireLockRequest:
- Protobuf: ~50 bytes
- JSON:     ~120 bytes
→ 2.4x smaller payload

Serialization speed:
- Protobuf: ~50 ns/op
- JSON:     ~500 ns/op
→ 10x faster serialization

2. Type safety - Compile-time guarantees

// lock.proto - the schema IS the contract
message AcquireLockRequest {
  string lock_name = 1;
  string owner_id = 2;
  uint64 lease_id = 3;
}

With JSON, you find bugs at runtime:

// JSON - runtime error
{"lock_name": 123}  // oops, should be string!

// Protobuf - compile-time error
req.LockName = 123  // won't compile, type mismatch

3. Schema evolution - Backward compatible

// Version 1
message Lock {
  string name = 1;
  uint64 token = 2;
}

// Version 2 - add optional field, old clients still work
message Lock {
  string name = 1;
  uint64 token = 2;
  int64 expires_at = 3;  // new field, backwards compatible
}

4. Multi-language support - One schema, all languages

Generate clients automatically:

protoc --go_out=. lock.proto        # Go
protoc --python_out=. lock.proto    # Python
protoc --java_out=. lock.proto      # Java
protoc --rust_out=. lock.proto      # Rust

5. gRPC integration - Native support

Protobuf is the native serialization for gRPC. You get:

Bidirectional streaming (for heartbeats)
HTTP/2 multiplexing (multiple RPCs on one connection)
Built-in authentication and load balancing

But wait, what about JSON users?

We support both! Using grpc-gateway:

Client sends JSON to /v1/lock/acquire

grpc-gateway translates to Protobuf
gRPC server processes Protobuf
grpc-gateway translates response back to JSON
Client receives JSON

Best of both worlds:

Internal efficiency (Protobuf)
External flexibility (JSON)

The tradeoff:

Protobuf requires schema management and code generation. But for a distributed lock service where every millisecond matters, the performance and type safety are worth it.

For lowkey specifically:

3.24ms average latency - every microsecond counts
Strong typing prevents token confusion (uint64, not string)
Schema evolution allows adding metrics without breaking clients

Layer 4: Client SDK

The SDK makes it easy:

// Create client
c, _ := client.NewClient("localhost:9000", "worker-1")
defer c.Stop()

// Start lease (automatic heartbeats every TTL/3)
c.Start(ctx, 10*time.Second)

// Acquire lock
lock, err := c.Acquire(ctx, "migration-job")
if err != nil {
    log.Printf("Another instance is running")
    return
}
defer lock.Release(ctx)

// Get fencing token
token := lock.Token()

// Use token in all protected operations
database.Migrate(token)

The SDK handles:

Automatic heartbeats in background
Error handling and logging
Reconnection on leader changes
Clean shutdown

Benchmarks: The Numbers

Benchmarked on AMD Ryzen 7 5800HS (16 cores), single-node cluster:

lowkey Benchmarks:
BenchmarkSequential-16       4460 ops     3.24ms/op
BenchmarkParallel-16        19911 ops     0.60ms/op
BenchmarkContention-16      10000 ops     1.40ms/op

What these numbers mean:

3.24ms sequential - Baseline Raft consensus latency for lock acquisition
0.60ms parallel - With no lock contention, throughput is 5.4x higher
1.40ms contention - Realistic scenario with competing clients

Why these numbers:

Optimized heartbeats → Leader-only renewal (no Raft overhead)
HashiCorp Raft → Battle-tested, highly optimized
Go → Concurrent, low-latency runtime
Smart design → Only critical ops need consensus

The key insight from benchmarks:

Sequential (3.24ms): Baseline Raft consensus latency
Parallel (0.60ms): No lock contention → 5.4x speedup
Contention (1.40ms): Realistic scenario → 2.3x faster than sequential

This proves the heartbeat optimization works - parallel throughput is dramatically higher because heartbeats don't bottleneck the system.

Latency Percentiles: What Users Actually Experience

Average latency (3.24ms) tells part of the story. But what about tail latency? The 1% of requests that take longer?

For distributed locks, tail latency matters:

p50 (median) - Half of requests are faster
p90 - 90% of requests are faster (most users)
p99 - 99% of requests are faster (outliers)
p99.9 - 99.9% of requests are faster (worst case)

lowkey percentile benchmarks (1000 samples, sequential):

Percentile Latency Benchmarks:
p50  (median):     2.87ms
p90  (90th):       4.12ms
p95  (95th):       4.58ms
p99  (99th):       5.94ms
p99.9 (99.9th):    8.21ms

What this tells us:

Percentile	Latency	Insight
p50	2.87ms	Half of lock acquisitions complete in under 3ms
p90	4.12ms	90% of users see sub-5ms latency
p99	5.94ms	Even outliers stay under 6ms
p99.9	8.21ms	Worst case is still under 10ms

Why this matters for production:

Scenario: API endpoint acquires lock before processing

With lowkey p99 = 5.94ms:
  → 99% of requests add <6ms overhead
  → Predictable, low tail latency
  → Good user experience

With slower system (p99 = 50ms):
  → 1% of requests add 50ms
  → Visible lag for users
  → Unpredictable performance

The distribution is tight:

p50 → p99 spread: 3.07ms (2.87ms to 5.94ms)
No huge outliers (p99.9 is only 8.21ms)
Raft consensus is predictable, not spiky

Takeaway: lowkey's tight percentile distribution means your p99 users don't get a worse experience. Consensus is expensive, but at least it's predictable.

Run the percentiles yourself:

make bench-percentiles

Design Decisions Recap

CP over AP → Consistency is non-negotiable for locks
Monotonic time → Wall clocks are lies, monotonic clocks are truth
Lease-based locks → Auto-cleanup on client failure
Fencing tokens → Math prevents stale writes
Raft consensus → Proven, understood, battle-tested
Leader-only heartbeats → 10-100x performance improvement
gRPC + HTTP → Flexibility for all languages

What I Learned Building This

Time is broken in distributed systems:

Wall clocks jump backward (NTP)
Processes pause unpredictably (GC)
Network delays are unbounded → Solution: Monotonic time + fencing tokens

You can't cheat the CAP theorem:

AP (Redis Redlock) → Split-brain under partition
CP (lowkey) → Only majority can make progress → For locks, CP is the only correct choice

Consensus is expensive:

Raft consensus: ~3-5ms per operation
Leader-only ops: ~0.5ms per operation → Use consensus sparingly, optimize the rest

Fencing tokens are mandatory:

Timeouts alone cannot prevent stale writes
Process pauses are real and unpredictable → The protected resource MUST validate tokens

Try It Yourself

// Clone the repository
git clone https://github.com/pixperk/lowkey.git
cd lowkey

// Build the binary
go build -o lowkey cmd/lowkey/main.go

// Start the server
./lowkey --bootstrap --data-dir ./data

// Use the SDK
go get github.com/pixperk/lowkey/pkg/client

Full documentation, examples, and benchmarks on GitHub.

The Bottom Line

Distributed locks are genuinely hard. Here's why:

Time is unreliable (wall clocks jump, processes pause)
Networks are unreliable (partitions happen)
Processes are unreliable (crashes, GC pauses)

But they're solvable with the right primitives:

Raft consensus → Strong consistency (CP in CAP)
Fencing tokens → Mathematical safety against stale writes
Monotonic time → Immune to clock drift
Leases → Automatic cleanup on failures

lowkey proves you can build a distributed lock service that's both correct (no split-brain, no stale writes) and fast (3.24ms, faster than etcd).

The real lesson? Distributed systems require paranoia.

Assume clocks lie. Assume networks partition. Assume processes pause at the worst possible moment. Then design systems that work anyway.

That's lowkey.

Writes done Right : Atomicity and Idempotency with Redis, Lua, and Go

Yashaswi Kumar Mishra — Sun, 30 Nov 2025 06:53:28 +0000

Life would have been easy if the world were filled with monolithic functions. Simple functions that execute once, and if they crash, we could try again.
But we build distributed systems and life isn’t that easy (but it is fun).

A classic scenario is : A user on our system clicks “Pay Now”. The backend needs to do two things :

Deduct the balance from the user’s wallet (Postgres).
Publish an event to send a confirmation email and notify the warehouse (Redis/Kafka).

This looks simple enough in code. But what if the database commits the transaction but the network acts up before the event is published onto the communication queue?
The user is charged, but the email is never sent, and so the warehouse never ships the item.
And if we try reversing it, the email is sent, but we do not get the payment.

This is the Dual Write Problem, and it is the silent killer of data integrity in microservices. To solve it, we cannot rely on luck; we need two architectural pillars: Atomicity and Idempotency.

In this one, we are going to move beyond the "happy path." We will architect a robust backend system using Go, Redis, and Postgres. We will implement the Transactional Outbox Pattern to guarantee that our database writes and event publications happen together (Atomicity), and we will use Redis to ensure that even if a user clicks the button ten times, they are only charged once (Idempotency).

You can find the complete, runnable source code for this implementation on my GitHub:
HERE

The Transactional Outbox Pattern

We have to solve the “Dual Write Problem”. For this, we need to stop treating our database and our message broker as two separate entities during the request lifecycle. They need to behave as one.

Since we cannot make Redis (our message broker here) participate in a Postgres transaction (without complex and slow 2-phase commits), we have to bring the message queue to the database.

This is the Transactional Outbox Pattern.

Instead of publishing a message directly to Redis, we insert the message into a local SQL table called outbox within the same transaction that modifies our business data.
This is an outgoing mailbox. We simply place the mail here and it is taken to the post-office (redis) by a mail carrier (background worker).
If the mailbox burns down (database rollback), the letter is burnt with that, so letter is never delivered and the recipient is not confused.
This preserves Consistency.

Our system needs to behave like this :

Let us set our schema up for our business table (orders) and our outbox (i.e the mailbox).

-- 1. The Business Table: Stores the actual state
CREATE TABLE orders (
    id UUID PRIMARY KEY,
    user_id UUID NOT NULL,
    amount INT NOT NULL, 
    status VARCHAR(50) DEFAULT 'PENDING',
    created_at TIMESTAMP DEFAULT NOW()
);

-- 2. The Outbox Table: Stores the intent to publish
CREATE TABLE outbox (
    id UUID PRIMARY KEY,
    event_type VARCHAR(255) NOT NULL, -- e.g., "order.created"
    payload JSONB NOT NULL,           -- The actual data to publish
    status VARCHAR(50) DEFAULT 'PENDING',
    created_at TIMESTAMP DEFAULT NOW()
);

As we have a clear picture of what to do now, we can write the go code to perform the operation. Let’s write the CreateOrder function.

type Order struct {
    ID     uuid.UUID `json:"id"`
    UserID uuid.UUID `json:"user_id"`
    Amount int       `json:"amount"`
}

func CreateOrder(ctx context.Context, db *sql.DB, order Order) error {
    // 1. Start the Transaction
    // This is the boundary of our atomicity.
    tx, err := db.BeginTx(ctx, nil)
    if err != nil {
        return fmt.Errorf("failed to begin transaction: %w", err)
    }

    // Defer a rollback. If the function exits without a Commit,
    // all changes (both order and outbox) are discarded.
    defer tx.Rollback()

    // 2. Insert the Business Record
    _, err = tx.ExecContext(ctx, 
        `INSERT INTO orders (id, user_id, amount) VALUES ($1, $2, $3)`,
        order.ID, order.UserID, order.Amount)
    if err != nil {
        return fmt.Errorf("failed to insert order: %w", err)
    }

    // 3. Insert the Outbox Record
    // We marshal the order data to JSON to serve as the event payload.
    payload, err := json.Marshal(order)
    if err != nil {
        return fmt.Errorf("failed to marshal payload: %w", err)
    }

    _, err = tx.ExecContext(ctx,
        `INSERT INTO outbox (id, event_type, payload) VALUES ($1, $2, $3)`,
        uuid.New(), "order.created", payload)
    if err != nil {
        return fmt.Errorf("failed to insert outbox event: %w", err)
    }

    // 4. Commit the Transaction
    // This is the moment of truth. Both the order and the event become visible
    // to the rest of the system at the exact same instant.
    if err := tx.Commit(); err != nil {
        return fmt.Errorf("failed to commit transaction: %w", err)
    }

    return nil
}

In this function, we aren't talking to Redis yet. We are strictly talking to Postgres. We prepare the order, we prepare the message, and we commit them together. All or nothing. (This is Atomicity).

By the end of this function, we have guaranteed that if an order exists, a corresponding event exists in the outbox table waiting to be picked up. We have successfully persisted our intent.

The Guard : Implementing Idempotency

In distributed systems, we cannot really trust the network.

A client might send a request, the server processes it, but the acknowledgment gets lost on the way back. The client, thinking the request failed, retries.

If we aren't careful, we just processed the same order twice.

To prevent this, we use Idempotency Keys. The client generates a unique ID (like a UUID) for every specific action and sends it in the header (Idempotency-Key).

We will use Redis to track these keys because it's fast and supports atomic operations.

Three States of a Request

To handle the robustness, a request key (idempotency key) isn’t just "present" or "absent.” It has three life stages :

Null (New): We have never seen this key. Lock it and proceed.
"PENDING" (In Progress): We are currently processing this key on another thread/machine. Tell the user to wait (409 Conflict).
JSON Payload (Completed): We finished this request 5 minutes ago. Return the saved response immediately.

For this, we will need Lua. WHY?

You might be tempted to write code like this :

// BAD CODE: DO NOT USE
if redis.Get(key) == "" {
    redis.Set(key, "PENDING")
    // ... process ...
}

This is dangerous. If two requests hit the same line at the exact same microsecond, both will see Get(key) == "" and both will proceed. This is a Race Condition.

Here is the disaster scenario:

Time 0.00s: Request A asks Redis: "Does Key X exist?" → Redis says: "No."
Time 0.00s: Request B asks Redis: "Does Key X exist?" → Redis says: "No."
Time 0.01s: Request A thinks: "Great, I'm first!" → Starts processing payment.
Time 0.01s: Request B thinks: "Great, I'm first!" → Starts processing payment.

You just charged the user twice.
The solution is Atomic Lock and Proceed.
To fix this, the "Check" and the "Lock" must happen as one indivisible action. In Redis, we do this with a Lua script. Redis guarantees that a Lua script is atomic, no other command runs while the script is executing.

The following is the Lua Script we write :

-- script.lua
local key = KEYS[1]
local pending_status = ARGV[1] -- Value: "PENDING"
local ttl = ARGV[2]            -- Expiration: e.g., 60s

-- 1. Check if key exists
local value = redis.call("GET", key)

-- 2. If it exists, return the value 
-- (This could be "PENDING" or the final JSON response)
if value then
    return value 
end

-- 3. If not, lock it with "PENDING" status and a TTL
-- The TTL is crucial: If our server crashes mid-process, 
-- this key must eventually expire so the user can retry.
redis.call("SET", key, pending_status, "EX", ttl)
return nil

Why the TTL (Time-To-Live) matters?

Imagine your Go server crashes exactly after acquiring the lock (setting "PENDING") but before it writes to the database.

The key is set to "PENDING".
The server is dead. It never deletes the key or updates it with a result.
The user retries. Redis sees "PENDING" and blocks them.
The user is locked out of their account forever.

The TTL is the "Dead Man's Switch." If the server dies, Redis will automatically delete the "PENDING" key after 60 seconds, allowing the user to try again.

Now we can wrap the business logic with our guard.

func HandleCreateOrder(w http.ResponseWriter, r *http.Request, db *sql.DB, rdb *redis.Client) {
    // 1. Get the Idempotency Key from headers
    idempotencyKey := r.Header.Get("Idempotency-Key")
    if idempotencyKey == "" {
        http.Error(w, "Missing Idempotency-Key", http.StatusBadRequest)
        return
    }

    // 2. Execute the Lua Guard
    ctx := r.Context()
    // Returns: nil (acquired lock), "PENDING", or "{"id":...}"
    val, err := rdb.Eval(ctx, luaScript, []string{idempotencyKey}, "PENDING", 60).Result()

    if err != nil && err != redis.Nil {
        http.Error(w, "Internal Server Error", http.StatusInternalServerError)
        return
    }

    // CASE A: Request is already running (Thundering Herd Protection)
    // If the user double-clicked, we stop the second request here.
    if val == "PENDING" {
        http.Error(w, "Request is processing, please retry shortly", http.StatusConflict)
        return
    }

    // CASE B: Request was already completed
    if val != nil {
        // We found a cached response! Return it and skip the DB entirely.
        w.Header().Set("Content-Type", "application/json")
        w.Write([]byte(fmt.Sprintf("%v", val)))
        return
    }

    // CASE C: New Request (val is nil)
    // We have acquired the lock. logic proceeds...

    orderID := uuid.New()
    order := Order{ID: orderID, UserID: uuid.New(), Amount: 1000}

    // --- CALL THE ATOMIC TRANSACTION WE WROTE EARLIER ---
    if err := CreateOrder(ctx, db, order); err != nil {
        // CRITICAL: If the DB transaction fails, we MUST delete the key.
        // If we don't, the key stays "PENDING" until it expires (60s), 
        // locking the user out.
        rdb.Del(ctx, idempotencyKey)
        http.Error(w, "Transaction failed", http.StatusInternalServerError)
        return
    }

    // 3. Update Redis with the Final Result
    // The transaction succeeded. We overwrite "PENDING" with the actual success JSON.
    response := fmt.Sprintf(`{"status": "success", "order_id": "%s"}`, orderID)

    // Cache this result for 24 hours so retries tomorrow still get the receipt.
    rdb.Set(ctx, idempotencyKey, response, 24*time.Hour) 

    w.Header().Set("Content-Type", "application/json")
    w.Write([]byte(response))
}

Now we have a fortress.

The Database is protected by SQL Transactions (Atomicity).
The API is protected by Redis Lua Scripts (Idempotency).

But what about those pending events in the outbox? The emails haven’t been sent.
The final piece of the puzzle is The Background Worker, the process that polls the outbox table and actually delivers the messages to the world.

Let us write the worker.

The Background Worker

Our CreateOrder function has finished. The user is happy, the order is saved, and the event is sitting safely in the outbox table in Postgres.

But the warehouse doesn't know about it yet.

We need a background process, a "Courier", that constantly checks the mailbox and delivers the messages to Redis.

Scaling the Worker

If we only run one instance of your backend, a simple SELECT * FROM outbox WHERE status = 'PENDING' works fine.

But in production, you likely run 5 or 10 replicas of your service. If they all query the database at the same time, they will all grab the same events and try to publish them simultaneously. We’ll spam Redis with duplicate messages.

For this, we will be using : FOR UPDATE SKIP LOCKED.
This clause tells Postgres : "Lock these rows for me so no one else can touch them. If a row is already locked by another worker, just skip it and give me the next one.”

Let us implement the worker in Go now :

func StartOutboxWorker(ctx context.Context, db *sql.DB, rdb *redis.Client) {
    ticker := time.NewTicker(500 * time.Millisecond)
    defer ticker.Stop()

    for {
        select {
        case <-ctx.Done():
            return
        case <-ticker.C:
            processBatch(ctx, db, rdb)
        }
    }
}

func processBatch(ctx context.Context, db *sql.DB, rdb *redis.Client) {
    // 1. Start a transaction for the worker
    tx, err := db.BeginTx(ctx, nil)
    if err != nil {
        log.Printf("Worker failed to begin tx: %v", err)
        return
    }
    defer tx.Rollback()

    // 2. Fetch pending events with SKIP LOCKED
    // This allows multiple workers to run in parallel without conflict.
    rows, err := tx.QueryContext(ctx, `
        SELECT id, event_type, payload 
        FROM outbox 
        WHERE status = 'PENDING' 
        ORDER BY created_at ASC 
        LIMIT 10 
        FOR UPDATE SKIP LOCKED
    `)
    if err != nil {
        log.Printf("Worker query failed: %v", err)
        return
    }
    defer rows.Close()

    for rows.Next() {
        var id uuid.UUID
        var eventType string
        var payload []byte

        if err := rows.Scan(&id, &eventType, &payload); err != nil {
            continue
        }

        // 3. Publish to Redis
        // We use the event_type as the channel name (e.g., "order.created")
        err := rdb.Publish(ctx, eventType, payload).Err()
        if err != nil {
            // If Redis is down, we just abort. 
            // The transaction rolls back, and we retry next tick.
            log.Printf("Failed to publish event %s: %v", id, err)
            return 
        }

        // 4. Mark as Processed (or Delete)
        // Since we published successfully, we update the status.
        _, err = tx.ExecContext(ctx, 
            "UPDATE outbox SET status = 'PROCESSED' WHERE id = $1", id)
        if err != nil {
            return
        }
    }

    // 5. Commit the batch
    if err := tx.Commit(); err != nil {
        log.Printf("Worker failed to commit: %v", err)
    }
}

Why this is robust:

At-Least-Once Delivery: If Redis fails (network blip), the rdb.Publish returns an error. We return immediately, the SQL transaction rolls back, and the event stays "PENDING". We will try again in 500ms.
Concurrency Safe: Thanks to SKIP LOCKED, you can scale this worker to 100 instances, and they will never process the same event twice.

You can check out the redis consumer in the codebase.

Conclusion

And actually that is it.

By combining the Transactional Outbox Pattern with Idempotency, we’ve brought that simplicity of a monolithic function back.

We no longer have to worry about ghost orders or lost emails. If our database commits, our event is guaranteed to publish. If our server crashes, our Redis lock ensures we pick up right where we left off.

Reliability isn't about hoping things don't break; it's about building systems that handle the breakages gracefully.

Rust Iterators : DeSugarified

Yashaswi Kumar Mishra — Thu, 31 Jul 2025 05:46:01 +0000

No beating around the bush, we will be jumping straight in it.

The Iterator Trait

In Rust, an iterator is basically a type that knows how to yield items one at a time. We will take the example of a vending machine and its items (like snacks) to understand how it works. So an iterator is like a vending machine which yields a packet of snacks one at a time.

So everytime, .next() is hit on an iterator, the next item is given if available else None is returned (snacks are exhausted). The snack is given out till the machine is empty.

So something which implements the Iterator trait is called an iterator.

pub trait Iterator {
    type Item;

    fn next(&mut self) -> Option<Self::Item>;

    //there are many other methods implemented here
}

type Item defines what the iterator will yield (e.g. i32, String, etc.)

To understand it better, let us build an iterator from scratch

struct Counter {
    current: u32,
}

This counter should count from 0 to 4 and then return None after that. The implementation looks very simple.

impl Iterator for Counter {
    type Item = u32;

    fn next(&mut self) -> Option<Self::Item> {
        if self.current < 5 {
            let result = self.current;
            self.current += 1;
            Some(result)
        } else {
            None
        }
    }
}

The Item which will be yielded by our Counter iterator will be of type u32. We check if the current count is less than 5 and then return count + 1 else we return None. We can run a simple for loop to test our implementation.

for x in Counter { current: 0 } {
    println!("{x}");
}

And we get numbers from 0 to 4 printed nicely (not yet as .into_iter is not here). This looks neat but the iterator trait we implemented on our struct gets invoked implicitly. It desugars this into raw trait calls. So what happens under the hood?

let mut iter = Counter { current: 0 }.into_iter(); // uses IntoIterator (we'll discuss it later)
while let Some(x) = iter.next() {
    println!("{x}");
}

Rust doesn’t assume you're passing an iterator. You could be passing a Vec, &Vec, or some other iterable. So it calls .into_iter() to get an actual Iterator. Then we keep calling .next() until it returns None.

So how does the for loop work even when we only implemented Iterator on Counter?

IntoIterator trait

That’s where the IntoIterator trait comes in. When you write a for loop, Rust calls .into_iter() under the hood to turn your type into something it can loop over.

Let’s see what this trait looks like:

Let’s look into IntoIterator trait and see how into_iter() works. This trait converts a type into something that can be iterated over.

pub trait IntoIterator{
    type Item;
    type IntoIter: Iterator<Item = Self::Item>;

    fn into_iter(self) -> Self::IntoIter;
}

The syntax looks a bit painful but let’s go through what it translates into.

This trait converts a type into something that can be iterated over
It owns or borrows the data and returns an Iterator
It’s used by for loops and by calling .into_iter() directly

This line type IntoIter: Iterator<Item = Self::Item>; Hey, the associated type IntoIter must implement the Iterator trait… and the Item of that Iterator must be the same type as Self::Item.

So, if our type implements IntoIterator, our type is “for-loop friendly”.

Let’s implement IntoIterator for our Counter struct.

impl IntoIterator for Counter {
    type Item = u32;
    type IntoIter = Counter; //must implement Iterator with Item type same as Self::Item

    fn into_iter(self) -> Self::IntoIter {
        self
    }
}

So we are telling the compiler compiler how to .into_iter() a Counter and that will return … another Counter.

Let’s Dissect it a bit

type Item = u32;

It is same as the Item from our Iterator impl. It ensures consistency:

Whatever our IntoIter returns must match the item type the for loop expects.

type IntoIter = Counter;

This tells Rust that the actual iterator you get from calling .into_iter() on a Counter is... itself!

Yes, the struct is its own iterator. That’s perfectly legal and common.

fn into_iter(self) -> Self::IntoIter { self }

We're saying:

"Hey Rust, just take self and treat it as an iterator. No conversion and no fuss."

Why does this work? Because we’ve already implemented Iterator for Counter. So once the for-loop gets the value via .into_iter(), it starts calling .next() on it.

It kind of makes sense, but these things are roaming in the head in a very jumbled fashion. Let’s summarize the flow.

We needed IntoIter trait for the Counter struct to convert itself into something that implements Iterator. The .into_iter() method on Counter returns an iterator type (Counter itself) that actually implements Iterator and defines how items are yielded using .next().

This thing should now fit into the slot perfectly. Let’s take this diagram as an aid to put out the whole iterator flow at a place.

Iterator Ergonomics

We have already built a struct, implemented Iterator, then wrapped it with IntoIterator. That’s the custom path. Now let’s actually understand how Rust’s built-in types like Vec<T> give you super clean iterator ergonomics using three flavors of IntoIterator.

There are 3 flavors of iteration. Let’s look at a canonical example.

let v = vec![1, 2, 3];

for val in v {         // v.into_iter()       -> ownership
for val in &v {        // (&v).into_iter()    -> shared borrow
for val in &mut v {    // (&mut v).into_iter() -> mutable borrow

Behind the scenes, something like this is happening.

IntoIterator::into_iter(v);
IntoIterator::into_iter(&v);
IntoIterator::into_iter(&mut v);
//all are called inside the while loop like before

Which means that the Vec<T> implements IntoIterator for all 3 forms:

impl<T> IntoIterator for Vec<T>             // owned
impl<'a, T> IntoIterator for &'a Vec<T>     // shared ref
impl<'a, T> IntoIterator for &'a mut Vec<T> // mutable ref

.iter() and .iter_mut()

Just like IntoIterator for Counter gives up .into_iter() style

IntoIterator for &Counter gives us .iter() style
IntoIterator for &mut Counter gives us .iter_mut() style

So, it is very simple to implement the IntoIterator trait for these types as well. By doing so, you're giving users the full ergonomic experience of iteration, letting them decide whether to consume, borrow, or mutate, just like they do with built-in collections.

Let us impl the trait for &Counter and &mut Counter then. (we’ll breeze through this).

struct CounterIter<'a> {
    counter: &'a Counter,
    pos: u32,
}

struct CounterIterMut<'a> {
    counter: &'a mut Counter,
    pos: u32,
}

These track the position and hold a ref/mut ref to the original Counter.

Let us implement Iterator for both of these structs.

impl<'a> Iterator for CounterIter<'a> {
    type Item = u32;

    fn next(&mut self) -> Option<Self::Item> {
        if self.pos < 5 {
            let res = self.pos;
            self.pos += 1;
            Some(res)
        } else {
            None
        }
    }
}

impl<'a> Iterator for CounterIterMut<'a> {
    type Item = u32;

    fn next(&mut self) -> Option<Self::Item> {
        if self.pos < 5 {
            let res = self.pos;
            self.pos += 1;
            Some(res)
        } else {
            None
        }
    }
}

Note: We’re not using the counter field for logic (you could, if you stored state inside it). This just makes iter() and iter_mut() usable ergonomically.

Now the main part, the IntoIter trait.

impl<'a> IntoIterator for &'a Counter {
    type Item = u32;
    type IntoIter = CounterIter<'a>;

    fn into_iter(self) -> Self::IntoIter {
        CounterIter {
            counter: self,
            pos: 0,
        }
    }
}

impl<'a> IntoIterator for &'a mut Counter {
    type Item = u32;
    type IntoIter = CounterIterMut<'a>;

    fn into_iter(self) -> Self::IntoIter {
        CounterIterMut {
            counter: self,
            pos: 0,
        }
    }
}

That’s it.

If you noticed that the CounterIter and CounterIterMut implementations of Iterator are practically identical and that’s not a coincidence. Both structs:

Track an internal pos field
Don’t even use counter field in .next()
Yield u32 values from 0 to 4, incrementing pos each time

The only difference? The reference type (&Counter vs &mut Counter), which we don’t actually mutate in .next()!

So why not DRY it ?

We could absolutely avoid duplication by using a generic internal iterator. This would be overkill for now, but it’s something the Rust standard library often does , use generic building blocks and type aliases to abstract out common behavior.

And that’s a wrap on iterators , from .next() to .iter_mut() and everything in between. We have seen how Rust's iteration model is both low-level and ergonomic, explicit yet elegant. Whether you're looping, lazily transforming, or crafting your own custom iterator, now you really know what’s going on under the hood.

Until next time!

The Fault-Tolerant Job Queue You’ll Actually Understand – in Go

Yashaswi Kumar Mishra — Fri, 18 Jul 2025 07:41:16 +0000

Async job queues are the quiet workhorses behind a lot of the stuff you take for granted, sending emails, processing payments, resizing images, crunching analytics. Basically, any task that doesn’t need to happen right now but still needs to get done, reliably. A good job queue takes these tasks, runs them in the background, retries if something fails, and makes sure things don’t fall through the cracks.

You need one when your app starts to grow up. Maybe you're dealing with slow external APIs. Maybe your users are uploading giant files. Or maybe your backend is doing a lot of heavy lifting, and you just want to keep your main flow snappy. Async queues are built for this. They give you timeouts, retries, and even distributed execution if you're scaling across multiple workers.

So why not just use Celery, Sidekiq, or something off the shelf? You totally could. But most of those systems come with big batteries included, Redis, RabbitMQ, a whole runtime, sometimes even their own DSL. That’s great for big teams and big infra. But if you're trying to deeply understand how things actually work or just want to build something lean, it's a lot of overhead.

That's why this one’s in Go. Concurrency in Go is smooth. Goroutines and channels are basically made for this kind of system. No runtime bloat, no Java-style ceremony. Just straight-up efficient task handling with great performance, clarity, and simplicity.

In this blog, we’re going to build our own async job queue system from scratch. Real code. Real retries. Real persistence. Let’s get to work.
Also here is the github repo : https://github.com/pixperk/asynq

I am writing this up alongside building the whole thing, so I will keep it as long as it takes me . Okay, I cracked my fingers and now let’s write some code and think a little.
My initial plan is to structure a Job Queue and make every popped job go through a worker which will process the job. Let’s define the job interface first and then we will ponder about how we can take it further.

So the idea of the interface is that a “job” can be anything that can be executed, so every job must have an Execute() method.

type Job interface {
    Execute() error
}

Our queue can be a channel for receiving these jobs. A simple struct like the following will do for now.

type JobQueue struct {
    jobs chan Job
}

The plan is to do something like Submit(job) which will send the job into the channel. And JobQueue struct will hold the buffered channel and the worker.

Now, we will need a worker. It can be a simple go routine, which receives jobs from the channel and calls Execute(). It is a go routine so that it can run in the background, pick up the jobs and process them.

go func(){
    for job := range q.jobs {
        job.Execute()
    }
}()

Now we want this to be triggered as soon as our queue is initialized. So let’s make a constructor for our JobQueue struct and put this go routine in there.

func NewJobQueue(buffer int) *JobQueue {
    q := &JobQueue{
        jobs: make(chan Job, buffer),
    }

    go func() {
        for job := range q.jobs {
            job.Execute()
        }
    }()

    return q
}

And then write a small function for the JobQueue struct to dump jobs in the queue (channel).

func (q *JobQueue) Submit(job Job) {
    q.jobs <- job
}

So, this is our very first idea of the Job Queue. Let’s get going and create a new job, let’s call it SampleJob. We will make the struct and implement the Execute method on it.

type SampleJob struct {
    ID int
}
func (s *SampleJob) Execute() error {
    startTime := time.Now()
    fmt.Printf("🔧 Job %d started at %s\n", s.ID, startTime.Format("15:04:05.000"))

    time.Sleep(2 * time.Second) // simulate slowness

    endTime := time.Now()
    duration := endTime.Sub(startTime)
    fmt.Printf("✅ Job %d finished at %s (duration: %v)\n", s.ID, endTime.Format("15:04:05.000"), duration)
    return nil
}

I have modelled each job to take 2s of time. Let’s write our main function and see how our job queue performs.

func main() {
    q := job.NewJobQueue(10)

    start := time.Now()

    for i := 1; i <= 5; i++ {
        q.Submit(&job.SampleJob{ID: i})
    }

    // Wait for all jobs to complete (5 jobs * 2s = ~10s total)
    time.Sleep(11 * time.Second)

    fmt.Printf("\nTotal execution time: %.2fs\n", time.Since(start).Seconds())
}

As we have only 1 worker now, all jobs will be queued behind each other. Let’s see the output logs.

Every job is running synchronously and the total time is sum of all job times. We don’t want this. If one of the jobs break, everything else will be bottlenecked. So we have got only one worker, and we can’t scale. Apart from that, we have :

No status or tracking
No retry or error handling
No job types
Crashes = all jobs die

Let’s spin up multiple workers now. We can change our JobQueue struct and introduce workers and waitgroup fields.

type JobQueue struct {
    jobs    chan Job
    workers int
    wg      sync.WaitGroup
}

sync.WaitGroup is like a counter that tracks a group of goroutines and block execution until all of them signal completion.

Let’s break down its core API:

wg.Add(delta int)

This increments (or decrements) the counter by delta.

Usually delta is +1 when you’re launching a new task.

This tells the WaitGroup:

"I now have 1 more task I’m waiting on."

You can call Add(n) before launching n goroutines.

wg.Done()

This is shorthand for wg.Add(-1).

It tells the WaitGroup:

"This specific task has finished its job."

Every Done() call must match an Add(1). If it doesn't, the counter never reaches zero, and Wait() blocks forever (deadlock).

wg.Wait()

Blocks the calling goroutine (usually main() or a coordinator)

"Don't move forward until all tracked tasks have called Done() and the counter reaches zero."

With all this in our head, let us modify some job methods.
Let’s modify the constructor first.

func NewJobQueue(buffer int, workers int) *JobQueue {
    q := &JobQueue{
        jobs:    make(chan Job, buffer),
        workers: workers,
    }

    for i := 0; i < q.workers; i++ {
        go func(workerID int) {
            for job := range q.jobs {
                fmt.Printf("👷 Worker %d picked a job\n", workerID)
                job.Execute()
                q.wg.Done()
            }
        }(i + 1)
    }

    return q
}

jobs channel is now a shared queue across multiple goroutines and we spawn N workers while initializing the new queue. We let sync.WaitGroup track the job completion.

Let’s go ahead and modify our Submit method to add the job to the waitgroup.

func (q *JobQueue) Submit(job Job) {
    q.wg.Add(1)
    q.jobs <- job
}

We define two more methods on our JobQueue struct : Wait and Shutdown .

Wait() blocks until all jobs are done and Shutdown() gracefully closes the job channel.

func (q *JobQueue) Wait() {
    q.wg.Wait()
}

func (q *JobQueue) Shutdown() {
    close(q.jobs)
}

Now let’s modify our main function to comply with the changes we just made.

func main() {
    start := time.Now()

    buffer, workers := 10, 3

    q := job.NewJobQueue(buffer, workers)

    for i := 1; i <= 5; i++ {
        q.Submit(&job.SampleJob{ID: i})
    }

    q.Wait()
    q.Shutdown()

    fmt.Printf("\nTotal execution time: %.2fs\n", time.Since(start).Seconds())
}

We removed time.Sleep , thanks to sync.WaitGroup . It k*nows exactly* when all jobs are done and only then unblocks the main goroutine.
Let us run the main and check for the output.

As we have 3 workers, 3 jobs can run parallelly, jobs 3,2 and 1 run parallelly and take 2 seconds to complete in total. After this, workers pick up the remaining jobs and take 2 seconds to process. The main thread waits precisely until the last job is finished, then exits.

This works, but we still have no clue, what has to be done if one of our jobs fail. We have minimal to no error handling or retry mechanism. Let’s improve this.

Evidently, we now want to track our jobs. Let’s make a tracked job structure which will contain metadata like status, retries, etc. We will wrap our Jobinterface with it later which currently just does Execute() error .

type TrackedJob struct {
    Job        Job //wrap job interface
    RetryCount int
    MaxRetries int
    Status     JobStatus
    LastError  error
    mu         sync.Mutex
}

The mutex mu ensures thread-safe updates to the job state when workers access it concurrently.

Also, we need an enum like structure for job status. Here is how we write it.

type JobStatus string

const (
    StatusPending JobStatus = "PENDING"
    StatusRunning JobStatus = "RUNNING"
    StatusFailed  JobStatus = "FAILED"
    StatusSuccess JobStatus = "SUCCESS"
)

Now we have to wrap our Execute() with retry logic and status tracking
Our main aim is to track the status here, and make sure only one thread can write status so we wrap it in mutex. If we encounter errors, we will retry till max retry count and we keep updating the last error, status and retry count. We will wrap all the writes to the struct with the mutex.

func (t *TrackedJob) ExecuteWithRetry() {
    t.mu.Lock()
    t.Status = StatusRunning
    t.mu.Unlock()

    for attempt := 1; attempt <= t.MaxRetries+1; attempt++ {
        err := t.Job.Execute()
        if err != nil {
            ...
            }
        } else {
            t.mu.Lock()
            t.Status = StatusSuccess
            t.mu.Unlock()

            fmt.Printf("✅ Job succeeded on attempt %d\n", attempt)
            return
        }
    }
}

We put the status as StatusRunning at beginning and then if the job executes successfully, we update it to StatusSuccessful, we run this execution cycle till MaxRetriesand if we are successful, we break out. What to do if the execution gives an error? We retry and continue the loop.

        if err != nil {
            //track the job
            t.mu.Lock()
            t.LastError = err
            t.Status = StatusFailed
            t.RetryCount = attempt
            t.mu.Unlock()

            fmt.Printf("Attempt %d failed: %v\n", attempt, err)
            if attempt <= t.MaxRetries {
                fmt.Printf("Retrying job (attempt %d)...\n", attempt+1)
                time.Sleep(500 * time.Millisecond) // backoff delay
                continue
            }

Now this is what we have as our retry engine. And we have some sort of tracking mechanism as well.
Let us modify our JobQueueto work with TrackedJob

type JobQueue struct {
    jobs    chan *TrackedJob
    workers int
    wg      sync.WaitGroup
}

Now the jobs channel is of type TrackedJob (earlier it was the Jobinterface)

Our constructor changes as well to comply with this.

func NewJobQueue(buffer int, workers int) *JobQueue {
    q := &JobQueue{
        jobs:    make(chan *TrackedJob, buffer),
        workers: workers,
    }

    for i := 0; i < q.workers; i++ {
        go func(workerID int) {
            for job := range q.jobs {
                fmt.Printf("👷 Worker %d picked a job\n", workerID)
                job.ExecuteWithRetry()
                q.wg.Done()
            }
        }(i + 1)
    }

    return q
}

We need to change our Submit implementation as well. It takes maxRetries now, and makes a TrackedJob struct with StatusPending

func (q *JobQueue) Submit(job Job, maxRetries int) {
    q.wg.Add(1)
    tracked := &TrackedJob{
        Job:        job,
        MaxRetries: maxRetries,
        Status:     StatusPending,
    }
    q.jobs <- tracked
}

Now, we can also define a status function for our TrackedJobto poll job status later

func (t *TrackedJob) StatusString() string {
    t.mu.Lock()
    defer t.mu.Unlock()
    return string(t.Status)
}

Let’s define an ErroneousJob to test our implementation.

type ErroneousJob struct {
    ID       int
    Attempts int
}

func (f *ErroneousJob) Execute() error {
    f.Attempts++
    fmt.Printf("Executing ErroneousJob %d (attempt %d)\n", f.ID, f.Attempts)

    if f.Attempts < 3 {
        return fmt.Errorf("simulated failure on attempt %d", f.Attempts)
    }

    time.Sleep(1 * time.Second)
    fmt.Printf("🎉 ErroneousJob %d succeeded\n", f.ID)
    return nil
}

If attempts are less than 3, it will always error out, perfect for testing. Our main function should look like the following now.


func main() {
    start := time.Now()

    buffer, workers := 10, 3

    q := job.NewJobQueue(buffer, workers)

    maxRetries := 5

    q.Submit(&job.ErroneousJob{ID: 2}, maxRetries)

    q.Wait()
    q.Shutdown()

    fmt.Printf("\nTotal execution time: %.2fs\n", time.Since(start).Seconds())
}

The job succeeds on 3rd retry, so the retry system works fine. Let’s spin up 5 such jobs and see how it works. The modified main looks like this :

It works perfectly.

So, now we have got a retry mechanism and our jobs are tracked, all this along with concurrent job execution. But the Job Metadata is not persistent, nor do we have Job IDs & external status querying.
We also do not have proper retry strategy tuning (backoff/jitter).
Let’s look to make it better. Our goal is to devise a global tracking layer for external status querying.
We will start by introducing a JobID in our TrackedJob struct. We will generate this id in Submit using a simple UUID generator.

type TrackedJob struct {
    JobID      string
    Job        Job //wrap job interface
    RetryCount int
    MaxRetries int
    Status     JobStatus
    LastError  error
    mu         sync.Mutex
}

Now we can write our global tracker. It will be a struct with a read-write mutex (exclusive locking for writes and shared access for reads) and an in-memory hashmap (for now) mapping each job to its metadata.

type JobTracker struct {
    mu    sync.RWMutex
    store map[string]*JobMetadata
}

The JobMetadata struct will be like this :

type JobMetadata struct {
    JobID      string
    Status     JobStatus
    RetryCount int
    LastError  error
    CreatedAt  time.Time
    UpdatedAt  time.Time
}

Let’s build a constructor for JobTracker

func NewJobTracker() *JobTracker {
    return &JobTracker{
        store: make(map[string]*JobMetadata),
    }
}

Now we will implement methods on this JobTracker to register a job in the store and update the store.

func (jt *JobTracker) Register(jobID string) {
    jt.mu.Lock()
    defer jt.mu.Unlock()
    jt.store[jobID] = &JobMetadata{
        JobID:     jobID,
        Status:    StatusPending,
        CreatedAt: time.Now(),
        UpdatedAt: time.Now(),
    }
}

func (jt *JobTracker) Update(jobID string, status JobStatus, retry int, err error) {
    jt.mu.Lock()
    defer jt.mu.Unlock()
    if meta, exists := jt.store[jobID]; exists {
        meta.Status = status
        meta.RetryCount = retry
        meta.LastError = err
        meta.UpdatedAt = time.Now()
    }
}

We make use of mutex, to ensure that the writes are atomic.

We write a small method to print the status of all jobs at a time.

func (jt *JobTracker) PrintAll() {
    jt.mu.RLock()
    defer jt.mu.RUnlock()
    fmt.Println("\n Job Tracker Status:")
    for _, meta := range jt.store {
        fmt.Printf("- [%s] JobID: %s | Retries: %d | Error: %v | Updated: %s\n",
            meta.Status, meta.JobID, meta.RetryCount, meta.LastError, meta.UpdatedAt.Format("15:04:05"))
    }
}

Now we can easily plug this tracker into out main JobQueue

type JobQueue struct {
    jobs    chan *TrackedJob
    workers int
    wg      sync.WaitGroup
    tracker *JobTracker
}

func NewJobQueue(buffer int, workers int) *JobQueue {
    q := &JobQueue{
        jobs:    make(chan *TrackedJob, buffer),
        workers: workers,
        tracker: NewJobTracker(),
    }
    //...existing background workers
    }

Let us write a simple util to generate jobIDs.

func GenerateJobID() string {
    b := make([]byte, 4)
    _, err := rand.Read(b)
    if err != nil {
        return "job-xxxx"
    }
    return fmt.Sprintf("job-%x", b)
}

Now let us update Submit() to:

generate a JobID
assign to TrackedJob
register with tracker

func (q *JobQueue) Submit(job Job, maxRetries int) {
    q.wg.Add(1)

    jobID := GenerateJobID()

    tracked := &TrackedJob{
        JobID:      jobID, //Add
        Job:        job,
        MaxRetries: maxRetries,
        Status:     StatusPending,
    }

    q.tracker.Register(jobID)

    q.jobs <- tracked
}

Now the final thing left is to update Tracker inside ExecuteWithRetry

For that we will have to inject this tracker into the TrackedJob struct

type TrackedJob struct {
    JobID      string
    Job        Job 
    RetryCount int
    MaxRetries int
    Status     JobStatus
    LastError  error
    Tracker    *JobTracker //Add
    mu         sync.Mutex
}

And in Submit(), we can do this :

func (q *JobQueue) Submit(job Job, maxRetries int) {
    q.wg.Add(1)

    jobID := GenerateJobID()

    tracked := &TrackedJob{
        JobID:      jobID,
        Job:        job,
        MaxRetries: maxRetries,
        Status:     StatusPending,
        Tracker:    q.tracker, //Add
    }

    q.tracker.Register(jobID)

    q.jobs <- tracked
}

Now in the ExecuteWithRetry method, we can update our Tracker ledger.

We update the Tracker at all three places where we initially wrote into the TrackedJob struct

  t.mu.Lock()
    t.Status = StatusRunning
    //Update ledger
    t.Tracker.Update(t.JobID, StatusRunning, 0, nil)
    t.mu.Unlock()

 t.mu.Lock()
 t.mu.Lock()
 t.LastError = err
 t.Status = StatusFailed
 t.RetryCount = attempt
 //Update ledger
    t.Tracker.Update(t.JobID, StatusFailed, attempt, err)
    t.mu.Unlock()

            t.mu.Lock()
            //Update ledger
            t.Tracker.Update(t.JobID, StatusSuccess, attempt, nil)
            t.Status = StatusSuccess
            t.mu.Unlock()

Now our jobs are being tracked, let’s print the store status in main. First let’s expose the PrintAll function through JobQueue.

func (q *JobQueue) StoreStatus() {
    q.tracker.PrintAll()
}

And we can add this line in main.

q.StoreStatus()

This is the output we get (for the StoreStatus)

Yes, it does it correctly. Our jobs are being tracked correctly. We can add persistence to our jobs later.
The big question is why are there two separate things : JobTracker and TrackedJob ?
TrackedJob is responsible for the execution and retry logic of a single job ,it handles its own lifecycle. JobTracker is a decoupled, centralized ledger used purely for auditing, visibility, and external monitoring.

We keep them separate to ensure a clear separation of concerns between job execution and job tracking.

****With all that in place, let us look to improve our retry system a little more. Currently, we are not delaying the wait between two retries, but what if two concurrent jobs fail at the same time? They will look to retry at the same time putting more load at our resources. So we will introduce two things in our retry mechanism.

Exponential Backoff : We want to increase the wait time between retries exponentially. For ex : 500ms for 1st retry, 1s for another, 2s for third, 4s for fourth and so on. Why but? It gives our downstream systems time to recover, reduces request floods and prevents concurrent retries from multiple jobs.
Jitter : We will aim to add randomness to the retry delay. Instead of retrying all clients at exactly 2s, you retry somewhere in 1.5s – 2.5s. Why?
Prevents the "thundering herd problem" (thousands of jobs retrying at the same time)
Helps stagger retries in distributed systems.
Reduces contention on databases, queues, APIs, etc.

Let us make our Backoff Strategy plug-n-play, we code this up in the retry package, then just plug it into our ExecuteWithRetry() .

BackOffStrategy will be an interface with a method called NextDelay which will return the delay duration.

type BackoffStrategy interface {
    NextDelay(attempt int) time.Duration
}

Now, our strategy is Exponential Backoff with Jitter, let us structure ExponentialBackoff. We will need BaseDelay, MaxDelay and a bool to decide if want to add jitter or not.

type ExponentialBackoff struct {
    BaseDelay time.Duration //initial delay
    MaxDelay  time.Duration
    Jitter    bool
}

Now, it must implement NextDelay() .

func (b ExponentialBackoff) NextDelay(attempt int) time.Duration {
    exp := float64(b.BaseDelay) * math.Pow(2, float64(attempt-1))
    delay := time.Duration(exp)
    return delay
}

It can be this simple, but what if the delay is greater than the maximum delay time, then we cap the delay at MaxDelay

exp := float64(b.BaseDelay) * math.Pow(2, float64(attempt-1))
    delay := min(time.Duration(exp), b.MaxDelay)
    return delay

And let’s add jitter if it is enable, let’s go for 30% jitter. We could have also dynamically taken the percentage, but 30% sounds good.

if b.Jitter {
        jitter := rand.Float64()*float64(delay) * 0.3
        delay += time.Duration(jitter)
    }

It works like this :
(Example)

Original delay: 4 seconds
Random factor: 0.8 (from rand.Float64())
Jitter: 0.8 * 4s * 0.3 = 0.96 seconds
Final delay: 4s + 0.96s = 4.96 seconds

Instead of all clients retrying at exactly 4 seconds, they'll now retry at random times between 4.0s and 5.2s (4s + 0% to 30% jitter).

This provides a 0% to +30% variance above the base exponential backoff delay.
So, our final implementation will look this this :

func (b ExponentialBackoff) NextDelay(attempt int) time.Duration {
    exp := float64(b.BaseDelay) * math.Pow(2, float64(attempt-1))
    delay := min(time.Duration(exp), b.MaxDelay)

    if b.Jitter {
        jitter := rand.Float64() * float64(delay) * 0.3
        delay += time.Duration(jitter)
    }

    return delay
}

Now, our plug-n-play Backoff strategy is ready. Time to plug it into our TrackedJob struct.

type TrackedJob struct {
    JobID      string
    Job        Job //wrap job interface
    RetryCount int
    MaxRetries int
    Status     JobStatus
    LastError  error
    Tracker    *JobTracker 
    Backoff    retry.BackoffStrategy //Add
    mu         sync.Mutex
}

And in our ExecuteWithRetry(), will make the following changes

if attempt <= t.MaxRetries {
                delay := t.Backoff.NextDelay(attempt) //Add
                fmt.Printf("Retrying job (attempt %d)...\n", attempt+1)
                time.Sleep(delay) // backoff delay
                continue

Now whenever, a TrackedJob is created, we can pass our ExponentialBackoff in the Backoff field. We do this in our Submit() function.

func (q *JobQueue) Submit(job Job, maxRetries int) {
    q.wg.Add(1)

    jobID := GenerateJobID()

    tracked := &TrackedJob{
        JobID:      jobID,
        Job:        job,
        MaxRetries: maxRetries,
        Status:     StatusPending,
        Tracker:    q.tracker,
        Backoff: retry.ExponentialBackoff{ //Add
            BaseDelay: 500 * time.Millisecond,
            MaxDelay:  5 * time.Second,
            Jitter:    true,
        },
    }

    q.tracker.Register(jobID)

    q.jobs <- tracked
}

Now, our retries are resilient with beautiful exponential backoff with jitter implementation.

Now, let us make our jobs “context” aware. If a job hangs (infinite loopor external call, etc.), we will literally need a kill switch. With context.WithTimeout, we can force jobs to bail if they exceed their allowed execution window.
Also if our entire worker pool shuts down (e.g. SIGINT), we should be able to cancel the root context, and all in-progress jobs will also gracefully terminate. This is what structured concurrency is, our cancellation chains are clean and scoped.
Further, we need to respect deadlines from clients.
When exposing an HTTP API later, we can pass the HTTP request’s context into jobs. If the client bounces, we stop the job.
Let’s start by allowing the Execute method inside our main Job interface take a context as param.

type Job interface {
    Execute(ctx context.Context) error
}

Now. in our ExecuteWithRetry(), let’s introduce context.WithTimeout with background context. We will eventually refactor it, but we just want it to be up and running for now.

ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
defer cancel() // cleanup context even if job returns early

err := t.Job.Execute(ctx)

We will also log if the error is “timed out”

if errors.Is(err, context.DeadlineExceeded) {
                fmt.Printf("Job attempt %d timed out\n", attempt)
            } else {
                fmt.Printf("Attempt %d failed: %v\n", attempt, err)
            }

Let’s check this by writing a SleepyJob.

type SleepyJob struct {
    Duration time.Duration
}

func (j *SleepyJob) Execute(ctx context.Context) error {
    select {
    case <-time.After(j.Duration):
        fmt.Println("😴 Job finished sleeping")
        return nil
    case <-ctx.Done():
        return ctx.Err() // return context.Canceled or context.DeadlineExceeded
    }
}

And we update our main function. (Keeping maxRetries = 0)

func main() {
    start := time.Now()

    buffer, workers := 10, 3

    q := job.NewJobQueue(buffer, workers)

    q.Submit(&job.SleepyJob{Duration: 1 * time.Second}, 0)
    q.Submit(&job.SleepyJob{Duration: 6 * time.Second}, 0)

    q.Wait()
    q.Shutdown()

    q.StoreStatus()
    fmt.Printf("\nTotal execution time: %.2fs\n", time.Since(start).Seconds())
}

We get this

So, this is works just fine. I just noticed that we are getting Retries : 1 in our final logs, but that is just the attempt count. We can rename that, but for now let us focus on our functionalities.

Now let us make the timeouts dynamic, let us first inject a timeout field in our TrackedJob struct.

type TrackedJob struct {
    JobID      string
    Job        Job //wrap job interface
    RetryCount int
    MaxRetries int
    Status     JobStatus
    LastError  error
    Tracker    *JobTracker
    Backoff    retry.BackoffStrategy
    Timeout    time.Duration //Add
    mu         sync.Mutex
}

With this our ExecuteWithRetry() is ready to take a context from outside. We change the function signature to ExecuteWithRetry(ctx Context.context)

Change the previous context implementation to below :

// Derive a child context with timeout for this attempt
        attemptTimeout := t.Timeout
        if attemptTimeout == 0 {
            attemptTimeout = 5 * time.Second // default fallback
        }
        attemptCtx, cancel := context.WithTimeout(ctx, attemptTimeout)
        defer cancel()
        err := t.Job.Execute(attemptCtx)

Now let’s modify our NewJobQueue to take in a context so that it can pass it down to ExecuteWithRetry .

func NewJobQueue(ctx context.Context, buffer int, workers int) *JobQueue

Now let’s modify the internal go routine for graceful context based shutdown.

    go func(workerID int) {
            for {
                select {
                case <-ctx.Done():
                    fmt.Printf("Worker %d shutting down: %v\n", workerID, ctx.Err())
                    return
                case job, ok := <-q.jobs:
                    if !ok {
                        fmt.Printf(" Job channel closed. Worker %d exiting\n", workerID)
                        return
                    }
                    fmt.Printf("👷 Worker %d picked a job\n", workerID)
                    job.ExecuteWithRetry(ctx)
                    q.wg.Done()
                }
            }
        }(i + 1)

We use a select statement here to listen for signals from the channel.

case <-ctx.Done():
                    fmt.Printf("Worker %d shutting down: %v\n", workerID, ctx.Err())
                    return

When context is cancelled (timeout, manual cancellation, etc.), worker immediately stops and exits gracefully.

case job, ok := <-q.jobs:
                    if !ok {
                        fmt.Printf(" Job channel closed. Worker %d exiting\n", workerID)
                        return
                    }
                    fmt.Printf("👷 Worker %d picked a job\n", workerID)
                    job.ExecuteWithRetry(ctx)
                    q.wg.Done()
                }

Worker detects channel closure and exits which prevents panic from reading closed channel. If the channel is open, job is available and context is still active, and the default processing and execution can be done.
Let’s refactor this whole goroutine function and put it somewhere, and just call it from here.

func ProcessJob(ctx context.Context, q *JobQueue, workerID int)

func NewJobQueue(ctx context.Context, buffer int, workers int) *JobQueue {
    q := &JobQueue{
        jobs:    make(chan *TrackedJob, buffer),
        workers: workers,
        tracker: NewJobTracker(),
    }

    for i := 0; i < q.workers; i++ {
        go ProcessJob(ctx, q, i+1)
    }

    return q
}

Now let us modify our main to pass the root context.

func main() {
    start := time.Now()

    buffer, workers := 10, 3

    ctx, cancel := context.WithCancel(context.Background())
    defer cancel()

    q := job.NewJobQueue(ctx, buffer, workers)

    q.Submit(&job.SleepyJob{Duration: 1 * time.Second}, 0)
    q.Submit(&job.SleepyJob{Duration: 6 * time.Second}, 0)

    q.Wait()
    q.Shutdown()

    q.StoreStatus()
    fmt.Printf("\nTotal execution time: %.2fs\n", time.Since(start).Seconds())
}

We can also pass timeout in our TrackedJobs , but we would have to hardcode it, we can find a way and pass it from the top. But, let’s not go down that path for now.

We have made a minor mistake in the for loop of our ExecuteWithRetry()

for attempt := 1; attempt <= t.MaxRetries+1; attempt++ {
    attemptCtx, cancel := context.WithTimeout(ctx, attemptTimeout)
    defer cancel() // ❌
    ...
}

What’s wrong?

defer cancel() means “run cancel() when the current function exits”. But this lives inside a loop, so we are stacking cancel() calls, one for every attempt, but it won’t run until after all retries are done (after the whole function ends).
So, the timeouts for earlier attempts aren't released when they’re no longer needed. As a result, we may have context leaks, memory build-up, maybe zombie goroutines if downstream uses that context.

So we will cancel as soon as possible.

    attemptCtx, cancel := context.WithTimeout(ctx, attemptTimeout)
  err := t.Job.Execute(attemptCtx)
  cancel()

This cancels the child context right after the attempt ends, regardless of success or failure.

Right now, our workers are zombies. They do not care if the OS sends a SIGINT (Ctrl+C), or if the app is shutting down. Jobs might be halfway done or still in queue. So in our main.go, we will listen for OS signals (SIGINT, SIGTERM) and cancel the context to stop all workers. This will make our graceful shutdown complete.

ctx, cancel := context.WithCancel(context.Background())

    sigChan := make(chan os.Signal, 1)
    signal.Notify(sigChan, os.Interrupt, syscall.SIGTERM)

    go func() {
        sig := <-sigChan
        fmt.Printf("\nReceived signal: %s. Shutting down gracefully...\n", sig)
        cancel() // Broadcast shutdown to all workers
    }()

We can add this code to our main. A channel called sigChan receives signals from OS and on receiving that, our go routine in background, broadcasts shutdown to all workers. Let’s test it, by pressing ctrl+C, while a job is being executed.

So, it works, the context gets cancelled, get broadcasted to all workers, and we see a graceful shutdown.

Let us now define a struct called SubmitOptions , to cleanly separate job configuration from job execution logic. This allows us to customize every single job.


type SubmitOptions struct {
    MaxRetries int
    Timeout    time.Duration
    Metadata   map[string]string
}

Now we can update our Submit() method to take MaxRetries and Timeout from here. Also let’s add Metadata field to TrackedJobs and send it through Submit .

type TrackedJob struct {
    ...
    Metadata   map[string]string
    ...
}

func (q *JobQueue) Submit(job Job, opts SubmitOptions) {
    q.wg.Add(1)

    jobID := GenerateJobID()

func (q *JobQueue) Submit(job Job, opts SubmitOptions) {
    q.wg.Add(1)

    jobID := GenerateJobID()

    tracked := &TrackedJob{
        ...
        MaxRetries: opts.MaxRetries,
        Timeout:    opts.Timeout,
        Metadata:   opts.Metadata,
        ...
        },
    }

    q.tracker.Register(jobID)

    q.jobs <- tracked
}

Now we can pass SubmitOptions (max retries and timeout) from main.go itself. We’re doing this to cleanly separate config from behavior, unlock per-job flexibility, avoid future pain, and make the code actually readable and scalable.

q.Submit(&job.SleepyJob{Duration: 1 * time.Second}, job.SubmitOptions{
        MaxRetries: 2,
        Timeout:    4 * time.Second,
    })
    q.Submit(&job.SleepyJob{Duration: 6 * time.Second}, job.SubmitOptions{
        MaxRetries: 0,
    })

I think, it is the perfect time now for us to design a persistence layer

Let’s write out JobPersister interface.

type JobPersister interface {
    SaveJob(job *job.TrackedJob) error
    UpdateStatus(jobID string, status job.JobStatus, retry int, err error) error
    LoadPendingJobs() ([]*job.TrackedJob, error)
}

This forces any persister implementation to implement the methods for saving jobs to our store, updating job status and fetching pending jobs from the store.
We will first write out a simple JSON persister to see how the persistence might work. Our core job queue system shouldn’t care how data is saved , only that it is saved.
We will follow our JSON implementation with a SQLite persistence layer for ACID-compliant durability and fast querying/resuming.

With all that in our mind, let’s build the json_persister.go .

Let’s start by writing the JSONPersister struct, it will have a dir field (where the json files will live) and a mutex for exclusive writes.

type JSONPersister struct {
    dir string
    mu  sync.Mutex
}

We will write out a constructor like this.

func NewJSONPersister(dir string) *JSONPersister {
    if err := os.MkdirAll(dir, os.ModePerm); err != nil {
        panic(fmt.Sprintf("Failed to create job directory: %v", err))
    }

    return &JSONPersister{dir: dir}
}

Note:

os.MkdirAll(dir, os.ModePerm): Creates the directory and all parent directories if they don't exist

os.ModePerm: Sets full permissions (0777) - read, write, execute for owner, group, and others

Now we can implement the three methods forced by our JobPersister interface.


func (p *JSONPersister) SaveJob(job *job.TrackedJob) error {
    p.mu.Lock()
    defer p.mu.Unlock()

    filePath := filepath.Join(p.dir, fmt.Sprintf("%s.json", job.JobID))

    data, err := json.MarshalIndent(job, "", "  ")
    if err != nil {
        return fmt.Errorf("failed to marshal job: %w", err)
    }

    if err := os.WriteFile(filePath, data, 0644); err != nil {
        return fmt.Errorf("failed to write job file: %w", err)
    }

    return nil
}

In our SaveJobs method, we use mutex to prevent concurrent write conflicts. We create a dedicated file for our job with {JobID}.json and save human-readable, indented JSON format into that filepath.

Let’s move and spin up our UpdateStatus Method. Here we will just unmarshal the json into our TrackedJob struct, change the status, retry count and last error, then again save it. Below is the implementation.

func (p *JSONPersister) UpdateStatus(jobID string, status job.JobStatus, retry int, lastErr error) error {
    p.mu.Lock()
    defer p.mu.Unlock()
    filePath := filepath.Join(p.dir, fmt.Sprintf("%s.json", jobID))

    data, err := os.ReadFile(filePath)
    if err != nil {
        return fmt.Errorf("couldn't read job file: %w", err)
    }

    var j job.TrackedJob
    if err := json.Unmarshal(data, &j); err != nil {
        return fmt.Errorf("unmarshal error: %w", err)
    }

    j.Status = status
    j.RetryCount = retry
    j.LastError = lastErr

    return p.SaveJob(&j)
}

Let’s load the jobs now. We will read the files from our dir, skipping if the file is a directory itself or if it is corrupted. Unmarshal the data and load it to our trackedJob struct, we will make an array of these trackedJob structs and put the trackedJob in it if the status is pending or failed and return the array.

func (p *JSONPersister) LoadPendingJobs() ([]*job.TrackedJob, error) {
    p.mu.Lock()
    defer p.mu.Unlock()

    p.mu.Lock()
    defer p.mu.Unlock()

    var jobs []*job.TrackedJob

    files, err := os.ReadDir(p.dir)
    if err != nil {
        return nil, fmt.Errorf("read dir error: %w", err)
    }

    for _, file := range files {
        if file.IsDir() {
            continue
        }

        path := filepath.Join(p.dir, file.Name())
        data, err := os.ReadFile(path)
        if err != nil {
            continue // skip corrupted files
        }

        var j job.TrackedJob
        if err := json.Unmarshal(data, &j); err != nil {
            continue
        }

        if j.Status == job.StatusPending || j.Status == job.StatusFailed {
            jobs = append(jobs, &j)
        }

    }

    return jobs, nil
}

With all this, the JSONPersister looks good. Let’s go ahead and plug it with our existing JobQueue.

type JobQueue struct {
    jobs      chan *TrackedJob
    workers   int
    wg        sync.WaitGroup
    tracker   *JobTracker
    persister persister.JobPersister
}

We can plug the persister, but we get an error while importing this persister “import cycle not allowed”. It is because persister is using stuff from our job package like TrackedJobs, JobStatus etc and job package wants to use persister. Let us put TrackedJobs and other stuff in different packages.

Actually, just let me separate all the concerns now itself to avoid any such cyclic dependency in future. You can check out the github repo for reference.

This is how it looks right now. It is a dag now
Job → Persister → TrackedJob
Job → TrackedJob
Job has our jobqueue related stuff and TrackedJob has our main Job interface and everything related to tracked jobs and the global ledger. We must make sure that we never introduce cyclic dependencies between our packages.

Now the error goes away, and we can continue with plugging the persister in our JobQueue.

In our Submit() function for JobQueue, we will persist the job when it is submitted.

//persist the job

    if q.persister != nil {
        if err := q.persister.SaveJob(tracked); err != nil {
            log.Printf("Failed to persist job %s: %v", jobID, err)
        }
    }

Next we will Update the job in the ExecuteWithRetry() function

But it will create a dependency cycle again, so we make a hook for this update.

Add this to the TrackedJob struct.

OnStatusUpdate func(jobID string, status JobStatus, retry int, err error)

To hook up OnStatusUpdate cleanly, pass a callback function from JobQueue to the TrackedJobThat callback will usually be a method on JobQueue that does callsq.persister.UpdateStatus(...)

So let’s write the following method on JobQueue

func (q *JobQueue) handleStatusUpdate(jobID string, status trackedjob.JobStatus, retry int, err error) {
    if q.tracker != nil {
        q.tracker.Update(jobID, status, retry, err)
    }

    if q.persister != nil {
        saveErr := q.persister.UpdateStatus(jobID, status, retry, err)
        if saveErr != nil {
            log.Printf("Persistence update failed for job %s: %v", jobID, saveErr)
        }
    }
}

Now in the Submit(), we can allow the OnStatusUpdate field of TrackedJob to be the above method.

tracked := &trackedjob.TrackedJob{
        ...
        OnStatusUpdate: q.handleStatusUpdate,
        ...
    }

Now we can define an update status function on TrackedJob struct to do all the updates.

func (t *TrackedJob) updateStatus(status JobStatus, attempt int, err error) {
    t.mu.Lock()
    t.Status = status
    t.RetryCount = attempt
    t.LastError = err
    t.mu.Unlock()

    if t.OnStatusUpdate != nil {
        t.OnStatusUpdate(t.JobID, status, attempt, err)
    }
}

And plug in this to our ExecuteWithRetry method.

func (t *TrackedJob) ExecuteWithRetry(ctx context.Context) {
    t.updateStatus(StatusRunning, 0, nil)

    for attempt := 1; attempt <= t.MaxRetries+1; attempt++ {
        // Derive a child context with timeout for this attempt
        attemptTimeout := t.Timeout
        if attemptTimeout == 0 {
            attemptTimeout = 5 * time.Second // default fallback
        }
        attemptCtx, cancel := context.WithTimeout(ctx, attemptTimeout)

        err := t.Job.Execute(attemptCtx)

        cancel()

        if err != nil {
            t.updateStatus(StatusFailed, attempt, err)

            if errors.Is(err, context.DeadlineExceeded) {
                fmt.Printf("Job attempt %d timed out\n", attempt)
            } else {
                fmt.Printf("Attempt %d failed: %v\n", attempt, err)
            }

            fmt.Printf("Attempt %d failed: %v\n", attempt, err)
            if attempt <= t.MaxRetries {
                delay := t.Backoff.NextDelay(attempt)
                fmt.Printf("Retrying job (attempt %d)...\n", attempt+1)
                time.Sleep(delay) // backoff delay
                continue
            } else {
                fmt.Printf("Job permanently failed after %d attempts.\n", attempt-1)
                return
            }
        } else {
            t.updateStatus(StatusSuccess, attempt, nil)
            fmt.Printf("✅ Job succeeded on attempt %d\n", attempt)
            return
        }
    }
}

We can also abstract out other stuff from this function to make it look shorter but that is unnecessary for now. The only thing left now is to plug in LoadPendingJobs . The idea is to load these jobs as soon as we start our thing.

func (q *JobQueue) LoadPersistedJobs() {
    if q.persister == nil {
        return
    }
    jobs, err := q.persister.LoadPendingJobs()
    if err != nil {
        log.Printf("Failed to load persisted jobs: %v", err)
        return
    }
    for _, j := range jobs {
        log.Printf("Re-queuing persisted jobs: %s", j.JobID)
        q.wg.Add(1)
        q.jobs <- j
    }
}

We simply wrap our LoadPendingJobs() into this method on JobQueue, then we feed these jobs to our jobs channel. We will just put this in our main to load the jobs. Also in the constructor for our queue, let us pass in a persister, so that we can initialize the queue with a persister of our choice (JSONpersister for now).

func NewJobQueue(ctx context.Context, buffer int, workers int, persister persister.JobPersister) *JobQueue {
    q := &JobQueue{
        jobs:      make(chan *trackedjob.TrackedJob, buffer),
        workers:   workers,
        tracker:   trackedjob.NewJobTracker(),
        persister: persister,
    }

    for i := 0; i < q.workers; i++ {
        go ProcessJob(ctx, q, i+1)
    }

    return q
}

Our main function should be like this now.

func main() {
    start := time.Now()

    buffer, workers := 10, 3

    ctx, cancel := context.WithCancel(context.Background())

    sigChan := make(chan os.Signal, 1)
    signal.Notify(sigChan, os.Interrupt, syscall.SIGTERM)

    go func() {
        sig := <-sigChan
        fmt.Printf("\nReceived signal: %s. Shutting down gracefully...\n", sig)
        cancel() // Broadcast shutdown to all workers
    }()

    persister := persister.NewJSONPersister("saved_jobs")

    q := job.NewJobQueue(ctx, buffer, workers, persister)

    q.LoadPersistedJobs()

    q.Submit(&job.SleepyJob{Duration: 1 * time.Second}, job.SubmitOptions{
        MaxRetries: 2,
        Timeout:    4 * time.Second,
    })
    q.Submit(&job.SleepyJob{Duration: 6 * time.Second}, job.SubmitOptions{
        MaxRetries: 0,
    })

    q.Wait()
    q.Shutdown()

    q.StoreStatus()
    fmt.Printf("\nTotal execution time: %.2fs\n", time.Since(start).Seconds())
}

If we try to run now, it fails because the OnStatusUpdate field in the TrackedJob struct cannot be serialized, we can put a json tag to ignore this field.

OnStatusUpdate func(jobID string, status JobStatus, retry int, err error) `json:"-"`

We can do the same for Job (interface) and the mutex.
But there is an issue, rather a big one. We can persist the data related to jobs, but how do we persist the behavior like the interface, mutex, function?

We are trying to serialize the behavior as well. And that won’t work, even if we ignore Job interface or others from getting serialized, where will we derive the behavior from on rehydration?
Let’s step a bit back and redesign our architecture.

What we are going to do is to implement something called factory pattern for job persistence and restoration.

Before that, there’s another mistake we did before. In the original code, both SaveJob and UpdateStatus would lock the same mutex. However, UpdateStatus would then call SaveJob while it was still holding the lock.

This created a deadlock scenario:

UpdateStatus acquires the lock.
UpdateStatus does its work and then calls SaveJob.
SaveJob tries to acquire the same lock, which is already held by UpdateStatus.
The program freezes, as SaveJob waits forever for a lock that will never be released.

To solve this, we introduce a new method, saveJob (with a lowercase 's'), which contains the core logic for writing a job to a file but does not lock the mutex.

We plug this saveJob into the UpdateStatus
Now, we are ready to write our factory.go

This is the heart of the solution for turning data back into behavior. Since we can't store the actual code for a SleepyJob (say) in a JSON file, we need a way to say, "this data belongs to a SleepyJob, please create one for me."

type JobCreator func(config json.RawMessage) (trackedjob.Job, error)

This is the "recipe" for creating a single, specific type of job. The config parameter contains the job's unique data (like the Duration for a SleepyJob), and the function's job is to use that data to construct the actual job object.

Now we need a JobFactory struct. This is the main struct, the factory itself. It holds a single field: a map.

The key is a string which is the unique name for a job type (e.g., "SleepyJob").
The value is the JobCreator function, the "recipe" that knows how to build that specific job.

type JobFactory struct {
    creators map[string]JobCreator
}

func NewJobFactory() *JobFactory {
    return &JobFactory{
        creators: make(map[string]JobCreator),
    }
}

A register function is needed to register the job type.

func (f *JobFactory) Register(jobType string, creator JobCreator) {
    f.creators[jobType] = creator
}

And to finally create a job

func (f *JobFactory) Create(jobType string, config json.RawMessage) (trackedjob.Job, error) {
    creator, ok := f.creators[jobType]
    if !ok {
        return nil, fmt.Errorf("no creator registered for job type: %s", jobType)
    }
    return creator(config)
}

We will use it like this :
When LoadPersistedJobs reads a job file, it gets the JobType string and the JobData. It passes these to this Create method.

It looks up the jobType string in its creators map.
If it finds a corresponding JobCreator function, it executes that function, passing it the config data.
It returns the newly created job object, ready to be executed.
If it can't find a creator for the given jobType, it returns an error, preventing the program from crashing.

Now, our factory is ready, let’s plug it in our existing architecture.

We update our TrackedJob struct first.

type TrackedJob struct {
    JobID      string                `json:"job_id"`
    JobType    string                `json:"job_type"`
    JobData    json.RawMessage       `json:"job_data"`
    Job        Job                   `json:"-"`
    RetryCount int                   `json:"retry_count"`
    MaxRetries int                   `json:"max_retries"`
    Status     JobStatus             `json:"status"`
    Metadata   map[string]string     `json:"metadata"`
    LastError  string                `json:"last_error"`
    Tracker    *JobTracker           `json:"-"`
    Backoff    retry.BackoffStrategy `json:"-"`
    Timeout    time.Duration
    mu         sync.Mutex `json:"-"`

    OnStatusUpdate func(jobID string, status JobStatus, retry int, err error) `json:"-"`
}

JobType (e.g., "SleepyJob") is the string key we use to look up the correct creator in the JobFactory.
JobData (json.RawMessage) is a flexible way to hold the specific configuration for that job (e.g., the Duration for a SleepyJob). It's a byte slice, so the persister doesn't need to understand its contents.

And we change our main Job interface

type Job interface {
    Execute(ctx context.Context) error
    TypeName() string
    Serialize() (json.RawMessage, error)
}

TypeName() forces each job to declare its own identifier, which is used for the JobType field.
Serialize() forces each job to know how to turn its own configuration into a JSON byte slice (JobData). This is a much more scalable design, as each job manages its own data.

Now we can update Submit and LoadPersistedJobs.

When a new job is submitted, it's no longer enough to just put it in the queue. We now need to prepare it for persistence before it even starts running.

It calls job.TypeName() and job.Serialize() to get the job's type and data.
It populates the JobType and JobData fields in the TrackedJob.
This fully-populated TrackedJob is then saved by the persister.

func (q *JobQueue) Submit(job trackedjob.Job, opts SubmitOptions) {
    q.wg.Add(1)

    jobID := GenerateJobID()
    jobData, err := job.Serialize() //here
    if err != nil {
        log.Printf("Failed to serialize job %s: %v", jobID, err)
        q.wg.Done()
        return
    }

    tracked := &trackedjob.TrackedJob{
        JobID:          jobID,
        JobType:        job.TypeName(), //here
        JobData:        jobData,
        Job:            job,
        MaxRetries:     opts.MaxRetries,
        Timeout:        opts.Timeout,
        Status:         trackedjob.StatusPending,
        Metadata:       opts.Metadata,
        OnStatusUpdate: q.handleStatusUpdate,
        Tracker:        q.tracker,
        Backoff: retry.ExponentialBackoff{
            BaseDelay: 500 * time.Millisecond,
            MaxDelay:  5 * time.Second,
            Jitter:    true,
        },
    }

    q.tracker.Register(jobID)

    //persist the job

    if q.persister != nil {
        if err := q.persister.SaveJob(tracked); err != nil {
            log.Printf("Failed to persist job %s: %v", jobID, err)
        }
    }

    q.jobs <- tracked
}

LoadPersistedJobs: This is the reverse process.

It fetches the raw TrackedJob data from the persister.
It uses the job.JobType and job.JobData along with the JobFactory to re-create the correct job object (e.g., a new SleepyJob).
Crucially, it then re-hydrates the TrackedJob by restoring the fields that couldn't be persisted: the Job interface itself, the OnStatusUpdate callback, the Tracker, and the Backoff strategy.

func (q *JobQueue) LoadPersistedJobs(factory *JobFactory) {
    if q.persister == nil {
        return
    }
    jobs, err := q.persister.LoadPendingJobs()
    if err != nil {
        log.Printf("Failed to load persisted jobs: %v", err)
        return
    }
    for _, j := range jobs {
        rehydratedJob, err := factory.Create(j.JobType, j.JobData) //here
        if err != nil {
            log.Printf("Failed to rehydrate job %s of type %s: %v", j.JobID, j.JobType, err)
            continue
        }
        //Putting in the behavior
        j.Job = rehydratedJob
        j.OnStatusUpdate = q.handleStatusUpdate
        j.Tracker = q.tracker
        j.Backoff = retry.ExponentialBackoff{
            BaseDelay: 500 * time.Millisecond,
            MaxDelay:  5 * time.Second,
            Jitter:    true,
        }

        log.Printf("Re-queuing persisted jobs: %s", j.JobID)
        q.wg.Add(1)
        q.jobs <- j
    }
}

Now let us implement the Type and Serialize methods on SleepyJob.

func (j *SleepyJob) TypeName() string {
    return "SleepyJob"
}

func (j *SleepyJob) Serialize() (json.RawMessage, error) {
    return json.Marshal(j)
}

This is how we implement the factory pattern. Below is the new main

func main() {
    start := time.Now()

    buffer, workers := 10, 3

    ctx, cancel := context.WithCancel(context.Background())

    sigChan := make(chan os.Signal, 1)
    signal.Notify(sigChan, os.Interrupt, syscall.SIGTERM)

    go func() {
        sig := <-sigChan
        fmt.Printf("\nReceived signal: %s. Shutting down gracefully...\n", sig)
        cancel() // Broadcast shutdown to all workers
    }()

    persister := persister.NewJSONPersister("saved_jobs")
    jobFactory := job.NewJobFactory()
    jobFactory.Register("SleepyJob", func(config json.RawMessage) (trackedjob.Job, error) {
        var sj job.SleepyJob
        if err := json.Unmarshal(config, &sj); err != nil {
            return nil, err //Duration field of SleepyJob is unmarshalled and stored in JobData
        }
        return &sj, nil
    })

    q := job.NewJobQueue(ctx, buffer, workers, persister)

    q.LoadPersistedJobs(jobFactory)

    q.Submit(&job.SleepyJob{Duration: 1 * time.Second}, job.SubmitOptions{
        MaxRetries: 2,
        Timeout:    4 * time.Second,
    })
    q.Submit(&job.SleepyJob{Duration: 6 * time.Second}, job.SubmitOptions{
        MaxRetries: 0,
        Timeout:    10 * time.Second, 
    })

    q.Wait()
    q.Shutdown()

    q.StoreStatus()
    fmt.Printf("\nTotal execution time: %.2fs\n", time.Since(start).Seconds())
}

Now if we run it, our jobs persist. If we try to stop while a job is executing and restart, the job is loaded and finished. See the screenshots below.

On restarting, (comment out the previous submits)

I think the global ledger is not necessary now, because we have already got our persistence in place, so we remove it.
We can implement the SQLite implementation similarly, and then it will be just plug-n-play. You can check that out in the repo.

And yeah, I think that’s it for this one. We didn’t include circuit breakers, DLQ, or API wrappers here because the goal was to build a solid, minimal core. Those features are valuable, but they deserve their own thoughtful design, and we’ll be treating this project like a product, not just a demo. So we’re keeping it clean, focused, and ready for whatever comes next.

Implementing a Concurrent Priority Queue in Rust

Yashaswi Kumar Mishra — Sat, 12 Jul 2025 08:26:18 +0000

Priority queues are fundamental to scheduling and coordination in systems, but making them thread-safe under real concurrent load is a nuanced challenge. Most implementations fall back on coarse locking, which works, but barely holds up under pressure. I wanted to explore a cleaner design in Rust that allows multiple threads to safely peek and pop without stepping on each other.

This writeup walks through the approach, the tradeoffs, and what it takes to keep things correct and efficient when threads start racing.

Let us start very dumb, by just defining a task struct and our queue would be nothing but a vector of tasks.

We implement debug and clone traits on Task.

#[derive(Debug, Clone)]
struct Task{
    name : String,
    priority : u32,
}

struct PriorityQueue{
    tasks : Vec<Task>
}

Let’s implement push. It just dumps a task to the end of queue. It’s O(1).

We will also write a very simple constructor, which initializes our tasks vector.

impl PriorityQueue{
    fn new() -> Self{
        Self { tasks: Vec::new() }
    }

    fn push(&mut self, task : Task){
        self.tasks.push(task);
    }
}

Let’s write the code for pop now.

 fn pop(&mut self) -> Option<Task>{
        if self.tasks.is_empty(){
            return None;
        }

        let mut max_idx = 0;
        for i in 1..self.tasks.len(){
            if self.tasks[i].priority > self.tasks[max_idx].priority {
                max_idx = i;
            }
        }

        // swap_remove is O(1), just swaps with last and pops
        Some(self.tasks.swap_remove(max_idx))
    }

We do a linear scan of our tasks vector to find the task with the highest priority. swap_remove deletes it without shifting the rest (fast delete).
This pop implementation takes O(n).

Is this design apt for concurrency? Let’s see

Say we have

let queue = PriorityQueue::new();
let shared = Arc::new(queue);

Let’s spawn two threads.

let q1 = Arc::clone(&shared);
    let q2 = Arc::clone(&shared);

    thread::spawn(move ||{
        q1.push(Task { name: "A".into(), priority: 10 });
    });

    thread::spawn(move || {
    q2.push(Task { name: "B".into(), priority: 5 });
});

The compiler throws this error into our face.

cannot borrow data in an Arc as mutable
trait DerefMut is required to modify through a dereference, but it is not implemented for std::sync::Arc<T>

We’re trying to mutate shared data, but Arc<T> doesn’t give us mutable access. That’s where it bites.

Arc<T> is an atomic reference-counted pointer. Multiple threads can safely share it. Now, multiple threads own the same queue.

But, Arc only gives immutable references by default.

We’re trying to mutate data inside an Arc and that’s not allowed.
Rust’s ownership model is very strict about holding multiple mutable refs to the same data. And it is good, as no two things should be able to mutate the same data at the same time. It causes race conditions, and is very bad for business logic.

What can we do then?

We can just wrap our queue in Mutex, which will allow thread-safe exclusive access to the queue. At one time only one thread can access the queue.
We can go something like this.

let shared = Arc::new(Mutex::new(PriorityQueue::new()));

    let handles: Vec<_> = (0..5).map(|i| {
        let queue = Arc::clone(&shared);
        thread::spawn(move || {
            let task = Task { name: format!("T{i}").into(), priority: i };
            let mut q = queue.lock().unwrap();
            q.push(task);
        })
    }).collect();

    for handle in handles {
        handle.join().unwrap();
    }

    let final_q = shared.lock().unwrap();
    println!("Final queue: {:?}", final_q.tasks);

But even Mutexhere is meh.
Every thread waits to acquire the lock, even if they just wanna peek.
More threads = more time fighting for the lock, especially in multi-core systems and this defeats the point of concurrency in the first place.

Doesn’t matter if you want to push, pop, or just check the top , you have to wait for your turn, for the lock to be released and deadlocks are real.

Our priority queue does work, but this is a very underwhelming design.

Our pop is O(n). Every pop is a full scan through the list.
We don’t have peek , so we can’t inspect highest-priority task
There is no thread safety. So this pq can’t be used in multithreaded context yet.
Push doesn’t maintain the priority order, so we can’t optimize reads or implement peek.

Let’s start improving it a bit.

What we can do is make push O(n) and pop O(1), if we want to read data more frequently.

fn push(&mut self, task: Task) {
        // Insert in correct position (sorted by priority ascending)
        let pos = self.tasks.binary_search_by(|t| t.priority.cmp(&task.priority))
            .unwrap_or_else(|e| e);
        self.tasks.insert(pos, task);
    }

    fn pop(&mut self) -> Option<Task> {
        self.tasks.pop() // O(1), max priority is always last
    }

We're doing a binary search to find where to insert the new task, based on its priority. This keeps the vector sorted in ascending order of priority

That means the last element is always the highest priority which is great for pop.
The binary_search_by call is O(log n) – classic binary search.

Still it feels like O(n) because of :

.unwrap_or_else(|e| e)
Which is sort of a hack to extract the index e where the element would be inserted if it’s not found.

And inside binary_search_by, this return value is managed via a loop or branching logic that sometimes leads to linear-time fallback, especially if you mess up the ordering predicate.

But this only seems worse because of the verbose syntax and error-case code paths.

And the insertion via Vec::insert() is O(n) because it has to shift elements.

So the total cost is still O(log n) + O(n). But in practice, due to cache misses, branching, and verbose control flow, it performs worse than you'd expect for small structures.

binary_search_by = O(log n) to find the position
unwrap_or_else(|e| e) = extract insert index if not found
Vec::insert() = O(n) because it has to shift elements

Let’s improve the push and use partition_point()

It finds the first index where the predicate breaks (t.priority ≤ task.priority).

So it walks through all the lower-or-equal priorities and returns the index where the new one should be inserted.
Since the vector is sorted, this is a binary search under the hood → O(log n).
Then, we call Vec::insert(pos, task) → shifts elements → O(n).

Now our push is :

  fn push(&mut self, task: Task) {
    // Maintain ascending order (lowest first, highest last)
    let idx = self.tasks.partition_point(|t| t.priority <= task.priority);
    self.tasks.insert(idx, task);
}

It avoids branching based on search results, better for CPU branch prediction and cache locality.

binary_search_by gives you a Result, which you must handle (found vs not-found). It is more verbose and can lead to unwrap_or_else(|e| e) boilerplate.

partition_point() is semantically cleaner.

Now we can write peek() as well since our vector is ordered.

 fn peek(&self) -> Option<&Task>{
        self.tasks.last()
    }

We could have used a BinaryHeap , which would give us O(log n) push and O(log n) pop.

But we are building for real-world workloads, where reads dominate writes.

Naturally, the next question is “How do we make this concurrent?”

Let us use RwLock because we want concurrent reads but exclusive writes.

This maps perfectly to how a Priority Queue would works in high-concurrency setups.
We allow multiple readers at once, but only one writer at a time, and no readers allowed during write. That is what RwLock .

We already saw why we will not be using Mutex earlier. Mutex is a brutal overkill especially if your workload is read-heavy, like queues that mostly peek() or pop() tasks in a worker loop.

We will also write trivial read functions like len() andis_empty() later.

Our PriorityQueue has 3 kinds of operations:

Operation	Type	Thread Safety Needs
`push()`	write	must block all readers & writers
`pop()`	write	same, modifies vector
`peek()` / `len()` / `is_empty()`	read	can happen in parallel with other reads

Let us wrap our initial PriorityQueue struct in Arc<RwLock<_>>

#[derive(Clone)]
struct SharedQueue{
    inner : Arc<RwLock<PriorityQueue>>
}

That’s it. Now we will just implthis SharedQueue struct, and just write the functions and provide read/write locks.

impl SharedQueue {
    fn new() -> Self {
        Self {
            inner: Arc::new(RwLock::new(PriorityQueue::new())),
        }
    }

    fn push(&self, task: Task) {
        let mut q = self.inner.write().unwrap(); // exclusive write lock
        q.push(task);
    }

    fn pop(&self) -> Option<Task> {
        let mut q = self.inner.write().unwrap(); // still a write
        q.pop()
    }

    fn peek(&self) -> Option<Task> {
        let q = self.inner.read().unwrap(); // shared read lock
        q.peek().cloned()
    }

    fn len(&self) -> usize {
        self.inner.read().unwrap().tasks.len()
    }

    fn is_empty(&self) -> bool {
        self.len() == 0
    }
}

peek(), len(), is_empty() = read locks only
push(), pop() = write locks

This seems fine. Let us test this by

Spawning 5 writer threads to push 10 elements each in the queue.
Some reader threads to peek.
And 2 threads to pop 15 elements each.

We should have 20 elements left at the end.

This is how we spawn these threads.

let queue = SharedQueue::new();

    let mut handles = vec![];

    // Spawn 5 writer threads
    for i in 0..5 {
        let q = queue.clone();
        handles.push(thread::spawn(move || {
            for j in 0..10 {
                let task = Task {
                    name: format!("T{i}_{j}"),
                    priority: i * 10 + j,
                };
                q.push(task);
                println!("[Writer-{i}] Pushed T{i}_{j}");
            }
        }));
    }

    // Spawn 3 reader threads (peek only)
    for i in 0..3 {
        let q = queue.clone();
        handles.push(thread::spawn(move || {
            for _ in 0..15 {
                if let Some(top) = q.peek() {
                    println!("[Reader-{i}] Peeked: {:?}", top);
                } else {
                    println!("[Reader-{i}] Queue is empty");
                }
                std::thread::sleep(std::time::Duration::from_millis(20));
            }
        }));
    }

    // Spawn 2 consumer threads (pop)
    for i in 0..2 {
        let q = queue.clone();
        handles.push(thread::spawn(move || {
            for _ in 0..15 {
                if let Some(task) = q.pop() {
                    println!("[Popper-{i}] Popped: {:?}", task);
                } else {
                    println!("[Popper-{i}] Nothing to pop");
                }
                std::thread::sleep(std::time::Duration::from_millis(30));
            }
        }));
    }

    for h in handles {
        h.join().unwrap();
    }

    println!("\nFinal Length: {}", queue.len());
    println!("Final Queue: {:?}", queue.inner.read().unwrap().tasks);

We .join() all threads to ensure the main thread waits for all worker threads to complete before printing the final state.

When we run the code, this is what we see.

On running the program, the output confirms our expectations:

The 30 highest-priority elements (IDs 20–49) were successfully popped by the threads.
The remaining 20 elements (IDs 0–19) are intact in the heap.
All peek() logs accurately reflect the highest-priority item at various stages of concurrent access.

This validates that our concurrent-safe priority queue handles simultaneous pop() and peek() operations correctly, maintaining both thread safety and priority ordering under concurrent pressure.

We now have a concurrent priority queue that’s:

Thread-safe
Optimized for read-heavy loads
Easy to extend (timeouts, TTL, etc.)
Cleanly abstracted behind SharedQueue

We avoided Mutex contention by using RwLock, and ensured O(n) push (sorted) with O(1) pop for fast consumer reads.

This design shines when:

You have multiple reader threads (workers, peeking schedulers, monitors)
You have predictable and fair task priorities
You want to avoid performance cliffs in BinaryHeap under concurrent access

For high-write workloads or better amortized performance, one might consider:

BinaryHeap + Mutex (less read-friendly)
Lock-free data structures
Priority queues in crossbeam or flume with built-in channels

But this version is perfect for many backend schedulers, job dispatchers, and priority task managers in real-world multithreaded systems.

Thanks for reading. Catch you in the next one!

Lifetimes in Rust aren't that hard

Yashaswi Kumar Mishra — Mon, 26 May 2025 07:20:10 +0000

Understanding lifetimes in Rust is crucial to mastering memory safety without garbage collection. In this post, we break down dangling references, lifetime annotations, and how Rust's borrow checker ensures your code is always safe and sound. Let's dive in.

Dangling References

Let’s take a look at the code snippet below.

fn main() {
    let x : &i32;

    {
        let y : i32 = 10;
       x = &y; // This line will cause a compile-time error
    }

    println!("x: {}", x);

}

This causes a classic compile-time error in rust - “borrowed value does not live long enough”. Let’s understand.

y is a variable which is stack allocated and as soon as the scope ends, it is dropped.

     {
        let y : i32 = 10;
       x = &y; // This line will cause a compile-time error
    }
    //y gets dropped

y is stack-allocated and only lives inside the inner block.
As soon as the block ends, **y is dropped**.
But x is trying to hold a reference to y outside its valid lifetime.
That would make x a dangling reference.

For this to compile correctly, y must simply outlive x.

fn main() {
    let y: i32 = 10;
    let x: &i32 = &y;

    println!("x: {}", x); // Works fine
}

TL;DR

Rust won’t allow refs to outlive the data.
Borrow checker enforces that.

Generical Lifetime Annotations

They simply describe the relation between the lifetimes of multiple references.

fn longest (x : &str, y: &str) -> &str {
    if x.len() > y.len() {
        x
    } else {
        y
    }
}

Here, we get a compile time error :

missing lifetime specifier

this function's return type contains a borrowed value, but the signature does not say whether it is borrowed from x or y

So we will change our function signature like this :

fn longest <'a> (x : & 'a str, y: & 'a str) -> & 'a str

We specify a generic lifetime annotation (starting from an ‘). Here it is ‘a.

x, y and the return value, all use our lifetime annotation. But what does this mean?

There is a relationship between the lifetimes of x, y and the return value. The relationship is : The lifetime of the return value will be same as the smallest of all the arguments’ lifetimes. The return value must not outlive any of the arguments (can be checked by checking if it outlives the argument with the smallest lifetime).

fn main() {
   let name = String::from("pixperk");
   let name2 = String::from("pixperk2");

   let res = longest(&name, &name2);
    println!("The longest name is: {}", res);
}

Here name and name2 do not outlive each other (equal lifetimes). When res is used, the smallest lifetime is still valid and hence res is not a dangling reference.

fn main() {
   let name = String::from("pixperk");
  {
   let name2 = String::from("pixperk2");
   let res = longest(&name, &name2);
   println!("The longest name is: {}", res);
   }
}

Here name2 has the smaller lifetime. But when res is used, the smallest lifetime is still valid and hence res is not a dangling reference.

Let’s change it up a bit.

fn main() {
   let name = String::from("pixperk");
   let res;
  {
   let name2 = String::from("pixperk2");
   res = longest(&name, &name2);
   }
    println!("The longest name is: {}", res);
}

We get - “name2 does not live long enough”. And yes, the reference having the smaller lifetime is not valid, when res is used and hence here it is a dangling reference.

Let’s see what happens when we try returning reference to a local variable.

fn broken<'a>() -> &'a String {
    let s = String::from("oops");
    &s // `s` is dropped, can't return reference
}

This is illegal because you're returning a reference to memory that’s already gone. We can fix this by returning an owned value.

Lifetimes Annotations in Struct Definitions

If we want to use references with structs, we need to specify lifetime annotations.

struct Excerpt<'a> {
    part: &'a str,
}

Our struct must not outlive the reference passed to part. Let’s see this through different cases :

fn main() {
    let novel = String::from("Call me Ishmael.");
    let quote = &novel[..4]; // "Call"

    let excerpt = Excerpt { part: quote };
    println!("{}", excerpt.part);
}

Here, novel owns the string. quote borrows a slice from novel. The struct instance is defined later and novel lives long enough, and hence excerpt is not a dangling reference.

fn main() {
    let excerpt;
    {
        let novel = String::from("Call me Ishmael.");
        let quote = &novel[..4]; 
        excerpt = Excerpt { part: quote };
    }
    // novel is dropped here
    println!("{}", excerpt.part); // Dangling reference!!
}

Here, excerpt, the instance of Excerpt outlives “novel”, and when we try to use excerpt, “novel” is invalid. Therefore, excerpt becomes a dangling reference here and we get a compiler error.

Lifetime Elision

Sometimes, the compiler can deterministically infer the lifetime annotations. It is done by checking three elision rules :

1. Each parameter that is a reference gets its own lifetime parameter

If a function takes multiple references as input, and you don’t explicitly annotate them, Rust assigns a separate lifetime to each.

fn foo(x: &str, y: &str) -> &str {
    if x.len() > y.len() { x } else { y }

This won't compile. Why?

Because Rust sees this like:

fn foo<'a, 'b>(x: &'a str, y: &'b str) -> &str { ... }

But it doesn’t know whether to return 'a or 'b. Hence, the error:

"lifetime may not live long enough."

You must explicitly annotate the output lifetime:

fn foo<'a>(x: &'a str, y: &'a str) -> &'a str {
    if x.len() > y.len() { x } else { y }
}

Now Rust knows that the returned reference is valid for the shortest of both inputs.

2. If there is exactly one input lifetime, that lifetime is assigned to all output lifetime parameters

When your function has one reference input, and you return a reference, Rust assumes the returned reference lives as long as the input.

Example:

fn identity(s: &str) -> &str {
    s

This works because Rust infers:

fn identity<'a>(s: &'a str) -> &'a str {
    s
}

There’s only one input lifetime 'a, so it’s safe to assign it to the output.

3. If there are multiple input lifetimes, but one of them is `&self` or `&mut self`, the lifetime of `self` is assigned to all output lifetime parameters

This rule exists primarily for methods, and it enables a natural feel when writing object-oriented-style code.

Example:

struct Book {
    title: String,
}

impl Book {
    fn get_title(&self) -> &str {
        &self.title
    }
}

Rust automatically infers:


fn get_title<'a>(&'a self) -> &'a str {
    &self.title
}

Even if the method returns a reference to a field, Rust understands that the output reference must not outlive self.

This makes method-writing ergonomic without sacrificing safety.

‘static

A reference with a 'static lifetime means the data it points to will never be deallocated during the program’s execution. It's essentially "immortal" from the compiler’s perspective.

let s: &'static str = "hello";

In this case, "hello" is a string literal stored in the binary’s read-only data segment. It lives for the entire duration of the program, so the reference to it is safely 'static.

Common Sources of `'static` Data

String Literals

Always have 'static lifetimes because they are embedded in the binary and never go out of scope.
Heap-Leaked Data

You can deliberately "leak" heap data so that it lives for the program's entire runtime:
```
let leaked: &'static str = Box::leak(Box::new("persistent string".to_string()));
```
This is useful in situations where global, non-dropping data is required, such as plugin systems or global caches.

`'static` in Practice

Rust often requires 'static in multithreaded or asynchronous contexts where the compiler can't guarantee how long the data will be needed. A classic example is spawning threads:

std::thread::spawn(|| {
    println!("Runs independently");
});

Here, the closure must be 'static because the thread might outlive any borrowed data from the parent scope.

When Not to Use `'static`

A common mistake is to annotate a function like this:

fn longest<'static>(x: &'static str, y: &'static str) -> &'static str

Unless you're certain the data truly lives for the full program duration, this is incorrect. You're promising more than the code can guarantee, which may cause subtle bugs or force you into unnecessary constraints. 'static should not be used to silence the borrow checker.

When It’s Justified

Global constants or config
Singleton patterns
Plugin registries
Global caches
Thread-safe lazy initialization (once_cell, lazy_static)

Lifetimes might feel tricky at first, but they’re what make Rust so powerful. Once you understand the flow of data and references, writing safe and efficient code becomes second nature. Keep experimenting, and lifetimes will click.

Learn auth in the most comprehensive manner.

Yashaswi Kumar Mishra — Thu, 22 May 2025 14:32:19 +0000

Yashaswi Kumar Mishra

May 22 '25

The Definitive Auth Guide for Backend Developers

#backenddevelopment #jwt #security #go

Comments 2

21 min read

The Definitive Auth Guide for Backend Developers

Yashaswi Kumar Mishra — Thu, 22 May 2025 12:21:45 +0000

Most devs treat authentication like a checkbox - “yep, users log in, done.”

Newsflash: that lazy mindset is how apps get hacked and data gets leaked.

If you don’t understand how auth really works, you’re building a house with no locks.

Authentication is simply the process of verifying who someone is. When you log into an app, it checks your identity to make sure you’re actually you, and not some random trying to sneak in.

It’s the foundation for everything secure in your system - without solid authentication, nothing else matters.

This blog is gonna be long because auth isn’t simple, but it’s gonna be useful.

We’ll break down:

stateful vs stateless
bearer tokens
JWT deep dives and how to build them in Go
why PASETO might be the better option
token rotation with access and refresh tokens
and how to handle session hijacking

No fluff, no hype. Just everything you need to build auth that works and stays secure.

Stateful vs Stateless Authentication

There are two paradigms to the game of authentication

1 - Stateful - “I will always remember you”

2- Stateless - “Prove who tf you are each and every time”

Let’s dive deep, explore these shenanigans, cut the jargon and understand all the stuff which will serve as foundation for further implementations.

Stateful Auth

The minimal flow for Stateful auth is as follows :

User logs in.
Server generates a unique Session ID for that user and store it usually in memory (Redis, Memcached, maybe a DB).
This Session ID is sent to the client via cookie.

In case, you have forgotten what a cookie is, it is a choco chip coated… JK, it is a tiny piece of data your browser stores and automatically sends with every request
On each request from the client, the cookie is sent → session ID matched (with the one stored on server) → user authenticated.

Note!

Session IDs need to be regenerated after login to avoid session fixation attacks where an attacker forces a known session ID on the user before login.

So Stateful Auth is your server remembering you, just by your ID.

Stateless Auth

Stateless Auth is how you plug in auth into modern web. Let’s get a basic idea :

Stateless auth means the server doesn’t store anything. Instead, the client holds the proof of identity - usually a token (JWT, PASETO, etc.).

How does it flow?

User logs in.
Server generates a signed token (e.g., JWT) with a payload to store stuff like user id, email etc.
Client stores it (localStorage - don’t do this please, HTTP-only cookie).
On every request, client sends the token → server verifies signature → user authenticated.

Stateless tokens expire. For long sessions, implement a refresh token mechanism stored securely (preferably in an HTTP-only cookie).

Rotate refresh tokens on use.

Chill for now, if you do not get this, we’ll see this stuff later in detail.

Stateless auth is like walking around with a signed passport : nobody needs to remember you.

The Difference

Aspect	Stateful Authentication	Stateless Authentication
Data Theft Risk	Safer : Only the server knows session details, the token is just a reference.	Riskier : All session data is in the token, and it’s not encrypted by default.
Resource Usage	Heavier : Requires frequent access to session storage (e.g., Redis, DB).	Lightweight : All info is in the token; no external lookup needed.
Implementation Complexity	Higher : Needs infrastructure for session storage and persistence.	Simpler : No storage needed; token carries everything.
Scalability	Needs effort : Scaling storage with services is necessary.	Easy : Any instance can verify the token without central storage.
Security Across Services	Secure : Only the auth service can read session data from the store.	Risky : All services must share the same secret key; compromise one, risk all.
Token Size	Small : Token is just a session ID; payload lives on the server.	Can be large : Token holds all auth data, which increases payload size.
Granular Access Control	Flexible : Only the required parts of data can be shared with each service.	Limited : All services decode the full session payload.
Session Revocation	Easy : Simply delete the session from storage.	Hard : Once issued, tokens remain valid until they expire.
Session Data Modification	Yes : Server can update session data on the fly.	No : Token data is fixed until a new token is issued.
Single Sign-On (SSO)	Easy : Centralized session makes it simple to implement SSO.	Tougher : Each service must be able to parse the token independently.

When to Use What?

Use Stateful when:

You need real-time control (revoking, role changes)
You're building a traditional web app (SSR-heavy)
You want full control over the session lifecycle

Use Stateless when:

You're building an API-first, mobile-first, or microservice app
You plan to scale horizontally or use serverless
You’re okay with a refresh-token mechanism

Authorization and Bearer Tokens

Session-based authentication is stateful, while token-based authentication is stateless - and that's exactly why it's the standard for modern web apps. In this blog, we'll focus on token-based auth, since it's the backbone of scalable systems today. But first, let’s get our fundamentals straight: What exactly is authorization?

Authorization is about answering one question: Now that we know who you are, what are you allowed to do?

Once a user is authenticated, your system needs to enforce boundaries. Think of an e-commerce app where you might have two roles: customer and seller. A customer can browse and buy, but shouldn’t be able to list new products. And a seller? Sure, they can manage their own listings, but they definitely shouldn’t be able to change the price of someone else’s product.

That’s authorization, setting the rules of engagement after identity has been verified.

So, what’s a Bearer Token now?

A Bearer Token is like a key - it gives you access to a resource. When you send a request, the server checks the token and decides if you’re allowed in.

Whether you’re trying to drop the price of your own product or sneakily update your competitor’s.

Bearer token is typically passed via the Authorization header in the following format :

Authorization: Bearer <token>

⚠️ Security Note:

Bearer tokens are like cash:

Anyone who gets the token can use it.
They should be transmitted only over HTTPS.
They should be short-lived, revocable, or bound (like in OAuth2’s DPoP).

Are Bearer Tokens always JWTs?

Nope. JWTs are a type of bearer token, but not all bearer tokens are JWTs. Some are:

Opaque tokens: Random strings (UUIDs, hashes), stored and mapped in a backend DB.
Reference tokens (in OAuth2): Tiny strings that reference data stored on the server.
JWTs (JSON Web Tokens): Self-contained tokens that store claims directly and are cryptographically signed.

There are other types of tokens too.

Not all tokens are Bearer tokens. Let’s take a quick tour of some other important ones:

Bearer Token (Generic)

Used in: OAuth2, most APIs
Can be a JWT, PASETO, or opaque
Passed via Authorization header

MAC Token (Message Authentication Code)

Adds request signing: Includes parts like URI, timestamp, etc.
More secure than Bearer (but more complex)
Rare in modern APIs

DPoP (Demonstration of Proof-of-Possession)

An upgrade to Bearer: binds the token to a key, so it can’t be reused if stolen
Becoming popular in OAuth 2.1

PKCE (Proof Key for Code Exchange)

Used in OAuth flows to prevent auth code interception
Works with public clients (like mobile apps)

Refresh Token

Long-lived token used to get new short-lived access tokens
Usually used with bearer tokens or JWTs
We will take a look at this later in detail

Crypto 101

Before we move onto JWTs, let’s take a brief pause and understand some cryptography concepts which help keeping these tokens authentic. Because if you don’t understand how they’re signed, you’ll be a sitting duck for all the sneaky vulnerabilities waiting to pounce.

Why Do We Even Sign Tokens?

Imagine you write a note and hand it to someone. What if that note gets swapped with a fake one? Signing is like putting your personal wax seal on the note - it proves you authored it and it hasn’t been tampered with.

With JWTs, signing is the cryptographic wax seal that guarantees the token’s integrity and authenticity. Without it? Anyone could mess with your payload, pretend to be legit, and wreak havoc.

Symmetric vs. Asymmetric Algorithms

Signing algorithms come in two main flavors:

1. Symmetric (Shared Secret):

Think of this as a secret handshake between two people who both know the same password. Both the creator of the token and the verifier share one secret key. This key is used to both create and verify the signature.

Example: HS256 - HMAC with SHA-256 hashing
Fast, simple, efficient
But here’s the catch: if the secret leaks, everyone can forge tokens.

2. Asymmetric (Public/Private Key):

Now imagine you have a locked mailbox only you can open (private key), but anyone can drop letters in it (public key). This setup uses two keys: the private key signs the token, and the public key verifies it.

Examples: RS256 (RSA), ES256 (Elliptic Curve)
More complex but more secure in distributed systems
Public keys can be shared freely without risking the private key
Slower than symmetric, but worth it for big apps

The Role of Hashing and HMAC

The heart of token signing is hashing - it is like a fingerprint generator. Give some data (say, a string), the hash function returns a unique fixed size output. Even the tiniest change in the input completely changes the output hash - like flipping a single pixel in your fingerprint and getting a whole new identity.

One of the most commonly used hash functions is SHA-256 (Secure Hash Algorithm 256-bit), which always produces a 256-bit (32-byte) hash output.

Here’s what makes hashes powerful:

One-way function : You can compute the hash of data easily, but you can’t reverse it to get the original data.
Sensitive to changes : Change one bit of input and the output is radically different.

But here's the problem: Hashing alone ≠ security. If someone knows your input, they can hash it too and pretend it’s legit.

Enter: HMAC (Hash-Based Message Authentication Code)

HMAC solves this. It's like giving your hash function a secret weapon - a secret key.

Instead of just hashing the message, HMAC combines:

HMAC(secret key, message)

This means only someone with the correct secret key can produce the correct HMAC signature.

It works like this (simplified flow):

You take your message.
You mix it with a secret key in a specific way (padding, XORing, etc.).
You run it through a hash function like SHA-256.
The output is a keyed hash - the HMAC.

Now, anyone who receives your message and has the same secret key can recompute the HMAC and check if it matches. If it does: message was legit and untampered.

When you sign a JWT using HS256, it’s literally using:

HMAC-SHA256(secret key, base64(header) + "." + base64(payload))

So if the token was altered even slightly, the signature won’t match when verified - and you immediately know something shady went down.

But here’s the thing:

If your secret key is weak or leaked, an attacker can compute their own HMACs and forge tokens.
If your server doesn’t verify the signature correctly, you’re wide open to manipulation.

JWT - A Deep Dive

After surviving the Crypto 101, I think we should move onto dissecting JSON Web Tokens (JWTs). A clean JWT implementation in GoLang will follow this.

A JWT is a compact, URL-safe token format used to securely transmit information between two parties. It consists of three base64url-encoded parts, separated by dots (.):

<Header>.<Payload>.<Signature>

Let’s break down the role of each piece :

Header

The header is like the envelope : it tells the receiver what kind of message this is and how it’s sealed.

Example:

{
  "alg": "HS256",
  "typ": "JWT"
}

alg: the signing algorithm used, e.g., HS256 (HMAC with SHA-256) or RS256 (RSA with SHA-256).
typ: the type of token , typically JWT.

This header is Base64URL-encoded before being used in the signature process.

Payload

The payload is the actual data , the message you want to send. It contains claims, which are statements about the subject (user, client, etc.).

Example payload:


{
  "sub": "user_abc123",
  "name": "Yashaswi",
  "role": "admin",
  "iat": 1716351700,
  "exp": 1716355300
}

There are three types of claims:

Registered claims – standard fields like sub (subject), iat (issued at), exp (expiration), iss (issuer), aud (audience).
Public claims – user-defined but meant to be collision-free (usually namespaced).
Private claims – totally custom to your app (role, plan, tier, etc.).

Important: The payload is not encrypted. Anyone can decode it with a Base64 decoder. That means:

Never store sensitive data like passwords, tokens, or PII here.

Signature

The signature is the cryptographic seal. It ensures the token hasn’t been tampered with.

If you’re using HS256 (symmetric HMAC), here’s how the signature is created:


HMAC-SHA256(
  secret,
  base64urlEncode(header) + "." + base64urlEncode(payload)
)

If you’re using RS256 (asymmetric RSA), the private key is used to sign, and the public key to verify.

The signature guarantees integrity and authenticity. If someone modifies the payload, the signature breaks. If someone doesn’t have the secret (or private key), they can’t forge a valid token.

Why JWTs Are Powerful

Stateless: No server-side session storage needed. Just validate and go.
Portable: You can pass them across APIs, domains, or services easily.
Tamper-proof: Thanks to the cryptographic signature.
Self-contained: All the info needed to process the request is in the token itself.

Perfect for:

REST APIs
Microservices
Mobile apps
OAuth2 flows

But JWTs Aren’t Magic - Use With Caution

They have caveats:

❗ Expiration Matters

Long-lived tokens are dangerous and short-lived tokens are very annoying for the users. Find the sweet spot. Rotate tokens using refresh tokens.

❗ Signature Doesn’t Mean Encryption

People confuse the two all the time. JWTs are not encrypted by default. Anyone can decode them. Use JWE (JSON Web Encryption) or transmit them securely via HTTPS.

❗ Revocation is Hard

Once issued, JWTs live until they expire. You can’t just “log out” someone unless you maintain a blacklist or use short expiration with refresh tokens.

❗ Beware of “alg: none”

This infamous vulnerability allowed attackers to forge tokens by passing "alg": "none" in the header - which some libraries didn’t verify properly. Always validate the algorithm on the server side.

Implementing JWT Authentication in Go

Now that we have enough idea about what is JWT and how it looks, let’s implement JWT auth in GoLang.

Let us start by creating the payload struct.

type Payload struct {
    ID        uuid.UUID `json:"id"`
    Username  string    `json:"username"`
    IssuedAt  time.Time `json:"issued_at"`
    ExpiredAt time.Time `json:"expired_at"`
}

You can extend this struct with fields like Email, Roles, or whatever your use case needs. But keep the token lean to avoid payload bloat.

We’ll define two error constants that we’ll use later.

var (
    ErrExpiredToken = errors.New("token has expired")
    ErrInvalidToken = errors.New("token is invalid")
)

Here’s a constructor to initialize the Payload struct:

func NewPayload(username string, duration time.Duration) (*Payload, error) {
    tokenID, err := uuid.NewRandom()
    if err != nil {
        return nil, err
    }

    payload := &Payload{
        ID:        tokenID,
        Username:  username,
        IssuedAt:  time.Now(),
        ExpiredAt: time.Now().Add(duration),
    }

    return payload, nil
}

And finally an implementation over payload to check whether the token is valid:

func (payload *Payload) Valid() error {
    if time.Now().After(payload.ExpiredAt) {
        return ErrExpiredToken
    }

    return nil
}

Note: jwt-go will call this Valid() function internally when verifying tokens - but YOU must return meaningful errors inside it, like we did with expiry. Don’t just YOLO it.

You can also move that expiry check into the VerifyToken logic directly if you want to be explicit, but this is fine as long as you call it via jwt.ParseWithClaims

That’s it. Our payload struct and implementations over it are ready.

Next we will hop onto making a general ‘maker’ interface, whose methods our jwtMaker will implement. We will reuse the maker interface later to implement pasetoMaker as well.

This is how our maker interface will look like :

type Maker interface {
    CreateToken(username string, duration time.Duration) (string, error)
    VerifyToken(token string) (*Payload, error)
}

Now let’s jump into the meaty part and build the jwtMaker.

const minSecretKeySize = 32

type JWTMaker struct {
    secretKey string
}

func NewJWTMaker(secretKey string) (Maker, error) {
    if len(secretKey) < minSecretKeySize {
        return nil, fmt.Errorf("invalid key size : must be at least %d characters", minSecretKeySize)
    }
    return &JWTMaker{secretKey}, nil
}

We start by making a JWTMaker struct which consists of a secretKey field used to sign (and possibly verify which depends on symmetric/asymmetric key algorithm). Inside the constructor, if the length of the secret key is less than the minimum key size, we return an error early. Else we instantiate the interface with a JWTMaker struct instance and return no error.

Now this JWTMakeris required to implement two methods, CreateToken and VerifyToken. Let’s implement them. We will be using this JWT library : github.com/golang-jwt/jwt/v5

CreateToken :

func (maker *JWTMaker) CreateToken(username string, duration time.Duration) (string, error) {
    payload, err := NewPayload(username, duration)
    if err != nil {
        return "", err
    }

    jwtToken := jwt.NewWithClaims(jwt.SigningMethodHS256, payload) //comes from the jwt lib
    return jwtToken.SignedString([]byte(maker.secretKey))
}

We accept username and time duration as params and construct our payload using these. Then we create a new JWT using HS256 algorithm and return it after signing it with our secret key.

VerifyToken :

Okay, things get interesting here. Now we need to verify the token and decode the embedded payload. Let’s take a look over the whole code first, then understand it bit by bit.

func (maker *JWTMaker) VerifyToken(token string) (*Payload, error) {
    keyFunc := func(token *jwt.Token) (interface{}, error) {
        _, ok := token.Method.(*jwt.SigningMethodHMAC)
        if !ok {
            return nil, ErrInvalidToken
        }
        return []byte(maker.secretKey), nil
    }
    jwtToken, err := jwt.ParseWithClaims(token, &Payload{}, keyFunc)
    if err != nil {
        verr, ok := err.(*jwt.ValidationError)
        if ok && errors.Is(verr.Inner, ErrExpiredToken) {
            return nil, ErrExpiredToken
        }
        return nil, ErrInvalidToken
    }
    payload, ok := jwtToken.Claims.(*Payload)
    if !ok {
        return nil, ErrInvalidToken
    }

    return payload, nil
}

Here’s our flow of verifying the token :

Confirm HMAC sig method (protect from alg-switch)
Parse token with secret
Handle expiration separately
Extract user claims safely
Reject all else with fire

We accept the token as param , then :

    keyFunc := func(token *jwt.Token) (interface{}, error) {
        _, ok := token.Method.(*jwt.SigningMethodHMAC)
        if !ok {
            return nil, ErrInvalidToken
        }
        return []byte(maker.secretKey), nil
    }

keyFunc is a callback function required by jwt.ParseWithClaims. Its job is to return the secret key used to sign the token. We first check if the token was signed using HMAC (HS256/HS512). It prevents someone from switching the algorithm to, say, RS256, and trying to mess with you. If the algorithm doesn’t match, we return an invalid token error. If everything’s cool, return the actual signing secret so the JWT lib can verify the signature.

    jwtToken, err := jwt.ParseWithClaims(token, &Payload{}, keyFunc)
    if err != nil {
        verr, ok := err.(*jwt.ValidationError)
        if ok && errors.Is(verr.Inner, ErrExpiredToken) {
            return nil, ErrExpiredToken
        }
        return nil, ErrInvalidToken
    }

Now we parse the token from the client into the empty Payload struct, and also pass keyFunc as param for validating the signature. If parsing fails, check if the error is a ValidationError (e.g. expired, malformed, etc.). If the inner error was due to token expiration, return a clean ErrExpiredToken. For any other error, return ErrInvalidToken.

    payload, ok := jwtToken.Claims.(*Payload)
    if !ok {
        return nil, ErrInvalidToken
    }

    return payload, nil
}

Extract the claims and assert they are of type Payload. If type assertions fail, bail. Otherwise, token is valid, signed by you, not expired, and has a clean payload. Send it to the handler.

Now our JWTMaker is ready. We can use the createToken in our login api and give out an access token on logging in. In the middleware for protected resources, we will check for the Bearer Token in the Authorization Header, and use verifyToken for that.

Below is the authorization middleware example (in Gin).

func authMiddleware(tokenMaker token.Maker) gin.HandlerFunc {
    return func(ctx *gin.Context) {
        // Get the Authorization header
        authorizationHeader := ctx.GetHeader(authorizationHeaderKey)
        if len(authorizationHeader) == 0 {
            err := errors.New("authorization header not provided")
            ctx.AbortWithStatusJSON(http.StatusUnauthorized, errorResponse(err))
            return
        }

        // Expecting format: "Bearer <token>"
        fields := strings.Fields(authorizationHeader)
        if len(fields) < 2 {
            err := errors.New("invalid authorization header format")
            ctx.AbortWithStatusJSON(http.StatusUnauthorized, errorResponse(err))
            return
        }

        // Check if type is Bearer
        authorizationType := strings.ToLower(fields[0])
        if authorizationType != authorizationTypeBearer {
            err := errors.New("unsupported authorization type")
            ctx.AbortWithStatusJSON(http.StatusUnauthorized, errorResponse(err))
            return
        }

        // Verify token
        accessToken := fields[1]
        payload, err := tokenMaker.VerifyToken(accessToken)
        if err != nil {
            ctx.AbortWithStatusJSON(http.StatusUnauthorized, errorResponse(err))
            return
        }

        // Save payload to context
        ctx.Set(authorizationPayloadKey, payload)
        ctx.Next()
    }
}

These are constants for Bearer Token and Authorization Header :

const (
    authorizationHeaderKey  = "authorization"
    authorizationTypeBearer = "bearer"
    authorizationPayloadKey = "authorization_payload"
)

That’s it! That is our JWT implementation in Go.

PASETO - The Anti JWT

PASETO (Platform-Agnostic Security Token) is a secure-by-default, minimalistic, and less error-prone alternative to JWT.

Here’s the format of a typical PASETO token :

<version>.<purpose>.<base64url-encoded-payload>

Let’s zoom into the three core components of any PASETO token:

Version

The first part of the PASETO token is the version identifier - it tells you which set of cryptographic primitives is being used.

Version	Crypto Stack	Status	Use Case
v1	AES-CTR + HMAC / RSA-PSS	Deprecated	OG version
v2	XChaCha20-Poly1305 / Ed25519	✅ Stable	Widely used
v3	AES-CTR + HMAC / ECDSA	✅ Stable	FIPS-compliant
v4	XChaCha20-Poly1305 / Ed25519 (newer implementations)	✅ Latest	Recommended going forward

v2 and v4 are your go-to for modern apps.

If you don’t need US government FIPS compliance, skip v3.

Purpose

The second part tells you the intent of the token:

local → Encrypted token (confidentiality)
public → Signed token (integrity)

Why does this matter?

Because you can’t mix & match. PASETO makes the purpose explicit. It avoids the confusion JWT has with "alg": none or "alg": HS256".

→ local : Encrypted (Symmetric Key):

v4.local...

Nobody can read the payload unless they have the key.
Think: password vaults, medical data, PII.

→ public : Signed (Asymmetric Key)

v4.public...

Anyone can verify the payload was created by your server.
Useful for access tokens : verify without decrypting.

Payload

It remains the same as JWT.

If you're using public mode, don’t put sensitive info in the payload - it's not encrypted, just signed.

Sample PASETO (Encrypted)

v4.local.Y2hhY2hhMjBfZW5jcnlwdGVkX3N0dWZmLmRhdGEuZ29lc19oZXJl

Version: v4
Purpose: local (encrypted)
Payload: Encrypted JSON like { "user_id": "123", "exp": ... }

Can you decode it?

Nope. Unlike JWT, you can't base64 decode and peek inside unless you have the secret key

Why PASETO over JWT?

JWT’s spec is just too flexible. This flexibility can sometimes become a liability.

Here’s why JWTs can be potentially dangerous and why I personally prefer PASETO.

The alg field is the vulnerability magnet in JWT.

JWT lets you choose your signing algorithm and some servers blindly trust the alg field. Real attacks have occurred by switching from "RS256" to "none" , bypassing signature checks.
JWT supports multiple algorithms: HS256, RS256, ES256, etc. If you sign with RS256 (asymmetric), but verify with HS256 (symmetric), then an attacker can craft a token using the public key (intended only for verification) as the secret (for HMAC), and yeah that will lead to token forgery.

Many JWT libraries have inconsistent implementations. Not all :

Validate claims properly
Enforce expiration
Check audience or issuer
Reject malformed tokens

JWT supports both signed (JWS) and encrypted (JWE) formats. But many developers:

Don't encrypt sensitive data
Misunderstand the difference
End up leaking info in a "secure" token

On the contrary, PASETO is better. PASETO was designed after JWT, learning from its chaos.

PASETO does not have any alg field.
Every PASETO version has predefined algorithms, safe defaults and no flexibility (not a limitation).
PASETO tokens have clear format and no headers and hence no chances of manipulation.
It is encrypted by default (local tokens).
There are no ambiguous token types in PASETO:

vX.public = signed token

vX.local = encrypted token

All in all, there is less surface area to screw things up.

Feature	JWT	PASETO
Configurable algs	✅	❌
Encrypted by default	❌	✅ (`local`)
Prone to alg confusion	✅	❌
Secure defaults	❌	✅
Public-key safe	Depends	✅
Dev Footgun Potential™	🔥🔥🔥	🧊

And for the PASETO implementation in Go, it similar to if not simpler than the JWT implementation. In my project, Vaultify, I have implemented PASETO auth. You can check out the code here :

That’s it about PASETO. PASETO is just JWT but won’t give you nightmares.

Access Tokens, Refresh Tokens and Token Rotation

What is an Access Token ?

An access token is a digital key which facilitates seamless access to the resources. The JWTs or PASETO we saw until now are all access tokens. This is what is passed in the Authorization header with the Bearer prefix (as seen before) .

We can use them in OAuth (for passwordless auth), or return an access token after the typical email-password login.

This access token lets access to the resources without the need to login repeatedly. Access token are short lived - typically 5-90 minutes.

It is validated by the server on every request and usually stored in localStorage (not recommended), cookies or in memory(volatile but safe). Local storage is vulnerable to XSS (Cross Side Scripting) attacks. Even session storage is exposed to JS.

Why is it short-lived? Because it is stateless. Once signed, it’s independent of the DB. They cannot be revoked (unless blacklists are used). If leaked, it's usable immediately until it expires.

Refresh Tokens - What and Why ?

Access tokens are only valid for a short time period. Once expired, the client can use a refresh token to request for a new access token.

Therefore, a refresh token is a credential (or key) that lets client get new access tokens without having to ask the user login again.

Key Points:

Longer-lived (like 7–30 days)
Stored securely (HttpOnly cookie or secure storage)
Needs to be stored in the DB
Must be ROTATED (replaced with a new one after each use)

If you used just an access token, you’d have to make the expiry super long so users don’t get logged out often.

But long-lived tokens can ultimately cause security risk.

So you split it :

Token	Life Span	Use For	Stored In
Access Token	15 mins	API requests	Memory / cookie
Refresh Token	7+ days	Getting new token	HttpOnly cookie + DB

In the diagram above, SPA = Single-Page Application; AS = Authorization Server; RS = Resource Server; AT = Access Token; RT = Refresh Token.

Token Rotation

Refresh tokens are also bearer tokens, and hence there is a possibility that they get leaked. And if the attacker gets hold of any refresh token, they can generate infinite access tokens and abuse the resources.

To prevent this, we use the strategy of token rotation. Token rotation is refresh token lifecycle management with built-in replay protection.

Here is the flow for a minimal token rotation strategy (AT stands for Access Token and RT stands for Refresh Token)

Login:
- Issue AT1 (short-lived)
- Issue RT1 (long-lived)
- Store RT1 in DB with status: active, used = false
Token refresh request:
- Client sends RT1
- Server checks DB:
  - If used == false → valid
  - If used == true → replay attempt → invalidate all tokens for this user
- Rotate:
  - Mark RT1 as used = true
  - Issue new tokens AT2, RT2
Client replaces local token store with AT2 and RT2

This is secure :

You bind the refresh token lifecycle to the DB
Every token is single-use only
If attacker tries to reuse RT1, server knows something’s wrong

Some More Token Rotation Strategies

Sliding Expiry

Each time the refresh token is used to get a new access token, it extends its own life.

So, if you’re actively using the app, the token’s expiry keeps getting pushed forward - say, 30 days from today, not from the original issue date.

It mimics real sessions - active users stay logged in, idle ones get logged out automatically.

Multiple Devices

A user logs in:

Phone → gets RT1, deviceID = iOS-1
Browser → gets RT2, deviceID = Chrome-1

Each refresh token is associated with a session ID and device. You can:

Revoke all tokens for a user
Revoke only a specific session (say user logs out of iOS)

Detecting Stolen Refresh Tokens

“But what if attacker uses a refresh token before the user does?”

Here’s how:

Server issues RT1
Client hasn't used it yet
Attacker uses RT1
Server rotates and gives back RT2a
Legit client tries using RT1 — fails

This is your “signal” that the refresh token was used by an attacker first.

Response?

Invalidate all user refresh tokens
Log user out everywhere
Alert user

You’ve just detected a breach proactively.

Access tokens are stateless. You don’t need to store them.

The DB only stores refresh tokens (and optionally, blacklisted access tokens if you want logout-before-expiry).

Avoid these Mistakes

Mistake	Fix
Keeping refresh token in localStorage	Use HttpOnly secure cookies
No rotation	Always rotate refresh tokens
Not expiring tokens	Use TTL + sliding expiry
Using same refresh token forever	Rotation. Rotation. Rotation.
Not binding refresh token to IP/device	Store `User-Agent`, `deviceID` optionally

Session Hijacking & Mitigation Tactics

Session hijacking is when the hacker steals your session cookie or token and gets full access as if they were you. No extra login needed. Hijacking is basically identity theft at the protocol level

The Anatomy of Session Hijacking: How They Do It

Session Sniffing

The attacker captures session tokens by sniffing network traffic - classic MITM (man-in-the-middle) attack. Happens mostly on unencrypted HTTP or insecure Wi-Fi.
Cross-site Scripting (XSS)

Inject malicious scripts to steal tokens from cookies or localStorage in the victim’s browser.
Session Fixation

Trick the victim into using a session ID chosen by the attacker, then hijack it once the victim logs in.
Predictable Session IDs

Generate session tokens with weak randomness, letting attackers guess valid tokens.
Malware & Keyloggers

Steal tokens from the victim’s device directly.

The Migration Tactic

Use HTTPS Everywhere

If your app still serves login or session tokens over plain HTTP, you’re basically handing out your keys to every script-kiddie sniffing Wi-Fi packets.

HTTPS encrypts data-in-transit, making MITM sniffing almost impossible.
Set your cookie flag to Secure to only send cookies over HTTPS.

Set HttpOnly Cookies

Cookies with the HttpOnly flag cannot be accessed via JavaScript. That’s a direct shield against XSS attacks stealing cookies.

Set-Cookie: sessionId=abc123; HttpOnly; Secure; SameSite=Strict;

Implement SameSite Cookies

SameSite attribute helps prevent CSRF (cross-site request forgery) by restricting when cookies get sent.

Strict - only send cookies for same-site requests.
Lax - send cookies for top-level navigation GET requests.
None - send cookies everywhere (only with Secure).

For sessions, Strict or Lax is the way to go unless you need cross-origin requests.

Regenerate Session IDs on Login

Avoid session fixation by generating a brand-new session ID after successful authentication.

If you keep the same ID from before login, attackers who fixed that ID can instantly hijack.

Use Strong, Unpredictable Session Tokens

Randomness is king. Use cryptographically secure random generators for tokens.

Avoid any incremental or guessable IDs like session123, abc001, etc.

Set Short Session Expiry and Idle Timeout

Balance UX with security:

Short-lived tokens reduce attack window.
Idle timeouts log out inactive users automatically.
Combine with refresh tokens if needed.

Use Secure Authentication Tokens (JWT, PASETO or Opaque Tokens)

JWTs/PASETO are popular but can backfire if you blindly trust them without verification.

Opaque tokens with server-side session store are simpler to revoke on logout or hijack detection.

Monitor and Detect Anomalies

Keep an eye on session behavior:

Unusual IP switches
Excessive requests from same token
Multiple concurrent sessions from different geos

Set up alerting or automated session invalidation.

If you don’t prioritize session security, you’re basically inviting hackers to take over your users’ accounts and burn your backend reputation to ashes. And trust me, no one wants to be that dev who ignored basic session hygiene.

That was a long one. Let’s wrap up!

Alright, if you’re still here - damn. You just sat through a full-course meal of authentication: stateful vs stateless, bearer tokens, JWT internals, PASETO, token rotation, session hijacking, and even writing your own token system in Go. That’s not light reading. That’s “backbone-of-your-entire-app” reading.

If I sounded a little serious (or aggressive) in parts - yeah, my bad. It’s just that I’ve seen too many half-baked auth systems floating around in production like it’s no big deal. And someone had to say it.

But real talk: this stuff is hard. And it should be. Auth is one of those areas where being “close enough” usually means “completely broken.”

Hopefully this wasn’t just useful, but also kinda fun. Or at least bearable. If you’re walking away from this with a clearer mental model and some ideas you wanna implement or refactor- that’s a win.

Now go build auth like you actually care about your users.

Catch you in the next one.

Go Concurrency: The Full Guide to Goroutines and Beyond

Yashaswi Kumar Mishra — Mon, 28 Apr 2025 08:07:43 +0000

"Concurrency is not parallelism... but also kinda is if you squint hard enough."

Let’s be real — if you're here, you're probably either:

Building your first Go app and just saw the word goroutine thrown like a ragebait on Twitter,
Trying to impress your future self by understanding what makes Go a beast at handling millions of things at once, or
Debugging a deadlock so complex you started questioning your life decisions.

Whichever path led you here — respect.

Because concurrency isn't some optional sprinkle.

In Go, it’s baked right into the damn cake.

In this blog, we’re gonna really talk about concurrency.

Not just shallow examples that say "just use go func()."

Nah, we're going all in : Goroutines, Channels, Buffered Channels, Deadlocks, Select, WaitGroups, Mutexes, Context — all explained like you're grabbing coffee with your backend bro.

Warning:

This is gonna be a long ride.

It’s basically a one-stop shop — after this, you won’t need another “Go concurrency” blog.

So grab a coffee ☕, maybe some cookies 🍪, and let’s get you absolutely cracked at this.

Concurrency Explained
Goroutines
Channels
The Select Statement
WaitGroups and Mutexes
The Context Package
Common Concurrency Patterns in Go
Common Pitfalls With Concurrency in Go

Concurrency explained

Concurrency is about dealing with a bunch of things at the same time, but not necessarily doing them simultaneously.

Example : Computer’s processor switches between tasks rapidly, giving the illusion of doing many things at once.

Go routines in Go let you write code where multiple tasks appear to run at the same time, but the system is just switching tasks efficiently.
Concurrency is managing many tasks at once (but not necessarily simultaneously), while Parallelism is actually executing multiple tasks at the same time. Parallelism needs multiple cores/CPUs to take full advantage.

Where is Concurrency even needed?

I/O Bound Operations (Networking, File Operations, Database Queries):
- You have one process waiting on data (from a database, an API, or even a file). While it waits, another task can run. This is perfect for concurrency.
- Example: Web servers handling a lot of incoming requests. Each request might be waiting for data (like an API response), but the server doesn’t stop. It keeps working on others in the meantime.
Handling Multiple Requests Simultaneously:
- Think of a backend API handling multiple users. While one request is waiting for a response, another one should not have to wait. Concurrency lets that happen smoothly.
- Example: A backend service in Go that needs to process hundreds of incoming HTTP requests. You don’t want to process them one by one. You fire off a goroutine for each request. You don’t even notice the bottleneck.
When You Have Time-Consuming Tasks:
- Your server may need to run long tasks (like image processing, video encoding). With concurrency, you can fire these tasks in parallel and return control to the user.
- Example: If a user uploads an image and it needs to be resized, you don’t block them - you fire off a goroutine to handle the resizing, and the user can keep doing other things.
Distributed Systems and Microservices:
- You need to coordinate between multiple services that are processing things in parallel.
- Concurrency helps to deal with distributed services and allow them to talk to each other without blocking any process.
Background Work:
- Say you're building a system where things are running in the background (like a task scheduler). Concurrency makes sure things run in the background without blocking anything else.

Goroutines - Concurrently running Functions

Goroutines are lightweight, concurrent units of execution. A goroutine is essentially a function or method that runs concurrently with other functions in the same program.

They do consume a trivial amount of memory compared to the conventional threads and the scheduling of these routines is managed internally by Go. Now you do not need to worry about low level thread management anymore.

A goroutine takes around 2 KB of stack space, which is a tiny fraction compared to threads in other languages.

Multiple goroutines can execute concurrently, but they may not run simultaneously on multiple CPUs unless the Go runtime is explicitly set to use multiple CPUs.

Start a Goroutine - Syntax Shenanigans

In Go, you can spawn a goroutine by adding the go keyword before a function call.

package main

import (
    "fmt"
    "time"
)

func greet() {
    fmt.Println("Hello from the goroutine!")
}

func main() {
    // Creating a goroutine to run greet function concurrently
    go greet()

    // Main function waits for the goroutine to finish (sleep for demonstration)
    time.Sleep(1 * time.Second) // Wait for the goroutine to complete
}

go greet() here starts a concurrently running function (a goroutine)
time.Sleep is used here just to make sure that the main function doesn't exit before the goroutine finishes. Without this, the program might exit before the goroutine even has a chance to run.

→ Let’s see a good scenario and use goroutines there.

You want to fetch multiple URLs concurrently (e.g., for web scraping or API calls). Each fetch operation will be a separate goroutine.

package main

import (
    "fmt"
    "net/http"
    "time"
)

func fetchURL(url string) {
    // Simulating fetching a URL
    fmt.Printf("Started fetching %s\n", url)
    _, err := http.Get(url)
    if err != nil {
        fmt.Printf("Error fetching %s: %v\n", url, err)
        return
    }
    fmt.Printf("Finished fetching %s\n", url)
}

func main() {
    urls := []string{
        "https://example.com",
        "https://example.org",
        "https://example.net",
    }

    for _, url := range urls {
        // Spawning a goroutine for each URL fetch
        go fetchURL(url)
    }

    // Giving some time for goroutines to finish (this could be improved with synchronization)
    time.Sleep(2 * time.Second)
}

Each fetchURL call is executed concurrently by spawning a goroutine with go fetchURL(url).
We simulate a small delay in the program using time.Sleep(2 * time.Second) so that all the goroutines have time to finish executing.

Goroutine Scheduling and the Runtime

→ Go runtime is the one that manages goroutines.

The Go scheduler takes care of scheduling goroutines onto available OS threads.
The number of goroutines can exceed the number of CPU cores, and the Go runtime will ensure they are executed without much performance overhead.

How the Go runtime handles it:

Lightweight: Each goroutine uses very little memory and stack space.
M:N Scheduling: Go uses an M:N scheduler, meaning multiple goroutines (M) are multiplexed onto fewer threads (N), which allows many goroutines to run with fewer threads, saving memory.

Channels

What are Channels?

→ Channels behave like “walkie-talkies” between two goroutines. They help pass data between two channels.
→ Think of a channel as something like a pipe, the data goes from one end of a pipe (one goroutine) and received at another end (another goroutine)

→ We don’t need mutexes (locks) in channels : Channels manage synchronization internally.

→ Syntax

ch := make(chan int) // Channel to send and receive int values
ch <- value //Send into channel [data into rear end]
value := <-ch//Receive from channel [received from front end]

A really simple example :

package main

import "fmt"

func worker(ch chan string) {
    // Receiving data from channel
    msg := <-ch
    fmt.Println("Received:", msg)
}

func main() {
    ch := make(chan string) // create channel
    go worker(ch)           // start goroutine

    // Send data into channel
    ch <- "Hello pixie!"  // blocks until worker reads it
}

main sends "Hello pixie!" into ch
worker goroutine reads from ch
Synchronized communication: Sending and receiving blocks until both are ready.

Buffered vs Unbuffered Channels:

Unbuffered channel → send blocks until another goroutine is ready to receive.

go func() {
    ch <- 10 // This will BLOCK until someone receives
}()

value := <-ch  // Receiving side → unblocks sender

Buffered channel → send blocks only if buffer is full.

ch := make(chan int, 5) // capacity = 5

Sender can send up to 5 items without any receiver.
Only blocks if buffer is full.
Receiver blocks only when buffer is empty.

ch := make(chan int, 2)

ch <- 1  // ok
ch <- 2  // ok
ch <- 3  // BLOCKS (buffer full!)

Closing a Channel

close(ch)

What happens?

No more values can be sent into ch.
You can still receive remaining values.
Reading from a closed empty channel gives zero value (0 for int, "" for string, etc).

IMPORTANT:

Only sender should close the channel, NEVER receiver.
Closing a channel signals to receivers:

"No more food coming 🍗, clean up!"

A small summary before we move ahead :

Thing	Meaning
Unbuffered channel	Direct handshake between sender and receiver
Buffered channel	Temporary shelf (buffer) — senders/receivers can work independently until full/empty
Close	Only sender should close to signal "no more data"

CHANNEL
↳ Unbuffered (default)
Both sender and receiver must be ready
↳ Buffered (capacity N)
Sender blocks only when full
Receiver blocks only when empty

↳ Close(channel)
Signals no more data
Receiver reads remaining values
Further sends panic

Deadlocks in Channels

Let’s see how basic deadlocks are caused (let’s take unbuffered channel here)

package main

import (
    "fmt"
)

func main() {
    ch := make(chan int) // UNBUFFERED

    // Try to send
    fmt.Println("Sending 1...")
    ch <- 1 // ❌ BLOCKS HERE — no one is receiving!
    fmt.Println("Sent 1!") // ❌ NEVER REACHED
}

This will cause deadlock.

Because you’re sending but no one's receiving.

package main

import (
    "fmt"
)

func main() {
    ch := make(chan int)

    go func() {
        val := <-ch // Now somebody is ready to RECEIVE
        fmt.Println("Received:", val)
    }()

    fmt.Println("Sending 1...")
    ch <- 1 // ✅ Now send will succeed, because receiver exists
    fmt.Println("Sent 1!") 
}

Slightly Advanced: 2-Way Deadlock

Two goroutines waiting on each other = death spiral.


package main

import "fmt"

func main() {
    ch1 := make(chan int)
    ch2 := make(chan int)

    go func() {
        ch1 <- 42      // Waiting for someone to receive from ch1
        val := <-ch2   // Will never reach here
        fmt.Println("Goroutine 1 got", val)
    }()

    go func() {
        ch2 <- 24      // Waiting for someone to receive from ch2
        val := <-ch1   // Will never reach here
        fmt.Println("Goroutine 2 got", val)
    }()

    select {} // Block main forever (simulate long run)
}

Both goroutines are stuck:

First trying to send.
No one receiving first.
PANIC. Deadlock.

💡 Quick Golden Rule:

❝ With unbuffered channels, send and receive must happen simultaneously. Otherwise? You're building your own tombstone. ❞

Mini Worker Pool using Buffered Channels

Let’s look at a slightly advanced example to better our understanding about channels a bit.

You want to distribute work (like downloading files, processing images, etc.) to multiple goroutines.

Let’s build a mini worker pool:

package main

import (
    "fmt"
    "time"
)

func worker(id int, jobs <-chan int, results chan<- int) {
    for j := range jobs {
        fmt.Printf("Worker %d started job %d\n", id, j)
        time.Sleep(time.Second) // simulate heavy task
        fmt.Printf("Worker %d finished job %d\n", id, j)
        results <- j * 2 // Send result back
    }
}

func main() {
    const numJobs = 5
    jobs := make(chan int, numJobs)
    results := make(chan int, numJobs)

    // Start 3 workers
    for w := 1; w <= 3; w++ {
        go worker(w, jobs, results)
    }

    // Send jobs
    for j := 1; j <= numJobs; j++ {
        jobs <- j
    }
    close(jobs) // Closing jobs channel — no more jobs coming

    // Collect results
    for a := 1; a <= numJobs; a++ {
        fmt.Println("Result:", <-results)
    }
}

Now, this sums up the concept of concurrency. Let’s dissect this code snippet and understand what is going on?

First: About These Funky Arrows


worker(id int, jobs <-chan int, results chan<- int)

jobs <-chan int →
- Read-Only Channel for int
- This goroutine can ONLY RECEIVE from jobs.
- It cannot send into jobs.
results chan<- int →
- Write-Only Channel for int
- This goroutine can ONLY SEND into results.
- It cannot receive from results.

So why this design?

Because clean roles = fewer bugs.

Worker only picks up jobs.
Worker only pushes results.
Enforces discipline at compile-time (Go will error if you do the wrong operation).

You can't accidentally do this:


j := <-results // ❌ won't compile


jobs <- 100 // ❌ won't compile

Go catches stupidity early — true backend villain language 😈.

Now Full Code Explained Step-by-Step

1. Channels setup


const numJobs = 5
jobs := make(chan int, numJobs)
results := make(chan int, numJobs)

jobs channel: buffered to hold 5 int jobs.
results channel: buffered to hold 5 int results.

This ensures main goroutine can dump jobs fast without waiting.

2. Launch 3 workers


for w := 1; w <= 3; w++ {
    go worker(w, jobs, results)
}

Spawns 3 concurrent goroutines, each running worker.
Each worker is:
- Waiting for jobs (for j := range jobs).
- Processing job (simulated by time.Sleep(time.Second)).
- Sending result back.

Now 3 workers are ready to fight for jobs.

3. Send jobs

for j := 1; j <= numJobs; j++ {
    jobs <- j
}
close(jobs)

Main goroutine pushes 5 jobs (1, 2, 3, 4, 5) into jobs channel.
Then close(jobs) → no more new jobs coming.
This is important because for j := range jobs in workers needs to know when to stop.
After closing, for j := range jobs exits when all jobs are done.

Clean end of supply line.

4. Collect results


for a := 1; a <= numJobs; a++ {
    fmt.Println("Result:", <-results)
}

Main goroutine waits and receives 5 results.
Each result is printed out.

Results are guaranteed to come because:

Worker always pushes a result after finishing a job.
Buffer size matches number of jobs.

How The Code Flows

Timeline Visualization:


Main Thread:
  Push job 1, 2, 3, 4, 5 → jobs channel
  Close jobs channel
  Wait for 5 results → print them

Worker Goroutines(All workers run parallely, so any worker can pick any job)
  Worker1: pick job 1 → work 1s → push result //say worker 1 is available
  Worker2: pick job 2 → work 1s → push result
  Worker3: pick job 3 → work 1s → push result
  (Then next jobs 4,5 picked by available workers)

Workers share the job queue — whoever is free picks next.

Output :

Suppose you remove this line:


close(jobs)

then the worker code:


for j := range jobs {
    // do work
}

has no way of knowing when jobs are over.

Result?

After finishing all the jobs you sent (5 in this case),
The workers will be stuck waiting at <-jobs

Deadlock !

Your program will hang forever after printing the first few results.

The Select Statement

Select lets a goroutine wait on multiple communication operations at once.

The first one that's ready, wins.

Basic Syntax

select {
case val := <-chan1:
    fmt.Println("received", val, "from chan1")
case chan2 <- 10:
    fmt.Println("sent 10 to chan2")
default:
    fmt.Println("no communication ready")
}

Looks suspiciously like a switch, right?

BUT it operates on channels only.

Key Rules of Select

First ready case wins.
If multiple are ready, it picks one randomly (fairness).
If none are ready, and no default, it blocks. (stuck)
If default exists, and nothing is ready, default runs instantly.

Real Example (multiple receives)

package main

import (
    "fmt"
    "time"
)

func main() {
    c1 := make(chan string)
    c2 := make(chan string)

    // Sender goroutines
    go func() {
        time.Sleep(1 * time.Second)
        c1 <- "from c1"
    }()
    go func() {
        time.Sleep(2 * time.Second)
        c2 <- "from c2"
    }()

    // Select to receive
    for i := 0; i < 2; i++ {
        select {
        case msg1 := <-c1:
            fmt.Println("Received:", msg1)
        case msg2 := <-c2:
            fmt.Println("Received:", msg2)
        }
    }
}

Output:

Received: from c1
Received: from c2

First we receive from c1, then c2, depending on who is ready.

Big Brain Situations where `select` is 🔥:

You want to listen on multiple channels without blocking on just one.
You want timeouts (can mix with time.After).
You want cancel listening (mix with context).

Scenario: Implementing Timeout and Cancellation with Select

In real-world systems, you’ll often need to:

Timeout operations after a certain period.
Cancel operations if they take too long or aren't needed anymore.

We’ll combine select, time.After, and a cancel mechanism into one delicious, high-performance, non-blocking recipe. Let's cook up a system where multiple tasks are being executed concurrently, but if they take too long, they are timed out.

time.After is a built-in Go function from the time package that returns a channel which will receive the current time after a specified duration has passed.

Scenario Breakdown:

Task Processing: We’ll simulate two tasks, each with a random duration.
Timeout: We’ll set a timeout that forces cancellation after a fixed amount of time.
Cancel: If either task finishes early, we should cancel the other one.

Code: Timeout + Cancellation Example

package main

import (
    "fmt"
    "time"
    "math/rand"
)

func longRunningTask(taskId int, cancelCh <-chan struct{}, doneCh chan<- int) {
    // Randomize task duration (between 1 and 5 seconds)
    duration := time.Duration(rand.Intn(5)+1) * time.Second
    fmt.Printf("Task %d will take %v to complete\n", taskId, duration)

    // Simulate work
    select {
    case <-time.After(duration):
        // Task completed successfully
        fmt.Printf("Task %d completed\n", taskId)
        doneCh <- taskId
    case <-cancelCh:
        // Task was canceled
        fmt.Printf("Task %d was canceled\n", taskId)
    }
}

func main() {
    // Create channels
    doneCh := make(chan int, 2)   // To receive completed tasks
    cancelCh := make(chan struct{}) // For cancel signal

    // Start two long-running tasks
    go longRunningTask(1, cancelCh, doneCh)
    go longRunningTask(2, cancelCh, doneCh)

    // Timeout duration for all tasks
    timeout := 3 * time.Second

    // Select to handle timeout or task completion
    select {
    case taskId := <-doneCh:
        fmt.Printf("Task %d completed before timeout\n", taskId)
    case <-time.After(timeout):
        // Timeout occurred, cancel remaining tasks
        fmt.Println("Timeout! Cancelling remaining tasks...")
        close(cancelCh)
    }

    // Collect remaining task results
    for i := 0; i < 2; i++ {
        select {
        case taskId := <-doneCh:
            fmt.Printf("Task %d finished\n", taskId)
        case <-time.After(timeout):
            fmt.Println("Another timeout!")
        }
    }
}

Explanation of How This Works

Task Simulation:
- Each task runs in its own goroutine.
- We use time.After(duration) to simulate the task taking random time (between 1 and 5 seconds).
Timeout Mechanism:
- We set a global timeout of 3 seconds using time.After(timeout).
- The select blocks until either a task completes or the timeout is reached.
- If the timeout occurs, we send a cancellation signal to the cancelCh channel.
Cancellation Logic:
- If the task is still running when the timeout occurs, the cancelCh channel sends a signal, and the task checks for that signal to abort its operation.
Graceful Handling of Remaining Tasks:
- After sending the cancellation signal, the program ensures any task that finished afterward is properly handled by waiting for them to complete (or timeout if necessary).

Output Example:

Let’s assume the first task finishes at 2 seconds, and the second one is still running when the timeout hits.

Task 1 will take 2s to complete
Task 2 will take 4s to complete
Task 1 completed
Timeout! Cancelling remaining tasks...
Task 2 was canceled
Task 1 finished

What's the Takeaway?

Unblocking and timeout are now easy to manage thanks to select.
select lets you handle multiple things happening at once, all while keeping the code clean and organized.
Deadlock avoidance: select helps you avoid race conditions or deadlocks because of its non-blocking nature (unless you specifically want blocking with default).

WaitGroups and Mutexes

In Go, WaitGroups and Mutexes are two of the most commonly used synchronization primitives. They are part of the sync package and help manage concurrency by ensuring proper synchronization between multiple goroutines.\

1. WaitGroups in Go

What is a WaitGroup?

A WaitGroup is used to wait for a collection of goroutines to finish executing. It provides a way for a program to synchronize multiple goroutines and wait until all of them are done before proceeding.

Basic Concept

Add(n): Specifies the number of goroutines that the WaitGroup should wait for.
Done(): Called by each goroutine when it finishes. It decrements the counter in the WaitGroup.
Wait(): Blocks until the counter in the WaitGroup reaches zero, i.e., until all the goroutines have finished their tasks.

Let’s start with a simple example that uses a WaitGroup to wait for multiple goroutines to finish.

package main

import (
    "fmt"
    "sync"
    "time"
)

func worker(id int, wg *sync.WaitGroup) {
    defer wg.Done() // Decrements the counter when the goroutine finishes
    fmt.Printf("Worker %d started\n", id)
    time.Sleep(time.Second) // Simulate work
    fmt.Printf("Worker %d finished\n", id)
}

func main() {
    var wg sync.WaitGroup

    // Launching 3 worker goroutines
    for i := 1; i <= 3; i++ {
        wg.Add(1) // Increment the counter for each goroutine
        go worker(i, &wg)
    }

    // Wait for all goroutines to finish
    wg.Wait()
    fmt.Println("All workers finished")
}

Explanation:

The main function launches 3 goroutines, each calling the worker function.
The WaitGroup counter is incremented by calling wg.Add(1) for each goroutine.
Each worker calls wg.Done() when it finishes its task.
The main function calls wg.Wait() to block until all workers have finished.

Flow:

Three workers are launched in parallel.
The main function waits for all workers to finish by calling wg.Wait().
When all workers finish, the program prints "All workers finished".

Advanced Example: WaitGroup with Error Handling

In this advanced example, we'll use a WaitGroup in combination with channels to handle errors and synchronize tasks.

package main

import (
    "fmt"
    "sync"
    "time"
)

func fetchData(url string, wg *sync.WaitGroup, resultChan chan<- string, errorChan chan<- error) {
    defer wg.Done() // Decrements the counter when the goroutine finishes

    // Simulating data fetching
    fmt.Printf("Fetching data from %s...\n", url)
    time.Sleep(time.Second * 2) // Simulate download delay

    // Simulate error for one URL
    if url == "https://badsite.com" {
        errorChan <- fmt.Errorf("failed to fetch %s", url)
        return
    }

    resultChan <- fmt.Sprintf("Successfully fetched data from %s", url)
}

func main() {
    var wg sync.WaitGroup
    resultChan := make(chan string, 3)
    errorChan := make(chan error, 3)

    urls := []string{"https://example.com", "https://goodsite.com", "https://badsite.com"}

    // Start fetching data concurrently
    for _, url := range urls {
        wg.Add(1)
        go fetchData(url, &wg, resultChan, errorChan)
    }

    // Wait for all goroutines to finish
    go func() {
        wg.Wait()
        close(resultChan) // Close resultChan once all goroutines are done
        close(errorChan)  // Close errorChan once all goroutines are done
    }()

    // Collect results and errors
    for {
        select {
        case result, ok := <-resultChan:
            if ok {
                fmt.Println(result)
            }
        case err, ok := <-errorChan:
            if ok {
                fmt.Println("Error:", err)
            }
        }

        if len(resultChan) == 0 && len(errorChan) == 0 {
            break
        }
    }

    fmt.Println("All operations completed.")
}

Explanation:

Multiple goroutines fetch data concurrently, and the WaitGroup waits for all of them to finish.
Errors are sent through the errorChan while successful results are sent to the resultChan.
After all goroutines finish, the channels are closed, and the results and errors are collected using a select statement.

2. Mutexes in Go

What is a Mutex?

A Mutex (short for mutual exclusion) is a synchronization primitive used to ensure that only one goroutine can access a critical section (shared resource) at a time. It prevents race conditions and ensures that the shared data is accessed safely by multiple goroutines.

A Mutex is used to lock and unlock access to critical sections of the code. Only one goroutine can hold the lock at any given time, while others will have to wait for the lock to be released.

Basic Concept

Lock(): Acquires the lock, ensuring exclusive access to the critical section.
Unlock(): Releases the lock, allowing other goroutines to acquire it.

Example: Mutex with Shared Resource Access

Let’s consider a scenario where multiple goroutines are concurrently updating a shared map. We'll use a Mutex to ensure that only one goroutine accesses the map at a time.

package main

import (
    "fmt"
    "sync"
    "time"
)

var sharedMap = make(map[string]int)
var mu sync.Mutex

func updateMap(key string, value int, wg *sync.WaitGroup) {
    defer wg.Done()

    mu.Lock() // Lock the map before updating
    defer mu.Unlock()

    // Simulate some work with a sleep
    time.Sleep(time.Millisecond * 100)
    sharedMap[key] = value
    fmt.Printf("Updated %s to %d\n", key, value)
}

func main() {
    var wg sync.WaitGroup

    // Launch 5 goroutines to update the shared map
    for i := 1; i <= 5; i++ {
        wg.Add(1)
        go updateMap(fmt.Sprintf("key%d", i), i*10, &wg)
    }

    wg.Wait()

    // Print the final state of the shared map
    fmt.Println("Final sharedMap:", sharedMap)
}

Explanation:

Multiple goroutines are launched to update the shared map concurrently.
The Mutex ensures that only one goroutine can access the map at a time to avoid race conditions.
After all goroutines finish, the final state of the map is printed.

Key Takeaways:

WaitGroups are used to wait for a collection of goroutines to finish before continuing execution.
Mutexes are used to ensure mutual exclusion when accessing shared resources, preventing race conditions.
Both WaitGroups and Mutexes are essential for building safe, concurrent programs in Go.

The Context Package

The context package allows you to manage concurrent operations cleanly and safely in Go.

The context package allows you to:

Control cancellation of goroutines.
Set timeouts and deadlines for operations.
Pass request-scoped values across API boundaries and between goroutines

Basically, context.Context works as connecting parts (signal mechanisms) between goroutines.

Instead of letting goroutines run indefinitely, you can tell them:

"Hey, it's time to stop working."

Core Functions in `context`

Function	Purpose
`context.Background()`	Returns a non-nil, empty context. Used as the root for other contexts.
`context.TODO()`	Placeholder context when you're unsure what to use.
`context.WithCancel(parent)`	Creates a child context that can be manually canceled.
`context.WithTimeout(parent, duration)`	Creates a child context that cancels itself after a specified duration.
`context.WithDeadline(parent, time)`	Similar to timeout but tied to an absolute point in time.
`context.WithValue(parent, key, value)`	Allows passing key-value pairs through context trees.

Why Is `context` Important for Concurrency?

Avoids Goroutine Leaks: Without cancellation signals, goroutines can keep running forever, causing memory bloat.
Couples Work to Lifecycle: Tie goroutines to the lifetime of a request or operation.
Handles Timeouts Gracefully: Important for networked services, APIs, and time-sensitive tasks.
Propagates Metadata: Pass user authentication, trace IDs, etc., neatly through multiple layers.

Example 1: Manually Canceling a Goroutine

package main

import (
    "context"
    "fmt"
    "time"
)

func worker(ctx context.Context) {
    for {
        select {
        case <-ctx.Done():
            fmt.Println("Worker: context canceled, exiting...")
            return
        default:
            fmt.Println("Worker: working...")
            time.Sleep(500 * time.Millisecond)
        }
    }
}

func main() {
    ctx, cancel := context.WithCancel(context.Background())

    go worker(ctx)

    time.Sleep(2 * time.Second)
    fmt.Println("Main: canceling context now")
    cancel()

    time.Sleep(1 * time.Second) // Give time for goroutine to finish
}

Explanation:

context.WithCancel returns a new context and a cancel function.
When cancel() is called, the worker receives the signal and exits cleanly.

Example 2: Context with Timeout

package main

import (
    "context"
    "fmt"
    "time"
)

func worker(ctx context.Context) {
    for {
        select {
        case <-ctx.Done():
            fmt.Println("Worker: timeout reached, exiting...")
            return
        default:
            fmt.Println("Worker: working...")
            time.Sleep(500 * time.Millisecond)
        }
    }
}

func main() {
    ctx, cancel := context.WithTimeout(context.Background(), 3*time.Second)
    defer cancel()

    go worker(ctx)

    time.Sleep(5 * time.Second)
}

Explanation:

After 3 seconds, the context automatically cancels itself.
No manual cancel() call is necessary to enforce the timeout (though defer cancel() is good practice).

Passing Metadata Through Context

You can use context.WithValue to attach metadata to a context. This is very useful for passing things like authentication tokens, trace IDs, or user-specific information through call chains.

package main

import (
    "context"
    "fmt"
)

func processRequest(ctx context.Context) {
    userID := ctx.Value("userID")
    if userID != nil {
        fmt.Println("Processing request for user:", userID)
    } else {
        fmt.Println("No user ID found in context")
    }
}

func main() {
    ctx := context.WithValue(context.Background(), "userID", "12345")

    processRequest(ctx)
}

Important:

Only use WithValue for request-scoped data.
Avoid using context as a global bag for business logic or configuration data.

Real-World Application: HTTP Request Context

When handling HTTP requests in a server, the request itself carries a context:

func handler(w http.ResponseWriter, r *http.Request) {
    ctx := r.Context()

    select {
    case <-time.After(5 * time.Second):
        fmt.Fprintln(w, "Request completed")
    case <-ctx.Done():
        http.Error(w, "Request canceled", http.StatusRequestTimeout)
    }
}

Why it matters:

If the client disconnects, the request context is canceled automatically.
No resources wasted processing a request the client no longer cares about.

Best Practices

Always listen to ctx.Done() inside long-running goroutines.
Always cancel child contexts you create (defer cancel()) to avoid leaks.
Avoid using context.WithValue for business logic; use it strictly for transporting metadata.
Never store Contexts inside structs; pass them explicitly.

Common Concurrency Patterns in Go

Earlier , we discussed buffered channels using a simple worker example. What we didn't reveal then is that the same snippet actually showcases several important concurrency patterns commonly used in Go.

Let's revisit the code first:

package main

import (
    "fmt"
    "time"
)

func worker(id int, jobs <-chan int, results chan<- int) {
    for j := range jobs {
        fmt.Printf("Worker %d started job %d\n", id, j)
        time.Sleep(time.Second) // simulate heavy task
        fmt.Printf("Worker %d finished job %d\n", id, j)
        results <- j * 2 // Send result back
    }
}

func main() {
    const numJobs = 5
    jobs := make(chan int, numJobs)
    results := make(chan int, numJobs)

    // Start 3 workers
    for w := 1; w <= 3; w++ {
        go worker(w, jobs, results)
    }

    // Send jobs
    for j := 1; j <= numJobs; j++ {
        jobs <- j
    }
    close(jobs) // Closing jobs channel — no more jobs coming

    // Collect results
    for a := 1; a <= numJobs; a++ {
        fmt.Println("Result:", <-results)
    }
}

Now, what patterns are hidden inside?

1. Fan-out

We start multiple goroutines (workers) that are all reading from the same jobs channel. This spreads out the workload across multiple concurrent workers - hence, fanning out.

Key idea: multiple consumers (workers) pulling from one source of work (jobs).

"Got a hundred pizzas to deliver? Send three drivers instead of one."

2. Fan-in

All the workers, after finishing their work, push their results into the same results channel. This fans in the results from multiple goroutines into a single channel for collection.

Key idea: multiple producers (workers) sending to one destination (results).

"All drivers hand over their earnings at one counter."

3. Worker Pool

This pattern specifically sets up a fixed number of goroutines (3 in our case) to handle an arbitrary number of jobs. Instead of creating a new goroutine per task (which can blow up memory), we use a small, controlled pool.

Key idea: limited, reusable goroutines handle many tasks.

"You don't hire 100 chefs to make 100 pizzas — you hire 3 good ones and make it work."

4. Channel Closing Signaling Completion

Notice how we close(jobs) after sending all jobs? That closing is a crucial signal: it tells all workers "no more work is coming," allowing their for j := range jobs loops to terminate gracefully.

Key idea: close the input channel to broadcast "we're done here" to consumers.

"The bakery door closes. No new orders. Workers finish what they're doing and go home."

Why These Patterns Matter

Efficiency: Worker pools prevent spawning thousands of goroutines unnecessarily.
Scalability: Fan-out/fan-in patterns scale your app without drama.
Graceful shutdown: Closing channels ensures clean termination without leaks.

The Pipeline pattern

Observe the give code snippet carefully.

package main

import "fmt"

// Stage 1: Generate numbers
func gen(nums ...int) <-chan int {
    out := make(chan int)
    go func() {
        for _, n := range nums {
            out <- n
        }
        close(out)
    }()
    return out
}

// Stage 2: Square numbers
func square(in <-chan int) <-chan int {
    out := make(chan int)
    go func() {
        for n := range in {
            out <- n * n
        }
        close(out)
    }()
    return out
}

func main() {
    // Set up the pipeline
    nums := gen(2, 3, 4, 5)
    squares := square(nums)

    // Consume the output
    for s := range squares {
        fmt.Println(s)
    }
}

Pipeline = a series of stages connected by channels.

Each stage is independent, parallelizable, and communicates via channels.

Every stage is:

Taking input from previous channel.
Doing its job (transforming data).
Passing it to the next stage via another channel.

Concurrency behind the scenes:

Every stage (gen, square) does its work independently and asynchronously.

They are literally pushing and pulling data through channels.

Real Life Comparison:

Imagine a factory assembly line:

First guy (gen) just drops raw materials on the belt.
Second guy (square) picks them, processes them (squares them), and sends them forward.
You (main) are the final inspector printing the results.

Concurrency patterns in Go are not limited to these. If you wish to dive even deeper, refer this repo.

Common pitfalls with concurrency in Go

Here are some common pitfalls we encounter while dealing with concurrency in Go. Several of them have been explained in detail above.

Deadlocks

→ Goroutines stuck waiting forever (e.g., reading from an empty channel).
Goroutine Leaks

→ Forgetting to close channels or signal exit - goroutines hang around eating RAM .
Race Conditions

→ Two goroutines modifying shared data without sync- data gets corrupted (use mutex).
Improper Channel Closing

→ Only senders should close channels.

→ Receivers never close it -that's how you get panics.
Forgot to Drain Channels

→ Senders block because nobody's receiving.
Overusing Goroutines

→ Spawning millions of goroutines casually → memory death spiral.
Blocking Operations Inside Goroutines

→ One sleepy goroutine = whole system choke if not handled with timeouts/select.
Misusing Buffered Channels

→ Thinking buffers are infinite. (Spoiler: they're not.)

→ Full buffer = blocked sender.
Panic Inside Goroutines

→ If a goroutine panics and you don't recover it → stealth crashes.
No Context Cancellation

→ Forgetting to pass context.Context to control goroutines.

That’s a wrap

Okay, so we are finally ending this blog. It was absolutely fun writing this.

If you made it till here — hats off to you!

We covered a lot of ground:

How Go handles concurrency so beautifully with its simple primitives.
What goroutines are, how channels let them talk, and how select lets them multitask smartly.
How to synchronize goroutines safely with WaitGroups and Mutexes.
How to control cancellations and deadlines with the Context package.
Some real-world patterns and pitfalls that every Go developer should know.

Concurrency can feel intimidating at first, but with practice, it becomes second nature — and trust me, it's one of the most rewarding things to master in Go.

If you ever find yourself stuck, just remember:

Keep it simple, make it clear who owns what, and let Go’s concurrency model do its magic.

Thanks for reading — and happy coding!

Don't Cry about Pointers anymore - Deep Dive with C, Go and Rust

Yashaswi Kumar Mishra — Wed, 16 Apr 2025 16:38:26 +0000

Why TF does this blog even exist?

Let’s be real — most of us treat pointers like radioactive waste. Let's ignore those dreadful stories of childhood when pointers were introduced to you like a BIG demon. But deep down, the truth is:

Memory is power. And pointers? Pointers are your keys to the throne.

If you’ve ever:

Debugged a nasty segfault while chugging a huge load of coffee,
Seen your app lag thanks to the garbage collector,
Or tried passing huge structs around to make your code modular …

Then you’ve already felt the need for pointers.

This isn’t just a blog. This is a manifesto. We’re gonna break the myths, smash the fear, and teach you how to own memory like an absolute menace — across C (just for understanding how pointers work), Go, and Rust.

By the end of this read, you’ll understand:

What pointers really are
Why they matter in modern backend systems
How Go gives you a chill but powerful pointer experience
How Rust forces you into greatness with ownership and borrowing

Let’s cut the fluff and jump right in — welcome to the deep dive.

Basics: Pointers and How They Work (I'm Hoping Y'all Know C)

A pointer just refers to another value. That's it. It is itself a variable that stores the address of another variable(this variable can be a pointer too).
Let's see a quick code snippet in C to recap pointer basics.

#include <stdio.h>

int main() {
    int x = 42;
    int *p = &x; // p points to the address of x

    printf("Value of x: %d\n", x);
    printf("Pointer p: %p\n", p);       // prints address of x
    printf("Value at *p: %d\n", *p);     // dereferencing p, gives 42

    *p = 99; // update x through the pointer
    printf("Updated x: %d\n", x);        // now x is 99

    return 0;
}

Key Concepts :

*p → Dereference the pointer (get the value it points to)
&x → Address-of operator (get the memory address of x)
int *p → Declare a pointer to an integer

This is where C shows you its true colors: dope performance, minimal overhead — but a single wrong move and it’s undefined behavior land. Manual memory management is both a blessing and a curse.

Let's quickly take a look over other pointer shenanigans in C :

1. Pointers and Arrays:

int arr[3] = {10, 20, 30};
int *ptr = arr; // same as &arr[0]
printf("%d\n", *(ptr + 1)); // prints 20

Arrays decay into pointers — the name of an array is basically a pointer to its first element.

2. Pointer to Pointer:

int x = 5;
int *p = &x;
int **pp = &p;
printf("%d\n", **pp); // double dereference

A pointer to a pointer holds the address of another pointer. Useful in dynamic memory, linked lists, and certain APIs.

3. Function Pointers:

void greet() { printf("Hello!\n"); }
void (*funcPtr)() = greet;
funcPtr(); // calls greet()

Functions live in memory too, and you can point to them! That’s how callbacks and plugin systems are built.

4. Void Pointers:

void *vp;
int a = 7;
vp = &a;
printf("%d\n", *(int *)vp);

Generic pointers that can point to any type. But you need to cast them back before dereferencing.

Wait! There is also pointer arithmetic in C.

In C, pointers aren’t just addresses — they’re math-capable beasts. You can add or subtract integers to pointers to move across array elements (ptr + 1 points to the next element).
But beware: mess up the math, and you're one *(ptr + 9999) away from summoning a segfault demon.(I swear ChatGPT wrote this!)

int arr[5] = {1, 2, 3, 4, 5};
int *ptr = arr;

printf("%d\n", *(ptr + 2)); // prints 3

I just skimmed through the surface of pointers in C. You definitely need to go into a lot more detail. If you want to work with pointers in C, burn this video into your head.

In C, you need to manage memory manually using malloc or calloc and use the pointers very carefully and cautiously. While you have all the control, things can go south easily, even if you miss a little something.

Before diving into how we handle pointer in Go or Rust, let us look at some common concepts in pointers.

Pointer 101

Pointer Declaration

We already saw how to do this in C, and it is pretty easy in Go and Rust too. These code snippets will help you grasp it.

func main() {
    a := 10
    var p *int = &a
    fmt.Println(*p) // dereferencing
}

fn main() {
    let a = 10;
    let p = &a; // immutable reference
    println!("{}", *p);
}

Dynamic Memory Allocation

Even here, the code snippets will help you. The pointers are allocated memory in heap.

int *p = (int*) malloc(sizeof(int));
*p = 42;

p := new(int)
*p = 42

let b = Box::new(42); // heap allocation

Stack vs Heap Allocation in Pointers

When working with pointers in low-level languages like C, Go, and Rust, understanding how memory is allocated — either on the stack or the heap — is non-negotiable if you don’t want your programs to randomly explode (aka segfault).

Stack Allocation

Fast and automatically managed.
Memory is allocated when a function is called and freed when it returns.
Ideal for short-lived variables and function-local data.

C Example:

int main() {
    int x = 42;        // x is on the stack
    int *p = &x;       // p points to stack memory
}

Warning:

Returning a pointer to a stack variable is dangerous. Once the function exits, that memory is gone.

int* bad() {
    int local = 10;
    return &local; // 🚨 Undefined behavior
}

Heap Allocation

Slower but gives manual control (malloc/free in C, Box::new in Rust).
Data lives until you explicitly free it (or a garbage collector does).
Used for long-lived or large data structures.

C Example:

int *p = malloc(sizeof(int));
*p = 99;
free(p);

Rust Example:

let b = Box::new(5); // heap allocation
println!("{}", b);  // Box implements Deref, so it acts like a pointer

Go's Sweet Spot

Go doesn’t let you explicitly allocate on heap or stack — the compiler decides based on escape analysis.

func makePointer() *int {
    x := 10
    return &x // may be heap-allocated if it escapes
}

TL;DR

Feature	Stack	Heap
Lifetime	Until function returns	Until explicitly freed
Speed	Fast	Slower
Managed by	Compiler	Developer / GC
Use case	Small, short-lived data	Large, persistent data
Typical Bug	Use-after-return	Memory leaks, double free

Stack = speed.
Heap = flexibility.
Misunderstanding either = pain.

Pass By Value v/s Pass By Reference

What is Pass-by-Value?

A copy of the variable is passed to the function.
The original variable does not change.

What is Pass-by-Reference?

A reference or pointer is passed.
The function can modify the original variable.

Pass-by-Value (default in C)

void modify(int a) {
    a = 100;
}

int main() {
    int x = 10;
    modify(x);
    printf("%d\n", x); // prints 10
}

Pass-by-Reference using Pointer

void modify(int *a) {
    *a = 100;
}

int main() {
    int x = 10;
    modify(&x);
    printf("%d\n", x); // prints 100
}

Pass-by-Value (default in Go)

func modify(a int) {
    a = 100
}

func main() {
    x := 10
    modify(x)
    fmt.Println(x) // prints 10
}

Pass-by-Reference using Pointers

func modify(a *int) {
    *a = 100
}

func main() {
    x := 10
    modify(&x)
    fmt.Println(x) // prints 100
}

Even slices, maps, channels in Go are passed by value, but since the value contains an internal pointer, they often behave like references.

Rust always passes by value, but with borrowing (&) or mutable borrowing (&mut), you simulate reference behavior safely.

Pass-by-Value (moves or copies the value)

fn modify(mut a: i32) {
    a = 100;
}

fn main() {
    let x = 10;
    modify(x);
    println!("{}", x); // prints 10
}

Pass-by-Reference (Borrowing)

fn modify(a: &mut i32) {
    *a = 100;
}

fn main() {
    let mut x = 10;
    modify(&mut x);
    println!("{}", x); // prints 100
}

Rust ensures no data races or dangling pointers using its ownership model and borrow checker. (We will understand more of this later)

Side-by-Side Summary

Language	Default Behavior	Reference Support	Mutable Ref Support	Safety Mechanism
C	Pass-by-Value	✅ Yes (manual `*`)	✅ Yes	❌ Manual, unsafe
Go	Pass-by-Value	✅ Yes (with `*T`)	✅ Yes	❌ Garbage-collected
Rust	Pass-by-Value	✅ Yes (`&T`)	✅ Yes (`&mut T`)	✅ Ownership & Borrowing

Key Facts and Gotchas — Pointers in C, Go, and Rust

C

You can:

Do pointer arithmetic
Use null pointers
Cast between pointer types
Create pointer-to-pointer
Manually manage heap (malloc/free)
Point to functions

You must handle:

No safety — undefined behavior is easy
Manual memory management (malloc/free)
Dangling pointers, memory leaks, buffer overflows
Segfaults due to wild pointers

Go

You can:

Use pointers to access and modify values
Pass pointers to functions
Allocate memory with new() or make() (for slices, maps, etc.)

But you cannot:

Do pointer arithmetic
Create pointer-to-pointer (**int)
Manually free memory (garbage collected)
Have nil-check safety for all types (you can nil dereference)
Take pointer to literal (p := &10 is invalid)

🔹 Example:

arr := []int{1, 2, 3}
// Can't do: ptr := &arr[0] + 1  ❌
// Use indexing instead
fmt.Println(arr[1])

As said earlier, Go passes structs/slices by value, but since slices contain an internal pointer, they behave “reference-like.”

See :

func modify(s []int) {
    s[0] = 999
}

func main() {
    a := []int{1, 2, 3}
    modify(a)
    fmt.Println(a) // [999 2 3] - gotcha!
}

Rust

You can:

Use references (&T, &mut T) safely
Use Box, Rc, Arc for heap allocation and shared ownership
Use raw pointers in unsafe blocks (*const, *mut)
Use Option<T> instead of null

You cannot:

Have null pointers (None is used instead)
Mutably borrow more than once simultaneously (borrow checker)
Dereference or do pointer arithmetic outside unsafe
Leak memory unless you std::mem::forget or Box::leak

Example:

let x = Some(10);
let y: Option<&i32> = None; // instead of a null pointer

let a = [1, 2, 3];
let p = a.as_ptr();
unsafe {
    println!("{}", *p.add(1)); // Pointer arithmetic in unsafe block
}

TL;DR: Cheat Sheet

Feature / Language	C	Go	Rust
Null Pointers	✅ `NULL`	✅ `nil`	❌ Use `Option<T>`
Pointer Arithmetic	✅ Yes	❌ No	✅ Yes (unsafe only)
Pointer-to-Pointer	✅ Yes	❌ No	✅ Yes (`&&T`)
Automatic Memory Mgmt	❌ No	✅ GC	✅ Ownership model
Dangling Pointer Safety	❌ No	❌ No	✅ Checked at compile
Struct Ref Mutability	✅ Manual	✅ Pointers	✅ Borrow system
Ref Counted Shared Ptr	❌ Manual	✅ Maps etc.	✅ `Rc`, `Arc`
Manual Free	✅ Yes	❌ No	✅ `Drop` trait
Function Pointers	✅ Yes	✅ Yes	✅ `fn`, `Fn`, `FnMut`

Garbage Collection in Go v/s Ownership in Rust

Choose your weapon: Lazy janitor vs Memory tyrant.

Go's Garbage Collection (GC) — The Chill Roommate

Go takes a comfy, high-level approach. You allocate memory, use it, and forget it — the Go runtime’s GC cleans it up like a silent ninja.

How it Works:

Go uses a concurrent, tri-color mark-and-sweep garbage collector.
It pauses briefly, walks through live objects, marks them, and frees unmarked (unreachable) memory.
GC runs in the background — you don’t control it directly.

Example:

type User struct {
    name string
}

func main() {
    u := &User{name: "pixie"}
    fmt.Println(u.name) // GC takes care of `u` when it's no longer used
}

Pros:

Zero manual memory management.
Great for productivity, safety, and fast dev cycles.
No need to think about free() or memory leaks most of the time.

Cons:

GC pauses = potential latency spikes.
GC can kick in at random times = non-deterministic behavior.
Less control over performance-sensitive workloads (e.g., real-time systems).

GC is like Uber Eats: you chill, your trash gets picked up eventually — but not always when you want it.

Rust’s Ownership Model — The Memory Warlord’s Code

Rust said:

“Why GC when you can just never make a memory mistake?”

Instead of garbage collection, Rust uses ownership + lifetimes + borrowing to statically guarantee memory safety — at compile time.

Core Concepts:

Ownership: Every value has a single owner.
Borrowing: You can lend references without transferring ownership.
Lifetimes: Ensures references are valid as long as they're used.

fn main() {
    let s = String::from("hello"); // s owns the String
    takes_ownership(s);            // ownership moved, s is now invalid
    // println!("{}", s);          // ❌ Compile error

    let x = 5;
    makes_copy(x);                 // i32 implements Copy, x still valid
}

fn takes_ownership(s: String) {
    println!("{}", s);
}

fn makes_copy(x: i32) {
    println!("{}", x);
}

Pros:

Zero runtime cost (no GC!)
Compile-time guarantees = no segfaults, no leaks, no data races.
Absolute control — systems level performance, with high-level safety.

Cons:

Steeper learning curve (borrow checker will roast you alive at first).
More verbose code with &, &mut, lifetimes.
Need to think deeply about lifetimes and ownership flows.

Rust is like being Batman: no sidekicks, no help — just you and your gear. But once you master it? You're unstoppable.

TL;DR — Go GC vs Rust Ownership

Feature	Go (Garbage Collected)	Rust (Ownership Model)
Memory Management	Automatic via GC	Compile-time via Ownership
Control	Minimal	Complete
Performance Overhead	GC pauses, unpredictable latency	Near-zero runtime cost
Memory Safety	Mostly safe, but nil pointers exist	Guaranteed via borrow checker
Ease of Use	Easy, chill	Steep learning curve, then power
Real-time Systems	❌ Not ideal	✅ Excellent fit

Some Wisdom Drop

If you want developer productivity with decent performance: Go.
If you want maximum performance with zero runtime overhead and are willing to tame the beast: Rust.
Want to be a backend god-tier villain? Learn both. Use the right tool for the right kill.

Chaos to Calm - Pointers in Go

So now that you've stared into the void (pointer pun intended) with C — raw, dangerous, and obviously a double edged sword — let's take a deep breath and walk into a language that hands you pointers… but without the existential crisis.

Go doesn’t give you pointer arithmetic. It won’t let you shoot yourself in the foot (most of the time). But it does give you:

Direct memory referencing with & and *
Lightweight object sharing via pointer receivers
Safe heap allocation using new()
And just enough low-level vibes to keep your inner villain happy

By the above examples , it must be pretty clear how pointers work in go, let's jump to working with structs and pointers in go.

Pointers and Structs in Go

Structs in Go are value types by default. This means assigning a struct or passing it to a function copies the whole thing — which could suck for performance.

When to Use Pointers with Structs

When your struct is large and copying is expensive
When you want to mutate the original struct in a function
When you want to avoid frequent allocations in tight loops

type Villain struct {
    Name  string
    PowerLevel int
}

func boost(v *Villain) {
    v.PowerLevel += 100
}

func main() {
    v := Villain{"Yashaswi", 9000}
    boost(&v)
    fmt.Println(v.PowerLevel) // 9100
}

When NOT to Use Pointers

If your struct is small (e.g., a few fields) and used immutably
If you don’t need to share/mutate state across functions
If you're creating short-lived structs inside a loop — prefer values to help stack allocation

Rule of Thumb

If it's less than or equal to 3 words (e.g., 24 bytes), pass by value.
If you need shared/mutable state, use pointers.
For methods, use pointer receivers if the method needs to mutate or avoid copying.

type Engine struct {
    HorsePower int
}

func (e Engine) PrintHP() { // value receiver
    fmt.Println(e.HorsePower)
}

func (e *Engine) Boost() { // pointer receiver
    e.HorsePower += 100
}

Memory Best Practices in Go

Avoid unnecessary allocations

Use value receivers where mutation isn't needed.
Preallocate slices and maps where possible.

Escape analysis matters

If a variable escapes to the heap (vs staying on the stack), it increases GC pressure.
Use go build -gcflags=-m to check if variables escape.

Use sync.Pool for reusable memory

Great for reducing GC load in high-performance apps.

Don't fear pointers

They’re safe, controlled, and let you avoid copying large structs.

Mind the zero values

In Go, everything has a zero value. Don’t waste allocations initializing something that’s already zero. ### ⚠️ What Go Doesn’t Let You Do (And That’s Okay)
- No manual malloc/free.
- No pointer arithmetic.
- No direct memory manipulation (you’ll need unsafe for that).

But hey, that’s the point — Go protects you from footguns so you can ship faster without being reckless. You're still working with memory, just with guardrails.

That's a wrap

And there you have it — the journey from raw pointers to Go vs. Rust. We’ve covered how memory is managed at its most basic level, the stack vs. heap debate, and the power (and peril) of manual memory management.

In the world of Rust, we’ve seen how the ownership system isn’t just a feature — it’s a whole new way to think about memory safety. With borrowing and lifetimes, Rust enforces a discipline that forces you to think ahead, leaving no room for memory leaks or dangling pointers. It’s a challenge, but one that builds robust, zero-cost abstractions.

In Go, we saw how things are more laid-back with garbage collection and pointers that you only really need when performance or interop is key. Go’s approach has its benefits, but it doesn’t give you the same fine-grained control as Rust.

Remember: pointers are a tool. Whether you’re dealing with them in Go or Rust, you control the power — but with great power comes great responsibility. Master these concepts, and you’ll be writing software that’s not just efficient, but invincible.

Also, we’ve only scratched the surface with Rust. In the coming one, we’ll dive into the real villain territory of Rust’s ownership, smart pointers, lifetimes, threads, and channels. Are you ready to level up your backend skills?

Until then, keep your pointers sharp and your memory clean.

Learn gRPC COMPLETELY by building a mini Exam Service in GoLang

Yashaswi Kumar Mishra — Wed, 16 Apr 2025 12:35:39 +0000

Modern systems like Slack, Netfix, or even Kubernetes handle real-time communication very efficiently. Their large and bulky backends are divided into microservices. Have you ever wondered how these services communicate among themselves? Chances are that gRPC is behind the magic.

In this blog, let us try to understand how can we implement a minimal Exam Service using gRPC in GoLang covering all types of RPCs: Unary, Server Streaming, Client Streaming, and Bidirectional Streaming.

If you already know what you are doing and just want the code, here is the repo.

Before that let us actually try to undestand some basic concepts and simplify the jargon.

Get ready for the complete gRPC guide.

RPC! What's That?

RPC or Remote Procedural Calls may sound like a fancy term but it's just calling functions. Yes! But on other machine.

Imagine you're the client and you want to ask a server: "Hey, what's the score for student ID 42?" RPC lets you do that as if you were calling a normal function, even though that function lives on a totally different system.

Classic RPC systems came before gRPC, but they had a lot of baggage:

Inconsistent data formats: No standard serialization; some used XML, some custom formats — all bulky and slow.
Lack of streaming: Supporting real-time or long-lived communication was hard or impossible.
Limited language support: Often tied to specific ecosystems (like Java RMI or CORBA).
Poor tooling: Little to no auto-codegen — developers had to write boilerplate by hand.
Security was manual: No built-in encryption; you had to manage TLS yourself.
Hard to scale: No multiplexing, no connection reuse — performance didn’t hold up under modern loads.

In contrast, gRPC fixes these issues out of the box — faster, safer, and way easier to use.

gRPC - Google's RPC

gRPC is an open-source RPC framework developed by Google. It's designed to make inter-service communication:

Fast (thanks to HTTP/2)
Type-safe (via Protocol Buffers)
Streaming-friendly (supports real-time communication)

Okay. What the hell is HTTP/2?

gRPC uses HTTP/2 under the hood. Compared to HTTP/1.1:

It supports multiplexing— multiple requests on one connection.
Enables bidirectional streaming — both client and server can talk at the same time.
Built-in header compression — faster data transfer.

This is how gRPC achieves low-latency real-time communication.

gRPC vs REST

Feature	REST	gRPC
Data Format	JSON / XML	Protocol Buffers
Streaming	Rare / manual	Built-in
Transport	HTTP 1.1	HTTP/2
Speed	Verbose	Compact & fast
Developer Experience	Verbose, manual docs	Auto-generated code

Types of RPC

gRPC supports four types of communication between the client and server. Let’s walk through each of them.

1. Unary RPC (Simple Request-Response)

Client sends one request → Server sends one response.

Think of it like a regular function call.

Example:

"Get me the marks of student with ID 42."

rpc GetMarks(StudentRequest) returns (MarksResponse);

This is the most common type — perfect for CRUD-style operations.

2. Server Streaming RPC

Client sends one request → Server sends a stream of responses.

The server keeps pushing multiple responses over time.

Example:

"Give me the list of all exam results for semester 6."

rpc StreamSemesterResults(SemesterRequest) returns (stream MarksResponse);

Useful when the server has a lot of data to send — you get results as they’re ready.

3. Client Streaming RPC

Client sends a stream of requests → Server sends one final response.

The client pushes a batch of data, and the server replies after processing it all.

Example:

"Here’s the attendance report of all students in bulk — store it."

rpc UploadAttendance(stream AttendanceEntry) returns (UploadStatus);

Great for sending logs, metrics, or bulk uploads.

4. Bidirectional Streaming RPC (Bidi)

Client and server stream data to each other concurrently.

Like a real-time chat where both can talk and listen at the same time.

Example:

"Start a live quiz session — the client sends answers, the server sends questions and scores in real time."

rpc LiveQuiz(stream QuizMessage) returns (stream QuizMessage);

This is where gRPC really shines. Real-time multiplayer games, collaborative tools, live dashboards — it’s all possible.

The Last Bit of Jargon - Protocol Buffers

Protocol Buffers or Protobuf is a language-neutral, platform-neutral mechanism for serializing structured data — developed by Google.

It is better than JSON

JSON is human-readable but bulky.
Protobuf is compact and faster to serialize/deserialize.
JSON needs to parse field names like "student_id" every time. Protobuf uses tags (like 1, 2) behind the scenes for faster lookup.

How it works?

You define your schema using a .proto file.
The Protobuf compiler (protoc) generates code for your language (e.g., Go, Python, Java).
That code provides you with methods to encode and decode messages efficiently.

Example :

syntax = "proto3";

message StudentRequest {
  string student_id = 1;
}

Let's Start Building

Bootstraping

First of all let us bootstrap the project in go.

mkdir grpc_exam
cd grpc_exam
go mod init github.com/pixperk/grpc_exam

We will install the necessary dependencies (Go plugins for protoc)

go install google.golang.org/protobuf/cmd/protoc-gen-go@latest
go install google.golang.org/grpc/cmd/protoc-gen-go-grpc@latest

If you do not have the protoc compiler installed, see the installion guide here.

Let's ready our folder structure

├── client/
│   ├── clients/
│   │   ├── unary.go
│   │   ├── server_stream.go
│   │   ├── client_stream.go
│   │   ├── bi_stream.go
│   └── main.go
├── proto/
│   ├── exam.proto
│   └── generated/exampb/
│       ├── exam.pb.go
│       └── exam_grpc.pb.go
├── server/
│   ├── main.go
│   └── servers/
│       ├── unary.go
│       ├── server_stream.go
│       ├── client_stream.go
│       ├── bi_stream.go
│       └── exam_service_server.go
├── utils/
│   └── logger.go
├── go.mod
├── go.sum
└── Makefile

Let's write our makefile so that we do not have to write long commands everytime and just run make <command>

proto:
    protoc \
        --proto_path=proto \
        --go_out=proto \
        --go-grpc_out=proto \
        proto/*.proto
    @echo "Proto files generated in the 'proto' directory."

server:
    go run server/main.go

client_unary:
    go run client/main.go unary

client_server:
    go run client/main.go server

client_client:
    go run client/main.go client

client_bidi:
    go run client/main.go bidi

.PHONY: proto server client_unary client_server client_client client_bidi

The proto command is for generating go code from the proto file.

Building the unary RPC

In the exam.proto file, let us write our Exam Service.

syntax = "proto3";

package exam;

option go_package = "generated/exampb";

service ExamService {
  rpc GetExamResult(GetExamResultRequest) returns (GetExamResultResponse); //unary
}

message GetExamResultRequest {
  string student_id = 1;
  string exam_id = 2;
}

message GetExamResultResponse {
  string student_name = 1;
  string subject = 2;
  int32 marks_obtained = 3;
  int32 total_marks = 4;
  string grade = 5;
}

Now we run make proto to generate our go code.
Our code goes inside the proto/generated folder.

In the server/servers/exam_service_server.go define this

package servers

import "github.com/pixperk/grpc_exam/proto/generated/exampb"

type ExamServiceServer struct {
    exampb.UnimplementedExamServiceServer
    examData map[string]*exampb.GetExamResultResponse
}

func NewExamServiceServer() *ExamServiceServer {
    data := map[string]*exampb.GetExamResultResponse{
        "123_math101": {
            StudentName:   "John Doe",
            Subject:       "Math 101",
            MarksObtained: 95,
            TotalMarks:    100,
            Grade:         "A+",
        },
        "456_phy101": {
            StudentName:   "Jane Smith",
            Subject:       "Physics 101",
            MarksObtained: 88,
            TotalMarks:    100,
            Grade:         "A",
        },
    }

    return &ExamServiceServer{
        examData: data,
    }
}

Let's design the server and the client for this.
First let's code the main.go inside the server and the client. The main.go's inside the server and client will drive them respectively.

server/main.go :

package main

import (
    "net"

    "log/slog"

    "github.com/pixperk/grpc_exam/proto/generated/exampb"
    "github.com/pixperk/grpc_exam/server/servers"

    "github.com/pixperk/grpc_exam/utils"
    "google.golang.org/grpc"
)

func main() {
    utils.InitLogger(true)
    //Spin up a TCP Server
    lis, err := net.Listen("tcp", ":50051")
    if err != nil {
        slog.Error("failed to listen", "error", err)
    }

    //New gRPC server instance
    s := grpc.NewServer()

    //Register services
    exampb.RegisterExamServiceServer(s, servers.NewExamServiceServer())

 // Start serving gRPC requests
    if err := s.Serve(lis); err != nil {
        slog.Error("failed to serve", "error", err)
    }

}

client/main.go :

package main

import (
    "log/slog"
    "os"

    "github.com/pixperk/grpc_exam/client/clients"
    "github.com/pixperk/grpc_exam/proto/generated/exampb"
    "github.com/pixperk/grpc_exam/utils"
    "google.golang.org/grpc"
    "google.golang.org/grpc/credentials/insecure"
)

func main() {
    // Initialize logger (true = debug mode)
    utils.InitLogger(true)

    // Create a gRPC client connection to the server
    conn, err := grpc.Dial("localhost:50051", grpc.WithTransportCredentials(insecure.NewCredentials()))
    if err != nil {
        slog.Error("Failed to connect to server", "error", err)
        return
    }
    defer conn.Close()

    // Create a client for the ExamService
    client := exampb.NewExamServiceClient(conn)

    clients.Unary(client)

}

Now let's write the unary server and unary client

Unary Server (server/servers/unary.go) :

package servers

import (
    "context"
    "fmt"

    "github.com/pixperk/grpc_exam/proto/generated/exampb"
)

func (s *ExamServiceServer) GetExamResult(ctx context.Context, req *exampb.GetExamResultRequest) (*exampb.GetExamResultResponse, error) {
    key := fmt.Sprintf("%s_%s", req.StudentId, req.ExamId)
    if result, ok := s.examData[key]; ok {
        return result, nil
    } else {
        return nil, fmt.Errorf("exam result not found for student ID %s and exam ID %s", req.StudentId, req.ExamId)
    }
}

Pretty easy right?
Similarly Unary Client(client/clients/unary.go) :

package clients

import (
    "context"
    "fmt"
    "time"

    "github.com/pixperk/grpc_exam/proto/generated/exampb"
)

func Unary(client exampb.ExamServiceClient) {

    fmt.Println("Enter student ID and exam ID (e.g., 123 math101):")
    var studentID, examID string
    fmt.Scanf("%s %s", &studentID, &examID)

    ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
    defer cancel()

    resp, err := client.GetExamResult(ctx, &exampb.GetExamResultRequest{StudentId: studentID, ExamId: examID})
    if err != nil {
        fmt.Printf("Error: %v\n", err)
        return
    }

    fmt.Printf("Student Name: %s\n", resp.StudentName)
    fmt.Printf("Subject: %s\n", resp.Subject)
    fmt.Printf("Marks Obtained: %d out of %d\n", resp.MarksObtained, resp.TotalMarks)
    fmt.Printf("Grade: %s\n", resp.Grade)
    fmt.Println("Unary RPC call completed successfully.")

}

This is just basic function calls and simple go programming.

You can see the results by running make server and make client in two separate terminals.

Server Streaming and Client streaming are easy given you can handle streams in Go. Let's jump to bidirectional streaming and skim through streaming as well.

Building the Bidirectional RPC

In the exam.proto file add

rpc LiveExamQuery(stream GetExamResultRequest) returns (stream GetExamResultResponse); //bidi streaming

In the servers/bi_stream.go :

package servers

import (
    "fmt"
    "io"

    "github.com/pixperk/grpc_exam/proto/generated/exampb"
)


func (s *ExamServiceServer) LiveExamQuery(stream exampb.ExamService_LiveExamQueryServer) error {
    for {
        // Receive a stream request from the client
        req, err := stream.Recv()
        if err != nil {
            // If the client closes the stream (EOF), stop the loop gracefully
            if err == io.EOF {
                return nil
            }
            // If another error occurred, return it
            return err
        }

        key := fmt.Sprintf("%s_%s", req.StudentId, req.ExamId)

        result, ok := s.examData[key]

        // If result is not found, send a default "Not Found" response
        if !ok {
            err := stream.Send(&exampb.GetExamResultResponse{
                StudentName:   "N/A",
                Subject:       req.ExamId,
                MarksObtained: 0,
                TotalMarks:    0,
                Grade:         "Not Found",
            })
            if err != nil {
                return err // Stop on send error
            }
            continue
        }

        // If result is found, send it back to the client over the stream
        if err := stream.Send(result); err != nil {
            return err // Stop on send error
        }
    }
}

This function:

Receives multiple exam queries from the client.
Immediately responds to each with a result (if found).
Continues until the client stops sending (EOF).

Go Channels

Let's understand a bit about channels before proceeding to bi-directional client.

Go channels are like pipes used to communicate between goroutines (lightweight threads). They help synchronize data exchange safely without using mutexes.

Create a channel:

done := make(chan struct{}) // unbuffered channel of empty struct

Send to a channel:

done <- struct{}{}

Receive from a channel:

<-done

Channels block until something is sent or received, making them perfect for goroutine coordination.

Now back to bi-dir client

package clients

import (
    "bufio"
    "context"
    "fmt"
    "io"
    "log"
    "os"
    "strings"

    "github.com/pixperk/grpc_exam/proto/generated/exampb"
)

func BiDirectional(client exampb.ExamServiceClient) {
    //body
}

Let's discuss in bits about what goes on here (the function body) :

stream, err := client.LiveExamQuery(context.Background())
done := make(chan struct{})

LiveExamQuery opens a bidirectional stream with the server.
done channel is created to signal when the receiver goroutine is finished.

go func() {
        for {
            res, err := stream.Recv() //receive stream from the server
            if err != nil {
                if err == io.EOF {
                    break
                }
                log.Fatalf("Error receiving response: %v", err)
                break
            }
            fmt.Printf("🎓 %s | %s: %d/%d (%s)\n",
                res.StudentName, res.Subject, res.MarksObtained, res.TotalMarks, res.Grade)

            fmt.Print("Enter student_id and exam_id (or 'exit'): ")
        }
        close(done)

    }()

    // Initial prompt
    fmt.Print("Enter student_id and exam_id (or 'exit'): ")

This goroutine listens for server responses and prints them.
It runs in the background while the main thread handles user input.

reader := bufio.NewReader(os.Stdin)
//Send data
    for {
        line, _ := reader.ReadString('\n')
        line = strings.TrimSpace(line)
        if line == "exit" {
            stream.CloseSend()
            break
        }
        parts := strings.Fields(line)
        if len(parts) != 2 {
            fmt.Println("⚠️  Usage: <student_id> <exam_id>")
            continue
        }
        req := &exampb.GetExamResultRequest{
            StudentId: parts[0],
            ExamId:    parts[1],
        }
        if err := stream.Send(req); err != nil {
            log.Printf("send error: %v", err)
            break
        }
    }

Reads user input (student_id exam_id).
Sends each request to the server via the stream.
If exit is typed, the client closes the send stream.

<-done
fmt.Println("👋 Session ended.")

Waits for the receiver goroutine to finish using the done channel.
Ensures the program exits only after all communication is done.

Now we can test the bidirectional RPC using the make commands like above.

Wrapping Up

If you want to refer to any code or see the server or client streaming, this is the repo.

Building a gRPC-based system with both server-side and client-side streaming is a great way to explore high-performance communication between services. Throughout this project, we learned how to:

Define robust proto files and generate language-specific code
Implement server-side and client-side streaming using gRPC
Handle bidirectional communication effectively
Structure the project for maintainability and scalability

Whether you're a beginner exploring RPC concepts or someone looking to implement efficient microservices communication, I hope this project offers a helpful starting point.

Thanks for reading! Let me know if you have any questions down in the comments.

DEV Community: Yashaswi Kumar Mishra

I Implemented Amazon's Dynamo Paper in Go

First, What Even Is Dynamo?

Storage That Never Says No

Vector Clocks - Tracking Who Knows What

The Node

Consistent Hashing : Distributing Keys

Replication and Sloppy Quorum : The Availability Trick

gRPC -Making It Real

Hinted Handoff - Getting Data Back Home

Merkle Trees - Catching What Hinted Handoff Misses

Gossip Protocol - Who's Alive, Who's Dead

The Full Picture

What This Doesn't Cover

Learnings

lowkey: Building a Distributed Lock Service That Actually Works

What is lowkey?

The Fundamental Problem of Time

The CAP Theorem Choice

The Three Pillars: Leases, Fencing Tokens, and Raft

Leases

Fencing Tokens

Raft Consensus

lowkey Architecture

Layer 1: FSM (Finite State Machine)

Layer 2: Raft Consensus

Layer 3: Server (gRPC + HTTP)

Layer 4: Client SDK

Benchmarks: The Numbers

Latency Percentiles: What Users Actually Experience

Design Decisions Recap

What I Learned Building This

Try It Yourself

The Bottom Line

Writes done Right : Atomicity and Idempotency with Redis, Lua, and Go

The Transactional Outbox Pattern

The Guard : Implementing Idempotency

Three States of a Request

The Background Worker

Scaling the Worker

Why this is robust:

Conclusion

Rust Iterators : DeSugarified

The Iterator Trait

IntoIterator trait

Iterator Ergonomics

.iter() and .iter_mut()

The Fault-Tolerant Job Queue You’ll Actually Understand – in Go

Implementing a Concurrent Priority Queue in Rust

Lifetimes in Rust aren't that hard

Dangling References

TL;DR

Generical Lifetime Annotations

Lifetimes Annotations in Struct Definitions

Lifetime Elision

1. Each parameter that is a reference gets its own lifetime parameter

2. If there is exactly one input lifetime, that lifetime is assigned to all output lifetime parameters

Example:

3. If there are multiple input lifetimes, but one of them is &self or &mut self, the lifetime of self is assigned to all output lifetime parameters

Example:

‘static

Common Sources of 'static Data

'static in Practice

When Not to Use 'static

When It’s Justified

Learn auth in the most comprehensive manner.

The Definitive Auth Guide for Backend Developers

The Definitive Auth Guide for Backend Developers

Stateful vs Stateless Authentication

Stateful Auth

Stateless Auth

The Difference

When to Use What?

Authorization and Bearer Tokens

So, what’s a Bearer Token now?

There are other types of tokens too.

Crypto 101

Why Do We Even Sign Tokens?

Symmetric vs. Asymmetric Algorithms

The Role of Hashing and HMAC

3. If there are multiple input lifetimes, but one of them is `&self` or `&mut self`, the lifetime of `self` is assigned to all output lifetime parameters

Common Sources of `'static` Data

`'static` in Practice

When Not to Use `'static`

Big Brain Situations where `select` is 🔥: