From 2a5ad5e1cc9b60a014b72ce10d440a09a56aa62a Mon Sep 17 00:00:00 2001 From: LucaCappelletti94 Date: Fri, 18 Sep 2026 19:00:03 +0200 Subject: [PATCH] Run the state setter on every path reaching a committed manifest --- crates/connetto-file-server/src/upload.rs | 22 ++- .../connetto-file-server/tests/it/upload.rs | 155 ++++++++++++++++++ 2 files changed, 174 insertions(+), 3 deletions(-) diff --git a/crates/connetto-file-server/src/upload.rs b/crates/connetto-file-server/src/upload.rs index 1e340e90..ec40ea23 100644 --- a/crates/connetto-file-server/src/upload.rs +++ b/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( /// /// 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( .transaction::(async move |conn| { let manifest = match db::load_manifest_locked::(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::>>(conn) + .await?; + return Ok(StatusCode::OK); + } Some(db::ManifestState::Uncommitted(manifest)) => manifest, }; verify_file_identity(store, &manifest).await?; diff --git a/crates/connetto-file-server/tests/it/upload.rs b/crates/connetto-file-server/tests/it/upload.rs index 78fecc7c..6399196b 100644 --- a/crates/connetto-file-server/tests/it/upload.rs +++ b/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 = 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 = 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::(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)]