diff --git a/payload_generator/ab_generator.cc b/payload_generator/ab_generator.cc
index f4cc9fb..d9b9d88 100644
--- a/payload_generator/ab_generator.cc
+++ b/payload_generator/ab_generator.cc
@@ -276,7 +276,7 @@
       target_part_path, dst_extents, &data, data.size(), kBlockSize));
 
   brillo::Blob blob;
-  InstallOperation_Type op_type;
+  InstallOperation::Type op_type;
   TEST_AND_RETURN_FALSE(
       diff_utils::GenerateBestFullOperation(data, version, &blob, &op_type));
 
diff --git a/payload_generator/ab_generator_unittest.cc b/payload_generator/ab_generator_unittest.cc
index 170e0e3..7a95284 100644
--- a/payload_generator/ab_generator_unittest.cc
+++ b/payload_generator/ab_generator_unittest.cc
@@ -49,7 +49,7 @@
 }
 
 // Tests splitting of a REPLACE/REPLACE_XZ operation.
-void TestSplitReplaceOrReplaceXzOperation(InstallOperation_Type orig_type,
+void TestSplitReplaceOrReplaceXzOperation(InstallOperation::Type orig_type,
                                           bool compressible) {
   const size_t op_ex1_start_block = 2;
   const size_t op_ex1_num_blocks = 2;
@@ -124,7 +124,7 @@
       version, aop, part_file.path(), &result_ops, &blob_file));
 
   // Check the result.
-  InstallOperation_Type expected_type =
+  InstallOperation::Type expected_type =
       compressible ? InstallOperation::REPLACE_XZ : InstallOperation::REPLACE;
 
   ASSERT_EQ(2U, result_ops.size());
@@ -200,7 +200,7 @@
 }
 
 // Tests merging of REPLACE/REPLACE_XZ operations.
-void TestMergeReplaceOrReplaceXzOperations(InstallOperation_Type orig_type,
+void TestMergeReplaceOrReplaceXzOperations(InstallOperation::Type orig_type,
                                            bool compressible) {
   const size_t first_op_num_blocks = 1;
   const size_t second_op_num_blocks = 2;
@@ -287,7 +287,7 @@
       &aops, version, 5, part_file.path(), &blob_file));
 
   // Check the result.
-  InstallOperation_Type expected_op_type =
+  InstallOperation::Type expected_op_type =
       compressible ? InstallOperation::REPLACE_XZ : InstallOperation::REPLACE;
   EXPECT_EQ(1U, aops.size());
   InstallOperation new_op = aops[0].op;
diff --git a/payload_generator/blob_file_writer.cc b/payload_generator/blob_file_writer.cc
index 7cdeb35..a1afe87 100644
--- a/payload_generator/blob_file_writer.cc
+++ b/payload_generator/blob_file_writer.cc
@@ -38,9 +38,9 @@
   return result;
 }
 
-void BlobFileWriter::SetTotalBlobs(size_t total_blobs) {
-  total_blobs_ = total_blobs;
-  stored_blobs_ = 0;
+void BlobFileWriter::IncTotalBlobs(size_t increment) {
+  base::AutoLock auto_lock(blob_mutex_);
+  total_blobs_ += increment;
 }
 
 }  // namespace chromeos_update_engine
diff --git a/payload_generator/blob_file_writer.h b/payload_generator/blob_file_writer.h
index 48553be..bdd4c08 100644
--- a/payload_generator/blob_file_writer.h
+++ b/payload_generator/blob_file_writer.h
@@ -35,10 +35,8 @@
   // was stored, or -1 in case of failure.
   off_t StoreBlob(const brillo::Blob& blob);
 
-  // The number of |total_blobs| is the number of blobs that will be stored but
-  // is only used for logging purposes. If not set or set to 0, logging will be
-  // skipped. This function will also reset the number of stored blobs to 0.
-  void SetTotalBlobs(size_t total_blobs);
+  // Increase |total_blobs| by |increment|. Thread safe.
+  void IncTotalBlobs(size_t increment);
 
  private:
   size_t total_blobs_{0};
diff --git a/payload_generator/boot_img_filesystem.cc b/payload_generator/boot_img_filesystem.cc
index 19de410..89b175e 100644
--- a/payload_generator/boot_img_filesystem.cc
+++ b/payload_generator/boot_img_filesystem.cc
@@ -17,6 +17,7 @@
 #include "update_engine/payload_generator/boot_img_filesystem.h"
 
 #include <base/logging.h>
+#include <bootimg.h>
 #include <brillo/secure_blob.h>
 #include <puffin/utils.h>
 
@@ -35,16 +36,61 @@
   if (filename.empty())
     return nullptr;
 
-  brillo::Blob header;
-  if (!utils::ReadFileChunk(filename, 0, sizeof(boot_img_hdr), &header) ||
-      header.size() != sizeof(boot_img_hdr) ||
-      memcmp(header.data(), BOOT_MAGIC, BOOT_MAGIC_SIZE) != 0) {
+  if (brillo::Blob header_magic;
+      !utils::ReadFileChunk(filename, 0, BOOT_MAGIC_SIZE, &header_magic) ||
+      memcmp(header_magic.data(), BOOT_MAGIC, BOOT_MAGIC_SIZE) != 0) {
+    return nullptr;
+  }
+
+  // The order of image header fields are different in version 3 from the
+  // previous versions. But the position of "header_version" is fixed at #9
+  // across all image headers.
+  // See details in system/tools/mkbootimg/include/bootimg/bootimg.h
+  constexpr size_t header_version_offset =
+      BOOT_MAGIC_SIZE + 8 * sizeof(uint32_t);
+  brillo::Blob header_version_blob;
+  if (!utils::ReadFileChunk(filename,
+                            header_version_offset,
+                            sizeof(uint32_t),
+                            &header_version_blob)) {
+    return nullptr;
+  }
+  uint32_t header_version =
+      *reinterpret_cast<uint32_t*>(header_version_blob.data());
+  if (header_version > 3) {
+    LOG(WARNING) << "Boot image header version " << header_version
+                 << " isn't supported for parsing";
+    return nullptr;
+  }
+
+  // Read the bytes of boot image header based on the header version.
+  size_t header_size =
+      header_version == 3 ? sizeof(boot_img_hdr_v3) : sizeof(boot_img_hdr_v0);
+  brillo::Blob header_blob;
+  if (!utils::ReadFileChunk(filename, 0, header_size, &header_blob)) {
     return nullptr;
   }
 
   unique_ptr<BootImgFilesystem> result(new BootImgFilesystem());
   result->filename_ = filename;
-  memcpy(&result->hdr_, header.data(), header.size());
+  if (header_version < 3) {
+    auto hdr_v0 = reinterpret_cast<boot_img_hdr_v0*>(header_blob.data());
+    CHECK_EQ(0, memcmp(hdr_v0->magic, BOOT_MAGIC, BOOT_MAGIC_SIZE));
+    CHECK_LT(hdr_v0->header_version, 3u);
+    result->kernel_size_ = hdr_v0->kernel_size;
+    result->ramdisk_size_ = hdr_v0->ramdisk_size;
+    result->page_size_ = hdr_v0->page_size;
+  } else {
+    auto hdr_v3 = reinterpret_cast<boot_img_hdr_v3*>(header_blob.data());
+    CHECK_EQ(0, memcmp(hdr_v3->magic, BOOT_MAGIC, BOOT_MAGIC_SIZE));
+    CHECK_EQ(3u, hdr_v3->header_version);
+    result->kernel_size_ = hdr_v3->kernel_size;
+    result->ramdisk_size_ = hdr_v3->ramdisk_size;
+    result->page_size_ = 4096;
+  }
+
+  CHECK_GT(result->page_size_, 0u);
+
   return result;
 }
 
@@ -87,13 +133,13 @@
   files->clear();
   const uint64_t file_size = utils::FileSize(filename_);
   // The first page is header.
-  uint64_t offset = hdr_.page_size;
-  if (hdr_.kernel_size > 0 && offset + hdr_.kernel_size <= file_size) {
-    files->emplace_back(GetFile("<kernel>", offset, hdr_.kernel_size));
+  uint64_t offset = page_size_;
+  if (kernel_size_ > 0 && offset + kernel_size_ <= file_size) {
+    files->emplace_back(GetFile("<kernel>", offset, kernel_size_));
   }
-  offset += utils::RoundUp(hdr_.kernel_size, hdr_.page_size);
-  if (hdr_.ramdisk_size > 0 && offset + hdr_.ramdisk_size <= file_size) {
-    files->emplace_back(GetFile("<ramdisk>", offset, hdr_.ramdisk_size));
+  offset += utils::RoundUp(kernel_size_, page_size_);
+  if (ramdisk_size_ > 0 && offset + ramdisk_size_ <= file_size) {
+    files->emplace_back(GetFile("<ramdisk>", offset, ramdisk_size_));
   }
   return true;
 }
diff --git a/payload_generator/boot_img_filesystem.h b/payload_generator/boot_img_filesystem.h
index 87725d4..61f755c 100644
--- a/payload_generator/boot_img_filesystem.h
+++ b/payload_generator/boot_img_filesystem.h
@@ -52,23 +52,9 @@
   // The boot.img file path.
   std::string filename_;
 
-// https://android.googlesource.com/platform/system/core/+/master/mkbootimg/include/bootimg/bootimg.h
-#define BOOT_MAGIC "ANDROID!"
-#define BOOT_MAGIC_SIZE 8
-  struct boot_img_hdr {
-    // Must be BOOT_MAGIC.
-    uint8_t magic[BOOT_MAGIC_SIZE];
-    uint32_t kernel_size;  /* size in bytes */
-    uint32_t kernel_addr;  /* physical load addr */
-    uint32_t ramdisk_size; /* size in bytes */
-    uint32_t ramdisk_addr; /* physical load addr */
-    uint32_t second_size;  /* size in bytes */
-    uint32_t second_addr;  /* physical load addr */
-    uint32_t tags_addr;    /* physical addr for kernel tags */
-    uint32_t page_size;    /* flash page size we assume */
-  } __attribute__((packed));
-  // The boot image header.
-  boot_img_hdr hdr_;
+  uint32_t kernel_size_;  /* size in bytes */
+  uint32_t ramdisk_size_; /* size in bytes */
+  uint32_t page_size_;    /* flash page size we assume */
 
   DISALLOW_COPY_AND_ASSIGN(BootImgFilesystem);
 };
diff --git a/payload_generator/boot_img_filesystem_stub.cc b/payload_generator/boot_img_filesystem_stub.cc
new file mode 100644
index 0000000..4928fa1
--- /dev/null
+++ b/payload_generator/boot_img_filesystem_stub.cc
@@ -0,0 +1,48 @@
+//
+// Copyright (C) 2020 The Android Open Source Project
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//      http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+//
+
+#include "update_engine/payload_generator/boot_img_filesystem.h"
+
+namespace chromeos_update_engine {
+std::unique_ptr<BootImgFilesystem> BootImgFilesystem::CreateFromFile(
+    const std::string& /* filename */) {
+  return nullptr;
+}
+
+size_t BootImgFilesystem::GetBlockSize() const {
+  return 4096;
+}
+
+size_t BootImgFilesystem::GetBlockCount() const {
+  return 0;
+}
+
+FilesystemInterface::File BootImgFilesystem::GetFile(
+    const std::string& /* name */,
+    uint64_t /* offset */,
+    uint64_t /* size */) const {
+  return {};
+}
+
+bool BootImgFilesystem::GetFiles(std::vector<File>* /* files */) const {
+  return false;
+}
+
+bool BootImgFilesystem::LoadSettings(brillo::KeyValueStore* /* store */) const {
+  return false;
+}
+
+}  // namespace chromeos_update_engine
diff --git a/payload_generator/boot_img_filesystem_unittest.cc b/payload_generator/boot_img_filesystem_unittest.cc
index b1e0d99..0b115e0 100644
--- a/payload_generator/boot_img_filesystem_unittest.cc
+++ b/payload_generator/boot_img_filesystem_unittest.cc
@@ -18,6 +18,7 @@
 
 #include <vector>
 
+#include <bootimg.h>
 #include <brillo/secure_blob.h>
 #include <gtest/gtest.h>
 
@@ -32,18 +33,32 @@
 class BootImgFilesystemTest : public ::testing::Test {
  protected:
   brillo::Blob GetBootImg(const brillo::Blob& kernel,
-                          const brillo::Blob& ramdisk) {
+                          const brillo::Blob& ramdisk,
+                          bool header_version3 = false) {
     brillo::Blob boot_img(16 * 1024);
-    BootImgFilesystem::boot_img_hdr hdr;
-    memcpy(hdr.magic, BOOT_MAGIC, BOOT_MAGIC_SIZE);
-    hdr.kernel_size = kernel.size();
-    hdr.ramdisk_size = ramdisk.size();
-    hdr.page_size = 4096;
+    constexpr uint32_t page_size = 4096;
+
     size_t offset = 0;
-    memcpy(boot_img.data() + offset, &hdr, sizeof(hdr));
-    offset += utils::RoundUp(sizeof(hdr), hdr.page_size);
+    if (header_version3) {
+      boot_img_hdr_v3 hdr_v3{};
+      memcpy(hdr_v3.magic, BOOT_MAGIC, BOOT_MAGIC_SIZE);
+      hdr_v3.kernel_size = kernel.size();
+      hdr_v3.ramdisk_size = ramdisk.size();
+      hdr_v3.header_version = 3;
+      memcpy(boot_img.data() + offset, &hdr_v3, sizeof(hdr_v3));
+      offset += utils::RoundUp(sizeof(hdr_v3), page_size);
+    } else {
+      boot_img_hdr_v0 hdr_v0{};
+      memcpy(hdr_v0.magic, BOOT_MAGIC, BOOT_MAGIC_SIZE);
+      hdr_v0.kernel_size = kernel.size();
+      hdr_v0.ramdisk_size = ramdisk.size();
+      hdr_v0.page_size = page_size;
+      hdr_v0.header_version = 0;
+      memcpy(boot_img.data() + offset, &hdr_v0, sizeof(hdr_v0));
+      offset += utils::RoundUp(sizeof(hdr_v0), page_size);
+    }
     memcpy(boot_img.data() + offset, kernel.data(), kernel.size());
-    offset += utils::RoundUp(kernel.size(), hdr.page_size);
+    offset += utils::RoundUp(kernel.size(), page_size);
     memcpy(boot_img.data() + offset, ramdisk.data(), ramdisk.size());
     return boot_img;
   }
@@ -76,6 +91,31 @@
   EXPECT_TRUE(files[1].deflates.empty());
 }
 
+TEST_F(BootImgFilesystemTest, ImageHeaderVersion3) {
+  test_utils::WriteFileVector(
+      boot_file_.path(),
+      GetBootImg(brillo::Blob(1000, 'k'), brillo::Blob(5000, 'r'), true));
+  unique_ptr<BootImgFilesystem> fs =
+      BootImgFilesystem::CreateFromFile(boot_file_.path());
+  EXPECT_NE(nullptr, fs);
+
+  vector<FilesystemInterface::File> files;
+  EXPECT_TRUE(fs->GetFiles(&files));
+  ASSERT_EQ(2u, files.size());
+
+  EXPECT_EQ("<kernel>", files[0].name);
+  EXPECT_EQ(1u, files[0].extents.size());
+  EXPECT_EQ(1u, files[0].extents[0].start_block());
+  EXPECT_EQ(1u, files[0].extents[0].num_blocks());
+  EXPECT_TRUE(files[0].deflates.empty());
+
+  EXPECT_EQ("<ramdisk>", files[1].name);
+  EXPECT_EQ(1u, files[1].extents.size());
+  EXPECT_EQ(2u, files[1].extents[0].start_block());
+  EXPECT_EQ(2u, files[1].extents[0].num_blocks());
+  EXPECT_TRUE(files[1].deflates.empty());
+}
+
 TEST_F(BootImgFilesystemTest, BadImageTest) {
   brillo::Blob boot_img = GetBootImg({}, {});
   boot_img[7] = '?';
diff --git a/payload_generator/deflate_utils.cc b/payload_generator/deflate_utils.cc
index 5d7a766..c874bfd 100644
--- a/payload_generator/deflate_utils.cc
+++ b/payload_generator/deflate_utils.cc
@@ -74,6 +74,15 @@
   return false;
 }
 
+bool IsRegularFile(const FilesystemInterface::File& file) {
+  // If inode is 0, then stat information is invalid for some psuedo files
+  if (file.file_stat.st_ino != 0 &&
+      (file.file_stat.st_mode & S_IFMT) == S_IFREG) {
+    return true;
+  }
+  return false;
+}
+
 // Realigns subfiles |files| of a splitted file |file| into its correct
 // positions. This can be used for squashfs, zip, apk, etc.
 bool RealignSplittedFiles(const FilesystemInterface::File& file,
@@ -265,7 +274,9 @@
   result_files->reserve(tmp_files.size());
 
   for (auto& file : tmp_files) {
-    if (IsSquashfsImage(part.path, file)) {
+    auto is_regular_file = IsRegularFile(file);
+
+    if (is_regular_file && IsSquashfsImage(part.path, file)) {
       // Read the image into a file.
       base::FilePath path;
       TEST_AND_RETURN_FALSE(base::CreateTemporaryFile(&path));
@@ -296,7 +307,7 @@
       }
     }
 
-    if (extract_deflates && !file.is_compressed) {
+    if (is_regular_file && extract_deflates && !file.is_compressed) {
       // Search for deflates if the file is in zip or gzip format.
       // .zvoice files may eventually move out of rootfs. If that happens,
       // remove ".zvoice" (crbug.com/782918).
diff --git a/payload_generator/delta_diff_generator.cc b/payload_generator/delta_diff_generator.cc
index 595a41e..c2b35ee 100644
--- a/payload_generator/delta_diff_generator.cc
+++ b/payload_generator/delta_diff_generator.cc
@@ -29,14 +29,17 @@
 #include <vector>
 
 #include <base/logging.h>
+#include <base/threading/simple_thread.h>
 
 #include "update_engine/common/utils.h"
 #include "update_engine/payload_consumer/delta_performer.h"
 #include "update_engine/payload_consumer/payload_constants.h"
 #include "update_engine/payload_generator/ab_generator.h"
+#include "update_engine/payload_generator/annotated_operation.h"
 #include "update_engine/payload_generator/blob_file_writer.h"
 #include "update_engine/payload_generator/delta_diff_utils.h"
 #include "update_engine/payload_generator/full_update_generator.h"
+#include "update_engine/payload_generator/merge_sequence_generator.h"
 #include "update_engine/payload_generator/payload_file.h"
 
 using std::string;
@@ -49,6 +52,59 @@
 const size_t kRootFSPartitionSize = static_cast<size_t>(2) * 1024 * 1024 * 1024;
 const size_t kBlockSize = 4096;  // bytes
 
+class PartitionProcessor : public base::DelegateSimpleThread::Delegate {
+ public:
+  explicit PartitionProcessor(
+      const PayloadGenerationConfig& config,
+      const PartitionConfig& old_part,
+      const PartitionConfig& new_part,
+      BlobFileWriter* file_writer,
+      std::vector<AnnotatedOperation>* aops,
+      std::vector<CowMergeOperation>* cow_merge_sequence,
+      std::unique_ptr<chromeos_update_engine::OperationsGenerator> strategy)
+      : config_(config),
+        old_part_(old_part),
+        new_part_(new_part),
+        file_writer_(file_writer),
+        aops_(aops),
+        cow_merge_sequence_(cow_merge_sequence),
+        strategy_(std::move(strategy)) {}
+  PartitionProcessor(PartitionProcessor&&) noexcept = default;
+  void Run() override {
+    LOG(INFO) << "Started an async task to process partition "
+              << old_part_.name;
+    bool success = strategy_->GenerateOperations(
+        config_, old_part_, new_part_, file_writer_, aops_);
+    if (!success) {
+      // ABORT the entire process, so that developer can look
+      // at recent logs and diagnose what happened
+      LOG(FATAL) << "GenerateOperations(" << old_part_.name << ", "
+                 << new_part_.name << ") failed";
+    }
+
+    bool snapshot_enabled =
+        config_.target.dynamic_partition_metadata &&
+        config_.target.dynamic_partition_metadata->snapshot_enabled();
+    if (old_part_.path.empty() || !snapshot_enabled) {
+      return;
+    }
+    auto generator = MergeSequenceGenerator::Create(*aops_);
+    if (!generator || !generator->Generate(cow_merge_sequence_)) {
+      LOG(FATAL) << "Failed to generate merge sequence";
+    }
+  }
+
+ private:
+  const PayloadGenerationConfig& config_;
+  const PartitionConfig& old_part_;
+  const PartitionConfig& new_part_;
+  BlobFileWriter* file_writer_;
+  std::vector<AnnotatedOperation>* aops_;
+  std::vector<CowMergeOperation>* cow_merge_sequence_;
+  std::unique_ptr<chromeos_update_engine::OperationsGenerator> strategy_;
+  DISALLOW_COPY_AND_ASSIGN(PartitionProcessor);
+};
+
 bool GenerateUpdatePayloadFile(const PayloadGenerationConfig& config,
                                const string& output_path,
                                const string& private_key_path,
@@ -80,6 +136,15 @@
                             config.target.partitions.size());
     }
     PartitionConfig empty_part("");
+    std::vector<std::vector<AnnotatedOperation>> all_aops;
+    all_aops.resize(config.target.partitions.size());
+    std::vector<std::vector<CowMergeOperation>> all_merge_sequences;
+    all_merge_sequences.resize(config.target.partitions.size());
+    std::vector<PartitionProcessor> partition_tasks{};
+    auto thread_count = std::min<int>(diff_utils::GetMaxThreads(),
+                                      config.target.partitions.size());
+    base::DelegateSimpleThreadPool thread_pool{"partition-thread-pool",
+                                               thread_count};
     for (size_t i = 0; i < config.target.partitions.size(); i++) {
       const PartitionConfig& old_part =
           config.is_delta ? config.source.partitions[i] : empty_part;
@@ -99,12 +164,30 @@
         strategy.reset(new FullUpdateGenerator());
       }
 
-      vector<AnnotatedOperation> aops;
       // Generate the operations using the strategy we selected above.
-      TEST_AND_RETURN_FALSE(strategy->GenerateOperations(
-          config, old_part, new_part, &blob_file, &aops));
+      partition_tasks.push_back(PartitionProcessor(config,
+                                                   old_part,
+                                                   new_part,
+                                                   &blob_file,
+                                                   &all_aops[i],
+                                                   &all_merge_sequences[i],
+                                                   std::move(strategy)));
+    }
+    thread_pool.Start();
+    for (auto& processor : partition_tasks) {
+      thread_pool.AddWork(&processor);
+    }
+    thread_pool.JoinAll();
 
-      TEST_AND_RETURN_FALSE(payload.AddPartition(old_part, new_part, aops));
+    for (size_t i = 0; i < config.target.partitions.size(); i++) {
+      const PartitionConfig& old_part =
+          config.is_delta ? config.source.partitions[i] : empty_part;
+      const PartitionConfig& new_part = config.target.partitions[i];
+      TEST_AND_RETURN_FALSE(
+          payload.AddPartition(old_part,
+                               new_part,
+                               std::move(all_aops[i]),
+                               std::move(all_merge_sequences[i])));
     }
   }
 
diff --git a/payload_generator/delta_diff_utils.cc b/payload_generator/delta_diff_utils.cc
index ded30fb..220c7ae 100644
--- a/payload_generator/delta_diff_utils.cc
+++ b/payload_generator/delta_diff_utils.cc
@@ -637,7 +637,7 @@
 bool GenerateBestFullOperation(const brillo::Blob& new_data,
                                const PayloadVersion& version,
                                brillo::Blob* out_blob,
-                               InstallOperation_Type* out_type) {
+                               InstallOperation::Type* out_type) {
   if (new_data.empty())
     return false;
 
@@ -739,7 +739,7 @@
 
   // Try generating a full operation for the given new data, regardless of the
   // old_data.
-  InstallOperation_Type op_type;
+  InstallOperation::Type op_type;
   TEST_AND_RETURN_FALSE(
       GenerateBestFullOperation(new_data, version, &data_blob, &op_type));
   operation.set_type(op_type);
@@ -766,7 +766,7 @@
         ScopedPathUnlinker unlinker(patch.value());
 
         std::unique_ptr<bsdiff::PatchWriterInterface> bsdiff_patch_writer;
-        InstallOperation_Type operation_type = InstallOperation::SOURCE_BSDIFF;
+        InstallOperation::Type operation_type = InstallOperation::SOURCE_BSDIFF;
         if (version.OperationAllowed(InstallOperation::BROTLI_BSDIFF)) {
           bsdiff_patch_writer =
               bsdiff::CreateBSDF2PatchWriter(patch.value(),
@@ -872,13 +872,13 @@
   return true;
 }
 
-bool IsAReplaceOperation(InstallOperation_Type op_type) {
+bool IsAReplaceOperation(InstallOperation::Type op_type) {
   return (op_type == InstallOperation::REPLACE ||
           op_type == InstallOperation::REPLACE_BZ ||
           op_type == InstallOperation::REPLACE_XZ);
 }
 
-bool IsNoSourceOperation(InstallOperation_Type op_type) {
+bool IsNoSourceOperation(InstallOperation::Type op_type) {
   return (IsAReplaceOperation(op_type) || op_type == InstallOperation::ZERO ||
           op_type == InstallOperation::DISCARD);
 }
@@ -938,7 +938,7 @@
   if (magic != EXT2_SUPER_MAGIC)
     return false;
 
-  // Sanity check the parameters.
+  // Validation check the parameters.
   TEST_AND_RETURN_FALSE(log_block_size >= EXT2_MIN_BLOCK_LOG_SIZE &&
                         log_block_size <= EXT2_MAX_BLOCK_LOG_SIZE);
   TEST_AND_RETURN_FALSE(block_count > 0);
diff --git a/payload_generator/delta_diff_utils.h b/payload_generator/delta_diff_utils.h
index a062327..c75d16d 100644
--- a/payload_generator/delta_diff_utils.h
+++ b/payload_generator/delta_diff_utils.h
@@ -119,13 +119,13 @@
 bool GenerateBestFullOperation(const brillo::Blob& new_data,
                                const PayloadVersion& version,
                                brillo::Blob* out_blob,
-                               InstallOperation_Type* out_type);
+                               InstallOperation::Type* out_type);
 
 // Returns whether |op_type| is one of the REPLACE full operations.
-bool IsAReplaceOperation(InstallOperation_Type op_type);
+bool IsAReplaceOperation(InstallOperation::Type op_type);
 
 // Returns true if an operation with type |op_type| has no |src_extents|.
-bool IsNoSourceOperation(InstallOperation_Type op_type);
+bool IsNoSourceOperation(InstallOperation::Type op_type);
 
 bool InitializePartitionInfo(const PartitionConfig& partition,
                              PartitionInfo* info);
diff --git a/payload_generator/delta_diff_utils_unittest.cc b/payload_generator/delta_diff_utils_unittest.cc
index 8a97b1b..0857f9c 100644
--- a/payload_generator/delta_diff_utils_unittest.cc
+++ b/payload_generator/delta_diff_utils_unittest.cc
@@ -231,7 +231,7 @@
     EXPECT_FALSE(data.empty());
 
     EXPECT_TRUE(op.has_type());
-    const InstallOperation_Type expected_type =
+    const InstallOperation::Type expected_type =
         (i == 0 ? InstallOperation::REPLACE : InstallOperation::REPLACE_BZ);
     EXPECT_EQ(expected_type, op.type());
     EXPECT_FALSE(op.has_data_offset());
diff --git a/payload_generator/extent_ranges.cc b/payload_generator/extent_ranges.cc
index 4600efe..2098639 100644
--- a/payload_generator/extent_ranges.cc
+++ b/payload_generator/extent_ranges.cc
@@ -202,6 +202,15 @@
   }
 }
 
+bool ExtentRanges::OverlapsWithExtent(const Extent& extent) const {
+  for (const auto& entry : extent_set_) {
+    if (ExtentsOverlap(entry, extent)) {
+      return true;
+    }
+  }
+  return false;
+}
+
 bool ExtentRanges::ContainsBlock(uint64_t block) const {
   auto lower = extent_set_.lower_bound(ExtentForRange(block, 1));
   // The block could be on the extent before the one in |lower|.
diff --git a/payload_generator/extent_ranges.h b/payload_generator/extent_ranges.h
index 62ffff4..68aa27f 100644
--- a/payload_generator/extent_ranges.h
+++ b/payload_generator/extent_ranges.h
@@ -63,6 +63,9 @@
   void AddRanges(const ExtentRanges& ranges);
   void SubtractRanges(const ExtentRanges& ranges);
 
+  // Returns true if the input extent overlaps with the current ExtentRanges.
+  bool OverlapsWithExtent(const Extent& extent) const;
+
   // Returns whether the block |block| is in this ExtentRange.
   bool ContainsBlock(uint64_t block) const;
 
diff --git a/payload_generator/full_update_generator.cc b/payload_generator/full_update_generator.cc
index 4d8b2f9..4a5f63a 100644
--- a/payload_generator/full_update_generator.cc
+++ b/payload_generator/full_update_generator.cc
@@ -99,7 +99,7 @@
       fd_, buffer_in_.data(), buffer_in_.size(), offset_, &bytes_read));
   TEST_AND_RETURN_FALSE(bytes_read == static_cast<ssize_t>(size_));
 
-  InstallOperation_Type op_type;
+  InstallOperation::Type op_type;
   TEST_AND_RETURN_FALSE(diff_utils::GenerateBestFullOperation(
       buffer_in_, version_, &op_blob, &op_type));
 
@@ -153,7 +153,7 @@
   aops->resize(num_chunks);
   vector<ChunkProcessor> chunk_processors;
   chunk_processors.reserve(num_chunks);
-  blob_file->SetTotalBlobs(num_chunks);
+  blob_file->IncTotalBlobs(num_chunks);
 
   for (size_t i = 0; i < num_chunks; ++i) {
     size_t start_block = i * chunk_blocks;
@@ -187,9 +187,6 @@
     thread_pool.AddWork(&processor);
   thread_pool.JoinAll();
 
-  // All the work done, disable logging.
-  blob_file->SetTotalBlobs(0);
-
   // All the operations must have a type set at this point. Otherwise, a
   // ChunkProcessor failed to complete.
   for (const AnnotatedOperation& aop : *aops) {
diff --git a/payload_generator/generate_delta_main.cc b/payload_generator/generate_delta_main.cc
index fe0a10b..5c1fb47 100644
--- a/payload_generator/generate_delta_main.cc
+++ b/payload_generator/generate_delta_main.cc
@@ -14,6 +14,7 @@
 // limitations under the License.
 //
 
+#include <map>
 #include <string>
 #include <vector>
 
@@ -22,6 +23,7 @@
 #include <base/logging.h>
 #include <base/strings/string_number_conversions.h>
 #include <base/strings/string_split.h>
+#include <base/strings/string_util.h>
 #include <brillo/flag_helper.h>
 #include <brillo/key_value_store.h>
 #include <brillo/message_loops/base_message_loop.h>
@@ -47,6 +49,7 @@
 // and an output file as arguments and the path to an output file and
 // generates a delta that can be sent to Chrome OS clients.
 
+using std::map;
 using std::string;
 using std::vector;
 
@@ -58,18 +61,15 @@
 constexpr char kPayloadPropertiesFormatJson[] = "json";
 
 void ParseSignatureSizes(const string& signature_sizes_flag,
-                         vector<int>* signature_sizes) {
+                         vector<size_t>* signature_sizes) {
   signature_sizes->clear();
   vector<string> split_strings = base::SplitString(
       signature_sizes_flag, ":", base::TRIM_WHITESPACE, base::SPLIT_WANT_ALL);
   for (const string& str : split_strings) {
-    int size = 0;
-    bool parsing_successful = base::StringToInt(str, &size);
+    size_t size = 0;
+    bool parsing_successful = base::StringToSizeT(str, &size);
     LOG_IF(FATAL, !parsing_successful) << "Invalid signature size: " << str;
 
-    LOG_IF(FATAL, size != (2048 / 8))
-        << "Only signature sizes of 256 bytes are supported.";
-
     signature_sizes->push_back(size);
   }
 }
@@ -106,7 +106,7 @@
   return true;
 }
 
-void CalculateHashForSigning(const vector<int>& sizes,
+void CalculateHashForSigning(const vector<size_t>& sizes,
                              const string& out_hash_file,
                              const string& out_metadata_hash_file,
                              const string& in_file) {
@@ -142,6 +142,7 @@
 
 void SignPayload(const string& in_file,
                  const string& out_file,
+                 const vector<size_t>& signature_sizes,
                  const string& payload_signature_file,
                  const string& metadata_signature_file,
                  const string& out_metadata_size_file) {
@@ -155,6 +156,7 @@
   SignatureFileFlagToBlobs(metadata_signature_file, &metadata_signatures);
   uint64_t final_metadata_size;
   CHECK(PayloadSigner::AddSignatureToPayload(in_file,
+                                             signature_sizes,
                                              payload_signatures,
                                              metadata_signatures,
                                              out_file,
@@ -253,8 +255,8 @@
                                        nullptr,
                                        new FileFetcher(),
                                        true /* interactive */);
-  auto filesystem_verifier_action =
-      std::make_unique<FilesystemVerifierAction>();
+  auto filesystem_verifier_action = std::make_unique<FilesystemVerifierAction>(
+      fake_boot_control.GetDynamicPartitionControl());
 
   BondActions(install_plan_action.get(), download_action.get());
   BondActions(download_action.get(), filesystem_verifier_action.get());
@@ -295,6 +297,39 @@
   return true;
 }
 
+template <typename Key, typename Val>
+string ToString(const map<Key, Val>& map) {
+  vector<string> result;
+  result.reserve(map.size());
+  for (const auto& it : map) {
+    result.emplace_back(it.first + ": " + it.second);
+  }
+  return "{" + base::JoinString(result, ",") + "}";
+}
+
+bool ParsePerPartitionTimestamps(const string& partition_timestamps,
+                                 PayloadGenerationConfig* config) {
+  base::StringPairs pairs;
+  CHECK(base::SplitStringIntoKeyValuePairs(
+      partition_timestamps, ':', ',', &pairs))
+      << "--partition_timestamps accepts commad "
+         "separated pairs. e.x. system:1234,vendor:5678";
+  map<string, string> partition_timestamps_map{
+      std::move_iterator(pairs.begin()), std::move_iterator(pairs.end())};
+  for (auto&& partition : config->target.partitions) {
+    auto&& it = partition_timestamps_map.find(partition.name);
+    if (it != partition_timestamps_map.end()) {
+      partition.version = std::move(it->second);
+      partition_timestamps_map.erase(it);
+    }
+  }
+  if (!partition_timestamps_map.empty()) {
+    LOG(ERROR) << "Unused timestamps: " << ToString(partition_timestamps_map);
+    return false;
+  }
+  return true;
+}
+
 int Main(int argc, char** argv) {
   DEFINE_string(old_image, "", "Path to the old rootfs");
   DEFINE_string(new_image, "", "Path to the new rootfs");
@@ -385,6 +420,11 @@
                0,
                "The maximum timestamp of the OS allowed to apply this "
                "payload.");
+  DEFINE_string(
+      partition_timestamps,
+      "",
+      "The per-partition maximum timestamps which the OS allowed to apply this "
+      "payload. Passed in comma separated pairs, e.x. system:1234,vendor:5678");
 
   DEFINE_string(old_channel,
                 "",
@@ -439,6 +479,17 @@
                 "",
                 "An info file specifying dynamic partition metadata. "
                 "Only allowed in major version 2 or newer.");
+  DEFINE_bool(disable_fec_computation,
+              false,
+              "Disables the fec data computation on device.");
+  DEFINE_string(
+      out_maximum_signature_size_file,
+      "",
+      "Path to the output maximum signature size given a private key.");
+  DEFINE_bool(is_partial_update,
+              false,
+              "The payload only targets a subset of partitions on the device,"
+              "e.g. generic kernel image update.");
 
   brillo::FlagHelper::Init(
       argc,
@@ -464,8 +515,34 @@
   // Initialize the Xz compressor.
   XzCompressInit();
 
-  vector<int> signature_sizes;
-  ParseSignatureSizes(FLAGS_signature_size, &signature_sizes);
+  if (!FLAGS_out_maximum_signature_size_file.empty()) {
+    LOG_IF(FATAL, FLAGS_private_key.empty())
+        << "Private key is not provided when calculating the maximum signature "
+           "size.";
+
+    size_t maximum_signature_size;
+    if (!PayloadSigner::GetMaximumSignatureSize(FLAGS_private_key,
+                                                &maximum_signature_size)) {
+      LOG(ERROR) << "Failed to get the maximum signature size of private key: "
+                 << FLAGS_private_key;
+      return 1;
+    }
+    // Write the size string to output file.
+    string signature_size_string = std::to_string(maximum_signature_size);
+    if (!utils::WriteFile(FLAGS_out_maximum_signature_size_file.c_str(),
+                          signature_size_string.c_str(),
+                          signature_size_string.size())) {
+      PLOG(ERROR) << "Failed to write the maximum signature size to "
+                  << FLAGS_out_maximum_signature_size_file << ".";
+      return 1;
+    }
+    return 0;
+  }
+
+  vector<size_t> signature_sizes;
+  if (!FLAGS_signature_size.empty()) {
+    ParseSignatureSizes(FLAGS_signature_size, &signature_sizes);
+  }
 
   if (!FLAGS_out_hash_file.empty() || !FLAGS_out_metadata_hash_file.empty()) {
     CHECK(FLAGS_out_metadata_size_file.empty());
@@ -478,6 +555,7 @@
   if (!FLAGS_payload_signature_file.empty()) {
     SignPayload(FLAGS_in_file,
                 FLAGS_out_file,
+                signature_sizes,
                 FLAGS_payload_signature_file,
                 FLAGS_metadata_signature_file,
                 FLAGS_out_metadata_size_file);
@@ -546,6 +624,8 @@
         << "Partition name can't be empty, see --partition_names.";
     payload_config.target.partitions.emplace_back(partition_names[i]);
     payload_config.target.partitions.back().path = new_partitions[i];
+    payload_config.target.partitions.back().disable_fec_computation =
+        FLAGS_disable_fec_computation;
     if (i < new_mapfiles.size())
       payload_config.target.partitions.back().mapfile_path = new_mapfiles[i];
   }
@@ -598,6 +678,10 @@
     CHECK(payload_config.target.ValidateDynamicPartitionMetadata());
   }
 
+  if (FLAGS_is_partial_update) {
+    payload_config.is_partial_update = true;
+  }
+
   CHECK(!FLAGS_out_file.empty());
 
   // Ignore failures. These are optional arguments.
@@ -670,8 +754,13 @@
   }
 
   payload_config.max_timestamp = FLAGS_max_timestamp;
+  if (!FLAGS_partition_timestamps.empty()) {
+    CHECK(ParsePerPartitionTimestamps(FLAGS_partition_timestamps,
+                                      &payload_config));
+  }
 
-  if (payload_config.version.minor >= kVerityMinorPayloadVersion)
+  if (payload_config.is_delta &&
+      payload_config.version.minor >= kVerityMinorPayloadVersion)
     CHECK(payload_config.target.LoadVerityConfig());
 
   LOG(INFO) << "Generating " << (payload_config.is_delta ? "delta" : "full")
diff --git a/payload_generator/merge_sequence_generator.cc b/payload_generator/merge_sequence_generator.cc
new file mode 100644
index 0000000..eaffeac
--- /dev/null
+++ b/payload_generator/merge_sequence_generator.cc
@@ -0,0 +1,269 @@
+//
+// Copyright (C) 2020 The Android Open Source Project
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//      http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+//
+
+#include "update_engine/payload_generator/merge_sequence_generator.h"
+
+#include <algorithm>
+
+#include "update_engine/payload_generator/extent_utils.h"
+
+namespace chromeos_update_engine {
+
+CowMergeOperation CreateCowMergeOperation(const Extent& src_extent,
+                                          const Extent& dst_extent) {
+  CowMergeOperation ret;
+  ret.set_type(CowMergeOperation::COW_COPY);
+  *ret.mutable_src_extent() = src_extent;
+  *ret.mutable_dst_extent() = dst_extent;
+  return ret;
+}
+
+std::ostream& operator<<(std::ostream& os,
+                         const CowMergeOperation& merge_operation) {
+  os << "CowMergeOperation src extent: "
+     << ExtentsToString({merge_operation.src_extent()})
+     << ", dst extent: " << ExtentsToString({merge_operation.dst_extent()});
+  return os;
+}
+
+// The OTA generation guarantees that all blocks in the dst extent will be
+// written only once. So we can use it to order the CowMergeOperation.
+bool operator<(const CowMergeOperation& op1, const CowMergeOperation& op2) {
+  return op1.dst_extent().start_block() < op2.dst_extent().start_block();
+}
+
+bool operator==(const CowMergeOperation& op1, const CowMergeOperation& op2) {
+  return op1.type() == op2.type() && op1.src_extent() == op2.src_extent() &&
+         op1.dst_extent() == op2.dst_extent();
+}
+
+std::unique_ptr<MergeSequenceGenerator> MergeSequenceGenerator::Create(
+    const std::vector<AnnotatedOperation>& aops) {
+  std::vector<CowMergeOperation> sequence;
+  for (const auto& aop : aops) {
+    // Only handle SOURCE_COPY now for the cow size optimization.
+    if (aop.op.type() != InstallOperation::SOURCE_COPY) {
+      continue;
+    }
+    if (aop.op.dst_extents().size() != 1) {
+      std::vector<Extent> out_extents;
+      ExtentsToVector(aop.op.dst_extents(), &out_extents);
+      LOG(ERROR) << "The dst extents for source_copy expects to be contiguous,"
+                 << " dst extents: " << ExtentsToString(out_extents);
+      return nullptr;
+    }
+
+    // Split the source extents.
+    size_t used_blocks = 0;
+    for (const auto& src_extent : aop.op.src_extents()) {
+      // The dst_extent in the merge sequence will be a subset of
+      // InstallOperation's dst_extent. This will simplify the OTA -> COW
+      // conversion when we install the payload.
+      Extent dst_extent =
+          ExtentForRange(aop.op.dst_extents(0).start_block() + used_blocks,
+                         src_extent.num_blocks());
+      sequence.emplace_back(CreateCowMergeOperation(src_extent, dst_extent));
+      used_blocks += src_extent.num_blocks();
+    }
+
+    if (used_blocks != aop.op.dst_extents(0).num_blocks()) {
+      LOG(ERROR) << "Number of blocks in src extents doesn't equal to the"
+                 << " ones in the dst extents, src blocks " << used_blocks
+                 << ", dst blocks " << aop.op.dst_extents(0).num_blocks();
+      return nullptr;
+    }
+  }
+
+  std::sort(sequence.begin(), sequence.end());
+  return std::unique_ptr<MergeSequenceGenerator>(
+      new MergeSequenceGenerator(sequence));
+}
+
+bool MergeSequenceGenerator::FindDependency(
+    std::map<CowMergeOperation, std::set<CowMergeOperation>>* result) const {
+  CHECK(result);
+  LOG(INFO) << "Finding dependencies";
+
+  // Since the OTA operation may reuse some source blocks, use the binary
+  // search on sorted dst extents to find overlaps.
+  std::map<CowMergeOperation, std::set<CowMergeOperation>> merge_after;
+  for (const auto& op : operations_) {
+    // lower bound (inclusive): dst extent's end block >= src extent's start
+    // block.
+    const auto lower_it = std::lower_bound(
+        operations_.begin(),
+        operations_.end(),
+        op,
+        [](const CowMergeOperation& it, const CowMergeOperation& op) {
+          auto dst_end_block =
+              it.dst_extent().start_block() + it.dst_extent().num_blocks() - 1;
+          return dst_end_block < op.src_extent().start_block();
+        });
+    // upper bound: dst extent's start block > src extent's end block
+    const auto upper_it = std::upper_bound(
+        lower_it,
+        operations_.end(),
+        op,
+        [](const CowMergeOperation& op, const CowMergeOperation& it) {
+          auto src_end_block =
+              op.src_extent().start_block() + op.src_extent().num_blocks() - 1;
+          return src_end_block < it.dst_extent().start_block();
+        });
+
+    // TODO(xunchang) skip inserting the empty set to merge_after.
+    if (lower_it == upper_it) {
+      merge_after.insert({op, {}});
+    } else {
+      std::set<CowMergeOperation> operations(lower_it, upper_it);
+      auto it = operations.find(op);
+      if (it != operations.end()) {
+        LOG(INFO) << "Self overlapping " << op;
+        operations.erase(it);
+      }
+      auto ret = merge_after.emplace(op, std::move(operations));
+      // Check the insertion indeed happens.
+      CHECK(ret.second);
+    }
+  }
+
+  *result = std::move(merge_after);
+  return true;
+}
+
+bool MergeSequenceGenerator::Generate(
+    std::vector<CowMergeOperation>* sequence) const {
+  sequence->clear();
+  std::map<CowMergeOperation, std::set<CowMergeOperation>> merge_after;
+  if (!FindDependency(&merge_after)) {
+    LOG(ERROR) << "Failed to find dependencies";
+    return false;
+  }
+
+  LOG(INFO) << "Generating sequence";
+
+  // Use the non-DFS version of the topology sort. So we can control the
+  // operations to discard to break cycles; thus yielding a deterministic
+  // sequence.
+  std::map<CowMergeOperation, int> incoming_edges;
+  for (const auto& it : merge_after) {
+    for (const auto& blocked : it.second) {
+      // Value is default initialized to 0.
+      incoming_edges[blocked] += 1;
+    }
+  }
+
+  std::set<CowMergeOperation> free_operations;
+  for (const auto& op : operations_) {
+    if (incoming_edges.find(op) == incoming_edges.end()) {
+      free_operations.insert(op);
+    }
+  }
+
+  std::vector<CowMergeOperation> merge_sequence;
+  std::set<CowMergeOperation> convert_to_raw;
+  while (!incoming_edges.empty()) {
+    if (!free_operations.empty()) {
+      merge_sequence.insert(
+          merge_sequence.end(), free_operations.begin(), free_operations.end());
+    } else {
+      auto to_convert = incoming_edges.begin()->first;
+      free_operations.insert(to_convert);
+      convert_to_raw.insert(to_convert);
+      LOG(INFO) << "Converting operation to raw " << to_convert;
+    }
+
+    std::set<CowMergeOperation> next_free_operations;
+    for (const auto& op : free_operations) {
+      incoming_edges.erase(op);
+
+      // Now that this particular operation is merged, other operations blocked
+      // by this one may be free. Decrement the count of blocking operations,
+      // and set up the free operations for the next iteration.
+      for (const auto& blocked : merge_after[op]) {
+        auto it = incoming_edges.find(blocked);
+        if (it == incoming_edges.end()) {
+          continue;
+        }
+
+        auto blocking_transfer_count = &it->second;
+        if (*blocking_transfer_count <= 0) {
+          LOG(ERROR) << "Unexpected count in merge after map "
+                     << blocking_transfer_count;
+          return false;
+        }
+        // This operation is no longer blocked by anyone. Add it to the merge
+        // sequence in the next iteration.
+        *blocking_transfer_count -= 1;
+        if (*blocking_transfer_count == 0) {
+          next_free_operations.insert(blocked);
+        }
+      }
+    }
+
+    LOG(INFO) << "Remaining transfers " << incoming_edges.size()
+              << ", free transfers " << free_operations.size()
+              << ", merge_sequence size " << merge_sequence.size();
+    free_operations = std::move(next_free_operations);
+  }
+
+  if (!free_operations.empty()) {
+    merge_sequence.insert(
+        merge_sequence.end(), free_operations.begin(), free_operations.end());
+  }
+
+  CHECK_EQ(operations_.size(), merge_sequence.size() + convert_to_raw.size());
+
+  size_t blocks_in_sequence = 0;
+  for (const CowMergeOperation& transfer : merge_sequence) {
+    blocks_in_sequence += transfer.dst_extent().num_blocks();
+  }
+
+  size_t blocks_in_raw = 0;
+  for (const CowMergeOperation& transfer : convert_to_raw) {
+    blocks_in_raw += transfer.dst_extent().num_blocks();
+  }
+
+  LOG(INFO) << "Blocks in merge sequence " << blocks_in_sequence
+            << ", blocks in raw " << blocks_in_raw;
+  if (!ValidateSequence(merge_sequence)) {
+    return false;
+  }
+
+  *sequence = std::move(merge_sequence);
+  return true;
+}
+
+bool MergeSequenceGenerator::ValidateSequence(
+    const std::vector<CowMergeOperation>& sequence) {
+  LOG(INFO) << "Validating merge sequence";
+  ExtentRanges visited;
+  for (const auto& op : sequence) {
+    if (visited.OverlapsWithExtent(op.src_extent())) {
+      LOG(ERROR) << "Transfer violates the merge sequence " << op
+                 << "Visited extent ranges: ";
+      visited.Dump();
+      return false;
+    }
+
+    CHECK(!visited.OverlapsWithExtent(op.dst_extent()))
+        << "dst extent should write only once.";
+    visited.AddExtent(op.dst_extent());
+  }
+
+  return true;
+}
+
+}  // namespace chromeos_update_engine
diff --git a/payload_generator/merge_sequence_generator.h b/payload_generator/merge_sequence_generator.h
new file mode 100644
index 0000000..bc0158e
--- /dev/null
+++ b/payload_generator/merge_sequence_generator.h
@@ -0,0 +1,74 @@
+//
+// Copyright (C) 2020 The Android Open Source Project
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//      http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+//
+
+#ifndef UPDATE_ENGINE_PAYLOAD_GENERATOR_MERGE_SEQUENCE_GENERATOR_H_
+#define UPDATE_ENGINE_PAYLOAD_GENERATOR_MERGE_SEQUENCE_GENERATOR_H_
+
+#include <map>
+#include <memory>
+#include <set>
+#include <utility>
+#include <vector>
+
+#include "update_engine/payload_generator/annotated_operation.h"
+#include "update_engine/payload_generator/extent_ranges.h"
+#include "update_engine/payload_generator/extent_utils.h"
+#include "update_engine/update_metadata.pb.h"
+
+namespace chromeos_update_engine {
+// Constructs CowMergeOperation from src & dst extents
+CowMergeOperation CreateCowMergeOperation(const Extent& src_extent,
+                                          const Extent& dst_extent);
+
+// Comparator for CowMergeOperation.
+bool operator<(const CowMergeOperation& op1, const CowMergeOperation& op2);
+bool operator==(const CowMergeOperation& op1, const CowMergeOperation& op2);
+
+std::ostream& operator<<(std::ostream& os,
+                         const CowMergeOperation& merge_operation);
+
+// This class takes a list of CowMergeOperations; and sorts them so that no
+// read after write will happen by following the sequence. When there is a
+// cycle, we will omit some operations in the list. Therefore, the result
+// sequence may not contain all blocks in the input list.
+class MergeSequenceGenerator {
+ public:
+  // Creates an object from a list of OTA InstallOperations. Returns nullptr on
+  // failure.
+  static std::unique_ptr<MergeSequenceGenerator> Create(
+      const std::vector<AnnotatedOperation>& aops);
+  // Checks that no read after write happens in the given sequence.
+  static bool ValidateSequence(const std::vector<CowMergeOperation>& sequence);
+
+  // Generates a merge sequence from |operations_|, puts the result in
+  // |sequence|. Returns false on failure.
+  bool Generate(std::vector<CowMergeOperation>* sequence) const;
+
+ private:
+  friend class MergeSequenceGeneratorTest;
+  explicit MergeSequenceGenerator(std::vector<CowMergeOperation> transfers)
+      : operations_(std::move(transfers)) {}
+
+  // For a given merge operation, finds all the operations that should merge
+  // after myself. Put the result in |merge_after|.
+  bool FindDependency(std::map<CowMergeOperation, std::set<CowMergeOperation>>*
+                          merge_after) const;
+  // The list of CowMergeOperations to sort.
+  std::vector<CowMergeOperation> operations_;
+};
+
+}  // namespace chromeos_update_engine
+#endif
diff --git a/payload_generator/merge_sequence_generator_unittest.cc b/payload_generator/merge_sequence_generator_unittest.cc
new file mode 100644
index 0000000..567ede1
--- /dev/null
+++ b/payload_generator/merge_sequence_generator_unittest.cc
@@ -0,0 +1,196 @@
+//
+// Copyright (C) 2020 The Android Open Source Project
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//      http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+//
+
+#include <algorithm>
+#include <vector>
+
+#include <gtest/gtest.h>
+
+#include "update_engine/payload_consumer/payload_constants.h"
+#include "update_engine/payload_generator/extent_utils.h"
+#include "update_engine/payload_generator/merge_sequence_generator.h"
+
+namespace chromeos_update_engine {
+class MergeSequenceGeneratorTest : public ::testing::Test {
+ protected:
+  void VerifyTransfers(MergeSequenceGenerator* generator,
+                       const std::vector<CowMergeOperation>& expected) {
+    ASSERT_EQ(expected, generator->operations_);
+  }
+
+  void FindDependency(
+      std::vector<CowMergeOperation> transfers,
+      std::map<CowMergeOperation, std::set<CowMergeOperation>>* result) {
+    std::sort(transfers.begin(), transfers.end());
+    MergeSequenceGenerator generator(std::move(transfers));
+    ASSERT_TRUE(generator.FindDependency(result));
+  }
+
+  void GenerateSequence(std::vector<CowMergeOperation> transfers,
+                        const std::vector<CowMergeOperation>& expected) {
+    std::sort(transfers.begin(), transfers.end());
+    MergeSequenceGenerator generator(std::move(transfers));
+    std::vector<CowMergeOperation> sequence;
+    ASSERT_TRUE(generator.Generate(&sequence));
+    ASSERT_EQ(expected, sequence);
+  }
+};
+
+TEST_F(MergeSequenceGeneratorTest, Create) {
+  std::vector<AnnotatedOperation> aops{{"file1", {}}, {"file2", {}}};
+  aops[0].op.set_type(InstallOperation::SOURCE_COPY);
+  *aops[0].op.add_src_extents() = ExtentForRange(10, 10);
+  *aops[0].op.add_dst_extents() = ExtentForRange(30, 10);
+
+  aops[1].op.set_type(InstallOperation::SOURCE_COPY);
+  *aops[1].op.add_src_extents() = ExtentForRange(20, 10);
+  *aops[1].op.add_dst_extents() = ExtentForRange(40, 10);
+
+  auto generator = MergeSequenceGenerator::Create(aops);
+  ASSERT_TRUE(generator);
+  std::vector<CowMergeOperation> expected = {
+      CreateCowMergeOperation(ExtentForRange(10, 10), ExtentForRange(30, 10)),
+      CreateCowMergeOperation(ExtentForRange(20, 10), ExtentForRange(40, 10))};
+  VerifyTransfers(generator.get(), expected);
+
+  *aops[1].op.add_src_extents() = ExtentForRange(30, 5);
+  *aops[1].op.add_dst_extents() = ExtentForRange(50, 5);
+  generator = MergeSequenceGenerator::Create(aops);
+  ASSERT_FALSE(generator);
+}
+
+TEST_F(MergeSequenceGeneratorTest, Create_SplitSource) {
+  InstallOperation op;
+  op.set_type(InstallOperation::SOURCE_COPY);
+  *(op.add_src_extents()) = ExtentForRange(2, 3);
+  *(op.add_src_extents()) = ExtentForRange(6, 1);
+  *(op.add_src_extents()) = ExtentForRange(8, 4);
+  *(op.add_dst_extents()) = ExtentForRange(10, 8);
+
+  AnnotatedOperation aop{"file1", op};
+  auto generator = MergeSequenceGenerator::Create({aop});
+  ASSERT_TRUE(generator);
+  std::vector<CowMergeOperation> expected = {
+      CreateCowMergeOperation(ExtentForRange(2, 3), ExtentForRange(10, 3)),
+      CreateCowMergeOperation(ExtentForRange(6, 1), ExtentForRange(13, 1)),
+      CreateCowMergeOperation(ExtentForRange(8, 4), ExtentForRange(14, 4))};
+  VerifyTransfers(generator.get(), expected);
+}
+
+TEST_F(MergeSequenceGeneratorTest, FindDependency) {
+  std::vector<CowMergeOperation> transfers = {
+      CreateCowMergeOperation(ExtentForRange(10, 10), ExtentForRange(15, 10)),
+      CreateCowMergeOperation(ExtentForRange(40, 10), ExtentForRange(50, 10)),
+  };
+
+  std::map<CowMergeOperation, std::set<CowMergeOperation>> merge_after;
+  FindDependency(transfers, &merge_after);
+  ASSERT_EQ(std::set<CowMergeOperation>(), merge_after.at(transfers[0]));
+  ASSERT_EQ(std::set<CowMergeOperation>(), merge_after.at(transfers[1]));
+
+  transfers = {
+      CreateCowMergeOperation(ExtentForRange(10, 10), ExtentForRange(25, 10)),
+      CreateCowMergeOperation(ExtentForRange(24, 5), ExtentForRange(35, 5)),
+      CreateCowMergeOperation(ExtentForRange(30, 10), ExtentForRange(15, 10)),
+  };
+
+  FindDependency(transfers, &merge_after);
+  ASSERT_EQ(std::set<CowMergeOperation>({transfers[2]}),
+            merge_after.at(transfers[0]));
+  ASSERT_EQ(std::set<CowMergeOperation>({transfers[0], transfers[2]}),
+            merge_after.at(transfers[1]));
+  ASSERT_EQ(std::set<CowMergeOperation>({transfers[0], transfers[1]}),
+            merge_after.at(transfers[2]));
+}
+
+TEST_F(MergeSequenceGeneratorTest, FindDependency_ReusedSourceBlocks) {
+  std::vector<CowMergeOperation> transfers = {
+      CreateCowMergeOperation(ExtentForRange(5, 10), ExtentForRange(15, 10)),
+      CreateCowMergeOperation(ExtentForRange(6, 5), ExtentForRange(30, 5)),
+      CreateCowMergeOperation(ExtentForRange(50, 5), ExtentForRange(5, 5)),
+  };
+
+  std::map<CowMergeOperation, std::set<CowMergeOperation>> merge_after;
+  FindDependency(transfers, &merge_after);
+  ASSERT_EQ(std::set<CowMergeOperation>({transfers[2]}),
+            merge_after.at(transfers[0]));
+  ASSERT_EQ(std::set<CowMergeOperation>({transfers[2]}),
+            merge_after.at(transfers[1]));
+}
+
+TEST_F(MergeSequenceGeneratorTest, ValidateSequence) {
+  std::vector<CowMergeOperation> transfers = {
+      CreateCowMergeOperation(ExtentForRange(10, 10), ExtentForRange(15, 10)),
+      CreateCowMergeOperation(ExtentForRange(30, 10), ExtentForRange(40, 10)),
+  };
+
+  // Self overlapping
+  ASSERT_TRUE(MergeSequenceGenerator::ValidateSequence(transfers));
+
+  transfers = {
+      CreateCowMergeOperation(ExtentForRange(30, 10), ExtentForRange(20, 10)),
+      CreateCowMergeOperation(ExtentForRange(15, 10), ExtentForRange(10, 10)),
+  };
+  ASSERT_FALSE(MergeSequenceGenerator::ValidateSequence(transfers));
+}
+
+TEST_F(MergeSequenceGeneratorTest, GenerateSequenceNoCycles) {
+  std::vector<CowMergeOperation> transfers = {
+      CreateCowMergeOperation(ExtentForRange(10, 10), ExtentForRange(15, 10)),
+      // file3 should merge before file2
+      CreateCowMergeOperation(ExtentForRange(40, 5), ExtentForRange(25, 5)),
+      CreateCowMergeOperation(ExtentForRange(25, 10), ExtentForRange(30, 10)),
+  };
+
+  std::vector<CowMergeOperation> expected{
+      transfers[0], transfers[2], transfers[1]};
+  GenerateSequence(transfers, expected);
+}
+
+TEST_F(MergeSequenceGeneratorTest, GenerateSequenceWithCycles) {
+  std::vector<CowMergeOperation> transfers = {
+      CreateCowMergeOperation(ExtentForRange(25, 10), ExtentForRange(30, 10)),
+      CreateCowMergeOperation(ExtentForRange(30, 10), ExtentForRange(40, 10)),
+      CreateCowMergeOperation(ExtentForRange(40, 10), ExtentForRange(25, 10)),
+      CreateCowMergeOperation(ExtentForRange(10, 10), ExtentForRange(15, 10)),
+  };
+
+  // file 1,2,3 form a cycle. And file3, whose dst ext has smallest offset, will
+  // be converted to raw blocks
+  std::vector<CowMergeOperation> expected{
+      transfers[3], transfers[1], transfers[0]};
+  GenerateSequence(transfers, expected);
+}
+
+TEST_F(MergeSequenceGeneratorTest, GenerateSequenceMultipleCycles) {
+  std::vector<CowMergeOperation> transfers = {
+      // cycle 1
+      CreateCowMergeOperation(ExtentForRange(10, 10), ExtentForRange(25, 10)),
+      CreateCowMergeOperation(ExtentForRange(24, 5), ExtentForRange(35, 5)),
+      CreateCowMergeOperation(ExtentForRange(30, 10), ExtentForRange(15, 10)),
+      // cycle 2
+      CreateCowMergeOperation(ExtentForRange(55, 10), ExtentForRange(60, 10)),
+      CreateCowMergeOperation(ExtentForRange(60, 10), ExtentForRange(70, 10)),
+      CreateCowMergeOperation(ExtentForRange(70, 10), ExtentForRange(55, 10)),
+  };
+
+  // file 3, 6 will be converted to raw.
+  std::vector<CowMergeOperation> expected{
+      transfers[1], transfers[0], transfers[4], transfers[3]};
+  GenerateSequence(transfers, expected);
+}
+
+}  // namespace chromeos_update_engine
diff --git a/payload_generator/payload_file.cc b/payload_generator/payload_file.cc
index b55d03c..49dff4e 100644
--- a/payload_generator/payload_file.cc
+++ b/payload_generator/payload_file.cc
@@ -78,17 +78,23 @@
     *(manifest_.mutable_dynamic_partition_metadata()) =
         *(config.target.dynamic_partition_metadata);
 
+  if (config.is_partial_update) {
+    manifest_.set_partial_update(true);
+  }
   return true;
 }
 
 bool PayloadFile::AddPartition(const PartitionConfig& old_conf,
                                const PartitionConfig& new_conf,
-                               const vector<AnnotatedOperation>& aops) {
+                               vector<AnnotatedOperation> aops,
+                               vector<CowMergeOperation> merge_sequence) {
   Partition part;
   part.name = new_conf.name;
-  part.aops = aops;
+  part.aops = std::move(aops);
+  part.cow_merge_sequence = std::move(merge_sequence);
   part.postinstall = new_conf.postinstall;
   part.verity = new_conf.verity;
+  part.version = new_conf.version;
   // Initialize the PartitionInfo objects if present.
   if (!old_conf.path.empty())
     TEST_AND_RETURN_FALSE(
@@ -129,6 +135,9 @@
   for (const auto& part : part_vec_) {
     PartitionUpdate* partition = manifest_.add_partitions();
     partition->set_partition_name(part.name);
+    if (!part.version.empty()) {
+      partition->set_version(part.version);
+    }
     if (part.postinstall.run) {
       partition->set_run_postinstall(true);
       if (!part.postinstall.path.empty())
@@ -156,6 +165,10 @@
     for (const AnnotatedOperation& aop : part.aops) {
       *partition->add_operations() = aop.op;
     }
+    for (const auto& merge_op : part.cow_merge_sequence) {
+      *partition->add_merge_operations() = merge_op;
+    }
+
     if (part.old_info.has_size() || part.old_info.has_hash())
       *(partition->mutable_old_partition_info()) = part.old_info;
     if (part.new_info.has_size() || part.new_info.has_hash())
@@ -167,16 +180,14 @@
   uint64_t signature_blob_length = 0;
   if (!private_key_path.empty()) {
     TEST_AND_RETURN_FALSE(PayloadSigner::SignatureBlobLength(
-        vector<string>(1, private_key_path), &signature_blob_length));
+        {private_key_path}, &signature_blob_length));
     PayloadSigner::AddSignatureToManifest(
-        next_blob_offset,
-        signature_blob_length,
-        &manifest_);
+        next_blob_offset, signature_blob_length, &manifest_);
   }
 
   // Serialize protobuf
   string serialized_manifest;
-  TEST_AND_RETURN_FALSE(manifest_.AppendToString(&serialized_manifest));
+  TEST_AND_RETURN_FALSE(manifest_.SerializeToString(&serialized_manifest));
 
   uint64_t metadata_size =
       sizeof(kDeltaMagic) + 2 * sizeof(uint64_t) + serialized_manifest.size();
@@ -215,13 +226,12 @@
 
   // Write metadata signature blob.
   if (!private_key_path.empty()) {
-    brillo::Blob metadata_hash, metadata_signature;
+    brillo::Blob metadata_hash;
     TEST_AND_RETURN_FALSE(HashCalculator::RawHashOfFile(
         payload_file, metadata_size, &metadata_hash));
-    TEST_AND_RETURN_FALSE(
-        PayloadSigner::SignHashWithKeys(metadata_hash,
-                                        vector<string>(1, private_key_path),
-                                        &metadata_signature));
+    string metadata_signature;
+    TEST_AND_RETURN_FALSE(PayloadSigner::SignHashWithKeys(
+        metadata_hash, {private_key_path}, &metadata_signature));
     TEST_AND_RETURN_FALSE_ERRNO(
         writer.Write(metadata_signature.data(), metadata_signature.size()));
   }
@@ -245,16 +255,16 @@
   // Write payload signature blob.
   if (!private_key_path.empty()) {
     LOG(INFO) << "Signing the update...";
-    brillo::Blob signature_blob;
+    string signature;
     TEST_AND_RETURN_FALSE(PayloadSigner::SignPayload(
         payload_file,
-        vector<string>(1, private_key_path),
+        {private_key_path},
         metadata_size,
         metadata_signature_size,
         metadata_size + metadata_signature_size + manifest_.signatures_offset(),
-        &signature_blob));
+        &signature));
     TEST_AND_RETURN_FALSE_ERRNO(
-        writer.Write(signature_blob.data(), signature_blob.size()));
+        writer.Write(signature.data(), signature.size()));
   }
 
   ReportPayloadUsage(metadata_size);
@@ -330,15 +340,15 @@
     const DeltaObject& object = object_count.first;
     // Use printf() instead of LOG(INFO) because timestamp makes it difficult to
     // compare two reports.
-    printf(
-        kFormatString,
-        object.size * 100.0 / total_size,
-        object.size,
-        (object.type >= 0 ? InstallOperationTypeName(
-                                static_cast<InstallOperation_Type>(object.type))
-                          : "-"),
-        object.name.c_str(),
-        object_count.second);
+    printf(kFormatString,
+           object.size * 100.0 / total_size,
+           object.size,
+           (object.type >= 0
+                ? InstallOperationTypeName(
+                      static_cast<InstallOperation::Type>(object.type))
+                : "-"),
+           object.name.c_str(),
+           object_count.second);
   }
   printf(kFormatString, 100.0, total_size, "", "<total>", total_op);
   fflush(stdout);
diff --git a/payload_generator/payload_file.h b/payload_generator/payload_file.h
index 9dc80a7..8b17956 100644
--- a/payload_generator/payload_file.h
+++ b/payload_generator/payload_file.h
@@ -43,7 +43,8 @@
   // reference a blob stored in the file provided to WritePayload().
   bool AddPartition(const PartitionConfig& old_conf,
                     const PartitionConfig& new_conf,
-                    const std::vector<AnnotatedOperation>& aops);
+                    std::vector<AnnotatedOperation> aops,
+                    std::vector<CowMergeOperation> merge_sequence);
 
   // Write the payload to the |payload_file| file. The operations reference
   // blobs in the |data_blobs_path| file and the blobs will be reordered in the
@@ -60,9 +61,9 @@
   // Computes a SHA256 hash of the given buf and sets the hash value in the
   // operation so that update_engine could verify. This hash should be set
   // for all operations that have a non-zero data blob. One exception is the
-  // dummy operation for signature blob because the contents of the signature
+  // fake operation for signature blob because the contents of the signature
   // blob will not be available at payload creation time. So, update_engine will
-  // gracefully ignore the dummy signature operation.
+  // gracefully ignore the fake signature operation.
   static bool AddOperationHash(InstallOperation* op, const brillo::Blob& buf);
 
   // Install operations in the manifest may reference data blobs, which
@@ -90,12 +91,15 @@
 
     // The operations to be performed to this partition.
     std::vector<AnnotatedOperation> aops;
+    std::vector<CowMergeOperation> cow_merge_sequence;
 
     PartitionInfo old_info;
     PartitionInfo new_info;
 
     PostInstallConfig postinstall;
     VerityConfig verity;
+    // Per partition timestamp.
+    std::string version;
   };
 
   std::vector<Partition> part_vec_;
diff --git a/payload_generator/payload_generation_config.cc b/payload_generator/payload_generation_config.cc
index 7158796..35a95dc 100644
--- a/payload_generator/payload_generation_config.cc
+++ b/payload_generator/payload_generation_config.cc
@@ -150,25 +150,34 @@
   for (const auto& group_name : group_names) {
     DynamicPartitionGroup* group = metadata->add_groups();
     group->set_name(group_name);
-    if (!store.GetString(group_name + "_size", &buf)) {
-      LOG(ERROR) << "Missing " << group_name + "_size.";
+    if (!store.GetString("super_" + group_name + "_group_size", &buf) &&
+        !store.GetString(group_name + "_size", &buf)) {
+      LOG(ERROR) << "Missing super_" << group_name + "_group_size or "
+                 << group_name << "_size.";
       return false;
     }
 
     uint64_t max_size;
     if (!base::StringToUint64(buf, &max_size)) {
-      LOG(ERROR) << group_name << "_size=" << buf << " is not an integer.";
+      LOG(ERROR) << "Group size for " << group_name << " = " << buf
+                 << " is not an integer.";
       return false;
     }
     group->set_size(max_size);
 
-    if (store.GetString(group_name + "_partition_list", &buf)) {
+    if (store.GetString("super_" + group_name + "_partition_list", &buf) ||
+        store.GetString(group_name + "_partition_list", &buf)) {
       auto partition_names = brillo::string_utils::Split(buf, " ");
       for (const auto& partition_name : partition_names) {
         group->add_partition_names()->assign(partition_name);
       }
     }
   }
+
+  bool snapshot_enabled = false;
+  store.GetBoolean("virtual_ab", &snapshot_enabled);
+  metadata->set_snapshot_enabled(snapshot_enabled);
+
   dynamic_partition_metadata = std::move(metadata);
   return true;
 }
@@ -225,11 +234,12 @@
                         minor == kOpSrcHashMinorPayloadVersion ||
                         minor == kBrotliBsdiffMinorPayloadVersion ||
                         minor == kPuffdiffMinorPayloadVersion ||
-                        minor == kVerityMinorPayloadVersion);
+                        minor == kVerityMinorPayloadVersion ||
+                        minor == kPartialUpdateMinorPayloadVersion);
   return true;
 }
 
-bool PayloadVersion::OperationAllowed(InstallOperation_Type operation) const {
+bool PayloadVersion::OperationAllowed(InstallOperation::Type operation) const {
   switch (operation) {
     // Full operations:
     case InstallOperation::REPLACE:
@@ -260,13 +270,14 @@
   return false;
 }
 
-bool PayloadVersion::IsDelta() const {
+bool PayloadVersion::IsDeltaOrPartial() const {
   return minor != kFullPayloadMinorVersion;
 }
 
 bool PayloadGenerationConfig::Validate() const {
   TEST_AND_RETURN_FALSE(version.Validate());
-  TEST_AND_RETURN_FALSE(version.IsDelta() == is_delta);
+  TEST_AND_RETURN_FALSE(version.IsDeltaOrPartial() ==
+                        (is_delta || is_partial_update));
   if (is_delta) {
     for (const PartitionConfig& part : source.partitions) {
       if (!part.path.empty()) {
@@ -294,6 +305,10 @@
       TEST_AND_RETURN_FALSE(part.verity.IsEmpty());
   }
 
+  if (version.minor < kPartialUpdateMinorPayloadVersion) {
+    TEST_AND_RETURN_FALSE(!is_partial_update);
+  }
+
   TEST_AND_RETURN_FALSE(hard_chunk_size == -1 ||
                         hard_chunk_size % block_size == 0);
   TEST_AND_RETURN_FALSE(soft_chunk_size % block_size == 0);
diff --git a/payload_generator/payload_generation_config.h b/payload_generator/payload_generation_config.h
index 32f1229..ec63043 100644
--- a/payload_generator/payload_generation_config.h
+++ b/payload_generator/payload_generation_config.h
@@ -116,6 +116,12 @@
 
   PostInstallConfig postinstall;
   VerityConfig verity;
+
+  // Enables the on device fec data computation by default.
+  bool disable_fec_computation = false;
+
+  // Per-partition version, usually a number representing timestamp.
+  std::string version;
 };
 
 // The ImageConfig struct describes a pair of binaries kernel and rootfs and the
@@ -165,10 +171,10 @@
   bool Validate() const;
 
   // Return whether the passed |operation| is allowed by this payload.
-  bool OperationAllowed(InstallOperation_Type operation) const;
+  bool OperationAllowed(InstallOperation::Type operation) const;
 
-  // Whether this payload version is a delta payload.
-  bool IsDelta() const;
+  // Whether this payload version is a delta or partial payload.
+  bool IsDeltaOrPartial() const;
 
   // The major version of the payload.
   uint64_t major;
@@ -195,6 +201,10 @@
   // Whether the requested payload is a delta payload.
   bool is_delta = false;
 
+  // Whether the requested payload is a partial payload, i.e. only update a
+  // subset of partitions on device.
+  bool is_partial_update = false;
+
   // The major/minor version of the payload.
   PayloadVersion version;
 
diff --git a/payload_generator/payload_generation_config_android.cc b/payload_generator/payload_generation_config_android.cc
index 90c053f..d950092 100644
--- a/payload_generator/payload_generation_config_android.cc
+++ b/payload_generator/payload_generation_config_android.cc
@@ -63,11 +63,13 @@
   part->verity.hash_tree_extent = ExtentForBytes(
       hashtree.hash_block_size, hashtree.tree_offset, hashtree.tree_size);
 
-  part->verity.fec_data_extent =
-      ExtentForBytes(hashtree.data_block_size, 0, hashtree.fec_offset);
-  part->verity.fec_extent = ExtentForBytes(
-      hashtree.data_block_size, hashtree.fec_offset, hashtree.fec_size);
-  part->verity.fec_roots = hashtree.fec_num_roots;
+  if (!part->disable_fec_computation) {
+    part->verity.fec_data_extent =
+        ExtentForBytes(hashtree.data_block_size, 0, hashtree.fec_offset);
+    part->verity.fec_extent = ExtentForBytes(
+        hashtree.data_block_size, hashtree.fec_offset, hashtree.fec_size);
+    part->verity.fec_roots = hashtree.fec_num_roots;
+  }
   return true;
 }
 
@@ -205,7 +207,8 @@
               ExtentForRange(hash_start_block, tree_size / block_size);
         }
         fec_ecc_metadata ecc_data;
-        if (fh.get_ecc_metadata(ecc_data) && ecc_data.valid) {
+        if (!part.disable_fec_computation && fh.get_ecc_metadata(ecc_data) &&
+            ecc_data.valid) {
           TEST_AND_RETURN_FALSE(block_size == FEC_BLOCKSIZE);
           part.verity.fec_data_extent = ExtentForRange(0, ecc_data.blocks);
           part.verity.fec_extent =
diff --git a/payload_generator/payload_generation_config_android_unittest.cc b/payload_generator/payload_generation_config_android_unittest.cc
index 53378c2..44eaf55 100644
--- a/payload_generator/payload_generation_config_android_unittest.cc
+++ b/payload_generator/payload_generation_config_android_unittest.cc
@@ -160,6 +160,24 @@
   EXPECT_EQ(2u, verity.fec_roots);
 }
 
+TEST_F(PayloadGenerationConfigAndroidTest, LoadVerityConfigDisableFecTest) {
+  brillo::Blob part = GetAVBPartition();
+  test_utils::WriteFileVector(temp_file_.path(), part);
+  image_config_.partitions[0].disable_fec_computation = true;
+  EXPECT_TRUE(image_config_.LoadImageSize());
+  EXPECT_TRUE(image_config_.partitions[0].OpenFilesystem());
+  EXPECT_TRUE(image_config_.LoadVerityConfig());
+  const VerityConfig& verity = image_config_.partitions[0].verity;
+  EXPECT_FALSE(verity.IsEmpty());
+  EXPECT_EQ(ExtentForRange(0, 2), verity.hash_tree_data_extent);
+  EXPECT_EQ(ExtentForRange(2, 1), verity.hash_tree_extent);
+  EXPECT_EQ("sha1", verity.hash_tree_algorithm);
+  brillo::Blob salt(kHashTreeSalt, std::end(kHashTreeSalt));
+  EXPECT_EQ(salt, verity.hash_tree_salt);
+  EXPECT_EQ(0u, verity.fec_data_extent.num_blocks());
+  EXPECT_EQ(0u, verity.fec_extent.num_blocks());
+}
+
 TEST_F(PayloadGenerationConfigAndroidTest,
        LoadVerityConfigInvalidHashTreeTest) {
   brillo::Blob part = GetAVBPartition();
diff --git a/payload_generator/payload_generation_config_unittest.cc b/payload_generator/payload_generation_config_unittest.cc
index 70a3df3..aca9655 100644
--- a/payload_generator/payload_generation_config_unittest.cc
+++ b/payload_generator/payload_generation_config_unittest.cc
@@ -59,7 +59,7 @@
   ASSERT_TRUE(
       store.LoadFromString("super_partition_groups=group_a group_b\n"
                            "group_a_size=3221225472\n"
-                           "group_a_partition_list=system product_services\n"
+                           "group_a_partition_list=system system_ext\n"
                            "group_b_size=2147483648\n"
                            "group_b_partition_list=vendor\n"));
   EXPECT_TRUE(image_config.LoadDynamicPartitionMetadata(store));
@@ -72,7 +72,7 @@
   EXPECT_EQ(3221225472u, group_a.size());
   ASSERT_EQ(2, group_a.partition_names_size());
   EXPECT_EQ("system", group_a.partition_names(0));
-  EXPECT_EQ("product_services", group_a.partition_names(1));
+  EXPECT_EQ("system_ext", group_a.partition_names(1));
 
   const auto& group_b = image_config.dynamic_partition_metadata->groups(1);
   EXPECT_EQ("group_b", group_b.name());
@@ -108,17 +108,17 @@
 
   PartitionConfig system("system");
   system.size = 2147483648u;
-  PartitionConfig product_services("product_services");
-  product_services.size = 1073741824u;
+  PartitionConfig system_ext("system_ext");
+  system_ext.size = 1073741824u;
 
   image_config.partitions.push_back(std::move(system));
-  image_config.partitions.push_back(std::move(product_services));
+  image_config.partitions.push_back(std::move(system_ext));
 
   brillo::KeyValueStore store;
   ASSERT_TRUE(
       store.LoadFromString("super_partition_groups=foo\n"
                            "foo_size=3221225472\n"
-                           "foo_partition_list=system product_services\n"));
+                           "foo_partition_list=system system_ext\n"));
   EXPECT_TRUE(image_config.LoadDynamicPartitionMetadata(store));
   EXPECT_NE(nullptr, image_config.dynamic_partition_metadata);
 
diff --git a/payload_generator/payload_properties_unittest.cc b/payload_generator/payload_properties_unittest.cc
index db3902c..e0072fc 100644
--- a/payload_generator/payload_properties_unittest.cc
+++ b/payload_generator/payload_properties_unittest.cc
@@ -98,7 +98,7 @@
     EXPECT_TRUE(strategy->GenerateOperations(
         config, old_part, new_part, &blob_file_writer, &aops));
 
-    payload.AddPartition(old_part, new_part, aops);
+    payload.AddPartition(old_part, new_part, aops, {});
 
     uint64_t metadata_size;
     EXPECT_TRUE(payload.WritePayload(
diff --git a/payload_generator/payload_signer.cc b/payload_generator/payload_signer.cc
index 420329f..9a44f94 100644
--- a/payload_generator/payload_signer.cc
+++ b/payload_generator/payload_signer.cc
@@ -18,6 +18,7 @@
 
 #include <endian.h>
 
+#include <memory>
 #include <utility>
 
 #include <base/logging.h>
@@ -28,6 +29,7 @@
 #include <openssl/err.h>
 #include <openssl/pem.h>
 
+#include "update_engine/common/constants.h"
 #include "update_engine/common/hash_calculator.h"
 #include "update_engine/common/subprocess.h"
 #include "update_engine/common/utils.h"
@@ -45,45 +47,49 @@
 namespace chromeos_update_engine {
 
 namespace {
-
-// The payload verifier will check all the signatures included in the payload
-// regardless of the version field. Old version of the verifier require the
-// version field to be included and be 1.
-const uint32_t kSignatureMessageLegacyVersion = 1;
-
 // Given raw |signatures|, packs them into a protobuf and serializes it into a
-// binary blob. Returns true on success, false otherwise.
-bool ConvertSignatureToProtobufBlob(const vector<brillo::Blob>& signatures,
-                                    brillo::Blob* out_signature_blob) {
+// string. Returns true on success, false otherwise.
+bool ConvertSignaturesToProtobuf(const vector<brillo::Blob>& signatures,
+                                 const vector<size_t>& padded_signature_sizes,
+                                 string* out_serialized_signature) {
+  TEST_AND_RETURN_FALSE(signatures.size() == padded_signature_sizes.size());
   // Pack it into a protobuf
   Signatures out_message;
-  for (const brillo::Blob& signature : signatures) {
-    Signatures_Signature* sig_message = out_message.add_signatures();
-    // Set all the signatures with the same version number.
-    sig_message->set_version(kSignatureMessageLegacyVersion);
-    sig_message->set_data(signature.data(), signature.size());
+  for (size_t i = 0; i < signatures.size(); i++) {
+    const auto& signature = signatures[i];
+    const auto& padded_signature_size = padded_signature_sizes[i];
+    TEST_AND_RETURN_FALSE(padded_signature_size >= signature.size());
+    Signatures::Signature* sig_message = out_message.add_signatures();
+    // Skip assigning the same version number because we don't need to be
+    // compatible with old major version 1 client anymore.
+
+    // TODO(Xunchang) don't need to set the unpadded_signature_size field for
+    // RSA key signed signatures.
+    sig_message->set_unpadded_signature_size(signature.size());
+    brillo::Blob padded_signature = signature;
+    padded_signature.insert(
+        padded_signature.end(), padded_signature_size - signature.size(), 0);
+    sig_message->set_data(padded_signature.data(), padded_signature.size());
   }
 
   // Serialize protobuf
-  string serialized;
-  TEST_AND_RETURN_FALSE(out_message.AppendToString(&serialized));
-  out_signature_blob->insert(
-      out_signature_blob->end(), serialized.begin(), serialized.end());
-  LOG(INFO) << "Signature blob size: " << out_signature_blob->size();
+  TEST_AND_RETURN_FALSE(
+      out_message.SerializeToString(out_serialized_signature));
+  LOG(INFO) << "Signature blob size: " << out_serialized_signature->size();
   return true;
 }
 
-// Given an unsigned payload under |payload_path| and the |signature_blob| and
-// |metadata_signature_blob| generates an updated payload that includes the
+// Given an unsigned payload under |payload_path| and the |payload_signature|
+// and |metadata_signature| generates an updated payload that includes the
 // signatures. It populates |out_metadata_size| with the size of the final
-// manifest after adding the dummy signature operation, and
+// manifest after adding the fake signature operation, and
 // |out_signatures_offset| with the expected offset for the new blob, and
-// |out_metadata_signature_size| which will be size of |metadata_signature_blob|
+// |out_metadata_signature_size| which will be size of |metadata_signature|
 // if the payload major version supports metadata signature, 0 otherwise.
 // Returns true on success, false otherwise.
 bool AddSignatureBlobToPayload(const string& payload_path,
-                               const brillo::Blob& signature_blob,
-                               const brillo::Blob& metadata_signature_blob,
+                               const string& payload_signature,
+                               const string& metadata_signature,
                                brillo::Blob* out_payload,
                                uint64_t* out_metadata_size,
                                uint32_t* out_metadata_signature_size,
@@ -99,7 +105,7 @@
   uint32_t metadata_signature_size =
       payload_metadata.GetMetadataSignatureSize();
   // Write metadata signature size in header.
-  uint32_t metadata_signature_size_be = htobe32(metadata_signature_blob.size());
+  uint32_t metadata_signature_size_be = htobe32(metadata_signature.size());
   memcpy(payload.data() + manifest_offset,
          &metadata_signature_size_be,
          sizeof(metadata_signature_size_be));
@@ -108,9 +114,9 @@
   payload.erase(payload.begin() + metadata_size,
                 payload.begin() + metadata_size + metadata_signature_size);
   payload.insert(payload.begin() + metadata_size,
-                 metadata_signature_blob.begin(),
-                 metadata_signature_blob.end());
-  metadata_signature_size = metadata_signature_blob.size();
+                 metadata_signature.begin(),
+                 metadata_signature.end());
+  metadata_signature_size = metadata_signature.size();
   LOG(INFO) << "Metadata signature size: " << metadata_signature_size;
 
   DeltaArchiveManifest manifest;
@@ -122,10 +128,10 @@
     // contents. We don't allow the manifest to change if there is already an op
     // present, because that might invalidate previously generated
     // hashes/signatures.
-    if (manifest.signatures_size() != signature_blob.size()) {
+    if (manifest.signatures_size() != payload_signature.size()) {
       LOG(ERROR) << "Attempt to insert different signature sized blob. "
                  << "(current:" << manifest.signatures_size()
-                 << "new:" << signature_blob.size() << ")";
+                 << "new:" << payload_signature.size() << ")";
       return false;
     }
 
@@ -134,7 +140,7 @@
     // Updates the manifest to include the signature operation.
     PayloadSigner::AddSignatureToManifest(
         payload.size() - metadata_size - metadata_signature_size,
-        signature_blob.size(),
+        payload_signature.size(),
         &manifest);
 
     // Updates the payload to include the new manifest.
@@ -160,8 +166,8 @@
   LOG(INFO) << "Signature Blob Offset: " << signatures_offset;
   payload.resize(signatures_offset);
   payload.insert(payload.begin() + signatures_offset,
-                 signature_blob.begin(),
-                 signature_blob.end());
+                 payload_signature.begin(),
+                 payload_signature.end());
 
   *out_payload = std::move(payload);
   *out_metadata_size = metadata_size;
@@ -201,8 +207,35 @@
   return true;
 }
 
+std::unique_ptr<EVP_PKEY, decltype(&EVP_PKEY_free)> CreatePrivateKeyFromPath(
+    const string& private_key_path) {
+  FILE* fprikey = fopen(private_key_path.c_str(), "rb");
+  if (!fprikey) {
+    PLOG(ERROR) << "Failed to read " << private_key_path;
+    return {nullptr, nullptr};
+  }
+
+  auto private_key = std::unique_ptr<EVP_PKEY, decltype(&EVP_PKEY_free)>(
+      PEM_read_PrivateKey(fprikey, nullptr, nullptr, nullptr), EVP_PKEY_free);
+  fclose(fprikey);
+  return private_key;
+}
+
 }  // namespace
 
+bool PayloadSigner::GetMaximumSignatureSize(const string& private_key_path,
+                                            size_t* signature_size) {
+  *signature_size = 0;
+  auto private_key = CreatePrivateKeyFromPath(private_key_path);
+  if (!private_key) {
+    LOG(ERROR) << "Failed to create private key from " << private_key_path;
+    return false;
+  }
+
+  *signature_size = EVP_PKEY_size(private_key.get());
+  return true;
+}
+
 void PayloadSigner::AddSignatureToManifest(uint64_t signature_blob_offset,
                                            uint64_t signature_blob_length,
                                            DeltaArchiveManifest* manifest) {
@@ -236,21 +269,22 @@
                                                  signatures_offset,
                                                  &payload_hash,
                                                  &metadata_hash));
-  brillo::Blob signature_blob(payload.begin() + signatures_offset,
-                              payload.end());
+  string signature(payload.begin() + signatures_offset, payload.end());
   string public_key;
   TEST_AND_RETURN_FALSE(utils::ReadFile(public_key_path, &public_key));
-  TEST_AND_RETURN_FALSE(PayloadVerifier::PadRSA2048SHA256Hash(&payload_hash));
-  TEST_AND_RETURN_FALSE(PayloadVerifier::VerifySignature(
-      signature_blob, public_key, payload_hash));
+  TEST_AND_RETURN_FALSE(payload_hash.size() == kSHA256Size);
+
+  auto payload_verifier = PayloadVerifier::CreateInstance(public_key);
+  TEST_AND_RETURN_FALSE(payload_verifier != nullptr);
+
+  TEST_AND_RETURN_FALSE(
+      payload_verifier->VerifySignature(signature, payload_hash));
   if (metadata_signature_size) {
-    signature_blob.assign(
-        payload.begin() + metadata_size,
-        payload.begin() + metadata_size + metadata_signature_size);
+    signature.assign(payload.begin() + metadata_size,
+                     payload.begin() + metadata_size + metadata_signature_size);
+    TEST_AND_RETURN_FALSE(metadata_hash.size() == kSHA256Size);
     TEST_AND_RETURN_FALSE(
-        PayloadVerifier::PadRSA2048SHA256Hash(&metadata_hash));
-    TEST_AND_RETURN_FALSE(PayloadVerifier::VerifySignature(
-        signature_blob, public_key, metadata_hash));
+        payload_verifier->VerifySignature(signature, metadata_hash));
   }
   return true;
 }
@@ -260,49 +294,97 @@
                              brillo::Blob* out_signature) {
   LOG(INFO) << "Signing hash with private key: " << private_key_path;
   // We expect unpadded SHA256 hash coming in
-  TEST_AND_RETURN_FALSE(hash.size() == 32);
-  brillo::Blob padded_hash(hash);
-  PayloadVerifier::PadRSA2048SHA256Hash(&padded_hash);
-
+  TEST_AND_RETURN_FALSE(hash.size() == kSHA256Size);
   // The code below executes the equivalent of:
   //
   // openssl rsautl -raw -sign -inkey |private_key_path|
   //   -in |padded_hash| -out |out_signature|
 
-  FILE* fprikey = fopen(private_key_path.c_str(), "rb");
-  TEST_AND_RETURN_FALSE(fprikey != nullptr);
-  RSA* rsa = PEM_read_RSAPrivateKey(fprikey, nullptr, nullptr, nullptr);
-  fclose(fprikey);
-  TEST_AND_RETURN_FALSE(rsa != nullptr);
-  brillo::Blob signature(RSA_size(rsa));
-  ssize_t signature_size = RSA_private_encrypt(padded_hash.size(),
-                                               padded_hash.data(),
-                                               signature.data(),
-                                               rsa,
-                                               RSA_NO_PADDING);
-  RSA_free(rsa);
-  if (signature_size < 0) {
-    LOG(ERROR) << "Signing hash failed: "
-               << ERR_error_string(ERR_get_error(), nullptr);
+  auto private_key = CreatePrivateKeyFromPath(private_key_path);
+  if (!private_key) {
+    LOG(ERROR) << "Failed to create private key from " << private_key_path;
     return false;
   }
-  TEST_AND_RETURN_FALSE(static_cast<size_t>(signature_size) ==
-                        signature.size());
+
+  int key_type = EVP_PKEY_id(private_key.get());
+  brillo::Blob signature;
+  if (key_type == EVP_PKEY_RSA) {
+    // TODO(b/158580694): Switch back to get0 version and remove manual freeing
+    // of the object once the bug is resolved or gale has been moved to
+    // informational.
+    RSA* rsa = EVP_PKEY_get1_RSA(private_key.get());
+    TEST_AND_RETURN_FALSE(rsa != nullptr);
+
+    brillo::Blob padded_hash = hash;
+    PayloadVerifier::PadRSASHA256Hash(&padded_hash, RSA_size(rsa));
+
+    signature.resize(RSA_size(rsa));
+    ssize_t signature_size = RSA_private_encrypt(padded_hash.size(),
+                                                 padded_hash.data(),
+                                                 signature.data(),
+                                                 rsa,
+                                                 RSA_NO_PADDING);
+    if (signature_size < 0) {
+      LOG(ERROR) << "Signing hash failed: "
+                 << ERR_error_string(ERR_get_error(), nullptr);
+      RSA_free(rsa);
+      return false;
+    }
+    RSA_free(rsa);
+    TEST_AND_RETURN_FALSE(static_cast<size_t>(signature_size) ==
+                          signature.size());
+  } else if (key_type == EVP_PKEY_EC) {
+    // TODO(b/158580694): Switch back to get0 version and remove manual freeing
+    // of the object once the bug is resolved or gale has been moved to
+    // informational.
+    EC_KEY* ec_key = EVP_PKEY_get1_EC_KEY(private_key.get());
+    TEST_AND_RETURN_FALSE(ec_key != nullptr);
+
+    signature.resize(ECDSA_size(ec_key));
+    unsigned int signature_size;
+    if (ECDSA_sign(0,
+                   hash.data(),
+                   hash.size(),
+                   signature.data(),
+                   &signature_size,
+                   ec_key) != 1) {
+      LOG(ERROR) << "Signing hash failed: "
+                 << ERR_error_string(ERR_get_error(), nullptr);
+      EC_KEY_free(ec_key);
+      return false;
+    }
+    EC_KEY_free(ec_key);
+
+    // NIST P-256
+    LOG(ERROR) << "signature max size " << signature.size() << " size "
+               << signature_size;
+    TEST_AND_RETURN_FALSE(signature.size() >= signature_size);
+    signature.resize(signature_size);
+  } else {
+    LOG(ERROR) << "key_type " << key_type << " isn't supported for signing";
+    return false;
+  }
   out_signature->swap(signature);
   return true;
 }
 
 bool PayloadSigner::SignHashWithKeys(const brillo::Blob& hash_data,
                                      const vector<string>& private_key_paths,
-                                     brillo::Blob* out_signature_blob) {
+                                     string* out_serialized_signature) {
   vector<brillo::Blob> signatures;
+  vector<size_t> padded_signature_sizes;
   for (const string& path : private_key_paths) {
     brillo::Blob signature;
     TEST_AND_RETURN_FALSE(SignHash(hash_data, path, &signature));
     signatures.push_back(signature);
+
+    size_t padded_signature_size;
+    TEST_AND_RETURN_FALSE(
+        GetMaximumSignatureSize(path, &padded_signature_size));
+    padded_signature_sizes.push_back(padded_signature_size);
   }
-  TEST_AND_RETURN_FALSE(
-      ConvertSignatureToProtobufBlob(signatures, out_signature_blob));
+  TEST_AND_RETURN_FALSE(ConvertSignaturesToProtobuf(
+      signatures, padded_signature_sizes, out_serialized_signature));
   return true;
 }
 
@@ -311,7 +393,7 @@
                                 const uint64_t metadata_size,
                                 const uint32_t metadata_signature_size,
                                 const uint64_t signatures_offset,
-                                brillo::Blob* out_signature_blob) {
+                                string* out_serialized_signature) {
   brillo::Blob payload;
   TEST_AND_RETURN_FALSE(utils::ReadFile(unsigned_payload_path, &payload));
   brillo::Blob hash_data;
@@ -322,16 +404,16 @@
                                                  &hash_data,
                                                  nullptr));
   TEST_AND_RETURN_FALSE(
-      SignHashWithKeys(hash_data, private_key_paths, out_signature_blob));
+      SignHashWithKeys(hash_data, private_key_paths, out_serialized_signature));
   return true;
 }
 
 bool PayloadSigner::SignatureBlobLength(const vector<string>& private_key_paths,
                                         uint64_t* out_length) {
   DCHECK(out_length);
-  brillo::Blob x_blob(1, 'x'), hash_blob, sig_blob;
-  TEST_AND_RETURN_FALSE(
-      HashCalculator::RawHashOfBytes(x_blob.data(), x_blob.size(), &hash_blob));
+  brillo::Blob hash_blob;
+  TEST_AND_RETURN_FALSE(HashCalculator::RawHashOfData({'x'}, &hash_blob));
+  string sig_blob;
   TEST_AND_RETURN_FALSE(
       SignHashWithKeys(hash_blob, private_key_paths, &sig_blob));
   *out_length = sig_blob.size();
@@ -339,7 +421,7 @@
 }
 
 bool PayloadSigner::HashPayloadForSigning(const string& payload_path,
-                                          const vector<int>& signature_sizes,
+                                          const vector<size_t>& signature_sizes,
                                           brillo::Blob* out_payload_hash_data,
                                           brillo::Blob* out_metadata_hash) {
   // Create a signature blob with signatures filled with 0.
@@ -348,17 +430,17 @@
   for (int signature_size : signature_sizes) {
     signatures.emplace_back(signature_size, 0);
   }
-  brillo::Blob signature_blob;
+  string signature;
   TEST_AND_RETURN_FALSE(
-      ConvertSignatureToProtobufBlob(signatures, &signature_blob));
+      ConvertSignaturesToProtobuf(signatures, signature_sizes, &signature));
 
   brillo::Blob payload;
   uint64_t metadata_size, signatures_offset;
   uint32_t metadata_signature_size;
   // Prepare payload for hashing.
   TEST_AND_RETURN_FALSE(AddSignatureBlobToPayload(payload_path,
-                                                  signature_blob,
-                                                  signature_blob,
+                                                  signature,
+                                                  signature,
                                                   &payload,
                                                   &metadata_size,
                                                   &metadata_signature_size,
@@ -374,6 +456,7 @@
 
 bool PayloadSigner::AddSignatureToPayload(
     const string& payload_path,
+    const vector<size_t>& padded_signature_sizes,
     const vector<brillo::Blob>& payload_signatures,
     const vector<brillo::Blob>& metadata_signatures,
     const string& signed_payload_path,
@@ -381,19 +464,19 @@
   // TODO(petkov): Reduce memory usage -- the payload is manipulated in memory.
 
   // Loads the payload and adds the signature op to it.
-  brillo::Blob signature_blob, metadata_signature_blob;
-  TEST_AND_RETURN_FALSE(
-      ConvertSignatureToProtobufBlob(payload_signatures, &signature_blob));
+  string payload_signature, metadata_signature;
+  TEST_AND_RETURN_FALSE(ConvertSignaturesToProtobuf(
+      payload_signatures, padded_signature_sizes, &payload_signature));
   if (!metadata_signatures.empty()) {
-    TEST_AND_RETURN_FALSE(ConvertSignatureToProtobufBlob(
-        metadata_signatures, &metadata_signature_blob));
+    TEST_AND_RETURN_FALSE(ConvertSignaturesToProtobuf(
+        metadata_signatures, padded_signature_sizes, &metadata_signature));
   }
   brillo::Blob payload;
   uint64_t signatures_offset;
   uint32_t metadata_signature_size;
   TEST_AND_RETURN_FALSE(AddSignatureBlobToPayload(payload_path,
-                                                  signature_blob,
-                                                  metadata_signature_blob,
+                                                  payload_signature,
+                                                  metadata_signature,
                                                   &payload,
                                                   out_metadata_size,
                                                   &metadata_signature_size,
diff --git a/payload_generator/payload_signer.h b/payload_generator/payload_signer.h
index 71f4983..9676b71 100644
--- a/payload_generator/payload_signer.h
+++ b/payload_generator/payload_signer.h
@@ -51,27 +51,27 @@
                        brillo::Blob* out_signature);
 
   // Sign |hash_data| blob with all private keys in |private_key_paths|, then
-  // convert the signatures to protobuf blob.
+  // convert the signatures to serialized protobuf.
   static bool SignHashWithKeys(
       const brillo::Blob& hash_data,
       const std::vector<std::string>& private_key_paths,
-      brillo::Blob* out_signature_blob);
+      std::string* out_serialized_signature);
 
   // Given an unsigned payload in |unsigned_payload_path|, private keys in
   // |private_key_path|, metadata size in |metadata_size|, metadata signature
   // size in |metadata_signature_size| and signatures offset in
   // |signatures_offset|, calculates the payload signature blob into
-  // |out_signature_blob|. Note that the payload must already have an
-  // updated manifest that includes the dummy signature op and correct metadata
+  // |out_serialized_signature|. Note that the payload must already have an
+  // updated manifest that includes the fake signature op and correct metadata
   // signature size in header. Returns true on success, false otherwise.
   static bool SignPayload(const std::string& unsigned_payload_path,
                           const std::vector<std::string>& private_key_paths,
                           const uint64_t metadata_size,
                           const uint32_t metadata_signature_size,
                           const uint64_t signatures_offset,
-                          brillo::Blob* out_signature_blob);
+                          std::string* out_serialized_signature);
 
-  // Returns the length of out_signature_blob that will result in a call
+  // Returns the length of out_serialized_signature that will result in a call
   // to SignPayload with the given private keys. Returns true on success.
   static bool SignatureBlobLength(
       const std::vector<std::string>& private_key_paths, uint64_t* out_length);
@@ -88,11 +88,11 @@
   //
   // The changes to payload are not preserved or written to disk.
   static bool HashPayloadForSigning(const std::string& payload_path,
-                                    const std::vector<int>& signature_sizes,
+                                    const std::vector<size_t>& signature_sizes,
                                     brillo::Blob* out_payload_hash_data,
                                     brillo::Blob* out_metadata_hash);
 
-  // Given an unsigned payload in |payload_path| (with no dummy signature op)
+  // Given an unsigned payload in |payload_path| (with no fake signature op)
   // and the raw |payload_signatures| and |metadata_signatures| updates the
   // payload to include the signature thus turning it into a signed payload. The
   // new payload is stored in |signed_payload_path|. |payload_path| and
@@ -102,6 +102,7 @@
   // otherwise.
   static bool AddSignatureToPayload(
       const std::string& payload_path,
+      const std::vector<size_t>& padded_signature_sizes,
       const std::vector<brillo::Blob>& payload_signatures,
       const std::vector<brillo::Blob>& metadata_signatures,
       const std::string& signed_payload_path,
@@ -116,6 +117,16 @@
                                    const std::string& private_key_path,
                                    std::string* out_signature);
 
+  static bool ExtractPayloadProperties(const std::string& payload_path,
+                                       brillo::KeyValueStore* properties);
+
+  // This function calculates the maximum size, in bytes, of a signature signed
+  // by private_key_path. For an RSA key, this returns the number of bytes
+  // needed to represent the modulus. For an EC key, this returns the maximum
+  // size of a DER-encoded ECDSA signature.
+  static bool GetMaximumSignatureSize(const std::string& private_key_path,
+                                      size_t* signature_size);
+
  private:
   // This should never be constructed
   DISALLOW_IMPLICIT_CONSTRUCTORS(PayloadSigner);
diff --git a/payload_generator/payload_signer_unittest.cc b/payload_generator/payload_signer_unittest.cc
index f7f9c69..fe62997 100644
--- a/payload_generator/payload_signer_unittest.cc
+++ b/payload_generator/payload_signer_unittest.cc
@@ -45,6 +45,10 @@
 const char* kUnittestPublicKeyPath = "unittest_key.pub.pem";
 const char* kUnittestPrivateKey2Path = "unittest_key2.pem";
 const char* kUnittestPublicKey2Path = "unittest_key2.pub.pem";
+const char* kUnittestPrivateKeyRSA4096Path = "unittest_key_RSA4096.pem";
+const char* kUnittestPublicKeyRSA4096Path = "unittest_key_RSA4096.pub.pem";
+const char* kUnittestPrivateKeyECPath = "unittest_key_EC.pem";
+const char* kUnittestPublicKeyECPath = "unittest_key_EC.pub.pem";
 
 // Some data and its corresponding hash and signature:
 const char kDataToSign[] = "This is some data to sign.";
@@ -87,44 +91,34 @@
     0x43, 0xb9, 0xab, 0x7d};
 
 namespace {
-void SignSampleData(brillo::Blob* out_signature_blob,
-                    const vector<string>& private_keys) {
-  brillo::Blob data_blob(std::begin(kDataToSign),
-                         std::begin(kDataToSign) + strlen(kDataToSign));
+void SignSampleData(string* out_signature, const vector<string>& private_keys) {
   uint64_t length = 0;
   EXPECT_TRUE(PayloadSigner::SignatureBlobLength(private_keys, &length));
   EXPECT_GT(length, 0U);
   brillo::Blob hash_blob;
   EXPECT_TRUE(HashCalculator::RawHashOfBytes(
-      data_blob.data(), data_blob.size(), &hash_blob));
-  EXPECT_TRUE(PayloadSigner::SignHashWithKeys(
-      hash_blob, private_keys, out_signature_blob));
-  EXPECT_EQ(length, out_signature_blob->size());
+      kDataToSign, strlen(kDataToSign), &hash_blob));
+  EXPECT_TRUE(
+      PayloadSigner::SignHashWithKeys(hash_blob, private_keys, out_signature));
+  EXPECT_EQ(length, out_signature->size());
 }
 }  // namespace
 
 class PayloadSignerTest : public ::testing::Test {
  protected:
-  void SetUp() override {
-    PayloadVerifier::PadRSA2048SHA256Hash(&padded_hash_data_);
-  }
-
-  brillo::Blob padded_hash_data_{std::begin(kDataHash), std::end(kDataHash)};
+  brillo::Blob hash_data_{std::begin(kDataHash), std::end(kDataHash)};
 };
 
 TEST_F(PayloadSignerTest, SignSimpleTextTest) {
-  brillo::Blob signature_blob;
-  SignSampleData(&signature_blob,
-                 {GetBuildArtifactsPath(kUnittestPrivateKeyPath)});
+  string signature;
+  SignSampleData(&signature, {GetBuildArtifactsPath(kUnittestPrivateKeyPath)});
 
   // Check the signature itself
   Signatures signatures;
-  EXPECT_TRUE(
-      signatures.ParseFromArray(signature_blob.data(), signature_blob.size()));
+  EXPECT_TRUE(signatures.ParseFromString(signature));
   EXPECT_EQ(1, signatures.signatures_size());
-  const Signatures_Signature& signature = signatures.signatures(0);
-  EXPECT_EQ(1U, signature.version());
-  const string& sig_data = signature.data();
+  const Signatures::Signature& sig = signatures.signatures(0);
+  const string& sig_data = sig.data();
   ASSERT_EQ(base::size(kDataSignature), sig_data.size());
   for (size_t i = 0; i < base::size(kDataSignature); i++) {
     EXPECT_EQ(kDataSignature[i], static_cast<uint8_t>(sig_data[i]));
@@ -132,38 +126,44 @@
 }
 
 TEST_F(PayloadSignerTest, VerifyAllSignatureTest) {
-  brillo::Blob signature_blob;
-  SignSampleData(&signature_blob,
+  string signature;
+  SignSampleData(&signature,
                  {GetBuildArtifactsPath(kUnittestPrivateKeyPath),
-                  GetBuildArtifactsPath(kUnittestPrivateKey2Path)});
+                  GetBuildArtifactsPath(kUnittestPrivateKey2Path),
+                  GetBuildArtifactsPath(kUnittestPrivateKeyRSA4096Path),
+                  GetBuildArtifactsPath(kUnittestPrivateKeyECPath)});
 
   // Either public key should pass the verification.
-  string public_key;
-  EXPECT_TRUE(utils::ReadFile(GetBuildArtifactsPath(kUnittestPublicKeyPath),
-                              &public_key));
-  EXPECT_TRUE(PayloadVerifier::VerifySignature(
-      signature_blob, public_key, padded_hash_data_));
-  EXPECT_TRUE(utils::ReadFile(GetBuildArtifactsPath(kUnittestPublicKey2Path),
-                              &public_key));
-  EXPECT_TRUE(PayloadVerifier::VerifySignature(
-      signature_blob, public_key, padded_hash_data_));
+  for (const auto& path : {kUnittestPublicKeyPath,
+                           kUnittestPublicKey2Path,
+                           kUnittestPublicKeyRSA4096Path,
+                           kUnittestPublicKeyECPath}) {
+    string public_key;
+    EXPECT_TRUE(utils::ReadFile(GetBuildArtifactsPath(path), &public_key));
+    auto payload_verifier = PayloadVerifier::CreateInstance(public_key);
+    EXPECT_TRUE(payload_verifier != nullptr);
+    EXPECT_TRUE(payload_verifier->VerifySignature(signature, hash_data_));
+  }
 }
 
 TEST_F(PayloadSignerTest, VerifySignatureTest) {
-  brillo::Blob signature_blob;
-  SignSampleData(&signature_blob,
-                 {GetBuildArtifactsPath(kUnittestPrivateKeyPath)});
+  string signature;
+  SignSampleData(&signature, {GetBuildArtifactsPath(kUnittestPrivateKeyPath)});
 
   string public_key;
   EXPECT_TRUE(utils::ReadFile(GetBuildArtifactsPath(kUnittestPublicKeyPath),
                               &public_key));
-  EXPECT_TRUE(PayloadVerifier::VerifySignature(
-      signature_blob, public_key, padded_hash_data_));
+  auto payload_verifier = PayloadVerifier::CreateInstance(public_key);
+  EXPECT_TRUE(payload_verifier != nullptr);
+  EXPECT_TRUE(payload_verifier->VerifySignature(signature, hash_data_));
+
   // Passing the invalid key should fail the verification.
+  public_key.clear();
   EXPECT_TRUE(utils::ReadFile(GetBuildArtifactsPath(kUnittestPublicKey2Path),
                               &public_key));
-  EXPECT_TRUE(PayloadVerifier::VerifySignature(
-      signature_blob, public_key, padded_hash_data_));
+  payload_verifier = PayloadVerifier::CreateInstance(public_key);
+  EXPECT_TRUE(payload_verifier != nullptr);
+  EXPECT_FALSE(payload_verifier->VerifySignature(signature, hash_data_));
 }
 
 TEST_F(PayloadSignerTest, SkipMetadataSignatureTest) {
@@ -175,7 +175,7 @@
   uint64_t metadata_size;
   EXPECT_TRUE(payload.WritePayload(
       payload_file.path(), "/dev/null", "", &metadata_size));
-  const vector<int> sizes = {256};
+  const vector<size_t> sizes = {256};
   brillo::Blob unsigned_payload_hash, unsigned_metadata_hash;
   EXPECT_TRUE(PayloadSigner::HashPayloadForSigning(payload_file.path(),
                                                    sizes,
diff --git a/payload_generator/squashfs_filesystem.cc b/payload_generator/squashfs_filesystem.cc
index eb4fda3..6152d7d 100644
--- a/payload_generator/squashfs_filesystem.cc
+++ b/payload_generator/squashfs_filesystem.cc
@@ -275,7 +275,7 @@
     auto last = std::unique(zlib_blks.begin(), zlib_blks.end());
     zlib_blks.erase(last, zlib_blks.end());
 
-    // Sanity check. Make sure zlib blocks are not overlapping.
+    // Make sure zlib blocks are not overlapping.
     auto result = std::adjacent_find(
         zlib_blks.begin(),
         zlib_blks.end(),