Rebuilding Redis in Go, Part 3: Streams and the Art of the Naive Implementation

Section 3 is done, and it was all about streams. XADD, XRANGE, XREAD, and blocking reads, plus a TYPE command. This was the first section where I really felt the pull between "make it work" and "make it fast," and I want to talk about which one won.

Also new this section: I started letting AI write tests for me. Not the implementation, that's still mine and hand-written, but tests I can run locally to verify things are working. That felt like a fair trade.

TYPE:

The section opened with the TYPE command, which returns the data type of a key. Easy enough, and a good excuse to write my first Go enum:

type Types string

func (t Types) String() string {
	return string(t)
}

const (
	String    Types = "string"
	List      Types = "list"
	Set       Types = "set"
	ZSet      Types = "zset"
	Hash      Types = "hash"
	Stream    Types = "stream"
	VectorSet Types = "vectorset"
	None      Types = "none"
)

For now I only send back string or none. Key exists, it's a string. Doesn't exist, it's none. ezpz.

XADD:

Then came XADD, which introduced me to streams. You send an ID and some key-value pairs, and they get stored in memory. I also noticed CodeCrafters a comment pointing out that Redis implements streams as radix tries.

I have not given much thought to data structures or optimizations so far. I've mostly been doing whatever comes to me first, and I know that sounds bad, like I should be thinking about optimization immediately. But I want to get it working first, then optimize it. That's the deal I've made with myself.

So here's my naive implementation:

type entry struct {
	id     string
	values []kvp
}

type kvp struct {
	key   string
	value string
}

var streamMap = make(map[string][]entry)

func XAdd(c helpers.Connection, args []string) error {
	key := args[0]
	id := args[1]
	values := []kvp{}

	for i := 2; i < len(args); i += 2 {
		values = append(values, kvp{
			key:   args[i],
			value: args[i+1],
		})
	}

	streamMap[key] = append(streamMap[key], entry{
		id:     id,
		values: values,
	})

	return c.SendBulk(id)
}

A map of keys to slices of entries. I can already see where this hurts: range queries like "give me 10 entries after this one" mean walking the slice one by one, which is O(N). The radix trie is definitely the way to go, but it works, and I'll optimize it later.

One thing I'm already smelling, though: my TYPE command now chains map checks to figure out where a key lives:

if _, ok := strings.GetMap()[key]; ok {
	valueType = String
} else if _, ok := streams.GetMap()[key]; ok {
	valueType = Stream
} else {
	valueType = None
}

Every new data type adds another link to this chain. That's getting fixed sooner rather than later.

XRANGE:

Querying ranges from a stream took longer than any lesson so far, and it had me seriously considering a rewrite of my stream implementation into something built for range queries on millisecond timestamps. I know I could keep going with the naive version, but it wouldn't be optimal. O(N) lookups on sequentially stored data, and the validation rules mean older data can't come before newer data, so at least it never degrades into O(N^2).

My actual blocker ended up being dumber than performance: figuring out how to print the data. The range query response is a nested structure, and getting the RESP encoding right took some fiddling. I did also change the ms field from a string to time.Time, which made the comparisons clearer to read.

The - and + range bounds were a nice win, though. I had a helper to parse the IDs, so - just becomes 0-0 (everything is above it) and + becomes maxInt64-maxInt (everything is before it). No special-casing in the range logic itself.

That printing blocker also pushed me into a refactor I took from an AI suggestion: flipping my message helpers so the message is the receiver instead of the connection. Now there are message types behind an interface, and each one implements its own SendTo(connection). Sending an array just calls SendTo on each element, which is much cleaner. I kept the old connection.SendBulk(message) style methods around for now, but I'll probably remove them, and settle on this new approach as my new standard.

XREAD:

I went into XREAD thinking I could reuse most of XRANGE. That didn't work out as planned. What I did reuse were the parseRangeID and rangeIDBefore helpers, which was most of the XRANGE code anyway. The key difference: XREAD excludes the given ID, while XRANGE includes it. So I add data to the result array only when the start is before the incoming ID.

Multiple streams threw another wrench in. Redis sends all the keys first, then all the IDs, which looked strange until I realized I could just cut the args in half to get the stream count, pair each key with its ID, and run the same single-stream logic on each pair.

Blocking Reads:

Blocking XREAD took extra time too. Concurrency in Go is proving to be a bit of a challenge for me. I'm understanding it slowly, and I'll keep learning more about it.

The last lesson of the section was the $ ID, which means "only entries added after this point." The trick was resolving it up front:

if ids[0] == "$" {
	lastID, ok := GetLastID(keys[0])
	if ok {
		ids[0] = lastID.String()
	} else {
		ids[0] = "0-0"
	}
}

Once $ becomes a concrete ID, the rest of the blocking logic doesn't even know it was special. That felt good.

What's Next

Section 3 was the biggest one yet: streams from scratch, my first real brush with "this needs a better data structure," a message helper refactor, and more concurrency homework. The naive implementation survives for now, but the radix trie comment is keeping me up at night and is probably going to hurt me the longer I keep it naive.

Onward to section 4!!