core: rewrite ufest decompress and unpack logic; introduce MaxChunkCount

* introduced `MaxChunkCount` const to limit number of chunks for both
  storing and loading paths.
* updated `packedChunkSize` to reflect worst-case size per chunk.
* refactored `unpack` to use SGL buffer, allowing dynamic growth up
  to the defined size limit `MaxChunkCount * packedChunkSize`.

Signed-off-by: Tony Chen <a122774007@gmail.com>

Author: Nahemah1022

ais/test/object_test.go

@@ -28,6 +28,7 @@ import (
 	"github.com/NVIDIA/aistore/core"
 	"github.com/NVIDIA/aistore/core/meta"
 	"github.com/NVIDIA/aistore/core/mock"
+	"github.com/NVIDIA/aistore/sys"
 	"github.com/NVIDIA/aistore/tools"
 	"github.com/NVIDIA/aistore/tools/docker"
 	"github.com/NVIDIA/aistore/tools/readers"
@@ -2173,6 +2174,125 @@ func TestMultipartUploadParallel(t *testing.T) {
 	tlog.Logfln("parallel multipart upload completed successfully with correct content ordering")
 }
 
+func TestMultipartMaxChunks(t *testing.T) {
+	tools.CheckSkip(t, &tools.SkipTestArgs{Long: true})
+
+	var (
+		proxyURL   = tools.RandomProxyURL(t)
+		baseParams = tools.BaseAPIParams(proxyURL)
+		bck        = cmn.Bck{
+			Name:     trand.String(10),
+			Provider: apc.AIS,
+		}
+		miniPartData = []byte("x") // Minimal 1-byte data per part
+	)
+
+	tools.CreateBucket(t, proxyURL, bck, nil, true /*cleanup*/)
+
+	t.Run("exceed-limit", func(t *testing.T) {
+		var (
+			objName  = "test-multipart-exceed-limit"
+			numParts = core.MaxChunkCount + 100 // Exceed limit by 100
+		)
+
+		uploadID, err := api.CreateMultipartUpload(baseParams, bck, objName)
+		tassert.CheckFatal(t, err)
+
+		err = uploadPartsInParallel(objName, uploadID, numParts, bck, miniPartData)
+
+		// We expect an error because we're exceeding MaxChunkCount
+		tassert.Fatalf(t, err != nil, "expected error when exceeding MaxChunkCount, but upload succeeded")
+		herr := cmn.UnwrapErrHTTP(err)
+		tassert.Fatalf(t, herr != nil, "expected ErrHTTP, got %v", err)
+		tassert.Fatalf(t, strings.Contains(herr.Message, "exceeds the maximum allowed"),
+			"expected error message to contain 'exceeds the maximum allowed', got: %v", err)
+
+		tlog.Logfln("multipart upload correctly rejected when exceeding MaxChunkCount (%d)", core.MaxChunkCount)
+
+		// Cleanup: abort the upload
+		_ = api.AbortMultipartUpload(baseParams, bck, objName, uploadID)
+	})
+
+	t.Run("equal-to-limit", func(t *testing.T) {
+		var (
+			objName  = "test-multipart-at-limit"
+			numParts = core.MaxChunkCount // Exactly at the limit
+		)
+
+		uploadID, err := api.CreateMultipartUpload(baseParams, bck, objName)
+		tassert.CheckFatal(t, err)
+		tassert.Fatalf(t, uploadID != "", "upload ID should not be empty")
+
+		// Upload parts in parallel - all should succeed
+		err = uploadPartsInParallel(objName, uploadID, numParts, bck, miniPartData)
+		tassert.CheckFatal(t, err)
+
+		// Complete multipart upload
+		partNumbers := make([]int, numParts)
+		for i := range numParts {
+			partNumbers[i] = i + 1
+		}
+		err = api.CompleteMultipartUpload(baseParams, bck, objName, uploadID, partNumbers)
+		tassert.CheckFatal(t, err)
+
+		tlog.Logfln("multipart upload completed successfully with %d parts at MaxChunkCount", numParts)
+
+		// Verify the uploaded object
+		hargs := api.HeadArgs{FltPresence: apc.FltPresent}
+		objAttrs, err := api.HeadObject(baseParams, bck, objName, hargs)
+		tassert.CheckFatal(t, err)
+
+		expectedSize := int64(numParts * len(miniPartData))
+		tassert.Fatalf(t, objAttrs.Size == expectedSize, "object size mismatch: expected %d, got %d", expectedSize, objAttrs.Size)
+
+		// GET the object and validate content
+		writer := bytes.NewBuffer(nil)
+		getArgs := api.GetArgs{Writer: writer}
+		_, err = api.GetObject(baseParams, bck, objName, &getArgs)
+		tassert.CheckFatal(t, err)
+
+		downloadedContent := writer.Bytes()
+		tassert.Errorf(t, len(downloadedContent) == numParts,
+			"content length mismatch: expected %d bytes, got %d", numParts, len(downloadedContent))
+
+		// Validate all bytes are 'x'
+		for i, b := range downloadedContent {
+			if b != 'x' {
+				t.Fatalf("byte at position %d is %q, expected 'x'", i, b)
+			}
+		}
+
+		tlog.Logfln("object content validated successfully: %d bytes, all 'x'", len(downloadedContent))
+	})
+}
+
+func uploadPartsInParallel(objName, uploadID string, numParts int, bck cmn.Bck, data []byte) error {
+	g := errgroup.Group{}
+	g.SetLimit(sys.MaxParallelism())
+
+	for partNum := 1; partNum <= numParts; partNum++ {
+		pn := partNum
+
+		g.Go(func() error {
+			putPartArgs := &api.PutPartArgs{
+				PutArgs: api.PutArgs{
+					BaseParams: baseParams,
+					Bck:        bck,
+					ObjName:    objName,
+					Reader:     readers.NewBytes(data),
+					Size:       uint64(len(data)),
+				},
+				UploadID:   uploadID,
+				PartNumber: pn,
+			}
+
+			return api.UploadPart(putPartArgs)
+		})
+	}
+
+	return g.Wait()
+}
+
 func TestMultipartUploadAbort(t *testing.T) {
 	var (
 		proxyURL   = tools.RandomProxyURL(t)

core/lchunk_test.go

@@ -207,10 +207,11 @@ var _ = Describe("Ufest Core Functionality", func() {
 			// Test chunk number exceeding uint16 limit
 			_, err := manifest.NewChunk(70000, lom) // > math.MaxUint16
 			Expect(err).To(HaveOccurred())
-			Expect(err.Error()).To(ContainSubstring("invalid chunk number"))
+			Expect(err.Error()).To(ContainSubstring("exceeds the maximum allowed"))
 
 			_, err = manifest.NewChunk(-1, lom)
 			Expect(err).To(HaveOccurred())
+			Expect(err.Error()).To(ContainSubstring("invalid chunk number"))
 		})
 
 		It("should handle concurrent chunk additions safely", func() {
@@ -534,6 +535,53 @@ var _ = Describe("Ufest Core Functionality", func() {
 				"expected ErrLmetaCorrupted or ErrBadCksum")
 		})
 
+		It("detects corrupted compressed data via checksum or decompression failure", func() {
+			testObject := "mpu/checksum-corrupt.bin"
+			localFQN := mix.MakePathFQN(&localBck, fs.ObjCT, testObject)
+			createTestFile(localFQN, 0)
+			lom := newBasicLom(localFQN)
+
+			u, err := core.NewUfest("cksum-test-"+cos.GenTie(), lom, false)
+			Expect(err).NotTo(HaveOccurred())
+
+			// Add multiple chunks to ensure we have enough data
+			for i := 1; i <= 3; i++ {
+				c, err := u.NewChunk(i, lom)
+				Expect(err).NotTo(HaveOccurred())
+				createTestChunk(c.Path(), 8*cos.KiB, nil)
+				Expect(u.Add(c, 8*cos.KiB, int64(i))).NotTo(HaveOccurred())
+			}
+			Expect(lom.CompleteUfest(u, false)).NotTo(HaveOccurred())
+
+			mfqn := lom.GenFQN(fs.ChunkMetaCT)
+			buf, err := os.ReadFile(mfqn)
+			Expect(err).NotTo(HaveOccurred())
+			Expect(len(buf)).To(BeNumerically(">", 16)) // Need enough data
+
+			// Corrupt the compressed data (not the trailing checksum)
+			// Trailing checksum is last 8 bytes, so corrupt byte before that
+			corruptIdx := len(buf) - 16
+			buf[corruptIdx] ^= 0xFF
+			Expect(os.WriteFile(mfqn, buf, 0o644)).NotTo(HaveOccurred())
+
+			// Reload and expect either decompression failure or checksum mismatch
+			loaded, err := core.NewUfest("", lom, true)
+			Expect(err).NotTo(HaveOccurred())
+			lom.Lock(false)
+			err = loaded.LoadCompleted(lom)
+			lom.Unlock(false)
+
+			// Corrupting compressed data can fail in two ways:
+			// 1. LZ4 decompression fails (ErrLmetaCorrupted) - most common
+			// 2. Checksum validation fails (ErrBadCksum) - if corruption doesn't break LZ4
+			var (
+				lmerr *cmn.ErrLmetaCorrupted
+				bcerr *cos.ErrBadCksum
+			)
+			Expect(errors.As(err, &lmerr) || errors.As(err, &bcerr)).To(BeTrue(),
+				"expected ErrLmetaCorrupted or ErrBadCksum for corrupted compressed data")
+		})
+
 	})
 
 	Describe("Locking Behavior", func() {
@@ -593,6 +641,50 @@ var _ = Describe("Ufest Core Functionality", func() {
 			// Test exceeding uint16 limit
 			_, err := manifest.NewChunk(70000, manifest.Lom()) // > math.MaxUint16 (65535)
 			Expect(err).To(HaveOccurred())
+			Expect(err.Error()).To(ContainSubstring("exceeds the maximum allowed"))
+		})
+	})
+
+	Describe("MaxChunkCount Limit", func() {
+		var (
+			manifest *core.Ufest
+			lom      *core.LOM
+		)
+
+		BeforeEach(func() {
+			var err error
+			testObjectName := "test-objects/maxchunks-test.bin"
+			testFileSize := int64(1 * cos.TiB)
+			localFQN := mix.MakePathFQN(&localBck, fs.ObjCT, testObjectName)
+			lom = newBasicLom(localFQN, testFileSize)
+			manifest, err = core.NewUfest("test-maxchunks-"+trand.String(8), lom, false)
+			Expect(err).NotTo(HaveOccurred())
+		})
+
+		It("should reject chunk number exceeding MaxChunkCount", func() {
+			invalidChunkNum := core.MaxChunkCount + 1
+			_, err := manifest.NewChunk(invalidChunkNum, lom)
+			Expect(err).To(HaveOccurred())
+			Expect(err.Error()).To(ContainSubstring("exceeds the maximum allowed"))
+		})
+
+		It("should accept chunk number at MaxChunkCount", func() {
+			validChunkNum := core.MaxChunkCount
+			chunk, err := manifest.NewChunk(validChunkNum, lom)
+			Expect(err).NotTo(HaveOccurred())
+			Expect(chunk).NotTo(BeNil())
+			Expect(chunk.Num()).To(Equal(uint16(validChunkNum)))
+		})
+
+		It("should reject chunk number at zero", func() {
+			_, err := manifest.NewChunk(0, lom)
+			Expect(err).To(HaveOccurred())
+			Expect(err.Error()).To(ContainSubstring("invalid chunk number"))
+		})
+
+		It("should reject negative chunk number", func() {
+			_, err := manifest.NewChunk(-1, lom)
+			Expect(err).To(HaveOccurred())
 			Expect(err.Error()).To(ContainSubstring("invalid chunk number"))
 		})
 	})