Patch for 1557 (#1645)

* Attempting a retry-attempt on the broken part of lsblk

* Improved logging

* Adding a retry to Partition._call_lsblk()

* Added error checks if lsblk returns nothing, also handles empty Partition().info instance.

* Added missing check of disk encryption is None or not.

* Added tweak to catching output from lsblk.stderr

* Added missing check of disk encryption is None or not.

* Fixed a logic test for empty lsblk info

* Fixed instances of None being interated

* Added some errro handling for weird block devices

* Fixed flake8

* Added /etc/vconsole.conf generation in Installer.mkinitcpio() as it's a dependency for it to generate properly without errors. Otherwise we'll get ==> ERRROR: file not found: '/etc/vconsole.conf'

* Prep for tagging RC1 of 2.5.3

* Corrected helpers.py get_blockdevice_info() to deal with empty lsblk results

3 files changed, 128 insertions, 87 deletions

diff --git a/archinstall/lib/disk/filesystem.py b/archinstall/lib/disk/filesystem.py
index bdfa502a..1e722ce5 100644
--- a/archinstall/lib/disk/filesystem.py
+++ b/archinstall/lib/disk/filesystem.py
@@ -113,7 +113,7 @@ class Filesystem:
 				format_options = partition.get('options',[]) + partition.get('filesystem',{}).get('format_options',[])
 				disk_encryption: DiskEncryption = storage['arguments'].get('disk_encryption')
 
-				if partition in disk_encryption.partitions:
+				if disk_encryption and partition in disk_encryption.partitions:
 					if not partition['device_instance']:
 						raise DiskError(f"Internal error caused us to loose the partition. Please report this issue upstream!")
 
diff --git a/archinstall/lib/disk/helpers.py b/archinstall/lib/disk/helpers.py
index a5164b76..aea794f5 100644
--- a/archinstall/lib/disk/helpers.py
+++ b/archinstall/lib/disk/helpers.py
@@ -212,6 +212,47 @@ def all_disks() -> List[BlockDevice]:
 	log(f"[Deprecated] archinstall.all_disks() is deprecated. Use archinstall.all_blockdevices() with the appropriate filters instead.", level=logging.WARNING, fg="yellow")
 	return all_blockdevices(partitions=False, mappers=False)
 
+def get_blockdevice_info(device_path, exclude_iso_dev :bool = True) -> Dict[str, Any]:
+	for retry_attempt in range(storage['DISK_RETRY_ATTEMPTS']):
+		partprobe(device_path)
+		time.sleep(max(0.1, storage['DISK_TIMEOUTS'] * retry_attempt))
+		
+		try:
+			if exclude_iso_dev:
+				# exclude all devices associated with the iso boot locations
+				iso_devs = ['/run/archiso/airootfs', '/run/archiso/bootmnt']
+
+				try:
+					lsblk_info = get_lsblk_info(device_path)
+				except DiskError:
+					continue
+
+				if any([dev in lsblk_info.mountpoints for dev in iso_devs]):
+					continue
+
+			information = blkid(f'blkid -p -o export {device_path}')
+			return enrich_blockdevice_information(information)
+		except SysCallError as ex:
+			if ex.exit_code in (512, 2):
+				# Assume that it's a loop device, and try to get info on it
+				try:
+					resolved_device_name = device_path.readlink().name
+				except OSError:
+					resolved_device_name = device_path.name
+
+				try:
+					information = get_loop_info(device_path)
+					if not information:
+						raise SysCallError(f"Could not get loop information for {resolved_device_name}", exit_code=1)
+					return enrich_blockdevice_information(information)
+
+				except SysCallError:
+					information = get_blockdevice_uevent(resolved_device_name)
+					return enrich_blockdevice_information(information)
+			else:
+				# We could not reliably get any information, perhaps the disk is clean of information?
+				if retry_attempt == storage['DISK_RETRY_ATTEMPTS'] - 1:
+					raise ex
 
 def all_blockdevices(
 	mappers: bool = False,
@@ -230,40 +271,18 @@ def all_blockdevices(
 	# we'll iterate the /sys/class definitions and find the information
 	# from there.
 	for block_device in glob.glob("/sys/class/block/*"):
-		device_path = pathlib.Path(f"/dev/{pathlib.Path(block_device).readlink().name}")
+		try:
+			device_path = pathlib.Path(f"/dev/{pathlib.Path(block_device).readlink().name}")
+		except FileNotFoundError:
+			log(f"Unknown device found by '/sys/class/block/*', ignoring: {device_path}", level=logging.WARNING, fg="yellow")
 
 		if device_path.exists() is False:
 			log(f"Unknown device found by '/sys/class/block/*', ignoring: {device_path}", level=logging.WARNING, fg="yellow")
 			continue
 
-		try:
-			if exclude_iso_dev:
-				# exclude all devices associated with the iso boot locations
-				iso_devs = ['/run/archiso/airootfs', '/run/archiso/bootmnt']
-				lsblk_info = get_lsblk_info(device_path)
-				if any([dev in lsblk_info.mountpoints for dev in iso_devs]):
-					continue
-
-			information = blkid(f'blkid -p -o export {device_path}')
-		except SysCallError as ex:
-			if ex.exit_code in (512, 2):
-				# Assume that it's a loop device, and try to get info on it
-				try:
-					information = get_loop_info(device_path)
-					if not information:
-						print("Exit code for blkid -p -o export was:", ex.exit_code)
-						raise SysCallError("Could not get loop information", exit_code=1)
-
-				except SysCallError:
-					print("Not a loop device, trying uevent rules.")
-					information = get_blockdevice_uevent(pathlib.Path(block_device).readlink().name)
-			else:
-				# We could not reliably get any information, perhaps the disk is clean of information?
-				print("Raising ex because:", ex.exit_code)
-				raise ex
-				# return instances
-
-		information = enrich_blockdevice_information(information)
+		information = get_blockdevice_info(device_path)
+		if not information:
+			continue
 
 		for path, path_info in information.items():
 			if path_info.get('DMCRYPT_NAME'):
@@ -409,7 +428,6 @@ def get_partitions_in_use(mountpoint :str) -> Dict[str, Any]:
 		return {}
 
 	output = json.loads(output)
-	# print(output)
 
 	mounts = {}
 
@@ -421,11 +439,13 @@ def get_partitions_in_use(mountpoint :str) -> Dict[str, Any]:
 			continue
 
 		if isinstance(blockdev, Partition):
-			for blockdev_mountpoint in blockdev.mountpoints:
-				block_devices_mountpoints[blockdev_mountpoint] = blockdev
+			if blockdev.mountpoints:
+				for blockdev_mountpoint in blockdev.mountpoints:
+					block_devices_mountpoints[blockdev_mountpoint] = blockdev
 		else:
-			for blockdev_mountpoint in blockdev.mount_information:
-				block_devices_mountpoints[blockdev_mountpoint['target']] = blockdev
+			if blockdev.mount_information:
+				for blockdev_mountpoint in blockdev.mount_information:
+					block_devices_mountpoints[blockdev_mountpoint['target']] = blockdev
 
 	log(f'Filtering available mounts {block_devices_mountpoints} to those under {mountpoint}', level=logging.DEBUG)
 
diff --git a/archinstall/lib/disk/partition.py b/archinstall/lib/disk/partition.py
index 9febf102..12b1a9a6 100644
--- a/archinstall/lib/disk/partition.py
+++ b/archinstall/lib/disk/partition.py
@@ -98,17 +98,18 @@ class Partition:
 		if mountpoint:
 			self.mount(mountpoint)
 
-		self._partition_info = self._fetch_information()
-
-		if not autodetect_filesystem and filesystem:
-			self._partition_info.filesystem_type = filesystem
+		try:
+			self._partition_info = self._fetch_information()
+	
+			if not autodetect_filesystem and filesystem:
+				self._partition_info.filesystem_type = filesystem
 
-		if self._partition_info.filesystem_type == 'crypto_LUKS':
-			self._encrypted = True
+			if self._partition_info.filesystem_type == 'crypto_LUKS':
+				self._encrypted = True
+		except DiskError:
+			self._partition_info = None
 
-	# I hate doint this but I'm currently unsure where this
-	# is acutally used to be able to fix the typing issues properly
-	@typing.no_type_check
+	@typing.no_type_check # I hate doint this but I'm currently unsure where this is used.
 	def __lt__(self, left_comparitor :BlockDevice) -> bool:
 		if type(left_comparitor) == Partition:
 			left_comparitor = left_comparitor.path
@@ -120,14 +121,17 @@ class Partition:
 
 	def __repr__(self, *args :str, **kwargs :str) -> str:
 		mount_repr = ''
-		if mountpoint := self._partition_info.get_first_mountpoint():
-			mount_repr = f", mounted={mountpoint}"
-		elif self._target_mountpoint:
-			mount_repr = f", rel_mountpoint={self._target_mountpoint}"
+		if self._partition_info:
+			if mountpoint := self._partition_info.get_first_mountpoint():
+				mount_repr = f", mounted={mountpoint}"
+			elif self._target_mountpoint:
+				mount_repr = f", rel_mountpoint={self._target_mountpoint}"
 
 		classname = self.__class__.__name__
 
-		if self._encrypted:
+		if not self._partition_info:
+			return f'{classname}(path={self._path})'
+		elif self._encrypted:
 			return f'{classname}(path={self._path}, size={self.size}, PARTUUID={self.part_uuid}, parent={self.real_device}, fs={self._partition_info.filesystem_type}{mount_repr})'
 		else:
 			return f'{classname}(path={self._path}, size={self.size}, PARTUUID={self.part_uuid}, fs={self._partition_info.filesystem_type}{mount_repr})'
@@ -146,7 +150,7 @@ class Partition:
 			'encrypted': self._encrypted,
 			'start': self.start,
 			'size': self.end,
-			'filesystem': self._partition_info.filesystem_type
+			'filesystem': self._partition_info.filesystem_type if self._partition_info else 'Unknown'
 		}
 
 		return partition_info
@@ -164,34 +168,38 @@ class Partition:
 			'start': self.start,
 			'size': self.end,
 			'filesystem': {
-				'format': self._partition_info.filesystem_type
+				'format': self._partition_info.filesystem_type if self._partition_info else 'None'
 			}
 		}
 
 	def _call_lsblk(self) -> Dict[str, Any]:
-		self.partprobe()
-		# This sleep might be overkill, but lsblk is known to
-		# work against a chaotic cache that can change during call
-		# causing no information to be returned (blkid is better)
-		# time.sleep(1)
+		for retry_attempt in range(storage['DISK_RETRY_ATTEMPTS']):
+			self.partprobe()
+			time.sleep(max(0.1, storage['DISK_TIMEOUTS'] * retry_attempt)) # TODO: Remove, we should be relying on blkid instead of lsblk
+			# This sleep might be overkill, but lsblk is known to
+			# work against a chaotic cache that can change during call
+			# causing no information to be returned (blkid is better)
+			# time.sleep(1)
 
-		# TODO: Maybe incorporate a re-try system here based on time.sleep(max(0.1, storage.get('DISK_TIMEOUTS', 1)))
+			# TODO: Maybe incorporate a re-try system here based on time.sleep(max(0.1, storage.get('DISK_TIMEOUTS', 1)))
 
-		try:
-			output = SysCommand(f"lsblk --json -b -o+LOG-SEC,SIZE,PTTYPE,PARTUUID,UUID,FSTYPE {self.device_path}").decode('UTF-8')
-		except SysCallError as error:
-			# It appears as if lsblk can return exit codes like 8192 to indicate something.
-			# But it does return output so we'll try to catch it.
-			output = error.worker.decode('UTF-8')
-
-		if output:
 			try:
-				lsblk_info = json.loads(output)
-				return lsblk_info
-			except json.decoder.JSONDecodeError:
-				log(f"Could not decode JSON: {output}", fg="red", level=logging.ERROR)
-
-		raise DiskError(f'Failed to read disk "{self.device_path}" with lsblk')
+				output = SysCommand(f"lsblk --json -b -o+LOG-SEC,SIZE,PTTYPE,PARTUUID,UUID,FSTYPE {self.device_path}").decode('UTF-8')
+			except SysCallError as error:
+				# It appears as if lsblk can return exit codes like 8192 to indicate something.
+				# But it does return output in stderr so we'll try to catch it minus the message/info.
+				output = error.worker.decode('UTF-8')
+				if '{' in output:
+					output = output[output.find('{'):]
+
+			if output:
+				try:
+					lsblk_info = json.loads(output)
+					return lsblk_info
+				except json.decoder.JSONDecodeError:
+					log(f"Could not decode JSON: {output}", fg="red", level=logging.ERROR)
+		
+		raise DiskError(f'Failed to get partition information "{self.device_path}" with lsblk')
 
 	def _call_sfdisk(self) -> Dict[str, Any]:
 		output = SysCommand(f"sfdisk --json {self.block_device.path}").decode('UTF-8')
@@ -212,9 +220,12 @@ class Partition:
 		lsblk_info = self._call_lsblk()
 		sfdisk_info = self._call_sfdisk()
 
-		if not (device := lsblk_info.get('blockdevices', [None])[0]):
+		if not (device := lsblk_info.get('blockdevices', [])):
 			raise DiskError(f'Failed to retrieve information for "{self.device_path}" with lsblk')
 
+		# Grab the first (and only) block device in the list as we're targeting a specific partition
+		device = device[0]
+
 		mountpoints = [Path(mountpoint) for mountpoint in device['mountpoints'] if mountpoint]
 		bootable = sfdisk_info.get('bootable', False) or sfdisk_info.get('type', '') == 'C12A7328-F81F-11D2-BA4B-00A0C93EC93B'
 
@@ -243,7 +254,8 @@ class Partition:
 
 	@property
 	def filesystem(self) -> str:
-		return self._partition_info.filesystem_type
+		if self._partition_info:
+			return self._partition_info.filesystem_type
 
 	@property
 	def mountpoint(self) -> Optional[Path]:
@@ -253,43 +265,51 @@ class Partition:
 
 	@property
 	def mountpoints(self) -> List[Path]:
-		return self._partition_info.mountpoints
+		if self._partition_info:
+			return self._partition_info.mountpoints
 
 	@property
 	def sector_size(self) -> int:
-		return self._partition_info.sector_size
+		if self._partition_info:
+			return self._partition_info.sector_size
 
 	@property
 	def start(self) -> Optional[int]:
-		return self._partition_info.start
+		if self._partition_info:
+			return self._partition_info.start
 
 	@property
 	def end(self) -> Optional[int]:
-		return self._partition_info.end
+		if self._partition_info:
+			return self._partition_info.end
 
 	@property
 	def end_sectors(self) -> Optional[int]:
-		start = self._partition_info.start
-		end = self._partition_info.end
-		if start and end:
-			return start + end
-		return None
+		if self._partition_info:
+			start = self._partition_info.start
+			end = self._partition_info.end
+			if start and end:
+				return start + end
 
 	@property
 	def size(self) -> Optional[float]:
-		return self._partition_info.size
+		if self._partition_info:
+			return self._partition_info.size
 
 	@property
 	def boot(self) -> bool:
-		return self._partition_info.bootable
+		if self._partition_info:
+			return self._partition_info.bootable
 
 	@property
 	def partition_type(self) -> Optional[str]:
-		return self._partition_info.pttype
+		if self._partition_info:
+			return self._partition_info.pttype
 
 	@property
 	def part_uuid(self) -> str:
-		return self._partition_info.partuuid
+		if self._partition_info:
+			return self._partition_info.partuuid
 
 	@property
 	def uuid(self) -> Optional[str]:
@@ -355,7 +375,8 @@ class Partition:
 
 			log(f"Could not get PARTUUID of partition using 'blkid -s PARTUUID -o value {self.device_path}': {error}")
 
-		return self._partition_info.uuid
+		if self._partition_info:
+			return self._partition_info.uuid
 
 	@property
 	def encrypted(self) -> Union[bool, None]: