Harden stream queues and hostile network tests
ci / test (push) Has been cancelled
ci / fuzz-smoke (push) Has been cancelled
ci / windows-client (push) Has been cancelled
ci / package-release (linux-x86_64, ubuntu-latest) (push) Has been cancelled
ci / package-release (macos-aarch64, macos-14) (push) Has been cancelled
ci / package-release (macos-x86_64, macos-13) (push) Has been cancelled
ci / package-release (windows-x86_64, windows-latest) (push) Has been cancelled
ci / remote-bench (push) Has been cancelled
ci / test (push) Has been cancelled
ci / fuzz-smoke (push) Has been cancelled
ci / windows-client (push) Has been cancelled
ci / package-release (linux-x86_64, ubuntu-latest) (push) Has been cancelled
ci / package-release (macos-aarch64, macos-14) (push) Has been cancelled
ci / package-release (macos-x86_64, macos-13) (push) Has been cancelled
ci / package-release (windows-x86_64, windows-latest) (push) Has been cancelled
ci / remote-bench (push) Has been cancelled
This commit is contained in:
@@ -6253,7 +6253,9 @@ async fn flush_stream_pending_data(
|
||||
if credit < bytes.len() {
|
||||
break;
|
||||
}
|
||||
let bytes = pending.pop_front().expect("pending front exists");
|
||||
let Some(bytes) = pending.pop_front() else {
|
||||
break;
|
||||
};
|
||||
*stream_send_credit.entry(stream_id).or_default() -= bytes.len();
|
||||
let offset = *stream_next_send_offset.entry(stream_id).or_default();
|
||||
stream_next_send_offset.insert(stream_id, offset.saturating_add(bytes.len() as u64));
|
||||
|
||||
+10
-4
@@ -2752,10 +2752,13 @@ async fn handle_file_request(
|
||||
atomic,
|
||||
});
|
||||
if let Some(mode) = mode {
|
||||
let target = upload
|
||||
let Some(active_upload) = upload.as_ref() else {
|
||||
return Err(anyhow!("upload state missing after start"));
|
||||
};
|
||||
let target = active_upload
|
||||
.temp_path
|
||||
.as_ref()
|
||||
.and_then(|state| state.temp_path.as_ref())
|
||||
.unwrap_or_else(|| &upload.as_ref().unwrap().final_path)
|
||||
.unwrap_or(&active_upload.final_path)
|
||||
.clone();
|
||||
set_mode(&target, mode)?;
|
||||
}
|
||||
@@ -3243,7 +3246,10 @@ async fn flush_stream_pending_data_to_client(
|
||||
if credit < bytes.len() {
|
||||
return Ok(());
|
||||
}
|
||||
let bytes = pending.pop_front().expect("pending front exists");
|
||||
let Some(bytes) = pending.pop_front() else {
|
||||
client.stream_pending_data.remove(&stream_id);
|
||||
return Ok(());
|
||||
};
|
||||
if pending.is_empty() {
|
||||
client.stream_pending_data.remove(&stream_id);
|
||||
}
|
||||
|
||||
+4
-2
@@ -596,11 +596,13 @@ impl StreamMux {
|
||||
if self.send_credit.get(&stream_id).copied().unwrap_or(0) < front_len {
|
||||
break;
|
||||
}
|
||||
let bytes = self
|
||||
let Some(bytes) = self
|
||||
.pending_data
|
||||
.get_mut(&stream_id)
|
||||
.and_then(VecDeque::pop_front)
|
||||
.expect("pending front exists");
|
||||
else {
|
||||
break;
|
||||
};
|
||||
if self
|
||||
.pending_data
|
||||
.get(&stream_id)
|
||||
|
||||
Reference in New Issue
Block a user