From aa53d75f96a4eced7d0b76d18216271af5aacf3e Mon Sep 17 00:00:00 2001 From: Claude Date: Thu, 24 Sep 2026 03:32:59 +0000 Subject: [PATCH] Swap: keep the journal replayable across failures and crashes - Replay recovers every complete record before a torn final record (the normal result of a crash mid-append) and reports it through err, instead of rejecting the whole journal. SwapReplay_TruncatedLog is updated to the new behaviour. - A record that cannot be written (open/write failure, breaker open) marks the journal as having a gap. Later position-based records are dropped until a checkpoint re-establishes the content; recorders request one on the next edit (at most once per second). - A failed append is truncated back off the file, and a failed header write closes the file so the next attempt starts fresh. - Inserts over the 24-bit record limit are split into consecutive records; buffers too large to checkpoint are no longer snapshotted and rejected on every checkpoint. - Declining to delete an unreadable swap moves it to .corrupt so new edits are not appended behind the bad data. - NotifyFilenameChanged no longer writes a full checkpoint into the journal it is about to delete. Audit findings D7, P5. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_015rcVQiiWyTzoNrsvLSaAfB --- Editor.cc | 12 ++- Swap.cc | 144 ++++++++++++++++++++++------- Swap.h | 5 + SyscallWrappers.cc | 11 +++ SyscallWrappers.h | 5 + tests/test_swap_recovery_prompt.cc | 54 +++++++++++ tests/test_swap_replay.cc | 112 +++++++++++++++++++++- 7 files changed, 304 insertions(+), 39 deletions(-) diff --git a/Editor.cc b/Editor.cc index 419b6a4..81b9db3 100644 --- a/Editor.cc +++ b/Editor.cc @@ -386,8 +386,16 @@ Editor::ResolveRecoveryPrompt(const bool yes) return true; } if (kind == RecoveryPromptKind::DeleteCorruptSwap) { + std::string kept_as; if (yes) { (void) std::remove(swp.c_str()); + } else { + // Keep the unreadable journal for inspection, but out of the way: + // left in place, the new session's records would be appended + // behind the corrupt data and could never be replayed. + kept_as = swp + ".corrupt"; + if (std::rename(swp.c_str(), kept_as.c_str()) != 0) + kept_as.clear(); } if (!OpenFile(req.path, err)) { SetStatus(err); @@ -395,7 +403,9 @@ Editor::ResolveRecoveryPrompt(const bool yes) } apply_pending_line(*this, req.line1); // Include a short hint that the swap was corrupt. - if (!rerr_s.empty()) { + if (!kept_as.empty()) { + SetStatus("Opened " + req.path + " (unreadable swap kept as " + kept_as + ")"); + } else if (!rerr_s.empty()) { SetStatus("Opened " + req.path + " (swap unreadable)"); } else { SetStatus("Opened " + req.path); diff --git a/Swap.cc b/Swap.cc index 10abcb9..4b309a7 100644 --- a/Swap.cc +++ b/Swap.cc @@ -20,6 +20,13 @@ 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 + namespace { constexpr std::uint8_t MAGIC[8] = {'K', 'T', 'E', '_', 'S', 'W', 'P', '\0'}; constexpr std::uint32_t VERSION = 1; @@ -489,8 +496,9 @@ SwapManager::PruneSwapDir() void SwapManager::NotifyFilenameChanged(Buffer &buf) { - // Best-effort: checkpoint the old journal before switching paths. - RecordCheckpoint(buf, true); + // 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; { std::lock_guard lg(mtx_); @@ -709,6 +717,11 @@ SwapManager::open_ctx(JournalCtx &ctx, const std::string &path, std::string &err 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 @@ -910,19 +923,37 @@ SwapManager::RecordInsert(Buffer &buf, int row, int col, std::string_view text) if (it == journals_.end() || it->second.suspended) return; } - Pending p; - p.buf = &buf; - p.type = SwapRecType::INS; // payload v1: [encver u8=1][row u32][col u32][nbytes u32][bytes] - if (text.size() > 0xFFFFFFFFu) - return; - p.payload.push_back(1); - put_le32(p.payload, static_cast(std::max(0, row))); - put_le32(p.payload, static_cast(std::max(0, col))); - put_le32(p.payload, static_cast(text.size())); - p.payload.insert(p.payload.end(), reinterpret_cast(text.data()), - reinterpret_cast(text.data()) + text.size()); - enqueue(std::move(p)); + // 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::max(0, row)); + std::size_t c = static_cast(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(r)); + put_le32(p.payload, static_cast(c)); + put_le32(p.payload, static_cast(chunk.size())); + p.payload.insert(p.payload.end(), reinterpret_cast(chunk.data()), + reinterpret_cast(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::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()); } @@ -1000,11 +1031,11 @@ SwapManager::maybe_request_checkpoint(Buffer &buf, const std::size_t approx_edit { std::lock_guard lg(mtx_); cfg = cfg_; - if (cfg.checkpoint_bytes == 0 && cfg.checkpoint_interval_ms == 0) - return; 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(); @@ -1014,7 +1045,10 @@ SwapManager::maybe_request_checkpoint(Buffer &buf, const std::size_t approx_edit 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); - if (bytes_hit || time_hit) { + 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; @@ -1036,8 +1070,15 @@ 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) + return; const std::string bytes = snapshot_buffer_bytes(buf); - if (bytes.size() > 0xFFFFFFFFu) + if (bytes.size() > kMaxRecordPayload - 5) return; Pending p; @@ -1128,6 +1169,14 @@ 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 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; { @@ -1153,6 +1202,7 @@ SwapManager::process_one(const Pending &p) ? p.buf->Filename() : ""); } + mark_gap(); return; } @@ -1167,6 +1217,9 @@ SwapManager::process_one(const Pending &p) auto it = journals_.find(p.buf); if (it == journals_.end()) 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; @@ -1182,14 +1235,14 @@ SwapManager::process_one(const Pending &p) std::lock_guard lg(mtx_); circuit_breaker_.RecordFailure(); } + mark_gap(); return; } - if (p.payload.size() > 0xFFFFFFu) { + 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); - { - std::lock_guard lg(mtx_); - circuit_breaker_.RecordFailure(); - } + mark_gap(); return; } @@ -1220,17 +1273,29 @@ SwapManager::process_one(const Pending &p) 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 lg(mtx_); circuit_breaker_.RecordFailure(); } + mark_gap(); return; } + if (p.type == SwapRecType::CHKPT) { + std::lock_guard lg(mtx_); + ctxp->gap = false; + } ctxp->approx_size_bytes += static_cast(rec.size()); if (p.urgent_flush) { if (kte::syscall::Fsync(ctxp->fd) != 0) { @@ -1259,12 +1324,14 @@ SwapManager::process_one(const Pending &p) std::lock_guard lg(mtx_); circuit_breaker_.RecordFailure(); } + mark_gap(); } catch (...) { report_error("Unknown exception in process_one", p.buf); { std::lock_guard lg(mtx_); circuit_breaker_.RecordFailure(); } + mark_gap(); } } @@ -1343,25 +1410,32 @@ SwapManager::ReplayFile(Buffer &buf, const std::string &swap_path, std::string & if (got_head == 0 && in.eof()) { return true; // clean EOF } - if (got_head != sizeof(head)) { - err = "Swap file truncated (record header): " + swap_path; - return false; - } + // 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(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 payload; payload.resize(len); - if (len > 0 && !read_exact(in, payload.data(), len)) { - err = "Swap file truncated (payload): " + swap_path; - return false; - } + 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))) { - err = "Swap file truncated (crc): " + swap_path; - return false; - } + 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); diff --git a/Swap.h b/Swap.h index 645d651..4d7fdac 100644 --- a/Swap.h +++ b/Swap.h @@ -192,6 +192,11 @@ private: std::uint64_t last_chkpt_ns{0}; std::uint64_t edit_bytes_since_chkpt{0}; std::uint64_t approx_size_bytes{0}; + // A record was lost (write failure, breaker open, oversize). Records are + // position-based, so later ones no longer apply to the journal's state: + // drop them until a checkpoint re-establishes the full content. + bool gap{false}; + std::uint64_t gap_chkpt_request_ns{0}; }; struct Pending { diff --git a/SyscallWrappers.cc b/SyscallWrappers.cc index 8b1a528..189ff07 100644 --- a/SyscallWrappers.cc +++ b/SyscallWrappers.cc @@ -52,6 +52,17 @@ Fstat(int fd, struct stat *buf) } +int +Ftruncate(int fd, off_t length) +{ + int ret; + do { + ret = ::ftruncate(fd, length); + } while (ret == -1 && errno == EINTR); + return ret; +} + + int Fchmod(int fd, mode_t mode) { diff --git a/SyscallWrappers.h b/SyscallWrappers.h index 604b0c5..6e59b3b 100644 --- a/SyscallWrappers.h +++ b/SyscallWrappers.h @@ -34,6 +34,11 @@ int Fstat(int fd, struct stat *buf); // Automatically retries on EINTR. int Fchmod(int fd, mode_t mode); +// EINTR-safe wrapper for ftruncate(2). +// Returns 0 on success, -1 on failure (errno set). +// Automatically retries on EINTR. +int Ftruncate(int fd, off_t length); + // EINTR-safe wrapper for mkstemp(3). // Returns file descriptor on success, -1 on failure (errno set). // Automatically retries on EINTR. diff --git a/tests/test_swap_recovery_prompt.cc b/tests/test_swap_recovery_prompt.cc index ba2d269..c0fd4f7 100644 --- a/tests/test_swap_recovery_prompt.cc +++ b/tests/test_swap_recovery_prompt.cc @@ -278,3 +278,57 @@ TEST(SwapRecoveryPrompt_CorruptSwap_OffersDelete) std::remove(file_path.c_str()); std::filesystem::remove_all(xdg_root); } + + +// Declining to delete an unreadable swap keeps it aside as ".corrupt" so the +// new session journals into a fresh file instead of behind the bad data. +TEST(SwapRecoveryPrompt_CorruptSwap_Decline_KeepsAsideAndStartsFresh) +{ + ktet::InstallDefaultCommandsOnce(); + + const std::filesystem::path xdg_root = std::filesystem::temp_directory_path() / + (std::string("kte_ut_xdg_state_corrupt_keep_") + + std::to_string((int) ::getpid())); + std::filesystem::remove_all(xdg_root); + std::filesystem::create_directories(xdg_root); + const ScopedXdgStateHome scoped(xdg_root.string()); + + const std::filesystem::path work = xdg_root / "work"; + std::filesystem::create_directories(work); + const std::string file_path = (work / "corrupt_keep.txt").string(); + write_file_bytes(file_path, "base\n"); + + Buffer b; + std::string err; + ASSERT_TRUE(b.OpenFromFile(file_path, err)); + const std::string swap_path = kte::SwapManager::ComputeSwapPathForTests(b); + std::filesystem::create_directories(std::filesystem::path(swap_path).parent_path()); + write_file_bytes(swap_path, std::string(80, 'z')); // long enough to be "trusted", bad magic + + Editor ed; + ed.SetDimensions(24, 80); + ed.AddBuffer(Buffer()); + ed.RequestOpenFile(b.Filename()); + ASSERT_EQ(ed.ProcessPendingOpens(), false); + ASSERT_EQ(ed.PendingRecoveryPrompt(), Editor::RecoveryPromptKind::DeleteCorruptSwap); + + ASSERT_TRUE(Execute(ed, CommandId::InsertText, "n")); + ASSERT_TRUE(Execute(ed, CommandId::Newline)); + ASSERT_EQ(ed.PendingRecoveryPrompt(), Editor::RecoveryPromptKind::None); + ASSERT_TRUE(std::filesystem::exists(swap_path + ".corrupt")); + + // New edits journal into a fresh, replayable file. + Buffer *cur = ed.CurrentBuffer(); + ASSERT_TRUE(cur != nullptr); + cur->insert_text(0, 0, std::string("X")); + ed.Swap()->Flush(cur); + + Buffer check; + ASSERT_TRUE(check.OpenFromFile(file_path, err)); + std::string rerr; + ASSERT_TRUE(kte::SwapManager::ReplayFile(check, swap_path, rerr)); + ASSERT_EQ(check.GetLineString(0), std::string("Xbase")); + + std::remove(file_path.c_str()); + std::filesystem::remove_all(xdg_root); +} diff --git a/tests/test_swap_replay.cc b/tests/test_swap_replay.cc index 3f617fe..cdf5d41 100644 --- a/tests/test_swap_replay.cc +++ b/tests/test_swap_replay.cc @@ -6,6 +6,7 @@ #include #include #include +#include #include #include @@ -103,7 +104,10 @@ TEST(SwapReplay_RecordFlushReopenReplay_ExactBytesMatch) } -TEST(SwapReplay_TruncatedLog_FailsSafely) +// A crash mid-append leaves a torn final record. Replay keeps every complete +// record before it (that tail is exactly what the journal exists to save), +// succeeds, and reports the incomplete record through err. +TEST(SwapReplay_TruncatedLog_RecoversPrefix) { const std::string path = "./.kte_ut_swap_replay_2.txt"; std::remove(path.c_str()); @@ -113,10 +117,13 @@ TEST(SwapReplay_TruncatedLog_FailsSafely) std::string err; ASSERT_TRUE(b.OpenFromFile(path, err)); + // Start from a fresh journal even if an earlier failed run left one. + std::remove(kte::SwapManager::ComputeSwapPathForTests(b).c_str()); kte::SwapManager sm; sm.Attach(&b); b.SetSwapRecorder(sm.RecorderFor(&b)); b.insert_text(0, 0, std::string("X")); + b.insert_text(0, 1, std::string("Y")); sm.Flush(&b); const std::string swap_path = kte::SwapManager::ComputeSwapPathForTests(b); b.SetSwapRecorder(nullptr); @@ -131,8 +138,11 @@ TEST(SwapReplay_TruncatedLog_FailsSafely) Buffer b2; ASSERT_TRUE(b2.OpenFromFile(path, err)); std::string rerr; - ASSERT_EQ(kte::SwapManager::ReplayFile(b2, trunc_path, rerr), false); - ASSERT_EQ(rerr.empty(), false); + ASSERT_TRUE(kte::SwapManager::ReplayFile(b2, trunc_path, rerr)); + ASSERT_TRUE(rerr.find("incomplete record") != std::string::npos); + // Detach appended a final checkpoint; that is the torn record, so both + // inserts before it are recovered. + ASSERT_EQ(b2.GetLineString(0), std::string("XYhello")); std::remove(path.c_str()); std::remove(swap_path.c_str()); @@ -225,3 +235,99 @@ TEST(SwapCompaction_RewritesToSingleCheckpoint) std::remove(path.c_str()); std::remove(swap_path.c_str()); } + + +// Inserts larger than one record's 24-bit payload are split into several INS +// records; each must be positioned after the previous chunk, including when a +// chunk boundary falls mid-line. +TEST(SwapReplay_LargeInsert_SplitAcrossRecords) +{ + const std::string path = "./.kte_ut_swap_replay_large.txt"; + std::remove(path.c_str()); + write_file_bytes(path, "tail\n"); + + Buffer b; + std::string err; + ASSERT_TRUE(b.OpenFromFile(path, err)); + std::remove(kte::SwapManager::ComputeSwapPathForTests(b).c_str()); + + // ~17 MiB of 1000-byte lines, so the 16 MiB chunk boundary splits a line. + std::string big; + const std::string line = std::string(999, 'a') + "\n"; + while (big.size() < (17u << 20)) + big += line; + big += "mid"; + + kte::SwapManager sm; + sm.Attach(&b); + b.SetSwapRecorder(sm.RecorderFor(&b)); + b.insert_text(0, 0, big); + sm.Flush(&b); + const std::string swap_path = kte::SwapManager::ComputeSwapPathForTests(b); + const std::string expected = buffer_bytes_via_views(b); + b.SetSwapRecorder(nullptr); + sm.Detach(&b, false); + + Buffer b2; + ASSERT_TRUE(b2.OpenFromFile(path, err)); + std::string rerr; + ASSERT_TRUE(kte::SwapManager::ReplayFile(b2, swap_path, rerr)); + ASSERT_TRUE(buffer_bytes_via_views(b2) == expected); + + std::remove(path.c_str()); + std::remove(swap_path.c_str()); +} + + +// When a record cannot be written, later position-based records no longer +// apply to the journal's state. The journal must drop them until a full +// checkpoint re-establishes the content, so replay still matches the buffer. +TEST(SwapReplay_LostRecord_ResyncsWithCheckpoint) +{ + const std::string path = "./.kte_ut_swap_replay_gap.txt"; + std::remove(path.c_str()); + write_file_bytes(path, "base\n"); + + Buffer b; + std::string err; + ASSERT_TRUE(b.OpenFromFile(path, err)); + const std::string swap_path = kte::SwapManager::ComputeSwapPathForTests(b); + std::remove(swap_path.c_str()); + + kte::SwapManager sm; + sm.Attach(&b); + b.SetSwapRecorder(sm.RecorderFor(&b)); + + b.insert_text(0, 0, std::string("A")); + sm.Flush(&b); + + // Make the journal unopenable so the next record is lost. + sm.ResetJournal(b); // closes and removes the file; base is now the disk file + ASSERT_TRUE(b.SaveAs(path, err)); // disk = "Abase" + std::filesystem::create_directory(swap_path); + b.insert_text(0, 1, std::string("B")); + sm.Flush(&b); + std::filesystem::remove(swap_path); + + b.insert_text(0, 2, std::string("C")); // dropped; triggers a resync checkpoint + sm.Flush(&b); + b.insert_text(0, 3, std::string("D")); + sm.Flush(&b); + const std::string expected = buffer_bytes_via_views(b); + ASSERT_EQ(expected, std::string("ABCDbase\n")); + // Replay the journal as it stands mid-session (as after a crash); Detach + // would append a final checkpoint that hides the problem. + const std::string crash_copy = swap_path + ".crash"; + write_file_bytes(crash_copy, read_file_bytes(swap_path)); + b.SetSwapRecorder(nullptr); + sm.Detach(&b, true); + + Buffer b2; + ASSERT_TRUE(b2.OpenFromFile(path, err)); + std::string rerr; + ASSERT_TRUE(kte::SwapManager::ReplayFile(b2, crash_copy, rerr)); + ASSERT_EQ(buffer_bytes_via_views(b2), expected); + + std::remove(path.c_str()); + std::remove(crash_copy.c_str()); +}