Skip to main content

max / synckit

Observe the two resume shapes nothing was looking at Measured with cargo-mutants on astra against src/client/blob.rs. Two gaps, both in the resume path, both closed here. A previous attempt that got every part to S3 and died on the assemble call takes a branch no test reached: the recorded parts cover the whole ciphertext, so the streaming loop sends nothing and the client owes the server only the part list it already holds. It is reachable because a failed complete is the one failure that keeps the session, stream_blob_parts having returned Ok. The age limit on a resume record had never been observed at all. A store double makes age settable without a clock seam in the client, and a record that names every part gives the widest oracle the question allows: keeping it costs zero PUTs, rejecting it costs the whole blob. That pins the comparison at twelve hours, and the constant behind it.
Co-Authored-By
Claude Opus 5 (1M context) <noreply@anthropic.com>
Claude-Session
https://claude.ai/code/session_01MptwXZ8k65v19rFmdGAyki
Author: Max Johnson <me@maxj.phd> · 2026-08-31 22:49 UTC
Signed with PGP, not checked
Commit: 90bdad1051fbd4dd7046ee28f3c0c323b9e847ce
Parent: 0bfe094
1 file changed, +267 insertions, -0 deletions
M synckit-client/tests/integration/blob_multipart.rs +267
@@ -665,6 +665,273 @@
665 665 std::fs::remove_file(&file).ok();
666 666 }
667 667
668 + /// A previous attempt that got every part to S3 and died on the assemble call
669 + /// must assemble on the next attempt without re-sending a byte.
670 + ///
671 + /// This is the one resume shape where the streaming loop sends nothing at all:
672 + /// the recorded parts cover the whole ciphertext, so the boundary lands past
673 + /// the last chunk and every chunk is read for the content-address check and
674 + /// then skipped. What the client owes the server is the part list it already
675 + /// has, and only `complete` is left to do.
676 + ///
677 + /// It is reachable because a failed `complete` is the one failure that keeps
678 + /// the session and the record: `stream_blob_parts` returned `Ok`, so the abort
679 + /// and clear that guard a failed transfer are never run.
680 + #[tokio::test]
681 + async fn a_resume_that_already_holds_every_part_assembles_without_sending_one() {
682 + let kit = MockKit::start().await;
683 + let key = synckit_client::crypto::generate_master_key();
684 + let store = resume_store("complete-died");
685 +
686 + let plaintext: Vec<u8> = (0..(synckit_client::crypto::BLOB_CHUNK_SIZE * 2 + 11))
687 + .map(|i| i as u8)
688 + .collect();
689 + let hash = hex::encode(sha2::Sha256::digest(&plaintext));
690 + let file = temp_blob("complete-died.bin", &plaintext);
691 + let cipher_len = synckit_client::crypto::blob_encrypted_len(plaintext.len());
692 + let part_size = 700 * 1024;
693 + let part_count = cipher_len.div_ceil(part_size);
694 + assert!(
695 + part_count > 1,
696 + "the fixture must be a real multipart upload"
697 + );
698 +
699 + // ── First attempt: every part lands, the assemble call is refused ──
700 + let client = kit.authed();
701 + client.set_master_key_raw(key);
702 + client.set_resume_store(Arc::clone(&store));
703 +
704 + kit.post(START_PATH)
705 + .json(json!({
706 + "upload_id": "test-upload-id",
707 + "part_size": part_size,
708 + "part_count": part_count,
709 + "already_exists": false,
710 + }))
711 + .await;
712 + kit.post(PARTS_PATH)
713 + .responder(PartsResponder {
714 + cipher_len,
715 + part_size,
716 + base: kit.uri(),
717 + })
718 + .await;
719 + kit.put(PART_PUT_PATH)
720 + .responder(DiesAfter {
721 + // Never dies: this run is about what happens after the parts are up.
722 + ok: usize::MAX,
723 + seen: std::sync::atomic::AtomicUsize::new(0),
724 + })
725 + .await;
726 + // 403 rather than 500 so the client treats it as permanent and the test does
727 + // not sit through the retry backoff.
728 + kit.post(COMPLETE_PATH)
729 + .code(403)
730 + .json(json!({ "message": "assemble refused" }))
731 + .await;
732 + kit.post(ABORT_PATH).code(204).empty().await;
733 +
734 + let err = client
735 + .blob_upload_streaming(&hash, &file)
736 + .await
737 + .unwrap_err();
738 + assert!(
739 + matches!(err, SyncKitError::Server { status: 403, .. }),
740 + "got {err:?}"
741 + );
742 + let sent = put_bodies(&kit).await;
743 + assert_eq!(sent.len(), part_count, "the first attempt sent every part");
744 +
745 + let record = store
746 + .load(&hash)
747 + .unwrap()
748 + .expect("a failed complete keeps the session: it is what the retry needs");
749 + assert_eq!(
750 + record.usable_parts().len(),
751 + part_count,
752 + "every part must be recorded, or this is a different resume shape"
753 + );
754 +
755 + // ── Second attempt: nothing left to send ──
756 + kit.reset().await;
757 + mount_session(&kit, cipher_len, part_size).await;
758 + kit.post(ABORT_PATH).code(204).empty().await;
759 +
760 + let restarted = kit.authed();
761 + restarted.set_master_key_raw(key);
762 + restarted.set_resume_store(Arc::clone(&store));
763 + restarted.blob_upload_streaming(&hash, &file).await.unwrap();
764 +
765 + assert!(
766 + put_bodies(&kit).await.is_empty(),
767 + "a resume holding every part must not re-send one"
768 + );
769 + // The redundant session `start` opened is released rather than left to the reaper.
770 + assert_eq!(kit.hits(ABORT_PATH).await, 1);
771 +
772 + // What it did instead: named the parts the first run uploaded, with the
773 + // ETags that run was given.
774 + let complete = kit.body(COMPLETE_PATH).await;
775 + let named = complete["parts"].as_array().unwrap();
776 + assert_eq!(named.len(), part_count, "complete must name every part");
777 + for (i, part) in named.iter().enumerate() {
778 + assert_eq!(part["part_number"].as_u64(), Some(i as u64 + 1));
779 + assert_eq!(
780 + part["etag"].as_str(),
781 + Some(format!("\"etag-{}\"", i + 1)).as_deref(),
782 + "part {} lost the ETag the first run was given",
783 + i + 1
784 + );
785 + }
786 + assert_eq!(complete["upload_id"].as_str().unwrap(), "test-upload-id");
787 +
788 + // Guard against a vacuous pass: the bytes the first run sent really were the
789 + // whole blob, so assembling them is the right thing to have done.
790 + let assembled: Vec<u8> = sent.into_iter().flatten().collect();
791 + assert_eq!(assembled.len(), cipher_len);
792 + assert_eq!(
793 + synckit_client::crypto::decrypt_blob_chunked(&assembled, &key, &hash).unwrap(),
794 + plaintext
795 + );
796 +
797 + // Assembled means the record describes nothing.
798 + assert!(store.load(&hash).unwrap().is_none());
799 +
800 + std::fs::remove_file(&file).ok();
801 + }
802 +
803 + // ── How old a resume record may be ──
804 + //
805 + // `load_resume` drops a record older than half the server's 24h orphan-reaper
806 + // window, because past it the `upload_id` is gone and resuming costs a doomed
807 + // round trip. Nothing observed the boundary: every other test here builds its
808 + // record seconds ago, so the comparison and the constant behind it were free to
809 + // be anything.
810 + //
811 + // A store of the test's own is what makes the age settable. `age_secs` is a
812 + // field on the record the store hands back, so a double can hand back any age
813 + // it likes without a clock seam in the client.
814 +
815 + /// A resume store that answers with one record the test chose, and counts the
816 + /// `clear` calls that say the client rejected it.
817 + struct SeededResume {
818 + record: std::sync::Mutex<Option<synckit_client::client::resume::ResumeRecord>>,
819 + }
820 +
821 + impl BlobResumeStore for SeededResume {
822 + fn load(
823 + &self,
824 + _hash: &str,
825 + ) -> synckit_client::Result<Option<synckit_client::client::resume::ResumeRecord>> {
826 + Ok(self.record.lock().unwrap().clone())
827 + }
828 +
829 + fn begin(
830 + &self,
831 + _hash: &str,
832 + _session: &synckit_client::client::resume::ResumeSession,
833 + ) -> synckit_client::Result<()> {
834 + Ok(())
835 + }
836 +
837 + fn record_part(
838 + &self,
839 + _hash: &str,
840 + _part: &synckit_client::client::resume::ResumePart,
841 + _chunks: &[synckit_client::client::resume::ResumeChunk],
842 + ) -> synckit_client::Result<()> {
843 + Ok(())
844 + }
845 +
846 + fn clear(&self, _hash: &str) -> synckit_client::Result<()> {
847 + *self.record.lock().unwrap() = None;
848 + Ok(())
849 + }
850 + }
851 +
852 + /// Run an upload against a resume record of the given age that names every
853 + /// part, and report how many parts went over the wire.
854 + ///
855 + /// A record the client keeps costs zero PUTs: it already names every part, so
856 + /// there is nothing left to send and only `complete` runs. A record the client
857 + /// rejects costs the whole blob. The two answers are as far apart as the
858 + /// question allows, which is what makes this a usable oracle for one `>`.
859 + async fn parts_sent_resuming_a_record_aged(age_secs: i64) -> (usize, usize) {
860 + let kit = MockKit::start().await;
861 + let key = synckit_client::crypto::generate_master_key();
862 +
863 + let plaintext: Vec<u8> = (0..(synckit_client::crypto::BLOB_CHUNK_SIZE * 2 + 11))
864 + .map(|i| i as u8)
865 + .collect();
866 + let hash = hex::encode(sha2::Sha256::digest(&plaintext));
867 + let file = temp_blob("aged.bin", &plaintext);
868 + let cipher_len = synckit_client::crypto::blob_encrypted_len(plaintext.len());
869 + let part_size = 700 * 1024;
870 + let part_count = cipher_len.div_ceil(part_size);
871 +
872 + // A record that fits the file exactly and names every part, so keeping it
873 + // means sending nothing. No chunk records are needed: with every part
874 + // already up, the resume boundary lands past the last chunk and no chunk is
875 + // ever re-sealed.
876 + let record = synckit_client::client::resume::ResumeRecord {
877 + session: synckit_client::client::resume::ResumeSession {
878 + upload_id: "test-upload-id".into(),
879 + part_size: part_size as u64,
880 + part_count: part_count as u32,
881 + size_bytes: cipher_len as u64,
882 + },
883 + age_secs,
884 + parts: (1..=part_count)
885 + .map(|n| synckit_client::client::resume::ResumePart {
886 + part_number: n as u32,
887 + etag: format!("\"etag-{n}\""),
888 + })
889 + .collect(),
890 + chunks: Vec::new(),
891 + };
892 + let store: Arc<dyn BlobResumeStore> = Arc::new(SeededResume {
893 + record: std::sync::Mutex::new(Some(record)),
894 + });
895 +
896 + mount_session(&kit, cipher_len, part_size).await;
897 + kit.post(ABORT_PATH).code(204).empty().await;
898 +
899 + let client = kit.authed();
900 + client.set_master_key_raw(key);
901 + client.set_resume_store(store);
902 + client.blob_upload_streaming(&hash, &file).await.unwrap();
903 +
904 + let sent = put_bodies(&kit).await.len();
905 + std::fs::remove_file(&file).ok();
906 + (sent, part_count)
907 + }
908 +
909 + /// The record survives to exactly twelve hours and no further.
910 + ///
911 + /// Both halves are needed and they pin different things. The `at` case pins the
912 + /// comparison (`>` rather than `>=`, so a record on the boundary is still
913 + /// usable) and the constant behind it: a `RESUME_MAX_AGE_SECS` that arithmetic
914 + /// had made smaller would discard this record too. The `past` case is what
915 + /// stops the whole guard from being deleted.
916 + #[tokio::test]
917 + async fn a_resume_record_survives_to_twelve_hours_and_not_past_it() {
918 + /// `RESUME_MAX_AGE_SECS` from `src/client/blob.rs`, private there. Half the
919 + /// server's 24h orphan-reaper window.
920 + const RESUME_MAX_AGE_SECS: i64 = 12 * 60 * 60;
921 +
922 + let (at_the_limit, part_count) = parts_sent_resuming_a_record_aged(RESUME_MAX_AGE_SECS).await;
923 + assert_eq!(
924 + at_the_limit, 0,
925 + "a record exactly at the age limit is still usable: it names every part, so a resume sends nothing"
926 + );
927 +
928 + let (past_the_limit, _) = parts_sent_resuming_a_record_aged(RESUME_MAX_AGE_SECS + 1).await;
929 + assert_eq!(
930 + past_the_limit, part_count,
931 + "a record one second past the limit must be discarded and the blob sent from part one"
932 + );
933 + }
934 +
668 935 // ── Part-boundary arithmetic ──
669 936 //
670 937 // The part plan is arithmetic on the ciphertext length, and the cases that break