fix: auto-size multipart parts for Dir.from_local_sync uploads

rs_controller/src/core.rs

@@ -67,6 +67,14 @@ pub enum StateClient {
     Authenticated(StateServiceClient<crate::auth::AuthService<tonic::transport::Channel>>),
 }
 
+// The three client enums here exist only to erase the Plain/Authenticated
+// distinction, so each method is a drop-in replacement for the tonic client method
+// it forwards to -- error type included. `clippy::result_large_err` fires because
+// `tonic::Status` is 176 bytes, but boxing it in these signatures would break that
+// drop-in property and make every caller unbox a `Status` it is about to hand back
+// to tonic-shaped code. `ControllerError` already boxes it where it is *stored*
+// (see error.rs), which is where the size actually costs something.
+#[allow(clippy::result_large_err)]
 impl StateClient {
     pub async fn watch(
         &mut self,
@@ -85,6 +93,7 @@ pub enum QueueClient {
     Authenticated(QueueServiceClient<crate::auth::AuthService<tonic::transport::Channel>>),
 }
 
+#[allow(clippy::result_large_err)]
 impl QueueClient {
     pub async fn enqueue_action(
         &mut self,
@@ -103,6 +112,7 @@ pub enum ActionsClient {
     Authenticated(ActionsServiceClient<crate::auth::AuthService<tonic::transport::Channel>>),
 }
 
+#[allow(clippy::result_large_err)]
 impl ActionsClient {
     pub async fn enqueue(
         &mut self,

src/flyte/io/_dir.py

@@ -786,9 +786,14 @@ async def _lazy_uploader() -> tuple[str | None, str]:
             output_path = str(Path(local_path).absolute())
             return cls(path=output_path, name=dirname, hash=dir_cache_key)
 
-        fs = storage.get_underlying_filesystem(path=resolved_remote_path)
-        fs.put(local_path_str, resolved_remote_path, recursive=True)
-        return cls(path=resolved_remote_path, name=dirname, hash=dir_cache_key)
+        # Route through the obstore-aware storage.put (via syncify) rather than the raw fsspec
+        # fs.put(..., recursive=True). fsspec's obstore backend calls store.put_async WITHOUT a
+        # chunk_size, pinning every multipart part to obstore's 5 MiB default and thus a
+        # ~48.8 GiB per-file ceiling (10,000-part S3 limit). storage.put auto-sizes the part
+        # size to each file, so large files upload correctly. Mirrors the async from_local()
+        # above and download_sync().
+        output_path = syncify(storage.put)(from_path=local_path_str, to_path=resolved_remote_path, recursive=True)
+        return cls(path=output_path, name=dirname, hash=dir_cache_key)
 
     @classmethod
     def new_remote(cls, dir_name: Optional[str] = None, hash: Optional[str] = None) -> Dir[T]: