diff --git a/bootstat/OWNERS b/bootstat/OWNERS
index 50b2097..f66b309 100644
--- a/bootstat/OWNERS
+++ b/bootstat/OWNERS
@@ -1,2 +1,2 @@
 jhawkins@google.com
-salyzyn@google.com
+dvander@google.com
diff --git a/fs_mgr/libsnapshot/include/libsnapshot/snapshot.h b/fs_mgr/libsnapshot/include/libsnapshot/snapshot.h
index d2ffaa7..ff7a727 100644
--- a/fs_mgr/libsnapshot/include/libsnapshot/snapshot.h
+++ b/fs_mgr/libsnapshot/include/libsnapshot/snapshot.h
@@ -379,6 +379,7 @@
     FRIEND_TEST(SnapshotUpdateTest, MergeCannotRemoveCow);
     FRIEND_TEST(SnapshotUpdateTest, MergeInRecovery);
     FRIEND_TEST(SnapshotUpdateTest, SnapshotStatusFileWithoutCow);
+    FRIEND_TEST(SnapshotUpdateTest, SpaceSwapUpdate);
     friend class SnapshotTest;
     friend class SnapshotUpdateTest;
     friend class FlashAfterUpdateTest;
@@ -717,6 +718,8 @@
     bool PerformInitTransition(InitTransition transition,
                                std::vector<std::string>* snapuserd_argv = nullptr);
 
+    SnapuserdClient* snapuserd_client() const { return snapuserd_client_.get(); }
+
     std::string gsid_dir_;
     std::string metadata_dir_;
     std::unique_ptr<IDeviceInfo> device_;
diff --git a/fs_mgr/libsnapshot/include_test/libsnapshot/test_helpers.h b/fs_mgr/libsnapshot/include_test/libsnapshot/test_helpers.h
index ff0047e..b038527 100644
--- a/fs_mgr/libsnapshot/include_test/libsnapshot/test_helpers.h
+++ b/fs_mgr/libsnapshot/include_test/libsnapshot/test_helpers.h
@@ -146,6 +146,7 @@
 bool WriteRandomData(const std::string& path, std::optional<size_t> expect_size = std::nullopt,
                      std::string* hash = nullptr);
 bool WriteRandomData(ICowWriter* writer, std::string* hash = nullptr);
+std::string HashSnapshot(ISnapshotWriter* writer);
 
 std::optional<std::string> GetHash(const std::string& path);
 
diff --git a/fs_mgr/libsnapshot/snapshot_test.cpp b/fs_mgr/libsnapshot/snapshot_test.cpp
index 0b8a03a..95e7d89 100644
--- a/fs_mgr/libsnapshot/snapshot_test.cpp
+++ b/fs_mgr/libsnapshot/snapshot_test.cpp
@@ -225,7 +225,7 @@
     }
 
     AssertionResult MapUpdateSnapshot(const std::string& name,
-                                      std::unique_ptr<ICowWriter>* writer) {
+                                      std::unique_ptr<ISnapshotWriter>* writer) {
         TestPartitionOpener opener(fake_super);
         CreateLogicalPartitionParams params{
                 .block_device = fake_super,
@@ -279,6 +279,7 @@
             return AssertionFailure() << "Cannot unmap image " << snapshot << "-cow-img";
         }
         if (!(res = DeleteDevice(snapshot + "-base"))) return res;
+        if (!(res = DeleteDevice(snapshot + "-src"))) return res;
         return AssertionSuccess();
     }
 
@@ -319,7 +320,7 @@
 
     // Prepare A/B slot for a partition named "test_partition".
     AssertionResult PrepareOneSnapshot(uint64_t device_size,
-                                       std::unique_ptr<ICowWriter>* writer = nullptr) {
+                                       std::unique_ptr<ISnapshotWriter>* writer = nullptr) {
         lock_ = nullptr;
 
         DeltaArchiveManifest manifest;
@@ -511,7 +512,7 @@
 
     static const uint64_t kDeviceSize = 1024 * 1024;
 
-    std::unique_ptr<ICowWriter> writer;
+    std::unique_ptr<ISnapshotWriter> writer;
     ASSERT_TRUE(PrepareOneSnapshot(kDeviceSize, &writer));
 
     // Release the lock.
@@ -827,11 +828,14 @@
 
         opener_ = std::make_unique<TestPartitionOpener>(fake_super);
 
+        auto dynamic_partition_metadata = manifest_.mutable_dynamic_partition_metadata();
+        dynamic_partition_metadata->set_vabc_enabled(IsCompressionEnabled());
+
         // Create a fake update package metadata.
         // Not using full name "system", "vendor", "product" because these names collide with the
         // mapped partitions on the running device.
         // Each test modifies manifest_ slightly to indicate changes to the partition layout.
-        group_ = manifest_.mutable_dynamic_partition_metadata()->add_groups();
+        group_ = dynamic_partition_metadata->add_groups();
         group_->set_name("group");
         group_->set_size(kGroupSize);
         group_->add_partition_names("sys");
@@ -945,7 +949,7 @@
 
     AssertionResult MapOneUpdateSnapshot(const std::string& name) {
         if (IsCompressionEnabled()) {
-            std::unique_ptr<ICowWriter> writer;
+            std::unique_ptr<ISnapshotWriter> writer;
             return MapUpdateSnapshot(name, &writer);
         } else {
             std::string path;
@@ -955,7 +959,7 @@
 
     AssertionResult WriteSnapshotAndHash(const std::string& name) {
         if (IsCompressionEnabled()) {
-            std::unique_ptr<ICowWriter> writer;
+            std::unique_ptr<ISnapshotWriter> writer;
             auto res = MapUpdateSnapshot(name, &writer);
             if (!res) {
                 return res;
@@ -984,6 +988,42 @@
                                   << ", hash: " << hashes_[name];
     }
 
+    // Generate a snapshot that moves all the upper blocks down to the start.
+    // It doesn't really matter the order, we just want copies that reference
+    // blocks that won't exist if the partition shrinks.
+    AssertionResult ShiftAllSnapshotBlocks(const std::string& name, uint64_t old_size) {
+        std::unique_ptr<ISnapshotWriter> writer;
+        if (auto res = MapUpdateSnapshot(name, &writer); !res) {
+            return res;
+        }
+        if (!writer->options().max_blocks || !*writer->options().max_blocks) {
+            return AssertionFailure() << "No max blocks set for " << name << " writer";
+        }
+
+        uint64_t src_block = (old_size / writer->options().block_size) - 1;
+        uint64_t dst_block = 0;
+        uint64_t max_blocks = *writer->options().max_blocks;
+        while (dst_block < max_blocks && dst_block < src_block) {
+            if (!writer->AddCopy(dst_block, src_block)) {
+                return AssertionFailure() << "Unable to add copy for " << name << " for blocks "
+                                          << src_block << ", " << dst_block;
+            }
+            dst_block++;
+            src_block--;
+        }
+        if (!writer->Finalize()) {
+            return AssertionFailure() << "Unable to finalize writer for " << name;
+        }
+
+        auto hash = HashSnapshot(writer.get());
+        if (hash.empty()) {
+            return AssertionFailure() << "Unable to hash snapshot writer for " << name;
+        }
+        hashes_[name] = hash;
+
+        return AssertionSuccess();
+    }
+
     AssertionResult MapUpdateSnapshots(const std::vector<std::string>& names = {"sys_b", "vnd_b",
                                                                                 "prd_b"}) {
         for (const auto& name : names) {
@@ -1092,8 +1132,132 @@
     // Initiate the merge and wait for it to be completed.
     ASSERT_TRUE(init->InitiateMerge());
     ASSERT_EQ(init->IsSnapuserdRequired(), IsCompressionEnabled());
+    {
+        // We should have started in SECOND_PHASE since nothing shrinks.
+        ASSERT_TRUE(AcquireLock());
+        auto local_lock = std::move(lock_);
+        auto status = init->ReadSnapshotUpdateStatus(local_lock.get());
+        ASSERT_EQ(status.merge_phase(), MergePhase::SECOND_PHASE);
+    }
     ASSERT_EQ(UpdateState::MergeCompleted, init->ProcessUpdateState());
 
+    // Make sure the second phase ran and deleted snapshots.
+    {
+        ASSERT_TRUE(AcquireLock());
+        auto local_lock = std::move(lock_);
+        std::vector<std::string> snapshots;
+        ASSERT_TRUE(init->ListSnapshots(local_lock.get(), &snapshots));
+        ASSERT_TRUE(snapshots.empty());
+    }
+
+    // Check that the target partitions have the same content after the merge.
+    for (const auto& name : {"sys_b", "vnd_b", "prd_b"}) {
+        ASSERT_TRUE(IsPartitionUnchanged(name))
+                << "Content of " << name << " changes after the merge";
+    }
+}
+
+// Test that shrinking and growing partitions at the same time is handled
+// correctly in VABC.
+TEST_F(SnapshotUpdateTest, SpaceSwapUpdate) {
+    // OTA client blindly unmaps all partitions that are possibly mapped.
+    for (const auto& name : {"sys_b", "vnd_b", "prd_b"}) {
+        ASSERT_TRUE(sm->UnmapUpdateSnapshot(name));
+    }
+
+    auto old_sys_size = GetSize(sys_);
+    auto old_prd_size = GetSize(prd_);
+
+    // Grow |sys| but shrink |prd|.
+    SetSize(sys_, old_sys_size * 2);
+    sys_->set_estimate_cow_size(8_MiB);
+    SetSize(prd_, old_prd_size / 2);
+    prd_->set_estimate_cow_size(1_MiB);
+
+    AddOperationForPartitions();
+
+    ASSERT_TRUE(sm->BeginUpdate());
+    ASSERT_TRUE(sm->CreateUpdateSnapshots(manifest_));
+
+    // Check that the old partition sizes were saved correctly.
+    {
+        ASSERT_TRUE(AcquireLock());
+        auto local_lock = std::move(lock_);
+
+        SnapshotStatus status;
+        ASSERT_TRUE(sm->ReadSnapshotStatus(local_lock.get(), "prd_b", &status));
+        ASSERT_EQ(status.old_partition_size(), 3145728);
+        ASSERT_TRUE(sm->ReadSnapshotStatus(local_lock.get(), "sys_b", &status));
+        ASSERT_EQ(status.old_partition_size(), 3145728);
+    }
+
+    ASSERT_TRUE(WriteSnapshotAndHash("sys_b"));
+    ASSERT_TRUE(WriteSnapshotAndHash("vnd_b"));
+    ASSERT_TRUE(ShiftAllSnapshotBlocks("prd_b", old_prd_size));
+
+    sync();
+
+    // Assert that source partitions aren't affected.
+    for (const auto& name : {"sys_a", "vnd_a", "prd_a"}) {
+        ASSERT_TRUE(IsPartitionUnchanged(name));
+    }
+
+    ASSERT_TRUE(sm->FinishedSnapshotWrites(false));
+
+    // Simulate shutting down the device.
+    ASSERT_TRUE(UnmapAll());
+
+    // After reboot, init does first stage mount.
+    auto init = NewManagerForFirstStageMount("_b");
+    ASSERT_NE(init, nullptr);
+    ASSERT_TRUE(init->NeedSnapshotsInFirstStageMount());
+    ASSERT_TRUE(init->CreateLogicalAndSnapshotPartitions("super", snapshot_timeout_));
+
+    auto indicator = sm->GetRollbackIndicatorPath();
+    ASSERT_NE(access(indicator.c_str(), R_OK), 0);
+
+    // Check that the target partitions have the same content.
+    for (const auto& name : {"sys_b", "vnd_b", "prd_b"}) {
+        ASSERT_TRUE(IsPartitionUnchanged(name));
+    }
+
+    // Initiate the merge and wait for it to be completed.
+    ASSERT_TRUE(init->InitiateMerge());
+    ASSERT_EQ(init->IsSnapuserdRequired(), IsCompressionEnabled());
+    {
+        // Check that the merge phase is FIRST_PHASE until at least one call
+        // to ProcessUpdateState() occurs.
+        ASSERT_TRUE(AcquireLock());
+        auto local_lock = std::move(lock_);
+        auto status = init->ReadSnapshotUpdateStatus(local_lock.get());
+        ASSERT_EQ(status.merge_phase(), MergePhase::FIRST_PHASE);
+    }
+
+    // Simulate shutting down the device and creating partitions again.
+    ASSERT_TRUE(UnmapAll());
+    ASSERT_TRUE(init->CreateLogicalAndSnapshotPartitions("super", snapshot_timeout_));
+
+    // Check that we used the correct types after rebooting mid-merge.
+    DeviceMapper::TargetInfo target;
+    ASSERT_TRUE(init->IsSnapshotDevice("prd_b", &target));
+    ASSERT_EQ(DeviceMapper::GetTargetType(target.spec), "snapshot-merge");
+    ASSERT_TRUE(init->IsSnapshotDevice("sys_b", &target));
+    ASSERT_EQ(DeviceMapper::GetTargetType(target.spec), "snapshot");
+    ASSERT_TRUE(init->IsSnapshotDevice("vnd_b", &target));
+    ASSERT_EQ(DeviceMapper::GetTargetType(target.spec), "snapshot");
+
+    // Complete the merge.
+    ASSERT_EQ(UpdateState::MergeCompleted, init->ProcessUpdateState());
+
+    // Make sure the second phase ran and deleted snapshots.
+    {
+        ASSERT_TRUE(AcquireLock());
+        auto local_lock = std::move(lock_);
+        std::vector<std::string> snapshots;
+        ASSERT_TRUE(init->ListSnapshots(local_lock.get(), &snapshots));
+        ASSERT_TRUE(snapshots.empty());
+    }
+
     // Check that the target partitions have the same content after the merge.
     for (const auto& name : {"sys_b", "vnd_b", "prd_b"}) {
         ASSERT_TRUE(IsPartitionUnchanged(name))
@@ -1814,6 +1978,13 @@
 
     ASSERT_TRUE(init->PerformInitTransition(SnapshotManager::InitTransition::SECOND_STAGE));
 
+    // :TODO: this is a workaround to ensure the handler list stays empty. We
+    // should make this test more like actual init, and spawn two copies of
+    // snapuserd, given how many other tests we now have for normal snapuserd.
+    ASSERT_TRUE(init->snapuserd_client()->WaitForDeviceDelete("sys_b-user-cow-init"));
+    ASSERT_TRUE(init->snapuserd_client()->WaitForDeviceDelete("vnd_b-user-cow-init"));
+    ASSERT_TRUE(init->snapuserd_client()->WaitForDeviceDelete("prd_b-user-cow-init"));
+
     // The control device should have been renamed.
     ASSERT_TRUE(android::fs_mgr::WaitForFileDeleted("/dev/dm-user/sys_b-user-cow-init", 10s));
     ASSERT_EQ(access("/dev/dm-user/sys_b-user-cow", F_OK), 0);
diff --git a/fs_mgr/libsnapshot/test_helpers.cpp b/fs_mgr/libsnapshot/test_helpers.cpp
index 6104c82..e3e3af8 100644
--- a/fs_mgr/libsnapshot/test_helpers.cpp
+++ b/fs_mgr/libsnapshot/test_helpers.cpp
@@ -23,6 +23,7 @@
 #include <android-base/unique_fd.h>
 #include <gtest/gtest.h>
 #include <openssl/sha.h>
+#include <payload_consumer/file_descriptor.h>
 
 namespace android {
 namespace snapshot {
@@ -169,6 +170,37 @@
     return true;
 }
 
+std::string HashSnapshot(ISnapshotWriter* writer) {
+    auto reader = writer->OpenReader();
+    if (!reader) {
+        return {};
+    }
+
+    SHA256_CTX ctx;
+    SHA256_Init(&ctx);
+
+    uint64_t remaining = reader->BlockDevSize();
+    char buffer[4096];
+    while (remaining) {
+        size_t to_read =
+                static_cast<size_t>(std::min(remaining, static_cast<uint64_t>(sizeof(buffer))));
+        ssize_t read = reader->Read(&buffer, to_read);
+        if (read <= 0) {
+            if (read < 0) {
+                LOG(ERROR) << "Failed to read from snapshot writer";
+                return {};
+            }
+            break;
+        }
+        SHA256_Update(&ctx, buffer, to_read);
+        remaining -= static_cast<size_t>(read);
+    }
+
+    uint8_t out[32];
+    SHA256_Final(out, &ctx);
+    return ToHexString(out, sizeof(out));
+}
+
 std::optional<std::string> GetHash(const std::string& path) {
     std::string content;
     if (!android::base::ReadFileToString(path, &content, true)) {