diff --git a/libs/binder/IPCThreadState.cpp b/libs/binder/IPCThreadState.cpp
index d2919e7..e933f7e 100644
--- a/libs/binder/IPCThreadState.cpp
+++ b/libs/binder/IPCThreadState.cpp
@@ -516,14 +516,16 @@
 
 bool IPCThreadState::flushIfNeeded()
 {
-    if (mIsLooper || mServingStackPointer != nullptr) {
+    if (mIsLooper || mServingStackPointer != nullptr || mIsFlushing) {
         return false;
     }
+    mIsFlushing = true;
     // In case this thread is not a looper and is not currently serving a binder transaction,
     // there's no guarantee that this thread will call back into the kernel driver any time
     // soon. Therefore, flush pending commands such as BC_FREE_BUFFER, to prevent them from getting
     // stuck in this thread's out buffer.
     flushCommands();
+    mIsFlushing = false;
     return true;
 }
 
@@ -880,6 +882,7 @@
         mWorkSource(kUnsetWorkSource),
         mPropagateWorkSource(false),
         mIsLooper(false),
+        mIsFlushing(false),
         mStrictModePolicy(0),
         mLastTransactionBinderFlags(0),
         mCallRestriction(mProcess->mCallRestriction) {
diff --git a/libs/binder/RpcServer.cpp b/libs/binder/RpcServer.cpp
index 73facc1..540c346 100644
--- a/libs/binder/RpcServer.cpp
+++ b/libs/binder/RpcServer.cpp
@@ -128,6 +128,17 @@
     return ret;
 }
 
+static void joinRpcServer(sp<RpcServer>&& thiz) {
+    thiz->join();
+}
+
+void RpcServer::start() {
+    LOG_ALWAYS_FATAL_IF(!mAgreedExperimental, "no!");
+    std::lock_guard<std::mutex> _l(mLock);
+    LOG_ALWAYS_FATAL_IF(mJoinThread.get(), "Already started!");
+    mJoinThread = std::make_unique<std::thread>(&joinRpcServer, sp<RpcServer>::fromExisting(this));
+}
+
 void RpcServer::join() {
     LOG_ALWAYS_FATAL_IF(!mAgreedExperimental, "no!");
 
@@ -173,10 +184,29 @@
 
 bool RpcServer::shutdown() {
     std::unique_lock<std::mutex> _l(mLock);
-    if (mShutdownTrigger == nullptr) return false;
+    if (mShutdownTrigger == nullptr) {
+        LOG_RPC_DETAIL("Cannot shutdown. No shutdown trigger installed.");
+        return false;
+    }
 
     mShutdownTrigger->trigger();
-    while (mJoinThreadRunning) mShutdownCv.wait(_l);
+    while (mJoinThreadRunning || !mConnectingThreads.empty()) {
+        ALOGI("Waiting for RpcServer to shut down. Join thread running: %d, Connecting threads: "
+              "%zu",
+              mJoinThreadRunning, mConnectingThreads.size());
+        mShutdownCv.wait(_l);
+    }
+
+    // At this point, we know join() is about to exit, but the thread that calls
+    // join() may not have exited yet.
+    // If RpcServer owns the join thread (aka start() is called), make sure the thread exits;
+    // otherwise ~thread() may call std::terminate(), which may crash the process.
+    // If RpcServer does not own the join thread (aka join() is called directly),
+    // then the owner of RpcServer is responsible for cleaning up that thread.
+    if (mJoinThread.get()) {
+        mJoinThread->join();
+        mJoinThread.reset();
+    }
 
     mShutdownTrigger = nullptr;
     return true;
@@ -200,17 +230,21 @@
 void RpcServer::establishConnection(sp<RpcServer>&& server, base::unique_fd clientFd) {
     // TODO(b/183988761): cannot trust this simple ID
     LOG_ALWAYS_FATAL_IF(!server->mAgreedExperimental, "no!");
-    bool idValid = true;
+
+    // mShutdownTrigger can only be cleared once connection threads have joined.
+    // It must be set before this thread is started
+    LOG_ALWAYS_FATAL_IF(server->mShutdownTrigger == nullptr);
+
     int32_t id;
-    if (sizeof(id) != read(clientFd.get(), &id, sizeof(id))) {
-        ALOGE("Could not read ID from fd %d", clientFd.get());
-        idValid = false;
+    bool idValid = server->mShutdownTrigger->interruptableRecv(clientFd.get(), &id, sizeof(id));
+    if (!idValid) {
+        ALOGE("Failed to read ID for client connecting to RPC server.");
     }
 
     std::thread thisThread;
     sp<RpcSession> session;
     {
-        std::lock_guard<std::mutex> _l(server->mLock);
+        std::unique_lock<std::mutex> _l(server->mLock);
 
         auto threadId = server->mConnectingThreads.find(std::this_thread::get_id());
         LOG_ALWAYS_FATAL_IF(threadId == server->mConnectingThreads.end(),
@@ -219,6 +253,16 @@
         ScopeGuard detachGuard = [&]() { thisThread.detach(); };
         server->mConnectingThreads.erase(threadId);
 
+        // TODO(b/185167543): we currently can't disable this because we don't
+        // shutdown sessions as well, only the server itself. So, we need to
+        // keep this separate from the detachGuard, since we temporarily want to
+        // give a notification even when we pass ownership of the thread to
+        // a session.
+        ScopeGuard threadLifetimeGuard = [&]() {
+            _l.unlock();
+            server->mShutdownCv.notify_all();
+        };
+
         if (!idValid) {
             return;
         }
diff --git a/libs/binder/RpcSession.cpp b/libs/binder/RpcSession.cpp
index ea82f36..771d738 100644
--- a/libs/binder/RpcSession.cpp
+++ b/libs/binder/RpcSession.cpp
@@ -144,6 +144,22 @@
     }
 }
 
+bool RpcSession::FdTrigger::interruptableRecv(base::borrowed_fd fd, void* data, size_t size) {
+    uint8_t* buffer = reinterpret_cast<uint8_t*>(data);
+    uint8_t* end = buffer + size;
+
+    while (triggerablePollRead(fd)) {
+        ssize_t readSize = TEMP_FAILURE_RETRY(recv(fd.get(), buffer, end - buffer, MSG_NOSIGNAL));
+        if (readSize < 0) {
+            ALOGE("Failed to read %s", strerror(errno));
+            return false;
+        }
+        buffer += readSize;
+        if (buffer == end) return true;
+    }
+    return false;
+}
+
 status_t RpcSession::readId() {
     {
         std::lock_guard<std::mutex> _l(mMutex);
diff --git a/libs/binder/include/binder/IPCThreadState.h b/libs/binder/include/binder/IPCThreadState.h
index ee661a5..204926d 100644
--- a/libs/binder/include/binder/IPCThreadState.h
+++ b/libs/binder/include/binder/IPCThreadState.h
@@ -62,7 +62,7 @@
              * call. If not in a binder call, this will return getpid. If the
              * call is oneway, this will return 0.
              */
-            pid_t               getCallingPid() const;
+            [[nodiscard]] pid_t getCallingPid() const;
 
             /**
              * Returns the SELinux security identifier of the process which has
@@ -73,13 +73,13 @@
              * This can't be restored once it's cleared, and it does not return the
              * context of the current process when not in a binder call.
              */
-            const char*         getCallingSid() const;
+            [[nodiscard]] const char* getCallingSid() const;
 
             /**
              * Returns the UID of the process which has made the current binder
              * call. If not in a binder call, this will return 0.
              */
-            uid_t               getCallingUid() const;
+            [[nodiscard]] uid_t getCallingUid() const;
 
             /**
              * Make it an abort to rely on getCalling* for a section of
@@ -243,6 +243,7 @@
             // Whether the work source should be propagated.
             bool                mPropagateWorkSource;
             bool                mIsLooper;
+            bool mIsFlushing;
             int32_t             mStrictModePolicy;
             int32_t             mLastTransactionBinderFlags;
             CallRestriction     mCallRestriction;
diff --git a/libs/binder/include/binder/RpcServer.h b/libs/binder/include/binder/RpcServer.h
index f76ecc2..50770f1 100644
--- a/libs/binder/include/binder/RpcServer.h
+++ b/libs/binder/include/binder/RpcServer.h
@@ -117,6 +117,11 @@
     sp<IBinder> getRootObject();
 
     /**
+     * Runs join() in a background thread. Immediately returns.
+     */
+    void start();
+
+    /**
      * You must have at least one client session before calling this.
      *
      * If a client needs to actively terminate join, call shutdown() in a separate thread.
@@ -159,12 +164,13 @@
     base::unique_fd mServer; // socket we are accepting sessions on
 
     std::mutex mLock; // for below
+    std::unique_ptr<std::thread> mJoinThread;
+    bool mJoinThreadRunning = false;
     std::map<std::thread::id, std::thread> mConnectingThreads;
     sp<IBinder> mRootObject;
     wp<IBinder> mRootObjectWeak;
     std::map<int32_t, sp<RpcSession>> mSessions;
     int32_t mSessionIdCounter = 0;
-    bool mJoinThreadRunning = false;
     std::unique_ptr<RpcSession::FdTrigger> mShutdownTrigger;
     std::condition_variable mShutdownCv;
 };
diff --git a/libs/binder/include/binder/RpcSession.h b/libs/binder/include/binder/RpcSession.h
index bcef8dc..3d22002 100644
--- a/libs/binder/include/binder/RpcSession.h
+++ b/libs/binder/include/binder/RpcSession.h
@@ -134,6 +134,15 @@
          */
         bool triggerablePollRead(base::borrowed_fd fd);
 
+        /**
+         * Read, but allow the read to be interrupted by this trigger.
+         *
+         * Return:
+         *   true - read succeeded at 'size'
+         *   false - interrupted (failure or trigger)
+         */
+        bool interruptableRecv(base::borrowed_fd fd, void* data, size_t size);
+
     private:
         base::unique_fd mWrite;
         base::unique_fd mRead;
diff --git a/libs/binder/tests/Android.bp b/libs/binder/tests/Android.bp
index 2809da5..9cf433d 100644
--- a/libs/binder/tests/Android.bp
+++ b/libs/binder/tests/Android.bp
@@ -135,6 +135,9 @@
         darwin: {
             enabled: false,
         },
+        android: {
+            test_suites: ["vts"],
+        },
     },
     defaults: [
         "binder_test_defaults",
diff --git a/libs/binder/tests/parcel_fuzzer/main.cpp b/libs/binder/tests/parcel_fuzzer/main.cpp
index f426fd3..2a79e85 100644
--- a/libs/binder/tests/parcel_fuzzer/main.cpp
+++ b/libs/binder/tests/parcel_fuzzer/main.cpp
@@ -23,7 +23,8 @@
 #include <iostream>
 
 #include <android-base/logging.h>
-#include <binder/RpcSession.h>
+#include <android/binder_auto_utils.h>
+#include <android/binder_libbinder.h>
 #include <fuzzbinder/random_parcel.h>
 #include <fuzzer/FuzzedDataProvider.h>
 
@@ -33,7 +34,6 @@
 #include <sys/time.h>
 
 using android::fillRandomParcel;
-using android::RpcSession;
 using android::sp;
 
 void fillRandomParcel(::android::hardware::Parcel* p, FuzzedDataProvider&& provider) {
@@ -46,9 +46,22 @@
     fillRandomParcel(p->parcel(), std::move(provider));
 }
 
+template <typename P, typename B>
+void doTransactFuzz(const char* backend, const sp<B>& binder, FuzzedDataProvider&& provider) {
+    uint32_t code = provider.ConsumeIntegral<uint32_t>();
+    uint32_t flag = provider.ConsumeIntegral<uint32_t>();
+
+    FUZZ_LOG() << "backend: " << backend;
+
+    P reply;
+    P data;
+    fillRandomParcel(&data, std::move(provider));
+    (void)binder->transact(code, data, &reply, flag);
+}
+
 template <typename P>
-void doFuzz(const char* backend, const std::vector<ParcelRead<P>>& reads,
-            FuzzedDataProvider&& provider) {
+void doReadFuzz(const char* backend, const std::vector<ParcelRead<P>>& reads,
+                FuzzedDataProvider&& provider) {
     // Allow some majority of the bytes to be dedicated to telling us what to
     // do. The fixed value added here represents that we want to test doing a
     // lot of 'instructions' even on really short parcels.
@@ -59,18 +72,7 @@
             provider.ConsumeIntegralInRange<size_t>(0, maxInstructions));
 
     P p;
-    if constexpr (std::is_same_v<P, android::Parcel>) {
-        if (provider.ConsumeBool()) {
-            auto session = sp<RpcSession>::make();
-            CHECK(session->addNullDebuggingClient());
-            p.markForRpc(session);
-            fillRandomParcelData(&p, std::move(provider));
-        } else {
-            fillRandomParcel(&p, std::move(provider));
-        }
-    } else {
-        fillRandomParcel(&p, std::move(provider));
-    }
+    fillRandomParcel(&p, std::move(provider));
 
     // since we are only using a byte to index
     CHECK(reads.size() <= 255) << reads.size();
@@ -95,6 +97,17 @@
     }
 }
 
+void* NothingClass_onCreate(void* args) {
+    return args;
+}
+void NothingClass_onDestroy(void* /*userData*/) {}
+binder_status_t NothingClass_onTransact(AIBinder*, transaction_code_t, const AParcel*, AParcel*) {
+    return STATUS_UNKNOWN_ERROR;
+}
+static AIBinder_Class* kNothingClass =
+        AIBinder_Class_define("nothing", NothingClass_onCreate, NothingClass_onDestroy,
+                              NothingClass_onTransact);
+
 extern "C" int LLVMFuzzerTestOneInput(const uint8_t* data, size_t size) {
     if (size <= 1) return 0;  // no use
 
@@ -103,18 +116,35 @@
 
     FuzzedDataProvider provider = FuzzedDataProvider(data, size);
 
-    const std::function<void(FuzzedDataProvider &&)> fuzzBackend[3] = {
+    const std::function<void(FuzzedDataProvider &&)> fuzzBackend[] = {
             [](FuzzedDataProvider&& provider) {
-                doFuzz<::android::hardware::Parcel>("hwbinder", HWBINDER_PARCEL_READ_FUNCTIONS,
-                                                    std::move(provider));
+                doTransactFuzz<
+                        ::android::hardware::Parcel>("hwbinder",
+                                                     sp<::android::hardware::BHwBinder>::make(),
+                                                     std::move(provider));
             },
             [](FuzzedDataProvider&& provider) {
-                doFuzz<::android::Parcel>("binder", BINDER_PARCEL_READ_FUNCTIONS,
-                                          std::move(provider));
+                doTransactFuzz<::android::Parcel>("binder", sp<::android::BBinder>::make(),
+                                                  std::move(provider));
             },
             [](FuzzedDataProvider&& provider) {
-                doFuzz<NdkParcelAdapter>("binder_ndk", BINDER_NDK_PARCEL_READ_FUNCTIONS,
-                                         std::move(provider));
+                // fuzz from the libbinder layer since it's a superset of the
+                // interface you get at the libbinder_ndk layer
+                auto ndkBinder = ndk::SpAIBinder(AIBinder_new(kNothingClass, nullptr));
+                auto binder = AIBinder_toPlatformBinder(ndkBinder.get());
+                doTransactFuzz<::android::Parcel>("binder_ndk", binder, std::move(provider));
+            },
+            [](FuzzedDataProvider&& provider) {
+                doReadFuzz<::android::hardware::Parcel>("hwbinder", HWBINDER_PARCEL_READ_FUNCTIONS,
+                                                        std::move(provider));
+            },
+            [](FuzzedDataProvider&& provider) {
+                doReadFuzz<::android::Parcel>("binder", BINDER_PARCEL_READ_FUNCTIONS,
+                                              std::move(provider));
+            },
+            [](FuzzedDataProvider&& provider) {
+                doReadFuzz<NdkParcelAdapter>("binder_ndk", BINDER_NDK_PARCEL_READ_FUNCTIONS,
+                                             std::move(provider));
             },
     };
 
diff --git a/libs/binder/tests/parcel_fuzzer/random_parcel.cpp b/libs/binder/tests/parcel_fuzzer/random_parcel.cpp
index b045a22..92fdc72 100644
--- a/libs/binder/tests/parcel_fuzzer/random_parcel.cpp
+++ b/libs/binder/tests/parcel_fuzzer/random_parcel.cpp
@@ -18,6 +18,7 @@
 
 #include <android-base/logging.h>
 #include <binder/IServiceManager.h>
+#include <binder/RpcSession.h>
 #include <fuzzbinder/random_fd.h>
 #include <utils/String16.h>
 
@@ -33,6 +34,14 @@
 };
 
 void fillRandomParcel(Parcel* p, FuzzedDataProvider&& provider) {
+    if (provider.ConsumeBool()) {
+        auto session = sp<RpcSession>::make();
+        CHECK(session->addNullDebuggingClient());
+        p->markForRpc(session);
+        fillRandomParcelData(p, std::move(provider));
+        return;
+    }
+
     while (provider.remaining_bytes() > 0) {
         auto fillFunc = provider.PickValueInArray<const std::function<void()>>({
                 // write data
diff --git a/libs/binder/tests/rpc_fuzzer/main.cpp b/libs/binder/tests/rpc_fuzzer/main.cpp
index 84f5974..e6fd392 100644
--- a/libs/binder/tests/rpc_fuzzer/main.cpp
+++ b/libs/binder/tests/rpc_fuzzer/main.cpp
@@ -20,6 +20,7 @@
 #include <binder/Parcel.h>
 #include <binder/RpcServer.h>
 #include <binder/RpcSession.h>
+#include <fuzzer/FuzzedDataProvider.h>
 
 #include <sys/resource.h>
 #include <sys/un.h>
@@ -53,6 +54,7 @@
 
 extern "C" int LLVMFuzzerTestOneInput(const uint8_t* data, size_t size) {
     if (size > 50000) return 0;
+    FuzzedDataProvider provider(data, size);
 
     unlink(kSock.c_str());
 
@@ -84,14 +86,21 @@
     CHECK(base::WriteFully(clientFd, &id, sizeof(id)));
 #endif
 
-    CHECK(base::WriteFully(clientFd, data, size));
+    bool hangupBeforeShutdown = provider.ConsumeBool();
 
-    clientFd.reset();
+    std::vector<uint8_t> writeData = provider.ConsumeRemainingBytes<uint8_t>();
+    CHECK(base::WriteFully(clientFd, writeData.data(), writeData.size()));
+
+    if (hangupBeforeShutdown) {
+        clientFd.reset();
+    }
 
     // TODO(185167543): currently this is okay because we only shutdown the one
     // thread, but once we can shutdown other sessions, we'll need to change
     // this behavior in order to make sure all of the input is actually read.
     while (!server->shutdown()) usleep(100);
+
+    clientFd.reset();
     serverThread.join();
 
     // TODO(b/185167543): better way to force a server to shutdown