Keep disabled local backup disks scoped to their node (#3361)

Do not let a successful same-named peer erase an unread local disk or
qualify its backup history as complete. Resolve ambiguous disabled-node
membership from the bounded datacenter catalogue without querying
disabled content; preserve unknown scopes, accepted shared reads and
cancellation.

---
Fleet candidate `20261026T213457Z-core-runtime` from the `core-runtime`
lane. Reviewed tip `24610bdf267b5dc9c6373a460e752e34010a2542` on `main`.

[How Pulse handles
triage](https://github.com/rcourtman/Pulse/blob/main/docs/AI_TRANSPARENCY.md)

docs/release-control/v6/internal/subsystems/monitoring.md

@@ -17,6 +17,35 @@
 
 ## Purpose
 
+### Disabled local backup targets keep their own unread scope
+
+A skipped local backup-content read cannot be covered by a successful
+same-named disk on another PVE node. Known live files retain their exact read
+origin, identity and dates; durable points retain their prior evidence and an
+incomplete provider observation, including when the live cache is empty.
+Accepted sibling rows still update or disappear on an accepted empty read.
+Disabled content endpoints are never queried to recover membership.
+
+PVE also lists a node-restricted local target as disabled on excluded nodes.
+Only for skipped local targets with an accepted same-name peer, one ordinary
+bounded datacenter storage-catalogue read resolves that ambiguity. An explicit
+`nodes` restriction can establish exclusion; matching names, missing/unrestricted
+configuration and failed config reads cannot. Those unknown scopes stay unread,
+not permission-denied or confirmed absent. Shared/inherently remote storage
+retains its existing accepted-peer semantics. A cancelled catalogue read cannot
+publish new inventory or provider evidence. Normal re-enable and complete reads
+still establish absence and recovery. No public fields, identifiers, freshness
+policy, access grants or guest/backup/recovery operations change.
+
+`pve_disabled_local_storage_test.go` covers VM/LXC subjects, both node orders,
+warm/cold live inventory, repeated skips, empty readable peers, unavailable or
+denied configuration, explicit node restrictions, shared controls, cancellation,
+connection isolation, all-disabled reads and ordinary recovery. Its normal-suite
+lifecycle optionally exports actual poller/ingest/posture frames for production
+Backups/API-hook desktop/narrow browser proof. This is monitoring correctness,
+not archive availability, installed guest/filesystem recovery or a diagnosis of
+#3248's original stale dates.
+
 ### PBS unread namespaces retain only their own last accepted live files
 
 When a PBS group-list request fails transiently, its previous live snapshot
@@ -54,8 +83,8 @@ public `node` remains guest attribution and may differ after migration. Hard
 content failures and unread node catalogues/offline nodes retain only their
 known local rows. Older rows without a read origin are retained conservatively
 until an ordinary complete read replaces them. An accepted shared-storage read
-still covers that shared repository; disabled/node-restricted query admission
-and its cross-node success handling are unchanged.
+still covers that shared repository. Disabled content-query admission is
+unchanged; skipped local membership follows the scoped rule above.
 
 Carried rows keep their exact dates and IDs, do not enter the observed-point
 batch and cannot renew point evidence. Partial scans remain partial and cannot

docs/release-control/v6/internal/subsystems/storage-recovery.md

@@ -21,6 +21,35 @@
 
 ## Purpose
 
+### Disabled local backup targets keep their own unread scope
+
+A skipped local backup-content read cannot be covered by a successful
+same-named disk on another PVE node. Known live files retain their exact read
+origin, identity and dates; durable points retain their prior evidence and an
+incomplete provider observation, including when the live cache is empty.
+Accepted sibling rows still update or disappear on an accepted empty read.
+Disabled content endpoints are never queried to recover membership.
+
+PVE also lists a node-restricted local target as disabled on excluded nodes.
+Only for skipped local targets with an accepted same-name peer, one ordinary
+bounded datacenter storage-catalogue read resolves that ambiguity. An explicit
+`nodes` restriction can establish exclusion; matching names, missing/unrestricted
+configuration and failed config reads cannot. Those unknown scopes stay unread,
+not permission-denied or confirmed absent. Shared/inherently remote storage
+retains its existing accepted-peer semantics. A cancelled catalogue read cannot
+publish new inventory or provider evidence. Normal re-enable and complete reads
+still establish absence and recovery. No public fields, identifiers, freshness
+policy, access grants or guest/backup/recovery operations change.
+
+`pve_disabled_local_storage_test.go` covers VM/LXC subjects, both node orders,
+warm/cold live inventory, repeated skips, empty readable peers, unavailable or
+denied configuration, explicit node restrictions, shared controls, cancellation,
+connection isolation, all-disabled reads and ordinary recovery. Its normal-suite
+lifecycle optionally exports actual poller/ingest/posture frames for production
+Backups/API-hook desktop/narrow browser proof. This is monitoring correctness,
+not archive availability, installed guest/filesystem recovery or a diagnosis of
+#3248's original stale dates.
+
 ### PBS unread namespaces retain only their own last accepted live files
 
 When a PBS group-list request fails transiently, its previous live snapshot
@@ -94,8 +123,8 @@ public `node` remains guest attribution and may differ after migration. Hard
 content failures and unread node catalogues/offline nodes retain only their
 known local rows. Older rows without a read origin are retained conservatively
 until an ordinary complete read replaces them. An accepted shared-storage read
-still covers that shared repository; disabled/node-restricted query admission
-and its cross-node success handling are unchanged.
+still covers that shared repository. Disabled content-query admission is
+unchanged; skipped local membership follows the scoped rule above.
 
 Carried rows keep their exact dates and IDs, do not enter the observed-point
 batch and cannot renew point evidence. Partial scans remain partial and cannot

internal/monitoring/monitor_backups.go

@@ -195,6 +195,7 @@ func (m *Monitor) pollStorageBackupsWithNodes(ctx context.Context, instanceName
 	storageSuccess := map[string]struct{}{}
 	sharedStorageSuccess := map[string]struct{}{}
 	unqueryableStorages := map[string]struct{}{}
+	skippedLocalStorages := map[pveLocalBackupStorage]struct{}{}
 	readState := m.backupReadStateForInstance(instanceName)
 
 	// Build guest lookup map to find actual node for each VMID
@@ -285,26 +286,24 @@ func (m *Monitor) pollStorageBackupsWithNodes(ctx context.Context, instanceName
 				continue
 			}
 			storagesWithBackup++
-			if !storageContentQueryable(storage) {
-				// Configured but unreadable is not an empty content listing.
-				// Preserve the no-query boundary (notably disabled storage)
-				// without letting this skipped scope delete known history or
-				// qualify the provider-wide observation as complete. PVE also
-				// reports storage restricted to other nodes as disabled here,
-				// so whether it went unread is decided after every node.
-				unqueryableStorages[storage.Storage] = struct{}{}
-				if _, ok := storageSuccess[storage.Storage]; !ok {
-					storagePreserveNeeded[storage.Storage] = struct{}{}
-				}
-				continue
-			}
-
-			// A shared listing can cover several nodes. A same-named local
-			// disk cannot: keep its read origin separate from guest location.
+			// Establish repository scope before admission: a disabled local
+			// disk is no more covered by a peer than a failed content read.
 			storageNode := ""
 			if storage.Shared == 0 && !isInherentlySharedStorageType(storage.Type) {
 				storageNode = node.Node
 			}
+			if !storageContentQueryable(storage) {
+				// Do not query disabled content to resolve its membership. A
+				// shared read may cover a skipped peer, but local assignment
+				// needs independent evidence after every node has been read.
+				if storageNode != "" {
+					skippedLocalStorages[pveLocalBackupStorage{node: storageNode, storage: storage.Storage}] = struct{}{}
+				} else {
+					unqueryableStorages[storage.Storage] = struct{}{}
+					storagePreserveNeeded[storage.Storage] = struct{}{}
+				}
+				continue
+			}
 
 			// Get storage content
 			contents, err := client.GetStorageContent(ctx, node.Node, storage.Storage)
@@ -465,13 +464,41 @@ func (m *Monitor) pollStorageBackupsWithNodes(ctx context.Context, instanceName
 		}
 	}
 
+	if ctx.Err() != nil {
+		return
+	}
+
+	// PVE also lists node-restricted local storage as disabled on excluded
+	// nodes. A peer's matching name alone cannot distinguish that from an
+	// unread local disk. Resolve only that ambiguity with the ordinary bounded
+	// datacenter catalogue, once per poll; failed/missing config is not absence.
+	storageAssignments := make(map[string]string)
+	for scope := range skippedLocalStorages {
+		if _, readElsewhere := storageSuccess[scope.storage]; !readElsewhere {
+			continue
+		}
+		if configured, err := client.GetAllStorage(ctx); err == nil {
+			for _, storage := range configured {
+				storageAssignments[storage.Storage] = storage.Nodes
+			}
+		}
+		break
+	}
 	if ctx.Err() != nil {
 		// A cancelled enumeration cannot publish a successful empty inventory
 		// or renew provider evidence for a payload we did not accept.
 		return
 	}
+	for scope := range skippedLocalStorages {
+		if assignment, known := storageAssignments[scope.storage]; known &&
+			clusterStorageRestrictedToOtherNodes(assignment, scope.node) {
+			continue
+		}
+		failedLocalStorages[scope] = struct{}{}
+		contentFailures++
+	}
 
-	// A storage skipped on one node is unread only if no node listed it.
+	// A shared storage skipped on one node is unread only if no node listed it.
 	for storageName := range unqueryableStorages {
 		if _, ok := storageSuccess[storageName]; !ok {
 			contentFailures++
@@ -785,8 +812,8 @@ type pveLocalBackupStorage struct {
 	storage string
 }
 
-// Name-wide failures cover unread/disabled targets and shared storage. Actual
-// local content failures and unread nodes must survive another node's success.
+// Name-wide failures cover shared storage. Local failed/skipped content reads
+// and unread nodes must survive another node's success.
 // Legacy rows without provenance are retained conservatively until a complete
 // read replaces them; guest location is not evidence of storage read origin.
 type pveBackupReadFailures struct {

internal/monitoring/pve_disabled_local_storage_test.go

@@ -0,0 +1,378 @@
+package monitoring
+
+import (
+	"context"
+	"crypto/sha256"
+	"encoding/json"
+	"fmt"
+	"os"
+	"reflect"
+	"strings"
+	"testing"
+	"time"
+
+	"github.com/rcourtman/pulse-go-rewrite/internal/models"
+	"github.com/rcourtman/pulse-go-rewrite/internal/operationaltrust"
+	"github.com/rcourtman/pulse-go-rewrite/internal/recovery"
+	"github.com/rcourtman/pulse-go-rewrite/internal/unifiedresources"
+	"github.com/rcourtman/pulse-go-rewrite/pkg/proxmox"
+)
+
+// Identically named local disks are independent scopes even when one is
+// disabled. No disabled content endpoint may be queried to prove that fact.
+type disabledLocalBackupClient struct {
+	*pveSameNameStorageClient
+	disabled       map[string]bool
+	kind           string
+	shared         bool
+	storageType    string
+	config         []proxmox.Storage
+	configErr      error
+	configCalls    int
+	contentQueries map[string]int
+	afterConfig    func()
+}
+
+func (c *disabledLocalBackupClient) GetStorage(ctx context.Context, node string) ([]proxmox.Storage, error) {
+	rows, err := c.pveSameNameStorageClient.GetStorage(ctx, node)
+	for i := range rows {
+		if c.disabled[node] {
+			rows[i].Enabled = 0
+		}
+		if c.shared {
+			rows[i].Shared = 1
+		}
+		if c.storageType != "" {
+			rows[i].Type = c.storageType
+		}
+	}
+	return rows, err
+}
+
+func (c *disabledLocalBackupClient) GetStorageContent(ctx context.Context, node, storage string) ([]proxmox.StorageContent, error) {
+	c.contentQueries[node]++
+	if c.disabled[node] {
+		return nil, fmt.Errorf("unexpected disabled content read")
+	}
+	rows, err := c.pveSameNameStorageClient.GetStorageContent(ctx, node, storage)
+	if c.kind == "lxc" {
+		for i := range rows {
+			rows[i].Volid = strings.ReplaceAll(rows[i].Volid, "qemu", "lxc")
+		}
+	}
+	return rows, err
+}
+
+func (c *disabledLocalBackupClient) GetAllStorage(context.Context) ([]proxmox.Storage, error) {
+	c.configCalls++
+	if c.afterConfig != nil {
+		c.afterConfig()
+	}
+	return c.config, c.configErr
+}
+
+func newDisabledLocalBackupFixture(t *testing.T, kind string) (*pveReconciliationFixture, *disabledLocalBackupClient) {
+	t.Helper()
+	f := newPVEReconciliationFixture(t)
+	c := &disabledLocalBackupClient{pveSameNameStorageClient: &pveSameNameStorageClient{f.client}, kind: kind,
+		config: []proxmox.Storage{{Storage: "local", Type: "dir"}}, contentQueries: make(map[string]int)}
+	if kind == "lxc" {
+		f.m.state.UpdateVMs(nil)
+		f.m.state.UpdateContainers([]models.Container{
+			{ID: makeGuestID("cluster", "node-a", 101), VMID: 101, Instance: "cluster", Node: "node-a", Name: "guest-101", Status: "running"},
+			{ID: makeGuestID("cluster", "node-b", 102), VMID: 102, Instance: "cluster", Node: "node-b", Name: "guest-102", Status: "running"},
+		})
+		for i, id := range []int{101, 102} {
+			f.ids[i] = unifiedresources.ProxmoxGuestCanonicalID(unifiedresources.ResourceTypeSystemContainer, "cluster", id)
+		}
+	}
+	return f, c
+}
+
+func TestPVEDisabledLocalStorageDoesNotLoseKnownHistory(t *testing.T) {
+	for _, kind := range []string{"qemu", "lxc"} {
+		for _, reverse := range []bool{false, true} {
+			for _, cold := range []bool{false, true} {
+				for _, emptyPeer := range []bool{false, true} {
+					t.Run(fmt.Sprintf("%s/reverse=%t/cold=%t/empty-peer=%t", kind, reverse, cold, emptyPeer), func(t *testing.T) {
+						f, c := newDisabledLocalBackupFixture(t, kind)
+						if reverse {
+							f.nodes[0], f.nodes[1] = f.nodes[1], f.nodes[0]
+						}
+						poll := func() {
+							f.m.pollStorageBackupsWithNodes(context.Background(), "cluster", c, f.nodes, f.online)
+							f.wait(t)
+						}
+						poll()
+						before, raw := f.points(t), f.m.state.GetSnapshot().PVEBackups.StorageBackups
+						if len(before) != 2 || len(raw) != 2 {
+							t.Fatal("healthy controls did not produce two independent backups")
+						}
+						if cold {
+							f.m.state.UpdateStorageBackupsForInstance("cluster", nil)
+						}
+						c.disabled = map[string]bool{"node-b": true}
+						c.contentQueries = make(map[string]int)
+						f.client.at = f.client.at.Add(15 * time.Minute)
+						f.client.emptyAll = emptyPeer
+						for range 2 {
+							poll()
+							points := f.points(t)
+							if len(points) != 2 {
+								t.Errorf("disabled local disk was reconciled as absent: points=%d want2", len(points))
+							}
+							for id, p := range before {
+								if p.SubjectResourceID == f.ids[1] && !reflect.DeepEqual(p, points[id]) {
+									t.Error("unread local point disappeared or renewed its evidence")
+								}
+							}
+							rows := f.m.state.GetSnapshot().PVEBackups.StorageBackups
+							want := 2
+							if cold {
+								want--
+							}
+							if emptyPeer {
+								want--
+							}
+							if len(rows) != want {
+								t.Errorf("live local inventory=%d want%d; readable sibling cannot cover unread disk", len(rows), want)
+							}
+							if !cold {
+								for _, old := range raw {
+									if old.VMID != 102 {
+										continue
+									}
+									var retained models.StorageBackup
+									for _, r := range rows {
+										if r.VMID == 102 {
+											retained = r
+										}
+									}
+									if !reflect.DeepEqual(old, retained) {
+										t.Error("disabled raw row changed or disappeared")
+									}
+								}
+							}
+							f.assertHistory(t, recovery.ProtectionHistoryPartial, operationaltrust.EvidencePermissionsUnknown)
+						}
+						if c.contentQueries["node-b"] != 0 || c.contentQueries["node-a"] != 2 {
+							t.Fatalf("no-query/independent-read boundary failed: %+v", c.contentQueries)
+						}
+						if c.configCalls != 2 {
+							t.Errorf("ambiguous membership reads=%d want one per poll", c.configCalls)
+						}
+						c.disabled = nil
+						f.client.emptyAll = false
+						f.client.emptyB = true
+						poll()
+						if len(f.points(t)) != 1 || len(f.m.state.GetSnapshot().PVEBackups.StorageBackups) != 1 {
+							t.Error("successful empty read after re-enable did not remove only its absent point")
+						}
+						f.client.emptyB = false
+						poll()
+						if len(f.points(t)) != 2 || len(f.m.state.GetSnapshot().PVEBackups.StorageBackups) != 2 {
+							t.Error("ordinary accepted read did not restore both backups")
+						}
+						f.assertHistory(t, recovery.ProtectionHistoryComplete, operationaltrust.EvidencePermissionsSufficient)
+					})
+				}
+			}
+		}
+	}
+}
+
+func TestPVEDisabledLocalStorageUnknownAssignmentIsNotAbsence(t *testing.T) {
+	for _, tc := range []struct {
+		name   string
+		config []proxmox.Storage
+		err    error
+	}{
+		{name: "missing-local-config"},
+		{name: "transient-config", err: fmt.Errorf("503 unavailable")},
+		{name: "denied-config", err: fmt.Errorf("403 forbidden")},
+		{name: "allowed-node-not-peer", config: []proxmox.Storage{{Storage: "local", Nodes: "node-a,node-b"}}},
+	} {
+		t.Run(tc.name, func(t *testing.T) {
+			f, c := newDisabledLocalBackupFixture(t, "qemu")
+			f.m.pollStorageBackupsWithNodes(context.Background(), "cluster", c, f.nodes, f.online)
+			f.wait(t)
+			before := f.points(t)
+			c.disabled = map[string]bool{"node-b": true}
+			c.config = tc.config
+			c.configErr = tc.err
+			f.client.emptyAll = true
+			f.m.pollStorageBackupsWithNodes(context.Background(), "cluster", c, f.nodes, f.online)
+			f.wait(t)
+			if len(f.points(t)) != 2 || len(f.m.state.GetSnapshot().PVEBackups.StorageBackups) != 1 {
+				t.Error("unknown assignment deleted local history")
+			}
+			for id, p := range before {
+				if p.SubjectResourceID == f.ids[1] && !reflect.DeepEqual(p, f.points(t)[id]) {
+					t.Error("unknown assignment renewed unread evidence")
+				}
+			}
+			f.assertHistory(t, recovery.ProtectionHistoryPartial, operationaltrust.EvidencePermissionsUnknown)
+		})
+	}
+}
+
+func TestPVEDisabledLocalStorageRespectsConfirmedRestriction(t *testing.T) {
+	for _, reverse := range []bool{false, true} {
+		t.Run(fmt.Sprintf("reverse=%t", reverse), func(t *testing.T) {
+			f, c := newDisabledLocalBackupFixture(t, "qemu")
+			if reverse {
+				f.nodes[0], f.nodes[1] = f.nodes[1], f.nodes[0]
+			}
+			c.disabled = map[string]bool{"node-b": true}
+			c.config = []proxmox.Storage{{Storage: "local", Nodes: " node-a "}}
+			f.m.pollStorageBackupsWithNodes(context.Background(), "cluster", c, f.nodes, f.online)
+			f.wait(t)
+			if len(f.points(t)) != 1 || len(f.m.state.GetSnapshot().PVEBackups.StorageBackups) != 1 || c.contentQueries["node-b"] != 0 {
+				t.Error("confirmed node restriction lost its no-query/accepted-scope boundary")
+			}
+			for _, p := range f.postures(t) {
+				if p.SubjectResourceID == f.ids[0] && (p.State != recovery.ProtectionStateProtected || len(p.ProviderStates) != 1 || p.ProviderStates[0].HistoryCompleteness != recovery.ProtectionHistoryComplete) {
+					t.Errorf("confirmed assigned scope not complete: %+v", p)
+				}
+			}
+		})
+	}
+}
+
+func TestPVEDisabledLocalStorageSharedReadStillCoversPeer(t *testing.T) {
+	for _, storageType := range []string{"dir", "nfs"} {
+		t.Run(storageType, func(t *testing.T) {
+			f, c := newDisabledLocalBackupFixture(t, "qemu")
+			c.storageType = storageType
+			c.shared = storageType == "dir"
+			f.m.pollStorageBackupsWithNodes(context.Background(), "cluster", c, f.nodes, f.online)
+			f.wait(t)
+			c.disabled = map[string]bool{"node-b": true}
+			f.client.emptyAll = true
+			f.m.pollStorageBackupsWithNodes(context.Background(), "cluster", c, f.nodes, f.online)
+			f.wait(t)
+			if len(f.points(t)) != 0 || len(f.m.state.GetSnapshot().PVEBackups.StorageBackups) != 0 {
+				t.Error("skipped shared peer overrode accepted shared absence")
+			}
+			if c.configCalls != 0 {
+				t.Error("unambiguous shared read acquired unnecessary config")
+			}
+		})
+	}
+}
+
+func TestPVEDisabledLocalStorageConfigCancellationDoesNotPublish(t *testing.T) {
+	f, c := newDisabledLocalBackupFixture(t, "qemu")
+	f.m.pollStorageBackupsWithNodes(context.Background(), "cluster", c, f.nodes, f.online)
+	f.wait(t)
+	before, raw := f.points(t), f.m.state.GetSnapshot().PVEBackups.StorageBackups
+	postures := f.postures(t)
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+	c.disabled = map[string]bool{"node-b": true}
+	c.afterConfig = cancel
+	f.client.emptyAll = true
+	f.m.pollStorageBackupsWithNodes(ctx, "cluster", c, f.nodes, f.online)
+	f.wait(t)
+	if !reflect.DeepEqual(raw, f.m.state.GetSnapshot().PVEBackups.StorageBackups) || !reflect.DeepEqual(before, f.points(t)) {
+		t.Error("cancelled assignment read published inventory or reconciled points")
+	}
+	after := f.postures(t)
+	for i := range after {
+		if !reflect.DeepEqual(postures[i].ProviderStates, after[i].ProviderStates) {
+			t.Error("cancelled assignment read renewed provider evidence")
+		}
+	}
+}
+
+// The standard Go suite checks this producer's lifecycle. Its optional export
+// feeds the real Backups consumer as evidence, not a separate application.
+func TestPVEDisabledLocalStorageAcceptedReadLifecycle(t *testing.T) {
+	f, c := newDisabledLocalBackupFixture(t, "lxc")
+	frames := map[string]any{}
+	for _, phase := range []string{"observed", "disabled", "assignment-unavailable", "recovered", "confirmed-empty"} {
+		switch phase {
+		case "disabled":
+			c.disabled = map[string]bool{"node-b": true}
+			f.client.at = f.client.at.Add(15 * time.Minute)
+		case "assignment-unavailable":
+			c.configErr = fmt.Errorf("503 unavailable")
+		case "recovered":
+			c.disabled = nil
+			c.configErr = nil
+		case "confirmed-empty":
+			f.client.emptyB = true
+		}
+		f.m.pollStorageBackupsWithNodes(context.Background(), "cluster", c, f.nodes, f.online)
+		f.wait(t)
+		rows, points, postures := f.m.state.GetSnapshot().PVEBackups, f.points(t), f.postures(t)
+		want := 2
+		if phase == "confirmed-empty" {
+			want = 1
+		}
+		if len(rows.StorageBackups) != want || len(points) != want {
+			t.Fatalf("phase %s inventory/points=%d/%d want%d", phase, len(rows.StorageBackups), len(points), want)
+		}
+		if phase == "disabled" || phase == "assignment-unavailable" {
+			f.assertHistory(t, recovery.ProtectionHistoryPartial, operationaltrust.EvidencePermissionsUnknown)
+		}
+		frames[phase] = map[string]any{"pve": map[string]any{"data": rows}, "postures": map[string]any{"data": postures, "policy": recovery.DefaultProtectionPosturePolicy.Payload()}, "points": points}
+	}
+	if os.Getenv("PULSE_DISABLED_LOCAL_FRAMES") != "1" {
+		return
+	}
+	source, err := os.ReadFile("monitor_backups.go")
+	if err != nil {
+		t.Fatal(err)
+	}
+	data, err := json.Marshal(map[string]any{"runtime_sha256": fmt.Sprintf("%x", sha256.Sum256(source)), "observed_at": time.Now().UTC(), "resource_ids": f.ids, "frames": frames})
+	if err != nil {
+		t.Fatal(err)
+	}
+	t.Log("DISABLED_LOCAL_FRAMES=" + string(data))
+}
+
+func TestPVEDisabledLocalStorageAllSkippedRemainsUnavailable(t *testing.T) {
+	f, c := newDisabledLocalBackupFixture(t, "lxc")
+	f.m.pollStorageBackupsWithNodes(context.Background(), "cluster", c, f.nodes, f.online)
+	f.wait(t)
+	before, raw := f.points(t), f.m.state.GetSnapshot().PVEBackups.StorageBackups
+	c.disabled = map[string]bool{"node-a": true, "node-b": true}
+	c.contentQueries = make(map[string]int)
+	for range 2 {
+		f.m.pollStorageBackupsWithNodes(context.Background(), "cluster", c, f.nodes, f.online)
+		f.wait(t)
+		if !reflect.DeepEqual(before, f.points(t)) || !reflect.DeepEqual(raw, f.m.state.GetSnapshot().PVEBackups.StorageBackups) {
+			t.Error("all-skipped collection changed accepted facts")
+		}
+		f.assertHistory(t, recovery.ProtectionHistoryUnavailable, operationaltrust.EvidencePermissionsUnknown)
+	}
+	if c.configCalls != 0 || len(c.contentQueries) != 0 {
+		t.Error("unambiguously unread collection acquired unnecessary requests")
+	}
+}
+
+func TestPVEDisabledLocalStorageDoesNotChangeAnotherConnection(t *testing.T) {
+	f, c := newDisabledLocalBackupFixture(t, "qemu")
+	otherRaw := []models.StorageBackup{{ID: "other-cluster-local-file", Instance: "other-cluster", Node: "node-b", StorageNode: "node-b", Storage: "local", VMID: 102, Time: f.client.at, CTime: f.client.at.Unix()}}
+	otherPoint := recoveryIngestTestPoint("pve-backup:other-cluster-file", "other-cluster", f.client.at)
+	if err := f.store.UpsertPoints(context.Background(), []recovery.RecoveryPoint{otherPoint}); err != nil {
+		t.Fatal(err)
+	}
+	otherPoint = f.points(t)[otherPoint.ID]
+	f.m.state.UpdateStorageBackupsForInstance("other-cluster", otherRaw)
+	for _, disabled := range []bool{false, true, false} {
+		c.disabled = map[string]bool{"node-b": disabled}
+		f.m.pollStorageBackupsWithNodes(context.Background(), "cluster", c, f.nodes, f.online)
+		f.wait(t)
+		var gotOther []models.StorageBackup
+		for _, row := range f.m.state.GetSnapshot().PVEBackups.StorageBackups {
+			if row.Instance == "other-cluster" {
+				gotOther = append(gotOther, row)
+			}
+		}
+		if !reflect.DeepEqual(otherRaw, gotOther) || !reflect.DeepEqual(otherPoint, f.points(t)[otherPoint.ID]) {
+			t.Error("local skip crossed connection scope")
+		}
+	}
+}

internal/monitoring/pve_protection_observation_test.go

@@ -331,6 +331,17 @@ func TestPVEProtectionDisabledStorageIsNotAnEmptyRead(t *testing.T) {
 type nodeRestrictedBackupStorageClient struct {
 	*pveReconciliationClient
 	contentQueries map[string]int
+	configQueries  int
+}
+
+// Matching names alone do not prove membership: the datacenter catalogue
+// explicitly excludes each node from the other node's local storage.
+func (c *nodeRestrictedBackupStorageClient) GetAllStorage(context.Context) ([]proxmox.Storage, error) {
+	c.configQueries++
+	return []proxmox.Storage{
+		{Storage: "archive-node-a", Type: "dir", Content: "backup", Nodes: "node-a"},
+		{Storage: "archive-node-b", Type: "dir", Content: "backup", Nodes: "node-b"},
+	}, nil
 }
 
 func (c *nodeRestrictedBackupStorageClient) GetStorage(ctx context.Context, node string) ([]proxmox.Storage, error) {
@@ -364,6 +375,9 @@ func TestPVEProtectionNodeRestrictedStorageReadElsewhereIsComplete(t *testing.T)
 			if client.contentQueries["node-a/archive-node-b"] != 0 || client.contentQueries["node-b/archive-node-a"] != 0 {
 				t.Fatalf("storage was queried on a node it is not available on: %+v", client.contentQueries)
 			}
+			if client.configQueries != 1 {
+				t.Fatalf("ambiguous local scopes need one shared catalogue read, got %d", client.configQueries)
+			}
 			f.assertHistory(t, recovery.ProtectionHistoryComplete, operationaltrust.EvidencePermissionsSufficient)
 		})
 	}