Swap: fsync without blocking edits; base identity off the lock; races

- The writer's periodic fsync now holds a separate io_mtx_, taken also
  (before mtx_) by the main-thread paths that close journal fds, instead
  of mtx_. Every keystroke takes mtx_ to enqueue its record, so a slow
  fsync (writeback pressure, NFS) froze typing.
- A journal's base identity (stat + content CRC) is computed before
  taking mtx_ and applied under it; the CRC is skipped above 64 MiB
  (size and mtime still identify the file). It re-read the whole file
  under the lock on every open and save.
- A buffer that was never journaled (a virtual buffer saved under a
  real name) is attached by NotifyFilenameChanged.
- ThreadSanitizer: open_ctx (writer thread) no longer writes ctx.path,
  and locked_out is recorded by the caller under mtx_; both raced with
  main-thread reads.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_015rcVQiiWyTzoNrsvLSaAfB

Swap.cc

@@ -348,13 +348,19 @@ 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 fresh = journals_.find(buf) == journals_.end();
+	const bool still_fresh = journals_.find(buf) == journals_.end();
-	JournalCtx &ctx  = journals_[buf];
+	JournalCtx &ctx        = journals_[buf];
 	if (ctx.path.empty())
 		ctx.path = ComputeSidecarPath(*buf);
-	if (fresh)
+	if (fresh && still_fresh)
-		capture_base(buf->Filename(), ctx);
+		apply_base(ctx, base);
 	// Ensure a recorder exists as well.
 	if (recorders_.find(buf) == recorders_.end()) {
 		recorders_[buf] = std::make_unique<BufferRecorder>(*this, *buf);
@@ -385,6 +391,7 @@ SwapManager::Detach(Buffer *buf, const bool remove_file)
 	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()) {
@@ -419,8 +426,10 @@ SwapManager::ResetJournal(Buffer &buf)
 	}
 
 	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())
@@ -433,7 +442,7 @@ SwapManager::ResetJournal(Buffer &buf)
 		ctx.gap_chkpt_request_ns   = 0;
 		ctx.gap_unfixable_reported = false;
 		ctx.locked_out             = false;
-		capture_base(buf.Filename(), ctx);
+		apply_base(ctx, base);
 		ctx.last_flush_ns          = 0;
 		ctx.last_fsync_ns          = 0;
 		ctx.last_chkpt_ns          = 0;
@@ -541,15 +550,28 @@ SwapManager::NotifyFilenameChanged(Buffer &buf)
 	// 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())
+		if (it != journals_.end()) {
-			return;
+			attached             = true;
-		old_path             = it->second.path;
+			old_path             = it->second.path;
-		it->second.suspended = true;
+			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())
@@ -566,7 +588,7 @@ SwapManager::NotifyFilenameChanged(Buffer &buf)
 	ctx.gap_chkpt_request_ns    = 0;
 	ctx.gap_unfixable_reported  = false;
 	ctx.locked_out              = false;
-	capture_base(buf.Filename(), ctx);
+	apply_base(ctx, base);
 	ctx.path                    = new_path;
 	ctx.suspended              = false;
 	ctx.header_ok              = false;
@@ -714,7 +736,7 @@ SwapManager::write_header(int fd, const JournalCtx &ctx)
 
 
 bool
-SwapManager::open_ctx(JournalCtx &ctx, const std::string &path, std::string &err)
+SwapManager::open_ctx(JournalCtx &ctx, const std::string &path, std::string &err, bool *locked_out)
 {
 	err.clear();
 	if (ctx.fd >= 0)
@@ -747,8 +769,9 @@ SwapManager::open_ctx(JournalCtx &ctx, const std::string &path, std::string &err
 	// its records with ours.
 	if (::flock(fd, LOCK_EX | LOCK_NB) != 0 && errno == EWOULDBLOCK) {
 		kte::syscall::Close(fd);
-		ctx.locked_out = true;
+		if (locked_out)
-		err            = "Swap file is in use by another kte process: " + path;
+			*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.
@@ -786,15 +809,17 @@ SwapManager::open_ctx(JournalCtx &ctx, const std::string &path, std::string &err
 		}
 		if (::flock(fd, LOCK_EX | LOCK_NB) != 0 && errno == EWOULDBLOCK) {
 			kte::syscall::Close(fd);
-			ctx.locked_out = true;
+			if (locked_out)
-			err            = "Swap file is in use by another kte process: " + path;
+				*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.fd   = fd;
+	// (ctx.path is not written here: this runs on the writer thread and the
-	ctx.path = path;
+	// 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;
@@ -1237,22 +1262,29 @@ SwapManager::writer_loop()
 		// Throttled fsync: best-effort (grouped)
 		try {
 			std::uint64_t now = now_ns();
-			// fsync under mtx_: the main thread closes journal fds under the
+			// Hold io_mtx_ across the fsyncs: journal fds are only closed
-			// same lock. Syncing outside it raced with those closes: the fd
+			// under it (by the main thread), so none of these can be closed or
-			// could be closed (or reused) mid-fsync, and the in-flight fsync
+			// reused mid-fsync. Unsynchronised, an in-flight fsync also kept a
-			// kept the closed journal's flock alive, so the owner's own
+			// just-closed journal's flock alive, so the owner's own cleanup saw
-			// cleanup saw its journal as held by someone else.
+			// its journal as held. mtx_ is held only to collect the fds, so
-			std::lock_guard<std::mutex> lg(mtx_);
+			// edits are not blocked while fsync runs.
-			for (auto &kv: journals_) {
+			std::lock_guard<std::mutex> io(io_mtx_);
-				JournalCtx &ctx = kv.second;
+			std::vector<int> to_sync;
-				if (ctx.fd >= 0) {
+			{
-					if (ctx.last_fsync_ns == 0 || (now - ctx.last_fsync_ns) / 1000000ULL >=
+				std::lock_guard<std::mutex> lg(mtx_);
-					    cfg_.fsync_interval_ms) {
+				for (auto &kv: journals_) {
-						ctx.last_fsync_ns = now;
+					JournalCtx &ctx = kv.second;
-						(void) kte::syscall::Fsync(ctx.fd);
+					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 (...) {
@@ -1340,10 +1372,17 @@ SwapManager::process_one(const Pending &p)
 		if (!ctxp)
 			return;
 		std::string open_err;
-		if (!open_ctx(*ctxp, path, open_err)) {
+		bool locked_out = false;
-			report_error(open_err, p.buf);
+		if (!open_ctx(*ctxp, path, open_err, &locked_out)) {
-			if (ctxp->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();
@@ -1560,11 +1599,27 @@ SwapManager::file_crc32(const std::string &path, std::uint32_t &out)
 }
 
 
-void
+SwapManager::BaseId
-SwapManager::capture_base(const std::string &file, JournalCtx &ctx)
+SwapManager::compute_base(const std::string &file)
 {
-	ctx.has_base     = stat_base(file, ctx.base_size, ctx.base_mtime_ns);
+	// Content CRCs are skipped for very large files (reading them again on
-	ctx.has_base_crc = ctx.has_base && file_crc32(file, ctx.base_crc);
+	// 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;
 }
 
 

Swap.h

@@ -247,7 +247,7 @@ private:
 
 	static bool write_header(int fd, const JournalCtx &ctx);
 
-	static bool open_ctx(JournalCtx &ctx, const std::string &path, std::string &err);
+	static bool open_ctx(JournalCtx &ctx, const std::string &path, std::string &err, bool *locked_out = nullptr);
 
 	static void close_ctx(JournalCtx &ctx);
 
@@ -259,9 +259,20 @@ private:
 	// CRC-32 of a file's content; false if it cannot be read.
 	static bool file_crc32(const std::string &path, std::uint32_t &out);
 
-	// Record the identity (size, mtime, content CRC) of the file a journal's
+	// Identity (size, mtime, content CRC) of the file a journal's records
-	// records will apply to.
+	// apply to. Computed without holding mtx_ (it reads the file), then
-	static void capture_base(const std::string &file, JournalCtx &ctx);
+	// stored into the JournalCtx under it.
+	struct BaseId {
+		bool has{false};
+		std::uint64_t size{0};
+		std::int64_t mtime_ns{0};
+		bool has_crc{false};
+		std::uint32_t crc{0};
+	};
+
+	static BaseId compute_base(const std::string &file);
+
+	static void apply_base(JournalCtx &ctx, const BaseId &id);
 
 	static void put_le32(std::vector<std::uint8_t> &out, std::uint32_t v);
 
@@ -283,6 +294,11 @@ private:
 	std::unordered_map<Buffer *, JournalCtx> journals_;
 	std::unordered_map<Buffer *, std::unique_ptr<BufferRecorder> > recorders_;
 	mutable std::mutex mtx_;
+	// Serialises closing journal fds with the writer's periodic fsync, so
+	// neither can act on an fd the other has closed (or that was reused).
+	// Taken before mtx_ when both are needed; edits (which take mtx_) never
+	// wait for an fsync.
+	mutable std::mutex io_mtx_;
 	std::condition_variable cv_;
 	std::vector<Pending> queue_;
 	std::uint64_t next_seq_{0};

tests/test_audit_regressions.cc

@@ -1126,3 +1126,23 @@ TEST(Audit_Save_HardLink_LeftoverCopyDoesNotBlock)
 	ASSERT_TRUE(b.Save(err));
 	ASSERT_EQ(slurp(d.path / "h2.txt"), std::string("new old\n"));
 }
+
+
+// A virtual buffer saved under a real name is journaled from then on.
+TEST(Audit_VirtualBuffer_SavedAs_GetsJournal)
+{
+	TempDir d("virtual_saveas");
+	TestHarness h;
+	Editor &ed = h.EditorRef();
+	ASSERT_TRUE(h.Exec(CommandId::ShowHelp));
+	ASSERT_TRUE(h.Exec(CommandId::ToggleReadOnly));
+	const std::string target = (d.path / "notes.txt").string();
+	ASSERT_TRUE(Execute(ed, "save-as", target));
+	Buffer *b = ed.CurrentBuffer();
+	ASSERT_TRUE(!b->IsVirtual());
+	ASSERT_TRUE(h.Exec(CommandId::InsertText, "x"));
+	ed.Swap()->Flush(b);
+	const std::string swp = kte::SwapManager::ComputeSwapPathForTests(*b);
+	ASSERT_TRUE(std::filesystem::exists(swp));
+	std::filesystem::remove(swp);
+}