when checksum is 'none' (refactoring)

* for fear of compiler-optimized nil deref
* none-hash: remove encoding.BinaryMarshaler interface
* separately, limit chunk-manifest's metadata entry size <= 4K

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

Author: alex-aizman

ais/bucketmeta.go

@@ -430,7 +430,7 @@ func loadBMD(mpaths fs.MPI, path string) (mainBMD *bucketMD) {
 			mainBMD = bmd
 			continue
 		}
-		if mainBMD.cksum.IsEmpty() {
+		if cos.NoneC(mainBMD.cksum) {
 			cos.ExitLogf("BMD is not checksummed (%q): %v", mpath, mainBMD)
 		}
 		if mainBMD.cksum.Equal(bmd.cksum) {

ais/etlmeta.go

@@ -284,7 +284,7 @@ func loadEtlMD(mpaths fs.MPI, path string) (mainEtlMD *etlMD) {
 			mainEtlMD = etlMD
 			continue
 		}
-		if mainEtlMD.cksum.IsEmpty() {
+		if cos.NoneC(mainEtlMD.cksum) {
 			cos.ExitLogf("EtlMD is not checksummed (%q): %v", mpath, mainEtlMD)
 		}
 		if mainEtlMD.cksum.Equal(etlMD.cksum) {

ais/test/object_test.go

@@ -292,7 +292,7 @@ func TestAppendObject(t *testing.T) {
 			writer := bytes.NewBuffer(nil)
 			getArgs := api.GetArgs{Writer: writer}
 			oah, err := api.GetObjectWithValidation(baseParams, bck, objName, &getArgs)
-			if !cksum.IsEmpty() {
+			if !cos.NoneH(cksum) {
 				tassert.CheckFatal(t, err)
 			}
 			tassert.Errorf(

ais/tgtobj.go

@@ -489,7 +489,7 @@ func (poi *putOI) write() (buf []byte, slab *memsys.Slab, lmfh cos.LomWriter, er
 		// not using `ReadFrom` of the `*os.File` -
 		// ultimately, https://github.com/golang/go/blob/master/src/internal/poll/copy_file_range_linux.go#L100
 		written, err = cos.CopyBuffer(lmfh, poi.r, buf)
-	case !poi.cksumToUse.IsEmpty() && !poi.validateCksum(ckconf):
+	case !cos.NoneC(poi.cksumToUse) && !poi.validateCksum(ckconf):
 		// if the corresponding validation is not configured/enabled we just go ahead
 		// and use the checksum that has arrived with the object
 		poi.lom.SetCksum(poi.cksumToUse)
@@ -499,7 +499,7 @@ func (poi *putOI) write() (buf []byte, slab *memsys.Slab, lmfh cos.LomWriter, er
 		writers := make([]io.Writer, 0, 3)
 		cksums.store = cos.NewCksumHash(ckconf.Type) // always according to the bucket
 		writers = append(writers, cksums.store.H)
-		if !poi.skipVC && !poi.cksumToUse.IsEmpty() && poi.validateCksum(ckconf) {
+		if !poi.skipVC && !cos.NoneC(poi.cksumToUse) && poi.validateCksum(ckconf) {
 			cksums.expct = poi.cksumToUse
 			if poi.cksumToUse.Type() == cksums.store.Type() {
 				cksums.compt = cksums.store
@@ -1450,7 +1450,7 @@ func (a *apndOI) flush() (int, error) {
 	debug.Assert(a.hdl.partialCksum != nil)
 	a.hdl.partialCksum.Finalize()
 	partialCksum := a.hdl.partialCksum.Clone()
-	if !a.cksum.IsEmpty() && !partialCksum.Equal(a.cksum) {
+	if !cos.NoneC(a.cksum) && !partialCksum.Equal(a.cksum) {
 		return http.StatusInternalServerError, cos.NewErrDataCksum(partialCksum, a.cksum)
 	}
 

ais/tpromote.go

@@ -109,7 +109,7 @@ func (t *target) _promLocal(params *core.PromoteParams, lom *core.LOM) (fileSize
 		}
 	}
 
-	if params.Cksum != nil && cksum != nil && !cksum.IsEmpty() {
+	if params.Cksum != nil && !cos.NoneH(cksum) {
 		if !cksum.Equal(params.Cksum) {
 			return 0, 0, cos.NewErrDataCksum(
 				cksum.Clone(),

cmn/cos/cksum.go

@@ -44,7 +44,7 @@ const (
 )
 
 type (
-	noopHash struct{}
+	noneHash struct{}
 
 	// in-cluster checksum validation (compare with cmn.ErrInvalidCksum)
 	ErrBadCksum struct {
@@ -77,16 +77,21 @@ var checksums = StrSet{
 	ChecksumSHA512: {},
 }
 
-// interface guard
-var (
-	_ hash.Hash                  = (*noopHash)(nil)
-	_ encoding.BinaryUnmarshaler = (*noopHash)(nil)
-	_ encoding.BinaryUnmarshaler = (*noopHash)(nil)
+var NoneCksum = NewCksum(ChecksumNone, "")
 
-	_ io.Writer = (*CksumHashSize)(nil)
-)
+func NoneC(ck *Cksum) bool {
+	if ck == nil {
+		return true
+	}
+	return ck.ty == "" || ck.ty == ChecksumNone || ck.value == ""
+}
 
-var NoneCksum = NewCksum(ChecksumNone, "")
+func NoneH(ck *CksumHash) bool {
+	if ck == nil {
+		return true
+	}
+	return NoneC(&ck.Cksum)
+}
 
 ///////////////
 // CksumHash //
@@ -121,7 +126,7 @@ func (ck *CksumHash) Init(ty string) {
 	ck.ty = ty
 	switch ty {
 	case ChecksumNone, "":
-		ck.ty, ck.H = ChecksumNone, newNoopHash()
+		ck.ty, ck.H = ChecksumNone, &noneHash{}
 	case ChecksumOneXxh:
 		ck.H = onexxh.New64()
 	case ChecksumCesXxh:
@@ -153,6 +158,11 @@ func (ck *CksumHash) Finalize() {
 // CksumHashSize //
 ///////////////////
 
+// interface guard
+var (
+	_ io.Writer = (*CksumHashSize)(nil)
+)
+
 func (ck *CksumHashSize) Write(b []byte) (n int, err error) {
 	n, err = ck.H.Write(b)
 	ck.Size += int64(n)
@@ -163,10 +173,6 @@ func (ck *CksumHashSize) Write(b []byte) (n int, err error) {
 // Cksum //
 ///////////
 
-func (ck *Cksum) IsEmpty() bool {
-	return ck == nil || ck.ty == "" || ck.ty == ChecksumNone || ck.value == ""
-}
-
 func NewCksum(ty, value string) *Cksum {
 	if err := ValidateCksumType(ty, true /*empty OK*/); err != nil {
 		AssertMsg(false, err.Error())
@@ -179,8 +185,8 @@ func NewCksum(ty, value string) *Cksum {
 
 // NOTE [caution]: empty checksums are also equal (compare with lom.EqCksum and friends)
 func (ck *Cksum) Equal(to *Cksum) bool {
-	if ck.IsEmpty() || to.IsEmpty() {
-		return ck.IsEmpty() == to.IsEmpty()
+	if a, b := NoneC(ck), NoneC(to); a || b {
+		return a && b
 	}
 	return ck.ty == to.ty && ck.value == to.value
 }
@@ -258,17 +264,23 @@ func ValidateCksumType(ty string, emptyOK ...bool) (err error) {
 }
 
 //
-// noopHash
+// noneHash
 //
 
-func newNoopHash() hash.Hash                     { return &noopHash{} }
-func (*noopHash) Write(b []byte) (int, error)    { return len(b), nil }
-func (*noopHash) Sum([]byte) []byte              { return nil }
-func (*noopHash) Reset()                         {}
-func (*noopHash) Size() int                      { return 0 }
-func (*noopHash) BlockSize() int                 { return KiB }
-func (*noopHash) MarshalBinary() ([]byte, error) { return nil, nil }
-func (*noopHash) UnmarshalBinary([]byte) error   { return nil }
+// interface guard
+var (
+	_ hash.Hash                = (*noneHash)(nil)
+	_ encoding.BinaryMarshaler = (*noneHash)(nil) // usage: append object (to serialize append handle)
+)
+
+func (*noneHash) Write(b []byte) (int, error) { return len(b), nil }
+func (*noneHash) Sum([]byte) []byte           { return nil }
+func (*noneHash) Reset()                      {}
+func (*noneHash) Size() int                   { return 0 }
+func (*noneHash) BlockSize() int              { return KiB }
+
+func (*noneHash) MarshalBinary() ([]byte, error) { return nil, nil }
+func (*noneHash) UnmarshalBinary([]byte) error   { return nil }
 
 //
 // errors

cmn/objattrs.go

@@ -111,7 +111,7 @@ func (oa *ObjAttrs) Checksum() *cos.Cksum    { return oa.Cksum }
 func (oa *ObjAttrs) SetCksum(ty, val string) { oa.Cksum = cos.NewCksum(ty, val) }
 
 func (oa *ObjAttrs) EqCksum(cksum *cos.Cksum) bool {
-	return !oa.Cksum.IsEmpty() && oa.Cksum.Equal(cksum)
+	return !cos.NoneC(oa.Cksum) && oa.Cksum.Equal(cksum)
 }
 
 func (oa *ObjAttrs) Version(_ ...bool) string {
@@ -189,7 +189,7 @@ func ToHeader(oah cos.OAH, hdr http.Header, size int64, cksums ...*cos.Cksum) {
 	} else {
 		cksum = oah.Checksum()
 	}
-	if !cksum.IsEmpty() {
+	if !cos.NoneC(cksum) {
 		hdr.Set(apc.HdrObjCksumType, cksum.Ty())
 		hdr.Set(apc.HdrObjCksumVal, cksum.Val())
 	}
@@ -270,12 +270,12 @@ func (oa *ObjAttrs) CheckEq(rem cos.OAH) error {
 		sameEtag  bool
 		sameCksum bool
 	)
-	// size check
+	// 1. size
 	if remSize := rem.Lsize(true); oa.Size != 0 && remSize != 0 && oa.Size != remSize {
 		return fmt.Errorf("size %d != %d remote", oa.Size, remSize)
 	}
 
-	// Cloud version check (NOTE: ais own version is currently a non-unique sequence number)
+	// 2. version (note that ais own version is a simple sequence number)
 	if remMeta, ok := rem.GetCustomKey(VersionObjMD); ok && remMeta != "" {
 		if locMeta, ok := oa.GetCustomKey(VersionObjMD); ok && locMeta != "" {
 			if remMeta != locMeta {
@@ -286,32 +286,28 @@ func (oa *ObjAttrs) CheckEq(rem cos.OAH) error {
 		}
 	}
 
-	// checksum check
-	if a, b := rem.Checksum(), oa.Cksum; a != nil && b != nil {
-		cksumType := a.Ty()
-		if !a.IsEmpty() && !b.IsEmpty() && cksumType == b.Ty() {
+	// 3. checksum
+	if a := rem.Checksum(); !cos.NoneC(a) {
+		b := oa.Cksum
+		if !cos.NoneC(b) {
+			ty := a.Ty()
 			if !a.Equal(b) {
-				return fmt.Errorf("%s checksum %s != %s remote", cksumType, b, a)
+				return fmt.Errorf("%s checksum %s != %s remote", ty, b, a)
 			}
 			cksumVal = a.Val()
-
-			// [NOTE]
-			// unless overridden via feature flag
-			// trust two checksums, namely md5 and xxhash, that are _not_ cryptographically secure
-
 			switch {
 			case Rom.Features().IsSet(feat.TrustCryptoSafeChecksums):
-				sameCksum = (cksumType == cos.ChecksumSHA256 || cksumType == cos.ChecksumSHA512)
+				sameCksum = (ty == cos.ChecksumSHA256 || ty == cos.ChecksumSHA512)
 			default:
-				debug.Assert(cksumType != cos.ChecksumNone)
-				sameCksum = cksumType != cos.ChecksumCRC32C
+				// NOTE trust non-cryptographic checksums except crc (unless overridden by feature flag)
+				debug.Assert(ty != cos.ChecksumNone)
+				sameCksum = ty != cos.ChecksumCRC32C
 			}
-
 			count++
 		}
 	}
 
-	// custom MD: ETag check (ignoring enclosing quotes)
+	// 4. custom MD: ETag check (ignoring enclosing quotes)
 	if remMeta, ok := rem.GetCustomKey(ETag); ok && remMeta != "" {
 		if locMeta, ok := oa.GetCustomKey(ETag); ok && locMeta != "" {
 			if !_eqIgnoreQuotes(remMeta, locMeta) {
@@ -324,7 +320,7 @@ func (oa *ObjAttrs) CheckEq(rem cos.OAH) error {
 			}
 		}
 	}
-	// custom MD: CRC check
+	// 4.1. custom MD: CRC
 	if remMeta, ok := rem.GetCustomKey(CRC32CObjMD); ok && remMeta != "" {
 		if locMeta, ok := oa.GetCustomKey(CRC32CObjMD); ok && locMeta != "" {
 			if remMeta != locMeta {
@@ -336,7 +332,7 @@ func (oa *ObjAttrs) CheckEq(rem cos.OAH) error {
 		}
 	}
 
-	// custom MD: MD5 check iff count < 2
+	// 4.2. custom MD: MD5 iff count < 2
 	// (ETag ambiguity, see: https://docs.aws.amazon.com/AmazonS3/latest/API/API_Object.htm)
 	if !sameEtag {
 		if remMeta, ok := rem.GetCustomKey(MD5ObjMD); ok && remMeta != "" {

core/lcopy.go

@@ -278,7 +278,7 @@ func (lom *LOM) copy2fqn(dst *LOM, buf []byte) (err error) {
 		srcCksum  = lom.Checksum()
 		cksumType = cos.ChecksumNone
 	)
-	if !srcCksum.IsEmpty() {
+	if !cos.NoneC(srcCksum) {
 		cksumType = srcCksum.Ty()
 	}
 	if dst.isMirror(lom) && lom.md.copies != nil {

core/lom.go

@@ -171,7 +171,7 @@ func (lom *LOM) SetSize(size int64) { lom.md.Size = size }
 func (lom *LOM) Checksum() *cos.Cksum      { return lom.md.Cksum }
 func (lom *LOM) SetCksum(cksum *cos.Cksum) { lom.md.Cksum = cksum }
 func (lom *LOM) EqCksum(cksum *cos.Cksum) bool {
-	return !lom.md.Cksum.IsEmpty() && lom.md.Cksum.Equal(cksum)
+	return !cos.NoneC(lom.md.Cksum) && lom.md.Cksum.Equal(cksum)
 }
 
 func (lom *LOM) Atime() time.Time      { return time.Unix(0, lom.md.Atime) }
@@ -331,13 +331,13 @@ recomp:
 	if cksumType == cos.ChecksumNone { // as far as do-no-checksum-checking bucket rules
 		return nil
 	}
-	if !lom.md.Cksum.IsEmpty() {
+	if !cos.NoneC(lom.md.Cksum) {
 		cksumType = lom.md.Cksum.Ty() // takes precedence on the other hand
 	}
 	if cksums.comp, err = lom.ComputeCksum(cksumType, locked); err != nil {
 		return err
 	}
-	if lom.md.Cksum.IsEmpty() { // store computed
+	if cos.NoneC(lom.md.Cksum) { // store computed
 		lom.md.Cksum = cksums.comp.Clone()
 		if !lom.loaded() {
 			lom.SetAtimeUnix(time.Now().UnixNano())

core/ufest.go

@@ -37,7 +37,9 @@ const (
 	iniChunksCap = 16
 
 	maxChunkSize = 5 * cos.GiB
-	maxMetaKeys  = 1000
+
+	maxMetaKeys     = 1000
+	maxMetaEntryLen = 4 * cos.KiB
 )
 
 const (
@@ -108,6 +110,11 @@ func (u *Ufest) SetMeta(md map[string]string) error {
 	if l := len(md); l > maxMetaKeys {
 		return fmt.Errorf("%s: number of metadata entries %d exceeds %d limit", utag, l, maxMetaKeys)
 	}
+	for k, v := range md {
+		if len(k) > maxMetaEntryLen || len(v) > maxMetaEntryLen {
+			return fmt.Errorf("%s: metadata entry too large (%d, %d, %d)", utag, len(k), len(v), maxMetaEntryLen)
+		}
+	}
 	u.mdmap = md
 	return nil
 }
@@ -419,7 +426,7 @@ func _packStr(w io.Writer, s string) {
 }
 
 func _packCksum(w io.Writer, cksum *cos.Cksum) {
-	if cksum == nil || cksum.IsEmpty() {
+	if cos.NoneC(cksum) {
 		_packStr(w, "")
 		return
 	}