Leased parts always have leases now

The only case where this wasn't true was exchanging over a network, and
we no longer use leased part for that.

Author: bkirwi

src/persist-client/src/fetch.rs

@@ -481,7 +481,7 @@ pub struct LeasedBatchPart<T> {
     pub(crate) part: BatchPart<T>,
     /// The lease that prevents this part from being GCed. Code should ensure that this lease
     /// lives as long as the part is needed.
-    pub(crate) lease: Option<Lease>,
+    pub(crate) lease: Lease,
     pub(crate) filter_pushdown_audit: bool,
 }
 
@@ -498,9 +498,9 @@ where
     /// that can't travel across process boundaries. The caller is responsible for
     /// ensuring that the lease is held for as long as the batch part may be in use:
     /// dropping it too early may cause a fetch to fail.
-    pub(crate) fn into_exchangeable_part(mut self) -> (ExchangeableBatchPart<T>, Option<Lease>) {
+    pub(crate) fn into_exchangeable_part(self) -> (ExchangeableBatchPart<T>, Lease) {
         // If `x` has a lease, we've effectively transferred it to `r`.
-        let lease = self.lease.take();
+        let lease = self.lease.clone();
         let part = ExchangeableBatchPart {
             shard_id: self.shard_id,
             encoded_size_bytes: self.part.encoded_size_bytes(),

src/persist-client/src/operators/shard_source.rs

@@ -572,9 +572,7 @@ where
                 // seemed to work okay so far. Continue to revisit as necessary.
                 let worker_idx = usize::cast_from(Instant::now().hashed()) % num_workers;
                 let (part, lease) = part_desc.into_exchangeable_part();
-                if let Some(lease) = lease {
-                    leases.borrow_mut().push_at(current_ts.clone(), lease);
-                }
+                leases.borrow_mut().push_at(current_ts.clone(), lease);
                 descs_output.give(&session_cap, (worker_idx, part));
             }
 

src/persist-client/src/read.rs

@@ -726,7 +726,7 @@ where
             filter,
             desc,
             part,
-            lease: Some(self.lease_seqno()),
+            lease: self.lease_seqno(),
             filter_pushdown_audit: false,
         }
     }
@@ -1450,7 +1450,7 @@ mod tests {
 
         // Repeat the same process as above, more or less, while fetching + returning parts
         for (mut i, part) in parts.into_iter().enumerate() {
-            let part_seqno = part.lease.as_ref().unwrap().seqno();
+            let part_seqno = part.lease.seqno();
             let last_seqno = this_seqno;
             this_seqno = part_seqno;
             assert!(this_seqno >= last_seqno);
@@ -1463,9 +1463,8 @@ mod tests {
             for event in subscribe.next(None).await {
                 if let ListenEvent::Updates(parts) = event {
                     for part in parts {
-                        if let (_, Some(lease)) = part.into_exchangeable_part() {
-                            subsequent_parts.push(lease);
-                        }
+                        let (_, lease) = part.into_exchangeable_part();
+                        subsequent_parts.push(lease);
                     }
                 }
             }