Skip to main content

max / makenotwork

mnw-cli: upload large files through the multipart session The CLI read whole files into memory and issued a single presigned PUT, which is the same one-shot transport the browser is refused for above 5 GiB. So the server's "Files larger than 5 GB must be uploaded with the makenot.work CLI" message pointed at a client that could not do it either, and the 5 GiB to 20 GB tier band was unreachable by every path. Add a multipart client (start/parts/complete/abort) and a driver that holds one part in memory at a time: read a part, PUT it to its presigned URL, collect the ETag, complete. Part URLs are pulled in windows rather than all at once, each sent with the exact Content-Length its signature binds. Part uploads retry transient failures, and any failure past the session opening aborts it from a single site so a half-finished transfer does not leave parts billing. publish_file routes files above 64 MiB through it and keeps the simpler single PUT below that. Confirm is unchanged either way: multipart replaces the transport only. Also read the OTA artifact with tokio::fs, since a blocking std::fs::read of a multi-hundred-MB artifact stalled the runtime on every publish. 58 tests green. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
Author: Max Johnson <me@maxj.phd> · 2026-07-21 13:32 UTC
Commit: 35d06a3a708d0f274ec4b7b363df23ab070fb707
Parent: 98e7254
3 files changed, +349 insertions, -14 deletions
M mnw-cli/src/api.rs +310
@@ -72,6 +72,40 @@ pub struct PresignResponse {
72 72 pub cache_control: Option<String>,
73 73 }
74 74
75 + /// Above this size an upload goes through a multipart session instead of one
76 + /// presigned PUT. The single-PUT path reads the whole file into memory, which is
77 + /// fine for a small file and unacceptable for a multi-GB one; the multipart path
78 + /// holds one part at a time. It is also the only path that can carry a file past
79 + /// S3's 5 GiB single-PUT ceiling, which is what the tier limits allow for.
80 + pub const MULTIPART_THRESHOLD_BYTES: u64 = 64 * 1024 * 1024;
81 +
82 + /// How many presigned part URLs to request at a time. Must not exceed the
83 + /// server's own window cap.
84 + const PART_URL_WINDOW: u32 = 100;
85 +
86 + /// An opened multipart upload session.
87 + #[derive(Debug, Clone, Deserialize)]
88 + pub struct MultipartStart {
89 + pub upload_id: String,
90 + pub s3_key: String,
91 + pub part_size: u64,
92 + pub part_count: u32,
93 + pub expires_in: u64,
94 + }
95 +
96 + /// One presigned part target, with the exact length the signature binds.
97 + #[derive(Debug, Clone, Deserialize)]
98 + pub struct MultipartPartUrl {
99 + pub part_number: i32,
100 + pub content_length: u64,
101 + pub url: String,
102 + }
103 +
104 + #[derive(Debug, Deserialize)]
105 + struct MultipartPartsResponse {
106 + parts: Vec<MultipartPartUrl>,
107 + }
108 +
75 109 /// Full item detail returned from the get/update endpoints.
76 110 #[derive(Debug, Clone, Deserialize, Serialize)]
77 111 pub struct ItemDetail {
@@ -686,6 +720,282 @@ impl MnwApiClient {
686 720 json_response(resp, "get_item_versions").await
687 721 }
688 722
723 + // ── Multipart upload session (large files) ──
724 +
725 + /// Open a multipart upload session and get the part geometry.
726 + pub async fn multipart_start(
727 + &self,
728 + item_id: &str,
729 + file_type: &str,
730 + file_name: &str,
731 + content_type: &str,
732 + file_size_bytes: u64,
733 + ) -> anyhow::Result<MultipartStart> {
734 + let url = format!("{}/api/internal/upload/multipart/start", self.base_url);
735 + let resp = self
736 + .http
737 + .post(&url)
738 + .bearer_auth(&self.service_token)
739 + .header("X-MNW-Actor", self.actor_header())
740 + .json(&serde_json::json!({
741 + "item_id": item_id,
742 + "file_type": file_type,
743 + "file_name": file_name,
744 + "content_type": content_type,
745 + "file_size_bytes": file_size_bytes,
746 + }))
747 + .send()
748 + .await?;
749 +
750 + json_response(resp, "multipart_start").await
751 + }
752 +
753 + /// Fetch a bounded window of presigned part URLs.
754 + async fn multipart_parts(
755 + &self,
756 + s3_key: &str,
757 + upload_id: &str,
758 + file_size_bytes: u64,
759 + first_part: u32,
760 + count: u32,
761 + ) -> anyhow::Result<Vec<MultipartPartUrl>> {
762 + let url = format!("{}/api/internal/upload/multipart/parts", self.base_url);
763 + let resp = self
764 + .http
765 + .post(&url)
766 + .bearer_auth(&self.service_token)
767 + .header("X-MNW-Actor", self.actor_header())
768 + .json(&serde_json::json!({
769 + "s3_key": s3_key,
770 + "upload_id": upload_id,
771 + "file_size_bytes": file_size_bytes,
772 + "first_part": first_part,
773 + "count": count,
774 + }))
775 + .send()
776 + .await?;
777 +
778 + let parts: MultipartPartsResponse = json_response(resp, "multipart_parts").await?;
779 + Ok(parts.parts)
780 + }
781 +
782 + /// Assemble the uploaded parts into the staging object.
783 + async fn multipart_complete(
784 + &self,
785 + s3_key: &str,
786 + upload_id: &str,
787 + parts: &[(i32, String)],
788 + ) -> anyhow::Result<()> {
789 + let url = format!("{}/api/internal/upload/multipart/complete", self.base_url);
790 + let parts: Vec<serde_json::Value> = parts
791 + .iter()
792 + .map(|(n, etag)| serde_json::json!({ "part_number": n, "etag": etag }))
793 + .collect();
794 + let resp = self
795 + .http
796 + .post(&url)
797 + .bearer_auth(&self.service_token)
798 + .header("X-MNW-Actor", self.actor_header())
799 + .json(&serde_json::json!({
800 + "s3_key": s3_key,
801 + "upload_id": upload_id,
802 + "parts": parts,
803 + }))
804 + .send()
805 + .await?;
806 +
807 + if !resp.status().is_success() {
808 + anyhow::bail!(
809 + "multipart_complete failed: HTTP {} {}",
810 + resp.status(),
811 + resp.text().await.unwrap_or_default()
812 + );
813 + }
814 + Ok(())
815 + }
816 +
817 + /// Release the parts of an abandoned session. Incomplete multipart uploads
818 + /// bill for their parts until aborted.
819 + pub async fn multipart_abort(&self, s3_key: &str, upload_id: &str) -> anyhow::Result<()> {
820 + let url = format!("{}/api/internal/upload/multipart/abort", self.base_url);
821 + let resp = self
822 + .http
823 + .post(&url)
824 + .bearer_auth(&self.service_token)
825 + .header("X-MNW-Actor", self.actor_header())
826 + .json(&serde_json::json!({ "s3_key": s3_key, "upload_id": upload_id }))
827 + .send()
828 + .await?;
829 +
830 + if !resp.status().is_success() {
831 + anyhow::bail!("multipart_abort failed: HTTP {}", resp.status());
832 + }
833 + Ok(())
834 + }
835 +
836 + /// Upload a file through a multipart session, holding one part in memory at
837 + /// a time. Returns the staging key to confirm against.
838 + ///
839 + /// `on_progress` is called with `(bytes_uploaded, total)` after each part.
840 + /// Any failure past the session opening aborts it, so a half-finished upload
841 + /// does not leave parts billing indefinitely — the single abort site means a
842 + /// future failure path added inside cannot forget to.
843 + #[allow(clippy::too_many_arguments)]
844 + pub async fn upload_file_multipart(
845 + &self,
846 + item_id: &str,
847 + file_type: &str,
848 + file_name: &str,
849 + content_type: &str,
850 + file_path: &std::path::Path,
851 + file_size: u64,
852 + mut on_progress: impl FnMut(u64, u64),
853 + ) -> anyhow::Result<String> {
854 + let start = self
855 + .multipart_start(item_id, file_type, file_name, content_type, file_size)
856 + .await?;
857 +
858 + tracing::info!(
859 + s3_key = %start.s3_key,
860 + part_count = start.part_count,
861 + part_size = start.part_size,
862 + expires_in = start.expires_in,
863 + file_size,
864 + "multipart upload session opened"
865 + );
866 +
867 + match self
868 + .run_multipart_upload(&start, file_path, file_size, &mut on_progress)
869 + .await
870 + {
871 + Ok(()) => Ok(start.s3_key),
872 + Err(e) => {
873 + if let Err(abort_err) = self.multipart_abort(&start.s3_key, &start.upload_id).await
874 + {
875 + tracing::warn!(
876 + error = %abort_err, s3_key = %start.s3_key,
877 + "failed to abort multipart upload after a failed transfer"
878 + );
879 + }
880 + Err(e)
881 + }
882 + }
883 + }
884 +
885 + /// Read and upload every part, then complete. Returns `Err` without
886 + /// aborting; the caller owns the single abort.
887 + async fn run_multipart_upload(
888 + &self,
889 + start: &MultipartStart,
890 + file_path: &std::path::Path,
891 + file_size: u64,
892 + on_progress: &mut impl FnMut(u64, u64),
893 + ) -> anyhow::Result<()> {
894 + use tokio::io::AsyncReadExt;
895 +
896 + let mut file = tokio::fs::File::open(file_path).await.map_err(|e| {
897 + anyhow::anyhow!("opening {} for upload: {e}", file_path.display())
898 + })?;
899 +
900 + let mut completed: Vec<(i32, String)> = Vec::with_capacity(start.part_count as usize);
901 + let mut uploaded: u64 = 0;
902 + let mut next: u32 = 1;
903 +
904 + while next <= start.part_count {
905 + let count = PART_URL_WINDOW.min(start.part_count - next + 1);
906 + let urls = self
907 + .multipart_parts(&start.s3_key, &start.upload_id, file_size, next, count)
908 + .await?;
909 + if urls.is_empty() {
910 + anyhow::bail!("server returned no part URLs for part {next}");
911 + }
912 +
913 + for part in urls {
914 + // One part resident at a time — this is the whole point of the
915 + // multipart path over the single-PUT one.
916 + let mut buf = vec![0u8; part.content_length as usize];
917 + file.read_exact(&mut buf).await.map_err(|e| {
918 + anyhow::anyhow!(
919 + "reading part {} ({} bytes) from {}: {e}",
920 + part.part_number,
921 + part.content_length,
922 + file_path.display()
923 + )
924 + })?;
925 +
926 + let etag = self.put_part(&part, buf).await?;
927 + completed.push((part.part_number, etag));
928 + uploaded += part.content_length;
929 + on_progress(uploaded, file_size);
930 + next += 1;
931 + }
932 + }
933 +
934 + self.multipart_complete(&start.s3_key, &start.upload_id, &completed)
935 + .await
936 + }
937 +
938 + /// PUT one part to its presigned URL, returning the ETag the completion call
939 + /// needs. Retries transient failures — losing a part to a network blip
940 + /// should not discard the whole transfer.
941 + async fn put_part(&self, part: &MultipartPartUrl, body: Vec<u8>) -> anyhow::Result<String> {
942 + // `Bytes` so a retry clones a refcount rather than re-copying the part.
943 + let body = bytes::Bytes::from(body);
944 + let mut attempt: u32 = 0;
945 + loop {
946 + attempt += 1;
947 + let sent = self
948 + .http
949 + .put(&part.url)
950 + .header(reqwest::header::CONTENT_LENGTH, part.content_length)
951 + .body(body.clone())
952 + .send()
953 + .await;
954 +
955 + let retriable = match sent {
956 + Ok(resp) if resp.status().is_success() => {
957 + let etag = resp
958 + .headers()
959 + .get(reqwest::header::ETAG)
960 + .and_then(|v| v.to_str().ok())
961 + .unwrap_or_default()
962 + .to_string();
963 + if etag.is_empty() {
964 + anyhow::bail!(
965 + "S3 returned no ETag for part {}; cannot complete the upload",
966 + part.part_number
967 + );
968 + }
969 + return Ok(etag);
970 + }
971 + Ok(resp) => {
972 + let status = resp.status();
973 + if attempt >= 3 {
974 + anyhow::bail!(
975 + "part {} failed after {attempt} attempts: HTTP {status}",
976 + part.part_number
977 + );
978 + }
979 + format!("HTTP {status}")
980 + }
981 + Err(e) => {
982 + if attempt >= 3 {
983 + return Err(anyhow::Error::new(e)
984 + .context(format!("part {} failed after {attempt} attempts", part.part_number)));
985 + }
986 + e.to_string()
987 + }
988 + };
989 +
990 + let delay_ms = 200u64 * (1u64 << ((attempt - 1) * 2));
991 + tracing::warn!(
992 + part_number = part.part_number, attempt, delay_ms, error = %retriable,
993 + "part upload transient failure, retrying"
994 + );
995 + tokio::time::sleep(std::time::Duration::from_millis(delay_ms)).await;
996 + }
997 + }
998 +
689 999 /// Upload a file to S3 using a presigned URL.
690 1000 pub async fn upload_to_s3(
691 1001 &self,
M mnw-cli/src/ota.rs +4 -1
@@ -182,7 +182,10 @@ fn parse_args(flags: &[String]) -> Result<PublishArgs> {
182 182 async fn publish(flags: &[String]) -> Result<()> {
183 183 let args = parse_args(flags)?;
184 184
185 - let bytes = std::fs::read(&args.artifact)
185 + // Async read: a blocking `std::fs::read` of a multi-hundred-MB artifact
186 + // stalls the runtime for the whole read on every publish.
187 + let bytes = tokio::fs::read(&args.artifact)
188 + .await
186 189 .with_context(|| format!("reading artifact {}", args.artifact.display()))?;
187 190 let file_size: i64 = bytes
188 191 .len()
M mnw-cli/src/tui/loading.rs +35 -13
@@ -286,22 +286,44 @@ pub(super) async fn publish_file(
286 286 .create_item(user_id, project_id, title, item_type, price_cents)
287 287 .await?;
288 288
289 - // Step 2: Get presigned URL
290 - let presign = api
291 - .presign_upload(user_id, &item.item_id, file_type, filename, content_type)
292 - .await?;
289 + // Steps 2+3: get an upload target and send the bytes. A large file goes
290 + // through a multipart session (one part resident at a time, and the only
291 + // path that clears S3's 5 GiB single-PUT ceiling); a small one keeps the
292 + // simpler single presigned PUT.
293 + let file_size = tokio::fs::metadata(file_path)
294 + .await
295 + .map_err(|e| anyhow::anyhow!("reading {}: {e}", file_path.display()))?
296 + .len();
293 297
294 - // Step 3: Upload to S3
295 - api.upload_to_s3(
296 - &presign.upload_url,
297 - file_path,
298 - content_type,
299 - presign.cache_control.as_deref(),
300 - )
301 - .await?;
298 + let s3_key = if file_size > crate::api::MULTIPART_THRESHOLD_BYTES {
299 + api.upload_file_multipart(
300 + &item.item_id,
301 + file_type,
302 + filename,
303 + content_type,
304 + file_path,
305 + file_size,
306 + |uploaded, total| {
307 + tracing::debug!(uploaded, total, "multipart upload progress");
308 + },
309 + )
310 + .await?
311 + } else {
312 + let presign = api
313 + .presign_upload(user_id, &item.item_id, file_type, filename, content_type)
314 + .await?;
315 + api.upload_to_s3(
316 + &presign.upload_url,
317 + file_path,
318 + content_type,
319 + presign.cache_control.as_deref(),
320 + )
321 + .await?;
322 + presign.s3_key
323 + };
302 324
303 325 // Step 4: Confirm upload
304 - api.confirm_upload(user_id, &item.item_id, file_type, &presign.s3_key)
326 + api.confirm_upload(user_id, &item.item_id, file_type, &s3_key)
305 327 .await?;
306 328
307 329 // Step 5: Delete staging file