func dbPutManifestList()

in registry/handlers/manifests.go [1143:1264]
90 lines of code
17 McCabe index (conditional complexity)

func dbPutManifestList(imh *manifestHandler, manifestList *manifestlist.DeserializedManifestList, payload []byte) error {
	if mlcompat.LikelyBuildxCache(manifestList) {
		return dbPutBuildkitIndex(imh, manifestList, payload)
	}

	repoPath := imh.Repository.Named().Name()
	l := log.GetLogger(log.WithContext(imh)).WithFields(log.Fields{
		"repository":      repoPath,
		"manifest_digest": imh.Digest,
	})
	l.Debug("putting manifest list")

	var opts []datastore.RepositoryStoreOption
	if imh.GetRepoCache() != nil {
		opts = append(opts, datastore.WithRepositoryCache(imh.GetRepoCache()))
	}
	rStore := datastore.NewRepositoryStore(imh.App.db.Primary(), opts...)
	v := validation.NewManifestListValidator(
		&datastore.RepositoryManifestService{
			RepositoryReader: rStore,
			RepositoryPath:   repoPath,
		},
		&datastore.RepositoryBlobService{RepositoryReader: rStore, RepositoryPath: repoPath},
		imh.App.manifestRefLimit,
		imh.App.manifestPayloadSizeLimit,
	)

	if err := v.Validate(imh, manifestList); err != nil {
		return err
	}

	// create or find target repository
	r, err := rStore.CreateOrFindByPath(imh.Context, repoPath)
	if err != nil {
		return err
	}

	ml, err := rStore.FindManifestByDigest(imh.Context, r, imh.Digest)
	if err != nil {
		return err
	}
	if ml != nil {
		return nil
	}

	// Media type can be either Docker (`application/vnd.docker.distribution.manifest.list.v2+json`) or OCI (empty).
	// We need to make it explicit if empty, otherwise we're not able to distinguish between media types.
	mediaType := manifestList.MediaType
	if mediaType == "" {
		mediaType = v1.MediaTypeImageIndex
	}

	ml = &models.Manifest{
		NamespaceID:   r.NamespaceID,
		RepositoryID:  r.ID,
		SchemaVersion: manifestList.SchemaVersion,
		MediaType:     mediaType,
		Digest:        imh.Digest,
		Payload:       payload,
	}

	// We need to find and lock referenced manifests to ensure we lock any related online GC tasks to prevent race
	// conditions around the manifest list insert. See:
	// https://gitlab.com/gitlab-org/container-registry/-/blob/master/docs/spec/gitlab/online-garbage-collection.md#creating-a-manifest-list-referencing-an-unreferenced-manifest
	mm := make([]*models.Manifest, 0, len(manifestList.Manifests))
	ids := make([]int64, 0, len(mm))
	for _, desc := range manifestList.Manifests {
		m, err := dbFindManifestListManifest(imh.Context, rStore, r, desc.Digest, r.Path)
		if err != nil {
			return err
		}
		mm = append(mm, m)
		ids = append(ids, m.ID)
	}

	tx, err := imh.db.Primary().BeginTx(imh.Context, nil)
	if err != nil {
		return fmt.Errorf("creating database transaction: %w", err)
	}
	defer tx.Rollback()

	// Prevent long running transactions by setting an upper limit of manifestListCreateGCLockTimeout. If the GC is
	// holding the lock of a related review record, the processing there should be fast enough to avoid this.
	// Regardless, we should not let transactions open (and clients waiting) for too long. If this sensible timeout
	// is exceeded, abort the request and let the client retry. This will bubble up and lead to a 503 Service
	// Unavailable response.
	ctx, cancel := context.WithTimeout(imh.Context, manifestListCreateGCLockTimeout)
	defer cancel()

	mts := datastore.NewGCManifestTaskStore(tx)
	if _, err := mts.FindAndLockNBefore(ctx, r.NamespaceID, r.ID, ids, time.Now().Add(manifestListCreateGCReviewWindow)); err != nil {
		return err
	}

	// use CreateOrFind to prevent race conditions when the same digest is used by different tags
	// and pushed at the same time https://gitlab.com/gitlab-org/container-registry/-/issues/1132
	mStore := datastore.NewManifestStore(tx)
	if err := mStore.CreateOrFind(imh, ml); err != nil {
		return err
	}

	// Associate manifests to the manifest list.
	for _, m := range mm {
		if err := mStore.AssociateManifest(imh.Context, ml, m); err != nil {
			if errors.Is(err, datastore.ErrRefManifestNotFound) {
				// This can only happen if the online GC deleted one of the referenced manifests (because they were
				// untagged/unreferenced) between the call to `FindAndLockNBefore` and `AssociateManifest`. For now
				// we need to return this error to mimic the behavior of the corresponding filesystem validation.
				return distribution.ErrManifestVerification{
					distribution.ErrManifestBlobUnknown{Digest: m.Digest},
				}
			}
			return err
		}
	}

	if err := tx.Commit(); err != nil {
		return fmt.Errorf("failed to commit database transaction: %w", err)
	}

	return nil
}