diff --git a/base b/base
deleted file mode 120000
index 622c552..0000000
--- a/base
+++ /dev/null
@@ -1 +0,0 @@
-../libbase
\ No newline at end of file
diff --git a/fs_mgr/libsnapshot/Android.bp b/fs_mgr/libsnapshot/Android.bp
index 95606d7..059a469 100644
--- a/fs_mgr/libsnapshot/Android.bp
+++ b/fs_mgr/libsnapshot/Android.bp
@@ -35,9 +35,11 @@
         "update_metadata-protos",
     ],
     whole_static_libs: [
+        "libcutils",
         "libext2_uuid",
         "libext4_utils",
         "libfstab",
+        "libsnapshot_snapuserd",
     ],
     header_libs: [
         "libchrome",
@@ -239,6 +241,7 @@
     srcs: [
         "partition_cow_creator_test.cpp",
         "snapshot_metadata_updater_test.cpp",
+        "snapshot_reader_test.cpp",
         "snapshot_test.cpp",
     ],
     shared_libs: [
@@ -571,3 +574,27 @@
     auto_gen_config: true,
     require_root: false,
 }
+
+cc_binary {
+    name: "inspect_cow",
+    host_supported: true,
+    device_supported: true,
+    cflags: [
+        "-D_FILE_OFFSET_BITS=64",
+        "-Wall",
+        "-Werror",
+    ],
+    static_libs: [
+        "libbase",
+        "libbrotli",
+        "libcrypto_static",
+        "liblog",
+        "libsnapshot_cow",
+        "libz",
+    ],
+    shared_libs: [
+    ],
+    srcs: [
+        "inspect_cow.cpp",
+    ],
+}
diff --git a/fs_mgr/libsnapshot/cow_reader.cpp b/fs_mgr/libsnapshot/cow_reader.cpp
index 5fac0ac..f10ccb6 100644
--- a/fs_mgr/libsnapshot/cow_reader.cpp
+++ b/fs_mgr/libsnapshot/cow_reader.cpp
@@ -18,6 +18,7 @@
 #include <unistd.h>
 
 #include <limits>
+#include <optional>
 #include <vector>
 
 #include <android-base/file.h>
@@ -117,8 +118,7 @@
         PLOG(ERROR) << "lseek ops failed";
         return false;
     }
-    uint64_t next_last_label = 0;
-    bool has_next = 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;
@@ -146,11 +146,23 @@
                 has_last_label_ = true;
                 last_label_ = current_op.source;
             } else {
-                last_label_ = next_last_label;
-                if (has_next) has_last_label_ = true;
-                next_last_label = current_op.source;
-                has_next = true;
+                if (next_last_label) {
+                    last_label_ = next_last_label.value();
+                    has_last_label_ = true;
+                }
+                next_last_label = {current_op.source};
             }
+        } else if (current_op.type == kCowFooterOp) {
+            memcpy(&footer_.op, &current_op, sizeof(footer_.op));
+
+            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;
+                }
+            }
+            break;
         }
     }
 
diff --git a/fs_mgr/libsnapshot/include/libsnapshot/cow_reader.h b/fs_mgr/libsnapshot/include/libsnapshot/cow_reader.h
index 814e104..b863ff2 100644
--- a/fs_mgr/libsnapshot/include/libsnapshot/cow_reader.h
+++ b/fs_mgr/libsnapshot/include/libsnapshot/cow_reader.h
@@ -39,11 +39,17 @@
     // maximum number of bytes that can be written to the returned buffer.
     //
     // The returned buffer is owned by IByteSink, but must remain valid until
-    // the ready operation has completed (or the entire buffer has been
+    // the read operation has completed (or the entire buffer has been
     // covered by calls to ReturnData).
     //
     // After calling GetBuffer(), all previous buffers returned are no longer
     // valid.
+    //
+    // GetBuffer() is intended to be sequential. A returned size of N indicates
+    // that the output stream will advance by N bytes, and the ReturnData call
+    // indicates that those bytes have been fulfilled. Therefore, it is
+    // possible to have ReturnBuffer do nothing, if the implementation doesn't
+    // care about incremental writes.
     virtual void* GetBuffer(size_t requested, size_t* actual) = 0;
 
     // Called when a section returned by |GetBuffer| has been filled with data.
diff --git a/fs_mgr/libsnapshot/include/libsnapshot/mock_snapshot.h b/fs_mgr/libsnapshot/include/libsnapshot/mock_snapshot.h
index 6dee3d4..92e7910 100644
--- a/fs_mgr/libsnapshot/include/libsnapshot/mock_snapshot.h
+++ b/fs_mgr/libsnapshot/include/libsnapshot/mock_snapshot.h
@@ -47,6 +47,8 @@
     MOCK_METHOD(bool, CreateLogicalAndSnapshotPartitions,
                 (const std::string& super_device, const std::chrono::milliseconds& timeout_ms),
                 (override));
+    MOCK_METHOD(bool, MapAllSnapshots, (const std::chrono::milliseconds& timeout_ms), (override));
+    MOCK_METHOD(bool, UnmapAllSnapshots, (), (override));
     MOCK_METHOD(bool, HandleImminentDataWipe, (const std::function<void()>& callback), (override));
     MOCK_METHOD(bool, FinishMergeInRecovery, (), (override));
     MOCK_METHOD(CreateResult, RecoveryCreateSnapshotDevices, (), (override));
diff --git a/fs_mgr/libsnapshot/include/libsnapshot/snapshot.h b/fs_mgr/libsnapshot/include/libsnapshot/snapshot.h
index 4bbdca3..35ed04a 100644
--- a/fs_mgr/libsnapshot/include/libsnapshot/snapshot.h
+++ b/fs_mgr/libsnapshot/include/libsnapshot/snapshot.h
@@ -37,6 +37,7 @@
 #include <libsnapshot/auto_device.h>
 #include <libsnapshot/return.h>
 #include <libsnapshot/snapshot_writer.h>
+#include <libsnapshot/snapuserd_client.h>
 
 #ifndef FRIEND_TEST
 #define FRIEND_TEST(test_set_name, individual_test) \
@@ -185,6 +186,9 @@
     // must be suffixed. If a source partition exists, it must be specified as well. The source
     // partition will only be used if raw bytes are needed. The source partition should be an
     // absolute path to the device, not a partition name.
+    //
+    // After calling OpenSnapshotWriter, the caller must invoke Initialize or InitializeForAppend
+    // before invoking write operations.
     virtual std::unique_ptr<ISnapshotWriter> OpenSnapshotWriter(
             const android::fs_mgr::CreateLogicalPartitionParams& params,
             const std::optional<std::string>& source_device) = 0;
@@ -203,6 +207,14 @@
     virtual bool CreateLogicalAndSnapshotPartitions(
             const std::string& super_device, const std::chrono::milliseconds& timeout_ms = {}) = 0;
 
+    // Map all snapshots. This is analogous to CreateLogicalAndSnapshotPartitions, except it maps
+    // the target slot rather than the current slot. It should only be used immediately after
+    // applying an update, before rebooting to the new slot.
+    virtual bool MapAllSnapshots(const std::chrono::milliseconds& timeout_ms = {}) = 0;
+
+    // Unmap all snapshots. This should be called to undo MapAllSnapshots().
+    virtual bool UnmapAllSnapshots() = 0;
+
     // This method should be called preceding any wipe or flash of metadata or
     // userdata. It is only valid in recovery or fastbootd, and it ensures that
     // a merge has been completed.
@@ -318,6 +330,8 @@
     bool Dump(std::ostream& os) override;
     std::unique_ptr<AutoDevice> EnsureMetadataMounted() override;
     ISnapshotMergeStats* GetSnapshotMergeStatsInstance() override;
+    bool MapAllSnapshots(const std::chrono::milliseconds& timeout_ms = {}) override;
+    bool UnmapAllSnapshots() override;
 
   private:
     FRIEND_TEST(SnapshotTest, CleanFirstStageMount);
@@ -355,6 +369,9 @@
     // This is created lazily since it can connect via binder.
     bool EnsureImageManager();
 
+    // Ensure we're connected to snapuserd.
+    bool EnsureSnapuserdConnected();
+
     // Helper for first-stage init.
     bool ForceLocalImageManager();
 
@@ -411,6 +428,11 @@
                      const std::string& cow_device, const std::chrono::milliseconds& timeout_ms,
                      std::string* dev_path);
 
+    // Create a dm-user device for a given snapshot.
+    bool MapDmUserCow(LockedFile* lock, const std::string& name, const std::string& cow_file,
+                      const std::string& base_device, const std::chrono::milliseconds& timeout_ms,
+                      std::string* path);
+
     // Map a COW image that was previous created with CreateCowImage.
     std::optional<std::string> MapCowImage(const std::string& name,
                                            const std::chrono::milliseconds& timeout_ms);
@@ -639,6 +661,7 @@
     std::unique_ptr<IImageManager> images_;
     bool has_local_image_manager_ = false;
     bool in_factory_data_reset_ = false;
+    std::unique_ptr<SnapuserdClient> snapuserd_client_;
 };
 
 }  // namespace snapshot
diff --git a/fs_mgr/libsnapshot/include/libsnapshot/snapshot_stub.h b/fs_mgr/libsnapshot/include/libsnapshot/snapshot_stub.h
index ed790a0..cba3560 100644
--- a/fs_mgr/libsnapshot/include/libsnapshot/snapshot_stub.h
+++ b/fs_mgr/libsnapshot/include/libsnapshot/snapshot_stub.h
@@ -52,6 +52,8 @@
     bool Dump(std::ostream& os) override;
     std::unique_ptr<AutoDevice> EnsureMetadataMounted() override;
     ISnapshotMergeStats* GetSnapshotMergeStatsInstance() override;
+    bool MapAllSnapshots(const std::chrono::milliseconds& timeout_ms) override;
+    bool UnmapAllSnapshots() override;
 };
 
 }  // namespace android::snapshot
diff --git a/fs_mgr/libsnapshot/include/libsnapshot/snapshot_writer.h b/fs_mgr/libsnapshot/include/libsnapshot/snapshot_writer.h
index f76f545..4732c2d 100644
--- a/fs_mgr/libsnapshot/include/libsnapshot/snapshot_writer.h
+++ b/fs_mgr/libsnapshot/include/libsnapshot/snapshot_writer.h
@@ -14,6 +14,8 @@
 
 #pragma once
 
+#include <optional>
+
 #include <android-base/unique_fd.h>
 
 #include <libsnapshot/cow_writer.h>
@@ -37,14 +39,22 @@
     // device is only opened on the first operation that requires it.
     void SetSourceDevice(const std::string& source_device);
 
+    // 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
+    // from. See CowWriter::InitializeAppend.
+    virtual bool InitializeAppend(std::optional<uint64_t> label = {}) = 0;
+
     virtual std::unique_ptr<FileDescriptor> OpenReader() = 0;
 
   protected:
     android::base::borrowed_fd GetSourceFd();
 
+    std::optional<std::string> source_device_;
+
   private:
     android::base::unique_fd source_fd_;
-    std::optional<std::string> source_device_;
 };
 
 // Send writes to a COW or a raw device directly, based on a threshold.
@@ -52,9 +62,11 @@
   public:
     CompressedSnapshotWriter(const CowOptions& options);
 
-    // Sets the COW device, if needed.
+    // Sets the COW device; this is required.
     bool SetCowDevice(android::base::unique_fd&& cow_device);
 
+    bool Initialize() override;
+    bool InitializeAppend(std::optional<uint64_t> label = {}) override;
     bool Finalize() override;
     uint64_t GetCowSize() override;
     std::unique_ptr<FileDescriptor> OpenReader() override;
@@ -79,6 +91,9 @@
     // Set the device used for all writes.
     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 Finalize() override;
     uint64_t GetCowSize() override { return cow_size_; }
     std::unique_ptr<FileDescriptor> OpenReader() override;
diff --git a/fs_mgr/libsnapshot/include/libsnapshot/snapuserd.h b/fs_mgr/libsnapshot/include/libsnapshot/snapuserd.h
index 2f727d6..80f87d9 100644
--- a/fs_mgr/libsnapshot/include/libsnapshot/snapuserd.h
+++ b/fs_mgr/libsnapshot/include/libsnapshot/snapuserd.h
@@ -77,7 +77,7 @@
     int ReadDiskExceptions(chunk_t chunk, size_t size);
     int ReadData(chunk_t chunk, size_t size);
 
-    std::string GetControlDevicePath() { return control_device_; }
+    const std::string& GetControlDevicePath() { return control_device_; }
 
   private:
     int ProcessReplaceOp(const CowOperation* cow_op);
diff --git a/fs_mgr/libsnapshot/include/libsnapshot/snapuserd_client.h b/fs_mgr/libsnapshot/include/libsnapshot/snapuserd_client.h
index dffd481..0bbdaa5 100644
--- a/fs_mgr/libsnapshot/include/libsnapshot/snapuserd_client.h
+++ b/fs_mgr/libsnapshot/include/libsnapshot/snapuserd_client.h
@@ -53,6 +53,10 @@
     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);
+
+    // Wait for snapuserd to disassociate with a dm-user control device. This
+    // must ONLY be called if the control device has already been deleted.
+    bool WaitForDeviceDelete(const std::string& control_device);
 };
 
 }  // namespace snapshot
diff --git a/fs_mgr/libsnapshot/include/libsnapshot/snapuserd_server.h b/fs_mgr/libsnapshot/include/libsnapshot/snapuserd_server.h
index 357acac..181ee33 100644
--- a/fs_mgr/libsnapshot/include/libsnapshot/snapuserd_server.h
+++ b/fs_mgr/libsnapshot/include/libsnapshot/snapuserd_server.h
@@ -21,12 +21,14 @@
 #include <functional>
 #include <future>
 #include <iostream>
+#include <mutex>
 #include <sstream>
 #include <string>
 #include <thread>
 #include <vector>
 
 #include <android-base/unique_fd.h>
+#include <libsnapshot/snapuserd.h>
 
 namespace android {
 namespace snapshot {
@@ -37,19 +39,23 @@
     START,
     QUERY,
     STOP,
+    DELETE,
     INVALID,
 };
 
 class DmUserHandler {
   private:
-    std::unique_ptr<std::thread> threadHandler_;
+    std::thread thread_;
+    std::unique_ptr<Snapuserd> snapuserd_;
 
   public:
-    void SetThreadHandler(std::function<void(void)> func) {
-        threadHandler_ = std::make_unique<std::thread>(func);
-    }
+    explicit DmUserHandler(std::unique_ptr<Snapuserd>&& snapuserd)
+        : snapuserd_(std::move(snapuserd)) {}
 
-    std::unique_ptr<std::thread>& GetThreadHandler() { return threadHandler_; }
+    const std::unique_ptr<Snapuserd>& snapuserd() const { return snapuserd_; }
+    std::thread& thread() { return thread_; }
+
+    const std::string& GetControlDevice() const;
 };
 
 class Stoppable {
@@ -61,9 +67,6 @@
 
     virtual ~Stoppable() {}
 
-    virtual void ThreadStart(std::string cow_device, std::string backing_device,
-                             std::string control_device) = 0;
-
     bool StopRequested() {
         // checks if value in future object is available
         if (futureObj_.wait_for(std::chrono::milliseconds(0)) == std::future_status::timeout)
@@ -78,27 +81,33 @@
   private:
     android::base::unique_fd sockfd_;
     bool terminating_;
-    std::vector<std::unique_ptr<DmUserHandler>> dm_users_;
     std::vector<struct pollfd> watched_fds_;
 
+    std::mutex lock_;
+    std::vector<std::unique_ptr<DmUserHandler>> dm_users_;
+
     void AddWatchedFd(android::base::borrowed_fd fd);
     void AcceptClient();
     bool HandleClient(android::base::borrowed_fd fd, int revents);
     bool Recv(android::base::borrowed_fd fd, std::string* data);
     bool Sendmsg(android::base::borrowed_fd fd, const std::string& msg);
-    bool Receivemsg(android::base::borrowed_fd fd, const std::string& msg);
+    bool Receivemsg(android::base::borrowed_fd fd, const std::string& str);
 
-    void ThreadStart(std::string cow_device, std::string backing_device,
-                     std::string control_device) override;
     void ShutdownThreads();
+    bool WaitForDelete(const std::string& control_device);
     DaemonOperations Resolveop(std::string& input);
     std::string GetDaemonStatus();
     void Parsemsg(std::string const& msg, const char delim, std::vector<std::string>& out);
 
     void SetTerminating() { terminating_ = true; }
-
     bool IsTerminating() { return terminating_; }
 
+    void RunThread(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);
+
   public:
     SnapuserdServer() { terminating_ = false; }
     ~SnapuserdServer();
diff --git a/fs_mgr/libsnapshot/inspect_cow.cpp b/fs_mgr/libsnapshot/inspect_cow.cpp
new file mode 100644
index 0000000..6046bad
--- /dev/null
+++ b/fs_mgr/libsnapshot/inspect_cow.cpp
@@ -0,0 +1,90 @@
+//
+// Copyright (C) 2020 The Android Open Source Project
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//      http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+//
+#include <stdio.h>
+
+#include <iostream>
+#include <string>
+
+#include <android-base/logging.h>
+#include <android-base/unique_fd.h>
+#include <libsnapshot/cow_reader.h>
+
+namespace android {
+namespace snapshot {
+
+void MyLogger(android::base::LogId, android::base::LogSeverity severity, const char*, const char*,
+              unsigned int, const char* message) {
+    if (severity == android::base::ERROR) {
+        fprintf(stderr, "%s\n", message);
+    } else {
+        fprintf(stdout, "%s\n", message);
+    }
+}
+
+static bool Inspect(const std::string& path) {
+    android::base::unique_fd fd(open(path.c_str(), O_RDONLY));
+    if (fd < 0) {
+        PLOG(ERROR) << "open failed: " << path;
+        return false;
+    }
+
+    CowReader reader;
+    if (!reader.Parse(fd)) {
+        LOG(ERROR) << "parse failed: " << path;
+        return false;
+    }
+
+    CowHeader header;
+    if (!reader.GetHeader(&header)) {
+        LOG(ERROR) << "could not get header: " << path;
+        return false;
+    }
+
+    std::cout << "Major version: " << header.major_version << "\n";
+    std::cout << "Minor version: " << header.minor_version << "\n";
+    std::cout << "Header size: " << header.header_size << "\n";
+    std::cout << "Footer size: " << header.footer_size << "\n";
+    std::cout << "Block size: " << header.block_size << "\n";
+    std::cout << "\n";
+
+    auto iter = reader.GetOpIter();
+    while (!iter->Done()) {
+        const CowOperation& op = iter->Get();
+
+        std::cout << op << "\n";
+
+        iter->Next();
+    }
+
+    return true;
+}
+
+}  // namespace snapshot
+}  // namespace android
+
+int main(int argc, char** argv) {
+    android::base::InitLogging(argv, android::snapshot::MyLogger);
+
+    if (argc < 2) {
+        LOG(ERROR) << "Usage: inspect_cow <COW_FILE>";
+        return 1;
+    }
+
+    if (!android::snapshot::Inspect(argv[1])) {
+        return 1;
+    }
+    return 0;
+}
diff --git a/fs_mgr/libsnapshot/snapshot.cpp b/fs_mgr/libsnapshot/snapshot.cpp
index 6574457..7061d56 100644
--- a/fs_mgr/libsnapshot/snapshot.cpp
+++ b/fs_mgr/libsnapshot/snapshot.cpp
@@ -56,6 +56,7 @@
 using android::dm::DmTable;
 using android::dm::DmTargetLinear;
 using android::dm::DmTargetSnapshot;
+using android::dm::DmTargetUser;
 using android::dm::kSectorSize;
 using android::dm::SnapshotStorageMode;
 using android::fiemap::FiemapStatus;
@@ -115,6 +116,10 @@
     return snapshot_name + "-cow";
 }
 
+static std::string GetDmUserCowName(const std::string& snapshot_name) {
+    return snapshot_name + "-user-cow";
+}
+
 static std::string GetCowImageDeviceName(const std::string& snapshot_name) {
     return snapshot_name + "-cow-img";
 }
@@ -370,6 +375,45 @@
     return Return(images_->CreateBackingImage(cow_image_name, status.cow_file_size(), cow_flags));
 }
 
+bool SnapshotManager::MapDmUserCow(LockedFile* lock, const std::string& name,
+                                   const std::string& cow_file, const std::string& base_device,
+                                   const std::chrono::milliseconds& timeout_ms, std::string* path) {
+    CHECK(lock);
+
+    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;
+    }
+
+    DmTable table;
+    table.Emplace<DmTargetUser>(0, base_sectors, name);
+    if (!dm.CreateDevice(name, table, path, timeout_ms)) {
+        return false;
+    }
+
+    if (!EnsureSnapuserdConnected()) {
+        return false;
+    }
+
+    auto control_device = "/dev/dm-user/" + name;
+    return snapuserd_client_->InitializeSnapuserd(cow_file, base_device, control_device);
+}
+
 bool SnapshotManager::MapSnapshot(LockedFile* lock, const std::string& name,
                                   const std::string& base_device, const std::string& cow_device,
                                   const std::chrono::milliseconds& timeout_ms,
@@ -1728,6 +1772,30 @@
         return true;
     }
 
+    if (IsCompressionEnabled()) {
+        auto name = GetDmUserCowName(params.GetPartitionName());
+
+        // :TODO: need to force init to process uevents for these in first-stage.
+        std::string cow_path;
+        if (!GetMappedImageDevicePath(cow_name, &cow_path)) {
+            LOG(ERROR) << "Could not determine path for: " << cow_name;
+            return false;
+        }
+
+        std::string new_cow_device;
+        if (!MapDmUserCow(lock, name, cow_path, base_path, remaining_time, &new_cow_device)) {
+            LOG(ERROR) << "Could not map dm-user device for partition "
+                       << params.GetPartitionName();
+            return false;
+        }
+        created_devices.EmplaceBack<AutoUnmapDevice>(&dm, name);
+
+        remaining_time = GetRemainingTime(params.timeout_ms, begin);
+        if (remaining_time.count() < 0) return false;
+
+        cow_device = new_cow_device;
+    }
+
     std::string path;
     if (!MapSnapshot(lock, params.GetPartitionName(), base_device, cow_device, remaining_time,
                      &path)) {
@@ -1847,6 +1915,22 @@
     if (!EnsureImageManager()) return false;
 
     auto& dm = DeviceMapper::Instance();
+
+    auto dm_user_name = GetDmUserCowName(name);
+    if (IsCompressionEnabled() && dm.GetState(dm_user_name) != DmDeviceState::INVALID) {
+        if (!EnsureSnapuserdConnected()) {
+            return false;
+        }
+        if (!dm.DeleteDevice(dm_user_name)) {
+            LOG(ERROR) << "Cannot unmap " << dm_user_name;
+            return false;
+        }
+        if (!snapuserd_client_->WaitForDeviceDelete("/dev/dm-user/" + dm_user_name)) {
+            LOG(ERROR) << "Failed to wait for " << dm_user_name << " control device to delete";
+            return false;
+        }
+    }
+
     auto cow_name = GetCowName(name);
     if (!dm.DeleteDeviceIfExists(cow_name)) {
         LOG(ERROR) << "Cannot unmap " << cow_name;
@@ -1861,6 +1945,16 @@
     return true;
 }
 
+bool SnapshotManager::MapAllSnapshots(const std::chrono::milliseconds&) {
+    LOG(ERROR) << "Not yet implemented.";
+    return false;
+}
+
+bool SnapshotManager::UnmapAllSnapshots() {
+    LOG(ERROR) << "Not yet implemented.";
+    return false;
+}
+
 auto SnapshotManager::OpenFile(const std::string& file, int lock_flags)
         -> std::unique_ptr<LockedFile> {
     unique_fd fd(open(file.c_str(), O_RDONLY | O_CLOEXEC | O_NOFOLLOW));
@@ -2117,6 +2211,20 @@
     return true;
 }
 
+bool SnapshotManager::EnsureSnapuserdConnected() {
+    if (!snapuserd_client_) {
+        if (!EnsureSnapuserdStarted()) {
+            return false;
+        }
+        snapuserd_client_ = SnapuserdClient::Connect(kSnapuserdSocket, 10s);
+        if (!snapuserd_client_) {
+            LOG(ERROR) << "Unable to connect to snapuserd";
+            return false;
+        }
+    }
+    return true;
+}
+
 bool SnapshotManager::ForceLocalImageManager() {
     images_ = android::fiemap::ImageManager::Open(gsid_dir_);
     if (!images_) {
diff --git a/fs_mgr/libsnapshot/snapshot_reader.cpp b/fs_mgr/libsnapshot/snapshot_reader.cpp
index 0d47468..a4a652a 100644
--- a/fs_mgr/libsnapshot/snapshot_reader.cpp
+++ b/fs_mgr/libsnapshot/snapshot_reader.cpp
@@ -14,13 +14,17 @@
 // limitations under the License.
 //
 
-#include <ext4_utils/ext4_utils.h>
-
 #include "snapshot_reader.h"
 
+#include <android-base/file.h>
+#include <android-base/logging.h>
+#include <ext4_utils/ext4_utils.h>
+
 namespace android {
 namespace snapshot {
 
+using android::base::borrowed_fd;
+
 // Not supported.
 bool ReadOnlyFileDescriptor::Open(const char*, int, mode_t) {
     errno = EINVAL;
@@ -73,5 +77,252 @@
     return true;
 }
 
+bool CompressedSnapshotReader::SetCow(std::unique_ptr<CowReader>&& cow) {
+    cow_ = std::move(cow);
+
+    CowHeader header;
+    if (!cow_->GetHeader(&header)) {
+        return false;
+    }
+    block_size_ = header.block_size;
+
+    // Populate the operation map.
+    op_iter_ = cow_->GetOpIter();
+    while (!op_iter_->Done()) {
+        const CowOperation* op = &op_iter_->Get();
+        if (op->new_block >= ops_.size()) {
+            ops_.resize(op->new_block + 1, nullptr);
+        }
+        ops_[op->new_block] = op;
+        op_iter_->Next();
+    }
+
+    return true;
+}
+
+void CompressedSnapshotReader::SetSourceDevice(const std::string& source_device) {
+    source_device_ = {source_device};
+}
+
+void CompressedSnapshotReader::SetBlockDeviceSize(uint64_t block_device_size) {
+    block_device_size_ = block_device_size;
+}
+
+borrowed_fd CompressedSnapshotReader::GetSourceFd() {
+    if (source_fd_ < 0) {
+        if (!source_device_) {
+            LOG(ERROR) << "CompressedSnapshotReader needs source device, but none was set";
+            errno = EINVAL;
+            return {-1};
+        }
+        source_fd_.reset(open(source_device_->c_str(), O_RDONLY | O_CLOEXEC));
+        if (source_fd_ < 0) {
+            PLOG(ERROR) << "open " << *source_device_;
+            return {-1};
+        }
+    }
+    return source_fd_;
+}
+
+class MemoryByteSink : public IByteSink {
+  public:
+    MemoryByteSink(void* buf, size_t count) {
+        buf_ = reinterpret_cast<uint8_t*>(buf);
+        pos_ = buf_;
+        end_ = buf_ + count;
+    }
+
+    void* GetBuffer(size_t requested, size_t* actual) override {
+        *actual = std::min(remaining(), requested);
+        if (!*actual) {
+            return nullptr;
+        }
+
+        uint8_t* start = pos_;
+        pos_ += *actual;
+        return start;
+    }
+
+    bool ReturnData(void*, size_t) override { return true; }
+
+    uint8_t* buf() const { return buf_; }
+    uint8_t* pos() const { return pos_; }
+    size_t remaining() const { return end_ - pos_; }
+
+  private:
+    uint8_t* buf_;
+    uint8_t* pos_;
+    uint8_t* end_;
+};
+
+ssize_t CompressedSnapshotReader::Read(void* buf, size_t count) {
+    // Find the start and end chunks, inclusive.
+    uint64_t start_chunk = offset_ / block_size_;
+    uint64_t end_chunk = (offset_ + count - 1) / block_size_;
+
+    // Chop off the first N bytes if the position is not block-aligned.
+    size_t start_offset = offset_ % block_size_;
+
+    MemoryByteSink sink(buf, count);
+
+    size_t initial_bytes = std::min(block_size_ - start_offset, sink.remaining());
+    ssize_t rv = ReadBlock(start_chunk, &sink, start_offset, initial_bytes);
+    if (rv < 0) {
+        return -1;
+    }
+    offset_ += rv;
+
+    for (uint64_t chunk = start_chunk + 1; chunk < end_chunk; chunk++) {
+        ssize_t rv = ReadBlock(chunk, &sink, 0);
+        if (rv < 0) {
+            return -1;
+        }
+        offset_ += rv;
+    }
+
+    if (sink.remaining()) {
+        ssize_t rv = ReadBlock(end_chunk, &sink, 0, {sink.remaining()});
+        if (rv < 0) {
+            return -1;
+        }
+        offset_ += rv;
+    }
+
+    errno = 0;
+
+    DCHECK(sink.pos() - sink.buf() == count);
+    return count;
+}
+
+// Discard the first N bytes of a sink request, or any excess bytes.
+class PartialSink : public MemoryByteSink {
+  public:
+    PartialSink(void* buffer, size_t size, size_t ignore_start)
+        : MemoryByteSink(buffer, size), ignore_start_(ignore_start) {}
+
+    void* GetBuffer(size_t requested, size_t* actual) override {
+        // Throw away the first N bytes if needed.
+        if (ignore_start_) {
+            *actual = std::min({requested, ignore_start_, sizeof(discard_)});
+            ignore_start_ -= *actual;
+            return discard_;
+        }
+        // Throw away any excess bytes if needed.
+        if (remaining() == 0) {
+            *actual = std::min(requested, sizeof(discard_));
+            return discard_;
+        }
+        return MemoryByteSink::GetBuffer(requested, actual);
+    }
+
+  private:
+    size_t ignore_start_;
+    char discard_[4096];
+};
+
+ssize_t CompressedSnapshotReader::ReadBlock(uint64_t chunk, IByteSink* sink, size_t start_offset,
+                                            const std::optional<uint64_t>& max_bytes) {
+    size_t bytes_to_read = block_size_;
+    if (max_bytes) {
+        bytes_to_read = *max_bytes;
+    }
+
+    // The offset is relative to the chunk; we should be reading no more than
+    // one chunk.
+    CHECK(start_offset + bytes_to_read <= block_size_);
+
+    const CowOperation* op = nullptr;
+    if (chunk < ops_.size()) {
+        op = ops_[chunk];
+    }
+
+    size_t actual;
+    void* buffer = sink->GetBuffer(bytes_to_read, &actual);
+    if (!buffer || actual < bytes_to_read) {
+        // This should never happen unless we calculated the read size wrong
+        // somewhere. MemoryByteSink always fulfills the entire requested
+        // region unless there's not enough buffer remaining.
+        LOG(ERROR) << "Asked for buffer of size " << bytes_to_read << ", got " << actual;
+        errno = EINVAL;
+        return -1;
+    }
+
+    if (!op || op->type == kCowCopyOp) {
+        borrowed_fd fd = GetSourceFd();
+        if (fd < 0) {
+            // GetSourceFd sets errno.
+            return -1;
+        }
+
+        if (op) {
+            chunk = op->source;
+        }
+
+        off64_t offset = (chunk * block_size_) + start_offset;
+        if (!android::base::ReadFullyAtOffset(fd, buffer, bytes_to_read, offset)) {
+            PLOG(ERROR) << "read " << *source_device_;
+            // ReadFullyAtOffset sets errno.
+            return -1;
+        }
+    } else if (op->type == kCowZeroOp) {
+        memset(buffer, 0, bytes_to_read);
+    } else if (op->type == kCowReplaceOp) {
+        PartialSink partial_sink(buffer, bytes_to_read, start_offset);
+        if (!cow_->ReadData(*op, &partial_sink)) {
+            LOG(ERROR) << "CompressedSnapshotReader failed to read replace op";
+            errno = EIO;
+            return -1;
+        }
+    } else {
+        LOG(ERROR) << "CompressedSnapshotReader unknown op type: " << op->type;
+        errno = EINVAL;
+        return -1;
+    }
+
+    // MemoryByteSink doesn't do anything in ReturnBuffer, so don't bother calling it.
+    return bytes_to_read;
+}
+
+off64_t CompressedSnapshotReader::Seek(off64_t offset, int whence) {
+    switch (whence) {
+        case SEEK_SET:
+            offset_ = offset;
+            break;
+        case SEEK_END:
+            offset_ = static_cast<off64_t>(block_device_size_) + offset;
+            break;
+        case SEEK_CUR:
+            offset_ += offset;
+            break;
+        default:
+            LOG(ERROR) << "Unrecognized seek whence: " << whence;
+            errno = EINVAL;
+            return -1;
+    }
+    return offset_;
+}
+
+uint64_t CompressedSnapshotReader::BlockDevSize() {
+    return block_device_size_;
+}
+
+bool CompressedSnapshotReader::Close() {
+    cow_ = nullptr;
+    source_fd_ = {};
+    return true;
+}
+
+bool CompressedSnapshotReader::IsSettingErrno() {
+    return true;
+}
+
+bool CompressedSnapshotReader::IsOpen() {
+    return cow_ != nullptr;
+}
+
+bool CompressedSnapshotReader::Flush() {
+    return true;
+}
+
 }  // namespace snapshot
 }  // namespace android
diff --git a/fs_mgr/libsnapshot/snapshot_reader.h b/fs_mgr/libsnapshot/snapshot_reader.h
index 1f2ffe2..a3e7e22 100644
--- a/fs_mgr/libsnapshot/snapshot_reader.h
+++ b/fs_mgr/libsnapshot/snapshot_reader.h
@@ -16,7 +16,11 @@
 
 #pragma once
 
+#include <optional>
+#include <vector>
+
 #include <android-base/file.h>
+#include <libsnapshot/cow_reader.h>
 #include <payload_consumer/file_descriptor.h>
 
 namespace android {
@@ -46,5 +50,36 @@
     android::base::unique_fd fd_;
 };
 
+class CompressedSnapshotReader : public ReadOnlyFileDescriptor {
+  public:
+    bool SetCow(std::unique_ptr<CowReader>&& cow);
+    void SetSourceDevice(const std::string& source_device);
+    void SetBlockDeviceSize(uint64_t block_device_size);
+
+    ssize_t Read(void* buf, size_t count) override;
+    off64_t Seek(off64_t offset, int whence) override;
+    uint64_t BlockDevSize() override;
+    bool Close() override;
+    bool IsSettingErrno() override;
+    bool IsOpen() override;
+    bool Flush() override;
+
+  private:
+    ssize_t ReadBlock(uint64_t chunk, IByteSink* sink, size_t start_offset,
+                      const std::optional<uint64_t>& max_bytes = {});
+    android::base::borrowed_fd GetSourceFd();
+
+    std::unique_ptr<CowReader> cow_;
+    std::unique_ptr<ICowOpIter> op_iter_;
+    uint32_t block_size_ = 0;
+
+    std::optional<std::string> source_device_;
+    android::base::unique_fd source_fd_;
+    uint64_t block_device_size_ = 0;
+    off64_t offset_ = 0;
+
+    std::vector<const CowOperation*> ops_;
+};
+
 }  // namespace snapshot
 }  // namespace android
diff --git a/fs_mgr/libsnapshot/snapshot_reader_test.cpp b/fs_mgr/libsnapshot/snapshot_reader_test.cpp
new file mode 100644
index 0000000..4202d22
--- /dev/null
+++ b/fs_mgr/libsnapshot/snapshot_reader_test.cpp
@@ -0,0 +1,166 @@
+// Copyright (C) 2018 The Android Open Source Project
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//      http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+#include <libsnapshot/snapshot.h>
+
+#include <unordered_set>
+
+#include <android-base/file.h>
+#include <gtest/gtest.h>
+#include <libsnapshot/cow_writer.h>
+#include <payload_consumer/file_descriptor.h>
+
+namespace android {
+namespace snapshot {
+
+using android::base::unique_fd;
+using chromeos_update_engine::FileDescriptor;
+
+static constexpr uint32_t kBlockSize = 4096;
+static constexpr size_t kBlockCount = 10;
+
+class OfflineSnapshotTest : public ::testing::Test {
+  protected:
+    virtual void SetUp() override {
+        base_ = std::make_unique<TemporaryFile>();
+        ASSERT_GE(base_->fd, 0) << strerror(errno);
+
+        cow_ = std::make_unique<TemporaryFile>();
+        ASSERT_GE(cow_->fd, 0) << strerror(errno);
+
+        WriteBaseDevice();
+    }
+
+    virtual void TearDown() override {
+        base_ = nullptr;
+        cow_ = nullptr;
+        base_blocks_ = {};
+    }
+
+    void WriteBaseDevice() {
+        unique_fd random(open("/dev/urandom", O_RDONLY));
+        ASSERT_GE(random, 0);
+
+        for (size_t i = 0; i < kBlockCount; i++) {
+            std::string block(kBlockSize, 0);
+            ASSERT_TRUE(android::base::ReadFully(random, block.data(), block.size()));
+            ASSERT_TRUE(android::base::WriteFully(base_->fd, block.data(), block.size()));
+            base_blocks_.emplace_back(std::move(block));
+        }
+        ASSERT_EQ(fsync(base_->fd), 0);
+    }
+
+    void WriteCow(ISnapshotWriter* writer) {
+        std::string new_block = MakeNewBlockString();
+
+        ASSERT_TRUE(writer->AddCopy(3, 0));
+        ASSERT_TRUE(writer->AddRawBlocks(5, new_block.data(), new_block.size()));
+        ASSERT_TRUE(writer->AddZeroBlocks(7, 2));
+        ASSERT_TRUE(writer->Finalize());
+    }
+
+    void TestBlockReads(ISnapshotWriter* writer) {
+        auto reader = writer->OpenReader();
+        ASSERT_NE(reader, nullptr);
+
+        // Test that unchanged blocks are not modified.
+        std::unordered_set<size_t> changed_blocks = {3, 5, 7, 8};
+        for (size_t i = 0; i < kBlockCount; i++) {
+            if (changed_blocks.count(i)) {
+                continue;
+            }
+
+            std::string block(kBlockSize, 0);
+            ASSERT_EQ(reader->Seek(i * kBlockSize, SEEK_SET), i * kBlockSize);
+            ASSERT_EQ(reader->Read(block.data(), block.size()), kBlockSize);
+            ASSERT_EQ(block, base_blocks_[i]);
+        }
+
+        // Test that we can read back our modified blocks.
+        std::string block(kBlockSize, 0);
+        ASSERT_EQ(reader->Seek(3 * kBlockSize, SEEK_SET), 3 * kBlockSize);
+        ASSERT_EQ(reader->Read(block.data(), block.size()), kBlockSize);
+        ASSERT_EQ(block, base_blocks_[0]);
+
+        ASSERT_EQ(reader->Seek(5 * kBlockSize, SEEK_SET), 5 * kBlockSize);
+        ASSERT_EQ(reader->Read(block.data(), block.size()), kBlockSize);
+        ASSERT_EQ(block, MakeNewBlockString());
+
+        std::string two_blocks(kBlockSize * 2, 0x7f);
+        std::string zeroes(kBlockSize * 2, 0);
+        ASSERT_EQ(reader->Seek(7 * kBlockSize, SEEK_SET), 7 * kBlockSize);
+        ASSERT_EQ(reader->Read(two_blocks.data(), two_blocks.size()), two_blocks.size());
+        ASSERT_EQ(two_blocks, zeroes);
+    }
+
+    void TestByteReads(ISnapshotWriter* writer) {
+        auto reader = writer->OpenReader();
+        ASSERT_NE(reader, nullptr);
+
+        std::string blob(kBlockSize * 3, 'x');
+
+        // Test that we can read in the middle of a block.
+        static constexpr size_t kOffset = 970;
+        off64_t offset = 3 * kBlockSize + kOffset;
+        ASSERT_EQ(reader->Seek(0, SEEK_SET), 0);
+        ASSERT_EQ(reader->Seek(offset, SEEK_CUR), offset);
+        ASSERT_EQ(reader->Read(blob.data(), blob.size()), blob.size());
+        ASSERT_EQ(blob.substr(0, 100), base_blocks_[0].substr(kOffset, 100));
+        ASSERT_EQ(blob.substr(kBlockSize - kOffset, kBlockSize), base_blocks_[4]);
+        ASSERT_EQ(blob.substr(kBlockSize * 2 - kOffset, 100), MakeNewBlockString().substr(0, 100));
+        ASSERT_EQ(blob.substr(blob.size() - kOffset), base_blocks_[6].substr(0, kOffset));
+
+        // Pull a random byte from the compressed block.
+        char value;
+        offset = 5 * kBlockSize + 1000;
+        ASSERT_EQ(reader->Seek(offset, SEEK_SET), offset);
+        ASSERT_EQ(reader->Read(&value, sizeof(value)), sizeof(value));
+        ASSERT_EQ(value, MakeNewBlockString()[1000]);
+    }
+
+    void TestReads(ISnapshotWriter* writer) {
+        ASSERT_NO_FATAL_FAILURE(TestBlockReads(writer));
+        ASSERT_NO_FATAL_FAILURE(TestByteReads(writer));
+    }
+
+    std::string MakeNewBlockString() {
+        std::string new_block = "This is a new block";
+        new_block.resize(kBlockSize / 2, '*');
+        new_block.resize(kBlockSize, '!');
+        return new_block;
+    }
+
+    std::unique_ptr<TemporaryFile> base_;
+    std::unique_ptr<TemporaryFile> cow_;
+    std::vector<std::string> base_blocks_;
+};
+
+TEST_F(OfflineSnapshotTest, CompressedSnapshot) {
+    CowOptions options;
+    options.compression = "gz";
+    options.max_blocks = {kBlockCount};
+
+    unique_fd cow_fd(dup(cow_->fd));
+    ASSERT_GE(cow_fd, 0);
+
+    auto writer = std::make_unique<CompressedSnapshotWriter>(options);
+    writer->SetSourceDevice(base_->path);
+    ASSERT_TRUE(writer->SetCowDevice(std::move(cow_fd)));
+    ASSERT_TRUE(writer->Initialize());
+    ASSERT_NO_FATAL_FAILURE(WriteCow(writer.get()));
+    ASSERT_NO_FATAL_FAILURE(TestReads(writer.get()));
+}
+
+}  // namespace snapshot
+}  // namespace android
diff --git a/fs_mgr/libsnapshot/snapshot_stub.cpp b/fs_mgr/libsnapshot/snapshot_stub.cpp
index 5be3c10..26b9129 100644
--- a/fs_mgr/libsnapshot/snapshot_stub.cpp
+++ b/fs_mgr/libsnapshot/snapshot_stub.cpp
@@ -136,4 +136,14 @@
     return nullptr;
 }
 
+bool SnapshotManagerStub::MapAllSnapshots(const std::chrono::milliseconds&) {
+    LOG(ERROR) << __FUNCTION__ << " should never be called.";
+    return false;
+}
+
+bool SnapshotManagerStub::UnmapAllSnapshots() {
+    LOG(ERROR) << __FUNCTION__ << " should never be called.";
+    return false;
+}
+
 }  // namespace android::snapshot
diff --git a/fs_mgr/libsnapshot/snapshot_test.cpp b/fs_mgr/libsnapshot/snapshot_test.cpp
index 7327629..9660357 100644
--- a/fs_mgr/libsnapshot/snapshot_test.cpp
+++ b/fs_mgr/libsnapshot/snapshot_test.cpp
@@ -907,6 +907,9 @@
         if (!result) {
             return AssertionFailure() << "Cannot open snapshot for writing: " << name;
         }
+        if (!result->Initialize()) {
+            return AssertionFailure() << "Cannot initialize snapshot for writing: " << name;
+        }
 
         if (writer) {
             *writer = std::move(result);
@@ -950,6 +953,9 @@
             if (!WriteRandomData(writer.get(), &hashes_[name])) {
                 return AssertionFailure() << "Unable to write random data to snapshot " << name;
             }
+            if (!writer->Finalize()) {
+                return AssertionFailure() << "Unable to finalize COW for " << name;
+            }
         } else {
             std::string path;
             auto res = MapUpdateSnapshot(name, &path);
diff --git a/fs_mgr/libsnapshot/snapshot_writer.cpp b/fs_mgr/libsnapshot/snapshot_writer.cpp
index 9b1ab97..85ed156 100644
--- a/fs_mgr/libsnapshot/snapshot_writer.cpp
+++ b/fs_mgr/libsnapshot/snapshot_writer.cpp
@@ -56,9 +56,9 @@
 bool CompressedSnapshotWriter::SetCowDevice(android::base::unique_fd&& cow_device) {
     cow_device_ = std::move(cow_device);
     cow_ = std::make_unique<CowWriter>(options_);
-
-    return cow_->Initialize(cow_device_);
+    return true;
 }
+
 bool CompressedSnapshotWriter::Finalize() {
     return cow_->Finalize();
 }
@@ -68,7 +68,31 @@
 }
 
 std::unique_ptr<FileDescriptor> CompressedSnapshotWriter::OpenReader() {
-    return nullptr;
+    unique_fd cow_fd(dup(cow_device_.get()));
+    if (cow_fd < 0) {
+        PLOG(ERROR) << "dup COW device";
+        return nullptr;
+    }
+
+    auto cow = std::make_unique<CowReader>();
+    if (!cow->Parse(std::move(cow_fd))) {
+        LOG(ERROR) << "Unable to read COW";
+        return nullptr;
+    }
+
+    auto reader = std::make_unique<CompressedSnapshotReader>();
+    if (!reader->SetCow(std::move(cow))) {
+        LOG(ERROR) << "Unable to initialize COW reader";
+        return nullptr;
+    }
+    if (source_device_) {
+        reader->SetSourceDevice(*source_device_);
+    }
+
+    const auto& cow_options = options();
+    reader->SetBlockDeviceSize(*cow_options.max_blocks * cow_options.block_size);
+
+    return reader;
 }
 
 bool CompressedSnapshotWriter::EmitCopy(uint64_t new_block, uint64_t old_block) {
@@ -88,6 +112,17 @@
     return cow_->AddLabel(label);
 }
 
+bool CompressedSnapshotWriter::Initialize() {
+    return cow_->Initialize(cow_device_, CowWriter::OpenMode::WRITE);
+}
+
+bool CompressedSnapshotWriter::InitializeAppend(std::optional<uint64_t> label) {
+    if (label) {
+        return cow_->InitializeAppend(cow_device_, *label);
+    }
+    return cow_->Initialize(cow_device_, CowWriter::OpenMode::APPEND);
+}
+
 OnlineKernelSnapshotWriter::OnlineKernelSnapshotWriter(const CowOptions& options)
     : ISnapshotWriter(options) {}
 
diff --git a/fs_mgr/libsnapshot/snapuserd.cpp b/fs_mgr/libsnapshot/snapuserd.cpp
index 6e772ad..40f26d6 100644
--- a/fs_mgr/libsnapshot/snapuserd.cpp
+++ b/fs_mgr/libsnapshot/snapuserd.cpp
@@ -379,7 +379,11 @@
         struct disk_exception* de =
                 reinterpret_cast<struct disk_exception*>((char*)de_ptr.get() + offset);
 
-        if (cow_op->type == kCowFooterOp || cow_op->type == kCowLabelOp) continue;
+        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: " << cow_op->type;
diff --git a/fs_mgr/libsnapshot/snapuserd_client.cpp b/fs_mgr/libsnapshot/snapuserd_client.cpp
index 532e585..35bb29b 100644
--- a/fs_mgr/libsnapshot/snapuserd_client.cpp
+++ b/fs_mgr/libsnapshot/snapuserd_client.cpp
@@ -123,34 +123,32 @@
     return true;
 }
 
-std::string SnapuserdClient::Receivemsg() {
-    int ret;
-    struct timeval tv;
-    fd_set set;
-    char msg[PACKET_SIZE];
-    std::string msgStr("fail");
-
-    tv.tv_sec = 2;
-    tv.tv_usec = 0;
-    FD_ZERO(&set);
-    FD_SET(sockfd_, &set);
-    ret = select(sockfd_ + 1, &set, NULL, NULL, &tv);
-    if (ret == -1) {  // select failed
-        LOG(ERROR) << "Snapuserd:client: Select call failed";
-    } else if (ret == 0) {  // timeout
-        LOG(ERROR) << "Snapuserd:client: Select call timeout";
-    } else {
-        ret = TEMP_FAILURE_RETRY(recv(sockfd_, msg, PACKET_SIZE, 0));
-        if (ret < 0) {
-            PLOG(ERROR) << "Snapuserd:client: recv failed";
-        } else if (ret == 0) {
-            LOG(DEBUG) << "Snapuserd:client disconnected";
-        } else {
-            msgStr.clear();
-            msgStr = msg;
-        }
+bool SnapuserdClient::WaitForDeviceDelete(const std::string& control_device) {
+    std::string msg = "delete," + control_device;
+    if (!Sendmsg(msg)) {
+        LOG(ERROR) << "Failed to send message " << msg << " to snapuserd";
+        return false;
     }
-    return msgStr;
+    std::string response = Receivemsg();
+    if (response != "success") {
+        LOG(ERROR) << "Failed waiting to delete device " << control_device;
+        return false;
+    }
+    return true;
+}
+
+std::string SnapuserdClient::Receivemsg() {
+    char msg[PACKET_SIZE];
+    ssize_t ret = TEMP_FAILURE_RETRY(recv(sockfd_, msg, sizeof(msg), 0));
+    if (ret < 0) {
+        PLOG(ERROR) << "Snapuserd:client: recv failed";
+        return {};
+    }
+    if (ret == 0) {
+        LOG(DEBUG) << "Snapuserd:client disconnected";
+        return {};
+    }
+    return std::string(msg, ret);
 }
 
 bool SnapuserdClient::StopSnapuserd() {
diff --git a/fs_mgr/libsnapshot/snapuserd_server.cpp b/fs_mgr/libsnapshot/snapuserd_server.cpp
index 48a3b2a..6b8cdd9 100644
--- a/fs_mgr/libsnapshot/snapuserd_server.cpp
+++ b/fs_mgr/libsnapshot/snapuserd_server.cpp
@@ -36,6 +36,7 @@
     if (input == "start") return DaemonOperations::START;
     if (input == "stop") return DaemonOperations::STOP;
     if (input == "query") return DaemonOperations::QUERY;
+    if (input == "delete") return DaemonOperations::DELETE;
 
     return DaemonOperations::INVALID;
 }
@@ -68,33 +69,25 @@
     }
 }
 
-// new thread
-void SnapuserdServer::ThreadStart(std::string cow_device, std::string backing_device,
-                                  std::string control_device) {
-    Snapuserd snapd(cow_device, backing_device, control_device);
-    if (!snapd.Init()) {
-        LOG(ERROR) << "Snapuserd: Init failed";
-        return;
-    }
-
-    while (StopRequested() == false) {
-        int ret = snapd.Run();
-
-        if (ret < 0) {
-            LOG(ERROR) << "Snapuserd: Thread terminating as control device is de-registered";
-            break;
-        }
-    }
-}
-
 void SnapuserdServer::ShutdownThreads() {
     StopThreads();
 
-    for (auto& client : dm_users_) {
-        auto& th = client->GetThreadHandler();
-
-        if (th->joinable()) th->join();
+    // Acquire the thread list within the lock.
+    std::vector<std::unique_ptr<DmUserHandler>> dm_users;
+    {
+        std::lock_guard<std::mutex> guard(lock_);
+        dm_users = std::move(dm_users_);
     }
+
+    for (auto& client : dm_users) {
+        auto& th = client->thread();
+
+        if (th.joinable()) th.join();
+    }
+}
+
+const std::string& DmUserHandler::GetControlDevice() const {
+    return snapuserd_->GetControlDevicePath();
 }
 
 bool SnapuserdServer::Sendmsg(android::base::borrowed_fd fd, const std::string& msg) {
@@ -135,10 +128,25 @@
             // start,<cow_device_path>,<source_device_path>,<control_device>
             //
             // Start the new thread which binds to dm-user misc device
-            auto handler = std::make_unique<DmUserHandler>();
-            handler->SetThreadHandler(
-                    std::bind(&SnapuserdServer::ThreadStart, this, out[1], out[2], out[3]));
-            dm_users_.push_back(std::move(handler));
+            if (out.size() != 4) {
+                LOG(ERROR) << "Malformed start message, " << out.size() << " parts";
+                return Sendmsg(fd, "fail");
+            }
+
+            auto snapuserd = std::make_unique<Snapuserd>(out[1], out[2], out[3]);
+            if (!snapuserd->Init()) {
+                LOG(ERROR) << "Failed to initialize Snapuserd";
+                return Sendmsg(fd, "fail");
+            }
+
+            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()));
+                dm_users_.push_back(std::move(handler));
+            }
             return Sendmsg(fd, "success");
         }
         case DaemonOperations::STOP: {
@@ -162,6 +170,18 @@
             // be ready to receive control message.
             return Sendmsg(fd, GetDaemonStatus());
         }
+        case DaemonOperations::DELETE: {
+            // Message format:
+            // delete,<cow_device_path>
+            if (out.size() != 2) {
+                LOG(ERROR) << "Malformed delete message, " << out.size() << " parts";
+                return Sendmsg(fd, "fail");
+            }
+            if (!WaitForDelete(out[1])) {
+                return Sendmsg(fd, "fail");
+            }
+            return Sendmsg(fd, "success");
+        }
         default: {
             LOG(ERROR) << "Received unknown message type from client";
             Sendmsg(fd, "fail");
@@ -170,6 +190,23 @@
     }
 }
 
+void SnapuserdServer::RunThread(DmUserHandler* handler) {
+    while (!StopRequested()) {
+        if (handler->snapuserd()->Run() < 0) {
+            LOG(INFO) << "Snapuserd: Thread terminating as control device is de-registered";
+            break;
+        }
+    }
+
+    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();
+    }
+}
+
 bool SnapuserdServer::Start(const std::string& socketname) {
     sockfd_.reset(android_get_control_socket(socketname.c_str()));
     if (sockfd_ >= 0) {
@@ -260,5 +297,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++;
+        }
+    }
+    return client;
+}
+
+bool SnapuserdServer::WaitForDelete(const std::string& control_device) {
+    auto client = RemoveHandler(control_device);
+
+    // Client already deleted.
+    if (!client) {
+        return true;
+    }
+
+    auto& th = client->thread();
+    if (th.joinable()) {
+        th.join();
+    }
+    return true;
+}
+
 }  // namespace snapshot
 }  // namespace android
diff --git a/init/Android.mk b/init/Android.mk
index 895f50f..ac31ef1 100644
--- a/init/Android.mk
+++ b/init/Android.mk
@@ -78,7 +78,6 @@
 LOCAL_POST_INSTALL_CMD := mkdir -p \
     $(TARGET_RAMDISK_OUT)/debug_ramdisk \
     $(TARGET_RAMDISK_OUT)/dev \
-    $(TARGET_RAMDISK_OUT)/first_stage_ramdisk \
     $(TARGET_RAMDISK_OUT)/mnt \
     $(TARGET_RAMDISK_OUT)/proc \
     $(TARGET_RAMDISK_OUT)/second_stage_resources \
diff --git a/init/README.md b/init/README.md
index 6439393..ab6a885 100644
--- a/init/README.md
+++ b/init/README.md
@@ -172,9 +172,12 @@
   This option connects stdin, stdout, and stderr to the console. It is mutually exclusive with the
   stdio_to_kmsg option, which only connects stdout and stderr to kmsg.
 
-`critical`
+`critical [window=<fatal crash window mins>] [target=<fatal reboot target>]`
 > This is a device-critical service. If it exits more than four times in
-  four minutes or before boot completes, the device will reboot into bootloader.
+  _fatal crash window mins_ minutes or before boot completes, the device
+  will reboot into _fatal reboot target_.
+  The default value of _fatal crash window mins_ is 4, and default value
+  of _fatal reboot target_ is 'bootloader'.
 
 `disabled`
 > This service will not automatically start with its class.
diff --git a/init/host_init_stubs.h b/init/host_init_stubs.h
index caa8f8d..2a8bf6c 100644
--- a/init/host_init_stubs.h
+++ b/init/host_init_stubs.h
@@ -20,6 +20,7 @@
 #include <sys/socket.h>
 #include <sys/types.h>
 
+#include <optional>
 #include <string>
 
 #include <android-base/properties.h>
@@ -41,7 +42,7 @@
 }
 
 // reboot_utils.h
-inline void SetFatalRebootTarget() {}
+inline void SetFatalRebootTarget(const std::optional<std::string>& = std::nullopt) {}
 inline void __attribute__((noreturn)) InitFatalReboot(int signal_number) {
     abort();
 }
diff --git a/init/reboot_utils.cpp b/init/reboot_utils.cpp
index 76460a5..98f6857 100644
--- a/init/reboot_utils.cpp
+++ b/init/reboot_utils.cpp
@@ -19,6 +19,7 @@
 #include <sys/syscall.h>
 #include <unistd.h>
 
+#include <optional>
 #include <string>
 
 #include <android-base/file.h>
@@ -37,7 +38,7 @@
 static std::string init_fatal_reboot_target = "bootloader";
 static bool init_fatal_panic = false;
 
-void SetFatalRebootTarget() {
+void SetFatalRebootTarget(const std::optional<std::string>& reboot_target) {
     std::string cmdline;
     android::base::ReadFileToString("/proc/cmdline", &cmdline);
     cmdline = android::base::Trim(cmdline);
@@ -45,6 +46,11 @@
     const char kInitFatalPanicString[] = "androidboot.init_fatal_panic=true";
     init_fatal_panic = cmdline.find(kInitFatalPanicString) != std::string::npos;
 
+    if (reboot_target) {
+        init_fatal_reboot_target = *reboot_target;
+        return;
+    }
+
     const char kRebootTargetString[] = "androidboot.init_fatal_reboot_target=";
     auto start_pos = cmdline.find(kRebootTargetString);
     if (start_pos == std::string::npos) {
diff --git a/init/reboot_utils.h b/init/reboot_utils.h
index 05bb9ae..a0023b9 100644
--- a/init/reboot_utils.h
+++ b/init/reboot_utils.h
@@ -16,6 +16,7 @@
 
 #pragma once
 
+#include <optional>
 #include <string>
 
 #define PROC_SYSRQ "/proc/sysrq-trigger"
@@ -23,7 +24,7 @@
 namespace android {
 namespace init {
 
-void SetFatalRebootTarget();
+void SetFatalRebootTarget(const std::optional<std::string>& reboot_target = std::nullopt);
 // Determines whether the system is capable of rebooting. This is conservative,
 // so if any of the attempts to determine this fail, it will still return true.
 bool IsRebootCapable();
diff --git a/init/service.cpp b/init/service.cpp
index 68365b3..ecc86d9 100644
--- a/init/service.cpp
+++ b/init/service.cpp
@@ -45,6 +45,7 @@
 #include <android/api-level.h>
 
 #include "mount_namespace.h"
+#include "reboot_utils.h"
 #include "selinux.h"
 #else
 #include "host_init_stubs.h"
@@ -312,20 +313,24 @@
 #endif
     const bool is_process_updatable = !pre_apexd_ && is_apex_updatable;
 
-    // If we crash > 4 times in 4 minutes or before boot_completed,
+    // If we crash > 4 times in 'fatal_crash_window_' minutes or before boot_completed,
     // reboot into bootloader or set crashing property
     boot_clock::time_point now = boot_clock::now();
     if (((flags_ & SVC_CRITICAL) || is_process_updatable) && !(flags_ & SVC_RESTART)) {
         bool boot_completed = android::base::GetBoolProperty("sys.boot_completed", false);
-        if (now < time_crashed_ + 4min || !boot_completed) {
+        if (now < time_crashed_ + fatal_crash_window_ || !boot_completed) {
             if (++crash_count_ > 4) {
+                auto exit_reason = boot_completed ?
+                    "in " + std::to_string(fatal_crash_window_.count()) + " minutes" :
+                    "before boot completed";
                 if (flags_ & SVC_CRITICAL) {
-                    // Aborts into bootloader
+                    // Aborts into `fatal_reboot_target_'.
+                    SetFatalRebootTarget(fatal_reboot_target_);
                     LOG(FATAL) << "critical process '" << name_ << "' exited 4 times "
-                               << (boot_completed ? "in 4 minutes" : "before boot completed");
+                               << exit_reason;
                 } else {
                     LOG(ERROR) << "updatable process '" << name_ << "' exited 4 times "
-                               << (boot_completed ? "in 4 minutes" : "before boot completed");
+                               << exit_reason;
                     // Notifies update_verifier and apexd
                     SetProperty("sys.init.updatable_crashing_process_name", name_);
                     SetProperty("sys.init.updatable_crashing", "1");
diff --git a/init/service.h b/init/service.h
index 34ed5ef..bc5c90f 100644
--- a/init/service.h
+++ b/init/service.h
@@ -155,6 +155,8 @@
     android::base::boot_clock::time_point time_started_;  // time of last start
     android::base::boot_clock::time_point time_crashed_;  // first crash within inspection window
     int crash_count_;                     // number of times crashed within window
+    std::chrono::minutes fatal_crash_window_ = 4min;  // fatal() when more than 4 crashes in it
+    std::optional<std::string> fatal_reboot_target_;  // reboot target of fatal handler
 
     std::optional<CapSet> capabilities_;
     ProcessAttributes proc_attr_;
diff --git a/init/service_parser.cpp b/init/service_parser.cpp
index bdac077..97621da 100644
--- a/init/service_parser.cpp
+++ b/init/service_parser.cpp
@@ -93,6 +93,39 @@
 }
 
 Result<void> ServiceParser::ParseCritical(std::vector<std::string>&& args) {
+    std::optional<std::string> fatal_reboot_target;
+    std::optional<std::chrono::minutes> fatal_crash_window;
+
+    for (auto it = args.begin() + 1; it != args.end(); ++it) {
+        auto arg = android::base::Split(*it, "=");
+        if (arg.size() != 2) {
+            return Error() << "critical: Argument '" << *it << "' is not supported";
+        } else if (arg[0] == "target") {
+            fatal_reboot_target = arg[1];
+        } else if (arg[0] == "window") {
+            int minutes;
+            auto window = ExpandProps(arg[1]);
+            if (!window.ok()) {
+                return Error() << "critical: Could not expand argument ': " << arg[1];
+            }
+            if (*window == "off") {
+                return {};
+            }
+            if (!ParseInt(*window, &minutes, 0)) {
+                return Error() << "critical: 'fatal_crash_window' must be an integer > 0";
+            }
+            fatal_crash_window = std::chrono::minutes(minutes);
+        } else {
+            return Error() << "critical: Argument '" << *it << "' is not supported";
+        }
+    }
+
+    if (fatal_reboot_target) {
+        service_->fatal_reboot_target_ = *fatal_reboot_target;
+    }
+    if (fatal_crash_window) {
+        service_->fatal_crash_window_ = *fatal_crash_window;
+    }
     service_->flags_ |= SVC_CRITICAL;
     return {};
 }
@@ -506,7 +539,7 @@
         {"capabilities",            {0,     kMax, &ServiceParser::ParseCapabilities}},
         {"class",                   {1,     kMax, &ServiceParser::ParseClass}},
         {"console",                 {0,     1,    &ServiceParser::ParseConsole}},
-        {"critical",                {0,     0,    &ServiceParser::ParseCritical}},
+        {"critical",                {0,     2,    &ServiceParser::ParseCritical}},
         {"disabled",                {0,     0,    &ServiceParser::ParseDisabled}},
         {"enter_namespace",         {2,     2,    &ServiceParser::ParseEnterNamespace}},
         {"file",                    {2,     2,    &ServiceParser::ParseFile}},
diff --git a/libutils/Unicode.cpp b/libutils/Unicode.cpp
index b6e457b..843a81a 100644
--- a/libutils/Unicode.cpp
+++ b/libutils/Unicode.cpp
@@ -359,49 +359,6 @@
 // UTF-8
 // --------------------------------------------------------------------------
 
-ssize_t utf8_length(const char *src)
-{
-    const char *cur = src;
-    size_t ret = 0;
-    while (*cur != '\0') {
-        const char first_char = *cur++;
-        if ((first_char & 0x80) == 0) { // ASCII
-            ret += 1;
-            continue;
-        }
-        // (UTF-8's character must not be like 10xxxxxx,
-        //  but 110xxxxx, 1110xxxx, ... or 1111110x)
-        if ((first_char & 0x40) == 0) {
-            return -1;
-        }
-
-        int32_t mask, to_ignore_mask;
-        size_t num_to_read = 0;
-        char32_t utf32 = 0;
-        for (num_to_read = 1, mask = 0x40, to_ignore_mask = 0x80;
-             num_to_read < 5 && (first_char & mask);
-             num_to_read++, to_ignore_mask |= mask, mask >>= 1) {
-            if ((*cur & 0xC0) != 0x80) { // must be 10xxxxxx
-                return -1;
-            }
-            // 0x3F == 00111111
-            utf32 = (utf32 << 6) + (*cur++ & 0x3F);
-        }
-        // "first_char" must be (110xxxxx - 11110xxx)
-        if (num_to_read == 5) {
-            return -1;
-        }
-        to_ignore_mask |= mask;
-        utf32 |= ((~to_ignore_mask) & first_char) << (6 * (num_to_read - 1));
-        if (utf32 > kUnicodeMaxCodepoint) {
-            return -1;
-        }
-
-        ret += num_to_read;
-    }
-    return ret;
-}
-
 ssize_t utf16_to_utf8_length(const char16_t *src, size_t src_len)
 {
     if (src == nullptr || src_len == 0) {
diff --git a/libutils/include/utils/Unicode.h b/libutils/include/utils/Unicode.h
index fc6712d..0087383 100644
--- a/libutils/include/utils/Unicode.h
+++ b/libutils/include/utils/Unicode.h
@@ -111,24 +111,6 @@
 void utf16_to_utf8(const char16_t* src, size_t src_len, char* dst, size_t dst_len);
 
 /**
- * Returns the length of "src" when "src" is valid UTF-8 string.
- * Returns 0 if src is NULL or 0-length string. Returns -1 when the source
- * is an invalid string.
- *
- * This function should be used to determine whether "src" is valid UTF-8
- * characters with valid unicode codepoints. "src" must be nul-terminated.
- *
- * If you are going to use other utf8_to_... functions defined in this header
- * with string which may not be valid UTF-8 with valid codepoint (form 0 to
- * 0x10FFFF), you should use this function before calling others, since the
- * other functions do not check whether the string is valid UTF-8 or not.
- *
- * If you do not care whether "src" is valid UTF-8 or not, you should use
- * strlen() as usual, which should be much faster.
- */
-ssize_t utf8_length(const char *src);
-
-/**
  * Returns the UTF-16 length of UTF-8 string "src". Returns -1 in case
  * it's invalid utf8. No buffer over-read occurs because of bound checks. Using overreadIsFatal you
  * can ask to log a message and fail in case the invalid utf8 could have caused an override if no
diff --git a/rootdir/init.zygote32.rc b/rootdir/init.zygote32.rc
index e827cf5..9469a48 100644
--- a/rootdir/init.zygote32.rc
+++ b/rootdir/init.zygote32.rc
@@ -13,3 +13,4 @@
     onrestart restart netd
     onrestart restart wificond
     writepid /dev/cpuset/foreground/tasks
+    critical window=${zygote.critical_window.minute:-off} target=zygote-fatal
diff --git a/rootdir/init.zygote64.rc b/rootdir/init.zygote64.rc
index adc7031..98dc088 100644
--- a/rootdir/init.zygote64.rc
+++ b/rootdir/init.zygote64.rc
@@ -13,3 +13,4 @@
     onrestart restart netd
     onrestart restart wificond
     writepid /dev/cpuset/foreground/tasks
+    critical window=${zygote.critical_window.minute:-off} target=zygote-fatal
diff --git a/rootdir/init.zygote64_32.rc b/rootdir/init.zygote64_32.rc
index fb9e99b..3eee180 100644
--- a/rootdir/init.zygote64_32.rc
+++ b/rootdir/init.zygote64_32.rc
@@ -13,6 +13,7 @@
     onrestart restart netd
     onrestart restart wificond
     task_profiles ProcessCapacityHigh MaxPerformance
+    critical window=${zygote.critical_window.minute:-off} target=zygote-fatal
 
 service zygote_secondary /system/bin/app_process32 -Xzygote /system/bin --zygote --socket-name=zygote_secondary --enable-lazy-preload
     class main