src/storage/blobfs/blob-loader.cc - fuchsia - Git at Google

 // Copyright 2020 The Fuchsia Authors. All rights reserved.
 // Use of this source code is governed by a BSD-style license that can be
 // found in the LICENSE file.

 #include "blob-loader.h"

 #include <lib/fzl/owned-vmo-mapper.h>
 #include <lib/zx/status.h>
 #include <zircon/assert.h>
 #include <zircon/errors.h>
 #include <zircon/status.h>

 #include <algorithm>
 #include <memory>

 #include <blobfs/common.h>
 #include <blobfs/compression-settings.h>
 #include <blobfs/format.h>
 #include <fbl/auto_call.h>
 #include <fbl/string_buffer.h>
 #include <storage/buffer/owned_vmoid.h>

 #include "blob-verifier.h"
 #include "compression/chunked.h"
 #include "compression/decompressor.h"
 #include "compression/seekable-decompressor.h"
 #include "compression/zstd-seekable-blob-collection.h"
 #include "iterator/block-iterator.h"

 namespace blobfs {

 namespace {

 // TODO(jfsulliv): Rationalize this with the size limits for chunk-compression headers.
 constexpr size_t kScratchBufferSize = 4 * kBlobfsBlockSize;

 }  // namespace

 BlobLoader::BlobLoader(TransactionManager* txn_manager, BlockIteratorProvider* block_iter_provider,
                        NodeFinder* node_finder, pager::UserPager* pager, BlobfsMetrics* metrics,
                        ZSTDSeekableBlobCollection* zstd_seekable_blob_collection,
                        fzl::OwnedVmoMapper scratch_vmo)
     : txn_manager_(txn_manager),
       block_iter_provider_(block_iter_provider),
       node_finder_(node_finder),
       pager_(pager),
       metrics_(metrics),
       zstd_seekable_blob_collection_(zstd_seekable_blob_collection),
       scratch_vmo_(std::move(scratch_vmo)) {}

 zx::status<BlobLoader> BlobLoader::Create(
     TransactionManager* txn_manager, BlockIteratorProvider* block_iter_provider,
     NodeFinder* node_finder, pager::UserPager* pager, BlobfsMetrics* metrics,
     ZSTDSeekableBlobCollection* zstd_seekable_blob_collection) {
   fzl::OwnedVmoMapper scratch_vmo;
   zx_status_t status = scratch_vmo.CreateAndMap(kScratchBufferSize, "blobfs-loader");
   if (status != ZX_OK) {
     FS_TRACE_ERROR("blobfs: Failed to map scratch vmo: %s\n", zx_status_get_string(status));
     return zx::error(status);
   }
   return zx::ok(BlobLoader(txn_manager, block_iter_provider, node_finder, pager, metrics,
                            zstd_seekable_blob_collection, std::move(scratch_vmo)));
 }

 void BlobLoader::Reset() { scratch_vmo_.Reset(); }

 zx_status_t BlobLoader::LoadBlob(uint32_t node_index,
                                  const BlobCorruptionNotifier* corruption_notifier,
                                  fzl::OwnedVmoMapper* data_out, fzl::OwnedVmoMapper* merkle_out) {
   ZX_DEBUG_ASSERT(scratch_vmo_.vmo().is_valid());
   const InodePtr inode = node_finder_->GetNode(node_index);
   // LoadBlob should only be called for Inodes. If this doesn't hold, one of two things happened:
   //   - Programmer error
   //   - Corruption of a blob's Inode
   // In either case it is preferable to ASSERT than to return an error here, since the first case
   // should happen only during development and in the second case there may be more corruption and
   // we want to unmount the filesystem before any more damage is done.
   ZX_ASSERT(inode->header.IsInode() && inode->header.IsAllocated());

   TRACE_DURATION("blobfs", "BlobLoader::LoadBlob", "blob_size", inode->blob_size);

   const uint64_t num_data_blocks = BlobDataBlocks(*inode);
   if (num_data_blocks == 0) {
     // No data to load for the null blob.
     return ZX_OK;
   }

   fzl::OwnedVmoMapper merkle_mapper;
   std::unique_ptr<BlobVerifier> verifier;
   zx_status_t status;
   if ((status = InitMerkleVerifier(node_index, *inode, corruption_notifier, &merkle_mapper,
                                    &verifier)) != ZX_OK) {
     return status;
   }

   size_t data_vmo_size;
   if (mul_overflow(num_data_blocks, kBlobfsBlockSize, &data_vmo_size)) {
     FS_TRACE_ERROR("Multiplication overflow");
     return ZX_ERR_OUT_OF_RANGE;
   }

   fbl::StringBuffer<ZX_MAX_NAME_LEN> data_vmo_name;
   FormatBlobDataVmoName(node_index, &data_vmo_name);

   fzl::OwnedVmoMapper data_mapper;
   status = data_mapper.CreateAndMap(data_vmo_size, data_vmo_name.c_str());
   if (status != ZX_OK) {
     FS_TRACE_ERROR("blobfs: Failed to initialize data vmo; error: %s\n",
                    zx_status_get_string(status));
     return status;
   }
   status = inode->IsCompressed() ? LoadAndDecompressData(node_index, *inode, data_mapper)
                                  : LoadData(node_index, *inode, data_mapper);
   if (status != ZX_OK) {
     return status;
   }

   if ((status = verifier->Verify(data_mapper.start(), inode->blob_size, data_vmo_size)) != ZX_OK) {
     return status;
   }

   *data_out = std::move(data_mapper);
   if (merkle_mapper.vmo().is_valid()) {
     *merkle_out = std::move(merkle_mapper);
   }
   return ZX_OK;
 }

 zx_status_t BlobLoader::LoadBlobPaged(uint32_t node_index,
                                       const BlobCorruptionNotifier* corruption_notifier,
                                       std::unique_ptr<pager::PageWatcher>* page_watcher_out,
                                       fzl::OwnedVmoMapper* data_out,
                                       fzl::OwnedVmoMapper* merkle_out) {
   ZX_DEBUG_ASSERT(scratch_vmo_.vmo().is_valid());
   const InodePtr inode = node_finder_->GetNode(node_index);
   // LoadBlobPaged should only be called for Inodes. If this doesn't hold, one of two things
   // happened:
   //   - Programmer error
   //   - Corruption of a blob's Inode
   // In either case it is preferable to ASSERT than to return an error here, since the first case
   // should happen only during development and in the second case there may be more corruption and
   // we want to unmount the filesystem before any more damage is done.
   ZX_ASSERT(inode->header.IsInode() && inode->header.IsAllocated());

   TRACE_DURATION("blobfs", "BlobLoader::LoadBlobPaged", "blob_size", inode->blob_size);

   const uint64_t num_data_blocks = BlobDataBlocks(*inode);
   if (num_data_blocks == 0) {
     // No data to load for the null blob.
     return ZX_OK;
   }

   fzl::OwnedVmoMapper merkle_mapper;
   std::unique_ptr<BlobVerifier> verifier;
   zx_status_t status;
   if ((status = InitMerkleVerifier(node_index, *inode, corruption_notifier, &merkle_mapper,
                                    &verifier)) != ZX_OK) {
     return status;
   }

   std::unique_ptr<SeekableDecompressor> decompressor;
   ZSTDSeekableBlobCollection* zstd_seekable_blob_collection = nullptr;
   if ((status = InitForDecompression(node_index, *inode, *verifier, &decompressor,
                                      &zstd_seekable_blob_collection)) != ZX_OK) {
     return status;
   }

   pager::UserPagerInfo userpager_info;
   userpager_info.identifier = node_index;
   userpager_info.data_start_bytes = ComputeNumMerkleTreeBlocks(*inode) * kBlobfsBlockSize;
   userpager_info.data_length_bytes = inode->blob_size;
   userpager_info.verifier = std::move(verifier);
   userpager_info.decompressor = std::move(decompressor);
   userpager_info.zstd_seekable_blob_collection = zstd_seekable_blob_collection;
   auto page_watcher = std::make_unique<pager::PageWatcher>(pager_, std::move(userpager_info));

   fbl::StringBuffer<ZX_MAX_NAME_LEN> data_vmo_name;
   FormatBlobDataVmoName(node_index, &data_vmo_name);

   zx::vmo data_vmo;
   size_t data_vmo_size;
   if (mul_overflow(num_data_blocks, kBlobfsBlockSize, &data_vmo_size)) {
     FS_TRACE_ERROR("Multiplication overflow");
     return ZX_ERR_OUT_OF_RANGE;
   }
   if ((status = page_watcher->CreatePagedVmo(data_vmo_size, &data_vmo)) != ZX_OK) {
     return status;
   }
   data_vmo.set_property(ZX_PROP_NAME, data_vmo_name.c_str(), data_vmo_name.length());

   fzl::OwnedVmoMapper data_mapper;
   if ((status = data_mapper.Map(std::move(data_vmo))) != ZX_OK) {
     FS_TRACE_ERROR("blobfs: Failed to create mapping for data vmo: %s\n",
                    zx_status_get_string(status));
     return status;
   }

   *page_watcher_out = std::move(page_watcher);
   *data_out = std::move(data_mapper);
   if (merkle_mapper.vmo().is_valid()) {
     *merkle_out = std::move(merkle_mapper);
   }
   return ZX_OK;
 }

 zx_status_t BlobLoader::InitMerkleVerifier(uint32_t node_index, const Inode& inode,
                                            const BlobCorruptionNotifier* notifier,
                                            fzl::OwnedVmoMapper* vmo_out,
                                            std::unique_ptr<BlobVerifier>* verifier_out) {
   uint64_t num_merkle_blocks = ComputeNumMerkleTreeBlocks(inode);
   if (num_merkle_blocks == 0) {
     return BlobVerifier::CreateWithoutTree(digest::Digest(inode.merkle_root_hash), metrics_,
                                            inode.blob_size, notifier, verifier_out);
   }

   fzl::OwnedVmoMapper merkle_mapper;
   std::unique_ptr<BlobVerifier> verifier;

   size_t merkle_vmo_size;
   if (mul_overflow(num_merkle_blocks, kBlobfsBlockSize, &merkle_vmo_size)) {
     FS_TRACE_ERROR("Multiplication overflow");
     return ZX_ERR_OUT_OF_RANGE;
   }

   fbl::StringBuffer<ZX_MAX_NAME_LEN> merkle_vmo_name;
   FormatBlobMerkleVmoName(node_index, &merkle_vmo_name);

   zx_status_t status;
   if ((status = merkle_mapper.CreateAndMap(merkle_vmo_size, merkle_vmo_name.c_str())) != ZX_OK) {
     FS_TRACE_ERROR("blobfs: Failed to initialize merkle vmo; error: %s\n",
                    zx_status_get_string(status));
     return status;
   }

   if ((status = LoadMerkle(node_index, inode, merkle_mapper)) != ZX_OK) {
     return status;
   }

   if ((status = BlobVerifier::Create(digest::Digest(inode.merkle_root_hash), metrics_,
                                      merkle_mapper.start(), merkle_vmo_size, inode.blob_size,
                                      notifier, &verifier)) != ZX_OK) {
     return status;
   }

   *vmo_out = std::move(merkle_mapper);
   *verifier_out = std::move(verifier);
   return ZX_OK;
 }

 zx_status_t BlobLoader::InitForDecompression(
     uint32_t node_index, const Inode& inode, const BlobVerifier& verifier,
     std::unique_ptr<SeekableDecompressor>* decompressor_out,
     ZSTDSeekableBlobCollection** zstd_seekable_blob_collection_out) {
   zx::status<CompressionAlgorithm> algorithm_status = AlgorithmForInode(inode);
   if (algorithm_status.is_error()) {
     FS_TRACE_ERROR("blobfs: Cannot decode blob due to multiple compression flags.\n");
     return algorithm_status.status_value();
   }
   CompressionAlgorithm algorithm = algorithm_status.value();

   switch (algorithm) {
     case CompressionAlgorithm::UNCOMPRESSED:
       return ZX_OK;
     case CompressionAlgorithm::ZSTD_SEEKABLE:
       *zstd_seekable_blob_collection_out = zstd_seekable_blob_collection_;
       break;
     case CompressionAlgorithm::CHUNKED:
       break;
     case CompressionAlgorithm::LZ4:
     case CompressionAlgorithm::ZSTD:
       // Callers should have guarded against calling this code path with an algorithm that
       // does not support paging.
       FS_TRACE_ERROR(
           "Algorithm %s does not support paging; this path should not be called.\n"
           "This is most likely programmer error.\n",
           CompressionAlgorithmToString(algorithm));
       ZX_DEBUG_ASSERT(false);
       return ZX_ERR_NOT_SUPPORTED;
   }

   TRACE_DURATION("blobfs", "BlobLoader::InitDecompressor");

   // The first few blocks of data contain the seek table, which we need to read to initialize
   // the decompressor. Read these from disk.

   zx_status_t status;

   uint32_t merkle_blocks = ComputeNumMerkleTreeBlocks(inode);
   // We don't know exactly how long the header is, so fill up as much of the scratch VMO as we can.
   // (The header should never be bigger than the size of the scratch VMO.)
   ZX_DEBUG_ASSERT(scratch_vmo_.size() % kBlobfsBlockSize == 0);
   uint32_t num_data_blocks = static_cast<uint32_t>(scratch_vmo_.size()) / kBlobfsBlockSize;
   num_data_blocks = std::min(num_data_blocks, inode.block_count - merkle_blocks);
   if (num_data_blocks == 0) {
     FS_TRACE_ERROR("blobfs: No data blocks; corrupted inode?\n");
     return ZX_ERR_BAD_STATE;
   }

   auto bytes_read = LoadBlocks(node_index, merkle_blocks, num_data_blocks, scratch_vmo_);
   if (bytes_read.is_error()) {
     FS_TRACE_ERROR("blobfs: Failed to load compression header: %s\n", bytes_read.status_string());
     return bytes_read.error_value();
   }

   if ((status = SeekableChunkedDecompressor::CreateDecompressor(
            scratch_vmo_.start(), num_data_blocks * kBlobfsBlockSize, inode.blob_size,
            decompressor_out)) != ZX_OK) {
     FS_TRACE_ERROR("blobfs: Failed to init decompressor: %s\n", zx_status_get_string(status));
     return status;
   }

   return ZX_OK;
 }

 zx_status_t BlobLoader::LoadMerkle(uint32_t node_index, const Inode& inode,
                                    const fzl::OwnedVmoMapper& vmo) const {
   const uint32_t num_merkle_blocks = ComputeNumMerkleTreeBlocks(inode);
   fs::Ticker ticker(metrics_->Collecting());
   auto bytes_read = LoadBlocks(node_index, /*block_offset=*/0, num_merkle_blocks, vmo);
   if (bytes_read.is_error()) {
     FS_TRACE_ERROR("blobfs: Failed to load Merkle tree: %s\n", bytes_read.status_string());
     return bytes_read.error_value();
   }

   metrics_->IncrementMerkleDiskRead(bytes_read.value(), ticker.End());
   return ZX_OK;
 }

 zx_status_t BlobLoader::LoadData(uint32_t node_index, const Inode& inode,
                                  const fzl::OwnedVmoMapper& vmo) const {
   TRACE_DURATION("blobfs", "BlobLoader::LoadData");

   const uint32_t num_merkle_blocks = ComputeNumMerkleTreeBlocks(inode);
   const uint32_t num_data_blocks = inode.block_count - num_merkle_blocks;
   fs::Ticker ticker(metrics_->Collecting());
   auto bytes_read = LoadBlocks(node_index, num_merkle_blocks, num_data_blocks, vmo);
   if (bytes_read.is_error()) {
     return bytes_read.error_value();
   }
   metrics_->unpaged_read_metrics().IncrementDiskRead(CompressionAlgorithm::UNCOMPRESSED,
                                                      bytes_read.value(), ticker.End());
   return ZX_OK;
 }

 zx_status_t BlobLoader::LoadAndDecompressData(uint32_t node_index, const Inode& inode,
                                               const fzl::OwnedVmoMapper& vmo) const {
   zx::status<CompressionAlgorithm> algorithm_or = AlgorithmForInode(inode);
   if (algorithm_or.is_error()) {
     FS_TRACE_ERROR("Blob has no known compression format\n");
     return algorithm_or.status_value();
   }
   CompressionAlgorithm algorithm = algorithm_or.value();
   ZX_DEBUG_ASSERT(algorithm != CompressionAlgorithm::UNCOMPRESSED);

   const uint32_t num_merkle_blocks = ComputeNumMerkleTreeBlocks(inode);
   const uint32_t num_data_blocks = inode.block_count - num_merkle_blocks;
   size_t compressed_size;
   if (mul_overflow(num_data_blocks, kBlobfsBlockSize, &compressed_size)) {
     FS_TRACE_ERROR("Multiplication overflow\n");
     return ZX_ERR_OUT_OF_RANGE;
   }

   TRACE_DURATION("blobfs", "BlobLoader::LoadAndDecompressData", "compressed_size", compressed_size,
                  "blob_size", inode.blob_size);

   // Create and attach a transfer VMO for fetching the compressed contents from the block FIFO.
   fbl::StringBuffer<ZX_MAX_NAME_LEN> vmo_name;
   FormatBlobCompressedVmoName(node_index, &vmo_name);
   fzl::OwnedVmoMapper compressed_mapper;
   zx_status_t status = compressed_mapper.CreateAndMap(compressed_size, vmo_name.c_str());
   if (status != ZX_OK) {
     FS_TRACE_ERROR("Failed to initialized compressed vmo; error: %d\n", status);
     return status;
   }

   fs::Ticker read_ticker(metrics_->Collecting());
   auto bytes_read = LoadBlocks(node_index, num_merkle_blocks, num_data_blocks, compressed_mapper);
   if (bytes_read.is_error()) {
     return bytes_read.error_value();
   }
   metrics_->unpaged_read_metrics().IncrementDiskRead(algorithm, bytes_read.value(),
                                                      read_ticker.End());

   fs::Ticker ticker(metrics_->Collecting());

   // Decompress into the target buffer.
   size_t target_size = inode.blob_size;
   const void* compressed_buffer = compressed_mapper.start();
   std::unique_ptr<Decompressor> decompressor;
   status = Decompressor::Create(algorithm, &decompressor);
   if (status != ZX_OK) {
     FS_TRACE_ERROR("Failed to create decompressor, status=%d", status);
     return status;
   }

   status = decompressor->Decompress(vmo.start(), &target_size, compressed_buffer, compressed_size);
   if (status != ZX_OK) {
     FS_TRACE_ERROR("Failed to decompress data: %s\n", zx_status_get_string(status));
     return status;
   } else if (target_size != inode.blob_size) {
     FS_TRACE_ERROR("Failed to fully decompress blob (%zu of %zu expected)\n", target_size,
                    inode.blob_size);
     return ZX_ERR_IO_DATA_INTEGRITY;
   }

   metrics_->unpaged_read_metrics().IncrementDecompression(algorithm, inode.blob_size, ticker.End());

   return ZX_OK;
 }

 zx::status<uint64_t> BlobLoader::LoadBlocks(uint32_t node_index, uint32_t block_offset,
                                             uint32_t block_count,
                                             const fzl::OwnedVmoMapper& vmo) const {
   TRACE_DURATION("blobfs", "BlobLoader::LoadBlocks", "block_count", block_count);

   zx_status_t status;
   // Attach |vmo| for transfer to the block FIFO.
   storage::OwnedVmoid vmoid(txn_manager_);
   if ((status = vmoid.AttachVmo(vmo.vmo())) != ZX_OK) {
     FS_TRACE_ERROR("Failed to attach VMO to block device; error: %s\n",
                    zx_status_get_string(status));
     return zx::error(status);
   }

   fs::ReadTxn txn(txn_manager_);

   const uint64_t kDataStart = DataStartBlock(txn_manager_->Info());
   BlockIterator block_iter = block_iter_provider_->BlockIteratorByNodeIndex(node_index);
   if ((status = IterateToBlock(&block_iter, block_offset)) != ZX_OK) {
     FS_TRACE_ERROR("blobfs: Failed to seek to starting block: %s\n", zx_status_get_string(status));
     return zx::error(status);
   }

   status = StreamBlocks(
       &block_iter, block_count, [&](uint64_t vmo_offset, uint64_t dev_offset, uint32_t length) {
         txn.Enqueue(vmoid.get(), vmo_offset - block_offset, kDataStart + dev_offset, length);
         return ZX_OK;
       });
   if (status != ZX_OK) {
     FS_TRACE_ERROR("blobfs: Failed to stream blocks: %s\n", zx_status_get_string(status));
     return zx::error(status);
   }
   if ((status = txn.Transact()) != ZX_OK) {
     FS_TRACE_ERROR("blobfs: Failed to flush read transaction: %s\n", zx_status_get_string(status));
     return zx::error(status);
   }

   return zx::ok(uint64_t{block_count} * kBlobfsBlockSize);
 }

 }  // namespace blobfs
	// Copyright 2020 The Fuchsia Authors. All rights reserved.
	// Use of this source code is governed by a BSD-style license that can be
	// found in the LICENSE file.

	#include "blob-loader.h"

	#include <lib/fzl/owned-vmo-mapper.h>
	#include <lib/zx/status.h>
	#include <zircon/assert.h>
	#include <zircon/errors.h>
	#include <zircon/status.h>

	#include <algorithm>
	#include <memory>

	#include <blobfs/common.h>
	#include <blobfs/compression-settings.h>
	#include <blobfs/format.h>
	#include <fbl/auto_call.h>
	#include <fbl/string_buffer.h>
	#include <storage/buffer/owned_vmoid.h>

	#include "blob-verifier.h"
	#include "compression/chunked.h"
	#include "compression/decompressor.h"
	#include "compression/seekable-decompressor.h"
	#include "compression/zstd-seekable-blob-collection.h"
	#include "iterator/block-iterator.h"

	namespace blobfs {

	namespace {

	// TODO(jfsulliv): Rationalize this with the size limits for chunk-compression headers.
	constexpr size_t kScratchBufferSize = 4 * kBlobfsBlockSize;

	} // namespace

	BlobLoader::BlobLoader(TransactionManager* txn_manager, BlockIteratorProvider* block_iter_provider,
	NodeFinder* node_finder, pager::UserPager* pager, BlobfsMetrics* metrics,
	ZSTDSeekableBlobCollection* zstd_seekable_blob_collection,
	fzl::OwnedVmoMapper scratch_vmo)
	: txn_manager_(txn_manager),
	block_iter_provider_(block_iter_provider),
	node_finder_(node_finder),
	pager_(pager),
	metrics_(metrics),
	zstd_seekable_blob_collection_(zstd_seekable_blob_collection),
	scratch_vmo_(std::move(scratch_vmo)) {}

	zx::status<BlobLoader> BlobLoader::Create(
	TransactionManager* txn_manager, BlockIteratorProvider* block_iter_provider,
	NodeFinder* node_finder, pager::UserPager* pager, BlobfsMetrics* metrics,
	ZSTDSeekableBlobCollection* zstd_seekable_blob_collection) {
	fzl::OwnedVmoMapper scratch_vmo;
	zx_status_t status = scratch_vmo.CreateAndMap(kScratchBufferSize, "blobfs-loader");
	if (status != ZX_OK) {
	FS_TRACE_ERROR("blobfs: Failed to map scratch vmo: %s\n", zx_status_get_string(status));
	return zx::error(status);
	}
	return zx::ok(BlobLoader(txn_manager, block_iter_provider, node_finder, pager, metrics,
	zstd_seekable_blob_collection, std::move(scratch_vmo)));
	}

	void BlobLoader::Reset() { scratch_vmo_.Reset(); }

	zx_status_t BlobLoader::LoadBlob(uint32_t node_index,
	const BlobCorruptionNotifier* corruption_notifier,
	fzl::OwnedVmoMapper* data_out, fzl::OwnedVmoMapper* merkle_out) {
	ZX_DEBUG_ASSERT(scratch_vmo_.vmo().is_valid());
	const InodePtr inode = node_finder_->GetNode(node_index);
	// LoadBlob should only be called for Inodes. If this doesn't hold, one of two things happened:
	// - Programmer error
	// - Corruption of a blob's Inode
	// In either case it is preferable to ASSERT than to return an error here, since the first case
	// should happen only during development and in the second case there may be more corruption and
	// we want to unmount the filesystem before any more damage is done.
	ZX_ASSERT(inode->header.IsInode() && inode->header.IsAllocated());

	TRACE_DURATION("blobfs", "BlobLoader::LoadBlob", "blob_size", inode->blob_size);

	const uint64_t num_data_blocks = BlobDataBlocks(*inode);
	if (num_data_blocks == 0) {
	// No data to load for the null blob.
	return ZX_OK;
	}

	fzl::OwnedVmoMapper merkle_mapper;
	std::unique_ptr<BlobVerifier> verifier;
	zx_status_t status;
	if ((status = InitMerkleVerifier(node_index, *inode, corruption_notifier, &merkle_mapper,
	&verifier)) != ZX_OK) {
	return status;
	}

	size_t data_vmo_size;
	if (mul_overflow(num_data_blocks, kBlobfsBlockSize, &data_vmo_size)) {
	FS_TRACE_ERROR("Multiplication overflow");
	return ZX_ERR_OUT_OF_RANGE;
	}

	fbl::StringBuffer<ZX_MAX_NAME_LEN> data_vmo_name;
	FormatBlobDataVmoName(node_index, &data_vmo_name);

	fzl::OwnedVmoMapper data_mapper;
	status = data_mapper.CreateAndMap(data_vmo_size, data_vmo_name.c_str());
	if (status != ZX_OK) {
	FS_TRACE_ERROR("blobfs: Failed to initialize data vmo; error: %s\n",
	zx_status_get_string(status));
	return status;
	}
	status = inode->IsCompressed() ? LoadAndDecompressData(node_index, *inode, data_mapper)
	: LoadData(node_index, *inode, data_mapper);
	if (status != ZX_OK) {
	return status;
	}

	if ((status = verifier->Verify(data_mapper.start(), inode->blob_size, data_vmo_size)) != ZX_OK) {
	return status;
	}

	*data_out = std::move(data_mapper);
	if (merkle_mapper.vmo().is_valid()) {
	*merkle_out = std::move(merkle_mapper);
	}
	return ZX_OK;
	}

	zx_status_t BlobLoader::LoadBlobPaged(uint32_t node_index,
	const BlobCorruptionNotifier* corruption_notifier,
	std::unique_ptr<pager::PageWatcher>* page_watcher_out,
	fzl::OwnedVmoMapper* data_out,
	fzl::OwnedVmoMapper* merkle_out) {
	ZX_DEBUG_ASSERT(scratch_vmo_.vmo().is_valid());
	const InodePtr inode = node_finder_->GetNode(node_index);
	// LoadBlobPaged should only be called for Inodes. If this doesn't hold, one of two things
	// happened:
	// - Programmer error
	// - Corruption of a blob's Inode
	// In either case it is preferable to ASSERT than to return an error here, since the first case
	// should happen only during development and in the second case there may be more corruption and
	// we want to unmount the filesystem before any more damage is done.
	ZX_ASSERT(inode->header.IsInode() && inode->header.IsAllocated());

	TRACE_DURATION("blobfs", "BlobLoader::LoadBlobPaged", "blob_size", inode->blob_size);

	const uint64_t num_data_blocks = BlobDataBlocks(*inode);
	if (num_data_blocks == 0) {
	// No data to load for the null blob.
	return ZX_OK;
	}

	fzl::OwnedVmoMapper merkle_mapper;
	std::unique_ptr<BlobVerifier> verifier;
	zx_status_t status;
	if ((status = InitMerkleVerifier(node_index, *inode, corruption_notifier, &merkle_mapper,
	&verifier)) != ZX_OK) {
	return status;
	}

	std::unique_ptr<SeekableDecompressor> decompressor;
	ZSTDSeekableBlobCollection* zstd_seekable_blob_collection = nullptr;
	if ((status = InitForDecompression(node_index, inode, verifier, &decompressor,
	&zstd_seekable_blob_collection)) != ZX_OK) {
	return status;
	}

	pager::UserPagerInfo userpager_info;
	userpager_info.identifier = node_index;
	userpager_info.data_start_bytes = ComputeNumMerkleTreeBlocks(inode) kBlobfsBlockSize;
	userpager_info.data_length_bytes = inode->blob_size;
	userpager_info.verifier = std::move(verifier);
	userpager_info.decompressor = std::move(decompressor);
	userpager_info.zstd_seekable_blob_collection = zstd_seekable_blob_collection;
	auto page_watcher = std::make_unique<pager::PageWatcher>(pager_, std::move(userpager_info));

	fbl::StringBuffer<ZX_MAX_NAME_LEN> data_vmo_name;
	FormatBlobDataVmoName(node_index, &data_vmo_name);

	zx::vmo data_vmo;
	size_t data_vmo_size;
	if (mul_overflow(num_data_blocks, kBlobfsBlockSize, &data_vmo_size)) {
	FS_TRACE_ERROR("Multiplication overflow");
	return ZX_ERR_OUT_OF_RANGE;
	}
	if ((status = page_watcher->CreatePagedVmo(data_vmo_size, &data_vmo)) != ZX_OK) {
	return status;
	}
	data_vmo.set_property(ZX_PROP_NAME, data_vmo_name.c_str(), data_vmo_name.length());

	fzl::OwnedVmoMapper data_mapper;
	if ((status = data_mapper.Map(std::move(data_vmo))) != ZX_OK) {
	FS_TRACE_ERROR("blobfs: Failed to create mapping for data vmo: %s\n",
	zx_status_get_string(status));
	return status;
	}

	*page_watcher_out = std::move(page_watcher);
	*data_out = std::move(data_mapper);
	if (merkle_mapper.vmo().is_valid()) {
	*merkle_out = std::move(merkle_mapper);
	}
	return ZX_OK;
	}

	zx_status_t BlobLoader::InitMerkleVerifier(uint32_t node_index, const Inode& inode,
	const BlobCorruptionNotifier* notifier,
	fzl::OwnedVmoMapper* vmo_out,
	std::unique_ptr<BlobVerifier>* verifier_out) {
	uint64_t num_merkle_blocks = ComputeNumMerkleTreeBlocks(inode);
	if (num_merkle_blocks == 0) {
	return BlobVerifier::CreateWithoutTree(digest::Digest(inode.merkle_root_hash), metrics_,
	inode.blob_size, notifier, verifier_out);
	}

	fzl::OwnedVmoMapper merkle_mapper;
	std::unique_ptr<BlobVerifier> verifier;

	size_t merkle_vmo_size;
	if (mul_overflow(num_merkle_blocks, kBlobfsBlockSize, &merkle_vmo_size)) {
	FS_TRACE_ERROR("Multiplication overflow");
	return ZX_ERR_OUT_OF_RANGE;
	}

	fbl::StringBuffer<ZX_MAX_NAME_LEN> merkle_vmo_name;
	FormatBlobMerkleVmoName(node_index, &merkle_vmo_name);

	zx_status_t status;
	if ((status = merkle_mapper.CreateAndMap(merkle_vmo_size, merkle_vmo_name.c_str())) != ZX_OK) {
	FS_TRACE_ERROR("blobfs: Failed to initialize merkle vmo; error: %s\n",
	zx_status_get_string(status));
	return status;
	}

	if ((status = LoadMerkle(node_index, inode, merkle_mapper)) != ZX_OK) {
	return status;
	}

	if ((status = BlobVerifier::Create(digest::Digest(inode.merkle_root_hash), metrics_,
	merkle_mapper.start(), merkle_vmo_size, inode.blob_size,
	notifier, &verifier)) != ZX_OK) {
	return status;
	}

	*vmo_out = std::move(merkle_mapper);
	*verifier_out = std::move(verifier);
	return ZX_OK;
	}

	zx_status_t BlobLoader::InitForDecompression(
	uint32_t node_index, const Inode& inode, const BlobVerifier& verifier,
	std::unique_ptr<SeekableDecompressor>* decompressor_out,
	ZSTDSeekableBlobCollection** zstd_seekable_blob_collection_out) {
	zx::status<CompressionAlgorithm> algorithm_status = AlgorithmForInode(inode);
	if (algorithm_status.is_error()) {
	FS_TRACE_ERROR("blobfs: Cannot decode blob due to multiple compression flags.\n");
	return algorithm_status.status_value();
	}
	CompressionAlgorithm algorithm = algorithm_status.value();

	switch (algorithm) {
	case CompressionAlgorithm::UNCOMPRESSED:
	return ZX_OK;
	case CompressionAlgorithm::ZSTD_SEEKABLE:
	*zstd_seekable_blob_collection_out = zstd_seekable_blob_collection_;
	break;
	case CompressionAlgorithm::CHUNKED:
	break;
	case CompressionAlgorithm::LZ4:
	case CompressionAlgorithm::ZSTD:
	// Callers should have guarded against calling this code path with an algorithm that
	// does not support paging.
	FS_TRACE_ERROR(
	"Algorithm %s does not support paging; this path should not be called.\n"
	"This is most likely programmer error.\n",
	CompressionAlgorithmToString(algorithm));
	ZX_DEBUG_ASSERT(false);
	return ZX_ERR_NOT_SUPPORTED;
	}

	TRACE_DURATION("blobfs", "BlobLoader::InitDecompressor");

	// The first few blocks of data contain the seek table, which we need to read to initialize
	// the decompressor. Read these from disk.

	zx_status_t status;

	uint32_t merkle_blocks = ComputeNumMerkleTreeBlocks(inode);
	// We don't know exactly how long the header is, so fill up as much of the scratch VMO as we can.
	// (The header should never be bigger than the size of the scratch VMO.)
	ZX_DEBUG_ASSERT(scratch_vmo_.size() % kBlobfsBlockSize == 0);
	uint32_t num_data_blocks = static_cast<uint32_t>(scratch_vmo_.size()) / kBlobfsBlockSize;
	num_data_blocks = std::min(num_data_blocks, inode.block_count - merkle_blocks);
	if (num_data_blocks == 0) {
	FS_TRACE_ERROR("blobfs: No data blocks; corrupted inode?\n");
	return ZX_ERR_BAD_STATE;
	}

	auto bytes_read = LoadBlocks(node_index, merkle_blocks, num_data_blocks, scratch_vmo_);
	if (bytes_read.is_error()) {
	FS_TRACE_ERROR("blobfs: Failed to load compression header: %s\n", bytes_read.status_string());
	return bytes_read.error_value();
	}

	if ((status = SeekableChunkedDecompressor::CreateDecompressor(
	scratch_vmo_.start(), num_data_blocks * kBlobfsBlockSize, inode.blob_size,
	decompressor_out)) != ZX_OK) {
	FS_TRACE_ERROR("blobfs: Failed to init decompressor: %s\n", zx_status_get_string(status));
	return status;
	}

	return ZX_OK;
	}

	zx_status_t BlobLoader::LoadMerkle(uint32_t node_index, const Inode& inode,
	const fzl::OwnedVmoMapper& vmo) const {
	const uint32_t num_merkle_blocks = ComputeNumMerkleTreeBlocks(inode);
	fs::Ticker ticker(metrics_->Collecting());
	auto bytes_read = LoadBlocks(node_index, /block_offset=/0, num_merkle_blocks, vmo);
	if (bytes_read.is_error()) {
	FS_TRACE_ERROR("blobfs: Failed to load Merkle tree: %s\n", bytes_read.status_string());
	return bytes_read.error_value();
	}

	metrics_->IncrementMerkleDiskRead(bytes_read.value(), ticker.End());
	return ZX_OK;
	}

	zx_status_t BlobLoader::LoadData(uint32_t node_index, const Inode& inode,
	const fzl::OwnedVmoMapper& vmo) const {
	TRACE_DURATION("blobfs", "BlobLoader::LoadData");

	const uint32_t num_merkle_blocks = ComputeNumMerkleTreeBlocks(inode);
	const uint32_t num_data_blocks = inode.block_count - num_merkle_blocks;
	fs::Ticker ticker(metrics_->Collecting());
	auto bytes_read = LoadBlocks(node_index, num_merkle_blocks, num_data_blocks, vmo);
	if (bytes_read.is_error()) {
	return bytes_read.error_value();
	}
	metrics_->unpaged_read_metrics().IncrementDiskRead(CompressionAlgorithm::UNCOMPRESSED,
	bytes_read.value(), ticker.End());
	return ZX_OK;
	}

	zx_status_t BlobLoader::LoadAndDecompressData(uint32_t node_index, const Inode& inode,
	const fzl::OwnedVmoMapper& vmo) const {
	zx::status<CompressionAlgorithm> algorithm_or = AlgorithmForInode(inode);
	if (algorithm_or.is_error()) {
	FS_TRACE_ERROR("Blob has no known compression format\n");
	return algorithm_or.status_value();
	}
	CompressionAlgorithm algorithm = algorithm_or.value();
	ZX_DEBUG_ASSERT(algorithm != CompressionAlgorithm::UNCOMPRESSED);

	const uint32_t num_merkle_blocks = ComputeNumMerkleTreeBlocks(inode);
	const uint32_t num_data_blocks = inode.block_count - num_merkle_blocks;
	size_t compressed_size;
	if (mul_overflow(num_data_blocks, kBlobfsBlockSize, &compressed_size)) {
	FS_TRACE_ERROR("Multiplication overflow\n");
	return ZX_ERR_OUT_OF_RANGE;
	}

	TRACE_DURATION("blobfs", "BlobLoader::LoadAndDecompressData", "compressed_size", compressed_size,
	"blob_size", inode.blob_size);

	// Create and attach a transfer VMO for fetching the compressed contents from the block FIFO.
	fbl::StringBuffer<ZX_MAX_NAME_LEN> vmo_name;
	FormatBlobCompressedVmoName(node_index, &vmo_name);
	fzl::OwnedVmoMapper compressed_mapper;
	zx_status_t status = compressed_mapper.CreateAndMap(compressed_size, vmo_name.c_str());
	if (status != ZX_OK) {
	FS_TRACE_ERROR("Failed to initialized compressed vmo; error: %d\n", status);
	return status;
	}

	fs::Ticker read_ticker(metrics_->Collecting());
	auto bytes_read = LoadBlocks(node_index, num_merkle_blocks, num_data_blocks, compressed_mapper);
	if (bytes_read.is_error()) {
	return bytes_read.error_value();
	}
	metrics_->unpaged_read_metrics().IncrementDiskRead(algorithm, bytes_read.value(),
	read_ticker.End());

	fs::Ticker ticker(metrics_->Collecting());

	// Decompress into the target buffer.
	size_t target_size = inode.blob_size;
	const void* compressed_buffer = compressed_mapper.start();
	std::unique_ptr<Decompressor> decompressor;
	status = Decompressor::Create(algorithm, &decompressor);
	if (status != ZX_OK) {
	FS_TRACE_ERROR("Failed to create decompressor, status=%d", status);
	return status;
	}

	status = decompressor->Decompress(vmo.start(), &target_size, compressed_buffer, compressed_size);
	if (status != ZX_OK) {
	FS_TRACE_ERROR("Failed to decompress data: %s\n", zx_status_get_string(status));
	return status;
	} else if (target_size != inode.blob_size) {
	FS_TRACE_ERROR("Failed to fully decompress blob (%zu of %zu expected)\n", target_size,
	inode.blob_size);
	return ZX_ERR_IO_DATA_INTEGRITY;
	}

	metrics_->unpaged_read_metrics().IncrementDecompression(algorithm, inode.blob_size, ticker.End());

	return ZX_OK;
	}

	zx::status<uint64_t> BlobLoader::LoadBlocks(uint32_t node_index, uint32_t block_offset,
	uint32_t block_count,
	const fzl::OwnedVmoMapper& vmo) const {
	TRACE_DURATION("blobfs", "BlobLoader::LoadBlocks", "block_count", block_count);

	zx_status_t status;
	// Attach \|vmo\| for transfer to the block FIFO.
	storage::OwnedVmoid vmoid(txn_manager_);
	if ((status = vmoid.AttachVmo(vmo.vmo())) != ZX_OK) {
	FS_TRACE_ERROR("Failed to attach VMO to block device; error: %s\n",
	zx_status_get_string(status));
	return zx::error(status);
	}

	fs::ReadTxn txn(txn_manager_);

	const uint64_t kDataStart = DataStartBlock(txn_manager_->Info());
	BlockIterator block_iter = block_iter_provider_->BlockIteratorByNodeIndex(node_index);
	if ((status = IterateToBlock(&block_iter, block_offset)) != ZX_OK) {
	FS_TRACE_ERROR("blobfs: Failed to seek to starting block: %s\n", zx_status_get_string(status));
	return zx::error(status);
	}

	status = StreamBlocks(
	&block_iter, block_count, [&](uint64_t vmo_offset, uint64_t dev_offset, uint32_t length) {
	txn.Enqueue(vmoid.get(), vmo_offset - block_offset, kDataStart + dev_offset, length);
	return ZX_OK;
	});
	if (status != ZX_OK) {
	FS_TRACE_ERROR("blobfs: Failed to stream blocks: %s\n", zx_status_get_string(status));
	return zx::error(status);
	}
	if ((status = txn.Transact()) != ZX_OK) {
	FS_TRACE_ERROR("blobfs: Failed to flush read transaction: %s\n", zx_status_get_string(status));
	return zx::error(status);
	}

	return zx::ok(uint64_t{block_count} * kBlobfsBlockSize);
	}

	} // namespace blobfs