diff --git a/adb/Android.bp b/adb/Android.bp
index 414d3d0..81d20c1 100644
--- a/adb/Android.bp
+++ b/adb/Android.bp
@@ -317,6 +317,7 @@
         "libandroidfw",
         "libapp_processes_protos_full",
         "libbase",
+        "libbrotli",
         "libcutils",
         "libcrypto_utils",
         "libcrypto",
@@ -469,6 +470,7 @@
     static_libs: [
         "libadbconnection_server",
         "libadbd_core",
+        "libbrotli",
         "libdiagnose_usb",
     ],
 
@@ -567,6 +569,7 @@
     },
 
     static_libs: [
+        "libbrotli",
         "libcutils_sockets",
         "libdiagnose_usb",
         "libmdnssd",
@@ -606,6 +609,7 @@
         "libapp_processes_protos_lite",
         "libasyncio",
         "libbase",
+        "libbrotli",
         "libcap",
         "libcrypto_utils",
         "libcutils_sockets",
diff --git a/adb/adb.cpp b/adb/adb.cpp
index 98db191..b514368 100644
--- a/adb/adb.cpp
+++ b/adb/adb.cpp
@@ -1016,8 +1016,12 @@
         if (kill_forward) {
             r = remove_listener(pieces[0].c_str(), transport);
         } else {
-            r = install_listener(pieces[0], pieces[1].c_str(), transport, no_rebind,
-                                 &resolved_tcp_port, &error);
+            int flags = 0;
+            if (no_rebind) {
+                flags |= INSTALL_LISTENER_NO_REBIND;
+            }
+            r = install_listener(pieces[0], pieces[1].c_str(), transport, flags, &resolved_tcp_port,
+                                 &error);
         }
         if (r == INSTALL_STATUS_OK) {
 #if ADB_HOST
diff --git a/adb/adb_listeners.cpp b/adb/adb_listeners.cpp
index 29909a5..43a9252 100644
--- a/adb/adb_listeners.cpp
+++ b/adb/adb_listeners.cpp
@@ -164,6 +164,15 @@
     }
 }
 
+void enable_daemon_sockets() EXCLUDES(listener_list_mutex) {
+    std::lock_guard<std::mutex> lock(listener_list_mutex);
+    for (auto& l : listener_list) {
+        if (l->connect_to == "*smartsocket*") {
+            fdevent_set(l->fde, FDE_READ);
+        }
+    }
+}
+
 void close_smartsockets() EXCLUDES(listener_list_mutex) {
     std::lock_guard<std::mutex> lock(listener_list_mutex);
     auto pred = [](const std::unique_ptr<alistener>& listener) {
@@ -173,7 +182,7 @@
 }
 
 InstallStatus install_listener(const std::string& local_name, const char* connect_to,
-                               atransport* transport, int no_rebind, int* resolved_tcp_port,
+                               atransport* transport, int flags, int* resolved_tcp_port,
                                std::string* error) EXCLUDES(listener_list_mutex) {
     std::lock_guard<std::mutex> lock(listener_list_mutex);
     for (auto& l : listener_list) {
@@ -184,8 +193,8 @@
                 return INSTALL_STATUS_INTERNAL_ERROR;
             }
 
-            // Can't repurpose a listener if 'no_rebind' is true.
-            if (no_rebind) {
+            // Can't repurpose a listener if INSTALL_LISTENER_NO_REBIND is set
+            if (flags & INSTALL_LISTENER_NO_REBIND) {
                 *error = "cannot rebind";
                 return INSTALL_STATUS_CANNOT_REBIND;
             }
@@ -222,7 +231,9 @@
     } else {
         listener->fde = fdevent_create(listener->fd, listener_event_func, listener.get());
     }
-    fdevent_set(listener->fde, FDE_READ);
+    if ((flags & INSTALL_LISTENER_DISABLED) == 0) {
+        fdevent_set(listener->fde, FDE_READ);
+    }
 
     listener->transport = transport;
 
diff --git a/adb/adb_listeners.h b/adb/adb_listeners.h
index 70a2ee1..354dcc5 100644
--- a/adb/adb_listeners.h
+++ b/adb/adb_listeners.h
@@ -32,8 +32,11 @@
   INSTALL_STATUS_LISTENER_NOT_FOUND = -4,
 };
 
+inline constexpr int INSTALL_LISTENER_NO_REBIND = 1 << 0;
+inline constexpr int INSTALL_LISTENER_DISABLED = 1 << 1;
+
 InstallStatus install_listener(const std::string& local_name, const char* connect_to,
-                               atransport* transport, int no_rebind, int* resolved_tcp_port,
+                               atransport* transport, int flags, int* resolved_tcp_port,
                                std::string* error);
 
 std::string format_listeners();
@@ -41,6 +44,7 @@
 InstallStatus remove_listener(const char* local_name, atransport* transport);
 void remove_all_listeners(void);
 
+void enable_daemon_sockets();
 void close_smartsockets();
 
 #endif /* __ADB_LISTENERS_H */
diff --git a/adb/brotli_utils.h b/adb/brotli_utils.h
new file mode 100644
index 0000000..c5be73d
--- /dev/null
+++ b/adb/brotli_utils.h
@@ -0,0 +1,144 @@
+/*
+ * 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.
+ */
+
+#pragma once
+
+#include <span>
+
+#include <brotli/decode.h>
+#include <brotli/encode.h>
+
+#include "types.h"
+
+enum class BrotliDecodeResult {
+    Error,
+    Done,
+    NeedInput,
+    MoreOutput,
+};
+
+struct BrotliDecoder {
+    explicit BrotliDecoder(std::span<char> output_buffer)
+        : output_buffer_(output_buffer),
+          decoder_(BrotliDecoderCreateInstance(nullptr, nullptr, nullptr),
+                   BrotliDecoderDestroyInstance) {}
+
+    void Append(Block&& block) { input_buffer_.append(std::move(block)); }
+
+    BrotliDecodeResult Decode(std::span<char>* output) {
+        size_t available_in = input_buffer_.front_size();
+        const uint8_t* next_in = reinterpret_cast<const uint8_t*>(input_buffer_.front_data());
+
+        size_t available_out = output_buffer_.size();
+        uint8_t* next_out = reinterpret_cast<uint8_t*>(output_buffer_.data());
+
+        BrotliDecoderResult r = BrotliDecoderDecompressStream(
+                decoder_.get(), &available_in, &next_in, &available_out, &next_out, nullptr);
+
+        size_t bytes_consumed = input_buffer_.front_size() - available_in;
+        input_buffer_.drop_front(bytes_consumed);
+
+        size_t bytes_emitted = output_buffer_.size() - available_out;
+        *output = std::span<char>(output_buffer_.data(), bytes_emitted);
+
+        switch (r) {
+            case BROTLI_DECODER_RESULT_SUCCESS:
+                return BrotliDecodeResult::Done;
+            case BROTLI_DECODER_RESULT_ERROR:
+                return BrotliDecodeResult::Error;
+            case BROTLI_DECODER_RESULT_NEEDS_MORE_INPUT:
+                // Brotli guarantees as one of its invariants that if it returns NEEDS_MORE_INPUT,
+                // it will consume the entire input buffer passed in, so we don't have to worry
+                // about bytes left over in the front block with more input remaining.
+                return BrotliDecodeResult::NeedInput;
+            case BROTLI_DECODER_RESULT_NEEDS_MORE_OUTPUT:
+                return BrotliDecodeResult::MoreOutput;
+        }
+    }
+
+  private:
+    IOVector input_buffer_;
+    std::span<char> output_buffer_;
+    std::unique_ptr<BrotliDecoderState, void (*)(BrotliDecoderState*)> decoder_;
+};
+
+enum class BrotliEncodeResult {
+    Error,
+    Done,
+    NeedInput,
+    MoreOutput,
+};
+
+template <size_t OutputBlockSize>
+struct BrotliEncoder {
+    explicit BrotliEncoder()
+        : output_block_(OutputBlockSize),
+          output_bytes_left_(OutputBlockSize),
+          encoder_(BrotliEncoderCreateInstance(nullptr, nullptr, nullptr),
+                   BrotliEncoderDestroyInstance) {
+        BrotliEncoderSetParameter(encoder_.get(), BROTLI_PARAM_QUALITY, 1);
+    }
+
+    void Append(Block input) { input_buffer_.append(std::move(input)); }
+    void Finish() { finished_ = true; }
+
+    BrotliEncodeResult Encode(Block* output) {
+        output->clear();
+        while (true) {
+            size_t available_in = input_buffer_.front_size();
+            const uint8_t* next_in = reinterpret_cast<const uint8_t*>(input_buffer_.front_data());
+
+            size_t available_out = output_bytes_left_;
+            uint8_t* next_out = reinterpret_cast<uint8_t*>(output_block_.data() +
+                                                           (OutputBlockSize - output_bytes_left_));
+
+            BrotliEncoderOperation op = BROTLI_OPERATION_PROCESS;
+            if (finished_) {
+                op = BROTLI_OPERATION_FINISH;
+            }
+
+            if (!BrotliEncoderCompressStream(encoder_.get(), op, &available_in, &next_in,
+                                             &available_out, &next_out, nullptr)) {
+                return BrotliEncodeResult::Error;
+            }
+
+            size_t bytes_consumed = input_buffer_.front_size() - available_in;
+            input_buffer_.drop_front(bytes_consumed);
+
+            output_bytes_left_ = available_out;
+
+            if (BrotliEncoderIsFinished(encoder_.get())) {
+                output_block_.resize(OutputBlockSize - output_bytes_left_);
+                *output = std::move(output_block_);
+                return BrotliEncodeResult::Done;
+            } else if (output_bytes_left_ == 0) {
+                *output = std::move(output_block_);
+                output_block_.resize(OutputBlockSize);
+                output_bytes_left_ = OutputBlockSize;
+                return BrotliEncodeResult::MoreOutput;
+            } else if (input_buffer_.empty()) {
+                return BrotliEncodeResult::NeedInput;
+            }
+        }
+    }
+
+  private:
+    bool finished_ = false;
+    IOVector input_buffer_;
+    Block output_block_;
+    size_t output_bytes_left_;
+    std::unique_ptr<BrotliEncoderState, void (*)(BrotliEncoderState*)> encoder_;
+};
diff --git a/adb/client/adb_install.cpp b/adb/client/adb_install.cpp
index 21b8f49..e4d010c 100644
--- a/adb/client/adb_install.cpp
+++ b/adb/client/adb_install.cpp
@@ -22,11 +22,12 @@
 #include <stdlib.h>
 #include <unistd.h>
 #include <algorithm>
-#include <iostream>
 #include <string>
+#include <string_view>
 #include <vector>
 
 #include <android-base/file.h>
+#include <android-base/parsebool.h>
 #include <android-base/stringprintf.h>
 #include <android-base/strings.h>
 
@@ -39,8 +40,9 @@
 #include "fastdeploy.h"
 #include "incremental.h"
 
+using namespace std::literals;
+
 static constexpr int kFastDeployMinApi = 24;
-static constexpr int kIncrementalMinApi = 29;
 
 namespace {
 
@@ -50,6 +52,8 @@
     INSTALL_STREAM,
     INSTALL_INCREMENTAL,
 };
+
+enum class CmdlineOption { None, Enable, Disable };
 }
 
 static bool can_use_feature(const char* feature) {
@@ -286,7 +290,7 @@
         }
     }
 
-    if (do_sync_push(apk_file, apk_dest.c_str(), false)) {
+    if (do_sync_push(apk_file, apk_dest.c_str(), false, true)) {
         result = pm_command(argc, argv);
         delete_device_file(apk_dest);
     }
@@ -299,28 +303,23 @@
     return std::chrono::duration_cast<std::chrono::milliseconds>(end - start).count();
 }
 
-static int install_app_incremental(int argc, const char** argv) {
-    printf("Performing Incremental Install\n");
+static int install_app_incremental(int argc, const char** argv, bool wait, bool silent) {
     using clock = std::chrono::high_resolution_clock;
     const auto start = clock::now();
     int first_apk = -1;
     int last_apk = -1;
-    std::string cert_path;
-    bool wait = false;
-    std::vector<std::string_view> args = {"package"};
+    std::vector<std::string_view> args = {"package"sv};
     for (int i = 0; i < argc; ++i) {
         const auto arg = std::string_view(argv[i]);
-        if (android::base::EndsWithIgnoreCase(arg, ".apk")) {
+        if (android::base::EndsWithIgnoreCase(arg, ".apk"sv)) {
             last_apk = i;
             if (first_apk == -1) {
                 first_apk = i;
             }
-        } else if (arg == "--wait") {
-            wait = true;
-        } else if (arg.starts_with("install-")) {
+        } else if (arg.starts_with("install-"sv)) {
             // incremental installation command on the device is the same for all its variations in
             // the adb, e.g. install-multiple or install-multi-package
-            args.push_back("install");
+            args.push_back("install"sv);
         } else {
             args.push_back(arg);
         }
@@ -328,16 +327,23 @@
 
     if (first_apk == -1) error_exit("Need at least one APK file on command line");
 
-    const auto afterApk = clock::now();
+    auto files = incremental::Files{argv + first_apk, argv + last_apk + 1};
+    if (silent) {
+        // For a silent installation we want to do the lightweight check first and bail early and
+        // quietly if it fails.
+        if (!incremental::can_install(files)) {
+            return -1;
+        }
+    }
 
-    auto server_process = incremental::install({argv + first_apk, argv + last_apk + 1});
+    printf("Performing Incremental Install\n");
+    auto server_process = incremental::install(files, silent);
     if (!server_process) {
         return -1;
     }
 
     const auto end = clock::now();
-    printf("Install command complete (ms: %d total, %d apk prep, %d install)\n",
-           msBetween(start, end), msBetween(start, afterApk), msBetween(afterApk, end));
+    printf("Install command complete in %d ms\n", msBetween(start, end));
 
     if (wait) {
         (*server_process).wait();
@@ -346,66 +352,134 @@
     return 0;
 }
 
+static std::pair<InstallMode, std::optional<InstallMode>> calculateInstallMode(
+        InstallMode modeFromArgs, bool fastdeploy, CmdlineOption incrementalRequest) {
+    if (incrementalRequest == CmdlineOption::Enable) {
+        if (fastdeploy) {
+            error_exit(
+                    "--incremental and --fast-deploy options are incompatible. "
+                    "Please choose one");
+        }
+    }
+
+    if (modeFromArgs != INSTALL_DEFAULT) {
+        if (incrementalRequest == CmdlineOption::Enable) {
+            error_exit("--incremental is not compatible with other installation modes");
+        }
+        return {modeFromArgs, std::nullopt};
+    }
+
+    if (incrementalRequest != CmdlineOption::Disable && !is_abb_exec_supported()) {
+        if (incrementalRequest == CmdlineOption::None) {
+            incrementalRequest = CmdlineOption::Disable;
+        } else {
+            error_exit("Device doesn't support incremental installations");
+        }
+    }
+    if (incrementalRequest == CmdlineOption::None) {
+        // check if the host is ok with incremental by default
+        if (const char* incrementalFromEnv = getenv("ADB_INSTALL_DEFAULT_INCREMENTAL")) {
+            using namespace android::base;
+            auto val = ParseBool(incrementalFromEnv);
+            if (val == ParseBoolResult::kFalse) {
+                incrementalRequest = CmdlineOption::Disable;
+            }
+        }
+    }
+    if (incrementalRequest == CmdlineOption::None) {
+        // still ok: let's see if the device allows using incremental by default
+        // it starts feeling like we're looking for an excuse to not to use incremental...
+        std::string error;
+        std::vector<std::string> args = {"settings", "get",
+                                         "enable_adb_incremental_install_default"};
+        auto fd = send_abb_exec_command(args, &error);
+        if (!fd.ok()) {
+            fprintf(stderr, "adb: retrieving the default device installation mode failed: %s",
+                    error.c_str());
+        } else {
+            char buf[BUFSIZ] = {};
+            read_status_line(fd.get(), buf, sizeof(buf));
+            using namespace android::base;
+            auto val = ParseBool(buf);
+            if (val == ParseBoolResult::kFalse) {
+                incrementalRequest = CmdlineOption::Disable;
+            }
+        }
+    }
+
+    if (incrementalRequest == CmdlineOption::Enable) {
+        // explicitly requested - no fallback
+        return {INSTALL_INCREMENTAL, std::nullopt};
+    }
+    const auto bestMode = best_install_mode();
+    if (incrementalRequest == CmdlineOption::None) {
+        // no opinion - use incremental, fallback to regular on a failure.
+        return {INSTALL_INCREMENTAL, bestMode};
+    }
+    // incremental turned off - use the regular best mode without a fallback.
+    return {bestMode, std::nullopt};
+}
+
 int install_app(int argc, const char** argv) {
     std::vector<int> processedArgIndices;
     InstallMode installMode = INSTALL_DEFAULT;
     bool use_fastdeploy = false;
     bool is_reinstall = false;
+    bool wait = false;
+    auto incremental_request = CmdlineOption::None;
     FastDeploy_AgentUpdateStrategy agent_update_strategy = FastDeploy_AgentUpdateDifferentVersion;
 
     for (int i = 1; i < argc; i++) {
-        if (!strcmp(argv[i], "--streaming")) {
+        if (argv[i] == "--streaming"sv) {
             processedArgIndices.push_back(i);
             installMode = INSTALL_STREAM;
-        } else if (!strcmp(argv[i], "--no-streaming")) {
+        } else if (argv[i] == "--no-streaming"sv) {
             processedArgIndices.push_back(i);
             installMode = INSTALL_PUSH;
-        } else if (!strcmp(argv[i], "-r")) {
+        } else if (argv[i] == "-r"sv) {
             // Note that this argument is not added to processedArgIndices because it
             // must be passed through to pm
             is_reinstall = true;
-        } else if (!strcmp(argv[i], "--fastdeploy")) {
+        } else if (argv[i] == "--fastdeploy"sv) {
             processedArgIndices.push_back(i);
             use_fastdeploy = true;
-        } else if (!strcmp(argv[i], "--no-fastdeploy")) {
+        } else if (argv[i] == "--no-fastdeploy"sv) {
             processedArgIndices.push_back(i);
             use_fastdeploy = false;
-        } else if (!strcmp(argv[i], "--force-agent")) {
+        } else if (argv[i] == "--force-agent"sv) {
             processedArgIndices.push_back(i);
             agent_update_strategy = FastDeploy_AgentUpdateAlways;
-        } else if (!strcmp(argv[i], "--date-check-agent")) {
+        } else if (argv[i] == "--date-check-agent"sv) {
             processedArgIndices.push_back(i);
             agent_update_strategy = FastDeploy_AgentUpdateNewerTimeStamp;
-        } else if (!strcmp(argv[i], "--version-check-agent")) {
+        } else if (argv[i] == "--version-check-agent"sv) {
             processedArgIndices.push_back(i);
             agent_update_strategy = FastDeploy_AgentUpdateDifferentVersion;
-        } else if (!strcmp(argv[i], "--incremental")) {
+        } else if (strlen(argv[i]) >= "--incr"sv.size() && "--incremental"sv.starts_with(argv[i])) {
             processedArgIndices.push_back(i);
-            installMode = INSTALL_INCREMENTAL;
-        } else if (!strcmp(argv[i], "--no-incremental")) {
+            incremental_request = CmdlineOption::Enable;
+        } else if (strlen(argv[i]) >= "--no-incr"sv.size() &&
+                   "--no-incremental"sv.starts_with(argv[i])) {
             processedArgIndices.push_back(i);
-            installMode = INSTALL_DEFAULT;
+            incremental_request = CmdlineOption::Disable;
+        } else if (argv[i] == "--wait"sv) {
+            processedArgIndices.push_back(i);
+            wait = true;
         }
     }
 
-    if (installMode == INSTALL_INCREMENTAL) {
-        if (get_device_api_level() < kIncrementalMinApi || !is_abb_exec_supported()) {
-            error_exit("Attempting to use incremental install on unsupported device");
-        }
-    }
-
-    if (installMode == INSTALL_DEFAULT) {
-        installMode = best_install_mode();
-    }
-
-    if (installMode == INSTALL_STREAM && best_install_mode() == INSTALL_PUSH) {
+    auto [primaryMode, fallbackMode] =
+            calculateInstallMode(installMode, use_fastdeploy, incremental_request);
+    if ((primaryMode == INSTALL_STREAM || fallbackMode.value_or(INSTALL_PUSH) == INSTALL_STREAM) &&
+        best_install_mode() == INSTALL_PUSH) {
         error_exit("Attempting to use streaming install on unsupported device");
     }
 
     if (use_fastdeploy && get_device_api_level() < kFastDeployMinApi) {
-        printf("Fast Deploy is only compatible with devices of API version %d or higher, "
-               "ignoring.\n",
-               kFastDeployMinApi);
+        fprintf(stderr,
+                "Fast Deploy is only compatible with devices of API version %d or higher, "
+                "ignoring.\n",
+                kFastDeployMinApi);
         use_fastdeploy = false;
     }
     fastdeploy_set_agent_update_strategy(agent_update_strategy);
@@ -421,19 +495,27 @@
         error_exit("install requires an apk argument");
     }
 
-    switch (installMode) {
-        case INSTALL_PUSH:
-            return install_app_legacy(passthrough_argv.size(), passthrough_argv.data(),
-                                      use_fastdeploy);
-        case INSTALL_STREAM:
-            return install_app_streamed(passthrough_argv.size(), passthrough_argv.data(),
-                                        use_fastdeploy);
-        case INSTALL_INCREMENTAL:
-            return install_app_incremental(passthrough_argv.size(), passthrough_argv.data());
-        case INSTALL_DEFAULT:
-        default:
-            return 1;
+    auto runInstallMode = [&](InstallMode installMode, bool silent) {
+        switch (installMode) {
+            case INSTALL_PUSH:
+                return install_app_legacy(passthrough_argv.size(), passthrough_argv.data(),
+                                          use_fastdeploy);
+            case INSTALL_STREAM:
+                return install_app_streamed(passthrough_argv.size(), passthrough_argv.data(),
+                                            use_fastdeploy);
+            case INSTALL_INCREMENTAL:
+                return install_app_incremental(passthrough_argv.size(), passthrough_argv.data(),
+                                               wait, silent);
+            case INSTALL_DEFAULT:
+            default:
+                return 1;
+        }
+    };
+    auto res = runInstallMode(primaryMode, fallbackMode.has_value());
+    if (res && fallbackMode.value_or(primaryMode) != primaryMode) {
+        res = runInstallMode(*fallbackMode, false);
     }
+    return res;
 }
 
 int install_multiple_app(int argc, const char** argv) {
diff --git a/adb/client/bugreport.cpp b/adb/client/bugreport.cpp
index 8ca44e8..ab93f7d 100644
--- a/adb/client/bugreport.cpp
+++ b/adb/client/bugreport.cpp
@@ -282,5 +282,5 @@
 
 bool Bugreport::DoSyncPull(const std::vector<const char*>& srcs, const char* dst, bool copy_attrs,
                            const char* name) {
-    return do_sync_pull(srcs, dst, copy_attrs, name);
+    return do_sync_pull(srcs, dst, copy_attrs, false, name);
 }
diff --git a/adb/client/commandline.cpp b/adb/client/commandline.cpp
index 0c96104..04b250d 100644
--- a/adb/client/commandline.cpp
+++ b/adb/client/commandline.cpp
@@ -129,15 +129,21 @@
         " reverse --remove-all     remove all reverse socket connections from device\n"
         "\n"
         "file transfer:\n"
-        " push [--sync] LOCAL... REMOTE\n"
+        " push [--sync] [-zZ] LOCAL... REMOTE\n"
         "     copy local files/directories to device\n"
         "     --sync: only push files that are newer on the host than the device\n"
-        " pull [-a] REMOTE... LOCAL\n"
+        "     -z: enable compression\n"
+        "     -Z: disable compression\n"
+        " pull [-azZ] REMOTE... LOCAL\n"
         "     copy files/dirs from device\n"
         "     -a: preserve file timestamp and mode\n"
-        " sync [all|data|odm|oem|product|system|system_ext|vendor]\n"
+        "     -z: enable compression\n"
+        "     -Z: disable compression\n"
+        " sync [-lzZ] [all|data|odm|oem|product|system|system_ext|vendor]\n"
         "     sync a local build from $ANDROID_PRODUCT_OUT to the device (default all)\n"
         "     -l: list files that would be copied, but don't copy them\n"
+        "     -z: enable compression\n"
+        "     -Z: disable compression\n"
         "\n"
         "shell:\n"
         " shell [-e ESCAPE] [-n] [-Tt] [-x] [COMMAND...]\n"
@@ -1309,8 +1315,12 @@
 }
 
 static void parse_push_pull_args(const char** arg, int narg, std::vector<const char*>* srcs,
-                                 const char** dst, bool* copy_attrs, bool* sync) {
+                                 const char** dst, bool* copy_attrs, bool* sync, bool* compressed) {
     *copy_attrs = false;
+    const char* adb_compression = getenv("ADB_COMPRESSION");
+    if (adb_compression && strcmp(adb_compression, "0") == 0) {
+        *compressed = false;
+    }
 
     srcs->clear();
     bool ignore_flags = false;
@@ -1322,6 +1332,14 @@
                 // Silently ignore for backwards compatibility.
             } else if (!strcmp(*arg, "-a")) {
                 *copy_attrs = true;
+            } else if (!strcmp(*arg, "-z")) {
+                if (compressed != nullptr) {
+                    *compressed = true;
+                }
+            } else if (!strcmp(*arg, "-Z")) {
+                if (compressed != nullptr) {
+                    *compressed = false;
+                }
             } else if (!strcmp(*arg, "--sync")) {
                 if (sync != nullptr) {
                     *sync = true;
@@ -1876,20 +1894,22 @@
     } else if (!strcmp(argv[0], "push")) {
         bool copy_attrs = false;
         bool sync = false;
+        bool compressed = true;
         std::vector<const char*> srcs;
         const char* dst = nullptr;
 
-        parse_push_pull_args(&argv[1], argc - 1, &srcs, &dst, &copy_attrs, &sync);
+        parse_push_pull_args(&argv[1], argc - 1, &srcs, &dst, &copy_attrs, &sync, &compressed);
         if (srcs.empty() || !dst) error_exit("push requires an argument");
-        return do_sync_push(srcs, dst, sync) ? 0 : 1;
+        return do_sync_push(srcs, dst, sync, compressed) ? 0 : 1;
     } else if (!strcmp(argv[0], "pull")) {
         bool copy_attrs = false;
+        bool compressed = true;
         std::vector<const char*> srcs;
         const char* dst = ".";
 
-        parse_push_pull_args(&argv[1], argc - 1, &srcs, &dst, &copy_attrs, nullptr);
+        parse_push_pull_args(&argv[1], argc - 1, &srcs, &dst, &copy_attrs, nullptr, &compressed);
         if (srcs.empty()) error_exit("pull requires an argument");
-        return do_sync_pull(srcs, dst, copy_attrs) ? 0 : 1;
+        return do_sync_pull(srcs, dst, copy_attrs, compressed) ? 0 : 1;
     } else if (!strcmp(argv[0], "install")) {
         if (argc < 2) error_exit("install requires an argument");
         return install_app(argc, argv);
@@ -1905,18 +1925,38 @@
     } else if (!strcmp(argv[0], "sync")) {
         std::string src;
         bool list_only = false;
-        if (argc < 2) {
-            // No partition specified: sync all of them.
-        } else if (argc >= 2 && strcmp(argv[1], "-l") == 0) {
-            list_only = true;
-            if (argc == 3) src = argv[2];
-        } else if (argc == 2) {
-            src = argv[1];
-        } else {
-            error_exit("usage: adb sync [-l] [PARTITION]");
+        bool compressed = true;
+
+        const char* adb_compression = getenv("ADB_COMPRESSION");
+        if (adb_compression && strcmp(adb_compression, "0") == 0) {
+            compressed = false;
         }
 
-        if (src.empty()) src = "all";
+        int opt;
+        while ((opt = getopt(argc, const_cast<char**>(argv), "lzZ")) != -1) {
+            switch (opt) {
+                case 'l':
+                    list_only = true;
+                    break;
+                case 'z':
+                    compressed = true;
+                    break;
+                case 'Z':
+                    compressed = false;
+                    break;
+                default:
+                    error_exit("usage: adb sync [-lzZ] [PARTITION]");
+            }
+        }
+
+        if (optind == argc) {
+            src = "all";
+        } else if (optind + 1 == argc) {
+            src = argv[optind];
+        } else {
+            error_exit("usage: adb sync [-lzZ] [PARTITION]");
+        }
+
         std::vector<std::string> partitions{"data",   "odm",        "oem",   "product",
                                             "system", "system_ext", "vendor"};
         bool found = false;
@@ -1925,7 +1965,7 @@
                 std::string src_dir{product_file(partition)};
                 if (!directory_exists(src_dir)) continue;
                 found = true;
-                if (!do_sync_sync(src_dir, "/" + partition, list_only)) return 1;
+                if (!do_sync_sync(src_dir, "/" + partition, list_only, compressed)) return 1;
             }
         }
         if (!found) error_exit("don't know how to sync %s partition", src.c_str());
diff --git a/adb/client/fastdeploy.cpp b/adb/client/fastdeploy.cpp
index c5fc12f..de82e14 100644
--- a/adb/client/fastdeploy.cpp
+++ b/adb/client/fastdeploy.cpp
@@ -112,7 +112,7 @@
     // but can't be removed until after the push.
     unix_close(tf.release());
 
-    if (!do_sync_push(srcs, dst, sync)) {
+    if (!do_sync_push(srcs, dst, sync, true)) {
         error_exit("Failed to push fastdeploy agent to device.");
     }
 }
diff --git a/adb/client/file_sync_client.cpp b/adb/client/file_sync_client.cpp
index 599a3e3..0844428 100644
--- a/adb/client/file_sync_client.cpp
+++ b/adb/client/file_sync_client.cpp
@@ -42,6 +42,7 @@
 #include "adb_client.h"
 #include "adb_io.h"
 #include "adb_utils.h"
+#include "brotli_utils.h"
 #include "file_sync_protocol.h"
 #include "line_printer.h"
 #include "sysdeps/errno.h"
@@ -233,6 +234,8 @@
         } else {
             have_stat_v2_ = CanUseFeature(features_, kFeatureStat2);
             have_ls_v2_ = CanUseFeature(features_, kFeatureLs2);
+            have_sendrecv_v2_ = CanUseFeature(features_, kFeatureSendRecv2);
+            have_sendrecv_v2_brotli_ = CanUseFeature(features_, kFeatureSendRecv2Brotli);
             fd.reset(adb_connect("sync:", &error));
             if (fd < 0) {
                 Error("connect failed: %s", error.c_str());
@@ -256,6 +259,9 @@
         line_printer_.KeepInfoLine();
     }
 
+    bool HaveSendRecv2() const { return have_sendrecv_v2_; }
+    bool HaveSendRecv2Brotli() const { return have_sendrecv_v2_brotli_; }
+
     const FeatureSet& Features() const { return features_; }
 
     bool IsValid() { return fd >= 0; }
@@ -314,6 +320,62 @@
         req->path_length = path.length();
         char* data = reinterpret_cast<char*>(req + 1);
         memcpy(data, path.data(), path.length());
+        return WriteFdExactly(fd, buf.data(), buf.size());
+    }
+
+    bool SendSend2(std::string_view path, mode_t mode, bool compressed) {
+        if (path.length() > 1024) {
+            Error("SendRequest failed: path too long: %zu", path.length());
+            errno = ENAMETOOLONG;
+            return false;
+        }
+
+        Block buf;
+
+        SyncRequest req;
+        req.id = ID_SEND_V2;
+        req.path_length = path.length();
+
+        syncmsg msg;
+        msg.send_v2_setup.id = ID_SEND_V2;
+        msg.send_v2_setup.mode = mode;
+        msg.send_v2_setup.flags = compressed ? kSyncFlagBrotli : kSyncFlagNone;
+
+        buf.resize(sizeof(SyncRequest) + path.length() + sizeof(msg.send_v2_setup));
+
+        void* p = buf.data();
+
+        p = mempcpy(p, &req, sizeof(SyncRequest));
+        p = mempcpy(p, path.data(), path.length());
+        p = mempcpy(p, &msg.send_v2_setup, sizeof(msg.send_v2_setup));
+
+        return WriteFdExactly(fd, buf.data(), buf.size());
+    }
+
+    bool SendRecv2(const std::string& path) {
+        if (path.length() > 1024) {
+            Error("SendRequest failed: path too long: %zu", path.length());
+            errno = ENAMETOOLONG;
+            return false;
+        }
+
+        Block buf;
+
+        SyncRequest req;
+        req.id = ID_RECV_V2;
+        req.path_length = path.length();
+
+        syncmsg msg;
+        msg.recv_v2_setup.id = ID_RECV_V2;
+        msg.recv_v2_setup.flags = kSyncFlagBrotli;
+
+        buf.resize(sizeof(SyncRequest) + path.length() + sizeof(msg.recv_v2_setup));
+
+        void* p = buf.data();
+
+        p = mempcpy(p, &req, sizeof(SyncRequest));
+        p = mempcpy(p, path.data(), path.length());
+        p = mempcpy(p, &msg.recv_v2_setup, sizeof(msg.recv_v2_setup));
 
         return WriteFdExactly(fd, buf.data(), buf.size());
     }
@@ -370,8 +432,8 @@
             }
 
             if (msg.stat_v1.id != ID_LSTAT_V1) {
-                PLOG(FATAL) << "protocol fault: stat response has wrong message id: "
-                            << msg.stat_v1.id;
+                LOG(FATAL) << "protocol fault: stat response has wrong message id: "
+                           << msg.stat_v1.id;
             }
 
             if (msg.stat_v1.mode == 0 && msg.stat_v1.size == 0 && msg.stat_v1.mtime == 0) {
@@ -445,7 +507,7 @@
         char* p = &buf[0];
 
         SyncRequest* req_send = reinterpret_cast<SyncRequest*>(p);
-        req_send->id = ID_SEND;
+        req_send->id = ID_SEND_V1;
         req_send->path_length = path_and_mode.length();
         p += sizeof(SyncRequest);
         memcpy(p, path_and_mode.data(), path_and_mode.size());
@@ -471,11 +533,92 @@
         return true;
     }
 
+    bool SendLargeFileCompressed(const std::string& path, mode_t mode, const std::string& lpath,
+                                 const std::string& rpath, unsigned mtime) {
+        if (!SendSend2(path, mode, true)) {
+            Error("failed to send ID_SEND_V2 message '%s': %s", path.c_str(), strerror(errno));
+            return false;
+        }
+
+        struct stat st;
+        if (stat(lpath.c_str(), &st) == -1) {
+            Error("cannot stat '%s': %s", lpath.c_str(), strerror(errno));
+            return false;
+        }
+
+        uint64_t total_size = st.st_size;
+        uint64_t bytes_copied = 0;
+
+        unique_fd lfd(adb_open(lpath.c_str(), O_RDONLY | O_CLOEXEC));
+        if (lfd < 0) {
+            Error("opening '%s' locally failed: %s", lpath.c_str(), strerror(errno));
+            return false;
+        }
+
+        syncsendbuf sbuf;
+        sbuf.id = ID_DATA;
+
+        BrotliEncoder<SYNC_DATA_MAX> encoder;
+        bool sending = true;
+        while (sending) {
+            Block input(SYNC_DATA_MAX);
+            int r = adb_read(lfd.get(), input.data(), input.size());
+            if (r < 0) {
+                Error("reading '%s' locally failed: %s", lpath.c_str(), strerror(errno));
+                return false;
+            }
+
+            if (r == 0) {
+                encoder.Finish();
+            } else {
+                input.resize(r);
+                encoder.Append(std::move(input));
+                RecordBytesTransferred(r);
+                bytes_copied += r;
+                ReportProgress(rpath, bytes_copied, total_size);
+            }
+
+            while (true) {
+                Block output;
+                BrotliEncodeResult result = encoder.Encode(&output);
+                if (result == BrotliEncodeResult::Error) {
+                    Error("compressing '%s' locally failed", lpath.c_str());
+                    return false;
+                }
+
+                if (!output.empty()) {
+                    sbuf.size = output.size();
+                    memcpy(sbuf.data, output.data(), output.size());
+                    WriteOrDie(lpath, rpath, &sbuf, sizeof(SyncRequest) + output.size());
+                }
+
+                if (result == BrotliEncodeResult::Done) {
+                    sending = false;
+                    break;
+                } else if (result == BrotliEncodeResult::NeedInput) {
+                    break;
+                } else if (result == BrotliEncodeResult::MoreOutput) {
+                    continue;
+                }
+            }
+        }
+
+        syncmsg msg;
+        msg.data.id = ID_DONE;
+        msg.data.size = mtime;
+        RecordFileSent(lpath, rpath);
+        return WriteOrDie(lpath, rpath, &msg.data, sizeof(msg.data));
+    }
+
     bool SendLargeFile(const std::string& path, mode_t mode, const std::string& lpath,
-                       const std::string& rpath, unsigned mtime) {
+                       const std::string& rpath, unsigned mtime, bool compressed) {
+        if (compressed && HaveSendRecv2Brotli()) {
+            return SendLargeFileCompressed(path, mode, lpath, rpath, mtime);
+        }
+
         std::string path_and_mode = android::base::StringPrintf("%s,%d", path.c_str(), mode);
-        if (!SendRequest(ID_SEND, path_and_mode)) {
-            Error("failed to send ID_SEND message '%s': %s", path_and_mode.c_str(),
+        if (!SendRequest(ID_SEND_V1, path_and_mode)) {
+            Error("failed to send ID_SEND_V1 message '%s': %s", path_and_mode.c_str(),
                   strerror(errno));
             return false;
         }
@@ -489,7 +632,7 @@
         uint64_t total_size = st.st_size;
         uint64_t bytes_copied = 0;
 
-        unique_fd lfd(adb_open(lpath.c_str(), O_RDONLY));
+        unique_fd lfd(adb_open(lpath.c_str(), O_RDONLY | O_CLOEXEC));
         if (lfd < 0) {
             Error("opening '%s' locally failed: %s", lpath.c_str(), strerror(errno));
             return false;
@@ -497,8 +640,9 @@
 
         syncsendbuf sbuf;
         sbuf.id = ID_DATA;
+
         while (true) {
-            int bytes_read = adb_read(lfd, sbuf.data, max - sizeof(SyncRequest));
+            int bytes_read = adb_read(lfd, sbuf.data, max);
             if (bytes_read == -1) {
                 Error("reading '%s' locally failed: %s", lpath.c_str(), strerror(errno));
                 return false;
@@ -511,7 +655,6 @@
 
             RecordBytesTransferred(bytes_read);
             bytes_copied += bytes_read;
-
             ReportProgress(rpath, bytes_copied, total_size);
         }
 
@@ -695,6 +838,8 @@
     FeatureSet features_;
     bool have_stat_v2_;
     bool have_ls_v2_;
+    bool have_sendrecv_v2_;
+    bool have_sendrecv_v2_brotli_;
 
     TransferLedger global_ledger_;
     TransferLedger current_ledger_;
@@ -776,7 +921,7 @@
 }
 
 static bool sync_send(SyncConnection& sc, const std::string& lpath, const std::string& rpath,
-                      unsigned mtime, mode_t mode, bool sync) {
+                      unsigned mtime, mode_t mode, bool sync, bool compressed) {
     if (sync) {
         struct stat st;
         if (sync_lstat(sc, rpath, &st)) {
@@ -819,16 +964,16 @@
             return false;
         }
     } else {
-        if (!sc.SendLargeFile(rpath, mode, lpath, rpath, mtime)) {
+        if (!sc.SendLargeFile(rpath, mode, lpath, rpath, mtime, compressed)) {
             return false;
         }
     }
     return sc.ReadAcknowledgements();
 }
 
-static bool sync_recv(SyncConnection& sc, const char* rpath, const char* lpath,
-                      const char* name, uint64_t expected_size) {
-    if (!sc.SendRequest(ID_RECV, rpath)) return false;
+static bool sync_recv_v1(SyncConnection& sc, const char* rpath, const char* lpath, const char* name,
+                         uint64_t expected_size) {
+    if (!sc.SendRequest(ID_RECV_V1, rpath)) return false;
 
     adb_unlink(lpath);
     unique_fd lfd(adb_creat(lpath, 0644));
@@ -881,6 +1026,114 @@
     return true;
 }
 
+static bool sync_recv_v2(SyncConnection& sc, const char* rpath, const char* lpath, const char* name,
+                         uint64_t expected_size) {
+    if (!sc.SendRecv2(rpath)) return false;
+
+    adb_unlink(lpath);
+    unique_fd lfd(adb_creat(lpath, 0644));
+    if (lfd < 0) {
+        sc.Error("cannot create '%s': %s", lpath, strerror(errno));
+        return false;
+    }
+
+    uint64_t bytes_copied = 0;
+
+    Block buffer(SYNC_DATA_MAX);
+    BrotliDecoder decoder(std::span(buffer.data(), buffer.size()));
+    bool reading = true;
+    while (reading) {
+        syncmsg msg;
+        if (!ReadFdExactly(sc.fd, &msg.data, sizeof(msg.data))) {
+            adb_unlink(lpath);
+            return false;
+        }
+
+        if (msg.data.id == ID_DONE) {
+            adb_unlink(lpath);
+            sc.Error("unexpected ID_DONE");
+            return false;
+        }
+
+        if (msg.data.id != ID_DATA) {
+            adb_unlink(lpath);
+            sc.ReportCopyFailure(rpath, lpath, msg);
+            return false;
+        }
+
+        if (msg.data.size > sc.max) {
+            sc.Error("msg.data.size too large: %u (max %zu)", msg.data.size, sc.max);
+            adb_unlink(lpath);
+            return false;
+        }
+
+        Block block(msg.data.size);
+        if (!ReadFdExactly(sc.fd, block.data(), msg.data.size)) {
+            adb_unlink(lpath);
+            return false;
+        }
+        decoder.Append(std::move(block));
+
+        while (true) {
+            std::span<char> output;
+            BrotliDecodeResult result = decoder.Decode(&output);
+
+            if (result == BrotliDecodeResult::Error) {
+                sc.Error("decompress failed");
+                adb_unlink(lpath);
+                return false;
+            }
+
+            if (!output.empty()) {
+                if (!WriteFdExactly(lfd, output.data(), output.size())) {
+                    sc.Error("cannot write '%s': %s", lpath, strerror(errno));
+                    adb_unlink(lpath);
+                    return false;
+                }
+            }
+
+            bytes_copied += output.size();
+
+            sc.RecordBytesTransferred(msg.data.size);
+            sc.ReportProgress(name != nullptr ? name : rpath, bytes_copied, expected_size);
+
+            if (result == BrotliDecodeResult::NeedInput) {
+                break;
+            } else if (result == BrotliDecodeResult::MoreOutput) {
+                continue;
+            } else if (result == BrotliDecodeResult::Done) {
+                reading = false;
+                break;
+            } else {
+                LOG(FATAL) << "invalid BrotliDecodeResult: " << static_cast<int>(result);
+            }
+        }
+    }
+
+    syncmsg msg;
+    if (!ReadFdExactly(sc.fd, &msg.data, sizeof(msg.data))) {
+        sc.Error("failed to read ID_DONE");
+        return false;
+    }
+
+    if (msg.data.id != ID_DONE) {
+        sc.Error("unexpected message after transfer: id = %d (expected ID_DONE)", msg.data.id);
+        return false;
+    }
+
+    sc.RecordFilesTransferred(1);
+    return true;
+}
+
+static bool sync_recv(SyncConnection& sc, const char* rpath, const char* lpath, const char* name,
+                      uint64_t expected_size, bool compressed) {
+    if (sc.HaveSendRecv2() && compressed) {
+        return sync_recv_v2(sc, rpath, lpath, name, expected_size);
+    } else {
+        return sync_recv_v1(sc, rpath, lpath, name, expected_size);
+    }
+}
+
 bool do_sync_ls(const char* path) {
     SyncConnection sc;
     if (!sc.IsValid()) return false;
@@ -956,9 +1209,8 @@
     return true;
 }
 
-static bool copy_local_dir_remote(SyncConnection& sc, std::string lpath,
-                                  std::string rpath, bool check_timestamps,
-                                  bool list_only) {
+static bool copy_local_dir_remote(SyncConnection& sc, std::string lpath, std::string rpath,
+                                  bool check_timestamps, bool list_only, bool compressed) {
     sc.NewTransfer();
 
     // Make sure that both directory paths end in a slash.
@@ -1040,7 +1292,7 @@
             if (list_only) {
                 sc.Println("would push: %s -> %s", ci.lpath.c_str(), ci.rpath.c_str());
             } else {
-                if (!sync_send(sc, ci.lpath, ci.rpath, ci.time, ci.mode, false)) {
+                if (!sync_send(sc, ci.lpath, ci.rpath, ci.time, ci.mode, false, compressed)) {
                     return false;
                 }
             }
@@ -1055,7 +1307,8 @@
     return success;
 }
 
-bool do_sync_push(const std::vector<const char*>& srcs, const char* dst, bool sync) {
+bool do_sync_push(const std::vector<const char*>& srcs, const char* dst, bool sync,
+                  bool compressed) {
     SyncConnection sc;
     if (!sc.IsValid()) return false;
 
@@ -1120,7 +1373,7 @@
                 dst_dir.append(android::base::Basename(src_path));
             }
 
-            success &= copy_local_dir_remote(sc, src_path, dst_dir, sync, false);
+            success &= copy_local_dir_remote(sc, src_path, dst_dir, sync, false, compressed);
             continue;
         } else if (!should_push_file(st.st_mode)) {
             sc.Warning("skipping special file '%s' (mode = 0o%o)", src_path, st.st_mode);
@@ -1141,7 +1394,7 @@
 
         sc.NewTransfer();
         sc.SetExpectedTotalBytes(st.st_size);
-        success &= sync_send(sc, src_path, dst_path, st.st_mtime, st.st_mode, sync);
+        success &= sync_send(sc, src_path, dst_path, st.st_mtime, st.st_mode, sync, compressed);
         sc.ReportTransferRate(src_path, TransferDirection::push);
     }
 
@@ -1226,8 +1479,8 @@
     return r1 ? r1 : r2;
 }
 
-static bool copy_remote_dir_local(SyncConnection& sc, std::string rpath,
-                                  std::string lpath, bool copy_attrs) {
+static bool copy_remote_dir_local(SyncConnection& sc, std::string rpath, std::string lpath,
+                                  bool copy_attrs, bool compressed) {
     sc.NewTransfer();
 
     // Make sure that both directory paths end in a slash.
@@ -1257,7 +1510,7 @@
                 continue;
             }
 
-            if (!sync_recv(sc, ci.rpath.c_str(), ci.lpath.c_str(), nullptr, ci.size)) {
+            if (!sync_recv(sc, ci.rpath.c_str(), ci.lpath.c_str(), nullptr, ci.size, compressed)) {
                 return false;
             }
 
@@ -1274,8 +1527,8 @@
     return true;
 }
 
-bool do_sync_pull(const std::vector<const char*>& srcs, const char* dst,
-                  bool copy_attrs, const char* name) {
+bool do_sync_pull(const std::vector<const char*>& srcs, const char* dst, bool copy_attrs,
+                  bool compressed, const char* name) {
     SyncConnection sc;
     if (!sc.IsValid()) return false;
 
@@ -1349,7 +1602,7 @@
                 dst_dir.append(android::base::Basename(src_path));
             }
 
-            success &= copy_remote_dir_local(sc, src_path, dst_dir, copy_attrs);
+            success &= copy_remote_dir_local(sc, src_path, dst_dir, copy_attrs, compressed);
             continue;
         } else if (!should_pull_file(src_st.st_mode)) {
             sc.Warning("skipping special file '%s' (mode = 0o%o)", src_path, src_st.st_mode);
@@ -1368,7 +1621,7 @@
 
         sc.NewTransfer();
         sc.SetExpectedTotalBytes(src_st.st_size);
-        if (!sync_recv(sc, src_path, dst_path, name, src_st.st_size)) {
+        if (!sync_recv(sc, src_path, dst_path, name, src_st.st_size, compressed)) {
             success = false;
             continue;
         }
@@ -1384,11 +1637,12 @@
     return success;
 }
 
-bool do_sync_sync(const std::string& lpath, const std::string& rpath, bool list_only) {
+bool do_sync_sync(const std::string& lpath, const std::string& rpath, bool list_only,
+                  bool compressed) {
     SyncConnection sc;
     if (!sc.IsValid()) return false;
 
-    bool success = copy_local_dir_remote(sc, lpath, rpath, true, list_only);
+    bool success = copy_local_dir_remote(sc, lpath, rpath, true, list_only, compressed);
     if (!list_only) {
         sc.ReportOverallTransferRate(TransferDirection::push);
     }
diff --git a/adb/client/file_sync_client.h b/adb/client/file_sync_client.h
index df7f14c..de3f192 100644
--- a/adb/client/file_sync_client.h
+++ b/adb/client/file_sync_client.h
@@ -20,8 +20,10 @@
 #include <vector>
 
 bool do_sync_ls(const char* path);
-bool do_sync_push(const std::vector<const char*>& srcs, const char* dst, bool sync);
+bool do_sync_push(const std::vector<const char*>& srcs, const char* dst, bool sync,
+                  bool compressed);
 bool do_sync_pull(const std::vector<const char*>& srcs, const char* dst, bool copy_attrs,
-                  const char* name = nullptr);
+                  bool compressed, const char* name = nullptr);
 
-bool do_sync_sync(const std::string& lpath, const std::string& rpath, bool list_only);
+bool do_sync_sync(const std::string& lpath, const std::string& rpath, bool list_only,
+                  bool compressed);
diff --git a/adb/client/incremental.cpp b/adb/client/incremental.cpp
index a9e65dc..9765292 100644
--- a/adb/client/incremental.cpp
+++ b/adb/client/incremental.cpp
@@ -90,38 +90,58 @@
     return total_tree_block_count * INCFS_DATA_FILE_BLOCK_SIZE;
 }
 
-// Base64-encode signature bytes. Keeping fd at the position of start of verity tree.
-static std::pair<unique_fd, std::string> read_and_encode_signature(Size file_size,
-                                                                   std::string signature_file) {
+// Read, verify and return the signature bytes. Keeping fd at the position of start of verity tree.
+static std::pair<unique_fd, std::vector<char>> read_signature(Size file_size,
+                                                              std::string signature_file,
+                                                              bool silent) {
     signature_file += IDSIG;
 
     struct stat st;
     if (stat(signature_file.c_str(), &st)) {
-        fprintf(stderr, "Failed to stat signature file %s. Abort.\n", signature_file.c_str());
+        if (!silent) {
+            fprintf(stderr, "Failed to stat signature file %s. Abort.\n", signature_file.c_str());
+        }
         return {};
     }
 
     unique_fd fd(adb_open(signature_file.c_str(), O_RDONLY | O_CLOEXEC));
     if (fd < 0) {
-        fprintf(stderr, "Failed to open signature file: %s. Abort.\n", signature_file.c_str());
+        if (!silent) {
+            fprintf(stderr, "Failed to open signature file: %s. Abort.\n", signature_file.c_str());
+        }
         return {};
     }
 
     auto [signature, tree_size] = read_id_sig_headers(fd);
     if (auto expected = verity_tree_size_for_file(file_size); tree_size != expected) {
-        fprintf(stderr,
-                "Verity tree size mismatch in signature file: %s [was %lld, expected %lld].\n",
-                signature_file.c_str(), (long long)tree_size, (long long)expected);
+        if (!silent) {
+            fprintf(stderr,
+                    "Verity tree size mismatch in signature file: %s [was %lld, expected %lld].\n",
+                    signature_file.c_str(), (long long)tree_size, (long long)expected);
+        }
+        return {};
+    }
+
+    return {std::move(fd), std::move(signature)};
+}
+
+// Base64-encode signature bytes. Keeping fd at the position of start of verity tree.
+static std::pair<unique_fd, std::string> read_and_encode_signature(Size file_size,
+                                                                   std::string signature_file,
+                                                                   bool silent) {
+    auto [fd, signature] = read_signature(file_size, std::move(signature_file), silent);
+    if (!fd.ok()) {
         return {};
     }
 
     size_t base64_len = 0;
     if (!EVP_EncodedLength(&base64_len, signature.size())) {
-        fprintf(stderr, "Fail to estimate base64 encoded length. Abort.\n");
+        if (!silent) {
+            fprintf(stderr, "Fail to estimate base64 encoded length. Abort.\n");
+        }
         return {};
     }
-    std::string encoded_signature;
-    encoded_signature.resize(base64_len);
+    std::string encoded_signature(base64_len, '\0');
     encoded_signature.resize(EVP_EncodeBlock((uint8_t*)encoded_signature.data(),
                                              (const uint8_t*)signature.data(), signature.size()));
 
@@ -130,7 +150,7 @@
 
 // Send install-incremental to the device along with properly configured file descriptors in
 // streaming format. Once connection established, send all fs-verity tree bytes.
-static unique_fd start_install(const std::vector<std::string>& files) {
+static unique_fd start_install(const Files& files, bool silent) {
     std::vector<std::string> command_args{"package", "install-incremental"};
 
     // fd's with positions at the beginning of fs-verity
@@ -141,11 +161,13 @@
 
         struct stat st;
         if (stat(file.c_str(), &st)) {
-            fprintf(stderr, "Failed to stat input file %s. Abort.\n", file.c_str());
+            if (!silent) {
+                fprintf(stderr, "Failed to stat input file %s. Abort.\n", file.c_str());
+            }
             return {};
         }
 
-        auto [signature_fd, signature] = read_and_encode_signature(st.st_size, file);
+        auto [signature_fd, signature] = read_and_encode_signature(st.st_size, file, silent);
         if (!signature_fd.ok()) {
             return {};
         }
@@ -161,15 +183,19 @@
     std::string error;
     auto connection_fd = unique_fd(send_abb_exec_command(command_args, &error));
     if (connection_fd < 0) {
-        fprintf(stderr, "Failed to run: %s, error: %s\n",
-                android::base::Join(command_args, " ").c_str(), error.c_str());
+        if (!silent) {
+            fprintf(stderr, "Failed to run: %s, error: %s\n",
+                    android::base::Join(command_args, " ").c_str(), error.c_str());
+        }
         return {};
     }
 
     // Pushing verity trees for all installation files.
     for (auto&& local_fd : signature_fds) {
         if (!copy_to_file(local_fd.get(), connection_fd.get())) {
-            fprintf(stderr, "Failed to stream tree bytes: %s. Abort.\n", strerror(errno));
+            if (!silent) {
+                fprintf(stderr, "Failed to stream tree bytes: %s. Abort.\n", strerror(errno));
+            }
             return {};
         }
     }
@@ -179,30 +205,45 @@
 
 }  // namespace
 
-std::optional<Process> install(std::vector<std::string> files) {
-    auto connection_fd = start_install(files);
+bool can_install(const Files& files) {
+    for (const auto& file : files) {
+        struct stat st;
+        if (stat(file.c_str(), &st)) {
+            return false;
+        }
+
+        auto [fd, _] = read_signature(st.st_size, file, true);
+        if (!fd.ok()) {
+            return false;
+        }
+    }
+    return true;
+}
+
+std::optional<Process> install(const Files& files, bool silent) {
+    auto connection_fd = start_install(files, silent);
     if (connection_fd < 0) {
-        fprintf(stderr, "adb: failed to initiate installation on device.\n");
+        if (!silent) {
+            fprintf(stderr, "adb: failed to initiate installation on device.\n");
+        }
         return {};
     }
 
     std::string adb_path = android::base::GetExecutablePath();
 
-    auto osh = adb_get_os_handle(connection_fd.get());
-#ifdef _WIN32
-    auto fd_param = std::to_string(reinterpret_cast<intptr_t>(osh));
-#else /* !_WIN32 a.k.a. Unix */
+    auto osh = cast_handle_to_int(adb_get_os_handle(connection_fd.get()));
     auto fd_param = std::to_string(osh);
-#endif
 
     // pipe for child process to write output
     int print_fds[2];
     if (adb_socketpair(print_fds) != 0) {
-        fprintf(stderr, "Failed to create socket pair for child to print to parent\n");
+        if (!silent) {
+            fprintf(stderr, "Failed to create socket pair for child to print to parent\n");
+        }
         return {};
     }
     auto [pipe_read_fd, pipe_write_fd] = print_fds;
-    auto pipe_write_fd_param = std::to_string(intptr_t(adb_get_os_handle(pipe_write_fd)));
+    auto pipe_write_fd_param = std::to_string(cast_handle_to_int(adb_get_os_handle(pipe_write_fd)));
     close_on_exec(pipe_read_fd);
 
     std::vector<std::string> args(std::move(files));
@@ -210,7 +251,9 @@
     auto child =
             adb_launch_process(adb_path, std::move(args), {connection_fd.get(), pipe_write_fd});
     if (!child) {
-        fprintf(stderr, "adb: failed to fork: %s\n", strerror(errno));
+        if (!silent) {
+            fprintf(stderr, "adb: failed to fork: %s\n", strerror(errno));
+        }
         return {};
     }
 
diff --git a/adb/client/incremental.h b/adb/client/incremental.h
index 731e6fb..1fb1e0b 100644
--- a/adb/client/incremental.h
+++ b/adb/client/incremental.h
@@ -25,7 +25,10 @@
 
 namespace incremental {
 
-std::optional<Process> install(std::vector<std::string> files);
+using Files = std::vector<std::string>;
+
+bool can_install(const Files& files);
+std::optional<Process> install(const Files& files, bool silent);
 
 enum class Result { Success, Failure, None };
 Result wait_for_installation(int read_fd);
diff --git a/adb/client/incremental_utils.cpp b/adb/client/incremental_utils.cpp
index caadb26..fa501e4 100644
--- a/adb/client/incremental_utils.cpp
+++ b/adb/client/incremental_utils.cpp
@@ -23,6 +23,7 @@
 #include <ziparchive/zip_archive.h>
 #include <ziparchive/zip_writer.h>
 
+#include <array>
 #include <cinttypes>
 #include <numeric>
 #include <unordered_set>
@@ -30,6 +31,8 @@
 #include "adb_trace.h"
 #include "sysdeps.h"
 
+using namespace std::literals;
+
 static constexpr int kBlockSize = 4096;
 
 static constexpr inline int32_t offsetToBlockIndex(int64_t offset) {
@@ -217,16 +220,24 @@
     return {zip, std::move(mapping)};
 }
 
-// TODO(b/151676293): avoid using libziparchive as it reads local file headers
-// which causes additional performance cost. Instead, only read from central directory.
 static std::vector<int32_t> InstallationPriorityBlocks(int fd, int64_t fileSize) {
+    static constexpr std::array<std::string_view, 3> additional_matches = {
+            "resources.arsc"sv, "AndroidManifest.xml"sv, "classes.dex"sv};
     auto [zip, _] = openZipArchive(fd, fileSize);
     if (!zip) {
         return {};
     }
 
+    auto matcher = [](std::string_view entry_name) {
+        if (entry_name.starts_with("lib/"sv) && entry_name.ends_with(".so"sv)) {
+            return true;
+        }
+        return std::any_of(additional_matches.begin(), additional_matches.end(),
+                           [entry_name](std::string_view i) { return i == entry_name; });
+    };
+
     void* cookie = nullptr;
-    if (StartIteration(zip, &cookie) != 0) {
+    if (StartIteration(zip, &cookie, std::move(matcher)) != 0) {
         D("%s failed at StartIteration: %d", __func__, errno);
         return {};
     }
@@ -235,8 +246,12 @@
     ZipEntry entry;
     std::string_view entryName;
     while (Next(cookie, &entry, &entryName) == 0) {
-        if (entryName == "resources.arsc" || entryName == "AndroidManifest.xml" ||
-            entryName.starts_with("lib/")) {
+        if (entryName == "classes.dex"sv) {
+            // Only the head is needed for installation
+            int32_t startBlockIndex = offsetToBlockIndex(entry.offset);
+            appendBlocks(startBlockIndex, 1, &installationPriorityBlocks);
+            D("\tadding to priority blocks: '%.*s' 1", (int)entryName.size(), entryName.data());
+        } else {
             // Full entries are needed for installation
             off64_t entryStartOffset = entry.offset;
             off64_t entryEndOffset =
@@ -248,11 +263,8 @@
             int32_t endBlockIndex = offsetToBlockIndex(entryEndOffset);
             int32_t numNewBlocks = endBlockIndex - startBlockIndex + 1;
             appendBlocks(startBlockIndex, numNewBlocks, &installationPriorityBlocks);
-            D("\tadding to priority blocks: '%.*s'", (int)entryName.size(), entryName.data());
-        } else if (entryName == "classes.dex") {
-            // Only the head is needed for installation
-            int32_t startBlockIndex = offsetToBlockIndex(entry.offset);
-            appendBlocks(startBlockIndex, 1, &installationPriorityBlocks);
+            D("\tadding to priority blocks: '%.*s' (%d)", (int)entryName.size(), entryName.data(),
+              numNewBlocks);
         }
     }
 
diff --git a/adb/client/main.cpp b/adb/client/main.cpp
index a85a18c..33e0716 100644
--- a/adb/client/main.cpp
+++ b/adb/client/main.cpp
@@ -139,9 +139,10 @@
     auto start = std::chrono::steady_clock::now();
 
     // If we told a previous adb server to quit because of version mismatch, we can get to this
-    // point before it's finished exiting. Retry for a while to give it some time.
-    while (install_listener(socket_spec, "*smartsocket*", nullptr, 0, nullptr, &error) !=
-           INSTALL_STATUS_OK) {
+    // point before it's finished exiting. Retry for a while to give it some time. Don't actually
+    // accept any connections until adb_wait_for_device_initialization finishes below.
+    while (install_listener(socket_spec, "*smartsocket*", nullptr, INSTALL_LISTENER_DISABLED,
+                            nullptr, &error) != INSTALL_STATUS_OK) {
         if (std::chrono::steady_clock::now() - start > 0.5s) {
             LOG(FATAL) << "could not install *smartsocket* listener: " << error;
         }
@@ -162,12 +163,14 @@
             PLOG(FATAL) << "setsid() failed";
         }
 #endif
+    }
 
-        // Wait for the USB scan to complete before notifying the parent that we're up.
-        // We need to perform this in a thread, because we would otherwise block the event loop.
-        std::thread notify_thread([ack_reply_fd]() {
-            adb_wait_for_device_initialization();
+    // Wait for the USB scan to complete before notifying the parent that we're up.
+    // We need to perform this in a thread, because we would otherwise block the event loop.
+    std::thread notify_thread([ack_reply_fd]() {
+        adb_wait_for_device_initialization();
 
+        if (ack_reply_fd >= 0) {
             // Any error output written to stderr now goes to adb.log. We could
             // keep around a copy of the stderr fd and use that to write any errors
             // encountered by the following code, but that is probably overkill.
@@ -193,9 +196,13 @@
             }
             unix_close(ack_reply_fd);
 #endif
-        });
-        notify_thread.detach();
-    }
+        }
+        // We don't accept() client connections until this point: this way, clients
+        // can't see wonky state early in startup even if they're connecting directly
+        // to the server instead of going through the adb program.
+        fdevent_run_on_main_thread([] { enable_daemon_sockets(); });
+    });
+    notify_thread.detach();
 
 #if defined(__linux__)
     // Write our location to .android/adb.$PORT, so that older clients can exec us.
diff --git a/adb/client/pairing/pairing_client.cpp b/adb/client/pairing/pairing_client.cpp
index 04bbceb..937a5bd 100644
--- a/adb/client/pairing/pairing_client.cpp
+++ b/adb/client/pairing/pairing_client.cpp
@@ -141,11 +141,7 @@
                                           cert_.size(), priv_key_.data(), priv_key_.size()));
     CHECK(connection_);
 
-#ifdef _WIN32
     int osh = cast_handle_to_int(adb_get_os_handle(fd.release()));
-#else
-    int osh = adb_get_os_handle(fd.release());
-#endif
     if (!pairing_connection_start(connection_.get(), osh, OnPairingResult, this)) {
         LOG(ERROR) << "PairingClient failed to start the PairingConnection";
         state_ = State::Stopped;
diff --git a/adb/daemon/file_sync_service.cpp b/adb/daemon/file_sync_service.cpp
index edf5683..07f6e65 100644
--- a/adb/daemon/file_sync_service.cpp
+++ b/adb/daemon/file_sync_service.cpp
@@ -32,6 +32,8 @@
 #include <utime.h>
 
 #include <memory>
+#include <optional>
+#include <span>
 #include <string>
 #include <vector>
 
@@ -55,10 +57,12 @@
 #include "adb_io.h"
 #include "adb_trace.h"
 #include "adb_utils.h"
+#include "brotli_utils.h"
 #include "file_sync_protocol.h"
 #include "security_log_tags.h"
 #include "sysdeps/errno.h"
 
+using android::base::borrowed_fd;
 using android::base::Dirname;
 using android::base::StringPrintf;
 
@@ -249,7 +253,7 @@
 // Make sure that SendFail from adb_io.cpp isn't accidentally used in this file.
 #pragma GCC poison SendFail
 
-static bool SendSyncFail(int fd, const std::string& reason) {
+static bool SendSyncFail(borrowed_fd fd, const std::string& reason) {
     D("sync: failure: %s", reason.c_str());
 
     syncmsg msg;
@@ -258,13 +262,89 @@
     return WriteFdExactly(fd, &msg.data, sizeof(msg.data)) && WriteFdExactly(fd, reason);
 }
 
-static bool SendSyncFailErrno(int fd, const std::string& reason) {
+static bool SendSyncFailErrno(borrowed_fd fd, const std::string& reason) {
     return SendSyncFail(fd, StringPrintf("%s: %s", reason.c_str(), strerror(errno)));
 }
 
-static bool handle_send_file(int s, const char* path, uint32_t* timestamp, uid_t uid, gid_t gid,
-                             uint64_t capabilities, mode_t mode, std::vector<char>& buffer,
-                             bool do_unlink) {
+static bool handle_send_file_compressed(borrowed_fd s, unique_fd fd, uint32_t* timestamp) {
+    syncmsg msg;
+    Block decode_buffer(SYNC_DATA_MAX);
+    BrotliDecoder decoder(std::span(decode_buffer.data(), decode_buffer.size()));
+    while (true) {
+        if (!ReadFdExactly(s, &msg.data, sizeof(msg.data))) return false;
+
+        if (msg.data.id != ID_DATA) {
+            if (msg.data.id == ID_DONE) {
+                *timestamp = msg.data.size;
+                return true;
+            }
+            SendSyncFail(s, "invalid data message");
+            return false;
+        }
+
+        Block block(msg.data.size);
+        if (!ReadFdExactly(s, block.data(), msg.data.size)) return false;
+        decoder.Append(std::move(block));
+
+        while (true) {
+            std::span<char> output;
+            BrotliDecodeResult result = decoder.Decode(&output);
+            if (result == BrotliDecodeResult::Error) {
+                SendSyncFailErrno(s, "decompress failed");
+                return false;
+            }
+
+            if (!WriteFdExactly(fd, output.data(), output.size())) {
+                SendSyncFailErrno(s, "write failed");
+                return false;
+            }
+
+            if (result == BrotliDecodeResult::NeedInput) {
+                break;
+            } else if (result == BrotliDecodeResult::MoreOutput) {
+                continue;
+            } else if (result == BrotliDecodeResult::Done) {
+                break;
+            } else {
+                LOG(FATAL) << "invalid BrotliDecodeResult: " << static_cast<int>(result);
+            }
+        }
+    }
+
+    __builtin_unreachable();
+}
+
+static bool handle_send_file_uncompressed(borrowed_fd s, unique_fd fd, uint32_t* timestamp,
+                                          std::vector<char>& buffer) {
+    syncmsg msg;
+
+    while (true) {
+        if (!ReadFdExactly(s, &msg.data, sizeof(msg.data))) return false;
+
+        if (msg.data.id != ID_DATA) {
+            if (msg.data.id == ID_DONE) {
+                *timestamp = msg.data.size;
+                return true;
+            }
+            SendSyncFail(s, "invalid data message");
+            return false;
+        }
+
+        if (msg.data.size > buffer.size()) {  // TODO: resize buffer?
+            SendSyncFail(s, "oversize data message");
+            return false;
+        }
+        if (!ReadFdExactly(s, &buffer[0], msg.data.size)) return false;
+        if (!WriteFdExactly(fd, &buffer[0], msg.data.size)) {
+            SendSyncFailErrno(s, "write failed");
+            return false;
+        }
+    }
+}
+
+static bool handle_send_file(borrowed_fd s, const char* path, uint32_t* timestamp, uid_t uid,
+                             gid_t gid, uint64_t capabilities, mode_t mode, bool compressed,
+                             std::vector<char>& buffer, bool do_unlink) {
     int rc;
     syncmsg msg;
 
@@ -302,45 +382,33 @@
         fchmod(fd.get(), mode);
     }
 
-    rc = posix_fadvise(fd.get(), 0, 0,
-                       POSIX_FADV_SEQUENTIAL | POSIX_FADV_NOREUSE | POSIX_FADV_WILLNEED);
-    if (rc != 0) {
-        D("[ Failed to fadvise: %s ]", strerror(rc));
-    }
-
-    while (true) {
-        if (!ReadFdExactly(s, &msg.data, sizeof(msg.data))) goto fail;
-
-        if (msg.data.id != ID_DATA) {
-            if (msg.data.id == ID_DONE) {
-                *timestamp = msg.data.size;
-                break;
-            }
-            SendSyncFail(s, "invalid data message");
-            goto abort;
+    {
+        rc = posix_fadvise(fd.get(), 0, 0,
+                           POSIX_FADV_SEQUENTIAL | POSIX_FADV_NOREUSE | POSIX_FADV_WILLNEED);
+        if (rc != 0) {
+            D("[ Failed to fadvise: %s ]", strerror(rc));
         }
 
-        if (msg.data.size > buffer.size()) {  // TODO: resize buffer?
-            SendSyncFail(s, "oversize data message");
-            goto abort;
+        bool result;
+        if (compressed) {
+            result = handle_send_file_compressed(s, std::move(fd), timestamp);
+        } else {
+            result = handle_send_file_uncompressed(s, std::move(fd), timestamp, buffer);
         }
 
-        if (!ReadFdExactly(s, &buffer[0], msg.data.size)) goto abort;
-
-        if (!WriteFdExactly(fd.get(), &buffer[0], msg.data.size)) {
-            SendSyncFailErrno(s, "write failed");
+        if (!result) {
             goto fail;
         }
-    }
 
-    if (!update_capabilities(path, capabilities)) {
-        SendSyncFailErrno(s, "update_capabilities failed");
-        goto fail;
-    }
+        if (!update_capabilities(path, capabilities)) {
+            SendSyncFailErrno(s, "update_capabilities failed");
+            goto fail;
+        }
 
-    msg.status.id = ID_OKAY;
-    msg.status.msglen = 0;
-    return WriteFdExactly(s, &msg.status, sizeof(msg.status));
+        msg.status.id = ID_OKAY;
+        msg.status.msglen = 0;
+        return WriteFdExactly(s, &msg.status, sizeof(msg.status));
+    }
 
 fail:
     // If there's a problem on the device, we'll send an ID_FAIL message and
@@ -371,7 +439,6 @@
         if (!ReadFdExactly(s, &buffer[0], msg.data.size)) break;
     }
 
-abort:
     if (do_unlink) adb_unlink(path);
     return false;
 }
@@ -432,23 +499,8 @@
 }
 #endif
 
-static bool do_send(int s, const std::string& spec, std::vector<char>& buffer) {
-    // 'spec' is of the form "/some/path,0755". Break it up.
-    size_t comma = spec.find_last_of(',');
-    if (comma == std::string::npos) {
-        SendSyncFail(s, "missing , in ID_SEND");
-        return false;
-    }
-
-    std::string path = spec.substr(0, comma);
-
-    errno = 0;
-    mode_t mode = strtoul(spec.substr(comma + 1).c_str(), nullptr, 0);
-    if (errno != 0) {
-        SendSyncFail(s, "bad mode");
-        return false;
-    }
-
+static bool send_impl(int s, const std::string& path, mode_t mode, bool compressed,
+                      std::vector<char>& buffer) {
     // Don't delete files before copying if they are not "regular" or symlinks.
     struct stat st;
     bool do_unlink = (lstat(path.c_str(), &st) == -1) || S_ISREG(st.st_mode) ||
@@ -474,8 +526,8 @@
             adbd_fs_config(path.c_str(), 0, nullptr, &uid, &gid, &mode, &capabilities);
         }
 
-        result = handle_send_file(s, path.c_str(), &timestamp, uid, gid, capabilities, mode, buffer,
-                                  do_unlink);
+        result = handle_send_file(s, path.c_str(), &timestamp, uid, gid, capabilities, mode,
+                                  compressed, buffer, do_unlink);
     }
 
     if (!result) {
@@ -491,7 +543,125 @@
     return true;
 }
 
-static bool do_recv(int s, const char* path, std::vector<char>& buffer) {
+static bool do_send_v1(int s, const std::string& spec, std::vector<char>& buffer) {
+    // 'spec' is of the form "/some/path,0755". Break it up.
+    size_t comma = spec.find_last_of(',');
+    if (comma == std::string::npos) {
+        SendSyncFail(s, "missing , in ID_SEND_V1");
+        return false;
+    }
+
+    std::string path = spec.substr(0, comma);
+
+    errno = 0;
+    mode_t mode = strtoul(spec.substr(comma + 1).c_str(), nullptr, 0);
+    if (errno != 0) {
+        SendSyncFail(s, "bad mode");
+        return false;
+    }
+
+    return send_impl(s, path, mode, false, buffer);
+}
+
+static bool do_send_v2(int s, const std::string& path, std::vector<char>& buffer) {
+    // Read the setup packet.
+    syncmsg msg;
+    int rc = ReadFdExactly(s, &msg.send_v2_setup, sizeof(msg.send_v2_setup));
+    if (rc == 0) {
+        LOG(ERROR) << "failed to read send_v2 setup packet: EOF";
+        return false;
+    } else if (rc < 0) {
+        PLOG(ERROR) << "failed to read send_v2 setup packet";
+    }
+
+    bool compressed = false;
+    if (msg.send_v2_setup.flags & kSyncFlagBrotli) {
+        msg.send_v2_setup.flags &= ~kSyncFlagBrotli;
+        compressed = true;
+    }
+    if (msg.send_v2_setup.flags) {
+        SendSyncFail(s, android::base::StringPrintf("unknown flags: %d", msg.send_v2_setup.flags));
+        return false;
+    }
+
+    errno = 0;
+    return send_impl(s, path, msg.send_v2_setup.mode, compressed, buffer);
+}
+
+static bool recv_uncompressed(borrowed_fd s, unique_fd fd, std::vector<char>& buffer) {
+    syncmsg msg;
+    msg.data.id = ID_DATA;
+    std::optional<BrotliEncoder<SYNC_DATA_MAX>> encoder;
+    while (true) {
+        int r = adb_read(fd.get(), &buffer[0], buffer.size() - sizeof(msg.data));
+        if (r <= 0) {
+            if (r == 0) break;
+            SendSyncFailErrno(s, "read failed");
+            return false;
+        }
+        msg.data.size = r;
+
+        if (!WriteFdExactly(s, &msg.data, sizeof(msg.data)) || !WriteFdExactly(s, &buffer[0], r)) {
+            return false;
+        }
+    }
+
+    return true;
+}
+
+static bool recv_compressed(borrowed_fd s, unique_fd fd) {
+    syncmsg msg;
+    msg.data.id = ID_DATA;
+
+    BrotliEncoder<SYNC_DATA_MAX> encoder;
+
+    bool sending = true;
+    while (sending) {
+        Block input(SYNC_DATA_MAX);
+        int r = adb_read(fd.get(), input.data(), input.size());
+        if (r < 0) {
+            SendSyncFailErrno(s, "read failed");
+            return false;
+        }
+
+        if (r == 0) {
+            encoder.Finish();
+        } else {
+            input.resize(r);
+            encoder.Append(std::move(input));
+        }
+
+        while (true) {
+            Block output;
+            BrotliEncodeResult result = encoder.Encode(&output);
+            if (result == BrotliEncodeResult::Error) {
+                SendSyncFailErrno(s, "compress failed");
+                return false;
+            }
+
+            if (!output.empty()) {
+                msg.data.size = output.size();
+                if (!WriteFdExactly(s, &msg.data, sizeof(msg.data)) ||
+                    !WriteFdExactly(s, output.data(), output.size())) {
+                    return false;
+                }
+            }
+
+            if (result == BrotliEncodeResult::Done) {
+                sending = false;
+                break;
+            } else if (result == BrotliEncodeResult::NeedInput) {
+                break;
+            } else if (result == BrotliEncodeResult::MoreOutput) {
+                continue;
+            }
+        }
+    }
+
+    return true;
+}
+
+static bool recv_impl(borrowed_fd s, const char* path, bool compressed, std::vector<char>& buffer) {
     __android_log_security_bswrite(SEC_TAG_ADB_RECV_FILE, path);
 
     unique_fd fd(adb_open(path, O_RDONLY | O_CLOEXEC));
@@ -505,26 +675,51 @@
         D("[ Failed to fadvise: %s ]", strerror(rc));
     }
 
-    syncmsg msg;
-    msg.data.id = ID_DATA;
-    while (true) {
-        int r = adb_read(fd.get(), &buffer[0], buffer.size() - sizeof(msg.data));
-        if (r <= 0) {
-            if (r == 0) break;
-            SendSyncFailErrno(s, "read failed");
-            return false;
-        }
-        msg.data.size = r;
-        if (!WriteFdExactly(s, &msg.data, sizeof(msg.data)) || !WriteFdExactly(s, &buffer[0], r)) {
-            return false;
-        }
+    bool result;
+    if (compressed) {
+        result = recv_compressed(s, std::move(fd));
+    } else {
+        result = recv_uncompressed(s, std::move(fd), buffer);
     }
 
+    if (!result) {
+        return false;
+    }
+
+    syncmsg msg;
     msg.data.id = ID_DONE;
     msg.data.size = 0;
     return WriteFdExactly(s, &msg.data, sizeof(msg.data));
 }
 
+static bool do_recv_v1(borrowed_fd s, const char* path, std::vector<char>& buffer) {
+    return recv_impl(s, path, false, buffer);
+}
+
+static bool do_recv_v2(borrowed_fd s, const char* path, std::vector<char>& buffer) {
+    syncmsg msg;
+    // Read the setup packet.
+    int rc = ReadFdExactly(s, &msg.recv_v2_setup, sizeof(msg.recv_v2_setup));
+    if (rc == 0) {
+        LOG(ERROR) << "failed to read recv_v2 setup packet: EOF";
+        return false;
+    } else if (rc < 0) {
+        PLOG(ERROR) << "failed to read recv_v2 setup packet";
+    }
+
+    bool compressed = false;
+    if (msg.recv_v2_setup.flags & kSyncFlagBrotli) {
+        msg.recv_v2_setup.flags &= ~kSyncFlagBrotli;
+        compressed = true;
+    }
+    if (msg.recv_v2_setup.flags) {
+        SendSyncFail(s, android::base::StringPrintf("unknown flags: %d", msg.recv_v2_setup.flags));
+        return false;
+    }
+
+    return recv_impl(s, path, compressed, buffer);
+}
+
 static const char* sync_id_to_name(uint32_t id) {
   switch (id) {
     case ID_LSTAT_V1:
@@ -537,10 +732,14 @@
       return "list_v1";
     case ID_LIST_V2:
       return "list_v2";
-    case ID_SEND:
-      return "send";
-    case ID_RECV:
-      return "recv";
+    case ID_SEND_V1:
+        return "send_v1";
+    case ID_SEND_V2:
+        return "send_v2";
+    case ID_RECV_V1:
+        return "recv_v1";
+    case ID_RECV_V2:
+        return "recv_v2";
     case ID_QUIT:
         return "quit";
     default:
@@ -585,11 +784,17 @@
         case ID_LIST_V2:
             if (!do_list_v2(fd, name)) return false;
             break;
-        case ID_SEND:
-            if (!do_send(fd, name, buffer)) return false;
+        case ID_SEND_V1:
+            if (!do_send_v1(fd, name, buffer)) return false;
             break;
-        case ID_RECV:
-            if (!do_recv(fd, name, buffer)) return false;
+        case ID_SEND_V2:
+            if (!do_send_v2(fd, name, buffer)) return false;
+            break;
+        case ID_RECV_V1:
+            if (!do_recv_v1(fd, name, buffer)) return false;
+            break;
+        case ID_RECV_V2:
+            if (!do_recv_v2(fd, name, buffer)) return false;
             break;
         case ID_QUIT:
             return false;
diff --git a/adb/file_sync_protocol.h b/adb/file_sync_protocol.h
index 87ede0c..fd9a516 100644
--- a/adb/file_sync_protocol.h
+++ b/adb/file_sync_protocol.h
@@ -27,8 +27,10 @@
 #define ID_DENT_V1 MKID('D', 'E', 'N', 'T')
 #define ID_DENT_V2 MKID('D', 'N', 'T', '2')
 
-#define ID_SEND MKID('S', 'E', 'N', 'D')
-#define ID_RECV MKID('R', 'E', 'C', 'V')
+#define ID_SEND_V1 MKID('S', 'E', 'N', 'D')
+#define ID_SEND_V2 MKID('S', 'N', 'D', '2')
+#define ID_RECV_V1 MKID('R', 'E', 'C', 'V')
+#define ID_RECV_V2 MKID('R', 'C', 'V', '2')
 #define ID_DONE MKID('D', 'O', 'N', 'E')
 #define ID_DATA MKID('D', 'A', 'T', 'A')
 #define ID_OKAY MKID('O', 'K', 'A', 'Y')
@@ -87,6 +89,26 @@
     uint32_t namelen;
 };  // followed by `namelen` bytes of the name.
 
+enum SyncFlag : uint32_t {
+    kSyncFlagNone = 0,
+    kSyncFlagBrotli = 1,
+};
+
+// send_v1 sent the path in a buffer, followed by a comma and the mode as a string.
+// send_v2 sends just the path in the first request, and then sends another syncmsg (with the
+// same ID!) with details.
+struct __attribute__((packed)) sync_send_v2 {
+    uint32_t id;
+    uint32_t mode;
+    uint32_t flags;
+};
+
+// Likewise, recv_v1 just sent the path without any accompanying data.
+struct __attribute__((packed)) sync_recv_v2 {
+    uint32_t id;
+    uint32_t flags;
+};
+
 struct __attribute__((packed)) sync_data {
     uint32_t id;
     uint32_t size;
@@ -104,6 +126,8 @@
     sync_dent_v2 dent_v2;
     sync_data data;
     sync_status status;
+    sync_send_v2 send_v2_setup;
+    sync_recv_v2 recv_v2_setup;
 };
 
 #define SYNC_DATA_MAX (64 * 1024)
diff --git a/adb/sysdeps.h b/adb/sysdeps.h
index 3e781b8..9a879b5 100644
--- a/adb/sysdeps.h
+++ b/adb/sysdeps.h
@@ -665,6 +665,10 @@
     return fd.get();
 }
 
+static __inline__ int cast_handle_to_int(int fd) {
+    return fd;
+}
+
 // A very simple wrapper over a launched child process
 class Process {
   public:
diff --git a/adb/transport.cpp b/adb/transport.cpp
index be440e2..6dccb7f 100644
--- a/adb/transport.cpp
+++ b/adb/transport.cpp
@@ -82,6 +82,8 @@
 const char* const kFeatureAbbExec = "abb_exec";
 const char* const kFeatureRemountShell = "remount_shell";
 const char* const kFeatureTrackApp = "track_app";
+const char* const kFeatureSendRecv2 = "sendrecv_v2";
+const char* const kFeatureSendRecv2Brotli = "sendrecv_v2_brotli";
 
 namespace {
 
@@ -498,12 +500,8 @@
     auto x509 = GenerateX509Certificate(evp_pkey.get());
     auto x509_str = X509ToPEMString(x509.get());
     auto evp_str = Key::ToPEMString(evp_pkey.get());
-#ifdef _WIN32
-    int osh = cast_handle_to_int(adb_get_os_handle(fd_));
-#else
-    int osh = adb_get_os_handle(fd_);
-#endif
 
+    int osh = cast_handle_to_int(adb_get_os_handle(fd_));
 #if ADB_HOST
     tls_ = TlsConnection::Create(TlsConnection::Role::Client,
 #else
@@ -1183,6 +1181,8 @@
             kFeatureAbbExec,
             kFeatureRemountShell,
             kFeatureTrackApp,
+            kFeatureSendRecv2,
+            kFeatureSendRecv2Brotli,
             // Increment ADB_SERVER_VERSION when adding a feature that adbd needs
             // to know about. Otherwise, the client can be stuck running an old
             // version of the server even after upgrading their copy of adb.
diff --git a/adb/transport.h b/adb/transport.h
index a62349e..3a77cf6 100644
--- a/adb/transport.h
+++ b/adb/transport.h
@@ -81,9 +81,14 @@
 extern const char* const kFeatureAbbExec;
 // adbd properly updates symlink timestamps on push.
 extern const char* const kFeatureFixedPushSymlinkTimestamp;
+// Implement `adb remount` via shelling out to /system/bin/remount.
 extern const char* const kFeatureRemountShell;
 // adbd supports `track-app` service reporting debuggable/profileable apps.
 extern const char* const kFeatureTrackApp;
+// adbd supports version 2 of send/recv.
+extern const char* const kFeatureSendRecv2;
+// adbd supports brotli for send/recv v2.
+extern const char* const kFeatureSendRecv2Brotli;
 
 TransportId NextTransportId();
 
diff --git a/adb/types.h b/adb/types.h
index c619fff..deca7ea 100644
--- a/adb/types.h
+++ b/adb/types.h
@@ -150,6 +150,22 @@
     IOVector& operator=(const IOVector& copy) = delete;
     IOVector& operator=(IOVector&& move) noexcept;
 
+    const value_type* front_data() const {
+        if (chain_.empty()) {
+            return nullptr;
+        }
+
+        return chain_.front().data() + begin_offset_;
+    }
+
+    size_type front_size() const {
+        if (chain_.empty()) {
+            return 0;
+        }
+
+        return chain_.front().size() - begin_offset_;
+    }
+
     size_type size() const { return chain_length_ - begin_offset_; }
     bool empty() const { return size() == 0; }
 
diff --git a/fastboot/device/commands.cpp b/fastboot/device/commands.cpp
index ca120c6..b8eee4a 100644
--- a/fastboot/device/commands.cpp
+++ b/fastboot/device/commands.cpp
@@ -31,6 +31,7 @@
 #include <cutils/android_reboot.h>
 #include <ext4_utils/wipe.h>
 #include <fs_mgr.h>
+#include <fs_mgr/roots.h>
 #include <libgsi/libgsi.h>
 #include <liblp/builder.h>
 #include <liblp/liblp.h>
diff --git a/fastboot/device/fastboot_device.cpp b/fastboot/device/fastboot_device.cpp
index 31fc359..bb085c5 100644
--- a/fastboot/device/fastboot_device.cpp
+++ b/fastboot/device/fastboot_device.cpp
@@ -16,18 +16,22 @@
 
 #include "fastboot_device.h"
 
+#include <algorithm>
+
 #include <android-base/logging.h>
 #include <android-base/strings.h>
 #include <android/hardware/boot/1.0/IBootControl.h>
 #include <android/hardware/fastboot/1.0/IFastboot.h>
+#include <fs_mgr.h>
+#include <fs_mgr/roots.h>
 #include <healthhalutils/HealthHalUtils.h>
 
-#include <algorithm>
-
 #include "constants.h"
 #include "flashing.h"
 #include "usb_client.h"
 
+using android::fs_mgr::EnsurePathUnmounted;
+using android::fs_mgr::Fstab;
 using ::android::hardware::hidl_string;
 using ::android::hardware::boot::V1_0::IBootControl;
 using ::android::hardware::boot::V1_0::Slot;
@@ -64,6 +68,13 @@
     if (boot_control_hal_) {
         boot1_1_ = android::hardware::boot::V1_1::IBootControl::castFrom(boot_control_hal_);
     }
+
+    // Make sure cache is unmounted, since recovery will have mounted it for
+    // logging.
+    Fstab fstab;
+    if (ReadDefaultFstab(&fstab)) {
+        EnsurePathUnmounted(&fstab, "/cache");
+    }
 }
 
 FastbootDevice::~FastbootDevice() {
diff --git a/fs_mgr/libdm/Android.bp b/fs_mgr/libdm/Android.bp
index c2c2cc4..0499e8d 100644
--- a/fs_mgr/libdm/Android.bp
+++ b/fs_mgr/libdm/Android.bp
@@ -58,11 +58,11 @@
     defaults: ["fs_mgr_defaults"],
     static_libs: [
         "libdm",
+        "libext2_uuid",
+        "libfs_mgr",
     ],
     shared_libs: [
         "libbase",
-        "libext2_uuid",
-        "libfs_mgr",
         "liblog",
     ],
     srcs: [":libdm_test_srcs"],
diff --git a/fs_mgr/libsnapshot/include/libsnapshot/snapshot.h b/fs_mgr/libsnapshot/include/libsnapshot/snapshot.h
index 5c276b4..1daa83b 100644
--- a/fs_mgr/libsnapshot/include/libsnapshot/snapshot.h
+++ b/fs_mgr/libsnapshot/include/libsnapshot/snapshot.h
@@ -142,7 +142,9 @@
     // be created, and the device must either cancel the OTA (either before
     // rebooting or after rolling back), or merge the OTA.
     // Before calling this function, all snapshots must be mapped.
-    bool FinishedSnapshotWrites();
+    // If |wipe| is set to true, wipe is scheduled after reboot, and snapshots
+    // may need to be merged before wiping.
+    bool FinishedSnapshotWrites(bool wipe);
 
     // Initiate a merge on all snapshot devices. This should only be used after an
     // update has been marked successful after booting.
@@ -452,6 +454,7 @@
 
     std::string GetSnapshotBootIndicatorPath();
     std::string GetRollbackIndicatorPath();
+    std::string GetForwardMergeIndicatorPath();
 
     // Return the name of the device holding the "snapshot" or "snapshot-merge"
     // target. This may not be the final device presented via MapSnapshot(), if
@@ -522,6 +525,17 @@
     bool ShouldDeleteSnapshot(LockedFile* lock, const std::map<std::string, bool>& flashing_status,
                               Slot current_slot, const std::string& name);
 
+    // Create or delete forward merge indicator given |wipe|. Iff wipe is scheduled,
+    // allow forward merge on FDR.
+    bool UpdateForwardMergeIndicator(bool wipe);
+
+    // Helper for HandleImminentDataWipe.
+    // Call ProcessUpdateState and handle states with special rules before data wipe. Specifically,
+    // if |allow_forward_merge| and allow-forward-merge indicator exists, initiate merge if
+    // necessary.
+    bool ProcessUpdateStateOnDataWipe(bool allow_forward_merge,
+                                      const std::function<bool()>& callback);
+
     std::string gsid_dir_;
     std::string metadata_dir_;
     std::unique_ptr<IDeviceInfo> device_;
diff --git a/fs_mgr/libsnapshot/snapshot.cpp b/fs_mgr/libsnapshot/snapshot.cpp
index 1eec6a4..c9fa28e 100644
--- a/fs_mgr/libsnapshot/snapshot.cpp
+++ b/fs_mgr/libsnapshot/snapshot.cpp
@@ -228,10 +228,17 @@
         return false;
     }
 
-    // It's okay if these fail - first-stage init performs a deeper check after
+    // It's okay if these fail:
+    // - For SnapshotBoot and Rollback, first-stage init performs a deeper check after
     // reading the indicator file, so it's not a problem if it still exists
     // after the update completes.
-    std::vector<std::string> files = {GetSnapshotBootIndicatorPath(), GetRollbackIndicatorPath()};
+    // - For ForwardMerge, FinishedSnapshotWrites asserts that the existence of the indicator
+    // matches the incoming update.
+    std::vector<std::string> files = {
+            GetSnapshotBootIndicatorPath(),
+            GetRollbackIndicatorPath(),
+            GetForwardMergeIndicatorPath(),
+    };
     for (const auto& file : files) {
         RemoveFileIfExists(file);
     }
@@ -241,7 +248,7 @@
     return WriteUpdateState(lock, UpdateState::None);
 }
 
-bool SnapshotManager::FinishedSnapshotWrites() {
+bool SnapshotManager::FinishedSnapshotWrites(bool wipe) {
     auto lock = LockExclusive();
     if (!lock) return false;
 
@@ -261,6 +268,10 @@
         return false;
     }
 
+    if (!UpdateForwardMergeIndicator(wipe)) {
+        return false;
+    }
+
     // This file is written on boot to detect whether a rollback occurred. It
     // MUST NOT exist before rebooting, otherwise, we're at risk of deleting
     // snapshots too early.
@@ -992,6 +1003,10 @@
     return metadata_dir_ + "/" + android::base::Basename(kRollbackIndicatorPath);
 }
 
+std::string SnapshotManager::GetForwardMergeIndicatorPath() {
+    return metadata_dir_ + "/allow-forward-merge";
+}
+
 void SnapshotManager::AcknowledgeMergeSuccess(LockedFile* lock) {
     RemoveAllUpdateState(lock);
 }
@@ -2438,6 +2453,9 @@
     ss << "Rollback indicator: "
        << (access(GetRollbackIndicatorPath().c_str(), F_OK) == 0 ? "exists" : strerror(errno))
        << std::endl;
+    ss << "Forward merge indicator: "
+       << (access(GetForwardMergeIndicatorPath().c_str(), F_OK) == 0 ? "exists" : strerror(errno))
+       << std::endl;
 
     bool ok = true;
     std::vector<std::string> snapshots;
@@ -2504,17 +2522,39 @@
         return false;
     }
 
-    UpdateState state = ProcessUpdateState([&]() -> bool {
-        callback();
+    auto process_callback = [&]() -> bool {
+        if (callback) {
+            callback();
+        }
         return true;
-    });
+    };
+    if (!ProcessUpdateStateOnDataWipe(true /* allow_forward_merge */, process_callback)) {
+        return false;
+    }
+
+    // Nothing should be depending on partitions now, so unmap them all.
+    if (!UnmapAllPartitions()) {
+        LOG(ERROR) << "Unable to unmap all partitions; fastboot may fail to flash.";
+    }
+    return true;
+}
+
+bool SnapshotManager::ProcessUpdateStateOnDataWipe(bool allow_forward_merge,
+                                                   const std::function<bool()>& callback) {
+    auto slot_number = SlotNumberForSlotSuffix(device_->GetSlotSuffix());
+    UpdateState state = ProcessUpdateState(callback);
     LOG(INFO) << "Update state in recovery: " << state;
     switch (state) {
         case UpdateState::MergeFailed:
             LOG(ERROR) << "Unrecoverable merge failure detected.";
             return false;
         case UpdateState::Unverified: {
-            // If an OTA was just applied but has not yet started merging, we
+            // If an OTA was just applied but has not yet started merging:
+            //
+            // - if forward merge is allowed, initiate merge and call
+            // ProcessUpdateState again.
+            //
+            // - if forward merge is not allowed, we
             // have no choice but to revert slots, because the current slot will
             // immediately become unbootable. Rather than wait for the device
             // to reboot N times until a rollback, we proactively disable the
@@ -2524,8 +2564,17 @@
             // as an error here.
             auto slot = GetCurrentSlot();
             if (slot == Slot::Target) {
+                if (allow_forward_merge &&
+                    access(GetForwardMergeIndicatorPath().c_str(), F_OK) == 0) {
+                    LOG(INFO) << "Forward merge allowed, initiating merge now.";
+                    return InitiateMerge() &&
+                           ProcessUpdateStateOnDataWipe(false /* allow_forward_merge */, callback);
+                }
+
                 LOG(ERROR) << "Reverting to old slot since update will be deleted.";
                 device_->SetSlotAsUnbootable(slot_number);
+            } else {
+                LOG(INFO) << "Booting from " << slot << " slot, no action is taken.";
             }
             break;
         }
@@ -2537,11 +2586,6 @@
         default:
             break;
     }
-
-    // Nothing should be depending on partitions now, so unmap them all.
-    if (!UnmapAllPartitions()) {
-        LOG(ERROR) << "Unable to unmap all partitions; fastboot may fail to flash.";
-    }
     return true;
 }
 
@@ -2622,5 +2666,24 @@
     return CreateResult::CREATED;
 }
 
+bool SnapshotManager::UpdateForwardMergeIndicator(bool wipe) {
+    auto path = GetForwardMergeIndicatorPath();
+
+    if (!wipe) {
+        LOG(INFO) << "Wipe is not scheduled. Deleting forward merge indicator.";
+        return RemoveFileIfExists(path);
+    }
+
+    // TODO(b/152094219): Don't forward merge if no CoW file is allocated.
+
+    LOG(INFO) << "Wipe will be scheduled. Allowing forward merge of snapshots.";
+    if (!android::base::WriteStringToFile("1", path)) {
+        PLOG(ERROR) << "Unable to write forward merge indicator: " << path;
+        return false;
+    }
+
+    return true;
+}
+
 }  // namespace snapshot
 }  // namespace android
diff --git a/fs_mgr/libsnapshot/snapshot_test.cpp b/fs_mgr/libsnapshot/snapshot_test.cpp
index 855451d..f82c082 100644
--- a/fs_mgr/libsnapshot/snapshot_test.cpp
+++ b/fs_mgr/libsnapshot/snapshot_test.cpp
@@ -320,7 +320,7 @@
     // Simulate a reboot into the new slot.
     AssertionResult SimulateReboot() {
         lock_ = nullptr;
-        if (!sm->FinishedSnapshotWrites()) {
+        if (!sm->FinishedSnapshotWrites(false)) {
             return AssertionFailure();
         }
         if (!dm_.DeleteDevice("test_partition_b")) {
@@ -424,7 +424,7 @@
 }
 
 TEST_F(SnapshotTest, NoMergeBeforeReboot) {
-    ASSERT_TRUE(sm->FinishedSnapshotWrites());
+    ASSERT_TRUE(sm->FinishedSnapshotWrites(false));
 
     // Merge should fail, since the slot hasn't changed.
     ASSERT_FALSE(sm->InitiateMerge());
@@ -440,7 +440,7 @@
 }
 
 TEST_F(SnapshotTest, FirstStageMountAfterRollback) {
-    ASSERT_TRUE(sm->FinishedSnapshotWrites());
+    ASSERT_TRUE(sm->FinishedSnapshotWrites(false));
 
     // We didn't change the slot, so we shouldn't need snapshots.
     TestDeviceInfo* info = new TestDeviceInfo(fake_super);
@@ -476,7 +476,7 @@
     lock_ = nullptr;
 
     // Done updating.
-    ASSERT_TRUE(sm->FinishedSnapshotWrites());
+    ASSERT_TRUE(sm->FinishedSnapshotWrites(false));
 
     test_device->set_slot_suffix("_b");
     ASSERT_TRUE(sm->InitiateMerge());
@@ -1007,7 +1007,7 @@
         ASSERT_TRUE(IsPartitionUnchanged(name));
     }
 
-    ASSERT_TRUE(sm->FinishedSnapshotWrites());
+    ASSERT_TRUE(sm->FinishedSnapshotWrites(false));
 
     // Simulate shutting down the device.
     ASSERT_TRUE(UnmapAll());
@@ -1139,7 +1139,7 @@
         ASSERT_TRUE(IsPartitionUnchanged(name));
     }
 
-    ASSERT_TRUE(sm->FinishedSnapshotWrites());
+    ASSERT_TRUE(sm->FinishedSnapshotWrites(false));
 
     // Simulate shutting down the device.
     ASSERT_TRUE(UnmapAll());
@@ -1171,7 +1171,7 @@
 // Test that if an update is applied but not booted into, it can be canceled.
 TEST_F(SnapshotUpdateTest, CancelAfterApply) {
     ASSERT_TRUE(sm->BeginUpdate());
-    ASSERT_TRUE(sm->FinishedSnapshotWrites());
+    ASSERT_TRUE(sm->FinishedSnapshotWrites(false));
     ASSERT_TRUE(sm->CancelUpdate());
 }
 
@@ -1188,7 +1188,7 @@
     ASSERT_TRUE(sm->BeginUpdate());
     ASSERT_TRUE(sm->CreateUpdateSnapshots(manifest_));
     ASSERT_TRUE(MapUpdateSnapshots());
-    ASSERT_TRUE(sm->FinishedSnapshotWrites());
+    ASSERT_TRUE(sm->FinishedSnapshotWrites(false));
 
     // Simulate shutting down the device.
     ASSERT_TRUE(UnmapAll());
@@ -1295,7 +1295,7 @@
         ASSERT_TRUE(IsPartitionUnchanged(name));
     }
 
-    ASSERT_TRUE(sm->FinishedSnapshotWrites());
+    ASSERT_TRUE(sm->FinishedSnapshotWrites(false));
 }
 
 TEST_F(SnapshotUpdateTest, MergeCannotRemoveCow) {
@@ -1324,7 +1324,7 @@
     ASSERT_TRUE(sm->BeginUpdate());
     ASSERT_TRUE(sm->CreateUpdateSnapshots(manifest_));
     ASSERT_TRUE(MapUpdateSnapshots());
-    ASSERT_TRUE(sm->FinishedSnapshotWrites());
+    ASSERT_TRUE(sm->FinishedSnapshotWrites(false));
 
     // Simulate shutting down the device.
     ASSERT_TRUE(UnmapAll());
@@ -1428,7 +1428,7 @@
     ASSERT_TRUE(sm->BeginUpdate());
     ASSERT_TRUE(sm->CreateUpdateSnapshots(manifest_));
     ASSERT_TRUE(MapUpdateSnapshots());
-    ASSERT_TRUE(sm->FinishedSnapshotWrites());
+    ASSERT_TRUE(sm->FinishedSnapshotWrites(false));
 
     // Simulate shutting down the device.
     ASSERT_TRUE(UnmapAll());
@@ -1460,7 +1460,7 @@
     ASSERT_TRUE(sm->BeginUpdate());
     ASSERT_TRUE(sm->CreateUpdateSnapshots(manifest_));
     ASSERT_TRUE(MapUpdateSnapshots());
-    ASSERT_TRUE(sm->FinishedSnapshotWrites());
+    ASSERT_TRUE(sm->FinishedSnapshotWrites(false));
 
     // Simulate shutting down the device.
     ASSERT_TRUE(UnmapAll());
@@ -1485,7 +1485,7 @@
     ASSERT_TRUE(sm->BeginUpdate());
     ASSERT_TRUE(sm->CreateUpdateSnapshots(manifest_));
     ASSERT_TRUE(MapUpdateSnapshots());
-    ASSERT_TRUE(sm->FinishedSnapshotWrites());
+    ASSERT_TRUE(sm->FinishedSnapshotWrites(false));
 
     // Simulate shutting down the device.
     ASSERT_TRUE(UnmapAll());
@@ -1498,7 +1498,46 @@
     ASSERT_TRUE(new_sm->HandleImminentDataWipe());
     EXPECT_EQ(new_sm->GetUpdateState(), UpdateState::None);
     EXPECT_FALSE(test_device->IsSlotUnbootable(0));
-    EXPECT_FALSE(test_device->IsSlotUnbootable(0));
+    EXPECT_FALSE(test_device->IsSlotUnbootable(1));
+}
+
+// Test update package that requests data wipe.
+TEST_F(SnapshotUpdateTest, DataWipeRequiredInPackage) {
+    AddOperationForPartitions();
+    // Execute the update.
+    ASSERT_TRUE(sm->BeginUpdate());
+    ASSERT_TRUE(sm->CreateUpdateSnapshots(manifest_));
+
+    // Write some data to target partitions.
+    for (const auto& name : {"sys_b", "vnd_b", "prd_b"}) {
+        ASSERT_TRUE(WriteSnapshotAndHash(name)) << name;
+    }
+
+    ASSERT_TRUE(sm->FinishedSnapshotWrites(true /* wipe */));
+
+    // Simulate shutting down the device.
+    ASSERT_TRUE(UnmapAll());
+
+    // Simulate a reboot into recovery.
+    auto test_device = new TestDeviceInfo(fake_super, "_b");
+    test_device->set_recovery(true);
+    auto new_sm = SnapshotManager::NewForFirstStageMount(test_device);
+
+    ASSERT_TRUE(new_sm->HandleImminentDataWipe());
+    // Manually mount metadata so that we can call GetUpdateState() below.
+    MountMetadata();
+    EXPECT_EQ(new_sm->GetUpdateState(), UpdateState::None);
+    ASSERT_FALSE(test_device->IsSlotUnbootable(1));
+    ASSERT_FALSE(test_device->IsSlotUnbootable(0));
+
+    // Now reboot into new slot.
+    test_device = new TestDeviceInfo(fake_super, "_b");
+    auto init = SnapshotManager::NewForFirstStageMount(test_device);
+    ASSERT_TRUE(init->CreateLogicalAndSnapshotPartitions("super", snapshot_timeout_));
+    // Verify that we are on the downgraded build.
+    for (const auto& name : {"sys_b", "vnd_b", "prd_b"}) {
+        ASSERT_TRUE(IsPartitionUnchanged(name)) << name;
+    }
 }
 
 TEST_F(SnapshotUpdateTest, Hashtree) {
@@ -1533,7 +1572,7 @@
     ASSERT_TRUE(WriteSnapshotAndHash("sys_b", partition_size));
 
     // Finish update.
-    ASSERT_TRUE(sm->FinishedSnapshotWrites());
+    ASSERT_TRUE(sm->FinishedSnapshotWrites(false));
 
     // Simulate shutting down the device.
     ASSERT_TRUE(UnmapAll());
@@ -1569,7 +1608,7 @@
     ASSERT_EQ(1u, table.size());
     EXPECT_TRUE(table[0].IsOverflowSnapshot());
 
-    ASSERT_FALSE(sm->FinishedSnapshotWrites())
+    ASSERT_FALSE(sm->FinishedSnapshotWrites(false))
             << "FinishedSnapshotWrites should detect overflow of CoW device.";
 }
 
@@ -1623,7 +1662,7 @@
     ASSERT_TRUE(sm->BeginUpdate());
     ASSERT_TRUE(sm->CreateUpdateSnapshots(manifest_));
     ASSERT_TRUE(MapUpdateSnapshots());
-    ASSERT_TRUE(sm->FinishedSnapshotWrites());
+    ASSERT_TRUE(sm->FinishedSnapshotWrites(false));
 
     // Simulate shutting down the device.
     ASSERT_TRUE(UnmapAll());
diff --git a/init/init.cpp b/init/init.cpp
index 81a097e..a9d6301 100644
--- a/init/init.cpp
+++ b/init/init.cpp
@@ -22,6 +22,7 @@
 #include <signal.h>
 #include <stdlib.h>
 #include <string.h>
+#include <sys/eventfd.h>
 #include <sys/mount.h>
 #include <sys/signalfd.h>
 #include <sys/types.h>
@@ -115,30 +116,26 @@
 // to fill that socket and deadlock the system.  Now we use locks to handle the property changes
 // directly in the property thread, however we still must wake the epoll to inform init that there
 // is a change to process, so we use this FD.  It is non-blocking, since we do not care how many
-// times WakeEpoll() is called, only that the epoll will wake.
-static int wake_epoll_fd = -1;
+// times WakeMainInitThread() is called, only that the epoll will wake.
+static int wake_main_thread_fd = -1;
 static void InstallInitNotifier(Epoll* epoll) {
-    int sockets[2];
-    if (socketpair(AF_UNIX, SOCK_STREAM | SOCK_CLOEXEC | SOCK_NONBLOCK, 0, sockets) != 0) {
-        PLOG(FATAL) << "Failed to socketpair() between property_service and init";
+    wake_main_thread_fd = eventfd(0, EFD_CLOEXEC);
+    if (wake_main_thread_fd == -1) {
+        PLOG(FATAL) << "Failed to create eventfd for waking init";
     }
-    int epoll_fd = sockets[0];
-    wake_epoll_fd = sockets[1];
-
-    auto drain_socket = [epoll_fd] {
-        char buf[512];
-        while (read(epoll_fd, buf, sizeof(buf)) > 0) {
-        }
+    auto clear_eventfd = [] {
+        uint64_t counter;
+        TEMP_FAILURE_RETRY(read(wake_main_thread_fd, &counter, sizeof(counter)));
     };
 
-    if (auto result = epoll->RegisterHandler(epoll_fd, drain_socket); !result.ok()) {
+    if (auto result = epoll->RegisterHandler(wake_main_thread_fd, clear_eventfd); !result.ok()) {
         LOG(FATAL) << result.error();
     }
 }
 
-static void WakeEpoll() {
-    constexpr char value[] = "1";
-    write(wake_epoll_fd, value, sizeof(value));
+static void WakeMainInitThread() {
+    uint64_t counter = 1;
+    TEMP_FAILURE_RETRY(write(wake_main_thread_fd, &counter, sizeof(counter)));
 }
 
 static class PropWaiterState {
@@ -182,7 +179,7 @@
                 LOG(INFO) << "Wait for property '" << wait_prop_name_ << "=" << wait_prop_value_
                           << "' took " << *waiting_for_prop_;
                 ResetWaitForPropLocked();
-                WakeEpoll();
+                WakeMainInitThread();
             }
         }
     }
@@ -238,18 +235,9 @@
         // action queue.  Instead we set this flag and ensure that shutdown happens before the next
         // command is run in the main init loop.
         auto lock = std::lock_guard{shutdown_command_lock_};
-        if (do_shutdown_) {
-            LOG(ERROR) << "TriggerShutdown called while a previous shutdown command '"
-                       << shutdown_command_ << "' has not been handled";
-            UnwindMainThreadStack();
-        }
-        if (IsShuttingDown()) {
-            LOG(ERROR) << "TriggerShutdown called while init is already shutting down";
-            UnwindMainThreadStack();
-        }
         shutdown_command_ = command;
         do_shutdown_ = true;
-        WakeEpoll();
+        WakeMainInitThread();
     }
 
     std::optional<std::string> CheckShutdown() {
@@ -261,12 +249,27 @@
         return {};
     }
 
+    bool do_shutdown() const { return do_shutdown_; }
+
   private:
     std::mutex shutdown_command_lock_;
     std::string shutdown_command_;
     bool do_shutdown_ = false;
 } shutdown_state;
 
+void DebugRebootLogging() {
+    LOG(INFO) << "do_shutdown: " << shutdown_state.do_shutdown()
+              << " IsShuttingDown: " << IsShuttingDown();
+    if (shutdown_state.do_shutdown()) {
+        LOG(ERROR) << "sys.powerctl set while a previous shutdown command has not been handled";
+        UnwindMainThreadStack();
+    }
+    if (IsShuttingDown()) {
+        LOG(ERROR) << "sys.powerctl set while init is already shutting down";
+        UnwindMainThreadStack();
+    }
+}
+
 void DumpState() {
     ServiceList::GetInstance().DumpState();
     ActionManager::GetInstance().DumpState();
@@ -333,7 +336,7 @@
 
     if (property_triggers_enabled) {
         ActionManager::GetInstance().QueuePropertyChange(name, value);
-        WakeEpoll();
+        WakeMainInitThread();
     }
 
     prop_waiter_state.CheckAndResetWait(name, value);
@@ -459,7 +462,7 @@
         return false;
     }
     pending_control_messages.push({message, name, pid, fd});
-    WakeEpoll();
+    WakeMainInitThread();
     return true;
 }
 
@@ -485,7 +488,7 @@
     }
     // If we still have items to process, make sure we wake back up to do so.
     if (!pending_control_messages.empty()) {
-        WakeEpoll();
+        WakeMainInitThread();
     }
 }
 
@@ -904,7 +907,9 @@
                 (*function)();
             }
         }
-        HandleControlMessages();
+        if (!IsShuttingDown()) {
+            HandleControlMessages();
+        }
     }
 
     return 0;
diff --git a/init/init.h b/init/init.h
index 27f64e2..4f686cb 100644
--- a/init/init.h
+++ b/init/init.h
@@ -42,6 +42,8 @@
 void PropertyChanged(const std::string& name, const std::string& value);
 bool QueueControlMessage(const std::string& message, const std::string& name, pid_t pid, int fd);
 
+void DebugRebootLogging();
+
 int SecondStageMain(int argc, char** argv);
 
 }  // namespace init
diff --git a/init/oneshot_on_test.cpp b/init/oneshot_on_test.cpp
index 7e7cc36..487e0ea 100644
--- a/init/oneshot_on_test.cpp
+++ b/init/oneshot_on_test.cpp
@@ -25,6 +25,7 @@
 using android::base::WaitForProperty;
 using namespace std::literals;
 
+#if 0  // TODO(b/152637928): this test is flaky
 TEST(init, oneshot_on) {
     // Bootanim shouldn't be running once the device has booted.
     ASSERT_EQ("stopped", GetProperty("init.svc.bootanim", ""));
@@ -42,3 +43,4 @@
     // Now that oneshot is enabled again, bootanim should transition into the 'stopped' state.
     EXPECT_TRUE(WaitForProperty("init.svc.bootanim", "stopped", 10s));
 }
+#endif
diff --git a/init/property_service.cpp b/init/property_service.cpp
index 8206522..1e4e127 100644
--- a/init/property_service.cpp
+++ b/init/property_service.cpp
@@ -125,7 +125,7 @@
 
 void StopSendingMessages() {
     auto lock = std::lock_guard{accept_messages_lock};
-    accept_messages = true;
+    accept_messages = false;
 }
 
 bool CanReadProperty(const std::string& source_context, const std::string& name) {
@@ -489,6 +489,7 @@
         }
         LOG(INFO) << "Received sys.powerctl='" << value << "' from pid: " << cr.pid
                   << process_log_string;
+        DebugRebootLogging();
     }
 
     // If a process other than init is writing a non-empty value, it means that process is