Run the state setter on every path reaching a committed manifest

crates/connetto-file-server/src/upload.rs

@@ -7,7 +7,7 @@ use axum::{
 };
 use bytes::Bytes;
 use connetto_file_core::{ChunkHash, ChunkMeta, FileId};
-use diesel_async::AsyncConnection;
+use diesel_async::{AsyncConnection, RunQueryDsl};
 use serde::{Deserialize, Serialize};
 
 use crate::{
@@ -183,7 +183,10 @@ pub(crate) async fn put_chunk<S: ConnettoFileSchema>(
 ///
 /// Four outcomes by manifest state:
 /// - Absent (`file_id`, `caller`) row: 404.
-/// - Already committed (sequential retry after a lost response): idempotent 200.
+/// - Already committed (sequential retry after a lost response, or a second
+///   session staging bytes identical to a file a previous session committed):
+///   idempotent 200, with the state setter re-run so metadata rows inserted
+///   since the first commit are still flipped.
 /// - Uncommitted: verify all chunks are satisfied (stored through this upload OR deduped
 ///   from a committed manifest visible to the caller), then verify identity, then commit.
 ///   The caller must have declared the manifest (ownership is implicit in the composite key).
@@ -220,7 +223,20 @@ pub(crate) async fn post_commit<S: ConnettoFileSchema>(
         .transaction::<StatusCode, ServerError, _>(async move |conn| {
             let manifest = match db::load_manifest_locked::<S>(conn, &file_id, &caller).await? {
                 None => return Err(ServerError::NotFound),
-                Some(db::ManifestState::Committed) => return Ok(StatusCode::OK),
+                Some(db::ManifestState::Committed) => {
+                    // Already committed from a previous call or session: re-run the
+                    // setter so metadata rows inserted since the first commit are
+                    // updated. The setter is UPDATE ... WHERE content_id = $1 and
+                    // is idempotent for rows already at the target state.
+                    diesel::select(crate::functions::connetto_set_content_state(
+                        file_id.as_bytes().to_vec().as_slice(),
+                        "available",
+                        caller.as_str(),
+                    ))
+                    .get_result::<Option<Vec<u8>>>(conn)
+                    .await?;
+                    return Ok(StatusCode::OK);
+                }
                 Some(db::ManifestState::Uncommitted(manifest)) => manifest,
             };
             verify_file_identity(store, &manifest).await?;

crates/connetto-file-server/tests/it/upload.rs

@@ -1164,6 +1164,161 @@ async fn commit_setter_failure_rolls_back_retry_succeeds() {
     assert_eq!(resp.status(), StatusCode::OK, "retry commit must succeed");
 }
 
+/// A commit that finds the manifest already committed, a second session
+/// staging bytes identical to a file a previous session committed, must
+/// still run the deployment state setter, because the metadata rows this
+/// session wrote for the same content can only be flipped by that call.
+#[tokio::test]
+#[expect(
+    clippy::too_many_lines,
+    reason = "proves one session's commit sequence and a second session's re-commit in strict order against a recorded setter; splitting hides the causal chain"
+)]
+async fn recommit_of_committed_manifest_still_runs_the_setter() {
+    async fn upload(
+        app: &axum::Router,
+        signer: &connetto_file_server::TicketSigner,
+        data: &[u8],
+    ) -> FileId {
+        let mem = MemStore::new();
+        let manifest = process_file(data, MimeClass::Generic, &mem).await.unwrap();
+        let file_id = manifest.file_id();
+        let file_hex = format!("{file_id}");
+        let file_size = u64::try_from(data.len()).unwrap();
+        let ticket = write_payload(signer, &file_id, file_size + 1024);
+        let chunks_json: Vec<serde_json::Value> = manifest
+            .chunks()
+            .iter()
+            .map(|c| serde_json::json!({ "hash": format!("{}", c.hash), "len": c.len }))
+            .collect();
+        let body = serde_json::json!({ "total_len": file_size, "chunks": chunks_json });
+        let req = axum::http::Request::builder()
+            .method("POST")
+            .uri(format!("/files/{file_hex}/intent?t={ticket}"))
+            .header("content-type", "application/json")
+            .body(axum::body::Body::from(serde_json::to_vec(&body).unwrap()))
+            .unwrap();
+        assert_eq!(
+            app.clone().oneshot(req).await.unwrap().status(),
+            StatusCode::OK,
+            "intent"
+        );
+        for c in manifest.chunks() {
+            let chunk_data = mem.read_chunk(&c.hash).await.unwrap();
+            let hash_hex = format!("{}", c.hash);
+            let req = axum::http::Request::builder()
+                .method("PUT")
+                .uri(format!("/chunks/{hash_hex}?t={ticket}"))
+                .header("content-type", "application/octet-stream")
+                .body(axum::body::Body::from(chunk_data))
+                .unwrap();
+            assert_eq!(
+                app.clone().oneshot(req).await.unwrap().status(),
+                StatusCode::NO_CONTENT,
+                "chunk PUT"
+            );
+        }
+        let req = axum::http::Request::builder()
+            .method("POST")
+            .uri(format!("/files/{file_hex}/commit?t={ticket}"))
+            .body(axum::body::Body::empty())
+            .unwrap();
+        assert_eq!(
+            app.clone().oneshot(req).await.unwrap().status(),
+            StatusCode::OK,
+            "commit"
+        );
+        file_id
+    }
+    let pg = Pg::start().await;
+    let dir = tempfile::TempDir::new().unwrap();
+    let (app, signer) = build_router(&pg, fs_store(&dir)).await;
+
+    let data = b"identical bytes staged by two devices";
+    let file_id = upload(&app, &signer, data).await;
+    let file_hex = format!("{file_id}");
+
+    // Record every setter call from here on, as a deployment's UPDATE by
+    // content_id would be called by the second session's commit.
+    let mut admin_conn = connect_admin(&pg.url_admin).await;
+    diesel::sql_query(
+        "CREATE TABLE IF NOT EXISTS public.setter_calls (
+             p_file_id BYTEA, p_new_state TEXT, p_caller TEXT
+         )",
+    )
+    .execute(&mut admin_conn)
+    .await
+    .unwrap();
+    diesel::sql_query(
+        "CREATE OR REPLACE FUNCTION connetto_set_content_state(
+             p_file_id BYTEA, p_new_state TEXT, p_caller TEXT
+         ) RETURNS BYTEA LANGUAGE plpgsql SECURITY DEFINER
+             SET search_path TO '' AS $$
+         BEGIN
+             INSERT INTO public.setter_calls VALUES (p_file_id, p_new_state, p_caller);
+             RETURN p_file_id;
+         END; $$",
+    )
+    .execute(&mut admin_conn)
+    .await
+    .unwrap();
+
+    // The second session: a fresh ticket, intent answered by dedup, commit.
+    let manifest = process_file(data, MimeClass::Generic, &MemStore::new())
+        .await
+        .unwrap();
+    let chunks_json: Vec<serde_json::Value> = manifest
+        .chunks()
+        .iter()
+        .map(|c| serde_json::json!({ "hash": format!("{}", c.hash), "len": c.len }))
+        .collect();
+    let body = serde_json::json!({ "total_len": u64::try_from(data.len()).unwrap(), "chunks": chunks_json });
+    let ticket2 = write_payload(&signer, &file_id, 8192);
+    let req = axum::http::Request::builder()
+        .method("POST")
+        .uri(format!("/files/{file_hex}/intent?t={ticket2}"))
+        .header("content-type", "application/json")
+        .body(axum::body::Body::from(serde_json::to_vec(&body).unwrap()))
+        .unwrap();
+    assert_eq!(
+        app.clone().oneshot(req).await.unwrap().status(),
+        StatusCode::OK,
+        "second intent on the committed manifest"
+    );
+    let req = axum::http::Request::builder()
+        .method("POST")
+        .uri(format!("/files/{file_hex}/commit?t={ticket2}"))
+        .body(axum::body::Body::empty())
+        .unwrap();
+    assert_eq!(
+        app.clone().oneshot(req).await.unwrap().status(),
+        StatusCode::OK,
+        "second commit on the committed manifest"
+    );
+
+    let calls: i64 = diesel_async::RunQueryDsl::get_result(
+        diesel::sql_query(
+            "SELECT COUNT(*) AS calls FROM public.setter_calls \
+             WHERE p_file_id = $1 AND p_new_state = 'available' AND p_caller = 'alice'",
+        )
+        .bind::<diesel::sql_types::Bytea, _>(file_id.as_bytes().as_ref()),
+        &mut admin_conn,
+    )
+    .await
+    .map(|r: SetterCallCount| r.calls)
+    .unwrap();
+    assert_eq!(
+        calls, 1,
+        "the re-commit must run the setter exactly once so the second session's \
+         metadata row for the same content is flipped"
+    );
+}
+
+#[derive(diesel::QueryableByName)]
+struct SetterCallCount {
+    #[diesel(sql_type = diesel::sql_types::BigInt)]
+    calls: i64,
+}
+
 #[derive(diesel::QueryableByName)]
 struct CommittedRow {
     #[diesel(sql_type = diesel::sql_types::Bool)]