New WAL implementation with configurable variables (#1356)

config/config.go

@@ -82,7 +82,7 @@ memory.keys_limit = 200000000
 memory.lfu_log_factor = 10
 
 # Persistence Configuration
-persistence.enabled = true
+persistence.enabled = false
 persistence.aof_file = "./dice-master.aof"
 persistence.persistence_enabled = true
 persistence.write_aof_on_cleanup = false
@@ -100,7 +100,22 @@ auth.password = ""
 
 # Network Configuration
 network.io_buffer_length = 512
-network.io_buffer_length_max = 51200`
+network.io_buffer_length_max = 51200
+
+# WAL Configuration
+LogDir = "tmp/dicedb-wal"
+Enabled = "true"
+WalMode = "buffered"
+WriteMode = "default"
+BufferSizeMB = 1
+RotationMode = "segemnt-size"
+MaxSegmentSizeMB = 16
+MaxSegmentRotationTime = 60s
+BufferSyncInterval = 200ms
+RetentionMode = "num-segments" 
+MaxSegmentCount = 10
+MaxSegmentRetentionDuration = 600s
+RecoveryMode = "strict"`
 )
 
 var (
@@ -120,6 +135,7 @@ type Config struct {
 	Persistence persistence `config:"persistence"`
 	Logging     logging     `config:"logging"`
 	Network     network     `config:"network"`
+	WAL         WALConfig   `config:"WAL"`
 }
 
 type auth struct {
@@ -163,19 +179,47 @@ type memory struct {
 	MaxMemory      int64   `config:"max_memory" default:"0" validate:"min=0"`
 	EvictionPolicy string  `config:"eviction_policy" default:"allkeys-lfu" validate:"oneof=simple-first allkeys-random allkeys-lru allkeys-lfu"`
 	EvictionRatio  float64 `config:"eviction_ratio" default:"0.9" validate:"min=0,lte=1"`
-	KeysLimit      int     `config:"keys_limit" default:"200000000" validate:"min=0"`
+	KeysLimit      int     `config:"keys_limit" default:"200000000" validate:"min=10"`
 	LFULogFactor   int     `config:"lfu_log_factor" default:"10" validate:"min=0"`
 }
 
 type persistence struct {
 	Enabled           bool   `config:"enabled" default:"false"`
 	AOFFile           string `config:"aof_file" default:"./dice-master.aof" validate:"filepath"`
 	WriteAOFOnCleanup bool   `config:"write_aof_on_cleanup" default:"false"`
-	WALDir            string `config:"wal-dir" default:"./" validate:"dirpath"`
 	RestoreFromWAL    bool   `config:"restore-wal" default:"false"`
 	WALEngine         string `config:"wal-engine" default:"aof" validate:"oneof=sqlite aof"`
 }
 
+type WALConfig struct {
+	// Directory where WAL log files will be stored
+	LogDir string `config:"log_dir" default:"tmp/dicedb-wal"`
+	// Whether WAL is enabled
+	Enabled bool `config:"enabled" default:"true"`
+	// WAL buffering mode: 'buffered' (writes buffered in memory) or 'unbuffered' (immediate disk writes)
+	WalMode string `config:"wal_mode" default:"buffered" validate:"oneof=buffered unbuffered"`
+	// Write mode: 'default' (OS handles syncing) or 'fsync' (explicit fsync after writes)
+	WriteMode string `config:"write_mode" default:"default" validate:"oneof=default fsync"`
+	// Size of the write buffer in megabytes
+	BufferSizeMB int `config:"buffer_size_mb" default:"1" validate:"min=1"`
+	// How WAL rotation is triggered: 'segment-size' (based on file size) or 'time' (based on duration)
+	RotationMode string `config:"rotation_mode" default:"segemnt-size" validate:"oneof=segment-size time"`
+	// Maximum size of a WAL segment file in megabytes before rotation
+	MaxSegmentSizeMB int `config:"max_segment_size_mb" default:"16" validate:"min=1"`
+	// Time interval in seconds after which WAL segment is rotated when using time-based rotation
+	MaxSegmentRotationTime time.Duration `config:"max_segment_rotation_time" default:"60s" validate:"min=1s"`
+	// Time interval in Milliseconds after which buffered WAL data is synced to disk
+	BufferSyncInterval time.Duration `config:"buffer_sync_interval" default:"200ms" validate:"min=1ms"`
+	// How old segments are removed: 'num-segments' (keep N latest), 'time' (by age), or 'checkpoint' (after checkpoint)
+	RetentionMode string `config:"retention_mode" default:"num-segments" validate:"oneof=num-segments time checkpoint"`
+	// Maximum number of WAL segment files to retain when using num-segments retention
+	MaxSegmentCount int `config:"max_segment_count" default:"10" validate:"min=1"`
+	// Time interval in Seconds till which WAL segments are retained when using time-based retention
+	MaxSegmentRetentionDuration time.Duration `config:"max_segment_retention_duration" default:"600s" validate:"min=1s"`
+	// How to handle WAL corruption on recovery: 'strict' (fail), 'truncate' (truncate at corruption), 'ignore' (skip corrupted)
+	RecoveryMode string `config:"recovery_mode" default:"strict" validate:"oneof=strict truncate ignore"`
+}
+
 type logging struct {
 	LogLevel string `config:"log_level" default:"info" validate:"oneof=debug info warn error"`
 	LogDir   string `config:"log_dir" default:"/tmp/dicedb" validate:"dirpath"`

config/validator.go

@@ -28,6 +28,7 @@ import (
 func validateConfig(config *Config) error {
 	validate := validator.New()
 	validate.RegisterStructValidation(validateShardCount, Config{})
+	validate.RegisterStructValidation(validateWALConfig, Config{})
 
 	if err := validate.Struct(config); err != nil {
 		validationErrors, ok := err.(validator.ValidationErrors)
@@ -111,3 +112,36 @@ func applyDefaultValuesFromTags(config *Config, fieldName string) error {
 	log.Printf("Setting default value for %s to: %s", fieldName, defaultValue)
 	return nil
 }
+
+func validateWALConfig(sl validator.StructLevel) {
+	config := sl.Current().Interface().(Config)
+
+	// LogDir validation
+	if config.WAL.LogDir == "" {
+		sl.ReportError(config.WAL.LogDir, "LogDir", "LogDir", "required", "cannot be empty")
+	}
+
+	// MaxSegmentSize validation
+	if config.WAL.MaxSegmentSizeMB <= 0 {
+		sl.ReportError(config.WAL.MaxSegmentSizeMB, "MaxSegmentSize", "MaxSegmentSize", "gt", "must be greater than 0")
+	}
+
+	// MaxSegmentCount validation
+	if config.WAL.MaxSegmentCount <= 0 {
+		sl.ReportError(config.WAL.MaxSegmentCount, "MaxSegmentCount", "MaxSegmentCount", "gt", "must be greater than 0")
+	}
+
+	// BufferSize validation
+	if config.WAL.BufferSizeMB <= 0 {
+		sl.ReportError(config.WAL.BufferSizeMB, "BufferSize", "BufferSize", "gt", "must be greater than 0")
+	}
+
+	// WALMode and WriteMode compatibility checks
+	if config.WAL.WalMode == "buffered" && config.WAL.WriteMode == "fsync" {
+		sl.ReportError(config.WAL.WalMode, "WALMode", "WALMode", "incompatible", "walMode 'buffered' cannot be used with writeMode 'fsync'")
+	}
+
+	if config.WAL.WalMode == "unbuffered" && config.WAL.WriteMode == "default" {
+		sl.ReportError(config.WAL.WalMode, "WALMode", "WALMode", "incompatible", "walMode 'unbuffered' cannot have writeMode as 'default'")
+	}
+}

dicedb.conf

@@ -39,7 +39,7 @@ memory.keys_limit = 200000000
 memory.lfu_log_factor = 10
 
 # Persistence Configuration
-persistence.enabled = true
+persistence.enabled = false
 persistence.aof_file = "./dice-master.aof"
 persistence.persistence_enabled = true
 persistence.write_aof_on_cleanup = false
@@ -56,4 +56,19 @@ auth.password = ""
 
 # Network Configuration
 network.io_buffer_length = 512
-network.io_buffer_length_max = 51200
+network.io_buffer_length_max = 51200
+
+# WAL Configuration
+LogDir = "tmp/dicedb-wal"
+Enabled = "true"
+WalMode = "buffered"
+WriteMode = "default"
+BufferSizeMB = 1
+RotationMode = "segemnt-size"
+MaxSegmentSizeMB = 16
+MaxSegmentRotationTime = 60s
+BufferSyncInterval = 200ms
+RetentionMode = "num-segments" 
+MaxSegmentCount = 10
+MaxSegmentRetentionDuration = 600s
+RecoveryMode = "strict"

internal/iothread/iothread.go

@@ -23,6 +23,7 @@ import (
 	"log/slog"
 	"net"
 	"strconv"
+	"strings"
 	"sync/atomic"
 	"syscall"
 	"time"
@@ -579,13 +580,17 @@ func (t *BaseIOThread) handleCommand(ctx context.Context, cmdMeta CmdMeta, diceD
 		}
 
 		if err == nil && t.wl != nil {
-			t.wl.LogCommand(diceDBCmd)
+			if err := t.wl.LogCommand([]byte(fmt.Sprintf("%s %s", diceDBCmd.Cmd, strings.Join(diceDBCmd.Args, " ")))); err != nil {
+				return err
+			}
 		}
 	case MultiShard, AllShard:
 		err = t.writeResponse(ctx, cmdMeta.composeResponse(storeOp...))
 
 		if err == nil && t.wl != nil {
-			t.wl.LogCommand(diceDBCmd)
+			if err := t.wl.LogCommand([]byte(fmt.Sprintf("%s %s", diceDBCmd.Cmd, strings.Join(diceDBCmd.Args, " ")))); err != nil {
+				return err
+			}
 		}
 	default:
 		slog.Error("Unknown command type",

internal/wal/gen_wal_proto.sh

@@ -0,0 +1,50 @@
+#!/bin/bash
+
+# Exit immediately if a command exits with a non-zero status
+set -e
+
+# Define the directory where the proto file resides
+PROTO_DIR=$(dirname "$0")
+
+# Define the proto file and the output file
+PROTO_FILE="$PROTO_DIR/wal.proto"
+OUTPUT_DIR="."
+
+# Function to install protoc if not present
+install_protoc() {
+    echo "protoc not found. Installing Protocol Buffers compiler..."
+    if [[ "$(uname)" == "Darwin" ]]; then
+        # MacOS installation
+        brew install protobuf
+    elif [[ "$(uname)" == "Linux" ]]; then
+        # Linux installation
+        sudo apt update && sudo apt install -y protobuf-compiler
+    else
+        echo "Unsupported OS. Please install 'protoc' manually."
+        exit 1
+    fi
+}
+
+# Function to install the Go plugin for protoc if not present
+install_protoc_gen_go() {
+    echo "protoc-gen-go not found. Installing Go plugin for protoc..."
+    go install google.golang.org/protobuf/cmd/protoc-gen-go@latest
+    export PATH="$PATH:$(go env GOPATH)/bin"
+    echo "Make sure $(go env GOPATH)/bin is in your PATH."
+}
+
+# Check if protoc is installed, install if necessary
+if ! command -v protoc &>/dev/null; then
+    install_protoc
+fi
+
+# Check if the Go plugin for protoc is installed, install if necessary
+if ! command -v protoc-gen-go &>/dev/null; then
+    install_protoc_gen_go
+fi
+
+# Generate the wal.pb.go file
+echo "Generating wal.pb.go from wal.proto..."
+protoc --go_out="$OUTPUT_DIR" --go_opt=paths=source_relative "$PROTO_FILE"
+
+echo "Generation complete. File created at $OUTPUT_DIR/wal.pb.go"

internal/wal/wal.go

@@ -26,7 +26,7 @@ import (
 )
 
 type AbstractWAL interface {
-	LogCommand(c *cmd.DiceDBCmd)
+	LogCommand([]byte) error
 	Close() error
 	Init(t time.Time) error
 	ForEachCommand(f func(c cmd.DiceDBCmd) error) error