Allow adding metadata to objects

.github/workflows/ui.yml

@@ -26,11 +26,11 @@ jobs:
         run: |
           # Fetch tags with pagination
           TAGS_JSON=$(gh api --paginate repos/SiaFoundation/web/tags)
-      
+
           # Extract tags that start with "renterd/", sort them in version order, and pick the highest version
           LATEST_RENTERD_GO_TAG=$(echo "$TAGS_JSON" | jq -r '.[] | select(.name | startswith("renterd/")).name' | sort -Vr | head -n 1)
           LATEST_RENTERD_VERSION=$(echo "$LATEST_RENTERD_GO_TAG" | sed 's/renterd\///')
-      
+
           echo "Latest renterd tag is $LATEST_RENTERD_GO_TAG"
           echo "GO_TAG=$LATEST_RENTERD_GO_TAG" >> $GITHUB_ENV
           echo "VERSION=$LATEST_RENTERD_VERSION" >> $GITHUB_ENV

api/multipart.go

@@ -48,6 +48,7 @@ type (
 	CreateMultipartOptions struct {
 		Key      object.EncryptionKey
 		MimeType string
+		Metadata ObjectUserMetadata
 	}
 )
 
@@ -84,6 +85,7 @@ type (
 		Path     string               `json:"path"`
 		Key      object.EncryptionKey `json:"key"`
 		MimeType string               `json:"mimeType"`
+		Metadata ObjectUserMetadata   `json:"metadata"`
 	}
 
 	MultipartCreateResponse struct {

api/object.go

@@ -3,6 +3,7 @@ package api
 import (
 	"errors"
 	"fmt"
+	"io"
 	"mime"
 	"net/http"
 	"net/url"
@@ -13,6 +14,8 @@ import (
 )
 
 const (
+	ObjectMetadataPrefix = "X-Sia-Meta-"
+
 	ObjectsRenameModeSingle = "single"
 	ObjectsRenameModeMulti  = "multi"
 
@@ -45,6 +48,7 @@ var (
 type (
 	// Object wraps an object.Object with its metadata.
 	Object struct {
+		Metadata ObjectUserMetadata `json:"metadata,omitempty"`
 		ObjectMetadata
 		object.Object
 	}
@@ -53,20 +57,20 @@ type (
 	ObjectMetadata struct {
 		ETag     string      `json:"eTag,omitempty"`
 		Health   float64     `json:"health"`
-		MimeType string      `json:"mimeType,omitempty"`
 		ModTime  TimeRFC3339 `json:"modTime"`
 		Name     string      `json:"name"`
 		Size     int64       `json:"size"`
+		MimeType string      `json:"mimeType,omitempty"`
 	}
 
-	// ObjectAddRequest is the request type for the /bus/object/*key endpoint.
-	ObjectAddRequest struct {
-		Bucket      string        `json:"bucket"`
-		ContractSet string        `json:"contractSet"`
-		Object      object.Object `json:"object"`
-		MimeType    string        `json:"mimeType"`
-		ETag        string        `json:"eTag"`
-	}
+	// ObjectUserMetadata contains user-defined metadata about an object and can
+	// be provided through `X-Sia-Meta-` meta headers.
+	//
+	// NOTE: `X-Amz-Meta-` headers are supported and will be converted to sia
+	// metadata headers internally, this means that S3 clients can safely keep
+	// using Amazon headers and find the metadata will be persisted in Sia as
+	// well
+	ObjectUserMetadata map[string]string
 
 	// ObjectsResponse is the response type for the /bus/objects endpoint.
 	ObjectsResponse struct {
@@ -75,15 +79,14 @@ type (
 		Object  *Object          `json:"object,omitempty"`
 	}
 
-	// ObjectsCopyRequest is the request type for the /bus/objects/copy endpoint.
-	ObjectsCopyRequest struct {
-		SourceBucket string `json:"sourceBucket"`
-		SourcePath   string `json:"sourcePath"`
-
-		DestinationBucket string `json:"destinationBucket"`
-		DestinationPath   string `json:"destinationPath"`
-
-		MimeType string `json:"mimeType"`
+	// GetObjectResponse is the response type for the /worker/object endpoint.
+	GetObjectResponse struct {
+		Content      io.ReadCloser      `json:"content"`
+		ContentType  string             `json:"contentType"`
+		LastModified string             `json:"lastModified"`
+		Range        *DownloadRange     `json:"range,omitempty"`
+		Size         int64              `json:"size"`
+		Metadata     ObjectUserMetadata `json:"metadata"`
 	}
 
 	// ObjectsDeleteRequest is the request type for the /bus/objects/list endpoint.
@@ -124,6 +127,16 @@ type (
 	}
 )
 
+func ExtractObjectUserMetadataFrom(metadata map[string]string) ObjectUserMetadata {
+	oum := make(map[string]string)
+	for k, v := range metadata {
+		if strings.HasPrefix(strings.ToLower(k), strings.ToLower(ObjectMetadataPrefix)) {
+			oum[k[len(ObjectMetadataPrefix):]] = v
+		}
+	}
+	return oum
+}
+
 // LastModified returns the object's ModTime formatted for use in the
 // 'Last-Modified' header
 func (o ObjectMetadata) LastModified() string {
@@ -146,13 +159,39 @@ func (o ObjectMetadata) ContentType() string {
 }
 
 type (
+	// AddObjectOptions is the options type for the bus client.
 	AddObjectOptions struct {
-		MimeType string
 		ETag     string
+		MimeType string
+		Metadata ObjectUserMetadata
 	}
 
+	// AddObjectRequest is the request type for the /bus/object/*key endpoint.
+	AddObjectRequest struct {
+		Bucket      string             `json:"bucket"`
+		ContractSet string             `json:"contractSet"`
+		Object      object.Object      `json:"object"`
+		ETag        string             `json:"eTag"`
+		MimeType    string             `json:"mimeType"`
+		Metadata    ObjectUserMetadata `json:"metadata"`
+	}
+
+	// CopyObjectOptions is the options type for the bus client.
 	CopyObjectOptions struct {
 		MimeType string
+		Metadata ObjectUserMetadata
+	}
+
+	// CopyObjectsRequest is the request type for the /bus/objects/copy endpoint.
+	CopyObjectsRequest struct {
+		SourceBucket string `json:"sourceBucket"`
+		SourcePath   string `json:"sourcePath"`
+
+		DestinationBucket string `json:"destinationBucket"`
+		DestinationPath   string `json:"destinationPath"`
+
+		MimeType string             `json:"mimeType"`
+		Metadata ObjectUserMetadata `json:"metadata"`
 	}
 
 	DeleteObjectOptions struct {
@@ -186,14 +225,16 @@ type (
 		Limit  int
 	}
 
+	// UploadObjectOptions is the options type for the worker client.
 	UploadObjectOptions struct {
 		Offset                       int
 		MinShards                    int
 		TotalShards                  int
 		ContractSet                  string
-		MimeType                     string
 		DisablePreshardingEncryption bool
 		ContentLength                int64
+		MimeType                     string
+		Metadata                     ObjectUserMetadata
 	}
 
 	UploadMultipartUploadPartOptions struct {
@@ -203,7 +244,7 @@ type (
 	}
 )
 
-func (opts UploadObjectOptions) Apply(values url.Values) {
+func (opts UploadObjectOptions) ApplyValues(values url.Values) {
 	if opts.Offset != 0 {
 		values.Set("offset", fmt.Sprint(opts.Offset))
 	}
@@ -224,6 +265,12 @@ func (opts UploadObjectOptions) Apply(values url.Values) {
 	}
 }
 
+func (opts UploadObjectOptions) ApplyHeaders(h http.Header) {
+	for k, v := range opts.Metadata {
+		h.Set(ObjectMetadataPrefix+k, v)
+	}
+}
+
 func (opts UploadMultipartUploadPartOptions) Apply(values url.Values) {
 	if opts.DisablePreshardingEncryption {
 		values.Set("disablepreshardingencryption", "true")

api/param.go

@@ -105,7 +105,7 @@ func (t *TimeRFC3339) UnmarshalText(b []byte) error {
 
 // MarshalJSON implements json.Marshaler.
 func (t TimeRFC3339) MarshalJSON() ([]byte, error) {
-	return []byte(fmt.Sprintf(`"%s"`, (time.Time)(t).UTC().Format(time.RFC3339))), nil
+	return []byte(fmt.Sprintf(`"%s"`, (time.Time)(t).UTC().Format(time.RFC3339Nano))), nil
 }
 
 // String implements fmt.Stringer.

api/worker.go

@@ -3,7 +3,6 @@ package api
 import (
 	"errors"
 	"fmt"
-	"io"
 	"strconv"
 	"strings"
 
@@ -217,14 +216,6 @@ type (
 	UploadMultipartUploadPartResponse struct {
 		ETag string `json:"etag"`
 	}
-
-	GetObjectResponse struct {
-		Content     io.ReadCloser  `json:"content"`
-		ContentType string         `json:"contentType"`
-		ModTime     TimeRFC3339    `json:"modTime"`
-		Range       *DownloadRange `json:"range,omitempty"`
-		Size        int64          `json:"size"`
-	}
 )
 
 type DownloadRange struct {

bus/bus.go

@@ -134,7 +134,7 @@ type (
 		ListBuckets(_ context.Context) ([]api.Bucket, error)
 		UpdateBucketPolicy(ctx context.Context, bucketName string, policy api.BucketPolicy) error
 
-		CopyObject(ctx context.Context, srcBucket, dstBucket, srcPath, dstPath, mimeType string) (api.ObjectMetadata, error)
+		CopyObject(ctx context.Context, srcBucket, dstBucket, srcPath, dstPath, mimeType string, metadata api.ObjectUserMetadata) (api.ObjectMetadata, error)
 		ListObjects(ctx context.Context, bucketName, prefix, sortBy, sortDir, marker string, limit int) (api.ObjectsListResponse, error)
 		Object(ctx context.Context, bucketName, path string) (api.Object, error)
 		ObjectEntries(ctx context.Context, bucketName, path, prefix, sortBy, sortDir, marker string, offset, limit int) ([]api.ObjectMetadata, bool, error)
@@ -145,12 +145,12 @@ type (
 		RenameObject(ctx context.Context, bucketName, from, to string, force bool) error
 		RenameObjects(ctx context.Context, bucketName, from, to string, force bool) error
 		SearchObjects(ctx context.Context, bucketName, substring string, offset, limit int) ([]api.ObjectMetadata, error)
-		UpdateObject(ctx context.Context, bucketName, path, contractSet, ETag, mimeType string, o object.Object) error
+		UpdateObject(ctx context.Context, bucketName, path, contractSet, ETag, mimeType string, metadata api.ObjectUserMetadata, o object.Object) error
 
 		AbortMultipartUpload(ctx context.Context, bucketName, path string, uploadID string) (err error)
 		AddMultipartPart(ctx context.Context, bucketName, path, contractSet, eTag, uploadID string, partNumber int, slices []object.SlabSlice) (err error)
 		CompleteMultipartUpload(ctx context.Context, bucketName, path, uploadID string, parts []api.MultipartCompletedPart) (_ api.MultipartCompleteResponse, err error)
-		CreateMultipartUpload(ctx context.Context, bucketName, path string, ec object.EncryptionKey, mimeType string) (api.MultipartCreateResponse, error)
+		CreateMultipartUpload(ctx context.Context, bucketName, path string, ec object.EncryptionKey, mimeType string, metadata api.ObjectUserMetadata) (api.MultipartCreateResponse, error)
 		MultipartUpload(ctx context.Context, uploadID string) (resp api.MultipartUpload, _ error)
 		MultipartUploads(ctx context.Context, bucketName, prefix, keyMarker, uploadIDMarker string, maxUploads int) (resp api.MultipartListUploadsResponse, _ error)
 		MultipartUploadParts(ctx context.Context, bucketName, object string, uploadID string, marker int, limit int64) (resp api.MultipartListPartsResponse, _ error)
@@ -1248,22 +1248,21 @@ func (b *bus) objectEntriesHandlerGET(jc jape.Context, path string) {
 }
 
 func (b *bus) objectsHandlerPUT(jc jape.Context) {
-	var aor api.ObjectAddRequest
+	var aor api.AddObjectRequest
 	if jc.Decode(&aor) != nil {
 		return
 	} else if aor.Bucket == "" {
 		aor.Bucket = api.DefaultBucketName
 	}
-	jc.Check("couldn't store object", b.ms.UpdateObject(jc.Request.Context(), aor.Bucket, jc.PathParam("path"), aor.ContractSet, aor.ETag, aor.MimeType, aor.Object))
+	jc.Check("couldn't store object", b.ms.UpdateObject(jc.Request.Context(), aor.Bucket, jc.PathParam("path"), aor.ContractSet, aor.ETag, aor.MimeType, aor.Metadata, aor.Object))
 }
 
 func (b *bus) objectsCopyHandlerPOST(jc jape.Context) {
-	var orr api.ObjectsCopyRequest
+	var orr api.CopyObjectsRequest
 	if jc.Decode(&orr) != nil {
 		return
 	}
-
-	om, err := b.ms.CopyObject(jc.Request.Context(), orr.SourceBucket, orr.DestinationBucket, orr.SourcePath, orr.DestinationPath, orr.MimeType)
+	om, err := b.ms.CopyObject(jc.Request.Context(), orr.SourceBucket, orr.DestinationBucket, orr.SourcePath, orr.DestinationPath, orr.MimeType, orr.Metadata)
 	if jc.Check("couldn't copy object", err) != nil {
 		return
 	}
@@ -2176,7 +2175,7 @@ func (b *bus) multipartHandlerCreatePOST(jc jape.Context) {
 		key = object.NoOpKey
 	}
 
-	resp, err := b.ms.CreateMultipartUpload(jc.Request.Context(), req.Bucket, req.Path, key, req.MimeType)
+	resp, err := b.ms.CreateMultipartUpload(jc.Request.Context(), req.Bucket, req.Path, key, req.MimeType, req.Metadata)
 	if jc.Check("failed to create multipart upload", err) != nil {
 		return
 	}

bus/client/multipart-upload.go

@@ -50,6 +50,7 @@ func (c *Client) CreateMultipartUpload(ctx context.Context, bucket, path string,
 		Path:     path,
 		Key:      opts.Key,
 		MimeType: opts.MimeType,
+		Metadata: opts.Metadata,
 	}, &resp)
 	return
 }

bus/client/objects.go

@@ -12,26 +12,27 @@ import (
 // AddObject stores the provided object under the given path.
 func (c *Client) AddObject(ctx context.Context, bucket, path, contractSet string, o object.Object, opts api.AddObjectOptions) (err error) {
 	path = api.ObjectPathEscape(path)
-	err = c.c.WithContext(ctx).PUT(fmt.Sprintf("/objects/%s", path), api.ObjectAddRequest{
+	err = c.c.WithContext(ctx).PUT(fmt.Sprintf("/objects/%s", path), api.AddObjectRequest{
 		Bucket:      bucket,
 		ContractSet: contractSet,
 		Object:      o,
-		MimeType:    opts.MimeType,
 		ETag:        opts.ETag,
+		MimeType:    opts.MimeType,
+		Metadata:    opts.Metadata,
 	})
 	return
 }
 
 // CopyObject copies the object from the source bucket and path to the
 // destination bucket and path.
 func (c *Client) CopyObject(ctx context.Context, srcBucket, dstBucket, srcPath, dstPath string, opts api.CopyObjectOptions) (om api.ObjectMetadata, err error) {
-	err = c.c.WithContext(ctx).POST("/objects/copy", api.ObjectsCopyRequest{
+	err = c.c.WithContext(ctx).POST("/objects/copy", api.CopyObjectsRequest{
 		SourceBucket:      srcBucket,
 		DestinationBucket: dstBucket,
 		SourcePath:        srcPath,
 		DestinationPath:   dstPath,
-
-		MimeType: opts.MimeType,
+		MimeType:          opts.MimeType,
+		Metadata:          opts.Metadata,
 	}, &om)
 	return
 }

internal/node/node.go

@@ -124,6 +124,7 @@ func NewBus(cfg BusConfig, dir string, seed types.PrivateKey, l *zap.Logger) (ht
 		GormLogger:                    sqlLogger,
 		SlabPruningInterval:           cfg.SlabPruningInterval,
 		SlabPruningCooldown:           cfg.SlabPruningCooldown,
+		RetryTransactionIntervals:     []time.Duration{200 * time.Millisecond, 500 * time.Millisecond, time.Second, 3 * time.Second, 10 * time.Second, 10 * time.Second},
 	})
 	if err != nil {
 		return nil, nil, err