第 7 章

[Day 7] Data Persistence

We have a TCP server supports RESP2 and PING, GET and SET Redis commands.

However, when the server is restarted, all of data will be reset because data is stored in memory by repo.mapStorage{}.

Data Persistence in Redis

Ther’re 2 methods to keep data persistence in Redis:

Append Only File (AOF): a log file write all commands which changes storage in RESP commands.
Redis Databae (RDB): a snapshot of storage at point-in-time.

I choosed AOF because it is easier to implement. (Maybe I’ll add RDB as appendix in future.)

AOF Format

The content of AOF is raw RESP marshaled bytes:

$ redis-cli -p 16879 SET foo bar
OK
$ cat database.aof
*3
$3
set
$3
foo
$3
bar
$ xxd database.aof
00000000: 2a33 0d0a 2433 0d0a 7365 740d 0a24 330d  *3..$3..set..$3.
00000010: 0a66 6f6f 0d0a 2433 0d0a 6261 720d 0a    .foo..$3..bar..

The logic is clear: when a storage changes Redis command execute successful, append RESP mashaled bytes into AOF.

AOF Reader and Writer

package service

import (
	"fmt"
	"olivine/pkg/resp"
	"os"
	"sync"
)

type AOF interface {
	Read() (*resp.Command, error)
	Write(*resp.Command) error
	Close() error
}

type file struct {
	f  *os.File
	rd *resp.Reader
	mu sync.Mutex
}

func (aof *file) Read() (*resp.Command, error) {
	return resp.ReadCommand(aof.rd)
}

func (aof *file) Write(v *resp.Command) error {
	aof.mu.Lock()
	defer aof.mu.Unlock()

	if !v.Dirty() {
		return nil
	}

	marshaled := v.Marshal()
	n, err := aof.f.Write(marshaled)
	if err != nil {
		return err
	}
	if n != len(marshaled) {
		return fmt.Errorf("wrote length mismatch: got %d want %d", n, len(marshaled))
	}

	return nil
}

func (aof *file) Close() error {
	if aof.f == nil {
		return nil
	}

	aof.mu.Lock()
	defer aof.mu.Unlock()

	return aof.f.Close()
}

Integrate AOF with Middleware

Middleware is an useful design pattern for a request-response model, such as HTTP server.

Middleware patttern in Go usually shows as function with handler, add HandlerFunc and Middleware type in internal/server/handler.go:

type HandlerFunc func(context.Context, *resp.Command) (resp.Value, error)
type Middleware func(HandlerFunc) HandlerFunc

func NewHandler(cmds []cmd.Command, middlewares ...Middleware) Handler {
	h := simpleHandler{}
	h.factory = make(map[string]HandlerFunc)
	for _, cmd := range cmds {
		h.factory[cmd.Command()] = cmd.Exec
	}

	h.serve = h.exec
	for i := len(middlewares) - 1; i >= 0; i-- {
		h.serve = middlewares[i](h.serve)
	}

	return &h
}

Add AOFMiddleware constructor

Add NewAOFMiddleware() constructor in internal/server/middleware.go:

package server

import (
	"context"
	"fmt"
	"olivine/internal/service"
	"olivine/pkg/resp"
)

func NewAOFMiddleware(aof service.AOF) Middleware {
	return func(next HandlerFunc) HandlerFunc {
		return func(ctx context.Context, cmd *resp.Command) (resp.Value, error) {
			ret, err := next(ctx, cmd)
			if err != nil {
				return ret, err
			}
			if err := aof.Write(cmd); err != nil {
				return nil, fmt.Errorf("%w: failed to write AOF: %w", ErrServer, err)
			}

			return ret, nil
		}
	}
}

Register Middleware in `App{}`

func NewApp() (*App, error) {
	aof, err := service.NewAOF("database.aof")
	if err != nil {
		return nil, err
	}

	return &App{
		srv: server.NewServer(server.NewHandler(cmd.NewCommands(repo.NewStorage()), server.NewAOFMiddleware(aof))),
	}, nil
}

Restore from AOF when server startup

`Restorer{}` interface and `aofRestorer{}` implementation

I’m going to add RDB support in future, define a Restorer{} interface for aofRestorer{} and rdbRestorer{} implementations is better.

package server

import (
	"context"
	"errors"
	"io"
	
	"olivine/internal/service"
)

type Restorer interface {
	LoadFromDisk() error
}

func NewRestorer(aof service.AOF, handler Handler) Restorer {
	return &aofRestorer{aof: aof, handler: handler}
}

type aofRestorer struct {
	aof     service.AOF
	handler Handler
}

func (r *aofRestorer) LoadFromDisk() error {
	for {
		cmd, err := r.aof.Read()
		if err != nil {
			if errors.Is(err, io.EOF) {
				return nil
			}
			return err
		}

		if _, err := r.handler.Exec(context.Background(), cmd); err != nil {
			return err
		}
	}
}

The NewRestorer{} requires service.AOF as argument, it is not good because this constructor should decide using aofRestorer{} or rdbRestorer{} by file header, but it’s OK at this moment.

Use `Restorer{}`

Apply Restorer{} in Server{}:

type Server interface {
	ListenAndServe() error
	RestoreFromDisk() error
}

func NewServer(handler Handler, restorer Restorer) Server {
	return &simpleSrv{
		handler:  handler,
		restorer: restorer,
	}
}

type simpleSrv struct {
	handler  Handler
	restorer Restorer

	listener net.Listener
}

func (s *simpleSrv) RestoreFromDisk() error {
	return s.restorer.LoadFromDisk()
}

And register Restorer{} into Server{} in App{}:

func NewApp() (*App, error) {
	aof, err := service.NewAOF(AOFPath)
	if err != nil {
		return nil, err
	}

	handler := server.NewHandler(cmd.NewCommands(repo.NewStorage()), server.NewAOFMiddleware(aof))

	return &App{
		srv: server.NewServer(
			handler,
			server.NewRestorer(aof, handler),
		),
	}, nil
}

What’s Next?

Congratulations! We have imlemented a TCP server supports RESP and some Redis commands.

However, there’s a lot of things todo:

Connection context not in use
Graceful shutdown
Testing (it complained by Gemini, I’ll add it in next week.)
RDB support, AOF rewrite support
Sentiel and cluster mode
…

However, I hope that the basic section of tutorial can help both you and me. See you in advance section!