第 11 章

[Day 11] AOF Fsync and Config

Is there possible losing data when shutdown but AOF haven’t been written?

Sure.

As I called os.file.Write(), the OS (POSIX-Compliant) could write data into page buffer, but not put into disk immediately. If there’s any unexpected termination (e.g. power-off) before flushing, data loss forever.

In most of database service (MySQL, Redis, …) uses fsync() syscall to force flushing page buffer into disk.

AOF Sync

Redis provides 3 policies for AOF syncing:

no: do not sync manually, wait for OS
always: sync every write of AOF
everysec: dispatch a thread, sync AOF every second.

It is configurable by redis.conf

Redis configuration

Redis can be configured by redis.conf, it is a text based file. More info

We can write a parser easily:

package data

import (
	"bufio"
	"errors"
	"fmt"
	"io"
	"os"
	"strings"
)

var (
	ErrCommand = errors.New("invalid command")
)

const (
	AOFFsyncNo = iota + 1
	AOFFsyncAlways
	AOFFsyncEverySec
)

type Config struct {
	AOFEnabled bool
	AOFFsync   int
}

func NewConfig(path string) (*Config, error) {
	f, err := os.Open(path)
	if err != nil {
		return nil, err
	}
	cfg := Config{}

	rd := bufio.NewReader(f)
	ln := 0
	for {
		ln++
		line, err := rd.ReadString('\n')
		if err != nil {
			if errors.Is(err, io.EOF) {
				return &cfg, nil
			}

			return nil, err
		}

		line = strings.Trim(line, " \n")
		if len(line) == 0 || line[0] == '#' {
			continue
		}

		parsed := strings.SplitN(line, " ", 2)
		if len(parsed) != 2 {
			return nil, fmt.Errorf("%w: (line: %d): %+v", ErrCommand, ln, parsed)
		}
		cmd, arg := parsed[0], parsed[1]

		switch cmd {
		case "appendonly":
			if arg == "yes" {
				cfg.AOFEnabled = true
			}
		case "appendfsync":
			switch arg {
			case "no":
				cfg.AOFFsync = AOFFsyncNo
			case "always":
				cfg.AOFFsync = AOFFsyncAlways
			case "everysec":
				cfg.AOFFsync = AOFFsyncEverySec
			}
		}
	}
}

Note: This isn’t a full Redis configration parser.

Apply configuration

Initialize config in main.go:

func main() {
	cfg, err := data.NewConfig("redis.conf")
	if err != nil {
		slog.Error("failed to load config", slog.Any("error", err))
		return
	}

Use config in app.go:

func NewApp(cfg *data.Config) (*App, error) {
	var handler server.Handler
	var restorer server.Restorer
	if cfg.AOFEnabled {
		aof, err := service.NewAOF(AOFPath)
		if err != nil {
			return nil, err
		}

		handler = server.NewHandler(cmd.NewCommands(repo.NewStorage()), server.NewAOFMiddleware(aof))
		restorer = server.NewRestorer(aof, handler)
	} else {
		handler = server.NewHandler(cmd.NewCommands(repo.NewStorage()))
	}

	return &App{
		cfg: cfg,
		srv: server.NewServer(handler, restorer),
	}, nil
}

When restorer is nil, ignore it:

func (s *simpleSrv) RestoreFromDisk() error {
	if s.restorer == nil {
		return nil
	}

	return s.restorer.LoadFromDisk()
}

Sync Always

If appendfsync always is set, call aof.Sync() after writing.

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

	if aof.cfg.AOFFsync == data.AOFFsyncAlways {
		return aof.f.Sync()
	}

	return nil
}

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

+	if aof.cfg.AOFFsync == data.AOFFsyncAlways {
+		return aof.f.Sync()
+	}

	return nil

Sync Every Second

If appendfsync everysec is set, add a goroutine in cmd/olivine/app.go to call sync every second.

	if app.cfg.AOFEnabled && app.cfg.AOFFsync == data.AOFFsyncEverySec {
		go func() {
			for {
				time.Sleep(time.Second)
				if err := app.aof.Sync(); err != nil {
					errch <- err
				}
			}
		}()
	}

Don’t forget add Sync() in service.AOF{}

// file: internal/service/aof.go
type AOF interface {
	Read() (*resp.Command, error)
	Write(*resp.Command) error
+	Sync() error
	Close() error
}

+func (aof *file) Sync() error {
+	aof.mu.Lock()
+	defer aof.mu.Unlock()
+
+	return aof.f.Sync()
+}

Redis Config Parser

I’ve implemented a incomplete configuration parser, but it includes some problems:

Because of rd.ReadString('\n'), it returns EOF if this file is not includes tailing \n
Because of strings.Trim(line, " \n") and strings.SplitN(line, " ", 2), it’s hard to understand.

I told Codex (GPT-5.5 medium) to fix these problems, and it gave me:

func NewConfig(path string) (*Config, error) {
	f, err := os.Open(path)
	if err != nil {
		return nil, err
	}
	defer f.Close()

	cfg := Config{}

	rd := bufio.NewReader(f)
	ln := 0
	for {
		ln++
		line, err := rd.ReadString('\n')
		if err != nil {
			if errors.Is(err, io.EOF) && len(line) == 0 {
				return &cfg, nil
			}
			if !errors.Is(err, io.EOF) {
				return nil, err
			}
		}

		if err := parse(&cfg, ln, line); err != nil {
			return nil, err
		}

		if errors.Is(err, io.EOF) {
			return &cfg, nil
		}
	}
}

func parse(cfg *Config, ln int, line string) error {
	parsed := strings.Fields(line)
	if len(parsed) == 0 || strings.HasPrefix(parsed[0], "#") {
		return nil
	}

	if len(parsed) != 2 {
		return fmt.Errorf("%w: (line: %d): %+v", ErrCommand, ln, parsed)
	}
	cmd, arg := parsed[0], parsed[1]

	switch cmd {
	case "appendonly":
		if arg == "yes" {
			cfg.AOFEnabled = true
		}
	case "appendfsync":
		switch arg {
		case "no":
			cfg.AOFFsync = AOFFsyncNo
		case "always":
			cfg.AOFFsync = AOFFsyncAlways
		case "everysec":
			cfg.AOFFsync = AOFFsyncEverySec
		}
	}

	return nil
}

Ohh…, what the fxxk…, it made everything hard to read: io.EOF checking everywhere.

There’s an important context: Redis configuration is provided by server owner, and it may not a very large file.

Therefore, we completely load whole file into memory and parse it directly instead of read line by line. The tailing \n can be added after read in memory, we don’t need to hanle it in every loop.

There’s my answer:

func NewConfigFromBytes(content []byte) (*Config, error) {
	cfg := Config{}
	rd := bytes.NewBuffer(append(content, '\n'))
	ln := 0
	for {
		ln++
		line, err := rd.ReadString('\n')
		if err != nil {
			if errors.Is(err, io.EOF) {
				return &cfg, nil
			}
			return &cfg, fmt.Errorf("%w: (line: %d): %w", ErrCommand, ln, err)
		}

		if err := parse(&cfg, line); err != nil {
			return &cfg, fmt.Errorf("%w: (line: %d): %w", ErrCommand, ln, err)
		}
	}
}

Constant Enum

There’s an important design is enum of AOFFsync:

const (
	AOFFsyncNo = iota + 1
	AOFFsyncAlways
	AOFFsyncEverySec
)

Use iota+1 instead of iota, because the variable will be initialize as zero value in Go. We don’t know if AOFFsync be set as AOFFsyncNo or default zero value if using iota.

Ticker

I used time.Sleep(time.Second) for AOF syncing in cmd/olivine/app.go:

		go func() {
			for {
				time.Sleep(time.Second)
				if err := app.aof.Sync(); err != nil {
					errch <- err
				}
			}
		}()

It ignores shutdown signals in this goroutine, it is safe because program exit recently, but it’s not elegant.

Use time.Ticker instead:

		go func() {
			ticker := time.NewTicker(time.Second)
			defer ticker.Stop()
			for {
				select {
				case <-ctx.Done():
					return
				case <-ticker.C:
					if err := app.aof.Sync(); err != nil {
						errch <- err
						return
					}
				}
			}
		}()