Claude
1e312fcd0b
After a lost record the journal drops position-based records until a full checkpoint. That checkpoint was only requested from later edits, rate-limited to once a second; when the writer handled the failing record before its edit asked for a checkpoint, the checkpoint failed as well and consumed the slot, so edits made within the next second were not recovered and, if the user stopped typing, the journal stayed behind indefinitely. SwapManager::RetryGapCheckpoints(), called once per frame from Editor::ProcessPendingOpens, requests it once the interval has passed. This also made SwapReplay_LostRecord_ResyncsWithCheckpoint flaky under load; the test now exercises the retry. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_015rcVQiiWyTzoNrsvLSaAfB
1919 lines
53 KiB
C++
1919 lines
53 KiB
C++
#include "Swap.h"
|
|
#include "Buffer.h"
|
|
#include "ErrorHandler.h"
|
|
#include "SyscallWrappers.h"
|
|
#include "ErrorRecovery.h"
|
|
|
|
#include <algorithm>
|
|
#include <array>
|
|
#include <chrono>
|
|
#include <cstdio>
|
|
#include <cstring>
|
|
#include <ctime>
|
|
#include <cstdlib>
|
|
#include <fstream>
|
|
#include <filesystem>
|
|
#include <fcntl.h>
|
|
#include <unistd.h>
|
|
#include <sys/file.h>
|
|
#include <sys/stat.h>
|
|
#include <cerrno>
|
|
|
|
namespace fs = std::filesystem;
|
|
|
|
namespace kte {
|
|
namespace {
|
|
// Records store their payload length in 24 bits.
|
|
constexpr std::size_t kMaxRecordPayload = 0xFFFFFFu;
|
|
// While a journal has a gap, request a recovery checkpoint at most this often.
|
|
constexpr std::uint64_t kGapCheckpointIntervalNs = 1000000000ULL;
|
|
} // namespace
|
|
|
|
static void remove_journal_unless_held(const std::string &path, bool locked_out);
|
|
|
|
namespace {
|
|
constexpr std::uint8_t MAGIC[8] = {'K', 'T', 'E', '_', 'S', 'W', 'P', '\0'};
|
|
constexpr std::uint32_t VERSION = 1;
|
|
|
|
|
|
static std::string
|
|
snapshot_buffer_bytes(const Buffer &b)
|
|
{
|
|
const std::size_t nrows = b.Nrows();
|
|
std::string out;
|
|
// Cheap lower bound: sum of row sizes.
|
|
std::size_t approx = 0;
|
|
for (std::size_t i = 0; i < nrows; i++)
|
|
approx += b.GetLineView(i).size();
|
|
out.reserve(approx);
|
|
for (std::size_t i = 0; i < nrows; i++) {
|
|
auto v = b.GetLineView(i);
|
|
out.append(v.data(), v.size());
|
|
}
|
|
return out;
|
|
}
|
|
|
|
|
|
static fs::path
|
|
xdg_state_home()
|
|
{
|
|
if (const char *p = std::getenv("XDG_STATE_HOME")) {
|
|
if (*p)
|
|
return fs::path(p);
|
|
}
|
|
if (const char *home = std::getenv("HOME")) {
|
|
if (*home)
|
|
return fs::path(home) / ".local" / "state";
|
|
}
|
|
// Last resort: still provide a stable per-user-ish location.
|
|
return fs::temp_directory_path() / "kte" / "state";
|
|
}
|
|
|
|
|
|
static fs::path
|
|
swap_root_dir()
|
|
{
|
|
return xdg_state_home() / "kte" / "swap";
|
|
}
|
|
|
|
|
|
static std::uint64_t
|
|
fnv1a64(std::string_view s)
|
|
{
|
|
std::uint64_t h = 14695981039346656037ULL;
|
|
for (unsigned char ch: s) {
|
|
h ^= (std::uint64_t) ch;
|
|
h *= 1099511628211ULL;
|
|
}
|
|
return h;
|
|
}
|
|
|
|
|
|
static std::string
|
|
hex_u64(std::uint64_t v)
|
|
{
|
|
static const char *kHex = "0123456789abcdef";
|
|
char out[16];
|
|
for (int i = 15; i >= 0; --i) {
|
|
out[i] = kHex[v & 0xFULL];
|
|
v >>= 4;
|
|
}
|
|
return std::string(out, sizeof(out));
|
|
}
|
|
|
|
|
|
// Write all bytes in buf to fd, handling EINTR and partial writes.
|
|
static bool
|
|
write_full(int fd, const void *buf, size_t len)
|
|
{
|
|
const std::uint8_t *p = static_cast<const std::uint8_t *>(buf);
|
|
while (len > 0) {
|
|
ssize_t n = ::write(fd, p, len);
|
|
if (n < 0) {
|
|
if (errno == EINTR)
|
|
continue;
|
|
return false;
|
|
}
|
|
if (n == 0)
|
|
return false; // shouldn't happen for regular files; treat as error
|
|
p += static_cast<size_t>(n);
|
|
len -= static_cast<size_t>(n);
|
|
}
|
|
return true;
|
|
}
|
|
|
|
|
|
static std::string
|
|
encode_path_key(std::string s)
|
|
{
|
|
// Turn an absolute path like "/home/kyle/tmp/test.txt" into
|
|
// "home!kyle!tmp!test.txt" so swap files are human-identifiable.
|
|
//
|
|
// Notes:
|
|
// - We strip a single leading path separator so absolute paths don't start with '!'.
|
|
// - We replace both '/' and '\\' with '!'.
|
|
// - We leave other characters as-is (spaces are OK on POSIX).
|
|
// - '!' and '%' in the path itself are escaped as %21 and %25, so the
|
|
// mapping is one-to-one: "/x/a!b" and "/x/a/b" used to share a journal.
|
|
if (!s.empty() && (s[0] == '/' || s[0] == '\\'))
|
|
s.erase(0, 1);
|
|
std::string out;
|
|
out.reserve(s.size());
|
|
for (char ch: s) {
|
|
if (ch == '/' || ch == '\\')
|
|
out.push_back('!');
|
|
else if (ch == '!')
|
|
out += "%21";
|
|
else if (ch == '%')
|
|
out += "%25";
|
|
else
|
|
out.push_back(ch);
|
|
}
|
|
return out;
|
|
}
|
|
|
|
|
|
// stat() a file for journal base identity. False if it is not a regular file.
|
|
static bool
|
|
stat_base(const std::string &file, std::uint64_t &size, std::int64_t &mtime_ns)
|
|
{
|
|
struct stat st{};
|
|
if (file.empty() || ::stat(file.c_str(), &st) != 0 || !S_ISREG(st.st_mode))
|
|
return false;
|
|
size = static_cast<std::uint64_t>(st.st_size);
|
|
#if defined(__APPLE__)
|
|
mtime_ns = static_cast<std::int64_t>(st.st_mtimespec.tv_sec) * 1000000000LL + st.st_mtimespec.tv_nsec;
|
|
#else
|
|
mtime_ns = static_cast<std::int64_t>(st.st_mtim.tv_sec) * 1000000000LL + st.st_mtim.tv_nsec;
|
|
#endif
|
|
return true;
|
|
}
|
|
|
|
|
|
static std::string
|
|
compute_swap_path_for_filename(const std::string &filename)
|
|
{
|
|
if (filename.empty())
|
|
return std::string();
|
|
// Always place swap under an XDG home-appropriate state directory.
|
|
// This avoids cluttering working directories and prevents stomping on
|
|
// swap files when multiple different paths share the same basename.
|
|
fs::path root = swap_root_dir();
|
|
|
|
fs::path p(filename);
|
|
std::string key;
|
|
try {
|
|
key = fs::weakly_canonical(p).string();
|
|
} catch (...) {
|
|
try {
|
|
key = fs::absolute(p).string();
|
|
} catch (...) {
|
|
key = filename;
|
|
}
|
|
}
|
|
std::string encoded = encode_path_key(key);
|
|
if (!encoded.empty()) {
|
|
std::string name = encoded + ".swp";
|
|
// Avoid filesystem/path length issues; fall back to hashed naming.
|
|
// NAME_MAX is often 255 on POSIX, but keep extra headroom.
|
|
if (name.size() <= 200) {
|
|
return (root / name).string();
|
|
}
|
|
}
|
|
|
|
// Fallback: stable, shorter name based on basename + hash.
|
|
std::string base = p.filename().string();
|
|
const std::string name = base + "." + hex_u64(fnv1a64(key)) + ".swp";
|
|
return (root / name).string();
|
|
}
|
|
}
|
|
|
|
|
|
SwapManager::SwapManager()
|
|
{
|
|
running_.store(true);
|
|
worker_ = std::thread([this] {
|
|
this->writer_loop();
|
|
});
|
|
// Best-effort prune of old swap files.
|
|
// Safe early in startup: journals_ is still empty and no fds are open yet.
|
|
if (cfg_.prune_on_startup) {
|
|
PruneSwapDir();
|
|
}
|
|
}
|
|
|
|
|
|
SwapManager::~SwapManager()
|
|
{
|
|
// Best-effort: drain queued records before stopping the writer.
|
|
Flush();
|
|
running_.store(false);
|
|
cv_.notify_all();
|
|
if (worker_.joinable())
|
|
worker_.join();
|
|
// Close all journals
|
|
for (auto &kv: journals_) {
|
|
close_ctx(kv.second);
|
|
}
|
|
}
|
|
|
|
|
|
void
|
|
SwapManager::Checkpoint(Buffer *buf)
|
|
{
|
|
if (buf) {
|
|
RecordCheckpoint(*buf, false);
|
|
return;
|
|
}
|
|
// All buffers
|
|
std::vector<Buffer *> bufs;
|
|
{
|
|
std::lock_guard<std::mutex> lg(mtx_);
|
|
bufs.reserve(journals_.size());
|
|
for (auto &kv: journals_) {
|
|
bufs.push_back(kv.first);
|
|
}
|
|
}
|
|
for (Buffer *b: bufs) {
|
|
if (b)
|
|
RecordCheckpoint(*b, false);
|
|
}
|
|
}
|
|
|
|
|
|
void
|
|
SwapManager::Flush(Buffer *buf)
|
|
{
|
|
(void) buf; // stage 1: flushes all buffers
|
|
std::unique_lock<std::mutex> lk(mtx_);
|
|
const std::uint64_t target = next_seq_;
|
|
// Wake the writer in case it's waiting on the interval.
|
|
cv_.notify_one();
|
|
cv_.wait(lk, [&] {
|
|
return queue_.empty() && inflight_ == 0 && last_processed_ >= target;
|
|
});
|
|
}
|
|
|
|
|
|
void
|
|
SwapManager::BufferRecorder::OnInsert(int row, int col, std::string_view bytes)
|
|
{
|
|
m_.RecordInsert(buf_, row, col, bytes);
|
|
}
|
|
|
|
|
|
void
|
|
SwapManager::BufferRecorder::OnDelete(int row, int col, std::size_t len)
|
|
{
|
|
m_.RecordDelete(buf_, row, col, len);
|
|
}
|
|
|
|
|
|
SwapRecorder *
|
|
SwapManager::RecorderFor(Buffer *buf)
|
|
{
|
|
if (!buf)
|
|
return nullptr;
|
|
std::lock_guard<std::mutex> lg(mtx_);
|
|
auto it = recorders_.find(buf);
|
|
if (it != recorders_.end())
|
|
return it->second.get();
|
|
// Create on-demand. Recording calls will no-op until Attach() has been called.
|
|
auto rec = std::make_unique<BufferRecorder>(*this, *buf);
|
|
SwapRecorder *ptr = rec.get();
|
|
recorders_[buf] = std::move(rec);
|
|
return ptr;
|
|
}
|
|
|
|
|
|
SwapRecorder *
|
|
SwapManager::Rehome(Buffer *old_addr, Buffer *new_addr)
|
|
{
|
|
if (!old_addr || !new_addr || old_addr == new_addr)
|
|
return nullptr;
|
|
std::lock_guard<std::mutex> lg(mtx_);
|
|
SwapRecorder *result = nullptr;
|
|
|
|
auto jit = journals_.find(old_addr);
|
|
if (jit != journals_.end()) {
|
|
JournalCtx ctx = std::move(jit->second);
|
|
journals_.erase(jit);
|
|
journals_[new_addr] = std::move(ctx);
|
|
}
|
|
|
|
auto rit = recorders_.find(old_addr);
|
|
if (rit != recorders_.end()) {
|
|
recorders_.erase(rit);
|
|
// BufferRecorder binds a Buffer& at construction, so it can't be
|
|
// repointed in place; rebuild it against the buffer's new address.
|
|
auto rec = std::make_unique<BufferRecorder>(*this, *new_addr);
|
|
result = rec.get();
|
|
recorders_[new_addr] = std::move(rec);
|
|
}
|
|
|
|
// Defensive: any record still queued (not yet drained by the writer thread)
|
|
// for the old address must follow the buffer to its new location. Callers
|
|
// are expected to Flush() before rehoming so this should normally be a no-op.
|
|
for (auto &p: queue_) {
|
|
if (p.buf == old_addr)
|
|
p.buf = new_addr;
|
|
}
|
|
|
|
return result;
|
|
}
|
|
|
|
|
|
void
|
|
SwapManager::Attach(Buffer *buf)
|
|
{
|
|
if (!buf || buf->IsVirtual())
|
|
return; // e.g. +HELP+: nothing to recover, and its name is not a path
|
|
bool fresh = false;
|
|
{
|
|
std::lock_guard<std::mutex> lg(mtx_);
|
|
fresh = journals_.find(buf) == journals_.end();
|
|
}
|
|
const BaseId base = fresh ? compute_base(buf->Filename()) : BaseId{};
|
|
std::lock_guard<std::mutex> lg(mtx_);
|
|
const bool still_fresh = journals_.find(buf) == journals_.end();
|
|
JournalCtx &ctx = journals_[buf];
|
|
if (ctx.path.empty())
|
|
ctx.path = ComputeSidecarPath(*buf);
|
|
if (fresh && still_fresh)
|
|
apply_base(ctx, base);
|
|
// Ensure a recorder exists as well.
|
|
if (recorders_.find(buf) == recorders_.end()) {
|
|
recorders_[buf] = std::make_unique<BufferRecorder>(*this, *buf);
|
|
}
|
|
}
|
|
|
|
|
|
void
|
|
SwapManager::Detach(Buffer *buf, const bool remove_file)
|
|
{
|
|
if (!buf) {
|
|
return;
|
|
}
|
|
|
|
// Write a best-effort final checkpoint before suspending and closing.
|
|
// If the caller requested removal, skip the final checkpoint so the file can be deleted.
|
|
if (!remove_file)
|
|
RecordCheckpoint(*buf, true);
|
|
{
|
|
std::lock_guard<std::mutex> lg(mtx_);
|
|
auto it = journals_.find(buf);
|
|
if (it != journals_.end()) {
|
|
it->second.suspended = true;
|
|
}
|
|
}
|
|
|
|
Flush(buf);
|
|
std::string path;
|
|
bool locked_out = false;
|
|
{
|
|
std::lock_guard<std::mutex> io(io_mtx_);
|
|
std::lock_guard<std::mutex> lg(mtx_);
|
|
auto it = journals_.find(buf);
|
|
if (it != journals_.end()) {
|
|
path = it->second.path;
|
|
locked_out = it->second.locked_out;
|
|
close_ctx(it->second);
|
|
journals_.erase(it);
|
|
}
|
|
recorders_.erase(buf);
|
|
}
|
|
|
|
if (remove_file)
|
|
remove_journal_unless_held(path, locked_out);
|
|
}
|
|
|
|
|
|
void
|
|
SwapManager::ResetJournal(Buffer &buf)
|
|
{
|
|
bool was_locked_out = false;
|
|
std::string path;
|
|
{
|
|
std::lock_guard<std::mutex> lg(mtx_);
|
|
auto it = journals_.find(&buf);
|
|
if (it == journals_.end())
|
|
return;
|
|
JournalCtx &ctx = it->second;
|
|
if (ctx.path.empty())
|
|
ctx.path = ComputeSidecarPath(buf);
|
|
path = ctx.path;
|
|
ctx.suspended = true;
|
|
}
|
|
|
|
Flush(&buf);
|
|
const BaseId base = compute_base(buf.Filename());
|
|
|
|
{
|
|
std::lock_guard<std::mutex> io(io_mtx_);
|
|
std::lock_guard<std::mutex> lg(mtx_);
|
|
auto it = journals_.find(&buf);
|
|
if (it == journals_.end())
|
|
return;
|
|
JournalCtx &ctx = it->second;
|
|
close_ctx(ctx);
|
|
was_locked_out = ctx.locked_out;
|
|
ctx.header_ok = false;
|
|
ctx.gap = false; // the journal restarts from the saved file
|
|
ctx.gap_chkpt_request_ns = 0;
|
|
ctx.gap_unfixable_reported = false;
|
|
ctx.locked_out = false;
|
|
apply_base(ctx, base);
|
|
ctx.last_flush_ns = 0;
|
|
ctx.last_fsync_ns = 0;
|
|
ctx.last_chkpt_ns = 0;
|
|
ctx.edit_bytes_since_chkpt = 0;
|
|
ctx.approx_size_bytes = 0;
|
|
ctx.suspended = false;
|
|
}
|
|
|
|
remove_journal_unless_held(path, was_locked_out);
|
|
}
|
|
|
|
|
|
std::string
|
|
SwapManager::SwapDirRoot()
|
|
{
|
|
return swap_root_dir().string();
|
|
}
|
|
|
|
|
|
void
|
|
SwapManager::PruneSwapDir()
|
|
{
|
|
SwapConfig cfg;
|
|
std::vector<std::string> active;
|
|
{
|
|
std::lock_guard<std::mutex> lg(mtx_);
|
|
cfg = cfg_;
|
|
active.reserve(journals_.size());
|
|
for (const auto &kv: journals_) {
|
|
if (!kv.second.path.empty())
|
|
active.push_back(kv.second.path);
|
|
}
|
|
}
|
|
|
|
const fs::path root = swap_root_dir();
|
|
std::error_code ec;
|
|
if (!fs::exists(root, ec) || ec)
|
|
return;
|
|
|
|
struct Entry {
|
|
fs::path path;
|
|
std::filesystem::file_time_type mtime;
|
|
};
|
|
std::vector<Entry> swps;
|
|
for (auto it = fs::directory_iterator(root, ec); !ec && it != fs::directory_iterator(); it.increment(ec)) {
|
|
const fs::path p = it->path();
|
|
if (p.extension() != ".swp")
|
|
continue;
|
|
// Never delete active journals.
|
|
const std::string ps = p.string();
|
|
bool is_active = false;
|
|
for (const auto &a: active) {
|
|
if (a == ps) {
|
|
is_active = true;
|
|
break;
|
|
}
|
|
}
|
|
if (is_active)
|
|
continue;
|
|
std::error_code ec2;
|
|
if (!it->is_regular_file(ec2) || ec2)
|
|
continue;
|
|
auto tm = fs::last_write_time(p, ec2);
|
|
if (ec2)
|
|
continue;
|
|
swps.push_back({p, tm});
|
|
}
|
|
|
|
if (swps.empty())
|
|
return;
|
|
|
|
// Sort newest first.
|
|
std::sort(swps.begin(), swps.end(), [](const Entry &a, const Entry &b) {
|
|
return a.mtime > b.mtime;
|
|
});
|
|
|
|
// Convert age threshold.
|
|
auto now = std::filesystem::file_time_type::clock::now();
|
|
auto max_age = std::chrono::hours(24) * static_cast<long long>(cfg.prune_max_age_days);
|
|
|
|
std::size_t kept = 0;
|
|
for (const auto &e: swps) {
|
|
bool too_old = false;
|
|
if (cfg.prune_max_age_days > 0) {
|
|
// If file_time_type isn't system_clock, duration arithmetic still works.
|
|
if (now - e.mtime > max_age)
|
|
too_old = true;
|
|
}
|
|
bool over_limit = (cfg.prune_max_files > 0) && (kept >= cfg.prune_max_files);
|
|
// Never prune a journal another running session holds.
|
|
if ((too_old || over_limit) && !JournalInUse(e.path.string())) {
|
|
std::error_code ec3;
|
|
fs::remove(e.path, ec3);
|
|
} else {
|
|
++kept;
|
|
}
|
|
}
|
|
}
|
|
|
|
|
|
void
|
|
SwapManager::NotifyFilenameChanged(Buffer &buf)
|
|
{
|
|
// No checkpoint here: the old journal is deleted below, and every caller
|
|
// has just loaded or saved the file, so the new journal's base is the
|
|
// file's content.
|
|
std::string old_path;
|
|
bool attached = false;
|
|
{
|
|
std::lock_guard<std::mutex> lg(mtx_);
|
|
auto it = journals_.find(&buf);
|
|
if (it != journals_.end()) {
|
|
attached = true;
|
|
old_path = it->second.path;
|
|
it->second.suspended = true;
|
|
}
|
|
}
|
|
if (!attached) {
|
|
// A buffer that was never journaled (a virtual buffer such as +HELP+,
|
|
// now saved under a real name) starts journaling here.
|
|
if (!buf.IsVirtual() && !buf.Filename().empty()) {
|
|
Attach(&buf);
|
|
buf.SetSwapRecorder(RecorderFor(&buf));
|
|
}
|
|
return;
|
|
}
|
|
Flush(&buf);
|
|
const BaseId base = compute_base(buf.Filename());
|
|
std::lock_guard<std::mutex> io(io_mtx_);
|
|
std::lock_guard<std::mutex> lg(mtx_);
|
|
auto it = journals_.find(&buf);
|
|
if (it == journals_.end())
|
|
return;
|
|
JournalCtx &ctx = it->second;
|
|
close_ctx(ctx);
|
|
const std::string new_path = ComputeSidecarPath(buf);
|
|
// When the path is unchanged (Attach already pointed the journal at this
|
|
// file's swap), that file may be the one the caller is about to replay
|
|
// for recovery: removing it here lost the recovery.
|
|
if (!old_path.empty() && old_path != new_path)
|
|
remove_journal_unless_held(old_path, ctx.locked_out);
|
|
ctx.gap = false;
|
|
ctx.gap_chkpt_request_ns = 0;
|
|
ctx.gap_unfixable_reported = false;
|
|
ctx.locked_out = false;
|
|
apply_base(ctx, base);
|
|
ctx.path = new_path;
|
|
ctx.suspended = false;
|
|
ctx.header_ok = false;
|
|
ctx.last_flush_ns = 0;
|
|
ctx.last_fsync_ns = 0;
|
|
ctx.last_chkpt_ns = 0;
|
|
ctx.edit_bytes_since_chkpt = 0;
|
|
ctx.approx_size_bytes = 0;
|
|
}
|
|
|
|
|
|
void
|
|
SwapManager::SetSuspended(Buffer &buf, bool on)
|
|
{
|
|
std::lock_guard<std::mutex> lg(mtx_);
|
|
auto it = journals_.find(&buf);
|
|
if (it == journals_.end())
|
|
return;
|
|
it->second.suspended = on;
|
|
}
|
|
|
|
|
|
SwapManager::SuspendGuard::SuspendGuard(SwapManager &m, Buffer *b)
|
|
: m_(m), buf_(b), prev_(false)
|
|
{
|
|
if (!buf_)
|
|
return;
|
|
{
|
|
std::lock_guard<std::mutex> lg(m_.mtx_);
|
|
auto it = m_.journals_.find(buf_);
|
|
if (it != m_.journals_.end()) {
|
|
prev_ = it->second.suspended;
|
|
it->second.suspended = true;
|
|
}
|
|
}
|
|
}
|
|
|
|
|
|
SwapManager::SuspendGuard::~SuspendGuard()
|
|
{
|
|
if (!buf_)
|
|
return;
|
|
std::lock_guard<std::mutex> lg(m_.mtx_);
|
|
auto it = m_.journals_.find(buf_);
|
|
if (it != m_.journals_.end()) {
|
|
it->second.suspended = prev_;
|
|
}
|
|
}
|
|
|
|
|
|
std::string
|
|
SwapManager::ComputeSidecarPath(const Buffer &buf)
|
|
{
|
|
fs::path root = swap_root_dir();
|
|
if (!buf.Filename().empty()) {
|
|
return compute_swap_path_for_filename(buf.Filename());
|
|
}
|
|
|
|
// Unnamed buffers: unique across processes (pid) and within one (counter).
|
|
// A per-process counter alone let a new session's scratch buffer claim,
|
|
// and later delete, another running session's unnamed journal.
|
|
static std::atomic<std::uint64_t> ctr{0};
|
|
const std::uint64_t n = ++ctr;
|
|
const std::string name = "unnamed-" + std::to_string(static_cast<long>(::getpid())) + "-" +
|
|
std::to_string(n) + ".swp";
|
|
return (root / name).string();
|
|
}
|
|
|
|
|
|
std::string
|
|
SwapManager::ComputeSwapPathForFilename(const std::string &filename)
|
|
{
|
|
return ComputeSidecarPathForFilename(filename);
|
|
}
|
|
|
|
|
|
std::string
|
|
SwapManager::ComputeSidecarPathForFilename(const std::string &filename)
|
|
{
|
|
return compute_swap_path_for_filename(filename);
|
|
}
|
|
|
|
|
|
std::uint64_t
|
|
SwapManager::now_ns()
|
|
{
|
|
using namespace std::chrono;
|
|
return duration_cast<nanoseconds>(steady_clock::now().time_since_epoch()).count();
|
|
}
|
|
|
|
|
|
bool
|
|
SwapManager::ensure_parent_dir(const std::string &path)
|
|
{
|
|
try {
|
|
fs::path p(path);
|
|
fs::path dir = p.parent_path();
|
|
if (dir.empty())
|
|
return true;
|
|
if (!fs::exists(dir))
|
|
fs::create_directories(dir);
|
|
return true;
|
|
} catch (...) {
|
|
return false;
|
|
}
|
|
}
|
|
|
|
|
|
bool
|
|
SwapManager::write_header(int fd, const JournalCtx &ctx)
|
|
{
|
|
if (fd < 0)
|
|
return false;
|
|
// Fixed 64-byte header (v1)
|
|
// [magic 8][version u32][flags u32][created_time u64]
|
|
// [base_size u64][base_mtime_ns i64] (valid when flags bit 0 is set)
|
|
// [reserved/padding]
|
|
std::uint8_t hdr[64];
|
|
std::memset(hdr, 0, sizeof(hdr));
|
|
std::memcpy(hdr, MAGIC, 8);
|
|
// version (little-endian)
|
|
hdr[8] = static_cast<std::uint8_t>(VERSION & 0xFFu);
|
|
hdr[9] = static_cast<std::uint8_t>((VERSION >> 8) & 0xFFu);
|
|
hdr[10] = static_cast<std::uint8_t>((VERSION >> 16) & 0xFFu);
|
|
hdr[11] = static_cast<std::uint8_t>((VERSION >> 24) & 0xFFu);
|
|
// flags: bit 0 = base file identity present
|
|
// flags: bit 1 = base content CRC-32 present (bytes 40..43)
|
|
if (ctx.has_base) {
|
|
hdr[12] = 1;
|
|
put_le64(hdr + 24, ctx.base_size);
|
|
put_le64(hdr + 32, static_cast<std::uint64_t>(ctx.base_mtime_ns));
|
|
if (ctx.has_base_crc) {
|
|
hdr[12] |= 2;
|
|
hdr[40] = static_cast<std::uint8_t>(ctx.base_crc & 0xFFu);
|
|
hdr[41] = static_cast<std::uint8_t>((ctx.base_crc >> 8) & 0xFFu);
|
|
hdr[42] = static_cast<std::uint8_t>((ctx.base_crc >> 16) & 0xFFu);
|
|
hdr[43] = static_cast<std::uint8_t>((ctx.base_crc >> 24) & 0xFFu);
|
|
}
|
|
}
|
|
// created_time (unix seconds; little-endian)
|
|
std::uint64_t ts = static_cast<std::uint64_t>(std::time(nullptr));
|
|
put_le64(hdr + 16, ts);
|
|
return write_full(fd, hdr, sizeof(hdr));
|
|
}
|
|
|
|
|
|
bool
|
|
SwapManager::open_ctx(JournalCtx &ctx, const std::string &path, std::string &err, bool *locked_out)
|
|
{
|
|
err.clear();
|
|
if (ctx.fd >= 0)
|
|
return true;
|
|
if (!ensure_parent_dir(path)) {
|
|
err = "Failed to create parent directory for swap file: " + path;
|
|
return false;
|
|
}
|
|
int flags = O_CREAT | O_WRONLY | O_APPEND;
|
|
#ifdef O_CLOEXEC
|
|
flags |= O_CLOEXEC;
|
|
#endif
|
|
|
|
// Retry on transient errors (ENOSPC, EDQUOT, EBUSY, etc.)
|
|
int fd = -1;
|
|
auto open_fn = [&]() -> bool {
|
|
fd = kte::syscall::Open(path.c_str(), flags, 0600);
|
|
return fd >= 0;
|
|
};
|
|
|
|
if (!RetryOnTransientError(open_fn, RetryPolicy::Aggressive(), err)) {
|
|
if (fd < 0) {
|
|
int saved_errno = errno;
|
|
err = "Failed to open swap file '" + path + "': " + std::strerror(saved_errno) + err;
|
|
}
|
|
return false;
|
|
}
|
|
// Take ownership of the journal. A lock held by another kte process means
|
|
// that session is journaling this file: writing here too would interleave
|
|
// its records with ours.
|
|
if (::flock(fd, LOCK_EX | LOCK_NB) != 0 && errno == EWOULDBLOCK) {
|
|
kte::syscall::Close(fd);
|
|
if (locked_out)
|
|
*locked_out = true; // caller records it under mtx_
|
|
err = "Swap file is in use by another kte process: " + path;
|
|
return false;
|
|
}
|
|
// Ensure permissions even if file already existed.
|
|
(void) kte::syscall::Fchmod(fd, 0600);
|
|
struct stat st{};
|
|
if (kte::syscall::Fstat(fd, &st) != 0) {
|
|
int saved_errno = errno;
|
|
kte::syscall::Close(fd);
|
|
err = "Failed to fstat swap file '" + path + "': " + std::strerror(saved_errno);
|
|
return false;
|
|
}
|
|
// If an existing file is too small to contain the fixed header, truncate
|
|
// and restart.
|
|
if (st.st_size > 0 && st.st_size < 64) {
|
|
kte::syscall::Close(fd);
|
|
int tflags = O_CREAT | O_WRONLY | O_TRUNC | O_APPEND;
|
|
#ifdef O_CLOEXEC
|
|
tflags |= O_CLOEXEC;
|
|
#endif
|
|
|
|
// Retry on transient errors for truncation open
|
|
fd = -1;
|
|
auto reopen_fn = [&]() -> bool {
|
|
fd = kte::syscall::Open(path.c_str(), tflags, 0600);
|
|
return fd >= 0;
|
|
};
|
|
|
|
if (!RetryOnTransientError(reopen_fn, RetryPolicy::Aggressive(), err)) {
|
|
if (fd < 0) {
|
|
int saved_errno = errno;
|
|
err = "Failed to reopen swap file for truncation '" + path + "': " + std::strerror(
|
|
saved_errno) + err;
|
|
}
|
|
return false;
|
|
}
|
|
if (::flock(fd, LOCK_EX | LOCK_NB) != 0 && errno == EWOULDBLOCK) {
|
|
kte::syscall::Close(fd);
|
|
if (locked_out)
|
|
*locked_out = true; // caller records it under mtx_
|
|
err = "Swap file is in use by another kte process: " + path;
|
|
return false;
|
|
}
|
|
(void) kte::syscall::Fchmod(fd, 0600);
|
|
st.st_size = 0;
|
|
}
|
|
// (ctx.path is not written here: this runs on the writer thread and the
|
|
// main thread reads ctx.path under mtx_; callers pass ctx.path itself.)
|
|
ctx.fd = fd;
|
|
if (st.st_size == 0) {
|
|
ctx.header_ok = write_header(fd, ctx);
|
|
ctx.approx_size_bytes = ctx.header_ok ? 64 : 0;
|
|
if (!ctx.header_ok) {
|
|
err = "Failed to write swap file header: " + path;
|
|
// Drop any partial header and close, so the next attempt starts a
|
|
// fresh file instead of appending records behind a torn header.
|
|
(void) kte::syscall::Ftruncate(fd, 0);
|
|
kte::syscall::Close(fd);
|
|
ctx.fd = -1;
|
|
}
|
|
} else {
|
|
ctx.header_ok = true; // stage 1: trust existing header
|
|
ctx.approx_size_bytes = static_cast<std::uint64_t>(st.st_size);
|
|
}
|
|
return ctx.header_ok;
|
|
}
|
|
|
|
|
|
void
|
|
SwapManager::close_ctx(JournalCtx &ctx)
|
|
{
|
|
if (ctx.fd >= 0) {
|
|
(void) kte::syscall::Fsync(ctx.fd);
|
|
kte::syscall::Close(ctx.fd);
|
|
ctx.fd = -1;
|
|
}
|
|
ctx.header_ok = false;
|
|
}
|
|
|
|
|
|
bool
|
|
SwapManager::compact_to_checkpoint(JournalCtx &ctx, const std::vector<std::uint8_t> &chkpt_record, std::string &err)
|
|
{
|
|
err.clear();
|
|
if (ctx.path.empty()) {
|
|
err = "Compact failed: empty path";
|
|
return false;
|
|
}
|
|
if (chkpt_record.empty()) {
|
|
err = "Compact failed: empty checkpoint record";
|
|
return false;
|
|
}
|
|
|
|
// Keep the current (locked) journal fd open until the compacted file has
|
|
// replaced it, so the journal is never unlocked in between: another
|
|
// session would otherwise see it as abandoned and offer to recover or
|
|
// discard it.
|
|
const std::string tmp_path = ctx.path + ".tmp";
|
|
// Create the compacted file: header + checkpoint record.
|
|
if (!ensure_parent_dir(tmp_path)) {
|
|
err = "Failed to create parent directory for temp swap file: " + tmp_path;
|
|
return false;
|
|
}
|
|
|
|
int flags = O_CREAT | O_WRONLY | O_TRUNC;
|
|
#ifdef O_CLOEXEC
|
|
flags |= O_CLOEXEC;
|
|
#endif
|
|
|
|
// Retry on transient errors for temp file creation
|
|
int tfd = -1;
|
|
auto open_tmp_fn = [&]() -> bool {
|
|
tfd = kte::syscall::Open(tmp_path.c_str(), flags, 0600);
|
|
return tfd >= 0;
|
|
};
|
|
|
|
if (!RetryOnTransientError(open_tmp_fn, RetryPolicy::Aggressive(), err)) {
|
|
if (tfd < 0) {
|
|
int saved_errno = errno;
|
|
err = "Failed to open temp swap file '" + tmp_path + "': " + std::strerror(saved_errno) + err;
|
|
}
|
|
return false;
|
|
}
|
|
(void) kte::syscall::Fchmod(tfd, 0600);
|
|
bool ok = write_header(tfd, ctx);
|
|
if (ok)
|
|
ok = write_full(tfd, chkpt_record.data(), chkpt_record.size());
|
|
if (ok) {
|
|
if (kte::syscall::Fsync(tfd) != 0) {
|
|
int saved_errno = errno;
|
|
err = "Failed to fsync temp swap file '" + tmp_path + "': " + std::strerror(saved_errno);
|
|
ok = false;
|
|
}
|
|
}
|
|
if (ok)
|
|
(void) ::flock(tfd, LOCK_EX | LOCK_NB); // new file: nobody else has it
|
|
if (!ok) {
|
|
kte::syscall::Close(tfd);
|
|
if (err.empty()) {
|
|
err = "Failed to write temp swap file: " + tmp_path;
|
|
}
|
|
std::remove(tmp_path.c_str());
|
|
return false;
|
|
}
|
|
|
|
// Atomic replace.
|
|
if (::rename(tmp_path.c_str(), ctx.path.c_str()) != 0) {
|
|
int saved_errno = errno;
|
|
kte::syscall::Close(tfd);
|
|
err = "Failed to rename temp swap file '" + tmp_path + "' to '" + ctx.path + "': " + std::strerror(
|
|
saved_errno);
|
|
std::remove(tmp_path.c_str());
|
|
return false;
|
|
}
|
|
// The compacted file is the journal now; retire the old descriptor.
|
|
if (ctx.fd >= 0)
|
|
kte::syscall::Close(ctx.fd);
|
|
const int fl = ::fcntl(tfd, F_GETFL);
|
|
if (fl >= 0)
|
|
(void) ::fcntl(tfd, F_SETFL, fl | O_APPEND);
|
|
ctx.fd = tfd;
|
|
ctx.header_ok = true;
|
|
|
|
// Best-effort: fsync parent dir to persist the rename.
|
|
try {
|
|
fs::path p(ctx.path);
|
|
fs::path dir = p.parent_path();
|
|
if (!dir.empty()) {
|
|
int dflags = O_RDONLY;
|
|
#ifdef O_DIRECTORY
|
|
dflags |= O_DIRECTORY;
|
|
#endif
|
|
int dfd = kte::syscall::Open(dir.string().c_str(), dflags);
|
|
if (dfd >= 0) {
|
|
(void) kte::syscall::Fsync(dfd);
|
|
kte::syscall::Close(dfd);
|
|
}
|
|
}
|
|
} catch (...) {
|
|
// ignore
|
|
}
|
|
|
|
// Re-open for further appends.
|
|
ctx.approx_size_bytes = 64 + static_cast<std::uint64_t>(chkpt_record.size());
|
|
return true;
|
|
}
|
|
|
|
|
|
std::uint32_t
|
|
SwapManager::crc32(const std::uint8_t *data, std::size_t len, std::uint32_t seed)
|
|
{
|
|
// Built once, thread-safely (used by the writer thread and by replay on
|
|
// the main thread; the old lazy init with a plain bool was a data race).
|
|
static const std::array<std::uint32_t, 256> table = [] {
|
|
std::array<std::uint32_t, 256> t{};
|
|
for (std::uint32_t i = 0; i < 256; ++i) {
|
|
std::uint32_t c = i;
|
|
for (int j = 0; j < 8; ++j)
|
|
c = (c & 1) ? (0xEDB88320u ^ (c >> 1)) : (c >> 1);
|
|
t[i] = c;
|
|
}
|
|
return t;
|
|
}();
|
|
std::uint32_t c = ~seed;
|
|
for (std::size_t i = 0; i < len; ++i)
|
|
c = table[(c ^ data[i]) & 0xFFu] ^ (c >> 8);
|
|
return ~c;
|
|
}
|
|
|
|
|
|
void
|
|
SwapManager::put_le32(std::vector<std::uint8_t> &out, std::uint32_t v)
|
|
{
|
|
out.push_back(static_cast<std::uint8_t>(v & 0xFFu));
|
|
out.push_back(static_cast<std::uint8_t>((v >> 8) & 0xFFu));
|
|
out.push_back(static_cast<std::uint8_t>((v >> 16) & 0xFFu));
|
|
out.push_back(static_cast<std::uint8_t>((v >> 24) & 0xFFu));
|
|
}
|
|
|
|
|
|
void
|
|
SwapManager::put_le64(std::uint8_t *dst, std::uint64_t v)
|
|
{
|
|
dst[0] = static_cast<std::uint8_t>(v & 0xFFu);
|
|
dst[1] = static_cast<std::uint8_t>((v >> 8) & 0xFFu);
|
|
dst[2] = static_cast<std::uint8_t>((v >> 16) & 0xFFu);
|
|
dst[3] = static_cast<std::uint8_t>((v >> 24) & 0xFFu);
|
|
dst[4] = static_cast<std::uint8_t>((v >> 32) & 0xFFu);
|
|
dst[5] = static_cast<std::uint8_t>((v >> 40) & 0xFFu);
|
|
dst[6] = static_cast<std::uint8_t>((v >> 48) & 0xFFu);
|
|
dst[7] = static_cast<std::uint8_t>((v >> 56) & 0xFFu);
|
|
}
|
|
|
|
|
|
void
|
|
SwapManager::put_u24_le(std::uint8_t dst[3], std::uint32_t v)
|
|
{
|
|
dst[0] = static_cast<std::uint8_t>(v & 0xFFu);
|
|
dst[1] = static_cast<std::uint8_t>((v >> 8) & 0xFFu);
|
|
dst[2] = static_cast<std::uint8_t>((v >> 16) & 0xFFu);
|
|
}
|
|
|
|
|
|
void
|
|
SwapManager::enqueue(Pending &&p)
|
|
{
|
|
{
|
|
std::lock_guard<std::mutex> lg(mtx_);
|
|
p.seq = ++next_seq_;
|
|
queue_.emplace_back(std::move(p));
|
|
}
|
|
cv_.notify_one();
|
|
}
|
|
|
|
|
|
void
|
|
SwapManager::RecordInsert(Buffer &buf, int row, int col, std::string_view text)
|
|
{
|
|
{
|
|
std::lock_guard<std::mutex> lg(mtx_);
|
|
auto it = journals_.find(&buf);
|
|
if (it == journals_.end() || it->second.suspended)
|
|
return;
|
|
}
|
|
// payload v1: [encver u8=1][row u32][col u32][nbytes u32][bytes]
|
|
// A record holds at most kMaxRecordPayload bytes, so large inserts are
|
|
// split into consecutive INS records, each positioned after the last.
|
|
constexpr std::size_t kInsHeader = 13;
|
|
constexpr std::size_t kChunk = kMaxRecordPayload - kInsHeader;
|
|
std::size_t r = static_cast<std::size_t>(std::max(0, row));
|
|
std::size_t c = static_cast<std::size_t>(std::max(0, col));
|
|
std::size_t off = 0;
|
|
do {
|
|
const std::string_view chunk = text.substr(off, kChunk);
|
|
Pending p;
|
|
p.buf = &buf;
|
|
p.type = SwapRecType::INS;
|
|
p.payload.reserve(kInsHeader + chunk.size());
|
|
p.payload.push_back(1);
|
|
put_le32(p.payload, static_cast<std::uint32_t>(r));
|
|
put_le32(p.payload, static_cast<std::uint32_t>(c));
|
|
put_le32(p.payload, static_cast<std::uint32_t>(chunk.size()));
|
|
p.payload.insert(p.payload.end(), reinterpret_cast<const std::uint8_t *>(chunk.data()),
|
|
reinterpret_cast<const std::uint8_t *>(chunk.data()) + chunk.size());
|
|
enqueue(std::move(p));
|
|
// Advance (r, c) past the chunk for the next record.
|
|
const std::size_t last_nl = chunk.rfind('\n');
|
|
if (last_nl == std::string_view::npos) {
|
|
c += chunk.size();
|
|
} else {
|
|
r += static_cast<std::size_t>(std::count(chunk.begin(), chunk.end(), '\n'));
|
|
c = chunk.size() - last_nl - 1;
|
|
}
|
|
off += chunk.size();
|
|
} while (off < text.size());
|
|
maybe_request_checkpoint(buf, text.size());
|
|
}
|
|
|
|
|
|
void
|
|
SwapManager::RecordDelete(Buffer &buf, int row, int col, std::size_t len)
|
|
{
|
|
{
|
|
std::lock_guard<std::mutex> lg(mtx_);
|
|
auto it = journals_.find(&buf);
|
|
if (it == journals_.end() || it->second.suspended)
|
|
return;
|
|
}
|
|
if (len > 0xFFFFFFFFu)
|
|
return;
|
|
Pending p;
|
|
p.buf = &buf;
|
|
p.type = SwapRecType::DEL;
|
|
// payload v1: [encver u8=1][row u32][col u32][len u32]
|
|
p.payload.push_back(1);
|
|
put_le32(p.payload, static_cast<std::uint32_t>(std::max(0, row)));
|
|
put_le32(p.payload, static_cast<std::uint32_t>(std::max(0, col)));
|
|
put_le32(p.payload, static_cast<std::uint32_t>(len));
|
|
enqueue(std::move(p));
|
|
maybe_request_checkpoint(buf, len);
|
|
}
|
|
|
|
|
|
void
|
|
SwapManager::RecordSplit(Buffer &buf, int row, int col)
|
|
{
|
|
{
|
|
std::lock_guard<std::mutex> lg(mtx_);
|
|
auto it = journals_.find(&buf);
|
|
if (it == journals_.end() || it->second.suspended)
|
|
return;
|
|
}
|
|
Pending p;
|
|
p.buf = &buf;
|
|
p.type = SwapRecType::SPLIT;
|
|
// payload v1: [encver u8=1][row u32][col u32]
|
|
p.payload.push_back(1);
|
|
put_le32(p.payload, static_cast<std::uint32_t>(std::max(0, row)));
|
|
put_le32(p.payload, static_cast<std::uint32_t>(std::max(0, col)));
|
|
enqueue(std::move(p));
|
|
maybe_request_checkpoint(buf, 1);
|
|
}
|
|
|
|
|
|
void
|
|
SwapManager::RecordJoin(Buffer &buf, int row)
|
|
{
|
|
{
|
|
std::lock_guard<std::mutex> lg(mtx_);
|
|
auto it = journals_.find(&buf);
|
|
if (it == journals_.end() || it->second.suspended)
|
|
return;
|
|
}
|
|
Pending p;
|
|
p.buf = &buf;
|
|
p.type = SwapRecType::JOIN;
|
|
// payload v1: [encver u8=1][row u32]
|
|
p.payload.push_back(1);
|
|
put_le32(p.payload, static_cast<std::uint32_t>(std::max(0, row)));
|
|
enqueue(std::move(p));
|
|
maybe_request_checkpoint(buf, 1);
|
|
}
|
|
|
|
|
|
void
|
|
SwapManager::maybe_request_checkpoint(Buffer &buf, const std::size_t approx_edit_bytes)
|
|
{
|
|
SwapConfig cfg;
|
|
bool do_chkpt = false;
|
|
{
|
|
std::lock_guard<std::mutex> lg(mtx_);
|
|
cfg = cfg_;
|
|
auto it = journals_.find(&buf);
|
|
if (it == journals_.end() || it->second.suspended)
|
|
return;
|
|
if (cfg.checkpoint_bytes == 0 && cfg.checkpoint_interval_ms == 0 && !it->second.gap)
|
|
return;
|
|
JournalCtx &ctx = it->second;
|
|
ctx.edit_bytes_since_chkpt += approx_edit_bytes;
|
|
const std::uint64_t now = now_ns();
|
|
if (ctx.last_chkpt_ns == 0)
|
|
ctx.last_chkpt_ns = now;
|
|
const bool bytes_hit = (cfg.checkpoint_bytes > 0) && (
|
|
ctx.edit_bytes_since_chkpt >= cfg.checkpoint_bytes);
|
|
const bool time_hit = (cfg.checkpoint_interval_ms > 0) &&
|
|
(((now - ctx.last_chkpt_ns) / 1000000ULL) >= cfg.checkpoint_interval_ms);
|
|
const bool gap_hit = ctx.gap && (now - ctx.gap_chkpt_request_ns) >= kGapCheckpointIntervalNs;
|
|
if (gap_hit)
|
|
ctx.gap_chkpt_request_ns = now;
|
|
if (bytes_hit || time_hit || gap_hit) {
|
|
ctx.edit_bytes_since_chkpt = 0;
|
|
ctx.last_chkpt_ns = now;
|
|
do_chkpt = true;
|
|
}
|
|
}
|
|
if (do_chkpt) {
|
|
RecordCheckpoint(buf, false);
|
|
}
|
|
}
|
|
|
|
|
|
void
|
|
SwapManager::RetryGapCheckpoints()
|
|
{
|
|
std::vector<Buffer *> due;
|
|
{
|
|
std::lock_guard<std::mutex> lg(mtx_);
|
|
const std::uint64_t now = now_ns();
|
|
for (auto &[b, ctx]: journals_) {
|
|
if (!b || !ctx.gap || ctx.suspended || ctx.locked_out || ctx.gap_unfixable_reported)
|
|
continue;
|
|
if (now - ctx.gap_chkpt_request_ns < kGapCheckpointIntervalNs)
|
|
continue;
|
|
ctx.gap_chkpt_request_ns = now;
|
|
due.push_back(b);
|
|
}
|
|
}
|
|
for (Buffer *b: due)
|
|
RecordCheckpoint(*b, false);
|
|
}
|
|
|
|
|
|
void
|
|
SwapManager::RecordCheckpoint(Buffer &buf, const bool urgent_flush)
|
|
{
|
|
{
|
|
std::lock_guard<std::mutex> lg(mtx_);
|
|
auto it = journals_.find(&buf);
|
|
if (it == journals_.end() || it->second.suspended)
|
|
return;
|
|
}
|
|
|
|
// payload = encver(1) + nbytes(4) + bytes; larger buffers cannot be
|
|
// checkpointed in this record format (the writer would reject them), so
|
|
// check the size before paying for a snapshot.
|
|
const std::size_t nrows = buf.Nrows();
|
|
const std::size_t total_size = nrows ? buf.GetLineRange(nrows - 1).second : 0;
|
|
if (total_size > kMaxRecordPayload - 5) {
|
|
// If the journal has a gap, only a checkpoint could close it: say so
|
|
// once rather than silently dropping every later record.
|
|
bool report = false;
|
|
{
|
|
std::lock_guard<std::mutex> lg(mtx_);
|
|
auto it = journals_.find(&buf);
|
|
if (it != journals_.end() && it->second.gap && !it->second.gap_unfixable_reported) {
|
|
it->second.gap_unfixable_reported = true;
|
|
report = true;
|
|
}
|
|
}
|
|
if (report)
|
|
report_error("Swap journal lost a record and the buffer is too large to checkpoint; "
|
|
"crash recovery will not include edits made until the file is saved", &buf);
|
|
return;
|
|
}
|
|
const std::string bytes = snapshot_buffer_bytes(buf);
|
|
if (bytes.size() > kMaxRecordPayload - 5)
|
|
return;
|
|
|
|
Pending p;
|
|
p.buf = &buf;
|
|
p.type = SwapRecType::CHKPT;
|
|
p.urgent_flush = urgent_flush;
|
|
// payload v1: [encver u8=1][nbytes u32][bytes]
|
|
p.payload.push_back(1);
|
|
put_le32(p.payload, static_cast<std::uint32_t>(bytes.size()));
|
|
p.payload.insert(p.payload.end(), reinterpret_cast<const std::uint8_t *>(bytes.data()),
|
|
reinterpret_cast<const std::uint8_t *>(bytes.data()) + bytes.size());
|
|
enqueue(std::move(p));
|
|
}
|
|
|
|
|
|
void
|
|
SwapManager::writer_loop()
|
|
{
|
|
for (;;) {
|
|
std::vector<Pending> batch;
|
|
{
|
|
std::unique_lock<std::mutex> lk(mtx_);
|
|
if (queue_.empty()) {
|
|
if (!running_.load())
|
|
break;
|
|
cv_.wait_for(lk, std::chrono::milliseconds(cfg_.flush_interval_ms));
|
|
}
|
|
if (!queue_.empty()) {
|
|
batch.swap(queue_);
|
|
inflight_ += batch.size();
|
|
}
|
|
}
|
|
if (batch.empty())
|
|
continue;
|
|
|
|
for (const Pending &p: batch) {
|
|
try {
|
|
process_one(p);
|
|
} catch (const std::exception &e) {
|
|
report_error(std::string("Exception in process_one: ") + e.what(), p.buf);
|
|
} catch (...) {
|
|
report_error("Unknown exception in process_one", p.buf);
|
|
}
|
|
{
|
|
std::lock_guard<std::mutex> lg(mtx_);
|
|
if (p.seq > last_processed_)
|
|
last_processed_ = p.seq;
|
|
if (inflight_ > 0)
|
|
--inflight_;
|
|
}
|
|
cv_.notify_all();
|
|
}
|
|
|
|
// Throttled fsync: best-effort (grouped)
|
|
try {
|
|
std::uint64_t now = now_ns();
|
|
// Hold io_mtx_ across the fsyncs: journal fds are only closed
|
|
// under it (by the main thread), so none of these can be closed or
|
|
// reused mid-fsync. Unsynchronised, an in-flight fsync also kept a
|
|
// just-closed journal's flock alive, so the owner's own cleanup saw
|
|
// its journal as held. mtx_ is held only to collect the fds, so
|
|
// edits are not blocked while fsync runs.
|
|
std::lock_guard<std::mutex> io(io_mtx_);
|
|
std::vector<int> to_sync;
|
|
{
|
|
std::lock_guard<std::mutex> lg(mtx_);
|
|
for (auto &kv: journals_) {
|
|
JournalCtx &ctx = kv.second;
|
|
if (ctx.fd >= 0) {
|
|
if (ctx.last_fsync_ns == 0 || (now - ctx.last_fsync_ns) / 1000000ULL >=
|
|
cfg_.fsync_interval_ms) {
|
|
ctx.last_fsync_ns = now;
|
|
to_sync.push_back(ctx.fd);
|
|
}
|
|
}
|
|
}
|
|
}
|
|
for (int fd: to_sync)
|
|
(void) kte::syscall::Fsync(fd);
|
|
} catch (const std::exception &e) {
|
|
report_error(std::string("Exception in fsync operations: ") + e.what());
|
|
} catch (...) {
|
|
report_error("Unknown exception in fsync operations");
|
|
}
|
|
}
|
|
// Wake any waiters.
|
|
cv_.notify_all();
|
|
}
|
|
|
|
|
|
void
|
|
SwapManager::process_one(const Pending &p)
|
|
{
|
|
if (!p.buf)
|
|
return;
|
|
|
|
// Any record that is not written leaves a gap in the journal.
|
|
auto mark_gap = [&]() {
|
|
std::lock_guard<std::mutex> lg(mtx_);
|
|
auto it = journals_.find(p.buf);
|
|
if (it != journals_.end())
|
|
it->second.gap = true;
|
|
};
|
|
|
|
// Check circuit breaker before processing
|
|
bool circuit_open = false;
|
|
{
|
|
std::lock_guard<std::mutex> lg(mtx_);
|
|
if (!circuit_breaker_.AllowRequest()) {
|
|
circuit_open = true;
|
|
}
|
|
}
|
|
|
|
if (circuit_open) {
|
|
// Circuit is open - graceful degradation: skip swap write
|
|
// This prevents repeated failures from overwhelming the system
|
|
// Swap recording will resume when circuit closes
|
|
static std::atomic<std::uint64_t> last_warning_ns{0};
|
|
const std::uint64_t now = now_ns();
|
|
const std::uint64_t last = last_warning_ns.load();
|
|
// Log warning at most once per 60 seconds to avoid spam
|
|
if (now - last > 60000000000ULL) {
|
|
last_warning_ns.store(now);
|
|
// Journal path read under mtx_, not Buffer::Filename(), which the
|
|
// main thread may be assigning (see report_error).
|
|
std::string context = "<unnamed>";
|
|
{
|
|
std::lock_guard<std::mutex> lg(mtx_);
|
|
auto it = journals_.find(p.buf);
|
|
if (it != journals_.end() && !it->second.path.empty())
|
|
context = it->second.path;
|
|
}
|
|
ErrorHandler::Instance().Warning("SwapManager",
|
|
"Swap operations temporarily disabled due to repeated failures (circuit breaker open)",
|
|
context);
|
|
}
|
|
mark_gap();
|
|
return;
|
|
}
|
|
|
|
try {
|
|
Buffer &buf = *p.buf;
|
|
|
|
JournalCtx *ctxp = nullptr;
|
|
std::string path;
|
|
std::size_t compact_bytes = 0;
|
|
{
|
|
std::lock_guard<std::mutex> lg(mtx_);
|
|
auto it = journals_.find(p.buf);
|
|
if (it == journals_.end())
|
|
return;
|
|
// Another process owns this journal (reported once when found).
|
|
if (it->second.locked_out)
|
|
return;
|
|
// After a gap only a checkpoint can bring the journal back in step.
|
|
if (it->second.gap && p.type != SwapRecType::CHKPT)
|
|
return;
|
|
if (it->second.path.empty())
|
|
it->second.path = ComputeSidecarPath(buf);
|
|
path = it->second.path;
|
|
ctxp = &it->second;
|
|
compact_bytes = cfg_.compact_bytes;
|
|
}
|
|
if (!ctxp)
|
|
return;
|
|
std::string open_err;
|
|
bool locked_out = false;
|
|
if (!open_ctx(*ctxp, path, open_err, &locked_out)) {
|
|
if (locked_out) {
|
|
{
|
|
std::lock_guard<std::mutex> lg(mtx_);
|
|
ctxp->locked_out = true;
|
|
}
|
|
report_error(open_err, p.buf);
|
|
return; // not an I/O failure; later records are skipped
|
|
}
|
|
report_error(open_err, p.buf);
|
|
{
|
|
std::lock_guard<std::mutex> lg(mtx_);
|
|
circuit_breaker_.RecordFailure();
|
|
}
|
|
mark_gap();
|
|
return;
|
|
}
|
|
if (p.payload.size() > kMaxRecordPayload) {
|
|
// Recorders keep payloads within the limit, so this is a bug, not an
|
|
// I/O failure: do not count it against the circuit breaker.
|
|
report_error("Payload too large: " + std::to_string(p.payload.size()) + " bytes", p.buf);
|
|
mark_gap();
|
|
return;
|
|
}
|
|
|
|
// Build record: [type u8][len u24][payload][crc32 u32]
|
|
std::uint8_t len3[3];
|
|
put_u24_le(len3, static_cast<std::uint32_t>(p.payload.size()));
|
|
|
|
std::uint8_t head[4];
|
|
head[0] = static_cast<std::uint8_t>(p.type);
|
|
head[1] = len3[0];
|
|
head[2] = len3[1];
|
|
head[3] = len3[2];
|
|
|
|
std::uint32_t c = 0;
|
|
c = crc32(head, sizeof(head), c);
|
|
if (!p.payload.empty())
|
|
c = crc32(p.payload.data(), p.payload.size(), c);
|
|
std::uint8_t crcbytes[4];
|
|
crcbytes[0] = static_cast<std::uint8_t>(c & 0xFFu);
|
|
crcbytes[1] = static_cast<std::uint8_t>((c >> 8) & 0xFFu);
|
|
crcbytes[2] = static_cast<std::uint8_t>((c >> 16) & 0xFFu);
|
|
crcbytes[3] = static_cast<std::uint8_t>((c >> 24) & 0xFFu);
|
|
|
|
std::vector<std::uint8_t> rec;
|
|
rec.reserve(sizeof(head) + p.payload.size() + sizeof(crcbytes));
|
|
rec.insert(rec.end(), head, head + sizeof(head));
|
|
if (!p.payload.empty())
|
|
rec.insert(rec.end(), p.payload.begin(), p.payload.end());
|
|
rec.insert(rec.end(), crcbytes, crcbytes + sizeof(crcbytes));
|
|
|
|
// Remember where this record starts so a partial write can be undone;
|
|
// otherwise later records would be appended behind the torn bytes.
|
|
struct stat before{};
|
|
const bool have_size = kte::syscall::Fstat(ctxp->fd, &before) == 0;
|
|
|
|
// Write (handle partial writes and check results)
|
|
bool ok = write_full(ctxp->fd, rec.data(), rec.size());
|
|
if (!ok) {
|
|
int err = errno;
|
|
report_error("Failed to write swap record to '" + path + "': " + std::strerror(err), p.buf);
|
|
if (have_size)
|
|
(void) kte::syscall::Ftruncate(ctxp->fd, before.st_size);
|
|
{
|
|
std::lock_guard<std::mutex> lg(mtx_);
|
|
circuit_breaker_.RecordFailure();
|
|
}
|
|
mark_gap();
|
|
return;
|
|
}
|
|
if (p.type == SwapRecType::CHKPT) {
|
|
std::lock_guard<std::mutex> lg(mtx_);
|
|
ctxp->gap = false;
|
|
}
|
|
ctxp->approx_size_bytes += static_cast<std::uint64_t>(rec.size());
|
|
if (p.urgent_flush) {
|
|
if (kte::syscall::Fsync(ctxp->fd) != 0) {
|
|
int err = errno;
|
|
report_error("Failed to fsync swap file '" + path + "': " + std::strerror(err), p.buf);
|
|
}
|
|
ctxp->last_fsync_ns = now_ns();
|
|
}
|
|
if (p.type == SwapRecType::CHKPT && compact_bytes > 0 &&
|
|
ctxp->approx_size_bytes >= static_cast<std::uint64_t>(compact_bytes)) {
|
|
std::string compact_err;
|
|
if (!compact_to_checkpoint(*ctxp, rec, compact_err)) {
|
|
report_error(compact_err, p.buf);
|
|
// Note: compaction failure is not fatal, don't record circuit breaker failure
|
|
}
|
|
}
|
|
|
|
// Record success for circuit breaker
|
|
{
|
|
std::lock_guard<std::mutex> lg(mtx_);
|
|
circuit_breaker_.RecordSuccess();
|
|
}
|
|
} catch (const std::exception &e) {
|
|
report_error(std::string("Exception in process_one: ") + e.what(), p.buf);
|
|
{
|
|
std::lock_guard<std::mutex> lg(mtx_);
|
|
circuit_breaker_.RecordFailure();
|
|
}
|
|
mark_gap();
|
|
} catch (...) {
|
|
report_error("Unknown exception in process_one", p.buf);
|
|
{
|
|
std::lock_guard<std::mutex> lg(mtx_);
|
|
circuit_breaker_.RecordFailure();
|
|
}
|
|
mark_gap();
|
|
}
|
|
}
|
|
|
|
|
|
static bool
|
|
read_exact(std::ifstream &in, void *dst, std::size_t n)
|
|
{
|
|
in.read(static_cast<char *>(dst), static_cast<std::streamsize>(n));
|
|
return in.good() && static_cast<std::size_t>(in.gcount()) == n;
|
|
}
|
|
|
|
|
|
static std::uint32_t
|
|
read_le32(const std::uint8_t b[4])
|
|
{
|
|
return (std::uint32_t) b[0] | ((std::uint32_t) b[1] << 8) | ((std::uint32_t) b[2] << 16) | (
|
|
(std::uint32_t) b[3] << 24);
|
|
}
|
|
|
|
|
|
static bool
|
|
parse_u32_le(const std::vector<std::uint8_t> &p, std::size_t &off, std::uint32_t &out)
|
|
{
|
|
if (off + 4 > p.size())
|
|
return false;
|
|
out = (std::uint32_t) p[off] | ((std::uint32_t) p[off + 1] << 8) | ((std::uint32_t) p[off + 2] << 16) |
|
|
((std::uint32_t) p[off + 3] << 24);
|
|
off += 4;
|
|
return true;
|
|
}
|
|
|
|
|
|
// Remove a journal this session is done with, unless another session holds
|
|
// it: a session that was locked out of the journal (or finds it locked) must
|
|
// not unlink the live journal of the session that owns it.
|
|
static void
|
|
remove_journal_unless_held(const std::string &path, bool locked_out)
|
|
{
|
|
if (path.empty() || locked_out || SwapManager::JournalInUse(path))
|
|
return;
|
|
(void) std::remove(path.c_str());
|
|
}
|
|
|
|
|
|
bool
|
|
SwapManager::JournalInUse(const std::string &swap_path)
|
|
{
|
|
int flags = O_RDONLY;
|
|
#ifdef O_CLOEXEC
|
|
flags |= O_CLOEXEC;
|
|
#endif
|
|
const int fd = kte::syscall::Open(swap_path.c_str(), flags);
|
|
if (fd < 0)
|
|
return false;
|
|
const bool in_use = ::flock(fd, LOCK_SH | LOCK_NB) != 0 && errno == EWOULDBLOCK;
|
|
kte::syscall::Close(fd); // also releases our shared lock, if taken
|
|
return in_use;
|
|
}
|
|
|
|
|
|
bool
|
|
SwapManager::JournalMatchesFile(const std::string &swap_path, const std::string &file_path)
|
|
{
|
|
std::ifstream in(swap_path, std::ios::binary);
|
|
std::uint8_t hdr[64];
|
|
if (!in || !in.read(reinterpret_cast<char *>(hdr), sizeof(hdr)))
|
|
return true; // unreadable: left to ReplayFile to report
|
|
if (std::memcmp(hdr, MAGIC, 8) != 0 || (hdr[12] & 1u) == 0)
|
|
return true; // no base identity recorded
|
|
std::uint64_t want_size = 0, want_mtime = 0;
|
|
for (int i = 7; i >= 0; --i) {
|
|
want_size = (want_size << 8) | hdr[24 + i];
|
|
want_mtime = (want_mtime << 8) | hdr[32 + i];
|
|
}
|
|
std::uint64_t size = 0;
|
|
std::int64_t mtime = 0;
|
|
if (!stat_base(file_path, size, mtime))
|
|
return false; // the file the journal was based on is gone
|
|
if (size != want_size)
|
|
return false;
|
|
if (static_cast<std::uint64_t>(mtime) == want_mtime)
|
|
return true;
|
|
// Timestamp changed (touch, checkout back and forth, sync tools): the
|
|
// content may still be the same, which is what matters.
|
|
if ((hdr[12] & 2u) == 0)
|
|
return false;
|
|
const std::uint32_t want_crc = static_cast<std::uint32_t>(hdr[40]) | (static_cast<std::uint32_t>(hdr[41]) << 8) |
|
|
(static_cast<std::uint32_t>(hdr[42]) << 16) |
|
|
(static_cast<std::uint32_t>(hdr[43]) << 24);
|
|
std::uint32_t crc = 0;
|
|
return file_crc32(file_path, crc) && crc == want_crc;
|
|
}
|
|
|
|
|
|
bool
|
|
SwapManager::file_crc32(const std::string &path, std::uint32_t &out)
|
|
{
|
|
std::ifstream in(path, std::ios::binary);
|
|
if (!in)
|
|
return false;
|
|
std::uint32_t c = 0;
|
|
std::uint8_t chunk[1 << 16];
|
|
while (in) {
|
|
in.read(reinterpret_cast<char *>(chunk), sizeof(chunk));
|
|
const auto n = in.gcount();
|
|
if (n > 0)
|
|
c = crc32(chunk, static_cast<std::size_t>(n), c);
|
|
}
|
|
if (in.bad())
|
|
return false;
|
|
out = c;
|
|
return true;
|
|
}
|
|
|
|
|
|
SwapManager::BaseId
|
|
SwapManager::compute_base(const std::string &file)
|
|
{
|
|
// Content CRCs are skipped for very large files (reading them again on
|
|
// every open and save is too slow); size and mtime still identify them.
|
|
constexpr std::uint64_t kMaxCrcBytes = std::uint64_t{64} << 20;
|
|
BaseId id;
|
|
id.has = stat_base(file, id.size, id.mtime_ns);
|
|
id.has_crc = id.has && id.size <= kMaxCrcBytes && file_crc32(file, id.crc);
|
|
return id;
|
|
}
|
|
|
|
|
|
void
|
|
SwapManager::apply_base(JournalCtx &ctx, const BaseId &id)
|
|
{
|
|
ctx.has_base = id.has;
|
|
ctx.base_size = id.size;
|
|
ctx.base_mtime_ns = id.mtime_ns;
|
|
ctx.has_base_crc = id.has_crc;
|
|
ctx.base_crc = id.crc;
|
|
}
|
|
|
|
|
|
bool
|
|
SwapManager::ReplayFile(Buffer &buf, const std::string &swap_path, std::string &err,
|
|
std::uint64_t *valid_bytes)
|
|
{
|
|
err.clear();
|
|
std::ifstream in(swap_path, std::ios::binary);
|
|
if (!in) {
|
|
err = "Failed to open swap file for replay: " + swap_path;
|
|
return false;
|
|
}
|
|
|
|
std::uint8_t hdr[64];
|
|
if (!read_exact(in, hdr, sizeof(hdr))) {
|
|
err = "Swap file truncated (header): " + swap_path;
|
|
return false;
|
|
}
|
|
if (std::memcmp(hdr, MAGIC, 8) != 0) {
|
|
err = "Swap file has bad magic: " + swap_path;
|
|
return false;
|
|
}
|
|
const std::uint32_t ver = read_le32(hdr + 8);
|
|
if (ver != VERSION) {
|
|
err = "Unsupported swap version: " + std::to_string(ver);
|
|
return false;
|
|
}
|
|
|
|
// Ensure replayed edits don't get re-journaled if the caller forgot to detach/suspend.
|
|
kte::SwapRecorder *prev_rec = buf.SwapRecorder();
|
|
buf.SetSwapRecorder(nullptr);
|
|
struct RestoreSwapRecorder {
|
|
Buffer &b;
|
|
kte::SwapRecorder *prev;
|
|
|
|
|
|
~RestoreSwapRecorder()
|
|
{
|
|
b.SetSwapRecorder(prev);
|
|
}
|
|
} restore{buf, prev_rec};
|
|
|
|
for (;;) {
|
|
// Everything before this offset replayed cleanly.
|
|
const std::streamoff record_start = in.tellg();
|
|
if (valid_bytes && record_start >= 0)
|
|
*valid_bytes = static_cast<std::uint64_t>(record_start);
|
|
std::uint8_t head[4];
|
|
in.read(reinterpret_cast<char *>(head), sizeof(head));
|
|
const std::size_t got_head = static_cast<std::size_t>(in.gcount());
|
|
if (got_head == 0 && in.eof()) {
|
|
return true; // clean EOF
|
|
}
|
|
// A record cut short by end-of-file is the expected result of a crash or
|
|
// power loss mid-append. Keep everything replayed so far and stop; the
|
|
// caller still sees err (with a true result) and can report it.
|
|
auto torn_tail = [&](const char *what) {
|
|
if (in.bad()) {
|
|
err = std::string("Failed to read swap file (") + what + "): " + swap_path;
|
|
return false;
|
|
}
|
|
err = std::string("Swap file ends with an incomplete record (") + what + "); recovered the records before it: " +
|
|
swap_path;
|
|
kte::ErrorHandler::Instance().Warning("SwapManager", err, swap_path);
|
|
return true;
|
|
};
|
|
if (got_head != sizeof(head))
|
|
return torn_tail("record header");
|
|
|
|
const SwapRecType type = static_cast<SwapRecType>(head[0]);
|
|
const std::size_t len = (std::size_t) head[1] | ((std::size_t) head[2] << 8) | (
|
|
(std::size_t) head[3] << 16);
|
|
std::vector<std::uint8_t> payload;
|
|
payload.resize(len);
|
|
if (len > 0 && !read_exact(in, payload.data(), len))
|
|
return torn_tail("payload");
|
|
std::uint8_t crcbytes[4];
|
|
if (!read_exact(in, crcbytes, sizeof(crcbytes)))
|
|
return torn_tail("crc");
|
|
const std::uint32_t want_crc = read_le32(crcbytes);
|
|
std::uint32_t got_crc = 0;
|
|
got_crc = crc32(head, sizeof(head), got_crc);
|
|
if (!payload.empty())
|
|
got_crc = crc32(payload.data(), payload.size(), got_crc);
|
|
if (got_crc != want_crc) {
|
|
err = "Swap file CRC mismatch: " + swap_path;
|
|
return false;
|
|
}
|
|
|
|
// Apply record
|
|
switch (type) {
|
|
case SwapRecType::INS: {
|
|
std::size_t off = 0;
|
|
// INS payload: encver(1) + row(4) + col(4) + nbytes(4) + data(nbytes)
|
|
// Minimum: 1 + 4 + 4 + 4 = 13 bytes
|
|
if (payload.size() < 13) {
|
|
err = "INS payload too short (need at least 13 bytes)";
|
|
return false;
|
|
}
|
|
const std::uint8_t encver = payload[off++];
|
|
if (encver != 1) {
|
|
err = "Unsupported swap payload encoding";
|
|
return false;
|
|
}
|
|
std::uint32_t row = 0, col = 0, nbytes = 0;
|
|
if (!parse_u32_le(payload, off, row) || !parse_u32_le(payload, off, col) || !parse_u32_le(
|
|
payload, off, nbytes)) {
|
|
err = "Malformed INS payload (failed to parse row/col/nbytes)";
|
|
return false;
|
|
}
|
|
if (off + nbytes > payload.size()) {
|
|
err = "Truncated INS payload bytes";
|
|
return false;
|
|
}
|
|
buf.insert_text((int) row, (int) col,
|
|
std::string_view(reinterpret_cast<const char *>(payload.data() + off), nbytes));
|
|
break;
|
|
}
|
|
case SwapRecType::DEL: {
|
|
std::size_t off = 0;
|
|
// DEL payload: encver(1) + row(4) + col(4) + dlen(4)
|
|
// Minimum: 1 + 4 + 4 + 4 = 13 bytes
|
|
if (payload.size() < 13) {
|
|
err = "DEL payload too short (need at least 13 bytes)";
|
|
return false;
|
|
}
|
|
const std::uint8_t encver = payload[off++];
|
|
if (encver != 1) {
|
|
err = "Unsupported swap payload encoding";
|
|
return false;
|
|
}
|
|
std::uint32_t row = 0, col = 0, dlen = 0;
|
|
if (!parse_u32_le(payload, off, row) || !parse_u32_le(payload, off, col) || !parse_u32_le(
|
|
payload, off, dlen)) {
|
|
err = "Malformed DEL payload (failed to parse row/col/dlen)";
|
|
return false;
|
|
}
|
|
buf.delete_text((int) row, (int) col, (std::size_t) dlen);
|
|
break;
|
|
}
|
|
case SwapRecType::SPLIT: {
|
|
std::size_t off = 0;
|
|
// SPLIT payload: encver(1) + row(4) + col(4)
|
|
// Minimum: 1 + 4 + 4 = 9 bytes
|
|
if (payload.size() < 9) {
|
|
err = "SPLIT payload too short (need at least 9 bytes)";
|
|
return false;
|
|
}
|
|
const std::uint8_t encver = payload[off++];
|
|
if (encver != 1) {
|
|
err = "Unsupported swap payload encoding";
|
|
return false;
|
|
}
|
|
std::uint32_t row = 0, col = 0;
|
|
if (!parse_u32_le(payload, off, row) || !parse_u32_le(payload, off, col)) {
|
|
err = "Malformed SPLIT payload (failed to parse row/col)";
|
|
return false;
|
|
}
|
|
buf.split_line((int) row, (int) col);
|
|
break;
|
|
}
|
|
case SwapRecType::JOIN: {
|
|
std::size_t off = 0;
|
|
// JOIN payload: encver(1) + row(4)
|
|
// Minimum: 1 + 4 = 5 bytes
|
|
if (payload.size() < 5) {
|
|
err = "JOIN payload too short (need at least 5 bytes)";
|
|
return false;
|
|
}
|
|
const std::uint8_t encver = payload[off++];
|
|
if (encver != 1) {
|
|
err = "Unsupported swap payload encoding";
|
|
return false;
|
|
}
|
|
std::uint32_t row = 0;
|
|
if (!parse_u32_le(payload, off, row)) {
|
|
err = "Malformed JOIN payload (failed to parse row)";
|
|
return false;
|
|
}
|
|
buf.join_lines((int) row);
|
|
break;
|
|
}
|
|
case SwapRecType::CHKPT: {
|
|
std::size_t off = 0;
|
|
// CHKPT payload: encver(1) + nbytes(4) + data(nbytes)
|
|
// Minimum: 1 + 4 = 5 bytes
|
|
if (payload.size() < 5) {
|
|
err = "CHKPT payload too short (need at least 5 bytes)";
|
|
return false;
|
|
}
|
|
const std::uint8_t encver = payload[off++];
|
|
if (encver != 1) {
|
|
err = "Unsupported swap checkpoint encoding";
|
|
return false;
|
|
}
|
|
std::uint32_t nbytes = 0;
|
|
if (!parse_u32_le(payload, off, nbytes)) {
|
|
err = "Malformed CHKPT payload (failed to parse nbytes)";
|
|
return false;
|
|
}
|
|
if (off + nbytes > payload.size()) {
|
|
err = "Truncated CHKPT payload bytes";
|
|
return false;
|
|
}
|
|
buf.replace_all_bytes(std::string_view(reinterpret_cast<const char *>(payload.data() + off),
|
|
(std::size_t) nbytes));
|
|
break;
|
|
}
|
|
default:
|
|
// Ignore unknown types for forward-compat
|
|
break;
|
|
}
|
|
}
|
|
}
|
|
|
|
|
|
void
|
|
SwapManager::report_error(const std::string &message, Buffer *buf)
|
|
{
|
|
// Identify the buffer by its journal path, read under mtx_. This runs on
|
|
// the writer thread, where reading Buffer::Filename() raced with the main
|
|
// thread assigning it (SaveAs, OpenFromFile).
|
|
std::string context = "<unknown>";
|
|
{
|
|
std::lock_guard<std::mutex> lg(mtx_);
|
|
if (buf) {
|
|
auto it = journals_.find(buf);
|
|
context = (it != journals_.end() && !it->second.path.empty()) ? it->second.path : "<unnamed>";
|
|
}
|
|
}
|
|
|
|
// Report to centralized error handler
|
|
ErrorHandler::Instance().Error("SwapManager", message, context);
|
|
|
|
// Maintain local error tracking for backward compatibility
|
|
std::lock_guard<std::mutex> lg(mtx_);
|
|
SwapError err;
|
|
err.timestamp_ns = now_ns();
|
|
err.message = message;
|
|
err.buffer_name = context;
|
|
errors_.push_back(err);
|
|
// Bound the error queue to 100 entries
|
|
while (errors_.size() > 100) {
|
|
errors_.pop_front();
|
|
}
|
|
++total_error_count_;
|
|
}
|
|
|
|
|
|
bool
|
|
SwapManager::HasErrors() const
|
|
{
|
|
std::lock_guard<std::mutex> lg(mtx_);
|
|
return !errors_.empty();
|
|
}
|
|
|
|
|
|
std::string
|
|
SwapManager::GetLastError() const
|
|
{
|
|
std::lock_guard<std::mutex> lg(mtx_);
|
|
if (errors_.empty())
|
|
return "";
|
|
const SwapError &e = errors_.back();
|
|
return "[" + e.buffer_name + "] " + e.message;
|
|
}
|
|
|
|
|
|
std::size_t
|
|
SwapManager::GetErrorCount() const
|
|
{
|
|
std::lock_guard<std::mutex> lg(mtx_);
|
|
return total_error_count_;
|
|
}
|
|
} // namespace kte
|