diff --git a/connections/implementation/endpoint_manager.cc b/connections/implementation/endpoint_manager.cc
index 6eeadb47..2038f1d4 100644
--- a/connections/implementation/endpoint_manager.cc
+++ b/connections/implementation/endpoint_manager.cc
@@ -15,6 +15,7 @@
 #include "connections/implementation/endpoint_manager.h"
 
 #include <algorithm>
+#include <cstdint>
 #include <functional>
 #include <memory>
 #include <string>
@@ -30,6 +31,7 @@
 #include "connections/implementation/payload_manager.h"
 #include "connections/implementation/proto/offline_wire_formats.pb.h"
 #include "connections/implementation/service_id_constants.h"
+#include "internal/platform/byte_array.h"
 #include "internal/platform/count_down_latch.h"
 #include "internal/platform/exception.h"
 #include "internal/platform/feature_flags.h"
@@ -836,7 +838,9 @@ bool EndpointManager::ApplySafeToDisconnect(const std::string& endpoint_id,
 
   NEARBY_LOGS(WARNING) << "[safe-to-disconnect] Wait for "
                        << (is_wait_for_ack ? "ack" : "disconnection")
-                       << ", timeout in " << timeout_millis;
+                       << " from endpoint: " << endpoint_id
+                       << " for reason: " << reason << ", timeout in "
+                       << timeout_millis;
   bool state = channel_manager_->CreateNewTimeoutDisconnectedState(
       endpoint_id, timeout_millis);
   if (!state) return is_safe_disconnection;
@@ -901,6 +905,21 @@ CountDownLatch EndpointManager::NotifyFrameProcessorsOnEndpointDisconnect(
   return barrier;
 }
 
+std::vector<std::string> EndpointManager::SendPayloadAck(
+std::int64_t payload_id,
+      const std::vector<std::string>& endpoint_ids) {
+  ByteArray bytes = parser::ForPayloadAckPayloadTransfer(payload_id);
+  PacketMetaData packet_meta_data;
+
+  return SendTransferFrameBytes(
+      endpoint_ids, bytes, payload_id,
+      /* offset= */ -1,
+      /*packet_type=*/
+      PayloadTransferFrame::PacketType_Name(PayloadTransferFrame::PAYLOAD_ACK),
+      packet_meta_data);
+}
+
+
 std::vector<std::string> EndpointManager::SendTransferFrameBytes(
     const std::vector<std::string>& endpoint_ids, const ByteArray& bytes,
     std::int64_t payload_id, std::int64_t offset,
diff --git a/connections/implementation/endpoint_manager.h b/connections/implementation/endpoint_manager.h
index 2081c22c..8b2b0b81 100644
--- a/connections/implementation/endpoint_manager.h
+++ b/connections/implementation/endpoint_manager.h
@@ -141,7 +141,10 @@ class EndpointManager {
       const location::nearby::connections::PayloadTransferFrame::ControlMessage&
           control_message,
       const std::vector<std::string>& endpoint_ids);
-
+  // Receiver sends this frame when all the packets are received. Returns the
+  // list of endpoints to which sending this frame failed.
+  std::vector<std::string> SendPayloadAck(
+      std::int64_t payload_id, const std::vector<std::string>& endpoint_ids);
   // Called when we internally want to get rid of the endpoint, without the
   // client directly telling us to. For example...
   //    a) We failed to read from the endpoint in its dedicated reader thread.
diff --git a/connections/implementation/endpoint_manager_test.cc b/connections/implementation/endpoint_manager_test.cc
index d2047130..89f83ac1 100644
--- a/connections/implementation/endpoint_manager_test.cc
+++ b/connections/implementation/endpoint_manager_test.cc
@@ -15,6 +15,7 @@
 #include "connections/implementation/endpoint_manager.h"
 
 #include <atomic>
+#include <cstdint>
 #include <memory>
 #include <string>
 #include <utility>
@@ -31,12 +32,15 @@
 #include "connections/implementation/endpoint_channel_manager.h"
 #include "connections/implementation/flags/nearby_connections_feature_flags.h"
 #include "connections/implementation/offline_frames.h"
+#include "connections/listeners.h"
+#include "connections/status.h"
 #include "internal/flags/nearby_flags.h"
 #include "internal/platform/byte_array.h"
 #include "internal/platform/count_down_latch.h"
 #include "internal/platform/exception.h"
 // #include "internal/platform/feature_flags.h"
 #include "internal/platform/logging.h"
+#include "internal/platform/single_thread_executor.h"
 #include "internal/test/fake_single_thread_executor.h"
 #include "proto/connections_enums.pb.h"
 
@@ -122,11 +126,24 @@ class MockFrameProcessor : public EndpointManager::FrameProcessor {
 
 class SetSafeToDisconnect {
  public:
-  explicit SetSafeToDisconnect(bool safe_to_disconnect) {
+  SetSafeToDisconnect(bool safe_to_disconnect, bool auto_reconnect,
+                               bool payload_received_ack,
+                               std::int32_t safe_to_disconnect_version) {
     NearbyFlags::GetInstance().OverrideBoolFlagValue(
         config_package_nearby::nearby_connections_feature::
             kEnableSafeToDisconnect,
         safe_to_disconnect);
+    NearbyFlags::GetInstance().OverrideBoolFlagValue(
+        config_package_nearby::nearby_connections_feature::kEnableAutoReconnect,
+        auto_reconnect);
+    NearbyFlags::GetInstance().OverrideBoolFlagValue(
+        config_package_nearby::nearby_connections_feature::
+            kEnablePayloadReceivedAck,
+        payload_received_ack);
+    NearbyFlags::GetInstance().OverrideInt64FlagValue(
+        config_package_nearby::nearby_connections_feature::
+            kSafeToDisconnectVersion,
+        safe_to_disconnect_version);
   }
 };
 
@@ -155,12 +172,12 @@ class EndpointManagerTest : public ::testing::Test {
     EXPECT_CALL(mock_listener_.initiated_cb, Call).Times(1);
     em_.RegisterEndpoint(client_.get(), endpoint_id_, info_,
                          connection_options_, std::move(channel), listener_,
-                         connection_token);
+                         connection_token_);
     if (should_close) {
       EXPECT_TRUE(done.Await(absl::Milliseconds(1000)).result());
     }
   }
-  SetSafeToDisconnect set_safe_to_disconnect_{true};
+  SetSafeToDisconnect set_safe_to_disconnect_{true, false, true, 5};
   std::unique_ptr<ClientProxy> client_ = std::make_unique<ClientProxy>();
   ConnectionOptions connection_options_{
       .keep_alive_interval_millis = 5000,
@@ -198,7 +215,7 @@ class EndpointManagerTest : public ::testing::Test {
       .bandwidth_changed_cb =
           mock_listener_.bandwidth_changed_cb.AsStdFunction(),
   };
-  std::string connection_token = "conntokn";
+  std::string connection_token_ = "conntokn";
   absl::Time start_time_{absl::Now()};
 };
 
@@ -293,7 +310,7 @@ TEST_F(EndpointManagerTest, UnregisterFrameProcessorWorks) {
   em_.UnregisterEndpoint(client_.get(), endpoint_id_);
 }
 
-TEST_F(EndpointManagerTest, SendControlMessageWorks) {
+TEST_F(EndpointManagerTest, SendControlMessageAndPayloadAckWorks) {
   auto endpoint_channel = std::make_unique<MockEndpointChannel>();
   PayloadTransferFrame::PayloadHeader header;
   PayloadTransferFrame::ControlMessage control;
@@ -322,9 +339,12 @@ TEST_F(EndpointManagerTest, SendControlMessageWorks) {
       .WillRepeatedly(Return(Exception{Exception::kSuccess}));
 
   RegisterEndpoint(std::move(endpoint_channel), false);
-  auto failed_ids =
+  auto failed_ids_1 =
       em_.SendControlMessage(header, control, std::vector{endpoint_id_});
-  EXPECT_EQ(failed_ids, std::vector<std::string>{});
+  EXPECT_EQ(failed_ids_1, std::vector<std::string>{});
+  auto failed_ids_2 = em_.SendPayloadAck(header.id(),
+      std::vector<std::string>{endpoint_id_});
+  EXPECT_EQ(failed_ids_2, std::vector<std::string>{});
   NEARBY_LOG(INFO, "Will unregister endpoint now");
   em_.UnregisterEndpoint(client_.get(), endpoint_id_);
   NEARBY_LOG(INFO, "Will call destructors now");
diff --git a/connections/implementation/offline_frames.cc b/connections/implementation/offline_frames.cc
index ca468be2..03ec322c 100644
--- a/connections/implementation/offline_frames.cc
+++ b/connections/implementation/offline_frames.cc
@@ -21,6 +21,7 @@
 #include <vector>
 
 #include "connections/implementation/flags/nearby_connections_feature_flags.h"
+#include "connections/implementation/internal_payload.h"
 #include "connections/implementation/offline_frames_validator.h"
 #include "connections/implementation/proto/offline_wire_formats.pb.h"
 #include "connections/medium_selector.h"
@@ -227,6 +228,23 @@ ByteArray ForControlPayloadTransfer(
   return ToBytes(std::move(frame));
 }
 
+ByteArray ForPayloadAckPayloadTransfer(std::int64_t payload_id) {
+  OfflineFrame frame;
+
+  frame.set_version(OfflineFrame::V1);
+  auto* v1_frame = frame.mutable_v1();
+  v1_frame->set_type(V1Frame::PAYLOAD_TRANSFER);
+  auto* sub_frame = v1_frame->mutable_payload_transfer();
+  sub_frame->set_packet_type(PayloadTransferFrame::PAYLOAD_ACK);
+
+  PayloadTransferFrame::PayloadHeader header;
+  header.set_id(payload_id);
+  header.set_total_size(InternalPayload::kIndeterminateSize);
+  *sub_frame->mutable_payload_header() = header;
+
+  return ToBytes(std::move(frame));
+}
+
 ByteArray ForBwuWifiHotspotPathAvailable(const std::string& ssid,
                                          const std::string& password,
                                          std::int32_t port,
diff --git a/connections/implementation/offline_frames.h b/connections/implementation/offline_frames.h
index c34fcc85..33b4a2a6 100644
--- a/connections/implementation/offline_frames.h
+++ b/connections/implementation/offline_frames.h
@@ -66,6 +66,7 @@ ByteArray ForControlPayloadTransfer(
         header,
     const location::nearby::connections::PayloadTransferFrame::ControlMessage&
         control);
+ByteArray ForPayloadAckPayloadTransfer(std::int64_t payload_id);
 
 // Builds Bandwidth Upgrade [BWU] messages.
 ByteArray ForBwuIntroduction(const std::string& endpoint_id,
diff --git a/connections/implementation/offline_frames_test.cc b/connections/implementation/offline_frames_test.cc
index f70723cb..af932d18 100644
--- a/connections/implementation/offline_frames_test.cc
+++ b/connections/implementation/offline_frames_test.cc
@@ -17,7 +17,6 @@
 #include <array>
 #include <memory>
 #include <string>
-#include <utility>
 #include <vector>
 
 #include "gmock/gmock.h"
@@ -337,6 +336,24 @@ TEST(OfflineFramesTest, CanGenerateDataPayloadTransfer) {
   EXPECT_THAT(message, EqualsProto(kExpected));
 }
 
+TEST(OfflineFramesTest, CanGeneratePayloadAckPayloadTransfer) {
+  constexpr absl::string_view kExpected =
+      R"pb(
+    version: V1
+    v1: <
+      type: PAYLOAD_TRANSFER
+      payload_transfer: <
+        packet_type: PAYLOAD_ACK,
+        payload_header: < id: 12345 total_size: -1 >
+      >
+    >)pb";
+  ByteArray bytes = ForPayloadAckPayloadTransfer(12345);
+  auto response = FromBytes(bytes);
+  ASSERT_TRUE(response.ok());
+  OfflineFrame message = response.result();
+  EXPECT_THAT(message, EqualsProto(kExpected));
+}
+
 TEST(OfflineFramesTest, CanGenerateBwuWifiHotspotPathAvailable) {
   constexpr absl::string_view kExpected =
       R"pb(
diff --git a/connections/implementation/offline_frames_validator.cc b/connections/implementation/offline_frames_validator.cc
index 5df7a1d3..9be7d18d 100644
--- a/connections/implementation/offline_frames_validator.cc
+++ b/connections/implementation/offline_frames_validator.cc
@@ -21,6 +21,7 @@
 #include "connections/implementation/internal_payload.h"
 #include "connections/implementation/offline_frames.h"
 #include "connections/implementation/proto/offline_wire_formats.pb.h"
+#include "internal/platform/exception.h"
 #include "internal/platform/implementation/platform.h"
 #include "internal/platform/logging.h"
 
@@ -126,6 +127,11 @@ Exception EnsureValidPayloadTransferControlFrame(
 
 Exception EnsureValidPayloadTransferFrame(const PayloadTransferFrame& frame) {
   if (!frame.has_payload_header()) return {Exception::kInvalidProtocolBuffer};
+  if (frame.packet_type() == PayloadTransferFrame::PAYLOAD_ACK) {
+    // Phone side code doesn't set "total_size" for "payload_header", so skip
+    // checking it.
+    return {Exception::kSuccess};
+  }
   if (!frame.payload_header().has_total_size() ||
       (frame.payload_header().total_size() < 0 &&
        frame.payload_header().total_size() !=
diff --git a/connections/implementation/payload_manager.cc b/connections/implementation/payload_manager.cc
index 349b8cb2..481922d8 100644
--- a/connections/implementation/payload_manager.cc
+++ b/connections/implementation/payload_manager.cc
@@ -15,6 +15,7 @@
 #include "connections/implementation/payload_manager.h"
 
 #include <algorithm>
+#include <cstddef>
 #include <cstdint>
 #include <limits>
 #include <memory>
@@ -32,8 +33,11 @@
 #include "connections/implementation/endpoint_channel_manager.h"
 #include "connections/implementation/flags/nearby_connections_feature_flags.h"
 #include "connections/implementation/internal_payload_factory.h"
+#include "connections/implementation/proto/offline_wire_formats.pb.h"
+#include "connections/listeners.h"
 #include "connections/payload_type.h"
 #include "internal/flags/nearby_flags.h"
+#include "internal/platform/byte_array.h"
 #include "internal/platform/count_down_latch.h"
 #include "internal/platform/exception.h"
 #include "internal/platform/feature_flags.h"
@@ -59,7 +63,7 @@ constexpr absl::Duration PayloadManager::kWaitCloseTimeout;
 bool PayloadManager::SendPayloadLoop(
     ClientProxy* client, PendingPayload& pending_payload,
     PayloadTransferFrame::PayloadHeader& payload_header,
-    std::int64_t& next_chunk_offset, size_t resume_offset) {
+    std::int64_t& next_chunk_offset, size_t resume_offset, int index) {
   // in lieu of structured binding:
   auto pair = GetAvailableAndUnavailableEndpoints(pending_payload);
   const EndpointIds& available_endpoint_ids =
@@ -151,7 +155,7 @@ bool PayloadManager::SendPayloadLoop(
   // In other cases, the offset should only be used in both side logs when error
   // happened.
   PayloadTransferFrame::PayloadChunk payload_chunk(CreatePayloadChunk(
-      next_chunk_offset - resume_offset, std::move(next_chunk)));
+      next_chunk_offset - resume_offset, std::move(next_chunk), index));
   const EndpointIds& failed_endpoint_ids = endpoint_manager_->SendPayloadChunk(
       payload_header, payload_chunk, available_endpoint_ids, packet_meta_data);
   // Check whether at least one endpoint failed.
@@ -349,6 +353,7 @@ PayloadManager::~PayloadManager() {
   bytes_payload_executor_.Shutdown();
   stream_payload_executor_.Shutdown();
   file_payload_executor_.Shutdown();
+  send_payload_ack_executor_.Shutdown();
 
   CountDownLatch stop_latch(1);
   // Clear our tracked pending payloads.
@@ -459,6 +464,7 @@ void PayloadManager::SendPayload(ClientProxy* client,
 
         bool should_continue = true;
         std::int64_t next_chunk_offset = 0;
+        int index = 0;
 
         ThroughputRecorderContainer::GetInstance()
             .GetTPRecorder(payload_id, PayloadDirection::OUTGOING_PAYLOAD)
@@ -466,7 +472,8 @@ void PayloadManager::SendPayload(ClientProxy* client,
         while (should_continue && !shutdown_.Get()) {
           should_continue =
               SendPayloadLoop(client, *pending_payload, payload_header,
-                              next_chunk_offset, resume_offset);
+                              next_chunk_offset, resume_offset, index);
+          index++;
         }
 
         RunOnStatusUpdateThread("destroy-payload",
@@ -525,6 +532,11 @@ void PayloadManager::OnIncomingFrame(
       ProcessDataPacket(to_client, from_endpoint_id, frame, current_medium,
                         packet_meta_data);
       break;
+    case PayloadTransferFrame::PAYLOAD_ACK:
+      NEARBY_LOGS(INFO) << "[safe-to-disconnect][PAYLOAD_RECEIVED_ACK] sender "
+                           "received payload ack from " << from_endpoint_id;
+      ProcessPayloadAckPacket(from_endpoint_id, frame);
+      break;
     default:
       NEARBY_LOGS(WARNING)
           << "PayloadManager: invalid frame; remote endpoint: self=" << this
@@ -694,7 +706,8 @@ PayloadTransferFrame::PayloadHeader PayloadManager::CreatePayloadHeader(
 }
 
 PayloadTransferFrame::PayloadChunk PayloadManager::CreatePayloadChunk(
-    std::int64_t payload_chunk_offset, ByteArray payload_chunk_body) {
+    std::int64_t payload_chunk_offset, ByteArray payload_chunk_body,
+    int index) {
   PayloadTransferFrame::PayloadChunk payload_chunk;
 
   payload_chunk.set_offset(payload_chunk_offset);
@@ -705,6 +718,7 @@ PayloadTransferFrame::PayloadChunk PayloadManager::CreatePayloadChunk(
     payload_chunk.set_flags(payload_chunk.flags() |
                             PayloadTransferFrame::PayloadChunk::LAST_CHUNK);
   }
+  payload_chunk.set_index(index);
 
   return payload_chunk;
 }
@@ -831,21 +845,24 @@ void PayloadManager::SendControlMessage(
 
 void PayloadManager::SendPayloadReceivedAck(
     ClientProxy* client, PendingPayload& pending_payload,
-    const std::string& endpoint_id,
-    const PayloadTransferFrame::PayloadHeader& payload_header,
-    std::int64_t chunk_size, bool is_last_chunk) {
+    const std::string& endpoint_id, bool is_last_chunk) {
   if (!is_last_chunk ||
       !IsPayloadReceivedAckEnabled(client, endpoint_id, pending_payload)) {
     return;
   }
-  // Send the PAYLOAD_RECEIVED_ACK to the remote endpoint for the sender asap.
-  NEARBY_LOGS(INFO)
-      << "[PAYLOAD_RECEIVED_ACK] isLastChunk, receiver send ack to "
-      << endpoint_id;
 
-  SendControlMessage(
-      {endpoint_id}, payload_header, chunk_size,
-      PayloadTransferFrame::ControlMessage::PAYLOAD_RECEIVED_ACK);
+  send_payload_ack_executor_.Execute(
+      "send_payload_ack", [this, &pending_payload, endpoint_id]() {
+        endpoint_manager_->SendPayloadAck(pending_payload.GetId(),
+                                          {endpoint_id});
+        NEARBY_LOGS(INFO) << "[safe-to-disconnect] Send "
+                             "PAYLOAD_RECEIVED_ACK frame to: "
+                          << endpoint_id << " done";
+      });
+  // Send the PAYLOAD_RECEIVED_ACK to the remote endpoint for the sender asap.
+  NEARBY_LOGS(INFO) << "[safe-to-disconnect] [PAYLOAD_RECEIVED_ACK] "
+                       "isLastChunk, receiver send payload ack to "
+                    << endpoint_id;
 }
 
 bool PayloadManager::WaitForReceivedAck(
@@ -866,11 +883,17 @@ bool PayloadManager::WaitForReceivedAck(
         GetPayload(payload_header.id());
     // Make sure we're still tracking this payload and its associated endpoint.
     if (!latest_pending_payload) {
+      NEARBY_LOGS(INFO) << "[safe-to-disconnect] short-circuiting "
+                           "latest_pending_payload is null for "
+                        << payload_header.id() << ", stop wait ack.";
       return false;
     }
 
     auto* endpoint_info = latest_pending_payload->GetEndpoint(endpoint_id);
     if (endpoint_info == nullptr) {
+      NEARBY_LOGS(INFO) << "[safe-to-disconnect] short-circuiting "
+                           "endpointInfo is null for "
+                        << payload_header.id() << ", stop wait ack.";
       return false;
     }
 
@@ -880,6 +903,9 @@ bool PayloadManager::WaitForReceivedAck(
                                     payload_chunk_offset,
                                     location::nearby::proto::connections::
                                         PayloadStatus::LOCAL_CANCELLATION);
+      NEARBY_LOGS(INFO) << "[safe-to-disconnect] short-circuiting local "
+                           "payload cancellation for "
+                        << payload_header.id() << ", stop wait ack.";
       return false;
     }
     // Remote payload cancellation, etc
@@ -888,6 +914,9 @@ bool PayloadManager::WaitForReceivedAck(
       HandleFinishedOutgoingPayload(
           client, {endpoint_id}, payload_header, payload_chunk_offset,
           EndpointInfoStatusToPayloadStatus(endpoint_info->status.Get()));
+      NEARBY_LOGS(INFO) << "[safe-to-disconnect] short-circuiting remote "
+                           "payload cancellation for "
+                        << payload_header.id() << ", stop wait ack.";
       return false;
     }
     {
@@ -902,8 +931,15 @@ bool PayloadManager::WaitForReceivedAck(
               .wait_payload_received_ack_millis);
       endpoint_info->is_payload_received_ack = false;
       if (!wait_exception.Ok()) {
+        NEARBY_LOGS(INFO)
+            << "[safe-to-disconnect][PAYLOAD_RECEIVED_ACK] sender wait for "
+               "received payload ack from "
+            << endpoint_id << " end with exception: " << wait_exception.value;
         return false;
       }
+      NEARBY_LOGS(INFO) << "[safe-to-disconnect][PAYLOAD_RECEIVED_ACK] sender "
+                           "received payload ack from "
+                        << endpoint_id;
       return true;
     }
   }
@@ -1261,8 +1297,7 @@ void PayloadManager::ProcessDataPacket(
   bool is_last_chunk = (payload_chunk.flags() &
                         PayloadTransferFrame::PayloadChunk::LAST_CHUNK) != 0;
   SendPayloadReceivedAck(
-      to_client, *pending_payload, from_endpoint_id, payload_header,
-      payload_chunk.offset() + payload_body_size, is_last_chunk);
+      to_client, *pending_payload, from_endpoint_id, is_last_chunk);
 
   HandleSuccessfulIncomingChunk(to_client, from_endpoint_id, payload_header,
                                 payload_chunk.flags(), payload_chunk.offset(),
@@ -1330,17 +1365,6 @@ void PayloadManager::ProcessControlPacket(
                                                              control_message);
       }
       break;
-    case PayloadTransferFrame::ControlMessage::PAYLOAD_RECEIVED_ACK:
-      if (!pending_payload->IsIncoming() &&
-          IsPayloadReceivedAckEnabled(to_client, from_endpoint_id,
-                                      *pending_payload)) {
-        NEARBY_LOGS(INFO) << "[safe-to-disconnect]Sender received "
-                             "PAYLOAD_RECEIVED_ACK frame with id:"
-                          << pending_payload->GetInternalPayload()->GetId()
-                          << " from endpoint_id=" << from_endpoint_id;
-        pending_payload->MarkReceivedAckFromEndpoint(from_endpoint_id);
-      }
-      break;
     default:
       NEARBY_LOGS(INFO) << "Unhandled control message "
                         << control_message.event() << " for payload_id="
@@ -1349,6 +1373,30 @@ void PayloadManager::ProcessControlPacket(
   }
 }
 
+void PayloadManager::ProcessPayloadAckPacket(
+    const std::string& from_endpoint_id,
+    PayloadTransferFrame& payload_transfer_frame) {
+  auto payload_header = payload_transfer_frame.payload_header();
+  PendingPayloadHandle pending_payload = GetPayload(payload_header.id());
+  if (!pending_payload) {
+    NEARBY_LOGS(INFO) << "[safe-to-disconnect][PAYLOAD_RECEIVED_ACK] "
+                         "short-circuiting got payload "
+                         "ack for unknown payload "
+                      << payload_header.id() << ", ignoring";
+    return;
+  }
+  if (pending_payload->IsIncoming()) {
+    NEARBY_LOGS(INFO) << "[safe-to-disconnect][PAYLOAD_RECEIVED_ACK] "
+                         "short-circuiting got Payload "
+                         "ack for incoming payload "
+                      << payload_header.id() << ", ignoring";
+  }
+  NEARBY_LOGS(INFO)
+      << "[safe-to-disconnect][PAYLOAD_RECEIVED_ACK] sender received payload "
+      << payload_header.id() << " ack from " << from_endpoint_id;
+  pending_payload->MarkReceivedAckFromEndpoint(from_endpoint_id);
+}
+
 // @PayloadManagerStatusUpdateThread
 void PayloadManager::NotifyClientOfIncomingPayloadProgressInfo(
     ClientProxy* client, const std::string& endpoint_id,
diff --git a/connections/implementation/payload_manager.h b/connections/implementation/payload_manager.h
index 768841d1..29d178c0 100644
--- a/connections/implementation/payload_manager.h
+++ b/connections/implementation/payload_manager.h
@@ -31,6 +31,7 @@
 #include "connections/implementation/internal_payload.h"
 #include "connections/listeners.h"
 #include "connections/payload.h"
+#include "connections/payload_type.h"
 #include "connections/status.h"
 #include "internal/platform/atomic_boolean.h"
 #include "internal/platform/atomic_reference.h"
@@ -38,6 +39,7 @@
 #include "internal/platform/condition_variable.h"
 #include "internal/platform/count_down_latch.h"
 #include "internal/platform/mutex.h"
+#include "internal/platform/single_thread_executor.h"
 
 namespace nearby {
 namespace connections {
@@ -270,7 +272,8 @@ class PayloadManager : public EndpointManager::FrameProcessor {
 
   bool SendPayloadLoop(ClientProxy* client, PendingPayload& pending_payload,
                        PayloadTransferFrame::PayloadHeader& payload_header,
-                       std::int64_t& next_chunk_offset, size_t resume_offset);
+                       std::int64_t& next_chunk_offset, size_t resume_offset,
+                       int index);
   void SendClientCallbacksForFinishedIncomingPayloadRunnable(
       ClientProxy* client, const std::string& endpoint_id,
       const PayloadTransferFrame::PayloadHeader& payload_header,
@@ -298,7 +301,8 @@ class PayloadManager : public EndpointManager::FrameProcessor {
       const std::string& parent_folder, const std::string& file_name);
 
   PayloadTransferFrame::PayloadChunk CreatePayloadChunk(std::int64_t offset,
-                                                        ByteArray body);
+                                                        ByteArray body,
+                                                        int index);
   bool IsLastChunk(PayloadTransferFrame::PayloadChunk payload_chunk) {
     return ((payload_chunk.flags() &
              PayloadTransferFrame::PayloadChunk::LAST_CHUNK) != 0);
@@ -331,9 +335,7 @@ class PayloadManager : public EndpointManager::FrameProcessor {
 
   void SendPayloadReceivedAck(
       ClientProxy* client, PendingPayload& pending_payload,
-      const std::string& endpoint_id,
-      const PayloadTransferFrame::PayloadHeader& payload_header,
-      std::int64_t chunk_size, bool is_last_chunk);
+      const std::string& endpoint_id, bool is_last_chunk);
 
   bool WaitForReceivedAck(
       ClientProxy* client, const std::string& endpoint_id,
@@ -377,6 +379,8 @@ class PayloadManager : public EndpointManager::FrameProcessor {
   void ProcessControlPacket(ClientProxy* to_client,
                             const std::string& from_endpoint_id,
                             PayloadTransferFrame& payload_transfer_frame);
+  void ProcessPayloadAckPacket(const std::string& from_endpoint_id,
+                            PayloadTransferFrame& payload_transfer_frame);
 
   void NotifyClientOfIncomingPayloadProgressInfo(
       ClientProxy* client, const std::string& endpoint_id,
@@ -420,6 +424,7 @@ class PayloadManager : public EndpointManager::FrameProcessor {
   SingleThreadExecutor file_payload_executor_;
   SingleThreadExecutor stream_payload_executor_;
   SingleThreadExecutor payload_status_update_executor_;
+  SingleThreadExecutor send_payload_ack_executor_;
   PendingPayloads pending_payloads_;
   EndpointManager* endpoint_manager_;
 
diff --git a/connections/implementation/simulation_user.h b/connections/implementation/simulation_user.h
index 9fba5c1b..417ab2f8 100644
--- a/connections/implementation/simulation_user.h
+++ b/connections/implementation/simulation_user.h
@@ -81,7 +81,7 @@ class SimulationUser {
   SimulationUser(const std::string& device_name,
                  BooleanMediumSelector allowed = BooleanMediumSelector(),
                  SetSafeToDisconnect set_safe_to_disconnect =
-                     SetSafeToDisconnect(true, false, true, 2))
+                     SetSafeToDisconnect(true, false, true, 5))
       : info_{ByteArray{device_name}},
         advertising_options_{
             {
diff --git a/internal/platform/feature_flags.h b/internal/platform/feature_flags.h
index 43d602e1..544a0d90 100644
--- a/internal/platform/feature_flags.h
+++ b/internal/platform/feature_flags.h
@@ -62,6 +62,7 @@ class FeatureFlags {
     // requested service id before attempting to connect over rfcomm. SDP fails
     // on Windows when connecting to FP service id but the rfcomm is successful.
     bool skip_service_discovery_before_connecting_to_rfcomm = false;
+
     std::int32_t min_nc_version_supports_safe_to_disconnect = 1;
     std::int32_t min_nc_version_supports_auto_reconnect = 3;
     absl::Duration auto_reconnect_retry_delay_millis = absl::Milliseconds(5000);