Method ReceiveBlob

pkg/index/receive.go:192–289

(ctx context.Context, blobRef blob.Ref, source io.Reader)

}

func (ix *Index) ReceiveBlob(ctx context.Context, blobRef blob.Ref, source io.Reader) (blob.SizedRef, error) {
	// Read from source before acquiring ix.Lock (Issue 878):
	sniffer := NewBlobSniffer(blobRef)
	written, err := io.Copy(sniffer, source)
	if err != nil {
		return blob.SizedRef{}, err
	}
	sbr := blob.SizedRef{Ref: blobRef, Size: uint32(written)}

	ix.Lock()
	defer ix.Unlock()

	missingDeps := false
	defer func() {
		if err == nil {
			ix.noteBlobIndexed(blobRef)
			if !missingDeps {
				ix.removeAllMissingEdges(blobRef)
			}
		}
	}()

	// By default, return immediately if it looks like we already
	// have indexed this blob before.  But if the user has
	// CAMLI_REDO_INDEX_ON_RECEIVE set in their environment,
	// always index it. This is generally only useful when working
	// on the indexing code and retroactively indexing a subset of
	// content without forcing a global reindexing.
	if haveVal, haveErr := ix.s.Get("have:" + blobRef.String()); haveErr == nil {
		if strings.HasSuffix(haveVal, "|indexed") {
			if allowReindex, _ := strconv.ParseBool(os.Getenv("CAMLI_REDO_INDEX_ON_RECEIVE")); allowReindex {
				if debugEnv {
					log.Printf("index: reindexing %v", sbr)
				}
			} else {
				if debugEnv {
					log.Printf("index: ignoring upload of already-indexed %v", sbr)
				}
				return sbr, nil
			}
		}
	}

	sniffer.Parse()

	fetcher := &missTrackFetcher{
		fetcher: ix.blobSource,
	}

	mm, err := ix.populateMutationMap(ctx, fetcher, blobRef, sniffer)
	if debugEnv {
		log.Printf("index of %v: mm=%v, err=%v", blobRef, mm, err)
	}
	if err != nil {
		if err != errMissingDep {
			return blob.SizedRef{}, err
		}
		fetcher.mu.Lock()