第 7 章

Day 7：資料持久化

目前的 TCP server 已支援 RESP2，以及 PING、GET、SET 指令。

但資料存在 repo.mapStorage 的記憶體中，服務重新啟動後就會全部消失。

Redis 的資料持久化

Redis 有兩種持久化方式：

Append Only File（AOF）：以 RESP 指令記錄所有會改變儲存內容的操作。
RDB：記錄某個時間點的儲存快照。

我選擇先實作較簡單的 AOF，未來也許會以附錄補上 RDB。

AOF 格式

AOF 的內容是經過 RESP marshal 的原始 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..

當改變資料的 Redis Command 成功執行後，就將它的 RESP bytes 附加到 AOF。

AOF Reader 與 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()
}

透過 Middleware 整合 AOF

Middleware 適合 HTTP server 這類 request-response 模型。

Go 的 middleware 通常以接收 handler 的函式表示。在 internal/server/handler.go 加入 HandlerFunc 與 Middleware：

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
}

加入 AOFMiddleware constructor

在 internal/server/middleware.go 加入 NewAOFMiddleware()：

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
		}
	}
}

在 `App` 註冊 Middleware

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
}

啟動時從 AOF 還原

`Restorer` interface 與 `aofRestorer` 實作

考量未來可能加入 RDB，先定義 Restorer interface，讓 aofRestorer 和未來的 rdbRestorer 共用介面。

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
		}
	}
}

目前 NewRestorer 需要 service.AOF 參數。理想上，它應該根據檔案 header 判斷使用 AOF 或 RDB，但目前先接受這個限制。

使用 `Restorer`

在 Server 套用 Restorer：

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()
}

再於 App 建立 Server 時註冊：

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
}

接下來呢？

恭喜，我們已實作能處理 RESP 與部分 Redis Commands 的 TCP server。

但仍有許多工作：

使用 connection context。
Graceful shutdown。
測試（Gemini 提醒了這件事，我打算下一週補上）。
RDB 與 AOF rewrite。
Sentinel 與 cluster mode。
其他功能。

希望基礎篇對你我都有幫助，進階篇再見！