diff --git a/.github/workflows/pagestore-test.yml b/.github/workflows/pagestore-test.yml index 258f5ffab3d6c..34d974f7c0ae1 100644 --- a/.github/workflows/pagestore-test.yml +++ b/.github/workflows/pagestore-test.yml @@ -73,30 +73,39 @@ jobs: - name: Compile daemon and test harness working-directory: contrib/pagestore run: | - cc -O2 -Wall -Wextra -Werror -o pagestore_daemon pagestore_daemon.c pagestore_core.c pagestore_fault.c storage_posix.c pagestore_layer.c pagestore_layer_store.c pagestore_manifest.c pagestore_memtable.c pagestore_pgcache.c pagestore_prune.c pagestore_retention.c pagestore_wal_store.c pagestore_wal_segment.c pagestore_walidx_prune.c pagestore_walidx_snapshot.c pagestore_forkmeta_prune.c pagestore_forkmeta_snapshot.c -lrt -lpthread + cc -O2 -Wall -Wextra -Werror -o pagestore_daemon pagestore_daemon.c pagestore_core.c pagestore_fault.c storage_posix.c pagestore_layer.c pagestore_layer_store.c pagestore_manifest.c pagestore_memtable.c pagestore_pgcache.c pagestore_prune.c pagestore_retention.c pagestore_wal_store.c pagestore_wal_segment.c pagestore_walidx_prune.c pagestore_walidx_snapshot.c pagestore_forkmeta_prune.c pagestore_forkmeta_snapshot.c pagestore_store_owner.c -lrt -lpthread cc -O2 -Wall -Wextra -Werror -o pagestore_inspect pagestore_inspect.c -lrt cc -O2 -Wall -Wextra -Werror -o pagestore_test pagestore_test.c -lrt - cc -O2 -Wall -Wextra -Werror -o pagestore_layer_test pagestore_layer_test.c pagestore_layer.c pagestore_layer_store.c pagestore_fault.c -lrt -pthread - cc -O2 -Wall -Wextra -Werror -o pagestore_layer_store_test pagestore_layer_store_test.c pagestore_layer_store.c pagestore_layer.c pagestore_fault.c -lrt -pthread + cc -O2 -Wall -Wextra -Werror -o pagestore_layer_test pagestore_layer_test.c pagestore_layer.c pagestore_layer_store.c pagestore_fault.c pagestore_store_owner.c -lrt -pthread + cc -O2 -Wall -Wextra -Werror -o pagestore_layer_store_test pagestore_layer_store_test.c pagestore_layer_store.c pagestore_layer.c pagestore_fault.c pagestore_store_owner.c -lrt -pthread cc -O2 -Wall -Wextra -Werror -o pagestore_layer_crash_client pagestore_layer_crash_client.c -lrt -pthread - cc -O2 -Wall -Wextra -Werror -o pagestore_tiering_test pagestore_tiering_test.c pagestore_core.c pagestore_fault.c storage_posix.c pagestore_layer.c pagestore_layer_store.c pagestore_manifest.c pagestore_memtable.c pagestore_pgcache.c pagestore_prune.c pagestore_retention.c pagestore_wal_store.c pagestore_wal_segment.c pagestore_walidx_prune.c pagestore_walidx_snapshot.c pagestore_forkmeta_prune.c pagestore_forkmeta_snapshot.c -lrt -lpthread - cc -O2 -Wall -Wextra -Werror -o pagestore_forkmeta_cutover_test pagestore_forkmeta_cutover_test.c pagestore_core.c pagestore_fault.c storage_posix.c pagestore_layer.c pagestore_layer_store.c pagestore_manifest.c pagestore_memtable.c pagestore_pgcache.c pagestore_prune.c pagestore_retention.c pagestore_wal_store.c pagestore_wal_segment.c pagestore_walidx_prune.c pagestore_walidx_snapshot.c pagestore_forkmeta_prune.c pagestore_forkmeta_snapshot.c -lrt -lpthread - cc -O2 -Wall -Wextra -Werror -o pagestore_forkmeta_crash_matrix_test pagestore_forkmeta_crash_matrix_test.c pagestore_core.c pagestore_fault.c storage_posix.c pagestore_layer.c pagestore_layer_store.c pagestore_manifest.c pagestore_memtable.c pagestore_pgcache.c pagestore_prune.c pagestore_retention.c pagestore_wal_store.c pagestore_wal_segment.c pagestore_walidx_prune.c pagestore_walidx_snapshot.c pagestore_forkmeta_prune.c pagestore_forkmeta_snapshot.c -lrt -lpthread + cc -O2 -Wall -Wextra -Werror -o pagestore_tiering_test pagestore_tiering_test.c pagestore_core.c pagestore_fault.c storage_posix.c pagestore_layer.c pagestore_layer_store.c pagestore_manifest.c pagestore_memtable.c pagestore_pgcache.c pagestore_prune.c pagestore_retention.c pagestore_wal_store.c pagestore_wal_segment.c pagestore_walidx_prune.c pagestore_walidx_snapshot.c pagestore_forkmeta_prune.c pagestore_forkmeta_snapshot.c pagestore_store_owner.c -lrt -lpthread + cc -O2 -Wall -Wextra -Werror -o pagestore_forkmeta_cutover_test pagestore_forkmeta_cutover_test.c pagestore_core.c pagestore_fault.c storage_posix.c pagestore_layer.c pagestore_layer_store.c pagestore_manifest.c pagestore_memtable.c pagestore_pgcache.c pagestore_prune.c pagestore_retention.c pagestore_wal_store.c pagestore_wal_segment.c pagestore_walidx_prune.c pagestore_walidx_snapshot.c pagestore_forkmeta_prune.c pagestore_forkmeta_snapshot.c pagestore_store_owner.c -lrt -lpthread + cc -O2 -Wall -Wextra -Werror -o pagestore_forkmeta_crash_matrix_test pagestore_forkmeta_crash_matrix_test.c pagestore_core.c pagestore_fault.c storage_posix.c pagestore_layer.c pagestore_layer_store.c pagestore_manifest.c pagestore_memtable.c pagestore_pgcache.c pagestore_prune.c pagestore_retention.c pagestore_wal_store.c pagestore_wal_segment.c pagestore_walidx_prune.c pagestore_walidx_snapshot.c pagestore_forkmeta_prune.c pagestore_forkmeta_snapshot.c pagestore_store_owner.c -lrt -lpthread cc -O2 -Wall -Wextra -Werror -o pagestore_pgcache_test pagestore_pgcache_test.c pagestore_pgcache.c -lrt -lpthread - cc -O2 -Wall -Wextra -Werror -o pagestore_manifest_test pagestore_manifest_test.c pagestore_manifest.c pagestore_layer.c pagestore_layer_store.c pagestore_fault.c -lrt -pthread + cc -O2 -Wall -Wextra -Werror -o pagestore_manifest_test pagestore_manifest_test.c pagestore_manifest.c pagestore_layer.c pagestore_layer_store.c pagestore_fault.c pagestore_store_owner.c -lrt -pthread cc -O2 -Wall -Wextra -Werror -o pagestore_retention_test pagestore_retention_test.c pagestore_retention.c -lrt -lpthread - cc -O2 -Wall -Wextra -Werror -o pagestore_gc_test pagestore_gc_test.c pagestore_core.c pagestore_fault.c storage_posix.c pagestore_layer.c pagestore_layer_store.c pagestore_manifest.c pagestore_memtable.c pagestore_pgcache.c pagestore_prune.c pagestore_retention.c pagestore_wal_store.c pagestore_wal_segment.c pagestore_walidx_prune.c pagestore_walidx_snapshot.c pagestore_forkmeta_prune.c pagestore_forkmeta_snapshot.c -lrt -lpthread + cc -O2 -Wall -Wextra -Werror -o pagestore_gc_test pagestore_gc_test.c pagestore_core.c pagestore_fault.c storage_posix.c pagestore_layer.c pagestore_layer_store.c pagestore_manifest.c pagestore_memtable.c pagestore_pgcache.c pagestore_prune.c pagestore_retention.c pagestore_wal_store.c pagestore_wal_segment.c pagestore_walidx_prune.c pagestore_walidx_snapshot.c pagestore_forkmeta_prune.c pagestore_forkmeta_snapshot.c pagestore_store_owner.c -lrt -lpthread cc -O2 -Wall -Wextra -Werror -o pagestore_walidx_prune_test pagestore_walidx_prune_test.c pagestore_walidx_prune.c -lrt cc -O2 -Wall -Wextra -Werror -o pagestore_forkmeta_prune_test pagestore_forkmeta_prune_test.c pagestore_forkmeta_prune.c -lrt cc -O2 -Wall -Wextra -Werror -o pagestore_walidx_snapshot_test pagestore_walidx_snapshot_test.c pagestore_walidx_snapshot.c -lrt cc -O2 -Wall -Wextra -Werror -o pagestore_forkmeta_snapshot_test pagestore_forkmeta_snapshot_test.c pagestore_forkmeta_snapshot.c -lrt - cc -O2 -Wall -Wextra -Werror -o storage_posix_wal_test storage_posix_wal_test.c storage_posix.c -lrt -lpthread + cc -O2 -Wall -Wextra -Werror -o storage_posix_wal_test storage_posix_wal_test.c storage_posix.c pagestore_store_owner.c -lrt -lpthread + cc -O2 -Wall -Wextra -Werror -o storage_posix_fork_owner_test storage_posix_fork_owner_test.c storage_posix.c pagestore_store_owner.c pagestore_manifest.c pagestore_layer.c pagestore_layer_store.c pagestore_fault.c -lrt -lpthread cc -O2 -Wall -Wextra -Werror -o pagestore_fault_test pagestore_fault_test.c pagestore_fault.c -pthread - name: Run standalone test suite working-directory: contrib/pagestore run: ./pagestore_test ./pagestore_daemon ./pagestore_inspect + - name: Run store ownership and orphan recovery tests + working-directory: contrib/pagestore + run: | + ./storage_posix_fork_owner_test + python3 harness/tests/test_store_owner_recovery.py \ + --daemon-binary ./pagestore_daemon \ + --inspect-binary ./pagestore_inspect + - name: Run named fault registry unit test working-directory: contrib/pagestore run: ./pagestore_fault_test diff --git a/contrib/pagestore/MVP_COMPLETION_PLAN.md b/contrib/pagestore/MVP_COMPLETION_PLAN.md index 9867d81e75c6a..24fda75069200 100644 --- a/contrib/pagestore/MVP_COMPLETION_PLAN.md +++ b/contrib/pagestore/MVP_COMPLETION_PLAN.md @@ -809,6 +809,25 @@ Acceptance: Expected scope: one or two PRs. +### Local POSIX store ownership and orphan-layer recovery + +Store recovery and local provider mutations share an exclusive advisory store +lease. Startup may remove canonical, unreferenced local layer files only after +validating the manifest and the complete candidate namespace. Manifest-owned +IDs, including deleting and remote-only records, remain protected. Unknown +non-layer files and object-tier artifacts are not part of this sweep. + +Ambiguous manifest-tail repair must durably inhibit orphan sweeping before +truncating the manifest, and the inhibition persists across restart. Missing +manifest metadata does not authorize deletion. Recovery retries must preserve +referenced data after partial unlink or directory-sync failures. + +Acceptance includes competing owners with distinct SHM names, release after +process death, failed-open cleanup, canonical orphan reclamation, namespace +validation before deletion, and continued H1 sentinel recovery. This closes +local orphan cleanup only; the other H1 crash families and R6 space acceptance +remain separate gates. + ### H1. Compose process-level crash scenarios Status: **materializer replay/restartpoint, branch prepared-receipt/service- diff --git a/contrib/pagestore/MVP_STATUS.md b/contrib/pagestore/MVP_STATUS.md index e9507971b4298..718435c033317 100644 --- a/contrib/pagestore/MVP_STATUS.md +++ b/contrib/pagestore/MVP_STATUS.md @@ -86,9 +86,31 @@ harness keeps the pre-recovery physical snapshot, then checks a sentinel page/LSN, the expected manifest state, and one additional restart for idempotence. A crash after ADD but before its flush watermark conservatively retains the durable layer and republishes segment-backed coverage once; the -intervening clean shutdown compacts that conservative duplicate, and the second -restart proves convergence to one layer. The slice does not delete crash -orphans; cross-process ownership is required before adding that policy. +second restart waits for background maintenance to compact that conservative +duplicate to one layer. Clean shutdown alone does not guarantee compaction. + +POSIX store opens now hold an exclusive advisory ownership lease across recovery +and provider teardown. Cooperating storage and local-layer +provider users share the same ownership mechanism. After successful manifest +replay, startup reconciles canonical local layer files against manifest-owned +IDs, preserving referenced layers and removing validated unreferenced files. +Legacy relative, symlinked-directory, and dot-dot local URI spellings are +normalized in the replayed map when their parent resolves to the owned store +and their filename matches the layer ID. Unresolvable or foreign-store paths +still fail closed; this is not an arbitrary store-relocation mechanism. +An invalid layer namespace or unsafe file type fails closed before deletion; +unrelated files and object-tier contents are outside this reconciliation. +Missing manifests do not authorize a sweep. Before accepting an ambiguous +manifest-tail repair, recovery durably records an orphan-sweep inhibition +marker; automatic cleanup remains disabled across subsequent restarts because +the repaired manifest cannot prove that omitted files were never referenced. +The persistent lock file must not be removed while a store is in use. Older +binaries and external tools that do not acquire the lock must remain stopped +during recovery; the lock is advisory, not a fence against arbitrary filesystem +writes. This is local POSIX recovery, not SPDK or power-loss certification. +Child processes cannot mutate through inherited provider handles; a child that +inherits an open core must exec a fresh process before using the core. SPDK +storage retains its original caller-owned teardown contract. ## MVP gates diff --git a/contrib/pagestore/harness/tests/test_store_owner_recovery.py b/contrib/pagestore/harness/tests/test_store_owner_recovery.py new file mode 100644 index 0000000000000..c0c4368f304ed --- /dev/null +++ b/contrib/pagestore/harness/tests/test_store_owner_recovery.py @@ -0,0 +1,707 @@ +#!/usr/bin/env python3 +"""Standalone POSIX process-level acceptance tests for pagestore ownership/recovery. + +This file intentionally has no unittest test cases. It is invoked explicitly +with the two standalone binaries, for example: + + python3 contrib/pagestore/harness/tests/test_store_owner_recovery.py \ + --daemon-binary build/contrib/pagestore/pagestore_daemon \ + --inspect-binary build/contrib/pagestore/pagestore_inspect + +The scenarios use only the daemon and its read-only inspector; they do not +start PostgreSQL or SPDK. The temporary root is removed unless --keep is set. +""" + +from __future__ import annotations + +import argparse +import ctypes +import errno +import hashlib +import json +import os +from pathlib import Path +import signal +import struct +import subprocess +import sys +import tempfile +import time +from dataclasses import dataclass +import stat +from typing import Any + + +DEFAULT_TIMEOUT = 10.0 +PAGE_SIZE = 8192 +SHARD_COUNT = 1 +MANIFEST_MAGIC = 0x504D414E # "PMAN" +MANIFEST_VERSION = 3 +MANIFEST_REMOVE_LAYER = 5 +ORPHAN_ID = 0x0000000000004000 + + +class TestFailure(RuntimeError): + """An acceptance assertion failed.""" + + +def require(condition: bool, message: str) -> None: + if not condition: + raise TestFailure(message) + + +def private_environment() -> dict[str, str]: + """Keep unrelated cluster and pagestore settings out of the subprocesses.""" + return { + key: value + for key, value in os.environ.items() + if not key.startswith("PG") and not key.startswith("PAGESTORE_") + } + + +def shm_unlink(name: str) -> None: + """Remove a POSIX shm object without depending on /dev/shm layout.""" + libc = ctypes.CDLL(None, use_errno=True) + unlink = libc.shm_unlink + unlink.argtypes = [ctypes.c_char_p] + unlink.restype = ctypes.c_int + if unlink(name.encode("utf-8")) != 0: + error = ctypes.get_errno() + if error != errno.ENOENT: + raise OSError(error, os.strerror(error), name) + + +def next_shm_name() -> str: + # Linux limits POSIX shm names to NAME_MAX, so keep this deliberately short. + return f"/psowner_{os.getpid()}_{time.monotonic_ns() & 0xFFFFFFFF:x}" + + +def daemon_command(daemon: Path, shm: str, store: Path) -> list[str]: + return [ + str(daemon), + "--shm", + shm, + "--store", + str(store), + "--page-size", + str(PAGE_SIZE), + "--nshards", + str(SHARD_COUNT), + "--storage", + "posix", + ] + + +@dataclass +class DaemonProcess: + process: subprocess.Popen[str] + shm: str + log: Path + + def cleanup(self) -> None: + if self.process.poll() is None: + try: + os.kill(self.process.pid, signal.SIGKILL) + except ProcessLookupError: + pass + try: + self.process.wait(timeout=2.0) + except subprocess.TimeoutExpired: + pass + shm_unlink(self.shm) + + +def launch_daemon(daemon: Path, store: Path, log: Path) -> DaemonProcess: + shm = next_shm_name() + with log.open("ab") as stream: + process = subprocess.Popen( + daemon_command(daemon, shm, store), + stdout=stream, + stderr=subprocess.STDOUT, + env=private_environment(), + start_new_session=True, + ) + return DaemonProcess(process, shm, log) + + +def inspect_health(inspector: Path, shm: str, timeout: float) -> dict[str, Any] | None: + try: + result = subprocess.run( + [str(inspector), "--shm", shm, "health"], + stdout=subprocess.PIPE, + stderr=subprocess.PIPE, + text=True, + env=private_environment(), + timeout=max(0.05, timeout), + check=False, + ) + except (OSError, subprocess.TimeoutExpired): + return None + if result.returncode != 0: + return None + try: + value = json.loads(result.stdout) + except (ValueError, TypeError): + return None + return value if isinstance(value, dict) else None + + +def wait_ready( + process: DaemonProcess, inspector: Path, timeout: float, label: str +) -> dict[str, Any]: + deadline = time.monotonic() + timeout + last_health: dict[str, Any] | None = None + while True: + returncode = process.process.poll() + if returncode is not None: + log_tail = process.log.read_text(encoding="utf-8", errors="replace")[-2000:] + raise TestFailure( + f"{label}: daemon exited before health with code {returncode}\n{log_tail}" + ) + remaining = deadline - time.monotonic() + if remaining <= 0: + raise TestFailure(f"{label}: health deadline expired; last={last_health!r}") + last_health = inspect_health(inspector, process.shm, min(0.5, remaining)) + if last_health is not None: + require(last_health.get("page_size") == PAGE_SIZE, + f"{label}: unexpected page size {last_health!r}") + require(last_health.get("nshards") == SHARD_COUNT, + f"{label}: unexpected shard count {last_health!r}") + return last_health + time.sleep(min(0.01, remaining)) + + +def wait_exit(process: DaemonProcess, timeout: float, expected: int, label: str) -> None: + try: + result = process.process.wait(timeout=timeout) + except subprocess.TimeoutExpired as error: + raise TestFailure(f"{label}: did not exit before deadline") from error + require(result == expected, f"{label}: expected exit code {expected}, got {result}") + + +def stop_clean(process: DaemonProcess, timeout: float, label: str) -> None: + if process.process.poll() is not None: + require(process.process.returncode == 0, + f"{label}: daemon had unexpected exit {process.process.returncode}") + return + os.kill(process.process.pid, signal.SIGTERM) + wait_exit(process, timeout, 0, label) + + +def kill_owner(process: DaemonProcess, timeout: float, label: str) -> None: + os.kill(process.process.pid, signal.SIGKILL) + wait_exit(process, timeout, -signal.SIGKILL, label) + + +def assert_regular(path: Path, label: str) -> None: + require(path.exists() and not path.is_symlink() and path.is_file(), + f"{label}: expected regular file {path}") + + +def snapshot_tree(root: Path) -> dict[str, tuple[Any, ...]]: + """Capture file identity/content without following symlinks.""" + result: dict[str, tuple[Any, ...]] = {} + + def visit(directory: Path) -> None: + for entry in sorted(os.scandir(directory), key=lambda item: item.name): + path = Path(entry.path) + relative = str(path.relative_to(root)) + stat_result = path.lstat() + mode = stat_result.st_mode + if path.is_symlink(): + result[relative] = ("symlink", stat_result.st_mode & 0o7777, + os.readlink(path)) + elif path.is_dir(): + result[relative] = ("directory", stat_result.st_mode & 0o7777) + visit(path) + elif path.is_file(): + result[relative] = ("file", stat_result.st_mode & 0o7777, + hashlib.sha256(path.read_bytes()).digest(), + stat_result.st_size) + else: + result[relative] = ("other", stat_result.st_mode & 0o7777) + + visit(root) + return result + + +def assert_tree_unchanged(before: dict[str, tuple[Any, ...]], root: Path, label: str) -> None: + after = snapshot_tree(root) + require(after == before, f"{label}: store files changed\nbefore={before!r}\nafter={after!r}") + + +def layer_path(store: Path, layer_id: int) -> Path: + shard = (layer_id >> 48) & 0xFFFF + return store / f"layer_{shard}_{layer_id:016x}" + + +def seed_store(daemon: Path, inspector: Path, store: Path, log: Path, timeout: float) -> None: + process = launch_daemon(daemon, store, log) + try: + wait_ready(process, inspector, timeout, "seed store") + lock = store / ".pagestore.lock" + assert_regular(lock, "seed store") + stop_clean(process, timeout, "seed store shutdown") + finally: + process.cleanup() + + +def test_duplicate_shm_does_not_touch_store( + daemon: Path, inspector: Path, root: Path, timeout: float +) -> None: + store = root / "duplicate-store" + store.mkdir() + first = launch_daemon(daemon, store, root / "duplicate-first.log") + second: DaemonProcess | None = None + try: + wait_ready(first, inspector, timeout, "duplicate owner") + lock = store / ".pagestore.lock" + assert_regular(lock, "duplicate owner") + before = snapshot_tree(store) + + second = launch_daemon(daemon, store, root / "duplicate-second.log") + wait_exit(second, timeout, 1, "duplicate owner") + require(inspect_health(inspector, first.shm, timeout) is not None, + "duplicate owner: first daemon lost health") + assert_tree_unchanged(before, store, "duplicate owner") + finally: + if second is not None: + second.cleanup() + first.cleanup() + + +def test_sigkill_releases_owner( + daemon: Path, inspector: Path, root: Path, timeout: float +) -> None: + store = root / "sigkill-store" + store.mkdir() + first = launch_daemon(daemon, store, root / "sigkill-first.log") + second: DaemonProcess | None = None + try: + wait_ready(first, inspector, timeout, "SIGKILL owner") + kill_owner(first, timeout, "SIGKILL owner") + second = launch_daemon(daemon, store, root / "sigkill-second.log") + wait_ready(second, inspector, timeout, "post-SIGKILL restart") + stop_clean(second, timeout, "post-SIGKILL shutdown") + finally: + if second is not None: + second.cleanup() + first.cleanup() + + +def test_empty_orphan_is_removed( + daemon: Path, inspector: Path, root: Path, timeout: float +) -> None: + store = root / "orphan-store" + store.mkdir() + seed_store(daemon, inspector, store, root / "orphan-seed.log", timeout) + establish_clean_manifest( + daemon, inspector, store, root / "orphan-clean-manifest.log", timeout + ) + orphan = layer_path(store, ORPHAN_ID) + unrelated = store / "unrelated-preserved.bin" + orphan.touch() + unrelated.write_bytes(b"do not remove\n") + + process = launch_daemon(daemon, store, root / "orphan-restart.log") + try: + wait_ready(process, inspector, timeout, "orphan cleanup restart") + require(not orphan.exists() and not orphan.is_symlink(), + f"orphan cleanup restart: empty orphan survived: {orphan}") + require(unrelated.read_bytes() == b"do not remove\n", + "orphan cleanup restart: unrelated file changed") + stop_clean(process, timeout, "orphan cleanup shutdown") + finally: + process.cleanup() + + +def test_missing_manifest_skips_sweep( + daemon: Path, inspector: Path, root: Path, timeout: float +) -> None: + store = root / "missing-manifest-store" + store.mkdir() + seed_store(daemon, inspector, store, root / "missing-seed.log", timeout) + manifest = store / "layers.manifest" + if manifest.exists() or manifest.is_symlink(): + manifest.unlink() + orphan = layer_path(store, ORPHAN_ID) + orphan.touch() + + process = launch_daemon(daemon, store, root / "missing-manifest.log") + try: + wait_ready(process, inspector, timeout, "missing manifest") + require(orphan.exists() and not orphan.is_symlink() and orphan.stat().st_size == 0, + "missing manifest: orphan was swept") + stop_clean(process, timeout, "missing manifest shutdown") + finally: + process.cleanup() + + +def test_unsafe_layer_entries_fail_closed( + daemon: Path, inspector: Path, root: Path, timeout: float +) -> None: + for case, make_unsafe in (("malformed", _make_malformed), ("symlink", _make_symlink)): + store = root / f"unsafe-{case}-store" + store.mkdir() + seed_store(daemon, inspector, store, root / f"unsafe-{case}-seed.log", timeout) + establish_clean_manifest( + daemon, inspector, store, root / f"unsafe-{case}-clean-manifest.log", timeout + ) + orphan = layer_path(store, ORPHAN_ID) + orphan.touch() + keep = store / "unrelated-preserved.bin" + keep.write_bytes(b"preserve\n") + unsafe = make_unsafe(store) + outside = store.parent / f"{case}-outside-target" + if outside.exists(): + outside.unlink() + if case == "symlink": + outside.write_bytes(b"outside target\n") + unsafe.symlink_to(outside) + + process = launch_daemon(daemon, store, root / f"unsafe-{case}.log") + try: + wait_exit(process, timeout, 1, f"unsafe {case} layer") + require(orphan.exists() and not orphan.is_symlink() and orphan.stat().st_size == 0, + f"unsafe {case} layer: valid orphan was removed") + require(keep.read_bytes() == b"preserve\n", + f"unsafe {case} layer: unrelated file changed") + require(unsafe.is_symlink() if case == "symlink" else unsafe.exists(), + f"unsafe {case} layer: unsafe entry disappeared") + finally: + process.cleanup() + + +def test_unsafe_copy_temporaries_fail_closed( + daemon: Path, inspector: Path, root: Path, timeout: float +) -> None: + exited = subprocess.Popen([sys.executable, "-c", "pass"], env=private_environment()) + require(exited.wait(timeout=timeout) == 0, "prepare exited copy-owner PID") + for label, pid in (("live", os.getpid()), ("exited", exited.pid)): + for kind in ("directory", "symlink", "hardlink"): + case = f"copy-{label}-{kind}" + store = root / f"{case}-store" + store.mkdir() + seed_store(daemon, inspector, store, root / f"{case}-seed.log", timeout) + establish_clean_manifest( + daemon, inspector, store, root / f"{case}-manifest.log", timeout + ) + orphan = layer_path(store, ORPHAN_ID) + orphan.touch() + unsafe = store / f"{layer_path(store, ORPHAN_ID + 1).name}.tmp.{pid}.1" + outside = root / f"{case}-outside" + outside.write_bytes(b"outside bytes\n") + if kind == "directory": + unsafe.mkdir() + elif kind == "symlink": + unsafe.symlink_to(outside) + else: + os.link(outside, unsafe) + before = snapshot_tree(store) + process = launch_daemon(daemon, store, root / f"{case}.log") + try: + wait_exit(process, timeout, 1, case) + assert_tree_unchanged(before, store, case) + require(outside.read_bytes() == b"outside bytes\n", + f"{case}: outside target changed") + finally: + process.cleanup() + + +def test_malformed_inhibition_markers_fail_closed( + daemon: Path, inspector: Path, root: Path, timeout: float +) -> None: + for case in ("symlink", "fifo", "nonempty"): + store = root / f"malformed-marker-{case}-store" + store.mkdir() + seed_store(daemon, inspector, store, root / f"marker-{case}-seed.log", timeout) + establish_clean_manifest( + daemon, inspector, store, root / f"marker-{case}-clean-manifest.log", timeout + ) + orphan = layer_path(store, ORPHAN_ID) + orphan.touch() + marker = store / ".pagestore-orphan-sweep-inhibited" + marker_target = store.parent / f"marker-{case}-target" + if case == "symlink": + marker_target.write_bytes(b"marker target\n") + marker.symlink_to(marker_target) + elif case == "fifo": + os.mkfifo(marker, 0o600) + else: + marker.write_bytes(b"not an empty marker\n") + + process = launch_daemon(daemon, store, root / f"marker-{case}.log") + try: + wait_exit(process, timeout, 1, f"malformed inhibition marker {case}") + require(orphan.exists() and not orphan.is_symlink() and orphan.stat().st_size == 0, + f"malformed inhibition marker {case}: orphan was removed") + if case == "symlink": + require(marker.is_symlink() and marker.resolve() == marker_target.resolve(), + "malformed inhibition marker symlink: marker changed") + require(marker_target.read_bytes() == b"marker target\n", + "malformed inhibition marker symlink: target changed") + elif case == "fifo": + require(stat.S_ISFIFO(marker.lstat().st_mode), + "malformed inhibition marker FIFO: marker changed") + else: + require(marker.read_bytes() == b"not an empty marker\n", + "malformed inhibition marker nonempty: marker changed") + finally: + process.cleanup() + + +def test_manifest_aliases_fail_closed( + daemon: Path, inspector: Path, root: Path, timeout: float +) -> None: + for case in ("symlink", "fifo"): + store = root / f"manifest-alias-{case}-store" + store.mkdir() + seed_store(daemon, inspector, store, root / f"manifest-{case}-seed.log", timeout) + establish_clean_manifest( + daemon, inspector, store, root / f"manifest-{case}-clean-manifest.log", timeout + ) + orphan = layer_path(store, ORPHAN_ID) + orphan.touch() + manifest = store / "layers.manifest" + manifest.unlink() + if case == "symlink": + target = store.parent / "manifest-symlink-target" + target.write_bytes( + manifest_record(MANIFEST_REMOVE_LAYER, struct.pack(" Path: + path = store / "layer_0_not-a-hex-id" + path.write_bytes(b"malformed layer name\n") + return path + + +def _make_symlink(store: Path) -> Path: + return store / "layer_0_0000000000005000" + + +def fnv1a(data: bytes) -> int: + value = 2166136261 + for byte in data: + value = ((value ^ byte) * 16777619) & 0xFFFFFFFF + return value + + +def manifest_record( + event_type: int, payload: bytes, corrupt_crc: bool = False +) -> bytes: + header_without_crc = struct.pack( + " None: + """Make a valid manifest and prove the daemon can replay it cleanly.""" + manifest = store / "layers.manifest" + manifest.write_bytes( + manifest_record(MANIFEST_REMOVE_LAYER, struct.pack(" None: + store = root / "corrupt-manifest-store" + store.mkdir() + seed_store(daemon, inspector, store, root / "corrupt-seed.log", timeout) + orphan = layer_path(store, ORPHAN_ID) + orphan.touch() + + # A valid v3 no-op record anchors the file format. The second complete + # record has a bad CRC, so replay must truncate only that final record. + valid = manifest_record(MANIFEST_REMOVE_LAYER, struct.pack(" None: + store = root / "interior-corrupt-manifest-store" + store.mkdir() + seed_store(daemon, inspector, store, root / "interior-seed.log", timeout) + orphan = layer_path(store, ORPHAN_ID) + orphan.touch() + valid = manifest_record(MANIFEST_REMOVE_LAYER, struct.pack(" None: + store = root / "torn-first-manifest-store" + store.mkdir() + seed_store(daemon, inspector, store, root / "torn-first-seed.log", timeout) + orphan = layer_path(store, ORPHAN_ID) + orphan.touch() + valid = manifest_record(MANIFEST_REMOVE_LAYER, struct.pack(" None: + daemon = Path(args.daemon_binary).resolve() + inspector = Path(args.inspect_binary).resolve() + require(daemon.is_file() and os.access(daemon, os.X_OK), + f"daemon binary is not executable: {daemon}") + require(inspector.is_file() and os.access(inspector, os.X_OK), + f"inspector binary is not executable: {inspector}") + require(args.timeout > 0, "--timeout must be positive") + + root = Path(tempfile.mkdtemp(prefix="pagestore-owner-recovery-", dir=args.work_dir)) + completed = False + try: + tests = ( + ("duplicate shm ownership", test_duplicate_shm_does_not_touch_store), + ("SIGKILL owner release", test_sigkill_releases_owner), + ("empty orphan cleanup", test_empty_orphan_is_removed), + ("missing manifest skips sweep", test_missing_manifest_skips_sweep), + ("unsafe layer fail-closed", test_unsafe_layer_entries_fail_closed), + ("unsafe copy temporaries fail-closed", test_unsafe_copy_temporaries_fail_closed), + ("malformed inhibition markers fail-closed", test_malformed_inhibition_markers_fail_closed), + ("manifest aliases fail-closed", test_manifest_aliases_fail_closed), + ("corrupt manifest tail safety", test_corrupt_manifest_tail_does_not_delete_orphan), + ("interior manifest fail-closed", test_interior_manifest_corruption_fails_closed), + ("torn first manifest quarantine", test_torn_first_manifest_record_stays_quarantined), + ) + for label, test in tests: + test(daemon, inspector, root, args.timeout) + print(f"ok - {label}") + completed = True + finally: + if args.keep or not completed: + destination = "temporary root preserved" if not completed else "temporary root" + print(f"{destination}: {root}", file=sys.stderr if not completed else sys.stdout) + else: + import shutil + + shutil.rmtree(root, ignore_errors=True) + + +def main(argv: list[str] | None = None) -> int: + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("--daemon-binary", required=True, + help="path to pagestore_daemon") + parser.add_argument("--inspect-binary", required=True, + help="path to pagestore_inspect") + parser.add_argument("--timeout", type=float, default=DEFAULT_TIMEOUT, + help=f"per-process deadline in seconds (default: {DEFAULT_TIMEOUT:g})") + parser.add_argument("--work-dir", type=Path, default=None, + help="parent directory for temporary test data") + parser.add_argument("--keep", action="store_true", + help="retain the temporary root after the run") + args = parser.parse_args(argv) + try: + run(args) + except (OSError, TestFailure) as error: + print(f"not ok - {error}", file=sys.stderr) + return 1 + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/contrib/pagestore/meson.build b/contrib/pagestore/meson.build index 0b728f0f35028..1ffbf118ed765 100644 --- a/contrib/pagestore/meson.build +++ b/contrib/pagestore/meson.build @@ -70,7 +70,7 @@ endif pagestore_daemon = executable('pagestore_daemon', files('pagestore_daemon.c', 'pagestore_core.c', 'storage_posix.c', 'pagestore_fault.c', - 'pagestore_layer.c', 'pagestore_layer_store.c', 'pagestore_manifest.c', + 'pagestore_layer.c', 'pagestore_store_owner.c', 'pagestore_layer_store.c', 'pagestore_manifest.c', 'pagestore_memtable.c', 'pagestore_pgcache.c', 'pagestore_prune.c', 'pagestore_retention.c', 'pagestore_wal_store.c', 'pagestore_wal_segment.c', 'pagestore_walidx_prune.c', @@ -184,7 +184,7 @@ test('pagestore_inspect_mailbox', # Unit test for the immutable image-layer file format (no daemon, no PG). pagestore_layer_test = executable('pagestore_layer_test', - files('pagestore_layer_test.c', 'pagestore_layer.c', 'pagestore_layer_store.c', + files('pagestore_layer_test.c', 'pagestore_layer.c', 'pagestore_store_owner.c', 'pagestore_layer_store.c', 'pagestore_fault.c'), dependencies: [dependency('threads')], install: false, @@ -197,7 +197,7 @@ test('pagestore_layer', # Unit test for the filesystem-backed remote object provider (no daemon, no PG). pagestore_layer_store_test = executable('pagestore_layer_store_test', - files('pagestore_layer_store_test.c', 'pagestore_layer_store.c', 'pagestore_layer.c', + files('pagestore_layer_store_test.c', 'pagestore_store_owner.c', 'pagestore_layer_store.c', 'pagestore_layer.c', 'pagestore_fault.c'), dependencies: [dependency('threads')], install: false, @@ -211,7 +211,7 @@ test('pagestore_layer_store', # Integration test for idle layer upload and durable remote manifest state. pagestore_tiering_test = executable('pagestore_tiering_test', files('pagestore_tiering_test.c', 'pagestore_core.c', 'pagestore_fault.c', 'storage_posix.c', - 'pagestore_layer.c', 'pagestore_layer_store.c', 'pagestore_manifest.c', + 'pagestore_layer.c', 'pagestore_store_owner.c', 'pagestore_layer_store.c', 'pagestore_manifest.c', 'pagestore_memtable.c', 'pagestore_pgcache.c', 'pagestore_prune.c', 'pagestore_retention.c', 'pagestore_wal_store.c', 'pagestore_wal_segment.c', 'pagestore_walidx_prune.c', @@ -231,7 +231,7 @@ test('pagestore_tiering', pagestore_forkmeta_cutover_test = executable('pagestore_forkmeta_cutover_test', files('pagestore_forkmeta_cutover_test.c', 'pagestore_core.c', 'pagestore_fault.c', 'storage_posix.c', 'pagestore_layer.c', - 'pagestore_layer_store.c', 'pagestore_manifest.c', + 'pagestore_store_owner.c', 'pagestore_layer_store.c', 'pagestore_manifest.c', 'pagestore_memtable.c', 'pagestore_pgcache.c', 'pagestore_prune.c', 'pagestore_retention.c', 'pagestore_wal_store.c', 'pagestore_wal_segment.c', 'pagestore_walidx_prune.c', @@ -252,7 +252,7 @@ test('pagestore_forkmeta_cutover', pagestore_forkmeta_crash_matrix_test = executable('pagestore_forkmeta_crash_matrix_test', files('pagestore_forkmeta_crash_matrix_test.c', 'pagestore_core.c', 'pagestore_fault.c', 'storage_posix.c', 'pagestore_layer.c', - 'pagestore_layer_store.c', 'pagestore_manifest.c', + 'pagestore_store_owner.c', 'pagestore_layer_store.c', 'pagestore_manifest.c', 'pagestore_memtable.c', 'pagestore_pgcache.c', 'pagestore_prune.c', 'pagestore_retention.c', 'pagestore_wal_store.c', 'pagestore_wal_segment.c', 'pagestore_walidx_prune.c', @@ -273,7 +273,7 @@ test('pagestore_forkmeta_crash_matrix', pagestore_timeline_test = executable('pagestore_timeline_test', files('pagestore_timeline_test.c', 'pagestore_core.c', 'pagestore_fault.c', 'storage_posix.c', 'pagestore_layer.c', - 'pagestore_layer_store.c', 'pagestore_manifest.c', + 'pagestore_store_owner.c', 'pagestore_layer_store.c', 'pagestore_manifest.c', 'pagestore_memtable.c', 'pagestore_pgcache.c', 'pagestore_prune.c', 'pagestore_retention.c', 'pagestore_wal_store.c', 'pagestore_wal_segment.c', 'pagestore_walidx_prune.c', @@ -293,7 +293,7 @@ test('pagestore_timeline', pagestore_wal_reclaim_core_test = executable('pagestore_wal_reclaim_core_test', files('pagestore_wal_reclaim_core_test.c', 'pagestore_core.c', 'pagestore_fault.c', 'storage_posix.c', 'pagestore_layer.c', - 'pagestore_layer_store.c', 'pagestore_manifest.c', + 'pagestore_store_owner.c', 'pagestore_layer_store.c', 'pagestore_manifest.c', 'pagestore_memtable.c', 'pagestore_pgcache.c', 'pagestore_prune.c', 'pagestore_retention.c', 'pagestore_wal_store.c', 'pagestore_wal_segment.c', 'pagestore_walidx_prune.c', @@ -312,7 +312,7 @@ test('pagestore_wal_reclaim_core', pagestore_backpressure_test = executable('pagestore_backpressure_test', files('pagestore_backpressure_test.c', 'pagestore_core.c', 'pagestore_fault.c', 'storage_posix.c', 'pagestore_layer.c', - 'pagestore_layer_store.c', 'pagestore_manifest.c', + 'pagestore_store_owner.c', 'pagestore_layer_store.c', 'pagestore_manifest.c', 'pagestore_memtable.c', 'pagestore_pgcache.c', 'pagestore_prune.c', 'pagestore_retention.c', 'pagestore_wal_store.c', 'pagestore_wal_segment.c', 'pagestore_walidx_prune.c', @@ -346,7 +346,7 @@ test('pagestore_backpressure_daemon', # Unit test for the durable layer manifest replay (no daemon, no PG). pagestore_manifest_test = executable('pagestore_manifest_test', files('pagestore_manifest_test.c', 'pagestore_manifest.c', 'pagestore_layer.c', - 'pagestore_layer_store.c', 'pagestore_fault.c'), + 'pagestore_store_owner.c', 'pagestore_layer_store.c', 'pagestore_fault.c'), dependencies: [dependency('threads')], install: false, ) @@ -371,7 +371,7 @@ test('pagestore_retention', # Unit test for crash-safe compaction/GC recovery (no daemon, no PG). pagestore_gc_test = executable('pagestore_gc_test', files('pagestore_gc_test.c', 'pagestore_manifest.c', 'pagestore_layer.c', - 'pagestore_layer_store.c', 'pagestore_fault.c'), + 'pagestore_store_owner.c', 'pagestore_layer_store.c', 'pagestore_fault.c'), dependencies: [dependency('threads')], install: false, ) @@ -439,7 +439,7 @@ test('pagestore_wal_store', # Crash-atomic prefix replacement for the authoritative flat shipped-WAL log. storage_posix_wal_test = executable('storage_posix_wal_test', - files('storage_posix_wal_test.c', 'storage_posix.c'), + files('storage_posix_wal_test.c', 'storage_posix.c', 'pagestore_store_owner.c'), dependencies: [dependency('threads')], install: false, ) @@ -449,6 +449,19 @@ test('storage_posix_wal', timeout: 30, ) +storage_posix_fork_owner_test = executable('storage_posix_fork_owner_test', + files('storage_posix_fork_owner_test.c', 'storage_posix.c', + 'pagestore_store_owner.c', 'pagestore_manifest.c', 'pagestore_layer.c', + 'pagestore_layer_store.c', 'pagestore_fault.c'), + dependencies: [dependency('threads')], + install: false, +) +test('storage_posix_fork_owner', + storage_posix_fork_owner_test, + suite: ['pagestore'], + timeout: 30, +) + # Crash-atomic multi-shard WAL-index snapshot generation publication. pagestore_walidx_snapshot_test = executable('pagestore_walidx_snapshot_test', files('pagestore_walidx_snapshot_test.c', 'pagestore_walidx_snapshot.c'), @@ -488,6 +501,15 @@ test('pagestore_pgcache', # the regular pagestore suite so scenario/capability changes fail before the # slower PostgreSQL integration coverage runs. pagestore_harness_python = find_program('python3') +test('pagestore_store_owner_recovery', + pagestore_harness_python, + args: [files('harness/tests/test_store_owner_recovery.py'), + '--daemon-binary', pagestore_daemon, + '--inspect-binary', pagestore_inspect], + depends: [pagestore_daemon, pagestore_inspect], + suite: ['pagestore'], + timeout: 120, +) test('pagestore_harness_plan', pagestore_harness_python, args: [files('harness/tests/test_plan.py')], diff --git a/contrib/pagestore/pagestore_core.c b/contrib/pagestore/pagestore_core.c index ff3fb97e4e522..bf4ea649666f2 100644 --- a/contrib/pagestore/pagestore_core.c +++ b/contrib/pagestore/pagestore_core.c @@ -93,6 +93,32 @@ static int tier_one_layer(void); static int finish_upload(const PsLayerDesc *candidate); static int map_locks_ready; static int core_opened; +static pid_t core_pid; + +/* A fork inherits mutexes and buffered mutations, not a usable core instance. + * Check before taking any core lock or flushing inherited state. */ +static int +core_process_valid(void) +{ + pid_t pid = __atomic_load_n(&core_pid, __ATOMIC_ACQUIRE); + + if (pid != 0 && pid != getpid()) + { + errno = ECHILD; + return 0; + } + return 1; +} + +/* POSIX close is idempotent and participates in this PR's store lease. + * Other providers retain their existing caller-owned teardown contract. */ +static void +core_close_posix_storage(void) +{ + if (ps_storage != NULL && ps_storage->name != NULL && + strcmp(ps_storage->name, "posix") == 0 && ps_storage->close != NULL) + ps_storage->close(); +} /* Every snapshot writer is serialized here. Mutations refresh at their next * lock-safe completion point. Maintenance may also attempt an opportunistic * fallback between work items, with 100ms as a minimum spacing between such @@ -109,7 +135,7 @@ static int inspection_timeline_cache_retention_usable; static uint64_t inspection_timeline_cache_retention_epoch; static PsInspectionTimeline inspection_timeline_cache[PS_INSPECTION_MAX_TIMELINES]; static int fork_meta_reclaim_baseline_init(void); -static int ps_core_open_impl(const char *store_dir); +static int ps_core_open_impl(const char *store_dir, int *storage_opened); static void ps_core_close_impl(void); static const PsLayerLocation *tier_local_location(const PsLayerDesc *layer); static int refresh_remote_only_layer(const PsLayerDesc *layer); @@ -4742,8 +4768,11 @@ int fork_grow(uint32_t timeline, const PsKey *key, uint32_t to_nblocks, uint64_t lsn) { - uint64_t admission_seq = admission_seq_alloc(); + uint64_t admission_seq; + if (!core_process_valid()) + return -1; + admission_seq = admission_seq_alloc(); if (admission_seq == 0) return -1; return fork_grow_with_seq(timeline, key, to_nblocks, lsn, admission_seq); @@ -5369,8 +5398,11 @@ PageVer * read_through(uint32_t timeline, const PsKey *key, uint32_t block, uint64_t read_lsn, uint64_t read_seq) { - TlWalk w = tl_walk_first(timeline, read_lsn); + TlWalk w; + if (!core_process_valid()) + return NULL; + w = tl_walk_first(timeline, read_lsn); do { ForkEnt *fe = fork_find(w.tl, key); @@ -12312,16 +12344,22 @@ append_page(uint32_t timeline, const PsKey *key, uint32_t block, uint64_t hdr_grow_lsn = 0; uint64_t order_id = 0; uint64_t page_version; - uint64_t admission_seq = admission_seq_alloc(); + uint64_t admission_seq; int clamped = 0; int ordered_record = 0; int segment_grows = 0; int zero_version = 0; - Shard *s = shard_for(key); - ForkEnt *fe = fork_find(timeline, key); + Shard *s; + ForkEnt *fe; uint64_t branch_floor = 0; - uint64_t growth_floor = fe ? fe->last_def_lsn : 0; + uint64_t growth_floor; + if (!core_process_valid()) + return -1; + admission_seq = admission_seq_alloc(); + s = shard_for(key); + fe = fork_find(timeline, key); + growth_floor = fe ? fe->last_def_lsn : 0; if (admission_seq == 0) return -1; @@ -12561,6 +12599,8 @@ append_page(uint32_t timeline, const PsKey *key, uint32_t block, int read_version(const PageVer *v, unsigned char *out) { + if (!core_process_valid()) + return -1; if (v->seg < 0) /* layer-origin version (no segment copy) */ return -1; if (ps_storage->seg_read(v->shard, v->seg, v->off, out, page_size) != 0) @@ -12701,11 +12741,15 @@ read_resolve(uint32_t timeline, const PsKey *key, uint32_t block, uint64_t read_lsn, uint64_t read_seq, unsigned char *out, uint64_t *out_ver) { - Shard *s = shard_for(key); /* same shard across the ancestry walk */ + Shard *s; /* same shard across the ancestry walk */ TlWalk walk[MAX_TIMELINES]; uint32_t levels = 0; - TlWalk w = tl_walk_first(timeline, read_lsn); + TlWalk w; + if (!core_process_valid()) + return -1; + s = shard_for(key); + w = tl_walk_first(timeline, read_lsn); /* * A durable compaction frontier makes older page history unavailable even * while a crash-recovery pass still has its source layers to clean up. @@ -14363,6 +14407,11 @@ ps_handle_meta(PsChannel *ch) { uint32_t tl = ch->timeline; + if (!core_process_valid()) + { + ch->status = PS_STATUS_ERROR; + return 1; + } if (!timeline_op_allowed(tl, (PsOpcode) ch->opcode, ch->incarnation)) { ch->status = PS_STATUS_ERROR; @@ -15090,8 +15139,14 @@ ps_handle_meta(PsChannel *ch) void ps_core_close(void) { + if (!core_process_valid()) + return; pthread_mutex_lock(&core_state_lock); - ps_core_close_impl(); + /* Failed startup already unwinds providers, and a prior close released + * their leases. Neither state may enter the flushing shutdown path. */ + if (__atomic_load_n(&core_opened, __ATOMIC_ACQUIRE)) + ps_core_close_impl(); + __atomic_store_n(&core_pid, 0, __ATOMIC_RELEASE); pthread_mutex_unlock(&core_state_lock); } @@ -15210,6 +15265,11 @@ ps_core_close_impl(void) * streams before the storage provider is closed or the store is reopened. */ ps_forkmeta_snapshot_gc_reset(); ps_manifest_close(); + /* Release local provider leases after all core users have stopped. SPDK + * storage is still closed by its daemon; it is not an idempotent provider. */ + if (ps_layer_store != NULL && ps_layer_store->close != NULL) + ps_layer_store->close(); + core_close_posix_storage(); free_page_fork_indexes(); free_walidx_indexes(); } @@ -16179,6 +16239,8 @@ ps_core_maintenance(void) { int did; + if (!core_process_valid()) + return -1; /* Keep lifecycle-rd across the complete synchronous call. Any asynchronous * worker started within it reserves an additional reader before create and * releases that reservation from its thread cleanup handler. */ @@ -16215,15 +16277,37 @@ int ps_core_open(const char *store_dir) { int rc; + int save_errno; + int storage_opened = 0; + if (!core_process_valid()) + return -1; pthread_mutex_lock(&core_state_lock); - rc = ps_core_open_impl(store_dir); + __atomic_store_n(&core_pid, getpid(), __ATOMIC_RELEASE); + rc = ps_core_open_impl(store_dir, &storage_opened); + if (rc != 0) + { + /* Provider opens own the store lease. Unwind all lifecycle refs on every + * startup failure, including failures after manifest replay begins. */ + save_errno = errno; + __atomic_store_n(&core_opened, 0, __ATOMIC_RELEASE); + ps_manifest_close(); + if (ps_layer_store != NULL && ps_layer_store->close != NULL) + ps_layer_store->close(); + /* A fully initialized provider may use its ordinary close, including + * SPDK after a later replay failure. A failed provider open must unwind + * itself: its normal close may publish uninitialized persistent state. */ + if (storage_opened && ps_storage->close != NULL) + ps_storage->close(); + __atomic_store_n(&core_pid, 0, __ATOMIC_RELEASE); + errno = save_errno; + } pthread_mutex_unlock(&core_state_lock); return rc; } static int -ps_core_open_impl(const char *store_dir) +ps_core_open_impl(const char *store_dir, int *storage_opened) { uint32_t ns = core_shards(); int publish_shard_count = 0; @@ -16387,6 +16471,7 @@ ps_core_open_impl(const char *store_dir) return -1; if (ps_storage->open(runtime_store_dir, segment_size) != 0) return -1; + *storage_opened = 1; memcpy(wal_segment_root, next_wal_segment_root, strlen(next_wal_segment_root) + 1); memcpy(fork_meta_snapshot_dir, next_fork_meta_snapshot_dir, @@ -16401,6 +16486,26 @@ ps_core_open_impl(const char *store_dir) if (validate_store_shard_count(runtime_store_dir, &publish_shard_count) != 0) return -1; + if (use_layers && ps_layer_store->validate_local_layers != NULL && + ps_layer_store->validate_local_layers(&ps_layer_map) != 0) + return -1; + if (use_layers && mark_legacy_shard_zero_layers() != 0) + return -1; + /* The map is now a complete, shard-compatible replay result. A tolerated + * manifest tail repair is deliberately not authority for destructive orphan + * cleanup; its durable quarantine marker suppresses this and all future + * sweeps until an operator/repair workflow removes the ambiguity. */ + if (use_layers && ps_layer_store->recover_local_layers != NULL) + { + int sweep_inhibited = ps_manifest_orphan_sweep_inhibited(); + + if (sweep_inhibited < 0) + return -1; + if (ps_manifest_replay_had_manifest() && + !ps_manifest_replay_repaired() && !sweep_inhibited && + ps_layer_store->recover_local_layers(&ps_layer_map) != 0) + return -1; + } /* Leave deleting layers for asynchronous maintenance: recovery must not * block on an unavailable remote object that is already excluded from reads. */ @@ -16445,9 +16550,6 @@ ps_core_open_impl(const char *store_dir) if (sh < ns && lid + 1 > g_shards[sh].next_layer_id) g_shards[sh].next_layer_id = lid + 1; } - if (use_layers && mark_legacy_shard_zero_layers() != 0) - return -1; - /* the LSM write side (memtable/flush/compaction) runs only when layers are * the read path; the SPDK daemon stays on the segment path for now. One * memtable per shard. */ diff --git a/contrib/pagestore/pagestore_core.h b/contrib/pagestore/pagestore_core.h index 08d482ec4b5f6..6969415faa520 100644 --- a/contrib/pagestore/pagestore_core.h +++ b/contrib/pagestore/pagestore_core.h @@ -55,7 +55,11 @@ extern uint64_t forkmeta_reclaim_catchup_bytes; extern const PsStorage *ps_storage; extern uint32_t ps_nshards; /* logical shards configured for this daemon */ -/* Open the store and rebuild all in-memory state (timelines, indexes, WAL). */ +/* Open the store and rebuild all in-memory state (timelines, indexes, WAL). + * A core inherited while open across fork is unusable in the child; exec a + * fresh process instead of reopening or flushing inherited mutex/buffer state. + * On success non-POSIX storage retains caller-owned close. On failure core + * closes only providers whose open completed; failed opens clean themselves. */ extern int ps_core_open(const char *store_dir); /* Clean-shutdown: flush the memtable into a layer and close the manifest. */ diff --git a/contrib/pagestore/pagestore_layer_store.c b/contrib/pagestore/pagestore_layer_store.c index eafee14846dbf..8839202132205 100644 --- a/contrib/pagestore/pagestore_layer_store.c +++ b/contrib/pagestore/pagestore_layer_store.c @@ -12,6 +12,7 @@ #include #include #include +#include #include #include #include @@ -21,6 +22,7 @@ #include "pagestore_fault.h" #include "pagestore_layer_store.h" +#include "pagestore_store_owner.h" static uint32_t layer_page_size = PS_DEFAULT_PAGE_SIZE; @@ -34,6 +36,7 @@ ps_layer_store_set_page_size(uint32_t value) * one; keep provider roots large enough for canonicalized object directories. */ static char layer_dir[4096]; static char object_dir[4096]; +static PsStoreOwner *layer_owner; static int layer_id_shard(uint64_t layer_id); static int fsync_dir(const char *dir); @@ -41,6 +44,16 @@ static int cleanup_stale_copy_temps(const char *dir); static const PsLayerLocation *remote_location(const PsLayerDesc *layer); static int local_download_layer(const PsLayerDesc *layer); static int local_refresh_layer_cache(const PsLayerDesc *layer); +static void local_close(void); + +typedef struct LocalLayerCandidate +{ + char name[NAME_MAX + 1]; + uint64_t layer_id; + dev_t dev; + ino_t ino; + off_t size; +} LocalLayerCandidate; /* * Object directories are deliberately single-store resources. Layer IDs are @@ -192,16 +205,35 @@ local_open(const char *store_dir) struct stat store_st; struct stat object_st; int n; + int save_errno; char probe[PS_LAYER_URI_MAX]; - n = snprintf(layer_dir, sizeof(layer_dir), "%s", store_dir); - if (n < 0 || (size_t) n >= sizeof(layer_dir)) - return -1; - if (stat(layer_dir, &store_st) != 0 || !S_ISDIR(store_st.st_mode)) + if (layer_owner != NULL) + { + /* Do not enter the process-global owner mutex from a fork child. A + * provider inherited across fork must exec before it can be reopened. */ + if (ps_store_owner_require_current(layer_owner) != 0) + return -1; + local_close(); + } + if (ps_store_owner_acquire(store_dir, &layer_owner) != 0) return -1; + if (ps_store_owner_require_current(layer_owner) != 0) + { + goto fail; + } + n = snprintf(layer_dir, sizeof(layer_dir), "%s", + ps_store_owner_root(layer_owner)); + if (n < 0 || (size_t) n >= sizeof(layer_dir)) + { + errno = ENAMETOOLONG; + goto fail; + } /* Reap interrupted copies once at provider startup, not on every copy. */ if (cleanup_stale_copy_temps(layer_dir) != 0) - return -1; + goto fail; + if (stat(layer_dir, &store_st) != 0 || !S_ISDIR(store_st.st_mode)) + goto fail; object_dir[0] = '\0'; configured_object_dir = getenv("PAGESTORE_OBJECT_DIR"); if (configured_object_dir == NULL || configured_object_dir[0] == '\0') @@ -214,8 +246,14 @@ local_open(const char *store_dir) PS_MAX_CHANNELS - 1, (unsigned long long) UINT64_MAX) >= (int) sizeof(probe) || claim_object_dir() != 0 || cleanup_stale_copy_temps(object_dir) != 0) - return -1; + goto fail; return 0; + +fail: + save_errno = errno; + local_close(); + errno = save_errno; + return -1; } static void @@ -223,6 +261,11 @@ local_close(void) { layer_dir[0] = '\0'; object_dir[0] = '\0'; + if (layer_owner != NULL) + { + ps_store_owner_release(layer_owner); + layer_owner = NULL; + } } static int @@ -261,6 +304,575 @@ local_layer_path(uint64_t layer_id, char *buf, size_t buflen) return 0; } +static int +local_owner_current(void) +{ + if (layer_owner == NULL) + { + errno = EPERM; + return 0; + } + return ps_store_owner_require_current(layer_owner) == 0; +} + +static int +hex_digit(unsigned char c) +{ + if (c >= '0' && c <= '9') + return c - '0'; + if (c >= 'a' && c <= 'f') + return c - 'a' + 10; + return -1; +} + +/* The writer emits layer__<16 lowercase hex id>. */ +static int +parse_canonical_layer_name(const char *name, uint64_t *layer_id) +{ + const char *p; + const char *hex; + char *end; + unsigned long shard; + uint64_t id = 0; + char canonical[NAME_MAX + 1]; + + if (strncmp(name, "layer_", 6) != 0) + return 0; + p = name + 6; + if (*p == '\0' || (*p == '0' && p[1] >= '0' && p[1] <= '9')) + return -1; + errno = 0; + shard = strtoul(p, &end, 10); + if (errno != 0 || end == p || shard > 0xFFFF || *end != '_') + return -1; + hex = end + 1; + if (strlen(hex) != 16) + return -1; + for (int i = 0; i < 16; i++) + { + int digit = hex_digit((unsigned char) hex[i]); + + if (digit < 0) + return -1; + id = (id << 4) | (uint64_t) digit; + } + if (((id >> 48) & UINT64_C(0xFFFF)) != shard || + snprintf(canonical, sizeof(canonical), "layer_%lu_%016llx", shard, + (unsigned long long) id) < 0 || strcmp(canonical, name) != 0) + return -1; + *layer_id = id; + return 1; +} + +/* A copy temp is still provider-owned namespace, even when its PID is alive + * because that PID has since been reused. Recovery must recognize the name + * before applying canonical-layer validation, then leave it for the copier. */ +static int +parse_copy_temp_name(const char *name, pid_t *pid_out) +{ + const char *tmp; + const char *pid_text; + const char *attempt_text; + char base[NAME_MAX + 1]; + char canonical_suffix[64]; + char *end; + unsigned long attempt; + long pid; + uint64_t layer_id; + size_t base_len; + + if (strncmp(name, "layer_", 6) != 0) + return 0; + tmp = strstr(name, ".tmp."); + if (tmp == NULL) + return 0; + base_len = (size_t) (tmp - name); + if (base_len == 0 || base_len > NAME_MAX) + return -1; + memcpy(base, name, base_len); + base[base_len] = '\0'; + if (parse_canonical_layer_name(base, &layer_id) != 1) + return -1; + + pid_text = tmp + strlen(".tmp."); + if (*pid_text < '0' || *pid_text > '9') + return -1; + errno = 0; + pid = strtol(pid_text, &end, 10); + if (errno == ERANGE || end == pid_text || pid <= 0 || + (pid_t) pid != pid || *end != '.') + return -1; + attempt_text = end + 1; + if (*attempt_text < '0' || *attempt_text > '9') + return -1; + errno = 0; + attempt = strtoul(attempt_text, &end, 10); + if (errno == ERANGE || end == attempt_text || *end != '\0' || + attempt > UINT_MAX || + snprintf(canonical_suffix, sizeof(canonical_suffix), "%ld.%u", pid, + (unsigned int) attempt) < 0 || + strcmp(canonical_suffix, pid_text) != 0) + return -1; + (void) layer_id; + (void) attempt; + if (pid_out != NULL) + *pid_out = (pid_t) pid; + return 1; +} + +static int +compare_layer_ids(const void *left, const void *right) +{ + const uint64_t a = *(const uint64_t *) left; + const uint64_t b = *(const uint64_t *) right; + + return a < b ? -1 : (a > b ? 1 : 0); +} + +static int +layer_map_has_id(const uint64_t *map_ids, size_t nmap_ids, uint64_t layer_id) +{ + if (nmap_ids == 0) + return 0; + return bsearch(&layer_id, map_ids, nmap_ids, sizeof(*map_ids), + compare_layer_ids) != NULL; +} + +static int +canonicalize_local_layer_uri(PsLayerLocation *location, + const char *expected) +{ + char resolved[4096]; + char parent[PS_LAYER_URI_MAX]; + char joined[4096]; + const char *slash; + const char *basename; + const char *expected_basename; + struct stat st; + struct stat parent_st; + struct stat root_st; + size_t uri_len; + size_t parent_len; + int n; + + uri_len = strnlen(location->uri, sizeof(location->uri)); + if (uri_len == sizeof(location->uri)) + { + errno = EINVAL; + return -1; + } + slash = strrchr(location->uri, '/'); + if (slash == NULL) + { + errno = EINVAL; + return -1; + } + parent_len = (size_t) (slash - location->uri); + if (parent_len == 0) + parent_len = 1; + if (parent_len >= sizeof(parent)) + { + errno = EINVAL; + return -1; + } + memcpy(parent, location->uri, parent_len); + if (slash == location->uri) + parent[0] = '/'; + parent[parent_len] = '\0'; + basename = slash + 1; + if (basename[0] == '\0') + { + errno = EINVAL; + return -1; + } + expected_basename = strrchr(expected, '/') + 1; + if (strcmp(basename, expected_basename) != 0 || + realpath(parent, resolved) == NULL) + { + errno = EINVAL; + return -1; + } + if (stat(resolved, &parent_st) != 0 || stat(layer_dir, &root_st) != 0 || + parent_st.st_dev != root_st.st_dev || parent_st.st_ino != root_st.st_ino) + { + errno = EINVAL; + return -1; + } + n = snprintf(joined, sizeof(joined), "%s/%s", resolved, basename); + if (n < 0 || (size_t) n >= sizeof(joined) || + strcmp(joined, expected) != 0) + { + errno = EINVAL; + return -1; + } + /* The leaf may be absent while a remote-durable or deleting layer is + * replayed. If present, it must already be a non-symlink regular file. */ + if (lstat(location->uri, &st) == 0) + { + if (!S_ISREG(st.st_mode)) + { + errno = EINVAL; + return -1; + } + } + else if (errno != ENOENT) + { + errno = EINVAL; + return -1; + } + if (strlen(expected) >= sizeof(location->uri)) + { + errno = EINVAL; + return -1; + } + if (strcmp(location->uri, expected) != 0) + snprintf(location->uri, sizeof(location->uri), "%s", expected); + return 0; +} + +static int +local_validate_layer_locations(PsLayerMap *map) +{ + if (map == NULL) + { + errno = EINVAL; + return -1; + } + if (!local_owner_current()) + return -1; + for (uint32_t i = 0; i < map->nlayers; i++) + { + PsLayerDesc *layer = &map->layers[i]; + char expected[4096]; + int local_count = 0; + + if (layer->location_count > PS_LAYER_MAX_LOCATIONS) + { + errno = EINVAL; + return -1; + } + for (uint32_t j = 0; j < layer->location_count; j++) + { + PsLayerLocation *location = &layer->locations[j]; + + switch (location->tier) + { + case PS_LAYER_TIER_LOCAL_HOT: + case PS_LAYER_TIER_LOCAL_COLD: + if (++local_count > 1 || + local_layer_path(layer->layer_id, expected, + sizeof(expected)) != 0 || + canonicalize_local_layer_uri(location, + expected) != 0) + { + errno = EINVAL; + return -1; + } + break; + case PS_LAYER_TIER_REMOTE_OBJECT: + /* Object locations are not scanned or unlinked here, but a + * replayed location still needs a bounded nonempty URI. */ + if (location->uri[0] == '\0') + { + errno = EINVAL; + return -1; + } + break; + default: + errno = EINVAL; + return -1; + } + } + } + return 0; +} + +static int +validate_layer_candidate(int dirfd, const LocalLayerCandidate *candidate) +{ + struct stat st; + int fd; + + if (fstatat(dirfd, candidate->name, &st, AT_SYMLINK_NOFOLLOW) != 0) + return -1; + if (!S_ISREG(st.st_mode) || st.st_nlink != 1 || + st.st_dev != candidate->dev || st.st_ino != candidate->ino) + { + errno = EINVAL; + return -1; + } + fd = openat(dirfd, candidate->name, O_RDONLY | O_CLOEXEC | O_NOFOLLOW); + if (fd < 0) + return -1; + if (fstat(fd, &st) != 0 || !S_ISREG(st.st_mode) || st.st_nlink != 1 || + st.st_dev != candidate->dev || st.st_ino != candidate->ino) + { + int save_errno = errno; + + close(fd); + errno = save_errno != 0 ? save_errno : EINVAL; + return -1; + } + if (close(fd) != 0) + return -1; + return 0; +} + +static int +local_recover_local_layers(PsLayerMap *map) +{ + LocalLayerCandidate *candidates = NULL; + uint64_t *map_ids = NULL; + uint32_t ncandidates = 0; + uint32_t capacity = 0; + DIR *dir = NULL; + struct dirent *ent; + int scanfd = -1; + int unlinkfd = -1; + int changed = 0; + int rc = -1; + int save_errno = 0; + size_t map_count = (size_t) 0; + + if (map == NULL || layer_dir[0] == '\0') + { + errno = EINVAL; + return -1; + } + if (!local_owner_current()) + return -1; + if (local_validate_layer_locations(map) != 0) + return -1; + map_count = (size_t) map->nlayers; + if (map_count > 0) + { + /* On 64-bit builds uint32_t nlayers cannot overflow this allocation. + * Keep the check for narrower size_t targets without provoking a + * -Wtype-limits diagnostic on the normal build. */ +#if SIZE_MAX < UINT64_MAX + if ((uint64_t) map_count > (uint64_t) SIZE_MAX / sizeof(*map_ids)) + { + errno = EOVERFLOW; + return -1; + } +#endif + map_ids = malloc(map_count * sizeof(*map_ids)); + if (map_ids == NULL) + return -1; + for (size_t i = 0; i < map_count; i++) + map_ids[i] = map->layers[i].layer_id; + qsort(map_ids, map_count, sizeof(*map_ids), compare_layer_ids); + } + scanfd = open(layer_dir, O_RDONLY | O_DIRECTORY | O_CLOEXEC | O_NOFOLLOW); + if (scanfd < 0) + goto done; + unlinkfd = dup(scanfd); + if (unlinkfd < 0 || fcntl(unlinkfd, F_SETFD, FD_CLOEXEC) != 0) + { + save_errno = errno; + goto done; + } + dir = fdopendir(scanfd); + if (dir == NULL) + { + save_errno = errno; + goto done; + } + scanfd = -1; /* owned by DIR now */ + errno = 0; + while ((ent = readdir(dir)) != NULL) + { + LocalLayerCandidate candidate; + uint64_t layer_id; + int parsed; + int copy_temp; + struct stat st; + + if (strcmp(ent->d_name, ".") == 0 || strcmp(ent->d_name, "..") == 0) + continue; + copy_temp = parse_copy_temp_name(ent->d_name, NULL); + if (copy_temp == 1) + { + if (fstatat(dirfd(dir), ent->d_name, &st, AT_SYMLINK_NOFOLLOW) != 0) + { + if (errno == ENOENT) + { + errno = 0; + continue; + } + goto done; + } + if (!S_ISREG(st.st_mode) || st.st_nlink != 1) + { + errno = EINVAL; + goto done; + } + continue; + } + if (copy_temp < 0) + { + errno = EINVAL; + goto done; + } + parsed = parse_canonical_layer_name(ent->d_name, &layer_id); + if (parsed == 0) + continue; + if (parsed < 0 || strlen(ent->d_name) > NAME_MAX) + { + errno = EINVAL; + goto done; + } + memset(&candidate, 0, sizeof(candidate)); + snprintf(candidate.name, sizeof(candidate.name), "%s", ent->d_name); + candidate.layer_id = layer_id; + if (fstatat(dirfd(dir), candidate.name, &st, AT_SYMLINK_NOFOLLOW) != 0) + goto done; + if (!S_ISREG(st.st_mode) || st.st_nlink != 1) + { + errno = EINVAL; + goto done; + } + candidate.dev = st.st_dev; + candidate.ino = st.st_ino; + candidate.size = st.st_size; + if (validate_layer_candidate(dirfd(dir), &candidate) != 0) + goto done; + if (ncandidates == capacity) + { + uint32_t new_capacity = capacity == 0 ? 16 : capacity * 2; + LocalLayerCandidate *grown = realloc(candidates, + (size_t) new_capacity * sizeof(*grown)); + + if (grown == NULL) + goto done; + candidates = grown; + capacity = new_capacity; + } + candidates[ncandidates++] = candidate; + } + if (errno != 0) + goto done; + if (closedir(dir) != 0) + { + dir = NULL; + goto done; + } + dir = NULL; + /* Validate all map-owned local paths before deleting any orphan. IDs remain + * protected even when their manifest record is deleting, remote-only, or has + * no currently available local location. */ + for (uint32_t i = 0; i < map->nlayers; i++) + { + for (uint32_t j = 0; j < map->layers[i].location_count; j++) + { + const PsLayerLocation *location = &map->layers[i].locations[j]; + char expected[4096]; + struct stat st; + + if (location->tier != PS_LAYER_TIER_LOCAL_HOT && + location->tier != PS_LAYER_TIER_LOCAL_COLD) + continue; + if (local_layer_path(map->layers[i].layer_id, expected, + sizeof(expected)) != 0 || + strcmp(location->uri, expected) != 0) + { + errno = EINVAL; + goto done; + } + if (map->layers[i].remote_durable) + { + if (stat(expected, &st) != 0) + { + if (errno == ENOENT) + continue; + goto done; + } + if (!S_ISREG(st.st_mode) || st.st_nlink != 1 || + (location->size != 0 && + (st.st_size < 0 || + (uint64_t) st.st_size != location->size))) + { + errno = EIO; + goto done; + } + } + else if (location->available && !map->layers[i].deleting) + { + if (stat(expected, &st) != 0 || !S_ISREG(st.st_mode) || + st.st_nlink != 1 || (location->size != 0 && + (st.st_size < 0 || + (uint64_t) st.st_size != location->size))) + { + errno = errno == ENOENT ? ENOENT : EIO; + goto done; + } + } + } + } + + for (uint32_t i = 0; i < ncandidates; i++) + { + struct stat st; + int fd; + + if (layer_map_has_id(map_ids, map_count, candidates[i].layer_id)) + continue; + if (fstatat(unlinkfd, candidates[i].name, &st, AT_SYMLINK_NOFOLLOW) != 0) + { + if (errno == ENOENT) + continue; + goto done; + } + if (st.st_dev != candidates[i].dev || st.st_ino != candidates[i].ino || + !S_ISREG(st.st_mode) || st.st_nlink != 1) + { + errno = EINVAL; + goto done; + } + fd = openat(unlinkfd, candidates[i].name, + O_RDONLY | O_CLOEXEC | O_NOFOLLOW); + if (fd < 0) + goto done; + if (fstat(fd, &st) != 0 || st.st_dev != candidates[i].dev || + st.st_ino != candidates[i].ino || !S_ISREG(st.st_mode) || + st.st_nlink != 1) + { + if (fd >= 0) + close(fd); + errno = EINVAL; + goto done; + } + if (close(fd) != 0) + goto done; + if (unlinkat(unlinkfd, candidates[i].name, 0) != 0 && errno != ENOENT) + goto done; + changed = 1; + } + rc = 0; + +done: + if (changed && fsync(unlinkfd) != 0 && rc == 0) + { + rc = -1; + save_errno = errno; + } + if (dir != NULL) + closedir(dir); + if (unlinkfd >= 0) + close(unlinkfd); + if (scanfd >= 0) + close(scanfd); + free(map_ids); + free(candidates); + if (rc != 0) + { + if (save_errno == 0) + save_errno = errno != 0 ? errno : EIO; + errno = save_errno; + } + return rc; +} + static int local_fsync_dir(void) { @@ -333,31 +945,69 @@ cleanup_stale_copy_temps(const char *dir) { DIR *d; struct dirent *ent; - char path[4096]; - char *tmp; - char *end; - long pid; - int n; + struct stat st; + pid_t pid; + int changed = 0; + int rc = 0; + int save_errno = 0; d = opendir(dir); if (d == NULL) return -1; - while ((ent = readdir(d)) != NULL) + for (;;) { - if (strncmp(ent->d_name, "layer_", strlen("layer_")) != 0 || - (tmp = strstr(ent->d_name, ".tmp.")) == NULL) - continue; errno = 0; - pid = strtol(tmp + strlen(".tmp."), &end, 10); - if (errno != 0 || end == tmp + strlen(".tmp.") || *end != '.' || pid <= 0 || - (kill((pid_t) pid, 0) != -1 || errno != ESRCH)) + ent = readdir(d); + if (ent == NULL) + { + if (errno != 0) + { + rc = -1; + save_errno = errno; + } + break; + } + if (parse_copy_temp_name(ent->d_name, &pid) != 1) continue; - n = snprintf(path, sizeof(path), "%s/%s", dir, ent->d_name); - if (n >= 0 && (size_t) n < sizeof(path)) - unlink(path); + if (fstatat(dirfd(d), ent->d_name, &st, AT_SYMLINK_NOFOLLOW) != 0) + { + if (errno == ENOENT) + continue; + rc = -1; + save_errno = errno; + break; + } + if (!S_ISREG(st.st_mode) || st.st_nlink != 1) + { + rc = -1; + save_errno = EINVAL; + break; + } + if (kill(pid, 0) != -1 || errno != ESRCH) + continue; + if (unlinkat(dirfd(d), ent->d_name, 0) != 0) + { + if (errno == ENOENT) + continue; + rc = -1; + save_errno = errno; + break; + } + changed = 1; } - closedir(d); - return 0; + if (changed && fsync(dirfd(d)) != 0 && rc == 0) + { + rc = -1; + save_errno = errno; + } + if (closedir(d) != 0 && rc == 0) + { + rc = -1; + save_errno = errno; + } + if (rc != 0) + errno = save_errno != 0 ? save_errno : EIO; + return rc; } static int @@ -438,6 +1088,8 @@ local_create_local_layer(uint64_t layer_id, char *uri, uint32_t uri_len) int fd; int n; + if (!local_owner_current()) + return -1; if (local_layer_path(layer_id, path, sizeof(path)) != 0) return -1; fd = open(path, O_RDWR | O_CREAT | O_EXCL, 0600); @@ -465,6 +1117,8 @@ local_layer_exists(uint64_t layer_id) { char path[4096]; + if (!local_owner_current()) + return -1; if (local_layer_path(layer_id, path, sizeof(path)) != 0) return -1; if (access(path, F_OK) == 0) @@ -480,6 +1134,8 @@ local_write_local_layer(uint64_t layer_id, const void *buf, uint64_t len) const char *p = buf; uint64_t done = 0; + if (!local_owner_current()) + return -1; if (local_layer_path(layer_id, path, sizeof(path)) != 0) return -1; fd = open(path, O_WRONLY); @@ -507,6 +1163,8 @@ local_seal_local_layer(uint64_t layer_id) int fd; int rc; + if (!local_owner_current()) + return -1; if (local_layer_path(layer_id, path, sizeof(path)) != 0) return -1; fd = open(path, O_RDONLY); @@ -528,6 +1186,8 @@ local_read_layer_block(const PsLayerDesc *layer, uint64_t off, ssize_t n; uint32_t nlocs; + if (!local_owner_current()) + return -1; nlocs = layer->location_count; if (nlocs > PS_LAYER_MAX_LOCATIONS) return -1; @@ -588,6 +1248,9 @@ local_refresh_layer_cache(const PsLayerDesc *layer) const PsLayerLocation *local_loc; char local[4096]; + if (!local_owner_current()) + return -1; + local_loc = local_location(layer); /* Before upload durability, a manifest-owned local layer is the only source * of truth. After remote durability, the verified remote object may repair @@ -609,6 +1272,8 @@ local_remote_uri(uint64_t layer_id, char *uri, uint32_t uri_len) char path[4096]; int n; + if (!local_owner_current()) + return -1; if (object_layer_path(layer_id, path, sizeof(path)) != 0) return -1; n = snprintf(uri, uri_len, "%s", path); @@ -632,6 +1297,9 @@ local_upload_layer(const PsLayerDesc *layer) const PsLayerLocation *published; char remote[4096]; + if (!local_owner_current()) + return -1; + source = local_location(layer); published = remote_location(layer); if (source == NULL || @@ -677,6 +1345,9 @@ local_download_layer(const PsLayerDesc *layer) char local[4096]; struct stat st; + if (!local_owner_current()) + return -1; + source = remote_location(layer); if (source == NULL || local_layer_path(layer->layer_id, local, sizeof(local)) != 0) return -1; @@ -724,6 +1395,9 @@ local_delete_remote_layer(const PsLayerDesc *layer) const PsLayerLocation *location; char expected[4096]; + if (!local_owner_current()) + return -1; + location = remote_location(layer); if (location == NULL || object_layer_path(layer->layer_id, expected, sizeof(expected)) != 0 || @@ -741,6 +1415,9 @@ local_delete_local_layer(const PsLayerDesc *layer) int unlinked = 0; uint32_t nlocs; + if (!local_owner_current()) + return -1; + nlocs = layer->location_count; if (nlocs > PS_LAYER_MAX_LOCATIONS) return -1; @@ -781,6 +1458,8 @@ local_layer_exists_remote(const PsLayerDesc *layer) const PsLayerLocation *location; char expected[4096]; + if (!local_owner_current()) + return -1; location = remote_location(layer); if (location == NULL || object_layer_path(layer->layer_id, expected, sizeof(expected)) != 0 || @@ -799,6 +1478,8 @@ local_verify_remote_layer(const PsLayerDesc *layer) char expected[4096]; struct stat st; + if (!local_owner_current()) + return -1; location = remote_location(layer); if (location == NULL || object_layer_path(layer->layer_id, expected, sizeof(expected)) != 0 || @@ -831,6 +1512,8 @@ const PsLayerStore PsLayerStoreLocal = { .name = "local", .open = local_open, .close = local_close, + .validate_local_layers = local_validate_layer_locations, + .recover_local_layers = local_recover_local_layers, .create_local_layer = local_create_local_layer, .layer_exists_local = local_layer_exists, .write_local_layer = local_write_local_layer, diff --git a/contrib/pagestore/pagestore_layer_store.h b/contrib/pagestore/pagestore_layer_store.h index 05740aaa21091..3ad0d1d83e54b 100644 --- a/contrib/pagestore/pagestore_layer_store.h +++ b/contrib/pagestore/pagestore_layer_store.h @@ -21,6 +21,10 @@ typedef struct PsLayerStore int (*open) (const char *store_dir); void (*close) (void); + /* Validate and normalize replayed local URIs in-place before core handoff. */ + int (*validate_local_layers) (PsLayerMap *map); + /* Reconcile canonical local layers after a complete manifest replay. */ + int (*recover_local_layers) (PsLayerMap *map); int (*create_local_layer) (uint64_t layer_id, char *uri, uint32_t uri_len); int (*layer_exists_local) (uint64_t layer_id); diff --git a/contrib/pagestore/pagestore_layer_store_test.c b/contrib/pagestore/pagestore_layer_store_test.c index 4552bd1e43c5a..a4c1fef705560 100644 --- a/contrib/pagestore/pagestore_layer_store_test.c +++ b/contrib/pagestore/pagestore_layer_store_test.c @@ -6,12 +6,16 @@ *------------------------------------------------------------------------- */ #include +#include #include #include #include +#include +#include #include #include "pagestore_layer_store.h" +#include "pagestore_store_owner.h" static int run = 0, failed = 0; @@ -50,6 +54,9 @@ main(void) char object_dir[] = "/tmp/pslayerstoreobjectXXXXXX"; char owner_path[sizeof(object_dir) + 32]; char stale_path[sizeof(object_dir) + 64]; + char retained_path[sizeof(local_dir) + 64]; + char alias_dir[] = "/tmp/pslayerstorealiasXXXXXX"; + char saved_cwd[4096]; char configured_object_dir[sizeof(object_dir) + 2]; char expected_remote_uri[PS_LAYER_URI_MAX]; char local_uri[PS_LAYER_URI_MAX]; @@ -73,7 +80,7 @@ main(void) ps_layer_store->open(local_dir) == 0, "open exclusive object directory"); ps_layer_store->close(); - snprintf(stale_path, sizeof(stale_path), "%s/layer_3_0000000000000011.tmp.999999.0", + snprintf(stale_path, sizeof(stale_path), "%s/layer_3_0003000000000011.tmp.999999.0", object_dir); { int fd = open(stale_path, O_WRONLY | O_CREAT | O_EXCL, 0600); @@ -82,9 +89,19 @@ main(void) close(fd); check(fd >= 0, "create interrupted-copy temporary"); } + snprintf(retained_path, sizeof(retained_path), + "%s/layer_3_0003000000000012.tmp.%ld.0", local_dir, + (long) getpid()); + { + int fd = open(retained_path, O_WRONLY | O_CREAT | O_EXCL, 0600); + + check(fd >= 0 && close(fd) == 0, + "create a copy temporary with the current PID"); + } snprintf(configured_object_dir, sizeof(configured_object_dir), "%s/", object_dir); check(setenv("PAGESTORE_OBJECT_DIR", configured_object_dir, 1) == 0 && - ps_layer_store->open(local_dir) == 0 && access(stale_path, F_OK) != 0, + ps_layer_store->open(local_dir) == 0 && access(stale_path, F_OK) != 0 && + access(retained_path, F_OK) == 0, "canonicalize object directory and reap interrupted copies at startup"); ps_layer_store->close(); check(ps_layer_store->open(other_local_dir) != 0, @@ -95,6 +112,103 @@ main(void) fprintf(stderr, "could not reopen object directory\n"); return 2; } + { + PsLayerMap empty; + + ps_layer_map_init(&empty); + check(ps_layer_store->recover_local_layers(&empty) == 0 && + access(retained_path, F_OK) == 0, + "recovery skips a live-PID copy temporary by its known grammar"); + ps_layer_map_free(&empty); + } + { + const uint64_t fork_layer_id = (3ULL << 48) | 16; + const uint64_t child_layer_id = (3ULL << 48) | 26; + char fork_layer_uri[PS_LAYER_URI_MAX]; + char child_layer_uri[PS_LAYER_URI_MAX]; + pid_t pid; + int status = 0; + + fork_layer_uri[0] = '\0'; + check(ps_layer_store->create_local_layer(fork_layer_id, + fork_layer_uri, + sizeof(fork_layer_uri)) == 0, + "create a provider layer before fork"); + snprintf(child_layer_uri, sizeof(child_layer_uri), + "%s/layer_3_%016llx", local_dir, + (unsigned long long) child_layer_id); + pid = fork(); + if (pid == 0) + { + char child_uri[PS_LAYER_URI_MAX]; + int open_rc; + int open_errno; + int create_rc; + int write_rc; + + open_rc = ps_layer_store->open(local_dir); + open_errno = errno; + create_rc = ps_layer_store->create_local_layer(child_layer_id, + child_uri, + sizeof(child_uri)); + write_rc = ps_layer_store->write_local_layer(fork_layer_id, + "child", 5); + _exit(open_rc != 0 && open_errno == ECHILD && + create_rc != 0 && write_rc != 0 ? 0 : 1); + } + check(pid > 0 && waitpid(pid, &status, 0) == pid && + WIFEXITED(status) && WEXITSTATUS(status) == 0 && + access(child_layer_uri, F_OK) != 0, + "forked provider rejects inherited-owner reopen and mutations"); + if (fork_layer_uri[0] != '\0') + unlink(fork_layer_uri); + unlink(child_layer_uri); + } + { + PsStoreOwner *owner1 = NULL; + PsStoreOwner *owner2 = NULL; + char canonical[4096]; + + check(realpath(local_dir, canonical) != NULL && + ps_store_owner_acquire(local_dir, &owner1) == 0 && + ps_store_owner_acquire(local_dir, &owner2) == 0 && + strcmp(ps_store_owner_root(owner1), canonical) == 0 && + strcmp(ps_store_owner_root(owner2), canonical) == 0, + "same-process owner references share the canonical lease"); + ps_store_owner_release(owner1); + ps_store_owner_release(owner2); + } + { + PsStoreOwner *held = NULL; + PsStoreOwner *after = NULL; + pid_t pid; + int status = 0; + + check(ps_store_owner_acquire(local_dir, &held) == 0, + "hold owner lease across fork test"); + pid = fork(); + if (pid == 0) + { + PsStoreOwner *child = NULL; + int rc; + + /* This inherited handle is not a child lease; release must not + * decrement the parent's refcount or unlock its flock. */ + ps_store_owner_release(held); + errno = 0; + rc = ps_store_owner_acquire(local_dir, &child); + if (child != NULL) + ps_store_owner_release(child); + _exit(rc != 0 && (errno == EWOULDBLOCK || errno == EAGAIN) ? 0 : 1); + } + check(pid > 0 && waitpid(pid, &status, 0) == pid && + WIFEXITED(status) && WEXITSTATUS(status) == 0, + "forked child cannot reuse inherited owner lease"); + ps_store_owner_release(held); + check(ps_store_owner_acquire(local_dir, &after) == 0, + "owner lease is reacquirable after parent release"); + ps_store_owner_release(after); + } memset(&layer, 0, sizeof(layer)); layer.layer_id = (3ULL << 48) | 17; @@ -110,6 +224,214 @@ main(void) "write and seal local layer"); snprintf(layer.locations[0].uri, sizeof(layer.locations[0].uri), "%s", local_uri); layer.locations[0].size = strlen(contents); + { + char child_buf[64]; + pid_t pid; + int status = 0; + + pid = fork(); + if (pid == 0) + { + int read_rc; + int read_errno; + + errno = 0; + read_rc = ps_layer_store->read_layer_block(&layer, 0, + child_buf, strlen(contents)); + read_errno = errno; + _exit(read_rc != 0 && read_errno == ECHILD ? 0 : 1); + } + check(pid > 0 && waitpid(pid, &status, 0) == pid && + WIFEXITED(status) && WEXITSTATUS(status) == 0, + "forked provider rejects inherited-owner reads"); + } + { + PsLayerMap legacy; + char legacy_uri[PS_LAYER_URI_MAX]; + char leaf_symlink[PS_LAYER_URI_MAX]; + char validation_orphan[PS_LAYER_URI_MAX]; + const char *basename = strrchr(local_uri, '/') + 1; + int alias_ready; + int relative_ok = 0; + int leaf_linked; + + ps_layer_map_init(&legacy); + alias_ready = mkdtemp(alias_dir) != NULL && rmdir(alias_dir) == 0 && + symlink(local_dir, alias_dir) == 0; + check(alias_ready, "create a legacy store spelling alias"); + snprintf(validation_orphan, sizeof(validation_orphan), + "%s/layer_3_0003000000000019", local_dir); + check(ps_layer_map_add(&legacy, &layer) == 0 && + ps_layer_store->create_local_layer((3ULL << 48) | 25, + validation_orphan, + sizeof(validation_orphan)) == 0 && + ps_layer_store->write_local_layer((3ULL << 48) | 25, contents, + strlen(contents)) == 0, + "seed an orphan for invalid URI recovery checks"); + if (alias_ready) + { + snprintf(legacy_uri, sizeof(legacy_uri), "%s/./%s", alias_dir, + basename); + check(snprintf(legacy.layers[0].locations[0].uri, + sizeof(legacy.layers[0].locations[0].uri), "%s", + legacy_uri) >= 0 && + ps_layer_store->validate_local_layers(&legacy) == 0 && + strcmp(legacy.layers[0].locations[0].uri, local_uri) == 0, + "normalize a symlinked legacy local URI"); + + if (getcwd(saved_cwd, sizeof(saved_cwd)) != NULL && + chdir(local_dir) == 0) + { + relative_ok = snprintf(legacy.layers[0].locations[0].uri, + sizeof(legacy.layers[0].locations[0].uri), "./%s", + basename) >= 0 && + ps_layer_store->validate_local_layers(&legacy) == 0 && + strcmp(legacy.layers[0].locations[0].uri, local_uri) == 0; + if (chdir(saved_cwd) != 0) + relative_ok = 0; + } + check(relative_ok, "normalize a same-directory relative local URI"); + + snprintf(legacy.layers[0].locations[0].uri, + sizeof(legacy.layers[0].locations[0].uri), "%s/%s", object_dir, + basename); + check(ps_layer_store->validate_local_layers(&legacy) != 0 && + ps_layer_store->recover_local_layers(&legacy) != 0 && + access(validation_orphan, F_OK) == 0, + "reject a foreign parent before orphan recovery"); + snprintf(legacy.layers[0].locations[0].uri, + sizeof(legacy.layers[0].locations[0].uri), "%s/layer_3_%016llx", + local_dir, (unsigned long long) layer.layer_id + 1); + check(ps_layer_store->validate_local_layers(&legacy) != 0 && + ps_layer_store->recover_local_layers(&legacy) != 0 && + access(validation_orphan, F_OK) == 0, + "reject a wrong layer ID before orphan recovery"); + snprintf(legacy.layers[0].locations[0].uri, + sizeof(legacy.layers[0].locations[0].uri), + "%s/layer_3_0003000000000014.bad", local_dir); + check(ps_layer_store->validate_local_layers(&legacy) != 0 && + ps_layer_store->recover_local_layers(&legacy) != 0 && + access(validation_orphan, F_OK) == 0, + "reject a malformed leaf before orphan recovery"); + + snprintf(leaf_symlink, sizeof(leaf_symlink), + "%s/layer_3_0003000000000016", local_dir); + leaf_linked = symlink(local_uri, leaf_symlink) == 0; + legacy.layers[0].layer_id = (3ULL << 48) | 22; + snprintf(legacy.layers[0].locations[0].uri, + sizeof(legacy.layers[0].locations[0].uri), "%s", leaf_symlink); + check(leaf_linked && ps_layer_store->validate_local_layers(&legacy) != 0 && + ps_layer_store->recover_local_layers(&legacy) != 0 && + access(validation_orphan, F_OK) == 0, + "reject a symlinked leaf before orphan recovery"); + if (leaf_linked) + unlink(leaf_symlink); + legacy.layers[0].layer_id = layer.layer_id; + + snprintf(legacy.layers[0].locations[0].uri, + sizeof(legacy.layers[0].locations[0].uri), "%s/./%s", alias_dir, + basename); + check(unlink(alias_dir) == 0 && + ps_layer_store->validate_local_layers(&legacy) != 0 && + ps_layer_store->recover_local_layers(&legacy) != 0 && + access(validation_orphan, F_OK) == 0, + "reject a disappeared alias before orphan recovery"); + + snprintf(legacy.layers[0].locations[0].uri, + sizeof(legacy.layers[0].locations[0].uri), "%s", local_uri); + check(ps_layer_store->recover_local_layers(&legacy) == 0 && + access(validation_orphan, F_OK) != 0, + "valid canonical recovery can sweep the seeded orphan"); + } + else + unlink(validation_orphan); + ps_layer_map_free(&legacy); + if (alias_ready) + unlink(alias_dir); + } + { + PsLayerMap unsafe_map; + char unsafe_orphan[PS_LAYER_URI_MAX]; + char unsafe_temp[PS_LAYER_URI_MAX]; + char hard_target[PS_LAYER_URI_MAX]; + int fd; + int temp_ready; + + ps_layer_map_init(&unsafe_map); + snprintf(unsafe_orphan, sizeof(unsafe_orphan), + "%s/layer_3_0003000000000019", local_dir); + snprintf(unsafe_temp, sizeof(unsafe_temp), + "%s/layer_3_0003000000000027.tmp.%ld.1", local_dir, + (long) getpid()); + check(ps_layer_map_add(&unsafe_map, &layer) == 0 && + ps_layer_store->create_local_layer((3ULL << 48) | 25, + unsafe_orphan, + sizeof(unsafe_orphan)) == 0 && + ps_layer_store->write_local_layer((3ULL << 48) | 25, contents, + strlen(contents)) == 0, + "seed an orphan before unsafe copy-temp checks"); + + temp_ready = mkdir(unsafe_temp, 0700) == 0; + check(temp_ready && ps_layer_store->recover_local_layers(&unsafe_map) != 0 && + access(unsafe_temp, F_OK) == 0 && + access(unsafe_orphan, F_OK) == 0, + "reject a directory copy temporary before orphan cleanup"); + if (temp_ready) + rmdir(unsafe_temp); + + temp_ready = symlink(local_uri, unsafe_temp) == 0; + check(temp_ready && ps_layer_store->recover_local_layers(&unsafe_map) != 0 && + access(unsafe_temp, F_OK) == 0 && + access(unsafe_orphan, F_OK) == 0, + "reject a symlink copy temporary before orphan cleanup"); + if (temp_ready) + unlink(unsafe_temp); + + snprintf(hard_target, sizeof(hard_target), "%s/copy-temp-target", local_dir); + fd = open(hard_target, O_WRONLY | O_CREAT | O_EXCL, 0600); + temp_ready = fd >= 0 && close(fd) == 0 && link(hard_target, unsafe_temp) == 0; + check(temp_ready && ps_layer_store->recover_local_layers(&unsafe_map) != 0 && + access(unsafe_temp, F_OK) == 0 && + access(unsafe_orphan, F_OK) == 0, + "reject a hard-linked copy temporary before orphan cleanup"); + if (temp_ready) + unlink(unsafe_temp); + unlink(hard_target); + + check(ps_layer_store->recover_local_layers(&unsafe_map) == 0 && + access(unsafe_orphan, F_OK) != 0, + "safe recovery removes the orphan after unsafe temps are gone"); + ps_layer_map_free(&unsafe_map); + } + { + char stale_dir[PS_LAYER_URI_MAX]; + pid_t dead_pid; + int status = 0; + int made = 0; + + dead_pid = fork(); + if (dead_pid == 0) + _exit(0); + if (dead_pid > 0 && waitpid(dead_pid, &status, 0) == dead_pid && + WIFEXITED(status)) + { + snprintf(stale_dir, sizeof(stale_dir), + "%s/layer_3_0003000000000028.tmp.%ld.1", local_dir, + (long) dead_pid); + made = mkdir(stale_dir, 0700) == 0; + } + check(made, "seed an unlink-failing stale copy temporary"); + if (made) + { + ps_layer_store->close(); + check(ps_layer_store->open(local_dir) != 0, + "propagate stale copy-temp unlink failure at startup"); + check(rmdir(stale_dir) == 0, + "remove the stale cleanup failure fixture"); + check(ps_layer_store->open(local_dir) == 0, + "reopen after stale cleanup failure is repaired"); + } + } check(ps_layer_store->remote_uri(layer.layer_id, remote_uri, sizeof(remote_uri)) == 0, @@ -186,9 +508,135 @@ main(void) check(ps_layer_store->delete_remote_layer(&layer) == 0 && ps_layer_store->layer_exists_remote(&layer) == 0, "remote delete is idempotent"); + { + PsLayerMap map; + char orphan[PS_LAYER_URI_MAX]; + char protected_path[PS_LAYER_URI_MAX]; + char bad_name[PS_LAYER_URI_MAX]; + char hard_name[PS_LAYER_URI_MAX]; + int fd; + uint64_t orphan_id = (3ULL << 48) | 19; + uint64_t bad_id = (3ULL << 48) | 20; + uint64_t protected_id = (3ULL << 48) | 5; + PsLayerDesc protected_layer; + + ps_layer_map_init(&map); + check(ps_layer_map_add(&map, &layer) == 0, + "build recovery reference map"); + memset(&protected_layer, 0, sizeof(protected_layer)); + protected_layer.layer_id = protected_id; + check(ps_layer_map_add(&map, &protected_layer) == 0, + "add an out-of-order manifest layer ID"); + check(ps_layer_store->create_local_layer(protected_id, protected_path, + sizeof(protected_path)) == 0 && + ps_layer_store->recover_local_layers(&map) == 0 && + access(protected_path, F_OK) == 0, + "sorted manifest IDs protect a referenced layer"); + unlink(protected_path); + /* The recovery corruption case models a live, not-yet-remote-durable + * reference; test remote-durable/unavailable semantics separately below. */ + map.layers[0].remote_durable = false; + check(ps_layer_store->create_local_layer(orphan_id, orphan, + sizeof(orphan)) == 0 && + ps_layer_store->write_local_layer(orphan_id, contents, + strlen(contents)) == 0, + "create an unreferenced canonical layer"); + check(ps_layer_store->recover_local_layers(&map) == 0 && + access(orphan, F_OK) != 0 && access(local_uri, F_OK) == 0, + "reconcile removes only unreferenced canonical layers"); + check(ps_layer_store->recover_local_layers(&map) == 0, + "reconciliation is retry-safe after a completed sweep"); + + snprintf(bad_name, sizeof(bad_name), "%s/layer_3_0003000000000014.bad", + local_dir); + fd = open(bad_name, O_WRONLY | O_CREAT | O_EXCL, 0600); + if (fd >= 0) + close(fd); + check(fd >= 0 && ps_layer_store->create_local_layer(bad_id, orphan, + sizeof(orphan)) == 0 && + ps_layer_store->recover_local_layers(&map) != 0 && + access(bad_name, F_OK) == 0 && access(orphan, F_OK) == 0, + "malformed layer namespace fails closed without unlinking"); + unlink(bad_name); + unlink(orphan); + + snprintf(hard_name, sizeof(hard_name), "%s/layer_3_0003000000000015", + local_dir); + fd = link(local_uri, hard_name); + check(fd == 0 && ps_layer_store->recover_local_layers(&map) != 0 && + access(hard_name, F_OK) == 0, + "hard-linked canonical layer fails closed"); + if (fd == 0) + unlink(hard_name); + check(ps_layer_store->recover_local_layers(&map) == 0, + "hard-link rejection can be retried safely"); + check(truncate(local_uri, 1) == 0, + "prepare a size-corrupt referenced layer"); + check(ps_layer_store->create_local_layer((3ULL << 48) | 21, orphan, + sizeof(orphan)) == 0 && + ps_layer_store->recover_local_layers(&map) != 0 && + access(orphan, F_OK) == 0, + "size-corrupt live layer fails closed"); + fd = open(local_uri, O_WRONLY | O_TRUNC); + if (fd >= 0) + { + ssize_t nw = write(fd, contents, strlen(contents)); + int close_rc = close(fd); + + check(nw == (ssize_t) strlen(contents) && close_rc == 0, + "restore the referenced layer after corruption"); + } + else + check(0, "open the referenced layer for restoration"); + check(unlink(local_uri) == 0, + "remove referenced local layer for deleting recovery"); + map.layers[0].deleting = true; + check(ps_layer_store->recover_local_layers(&map) == 0 && + access(orphan, F_OK) != 0 && access(local_uri, F_OK) != 0, + "missing deleting layer does not block retry cleanup"); + map.layers[0].deleting = false; + map.layers[0].remote_durable = true; + map.layers[0].locations[0].available = false; + check(ps_layer_store->create_local_layer(layer.layer_id, local_uri, + sizeof(local_uri)) == 0 && + ps_layer_store->write_local_layer(layer.layer_id, contents, + strlen(contents)) == 0 && + ps_layer_store->create_local_layer((3ULL << 48) | 22, orphan, + sizeof(orphan)) == 0 && + ps_layer_store->recover_local_layers(&map) == 0 && + access(local_uri, F_OK) == 0 && access(orphan, F_OK) != 0, + "unavailable remote-durable reference remains protected"); + check(truncate(local_uri, 1) == 0 && + ps_layer_store->create_local_layer((3ULL << 48) | 24, orphan, + sizeof(orphan)) == 0 && + ps_layer_store->recover_local_layers(&map) != 0 && + access(orphan, F_OK) == 0 && + ps_layer_store->write_local_layer(layer.layer_id, contents, + strlen(contents)) == 0 && + ps_layer_store->recover_local_layers(&map) == 0 && + access(orphan, F_OK) != 0, + "present corrupt remote-durable cache fails closed"); + map.layers[0].remote_durable = false; + map.layers[0].locations[0].available = true; + map.layers[0].location_count = 2; + snprintf(map.layers[0].locations[1].uri, + sizeof(map.layers[0].locations[1].uri), "%s/not-canonical", + local_dir); + map.layers[0].locations[1].tier = PS_LAYER_TIER_LOCAL_COLD; + map.layers[0].locations[1].available = false; + check(ps_layer_store->create_local_layer((3ULL << 48) | 23, orphan, + sizeof(orphan)) == 0 && + ps_layer_store->recover_local_layers(&map) != 0 && + access(orphan, F_OK) == 0, + "conflicting second local location fails closed"); + map.layers[0].location_count = 1; + unlink(orphan); + ps_layer_map_free(&map); + } ps_layer_store->close(); unsetenv("PAGESTORE_OBJECT_DIR"); + unlink(retained_path); snprintf(owner_path, sizeof(owner_path), "%s/.pagestore-owner", object_dir); unlink(owner_path); snprintf(owner_path, sizeof(owner_path), "%s/.pagestore-store-id", local_dir); diff --git a/contrib/pagestore/pagestore_manifest.c b/contrib/pagestore/pagestore_manifest.c index 48cf3989b0ea0..6fbb27c25b22c 100644 --- a/contrib/pagestore/pagestore_manifest.c +++ b/contrib/pagestore/pagestore_manifest.c @@ -19,6 +19,7 @@ #include "pagestore_manifest.h" #include "pagestore_fault.h" +#include "pagestore_store_owner.h" #define PS_MANIFEST_MAGIC 0x504d414e /* "PMAN" */ #define PS_MANIFEST_VERSION 3 /* 3: per-record CRC (over the header + payload) */ @@ -99,6 +100,7 @@ typedef struct PsManifestLayerDisk static char manifest_path[4096]; static char manifest_dir[2048]; +static PsStoreOwner *manifest_owner; PsLayerMap ps_layer_map; static PsFlushWatermark flush_watermarks[PS_MAX_CHANNELS]; static uint8_t flush_watermark_valid[PS_MAX_CHANNELS]; @@ -116,6 +118,8 @@ static uint8_t flush_watermark_valid[PS_MAX_CHANNELS]; * reads the flag), so all reads/writes go through __atomic. */ static int manifest_poisoned = 0; +static int manifest_replay_repaired; +static int manifest_replay_had_file; /* * Records currently in the on-disk log (set by replay, bumped by append, reset by @@ -139,6 +143,110 @@ manifest_fsync_dir(void) return rc; } +static int +manifest_repair_marker_path(char *path, size_t path_len) +{ + int n = snprintf(path, path_len, "%s/.pagestore-orphan-sweep-inhibited", + manifest_dir); + + if (n < 0 || (size_t) n >= path_len) + { + errno = ENAMETOOLONG; + return -1; + } + return 0; +} + +static int +manifest_validate_repair_marker(void) +{ + char path[4096]; + struct stat st; + int fd; + + if (manifest_repair_marker_path(path, sizeof(path)) != 0) + return -1; + fd = open(path, O_RDONLY | O_CLOEXEC | O_NOFOLLOW | O_NONBLOCK); + if (fd < 0) + return errno == ENOENT ? 0 : -1; + if (fstat(fd, &st) != 0) + { + int save_errno = errno; + + close(fd); + errno = save_errno; + return -1; + } + if (!S_ISREG(st.st_mode) || st.st_nlink != 1 || st.st_size != 0) + { + close(fd); + errno = EINVAL; + return -1; + } + if (close(fd) != 0) + return -1; + return 1; +} + +/* Install the quarantine before changing the manifest. A zero-length regular + * file is enough: its durable existence, not its contents, is the provenance. */ +static int +manifest_install_repair_marker(void) +{ + char path[4096]; + struct stat st; + int fd; + + if (manifest_repair_marker_path(path, sizeof(path)) != 0) + return -1; + fd = open(path, O_WRONLY | O_CREAT | O_EXCL | O_CLOEXEC | O_NOFOLLOW, 0600); + if (fd < 0) + { + if (errno != EEXIST) + return -1; + if (manifest_validate_repair_marker() != 1) + return -1; + fd = open(path, O_WRONLY | O_CLOEXEC | O_NOFOLLOW | O_NONBLOCK); + if (fd < 0) + return -1; + if (fsync(fd) != 0) + { + int save_errno = errno; + + close(fd); + errno = save_errno; + return -1; + } + if (close(fd) != 0) + return -1; + return manifest_fsync_dir(); + } + { + int fstat_rc = fstat(fd, &st); + + if (fstat_rc != 0 || !S_ISREG(st.st_mode) || st.st_nlink != 1) + { + int save_errno = fstat_rc != 0 ? errno : EINVAL; + + close(fd); + (void) unlink(path); + errno = save_errno; + return -1; + } + } + if (fsync(fd) != 0) + { + int save_errno = errno; + + close(fd); + errno = save_errno; + return -1; + } + if (close(fd) != 0) + return -1; + return manifest_fsync_dir(); +} + /* FNV-1a (streaming): not cryptographic, just integrity. Matches img_crc(). */ static uint32_t manifest_fnv1a(uint32_t h, const void *p, size_t n) @@ -214,13 +322,32 @@ manifest_append(uint32_t type, const void *payload, uint32_t len) int rc = 0; int created = 0; + if (ps_store_owner_require_current(manifest_owner) != 0) + return -1; + /* once the tail may be torn, never append again (see manifest_poisoned) */ if (__atomic_load_n(&manifest_poisoned, __ATOMIC_ACQUIRE)) return -1; - fd = open(manifest_path, O_WRONLY | O_APPEND | O_CREAT, 0600); + fd = open(manifest_path, + O_WRONLY | O_APPEND | O_CREAT | O_CLOEXEC | O_NOFOLLOW | + O_NONBLOCK, 0600); if (fd < 0) return -1; /* nothing written; tail not torn */ + { + struct stat st; + int fstat_rc; + + fstat_rc = fstat(fd, &st); + if (fstat_rc != 0 || !S_ISREG(st.st_mode) || st.st_nlink != 1) + { + int save_errno = fstat_rc != 0 ? errno : EINVAL; + + close(fd); + errno = save_errno; + return -1; + } + } if (lseek(fd, 0, SEEK_END) == 0) created = 1; @@ -388,27 +515,66 @@ int ps_manifest_open(const char *store_dir) { int n; + int save_errno; - n = snprintf(manifest_dir, sizeof(manifest_dir), "%s", store_dir); - if (n < 0 || (size_t) n >= sizeof(manifest_dir)) + if (manifest_owner != NULL) + { + if (ps_store_owner_require_current(manifest_owner) != 0) + return -1; + ps_manifest_close(); + } + if (ps_store_owner_acquire(store_dir, &manifest_owner) != 0) return -1; - n = snprintf(manifest_path, sizeof(manifest_path), "%s/layers.manifest", store_dir); + n = snprintf(manifest_dir, sizeof(manifest_dir), "%s", + ps_store_owner_root(manifest_owner)); + if (n < 0 || (size_t) n >= sizeof(manifest_dir)) + { + errno = ENAMETOOLONG; + goto fail; + } + n = snprintf(manifest_path, sizeof(manifest_path), "%s/layers.manifest", + manifest_dir); if (n < 0 || (size_t) n >= sizeof(manifest_path)) - return -1; + { + errno = ENAMETOOLONG; + goto fail; + } + if (manifest_validate_repair_marker() < 0) + goto fail; /* replay truncates any torn tail; start clean */ __atomic_store_n(&manifest_poisoned, 0, __ATOMIC_RELEASE); manifest_nrecords = 0; + manifest_replay_repaired = 0; + manifest_replay_had_file = 0; memset(flush_watermarks, 0, sizeof(flush_watermarks)); memset(flush_watermark_valid, 0, sizeof(flush_watermark_valid)); ps_layer_map_init(&ps_layer_map); return 0; + +fail: + save_errno = errno; + ps_store_owner_release(manifest_owner); + manifest_owner = NULL; + manifest_dir[0] = '\0'; + manifest_path[0] = '\0'; + errno = save_errno; + return -1; } void ps_manifest_close(void) { + if (manifest_owner != NULL && + ps_store_owner_require_current(manifest_owner) != 0) + return; ps_layer_map_free(&ps_layer_map); manifest_path[0] = '\0'; + manifest_dir[0] = '\0'; + if (manifest_owner != NULL) + { + ps_store_owner_release(manifest_owner); + manifest_owner = NULL; + } } /* @@ -423,6 +589,26 @@ ps_manifest_poisoned(void) return __atomic_load_n(&manifest_poisoned, __ATOMIC_ACQUIRE); } +int +ps_manifest_replay_repaired(void) +{ + return manifest_replay_repaired; +} + +int +ps_manifest_replay_had_manifest(void) +{ + return manifest_replay_had_file; +} + +int +ps_manifest_orphan_sweep_inhibited(void) +{ + if (ps_store_owner_require_current(manifest_owner) != 0) + return -1; + return manifest_validate_repair_marker(); +} + /* * Does a fully-intact record begin at byte offset 'off'? "Fully intact" means the * header fits, magic and version are exact, the type is known, the len field equals @@ -505,18 +691,30 @@ ps_manifest_replay(PsLayerMap *map) off_t file_size; struct stat st; + if (ps_store_owner_require_current(manifest_owner) != 0) + return -1; + manifest_nrecords = 0; - fd = open(manifest_path, O_RDWR); + fd = open(manifest_path, + O_RDWR | O_CLOEXEC | O_NOFOLLOW | O_NONBLOCK); if (fd < 0) { if (errno == ENOENT) return 0; return -1; } - if (fstat(fd, &st) != 0) + manifest_replay_had_file = 1; { - close(fd); - return -1; + int fstat_rc = fstat(fd, &st); + + if (fstat_rc != 0 || !S_ISREG(st.st_mode) || st.st_nlink != 1) + { + int save_errno = fstat_rc != 0 ? errno : EINVAL; + + close(fd); + errno = save_errno; + return -1; + } } file_size = st.st_size; @@ -758,12 +956,19 @@ ps_manifest_replay(PsLayerMap *map) manifest_nrecords++; } + if (truncate_tail && manifest_install_repair_marker() != 0) + { + close(fd); + return -1; + } if (truncate_tail && (ftruncate(fd, good_off) != 0 || fsync(fd) != 0)) { close(fd); return -1; } + if (truncate_tail) + manifest_replay_repaired = 1; close(fd); /* @@ -1002,15 +1207,33 @@ ps_manifest_compact(void) uint64_t nrec = 0; int n; + if (ps_store_owner_require_current(manifest_owner) != 0) + return -1; + if (__atomic_load_n(&manifest_poisoned, __ATOMIC_ACQUIRE)) return -1; n = snprintf(tmp, sizeof(tmp), "%s.tmp", manifest_path); if (n < 0 || (size_t) n >= sizeof(tmp)) return -1; - fd = open(tmp, O_WRONLY | O_CREAT | O_TRUNC, 0600); + fd = open(tmp, O_WRONLY | O_CREAT | O_TRUNC | O_CLOEXEC | + O_NOFOLLOW | O_NONBLOCK, 0600); if (fd < 0) return -1; + { + struct stat st; + int fstat_rc = fstat(fd, &st); + + if (fstat_rc != 0 || !S_ISREG(st.st_mode) || st.st_nlink != 1) + { + int save_errno = fstat_rc != 0 ? errno : EINVAL; + + close(fd); + unlink(tmp); + errno = save_errno; + return -1; + } + } for (uint32_t i = 0; i < ps_layer_map.nlayers && rc == 0; i++) { PsManifestLayerDisk disk; diff --git a/contrib/pagestore/pagestore_manifest.h b/contrib/pagestore/pagestore_manifest.h index 0d4c8296b5f66..54fd2451cec2c 100644 --- a/contrib/pagestore/pagestore_manifest.h +++ b/contrib/pagestore/pagestore_manifest.h @@ -24,6 +24,11 @@ typedef struct PsFlushWatermark extern int ps_manifest_open(const char *store_dir); extern void ps_manifest_close(void); extern int ps_manifest_poisoned(void); +extern int ps_manifest_replay_repaired(void); +extern int ps_manifest_replay_had_manifest(void); +/* 1 means a durable replay-repair quarantine marker exists, 0 means absent, + * and -1 means the marker itself is invalid. */ +extern int ps_manifest_orphan_sweep_inhibited(void); extern int ps_manifest_replay(PsLayerMap *map); extern int ps_manifest_add_layer(const PsLayerDesc *desc); extern int ps_manifest_set_remote_location(uint64_t layer_id, diff --git a/contrib/pagestore/pagestore_store_owner.c b/contrib/pagestore/pagestore_store_owner.c new file mode 100644 index 0000000000000..25cdebdf41c99 --- /dev/null +++ b/contrib/pagestore/pagestore_store_owner.c @@ -0,0 +1,349 @@ +/*------------------------------------------------------------------------- + * + * pagestore_store_owner.c + * + * The lease is a persistent .pagestore.lock inode protected by flock(). + * Providers in one process share one open lock and reference count; separate + * processes contend on the kernel lock. The child side of fork discards its + * inherited references and descriptors before it can acquire a new lease. + * + *------------------------------------------------------------------------- + */ +#ifndef _GNU_SOURCE +#define _GNU_SOURCE +#endif + +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include + +#include "pagestore_store_owner.h" + +typedef struct PsStoreOwnerEntry +{ + dev_t root_dev; + ino_t root_ino; + char root[PATH_MAX]; + int root_fd; + int lock_fd; + unsigned int refs; + struct PsStoreOwnerEntry *next; +} PsStoreOwnerEntry; + +struct PsStoreOwner +{ + PsStoreOwnerEntry *entry; + pid_t pid; +}; + +static pthread_mutex_t owner_lock = PTHREAD_MUTEX_INITIALIZER; +static pthread_once_t owner_atfork_once = PTHREAD_ONCE_INIT; +static PsStoreOwnerEntry *owner_entries; +static int owner_atfork_rc; + +static void +owner_prepare(void) +{ + (void) pthread_mutex_lock(&owner_lock); +} + +static void +owner_parent(void) +{ + (void) pthread_mutex_unlock(&owner_lock); +} + +static void +owner_child(void) +{ + PsStoreOwnerEntry *entry; + + /* Closing this process's descriptor does not release the parent's open-file + * description. Do not free entries: inherited handles still point at them + * and release() must be able to identify them as child-invalid handles. */ + for (entry = owner_entries; entry != NULL; entry = entry->next) + { + if (entry->lock_fd >= 0) + (void) close(entry->lock_fd); + if (entry->root_fd >= 0) + (void) close(entry->root_fd); + entry->root_fd = -1; + entry->lock_fd = -1; + entry->refs = 0; + } + (void) pthread_mutex_unlock(&owner_lock); +} + +static void +owner_register_atfork(void) +{ + owner_atfork_rc = pthread_atfork(owner_prepare, owner_parent, owner_child); +} + +static int +owner_prepare_root(const char *store_dir, char *root, size_t root_len, + struct stat *root_st) +{ + int n; + + if (store_dir == NULL || store_dir[0] == '\0') + { + errno = EINVAL; + return -1; + } + if (mkdir(store_dir, 0700) != 0 && errno != EEXIST) + return -1; + if (realpath(store_dir, root) == NULL) + return -1; + n = (int) strlen(root); + if (n <= 0 || (size_t) n >= root_len) + { + errno = ENAMETOOLONG; + return -1; + } + if (stat(root, root_st) != 0) + return -1; + if (!S_ISDIR(root_st->st_mode)) + { + errno = ENOTDIR; + return -1; + } + return 0; +} + +static int +owner_open_lock(const char *root, const struct stat *root_st, int *root_fd_out) +{ + char path[PATH_MAX]; + struct stat st; + int root_fd; + int fd; + int n; + + root_fd = open(root, O_RDONLY | O_DIRECTORY | O_CLOEXEC | O_NOFOLLOW); + if (root_fd < 0) + return -1; + if (fstat(root_fd, &st) != 0 || !S_ISDIR(st.st_mode) || + st.st_dev != root_st->st_dev || st.st_ino != root_st->st_ino) + { + int save_errno = errno; + + close(root_fd); + errno = save_errno != 0 ? save_errno : EAGAIN; + return -1; + } + n = snprintf(path, sizeof(path), ".pagestore.lock"); + if (n < 0 || (size_t) n >= sizeof(path)) + { + close(root_fd); + errno = ENAMETOOLONG; + return -1; + } + fd = openat(root_fd, path, + O_RDWR | O_CREAT | O_CLOEXEC | O_NOFOLLOW, 0600); + if (fd < 0) + { + int save_errno = errno; + + close(root_fd); + errno = save_errno; + return -1; + } + if (fstat(fd, &st) != 0) + { + int save_errno = errno; + + close(fd); + close(root_fd); + errno = save_errno; + return -1; + } + if (!S_ISREG(st.st_mode) || st.st_nlink != 1) + { + close(fd); + close(root_fd); + errno = EINVAL; + return -1; + } + if (flock(fd, LOCK_EX | LOCK_NB) != 0) + { + int save_errno = errno; + + close(fd); + close(root_fd); + errno = save_errno; + return -1; + } + *root_fd_out = root_fd; + return fd; +} + +int +ps_store_owner_acquire(const char *store_dir, PsStoreOwner **owner_out) +{ + char root[PATH_MAX]; + struct stat root_st; + PsStoreOwnerEntry *entry; + PsStoreOwner *handle; + int fd = -1; + int root_fd = -1; + int save_errno; + int new_entry = 0; + + if (owner_out == NULL) + { + errno = EINVAL; + return -1; + } + *owner_out = NULL; + handle = malloc(sizeof(*handle)); + if (handle == NULL) + return -1; + if (pthread_once(&owner_atfork_once, owner_register_atfork) != 0 || + owner_atfork_rc != 0) + { + free(handle); + errno = EAGAIN; + return -1; + } + if (owner_prepare_root(store_dir, root, sizeof(root), &root_st) != 0) + { + save_errno = errno; + free(handle); + errno = save_errno; + return -1; + } + if (pthread_mutex_lock(&owner_lock) != 0) + { + free(handle); + errno = EBUSY; + return -1; + } + for (entry = owner_entries; entry != NULL; entry = entry->next) + if (entry->root_dev == root_st.st_dev && entry->root_ino == root_st.st_ino) + break; + if (entry != NULL && entry->refs != 0) + { + entry->refs++; + pthread_mutex_unlock(&owner_lock); + } + else + { + if (entry == NULL) + { + entry = calloc(1, sizeof(*entry)); + if (entry == NULL) + { + save_errno = errno; + pthread_mutex_unlock(&owner_lock); + free(handle); + errno = save_errno; + return -1; + } + entry->root_dev = root_st.st_dev; + entry->root_ino = root_st.st_ino; + snprintf(entry->root, sizeof(entry->root), "%s", root); + entry->root_fd = -1; + entry->lock_fd = -1; + entry->next = owner_entries; + owner_entries = entry; + new_entry = 1; + } + fd = owner_open_lock(root, &root_st, &root_fd); + if (fd < 0) + { + save_errno = errno; + if (new_entry && entry->refs == 0 && entry->lock_fd < 0) + { + owner_entries = entry->next; + free(entry); + } + pthread_mutex_unlock(&owner_lock); + free(handle); + errno = save_errno; + return -1; + } + entry->root_fd = root_fd; + entry->lock_fd = fd; + entry->refs = 1; + pthread_mutex_unlock(&owner_lock); + } + handle->entry = entry; + handle->pid = getpid(); + *owner_out = handle; + return 0; +} + +const char * +ps_store_owner_root(const PsStoreOwner *owner) +{ + if (owner == NULL || owner->entry == NULL || owner->pid != getpid()) + return NULL; + return owner->entry->root; +} + +int +ps_store_owner_require_current(const PsStoreOwner *owner) +{ + if (owner == NULL || owner->entry == NULL) + { + errno = EINVAL; + return -1; + } + if (owner->pid != getpid()) + { + errno = ECHILD; + return -1; + } + return 0; +} + +void +ps_store_owner_release(PsStoreOwner *owner) +{ + PsStoreOwnerEntry **link; + + if (owner == NULL) + return; + if (owner->pid != getpid()) + { + free(owner); + return; + } + if (pthread_mutex_lock(&owner_lock) != 0) + { + /* Never mutate the shared entry without its mutex. A mutex failure is + * catastrophic and deliberately leaks this handle/lease rather than + * risking an unsynchronised decrement or a premature unlock. */ + return; + } + if (owner->entry != NULL && owner->entry->refs > 0) + { + owner->entry->refs--; + if (owner->entry->refs == 0) + { + (void) close(owner->entry->lock_fd); + (void) close(owner->entry->root_fd); + owner->entry->lock_fd = -1; + owner->entry->root_fd = -1; + link = &owner_entries; + while (*link != owner->entry && *link != NULL) + link = &(*link)->next; + if (*link == owner->entry) + { + *link = owner->entry->next; + free(owner->entry); + } + } + } + pthread_mutex_unlock(&owner_lock); + free(owner); +} diff --git a/contrib/pagestore/pagestore_store_owner.h b/contrib/pagestore/pagestore_store_owner.h new file mode 100644 index 0000000000000..45410105a8136 --- /dev/null +++ b/contrib/pagestore/pagestore_store_owner.h @@ -0,0 +1,29 @@ +/*------------------------------------------------------------------------- + * + * pagestore_store_owner.h + * + * Same-process reference-counted ownership of a POSIX pagestore root. + * + *------------------------------------------------------------------------- + */ +#ifndef PAGESTORE_STORE_OWNER_H +#define PAGESTORE_STORE_OWNER_H + +typedef struct PsStoreOwner PsStoreOwner; + +/* Acquire an exclusive, non-blocking lease for the canonical store root. */ +extern int ps_store_owner_acquire(const char *store_dir, + PsStoreOwner **owner_out); + +/* The canonical root spelling held by an acquired lease. */ +extern const char *ps_store_owner_root(const PsStoreOwner *owner); + +/* Require that a lease belongs to this process. Inherited handles fail with + * ECHILD before callers touch cached descriptors, paths, or provider locks. */ +extern int ps_store_owner_require_current(const PsStoreOwner *owner); + +/* Release one same-process reference. A handle inherited across fork is a + * no-op in the child, so it cannot release or reuse the parent's ownership. */ +extern void ps_store_owner_release(PsStoreOwner *owner); + +#endif /* PAGESTORE_STORE_OWNER_H */ diff --git a/contrib/pagestore/pagestore_tiering_test.c b/contrib/pagestore/pagestore_tiering_test.c index 6dfb79b1ae533..3eef453c3aecf 100644 --- a/contrib/pagestore/pagestore_tiering_test.c +++ b/contrib/pagestore/pagestore_tiering_test.c @@ -5,7 +5,12 @@ * *------------------------------------------------------------------------- */ +#include +#include +#include #include +#include +#include #include #include #include @@ -44,6 +49,214 @@ static BlockingGate publication_gate; static BlockingGate verification_gate; static PsLayerStore blocking_store; +static int mock_close_calls; + +static void +mock_storage_close(void) +{ + mock_close_calls++; + PsStoragePosix.close(); +} + +static int +mock_storage_failed_open(const char *path, uint64_t size) +{ + /* Model an attach failure after acquiring the delegated POSIX lease. */ + if (PsStoragePosix.open(path, size) != 0) + return -1; + PsStoragePosix.close(); + errno = EIO; + return -1; +} + +static int +mock_layer_failed_open(const char *path) +{ + (void) path; + errno = EIO; + return -1; +} + +static void +test_core_provider_lifecycle(void) +{ + char dir[] = "/tmp/ps-core-owner-XXXXXX"; + PsStorage mock = PsStoragePosix; + PsLayerStore mock_layer = PsLayerStoreLocal; + PsKey key = {1, 1, 5, 0, PS_KLASS_RELATION}; + unsigned char page[PSZ] = {0}; + pid_t pid; + int status; + int rc; + + if (mkdtemp(dir) == NULL) + { + check(0, "create core lifecycle test store"); + return; + } + page_size = PSZ; + segment_size = 1024 * 1024; + flush_pages = 1024; + cache_pages = 0; + use_layers = 1; + ps_nshards = 1; + mock.name = "mock-non-posix"; + mock.close = mock_storage_close; + mock.open = mock_storage_failed_open; + ps_storage = &mock; + mock_close_calls = 0; + errno = 0; + rc = ps_core_open(dir); + check(rc != 0 && errno == EIO && mock_close_calls == 0, + "failed non-POSIX open never calls catalog-publishing close"); + ps_core_close(); + ps_core_close(); + check(mock_close_calls == 0, "repeated close after provider-open failure is harmless"); + mock.open = PsStoragePosix.open; + mock_layer.open = mock_layer_failed_open; + ps_layer_store = &mock_layer; + errno = 0; + rc = ps_core_open(dir); + check(rc != 0 && errno == EIO && mock_close_calls == 1, + "late core open failure closes fully initialized non-POSIX storage once"); + ps_core_close(); + ps_core_close(); + check(mock_close_calls == 1, "close after late startup failure does not close again"); + ps_layer_store = &PsLayerStoreLocal; + mock_close_calls = 0; + check(ps_core_open(dir) == 0, "reopen after failed provider initialization"); + ps_core_close(); + check(mock_close_calls == 0, "core leaves non-POSIX teardown to its caller"); + ps_storage->close(); + check(mock_close_calls == 1, "caller closes non-POSIX provider exactly once"); + + ps_storage = &PsStoragePosix; + check(ps_core_open(dir) == 0, "POSIX core reopens after caller teardown"); + check(append_page(0, &key, 0, page, 1, NULL) == 0, + "buffer a parent page before fork"); + pid = fork(); + if (pid == 0) + { + PsChannel channel; + PageVer version = {.seg = 0}; + int good = 1; + + alarm(3); + errno = 0; + ps_core_close(); + good = good && errno == ECHILD; + good = good && ps_core_maintenance() == -1 && errno == ECHILD; + good = good && append_page(0, &key, 1, page, 2, NULL) == -1 && + errno == ECHILD; + good = good && fork_grow(0, &key, 3, 3) == -1 && errno == ECHILD; + good = good && read_through(0, &key, 0, UINT64_MAX, 0) == NULL && + errno == ECHILD; + good = good && read_version(&version, page) == -1 && errno == ECHILD; + good = good && read_resolve(0, &key, 0, UINT64_MAX, 0, page, NULL) == -1 && + errno == ECHILD; + memset(&channel, 0, sizeof(channel)); + channel.opcode = PS_OP_CREATE; + channel.key = key; + good = good && ps_handle_meta(&channel) == 1 && + channel.status == PS_STATUS_ERROR; + good = good && ps_core_open(dir) == -1 && errno == ECHILD; + _exit(good ? 0 : 1); + } + check(pid > 0 && waitpid(pid, &status, 0) == pid && + WIFEXITED(status) && WEXITSTATUS(status) == 0, + "forked core cannot flush, mutate, or reopen inherited state"); + check(append_page(0, &key, 1, page, 2, NULL) == 0, + "parent remains writable after child rejects inherited core"); + ps_core_close(); + ps_core_close(); + check(ps_core_open(dir) == 0, "POSIX core reopens after repeated close"); + ps_core_close(); +} + +static void +test_legacy_local_uri_reopen(void) +{ + char root[] = "/tmp/ps-legacy-uri-XXXXXX"; + char store[PATH_MAX]; + char previous_cwd[PATH_MAX]; + const char *spellings[] = {"store", "alias", "store/../store"}; + PsKey key = {1, 1, 5, 0, PS_KLASS_RELATION}; + unsigned char page[PSZ]; + unsigned char out[PSZ]; + + if (getcwd(previous_cwd, sizeof(previous_cwd)) == NULL || + mkdtemp(root) == NULL) + { + check(0, "prepare legacy URI fixture"); + return; + } + snprintf(store, sizeof(store), "%s/store", root); + if (chdir(root) != 0) + { + check(0, "enter legacy URI fixture directory"); + return; + } + memset(page, 0x5a, sizeof(page)); + flush_pages = 1; + check(ps_core_open(store) == 0 && + append_page(0, &key, 0, page, 1, NULL) == 0, + "persist a page for legacy path upgrade"); + ps_core_close(); + check(symlink("store", "alias") == 0, "create legacy store alias"); + for (size_t n = 0; n < sizeof(spellings) / sizeof(spellings[0]); n++) + { + char orphan[PATH_MAX]; + int fd; + int opened; + int normalized = 1; + + check(ps_manifest_open(store) == 0 && + ps_manifest_replay(&ps_layer_map) == 0 && ps_layer_map.nlayers > 0, + "replay manifest for legacy spelling fixture"); + /* Model an older binary's on-disk spelling, without relying on the new + * provider's canonical path writer to produce that old format. */ + for (uint32_t i = 0; i < ps_layer_map.nlayers; i++) + { + PsLayerDesc *layer = &ps_layer_map.layers[i]; + + for (uint32_t j = 0; j < layer->location_count; j++) + if (layer->locations[j].tier == PS_LAYER_TIER_LOCAL_HOT || + layer->locations[j].tier == PS_LAYER_TIER_LOCAL_COLD) + snprintf(layer->locations[j].uri, sizeof(layer->locations[j].uri), + "%s/layer_%u_%016llx", spellings[n], + (unsigned int) (layer->layer_id >> 48), + (unsigned long long) layer->layer_id); + } + check(ps_manifest_compact() == 0, "persist old local URI spelling"); + ps_manifest_close(); + snprintf(orphan, sizeof(orphan), "%s/layer_0_000000000000ffff", "store"); + fd = open(orphan, O_CREAT | O_EXCL | O_WRONLY, 0600); + check(fd >= 0 && close(fd) == 0, "seed canonical orphan before upgrade"); + opened = ps_core_open(spellings[n]) == 0; + check(opened, "upgrade reopens relative, symlinked, or dot-dot store spelling"); + if (!opened) + continue; + check(read_resolve(0, &key, 0, UINT64_MAX, 0, out, NULL) == 1 && + memcmp(page, out, sizeof(page)) == 0, + "upgrade preserves referenced page bytes"); + for (uint32_t i = 0; i < ps_layer_map.nlayers; i++) + for (uint32_t j = 0; j < ps_layer_map.layers[i].location_count; j++) + { + const PsLayerLocation *location = &ps_layer_map.layers[i].locations[j]; + + if ((location->tier == PS_LAYER_TIER_LOCAL_HOT || + location->tier == PS_LAYER_TIER_LOCAL_COLD) && + strncmp(location->uri, store, strlen(store)) != 0) + normalized = 0; + } + check(normalized && access(orphan, F_OK) != 0, + "upgrade canonicalizes live URIs and reclaims only the orphan"); + check(ps_manifest_compact() == 0, "canonical spelling can be persisted"); + ps_core_close(); + } + check(chdir(previous_cwd) == 0, "restore test working directory"); +} + static int blocking_upload(const PsLayerDesc *layer) { @@ -512,6 +725,8 @@ main(void) pthread_cond_destroy(&verification_gate.cond); pthread_mutex_destroy(&verification_gate.mutex); unsetenv("PAGESTORE_OBJECT_DIR"); + test_core_provider_lifecycle(); + test_legacy_local_uri_reopen(); printf("pagestore_tiering_test: %d checks, %d failed\n", run, failed); return failed ? 1 : 0; } diff --git a/contrib/pagestore/pagestore_wal_reclaim_core_test.c b/contrib/pagestore/pagestore_wal_reclaim_core_test.c index e89bbf8a9ac33..b96933d4bf845 100644 --- a/contrib/pagestore/pagestore_wal_reclaim_core_test.c +++ b/contrib/pagestore/pagestore_wal_reclaim_core_test.c @@ -592,6 +592,9 @@ test_residual_prefix_retry_after_reopen(void) configure_core(); check(prepare_store(store, WAL_TOTAL, 0, 0, 1), "construct reclaimable WAL for a residual-prefix retry"); + /* Crash the child-owned recovery instance, not a fork-inherited provider. + * Ownership fencing deliberately forbids mutation through the latter. */ + close_store(); { pid_t pid; int status = 0; @@ -601,6 +604,8 @@ test_residual_prefix_retry_after_reopen(void) pid = fork(); if (pid == 0) { + if (ps_core_open(store) != 0) + _exit(2); (void) ps_test_wal_reclaim_maintenance(); _exit(1); } @@ -610,7 +615,6 @@ test_residual_prefix_retry_after_reopen(void) "crash leaves the authorized residual WAL prefix on disk"); unsetenv("PAGESTORE_TEST_WAL_RECLAIM_CRASH_BEFORE_UNLINK"); } - close_store(); memset(&metrics, 0, sizeof(metrics)); ps_core_set_metrics_header(&metrics); check(ps_backpressure_configure(0, 0, WAL_SEGMENT, WAL_SEGMENT / 2) == 0 && @@ -654,6 +658,8 @@ test_residual_prefix_and_suffix_debt(void) remove_tree(store); return; } + close_store(); + opened = 0; { pid_t pid; int status = 0; @@ -664,6 +670,8 @@ test_residual_prefix_and_suffix_debt(void) pid = fork(); if (pid == 0) { + if (ps_core_open(store) != 0) + _exit(2); (void) ps_test_wal_reclaim_maintenance(); _exit(1); } @@ -673,8 +681,6 @@ test_residual_prefix_and_suffix_debt(void) "crash leaves one residual and three post-frontier segments"); unsetenv("PAGESTORE_TEST_WAL_RECLAIM_CRASH_AFTER_UNLINK_SEGMENT_NO"); } - close_store(); - opened = 0; memset(&metrics, 0, sizeof(metrics)); ps_core_set_metrics_header(&metrics); check(ps_backpressure_configure(0, 0, 3 * (uint64_t) WAL_SEGMENT, diff --git a/contrib/pagestore/spdk_build.sh b/contrib/pagestore/spdk_build.sh index 667b85976cb4d..1cd1e3219d0e5 100755 --- a/contrib/pagestore/spdk_build.sh +++ b/contrib/pagestore/spdk_build.sh @@ -38,7 +38,7 @@ cc -O2 -Wall -Wextra -DPAGESTORE_SPDK -I"$here" $cflags \ -o "$out" \ "$here/pagestore_daemon_spdk.c" "$here/pagestore_core.c" \ "$here/pagestore_fault.c" \ - "$here/storage_spdk.c" "$here/storage_posix.c" \ + "$here/pagestore_store_owner.c" "$here/storage_spdk.c" "$here/storage_posix.c" \ "$here/pagestore_layer.c" "$here/pagestore_layer_store.c" \ "$here/pagestore_manifest.c" "$here/pagestore_memtable.c" "$here/pagestore_pgcache.c" \ "$here/pagestore_prune.c" "$here/pagestore_retention.c" \ diff --git a/contrib/pagestore/storage_posix.c b/contrib/pagestore/storage_posix.c index ebfd8641855b7..6f8fe8322eaf1 100644 --- a/contrib/pagestore/storage_posix.c +++ b/contrib/pagestore/storage_posix.c @@ -27,10 +27,12 @@ #include "pagestore_storage.h" #include "pagestore_ipc.h" +#include "pagestore_store_owner.h" #include "pagestore_wal_store.h" /* bounded well under the 4096-byte path buffers so suffixes never truncate */ static char posix_dir[2048]; +static PsStoreOwner *posix_owner; /* * One cached OS fd per segment shard+id (opened lazily, never closed during a @@ -97,6 +99,14 @@ static pthread_mutex_t posix_walidx_locks_lock = PTHREAD_MUTEX_INITIALIZER; static void posix_wal_locks_clear(void); static void posix_walidx_locks_clear(void); +static void posix_close(void); + +/* Check before touching copied caches, descriptors, or provider mutexes. */ +static int +posix_require_owner(void) +{ + return ps_store_owner_require_current(posix_owner); +} static int posix_log_read(const char *name, uint64_t off, void *buf, uint32_t len); static int posix_log_append(const char *name, const void *buf, uint32_t len); @@ -263,10 +273,27 @@ posix_open(const char *path, uint64_t segment_size) const char *fail_seg_remove_dir_fsync; const char *fail_seg_size; int dfd; + int save_errno; (void) segment_size; /* the file backend has no fixed-region layout */ - if (mkdir(path, 0700) != 0 && errno != EEXIST) + /* Explicit OPEN has historically replaced the previous POSIX instance. Keep + * that reopen behavior, but make the replacement release its old lease before + * attempting to claim a different root. */ + if (posix_owner != NULL) + { + if (ps_store_owner_require_current(posix_owner) != 0) + return -1; + posix_close(); + } + if (ps_store_owner_acquire(path, &posix_owner) != 0) return -1; + if (snprintf(posix_dir, sizeof(posix_dir), "%s", + ps_store_owner_root(posix_owner)) < 0 || + strlen(ps_store_owner_root(posix_owner)) >= sizeof(posix_dir)) + { + errno = ENAMETOOLONG; + goto fail; + } /* Backend ownership permits an explicit OPEN to abandon a previous POSIX * instance. Serialize teardown before installing the new path so stale fd @@ -328,7 +355,6 @@ posix_open(const char *path, uint64_t segment_size) value = getenv("PAGESTORE_TEST_FAIL_FORK_META_REWRITE_DIR_FSYNC"); test_fail_fork_meta_rewrite_dir_fsync = value ? atoi(value) : 0; } - snprintf(posix_dir, sizeof(posix_dir), "%s", path); /* A prior metadata rename whose directory sync failed must be made durable * before this process can accept writes against its visible replacement. */ dfd = open(posix_dir, O_RDONLY | O_DIRECTORY); @@ -336,23 +362,42 @@ posix_open(const char *path, uint64_t segment_size) { if (dfd >= 0) close(dfd); - return -1; + goto fail; } if (close(dfd) != 0) - return -1; + goto fail; /* A successful reopen has reconciled any ambiguous post-rename state. */ posix_wal_locks_clear(); return 0; + +fail: + save_errno = errno; + posix_close(); + errno = save_errno; + return -1; } static void posix_close(void) { + if (posix_owner != NULL && ps_store_owner_require_current(posix_owner) != 0) + { + /* An inherited provider remains unusable until exec. Keep the stale + * token so repeated close calls take this branch without touching any + * copied provider mutex or cache. */ + return; + } pthread_mutex_lock(&seg_fds_lock); free_shard_caches(); pthread_mutex_unlock(&seg_fds_lock); posix_wal_locks_clear(); posix_walidx_locks_clear(); + posix_dir[0] = '\0'; + if (posix_owner != NULL) + { + ps_store_owner_release(posix_owner); + posix_owner = NULL; + } } static int @@ -360,6 +405,9 @@ posix_sync(void) { int rc = 0; + if (posix_require_owner() != 0) + return -1; + /* * Walk the shared seg-fd cache under seg_fds_lock so a concurrent shard * worker cannot lazily open a segment and realloc seg_fds[]/seg_fds_caps[] @@ -409,6 +457,9 @@ posix_seg_remove(uint32_t shard, int seg) int dfd; int rc = 0; + if (posix_require_owner() != 0) + return -1; + seg_path(path, sizeof(path), shard, seg); pthread_mutex_lock(&seg_fds_lock); if (shard < (uint32_t) seg_shards_cap && seg >= 0 && @@ -452,12 +503,16 @@ static int posix_seg_write(uint32_t shard, int seg, uint64_t off, const void *buf, uint32_t len) { + int fd; + + if (posix_require_owner() != 0) + return -1; if (posix_seg_rewrite_poisoned) { errno = EIO; return -1; } - int fd = seg_fd(shard, seg, 1); + fd = seg_fd(shard, seg, 1); if (fd < 0) return -1; @@ -478,12 +533,16 @@ posix_seg_write(uint32_t shard, int seg, uint64_t off, const void *buf, static int posix_seg_read(uint32_t shard, int seg, uint64_t off, void *buf, uint32_t len) { + int fd; + + if (posix_require_owner() != 0) + return -1; if (posix_seg_rewrite_poisoned) { errno = EIO; return -1; } - int fd = seg_fd(shard, seg, 0); + fd = seg_fd(shard, seg, 0); if (fd < 0) return -1; @@ -498,6 +557,9 @@ posix_seg_size(uint32_t shard, int seg) char path[4096]; struct stat st; + if (posix_require_owner() != 0) + return -1; + if (posix_seg_rewrite_poisoned) { errno = EIO; @@ -529,6 +591,9 @@ posix_seg_rewrite(uint32_t shard, int seg, const void *buf, uint64_t len) int cache_locked = 0; int rc = -1; + if (posix_require_owner() != 0) + return -1; + if (posix_seg_rewrite_poisoned) { errno = EIO; @@ -750,9 +815,13 @@ static int posix_wal_append(uint32_t tl, const void *a, uint32_t alen, const void *b, uint32_t blen) { - PosixWalLock *lock = posix_wal_lock_for(tl); + PosixWalLock *lock; int rc; + if (posix_require_owner() != 0) + return -1; + lock = posix_wal_lock_for(tl); + if (lock == NULL) return -1; pthread_mutex_lock(&lock->lock); @@ -765,12 +834,16 @@ posix_wal_append(uint32_t tl, const void *a, uint32_t alen, static int posix_wal_read(uint32_t tl, uint64_t off, void *buf, uint32_t len) { - PosixWalLock *lock = posix_wal_lock_for(tl); + PosixWalLock *lock; char path[4096]; int fd; ssize_t n; uint32_t done = 0; + if (posix_require_owner() != 0) + return -1; + lock = posix_wal_lock_for(tl); + if (lock == NULL) return -1; pthread_mutex_lock(&lock->lock); @@ -827,9 +900,13 @@ posix_wal_truncate_locked(uint32_t tl, uint64_t len) static int posix_wal_truncate(uint32_t tl, uint64_t len) { - PosixWalLock *lock = posix_wal_lock_for(tl); + PosixWalLock *lock; int rc; + if (posix_require_owner() != 0) + return -1; + lock = posix_wal_lock_for(tl); + if (lock == NULL) return -1; pthread_mutex_lock(&lock->lock); @@ -848,7 +925,7 @@ posix_wal_truncate(uint32_t tl, uint64_t len) static int posix_wal_rewrite_prefix(uint32_t tl, uint64_t keep_off) { - PosixWalLock *lock = posix_wal_lock_for(tl); + PosixWalLock *lock; char path[4096]; char tmp[4096]; unsigned char buf[64 * 1024]; @@ -859,6 +936,10 @@ posix_wal_rewrite_prefix(uint32_t tl, uint64_t keep_off) int dfd = -1; int rc = -1; + if (posix_require_owner() != 0) + return -1; + lock = posix_wal_lock_for(tl); + if (lock == NULL) return -1; pthread_mutex_lock(&lock->lock); @@ -1185,6 +1266,9 @@ posix_walidx_append(uint32_t tl, uint32_t shard, uint64_t epoch, uint64_t old_length = 0; int rc; + if (posix_require_owner() != 0) + return -1; + if (posix_walidx_name(tl, shard, epoch, name, sizeof(name)) != 0) return -1; lock = posix_walidx_lock_for(tl, shard); @@ -1274,6 +1358,9 @@ posix_walidx_read(uint32_t tl, uint32_t shard, uint64_t epoch, uint64_t off, uint64_t length; int rc; + if (posix_require_owner() != 0) + return -1; + if (posix_walidx_name(tl, shard, epoch, name, sizeof(name)) != 0) return -1; if (epoch == 0) @@ -1307,6 +1394,9 @@ posix_walidx_truncate(uint32_t tl, uint32_t shard, uint64_t epoch, uint64_t len) int rc = 0; PosixWalIdxLock *lock; + if (posix_require_owner() != 0) + return -1; + if (posix_walidx_name(tl, shard, epoch, name, sizeof(name)) != 0 || snprintf(path, sizeof(path), "%s/%s", posix_dir, name) < 0 || strlen(posix_dir) + 1 + strlen(name) >= sizeof(path)) @@ -1351,6 +1441,9 @@ posix_walidx_epoch_create(uint32_t tl, uint32_t shard, uint64_t epoch) int fd = -1; int rc = -1; + if (posix_require_owner() != 0) + return -1; + if (epoch == 0 || posix_walidx_name(tl, shard, epoch, name, sizeof(name)) != 0 || snprintf(path, sizeof(path), "%s/%s", posix_dir, name) < 0 || @@ -1384,7 +1477,7 @@ posix_walidx_epoch_create(uint32_t tl, uint32_t shard, uint64_t epoch) static int posix_walidx_epoch_gc(uint32_t tl, const uint64_t *keep_epochs, - uint32_t nshards) + uint32_t nshards) { char prefix[128]; struct dirent *entry; @@ -1395,6 +1488,9 @@ posix_walidx_epoch_gc(uint32_t tl, const uint64_t *keep_epochs, int rc = -1; int n; + if (posix_require_owner() != 0) + return -1; + if (keep_epochs == NULL || nshards == 0) return -1; /* The manifest cutover drains old-epoch appenders before selecting each @@ -1505,6 +1601,9 @@ posix_walidx_reclaim_bytes(uint32_t tl, const uint64_t *keep_epochs, int rc = -1; int n; + if (posix_require_owner() != 0) + return -1; + if (keep_epochs == NULL || covered_offsets == NULL || observed_offsets == NULL || nshards == 0 || tail_bytes == NULL || obsolete_bytes == NULL) @@ -2141,6 +2240,9 @@ posix_timeline_wal_cleanup(uint32_t tl) int rc = -1; int readdir_errno; + if (posix_require_owner() != 0) + return -1; + if (tl == 0 || tl == UINT32_MAX || snprintf(wal_name, sizeof(wal_name), "wal_%u", tl) < 0 || snprintf(wal_rewrite_name, sizeof(wal_rewrite_name), @@ -2433,6 +2535,9 @@ posix_log_read(const char *name, uint64_t off, void *buf, uint32_t len) ssize_t n; uint32_t done = 0; + if (posix_require_owner() != 0) + return -1; + snprintf(path, sizeof(path), "%s/%s", posix_dir, name); fd = open(path, O_RDONLY); if (fd < 0) @@ -2466,6 +2571,9 @@ posix_log_append(const char *name, const void *buf, uint32_t len) { int rc; + if (posix_require_owner() != 0) + return -1; + pthread_mutex_lock(&posix_log_lock); rc = posix_log_append_locked(name, buf, len); pthread_mutex_unlock(&posix_log_lock); @@ -2491,6 +2599,9 @@ posix_log_truncate(const char *name, uint64_t len) int fd; int rc = 0; + if (posix_require_owner() != 0) + return -1; + snprintf(path, sizeof(path), "%s/%s", posix_dir, name); pthread_mutex_lock(&posix_log_lock); fd = open(path, O_WRONLY); @@ -2520,6 +2631,9 @@ posix_meta_rewrite(const void *buf, uint32_t len) int fd = -1, dfd = -1, rc = -1; ssize_t n; + if (posix_require_owner() != 0) + return -1; + snprintf(path, sizeof(path), "%s/timelines", posix_dir); snprintf(tmp, sizeof(tmp), "%s/timelines.tmp", posix_dir); pthread_mutex_lock(&posix_log_lock); @@ -2568,6 +2682,8 @@ posix_meta_rewrite(const void *buf, uint32_t len) static int posix_fork_meta_append(const void *buf, uint32_t len) { + if (posix_require_owner() != 0) + return -1; /* Standalone-test fault injection; ordinary deployments leave this zero. */ if (test_fail_fork_meta_append_at > 0 && --test_fail_fork_meta_append_at == 0) @@ -2597,6 +2713,9 @@ posix_fork_meta_size(uint64_t *len) struct stat st; int n; + if (posix_require_owner() != 0) + return -1; + if (len == NULL) { errno = EINVAL; @@ -2633,6 +2752,9 @@ posix_fork_meta_rewrite(const void *buf, uint32_t len) int fd = -1, dfd = -1, rc = -1; uint32_t off = 0; + if (posix_require_owner() != 0) + return -1; + if (buf == NULL && len != 0) return -1; pthread_mutex_lock(&posix_log_lock); diff --git a/contrib/pagestore/storage_posix_fork_owner_test.c b/contrib/pagestore/storage_posix_fork_owner_test.c new file mode 100644 index 0000000000000..62fe0f8c14078 --- /dev/null +++ b/contrib/pagestore/storage_posix_fork_owner_test.c @@ -0,0 +1,119 @@ +/* Focused fork invalidation tests for the POSIX provider lease. */ +#ifndef _GNU_SOURCE +#define _GNU_SOURCE +#endif + +#include +#include +#include +#include +#include +#include +#include +#include +#include + +#include "pagestore_layer.h" +#include "pagestore_manifest.h" +#include "pagestore_storage.h" + +static int checks; +static int failed; + +static void +check(int condition, const char *name) +{ + checks++; + if (!condition) + { + failed++; + fprintf(stderr, "FAIL: %s\n", name); + } +} + +int +main(void) +{ + char store[] = "/tmp/pagestore-fork-owner-XXXXXX"; + char path[1024]; + char got[sizeof("parent")]; + PsLayerDesc manifest_layer; + pid_t pid; + int status = 0; + + check(mkdtemp(store) != NULL, "create fork-owner store"); + check(PsStoragePosix.open(store, 0) == 0, "open fork-owner store"); + check(PsStoragePosix.seg_write(0, 0, 0, "parent", 6) == 0, + "write parent segment bytes"); + memset(&manifest_layer, 0, sizeof(manifest_layer)); + manifest_layer.layer_id = (3ULL << 48) | 1; + manifest_layer.kind = PS_LAYER_IMAGE; + check(ps_manifest_open(store) == 0 && + ps_manifest_add_layer(&manifest_layer) == 0, + "open and append parent manifest"); + + pid = fork(); + if (pid == 0) + { + int rc; + + alarm(2); + errno = 0; + rc = PsStoragePosix.seg_write(0, 0, 0, "child!", 6); + if (rc != -1 || errno != ECHILD) + _exit(1); + PsStoragePosix.close(); + PsStoragePosix.close(); + errno = 0; + rc = PsStoragePosix.open(store, 0); + _exit(rc == -1 && errno == ECHILD ? 0 : 2); + } + check(pid > 0 && waitpid(pid, &status, 0) == pid && + WIFEXITED(status) && WEXITSTATUS(status) == 0, + "forked provider rejects writes and reopen while parent owns store"); + pid = fork(); + if (pid == 0) + { + PsLayerDesc child_layer = manifest_layer; + int rc; + + alarm(2); + child_layer.layer_id++; + errno = 0; + rc = ps_manifest_add_layer(&child_layer); + if (rc != -1 || errno != ECHILD) + _exit(3); + errno = 0; + rc = ps_manifest_compact(); + if (rc != -1 || errno != ECHILD) + _exit(4); + errno = 0; + rc = ps_manifest_replay(&ps_layer_map); + if (rc != -1 || errno != ECHILD) + _exit(5); + ps_manifest_close(); + ps_manifest_close(); + _exit(0); + } + check(pid > 0 && waitpid(pid, &status, 0) == pid && + WIFEXITED(status) && WEXITSTATUS(status) == 0, + "forked manifest rejects append/compact/replay without hanging"); + + memset(got, 0, sizeof(got)); + check(PsStoragePosix.seg_read(0, 0, 0, got, 6) == 0 && + memcmp(got, "parent", 6) == 0, + "forked child cannot modify parent cached segment"); + ps_manifest_close(); + PsStoragePosix.close(); + + snprintf(path, sizeof(path), "%s/seg_00000000", store); + unlink(path); + snprintf(path, sizeof(path), "%s/layers.manifest", store); + unlink(path); + snprintf(path, sizeof(path), "%s/.pagestore.lock", store); + unlink(path); + rmdir(store); + printf("storage_posix_fork_owner_test: %d checks, %d failed\n", + checks, failed); + return failed != 0; +} diff --git a/contrib/pagestore/storage_spdk.c b/contrib/pagestore/storage_spdk.c index f331df4637ddf..16ce4cd43d91e 100644 --- a/contrib/pagestore/storage_spdk.c +++ b/contrib/pagestore/storage_spdk.c @@ -532,7 +532,7 @@ spdk_open(const char *path, uint64_t segment_size) { fprintf(stderr, "storage_spdk: spdk_env_init failed\n"); spdk_emit_env_diag(pci); - return -1; + goto fail; } memset(&trid, 0, sizeof(trid)); @@ -542,21 +542,21 @@ spdk_open(const char *path, uint64_t segment_size) { fprintf(stderr, "storage_spdk: could not attach NVMe at %s " "(bound to vfio-pci? see spdk_setup.sh)\n", pci); - return -1; + goto fail; } g_ns = spdk_nvme_ctrlr_get_ns(g_ctrlr, 1); if (!g_ns || !spdk_nvme_ns_is_active(g_ns)) { fprintf(stderr, "storage_spdk: namespace 1 not active\n"); - return -1; + goto fail; } g_sector = spdk_nvme_ns_get_sector_size(g_ns); if (g_segsize % g_sector != 0) { fprintf(stderr, "storage_spdk: segment size %llu not a multiple of " "sector %u\n", (unsigned long long) g_segsize, g_sector); - return -1; + goto fail; } g_secs_per_seg = (uint32_t) (g_segsize / g_sector); @@ -571,20 +571,31 @@ spdk_open(const char *path, uint64_t segment_size) fprintf(stderr, "storage_spdk: failed to allocate shard-%u context\n", i); for (uint32_t j = 0; j < i; j++) thread_free(&g_threads[j]); - if (g_ctrlr) - { - spdk_nvme_detach(g_ctrlr); - g_ctrlr = NULL; - } - g_ns = NULL; - PsStoragePosix.close(); - return -1; + goto fail; } } fprintf(stderr, "storage_spdk: %s ns1 sector=%u segsize=%llu nshards=%u\n", pci, g_sector, (unsigned long long) g_segsize, g_nshards); return 0; + +fail: + /* Initialization never established a writable segment catalog. Release + * the delegated POSIX lease without spdk_close()/super_write(), which would + * publish uninitialized or partially cleared segment counts. SPDK attach + * failures remain process-terminal (the frontend exits); this is not a + * same-process spdk_env_init retry path. */ + { + int save_errno = errno; + + if (g_ctrlr) + spdk_nvme_detach(g_ctrlr); + g_ctrlr = NULL; + g_ns = NULL; + PsStoragePosix.close(); + errno = save_errno; + } + return -1; } static void