object: stop verifying object length in VerifyObject

With compression, the decompressed object length is
not known until we read the content.

cli/command_snapshot_verify.go

@@ -104,14 +104,14 @@ func (v *verifier) enqueueVerifyDirectory(ctx context.Context, oid object.ID, pa
 	})
 }
 
-func (v *verifier) enqueueVerifyObject(ctx context.Context, oid object.ID, path string, expectedLength int64) {
+func (v *verifier) enqueueVerifyObject(ctx context.Context, oid object.ID, path string) {
 	// push to the back of the queue, so that we process non-directories at the end.
 	if !v.shouldEnqueue(oid) {
 		return
 	}
 
 	v.workQueue.EnqueueBack(func() error {
-		return v.doVerifyObject(ctx, oid, path, expectedLength)
+		return v.doVerifyObject(ctx, oid, path)
 	})
 }
 
@@ -137,33 +137,20 @@ func (v *verifier) doVerifyDirectory(ctx context.Context, oid object.ID, path st
 		if e.IsDir() {
 			v.enqueueVerifyDirectory(ctx, objectID, childPath)
 		} else {
-			v.enqueueVerifyObject(ctx, objectID, childPath, e.Size())
+			v.enqueueVerifyObject(ctx, objectID, childPath)
 		}
 	}
 
 	return nil
 }
 
-func (v *verifier) doVerifyObject(ctx context.Context, oid object.ID, path string, expectedLength int64) error {
-	if expectedLength < 0 {
-		log.Debugf("verifying object %v", oid)
-	} else {
-		log.Debugf("verifying object %v (%v) with length %v", path, oid, expectedLength)
-	}
+func (v *verifier) doVerifyObject(ctx context.Context, oid object.ID, path string) error {
+	log.Debugf("verifying object %v", oid)
 
-	var length int64
-
-	var err error
-
-	length, _, err = v.om.VerifyObject(ctx, oid)
-	if err != nil {
+	if _, err := v.om.VerifyObject(ctx, oid); err != nil {
 		v.reportError(path, errors.Wrapf(err, "error verifying %v", oid))
 	}
 
-	if expectedLength >= 0 && length != expectedLength {
-		v.reportError(path, errors.Errorf("invalid object length %q, %v, expected %v", oid, length, expectedLength))
-	}
-
 	if rand.Intn(100) < *verifyCommandFilesPercent {
 		if err := v.readEntireObject(ctx, oid, path); err != nil {
 			v.reportError(path, errors.Wrapf(err, "error reading object %v", oid))
@@ -231,7 +218,7 @@ func enqueueRootsToVerify(ctx context.Context, v *verifier, rep *repo.Repository
 		if man.RootEntry.Type == snapshot.EntryTypeDirectory {
 			v.enqueueVerifyDirectory(ctx, man.RootObjectID(), path)
 		} else {
-			v.enqueueVerifyObject(ctx, man.RootObjectID(), path, -1)
+			v.enqueueVerifyObject(ctx, man.RootObjectID(), path)
 		}
 	}
 
@@ -250,7 +237,7 @@ func enqueueRootsToVerify(ctx context.Context, v *verifier, rep *repo.Repository
 			return err
 		}
 
-		v.enqueueVerifyObject(ctx, oid, oidStr, -1)
+		v.enqueueVerifyObject(ctx, oid, oidStr)
 	}
 
 	return nil

repo/object/object_manager.go

@@ -57,6 +57,10 @@ func (om *Manager) NewWriter(ctx context.Context, opt WriterOptions) Writer {
 
 // Open creates new ObjectReader for reading given object from a repository.
 func (om *Manager) Open(ctx context.Context, objectID ID) (Reader, error) {
+	return om.openAndAssertLength(ctx, objectID, -1)
+}
+
+func (om *Manager) openAndAssertLength(ctx context.Context, objectID ID, assertLength int64) (Reader, error) {
 	if indexObjectID, ok := objectID.IndexObjectID(); ok {
 		rd, err := om.Open(ctx, indexObjectID)
 		if err != nil {
@@ -79,71 +83,63 @@ func (om *Manager) Open(ctx context.Context, objectID ID) (Reader, error) {
 		}, nil
 	}
 
-	return om.newRawReader(ctx, objectID)
+	return om.newRawReader(ctx, objectID, assertLength)
 }
 
 // VerifyObject ensures that all objects backing ObjectID are present in the repository
-// and returns the total length of the object and content IDs of which it is composed.
-func (om *Manager) VerifyObject(ctx context.Context, oid ID) (int64, []content.ID, error) {
+// and returns the content IDs of which it is composed.
+func (om *Manager) VerifyObject(ctx context.Context, oid ID) ([]content.ID, error) {
 	tracker := &contentIDTracker{}
 
-	l, err := om.verifyObjectInternal(ctx, oid, tracker)
-	if err != nil {
-		return 0, nil, err
+	if err := om.verifyObjectInternal(ctx, oid, tracker); err != nil {
+		return nil, err
 	}
 
-	return l, tracker.contentIDs(), nil
+	return tracker.contentIDs(), nil
 }
 
-func (om *Manager) verifyIndirectObjectInternal(ctx context.Context, indexObjectID ID, tracker *contentIDTracker) (int64, error) {
-	if _, err := om.verifyObjectInternal(ctx, indexObjectID, tracker); err != nil {
-		return 0, errors.Wrap(err, "unable to read index")
+func (om *Manager) verifyIndirectObjectInternal(ctx context.Context, indexObjectID ID, tracker *contentIDTracker) error {
+	if err := om.verifyObjectInternal(ctx, indexObjectID, tracker); err != nil {
+		return errors.Wrap(err, "unable to read index")
 	}
 
 	rd, err := om.Open(ctx, indexObjectID)
 	if err != nil {
-		return 0, err
+		return err
 	}
 	defer rd.Close() //nolint:errcheck
 
 	seekTable, err := om.flattenListChunk(rd)
 	if err != nil {
-		return 0, err
+		return err
 	}
 
-	for i, m := range seekTable {
-		l, err := om.verifyObjectInternal(ctx, m.Object, tracker)
+	for _, m := range seekTable {
+		err := om.verifyObjectInternal(ctx, m.Object, tracker)
 		if err != nil {
-			return 0, err
-		}
-
-		if l != m.Length {
-			return 0, errors.Errorf("unexpected length of part %#v of indirect object %q: %v %v, expected %v", i, indexObjectID, m.Object, l, m.Length)
+			return err
 		}
 	}
 
-	totalLength := seekTable[len(seekTable)-1].endOffset()
-
-	return totalLength, nil
+	return nil
 }
 
-func (om *Manager) verifyObjectInternal(ctx context.Context, oid ID, tracker *contentIDTracker) (int64, error) {
+func (om *Manager) verifyObjectInternal(ctx context.Context, oid ID, tracker *contentIDTracker) error {
 	if indexObjectID, ok := oid.IndexObjectID(); ok {
 		return om.verifyIndirectObjectInternal(ctx, indexObjectID, tracker)
 	}
 
 	if contentID, ok := oid.ContentID(); ok {
-		p, err := om.contentMgr.ContentInfo(ctx, contentID)
-		if err != nil {
-			return 0, err
+		if _, err := om.contentMgr.ContentInfo(ctx, contentID); err != nil {
+			return err
 		}
 
 		tracker.addContentID(contentID)
 
-		return int64(p.Length), nil
+		return nil
 	}
 
-	return 0, errors.Errorf("unrecognized object type: %v", oid)
+	return errors.Errorf("unrecognized object type: %v", oid)
 }
 
 func nullTrace(message string, args ...interface{}) {
@@ -210,7 +206,7 @@ func (om *Manager) flattenListChunk(rawReader io.Reader) ([]indirectObjectEntry,
 	return ind.Entries, nil
 }
 
-func (om *Manager) newRawReader(ctx context.Context, objectID ID) (Reader, error) {
+func (om *Manager) newRawReader(ctx context.Context, objectID ID, assertLength int64) (Reader, error) {
 	if contentID, ok := objectID.ContentID(); ok {
 		payload, err := om.contentMgr.GetContent(ctx, contentID)
 		if err == content.ErrContentNotFound {
@@ -221,6 +217,10 @@ func (om *Manager) newRawReader(ctx context.Context, objectID ID) (Reader, error
 			return nil, errors.Wrap(err, "unexpected content error")
 		}
 
+		if assertLength != -1 && int64(len(payload)) != assertLength {
+			return nil, errors.Wrapf(err, "unexpected chunk length %v, expected %v", len(payload), assertLength)
+		}
+
 		return newObjectReaderWithData(payload), nil
 	}
 

repo/object/object_manager_test.go

@@ -205,15 +205,11 @@ func TestIndirection(t *testing.T) {
 			t.Errorf("unexpected blob count for %v: %v, expected %v", c.dataLength, got, want)
 		}
 
-		l, b, err := om.VerifyObject(ctx, result)
+		b, err := om.VerifyObject(ctx, result)
 		if err != nil {
 			t.Errorf("error verifying %q: %v", result, err)
 		}
 
-		if got, want := int(l), len(contentBytes); got != want {
-			t.Errorf("got invalid byte count for %q: %v, wanted %v", result, got, want)
-		}
-
 		if got, want := len(b), c.expectedBlobCount; got != want {
 			t.Errorf("invalid blob count for %v, got %v, wanted %v", result, got, want)
 		}
@@ -316,6 +312,7 @@ func TestEndToEndReadAndSeek(t *testing.T) {
 		}
 
 		objectID, err := writer.Result()
+		t.Logf("oid: %v", objectID)
 
 		writer.Close()
 

repo/object/object_reader.go

@@ -79,7 +79,7 @@ func (r *objectReader) Read(buffer []byte) (int, error) {
 func (r *objectReader) openCurrentChunk() error {
 	st := r.seekTable[r.currentChunkIndex]
 
-	rd, err := r.repo.Open(r.ctx, st.Object)
+	rd, err := r.repo.openAndAssertLength(r.ctx, st.Object, st.Length)
 	if err != nil {
 		return err
 	}

snapshot/gc/gc.go

@@ -50,7 +50,7 @@ func findInUseContentIDs(ctx context.Context, rep *repo.Repository, used *sync.M
 
 	w.ObjectCallback = func(entry fs.Entry) error {
 		oid := oidOf(entry)
-		_, contentIDs, err := rep.Objects.VerifyObject(ctx, oid)
+		contentIDs, err := rep.Objects.VerifyObject(ctx, oid)
 
 		if err != nil {
 			return errors.Wrapf(err, "error verifying %v", oid)