Complete phase A replica repair safety (dc95ba15) · Commits · hpc / gekkofs

REPLICA.md

+11 −9

Changes for REPLICA.md: 11 added lines, 9 removed lines.
Original line number	Diff line number	Diff line
		@@ -1104,7 +1104,8 @@ testable changes over new availability features.

		### Phase A — repair correctness and restart safety

		Phase A is the current work phase.
		Phase A is complete for the current client-local repair scope. The remaining
		limitations below are intentionally deferred to later phases.

		1. Repair operation ordering
		- Make append, inline-data migration, truncate, remove, and metadata-size
		@@ -1115,16 +1116,17 @@ Phase A is the current work phase.
		truncate during repair, and remove during repair.

		2. Durable repair-journal integration
		- In progress: test a partial replica write, client exit, client restart,
		journal reload, and eventual repair completion.
		- Validate malformed, truncated, empty, and atomically replaced journals.
		- Define journal ownership and behavior when multiple clients use the same
		path.
		- Tested: partial replica write, client exit, client restart, journal reload,
		and eventual repair completion.
		- Malformed, truncated, empty, invalid-kind, duplicate, and atomic-replace
		failure handling are bounded and tested where observable.
		- A journal path is single-owner: concurrent client processes must not share
		it. Cross-client locking and distributed ownership are Phase C work.

		3. Topology-aware journal validation
		- Record enough host/topology identity to detect a journal created for a
		different active host layout.
		- Do not silently send a recovered task to a different host identity.
		- Deferred: durable host/topology identity is not part of the current local
		journal format. Recovered tasks continue to use current copy placement.
		- Do not share a journal across clients or topology changes.

		Phase A impact: high correctness and operational impact, with no intended
		change to the storage format or default write policy. This phase should reduce

src/client/gkfs_metadata.cpp

+37 −0

Changes for src/client/gkfs_metadata.cpp: 37 added lines, 0 removed lines.
Original line number	Diff line number	Diff line
		@@ -62,6 +62,7 @@

		#include <common/rpc/distributor.hpp>
		#include <common/rpc/rpc_types_thallium.hpp>
		#include <common/arithmetic/arithmetic.hpp>
		#include <common/path_util.hpp>
		#include <thallium.hpp>
		#ifdef GKFS_ENABLE_CLIENT_METRICS
		@@ -516,6 +517,20 @@ gkfs_remove(const std::string& path, bool is_rename_stub) {
		uint32_t mode = 0;
		std::string target_path;

		if(CTX->get_replicas() > 0) {
		// Invalidate metadata repairs before removal is sent. The authoritative
		// remove response provides the file size needed to invalidate old data
		// chunk repairs below.
		CTX->next_repair_generation(path,
		gkfs::rpc::repair_kind::metadata_create);
		CTX->next_repair_generation(
		path, gkfs::rpc::repair_kind::metadata_create_inline);
		CTX->next_repair_generation(path,
		gkfs::rpc::repair_kind::metadata_size);
		CTX->next_repair_generation(path,
		gkfs::rpc::repair_kind::metadata_inline);
		}

		if(gkfs::config::proxy::fwd_remove && CTX->use_proxy()) {
		err = gkfs::rpc::forward_remove_proxy(path, false);
		} else {
		@@ -527,6 +542,16 @@ gkfs_remove(const std::string& path, bool is_rename_stub) {
		return -1;
		}

		if(CTX->get_replicas() > 0 && size > 0) {
		const auto chunk_count = (static_cast<uint64_t>(size) +
		gkfs::config::rpc::chunksize - 1) /
		gkfs::config::rpc::chunksize;
		for(uint64_t chunk_id = 0; chunk_id < chunk_count; ++chunk_id) {
		CTX->next_repair_generation(
		path, gkfs::rpc::repair_kind::data_chunk, chunk_id);
		}
		}

		if(gkfs::config::metadata::rename_support && !is_rename_stub) {
		if(!target_path.empty() && !S_ISLNK(mode)) {
		// It was a rename link! We recursively remove the target path.
		@@ -998,6 +1023,18 @@ gkfs_truncate(const std::string& path, off_t old_size, off_t new_size) {
		if(new_size == old_size) {
		return 0;
		}
		if(CTX->get_replicas() > 0) {
		CTX->next_repair_generation(path,
		gkfs::rpc::repair_kind::metadata_size);
		const auto first_chunk = gkfs::utils::arithmetic::block_index(
		new_size, gkfs::config::rpc::chunksize);
		const auto last_chunk = gkfs::utils::arithmetic::block_index(
		old_size - 1, gkfs::config::rpc::chunksize);
		for(auto chunk_id = first_chunk; chunk_id <= last_chunk; ++chunk_id) {
		CTX->next_repair_generation(
		path, gkfs::rpc::repair_kind::data_chunk, chunk_id);
		}
		}
		int err = 0;
		// decrease size on metadata server first
		if(gkfs::config::proxy::fwd_truncate && CTX->use_proxy()) {

src/client/preload_context.cpp

+37 −3

Changes for src/client/preload_context.cpp: 37 added lines, 3 removed lines.
Original line number	Diff line number	Diff line
		@@ -64,6 +64,7 @@
		#include <iomanip>
		#include <sstream>
		#include <string>
		#include <unordered_set>
		#include <utility>
		#include <cstdio>
		#include <unistd.h>
		@@ -1367,18 +1368,37 @@ PreloadContext::load_repair_journal() {
		}

		std::vector<gkfs::rpc::repair_task> tasks;
		while(input) {
		std::string line;
		while(std::getline(input, line)) {
		if(line.empty()) {
		continue;
		}
		gkfs::rpc::repair_task task;
		int kind = 0;
		int data_chunk = 0;
		int append = 0;
		int clear_inline = 0;
		if(!(input >> std::quoted(task.path) >> task.chunk_id >>
		std::istringstream record(line);
		if(!(record >> std::quoted(task.path) >> task.chunk_id >>
		task.source_copy >> task.target_copy >> task.generation >>
		task.retry_count >> task.last_error >> data_chunk >> kind >>
		task.mode >> task.size >> task.offset >> append >> clear_inline >>
		std::quoted(task.data))) {
		break;
		LOG(WARNING, "Ignoring malformed replica repair journal record.");
		continue;
		}
		std::string trailing;
		if(record >> trailing) {
		LOG(WARNING,
		"Ignoring replica repair journal record with trailing data.");
		continue;
		}
		if(kind < static_cast<int>(gkfs::rpc::repair_kind::metadata_create) \|\|
		kind > static_cast<int>(gkfs::rpc::repair_kind::data_chunk)) {
		LOG(WARNING,
		"Ignoring replica repair journal record with invalid kind {}.",
		kind);
		continue;
		}
		task.data_chunk = data_chunk != 0;
		task.kind = static_cast<gkfs::rpc::repair_kind>(kind);
		@@ -1406,7 +1426,19 @@ PreloadContext::persist_repair_journal_locked() const {
		return;
		}

		std::unordered_set<std::string> written_tasks;
		auto task_key = [](const auto& task) {
		std::ostringstream key;
		key << std::quoted(task.path) << ':' << task.chunk_id << ':'
		<< static_cast<int>(task.source_copy) << ':'
		<< static_cast<int>(task.target_copy) << ':'
		<< static_cast<int>(task.kind) << ':' << task.generation;
		return key.str();
		};
		auto write_task = [&](const auto& task) {
		if(!written_tasks.emplace(task_key(task)).second) {
		return;
		}
		output << std::quoted(task.path) << ' ' << task.chunk_id << ' '
		<< static_cast<int>(task.source_copy) << ' '
		<< static_cast<int>(task.target_copy) << ' ' << task.generation
		@@ -1427,6 +1459,8 @@ PreloadContext::persist_repair_journal_locked() const {
		std::error_code ec;
		std::filesystem::rename(temporary, repair_journal_path_, ec);
		if(ec) {
		LOG(WARNING, "Unable to replace replica repair journal '{}': {}",
		repair_journal_path_, ec.message());
		std::filesystem::remove(temporary, ec);
		}
		}

tests/integration/data/test_replication.py

+37 −0

Changes for tests/integration/data/test_replication.py: 37 added lines, 0 removed lines.
Original line number	Diff line number	Diff line
		@@ -348,6 +348,43 @@ def test_replication_repair_journal_survives_client_restart(
		d00.shutdown()


		def test_replication_repair_journal_ignores_bad_records(
		test_workspace, gkfs_shell, request
		):
		"""Malformed and empty journals do not prevent a fresh client from starting."""
		interface = request.config.getoption("--interface")
		daemon = Daemon(
		interface,
		"rocksdb",
		test_workspace,
		rootdir_suffix="journal_validation",
		keep_hosts=True,
		).run()
		journal_path = Path(test_workspace.twd) / "replica-invalid.journal"
		gkfs_shell._env["LIBGKFS_REPAIR_JOURNAL_PATH"] = str(journal_path)
		file_path = daemon.mountdir / "journal_validation_file"

		try:
		journal_path.write_text(
		"not a repair record\n"
		'"/valid" 0 0 1 1 0 5 0 99 0 0 0 0 0 ""\n'
		)
		result = gkfs_shell.script(f"stat '{daemon.mountdir}'", timeout=60)
		assert result.exit_code == 0, result.stderr.decode(errors="replace")
		assert journal_path.read_text() != ""

		journal_path.write_text("")
		result = gkfs_shell.script(
		f"touch '{file_path}' && stat '{file_path}'", timeout=60
		)
		assert result.exit_code == 0, result.stderr.decode(errors="replace")
		assert journal_path.read_text() == ""
		finally:
		journal_path.unlink(missing_ok=True)
		if daemon._proc is not None and daemon._proc.poll() is None:
		daemon.shutdown()


		def test_replication_rejects_mutation_safely(
		test_workspace, gkfs_client, gkfs_shell, request
		):

tests/unit/test_repair_queue.cpp

+13 −0

Changes for tests/unit/test_repair_queue.cpp: 13 added lines, 0 removed lines.
Original line number	Diff line number	Diff line
		@@ -216,3 +216,16 @@ TEST_CASE("repair queue keeps append metadata repairs executable",
		REQUIRE(second->append);
		REQUIRE(second->kind == gkfs::rpc::repair_kind::metadata_inline);
		}

		TEST_CASE("repair queue coalesces journal duplicates by generation",
		"[replication][repair]") {
		gkfs::rpc::repair_queue queue;
		const auto now = std::chrono::steady_clock::now();
		gkfs::rpc::repair_task task{"/journal", 2, 0, 1, 8, 0, EIO, now};
		task.kind = gkfs::rpc::repair_kind::data_chunk;

		REQUIRE(queue.enqueue(task));
		REQUIRE(queue.enqueue(task));
		REQUIRE(queue.size() == 1);
		REQUIRE(queue.stats().coalesced == 1);
		}
		No newline at end of file