Move accept timeout into incoming share session.

PiperOrigin-RevId: 651932393

sharing/BUILD

@@ -201,6 +201,7 @@ cc_library(
         ":connection_types",
         ":incoming_frame_reader",
         ":paired_key_verification_runner",
+        ":thread_timer",
         ":transfer_metadata",
         ":types",
         "//internal/base:files",
@@ -813,6 +814,7 @@ cc_test(
         ":share_session",
         ":test_support",
         ":transfer_metadata",
+        ":transfer_metadata_matchers",
         ":types",
         "//internal/analytics:mock_event_logger",
         "//internal/platform/implementation/g3",  # fixdeps: keep

sharing/incoming_share_session.cc

@@ -42,6 +42,7 @@
 #include "sharing/share_session.h"
 #include "sharing/share_target.h"
 #include "sharing/text_attachment.h"
+#include "sharing/thread_timer.h"
 #include "sharing/transfer_metadata.h"
 #include "sharing/transfer_metadata_builder.h"
 #include "sharing/wifi_credentials_attachment.h"
@@ -52,6 +53,7 @@ namespace {
 using ::location::nearby::proto::sharing::OSType;
 using ::nearby::sharing::service::proto::ConnectionResponseFrame;
 using ::nearby::sharing::service::proto::IntroductionFrame;
+using ::nearby::sharing::service::proto::ProgressUpdateFrame;
 using ::location::nearby::proto::sharing::ResponseToIntroduction;
 using ::nearby::sharing::service::proto::V1Frame;
 using ::nearby::sharing::service::proto::WifiCredentials;
@@ -182,6 +184,7 @@ bool IncomingShareSession::ProcessKeyVerificationResult(
 }
 
 bool IncomingShareSession::ReadyForTransfer(
+    std::function<void()> accept_timeout_callback,
     std::function<void(std::optional<V1Frame> frame)> frame_read_callback) {
   if (!IsConnected()) {
     NL_LOG(WARNING) << __func__ << ": out of order API call.";
@@ -191,6 +194,9 @@ bool IncomingShareSession::ReadyForTransfer(
   set_disconnect_status(
       TransferMetadata::Status::kUnexpectedDisconnection);
 
+  mutual_acceptance_timeout_ = std::make_unique<ThreadTimer>(
+      service_thread(), "incoming_mutual_acceptance_timeout",
+      kReadResponseFrameTimeout, std::move(accept_timeout_callback));
   frames_reader()->ReadFrame(std::move(frame_read_callback));
 
   if (!self_share()) {
@@ -256,6 +262,35 @@ bool IncomingShareSession::AcceptTransfer(
   return true;
 }
 
+void IncomingShareSession::HandleProgressUpdate(
+    NearbyConnectionsManager& connections_manager,
+    const ProgressUpdateFrame& progress_update) {
+  if (!IsConnected()) {
+    NL_LOG(ERROR) << "Received ProgressUpdate Frame on disconnected session";
+    return;
+  }
+  if (progress_update.start_transfer()) {
+    // Cancel timeout when progress update is received.
+    mutual_acceptance_timeout_.reset();
+    NL_LOG(INFO) << __func__ << ": Received progress for ShareTarget "
+                 << share_target().id;
+    // TODO(b/338468927): Check if this is actually needed.
+    // Bandwidth upgrade was already requested in Accept.
+    if (TryUpgradeBandwidth(connections_manager)) {
+      NL_LOG(INFO)
+          << __func__
+          << ": Upgrade bandwidth when receiving progress update frame "
+             "for endpoint "
+          << endpoint_id();
+    }
+  }
+
+  if (progress_update.has_progress()) {
+    NL_VLOG(1) << __func__ << ": Current progress for ShareTarget "
+               << share_target().id << " is " << progress_update.progress();
+  }
+}
+
 bool IncomingShareSession::UpdateFilePayloadPaths(
     const NearbyConnectionsManager& connections_manager) {
   AttachmentContainer& container = mutable_attachment_container();
@@ -415,4 +450,33 @@ bool IncomingShareSession::TryUpgradeBandwidth(
   return false;
 }
 
+void IncomingShareSession::SendFailureResponse(
+    TransferMetadata::Status status) {
+  // Send response to remote device.
+  ConnectionResponseFrame::Status response_status;
+  switch (status) {
+    case TransferMetadata::Status::kNotEnoughSpace:
+      response_status = ConnectionResponseFrame::NOT_ENOUGH_SPACE;
+      break;
+
+    case TransferMetadata::Status::kUnsupportedAttachmentType:
+      response_status = ConnectionResponseFrame::UNSUPPORTED_ATTACHMENT_TYPE;
+      break;
+
+    case TransferMetadata::Status::kTimedOut:
+      response_status = ConnectionResponseFrame::TIMED_OUT;
+      break;
+
+    default:
+      response_status = ConnectionResponseFrame::UNKNOWN;
+      break;
+  }
+
+  WriteResponseFrame(response_status);
+  NL_DCHECK(TransferMetadata::IsFinalStatus(status))
+      << "SendFailureResponse should only be called with a final status";
+  UpdateTransferMetadata(
+      TransferMetadataBuilder().set_status(status).build());
+}
+
 }  // namespace nearby::sharing

sharing/incoming_share_session.h

@@ -18,6 +18,7 @@
 #include <cstdint>
 #include <filesystem>  // NOLINT
 #include <functional>
+#include <memory>
 #include <optional>
 #include <string>
 #include <vector>
@@ -31,6 +32,7 @@
 #include "sharing/proto/wire_format.pb.h"
 #include "sharing/share_session.h"
 #include "sharing/share_target.h"
+#include "sharing/thread_timer.h"
 #include "sharing/transfer_metadata.h"
 
 namespace nearby::sharing {
@@ -76,6 +78,7 @@ class IncomingShareSession : public ShareSession {
   // immediately.
   // Returns false if user needs to accept the transfer.
   bool ReadyForTransfer(
+      std::function<void()> accept_timeout_callback,
       std::function<
           void(std::optional<nearby::sharing::service::proto::V1Frame> frame)>
           frame_read_callback);
@@ -86,6 +89,11 @@ class IncomingShareSession : public ShareSession {
       Clock* clock, NearbyConnectionsManager& connections_manager,
       std::function<void(int64_t, TransferMetadata)> update_callback);
 
+  void HandleProgressUpdate(
+      NearbyConnectionsManager& connections_manager,
+      const nearby::sharing::service::proto::ProgressUpdateFrame&
+          progress_update);
+
   // Once transfer has completed, make payload content available in the
   // corresponding Attachment.
   // Returns true if all payloads were successfully finalized.
@@ -98,6 +106,11 @@ class IncomingShareSession : public ShareSession {
   // Returns true if bandwidth upgrade was requested.
   bool TryUpgradeBandwidth(NearbyConnectionsManager& connections_manager);
 
+  // Send TransferMetadataUpdate with the final |status|.
+  // Map |status| to corresponding ConnectionResponseFrame::Status and send
+  // response to remote device.
+  void SendFailureResponse(TransferMetadata::Status status);
+
  protected:
   void InvokeTransferUpdateCallback(const TransferMetadata& metadata) override;
   bool OnNewConnection(NearbyConnection* connection) override;
@@ -112,6 +125,9 @@ class IncomingShareSession : public ShareSession {
 
   bool bandwidth_upgrade_requested_ = false;
   bool ready_for_accept_ = false;
+  // This alarm is used to disconnect the sharing connection if both sides do
+  // not press accept within the timeout.
+  std::unique_ptr<ThreadTimer> mutual_acceptance_timeout_;
 };
 
 }  // namespace nearby::sharing

sharing/incoming_share_session_test.cc

@@ -28,6 +28,7 @@
 #include "gtest/gtest.h"
 #include "absl/strings/string_view.h"
 #include "absl/time/clock.h"
+#include "absl/time/time.h"
 #include "internal/analytics/mock_event_logger.h"
 #include "internal/analytics/sharing_log_matchers.h"
 #include "internal/test/fake_clock.h"
@@ -47,6 +48,7 @@
 #include "sharing/share_target.h"
 #include "sharing/text_attachment.h"
 #include "sharing/transfer_metadata.h"
+#include "sharing/transfer_metadata_matchers.h"
 #include "sharing/wifi_credentials_attachment.h"
 #include "google/protobuf/text_format.h"
 
@@ -66,6 +68,7 @@ using ::nearby::sharing::service::proto::ConnectionResponseFrame;
 using ::nearby::sharing::service::proto::FileMetadata;
 using ::nearby::sharing::service::proto::Frame;
 using ::nearby::sharing::service::proto::IntroductionFrame;
+using ::nearby::sharing::service::proto::ProgressUpdateFrame;
 using ::nearby::sharing::service::proto::TextMetadata;
 using ::nearby::sharing::service::proto::V1Frame;
 using ::nearby::sharing::service::proto::WifiCredentials;
@@ -594,8 +597,9 @@ TEST_F(IncomingShareSessionTest, ReadyForTransferNotConnected) {
 
   FakeNearbyConnectionsManager connections_manager;
   FakeClock clock;
-  EXPECT_THAT(session_.ReadyForTransfer([](std::optional<V1Frame> frame) {}),
-              IsFalse());
+  EXPECT_THAT(
+      session_.ReadyForTransfer([]() {}, [](std::optional<V1Frame> frame) {}),
+      IsFalse());
 }
 
 TEST_F(IncomingShareSessionTest, ReadyForTransferNotSelfShare) {
@@ -604,9 +608,13 @@ TEST_F(IncomingShareSessionTest, ReadyForTransferNotSelfShare) {
   session_.set_session_id(1234);
   EXPECT_TRUE(
       session_.OnConnected(nearby_sharing_decoder, absl::Now(), &connection));
+  EXPECT_CALL(
+      transfer_metadata_callback_,
+      Call(_, HasStatus(TransferMetadata::Status::kAwaitingLocalConfirmation)));
 
-  EXPECT_THAT(session_.ReadyForTransfer([](std::optional<V1Frame> frame) {}),
-              IsFalse());
+  EXPECT_THAT(
+      session_.ReadyForTransfer([]() {}, [](std::optional<V1Frame> frame) {}),
+      IsFalse());
 }
 
 TEST_F(IncomingShareSessionTest, ReadyForTransferSelfShare) {
@@ -620,9 +628,99 @@ TEST_F(IncomingShareSessionTest, ReadyForTransferSelfShare) {
   session.set_session_id(1234);
   EXPECT_TRUE(
       session.OnConnected(nearby_sharing_decoder, absl::Now(), &connection));
+  EXPECT_CALL(
+      transfer_metadata_callback_,
+      Call(_, HasStatus(TransferMetadata::Status::kAwaitingLocalConfirmation)))
+      .Times(0);
 
-  EXPECT_THAT(session.ReadyForTransfer([](std::optional<V1Frame> frame) {}),
-              IsTrue());
+  EXPECT_THAT(
+      session.ReadyForTransfer([]() {}, [](std::optional<V1Frame> frame) {}),
+      IsTrue());
+}
+
+TEST_F(IncomingShareSessionTest, ReadyForTransferTimeout) {
+  NearbySharingDecoderImpl nearby_sharing_decoder;
+  FakeNearbyConnection connection;
+  session_.set_session_id(1234);
+  EXPECT_TRUE(
+      session_.OnConnected(nearby_sharing_decoder, absl::Now(), &connection));
+  EXPECT_CALL(
+      transfer_metadata_callback_,
+      Call(_, HasStatus(TransferMetadata::Status::kAwaitingLocalConfirmation)));
+  bool accept_timeout_called = false;
+
+  EXPECT_THAT(session_.ReadyForTransfer(
+                  [&accept_timeout_called]() { accept_timeout_called = true; },
+                  [](std::optional<V1Frame> frame) {}),
+              IsFalse());
+  clock_.FastForward(absl::Seconds(60));
+  task_runner_.SyncWithTimeout(absl::Milliseconds(100));
+
+  EXPECT_THAT(accept_timeout_called, IsTrue());
+}
+
+TEST_F(IncomingShareSessionTest, ReadyForTransferTimeoutCancelled) {
+  NearbySharingDecoderImpl nearby_sharing_decoder;
+  FakeNearbyConnection connection;
+  session_.set_session_id(1234);
+  EXPECT_TRUE(
+      session_.OnConnected(nearby_sharing_decoder, absl::Now(), &connection));
+  EXPECT_CALL(
+      transfer_metadata_callback_,
+      Call(_, HasStatus(TransferMetadata::Status::kAwaitingLocalConfirmation)));
+  bool accept_timeout_called = false;
+
+  EXPECT_THAT(session_.ReadyForTransfer(
+                  [&accept_timeout_called]() { accept_timeout_called = true; },
+                  [](std::optional<V1Frame> frame) {}),
+              IsFalse());
+  FakeNearbyConnectionsManager connections_manager;
+  ProgressUpdateFrame progress_update_frame;
+  progress_update_frame.set_start_transfer(true);
+  session_.HandleProgressUpdate(connections_manager, progress_update_frame);
+  clock_.FastForward(absl::Seconds(60));
+  task_runner_.SyncWithTimeout(absl::Milliseconds(100));
+
+  EXPECT_THAT(accept_timeout_called, IsFalse());
+}
+
+TEST_F(IncomingShareSessionTest, HandleProgressUpdateNotConnected) {
+  FakeNearbyConnectionsManager connections_manager;
+  ProgressUpdateFrame progress_update_frame;
+  progress_update_frame.set_start_transfer(true);
+  session_.HandleProgressUpdate(connections_manager, progress_update_frame);
+
+  EXPECT_THAT(connections_manager.DidUpgradeBandwidth(kEndpointId), IsFalse());
+}
+
+TEST_F(IncomingShareSessionTest, HandleProgressUpdateTryUpgradeBandwidth) {
+  IntroductionFrame introduction_frame;
+  NL_CHECK(
+      proto2::TextFormat::ParseFromString(R"pb(
+                                            file_metadata {
+                                              id: 1234
+                                              size: 2000000
+                                              name: "file_name1"
+                                              mime_type: "application/pdf"
+                                              type: DOCUMENT
+                                              parent_folder: "parent_folder1"
+                                              payload_id: 9876
+                                            }
+                                          )pb",
+                                          &introduction_frame));
+  NearbySharingDecoderImpl nearby_sharing_decoder;
+  FakeNearbyConnection connection;
+  session_.set_session_id(1234);
+  EXPECT_TRUE(
+      session_.OnConnected(nearby_sharing_decoder, absl::Now(), &connection));
+  EXPECT_THAT(session_.ProcessIntroduction(introduction_frame),
+              Eq(std::nullopt));
+  FakeNearbyConnectionsManager connections_manager;
+  ProgressUpdateFrame progress_update_frame;
+  progress_update_frame.set_start_transfer(true);
+  session_.HandleProgressUpdate(connections_manager, progress_update_frame);
+
+  EXPECT_THAT(connections_manager.DidUpgradeBandwidth(kEndpointId), IsTrue());
 }
 
 TEST_F(IncomingShareSessionTest, AcceptTransferNotConnected) {
@@ -659,14 +757,12 @@ TEST_F(IncomingShareSessionTest, AcceptTransferSuccess) {
       session_.OnConnected(nearby_sharing_decoder, absl::Now(), &connection));
   EXPECT_THAT(session_.ProcessIntroduction(introduction_frame_),
               Eq(std::nullopt));
-  EXPECT_THAT(session_.ReadyForTransfer([](std::optional<V1Frame> frame) {}),
-              IsFalse());
-  EXPECT_CALL(transfer_metadata_callback_, Call(_, _))
-      .WillOnce(Invoke([](const IncomingShareSession& session,
-                          const TransferMetadata& metadata) {
-        EXPECT_EQ(metadata.status(),
-                  TransferMetadata::Status::kAwaitingRemoteAcceptance);
-      }));
+  EXPECT_THAT(
+      session_.ReadyForTransfer([]() {}, [](std::optional<V1Frame> frame) {}),
+      IsFalse());
+  EXPECT_CALL(
+      transfer_metadata_callback_,
+      Call(_, HasStatus(TransferMetadata::Status::kAwaitingRemoteAcceptance)));
   EXPECT_CALL(
       mock_event_logger_,
       Log(Matcher<const SharingLog&>(AllOf(
@@ -851,6 +947,7 @@ TEST_F(IncomingShareSessionTest, TryUpgradeBandwidthNotNeeded) {
   session_.OnConnected(decoder, absl::Now(), &connection);
 
   EXPECT_THAT(session_.TryUpgradeBandwidth(connections_manager), IsFalse());
+  EXPECT_THAT(connections_manager.DidUpgradeBandwidth(kEndpointId), IsFalse());
 }
 
 TEST_F(IncomingShareSessionTest, TryUpgradeBandwidthNeeded) {
@@ -885,6 +982,34 @@ TEST_F(IncomingShareSessionTest, TryUpgradeBandwidthNeeded) {
               Eq(std::nullopt));
 
   EXPECT_THAT(session_.TryUpgradeBandwidth(connections_manager), IsTrue());
+  EXPECT_THAT(connections_manager.DidUpgradeBandwidth(kEndpointId), IsTrue());
+}
+
+TEST_F(IncomingShareSessionTest, SendFailureResponseNotConnected) {
+  EXPECT_CALL(
+      transfer_metadata_callback_,
+      Call(_, HasStatus(TransferMetadata::Status::kNotEnoughSpace)));
+
+  session_.SendFailureResponse(TransferMetadata::Status::kNotEnoughSpace);
+}
+
+TEST_F(IncomingShareSessionTest, SendFailureResponseConnected) {
+  NearbySharingDecoderImpl decoder;
+  FakeNearbyConnection connection;
+  session_.OnConnected(decoder, absl::Now(), &connection);
+  EXPECT_CALL(
+      transfer_metadata_callback_,
+      Call(_, HasStatus(TransferMetadata::Status::kNotEnoughSpace)));
+
+  session_.SendFailureResponse(TransferMetadata::Status::kNotEnoughSpace);
+
+  std::vector<uint8_t> frame_data = connection.GetWrittenData();
+  Frame frame;
+  ASSERT_TRUE(frame.ParseFromArray(frame_data.data(), frame_data.size()));
+  ASSERT_EQ(frame.version(), Frame::V1);
+  ASSERT_EQ(frame.v1().type(), V1Frame::RESPONSE);
+  EXPECT_EQ(frame.v1().connection_response().status(),
+            ConnectionResponseFrame::NOT_ENOUGH_SPACE);
 }
 
 }  // namespace

sharing/nearby_sharing_service_impl.cc

@@ -2631,36 +2631,7 @@ void NearbySharingServiceImpl::Fail(IncomingShareSession& session,
       absl::bind_front(&NearbySharingServiceImpl::CloseConnection, this,
                        session.share_target().id));
 
-  session.set_disconnect_status(status);
-
-  // Send response to remote device.
-  nearby::sharing::service::proto::ConnectionResponseFrame::Status
-      response_status;
-  switch (status) {
-    case TransferMetadata::Status::kNotEnoughSpace:
-      response_status = nearby::sharing::service::proto::
-          ConnectionResponseFrame::NOT_ENOUGH_SPACE;
-      break;
-
-    case TransferMetadata::Status::kUnsupportedAttachmentType:
-      response_status = nearby::sharing::service::proto::
-          ConnectionResponseFrame::UNSUPPORTED_ATTACHMENT_TYPE;
-      break;
-
-    case TransferMetadata::Status::kTimedOut:
-      response_status =
-          nearby::sharing::service::proto::ConnectionResponseFrame::TIMED_OUT;
-      break;
-
-    default:
-      response_status =
-          nearby::sharing::service::proto::ConnectionResponseFrame::UNKNOWN;
-      break;
-  }
-
-  session.WriteResponseFrame(response_status);
-  session.UpdateTransferMetadata(
-      TransferMetadataBuilder().set_status(status).build());
+  session.SendFailureResponse(status);
 }
 
 void NearbySharingServiceImpl::OnIncomingAdvertisementDecoded(
@@ -2996,13 +2967,9 @@ void NearbySharingServiceImpl::OnReceivedIntroduction(
     }
   }
 
-  std::filesystem::path download_path =
-      std::filesystem::u8path(settings_->GetCustomSavePath());
-
-  bool is_out_of_storage =
-      IsOutOfStorage(device_info_, download_path,
-                     session->attachment_container().GetStorageSize());
-  if (is_out_of_storage) {
+  if (IsOutOfStorage(device_info_,
+                     std::filesystem::u8path(settings_->GetCustomSavePath()),
+                     session->attachment_container().GetStorageSize())) {
     Fail(*session, TransferMetadata::Status::kNotEnoughSpace);
     NL_LOG(WARNING) << __func__
                     << ": Not enough space on the receiver. We have informed "
@@ -3114,32 +3081,28 @@ void NearbySharingServiceImpl::OnReceiveConnectionResponse(
 
 void NearbySharingServiceImpl::OnStorageCheckCompleted(
     IncomingShareSession& session) {
-  mutual_acceptance_timeout_alarm_ = std::make_unique<ThreadTimer>(
-      *service_thread_, "mutual_acceptance_timeout_alarm",
-      kReadResponseFrameTimeout,
-      [this, share_target_id = session.share_target().id]() {
-        NL_VLOG(1)
-            << __func__
-            << ": Incoming mutual acceptance timed out, closing connection for "
-            << share_target_id;
-        IncomingShareSession* session =
-            GetIncomingShareSession(share_target_id);
-        if (session != nullptr) {
-          Fail(*session, TransferMetadata::Status::kTimedOut);
-        }
-      });
-  if (session.ReadyForTransfer(
+  if (!session.ReadyForTransfer(
+          [this, share_target_id = session.share_target().id]() {
+            NL_VLOG(1) << "Incoming mutual acceptance timed out, closing "
+                          "connection for "
+                       << share_target_id;
+            IncomingShareSession* session =
+                GetIncomingShareSession(share_target_id);
+            if (session != nullptr) {
+              Fail(*session, TransferMetadata::Status::kTimedOut);
+            }
+          },
           absl::bind_front(&NearbySharingServiceImpl::OnFrameRead, this,
                            session.share_target().id))) {
-    // Don't need to send kAwaitingLocalConfirmation for auto accept of Self
-    // share.
-    NL_LOG(INFO) << __func__ << ": Auto-accepting self share.";
-    session.AcceptTransfer(
-        context_->GetClock(), *nearby_connections_manager_,
-        absl::bind_front(&NearbySharingServiceImpl::OnPayloadTransferUpdate,
-                         this));
-    OnTransferStarted(/*is_incoming=*/true);
+    return;
   }
+  // Don't need to wait for user to accept for Self share.
+  NL_LOG(INFO) << __func__ << ": Auto-accepting self share.";
+  session.AcceptTransfer(
+      context_->GetClock(), *nearby_connections_manager_,
+      absl::bind_front(&NearbySharingServiceImpl::OnPayloadTransferUpdate,
+                       this));
+  OnTransferStarted(/*is_incoming=*/true);
 }
 
 void NearbySharingServiceImpl::OnFrameRead(
@@ -3194,34 +3157,13 @@ void NearbySharingServiceImpl::HandleProgressUpdateFrame(
     int64_t share_target_id,
     const nearby::sharing::service::proto::ProgressUpdateFrame&
         progress_update_frame) {
-  if (progress_update_frame.has_start_transfer() &&
-      progress_update_frame.start_transfer()) {
-    IncomingShareSession* session = GetIncomingShareSession(share_target_id);
-
-    if (session == nullptr || !session->IsConnected()) {
-      NL_LOG(ERROR) << "Received ProgressUpdate Frame on unknown session";
-      return;
-    }
-    mutual_acceptance_timeout_alarm_.reset();
-    NL_LOG(INFO) << __func__ << ": Received progress for ShareTarget "
-                 << share_target_id << " : "
-                 << progress_update_frame.progress();
-    // TODO(b/338468927): Check if this is actually needed.
-    // Bandwidth upgrade was already requested in Accept.
-    if (session->TryUpgradeBandwidth(*nearby_connections_manager_)) {
-      NL_LOG(INFO)
-          << __func__
-          << ": Upgrade bandwidth when receiving progress update frame "
-             "for endpoint "
-          << session->endpoint_id();
-    }
-  }
-
-  if (progress_update_frame.has_progress()) {
-    NL_VLOG(1) << __func__ << ": Current progress for ShareTarget "
-                 << share_target_id << " is "
-                 << progress_update_frame.progress();
+  IncomingShareSession* session = GetIncomingShareSession(share_target_id);
+  if (session == nullptr) {
+    NL_LOG(ERROR) << "Received ProgressUpdate Frame on unknown session";
+    return;
   }
+  session->HandleProgressUpdate(*nearby_connections_manager_,
+                                progress_update_frame);
 }
 
 void NearbySharingServiceImpl::OnConnectionDisconnected(

sharing/share_session.h

@@ -151,6 +151,7 @@ class ShareSession {
     return analytics_recorder_;
   };
 
+  TaskRunner& service_thread() const { return service_thread_; }
   void SetAttachmentPayloadId(int64_t attachment_id, int64_t payload_id);
 
   void set_payload_tracker(std::shared_ptr<PayloadTracker> payload_tracker) {