yams/tests/unit/wal/wal_manager_catch2_test.cpp

#include <catch2/catch_test_macros.hpp>
#include <catch2/generators/catch_generators.hpp>

#include <chrono>
#include <filesystem>
#include <fstream>
#include <string>

#include <yams/core/types.h>
#include <yams/wal/wal_manager.h>

using namespace yams;
using namespace yams::wal;

namespace {

struct TempDir {
    TempDir() {
        auto stamp = std::to_string(std::chrono::steady_clock::now().time_since_epoch().count());
        path = std::filesystem::temp_directory_path() / ("yams_wal_manager_test_" + stamp);
        std::filesystem::create_directories(path);
    }

    ~TempDir() {
        std::error_code ec;
        std::filesystem::remove_all(path, ec);
    }

    std::filesystem::path path;
};

std::string makeHash(char fill) {
    return std::string(HASH_SIZE, fill);
}

} // namespace

TEST_CASE("WALManager initialize picks max sequence from existing log names",
          "[unit][wal][manager]") {
    TempDir temp;
    auto walDir = temp.path / "wal";
    std::filesystem::create_directories(walDir);

    {
        std::ofstream plain((walDir / "wal_20240101_010101_42.log").string(), std::ios::binary);
        REQUIRE(plain.good());
    }
    {
        std::ofstream compressed((walDir / "wal_20240101_010102_123.log.zst").string(),
                                 std::ios::binary);
        REQUIRE(compressed.good());
    }
    {
        std::ofstream ignored((walDir / "wal_invalid.log").string(), std::ios::binary);
        REQUIRE(ignored.good());
    }

    WALManager::Config cfg;
    cfg.walDirectory = walDir;
    cfg.compressOldLogs = false;
    cfg.enableGroupCommit = false;
    cfg.syncInterval = 1;

    WALManager manager(cfg);
    REQUIRE(manager.initialize());

    CHECK(manager.getCurrentSequence() == 123);

    auto payload = WALEntry::DeleteBlockData::encode(makeHash('s'));
    WALEntry entry(WALEntry::OpType::DeleteBlock, 0, 0, payload);
    auto writeResult = manager.writeEntry(entry);
    REQUIRE(writeResult);
    CHECK(writeResult.value() == 124);
    CHECK(manager.getCurrentSequence() == 124);

    REQUIRE(manager.shutdown());
}

TEST_CASE("WALManager pruneLogs removes old inactive logs and keeps active log",
          "[unit][wal][manager]") {
    TempDir temp;
    auto walDir = temp.path / "wal";

    WALManager::Config cfg;
    cfg.walDirectory = walDir;
    cfg.compressOldLogs = false;
    cfg.enableGroupCommit = false;
    cfg.syncInterval = 1;

    WALManager manager(cfg);
    REQUIRE(manager.initialize());

    auto activeLogPath = manager.getCurrentLogPath();
    REQUIRE(!activeLogPath.empty());
    REQUIRE(std::filesystem::exists(activeLogPath));

    auto oldInactiveLog = walDir / "wal_19990101_000000_1.log";
    {
        std::ofstream oldLog(oldInactiveLog.string(), std::ios::binary);
        REQUIRE(oldLog.good());
    }
    auto oldTimestamp = std::filesystem::file_time_type::clock::now() - std::chrono::hours(24);
    std::filesystem::last_write_time(oldInactiveLog, oldTimestamp);

    REQUIRE(manager.pruneLogs(std::chrono::hours(0)));

    CHECK(std::filesystem::exists(activeLogPath));
    CHECK_FALSE(std::filesystem::exists(oldInactiveLog));

    REQUIRE(manager.shutdown());
}

TEST_CASE("WALManager transaction lifecycle updates stats and enforces state",
          "[unit][wal][manager][transaction]") {
    TempDir temp;
    auto walDir = temp.path / "wal";

    WALManager::Config cfg;
    cfg.walDirectory = walDir;
    cfg.compressOldLogs = false;
    cfg.enableGroupCommit = false;
    cfg.syncInterval = 1;

    WALManager manager(cfg);
    REQUIRE(manager.initialize());

    auto txn = manager.beginTransaction();
    REQUIRE(txn);
    CHECK(txn->getState() == Transaction::State::Active);
    CHECK(manager.getStats().activeTransactions == 1);

    REQUIRE(txn->storeBlock(makeHash('a'), 123, 1));
    REQUIRE(txn->commit());
    CHECK(txn->getState() == Transaction::State::Committed);
    CHECK(manager.getStats().activeTransactions == 0);

    auto secondCommit = txn->commit();
    REQUIRE_FALSE(secondCommit);
    CHECK(secondCommit.error().code == ErrorCode::InvalidOperation);

    auto addAfterCommit = txn->deleteBlock(makeHash('b'));
    REQUIRE_FALSE(addAfterCommit);
    CHECK(addAfterCommit.error().code == ErrorCode::InvalidOperation);

    {
        auto autoRollbackTxn = manager.beginTransaction();
        REQUIRE(autoRollbackTxn);
        REQUIRE(autoRollbackTxn->updateReference(makeHash('c'), 2));
        CHECK(manager.getStats().activeTransactions == 1);
    }

    CHECK(manager.getStats().activeTransactions == 0);
    REQUIRE(manager.shutdown());
}

TEST_CASE("WALManager initialize preserves nonempty segments with a zero prefix",
          "[unit][wal][manager]") {
    TempDir temp;
    const auto walDir = temp.path / "wal";
    std::filesystem::create_directories(walDir);
    const auto damagedLog = walDir / "wal_20240101_010159_150.log";
    const auto size = GENERATE(std::size_t{3}, std::size_t{5}, std::size_t{64 * 1024},
                               std::size_t{128 * 1024 + 1});
    std::string original(size, '\0');
    original.back() = 'x'; // Include short reads, buffer boundaries, and a trailing partial block.
    {
        std::ofstream out(damagedLog, std::ios::binary);
        out.write(original.data(), static_cast<std::streamsize>(original.size()));
        REQUIRE(out.good());
    }

    WALManager::Config cfg;
    cfg.walDirectory = walDir;
    cfg.compressOldLogs = false;
    cfg.enableGroupCommit = false;
    WALManager manager(cfg);
    REQUIRE(manager.initialize());
    CHECK(manager.getCurrentSequence() == 150);
    REQUIRE(manager.shutdown());

    REQUIRE(std::filesystem::exists(damagedLog));
    REQUIRE(std::filesystem::file_size(damagedLog) == original.size());
    std::ifstream in(damagedLog, std::ios::binary);
    std::string preserved(original.size(), '\0');
    in.read(preserved.data(), static_cast<std::streamsize>(preserved.size()));
    REQUIRE(in.good());
    CHECK(preserved == original);
}

TEST_CASE("WALManager initialize rejects an unconfigured directory", "[unit][wal][manager]") {
    WALManager::Config cfg;
    cfg.enableGroupCommit = false;
    REQUIRE(cfg.walDirectory.empty());

    WALManager manager(cfg);
    auto result = manager.initialize();
    REQUIRE_FALSE(result);
    CHECK(result.error().code == ErrorCode::InvalidArgument);
}

TEST_CASE("WALManager initialize prunes empty log segments but keeps their sequence",
          "[unit][wal][manager]") {
    TempDir temp;
    auto walDir = temp.path / "wal";
    std::filesystem::create_directories(walDir);
    for (int i = 0; i < 50; ++i) {
        std::ofstream empty((walDir / ("wal_20240101_0101" + std::to_string(10 + i) + "_" +
                                       std::to_string(100 + i) + ".log"))
                                .string(),
                            std::ios::binary);
        REQUIRE(empty.good());
    }
    {
        // A killed writer leaves the mmap preallocation behind: 1 MiB of zeros, no entry.
        std::ofstream preallocated((walDir / "wal_20240101_010159_150.log").string(),
                                   std::ios::binary);
        std::string zeros(1024 * 1024, '\0');
        preallocated.write(zeros.data(), static_cast<std::streamsize>(zeros.size()));
        REQUIRE(preallocated.good());
    }
    {
        std::ofstream nonEmpty((walDir / "wal_20240101_010200_7.log").string(), std::ios::binary);
        nonEmpty << "not-a-real-entry";
        REQUIRE(nonEmpty.good());
    }

    WALManager::Config cfg;
    cfg.walDirectory = walDir;
    cfg.compressOldLogs = false;
    cfg.enableGroupCommit = false;
    WALManager manager(cfg);
    REQUIRE(manager.initialize());

    CHECK(manager.getCurrentSequence() == 150);

    std::size_t logFiles = 0;
    bool nonEmptySurvived = false;
    for (const auto& entry : std::filesystem::directory_iterator(walDir)) {
        if (entry.path().extension() == ".log") {
            ++logFiles;
            if (entry.path().filename() == "wal_20240101_010200_7.log") {
                nonEmptySurvived = true;
            }
        }
    }
    CHECK(nonEmptySurvived);
    // The non-empty segment plus the freshly opened active log.
    CHECK(logFiles == 2);
    REQUIRE(manager.shutdown());
}