Persistent Reservations

SCSI-3 Persistent Reservations (PR) — cluster-aware device locking that survives SCSI bus resets. Used for shared-storage fencing in HA clusters. Wraps sg_persist from sg3_utils.

sts.pr

SCSI Persistent Reservation (PR) device management.

Coordinates shared access to SCSI devices between multiple initiators via registration, reservation, preemption, and release operations.

PRDevice pydantic-model

Bases: StorageDevice

SCSI Persistent Reservation capable device.

Note

Construction performs no I/O. Call setup() to wait for udev, validate PR capability, optionally register the key, and load initial reservation state.

Example

pr_device = PRDevice(path='/dev/sda', register_key='0x1234').setup()
pr_device.create_reservation(PR_TYPE_WRITE_EXCLUSIVE)
pr_device.release_reservation()

Show JSON schema:

{
  "additionalProperties": false,
  "description": "SCSI Persistent Reservation capable device.\n\nNote:\n    Construction performs no I/O. Call `setup()` to wait for udev,\n    validate PR capability, optionally register the key, and load\n    initial reservation state.\n\nExample:\n    ```python\n    pr_device = PRDevice(path='/dev/sda', register_key='0x1234').setup()\n    pr_device.create_reservation(PR_TYPE_WRITE_EXCLUSIVE)\n    pr_device.release_reservation()\n    ```",
  "properties": {
    "path": {
      "anyOf": [
        {
          "type": "string"
        },
        {
          "format": "path",
          "type": "string"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "title": "Path"
    },
    "name": {
      "anyOf": [
        {
          "type": "string"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "title": "Name"
    },
    "size": {
      "anyOf": [
        {
          "minimum": 0,
          "type": "integer"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "title": "Size"
    },
    "model": {
      "anyOf": [
        {
          "type": "string"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "title": "Model"
    },
    "register_key": {
      "anyOf": [
        {
          "type": "string"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "title": "Register Key"
    },
    "transport_id": {
      "anyOf": [
        {
          "type": "string"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "title": "Transport Id"
    },
    "auto_register": {
      "default": false,
      "title": "Auto Register",
      "type": "boolean"
    }
  },
  "title": "PRDevice",
  "type": "object"
}

Fields:

• path (PathOrStr | None)
• name (str | None)
• size (int | None)
• model (str | None)
• register_key (str | None)
• transport_id (str | None)
• auto_register (bool)

Validators:

• _derive_fields

Source code in sts_libs/src/sts/pr.py

class PRDevice(StorageDevice):
    """SCSI Persistent Reservation capable device.

    Note:
        Construction performs no I/O. Call `setup()` to wait for udev,
        validate PR capability, optionally register the key, and load
        initial reservation state.

    Example:
        ```python
        pr_device = PRDevice(path='/dev/sda', register_key='0x1234').setup()
        pr_device.create_reservation(PR_TYPE_WRITE_EXCLUSIVE)
        pr_device.release_reservation()
        ```
    """

    # PR-specific attributes
    register_key: str | None = None
    transport_id: str | None = None
    auto_register: bool = False

    # Internal state management
    _sg_persist: SgPersist = PrivateAttr(default_factory=SgPersist)
    _current_reservation_type: int | None = PrivateAttr(default=None)
    _registered_keys: list[str] = PrivateAttr(default_factory=list)
    _reservation_holder: str | None = PrivateAttr(default=None)
    _supports_pr: bool | None = PrivateAttr(default=None)

    @model_validator(mode='after')
    def _derive_fields(self) -> Self:
        # Field derivation only — no I/O. See setup() for system initialization.
        if self.path and not self.name:
            self.name = Path(self.path).name
        return self

    def setup(self) -> Self:
        """Wait for udev, validate PR support, optionally register key, and load state.

        Raises:
            DeviceError: If device is not fully ready or does not support
                persistent reservations.
        """
        # Wait for device to be fully available before PR operations
        if not self.wait_udev():
            raise DeviceError(f'Device {self.path} is not fully ready')

        # Validate PR capability
        if not self.supports_pr:
            raise DeviceError(f'Device {self.path} does not support persistent reservations')

        # Auto-register key if requested
        if self.auto_register and self.register_key:
            success = self.register_key_on_device(self.register_key, self.transport_id)
            if not success:
                logger.warning(f'Failed to auto-register key {self.register_key}')

        # Load initial state
        self._refresh_state()

        return self

    def _check_pr_support(self) -> bool:
        """Check if device supports persistent reservations."""
        if not self.path:
            return False
        try:
            # report_capabilities returns bool directly
            return self._sg_persist.report_capabilities(self.path)
        except (OSError, RuntimeError, ValueError) as e:
            logger.debug(f'PR support check failed: {e}')
            return False

    def _refresh_state(self) -> None:
        """Refresh internal state from device."""
        if not self.path:
            return

        # Get current keys - read_keys returns list[str] directly
        self._registered_keys = self._sg_persist.read_keys(self.path)

        # Get current reservation - read_reservation returns tuple
        holder_key, reservation_type = self._sg_persist.read_reservation(self.path)

        if reservation_type:
            try:
                self._current_reservation_type = int(reservation_type)
            except ValueError:
                self._current_reservation_type = None
        else:
            self._current_reservation_type = None

        self._reservation_holder = holder_key

    # Core PR Operations

    def register_key_on_device(self, key: str, transport_id: str | None = None) -> bool:
        """Register a PR key (required before creating reservations).

        Args:
            key: Registration key (e.g. ``'0x1234'``).
            transport_id: Transport-specific identifier for SAS (e.g. ``'sas,5001...'``).
        """
        if not self.path:
            logger.error('Device path not set')
            return False

        logger.info(f'Registering PR key {key} for {self.path}')

        # register returns bool directly
        success = self._sg_persist.register(self.path, key, transport_id)

        if success:
            self.register_key = key
            self.transport_id = transport_id
            self._refresh_state()
            logger.debug(f'Successfully registered key {key}')
        else:
            logger.error(f'Failed to register key {key}')

        return success

    def unregister_key_from_device(self, key: str | None = None) -> bool:
        """Unregister a key (defaults to current ``register_key``)."""
        if not self.path:
            logger.error('Device path not set')
            return False

        target_key = key or self.register_key
        if not target_key:
            logger.error('No key specified for unregistration')
            return False

        logger.info(f'Unregistering PR key {target_key} for {self.path}')

        # unregister returns bool directly
        success = self._sg_persist.unregister(self.path, target_key)

        if success:
            if target_key == self.register_key:
                self.register_key = None
                self.transport_id = None
            self._refresh_state()
            logger.debug(f'Successfully unregistered key {target_key}')
        else:
            logger.error(f'Failed to unregister key {target_key}')

        return success

    def create_reservation(self, pr_type: int = PR_TYPE_WRITE_EXCLUSIVE, key: str | None = None) -> bool:
        """Create a persistent reservation (initiator must be registered first).

        Args:
            pr_type: Reservation type constant (default: write exclusive).
            key: Reservation key (defaults to current ``register_key``).
        """
        if not self.path:
            logger.error('Device path not set')
            return False

        target_key = key or self.register_key
        if not target_key:
            logger.error('No key specified for reservation')
            return False

        if pr_type not in PR_TYPE_NAMES:
            logger.error(f'Unsupported PR type: {pr_type}')
            return False

        logger.info(f'Creating {get_pr_type_name(pr_type)} reservation with key {target_key}')

        # reserve returns bool directly
        success = self._sg_persist.reserve(self.path, target_key, pr_type)

        if success:
            self._current_reservation_type = pr_type
            self._reservation_holder = target_key
            logger.debug(f'Successfully created reservation type {pr_type}')
        else:
            logger.error('Failed to create reservation')

        return success

    def release_reservation(self, pr_type: int | None = None, key: str | None = None) -> bool:
        """Release a persistent reservation.

        Args:
            pr_type: Reservation type (defaults to current type).
            key: Reservation key (defaults to current ``register_key``).
        """
        if not self.path:
            logger.error('Device path not set')
            return False

        target_key = key or self.register_key
        if not target_key:
            logger.error('No key specified for release')
            return False

        target_type = pr_type or self._current_reservation_type
        if target_type is None:
            logger.error('No reservation type specified for release')
            return False

        logger.info(f'Releasing reservation type {target_type} with key {target_key}')

        # release returns bool directly
        success = self._sg_persist.release(self.path, target_key, target_type)

        if success:
            self._current_reservation_type = None
            self._reservation_holder = None
            self._refresh_state()
            logger.debug('Successfully released reservation')
        else:
            logger.error('Failed to release reservation')

        return success

    # Status and State Methods

    def get_reservation_status(self) -> tuple[str | None, str | None]:
        """Get ``(holder_key, reservation_type)`` or ``(None, None)``."""
        if not self.path:
            return (None, None)

        return self._sg_persist.read_reservation(self.path)

    def get_registered_keys(self) -> list[str]:
        """Get all registered reservation keys from the device."""
        if not self.path:
            return []

        return self._sg_persist.read_keys(self.path)

    def get_full_status(self) -> dict[str, Any]:
        """Get complete reservation status including all keys and transport IDs."""
        if not self.path:
            return {}

        return self._sg_persist.read_full_status(self.path)

    def is_reservation_holder(self) -> bool:
        """Check if this initiator's key holds the current reservation."""
        if not self.register_key:
            return False

        return self.reservation_holder == self.register_key

    def is_registered(self, key: str | None = None) -> bool:
        """Check if a key is registered (defaults to current ``register_key``)."""
        target_key = key or self.register_key
        if not target_key:
            return False

        return target_key in self.all_registered_keys

    def has_reservation(self) -> bool:
        """Check if device has any active reservation."""
        return self.reservation_holder is not None

    # ==========================================================================
    # Properties - Organized for clarity and convenience
    # ==========================================================================

    # Core Device State Properties (direct access to internal state)
    # ----------------------------------------------------------------

    @property
    def supports_pr(self) -> bool:
        """Whether the device supports PR operations (cached)."""
        if self._supports_pr is None:
            self._supports_pr = self._check_pr_support()
        return self._supports_pr

    @property
    def reservation_holder(self) -> str | None:
        """Key of the current reservation holder, or None."""
        if self._reservation_holder is None:
            self._refresh_state()
        return self._reservation_holder

    @property
    def reservation_type(self) -> int | None:
        """Current reservation type number, or None."""
        if self._current_reservation_type is None:
            self._refresh_state()
        return self._current_reservation_type

    @property
    def all_registered_keys(self) -> list[str]:
        """All registered keys (returns a safe copy)."""
        if not self._registered_keys:
            self._refresh_state()
        return self._registered_keys.copy()

    @property
    def transport_ids(self) -> list[dict[str, str]]:
        """Transport IDs of all registered initiators.

        Each dict contains ``'type'``, ``'id'``, and ``'key'``.
        """
        if not self.path:
            return []

        return self._sg_persist.get_transport_ids(self.path)

    # Derived State Properties (computed from core state)
    # ---------------------------------------------------

    @property
    def current_reservation_type_name(self) -> str | None:
        """Human-readable name of the current reservation type, or None."""
        if self.reservation_type is not None:
            return get_pr_type_name(self.reservation_type)
        return None

    @property
    def num_registered_keys(self) -> int:
        """Count of registered keys on the device."""
        return len(self.all_registered_keys)

    # Boolean Status Properties (convenient checks)
    # ---------------------------------------------

    @property
    def is_key_registered(self) -> bool:
        """Whether current ``register_key`` is registered on the device."""
        return self.is_registered()

    @property
    def can_create_reservation(self) -> bool:
        """True if key is registered, no reservation exists, and register_key is set."""
        return self.is_key_registered and not self.has_reservation() and self.register_key is not None

    # Summary Properties (aggregated information)
    # ------------------------------------------

    @property
    def pr_status_summary(self) -> dict[str, Any]:
        """Concise dict of current PR state (holder, type, key counts, etc.)."""
        return {
            'has_reservation': self.has_reservation(),
            'holder': self.reservation_holder,
            'type': self.reservation_type,
            'type_name': self.current_reservation_type_name,
            'num_keys': self.num_registered_keys,
            'is_holder': self.is_reservation_holder(),
            'is_registered': self.is_key_registered,
            'can_create': self.can_create_reservation,
        }

    @property
    def reservation_info(self) -> dict[str, Any]:
        """Comprehensive reservation info dict for debugging and monitoring."""
        return {
            'device_path': str(self.path) if self.path else None,
            'register_key': self.register_key,
            'transport_id': self.transport_ids,
            'reservation_holder': self.reservation_holder,
            'reservation_type': self.reservation_type,
            'reservation_type_name': self.current_reservation_type_name,
            'registered_keys': self.all_registered_keys,
            'num_registered_keys': self.num_registered_keys,
            'is_reservation_holder': self.is_reservation_holder(),
            'is_key_registered': self.is_key_registered,
            'has_reservation': self.has_reservation(),
            'supports_pr': self.supports_pr,
            'can_create_reservation': self.can_create_reservation,
        }

    # Utility Methods

    @classmethod
    def from_device(cls, device: StorageDevice, **kwargs: Any) -> Self:
        """Create a PRDevice from an existing StorageDevice, calling ``setup()`` automatically.

        Example:
            ```python
            pr_device = PRDevice.from_device(scsi_device, register_key='0x1234')
            ```
        """
        # Copy relevant attributes from source device
        pr_kwargs: dict[str, Any] = {
            'path': device.path,
            'name': device.name,
            'size': device.size,
            'model': device.model,
        }
        pr_kwargs.update(kwargs)

        return cls(**pr_kwargs).setup()

    def __str__(self) -> str:
        """Return string representation of PR device."""
        status_parts: list[str] = []

        if self.register_key:
            status_parts.append(f'key={self.register_key}')

        if self.reservation_type is not None:
            type_name = get_pr_type_name(self.reservation_type)
            status_parts.append(f'reservation={type_name}')

        status = f'({", ".join(status_parts)})' if status_parts else '(unregistered)'

        return f'PRDevice({self.path or "unknown"}) {status}'

all_registered_keys property

All registered keys (returns a safe copy).

can_create_reservation property

True if key is registered, no reservation exists, and register_key is set.

current_reservation_type_name property

Human-readable name of the current reservation type, or None.

is_key_registered property

Whether current register_key is registered on the device.

num_registered_keys property

Count of registered keys on the device.

pr_status_summary property

Concise dict of current PR state (holder, type, key counts, etc.).

reservation_holder property

Key of the current reservation holder, or None.

reservation_info property

Comprehensive reservation info dict for debugging and monitoring.

reservation_type property

Current reservation type number, or None.

supports_pr property

Whether the device supports PR operations (cached).

transport_ids property

Transport IDs of all registered initiators.

Each dict contains 'type', 'id', and 'key'.

__str__()

Return string representation of PR device.

Source code in sts_libs/src/sts/pr.py

def __str__(self) -> str:
    """Return string representation of PR device."""
    status_parts: list[str] = []

    if self.register_key:
        status_parts.append(f'key={self.register_key}')

    if self.reservation_type is not None:
        type_name = get_pr_type_name(self.reservation_type)
        status_parts.append(f'reservation={type_name}')

    status = f'({", ".join(status_parts)})' if status_parts else '(unregistered)'

    return f'PRDevice({self.path or "unknown"}) {status}'

create_reservation(pr_type=PR_TYPE_WRITE_EXCLUSIVE, key=None)

Create a persistent reservation (initiator must be registered first).

Parameters:

Name	Type	Description	Default
pr_type	int	Reservation type constant (default: write exclusive).	PR_TYPE_WRITE_EXCLUSIVE
key	str | None	Reservation key (defaults to current register_key).	None

Source code in sts_libs/src/sts/pr.py

def create_reservation(self, pr_type: int = PR_TYPE_WRITE_EXCLUSIVE, key: str | None = None) -> bool:
    """Create a persistent reservation (initiator must be registered first).

    Args:
        pr_type: Reservation type constant (default: write exclusive).
        key: Reservation key (defaults to current ``register_key``).
    """
    if not self.path:
        logger.error('Device path not set')
        return False

    target_key = key or self.register_key
    if not target_key:
        logger.error('No key specified for reservation')
        return False

    if pr_type not in PR_TYPE_NAMES:
        logger.error(f'Unsupported PR type: {pr_type}')
        return False

    logger.info(f'Creating {get_pr_type_name(pr_type)} reservation with key {target_key}')

    # reserve returns bool directly
    success = self._sg_persist.reserve(self.path, target_key, pr_type)

    if success:
        self._current_reservation_type = pr_type
        self._reservation_holder = target_key
        logger.debug(f'Successfully created reservation type {pr_type}')
    else:
        logger.error('Failed to create reservation')

    return success

from_device(device, **kwargs) classmethod

Create a PRDevice from an existing StorageDevice, calling setup() automatically.

Example

pr_device = PRDevice.from_device(scsi_device, register_key='0x1234')

Source code in sts_libs/src/sts/pr.py

@classmethod
def from_device(cls, device: StorageDevice, **kwargs: Any) -> Self:
    """Create a PRDevice from an existing StorageDevice, calling ``setup()`` automatically.

    Example:
        ```python
        pr_device = PRDevice.from_device(scsi_device, register_key='0x1234')
        ```
    """
    # Copy relevant attributes from source device
    pr_kwargs: dict[str, Any] = {
        'path': device.path,
        'name': device.name,
        'size': device.size,
        'model': device.model,
    }
    pr_kwargs.update(kwargs)

    return cls(**pr_kwargs).setup()

get_full_status()

Get complete reservation status including all keys and transport IDs.

Source code in sts_libs/src/sts/pr.py

def get_full_status(self) -> dict[str, Any]:
    """Get complete reservation status including all keys and transport IDs."""
    if not self.path:
        return {}

    return self._sg_persist.read_full_status(self.path)

get_registered_keys()

Get all registered reservation keys from the device.

Source code in sts_libs/src/sts/pr.py

def get_registered_keys(self) -> list[str]:
    """Get all registered reservation keys from the device."""
    if not self.path:
        return []

    return self._sg_persist.read_keys(self.path)

get_reservation_status()

Get (holder_key, reservation_type) or (None, None).

Source code in sts_libs/src/sts/pr.py

def get_reservation_status(self) -> tuple[str | None, str | None]:
    """Get ``(holder_key, reservation_type)`` or ``(None, None)``."""
    if not self.path:
        return (None, None)

    return self._sg_persist.read_reservation(self.path)

has_reservation()

Check if device has any active reservation.

Source code in sts_libs/src/sts/pr.py

def has_reservation(self) -> bool:
    """Check if device has any active reservation."""
    return self.reservation_holder is not None

is_registered(key=None)

Check if a key is registered (defaults to current register_key).

Source code in sts_libs/src/sts/pr.py

def is_registered(self, key: str | None = None) -> bool:
    """Check if a key is registered (defaults to current ``register_key``)."""
    target_key = key or self.register_key
    if not target_key:
        return False

    return target_key in self.all_registered_keys

is_reservation_holder()

Check if this initiator's key holds the current reservation.

Source code in sts_libs/src/sts/pr.py

def is_reservation_holder(self) -> bool:
    """Check if this initiator's key holds the current reservation."""
    if not self.register_key:
        return False

    return self.reservation_holder == self.register_key

register_key_on_device(key, transport_id=None)

Register a PR key (required before creating reservations).

Parameters:

Name	Type	Description	Default
key	str	Registration key (e.g. '0x1234').	required
transport_id	str | None	Transport-specific identifier for SAS (e.g. 'sas,5001...').	None

Source code in sts_libs/src/sts/pr.py

def register_key_on_device(self, key: str, transport_id: str | None = None) -> bool:
    """Register a PR key (required before creating reservations).

    Args:
        key: Registration key (e.g. ``'0x1234'``).
        transport_id: Transport-specific identifier for SAS (e.g. ``'sas,5001...'``).
    """
    if not self.path:
        logger.error('Device path not set')
        return False

    logger.info(f'Registering PR key {key} for {self.path}')

    # register returns bool directly
    success = self._sg_persist.register(self.path, key, transport_id)

    if success:
        self.register_key = key
        self.transport_id = transport_id
        self._refresh_state()
        logger.debug(f'Successfully registered key {key}')
    else:
        logger.error(f'Failed to register key {key}')

    return success

release_reservation(pr_type=None, key=None)

Release a persistent reservation.

Parameters:

Name	Type	Description	Default
pr_type	int | None	Reservation type (defaults to current type).	None
key	str | None	Reservation key (defaults to current register_key).	None

Source code in sts_libs/src/sts/pr.py

def release_reservation(self, pr_type: int | None = None, key: str | None = None) -> bool:
    """Release a persistent reservation.

    Args:
        pr_type: Reservation type (defaults to current type).
        key: Reservation key (defaults to current ``register_key``).
    """
    if not self.path:
        logger.error('Device path not set')
        return False

    target_key = key or self.register_key
    if not target_key:
        logger.error('No key specified for release')
        return False

    target_type = pr_type or self._current_reservation_type
    if target_type is None:
        logger.error('No reservation type specified for release')
        return False

    logger.info(f'Releasing reservation type {target_type} with key {target_key}')

    # release returns bool directly
    success = self._sg_persist.release(self.path, target_key, target_type)

    if success:
        self._current_reservation_type = None
        self._reservation_holder = None
        self._refresh_state()
        logger.debug('Successfully released reservation')
    else:
        logger.error('Failed to release reservation')

    return success

setup()

Wait for udev, validate PR support, optionally register key, and load state.

Raises:

Type	Description
DeviceError	If device is not fully ready or does not support persistent reservations.

Source code in sts_libs/src/sts/pr.py

def setup(self) -> Self:
    """Wait for udev, validate PR support, optionally register key, and load state.

    Raises:
        DeviceError: If device is not fully ready or does not support
            persistent reservations.
    """
    # Wait for device to be fully available before PR operations
    if not self.wait_udev():
        raise DeviceError(f'Device {self.path} is not fully ready')

    # Validate PR capability
    if not self.supports_pr:
        raise DeviceError(f'Device {self.path} does not support persistent reservations')

    # Auto-register key if requested
    if self.auto_register and self.register_key:
        success = self.register_key_on_device(self.register_key, self.transport_id)
        if not success:
            logger.warning(f'Failed to auto-register key {self.register_key}')

    # Load initial state
    self._refresh_state()

    return self

unregister_key_from_device(key=None)

Unregister a key (defaults to current register_key).

Source code in sts_libs/src/sts/pr.py

def unregister_key_from_device(self, key: str | None = None) -> bool:
    """Unregister a key (defaults to current ``register_key``)."""
    if not self.path:
        logger.error('Device path not set')
        return False

    target_key = key or self.register_key
    if not target_key:
        logger.error('No key specified for unregistration')
        return False

    logger.info(f'Unregistering PR key {target_key} for {self.path}')

    # unregister returns bool directly
    success = self._sg_persist.unregister(self.path, target_key)

    if success:
        if target_key == self.register_key:
            self.register_key = None
            self.transport_id = None
        self._refresh_state()
        logger.debug(f'Successfully unregistered key {target_key}')
    else:
        logger.error(f'Failed to unregister key {target_key}')

    return success