diff --git a/cmd/server/main.go b/cmd/server/main.go index 960aa5d..8273cb4 100644 --- a/cmd/server/main.go +++ b/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) } diff --git a/internal/server/tcp_server.go b/internal/server/tcp_server.go index 29c512d..f4e8a3b 100644 --- a/internal/server/tcp_server.go +++ b/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 diff --git a/internal/wal/wal.go b/internal/wal/wal.go index bffd3c8..56f1750 100644 --- a/internal/wal/wal.go +++ b/internal/wal/wal.go @@ -3,7 +3,10 @@ package wal import ( "bufio" "context" + "encoding/binary" "fmt" + "hash/crc32" + "io" "iter" "log" "os" @@ -11,7 +14,6 @@ import ( "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: 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() +}