text: // Copyright 2017 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.h"
  
  #include <assert.h>
  #include <ctype.h>
  #include <fuchsia/device/c/fidl.h>
  #include <fuchsia/io/llcpp/fidl.h>
  #include <lib/sync/completion.h>
  #include <stdlib.h>
  #include <string.h>
  #include <zircon/device/vfs.h>
  #include <zircon/status.h>
  #include <zircon/syscalls.h>
  
  #include <utility>
  #include <vector>
  
  #include <digest/digest.h>
  #include <fbl/auto_call.h>
  #include <fbl/ref_ptr.h>
  #include <fbl/string_buffer.h>
  #include <fbl/string_piece.h>
  #include <fs/journal/data_streamer.h>
  #include <fs/metrics/events.h>
  #include <fs/transaction/writeback.h>
  #include <fs/vfs_types.h>
  
  #include "blobfs.h"
  #include "compression/lz4.h"
  #include "compression/zstd.h"
  #include "iterator/allocated-extent-iterator.h"
  #include "iterator/block-iterator.h"
  #include "iterator/extent-iterator.h"
  #include "iterator/node-populator.h"
  #include "iterator/vector-extent-iterator.h"
  #include "metrics.h"
  
  namespace blobfs {
  namespace {
  
  using digest::Digest;
  using digest::MerkleTreeCreator;
  using digest::MerkleTreeVerifier;
  
  // Blob's vmo names have following pattern
  // "blob-1abc8" or "compressedBlob-5c"
  constexpr char kBlobVmoNamePrefix[] = "blob";
  constexpr char kCompressedBlobVmoNamePrefix[] = "compressedBlob";
  
  void FormatVmoName(const char* prefix, fbl::StringBuffer<ZX_MAX_NAME_LEN>* vmo_name, size_t index) {
    vmo_name->Clear();
    vmo_name->AppendPrintf("%s-%lx", prefix, index);
  }
  
  }  // namespace
  
  zx_status_t Blob::Verify() const {
    TRACE_DURATION("blobfs", "Blobfs::Verify");
    fs::Ticker ticker(blobfs_->Metrics().Collecting());
  
    const void* data = inode_.blob_size ? GetData() : nullptr;
    const void* tree = inode_.blob_size ? GetMerkle() : nullptr;
    const uint64_t data_size = inode_.blob_size;
  
    // TODO(smklein): We could lazily verify more of the VMO if
    // we could fault in pages on-demand.
    //
    // For now, we aggressively verify the entire VMO up front.
    MerkleTreeVerifier mtv;
    zx_status_t status = mtv.SetDataLength(data_size);
    size_t merkle_size = mtv.GetTreeLength();
    if (status != ZX_OK ||
        (status = mtv.SetTree(tree, merkle_size, GetKey(), digest::kSha256Length)) != ZX_OK ||
        (status = mtv.Verify(data, data_size, 0)) != ZX_OK) {
      Digest digest(GetKey());
      FS_TRACE_ERROR("blobfs verify(%s) Failure: %s\n", digest.ToString().c_str(),
                     zx_status_get_string(status));
    }
    blobfs_->Metrics().UpdateMerkleVerify(data_size, merkle_size, ticker.End());
  
    return status;
  }
  
  zx_status_t Blob::InitVmos() {
    TRACE_DURATION("blobfs", "Blobfs::InitVmos");
  
    if (mapping_.vmo()) {
      return ZX_OK;
    }
  
    uint64_t data_blocks = BlobDataBlocks(inode_);
    uint64_t merkle_blocks = MerkleTreeBlocks(inode_);
    uint64_t num_blocks = data_blocks + merkle_blocks;
  
    if (num_blocks == 0) {
      // No need to initialize VMO for null blob.
      return ZX_OK;
    }
  
    // Reverts blob back to uninitialized state on error.
    auto cleanup = fbl::MakeAutoCall([this]() { BlobCloseHandles(); });
  
    size_t vmo_size;
    if (mul_overflow(num_blocks, kBlobfsBlockSize, &vmo_size)) {
      FS_TRACE_ERROR("Multiplication overflow");
      return ZX_ERR_OUT_OF_RANGE;
    }
  
    fbl::StringBuffer<ZX_MAX_NAME_LEN> vmo_name;
    FormatVmoName(kBlobVmoNamePrefix, &vmo_name, Ino());
  
    // Use the pager only if the blob is uncompressed AND blobfs has a pager set up.
    bool use_pager =
        (((inode_.header.flags & (kBlobFlagLZ4Compressed | kBlobFlagZSTDCompressed)) == 0) &&
         blobfs_->PagingEnabled());
  
    zx_status_t status;
    if (use_pager) {
      page_watcher_ = std::make_unique<PageWatcher>(blobfs_, GetMapIndex());
      zx::vmo vmo;
      status = page_watcher_->CreatePagedVmo(vmo_size, &vmo);
      if (status != ZX_OK) {
        return status;
      }
  
      vmo.set_property(ZX_PROP_NAME, vmo_name.c_str(), vmo_name.length());
      status = mapping_.Map(std::move(vmo));
    } else {
      status = mapping_.CreateAndMap(vmo_size, vmo_name.c_str());
    }
  
    if (status != ZX_OK) {
      FS_TRACE_ERROR("Failed to initialize vmo; error: %s\n", zx_status_get_string(status));
      return status;
    }
    if ((status = blobfs_->AttachVmo(mapping_.vmo(), &vmoid_)) != ZX_OK) {
      FS_TRACE_ERROR("Failed to attach VMO to block device; error: %s\n",
                     zx_status_get_string(status));
      return status;
    }
  
    if (use_pager) {
      // TODO(rashaeqbal): Remove this once Merkle tree verification no longer requires the entire
      // blob to be present (fxb/35678).
      // For now page in the entire blob. This is different from the InitUncompressed() path below,
      // which reads the data directly into the blob's VMO. This path exercises the blobfs pager.
      char bytes[kBlobfsBlockSize];
      for (uint64_t blk = 0; blk < num_blocks; blk++) {
        status = mapping_.vmo().read(static_cast<void*>(bytes), blk * kBlobfsBlockSize,
                                     kBlobfsBlockSize);
        if (status != ZX_OK) {
          FS_TRACE_ERROR("Failed to read in data at block %zu; error %s\n", blk,
                         zx_status_get_string(status));
          return status;
        }
      }
    } else if ((inode_.header.flags & kBlobFlagLZ4Compressed) != 0) {
      status = InitCompressed(CompressionAlgorithm::LZ4);
    } else if ((inode_.header.flags & kBlobFlagZSTDCompressed) != 0) {
      status = InitCompressed(CompressionAlgorithm::ZSTD);
    } else {
      status = InitUncompressed();
    }
  
    if (status != ZX_OK) {
      return status;
    }
    // TODO(rashaeqbal): Remove this once fxb/35678 is in. Instead verify pages as they are populated
    // by the pager.
    status = Verify();
    if (status != ZX_OK) {
      return status;
    }
  
    cleanup.cancel();
    return ZX_OK;
  }
  
  zx_status_t Blob::InitCompressed(CompressionAlgorithm algorithm) {
    TRACE_DURATION("blobfs", "Blobfs::InitCompressed", "size", inode_.blob_size, "blocks",
                   inode_.block_count);
    fs::Ticker ticker(blobfs_->Metrics().Collecting());
    fs::ReadTxn txn(blobfs_);
    uint32_t merkle_blocks = MerkleTreeBlocks(inode_);
  
    fzl::OwnedVmoMapper compressed_mapper;
    uint32_t compressed_blocks = (inode_.block_count - merkle_blocks);
    size_t compressed_size;
    if (mul_overflow(compressed_blocks, kBlobfsBlockSize, &compressed_size)) {
      FS_TRACE_ERROR("Multiplication overflow\n");
      return ZX_ERR_OUT_OF_RANGE;
    }
  
    fbl::StringBuffer<ZX_MAX_NAME_LEN> vmo_name;
    FormatVmoName(kCompressedBlobVmoNamePrefix, &vmo_name, Ino());
    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;
    }
    vmoid_t compressed_vmoid;
    status = blobfs_->AttachVmo(compressed_mapper.vmo(), &compressed_vmoid);
    if (status != ZX_OK) {
      FS_TRACE_ERROR("Failed to attach compressed VMO to blkdev: %d\n", status);
      return status;
    }
  
    auto detach =
        fbl::MakeAutoCall([this, &compressed_vmoid]() { blobfs_->DetachVmo(compressed_vmoid); });
  
    const uint64_t kDataStart = DataStartBlock(blobfs_->Info());
    AllocatedExtentIterator extent_iter(blobfs_->GetNodeFinder(), GetMapIndex());
    BlockIterator block_iter(&extent_iter);
  
    // Read the uncompressed merkle tree into the start of the blob's VMO.
    status = StreamBlocks(&block_iter, merkle_blocks,
                          [&](uint64_t vmo_offset, uint64_t dev_offset, uint32_t length) {
                            txn.Enqueue(vmoid_, vmo_offset, dev_offset + kDataStart, length);
                            return ZX_OK;
                          });
    if (status != ZX_OK) {
      return status;
    }
  
    // Read the compressed blocks into the compressed VMO, accounting for the merkle blocks
    // which have already been seen.
    ZX_DEBUG_ASSERT(block_iter.BlockIndex() == merkle_blocks);
  
    status = StreamBlocks(&block_iter, compressed_blocks,
                          [&](uint64_t vmo_offset, uint64_t dev_offset, uint32_t length) {
                            txn.Enqueue(compressed_vmoid, vmo_offset - merkle_blocks,
                                        dev_offset + kDataStart, length);
                            return ZX_OK;
                          });
  
    if (status != ZX_OK) {
      return status;
    }
  
    if ((status = txn.Transact()) != ZX_OK) {
      FS_TRACE_ERROR("Failed to flush read transaction: %d\n", status);
      return status;
    }
  
    fs::Duration read_time = ticker.End();
    ticker.Reset();
  
    // Decompress the compressed data into the target buffer.
    size_t target_size = inode_.blob_size;
    const void* compressed_buffer = compressed_mapper.start();
    switch (algorithm) {
      case CompressionAlgorithm::LZ4:
        status = LZ4Decompress(GetData(), &target_size, compressed_buffer, &compressed_size);
        break;
      case CompressionAlgorithm::ZSTD:
        status = ZSTDDecompress(GetData(), &target_size, compressed_buffer, &compressed_size);
        break;
      default:
        FS_TRACE_ERROR("Unsupported decompression algorithm");
        return ZX_ERR_NOT_SUPPORTED;
    }
  
    if (status != ZX_OK) {
      FS_TRACE_ERROR("Failed to decompress data: %d\n", 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;
    }
  
    blobfs_->Metrics().UpdateMerkleDecompress(compressed_blocks * kBlobfsBlockSize, inode_.blob_size,
                                              read_time, ticker.End());
    return ZX_OK;
  }
  
  zx_status_t Blob::InitUncompressed() {
    TRACE_DURATION("blobfs", "Blobfs::InitUncompressed", "size", inode_.blob_size, "blocks",
                   inode_.block_count);
    fs::Ticker ticker(blobfs_->Metrics().Collecting());
    fs::ReadTxn txn(blobfs_);
    AllocatedExtentIterator extent_iter(blobfs_->GetNodeFinder(), GetMapIndex());
    BlockIterator block_iter(&extent_iter);
    // Read both the uncompressed merkle tree and data.
    const uint64_t blob_data_blocks = BlobDataBlocks(inode_);
    const uint64_t merkle_blocks = MerkleTreeBlocks(inode_);
    if (blob_data_blocks + merkle_blocks > std::numeric_limits<uint32_t>::max()) {
      return ZX_ERR_IO_DATA_INTEGRITY;
    }
    const uint32_t length = static_cast<uint32_t>(blob_data_blocks + merkle_blocks);
    const uint64_t data_start = DataStartBlock(blobfs_->Info());
    zx_status_t status = StreamBlocks(
        &block_iter, length, [&](uint64_t vmo_offset, uint64_t dev_offset, uint32_t length) {
          txn.Enqueue(vmoid_, vmo_offset, dev_offset + data_start, length);
          return ZX_OK;
        });
  
    if (status != ZX_OK) {
      return status;
    }
  
    status = txn.Transact();
    if (status != ZX_OK) {
      return status;
    }
    blobfs_->Metrics().UpdateMerkleDiskRead(length * kBlobfsBlockSize, ticker.End());
    return status;
  }
  
  void Blob::PopulateInode(uint32_t node_index) {
    ZX_DEBUG_ASSERT(map_index_ == 0);
    SetState(kBlobStateReadable);
    map_index_ = node_index;
    Inode* inode = blobfs_->GetNode(node_index);
    inode_ = *inode;
  }
  
  uint64_t Blob::SizeData() const {
    if (GetState() == kBlobStateReadable) {
      return inode_.blob_size;
    }
    return 0;
  }
  
  Blob::Blob(Blobfs* bs, const Digest& digest)
      : CacheNode(digest),
        blobfs_(bs),
        flags_(kBlobStateEmpty),
        syncing_(false),
        clone_watcher_(this) {}
  
  void Blob::BlobCloseHandles() {
    page_watcher_.reset();
    mapping_.Reset();
    readable_event_.reset();
  }
  
  zx_status_t Blob::SpaceAllocate(uint64_t size_data) {
    TRACE_DURATION("blobfs", "Blobfs::SpaceAllocate", "size_data", size_data);
    fs::Ticker ticker(blobfs_->Metrics().Collecting());
  
    if (GetState() != kBlobStateEmpty) {
      return ZX_ERR_BAD_STATE;
    }
  
    auto write_info = std::make_unique<WritebackInfo>();
  
    // Initialize the inode with known fields.
    memset(inode_.merkle_root_hash, 0, sizeof(inode_.merkle_root_hash));
    inode_.blob_size = size_data;
    inode_.block_count = MerkleTreeBlocks(inode_) + static_cast<uint32_t>(BlobDataBlocks(inode_));
  
    // Special case for the null blob: We skip the write phase.
    if (inode_.blob_size == 0) {
      zx_status_t status = blobfs_->GetAllocator()->ReserveNodes(1, &write_info->node_indices);
      if (status != ZX_OK) {
        return status;
      }
      map_index_ = write_info->node_indices[0].index();
      write_info_ = std::move(write_info);
  
      if ((status = Verify()) != ZX_OK) {
        return status;
      }
      SetState(kBlobStateDataWrite);
  
      blobfs_->journal()->schedule_task(
          WriteMetadata().and_then([blob = fbl::RefPtr(this)]() { blob->CompleteSync(); }));
      return ZX_OK;
    }
  
    fbl::Vector<ReservedExtent> extents;
    fbl::Vector<ReservedNode> nodes;
  
    // Reserve space for the blob.
    zx_status_t status = blobfs_->GetAllocator()->ReserveBlocks(inode_.block_count, &extents);
    if (status != ZX_OK) {
      return status;
    }
    if (extents.size() > kMaxBlobExtents) {
      FS_TRACE_ERROR("Error: Block reservation requires too many extents (%zu vs %zu max)\n",
                     extents.size(), kMaxBlobExtents);
      return ZX_ERR_BAD_STATE;
    }
    const ExtentCountType extent_count = static_cast<ExtentCountType>(extents.size());
  
    // Reserve space for all the nodes necessary to contain this blob.
    size_t node_count = NodePopulator::NodeCountForExtents(extent_count);
    status = blobfs_->GetAllocator()->ReserveNodes(node_count, &nodes);
    if (status != ZX_OK) {
      return status;
    }
  
    fbl::StringBuffer<ZX_MAX_NAME_LEN> vmo_name;
    if (inode_.blob_size >= kCompressionMinBytesSaved) {
      write_info->compressor = BlobCompressor::Create(CompressionAlgorithm::ZSTD, inode_.blob_size);
      if (!write_info->compressor) {
        FS_TRACE_ERROR("blobfs: Failed to initialize compressor: %d\n", status);
        return status;
      }
    }
  
    // Open VMOs, so we can begin writing after allocate succeeds.
    fzl::OwnedVmoMapper mapping;
    FormatVmoName(kBlobVmoNamePrefix, &vmo_name, Ino());
    if ((status = mapping.CreateAndMap(inode_.block_count * kBlobfsBlockSize, vmo_name.c_str())) !=
        ZX_OK) {
      return status;
    }
    if ((status = blobfs_->AttachVmo(mapping.vmo(), &vmoid_)) != ZX_OK) {
      return status;
    }
  
    map_index_ = nodes[0].index();
    mapping_ = std::move(mapping);
    write_info->extents = std::move(extents);
    write_info->node_indices = std::move(nodes);
    write_info_ = std::move(write_info);
  
    SetState(kBlobStateDataWrite);
    blobfs_->Metrics().UpdateAllocation(size_data, ticker.End());
    return ZX_OK;
  }
  
  void* Blob::GetData() const {
    return fs::GetBlock(kBlobfsBlockSize, mapping_.start(), MerkleTreeBlocks(inode_));
  }
  
  void* Blob::GetMerkle() const { return mapping_.start(); }
  
  fit::promise<void, zx_status_t> Blob::WriteMetadata() {
    TRACE_DURATION("blobfs", "Blobfs::WriteMetadata");
    assert(GetState() == kBlobStateDataWrite);
  
    // Update the on-disk hash.
    memcpy(inode_.merkle_root_hash, GetKey(), digest::kSha256Length);
  
    // All data has been written to the containing VMO.
    SetState(kBlobStateReadable);
    if (readable_event_.is_valid()) {
      zx_status_t status = readable_event_.signal(0u, ZX_USER_SIGNAL_0);
      if (status != ZX_OK) {
        SetState(kBlobStateError);
        return fit::make_error_promise(status);
      }
    }
  
    atomic_store(&syncing_, true);
  
    storage::UnbufferedOperationsBuilder operations;
    if (inode_.block_count) {
      // We utilize the NodePopulator class to take our reserved blocks and nodes and fill the
      // persistent map with an allocated inode / container.
  
      // If |on_node| is invoked on a node, it means that node was necessary to represent this
      // blob. Persist the node back to durable storge.
      auto on_node = [this, &operations](const ReservedNode& node) {
        blobfs_->PersistNode(node.index(), &operations);
      };
  
      // If |on_extent| is invoked on an extent, it was necessary to represent this blob. Persist
      // the allocation of these blocks back to durable storage.
      //
      // Additionally, because of the compression feature of blobfs, it is possible we reserved
      // more extents than this blob ended up using. Decrement |remaining_blocks| to track if we
      // should exit early.
      size_t remaining_blocks = inode_.block_count;
      auto on_extent = [this, &operations, &remaining_blocks](ReservedExtent& extent) {
        ZX_DEBUG_ASSERT(remaining_blocks > 0);
        if (remaining_blocks >= extent.extent().Length()) {
          // Consume the entire extent.
          remaining_blocks -= extent.extent().Length();
        } else {
          // Consume only part of the extent; we're done iterating.
          extent.SplitAt(static_cast<BlockCountType>(remaining_blocks));
          remaining_blocks = 0;
        }
        blobfs_->PersistBlocks(extent, &operations);
        if (remaining_blocks == 0) {
          return NodePopulator::IterationCommand::Stop;
        }
        return NodePopulator::IterationCommand::Continue;
      };
  
      Inode* mapped_inode = blobfs_->GetNode(map_index_);
      *mapped_inode = inode_;
      NodePopulator populator(blobfs_->GetAllocator(), std::move(write_info_->extents),
                              std::move(write_info_->node_indices));
      ZX_ASSERT(populator.Walk(on_node, on_extent) == ZX_OK);
  
      // Ensure all non-allocation flags are propagated to the inode.
      const uint16_t non_allocation_flags = kBlobFlagZSTDCompressed | kBlobFlagLZ4Compressed;
      mapped_inode->header.flags |= (inode_.header.flags & non_allocation_flags);
    } else {
      // Special case: Empty node.
      ZX_DEBUG_ASSERT(write_info_->node_indices.size() == 1);
      *(blobfs_->GetNode(map_index_)) = inode_;
      const ReservedNode& node = write_info_->node_indices[0];
      blobfs_->GetAllocator()->MarkInodeAllocated(node);
      blobfs_->PersistNode(node.index(), &operations);
    }
  
    write_info_.reset();
  
    return blobfs_->journal()
        ->WriteMetadata(operations.TakeOperations())
        .and_then([blob = fbl::RefPtr(this)]() { blob->CompleteSync(); });
  }
  
  zx_status_t Blob::WriteInternal(const void* data, size_t len, size_t* actual) {
    TRACE_DURATION("blobfs", "Blobfs::WriteInternal", "data", data, "len", len);
  
    *actual = 0;
    if (len == 0) {
      return ZX_OK;
    }
  
    const uint64_t data_start = DataStartBlock(blobfs_->Info());
    const uint32_t merkle_blocks = MerkleTreeBlocks(inode_);
    const size_t merkle_bytes = merkle_blocks * kBlobfsBlockSize;
    if (GetState() == kBlobStateDataWrite) {
      size_t to_write = fbl::min(len, inode_.blob_size - write_info_->bytes_written);
      size_t offset = write_info_->bytes_written + merkle_bytes;
      zx_status_t status = mapping_.vmo().write(data, offset, to_write);
      if (status != ZX_OK) {
        return status;
      }
  
      *actual = to_write;
      write_info_->bytes_written += to_write;
  
      if (write_info_->compressor) {
        if ((status = write_info_->compressor->Update(data, to_write)) != ZX_OK) {
          return status;
        }
        ConsiderCompressionAbort();
      }
  
      // More data to write.
      if (write_info_->bytes_written < inode_.blob_size) {
        return ZX_OK;
      }
  
      auto set_error = fbl::MakeAutoCall([this]() { SetState(kBlobStateError); });
  
      // Only write data to disk once we've buffered the file into memory.
      // This gives us a chance to try compressing the blob before we write it back.
      if (write_info_->compressor) {
        if ((status = write_info_->compressor->End()) != ZX_OK) {
          return status;
        }
        ConsiderCompressionAbort();
      }
  
      // Since the merkle tree and data are co-allocated, use a block iterator
      // to parse their data in order.
      VectorExtentIterator extent_iter(write_info_->extents);
      BlockIterator block_iter(&extent_iter);
  
      fs::Duration generation_time;
      std::vector<fit::promise<void, zx_status_t>> promises;
      fs::DataStreamer streamer(blobfs_->journal(), blobfs_->WritebackCapacity());
  
      MerkleTreeCreator mtc;
      if ((status = mtc.SetDataLength(inode_.blob_size)) != ZX_OK) {
        return status;
      }
      size_t merkle_size = mtc.GetTreeLength();
      if (merkle_size > 0) {
        // Tracking generation time.
        fs::Ticker ticker(blobfs_->Metrics().Collecting());
  
        // TODO(smklein): As an optimization, use the Append method to create the merkle tree as we
        // write data, rather than waiting until the data is fully downloaded to create the tree.
        uint8_t root[digest::kSha256Length];
        if ((status = mtc.SetTree(GetMerkle(), merkle_size, root, sizeof(root))) != ZX_OK ||
            (status = mtc.Append(GetData(), inode_.blob_size)) != ZX_OK) {
          return status;
        }
  
        Digest expected(GetKey());
        Digest actual(root);
        if (expected != actual) {
          // Downloaded blob did not match provided digest.
          return ZX_ERR_IO_DATA_INTEGRITY;
        }
  
        status = StreamBlocks(&block_iter, merkle_blocks,
                              [&](uint64_t vmo_offset, uint64_t dev_offset, uint32_t length) {
                                storage::UnbufferedOperation op = {
                                    .vmo = zx::unowned_vmo(mapping_.vmo().get()),
                                    {
                                        .type = storage::OperationType::kWrite,
                                        .vmo_offset = vmo_offset,
                                        .dev_offset = dev_offset + data_start,
                                        .length = length,
                                    }};
                                streamer.StreamData(std::move(op));
                                return ZX_OK;
                              });
  
        if (status != ZX_OK) {
          return status;
        }
        generation_time = ticker.End();
      } else if ((status = Verify()) != ZX_OK) {
        // Small blobs may not have associated Merkle Trees, and will
        // require validation, since we are not regenerating and checking
        // the digest.
        return status;
      }
  
      if (write_info_->compressor) {
        uint64_t blocks64 =
            fbl::round_up(write_info_->compressor->Size(), kBlobfsBlockSize) / kBlobfsBlockSize;
        ZX_DEBUG_ASSERT(blocks64 <= std::numeric_limits<uint32_t>::max());
        uint32_t blocks = static_cast<uint32_t>(blocks64);
        int64_t vmo_bias = -static_cast<int64_t>(merkle_blocks);
        ZX_DEBUG_ASSERT(block_iter.BlockIndex() + vmo_bias == 0);
        status = StreamBlocks(&block_iter, blocks,
                              [&](uint64_t vmo_offset, uint64_t dev_offset, uint32_t length) {
                                storage::UnbufferedOperation op = {
                                    .vmo = zx::unowned_vmo(write_info_->compressor->Vmo().get()),
                                    {
                                        .type = storage::OperationType::kWrite,
                                        .vmo_offset = vmo_offset - merkle_blocks,
                                        .dev_offset = dev_offset + data_start,
                                        .length = length,
                                    }};
                                streamer.StreamData(std::move(op));
                                return ZX_OK;
                              });
  
        if (status != ZX_OK) {
          return status;
        }
        blocks += MerkleTreeBlocks(inode_);
        // By compressing, we used less blocks than we originally reserved.
        ZX_DEBUG_ASSERT(inode_.block_count > blocks);
  
        inode_.block_count = blocks;
        inode_.header.flags |= kBlobFlagZSTDCompressed;
      } else {
        uint64_t blocks64 = fbl::round_up(inode_.blob_size, kBlobfsBlockSize) / kBlobfsBlockSize;
        ZX_DEBUG_ASSERT(blocks64 <= std::numeric_limits<uint32_t>::max());
        uint32_t blocks = static_cast<uint32_t>(blocks64);
        status = StreamBlocks(
            &block_iter, blocks, [&](uint64_t vmo_offset, uint64_t dev_offset, uint32_t length) {
              storage::UnbufferedOperation op = {.vmo = zx::unowned_vmo(mapping_.vmo().get()),
                                                 {
                                                     .type = storage::OperationType::kWrite,
                                                     .vmo_offset = vmo_offset,
                                                     .dev_offset = dev_offset + data_start,
                                                     .length = length,
                                                 }};
              streamer.StreamData(std::move(op));
              return ZX_OK;
            });
        if (status != ZX_OK) {
          return status;
        }
      }
  
      // Enqueue the blob's final data work. Metadata must be enqueued separately.
      fs::Journal::Promise write_all_data = streamer.Flush();
  
      // No more data to write. Flush to disk.
      fs::Ticker ticker(blobfs_->Metrics().Collecting());  // Tracking enqueue time.
  
      // Wrap all pending writes with a strong reference to this Blob, so that it stays
      // alive while there are writes in progress acting on it.
      auto task = fs::wrap_reference(write_all_data.and_then(WriteMetadata()), fbl::RefPtr(this));
      blobfs_->journal()->schedule_task(std::move(task));
      blobfs_->Metrics().UpdateClientWrite(to_write, merkle_size, ticker.End(), generation_time);
      set_error.cancel();
      return ZX_OK;
    }
    return ZX_ERR_BAD_STATE;
  }
  
  void Blob::ConsiderCompressionAbort() {
    ZX_DEBUG_ASSERT(write_info_->compressor);
    if (inode_.blob_size - kCompressionMinBytesSaved < write_info_->compressor->Size()) {
      write_info_->compressor.reset();
    }
  }
  
  zx_status_t Blob::GetReadableEvent(zx::event* out) {
    TRACE_DURATION("blobfs", "Blobfs::GetReadableEvent");
    zx_status_t status;
    // This is the first 'wait until read event' request received.
    if (!readable_event_.is_valid()) {
      status = zx::event::create(0, &readable_event_);
      if (status != ZX_OK) {
        return status;
      } else if (GetState() == kBlobStateReadable) {
        readable_event_.signal(0u, ZX_USER_SIGNAL_0);
      }
    }
    zx::event out_event;
    status = readable_event_.duplicate(ZX_RIGHTS_BASIC, &out_event);
    if (status != ZX_OK) {
      return status;
    }
  
    *out = std::move(out_event);
    return ZX_OK;
  }
  
  zx_status_t Blob::CloneVmo(zx_rights_t rights, zx::vmo* out_vmo, size_t* out_size) {
    TRACE_DURATION("blobfs", "Blobfs::CloneVmo", "rights", rights);
    if (GetState() != kBlobStateReadable) {
      return ZX_ERR_BAD_STATE;
    }
    if (inode_.blob_size == 0) {
      return ZX_ERR_BAD_STATE;
    }
    zx_status_t status = InitVmos();
    if (status != ZX_OK) {
      return status;
    }
  
    const size_t merkle_bytes = MerkleTreeBlocks(inode_) * kBlobfsBlockSize;
    zx::vmo clone;
  
    zx_info_vmo_t info;
    status = mapping_.vmo().get_info(ZX_INFO_VMO, &info, sizeof(info), nullptr, nullptr);
    if (status != ZX_OK) {
      return status;
    }
    if (info.flags & ZX_INFO_VMO_PAGER_BACKED) {
      status = mapping_.vmo().create_child(ZX_VMO_CHILD_PRIVATE_PAGER_COPY, merkle_bytes,
                                           inode_.blob_size, &clone);
    } else {
      status = mapping_.vmo().create_child(ZX_VMO_CHILD_COPY_ON_WRITE, merkle_bytes,
                                           inode_.blob_size, &clone);
    }
    if (status != ZX_OK) {
      FS_TRACE_ERROR("blobfs: Failed to create child VMO: %s\n", zx_status_get_string(status));
      return status;
    }
  
    // Only add exec right to VMO if explictly requested.  (Saves a syscall if
    // we're just going to drop the right back again in replace() call below.)
    if (rights & ZX_RIGHT_EXECUTE) {
      if ((status = clone.replace_as_executable(zx::handle(), &clone)) != ZX_OK) {
        return status;
      }
    }
  
    // Narrow rights to those requested.
    if ((status = clone.replace(rights, &clone)) != ZX_OK) {
      return status;
    }
    *out_vmo = std::move(clone);
    *out_size = inode_.blob_size;
  
    if (clone_watcher_.object() == ZX_HANDLE_INVALID) {
      clone_watcher_.set_object(mapping_.vmo().get());
      clone_watcher_.set_trigger(ZX_VMO_ZERO_CHILDREN);
  
      // Keep a reference to "this" alive, preventing the blob
      // from being closed while someone may still be using the
      // underlying memory.
      //
      // We'll release it when no client-held VMOs are in use.
      clone_ref_ = fbl::RefPtr<Blob>(this);
      clone_watcher_.Begin(blobfs_->dispatcher());
    }
  
    return ZX_OK;
  }
  
  void Blob::HandleNoClones(async_dispatcher_t* dispatcher, async::WaitBase* wait, zx_status_t status,
                            const zx_packet_signal_t* signal) {
    ZX_DEBUG_ASSERT(status == ZX_OK);
    ZX_DEBUG_ASSERT((signal->observed & ZX_VMO_ZERO_CHILDREN) != 0);
    ZX_DEBUG_ASSERT(clone_watcher_.object() != ZX_HANDLE_INVALID);
    clone_watcher_.set_object(ZX_HANDLE_INVALID);
    clone_ref_ = nullptr;
  }
  
  zx_status_t Blob::ReadInternal(void* data, size_t len, size_t off, size_t* actual) {
    TRACE_DURATION("blobfs", "Blobfs::ReadInternal", "len", len, "off", off);
  
    if (GetState() != kBlobStateReadable) {
      return ZX_ERR_BAD_STATE;
    }
  
    if (inode_.blob_size == 0) {
      *actual = 0;
      return ZX_OK;
    }
  
    zx_status_t status = InitVmos();
    if (status != ZX_OK) {
      return status;
    }
  
    Digest d(GetKey());
  
    if (off >= inode_.blob_size) {
      *actual = 0;
      return ZX_OK;
    }
    if (len > (inode_.blob_size - off)) {
      len = inode_.blob_size - off;
    }
  
    const size_t merkle_bytes = MerkleTreeBlocks(inode_) * kBlobfsBlockSize;
    status = mapping_.vmo().read(data, merkle_bytes + off, len);
    if (status == ZX_OK) {
      *actual = len;
    }
    return status;
  }
  
  zx_status_t Blob::QueueUnlink() {
    flags_ |= kBlobFlagDeletable;
    // Attempt to purge in case the blob has been unlinked with no open fds
    return TryPurge();
  }
  
  zx_status_t Blob::VerifyBlob(Blobfs* bs, uint32_t node_index) {
    Inode* inode = bs->GetNode(node_index);
    Digest digest(inode->merkle_root_hash);
    fbl::RefPtr<Blob> vn = fbl::AdoptRef(new Blob(bs, digest));
  
    vn->PopulateInode(node_index);
  
    // If we are unable to read in the blob from disk, this should also be a VerifyBlob error.
    // Since InitVmos calls Verify as its final step, we can just return its result here.
    return vn->InitVmos();
  }
  
  BlobCache& Blob::Cache() { return blobfs_->Cache(); }
  
  bool Blob::ShouldCache() const {
    switch (GetState()) {
      // All "Valid", cacheable states, where the blob still exists on storage.
      case kBlobStateReadable:
        return true;
      default:
        return false;
    }
  }
  
  void Blob::ActivateLowMemory() {
    // We shouldn't be putting the blob into a low-memory state while it is still mapped.
    ZX_ASSERT(clone_watcher_.object() == ZX_HANDLE_INVALID);
    page_watcher_.reset();
    if (mapping_.vmo()) {
      blobfs_->DetachVmo(vmoid_);
    }
    mapping_.Reset();
  }
  
  Blob::~Blob() { ActivateLowMemory(); }
  
  fs::VnodeProtocolSet Blob::GetProtocols() const { return fs::VnodeProtocol::kFile; }
  
  bool Blob::ValidateRights(fs::Rights rights) {
    // To acquire write access to a blob, it must be empty.
    return !rights.write || (GetState() == kBlobStateEmpty);
  }
  
  zx_status_t Blob::GetNodeInfoForProtocol([[maybe_unused]] fs::VnodeProtocol protocol,
                                           [[maybe_unused]] fs::Rights rights,
                                           fs::VnodeRepresentation* info) {
    zx::event observer;
    zx_status_t status = GetReadableEvent(&observer);
    if (status != ZX_OK) {
      return status;
    }
    *info = fs::VnodeRepresentation::File{.observer = std::move(observer)};
    return ZX_OK;
  }
  
  zx_status_t Blob::Read(void* data, size_t len, size_t off, size_t* out_actual) {
    TRACE_DURATION("blobfs", "Blob::Read", "len", len, "off", off);
    auto event = blobfs_->Metrics().NewLatencyEvent(fs_metrics::Event::kRead);
  
    return ReadInternal(data, len, off, out_actual);
  }
  
  zx_status_t Blob::Write(const void* data, size_t len, size_t offset, size_t* out_actual) {
    TRACE_DURATION("blobfs", "Blob::Write", "len", len, "off", offset);
    auto event = blobfs_->Metrics().NewLatencyEvent(fs_metrics::Event::kWrite);
    return WriteInternal(data, len, out_actual);
  }
  
  zx_status_t Blob::Append(const void* data, size_t len, size_t* out_end, size_t* out_actual) {
    auto event = blobfs_->Metrics().NewLatencyEvent(fs_metrics::Event::kAppend);
    zx_status_t status = WriteInternal(data, len, out_actual);
    if (GetState() == kBlobStateDataWrite) {
      ZX_DEBUG_ASSERT(write_info_ != nullptr);
      *out_actual = write_info_->bytes_written;
    } else {
      *out_actual = inode_.blob_size;
    }
    return status;
  }
  
  zx_status_t Blob::GetAttributes(fs::VnodeAttributes* a) {
    auto event = blobfs_->Metrics().NewLatencyEvent(fs_metrics::Event::kGetAttr);
    *a = fs::VnodeAttributes();
    a->mode = V_TYPE_FILE | V_IRUSR;
    a->inode = Ino();
    a->content_size = SizeData();
    a->storage_size = inode_.block_count * kBlobfsBlockSize;
    a->link_count = 1;
    a->creation_time = 0;
    a->modification_time = 0;
    return ZX_OK;
  }
  
  zx_status_t Blob::Truncate(size_t len) {
    TRACE_DURATION("blobfs", "Blob::Truncate", "len", len);
    auto event = blobfs_->Metrics().NewLatencyEvent(fs_metrics::Event::kTruncate);
    return SpaceAllocate(len);
  }
  
  #ifdef __Fuchsia__
  
  constexpr const char kFsName[] = "blobfs";
  
  zx_status_t Blob::QueryFilesystem(::llcpp::fuchsia::io::FilesystemInfo* info) {
    static_assert(fbl::constexpr_strlen(kFsName) + 1 < ::llcpp::fuchsia::io::MAX_FS_NAME_BUFFER,
                  "Blobfs name too long");
  
    *info = {};
    info->block_size = kBlobfsBlockSize;
    info->max_filename_size = digest::kSha256HexLength;
    info->fs_type = VFS_TYPE_BLOBFS;
    info->fs_id = blobfs_->GetFsId();
    info->total_bytes = blobfs_->Info().data_block_count * blobfs_->Info().block_size;
    info->used_bytes = blobfs_->Info().alloc_block_count * blobfs_->Info().block_size;
    info->total_nodes = blobfs_->Info().inode_count;
    info->used_nodes = blobfs_->Info().alloc_inode_count;
    strlcpy(reinterpret_cast<char*>(info->name.data()), kFsName,
            ::llcpp::fuchsia::io::MAX_FS_NAME_BUFFER);
    return ZX_OK;
  }
  
  zx_status_t Blob::GetDevicePath(size_t buffer_len, char* out_name, size_t* out_len) {
    return blobfs_->Device()->GetDevicePath(buffer_len, out_name, out_len);
  }
  
  zx_status_t Blob::GetVmo(int flags, zx::vmo* out_vmo, size_t* out_size) {
    TRACE_DURATION("blobfs", "Blob::GetVmo", "flags", flags);
  
    if (flags & ::llcpp::fuchsia::io::VMO_FLAG_WRITE) {
      return ZX_ERR_NOT_SUPPORTED;
    } else if (flags & ::llcpp::fuchsia::io::VMO_FLAG_EXACT) {
      return ZX_ERR_NOT_SUPPORTED;
    }
  
    // Let clients map and set the names of their VMOs.
    zx_rights_t rights = ZX_RIGHTS_BASIC | ZX_RIGHT_MAP | ZX_RIGHTS_PROPERTY;
    // We can ignore fuchsia_io_VMO_FLAG_PRIVATE, since private / shared access
    // to the underlying VMO can both be satisfied with a clone due to
    // the immutability of blobfs blobs.
    rights |= (flags & ::llcpp::fuchsia::io::VMO_FLAG_READ) ? ZX_RIGHT_READ : 0;
    rights |= (flags & ::llcpp::fuchsia::io::VMO_FLAG_EXEC) ? ZX_RIGHT_EXECUTE : 0;
    return CloneVmo(rights, out_vmo, out_size);
  }
  
  #endif
  
  void Blob::Sync(SyncCallback closure) {
    auto event = blobfs_->Metrics().NewLatencyEvent(fs_metrics::Event::kSync);
    if (atomic_load(&syncing_)) {
      blobfs_->Sync(
          [this, evt = std::move(event), cb = std::move(closure)](zx_status_t status) mutable {
            if (status != ZX_OK) {
              cb(status);
              return;
            }
  
            fs::WriteTxn sync_txn(blobfs_);
            sync_txn.EnqueueFlush();
            status = sync_txn.Transact();
            cb(status);
          });
    } else {
      closure(ZX_OK);
    }
  }
  
  void Blob::CompleteSync() {
    atomic_store(&syncing_, false);
    // Drop the write info, since we no longer need it.
    write_info_.reset();
  }
  
  fbl::RefPtr<Blob> Blob::CloneWatcherTeardown() {
    if (clone_watcher_.is_pending()) {
      clone_watcher_.Cancel();
      clone_watcher_.set_object(ZX_HANDLE_INVALID);
      return std::move(clone_ref_);
    }
    return nullptr;
  }
  
  zx_status_t Blob::Open([[maybe_unused]] ValidatedOptions options,
                         fbl::RefPtr<Vnode>* out_redirect) {
    fd_count_++;
    return ZX_OK;
  }
  
  zx_status_t Blob::Close() {
    auto event = blobfs_->Metrics().NewLatencyEvent(fs_metrics::Event::kClose);
    ZX_DEBUG_ASSERT_MSG(fd_count_ > 0, "Closing blob with no fds open");
    fd_count_--;
    // Attempt purge in case blob was unlinked prior to close
    return TryPurge();
  }
  
  zx_status_t Blob::TryPurge() {
    if (Purgeable()) {
      return Purge();
    }
    return ZX_OK;
  }
  
  zx_status_t Blob::Purge() {
    ZX_DEBUG_ASSERT(fd_count_ == 0);
    ZX_DEBUG_ASSERT(Purgeable());
  
    if (GetState() == kBlobStateReadable) {
      // A readable blob should only be purged if it has been unlinked.
      ZX_ASSERT(DeletionQueued());
      storage::UnbufferedOperationsBuilder operations;
      fbl::Vector<storage::BufferedOperation> trim_data;
      blobfs_->FreeInode(GetMapIndex(), &operations, &trim_data);
  
      auto task = fs::wrap_reference(blobfs_->journal()->WriteMetadata(operations.TakeOperations()),
                                     fbl::RefPtr(this))
                      .and_then(blobfs_->journal()->TrimData(std::move(trim_data)));
      blobfs_->journal()->schedule_task(std::move(task));
    }
    ZX_ASSERT(Cache().Evict(fbl::RefPtr(this)) == ZX_OK);
    SetState(kBlobStatePurged);
    return ZX_OK;
  }
  
  }  // namespace blobfs
  