Repository: operator-framework/operator-lifecycle-manager
Pull request #2942 (2023-03-22 11:34:54)

Prevent OLM from creating `InstallPlan`s when bundle unpack fails
Prerequisites:
* https://github.com/operator-framework/api/pull/282
* https://github.com/operator-framework/operator-lifecycle-manager/pull/2954
* https://github.com/operator-framework/operator-lifecycle-manager/pull/2952

---

<!--

Before making a PR, please read our contributing guidelines https://github.com/operator-framework/operator-lifecycle-manager/blob/master/CONTRIBUTING.md

Note: Make sure your branch is rebased to the latest upstream master.

-->

**Description of the change:**

This PR changes how `InstallPlan`s are being created.

**Motivation for the change:**

Today OLM creates InstallPlans even if bundle images are not available for some reason. For example, catalog source contains references to an unreachable registry.

Consider this scenario:
* User installs an operator and enables auto upgrades
* Catalog source receives an update which references unreachable bundle images (e.g. pre-release from a private registry)
* OLM creates an `InstallPlan` for an upgrade which features unreachable bundle images
* At this point upgrade is stuck since the cluster can not pull bundle images
* Catalog source receives a corrective update which contains only reachable bundle images

Workaround: to resolve this issue users have to manually remove a failed `InstallPlan`. After that OLM is expected to create a new `InstallPlan` based on the latest state of catalog sources.

The goal: We would like to find a solution which prevents `InstallPlan` from being created with unreachable images in the first place. This eliminates the need to manually delete `InstallPlan` in scenarios like the one described above.

**Architectural changes:**

<!--
If necessary, briefly describe any architectural changes, other options considered, and/or link to any EPs or design docs
-->

Before the change:

* On a `Namespace` reconciliation (`syncResolvingNamespace`) OLM creates an `InstallPlan` if required (e.g. new sub is created or a new update in a catalog is available for an existing sub)
* On `InstallPlan` reconciliation (`syncInstallPlans`) OLM creates a `Job` to unpack a relevant bundle.
  * On unsuccessful job unpacking OLM reports failures as `InstallPlan` conditions

After this change:

* On a `Namespace` reconciliation (`syncResolvingNamespace`):
  * OLM first creates a `Job` to unpack a relevant bundle and waits for the job to sucessfully complete
    * On unsuccessful job unpacking OLM reports failures as `Subscription` conditions (set on all subs in the namespace)
  * Only after successfull completion OLM creates an `InstallPlan`



**Testing remarks:**

To reproduce:

* Create a test cluster and install OLM (`make run-local` from this repo can be helpful)
* Create a catalog source:
    ```yaml
    apiVersion: operators.coreos.com/v1alpha1
    kind: CatalogSource
    metadata:
      name: alex-operators
      namespace: olm
    spec:
      image: quay.io/agreene/index:unreachable-bundle-upgrade
      displayName: Alex Operators
      priority: -100
      publisher: Alex
      sourceType: grpc
      updateStrategy:
        registryPoll:
          interval: 10m0s
    ```
* And create a subscription:
    ```yaml
    apiVersion: operators.coreos.com/v1alpha1
    kind: Subscription
    metadata:
      name: quay
      namespace: operators
    spec:
      channel: unavailable-image-upgrade
      installPlanApproval: Automatic
      name: quay
      source: alex-operators
      sourceNamespace: olm
      startingCSV: quay-operator.v3.8.3
    ```

In master - you will see two `InstallPlan`s: one for `quay-operator.v3.8.3` (successfull install) and one for `quay-operator.v3.8.4` which will eventually (10min by default) report failure to unpack in bundle lookup conditions.

In this branch you will see only one `InstallPlan` for `quay-operator.v3.8.3` (successfull install). Eventually `Subscription` will get a condition reporting failure to unpack (`SubscriptionBundleUnpackFailed`). During unpacking it should report `UnpackingInProgress`.

**Reviewer Checklist**
- [ ] Implementation matches the proposed design, or proposal is updated to match implementation
- [ ] Sufficient unit test coverage
- [ ] Sufficient end-to-end test coverage
- [ ] Bug fixes are accompanied by regression test(s)
- [ ] e2e tests and flake fixes are accompanied evidence of flake testing, e.g. executing the test 100(0) times
- [ ] tech debt/todo is accompanied by issue link(s) in comments in the surrounding code
- [ ] Tests are comprehensible, e.g. Ginkgo DSL is being used appropriately
- [ ] Docs updated or added to `/doc`
- [ ] Commit messages sensible and descriptive
- [ ] Tests marked as `[FLAKE]` are truly flaky and have an issue
- [ ] Code is properly formatted


<!--

Note: If this PR is fixing an issue make sure to add a note saying:
Closes #<ISSUE_NUMBER>

-->


openshift-ci[bot]: Skipping CI for Draft Pull Request.
If you want CI signal for your change, please convert it to an actual PR.
You can still manually trigger a test run with `/test all`
openshift-ci[bot]: [APPROVALNOTIFIER] This PR is **NOT APPROVED**

This pull-request has been approved by: *<a href="https://github.com/operator-framework/operator-lifecycle-manager/pull/2942#" title="Author self-approved">m1kola</a>*
**Once this PR has been reviewed and has the lgtm label**, please assign benluddy for approval. For more information see [the Kubernetes Code Review Process](https://git.k8s.io/community/contributors/guide/owners.md#the-code-review-process).

The full list of commands accepted by this bot can be found [here](https://go.k8s.io/bot-commands?repo=operator-framework%2Foperator-lifecycle-manager).

<details open>
Needs approval from an approver in each of these files:

- **[OWNERS](https://github.com/operator-framework/operator-lifecycle-manager/blob/master/OWNERS)**

Approvers can indicate their approval by writing `/approve` in a comment
Approvers can cancel approval by writing `/approve cancel` in a comment
</details>
<!-- META={"approvers":["benluddy"]} -->
openshift-merge-robot: PR needs rebase.

<details>

Instructions for interacting with me using PR comments are available [here](https://git.k8s.io/community/contributors/guide/pull-requests.md).  If you have questions or suggestions related to my behavior, please file an issue against the [kubernetes/test-infra](https://github.com/kubernetes/test-infra/issues/new?title=Prow%20issue:) repository.
</details>
dtfranz: Looks good! Had a couple questions. 

You mentioned above:
> Only after successfull completion OLM creates an InstallPlan

In the example scenario then, where the CatalogSource is updated again with correct images, will OLM self-heal?

In the case where the bundle unpack is successful and then the following `InstallPlan` fails for an unrelated reason, what happens to the bundle? Can it be re-used on subsequent `InstallPlan` attempts, or will it need to be unpacked again on each retry?
m1kola: > In the example scenario then, where the CatalogSource is updated again with correct images, will OLM self-heal?

@dtfranz the answer is yes, but... I was going to point you to a e2e test demonstrating this and it looks like it is missing. I was sure that I covered it, but it looks like I didn't. I'll look into adding an e2e to demonstrate this. Good catch! 👍

I'll _probably_ submit it as a follow up PR. Given this PR already has decent coverage and is already too big it might be good idea to do that separately. In any way - I'll let you know once I add extra coverage here or elsewhere.

> In the case where the bundle unpack is successful and then the following `InstallPlan` fails for an unrelated reason, what happens to the bundle? Can it be re-used on subsequent `InstallPlan` attempts, or will it need to be unpacked again on each retry?

I did not touch the bit concerning unpack jobs (not directly at least), but I believe you are correct: unpack jobs are being reused. See in `pkg/controller/bundle`:
* Before creating a new job we first check whether job already exists by name [here](https://github.com/operator-framework/operator-lifecycle-manager/blob/cf13b7731a2a272f2eb38a5bef3b02c5fdca50cf/pkg/controller/bundle/bundle_unpacker.go#L628-L635)
* Name of the job comes from the name of config map via [this](https://github.com/operator-framework/operator-lifecycle-manager/blob/cf13b7731a2a272f2eb38a5bef3b02c5fdca50cf/pkg/controller/bundle/bundle_unpacker.go#L467) ref
* And the config name gets its name from [this](https://github.com/operator-framework/operator-lifecycle-manager/blob/cf13b7731a2a272f2eb38a5bef3b02c5fdca50cf/pkg/controller/bundle/bundle_unpacker.go#L87) hash which is based on bundle lookup path
* So as long as bundle lookup path (image reference) stays the same - jobs should be reusable

dtfranz: > I'll _probably_ submit it as a follow up PR. Given this PR already has decent coverage and is already too big it might be good idea to do that separately. In any way - I'll let you know once I add extra coverage here or elsewhere.

I agree with you there, I'm fine with this being submitted in a follow-up. Thanks for answering those questions as well!
/lgtm
m1kola: Putting this on hold as I had several unexpected `ConstraintsNotSatisfiable` while looking into additional E2E coverage.

/hold
openshift-ci[bot]: New changes are detected. LGTM label has been removed.
m1kola: Fixed the issue mentioned above (details are [here](https://github.com/operator-framework/operator-lifecycle-manager/pull/2942#discussion_r1186045060)). Extra E2E coverage added in a [follow up PR](https://github.com/operator-framework/operator-lifecycle-manager/pull/2961) just to not make things even more complicated in this one.

/hold cancel
dtfranz: /lgtm
m1kola: Had to rebase since something else got merged into master

---

@pgodowski thank you for the example. Very helpful. I think I mislead you on the community call yesterday because I misunderstood what you were asking about. I will try to explain the behaviour for this case better now.

> catalog source was updated with the GA content which surely contains valid operator bundles and then for some reason, the unpack Job fails.
> Today when unpack Job fails, we have to manually remove it.

Nothing changes in this scenario apart from the fact that there will be no `InstallPlan` created now. Failure will be reported on a subscription via the `BundleUnpackFailed` condition. You will still have to take a manual action in cases like this as `DeadlineExceeded` might indicate a legit issue (for example, cluster gets 403 when pulling a bundle image).

Your options are:

* Increase unpack timeout using `--bundle-unpack-timeout` (10m by default). This might be helpful if you are aware of a reason why unpacking takes longer than default timeout on your clusters. Might happen due to a variety of reasons such as insufficient resources for unpacking Pod or slow image pulling due to unreliable networking, etc. Increased timeout will allow Job to stay alive for longer period of time. This timeout translates into [Job's `.spec.activeDeadlineSeconds`][activeDeadlineSeconds]. Keep in mind that OLM sets `.spec.backoffLimit` to 3 still. 
* Keep removing the Job manually. OLM will recreated the Job for the bundle on the next reconciliation.

However once this PR merged it will be easier to introduce logic where we can instruct OLM to recreated unpack failed jobs automatically. E.g. "recreated failed unpack jobs that are older than 1 day". It will be easier because we won't have to worry about transitioning `InstallPlan` from the failed terminal state into non-terminal state since after this PR we won't be creating `InstallPlans` at all. I think it will come down to modifying [`ensureJob` here][ensureJob] + adding a param. But I personally not a big fan of this idea as it sounds like hiding root cause behind the retries.

It might be a good idea to create an issue for this and include details of root cause(s) for unpack jobs failures in your clusters.

To summarise - We are not trying to improve this case here. This PR aims at solving an issue when catalog source gets updated with a corrected references to bundle images.



[activeDeadlineSeconds]: https://kubernetes.io/docs/concepts/workloads/controllers/job/#job-termination-and-cleanup
[ensureJob]: https://github.com/operator-framework/operator-lifecycle-manager/blob/858c7cdb2f4cfe3d351225972aede54e3b114ce4/pkg/controller/bundle/bundle_unpacker.go#L626-L645
pgodowski: > To summarise - We are not trying to improve this case here. This PR aims at solving an issue when catalog source gets updated with a corrected references to bundle images.

Understood - thanks.
perdasilva: @m1kola I've rebased from this side
openshift-ci[bot]: [APPROVALNOTIFIER] This PR is **NOT APPROVED**

This pull-request has been approved by: *<a href="https://github.com/operator-framework/operator-lifecycle-manager/pull/2942#" title="Author self-approved">m1kola</a>*, *<a href="https://github.com/operator-framework/operator-lifecycle-manager/pull/2942#pullrequestreview-1437296933" title="Approved">tmshort</a>*
**Once this PR has been reviewed and has the lgtm label**, please assign benluddy for approval. For more information see [the Kubernetes Code Review Process](https://git.k8s.io/community/contributors/guide/owners.md#the-code-review-process).

The full list of commands accepted by this bot can be found [here](https://go.k8s.io/bot-commands?repo=operator-framework%2Foperator-lifecycle-manager).

<details open>
Needs approval from an approver in each of these files:

- **[OWNERS](https://github.com/operator-framework/operator-lifecycle-manager/blob/master/OWNERS)**

Approvers can indicate their approval by writing `/approve` in a comment
Approvers can cancel approval by writing `/approve cancel` in a comment
</details>
<!-- META={"approvers":["benluddy"]} -->
tmshort: /lgtm
openshift-ci[bot]: New changes are detected. LGTM label has been removed.
m1kola: I updated to address feedback.

Here is the short summary of the diff since the last push which ([diff can be found here](https://github.com/operator-framework/operator-lifecycle-manager/compare/67b8fc69b1d1945a4243f4d6110ddfcbee5b6e62..9ed3d8363609854f272963e4781e67548bcaf8f4)). It is arguably easier to comparae modified files to the master still.

* Updated `test/e2e/fail_forward_e2e_test.go` as per descusion [here](https://github.com/operator-framework/operator-lifecycle-manager/pull/2942#discussion_r1196646809): reverted (with necessary changes) to having failed `InstallPlan`.
* Updated `test/e2e/subscription_e2e_test.go` to add extra coverage to show how OLM auto-recovers as per [this](https://github.com/operator-framework/operator-lifecycle-manager/pull/2942#issuecomment-1527315191). Initially I wanted it to be a separate PR (#2961) but to address fail forward I needed extra test data file which I already had in that separate PR. So I squashed them together here.
* Rearanged test data in `test/e2e/testdata/fail-forward/` for above two to work.

No changes to the controller itself. Only tests.

I hate large PRs like this, but changes changes in `test/e2e/fail_forward_e2e_test.go` were necessary to ensure that we do not merge something which fails to catch a regression in fail forward feature. And it dragged other things in.

@tmshort @dtfranz @pgodowski @ankitathomas @perdasilva please take another look.
openshift-ci[bot]: [APPROVALNOTIFIER] This PR is **APPROVED**

This pull-request has been approved by: *<a href="https://github.com/operator-framework/operator-lifecycle-manager/pull/2942#" title="Author self-approved">m1kola</a>*, *<a href="https://github.com/operator-framework/operator-lifecycle-manager/pull/2942#pullrequestreview-1446111501" title="Approved">perdasilva</a>*, *<a href="https://github.com/operator-framework/operator-lifecycle-manager/pull/2942#pullrequestreview-1437296933" title="Approved">tmshort</a>*

The full list of commands accepted by this bot can be found [here](https://go.k8s.io/bot-commands?repo=operator-framework%2Foperator-lifecycle-manager).

The pull request process is described [here](https://git.k8s.io/community/contributors/guide/owners.md#the-code-review-process)

<details >
Needs approval from an approver in each of these files:

- ~~[OWNERS](https://github.com/operator-framework/operator-lifecycle-manager/blob/master/OWNERS)~~ [perdasilva]

Approvers can indicate their approval by writing `/approve` in a comment
Approvers can cancel approval by writing `/approve cancel` in a comment
</details>
<!-- META={"approvers":[]} -->
openshift-ci[bot]: [APPROVALNOTIFIER] This PR is **APPROVED**

This pull-request has been approved by: *<a href="https://github.com/operator-framework/operator-lifecycle-manager/pull/2942#" title="Author self-approved">m1kola</a>*, *<a href="https://github.com/operator-framework/operator-lifecycle-manager/pull/2942#pullrequestreview-1446111501" title="Approved">perdasilva</a>*, *<a href="https://github.com/operator-framework/operator-lifecycle-manager/pull/2942#pullrequestreview-1437296933" title="Approved">tmshort</a>*

The full list of commands accepted by this bot can be found [here](https://go.k8s.io/bot-commands?repo=operator-framework%2Foperator-lifecycle-manager).

The pull request process is described [here](https://git.k8s.io/community/contributors/guide/owners.md#the-code-review-process)

<details >
Needs approval from an approver in each of these files:

- ~~[OWNERS](https://github.com/operator-framework/operator-lifecycle-manager/blob/master/OWNERS)~~ [perdasilva]

Approvers can indicate their approval by writing `/approve` in a comment
Approvers can cancel approval by writing `/approve cancel` in a comment
</details>
<!-- META={"approvers":[]} -->

Patch:
diff --git a/pkg/controller/bundle/bundle_unpacker.go b/pkg/controller/bundle/bundle_unpacker.go
index 28bd61fb2..c7b71c067 100644
--- a/pkg/controller/bundle/bundle_unpacker.go
+++ b/pkg/controller/bundle/bundle_unpacker.go
@@ -35,7 +35,7 @@ import (
 
 const (
 	// TODO: This can be a spec field
-	// BundleUnpackTimeoutAnnotationKey allows setting a bundle unpack timeout per InstallPlan
+	// BundleUnpackTimeoutAnnotationKey allows setting a bundle unpack timeout per OperatorGroup
 	// and overrides the default specified by the --bundle-unpack-timeout flag
 	// The time duration should be in the same format as accepted by time.ParseDuration()
 	// e.g 1m30s
diff --git a/pkg/controller/operators/catalog/operator.go b/pkg/controller/operators/catalog/operator.go
index 6b0cde59b..0d24b363d 100644
--- a/pkg/controller/operators/catalog/operator.go
+++ b/pkg/controller/operators/catalog/operator.go
@@ -934,7 +934,13 @@ func (o *Operator) syncResolvingNamespace(obj interface{}) error {
 		return err
 	}
 
-	failForwardEnabled, err := resolver.IsFailForwardEnabled(o.lister.OperatorsV1().OperatorGroupLister().OperatorGroups(namespace))
+	ogLister := o.lister.OperatorsV1().OperatorGroupLister().OperatorGroups(namespace)
+	failForwardEnabled, err := resolver.IsFailForwardEnabled(ogLister)
+	if err != nil {
+		return err
+	}
+
+	unpackTimeout, err := bundle.OperatorGroupBundleUnpackTimeout(ogLister)
 	if err != nil {
 		return err
 	}
@@ -1028,6 +1034,80 @@ func (o *Operator) syncResolvingNamespace(obj interface{}) error {
 		return err
 	}
 
+	// Attempt to unpack bundles before installing
+	// Note: This should probably use the attenuated client to prevent users from resolving resources they otherwise don't have access to.
+	if len(bundleLookups) > 0 {
+		logger.Debug("unpacking bundles")
+
+		var unpacked bool
+		unpacked, steps, bundleLookups, err = o.unpackBundles(namespace, steps, bundleLookups, unpackTimeout)
+		if err != nil {
+			// If the error was fatal capture and fail
+			if olmerrors.IsFatal(err) {
+				_, updateErr := o.updateSubscriptionStatuses(
+					o.setSubsCond(subs, v1alpha1.SubscriptionCondition{
+						Type:    v1alpha1.SubscriptionBundleUnpackFailed,
+						Reason:  "ErrorPreventedUnpacking",
+						Message: err.Error(),
+						Status:  corev1.ConditionTrue,
+					}))
+				if updateErr != nil {
+					logger.WithError(updateErr).Debug("failed to update subs conditions")
+					return updateErr
+				}
+				return nil
+			}
+			// Retry sync if non-fatal error
+			return fmt.Errorf("bundle unpacking failed with an error: %w", err)
+		}
+
+		// Check BundleLookup status conditions to see if the BundleLookupFailed condtion is true
+		// which means bundle lookup has failed and subscriptions need to be updated
+		// with a condition indicating the failure.
+		isFailed, cond := hasBundleLookupFailureCondition(bundleLookups)
+		if isFailed {
+			err := fmt.Errorf("bundle unpacking failed. Reason: %v, and Message: %v", cond.Reason, cond.Message)
+			logger.Infof("%v", err)
+
+			_, updateErr := o.updateSubscriptionStatuses(
+				o.setSubsCond(subs, v1alpha1.SubscriptionCondition{
+					Type:    v1alpha1.SubscriptionBundleUnpackFailed,
+					Reason:  "BundleUnpackFailed",
+					Message: err.Error(),
+					Status:  corev1.ConditionTrue,
+				}))
+			if updateErr != nil {
+				logger.WithError(updateErr).Debug("failed to update subs conditions")
+				return updateErr
+			}
+			// Since this is likely requires intervention we do not want to
+			// requeue too often. We return no error here and rely on a
+			// periodic resync which will help to automatically resolve
+			// some issues such as unreachable bundle images caused by
+			// bad catalog updates.
+			return nil
+		}
+
+		// This means that the unpack job is still running (most likely) or
+		// there was some issue which we did not handle above.
+		if !unpacked {
+			_, updateErr := o.updateSubscriptionStatuses(
+				o.setSubsCond(subs, v1alpha1.SubscriptionCondition{
+					Type:   v1alpha1.SubscriptionBundleUnpacking,
+					Reason: "UnpackingInProgress",
+					Status: corev1.ConditionTrue,
+				}))
+			if updateErr != nil {
+				logger.WithError(updateErr).Debug("failed to update subs conditions")
+				return updateErr
+			}
+
+			logger.Debug("unpacking is not complete yet, requeueing")
+			o.nsResolveQueue.AddAfter(namespace, 5*time.Second)
+			return nil
+		}
+	}
+
 	// create installplan if anything updated
 	if len(updatedSubs) > 0 {
 		logger.Debug("resolution caused subscription changes, creating installplan")
@@ -1062,8 +1142,17 @@ func (o *Operator) syncResolvingNamespace(obj interface{}) error {
 		logger.Debugf("no subscriptions were updated")
 	}
 
+	// Make sure that we no longer indicate unpacking progress
+	subs = o.setSubsCond(subs, v1alpha1.SubscriptionCondition{
+		Type:   v1alpha1.SubscriptionBundleUnpacking,
+		Status: corev1.ConditionFalse,
+	})
+
+	// Remove BundleUnpackFailed condition from subscriptions
+	o.removeSubsCond(subs, v1alpha1.SubscriptionBundleUnpackFailed)
+
 	// Remove resolutionfailed condition from subscriptions
-	subs = o.removeSubsCond(subs, v1alpha1.SubscriptionResolutionFailed)
+	o.removeSubsCond(subs, v1alpha1.SubscriptionResolutionFailed)
 	newSub := true
 	for _, updatedSub := range updatedSubs {
 		updatedSub.Status.RemoveConditions(v1alpha1.SubscriptionResolutionFailed)
@@ -1408,19 +1497,27 @@ type UnpackedBundleReference struct {
 	Properties             string `json:"properties"`
 }
 
-func (o *Operator) unpackBundles(plan *v1alpha1.InstallPlan, unpackTimeout time.Duration) (bool, *v1alpha1.InstallPlan, error) {
-	out := plan.DeepCopy()
+func (o *Operator) unpackBundles(namespace string, installPlanSteps []*v1alpha1.Step, bundleLookups []v1alpha1.BundleLookup, unpackTimeout time.Duration) (bool, []*v1alpha1.Step, []v1alpha1.BundleLookup, error) {
 	unpacked := true
 
+	outBundleLookups := make([]v1alpha1.BundleLookup, len(bundleLookups))
+	for i := range bundleLookups {
+		bundleLookups[i].DeepCopyInto(&outBundleLookups[i])
+	}
+	outInstallPlanSteps := make([]*v1alpha1.Step, len(installPlanSteps))
+	for i := range installPlanSteps {
+		outInstallPlanSteps[i] = installPlanSteps[i].DeepCopy()
+	}
+
 	var errs []error
-	for i := 0; i < len(out.Status.BundleLookups); i++ {
-		lookup := out.Status.BundleLookups[i]
+	for i := 0; i < len(outBundleLookups); i++ {
+		lookup := outBundleLookups[i]
 		res, err := o.bundleUnpacker.UnpackBundle(&lookup, unpackTimeout)
 		if err != nil {
 			errs = append(errs, err)
 			continue
 		}
-		out.Status.BundleLookups[i] = *res.BundleLookup
+		outBundleLookups[i] = *res.BundleLookup
 
 		// if the failed condition is present it means the bundle unpacking has failed
 		failedCondition := res.GetCondition(v1alpha1.BundleLookupFailed)
@@ -1442,11 +1539,11 @@ func (o *Operator) unpackBundles(plan *v1alpha1.InstallPlan, unpackTimeout time.
 			continue
 		}
 
-		// Ensure that bundle can be applied by the current version of OLM by converting to steps
-		steps, err := resolver.NewStepsFromBundle(res.Bundle(), out.GetNamespace(), res.Replaces, res.CatalogSourceRef.Name, res.CatalogSourceRef.Namespace)
+		// Ensure that bundle can be applied by the current version of OLM by converting to bundleSteps
+		bundleSteps, err := resolver.NewStepsFromBundle(res.Bundle(), namespace, res.Replaces, res.CatalogSourceRef.Name, res.CatalogSourceRef.Namespace)
 		if err != nil {
 			if fatal := olmerrors.IsFatal(err); fatal {
-				return false, nil, err
+				return false, nil, nil, err
 			}
 
 			errs = append(errs, fmt.Errorf("failed to turn bundle into steps: %v", err))
@@ -1455,7 +1552,7 @@ func (o *Operator) unpackBundles(plan *v1alpha1.InstallPlan, unpackTimeout time.
 		}
 
 		// step manifests are replaced with references to the configmap containing them
-		for i, s := range steps {
+		for i, s := range bundleSteps {
 			ref := UnpackedBundleReference{
 				Kind:                   "ConfigMap",
 				Namespace:              res.CatalogSourceRef.Namespace,
@@ -1472,19 +1569,19 @@ func (o *Operator) unpackBundles(plan *v1alpha1.InstallPlan, unpackTimeout time.
 				continue
 			}
 			s.Resource.Manifest = string(r)
-			steps[i] = s
+			bundleSteps[i] = s
 		}
 		res.RemoveCondition(resolver.BundleLookupConditionPacked)
-		out.Status.BundleLookups[i] = *res.BundleLookup
-		out.Status.Plan = append(out.Status.Plan, steps...)
+		outBundleLookups[i] = *res.BundleLookup
+		outInstallPlanSteps = append(outInstallPlanSteps, bundleSteps...)
 	}
 
 	if err := utilerrors.NewAggregate(errs); err != nil {
 		o.logger.Debugf("failed to unpack bundles: %v", err)
-		return false, nil, err
+		return false, nil, nil, err
 	}
 
-	return unpacked, out, nil
+	return unpacked, outInstallPlanSteps, outBundleLookups, nil
 }
 
 // gcInstallPlans garbage collects installplans that are too old
@@ -1631,71 +1728,6 @@ func (o *Operator) syncInstallPlans(obj interface{}) (syncError error) {
 		}
 	}
 
-	ogLister := o.lister.OperatorsV1().OperatorGroupLister().OperatorGroups(plan.GetNamespace())
-	unpackTimeout, err := bundle.OperatorGroupBundleUnpackTimeout(ogLister)
-	if err != nil {
-		return err
-	}
-
-	// Attempt to unpack bundles before installing
-	// Note: This should probably use the attenuated client to prevent users from resolving resources they otherwise don't have access to.
-	if len(plan.Status.BundleLookups) > 0 {
-		unpacked, out, err := o.unpackBundles(plan, unpackTimeout)
-		if err != nil {
-			// If the error was fatal capture and fail
-			if fatal := olmerrors.IsFatal(err); fatal {
-				if err := o.transitionInstallPlanToFailed(plan, logger, v1alpha1.InstallPlanReasonInstallCheckFailed, err.Error()); err != nil {
-					// retry for failure to update status
-					syncError = err
-					return
-				}
-			}
-			// Retry sync if non-fatal error
-			syncError = fmt.Errorf("bundle unpacking failed: %v", err)
-			return
-		}
-
-		if !reflect.DeepEqual(plan.Status, out.Status) {
-			logger.Warnf("status not equal, updating...")
-			if _, err := o.client.OperatorsV1alpha1().InstallPlans(out.GetNamespace()).UpdateStatus(context.TODO(), out, metav1.UpdateOptions{}); err != nil {
-				syncError = fmt.Errorf("failed to update installplan bundle lookups: %v", err)
-			}
-
-			return
-		}
-
-		// Check BundleLookup status conditions to see if the BundleLookupPending condtion is false
-		// which means bundle lookup has failed and the InstallPlan should be failed as well
-		isFailed, cond := hasBundleLookupFailureCondition(plan)
-		if isFailed {
-			err := fmt.Errorf("bundle unpacking failed. Reason: %v, and Message: %v", cond.Reason, cond.Message)
-			// Mark the InstallPlan as failed for a fatal bundle unpack error
-			logger.Infof("%v", err)
-
-			if err := o.transitionInstallPlanToFailed(plan, logger, v1alpha1.InstallPlanReasonInstallCheckFailed, err.Error()); err != nil {
-				// retry for failure to update status
-				syncError = err
-				return
-			}
-
-			// Requeue subscription to propagate SubscriptionInstallPlanFailed condtion to subscription
-			o.requeueSubscriptionForInstallPlan(plan, logger)
-			return
-		}
-
-		// TODO: Remove in favor of job and configmap informer requeuing
-		if !unpacked {
-			err := o.ipQueueSet.RequeueAfter(plan.GetNamespace(), plan.GetName(), 5*time.Second)
-			if err != nil {
-				syncError = err
-				return
-			}
-			logger.Debug("install plan not yet populated from bundle image, requeueing")
-
-			return
-		}
-	}
-
 	outInstallPlan, syncError := transitionInstallPlanState(logger.Logger, o, *plan, o.now(), o.installPlanTimeout)
 
 	if syncError != nil {
@@ -1723,8 +1755,8 @@ func (o *Operator) syncInstallPlans(obj interface{}) (syncError error) {
 	return
 }
 
-func hasBundleLookupFailureCondition(plan *v1alpha1.InstallPlan) (bool, *v1alpha1.BundleLookupCondition) {
-	for _, bundleLookup := range plan.Status.BundleLookups {
+func hasBundleLookupFailureCondition(bundleLookups []v1alpha1.BundleLookup) (bool, *v1alpha1.BundleLookupCondition) {
+	for _, bundleLookup := range bundleLookups {
 		for _, cond := range bundleLookup.Conditions {
 			if cond.Type == v1alpha1.BundleLookupFailed && cond.Status == corev1.ConditionTrue {
 				return true, &cond
@@ -1734,27 +1766,6 @@ func hasBundleLookupFailureCondition(plan *v1alpha1.InstallPlan) (bool, *v1alpha
 	return false, nil
 }
 
-func (o *Operator) transitionInstallPlanToFailed(plan *v1alpha1.InstallPlan, logger logrus.FieldLogger, reason v1alpha1.InstallPlanConditionReason, message string) error {
-	now := o.now()
-	out := plan.DeepCopy()
-	out.Status.SetCondition(v1alpha1.ConditionFailed(v1alpha1.InstallPlanInstalled,
-		reason, message, &now))
-	out.Status.Phase = v1alpha1.InstallPlanPhaseFailed
-
-	logger.Info("transitioning InstallPlan to failed")
-	_, err := o.client.OperatorsV1alpha1().InstallPlans(plan.GetNamespace()).UpdateStatus(context.TODO(), out, metav1.UpdateOptions{})
-	if err == nil {
-		return nil
-	}
-
-	updateErr := errors.New("error updating InstallPlan status: " + err.Error())
-	logger = logger.WithField("updateError", updateErr)
-	logger.Errorf("error transitioning InstallPlan to failed")
-
-	// retry sync with error to update InstallPlan status
-	return fmt.Errorf("installplan failed: %s and error updating InstallPlan status as failed: %s", message, updateErr)
-}
-
 func (o *Operator) requeueSubscriptionForInstallPlan(plan *v1alpha1.InstallPlan, logger *logrus.Entry) {
 	// Notify subscription loop of installplan changes
 	owners := ownerutil.GetOwnersByKind(plan, v1alpha1.SubscriptionKind)


Test patch:
diff --git a/pkg/controller/operators/catalog/operator_test.go b/pkg/controller/operators/catalog/operator_test.go
index 6868f0520..cdf31e4d5 100644
--- a/pkg/controller/operators/catalog/operator_test.go
+++ b/pkg/controller/operators/catalog/operator_test.go
@@ -49,6 +49,8 @@ import (
 	"github.com/operator-framework/api/pkg/operators/v1alpha1"
 	"github.com/operator-framework/operator-lifecycle-manager/pkg/api/client/clientset/versioned/fake"
 	"github.com/operator-framework/operator-lifecycle-manager/pkg/api/client/informers/externalversions"
+	"github.com/operator-framework/operator-lifecycle-manager/pkg/controller/bundle"
+	"github.com/operator-framework/operator-lifecycle-manager/pkg/controller/bundle/bundlefakes"
 	olmerrors "github.com/operator-framework/operator-lifecycle-manager/pkg/controller/errors"
 	"github.com/operator-framework/operator-lifecycle-manager/pkg/controller/registry"
 	"github.com/operator-framework/operator-lifecycle-manager/pkg/controller/registry/grpc"
@@ -1127,6 +1129,12 @@ func TestSyncResolvingNamespace(t *testing.T) {
 	clockFake := utilclocktesting.NewFakeClock(time.Date(2018, time.January, 26, 20, 40, 0, 0, time.UTC))
 	now := metav1.NewTime(clockFake.Now())
 	testNamespace := "testNamespace"
+	og := &operatorsv1.OperatorGroup{
+		ObjectMeta: metav1.ObjectMeta{
+			Name:      "og",
+			Namespace: testNamespace,
+		},
+	}
 
 	type fields struct {
 		clientOptions   []clientfake.Option
@@ -1181,6 +1189,13 @@ func TestSyncResolvingNamespace(t *testing.T) {
 					Status: v1alpha1.SubscriptionStatus{
 						CurrentCSV: "",
 						State:      "",
+						Conditions: []v1alpha1.SubscriptionCondition{
+							{
+								Type:   v1alpha1.SubscriptionBundleUnpacking,
+								Status: corev1.ConditionFalse,
+							},
+						},
+						LastUpdated: now,
 					},
 				},
 			},
@@ -1319,7 +1334,7 @@ func TestSyncResolvingNamespace(t *testing.T) {
 			ctx, cancel := context.WithCancel(context.TODO())
 			defer cancel()
 
-			o, err := NewFakeOperator(ctx, testNamespace, []string{testNamespace}, withClock(clockFake), withClientObjs(tt.fields.existingOLMObjs...), withFakeClientOptions(tt.fields.clientOptions...))
+			o, err := NewFakeOperator(ctx, testNamespace, []string{testNamespace}, withClock(clockFake), withClientObjs(append(tt.fields.existingOLMObjs, og)...), withFakeClientOptions(tt.fields.clientOptions...))
 			require.NoError(t, err)
 
 			o.reconciler = &fakes.FakeRegistryReconcilerFactory{
@@ -1560,17 +1575,18 @@ func fakeConfigMapData() map[string]string {
 
 // fakeOperatorConfig is the configuration for a fake operator.
 type fakeOperatorConfig struct {
-	clock         utilclock.Clock
-	clientObjs    []runtime.Object
-	k8sObjs       []runtime.Object
-	extObjs       []runtime.Object
-	regObjs       []runtime.Object
-	clientOptions []clientfake.Option
-	logger        *logrus.Logger
-	resolver      resolver.StepResolver
-	recorder      record.EventRecorder
-	reconciler    reconciler.RegistryReconcilerFactory
-	sources       []sourceAddress
+	clock          utilclock.Clock
+	clientObjs     []runtime.Object
+	k8sObjs        []runtime.Object
+	extObjs        []runtime.Object
+	regObjs        []runtime.Object
+	clientOptions  []clientfake.Option
+	logger         *logrus.Logger
+	resolver       resolver.StepResolver
+	recorder       record.EventRecorder
+	reconciler     reconciler.RegistryReconcilerFactory
+	bundleUnpacker bundle.Unpacker
+	sources        []sourceAddress
 }
 
 // fakeOperatorOption applies an option to the given fake operator configuration.
@@ -1582,6 +1598,12 @@ func withResolver(res resolver.StepResolver) fakeOperatorOption {
 	}
 }
 
+func withBundleUnpacker(bundleUnpacker bundle.Unpacker) fakeOperatorOption {
+	return func(config *fakeOperatorConfig) {
+		config.bundleUnpacker = bundleUnpacker
+	}
+}
+
 func withSources(sources ...sourceAddress) fakeOperatorOption {
 	return func(config *fakeOperatorConfig) {
 		config.sources = sources
@@ -1627,10 +1649,11 @@ type sourceAddress struct {
 func NewFakeOperator(ctx context.Context, namespace string, namespaces []string, fakeOptions ...fakeOperatorOption) (*Operator, error) {
 	// Apply options to default config
 	config := &fakeOperatorConfig{
-		logger:   logrus.StandardLogger(),
-		clock:    utilclock.RealClock{},
-		resolver: &fakes.FakeStepResolver{},
-		recorder: &record.FakeRecorder{},
+		logger:         logrus.StandardLogger(),
+		clock:          utilclock.RealClock{},
+		resolver:       &fakes.FakeStepResolver{},
+		recorder:       &record.FakeRecorder{},
+		bundleUnpacker: &bundlefakes.FakeUnpacker{},
 	}
 	for _, option := range fakeOptions {
 		option(config)
@@ -1669,12 +1692,14 @@ func NewFakeOperator(ctx context.Context, namespace string, namespaces []string,
 	subInformer := operatorsFactory.Operators().V1alpha1().Subscriptions()
 	ipInformer := operatorsFactory.Operators().V1alpha1().InstallPlans()
 	csvInformer := operatorsFactory.Operators().V1alpha1().ClusterServiceVersions()
-	sharedInformers = append(sharedInformers, catsrcInformer.Informer(), subInformer.Informer(), ipInformer.Informer(), csvInformer.Informer())
+	ogInformer := operatorsFactory.Operators().V1().OperatorGroups()
+	sharedInformers = append(sharedInformers, catsrcInformer.Informer(), subInformer.Informer(), ipInformer.Informer(), csvInformer.Informer(), ogInformer.Informer())
 
 	lister.OperatorsV1alpha1().RegisterCatalogSourceLister(metav1.NamespaceAll, catsrcInformer.Lister())
 	lister.OperatorsV1alpha1().RegisterSubscriptionLister(metav1.NamespaceAll, subInformer.Lister())
 	lister.OperatorsV1alpha1().RegisterInstallPlanLister(metav1.NamespaceAll, ipInformer.Lister())
 	lister.OperatorsV1alpha1().RegisterClusterServiceVersionLister(metav1.NamespaceAll, csvInformer.Lister())
+	lister.OperatorsV1().RegisterOperatorGroupLister(metav1.NamespaceAll, ogInformer.Lister())
 
 	factory := informers.NewSharedInformerFactoryWithOptions(opClientFake.KubernetesInterface(), wakeupInterval, informers.WithNamespace(metav1.NamespaceAll))
 	roleInformer := factory.Rbac().V1().Roles()
@@ -1722,6 +1747,7 @@ func NewFakeOperator(ctx context.Context, namespace string, namespaces []string,
 		recorder:              config.recorder,
 		clientAttenuator:      scoped.NewClientAttenuator(logger, &rest.Config{}, opClientFake),
 		serviceAccountQuerier: scoped.NewUserDefinedServiceAccountQuerier(logger, clientFake),
+		bundleUnpacker:        config.bundleUnpacker,
 		catsrcQueueSet:        queueinformer.NewEmptyResourceQueueSet(),
 		clientFactory: &stubClientFactory{
 			operatorClient:   opClientFake,
diff --git a/pkg/controller/operators/catalog/subscriptions_test.go b/pkg/controller/operators/catalog/subscriptions_test.go
index 958aeb1d0..b84d24c9e 100644
--- a/pkg/controller/operators/catalog/subscriptions_test.go
+++ b/pkg/controller/operators/catalog/subscriptions_test.go
@@ -2,6 +2,7 @@ package catalog
 
 import (
 	"context"
+	"errors"
 	"fmt"
 	"testing"
 	"time"
@@ -10,9 +11,13 @@ import (
 	corev1 "k8s.io/api/core/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/runtime"
+	utilerrors "k8s.io/apimachinery/pkg/util/errors"
 	utilclocktesting "k8s.io/utils/clock/testing"
 
+	operatorsv1 "github.com/operator-framework/api/pkg/operators/v1"
 	"github.com/operator-framework/api/pkg/operators/v1alpha1"
+	"github.com/operator-framework/operator-lifecycle-manager/pkg/controller/bundle"
+	"github.com/operator-framework/operator-lifecycle-manager/pkg/controller/bundle/bundlefakes"
 	"github.com/operator-framework/operator-lifecycle-manager/pkg/controller/registry/reconciler"
 	"github.com/operator-framework/operator-lifecycle-manager/pkg/controller/registry/resolver"
 	"github.com/operator-framework/operator-lifecycle-manager/pkg/fakes"
@@ -23,12 +28,19 @@ func TestSyncSubscriptions(t *testing.T) {
 	clockFake := utilclocktesting.NewFakeClock(time.Date(2018, time.January, 26, 20, 40, 0, 0, time.UTC))
 	now := metav1.NewTime(clockFake.Now())
 	testNamespace := "testNamespace"
+	og := &operatorsv1.OperatorGroup{
+		ObjectMeta: metav1.ObjectMeta{
+			Name:      "og",
+			Namespace: testNamespace,
+		},
+	}
 
 	type fields struct {
 		clientOptions        []clientfake.Option
 		resolveSteps         []*v1alpha1.Step
 		resolveSubs          []*v1alpha1.Subscription
 		resolveBundleLookups []v1alpha1.BundleLookup
+		unpackBundleErr      error
 		existingOLMObjs      []runtime.Object
 	}
 	type args struct {
@@ -385,11 +397,10 @@ func TestSyncSubscriptions(t *testing.T) {
 						},
 						Conditions: []v1alpha1.BundleLookupCondition{
 							{
-								Type:               v1alpha1.BundleLookupPending,
-								Status:             corev1.ConditionTrue,
-								Reason:             "JobIncomplete",
-								Message:            "unpack job not completed",
-								LastTransitionTime: &now,
+								Type:    v1alpha1.BundleLookupPending,
+								Status:  corev1.ConditionTrue,
+								Reason:  "JobIncomplete",
+								Message: "unpack job not completed",
 							},
 						},
 					},
@@ -430,56 +441,203 @@ func TestSyncSubscriptions(t *testing.T) {
 						CatalogSourceNamespace: testNamespace,
 					},
 					Status: v1alpha1.SubscriptionStatus{
-						CurrentCSV: "",
-						State:      v1alpha1.SubscriptionStateUpgradePending,
-						Install: &v1alpha1.InstallPlanReference{
-							Kind:       v1alpha1.InstallPlanKind,
-							APIVersion: v1alpha1.InstallPlanAPIVersion,
+						CurrentCSV:  "",
+						State:       "",
+						LastUpdated: now,
+						Conditions: []v1alpha1.SubscriptionCondition{
+							{
+								Type:   v1alpha1.SubscriptionBundleUnpacking,
+								Status: corev1.ConditionTrue,
+								Reason: "UnpackingInProgress",
+							},
 						},
-						InstallPlanRef: &corev1.ObjectReference{
-							Namespace:  testNamespace,
-							Kind:       v1alpha1.InstallPlanKind,
-							APIVersion: v1alpha1.InstallPlanAPIVersion,
+					},
+				},
+			},
+		},
+		{
+			name: "NoStatus/NoCurrentCSV/BundleUnpackFailed",
+			fields: fields{
+				clientOptions: []clientfake.Option{clientfake.WithSelfLinks(t)},
+				existingOLMObjs: []runtime.Object{
+					&v1alpha1.Subscription{
+						TypeMeta: metav1.TypeMeta{
+							Kind:       v1alpha1.SubscriptionKind,
+							APIVersion: v1alpha1.SchemeGroupVersion.String(),
+						},
+						ObjectMeta: metav1.ObjectMeta{
+							Name:      "sub",
+							Namespace: testNamespace,
+						},
+						Spec: &v1alpha1.SubscriptionSpec{
+							CatalogSource:          "src",
+							CatalogSourceNamespace: testNamespace,
+						},
+						Status: v1alpha1.SubscriptionStatus{
+							CurrentCSV: "",
+							State:      "",
+						},
+					},
+				},
+				resolveSubs: []*v1alpha1.Subscription{
+					{
+						TypeMeta: metav1.TypeMeta{
+							Kind:       v1alpha1.SubscriptionKind,
+							APIVersion: v1alpha1.SchemeGroupVersion.String(),
+						},
+						ObjectMeta: metav1.ObjectMeta{
+							Name:      "sub",
+							Namespace: testNamespace,
+						},
+						Spec: &v1alpha1.SubscriptionSpec{
+							CatalogSource:          "src",
+							CatalogSourceNamespace: testNamespace,
+						},
+						Status: v1alpha1.SubscriptionStatus{
+							CurrentCSV: "",
+							State:      v1alpha1.SubscriptionStateUpgradePending,
+						},
+					},
+				},
+				resolveBundleLookups: []v1alpha1.BundleLookup{
+					{
+						Path:       "bundle-path-a",
+						Identifier: "bundle-a",
+						CatalogSourceRef: &corev1.ObjectReference{
+							Namespace: testNamespace,
+							Name:      "src",
+						},
+						Conditions: []v1alpha1.BundleLookupCondition{
+							{
+								Type:               v1alpha1.BundleLookupFailed,
+								Status:             corev1.ConditionTrue,
+								Reason:             "JobFailed",
+								Message:            "unpack job failed",
+								LastTransitionTime: &now,
+							},
 						},
-						InstallPlanGeneration: 1,
-						LastUpdated:           now,
 					},
 				},
 			},
-			wantInstallPlans: []v1alpha1.InstallPlan{
+			args: args{
+				obj: &v1alpha1.Subscription{
+					TypeMeta: metav1.TypeMeta{
+						Kind:       v1alpha1.SubscriptionKind,
+						APIVersion: v1alpha1.SchemeGroupVersion.String(),
+					},
+					ObjectMeta: metav1.ObjectMeta{
+						Name:      "sub",
+						Namespace: testNamespace,
+					},
+					Spec: &v1alpha1.SubscriptionSpec{
+						CatalogSource:          "src",
+						CatalogSourceNamespace: testNamespace,
+					},
+					Status: v1alpha1.SubscriptionStatus{
+						CurrentCSV: "",
+						State:      "",
+					},
+				},
+			},
+			wantSubscriptions: []*v1alpha1.Subscription{
 				{
-					Spec: v1alpha1.InstallPlanSpec{
-						ClusterServiceVersionNames: []string{"bundle-a"},
-						Approval:                   v1alpha1.ApprovalAutomatic,
-						Approved:                   true,
-						Generation:                 1,
+					TypeMeta: metav1.TypeMeta{
+						Kind:       v1alpha1.SubscriptionKind,
+						APIVersion: v1alpha1.SubscriptionCRDAPIVersion,
 					},
-					Status: v1alpha1.InstallPlanStatus{
-						Phase:          v1alpha1.InstallPlanPhaseInstalling,
-						CatalogSources: []string{},
-						BundleLookups: []v1alpha1.BundleLookup{
+					ObjectMeta: metav1.ObjectMeta{
+						Name:      "sub",
+						Namespace: testNamespace,
+					},
+					Spec: &v1alpha1.SubscriptionSpec{
+						CatalogSource:          "src",
+						CatalogSourceNamespace: testNamespace,
+					},
+					Status: v1alpha1.SubscriptionStatus{
+						CurrentCSV:  "",
+						State:       "",
+						LastUpdated: now,
+						Conditions: []v1alpha1.SubscriptionCondition{
 							{
-								Path:       "bundle-path-a",
-								Identifier: "bundle-a",
-								CatalogSourceRef: &corev1.ObjectReference{
-									Namespace: testNamespace,
-									Name:      "src",
-								},
-								Conditions: []v1alpha1.BundleLookupCondition{
-									{
-										Type:               v1alpha1.BundleLookupPending,
-										Status:             corev1.ConditionTrue,
-										Reason:             "JobIncomplete",
-										Message:            "unpack job not completed",
-										LastTransitionTime: &now,
-									},
-								},
+								Type:    v1alpha1.SubscriptionBundleUnpackFailed,
+								Reason:  "BundleUnpackFailed",
+								Message: "bundle unpacking failed. Reason: JobFailed, and Message: unpack job failed",
+								Status:  corev1.ConditionTrue,
 							},
 						},
 					},
 				},
 			},
 		},
+		{
+			name: "NoStatus/NoCurrentCSV/BundleLookupError",
+			fields: fields{
+				clientOptions: []clientfake.Option{clientfake.WithSelfLinks(t)},
+				existingOLMObjs: []runtime.Object{
+					&v1alpha1.Subscription{
+						TypeMeta: metav1.TypeMeta{
+							Kind:       v1alpha1.SubscriptionKind,
+							APIVersion: v1alpha1.SchemeGroupVersion.String(),
+						},
+						ObjectMeta: metav1.ObjectMeta{
+							Name:      "sub",
+							Namespace: testNamespace,
+						},
+						Spec: &v1alpha1.SubscriptionSpec{
+							CatalogSource:          "src",
+							CatalogSourceNamespace: testNamespace,
+						},
+						Status: v1alpha1.SubscriptionStatus{
+							CurrentCSV: "",
+							State:      "",
+						},
+					},
+				},
+				resolveBundleLookups: []v1alpha1.BundleLookup{{}},
+				unpackBundleErr:      errors.New("fake unpack error"),
+			},
+			args: args{
+				obj: &v1alpha1.Subscription{
+					TypeMeta: metav1.TypeMeta{
+						Kind:       v1alpha1.SubscriptionKind,
+						APIVersion: v1alpha1.SchemeGroupVersion.String(),
+					},
+					ObjectMeta: metav1.ObjectMeta{
+						Name:      "sub",
+						Namespace: testNamespace,
+					},
+					Spec: &v1alpha1.SubscriptionSpec{
+						CatalogSource:          "src",
+						CatalogSourceNamespace: testNamespace,
+					},
+					Status: v1alpha1.SubscriptionStatus{
+						CurrentCSV: "",
+						State:      "",
+					},
+				},
+			},
+			wantSubscriptions: []*v1alpha1.Subscription{
+				{
+					TypeMeta: metav1.TypeMeta{
+						Kind:       v1alpha1.SubscriptionKind,
+						APIVersion: v1alpha1.SubscriptionCRDAPIVersion,
+					},
+					ObjectMeta: metav1.ObjectMeta{
+						Name:      "sub",
+						Namespace: testNamespace,
+					},
+					Spec: &v1alpha1.SubscriptionSpec{
+						CatalogSource:          "src",
+						CatalogSourceNamespace: testNamespace,
+					},
+					Status: v1alpha1.SubscriptionStatus{
+						CurrentCSV: "",
+						State:      "",
+					},
+				},
+			},
+			wantErr: fmt.Errorf("bundle unpacking failed with an error: %w", utilerrors.NewAggregate([]error{errors.New("fake unpack error")})),
+		},
 		{
 			name: "Status/HaveCurrentCSV/UpdateFoundInCatalog",
 			fields: fields{
@@ -1013,7 +1171,12 @@ func TestSyncSubscriptions(t *testing.T) {
 			ctx, cancel := context.WithCancel(context.TODO())
 			defer cancel()
 
-			o, err := NewFakeOperator(ctx, testNamespace, []string{testNamespace}, withClock(clockFake), withClientObjs(tt.fields.existingOLMObjs...), withFakeClientOptions(tt.fields.clientOptions...))
+			fakeBundleUnpacker := &bundlefakes.FakeUnpacker{
+				UnpackBundleStub: func(lookup *v1alpha1.BundleLookup, timeout time.Duration) (*bundle.BundleUnpackResult, error) {
+					return &bundle.BundleUnpackResult{BundleLookup: lookup.DeepCopy()}, tt.fields.unpackBundleErr
+				},
+			}
+			o, err := NewFakeOperator(ctx, testNamespace, []string{testNamespace}, withClock(clockFake), withClientObjs(append(tt.fields.existingOLMObjs, og)...), withFakeClientOptions(tt.fields.clientOptions...), withBundleUnpacker(fakeBundleUnpacker))
 			require.NoError(t, err)
 
 			o.reconciler = &fakes.FakeRegistryReconcilerFactory{
diff --git a/test/e2e/catalog_e2e_test.go b/test/e2e/catalog_e2e_test.go
index b582f9ab6..41c6cca14 100644
--- a/test/e2e/catalog_e2e_test.go
+++ b/test/e2e/catalog_e2e_test.go
@@ -1010,7 +1010,7 @@ var _ = Describe("Starting CatalogSource e2e tests", func() {
 
 		catSrcImage := "quay.io/olmtest/busybox-dependencies-index"
 
-		// Create gRPC CatalogSource
+		By("creating gRPC CatalogSource")
 		source := &v1alpha1.CatalogSource{
 			TypeMeta: metav1.TypeMeta{
 				Kind:       v1alpha1.CatalogSourceKind,
@@ -1028,7 +1028,6 @@ var _ = Describe("Starting CatalogSource e2e tests", func() {
 				},
 			},
 		}
-
 		source, err := crc.OperatorsV1alpha1().CatalogSources(source.GetNamespace()).Create(context.Background(), source, metav1.CreateOptions{})
 		Expect(err).ShouldNot(HaveOccurred())
 		defer func() {
@@ -1036,22 +1035,22 @@ var _ = Describe("Starting CatalogSource e2e tests", func() {
 			Expect(err).ShouldNot(HaveOccurred())
 		}()
 
-		// Wait for the CatalogSource to be ready
+		By("waiting for the CatalogSource to be ready")
 		_, err = fetchCatalogSourceOnStatus(crc, source.GetName(), source.GetNamespace(), catalogSourceRegistryPodSynced)
 		Expect(err).ToNot(HaveOccurred(), "catalog source did not become ready")
 
-		// Create a Subscription for busybox
+		By("creating a Subscription for busybox")
 		subscriptionName := genName("sub-")
 		cleanupSubscription := createSubscriptionForCatalog(crc, source.GetNamespace(), subscriptionName, source.GetName(), packageName, channelName, "", v1alpha1.ApprovalAutomatic)
 		defer cleanupSubscription()
 
-		// Wait for the Subscription to succeed
+		By("waiting for the Subscription to succeed")
 		subscription, err := fetchSubscription(crc, ns.GetName(), subscriptionName, subscriptionStateAtLatestChecker)
 		Expect(err).ShouldNot(HaveOccurred())
 		Expect(subscription).ShouldNot(BeNil())
 		Expect(subscription.Status.InstalledCSV).To(Equal("busybox.v1.0.0"))
 
-		// Confirm that a subscription was created for busybox-dependency
+		By("confirming that a subscription was created for busybox-dependency")
 		subscriptionList, err := crc.OperatorsV1alpha1().Subscriptions(source.GetNamespace()).List(context.Background(), metav1.ListOptions{})
 		Expect(err).ShouldNot(HaveOccurred())
 		dependencySubscriptionName := ""
@@ -1062,13 +1061,13 @@ var _ = Describe("Starting CatalogSource e2e tests", func() {
 		}
 		Expect(dependencySubscriptionName).ToNot(BeEmpty())
 
-		// Wait for the Subscription to succeed
+		By("waiting for the Subscription to succeed")
 		subscription, err = fetchSubscription(crc, ns.GetName(), dependencySubscriptionName, subscriptionStateAtLatestChecker)
 		Expect(err).ShouldNot(HaveOccurred())
 		Expect(subscription).ShouldNot(BeNil())
 		Expect(subscription.Status.InstalledCSV).To(Equal("busybox-dependency.v1.0.0"))
 
-		// Update the catalog image
+		By("updating the catalog image")
 		Eventually(func() error {
 			existingSource, err := crc.OperatorsV1alpha1().CatalogSources(source.GetNamespace()).Get(context.Background(), sourceName, metav1.GetOptions{})
 			if err != nil {
@@ -1080,11 +1079,11 @@ var _ = Describe("Starting CatalogSource e2e tests", func() {
 			return err
 		}).Should(Succeed())
 
-		// Wait for the CatalogSource to be ready
+		By("waiting for the CatalogSource to be ready")
 		_, err = fetchCatalogSourceOnStatus(crc, source.GetName(), source.GetNamespace(), catalogSourceRegistryPodSynced)
 		Expect(err).ToNot(HaveOccurred(), "catalog source did not become ready")
 
-		// Wait for the busybox v2 Subscription to succeed
+		By("waiting for the busybox v2 Subscription to succeed")
 		subChecker := func(sub *v1alpha1.Subscription) bool {
 			return sub.Status.InstalledCSV == "busybox.v2.0.0"
 		}
@@ -1092,12 +1091,12 @@ var _ = Describe("Starting CatalogSource e2e tests", func() {
 		Expect(err).ShouldNot(HaveOccurred())
 		Expect(subscription).ShouldNot(BeNil())
 
-		// Wait for busybox v2 csv to succeed and check the replaces field
+		By("waiting for busybox v2 csv to succeed and check the replaces field")
 		csv, err := fetchCSV(crc, subscription.Status.CurrentCSV, subscription.GetNamespace(), csvSucceededChecker)
 		Expect(err).ShouldNot(HaveOccurred())
 		Expect(csv.Spec.Replaces).To(Equal("busybox.v1.0.0"))
 
-		// Wait for the busybox-dependency v2 Subscription to succeed
+		By("waiting for the busybox-dependency v2 Subscription to succeed")
 		subChecker = func(sub *v1alpha1.Subscription) bool {
 			return sub.Status.InstalledCSV == "busybox-dependency.v2.0.0"
 		}
@@ -1105,7 +1104,7 @@ var _ = Describe("Starting CatalogSource e2e tests", func() {
 		Expect(err).ShouldNot(HaveOccurred())
 		Expect(subscription).ShouldNot(BeNil())
 
-		// Wait for busybox-dependency v2 csv to succeed and check the replaces field
+		By("waiting for busybox-dependency v2 csv to succeed and check the replaces field")
 		csv, err = fetchCSV(crc, subscription.Status.CurrentCSV, subscription.GetNamespace(), csvSucceededChecker)
 		Expect(err).ShouldNot(HaveOccurred())
 		Expect(csv.Spec.Replaces).To(Equal("busybox-dependency.v1.0.0"))
@@ -1403,21 +1402,15 @@ var _ = Describe("Starting CatalogSource e2e tests", func() {
 				Expect(c.Create(context.Background(), subscription)).To(BeNil())
 			})
 
-			It("fails with a ResolutionFailed error condition, and a message that highlights the missing field in the CSV", func() {
-
-				subscription, err := fetchSubscription(crc, subscription.GetNamespace(), subscription.GetName(), subscriptionHasInstallPlanChecker)
-				Expect(err).Should(BeNil())
-				installPlanName := subscription.Status.Install.Name
-
-				// ensure we wait for the installPlan to fail before moving forward then fetch the subscription again
-				_, err = fetchInstallPlan(GinkgoT(), crc, installPlanName, subscription.GetNamespace(), buildInstallPlanPhaseCheckFunc(operatorsv1alpha1.InstallPlanPhaseFailed))
-				Expect(err).To(BeNil())
-				subscription, err = fetchSubscription(crc, subscription.GetNamespace(), subscription.GetName(), subscriptionHasInstallPlanChecker)
-				Expect(err).To(BeNil())
+			It("fails with a BundleUnpackFailed error condition, and a message that highlights the missing field in the CSV", func() {
+				Eventually(func(g Gomega) string {
+					fetchedSubscription, err := crc.OperatorsV1alpha1().Subscriptions(ns.GetName()).Get(context.Background(), subscription.GetName(), metav1.GetOptions{})
+					g.Expect(err).NotTo(HaveOccurred())
 
-				// expect the message that API missing
-				failingCondition := subscription.Status.GetCondition(operatorsv1alpha1.SubscriptionInstallPlanFailed)
-				Expect(failingCondition.Message).To(ContainSubstring("missing APIVersion"))
+					// expect the message that API missing
+					failingCondition := fetchedSubscription.Status.GetCondition(v1alpha1.SubscriptionBundleUnpackFailed)
+					return failingCondition.Message
+				}).Should(ContainSubstring("missing APIVersion"))
 			})
 		})
 	})
diff --git a/test/e2e/fail_forward_e2e_test.go b/test/e2e/fail_forward_e2e_test.go
index 0913271a7..3df0768f6 100644
--- a/test/e2e/fail_forward_e2e_test.go
+++ b/test/e2e/fail_forward_e2e_test.go
@@ -58,12 +58,18 @@ var _ = Describe("Fail Forward Upgrades", func() {
 	When("an InstallPlan is reporting a failed state", func() {
 
 		var (
-			magicCatalog      *MagicCatalog
-			catalogSourceName string
-			subscription      *operatorsv1alpha1.Subscription
+			magicCatalog           *MagicCatalog
+			catalogSourceName      string
+			subscription           *operatorsv1alpha1.Subscription
+			originalInstallPlanRef *corev1.ObjectReference
+			failedInstallPlanRef   *corev1.ObjectReference
 		)
 
 		BeforeEach(func() {
+			By("creating a service account with no permission")
+			saNameWithNoPerms := genName("scoped-sa-")
+			newServiceAccount(ctx.Ctx().KubeClient(), ns.GetName(), saNameWithNoPerms)
+
 			By("deploying the testing catalog")
 			provider, err := NewFileBasedFiledBasedCatalogProvider(filepath.Join(testdataDir, failForwardTestDataBaseDir, "example-operator.v0.1.0.yaml"))
 			Expect(err).To(BeNil())
@@ -90,19 +96,18 @@ var _ = Describe("Fail Forward Upgrades", func() {
 			subscription, err := fetchSubscription(crclient, subscription.GetNamespace(), subscription.GetName(), subscriptionHasInstallPlanChecker)
 			Expect(err).Should(BeNil())
 
-			originalInstallPlanRef := subscription.Status.InstallPlanRef
+			originalInstallPlanRef = subscription.Status.InstallPlanRef
 
 			By("waiting for the v0.1.0 CSV to report a succeeded phase")
 			_, err = fetchCSV(crclient, subscription.Status.CurrentCSV, ns.GetName(), buildCSVConditionChecker(operatorsv1alpha1.CSVPhaseSucceeded))
 			Expect(err).ShouldNot(HaveOccurred())
 
-			By("patching the OperatorGroup to reduce the bundle unpacking timeout")
-			addBundleUnpackTimeoutOGAnnotation(context.Background(), c, types.NamespacedName{Name: ogName, Namespace: ns.GetName()}, "1s")
+			By("updating the operator group to use the service account without required permissions to simulate InstallPlan failure")
+			Eventually(operatorGroupServiceAccountNameSetter(crclient, ns.GetName(), ogName, saNameWithNoPerms)).Should(Succeed())
 
-			By("updating the catalog with a broken v0.2.0 bundle image")
+			By("updating the catalog with v0.2.0 bundle image")
 			brokenProvider, err := NewFileBasedFiledBasedCatalogProvider(filepath.Join(testdataDir, failForwardTestDataBaseDir, "example-operator.v0.2.0.yaml"))
 			Expect(err).To(BeNil())
-
 			err = magicCatalog.UpdateCatalog(context.Background(), brokenProvider)
 			Expect(err).To(BeNil())
 
@@ -111,31 +116,29 @@ var _ = Describe("Fail Forward Upgrades", func() {
 			Expect(err).Should(BeNil())
 
 			By("waiting for the bad InstallPlan to report a failed installation state")
-			ref := subscription.Status.InstallPlanRef
-			_, err = fetchInstallPlan(GinkgoT(), crclient, ref.Name, ref.Namespace, buildInstallPlanPhaseCheckFunc(operatorsv1alpha1.InstallPlanPhaseFailed))
+			failedInstallPlanRef = subscription.Status.InstallPlanRef
+			_, err = fetchInstallPlan(GinkgoT(), crclient, failedInstallPlanRef.Name, failedInstallPlanRef.Namespace, buildInstallPlanPhaseCheckFunc(operatorsv1alpha1.InstallPlanPhaseFailed))
 			Expect(err).To(BeNil())
 
+			By("updating the operator group remove service account without permissions")
+			Eventually(operatorGroupServiceAccountNameSetter(crclient, ns.GetName(), ogName, "")).Should(Succeed())
 		})
 		AfterEach(func() {
 			By("removing the testing catalog resources")
 			Expect(magicCatalog.UndeployCatalog(context.Background())).To(BeNil())
 		})
 		It("eventually reports a successful state when multiple bad versions are rolled forward", func() {
-			By("patching the catalog with another bad bundle version")
-			badProvider, err := NewFileBasedFiledBasedCatalogProvider(filepath.Join(testdataDir, "fail-forward/multiple-bad-versions", "example-operator.v0.2.1.yaml"))
-			Expect(err).To(BeNil())
+			By("patching the OperatorGroup to reduce the bundle unpacking timeout")
+			addBundleUnpackTimeoutOGAnnotation(context.Background(), c, types.NamespacedName{Name: ogName, Namespace: ns.GetName()}, "1s")
 
+			By("patching the catalog with a bad bundle version")
+			badProvider, err := NewFileBasedFiledBasedCatalogProvider(filepath.Join(testdataDir, "fail-forward/multiple-bad-versions", "example-operator.v0.2.1-non-existent-tag.yaml"))
+			Expect(err).To(BeNil())
 			err = magicCatalog.UpdateCatalog(context.Background(), badProvider)
 			Expect(err).To(BeNil())
 
-			By("waiting for the subscription to have the example-operator.v0.2.1 status.updatedCSV")
-			subscription, err = fetchSubscription(crclient, subscription.GetNamespace(), subscription.GetName(), subscriptionHasCurrentCSV("example-operator.v0.2.1"))
-			Expect(err).Should(BeNil())
-
-			By("waiting for the bad v0.2.1 InstallPlan to report a failed installation state")
-			ref := subscription.Status.InstallPlanRef
-			_, err = fetchInstallPlan(GinkgoT(), crclient, ref.Name, ref.Namespace, buildInstallPlanPhaseCheckFunc(operatorsv1alpha1.InstallPlanPhaseFailed))
-			Expect(err).To(BeNil())
+			By("waiting for the subscription to maintain the example-operator.v0.2.0 status.currentCSV")
+			Consistently(subscriptionCurrentCSVGetter(crclient, subscription.GetNamespace(), subscription.GetName())).Should(Equal("example-operator.v0.2.0"))
 
 			By("patching the OperatorGroup to increase the bundle unpacking timeout")
 			addBundleUnpackTimeoutOGAnnotation(context.Background(), c, types.NamespacedName{Name: ogName, Namespace: ns.GetName()}, "5m")
@@ -146,57 +149,74 @@ var _ = Describe("Fail Forward Upgrades", func() {
 			err = magicCatalog.UpdateCatalog(context.Background(), fixedProvider)
 			Expect(err).To(BeNil())
 
-			By("waiting for the subscription to have the example-operator.v0.3.0 status.updatedCSV")
-			subscription, err = fetchSubscription(crclient, subscription.GetNamespace(), subscription.GetName(), subscriptionHasCurrentCSV("example-operator.v0.3.0"))
+			By("waiting for the subscription to have the example-operator.v0.3.0 status.currentCSV")
+			_, err = fetchSubscription(crclient, subscription.GetNamespace(), subscription.GetName(), subscriptionHasCurrentCSV("example-operator.v0.3.0"))
+			Expect(err).Should(BeNil())
+
+			By("verifying the subscription is referencing a new InstallPlan")
+			subscription, err = fetchSubscription(crclient, subscription.GetNamespace(), subscription.GetName(), subscriptionHasInstallPlanDifferentChecker(originalInstallPlanRef.Name))
 			Expect(err).Should(BeNil())
+
+			By("waiting for the fixed v0.3.0 InstallPlan to report a successful state")
+			ref := subscription.Status.InstallPlanRef
+			_, err = fetchInstallPlan(GinkgoT(), crclient, ref.Name, ref.Namespace, buildInstallPlanPhaseCheckFunc(operatorsv1alpha1.InstallPlanPhaseComplete))
+			Expect(err).To(BeNil())
 		})
 
 		It("eventually reports a successful state when using skip ranges", func() {
-			By("patching the OperatorGroup to increase the bundle unpacking timeout")
-			addBundleUnpackTimeoutOGAnnotation(context.Background(), c, types.NamespacedName{Name: ogName, Namespace: ns.GetName()}, "5m")
-
 			By("patching the catalog with a fixed version")
 			fixedProvider, err := NewFileBasedFiledBasedCatalogProvider(filepath.Join(testdataDir, "fail-forward/skip-range", "example-operator.v0.3.0.yaml"))
 			Expect(err).To(BeNil())
 			err = magicCatalog.UpdateCatalog(context.Background(), fixedProvider)
 			Expect(err).To(BeNil())
 
-			By("waiting for the subscription to have the example-operator.v0.3.0 status.updatedCSV")
-			subscription, err = fetchSubscription(crclient, subscription.GetNamespace(), subscription.GetName(), subscriptionHasCurrentCSV("example-operator.v0.3.0"))
+			By("waiting for the subscription to have the example-operator.v0.3.0 status.currentCSV")
+			_, err = fetchSubscription(crclient, subscription.GetNamespace(), subscription.GetName(), subscriptionHasCurrentCSV("example-operator.v0.3.0"))
+			Expect(err).Should(BeNil())
+
+			By("verifying the subscription is referencing a new InstallPlan")
+			subscription, err = fetchSubscription(crclient, subscription.GetNamespace(), subscription.GetName(), subscriptionHasInstallPlanDifferentChecker(originalInstallPlanRef.Name))
 			Expect(err).Should(BeNil())
+
+			By("waiting for the fixed v0.3.0 InstallPlan to report a successful state")
+			ref := subscription.Status.InstallPlanRef
+			_, err = fetchInstallPlan(GinkgoT(), crclient, ref.Name, ref.Namespace, buildInstallPlanPhaseCheckFunc(operatorsv1alpha1.InstallPlanPhaseComplete))
+			Expect(err).To(BeNil())
 		})
 		It("eventually reports a successful state when using skips", func() {
-			By("patching the OperatorGroup to increase the bundle unpacking timeout")
-			addBundleUnpackTimeoutOGAnnotation(context.Background(), c, types.NamespacedName{Name: ogName, Namespace: ns.GetName()}, "5m")
-
 			By("patching the catalog with a fixed version")
 			fixedProvider, err := NewFileBasedFiledBasedCatalogProvider(filepath.Join(testdataDir, "fail-forward/skips", "example-operator.v0.3.0.yaml"))
 			Expect(err).To(BeNil())
 			err = magicCatalog.UpdateCatalog(context.Background(), fixedProvider)
 			Expect(err).To(BeNil())
 
-			By("waiting for the subscription to have the example-operator.v0.3.0 status.updatedCSV")
-			subscription, err = fetchSubscription(crclient, subscription.GetNamespace(), subscription.GetName(), subscriptionHasCurrentCSV("example-operator.v0.3.0"))
+			By("waiting for the subscription to have the example-operator.v0.3.0 status.currentCSV")
+			_, err = fetchSubscription(crclient, subscription.GetNamespace(), subscription.GetName(), subscriptionHasCurrentCSV("example-operator.v0.3.0"))
+			Expect(err).Should(BeNil())
+
+			By("verifying the subscription is referencing a new InstallPlan")
+			subscription, err = fetchSubscription(crclient, subscription.GetNamespace(), subscription.GetName(), subscriptionHasInstallPlanDifferentChecker(originalInstallPlanRef.Name))
 			Expect(err).Should(BeNil())
+
+			By("waiting for the fixed v0.3.0 InstallPlan to report a successful state")
+			ref := subscription.Status.InstallPlanRef
+			_, err = fetchInstallPlan(GinkgoT(), crclient, ref.Name, ref.Namespace, buildInstallPlanPhaseCheckFunc(operatorsv1alpha1.InstallPlanPhaseComplete))
+			Expect(err).To(BeNil())
 		})
 		It("eventually reports a failed state when using replaces", func() {
-			By("patching the OperatorGroup to increase the bundle unpacking timeout")
-			addBundleUnpackTimeoutOGAnnotation(context.Background(), c, types.NamespacedName{Name: ogName, Namespace: ns.GetName()}, "5m")
-
 			By("patching the catalog with a fixed version")
 			fixedProvider, err := NewFileBasedFiledBasedCatalogProvider(filepath.Join(testdataDir, "fail-forward/replaces", "example-operator.v0.3.0.yaml"))
 			Expect(err).To(BeNil())
 			err = magicCatalog.UpdateCatalog(context.Background(), fixedProvider)
 			Expect(err).To(BeNil())
 
-			By("waiting for the subscription to maintain the example-operator.v0.2.0 status.updatedCSV")
-			Consistently(func() string {
-				subscription, err := crclient.OperatorsV1alpha1().Subscriptions(subscription.GetNamespace()).Get(context.Background(), subscription.GetName(), metav1.GetOptions{})
-				if err != nil || subscription == nil {
-					return ""
-				}
-				return subscription.Status.CurrentCSV
-			}).Should(Equal("example-operator.v0.2.0"))
+			By("waiting for the subscription to maintain the example-operator.v0.2.0 status.currentCSV")
+			Consistently(subscriptionCurrentCSVGetter(crclient, subscription.GetNamespace(), subscription.GetName())).Should(Equal("example-operator.v0.2.0"))
+
+			By("verifying the subscription is referencing the same InstallPlan")
+			subscription, err = fetchSubscription(crclient, subscription.GetNamespace(), subscription.GetName(), subscriptionHasInstallPlanChecker)
+			Expect(err).Should(BeNil())
+			Expect(subscription.Status.InstallPlanRef.Name).To(Equal(failedInstallPlanRef.Name))
 		})
 	})
 	When("a CSV resource is in a failed state", func() {
@@ -239,7 +259,7 @@ var _ = Describe("Fail Forward Upgrades", func() {
 			Expect(err).ShouldNot(HaveOccurred())
 
 			By("updating the catalog with a broken v0.2.0 csv")
-			brokenProvider, err := NewFileBasedFiledBasedCatalogProvider(filepath.Join(testdataDir, failForwardTestDataBaseDir, "example-operator.v0.2.0-2.yaml"))
+			brokenProvider, err := NewFileBasedFiledBasedCatalogProvider(filepath.Join(testdataDir, failForwardTestDataBaseDir, "example-operator.v0.2.0-invalid-csv.yaml"))
 			Expect(err).To(BeNil())
 
 			err = magicCatalog.UpdateCatalog(context.Background(), brokenProvider)
@@ -269,7 +289,7 @@ var _ = Describe("Fail Forward Upgrades", func() {
 			err = magicCatalog.UpdateCatalog(context.Background(), fixedProvider)
 			Expect(err).To(BeNil())
 
-			By("waiting for the subscription to have the example-operator.v0.3.0 status.updatedCSV")
+			By("waiting for the subscription to have the example-operator.v0.3.0 status.currentCSV")
 			subscription, err = fetchSubscription(crclient, subscription.GetNamespace(), subscription.GetName(), subscriptionHasCurrentCSV("example-operator.v0.3.0"))
 			Expect(err).Should(BeNil())
 		})
@@ -282,7 +302,7 @@ var _ = Describe("Fail Forward Upgrades", func() {
 			err = magicCatalog.UpdateCatalog(context.Background(), fixedProvider)
 			Expect(err).To(BeNil())
 
-			By("waiting for the subscription to have the example-operator.v0.3.0 status.updatedCSV")
+			By("waiting for the subscription to have the example-operator.v0.3.0 status.currentCSV")
 			subscription, err = fetchSubscription(crclient, subscription.GetNamespace(), subscription.GetName(), subscriptionHasCurrentCSV("example-operator.v0.3.0"))
 			Expect(err).Should(BeNil())
 		})
@@ -295,7 +315,7 @@ var _ = Describe("Fail Forward Upgrades", func() {
 			err = magicCatalog.UpdateCatalog(context.Background(), fixedProvider)
 			Expect(err).To(BeNil())
 
-			By("waiting for the subscription to have the example-operator.v0.3.0 status.updatedCSV")
+			By("waiting for the subscription to have the example-operator.v0.3.0 status.currentCSV")
 			subscription, err = fetchSubscription(crclient, subscription.GetNamespace(), subscription.GetName(), subscriptionHasCurrentCSV("example-operator.v0.3.0"))
 			Expect(err).Should(BeNil())
 		})
diff --git a/test/e2e/installplan_e2e_test.go b/test/e2e/installplan_e2e_test.go
index d349a4313..92d2c4c3e 100644
--- a/test/e2e/installplan_e2e_test.go
+++ b/test/e2e/installplan_e2e_test.go
@@ -30,7 +30,6 @@ import (
 	"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
 	"k8s.io/apimachinery/pkg/runtime"
 	k8sjson "k8s.io/apimachinery/pkg/runtime/serializer/json"
-	"k8s.io/apimachinery/pkg/types"
 	"k8s.io/apimachinery/pkg/util/diff"
 	"k8s.io/apimachinery/pkg/util/wait"
 	"k8s.io/apimachinery/pkg/watch"
@@ -3062,90 +3061,6 @@ var _ = Describe("Install Plan", func() {
 		require.Equal(GinkgoT(), operatorsv1alpha1.InstallPlanPhaseComplete, fetchedInstallPlan.Status.Phase)
 	})
 
-	It("unpacks bundle image", func() {
-		ns, err := c.KubernetesInterface().CoreV1().Namespaces().Create(context.Background(), &corev1.Namespace{
-			ObjectMeta: metav1.ObjectMeta{
-				Name: genName("ns-"),
-			},
-		}, metav1.CreateOptions{})
-		require.NoError(GinkgoT(), err)
-
-		og := &operatorsv1.OperatorGroup{}
-		og.SetName("og")
-		_, err = crc.OperatorsV1().OperatorGroups(ns.GetName()).Create(context.Background(), og, metav1.CreateOptions{})
-		require.NoError(GinkgoT(), err)
-
-		deleteOpts := &metav1.DeleteOptions{}
-		defer func() {
-			require.NoError(GinkgoT(), c.KubernetesInterface().CoreV1().Namespaces().Delete(context.Background(), ns.GetName(), *deleteOpts))
-		}()
-
-		catsrc := &operatorsv1alpha1.CatalogSource{
-			ObjectMeta: metav1.ObjectMeta{
-				Name:      genName("kiali-"),
-				Namespace: ns.GetName(),
-				Labels:    map[string]string{"olm.catalogSource": "kaili-catalog"},
-			},
-			Spec: operatorsv1alpha1.CatalogSourceSpec{
-				Image:      "quay.io/operator-framework/ci-index:latest",
-				SourceType: operatorsv1alpha1.SourceTypeGrpc,
-				GrpcPodConfig: &operatorsv1alpha1.GrpcPodConfig{
-					SecurityContextConfig: operatorsv1alpha1.Restricted,
-				},
-			},
-		}
-		catsrc, err = crc.OperatorsV1alpha1().CatalogSources(catsrc.GetNamespace()).Create(context.Background(), catsrc, metav1.CreateOptions{})
-		require.NoError(GinkgoT(), err)
-		defer func() {
-			Eventually(func() error {
-				return client.IgnoreNotFound(ctx.Ctx().Client().Delete(context.Background(), catsrc))
-			}).Should(Succeed())
-		}()
-
-		// Wait for the CatalogSource to be ready
-		catsrc, err = fetchCatalogSourceOnStatus(crc, catsrc.GetName(), catsrc.GetNamespace(), catalogSourceRegistryPodSynced)
-		require.NoError(GinkgoT(), err)
-
-		// Generate a Subscription
-		subName := genName("kiali-")
-		cleanUpSubscriptionFn := createSubscriptionForCatalog(crc, catsrc.GetNamespace(), subName, catsrc.GetName(), "kiali", stableChannel, "", operatorsv1alpha1.ApprovalAutomatic)
-		defer cleanUpSubscriptionFn()
-
-		sub, err := fetchSubscription(crc, catsrc.GetNamespace(), subName, subscriptionHasInstallPlanChecker)
-		require.NoError(GinkgoT(), err)
-
-		// Wait for the expected InstallPlan's execution to either fail or succeed
-		ipName := sub.Status.InstallPlanRef.Name
-		ip, err := waitForInstallPlan(crc, ipName, sub.GetNamespace(), buildInstallPlanPhaseCheckFunc(operatorsv1alpha1.InstallPlanPhaseFailed, operatorsv1alpha1.InstallPlanPhaseComplete))
-		require.NoError(GinkgoT(), err)
-		require.Equal(GinkgoT(), operatorsv1alpha1.InstallPlanPhaseComplete, ip.Status.Phase, "InstallPlan not complete")
-
-		// Ensure the InstallPlan contains the steps resolved from the bundle image
-		operatorName := "kiali-operator"
-		expectedSteps := map[registry.ResourceKey]struct{}{
-			{Name: operatorName, Kind: "ClusterServiceVersion"}:                                  {},
-			{Name: "kialis.kiali.io", Kind: "CustomResourceDefinition"}:                          {},
-			{Name: "monitoringdashboards.monitoring.kiali.io", Kind: "CustomResourceDefinition"}: {},
-			{Name: operatorName, Kind: "ServiceAccount"}:                                         {},
-			{Name: operatorName, Kind: "ClusterRole"}:                                            {},
-			{Name: operatorName, Kind: "ClusterRoleBinding"}:                                     {},
-		}
-		require.Lenf(GinkgoT(), ip.Status.Plan, len(expectedSteps), "number of expected steps does not match installed: %v", ip.Status.Plan)
-
-		for _, step := range ip.Status.Plan {
-			key := registry.ResourceKey{
-				Name: step.Resource.Name,
-				Kind: step.Resource.Kind,
-			}
-			for expected := range expectedSteps {
-				if strings.HasPrefix(key.Name, expected.Name) && key.Kind == expected.Kind {
-					delete(expectedSteps, expected)
-				}
-			}
-		}
-		require.Lenf(GinkgoT(), expectedSteps, 0, "Actual resource steps do not match expected: %#v", expectedSteps)
-	})
-
 	// This It spec verifies that, in cases where there are multiple options to fulfil a dependency
 	// across multiple catalogs, we only generate one installplan with one set of resolved resources.
 	//issue: https://github.com/operator-framework/operator-lifecycle-manager/issues/2633
@@ -3402,218 +3317,6 @@ var _ = Describe("Install Plan", func() {
 		})
 	})
 
-	When("waiting on the bundle unpacking job", func() {
-		var (
-			ns         *corev1.Namespace
-			catsrcName string
-			ip         *operatorsv1alpha1.InstallPlan
-			ogName     string
-		)
-
-		BeforeEach(func() {
-			ns = &corev1.Namespace{}
-			ns.SetName(genName("ns-"))
-			Eventually(func() error {
-				return ctx.Ctx().Client().Create(context.Background(), ns)
-			}, timeout, interval).Should(Succeed(), "could not create Namespace")
-
-			// Create a dummy CatalogSource to bypass the bundle unpacker's check for a CatalogSource
-			catsrc := &operatorsv1alpha1.CatalogSource{
-				ObjectMeta: metav1.ObjectMeta{
-					Name:      genName("dummy-catsrc-"),
-					Namespace: ns.GetName(),
-				},
-				Spec: operatorsv1alpha1.CatalogSourceSpec{
-					Image:      "localhost:0/not/exist:catsrc",
-					SourceType: operatorsv1alpha1.SourceTypeGrpc,
-					GrpcPodConfig: &operatorsv1alpha1.GrpcPodConfig{
-						SecurityContextConfig: operatorsv1alpha1.Restricted,
-					},
-				},
-			}
-			Eventually(func() error {
-				return ctx.Ctx().Client().Create(context.Background(), catsrc)
-			}, timeout, interval).Should(Succeed(), "could not create CatalogSource")
-
-			catsrcName = catsrc.GetName()
-
-			// Create the OperatorGroup
-			og := &operatorsv1.OperatorGroup{
-				ObjectMeta: metav1.ObjectMeta{
-					Name:      "og",
-					Namespace: ns.GetName(),
-				},
-				Spec: operatorsv1.OperatorGroupSpec{
-					TargetNamespaces: []string{ns.GetName()},
-				},
-			}
-			ogName = og.GetName()
-			Eventually(func() error {
-				return ctx.Ctx().Client().Create(context.Background(), og)
-			}, timeout, interval).Should(Succeed(), "could not create OperatorGroup")
-
-			// Wait for the OperatorGroup to be synced so the InstallPlan doesn't have to be resynced due to an invalid OperatorGroup
-			Eventually(
-				func() ([]string, error) {
-					err := ctx.Ctx().Client().Get(context.Background(), client.ObjectKeyFromObject(og), og)
-					ctx.Ctx().Logf("Waiting for OperatorGroup(%v) to be synced with status.namespaces: %v", og.Name, og.Status.Namespaces)
-					return og.Status.Namespaces, err
-				},
-				1*time.Minute,
-				interval,
-			).Should(ContainElement(ns.GetName()))
-
-			ip = &operatorsv1alpha1.InstallPlan{
-				ObjectMeta: metav1.ObjectMeta{
-					Name:      "ip",
-					Namespace: ns.GetName(),
-				},
-				Spec: operatorsv1alpha1.InstallPlanSpec{
-					ClusterServiceVersionNames: []string{"foobar"},
-					Approval:                   operatorsv1alpha1.ApprovalAutomatic,
-					Approved:                   true,
-				},
-			}
-		})
-
-		AfterEach(func() {
-			Eventually(func() error {
-				return client.IgnoreNotFound(ctx.Ctx().Client().Delete(context.Background(), ns))
-			}, timeout, interval).Should(Succeed(), "could not delete Namespace")
-			Eventually(func() error {
-				return client.IgnoreNotFound(ctx.Ctx().Client().Delete(context.Background(), ip))
-			}, timeout, interval).Should(Succeed(), "could not delete Namespace")
-		})
-
-		It("should show an error on the bundlelookup condition for a non-existent bundle image", func() {
-			// We wait for some time over the bundle unpack timeout (i.e ActiveDeadlineSeconds) so that the Job can eventually fail
-			// Since the default --bundle-unpack-timeout=10m, we override with a shorter timeout
-			By("patching the OperatorGroup to reduce the bundle unpacking timeout")
-			addBundleUnpackTimeoutOGAnnotation(context.Background(), ctx.Ctx().Client(), types.NamespacedName{Name: ogName, Namespace: ns.GetName()}, "1m")
-
-			Eventually(func() error {
-				return ctx.Ctx().Client().Create(context.Background(), ip)
-			}, timeout, interval).Should(Succeed(), "could not create InstallPlan")
-
-			now := metav1.Now()
-			// Create an InstallPlan status.bundleLookups.Path specified for a non-existent bundle image
-			ip.Status = operatorsv1alpha1.InstallPlanStatus{
-				Phase:          operatorsv1alpha1.InstallPlanPhaseInstalling,
-				CatalogSources: []string{},
-				BundleLookups: []operatorsv1alpha1.BundleLookup{
-					{
-						Path:       "localhost:0/not/exist:v0.0.1",
-						Identifier: "foobar.v0.0.1",
-						CatalogSourceRef: &corev1.ObjectReference{
-							Namespace: ns.GetName(),
-							Name:      catsrcName,
-						},
-						Conditions: []operatorsv1alpha1.BundleLookupCondition{
-							{
-								Type:               operatorsv1alpha1.BundleLookupPending,
-								Status:             corev1.ConditionTrue,
-								Reason:             "JobIncomplete",
-								Message:            "unpack job not completed",
-								LastTransitionTime: &now,
-							},
-						},
-					},
-				},
-			}
-
-			// The status gets ignored on create so we need to update it else the InstallPlan sync ignores
-			// InstallPlans without any steps or bundle lookups
-			Eventually(func() error {
-				return ctx.Ctx().Client().Status().Update(context.Background(), ip)
-			}, timeout, interval).Should(Succeed(), "could not update InstallPlan status")
-
-			// The InstallPlan's status.bundleLookup.conditions should have a BundleLookupPending condition
-			// with the container status from unpack pod that mentions an image pull failure for the non-existent
-			// image, e.g ErrImagePull or ImagePullBackOff
-			Eventually(
-				func() (string, error) {
-					err := ctx.Ctx().Client().Get(context.Background(), client.ObjectKeyFromObject(ip), ip)
-					if err != nil {
-						return "", err
-					}
-					for _, bl := range ip.Status.BundleLookups {
-						for _, cond := range bl.Conditions {
-							if cond.Type != operatorsv1alpha1.BundleLookupPending {
-								continue
-							}
-							return cond.Message, nil
-						}
-					}
-					return "", fmt.Errorf("%s condition not found", operatorsv1alpha1.BundleLookupPending)
-				},
-				1*time.Minute,
-				interval,
-			).Should(ContainSubstring("ErrImagePull"))
-
-			waitFor := 1*time.Minute + 30*time.Second
-			// The InstallPlan should eventually fail due to the ActiveDeadlineSeconds limit
-			Eventually(
-				func() (*operatorsv1alpha1.InstallPlan, error) {
-					err := ctx.Ctx().Client().Get(context.Background(), client.ObjectKeyFromObject(ip), ip)
-					return ip, err
-				},
-				waitFor,
-				interval,
-			).Should(HavePhase(operatorsv1alpha1.InstallPlanPhaseFailed))
-		})
-
-		It("should timeout and fail the InstallPlan for an invalid bundle image", func() {
-			Eventually(func() error {
-				return ctx.Ctx().Client().Create(context.Background(), ip)
-			}, timeout, interval).Should(Succeed(), "could not create InstallPlan")
-
-			// The status gets ignored on create so we need to update it else the InstallPlan sync ignores
-			// InstallPlans without any steps or bundle lookups
-			// Create an InstallPlan status.bundleLookups.Path specified for an invalid bundle image
-			now := metav1.Now()
-			ip.Status = operatorsv1alpha1.InstallPlanStatus{
-				Phase:          operatorsv1alpha1.InstallPlanPhaseInstalling,
-				CatalogSources: []string{},
-				BundleLookups: []operatorsv1alpha1.BundleLookup{
-					{
-						Path:       "alpine:3.13",
-						Identifier: "foobar.v0.0.1",
-						CatalogSourceRef: &corev1.ObjectReference{
-							Namespace: ns.GetName(),
-							Name:      catsrcName,
-						},
-						Conditions: []operatorsv1alpha1.BundleLookupCondition{
-							{
-								Type:               operatorsv1alpha1.BundleLookupPending,
-								Status:             corev1.ConditionTrue,
-								Reason:             "JobIncomplete",
-								Message:            "unpack job not completed",
-								LastTransitionTime: &now,
-							},
-						},
-					},
-				},
-			}
-
-			Eventually(func() error {
-				return ctx.Ctx().Client().Status().Update(context.Background(), ip)
-			}, timeout, interval).Should(Succeed(), "could not update InstallPlan status")
-
-			// The InstallPlan should fail after the unpack pod keeps failing and exceeds the job's
-			// BackoffLimit(set to 3), which for 4 failures is an exponential backoff (10s + 20s + 40s + 80s)= 2m30s
-			// so we wait a little over that.
-			Eventually(
-				func() (*operatorsv1alpha1.InstallPlan, error) {
-					err := ctx.Ctx().Client().Get(context.Background(), client.ObjectKeyFromObject(ip), ip)
-					return ip, err
-				},
-				5*time.Minute,
-				interval,
-			).Should(HavePhase(operatorsv1alpha1.InstallPlanPhaseFailed))
-		})
-
-	})
-
 	It("compresses installplan step resource manifests to configmap references", func() {
 		// Test ensures that all steps for index-based catalogs are references to configmaps. This avoids the problem
 		// of installplans growing beyond the etcd size limit when manifests are written to the ip status.
diff --git a/test/e2e/subscription_e2e_test.go b/test/e2e/subscription_e2e_test.go
index 766c0a531..0c968997f 100644
--- a/test/e2e/subscription_e2e_test.go
+++ b/test/e2e/subscription_e2e_test.go
@@ -4,6 +4,7 @@ import (
 	"context"
 	"encoding/json"
 	"fmt"
+	"path/filepath"
 	"reflect"
 	"strings"
 	"sync"
@@ -22,11 +23,14 @@ import (
 	apierrors "k8s.io/apimachinery/pkg/api/errors"
 	"k8s.io/apimachinery/pkg/api/resource"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/types"
 	"k8s.io/apimachinery/pkg/util/wait"
 	"k8s.io/client-go/discovery"
 	"sigs.k8s.io/controller-runtime/pkg/client"
 
 	"github.com/operator-framework/api/pkg/lib/version"
+	operatorsv1 "github.com/operator-framework/api/pkg/operators/v1"
+	"github.com/operator-framework/api/pkg/operators/v1alpha1"
 	operatorsv1alpha1 "github.com/operator-framework/api/pkg/operators/v1alpha1"
 	"github.com/operator-framework/operator-lifecycle-manager/pkg/api/client/clientset/versioned"
 	"github.com/operator-framework/operator-lifecycle-manager/pkg/controller/install"
@@ -50,6 +54,7 @@ const (
 var _ = Describe("Subscription", func() {
 	var (
 		generatedNamespace corev1.Namespace
+		operatorGroup      operatorsv1.OperatorGroup
 		c                  operatorclient.ClientInterface
 		crc                versioned.Interface
 	)
@@ -57,7 +62,15 @@ var _ = Describe("Subscription", func() {
 	BeforeEach(func() {
 		c = ctx.Ctx().KubeClient()
 		crc = ctx.Ctx().OperatorClient()
-		generatedNamespace = SetupGeneratedTestNamespace(genName("subscription-e2e-"))
+
+		nsName := genName("subscription-e2e-")
+		operatorGroup = operatorsv1.OperatorGroup{
+			ObjectMeta: metav1.ObjectMeta{
+				Name:      fmt.Sprintf("%s-operatorgroup", nsName),
+				Namespace: nsName,
+			},
+		}
+		generatedNamespace = SetupGeneratedTestNamespaceWithOperatorGroup(nsName, operatorGroup)
 	})
 
 	AfterEach(func() {
@@ -2329,6 +2342,186 @@ var _ = Describe("Subscription", func() {
 			)))
 		})
 	})
+
+	It("unpacks bundle image", func() {
+		catsrc := &operatorsv1alpha1.CatalogSource{
+			ObjectMeta: metav1.ObjectMeta{
+				Name:      genName("kiali-"),
+				Namespace: generatedNamespace.GetName(),
+				Labels:    map[string]string{"olm.catalogSource": "kaili-catalog"},
+			},
+			Spec: operatorsv1alpha1.CatalogSourceSpec{
+				Image:      "quay.io/operator-framework/ci-index:latest",
+				SourceType: operatorsv1alpha1.SourceTypeGrpc,
+				GrpcPodConfig: &operatorsv1alpha1.GrpcPodConfig{
+					SecurityContextConfig: operatorsv1alpha1.Restricted,
+				},
+			},
+		}
+		catsrc, err := crc.OperatorsV1alpha1().CatalogSources(catsrc.GetNamespace()).Create(context.Background(), catsrc, metav1.CreateOptions{})
+		require.NoError(GinkgoT(), err)
+		defer func() {
+			Eventually(func() error {
+				return client.IgnoreNotFound(ctx.Ctx().Client().Delete(context.Background(), catsrc))
+			}).Should(Succeed())
+		}()
+
+		By("waiting for the CatalogSource to be ready")
+		catsrc, err = fetchCatalogSourceOnStatus(crc, catsrc.GetName(), catsrc.GetNamespace(), catalogSourceRegistryPodSynced)
+		require.NoError(GinkgoT(), err)
+
+		By("generating a Subscription")
+		subName := genName("kiali-")
+		cleanUpSubscriptionFn := createSubscriptionForCatalog(crc, catsrc.GetNamespace(), subName, catsrc.GetName(), "kiali", stableChannel, "", operatorsv1alpha1.ApprovalAutomatic)
+		defer cleanUpSubscriptionFn()
+
+		By("waiting for the InstallPlan to get created for the subscription")
+		sub, err := fetchSubscription(crc, catsrc.GetNamespace(), subName, subscriptionHasInstallPlanChecker)
+		require.NoError(GinkgoT(), err)
+
+		By("waiting for the expected InstallPlan's execution to either fail or succeed")
+		ipName := sub.Status.InstallPlanRef.Name
+		ip, err := waitForInstallPlan(crc, ipName, sub.GetNamespace(), buildInstallPlanPhaseCheckFunc(operatorsv1alpha1.InstallPlanPhaseFailed, operatorsv1alpha1.InstallPlanPhaseComplete))
+		require.NoError(GinkgoT(), err)
+		require.Equal(GinkgoT(), operatorsv1alpha1.InstallPlanPhaseComplete, ip.Status.Phase, "InstallPlan not complete")
+
+		By("ensuring the InstallPlan contains the steps resolved from the bundle image")
+		operatorName := "kiali-operator"
+		expectedSteps := map[registry.ResourceKey]struct{}{
+			{Name: operatorName, Kind: "ClusterServiceVersion"}:                                  {},
+			{Name: "kialis.kiali.io", Kind: "CustomResourceDefinition"}:                          {},
+			{Name: "monitoringdashboards.monitoring.kiali.io", Kind: "CustomResourceDefinition"}: {},
+			{Name: operatorName, Kind: "ServiceAccount"}:                                         {},
+			{Name: operatorName, Kind: "ClusterRole"}:                                            {},
+			{Name: operatorName, Kind: "ClusterRoleBinding"}:                                     {},
+		}
+		require.Lenf(GinkgoT(), ip.Status.Plan, len(expectedSteps), "number of expected steps does not match installed: %v", ip.Status.Plan)
+
+		for _, step := range ip.Status.Plan {
+			key := registry.ResourceKey{
+				Name: step.Resource.Name,
+				Kind: step.Resource.Kind,
+			}
+			for expected := range expectedSteps {
+				if strings.HasPrefix(key.Name, expected.Name) && key.Kind == expected.Kind {
+					delete(expectedSteps, expected)
+				}
+			}
+		}
+		require.Lenf(GinkgoT(), expectedSteps, 0, "Actual resource steps do not match expected: %#v", expectedSteps)
+	})
+
+	When("unpacking bundle", func() {
+		var (
+			magicCatalog      *MagicCatalog
+			catalogSourceName string
+			subName           string
+		)
+
+		BeforeEach(func() {
+			By("deploying the testing catalog")
+			provider, err := NewFileBasedFiledBasedCatalogProvider(filepath.Join(testdataDir, failForwardTestDataBaseDir, "example-operator.v0.1.0.yaml"))
+			Expect(err).To(BeNil())
+			catalogSourceName = fmt.Sprintf("%s-catsrc", generatedNamespace.GetName())
+			magicCatalog = NewMagicCatalog(ctx.Ctx().Client(), generatedNamespace.GetName(), catalogSourceName, provider)
+			Expect(magicCatalog.DeployCatalog(context.Background())).To(BeNil())
+
+			By("creating the testing subscription")
+			subName = fmt.Sprintf("%s-packagea-sub", generatedNamespace.GetName())
+			createSubscriptionForCatalog(crc, generatedNamespace.GetName(), subName, catalogSourceName, "packageA", stableChannel, "", operatorsv1alpha1.ApprovalAutomatic)
+
+			By("waiting until the subscription has an IP reference")
+			subscription, err := fetchSubscription(crc, generatedNamespace.GetName(), subName, subscriptionHasInstallPlanChecker)
+			Expect(err).Should(BeNil())
+
+			By("waiting for the v0.1.0 CSV to report a succeeded phase")
+			_, err = fetchCSV(crc, subscription.Status.CurrentCSV, generatedNamespace.GetName(), buildCSVConditionChecker(operatorsv1alpha1.CSVPhaseSucceeded))
+			Expect(err).ShouldNot(HaveOccurred())
+		})
+
+		It("should not report unpacking progress or errors after successfull unpacking", func() {
+			By("verifying that the subscription is not reporting unpacking progress")
+			Eventually(
+				func() (corev1.ConditionStatus, error) {
+					fetched, err := crc.OperatorsV1alpha1().Subscriptions(generatedNamespace.GetName()).Get(context.Background(), subName, metav1.GetOptions{})
+					if err != nil {
+						return "", err
+					}
+					cond := fetched.Status.GetCondition(v1alpha1.SubscriptionBundleUnpacking)
+					return cond.Status, nil
+				},
+				5*time.Minute,
+				interval,
+			).Should(Equal(corev1.ConditionFalse))
+
+			By("verifying that the subscription is not reporting unpacking errors")
+			Eventually(
+				func() (corev1.ConditionStatus, error) {
+					fetched, err := crc.OperatorsV1alpha1().Subscriptions(generatedNamespace.GetName()).Get(context.Background(), subName, metav1.GetOptions{})
+					if err != nil {
+						return "", err
+					}
+					cond := fetched.Status.GetCondition(v1alpha1.SubscriptionBundleUnpackFailed)
+					return cond.Status, nil
+				},
+				5*time.Minute,
+				interval,
+			).Should(Equal(corev1.ConditionUnknown))
+		})
+
+		Context("with bundle which OLM will fail to unpack", func() {
+			BeforeEach(func() {
+				By("patching the OperatorGroup to reduce the bundle unpacking timeout")
+				ogNN := types.NamespacedName{Name: operatorGroup.GetName(), Namespace: generatedNamespace.GetName()}
+				addBundleUnpackTimeoutOGAnnotation(context.Background(), ctx.Ctx().Client(), ogNN, "1s")
+
+				By("updating the catalog with a broken v0.2.0 bundle image")
+				brokenProvider, err := NewFileBasedFiledBasedCatalogProvider(filepath.Join(testdataDir, failForwardTestDataBaseDir, "example-operator.v0.2.0-non-existent-tag.yaml"))
+				Expect(err).To(BeNil())
+				err = magicCatalog.UpdateCatalog(context.Background(), brokenProvider)
+				Expect(err).To(BeNil())
+			})
+
+			It("should expose a condition indicating failure to unpack", func() {
+				By("verifying that the subscription is reporting bundle unpack failure condition")
+				Eventually(
+					func() (string, error) {
+						fetched, err := crc.OperatorsV1alpha1().Subscriptions(generatedNamespace.GetName()).Get(context.Background(), subName, metav1.GetOptions{})
+						if err != nil {
+							return "", err
+						}
+						cond := fetched.Status.GetCondition(v1alpha1.SubscriptionBundleUnpackFailed)
+						if cond.Status != corev1.ConditionTrue || cond.Reason != "BundleUnpackFailed" {
+							return "", fmt.Errorf("%s condition not found", v1alpha1.SubscriptionBundleUnpackFailed)
+						}
+
+						return cond.Message, nil
+					},
+					5*time.Minute,
+					interval,
+				).Should(ContainSubstring("bundle unpacking failed. Reason: DeadlineExceeded"))
+
+				By("waiting for the subscription to maintain the example-operator.v0.1.0 status.currentCSV")
+				Consistently(subscriptionCurrentCSVGetter(crc, generatedNamespace.GetName(), subName)).Should(Equal("example-operator.v0.1.0"))
+			})
+
+			It("should be able to recover when catalog gets updated with a fixed version", func() {
+				By("patching the OperatorGroup to reduce the bundle unpacking timeout")
+				ogNN := types.NamespacedName{Name: operatorGroup.GetName(), Namespace: generatedNamespace.GetName()}
+				addBundleUnpackTimeoutOGAnnotation(context.Background(), ctx.Ctx().Client(), ogNN, "5m")
+
+				By("updating the catalog with a fixed v0.2.0 bundle image")
+				brokenProvider, err := NewFileBasedFiledBasedCatalogProvider(filepath.Join(testdataDir, failForwardTestDataBaseDir, "example-operator.v0.2.0.yaml"))
+				Expect(err).To(BeNil())
+				err = magicCatalog.UpdateCatalog(context.Background(), brokenProvider)
+				Expect(err).To(BeNil())
+
+				By("waiting for the subscription to have v0.2.0 installed")
+				_, err = fetchSubscription(crc, generatedNamespace.GetName(), subName, subscriptionHasCurrentCSV("example-operator.v0.2.0"))
+				Expect(err).Should(BeNil())
+			})
+		})
+	})
 })
 
 const (
@@ -2951,3 +3144,25 @@ func updateCatSrcPriority(crClient versioned.Interface, namespace string, catsrc
 	_, err := crClient.OperatorsV1alpha1().CatalogSources(namespace).Update(context.Background(), catsrc, metav1.UpdateOptions{})
 	Expect(err).Should(BeNil())
 }
+
+func subscriptionCurrentCSVGetter(crclient versioned.Interface, namespace, subName string) func() string {
+	return func() string {
+		subscription, err := crclient.OperatorsV1alpha1().Subscriptions(namespace).Get(context.Background(), subName, metav1.GetOptions{})
+		if err != nil || subscription == nil {
+			return ""
+		}
+		return subscription.Status.CurrentCSV
+	}
+}
+
+func operatorGroupServiceAccountNameSetter(crclient versioned.Interface, namespace, name, saName string) func() error {
+	return func() error {
+		toUpdate, err := crclient.OperatorsV1().OperatorGroups(namespace).Get(context.Background(), name, metav1.GetOptions{})
+		if err != nil {
+			return err
+		}
+		toUpdate.Spec.ServiceAccountName = saName
+		_, err = crclient.OperatorsV1().OperatorGroups(namespace).Update(context.Background(), toUpdate, metav1.UpdateOptions{})
+		return err
+	}
+}
diff --git a/test/e2e/testdata/fail-forward/base/example-operator.v0.2.0-2.yaml b/test/e2e/testdata/fail-forward/base/example-operator.v0.2.0-invalid-csv.yaml
similarity index 100%
rename from test/e2e/testdata/fail-forward/base/example-operator.v0.2.0-2.yaml
rename to test/e2e/testdata/fail-forward/base/example-operator.v0.2.0-invalid-csv.yaml
diff --git a/test/e2e/testdata/fail-forward/base/example-operator.v0.2.0-non-existent-tag.yaml b/test/e2e/testdata/fail-forward/base/example-operator.v0.2.0-non-existent-tag.yaml
new file mode 100644
index 000000000..555242e53
--- /dev/null
+++ b/test/e2e/testdata/fail-forward/base/example-operator.v0.2.0-non-existent-tag.yaml
@@ -0,0 +1,26 @@
+---
+schema: olm.package
+name: packageA
+defaultChannel: stable
+---
+schema: olm.channel
+package: packageA
+name: stable
+entries:
+  - name: example-operator.v0.2.0
+    replaces: example-operator.v0.1.0
+---
+schema: olm.bundle
+name: example-operator.v0.2.0
+package: packageA
+image: quay.io/olmtest/example-operator-bundle:non-existent-tag
+properties:
+  - type: olm.gvk
+    value:
+      group: example.com
+      kind: TestA
+      version: v1alpha1
+  - type: olm.package
+    value:
+      packageName: packageA
+      version: 1.0.1
diff --git a/test/e2e/testdata/fail-forward/base/example-operator.v0.2.0.yaml b/test/e2e/testdata/fail-forward/base/example-operator.v0.2.0.yaml
index 555242e53..3058376dd 100644
--- a/test/e2e/testdata/fail-forward/base/example-operator.v0.2.0.yaml
+++ b/test/e2e/testdata/fail-forward/base/example-operator.v0.2.0.yaml
@@ -13,7 +13,7 @@ entries:
 schema: olm.bundle
 name: example-operator.v0.2.0
 package: packageA
-image: quay.io/olmtest/example-operator-bundle:non-existent-tag
+image: quay.io/olmtest/example-operator-bundle:0.2.0
 properties:
   - type: olm.gvk
     value: