Method hourlyCompare

pkg/server/sync.go:1077–1133

Every hour, hourlyCompare picks blob names from a random point in the source, downloads up to hourlyBytes from the destination, and verifies them.

(hourlyBytes uint64)

// Every hour, hourlyCompare picks blob names from a random point in the source,
// downloads up to hourlyBytes from the destination, and verifies them.
func (sh *SyncHandler) hourlyCompare(hourlyBytes uint64) {
	ctx := context.TODO()
	ticker := time.NewTicker(time.Hour).C
	for {
		content := make([]byte, 16)
		if _, err := rand.Read(content); err != nil {
			panic(err)
		}
		after := blob.RefFromBytes(content).String()
		var roundBytes uint64
		var roundBlobs int
		err := blobserver.EnumerateAllFrom(ctx, sh.from, after, func(sr blob.SizedRef) error {
			sh.mu.Lock()
			if _, ok := sh.needCopy[sr.Ref]; ok {
				sh.mu.Unlock()
				return nil // skip blobs in the copy queue
			}
			sh.mu.Unlock()

			if roundBytes+uint64(sr.Size) > hourlyBytes {
				return errStopEnumerating
			}
			blob, size, err := sh.to.(blob.Fetcher).Fetch(ctx, sr.Ref)
			if err != nil {
				return fmt.Errorf("error fetching %s: %v", sr.Ref, err)
			}
			if size != sr.Size {
				return fmt.Errorf("%s: expected size %d, got %d", sr.Ref, sr.Size, size)
			}
			h := sr.Ref.Hash()
			if _, err := io.Copy(h, blob); err != nil {
				return fmt.Errorf("error while reading %s: %v", sr.Ref, err)
			}
			if !sr.HashMatches(h) {
				return fmt.Errorf("expected %s, got %x", sr.Ref, h.Sum(nil))
			}

			sh.mu.Lock()
			sh.comparedBlobs++
			sh.comparedBytes += uint64(size)
			sh.compLastBlob = sr.Ref.String()
			sh.mu.Unlock()
			roundBlobs++
			roundBytes += uint64(size)
			return nil
		})
		sh.mu.Lock()
		if err != nil && err != errStopEnumerating {
			sh.compareErrors = append(sh.compareErrors, fmt.Sprintf("%s %v", time.Now(), err))
			sh.logf("!! hourly compare error !!: %v", err)
		}
		sh.comparedRounds++
		sh.mu.Unlock()
		sh.logf("compared %d blobs (%d bytes)", roundBlobs, roundBytes)
		<-ticker
	}
}