m2.1 : Fix wal entry format, use checksums, fix wal file on startup if corrupted

cmd/server/main.go

@@ -14,6 +14,7 @@ import (
 const port = ":9000" // TODO configs
 const walFilePath = "bin/wal.log" // TODO configs
 const fsyncStrategy = wal.EVERY_SEC // TODO configs
+const maxWalEntryLength = 1 << 30 // TODO configs
 
 func main() {
 	ctx, cancel := context.WithCancel(context.Background())
@@ -24,7 +25,7 @@ func main() {
 	go handleFatalErrors(fatalErrChan, cancel)
 
 	store := store.NewStore()
-	wal, err := wal.NewWal(walFilePath, fsyncStrategy, ctx, fatalErrChan)
+	wal, err := wal.NewWal(walFilePath, fsyncStrategy, maxWalEntryLength, ctx, fatalErrChan)
 	if err != nil {
 		log.Fatal("unable to open wal file: ", err)
 	}

internal/server/tcp_server.go

@@ -26,7 +26,9 @@ func NewTCPServer(addr string, store *store.Store, wal *wal.Wal) *TCPServer {
 }
 
 func (s *TCPServer) Start(ctx context.Context) error {
-	err := replayWal(s.store, s.wal, 0)
+	if err := replayWal(s.store, s.wal, 0); err != nil {
+		return err
+	}
 	ln, err := net.Listen("tcp", s.addr)
 	if err != nil {
 		return err

internal/wal/wal.go

@@ -3,15 +3,17 @@ package wal
 import (
 	"bufio"
 	"context"
+	"encoding/binary"
 	"fmt"
+	"hash/crc32"
+	"io"
 	"iter"
 	"log"
 	"os"
 	"strings"
 	"bytes"
 	"sync"
 	"time"
-	"io"
 )
 
 type FsyncStrategy int
@@ -28,10 +30,14 @@ type Wal struct {
 	closed bool
 	fsyncStrategy FsyncStrategy
 	asyncBuffer *bytes.Buffer
+	entryBuf []byte
 	fatalErrChan chan error
+	maxEntryLen uint32
 }
 
-func NewWal(filepath string, fsyncStrategy FsyncStrategy, ctx context.Context, fatalErrChan chan error) (*Wal, error) {
+var byteOrder binary.ByteOrder = binary.LittleEndian
+
+func NewWal(filepath string, fsyncStrategy FsyncStrategy, maxWalEntryLength uint32, ctx context.Context, fatalErrChan chan error) (*Wal, error) {
 	file, err := os.OpenFile(filepath, os.O_WRONLY|os.O_CREATE|os.O_APPEND, 0644)
 	if err != nil {
 		return nil, err
@@ -42,6 +48,8 @@ func NewWal(filepath string, fsyncStrategy FsyncStrategy, ctx context.Context, f
 		closed: false,
 		fsyncStrategy: fsyncStrategy,
 		fatalErrChan: fatalErrChan,
+		maxEntryLen: maxWalEntryLength,
+		entryBuf: make([]byte, maxWalEntryLength),
 	}
 	if fsyncStrategy != ALWAYS {
 		w.asyncBuffer = new(bytes.Buffer)
@@ -51,36 +59,6 @@ func NewWal(filepath string, fsyncStrategy FsyncStrategy, ctx context.Context, f
 	return w, nil
 }
 
-func (w *Wal) Append(cmd ... string) (int, error) {
-	w.mu.Lock()
-	defer w.mu.Unlock()
-	if !w.closed {
-		if w.fsyncStrategy != ALWAYS {
-			bytes, err := fmt.Fprintf(w.asyncBuffer, "%s\n", strings.Join(cmd, " "))
-			return bytes, err
-		}
-
-		bytes, err := fmt.Fprintf(w.file, "%s\n", strings.Join(cmd, " "))
-		if err == nil {
-			err = w.file.Sync()
-		}
-		if err != nil {
-			log.Printf("wal append failure: %s, stopping the server", err)
-			w.fatalErrChan <- err
-		}
-		return bytes, err
-	}
-	return 0, fmt.Errorf("unable to append, wal already closed")
-}
-
-func (w *Wal) Close() error {
-	w.mu.Lock()
-	defer w.mu.Unlock()
-
-	w.closed = true
-	return w.file.Close()
-}
-
 func scheduleFsyncEverySec(w *Wal, ctx context.Context) {
 	ticker := time.NewTicker(1 * time.Second)
 	defer ticker.Stop()
@@ -128,19 +106,135 @@ func scheduleFsyncEverySec(w *Wal, ctx context.Context) {
 	}
 }
 
+func (w *Wal) Append(cmd ... string) (int, error) {
+	w.mu.Lock()
+	defer w.mu.Unlock()
+	if !w.closed {
+		line := strings.Join(cmd, " ")
+		entry, err := w.getWalEntry(line)
+		if err != nil {
+			return 0, err
+		}
+
+		if w.fsyncStrategy != ALWAYS {
+			bytes, err := w.asyncBuffer.Write(entry)
+			return bytes, err
+		}
+
+		bytes, err := w.file.Write(entry)
+		if err == nil {
+			err = w.file.Sync()
+		}
+		if err != nil {
+			log.Printf("wal append failure: %s, stopping the server", err)
+			w.fatalErrChan <- err
+		}
+		return bytes, err
+	}
+	return 0, fmt.Errorf("unable to append, wal already closed")
+}
+
+func (w *Wal) getWalEntry(line string) ([]byte, error) {
+	// wal entry format - [length: 4 bytes][line: <length> bytes][checksum: 4 bytes]
+	lineBytes := []byte(line)
+	length := len(lineBytes)
+	if length > int(w.maxEntryLen) {
+		return nil, fmt.Errorf("line length > maxLength, can't append")
+	}
+
+	byteOrder.PutUint32(w.entryBuf[:4], uint32(length))
+	copy(w.entryBuf[4:(length + 4)], lineBytes)
+
+	checksum := calcChecksum(w.entryBuf[:(length + 4)])
+	byteOrder.PutUint32(w.entryBuf[(length + 4):(length + 8)], checksum)
+
+	return w.entryBuf[:(length + 8)], nil
+}
+
+func calcChecksum(bytes []byte) uint32 {
+	return crc32.ChecksumIEEE(bytes)
+}
+
 func (w *Wal) WALIterator(fromOffset int64) iter.Seq2[string, error] {
+	// the iterator also validates the wal file and truncates till last read safe offset
     return func(yield func(string, error) bool) {
-        f, err := os.Open(w.file.Name()) // create a separate fd
-        if err != nil { yield("", err); return }
+		w.mu.Lock()
+		defer w.mu.Unlock()
+        f, err := os.OpenFile(w.file.Name(), os.O_RDWR, 0644) // create a separate fd
+        if err != nil {
+            yield("", err)
+            return
+        }
         defer f.Close()
 
-		_, err = f.Seek(fromOffset, io.SeekStart)
-		if err != nil { yield("", err); return }
+        if _, err := f.Seek(fromOffset, io.SeekStart); err != nil {
+            yield("", err)
+            return
+        }
+
+		var safeOffset int64 = 0
+		var currOffset int64 = 0
+
+		buf := make([]byte, w.maxEntryLen)
+        reader := bufio.NewReader(f)
+
+        for {
+			// read length
+            _, err := io.ReadFull(reader, buf[:4])
+            if err == io.EOF {
+                return
+            }
+            if err != nil {
+				log.Printf("wal corrupted, err %s, truncating till last safe read offset %d\n", err, safeOffset)
+				f.Truncate(safeOffset)
+				f.Sync()
+				yield("", err)
+                return
+            }
+            length := byteOrder.Uint32(buf[:4])
+			currOffset += 4
+			if length > w.maxEntryLen {
+				log.Printf("wal corrupted, length found too big, truncating till last safe read offset %d\n", safeOffset)
+				f.Truncate(safeOffset)
+				f.Sync()
+                return
+			}
+
+            // read full wal entry ([length | data | checksum])
+            _, err = io.ReadFull(reader, buf[4:(8 + length)])
+            if err != nil {
+				log.Printf("wal corrupted, err %s, truncating till last safe read offset %d\n", err, safeOffset)
+				f.Truncate(safeOffset)
+				f.Sync()
+				yield("", err)
+                return
+            }
+            data := buf[4:(length + 4)]
+			currOffset += int64(length) + 4
+
+			// validate checksum
+			checksum := buf[(4 + length):(8 + length)]
+			checksumFound := byteOrder.Uint32(checksum)
+			checksumCalc := calcChecksum(buf[:(4 + length)])
+			if checksumFound != checksumCalc {
+				log.Printf("wal corrupted, checksum mismatch, truncating till last safe read offset %d\n", safeOffset)
+				f.Truncate(safeOffset)
+				f.Sync()
+				return
+			}
+			safeOffset = currOffset
 
-        scanner := bufio.NewScanner(f)
-        for scanner.Scan() {
-            if !yield(scanner.Text(), nil) { return }
+			if !yield(string(data), nil) {
+                return
+            }
         }
-        if err := scanner.Err(); err != nil { yield("", err) }
     }
 }
+
+func (w *Wal) Close() error {
+	w.mu.Lock()
+	defer w.mu.Unlock()
+
+	w.closed = true
+	return w.file.Close()
+}