From 07991f172ea4d0922c73c1cd2e2cf9c22bc34077 Mon Sep 17 00:00:00 2001 From: euzu Date: Wed, 10 Dec 2025 17:30:34 +0100 Subject: [PATCH] Made cache storage more robust --- .../api/model/streams/persist_pipe_stream.rs | 145 ------------------ backend/src/tools/lru_cache.rs | 2 +- 2 files changed, 1 insertion(+), 146 deletions(-) diff --git a/backend/src/api/model/streams/persist_pipe_stream.rs b/backend/src/api/model/streams/persist_pipe_stream.rs index 8baac31fe..10f48880e 100644 --- a/backend/src/api/model/streams/persist_pipe_stream.rs +++ b/backend/src/api/model/streams/persist_pipe_stream.rs @@ -66,148 +66,3 @@ where S: tokio_stream::Stream> + Send + Unpin ReceiverStream::new(rx) } - -// -// -// /// `PersistPipeStream` -// /// -// /// Pipes bytes from an upstream stream to an async writer while tracking total size. -// /// Once the stream completes and the writer is flushed, the provided callback is invoked -// /// with the total number of bytes written. -// pub struct PersistPipeStream { -// inner: S, -// completed: bool, -// writer: W, -// size: AtomicUsize, -// callback: Arc, -// pending_writes: VecDeque, -// current_offset: usize, -// } -// -// impl PersistPipeStream -// where -// S: Stream + Unpin, -// W: AsyncWrite + Unpin + 'static, -// { -// pub fn new(inner: S, writer: W, callback: Arc) -> Self { -// Self { -// inner, -// completed: false, -// writer, -// size: AtomicUsize::new(0), -// callback, -// pending_writes: VecDeque::new(), -// current_offset: 0, -// } -// } -// -// fn enqueue_chunk(&mut self, bytes: Bytes) { -// self.pending_writes.push_back(bytes); -// } -// -// fn poll_pending_writes(&mut self, cx: &mut Context<'_>) -> Poll<()> { -// while let Some(chunk) = self.pending_writes.front() { -// let chunk_len = chunk.len(); -// if self.current_offset >= chunk_len { -// if let Some(finished) = self.pending_writes.pop_front() { -// self.size.fetch_add(finished.len(), Ordering::AcqRel); -// } -// self.current_offset = 0; -// continue; -// } -// -// let remaining = &chunk[self.current_offset..]; -// match Pin::new(&mut self.writer).poll_write(cx, remaining) { -// Poll::Pending => return Poll::Pending, -// Poll::Ready(Ok(written)) => { -// if written == 0 { -// cx.waker().wake_by_ref(); -// // Avoid tight loop if writer makes no progress. -// return Poll::Pending; -// } -// self.current_offset += written; -// } -// Poll::Ready(Err(err)) => { -// warn!( -// "Dropping {} buffered bytes after persistence write error: {err}", -// chunk_len.saturating_sub(self.current_offset) -// ); -// error!("Error writing to resource file: {err}"); -// self.pending_writes.pop_front(); -// self.current_offset = 0; -// } -// } -// } -// -// self.current_offset = 0; -// Poll::Ready(()) -// } -// -// fn poll_flush(&mut self, cx: &mut Context<'_>) -> Poll<()> { -// match Pin::new(&mut self.writer).poll_flush(cx) { -// Poll::Pending => Poll::Pending, -// Poll::Ready(Ok(())) => Poll::Ready(()), -// Poll::Ready(Err(err)) => { -// error!("Error flushing resource file: {err}"); -// Poll::Ready(()) -// } -// } -// } -// -// fn finalize(&mut self) { -// if !self.completed { -// self.completed = true; -// let size = self.size.load(Ordering::Acquire); -// debug!("Persisted {size} bytes to cache resource"); -// (self.callback)(size); -// } -// } -// } -// -// impl Stream for PersistPipeStream -// where -// S: Stream> + Unpin, -// W: AsyncWrite + Unpin + 'static, -// { -// type Item = Result; -// -// fn poll_next(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll> { -// let this = self.get_mut(); -// -// if !this.pending_writes.is_empty() && this.poll_pending_writes(cx).is_pending() { -// return Poll::Pending; -// } -// -// match Pin::new(&mut this.inner).poll_next(cx) { -// Poll::Pending => Poll::Pending, -// Poll::Ready(None) => { -// if !this.pending_writes.is_empty() { -// if this.poll_pending_writes(cx).is_pending() { -// return Poll::Pending; -// } -// } -// -// if !this.pending_writes.is_empty() { -// return Poll::Pending; -// } -// -// if this.poll_flush(cx).is_pending() { -// return Poll::Pending; -// } -// -// this.finalize(); -// Poll::Ready(None) -// } -// Poll::Ready(Some(item)) => { -// if let Ok(bytes) = &item { -// this.enqueue_chunk(bytes.clone()); -// } -// // Try to drain pending bytes after queuing new data. -// if this.poll_pending_writes(cx).is_pending() { -// // fall through: we still return the chunk to the caller even if persistence is pending -// } -// Poll::Ready(Some(item)) -// } -// } -// } -// } diff --git a/backend/src/tools/lru_cache.rs b/backend/src/tools/lru_cache.rs index 43c43bd84..d666ab1c4 100644 --- a/backend/src/tools/lru_cache.rs +++ b/backend/src/tools/lru_cache.rs @@ -127,7 +127,7 @@ impl LRUResourceCache { return Some(path.clone()); } { - trace_if_enabled!("Responding resource from cache with key: {key} for url: {}", sanitize_sensitive_info(url)); + trace_if_enabled!("Cache inconsistency: file missing for key: {key}, url: {}", sanitize_sensitive_info(url)); // this should not happen, someone deleted the file manually and the cache is not in sync self.current_size -= size; self.cache.remove(&key);