api: select xet object upload mode

Author: flaviusburca

api/swagger.yml

@@ -415,6 +415,19 @@ components:
           type: boolean
           default: false
 
+    ObjectUploadMode:
+      type: object
+      required:
+        - upload_mode
+      properties:
+        upload_mode:
+          type: string
+          enum: [regular, xet]
+          description: |
+            Server-selected upload mode for a regular object upload.
+            "regular" means the client should use POST /objects.
+            "xet" means the client may upload through XET and link an xet:// physical address.
+
     ObjectUserMetadata:
       type: object
       additionalProperties:
@@ -4900,13 +4913,24 @@ paths:
         required: true
         schema:
           type: string
+      - in: query
+        name: size_bytes
+        description: Size of the object content the client plans to upload.
+        required: false
+        schema:
+          type: integer
+          format: int64
     get:
       tags:
         - internal
       operationId: uploadObjectPreflight
       responses:
-        204:
-          description: User has permissions to upload this object. This does not guarantee that the upload will be successful or even possible. It indicates only the permission at the time of calling this endpoint
+        200:
+          description: User has permissions to upload this object and the server-selected upload mode.
+          content:
+            application/json:
+              schema:
+                $ref: "#/components/schemas/ObjectUploadMode"
         401:
           $ref: "#/components/responses/Unauthorized"
         403:

pkg/api/controller.go

@@ -3298,7 +3298,23 @@ func (c *Controller) UploadObjectPreflight(w http.ResponseWriter, r *http.Reques
 	ctx := r.Context()
 	c.LogAction(ctx, "put_object_preflight", r, repository, branch, "")
 
-	writeResponse(w, r, http.StatusNoContent, nil)
+	writeResponse(w, r, http.StatusOK, apigen.ObjectUploadMode{
+		UploadMode: c.objectUploadMode(params.SizeBytes),
+	})
+}
+
+func (c *Controller) objectUploadMode(sizeBytes *int64) string {
+	if sizeBytes == nil || *sizeBytes <= 0 {
+		return "regular"
+	}
+	minSize := c.Config.GetBaseConfig().XET.Upload.MinSizeBytes
+	if minSize <= 0 {
+		return "regular"
+	}
+	if *sizeBytes >= minSize {
+		return "xet"
+	}
+	return "regular"
 }
 
 func (c *Controller) UploadObject(w http.ResponseWriter, r *http.Request, owner, repository, branch string, params apigen.UploadObjectParams) {

pkg/api/controller_test.go

@@ -126,6 +126,59 @@ func TestController_LinkXETPhysicalAddressWritesFileRef(t *testing.T) {
 	require.Equal(t, []byte{}, fileRef.Value)
 }
 
+func TestController_UploadObjectPreflightReturnsRegularForSmallObject(t *testing.T) {
+	clt, deps := setupClientWithAdmin(t)
+	ctx := context.Background()
+	repo := testUniqueRepoName()
+	const branch = "main"
+	_, err := deps.catalog.CreateRepository(ctx, repo, "", onBlock(deps, "bucket/prefix"), branch, false)
+	require.NoError(t, err)
+
+	resp, err := clt.UploadObjectPreflightWithResponse(ctx, apigen.RepositoryOwner(repo), apigen.RepositoryName(repo), branch, &apigen.UploadObjectPreflightParams{
+		Path:      "small.txt",
+		SizeBytes: apiutil.Ptr(int64(1024)),
+	})
+	require.NoError(t, err)
+	require.Equal(t, http.StatusOK, resp.StatusCode())
+	require.NotNil(t, resp.JSON200)
+	require.Equal(t, "regular", resp.JSON200.UploadMode)
+}
+
+func TestController_UploadObjectPreflightReturnsXETForLargeObject(t *testing.T) {
+	clt, deps := setupClientWithAdmin(t)
+	ctx := context.Background()
+	repo := testUniqueRepoName()
+	const branch = "main"
+	_, err := deps.catalog.CreateRepository(ctx, repo, "", onBlock(deps, "bucket/prefix"), branch, false)
+	require.NoError(t, err)
+
+	resp, err := clt.UploadObjectPreflightWithResponse(ctx, apigen.RepositoryOwner(repo), apigen.RepositoryName(repo), branch, &apigen.UploadObjectPreflightParams{
+		Path:      "models/model.bin",
+		SizeBytes: apiutil.Ptr(int64(8 * 1024 * 1024)),
+	})
+	require.NoError(t, err)
+	require.Equal(t, http.StatusOK, resp.StatusCode())
+	require.NotNil(t, resp.JSON200)
+	require.Equal(t, "xet", resp.JSON200.UploadMode)
+}
+
+func TestController_UploadObjectPreflightReturnsRegularWhenSizeUnknown(t *testing.T) {
+	clt, deps := setupClientWithAdmin(t)
+	ctx := context.Background()
+	repo := testUniqueRepoName()
+	const branch = "main"
+	_, err := deps.catalog.CreateRepository(ctx, repo, "", onBlock(deps, "bucket/prefix"), branch, false)
+	require.NoError(t, err)
+
+	resp, err := clt.UploadObjectPreflightWithResponse(ctx, apigen.RepositoryOwner(repo), apigen.RepositoryName(repo), branch, &apigen.UploadObjectPreflightParams{
+		Path: "stream.bin",
+	})
+	require.NoError(t, err)
+	require.Equal(t, http.StatusOK, resp.StatusCode())
+	require.NotNil(t, resp.JSON200)
+	require.Equal(t, "regular", resp.JSON200.UploadMode)
+}
+
 func TestController_LinkXETPhysicalAddressRejectsMissingShard(t *testing.T) {
 	clt, deps := setupClientWithAdmin(t)
 	ctx := context.Background()

pkg/config/config.go

@@ -474,6 +474,9 @@ type BaseConfig struct {
 		PrepareInterval    time.Duration `mapstructure:"prepare_interval"`
 	} `mapstructure:"ugc"`
 	XET struct {
+		Upload struct {
+			MinSizeBytes int64 `mapstructure:"min_size_bytes"`
+		} `mapstructure:"upload"`
 		Verify struct {
 			MaxConcurrent int `mapstructure:"max_concurrent"`
 		} `mapstructure:"verify"`

pkg/config/defaults.go

@@ -170,6 +170,7 @@ func setBaseDefaults(cfgType string) {
 	viper.SetDefault("ugc.prepare_interval", time.Minute)
 	viper.SetDefault("ugc.prepare_max_file_size", 20*1024*1024)
 
+	viper.SetDefault("xet.upload.min_size_bytes", int64(5*1024*1024))
 	viper.SetDefault("xet.verify.max_concurrent", 0)
 	viper.SetDefault("xet.read.capability_scan_batch_size", 32)
 	viper.SetDefault("xet.gc.min_age", 24*time.Hour)