From cb9b6bdf4585a1a0ea6b10895171488ed4156aa6 Mon Sep 17 00:00:00 2001 From: Claude Date: Thu, 24 Sep 2026 04:36:21 +0000 Subject: [PATCH] Swap: lock journals to one session; record the base file in the header - open_ctx takes a non-blocking flock on the journal. A journal locked by another kte process is left alone: this session reports it once and does not write (records from two sessions used to interleave in one file). Opening a file whose journal is live in another session skips recovery instead of replaying or removing that journal. - The header records the size and mtime of the file the journal's records apply to (flags bit 0, bytes 24..39; previously reserved, so no version change). Before offering recovery, a journal whose base no longer matches the file on disk (changed after a crash, or the crash fell between a save's rename and the journal reset) is treated as stale and offered for deletion, instead of replaying its edits onto different text. Journals from older kte (flag clear) replay as before. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_015rcVQiiWyTzoNrsvLSaAfB --- Editor.cc | 25 ++++++++ Swap.cc | 97 ++++++++++++++++++++++++++++-- Swap.h | 20 +++++- tests/test_swap_recovery_prompt.cc | 63 +++++++++++++++++++ tests/test_swap_writer.cc | 6 +- 5 files changed, 202 insertions(+), 9 deletions(-) diff --git a/Editor.cc b/Editor.cc index 8288036..02768fa 100644 --- a/Editor.cc +++ b/Editor.cc @@ -512,6 +512,31 @@ Editor::ProcessPendingOpens() } catch (...) { swp_exists = false; } + if (swp_exists && kte::SwapManager::JournalInUse(swp)) { + // Another running kte is editing this file: its journal is live. + // Open without recovery and leave the journal alone; this + // session will not journal the file. + std::string err; + if (!OpenFile(req.path, err)) { + SetStatus(err); + continue; + } + apply_pending_line(*this, req.line1); + SetStatus(req.path + " is open in another kte; this copy has no crash recovery"); + return true; + } + if (swp_exists && !kte::SwapManager::JournalMatchesFile(swp, req.path)) { + // The file changed (or was removed) after the journal started, so + // its position-based edits no longer apply. + pending_recovery_prompt_ = RecoveryPromptKind::DeleteCorruptSwap; + pending_recovery_open_ = req; + pending_recovery_swap_path_ = swp; + pending_recovery_replay_err_ = "stale"; + StartPrompt(PromptKind::Confirm, "Swap", ""); + SetStatus("Swap file for " + req.path + + " predates changes to the file on disk and cannot be applied. Delete it? (y/N, C-g cancel)"); + return opened_any; + } if (swp_exists) { Buffer tmp; std::string oerr; diff --git a/Swap.cc b/Swap.cc index b10df80..dedf1cc 100644 --- a/Swap.cc +++ b/Swap.cc @@ -15,6 +15,7 @@ #include #include #include +#include #include #include @@ -150,6 +151,23 @@ encode_path_key(std::string s) } +// 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(st.st_size); +#if defined(__APPLE__) + mtime_ns = static_cast(st.st_mtimespec.tv_sec) * 1000000000LL + st.st_mtimespec.tv_nsec; +#else + mtime_ns = static_cast(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) { @@ -329,9 +347,12 @@ SwapManager::Attach(Buffer *buf) if (!buf) return; std::lock_guard lg(mtx_); - JournalCtx &ctx = journals_[buf]; + const bool fresh = journals_.find(buf) == journals_.end(); + JournalCtx &ctx = journals_[buf]; if (ctx.path.empty()) ctx.path = ComputeSidecarPath(*buf); + if (fresh) + ctx.has_base = stat_base(buf->Filename(), ctx.base_size, ctx.base_mtime_ns); // Ensure a recorder exists as well. if (recorders_.find(buf) == recorders_.end()) { recorders_[buf] = std::make_unique(*this, *buf); @@ -406,6 +427,8 @@ SwapManager::ResetJournal(Buffer &buf) 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; + ctx.has_base = stat_base(buf.Filename(), ctx.base_size, ctx.base_mtime_ns); ctx.last_flush_ns = 0; ctx.last_fsync_ns = 0; ctx.last_chkpt_ns = 0; @@ -538,6 +561,8 @@ SwapManager::NotifyFilenameChanged(Buffer &buf) ctx.gap = false; ctx.gap_chkpt_request_ns = 0; ctx.gap_unfixable_reported = false; + ctx.locked_out = false; + ctx.has_base = stat_base(buf.Filename(), ctx.base_size, ctx.base_mtime_ns); ctx.path = new_path; ctx.suspended = false; ctx.header_ok = false; @@ -644,12 +669,14 @@ SwapManager::ensure_parent_dir(const std::string &path) bool -SwapManager::write_header(int fd) +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][reserved/padding] + // [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); @@ -658,7 +685,12 @@ SwapManager::write_header(int fd) hdr[9] = static_cast((VERSION >> 8) & 0xFFu); hdr[10] = static_cast((VERSION >> 16) & 0xFFu); hdr[11] = static_cast((VERSION >> 24) & 0xFFu); - // flags = 0 + // flags: bit 0 = base file identity present + if (ctx.has_base) { + hdr[12] = 1; + put_le64(hdr + 24, ctx.base_size); + put_le64(hdr + 32, static_cast(ctx.base_mtime_ns)); + } // created_time (unix seconds; little-endian) std::uint64_t ts = static_cast(std::time(nullptr)); put_le64(hdr + 16, ts); @@ -695,6 +727,15 @@ SwapManager::open_ctx(JournalCtx &ctx, const std::string &path, std::string &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); + ctx.locked_out = true; + 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{}; @@ -728,13 +769,14 @@ SwapManager::open_ctx(JournalCtx &ctx, const std::string &path, std::string &err } return false; } + (void) ::flock(fd, LOCK_EX | LOCK_NB); (void) kte::syscall::Fchmod(fd, 0600); st.st_size = 0; } ctx.fd = fd; ctx.path = path; if (st.st_size == 0) { - ctx.header_ok = write_header(fd); + 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; @@ -812,7 +854,7 @@ SwapManager::compact_to_checkpoint(JournalCtx &ctx, const std::vectorsecond.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; @@ -1275,6 +1320,8 @@ SwapManager::process_one(const Pending &p) std::string open_err; if (!open_ctx(*ctxp, path, open_err)) { report_error(open_err, p.buf); + if (ctxp->locked_out) + return; // not an I/O failure; later records are skipped { std::lock_guard lg(mtx_); circuit_breaker_.RecordFailure(); @@ -1408,6 +1455,44 @@ parse_u32_le(const std::vector &p, std::size_t &off, std::uint32_t } +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(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 + return size == want_size && static_cast(mtime) == want_mtime; +} + + bool SwapManager::ReplayFile(Buffer &buf, const std::string &swap_path, std::string &err, std::uint64_t *valid_bytes) diff --git a/Swap.h b/Swap.h index 62f1303..966d65d 100644 --- a/Swap.h +++ b/Swap.h @@ -123,6 +123,16 @@ public: // Returns empty string if filename is empty. static std::string ComputeSwapPathForFilename(const std::string &filename); + // True if another process currently holds the journal at swap_path (it + // is live, not left by a crash): it must be neither replayed nor removed. + static bool JournalInUse(const std::string &swap_path); + + // False if the journal records which version of `file_path` it applies + // to and the file on disk no longer matches (edited or replaced since + // the journal started); replaying it would apply edits to the wrong text. + // Journals without that record (older kte) are assumed to match. + static bool JournalMatchesFile(const std::string &swap_path, const std::string &file_path); + // Test-only hook to keep swap path logic centralized. // (Avoid duplicating naming rules in unit tests.) #ifdef KTE_TESTS @@ -204,6 +214,14 @@ private: bool gap{false}; std::uint64_t gap_chkpt_request_ns{0}; bool gap_unfixable_reported{false}; // gap on a buffer too large to checkpoint + // Another kte process holds this journal's lock: do not write to it. + bool locked_out{false}; + // Identity of the file the journal's records apply to (captured when + // the journal starts) and written into its header, so recovery can + // tell whether the file changed since. + bool has_base{false}; + std::uint64_t base_size{0}; + std::int64_t base_mtime_ns{0}; }; struct Pending { @@ -225,7 +243,7 @@ private: static std::string SwapDirRoot(); - static bool write_header(int fd); + static bool write_header(int fd, const JournalCtx &ctx); static bool open_ctx(JournalCtx &ctx, const std::string &path, std::string &err); diff --git a/tests/test_swap_recovery_prompt.cc b/tests/test_swap_recovery_prompt.cc index d47b809..6f98194 100644 --- a/tests/test_swap_recovery_prompt.cc +++ b/tests/test_swap_recovery_prompt.cc @@ -520,3 +520,66 @@ TEST(SwapJournal_Reload_ResetsJournal) ASSERT_TRUE(kte::SwapManager::ReplayFile(check, kte::SwapManager::ComputeSwapPathForTests(*cur), rerr)); ASSERT_EQ(buffer_bytes_via_views(check), live); } + + +// A journal whose base file changed on disk afterwards is not replayed onto +// the new content; the user is offered to delete it instead. +TEST(SwapRecoveryPrompt_StaleJournal_NotReplayed) +{ + ktet::InstallDefaultCommandsOnce(); + XdgSandbox sb("stale"); + const std::string file = (sb.root / "work" / "s.txt").string(); + write_file_bytes(file, "base\n"); + std::string expected; + (void) make_journal(file, [](Buffer &b) { + b.insert_text(0, 0, std::string("A")); + }, expected); + write_file_bytes(file, "completely different content\n"); // changed after the crash + + Editor ed; + ed.SetDimensions(24, 80); + ed.AddBuffer(Buffer()); + ed.RequestOpenFile(file); + (void) ed.ProcessPendingOpens(); + ASSERT_EQ(ed.PendingRecoveryPrompt(), Editor::RecoveryPromptKind::DeleteCorruptSwap); + answer(ed, "y"); + ASSERT_EQ(buffer_bytes_via_views(*ed.CurrentBuffer()), std::string("completely different content\n")); +} + + +// A journal held by another live session is neither replayed, removed nor +// written to by a second session opening the same file. +TEST(SwapRecoveryPrompt_LiveJournalOfOtherSession_LeftAlone) +{ + ktet::InstallDefaultCommandsOnce(); + XdgSandbox sb("live"); + const std::string file = (sb.root / "work" / "l.txt").string(); + write_file_bytes(file, "base\n"); + + // "Other process": a session with the journal open (and locked). + Buffer other; + std::string err; + ASSERT_TRUE(other.OpenFromFile(file, err)); + kte::SwapManager other_sm; + other_sm.Attach(&other); + other.SetSwapRecorder(other_sm.RecorderFor(&other)); + other.insert_text(0, 0, std::string("O")); + other_sm.Flush(&other); + const std::string swp = kte::SwapManager::ComputeSwapPathForTests(other); + ASSERT_TRUE(kte::SwapManager::JournalInUse(swp)); + const std::string before = read_file_bytes(swp); + + Editor ed; + ed.SetDimensions(24, 80); + ed.AddBuffer(Buffer()); + ed.RequestOpenFile(file); + (void) ed.ProcessPendingOpens(); + ASSERT_EQ(ed.PendingRecoveryPrompt(), Editor::RecoveryPromptKind::None); + ASSERT_EQ(buffer_bytes_via_views(*ed.CurrentBuffer()), std::string("base\n")); + ed.CurrentBuffer()->insert_text(0, 0, std::string("X")); + ed.Swap()->Flush(ed.CurrentBuffer()); + ASSERT_EQ(read_file_bytes(swp), before); + + other.SetSwapRecorder(nullptr); + other_sm.Detach(&other, true); +} diff --git a/tests/test_swap_writer.cc b/tests/test_swap_writer.cc index a72a461..4082b23 100644 --- a/tests/test_swap_writer.cc +++ b/tests/test_swap_writer.cc @@ -124,8 +124,10 @@ TEST(SwapWriter_Header_Records_And_CRC) 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); - // flags currently 0 - ASSERT_EQ(read_le32(bytes.data() + 12), (std::uint32_t) 0); + // flags bit 0: the base file's identity (size, mtime) is recorded, since + // the buffer is file-backed. + ASSERT_EQ(read_le32(bytes.data() + 12), (std::uint32_t) 1); + ASSERT_EQ(read_le64(bytes.data() + 24), (std::uint64_t) std::filesystem::file_size(b.Filename())); ASSERT_TRUE(read_le64(bytes.data() + 16) != 0); // Records