perf(stream,core): take the disk cache off the read path, warm file edges, report crawl progress

A block the player waits for no longer waits on its disk-cache write and
two fsyncs; the put runs in a spawned task and only the block itself is
synced before the rename. Opening a stream can fetch the first and last
block together, where MKV cues and the MP4 moov live. Downloads open a
copying stream that reads the disk cache but never fills it.

crawl_reporting reports the running file count after each listing.

crates/pstr-core/src/shares.rs

@@ -358,6 +358,19 @@ impl SharedLibrary {
     /// and a catalog that reorders itself between two identical crawls is
     /// needlessly hard to diff.
     pub async fn crawl(&self, share_id: &str) -> Result<Vec<Node>> {
+        self.crawl_reporting(share_id, |_| {}).await
+    }
+
+    /// [`Self::crawl`], calling `progress` with how many nodes have been found
+    /// so far each time a folder's listing comes back.
+    ///
+    /// A crawl of a large share takes minutes, and "crawling…" for minutes
+    /// reads as stuck. A count that keeps rising does not.
+    pub async fn crawl_reporting(
+        &self,
+        share_id: &str,
+        progress: impl Fn(usize),
+    ) -> Result<Vec<Node>> {
         use futures::stream::{StreamExt as _, TryStreamExt as _};
 
         let client = self
@@ -392,6 +405,7 @@ impl SharedLibrary {
                     }
                     found.push(child);
                 }
+                progress(found.len());
             }
             level = next;
         }
@@ -428,6 +442,15 @@ fn client_configuration() -> ProtonClientConfiguration {
 /// Parsed here as well as in the SDK because the store needs a stable id for a
 /// share *before* it has opened it — and the id must not be derived from the
 /// password, which would put a secret in the config file by the back door.
+/// The share token a link names, or why it is not a usable share link.
+///
+/// Public so a form can say what is wrong with a link while it is still being
+/// pasted, rather than after a round trip through [`ShareStore::add`] — which
+/// asks exactly this first.
+pub fn share_token(url: &str) -> Result<String> {
+    token_from_url(url)
+}
+
 fn token_from_url(url: &str) -> Result<String> {
     let (_, tail) = url
         .split_once("/urls/")

crates/pstr-stream/src/disk.rs

@@ -321,11 +321,9 @@ async fn write_entry(path: &Path, block: &[u8]) -> Result<()> {
     }
     tokio::fs::rename(&temp, path).await?;
 
-    let mut sidecar = tokio::fs::File::create(meta_path(path)).await?;
-    sidecar
-        .write_all(block.len().to_string().as_bytes())
-        .await?;
-    sidecar.sync_all().await?;
+    // Not synced. The block's sync is the one the ordering depends on; a
+    // sidecar lost to a crash only turns a valid entry into a miss.
+    tokio::fs::write(meta_path(path), block.len().to_string()).await?;
     Ok(())
 }
 

crates/pstr-stream/src/source.rs

@@ -155,6 +155,24 @@ impl StreamSource {
             .await
     }
 
+    /// Open a file to copy it out whole, rather than to play it.
+    ///
+    /// Not the playback stream, and not shared with it: its blocks are read
+    /// once and written to a file, so caching them in memory only evicts the
+    /// blocks a viewer is about to need, and caching them on disk stores every
+    /// byte twice while pushing out what was just watched. It does still read
+    /// the disk cache — an episode watched and then downloaded is not fetched
+    /// again.
+    pub async fn open_for_copy(&self, share_id: &str, uid: &NodeUid) -> Result<VideoStream> {
+        let blocks = self.inner.opener.open(share_id, uid).await?;
+        Ok(VideoStream::copying(
+            uid.clone(),
+            blocks,
+            Arc::new(BlockRing::new(0)),
+            self.inner.disk.clone(),
+        ))
+    }
+
     /// Drop a stream and everything cached for it in memory.
     ///
     /// The disk cache is deliberately untouched: it exists precisely so the next

crates/pstr-stream/src/stream.rs

@@ -103,6 +103,11 @@ struct Inner {
     map: BlockMap,
     ring: Arc<BlockRing>,
     disk: Option<Arc<DiskCache>>,
+    /// Whether blocks fetched from the network are written to `disk`. Off for
+    /// a stream being copied out to a file, which reads what the cache already
+    /// holds but should not push what a viewer just watched out of it to make
+    /// room for bytes that are about to exist on disk anyway.
+    fill_disk: bool,
     /// One fetch per block, however many readers want it.
     flight: Arc<SingleFlight<BlockKey, CachedBlock>>,
     readahead: usize,
@@ -131,6 +136,28 @@ impl VideoStream {
         ring: Arc<BlockRing>,
         disk: Option<Arc<DiskCache>>,
         readahead: usize,
+    ) -> Self {
+        Self::build(uid, blocks, ring, disk, readahead, true)
+    }
+
+    /// A stream that reads through the disk cache but never adds to it. See
+    /// [`crate::StreamSource::open_for_copy`].
+    pub(crate) fn copying(
+        uid: NodeUid,
+        blocks: SharedBlocks,
+        ring: Arc<BlockRing>,
+        disk: Option<Arc<DiskCache>>,
+    ) -> Self {
+        Self::build(uid, blocks, ring, disk, 0, false)
+    }
+
+    fn build(
+        uid: NodeUid,
+        blocks: SharedBlocks,
+        ring: Arc<BlockRing>,
+        disk: Option<Arc<DiskCache>>,
+        readahead: usize,
+        fill_disk: bool,
     ) -> Self {
         let map = BlockMap::new(blocks.block_sizes());
         let readahead = clamp_readahead(readahead, ring.budget(), blocks.block_sizes());
@@ -141,6 +168,7 @@ impl VideoStream {
                 map,
                 ring,
                 disk,
+                fill_disk,
                 flight: Arc::new(SingleFlight::new()),
                 readahead,
                 readahead_slots: Arc::new(Semaphore::new(readahead.max(1))),
@@ -176,6 +204,38 @@ impl VideoStream {
         self.inner.blocks.block_sizes()
     }
 
+    /// Start fetching the first and last blocks, without waiting for them.
+    ///
+    /// Every demuxer reads the head of a file first, and the containers this
+    /// app plays keep their index at the tail — Matroska's cues, an MP4's
+    /// `moov` when it was not written for streaming. Opening a file is
+    /// therefore two round trips, one after the other. Asked for together,
+    /// they are one. The player's own reads join these through the
+    /// single-flight, so nothing is fetched twice.
+    ///
+    /// Detached rather than read-ahead: the jump to the tail looks like a
+    /// seek, and a seek cancels read-ahead — which here would cancel the very
+    /// fetch it was meant to have started.
+    pub fn warm_edges(&self) {
+        let Ok(runtime) = tokio::runtime::Handle::try_current() else {
+            return;
+        };
+        let count = self.inner.map.len();
+        let edges: &[usize] = match count {
+            0 => &[],
+            1 => &[0],
+            _ => &[0, count - 1],
+        };
+        for &index in edges {
+            let inner = Arc::clone(&self.inner);
+            runtime.spawn(async move {
+                if let Err(error) = block_at(inner, index).await {
+                    tracing::debug!(index, %error, "warming an edge block failed");
+                }
+            });
+        }
+    }
+
     /// Fill `buf` from `offset`, returning how many bytes were written.
     ///
     /// A read at or past EOF returns 0 rather than failing — that is how a
@@ -413,10 +473,17 @@ async fn block_at(inner: Arc<Inner>, index: usize) -> Result<CachedBlock> {
                 .fetched_bytes
                 .fetch_add(bytes.len() as u64, Ordering::Relaxed);
 
-            if let Some(disk) = inner.disk.as_ref() {
-                disk.put(&fetch_key, &bytes).await;
-            }
             let block: CachedBlock = bytes.into();
+            // Written behind the read, not before it: the cache is disposable,
+            // and a block the player is waiting for should not also wait for a
+            // 4 MiB write and its fsync. That wait was on every cold block,
+            // which is to say on the first frame and on every seek.
+            if let Some(disk) = inner.disk.as_ref().filter(|_| inner.fill_disk) {
+                let disk = Arc::clone(disk);
+                let key = fetch_key.clone();
+                let bytes = Arc::clone(&block);
+                tokio::spawn(async move { disk.put(&key, &bytes).await });
+            }
             inner.ring.insert(fetch_key, Arc::clone(&block));
             Ok(block)
         })
@@ -891,4 +958,85 @@ mod tests {
         assert_eq!(stream.read_at(80, &mut buf).await.unwrap(), 20);
         assert_eq!(&buf[..20], &expected(80, 20)[..]);
     }
+
+    async fn disk_in(tag: &str) -> Arc<DiskCache> {
+        let root = std::env::temp_dir().join(format!("pstr-stream-{tag}-{}", std::process::id()));
+        let _ = std::fs::remove_dir_all(&root);
+        Arc::new(
+            DiskCache::open(crate::disk::DiskCacheConfig::new(root))
+                .await
+                .unwrap(),
+        )
+    }
+
+    /// Waits on a detached task without sleeping for a guessed duration.
+    async fn eventually(mut done: impl FnMut() -> bool) -> bool {
+        for _ in 0..200 {
+            if done() {
+                return true;
+            }
+            tokio::time::sleep(std::time::Duration::from_millis(5)).await;
+        }
+        done()
+    }
+
+    #[tokio::test]
+    async fn warming_fetches_the_first_and_last_blocks_and_nothing_between() {
+        let fake = Fake::new(&[1000; 10]);
+        let stream = stream_over(Arc::clone(&fake), 0);
+
+        stream.warm_edges();
+        assert!(eventually(|| fake.fetches_of(0) == 1 && fake.fetches_of(9) == 1).await);
+        assert_eq!(fake.fetches(), 2);
+
+        // And the player's own reads are then served without a second fetch.
+        stream.read_range(0, 10).await.unwrap();
+        stream.read_range(9_990, 10).await.unwrap();
+        assert_eq!(fake.fetches(), 2);
+    }
+
+    #[tokio::test]
+    async fn a_fetched_block_reaches_the_disk_cache_without_the_read_waiting_for_it() {
+        let disk = disk_in("fill").await;
+        let fake = Fake::new(&[1000, 1000]);
+        let stream = VideoStream::new(
+            uid(),
+            fake,
+            Arc::new(BlockRing::new(DEFAULT_RING_BYTES)),
+            Some(Arc::clone(&disk)),
+            0,
+        );
+
+        assert_eq!(stream.read_range(0, 10).await.unwrap(), expected(0, 10));
+        assert!(eventually(|| disk.stats().entries == 1).await);
+    }
+
+    #[tokio::test]
+    async fn a_copying_stream_reads_the_disk_cache_but_never_adds_to_it() {
+        let disk = disk_in("copy").await;
+        let fake = Fake::new(&[1000, 1000]);
+        let key = BlockKey::new(&uid(), "rev1", 0);
+        disk.put(&key, &expected(0, 1000)).await;
+
+        let stream = VideoStream::copying(
+            uid(),
+            Arc::clone(&fake) as SharedBlocks,
+            Arc::new(BlockRing::new(0)),
+            Some(Arc::clone(&disk)),
+        );
+        assert_eq!(stream.read_range(0, 2000).await.unwrap(), expected(0, 2000));
+
+        assert_eq!(
+            fake.fetches_of(0),
+            0,
+            "block 0 should have come from the disk"
+        );
+        assert_eq!(fake.fetches_of(1), 1);
+        tokio::time::sleep(std::time::Duration::from_millis(50)).await;
+        assert_eq!(
+            disk.stats().entries,
+            1,
+            "the copy wrote into the playback cache"
+        );
+    }
 }