Repository: crossplane/crossplane

Issue:
Migrate xp-extract from up to crossplane CLI
TL;DR: `crossplane` CLI is missing the `xpkg xp-extract` command, but `up` has it, but `xpkg` is marked as depricated in favor of `crossplane xpkg`. Can we migrate `xp-extract` from `up` to `crossplane`? 😄 

### What problem are you facing?
I recently went down a little rabbit hole trying to install my custom provider without pushing it to a remote repository. I got a lot of help from the Crossplane Slack channel, but there is one issue. 
The solution depends on a subcommand in `up` that is both undocumented and the parent command itself is marked as deprecated in favor of the same command in the `crossplane` CLI. 

The problem is that the `crossplane` CLI does not have the `xp-extract` subcommand. Both the solutions to my issues, proposed by @haarchri and @jbw976, depends on this. 
1. https://github.com/jbw976/change-log-sidecar/blob/main/DEVELOPMENT.md
2. https://github.com/crossplane/build/blob/1ed19332b947c449795fd016f3c21ee0a64930fd/makelib/local.xpkg.mk#L34

Currently I have a working setup, but the deprecation of the `up xpkg` command and that it is missing from `crosplane xpkg` concerns me. 

### How could Crossplane help solve your problem?
Please migrate the `xp-extract` command to Crossplane, as it has been really useful so far!
I am of course open to any better alternatives, if they exist! The only thing I need it for is to create a `.gz` from a Crossplane `.xpkg` to use it in the Crossplane's package cache. 

Slack thread for context: https://crossplane.slack.com/archives/CEG3T90A1/p1737556112272989

Hints from issue comments:
@scoplin: Would also love to see this particular workflow ported, especially given the partially-hidden and [deprecated status](https://docs.upbound.io/reference/cli/rel-notes/#v0350) of the `xp-extract` function in the upbound cli v0.35.0.  Seems like local testing of composition functions (which may have a larger community of authors than even providers) will be in need of it.

Solution patch:
diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml
index 5de3a1013..205bf13ac 100644
--- a/.github/workflows/ci.yml
+++ b/.github/workflows/ci.yml
@@ -227,7 +227,6 @@ jobs:
       matrix:
         test-suite:
           - base
-          - realtime-compositions
           - package-dependency-updates
           - package-signature-verification
 
@@ -288,7 +287,7 @@ jobs:
       - name: Run E2E Tests
         run: |
           earthly --strict --allow-privileged --remote-cache ghcr.io/crossplane/earthly-cache:${{ github.job }}-${{ matrix.test-suite}} \
-            +e2e --FLAGS="-test.failfast -fail-fast -prior-crossplane-version=${CROSSPLANE_PRIOR_VERSION} --test-suite ${{ matrix.test-suite }}"
+            +e2e --GOTESTSUM_FORMAT="testname" --FLAGS="-test.failfast -fail-fast -prior-crossplane-version=${CROSSPLANE_PRIOR_VERSION} --test-suite ${{ matrix.test-suite }}"
 
       - name: Publish E2E Test Flakes
         if: '!cancelled()'
diff --git a/Earthfile b/Earthfile
index c35a30fff..d35883250 100644
--- a/Earthfile
+++ b/Earthfile
@@ -51,6 +51,7 @@ e2e:
   ARG GOARCH=${TARGETARCH}
   ARG GOOS=${TARGETOS}
   ARG FLAGS="-test-suite=base"
+  ARG GOTESTSUM_FORMAT="standard-verbose"
   # Using earthly image to allow compatibility with different development environments e.g. WSL
   FROM earthly/dind:alpine-3.20-docker-26.1.5-r0
   RUN wget https://dl.google.com/go/go${GO_VERSION}.${GOOS}-${GOARCH}.tar.gz
@@ -71,7 +72,12 @@ e2e:
     WITH DOCKER --load crossplane-e2e/crossplane:latest=(+image --CROSSPLANE_VERSION=v0.0.0-e2e)
       # TODO(negz:) Set GITHUB_ACTIONS=true and use RUN --raw-output when
       # https://github.com/earthly/earthly/issues/4143 is fixed.
-      RUN gotestsum --no-color=false --format testname --junitfile e2e-tests.xml --raw-command go tool test2json -t -p E2E ./e2e -test.v ${FLAGS}
+      RUN gotestsum \
+        --rerun-fails \
+      	--no-color=false \
+	--format ${GOTESTSUM_FORMAT} \
+	--junitfile e2e-tests.xml \
+	--raw-command go tool test2json -t -p E2E ./e2e -test.v ${FLAGS}
     END
   FINALLY
     SAVE ARTIFACT --if-exists e2e-tests.xml AS LOCAL _output/tests/e2e-tests.xml
diff --git a/OWNERS.md b/OWNERS.md
index 16f71732a..8a25b9ca5 100644
--- a/OWNERS.md
+++ b/OWNERS.md
@@ -15,11 +15,11 @@ See [CODEOWNERS](CODEOWNERS) for automatic PR assignment.
 * Hasan Turken <hasan@upbound.io> ([turkenh](https://github.com/turkenh))
 * Bob Haddleton <bob.haddleton@nokia.com> ([bobh66](https://github.com/bobh66))
 * Philippe Scorsolini <philippe.scorsolini@upbound.io> ([phisco](https://github.com/phisco))
+* Jared Watts <jared@upbound.io> ([jbw976](https://github.com/jbw976))
 
 ## Emeritus Maintainers
 
 * Bassam Tabbara <bassam@upbound.io> ([bassam](https://github.com/bassam))
-* Jared Watts <jared@upbound.io> ([jbw976](https://github.com/jbw976))
 * Illya Chekrygin <illya.chekrygin@gmail.com> ([ichekrygin](https://github.com/ichekrygin))
 * Daniel Mangum <georgedanielmangum@gmail.com> ([hasheddan](https://github.com/hasheddan))
 * Muvaffak Onus <me@muvaf.com> ([muvaf](https://github.com/muvaf))
diff --git a/apis/apiextensions/fn/proto/v1/run_function.pb.go b/apis/apiextensions/fn/proto/v1/run_function.pb.go
index cf1be5171..7a26d1dd1 100644
--- a/apis/apiextensions/fn/proto/v1/run_function.pb.go
+++ b/apis/apiextensions/fn/proto/v1/run_function.pb.go
@@ -283,7 +283,7 @@ type RunFunctionRequest struct {
 	// Optional input specific to this Function invocation. A JSON representation
 	// of the 'input' block of the relevant entry in a Composition's pipeline.
 	Input *structpb.Struct `protobuf:"bytes,4,opt,name=input,proto3,oneof" json:"input,omitempty"`
-	// Optional context. Crossplane may pass arbitary contextual information to a
+	// Optional context. Crossplane may pass arbitrary contextual information to a
 	// Function. A Function may also return context in its RunFunctionResponse,
 	// and that context will be passed to subsequent Functions. Crossplane
 	// discards all context returned by the last Function in the pipeline.
@@ -897,9 +897,9 @@ type ResponseMeta struct {
 	// An opaque string identifying the content of the request. Must match the
 	// meta.tag of the corresponding RunFunctionRequest.
 	Tag string `protobuf:"bytes,1,opt,name=tag,proto3" json:"tag,omitempty"`
-	// Time-to-live of this response. Deterministic Functions with no side-effects
-	// (e.g. simple templating Functions) may specify a TTL. Crossplane may choose
-	// to cache responses until the TTL expires.
+	// Time-to-live of this response. Crossplane will call the function again when
+	// the TTL expires. Crossplane may cache the response to avoid calling the
+	// function again until the TTL expires.
 	Ttl *durationpb.Duration `protobuf:"bytes,2,opt,name=ttl,proto3,oneof" json:"ttl,omitempty"`
 }
 
diff --git a/apis/apiextensions/fn/proto/v1/run_function.proto b/apis/apiextensions/fn/proto/v1/run_function.proto
index e8417b68e..3d5b5ab48 100644
--- a/apis/apiextensions/fn/proto/v1/run_function.proto
+++ b/apis/apiextensions/fn/proto/v1/run_function.proto
@@ -55,7 +55,7 @@ message RunFunctionRequest {
   // of the 'input' block of the relevant entry in a Composition's pipeline.
   optional google.protobuf.Struct input = 4;
 
-  // Optional context. Crossplane may pass arbitary contextual information to a
+  // Optional context. Crossplane may pass arbitrary contextual information to a
   // Function. A Function may also return context in its RunFunctionResponse,
   // and that context will be passed to subsequent Functions. Crossplane
   // discards all context returned by the last Function in the pipeline.
@@ -170,9 +170,9 @@ message ResponseMeta {
   // meta.tag of the corresponding RunFunctionRequest.
   string tag = 1;
 
-  // Time-to-live of this response. Deterministic Functions with no side-effects
-  // (e.g. simple templating Functions) may specify a TTL. Crossplane may choose
-  // to cache responses until the TTL expires.
+  // Time-to-live of this response. Crossplane will call the function again when
+  // the TTL expires. Crossplane may cache the response to avoid calling the
+  // function again until the TTL expires.
   optional google.protobuf.Duration ttl = 2;
 }
 
diff --git a/apis/apiextensions/fn/proto/v1beta1/zz_generated_run_function.pb.go b/apis/apiextensions/fn/proto/v1beta1/zz_generated_run_function.pb.go
index a07c45c61..e0d36b766 100644
--- a/apis/apiextensions/fn/proto/v1beta1/zz_generated_run_function.pb.go
+++ b/apis/apiextensions/fn/proto/v1beta1/zz_generated_run_function.pb.go
@@ -285,7 +285,7 @@ type RunFunctionRequest struct {
 	// Optional input specific to this Function invocation. A JSON representation
 	// of the 'input' block of the relevant entry in a Composition's pipeline.
 	Input *structpb.Struct `protobuf:"bytes,4,opt,name=input,proto3,oneof" json:"input,omitempty"`
-	// Optional context. Crossplane may pass arbitary contextual information to a
+	// Optional context. Crossplane may pass arbitrary contextual information to a
 	// Function. A Function may also return context in its RunFunctionResponse,
 	// and that context will be passed to subsequent Functions. Crossplane
 	// discards all context returned by the last Function in the pipeline.
@@ -899,9 +899,9 @@ type ResponseMeta struct {
 	// An opaque string identifying the content of the request. Must match the
 	// meta.tag of the corresponding RunFunctionRequest.
 	Tag string `protobuf:"bytes,1,opt,name=tag,proto3" json:"tag,omitempty"`
-	// Time-to-live of this response. Deterministic Functions with no side-effects
-	// (e.g. simple templating Functions) may specify a TTL. Crossplane may choose
-	// to cache responses until the TTL expires.
+	// Time-to-live of this response. Crossplane will call the function again when
+	// the TTL expires. Crossplane may cache the response to avoid calling the
+	// function again until the TTL expires.
 	Ttl *durationpb.Duration `protobuf:"bytes,2,opt,name=ttl,proto3,oneof" json:"ttl,omitempty"`
 }
 
diff --git a/apis/apiextensions/fn/proto/v1beta1/zz_generated_run_function.proto b/apis/apiextensions/fn/proto/v1beta1/zz_generated_run_function.proto
index 578b6a975..0d20d07e0 100644
--- a/apis/apiextensions/fn/proto/v1beta1/zz_generated_run_function.proto
+++ b/apis/apiextensions/fn/proto/v1beta1/zz_generated_run_function.proto
@@ -57,7 +57,7 @@ message RunFunctionRequest {
   // of the 'input' block of the relevant entry in a Composition's pipeline.
   optional google.protobuf.Struct input = 4;
 
-  // Optional context. Crossplane may pass arbitary contextual information to a
+  // Optional context. Crossplane may pass arbitrary contextual information to a
   // Function. A Function may also return context in its RunFunctionResponse,
   // and that context will be passed to subsequent Functions. Crossplane
   // discards all context returned by the last Function in the pipeline.
@@ -172,9 +172,9 @@ message ResponseMeta {
   // meta.tag of the corresponding RunFunctionRequest.
   string tag = 1;
 
-  // Time-to-live of this response. Deterministic Functions with no side-effects
-  // (e.g. simple templating Functions) may specify a TTL. Crossplane may choose
-  // to cache responses until the TTL expires.
+  // Time-to-live of this response. Crossplane will call the function again when
+  // the TTL expires. Crossplane may cache the response to avoid calling the
+  // function again until the TTL expires.
   optional google.protobuf.Duration ttl = 2;
 }
 
diff --git a/apis/apiextensions/v1alpha1/zz_generated.environment_config_types.go b/apis/apiextensions/v1alpha1/zz_generated.environment_config_types.go
index da7efdc6e..87b884290 100644
--- a/apis/apiextensions/v1alpha1/zz_generated.environment_config_types.go
+++ b/apis/apiextensions/v1alpha1/zz_generated.environment_config_types.go
@@ -24,7 +24,6 @@ import (
 )
 
 // +kubebuilder:object:root=true
-// +kubebuilder:storageversion
 // +genclient
 // +genclient:nonNamespaced
 
diff --git a/apis/apiextensions/v1beta1/environment_config_types.go b/apis/apiextensions/v1beta1/environment_config_types.go
index 8de6ae5bf..aaa36845c 100644
--- a/apis/apiextensions/v1beta1/environment_config_types.go
+++ b/apis/apiextensions/v1beta1/environment_config_types.go
@@ -22,6 +22,7 @@ import (
 )
 
 // +kubebuilder:object:root=true
+// +kubebuilder:storageversion
 // +genclient
 // +genclient:nonNamespaced
 
diff --git a/apis/generate.go b/apis/generate.go
index eb25fea67..f370b3262 100644
--- a/apis/generate.go
+++ b/apis/generate.go
@@ -31,7 +31,7 @@ limitations under the License.
 //go:generate ../hack/duplicate_api_type.sh apiextensions/v1/composition_patches.go apiextensions/v1beta1
 //go:generate ../hack/duplicate_api_type.sh apiextensions/v1/composition_transforms.go apiextensions/v1beta1
 
-//go:generate ../hack/duplicate_api_type.sh apiextensions/v1beta1/environment_config_types.go apiextensions/v1alpha1 true
+//go:generate ../hack/duplicate_api_type.sh apiextensions/v1beta1/environment_config_types.go apiextensions/v1alpha1 false
 //go:generate ../hack/duplicate_api_type.sh apiextensions/v1beta1/usage_types.go apiextensions/v1alpha1 true
 
 //go:generate ../hack/duplicate_api_type.sh pkg/v1/package_types.go pkg/v1beta1
diff --git a/cluster/charts/crossplane/README.md b/cluster/charts/crossplane/README.md
index d847b2f40..df0fbdc0a 100644
--- a/cluster/charts/crossplane/README.md
+++ b/cluster/charts/crossplane/README.md
@@ -119,6 +119,7 @@ and their default values.
 | `resourcesRBACManager.requests.cpu` | CPU resource requests for the RBAC Manager pod. | `"100m"` |
 | `resourcesRBACManager.requests.memory` | Memory resource requests for the RBAC Manager pod. | `"256Mi"` |
 | `revisionHistoryLimit` | The number of Crossplane ReplicaSets to retain. | `nil` |
+| `runtimeClassName` | The runtimeClassName name to apply to the Crossplane and RBAC Manager pods. | `""` |
 | `securityContextCrossplane.allowPrivilegeEscalation` | Enable `allowPrivilegeEscalation` for the Crossplane pod. | `false` |
 | `securityContextCrossplane.readOnlyRootFilesystem` | Set the Crossplane pod root file system as read-only. | `true` |
 | `securityContextCrossplane.runAsGroup` | The group ID used by the Crossplane pod. | `65532` |
diff --git a/cluster/charts/crossplane/templates/deployment.yaml b/cluster/charts/crossplane/templates/deployment.yaml
index ea66975db..48bb1c8a4 100644
--- a/cluster/charts/crossplane/templates/deployment.yaml
+++ b/cluster/charts/crossplane/templates/deployment.yaml
@@ -47,6 +47,9 @@ spec:
       {{- if .Values.priorityClassName }}
       priorityClassName: {{ .Values.priorityClassName  | quote }}
       {{- end }}
+      {{- if .Values.runtimeClassName }}
+      runtimeClassName: {{ .Values.runtimeClassName | quote }}
+      {{- end }}
       {{- if not .Values.serviceAccount.create }}
       serviceAccountName: {{ .Values.serviceAccount.name }}
       {{- else }}
diff --git a/cluster/charts/crossplane/templates/rbac-manager-deployment.yaml b/cluster/charts/crossplane/templates/rbac-manager-deployment.yaml
index cf91b99c1..a04d7232d 100644
--- a/cluster/charts/crossplane/templates/rbac-manager-deployment.yaml
+++ b/cluster/charts/crossplane/templates/rbac-manager-deployment.yaml
@@ -48,6 +48,9 @@ spec:
       priorityClassName: {{ .Values.priorityClassName | quote }}
       {{- end }}
       serviceAccountName: rbac-manager
+      {{- if .Values.runtimeClassName }}
+      runtimeClassName: {{ .Values.runtimeClassName | quote }}
+      {{- end }}
       initContainers:
       - image: "{{ .Values.image.repository }}:{{ .Values.image.tag | default (printf "v%s" .Chart.AppVersion) }}"
         args:
diff --git a/cluster/charts/crossplane/values.yaml b/cluster/charts/crossplane/values.yaml
index e21825091..34857580a 100755
--- a/cluster/charts/crossplane/values.yaml
+++ b/cluster/charts/crossplane/values.yaml
@@ -112,6 +112,9 @@ packageManager:
 # -- The PriorityClass name to apply to the Crossplane and RBAC Manager pods.
 priorityClassName: ""
 
+# -- The runtimeClassName name to apply to the Crossplane and RBAC Manager pods.
+runtimeClassName: ""
+
 resourcesCrossplane:
   limits:
     # -- CPU resource limits for the Crossplane pod.
diff --git a/cluster/crds/apiextensions.crossplane.io_environmentconfigs.yaml b/cluster/crds/apiextensions.crossplane.io_environmentconfigs.yaml
index 0919a0017..065e14ece 100644
--- a/cluster/crds/apiextensions.crossplane.io_environmentconfigs.yaml
+++ b/cluster/crds/apiextensions.crossplane.io_environmentconfigs.yaml
@@ -58,7 +58,7 @@ spec:
             type: object
         type: object
     served: true
-    storage: true
+    storage: false
     subresources: {}
   - additionalPrinterColumns:
     - jsonPath: .metadata.creationTimestamp
@@ -100,5 +100,5 @@ spec:
             type: object
         type: object
     served: true
-    storage: false
+    storage: true
     subresources: {}
diff --git a/cmd/crank/xpkg/extract.go b/cmd/crank/xpkg/extract.go
new file mode 100644
index 000000000..9cce9f04d
--- /dev/null
+++ b/cmd/crank/xpkg/extract.go
@@ -0,0 +1,213 @@
+/*
+Copyright 2025 The Crossplane Authors.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package xpkg
+
+import (
+	"archive/tar"
+	"compress/gzip"
+	"context"
+	"io"
+	"os"
+	"path/filepath"
+	"time"
+
+	"github.com/google/go-containerregistry/pkg/name"
+	v1 "github.com/google/go-containerregistry/pkg/v1"
+	"github.com/google/go-containerregistry/pkg/v1/daemon"
+	"github.com/google/go-containerregistry/pkg/v1/mutate"
+	"github.com/google/go-containerregistry/pkg/v1/remote"
+	"github.com/google/go-containerregistry/pkg/v1/tarball"
+	"github.com/spf13/afero"
+
+	"github.com/crossplane/crossplane-runtime/pkg/errors"
+	"github.com/crossplane/crossplane-runtime/pkg/logging"
+
+	"github.com/crossplane/crossplane/internal/xpkg"
+	"github.com/crossplane/crossplane/internal/xpkg/upbound"
+)
+
+const (
+	errMustProvideTag          = "must provide package tag if fetching from registry or daemon"
+	errInvalidTag              = "package tag is not a valid reference"
+	errFetchPackage            = "failed to fetch package from remote"
+	errGetManifest             = "failed to get package image manifest from remote"
+	errFetchLayer              = "failed to fetch annotated base layer from remote"
+	errGetUncompressed         = "failed to get uncompressed contents from layer"
+	errMultipleAnnotatedLayers = "package is invalid due to multiple annotated base layers"
+	errOpenPackageStream       = "failed to open package stream file"
+	errCreateOutputFile        = "failed to create output file"
+	errCreateGzipWriter        = "failed to create gzip writer"
+	errExtractPackageContents  = "failed to extract package contents"
+	cacheContentExt            = ".gz"
+)
+
+// fetchFn fetches a package from a source.
+type fetchFn func(context.Context, name.Reference) (v1.Image, error)
+
+// registryFetch fetches a package from the registry.
+func registryFetch(ctx context.Context, r name.Reference) (v1.Image, error) {
+	return remote.Image(r, remote.WithContext(ctx))
+}
+
+// daemonFetch fetches a package from the Docker daemon.
+func daemonFetch(ctx context.Context, r name.Reference) (v1.Image, error) {
+	return daemon.Image(r, daemon.WithContext(ctx))
+}
+
+func xpkgFetch(path string) fetchFn {
+	return func(_ context.Context, _ name.Reference) (v1.Image, error) {
+		return tarball.ImageFromPath(filepath.Clean(path), nil)
+	}
+}
+
+// AfterApply constructs and binds context to any subcommands
+// that have Run() methods that receive it.
+func (c *extractCmd) AfterApply() error {
+	c.fs = afero.NewOsFs()
+	c.fetch = registryFetch
+	if c.FromDaemon {
+		c.fetch = daemonFetch
+	}
+	if c.FromXpkg {
+		// If package is not defined, attempt to find single package in current
+		// directory.
+		if c.Package == "" {
+			wd, err := os.Getwd()
+			if err != nil {
+				return errors.Wrap(err, errGetwd)
+			}
+			path, err := xpkg.FindXpkgInDir(c.fs, wd)
+			if err != nil {
+				return errors.Wrap(err, errFindPackageinWd)
+			}
+			c.Package = path
+		}
+		c.fetch = xpkgFetch(c.Package)
+	}
+	if !c.FromXpkg {
+		if c.Package == "" {
+			return errors.New(errMustProvideTag)
+		}
+		upCtx, err := upbound.NewFromFlags(c.Flags)
+		if err != nil {
+			return err
+		}
+
+		name, err := name.ParseReference(c.Package, name.WithDefaultRegistry(upCtx.RegistryEndpoint.Hostname()))
+		if err != nil {
+			return errors.Wrap(err, errInvalidTag)
+		}
+		c.name = name
+	}
+	return nil
+}
+
+// extractCmd extracts package contents into a Crossplane cache compatible
+// format.
+type extractCmd struct {
+	fs    afero.Fs
+	name  name.Reference
+	fetch fetchFn
+
+	Package    string `arg:""                                                                                                                                                     help:"Name of the package to extract. Must be a valid OCI image tag or a path if using --from-xpkg." optional:""`
+	FromDaemon bool   `help:"Indicates that the image should be fetched from the Docker daemon."`
+	FromXpkg   bool   `help:"Indicates that the image should be fetched from a local xpkg. If package is not specified and only one exists in current directory it will be used."`
+	Output     string `default:"out.gz"                                                                                                                                           help:"Package output file path. Extension must be .gz or will be replaced."                          short:"o"`
+
+	// Common API configuration
+	Flags upbound.Flags `embed:""`
+}
+
+// Run runs the xpkg extract cmd.
+func (c *extractCmd) Run(logger logging.Logger) error { //nolint:gocyclo // xpkg extract for cli
+	ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
+	defer cancel()
+
+	// Fetch package.
+	img, err := c.fetch(ctx, c.name)
+	if err != nil {
+		return errors.Wrap(err, errFetchPackage)
+	}
+
+	// Get image manifest.
+	manifest, err := img.Manifest()
+	if err != nil {
+		return errors.Wrap(err, errGetManifest)
+	}
+
+	// Determine if the image is using annotated layers.
+	var tarc io.ReadCloser
+	foundAnnotated := false
+	for _, l := range manifest.Layers {
+		if a, ok := l.Annotations[xpkg.AnnotationKey]; !ok || a != xpkg.PackageAnnotation {
+			continue
+		}
+		if foundAnnotated {
+			return errors.New(errMultipleAnnotatedLayers)
+		}
+		foundAnnotated = true
+		layer, err := img.LayerByDigest(l.Digest)
+		if err != nil {
+			return errors.Wrap(err, errFetchLayer)
+		}
+		tarc, err = layer.Uncompressed()
+		if err != nil {
+			return errors.Wrap(err, errGetUncompressed)
+		}
+	}
+
+	// If we still don't have content then we need to flatten image filesystem.
+	if !foundAnnotated {
+		tarc = mutate.Extract(img)
+	}
+
+	// The ReadCloser is an uncompressed tarball, either consisting of annotated
+	// layer contents or flattened filesystem content. Either way, we only want
+	// the package YAML stream.
+	t := tar.NewReader(tarc)
+	var size int64
+	for {
+		h, err := t.Next()
+		if err != nil {
+			return errors.Wrap(err, errOpenPackageStream)
+		}
+		if h.Name == xpkg.StreamFile {
+			size = h.Size
+			break
+		}
+	}
+
+	out := xpkg.ReplaceExt(filepath.Clean(c.Output), cacheContentExt)
+	cf, err := c.fs.Create(out)
+	if err != nil {
+		return errors.Wrap(err, errCreateOutputFile)
+	}
+	defer cf.Close() //nolint:errcheck // defer close
+	w, err := gzip.NewWriterLevel(cf, gzip.BestSpeed)
+	if err != nil {
+		return errors.Wrap(err, errCreateGzipWriter)
+	}
+	if _, err = io.CopyN(w, t, size); err != nil {
+		return errors.Wrap(err, errExtractPackageContents)
+	}
+	if err := w.Close(); err != nil {
+		return errors.Wrap(err, errExtractPackageContents)
+	}
+
+	logger.Debug("xpkg contents extracted to %s", out)
+	return nil
+}
diff --git a/cmd/crank/xpkg/xpkg.go b/cmd/crank/xpkg/xpkg.go
index f996600ee..50a2c731a 100644
--- a/cmd/crank/xpkg/xpkg.go
+++ b/cmd/crank/xpkg/xpkg.go
@@ -29,13 +29,14 @@ type Cmd struct {
 	Logout  logoutCmd  `cmd:"" help:"Logout of the default package registry."`
 	Push    pushCmd    `cmd:"" help:"Push a package to a registry."`
 	Update  updateCmd  `cmd:"" help:"Update a package in a control plane."`
+	Extract extractCmd `cmd:"" help:"Extract package contents into a Crossplane cache compatible format. Fetches from a remote registry by default."`
 }
 
 // Help prints out the help for the xpkg command.
 func (c *Cmd) Help() string {
 	return `
 Crossplane can be extended using packages. Crossplane packages are called xpkgs.
-Crossplane supports configuration, provider and function packages. 
+Crossplane supports configuration, provider and function packages.
 
 A package is an opinionated OCI image that contains everything needed to extend
 a Crossplane control plane with new functionality. For example installing a
diff --git a/cmd/crossplane/core/core.go b/cmd/crossplane/core/core.go
index 6fb0a92c9..a743a1f60 100644
--- a/cmd/crossplane/core/core.go
+++ b/cmd/crossplane/core/core.go
@@ -116,7 +116,6 @@ type startCommand struct {
 	TLSClientCertsDir   string `env:"TLS_CLIENT_CERTS_DIR"   help:"The path of the folder which will store TLS client certificate of Crossplane."`
 
 	EnableExternalSecretStores      bool `group:"Alpha Features:" help:"Enable support for External Secret Stores."`
-	EnableRealtimeCompositions      bool `group:"Alpha Features:" help:"Enable support for realtime compositions, i.e. watching composed resources and reconciling compositions immediately when any of the composed resources is updated."`
 	EnableDependencyVersionUpgrades bool `group:"Alpha Features:" help:"Enable support for upgrading dependency versions when the parent package is updated."`
 	EnableSignatureVerification     bool `group:"Alpha Features:" help:"Enable support for package signature verification via ImageConfig API."`
 
@@ -124,6 +123,7 @@ type startCommand struct {
 	EnableDeploymentRuntimeConfigs           bool `default:"true" group:"Beta Features:" help:"Enable support for Deployment Runtime Configs."`
 	EnableUsages                             bool `default:"true" group:"Beta Features:" help:"Enable support for deletion ordering and resource protection with Usages."`
 	EnableSSAClaims                          bool `default:"true" group:"Beta Features:" help:"Enable support for using Kubernetes server-side apply to sync claims with composite resources (XRs)."`
+	EnableRealtimeCompositions               bool `default:"true" group:"Beta Features:" help:"Enable support for realtime compositions, i.e. watching composed resources and reconciling compositions immediately when any of the composed resources is updated."`
 
 	// These are GA features that previously had alpha or beta feature flags.
 	// You can't turn off a GA feature. We maintain the flags to avoid breaking
@@ -244,14 +244,14 @@ func (c *startCommand) Run(s *runtime.Scheme, log logging.Logger) error { //noli
 		return errors.Wrap(err, "cannot load client TLS certificates")
 	}
 
-	m := xfn.NewMetrics()
-	metrics.Registry.MustRegister(m)
+	xfnm := xfn.NewPrometheusMetrics()
+	metrics.Registry.MustRegister(xfnm)
 
 	// We want all XR controllers to share the same gRPC clients.
 	functionRunner := xfn.NewPackagedFunctionRunner(mgr.GetClient(),
 		xfn.WithLogger(log),
 		xfn.WithTLSConfig(clienttls),
-		xfn.WithInterceptorCreators(m),
+		xfn.WithInterceptorCreators(xfnm),
 	)
 
 	// Periodically remove clients for Functions that no longer exist.
@@ -283,8 +283,8 @@ func (c *startCommand) Run(s *runtime.Scheme, log logging.Logger) error { //noli
 		}
 	}
 	if c.EnableRealtimeCompositions {
-		o.Features.Enable(features.EnableAlphaRealtimeCompositions)
-		log.Info("Alpha feature enabled", "flag", features.EnableAlphaRealtimeCompositions)
+		o.Features.Enable(features.EnableBetaRealtimeCompositions)
+		log.Info("Beta feature enabled", "flag", features.EnableBetaRealtimeCompositions)
 	}
 	if c.EnableDeploymentRuntimeConfigs {
 		o.Features.Enable(features.EnableBetaDeploymentRuntimeConfigs)
@@ -375,6 +375,9 @@ func (c *startCommand) Run(s *runtime.Scheme, log logging.Logger) error { //noli
 		return errors.Wrap(err, "cannot create uncached client for API extension controllers")
 	}
 
+	cem := engine.NewPrometheusMetrics()
+	metrics.Registry.MustRegister(cem)
+
 	// It's important the engine's client is wrapped with unstructured.NewClient
 	// because controller-runtime always caches *unstructured.Unstructured, not
 	// our wrapper types like *composite.Unstructured. This client takes care of
@@ -384,6 +387,7 @@ func (c *startCommand) Run(s *runtime.Scheme, log logging.Logger) error { //noli
 		unstructured.NewClient(cached),
 		unstructured.NewClient(uncached),
 		engine.WithLogger(log),
+		engine.WithMetrics(cem),
 	)
 
 	// TODO(negz): Garbage collect informers for CRs that are still defined
diff --git a/internal/controller/apiextensions/claim/reconciler.go b/internal/controller/apiextensions/claim/reconciler.go
index baeb5c77c..29594af6a 100644
--- a/internal/controller/apiextensions/claim/reconciler.go
+++ b/internal/controller/apiextensions/claim/reconciler.go
@@ -98,7 +98,7 @@ type CompositeSyncer interface {
 // composite resource (XR).
 type CompositeSyncerFn func(ctx context.Context, cm *claim.Unstructured, xr *composite.Unstructured) error
 
-// Sync the supplied claim with the supplied composite resource..
+// Sync the supplied claim with the supplied composite resource.
 func (fn CompositeSyncerFn) Sync(ctx context.Context, cm *claim.Unstructured, xr *composite.Unstructured) error {
 	return fn(ctx, cm, xr)
 }
@@ -188,9 +188,8 @@ type Reconciler struct {
 	composite crComposite
 	claim     crClaim
 
-	log          logging.Logger
-	record       event.Recorder
-	pollInterval time.Duration
+	log    logging.Logger
+	record event.Recorder
 }
 
 type crComposite struct {
@@ -275,17 +274,6 @@ func WithRecorder(er event.Recorder) ReconcilerOption {
 	}
 }
 
-// WithPollInterval specifies how long the Reconciler should wait before queueing
-// a new reconciliation after a successful reconcile. The Reconciler requeues
-// after a specified duration when it is not actively waiting for an external
-// operation, but wishes to check whether resources it does not have a watch on
-// (i.e. composed resources) need to be reconciled.
-func WithPollInterval(after time.Duration) ReconcilerOption {
-	return func(r *Reconciler) {
-		r.pollInterval = after
-	}
-}
-
 // NewReconciler returns a Reconciler that reconciles composite resource claims of
 // the supplied CompositeClaimKind with resources of the supplied CompositeKind.
 // The returned Reconciler will apply only the ObjectMetaConfigurator by
diff --git a/internal/controller/apiextensions/composite/api.go b/internal/controller/apiextensions/composite/api.go
index c89ce4d57..89127e13d 100644
--- a/internal/controller/apiextensions/composite/api.go
+++ b/internal/controller/apiextensions/composite/api.go
@@ -128,13 +128,13 @@ func (a *APIFilteredSecretPublisher) UnpublishConnection(_ context.Context, _ re
 // for compatibility with existing Composition logic while CompositionRevisions
 // are in alpha.
 type APIRevisionFetcher struct {
-	ca resource.ClientApplicator
+	client client.Client
 }
 
 // NewAPIRevisionFetcher returns a RevisionFetcher that fetches the
 // Revision referenced by a composite resource.
-func NewAPIRevisionFetcher(ca resource.ClientApplicator) *APIRevisionFetcher {
-	return &APIRevisionFetcher{ca: ca}
+func NewAPIRevisionFetcher(c client.Client) *APIRevisionFetcher {
+	return &APIRevisionFetcher{client: c}
 }
 
 // Fetch the appropriate CompositionRevision for the supplied XR. Panics if the
@@ -148,7 +148,7 @@ func (f *APIRevisionFetcher) Fetch(ctx context.Context, cr resource.Composite) (
 	// Just fetch and return the selected revision.
 	if current != nil && pol != nil && *pol == xpv1.UpdateManual {
 		rev := &v1.CompositionRevision{}
-		err := f.ca.Get(ctx, types.NamespacedName{Name: current.Name}, rev)
+		err := f.client.Get(ctx, types.NamespacedName{Name: current.Name}, rev)
 		return rev, errors.Wrap(err, errGetCompositionRevision)
 	}
 
@@ -156,7 +156,7 @@ func (f *APIRevisionFetcher) Fetch(ctx context.Context, cr resource.Composite) (
 	// automatic. Either way we need to determine the latest revision.
 
 	comp := &v1.Composition{}
-	if err := f.ca.Get(ctx, meta.NamespacedNameOf(cr.GetCompositionReference()), comp); err != nil {
+	if err := f.client.Get(ctx, meta.NamespacedNameOf(cr.GetCompositionReference()), comp); err != nil {
 		return nil, errors.Wrap(err, errGetComposition)
 	}
 
@@ -172,7 +172,7 @@ func (f *APIRevisionFetcher) Fetch(ctx context.Context, cr resource.Composite) (
 
 	if current == nil || current.Name != latest.GetName() {
 		cr.SetCompositionRevisionReference(&corev1.LocalObjectReference{Name: latest.GetName()})
-		if err := f.ca.Apply(ctx, cr); err != nil {
+		if err := f.client.Update(ctx, cr); err != nil {
 			return nil, errors.Wrap(err, errUpdate)
 		}
 	}
@@ -190,7 +190,7 @@ func (f *APIRevisionFetcher) getCompositionRevisionList(ctx context.Context, cr
 	}
 
 	ml[v1.LabelCompositionName] = comp.GetName()
-	if err := f.ca.List(ctx, rl, ml); err != nil {
+	if err := f.client.List(ctx, rl, ml); err != nil {
 		return nil, errors.Wrap(err, errListCompositionRevisions)
 	}
 	return rl, nil
diff --git a/internal/controller/apiextensions/composite/composed.go b/internal/controller/apiextensions/composite/composed.go
index fbb4d1091..8d89a5c75 100644
--- a/internal/controller/apiextensions/composite/composed.go
+++ b/internal/controller/apiextensions/composite/composed.go
@@ -22,15 +22,6 @@ import (
 	v1 "github.com/crossplane/crossplane/apis/apiextensions/v1"
 )
 
-// A CompositeResource is an output of the composition process.
-type CompositeResource struct { //nolint:revive // stick with CompositeResource
-	// Ready indicated whether the composite resource should be marked as
-	// ready or unready regardless of the state of the composed resoureces.
-	// If it is nil the readiness of the composite is determined by the
-	// readiness of the composed resources.
-	Ready *bool
-}
-
 // A ResourceName uniquely identifies the composed resource within a Composition
 // and within Composition Function gRPC calls. This is not the metadata.name of
 // the actual composed resource instance; rather it is the name of an entry in a
diff --git a/internal/controller/apiextensions/composite/composition_functions.go b/internal/controller/apiextensions/composite/composition_functions.go
index 6b75ec952..e658999c9 100644
--- a/internal/controller/apiextensions/composite/composition_functions.go
+++ b/internal/controller/apiextensions/composite/composition_functions.go
@@ -21,6 +21,7 @@ import (
 	"fmt"
 	"sort"
 	"strings"
+	"time"
 
 	"google.golang.org/protobuf/encoding/protojson"
 	"google.golang.org/protobuf/types/known/structpb"
@@ -269,6 +270,11 @@ func (c *FunctionComposer) Compose(ctx context.Context, xr *composite.Unstructur
 		return CompositionResult{}, errors.Wrap(err, errBuildObserved)
 	}
 
+	// Time-to-live for this composition pipeline run. Each function returns
+	// a TTL. The pipeline's TTL will be the shortest non-zero TTL returned
+	// by any function. A TTL of zero means unlimited TTL.
+	var ttl time.Duration
+
 	// The Function pipeline starts with empty desired state.
 	d := &fnv1.State{}
 
@@ -319,6 +325,13 @@ func (c *FunctionComposer) Compose(ctx context.Context, xr *composite.Unstructur
 			return CompositionResult{}, errors.Wrapf(err, errFmtRunPipelineStep, fn.Step)
 		}
 
+		// If this Function specified a non-zero TTL that's less than
+		// the current recorded TTL for the pipeline, it's the new TTL
+		// for the pipeline.
+		if d := rsp.GetMeta().GetTtl().AsDuration(); d > 0 && (ttl == 0 || d < ttl) {
+			ttl = d
+		}
+
 		// Pass the desired state returned by this Function to the next one.
 		d = rsp.GetDesired()
 
@@ -426,16 +439,6 @@ func (c *FunctionComposer) Compose(ctx context.Context, xr *composite.Unstructur
 		}
 	}
 
-	compositeRes := CompositeResource{}
-
-	// Consider the explicit composite unready state in the function response:
-	switch d.GetComposite().GetReady() { //nolint:exhaustive // only check for false or true
-	case fnv1.Ready_READY_TRUE:
-		compositeRes.Ready = ptr.To(true)
-	case fnv1.Ready_READY_FALSE:
-		compositeRes.Ready = ptr.To(false)
-	}
-
 	// Garbage collect any observed resources that aren't part of our final
 	// desired state. We must do this before we update the XR's resource
 	// references to ensure that we don't forget and leak them if a delete
@@ -550,7 +553,26 @@ func (c *FunctionComposer) Compose(ctx context.Context, xr *composite.Unstructur
 		return CompositionResult{}, errors.Wrap(err, errApplyXRStatus)
 	}
 
-	return CompositionResult{ConnectionDetails: d.GetComposite().GetConnectionDetails(), Composite: compositeRes, Composed: resources, Events: events, Conditions: conditions}, nil
+	var ready *bool
+	switch d.GetComposite().GetReady() {
+	case fnv1.Ready_READY_TRUE:
+		ready = ptr.To(true)
+	case fnv1.Ready_READY_FALSE:
+		ready = ptr.To(false)
+	case fnv1.Ready_READY_UNSPECIFIED:
+		// Remains nil.
+	}
+
+	result := CompositionResult{
+		Composed:          resources,
+		ConnectionDetails: d.GetComposite().GetConnectionDetails(),
+		Ready:             ready,
+		Events:            events,
+		Conditions:        conditions,
+		TTL:               ttl,
+	}
+
+	return result, nil
 }
 
 // ComposedFieldOwnerName generates a unique field owner name
diff --git a/internal/controller/apiextensions/composite/reconciler.go b/internal/controller/apiextensions/composite/reconciler.go
index 0bf8379c9..8aba73c61 100644
--- a/internal/controller/apiextensions/composite/reconciler.go
+++ b/internal/controller/apiextensions/composite/reconciler.go
@@ -36,6 +36,7 @@ import (
 	xpv1 "github.com/crossplane/crossplane-runtime/apis/common/v1"
 	"github.com/crossplane/crossplane-runtime/pkg/errors"
 	"github.com/crossplane/crossplane-runtime/pkg/event"
+	"github.com/crossplane/crossplane-runtime/pkg/feature"
 	"github.com/crossplane/crossplane-runtime/pkg/logging"
 	"github.com/crossplane/crossplane-runtime/pkg/meta"
 	"github.com/crossplane/crossplane-runtime/pkg/reconciler/managed"
@@ -46,12 +47,12 @@ import (
 
 	v1 "github.com/crossplane/crossplane/apis/apiextensions/v1"
 	"github.com/crossplane/crossplane/internal/engine"
+	"github.com/crossplane/crossplane/internal/features"
 )
 
 const (
-	timeout             = 2 * time.Minute
-	defaultPollInterval = 1 * time.Minute
-	finalizer           = "composite.apiextensions.crossplane.io"
+	timeout   = 2 * time.Minute
+	finalizer = "composite.apiextensions.crossplane.io"
 )
 
 // Error strings.
@@ -66,6 +67,7 @@ const (
 	errFetchComp              = "cannot fetch Composition"
 	errConfigure              = "cannot configure composite resource"
 	errPublish                = "cannot publish connection details"
+	errWatch                  = "cannot watch resource for changes"
 	errUnpublish              = "cannot unpublish connection details"
 	errValidate               = "refusing to use invalid Composition"
 	errAssociate              = "cannot associate composed resources with Composition resource templates"
@@ -84,6 +86,7 @@ const (
 	reasonResolve event.Reason = "SelectComposition"
 	reasonCompose event.Reason = "ComposeResources"
 	reasonPublish event.Reason = "PublishConnectionSecret"
+	reasonWatch   event.Reason = "WatchComposedResources"
 	reasonInit    event.Reason = "InitializeCompositeResource"
 	reasonDelete  event.Reason = "DeleteCompositeResource"
 	reasonPaused  event.Reason = "ReconciliationPaused"
@@ -154,11 +157,23 @@ type CompositionRequest struct {
 
 // A CompositionResult is the result of the composition process.
 type CompositionResult struct {
-	Composite         CompositeResource
-	Composed          []ComposedResource
+	// Composed resource details.
+	Composed []ComposedResource
+
+	// XR connection details.
 	ConnectionDetails managed.ConnectionDetails
-	Events            []TargetedEvent
-	Conditions        []TargetedCondition
+
+	// XR readiness. When nil readiness is derived from composed resources.
+	Ready *bool
+
+	// XR and claim events.
+	Events []TargetedEvent
+
+	// XR and claim conditions.
+	Conditions []TargetedCondition
+
+	// TTL for this composition result.
+	TTL time.Duration
 }
 
 // A CompositionTarget is the target of a composition event or condition.
@@ -240,29 +255,20 @@ func WithRecorder(er event.Recorder) ReconcilerOption {
 	}
 }
 
-// A PollIntervalHook determines how frequently the XR should poll its composed
-// resources.
-type PollIntervalHook func(ctx context.Context, xr *composite.Unstructured) time.Duration
-
-// WithPollIntervalHook specifies how to determine how long the Reconciler
-// should wait before queueing a new reconciliation after a successful
-// reconcile.
-func WithPollIntervalHook(h PollIntervalHook) ReconcilerOption {
+// WithFeatures specifies what feature flags the Reconciler should enable.
+func WithFeatures(f *feature.Flags) ReconcilerOption {
 	return func(r *Reconciler) {
-		r.pollInterval = h
+		r.features = f
 	}
 }
 
 // WithPollInterval specifies how long the Reconciler should wait before
 // queueing a new reconciliation after a successful reconcile. The Reconciler
-// uses the interval jittered +/- 10% when all composed resources are ready. It
-// polls twice as frequently (i.e. at half the supplied interval) +/- 10% when
-// waiting for composed resources to become ready.
+// uses the interval jittered +/- 10%.
 func WithPollInterval(interval time.Duration) ReconcilerOption {
-	return WithPollIntervalHook(func(_ context.Context, _ *composite.Unstructured) time.Duration {
-		// Jitter the poll interval +/- 10%.
-		return interval + time.Duration((rand.Float64()-0.5)*2*(float64(interval)*0.1)) //nolint:gosec // No need for secure randomness
-	})
+	return func(r *Reconciler) {
+		r.pollInterval = interval
+	}
 }
 
 // WithCompositionRevisionFetcher specifies how the composition to be used should be
@@ -355,21 +361,23 @@ func (fn CompositionRevisionValidatorFn) Validate(c *v1.CompositionRevision) err
 // start watches when they compose new kinds of resources.
 type WatchStarter interface {
 	// StartWatches starts the supplied watches, if they're not running already.
-	StartWatches(name string, ws ...engine.Watch) error
+	StartWatches(ctx context.Context, name string, ws ...engine.Watch) error
 }
 
 // A NopWatchStarter does nothing.
 type NopWatchStarter struct{}
 
 // StartWatches does nothing.
-func (n *NopWatchStarter) StartWatches(_ string, _ ...engine.Watch) error { return nil }
+func (n *NopWatchStarter) StartWatches(_ context.Context, _ string, _ ...engine.Watch) error {
+	return nil
+}
 
 // A WatchStarterFn is a function that can start a new watch.
-type WatchStarterFn func(name string, ws ...engine.Watch) error
+type WatchStarterFn func(ctx context.Context, name string, ws ...engine.Watch) error
 
 // StartWatches starts the supplied watches, if they're not running already.
-func (fn WatchStarterFn) StartWatches(name string, ws ...engine.Watch) error {
-	return fn(name, ws...)
+func (fn WatchStarterFn) StartWatches(ctx context.Context, name string, ws ...engine.Watch) error {
+	return fn(ctx, name, ws...)
 }
 
 type compositeResource struct {
@@ -419,8 +427,6 @@ func NewReconciler(c, uc client.Client, of resource.CompositeKind, opts ...Recon
 
 		log:    logging.NewNopLogger(),
 		record: event.NewNopRecorder(),
-
-		pollInterval: func(_ context.Context, _ *composite.Unstructured) time.Duration { return defaultPollInterval },
 	}
 
 	for _, f := range opts {
@@ -435,6 +441,8 @@ type Reconciler struct {
 	client client.Client
 	gvk    schema.GroupVersionKind
 
+	features *feature.Flags
+
 	revision  revision
 	composite compositeResource
 
@@ -448,7 +456,7 @@ type Reconciler struct {
 	log    logging.Logger
 	record event.Recorder
 
-	pollInterval PollIntervalHook
+	pollInterval time.Duration
 }
 
 // Reconcile a composite resource.
@@ -519,6 +527,9 @@ func (r *Reconciler) Reconcile(ctx context.Context, req reconcile.Request) (reco
 
 	orig := xr.GetCompositionReference()
 	if err := r.composite.SelectComposition(ctx, xr); err != nil {
+		if kerrors.IsConflict(err) {
+			return reconcile.Result{Requeue: true}, nil
+		}
 		err = errors.Wrap(err, errSelectComp)
 		r.record.Event(xr, event.Warning(reasonResolve, err))
 		xr.SetConditions(xpv1.ReconcileError(err))
@@ -532,6 +543,9 @@ func (r *Reconciler) Reconcile(ctx context.Context, req reconcile.Request) (reco
 	origRev := xr.GetCompositionRevisionReference()
 	rev, err := r.revision.Fetch(ctx, xr)
 	if err != nil {
+		if kerrors.IsConflict(err) {
+			return reconcile.Result{Requeue: true}, nil
+		}
 		log.Debug(errFetchComp, "error", err)
 		err = errors.Wrap(err, errFetchComp)
 		r.record.Event(xr, event.Warning(reasonCompose, err))
@@ -542,8 +556,6 @@ func (r *Reconciler) Reconcile(ctx context.Context, req reconcile.Request) (reco
 		r.record.Event(xr, event.Normal(reasonResolve, fmt.Sprintf("Selected composition revision: %s", rev.Name)))
 	}
 
-	// TODO(negz): Update this to validate the revision? In practice that's what
-	// it's doing today when revis are enabled.
 	if err := r.revision.Validate(rev); err != nil {
 		log.Debug(errValidate, "error", err)
 		err = errors.Wrap(err, errValidate)
@@ -607,12 +619,13 @@ func (r *Reconciler) Reconcile(ctx context.Context, req reconcile.Request) (reco
 		ws[i] = engine.WatchFor(composed.New(composed.FromReference(ref)), engine.WatchTypeComposedResource, r.watchHandler)
 	}
 
-	// StartWatches is a no-op unless the realtime compositions feature flag is
-	// enabled. When the flag is enabled, the ControllerEngine that starts this
-	// controller also starts a garbage collector for its watches.
-	if err := r.engine.StartWatches(r.controllerName, ws...); err != nil {
-		// TODO(negz): If we stop polling this will be a more serious error.
-		log.Debug("Cannot start watches for composed resources. Relying on polling to know when they change.", "controller-name", r.controllerName, "error", err)
+	// The ControllerEngine that starts this controller also starts a
+	// garbage collector for its watches.
+	if err := r.engine.StartWatches(ctx, r.controllerName, ws...); err != nil {
+		err = errors.Wrap(err, errWatch)
+		r.record.Event(xr, event.Warning(reasonWatch, err))
+		xr.SetConditions(xpv1.ReconcileError(err))
+		return reconcile.Result{Requeue: true}, errors.Wrap(r.client.Status().Update(ctx, xr), errUpdateStatus)
 	}
 
 	published, err := r.composite.PublishConnection(ctx, xr, res.ConnectionDetails)
@@ -642,8 +655,8 @@ func (r *Reconciler) Reconcile(ctx context.Context, req reconcile.Request) (reco
 		log.Debug("Successfully composed resources")
 	}
 
-	var unready []ComposedResource
-	var unsynced []ComposedResource
+	var unsynced []string
+	var unready []string
 	for i, cd := range res.Composed {
 		// Specifying a name for P&T templates is optional but encouraged.
 		// If there was no name, fall back to using the index.
@@ -654,67 +667,56 @@ func (r *Reconciler) Reconcile(ctx context.Context, req reconcile.Request) (reco
 
 		if !cd.Synced {
 			log.Debug("Composed resource is not yet valid", "id", id)
-			unsynced = append(unsynced, cd)
+			unsynced = append(unsynced, id)
 			r.record.Event(xr, event.Normal(reasonCompose, fmt.Sprintf("Composed resource %q is not yet valid", id)))
 		}
 
 		if !cd.Ready {
 			log.Debug("Composed resource is not yet ready", "id", id)
-			unready = append(unready, cd)
+			unready = append(unready, id)
 			r.record.Event(xr, event.Normal(reasonCompose, fmt.Sprintf("Composed resource %q is not yet ready", id)))
 		}
 	}
 
-	if updateXRConditions(xr, unsynced, unready, res) {
-		// This requeue is subject to rate limiting. Requeues will exponentially
-		// backoff from 1 to 30 seconds. See the 'definition' (XRD) reconciler
-		// that sets up the ratelimiter.
-		return reconcile.Result{Requeue: true}, errors.Wrap(r.client.Status().Update(ctx, xr), errUpdateStatus)
-	}
-
-	// We requeue after our poll interval because we can't watch composed
-	// resources - we can't know what type of resources we might compose
-	// when this controller is started.
-	return reconcile.Result{RequeueAfter: r.pollInterval(ctx, xr)}, errors.Wrap(r.client.Status().Update(ctx, xr), errUpdateStatus)
-}
-
-// updateXRConditions updates the conditions of the supplied composite resource
-// based on the supplied composed resources. It returns true if the XR should be
-// requeued immediately.
-func updateXRConditions(xr *composite.Unstructured, unsynced, unready []ComposedResource, res CompositionResult) (requeueImmediately bool) {
-	readyCond := xpv1.Available()
-	syncedCond := xpv1.ReconcileSuccess()
+	synced := xpv1.ReconcileSuccess()
 	if len(unsynced) > 0 {
-		// We want to requeue to wait for our composed resources to
-		// become ready, since we can't watch them.
-		syncedCond = xpv1.ReconcileError(errors.New(errSyncResources)).WithMessage(fmt.Sprintf("Invalid resources: %s", resource.StableNAndSomeMore(resource.DefaultFirstN, getComposerResourcesNames(unsynced))))
-		requeueImmediately = true
+		synced = xpv1.ReconcileError(errors.New(errSyncResources)).WithMessage(fmt.Sprintf("Unsynced resources: %s", resource.StableNAndSomeMore(resource.DefaultFirstN, unsynced)))
 	}
+
+	ready := xpv1.Available()
 	if len(unready) > 0 {
-		// We want to requeue to wait for our composed resources to
-		// become ready, since we can't watch them.
-		readyCond = xpv1.Creating().WithMessage(fmt.Sprintf("Unready resources: %s", resource.StableNAndSomeMore(resource.DefaultFirstN, getComposerResourcesNames(unready))))
-		requeueImmediately = true
-	}
-	if res.Composite.Ready != nil {
-		if *res.Composite.Ready {
-			readyCond = xpv1.Available()
-		} else if readyCond.Status != corev1.ConditionFalse {
-			// To keep information about unready resources only set this status
-			// if the composite would be otherwise marked as ready.
-			readyCond = xpv1.Creating().WithMessage("Composite resource was explicitly marked as unready by the composer")
+		ready = xpv1.Creating().WithMessage(fmt.Sprintf("Unready resources: %s", resource.StableNAndSomeMore(resource.DefaultFirstN, unready)))
+	}
+
+	// If the composer explicitly specified the XR's readiness it
+	// supersedes readiness derived from composed resources.
+	if res.Ready != nil {
+		ready = xpv1.Creating()
+		if *res.Ready {
+			ready = xpv1.Available()
 		}
 	}
-	xr.SetConditions(syncedCond, readyCond)
-	return requeueImmediately
-}
 
-func getComposerResourcesNames(cds []ComposedResource) []string {
-	names := make([]string, len(cds))
-	for i, cd := range cds {
-		names[i] = string(cd.ResourceName)
+	xr.SetConditions(synced, ready)
+
+	// Requeue after the configured poll interval by default. If realtime
+	// compositions is enabled this'll be RequeueAfter: 0, i.e. no requeue.
+	result := reconcile.Result{RequeueAfter: jitter(r.pollInterval)}
+
+	switch {
+	case !r.features.Enabled(features.EnableBetaRealtimeCompositions) && len(unsynced)+len(unready) > 0:
+		// Realtime compositions isn't enabled, and one of our composed
+		// resources is unsynced or unready. Requeue immediately
+		// (subject to backoff) while we wait for them.
+		result = reconcile.Result{Requeue: true}
+	case res.TTL > 0:
+		// The composer (e.g. the function pipeline) explicitly returned
+		// a TTL for the composition result. Requeue after the TTL
+		// expires.
+		result = reconcile.Result{RequeueAfter: jitter(res.TTL)}
 	}
-	return names
+
+	return result, errors.Wrap(r.client.Status().Update(ctx, xr), errUpdateStatus)
 }
 
 type compositionResultMeta struct {
@@ -787,3 +789,8 @@ func getClaimFromXR(ctx context.Context, c client.Client, xr *composite.Unstruct
 	}
 	return cm, nil
 }
+
+// Jitter the supplied duration by up to +/- 10%.
+func jitter(d time.Duration) time.Duration {
+	return d + time.Duration((rand.Float64()-0.5)*2*(float64(d)*0.1)) //nolint:gosec // No need for secure randomness
+}
diff --git a/internal/controller/apiextensions/composite/watch/watch.go b/internal/controller/apiextensions/composite/watch/watch.go
index d8a139934..5933c1294 100644
--- a/internal/controller/apiextensions/composite/watch/watch.go
+++ b/internal/controller/apiextensions/composite/watch/watch.go
@@ -44,8 +44,8 @@ type ControllerEngine interface {
 // A GarbageCollector garbage collects watches for a single composite resource
 // (XR) controller. A watch is eligible for garbage collection when none of the
 // XRs the controller owns resource references its GVK. The garbage collector
-// periodically lists all of the controller's XRs to determine what GVKs they
-// still reference.
+// periodically lists all the controller's XRs to determine what GVKs they still
+// reference.
 type GarbageCollector struct {
 	controllerName string
 	xrGVK          schema.GroupVersionKind
@@ -103,7 +103,28 @@ func (gc *GarbageCollector) GarbageCollectWatches(ctx context.Context, interval
 // longer composed by any composite resource (XR). It's safe to call from
 // multiple goroutines.
 func (gc *GarbageCollector) GarbageCollectWatchesNow(ctx context.Context) error {
-	// List all XRs of the type we're interested in.
+	// Get the set of running watches.
+	running, err := gc.engine.GetWatches(gc.controllerName)
+	if err != nil {
+		return errors.Wrap(err, "cannot get running watches")
+	}
+
+	composed := make([]engine.WatchID, 0, len(running))
+	for _, wid := range running {
+		// Only stop watches for composed resources. The other watch
+		// types should only be stopped when the XR controller stops.
+		if wid.Type == engine.WatchTypeComposedResource {
+			composed = append(composed, wid)
+		}
+	}
+
+	// No composed resource watches exist. Nothing to do.
+	if len(composed) == 0 {
+		return nil
+	}
+
+	// List all XRs of the type we're interested in. This list is from
+	// cache, so it could be stale.
 	l := &kunstructured.UnstructuredList{}
 	l.SetAPIVersion(gc.xrGVK.GroupVersion().String())
 	l.SetKind(gc.xrGVK.Kind + "List")
@@ -112,25 +133,69 @@ func (gc *GarbageCollector) GarbageCollectWatchesNow(ctx context.Context) error
 	}
 
 	// Build the set of GVKs they still reference.
-	used := make(map[engine.WatchID]bool)
+	used := make(map[schema.GroupVersionKind]bool)
 	for _, u := range l.Items {
 		xr := &composite.Unstructured{Unstructured: u}
 		for _, ref := range xr.GetResourceReferences() {
-			used[engine.WatchID{Type: engine.WatchTypeComposedResource, GVK: schema.FromAPIVersionAndKind(ref.APIVersion, ref.Kind)}] = true
+			used[schema.FromAPIVersionAndKind(ref.APIVersion, ref.Kind)] = true
 		}
 	}
 
-	// Get the set of running watches.
-	running, err := gc.engine.GetWatches(gc.controllerName)
-	if err != nil {
-		return errors.Wrap(err, "cannot get running watches")
+	stop := make([]engine.WatchID, 0)
+	for _, wid := range composed {
+		// Only stop watches for types no XR composes.
+		if used[wid.GVK] {
+			continue
+		}
+		stop = append(stop, wid)
 	}
 
-	stop := make([]engine.WatchID, 0)
-	for _, wid := range running {
-		if !used[wid] {
-			stop = append(stop, wid)
+	// We listed from cache, so the set of 'used' GVKs still composed by at
+	// least one XR could be stale. For example a watch for kind: X exists,
+	// our stale cache told us an XR was still composing kind: X, but really
+	// it wasn't. That's okay. We'll stop the watch on the next GC cycle.
+	//
+	// What we really want to avoid is stopping a watch that we shouldn't.
+	// For example a watch for kind: X exists, our stale cache told us no XR
+	// was still composing kind: X, but really one was. e.g Because it
+	// started the watch very recently.
+	//
+	// So if it looks like there's no watches to stop we return early. If it
+	// looks like there _are_ watches to stop, we double check using an
+	// uncached client before stopping them.
+	if len(stop) == 0 {
+		return nil
+	}
+
+	// List all XRs again, this time using an uncached client.
+	l = &kunstructured.UnstructuredList{}
+	l.SetAPIVersion(gc.xrGVK.GroupVersion().String())
+	l.SetKind(gc.xrGVK.Kind + "List")
+	if err := gc.engine.GetUncached().List(ctx, l); err != nil {
+		return errors.Wrap(err, "cannot list composite resources")
+	}
+
+	// Build the set of GVKs they still reference.
+	used = make(map[schema.GroupVersionKind]bool)
+	for _, u := range l.Items {
+		xr := &composite.Unstructured{Unstructured: u}
+		for _, ref := range xr.GetResourceReferences() {
+			used[schema.FromAPIVersionAndKind(ref.APIVersion, ref.Kind)] = true
+		}
+	}
+
+	stop = make([]engine.WatchID, 0)
+	for _, wid := range composed {
+		// Only stop watches for composed resources. The other watch
+		// types should only be stopped when the XR controller stops.
+		if wid.Type != engine.WatchTypeComposedResource {
+			continue
+		}
+		// Only stop watches for types no XR composes.
+		if used[wid.GVK] {
+			continue
 		}
+		stop = append(stop, wid)
 	}
 
 	// No need to call StopWatches if there's nothing to stop.
@@ -145,10 +210,12 @@ func (gc *GarbageCollector) GarbageCollectWatchesNow(ctx context.Context) error
 	// watch was started that needs garbage collecting we'll get it eventually
 	// when GC runs again.
 	//
-	// It's also possible we'll stop a watch that's actually in use, if an XR
-	// started composing its GVK between when we built the map of used watches
-	// and here where we call StopWatches. It'll restart next time the XR
-	// controller calls StartWatches.
+	// It's also possible we'll stop a watch that's actually in use. This'd
+	// happen if an XR started composing its type after we listed XRs using
+	// the uncached client. We'll recover from this the next time the XR is
+	// reconciled, when it'll call StartWatches again. That could take some
+	// time though - at worst up to Crossplane's sync interval.
+	gc.log.Debug("Garbage collecting watches", "count", len(stop))
 	_, err = gc.engine.StopWatches(ctx, gc.controllerName, stop...)
 	return errors.Wrap(err, "cannot stop watches for composed resource types that are no longer referenced by any composite resource")
 }
diff --git a/internal/controller/apiextensions/definition/handlers.go b/internal/controller/apiextensions/definition/handlers.go
index 3bdad9f6e..8d9ff3208 100644
--- a/internal/controller/apiextensions/definition/handlers.go
+++ b/internal/controller/apiextensions/definition/handlers.go
@@ -43,39 +43,51 @@ func EnqueueForCompositionRevision(of resource.CompositeKind, c client.Reader, l
 		CreateFunc: func(ctx context.Context, e kevent.CreateEvent, q workqueue.TypedRateLimitingInterface[reconcile.Request]) {
 			rev, ok := e.Object.(*v1.CompositionRevision)
 			if !ok {
-				// should not happen
 				return
 			}
 
-			// TODO(negz): Check whether the revision's compositeTypeRef matches
-			// the supplied CompositeKind. If it doesn't, we can return early.
+			// We don't know what composition this revision is for,
+			// so we can't determine whether an XR might use it.
+			// This should never happen in practice - the
+			// composition controller sets this label when it
+			// creates a revision.
+			compName := rev.Labels[v1.LabelCompositionName]
+			if compName == "" {
+				return
+			}
+
+			// This handler is for a specific type of XR. This
+			// revisionisn't compatible with that type.
+			if rev.Spec.CompositeTypeRef.APIVersion != schema.GroupVersionKind(of).GroupVersion().String() {
+				return
+			}
+			if rev.Spec.CompositeTypeRef.Kind != of.Kind {
+				return
+			}
 
-			// get all XRs
 			xrs := kunstructured.UnstructuredList{}
 			xrs.SetGroupVersionKind(schema.GroupVersionKind(of))
 			xrs.SetKind(schema.GroupVersionKind(of).Kind + "List")
 			// TODO(negz): Index XRs by composition revision name?
 			if err := c.List(ctx, &xrs); err != nil {
-				// logging is most we can do here. This is a programming error if it happens.
+				// Logging is most we can do here. This is a programming error if it happens.
 				log.Info("cannot list in CompositionRevision handler", "type", schema.GroupVersionKind(of).String(), "error", err)
 				return
 			}
 
-			// enqueue all those that reference the Composition of this revision
-			compName := rev.Labels[v1.LabelCompositionName]
-			// TODO(negz): Check this before we get all XRs.
-			if compName == "" {
-				return
-			}
+			// Enqueue all those that reference the composition of
+			// this revision.
 			for _, u := range xrs.Items {
 				xr := composite.Unstructured{Unstructured: u}
 
-				// only automatic
+				// We only care about XRs that would
+				// automatically update to this new revision.
 				if pol := xr.GetCompositionUpdatePolicy(); pol != nil && *pol == xpv1.UpdateManual {
 					continue
 				}
 
-				// only those that reference the right Composition
+				// We only care about XRs that reference the
+				// composition this revision derives from.
 				if ref := xr.GetCompositionReference(); ref == nil || ref.Name != compName {
 					continue
 				}
diff --git a/internal/controller/apiextensions/definition/reconciler.go b/internal/controller/apiextensions/definition/reconciler.go
index d32a9bc41..e41a5fd3b 100644
--- a/internal/controller/apiextensions/definition/reconciler.go
+++ b/internal/controller/apiextensions/definition/reconciler.go
@@ -99,7 +99,7 @@ type ControllerEngine interface {
 	Stop(ctx context.Context, name string) error
 	IsRunning(name string) bool
 	GetWatches(name string) ([]engine.WatchID, error)
-	StartWatches(name string, ws ...engine.Watch) error
+	StartWatches(ctx context.Context, name string, ws ...engine.Watch) error
 	StopWatches(ctx context.Context, name string, ws ...engine.WatchID) (int, error)
 	GetCached() client.Client
 	GetUncached() client.Client
@@ -110,7 +110,9 @@ type ControllerEngine interface {
 type NopEngine struct{}
 
 // Start does nothing.
-func (e *NopEngine) Start(_ string, _ ...engine.ControllerOption) error { return nil }
+func (e *NopEngine) Start(_ string, _ ...engine.ControllerOption) error {
+	return nil
+}
 
 // Stop does nothing.
 func (e *NopEngine) Stop(_ context.Context, _ string) error { return nil }
@@ -122,7 +124,7 @@ func (e *NopEngine) IsRunning(_ string) bool { return true }
 func (e *NopEngine) GetWatches(_ string) ([]engine.WatchID, error) { return nil, nil }
 
 // StartWatches does nothing.
-func (e *NopEngine) StartWatches(_ string, _ ...engine.Watch) error { return nil }
+func (e *NopEngine) StartWatches(_ context.Context, _ string, _ ...engine.Watch) error { return nil }
 
 // StopWatches does nothing.
 func (e *NopEngine) StopWatches(_ context.Context, _ string, _ ...engine.WatchID) (int, error) {
@@ -220,7 +222,7 @@ func WithControllerEngine(c ControllerEngine) ReconcilerOption {
 	}
 }
 
-// WithCRDRenderer specifies how the Reconciler should render an
+// WithCRDRenderer specifies how the Reconciler should render a
 // CompositeResourceDefinition's corresponding CustomResourceDefinition.
 func WithCRDRenderer(c CRDRenderer) ReconcilerOption {
 	return func(r *Reconciler) {
@@ -494,7 +496,7 @@ func (r *Reconciler) Reconcile(ctx context.Context, req reconcile.Request) (reco
 	// ControllerOptions instead? It bothers me that this is the only feature
 	// flagged block outside that method.
 	co := []engine.ControllerOption{engine.WithRuntimeOptions(ko)}
-	if r.options.Features.Enabled(features.EnableAlphaRealtimeCompositions) {
+	if r.options.Features.Enabled(features.EnableBetaRealtimeCompositions) {
 		// If realtime composition is enabled we'll start watches dynamically,
 		// so we want to garbage collect watches for composed resource kinds
 		// that aren't used anymore.
@@ -516,7 +518,7 @@ func (r *Reconciler) Reconcile(ctx context.Context, req reconcile.Request) (reco
 	xr.SetGroupVersionKind(xrGVK)
 
 	crh := EnqueueForCompositionRevision(resource.CompositeKind(xrGVK), r.engine.GetCached(), log)
-	if err := r.engine.StartWatches(name,
+	if err := r.engine.StartWatches(ctx, name,
 		engine.WatchFor(xr, engine.WatchTypeCompositeResource, &handler.EnqueueRequestForObject{}),
 		engine.WatchFor(&v1.CompositionRevision{}, engine.WatchTypeCompositionRevision, crh),
 	); err != nil {
@@ -619,7 +621,7 @@ func (r *Reconciler) CompositeReconcilerOptions(ctx context.Context, d *v1.Compo
 
 	// If realtime compositions are enabled we pass the ControllerEngine to the
 	// XR reconciler so that it can start watches for composed resources.
-	if r.options.Features.Enabled(features.EnableAlphaRealtimeCompositions) {
+	if r.options.Features.Enabled(features.EnableBetaRealtimeCompositions) {
 		gvk := d.GetCompositeGroupVersionKind()
 		u := &kunstructured.Unstructured{}
 		u.SetAPIVersion(gvk.GroupVersion().String())
@@ -631,7 +633,10 @@ func (r *Reconciler) CompositeReconcilerOptions(ctx context.Context, d *v1.Compo
 		}
 
 		h := EnqueueCompositeResources(resource.CompositeKind(d.GetCompositeGroupVersionKind()), r.engine.GetCached(), r.log)
-		o = append(o, composite.WithWatchStarter(composite.ControllerName(d.GetName()), h, r.engine))
+		o = append(o,
+			composite.WithWatchStarter(composite.ControllerName(d.GetName()), h, r.engine),
+			composite.WithPollInterval(0), // Disable polling.
+		)
 	}
 
 	return o
diff --git a/internal/controller/apiextensions/offered/reconciler.go b/internal/controller/apiextensions/offered/reconciler.go
index 63840ffc5..5985b4f4b 100644
--- a/internal/controller/apiextensions/offered/reconciler.go
+++ b/internal/controller/apiextensions/offered/reconciler.go
@@ -94,7 +94,7 @@ type ControllerEngine interface {
 	Start(name string, o ...engine.ControllerOption) error
 	Stop(ctx context.Context, name string) error
 	IsRunning(name string) bool
-	StartWatches(name string, ws ...engine.Watch) error
+	StartWatches(ctx context.Context, name string, ws ...engine.Watch) error
 	GetCached() client.Client
 }
 
@@ -111,7 +111,7 @@ func (e *NopEngine) Stop(_ context.Context, _ string) error { return nil }
 func (e *NopEngine) IsRunning(_ string) bool { return true }
 
 // StartWatches does nothing.
-func (e *NopEngine) StartWatches(_ string, _ ...engine.Watch) error { return nil }
+func (e *NopEngine) StartWatches(_ context.Context, _ string, _ ...engine.Watch) error { return nil }
 
 // GetCached returns a nil client.
 func (e *NopEngine) GetCached() client.Client {
@@ -430,7 +430,6 @@ func (r *Reconciler) Reconcile(ctx context.Context, req reconcile.Request) (reco
 	o := []claim.ReconcilerOption{
 		claim.WithLogger(log.WithValues("controller", claim.ControllerName(d.GetName()))),
 		claim.WithRecorder(r.record.WithAnnotations("controller", claim.ControllerName(d.GetName()))),
-		claim.WithPollInterval(r.options.PollInterval),
 	}
 
 	// We only want to use the server-side XR syncer if the relevant feature
@@ -499,7 +498,7 @@ func (r *Reconciler) Reconcile(ctx context.Context, req reconcile.Request) (reco
 	xr := &kunstructured.Unstructured{}
 	xr.SetGroupVersionKind(d.GetCompositeGroupVersionKind())
 
-	if err := r.engine.StartWatches(claim.ControllerName(d.GetName()),
+	if err := r.engine.StartWatches(ctx, claim.ControllerName(d.GetName()),
 		engine.WatchFor(cm, engine.WatchTypeClaim, &handler.EnqueueRequestForObject{}),
 		engine.WatchFor(xr, engine.WatchTypeCompositeResource, &EnqueueRequestForClaim{}),
 	); err != nil {
diff --git a/internal/engine/engine.go b/internal/engine/engine.go
index aafe981c8..a1bdf2367 100644
--- a/internal/engine/engine.go
+++ b/internal/engine/engine.go
@@ -51,21 +51,22 @@ type ControllerEngine struct {
 	// managed by the engine should use these informers.
 	infs TrackingInformers
 
-	// The client used by the engine's controllers. The client must be backed by
-	// the above TrackingInformers.
+	// The cached client used by the engine's controllers. The client must
+	// be backed by the above TrackingInformers.
 	cached client.Client
 
-	// uncached is a non-cached client used when Unstructured resources
-	// are not found in the cache.
+	// The uncached client used by the engine's controllers. This client
+	// mustn't be cache backed. It's used to double-check operations that
+	// can't tolerate a stale cache, like checking whether a resource
+	// exists.
 	uncached client.Client
 
-	log logging.Logger
-
-	// Protects everything below.
-	mx sync.RWMutex
-
-	// Running controllers, by name.
+	// Running controllers, by name. Protected by the mutex.
 	controllers map[string]*controller
+	mx          sync.RWMutex
+
+	log     logging.Logger
+	metrics Metrics
 }
 
 // TrackingInformers is a set of Informers. It tracks which are active.
@@ -74,6 +75,21 @@ type TrackingInformers interface {
 	ActiveInformers() []schema.GroupVersionKind
 }
 
+// Metrics for the controller engine.
+type Metrics interface {
+	// ControllerStarted records a controller start.
+	ControllerStarted(name string)
+
+	// ControllerStopped records a controller stop.
+	ControllerStopped(name string)
+
+	// WatchStarted records a watch start for a controller.
+	WatchStarted(name string, t WatchType)
+
+	// WatchStopped records a watch stop for a controller.
+	WatchStopped(name string, t WatchType)
+}
+
 // New creates a new controller engine.
 func New(mgr manager.Manager, infs TrackingInformers, c client.Client, nc client.Client, o ...ControllerEngineOption) *ControllerEngine {
 	e := &ControllerEngine{
@@ -81,8 +97,9 @@ func New(mgr manager.Manager, infs TrackingInformers, c client.Client, nc client
 		infs:        infs,
 		cached:      c,
 		uncached:    nc,
-		log:         logging.NewNopLogger(),
 		controllers: make(map[string]*controller),
+		log:         logging.NewNopLogger(),
+		metrics:     &NopMetrics{},
 	}
 
 	for _, fn := range o {
@@ -92,7 +109,7 @@ func New(mgr manager.Manager, infs TrackingInformers, c client.Client, nc client
 	return e
 }
 
-// An ControllerEngineOption configures a controller engine.
+// A ControllerEngineOption configures a controller engine.
 type ControllerEngineOption func(*ControllerEngine)
 
 // WithLogger configures an Engine to use a logger.
@@ -102,6 +119,13 @@ func WithLogger(l logging.Logger) ControllerEngineOption {
 	}
 }
 
+// WithMetrics configures an Engine to expose metrics.
+func WithMetrics(m Metrics) ControllerEngineOption {
+	return func(e *ControllerEngine) {
+		e.metrics = m
+	}
+}
+
 type controller struct {
 	// The running controller.
 	ctrl kcontroller.Controller
@@ -211,6 +235,7 @@ func (e *ControllerEngine) Start(name string, o ...ControllerOption) error {
 		<-e.mgr.Elected()
 
 		e.log.Debug("Starting new controller", "controller", name)
+		e.metrics.ControllerStarted(name)
 
 		// Run the controller until its context is cancelled.
 		if err := c.Start(ctx); err != nil {
@@ -223,6 +248,7 @@ func (e *ControllerEngine) Start(name string, o ...ControllerOption) error {
 		}
 
 		e.log.Debug("Stopped controller", "controller", name)
+		e.metrics.ControllerStopped(name)
 	}()
 
 	if co.gc != nil {
@@ -270,6 +296,7 @@ func (e *ControllerEngine) Stop(ctx context.Context, name string) error {
 		if err := w.Stop(ctx); err != nil {
 			return errors.Wrapf(err, "cannot stop %q watch for %q", wid.Type, wid.GVK)
 		}
+
 		delete(c.sources, wid)
 		e.log.Debug("Stopped watching GVK", "controller", name, "watch-type", wid.Type, "watched-gvk", wid.GVK)
 	}
@@ -278,7 +305,6 @@ func (e *ControllerEngine) Stop(ctx context.Context, name string) error {
 	c.cancel()
 	delete(e.controllers, name)
 
-	e.log.Debug("Stopped controller", "controller", name)
 	return nil
 }
 
@@ -330,7 +356,7 @@ func WatchFor(kind client.Object, wt WatchType, h handler.EventHandler, p ...pre
 // The controller will only start a watch if it's not already watching the type
 // of object specified by the supplied Watch. StartWatches blocks other
 // operations on the same controller if and when it starts a watch.
-func (e *ControllerEngine) StartWatches(name string, ws ...Watch) error {
+func (e *ControllerEngine) StartWatches(ctx context.Context, name string, ws ...Watch) error {
 	e.mx.RLock()
 	c, running := e.controllers[name]
 	e.mx.RUnlock()
@@ -397,6 +423,14 @@ func (e *ControllerEngine) StartWatches(name string, ws ...Watch) error {
 	c.mx.Lock()
 	defer c.mx.Unlock()
 
+	// Refresh active informers in case they changed between when we lost
+	// the read lock and took the write lock.
+	a = e.infs.ActiveInformers()
+	activeInformer = make(map[schema.GroupVersionKind]bool, len(a))
+	for _, gvk := range a {
+		activeInformer[gvk] = true
+	}
+
 	// Start new sources.
 	for i, w := range ws {
 		wid := WatchID{Type: w.wt, GVK: gvks[i]}
@@ -417,7 +451,11 @@ func (e *ControllerEngine) StartWatches(name string, ws ...Watch) error {
 		// The watch will stop sending events when either the source is stopped,
 		// or its backing informer is stopped. The controller's work queue will
 		// stop processing events when the controller is stopped.
-		src := NewStoppableSource(e.infs, w.kind, w.handler, w.predicates...)
+		inf, err := e.infs.GetInformer(ctx, w.kind, cache.BlockUntilSynced(true))
+		if err != nil {
+			return errors.Wrapf(err, "cannot get informer for %q", wid.GVK)
+		}
+		src := NewStoppableSource(inf, w.handler, w.predicates...)
 		if err := c.ctrl.Watch(src); err != nil {
 			return errors.Wrapf(err, "cannot start %q watch for %q", wid.Type, wid.GVK)
 		}
@@ -426,6 +464,7 @@ func (e *ControllerEngine) StartWatches(name string, ws ...Watch) error {
 		c.sources[wid] = src
 
 		e.log.Debug("Started watching GVK", "controller", name, "watch-type", wid.Type, "watched-gvk", wid.GVK)
+		e.metrics.WatchStarted(name, wid.Type)
 	}
 
 	return nil
@@ -495,6 +534,7 @@ func (e *ControllerEngine) StopWatches(ctx context.Context, name string, ws ...W
 		}
 		delete(c.sources, wid)
 		e.log.Debug("Stopped watching GVK", "controller", name, "watch-type", wid.Type, "watched-gvk", wid.GVK)
+		e.metrics.WatchStopped(name, wid.Type)
 		stopped++
 	}
 
@@ -512,7 +552,7 @@ func (e *ControllerEngine) GarbageCollectCustomResourceInformers(ctx context.Con
 	}
 
 	h, err := i.AddEventHandler(kcache.ResourceEventHandlerFuncs{
-		DeleteFunc: func(obj interface{}) {
+		DeleteFunc: func(obj any) {
 			o := obj
 			if fsu, ok := obj.(kcache.DeletedFinalStateUnknown); ok {
 				o = fsu.Obj
@@ -533,6 +573,7 @@ func (e *ControllerEngine) GarbageCollectCustomResourceInformers(ctx context.Con
 				u := &unstructured.Unstructured{}
 				u.SetGroupVersionKind(gvk)
 
+				// This stops the informer if it was running.
 				if err := e.infs.RemoveInformer(ctx, u); err != nil {
 					e.log.Info("Cannot remove informer for type defined by deleted CustomResourceDefinition", "crd", crd.GetName(), "gvk", gvk)
 					continue
diff --git a/internal/engine/engine_metrics.go b/internal/engine/engine_metrics.go
new file mode 100644
index 000000000..75a68d87d
--- /dev/null
+++ b/internal/engine/engine_metrics.go
@@ -0,0 +1,113 @@
+/*
+Copyright 2025 The Crossplane Authors.
+
+Licensed under the Apache License, Version 2.0 (the "License"); you may not use
+this file except in compliance with the License. You may obtain a copy of the
+License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software distributed
+under the License is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR
+CONDITIONS OF ANY KIND, either express or implied. See the License for the
+specific language governing permissions and limitations under the License.
+*/
+
+package engine
+
+import (
+	"github.com/prometheus/client_golang/prometheus"
+)
+
+// NopMetrics does nothing.
+type NopMetrics struct{}
+
+// ControllerStarted does nothing.
+func (m *NopMetrics) ControllerStarted(_ string) {}
+
+// ControllerStopped does nothing.
+func (m *NopMetrics) ControllerStopped(_ string) {}
+
+// WatchStarted does nothing.
+func (m *NopMetrics) WatchStarted(_ string, _ WatchType) {}
+
+// WatchStopped does nothing.
+func (m *NopMetrics) WatchStopped(_ string, _ WatchType) {}
+
+// PrometheusMetrics for the controller engine.
+type PrometheusMetrics struct {
+	controllersStarted *prometheus.CounterVec
+	controllersStopped *prometheus.CounterVec
+
+	watchesStarted *prometheus.CounterVec
+	watchesStopped *prometheus.CounterVec
+}
+
+// NewPrometheusMetrics exposes controller engine metrics via Prometheus.
+func NewPrometheusMetrics() *PrometheusMetrics {
+	return &PrometheusMetrics{
+		controllersStarted: prometheus.NewCounterVec(prometheus.CounterOpts{
+			Subsystem: "composition",
+			Name:      "controllers_started_total",
+			Help:      "Total number of XR controllers started.",
+		}, []string{"controller"}),
+
+		controllersStopped: prometheus.NewCounterVec(prometheus.CounterOpts{
+			Subsystem: "composition",
+			Name:      "controllers_stopped_total",
+			Help:      "Total number of XR controllers stopped.",
+		}, []string{"controller"}),
+
+		watchesStarted: prometheus.NewCounterVec(prometheus.CounterOpts{
+			Subsystem: "composition",
+			Name:      "watches_started_total",
+			Help:      "Total number of watches started.",
+		}, []string{"controller", "type"}),
+
+		watchesStopped: prometheus.NewCounterVec(prometheus.CounterOpts{
+			Subsystem: "composition",
+			Name:      "watches_stopped_total",
+			Help:      "Total number of watches stopped.",
+		}, []string{"controller", "type"}),
+	}
+}
+
+// ControllerStarted records a controller start.
+func (m *PrometheusMetrics) ControllerStarted(name string) {
+	m.controllersStarted.With(prometheus.Labels{"controller": name}).Inc()
+}
+
+// ControllerStopped records a controller stop.
+func (m *PrometheusMetrics) ControllerStopped(name string) {
+	m.controllersStopped.With(prometheus.Labels{"controller": name}).Inc()
+}
+
+// WatchStarted records a watch start for a controller.
+func (m *PrometheusMetrics) WatchStarted(name string, t WatchType) {
+	m.watchesStarted.With(prometheus.Labels{"controller": name, "type": string(t)}).Inc()
+}
+
+// WatchStopped records a watch stop for a controller.
+func (m *PrometheusMetrics) WatchStopped(name string, t WatchType) {
+	m.watchesStopped.With(prometheus.Labels{"controller": name, "type": string(t)}).Inc()
+}
+
+// Describe sends the super-set of all possible descriptors of metrics
+// collected by this Collector to the provided channel and returns once
+// the last descriptor has been sent.
+func (m *PrometheusMetrics) Describe(ch chan<- *prometheus.Desc) {
+	m.controllersStarted.Describe(ch)
+	m.controllersStopped.Describe(ch)
+	m.watchesStarted.Describe(ch)
+	m.watchesStopped.Describe(ch)
+}
+
+// Collect is called by the Prometheus registry when collecting
+// metrics. The implementation sends each collected metric via the
+// provided channel and returns once the last metric has been sent.
+func (m *PrometheusMetrics) Collect(ch chan<- prometheus.Metric) {
+	m.controllersStarted.Collect(ch)
+	m.controllersStopped.Collect(ch)
+	m.watchesStarted.Collect(ch)
+	m.watchesStopped.Collect(ch)
+}
diff --git a/internal/engine/source.go b/internal/engine/source.go
index ea1c79a9c..8d295809a 100644
--- a/internal/engine/source.go
+++ b/internal/engine/source.go
@@ -35,15 +35,14 @@ import (
 var _ source.Source = &StoppableSource{}
 
 // NewStoppableSource returns a new watch source that can be stopped.
-func NewStoppableSource(infs cache.Informers, t client.Object, h handler.EventHandler, ps ...predicate.Predicate) *StoppableSource {
-	return &StoppableSource{infs: infs, Type: t, handler: h, predicates: ps}
+func NewStoppableSource(inf cache.Informer, h handler.EventHandler, ps ...predicate.Predicate) *StoppableSource {
+	return &StoppableSource{inf: inf, handler: h, predicates: ps}
 }
 
 // A StoppableSource is a controller-runtime watch source that can be stopped.
 type StoppableSource struct {
-	infs cache.Informers
+	inf cache.Informer
 
-	Type       client.Object
 	handler    handler.EventHandler
 	predicates []predicate.Predicate
 
@@ -53,12 +52,8 @@ type StoppableSource struct {
 // Start is internal and should be called only by the Controller to register
 // an EventHandler with the Informer to enqueue reconcile.Requests.
 func (s *StoppableSource) Start(ctx context.Context, q workqueue.TypedRateLimitingInterface[reconcile.Request]) error {
-	i, err := s.infs.GetInformer(ctx, s.Type, cache.BlockUntilSynced(true))
-	if err != nil {
-		return errors.Wrapf(err, "cannot get informer for %T", s.Type)
-	}
-
-	reg, err := i.AddEventHandler(NewEventHandler(ctx, q, s.handler, s.predicates...).HandlerFuncs())
+	// TODO(negz): Should we check if the informer is stopped first?
+	reg, err := s.inf.AddEventHandler(NewEventHandler(ctx, q, s.handler, s.predicates...).HandlerFuncs())
 	if err != nil {
 		return errors.Wrapf(err, "cannot add event handler")
 	}
@@ -69,17 +64,12 @@ func (s *StoppableSource) Start(ctx context.Context, q workqueue.TypedRateLimiti
 
 // Stop removes the EventHandler from the source's Informer. The Informer will
 // stop sending events to the source.
-func (s *StoppableSource) Stop(ctx context.Context) error {
-	if s.reg == nil {
+func (s *StoppableSource) Stop(_ context.Context) error {
+	if s.reg == nil || s.inf.IsStopped() {
 		return nil
 	}
 
-	i, err := s.infs.GetInformer(ctx, s.Type)
-	if err != nil {
-		return errors.Wrapf(err, "cannot get informer for %T", s.Type)
-	}
-
-	if err := i.RemoveEventHandler(s.reg); err != nil {
+	if err := s.inf.RemoveEventHandler(s.reg); err != nil {
 		return errors.Wrap(err, "cannot remove event handler")
 	}
 
@@ -119,7 +109,7 @@ func (e *EventHandler) HandlerFuncs() kcache.ResourceEventHandlerFuncs {
 }
 
 // OnAdd creates CreateEvent and calls Create on EventHandler.
-func (e *EventHandler) OnAdd(obj interface{}) {
+func (e *EventHandler) OnAdd(obj any) {
 	o, ok := obj.(client.Object)
 	if !ok {
 		return
@@ -138,7 +128,7 @@ func (e *EventHandler) OnAdd(obj interface{}) {
 }
 
 // OnUpdate creates UpdateEvent and calls Update on EventHandler.
-func (e *EventHandler) OnUpdate(oldObj, newObj interface{}) {
+func (e *EventHandler) OnUpdate(oldObj, newObj any) {
 	o, ok := oldObj.(client.Object)
 	if !ok {
 		return
@@ -163,7 +153,7 @@ func (e *EventHandler) OnUpdate(oldObj, newObj interface{}) {
 }
 
 // OnDelete creates DeleteEvent and calls Delete on EventHandler.
-func (e *EventHandler) OnDelete(obj interface{}) {
+func (e *EventHandler) OnDelete(obj any) {
 	var d event.DeleteEvent
 
 	switch o := obj.(type) {
diff --git a/internal/features/features.go b/internal/features/features.go
index 4e23447d8..e368d5e02 100644
--- a/internal/features/features.go
+++ b/internal/features/features.go
@@ -26,11 +26,6 @@ const (
 	// https://github.com/crossplane/crossplane/blob/390ddd/design/design-doc-external-secret-stores.md
 	EnableAlphaExternalSecretStores feature.Flag = "EnableAlphaExternalSecretStores"
 
-	// EnableAlphaRealtimeCompositions enables alpha support for realtime
-	// compositions, i.e. watching MRs and reconciling compositions immediately
-	// when any MR is updated.
-	EnableAlphaRealtimeCompositions feature.Flag = "EnableAlphaRealtimeCompositions"
-
 	// EnableAlphaDependencyVersionUpgrades enables alpha support for upgrading the version of a package's dependencies
 	// when needed.
 	EnableAlphaDependencyVersionUpgrades feature.Flag = "EnableAlphaDependencyVersionUpgrades"
@@ -61,4 +56,9 @@ const (
 	// the claim controller. See the below issue for more details:
 	// https://github.com/crossplane/crossplane/issues/4581
 	EnableBetaClaimSSA feature.Flag = "EnableBetaClaimSSA"
+
+	// EnableBetaRealtimeCompositions enables beta support for realtime
+	// compositions, i.e. watching composed resources and reconciling
+	// compositions immediately when any composed resource is updated.
+	EnableBetaRealtimeCompositions feature.Flag = "EnableBetaRealtimeCompositions"
 )
diff --git a/internal/xfn/function_runner_metrics.go b/internal/xfn/function_runner_metrics.go
index 8b6bf1adf..e148090ee 100644
--- a/internal/xfn/function_runner_metrics.go
+++ b/internal/xfn/function_runner_metrics.go
@@ -26,17 +26,17 @@ import (
 	fnv1 "github.com/crossplane/crossplane/apis/apiextensions/fn/proto/v1"
 )
 
-// Metrics are requests, errors, and duration (RED) metrics for composition
+// PrometheusMetrics are requests, errors, and duration (RED) metrics for composition
 // function runs.
-type Metrics struct {
+type PrometheusMetrics struct {
 	requests  *prometheus.CounterVec
 	responses *prometheus.CounterVec
 	duration  *prometheus.HistogramVec
 }
 
-// NewMetrics creates metrics for composition function runs.
-func NewMetrics() *Metrics {
-	return &Metrics{
+// NewPrometheusMetrics creates metrics for composition function runs.
+func NewPrometheusMetrics() *PrometheusMetrics {
+	return &PrometheusMetrics{
 		requests: prometheus.NewCounterVec(prometheus.CounterOpts{
 			Subsystem: "composition",
 			Name:      "run_function_request_total",
@@ -61,7 +61,7 @@ func NewMetrics() *Metrics {
 // Describe sends the super-set of all possible descriptors of metrics
 // collected by this Collector to the provided channel and returns once
 // the last descriptor has been sent.
-func (m *Metrics) Describe(ch chan<- *prometheus.Desc) {
+func (m *PrometheusMetrics) Describe(ch chan<- *prometheus.Desc) {
 	m.requests.Describe(ch)
 	m.responses.Describe(ch)
 	m.duration.Describe(ch)
@@ -70,7 +70,7 @@ func (m *Metrics) Describe(ch chan<- *prometheus.Desc) {
 // Collect is called by the Prometheus registry when collecting
 // metrics. The implementation sends each collected metric via the
 // provided channel and returns once the last metric has been sent.
-func (m *Metrics) Collect(ch chan<- prometheus.Metric) {
+func (m *PrometheusMetrics) Collect(ch chan<- prometheus.Metric) {
 	m.requests.Collect(ch)
 	m.responses.Collect(ch)
 	m.duration.Collect(ch)
@@ -78,7 +78,7 @@ func (m *Metrics) Collect(ch chan<- prometheus.Metric) {
 
 // CreateInterceptor returns a gRPC UnaryClientInterceptor for the named
 // function. The supplied package (pkg) should be the package's OCI reference.
-func (m *Metrics) CreateInterceptor(name, pkg string) grpc.UnaryClientInterceptor {
+func (m *PrometheusMetrics) CreateInterceptor(name, pkg string) grpc.UnaryClientInterceptor {
 	return func(ctx context.Context, method string, req, reply any, cc *grpc.ClientConn, invoker grpc.UnaryInvoker, opts ...grpc.CallOption) error {
 		l := prometheus.Labels{"function_name": name, "function_package": pkg, "grpc_target": cc.Target(), "grpc_method": method}
 
diff --git a/internal/xpkg/name.go b/internal/xpkg/name.go
index 17a42dc87..6ca02997e 100644
--- a/internal/xpkg/name.go
+++ b/internal/xpkg/name.go
@@ -144,3 +144,9 @@ func parseNameFromPackage(bs []byte) (string, error) {
 	err := yaml.Unmarshal(bs, p)
 	return p.Metadata.Name, err
 }
+
+// ReplaceExt replaces the file extension of the given path.
+func ReplaceExt(path, ext string) string {
+	old := filepath.Ext(path)
+	return path[0:len(path)-len(old)] + ext
+}



Test patch:
diff --git a/cmd/crank/xpkg/extract_test.go b/cmd/crank/xpkg/extract_test.go
new file mode 100644
index 000000000..6cbb21adb
--- /dev/null
+++ b/cmd/crank/xpkg/extract_test.go
@@ -0,0 +1,137 @@
+/*
+Copyright 2025 The Crossplane Authors.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package xpkg
+
+import (
+	"archive/tar"
+	"bytes"
+	"context"
+	"io"
+	"strings"
+	"testing"
+
+	"github.com/google/go-cmp/cmp"
+	"github.com/google/go-containerregistry/pkg/name"
+	v1 "github.com/google/go-containerregistry/pkg/v1"
+	"github.com/google/go-containerregistry/pkg/v1/empty"
+	"github.com/google/go-containerregistry/pkg/v1/mutate"
+	"github.com/google/go-containerregistry/pkg/v1/random"
+	"github.com/google/go-containerregistry/pkg/v1/tarball"
+	"github.com/google/go-containerregistry/pkg/v1/types"
+	"github.com/spf13/afero"
+
+	"github.com/crossplane/crossplane-runtime/pkg/errors"
+	"github.com/crossplane/crossplane-runtime/pkg/logging"
+	"github.com/crossplane/crossplane-runtime/pkg/test"
+
+	"github.com/crossplane/crossplane/internal/xpkg"
+)
+
+func TestExtractRun(t *testing.T) {
+	errBoom := errors.New("boom")
+	validTag := name.MustParseReference("crossplane/provider-aws:v0.24.1")
+	randLayer, _ := random.Layer(int64(1000), types.DockerLayer)
+	randImg, _ := mutate.Append(empty.Image, mutate.Addendum{
+		Layer: randLayer,
+		Annotations: map[string]string{
+			xpkg.AnnotationKey: xpkg.PackageAnnotation,
+		},
+	})
+
+	randImgDup, _ := mutate.Append(randImg, mutate.Addendum{
+		Layer: randLayer,
+		Annotations: map[string]string{
+			xpkg.AnnotationKey: xpkg.PackageAnnotation,
+		},
+	})
+
+	streamCont := "somestreamofyaml"
+	tarBuf := new(bytes.Buffer)
+	tw := tar.NewWriter(tarBuf)
+	hdr := &tar.Header{
+		Name: xpkg.StreamFile,
+		Mode: int64(xpkg.StreamFileMode),
+		Size: int64(len(streamCont)),
+	}
+	_ = tw.WriteHeader(hdr)
+	_, _ = io.Copy(tw, strings.NewReader(streamCont))
+	_ = tw.Close()
+
+	packLayer, _ := tarball.LayerFromOpener(func() (io.ReadCloser, error) {
+		// NOTE(hasheddan): we must construct a new reader each time as we
+		// ingest packImg in multiple tests below.
+		return io.NopCloser(bytes.NewReader(tarBuf.Bytes())), nil
+	})
+	packImg, _ := mutate.AppendLayers(empty.Image, packLayer)
+	cases := map[string]struct {
+		reason string
+		fs     afero.Fs
+		name   name.Reference
+		fetch  fetchFn
+		out    string
+		want   error
+	}{
+		"ErrorFetchPackage": {
+			reason: "Should return error if we fail to fetch package.",
+			name:   validTag,
+			fetch: func(_ context.Context, _ name.Reference) (v1.Image, error) {
+				return nil, errBoom
+			},
+			want: errors.Wrap(errBoom, errFetchPackage),
+		},
+		"ErrorMultipleAnnotatedLayers": {
+			reason: "Should return error if manifest contains multiple annotated layers.",
+			name:   validTag,
+			fetch: func(_ context.Context, _ name.Reference) (v1.Image, error) {
+				return randImgDup, nil
+			},
+			want: errors.New(errMultipleAnnotatedLayers),
+		},
+		"ErrorFetchBadPackage": {
+			reason: "Should return error if image with contents does not have package.yaml.",
+			name:   validTag,
+			fetch: func(_ context.Context, _ name.Reference) (v1.Image, error) {
+				return randImg, nil
+			},
+			want: errors.Wrap(io.EOF, errOpenPackageStream),
+		},
+		"Success": {
+			reason: "Should not return error if we successfully fetch package and extract contents.",
+			name:   validTag,
+			fetch: func(_ context.Context, _ name.Reference) (v1.Image, error) {
+				return packImg, nil
+			},
+			fs:  afero.NewMemMapFs(),
+			out: "out.gz",
+		},
+	}
+	for name, tc := range cases {
+		t.Run(name, func(t *testing.T) {
+			logger := logging.NewNopLogger()
+
+			err := (&extractCmd{
+				fs:     tc.fs,
+				fetch:  tc.fetch,
+				name:   tc.name,
+				Output: tc.out,
+			}).Run(logger)
+			if diff := cmp.Diff(tc.want, err, test.EquateErrors()); diff != "" {
+				t.Errorf("\n%s\nRun(...): -want error, +got error:\n%s", tc.reason, diff)
+			}
+		})
+	}
+}
diff --git a/internal/controller/apiextensions/composite/api_test.go b/internal/controller/apiextensions/composite/api_test.go
index 4f4cb75ae..73403654d 100644
--- a/internal/controller/apiextensions/composite/api_test.go
+++ b/internal/controller/apiextensions/composite/api_test.go
@@ -213,15 +213,15 @@ func TestFetchRevision(t *testing.T) {
 
 	cases := map[string]struct {
 		reason string
-		client resource.ClientApplicator
+		client client.Client
 		args   args
 		want   want
 	}{
 		"GetCompositionRevisionError": {
 			reason: "We should wrap and return errors encountered getting the CompositionRevision.",
-			client: resource.ClientApplicator{Client: &test.MockClient{
+			client: &test.MockClient{
 				MockGet: test.NewMockGetFn(errBoom),
-			}},
+			},
 			args: args{
 				cr: &fake.Composite{
 					CompositionRevisionReferencer: fake.CompositionRevisionReferencer{Ref: &corev1.LocalObjectReference{}},
@@ -235,12 +235,12 @@ func TestFetchRevision(t *testing.T) {
 		},
 		"UpdateManual": {
 			reason: "When we're using the manual update policy and a revision reference is set we should return that revision as a composition.",
-			client: resource.ClientApplicator{Client: &test.MockClient{
+			client: &test.MockClient{
 				MockGet: test.NewMockGetFn(nil, func(obj client.Object) error {
 					*obj.(*v1.CompositionRevision) = *rev3
 					return nil
 				}),
-			}},
+			},
 			args: args{
 				cr: &fake.Composite{
 					CompositionRevisionReferencer: fake.CompositionRevisionReferencer{Ref: &corev1.LocalObjectReference{}},
@@ -253,9 +253,9 @@ func TestFetchRevision(t *testing.T) {
 		},
 		"GetCompositionError": {
 			reason: "We should wrap and return errors encountered getting the Composition.",
-			client: resource.ClientApplicator{Client: &test.MockClient{
+			client: &test.MockClient{
 				MockGet: test.NewMockGetFn(errBoom),
-			}},
+			},
 			args: args{
 				cr: &fake.Composite{
 					CompositionReferencer: fake.CompositionReferencer{Ref: &corev1.ObjectReference{}},
@@ -267,10 +267,10 @@ func TestFetchRevision(t *testing.T) {
 		},
 		"ListCompositionRevisionsError": {
 			reason: "We should wrap and return errors encountered listing CompositionRevisions.",
-			client: resource.ClientApplicator{Client: &test.MockClient{
+			client: &test.MockClient{
 				MockGet:  test.NewMockGetFn(nil),
 				MockList: test.NewMockListFn(errBoom),
-			}},
+			},
 			args: args{
 				cr: &fake.Composite{
 					CompositionReferencer: fake.CompositionReferencer{Ref: &corev1.ObjectReference{}},
@@ -282,10 +282,10 @@ func TestFetchRevision(t *testing.T) {
 		},
 		"NoCompositionRevisionsError": {
 			reason: "We should return an error if we don't find any suitable CompositionRevisions.",
-			client: resource.ClientApplicator{Client: &test.MockClient{
+			client: &test.MockClient{
 				MockGet:  test.NewMockGetFn(nil),
 				MockList: test.NewMockListFn(nil),
-			}},
+			},
 			args: args{
 				cr: &fake.Composite{
 					CompositionReferencer: fake.CompositionReferencer{Ref: &corev1.ObjectReference{}},
@@ -297,33 +297,31 @@ func TestFetchRevision(t *testing.T) {
 		},
 		"AlreadyAtLatestRevision": {
 			reason: "We should return the latest revision without updating our reference if we already reference it.",
-			client: resource.ClientApplicator{
-				Client: &test.MockClient{
-					MockGet: test.NewMockGetFn(nil, func(obj client.Object) error {
-						*obj.(*v1.Composition) = *comp
-						return nil
-					}),
-					MockList: test.NewMockListFn(nil, func(obj client.ObjectList) error {
-						*obj.(*v1.CompositionRevisionList) = v1.CompositionRevisionList{
-							Items: []v1.CompositionRevision{
-								// We should ignore this revision because it does not have
-								// our composition above as its controller reference.
-								*rev3,
-
-								// This revision is owned by our composition, and is the
-								// latest revision.
-								*rev2,
-
-								// This revision is owned by our composition, but is not the
-								// latest revision.
-								*rev1,
-							},
-						}
-						return nil
-					}),
-				},
+			client: &test.MockClient{
+				MockGet: test.NewMockGetFn(nil, func(obj client.Object) error {
+					*obj.(*v1.Composition) = *comp
+					return nil
+				}),
+				MockList: test.NewMockListFn(nil, func(obj client.ObjectList) error {
+					*obj.(*v1.CompositionRevisionList) = v1.CompositionRevisionList{
+						Items: []v1.CompositionRevision{
+							// We should ignore this revision because it does not have
+							// our composition above as its controller reference.
+							*rev3,
+
+							// This revision is owned by our composition, and is the
+							// latest revision.
+							*rev2,
+
+							// This revision is owned by our composition, but is not the
+							// latest revision.
+							*rev1,
+						},
+					}
+					return nil
+				}),
 				// This should not be called.
-				Applicator: resource.ApplyFn(func(_ context.Context, _ client.Object, _ ...resource.ApplyOption) error { return errBoom }),
+				MockUpdate: test.NewMockUpdateFn(errBoom),
 			},
 			args: args{
 				cr: &fake.Composite{
@@ -342,25 +340,22 @@ func TestFetchRevision(t *testing.T) {
 		},
 		"NoRevisionSet": {
 			reason: "We should return the latest revision and update our reference if none is set.",
-			client: resource.ClientApplicator{
-				Client: &test.MockClient{
-					MockGet: test.NewMockGetFn(nil, func(obj client.Object) error {
-						*obj.(*v1.Composition) = *comp
-						return nil
-					}),
-					MockList: test.NewMockListFn(nil, func(obj client.ObjectList) error {
-						*obj.(*v1.CompositionRevisionList) = v1.CompositionRevisionList{
-							Items: []v1.CompositionRevision{
-								// This revision is owned by our composition, and is the
-								// latest revision.
-								*rev2,
-							},
-						}
-						return nil
-					}),
-				},
-				Applicator: resource.ApplyFn(func(_ context.Context, o client.Object, _ ...resource.ApplyOption) error {
-					// Ensure we were updated to reference the latest CompositionRevision.
+			client: &test.MockClient{
+				MockGet: test.NewMockGetFn(nil, func(obj client.Object) error {
+					*obj.(*v1.Composition) = *comp
+					return nil
+				}),
+				MockList: test.NewMockListFn(nil, func(obj client.ObjectList) error {
+					*obj.(*v1.CompositionRevisionList) = v1.CompositionRevisionList{
+						Items: []v1.CompositionRevision{
+							// This revision is owned by our composition, and is the
+							// latest revision.
+							*rev2,
+						},
+					}
+					return nil
+				}),
+				MockUpdate: test.NewMockUpdateFn(nil, func(obj client.Object) error {
 					want := &fake.Composite{
 						CompositionReferencer: fake.CompositionReferencer{
 							Ref: &corev1.ObjectReference{Name: comp.GetName()},
@@ -372,7 +367,7 @@ func TestFetchRevision(t *testing.T) {
 						},
 						CompositionUpdater: fake.CompositionUpdater{Policy: &manual},
 					}
-					if diff := cmp.Diff(want, o); diff != "" {
+					if diff := cmp.Diff(want, obj); diff != "" {
 						t.Errorf("Apply(): -want, +got: %s", diff)
 					}
 					return nil
@@ -394,27 +389,25 @@ func TestFetchRevision(t *testing.T) {
 		},
 		"OutdatedRevisionSet": {
 			reason: "We should return the latest revision and update our reference if an outdated revision is referenced.",
-			client: resource.ClientApplicator{
-				Client: &test.MockClient{
-					MockGet: test.NewMockGetFn(nil, func(obj client.Object) error {
-						*obj.(*v1.Composition) = *comp
-						return nil
-					}),
-					MockList: test.NewMockListFn(nil, func(obj client.ObjectList) error {
-						*obj.(*v1.CompositionRevisionList) = v1.CompositionRevisionList{
-							Items: []v1.CompositionRevision{
-								// This revision is owned by our composition, and is the
-								// latest revision.
-								*rev2,
-
-								// This is an outdated revision.
-								*rev1,
-							},
-						}
-						return nil
-					}),
-				},
-				Applicator: resource.ApplyFn(func(_ context.Context, o client.Object, _ ...resource.ApplyOption) error {
+			client: &test.MockClient{
+				MockGet: test.NewMockGetFn(nil, func(obj client.Object) error {
+					*obj.(*v1.Composition) = *comp
+					return nil
+				}),
+				MockList: test.NewMockListFn(nil, func(obj client.ObjectList) error {
+					*obj.(*v1.CompositionRevisionList) = v1.CompositionRevisionList{
+						Items: []v1.CompositionRevision{
+							// This revision is owned by our composition, and is the
+							// latest revision.
+							*rev2,
+
+							// This is an outdated revision.
+							*rev1,
+						},
+					}
+					return nil
+				}),
+				MockUpdate: test.NewMockUpdateFn(nil, func(obj client.Object) error {
 					// Ensure we were updated to reference the latest CompositionRevision.
 					want := &fake.Composite{
 						CompositionReferencer: fake.CompositionReferencer{
@@ -426,7 +419,7 @@ func TestFetchRevision(t *testing.T) {
 							},
 						},
 					}
-					if diff := cmp.Diff(want, o); diff != "" {
+					if diff := cmp.Diff(want, obj); diff != "" {
 						t.Errorf("Apply(): -want, +got: %s", diff)
 					}
 					return nil
@@ -451,26 +444,22 @@ func TestFetchRevision(t *testing.T) {
 		},
 		"SetRevisionError": {
 			reason: "We should return the latest revision and update our reference if none is set.",
-			client: resource.ClientApplicator{
-				Client: &test.MockClient{
-					MockGet: test.NewMockGetFn(nil, func(obj client.Object) error {
-						*obj.(*v1.Composition) = *comp
-						return nil
-					}),
-					MockList: test.NewMockListFn(nil, func(obj client.ObjectList) error {
-						*obj.(*v1.CompositionRevisionList) = v1.CompositionRevisionList{
-							Items: []v1.CompositionRevision{
-								// This revision is owned by our composition, and is the
-								// latest revision.
-								*rev2,
-							},
-						}
-						return nil
-					}),
-				},
-				Applicator: resource.ApplyFn(func(_ context.Context, _ client.Object, _ ...resource.ApplyOption) error {
-					return errBoom
+			client: &test.MockClient{
+				MockGet: test.NewMockGetFn(nil, func(obj client.Object) error {
+					*obj.(*v1.Composition) = *comp
+					return nil
+				}),
+				MockList: test.NewMockListFn(nil, func(obj client.ObjectList) error {
+					*obj.(*v1.CompositionRevisionList) = v1.CompositionRevisionList{
+						Items: []v1.CompositionRevision{
+							// This revision is owned by our composition, and is the
+							// latest revision.
+							*rev2,
+						},
+					}
+					return nil
 				}),
+				MockUpdate: test.NewMockUpdateFn(errBoom),
 			},
 			args: args{
 				cr: &fake.Composite{
diff --git a/internal/controller/apiextensions/composite/composition_functions_test.go b/internal/controller/apiextensions/composite/composition_functions_test.go
index 3a9c7462c..0f6943119 100644
--- a/internal/controller/apiextensions/composite/composition_functions_test.go
+++ b/internal/controller/apiextensions/composite/composition_functions_test.go
@@ -19,11 +19,13 @@ package composite
 import (
 	"context"
 	"testing"
+	"time"
 
 	"github.com/google/go-cmp/cmp"
 	"github.com/google/go-cmp/cmp/cmpopts"
 	"google.golang.org/protobuf/encoding/protojson"
 	"google.golang.org/protobuf/testing/protocmp"
+	"google.golang.org/protobuf/types/known/durationpb"
 	"google.golang.org/protobuf/types/known/structpb"
 	corev1 "k8s.io/api/core/v1"
 	kerrors "k8s.io/apimachinery/pkg/api/errors"
@@ -698,6 +700,9 @@ func TestFunctionCompose(t *testing.T) {
 				},
 				r: FunctionRunnerFn(func(_ context.Context, _ string, _ *fnv1.RunFunctionRequest) (*fnv1.RunFunctionResponse, error) {
 					rsp := &fnv1.RunFunctionResponse{
+						Meta: &fnv1.ResponseMeta{
+							Ttl: durationpb.New(5 * time.Minute),
+						},
 						Desired: &fnv1.State{
 							Composite: &fnv1.Resource{
 								Resource: MustStruct(map[string]any{
@@ -890,6 +895,7 @@ func TestFunctionCompose(t *testing.T) {
 							Target: CompositionTargetCompositeAndClaim,
 						},
 					},
+					TTL: 5 * time.Minute,
 				},
 				err: nil,
 			},
diff --git a/internal/controller/apiextensions/composite/reconciler_test.go b/internal/controller/apiextensions/composite/reconciler_test.go
index 65309d7d4..88195adc0 100644
--- a/internal/controller/apiextensions/composite/reconciler_test.go
+++ b/internal/controller/apiextensions/composite/reconciler_test.go
@@ -29,6 +29,7 @@ import (
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/runtime"
 	"k8s.io/apimachinery/pkg/runtime/schema"
+	"k8s.io/utils/ptr"
 	"sigs.k8s.io/controller-runtime/pkg/client"
 	"sigs.k8s.io/controller-runtime/pkg/reconcile"
 
@@ -452,7 +453,7 @@ func TestReconcile(t *testing.T) {
 				},
 			},
 			want: want{
-				r: reconcile.Result{RequeueAfter: defaultPollInterval},
+				r: reconcile.Result{},
 			},
 		},
 		"ComposedResourcesNotReady": {
@@ -575,7 +576,7 @@ func TestReconcile(t *testing.T) {
 							return true, nil
 						},
 					}),
-					WithWatchStarter("cool-controller", nil, WatchStarterFn(func(_ string, ws ...engine.Watch) error {
+					WithWatchStarter("cool-controller", nil, WatchStarterFn(func(_ context.Context, _ string, ws ...engine.Watch) error {
 						cd := composed.New(composed.FromReference(corev1.ObjectReference{
 							APIVersion: "example.org/v1",
 							Kind:       "ComposedResource",
@@ -591,7 +592,7 @@ func TestReconcile(t *testing.T) {
 				},
 			},
 			want: want{
-				r: reconcile.Result{RequeueAfter: defaultPollInterval},
+				r: reconcile.Result{},
 			},
 		},
 		"ReconciliationPausedSuccessful": {
@@ -626,7 +627,7 @@ func TestReconcile(t *testing.T) {
 			},
 		},
 		"ReconciliationResumes": {
-			reason: `If a composite resource has the pause annotation with some value other than "true" and the Synced=False/ReconcilePaused status condition, reconciliation should resume with requeueing.`,
+			reason: `If a composite resource has the pause annotation with some value other than "true" and the Synced=False/ReconcilePaused status condition, reconciliation should resume with requeuing.`,
 			args: args{
 				c: &test.MockClient{
 					MockGet: WithComposite(t, NewComposite(func(cr resource.Composite) {
@@ -670,11 +671,11 @@ func TestReconcile(t *testing.T) {
 				},
 			},
 			want: want{
-				r: reconcile.Result{RequeueAfter: defaultPollInterval},
+				r: reconcile.Result{},
 			},
 		},
 		"ReconciliationResumesAfterAnnotationRemoval": {
-			reason: `If a composite resource has the pause annotation removed and the Synced=False/ReconcilePaused status condition, reconciliation should resume with requeueing.`,
+			reason: `If a composite resource has the pause annotation removed and the Synced=False/ReconcilePaused status condition, reconciliation should resume with requeuing.`,
 			args: args{
 				c: &test.MockClient{
 					MockGet: WithComposite(t, NewComposite(func(cr resource.Composite) {
@@ -718,7 +719,7 @@ func TestReconcile(t *testing.T) {
 				},
 			},
 			want: want{
-				r: reconcile.Result{RequeueAfter: defaultPollInterval},
+				r: reconcile.Result{},
 			},
 		},
 		"CustomEventsAndConditions": {
@@ -870,7 +871,7 @@ func TestReconcile(t *testing.T) {
 				},
 			},
 			want: want{
-				r: reconcile.Result{RequeueAfter: defaultPollInterval},
+				r: reconcile.Result{},
 			},
 		},
 		"CustomEventsAndConditionFatal": {
@@ -1203,7 +1204,7 @@ func TestReconcile(t *testing.T) {
 				},
 			},
 			want: want{
-				r: reconcile.Result{RequeueAfter: defaultPollInterval},
+				r: reconcile.Result{},
 			},
 		},
 		"SystemConditionUpdate": {
@@ -1224,10 +1225,7 @@ func TestReconcile(t *testing.T) {
 					}),
 					MockStatusUpdate: WantComposite(t, NewComposite(func(cr resource.Composite) {
 						cr.SetCompositionReference(&corev1.ObjectReference{})
-						cr.SetConditions(
-							xpv1.ReconcileSuccess(),
-							xpv1.Creating().WithMessage("Composite resource was explicitly marked as unready by the composer"),
-						)
+						cr.SetConditions(xpv1.ReconcileSuccess(), xpv1.Creating())
 						cr.SetClaimReference(&reference.Claim{})
 					})),
 				},
@@ -1260,11 +1258,9 @@ func TestReconcile(t *testing.T) {
 					})),
 					WithComposer(ComposerFn(func(_ context.Context, _ *composite.Unstructured, _ CompositionRequest) (CompositionResult, error) {
 						return CompositionResult{
-							Composite: CompositeResource{
-								Ready: &valBoolFalse,
-							},
 							Composed:          []ComposedResource{},
 							ConnectionDetails: cd,
+							Ready:             ptr.To(false),
 							Events:            []TargetedEvent{},
 							Conditions:        []TargetedCondition{},
 						}, nil
@@ -1272,7 +1268,7 @@ func TestReconcile(t *testing.T) {
 				},
 			},
 			want: want{
-				r: reconcile.Result{RequeueAfter: defaultPollInterval},
+				r: reconcile.Result{},
 			},
 		},
 		"CustomEventsFailToGetClaim": {
@@ -1354,7 +1350,7 @@ func TestReconcile(t *testing.T) {
 				},
 			},
 			want: want{
-				r: reconcile.Result{RequeueAfter: defaultPollInterval},
+				r: reconcile.Result{},
 			},
 		},
 	}
diff --git a/internal/controller/apiextensions/composite/watch/watch_test.go b/internal/controller/apiextensions/composite/watch/watch_test.go
index d25542af9..5c1960d5c 100644
--- a/internal/controller/apiextensions/composite/watch/watch_test.go
+++ b/internal/controller/apiextensions/composite/watch/watch_test.go
@@ -83,13 +83,16 @@ func TestGarbageCollectWatchesNow(t *testing.T) {
 		args   args
 		want   want
 	}{
-		"ListXRsError": {
-			reason: "The method should return an error if it can't list XRs.",
+		"GetWatchesError": {
+			reason: "The method should return an error if it can't get watches.",
 			params: params{
 				ce: &MockEngine{
+					MockGetWatches: func(_ string) ([]engine.WatchID, error) {
+						return nil, errBoom
+					},
 					MockGetCached: func() client.Client {
 						return &test.MockClient{
-							MockList: test.NewMockListFn(errBoom),
+							MockList: test.NewMockListFn(nil),
 						}
 					},
 				},
@@ -98,18 +101,37 @@ func TestGarbageCollectWatchesNow(t *testing.T) {
 				err: cmpopts.AnyError,
 			},
 		},
-		"GetWatchesError": {
-			reason: "The method should return an error if it can't get watches.",
+		"NoComposedResourceWatches": {
+			reason: "The method should return early if there's no composed resource watches to potentially GC.",
+			params: params{
+				ce: &MockEngine{
+					MockGetWatches: func(_ string) ([]engine.WatchID, error) {
+						w := []engine.WatchID{{
+							Type: engine.WatchTypeCompositeResource,
+						}}
+						return w, nil
+					},
+				},
+			},
+			want: want{
+				err: nil,
+			},
+		},
+		"ListXRsError": {
+			reason: "The method should return an error if it can't list XRs.",
 			params: params{
 				ce: &MockEngine{
+					MockGetWatches: func(_ string) ([]engine.WatchID, error) {
+						w := []engine.WatchID{{
+							Type: engine.WatchTypeComposedResource,
+						}}
+						return w, nil
+					},
 					MockGetCached: func() client.Client {
 						return &test.MockClient{
-							MockList: test.NewMockListFn(nil),
+							MockList: test.NewMockListFn(errBoom),
 						}
 					},
-					MockGetWatches: func(_ string) ([]engine.WatchID, error) {
-						return nil, errBoom
-					},
 				},
 			},
 			want: want{
@@ -120,11 +142,6 @@ func TestGarbageCollectWatchesNow(t *testing.T) {
 			reason: "The method should return an error if it can't stop watches.",
 			params: params{
 				ce: &MockEngine{
-					MockGetCached: func() client.Client {
-						return &test.MockClient{
-							MockList: test.NewMockListFn(nil),
-						}
-					},
 					MockGetWatches: func(_ string) ([]engine.WatchID, error) {
 						w := []engine.WatchID{
 							{
@@ -134,6 +151,16 @@ func TestGarbageCollectWatchesNow(t *testing.T) {
 						}
 						return w, nil
 					},
+					MockGetCached: func() client.Client {
+						return &test.MockClient{
+							MockList: test.NewMockListFn(nil),
+						}
+					},
+					MockGetUncached: func() client.Client {
+						return &test.MockClient{
+							MockList: test.NewMockListFn(nil),
+						}
+					},
 					MockStopWatches: func(_ context.Context, _ string, _ ...engine.WatchID) (int, error) {
 						return 0, errBoom
 					},
@@ -144,7 +171,7 @@ func TestGarbageCollectWatchesNow(t *testing.T) {
 			},
 		},
 		"NothingToStop": {
-			reason: "StopWatches shouldn't be called if there's no watches to stop.",
+			reason: "The method shouldn't list from the uncached client if the cached client indicates there's no watches to stop.",
 			params: params{
 				ce: &MockEngine{
 					MockGetCached: func() client.Client {
@@ -152,17 +179,32 @@ func TestGarbageCollectWatchesNow(t *testing.T) {
 							MockList: test.NewMockListFn(nil),
 						}
 					},
+					// A list from uncached would panic,
+					// since it's not mocked.
 					MockGetWatches: func(_ string) ([]engine.WatchID, error) {
-						return nil, nil
+						w := []engine.WatchID{
+							{
+								Type: engine.WatchTypeCompositeResource,
+								GVK:  schema.GroupVersionKind{},
+							},
+							{
+								Type: engine.WatchTypeClaim,
+								GVK:  schema.GroupVersionKind{},
+							},
+							{
+								Type: engine.WatchTypeCompositionRevision,
+								GVK:  schema.GroupVersionKind{},
+							},
+						}
+						return w, nil
 					},
-					// StopWatches would panic if called, since it's not mocked.
 				},
 			},
 			want: want{
 				err: nil,
 			},
 		},
-		"UneededWatchesStopped": {
+		"UnneededWatchesStopped": {
 			reason: "StopWatches shouldn't be called if there's no watches to stop.",
 			params: params{
 				ce: &MockEngine{
@@ -184,6 +226,26 @@ func TestGarbageCollectWatchesNow(t *testing.T) {
 							}),
 						}
 					},
+					// Uncached result matches cached
+					// result.
+					MockGetUncached: func() client.Client {
+						return &test.MockClient{
+							MockList: test.NewMockListFn(nil, func(obj client.ObjectList) error {
+								xr := composite.New()
+								xr.SetResourceReferences([]corev1.ObjectReference{
+									{
+										APIVersion: "example.org/v1",
+										Kind:       "StillComposed",
+										// Name doesn't matter.
+									},
+								})
+
+								obj.(*unstructured.UnstructuredList).Items = []unstructured.Unstructured{xr.Unstructured}
+
+								return nil
+							}),
+						}
+					},
 					MockGetWatches: func(_ string) ([]engine.WatchID, error) {
 						w := []engine.WatchID{
 							// We want to keep this one.
diff --git a/internal/controller/apiextensions/definition/handlers_test.go b/internal/controller/apiextensions/definition/handlers_test.go
index 2d6f79a2d..67c242a5c 100644
--- a/internal/controller/apiextensions/definition/handlers_test.go
+++ b/internal/controller/apiextensions/definition/handlers_test.go
@@ -31,7 +31,7 @@ func TestEnqueueForCompositionRevisionFunc(t *testing.T) {
 		event  kevent.CreateEvent
 	}
 	type want struct {
-		added []interface{}
+		added []any
 	}
 
 	dog := schema.GroupVersionKind{Group: "example.com", Version: "v1", Kind: "Dog"}
@@ -89,11 +89,14 @@ func TestEnqueueForCompositionRevisionFunc(t *testing.T) {
 								v1.LabelCompositionName: "dachshund",
 							},
 						},
+						Spec: v1.CompositionRevisionSpec{
+							CompositeTypeRef: v1.TypeReferenceTo(dog),
+						},
 					},
 				},
 			},
 			want: want{
-				added: []interface{}{reconcile.Request{NamespacedName: types.NamespacedName{
+				added: []any{reconcile.Request{NamespacedName: types.NamespacedName{
 					Namespace: "ns",
 					Name:      "obj1",
 				}}},
@@ -125,6 +128,9 @@ func TestEnqueueForCompositionRevisionFunc(t *testing.T) {
 								v1.LabelCompositionName: "dachshund",
 							},
 						},
+						Spec: v1.CompositionRevisionSpec{
+							CompositeTypeRef: v1.TypeReferenceTo(dog),
+						},
 					},
 				},
 			},
@@ -156,6 +162,9 @@ func TestEnqueueForCompositionRevisionFunc(t *testing.T) {
 								v1.LabelCompositionName: "dachshund",
 							},
 						},
+						Spec: v1.CompositionRevisionSpec{
+							CompositeTypeRef: v1.TypeReferenceTo(dog),
+						},
 					},
 				},
 			},
@@ -203,11 +212,14 @@ func TestEnqueueForCompositionRevisionFunc(t *testing.T) {
 								v1.LabelCompositionName: "dachshund",
 							},
 						},
+						Spec: v1.CompositionRevisionSpec{
+							CompositeTypeRef: v1.TypeReferenceTo(dog),
+						},
 					},
 				},
 			},
 			want: want{
-				added: []interface{}{
+				added: []any{
 					reconcile.Request{NamespacedName: types.NamespacedName{Namespace: "ns", Name: "obj1"}},
 					reconcile.Request{NamespacedName: types.NamespacedName{Namespace: "ns", Name: "obj2"}},
 				},
@@ -229,7 +241,7 @@ func TestEnqueueForCompositionRevisionFunc(t *testing.T) {
 
 type rateLimitingQueueMock struct {
 	workqueue.TypedRateLimitingInterface[reconcile.Request]
-	added []interface{}
+	added []any
 }
 
 func (f *rateLimitingQueueMock) Add(item reconcile.Request) {
diff --git a/internal/controller/apiextensions/definition/reconciler_test.go b/internal/controller/apiextensions/definition/reconciler_test.go
index e9c9a5035..bfb70af9d 100644
--- a/internal/controller/apiextensions/definition/reconciler_test.go
+++ b/internal/controller/apiextensions/definition/reconciler_test.go
@@ -48,7 +48,7 @@ type MockEngine struct {
 	MockStop            func(ctx context.Context, name string) error
 	MockIsRunning       func(name string) bool
 	MockGetWatches      func(name string) ([]engine.WatchID, error)
-	MockStartWatches    func(name string, ws ...engine.Watch) error
+	MockStartWatches    func(ctx context.Context, name string, ws ...engine.Watch) error
 	MockStopWatches     func(ctx context.Context, name string, ws ...engine.WatchID) (int, error)
 	MockGetCached       func() client.Client
 	MockGetUncached     func() client.Client
@@ -71,8 +71,8 @@ func (m *MockEngine) GetWatches(name string) ([]engine.WatchID, error) {
 	return m.MockGetWatches(name)
 }
 
-func (m *MockEngine) StartWatches(name string, ws ...engine.Watch) error {
-	return m.MockStartWatches(name, ws...)
+func (m *MockEngine) StartWatches(ctx context.Context, name string, ws ...engine.Watch) error {
+	return m.MockStartWatches(ctx, name, ws...)
 }
 
 func (m *MockEngine) StopWatches(ctx context.Context, name string, ws ...engine.WatchID) (int, error) {
@@ -599,7 +599,7 @@ func TestReconcile(t *testing.T) {
 			},
 		},
 		"VersionChangedStopControllerError": {
-			reason: "We should return any error we encounter while stopping our controller because the XRD's referencable version changed.",
+			reason: "We should return any error we encounter while stopping our controller because the XRD's referenceable version changed.",
 			args: args{
 				ca: resource.ClientApplicator{
 					Client: &test.MockClient{
@@ -732,7 +732,7 @@ func TestReconcile(t *testing.T) {
 						MockStart: func(_ string, _ ...engine.ControllerOption) error {
 							return nil
 						},
-						MockStartWatches: func(_ string, _ ...engine.Watch) error {
+						MockStartWatches: func(_ context.Context, _ string, _ ...engine.Watch) error {
 							return errBoom
 						},
 						MockGetCached:   func() client.Client { return test.NewMockClient() },
@@ -746,7 +746,7 @@ func TestReconcile(t *testing.T) {
 			},
 		},
 		"SuccessfulStart": {
-			reason: "We should return without requeueing if we successfully ensured our CRD exists and controller is started.",
+			reason: "We should return without requeuing if we successfully ensured our CRD exists and controller is started.",
 			args: args{
 				ca: resource.ClientApplicator{
 					Client: &test.MockClient{
@@ -781,7 +781,7 @@ func TestReconcile(t *testing.T) {
 					WithControllerEngine(&MockEngine{
 						MockIsRunning:    func(_ string) bool { return false },
 						MockStart:        func(_ string, _ ...engine.ControllerOption) error { return nil },
-						MockStartWatches: func(_ string, _ ...engine.Watch) error { return nil },
+						MockStartWatches: func(_ context.Context, _ string, _ ...engine.Watch) error { return nil },
 						MockGetCached:    func() client.Client { return test.NewMockClient() },
 						MockGetUncached:  func() client.Client { return test.NewMockClient() },
 					}),
@@ -792,7 +792,7 @@ func TestReconcile(t *testing.T) {
 			},
 		},
 		"SuccessfulUpdateControllerVersion": {
-			reason: "We should return without requeueing if we successfully ensured our CRD exists, the old controller stopped, and the new one started.",
+			reason: "We should return without requeuing if we successfully ensured our CRD exists, the old controller stopped, and the new one started.",
 			args: args{
 				ca: resource.ClientApplicator{
 					Client: &test.MockClient{
@@ -841,7 +841,7 @@ func TestReconcile(t *testing.T) {
 						MockStart:        func(_ string, _ ...engine.ControllerOption) error { return nil },
 						MockStop:         func(_ context.Context, _ string) error { return nil },
 						MockIsRunning:    func(_ string) bool { return false },
-						MockStartWatches: func(_ string, _ ...engine.Watch) error { return nil },
+						MockStartWatches: func(_ context.Context, _ string, _ ...engine.Watch) error { return nil },
 						MockGetCached:    func() client.Client { return test.NewMockClient() },
 						MockGetUncached:  func() client.Client { return test.NewMockClient() },
 					}),
@@ -852,7 +852,7 @@ func TestReconcile(t *testing.T) {
 			},
 		},
 		"NotRestartingWithoutVersionChange": {
-			reason: "We should return without requeueing if we successfully ensured our CRD exists and controller is started.",
+			reason: "We should return without requeuing if we successfully ensured our CRD exists and controller is started.",
 			args: args{
 				ca: resource.ClientApplicator{
 					Client: &test.MockClient{
diff --git a/internal/controller/apiextensions/offered/reconciler_test.go b/internal/controller/apiextensions/offered/reconciler_test.go
index adbd049af..f270e111b 100644
--- a/internal/controller/apiextensions/offered/reconciler_test.go
+++ b/internal/controller/apiextensions/offered/reconciler_test.go
@@ -45,7 +45,7 @@ type MockEngine struct {
 	MockStart        func(name string, o ...engine.ControllerOption) error
 	MockStop         func(ctx context.Context, name string) error
 	MockIsRunning    func(name string) bool
-	MockStartWatches func(name string, ws ...engine.Watch) error
+	MockStartWatches func(ctx context.Context, name string, ws ...engine.Watch) error
 	MockGetClient    func() client.Client
 }
 
@@ -66,8 +66,8 @@ func (m *MockEngine) IsRunning(name string) bool {
 	return m.MockIsRunning(name)
 }
 
-func (m *MockEngine) StartWatches(name string, ws ...engine.Watch) error {
-	return m.MockStartWatches(name, ws...)
+func (m *MockEngine) StartWatches(ctx context.Context, name string, ws ...engine.Watch) error {
+	return m.MockStartWatches(ctx, name, ws...)
 }
 
 func (m *MockEngine) GetCached() client.Client {
@@ -601,7 +601,7 @@ func TestReconcile(t *testing.T) {
 			},
 		},
 		"VersionChangedStopControllerError": {
-			reason: "We should return any error we encounter while stopping our controller because the XRD's referencable version changed.",
+			reason: "We should return any error we encounter while stopping our controller because the XRD's referenceable version changed.",
 			args: args{
 				ca: resource.ClientApplicator{
 					Client: &test.MockClient{
@@ -730,7 +730,7 @@ func TestReconcile(t *testing.T) {
 						MockStart: func(_ string, _ ...engine.ControllerOption) error {
 							return nil
 						},
-						MockStartWatches: func(_ string, _ ...engine.Watch) error {
+						MockStartWatches: func(_ context.Context, _ string, _ ...engine.Watch) error {
 							return errBoom
 						},
 						MockGetClient: func() client.Client { return test.NewMockClient() },
@@ -778,7 +778,7 @@ func TestReconcile(t *testing.T) {
 					WithControllerEngine(&MockEngine{
 						MockIsRunning:    func(_ string) bool { return false },
 						MockStart:        func(_ string, _ ...engine.ControllerOption) error { return nil },
-						MockStartWatches: func(_ string, _ ...engine.Watch) error { return nil },
+						MockStartWatches: func(_ context.Context, _ string, _ ...engine.Watch) error { return nil },
 						MockGetClient:    func() client.Client { return test.NewMockClient() },
 					},
 					),
@@ -840,7 +840,7 @@ func TestReconcile(t *testing.T) {
 						MockStart:        func(_ string, _ ...engine.ControllerOption) error { return nil },
 						MockStop:         func(_ context.Context, _ string) error { return nil },
 						MockIsRunning:    func(_ string) bool { return false },
-						MockStartWatches: func(_ string, _ ...engine.Watch) error { return nil },
+						MockStartWatches: func(_ context.Context, _ string, _ ...engine.Watch) error { return nil },
 						MockGetClient:    func() client.Client { return test.NewMockClient() },
 					}),
 				},
@@ -850,7 +850,7 @@ func TestReconcile(t *testing.T) {
 			},
 		},
 		"NotRestartingWithoutVersionChange": {
-			reason: "We should return without requeueing if we successfully ensured our CRD exists and controller is started.",
+			reason: "We should return without requeuing if we successfully ensured our CRD exists and controller is started.",
 			args: args{
 				ca: resource.ClientApplicator{
 					Client: &test.MockClient{
diff --git a/internal/engine/engine_test.go b/internal/engine/engine_test.go
index f2f1c2aeb..52444d102 100644
--- a/internal/engine/engine_test.go
+++ b/internal/engine/engine_test.go
@@ -391,6 +391,9 @@ func TestStopController(t *testing.T) {
 					MockActiveInformers: func() []schema.GroupVersionKind {
 						return nil
 					},
+					MockGetInformer: func(_ context.Context, _ client.Object, _ ...cache.InformerGetOption) (cache.Informer, error) {
+						return nil, nil
+					},
 				},
 			},
 			args: args{
@@ -428,7 +431,7 @@ func TestStopController(t *testing.T) {
 			u := &unstructured.Unstructured{}
 			u.SetAPIVersion("test.crossplane.io/v1")
 			u.SetKind("Composed")
-			err = e.StartWatches(tc.args.name, WatchFor(u, WatchTypeComposedResource, nil))
+			err = e.StartWatches(tc.args.ctx, tc.args.name, WatchFor(u, WatchTypeComposedResource, nil))
 			if diff := cmp.Diff(nil, err, cmpopts.EquateErrors()); diff != "" {
 				t.Errorf("\n%s\ne.StartWatches(...): -want error, +got error:\n%s", tc.reason, diff)
 			}
@@ -462,6 +465,7 @@ func TestStartWatches(t *testing.T) {
 		opts []ControllerOption
 	}
 	type args struct {
+		ctx  context.Context
 		name string
 		ws   []Watch
 	}
@@ -497,6 +501,9 @@ func TestStartWatches(t *testing.T) {
 							},
 						}
 					},
+					MockGetInformer: func(_ context.Context, _ client.Object, _ ...cache.InformerGetOption) (cache.Informer, error) {
+						return nil, nil
+					},
 				},
 			},
 			argsStart: argsStart{
@@ -551,6 +558,9 @@ func TestStartWatches(t *testing.T) {
 							},
 						}
 					},
+					MockGetInformer: func(_ context.Context, _ client.Object, _ ...cache.InformerGetOption) (cache.Informer, error) {
+						return nil, nil
+					},
 				},
 			},
 			argsStart: argsStart{
@@ -627,14 +637,14 @@ func TestStartWatches(t *testing.T) {
 				t.Fatalf("\n%s\ne.Start(...): -want error, +got error:\n%s", tc.reason, diff)
 			}
 
-			err = e.StartWatches(tc.args.name, tc.args.ws...)
+			err = e.StartWatches(tc.args.ctx, tc.args.name, tc.args.ws...)
 			if diff := cmp.Diff(tc.want.err, err, cmpopts.EquateErrors()); diff != "" {
 				t.Errorf("\n%s\ne.StartWatches(...): -want error, +got error:\n%s", tc.reason, diff)
 			}
 
 			// Start the same watches again to exercise the code that ensures we
 			// only add each watch once.
-			err = e.StartWatches(tc.args.name, tc.args.ws...)
+			err = e.StartWatches(tc.args.ctx, tc.args.name, tc.args.ws...)
 			if diff := cmp.Diff(tc.want.err, err, cmpopts.EquateErrors()); diff != "" {
 				t.Errorf("\n%s\ne.StartWatches(...): -want error, +got error:\n%s", tc.reason, diff)
 			}
@@ -700,6 +710,9 @@ func TestStopWatches(t *testing.T) {
 					MockActiveInformers: func() []schema.GroupVersionKind {
 						return nil
 					},
+					MockGetInformer: func(_ context.Context, _ client.Object, _ ...cache.InformerGetOption) (cache.Informer, error) {
+						return nil, nil
+					},
 				},
 			},
 			args: args{
@@ -754,6 +767,9 @@ func TestStopWatches(t *testing.T) {
 					MockActiveInformers: func() []schema.GroupVersionKind {
 						return nil
 					},
+					MockGetInformer: func(_ context.Context, _ client.Object, _ ...cache.InformerGetOption) (cache.Informer, error) {
+						return nil, nil
+					},
 				},
 			},
 			args: args{
@@ -800,6 +816,9 @@ func TestStopWatches(t *testing.T) {
 					MockActiveInformers: func() []schema.GroupVersionKind {
 						return nil
 					},
+					MockGetInformer: func(_ context.Context, _ client.Object, _ ...cache.InformerGetOption) (cache.Informer, error) {
+						return nil, nil
+					},
 				},
 			},
 			args: args{
@@ -865,7 +884,7 @@ func TestStopWatches(t *testing.T) {
 			u1 := &unstructured.Unstructured{}
 			u1.SetAPIVersion("test.crossplane.io/v1")
 			u1.SetKind("Resource")
-			err = e.StartWatches(tc.args.name,
+			err = e.StartWatches(tc.args.ctx, tc.args.name,
 				WatchFor(u1, WatchTypeComposedResource, nil),
 				WatchFor(u1, WatchTypeCompositeResource, nil),
 			)
diff --git a/internal/engine/source_test.go b/internal/engine/source_test.go
index 8e4c5520e..e1a1aad8d 100644
--- a/internal/engine/source_test.go
+++ b/internal/engine/source_test.go
@@ -23,11 +23,9 @@ import (
 
 	"github.com/google/go-cmp/cmp"
 	"github.com/google/go-cmp/cmp/cmpopts"
-	"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
 	kcache "k8s.io/client-go/tools/cache"
 	"k8s.io/client-go/util/workqueue"
 	"sigs.k8s.io/controller-runtime/pkg/cache"
-	"sigs.k8s.io/controller-runtime/pkg/client"
 	"sigs.k8s.io/controller-runtime/pkg/handler"
 	"sigs.k8s.io/controller-runtime/pkg/predicate"
 	"sigs.k8s.io/controller-runtime/pkg/reconcile"
@@ -40,6 +38,7 @@ type MockInformer struct {
 
 	MockAddEventHandler    func(handler kcache.ResourceEventHandler) (kcache.ResourceEventHandlerRegistration, error)
 	MockRemoveEventHandler func(handle kcache.ResourceEventHandlerRegistration) error
+	MockIsStopped          func() bool
 }
 
 func (m *MockInformer) AddEventHandler(handler kcache.ResourceEventHandler) (kcache.ResourceEventHandlerRegistration, error) {
@@ -50,12 +49,15 @@ func (m *MockInformer) RemoveEventHandler(handle kcache.ResourceEventHandlerRegi
 	return m.MockRemoveEventHandler(handle)
 }
 
+func (m *MockInformer) IsStopped() bool {
+	return m.MockIsStopped()
+}
+
 func TestStartSource(t *testing.T) {
 	type params struct {
-		infs cache.Informers
-		t    client.Object
-		h    handler.EventHandler
-		ps   []predicate.Predicate
+		inf cache.Informer
+		h   handler.EventHandler
+		ps  []predicate.Predicate
 	}
 	type args struct {
 		ctx context.Context
@@ -71,37 +73,14 @@ func TestStartSource(t *testing.T) {
 		args   args
 		want   want
 	}{
-		"GetInformerError": {
-			reason: "Start should return an error if it can't get an informer for the supplied type.",
-			params: params{
-				infs: &MockCache{
-					MockGetInformer: func(_ context.Context, _ client.Object, _ ...cache.InformerGetOption) (cache.Informer, error) {
-						return nil, errors.New("boom")
-					},
-				},
-				t: &unstructured.Unstructured{},
-			},
-			args: args{
-				ctx: context.Background(),
-				q:   nil, // Not called, just plumbed down to the event handler.
-			},
-			want: want{
-				err: cmpopts.AnyError,
-			},
-		},
 		"AddEventHandlerError": {
 			reason: "Start should return an error if it can't add an event handler to the informer.",
 			params: params{
-				infs: &MockCache{
-					MockGetInformer: func(_ context.Context, _ client.Object, _ ...cache.InformerGetOption) (cache.Informer, error) {
-						return &MockInformer{
-							MockAddEventHandler: func(_ kcache.ResourceEventHandler) (kcache.ResourceEventHandlerRegistration, error) {
-								return nil, errors.New("boom")
-							},
-						}, nil
+				inf: &MockInformer{
+					MockAddEventHandler: func(_ kcache.ResourceEventHandler) (kcache.ResourceEventHandlerRegistration, error) {
+						return nil, errors.New("boom")
 					},
 				},
-				t: &unstructured.Unstructured{},
 			},
 			args: args{
 				ctx: context.Background(),
@@ -114,16 +93,11 @@ func TestStartSource(t *testing.T) {
 		"SuccessfulStart": {
 			reason: "Start should return nil if it successfully starts the source.",
 			params: params{
-				infs: &MockCache{
-					MockGetInformer: func(_ context.Context, _ client.Object, _ ...cache.InformerGetOption) (cache.Informer, error) {
-						return &MockInformer{
-							MockAddEventHandler: func(_ kcache.ResourceEventHandler) (kcache.ResourceEventHandlerRegistration, error) {
-								return nil, nil
-							},
-						}, nil
+				inf: &MockInformer{
+					MockAddEventHandler: func(_ kcache.ResourceEventHandler) (kcache.ResourceEventHandlerRegistration, error) {
+						return nil, nil
 					},
 				},
-				t: &unstructured.Unstructured{},
 			},
 			args: args{
 				ctx: context.Background(),
@@ -137,7 +111,7 @@ func TestStartSource(t *testing.T) {
 
 	for name, tc := range cases {
 		t.Run(name, func(t *testing.T) {
-			s := NewStoppableSource(tc.params.infs, tc.params.t, tc.params.h, tc.params.ps...)
+			s := NewStoppableSource(tc.params.inf, tc.params.h, tc.params.ps...)
 
 			err := s.Start(tc.args.ctx, tc.args.q)
 			if diff := cmp.Diff(tc.want.err, err, cmpopts.EquateErrors()); diff != "" {
@@ -155,10 +129,9 @@ func (m *MockRegistration) HasSynced() bool { return true }
 
 func TestStopSource(t *testing.T) {
 	type params struct {
-		infs cache.Informers
-		t    client.Object
-		h    handler.EventHandler
-		ps   []predicate.Predicate
+		inf cache.Informer
+		h   handler.EventHandler
+		ps  []predicate.Predicate
 	}
 	type args struct {
 		ctx context.Context
@@ -168,9 +141,6 @@ func TestStopSource(t *testing.T) {
 		err error
 	}
 
-	// Used to return an error only when getting an informer to stop it.
-	started := false
-
 	cases := map[string]struct {
 		reason string
 		params params
@@ -180,70 +150,36 @@ func TestStopSource(t *testing.T) {
 		"SuccessfulStop": {
 			reason: "Stop should return nil if it successfully stops the source.",
 			params: params{
-				infs: &MockCache{
-					MockGetInformer: func(_ context.Context, _ client.Object, _ ...cache.InformerGetOption) (cache.Informer, error) {
-						return &MockInformer{
-							MockAddEventHandler: func(_ kcache.ResourceEventHandler) (kcache.ResourceEventHandlerRegistration, error) {
-								return &MockRegistration{}, nil
-							},
-							MockRemoveEventHandler: func(_ kcache.ResourceEventHandlerRegistration) error {
-								return nil
-							},
-						}, nil
+				inf: &MockInformer{
+					MockAddEventHandler: func(_ kcache.ResourceEventHandler) (kcache.ResourceEventHandlerRegistration, error) {
+						return &MockRegistration{}, nil
 					},
-				},
-				t: &unstructured.Unstructured{},
-			},
-			args: args{
-				ctx: context.Background(),
-				q:   nil, // Not called, just plumbed down to the event handler.
-			},
-			want: want{
-				err: nil,
-			},
-		},
-		"GetInformerError": {
-			reason: "Stop should return an error if it can't get an informer.",
-			params: params{
-				infs: &MockCache{
-					MockGetInformer: func(_ context.Context, _ client.Object, _ ...cache.InformerGetOption) (cache.Informer, error) {
-						if !started {
-							started = true
-							return &MockInformer{
-								MockAddEventHandler: func(_ kcache.ResourceEventHandler) (kcache.ResourceEventHandlerRegistration, error) {
-									return &MockRegistration{}, nil
-								},
-							}, nil
-						}
-						return nil, errors.New("boom")
+					MockRemoveEventHandler: func(_ kcache.ResourceEventHandlerRegistration) error {
+						return nil
 					},
+					MockIsStopped: func() bool { return false },
 				},
-				t: &unstructured.Unstructured{},
 			},
 			args: args{
 				ctx: context.Background(),
 				q:   nil, // Not called, just plumbed down to the event handler.
 			},
 			want: want{
-				err: cmpopts.AnyError,
+				err: nil,
 			},
 		},
 		"RemoveEventHandlerError": {
 			reason: "Stop should return an error if it can't remove the source's event handler.",
 			params: params{
-				infs: &MockCache{
-					MockGetInformer: func(_ context.Context, _ client.Object, _ ...cache.InformerGetOption) (cache.Informer, error) {
-						return &MockInformer{
-							MockAddEventHandler: func(_ kcache.ResourceEventHandler) (kcache.ResourceEventHandlerRegistration, error) {
-								return &MockRegistration{}, nil
-							},
-							MockRemoveEventHandler: func(_ kcache.ResourceEventHandlerRegistration) error {
-								return errors.New("boom")
-							},
-						}, nil
+				inf: &MockInformer{
+					MockAddEventHandler: func(_ kcache.ResourceEventHandler) (kcache.ResourceEventHandlerRegistration, error) {
+						return &MockRegistration{}, nil
+					},
+					MockRemoveEventHandler: func(_ kcache.ResourceEventHandlerRegistration) error {
+						return errors.New("boom")
 					},
+					MockIsStopped: func() bool { return false },
 				},
-				t: &unstructured.Unstructured{},
 			},
 			args: args{
 				ctx: context.Background(),
@@ -257,7 +193,7 @@ func TestStopSource(t *testing.T) {
 
 	for name, tc := range cases {
 		t.Run(name, func(t *testing.T) {
-			s := NewStoppableSource(tc.params.infs, tc.params.t, tc.params.h, tc.params.ps...)
+			s := NewStoppableSource(tc.params.inf, tc.params.h, tc.params.ps...)
 
 			err := s.Start(tc.args.ctx, tc.args.q)
 			if diff := cmp.Diff(nil, err, cmpopts.EquateErrors()); diff != "" {
diff --git a/internal/xpkg/name_test.go b/internal/xpkg/name_test.go
index 1bb6e4a53..6234d6b27 100644
--- a/internal/xpkg/name_test.go
+++ b/internal/xpkg/name_test.go
@@ -231,3 +231,67 @@ func TestBuildPath(t *testing.T) {
 		})
 	}
 }
+
+func TestReplaceExt(t *testing.T) {
+	type args struct {
+		path string
+		ext  string
+	}
+
+	cases := map[string]struct {
+		reason string
+		args   args
+		want   string
+	}{
+		"ReplaceWithTxt": {
+			reason: "Should replace the existing extension with .txt",
+			args: args{
+				path: "file.doc",
+				ext:  ".txt",
+			},
+			want: "file.txt",
+		},
+		"ReplaceWithEmpty": {
+			reason: "Should remove the extension if an empty string is given",
+			args: args{
+				path: "file.doc",
+				ext:  "",
+			},
+			want: "file",
+		},
+		"NoExtensionToAdd": {
+			reason: "Should add an extension if there was none before",
+			args: args{
+				path: "file",
+				ext:  ".txt",
+			},
+			want: "file.txt",
+		},
+		"MultipleDots": {
+			reason: "Should correctly replace only the last extension",
+			args: args{
+				path: "archive.tar.gz",
+				ext:  ".zip",
+			},
+			want: "archive.tar.zip",
+		},
+		"HiddenFile": {
+			reason: "Should correctly replace extension of hidden files",
+			args: args{
+				path: ".hiddenfile.conf",
+				ext:  ".bak",
+			},
+			want: ".hiddenfile.bak",
+		},
+	}
+
+	for name, tc := range cases {
+		t.Run(name, func(t *testing.T) {
+			got := ReplaceExt(tc.args.path, tc.args.ext)
+
+			if diff := cmp.Diff(tc.want, got); diff != "" {
+				t.Errorf("\n%s\nReplaceExt(...): -want, +got:\n%s", tc.reason, diff)
+			}
+		})
+	}
+}
diff --git a/test/e2e/README.md b/test/e2e/README.md
index f7bc7cb87..77e3bed99 100644
--- a/test/e2e/README.md
+++ b/test/e2e/README.md
@@ -89,21 +89,6 @@ $ apk add kubectl
 # Now you can use kubectl to access the cluster
 ```
 
-### Test Logging
-
-The E2E tests are run using [`gotestsum`] that helps output the test results in
-a format that integrates easily with CI systems. The downside though is that
-test logging statements, like `t.Logf("foo")`, are not streamed in real time and
-they are only included in the output if there is a failure.
-
-To change this behavior when running the E2E tests locally, you can update the
-`Earthfile` to call `gotestsum` with a format like `--format standard-verbose`
-instead, for example:
-
-```Dockerfile
-RUN gotestsum --no-color=false --format standard-verbose ...
-```
-
 ### Running Tests Directly
 
 The E2E tests are typically run via `earthly` for convenience and consistency.
diff --git a/test/e2e/apiextensions_test.go b/test/e2e/apiextensions_test.go
index b4b6f6c6c..d793ad751 100644
--- a/test/e2e/apiextensions_test.go
+++ b/test/e2e/apiextensions_test.go
@@ -142,9 +142,13 @@ func TestCompositionPatchAndTransform(t *testing.T) {
 				funcs.ResourcesCreatedWithin(30*time.Second, manifests, "claim.yaml"),
 			)).
 			Assess("ClaimIsReady",
-				funcs.ResourcesHaveConditionWithin(5*time.Minute, manifests, "claim.yaml", xpv1.Available())).
+				funcs.ResourcesHaveConditionWithin(30*time.Second, manifests, "claim.yaml", xpv1.Available())).
+			// This is testing realtime composition. The composition
+			// patches this value from the XR to a composed resource
+			// and back to the XR. We rely on a watch of the
+			// composed resource to notice this within 30 seconds.
 			Assess("ClaimHasPatchedField",
-				funcs.ResourcesHaveFieldValueWithin(5*time.Minute, manifests, "claim.yaml", "status.coolerField", "I'M COOL!"),
+				funcs.ResourcesHaveFieldValueWithin(30*time.Second, manifests, "claim.yaml", "status.coolerField", "I'M COOL!"),
 			).
 			WithTeardown("DeleteClaim", funcs.AllOf(
 				funcs.DeleteResources(manifests, "claim.yaml"),
@@ -155,7 +159,7 @@ func TestCompositionPatchAndTransform(t *testing.T) {
 	)
 }
 
-func TestCompositionRealtimeRevisionSelection(t *testing.T) {
+func TestCompositionRevisionSelection(t *testing.T) {
 	manifests := "test/e2e/manifests/apiextensions/composition/realtime-revision-selection"
 	environment.Test(t,
 		features.NewWithDescription(t.Name(), "Tests Crossplane's Composition functionality to react in realtime to changes in a Composition by selecting the new CompositionRevision and reconcile the XRs.").
@@ -173,13 +177,13 @@ func TestCompositionRealtimeRevisionSelection(t *testing.T) {
 				funcs.ResourcesCreatedWithin(30*time.Second, manifests, "claim.yaml"),
 			)).
 			Assess("ClaimIsReady",
-				funcs.ResourcesHaveConditionWithin(5*time.Minute, manifests, "claim.yaml", xpv1.Available()),
+				funcs.ResourcesHaveConditionWithin(30*time.Second, manifests, "claim.yaml", xpv1.Available()),
 			).
 			Assess("UpdateComposition", funcs.AllOf(
 				funcs.ApplyResources(FieldManager, manifests, "composition-update.yaml"),
 			)).
 			Assess("ClaimHasPatchedField",
-				funcs.ResourcesHaveFieldValueWithin(10*time.Second, manifests, "claim.yaml", "status.coolerField", "I'M COOL!"),
+				funcs.ResourcesHaveFieldValueWithin(30*time.Second, manifests, "claim.yaml", "status.coolerField", "I'M COOL!"),
 			).
 			WithTeardown("DeleteClaim", funcs.AllOf(
 				funcs.DeleteResources(manifests, "claim.yaml"),
@@ -207,10 +211,14 @@ func TestCompositionFunctions(t *testing.T) {
 				funcs.ApplyResources(FieldManager, manifests, "claim.yaml"),
 				funcs.ResourcesCreatedWithin(30*time.Second, manifests, "claim.yaml"),
 			)).
+			// TODO(negz): This Assess consistently takes ~2 seconds
+			// on my M1 Max MacBook, but sometimes exceeds 30
+			// seconds in CI, even with larger runners. Perhaps
+			// slower I/O in CI?
 			Assess("ClaimIsReady",
-				funcs.ResourcesHaveConditionWithin(5*time.Minute, manifests, "claim.yaml", xpv1.Available())).
+				funcs.ResourcesHaveConditionWithin(60*time.Second, manifests, "claim.yaml", xpv1.Available())).
 			Assess("ClaimHasPatchedField",
-				funcs.ResourcesHaveFieldValueWithin(5*time.Minute, manifests, "claim.yaml", "status.coolerField", "I'M COOLER!"),
+				funcs.ResourcesHaveFieldValueWithin(30*time.Second, manifests, "claim.yaml", "status.coolerField", "I'M COOLER!"),
 			).
 			WithTeardown("DeleteClaim", funcs.AllOf(
 				funcs.DeleteResources(manifests, "claim.yaml"),
diff --git a/test/e2e/manifests/apiextensions/composition/realtime-compositions/claim.yaml b/test/e2e/manifests/apiextensions/composition/realtime-compositions/claim.yaml
deleted file mode 100644
index 7af5c687c..000000000
--- a/test/e2e/manifests/apiextensions/composition/realtime-compositions/claim.yaml
+++ /dev/null
@@ -1,12 +0,0 @@
-apiVersion: realtime-compositions.e2e.crossplane.io/v1alpha1
-kind: NopResource
-metadata:
-  namespace: default
-  name: realtime-composition
-  labels:
-    realtime-compositions: "true"
-spec:
-  coolField: "I'm cool!"
-  # This is necessary to ensure the claim's MRs are actually gone before we
-  # delete the Provider - https://github.com/crossplane/crossplane/issues/4251
-  compositeDeletePolicy: Foreground
diff --git a/test/e2e/manifests/apiextensions/composition/realtime-compositions/setup/composition.yaml b/test/e2e/manifests/apiextensions/composition/realtime-compositions/setup/composition.yaml
deleted file mode 100644
index fbf972f2d..000000000
--- a/test/e2e/manifests/apiextensions/composition/realtime-compositions/setup/composition.yaml
+++ /dev/null
@@ -1,26 +0,0 @@
-apiVersion: apiextensions.crossplane.io/v1
-kind: Composition
-metadata:
-  name: xnopresources.realtime-compositions.e2e.crossplane.io
-spec:
-  compositeTypeRef:
-    apiVersion: realtime-compositions.e2e.crossplane.io/v1alpha1
-    kind: XNopResource
-  resources:
-  - name: nop-resource-1
-    base:
-      apiVersion: nop.crossplane.io/v1alpha1
-      kind: NopResource
-      metadata:
-        labels:
-          realtime-compositions: "true"
-      spec:
-        forProvider:
-          conditionAfter:
-          - conditionType: Ready
-            conditionStatus: "True"
-            time: 0s
-    patches:
-    - type: ToCompositeFieldPath
-      fromFieldPath: metadata.annotations[cool-field]
-      toFieldPath: status.coolerField
diff --git a/test/e2e/manifests/apiextensions/composition/realtime-compositions/setup/definition.yaml b/test/e2e/manifests/apiextensions/composition/realtime-compositions/setup/definition.yaml
deleted file mode 100644
index 9255806d9..000000000
--- a/test/e2e/manifests/apiextensions/composition/realtime-compositions/setup/definition.yaml
+++ /dev/null
@@ -1,34 +0,0 @@
-apiVersion: apiextensions.crossplane.io/v1
-kind: CompositeResourceDefinition
-metadata:
-  name: xnopresources.realtime-compositions.e2e.crossplane.io
-spec:
-  group: realtime-compositions.e2e.crossplane.io
-  names:
-    kind: XNopResource
-    plural: xnopresources
-  claimNames:
-    kind: NopResource
-    plural: nopresources
-  connectionSecretKeys:
-  - test
-  versions:
-  - name: v1alpha1
-    served: true
-    referenceable: true
-    schema:
-     openAPIV3Schema:
-       type: object
-       properties:
-        spec:
-          type: object
-          properties:
-            coolField:
-              type: string
-          required:
-          - coolField
-        status:
-          type: object
-          properties:
-            coolerField:
-              type: string
\ No newline at end of file
diff --git a/test/e2e/manifests/apiextensions/composition/realtime-compositions/setup/provider.yaml b/test/e2e/manifests/apiextensions/composition/realtime-compositions/setup/provider.yaml
deleted file mode 100644
index d3fb1a026..000000000
--- a/test/e2e/manifests/apiextensions/composition/realtime-compositions/setup/provider.yaml
+++ /dev/null
@@ -1,7 +0,0 @@
-apiVersion: pkg.crossplane.io/v1
-kind: Provider
-metadata:
-  name: provider-nop
-spec:
-  package: xpkg.upbound.io/crossplane-contrib/provider-nop:v0.3.0
-  ignoreCrossplaneConstraints: true
\ No newline at end of file
diff --git a/test/e2e/manifests/apiextensions/usage/composition-pipeline/claim.yaml b/test/e2e/manifests/apiextensions/usage/composition-pipeline/claim.yaml
deleted file mode 100644
index 6df9dd014..000000000
--- a/test/e2e/manifests/apiextensions/usage/composition-pipeline/claim.yaml
+++ /dev/null
@@ -1,7 +0,0 @@
-apiVersion: nop.example.org/v1alpha1
-kind: NopResource
-metadata:
-  namespace: default
-  name: test
-spec:
-  coolField: "I'm cool!"
diff --git a/test/e2e/manifests/apiextensions/usage/composition-pipeline/setup/composition.yaml b/test/e2e/manifests/apiextensions/usage/composition-pipeline/setup/composition.yaml
deleted file mode 100644
index 1b33cde4b..000000000
--- a/test/e2e/manifests/apiextensions/usage/composition-pipeline/setup/composition.yaml
+++ /dev/null
@@ -1,87 +0,0 @@
-apiVersion: apiextensions.crossplane.io/v1
-kind: Composition
-metadata:
-  name: xnopresources.nop.example.org
-spec:
-  compositeTypeRef:
-    apiVersion: nop.example.org/v1alpha1
-    kind: XNopResource
-  mode: Pipeline
-  pipeline:
-    - step: be-a-dummy
-      functionRef:
-        name: function-dummy
-      input:
-        apiVersion: dummy.fn.crossplane.io/v1beta1
-        kind: Response
-        # This is a YAML-serialized RunFunctionResponse. function-dummy will
-        # overlay the desired state on any that was passed into it.
-        response:
-          desired:
-            resources:
-              used-resource:
-                resource:
-                  apiVersion: nop.crossplane.io/v1alpha1
-                  kind: NopResource
-                  metadata:
-                    labels:
-                      usage: used
-                  spec:
-                    forProvider:
-                      conditionAfter:
-                        - conditionType: "Synced"
-                          conditionStatus: "True"
-                          time: "5s"
-                        - conditionType: "Ready"
-                          conditionStatus: "True"
-                          time: "10s"
-              using-resource:
-                resource:
-                  apiVersion: nop.crossplane.io/v1alpha1
-                  kind: NopResource
-                  metadata:
-                    # We are delaying deletion of using resource with this finalizer. This is to ensure that the used resource is
-                    # not deleted before the using resource. Imagine a scenario where a Release resource using a Cluster resource
-                    # where we expect the Cluster resource not to be deleted until the Release resource is deleted. We are mimicking
-                    # this behavior with the help of a finalizer on a NopResource.
-                    finalizers:
-                      - delay-deletion-of-using-resource
-                    labels:
-                      usage: using
-                  spec:
-                    forProvider:
-                      conditionAfter:
-                        - conditionType: "Synced"
-                          conditionStatus: "True"
-                          time: "5s"
-                        - conditionType: "Ready"
-                          conditionStatus: "True"
-                          time: "10s"
-              usage-resource:
-                resource:
-                  apiVersion: apiextensions.crossplane.io/v1beta1
-                  kind: Usage
-                  metadata:
-                    labels:
-                      version: initial
-                  spec:
-                    of:
-                      apiVersion: nop.crossplane.io/v1alpha1
-                      kind: NopResource
-                      resourceSelector:
-                        matchControllerRef: true
-                        matchLabels:
-                          usage: used
-                    by:
-                      apiVersion: nop.crossplane.io/v1alpha1
-                      kind: NopResource
-                      resourceSelector:
-                        matchControllerRef: true
-                        matchLabels:
-                          usage: using
-          results:
-            - severity: SEVERITY_NORMAL
-              message: "I am doing a compose!"
-    - step: detect-readiness
-      functionRef:
-        name: function-auto-ready
\ No newline at end of file
diff --git a/test/e2e/manifests/apiextensions/usage/composition-pipeline/setup/definition.yaml b/test/e2e/manifests/apiextensions/usage/composition-pipeline/setup/definition.yaml
deleted file mode 100644
index bf70cb798..000000000
--- a/test/e2e/manifests/apiextensions/usage/composition-pipeline/setup/definition.yaml
+++ /dev/null
@@ -1,29 +0,0 @@
-apiVersion: apiextensions.crossplane.io/v1
-kind: CompositeResourceDefinition
-metadata:
-  name: xnopresources.nop.example.org
-spec:
-  group: nop.example.org
-  names:
-    kind: XNopResource
-    plural: xnopresources
-  claimNames:
-    kind: NopResource
-    plural: nopresources
-  connectionSecretKeys:
-  - test
-  versions:
-  - name: v1alpha1
-    served: true
-    referenceable: true
-    schema:
-     openAPIV3Schema:
-       type: object
-       properties:
-        spec:
-          type: object
-          properties:
-            coolField:
-              type: string
-          required:
-          - coolField
\ No newline at end of file
diff --git a/test/e2e/manifests/apiextensions/usage/composition-pipeline/setup/provider.yaml b/test/e2e/manifests/apiextensions/usage/composition-pipeline/setup/provider.yaml
deleted file mode 100644
index d3fb1a026..000000000
--- a/test/e2e/manifests/apiextensions/usage/composition-pipeline/setup/provider.yaml
+++ /dev/null
@@ -1,7 +0,0 @@
-apiVersion: pkg.crossplane.io/v1
-kind: Provider
-metadata:
-  name: provider-nop
-spec:
-  package: xpkg.upbound.io/crossplane-contrib/provider-nop:v0.3.0
-  ignoreCrossplaneConstraints: true
\ No newline at end of file
diff --git a/test/e2e/manifests/apiextensions/usage/composition-pipeline/composition-updated.yaml b/test/e2e/manifests/apiextensions/usage/composition/composition-updated.yaml
similarity index 100%
rename from test/e2e/manifests/apiextensions/usage/composition-pipeline/composition-updated.yaml
rename to test/e2e/manifests/apiextensions/usage/composition/composition-updated.yaml
diff --git a/test/e2e/manifests/apiextensions/usage/composition/setup/composition.yaml b/test/e2e/manifests/apiextensions/usage/composition/setup/composition.yaml
index 2f4b0d565..1b33cde4b 100644
--- a/test/e2e/manifests/apiextensions/usage/composition/setup/composition.yaml
+++ b/test/e2e/manifests/apiextensions/usage/composition/setup/composition.yaml
@@ -6,61 +6,82 @@ spec:
   compositeTypeRef:
     apiVersion: nop.example.org/v1alpha1
     kind: XNopResource
-  resources:
-  - name: used-resource
-    base:
-      apiVersion: nop.crossplane.io/v1alpha1
-      kind: NopResource
-      metadata:
-        labels:
-            usage: used
-      spec:
-        forProvider:
-          conditionAfter:
-          - conditionType: "Synced"
-            conditionStatus: "True"
-            time: "5s"
-          - conditionType: "Ready"
-            conditionStatus: "True"
-            time: "10s"
-  - name: using-resource
-    base:
-      apiVersion: nop.crossplane.io/v1alpha1
-      kind: NopResource
-      metadata:
-        # We are delaying deletion of using resource with this finalizer. This is to ensure that the used resource is
-        # not deleted before the using resource. Imagine a scenario where a Release resource using a Cluster resource
-        # where we expect the Cluster resource not to be deleted until the Release resource is deleted. We are mimicking
-        # this behavior with the help of a finalizer on a NopResource.
-        finalizers:
-          - delay-deletion-of-using-resource
-        labels:
-          usage: using
-      spec:
-        forProvider:
-          conditionAfter:
-          - conditionType: "Synced"
-            conditionStatus: "True"
-            time: "5s"
-          - conditionType: "Ready"
-            conditionStatus: "True"
-            time: "10s"
-  - name: usage-resource
-    base:
-      apiVersion: apiextensions.crossplane.io/v1beta1
-      kind: Usage
-      spec:
-        of:
-          apiVersion: nop.crossplane.io/v1alpha1
-          kind: NopResource
-          resourceSelector:
-            matchControllerRef: true
-            matchLabels:
-              usage: used
-        by:
-          apiVersion: nop.crossplane.io/v1alpha1
-          kind: NopResource
-          resourceSelector:
-            matchControllerRef: true
-            matchLabels:
-              usage: using
+  mode: Pipeline
+  pipeline:
+    - step: be-a-dummy
+      functionRef:
+        name: function-dummy
+      input:
+        apiVersion: dummy.fn.crossplane.io/v1beta1
+        kind: Response
+        # This is a YAML-serialized RunFunctionResponse. function-dummy will
+        # overlay the desired state on any that was passed into it.
+        response:
+          desired:
+            resources:
+              used-resource:
+                resource:
+                  apiVersion: nop.crossplane.io/v1alpha1
+                  kind: NopResource
+                  metadata:
+                    labels:
+                      usage: used
+                  spec:
+                    forProvider:
+                      conditionAfter:
+                        - conditionType: "Synced"
+                          conditionStatus: "True"
+                          time: "5s"
+                        - conditionType: "Ready"
+                          conditionStatus: "True"
+                          time: "10s"
+              using-resource:
+                resource:
+                  apiVersion: nop.crossplane.io/v1alpha1
+                  kind: NopResource
+                  metadata:
+                    # We are delaying deletion of using resource with this finalizer. This is to ensure that the used resource is
+                    # not deleted before the using resource. Imagine a scenario where a Release resource using a Cluster resource
+                    # where we expect the Cluster resource not to be deleted until the Release resource is deleted. We are mimicking
+                    # this behavior with the help of a finalizer on a NopResource.
+                    finalizers:
+                      - delay-deletion-of-using-resource
+                    labels:
+                      usage: using
+                  spec:
+                    forProvider:
+                      conditionAfter:
+                        - conditionType: "Synced"
+                          conditionStatus: "True"
+                          time: "5s"
+                        - conditionType: "Ready"
+                          conditionStatus: "True"
+                          time: "10s"
+              usage-resource:
+                resource:
+                  apiVersion: apiextensions.crossplane.io/v1beta1
+                  kind: Usage
+                  metadata:
+                    labels:
+                      version: initial
+                  spec:
+                    of:
+                      apiVersion: nop.crossplane.io/v1alpha1
+                      kind: NopResource
+                      resourceSelector:
+                        matchControllerRef: true
+                        matchLabels:
+                          usage: used
+                    by:
+                      apiVersion: nop.crossplane.io/v1alpha1
+                      kind: NopResource
+                      resourceSelector:
+                        matchControllerRef: true
+                        matchLabels:
+                          usage: using
+          results:
+            - severity: SEVERITY_NORMAL
+              message: "I am doing a compose!"
+    - step: detect-readiness
+      functionRef:
+        name: function-auto-ready
\ No newline at end of file
diff --git a/test/e2e/manifests/apiextensions/usage/composition-pipeline/setup/functions.yaml b/test/e2e/manifests/apiextensions/usage/composition/setup/functions.yaml
similarity index 100%
rename from test/e2e/manifests/apiextensions/usage/composition-pipeline/setup/functions.yaml
rename to test/e2e/manifests/apiextensions/usage/composition/setup/functions.yaml
diff --git a/test/e2e/realtimecompositions_test.go b/test/e2e/realtimecompositions_test.go
deleted file mode 100644
index 251115900..000000000
--- a/test/e2e/realtimecompositions_test.go
+++ /dev/null
@@ -1,107 +0,0 @@
-package e2e
-
-import (
-	"testing"
-	"time"
-
-	corev1 "k8s.io/api/core/v1"
-	"k8s.io/apimachinery/pkg/labels"
-	"sigs.k8s.io/e2e-framework/klient/k8s"
-	"sigs.k8s.io/e2e-framework/klient/k8s/resources"
-	"sigs.k8s.io/e2e-framework/pkg/features"
-	"sigs.k8s.io/e2e-framework/third_party/helm"
-
-	xpv1 "github.com/crossplane/crossplane-runtime/apis/common/v1"
-	"github.com/crossplane/crossplane-runtime/pkg/resource/unstructured/composed"
-
-	apiextensionsv1 "github.com/crossplane/crossplane/apis/apiextensions/v1"
-	pkgv1 "github.com/crossplane/crossplane/apis/pkg/v1"
-	"github.com/crossplane/crossplane/test/e2e/config"
-	"github.com/crossplane/crossplane/test/e2e/funcs"
-)
-
-const (
-	// SuiteRealtimeCompositions is the value for the config.LabelTestSuite
-	// label to be assigned to tests that should be part of the Realtime
-	// Compositions test suite.
-	SuiteRealtimeCompositions = "realtime-compositions"
-)
-
-func init() {
-	environment.AddTestSuite(SuiteRealtimeCompositions,
-		config.WithHelmInstallOpts(
-			helm.WithArgs("--set args={--debug,--enable-realtime-compositions}"),
-		),
-		config.WithLabelsToSelect(features.Labels{
-			config.LabelTestSuite: []string{SuiteRealtimeCompositions, config.TestSuiteDefault},
-		}),
-	)
-}
-
-// TestRealtimeCompositions tests scenarios for compositions with realtime
-// reconciles through MR updates.
-func TestRealtimeCompositions(t *testing.T) {
-	manifests := "test/e2e/manifests/apiextensions/composition/realtime-compositions"
-
-	claimList := composed.NewList(composed.FromReferenceToList(corev1.ObjectReference{
-		APIVersion: "realtime-revision-selection.e2e.crossplane.io/v1alpha1",
-		Kind:       "NopResource",
-	}))
-	xrList := composed.NewList(composed.FromReferenceToList(corev1.ObjectReference{
-		APIVersion: "realtime-revision-selection.e2e.crossplane.io/v1alpha1",
-		Kind:       "XNopResource",
-	}))
-	withTestLabels := resources.WithLabelSelector(labels.FormatLabels(map[string]string{"realtime-compositions": "true"}))
-
-	environment.Test(t,
-		features.NewWithDescription(t.Name(), "Tests scenarios for compositions with realtime reconciles through MR updates.").
-			WithLabel(LabelStage, LabelStageAlpha).
-			WithLabel(LabelArea, LabelAreaAPIExtensions).
-			WithLabel(LabelSize, LabelSizeSmall).
-			WithLabel(LabelModifyCrossplaneInstallation, LabelModifyCrossplaneInstallationTrue).
-			WithLabel(config.LabelTestSuite, SuiteRealtimeCompositions).
-			WithSetup("EnableAlphaRealtimeCompositions", funcs.AllOf(
-				funcs.AsFeaturesFunc(environment.HelmUpgradeCrossplaneToSuite(SuiteRealtimeCompositions)),
-				funcs.ReadyToTestWithin(1*time.Minute, namespace),
-			)).
-			WithSetup("PrerequisitesAreCreated", funcs.AllOf(
-				funcs.ApplyResources(FieldManager, manifests, "setup/*.yaml"),
-				funcs.ResourcesCreatedWithin(30*time.Second, manifests, "setup/*.yaml"),
-				funcs.ResourcesHaveConditionWithin(1*time.Minute, manifests, "setup/definition.yaml", apiextensionsv1.WatchingComposite()),
-				funcs.ResourcesHaveConditionWithin(2*time.Minute, manifests, "setup/provider.yaml", pkgv1.Healthy(), pkgv1.Active()),
-			)).
-			Assess("CreateClaim", funcs.AllOf(
-				funcs.ApplyResources(FieldManager, manifests, "claim.yaml"),
-				funcs.InBackground(funcs.LogResources(claimList, withTestLabels)),
-				funcs.InBackground(funcs.LogResources(xrList, withTestLabels)),
-				funcs.InBackground(funcs.LogResources(nopList, withTestLabels)),
-				funcs.ResourcesCreatedWithin(30*time.Second, manifests, "claim.yaml"),
-				funcs.ResourcesHaveConditionWithin(5*time.Minute, manifests, "claim.yaml", xpv1.Available()),
-			)).
-			Assess("UpdateMR", funcs.AllOf(
-				funcs.ListedResourcesModifiedWith(nopList, 1, func(object k8s.Object) {
-					anns := object.GetAnnotations()
-					if anns == nil {
-						anns = make(map[string]string)
-					}
-					anns["cool-field"] = "I'M COOL!"
-					object.SetAnnotations(anns)
-				}, withTestLabels),
-			)).
-			Assess("ClaimHasPatchedField",
-				// 10 seconds is a long time for a realtime composition, but
-				// considerably below the normal reconcile time for a XR.
-				funcs.ResourcesHaveFieldValueWithin(10*time.Second, manifests, "claim.yaml", "status.coolerField", "I'M COOL!"),
-			).
-			WithTeardown("DeleteClaim", funcs.AllOf(
-				funcs.DeleteResources(manifests, "claim.yaml"),
-				funcs.ResourcesDeletedWithin(2*time.Minute, manifests, "claim.yaml"),
-			)).
-			WithTeardown("DeletePrerequisites", funcs.ResourcesDeletedAfterListedAreGone(3*time.Minute, manifests, "setup/*.yaml", nopList)).
-			WithTeardown("DisableAlphaRealtimeCompositions", funcs.AllOf(
-				funcs.AsFeaturesFunc(environment.HelmUpgradeCrossplaneToBase()), // Disable our feature flag.
-				funcs.ReadyToTestWithin(1*time.Minute, namespace),
-			)).
-			Feature(),
-	)
-}
diff --git a/test/e2e/usage_test.go b/test/e2e/usage_test.go
index 783f0fab5..f26a77f15 100644
--- a/test/e2e/usage_test.go
+++ b/test/e2e/usage_test.go
@@ -83,11 +83,6 @@ func TestUsageStandalone(t *testing.T) {
 				funcs.DeleteResources(manifests, "setup/*.yaml"),
 				funcs.ResourcesDeletedWithin(3*time.Minute, manifests, "setup/*.yaml"),
 			)).
-			// Disable our feature flag.
-			WithTeardown("DisableAlphaUsages", funcs.AllOf(
-				funcs.AsFeaturesFunc(environment.HelmUpgradeCrossplaneToBase()),
-				funcs.ReadyToTestWithin(1*time.Minute, namespace),
-			)).
 			Feature(),
 	)
 }
@@ -100,83 +95,6 @@ func TestUsageComposition(t *testing.T) {
 		Kind:       "Usage",
 	}))
 
-	environment.Test(t,
-		features.NewWithDescription(t.Name(), "Tests scenarios for Crossplane's `Usage` resource as part of a composition.").
-			WithLabel(LabelStage, LabelStageBeta).
-			WithLabel(LabelArea, LabelAreaAPIExtensions).
-			WithLabel(LabelSize, LabelSizeSmall).
-			WithLabel(config.LabelTestSuite, config.TestSuiteDefault).
-			WithSetup("PrerequisitesAreCreated", funcs.AllOf(
-				funcs.ApplyResources(FieldManager, manifests, "setup/*.yaml"),
-				funcs.ResourcesCreatedWithin(30*time.Second, manifests, "setup/*.yaml"),
-				funcs.ResourcesHaveConditionWithin(1*time.Minute, manifests, "setup/definition.yaml", apiextensionsv1.WatchingComposite()),
-				funcs.ResourcesHaveConditionWithin(2*time.Minute, manifests, "setup/provider.yaml", pkgv1.Healthy(), pkgv1.Active()),
-			)).
-			Assess("ClaimCreatedAndReady", funcs.AllOf(
-				funcs.ApplyResources(FieldManager, manifests, "claim.yaml"),
-				funcs.ResourcesCreatedWithin(30*time.Second, manifests, "claim.yaml"),
-				funcs.ResourcesHaveConditionWithin(5*time.Minute, manifests, "claim.yaml", xpv1.Available()),
-			)).
-			Assess("UsedResourceHasInUseLabel", funcs.AllOf(
-				funcs.ComposedResourcesHaveFieldValueWithin(1*time.Minute, manifests, "claim.yaml", "metadata.labels[crossplane.io/in-use]", "true", func(object k8s.Object) bool {
-					return object.GetLabels()["usage"] == "used"
-				}),
-			)).
-			Assess("ClaimDeleted", funcs.AllOf(
-				funcs.DeleteResources(manifests, "claim.yaml"),
-				funcs.ResourcesDeletedWithin(1*time.Minute, manifests, "claim.yaml"),
-			)).
-			// NOTE(turkenh): At this point, the claim is deleted and hence the
-			// garbage collector started attempting to delete all composed
-			// resources. With the help of a finalizer (namely
-			// `delay-deletion-of-using-resource`, see in the composition),
-			// we know that the using resource is still there and hence the
-			// deletion of the used resource should be blocked. We will assess
-			// that below.
-			Assess("OthersDeletedExceptUsed", funcs.AllOf(
-				// Using resource should have a deletion timestamp (i.e. deleted by the garbage collector).
-				funcs.ListedResourcesValidatedWithin(1*time.Minute, nopList, 1, func(object k8s.Object) bool {
-					return object.GetDeletionTimestamp() != nil
-				}, resources.WithLabelSelector(labels.FormatLabels(map[string]string{"usage": "using"}))),
-				// Usage resource should not have a deletion timestamp since it is owned by the using resource.
-				funcs.ListedResourcesValidatedWithin(1*time.Minute, usageList, 1, func(object k8s.Object) bool {
-					return object.GetDeletionTimestamp() == nil
-				}),
-				// Used resource should not have a deletion timestamp since it is still in use.
-				funcs.ListedResourcesValidatedWithin(1*time.Minute, nopList, 1, func(object k8s.Object) bool {
-					return object.GetDeletionTimestamp() == nil
-				}, resources.WithLabelSelector(labels.FormatLabels(map[string]string{"usage": "used"}))),
-			)).
-			Assess("UsingDeletedAllGone", funcs.AllOf(
-				// Remove the finalizer from the using resource.
-				funcs.ListedResourcesModifiedWith(nopList, 1, func(object k8s.Object) {
-					object.SetFinalizers(nil)
-				}, resources.WithLabelSelector(labels.FormatLabels(map[string]string{"usage": "using"}))),
-				// All composed resources should now be deleted including the Usage itself.
-				funcs.ListedResourcesDeletedWithin(2*time.Minute, nopList),
-				funcs.ListedResourcesDeletedWithin(2*time.Minute, usageList),
-			)).
-			WithTeardown("DeletePrerequisites", funcs.AllOf(
-				funcs.DeleteResources(manifests, "setup/*.yaml"),
-				funcs.ResourcesDeletedWithin(3*time.Minute, manifests, "setup/*.yaml"),
-			)).
-			// Disable our feature flag.
-			WithTeardown("DisableAlphaUsages", funcs.AllOf(
-				funcs.AsFeaturesFunc(environment.HelmUpgradeCrossplaneToBase()),
-				funcs.ReadyToTestWithin(1*time.Minute, namespace),
-			)).
-			Feature(),
-	)
-}
-
-func TestUsageCompositionWithPipeline(t *testing.T) {
-	manifests := "test/e2e/manifests/apiextensions/usage/composition-pipeline"
-
-	usageList := composed.NewList(composed.FromReferenceToList(corev1.ObjectReference{
-		APIVersion: "apiextensions.crossplane.io/v1alpha1",
-		Kind:       "Usage",
-	}))
-
 	environment.Test(t,
 		features.NewWithDescription(t.Name(), "Tests scenarios for Crossplane's `Usage` resource as part of a composition pipeline and decomposed properly.").
 			WithLabel(LabelStage, LabelStageBeta).
@@ -252,11 +170,6 @@ func TestUsageCompositionWithPipeline(t *testing.T) {
 				funcs.DeleteResources(manifests, "setup/*.yaml"),
 				funcs.ResourcesDeletedWithin(3*time.Minute, manifests, "setup/*.yaml"),
 			)).
-			// Disable our feature flag.
-			WithTeardown("DisableAlphaUsages", funcs.AllOf(
-				funcs.AsFeaturesFunc(environment.HelmUpgradeCrossplaneToBase()),
-				funcs.ReadyToTestWithin(1*time.Minute, namespace),
-			)).
 			Feature(),
 	)
 }