nfs: consume XDR padding that arrives after a chunk's data

The continuation of a partial WRITE request or READ reply only covered the
file data, not the XDR padding that follows it in the record. When a delivery
of stream data ended on the last byte of file data, or inside the padding,
the remaining padding was parsed as the start of a new RPC record, which
failed and disabled the parser for the flow.

Keep the chunk active while its record still has unconsumed bytes, as tracked
by the record tail, and consume exactly those bytes: from the input that
completes the file data and, for what is not there yet, from the next
deliveries. If the file is still open the padding is passed to its file
tracker, which consumes it as fill bytes, never as file data.

A partial READ reply that already holds all of its file data now also has its
request removed from the request map, as is done when a chunk completes.

Builds on the file handle of the in-progress READ chunk added for #9192.

Ticket: #9194

rust/src/nfs/nfs.rs

@@ -1596,7 +1596,42 @@ impl NFSState {
         }
 
         if chunk_left == 0 {
-            return 0;
+            if chunk_tail == 0 {
+                return 0;
+            }
+            // The chunk's file data is complete, but the rest of its record
+            // (its XDR padding) was not part of that input. Consume it now,
+            // never past the record tail: for a still open file these bytes
+            // are the tracker's fill bytes, never file data.
+            let pad = cmp::min(chunk_tail, data.len() as u32);
+            let file_handle = if direction == Direction::ToServer {
+                self.ts_chunk_fh.to_vec()
+            } else {
+                self.tc_chunk_fh.to_vec()
+            };
+            if let Some(tx) = self.get_file_tx_by_handle(&file_handle, direction) {
+                if let Some(NFSTransactionTypeData::FILE(ref mut tdf)) = tx.type_data {
+                    // only a tracker with an open file owes fill bytes
+                    if !tdf.file_tracker.is_done() {
+                        filetracker_update(&mut tdf.file_tracker, &data[..pad as usize], gap_size);
+                    }
+                }
+            }
+            chunk_tail -= pad;
+            if direction == Direction::ToServer {
+                self.ts_chunk_tail = chunk_tail;
+                if chunk_tail == 0 {
+                    self.ts_chunk_xid = 0;
+                    self.ts_chunk_fh.clear();
+                }
+            } else {
+                self.tc_chunk_tail = chunk_tail;
+                if chunk_tail == 0 {
+                    self.tc_chunk_xid = 0;
+                    self.tc_chunk_fh.clear();
+                }
+            }
+            return pad;
         }
         // bytes of this call's input that belong to the in-progress chunk
         let chunk_total = chunk_left;
@@ -1623,16 +1658,14 @@ impl NFSState {
         // we have the data that we expect
         if chunk_left <= data.len() as u32 {
             chunk_left = 0;
-            chunk_tail = 0;
+            // chunk_tail is reduced by what is consumed below: if the record's
+            // XDR padding is not (all) in this input, the chunk stays active
+            // until it is (see the chunk_left == 0 case above)
 
             if direction == Direction::ToServer {
-                self.ts_chunk_xid = 0;
                 file_handle = self.ts_chunk_fh.to_vec();
-                self.ts_chunk_fh.clear();
             } else {
-                self.tc_chunk_xid = 0;
                 file_handle = self.tc_chunk_fh.to_vec();
-                self.tc_chunk_fh.clear();
 
                 // chunk done, remove requestmap entry (unless evicted)
                 self.requestmap.pop(&xid);
@@ -1766,6 +1799,13 @@ impl NFSState {
         if limit_event.is_none() && consumed > 0 {
             chunk_tail = chunk_tail.saturating_sub(consumed);
         }
+        // once the file data is done, the rest of the record that is in this
+        // input is consumed too, so it is not parsed as an RPC record
+        if limit_event.is_none() && chunk_left == 0 && chunk_tail > 0 {
+            let rest = cmp::min(chunk_tail, (data.len() as u32).saturating_sub(consumed));
+            consumed += rest;
+            chunk_tail -= rest;
+        }
 
         // persist after the limit bookkeeping: the rejected-chunk reset and
         // the normal-path tail decrement both land in the stream state
@@ -1776,6 +1816,16 @@ impl NFSState {
             self.tc_chunk_left = chunk_left;
             self.tc_chunk_tail = chunk_tail;
         }
+        // all of the chunk's record is consumed: forget the chunk
+        if chunk_left == 0 && chunk_tail == 0 {
+            if direction == Direction::ToServer {
+                self.ts_chunk_xid = 0;
+                self.ts_chunk_fh.clear();
+            } else {
+                self.tc_chunk_xid = 0;
+                self.tc_chunk_fh.clear();
+            }
+        }
 
         // Chunk-limit overflow: reset chunk state and skip the promised
         // remainder (event attached to the file tx under the borrow).
@@ -2082,6 +2132,12 @@ impl NFSState {
                 // continuation may consume or skip
                 self.tc_chunk_tail = r.prog_data_size.saturating_sub(r.prog_data.len() as u32);
                 self.tc_chunk_fh = file_handle;
+                // a partial reply that holds all of its data only has its padding
+                // left: done with the request, as at the end of a chunk in
+                // filetracker_update
+                if xidmapr.is_none() && self.tc_chunk_left == 0 {
+                    self.requestmap.pop(&r.hdr.xid);
+                }
             }
         }