Method validateShardPrefix

pkg/server/sync.go:776–831

(pfx string)

}

func (sh *SyncHandler) validateShardPrefix(pfx string) (err error) {
	defer func() {
		sh.mu.Lock()
		if err != nil {
			errs := fmt.Sprintf("Failed to validate prefix %s: %v", pfx, err)
			sh.logf("%s", errs)
			sh.vshardErrs = append(sh.vshardErrs, errs)
		}
		sh.vshardDone++
		sh.mu.Unlock()
	}()
	ctx, cancel := context.WithCancel(context.TODO())
	defer cancel()
	src, serrc := sh.startValidatePrefix(ctx, pfx, false)
	dst, derrc := sh.startValidatePrefix(ctx, pfx, true)
	srcErr := &chanError{
		C: serrc,
		Wrap: func(err error) error {
			return fmt.Errorf("Error enumerating source %s for validating shard %s: %v", sh.fromName, pfx, err)
		},
	}
	dstErr := &chanError{
		C: derrc,
		Wrap: func(err error) error {
			return fmt.Errorf("Error enumerating target %s for validating shard %s: %v", sh.toName, pfx, err)
		},
	}

	missingc := make(chan blob.SizedRef, 8)
	go blobserver.ListMissingDestinationBlobs(missingc, func(blob.Ref) {}, src, dst)

	var missing []blob.SizedRef
	for sb := range missingc {
		missing = append(missing, sb)
	}

	if err := srcErr.Get(); err != nil {
		return err
	}
	if err := dstErr.Get(); err != nil {
		return err
	}

	for _, sb := range missing {
		if enqErr := sh.enqueue(sb); enqErr != nil {
			if err == nil {
				err = enqErr
			}
		} else {
			sh.mu.Lock()
			sh.vmissing++
			sh.mu.Unlock()
		}
	}
	return err
}

var errNotPrefix = errors.New("sentinel error: hit blob into the next shard")