Carry staged files to the hub over the tab-to-worker content lane

.github/workflows/ci.yml

@@ -404,6 +404,7 @@ jobs:
               -p connetto-yew
               -p connetto-file-core
               -p connetto-file-server
+              -p connetto-file-client
             bins: ''
     steps:
       - uses: actions/checkout@v7
@@ -509,7 +510,6 @@ jobs:
           echo "${CHROMEWEBDRIVER}" >> "${GITHUB_PATH}"
           echo "CHROMEDRIVER=${CHROMEWEBDRIVER}/chromedriver" >> "${GITHUB_ENV}"
 
-
       # The identity-provider image is pulled inside each shard's startup
       # wait, which a congested ghcr.io can overrun. Pulling it once here
       # takes the registry out of the container startup budget.

crates/connetto-file-client/src/error.rs

@@ -112,6 +112,24 @@ pub enum ContentError {
     },
 }
 
+/// Why a staged commit did not land, keeping the row's refusal distinguishable
+/// from a bookkeeping failure so the caller can word the answer.
+#[derive(Debug, Error)]
+pub enum StageCommitError {
+    /// Recording the manifest or queueing the upload failed.
+    #[error("staged content bookkeeping: {0}")]
+    Bookkeeping(#[from] ContentError),
+    /// The row closure refused, carrying the detail it gave.
+    #[error("staged row refused: {0}")]
+    Row(String),
+}
+
+impl From<diesel::result::Error> for StageCommitError {
+    fn from(err: diesel::result::Error) -> Self {
+        Self::Bookkeeping(err.into())
+    }
+}
+
 /// What the outbox walk does with a failed upload attempt, decided by where the fact came from.
 #[derive(Debug, Clone, Copy, PartialEq, Eq)]
 pub enum AttemptOutcome {

crates/connetto-file-client/src/lib.rs

@@ -19,15 +19,15 @@ mod worker;
 pub use browser_store::{BrowserStore, BrowserStoreError};
 
 pub use client::{ContentClient, ContentEvent};
-pub use connetto_file_core::{ChunkHash, FileId};
-pub use error::{AttemptOutcome, ContentError};
+pub use connetto_file_core::{ChunkHash, FileId, FileIdHasher, MimeClass};
+pub use error::{AttemptOutcome, ContentError, StageCommitError};
 pub use import::ContentImportPlan;
 pub use resolve::{BoxedSource, ChunkStoreSource, LocalContentSource, Resolved, SourceFuture};
 #[cfg(not(all(target_family = "wasm", target_os = "unknown")))]
 pub use store::{FsStore, FsStoreError};
 pub use worker::{
     ChunkScan, ContentArchive, ContentFlush, ContentFlushStart, ContentFlushState, ContentUpload,
-    FlushCursor, ScanStep,
+    FlushCursor, PendingConnectionResolve, ResolveRoute, ResolveStart, ScanStep,
 };
 
 #[cfg(all(target_family = "wasm", target_os = "unknown"))]

crates/connetto-file-client/src/ticket.rs

@@ -88,7 +88,7 @@ where
     }
 }
 
-async fn ensure_pending_request<T: Transport>(
+pub(crate) async fn ensure_pending_request<T: Transport>(
     connection: &mut ConnettoConnection<T>,
     file_id: FileId,
     verb: ContentVerb,
@@ -170,6 +170,26 @@ fn resolve_connection_event(
     Some(result)
 }
 
+/// What one pumped event has to say about one in-flight ticket request,
+/// without consuming the event or the waiting slot.
+pub(crate) enum RouteAnswer {
+    /// The request's answer arrived, or the link carrying it is gone.
+    Settled(Result<String, ContentError>),
+    /// The event concerns something else.
+    Other,
+}
+
+pub(crate) fn route_connection_event(pending: &PendingTicket, event: &ClientEvent) -> RouteAnswer {
+    match classify(event.clone(), &pending.request_id, pending.file_id) {
+        TicketAnswer::Granted(url) => RouteAnswer::Settled(Ok(url)),
+        TicketAnswer::Failed(error) => RouteAnswer::Settled(Err(error)),
+        TicketAnswer::Unrelated(ClientEvent::Closed | ClientEvent::ServerClosed { .. }) => {
+            RouteAnswer::Settled(Err(ContentError::TicketAbandoned))
+        }
+        TicketAnswer::Unrelated(_) => RouteAnswer::Other,
+    }
+}
+
 fn next_request_id() -> String {
     format!("content-{}", NEXT_REQUEST.fetch_add(1, Ordering::Relaxed))
 }
@@ -185,3 +205,56 @@ fn refusal(detail: &str, file_id: FileId) -> ContentError {
     );
     ContentError::TicketRefused { file_id }
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    fn waiting() -> PendingTicket {
+        PendingTicket {
+            file_id: FileId::from_bytes([1; 32]),
+            request_id: "content-7".to_owned(),
+        }
+    }
+
+    #[test]
+    fn a_routed_event_settles_only_its_own_request() {
+        let pending = waiting();
+        let grant = ClientEvent::ContentTicket {
+            request_id: "content-7".to_owned(),
+            url: "https://content.invalid/f".to_owned(),
+        };
+        assert!(matches!(
+            route_connection_event(&pending, &grant),
+            RouteAnswer::Settled(Ok(url)) if url == "https://content.invalid/f"
+        ));
+        let someone_elses_grant = ClientEvent::ContentTicket {
+            request_id: "content-8".to_owned(),
+            url: "https://content.invalid/f".to_owned(),
+        };
+        assert!(matches!(
+            route_connection_event(&pending, &someone_elses_grant),
+            RouteAnswer::Other
+        ));
+        let refusal = ClientEvent::NonFatal {
+            related_to: Some("content-7".to_owned()),
+            detail: CONTENT_TICKET_REFUSED.to_owned(),
+        };
+        assert!(matches!(
+            route_connection_event(&pending, &refusal),
+            RouteAnswer::Settled(Err(_))
+        ));
+        let unrelated = ClientEvent::NonFatal {
+            related_to: Some("subscription-1".to_owned()),
+            detail: "something else failed".to_owned(),
+        };
+        assert!(matches!(
+            route_connection_event(&pending, &unrelated),
+            RouteAnswer::Other
+        ));
+        assert!(matches!(
+            route_connection_event(&pending, &ClientEvent::Closed),
+            RouteAnswer::Settled(Err(ContentError::TicketAbandoned))
+        ));
+    }
+}