test(storage): harden remote dedupe convergence checks

Signed-off-by: Ramkumar Chinchani <rchincha.dev@gmail.com>

Author: rchincha

pkg/storage/s3/s3_test.go

@@ -1069,6 +1069,55 @@ func TestNegativeCasesObjectsStorage(t *testing.T) {
 func TestS3Dedupe(t *testing.T) {
 	tskip.SkipS3(t)
 	tskip.SkipDynamo(t)
+
+	waitForBlobStatExists := func(storeDrv driver.StorageDriver, rootDir string, repo string,
+		digest godigest.Digest,
+	) error {
+		var statErr error
+
+		for range 300 {
+			_, statErr = storeDrv.Stat(context.Background(), path.Join(rootDir, repo, "blobs", "sha256", digest.Encoded()))
+			if statErr == nil {
+				return nil
+			}
+
+			time.Sleep(100 * time.Millisecond)
+		}
+
+		return fmt.Errorf("timed out waiting for stat on blob %s/%s: %w", repo, digest.Encoded(), statErr)
+	}
+
+	waitForBlobStatSize := func(storeDrv driver.StorageDriver, rootDir string, repo string,
+		digest godigest.Digest, expectedSize int64,
+	) error {
+		var (
+			statErr      error
+			observedSize int64 = -1
+		)
+
+		for range 300 {
+			fi, err := storeDrv.Stat(context.Background(), path.Join(rootDir, repo, "blobs", "sha256", digest.Encoded()))
+			if err == nil {
+				observedSize = fi.Size()
+				if observedSize == expectedSize {
+					return nil
+				}
+			} else {
+				statErr = err
+			}
+
+			time.Sleep(100 * time.Millisecond)
+		}
+
+		if statErr != nil {
+			return fmt.Errorf("timed out waiting for size %d on blob %s/%s: %w",
+				expectedSize, repo, digest.Encoded(), statErr)
+		}
+
+		return fmt.Errorf("timed out waiting for size %d on blob %s/%s, observed %d",
+			expectedSize, repo, digest.Encoded(), observedSize)
+	}
+
 	Convey("Dedupe", t, func(c C) {
 		uuid, err := guuid.NewV4()
 		if err != nil {
@@ -1478,16 +1527,19 @@ func TestS3Dedupe(t *testing.T) {
 
 				// rebuild with dedupe false, should have all blobs with content
 				imgStore.RunDedupeBlobs(time.Duration(0), taskScheduler)
-				// wait until rebuild finishes
 
-				time.Sleep(10 * time.Second)
+				err = waitForBlobStatExists(storeDriver, testDir, "dedupe1", blobDigest1)
+				So(err, ShouldBeNil)
 
 				taskScheduler.Shutdown()
 
 				fi1, err := storeDriver.Stat(context.Background(), path.Join(testDir, "dedupe1", "blobs", "sha256",
 					blobDigest1.Encoded()))
 				So(err, ShouldBeNil)
 
+				err = waitForBlobStatSize(storeDriver, testDir, "dedupe2", blobDigest2, fi1.Size())
+				So(err, ShouldBeNil)
+
 				blobContent1, err := imgStore.GetBlobContent("dedupe1", blobDigest1)
 				So(err, ShouldBeNil)
 				So(len(blobContent1), ShouldBeGreaterThan, 0)
@@ -1510,9 +1562,9 @@ func TestS3Dedupe(t *testing.T) {
 
 					// rebuild with dedupe false, should have all blobs with content
 					imgStore.RunDedupeBlobs(time.Duration(0), taskScheduler)
-					// wait until rebuild finishes
 
-					time.Sleep(10 * time.Second)
+					err = waitForBlobStatSize(storeDriver, testDir, "dedupe2", blobDigest2, 0)
+					So(err, ShouldBeNil)
 
 					taskScheduler.Shutdown()
 
@@ -1761,9 +1813,9 @@ func TestS3Dedupe(t *testing.T) {
 
 			// rebuild with dedupe false, should have all blobs with content
 			imgStore.RunDedupeBlobs(time.Duration(0), taskScheduler)
-			// wait until rebuild finishes
 
-			time.Sleep(10 * time.Second)
+			err = waitForBlobStatExists(storeDriver, testDir, "dedupe1", blobDigest1)
+			So(err, ShouldBeNil)
 
 			taskScheduler.Shutdown()
 
@@ -1841,9 +1893,9 @@ func TestS3Dedupe(t *testing.T) {
 
 				// rebuild with dedupe false, should have all blobs with content
 				imgStore.RunDedupeBlobs(time.Duration(0), taskScheduler)
-				// wait until rebuild finishes
 
-				time.Sleep(10 * time.Second)
+				err = waitForBlobStatSize(storeDriver, testDir, "dedupe2", blobDigest2, 0)
+				So(err, ShouldBeNil)
 
 				taskScheduler.Shutdown()
 

pkg/storage/storage_test.go

@@ -2518,18 +2518,39 @@ func TestReuploadCorruptedBlob(t *testing.T) {
 					ok       bool
 					size     int64
 					checkErr error
+					statOK   bool
+					statSize int64
+					statErr  error
 				)
 
-				for range 100 {
+				for range 300 {
 					ok, size, checkErr = imgStore.CheckBlob(context.Background(), repoName, digest)
-					if checkErr == nil && ok && size == int64(expectedSize) {
+					statOK, statSize, _, statErr = imgStore.StatBlob(repoName, digest)
+
+					if checkErr == nil && statErr == nil && ok && statOK &&
+						size == int64(expectedSize) && statSize == int64(expectedSize) {
 						return ok, size, nil
 					}
 
 					time.Sleep(100 * time.Millisecond)
 				}
 
-				return ok, size, checkErr
+				if checkErr != nil {
+					return ok, size, checkErr
+				}
+
+				if statErr != nil {
+					return statOK, statSize, statErr
+				}
+
+				if statOK {
+					return statOK, statSize,
+						fmt.Errorf("timed out waiting for blob %s size convergence: check=%d stat=%d expected=%d",
+							digest.String(), size, statSize, expectedSize)
+				}
+
+				return ok, size,
+					fmt.Errorf("timed out waiting for blob %s to be present with expected size %d", digest.String(), expectedSize)
 			}
 
 			Convey("Test reupload repair corrupted image", t, func() {