Commit bb6ba5cc authored by Ramon Nou's avatar Ramon Nou
Browse files

Test replica repair journal restart recovery

parent 6c7d71eb

REPLICA.md

+2 −2
Changes for REPLICA.md: 2 added lines, 2 removed lines.
Original line number Diff line number Diff line
@@ -1115,8 +1115,8 @@ Phase A is the current work phase. 
     truncate during repair, and remove during repair.



2. **Durable repair-journal integration**

   - Test a partial replica write, client exit, client restart, journal reload,

     and eventual repair completion.

   - 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.

tests/integration/data/test_replication.py

+74 −2
Changes for tests/integration/data/test_replication.py: 74 added lines, 2 removed lines.
Original line number Diff line number Diff line
@@ -143,8 +143,6 @@ def test_replication_survives_daemon_failure_and_recovers( 
    repair_env["LIBGKFS_NUM_REPL"] = "1"

    file_path = d00.mountdir / "replicated_failure_recovery"

    ready_marker = Path(test_workspace.twd) / "replica_failure_test_ready"

    repair_process = None



    try:

        result = setup_client.script(

            f"dd if=/dev/zero of='{file_path}' bs=1M count=2 conv=fsync status=none",
@@ -276,6 +274,80 @@ def test_replication_metadata_fallback_after_daemon_failure( 
            d00.shutdown()





def test_replication_repair_journal_survives_client_restart(

    test_workspace, gkfs_shell, request

):

    """Reload a persisted repair task in a later client process."""

    interface = request.config.getoption("--interface")

    d00 = Daemon(

        interface,

        "rocksdb",

        test_workspace,

        rootdir_suffix="journal_replica_0",

        keep_hosts=True,

    ).run()

    d01 = Daemon(

        interface,

        "rocksdb",

        test_workspace,

        rootdir_suffix="journal_replica_1",

        keep_hosts=True,

    ).run()



    journal_path = Path(test_workspace.twd) / "replica-repair.journal"

    directory_path = d00.mountdir / "replicated_journal_recovery"

    gkfs_shell._env["LIBGKFS_NUM_REPL"] = "1"

    gkfs_shell._env["LIBGKFS_REPAIR_JOURNAL_PATH"] = str(journal_path)



    try:

        result = gkfs_shell.script(f"mkdir '{directory_path}'", timeout=60)

        assert result.exit_code == 0, result.stderr.decode(errors="replace")



        os.kill(d01._proc.pid, signal.SIGKILL)

        d01._proc.wait(timeout=10)

        assert d01._proc.poll() is not None



        result = gkfs_shell.script(f"mkdir '{directory_path}.second'", timeout=60)

        assert result.exit_code == 0, result.stderr.decode(errors="replace")

        assert journal_path.exists()

        assert journal_path.read_text() != ""



        _remove_dead_daemon_registration(d01)

        d01.run()

        assert d01._proc.poll() is None



        reload_client = gkfs_shell

        deadline = time.monotonic() + 30

        while time.monotonic() < deadline:

            result = reload_client.script(

                f"stat '{directory_path}'",

                timeout=30,

            )

            if result.exit_code == 0 and journal_path.exists() and not journal_path.read_text():

                break

            time.sleep(0.5)

        assert journal_path.exists()

        assert journal_path.read_text() == ""



        os.kill(d00._proc.pid, signal.SIGKILL)

        d00._proc.wait(timeout=10)

        assert d00._proc.poll() is not None

        _remove_dead_daemon_registration(d00)

        gkfs_shell._env["LIBGKFS_NUM_REPL"] = "0"



        result = gkfs_shell.script(

            f"stat '{directory_path}.second'",

            timeout=60,

        )

        assert result.exit_code == 0, result.stderr.decode(errors="replace")

    finally:

        journal_path.unlink(missing_ok=True)

        if d01._proc is not None and d01._proc.poll() is None:

            d01.shutdown()

        if d00._proc is not None and d00._proc.poll() is None:

            d00.shutdown()





def test_replication_rejects_mutation_safely(

    test_workspace, gkfs_client, gkfs_shell, request

):