diff --git a/CLAUDE.md b/CLAUDE.md index 0f5a0b6..25b5e2b 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -24,7 +24,7 @@ cmake --build build --target kte_tests # test suite cmake --build build --target kte_tests && ./build/kte_tests ``` -There is no single-test runner; the test binary runs all tests. Tests use a minimal custom framework in `tests/Test.h` with `TEST()`, `ASSERT_EQ()`, `ASSERT_TRUE()`, `EXPECT_TRUE()` macros. +The test binary runs all tests, or only those whose names contain one of its arguments (`./build/kte_tests Swap_ Undo_`). Tests use a minimal custom framework in `tests/Test.h` with `TEST()`, `ASSERT_EQ()`, `ASSERT_TRUE()`, `EXPECT_TRUE()` macros. ### Key CMake Options diff --git a/Editor.cc b/Editor.cc index 247e900..ad532ac 100644 --- a/Editor.cc +++ b/Editor.cc @@ -395,6 +395,9 @@ Editor::CancelRecoveryPrompt() pending_recovery_open_ = PendingOpen{}; pending_recovery_swap_path_.clear(); pending_recovery_replay_err_.clear(); + std::string().swap(pending_recovery_content_); + pending_recovery_valid_bytes_ = 0; + pending_recovery_have_content_ = false; } @@ -407,6 +410,9 @@ Editor::ResolveRecoveryPrompt(const bool yes) const PendingOpen req = pending_recovery_open_; const std::string swp = pending_recovery_swap_path_; const std::string rerr_s = pending_recovery_replay_err_; + std::string content = std::move(pending_recovery_content_); + const bool have_content = pending_recovery_have_content_; + std::uint64_t valid_bytes = pending_recovery_valid_bytes_; CancelRecoveryPrompt(); std::string err; @@ -421,11 +427,16 @@ Editor::ResolveRecoveryPrompt(const bool yes) SetStatus("Recovery failed: no buffer"); return false; } - std::string rerr; - std::uint64_t valid_bytes = 0; - if (!kte::SwapManager::ReplayFile(*b, swp, rerr, &valid_bytes)) { - SetStatus("Swap recovery failed: " + rerr); - return false; + if (have_content) { + // The prompt already replayed the journal; install that. + b->replace_all_bytes(content); + std::string().swap(content); + } else { + std::string rerr; + if (!kte::SwapManager::ReplayFile(*b, swp, rerr, &valid_bytes)) { + SetStatus("Swap recovery failed: " + rerr); + return false; + } } // This session keeps appending to the same journal. Drop a torn // final record first, or new records would land behind it and the @@ -581,8 +592,9 @@ Editor::process_pending_opens_() if (tmp.OpenFromFile(req.path, oerr)) { const std::string orig = buffer_bytes_via_views(tmp); std::string rerr; - if (kte::SwapManager::ReplayFile(tmp, swp, rerr)) { - const std::string rec = buffer_bytes_via_views(tmp); + std::uint64_t valid = 0; + if (kte::SwapManager::ReplayFile(tmp, swp, rerr, &valid)) { + std::string rec = buffer_bytes_via_views(tmp); if (rec == orig) { // Nothing to recover. Remove the journal rather than // appending this session's records to it (it may end in a @@ -592,6 +604,9 @@ Editor::process_pending_opens_() pending_recovery_prompt_ = RecoveryPromptKind::RecoverOrDiscard; pending_recovery_open_ = req; pending_recovery_swap_path_ = swp; + pending_recovery_content_ = std::move(rec); + pending_recovery_valid_bytes_ = valid; + pending_recovery_have_content_ = true; StartPrompt(PromptKind::Confirm, "Recover", ""); SetStatus("Recover swap edits for " + req.path + "? (y/n, C-g cancel)"); return opened_any; diff --git a/Editor.h b/Editor.h index 86767ba..3f70dc3 100644 --- a/Editor.h +++ b/Editor.h @@ -717,6 +717,11 @@ private: PendingOpen pending_recovery_open_{}; std::string pending_recovery_swap_path_; std::string pending_recovery_replay_err_; + // The recovered content the prompt offers, and how much of the journal + // produced it: "y" installs it instead of replaying the journal again. + std::string pending_recovery_content_; + std::uint64_t pending_recovery_valid_bytes_ = 0; + bool pending_recovery_have_content_ = false; // GUI-only state (safe no-op in terminal builds) bool file_picker_visible_ = false; diff --git a/Swap.cc b/Swap.cc index 34faed8..1af3386 100644 --- a/Swap.cc +++ b/Swap.cc @@ -27,13 +27,20 @@ namespace { 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; +// Base files up to this size get their CRC computed on open/save; larger ones +// on the writer thread (see compute_base). Adjustable for tests. +std::uint64_t g_sync_crc_limit = std::uint64_t{64} << 20; +// Checkpoints of buffers larger than this are spaced by buffer size. +constexpr std::size_t kLargeCheckpointBytes = std::size_t{16} << 20; } // 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; +// Version 2 adds chunked checkpoints (CHKPT_BEGIN/DATA/END); version 1 +// journals (which never contain them) are still read. +constexpr std::uint32_t VERSION = 2; static std::string @@ -866,15 +873,85 @@ SwapManager::close_ctx(JournalCtx &ctx) } +void +SwapManager::frame_pending(const Pending &p, RecordFrames &f) +{ + // One record: [type u8][len u24][prefix][data][crc32 u32], CRC over all + // but itself. + auto frame = [&f](SwapRecType type, std::vector prefix, const char *data, std::size_t n) { + std::vector head(4); + head[0] = static_cast(type); + put_u24_le(head.data() + 1, static_cast(prefix.size() + n)); + std::uint32_t c = crc32(head.data(), head.size(), 0); + if (!prefix.empty()) + c = crc32(prefix.data(), prefix.size(), c); + if (n > 0) + c = crc32(reinterpret_cast(data), n, c); + head.insert(head.end(), prefix.begin(), prefix.end()); + f.owned.push_back(std::move(head)); + f.segs.emplace_back(f.owned.back().data(), f.owned.back().size()); + f.total += f.owned.back().size(); + if (n > 0) { + f.segs.emplace_back(reinterpret_cast(data), n); + f.total += n; + } + std::vector crc(4); + crc[0] = static_cast(c & 0xFFu); + crc[1] = static_cast((c >> 8) & 0xFFu); + crc[2] = static_cast((c >> 16) & 0xFFu); + crc[3] = static_cast((c >> 24) & 0xFFu); + f.owned.push_back(std::move(crc)); + f.segs.emplace_back(f.owned.back().data(), f.owned.back().size()); + f.total += 4; + }; + + if (p.type != SwapRecType::CHKPT) { + frame(p.type, p.payload, nullptr, 0); + return; + } + const std::string &bytes = p.chkpt; + if (bytes.size() <= kMaxRecordPayload - 5) { + // v1 checkpoint: [encver u8=1][nbytes u32][bytes] + std::vector prefix{1}; + put_le32(prefix, static_cast(bytes.size())); + frame(SwapRecType::CHKPT, std::move(prefix), bytes.data(), bytes.size()); + return; + } + // Chunked: BEGIN, DATA..., END (see SwapRecType). + std::vector begin{1}; + std::uint8_t total[8]; + put_le64(total, static_cast(bytes.size())); + begin.insert(begin.end(), total, total + 8); + put_le32(begin, crc32(reinterpret_cast(bytes.data()), bytes.size(), 0)); + frame(SwapRecType::CHKPT_BEGIN, std::move(begin), nullptr, 0); + for (std::size_t off = 0; off < bytes.size(); off += kMaxRecordPayload) { + const std::size_t n = std::min(kMaxRecordPayload, bytes.size() - off); + frame(SwapRecType::CHKPT_DATA, {}, bytes.data() + off, n); + } + frame(SwapRecType::CHKPT_END, std::vector{1}, nullptr, 0); +} + + bool -SwapManager::compact_to_checkpoint(JournalCtx &ctx, const std::vector &chkpt_record, std::string &err) +SwapManager::write_frames(int fd, const RecordFrames &f) +{ + for (const auto &[data, n]: f.segs) { + if (!write_full(fd, data, n)) + return false; + } + return true; +} + + +bool +SwapManager::compact_to_checkpoint(JournalCtx &ctx, const RecordFrames &chkpt_record, std::string &err) { err.clear(); if (ctx.path.empty()) { err = "Compact failed: empty path"; return false; } - if (chkpt_record.empty()) { + if (chkpt_record.total == 0) { err = "Compact failed: empty checkpoint record"; return false; } @@ -912,7 +989,7 @@ SwapManager::compact_to_checkpoint(JournalCtx &ctx, const std::vector(chkpt_record.size()); + ctx.approx_size_bytes = 64 + static_cast(chkpt_record.total); return true; } @@ -1168,11 +1245,20 @@ SwapManager::maybe_request_checkpoint(Buffer &buf, const std::size_t approx_edit 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); + // A checkpoint writes the whole buffer. For a large one, checkpoint + // after edits amounting to half its size (and on the timer only + // once edits reach a sixteenth), so the bytes written stay in + // proportion to the bytes edited; retry a gap less often too. + const std::size_t size = buf.ContentBytes(); + const bool large = size > kLargeCheckpointBytes; + const std::size_t bytes_limit = std::max(cfg.checkpoint_bytes, size / 2); + const bool bytes_hit = (cfg.checkpoint_bytes > 0) && ctx.edit_bytes_since_chkpt >= bytes_limit; 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; + (((now - ctx.last_chkpt_ns) / 1000000ULL) >= cfg.checkpoint_interval_ms) && + (!large || ctx.edit_bytes_since_chkpt >= size / 16); + const std::uint64_t gap_interval = kGapCheckpointIntervalNs * + (1 + size / kLargeCheckpointBytes); + const bool gap_hit = ctx.gap && (now - ctx.gap_chkpt_request_ns) >= gap_interval; if (gap_hit) ctx.gap_chkpt_request_ns = now; if (bytes_hit || time_hit || gap_hit) { @@ -1197,7 +1283,9 @@ SwapManager::RetryGapCheckpoints() 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) + const std::uint64_t interval = kGapCheckpointIntervalNs * + (1 + b->ContentBytes() / kLargeCheckpointBytes); + if (now - ctx.gap_chkpt_request_ns < interval) continue; ctx.gap_chkpt_request_ns = now; due.push_back(b); @@ -1238,41 +1326,12 @@ SwapManager::RecordCheckpoint(Buffer &buf, const bool urgent_flush) 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 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; - + // Any size: the writer frames large checkpoints as several records. 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(bytes.size())); - p.payload.insert(p.payload.end(), reinterpret_cast(bytes.data()), - reinterpret_cast(bytes.data()) + bytes.size()); + p.chkpt = snapshot_buffer_bytes(buf); enqueue(std::move(p)); } @@ -1435,6 +1494,8 @@ SwapManager::process_one(const Pending &p) } if (!ctxp) return; + if (ctxp->fd < 0) + complete_base_crc(*ctxp); std::string open_err; bool locked_out = false; if (!open_ctx(*ctxp, path, open_err, &locked_out)) { @@ -1473,32 +1534,26 @@ SwapManager::process_one(const Pending &p) return; } - // Build record: [type u8][len u24][payload][crc32 u32] - std::uint8_t len3[3]; - put_u24_le(len3, static_cast(p.payload.size())); + RecordFrames rec; + frame_pending(p, rec); - std::uint8_t head[4]; - head[0] = static_cast(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(c & 0xFFu); - crcbytes[1] = static_cast((c >> 8) & 0xFFu); - crcbytes[2] = static_cast((c >> 16) & 0xFFu); - crcbytes[3] = static_cast((c >> 24) & 0xFFu); - - std::vector 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)); + // A checkpoint that would trigger compaction goes straight into the + // compacted journal instead of being appended and then copied there. + // For a large buffer the thresholds scale with its size, so the cost + // of compacting stays proportional to what was journaled since. + const std::uint64_t compact_at = std::max(compact_bytes, 2 * p.chkpt.size()); + if (p.type == SwapRecType::CHKPT && compact_bytes > 0 && + ctxp->approx_size_bytes + rec.total >= compact_at) { + std::string compact_err; + if (compact_to_checkpoint(*ctxp, rec, compact_err)) { + std::lock_guard lg(mtx_); + ctxp->gap = false; + ctxp->gap_notified = false; + circuit_breaker_.RecordSuccess(); + return; + } + report_error(compact_err, p.buf); // fall back to appending + } // Remember where this record starts so a partial write can be undone; // otherwise later records would be appended behind the torn bytes. @@ -1506,7 +1561,7 @@ SwapManager::process_one(const Pending &p) 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()); + bool ok = write_frames(ctxp->fd, rec); if (!ok) { int err = errno; report_error("Failed to write swap record to '" + path + "': " + std::strerror(err), p.buf); @@ -1524,7 +1579,7 @@ SwapManager::process_one(const Pending &p) ctxp->gap = false; ctxp->gap_notified = false; } - ctxp->approx_size_bytes += static_cast(rec.size()); + ctxp->approx_size_bytes += static_cast(rec.total); if (p.urgent_flush) { if (kte::syscall::Fsync(ctxp->fd) != 0) { int err = errno; @@ -1532,14 +1587,6 @@ SwapManager::process_one(const Pending &p) } ctxp->last_fsync_ns = now_ns(); } - if (p.type == SwapRecType::CHKPT && compact_bytes > 0 && - ctxp->approx_size_bytes >= static_cast(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 { @@ -1675,15 +1722,58 @@ SwapManager::file_crc32(const std::string &path, std::uint32_t &out) } +#if defined(KTE_TESTS) +void +SwapManager::SetSyncCrcLimitForTests(std::uint64_t bytes) +{ + g_sync_crc_limit = bytes; +} +#endif + + +void +SwapManager::complete_base_crc(JournalCtx &ctx) +{ + std::string file; + std::uint64_t size = 0; + std::int64_t mtime = 0; + { + std::lock_guard lg(mtx_); + if (!ctx.has_base || ctx.has_base_crc || ctx.base_crc_file.empty()) + return; + file = ctx.base_crc_file; + size = ctx.base_size; + mtime = ctx.base_mtime_ns; + } + // Off the lock: this reads the whole file. Keep the CRC only if the file + // is still the one whose size and mtime the journal records. + std::uint32_t crc = 0; + std::uint64_t size2 = 0; + std::int64_t mtime2 = 0; + const bool ok = file_crc32(file, crc) && stat_base(file, size2, mtime2) && size2 == size && mtime2 == mtime; + std::lock_guard lg(mtx_); + if (ctx.base_crc_file != file || ctx.base_size != size || ctx.base_mtime_ns != mtime) + return; // the base changed meanwhile (save, rename) + ctx.base_crc_file.clear(); + if (ok) { + ctx.has_base_crc = true; + ctx.base_crc = crc; + } +} + + 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; + // Reading a very large file again on every open and save is too slow + // for the main thread: its CRC is left to the writer thread, which + // computes it before creating the journal (see process_one). + const std::uint64_t kMaxCrcBytes = g_sync_crc_limit; 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); + if (id.has && !id.has_crc && id.size > kMaxCrcBytes) + id.crc_file = file; return id; } @@ -1696,6 +1786,7 @@ SwapManager::apply_base(JournalCtx &ctx, const BaseId &id) ctx.base_mtime_ns = id.mtime_ns; ctx.has_base_crc = id.has_crc; ctx.base_crc = id.crc; + ctx.base_crc_file = id.crc_file; } @@ -1720,7 +1811,7 @@ SwapManager::ReplayFile(Buffer &buf, const std::string &swap_path, std::string & return false; } const std::uint32_t ver = read_le32(hdr + 8); - if (ver != VERSION) { + if (ver != 1 && ver != VERSION) { err = "Unsupported swap version: " + std::to_string(ver); return false; } @@ -1739,15 +1830,27 @@ SwapManager::ReplayFile(Buffer &buf, const std::string &swap_path, std::string & } } restore{buf, prev_rec}; + // A chunked checkpoint being assembled (see SwapRecType::CHKPT_BEGIN). + bool in_chkpt = false; + std::uint64_t chkpt_size = 0; + std::uint32_t chkpt_crc = 0; + std::string chkpt; + for (;;) { - // Everything before this offset replayed cleanly. + // Everything before this offset replayed cleanly. Inside a chunked + // checkpoint, that is where the checkpoint began (it only counts + // once complete). const std::streamoff record_start = in.tellg(); - if (valid_bytes && record_start >= 0) + if (valid_bytes && record_start >= 0 && !in_chkpt) *valid_bytes = static_cast(record_start); std::uint8_t head[4]; in.read(reinterpret_cast(head), sizeof(head)); const std::size_t got_head = static_cast(in.gcount()); if (got_head == 0 && in.eof()) { + if (in_chkpt) { + err = "Swap file ends inside a checkpoint; recovered the records before it: " + swap_path; + kte::ErrorHandler::Instance().Warning("SwapManager", err, swap_path); + } return true; // clean EOF } // A record cut short by end-of-file is the expected result of a crash or @@ -1786,8 +1889,52 @@ SwapManager::ReplayFile(Buffer &buf, const std::string &swap_path, std::string & return false; } + // Chunked checkpoint records; nothing else may come between them. + if (in_chkpt && type != SwapRecType::CHKPT_DATA && type != SwapRecType::CHKPT_END) { + err = "Swap file record inside a checkpoint: " + swap_path; + return false; + } + // Apply record switch (type) { + case SwapRecType::CHKPT_BEGIN: { + // [encver u8=1][total u64][crc32 u32] + if (payload.size() < 13 || payload[0] != 1) { + err = "Malformed CHKPT_BEGIN payload"; + return false; + } + chkpt_size = 0; + for (int i = 7; i >= 0; --i) + chkpt_size = (chkpt_size << 8) | payload[1 + static_cast(i)]; + chkpt_crc = read_le32(payload.data() + 9); + chkpt.clear(); + try { + chkpt.reserve(static_cast(chkpt_size)); + } catch (...) { + err = "Checkpoint too large to replay: " + std::to_string(chkpt_size) + " bytes"; + return false; + } + in_chkpt = true; + break; + } + case SwapRecType::CHKPT_DATA: + if (!in_chkpt || chkpt.size() + payload.size() > chkpt_size) { + err = "Swap file checkpoint data out of place: " + swap_path; + return false; + } + chkpt.append(reinterpret_cast(payload.data()), payload.size()); + break; + case SwapRecType::CHKPT_END: { + if (!in_chkpt || chkpt.size() != chkpt_size || + crc32(reinterpret_cast(chkpt.data()), chkpt.size(), 0) != chkpt_crc) { + err = "Swap file checkpoint incomplete or corrupt: " + swap_path; + return false; + } + buf.replace_all_bytes(chkpt); + std::string().swap(chkpt); + in_chkpt = false; + break; + } case SwapRecType::INS: { std::size_t off = 0; // INS payload: encver(1) + row(4) + col(4) + nbytes(4) + data(nbytes) diff --git a/Swap.h b/Swap.h index 6969f4e..1880962 100644 --- a/Swap.h +++ b/Swap.h @@ -27,6 +27,13 @@ enum class SwapRecType : std::uint8_t { SPLIT = 3, JOIN = 4, META = 0xF0, + // A checkpoint too large for one record (a record's length field is 24 + // bits) is written as BEGIN [encver u8][total u64][crc32 u32], DATA + // chunks (raw bytes), END [encver u8]. Replay applies it only at END, + // with the size and CRC-32 matching; an unfinished one is ignored. + CHKPT_BEGIN = 0xFB, + CHKPT_DATA = 0xFC, + CHKPT_END = 0xFD, CHKPT = 0xFE, }; @@ -149,6 +156,9 @@ public: // Test-only hook to keep swap path logic centralized. // (Avoid duplicating naming rules in unit tests.) #ifdef KTE_TESTS + // Base files larger than this get their CRC on the writer thread. + static void SetSyncCrcLimitForTests(std::uint64_t bytes); + static std::string ComputeSwapPathForTests(const Buffer &buf) { return ComputeSidecarPath(buf); @@ -239,16 +249,33 @@ private: std::int64_t base_mtime_ns{0}; bool has_base_crc{false}; // CRC-32 of the base file's content std::uint32_t base_crc{0}; + // A large base file's CRC is computed by the writer thread before it + // creates the journal (see process_one), not on open/save. + std::string base_crc_file; }; struct Pending { Buffer *buf{nullptr}; SwapRecType type{SwapRecType::INS}; std::vector payload; // framed payload only + std::string chkpt; // CHKPT: the buffer's content (framed by the writer) bool urgent_flush{false}; std::uint64_t seq{0}; }; + // A record (or, for a large checkpoint, several) as segments to write: + // small framing buffers owned here, and a reference to the checkpoint + // content (not copied). + struct RecordFrames { + std::deque > owned; + std::vector > segs; + std::size_t total = 0; + }; + + static void frame_pending(const Pending &p, RecordFrames &f); + + static bool write_frames(int fd, const RecordFrames &f); + // Helpers static std::string ComputeSidecarPath(const Buffer &buf); @@ -266,7 +293,7 @@ private: static void close_ctx(JournalCtx &ctx); - static bool compact_to_checkpoint(JournalCtx &ctx, const std::vector &chkpt_record, + static bool compact_to_checkpoint(JournalCtx &ctx, const RecordFrames &chkpt_record, std::string &err); static std::uint32_t crc32(const std::uint8_t *data, std::size_t len, std::uint32_t seed = 0); @@ -283,10 +310,14 @@ private: std::int64_t mtime_ns{0}; bool has_crc{false}; std::uint32_t crc{0}; + std::string crc_file; // set when the CRC is deferred to the writer thread }; static BaseId compute_base(const std::string &file); + // Writer thread: compute a deferred base CRC (see BaseId::crc_file). + void complete_base_crc(JournalCtx &ctx); + static void apply_base(JournalCtx &ctx, const BaseId &id); static void put_le32(std::vector &out, std::uint32_t v); diff --git a/docs/audits/claude-findings-20260923.md b/docs/audits/claude-findings-20260923.md index 283a771..483d066 100644 --- a/docs/audits/claude-findings-20260923.md +++ b/docs/audits/claude-findings-20260923.md @@ -526,6 +526,24 @@ before the fix (unless noted). through matches in the regex replace prompt; Up/Down in a regex prompt use the regex. +**Follow-up: journals for large buffers** +- Buffers over 16 MiB are checkpointed as CHKPT_BEGIN (size, CRC-32), + data chunks and CHKPT_END; replay applies the checkpoint only when it + is complete and its CRC matches, and treats an unfinished one as a + torn tail. Such buffers could not be checkpointed at all: a lost + record left the journal permanently behind and it was never + compacted. The journal header is now version 2 (version 1 journals + are still read; older kte builds will not read version 2). +- Checkpoint thresholds scale with the buffer: after edits amounting to + half its size, on the timer only after a sixteenth, compaction at + twice its size, gap retries spaced by size. A checkpoint that would + compact is written straight into the compacted journal. +- The recovery prompt keeps the content it replayed for the preview; + answering y installs it instead of replaying the journal again. +- The CRC of a base file over 64 MiB is computed by the journal writer + thread (it was skipped), so a touched but unchanged large file no + longer makes its journal unusable for recovery. + **Known limitations (not fixed)** - Catastrophic regex backtracking (e.g. `(a*)*b`) can still take very long; std::regex has no time limit. @@ -539,6 +557,3 @@ before the fix (unless noted). session that is locked out is told at its first edit instead. - A regex that matches rarely or not at all still scans the whole buffer with std::regex (1.6 s on 105 MB). -- A journal whose buffer exceeds 16 MiB cannot be checkpointed; after a - lost record such a journal stays incomplete until the file is saved - (reported to the user). diff --git a/tests/TestRunner.cc b/tests/TestRunner.cc index b2eb9b7..bfae86d 100644 --- a/tests/TestRunner.cc +++ b/tests/TestRunner.cc @@ -1,17 +1,32 @@ #include "Test.h" #include #include +#include +#include +// Usage: kte_tests [substring...] runs the tests whose names contain any of +// the substrings (all tests when none are given). int -main() +main(int argc, char **argv) { using namespace std::chrono; auto ® = ktet::registry(); + std::vector filters(argv + 1, argv + argc); + auto selected = [&filters](const std::string &name) { + if (filters.empty()) + return true; + for (const auto &f: filters) + if (name.find(f) != std::string::npos) + return true; + return false; + }; std::cout << "kte unit tests: " << reg.size() << " test(s)\n"; int failed = 0; auto t0 = steady_clock::now(); for (const auto &tc: reg) { + if (!selected(tc.name)) + continue; auto ts = steady_clock::now(); try { tc.fn(); diff --git a/tests/test_audit_regressions.cc b/tests/test_audit_regressions.cc index 88a90a1..f6c088c 100644 --- a/tests/test_audit_regressions.cc +++ b/tests/test_audit_regressions.cc @@ -18,6 +18,7 @@ #include #include #include +#include #include #include #include @@ -2012,3 +2013,169 @@ TEST(Search_PrevAcrossBlocks) } ASSERT_TRUE(h.Exec(CommandId::Refresh)); } + + +namespace { +// A 20 MiB buffer: over the 16 MiB a single checkpoint record can hold. +std::string +big_text() +{ + std::string t; + t.reserve(std::size_t{21} << 20); + for (int i = 0; t.size() < (std::size_t{20} << 20); ++i) + t += "line " + std::to_string(i) + " of a large buffer\n"; + return t; +} +} // namespace + + +// Buffers over 16 MiB are checkpointed as BEGIN/DATA.../END records, which +// replay to the buffer; edits after the checkpoint apply on top. +TEST(Swap_LargeCheckpointReplays) +{ + TempDir d("big_chkpt"); + const std::string path = (d.path / "big.txt").string(); + const std::string text = big_text(); + { + std::ofstream o(path, std::ios::binary); + o << text; + } + Buffer b; + std::string err; + ASSERT_TRUE(b.OpenFromFile(path, err)); + const std::string swp = kte::SwapManager::ComputeSwapPathForTests(b); + std::remove(swp.c_str()); + kte::SwapManager sm; + sm.Attach(&b); + b.SetSwapRecorder(sm.RecorderFor(&b)); + b.insert_text(0, 0, "A"); + sm.Checkpoint(&b); + b.insert_text(1, 0, "B"); + sm.Flush(&b); + const std::string want = b.BytesForTests(); + const std::string copy = (d.path / "copy.swp").string(); + { + std::ofstream o(copy, std::ios::binary | std::ios::trunc); + o << slurp(swp); + } + // The journal holds the chunked checkpoint. + ASSERT_TRUE(std::filesystem::file_size(copy) > text.size()); + Buffer r; + ASSERT_TRUE(r.OpenFromFile(path, err)); + std::uint64_t valid = 0; + ASSERT_TRUE(kte::SwapManager::ReplayFile(r, copy, err, &valid)); + ASSERT_TRUE(err.empty()); + ASSERT_TRUE(r.BytesForTests() == want); + ASSERT_EQ(valid, (std::uint64_t) std::filesystem::file_size(copy)); + + // Torn inside the checkpoint's data: the records before it replay, and + // the checkpoint is not counted as valid. + std::filesystem::resize_file(copy, 64 + 1000 + (std::size_t{8} << 20)); + Buffer t; + ASSERT_TRUE(t.OpenFromFile(path, err)); + valid = 0; + ASSERT_TRUE(kte::SwapManager::ReplayFile(t, copy, err, &valid)); + ASSERT_TRUE(!err.empty()); + ASSERT_EQ(t.BytesForTests(), "A" + text); + ASSERT_TRUE(valid < 64 + 1000); + + b.SetSwapRecorder(nullptr); + sm.Detach(&b, true); +} + + +// A lost record on a large buffer is closed by a checkpoint (it could not +// be before: the buffer was too large to checkpoint), and compaction +// rewrites the journal to that checkpoint alone. +TEST(Swap_LargeBufferGapResyncsAndCompacts) +{ + TempDir d("big_gap"); + const std::string path = (d.path / "big.txt").string(); + const std::string text = big_text(); + { + std::ofstream o(path, std::ios::binary); + o << text; + } + Buffer b; + std::string err; + ASSERT_TRUE(b.OpenFromFile(path, err)); + const std::string swp = kte::SwapManager::ComputeSwapPathForTests(b); + std::filesystem::remove_all(swp); + kte::SwapManager sm; + kte::SwapConfig cfg; + cfg.compact_bytes = 1024; + sm.SetConfig(cfg); + sm.Attach(&b); + b.SetSwapRecorder(sm.RecorderFor(&b)); + // The journal cannot be opened: the record is lost. + std::filesystem::create_directories(swp); + b.insert_text(0, 0, "X"); + sm.Flush(&b); + std::filesystem::remove_all(swp); + std::this_thread::sleep_for(std::chrono::milliseconds(2100)); // gap retry interval for 20 MiB + sm.RetryGapCheckpoints(); + sm.Flush(&b); + b.insert_text(0, 1, "Y"); + sm.Flush(&b); + const std::string copy = (d.path / "copy.swp").string(); + { + std::ofstream o(copy, std::ios::binary | std::ios::trunc); + o << slurp(swp); + } + Buffer r; + ASSERT_TRUE(r.OpenFromFile(path, err)); + ASSERT_TRUE(kte::SwapManager::ReplayFile(r, copy, err)); + ASSERT_TRUE(r.BytesForTests() == b.BytesForTests()); + // Compacted: header, the checkpoint (a few framing bytes over the + // content) and the one INS record after it. + ASSERT_TRUE(std::filesystem::file_size(copy) < text.size() + 4096); + b.SetSwapRecorder(nullptr); + sm.Detach(&b, true); +} + + +// A large base file's CRC is computed by the journal writer, not on open; +// the journal still records it, so a same-size, same-mtime change on disk is +// caught. +TEST(Swap_DeferredBaseCrc) +{ + TempDir d("deferred_crc"); + const std::string path = (d.path / "f.txt").string(); + const std::string text(20000, 'a'); + { + std::ofstream o(path, std::ios::binary); + o << text; + } + kte::SwapManager::SetSyncCrcLimitForTests(1024); + Buffer b; + std::string err; + ASSERT_TRUE(b.OpenFromFile(path, err)); + const std::string swp = kte::SwapManager::ComputeSwapPathForTests(b); + std::remove(swp.c_str()); + kte::SwapManager sm; + sm.Attach(&b); + b.SetSwapRecorder(sm.RecorderFor(&b)); + b.insert_text(0, 0, "X"); + sm.Flush(&b); + kte::SwapManager::SetSyncCrcLimitForTests(std::uint64_t{64} << 20); + const std::string j = slurp(swp); + ASSERT_TRUE(j.size() >= 64); + ASSERT_TRUE((static_cast(j[12]) & 2) != 0); // CRC recorded + ASSERT_TRUE(kte::SwapManager::JournalMatchesFile(swp, path)); + // Touched (new mtime, same content): the CRC shows it is still the same + // file, so recovery can use the journal. Without a CRC (large files + // before this), any touch made the journal unusable. + struct timespec later[2]; + later[0].tv_sec = later[1].tv_sec = ::time(nullptr) + 100; + later[0].tv_nsec = later[1].tv_nsec = 0; + ASSERT_TRUE(::utimensat(AT_FDCWD, path.c_str(), later, 0) == 0); + ASSERT_TRUE(kte::SwapManager::JournalMatchesFile(swp, path)); + // Changed content with a new mtime: stale. + { + std::ofstream o(path, std::ios::binary | std::ios::trunc); + o << std::string(20000, 'b'); + } + ASSERT_TRUE(!kte::SwapManager::JournalMatchesFile(swp, path)); + b.SetSwapRecorder(nullptr); + sm.Detach(&b, true); +} diff --git a/tests/test_swap_writer.cc b/tests/test_swap_writer.cc index 4697214..8e63315 100644 --- a/tests/test_swap_writer.cc +++ b/tests/test_swap_writer.cc @@ -123,7 +123,7 @@ TEST(SwapWriter_Header_Records_And_CRC) static const std::uint8_t magic[8] = {'K', 'T', 'E', '_', 'S', 'W', 'P', '\0'}; for (int i = 0; i < 8; ++i) ASSERT_EQ(bytes[(std::size_t) i], magic[i]); - ASSERT_EQ(read_le32(bytes.data() + 8), (std::uint32_t) 1); + ASSERT_EQ(read_le32(bytes.data() + 8), (std::uint32_t) 2); // v2: chunked checkpoints // flags bit 0: the base file's identity (size, mtime) is recorded, since // the buffer is file-backed; bit 1: its content CRC-32 too. ASSERT_EQ(read_le32(bytes.data() + 12), (std::uint32_t) 3);