1
A Redis-compatible server, from scratch, in Go[github]
1
2
A Redis-compatible server built from scratch in Go, speaking the
3
real RESP wire protocol over TCP — you can talk to it with
4
redis-cli. Concurrent clients are handled with a
5
goroutine-per-connection model over a thread-safe keyspace.
6
7
Implements strings with expiry, streams (XADD/XRANGE/XREAD),
8
transactions (MULTI/EXEC/DISCARD), optimistic locking (WATCH),
9
and full leader-follower replication with command propagation and
10
WAIT. Built through the CodeCrafters challenge — base stages plus
11
the Streams, Transactions, Optimistic Locking, and Replication
12
extensions.
13
14
15
17
18
func Parse(r Reader) (RESPValue, error) {
19
b, err := r.ReadByte()
20
if err != nil {
21
return nil, err
22
}
23
switch b {
24
case '*':
25
return ParseArray(r)
26
case '$':
27
return ParseBulkString(r)
28
case ':':
29
return ParseInteger(r)
30
case '+':
31
return ParseSimpleString(r)
32
33
default:
34
return nil, fmt.Errorf("unexpected input %q", b)
35
}
36
}
37
38
39
40
41
42
43
func Parse(r Reader) (RESPValue, error) {
b, err := r.ReadByte()
if err != nil {
return nil, err
}
switch b {
case '*':
return ParseArray(r)
case '$':
return ParseBulkString(r)
case ':':
return ParseInteger(r)
case '+':
return ParseSimpleString(r)
default:
return nil, fmt.Errorf("unexpected input %q", b)
}
}
func ParseBulkString(r Reader) (BulkString, error) {
contentLength, firstNonNumberByte, err := readNumber(r)
if err != nil {
return "", err
}
if firstNonNumberByte != '\r' {
return "", fmt.Errorf("expected '\\r', got %q", firstNonNumberByte)
}
// nums finished, skip next byte which should be '\r'
b, err := r.ReadByte()
if err != nil {
return "", err
}
if b != '\n' {
return "", fmt.Errorf("expected '\\n', got %q", b)
}
// read content
buf := make([]byte, contentLength)
_, err = io.ReadFull(r, buf)
if err != nil {
return "", err
}
func makeNthStreamID(id string, lastEntry Pairs) PairsID {
chunks := strings.Split(id, "-")
var ms int
var seq int
if chunks[0] == "*" {
ms = int(time.Now().UnixMilli())
if ms == lastEntry.ID.Milliseconds {
seq = lastEntry.ID.Sequence + 1
} else {
seq = 0
}
return PairsID{Milliseconds: ms, Sequence: seq}
} else {
ms, _ = strconv.Atoi(chunks[0])
}
if chunks[1] == "*" {
if ms == lastEntry.ID.Milliseconds {
seq = lastEntry.ID.Sequence + 1
} else {
seq = 0
}
} else {
seq, _ = strconv.Atoi(chunks[1])
}
return PairsID{Milliseconds: ms, Sequence: seq}
}
func (r *Registry) CountAcked(target int64) int {
r.mu.Lock()
defer r.mu.Unlock()
res := 0
for _, replica := range r.replicas {
if replica.AckOffset.Load() >= target {
res++
}
}
return res
}
func (r *Registry) RecordAck(conn net.Conn, target int64) {
r.mu.Lock()
defer r.mu.Unlock()
for _, replica := range r.replicas {
if replica.Conn == conn {
replica.AckOffset.Store(target)
select {
case r.ackSignals <- struct{}{}:
default:
}
}
}
}