Skip to content

Commit 10a9f5d

Browse files
authored
feat: add object attributes multipart parts
1 parent 9994997 commit 10a9f5d

15 files changed

Lines changed: 577 additions & 69 deletions

README.md

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -43,7 +43,7 @@ SynapS3 is an open-source, self-hosted S3-compatible gateway for storing objects
4343
| Object | `ListObjects` || Marker pagination |
4444
| Object | `ListObjectsV2` || Continuation-token pagination |
4545
| Object | `ListObjectVersions` || Lists object versions and delete markers |
46-
| Object | `GetObjectAttributes` || Reports ETag, checksum, size, and storage class |
46+
| Object | `GetObjectAttributes` || Reports metadata and multipart `ObjectParts`; `TotalPartsCount` is not emitted |
4747
| Multipart | `CreateMultipartUpload` || Starts an upload |
4848
| Multipart | `UploadPart` || Uploads one part |
4949
| Multipart | `UploadPartCopy` | ⚠️ | Whole-object copy only; range copy is not supported |

docs/en/reference/s3-compatibility.md

Lines changed: 2 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -33,7 +33,7 @@ SynapS3 focuses on path-style S3 bucket and object workflows needed for object s
3333
| Object | `ListObjects` | Supported | Marker pagination. |
3434
| Object | `ListObjectsV2` | Supported | Continuation-token pagination. |
3535
| Object | `ListObjectVersions` | Supported | Lists object versions and delete markers. |
36-
| Object | `GetObjectAttributes` | Supported | Reports ETag, checksum, size, and storage class. |
36+
| Object | `GetObjectAttributes` | Supported | Reports metadata and multipart `ObjectParts`; `TotalPartsCount` is not emitted. |
3737
| Multipart | `CreateMultipartUpload` | Supported | Starts an upload. |
3838
| Multipart | `UploadPart` | Supported | Uploads one part. |
3939
| Multipart | `UploadPartCopy` | Partial | Whole-object copy only; range copy is not supported. |
@@ -50,6 +50,7 @@ Buckets behave as versioning-enabled. A normal object delete creates a delete ma
5050

5151
- Bucket deletion.
5252
- Suspending bucket versioning.
53+
- `TotalPartsCount` in `GetObjectAttributes.ObjectParts`; the current VersityGW response type has no field for it.
5354
- Multipart range copy for `UploadPartCopy`.
5455
- Distributed coordination across multiple SynapS3 nodes.
5556

docs/zh/reference/s3-compatibility.md

Lines changed: 2 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -33,7 +33,7 @@ SynapS3 聚焦在把对象存储到 Filecoin 所需的 path-style S3 bucket 和
3333
| Object | `ListObjects` | 支持 | Marker 分页。 |
3434
| Object | `ListObjectsV2` | 支持 | Continuation-token 分页。 |
3535
| Object | `ListObjectVersions` | 支持 | 列出对象版本和 delete markers。 |
36-
| Object | `GetObjectAttributes` | 支持 | 返回 ETag、checksum、sizestorage class|
36+
| Object | `GetObjectAttributes` | 支持 | 返回 metadatamultipart `ObjectParts`;不返回 `TotalPartsCount`|
3737
| Multipart | `CreateMultipartUpload` | 支持 | 开始上传。 |
3838
| Multipart | `UploadPart` | 支持 | 上传单个 part。 |
3939
| Multipart | `UploadPartCopy` | 部分支持 | 仅支持整对象复制,不支持 range copy。 |
@@ -50,6 +50,7 @@ Bucket 表现为 versioning-enabled。普通 object delete 会创建 delete mark
5050

5151
- Bucket deletion。
5252
- 暂停 bucket versioning。
53+
- `GetObjectAttributes.ObjectParts` 中的 `TotalPartsCount`;当前 VersityGW response type 没有该字段。
5354
- `UploadPartCopy` 的 multipart range copy。
5455
- 多个 SynapS3 节点之间的分布式协调。
5556

internal/backend/multipart.go

Lines changed: 21 additions & 14 deletions
Original file line numberDiff line numberDiff line change
@@ -86,6 +86,7 @@ func (b *SynapseBackend) UploadPart(ctx context.Context, input *s3.UploadPartInp
8686
PartNumber: partNum,
8787
Size: cacheInfo.Size,
8888
ETag: cacheInfo.ETag,
89+
Checksum: &cacheInfo.Checksum,
8990
}
9091
if err := b.repos.Multiparts.CreatePart(ctx, part); err != nil {
9192
return nil, fmt.Errorf("recording part: %w", err)
@@ -147,6 +148,7 @@ func (b *SynapseBackend) UploadPartCopy(ctx context.Context, input *s3.UploadPar
147148
PartNumber: partNum,
148149
Size: cacheInfo.Size,
149150
ETag: cacheInfo.ETag,
151+
Checksum: &cacheInfo.Checksum,
150152
}
151153
if err := b.repos.Multiparts.CreatePart(ctx, part); err != nil {
152154
return s3response.CopyPartResult{}, fmt.Errorf("recording copied part: %w", err)
@@ -272,18 +274,19 @@ func (b *SynapseBackend) CompleteMultipartUpload(ctx context.Context, input *s3.
272274
}
273275

274276
version := &model.ObjectVersion{
275-
VersionID: versionID,
276-
BucketID: upload.BucketID,
277-
Key: keyName,
278-
Size: cacheInfo.Size,
279-
ETag: s3ETag,
280-
Checksum: cacheInfo.Checksum,
281-
ContentType: upload.ContentType,
282-
Metadata: upload.Metadata,
283-
CacheKey: cacheKey,
284-
StorageUploadID: reuse.StorageUploadID,
285-
InCache: true,
286-
State: reuse.State,
277+
VersionID: versionID,
278+
BucketID: upload.BucketID,
279+
Key: keyName,
280+
Size: cacheInfo.Size,
281+
ETag: s3ETag,
282+
Checksum: cacheInfo.Checksum,
283+
ContentType: upload.ContentType,
284+
Metadata: upload.Metadata,
285+
CacheKey: cacheKey,
286+
MultipartUploadID: &uploadID,
287+
StorageUploadID: reuse.StorageUploadID,
288+
InCache: true,
289+
State: reuse.State,
287290
}
288291
createdState = version.State
289292

@@ -431,9 +434,13 @@ func (b *SynapseBackend) ListParts(ctx context.Context, input *s3.ListPartsInput
431434
}
432435

433436
if len(parts) > maxParts {
434-
parts = parts[:maxParts]
437+
if maxParts > 0 {
438+
parts = parts[:maxParts]
439+
result.NextPartNumberMarker = parts[len(parts)-1].PartNumber
440+
} else {
441+
parts = nil
442+
}
435443
result.IsTruncated = true
436-
result.NextPartNumberMarker = parts[len(parts)-1].PartNumber
437444
}
438445

439446
for _, p := range parts {

internal/backend/multipart_test.go

Lines changed: 38 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -65,12 +65,13 @@ func TestUploadPart_HappyPath(t *testing.T) {
6565
}
6666

6767
partNum := int32(1)
68+
partBody := "part-1-data"
6869
partOut, err := tb.backend.UploadPart(ctx, &s3.UploadPartInput{
6970
Bucket: aws.String("up-bucket"),
7071
Key: aws.String("parts.bin"),
7172
UploadId: aws.String(initResult.UploadId),
7273
PartNumber: &partNum,
73-
Body: strings.NewReader("part-1-data"),
74+
Body: strings.NewReader(partBody),
7475
})
7576
if err != nil {
7677
t.Fatalf("UploadPart: %v", err)
@@ -90,6 +91,9 @@ func TestUploadPart_HappyPath(t *testing.T) {
9091
if len(parts) > 0 && parts[0].PartNumber != 1 {
9192
t.Errorf("part number = %d, want 1", parts[0].PartNumber)
9293
}
94+
if len(parts) > 0 && (parts[0].Checksum == nil || *parts[0].Checksum != testSHA256Hex(partBody)) {
95+
t.Fatalf("part checksum = %v, want %s", parts[0].Checksum, testSHA256Hex(partBody))
96+
}
9397
}
9498

9599
func TestUploadPartCopy_CopySourceVersionIDCopiesSpecifiedVersion(t *testing.T) {
@@ -122,6 +126,16 @@ func TestUploadPartCopy_CopySourceVersionIDCopiesSpecifiedVersion(t *testing.T)
122126
if partOut.CopySourceVersionId != firstOut.VersionID {
123127
t.Fatalf("CopySourceVersionId = %q, want %s", partOut.CopySourceVersionId, firstOut.VersionID)
124128
}
129+
parts, err := tb.repos.Multiparts.GetParts(ctx, initResult.UploadId, 0, 100)
130+
if err != nil {
131+
t.Fatalf("GetParts after UploadPartCopy: %v", err)
132+
}
133+
if len(parts) != 1 {
134+
t.Fatalf("copied parts = %d, want 1", len(parts))
135+
}
136+
if parts[0].Checksum == nil || *parts[0].Checksum != testSHA256Hex(validTestObjectBody("old")) {
137+
t.Fatalf("copied part checksum = %v, want %s", parts[0].Checksum, testSHA256Hex(validTestObjectBody("old")))
138+
}
125139

126140
_, versionID, err := tb.backend.CompleteMultipartUpload(ctx, &s3.CompleteMultipartUploadInput{
127141
Bucket: aws.String("up-copy-version-bucket"),
@@ -228,6 +242,9 @@ func TestCompleteMultipartUpload_HappyPath(t *testing.T) {
228242
if obj.State != model.ObjectStateCached {
229243
t.Errorf("object state = %q, want %q", obj.State, model.ObjectStateCached)
230244
}
245+
if obj.MultipartUploadID == nil || *obj.MultipartUploadID != uploadID {
246+
t.Fatalf("object multipart_upload_id = %v, want %s", obj.MultipartUploadID, uploadID)
247+
}
231248
task, err := tb.repos.Tasks.ClaimReady(ctx, model.TaskTypeUpload, time.Minute)
232249
if err != nil {
233250
t.Fatalf("ClaimReady: %v", err)
@@ -545,4 +562,24 @@ func TestListParts_HappyPath(t *testing.T) {
545562
t.Errorf("part[%d].PartNumber = %d, want %d", i, p.PartNumber, i+1)
546563
}
547564
}
565+
566+
maxParts := int32(0)
567+
zeroMaxResult, err := tb.backend.ListParts(ctx, &s3.ListPartsInput{
568+
Bucket: aws.String("lp-bucket"),
569+
Key: aws.String("parts-file.bin"),
570+
UploadId: aws.String(initResult.UploadId),
571+
MaxParts: &maxParts,
572+
})
573+
if err != nil {
574+
t.Fatalf("ListParts MaxParts=0: %v", err)
575+
}
576+
if len(zeroMaxResult.Parts) != 0 {
577+
t.Fatalf("ListParts MaxParts=0 parts count = %d, want 0", len(zeroMaxResult.Parts))
578+
}
579+
if !zeroMaxResult.IsTruncated {
580+
t.Fatal("ListParts MaxParts=0 IsTruncated = false, want true")
581+
}
582+
if zeroMaxResult.NextPartNumberMarker != 0 {
583+
t.Fatalf("ListParts MaxParts=0 NextPartNumberMarker = %d, want 0", zeroMaxResult.NextPartNumberMarker)
584+
}
548585
}

internal/backend/object.go

Lines changed: 98 additions & 14 deletions
Original file line numberDiff line numberDiff line change
@@ -8,6 +8,7 @@ import (
88
"net/http"
99
"net/url"
1010
"path"
11+
"strconv"
1112
"strings"
1213
"time"
1314

@@ -230,14 +231,22 @@ func (b *SynapseBackend) GetObjectAttributes(ctx context.Context, input *s3.GetO
230231
}
231232

232233
checksum := types.Checksum{ChecksumSHA256: &meta.Checksum}
233-
return s3response.GetObjectAttributesResponse{
234+
resp := s3response.GetObjectAttributesResponse{
234235
ETag: &meta.QuotedETag,
235236
ObjectSize: &meta.Size,
236237
StorageClass: types.StorageClassStandard,
237238
Checksum: &checksum,
238239
VersionId: &meta.VersionID,
239240
LastModified: &meta.LastModified,
240-
}, nil
241+
}
242+
if objectPartsRequested(input.ObjectAttributes) {
243+
objectParts, err := b.getObjectAttributeParts(ctx, meta.MultipartUploadID, input)
244+
if err != nil {
245+
return s3response.GetObjectAttributesResponse{}, err
246+
}
247+
resp.ObjectParts = objectParts
248+
}
249+
return resp, nil
241250
}
242251

243252
func (b *SynapseBackend) ListObjects(ctx context.Context, input *s3.ListObjectsInput) (s3response.ListObjectsResult, error) {
@@ -801,12 +810,13 @@ func derefStr(s *string) string {
801810
}
802811

803812
type objectMetadataResult struct {
804-
Size int64
805-
QuotedETag string
806-
Checksum string
807-
ContentType string
808-
VersionID string
809-
LastModified time.Time
813+
Size int64
814+
QuotedETag string
815+
Checksum string
816+
ContentType string
817+
VersionID string
818+
MultipartUploadID *string
819+
LastModified time.Time
810820
}
811821

812822
func (b *SynapseBackend) objectMetadata(ctx context.Context, bucketID int64, key, versionID string) (objectMetadataResult, error) {
@@ -816,15 +826,89 @@ func (b *SynapseBackend) objectMetadata(ctx context.Context, bucketID int64, key
816826
}
817827
etag := fmt.Sprintf(`"%s"`, version.ETag)
818828
return objectMetadataResult{
819-
Size: version.Size,
820-
QuotedETag: etag,
821-
Checksum: version.Checksum,
822-
ContentType: version.ContentType,
823-
VersionID: version.VersionID,
824-
LastModified: version.CreatedAt,
829+
Size: version.Size,
830+
QuotedETag: etag,
831+
Checksum: version.Checksum,
832+
ContentType: version.ContentType,
833+
VersionID: version.VersionID,
834+
MultipartUploadID: version.MultipartUploadID,
835+
LastModified: version.CreatedAt,
825836
}, nil
826837
}
827838

839+
func objectPartsRequested(attrs []types.ObjectAttributes) bool {
840+
if len(attrs) == 0 {
841+
return true
842+
}
843+
for _, attr := range attrs {
844+
if attr == types.ObjectAttributesObjectParts {
845+
return true
846+
}
847+
}
848+
return false
849+
}
850+
851+
func (b *SynapseBackend) getObjectAttributeParts(ctx context.Context, uploadID *string, input *s3.GetObjectAttributesInput) (*s3response.ObjectParts, error) {
852+
maxParts := 1000
853+
if input.MaxParts != nil {
854+
maxParts = int(*input.MaxParts)
855+
}
856+
857+
partMarker := 0
858+
if input.PartNumberMarker != nil {
859+
if v := *input.PartNumberMarker; v != "" {
860+
parsed, err := strconv.Atoi(v)
861+
if err != nil {
862+
return nil, s3err.GetInvalidMaxLimiterErr("part-number-marker")
863+
}
864+
if parsed < 0 {
865+
return nil, s3err.GetNegativeMaxLimiterErr("part-number-marker")
866+
}
867+
partMarker = parsed
868+
}
869+
}
870+
871+
result := &s3response.ObjectParts{
872+
Parts: []types.ObjectPart{},
873+
MaxParts: maxParts,
874+
PartNumberMarker: partMarker,
875+
}
876+
if uploadID == nil || *uploadID == "" {
877+
return result, nil
878+
}
879+
880+
parts, err := b.repos.Multiparts.GetParts(ctx, *uploadID, partMarker, maxParts+1)
881+
if err != nil {
882+
return nil, fmt.Errorf("listing object attribute parts: %w", err)
883+
}
884+
if len(parts) > maxParts {
885+
if maxParts > 0 {
886+
parts = parts[:maxParts]
887+
result.NextPartNumberMarker = parts[len(parts)-1].PartNumber
888+
} else {
889+
parts = nil
890+
}
891+
result.IsTruncated = true
892+
}
893+
894+
result.Parts = make([]types.ObjectPart, 0, len(parts))
895+
for _, p := range parts {
896+
partNumber := int32(p.PartNumber)
897+
size := p.Size
898+
part := types.ObjectPart{
899+
PartNumber: &partNumber,
900+
Size: &size,
901+
}
902+
if p.Checksum != nil && *p.Checksum != "" {
903+
checksum := *p.Checksum
904+
part.ChecksumSHA256 = &checksum
905+
}
906+
result.Parts = append(result.Parts, part)
907+
}
908+
909+
return result, nil
910+
}
911+
828912
func (b *SynapseBackend) versionForRead(ctx context.Context, bucketID int64, key, versionID string) (*model.ObjectVersion, error) {
829913
if versionID != "" {
830914
version, err := b.repos.Objects.GetVersionByBucketKeyAndID(ctx, bucketID, key, versionID)

0 commit comments

Comments
 (0)