s3 multipart; chunk manifest

* s3 multipart: disallow (with 501) partial completion
  - but allow unordered [1..count] parts with duplications
* split `validateChecksumEtag()` out of complete-mpt
  - still compute _whole_ checksum and ETag on the entire manifest
* hide manifest's chunks
* with refactoring

Signed-off-by: Alex Aizman <alex.aizman@gmail.com>

Author: alex-aizman

ais/s3/mpt.go

@@ -5,6 +5,9 @@
 package s3
 
 import (
+	"errors"
+	"fmt"
+	"net/http"
 	"sort"
 
 	"github.com/NVIDIA/aistore/api/apc"
@@ -67,9 +70,9 @@ func ListUploads(all []*core.Ufest, bckName, idMarker string, maxUploads int) *L
 
 func ListParts(manifest *core.Ufest) (parts []types.CompletedPart, ecode int, err error) {
 	manifest.Lock()
-	parts = make([]types.CompletedPart, 0, len(manifest.Chunks))
-	for i := range manifest.Chunks {
-		c := &manifest.Chunks[i]
+	parts = make([]types.CompletedPart, 0, manifest.Count())
+	for i := range manifest.Count() {
+		c := manifest.GetChunk(i+1, true)
 		etag := c.ETag
 		if etag == "" {
 			etag = cmn.MD5ToETag(c.MD5)
@@ -82,3 +85,56 @@ func ListParts(manifest *core.Ufest) (parts []types.CompletedPart, ecode int, er
 	manifest.Unlock()
 	return parts, 0, nil
 }
+
+// validate that the caller requests completion of
+// exactly all uploaded parts: the set {1..count} in ANY order
+// on success, normalize req.Parts in-place
+// on error return:
+// - 501 when partial completion is requested (len != count)
+// - 400 for malformed input (nil part number, out of range, duplicates)
+func EnforceCompleteAllParts(req *CompleteMptUpload, count int) (int, error) {
+	if req == nil {
+		return http.StatusBadRequest, errors.New("nil parts list")
+	}
+	if len(req.Parts) != count {
+		return http.StatusNotImplemented,
+			fmt.Errorf("partial completion is not allowed: requested %d parts, have %d",
+				len(req.Parts), count)
+	}
+	// fast path
+	for i := range count {
+		p := req.Parts[i]
+		if p.PartNumber == nil {
+			return http.StatusBadRequest, fmt.Errorf("nil part number at index %d", i)
+		}
+		if *p.PartNumber != int32(i+1) {
+			goto slow
+		}
+	}
+	return 0, nil
+
+slow:
+	sort.Slice(req.Parts, func(i, j int) bool {
+		pi, pj := req.Parts[i], req.Parts[j]
+		// nil can't occur here due to the fast-path scan, but be defensive
+		if pi.PartNumber == nil {
+			return false
+		}
+		if pj.PartNumber == nil {
+			return true
+		}
+		return *pi.PartNumber < *pj.PartNumber
+	})
+	for i := range count {
+		p := req.Parts[i]
+		if p.PartNumber == nil {
+			return http.StatusBadRequest, fmt.Errorf("nil part number after sort at index %d", i)
+		}
+		got := *p.PartNumber
+		if got != int32(i+1) {
+			return http.StatusBadRequest, fmt.Errorf("parts must be exactly 1..%d: got %d at position %d",
+				count, got, i)
+		}
+	}
+	return 0, nil
+}

ais/tgtmpt.go

@@ -47,29 +47,41 @@ func (ups *ups) init(id string, lom *core.LOM, rmd map[string]string) error {
 	if ups.m == nil {
 		ups.m = make(map[string]up, iniCapUploads)
 	}
-	ups._add(id, manifest, rmd)
+	err = ups._add(id, manifest, rmd)
 	ups.Unlock()
-	return nil
+	return err
 }
 
-func (ups *ups) _add(id string, manifest *core.Ufest, rmd map[string]string) {
+func (ups *ups) _add(id string, manifest *core.Ufest, rmd map[string]string) (err error) {
 	debug.Assert(manifest.Lom() != nil)
-	_, ok := ups.m[id]
-	debug.Assert(!ok, "duplicated upload ID: ", id, " ", manifest.Lom().Cname())
+	if _, ok := ups.m[id]; ok {
+		err = fmt.Errorf("duplicated upload ID: %q", id)
+		debug.AssertNoErr(err)
+		return
+	}
 	ups.m[id] = up{manifest, rmd}
+	return
 }
 
 func (ups *ups) get(id string) (manifest *core.Ufest) {
 	ups.RLock()
-	manifest = ups.m[id].u
+	up, ok := ups.m[id]
+	if ok {
+		manifest = up.u
+		debug.Assert(id == manifest.ID())
+	}
 	ups.RUnlock()
 	return
 }
 
 func (ups *ups) getWithMeta(id string) (manifest *core.Ufest, rmd map[string]string) {
 	ups.RLock()
-	manifest = ups.m[id].u
-	rmd = ups.m[id].rmd
+	up, ok := ups.m[id]
+	if ok {
+		manifest = up.u
+		rmd = up.rmd
+		debug.Assert(id == manifest.ID())
+	}
 	ups.RUnlock()
 	return
 }

ais/tgts3mpt.go

@@ -23,7 +23,6 @@ import (
 	"net/http"
 	"net/url"
 	"os"
-	"sort"
 	"strconv"
 
 	"github.com/NVIDIA/aistore/ais/s3"
@@ -134,7 +133,6 @@ func (t *target) putMptPart(w http.ResponseWriter, r *http.Request, items []stri
 	var (
 		etag         string
 		expectedSize int64
-		ecode        int
 		partSHA      = r.Header.Get(cos.S3HdrContentSHA256)
 		checkPartSHA = partSHA != "" && partSHA != cos.S3UnsignedPayload
 		cksumSHA     *cos.CksumHash
@@ -150,15 +148,17 @@ func (t *target) putMptPart(w http.ResponseWriter, r *http.Request, items []stri
 
 	if r.ContentLength <= 0 {
 		err := fmt.Errorf("put-part invalid size (%d)", r.ContentLength)
-		s3.WriteMptErr(w, r, err, ecode, lom, uploadID)
+		s3.WriteMptErr(w, r, err, 0, lom, uploadID)
 		return
 	}
 
 	// 3. write
 	// for remote buckets, use SGL buffering when memory is available;
 	// fall back to TeeReader to avoid high memory usage under pressure
-	mw := cos.IniWriterMulti(cksumMD5.H, cksumSHA.H, partFh)
-
+	var (
+		ecode int
+		mw    = cos.IniWriterMulti(cksumMD5.H, cksumSHA.H, partFh)
+	)
 	switch {
 	case !remote:
 		buf, slab := t.gmm.AllocSize(r.ContentLength)
@@ -183,7 +183,6 @@ func (t *target) putMptPart(w http.ResponseWriter, r *http.Request, items []stri
 		etag, ecode, err = t.ups.putPartRemote(lom, tr, r, q, uploadID, expectedSize, partNum)
 		remotePutLatency = mono.SinceNano(remoteStart)
 	}
-
 	cos.Close(partFh)
 	if err != nil {
 		if nerr := cos.RemoveFile(chunkPath); nerr != nil && !cos.IsNotExist(nerr) {
@@ -196,7 +195,7 @@ func (t *target) putMptPart(w http.ResponseWriter, r *http.Request, items []stri
 	size := mw.Size()
 	if size != expectedSize {
 		err := fmt.Errorf("part %d size mismatch (%d vs %d)", partNum, size, expectedSize)
-		s3.WriteMptErr(w, r, err, ecode, lom, uploadID)
+		s3.WriteMptErr(w, r, err, 0, lom, uploadID)
 		return
 	}
 
@@ -281,100 +280,36 @@ func (t *target) completeMpt(w http.ResponseWriter, r *http.Request, items []str
 		s3.WriteErr(w, r, err, http.StatusBadRequest)
 		return
 	}
-	if len(partList.Parts) == 0 {
-		s3.WriteErr(w, r, fmt.Errorf("upload %q: empty list of upload parts", uploadID), http.StatusBadRequest)
+	if partList == nil || len(partList.Parts) == 0 {
+		s3.WriteErr(w, r, errors.New("no parts"), http.StatusBadRequest)
 		return
 	}
+
 	objName := s3.ObjName(items)
 	lom := &core.LOM{ObjName: objName} // TODO: use core.AllocLOM()
 	if err := lom.InitBck(bck.Bucket()); err != nil {
 		s3.WriteErr(w, r, err, 0)
 		return
 	}
 
-	// TODO: compare with listMptUploads (that does fromFS)
+	// (compare with listMptUploads)
 	manifest, metadata := t.ups.getWithMeta(uploadID)
 	if manifest == nil {
 		s3.WriteMptErr(w, r, s3.NewErrNoSuchUpload(uploadID, nil), http.StatusNotFound, lom, uploadID)
 		return
 	}
 
-	// validate parts (note: maybe too strict)
-	var (
-		numParts = len(partList.Parts)
-	)
-	if numParts == 0 {
-		s3.WriteMptErr(w, r, errors.New("empty parts list"), 0, lom, uploadID)
-		return
-	}
-	sort.Slice(partList.Parts, func(i, j int) bool {
-		return *partList.Parts[i].PartNumber < *partList.Parts[j].PartNumber
-	})
-	for i, p := range partList.Parts {
-		if p.PartNumber == nil {
-			s3.WriteMptErr(w, r, fmt.Errorf("nil part number at index %d", i), http.StatusBadRequest, lom, uploadID)
-			return
-		}
-		want := int32(i + 1)
-		got := *p.PartNumber
-		if got != want {
-			s3.WriteMptErr(w, r,
-				fmt.Errorf("parts must be strictly ascending and contiguous (1..N): got %d at position %d", got, i),
-				http.StatusBadRequest, lom, uploadID)
-			return
-		}
-	}
-
-	manifest.Lock() // ------------------------------------------------ TODO -- FIXME: consider CoW
-	if len(manifest.Chunks) < numParts {
-		manifest.Unlock()
-		s3.WriteMptErr(w, r,
-			fmt.Errorf("have %d parts, requested %d", len(manifest.Chunks), numParts),
-			http.StatusBadRequest, lom, uploadID)
-		return
-	}
-	nparts := make([]*core.Uchunk, numParts)
-	for i := range numParts {
-		c := &manifest.Chunks[i]
-		if c.Num() != uint16(i+1) {
-			manifest.Unlock()
-			s3.WriteMptErr(w, r, fmt.Errorf("missing or out-of-order part %d (found %d)", i+1, c.Num()), 0, lom, uploadID)
-			return
-		}
-		nparts[i] = c
-	}
-
-	// compute "whole" checksum // TODO -- FIXME: sha256 may take precedence when implied by partSHA (see above)
-	var (
-		wholeCksum *cos.CksumHash
-		remote     = bck.IsRemoteS3() || bck.IsRemoteOCI()
-	)
-	if remote && lom.CksumConf().Type != cos.ChecksumNone {
-		wholeCksum = cos.NewCksumHash(lom.CksumConf().Type)
-	} else {
-		wholeCksum = cos.NewCksumHash(cos.ChecksumMD5)
-	}
-
-	if wholeCksum != nil {
-		if err := manifest.ComputeWholeChecksum(wholeCksum); err != nil {
-			s3.WriteMptErr(w, r, err, 0, lom, uploadID)
-			return
-		}
-		lom.SetCksum(&wholeCksum.Cksum)
-	}
-
-	// compute multipart-compliant ETag if need be
+	// validate/enforce parts, compute _whole" checksum and etag
 	var etag string
-	if !remote {
-		var err error
-		if etag, err = manifest.ETagS3(); err != nil {
-			s3.WriteMptErr(w, r, err, 0, lom, uploadID)
-			return
-		}
+	manifest.Lock()
+	etag, err = validateChecksumEtag(w, r, lom, manifest, partList)
+	manifest.Unlock()
+	if err != nil {
+		return
 	}
-	manifest.Unlock() // -------------------------------------------------- TODO -- FIXME: ditto
 
 	// call remote
+	remote := bck.IsRemoteS3() || bck.IsRemoteOCI()
 	if remote {
 		var err error
 		etag, err = t.ups.completeRemote(w, r, lom, q, uploadID, body, partList)
@@ -422,6 +357,50 @@ func (t *target) completeMpt(w http.ResponseWriter, r *http.Request, items []str
 	}
 }
 
+// (under manifest lock)
+func validateChecksumEtag(w http.ResponseWriter, r *http.Request, lom *core.LOM, manifest *core.Ufest, partList *s3.CompleteMptUpload) (string, error) {
+	uploadID := manifest.ID()
+	if err := manifest.Check(); err != nil {
+		s3.WriteMptErr(w, r, err, 0, lom, uploadID)
+		return "", err
+	}
+	if ecode, err := s3.EnforceCompleteAllParts(partList, manifest.Count()); err != nil {
+		s3.WriteMptErr(w, r, err, ecode, lom, uploadID)
+		return "", err
+	}
+
+	// compute "whole" checksum (TODO: sha256 may take precedence when implied by `partSHA`)
+	var (
+		wholeCksum *cos.CksumHash
+		bck        = lom.Bck()
+		remote     = bck.IsRemoteS3() || bck.IsRemoteOCI()
+	)
+	if remote && lom.CksumConf().Type != cos.ChecksumNone {
+		wholeCksum = cos.NewCksumHash(lom.CksumConf().Type)
+	} else {
+		wholeCksum = cos.NewCksumHash(cos.ChecksumMD5)
+	}
+	if wholeCksum != nil {
+		if err := manifest.ComputeWholeChecksum(wholeCksum); err != nil {
+			s3.WriteMptErr(w, r, err, 0, lom, uploadID)
+			return "", err
+		}
+		lom.SetCksum(&wholeCksum.Cksum)
+	}
+
+	// compute multipart-compliant ETag if need be
+	var etag string
+	if !remote {
+		var err error
+		if etag, err = manifest.ETagS3(); err != nil {
+			s3.WriteMptErr(w, r, err, 0, lom, uploadID)
+			return "", err
+		}
+	}
+
+	return etag, nil
+}
+
 // Abort an active multipart upload.
 // Body is empty, only URL query contains uploadID
 // 1. uploadID must exists
@@ -548,17 +527,17 @@ func (t *target) getMptPart(w http.ResponseWriter, r *http.Request, bck *meta.Bc
 		return
 	}
 
-	// Find the specific chunk
+	// get specific chunk
 	manifest.Lock()
-	chunk := manifest.GetChunk(uint16(partNum), true)
+	chunk := manifest.GetChunk(int(partNum), true)
 	manifest.Unlock()
 	if chunk == nil {
 		err := fmt.Errorf("part %d not found", partNum)
 		s3.WriteErr(w, r, err, http.StatusNotFound)
 		return
 	}
 
-	// Read chunk file directly
+	// read chunk file
 	fh, err := os.Open(chunk.Path)
 	if err != nil {
 		s3.WriteErr(w, r, err, 0)

core/lchunk_test.go

@@ -148,7 +148,7 @@ var _ = Describe("Ufest Core Functionality", func() {
 
 			// Verify all chunks are accessible
 			for _, c := range chunks {
-				retrievedChunk := manifest.GetChunk(uint16(c.num), false)
+				retrievedChunk := manifest.GetChunk(int(c.num), false)
 				Expect(retrievedChunk).NotTo(BeNil())
 				Expect(retrievedChunk.Path).To(Equal(c.path))
 				Expect(retrievedChunk.Size()).To(Equal(c.size))
@@ -234,7 +234,7 @@ var _ = Describe("Ufest Core Functionality", func() {
 
 			// Verify all chunks were added
 			for i := 1; i <= numWorkers; i++ {
-				chunk := manifest.GetChunk(uint16(i), false)
+				chunk := manifest.GetChunk(i, false)
 				Expect(chunk).NotTo(BeNil())
 				Expect(chunk.Num()).To(Equal(uint16(i)))
 			}
@@ -476,8 +476,8 @@ var _ = Describe("Ufest Core Functionality", func() {
 
 			// Verify chunks
 			for i := 1; i <= 3; i++ {
-				originalChunk := manifest.GetChunk(uint16(i), false)
-				loadedChunk := loadedManifest.GetChunk(uint16(i), false)
+				originalChunk := manifest.GetChunk(i, false)
+				loadedChunk := loadedManifest.GetChunk(i, false)
 
 				Expect(loadedChunk).NotTo(BeNil())
 				Expect(loadedChunk.Num()).To(Equal(originalChunk.Num()))