feat: 检查响应头 etag 和 last_modified 是否与之前一致

share121 · share121 · commit 5302e3325135 · 2025-08-28T15:06:04.000+08:00
diff --git a/crates/fast-down/src/http/mod.rs b/crates/fast-down/src/http/mod.rs
@@ -3,6 +3,7 @@ mod puller;
 pub use prefetch::*;
 pub use puller::*;
 
+use crate::url_info::FileId;
 use bytes::Bytes;
 use fast_pull::ProgressEntry;
 use std::{fmt::Debug, future::Future};
@@ -43,4 +44,5 @@ pub enum HttpError<Client: HttpClient> {
     Chunk(GetChunkError<Client>),
     GetHeader(GetHeaderError<Client>),
     Irrecoverable,
+    MismatchedBody(FileId),
 }
diff --git a/crates/fast-down/src/http/prefetch.rs b/crates/fast-down/src/http/prefetch.rs
@@ -1,6 +1,7 @@
 use crate::{
     UrlInfo,
     http::{GetResponse, HttpClient, HttpError, HttpHeaders, HttpRequestBuilder, HttpResponse},
+    url_info::FileId,
 };
 use content_disposition;
 use std::future::Future;
@@ -65,8 +66,7 @@ async fn prefetch<Client: HttpClient>(
             size,
             supports_range,
             fast_download: size > 0 && supports_range,
-            etag: headers.get("etag").ok().map(|s| s.to_string()),
-            last_modified: headers.get("last-modified").ok().map(|s| s.to_string()),
+            file_id: FileId::new(headers.get("etag").ok(), headers.get("last-modified").ok()),
         },
         resp,
     ))
diff --git a/crates/fast-down/src/http/puller.rs b/crates/fast-down/src/http/puller.rs
@@ -1,5 +1,6 @@
 use crate::http::{
-    GetRequestError, GetResponse, HttpClient, HttpError, HttpRequestBuilder, HttpResponse,
+    FileId, GetRequestError, GetResponse, HttpClient, HttpError, HttpHeaders, HttpRequestBuilder,
+    HttpResponse,
 };
 use bytes::Bytes;
 use fast_pull::{ProgressEntry, RandPuller, SeqPuller};
@@ -17,13 +18,20 @@ pub struct HttpPuller<Client: HttpClient> {
     pub(crate) client: Client,
     url: Url,
     resp: Arc<SpinMutex<Option<GetResponse<Client>>>>,
+    file_id: FileId,
 }
 impl<Client: HttpClient> HttpPuller<Client> {
-    pub fn new(url: Url, client: Client, resp: Option<GetResponse<Client>>) -> Self {
+    pub fn new(
+        url: Url,
+        client: Client,
+        resp: Option<GetResponse<Client>>,
+        file_id: FileId,
+    ) -> Self {
         Self {
             client,
             url,
             resp: Arc::new(SpinMutex::new(resp)),
+            file_id,
         }
     }
 }
@@ -54,6 +62,7 @@ impl<Client: HttpClient + 'static> RandPuller for HttpPuller<Client> {
             } else {
                 ResponseState::None
             },
+            file_id: self.file_id.clone(),
         }
     }
 }
@@ -63,6 +72,7 @@ struct RandRequestStream<Client: HttpClient + 'static> {
     start: u64,
     end: u64,
     state: ResponseState<Client>,
+    file_id: FileId,
 }
 impl<Client: HttpClient> Stream for RandRequestStream<Client> {
     type Item = Result<Bytes, HttpError<Client>>;
@@ -73,8 +83,16 @@ impl<Client: HttpClient> Stream for RandRequestStream<Client> {
                 return match resp.try_poll_unpin(cx) {
                     Poll::Ready(resp) => match resp {
                         Ok(resp) => {
-                            self.state = ResponseState::Ready(resp);
-                            self.poll_next(cx)
+                            let etag = resp.headers().get("etag").ok();
+                            let last_modified = resp.headers().get("last-modified").ok();
+                            let new_file_id = FileId::new(etag, last_modified);
+                            if new_file_id != self.file_id {
+                                self.state = ResponseState::None;
+                                Poll::Ready(Some(Err(HttpError::MismatchedBody(new_file_id))))
+                            } else {
+                                self.state = ResponseState::Ready(resp);
+                                self.poll_next(cx)
+                            }
                         }
                         Err(e) => {
                             self.state = ResponseState::None;
@@ -118,21 +136,21 @@ impl<Client: HttpClient> Stream for RandRequestStream<Client> {
 impl<Client: HttpClient + 'static> SeqPuller for HttpPuller<Client> {
     type Error = HttpError<Client>;
     fn pull(&mut self) -> impl TryStream<Ok = Bytes, Error = Self::Error> + Send + Unpin {
-        match self.resp.lock().take() {
-            Some(resp) => SeqRequestStream {
-                state: ResponseState::Ready(resp),
-            },
-            None => {
-                let req = self.client.get(self.url.clone(), None).send();
-                SeqRequestStream {
-                    state: ResponseState::Pending(Box::pin(req)),
+        SeqRequestStream {
+            state: match self.resp.lock().take() {
+                Some(resp) => ResponseState::Ready(resp),
+                None => {
+                    let req = self.client.get(self.url.clone(), None).send();
+                    ResponseState::Pending(Box::pin(req))
                 }
-            }
+            },
+            file_id: self.file_id.clone(),
         }
     }
 }
 struct SeqRequestStream<Client: HttpClient + 'static> {
     state: ResponseState<Client>,
+    file_id: FileId,
 }
 impl<Client: HttpClient> Stream for SeqRequestStream<Client> {
     type Item = Result<Bytes, HttpError<Client>>;
@@ -143,8 +161,16 @@ impl<Client: HttpClient> Stream for SeqRequestStream<Client> {
                 return match resp.try_poll_unpin(cx) {
                     Poll::Ready(resp) => match resp {
                         Ok(resp) => {
-                            self.state = ResponseState::Ready(resp);
-                            self.poll_next(cx)
+                            let etag = resp.headers().get("etag").ok();
+                            let last_modified = resp.headers().get("last-modified").ok();
+                            let new_file_id = FileId::new(etag, last_modified);
+                            if new_file_id != self.file_id {
+                                self.state = ResponseState::None;
+                                Poll::Ready(Some(Err(HttpError::MismatchedBody(new_file_id))))
+                            } else {
+                                self.state = ResponseState::Ready(resp);
+                                self.poll_next(cx)
+                            }
                         }
                         Err(e) => {
                             self.state = ResponseState::None;
diff --git a/crates/fast-down/src/reqwest/mod.rs b/crates/fast-down/src/reqwest/mod.rs
@@ -90,7 +90,10 @@ pub enum ReqwestGetHeaderError {
 #[cfg(test)]
 mod tests {
     use super::*;
-    use crate::http::{HttpError, HttpPuller, Prefetch};
+    use crate::{
+        http::{HttpError, HttpPuller, Prefetch},
+        url_info::FileId,
+    };
     use fast_pull::{
         Event, MergeProgress,
         mem::MemPusher,
@@ -185,6 +188,9 @@ mod tests {
                 HttpError::Chunk(_) => unreachable!(),
                 HttpError::GetHeader(_) => unreachable!(),
                 HttpError::Irrecoverable => unreachable!(),
+                HttpError::MismatchedBody(file_id) => {
+                    unreachable!("404 status code should not return mismatched body: {file_id:?}")
+                }
             },
         }
         mock1.assert_async().await;
@@ -227,6 +233,7 @@ mod tests {
             format!("{}/concurrent", server.url()).parse().unwrap(),
             Client::new(),
             None,
+            FileId::empty(),
         );
         let pusher = MemPusher::with_capacity(mock_data.len());
         #[allow(clippy::single_range_in_vec_init)]
@@ -280,6 +287,7 @@ mod tests {
             format!("{}/sequential", server.url()).parse().unwrap(),
             Client::new(),
             None,
+            FileId::empty(),
         );
         let pusher = MemPusher::with_capacity(mock_data.len());
         #[allow(clippy::single_range_in_vec_init)]
diff --git a/crates/fast-down/src/url_info.rs b/crates/fast-down/src/url_info.rs
@@ -1,3 +1,4 @@
+use std::sync::Arc;
 use url::Url;
 
 #[derive(Debug)]
@@ -7,6 +8,26 @@ pub struct UrlInfo {
     pub supports_range: bool,
     pub fast_download: bool,
     pub final_url: Url,
-    pub etag: Option<String>,
-    pub last_modified: Option<String>,
+    pub file_id: FileId,
+}
+
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub struct FileId {
+    pub etag: Option<Arc<str>>,
+    pub last_modified: Option<Arc<str>>,
+}
+
+impl FileId {
+    pub fn new(etag: Option<&str>, last_modified: Option<&str>) -> Self {
+        Self {
+            etag: etag.map(Arc::from),
+            last_modified: last_modified.map(Arc::from),
+        }
+    }
+    pub fn empty() -> Self {
+        Self {
+            etag: None,
+            last_modified: None,
+        }
+    }
 }
