diff --git a/debuggerd/crash_dump.cpp b/debuggerd/crash_dump.cpp
index 5280121..b3e81b0 100644
--- a/debuggerd/crash_dump.cpp
+++ b/debuggerd/crash_dump.cpp
@@ -299,11 +299,8 @@
       process_info->abort_msg_address = crash_info->data.s.abort_msg_address;
       *siginfo = crash_info->data.s.siginfo;
       if (signal_has_si_addr(siginfo)) {
-        // Make a copy of the ucontext field because otherwise it is not aligned enough (due to
-        // being in a packed struct) and clang complains about that.
-        ucontext_t ucontext = crash_info->data.s.ucontext;
         process_info->has_fault_address = true;
-        process_info->fault_address = get_fault_address(siginfo, &ucontext);
+        process_info->fault_address = reinterpret_cast<uintptr_t>(siginfo->si_addr);
       }
       regs->reset(unwindstack::Regs::CreateFromUcontext(unwindstack::Regs::CurrentArch(),
                                                         &crash_info->data.s.ucontext));
diff --git a/debuggerd/handler/debuggerd_handler.cpp b/debuggerd/handler/debuggerd_handler.cpp
index 121a074..85ffc98 100644
--- a/debuggerd/handler/debuggerd_handler.cpp
+++ b/debuggerd/handler/debuggerd_handler.cpp
@@ -167,7 +167,7 @@
  * mutex is being held, so we don't want to use any libc functions that
  * could allocate memory or hold a lock.
  */
-static void log_signal_summary(const siginfo_t* info, const ucontext_t* ucontext) {
+static void log_signal_summary(const siginfo_t* info) {
   char thread_name[MAX_TASK_NAME_LEN + 1];  // one more for termination
   if (prctl(PR_GET_NAME, reinterpret_cast<unsigned long>(thread_name), 0, 0, 0) != 0) {
     strcpy(thread_name, "<name unknown>");
@@ -186,8 +186,7 @@
   // Many signals don't have an address or sender.
   char addr_desc[32] = "";  // ", fault addr 0x1234"
   if (signal_has_si_addr(info)) {
-    async_safe_format_buffer(addr_desc, sizeof(addr_desc), ", fault addr %p",
-                             reinterpret_cast<void*>(get_fault_address(info, ucontext)));
+    async_safe_format_buffer(addr_desc, sizeof(addr_desc), ", fault addr %p", info->si_addr);
   }
   pid_t self_pid = __getpid();
   char sender_desc[32] = {};  // " from pid 1234, uid 666"
@@ -544,7 +543,7 @@
     return;
   }
 
-  log_signal_summary(info, ucontext);
+  log_signal_summary(info);
 
   debugger_thread_info thread_info = {
       .crashing_tid = __gettid(),
@@ -638,5 +637,11 @@
 
   // Use the alternate signal stack if available so we can catch stack overflows.
   action.sa_flags |= SA_ONSTACK;
+
+#define SA_EXPOSE_TAGBITS 0x00000800
+  // Request that the kernel set tag bits in the fault address. This is necessary for diagnosing MTE
+  // faults.
+  action.sa_flags |= SA_EXPOSE_TAGBITS;
+
   debuggerd_register_handlers(&action);
 }
diff --git a/debuggerd/libdebuggerd/include/libdebuggerd/utility.h b/debuggerd/libdebuggerd/include/libdebuggerd/utility.h
index 76155b1..29fb9a4 100644
--- a/debuggerd/libdebuggerd/include/libdebuggerd/utility.h
+++ b/debuggerd/libdebuggerd/include/libdebuggerd/utility.h
@@ -91,6 +91,4 @@
 const char* get_signame(const siginfo_t*);
 const char* get_sigcode(const siginfo_t*);
 
-uintptr_t get_fault_address(const siginfo_t* siginfo, const ucontext_t* ucontext);
-
 #endif // _DEBUGGERD_UTILITY_H
diff --git a/debuggerd/libdebuggerd/utility.cpp b/debuggerd/libdebuggerd/utility.cpp
index 4e6df09..d7067ca 100644
--- a/debuggerd/libdebuggerd/utility.cpp
+++ b/debuggerd/libdebuggerd/utility.cpp
@@ -451,40 +451,3 @@
     _LOG(log, logtype::BACKTRACE, "%s%s\n", prefix, unwinder->FormatFrame(i).c_str());
   }
 }
-
-#if defined(__aarch64__)
-#define FAR_MAGIC 0x46415201
-
-struct far_context {
-  struct _aarch64_ctx head;
-  __u64 far;
-};
-#endif
-
-uintptr_t get_fault_address(const siginfo_t* siginfo, const ucontext_t* ucontext) {
-  (void)ucontext;
-#if defined(__aarch64__)
-  // This relies on a kernel patch:
-  //   https://patchwork.kernel.org/patch/11435077/
-  // that hasn't been accepted into the kernel yet. TODO(pcc): Update this to
-  // use the official interface once it lands.
-  auto* begin = reinterpret_cast<const char*>(ucontext->uc_mcontext.__reserved);
-  auto* end = begin + sizeof(ucontext->uc_mcontext.__reserved);
-  auto* ptr = begin;
-  while (1) {
-    auto* ctx = reinterpret_cast<const _aarch64_ctx*>(ptr);
-    if (ctx->magic == 0) {
-      break;
-    }
-    if (ctx->magic == FAR_MAGIC) {
-      auto* far_ctx = reinterpret_cast<const far_context*>(ctx);
-      return far_ctx->far;
-    }
-    ptr += ctx->size;
-    if (ctx->size % sizeof(void*) != 0 || ptr < begin || ptr >= end) {
-      break;
-    }
-  }
-#endif
-  return reinterpret_cast<uintptr_t>(siginfo->si_addr);
-}
diff --git a/fastboot/device/flashing.cpp b/fastboot/device/flashing.cpp
index 1bf4c9c..333ca50 100644
--- a/fastboot/device/flashing.cpp
+++ b/fastboot/device/flashing.cpp
@@ -67,7 +67,7 @@
 
         if ((partition + device->GetCurrentSlot()) == partition_name) {
             mount_metadata.emplace();
-            fs_mgr_overlayfs_teardown(entry.mount_point.c_str());
+            android::fs_mgr::TeardownAllOverlayForMountPoint(entry.mount_point);
         }
     }
 }
@@ -194,7 +194,7 @@
         if (!FlashPartitionTable(super_name, *new_metadata.get())) {
             return device->WriteFail("Unable to flash new partition table");
         }
-        fs_mgr_overlayfs_teardown();
+        android::fs_mgr::TeardownAllOverlayForMountPoint();
         sync();
         return device->WriteOkay("Successfully flashed partition table");
     }
@@ -234,7 +234,7 @@
     if (!UpdateAllPartitionMetadata(device, super_name, *new_metadata.get())) {
         return device->WriteFail("Unable to write new partition table");
     }
-    fs_mgr_overlayfs_teardown();
+    android::fs_mgr::TeardownAllOverlayForMountPoint();
     sync();
     return device->WriteOkay("Successfully updated partition table");
 }
diff --git a/fastboot/fastboot.cpp b/fastboot/fastboot.cpp
index 4bf791e..62f6ac7 100644
--- a/fastboot/fastboot.cpp
+++ b/fastboot/fastboot.cpp
@@ -454,6 +454,8 @@
             " --skip-reboot              Don't reboot device after flashing.\n"
             " --disable-verity           Sets disable-verity when flashing vbmeta.\n"
             " --disable-verification     Sets disable-verification when flashing vbmeta.\n"
+            " --fs-options=OPTION[,OPTION]\n"
+            "                            Enable filesystem features. OPTION supports casefold, projid, compress\n"
 #if !defined(_WIN32)
             " --wipe-and-use-fbe         Enable file-based encryption, wiping userdata.\n"
 #endif
@@ -1581,7 +1583,7 @@
 static void fb_perform_format(
                               const std::string& partition, int skip_if_not_supported,
                               const std::string& type_override, const std::string& size_override,
-                              const std::string& initial_dir) {
+                              const std::string& initial_dir, const unsigned fs_options) {
     std::string partition_type, partition_size;
 
     struct fastboot_buffer buf;
@@ -1644,7 +1646,7 @@
     logicalBlkSize = fb_get_flash_block_size("logical-block-size");
 
     if (fs_generator_generate(gen, output.path, size, initial_dir,
-            eraseBlkSize, logicalBlkSize)) {
+            eraseBlkSize, logicalBlkSize, fs_options)) {
         die("Cannot generate image for %s", partition.c_str());
     }
 
@@ -1778,6 +1780,7 @@
     bool skip_secondary = false;
     bool set_fbe_marker = false;
     bool force_flash = false;
+    unsigned fs_options = 0;
     int longindex;
     std::string slot_override;
     std::string next_active;
@@ -1795,6 +1798,7 @@
         {"disable-verification", no_argument, 0, 0},
         {"disable-verity", no_argument, 0, 0},
         {"force", no_argument, 0, 0},
+        {"fs-options", required_argument, 0, 0},
         {"header-version", required_argument, 0, 0},
         {"help", no_argument, 0, 'h'},
         {"kernel-offset", required_argument, 0, 0},
@@ -1834,6 +1838,8 @@
                 g_disable_verity = true;
             } else if (name == "force") {
                 force_flash = true;
+            } else if (name == "fs-options") {
+                fs_options = ParseFsOption(optarg);
             } else if (name == "header-version") {
                 g_boot_img_hdr.header_version = strtoul(optarg, nullptr, 0);
             } else if (name == "dtb") {
@@ -1990,7 +1996,7 @@
             std::string partition = next_arg(&args);
 
             auto format = [&](const std::string& partition) {
-                fb_perform_format(partition, 0, type_override, size_override, "");
+                fb_perform_format(partition, 0, type_override, size_override, "", fs_options);
             };
             do_for_partitions(partition, slot_override, format, true);
         } else if (command == "signature") {
@@ -2180,10 +2186,10 @@
             if (partition == "userdata" && set_fbe_marker) {
                 fprintf(stderr, "setting FBE marker on initial userdata...\n");
                 std::string initial_userdata_dir = create_fbemarker_tmpdir();
-                fb_perform_format(partition, 1, partition_type, "", initial_userdata_dir);
+                fb_perform_format(partition, 1, partition_type, "", initial_userdata_dir, fs_options);
                 delete_fbemarker_tmpdir(initial_userdata_dir);
             } else {
-                fb_perform_format(partition, 1, partition_type, "", "");
+                fb_perform_format(partition, 1, partition_type, "", "", fs_options);
             }
         }
     }
@@ -2233,3 +2239,23 @@
     }
     hdr->SetOsVersion(major, minor, patch);
 }
+
+unsigned FastBootTool::ParseFsOption(const char* arg) {
+    unsigned fsOptions = 0;
+
+    std::vector<std::string> options = android::base::Split(arg, ",");
+    if (options.size() < 1)
+        syntax_error("bad options: %s", arg);
+
+    for (size_t i = 0; i < options.size(); ++i) {
+        if (options[i] == "casefold")
+            fsOptions |= (1 << FS_OPT_CASEFOLD);
+        else if (options[i] == "projid")
+            fsOptions |= (1 << FS_OPT_PROJID);
+        else if (options[i] == "compress")
+            fsOptions |= (1 << FS_OPT_COMPRESS);
+        else
+            syntax_error("unsupported options: %s", options[i].c_str());
+    }
+    return fsOptions;
+}
diff --git a/fastboot/fastboot.h b/fastboot/fastboot.h
index 9f18253..c23793a 100644
--- a/fastboot/fastboot.h
+++ b/fastboot/fastboot.h
@@ -34,4 +34,5 @@
 
     void ParseOsPatchLevel(boot_img_hdr_v1*, const char*);
     void ParseOsVersion(boot_img_hdr_v1*, const char*);
+    unsigned ParseFsOption(const char*);
 };
diff --git a/fastboot/fs.cpp b/fastboot/fs.cpp
index 8c0aa6b..8addcb6 100644
--- a/fastboot/fs.cpp
+++ b/fastboot/fs.cpp
@@ -113,7 +113,7 @@
 
 static int generate_ext4_image(const char* fileName, long long partSize,
                                const std::string& initial_dir, unsigned eraseBlkSize,
-                               unsigned logicalBlkSize) {
+                               unsigned logicalBlkSize, const unsigned fsOptions) {
     static constexpr int block_size = 4096;
     const std::string exec_dir = android::base::GetExecutableDirectory();
 
@@ -137,6 +137,12 @@
     mke2fs_args.push_back(ext_attr.c_str());
     mke2fs_args.push_back("-O");
     mke2fs_args.push_back("uninit_bg");
+
+    if (fsOptions & (1 << FS_OPT_PROJID)) {
+        mke2fs_args.push_back("-I");
+        mke2fs_args.push_back("512");
+    }
+
     mke2fs_args.push_back(fileName);
 
     std::string size_str = std::to_string(partSize / block_size);
@@ -162,9 +168,9 @@
     return exec_cmd(e2fsdroid_args[0], e2fsdroid_args.data(), nullptr);
 }
 
-static int generate_f2fs_image(const char* fileName, long long partSize, const std::string& initial_dir,
-                               unsigned /* unused */, unsigned /* unused */)
-{
+static int generate_f2fs_image(const char* fileName, long long partSize,
+                               const std::string& initial_dir, unsigned /* unused */,
+                               unsigned /* unused */, const unsigned fsOptions) {
     const std::string exec_dir = android::base::GetExecutableDirectory();
     const std::string mkf2fs_path = exec_dir + "/make_f2fs";
     std::vector<const char*> mkf2fs_args = {mkf2fs_path.c_str()};
@@ -174,6 +180,26 @@
     mkf2fs_args.push_back(size_str.c_str());
     mkf2fs_args.push_back("-g");
     mkf2fs_args.push_back("android");
+
+    if (fsOptions & (1 << FS_OPT_PROJID)) {
+        mkf2fs_args.push_back("-O");
+        mkf2fs_args.push_back("project_quota,extra_attr");
+    }
+
+    if (fsOptions & (1 << FS_OPT_CASEFOLD)) {
+        mkf2fs_args.push_back("-O");
+        mkf2fs_args.push_back("casefold");
+        mkf2fs_args.push_back("-C");
+        mkf2fs_args.push_back("utf8");
+    }
+
+    if (fsOptions & (1 << FS_OPT_COMPRESS)) {
+        mkf2fs_args.push_back("-O");
+        mkf2fs_args.push_back("compression");
+        mkf2fs_args.push_back("-O");
+        mkf2fs_args.push_back("extra_attr");
+    }
+
     mkf2fs_args.push_back(fileName);
     mkf2fs_args.push_back(nullptr);
 
@@ -198,7 +224,7 @@
 
     //returns 0 or error value
     int (*generate)(const char* fileName, long long partSize, const std::string& initial_dir,
-                    unsigned eraseBlkSize, unsigned logicalBlkSize);
+                    unsigned eraseBlkSize, unsigned logicalBlkSize, const unsigned fsOptions);
 
 } generators[] = {
     { "ext4", generate_ext4_image},
@@ -215,7 +241,7 @@
 }
 
 int fs_generator_generate(const struct fs_generator* gen, const char* fileName, long long partSize,
-    const std::string& initial_dir, unsigned eraseBlkSize, unsigned logicalBlkSize)
-{
-    return gen->generate(fileName, partSize, initial_dir, eraseBlkSize, logicalBlkSize);
+                          const std::string& initial_dir, unsigned eraseBlkSize,
+                          unsigned logicalBlkSize, const unsigned fsOptions) {
+    return gen->generate(fileName, partSize, initial_dir, eraseBlkSize, logicalBlkSize, fsOptions);
 }
diff --git a/fastboot/fs.h b/fastboot/fs.h
index 331100d..f832938 100644
--- a/fastboot/fs.h
+++ b/fastboot/fs.h
@@ -5,6 +5,13 @@
 
 struct fs_generator;
 
+enum FS_OPTION {
+    FS_OPT_CASEFOLD,
+    FS_OPT_PROJID,
+    FS_OPT_COMPRESS,
+};
+
 const struct fs_generator* fs_get_generator(const std::string& fs_type);
 int fs_generator_generate(const struct fs_generator* gen, const char* fileName, long long partSize,
-    const std::string& initial_dir, unsigned eraseBlkSize = 0, unsigned logicalBlkSize = 0);
+                          const std::string& initial_dir, unsigned eraseBlkSize = 0,
+                          unsigned logicalBlkSize = 0, unsigned fsOptions = 0);
diff --git a/fs_mgr/fs_mgr_fstab.cpp b/fs_mgr/fs_mgr_fstab.cpp
index 616a06f..42459ec 100644
--- a/fs_mgr/fs_mgr_fstab.cpp
+++ b/fs_mgr/fs_mgr_fstab.cpp
@@ -591,7 +591,7 @@
     FstabEntry userdata;
     if (FstabEntry* entry = GetEntryForMountPoint(fstab, "/data")) {
         userdata = *entry;
-        userdata.blk_device = "userdata_gsi";
+        userdata.blk_device = android::gsi::kDsuUserdata;
         userdata.fs_mgr_flags.logical = true;
         userdata.fs_mgr_flags.formattable = true;
         if (!userdata.metadata_key_dir.empty()) {
@@ -611,7 +611,11 @@
             continue;
         }
         // userdata has been handled
-        if (StartsWith(partition, "user")) {
+        if (partition == android::gsi::kDsuUserdata) {
+            continue;
+        }
+        // scratch is handled by fs_mgr_overlayfs
+        if (partition == android::gsi::kDsuScratch) {
             continue;
         }
         // dsu_partition_name = corresponding_partition_name + kDsuPostfix
@@ -688,11 +692,18 @@
         return false;
     }
     if (!is_proc_mounts && !access(android::gsi::kGsiBootedIndicatorFile, F_OK)) {
+        // This is expected to fail if host is android Q, since Q doesn't
+        // support DSU slotting. The DSU "active" indicator file would be
+        // non-existent or empty if DSU is enabled within the guest system.
+        // In that case, just use the default slot name "dsu".
         std::string dsu_slot;
         if (!android::gsi::GetActiveDsu(&dsu_slot)) {
-            PERROR << __FUNCTION__ << "(): failed to get active dsu slot";
-            return false;
+            PWARNING << __FUNCTION__ << "(): failed to get active dsu slot";
         }
+        if (dsu_slot.empty()) {
+            dsu_slot = "dsu";
+        }
+
         std::string lp_names;
         ReadFileToString(gsi::kGsiLpNamesFile, &lp_names);
         TransformFstabForDsu(fstab, dsu_slot, Split(lp_names, ","));
diff --git a/fs_mgr/fs_mgr_overlayfs.cpp b/fs_mgr/fs_mgr_overlayfs.cpp
index a7704de..388c296 100644
--- a/fs_mgr/fs_mgr_overlayfs.cpp
+++ b/fs_mgr/fs_mgr_overlayfs.cpp
@@ -113,8 +113,11 @@
 namespace android {
 namespace fs_mgr {
 
-void MapScratchPartitionIfNeeded(Fstab*,
-                                 const std::function<bool(const std::set<std::string>&)>&) {}
+void MapScratchPartitionIfNeeded(Fstab*, const std::function<bool(const std::set<std::string>&)>&) {
+}
+
+void TeardownAllOverlayForMountPoint(const std::string&) {}
+
 }  // namespace fs_mgr
 }  // namespace android
 
@@ -166,11 +169,34 @@
            (vst.f_bfree * vst.f_bsize) >= kSizeThreshold;
 }
 
+bool fs_mgr_in_recovery() {
+    // Check the existence of recovery binary instead of using the compile time
+    // macro, because first-stage-init is compiled with __ANDROID_RECOVERY__
+    // defined, albeit not in recovery. More details: system/core/init/README.md
+    return fs_mgr_access("/system/bin/recovery");
+}
+
+bool fs_mgr_is_dsu_running() {
+    // Since android::gsi::CanBootIntoGsi() or android::gsi::MarkSystemAsGsi() is
+    // never called in recovery, the return value of android::gsi::IsGsiRunning()
+    // is not well-defined. In this case, just return false as being in recovery
+    // implies not running a DSU system.
+    if (fs_mgr_in_recovery()) return false;
+    auto saved_errno = errno;
+    auto ret = android::gsi::IsGsiRunning();
+    errno = saved_errno;
+    return ret;
+}
+
 const auto kPhysicalDevice = "/dev/block/by-name/"s;
 constexpr char kScratchImageMetadata[] = "/metadata/gsi/remount/lp_metadata";
 
 // Note: this is meant only for recovery/first-stage init.
 bool ScratchIsOnData() {
+    // The scratch partition of DSU is managed by gsid.
+    if (fs_mgr_is_dsu_running()) {
+        return false;
+    }
     return fs_mgr_access(kScratchImageMetadata);
 }
 
@@ -464,6 +490,12 @@
     // umount and delete kScratchMountPoint storage if we have logical partitions
     if (overlay != kScratchMountPoint) return true;
 
+    // Validation check.
+    if (fs_mgr_is_dsu_running()) {
+        LERROR << "Destroying DSU scratch is not allowed.";
+        return false;
+    }
+
     auto save_errno = errno;
     if (fs_mgr_overlayfs_already_mounted(kScratchMountPoint, false)) {
         fs_mgr_overlayfs_umount_scratch();
@@ -512,10 +544,13 @@
 }
 
 bool fs_mgr_overlayfs_teardown_one(const std::string& overlay, const std::string& mount_point,
-                                   bool* change) {
+                                   bool* change, bool* should_destroy_scratch = nullptr) {
     const auto top = overlay + kOverlayTopDir;
 
-    if (!fs_mgr_access(top)) return fs_mgr_overlayfs_teardown_scratch(overlay, change);
+    if (!fs_mgr_access(top)) {
+        if (should_destroy_scratch) *should_destroy_scratch = true;
+        return true;
+    }
 
     auto cleanup_all = mount_point.empty();
     const auto partition_name = android::base::Basename(mount_point);
@@ -571,7 +606,7 @@
             PERROR << "rmdir " << top;
         }
     }
-    if (cleanup_all) ret &= fs_mgr_overlayfs_teardown_scratch(overlay, change);
+    if (should_destroy_scratch) *should_destroy_scratch = cleanup_all;
     return ret;
 }
 
@@ -881,12 +916,29 @@
     return "";
 }
 
+// Note: The scratch partition of DSU is managed by gsid, and should be initialized during
+// first-stage-mount. Just check if the DM device for DSU scratch partition is created or not.
+static std::string GetDsuScratchDevice() {
+    auto& dm = DeviceMapper::Instance();
+    std::string device;
+    if (dm.GetState(android::gsi::kDsuScratch) != DmDeviceState::INVALID &&
+        dm.GetDmDevicePathByName(android::gsi::kDsuScratch, &device)) {
+        return device;
+    }
+    return "";
+}
+
 // This returns the scratch device that was detected during early boot (first-
 // stage init). If the device was created later, for example during setup for
 // the adb remount command, it can return an empty string since it does not
 // query ImageManager. (Note that ImageManager in first-stage init will always
 // use device-mapper, since /data is not available to use loop devices.)
 static std::string GetBootScratchDevice() {
+    // Note: fs_mgr_is_dsu_running() always returns false in recovery or fastbootd.
+    if (fs_mgr_is_dsu_running()) {
+        return GetDsuScratchDevice();
+    }
+
     auto& dm = DeviceMapper::Instance();
 
     // If there is a scratch partition allocated in /data or on super, we
@@ -1045,7 +1097,7 @@
 
 static bool CreateScratchOnData(std::string* scratch_device, bool* partition_exists, bool* change) {
     *partition_exists = false;
-    *change = false;
+    if (change) *change = false;
 
     auto images = IImageManager::Open("remount", 10s);
     if (!images) {
@@ -1065,7 +1117,7 @@
         return false;
     }
 
-    *change = true;
+    if (change) *change = true;
 
     // Note: calling RemoveDisabledImages here ensures that we do not race with
     // clean_scratch_files and accidentally try to map an image that will be
@@ -1108,6 +1160,14 @@
 
 bool fs_mgr_overlayfs_create_scratch(const Fstab& fstab, std::string* scratch_device,
                                      bool* partition_exists, bool* change) {
+    // Use the DSU scratch device managed by gsid if within a DSU system.
+    if (fs_mgr_is_dsu_running()) {
+        *scratch_device = GetDsuScratchDevice();
+        *partition_exists = !scratch_device->empty();
+        *change = false;
+        return *partition_exists;
+    }
+
     // Try a physical partition first.
     *scratch_device = GetPhysicalScratchDevice();
     if (!scratch_device->empty() && fs_mgr_rw_access(*scratch_device)) {
@@ -1166,12 +1226,8 @@
 bool fs_mgr_overlayfs_invalid() {
     if (fs_mgr_overlayfs_valid() == OverlayfsValidResult::kNotSupported) return true;
 
-    // in recovery, fastbootd, or gsi mode, not allowed!
-    if (fs_mgr_access("/system/bin/recovery")) return true;
-    auto save_errno = errno;
-    auto ret = android::gsi::IsGsiRunning();
-    errno = save_errno;
-    return ret;
+    // in recovery or fastbootd, not allowed!
+    return fs_mgr_in_recovery();
 }
 
 }  // namespace
@@ -1314,6 +1370,8 @@
     return ret;
 }
 
+// Note: This function never returns the DSU scratch device in recovery or fastbootd,
+// because the DSU scratch is created in the first-stage-mount, which is not run in recovery.
 static bool EnsureScratchMapped(std::string* device, bool* mapped) {
     *mapped = false;
     *device = GetBootScratchDevice();
@@ -1321,6 +1379,11 @@
         return true;
     }
 
+    if (!fs_mgr_in_recovery()) {
+        errno = EINVAL;
+        return false;
+    }
+
     auto partition_name = android::base::Basename(kScratchMountPoint);
 
     // Check for scratch on /data first, before looking for a modified super
@@ -1362,10 +1425,27 @@
     return true;
 }
 
-static void UnmapScratchDevice() {
-    // This should only be reachable in recovery, where scratch is not
-    // automatically mapped and therefore can be unmapped.
-    DestroyLogicalPartition(android::base::Basename(kScratchMountPoint));
+// This should only be reachable in recovery, where DSU scratch is not
+// automatically mapped.
+static bool MapDsuScratchDevice(std::string* device) {
+    std::string dsu_slot;
+    if (!android::gsi::IsGsiInstalled() || !android::gsi::GetActiveDsu(&dsu_slot) ||
+        dsu_slot.empty()) {
+        // Nothing to do if no DSU installation present.
+        return false;
+    }
+
+    auto images = IImageManager::Open("dsu/" + dsu_slot, 10s);
+    if (!images || !images->BackingImageExists(android::gsi::kDsuScratch)) {
+        // Nothing to do if DSU scratch device doesn't exist.
+        return false;
+    }
+
+    images->UnmapImageDevice(android::gsi::kDsuScratch);
+    if (!images->MapImageDevice(android::gsi::kDsuScratch, 10s, device)) {
+        return false;
+    }
+    return true;
 }
 
 // Returns false if teardown not permitted, errno set to last error.
@@ -1377,21 +1457,27 @@
     // If scratch exists, but is not mounted, lets gain access to clean
     // specific override entries.
     auto mount_scratch = false;
-    bool unmap = false;
     if ((mount_point != nullptr) && !fs_mgr_overlayfs_already_mounted(kScratchMountPoint, false)) {
-        std::string scratch_device;
-        if (EnsureScratchMapped(&scratch_device, &unmap)) {
+        std::string scratch_device = GetBootScratchDevice();
+        if (!scratch_device.empty()) {
             mount_scratch = fs_mgr_overlayfs_mount_scratch(scratch_device,
                                                            fs_mgr_overlayfs_scratch_mount_type());
         }
     }
+    bool should_destroy_scratch = false;
     for (const auto& overlay_mount_point : kOverlayMountPoints) {
         ret &= fs_mgr_overlayfs_teardown_one(
-                overlay_mount_point, mount_point ? fs_mgr_mount_point(mount_point) : "", change);
+                overlay_mount_point, mount_point ? fs_mgr_mount_point(mount_point) : "", change,
+                overlay_mount_point == kScratchMountPoint ? &should_destroy_scratch : nullptr);
+    }
+    // Do not attempt to destroy DSU scratch if within a DSU system,
+    // because DSU scratch partition is managed by gsid.
+    if (should_destroy_scratch && !fs_mgr_is_dsu_running()) {
+        ret &= fs_mgr_overlayfs_teardown_scratch(kScratchMountPoint, change);
     }
     if (fs_mgr_overlayfs_valid() == OverlayfsValidResult::kNotSupported) {
         // After obligatory teardown to make sure everything is clean, but if
-        // we didn't want overlayfs in the the first place, we do not want to
+        // we didn't want overlayfs in the first place, we do not want to
         // waste time on a reboot (or reboot request message).
         if (change) *change = false;
     }
@@ -1405,9 +1491,6 @@
     if (mount_scratch) {
         fs_mgr_overlayfs_umount_scratch();
     }
-    if (unmap) {
-        UnmapScratchDevice();
-    }
     return ret;
 }
 
@@ -1475,6 +1558,54 @@
     }
 }
 
+void TeardownAllOverlayForMountPoint(const std::string& mount_point) {
+    if (!fs_mgr_in_recovery()) {
+        LERROR << __FUNCTION__ << "(): must be called within recovery.";
+        return;
+    }
+
+    // Empty string means teardown everything.
+    const std::string teardown_dir = mount_point.empty() ? "" : fs_mgr_mount_point(mount_point);
+    constexpr bool* ignore_change = nullptr;
+
+    // Teardown legacy overlay mount points that's not backed by a scratch device.
+    for (const auto& overlay_mount_point : kOverlayMountPoints) {
+        if (overlay_mount_point == kScratchMountPoint) {
+            continue;
+        }
+        fs_mgr_overlayfs_teardown_one(overlay_mount_point, teardown_dir, ignore_change);
+    }
+
+    // Map scratch device, mount kScratchMountPoint and teardown kScratchMountPoint.
+    bool mapped = false;
+    std::string scratch_device;
+    if (EnsureScratchMapped(&scratch_device, &mapped)) {
+        fs_mgr_overlayfs_umount_scratch();
+        if (fs_mgr_overlayfs_mount_scratch(scratch_device, fs_mgr_overlayfs_scratch_mount_type())) {
+            bool should_destroy_scratch = false;
+            fs_mgr_overlayfs_teardown_one(kScratchMountPoint, teardown_dir, ignore_change,
+                                          &should_destroy_scratch);
+            if (should_destroy_scratch) {
+                fs_mgr_overlayfs_teardown_scratch(kScratchMountPoint, nullptr);
+            }
+            fs_mgr_overlayfs_umount_scratch();
+        }
+        if (mapped) {
+            DestroyLogicalPartition(android::base::Basename(kScratchMountPoint));
+        }
+    }
+
+    // Teardown DSU overlay if present.
+    if (MapDsuScratchDevice(&scratch_device)) {
+        fs_mgr_overlayfs_umount_scratch();
+        if (fs_mgr_overlayfs_mount_scratch(scratch_device, fs_mgr_overlayfs_scratch_mount_type())) {
+            fs_mgr_overlayfs_teardown_one(kScratchMountPoint, teardown_dir, ignore_change);
+            fs_mgr_overlayfs_umount_scratch();
+        }
+        DestroyLogicalPartition(android::gsi::kDsuScratch);
+    }
+}
+
 }  // namespace fs_mgr
 }  // namespace android
 
diff --git a/fs_mgr/include/fs_mgr_overlayfs.h b/fs_mgr/include/fs_mgr_overlayfs.h
index 34aded9..d45e2de 100644
--- a/fs_mgr/include/fs_mgr_overlayfs.h
+++ b/fs_mgr/include/fs_mgr_overlayfs.h
@@ -49,5 +49,12 @@
                                  const std::function<bool(const std::set<std::string>&)>& init);
 void CleanupOldScratchFiles();
 
+// Teardown overlays of all sources (cache dir, scratch device, DSU) for |mount_point|.
+// Teardown all overlays if |mount_point| is empty.
+//
+// Note: This should be called if and only if in recovery or fastbootd to teardown
+// overlays if any partition is flashed or updated.
+void TeardownAllOverlayForMountPoint(const std::string& mount_point = {});
+
 }  // namespace fs_mgr
 }  // namespace android
diff --git a/fs_mgr/libdm/dm.cpp b/fs_mgr/libdm/dm.cpp
index bb7d8b3..d5b8a61 100644
--- a/fs_mgr/libdm/dm.cpp
+++ b/fs_mgr/libdm/dm.cpp
@@ -111,8 +111,10 @@
 
     // Check to make sure appropriate uevent is generated so ueventd will
     // do the right thing and remove the corresponding device node and symlinks.
-    CHECK(io.flags & DM_UEVENT_GENERATED_FLAG)
-            << "Didn't generate uevent for [" << name << "] removal";
+    if ((io.flags & DM_UEVENT_GENERATED_FLAG) == 0) {
+        LOG(ERROR) << "Didn't generate uevent for [" << name << "] removal";
+        return false;
+    }
 
     if (timeout_ms <= std::chrono::milliseconds::zero()) {
         return true;
diff --git a/fs_mgr/libsnapshot/Android.bp b/fs_mgr/libsnapshot/Android.bp
index f1b0031..910911e 100644
--- a/fs_mgr/libsnapshot/Android.bp
+++ b/fs_mgr/libsnapshot/Android.bp
@@ -46,6 +46,7 @@
     header_libs: [
         "libchrome",
         "libfiemap_headers",
+        "libstorage_literals_headers",
         "libupdate_engine_headers",
     ],
     export_static_lib_headers: [
diff --git a/fs_mgr/libsnapshot/cow_api_test.cpp b/fs_mgr/libsnapshot/cow_api_test.cpp
index 3d0321f..4db6584 100644
--- a/fs_mgr/libsnapshot/cow_api_test.cpp
+++ b/fs_mgr/libsnapshot/cow_api_test.cpp
@@ -264,15 +264,11 @@
     ASSERT_EQ(size_before, size_after);
     struct stat buf;
 
-    if (fstat(cow_->fd, &buf) < 0) {
-        perror("Fails to determine size of cow image written");
-        FAIL();
-    }
+    ASSERT_GE(fstat(cow_->fd, &buf), 0) << strerror(errno);
     ASSERT_EQ(buf.st_size, writer.GetCowSize());
 }
 
-TEST_F(CowTest, Append) {
-    cow_->DoNotRemove();
+TEST_F(CowTest, AppendLabelSmall) {
     CowOptions options;
     auto writer = std::make_unique<CowWriter>(options);
     ASSERT_TRUE(writer->Initialize(cow_->fd));
@@ -280,12 +276,13 @@
     std::string data = "This is some data, believe it";
     data.resize(options.block_size, '\0');
     ASSERT_TRUE(writer->AddRawBlocks(50, data.data(), data.size()));
+    ASSERT_TRUE(writer->AddLabel(3));
     ASSERT_TRUE(writer->Finalize());
 
     ASSERT_EQ(lseek(cow_->fd, 0, SEEK_SET), 0);
 
     writer = std::make_unique<CowWriter>(options);
-    ASSERT_TRUE(writer->Initialize(cow_->fd, CowWriter::OpenMode::APPEND));
+    ASSERT_TRUE(writer->InitializeAppend(cow_->fd, 3));
 
     std::string data2 = "More data!";
     data2.resize(options.block_size, '\0');
@@ -298,11 +295,12 @@
     ASSERT_EQ(fstat(cow_->fd, &buf), 0);
     ASSERT_EQ(buf.st_size, writer->GetCowSize());
 
-    // Read back both operations.
+    // Read back both operations, and label.
     CowReader reader;
     uint64_t label;
     ASSERT_TRUE(reader.Parse(cow_->fd));
-    ASSERT_FALSE(reader.GetLastLabel(&label));
+    ASSERT_TRUE(reader.GetLastLabel(&label));
+    ASSERT_EQ(label, 3);
 
     StringSink sink;
 
@@ -320,89 +318,6 @@
 
     ASSERT_FALSE(iter->Done());
     op = &iter->Get();
-    ASSERT_EQ(op->type, kCowReplaceOp);
-    ASSERT_TRUE(reader.ReadData(*op, &sink));
-    ASSERT_EQ(sink.stream(), data2);
-
-    iter->Next();
-    ASSERT_TRUE(iter->Done());
-}
-
-TEST_F(CowTest, AppendCorrupted) {
-    CowOptions options;
-    auto writer = std::make_unique<CowWriter>(options);
-    ASSERT_TRUE(writer->Initialize(cow_->fd));
-
-    std::string data = "This is some data, believe it";
-    data.resize(options.block_size, '\0');
-    ASSERT_TRUE(writer->AddLabel(0));
-    ASSERT_TRUE(writer->AddRawBlocks(50, data.data(), data.size()));
-    ASSERT_TRUE(writer->AddLabel(1));
-    ASSERT_TRUE(writer->AddZeroBlocks(50, 1));
-    ASSERT_TRUE(writer->Finalize());
-    // Drop the tail end of the header. Last entry may be corrupted.
-    ftruncate(cow_->fd, writer->GetCowSize() - 5);
-
-    ASSERT_EQ(lseek(cow_->fd, 0, SEEK_SET), 0);
-
-    writer = std::make_unique<CowWriter>(options);
-    ASSERT_TRUE(writer->Initialize(cow_->fd, CowWriter::OpenMode::APPEND));
-
-    ASSERT_TRUE(writer->AddLabel(2));
-    ASSERT_TRUE(writer->AddZeroBlocks(50, 1));
-
-    std::string data2 = "More data!";
-    data2.resize(options.block_size, '\0');
-    ASSERT_TRUE(writer->AddLabel(3));
-    ASSERT_TRUE(writer->AddRawBlocks(51, data2.data(), data2.size()));
-    ASSERT_TRUE(writer->Finalize());
-
-    ASSERT_EQ(lseek(cow_->fd, 0, SEEK_SET), 0);
-
-    struct stat buf;
-    ASSERT_EQ(fstat(cow_->fd, &buf), 0);
-    ASSERT_EQ(buf.st_size, writer->GetCowSize());
-
-    // Read back all three operations.
-    CowReader reader;
-    ASSERT_TRUE(reader.Parse(cow_->fd));
-
-    StringSink sink;
-
-    auto iter = reader.GetOpIter();
-    ASSERT_NE(iter, nullptr);
-
-    ASSERT_FALSE(iter->Done());
-    auto op = &iter->Get();
-    ASSERT_EQ(op->type, kCowLabelOp);
-    ASSERT_EQ(op->source, 0);
-
-    iter->Next();
-
-    ASSERT_FALSE(iter->Done());
-    op = &iter->Get();
-    ASSERT_EQ(op->type, kCowReplaceOp);
-    ASSERT_TRUE(reader.ReadData(*op, &sink));
-    ASSERT_EQ(sink.stream(), data);
-
-    iter->Next();
-    sink.Reset();
-
-    ASSERT_FALSE(iter->Done());
-    op = &iter->Get();
-    ASSERT_EQ(op->type, kCowLabelOp);
-    ASSERT_EQ(op->source, 2);
-
-    iter->Next();
-
-    ASSERT_FALSE(iter->Done());
-    op = &iter->Get();
-    ASSERT_EQ(op->type, kCowZeroOp);
-
-    iter->Next();
-
-    ASSERT_FALSE(iter->Done());
-    op = &iter->Get();
     ASSERT_EQ(op->type, kCowLabelOp);
     ASSERT_EQ(op->source, 3);
 
@@ -418,33 +333,86 @@
     ASSERT_TRUE(iter->Done());
 }
 
+TEST_F(CowTest, AppendLabelMissing) {
+    CowOptions options;
+    auto writer = std::make_unique<CowWriter>(options);
+    ASSERT_TRUE(writer->Initialize(cow_->fd));
+
+    ASSERT_TRUE(writer->AddLabel(0));
+    std::string data = "This is some data, believe it";
+    data.resize(options.block_size, '\0');
+    ASSERT_TRUE(writer->AddRawBlocks(50, data.data(), data.size()));
+    ASSERT_TRUE(writer->AddLabel(1));
+    // Drop the tail end of the last op header, corrupting it.
+    ftruncate(cow_->fd, writer->GetCowSize() - sizeof(CowFooter) - 3);
+
+    ASSERT_EQ(lseek(cow_->fd, 0, SEEK_SET), 0);
+
+    writer = std::make_unique<CowWriter>(options);
+    ASSERT_FALSE(writer->InitializeAppend(cow_->fd, 1));
+    ASSERT_TRUE(writer->InitializeAppend(cow_->fd, 0));
+
+    ASSERT_TRUE(writer->AddZeroBlocks(51, 1));
+    ASSERT_TRUE(writer->Finalize());
+
+    ASSERT_EQ(lseek(cow_->fd, 0, SEEK_SET), 0);
+
+    struct stat buf;
+    ASSERT_EQ(fstat(cow_->fd, &buf), 0);
+    ASSERT_EQ(buf.st_size, writer->GetCowSize());
+
+    // Read back both operations.
+    CowReader reader;
+    ASSERT_TRUE(reader.Parse(cow_->fd));
+
+    StringSink sink;
+
+    auto iter = reader.GetOpIter();
+    ASSERT_NE(iter, nullptr);
+
+    ASSERT_FALSE(iter->Done());
+    auto op = &iter->Get();
+    ASSERT_EQ(op->type, kCowLabelOp);
+    ASSERT_EQ(op->source, 0);
+
+    iter->Next();
+
+    ASSERT_FALSE(iter->Done());
+    op = &iter->Get();
+    ASSERT_EQ(op->type, kCowZeroOp);
+
+    iter->Next();
+
+    ASSERT_TRUE(iter->Done());
+}
+
 TEST_F(CowTest, AppendExtendedCorrupted) {
     CowOptions options;
     auto writer = std::make_unique<CowWriter>(options);
     ASSERT_TRUE(writer->Initialize(cow_->fd));
 
     ASSERT_TRUE(writer->AddLabel(5));
-    ASSERT_TRUE(writer->AddLabel(6));
 
     std::string data = "This is some data, believe it";
     data.resize(options.block_size * 2, '\0');
     ASSERT_TRUE(writer->AddRawBlocks(50, data.data(), data.size()));
+    ASSERT_TRUE(writer->AddLabel(6));
 
-    // fail to write the footer
+    // fail to write the footer. Cow Format does not know if Label 6 is valid
 
     ASSERT_EQ(lseek(cow_->fd, 0, SEEK_SET), 0);
 
     // Get the last known good label
     CowReader label_reader;
     uint64_t label;
-    ASSERT_TRUE(label_reader.Parse(cow_->fd));
+    ASSERT_TRUE(label_reader.Parse(cow_->fd, {5}));
     ASSERT_TRUE(label_reader.GetLastLabel(&label));
     ASSERT_EQ(label, 5);
 
     ASSERT_EQ(lseek(cow_->fd, 0, SEEK_SET), 0);
 
     writer = std::make_unique<CowWriter>(options);
-    ASSERT_TRUE(writer->Initialize(cow_->fd, CowWriter::OpenMode::APPEND));
+    ASSERT_TRUE(writer->InitializeAppend(cow_->fd, 5));
 
     ASSERT_TRUE(writer->Finalize());
 
@@ -452,7 +420,7 @@
     ASSERT_EQ(fstat(cow_->fd, &buf), 0);
     ASSERT_EQ(buf.st_size, writer->GetCowSize());
 
-    // Read back all three operations.
+    // Read back all valid operations
     CowReader reader;
     ASSERT_TRUE(reader.Parse(cow_->fd));
 
@@ -475,16 +443,20 @@
     auto writer = std::make_unique<CowWriter>(options);
     ASSERT_TRUE(writer->Initialize(cow_->fd));
 
-    ASSERT_TRUE(writer->AddLabel(4));
-
-    ASSERT_TRUE(writer->AddLabel(5));
     std::string data = "This is some data, believe it";
     data.resize(options.block_size * 2, '\0');
     ASSERT_TRUE(writer->AddRawBlocks(50, data.data(), data.size()));
 
-    ASSERT_TRUE(writer->AddLabel(6));
+    ASSERT_TRUE(writer->AddLabel(4));
+
     ASSERT_TRUE(writer->AddZeroBlocks(50, 2));
 
+    ASSERT_TRUE(writer->AddLabel(5));
+
+    ASSERT_TRUE(writer->AddCopy(5, 6));
+
+    ASSERT_TRUE(writer->AddLabel(6));
+
     ASSERT_EQ(lseek(cow_->fd, 0, SEEK_SET), 0);
 
     writer = std::make_unique<CowWriter>(options);
@@ -509,20 +481,6 @@
 
     ASSERT_FALSE(iter->Done());
     auto op = &iter->Get();
-    ASSERT_EQ(op->type, kCowLabelOp);
-    ASSERT_EQ(op->source, 4);
-
-    iter->Next();
-
-    ASSERT_FALSE(iter->Done());
-    op = &iter->Get();
-    ASSERT_EQ(op->type, kCowLabelOp);
-    ASSERT_EQ(op->source, 5);
-
-    iter->Next();
-
-    ASSERT_FALSE(iter->Done());
-    op = &iter->Get();
     ASSERT_EQ(op->type, kCowReplaceOp);
     ASSERT_TRUE(reader.ReadData(*op, &sink));
     ASSERT_EQ(sink.stream(), data.substr(0, options.block_size));
@@ -539,6 +497,31 @@
     iter->Next();
     sink.Reset();
 
+    ASSERT_FALSE(iter->Done());
+    op = &iter->Get();
+    ASSERT_EQ(op->type, kCowLabelOp);
+    ASSERT_EQ(op->source, 4);
+
+    iter->Next();
+
+    ASSERT_FALSE(iter->Done());
+    op = &iter->Get();
+    ASSERT_EQ(op->type, kCowZeroOp);
+
+    iter->Next();
+
+    ASSERT_FALSE(iter->Done());
+    op = &iter->Get();
+    ASSERT_EQ(op->type, kCowZeroOp);
+
+    iter->Next();
+    ASSERT_FALSE(iter->Done());
+    op = &iter->Get();
+    ASSERT_EQ(op->type, kCowLabelOp);
+    ASSERT_EQ(op->source, 5);
+
+    iter->Next();
+
     ASSERT_TRUE(iter->Done());
 }
 
diff --git a/fs_mgr/libsnapshot/cow_reader.cpp b/fs_mgr/libsnapshot/cow_reader.cpp
index f10ccb6..6b7ada5 100644
--- a/fs_mgr/libsnapshot/cow_reader.cpp
+++ b/fs_mgr/libsnapshot/cow_reader.cpp
@@ -31,14 +31,7 @@
 namespace android {
 namespace snapshot {
 
-CowReader::CowReader()
-    : fd_(-1),
-      header_(),
-      footer_(),
-      fd_size_(0),
-      has_footer_(false),
-      last_label_(0),
-      has_last_label_(false) {}
+CowReader::CowReader() : fd_(-1), header_(), fd_size_(0) {}
 
 static void SHA256(const void*, size_t, uint8_t[]) {
 #if 0
@@ -49,12 +42,12 @@
 #endif
 }
 
-bool CowReader::Parse(android::base::unique_fd&& fd) {
+bool CowReader::Parse(android::base::unique_fd&& fd, std::optional<uint64_t> label) {
     owned_fd_ = std::move(fd);
-    return Parse(android::base::borrowed_fd{owned_fd_});
+    return Parse(android::base::borrowed_fd{owned_fd_}, label);
 }
 
-bool CowReader::Parse(android::base::borrowed_fd fd) {
+bool CowReader::Parse(android::base::borrowed_fd fd, std::optional<uint64_t> label) {
     fd_ = fd;
 
     auto pos = lseek(fd_.get(), 0, SEEK_END);
@@ -99,85 +92,122 @@
         return false;
     }
 
-    auto footer_pos = lseek(fd_.get(), -header_.footer_size, SEEK_END);
-    if (footer_pos != fd_size_ - header_.footer_size) {
-        LOG(ERROR) << "Failed to read full footer!";
-        return false;
-    }
-    if (!android::base::ReadFully(fd_, &footer_, sizeof(footer_))) {
-        PLOG(ERROR) << "read footer failed";
-        return false;
-    }
-    has_footer_ = (footer_.op.type == kCowFooterOp);
-    return ParseOps();
+    return ParseOps(label);
 }
 
-bool CowReader::ParseOps() {
+bool CowReader::ParseOps(std::optional<uint64_t> label) {
     uint64_t pos = lseek(fd_.get(), sizeof(header_), SEEK_SET);
     if (pos != sizeof(header_)) {
         PLOG(ERROR) << "lseek ops failed";
         return false;
     }
-    std::optional<uint64_t> next_last_label;
+
     auto ops_buffer = std::make_shared<std::vector<CowOperation>>();
-    if (has_footer_) ops_buffer->reserve(footer_.op.num_ops);
-    uint64_t current_op_num = 0;
-    // Look until we reach the last possible non-footer position.
-    uint64_t last_pos = fd_size_ - (has_footer_ ? sizeof(footer_) : sizeof(CowOperation));
 
     // Alternating op and data
-    while (pos < last_pos) {
-        ops_buffer->resize(current_op_num + 1);
-        if (!android::base::ReadFully(fd_, ops_buffer->data() + current_op_num,
-                                      sizeof(CowOperation))) {
+    while (true) {
+        ops_buffer->emplace_back();
+        if (!android::base::ReadFully(fd_, &ops_buffer->back(), sizeof(CowOperation))) {
             PLOG(ERROR) << "read op failed";
             return false;
         }
-        auto& current_op = ops_buffer->data()[current_op_num];
-        pos = lseek(fd_.get(), GetNextOpOffset(current_op), SEEK_CUR);
-        if (pos == uint64_t(-1)) {
+
+        auto& current_op = ops_buffer->back();
+        off_t offs = lseek(fd_.get(), GetNextOpOffset(current_op), SEEK_CUR);
+        if (offs < 0) {
             PLOG(ERROR) << "lseek next op failed";
             return false;
         }
-        current_op_num++;
+        pos = static_cast<uint64_t>(offs);
+
         if (current_op.type == kCowLabelOp) {
-            // If we don't have a footer, the last label may be incomplete
-            if (has_footer_) {
-                has_last_label_ = true;
-                last_label_ = current_op.source;
-            } else {
-                if (next_last_label) {
-                    last_label_ = next_last_label.value();
-                    has_last_label_ = true;
-                }
-                next_last_label = {current_op.source};
+            last_label_ = {current_op.source};
+
+            // If we reach the requested label, stop reading.
+            if (label && label.value() == current_op.source) {
+                break;
             }
         } else if (current_op.type == kCowFooterOp) {
-            memcpy(&footer_.op, &current_op, sizeof(footer_.op));
+            footer_.emplace();
 
-            if (android::base::ReadFully(fd_, &footer_.data, sizeof(footer_.data))) {
-                has_footer_ = true;
-                if (next_last_label) {
-                    last_label_ = next_last_label.value();
-                    has_last_label_ = true;
-                }
+            CowFooter* footer = &footer_.value();
+            memcpy(&footer_->op, &current_op, sizeof(footer->op));
+
+            if (!android::base::ReadFully(fd_, &footer->data, sizeof(footer->data))) {
+                LOG(ERROR) << "Could not read COW footer";
+                return false;
             }
+
+            // Drop the footer from the op stream.
+            ops_buffer->pop_back();
             break;
         }
     }
 
+    // To successfully parse a COW file, we need either:
+    //  (1) a label to read up to, and for that label to be found, or
+    //  (2) a valid footer.
+    if (label) {
+        if (!last_label_) {
+            LOG(ERROR) << "Did not find label " << label.value()
+                       << " while reading COW (no labels found)";
+            return false;
+        }
+        if (last_label_.value() != label.value()) {
+            LOG(ERROR) << "Did not find label " << label.value()
+                       << ", last label=" << last_label_.value();
+            return false;
+        }
+    } else if (!footer_) {
+        LOG(ERROR) << "No COW footer found";
+        return false;
+    }
+
     uint8_t csum[32];
     memset(csum, 0, sizeof(uint8_t) * 32);
 
-    if (has_footer_) {
-        SHA256(ops_buffer.get()->data(), footer_.op.ops_size, csum);
-        if (memcmp(csum, footer_.data.ops_checksum, sizeof(csum)) != 0) {
+    if (footer_) {
+        if (ops_buffer->size() != footer_->op.num_ops) {
+            LOG(ERROR) << "num ops does not match";
+            return false;
+        }
+        if (ops_buffer->size() * sizeof(CowOperation) != footer_->op.ops_size) {
+            LOG(ERROR) << "ops size does not match ";
+            return false;
+        }
+        SHA256(&footer_->op, sizeof(footer_->op), footer_->data.footer_checksum);
+        if (memcmp(csum, footer_->data.ops_checksum, sizeof(csum)) != 0) {
+            LOG(ERROR) << "ops checksum does not match";
+            return false;
+        }
+        SHA256(ops_buffer.get()->data(), footer_->op.ops_size, csum);
+        if (memcmp(csum, footer_->data.ops_checksum, sizeof(csum)) != 0) {
             LOG(ERROR) << "ops checksum does not match";
             return false;
         }
     } else {
-        LOG(INFO) << "No Footer, recovered data";
+        LOG(INFO) << "No COW Footer, recovered data";
     }
+
+    if (header_.num_merge_ops > 0) {
+        uint64_t merge_ops = header_.num_merge_ops;
+        uint64_t metadata_ops = 0;
+        uint64_t current_op_num = 0;
+
+        CHECK(ops_buffer->size() >= merge_ops);
+        while (merge_ops) {
+            auto& current_op = ops_buffer->data()[current_op_num];
+            if (current_op.type == kCowLabelOp || current_op.type == kCowFooterOp) {
+                metadata_ops += 1;
+            } else {
+                merge_ops -= 1;
+            }
+            current_op_num += 1;
+        }
+        ops_buffer->erase(ops_buffer.get()->begin(),
+                          ops_buffer.get()->begin() + header_.num_merge_ops + metadata_ops);
+    }
+
     ops_ = ops_buffer;
     return true;
 }
@@ -188,14 +218,14 @@
 }
 
 bool CowReader::GetFooter(CowFooter* footer) {
-    if (!has_footer_) return false;
-    *footer = footer_;
+    if (!footer_) return false;
+    *footer = footer_.value();
     return true;
 }
 
 bool CowReader::GetLastLabel(uint64_t* label) {
-    if (!has_last_label_) return false;
-    *label = last_label_;
+    if (!last_label_) return false;
+    *label = last_label_.value();
     return true;
 }
 
@@ -231,14 +261,50 @@
     return (*op_iter_);
 }
 
+class CowOpReverseIter final : public ICowOpReverseIter {
+  public:
+    explicit CowOpReverseIter(std::shared_ptr<std::vector<CowOperation>> ops);
+
+    bool Done() override;
+    const CowOperation& Get() override;
+    void Next() override;
+
+  private:
+    std::shared_ptr<std::vector<CowOperation>> ops_;
+    std::vector<CowOperation>::reverse_iterator op_riter_;
+};
+
+CowOpReverseIter::CowOpReverseIter(std::shared_ptr<std::vector<CowOperation>> ops) {
+    ops_ = ops;
+    op_riter_ = ops_.get()->rbegin();
+}
+
+bool CowOpReverseIter::Done() {
+    return op_riter_ == ops_.get()->rend();
+}
+
+void CowOpReverseIter::Next() {
+    CHECK(!Done());
+    op_riter_++;
+}
+
+const CowOperation& CowOpReverseIter::Get() {
+    CHECK(!Done());
+    return (*op_riter_);
+}
+
 std::unique_ptr<ICowOpIter> CowReader::GetOpIter() {
     return std::make_unique<CowOpIter>(ops_);
 }
 
+std::unique_ptr<ICowOpReverseIter> CowReader::GetRevOpIter() {
+    return std::make_unique<CowOpReverseIter>(ops_);
+}
+
 bool CowReader::GetRawBytes(uint64_t offset, void* buffer, size_t len, size_t* read) {
     // Validate the offset, taking care to acknowledge possible overflow of offset+len.
-    if (offset < sizeof(header_) || offset >= fd_size_ - sizeof(footer_) || len >= fd_size_ ||
-        offset + len > fd_size_ - sizeof(footer_)) {
+    if (offset < sizeof(header_) || offset >= fd_size_ - sizeof(CowFooter) || len >= fd_size_ ||
+        offset + len > fd_size_ - sizeof(CowFooter)) {
         LOG(ERROR) << "invalid data offset: " << offset << ", " << len << " bytes";
         return false;
     }
diff --git a/fs_mgr/libsnapshot/cow_snapuserd_test.cpp b/fs_mgr/libsnapshot/cow_snapuserd_test.cpp
index ab15194..5483fd0 100644
--- a/fs_mgr/libsnapshot/cow_snapuserd_test.cpp
+++ b/fs_mgr/libsnapshot/cow_snapuserd_test.cpp
@@ -41,6 +41,10 @@
 class SnapuserdTest : public ::testing::Test {
   protected:
     void SetUp() override {
+        // TODO: Daemon started through first stage
+        // init does not have permission to read files
+        // from /data/nativetest.
+        system("setenforce 0");
         cow_system_ = std::make_unique<TemporaryFile>();
         ASSERT_GE(cow_system_->fd, 0) << strerror(errno);
 
@@ -53,10 +57,6 @@
         cow_product_1_ = std::make_unique<TemporaryFile>();
         ASSERT_GE(cow_product_1_->fd, 0) << strerror(errno);
 
-        // Create temp files in the PWD as selinux
-        // allows kernel domin to read from that directory only
-        // on userdebug/eng builds. Creating files under /data/local/tmp
-        // will have selinux denials.
         std::string path = android::base::GetExecutableDirectory();
 
         system_a_ = std::make_unique<TemporaryFile>(path);
@@ -65,10 +65,11 @@
         product_a_ = std::make_unique<TemporaryFile>(path);
         ASSERT_GE(product_a_->fd, 0) << strerror(errno);
 
-        size_ = 1_MiB;
+        size_ = 100_MiB;
     }
 
     void TearDown() override {
+        system("setenforce 1");
         cow_system_ = nullptr;
         cow_product_ = nullptr;
 
@@ -107,6 +108,8 @@
     std::unique_ptr<uint8_t[]> product_buffer_;
 
     void Init();
+    void InitCowDevices();
+    void InitDaemon();
     void CreateCowDevice(std::unique_ptr<TemporaryFile>& cow);
     void CreateSystemDmUser(std::unique_ptr<TemporaryFile>& cow);
     void CreateProductDmUser(std::unique_ptr<TemporaryFile>& cow);
@@ -116,10 +119,10 @@
     void SwitchSnapshotDevices();
 
     std::string GetSystemControlPath() {
-        return std::string("/dev/dm-user-") + system_device_ctrl_name_;
+        return std::string("/dev/dm-user/") + system_device_ctrl_name_;
     }
     std::string GetProductControlPath() {
-        return std::string("/dev/dm-user-") + product_device_ctrl_name_;
+        return std::string("/dev/dm-user/") + product_device_ctrl_name_;
     }
 
     void TestIO(unique_fd& snapshot_fd, std::unique_ptr<uint8_t[]>& buffer);
@@ -151,12 +154,12 @@
         offset += 1_MiB;
     }
 
-    for (size_t j = 0; j < (8_MiB / 1_MiB); j++) {
+    for (size_t j = 0; j < (800_MiB / 1_MiB); j++) {
         ASSERT_EQ(ReadFullyAtOffset(rnd_fd, (char*)random_buffer.get(), 1_MiB, 0), true);
         ASSERT_EQ(android::base::WriteFully(system_a_->fd, random_buffer.get(), 1_MiB), true);
     }
 
-    for (size_t j = 0; j < (8_MiB / 1_MiB); j++) {
+    for (size_t j = 0; j < (800_MiB / 1_MiB); j++) {
         ASSERT_EQ(ReadFullyAtOffset(rnd_fd, (char*)random_buffer.get(), 1_MiB, 0), true);
         ASSERT_EQ(android::base::WriteFully(product_a_->fd, random_buffer.get(), 1_MiB), true);
     }
@@ -237,12 +240,6 @@
     system_device_name_.clear();
     system_device_ctrl_name_.clear();
 
-    // Create a COW device. Number of sectors is chosen random which can
-    // hold at least 400MB of data
-
-    int err = ioctl(sys_fd_.get(), BLKGETSIZE, &system_blksize_);
-    ASSERT_GE(err, 0);
-
     std::string str(cow->path);
     std::size_t found = str.find_last_of("/\\");
     ASSERT_NE(found, std::string::npos);
@@ -279,12 +276,6 @@
     product_device_name_.clear();
     product_device_ctrl_name_.clear();
 
-    // Create a COW device. Number of sectors is chosen random which can
-    // hold at least 400MB of data
-
-    int err = ioctl(product_fd_.get(), BLKGETSIZE, &product_blksize_);
-    ASSERT_GE(err, 0);
-
     std::string str(cow->path);
     std::size_t found = str.find_last_of("/\\");
     ASSERT_NE(found, std::string::npos);
@@ -296,19 +287,29 @@
     system(cmd.c_str());
 }
 
+void SnapuserdTest::InitCowDevices() {
+    system_blksize_ = client_->InitDmUserCow(system_device_ctrl_name_, cow_system_->path,
+                                             system_a_loop_->device());
+    ASSERT_NE(system_blksize_, 0);
+
+    product_blksize_ = client_->InitDmUserCow(product_device_ctrl_name_, cow_product_->path,
+                                              product_a_loop_->device());
+    ASSERT_NE(product_blksize_, 0);
+}
+
+void SnapuserdTest::InitDaemon() {
+    bool ok = client_->AttachDmUser(system_device_ctrl_name_);
+    ASSERT_TRUE(ok);
+
+    ok = client_->AttachDmUser(product_device_ctrl_name_);
+    ASSERT_TRUE(ok);
+}
+
 void SnapuserdTest::StartSnapuserdDaemon() {
     ASSERT_TRUE(EnsureSnapuserdStarted());
 
     client_ = SnapuserdClient::Connect(kSnapuserdSocket, 5s);
     ASSERT_NE(client_, nullptr);
-
-    bool ok = client_->InitializeSnapuserd(cow_system_->path, system_a_loop_->device(),
-                                           GetSystemControlPath());
-    ASSERT_TRUE(ok);
-
-    ok = client_->InitializeSnapuserd(cow_product_->path, product_a_loop_->device(),
-                                      GetProductControlPath());
-    ASSERT_TRUE(ok);
 }
 
 void SnapuserdTest::CreateSnapshotDevices() {
@@ -434,10 +435,13 @@
     CreateCowDevice(cow_system_);
     CreateCowDevice(cow_product_);
 
+    StartSnapuserdDaemon();
+    InitCowDevices();
+
     CreateSystemDmUser(cow_system_);
     CreateProductDmUser(cow_product_);
 
-    StartSnapuserdDaemon();
+    InitDaemon();
 
     CreateSnapshotDevices();
 
@@ -451,42 +455,9 @@
 
     snapshot_fd.reset(-1);
 
-    // Sequence of operations for transition
-    CreateCowDevice(cow_system_1_);
-    CreateCowDevice(cow_product_1_);
-
-    // Create dm-user which creates new control devices
-    CreateSystemDmUser(cow_system_1_);
-    CreateProductDmUser(cow_product_1_);
-
-    // Send the path information to second stage daemon through vector
-    std::vector<std::vector<std::string>> vec{
-            {cow_system_1_->path, system_a_loop_->device(), GetSystemControlPath()},
-            {cow_product_1_->path, product_a_loop_->device(), GetProductControlPath()}};
-
-    // TODO: This is not switching snapshot device but creates a new table;
-    // Second stage daemon will be ready to serve the IO request. From now
-    // onwards, we can go ahead and shutdown the first stage daemon
-    SwitchSnapshotDevices();
-
     DeleteDmUser(cow_system_, "system-snapshot");
     DeleteDmUser(cow_product_, "product-snapshot");
 
-    // Test the IO again with the second stage daemon
-    snapshot_fd.reset(open("/dev/block/mapper/system-snapshot-1", O_RDONLY));
-    ASSERT_TRUE(snapshot_fd > 0);
-    TestIO(snapshot_fd, system_buffer_);
-
-    snapshot_fd.reset(open("/dev/block/mapper/product-snapshot-1", O_RDONLY));
-    ASSERT_TRUE(snapshot_fd > 0);
-    TestIO(snapshot_fd, product_buffer_);
-
-    snapshot_fd.reset(-1);
-
-    DeleteDmUser(cow_system_1_, "system-snapshot-1");
-    DeleteDmUser(cow_product_1_, "product-snapshot-1");
-
-    // Stop the second stage daemon
     ASSERT_TRUE(client_->StopSnapuserd());
 }
 
diff --git a/fs_mgr/libsnapshot/cow_writer.cpp b/fs_mgr/libsnapshot/cow_writer.cpp
index ec2dc96..957ba35 100644
--- a/fs_mgr/libsnapshot/cow_writer.cpp
+++ b/fs_mgr/libsnapshot/cow_writer.cpp
@@ -91,6 +91,7 @@
     header_.header_size = sizeof(CowHeader);
     header_.footer_size = sizeof(CowFooter);
     header_.block_size = options_.block_size;
+    header_.num_merge_ops = 0;
     footer_ = {};
     footer_.op.data_length = 64;
     footer_.op.type = kCowFooterOp;
@@ -110,27 +111,45 @@
     return true;
 }
 
-bool CowWriter::Initialize(unique_fd&& fd, OpenMode mode) {
-    owned_fd_ = std::move(fd);
-    return Initialize(borrowed_fd{owned_fd_}, mode);
+bool CowWriter::SetFd(android::base::borrowed_fd fd) {
+    if (fd.get() < 0) {
+        owned_fd_.reset(open("/dev/null", O_RDWR | O_CLOEXEC));
+        if (owned_fd_ < 0) {
+            PLOG(ERROR) << "open /dev/null failed";
+            return false;
+        }
+        fd_ = owned_fd_;
+        is_dev_null_ = true;
+    } else {
+        fd_ = fd;
+
+        struct stat stat;
+        if (fstat(fd.get(), &stat) < 0) {
+            PLOG(ERROR) << "fstat failed";
+            return false;
+        }
+        is_block_device_ = S_ISBLK(stat.st_mode);
+    }
+    return true;
 }
 
-bool CowWriter::Initialize(borrowed_fd fd, OpenMode mode) {
+void CowWriter::InitializeMerge(borrowed_fd fd, CowHeader* header) {
     fd_ = fd;
+    memcpy(&header_, header, sizeof(CowHeader));
+    merge_in_progress_ = true;
+}
 
-    if (!ParseOptions()) {
+bool CowWriter::Initialize(unique_fd&& fd) {
+    owned_fd_ = std::move(fd);
+    return Initialize(borrowed_fd{owned_fd_});
+}
+
+bool CowWriter::Initialize(borrowed_fd fd) {
+    if (!SetFd(fd) || !ParseOptions()) {
         return false;
     }
 
-    switch (mode) {
-        case OpenMode::WRITE:
-            return OpenForWrite();
-        case OpenMode::APPEND:
-            return OpenForAppend();
-        default:
-            LOG(ERROR) << "Unknown open mode in CowWriter";
-            return false;
-    }
+    return OpenForWrite();
 }
 
 bool CowWriter::InitializeAppend(android::base::unique_fd&& fd, uint64_t label) {
@@ -139,9 +158,7 @@
 }
 
 bool CowWriter::InitializeAppend(android::base::borrowed_fd fd, uint64_t label) {
-    fd_ = fd;
-
-    if (!ParseOptions()) {
+    if (!SetFd(fd) || !ParseOptions()) {
         return false;
     }
 
@@ -171,95 +188,35 @@
     return true;
 }
 
-bool CowWriter::OpenForAppend() {
-    auto reader = std::make_unique<CowReader>();
-    bool incomplete = false;
-    std::queue<CowOperation> toAdd;
-    if (!reader->Parse(fd_) || !reader->GetHeader(&header_)) {
-        return false;
-    }
-    incomplete = !reader->GetFooter(&footer_);
-
-    options_.block_size = header_.block_size;
-
-    // Reset this, since we're going to reimport all operations.
-    footer_.op.num_ops = 0;
-    next_op_pos_ = sizeof(header_);
-
-    auto iter = reader->GetOpIter();
-    while (!iter->Done()) {
-        CowOperation op = iter->Get();
-        if (op.type == kCowFooterOp) break;
-        if (incomplete) {
-            // Last operation translation may be corrupt. Wait to add it.
-            if (op.type == kCowLabelOp) {
-                while (!toAdd.empty()) {
-                    AddOperation(toAdd.front());
-                    toAdd.pop();
-                }
-            }
-            toAdd.push(op);
-        } else {
-            AddOperation(op);
-        }
-        iter->Next();
-    }
-
-    // Free reader so we own the descriptor position again.
-    reader = nullptr;
-
-    // Position for new writing
-    if (ftruncate(fd_.get(), next_op_pos_) != 0) {
-        PLOG(ERROR) << "Failed to trim file";
-        return false;
-    }
-    if (lseek(fd_.get(), 0, SEEK_END) < 0) {
-        PLOG(ERROR) << "lseek failed";
-        return false;
-    }
-    return true;
-}
-
 bool CowWriter::OpenForAppend(uint64_t label) {
     auto reader = std::make_unique<CowReader>();
     std::queue<CowOperation> toAdd;
-    if (!reader->Parse(fd_) || !reader->GetHeader(&header_)) {
+
+    if (!reader->Parse(fd_, {label}) || !reader->GetHeader(&header_)) {
         return false;
     }
 
     options_.block_size = header_.block_size;
-    bool found_label = false;
 
     // Reset this, since we're going to reimport all operations.
     footer_.op.num_ops = 0;
     next_op_pos_ = sizeof(header_);
+    ops_.resize(0);
 
     auto iter = reader->GetOpIter();
     while (!iter->Done()) {
-        CowOperation op = iter->Get();
-        if (op.type == kCowFooterOp) break;
-        if (op.type == kCowLabelOp) {
-            if (found_label) break;
-            if (op.source == label) found_label = true;
-        }
-        AddOperation(op);
+        AddOperation(iter->Get());
         iter->Next();
     }
 
-    if (!found_label) {
-        PLOG(ERROR) << "Failed to find last label";
-        return false;
-    }
-
     // Free reader so we own the descriptor position again.
     reader = nullptr;
 
-    // Position for new writing
-    if (ftruncate(fd_.get(), next_op_pos_) != 0) {
-        PLOG(ERROR) << "Failed to trim file";
+    // Remove excess data
+    if (!Truncate(next_op_pos_)) {
         return false;
     }
-    if (lseek(fd_.get(), 0, SEEK_END) < 0) {
+    if (lseek(fd_.get(), next_op_pos_, SEEK_SET) < 0) {
         PLOG(ERROR) << "lseek failed";
         return false;
     }
@@ -267,6 +224,7 @@
 }
 
 bool CowWriter::EmitCopy(uint64_t new_block, uint64_t old_block) {
+    CHECK(!merge_in_progress_);
     CowOperation op = {};
     op.type = kCowCopyOp;
     op.new_block = new_block;
@@ -277,6 +235,7 @@
 bool CowWriter::EmitRawBlocks(uint64_t new_block_start, const void* data, size_t size) {
     const uint8_t* iter = reinterpret_cast<const uint8_t*>(data);
     uint64_t pos;
+    CHECK(!merge_in_progress_);
     for (size_t i = 0; i < size / header_.block_size; i++) {
         CowOperation op = {};
         op.type = kCowReplaceOp;
@@ -315,6 +274,7 @@
 }
 
 bool CowWriter::EmitZeroBlocks(uint64_t new_block_start, uint64_t num_blocks) {
+    CHECK(!merge_in_progress_);
     for (uint64_t i = 0; i < num_blocks; i++) {
         CowOperation op = {};
         op.type = kCowZeroOp;
@@ -326,10 +286,11 @@
 }
 
 bool CowWriter::EmitLabel(uint64_t label) {
+    CHECK(!merge_in_progress_);
     CowOperation op = {};
     op.type = kCowLabelOp;
     op.source = label;
-    return WriteOperation(op);
+    return WriteOperation(op) && Sync();
 }
 
 std::basic_string<uint8_t> CowWriter::Compress(const void* data, size_t length) {
@@ -384,7 +345,7 @@
 }
 
 bool CowWriter::Finalize() {
-    footer_.op.ops_size = ops_.size() + sizeof(footer_.op);
+    footer_.op.ops_size = ops_.size();
     uint64_t pos;
 
     if (!GetDataPos(&pos)) {
@@ -408,7 +369,7 @@
         PLOG(ERROR) << "lseek ops failed";
         return false;
     }
-    return true;
+    return Sync();
 }
 
 uint64_t CowWriter::GetCowSize() {
@@ -429,10 +390,11 @@
     if (!android::base::WriteFully(fd_, reinterpret_cast<const uint8_t*>(&op), sizeof(op))) {
         return false;
     }
-    if (data != NULL && size > 0)
+    if (data != nullptr && size > 0) {
         if (!WriteRawData(data, size)) return false;
+    }
     AddOperation(op);
-    return !fsync(fd_.get());
+    return true;
 }
 
 void CowWriter::AddOperation(const CowOperation& op) {
@@ -448,5 +410,45 @@
     return true;
 }
 
+bool CowWriter::Sync() {
+    if (is_dev_null_) {
+        return true;
+    }
+    if (fsync(fd_.get()) < 0) {
+        PLOG(ERROR) << "fsync failed";
+        return false;
+    }
+    return true;
+}
+
+bool CowWriter::CommitMerge(int merged_ops) {
+    CHECK(merge_in_progress_);
+    header_.num_merge_ops += merged_ops;
+
+    if (lseek(fd_.get(), 0, SEEK_SET) < 0) {
+        PLOG(ERROR) << "lseek failed";
+        return false;
+    }
+
+    if (!android::base::WriteFully(fd_, reinterpret_cast<const uint8_t*>(&header_),
+                                   sizeof(header_))) {
+        PLOG(ERROR) << "WriteFully failed";
+        return false;
+    }
+
+    return Sync();
+}
+
+bool CowWriter::Truncate(off_t length) {
+    if (is_dev_null_ || is_block_device_) {
+        return true;
+    }
+    if (ftruncate(fd_.get(), length) < 0) {
+        PLOG(ERROR) << "Failed to truncate.";
+        return false;
+    }
+    return true;
+}
+
 }  // namespace snapshot
 }  // namespace android
diff --git a/fs_mgr/libsnapshot/device_info.cpp b/fs_mgr/libsnapshot/device_info.cpp
index 0e90100..8770255 100644
--- a/fs_mgr/libsnapshot/device_info.cpp
+++ b/fs_mgr/libsnapshot/device_info.cpp
@@ -120,5 +120,9 @@
 #endif
 }
 
+std::string DeviceInfo::GetSnapuserdFirstStagePidVar() const {
+    return kSnapuserdFirstStagePidVar;
+}
+
 }  // namespace snapshot
 }  // namespace android
diff --git a/fs_mgr/libsnapshot/device_info.h b/fs_mgr/libsnapshot/device_info.h
index d8d3d91..1f08860 100644
--- a/fs_mgr/libsnapshot/device_info.h
+++ b/fs_mgr/libsnapshot/device_info.h
@@ -39,6 +39,7 @@
     bool SetBootControlMergeStatus(MergeStatus status) override;
     bool SetSlotAsUnbootable(unsigned int slot) override;
     bool IsRecovery() const override;
+    std::string GetSnapuserdFirstStagePidVar() const override;
 
   private:
     bool EnsureBootHal();
diff --git a/fs_mgr/libsnapshot/include/libsnapshot/cow_format.h b/fs_mgr/libsnapshot/include/libsnapshot/cow_format.h
index 2291e30..80766ff 100644
--- a/fs_mgr/libsnapshot/include/libsnapshot/cow_format.h
+++ b/fs_mgr/libsnapshot/include/libsnapshot/cow_format.h
@@ -59,6 +59,9 @@
 
     // The size of block operations, in bytes.
     uint32_t block_size;
+
+    // Tracks merge operations completed
+    uint64_t num_merge_ops;
 } __attribute__((packed));
 
 // This structure is the same size of a normal Operation, but is repurposed for the footer.
diff --git a/fs_mgr/libsnapshot/include/libsnapshot/cow_reader.h b/fs_mgr/libsnapshot/include/libsnapshot/cow_reader.h
index b863ff2..be69225 100644
--- a/fs_mgr/libsnapshot/include/libsnapshot/cow_reader.h
+++ b/fs_mgr/libsnapshot/include/libsnapshot/cow_reader.h
@@ -18,6 +18,7 @@
 
 #include <functional>
 #include <memory>
+#include <optional>
 
 #include <android-base/unique_fd.h>
 #include <libsnapshot/cow_format.h>
@@ -26,6 +27,7 @@
 namespace snapshot {
 
 class ICowOpIter;
+class ICowOpReverseIter;
 
 // A ByteSink object handles requests for a buffer of a specific size. It
 // always owns the underlying buffer. It's designed to minimize potential
@@ -73,6 +75,9 @@
     // Return an iterator for retrieving CowOperation entries.
     virtual std::unique_ptr<ICowOpIter> GetOpIter() = 0;
 
+    // Return an reverse iterator for retrieving CowOperation entries.
+    virtual std::unique_ptr<ICowOpReverseIter> GetRevOpIter() = 0;
+
     // Get decoded bytes from the data section, handling any decompression.
     // All retrieved data is passed to the sink.
     virtual bool ReadData(const CowOperation& op, IByteSink* sink) = 0;
@@ -93,12 +98,29 @@
     virtual void Next() = 0;
 };
 
+// Reverse Iterate over a sequence of COW operations.
+class ICowOpReverseIter {
+  public:
+    virtual ~ICowOpReverseIter() {}
+
+    // True if there are more items to read, false otherwise.
+    virtual bool Done() = 0;
+
+    // Read the current operation.
+    virtual const CowOperation& Get() = 0;
+
+    // Advance to the next item.
+    virtual void Next() = 0;
+};
+
 class CowReader : public ICowReader {
   public:
     CowReader();
 
-    bool Parse(android::base::unique_fd&& fd);
-    bool Parse(android::base::borrowed_fd fd);
+    // Parse the COW, optionally, up to the given label. If no label is
+    // specified, the COW must have an intact footer.
+    bool Parse(android::base::unique_fd&& fd, std::optional<uint64_t> label = {});
+    bool Parse(android::base::borrowed_fd fd, std::optional<uint64_t> label = {});
 
     bool GetHeader(CowHeader* header) override;
     bool GetFooter(CowFooter* footer) override;
@@ -107,23 +129,26 @@
 
     // Create a CowOpIter object which contains footer_.num_ops
     // CowOperation objects. Get() returns a unique CowOperation object
-    // whose lifetime depends on the CowOpIter object
+    // whose lifetime depends on the CowOpIter object; the return
+    // value of these will never be null.
     std::unique_ptr<ICowOpIter> GetOpIter() override;
+    std::unique_ptr<ICowOpReverseIter> GetRevOpIter() override;
+
     bool ReadData(const CowOperation& op, IByteSink* sink) override;
 
     bool GetRawBytes(uint64_t offset, void* buffer, size_t len, size_t* read);
 
+    void UpdateMergeProgress(uint64_t merge_ops) { header_.num_merge_ops += merge_ops; }
+
   private:
-    bool ParseOps();
+    bool ParseOps(std::optional<uint64_t> label);
 
     android::base::unique_fd owned_fd_;
     android::base::borrowed_fd fd_;
     CowHeader header_;
-    CowFooter footer_;
+    std::optional<CowFooter> footer_;
     uint64_t fd_size_;
-    bool has_footer_;
-    uint64_t last_label_;
-    bool has_last_label_;
+    std::optional<uint64_t> last_label_;
     std::shared_ptr<std::vector<CowOperation>> ops_;
 };
 
diff --git a/fs_mgr/libsnapshot/include/libsnapshot/cow_writer.h b/fs_mgr/libsnapshot/include/libsnapshot/cow_writer.h
index c031d63..e9320b0 100644
--- a/fs_mgr/libsnapshot/include/libsnapshot/cow_writer.h
+++ b/fs_mgr/libsnapshot/include/libsnapshot/cow_writer.h
@@ -82,22 +82,24 @@
 
 class CowWriter : public ICowWriter {
   public:
-    enum class OpenMode { WRITE, APPEND };
-
     explicit CowWriter(const CowOptions& options);
 
     // Set up the writer.
-    // If opening for write, the file starts from the beginning.
-    // If opening for append, if the file has a footer, we start appending to the last op.
-    // If the footer isn't found, the last label is considered corrupt, and dropped.
-    bool Initialize(android::base::unique_fd&& fd, OpenMode mode = OpenMode::WRITE);
-    bool Initialize(android::base::borrowed_fd fd, OpenMode mode = OpenMode::WRITE);
+    // The file starts from the beginning.
+    //
+    // If fd is < 0, the CowWriter will be opened against /dev/null. This is for
+    // computing COW sizes without using storage space.
+    bool Initialize(android::base::unique_fd&& fd);
+    bool Initialize(android::base::borrowed_fd fd);
     // Set up a writer, assuming that the given label is the last valid label.
     // This will result in dropping any labels that occur after the given on, and will fail
     // if the given label does not appear.
     bool InitializeAppend(android::base::unique_fd&&, uint64_t label);
     bool InitializeAppend(android::base::borrowed_fd fd, uint64_t label);
 
+    void InitializeMerge(android::base::borrowed_fd fd, CowHeader* header);
+    bool CommitMerge(int merged_ops);
+
     bool Finalize() override;
 
     uint64_t GetCowSize() override;
@@ -112,15 +114,17 @@
     void SetupHeaders();
     bool ParseOptions();
     bool OpenForWrite();
-    bool OpenForAppend();
     bool OpenForAppend(uint64_t label);
-    bool ImportOps(std::unique_ptr<ICowOpIter> iter);
     bool GetDataPos(uint64_t* pos);
     bool WriteRawData(const void* data, size_t size);
     bool WriteOperation(const CowOperation& op, const void* data = nullptr, size_t size = 0);
     void AddOperation(const CowOperation& op);
     std::basic_string<uint8_t> Compress(const void* data, size_t length);
 
+    bool SetFd(android::base::borrowed_fd fd);
+    bool Sync();
+    bool Truncate(off_t length);
+
   private:
     android::base::unique_fd owned_fd_;
     android::base::borrowed_fd fd_;
@@ -128,6 +132,9 @@
     CowFooter footer_{};
     int compression_ = 0;
     uint64_t next_op_pos_ = 0;
+    bool is_dev_null_ = false;
+    bool merge_in_progress_ = false;
+    bool is_block_device_ = false;
 
     // :TODO: this is not efficient, but stringstream ubsan aborts because some
     // bytes overflow a signed char.
diff --git a/fs_mgr/libsnapshot/include/libsnapshot/mock_device_info.h b/fs_mgr/libsnapshot/include/libsnapshot/mock_device_info.h
index ef9d648..d5c263d 100644
--- a/fs_mgr/libsnapshot/include/libsnapshot/mock_device_info.h
+++ b/fs_mgr/libsnapshot/include/libsnapshot/mock_device_info.h
@@ -32,6 +32,7 @@
     MOCK_METHOD(bool, SetBootControlMergeStatus, (MergeStatus status), (override));
     MOCK_METHOD(bool, SetSlotAsUnbootable, (unsigned int slot), (override));
     MOCK_METHOD(bool, IsRecovery, (), (const, override));
+    MOCK_METHOD(std::string, GetSnapuserdFirstStagePidVar, (), (const, override));
 };
 
 }  // namespace android::snapshot
diff --git a/fs_mgr/libsnapshot/include/libsnapshot/snapshot.h b/fs_mgr/libsnapshot/include/libsnapshot/snapshot.h
index 8bed1b9..351dce7 100644
--- a/fs_mgr/libsnapshot/include/libsnapshot/snapshot.h
+++ b/fs_mgr/libsnapshot/include/libsnapshot/snapshot.h
@@ -107,6 +107,7 @@
         virtual bool SetSlotAsUnbootable(unsigned int slot) = 0;
         virtual bool IsRecovery() const = 0;
         virtual bool IsTestDevice() const { return false; }
+        virtual std::string GetSnapuserdFirstStagePidVar() const = 0;
     };
     virtual ~ISnapshotManager() = default;
 
diff --git a/fs_mgr/libsnapshot/include/libsnapshot/snapshot_writer.h b/fs_mgr/libsnapshot/include/libsnapshot/snapshot_writer.h
index 4732c2d..bf5ce8b 100644
--- a/fs_mgr/libsnapshot/include/libsnapshot/snapshot_writer.h
+++ b/fs_mgr/libsnapshot/include/libsnapshot/snapshot_writer.h
@@ -42,9 +42,9 @@
     // Open the writer in write mode (no append).
     virtual bool Initialize() = 0;
 
-    // Open the writer in append mode, optionally with the last label to resume
+    // Open the writer in append mode, with the last label to resume
     // from. See CowWriter::InitializeAppend.
-    virtual bool InitializeAppend(std::optional<uint64_t> label = {}) = 0;
+    virtual bool InitializeAppend(uint64_t label) = 0;
 
     virtual std::unique_ptr<FileDescriptor> OpenReader() = 0;
 
@@ -66,7 +66,7 @@
     bool SetCowDevice(android::base::unique_fd&& cow_device);
 
     bool Initialize() override;
-    bool InitializeAppend(std::optional<uint64_t> label = {}) override;
+    bool InitializeAppend(uint64_t label) override;
     bool Finalize() override;
     uint64_t GetCowSize() override;
     std::unique_ptr<FileDescriptor> OpenReader() override;
@@ -92,7 +92,7 @@
     void SetSnapshotDevice(android::base::unique_fd&& snapshot_fd, uint64_t cow_size);
 
     bool Initialize() override { return true; }
-    bool InitializeAppend(std::optional<uint64_t>) override { return true; }
+    bool InitializeAppend(uint64_t) override { return true; }
 
     bool Finalize() override;
     uint64_t GetCowSize() override { return cow_size_; }
diff --git a/fs_mgr/libsnapshot/include/libsnapshot/snapuserd.h b/fs_mgr/libsnapshot/include/libsnapshot/snapuserd.h
index 80f87d9..24b44fa 100644
--- a/fs_mgr/libsnapshot/include/libsnapshot/snapuserd.h
+++ b/fs_mgr/libsnapshot/include/libsnapshot/snapuserd.h
@@ -24,6 +24,7 @@
 #include <limits>
 #include <string>
 #include <thread>
+#include <unordered_map>
 #include <vector>
 
 #include <android-base/file.h>
@@ -60,52 +61,67 @@
 
 class Snapuserd final {
   public:
-    Snapuserd(const std::string& in_cow_device, const std::string& in_backing_store_device,
-              const std::string& in_control_device)
-        : cow_device_(in_cow_device),
-          backing_store_device_(in_backing_store_device),
-          control_device_(in_control_device),
-          metadata_read_done_(false) {}
-
-    bool Init();
-    int Run();
-    int ReadDmUserHeader();
-    int WriteDmUserPayload(size_t size);
-    int ConstructKernelCowHeader();
-    int ReadMetadata();
-    int ZerofillDiskExceptions(size_t read_size);
-    int ReadDiskExceptions(chunk_t chunk, size_t size);
-    int ReadData(chunk_t chunk, size_t size);
-
+    Snapuserd(const std::string& misc_name, const std::string& cow_device,
+              const std::string& backing_device);
+    bool InitBackingAndControlDevice();
+    bool InitCowDevice();
+    bool Run();
     const std::string& GetControlDevicePath() { return control_device_; }
+    const std::string& GetMiscName() { return misc_name_; }
+    uint64_t GetNumSectors() { return num_sectors_; }
+    bool IsAttached() const { return ctrl_fd_ >= 0; }
 
   private:
-    int ProcessReplaceOp(const CowOperation* cow_op);
-    int ProcessCopyOp(const CowOperation* cow_op);
-    int ProcessZeroOp();
+    bool ReadDmUserHeader();
+    bool ReadDmUserPayload(void* buffer, size_t size);
+    bool WriteDmUserPayload(size_t size);
+    void ConstructKernelCowHeader();
+    bool ReadMetadata();
+    bool ZerofillDiskExceptions(size_t read_size);
+    bool ReadDiskExceptions(chunk_t chunk, size_t size);
+    bool ReadData(chunk_t chunk, size_t size);
+    bool IsChunkIdMetadata(chunk_t chunk);
+    chunk_t GetNextAllocatableChunkId(chunk_t chunk);
+
+    bool ProcessReplaceOp(const CowOperation* cow_op);
+    bool ProcessCopyOp(const CowOperation* cow_op);
+    bool ProcessZeroOp();
+
+    loff_t GetMergeStartOffset(void* merged_buffer, void* unmerged_buffer,
+                               int* unmerged_exceptions);
+    int GetNumberOfMergedOps(void* merged_buffer, void* unmerged_buffer, loff_t offset,
+                             int unmerged_exceptions);
+    bool AdvanceMergedOps(int merged_ops_cur_iter);
+    bool ProcessMergeComplete(chunk_t chunk, void* buffer);
+    sector_t ChunkToSector(chunk_t chunk) { return chunk << CHUNK_SHIFT; }
+    chunk_t SectorToChunk(sector_t sector) { return sector >> CHUNK_SHIFT; }
 
     std::string cow_device_;
     std::string backing_store_device_;
     std::string control_device_;
+    std::string misc_name_;
 
     unique_fd cow_fd_;
     unique_fd backing_store_fd_;
     unique_fd ctrl_fd_;
 
     uint32_t exceptions_per_area_;
+    uint64_t num_sectors_;
 
     std::unique_ptr<ICowOpIter> cowop_iter_;
+    std::unique_ptr<ICowOpReverseIter> cowop_riter_;
     std::unique_ptr<CowReader> reader_;
+    std::unique_ptr<CowWriter> writer_;
 
     // Vector of disk exception which is a
     // mapping of old-chunk to new-chunk
     std::vector<std::unique_ptr<uint8_t[]>> vec_;
 
-    // Index - Chunk ID
+    // Key - Chunk ID
     // Value - cow operation
-    std::vector<const CowOperation*> chunk_vec_;
+    std::unordered_map<chunk_t, const CowOperation*> chunk_map_;
 
-    bool metadata_read_done_;
+    bool metadata_read_done_ = false;
     BufferSink bufsink_;
 };
 
diff --git a/fs_mgr/libsnapshot/include/libsnapshot/snapuserd_client.h b/fs_mgr/libsnapshot/include/libsnapshot/snapuserd_client.h
index aaec229..0e9ba9e 100644
--- a/fs_mgr/libsnapshot/include/libsnapshot/snapuserd_client.h
+++ b/fs_mgr/libsnapshot/include/libsnapshot/snapuserd_client.h
@@ -58,9 +58,19 @@
                                                     std::chrono::milliseconds timeout_ms);
 
     bool StopSnapuserd();
-    int RestartSnapuserd(std::vector<std::vector<std::string>>& vec);
-    bool InitializeSnapuserd(const std::string& cow_device, const std::string& backing_device,
-                             const std::string& control_device);
+
+    // Initializing a snapuserd handler is a three-step process:
+    //
+    //  1. Client invokes InitDmUserCow. This creates the snapuserd handler and validates the
+    //     COW. The number of sectors required for the dm-user target is returned.
+    //  2. Client creates the device-mapper device with the dm-user target.
+    //  3. Client calls AttachControlDevice.
+    //
+    // The misc_name must be the "misc_name" given to dm-user in step 2.
+    //
+    uint64_t InitDmUserCow(const std::string& misc_name, const std::string& cow_device,
+                           const std::string& backing_device);
+    bool AttachDmUser(const std::string& misc_name);
 
     // Wait for snapuserd to disassociate with a dm-user control device. This
     // must ONLY be called if the control device has already been deleted.
diff --git a/fs_mgr/libsnapshot/include/libsnapshot/snapuserd_kernel.h b/fs_mgr/libsnapshot/include/libsnapshot/snapuserd_kernel.h
index 1037c12..e8dbe6e 100644
--- a/fs_mgr/libsnapshot/include/libsnapshot/snapuserd_kernel.h
+++ b/fs_mgr/libsnapshot/include/libsnapshot/snapuserd_kernel.h
@@ -17,6 +17,13 @@
 namespace android {
 namespace snapshot {
 
+#define DM_USER_REQ_MAP_READ 0
+#define DM_USER_REQ_MAP_WRITE 1
+
+#define DM_USER_RESP_SUCCESS 0
+#define DM_USER_RESP_ERROR 1
+#define DM_USER_RESP_UNSUPPORTED 2
+
 // Kernel COW header fields
 static constexpr uint32_t SNAP_MAGIC = 0x70416e53;
 
diff --git a/fs_mgr/libsnapshot/include/libsnapshot/snapuserd_server.h b/fs_mgr/libsnapshot/include/libsnapshot/snapuserd_server.h
index 181ee33..cadfd71 100644
--- a/fs_mgr/libsnapshot/include/libsnapshot/snapuserd_server.h
+++ b/fs_mgr/libsnapshot/include/libsnapshot/snapuserd_server.h
@@ -36,6 +36,7 @@
 static constexpr uint32_t MAX_PACKET_SIZE = 512;
 
 enum class DaemonOperations {
+    INIT,
     START,
     QUERY,
     STOP,
@@ -55,7 +56,7 @@
     const std::unique_ptr<Snapuserd>& snapuserd() const { return snapuserd_; }
     std::thread& thread() { return thread_; }
 
-    const std::string& GetControlDevice() const;
+    const std::string& GetMiscName() const;
 };
 
 class Stoppable {
@@ -84,7 +85,9 @@
     std::vector<struct pollfd> watched_fds_;
 
     std::mutex lock_;
-    std::vector<std::unique_ptr<DmUserHandler>> dm_users_;
+
+    using HandlerList = std::vector<std::shared_ptr<DmUserHandler>>;
+    HandlerList dm_users_;
 
     void AddWatchedFd(android::base::borrowed_fd fd);
     void AcceptClient();
@@ -94,7 +97,7 @@
     bool Receivemsg(android::base::borrowed_fd fd, const std::string& str);
 
     void ShutdownThreads();
-    bool WaitForDelete(const std::string& control_device);
+    bool RemoveHandler(const std::string& control_device, bool wait);
     DaemonOperations Resolveop(std::string& input);
     std::string GetDaemonStatus();
     void Parsemsg(std::string const& msg, const char delim, std::vector<std::string>& out);
@@ -102,11 +105,11 @@
     void SetTerminating() { terminating_ = true; }
     bool IsTerminating() { return terminating_; }
 
-    void RunThread(DmUserHandler* handler);
+    void RunThread(std::shared_ptr<DmUserHandler> handler);
 
-    // Remove a DmUserHandler from dm_users_, searching by its control device.
-    // If none is found, return nullptr.
-    std::unique_ptr<DmUserHandler> RemoveHandler(const std::string& control_device);
+    // Find a DmUserHandler within a lock.
+    HandlerList::iterator FindHandler(std::lock_guard<std::mutex>* proof_of_lock,
+                                      const std::string& misc_name);
 
   public:
     SnapuserdServer() { terminating_ = false; }
diff --git a/fs_mgr/libsnapshot/include_test/libsnapshot/test_helpers.h b/fs_mgr/libsnapshot/include_test/libsnapshot/test_helpers.h
index 7aef086..1d1510a 100644
--- a/fs_mgr/libsnapshot/include_test/libsnapshot/test_helpers.h
+++ b/fs_mgr/libsnapshot/include_test/libsnapshot/test_helpers.h
@@ -96,6 +96,7 @@
         return true;
     }
     bool IsTestDevice() const override { return true; }
+    std::string GetSnapuserdFirstStagePidVar() const override { return {}; }
 
     bool IsSlotUnbootable(uint32_t slot) { return unbootable_slots_.count(slot) != 0; }
 
diff --git a/fs_mgr/libsnapshot/partition_cow_creator.cpp b/fs_mgr/libsnapshot/partition_cow_creator.cpp
index 0df5664..da6fc9d 100644
--- a/fs_mgr/libsnapshot/partition_cow_creator.cpp
+++ b/fs_mgr/libsnapshot/partition_cow_creator.cpp
@@ -18,6 +18,7 @@
 
 #include <android-base/logging.h>
 #include <android/snapshot/snapshot.pb.h>
+#include <storage_literals/storage_literals.h>
 
 #include "dm_snapshot_internals.h"
 #include "utility.h"
@@ -34,6 +35,10 @@
 namespace android {
 namespace snapshot {
 
+static constexpr uint64_t kBlockSize = 4096;
+
+using namespace android::storage_literals;
+
 // Intersect two linear extents. If no intersection, return an extent with length 0.
 static std::unique_ptr<Extent> Intersect(Extent* target_extent, Extent* existing_extent) {
     // Convert target_extent and existing_extent to linear extents. Zero extents
@@ -138,6 +143,22 @@
 }
 
 uint64_t PartitionCowCreator::GetCowSize() {
+    if (compression_enabled) {
+        if (update == nullptr || !update->has_estimate_cow_size()) {
+            LOG(ERROR) << "Update manifest does not include a COW size";
+            return 0;
+        }
+
+        // Add an extra 2MB of wiggle room for any minor differences in labels/metadata
+        // that might come up.
+        auto size = update->estimate_cow_size() + 2_MiB;
+
+        // Align to nearest block.
+        size += kBlockSize - 1;
+        size &= ~(kBlockSize - 1);
+        return size;
+    }
+
     // WARNING: The origin partition should be READ-ONLY
     const uint64_t logical_block_size = current_metadata->logical_block_size();
     const unsigned int sectors_per_block = logical_block_size / kSectorSize;
@@ -149,9 +170,9 @@
         WriteExtent(&sc, de, sectors_per_block);
     }
 
-    if (operations == nullptr) return sc.cow_size_bytes();
+    if (update == nullptr) return sc.cow_size_bytes();
 
-    for (const auto& iop : *operations) {
+    for (const auto& iop : update->operations()) {
         const InstallOperation* written_op = &iop;
         InstallOperation buf;
         // Do not allocate space for extents that are going to be skipped
@@ -213,6 +234,9 @@
 
     LOG(INFO) << "Remaining free space for COW: " << free_region_length << " bytes";
     auto cow_size = GetCowSize();
+    if (!cow_size) {
+        return {};
+    }
 
     // Compute the COW partition size.
     uint64_t cow_partition_size = std::min(cow_size, free_region_length);
diff --git a/fs_mgr/libsnapshot/partition_cow_creator.h b/fs_mgr/libsnapshot/partition_cow_creator.h
index 699f9a1..64d186b 100644
--- a/fs_mgr/libsnapshot/partition_cow_creator.h
+++ b/fs_mgr/libsnapshot/partition_cow_creator.h
@@ -36,6 +36,7 @@
     using MetadataBuilder = android::fs_mgr::MetadataBuilder;
     using Partition = android::fs_mgr::Partition;
     using InstallOperation = chromeos_update_engine::InstallOperation;
+    using PartitionUpdate = chromeos_update_engine::PartitionUpdate;
     template <typename T>
     using RepeatedPtrField = google::protobuf::RepeatedPtrField<T>;
 
@@ -50,11 +51,13 @@
     MetadataBuilder* current_metadata = nullptr;
     // The suffix of the current slot.
     std::string current_suffix;
-    // List of operations to be applied on the partition.
-    const RepeatedPtrField<InstallOperation>* operations = nullptr;
+    // Partition information from the OTA manifest.
+    const PartitionUpdate* update = nullptr;
     // Extra extents that are going to be invalidated during the update
     // process.
     std::vector<ChromeOSExtent> extra_extents = {};
+    // True if compression is enabled.
+    bool compression_enabled = false;
 
     struct Return {
         SnapshotStatus snapshot_status;
diff --git a/fs_mgr/libsnapshot/partition_cow_creator_test.cpp b/fs_mgr/libsnapshot/partition_cow_creator_test.cpp
index 2970222..e4b476f 100644
--- a/fs_mgr/libsnapshot/partition_cow_creator_test.cpp
+++ b/fs_mgr/libsnapshot/partition_cow_creator_test.cpp
@@ -145,13 +145,15 @@
 
     auto cow_device_size = [](const std::vector<InstallOperation>& iopv, MetadataBuilder* builder_a,
                               MetadataBuilder* builder_b, Partition* system_b) {
-        RepeatedInstallOperationPtr riop(iopv.begin(), iopv.end());
+        PartitionUpdate update;
+        *update.mutable_operations() = RepeatedInstallOperationPtr(iopv.begin(), iopv.end());
+
         PartitionCowCreator creator{.target_metadata = builder_b,
                                     .target_suffix = "_b",
                                     .target_partition = system_b,
                                     .current_metadata = builder_a,
                                     .current_suffix = "_a",
-                                    .operations = &riop};
+                                    .update = &update};
 
         auto ret = creator.Run();
 
@@ -218,7 +220,7 @@
                                 .target_partition = system_b,
                                 .current_metadata = builder_a.get(),
                                 .current_suffix = "_a",
-                                .operations = nullptr};
+                                .update = nullptr};
 
     auto ret = creator.Run();
 
@@ -228,6 +230,58 @@
     ASSERT_EQ(0u, ret->snapshot_status.cow_partition_size());
 }
 
+TEST_F(PartitionCowCreatorTest, CompressionEnabled) {
+    constexpr uint64_t super_size = 1_MiB;
+    auto builder_a = MetadataBuilder::New(super_size, 1_KiB, 2);
+    ASSERT_NE(builder_a, nullptr);
+
+    auto builder_b = MetadataBuilder::New(super_size, 1_KiB, 2);
+    ASSERT_NE(builder_b, nullptr);
+    auto system_b = builder_b->AddPartition("system_b", LP_PARTITION_ATTR_READONLY);
+    ASSERT_NE(system_b, nullptr);
+    ASSERT_TRUE(builder_b->ResizePartition(system_b, 128_KiB));
+
+    PartitionUpdate update;
+    update.set_estimate_cow_size(256_KiB);
+
+    PartitionCowCreator creator{.target_metadata = builder_b.get(),
+                                .target_suffix = "_b",
+                                .target_partition = system_b,
+                                .current_metadata = builder_a.get(),
+                                .current_suffix = "_a",
+                                .compression_enabled = true,
+                                .update = &update};
+
+    auto ret = creator.Run();
+    ASSERT_TRUE(ret.has_value());
+    ASSERT_EQ(ret->snapshot_status.cow_file_size(), 1458176);
+}
+
+TEST_F(PartitionCowCreatorTest, CompressionWithNoManifest) {
+    constexpr uint64_t super_size = 1_MiB;
+    auto builder_a = MetadataBuilder::New(super_size, 1_KiB, 2);
+    ASSERT_NE(builder_a, nullptr);
+
+    auto builder_b = MetadataBuilder::New(super_size, 1_KiB, 2);
+    ASSERT_NE(builder_b, nullptr);
+    auto system_b = builder_b->AddPartition("system_b", LP_PARTITION_ATTR_READONLY);
+    ASSERT_NE(system_b, nullptr);
+    ASSERT_TRUE(builder_b->ResizePartition(system_b, 128_KiB));
+
+    PartitionUpdate update;
+
+    PartitionCowCreator creator{.target_metadata = builder_b.get(),
+                                .target_suffix = "_b",
+                                .target_partition = system_b,
+                                .current_metadata = builder_a.get(),
+                                .current_suffix = "_a",
+                                .compression_enabled = true,
+                                .update = nullptr};
+
+    auto ret = creator.Run();
+    ASSERT_FALSE(ret.has_value());
+}
+
 TEST(DmSnapshotInternals, CowSizeCalculator) {
     SKIP_IF_NON_VIRTUAL_AB();
 
diff --git a/fs_mgr/libsnapshot/snapshot.cpp b/fs_mgr/libsnapshot/snapshot.cpp
index f9bb0dd..6595707 100644
--- a/fs_mgr/libsnapshot/snapshot.cpp
+++ b/fs_mgr/libsnapshot/snapshot.cpp
@@ -75,7 +75,7 @@
 using chromeos_update_engine::DeltaArchiveManifest;
 using chromeos_update_engine::Extent;
 using chromeos_update_engine::FileDescriptor;
-using chromeos_update_engine::InstallOperation;
+using chromeos_update_engine::PartitionUpdate;
 template <typename T>
 using RepeatedPtrField = google::protobuf::RepeatedPtrField<T>;
 using std::chrono::duration_cast;
@@ -116,10 +116,6 @@
     metadata_dir_ = device_->GetMetadataDir();
 }
 
-static inline bool IsCompressionEnabled() {
-    return android::base::GetBoolProperty("ro.virtual_ab.compression.enabled", false);
-}
-
 static std::string GetCowName(const std::string& snapshot_name) {
     return snapshot_name + "-cow";
 }
@@ -390,24 +386,6 @@
 
     auto& dm = DeviceMapper::Instance();
 
-    // Use the size of the base device for the COW device. It doesn't really
-    // matter, it just needs to look similar enough so the kernel doesn't complain
-    // about alignment or being too small.
-    uint64_t base_sectors = 0;
-    {
-        unique_fd fd(open(base_device.c_str(), O_RDONLY | O_CLOEXEC));
-        if (fd < 0) {
-            PLOG(ERROR) << "open failed: " << base_device;
-            return false;
-        }
-        auto dev_size = get_block_device_size(fd);
-        if (!dev_size) {
-            PLOG(ERROR) << "Could not determine block device size: " << base_device;
-            return false;
-        }
-        base_sectors = dev_size / kSectorSize;
-    }
-
     // Use an extra decoration for first-stage init, so we can transition
     // to a new table entry in second-stage.
     std::string misc_name = name;
@@ -415,18 +393,29 @@
         misc_name += "-init";
     }
 
+    if (!EnsureSnapuserdConnected()) {
+        return false;
+    }
+
+    uint64_t base_sectors = snapuserd_client_->InitDmUserCow(misc_name, cow_file, base_device);
+    if (base_sectors == 0) {
+        LOG(ERROR) << "Failed to retrieve base_sectors from Snapuserd";
+        return false;
+    }
+
     DmTable table;
     table.Emplace<DmTargetUser>(0, base_sectors, misc_name);
     if (!dm.CreateDevice(name, table, path, timeout_ms)) {
         return false;
     }
 
-    if (!EnsureSnapuserdConnected()) {
+    auto control_device = "/dev/dm-user/" + misc_name;
+    if (!android::fs_mgr::WaitForFile(control_device, timeout_ms)) {
+        LOG(ERROR) << "Timed out waiting for dm-user misc device: " << control_device;
         return false;
     }
 
-    auto control_device = "/dev/dm-user/" + misc_name;
-    return snapuserd_client_->InitializeSnapuserd(cow_file, base_device, control_device);
+    return snapuserd_client_->AttachDmUser(misc_name);
 }
 
 bool SnapshotManager::MapSnapshot(LockedFile* lock, const std::string& name,
@@ -1326,28 +1315,34 @@
     size_t num_cows = 0;
     size_t ok_cows = 0;
     for (const auto& snapshot : snapshots) {
-        std::string cow_name = GetDmUserCowName(snapshot);
-        if (dm.GetState(cow_name) == DmDeviceState::INVALID) {
+        std::string user_cow_name = GetDmUserCowName(snapshot);
+        if (dm.GetState(user_cow_name) == DmDeviceState::INVALID) {
             continue;
         }
 
         DeviceMapper::TargetInfo target;
-        if (!GetSingleTarget(cow_name, TableQuery::Table, &target)) {
+        if (!GetSingleTarget(user_cow_name, TableQuery::Table, &target)) {
             continue;
         }
 
         auto target_type = DeviceMapper::GetTargetType(target.spec);
         if (target_type != "user") {
-            LOG(ERROR) << "Unexpected target type for " << cow_name << ": " << target_type;
+            LOG(ERROR) << "Unexpected target type for " << user_cow_name << ": " << target_type;
             continue;
         }
 
         num_cows++;
 
+        SnapshotStatus snapshot_status;
+        if (!ReadSnapshotStatus(lock.get(), snapshot, &snapshot_status)) {
+            LOG(ERROR) << "Unable to read snapshot status: " << snapshot;
+            continue;
+        }
+
         DmTable table;
-        table.Emplace<DmTargetUser>(0, target.spec.length, cow_name);
-        if (!dm.LoadTableAndActivate(cow_name, table)) {
-            LOG(ERROR) << "Unable to swap tables for " << cow_name;
+        table.Emplace<DmTargetUser>(0, target.spec.length, user_cow_name);
+        if (!dm.LoadTableAndActivate(user_cow_name, table)) {
+            LOG(ERROR) << "Unable to swap tables for " << user_cow_name;
             continue;
         }
 
@@ -1357,23 +1352,42 @@
             continue;
         }
 
-        std::string cow_device;
-        if (!dm.GetDmDevicePathByName(GetCowName(snapshot), &cow_device)) {
-            LOG(ERROR) << "Could not get device path for " << GetCowName(snapshot);
+        // If no partition was created (the COW exists entirely on /data), the
+        // device-mapper layering is different than if we had a partition.
+        std::string cow_image_name;
+        if (snapshot_status.cow_partition_size() == 0) {
+            cow_image_name = GetCowImageDeviceName(snapshot);
+        } else {
+            cow_image_name = GetCowName(snapshot);
+        }
+
+        std::string cow_image_device;
+        if (!dm.GetDmDevicePathByName(cow_image_name, &cow_image_device)) {
+            LOG(ERROR) << "Could not get device path for " << cow_image_name;
             continue;
         }
 
         // Wait for ueventd to acknowledge and create the control device node.
-        std::string control_device = "/dev/dm-user/" + cow_name;
+        std::string control_device = "/dev/dm-user/" + user_cow_name;
         if (!android::fs_mgr::WaitForFile(control_device, 10s)) {
             LOG(ERROR) << "Could not find control device: " << control_device;
             continue;
         }
 
-        if (!snapuserd_client_->InitializeSnapuserd(cow_device, backing_device, control_device)) {
+        uint64_t base_sectors =
+                snapuserd_client_->InitDmUserCow(user_cow_name, cow_image_device, backing_device);
+        if (base_sectors == 0) {
+            // Unrecoverable as metadata reads from cow device failed
+            LOG(FATAL) << "Failed to retrieve base_sectors from Snapuserd";
+            return false;
+        }
+
+        CHECK(base_sectors == target.spec.length);
+
+        if (!snapuserd_client_->AttachDmUser(user_cow_name)) {
             // This error is unrecoverable. We cannot proceed because reads to
             // the underlying device will fail.
-            LOG(FATAL) << "Could not initialize snapuserd for " << cow_name;
+            LOG(FATAL) << "Could not initialize snapuserd for " << user_cow_name;
             return false;
         }
 
@@ -1385,8 +1399,13 @@
         return false;
     }
 
+    std::string pid_var = device_->GetSnapuserdFirstStagePidVar();
+    if (pid_var.empty()) {
+        return true;
+    }
+
     int pid;
-    const char* pid_str = getenv(kSnapuserdFirstStagePidVar);
+    const char* pid_str = getenv(pid_var.c_str());
     if (pid_str && android::base::ParseInt(pid_str, &pid)) {
         if (kill(pid, SIGTERM) < 0 && errno != ESRCH) {
             LOG(ERROR) << "kill snapuserd failed";
@@ -2055,13 +2074,13 @@
             return false;
         }
 
-        auto control_device = "/dev/dm-user/" + dm_user_name;
-        if (!snapuserd_client_->WaitForDeviceDelete(control_device)) {
+        if (!snapuserd_client_->WaitForDeviceDelete(dm_user_name)) {
             LOG(ERROR) << "Failed to wait for " << dm_user_name << " control device to delete";
             return false;
         }
 
         // Ensure the control device is gone so we don't run into ABA problems.
+        auto control_device = "/dev/dm-user/" + dm_user_name;
         if (!android::fs_mgr::WaitForFileDeleted(control_device, 10s)) {
             LOG(ERROR) << "Timed out waiting for " << control_device << " to unlink";
             return false;
@@ -2527,8 +2546,9 @@
             .target_partition = nullptr,
             .current_metadata = current_metadata.get(),
             .current_suffix = current_suffix,
-            .operations = nullptr,
+            .update = nullptr,
             .extra_extents = {},
+            .compression_enabled = IsCompressionEnabled(),
     };
 
     auto ret = CreateUpdateSnapshotsInternal(lock.get(), manifest, &cow_creator, &created_devices,
@@ -2572,12 +2592,11 @@
         return Return::Error();
     }
 
-    std::map<std::string, const RepeatedPtrField<InstallOperation>*> install_operation_map;
+    std::map<std::string, const PartitionUpdate*> partition_map;
     std::map<std::string, std::vector<Extent>> extra_extents_map;
     for (const auto& partition_update : manifest.partitions()) {
         auto suffixed_name = partition_update.partition_name() + target_suffix;
-        auto&& [it, inserted] =
-                install_operation_map.emplace(suffixed_name, &partition_update.operations());
+        auto&& [it, inserted] = partition_map.emplace(suffixed_name, &partition_update);
         if (!inserted) {
             LOG(ERROR) << "Duplicated partition " << partition_update.partition_name()
                        << " in update manifest.";
@@ -2595,10 +2614,10 @@
 
     for (auto* target_partition : ListPartitionsWithSuffix(target_metadata, target_suffix)) {
         cow_creator->target_partition = target_partition;
-        cow_creator->operations = nullptr;
-        auto operations_it = install_operation_map.find(target_partition->name());
-        if (operations_it != install_operation_map.end()) {
-            cow_creator->operations = operations_it->second;
+        cow_creator->update = nullptr;
+        auto iter = partition_map.find(target_partition->name());
+        if (iter != partition_map.end()) {
+            cow_creator->update = iter->second;
         } else {
             LOG(INFO) << target_partition->name()
                       << " isn't included in the payload, skipping the cow creation.";
@@ -2614,6 +2633,7 @@
         // Compute the device sizes for the partition.
         auto cow_creator_ret = cow_creator->Run();
         if (!cow_creator_ret.has_value()) {
+            LOG(ERROR) << "PartitionCowCreator returned no value for " << target_partition->name();
             return Return::Error();
         }
 
diff --git a/fs_mgr/libsnapshot/snapshot_fuzz_utils.h b/fs_mgr/libsnapshot/snapshot_fuzz_utils.h
index 5319e69..092ee0b 100644
--- a/fs_mgr/libsnapshot/snapshot_fuzz_utils.h
+++ b/fs_mgr/libsnapshot/snapshot_fuzz_utils.h
@@ -124,6 +124,7 @@
         return data_->allow_set_slot_as_unbootable();
     }
     bool IsRecovery() const override { return data_->is_recovery(); }
+    std::string GetSnapuserdFirstStagePidVar() const override { return {}; }
 
     void SwitchSlot() { switched_slot_ = !switched_slot_; }
 
diff --git a/fs_mgr/libsnapshot/snapshot_reader.cpp b/fs_mgr/libsnapshot/snapshot_reader.cpp
index a4a652a..b56d879 100644
--- a/fs_mgr/libsnapshot/snapshot_reader.cpp
+++ b/fs_mgr/libsnapshot/snapshot_reader.cpp
@@ -90,6 +90,10 @@
     op_iter_ = cow_->GetOpIter();
     while (!op_iter_->Done()) {
         const CowOperation* op = &op_iter_->Get();
+        if (op->type == kCowLabelOp || op->type == kCowFooterOp) {
+            op_iter_->Next();
+            continue;
+        }
         if (op->new_block >= ops_.size()) {
             ops_.resize(op->new_block + 1, nullptr);
         }
@@ -274,7 +278,7 @@
             return -1;
         }
     } else {
-        LOG(ERROR) << "CompressedSnapshotReader unknown op type: " << op->type;
+        LOG(ERROR) << "CompressedSnapshotReader unknown op type: " << uint32_t(op->type);
         errno = EINVAL;
         return -1;
     }
diff --git a/fs_mgr/libsnapshot/snapshot_test.cpp b/fs_mgr/libsnapshot/snapshot_test.cpp
index 7fc64e5..74558ab 100644
--- a/fs_mgr/libsnapshot/snapshot_test.cpp
+++ b/fs_mgr/libsnapshot/snapshot_test.cpp
@@ -82,7 +82,6 @@
 std::string fake_super;
 
 void MountMetadata();
-bool IsCompressionEnabled();
 
 class SnapshotTest : public ::testing::Test {
   public:
@@ -796,12 +795,15 @@
         group_->add_partition_names("prd");
         sys_ = manifest_.add_partitions();
         sys_->set_partition_name("sys");
+        sys_->set_estimate_cow_size(6_MiB);
         SetSize(sys_, 3_MiB);
         vnd_ = manifest_.add_partitions();
         vnd_->set_partition_name("vnd");
+        vnd_->set_estimate_cow_size(6_MiB);
         SetSize(vnd_, 3_MiB);
         prd_ = manifest_.add_partitions();
         prd_->set_partition_name("prd");
+        prd_->set_estimate_cow_size(6_MiB);
         SetSize(prd_, 3_MiB);
 
         // Initialize source partition metadata using |manifest_|.
@@ -1045,7 +1047,11 @@
     auto tgt = MetadataBuilder::New(*opener_, "super", 1);
     ASSERT_NE(tgt, nullptr);
     ASSERT_NE(nullptr, tgt->FindPartition("sys_b-cow"));
-    ASSERT_NE(nullptr, tgt->FindPartition("vnd_b-cow"));
+    if (IsCompressionEnabled()) {
+        ASSERT_EQ(nullptr, tgt->FindPartition("vnd_b-cow"));
+    } else {
+        ASSERT_NE(nullptr, tgt->FindPartition("vnd_b-cow"));
+    }
     ASSERT_EQ(nullptr, tgt->FindPartition("prd_b-cow"));
 
     // Write some data to target partitions.
@@ -1453,10 +1459,6 @@
     MetadataMountedTest().TearDown();
 }
 
-bool IsCompressionEnabled() {
-    return android::base::GetBoolProperty("ro.virtual_ab.compression.enabled", false);
-}
-
 TEST_F(MetadataMountedTest, Android) {
     auto device = sm->EnsureMetadataMounted();
     EXPECT_NE(nullptr, device);
diff --git a/fs_mgr/libsnapshot/snapshot_writer.cpp b/fs_mgr/libsnapshot/snapshot_writer.cpp
index 85ed156..8e379e4 100644
--- a/fs_mgr/libsnapshot/snapshot_writer.cpp
+++ b/fs_mgr/libsnapshot/snapshot_writer.cpp
@@ -113,14 +113,11 @@
 }
 
 bool CompressedSnapshotWriter::Initialize() {
-    return cow_->Initialize(cow_device_, CowWriter::OpenMode::WRITE);
+    return cow_->Initialize(cow_device_);
 }
 
-bool CompressedSnapshotWriter::InitializeAppend(std::optional<uint64_t> label) {
-    if (label) {
-        return cow_->InitializeAppend(cow_device_, *label);
-    }
-    return cow_->Initialize(cow_device_, CowWriter::OpenMode::APPEND);
+bool CompressedSnapshotWriter::InitializeAppend(uint64_t label) {
+    return cow_->InitializeAppend(cow_device_, label);
 }
 
 OnlineKernelSnapshotWriter::OnlineKernelSnapshotWriter(const CowOptions& options)
diff --git a/fs_mgr/libsnapshot/snapuserd.cpp b/fs_mgr/libsnapshot/snapuserd.cpp
index 40f26d6..954699c 100644
--- a/fs_mgr/libsnapshot/snapuserd.cpp
+++ b/fs_mgr/libsnapshot/snapuserd.cpp
@@ -28,9 +28,6 @@
 using namespace android::dm;
 using android::base::unique_fd;
 
-#define DM_USER_MAP_READ 0
-#define DM_USER_MAP_WRITE 1
-
 static constexpr size_t PAYLOAD_SIZE = (1UL << 16);
 
 static_assert(PAYLOAD_SIZE >= BLOCK_SIZE);
@@ -66,11 +63,19 @@
     return header;
 }
 
+Snapuserd::Snapuserd(const std::string& misc_name, const std::string& cow_device,
+                     const std::string& backing_device) {
+    misc_name_ = misc_name;
+    cow_device_ = cow_device;
+    backing_store_device_ = backing_device;
+    control_device_ = "/dev/dm-user/" + misc_name;
+}
+
 // Construct kernel COW header in memory
 // This header will be in sector 0. The IO
 // request will always be 4k. After constructing
 // the header, zero out the remaining block.
-int Snapuserd::ConstructKernelCowHeader() {
+void Snapuserd::ConstructKernelCowHeader() {
     void* buffer = bufsink_.GetPayloadBuffer(BLOCK_SIZE);
     CHECK(buffer != nullptr);
 
@@ -82,25 +87,23 @@
     dh->valid = SNAPSHOT_VALID;
     dh->version = SNAPSHOT_DISK_VERSION;
     dh->chunk_size = CHUNK_SIZE;
-
-    return BLOCK_SIZE;
 }
 
 // Start the replace operation. This will read the
 // internal COW format and if the block is compressed,
 // it will be de-compressed.
-int Snapuserd::ProcessReplaceOp(const CowOperation* cow_op) {
+bool Snapuserd::ProcessReplaceOp(const CowOperation* cow_op) {
     if (!reader_->ReadData(*cow_op, &bufsink_)) {
         LOG(ERROR) << "ReadData failed for chunk: " << cow_op->new_block;
-        return -EIO;
+        return false;
     }
 
-    return BLOCK_SIZE;
+    return true;
 }
 
 // Start the copy operation. This will read the backing
 // block device which is represented by cow_op->source.
-int Snapuserd::ProcessCopyOp(const CowOperation* cow_op) {
+bool Snapuserd::ProcessCopyOp(const CowOperation* cow_op) {
     void* buffer = bufsink_.GetPayloadBuffer(BLOCK_SIZE);
     CHECK(buffer != nullptr);
 
@@ -109,19 +112,19 @@
     if (!android::base::ReadFullyAtOffset(backing_store_fd_, buffer, BLOCK_SIZE,
                                           cow_op->source * BLOCK_SIZE)) {
         LOG(ERROR) << "Copy-op failed. Read from backing store at: " << cow_op->source;
-        return -1;
+        return false;
     }
 
-    return BLOCK_SIZE;
+    return true;
 }
 
-int Snapuserd::ProcessZeroOp() {
+bool Snapuserd::ProcessZeroOp() {
     // Zero out the entire block
     void* buffer = bufsink_.GetPayloadBuffer(BLOCK_SIZE);
     CHECK(buffer != nullptr);
 
     memset(buffer, 0, BLOCK_SIZE);
-    return BLOCK_SIZE;
+    return true;
 }
 
 /*
@@ -146,11 +149,9 @@
  * 3: Zero operation
  *
  */
-int Snapuserd::ReadData(chunk_t chunk, size_t size) {
-    int ret = 0;
-
+bool Snapuserd::ReadData(chunk_t chunk, size_t size) {
     size_t read_size = size;
-
+    bool ret = true;
     chunk_t chunk_key = chunk;
     uint32_t stride;
     lldiv_t divresult;
@@ -159,49 +160,49 @@
     CHECK((read_size & (BLOCK_SIZE - 1)) == 0);
 
     while (read_size > 0) {
-        const CowOperation* cow_op = chunk_vec_[chunk_key];
+        const CowOperation* cow_op = chunk_map_[chunk_key];
         CHECK(cow_op != nullptr);
-        int result;
 
         switch (cow_op->type) {
             case kCowReplaceOp: {
-                result = ProcessReplaceOp(cow_op);
+                ret = ProcessReplaceOp(cow_op);
                 break;
             }
 
             case kCowZeroOp: {
-                result = ProcessZeroOp();
+                ret = ProcessZeroOp();
                 break;
             }
 
             case kCowCopyOp: {
-                result = ProcessCopyOp(cow_op);
+                ret = ProcessCopyOp(cow_op);
                 break;
             }
 
             default: {
                 LOG(ERROR) << "Unknown operation-type found: " << cow_op->type;
-                ret = -EIO;
-                goto done;
+                ret = false;
+                break;
             }
         }
 
-        if (result < 0) {
-            ret = result;
-            goto done;
+        if (!ret) {
+            LOG(ERROR) << "ReadData failed for operation: " << cow_op->type;
+            return false;
         }
 
         // Update the buffer offset
         bufsink_.UpdateBufferOffset(BLOCK_SIZE);
 
         read_size -= BLOCK_SIZE;
-        ret += BLOCK_SIZE;
 
         // Start iterating the chunk incrementally; Since while
         // constructing the metadata, we know that the chunk IDs
         // are contiguous
         chunk_key += 1;
 
+        if (cow_op->type == kCowCopyOp) CHECK(read_size == 0);
+
         // This is similar to the way when chunk IDs were assigned
         // in ReadMetadata().
         //
@@ -221,8 +222,6 @@
         }
     }
 
-done:
-
     // Reset the buffer offset
     bufsink_.ResetBufferOffset();
     return ret;
@@ -239,16 +238,18 @@
  * When dm-snap starts parsing the buffer, it will stop
  * reading metadata page once the buffer content is zero.
  */
-int Snapuserd::ZerofillDiskExceptions(size_t read_size) {
+bool Snapuserd::ZerofillDiskExceptions(size_t read_size) {
     size_t size = exceptions_per_area_ * sizeof(struct disk_exception);
 
-    if (read_size > size) return -EINVAL;
+    if (read_size > size) {
+        return false;
+    }
 
     void* buffer = bufsink_.GetPayloadBuffer(size);
     CHECK(buffer != nullptr);
 
     memset(buffer, 0, size);
-    return size;
+    return true;
 }
 
 /*
@@ -264,7 +265,7 @@
  * Convert the chunk ID to index into the vector which gives us
  * the metadata page.
  */
-int Snapuserd::ReadDiskExceptions(chunk_t chunk, size_t read_size) {
+bool Snapuserd::ReadDiskExceptions(chunk_t chunk, size_t read_size) {
     uint32_t stride = exceptions_per_area_ + 1;
     size_t size;
 
@@ -274,17 +275,184 @@
     if (divresult.quot < vec_.size()) {
         size = exceptions_per_area_ * sizeof(struct disk_exception);
 
-        if (read_size > size) return -EINVAL;
+        if (read_size > size) {
+            return false;
+        }
 
         void* buffer = bufsink_.GetPayloadBuffer(size);
         CHECK(buffer != nullptr);
 
         memcpy(buffer, vec_[divresult.quot].get(), size);
     } else {
-        size = ZerofillDiskExceptions(read_size);
+        return ZerofillDiskExceptions(read_size);
     }
 
-    return size;
+    return true;
+}
+
+loff_t Snapuserd::GetMergeStartOffset(void* merged_buffer, void* unmerged_buffer,
+                                      int* unmerged_exceptions) {
+    loff_t offset = 0;
+    *unmerged_exceptions = 0;
+
+    while (*unmerged_exceptions <= exceptions_per_area_) {
+        struct disk_exception* merged_de =
+                reinterpret_cast<struct disk_exception*>((char*)merged_buffer + offset);
+        struct disk_exception* cow_de =
+                reinterpret_cast<struct disk_exception*>((char*)unmerged_buffer + offset);
+
+        // Unmerged op by the kernel
+        if (merged_de->old_chunk != 0) {
+            CHECK(merged_de->new_chunk != 0);
+            CHECK(merged_de->old_chunk == cow_de->old_chunk);
+            CHECK(merged_de->new_chunk == cow_de->new_chunk);
+
+            offset += sizeof(struct disk_exception);
+            *unmerged_exceptions += 1;
+            continue;
+        }
+
+        // Merge complete on this exception. However, we don't know how many
+        // merged in this cycle; hence break here.
+        CHECK(merged_de->new_chunk == 0);
+        CHECK(merged_de->old_chunk == 0);
+
+        break;
+    }
+
+    CHECK(!(*unmerged_exceptions == exceptions_per_area_));
+
+    LOG(DEBUG) << "Unmerged_Exceptions: " << *unmerged_exceptions << " Offset: " << offset;
+    return offset;
+}
+
+int Snapuserd::GetNumberOfMergedOps(void* merged_buffer, void* unmerged_buffer, loff_t offset,
+                                    int unmerged_exceptions) {
+    int merged_ops_cur_iter = 0;
+
+    // Find the operations which are merged in this cycle.
+    while ((unmerged_exceptions + merged_ops_cur_iter) <= exceptions_per_area_) {
+        struct disk_exception* merged_de =
+                reinterpret_cast<struct disk_exception*>((char*)merged_buffer + offset);
+        struct disk_exception* cow_de =
+                reinterpret_cast<struct disk_exception*>((char*)unmerged_buffer + offset);
+
+        CHECK(merged_de->new_chunk == 0);
+        CHECK(merged_de->old_chunk == 0);
+
+        if (cow_de->new_chunk != 0) {
+            merged_ops_cur_iter += 1;
+            offset += sizeof(struct disk_exception);
+            // zero out to indicate that operation is merged.
+            cow_de->old_chunk = 0;
+            cow_de->new_chunk = 0;
+        } else if (cow_de->old_chunk == 0) {
+            // Already merged op in previous iteration or
+            // This could also represent a partially filled area.
+            //
+            // If the op was merged in previous cycle, we don't have
+            // to count them.
+            CHECK(cow_de->new_chunk == 0);
+            break;
+        } else {
+            LOG(ERROR) << "Error in merge operation. Found invalid metadata";
+            LOG(ERROR) << "merged_de-old-chunk: " << merged_de->old_chunk;
+            LOG(ERROR) << "merged_de-new-chunk: " << merged_de->new_chunk;
+            LOG(ERROR) << "cow_de-old-chunk: " << cow_de->old_chunk;
+            LOG(ERROR) << "cow_de-new-chunk: " << cow_de->new_chunk;
+            return -1;
+        }
+    }
+
+    return merged_ops_cur_iter;
+}
+
+bool Snapuserd::AdvanceMergedOps(int merged_ops_cur_iter) {
+    // Advance the merge operation pointer in the
+    // vector.
+    //
+    // cowop_iter_ is already initialized in ReadMetadata(). Just resume the
+    // merge process
+    while (!cowop_iter_->Done() && merged_ops_cur_iter) {
+        const CowOperation* cow_op = &cowop_iter_->Get();
+        CHECK(cow_op != nullptr);
+
+        if (cow_op->type == kCowFooterOp || cow_op->type == kCowLabelOp) {
+            cowop_iter_->Next();
+            continue;
+        }
+
+        if (!(cow_op->type == kCowReplaceOp || cow_op->type == kCowZeroOp ||
+              cow_op->type == kCowCopyOp)) {
+            LOG(ERROR) << "Unknown operation-type found during merge: " << cow_op->type;
+            return false;
+        }
+
+        merged_ops_cur_iter -= 1;
+        LOG(DEBUG) << "Merge op found of type " << cow_op->type
+                   << "Pending-merge-ops: " << merged_ops_cur_iter;
+        cowop_iter_->Next();
+    }
+
+    if (cowop_iter_->Done()) {
+        CHECK(merged_ops_cur_iter == 0);
+        LOG(DEBUG) << "All cow operations merged successfully in this cycle";
+    }
+
+    return true;
+}
+
+bool Snapuserd::ProcessMergeComplete(chunk_t chunk, void* buffer) {
+    uint32_t stride = exceptions_per_area_ + 1;
+    CowHeader header;
+
+    if (!reader_->GetHeader(&header)) {
+        LOG(ERROR) << "Failed to get header";
+        return false;
+    }
+
+    // ChunkID to vector index
+    lldiv_t divresult = lldiv(chunk, stride);
+    CHECK(divresult.quot < vec_.size());
+    LOG(DEBUG) << "ProcessMergeComplete: chunk: " << chunk << " Metadata-Index: " << divresult.quot;
+
+    int unmerged_exceptions = 0;
+    loff_t offset = GetMergeStartOffset(buffer, vec_[divresult.quot].get(), &unmerged_exceptions);
+
+    int merged_ops_cur_iter =
+            GetNumberOfMergedOps(buffer, vec_[divresult.quot].get(), offset, unmerged_exceptions);
+
+    // There should be at least one operation merged in this cycle
+    CHECK(merged_ops_cur_iter > 0);
+    if (!AdvanceMergedOps(merged_ops_cur_iter)) return false;
+
+    header.num_merge_ops += merged_ops_cur_iter;
+    reader_->UpdateMergeProgress(merged_ops_cur_iter);
+    if (!writer_->CommitMerge(merged_ops_cur_iter)) {
+        LOG(ERROR) << "CommitMerge failed...";
+        return false;
+    }
+
+    LOG(DEBUG) << "Merge success";
+    return true;
+}
+
+bool Snapuserd::IsChunkIdMetadata(chunk_t chunk) {
+    uint32_t stride = exceptions_per_area_ + 1;
+    lldiv_t divresult = lldiv(chunk, stride);
+
+    return (divresult.rem == NUM_SNAPSHOT_HDR_CHUNKS);
+}
+
+// Find the next free chunk-id to be assigned. Check if the next free
+// chunk-id represents a metadata page. If so, skip it.
+chunk_t Snapuserd::GetNextAllocatableChunkId(chunk_t chunk) {
+    chunk_t next_chunk = chunk + 1;
+
+    if (IsChunkIdMetadata(next_chunk)) {
+        next_chunk += 1;
+    }
+    return next_chunk;
 }
 
 /*
@@ -304,12 +472,26 @@
  *    This represents the old_chunk in the kernel COW format
  * 4: We need to assign new_chunk for a corresponding old_chunk
  * 5: The algorithm is similar to how kernel assigns chunk number
- *    while creating exceptions.
+ *    while creating exceptions. However, there are few cases
+ *    which needs to be addressed here:
+ *      a: During merge process, kernel scans the metadata page
+ *      from backwards when merge is initiated. Since, we need
+ *      to make sure that the merge ordering follows our COW format,
+ *      we read the COW operation from backwards and populate the
+ *      metadata so that when kernel starts the merging from backwards,
+ *      those ops correspond to the beginning of our COW format.
+ *      b: Kernel can merge successive operations if the two chunk IDs
+ *      are contiguous. This can be problematic when there is a crash
+ *      during merge; specifically when the merge operation has dependency.
+ *      These dependencies can only happen during copy operations.
+ *
+ *      To avoid this problem, we make sure that no two copy-operations
+ *      do not have contiguous chunk IDs. Additionally, we make sure
+ *      that each copy operation is merged individually.
  * 6: Use a monotonically increasing chunk number to assign the
  *    new_chunk
  * 7: Each chunk-id represents either a: Metadata page or b: Data page
- * 8: Chunk-id representing a data page is stored in a vector. Index is the
- *    chunk-id and value is the pointer to the CowOperation
+ * 8: Chunk-id representing a data page is stored in a map.
  * 9: Chunk-id representing a metadata page is converted into a vector
  *    index. We store this in vector as kernel requests metadata during
  *    two stage:
@@ -324,130 +506,128 @@
  *    exceptions_per_area_
  * 12: Kernel will stop issuing metadata IO request when new-chunk ID is 0.
  */
-int Snapuserd::ReadMetadata() {
+bool Snapuserd::ReadMetadata() {
     reader_ = std::make_unique<CowReader>();
     CowHeader header;
-    CowFooter footer;
+    CowOptions options;
+    bool prev_copy_op = false;
+    bool metadata_found = false;
+
+    LOG(DEBUG) << "ReadMetadata Start...";
 
     if (!reader_->Parse(cow_fd_)) {
         LOG(ERROR) << "Failed to parse";
-        return 1;
+        return false;
     }
 
     if (!reader_->GetHeader(&header)) {
         LOG(ERROR) << "Failed to get header";
-        return 1;
-    }
-
-    if (!reader_->GetFooter(&footer)) {
-        LOG(ERROR) << "Failed to get footer";
-        return 1;
+        return false;
     }
 
     CHECK(header.block_size == BLOCK_SIZE);
 
-    LOG(DEBUG) << "Num-ops: " << std::hex << footer.op.num_ops;
-    LOG(DEBUG) << "ops-size: " << std::hex << footer.op.ops_size;
+    LOG(DEBUG) << "Merge-ops: " << header.num_merge_ops;
 
-    cowop_iter_ = reader_->GetOpIter();
+    writer_ = std::make_unique<CowWriter>(options);
+    writer_->InitializeMerge(cow_fd_.get(), &header);
 
-    if (cowop_iter_ == nullptr) {
-        LOG(ERROR) << "Failed to get cowop_iter";
-        return 1;
-    }
+    // Initialize the iterator for reading metadata
+    cowop_riter_ = reader_->GetRevOpIter();
 
     exceptions_per_area_ = (CHUNK_SIZE << SECTOR_SHIFT) / sizeof(struct disk_exception);
 
     // Start from chunk number 2. Chunk 0 represents header and chunk 1
     // represents first metadata page.
     chunk_t next_free = NUM_SNAPSHOT_HDR_CHUNKS + 1;
-    chunk_vec_.push_back(nullptr);
-    chunk_vec_.push_back(nullptr);
 
     loff_t offset = 0;
     std::unique_ptr<uint8_t[]> de_ptr =
             std::make_unique<uint8_t[]>(exceptions_per_area_ * sizeof(struct disk_exception));
 
     // This memset is important. Kernel will stop issuing IO when new-chunk ID
-    // is 0. When Area is not filled completely will all 256 exceptions,
+    // is 0. When Area is not filled completely with all 256 exceptions,
     // this memset will ensure that metadata read is completed.
     memset(de_ptr.get(), 0, (exceptions_per_area_ * sizeof(struct disk_exception)));
     size_t num_ops = 0;
 
-    while (!cowop_iter_->Done()) {
-        const CowOperation* cow_op = &cowop_iter_->Get();
+    while (!cowop_riter_->Done()) {
+        const CowOperation* cow_op = &cowop_riter_->Get();
         struct disk_exception* de =
                 reinterpret_cast<struct disk_exception*>((char*)de_ptr.get() + offset);
 
         if (cow_op->type == kCowFooterOp || cow_op->type == kCowLabelOp) {
-            cowop_iter_->Next();
+            cowop_riter_->Next();
             continue;
         }
 
         if (!(cow_op->type == kCowReplaceOp || cow_op->type == kCowZeroOp ||
               cow_op->type == kCowCopyOp)) {
             LOG(ERROR) << "Unknown operation-type found: " << cow_op->type;
-            return 1;
+            return false;
         }
 
+        metadata_found = true;
+        if ((cow_op->type == kCowCopyOp || prev_copy_op)) {
+            next_free = GetNextAllocatableChunkId(next_free);
+        }
+
+        prev_copy_op = (cow_op->type == kCowCopyOp);
+
         // Construct the disk-exception
         de->old_chunk = cow_op->new_block;
         de->new_chunk = next_free;
 
         LOG(DEBUG) << "Old-chunk: " << de->old_chunk << "New-chunk: " << de->new_chunk;
 
-        // Store operation pointer. Note, new-chunk ID is the index
-        chunk_vec_.push_back(cow_op);
-        CHECK(next_free == (chunk_vec_.size() - 1));
+        // Store operation pointer.
+        chunk_map_[next_free] = cow_op;
+        num_ops += 1;
 
         offset += sizeof(struct disk_exception);
 
-        cowop_iter_->Next();
+        cowop_riter_->Next();
 
-        // Find the next free chunk-id to be assigned. Check if the next free
-        // chunk-id represents a metadata page. If so, skip it.
-        next_free += 1;
-        uint32_t stride = exceptions_per_area_ + 1;
-        lldiv_t divresult = lldiv(next_free, stride);
-        num_ops += 1;
-
-        if (divresult.rem == NUM_SNAPSHOT_HDR_CHUNKS) {
-            CHECK(num_ops == exceptions_per_area_);
+        if (num_ops == exceptions_per_area_) {
             // Store it in vector at the right index. This maps the chunk-id to
             // vector index.
             vec_.push_back(std::move(de_ptr));
             offset = 0;
             num_ops = 0;
 
-            chunk_t metadata_chunk = (next_free - exceptions_per_area_ - NUM_SNAPSHOT_HDR_CHUNKS);
-
-            LOG(DEBUG) << "Area: " << vec_.size() - 1;
-            LOG(DEBUG) << "Metadata-chunk: " << metadata_chunk;
-            LOG(DEBUG) << "Sector number of Metadata-chunk: " << (metadata_chunk << CHUNK_SHIFT);
-
             // Create buffer for next area
             de_ptr = std::make_unique<uint8_t[]>(exceptions_per_area_ *
                                                  sizeof(struct disk_exception));
             memset(de_ptr.get(), 0, (exceptions_per_area_ * sizeof(struct disk_exception)));
 
-            // Since this is a metadata, store at this index
-            chunk_vec_.push_back(nullptr);
-
-            // Find the next free chunk-id
-            next_free += 1;
-            if (cowop_iter_->Done()) {
+            if (cowop_riter_->Done()) {
                 vec_.push_back(std::move(de_ptr));
+                LOG(DEBUG) << "ReadMetadata() completed; Number of Areas: " << vec_.size();
             }
         }
+
+        next_free = GetNextAllocatableChunkId(next_free);
     }
 
-    // Partially filled area
-    if (num_ops) {
-        LOG(DEBUG) << "Partially filled area num_ops: " << num_ops;
+    // Partially filled area or there is no metadata
+    // If there is no metadata, fill with zero so that kernel
+    // is aware that merge is completed.
+    if (num_ops || !metadata_found) {
         vec_.push_back(std::move(de_ptr));
+        LOG(DEBUG) << "ReadMetadata() completed. Partially filled area num_ops: " << num_ops
+                   << "Areas : " << vec_.size();
     }
 
-    return 0;
+    LOG(DEBUG) << "ReadMetadata() completed. chunk_id: " << next_free
+               << "Num Sector: " << ChunkToSector(next_free);
+
+    // Initialize the iterator for merging
+    cowop_iter_ = reader_->GetOpIter();
+
+    // Total number of sectors required for creating dm-user device
+    num_sectors_ = ChunkToSector(next_free);
+    metadata_read_done_ = true;
+    return true;
 }
 
 void MyLogger(android::base::LogId, android::base::LogSeverity severity, const char*, const char*,
@@ -461,52 +641,42 @@
 
 // Read Header from dm-user misc device. This gives
 // us the sector number for which IO is issued by dm-snapshot device
-int Snapuserd::ReadDmUserHeader() {
-    int ret;
-
-    ret = read(ctrl_fd_, bufsink_.GetBufPtr(), sizeof(struct dm_user_header));
-    if (ret < 0) {
-        PLOG(ERROR) << "Control-read failed with: " << ret;
-        return ret;
+bool Snapuserd::ReadDmUserHeader() {
+    if (!android::base::ReadFully(ctrl_fd_, bufsink_.GetBufPtr(), sizeof(struct dm_user_header))) {
+        PLOG(ERROR) << "ReadDmUserHeader failed";
+        return false;
     }
 
-    return sizeof(struct dm_user_header);
+    return true;
 }
 
 // Send the payload/data back to dm-user misc device.
-int Snapuserd::WriteDmUserPayload(size_t size) {
+bool Snapuserd::WriteDmUserPayload(size_t size) {
     if (!android::base::WriteFully(ctrl_fd_, bufsink_.GetBufPtr(),
                                    sizeof(struct dm_user_header) + size)) {
         PLOG(ERROR) << "Write to dm-user failed";
-        return -1;
-    }
-
-    return sizeof(struct dm_user_header) + size;
-}
-
-bool Snapuserd::Init() {
-    backing_store_fd_.reset(open(backing_store_device_.c_str(), O_RDONLY));
-    if (backing_store_fd_ < 0) {
-        PLOG(ERROR) << "Open Failed: " << backing_store_device_;
         return false;
     }
 
+    return true;
+}
+
+bool Snapuserd::ReadDmUserPayload(void* buffer, size_t size) {
+    if (!android::base::ReadFully(ctrl_fd_, buffer, size)) {
+        PLOG(ERROR) << "ReadDmUserPayload failed";
+        return false;
+    }
+
+    return true;
+}
+
+bool Snapuserd::InitCowDevice() {
     cow_fd_.reset(open(cow_device_.c_str(), O_RDWR));
     if (cow_fd_ < 0) {
         PLOG(ERROR) << "Open Failed: " << cow_device_;
         return false;
     }
 
-    std::string control_path = GetControlDevicePath();
-
-    LOG(DEBUG) << "Opening control device " << control_path;
-
-    ctrl_fd_.reset(open(control_path.c_str(), O_RDWR));
-    if (ctrl_fd_ < 0) {
-        PLOG(ERROR) << "Unable to open " << control_path;
-        return false;
-    }
-
     // Allocate the buffer which is used to communicate between
     // daemon and dm-user. The buffer comprises of header and a fixed payload.
     // If the dm-user requests a big IO, the IO will be broken into chunks
@@ -514,20 +684,34 @@
     size_t buf_size = sizeof(struct dm_user_header) + PAYLOAD_SIZE;
     bufsink_.Initialize(buf_size);
 
+    return ReadMetadata();
+}
+
+bool Snapuserd::InitBackingAndControlDevice() {
+    backing_store_fd_.reset(open(backing_store_device_.c_str(), O_RDONLY));
+    if (backing_store_fd_ < 0) {
+        PLOG(ERROR) << "Open Failed: " << backing_store_device_;
+        return false;
+    }
+
+    ctrl_fd_.reset(open(control_device_.c_str(), O_RDWR));
+    if (ctrl_fd_ < 0) {
+        PLOG(ERROR) << "Unable to open " << control_device_;
+        return false;
+    }
+
     return true;
 }
 
-int Snapuserd::Run() {
-    int ret = 0;
-
+bool Snapuserd::Run() {
     struct dm_user_header* header = bufsink_.GetHeaderPtr();
 
     bufsink_.Clear();
 
-    ret = ReadDmUserHeader();
-    if (ret < 0) return ret;
-
-    LOG(DEBUG) << "dm-user returned " << ret << " bytes";
+    if (!ReadDmUserHeader()) {
+        LOG(ERROR) << "ReadDmUserHeader failed";
+        return false;
+    }
 
     LOG(DEBUG) << "msg->seq: " << std::hex << header->seq;
     LOG(DEBUG) << "msg->type: " << std::hex << header->type;
@@ -536,12 +720,12 @@
     LOG(DEBUG) << "msg->len: " << std::hex << header->len;
 
     switch (header->type) {
-        case DM_USER_MAP_READ: {
+        case DM_USER_REQ_MAP_READ: {
             size_t remaining_size = header->len;
             loff_t offset = 0;
-            ret = 0;
             do {
                 size_t read_size = std::min(PAYLOAD_SIZE, remaining_size);
+                header->type = DM_USER_RESP_SUCCESS;
 
                 // Request to sector 0 is always for kernel
                 // representation of COW header. This IO should be only
@@ -549,81 +733,113 @@
                 // never see multiple IO requests. Additionally this IO
                 // will always be a single 4k.
                 if (header->sector == 0) {
-                    // Read the metadata from internal COW device
-                    // and build the in-memory data structures
-                    // for all the operations in the internal COW.
-                    if (!metadata_read_done_ && ReadMetadata()) {
-                        LOG(ERROR) << "Metadata read failed";
-                        return 1;
-                    }
-                    metadata_read_done_ = true;
-
+                    CHECK(metadata_read_done_ == true);
                     CHECK(read_size == BLOCK_SIZE);
-                    ret = ConstructKernelCowHeader();
-                    if (ret < 0) return ret;
+                    ConstructKernelCowHeader();
+                    LOG(DEBUG) << "Kernel header constructed";
                 } else {
                     // Convert the sector number to a chunk ID.
                     //
                     // Check if the chunk ID represents a metadata
                     // page. If the chunk ID is not found in the
                     // vector, then it points to a metadata page.
-                    chunk_t chunk = (header->sector >> CHUNK_SHIFT);
+                    chunk_t chunk = SectorToChunk(header->sector);
 
-                    if (chunk >= chunk_vec_.size()) {
-                        ret = ZerofillDiskExceptions(read_size);
-                        if (ret < 0) {
-                            LOG(ERROR) << "ZerofillDiskExceptions failed";
-                            return ret;
-                        }
-                    } else if (chunk_vec_[chunk] == nullptr) {
-                        ret = ReadDiskExceptions(chunk, read_size);
-                        if (ret < 0) {
-                            LOG(ERROR) << "ReadDiskExceptions failed";
-                            return ret;
+                    if (chunk_map_.find(chunk) == chunk_map_.end()) {
+                        if (!ReadDiskExceptions(chunk, read_size)) {
+                            LOG(ERROR) << "ReadDiskExceptions failed for chunk id: " << chunk
+                                       << "Sector: " << header->sector;
+                            header->type = DM_USER_RESP_ERROR;
+                        } else {
+                            LOG(DEBUG) << "ReadDiskExceptions success for chunk id: " << chunk
+                                       << "Sector: " << header->sector;
                         }
                     } else {
                         chunk_t num_chunks_read = (offset >> BLOCK_SHIFT);
-                        ret = ReadData(chunk + num_chunks_read, read_size);
-                        if (ret < 0) {
-                            LOG(ERROR) << "ReadData failed";
-                            // TODO: Bug 168259959: All the error paths from this function
-                            // should send error code to dm-user thereby IO
-                            // terminates with an error from dm-user. Returning
-                            // here without sending error code will block the
-                            // IO.
-                            return ret;
+                        if (!ReadData(chunk + num_chunks_read, read_size)) {
+                            LOG(ERROR) << "ReadData failed for chunk id: " << chunk
+                                       << "Sector: " << header->sector;
+                            header->type = DM_USER_RESP_ERROR;
+                        } else {
+                            LOG(DEBUG) << "ReadData success for chunk id: " << chunk
+                                       << "Sector: " << header->sector;
                         }
                     }
                 }
 
-                ssize_t written = WriteDmUserPayload(ret);
-                if (written < 0) return written;
-
-                remaining_size -= ret;
-                offset += ret;
-                if (remaining_size) {
-                    LOG(DEBUG) << "Write done ret: " << ret
-                               << " remaining size: " << remaining_size;
+                // Daemon will not be terminated if there is any error. We will
+                // just send the error back to dm-user.
+                if (!WriteDmUserPayload(read_size)) {
+                    return false;
                 }
+
+                remaining_size -= read_size;
+                offset += read_size;
             } while (remaining_size);
 
             break;
         }
 
-        case DM_USER_MAP_WRITE: {
-            // TODO: Bug: 168311203: After merge operation is completed, kernel issues write
-            // to flush all the exception mappings where the merge is
-            // completed. If dm-user routes the WRITE IO, we need to clear
-            // in-memory data structures representing those exception
-            // mappings.
-            abort();
+        case DM_USER_REQ_MAP_WRITE: {
+            // device mapper has the capability to allow
+            // targets to flush the cache when writes are completed. This
+            // is controlled by each target by a flag "flush_supported".
+            // This flag is set by dm-user. When flush is supported,
+            // a number of zero-length bio's will be submitted to
+            // the target for the purpose of flushing cache. It is the
+            // responsibility of the target driver - which is dm-user in this
+            // case, to remap these bio's to the underlying device. Since,
+            // there is no underlying device for dm-user, this zero length
+            // bio's gets routed to daemon.
+            //
+            // Flush operations are generated post merge by dm-snap by having
+            // REQ_PREFLUSH flag set. Snapuser daemon doesn't have anything
+            // to flush per se; hence, just respond back with a success message.
+            if (header->sector == 0) {
+                CHECK(header->len == 0);
+                header->type = DM_USER_RESP_SUCCESS;
+                if (!WriteDmUserPayload(0)) {
+                    return false;
+                }
+                break;
+            }
+
+            size_t remaining_size = header->len;
+            size_t read_size = std::min(PAYLOAD_SIZE, remaining_size);
+            CHECK(read_size == BLOCK_SIZE);
+
+            CHECK(header->sector > 0);
+            chunk_t chunk = SectorToChunk(header->sector);
+            CHECK(chunk_map_.find(chunk) == chunk_map_.end());
+
+            void* buffer = bufsink_.GetPayloadBuffer(read_size);
+            CHECK(buffer != nullptr);
+            header->type = DM_USER_RESP_SUCCESS;
+
+            if (!ReadDmUserPayload(buffer, read_size)) {
+                LOG(ERROR) << "ReadDmUserPayload failed for chunk id: " << chunk
+                           << "Sector: " << header->sector;
+                header->type = DM_USER_RESP_ERROR;
+            }
+
+            if (header->type == DM_USER_RESP_SUCCESS && !ProcessMergeComplete(chunk, buffer)) {
+                LOG(ERROR) << "ProcessMergeComplete failed for chunk id: " << chunk
+                           << "Sector: " << header->sector;
+                header->type = DM_USER_RESP_ERROR;
+            } else {
+                LOG(DEBUG) << "ProcessMergeComplete success for chunk id: " << chunk
+                           << "Sector: " << header->sector;
+            }
+
+            if (!WriteDmUserPayload(0)) {
+                return false;
+            }
+
             break;
         }
     }
 
-    LOG(DEBUG) << "read() finished, next message";
-
-    return 0;
+    return true;
 }
 
 }  // namespace snapshot
diff --git a/fs_mgr/libsnapshot/snapuserd_client.cpp b/fs_mgr/libsnapshot/snapuserd_client.cpp
index 5650139..a5d2061 100644
--- a/fs_mgr/libsnapshot/snapuserd_client.cpp
+++ b/fs_mgr/libsnapshot/snapuserd_client.cpp
@@ -27,9 +27,12 @@
 #include <unistd.h>
 
 #include <chrono>
+#include <sstream>
 
 #include <android-base/logging.h>
+#include <android-base/parseint.h>
 #include <android-base/properties.h>
+#include <android-base/strings.h>
 #include <libsnapshot/snapuserd_client.h>
 
 namespace android {
@@ -114,7 +117,7 @@
     std::string str = Receivemsg();
 
     // If the daemon is passive then fallback to secondary active daemon. Daemon
-    // is passive during transition phase. Please see RestartSnapuserd()
+    // is passive during transition phase.
     if (str.find("passive") != std::string::npos) {
         LOG(ERROR) << "Snapuserd is terminating";
         return false;
@@ -180,10 +183,8 @@
     return true;
 }
 
-bool SnapuserdClient::InitializeSnapuserd(const std::string& cow_device,
-                                          const std::string& backing_device,
-                                          const std::string& control_device) {
-    std::string msg = "start," + cow_device + "," + backing_device + "," + control_device;
+bool SnapuserdClient::AttachDmUser(const std::string& misc_name) {
+    std::string msg = "start," + misc_name;
     if (!Sendmsg(msg)) {
         LOG(ERROR) << "Failed to send message " << msg << " to snapuserd daemon";
         return false;
@@ -199,77 +200,33 @@
     return true;
 }
 
-/*
- * Transition from first stage snapuserd daemon to second stage daemon involves
- * series of steps viz:
- *
- * 1: Create new dm-user devices - This is done by libsnapshot
- *
- * 2: Spawn the new snapuserd daemon - This is the second stage daemon which
- * will start the server but the dm-user misc devices is not binded yet.
- *
- * 3: Vector to this function contains pair of cow_device and source device.
- *    Ex: {{system_cow,system_a}, {product_cow, product_a}, {vendor_cow,
- *    vendor_a}}. This vector will be populated by the libsnapshot.
- *
- * 4: Initialize the Second stage daemon passing the information from the
- * vector. This will bind the daemon with dm-user misc device and will be ready
- * to serve the IO. Up until this point, first stage daemon is still active.
- * However, client library will mark the first stage daemon as passive and hence
- * all the control message from hereon will be sent to active second stage
- * daemon.
- *
- * 5: Create new dm-snapshot table. This is done by libsnapshot. When new table
- * is created, kernel will issue metadata read once again which will be served
- * by second stage daemon. However, any active IO will still be served by first
- * stage daemon.
- *
- * 6: Swap the snapshot table atomically - This is done by libsnapshot. Once
- * the swapping is done, all the IO will be served by second stage daemon.
- *
- * 7: Stop the first stage daemon. After this point second stage daemon is
- * completely active to serve the IO and merging process.
- *
- */
-int SnapuserdClient::RestartSnapuserd(std::vector<std::vector<std::string>>& vec) {
-    std::string msg = "terminate-request";
+uint64_t SnapuserdClient::InitDmUserCow(const std::string& misc_name, const std::string& cow_device,
+                                        const std::string& backing_device) {
+    std::vector<std::string> parts = {"init", misc_name, cow_device, backing_device};
+    std::string msg = android::base::Join(parts, ",");
     if (!Sendmsg(msg)) {
         LOG(ERROR) << "Failed to send message " << msg << " to snapuserd daemon";
-        return -1;
+        return 0;
     }
 
     std::string str = Receivemsg();
 
-    if (str.find("fail") != std::string::npos) {
-        LOG(ERROR) << "Failed to receive ack for " << msg << " from snapuserd daemon";
-        return -1;
+    std::vector<std::string> input = android::base::Split(str, ",");
+
+    if (input.empty() || input[0] != "success") {
+        LOG(ERROR) << "Failed to receive number of sectors for " << msg << " from snapuserd daemon";
+        return 0;
     }
 
-    CHECK(str.find("success") != std::string::npos);
+    LOG(DEBUG) << "Snapuserd daemon COW device initialized: " << cow_device
+               << " Num-sectors: " << input[1];
 
-    // Start the new daemon
-    if (!EnsureSnapuserdStarted()) {
-        LOG(ERROR) << "Failed to start new daemon";
-        return -1;
+    uint64_t num_sectors = 0;
+    if (!android::base::ParseUint(input[1], &num_sectors)) {
+        LOG(ERROR) << "Failed to parse input string to sectors";
+        return 0;
     }
-
-    LOG(DEBUG) << "Second stage Snapuserd daemon created successfully";
-
-    // Vector contains all the device information to be passed to the new
-    // daemon. Note that the caller can choose to initialize separately
-    // by calling InitializeSnapuserd() API as well. In that case, vector
-    // should be empty
-    for (int i = 0; i < vec.size(); i++) {
-        std::string& cow_device = vec[i][0];
-        std::string& base_device = vec[i][1];
-        std::string& control_device = vec[i][2];
-
-        InitializeSnapuserd(cow_device, base_device, control_device);
-        LOG(DEBUG) << "Daemon initialized with " << cow_device << ", " << base_device << " and "
-                   << control_device;
-    }
-
-    return 0;
+    return num_sectors;
 }
 
 }  // namespace snapshot
diff --git a/fs_mgr/libsnapshot/snapuserd_server.cpp b/fs_mgr/libsnapshot/snapuserd_server.cpp
index 6a89218..9d57ab0 100644
--- a/fs_mgr/libsnapshot/snapuserd_server.cpp
+++ b/fs_mgr/libsnapshot/snapuserd_server.cpp
@@ -33,6 +33,7 @@
 namespace snapshot {
 
 DaemonOperations SnapuserdServer::Resolveop(std::string& input) {
+    if (input == "init") return DaemonOperations::INIT;
     if (input == "start") return DaemonOperations::START;
     if (input == "stop") return DaemonOperations::STOP;
     if (input == "query") return DaemonOperations::QUERY;
@@ -73,7 +74,7 @@
     StopThreads();
 
     // Acquire the thread list within the lock.
-    std::vector<std::unique_ptr<DmUserHandler>> dm_users;
+    std::vector<std::shared_ptr<DmUserHandler>> dm_users;
     {
         std::lock_guard<std::mutex> guard(lock_);
         dm_users = std::move(dm_users_);
@@ -86,8 +87,8 @@
     }
 }
 
-const std::string& DmUserHandler::GetControlDevice() const {
-    return snapuserd_->GetControlDevicePath();
+const std::string& DmUserHandler::GetMiscName() const {
+    return snapuserd_->GetMiscName();
 }
 
 bool SnapuserdServer::Sendmsg(android::base::borrowed_fd fd, const std::string& msg) {
@@ -123,30 +124,61 @@
     DaemonOperations op = Resolveop(out[0]);
 
     switch (op) {
-        case DaemonOperations::START: {
+        case DaemonOperations::INIT: {
             // Message format:
-            // start,<cow_device_path>,<source_device_path>,<control_device>
+            // init,<misc_name>,<cow_device_path>,<control_device>
             //
-            // Start the new thread which binds to dm-user misc device
+            // Reads the metadata and send the number of sectors
             if (out.size() != 4) {
-                LOG(ERROR) << "Malformed start message, " << out.size() << " parts";
+                LOG(ERROR) << "Malformed init message, " << out.size() << " parts";
                 return Sendmsg(fd, "fail");
             }
 
             auto snapuserd = std::make_unique<Snapuserd>(out[1], out[2], out[3]);
-            if (!snapuserd->Init()) {
+            if (!snapuserd->InitCowDevice()) {
                 LOG(ERROR) << "Failed to initialize Snapuserd";
                 return Sendmsg(fd, "fail");
             }
 
+            std::string retval = "success," + std::to_string(snapuserd->GetNumSectors());
+
             auto handler = std::make_unique<DmUserHandler>(std::move(snapuserd));
             {
                 std::lock_guard<std::mutex> lock(lock_);
-
-                handler->thread() =
-                        std::thread(std::bind(&SnapuserdServer::RunThread, this, handler.get()));
+                if (FindHandler(&lock, out[1]) != dm_users_.end()) {
+                    LOG(ERROR) << "Handler already exists: " << out[1];
+                    return Sendmsg(fd, "fail");
+                }
                 dm_users_.push_back(std::move(handler));
             }
+
+            return Sendmsg(fd, retval);
+        }
+        case DaemonOperations::START: {
+            // Message format:
+            // start,<misc_name>
+            //
+            // Start the new thread which binds to dm-user misc device
+            if (out.size() != 2) {
+                LOG(ERROR) << "Malformed start message, " << out.size() << " parts";
+                return Sendmsg(fd, "fail");
+            }
+
+            std::lock_guard<std::mutex> lock(lock_);
+            auto iter = FindHandler(&lock, out[1]);
+            if (iter == dm_users_.end()) {
+                LOG(ERROR) << "Could not find handler: " << out[1];
+                return Sendmsg(fd, "fail");
+            }
+            if ((*iter)->snapuserd()->IsAttached()) {
+                LOG(ERROR) << "Tried to re-attach control device: " << out[1];
+                return Sendmsg(fd, "fail");
+            }
+            if (!((*iter)->snapuserd()->InitBackingAndControlDevice())) {
+                LOG(ERROR) << "Failed to initialize control device: " << out[1];
+                return Sendmsg(fd, "fail");
+            }
+            (*iter)->thread() = std::thread(std::bind(&SnapuserdServer::RunThread, this, *iter));
             return Sendmsg(fd, "success");
         }
         case DaemonOperations::STOP: {
@@ -172,12 +204,12 @@
         }
         case DaemonOperations::DELETE: {
             // Message format:
-            // delete,<cow_device_path>
+            // delete,<misc_name>
             if (out.size() != 2) {
                 LOG(ERROR) << "Malformed delete message, " << out.size() << " parts";
                 return Sendmsg(fd, "fail");
             }
-            if (!WaitForDelete(out[1])) {
+            if (!RemoveHandler(out[1], true)) {
                 return Sendmsg(fd, "fail");
             }
             return Sendmsg(fd, "success");
@@ -190,25 +222,21 @@
     }
 }
 
-void SnapuserdServer::RunThread(DmUserHandler* handler) {
-    LOG(INFO) << "Entering thread for handler: " << handler->GetControlDevice();
+void SnapuserdServer::RunThread(std::shared_ptr<DmUserHandler> handler) {
+    LOG(INFO) << "Entering thread for handler: " << handler->GetMiscName();
 
     while (!StopRequested()) {
-        if (handler->snapuserd()->Run() < 0) {
-            LOG(INFO) << "Snapuserd: Thread terminating as control device is de-registered";
+        if (!handler->snapuserd()->Run()) {
+            LOG(INFO) << "Snapuserd: Thread terminating";
             break;
         }
     }
 
-    LOG(INFO) << "Exiting thread for handler: " << handler->GetControlDevice();
+    LOG(INFO) << "Exiting thread for handler: " << handler->GetMiscName();
 
-    if (auto client = RemoveHandler(handler->GetControlDevice())) {
-        // The main thread did not receive a WaitForDelete request for this
-        // control device. Since we transferred ownership within the lock,
-        // we know join() was never called, and will never be called. We can
-        // safely detach here.
-        client->thread().detach();
-    }
+    // If the main thread called /emoveHandler, the handler was already removed
+    // from within the lock, and calling RemoveHandler again has no effect.
+    RemoveHandler(handler->GetMiscName(), false);
 }
 
 bool SnapuserdServer::Start(const std::string& socketname) {
@@ -301,34 +329,37 @@
     SetTerminating();
 }
 
-std::unique_ptr<DmUserHandler> SnapuserdServer::RemoveHandler(const std::string& control_device) {
-    std::unique_ptr<DmUserHandler> client;
-    {
-        std::lock_guard<std::mutex> lock(lock_);
-        auto iter = dm_users_.begin();
-        while (iter != dm_users_.end()) {
-            if ((*iter)->GetControlDevice() == control_device) {
-                client = std::move(*iter);
-                iter = dm_users_.erase(iter);
-                break;
-            }
-            iter++;
+auto SnapuserdServer::FindHandler(std::lock_guard<std::mutex>* proof_of_lock,
+                                  const std::string& misc_name) -> HandlerList::iterator {
+    CHECK(proof_of_lock);
+
+    for (auto iter = dm_users_.begin(); iter != dm_users_.end(); iter++) {
+        if ((*iter)->GetMiscName() == misc_name) {
+            return iter;
         }
     }
-    return client;
+    return dm_users_.end();
 }
 
-bool SnapuserdServer::WaitForDelete(const std::string& control_device) {
-    auto client = RemoveHandler(control_device);
+bool SnapuserdServer::RemoveHandler(const std::string& misc_name, bool wait) {
+    std::shared_ptr<DmUserHandler> handler;
+    {
+        std::lock_guard<std::mutex> lock(lock_);
 
-    // Client already deleted.
-    if (!client) {
-        return true;
+        auto iter = FindHandler(&lock, misc_name);
+        if (iter == dm_users_.end()) {
+            // Client already deleted.
+            return true;
+        }
+        handler = std::move(*iter);
+        dm_users_.erase(iter);
     }
 
-    auto& th = client->thread();
-    if (th.joinable()) {
+    auto& th = handler->thread();
+    if (th.joinable() && wait) {
         th.join();
+    } else if (handler->snapuserd()->IsAttached()) {
+        th.detach();
     }
     return true;
 }
diff --git a/fs_mgr/libsnapshot/update_engine/update_metadata.proto b/fs_mgr/libsnapshot/update_engine/update_metadata.proto
index 202e39b..dda214e 100644
--- a/fs_mgr/libsnapshot/update_engine/update_metadata.proto
+++ b/fs_mgr/libsnapshot/update_engine/update_metadata.proto
@@ -62,6 +62,7 @@
     repeated InstallOperation operations = 8;
     optional Extent hash_tree_extent = 11;
     optional Extent fec_extent = 15;
+    optional uint64 estimate_cow_size = 19;
 }
 
 message DynamicPartitionGroup {
diff --git a/fs_mgr/libsnapshot/utility.cpp b/fs_mgr/libsnapshot/utility.cpp
index 4cae83a..7342fd4 100644
--- a/fs_mgr/libsnapshot/utility.cpp
+++ b/fs_mgr/libsnapshot/utility.cpp
@@ -22,6 +22,7 @@
 
 #include <android-base/file.h>
 #include <android-base/logging.h>
+#include <android-base/properties.h>
 #include <android-base/strings.h>
 #include <fs_mgr/roots.h>
 
@@ -182,5 +183,9 @@
     new_extent->set_num_blocks(num_blocks);
 }
 
+bool IsCompressionEnabled() {
+    return android::base::GetBoolProperty("ro.virtual_ab.compression.enabled", false);
+}
+
 }  // namespace snapshot
 }  // namespace android
diff --git a/fs_mgr/libsnapshot/utility.h b/fs_mgr/libsnapshot/utility.h
index 482888a..3e6873b 100644
--- a/fs_mgr/libsnapshot/utility.h
+++ b/fs_mgr/libsnapshot/utility.h
@@ -129,5 +129,7 @@
 void AppendExtent(google::protobuf::RepeatedPtrField<chromeos_update_engine::Extent>* extents,
                   uint64_t start_block, uint64_t num_blocks);
 
+bool IsCompressionEnabled();
+
 }  // namespace snapshot
 }  // namespace android
diff --git a/fs_mgr/tests/adb-remount-test.sh b/fs_mgr/tests/adb-remount-test.sh
index e995888..f5bbe35 100755
--- a/fs_mgr/tests/adb-remount-test.sh
+++ b/fs_mgr/tests/adb-remount-test.sh
@@ -1380,9 +1380,9 @@
 check_eq "${VENDOR_DEVT}" "`adb_sh stat --format=%D /vendor/hello </dev/null`" vendor devt after reboot
 check_eq "${SYSTEM_INO}" "`adb_sh stat --format=%i /system/hello </dev/null`" system inode after reboot
 check_eq "${VENDOR_INO}" "`adb_sh stat --format=%i /vendor/hello </dev/null`" vendor inode after reboot
-check_eq "${BASE_SYSTEM_DEVT}" "`adb_sh stat --format=%D /system/bin/stat </dev/null`" base system devt after reboot
-check_eq "${BASE_VENDOR_DEVT}" "`adb_sh stat --format=%D /vendor/bin/stat </dev/null`" base system devt after reboot
-check_eq "${BASE_SYSTEM_DEVT}" "`adb_sh stat --format=%D /system/xbin/su </dev/null`" devt for su after reboot
+check_eq "${BASE_SYSTEM_DEVT}" "`adb_sh stat --format=%D /system/bin/stat </dev/null`" --warning base system devt after reboot
+check_eq "${BASE_VENDOR_DEVT}" "`adb_sh stat --format=%D /vendor/bin/stat </dev/null`" --warning base vendor devt after reboot
+check_eq "${BASE_SYSTEM_DEVT}" "`adb_sh stat --format=%D /system/xbin/su </dev/null`" --warning devt for su after reboot
 
 # Feed log with selinux denials as a result of overlays
 adb_sh find ${MOUNTS} </dev/null >/dev/null 2>/dev/null
@@ -1509,8 +1509,8 @@
 
   check_eq "${SYSTEM_DEVT}" "`adb_sh stat --format=%D /system/hello </dev/null`" system devt after reboot
   check_eq "${SYSTEM_INO}" "`adb_sh stat --format=%i /system/hello </dev/null`" system inode after reboot
-  check_eq "${BASE_SYSTEM_DEVT}" "`adb_sh stat --format=%D /system/bin/stat </dev/null`" base system devt after reboot
-  check_eq "${BASE_SYSTEM_DEVT}" "`adb_sh stat --format=%D /system/xbin/su </dev/null`" devt for su after reboot
+  check_eq "${BASE_SYSTEM_DEVT}" "`adb_sh stat --format=%D /system/bin/stat </dev/null`" --warning base system devt after reboot
+  check_eq "${BASE_SYSTEM_DEVT}" "`adb_sh stat --format=%D /system/xbin/su </dev/null`" --warning devt for su after reboot
 
 fi
 
diff --git a/fs_mgr/tools/Android.bp b/fs_mgr/tools/Android.bp
index 4d4aae4..d6ccc4b 100644
--- a/fs_mgr/tools/Android.bp
+++ b/fs_mgr/tools/Android.bp
@@ -29,3 +29,15 @@
 
     cflags: ["-Werror"],
 }
+
+cc_binary {
+    name: "dmuserd",
+    srcs: ["dmuserd.cpp"],
+
+    shared_libs: [
+        "libbase",
+        "liblog",
+    ],
+
+    cflags: ["-Werror"],
+}
diff --git a/fs_mgr/tools/dmuserd.cpp b/fs_mgr/tools/dmuserd.cpp
new file mode 100644
index 0000000..92f5878
--- /dev/null
+++ b/fs_mgr/tools/dmuserd.cpp
@@ -0,0 +1,319 @@
+// SPDX-License-Identifier: Apache-2.0
+
+#define _LARGEFILE64_SOURCE
+
+#include <errno.h>
+#include <fcntl.h>
+#include <getopt.h>
+#include <poll.h>
+#include <stdio.h>
+#include <stdlib.h>
+#include <string.h>
+#include <sys/mman.h>
+#include <sys/prctl.h>
+#include <unistd.h>
+#include <iostream>
+
+#define SECTOR_SIZE ((__u64)512)
+#define BUFFER_BYTES 4096
+
+#define MAX(a, b) ((a) > (b) ? (a) : (b))
+
+/* This should be replaced with linux/dm-user.h. */
+#ifndef _LINUX_DM_USER_H
+#define _LINUX_DM_USER_H
+
+#include <linux/types.h>
+
+#define DM_USER_REQ_MAP_READ 0
+#define DM_USER_REQ_MAP_WRITE 1
+#define DM_USER_REQ_MAP_FLUSH 2
+#define DM_USER_REQ_MAP_DISCARD 3
+#define DM_USER_REQ_MAP_SECURE_ERASE 4
+#define DM_USER_REQ_MAP_WRITE_SAME 5
+#define DM_USER_REQ_MAP_WRITE_ZEROES 6
+#define DM_USER_REQ_MAP_ZONE_OPEN 7
+#define DM_USER_REQ_MAP_ZONE_CLOSE 8
+#define DM_USER_REQ_MAP_ZONE_FINISH 9
+#define DM_USER_REQ_MAP_ZONE_APPEND 10
+#define DM_USER_REQ_MAP_ZONE_RESET 11
+#define DM_USER_REQ_MAP_ZONE_RESET_ALL 12
+
+#define DM_USER_REQ_MAP_FLAG_FAILFAST_DEV 0x00001
+#define DM_USER_REQ_MAP_FLAG_FAILFAST_TRANSPORT 0x00002
+#define DM_USER_REQ_MAP_FLAG_FAILFAST_DRIVER 0x00004
+#define DM_USER_REQ_MAP_FLAG_SYNC 0x00008
+#define DM_USER_REQ_MAP_FLAG_META 0x00010
+#define DM_USER_REQ_MAP_FLAG_PRIO 0x00020
+#define DM_USER_REQ_MAP_FLAG_NOMERGE 0x00040
+#define DM_USER_REQ_MAP_FLAG_IDLE 0x00080
+#define DM_USER_REQ_MAP_FLAG_INTEGRITY 0x00100
+#define DM_USER_REQ_MAP_FLAG_FUA 0x00200
+#define DM_USER_REQ_MAP_FLAG_PREFLUSH 0x00400
+#define DM_USER_REQ_MAP_FLAG_RAHEAD 0x00800
+#define DM_USER_REQ_MAP_FLAG_BACKGROUND 0x01000
+#define DM_USER_REQ_MAP_FLAG_NOWAIT 0x02000
+#define DM_USER_REQ_MAP_FLAG_CGROUP_PUNT 0x04000
+#define DM_USER_REQ_MAP_FLAG_NOUNMAP 0x08000
+#define DM_USER_REQ_MAP_FLAG_HIPRI 0x10000
+#define DM_USER_REQ_MAP_FLAG_DRV 0x20000
+#define DM_USER_REQ_MAP_FLAG_SWAP 0x40000
+
+#define DM_USER_RESP_SUCCESS 0
+#define DM_USER_RESP_ERROR 1
+#define DM_USER_RESP_UNSUPPORTED 2
+
+struct dm_user_message {
+    __u64 seq;
+    __u64 type;
+    __u64 flags;
+    __u64 sector;
+    __u64 len;
+    __u8 buf[];
+};
+
+#endif
+
+static bool verbose = false;
+
+size_t write_all(int fd, void* buf, size_t len) {
+    char* buf_c = (char*)buf;
+    ssize_t total = 0;
+    ssize_t once;
+
+    while (total < len) {
+        once = write(fd, buf_c + total, len - total);
+        if (once < 0) return once;
+        if (once == 0) {
+            errno = ENOSPC;
+            return 0;
+        }
+        total += once;
+    }
+
+    return total;
+}
+
+size_t read_all(int fd, void* buf, size_t len) {
+    char* buf_c = (char*)buf;
+    ssize_t total = 0;
+    ssize_t once;
+
+    while (total < len) {
+        once = read(fd, buf_c + total, len - total);
+        if (once < 0) return once;
+        if (once == 0) {
+            errno = ENOSPC;
+            return 0;
+        }
+        total += once;
+    }
+
+    return total;
+}
+
+int not_splice(int from, int to, __u64 count) {
+    while (count > 0) {
+        char buf[BUFFER_BYTES];
+        __u64 max = count > BUFFER_BYTES ? BUFFER_BYTES : count;
+
+        if (read_all(from, buf, max) <= 0) {
+            perror("Unable to read");
+            return -EIO;
+        }
+
+        if (write_all(to, buf, max) <= 0) {
+            perror("Unable to write");
+            return -EIO;
+        }
+
+        count -= max;
+    }
+
+    return 0;
+}
+
+int simple_daemon(char* control_path, char* backing_path) {
+    int control_fd = open(control_path, O_RDWR);
+    if (control_fd < 0) {
+        fprintf(stderr, "Unable to open control device %s\n", control_path);
+        return -1;
+    }
+
+    int backing_fd = open(backing_path, O_RDWR);
+    if (backing_fd < 0) {
+        fprintf(stderr, "Unable to open backing device %s\n", backing_path);
+        return -1;
+    }
+
+    while (1) {
+        struct dm_user_message msg;
+        char* base;
+        __u64 type;
+
+        if (verbose) std::cerr << "dmuserd: Waiting for message...\n";
+
+        if (read_all(control_fd, &msg, sizeof(msg)) < 0) {
+            if (errno == ENOTBLK) return 0;
+
+            perror("unable to read msg");
+            return -1;
+        }
+
+        if (verbose) {
+            std::string type;
+            switch (msg.type) {
+                case DM_USER_REQ_MAP_WRITE:
+                    type = "write";
+                    break;
+                case DM_USER_REQ_MAP_READ:
+                    type = "read";
+                    break;
+                case DM_USER_REQ_MAP_FLUSH:
+                    type = "flush";
+                    break;
+                default:
+                    /*
+                     * FIXME: Can't I do "whatever"s here rather that
+                     * std::string("whatever")?
+                     */
+                    type = std::string("(unknown, id=") + std::to_string(msg.type) + ")";
+                    break;
+            }
+
+            std::string flags;
+            if (msg.flags & DM_USER_REQ_MAP_FLAG_SYNC) {
+                if (!flags.empty()) flags += "|";
+                flags += "S";
+            }
+            if (msg.flags & DM_USER_REQ_MAP_FLAG_META) {
+                if (!flags.empty()) flags += "|";
+                flags += "M";
+            }
+            if (msg.flags & DM_USER_REQ_MAP_FLAG_FUA) {
+                if (!flags.empty()) flags += "|";
+                flags += "FUA";
+            }
+            if (msg.flags & DM_USER_REQ_MAP_FLAG_PREFLUSH) {
+                if (!flags.empty()) flags += "|";
+                flags += "F";
+            }
+
+            std::cerr << "dmuserd: Got " << type << " request " << flags << " for sector "
+                      << std::to_string(msg.sector) << " with length " << std::to_string(msg.len)
+                      << "\n";
+        }
+
+        type = msg.type;
+        switch (type) {
+            case DM_USER_REQ_MAP_READ:
+                msg.type = DM_USER_RESP_SUCCESS;
+                break;
+            case DM_USER_REQ_MAP_WRITE:
+                if (msg.flags & DM_USER_REQ_MAP_FLAG_PREFLUSH ||
+                    msg.flags & DM_USER_REQ_MAP_FLAG_FUA) {
+                    if (fsync(backing_fd) < 0) {
+                        perror("Unable to fsync(), just sync()ing instead");
+                        sync();
+                    }
+                }
+                msg.type = DM_USER_RESP_SUCCESS;
+                if (lseek64(backing_fd, msg.sector * SECTOR_SIZE, SEEK_SET) < 0) {
+                    perror("Unable to seek");
+                    return -1;
+                }
+                if (not_splice(control_fd, backing_fd, msg.len) < 0) {
+                    if (errno == ENOTBLK) return 0;
+                    std::cerr << "unable to handle write data\n";
+                    return -1;
+                }
+                if (msg.flags & DM_USER_REQ_MAP_FLAG_FUA) {
+                    if (fsync(backing_fd) < 0) {
+                        perror("Unable to fsync(), just sync()ing instead");
+                        sync();
+                    }
+                }
+                break;
+            case DM_USER_REQ_MAP_FLUSH:
+                msg.type = DM_USER_RESP_SUCCESS;
+                if (fsync(backing_fd) < 0) {
+                    perror("Unable to fsync(), just sync()ing instead");
+                    sync();
+                }
+                break;
+            default:
+                std::cerr << "dmuserd: unsupported op " << std::to_string(msg.type) << "\n";
+                msg.type = DM_USER_RESP_UNSUPPORTED;
+                break;
+        }
+
+        if (verbose) std::cerr << "dmuserd: Responding to message\n";
+
+        if (write_all(control_fd, &msg, sizeof(msg)) < 0) {
+            if (errno == ENOTBLK) return 0;
+            perror("unable to write msg");
+            return -1;
+        }
+
+        switch (type) {
+            case DM_USER_REQ_MAP_READ:
+                if (verbose) std::cerr << "dmuserd: Sending read data\n";
+                if (lseek64(backing_fd, msg.sector * SECTOR_SIZE, SEEK_SET) < 0) {
+                    perror("Unable to seek");
+                    return -1;
+                }
+                if (not_splice(backing_fd, control_fd, msg.len) < 0) {
+                    if (errno == ENOTBLK) return 0;
+                    std::cerr << "unable to handle read data\n";
+                    return -1;
+                }
+                break;
+        }
+    }
+
+    /* The daemon doesn't actully terminate for this test. */
+    perror("Unable to read from control device");
+    return -1;
+}
+
+void usage(char* prog) {
+    printf("Usage: %s\n", prog);
+    printf("	Handles block requests in userspace, backed by memory\n");
+    printf("  -h			Display this help message\n");
+    printf("  -c <control dev>		Control device to use for the test\n");
+    printf("  -b <store path>		The file to use as a backing store, otherwise memory\n");
+    printf("  -v                        Enable verbose mode\n");
+}
+
+int main(int argc, char* argv[]) {
+    char* control_path = NULL;
+    char* backing_path = NULL;
+    char* store;
+    int c;
+
+    prctl(PR_SET_IO_FLUSHER, 0, 0, 0, 0);
+
+    while ((c = getopt(argc, argv, "h:c:s:b:v")) != -1) {
+        switch (c) {
+            case 'h':
+                usage(basename(argv[0]));
+                exit(0);
+            case 'c':
+                control_path = strdup(optarg);
+                break;
+            case 'b':
+                backing_path = strdup(optarg);
+                break;
+            case 'v':
+                verbose = true;
+                break;
+            default:
+                usage(basename(argv[0]));
+                exit(1);
+        }
+    }
+
+    int r = simple_daemon(control_path, backing_path);
+    if (r) fprintf(stderr, "simple_daemon() errored out\n");
+    return r;
+}
diff --git a/gatekeeperd/binder/android/service/gatekeeper/IGateKeeperService.aidl b/gatekeeperd/binder/android/service/gatekeeper/IGateKeeperService.aidl
index 4646efc..67e8633 100644
--- a/gatekeeperd/binder/android/service/gatekeeper/IGateKeeperService.aidl
+++ b/gatekeeperd/binder/android/service/gatekeeper/IGateKeeperService.aidl
@@ -27,6 +27,7 @@
  *
  * @hide
  */
+@SensitiveData
 interface IGateKeeperService {
     /**
      * Enrolls a password, returning the handle to the enrollment to be stored locally.
diff --git a/init/Android.mk b/init/Android.mk
index ac31ef1..4c1665b 100644
--- a/init/Android.mk
+++ b/init/Android.mk
@@ -82,6 +82,7 @@
     $(TARGET_RAMDISK_OUT)/proc \
     $(TARGET_RAMDISK_OUT)/second_stage_resources \
     $(TARGET_RAMDISK_OUT)/sys \
+    $(TARGET_RAMDISK_OUT)/metadata \
 
 LOCAL_STATIC_LIBRARIES := \
     libc++fs \
diff --git a/init/devices.cpp b/init/devices.cpp
index f8eb16a..5888c06 100644
--- a/init/devices.cpp
+++ b/init/devices.cpp
@@ -440,13 +440,6 @@
             }
         }
         unlink(devpath.c_str());
-
-        if (android::base::StartsWith(devpath, "/dev/dm-user/")) {
-            std::error_code ec;
-            if (std::filesystem::is_empty("/dev/dm-user/", ec)) {
-                rmdir("/dev/dm-user");
-            }
-        }
     }
 }
 
diff --git a/init/first_stage_console.cpp b/init/first_stage_console.cpp
index cfa0d99..0f01166 100644
--- a/init/first_stage_console.cpp
+++ b/init/first_stage_console.cpp
@@ -30,6 +30,41 @@
 #include <android-base/file.h>
 #include <android-base/logging.h>
 
+static bool KernelConsolePresent(const std::string& cmdline) {
+    size_t pos = 0;
+    while (true) {
+        pos = cmdline.find("console=", pos);
+        if (pos == std::string::npos) return false;
+        if (pos == 0 || cmdline[pos - 1] == ' ') return true;
+        pos++;
+    }
+}
+
+static bool SetupConsole() {
+    if (mknod("/dev/console", S_IFCHR | 0600, makedev(5, 1))) {
+        PLOG(ERROR) << "unable to create /dev/console";
+        return false;
+    }
+    int fd = -1;
+    int tries = 50;  // should timeout after 5s
+    // The device driver for console may not be ready yet so retry for a while in case of failure.
+    while (tries--) {
+        fd = open("/dev/console", O_RDWR);
+        if (fd != -1) break;
+        std::this_thread::sleep_for(100ms);
+    }
+    if (fd == -1) {
+        PLOG(ERROR) << "could not open /dev/console";
+        return false;
+    }
+    ioctl(fd, TIOCSCTTY, 0);
+    dup2(fd, STDIN_FILENO);
+    dup2(fd, STDOUT_FILENO);
+    dup2(fd, STDERR_FILENO);
+    close(fd);
+    return true;
+}
+
 static void RunScript() {
     LOG(INFO) << "Attempting to run /first_stage.sh...";
     pid_t pid = fork();
@@ -48,11 +83,9 @@
 namespace android {
 namespace init {
 
-void StartConsole() {
-    if (mknod("/dev/console", S_IFCHR | 0600, makedev(5, 1))) {
-        PLOG(ERROR) << "unable to create /dev/console";
-        return;
-    }
+void StartConsole(const std::string& cmdline) {
+    bool console = KernelConsolePresent(cmdline);
+
     pid_t pid = fork();
     if (pid != 0) {
         int status;
@@ -60,31 +93,15 @@
         LOG(ERROR) << "console shell exited with status " << status;
         return;
     }
-    int fd = -1;
-    int tries = 50; // should timeout after 5s
-    // The device driver for console may not be ready yet so retry for a while in case of failure.
-    while (tries--) {
-        fd = open("/dev/console", O_RDWR);
-        if (fd != -1) {
-            break;
-        }
-        std::this_thread::sleep_for(100ms);
-    }
-    if (fd == -1) {
-        LOG(ERROR) << "Could not open /dev/console, errno = " << errno;
-        _exit(127);
-    }
-    ioctl(fd, TIOCSCTTY, 0);
-    dup2(fd, STDIN_FILENO);
-    dup2(fd, STDOUT_FILENO);
-    dup2(fd, STDERR_FILENO);
-    close(fd);
 
+    if (console) console = SetupConsole();
     RunScript();
-    const char* path = "/system/bin/sh";
-    const char* args[] = {path, nullptr};
-    int rv = execv(path, const_cast<char**>(args));
-    LOG(ERROR) << "unable to execv, returned " << rv << " errno " << errno;
+    if (console) {
+        const char* path = "/system/bin/sh";
+        const char* args[] = {path, nullptr};
+        int rv = execv(path, const_cast<char**>(args));
+        LOG(ERROR) << "unable to execv, returned " << rv << " errno " << errno;
+    }
     _exit(127);
 }
 
diff --git a/init/first_stage_console.h b/init/first_stage_console.h
index 8f36a7c..d5744df 100644
--- a/init/first_stage_console.h
+++ b/init/first_stage_console.h
@@ -28,7 +28,7 @@
     MAX_PARAM_VALUE = IGNORE_FAILURE,
 };
 
-void StartConsole();
+void StartConsole(const std::string& cmdline);
 int FirstStageConsole(const std::string& cmdline);
 
 }  // namespace init
diff --git a/init/first_stage_init.cpp b/init/first_stage_init.cpp
index d2a952f..91aaffd 100644
--- a/init/first_stage_init.cpp
+++ b/init/first_stage_init.cpp
@@ -221,6 +221,7 @@
     CHECKCALL(mount("tmpfs", "/dev", "tmpfs", MS_NOSUID, "mode=0755"));
     CHECKCALL(mkdir("/dev/pts", 0755));
     CHECKCALL(mkdir("/dev/socket", 0755));
+    CHECKCALL(mkdir("/dev/dm-user", 0755));
     CHECKCALL(mount("devpts", "/dev/pts", "devpts", 0, NULL));
 #define MAKE_STR(x) __STRING(x)
     CHECKCALL(mount("proc", "/proc", "proc", 0, "hidepid=2,gid=" MAKE_STR(AID_READPROC)));
@@ -307,7 +308,7 @@
     }
 
     if (want_console == FirstStageConsoleParam::CONSOLE_ON_FAILURE) {
-        StartConsole();
+        StartConsole(cmdline);
     }
 
     if (access(kBootImageRamdiskProp, F_OK) == 0) {
diff --git a/init/selinux.cpp b/init/selinux.cpp
index 5a0255a..f03ca6b 100644
--- a/init/selinux.cpp
+++ b/init/selinux.cpp
@@ -534,6 +534,7 @@
     selinux_android_restorecon("/dev/__properties__", 0);
 
     selinux_android_restorecon("/dev/block", SELINUX_ANDROID_RESTORECON_RECURSE);
+    selinux_android_restorecon("/dev/dm-user", SELINUX_ANDROID_RESTORECON_RECURSE);
     selinux_android_restorecon("/dev/device-mapper", 0);
 
     selinux_android_restorecon("/apex", 0);
diff --git a/init/service.cpp b/init/service.cpp
index ecc86d9..7b98392 100644
--- a/init/service.cpp
+++ b/init/service.cpp
@@ -329,8 +329,8 @@
                     LOG(FATAL) << "critical process '" << name_ << "' exited 4 times "
                                << exit_reason;
                 } else {
-                    LOG(ERROR) << "updatable process '" << name_ << "' exited 4 times "
-                               << exit_reason;
+                    LOG(ERROR) << "process with updatable components '" << name_
+                               << "' exited 4 times " << exit_reason;
                     // Notifies update_verifier and apexd
                     SetProperty("sys.init.updatable_crashing_process_name", name_);
                     SetProperty("sys.init.updatable_crashing", "1");
diff --git a/libprocessgroup/task_profiles.cpp b/libprocessgroup/task_profiles.cpp
index a638fca..4e767db 100644
--- a/libprocessgroup/task_profiles.cpp
+++ b/libprocessgroup/task_profiles.cpp
@@ -24,6 +24,7 @@
 #include <android-base/file.h>
 #include <android-base/logging.h>
 #include <android-base/stringprintf.h>
+#include <android-base/strings.h>
 #include <android-base/threads.h>
 
 #include <cutils/android_filesystem_config.h>
@@ -38,6 +39,7 @@
 
 using android::base::GetThreadId;
 using android::base::StringPrintf;
+using android::base::StringReplace;
 using android::base::unique_fd;
 using android::base::WriteStringToFile;
 
@@ -257,6 +259,39 @@
     return true;
 }
 
+bool WriteFileAction::ExecuteForProcess(uid_t uid, pid_t pid) const {
+    std::string filepath(filepath_), value(value_);
+
+    filepath = StringReplace(filepath, "<uid>", std::to_string(uid), true);
+    filepath = StringReplace(filepath, "<pid>", std::to_string(pid), true);
+    value = StringReplace(value, "<uid>", std::to_string(uid), true);
+    value = StringReplace(value, "<pid>", std::to_string(pid), true);
+
+    if (!WriteStringToFile(value, filepath)) {
+        PLOG(ERROR) << "Failed to write '" << value << "' to " << filepath;
+        return false;
+    }
+
+    return true;
+}
+
+bool WriteFileAction::ExecuteForTask(int tid) const {
+    std::string filepath(filepath_), value(value_);
+    int uid = getuid();
+
+    filepath = StringReplace(filepath, "<uid>", std::to_string(uid), true);
+    filepath = StringReplace(filepath, "<pid>", std::to_string(tid), true);
+    value = StringReplace(value, "<uid>", std::to_string(uid), true);
+    value = StringReplace(value, "<pid>", std::to_string(tid), true);
+
+    if (!WriteStringToFile(value, filepath)) {
+        PLOG(ERROR) << "Failed to write '" << value << "' to " << filepath;
+        return false;
+    }
+
+    return true;
+}
+
 bool ApplyProfileAction::ExecuteForProcess(uid_t uid, pid_t pid) const {
     for (const auto& profile : profiles_) {
         if (!profile->ExecuteForProcess(uid, pid)) {
@@ -459,6 +494,18 @@
                 } else {
                     LOG(WARNING) << "SetClamps: invalid parameter: " << boost_value;
                 }
+            } else if (action_name == "WriteFile") {
+                std::string attr_filepath = params_val["FilePath"].asString();
+                std::string attr_value = params_val["Value"].asString();
+                if (!attr_filepath.empty() && !attr_value.empty()) {
+                    profile->Add(std::make_unique<WriteFileAction>(attr_filepath, attr_value));
+                } else if (attr_filepath.empty()) {
+                    LOG(WARNING) << "WriteFile: invalid parameter: "
+                                 << "empty filepath";
+                } else if (attr_value.empty()) {
+                    LOG(WARNING) << "WriteFile: invalid parameter: "
+                                 << "empty value";
+                }
             } else {
                 LOG(WARNING) << "Unknown profile action: " << action_name;
             }
diff --git a/libprocessgroup/task_profiles.h b/libprocessgroup/task_profiles.h
index 2983a09..98bcb0e 100644
--- a/libprocessgroup/task_profiles.h
+++ b/libprocessgroup/task_profiles.h
@@ -139,6 +139,19 @@
     bool IsFdValid() const { return fd_ > FDS_INACCESSIBLE; }
 };
 
+// Write to file action
+class WriteFileAction : public ProfileAction {
+  public:
+    WriteFileAction(const std::string& filepath, const std::string& value) noexcept
+        : filepath_(filepath), value_(value) {}
+
+    virtual bool ExecuteForProcess(uid_t uid, pid_t pid) const;
+    virtual bool ExecuteForTask(int tid) const;
+
+  private:
+    std::string filepath_, value_;
+};
+
 class TaskProfile {
   public:
     TaskProfile() : res_cached_(false) {}
diff --git a/libutils/String16_test.cpp b/libutils/String16_test.cpp
index f1f24c3..2505f44 100644
--- a/libutils/String16_test.cpp
+++ b/libutils/String16_test.cpp
@@ -215,4 +215,16 @@
     EXPECT_TRUE(tmp.isStaticString());
 }
 
+TEST(String16Test, OverreadUtf8Conversion) {
+    char tmp[] = {'a', static_cast<char>(0xe0), '\0'};
+    String16 another(tmp);
+    EXPECT_TRUE(another.size() == 0);
+}
+
+TEST(String16Test, ValidUtf8Conversion) {
+    String16 another("abcdef");
+    EXPECT_EQ(6U, another.size());
+    EXPECT_STR16EQ(another, u"abcdef");
+}
+
 }  // namespace android
diff --git a/libutils/String8_test.cpp b/libutils/String8_test.cpp
index 3947a5f..9efcc6f 100644
--- a/libutils/String8_test.cpp
+++ b/libutils/String8_test.cpp
@@ -96,4 +96,9 @@
     EXPECT_EQ(10U, string8.length());
 }
 
+TEST_F(String8Test, ValidUtf16Conversion) {
+    char16_t tmp[] = u"abcdef";
+    String8 valid = String8(String16(tmp));
+    EXPECT_STREQ(valid, "abcdef");
+}
 }
diff --git a/rootdir/Android.bp b/rootdir/Android.bp
index 96b5e0d..a21f686 100644
--- a/rootdir/Android.bp
+++ b/rootdir/Android.bp
@@ -24,3 +24,9 @@
     src: "ueventd.rc",
     recovery_available: true,
 }
+
+// TODO(b/147210213) Generate list of libraries during build and fill in at build time
+linker_config {
+    name: "system_linker_config",
+    src: "etc/linker.config.json",
+}
diff --git a/rootdir/avb/Android.mk b/rootdir/avb/Android.mk
index 9892ae7..c8fc1d6 100644
--- a/rootdir/avb/Android.mk
+++ b/rootdir/avb/Android.mk
@@ -1,11 +1,17 @@
 LOCAL_PATH:= $(call my-dir)
 
-ifeq ($(BOARD_USES_RECOVERY_AS_BOOT),true)
-  my_gsi_avb_keys_path := $(TARGET_RECOVERY_ROOT_OUT)/first_stage_ramdisk/avb
-else ifeq ($(BOARD_MOVE_GSI_AVB_KEYS_TO_VENDOR_BOOT),true)
-  my_gsi_avb_keys_path := $(TARGET_VENDOR_RAMDISK_OUT)/avb
+ifeq ($(BOARD_MOVE_GSI_AVB_KEYS_TO_VENDOR_BOOT),true) # AVB keys are installed to vendor ramdisk
+  ifeq ($(BOARD_MOVE_RECOVERY_RESOURCES_TO_VENDOR_BOOT),true) # no dedicated recovery partition
+    my_gsi_avb_keys_path := $(TARGET_VENDOR_RAMDISK_OUT)/first_stage_ramdisk/avb
+  else # device has a dedicated recovery partition
+    my_gsi_avb_keys_path := $(TARGET_VENDOR_RAMDISK_OUT)/avb
+  endif
 else
-  my_gsi_avb_keys_path := $(TARGET_RAMDISK_OUT)/avb
+  ifeq ($(BOARD_USES_RECOVERY_AS_BOOT),true) # no dedicated recovery partition
+    my_gsi_avb_keys_path := $(TARGET_RECOVERY_ROOT_OUT)/first_stage_ramdisk/avb
+  else # device has a dedicated recovery partition
+    my_gsi_avb_keys_path := $(TARGET_RAMDISK_OUT)/avb
+  endif
 endif
 
 #######################################
diff --git a/rootdir/etc/linker.config.json b/rootdir/etc/linker.config.json
new file mode 100644
index 0000000..d66ab73
--- /dev/null
+++ b/rootdir/etc/linker.config.json
@@ -0,0 +1,72 @@
+{
+  // These are list of libraries which has stub interface and installed
+  // in system image so other partition and APEX modules can link to it.
+  // TODO(b/147210213) : Generate this list on build and read from the file
+  "provideLibs": [
+    // LLNDK libraries
+    "libEGL.so",
+    "libGLESv1_CM.so",
+    "libGLESv2.so",
+    "libGLESv3.so",
+    "libRS.so",
+    "libandroid_net.so",
+    "libbinder_ndk.so",
+    "libc.so",
+    "libcgrouprc.so",
+    "libclang_rt.asan-arm-android.so",
+    "libclang_rt.asan-i686-android.so",
+    "libclang_rt.asan-x86_64-android.so",
+    "libdl.so",
+    "libft2.so",
+    "liblog.so",
+    "libm.so",
+    "libmediandk.so",
+    "libnativewindow.so",
+    "libsync.so",
+    "libvndksupport.so",
+    "libvulkan.so",
+    // NDK libraries
+    "libaaudio.so",
+    "libandroid.so",
+    // adb
+    "libadbd_auth.so",
+    "libadbd_fs.so",
+    // bionic
+    "libdl_android.so",
+    // statsd
+    "libincident.so",
+    // media
+    "libmediametrics.so",
+    // nn
+    "libneuralnetworks_packageinfo.so",
+    // SELinux
+    "libselinux.so"
+  ],
+  "requireLibs": [
+    // Keep in sync with the "platform" namespace in art/build/apex/ld.config.txt.
+    "libdexfile_external.so",
+    "libdexfiled_external.so",
+    "libnativebridge.so",
+    "libnativehelper.so",
+    "libnativeloader.so",
+    "libandroidicu.so",
+    "libicu.so",
+    // TODO(b/122876336): Remove libpac.so once it's migrated to Webview
+    "libpac.so",
+    // TODO(b/120786417 or b/134659294): libicuuc.so
+    // and libicui18n.so are kept for app compat.
+    "libicui18n.so",
+    "libicuuc.so",
+    // resolv
+    "libnetd_resolv.so",
+    // nn
+    "libneuralnetworks.so",
+    // statsd
+    "libstatspull.so",
+    "libstatssocket.so",
+    // adbd
+    "libadb_pairing_auth.so",
+    "libadb_pairing_connection.so",
+    "libadb_pairing_server.so"
+  ]
+}
\ No newline at end of file
diff --git a/rootdir/init.rc b/rootdir/init.rc
index 746fc61..4219e32 100644
--- a/rootdir/init.rc
+++ b/rootdir/init.rc
@@ -792,6 +792,7 @@
     # Delete these if need be, per b/139193659
     mkdir /data/rollback 0700 system system encryption=DeleteIfNecessary
     mkdir /data/rollback-observer 0700 system system encryption=DeleteIfNecessary
+    mkdir /data/rollback-history 0700 system system encryption=DeleteIfNecessary
 
     # Create root dir for Incremental Service
     mkdir /data/incremental 0771 system system encryption=Require
@@ -807,6 +808,10 @@
     wait_for_prop apexd.status activated
     perform_apex_config
 
+    # After apexes are mounted, tell keymaster early boot has ended, so it will
+    # stop allowing use of early-boot keys
+    exec - system system -- /system/bin/vdc keymaster early-boot-ended
+
     # Special-case /data/media/obb per b/64566063
     mkdir /data/media 0770 media_rw media_rw encryption=None
     exec - media_rw media_rw -- /system/bin/chattr +F /data/media
diff --git a/toolbox/start.cpp b/toolbox/start.cpp
index cffb89c..46314cf 100644
--- a/toolbox/start.cpp
+++ b/toolbox/start.cpp
@@ -37,7 +37,6 @@
 
 static void ControlDefaultServices(bool start) {
     std::vector<std::string> services = {
-        "iorapd",
         "netd",
         "surfaceflinger",
         "audioserver",
@@ -92,4 +91,4 @@
 
 extern "C" int stop_main(int argc, char** argv) {
     return StartStop(argc, argv, false);
-}
\ No newline at end of file
+}