[Cherry-Pick] RIA MustInclude annotation and VSC fix for 1.18 (#10101)

* design for RIA must-include-additional-items Design for `restore.velero.io/must-include-additional-items` annotation and its usage and interaction with existing filtering mechanism. Signed-off-by: Adam Zhang <adam.zhang@broadcom.com> (cherry picked from commit ac76402aa0) Signed-off-by: Adam Zhang <adam.zhang@broadcom.com> * Add RIA must-include additional items (#10082) Let RestoreItemActions opt in via annotation to bypass global restore filters for AdditionalItems, mirroring the backup-side must-include behavior. Signed-off-by: Adam Zhang <adam.zhang@broadcom.com> (cherry picked from commit a43a1bce6a) Signed-off-by: Adam Zhang <adam.zhang@broadcom.com> * update v1.18 custom-plugins docs for RIA must-include additional items Signed-off-by: Adam Zhang <adam.zhang@broadcom.com> * remove VolumeSnapshotContents from resourceMustHave list Stop force-including VolumeSnapshotContents via resourceMustHave on every restore; CSI VolumeSnapshot/PVC RestoreItemActions now set `restore.velero.io/must-include-additional-items` so bound snapshot dependencies are restored only when their parent is restored. Fixes: #9957 Signed-off-by: Adam Zhang <adam.zhang@broadcom.com> (cherry picked from commit ef100da89b) Signed-off-by: Adam Zhang <adam.zhang@broadcom.com> * add tests to cover pvc and vsc ria Signed-off-by: Adam Zhang <adam.zhang@broadcom.com> (cherry picked from commit 63cfddd18d) Signed-off-by: Adam Zhang <adam.zhang@broadcom.com> * fix change logs Signed-off-by: Adam Zhang <adam.zhang@broadcom.com> --------- Signed-off-by: Adam Zhang <adam.zhang@broadcom.com>
2026-08-03 05:46:05 +00:00 · 2026-07-29 19:20:52 -04:00
parent ad0f74ee9d
commit 8b761803ff
13 changed files with 1095 additions and 38 deletions
@@ -0,0 +1 @@
+Add restore.velero.io/must-include-additional-items so RestoreItemActions can opt in to bypassing global restore filters for AdditionalItems (mirrors the backup-side must-include annotation; no default behavior change for existing restores/plugins). Stop force-including VolumeSnapshotContents via resourceMustHave on every restore; CSI VolumeSnapshot/PVC RestoreItemActions now set restore.velero.io/must-include-additional-items so bound snapshot dependencies are restored only when their parent is restored (fixes #9957)
@@ -0,0 +1,357 @@
+# RestoreItemAction Must-Include Additional Items
+
+## Abstract
+
+Backup Item Actions (BIAs) can already mark additional items as must-include via `backup.velero.io/must-include-additional-items`, so Velero bypasses resource and namespace exclusion filters when backing those dependencies up.
+This proposal adds the same plugin-controlled escape hatch on restore: `restore.velero.io/must-include-additional-items`, so Restore Item Actions (RIAs) can force-restore declared `AdditionalItems` even when they would otherwise be dropped by global restore filters.
+
+## Glossary & Abbreviation
+
+**Additional Item**: A resource identifier returned by a Backup/Restore Item Action's `Execute()` result that Velero should process as a dependency of the current item.  
+**BIA**: Backup Item Action plugin.  
+**RIA**: Restore Item Action plugin.  
+**Must-Include**: A plugin-set annotation on the action's `UpdatedItem` that tells Velero to bypass global include/exclude filters for that action's `AdditionalItems`.  
+**Global Restore Filter**: `RestoreSpec` filters applied uniformly — `IncludedNamespaces`/`ExcludedNamespaces`, `IncludedResources`/`ExcludedResources`, `IncludeClusterResources`, and label selectors.  
+**Fine-Grained Restore Filter**: Per-namespace / cluster-scoped policies from `RestoreSpec.ResourcePolicy` (`namespacedFilterPolicies`, `clusterScopedFilterPolicy`), as described in [Fine Grained Restore Filters via Resource Policies](https://github.com/velero-io/velero/blob/main/design/restore-filter-enhancement/fine-grained-restore-filters-design.md).  
+**`resourceMustHave`**: A small hardcoded server-side set of resource types that bypass resource and namespace I/E checks inside `restoreItem()` today (but not `IncludeClusterResources=false`).  
+
+## Background
+
+### Backup-side precedent
+
+On backup, a BIA may set `backup.velero.io/must-include-additional-items: "true"` on the returned `UpdatedItem`.
+Velero strips that annotation (it is an internal signal, not intended to land on the live object) and passes `mustInclude=true` into recursive `backupItem` calls for that action's `AdditionalItems`.
+When `mustInclude` is true, `itemInclusionChecks` skips namespace/resource exclusion checks (and related exclusion labels / fine-grained name filters) so plugin-declared dependencies are not dropped by the user's backup filters.
+In-tree CSI BIAs already rely on this for VolumeSnapshot / VolumeSnapshotContent / VolumeSnapshotClass style dependency chains.
+
+### Restore-side gap
+
+On restore, RIAs can return `AdditionalItems`, and Velero recursively calls `restoreItem()` for each of them.
+That path already bypasses fine-grained restore filters and global label selectors, because those are evaluated earlier in `getOrderedResourceCollection` / `getSelectedRestoreableItems`.
+However, `restoreItem()` still enforces global resource includes/excludes, namespace includes/excludes, and `IncludeClusterResources=false`.
+
+The fine-grained restore filters design explicitly documents this remaining floor:
+
+> Note that these additional items must still pass global resource/namespace exclusions.
+
+There is no restore-side equivalent of the BIA must-include annotation.
+Plugins that need a hard dependency restored despite a selective restore configuration have no opt-in way to express that, short of relying on the server-side `resourceMustHave` list (which is global, not plugin-scoped, and does not bypass `IncludeClusterResources=false`).
+
+### Motivating scenario
+
+Consider a selective restore that includes only application namespaces and excludes storage/snapshot resource types, while a plugin knows that restoring a PVC correctly requires a related cluster-scoped or cross-namespace dependency that exists in the backup archive.
+Today the RIA can request that dependency as an `AdditionalItem`, but Velero will skip it at the global exclusion checks inside `restoreItem()`.
+With a restore must-include annotation, the plugin can declare the dependency as required and Velero will restore it (provided the object is present in the backup tarball).
+
+## Goals
+
+- Add `restore.velero.io/must-include-additional-items` with the same parent-annotation contract as the backup-side must-include annotation.
+- When an RIA sets the annotation on `UpdatedItem`, bypass global resource I/E, namespace I/E, and `IncludeClusterResources=false` for that RIA's `AdditionalItems`.
+- Keep the change opt-in and backward compatible: restores and plugins that do not set the annotation behave exactly as today.
+- Document the trust model, precedence rules, and interaction with existing restore gates for plugin authors and operators.
+
+## Non-Goals
+
+- Changing the plugin protobuf / `RestoreItemAction` interface shape (no new RPC fields).
+- Changing CRDs or adding CLI flags.
+- Changing the `resourceMustHave` list (including any narrowing related to VolumeSnapshotContent).
+- Updating in-tree RIAs (CSI or otherwise) to set the new annotation as part of this change.
+- Per-additional-item granularity (the annotation applies blanket to all `AdditionalItems` from that RIA invocation, matching BIA).
+- Materializing items that were never backed up.
+
+## High-Level Design
+
+Mirror the backup workflow:
+
+1. Introduce annotation constant `restore.velero.io/must-include-additional-items`.
+2. After each RIA `Execute()`, if `UpdatedItem` carries the annotation with value `"true"`, strip it and set `mustIncludeAdditionalItems=true`.
+3. Pass that boolean into recursive `restoreItem(..., mustInclude)` calls for the action's `AdditionalItems`.
+4. When `mustInclude` is true, skip the global resource/namespace/`IncludeClusterResources` exclusion checks inside `restoreItem()`.
+5. Keep all non-filter gates unchanged (tarball presence, already-restored, completed Jobs, API errors, wait-for-additional-items, etc.).
+
+Top-level items from the archive continue to be restored with `mustInclude=false`, so user filters still apply to the primary restore set.
+
+```mermaid
+flowchart TD
+    startRestore[Start Restore] --> readTarball[Read Item from Backup Tarball]
+    readTarball --> topLevelRestoreItem["restoreItem(..., mustInclude=false)"]
+
+    topLevelRestoreItem --> checkMustInclude{"mustInclude == true?"}
+
+    checkMustInclude -- No --> checkFilters{"Pass Global Resource/Namespace Filters?"}
+    checkFilters -- No --> skipItem[Skip Restore]
+    checkFilters -- Yes --> nonFilterGates["Other gates: isCompleted, already-restored, ..."]
+
+    checkMustInclude -- Yes --> nonFilterGates
+
+    nonFilterGates --> executeRIA[Execute RestoreItemAction]
+
+    executeRIA --> checkSkip{"SkipRestore?"}
+    checkSkip -- Yes --> skipItem
+    checkSkip -- No --> checkAnnotation{"Has must-include annotation?"}
+
+    checkAnnotation -- Yes --> stripAnnotation[Strip Annotation]
+    stripAnnotation --> setFlagTrue["mustIncludeAdditionalItems = true"]
+
+    checkAnnotation -- No --> setFlagFalse["mustIncludeAdditionalItems = false"]
+
+    setFlagTrue --> loopAdditionalItems[Loop over AdditionalItems]
+    setFlagFalse --> loopAdditionalItems
+
+    loopAdditionalItems --> existsInBackup{"Item file in tarball?"}
+    existsInBackup -- No --> warnSkip[Warn and skip]
+    existsInBackup -- Yes --> recursiveRestoreItem["restoreItem(..., mustInclude=mustIncludeAdditionalItems)"]
+    recursiveRestoreItem --> checkMustInclude
+```
+
+> The edge `recursiveRestoreItem --> checkMustInclude` is a recursive call (new `restoreItem` stack frame), not a same-frame loop.
+
+## Detailed Design
+
+### Annotation constant
+
+In `pkg/apis/velero/v1/labels_annotations.go`, next to the existing backup constant:
+
+```go
+// Velero checks this annotation to determine whether to skip resource excluding check.
+MustIncludeAdditionalItemAnnotation = "backup.velero.io/must-include-additional-items"
+
+// MustIncludeAdditionalItemRestoreAnnotation is set by RestoreItemActions on the UpdatedItem
+// to tell Velero to bypass global resource/namespace exclusion checks (and IncludeClusterResources=false)
+// for that action's AdditionalItems. Value must be "true". The annotation is stripped before
+// the item is applied to the cluster.
+//
+// Notice: SkipRestore on the Execute output takes precedence. If SkipRestore is true, the
+// annotation is never inspected and AdditionalItems are not processed.
+MustIncludeAdditionalItemRestoreAnnotation = "restore.velero.io/must-include-additional-items"
+```
+
+Only the string value `"true"` enables the bypass (same as backup).
+
+### `restoreItem` signature
+
+```go
+func (ctx *restoreContext) restoreItem(
+	obj *unstructured.Unstructured,
+	groupResource schema.GroupResource,
+	namespace string,
+	mustInclude bool,
+) (results.Result, results.Result, bool)
+```
+
+Call sites:
+
+| Site | `mustInclude` value |
+|---|---|
+| Top-level restore loop | `false` |
+| Recursive additional-item restore after an RIA | derived from that RIA's `UpdatedItem` annotation |
+
+### Bypass exclusion checks; keep namespace creation
+
+Today, namespace exclusion and `EnsureNamespaceExistsAndIsReady` share one `if namespace != ""` block in `restoreItem()`.
+If must-include only skipped the exclusion check without refactoring, an additional item targeting an excluded namespace would fail because its target namespace was never ensured.
+
+Required structure:
+
+```go
+if mustInclude {
+	restoreLogger.Info("Skipping the resource/namespace exclusion checks because the item is marked as must-include")
+} else {
+	if !ctx.resourceIncludesExcludes.ShouldInclude(groupResource.String()) && !ctx.resourceMustHave.Has(groupResource.String()) {
+		restoreLogger.Info("Not restoring item because resource is excluded")
+		return warnings, errs, itemExists
+	}
+
+	if namespace != "" {
+		if !ctx.namespaceIncludesExcludes.ShouldInclude(obj.GetNamespace()) && !ctx.resourceMustHave.Has(groupResource.String()) {
+			restoreLogger.Info("Not restoring item because namespace is excluded")
+			return warnings, errs, itemExists
+		}
+	} else {
+		if boolptr.IsSetToFalse(ctx.restore.Spec.IncludeClusterResources) {
+			restoreLogger.Info("Not restoring item because it's cluster-scoped")
+			return warnings, errs, itemExists
+		}
+	}
+}
+
+// Namespace creation runs regardless of mustInclude.
+if namespace != "" {
+	nsToEnsure := getNamespace(restoreLogger, archive.GetItemFilePath(ctx.restoreDir, "namespaces", "", obj.GetNamespace()), namespace)
+	_, nsCreated, err := kube.EnsureNamespaceExistsAndIsReady(nsToEnsure, ctx.namespaceClient, ctx.resourceTerminatingTimeout, ctx.resourceDeletionStatusTracker)
+	// ... existing error handling and restoredItems bookkeeping ...
+}
+```
+
+Namespace remapping is unchanged: exclusion checks use the original namespace (`obj.GetNamespace()`); namespace creation uses the remapped target `namespace` parameter.
+
+### Process the annotation after each RIA
+
+Inside the applicable-actions loop in `restoreItem()`, after `SkipRestore` handling and type-asserting `UpdatedItem`:
+
+```go
+obj = unstructuredObj
+
+mustIncludeAdditionalItems := false
+if annotations := obj.GetAnnotations(); annotations != nil &&
+	annotations[velerov1api.MustIncludeAdditionalItemRestoreAnnotation] == "true" {
+	mustIncludeAdditionalItems = true
+	restoreLogger.Info("RestoreItemAction marked additional items as must-include; bypassing resource/namespace exclusion checks for them")
+	delete(annotations, velerov1api.MustIncludeAdditionalItemRestoreAnnotation)
+	obj.SetAnnotations(annotations)
+}
+
+for _, additionalItem := range executeOutput.AdditionalItems {
+	// existing tarball stat / unmarshal / namespace mapping ...
+	w, e, additionalItemExists := ctx.restoreItem(
+		additionalObj,
+		additionalItem.GroupResource,
+		additionalItemNamespace,
+		mustIncludeAdditionalItems,
+	)
+	// existing merge / filteredAdditionalItems bookkeeping ...
+}
+```
+
+### Filter bypass matrix
+
+| Gate | Plain AdditionalItem | `resourceMustHave` | RIA `mustInclude=true` | BIA `mustInclude=true` (parity target) |
+|---|---|---|---|---|
+| Fine-grained policies (kind/name/label) | Bypass (never enter selection Phase B filters) | N/A in `restoreItem` | Bypass (same) | Bypass |
+| Global label selectors | Bypass (never re-enter selection) | N/A in `restoreItem` | Bypass (same) | Bypass |
+| Global resource I/E | Honored | Bypass | Bypass | Bypass |
+| Global namespace I/E | Honored | Bypass | Bypass | Bypass |
+| `IncludeClusterResources=false` | Honored | Honored (not bypassed) | Bypass | Bypass |
+| Item must exist in backup tarball | Required | Required | Required | N/A (fetched from cluster) |
+| `isCompleted` / already-restored / API errors | Still apply | Still apply | Still apply | `DeletionTimestamp` still applies on backup |
+
+RIA must-include is intentionally a **stronger** override than `resourceMustHave` because it also bypasses `IncludeClusterResources=false`.
+That matches BIA must-include semantics (plugin-trusted hard dependencies), rather than widening the hardcoded server list.
+
+### Interaction with fine-grained restore filters
+
+Per [Fine Grained Restore Filters via Resource Policies](../restore-filter-enhancement/fine-grained-restore-filters-design.md), plugin additional items already bypass `namespacedFilterPolicies` / `clusterScopedFilterPolicy` kind, name, and label checks.
+Those filters live in the selection phases; additional items enter `restoreItem()` directly.
+
+This proposal only changes the remaining global gates inside `restoreItem()`.
+With must-include set, an additional item effectively bypasses **all** restore filters (fine-grained and global).
+Without the annotation, behavior is unchanged: fine-grained filters are still bypassed, global exclusions still apply.
+
+### Interaction with existing restore gates
+
+#### `SkipRestore` precedence
+
+If `Execute()` returns `SkipRestore: true`, `restoreItem()` returns before inspecting the annotation, and no `AdditionalItems` are processed.
+This mirrors backup-side precedence where `velero.io/skip-from-backup` outranks must-include.
+
+#### Multi-RIA semantics
+
+Annotation handling is per RIA invocation inside the actions loop:
+
+1. RIA N executes → inspect/strip annotation on that `UpdatedItem` → restore that RIA's `AdditionalItems` with the derived flag.
+2. RIA N+1 sees the already-stripped object unless it sets the annotation again.
+
+A later RIA does not inherit an earlier RIA's must-include decision.
+
+#### Transitive propagation
+
+The parent's `mustInclude` flag admits the child additional item through filters.
+It does **not** automatically force-include grandchildren.
+Each RIA level that needs the escape hatch must set the annotation on its own `UpdatedItem`, matching BIA behavior.
+
+#### Non-filter gates that still apply
+
+Even when `mustInclude=true`:
+
+- Missing archive file → warn and skip (existing behavior).
+- `isCompleted` resources (e.g. completed Jobs) → skip.
+- Already present in `ctx.restoredItems` → skip.
+- Create/update API failures → errors as today.
+- `WaitForAdditionalItems` / `AreAdditionalItemsReady` polling after the additional-item loop → unchanged.
+
+### Relationship to `resourceMustHave`
+
+| Mechanism | Who decides | Bypasses resource/ns I/E | Bypasses `IncludeClusterResources=false` |
+|---|---|---|---|
+| `resourceMustHave` | Velero server (hardcoded) | Yes | No |
+| RIA must-include | Plugin author (annotation) | Yes | Yes |
+
+The two mechanisms coexist.
+This proposal does not migrate in-tree CSI (or other) RIAs onto the annotation.
+Doing so would be a separate behavior change: it could force-restore types users explicitly excluded, and would newly restore cluster-scoped dependencies even when `IncludeClusterResources=false`.
+
+### Plugin usage sketch
+
+```go
+func (p *myRestoreAction) Execute(input *velero.RestoreItemActionExecuteInput) (*velero.RestoreItemActionExecuteOutput, error) {
+	item := input.Item.(*unstructured.Unstructured)
+	annotations := item.GetAnnotations()
+	if annotations == nil {
+		annotations = map[string]string{}
+	}
+	annotations[velerov1api.MustIncludeAdditionalItemRestoreAnnotation] = "true"
+	item.SetAnnotations(annotations)
+
+	return &velero.RestoreItemActionExecuteOutput{
+		UpdatedItem: item,
+		AdditionalItems: []velero.ResourceIdentifier{
+			{GroupResource: schema.GroupResource{Group: "example.io", Resource: "dependencies"}, Namespace: "dep-ns", Name: "dep-1"},
+		},
+	}, nil
+}
+```
+
+Plugin authors must ensure the additional item was actually captured in the backup (typically via the corresponding BIA also using `backup.velero.io/must-include-additional-items`).
+
+### Tests
+
+Extend restore coverage (existing `TestRestoreActionAdditionalItems` patterns / focused cases) for:
+
+1. Resource exclusion bypass with annotation; still skipped without annotation.
+2. Namespace exclusion bypass **and** target namespace creation.
+3. `IncludeClusterResources=false` bypass for cluster-scoped additional items.
+4. Annotation stripped from the object applied to the cluster.
+5. `SkipRestore: true` prevents additional-item processing even if the annotation is set.
+6. Missing tarball entry still warns and skips.
+7. Transitive case: child RIA must re-set the annotation for grandchildren.
+8. Top-level restore path still passes `mustInclude=false` and honors filters.
+
+### Documentation
+
+- Constant doc comment (including `SkipRestore` precedence).
+- Plugin-author docs for Restore Item Actions: annotation key/value, blanket scope, filter-bypass matrix, namespace-creation side effect, tarball requirement.
+
+## Security Considerations
+
+Installing an RIA that sets this annotation grants that plugin authority to restore dependencies outside the operator's restore filters, including:
+
+- resources in namespaces the restore excluded (and creation of those target namespaces if needed);
+- resource types the restore excluded;
+- cluster-scoped resources even when `IncludeClusterResources=false`.
+
+This matches the existing BIA trust model: item-action plugins are already privileged components of the Velero deployment.
+Operators should treat RIA installation as a trust decision.
+The annotation is stripped before apply so it does not persist as attacker-controlled cluster state from the backup archive alone; a matching RIA must run and return `AdditionalItems` for the bypass to take effect.
+
+## Compatibility
+
+- No CRD or plugin interface changes.
+- Existing restores unchanged when no RIA sets the annotation.
+- Existing tests that assert additional items are dropped under namespace filters / `IncludeClusterResources=false` remain valid for the no-annotation path.
+- Compatible with fine-grained restore filters: additional items already bypass those filters; this proposal only addresses the documented global-exclusion floor.
+
+## Alternatives Considered
+
+### Per-item must-include on each `ResourceIdentifier`
+
+Pros: selective control within one `AdditionalItems` list.
+Cons: requires API changes to `ResourceIdentifier` or a parallel structure; diverges from BIA; plugins that need selectivity can already split across actions or omit non-required items.
+
+Rejected for this proposal; may be revisited later if plugin authors demonstrate a concrete need.
+
+### Widen `resourceMustHave` instead of a plugin annotation
+
+Pros: no plugin contract change.
+Cons: server-forced, global, not scoped to a plugin call; does not give third-party plugins a general tool; does not match BIA; conflicts with efforts to keep hardcoded force-include lists narrow.
+
+Rejected — wrong trust model for a general plugin escape hatch.
@@ -166,6 +166,14 @@ const (

 	// Velero checks this annotation to determine whether to skip resource excluding check.
 	MustIncludeAdditionalItemAnnotation = "backup.velero.io/must-include-additional-items"
+	// MustIncludeAdditionalItemRestoreAnnotation is set by RestoreItemActions on the UpdatedItem
+	// to tell Velero to bypass global resource/namespace exclusion checks (and IncludeClusterResources=false)
+	// for that action's AdditionalItems. Value must be "true" to enable the bypass. The annotation is
+	// always stripped before the item is applied to the cluster when present, including non-"true" values.
+	//
+	// Notice: SkipRestore on the Execute output takes precedence. If SkipRestore is true, the
+	// annotation is never inspected and AdditionalItems are not processed.
+	MustIncludeAdditionalItemRestoreAnnotation = "restore.velero.io/must-include-additional-items"
 	// SkippedNoCSIPVAnnotation - Velero checks this annotation on processed PVC to
 	// find out if the snapshot was skipped b/c the PV is not provisioned via CSI
 	SkippedNoCSIPVAnnotation = "backup.velero.io/skipped-no-csi-pv"
@@ -176,6 +176,15 @@ func (p *pvcRestoreItemAction) Execute(
 				Name:          vsName,
 				Namespace:     pvc.Namespace,
 			})
+
+			// Force-restore the VolumeSnapshot even when restore resource filters
+			// would otherwise exclude it (mirrors backup-side must-include).
+			annotations := pvc.GetAnnotations()
+			if annotations == nil {
+				annotations = map[string]string{}
+			}
+			annotations[velerov1api.MustIncludeAdditionalItemRestoreAnnotation] = "true"
+			pvc.SetAnnotations(annotations)
 		}
 	}

@@ -371,6 +371,7 @@ func TestExecute(t *testing.T) {
 		backup               *velerov1api.Backup
 		restore              *velerov1api.Restore
 		pvc                  *corev1api.PersistentVolumeClaim
+		pvcFromBackup        *corev1api.PersistentVolumeClaim
 		vs                   *snapshotv1api.VolumeSnapshot
 		dataUploadResult     *corev1api.ConfigMap
 		expectedErr          string
@@ -402,15 +403,40 @@ func TestExecute(t *testing.T) {
 			vs: builder.ForVolumeSnapshot("velero", vsName).ObjectMeta(
 				builder.WithAnnotations(velerov1api.VolumeSnapshotRestoreSize, "10Gi"),
 			).Result(),
-			expectedPVC: builder.ForPersistentVolumeClaim("velero", "testPVC").ObjectMeta(builder.WithAnnotations(velerov1api.VolumeSnapshotLabel, "vsName")).Result(),
+			expectedPVC: builder.ForPersistentVolumeClaim("velero", "testPVC").ObjectMeta(builder.WithAnnotations(
+				velerov1api.VolumeSnapshotLabel, "vsName",
+				velerov1api.MustIncludeAdditionalItemRestoreAnnotation, "true",
+			)).Result(),
 		},
 		{
-			name:        "Restore from VolumeSnapshot without volume-snapshot-name annotation",
-			backup:      builder.ForBackup("velero", "testBackup").Result(),
-			restore:     builder.ForRestore("velero", "testRestore").Backup("testBackup").Result(),
-			pvc:         builder.ForPersistentVolumeClaim("velero", "testPVC").ObjectMeta(builder.WithAnnotations(AnnSelectedNode, "node1")).Result(),
-			vs:          builder.ForVolumeSnapshot("velero", "testVS").ObjectMeta(builder.WithAnnotations(velerov1api.VolumeSnapshotRestoreSize, "10Gi")).Result(),
-			expectedPVC: builder.ForPersistentVolumeClaim("velero", "testPVC").ObjectMeta(builder.WithAnnotations(AnnSelectedNode, "node1")).Result(),
+			name:    "Restore from VolumeSnapshot with nil PVC annotations",
+			backup:  builder.ForBackup("velero", "testBackup").Result(),
+			restore: builder.ForRestore("velero", "testRestore").ObjectMeta(builder.WithUID("restoreUID")).Backup("testBackup").Result(),
+			pvc: &corev1api.PersistentVolumeClaim{
+				ObjectMeta: metav1.ObjectMeta{
+					Name:      "testPVC",
+					Namespace: "velero",
+				},
+			},
+			pvcFromBackup: builder.ForPersistentVolumeClaim("velero", "testPVC").ObjectMeta(builder.WithAnnotations(velerov1api.VolumeSnapshotLabel, "vsName")).Result(),
+			vs: builder.ForVolumeSnapshot("velero", vsName).ObjectMeta(
+				builder.WithAnnotations(velerov1api.VolumeSnapshotRestoreSize, "10Gi"),
+			).Result(),
+			expectedPVC: builder.ForPersistentVolumeClaim("velero", "testPVC").ObjectMeta(builder.WithAnnotations(
+				velerov1api.MustIncludeAdditionalItemRestoreAnnotation, "true",
+			)).Result(),
+		},
+		{
+			name:    "Restore from VolumeSnapshot without volume-snapshot-name annotation",
+			backup:  builder.ForBackup("velero", "testBackup").Result(),
+			restore: builder.ForRestore("velero", "testRestore").Backup("testBackup").Result(),
+			pvc:     builder.ForPersistentVolumeClaim("velero", "testPVC").ObjectMeta(builder.WithAnnotations(velerov1api.VolumeSnapshotLabel, "vsName", AnnSelectedNode, "node1")).Result(),
+			vs:      builder.ForVolumeSnapshot("velero", "testVS").ObjectMeta(builder.WithAnnotations(velerov1api.VolumeSnapshotRestoreSize, "10Gi")).Result(),
+			expectedPVC: builder.ForPersistentVolumeClaim("velero", "testPVC").ObjectMeta(builder.WithAnnotations(
+				velerov1api.VolumeSnapshotLabel, "vsName",
+				AnnSelectedNode, "node1",
+				velerov1api.MustIncludeAdditionalItemRestoreAnnotation, "true",
+			)).Result(),
 		},
 		{
 			name:        "DataUploadResult cannot be found",
@@ -480,7 +506,13 @@ func TestExecute(t *testing.T) {
 				require.NoError(t, err)

 				input.Item = &unstructured.Unstructured{Object: pvcMap}
-				input.ItemFromBackup = &unstructured.Unstructured{Object: pvcMap}
+				if tc.pvcFromBackup != nil {
+					pvcFromBackupMap, err := runtime.DefaultUnstructuredConverter.ToUnstructured(tc.pvcFromBackup)
+					require.NoError(t, err)
+					input.ItemFromBackup = &unstructured.Unstructured{Object: pvcFromBackupMap}
+				} else {
+					input.ItemFromBackup = &unstructured.Unstructured{Object: pvcMap}
+				}
 				input.Restore = tc.restore
 			}
 			if tc.preCreatePVC {
@@ -508,6 +540,12 @@ func TestExecute(t *testing.T) {
 				err := runtime.DefaultUnstructuredConverter.FromUnstructured(output.UpdatedItem.UnstructuredContent(), pvc)
 				require.NoError(t, err)
 				require.Equal(t, tc.expectedPVC.GetObjectMeta(), pvc.GetObjectMeta())
+				if tc.name == "Restore from VolumeSnapshot" {
+					require.Equal(t, "true", pvc.GetAnnotations()[velerov1api.MustIncludeAdditionalItemRestoreAnnotation])
+					require.Len(t, output.AdditionalItems, 1)
+					require.Equal(t, "volumesnapshots.snapshot.storage.k8s.io", output.AdditionalItems[0].GroupResource.String())
+					require.Equal(t, "vsName", output.AdditionalItems[0].Name)
+				}
 				if pvc.Spec.Selector != nil && pvc.Spec.Selector.MatchLabels != nil {
 					// This is used for long name and namespace case.
 					if len(tc.pvc.Namespace+"."+tc.pvc.Name) >= validation.DNS1035LabelMaxLength {
@@ -66,6 +66,9 @@ func resetVolumeSnapshotSpecForRestore(vs *snapshotv1api.VolumeSnapshot, vscName
 }

 func resetVolumeSnapshotAnnotation(vs *snapshotv1api.VolumeSnapshot) {
+	if vs.ObjectMeta.Annotations == nil {
+		vs.ObjectMeta.Annotations = make(map[string]string)
+	}
 	vs.ObjectMeta.Annotations[velerov1api.VSCDeletionPolicyAnnotation] =
 		string(snapshotv1api.VolumeSnapshotContentRetain)
 }
@@ -282,12 +285,6 @@ func (p *volumeSnapshotRestoreItemAction) Execute(
 			vs.Namespace, vs.Name)
 	}

-	vsMap, err := runtime.DefaultUnstructuredConverter.ToUnstructured(&vs)
-	if err != nil {
-		p.log.Errorf("Fail to convert VS %s to unstructured", vs.Namespace+"/"+vs.Name)
-		return nil, errors.WithStack(err)
-	}
-
 	if vsFromBackup.Status == nil ||
 		vsFromBackup.Status.BoundVolumeSnapshotContentName == nil {
 		p.log.Errorf("VS %s doesn't have bound VSC", vsFromBackup.Name)
@@ -299,6 +296,21 @@ func (p *volumeSnapshotRestoreItemAction) Execute(
 		Name:          *vsFromBackup.Status.BoundVolumeSnapshotContentName,
 	}

+	// Force-restore the bound VSC even when restore resource filters would
+	// otherwise exclude it (mirrors backup-side must-include for CSI deps).
+	annotations := vs.GetAnnotations()
+	if annotations == nil {
+		annotations = map[string]string{}
+	}
+	annotations[velerov1api.MustIncludeAdditionalItemRestoreAnnotation] = "true"
+	vs.SetAnnotations(annotations)
+
+	vsMap, err := runtime.DefaultUnstructuredConverter.ToUnstructured(&vs)
+	if err != nil {
+		p.log.Errorf("Fail to convert VS %s to unstructured", vs.Namespace+"/"+vs.Name)
+		return nil, errors.WithStack(err)
+	}
+
 	p.log.Infof(`Returning from VolumeSnapshotRestoreItemAction with 
 		VolumeSnapshotContent in additionalItems`)

@@ -103,6 +103,26 @@ func TestResetVolumeSnapshotSpecForRestore(t *testing.T) {
 	}
 }

+func TestResetVolumeSnapshotAnnotation(t *testing.T) {
+	t.Run("should set deletion policy annotation when annotations is nil", func(t *testing.T) {
+		vs := snapshotv1api.VolumeSnapshot{}
+		resetVolumeSnapshotAnnotation(&vs)
+		assert.NotNil(t, vs.ObjectMeta.Annotations)
+		assert.Equal(t, string(snapshotv1api.VolumeSnapshotContentRetain), vs.ObjectMeta.Annotations[velerov1api.VSCDeletionPolicyAnnotation])
+	})
+
+	t.Run("should preserve existing annotations and set deletion policy annotation", func(t *testing.T) {
+		vs := snapshotv1api.VolumeSnapshot{
+			ObjectMeta: metav1.ObjectMeta{
+				Annotations: map[string]string{"foo": "bar"},
+			},
+		}
+		resetVolumeSnapshotAnnotation(&vs)
+		assert.Equal(t, "bar", vs.ObjectMeta.Annotations["foo"])
+		assert.Equal(t, string(snapshotv1api.VolumeSnapshotContentRetain), vs.ObjectMeta.Annotations[velerov1api.VSCDeletionPolicyAnnotation])
+	})
+}
+
 func TestVSExecute(t *testing.T) {
 	newVscName := util.GenerateSha256FromRestoreUIDAndVsName("restoreUID", "vsName")
 	tests := []struct {
@@ -145,6 +165,18 @@ func TestVSExecute(t *testing.T) {
 			expectErr:  false,
 			expectedVS: builder.ForVolumeSnapshot("ns", "test").SourceVolumeSnapshotContentName(newVscName).Result(),
 		},
+		{
+			name: "Normal case with nil VS annotations, VSC should be created",
+			vs: builder.ForVolumeSnapshot("ns", "vsName").
+				SourceVolumeSnapshotContentName(newVscName).
+				VolumeSnapshotClass("vscClass").
+				Status().
+				BoundVolumeSnapshotContentName("vscName").
+				Result(),
+			restore:    builder.ForRestore("velero", "restore").ObjectMeta(builder.WithUID("restoreUID")).Result(),
+			expectErr:  false,
+			expectedVS: builder.ForVolumeSnapshot("ns", "test").SourceVolumeSnapshotContentName(newVscName).Result(),
+		},
 	}

 	for _, test := range tests {
@@ -184,6 +216,10 @@ func TestVSExecute(t *testing.T) {
 				require.NoError(t, runtime.DefaultUnstructuredConverter.FromUnstructured(
 					result.UpdatedItem.UnstructuredContent(), &vs))
 				require.Equal(t, test.expectedVS.Spec, vs.Spec)
+				require.Equal(t, "true", vs.GetAnnotations()[velerov1api.MustIncludeAdditionalItemRestoreAnnotation])
+				require.Len(t, result.AdditionalItems, 1)
+				require.Equal(t, "volumesnapshotcontents.snapshot.storage.k8s.io", result.AdditionalItems[0].GroupResource.String())
+				require.Equal(t, "vscName", result.AdditionalItems[0].Name)
 			}
 		})
 	}
@@ -87,7 +87,6 @@ const ObjectStatusRestoreAnnotationKey = "velero.io/restore-status"

 var resourceMustHave = []string{
 	"datauploads.velero.io",
-	"volumesnapshotcontents.snapshot.storage.k8s.io",
 }

 type VolumeSnapshotterGetter interface {
@@ -1060,7 +1059,7 @@ func (ctx *restoreContext) processSelectedResource(
 				continue
 			}

-			w, e, _ := ctx.restoreItem(obj, groupResource, targetNS)
+			w, e, _ := ctx.restoreItem(obj, groupResource, targetNS, false)
 			warnings.Merge(&w)
 			errs.Merge(&e)
 			processedItems++
@@ -1386,7 +1385,7 @@ func (ctx *restoreContext) getResource(groupResource schema.GroupResource, obj *
 	return u, nil
 }

-func (ctx *restoreContext) restoreItem(obj *unstructured.Unstructured, groupResource schema.GroupResource, namespace string) (results.Result, results.Result, bool) {
+func (ctx *restoreContext) restoreItem(obj *unstructured.Unstructured, groupResource schema.GroupResource, namespace string, mustInclude bool) (results.Result, results.Result, bool) {
 	warnings, errs := results.Result{}, results.Result{}
 	// itemExists bool is used to determine whether to include this item in the "wait for additional items" list
 	itemExists := false
@@ -1403,27 +1402,41 @@ func (ctx *restoreContext) restoreItem(obj *unstructured.Unstructured, groupReso
 	// Check if group/resource should be restored. We need to do this here since
 	// this method may be getting called for an additional item which is a group/resource
 	// that's excluded.
-	if !ctx.resourceIncludesExcludes.ShouldInclude(groupResource.String()) && !ctx.resourceMustHave.Has(groupResource.String()) {
-		restoreLogger.Info("Not restoring item because resource is excluded")
-		return warnings, errs, itemExists
-	}
-
-	// Check if namespace/cluster-scoped resource should be restored. We need
-	// to do this here since this method may be getting called for an additional
-	// item which is in a namespace that's excluded, or which is cluster-scoped
-	// and should be excluded. Note that we're checking the object's namespace (
-	// via obj.GetNamespace()) instead of the namespace parameter, because we want
-	// to check the *original* namespace, not the remapped one if it's been remapped.
 	//
 	// Note: Additional items intentionally bypass fine-grained resource filter policies
 	// (like per-namespace label/name selectors) to avoid breaking semantic dependencies,
-	// but they must still pass the global exclusions enforced below.
-	if namespace != "" {
-		if !ctx.namespaceIncludesExcludes.ShouldInclude(obj.GetNamespace()) && !ctx.resourceMustHave.Has(groupResource.String()) {
-			restoreLogger.Info("Not restoring item because namespace is excluded")
+	// but they must still pass the global exclusions enforced below unless mustInclude is set.
+	if mustInclude {
+		restoreLogger.Info("Skipping the resource/namespace exclusion checks because the item is marked as must-include")
+	} else {
+		if !ctx.resourceIncludesExcludes.ShouldInclude(groupResource.String()) && !ctx.resourceMustHave.Has(groupResource.String()) {
+			restoreLogger.Info("Not restoring item because resource is excluded")
 			return warnings, errs, itemExists
 		}

+		// Check if namespace/cluster-scoped resource should be restored. We need
+		// to do this here since this method may be getting called for an additional
+		// item which is in a namespace that's excluded, or which is cluster-scoped
+		// and should be excluded. Note that we're checking the object's namespace (
+		// via obj.GetNamespace()) instead of the namespace parameter, because we want
+		// to check the *original* namespace, not the remapped one if it's been remapped.
+		if namespace != "" {
+			if !ctx.namespaceIncludesExcludes.ShouldInclude(obj.GetNamespace()) && !ctx.resourceMustHave.Has(groupResource.String()) {
+				restoreLogger.Info("Not restoring item because namespace is excluded")
+				return warnings, errs, itemExists
+			}
+		} else {
+			if boolptr.IsSetToFalse(ctx.restore.Spec.IncludeClusterResources) {
+				restoreLogger.Info("Not restoring item because it's cluster-scoped")
+				return warnings, errs, itemExists
+			}
+		}
+	}
+
+	// Namespace creation runs unconditionally when namespace != "", regardless of
+	// mustInclude. This ensures target namespaces exist for additional items that
+	// bypass the namespace-exclusion check above.
+	if namespace != "" {
 		// If the namespace scoped resource should be restored, ensure that the
 		// namespace into which the resource is being restored into exists.
 		// This is the *remapped* namespace that we are ensuring exists.
@@ -1442,11 +1455,6 @@ func (ctx *restoreContext) restoreItem(obj *unstructured.Unstructured, groupReso
 			}
 			ctx.restoredItems[itemKey] = restoredItemStatus{action: ItemRestoreResultCreated, itemExists: true, createdName: nsToEnsure.Name}
 		}
-	} else {
-		if boolptr.IsSetToFalse(ctx.restore.Spec.IncludeClusterResources) {
-			restoreLogger.Info("Not restoring item because it's cluster-scoped")
-			return warnings, errs, itemExists
-		}
 	}

 	// Make a copy of object retrieved from backup to make it available unchanged
@@ -1668,6 +1676,21 @@ func (ctx *restoreContext) restoreItem(obj *unstructured.Unstructured, groupReso

 		obj = unstructuredObj

+		mustIncludeAdditionalItems := false
+		if annotations := obj.GetAnnotations(); annotations != nil {
+			if _, present := annotations[velerov1api.MustIncludeAdditionalItemRestoreAnnotation]; present {
+				// Only the string value "true" enables the bypass.
+				if annotations[velerov1api.MustIncludeAdditionalItemRestoreAnnotation] == "true" {
+					mustIncludeAdditionalItems = true
+					restoreLogger.Info("RestoreItemAction marked additional items as must-include; bypassing resource/namespace exclusion checks for them")
+				}
+				// Always strip the annotation so it never lands on the cluster,
+				// regardless of whether the value enabled the bypass.
+				delete(annotations, velerov1api.MustIncludeAdditionalItemRestoreAnnotation)
+				obj.SetAnnotations(annotations)
+			}
+		}
+
 		var filteredAdditionalItems []velero.ResourceIdentifier
 		for _, additionalItem := range executeOutput.AdditionalItems {
 			itemPath := archive.GetItemFilePath(ctx.restoreDir, additionalItem.GroupResource.String(), additionalItem.Namespace, additionalItem.Name)
@@ -1687,6 +1710,7 @@ func (ctx *restoreContext) restoreItem(obj *unstructured.Unstructured, groupReso
 			additionalObj, err := archive.Unmarshal(ctx.fileSystem, itemPath)
 			if err != nil {
 				errs.Add(namespace, errors.Wrapf(err, "error restoring additional item %s", additionalResourceID))
+				continue
 			}

 			additionalItemNamespace := additionalItem.Namespace
@@ -1696,7 +1720,7 @@ func (ctx *restoreContext) restoreItem(obj *unstructured.Unstructured, groupReso
 				}
 			}

-			w, e, additionalItemExists := ctx.restoreItem(additionalObj, additionalItem.GroupResource, additionalItemNamespace)
+			w, e, additionalItemExists := ctx.restoreItem(additionalObj, additionalItem.GroupResource, additionalItemNamespace, mustIncludeAdditionalItems)
 			if additionalItemExists {
 				filteredAdditionalItems = append(filteredAdditionalItems, additionalItem)
 			}
@@ -754,6 +754,29 @@ func TestRestoreResourceFiltering(t *testing.T) {
 			apiResources: []*test.APIResource{test.ServiceAccounts()},
 			want:         map[*test.APIResource][]string{test.ServiceAccounts(): {"ns-1/sa-1"}},
 		},
+		{
+			// Regression for #9957: VSC must not be force-included via resourceMustHave
+			// when the restore only selects unrelated resource types.
+			name:    "volumesnapshotcontents are not force-included for selective resource restores",
+			restore: defaultRestore().IncludedResources("storageclasses").IncludeClusterResources(true).Result(),
+			backup:  defaultBackup().Result(),
+			tarball: test.NewTarWriter(t).
+				AddItems("storageclasses.storage.k8s.io",
+					builder.ForStorageClass("sc-1").Result(),
+				).
+				AddItems("volumesnapshotcontents.snapshot.storage.k8s.io",
+					builder.ForVolumeSnapshotContent("vsc-1").Result(),
+				).
+				Done(),
+			apiResources: []*test.APIResource{
+				test.StorageClasses(),
+				test.VolumeSnapshotContents(),
+			},
+			want: map[*test.APIResource][]string{
+				test.StorageClasses():         {"/sc-1"},
+				test.VolumeSnapshotContents(): nil,
+			},
+		},
 	}

 	for _, tc := range tests {
@@ -2150,6 +2173,102 @@ func TestRestoreActionAdditionalItems(t *testing.T) {
 				test.PVs():  nil,
 			},
 		},
+		{
+			name:    "must-include annotation bypasses resource exclusion for additional items",
+			restore: defaultRestore().IncludedResources("pods").Result(),
+			backup:  defaultBackup().Result(),
+			tarball: test.NewTarWriter(t).
+				AddItems("pods", builder.ForPod("ns-1", "pod-1").Result()).
+				AddItems("persistentvolumes", builder.ForPersistentVolume("pv-1").Result()).
+				Done(),
+			apiResources: []*test.APIResource{test.Pods(), test.PVs()},
+			actions: []riav2.RestoreItemAction{
+				&pluggableAction{
+					executeFunc: func(input *velero.RestoreItemActionExecuteInput) (*velero.RestoreItemActionExecuteOutput, error) {
+						item := input.Item.(*unstructured.Unstructured)
+						annotations := item.GetAnnotations()
+						if annotations == nil {
+							annotations = map[string]string{}
+						}
+						annotations[velerov1api.MustIncludeAdditionalItemRestoreAnnotation] = "true"
+						item.SetAnnotations(annotations)
+						return &velero.RestoreItemActionExecuteOutput{
+							UpdatedItem: item,
+							AdditionalItems: []velero.ResourceIdentifier{
+								{GroupResource: kuberesource.PersistentVolumes, Name: "pv-1"},
+							},
+						}, nil
+					},
+				},
+			},
+			want: map[*test.APIResource][]string{
+				test.Pods(): {"ns-1/pod-1"},
+				test.PVs():  {"/pv-1"},
+			},
+		},
+		{
+			name:         "must-include annotation bypasses namespace exclusion for additional items",
+			restore:      defaultRestore().IncludedNamespaces("ns-1").Result(),
+			backup:       defaultBackup().Result(),
+			tarball:      test.NewTarWriter(t).AddItems("pods", builder.ForPod("ns-1", "pod-1").Result(), builder.ForPod("ns-2", "pod-2").Result()).Done(),
+			apiResources: []*test.APIResource{test.Pods()},
+			actions: []riav2.RestoreItemAction{
+				&pluggableAction{
+					selector: velero.ResourceSelector{IncludedNamespaces: []string{"ns-1"}},
+					executeFunc: func(input *velero.RestoreItemActionExecuteInput) (*velero.RestoreItemActionExecuteOutput, error) {
+						item := input.Item.(*unstructured.Unstructured)
+						annotations := item.GetAnnotations()
+						if annotations == nil {
+							annotations = map[string]string{}
+						}
+						annotations[velerov1api.MustIncludeAdditionalItemRestoreAnnotation] = "true"
+						item.SetAnnotations(annotations)
+						return &velero.RestoreItemActionExecuteOutput{
+							UpdatedItem: item,
+							AdditionalItems: []velero.ResourceIdentifier{
+								{GroupResource: kuberesource.Pods, Namespace: "ns-2", Name: "pod-2"},
+							},
+						}, nil
+					},
+				},
+			},
+			want: map[*test.APIResource][]string{
+				test.Pods(): {"ns-1/pod-1", "ns-2/pod-2"},
+			},
+		},
+		{
+			name:    "must-include annotation bypasses IncludeClusterResources=false for additional items",
+			restore: defaultRestore().IncludeClusterResources(false).Result(),
+			backup:  defaultBackup().Result(),
+			tarball: test.NewTarWriter(t).
+				AddItems("pods", builder.ForPod("ns-1", "pod-1").Result()).
+				AddItems("persistentvolumes", builder.ForPersistentVolume("pv-1").Result()).
+				Done(),
+			apiResources: []*test.APIResource{test.Pods(), test.PVs()},
+			actions: []riav2.RestoreItemAction{
+				&pluggableAction{
+					executeFunc: func(input *velero.RestoreItemActionExecuteInput) (*velero.RestoreItemActionExecuteOutput, error) {
+						item := input.Item.(*unstructured.Unstructured)
+						annotations := item.GetAnnotations()
+						if annotations == nil {
+							annotations = map[string]string{}
+						}
+						annotations[velerov1api.MustIncludeAdditionalItemRestoreAnnotation] = "true"
+						item.SetAnnotations(annotations)
+						return &velero.RestoreItemActionExecuteOutput{
+							UpdatedItem: item,
+							AdditionalItems: []velero.ResourceIdentifier{
+								{GroupResource: kuberesource.PersistentVolumes, Name: "pv-1"},
+							},
+						}, nil
+					},
+				},
+			},
+			want: map[*test.APIResource][]string{
+				test.Pods(): {"ns-1/pod-1"},
+				test.PVs():  {"/pv-1"},
+			},
+		},
 	}

 	for _, tc := range tests {
@@ -2180,6 +2299,370 @@ func TestRestoreActionAdditionalItems(t *testing.T) {
 	}
 }

+// TestRestoreMustIncludeAdditionalItems covers restore must-include edge cases beyond the
+// basic filter-bypass cases in TestRestoreActionAdditionalItems.
+func TestRestoreMustIncludeAdditionalItems(t *testing.T) {
+	t.Run("must-include annotation is stripped from the restored item", func(t *testing.T) {
+		h := newHarness(t)
+		h.AddItems(t, test.Pods())
+
+		data := &Request{
+			Log:     h.log,
+			Restore: defaultRestore().Result(),
+			Backup:  defaultBackup().Result(),
+			BackupReader: test.NewTarWriter(t).
+				AddItems("pods", builder.ForPod("ns-1", "pod-1").Result()).
+				Done(),
+		}
+		warnings, errs := h.restorer.Restore(
+			data,
+			[]riav2.RestoreItemAction{
+				&pluggableAction{
+					executeFunc: func(input *velero.RestoreItemActionExecuteInput) (*velero.RestoreItemActionExecuteOutput, error) {
+						item := input.Item.(*unstructured.Unstructured)
+						annotations := item.GetAnnotations()
+						if annotations == nil {
+							annotations = map[string]string{}
+						}
+						annotations[velerov1api.MustIncludeAdditionalItemRestoreAnnotation] = "true"
+						annotations["keep-me"] = "yes"
+						item.SetAnnotations(annotations)
+						return &velero.RestoreItemActionExecuteOutput{UpdatedItem: item}, nil
+					},
+				},
+			},
+			nil,
+		)
+
+		assertEmptyResults(t, warnings, errs)
+
+		got, err := h.DynamicClient.Resource(test.Pods().GVR()).Namespace("ns-1").Get(t.Context(), "pod-1", metav1.GetOptions{})
+		require.NoError(t, err)
+		annotations := got.GetAnnotations()
+		assert.NotContains(t, annotations, velerov1api.MustIncludeAdditionalItemRestoreAnnotation)
+		assert.Equal(t, "yes", annotations["keep-me"])
+	})
+
+	t.Run("non-true must-include annotation is stripped without bypassing filters", func(t *testing.T) {
+		h := newHarness(t)
+		h.AddItems(t, test.Pods())
+		h.AddItems(t, test.PVs())
+
+		data := &Request{
+			Log:     h.log,
+			Restore: defaultRestore().IncludedResources("pods").Result(),
+			Backup:  defaultBackup().Result(),
+			BackupReader: test.NewTarWriter(t).
+				AddItems("pods", builder.ForPod("ns-1", "pod-1").Result()).
+				AddItems("persistentvolumes", builder.ForPersistentVolume("pv-1").Result()).
+				Done(),
+		}
+		warnings, errs := h.restorer.Restore(
+			data,
+			[]riav2.RestoreItemAction{
+				&pluggableAction{
+					executeFunc: func(input *velero.RestoreItemActionExecuteInput) (*velero.RestoreItemActionExecuteOutput, error) {
+						item := input.Item.(*unstructured.Unstructured)
+						annotations := item.GetAnnotations()
+						if annotations == nil {
+							annotations = map[string]string{}
+						}
+						annotations[velerov1api.MustIncludeAdditionalItemRestoreAnnotation] = "True"
+						annotations["keep-me"] = "yes"
+						item.SetAnnotations(annotations)
+						return &velero.RestoreItemActionExecuteOutput{
+							UpdatedItem: item,
+							AdditionalItems: []velero.ResourceIdentifier{
+								{GroupResource: kuberesource.PersistentVolumes, Name: "pv-1"},
+							},
+						}, nil
+					},
+				},
+			},
+			nil,
+		)
+
+		assertEmptyResults(t, warnings, errs)
+		assertAPIContents(t, h, map[*test.APIResource][]string{
+			test.Pods(): {"ns-1/pod-1"},
+			test.PVs():  nil,
+		})
+
+		got, err := h.DynamicClient.Resource(test.Pods().GVR()).Namespace("ns-1").Get(t.Context(), "pod-1", metav1.GetOptions{})
+		require.NoError(t, err)
+		annotations := got.GetAnnotations()
+		assert.NotContains(t, annotations, velerov1api.MustIncludeAdditionalItemRestoreAnnotation)
+		assert.Equal(t, "yes", annotations["keep-me"])
+	})
+
+	t.Run("SkipRestore supersedes must-include annotation and skips additional items", func(t *testing.T) {
+		h := newHarness(t)
+		h.AddItems(t, test.Pods())
+		h.AddItems(t, test.PVs())
+
+		data := &Request{
+			Log:     h.log,
+			Restore: defaultRestore().IncludedResources("pods").Result(),
+			Backup:  defaultBackup().Result(),
+			BackupReader: test.NewTarWriter(t).
+				AddItems("pods", builder.ForPod("ns-1", "pod-1").Result()).
+				AddItems("persistentvolumes", builder.ForPersistentVolume("pv-1").Result()).
+				Done(),
+		}
+		warnings, errs := h.restorer.Restore(
+			data,
+			[]riav2.RestoreItemAction{
+				&pluggableAction{
+					executeFunc: func(input *velero.RestoreItemActionExecuteInput) (*velero.RestoreItemActionExecuteOutput, error) {
+						item := input.Item.(*unstructured.Unstructured)
+						annotations := item.GetAnnotations()
+						if annotations == nil {
+							annotations = map[string]string{}
+						}
+						annotations[velerov1api.MustIncludeAdditionalItemRestoreAnnotation] = "true"
+						item.SetAnnotations(annotations)
+						return &velero.RestoreItemActionExecuteOutput{
+							UpdatedItem: item,
+							SkipRestore: true,
+							AdditionalItems: []velero.ResourceIdentifier{
+								{GroupResource: kuberesource.PersistentVolumes, Name: "pv-1"},
+							},
+						}, nil
+					},
+				},
+			},
+			nil,
+		)
+
+		assertEmptyResults(t, warnings, errs)
+		assertAPIContents(t, h, map[*test.APIResource][]string{
+			test.Pods(): nil,
+			test.PVs():  nil,
+		})
+	})
+
+	t.Run("must-include does not restore additional items missing from the backup tarball", func(t *testing.T) {
+		h := newHarness(t)
+		h.AddItems(t, test.Pods())
+		h.AddItems(t, test.PVs())
+
+		data := &Request{
+			Log:     h.log,
+			Restore: defaultRestore().IncludedResources("pods").Result(),
+			Backup:  defaultBackup().Result(),
+			BackupReader: test.NewTarWriter(t).
+				AddItems("pods", builder.ForPod("ns-1", "pod-1").Result()).
+				Done(),
+		}
+		warnings, errs := h.restorer.Restore(
+			data,
+			[]riav2.RestoreItemAction{
+				&pluggableAction{
+					executeFunc: func(input *velero.RestoreItemActionExecuteInput) (*velero.RestoreItemActionExecuteOutput, error) {
+						item := input.Item.(*unstructured.Unstructured)
+						annotations := item.GetAnnotations()
+						if annotations == nil {
+							annotations = map[string]string{}
+						}
+						annotations[velerov1api.MustIncludeAdditionalItemRestoreAnnotation] = "true"
+						item.SetAnnotations(annotations)
+						return &velero.RestoreItemActionExecuteOutput{
+							UpdatedItem: item,
+							AdditionalItems: []velero.ResourceIdentifier{
+								{GroupResource: kuberesource.PersistentVolumes, Name: "pv-missing"},
+							},
+						}, nil
+					},
+				},
+			},
+			nil,
+		)
+
+		assertEmptyResults(t, errs)
+		assertNonEmptyResults(t, "warning", warnings)
+		assertAPIContents(t, h, map[*test.APIResource][]string{
+			test.Pods(): {"ns-1/pod-1"},
+			test.PVs():  nil,
+		})
+	})
+
+	t.Run("transitive must-include requires each RIA level to re-set the annotation", func(t *testing.T) {
+		h := newHarness(t)
+		h.AddItems(t, test.Pods())
+		h.AddItems(t, test.PVs())
+		h.AddItems(t, test.PVCs())
+
+		data := &Request{
+			Log:     h.log,
+			Restore: defaultRestore().IncludedResources("pods").Result(),
+			Backup:  defaultBackup().Result(),
+			BackupReader: test.NewTarWriter(t).
+				AddItems("pods", builder.ForPod("ns-1", "pod-1").Result()).
+				AddItems("persistentvolumes", builder.ForPersistentVolume("pv-1").Result()).
+				AddItems("persistentvolumeclaims", builder.ForPersistentVolumeClaim("ns-2", "pvc-1").Result()).
+				Done(),
+		}
+		warnings, errs := h.restorer.Restore(
+			data,
+			[]riav2.RestoreItemAction{
+				// Parent pod RIA force-includes the excluded PV.
+				&pluggableAction{
+					selector: velero.ResourceSelector{IncludedResources: []string{"pods"}},
+					executeFunc: func(input *velero.RestoreItemActionExecuteInput) (*velero.RestoreItemActionExecuteOutput, error) {
+						item := input.Item.(*unstructured.Unstructured)
+						annotations := item.GetAnnotations()
+						if annotations == nil {
+							annotations = map[string]string{}
+						}
+						annotations[velerov1api.MustIncludeAdditionalItemRestoreAnnotation] = "true"
+						item.SetAnnotations(annotations)
+						return &velero.RestoreItemActionExecuteOutput{
+							UpdatedItem: item,
+							AdditionalItems: []velero.ResourceIdentifier{
+								{GroupResource: kuberesource.PersistentVolumes, Name: "pv-1"},
+							},
+						}, nil
+					},
+				},
+				// Child PV RIA also re-sets the annotation to force-include an excluded PVC.
+				&pluggableAction{
+					selector: velero.ResourceSelector{IncludedResources: []string{"persistentvolumes"}},
+					executeFunc: func(input *velero.RestoreItemActionExecuteInput) (*velero.RestoreItemActionExecuteOutput, error) {
+						item := input.Item.(*unstructured.Unstructured)
+						annotations := item.GetAnnotations()
+						if annotations == nil {
+							annotations = map[string]string{}
+						}
+						annotations[velerov1api.MustIncludeAdditionalItemRestoreAnnotation] = "true"
+						item.SetAnnotations(annotations)
+						return &velero.RestoreItemActionExecuteOutput{
+							UpdatedItem: item,
+							AdditionalItems: []velero.ResourceIdentifier{
+								{GroupResource: kuberesource.PersistentVolumeClaims, Namespace: "ns-2", Name: "pvc-1"},
+							},
+						}, nil
+					},
+				},
+			},
+			nil,
+		)
+
+		assertEmptyResults(t, warnings, errs)
+		assertAPIContents(t, h, map[*test.APIResource][]string{
+			test.Pods(): {"ns-1/pod-1"},
+			test.PVs():  {"/pv-1"},
+			test.PVCs(): {"ns-2/pvc-1"},
+		})
+	})
+
+	t.Run("without re-annotating, transitive additional items still respect filters", func(t *testing.T) {
+		h := newHarness(t)
+		h.AddItems(t, test.Pods())
+		h.AddItems(t, test.PVs())
+		h.AddItems(t, test.PVCs())
+
+		data := &Request{
+			Log:     h.log,
+			Restore: defaultRestore().IncludedResources("pods").Result(),
+			Backup:  defaultBackup().Result(),
+			BackupReader: test.NewTarWriter(t).
+				AddItems("pods", builder.ForPod("ns-1", "pod-1").Result()).
+				AddItems("persistentvolumes", builder.ForPersistentVolume("pv-1").Result()).
+				AddItems("persistentvolumeclaims", builder.ForPersistentVolumeClaim("ns-2", "pvc-1").Result()).
+				Done(),
+		}
+		warnings, errs := h.restorer.Restore(
+			data,
+			[]riav2.RestoreItemAction{
+				&pluggableAction{
+					selector: velero.ResourceSelector{IncludedResources: []string{"pods"}},
+					executeFunc: func(input *velero.RestoreItemActionExecuteInput) (*velero.RestoreItemActionExecuteOutput, error) {
+						item := input.Item.(*unstructured.Unstructured)
+						annotations := item.GetAnnotations()
+						if annotations == nil {
+							annotations = map[string]string{}
+						}
+						annotations[velerov1api.MustIncludeAdditionalItemRestoreAnnotation] = "true"
+						item.SetAnnotations(annotations)
+						return &velero.RestoreItemActionExecuteOutput{
+							UpdatedItem: item,
+							AdditionalItems: []velero.ResourceIdentifier{
+								{GroupResource: kuberesource.PersistentVolumes, Name: "pv-1"},
+							},
+						}, nil
+					},
+				},
+				// Child PV RIA returns an additional PVC but does NOT set must-include.
+				&pluggableAction{
+					selector: velero.ResourceSelector{IncludedResources: []string{"persistentvolumes"}},
+					executeFunc: func(input *velero.RestoreItemActionExecuteInput) (*velero.RestoreItemActionExecuteOutput, error) {
+						return &velero.RestoreItemActionExecuteOutput{
+							UpdatedItem: input.Item,
+							AdditionalItems: []velero.ResourceIdentifier{
+								{GroupResource: kuberesource.PersistentVolumeClaims, Namespace: "ns-2", Name: "pvc-1"},
+							},
+						}, nil
+					},
+				},
+			},
+			nil,
+		)
+
+		assertEmptyResults(t, warnings, errs)
+		assertAPIContents(t, h, map[*test.APIResource][]string{
+			test.Pods(): {"ns-1/pod-1"},
+			test.PVs():  {"/pv-1"},
+			test.PVCs(): nil,
+		})
+	})
+
+	t.Run("VS must-include restores excluded VolumeSnapshotContent additional item", func(t *testing.T) {
+		h := newHarness(t)
+		h.AddItems(t, test.VolumeSnapshots())
+		h.AddItems(t, test.VolumeSnapshotContents())
+
+		data := &Request{
+			Log:     h.log,
+			Restore: defaultRestore().IncludedResources("volumesnapshots.snapshot.storage.k8s.io").IncludeClusterResources(true).Result(),
+			Backup:  defaultBackup().Result(),
+			BackupReader: test.NewTarWriter(t).
+				AddItems("volumesnapshots.snapshot.storage.k8s.io", builder.ForVolumeSnapshot("ns-1", "vs-1").Result()).
+				AddItems("volumesnapshotcontents.snapshot.storage.k8s.io", builder.ForVolumeSnapshotContent("vsc-1").Result()).
+				Done(),
+		}
+		warnings, errs := h.restorer.Restore(
+			data,
+			[]riav2.RestoreItemAction{
+				&pluggableAction{
+					selector: velero.ResourceSelector{IncludedResources: []string{"volumesnapshots.snapshot.storage.k8s.io"}},
+					executeFunc: func(input *velero.RestoreItemActionExecuteInput) (*velero.RestoreItemActionExecuteOutput, error) {
+						item := input.Item.(*unstructured.Unstructured)
+						annotations := item.GetAnnotations()
+						if annotations == nil {
+							annotations = map[string]string{}
+						}
+						annotations[velerov1api.MustIncludeAdditionalItemRestoreAnnotation] = "true"
+						item.SetAnnotations(annotations)
+						return &velero.RestoreItemActionExecuteOutput{
+							UpdatedItem: item,
+							AdditionalItems: []velero.ResourceIdentifier{
+								{GroupResource: kuberesource.VolumeSnapshotContents, Name: "vsc-1"},
+							},
+						}, nil
+					},
+				},
+			},
+			nil,
+		)
+
+		assertEmptyResults(t, warnings, errs)
+		assertAPIContents(t, h, map[*test.APIResource][]string{
+			test.VolumeSnapshots():        {"ns-1/vs-1"},
+			test.VolumeSnapshotContents(): {"/vsc-1"},
+		})
+	})
+}
+
 // TestShouldRestore runs the ShouldRestore function for various permutations of
 // existing/nonexisting/being-deleted PVs, PVCs, and namespaces, and verifies the
 // result/error matches expectations.
@@ -58,6 +58,9 @@ func NewAPIServer(t *testing.T) *APIServer {
 				{Group: "velero.io", Version: "v2alpha1", Resource: "datauploads"}:                         "DataUploadsList",
 				{Group: "mygroup.io", Version: "v1", Resource: "mycustomkinds"}:                            "MyCustomKindList",
 				{Group: "mygroup.io", Version: "v1", Resource: "myclustercustomkinds"}:                     "MyClusterCustomKindList",
+				{Group: "storage.k8s.io", Version: "v1", Resource: "storageclasses"}:                       "StorageClassList",
+				{Group: "snapshot.storage.k8s.io", Version: "v1", Resource: "volumesnapshots"}:             "VolumeSnapshotList",
+				{Group: "snapshot.storage.k8s.io", Version: "v1", Resource: "volumesnapshotcontents"}:      "VolumeSnapshotContentList",
 			})
 		discoveryClient = &DiscoveryClient{FakeDiscovery: kubeClient.Discovery().(*discoveryfake.FakeDiscovery)}
 	)
@@ -220,3 +220,37 @@ func DataUploads(items ...metav1.Object) *APIResource {
 		Items:      items,
 	}
 }
+
+func StorageClasses(items ...metav1.Object) *APIResource {
+	return &APIResource{
+		Group:      "storage.k8s.io",
+		Version:    "v1",
+		Name:       "storageclasses",
+		ShortName:  "sc",
+		Kind:       "StorageClass",
+		Namespaced: false,
+		Items:      items,
+	}
+}
+
+func VolumeSnapshotContents(items ...metav1.Object) *APIResource {
+	return &APIResource{
+		Group:      "snapshot.storage.k8s.io",
+		Version:    "v1",
+		Name:       "volumesnapshotcontents",
+		Kind:       "VolumeSnapshotContent",
+		Namespaced: false,
+		Items:      items,
+	}
+}
+
+func VolumeSnapshots(items ...metav1.Object) *APIResource {
+	return &APIResource{
+		Group:      "snapshot.storage.k8s.io",
+		Version:    "v1",
+		Name:       "volumesnapshots",
+		Kind:       "VolumeSnapshot",
+		Namespaced: true,
+		Items:      items,
+	}
+}
@@ -65,6 +65,32 @@ order in which item action plugins are invoked. However, if a single binary impl
 they may be invoked in the order in which they are registered but it is best to not depend on this
 implementation. This is not guaranteed officially and the implementation can change at any time.

+### Must-include additional items (Restore Item Actions)
+
+Restore Item Actions may return `AdditionalItems` that Velero restores as dependencies of the current item.
+By default those additional items must still pass the restore's global resource and namespace include/exclude
+filters (and `IncludeClusterResources=false` for cluster-scoped resources).
+
+To force-restore hard dependencies despite those filters, set the following annotation on the `UpdatedItem`
+returned from `Execute()`:
+
+```
+restore.velero.io/must-include-additional-items: "true"
+```
+
+Behavior:
+- Only the string value `"true"` enables the bypass.
+- The annotation applies blanket to all `AdditionalItems` from that RIA invocation (not per-item).
+- Velero strips the annotation before applying the item to the cluster.
+- `SkipRestore: true` takes precedence: if set, the annotation is never inspected and `AdditionalItems` are not processed.
+- Must-include only bypasses filters; the additional item must still exist in the backup tarball.
+- When an additional item targets an excluded namespace, Velero may still create that target namespace so the item can be restored.
+- Cluster-scoped additional items are restored even when `IncludeClusterResources=false`.
+- Transitive force-include requires each RIA level to re-set the annotation on its own `UpdatedItem`.
+
+This mirrors the backup-side annotation `backup.velero.io/must-include-additional-items` used by Backup Item Actions.
+Installing an RIA that sets this annotation is a trust decision: the plugin can restore resources outside the operator's restore filters.
+
 ## Plugin Logging

 Velero provides a [logger][2] that can be used by plugins to log structured information to the main Velero server log or
@@ -65,6 +65,32 @@ order in which item action plugins are invoked. However, if a single binary impl
 they may be invoked in the order in which they are registered but it is best to not depend on this
 implementation. This is not guaranteed officially and the implementation can change at any time.

+### Must-include additional items (Restore Item Actions)
+
+Restore Item Actions may return `AdditionalItems` that Velero restores as dependencies of the current item.
+By default those additional items must still pass the restore's global resource and namespace include/exclude
+filters (and `IncludeClusterResources=false` for cluster-scoped resources).
+
+To force-restore hard dependencies despite those filters, set the following annotation on the `UpdatedItem`
+returned from `Execute()`:
+
+```
+restore.velero.io/must-include-additional-items: "true"
+```
+
+Behavior:
+- Only the string value `"true"` enables the bypass.
+- The annotation applies blanket to all `AdditionalItems` from that RIA invocation (not per-item).
+- Velero strips the annotation before applying the item to the cluster.
+- `SkipRestore: true` takes precedence: if set, the annotation is never inspected and `AdditionalItems` are not processed.
+- Must-include only bypasses filters; the additional item must still exist in the backup tarball.
+- When an additional item targets an excluded namespace, Velero may still create that target namespace so the item can be restored.
+- Cluster-scoped additional items are restored even when `IncludeClusterResources=false`.
+- Transitive force-include requires each RIA level to re-set the annotation on its own `UpdatedItem`.
+
+This mirrors the backup-side annotation `backup.velero.io/must-include-additional-items` used by Backup Item Actions.
+Installing an RIA that sets this annotation is a trust decision: the plugin can restore resources outside the operator's restore filters.
+
 ## Plugin Logging

 Velero provides a [logger][2] that can be used by plugins to log structured information to the main Velero server log or
				`@@ -0,0 +1 @@`
				Add restore.velero.io/must-include-additional-items so RestoreItemActions can opt in to bypassing global restore filters for AdditionalItems (mirrors the backup-side must-include annotation; no default behavior change for existing restores/plugins). Stop force-including VolumeSnapshotContents via resourceMustHave on every restore; CSI VolumeSnapshot/PVC RestoreItemActions now set restore.velero.io/must-include-additional-items so bound snapshot dependencies are restored only when their parent is restored (fixes #9957)