"""Durable create-only local filesystem backend."""
from __future__ import annotations
import fcntl
import os
import stat
import tempfile
from collections.abc import Iterator
from contextlib import contextmanager
from pathlib import Path
from signal_dataset.errors import PublicationCollisionError, StorageError
from signal_dataset.storage.backends.local.capabilities import (
POSIX,
FilesystemCapabilities,
)
from signal_dataset.storage.backends.local.confinement import ConfinedDirectory
from signal_dataset.storage.contracts import ObjectInfo, ObjectStore, ObjectVersion
[docs]
class LocalObjectStore(ObjectStore):
"""Local storage, optionally confined with descriptor-relative operations.
A configured root is bound to the directory identity observed at
construction. Operations reject symlinks below it and cannot be redirected
by replacing the root or one of its ancestors.
"""
def __init__(
self,
*,
root: str | Path | None = None,
file_mode: int = 0o640,
capabilities: FilesystemCapabilities | None = None,
) -> None:
if file_mode < 0 or file_mode > 0o777:
raise ValueError("file_mode must be between 0o000 and 0o777")
self._root = None if root is None else Path(root).absolute()
self._file_mode = file_mode
self._capabilities = POSIX if capabilities is None else capabilities
self._confined = (
None
if self._root is None
else ConfinedDirectory(self._root, capabilities=self._capabilities)
)
# The annotation catalog is the only object ever modified in place, and
# swinging it needs a lock this filesystem may not have. Callers read
# this before doing any work rather than failing at the swap.
self.supports_compare_and_swap = self._capabilities.file_locks
def _path(self, uri: str) -> Path:
path = Path(uri).absolute()
if self._root is not None:
try:
path.relative_to(self._root)
except ValueError as exc:
raise ValueError(f"path {path} is outside local backend root {self._root}") from exc
return path
[docs]
def read(self, uri: str, *, generation: int | None = None) -> bytes:
if generation is not None:
raise ValueError("local objects do not have generations")
path = self._path(uri)
if self._confined is not None:
return self._confined.read(path)
return path.read_bytes()
[docs]
def read_version(self, uri: str) -> ObjectVersion:
path = self._path(uri)
generation = (
self._confined.generation(path)
if self._confined is not None
else path.stat().st_mtime_ns
)
return ObjectVersion(self.read(uri), generation)
[docs]
def info(self, uri: str, *, generation: int | None = None) -> ObjectInfo:
if generation is not None:
raise ValueError("local objects do not have generations")
path = self._path(uri)
if self._confined is not None:
return ObjectInfo(self._confined.size(path))
return ObjectInfo(path.stat().st_size)
[docs]
def create(self, uri: str, data: bytes, *, content_type: str) -> None:
del content_type
path = self._path(uri)
if self._confined is not None:
self._confined.create(path, data, mode=self._file_mode)
return
self._create_unconfined(path, data)
[docs]
def create_file(self, uri: str, path: Path, *, content_type: str) -> None:
del content_type
destination = self._path(uri)
if self._confined is not None:
with path.open("rb") as stream:
self._confined.create_stream(destination, stream, mode=self._file_mode)
return
self._create_unconfined_file(destination, path)
[docs]
def compare_and_swap(
self,
uri: str,
data: bytes,
*,
expected_generation: int | None,
content_type: str,
) -> None:
del content_type
path = self._path(uri)
if not self._capabilities.file_locks:
# Refused here as well as in ConfinedDirectory, and refused even
# for the create-only branch below. Letting the first update of a
# fresh object through and failing every later one is worse than
# not working: it looks supported until a second writer arrives.
raise StorageError(
f"cannot update {uri} through a mounted bucket: this object is "
"modified in place, which needs a compare-and-swap that neither "
"gcsfuse nor Mountpoint for Amazon S3 can provide. Address the "
"dataset by its native URI for this operation."
)
if expected_generation is None:
self.create(uri, data, content_type="application/json")
return
if self._confined is not None:
self._confined.compare_and_swap(
path,
data,
expected_generation=expected_generation,
mode=self._file_mode,
)
else:
path.parent.mkdir(parents=True, exist_ok=True)
with (path.parent / f".{path.name}.lock").open("a+b") as lock:
fcntl.flock(lock, fcntl.LOCK_EX)
try:
observed = path.stat().st_mtime_ns
except FileNotFoundError as exc:
raise PublicationCollisionError(uri) from exc
if observed != expected_generation:
raise PublicationCollisionError(uri)
self._replace_unconfined(path, data)
def _create_unconfined(self, path: Path, data: bytes) -> None:
path.parent.mkdir(parents=True, exist_ok=True)
descriptor, temporary_name = tempfile.mkstemp(prefix=f".{path.name}.", dir=path.parent)
temporary = Path(temporary_name)
try:
with os.fdopen(descriptor, "wb") as stream:
os.fchmod(stream.fileno(), self._file_mode)
stream.write(data)
stream.flush()
os.fsync(stream.fileno())
try:
os.link(temporary, path)
except FileExistsError as exc:
raise PublicationCollisionError(str(path)) from exc
directory_descriptor = os.open(path.parent, os.O_RDONLY)
try:
os.fsync(directory_descriptor)
finally:
os.close(directory_descriptor)
finally:
temporary.unlink(missing_ok=True)
def _create_unconfined_file(self, destination: Path, source: Path) -> None:
destination.parent.mkdir(parents=True, exist_ok=True)
try:
os.link(source, destination)
except FileExistsError as exc:
raise PublicationCollisionError(str(destination)) from exc
def _replace_unconfined(self, path: Path, data: bytes) -> None:
descriptor, temporary_name = tempfile.mkstemp(prefix=f".{path.name}.", dir=path.parent)
temporary = Path(temporary_name)
try:
with os.fdopen(descriptor, "wb") as stream:
stream.write(data)
stream.flush()
os.fsync(stream.fileno())
os.replace(temporary, path)
finally:
temporary.unlink(missing_ok=True)
[docs]
def list(self, prefix_uri: str) -> list[str]:
prefix = self._path(prefix_uri)
if self._confined is not None:
return self._confined.list(prefix, directory=prefix_uri.endswith("/"))
candidates = (
prefix.glob("*") if prefix_uri.endswith("/") else prefix.parent.glob(prefix.name + "*")
)
return sorted(str(path) for path in candidates if stat.S_ISREG(path.stat().st_mode))
[docs]
def generation(self, uri: str) -> int:
path = self._path(uri)
if self._confined is not None:
return self._confined.generation(path)
return path.stat().st_mtime_ns
[docs]
@contextmanager
def indexed_uri(self, uri: str) -> Iterator[str]:
"""Yield a stable descriptor path suitable for a native indexed reader."""
path = self._path(uri)
if self._confined is None:
yield str(path)
return
with self._confined.open_file(path) as descriptor:
yield f"/dev/fd/{descriptor}"