pw_transfer: Ignore no-op continuation packets

- Ignore PARAMETERS_CONTINUE packets that do not advance the window.
  This is unexpected, but can be ignored without causing problems.
- End transfers with RESOURCE_EXHAUSTED when the receiver requests zero
  bytes. This indicates that they are out of storage, but the transfer
  may still succeed if there is no more data to send. This aligns
  pw_transfer with the documented error code for this situation.

Change-Id: Ie07f6e8b8ab44d02bff8b62a93b3331b58f1b569
Reviewed-on: https://pigweed-review.googlesource.com/c/pigweed/pigweed/+/88464
Pigweed-Auto-Submit: Wyatt Hepler <hepler@google.com>
Reviewed-by: Tri Pho <tripho@google.com>
Reviewed-by: Keir Mierle <keir@google.com>
Reviewed-by: Alexei Frolov <frolv@google.com>
Commit-Queue: Auto-Submit <auto-submit@pigweed.google.com.iam.gserviceaccount.com>

4 files changed, 64 insertions, 19 deletions

diff --git a/pw_transfer/client_test.cc b/pw_transfer/client_test.cc
index ddc18c31a..cbf82d7ba 100644
--- a/pw_transfer/client_test.cc
+++ b/pw_transfer/client_test.cc
@@ -1171,9 +1171,9 @@ TEST_F(WriteTransfer, AbortIfZeroBytesAreRequested) {
   Chunk c1 = DecodeChunk(payloads[1]);
   EXPECT_EQ(c1.transfer_id, 9u);
   ASSERT_TRUE(c1.status.has_value());
-  EXPECT_EQ(c1.status.value(), Status::Internal());
+  EXPECT_EQ(c1.status.value(), Status::ResourceExhausted());
 
-  EXPECT_EQ(transfer_status, Status::Internal());
+  EXPECT_EQ(transfer_status, Status::ResourceExhausted());
 }
 
 TEST_F(WriteTransfer, Timeout_RetriesWithInitialChunk) {
diff --git a/pw_transfer/context.cc b/pw_transfer/context.cc
index 0e5ebbb93..47a0433bf 100644
--- a/pw_transfer/context.cc
+++ b/pw_transfer/context.cc
@@ -348,10 +348,10 @@ void Context::HandleTransferParametersUpdate(const Chunk& chunk) {
   // Parsed all of the parameters; start sending the window.
   set_transfer_state(TransferState::kTransmitting);
 
-  TransmitNextChunk();
+  TransmitNextChunk(retransmit);
 }
 
-void Context::TransmitNextChunk() {
+void Context::TransmitNextChunk(bool retransmit_requested) {
   ByteSpan buffer = thread_->encode_buffer();
 
   // Begin by doing a partial encode of all the metadata fields, leaving the
@@ -383,16 +383,20 @@ void Context::TransmitNextChunk() {
                  static_cast<unsigned>(transfer_id_));
   } else if (data.ok()) {
     if (offset_ == window_end_offset_) {
-      PW_LOG_DEBUG(
-          "Transfer %u is not finished, but the receiver cannot accept any "
-          "more data (offset == window_end_offset)",
-          static_cast<unsigned>(transfer_id_));
-
-      // TODO(frolv): The transfer shouldn't end here as this may just indicate
-      // that the receiver is temporarily busy. Instead, we should wait to see
-      // if more data is requested.
-      Finish(Status::Internal());
-      return;
+      if (retransmit_requested) {
+        PW_LOG_DEBUG(
+            "Transfer %u: received an empty retransmit request, but there is "
+            "still data to send; aborting with RESOURCE_EXHAUSTED",
+            id_for_log());
+        Finish(Status::ResourceExhausted());
+      } else {
+        PW_LOG_DEBUG(
+            "Transfer %u: ignoring continuation packet for transfer window "
+            "that has already been sent",
+            id_for_log());
+        SetTimeout(chunk_timeout_);
+      }
+      return;  // No data was requested, so there is nothing else to do.
     }
 
     PW_LOG_DEBUG("Transfer %u sending chunk offset=%u size=%u",
@@ -622,7 +626,7 @@ void Context::HandleTimeout() {
       // A timeout occurring in a TRANSMITTING state indicates that the transfer
       // has waited for its inter-chunk delay and should transmit its next
       // chunk.
-      TransmitNextChunk();
+      TransmitNextChunk(/*retransmit_requested=*/false);
       break;
 
     case TransferState::kWaiting:
@@ -691,7 +695,7 @@ void Context::Retry() {
   offset_ = last_chunk_offset_;
   pending_bytes_ += last_size_sent;
 
-  TransmitNextChunk();
+  TransmitNextChunk(/*retransmit_requested=*/false);
 }
 
 uint32_t Context::MaxWriteChunkSize(uint32_t max_chunk_size_bytes,
diff --git a/pw_transfer/public/pw_transfer/internal/context.h b/pw_transfer/public/pw_transfer/internal/context.h
index 0bc956e16..0a4d2b453 100644
--- a/pw_transfer/public/pw_transfer/internal/context.h
+++ b/pw_transfer/public/pw_transfer/internal/context.h
@@ -153,6 +153,12 @@ class Context {
 
   void set_transfer_state(TransferState state) { transfer_state_ = state; }
 
+  // The transfer ID as unsigned instead of uint32_t so it can be used with %u.
+  unsigned id_for_log() const {
+    static_assert(sizeof(unsigned) >= sizeof(transfer_id_));
+    return static_cast<unsigned>(transfer_id_);
+  }
+
   stream::Reader& reader() {
     PW_DASSERT(active() && type() == TransferType::kTransmit);
     return static_cast<stream::Reader&>(*stream_);
@@ -210,7 +216,7 @@ class Context {
   void HandleTransferParametersUpdate(const Chunk& chunk);
 
   // Sends the next chunk in a transmit transfer, if any.
-  void TransmitNextChunk();
+  void TransmitNextChunk(bool retransmit_requested);
 
   // Processes a chunk in a receive transfer.
   void HandleReceiveChunk(const Chunk& chunk);
diff --git a/pw_transfer/transfer_test.cc b/pw_transfer/transfer_test.cc
index 9d2d1d7e0..e2f8e2a40 100644
--- a/pw_transfer/transfer_test.cc
+++ b/pw_transfer/transfer_test.cc
@@ -270,6 +270,41 @@ TEST_F(ReadTransfer, MultiChunk) {
   EXPECT_EQ(handler_.finalize_read_status, OkStatus());
 }
 
+TEST_F(ReadTransfer, MultiChunk_RepeatedContinuePackets) {
+  ctx_.SendClientStream(
+      EncodeChunk({.transfer_id = 3,
+                   .window_end_offset = 16,
+                   .pending_bytes = 16,
+                   .offset = 0,
+                   .type = Chunk::Type::kParametersRetransmit}));
+
+  transfer_thread_.WaitUntilEventIsProcessed();
+
+  const auto continue_chunk =
+      EncodeChunk({.transfer_id = 3,
+                   .window_end_offset = 24,
+                   .pending_bytes = 8,
+                   .offset = 16,
+                   .type = Chunk::Type::kParametersContinue});
+  ctx_.SendClientStream(continue_chunk);
+
+  transfer_thread_.WaitUntilEventIsProcessed();
+
+  // Resend the CONTINUE packets that don't actually advance the window.
+  for (int i = 0; i < 3; ++i) {
+    ctx_.SendClientStream(continue_chunk);
+    transfer_thread_.WaitUntilEventIsProcessed();
+  }
+
+  ASSERT_EQ(ctx_.total_responses(), 2u);  // Only sent one packet
+  Chunk c1 = DecodeChunk(ctx_.responses()[1]);
+
+  EXPECT_EQ(c1.transfer_id, 3u);
+  EXPECT_EQ(c1.offset, 16u);
+  ASSERT_EQ(c1.data.size(), 8u);
+  EXPECT_EQ(std::memcmp(c1.data.data(), kData.data() + 16, c1.data.size()), 0);
+}
+
 TEST_F(ReadTransfer, PendingBytes_MultiChunk) {
   ctx_.SendClientStream(
       EncodeChunk({.transfer_id = 3, .pending_bytes = 16, .offset = 0}));
@@ -601,10 +636,10 @@ TEST_F(ReadTransfer, ZeroPendingBytesWithRemainingData_Aborts) {
 
   ASSERT_EQ(ctx_.total_responses(), 1u);
   ASSERT_TRUE(handler_.finalize_read_called);
-  EXPECT_EQ(handler_.finalize_read_status, Status::Internal());
+  EXPECT_EQ(handler_.finalize_read_status, Status::ResourceExhausted());
 
   Chunk chunk = DecodeChunk(ctx_.responses().back());
-  EXPECT_EQ(chunk.status, Status::Internal());
+  EXPECT_EQ(chunk.status, Status::ResourceExhausted());
 }
 
 TEST_F(ReadTransfer, ZeroPendingBytesNoRemainingData_Completes) {