publish compares the existing block file instead of writing a block it then discards

Author: bigerl

bench/addressing_bench.cpp

@@ -8,7 +8,7 @@
 //                  recipe entry, nothing written: the skip that generation
 //                  addressing gives up (it would rewrite the block)
 //   content_dedup  hash plus publish against an existing identical file:
-//                  the staged write, the refused link, the byte compare
+//                  the byte compare that answers the publication, no write
 //   generation     write under a counter name, no hash, always a fresh
 //                  file: what a generation-addressed commit does for every
 //                  dirty block, value-identical or not
@@ -120,8 +120,7 @@ namespace {
 	}
 
 	// the dedup hit: hash plus publish where an identical file already
-	// carries the name (the staged write, the refused link, the byte
-	// compare); the staged temp file is discarded by publish itself
+	// carries the name, which the compare against that file answers
 	void content_dedup(benchmark::State &state) {
 		size_t const len = block_len_of(state);
 		bench_store bench;

include/privateer/block_store.hpp

@@ -49,8 +49,9 @@ namespace privateer {
 
 		// Publishes data under its digest, atomically. Returns true when a
 		// new file was created, false when an identical file already carried
-		// the name (dedup). A different file under the name is a fatal
-		// hash_collision error. Thread-safe.
+		// the name (dedup). A name that already has a file is answered by
+		// comparing that file, so a duplicate writes nothing. A different
+		// file under the name is a fatal hash_collision error. Thread-safe.
 		result<bool> publish(block_digest const &name, std::span<std::byte const> data) const;
 
 		// path of the block file for a name, existing or not

include/privateer/file_util.hpp

@@ -26,6 +26,10 @@ namespace privateer {
 		// it to observe what a durability path paid.
 		extern std::atomic<uint64_t> sync_calls;
 
+		// Counts every staged backing file created. Tests reset and read it
+		// to observe whether a path wrote a file at all.
+		extern std::atomic<uint64_t> staged_files;
+
 	}  // namespace detail_file_util
 #endif  // PRIVATEER_TEST_HOOKS
 

include/privateer/region.hpp

@@ -179,7 +179,8 @@ namespace privateer {
 		uint64_t slots_hashed = 0;
 		// hash matched the recipe entry, so no file was written
 		uint64_t slots_skipped = 0;
-		// new name, an identical block file already existed
+		// new name whose block file already existed, so the compare against
+		// that file replaced the write and nothing was written either
 		uint64_t slots_deduped = 0;
 		// new name, a new block file was written
 		uint64_t slots_written = 0;

src/block_store.cpp

@@ -49,10 +49,22 @@ namespace privateer {
 			}
 		}
 
-		// byte-compares an existing block file against data
-		result<bool> file_equals(fs::path const &path, std::span<std::byte const> data) {
+		// what a block name already holds against the content being published
+		enum struct existing_block : int {
+			absent,   // the name has no file
+			equal,    // the file holds this content, so it is already published
+			differs,  // the file holds other content under the same name
+		};
+
+		// Byte-compares the file under path against data. A name with no file
+		// is absent rather than an error, because writing one is what a
+		// publisher does about it.
+		result<existing_block> compare_block(fs::path const &path, std::span<std::byte const> data) {
 			int const fd = ::open(path.c_str(), O_RDONLY | O_CLOEXEC);
 			if (fd < 0) {
+				if (errno == ENOENT) {
+					return existing_block::absent;
+				}
 				return fail_errno(errc::io_error, "open block for the dedup compare");
 			}
 			struct stat st {};
@@ -63,7 +75,7 @@ namespace privateer {
 			}
 			if (std::cmp_not_equal(st.st_size, data.size())) {
 				::close(fd);
-				return false;
+				return existing_block::differs;
 			}
 			std::vector<std::byte> buffer(std::min<size_t>(data.size(), size_t{1} << 20));
 			size_t offset = 0;
@@ -80,14 +92,22 @@ namespace privateer {
 				}
 				if (got == 0 || std::memcmp(buffer.data(), data.data() + offset, static_cast<size_t>(got)) != 0) {
 					::close(fd);
-					return false;
+					return existing_block::differs;
 				}
 				offset += static_cast<size_t>(got);
 			}
 			::close(fd);
-			return true;
+			return existing_block::equal;
 		}
 
+		// How many times publish resolves the name before it gives up. Each
+		// extra round needs another publisher to take the name in the window
+		// between this one's compare and its link, and something to unlink
+		// that file again before the next compare. A directory entry that
+		// cannot be opened at all, a dangling symlink in a tampered store,
+		// exhausts the rounds and reports io_error.
+		constexpr unsigned publish_attempts = 4;
+
 	}  // namespace
 
 	result<block_store> block_store::create(fs::path const &segment_dir, bool durable) {
@@ -147,28 +167,45 @@ namespace privateer {
 		if (name.size == 0 || data.empty()) {
 			return fail(errc::invalid_argument, "publish needs a name and content");
 		}
-		auto staged = staged_file::create_in(shard_path(name));
-		if (!staged) {
-			return std::unexpected{staged.error()};
-		}
-		if (auto written = staged->write(data); !written) {
-			return std::unexpected{written.error()};
-		}
-		auto published = staged->publish(to_hex(name), publish_mode::fail_if_exists);
-		if (!published) {
-			return std::unexpected{published.error()};
-		}
-		if (*published) {
-			return true;
-		}
-		auto equal = file_equals(block_path(name), data);
-		if (!equal) {
-			return std::unexpected{equal.error()};
-		}
-		if (!*equal) {
-			return fail(errc::hash_collision, "existing block differs under the same name");
+		fs::path const path = block_path(name);
+		for (unsigned attempt = 0; attempt < publish_attempts; ++attempt) {
+			// A file already under the name carries the content the name
+			// stands for, so the compare is the whole publication and the
+			// block is not written at all. Content that repeats is the
+			// common case on a workload that copies a structure and changes
+			// parts of it.
+			auto const existing = compare_block(path, data);
+			if (!existing) {
+				return std::unexpected{existing.error()};
+			}
+			switch (*existing) {
+				case existing_block::equal:
+					return false;
+				case existing_block::differs:
+					return fail(errc::hash_collision, "existing block differs under the same name");
+				case existing_block::absent:
+					break;  // the write below is what resolves it
+			}
+			auto staged = staged_file::create_in(shard_path(name));
+			if (!staged) {
+				return std::unexpected{staged.error()};
+			}
+			if (auto written = staged->write(data); !written) {
+				return std::unexpected{written.error()};
+			}
+			auto const published = staged->publish(to_hex(name), publish_mode::fail_if_exists);
+			if (!published) {
+				return std::unexpected{published.error()};
+			}
+			if (*published) {
+				return true;
+			}
+			// The atomic link is what resolves a race on one name, and this
+			// caller lost it. The next round compares against the winner's
+			// file, which is the dedup answer; only an unlink of that file
+			// inside the same window sends the round back to the write.
 		}
-		return false;
+		return fail(errc::io_error, "the block name neither opens nor accepts a link");
 	}
 
 	result<> block_store::make_durable(std::span<block_digest const> names, sync_fan_out const &fan_out) {

src/file_util.cpp

@@ -14,6 +14,7 @@ namespace privateer {
 	namespace detail_file_util {
 
 		std::atomic<uint64_t> sync_calls{0};
+		std::atomic<uint64_t> staged_files{0};
 
 	}  // namespace detail_file_util
 #endif
@@ -26,6 +27,12 @@ namespace privateer {
 #endif
 		}
 
+		void count_staged_file() noexcept {
+#ifdef PRIVATEER_TEST_HOOKS
+			detail_file_util::staged_files.fetch_add(1, std::memory_order_relaxed);
+#endif
+		}
+
 		// fsync with EINTR retry
 		int fsync_retry(int fd) noexcept {
 			int rc;
@@ -101,6 +108,7 @@ namespace privateer {
 				if (::access("/proc/self/fd", F_OK) == 0) {
 					f.fd_ = fd;
 					f.anonymous_ = true;
+					count_staged_file();
 					return f;
 				}
 				::close(fd);
@@ -122,6 +130,7 @@ namespace privateer {
 		}
 		f.fd_ = fd;
 		f.temp_path_ = name.data();
+		count_staged_file();
 		return f;
 	}
 

test/block_store_tests.cpp

@@ -20,6 +20,8 @@
 #include <thread>
 #include <vector>
 
+#include <unistd.h>
+
 using namespace privateer;
 namespace fs = std::filesystem;
 
@@ -124,6 +126,87 @@ namespace {
 		EXPECT_EQ(file_count(f.blocks_dir()), 1u);
 	}
 
+	TEST(BlockStore, PublishWritesNothingForADuplicate) {
+		store_fixture f;
+		auto const data = content(std::string(1 << 16, 'q'));
+		auto const name = name_of(data);
+
+		detail_file_util::staged_files.store(0);
+		ASSERT_TRUE(f.store.publish(name, data));
+		EXPECT_EQ(detail_file_util::staged_files.load(), 1u);
+
+		// the name has a file, so the compare answers the publication
+		detail_file_util::staged_files.store(0);
+		auto again = f.store.publish(name, data);
+		ASSERT_TRUE(again.has_value()) << to_string(again.error());
+		EXPECT_FALSE(*again);
+		EXPECT_EQ(detail_file_util::staged_files.load(), 0u);
+		EXPECT_EQ(read_file(f.store.block_path(name)).size(), data.size());
+	}
+
+	TEST(BlockStore, PublishWritesNothingForACollision) {
+		store_fixture f;
+		auto const data = content("first content");
+		auto const name = name_of(data);
+		ASSERT_TRUE(f.store.publish(name, data));
+
+		detail_file_util::staged_files.store(0);
+		auto collided = f.store.publish(name, content("other content"));
+		ASSERT_FALSE(collided.has_value());
+		EXPECT_EQ(collided.error().code, errc::hash_collision);
+		EXPECT_EQ(detail_file_util::staged_files.load(), 0u);
+	}
+
+	TEST(BlockStore, PublishSurvivesConcurrentUnlinks) {
+		store_fixture f;
+		auto const data = content(std::string(1 << 14, 'u'));
+		auto const name = name_of(data);
+		fs::path const path = f.store.block_path(name);
+
+		// A name that keeps disappearing under the publisher: every round is
+		// either a compare against the file or a fresh write, and neither
+		// leaves the store with a temp file or an error.
+		std::atomic<bool> stop{false};
+		std::thread unlinker{[&] {
+			while (!stop.load()) {
+				::unlink(path.c_str());
+				std::this_thread::yield();
+			}
+		}};
+		int created = 0;
+		for (int round = 0; round < 500; ++round) {
+			auto published = f.store.publish(name, data);
+			if (!published) {
+				stop.store(true);
+				unlinker.join();
+				FAIL() << to_string(published.error());
+			}
+			created += *published ? 1 : 0;
+		}
+		stop.store(true);
+		unlinker.join();
+
+		EXPECT_GT(created, 0);
+		for (auto const &shard : fs::directory_iterator{f.blocks_dir()}) {
+			for (auto const &entry : fs::directory_iterator{shard}) {
+				EXPECT_FALSE(entry.path().filename().string().starts_with(temp_name_prefix));
+			}
+		}
+	}
+
+	TEST(BlockStore, PublishGivesUpOnANameThatCannotBeOpened) {
+		store_fixture f;
+		auto const data = content("content behind a broken name");
+		auto const name = name_of(data);
+		// A dangling symlink is a name whose entry refuses the link and whose
+		// file cannot be opened, so no round of publish can resolve it.
+		fs::create_symlink("nowhere", f.store.block_path(name));
+
+		auto published = f.store.publish(name, data);
+		ASSERT_FALSE(published.has_value());
+		EXPECT_EQ(published.error().code, errc::io_error);
+	}
+
 	TEST(BlockStore, PublishDetectsAHashCollision) {
 		store_fixture f;
 		auto const data = content("first content");