atomic_store.c

#define _POSIX_C_SOURCE 200809L

#include "atomic_store.h"

#include <errno.h>
#include <fcntl.h>
#include <inttypes.h>
#include <stdint.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <sys/stat.h>
#include <sys/types.h>
#include <unistd.h>

/* ==== On-disk format =====================================================

File = zero or more snapshots.

Snapshot:
  [blob bytes][footer]

Footer (fixed size, little-endian):
  uint64_t magic      = 0x4157534E41505348ULL  ("AW SNAPH" arbitrary)
  uint64_t blob_len   = length in bytes of blob
  uint64_t checksum   = FNV-1a 64-bit over blob

Blob format (little-endian, simple packed):
  uint32_t file_count
  For each file:
    uint32_t path_len
    uint8_t  path[path_len]          (no NUL terminator)
    uint64_t data_len
    uint8_t  data[data_len]

We always append:
  write(blob), fsync(fd), write(footer), fsync(fd).
If crash happens before footer is fully written and fsynced,
last snapshot is ignored on next open because footer is invalid.

========================================================================= */

#define AW_MAGIC 0x4157534E41505348ULL /* "AWSNAPSH"-ish */

/* ==== FNV-1a 64-bit checksum ========================================== */

static uint64_t aw_fnv1a64(const void *buf, size_t len)
{
    const uint8_t *p = (const uint8_t *)buf;
    uint64_t h = 1469598103934665603ULL; 
    size_t i = 0;
    while (i < len)
    {
        h ^= (uint64_t)p[i];
        h *= 1099511628211ULL; /* FNV prime */
        i++;
    }
    return h;
}


static void write_u32_le(uint8_t *p, uint32_t v)
{
    p[0] = (uint8_t)(v & 0xFFu);
    p[1] = (uint8_t)((v >> 8) & 0xFFu);
    p[2] = (uint8_t)((v >> 16) & 0xFFu);
    p[3] = (uint8_t)((v >> 24) & 0xFFu);
}

static uint32_t read_u32_le(const uint8_t *p)
{
    return (uint32_t)p[0] | ((uint32_t)p[1] << 8) | ((uint32_t)p[2] << 16) | ((uint32_t)p[3] << 24);
}

static void write_u64_le(uint8_t *p, uint64_t v)
{
    p[0] = (uint8_t)(v & 0xFFu);
    p[1] = (uint8_t)((v >> 8) & 0xFFu);
    p[2] = (uint8_t)((v >> 16) & 0xFFu);
    p[3] = (uint8_t)((v >> 24) & 0xFFu);
    p[4] = (uint8_t)((v >> 32) & 0xFFu);
    p[5] = (uint8_t)((v >> 40) & 0xFFu);
    p[6] = (uint8_t)((v >> 48) & 0xFFu);
    p[7] = (uint8_t)((v >> 56) & 0xFFu);
}

static uint64_t read_u64_le(const uint8_t *p)
{
    return (uint64_t)p[0] | ((uint64_t)p[1] << 8) | ((uint64_t)p[2] << 16) | ((uint64_t)p[3] << 24) | ((uint64_t)p[4] << 32) | ((uint64_t)p[5] << 40) | ((uint64_t)p[6] << 48) | ((uint64_t)p[7] << 56);
}


struct aw_file_entry
{
    char *path;
    void *data;
    size_t len;
};

struct aw_store
{
    int fd;
    char *path;
    struct aw_file_entry *files;
    size_t file_count;
    size_t file_cap;
};

struct aw_tx
{
    aw_store_t *store;
    struct aw_file_entry *files;
    size_t file_count;
    size_t file_cap;
    int committed;
};


static void aw_free_entry(struct aw_file_entry *e)
{
    if (e == NULL)
        return;
    free(e->path);
    free(e->data);
    e->path = NULL;
    e->data = NULL;
    e->len = 0;
}

static void aw_free_entries(struct aw_file_entry *arr, size_t n)
{
    size_t i = 0;
    while (i < n)
    {
        aw_free_entry(&arr[i]);
        i++;
    }
}

static int aw_grow_entries(struct aw_file_entry **arr,
                           size_t *cap,
                           size_t new_count)
{
    if (new_count <= *cap)
    {
        return 0;
    }
    size_t new_cap = *cap == 0 ? 4 : *cap;
    while (new_cap < new_count)
    {
        new_cap = new_cap * 2;
    }
    struct aw_file_entry *tmp = (struct aw_file_entry *) realloc(*arr, new_cap * sizeof(*tmp));
    if (tmp == NULL)
    {
        return -1;
    }
    *arr = tmp;
    *cap = new_cap;
    return 0;
}

static int aw_find_entry(struct aw_file_entry *arr,
                         size_t n,
                         const char *path)
{
    size_t i = 0;
    while (i < n)
    {
        if (strcmp(arr[i].path, path) == 0)
        {
            return (int)i;
        }
        i++;
    }
    return -1;
}

static int aw_serialize_snapshot(struct aw_file_entry *files,
                                 size_t file_count,
                                 uint8_t **buf_out,
                                 size_t *len_out)
{
    size_t blob_len = 0;
    size_t i = 0;

    blob_len += 4;

    while (i < file_count)
    {
        size_t path_len = strlen(files[i].path);
        blob_len += 4;
        blob_len += path_len;
        blob_len += 8;
        blob_len += files[i].len;
        i++;
    }

    uint8_t *buf = (uint8_t *)malloc(blob_len);
    if (buf == NULL)
    {
        return -1;
    }

    uint8_t *p = buf;

    write_u32_le(p, (uint32_t)file_count);
    p += 4;

    i = 0;
    while (i < file_count)
    {
        size_t path_len = strlen(files[i].path);
        write_u32_le(p, (uint32_t)path_len);
        p += 4;
        memcpy(p, files[i].path, path_len);
        p += path_len;
        write_u64_le(p, (uint64_t)files[i].len);
        p += 8;
        if (files[i].len > 0 && files[i].data != NULL)
        {
            memcpy(p, files[i].data, files[i].len);
            p += files[i].len;
        }
        i++;
    }

    *buf_out = buf;
    *len_out = blob_len;
    return 0;
}

static int aw_parse_snapshot(const uint8_t *buf,
                             size_t len,
                             struct aw_file_entry **arr_out,
                             size_t *cnt_out)
{
    struct aw_file_entry *arr = NULL;
    size_t cap = 0;
    size_t count = 0;
    const uint8_t *p = buf;
    const uint8_t *end = buf + len;

    if ((size_t)(end - p) < 4)
    {
        errno = EINVAL;
        return -1;
    }

    uint32_t file_count = read_u32_le(p);
    p += 4;

    if (file_count == 0)
    {
        *arr_out = NULL;
        *cnt_out = 0;
        return 0;
    }

    if (aw_grow_entries(&arr, &cap, (size_t)file_count) != 0)
    {
        return -1;
    }

    while (count < (size_t)file_count)
    {
        if ((size_t)(end - p) < 4)
        {
            errno = EINVAL;
            aw_free_entries(arr, count);
            free(arr);
            return -1;
        }
        uint32_t path_len = read_u32_le(p);
        p += 4;

        if ((size_t)(end - p) < path_len)
        {
            errno = EINVAL;
            aw_free_entries(arr, count);
            free(arr);
            return -1;
        }
        char *path = (char *)malloc(path_len + 1);
        if (path == NULL)
        {
            aw_free_entries(arr, count);
            free(arr);
            return -1;
        }
        memcpy(path, p, path_len);
        path[path_len] = '\0';
        p += path_len;

        if ((size_t)(end - p) < 8)
        {
            errno = EINVAL;
            free(path);
            aw_free_entries(arr, count);
            free(arr);
            return -1;
        }
        uint64_t data_len64 = read_u64_le(p);
        p += 8;

        if ((uint64_t)(end - p) < data_len64)
        {
            errno = EINVAL;
            free(path);
            aw_free_entries(arr, count);
            free(arr);
            return -1;
        }

        size_t data_len = (size_t)data_len64;
        void *data = NULL;
        if (data_len > 0)
        {
            data = malloc(data_len);
            if (data == NULL)
            {
                free(path);
                aw_free_entries(arr, count);
                free(arr);
                return -1;
            }
            memcpy(data, p, data_len);
            p += data_len;
        }

        arr[count].path = path;
        arr[count].data = data;
        arr[count].len = data_len;
        count++;
    }

    *arr_out = arr;
    *cnt_out = count;
    return 0;
}

struct aw_footer
{
    uint64_t magic;
    uint64_t blob_len;
    uint64_t checksum;
};

static int aw_read_footer_at(int fd, off_t offset, struct aw_footer *out)
{
    uint8_t buf[24];
    ssize_t n = pread(fd, buf, sizeof(buf), offset);
    if (n != (ssize_t)sizeof(buf))
    {
        if (n < 0 && errno == EINTR)
        {
            /* retry logic */
        }
        return -1;
    }
    out->magic = read_u64_le(buf);
    out->blob_len = read_u64_le(buf + 8);
    out->checksum = read_u64_le(buf + 16);
    return 0;
}

static int aw_write_footer_append(int fd, uint64_t blob_len, uint64_t checksum)
{
    uint8_t buf[24];
    write_u64_le(buf, AW_MAGIC);
    write_u64_le(buf + 8, blob_len);
    write_u64_le(buf + 16, checksum);

    ssize_t n = write(fd, buf, sizeof(buf));
    if (n != (ssize_t)sizeof(buf))
    {
        return -1;
    }
    return 0;
}

static int aw_load_latest_snapshot(aw_store_t *st)
{
    struct stat stbuf;
    if (fstat(st->fd, &stbuf) != 0)
    {
        return -1;
    }

    off_t file_size = stbuf.st_size;
    if (file_size < (off_t)sizeof(struct aw_footer))
    {
        st->files = NULL;
        st->file_count = 0;
        st->file_cap = 0;
        return 0;
    }

    off_t cursor = file_size;

    int found = 0;
    struct aw_footer footer;
    uint8_t *blob = NULL;
    size_t blob_len = 0;

    while (!found && cursor >= (off_t)sizeof(struct aw_footer))
    {
        off_t footer_off = cursor - (off_t)sizeof(struct aw_footer);
        if (aw_read_footer_at(st->fd, footer_off, &footer) != 0)
        {
            return -1;
        }

        if (footer.magic != AW_MAGIC || footer.blob_len == 0)
        {
            /* Not valid here; move cursor back and try previous position. */
            cursor = footer_off;
            continue;
        }

        if (footer.blob_len > (uint64_t)footer_off)
        {
            cursor = footer_off;
            continue;
        }

        blob_len = (size_t)footer.blob_len;
        blob = (uint8_t *)malloc(blob_len);
        if (blob == NULL)
        {
            return -1;
        }

        off_t blob_off = footer_off - (off_t)blob_len;
        if (blob_off < 0)
        {
            free(blob);
            blob = NULL;
            cursor = footer_off;
            continue;
        }

        ssize_t n = pread(st->fd, blob, blob_len, blob_off);
        if (n != (ssize_t)blob_len)
        {
            free(blob);
            blob = NULL;
            cursor = footer_off;
            continue;
        }

        uint64_t chk = aw_fnv1a64(blob, blob_len);
        if (chk != footer.checksum)
        {
            free(blob);
            blob = NULL;
            cursor = footer_off;
            continue;
        }

        found = 1;
    }

    if (!found)
    {
        st->files = NULL;
        st->file_count = 0;
        st->file_cap = 0;
        return 0;
    }

    struct aw_file_entry *arr = NULL;
    size_t cnt = 0;
    int ok = aw_parse_snapshot(blob, blob_len, &arr, &cnt);
    free(blob);

    if (ok != 0)
    {
        return -1;
    }

    st->files = arr;
    st->file_count = cnt;
    st->file_cap = cnt;

    return 0;
}

aw_store_t *aw_store_open(const char *path, int create_if_missing)
{
    if (path == NULL)
    {
        errno = EINVAL;
        return NULL;
    }

    int flags = O_RDWR;
    if (create_if_missing)
    {
        flags |= O_CREAT;
    }

    int fd = open(path, flags, 0666);
    if (fd < 0)
    {
        return NULL;
    }

    aw_store_t *st = (aw_store_t *)calloc(1, sizeof(*st));
    if (st == NULL)
    {
        close(fd);
        return NULL;
    }

    st->fd = fd;
    st->path = strdup(path);
    st->files = NULL;
    st->file_count = 0;
    st->file_cap = 0;

    if (aw_load_latest_snapshot(st) != 0)
    {
        aw_free_entries(st->files, st->file_count);
        free(st->files);
        free(st->path);
        close(st->fd);
        free(st);
        return NULL;
    }

    return st;
}

void aw_store_close(aw_store_t *st)
{
    if (st == NULL)
    {
        return;
    }
    aw_free_entries(st->files, st->file_count);
    free(st->files);
    st->files = NULL;
    st->file_count = 0;
    st->file_cap = 0;

    if (st->fd >= 0)
    {
        close(st->fd);
    }
    free(st->path);
    free(st);
}

int aw_store_get(aw_store_t *st,
                 const char *logical_path,
                 void **data_out,
                 size_t *len_out)
{
    if (st == NULL || logical_path == NULL ||
        data_out == NULL || len_out == NULL)
    {
        errno = EINVAL;
        return -1;
    }

    int idx = aw_find_entry(st->files, st->file_count, logical_path);
    if (idx < 0)
    {
        errno = ENOENT;
        return -1;
    }

    struct aw_file_entry *e = &st->files[idx];
    void *buf = NULL;
    if (e->len > 0)
    {
        buf = malloc(e->len);
        if (buf == NULL)
        {
            return -1;
        }
        memcpy(buf, e->data, e->len);
    }

    *data_out = buf;
    *len_out = e->len;
    return 0;
}

aw_tx_t *aw_tx_begin(aw_store_t *st)
{
    if (st == NULL)
    {
        errno = EINVAL;
        return NULL;
    }

    aw_tx_t *tx = (aw_tx_t *)calloc(1, sizeof(*tx));
    if (tx == NULL)
    {
        return NULL;
    }

    tx->store = st;
    tx->files = NULL;
    tx->file_count = 0;
    tx->file_cap = 0;
    tx->committed = 0;

    if (st->file_count > 0)
    {
        if (aw_grow_entries(&tx->files, &tx->file_cap, st->file_count) != 0)
        {
            free(tx);
            return NULL;
        }

        size_t i = 0;
        while (i < st->file_count)
        {
            const struct aw_file_entry *src = &st->files[i];
            struct aw_file_entry *dst = &tx->files[i];

            dst->path = strdup(src->path);
            if (dst->path == NULL)
            {
                aw_free_entries(tx->files, i);
                free(tx->files);
                free(tx);
                return NULL;
            }

            dst->len = src->len;
            if (src->len > 0)
            {
                dst->data = malloc(src->len);
                if (dst->data == NULL)
                {
                    free(dst->path);
                    dst->path = NULL;
                    aw_free_entries(tx->files, i);
                    free(tx->files);
                    free(tx);
                    return NULL;
                }
                memcpy(dst->data, src->data, src->len);
            }
            else
            {
                dst->data = NULL;
            }
            i++;
        }
        tx->file_count = st->file_count;
    }

    return tx;
}

int aw_tx_add_buffer(aw_tx_t *tx,
                     const char *logical_path,
                     const void *data,
                     size_t len)
{
    if (tx == NULL || logical_path == NULL)
    {
        errno = EINVAL;
        return -1;
    }

    int idx = aw_find_entry(tx->files, tx->file_count, logical_path);

    if (idx >= 0)
    {
        struct aw_file_entry *e = &tx->files[idx];
        void *new_data = NULL;
        if (len > 0)
        {
            new_data = malloc(len);
            if (new_data == NULL)
            {
                return -1;
            }
            if (data != NULL)
            {
                memcpy(new_data, data, len);
            }
        }
        free(e->data);
        e->data = new_data;
        e->len = len;
        return 0;
    }

    if (aw_grow_entries(&tx->files, &tx->file_cap, tx->file_count + 1) != 0)
    {
        return -1;
    }

    struct aw_file_entry *e = &tx->files[tx->file_count];
    e->path = strdup(logical_path);
    if (e->path == NULL)
    {
        return -1;
    }

    e->len = len;
    e->data = NULL;
    if (len > 0)
    {
        e->data = malloc(len);
        if (e->data == NULL)
        {
            free(e->path);
            e->path = NULL;
            return -1;
        }
        if (data != NULL)
        {
            memcpy(e->data, data, len);
        }
    }

    tx->file_count++;
    return 0;
}

int aw_tx_remove(aw_tx_t *tx, const char *logical_path)
{
    if (tx == NULL || logical_path == NULL)
    {
        errno = EINVAL;
        return -1;
    }

    int idx = aw_find_entry(tx->files, tx->file_count, logical_path);
    if (idx < 0)
    {
        return 0;
    }

    aw_free_entry(&tx->files[idx]);

    if ((size_t)idx != tx->file_count - 1)
    {
        tx->files[idx] = tx->files[tx->file_count - 1];
    }
    tx->file_count--;

    return 0;
}

int aw_tx_commit(aw_tx_t *tx)
{
    if (tx == NULL || tx->store == NULL)
    {
        errno = EINVAL;
        return -1;
    }
    if (tx->committed)
    {
        return 0;
    }

    aw_store_t *st = tx->store;

    uint8_t *blob = NULL;
    size_t blob_len = 0;
    if (aw_serialize_snapshot(tx->files, tx->file_count, &blob, &blob_len) != 0)
    {
        return -1;
    }

    uint64_t checksum = aw_fnv1a64(blob, blob_len);

    int ok = 0;
    off_t end_off = lseek(st->fd, 0, SEEK_END);
    if (end_off < 0)
    {
        ok = -1;
    }

    if (ok == 0)
    {
        ssize_t n = write(st->fd, blob, blob_len);
        if (n != (ssize_t)blob_len)
        {
            ok = -1;
        }
    }

    if (ok == 0)
    {
        if (fsync(st->fd) != 0)
        {
            ok = -1;
        }
    }

    if (ok == 0)
    {
        if (aw_write_footer_append(st->fd,
                                   (uint64_t)blob_len,
                                   checksum) != 0)
        {
            ok = -1;
        }
    }

    if (ok == 0)
    {
        if (fsync(st->fd) != 0)
        {
            ok = -1;
        }
    }

    free(blob);
    blob = NULL;

    if (ok != 0)
    {
        return -1;
    }

    aw_free_entries(st->files, st->file_count);
    free(st->files);
    st->files = NULL;
    st->file_count = 0;
    st->file_cap = 0;

    st->files = (struct aw_file_entry *)calloc(tx->file_cap, sizeof(*st->files));
    if (st->files == NULL)
    {
        errno = ENOMEM;
        return -1;
    }
    st->file_cap = tx->file_cap;
    st->file_count = tx->file_count;

    size_t i = 0;
    while (i < tx->file_count)
    {
        const struct aw_file_entry *src = &tx->files[i];
        struct aw_file_entry *dst = &st->files[i];

        dst->path = strdup(src->path);
        if (dst->path == NULL)
        {
            aw_free_entries(st->files, i);
            free(st->files);
            st->files = NULL;
            st->file_count = 0;
            st->file_cap = 0;
            errno = ENOMEM;
            return -1;
        }

        dst->len = src->len;
        if (src->len > 0)
        {
            dst->data = malloc(src->len);
            if (dst->data == NULL)
            {
                free(dst->path);
                aw_free_entries(st->files, i);
                free(st->files);
                st->files = NULL;
                st->file_count = 0;
                st->file_cap = 0;
                errno = ENOMEM;
                return -1;
            }
            memcpy(dst->data, src->data, src->len);
        }
        else
        {
            dst->data = NULL;
        }
        i++;
    }

    tx->committed = 1;
    return 0;
}

void aw_tx_abort(aw_tx_t *tx)
{
    if (tx == NULL)
    {
        return;
    }
    if (tx->committed)
    {
        return;
    }
}

void aw_tx_free(aw_tx_t *tx)
{
    if (tx == NULL)
    {
        return;
    }
    aw_tx_abort(tx);
    aw_free_entries(tx->files, tx->file_count);
    free(tx->files);
    free(tx);
}