Merge pull request #15 from cozis/metadata-server-WAL

Implement metadata server WAL

.gitignore

@@ -8,3 +8,5 @@ coverage_report/
 *.gcov
 *.gcda
 *.gcno
+*.wal
+*.tmp

misc/DESIGN.txt

@@ -209,3 +209,12 @@ Chunk Management:
     already cached that chunk list or not. If it didn't, the chunk server sends the entire list
     in chunks during state updates, with an increased
     update frequency.
+
+Metadata Persistence & Crash Recovery:
+    The metadata server uses a write-ahead log (WAL) file to store its state on disk.
+
+    Log files start with a full snapshot of the metadata and continues with operation
+    log entries.
+
+    When a file gets too big, the metadata server creates a new WAL file by writing
+    a snapshot to it and continuing logging there.

misc/TODO.txt

@@ -13,3 +13,8 @@
 [ ] Add failt injections to the simulation
 [ ] Add authentication
 [ ] Add release build and allow switching between debug, release, coverage, sanitizer builds while calling make
+[ ] When an operation is committed to the WAL, it may then fail due to not deterministic reasons (out of memory). When the metadata server restarts and replays the log, the operation may succede and cause the system to diverge from the last instance of the process. Is this not a problem? How do we solve it?
+[ ] Add notes on who inspired the testing approach
+[ ] Add notes on benchmarks and that we don't do batching
+[ ] Fix endianess when writing to network and the WAL
+[ ] Add a change counter to FileTree. The counter is initialized to 0 and updated any time the tree is changed. When a new file is created, the current change counter is assigned to it. This makes it possible to verify that the file didn't change in between read and write operations without that clumsy previous hash list and previous chunk size.

src/file_system.c

@@ -57,6 +57,44 @@ void file_close(Handle fd)
 #endif
 }
 
+int file_set_offset(Handle fd, int off)
+{
+#ifdef __linux__
+    off_t ret = sys_lseek((int) fd.data, off, SEEK_SET);
+    if (ret < 0)
+        return -1;
+    return 0;
+#endif
+
+#ifdef _WIN32
+    LARGE_INTEGER distance;
+    distance.QuadPart = off;
+    if (!sys_SetFilePointer((HANDLE) fd.data, distance.LowPart, &distance.HighPart, FILE_BEGIN))
+        if (GetLastError() != NO_ERROR)
+            return -1;
+    return 0;
+#endif
+}
+
+int file_get_offset(Handle fd, int *off)
+{
+#ifdef __linux__
+    off_t ret = sys_lseek((int) fd.data, 0, SEEK_CUR);
+    if (ret < 0)
+        return -1;
+    *off = (int) ret;
+    return 0;
+#endif
+
+#ifdef _WIN32
+    DWORD pos = sys_SetFilePointer((HANDLE) fd.data, 0, NULL, FILE_CURRENT);
+    if (pos == INVALID_SET_FILE_POINTER && GetLastError() != NO_ERROR)
+        return -1;
+    *off = (int) pos;
+    return 0;
+#endif
+}
+
 int file_lock(Handle fd)
 {
 #ifdef __linux__

src/file_system.h

@@ -28,6 +28,8 @@ typedef struct {
 
 int  file_open(string path, Handle *fd);
 void file_close(Handle fd);
+int  file_set_offset(Handle fd, int off);
+int  file_get_offset(Handle fd, int *off);
 int  file_lock(Handle fd);
 int  file_unlock(Handle fd);
 int  file_sync(Handle fd);

src/file_tree.c

@@ -1,3 +1,4 @@
+#include <limits.h>
 #include <assert.h>
 #include <string.h>
 #include <stdlib.h>
@@ -105,6 +106,7 @@ static void dir_free(Dir *d)
 
 static void dir_remove(Dir *d, int idx)
 {
+    // TODO: pretty sure this leaks memory
     d->children[idx] = d->children[--d->num_children];
 }
 
@@ -373,7 +375,12 @@ int file_tree_write(FileTree *ft, string path,
     for (uint64_t i = first_chunk_index; i <= last_chunk_index; i++)
         f->chunks[i] = hashes[i - first_chunk_index];
 
-    // Now check which old hashes are no longer used anywhere in the tree
+    // Now check which old hashes are no longer used
+    // anywhere in the tree
+    //
+    // NOTE: If removed_hashes is NULL, the caller isn't
+    //       interested in which hashes are no longer reachable.
+    if (removed_hashes != NULL) {
         *num_removed = 0;
         for (uint64_t i = first_chunk_index; i <= last_chunk_index; i++) {
             SHA256 old_hash = prev_hashes[i - first_chunk_index];
@@ -397,6 +404,7 @@ int file_tree_write(FileTree *ft, string path,
                 (*num_removed)++;
             }
         }
+    }
 
     return 0;
 }
@@ -466,3 +474,221 @@ string file_tree_strerror(int code)
     }
     return S("Unknown error");
 }
+
+typedef struct {
+    int (*write_fn)(char*,int,void*);
+    void *write_data;
+    char *buffer;
+    int   buffer_size;
+    int   buffer_used;
+    bool  error;
+} SerializeContext;
+
+static void sc_flush(SerializeContext *sc)
+{
+    if (sc->error)
+        return;
+
+    int ret = sc->write_fn(sc->buffer, sc->buffer_used, sc->write_data);
+    if (ret < 0) {
+        sc->error = true;
+        return;
+    }
+
+    sc->buffer_used = 0;
+}
+
+static void sc_write_mem(SerializeContext *sc, char *src, int len)
+{
+    if (sc->error)
+        return;
+
+    if (sc->buffer_size - sc->buffer_used < len) {
+
+        if (len > sc->buffer_size) {
+            sc->error = true;
+            return;
+        }
+
+        sc_flush(sc);
+        if (sc->error)
+            return;
+    }
+
+    memcpy(sc->buffer + sc->buffer_used, src, len);
+    sc->buffer_used += len;
+}
+static void sc_write_u8  (SerializeContext *sc, uint8_t  value) { sc_write_mem(sc, (char*) &value, (int) sizeof(value)); }
+static void sc_write_u16 (SerializeContext *sc, uint16_t value) { sc_write_mem(sc, (char*) &value, (int) sizeof(value)); }
+static void sc_write_u64 (SerializeContext *sc, uint64_t value) { sc_write_mem(sc, (char*) &value, (int) sizeof(value)); }
+static void sc_write_hash(SerializeContext *sc, SHA256   value) { sc_write_mem(sc, (char*) &value, (int) sizeof(value)); }
+
+static void file_serialize(SerializeContext *sc, File *f)
+{
+    sc_write_u64(sc, f->chunk_size);
+    sc_write_u64(sc, f->num_chunks);
+    for (uint64_t i = 0; i < f->num_chunks; i++)
+        sc_write_hash(sc, f->chunks[i]);
+}
+
+static void entity_serialize(SerializeContext *sc, Entity *e);
+
+static void dir_serialize(SerializeContext *sc, Dir *d)
+{
+    sc_write_u64(sc, d->num_children);
+    for (uint64_t i = 0; i < d->num_children; i++)
+        entity_serialize(sc, &d->children[i]);
+}
+
+static void entity_serialize(SerializeContext *sc, Entity *e)
+{
+    sc_write_u16(sc, e->name_len);
+    sc_write_mem(sc, e->name, e->name_len);
+    sc_write_u8(sc, e->is_dir);
+    if (e->is_dir)
+        dir_serialize(sc, &e->d);
+    else
+        file_serialize(sc, &e->f);
+}
+
+int file_tree_serialize(FileTree *ft, int (*write_fn)(char*,int,void*), void *write_data)
+{
+    SerializeContext sc;
+    sc.write_fn = write_fn;
+    sc.write_data = write_data;
+    sc.buffer_used = 0;
+    sc.buffer_size = 1<<10;
+    sc.buffer = sys_malloc(sc.buffer_size);
+    sc.error = false;
+    if (sc.buffer == NULL)
+        sc.error = true;
+    entity_serialize(&sc, &ft->root);
+    sc_flush(&sc);
+    sys_free(sc.buffer);
+    if (sc.error)
+        return -1;
+    return 0;
+}
+
+typedef struct {
+    int (*read_fn)(char*,int,void*);
+    void *read_data;
+    char *buffer;
+    int   buffer_size;
+    int   buffer_used;
+    int   buffer_head;
+    bool  error;
+    uint64_t total_read;
+} DeserializeContext;
+
+static void dc_read_mem(DeserializeContext *dc, void *dst, int len)
+{
+    if (dc->error)
+        return;
+
+    if (dc->buffer_used < len) {
+
+        if (dc->buffer_size < len) {
+            dc->error = true;
+            return;
+        }
+
+        memmove(dc->buffer, dc->buffer + dc->buffer_head, dc->buffer_used);
+        dc->buffer_head = 0;
+
+        int ret = dc->read_fn(
+            dc->buffer      + dc->buffer_used,
+            dc->buffer_size - dc->buffer_used,
+            dc->read_data);
+        if (ret < 0) {
+            dc->error = true;
+            return;
+        }
+        dc->buffer_used += ret;
+
+        if (dc->buffer_used < len) {
+            dc->error = true;
+            return;
+        }
+    }
+
+    memcpy(dst, dc->buffer + dc->buffer_head, len);
+    dc->buffer_head += len;
+    dc->buffer_used -= len;
+    dc->total_read  += len;
+}
+static void dc_read_u8 (DeserializeContext *dc,  uint8_t  *dst) { dc_read_mem(dc, dst, sizeof(*dst)); }
+static void dc_read_u16(DeserializeContext *dc,  uint16_t *dst) { dc_read_mem(dc, dst, sizeof(*dst)); }
+static void dc_read_u64(DeserializeContext *dc,  uint64_t *dst) { dc_read_mem(dc, dst, sizeof(*dst)); }
+static void dc_read_hash(DeserializeContext *dc, SHA256   *dst) { dc_read_mem(dc, dst, sizeof(*dst)); }
+
+static void file_deserialize(DeserializeContext *dc, File *f)
+{
+    dc_read_u64(dc, &f->chunk_size);
+    dc_read_u64(dc, &f->num_chunks);
+
+    f->chunks = sys_malloc(f->num_chunks * sizeof(SHA256));
+    if (f->chunks == NULL) {
+        assert(0); // TODO
+    }
+
+    for (uint64_t i = 0; i < f->num_chunks; i++)
+        dc_read_hash(dc, &f->chunks[i]);
+}
+
+static void entity_deserialize(DeserializeContext *dc, Entity *e);
+
+static void dir_deserialize(DeserializeContext *dc, Dir *d)
+{
+    dc_read_u64(dc, &d->num_children);
+
+    d->max_children = d->num_children;
+    d->children = sys_malloc(d->num_children * sizeof(Entity));
+    if (d->children == NULL) {
+        assert(0); // TODO
+    }
+
+    // TODO: not checking for errors is not okay as
+    //       the code will branch based on garbage
+    //       values.
+    for (uint64_t i = 0; i < d->num_children; i++)
+        entity_deserialize(dc, &d->children[i]);
+}
+
+static void entity_deserialize(DeserializeContext *dc, Entity *e)
+{
+    dc_read_u16(dc, &e->name_len); // TODO: make sure this doesn't go over the static buffer
+    dc_read_mem(dc, e->name, e->name_len);
+
+    uint8_t is_dir;
+    dc_read_u8 (dc, &is_dir);
+    e->is_dir = (is_dir != 0);
+
+    if (e->is_dir)
+        dir_deserialize(dc, &e->d);
+    else
+        file_deserialize(dc, &e->f);
+}
+
+int file_tree_deserialize(FileTree *ft, int (*read_fn)(char*,int,void*), void *read_data)
+{
+    DeserializeContext dc;
+    dc.read_fn = read_fn;
+    dc.read_data = read_data;
+    dc.buffer_head = 0;
+    dc.buffer_used = 0;
+    dc.buffer_size = 1<<10;
+    dc.buffer = sys_malloc(dc.buffer_size);
+    dc.error = false;
+    if (dc.buffer == NULL)
+        dc.error = true;
+    dc.total_read = 0;
+    entity_deserialize(&dc, &ft->root);
+    sys_free(dc.buffer);
+    if (dc.error)
+        return -1;
+    if (dc.total_read > INT_MAX) {
+        assert(0); // TODO
+    }
+    return dc.total_read;
+}

src/file_tree.h

@@ -16,8 +16,8 @@ enum {
 typedef struct Entity Entity;
 
 typedef struct {
-    uint64_t chunk_size;
+    uint64_t chunk_size; // TODO: this should be an u32
-    uint64_t num_chunks;
+    uint64_t num_chunks; // TODO: and this too
     SHA256 *chunks;
 } File;
 
@@ -49,14 +49,16 @@ typedef struct {
 
 #define MAX_COMPS 32
 
-int    file_tree_init(FileTree *ft);
+int    file_tree_init          (FileTree *ft);
-void   file_tree_free(FileTree *ft);
+void   file_tree_free          (FileTree *ft);
-bool   file_tree_uses_hash(FileTree *ft, SHA256 hash);
+bool   file_tree_uses_hash     (FileTree *ft, SHA256 hash);
-int    file_tree_list(FileTree *ft, string path, ListItem *items, int max_items);
+int    file_tree_list          (FileTree *ft, string path, ListItem *items, int max_items);
-int    file_tree_create_entity(FileTree *ft, string path, bool is_dir, uint64_t chunk_size);
+int    file_tree_create_entity (FileTree *ft, string path, bool is_dir, uint64_t chunk_size);
-int    file_tree_delete_entity(FileTree *ft, string path);
+int    file_tree_delete_entity (FileTree *ft, string path);
-int    file_tree_write(FileTree *ft, string path, uint64_t off, uint64_t len, uint32_t num_chunks, uint32_t chunk_size, SHA256 *prev_hashes, SHA256 *hashes, SHA256 *removed_hashes, int *num_removed);
+int    file_tree_write         (FileTree *ft, string path, uint64_t off, uint64_t len, uint32_t num_chunks, uint32_t chunk_size, SHA256 *prev_hashes, SHA256 *hashes, SHA256 *removed_hashes, int *num_removed);
-int    file_tree_read(FileTree *ft, string path, uint64_t off, uint64_t len, uint64_t *chunk_size, SHA256 *hashes, int max_hashes);
+int    file_tree_read          (FileTree *ft, string path, uint64_t off, uint64_t len, uint64_t *chunk_size, SHA256 *hashes, int max_hashes);
-string file_tree_strerror(int code);
+string file_tree_strerror      (int code);
+int    file_tree_serialize     (FileTree *ft, int (*flush_fn)(char*,int,void*), void *flush_data);
+int    file_tree_deserialize   (FileTree *ft, int (*read_fn)(char*,int,void*), void *read_data);
 
 #endif // FILE_TREE_INCLUDED

src/metadata_server.c

@@ -179,6 +179,10 @@ process_client_create(MetadataServer *state, int conn_idx, ByteView msg)
     if (binary_read(&reader, NULL, 1))
         return -1;
 
+    if (wal_append_create(&state->wal, path, is_dir, chunk_size) < 0) {
+        assert(0); // TODO
+    }
+
     int ret = file_tree_create_entity(&state->file_tree, path, is_dir, chunk_size);
 
     if (ret < 0) {
@@ -238,6 +242,10 @@ process_client_delete(MetadataServer *state, int conn_idx, ByteView msg)
     if (binary_read(&reader, NULL, 1))
         return -1;
 
+    if (wal_append_delete(&state->wal, path) < 0) {
+        assert(0); // TODO
+    }
+
     // TODO: return unused hashes and add them to the ms_rem_list of holder chunk servers
     int ret = file_tree_delete_entity(&state->file_tree, path);
 
@@ -597,6 +605,10 @@ process_client_write(MetadataServer *state, int conn_idx, ByteView msg)
         new_hashes[i] = results[i].new_hash;
     }
 
+    if (wal_append_write(&state->wal, path, offset, length, num_chunks, chunk_size, old_hashes, new_hashes) < 0) {
+        assert(0); // TODO
+    }
+
     int ret = file_tree_write(&state->file_tree, path, offset, length,
         num_chunks, chunk_size, old_hashes, new_hashes, removed_hashes, &num_removed);
 
@@ -971,10 +983,15 @@ int metadata_server_init(MetadataServer *state, int argc, char **argv, void **co
     string addr      = getargs(argc, argv, "--addr", "127.0.0.1");
     int    port      = getargi(argc, argv, "--port", 8080);
     bool   trace     = getargb(argc, argv, "--trace");
+    string wal_file  = getargs(argc, argv, "--wal-file", "metadata.wal");
+    int    wal_limit = getargi(argc, argv, "--wal-limit", 1000); // TODO: Choose a good default limit
 
     if (port <= 0 || port >= 1<<16)
         return -1;
 
+    if (wal_limit < 0)
+        return -1;
+
     state->trace = trace;
     state->replication_factor = 3; // TODO: what about the REPLICATION_FACTOR macro?
     if (state->replication_factor > MAX_CHUNK_SERVERS)
@@ -998,6 +1015,10 @@ int metadata_server_init(MetadataServer *state, int argc, char **argv, void **co
         return -1;
     }
 
+    if (wal_open(&state->wal, &state->file_tree, wal_file, wal_limit) < 0) {
+        assert(0); // TODO
+    }
+
     printf("Metadata server set up (local=%.*s:%d)\n",
         addr.len,
         addr.ptr,
@@ -1010,6 +1031,7 @@ int metadata_server_init(MetadataServer *state, int argc, char **argv, void **co
 
 int metadata_server_free(MetadataServer *state)
 {
+    wal_close(&state->wal);
     file_tree_free(&state->file_tree);
     tcp_context_free(&state->tcp);
     return 0;

src/metadata_server.h

@@ -2,6 +2,7 @@
 #define METADATA_SERVER_INCLUDED
 
 #include "tcp.h"
+#include "wal.h"
 #include "file_tree.h"
 #include "config.h"
 #include "basic.h"
@@ -39,6 +40,7 @@ typedef struct {
 typedef struct {
 
     TCP tcp;
+    WAL wal;
 
     FileTree file_tree;
 

src/simulation_client.c

@@ -61,7 +61,7 @@ int simulation_client_step(SimulationClient *client, void **contexts,
     for (int i = 0; i < client->num_pending; i++) {
 
         ToastyResult result;
-        if (!toasty_get_result(client->toasty, client->pending[i].handle, &result))
+        if (toasty_get_result(client->toasty, client->pending[i].handle, &result) != 0)
             continue;
 
         PendingOperation pending = client->pending[i];

src/system.c

@@ -587,7 +587,7 @@ static int setup_poll_array(void **contexts, struct pollfd *polled)
         switch (desc->type) {
 
             case DESC_FILE:
-            assert(0); // TODO: error
+            // Ignore
             break;
 
             case DESC_SOCKET:
@@ -1493,6 +1493,24 @@ BOOL mock_WriteFile(HANDLE handle, char *src, DWORD len, DWORD *num, OVERLAPPED
     return WriteFile(desc->real_fd, src, len, num, ov);
 }
 
+DWORD mock_SetFilePointer(HANDLE hFile, LONG lDistanceToMove, PLONG lpDistanceToMoveHigh, DWORD dwMoveMethod)
+{
+    if (hFile == INVALID_HANDLE_VALUE || (int)hFile < 0 || (int)hFile >= MAX_DESCRIPTORS) {
+        SetLastError(ERROR_INVALID_HANDLE);
+        return INVALID_SET_FILE_POINTER;
+    }
+    int idx = (int) hFile;
+
+    Descriptor *desc = &current_process->desc[idx];
+    if (desc->type != DESC_FILE) {
+        SetLastError(ERROR_INVALID_HANDLE);
+        return INVALID_SET_FILE_POINTER;
+    }
+
+    // Forward to real SetFilePointer, last error is set by the real call
+    return SetFilePointer(desc->real_fd, lDistanceToMove, lpDistanceToMoveHigh, dwMoveMethod);
+}
+
 BOOL mock_GetFileSizeEx(HANDLE handle, LARGE_INTEGER *buf)
 {
     if (handle == INVALID_HANDLE_VALUE || (int)handle < 0 || (int)handle >= MAX_DESCRIPTORS) {
@@ -1648,6 +1666,12 @@ BOOL mock_FindClose(HANDLE hFindFile)
     return TRUE;
 }
 
+BOOL mock_MoveFileExW(WCHAR *lpExistingFileName, WCHAR *lpNewFileName, DWORD dwFlags)
+{
+    // Forward to real MoveFileExW, last error is set by the real call
+    return MoveFileExW(lpExistingFileName, lpNewFileName, dwFlags);
+}
+
 #else
 
 int mock_clock_gettime(clockid_t clockid, struct timespec *tp)
@@ -1785,6 +1809,24 @@ int mock_write(int fd, char *src, int len)
     }
 }
 
+off_t mock_lseek(int fd, off_t offset, int whence)
+{
+    if (fd < 0 || fd >= MAX_DESCRIPTORS) {
+        errno = EBADF;  // Bad file descriptor
+        return -1;
+    }
+    int idx = fd;
+
+    Descriptor *desc = &current_process->desc[idx];
+    if (desc->type != DESC_FILE) {
+        errno = EBADF;  // Not a file descriptor
+        return -1;
+    }
+
+    // Forward to real lseek, errno is set by the real call
+    return lseek(desc->real_fd, offset, whence);
+}
+
 int mock_fstat(int fd, struct stat *buf)
 {
     if (fd < 0 || fd >= MAX_DESCRIPTORS) {

src/system.h

@@ -65,6 +65,7 @@ BOOL     mock_UnlockFile(HANDLE hFile, DWORD dwFileOffsetLow, DWORD dwFileOffset
 BOOL     mock_FlushFileBuffers(HANDLE handle);
 BOOL     mock_ReadFile(HANDLE handle, char *dst, DWORD len, DWORD *num, OVERLAPPED *ov);
 BOOL     mock_WriteFile(HANDLE handle, char *src, DWORD len, DWORD *num, OVERLAPPED *ov);
+DWORD    mock_SetFilePointer(HANDLE hFile, LONG lDistanceToMove, PLONG lpDistanceToMoveHigh, DWORD dwMoveMethod);
 BOOL     mock_GetFileSizeEx(HANDLE handle, LARGE_INTEGER *buf);
 BOOL     mock_QueryPerformanceCounter(LARGE_INTEGER *lpPerformanceCount);
 BOOL     mock_QueryPerformanceFrequency(LARGE_INTEGER *lpFrequency);
@@ -73,6 +74,7 @@ int      mock__mkdir(char *path);
 HANDLE   mock_FindFirstFileA(char *lpFileName, WIN32_FIND_DATAA *lpFindFileData);
 BOOL     mock_FindNextFileA(HANDLE hFindFile, WIN32_FIND_DATAA *lpFindFileData);
 BOOL     mock_FindClose(HANDLE hFindFile);
+BOOL     mock_MoveFileExW(WCHAR *lpExistingFileName, WCHAR *lpNewFileName, DWORD dwFlags);
 #else
 int     mock_clock_gettime(clockid_t clockid, struct timespec *tp);
 int     mock_open(char *path, int flags, int mode);
@@ -81,6 +83,7 @@ int     mock_flock(int fd, int op);
 int     mock_fsync(int fd);
 int     mock_read(int fd, char *dst, int len);
 int     mock_write(int fd, char *src, int len);
+off_t   mock_lseek(int fd, off_t offset, int whence);
 int     mock_fstat(int fd, struct stat *buf);
 int     mock_mkstemp(char *path);
 char*   mock_realpath(char *path, char *dst);
@@ -117,6 +120,7 @@ int     mock_closedir(DIR *dirp);
 #define sys_FlushFileBuffers mock_FlushFileBuffers
 #define sys_ReadFile         mock_ReadFile
 #define sys_WriteFile        mock_WriteFile
+#define sys_SetFilePointer   mock_SetFilePointer
 #define sys_GetFileSizeEx    mock_GetFileSizeEx
 #define sys__fullpath        mock__fullpath
 #define sys_QueryPerformanceCounter   mock_QueryPerformanceCounter
@@ -124,6 +128,7 @@ int     mock_closedir(DIR *dirp);
 #define sys_FindFirstFileA   mock_FindFirstFileA
 #define sys_FindNextFileA    mock_FindNextFileA
 #define sys_FindClose        mock_FindClose
+#define sys_MoveFileExW      mock_MoveFileExW
 
 // Linux
 #define sys_mkdir            mock_mkdir
@@ -133,6 +138,7 @@ int     mock_closedir(DIR *dirp);
 #define sys_fsync            mock_fsync
 #define sys_read             mock_read
 #define sys_write            mock_write
+#define sys_lseek            mock_lseek
 #define sys_fstat            mock_fstat
 #define sys_mkstemp          mock_mkstemp
 #define sys_realpath         mock_realpath
@@ -171,6 +177,7 @@ int     mock_closedir(DIR *dirp);
 #define sys_FlushFileBuffers FlushFileBuffers
 #define sys_ReadFile         ReadFile
 #define sys_WriteFile        WriteFile
+#define sys_SetFilePointer   SetFilePointer
 #define sys_GetFileSizeEx    GetFileSizeEx
 #define sys__fullpath        _fullpath
 #define sys_QueryPerformanceCounter   QueryPerformanceCounter
@@ -178,6 +185,7 @@ int     mock_closedir(DIR *dirp);
 #define sys_FindFirstFileA   FindFirstFileA
 #define sys_FindNextFileA    FindNextFileA
 #define sys_FindClose        FindClose
+#define sys_MoveFileExW      MoveFileExW
 
 // Linux
 #define sys_mkdir            mkdir
@@ -187,6 +195,7 @@ int     mock_closedir(DIR *dirp);
 #define sys_fsync            fsync
 #define sys_read             read
 #define sys_write            write
+#define sys_lseek            lseek
 #define sys_fstat            fstat
 #define sys_mkstemp          mkstemp
 #define sys_realpath         realpath

src/wal.c

@@ -0,0 +1,639 @@
+#include <stddef.h>
+#include <limits.h>
+#include <assert.h>
+#include <string.h>
+
+#include "wal.h"
+#include "file_system.h"
+#include "file_tree.h"
+#include "system.h"
+
+#define WAL_MAGIC   0xcafebebe
+#define WAL_VERSION 1
+
+typedef struct {
+    uint32_t magic;
+    uint32_t version;
+    uint64_t reserved;
+} WALHeader;
+
+typedef enum {
+    WAL_ENTRY_CREATE,
+    WAL_ENTRY_DELETE,
+    WAL_ENTRY_WRITE,
+} WALEntryType;
+
+typedef struct {
+    WALEntryType type;
+
+    // create, delete, write
+    string path;
+
+    // create, write
+    uint64_t chunk_size;
+
+    // create
+    bool is_dir;
+
+    // write
+    uint64_t offset;
+    uint64_t length;
+    uint32_t num_chunks;
+    SHA256 *prev_hashes;
+    SHA256 *next_hashes;
+
+} WALEntry;
+
+static int write_exact(Handle handle, char *src, int len)
+{
+    int copied = 0;
+    while (copied < len) {
+        int ret = file_write(handle, src + copied, len - copied);
+        if (ret < 0)
+            return -1;
+        copied += ret;
+    }
+    return 0;
+}
+
+typedef struct {
+    Handle handle;
+} WriteSnapshotContext;
+
+static int
+serialize_callback(char *src, int num, void *data)
+{
+    WriteSnapshotContext *wsc = data;
+    return write_exact(wsc->handle, src, num);
+}
+
+static int write_snapshot(FileTree *file_tree, Handle handle)
+{
+    WriteSnapshotContext wsc;
+    wsc.handle = handle;
+    if (file_tree_serialize(file_tree, serialize_callback, &wsc) < 0)
+        return -1;
+    return 0;
+}
+
+typedef struct {
+    Handle handle;
+} ReadSnapshotContext;
+
+static int
+deserialize_callback(char *dst, int num, void *data)
+{
+    ReadSnapshotContext *rsc = data;
+    int copied = 0;
+    while (copied < num) {
+        int ret = file_read(rsc->handle, dst + copied, num - copied);
+        if (ret < 0)
+            return -1;
+        if (ret == 0)
+            break;
+        copied += ret;
+    }
+    return copied;
+}
+
+static int read_snapshot(FileTree *file_tree, Handle handle)
+{
+    ReadSnapshotContext rsc;
+    rsc.handle = handle;
+    int num = file_tree_deserialize(file_tree, deserialize_callback, &rsc);
+    if (num < 0)
+        return -1;
+    return num;
+}
+
+static int swap_file(WAL *wal)
+{
+    // Create a temporary file path
+    char temp_path_buf[1<<10];
+    if (wal->file_path.len + 5 > (int) sizeof(temp_path_buf))
+        return -1;
+
+    memcpy(temp_path_buf, wal->file_path.ptr, wal->file_path.len);
+    memcpy(temp_path_buf + wal->file_path.len, ".tmp", 4);
+    temp_path_buf[wal->file_path.len + 4] = '\0';
+
+    string temp_path = { temp_path_buf, wal->file_path.len + 4 };
+
+    // Create and open the temporary file
+    Handle temp_handle;
+    if (file_open(temp_path, &temp_handle) < 0)
+        return -1;
+
+    // Write the WAL header
+    WALHeader header;
+    header.magic = WAL_MAGIC;
+    header.version = WAL_VERSION;
+    header.reserved = 0;
+
+    if (write_exact(temp_handle, (char*) &header, sizeof(header)) < 0) {
+        file_close(temp_handle);
+        remove_file_or_dir(temp_path);
+        return -1;
+    }
+
+    // Serialize the current file tree to the new file
+    if (write_snapshot(wal->file_tree, temp_handle) < 0) {
+        file_close(temp_handle);
+        remove_file_or_dir(temp_path);
+        return -1;
+    }
+
+    // Sync the temporary file to ensure it's written to disk
+    if (file_sync(temp_handle) < 0) {
+        file_close(temp_handle);
+        remove_file_or_dir(temp_path);
+        return -1;
+    }
+
+    // Lock the new file before switching
+    if (file_lock(temp_handle) < 0) {
+        file_close(temp_handle);
+        remove_file_or_dir(temp_path);
+        return -1;
+    }
+
+    // Unlock and close the old file
+    file_unlock(wal->handle);
+    file_close(wal->handle);
+
+    // Atomically rename the temporary file to replace the old file
+    // On Unix: rename() atomically replaces the destination
+    // On Windows: we use MoveFileEx with MOVEFILE_REPLACE_EXISTING for atomicity
+#ifdef _WIN32
+    // On Windows, use MoveFileEx for atomic replace
+    char old_path_zt[1<<10];
+    if (wal->file_path.len >= (int) sizeof(old_path_zt)) {
+        file_unlock(temp_handle);
+        file_close(temp_handle);
+        remove_file_or_dir(temp_path);
+        return -1;
+    }
+    memcpy(old_path_zt, wal->file_path.ptr, wal->file_path.len);
+    old_path_zt[wal->file_path.len] = '\0';
+
+    WCHAR old_path_w[MAX_PATH];
+    WCHAR temp_path_w[MAX_PATH];
+
+    if (!MultiByteToWideChar(CP_UTF8, 0, old_path_zt, -1, old_path_w, MAX_PATH) ||
+        !MultiByteToWideChar(CP_UTF8, 0, temp_path_buf, -1, temp_path_w, MAX_PATH)) {
+        file_unlock(temp_handle);
+        file_close(temp_handle);
+        remove_file_or_dir(temp_path);
+        return -1;
+    }
+
+    // MOVEFILE_REPLACE_EXISTING allows atomic overwrite
+    if (!sys_MoveFileExW(temp_path_w, old_path_w, MOVEFILE_REPLACE_EXISTING)) {
+        file_unlock(temp_handle);
+        file_close(temp_handle);
+        remove_file_or_dir(temp_path);
+        return -1;
+    }
+#else
+    // On Unix/Linux, rename() atomically replaces the destination
+    if (rename_file_or_dir(temp_path, wal->file_path) < 0) {
+        file_unlock(temp_handle);
+        file_close(temp_handle);
+        remove_file_or_dir(temp_path);
+        return -1;
+    }
+#endif
+
+    // Update the WAL to use the new file handle
+    wal->handle = temp_handle;
+    wal->entry_count = 0;
+
+    return 0;
+}
+
+static int read_exact(Handle handle, char *dst, int len)
+{
+    int copied = 0;
+    while (copied < len) {
+        int ret = file_read(handle, dst + copied, len - copied);
+        if (ret < 0)
+            return -1;
+        if (ret == 0)
+            return 0; // EOF
+        copied += ret;
+    }
+    return copied;
+}
+
+static int read_u8(Handle handle, uint8_t *value)
+{
+    return read_exact(handle, (char*) value, sizeof(*value));
+}
+
+static int read_u16(Handle handle, uint16_t *value)
+{
+    return read_exact(handle, (char*) value, sizeof(*value));
+}
+
+static int read_u32(Handle handle, uint32_t *value)
+{
+    return read_exact(handle, (char*) value, sizeof(*value));
+}
+
+static int read_u64(Handle handle, uint64_t *value)
+{
+    return read_exact(handle, (char*) value, sizeof(*value));
+}
+
+static int next_entry(Handle handle, WALEntry *entry)
+{
+    // Initialize pointers to NULL for cleanup on error
+    entry->path.ptr = NULL;
+    entry->prev_hashes = NULL;
+    entry->next_hashes = NULL;
+
+    uint8_t type;
+    int ret = read_u8(handle, &type);
+    if (ret == 0)
+        return 0; // EOF
+    if (ret < 0)
+        return -1;
+
+    entry->type = (WALEntryType) type;
+
+    uint16_t path_len;
+    if (read_u16(handle, &path_len) <= 0)
+        return -1;
+
+    // Dynamically allocate path buffer
+    char *path_buffer = sys_malloc(path_len);
+    if (!path_buffer)
+        return -1;
+
+    if (read_exact(handle, path_buffer, path_len) <= 0) {
+        sys_free(path_buffer);
+        return -1;
+    }
+
+    entry->path.ptr = path_buffer;
+    entry->path.len = path_len;
+
+    switch (entry->type) {
+    case WAL_ENTRY_CREATE:
+        {
+            uint8_t is_dir;
+            if (read_u8(handle, &is_dir) <= 0)
+                goto cleanup_error;
+            entry->is_dir = is_dir;
+
+            if (!is_dir) {
+                if (read_u64(handle, &entry->chunk_size) <= 0)
+                    goto cleanup_error;
+            } else {
+                entry->chunk_size = 0;
+            }
+        }
+        break;
+
+    case WAL_ENTRY_DELETE:
+        // No additional fields
+        break;
+
+    case WAL_ENTRY_WRITE:
+        {
+            if (read_u64(handle, &entry->offset) <= 0)
+                goto cleanup_error;
+            if (read_u64(handle, &entry->length) <= 0)
+                goto cleanup_error;
+            if (read_u32(handle, &entry->num_chunks) <= 0)
+                goto cleanup_error;
+            if (read_u32(handle, (uint32_t*) &entry->chunk_size) <= 0)
+                goto cleanup_error;
+
+            // Dynamically allocate hash buffers
+            SHA256 *prev_hashes_buffer = sys_malloc(entry->num_chunks * sizeof(SHA256));
+            SHA256 *next_hashes_buffer = sys_malloc(entry->num_chunks * sizeof(SHA256));
+
+            if (!prev_hashes_buffer || !next_hashes_buffer) {
+                if (prev_hashes_buffer)
+                    sys_free(prev_hashes_buffer);
+                if (next_hashes_buffer)
+                    sys_free(next_hashes_buffer);
+                goto cleanup_error;
+            }
+
+            if (read_exact(handle, (char*) prev_hashes_buffer, entry->num_chunks * sizeof(SHA256)) <= 0) {
+                sys_free(prev_hashes_buffer);
+                sys_free(next_hashes_buffer);
+                goto cleanup_error;
+            }
+            if (read_exact(handle, (char*) next_hashes_buffer, entry->num_chunks * sizeof(SHA256)) <= 0) {
+                sys_free(prev_hashes_buffer);
+                sys_free(next_hashes_buffer);
+                goto cleanup_error;
+            }
+
+            entry->prev_hashes = prev_hashes_buffer;
+            entry->next_hashes = next_hashes_buffer;
+        }
+        break;
+
+    default:
+        goto cleanup_error;
+    }
+
+    return 1;
+
+cleanup_error:
+    if (entry->path.ptr)
+        sys_free((char*) entry->path.ptr);
+    if (entry->prev_hashes)
+        sys_free(entry->prev_hashes);
+    if (entry->next_hashes)
+        sys_free(entry->next_hashes);
+    return -1;
+}
+
+int wal_open(WAL *wal, FileTree *file_tree, string file_path, int entry_limit)
+{
+    wal->entry_count = 0;
+    wal->entry_limit = entry_limit;
+    wal->file_tree = file_tree;
+    wal->file_path.ptr = NULL;
+
+    // Copy file_path since the passed string may not have the same lifetime as WAL
+    char *path_copy = sys_malloc(file_path.len);
+    if (!path_copy)
+        return -1;
+    memcpy(path_copy, file_path.ptr, file_path.len);
+    wal->file_path.ptr = path_copy;
+    wal->file_path.len = file_path.len;
+
+    Handle handle;
+    if (file_open(file_path, &handle) < 0)
+        goto error_cleanup_path;
+
+    if (file_lock(handle) < 0) {
+        file_close(handle);
+        goto error_cleanup_path;
+    }
+
+    // Check if the file is empty (newly created) and initialize it
+    size_t size;
+    if (file_size(handle, &size) < 0) {
+        file_close(handle);
+        goto error_cleanup_path;
+    }
+
+    if (size == 0) {
+        // Initialize a new WAL file
+        WALHeader header;
+        header.magic = WAL_MAGIC;
+        header.version = WAL_VERSION;
+        header.reserved = 0;
+
+        if (write_exact(handle, (char*) &header, sizeof(header)) < 0) {
+            file_close(handle);
+            goto error_cleanup_path;
+        }
+
+        if (write_snapshot(file_tree, handle) < 0) {
+            file_close(handle);
+            goto error_cleanup_path;
+        }
+
+        if (file_sync(handle) < 0) {
+            file_close(handle);
+            goto error_cleanup_path;
+        }
+
+        // Reset to beginning after initialization
+        if (file_set_offset(handle, 0) < 0) {
+            file_close(handle);
+            goto error_cleanup_path;
+        }
+    }
+
+    // Read file header
+    // NOTE: For now we don't worry about fixing endianess
+    WALHeader header;
+    for (int copied = 0; copied < (int) sizeof(header); ) {
+        int ret = file_read(handle, (char*) &header + copied, (int) sizeof(header) - copied);
+        if (ret <= 0) {
+            file_close(handle); // TODO: what happens if I close a file without unlocking it?
+            goto error_cleanup_path;
+        }
+        copied += ret;
+    }
+
+    // Validate header fields
+    if (header.magic != WAL_MAGIC) {
+        file_close(handle);
+        goto error_cleanup_path;
+    }
+    if (header.version != WAL_VERSION) {
+        file_close(handle);
+        goto error_cleanup_path;
+    }
+
+    // The read_snapshot function may read more
+    // bytes than necessary from the buffer, so
+    // we need to save our current position to
+    // later restore it to this offset plus what
+    // read_snapshot really consumed.
+    int saved_offset;
+    if (file_get_offset(handle, &saved_offset) < 0) {
+        file_close(handle);
+        goto error_cleanup_path;
+    }
+
+    int num = read_snapshot(file_tree, handle);
+    if (num < 0) {
+        file_close(handle);
+        goto error_cleanup_path;
+    }
+
+    // Now restore the offset to the correct position.
+    if (num > INT_MAX - saved_offset) {
+        file_close(handle);
+        goto error_cleanup_path;
+    }
+    if (file_set_offset(handle, saved_offset + num) < 0) {
+        file_close(handle);
+        goto error_cleanup_path;
+    }
+
+    WALEntry entry;
+    for (;;) {
+
+        int ret = next_entry(handle, &entry);
+        if (ret == 0)
+            break;
+        if (ret < 0) {
+            file_close(handle);
+            goto error_cleanup_path;
+        }
+        assert(ret == 1);
+
+        switch (entry.type) {
+        case WAL_ENTRY_CREATE:
+            file_tree_create_entity(file_tree, entry.path, entry.is_dir, entry.chunk_size);
+            break;
+        case WAL_ENTRY_DELETE:
+            file_tree_delete_entity(file_tree, entry.path);
+            break;
+        case WAL_ENTRY_WRITE:
+            file_tree_write(file_tree, entry.path, entry.offset, entry.length, entry.num_chunks, entry.chunk_size, entry.prev_hashes, entry.next_hashes, NULL, NULL);
+            break;
+        default:
+            UNREACHABLE;
+        }
+
+        // Free dynamically allocated fields from next_entry
+        if (entry.path.ptr)
+            sys_free((char*) entry.path.ptr);
+        if (entry.prev_hashes)
+            sys_free(entry.prev_hashes);
+        if (entry.next_hashes)
+            sys_free(entry.next_hashes);
+
+        wal->entry_count++;
+    }
+
+    wal->handle = handle;
+
+    if (wal->entry_count >= wal->entry_limit) {
+        if (swap_file(wal) < 0) {
+            goto error_cleanup_path;
+        }
+    }
+    return 0;
+
+error_cleanup_path:
+    sys_free(path_copy);
+    return -1;
+}
+
+void wal_close(WAL *wal)
+{
+    file_unlock(wal->handle);
+    file_close(wal->handle);
+    if (wal->file_path.ptr)
+        sys_free((char*) wal->file_path.ptr);
+}
+
+static int write_u8(Handle handle, uint8_t value)
+{
+    return write_exact(handle, (char*) &value, sizeof(value));
+}
+
+static int write_u16(Handle handle, uint16_t value)
+{
+    return write_exact(handle, (char*) &value, sizeof(value));
+}
+
+static int write_u32(Handle handle, uint32_t value)
+{
+    return write_exact(handle, (char*) &value, sizeof(value));
+}
+
+static int write_u64(Handle handle, uint64_t value)
+{
+    return write_exact(handle, (char*) &value, sizeof(value));
+}
+
+static int write_str(Handle handle, string value)
+{
+    return write_exact(handle, value.ptr, value.len);
+}
+
+static int append_begin(WAL *wal)
+{
+    if (wal->entry_count >= wal->entry_limit) {
+        if (swap_file(wal) < 0)
+            return -1;
+    }
+    return 0;
+}
+
+static int append_end(WAL *wal)
+{
+    if (file_sync(wal->handle) < 0)
+        return -1;
+    wal->entry_count++;
+    return 0;
+}
+
+int wal_append_create(WAL *wal, string path, bool is_dir, uint64_t chunk_size)
+{
+    if (path.len > UINT16_MAX)
+        return -1;
+
+    if (append_begin(wal) < 0)
+        return -1;
+
+    write_u8(wal->handle, WAL_ENTRY_CREATE);
+    write_u16(wal->handle, path.len);
+    write_str(wal->handle, path);
+    write_u8(wal->handle, is_dir);
+    if (!is_dir)
+        write_u64(wal->handle, chunk_size);
+
+    if (append_end(wal) < 0)
+        return -1;
+
+    return 0;
+}
+
+int wal_append_delete(WAL *wal, string path)
+{
+    if (path.len > UINT16_MAX)
+        return -1;
+
+    if (append_begin(wal) < 0)
+        return -1;
+
+    write_u8(wal->handle, WAL_ENTRY_DELETE);
+    write_u16(wal->handle, path.len);
+    write_str(wal->handle, path);
+
+    if (append_end(wal) < 0)
+        return -1;
+
+    return 0;
+}
+
+int wal_append_write(WAL *wal, string path, uint64_t off,
+    uint64_t len, uint32_t num_chunks, uint32_t chunk_size,
+    SHA256 *prev_hashes, SHA256 *hashes)
+{
+    if (path.len > UINT16_MAX)
+        return -1;
+
+    if (append_begin(wal) < 0)
+        return -1;
+
+    if (write_u8(wal->handle, WAL_ENTRY_WRITE) < 0)
+        return -1;
+    if (write_u16(wal->handle, path.len) < 0)
+        return -1;
+    if (write_str(wal->handle, path) < 0)
+        return -1;
+    if (write_u64(wal->handle, off) < 0)
+        return -1;
+    if (write_u64(wal->handle, len) < 0)
+        return -1;
+    if (write_u32(wal->handle, num_chunks) < 0)
+        return -1;
+    if (write_u32(wal->handle, chunk_size) < 0)
+        return -1;
+    if (write_exact(wal->handle, (char*) prev_hashes, num_chunks * sizeof(SHA256)) < 0)
+        return -1;
+    if (write_exact(wal->handle, (char*) hashes, num_chunks * sizeof(SHA256)) < 0)
+        return -1;
+
+    if (append_end(wal) < 0)
+        return -1;
+
+    return 0;
+}

src/wal.h

@@ -0,0 +1,21 @@
+#ifndef WAL_INCLUDED
+#define WAL_INCLUDED
+
+#include "file_tree.h"
+#include "file_system.h"
+
+typedef struct {
+    Handle handle;
+    FileTree *file_tree;
+    string file_path;
+    int entry_count;
+    int entry_limit;
+} WAL;
+
+int  wal_open(WAL *wal, FileTree *file_tree, string file_path, int entry_limit);
+void wal_close(WAL *wal);
+int  wal_append_create(WAL *wal, string path, bool is_dir, uint64_t chunk_size);
+int  wal_append_delete(WAL *wal, string path);
+int  wal_append_write(WAL *wal, string path, uint64_t off, uint64_t len, uint32_t num_chunks, uint32_t chunk_size, SHA256 *prev_hashes, SHA256 *hashes);
+
+#endif // WAL_INCLUDED