func openWithConfig(ctx context.Context, st blob.Storage, lc *LocalConfig, password string, options *Options, cacheOpts *content.CachingOptions, configFile string) (DirectRepository, error) {
	cacheOpts = cacheOpts.CloneOrDefault()
	cmOpts := &content.ManagerOptions{
		TimeNow:            defaultTime(options.TimeNowFunc),
		DisableInternalLog: options.DisableInternalLog,
	}

	var ufb *unpackedFormatBlob

	if _, err := retry.WithExponentialBackoffMaxRetries(ctx, -1, "read repo config and wait for upgrade", func() (interface{}, error) {
		var internalErr error
		ufb, internalErr = readAndCacheRepoConfig(ctx, st, password, cacheOpts,
			lc.FormatBlobCacheDuration)
		if internalErr != nil {
			return nil, internalErr
		}

		// retry if upgrade lock has been taken
		if locked, _ := ufb.repoConfig.UpgradeLock.IsLocked(cmOpts.TimeNow()); locked && options.UpgradeOwnerID != ufb.repoConfig.UpgradeLock.OwnerID {
			return nil, ErrRepositoryUnavailableDueToUpgrageInProgress
		}

		return nil, nil
	}, func(internalErr error) bool {
		return errors.Is(internalErr, ErrRepositoryUnavailableDueToUpgrageInProgress)
	}); err != nil {
		// nolint:wrapcheck
		return nil, err
	}

	cmOpts.RepositoryFormatBytes = ufb.fb

	// Read blobcfg blob, potentially from cache.
	bb, _, err := readAndCacheRepositoryBlobBytes(ctx, st, cacheOpts.CacheDirectory, BlobCfgBlobID, lc.FormatBlobCacheDuration)
	if err != nil && !errors.Is(err, blob.ErrBlobNotFound) {
		return nil, errors.Wrap(err, "unable to read blobcfg blob")
	}

	blobcfg, err := deserializeBlobCfgBytes(ufb.f, bb, ufb.formatEncryptionKey)
	if err != nil {
		return nil, ErrInvalidPassword
	}

	if ufb.repoConfig.FormattingOptions.EnablePasswordChange {
		cacheOpts.HMACSecret = deriveKeyFromMasterKey(ufb.repoConfig.HMACSecret, ufb.f.UniqueID, localCacheIntegrityPurpose, localCacheIntegrityHMACSecretLength)
	} else {
		// deriving from ufb.formatEncryptionKey was actually a bug, that only matters will change when we change the password
		cacheOpts.HMACSecret = deriveKeyFromMasterKey(ufb.formatEncryptionKey, ufb.f.UniqueID, localCacheIntegrityPurpose, localCacheIntegrityHMACSecretLength)
	}

	fo := &ufb.repoConfig.FormattingOptions

	if fo.MaxPackSize == 0 {
		// legacy only, apply default
		fo.MaxPackSize = 20 << 20 // nolint:gomnd
	}

	// do not embed repository format info in pack blobs when password change is enabled.
	if fo.EnablePasswordChange {
		cmOpts.RepositoryFormatBytes = nil
	}

	limits := throttlingLimitsFromConnectionInfo(ctx, st.ConnectionInfo())
	if lc.Throttling != nil {
		limits = *lc.Throttling
	}

	st, throttler, err := addThrottler(st, limits)
	if err != nil {
		return nil, errors.Wrap(err, "unable to add throttler")
	}

	throttler.OnUpdate(func(l throttling.Limits) error {
		lc2, err2 := LoadConfigFromFile(configFile)
		if err2 != nil {
			return err2
		}

		lc2.Throttling = &l

		return lc2.writeToFile(configFile)
	})

	if blobcfg.IsRetentionEnabled() {
		st = wrapLockingStorage(st, blobcfg)
	}

	// background/interleaving upgrade lock storage monitor
	st = upgradeLockMonitor(options.UpgradeOwnerID, st, password, cacheOpts, lc.FormatBlobCacheDuration,
		ufb.cacheMTime, cmOpts.TimeNow)

	scm, err := content.NewSharedManager(ctx, st, fo, cacheOpts, cmOpts)
	if err != nil {
		return nil, errors.Wrap(err, "unable to create shared content manager")
	}

	cm := content.NewWriteManager(ctx, scm, content.SessionOptions{
		SessionUser: lc.Username,
		SessionHost: lc.Hostname,
	}, "")

	om, err := object.NewObjectManager(ctx, cm, ufb.repoConfig.Format)
	if err != nil {
		return nil, errors.Wrap(err, "unable to open object manager")
	}

	manifests, err := manifest.NewManager(ctx, cm, manifest.ManagerOptions{TimeNow: cmOpts.TimeNow})
	if err != nil {
		return nil, errors.Wrap(err, "unable to open manifests")
	}

	dr := &directRepository{
		cmgr:  cm,
		omgr:  om,
		blobs: st,
		mmgr:  manifests,
		sm:    scm,
		directRepositoryParameters: directRepositoryParameters{
			uniqueID:            ufb.f.UniqueID,
			cachingOptions:      *cacheOpts,
			formatBlob:          ufb.f,
			blobCfgBlob:         blobcfg,
			formatEncryptionKey: ufb.formatEncryptionKey,
			timeNow:             cmOpts.TimeNow,
			cliOpts:             lc.ClientOptions.ApplyDefaults(ctx, "Repository in "+st.DisplayName()),
			configFile:          configFile,
			nextWriterID:        new(int32),
			throttler:           throttler,
		},
		closed: make(chan struct{}),
	}

	return dr, nil
}
// openWithConfig opens the repository with a given configuration, avoiding the need for a config file.
// nolint:funlen,gocyclo
openWithConfig opens the repository with a given configuration, avoiding the need for a config file.
nolint:funlen,gocyclo
// openWithConfig opens the repository with a given configuration avoiding the need for a config file . // nolint : funlen gocyclo
func openWithConfig(ctx context.Context, st blob.Storage, lc *LocalConfig, password string, options *Options, cacheOpts *content.CachingOptions, configFile string) (DirectRepository, error) {
	cacheOpts = cacheOpts.CloneOrDefault()
	cmOpts := &content.ManagerOptions{
		TimeNow:            defaultTime(options.TimeNowFunc),
		DisableInternalLog: options.DisableInternalLog,
	}
	var ufb *unpackedFormatBlob
	if _, err := retry.WithExponentialBackoffMaxRetries(ctx, -1, "read repo config and wait for upgrade", func() (interface{}, error) {
		var internalErr error
		ufb, internalErr = readAndCacheRepoConfig(ctx, st, password, cacheOpts,
			lc.FormatBlobCacheDuration)
		if internalErr != nil {
			return nil, internalErr
		}
		if locked, _ := ufb.repoConfig.UpgradeLock.IsLocked(cmOpts.TimeNow()); locked && options.UpgradeOwnerID != ufb.repoConfig.UpgradeLock.OwnerID {
			return nil, ErrRepositoryUnavailableDueToUpgrageInProgress
		}
		return nil, nil
	}, func(internalErr error) bool {
		return errors.Is(internalErr, ErrRepositoryUnavailableDueToUpgrageInProgress)
	}); err != nil {
		return nil, err
	}
	cmOpts.RepositoryFormatBytes = ufb.fb
	bb, _, err := readAndCacheRepositoryBlobBytes(ctx, st, cacheOpts.CacheDirectory, BlobCfgBlobID, lc.FormatBlobCacheDuration)
	if err != nil && !errors.Is(err, blob.ErrBlobNotFound) {
		return nil, errors.Wrap(err, "unable to read blobcfg blob")
	}
	blobcfg, err := deserializeBlobCfgBytes(ufb.f, bb, ufb.formatEncryptionKey)
	if err != nil {
		return nil, ErrInvalidPassword
	}
	if ufb.repoConfig.FormattingOptions.EnablePasswordChange {
		cacheOpts.HMACSecret = deriveKeyFromMasterKey(ufb.repoConfig.HMACSecret, ufb.f.UniqueID, localCacheIntegrityPurpose, localCacheIntegrityHMACSecretLength)
	} else {
		cacheOpts.HMACSecret = deriveKeyFromMasterKey(ufb.formatEncryptionKey, ufb.f.UniqueID, localCacheIntegrityPurpose, localCacheIntegrityHMACSecretLength)
	}
	fo := &ufb.repoConfig.FormattingOptions
	if fo.MaxPackSize == 0 {
		fo.MaxPackSize = 20 << 20 
	}
	if fo.EnablePasswordChange {
		cmOpts.RepositoryFormatBytes = nil
	}
	limits := throttlingLimitsFromConnectionInfo(ctx, st.ConnectionInfo())
	if lc.Throttling != nil {
		limits = *lc.Throttling
	}
	st, throttler, err := addThrottler(st, limits)
	if err != nil {
		return nil, errors.Wrap(err, "unable to add throttler")
	}
	throttler.OnUpdate(func(l throttling.Limits) error {
		lc2, err2 := LoadConfigFromFile(configFile)
		if err2 != nil {
			return err2
		}
		lc2.Throttling = &l
		return lc2.writeToFile(configFile)
	})
	if blobcfg.IsRetentionEnabled() {
		st = wrapLockingStorage(st, blobcfg)
	}
	st = upgradeLockMonitor(options.UpgradeOwnerID, st, password, cacheOpts, lc.FormatBlobCacheDuration,
		ufb.cacheMTime, cmOpts.TimeNow)
	scm, err := content.NewSharedManager(ctx, st, fo, cacheOpts, cmOpts)
	if err != nil {
		return nil, errors.Wrap(err, "unable to create shared content manager")
	}
	cm := content.NewWriteManager(ctx, scm, content.SessionOptions{
		SessionUser: lc.Username,
		SessionHost: lc.Hostname,
	}, "")
	om, err := object.NewObjectManager(ctx, cm, ufb.repoConfig.Format)
	if err != nil {
		return nil, errors.Wrap(err, "unable to open object manager")
	}
	manifests, err := manifest.NewManager(ctx, cm, manifest.ManagerOptions{TimeNow: cmOpts.TimeNow})
	if err != nil {
		return nil, errors.Wrap(err, "unable to open manifests")
	}
	dr := &directRepository{
		cmgr:  cm,
		omgr:  om,
		blobs: st,
		mmgr:  manifests,
		sm:    scm,
		directRepositoryParameters: directRepositoryParameters{
			uniqueID:            ufb.f.UniqueID,
			cachingOptions:      *cacheOpts,
			formatBlob:          ufb.f,
			blobCfgBlob:         blobcfg,
			formatEncryptionKey: ufb.formatEncryptionKey,
			timeNow:             cmOpts.TimeNow,
			cliOpts:             lc.ClientOptions.ApplyDefaults(ctx, "Repository in "+st.DisplayName()),
			configFile:          configFile,
			nextWriterID:        new(int32),
			throttler:           throttler,
		},
		closed: make(chan struct{}),
	}
	return dr, nil
}
func openWithConfig ( ctx context . Context , st blob . Storage , lc * LocalConfig , password string , options * Options , cacheOpts * content . CachingOptions , configFile string ) ( DirectRepository , error ) { cacheOpts = cacheOpts . CloneOrDefault ( ) cmOpts := & content . ManagerOptions { TimeNow : defaultTime ( options . TimeNowFunc ) , DisableInternalLog : options . DisableInternalLog , } var ufb * unpackedFormatBlob if _ , err := retry . WithExponentialBackoffMaxRetries ( ctx , - 1 , " " , func ( ) ( interface { } , error ) { var internalErr error ufb , internalErr = readAndCacheRepoConfig ( ctx , st , password , cacheOpts , lc . FormatBlobCacheDuration ) if internalErr != nil { return nil , internalErr } if locked , _ := ufb . repoConfig . UpgradeLock . IsLocked ( cmOpts . TimeNow ( ) ) ; locked && options . UpgradeOwnerID != ufb . repoConfig . UpgradeLock . OwnerID { return nil , ErrRepositoryUnavailableDueToUpgrageInProgress } return nil , nil } , func ( internalErr error ) bool { return errors . Is ( internalErr , ErrRepositoryUnavailableDueToUpgrageInProgress ) } ) ; err != nil { return nil , err } cmOpts . RepositoryFormatBytes = ufb . fb bb , _ , err := readAndCacheRepositoryBlobBytes ( ctx , st , cacheOpts . CacheDirectory , BlobCfgBlobID , lc . FormatBlobCacheDuration ) if err != nil && ! errors . Is ( err , blob . ErrBlobNotFound ) { return nil , errors . Wrap ( err , " " ) } blobcfg , err := deserializeBlobCfgBytes ( ufb . f , bb , ufb . formatEncryptionKey ) if err != nil { return nil , ErrInvalidPassword } if ufb . repoConfig . FormattingOptions . EnablePasswordChange { cacheOpts . HMACSecret = deriveKeyFromMasterKey ( ufb . repoConfig . HMACSecret , ufb . f . UniqueID , localCacheIntegrityPurpose , localCacheIntegrityHMACSecretLength ) } else { cacheOpts . HMACSecret = deriveKeyFromMasterKey ( ufb . formatEncryptionKey , ufb . f . UniqueID , localCacheIntegrityPurpose , localCacheIntegrityHMACSecretLength ) } fo := & ufb . repoConfig . FormattingOptions if fo . MaxPackSize == 0 { fo . MaxPackSize = 20 << 20 } if fo . EnablePasswordChange { cmOpts . RepositoryFormatBytes = nil } limits := throttlingLimitsFromConnectionInfo ( ctx , st . ConnectionInfo ( ) ) if lc . Throttling != nil { limits = * lc . Throttling } st , throttler , err := addThrottler ( st , limits ) if err != nil { return nil , errors . Wrap ( err , " " ) } throttler . OnUpdate ( func ( l throttling . Limits ) error { lc2 , err2 := LoadConfigFromFile ( configFile ) if err2 != nil { return err2 } lc2 . Throttling = & l return lc2 . writeToFile ( configFile ) } ) if blobcfg . IsRetentionEnabled ( ) { st = wrapLockingStorage ( st , blobcfg ) } st = upgradeLockMonitor ( options . UpgradeOwnerID , st , password , cacheOpts , lc . FormatBlobCacheDuration , ufb . cacheMTime , cmOpts . TimeNow ) scm , err := content . NewSharedManager ( ctx , st , fo , cacheOpts , cmOpts ) if err != nil { return nil , errors . Wrap ( err , " " ) } cm := content . NewWriteManager ( ctx , scm , content . SessionOptions { SessionUser : lc . Username , SessionHost : lc . Hostname , } , " " ) om , err := object . NewObjectManager ( ctx , cm , ufb . repoConfig . Format ) if err != nil { return nil , errors . Wrap ( err , " " ) } manifests , err := manifest . NewManager ( ctx , cm , manifest . ManagerOptions { TimeNow : cmOpts . TimeNow } ) if err != nil { return nil , errors . Wrap ( err , " " ) } dr := & directRepository { cmgr : cm , omgr : om , blobs : st , mmgr : manifests , sm : scm , directRepositoryParameters : directRepositoryParameters { uniqueID : ufb . f . UniqueID , cachingOptions : * cacheOpts , formatBlob : ufb . f , blobCfgBlob : blobcfg , formatEncryptionKey : ufb . formatEncryptionKey , timeNow : cmOpts . TimeNow , cliOpts : lc . ClientOptions . ApplyDefaults ( ctx , " " + st . DisplayName ( ) ) , configFile : configFile , nextWriterID : new ( int32 ) , throttler : throttler , } , closed : make ( chan struct { } ) , } return dr , nil }
openWithConfig opens the repository with a given configuration, avoiding the need for a config file.
openWithConfig opens the repository with a given configuration avoiding the need for a config file .
// retry if upgrade lock has been taken // nolint:wrapcheck // Read blobcfg blob, potentially from cache. // deriving from ufb.formatEncryptionKey was actually a bug, that only matters will change when we change the password // legacy only, apply default // nolint:gomnd // do not embed repository format info in pack blobs when password change is enabled. // background/interleaving upgrade lock storage monitor