Judge source merges by the ingest's stale thresholds

mergeInto always passed nil thresholds to mergeMetrics, so every source
merge into an existing row judged metric freshness by the registry
defaults while the stale pass, the manual links and the monitor's
configuration used the caller's thresholds. A Proxmox node polled every
two minutes (four-minute threshold) merged with its auto-linked agent
showed the silent agent's readings while the stale pass called the
node's poll current, and a supplemental refresh of a linked vSphere VM
kept the in-guest agent's memory over vSphere's current reading.

Snapshot and record ingest now hold the caller's thresholds on the
registry while they run, the way the guest loop holds its agent index,
and mergeInto reads them. The ~50 snapshot helpers that reach
mergeInto through ingest() keep their signatures. The stale pass and
the metric merge now share one threshold rule, sourceStaleThreshold,
so the merge also honours a sighting's expected update interval (slow
Proxmox disk inventory) as the stale pass already did, and a source
without a default falls back to 120s in both (the merge used 60s).
Every defined source has a default, so that fallback reaches none.

Outside mock mode the monitor configures only Proxmox, PBS and PMG,
so those are the production sources this changes; mock mode also
configures TrueNAS, vSphere and availability.

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

@@ -2679,10 +2679,12 @@ truthfulness, not native thaw, containing-release or workload acceptance.
     snapshot-bridge adapters stay read-only. The rebuild, the live
     supplemental refresh and the read-state overlay ingest records with the
     adapter's configured stale thresholds, because record ingest joins
-    operator links and a link's metric merge judges each side's freshness
-    by them (unified-resources contract, "Operator links reach
-    record-ingested resources"). Regression coverage:
-    `TestMonitorAdapterJoinsLinkedRecordsWithConfiguredStaleThresholds` in
+    operator links and the freshness gate of every metric merge, a link's or
+    a source's into an existing row, reads them (unified-resources contract,
+    "Operator links reach record-ingested resources"). Regression
+    coverage:
+    `TestMonitorAdapterJoinsLinkedRecordsWithConfiguredStaleThresholds` and
+    `TestMonitorAdapterSourceMergesUseConfiguredStaleThresholds` in
     `internal/unifiedresources/monitor_adapter_read_state_test.go` and
     `TestManualLinkToSupplementalGuestHoldsWithAndWithoutContinuity` in
     `internal/monitoring/issue1913_host_continuity_test.go`.

docs/release-control/v6/internal/subsystems/unified-resources.md

@@ -1609,7 +1609,12 @@ Source freshness is a separate unified-resource status input, not a fixed
 global health timer. Snapshot rebuilds, resource seeding, supplemental record
 overlays, and cloned read-state overlays must preserve the monitoring-provided
 stale thresholds for Proxmox, PBS, and PMG sources so resources do not flap to
-warning/degraded between successful configured poll cycles.
+warning/degraded between successful configured poll cycles. A source merge
+into an existing row judges metric freshness by the ingest's thresholds and
+the stale pass's own threshold rule, so the merge's freshness gate never
+calls a source current that the stale pass marks stale, or the reverse. The
+gate decides only where no stronger rule does: a hypervisor-managed guest
+keeps its platform CPU even while that source is stale.
 
 Service-discovery readiness is a unified-resource payload contract, not a
 drawer-local decoration. Resource list/detail payloads that expose a
@@ -6556,8 +6561,21 @@ separate.
 
 - The monitor adapter passes its configured stale thresholds to record
   ingest, so a link joined there judges each side's metrics the way a
-  snapshot-time link does. Ordinary source merges into an existing row still
-  use the default thresholds on every ingest path.
+  snapshot-time link does. Ordinary source merges into an existing row use
+  the same thresholds on every ingest path: snapshot and record ingest hold
+  the caller's thresholds while they run, and `mergeInto` judges each
+  metric's freshness with them through `sourceStaleThreshold`, the rule the
+  stale pass reads. Before this, those merges used the defaults, so a
+  Proxmox node polled every two minutes showed a silent auto-linked agent's
+  readings while the stale pass called the node's poll current, and a
+  supplemental refresh of a linked vSphere VM kept the agent's memory over
+  vSphere's current reading. Outside mock mode the monitor configures only
+  Proxmox, PBS and PMG; mock mode also configures TrueNAS, vSphere and
+  availability. Regression coverage:
+  `TestMonitorAdapterSourceMergesUseConfiguredStaleThresholds` (rebuild,
+  supplemental refresh, overlay, node with auto-linked agent),
+  `TestSourceMergesJudgeFreshnessByTheIngestStaleThresholds` and
+  `TestMetricMergeFreshnessMatchesTheStalePass`.
 - Continuity records are the one record ingest that never joins links. A
   saved enrollment only fills an absent machine, so it stays its own row
   instead of lending its linked guest an offline verdict and an old agent

internal/unifiedresources/monitor_adapter_read_state_test.go

@@ -481,6 +481,153 @@ func TestMonitorAdapterJoinsLinkedRecordsWithConfiguredStaleThresholds(t *testin
 	}
 }
 
+// Every source merge into an existing row judges metric freshness by the
+// adapter's configured thresholds, as the rebuild's links and stale pass do.
+// A linked vSphere VM first reports no memory, so its in-guest agent supplies
+// it; a refresh then brings vSphere memory observed three minutes ago, current
+// under a five-minute threshold. The live supplemental refresh and the
+// read-state overlay merge that record into the existing row, and must show
+// vSphere's memory as a rebuild from the same records does. A Proxmox node
+// polled every two minutes is likewise current at ninety seconds, so the
+// rebuild keeps its readings over those of its auto-linked agent, silent for
+// seventy-five seconds.
+func TestMonitorAdapterSourceMergesUseConfiguredStaleThresholds(t *testing.T) {
+	now := time.Now().UTC()
+	thresholds := map[DataSource]time.Duration{SourceProxmox: 4 * time.Minute, SourceVMware: 5 * time.Minute}
+
+	t.Run("linked vSphere VM refresh", func(t *testing.T) {
+		snapshot := models.StateSnapshot{
+			Hosts: []models.Host{{
+				ID:        "host-app-guest",
+				Hostname:  "app-guest",
+				MachineID: "machine-app-guest",
+				Status:    "online",
+				LastSeen:  now,
+				Memory:    models.Memory{Total: 8 << 30, Used: 6 << 30, Free: 2 << 30, Usage: 75},
+			}},
+			LastUpdate: now,
+		}
+		vmRecord := func(lastSeen time.Time, metrics *ResourceMetrics) []IngestRecord {
+			return []IngestRecord{{
+				SourceID: "vc-1:vm:vm-42",
+				Resource: Resource{
+					Type:       ResourceTypeVM,
+					Technology: "vmware",
+					Name:       "app-guest",
+					Status:     StatusOnline,
+					LastSeen:   lastSeen,
+					Metrics:    metrics,
+					VMware:     &VMwareData{ConnectionID: "vc-1", ManagedObjectID: "vm-42", EntityType: "vm"},
+				},
+			}}
+		}
+		initial := vmRecord(now.Add(-4*time.Minute), &ResourceMetrics{CPU: &MetricValue{Percent: 10, Source: SourceVMware}})
+		refreshedAt := now.Add(-3 * time.Minute)
+		refreshed := vmRecord(refreshedAt, &ResourceMetrics{
+			CPU:    &MetricValue{Percent: 10, Source: SourceVMware},
+			Memory: &MetricValue{Percent: 40, Source: SourceVMware},
+		})
+
+		unlinked := NewMonitorAdapter(NewRegistry(nil))
+		unlinked.PopulateSnapshotAndSupplemental(snapshot, map[DataSource][]IngestRecord{SourceVMware: initial})
+		vms, hosts := unlinked.VMs(), unlinked.Hosts()
+		if len(vms) != 1 || len(hosts) != 1 {
+			t.Fatalf("unlinked estate = %d VMs, %d agents, want one of each", len(vms), len(hosts))
+		}
+		link := ResourceLink{ResourceA: vms[0].ID(), ResourceB: hosts[0].ID(), PrimaryID: hosts[0].ID()}
+
+		for _, path := range []struct {
+			name string
+			read func(adapter *MonitorAdapter) ReadState
+		}{
+			{"rebuild", func(adapter *MonitorAdapter) ReadState {
+				adapter.PopulateSnapshotAndSupplemental(snapshot, map[DataSource][]IngestRecord{SourceVMware: refreshed})
+				return adapter
+			}},
+			{"supplemental", func(adapter *MonitorAdapter) ReadState {
+				adapter.PopulateSnapshotAndSupplemental(snapshot, map[DataSource][]IngestRecord{SourceVMware: initial})
+				adapter.PopulateSupplementalRecords(SourceVMware, refreshed)
+				return adapter
+			}},
+			{"overlay", func(adapter *MonitorAdapter) ReadState {
+				adapter.PopulateSnapshotAndSupplemental(snapshot, map[DataSource][]IngestRecord{SourceVMware: initial})
+				return ReadStateWithRecords(adapter, SourceVMware, refreshed)
+			}},
+		} {
+			t.Run(path.name, func(t *testing.T) {
+				store := NewMemoryStore()
+				if err := store.AddLink(link); err != nil {
+					t.Fatalf("add link: %v", err)
+				}
+				readState := path.read(NewMonitorAdapterWithStaleThresholds(NewRegistry(store), thresholds))
+
+				vms := readState.VMs()
+				if len(vms) != 1 || vms[0].ID() != link.ResourceA || len(readState.Hosts()) != 0 {
+					t.Fatalf("estate = %d VMs, %d standalone agents, want only the linked VM %s", len(vms), len(readState.Hosts()), link.ResourceA)
+				}
+				if got := vms[0].r.SourceStatus[SourceVMware]; got.Status != "online" || !got.LastSeen.Equal(refreshedAt) {
+					t.Fatalf("vSphere sighting = %+v, want the refresh, current under the configured threshold", got)
+				}
+				if vms[0].r.Metrics == nil || vms[0].r.Metrics.Memory == nil {
+					t.Fatal("linked VM lost its memory reading")
+				}
+				if memory := vms[0].r.Metrics.Memory; memory.Source != SourceVMware || memory.Percent != 40 {
+					t.Fatalf("linked VM memory = %.0f%% from %s, want vSphere's current 40%%", memory.Percent, memory.Source)
+				}
+			})
+		}
+	})
+
+	t.Run("node with auto-linked agent", func(t *testing.T) {
+		adapter := NewMonitorAdapterWithStaleThresholds(NewRegistry(nil), thresholds)
+		adapter.PopulateFromSnapshot(models.StateSnapshot{
+			Nodes: []models.Node{{
+				ID:            "homelab-pve1",
+				Name:          "pve1",
+				Instance:      "homelab",
+				Status:        "online",
+				LinkedAgentID: "host-pve1",
+				CPU:           0.12,
+				Memory:        models.Memory{Total: 64 << 30, Used: 16 << 30, Free: 48 << 30, Usage: 25},
+				LastSeen:      now.Add(-90 * time.Second),
+			}},
+			Hosts: []models.Host{{
+				ID:              "host-pve1",
+				MachineID:       "machine-pve1",
+				Hostname:        "pve1",
+				LinkedNodeID:    "homelab-pve1",
+				Status:          "online",
+				CPUUsage:        80,
+				Memory:          models.Memory{Total: 64 << 30, Used: 48 << 30, Free: 16 << 30, Usage: 75},
+				LastSeen:        now.Add(-75 * time.Second),
+				IntervalSeconds: 30,
+			}},
+			LastUpdate: now,
+		})
+
+		nodes := adapter.Nodes()
+		if len(nodes) != 1 {
+			t.Fatalf("nodes = %d, want the node merged with its agent", len(nodes))
+		}
+		node := nodes[0].r
+		if poll := node.SourceStatus[SourceProxmox]; poll.Status != "online" || !poll.LastSeen.Equal(now.Add(-90*time.Second)) {
+			t.Fatalf("Proxmox sighting = %+v, want the current poll from ninety seconds ago", poll)
+		}
+		if got := node.SourceStatus[SourceAgent].Status; got != "stale" {
+			t.Fatalf("agent sighting = %q, want stale", got)
+		}
+		if node.Metrics == nil || node.Metrics.CPU == nil || node.Metrics.Memory == nil {
+			t.Fatalf("merged node lost its readings: %+v", node.Metrics)
+		}
+		if cpu := node.Metrics.CPU; cpu.Source != SourceProxmox || cpu.Percent != 12 {
+			t.Fatalf("node CPU = %.0f%% from %s, want the current poll's 12%%", cpu.Percent, cpu.Source)
+		}
+		if memory := node.Metrics.Memory; memory.Source != SourceProxmox || memory.Percent != 25 {
+			t.Fatalf("node memory = %.0f%% from %s, want the current poll's 25%%", memory.Percent, memory.Source)
+		}
+	})
+}
+
 func TestHostContinuityCannotOverwriteCurrentResource(t *testing.T) {
 	now := time.Now().UTC()
 	for _, mode := range []string{"machine-identity", "source-identity", "future-timestamp", "provider-only"} {

internal/unifiedresources/registry.go

@@ -152,6 +152,13 @@ type ResourceRegistry struct {
 	// what the full scan would see; scores stay live through the shared
 	// pointers.
 	agentNodeScanIndex map[string][]agentNodeCandidate
+
+	// ingestStaleThresholds holds the caller-owned freshness thresholds of
+	// the snapshot or record ingest in progress, so each source merge into an
+	// existing row judges metric freshness as that ingest's manual links and
+	// stale pass do. nil outside an ingest and for callers without
+	// thresholds, where all three use the defaults.
+	ingestStaleThresholds map[DataSource]time.Duration
 }
 
 // agentNodeCandidate pairs a resources-map key with its entry so the indexed
@@ -247,6 +254,10 @@ func (rr *ResourceRegistry) IngestSnapshotWithStaleThresholds(snapshot models.St
 }
 
 func (rr *ResourceRegistry) ingestSnapshot(snapshot models.StateSnapshot, thresholds map[DataSource]time.Duration) {
+	rr.mu.Lock()
+	rr.ingestStaleThresholds = thresholds
+	rr.mu.Unlock()
+
 	hostByID := make(map[string]*models.Host, len(snapshot.Hosts))
 	for i := range snapshot.Hosts {
 		host := snapshot.Hosts[i]
@@ -610,6 +621,7 @@ func (rr *ResourceRegistry) ingestSnapshot(snapshot models.StateSnapshot, thresh
 	rr.refreshCanonicalIdentitiesLocked()
 	rr.markStaleLocked(time.Now().UTC(), thresholds)
 	rr.invalidateViewsLocked()
+	rr.ingestStaleThresholds = nil
 	rr.mu.Unlock()
 }
 
@@ -625,6 +637,10 @@ func (rr *ResourceRegistry) IngestRecordsWithStaleThresholds(source DataSource,
 }
 
 func (rr *ResourceRegistry) ingestRecords(source DataSource, records []IngestRecord, onlyMissing bool, thresholds map[DataSource]time.Duration) {
+	rr.mu.Lock()
+	rr.ingestStaleThresholds = thresholds
+	rr.mu.Unlock()
+
 	var successions []CanonicalIDSuccession
 	supersededSeen := make(map[string]struct{})
 	for _, record := range records {
@@ -677,6 +693,7 @@ func (rr *ResourceRegistry) ingestRecords(source DataSource, records []IngestRec
 	rr.refreshProxmoxSensorSetupLocked()
 	rr.refreshCanonicalIdentitiesLocked()
 	rr.invalidateViewsLocked()
+	rr.ingestStaleThresholds = nil
 	rr.mu.Unlock()
 }
 
@@ -1805,26 +1822,15 @@ func sourceVerdictsRecorded(sightings map[DataSource]SourceStatus) bool {
 }
 
 func (rr *ResourceRegistry) markStaleLocked(now time.Time, thresholds map[DataSource]time.Duration) {
-	thresholds = effectiveStaleThresholds(thresholds)
-
 	changed := false
 	for _, resource := range rr.resources {
 		previousStatus := resource.Status
 		staleFound := false
 		for source, status := range resource.SourceStatus {
-			threshold, ok := thresholds[source]
-			if !ok {
-				threshold = 120 * time.Second
-			}
-			if status.ExpectedUpdateIntervalSeconds > 0 {
-				// Slow inventory polls have their own cadence. A source must miss
-				// two expected intervals before its retained observation is stale.
-				threshold = max(threshold, 2*time.Duration(status.ExpectedUpdateIntervalSeconds)*time.Second)
-			}
 			if status.LastSeen.IsZero() {
 				continue
 			}
-			if now.Sub(status.LastSeen) > threshold {
+			if now.Sub(status.LastSeen) > sourceStaleThreshold(source, status, thresholds) {
 				changed = changed || status.Status != "stale"
 				status.Status = "stale"
 				resource.SourceStatus[source] = status
@@ -3973,7 +3979,7 @@ func (rr *ResourceRegistry) mergeInto(existing *Resource, incoming Resource, sou
 	existing.ParentID = rr.resolveCanonicalParentID(existing)
 
 	existing.Status = chooseStatus(existing.Status, incoming.Status, source, existing.Sources)
-	existing.Metrics = mergeMetrics(existing, existing.Metrics, incoming.Metrics, source, now, existing.SourceStatus, nil)
+	existing.Metrics = mergeMetrics(existing, existing.Metrics, incoming.Metrics, source, now, existing.SourceStatus, rr.ingestStaleThresholds)
 	existing.Metrics = clearUnavailableSourceMemoryMetric(existing.Metrics, &incoming, source)
 
 	// Prefer agent naming when available
@@ -6327,6 +6333,27 @@ func mergeMetrics(
 	return &merged
 }
 
+// sourceStaleThreshold is how long a source may go without delivering before
+// its sighting is stale: the caller's threshold for the source, else the
+// default. The stale pass and the metric merge's freshness gate both read it,
+// so the gate never calls a source current that the pass marks stale, or the
+// reverse.
+func sourceStaleThreshold(source DataSource, sighting SourceStatus, thresholds map[DataSource]time.Duration) time.Duration {
+	threshold := thresholds[source]
+	if threshold <= 0 {
+		threshold = defaultStaleThresholds[source]
+	}
+	if threshold <= 0 {
+		threshold = 120 * time.Second
+	}
+	if sighting.ExpectedUpdateIntervalSeconds > 0 {
+		// Slow inventory polls have their own cadence. A source must miss
+		// two expected intervals before its retained observation is stale.
+		threshold = max(threshold, 2*time.Duration(sighting.ExpectedUpdateIntervalSeconds)*time.Second)
+	}
+	return threshold
+}
+
 // metricSourceStale reports whether a source's most recent report is older than
 // its stale threshold. A zero/unknown last-seen is treated as NOT stale so the
 // merge never demotes a source on missing information.
@@ -6343,17 +6370,7 @@ func metricSourceStale(
 	if !ok || st.LastSeen.IsZero() {
 		return false
 	}
-	threshold := time.Duration(0)
-	if configured := thresholds[source]; configured > 0 {
-		threshold = configured
-	}
-	if threshold <= 0 {
-		threshold = defaultStaleThresholds[source]
-	}
-	if threshold <= 0 {
-		threshold = 60 * time.Second
-	}
-	return now.Sub(st.LastSeen) > threshold
+	return now.Sub(st.LastSeen) > sourceStaleThreshold(source, st, thresholds)
 }
 
 func mergeMetric(

internal/unifiedresources/registry_merge_policy_test.go

@@ -247,6 +247,114 @@ func TestLeaseExpiredAgentOnLiveProxmoxNodeStaysOnline(t *testing.T) {
 	}
 }
 
+// A source merge into an existing row judges metric freshness by the ingest's
+// stale thresholds, as the stale pass that ends the ingest does. With PVE
+// polled every two minutes the monitor allows Proxmox four minutes, so a node
+// polled ninety seconds ago is current while its agent, silent for seventy-five
+// seconds, is stale, and the row shows the node's readings. Without thresholds
+// both sources are stale by the defaults and the agent keeps its precedence.
+func TestSourceMergesJudgeFreshnessByTheIngestStaleThresholds(t *testing.T) {
+	now := time.Now().UTC()
+	snapshot := models.StateSnapshot{
+		Nodes: []models.Node{{
+			ID:            "homelab-pve1",
+			Name:          "pve1",
+			Instance:      "homelab",
+			Status:        "online",
+			LinkedAgentID: "host-pve1",
+			CPU:           0.12,
+			Memory:        models.Memory{Total: 64 << 30, Used: 16 << 30, Free: 48 << 30, Usage: 25},
+			LastSeen:      now.Add(-90 * time.Second),
+		}},
+		Hosts: []models.Host{{
+			ID:              "host-pve1",
+			MachineID:       "machine-pve1",
+			Hostname:        "pve1",
+			LinkedNodeID:    "homelab-pve1",
+			Status:          "online",
+			CPUUsage:        80,
+			Memory:          models.Memory{Total: 64 << 30, Used: 48 << 30, Free: 16 << 30, Usage: 75},
+			LastSeen:        now.Add(-75 * time.Second),
+			IntervalSeconds: 30,
+		}},
+	}
+
+	for _, tc := range []struct {
+		name        string
+		thresholds  map[DataSource]time.Duration
+		proxmoxPoll string
+		want        DataSource
+		wantCPU     float64
+		wantMemory  float64
+	}{
+		{"configured", map[DataSource]time.Duration{SourceProxmox: 4 * time.Minute}, "online", SourceProxmox, 12, 25},
+		{"defaults", nil, "stale", SourceAgent, 80, 75},
+	} {
+		t.Run(tc.name, func(t *testing.T) {
+			rr := NewRegistry(nil)
+			rr.IngestSnapshotWithStaleThresholds(snapshot, tc.thresholds)
+
+			resource := onlyResourceOfType(t, rr, ResourceTypeAgent)
+			if got := resource.SourceStatus[SourceProxmox].Status; got != tc.proxmoxPoll {
+				t.Fatalf("Proxmox sighting = %q, want %q: %+v", got, tc.proxmoxPoll, resource.SourceStatus)
+			}
+			if got := resource.SourceStatus[SourceAgent].Status; got != "stale" {
+				t.Fatalf("agent sighting = %q, want stale", got)
+			}
+			if resource.Metrics == nil || resource.Metrics.CPU == nil || resource.Metrics.Memory == nil {
+				t.Fatalf("merged row lost its readings: %+v", resource.Metrics)
+			}
+			if cpu := resource.Metrics.CPU; cpu.Source != tc.want || cpu.Percent != tc.wantCPU {
+				t.Fatalf("CPU = %.0f%% from %s, want %.0f%% from %s", cpu.Percent, cpu.Source, tc.wantCPU, tc.want)
+			}
+			if memory := resource.Metrics.Memory; memory.Source != tc.want || memory.Percent != tc.wantMemory {
+				t.Fatalf("memory = %.0f%% from %s, want %.0f%% from %s", memory.Percent, memory.Source, tc.wantMemory, tc.want)
+			}
+		})
+	}
+}
+
+// The metric merge and the stale pass read one threshold rule, so a source
+// sighting the merge treats as current is never one the pass marks stale.
+func TestMetricMergeFreshnessMatchesTheStalePass(t *testing.T) {
+	now := time.Now().UTC()
+	configured := map[DataSource]time.Duration{SourceProxmox: 4 * time.Minute, SourceVMware: 5 * time.Minute}
+	for _, tc := range []struct {
+		name       string
+		source     DataSource
+		sighting   SourceStatus
+		thresholds map[DataSource]time.Duration
+		wantStale  bool
+	}{
+		{"proxmox by default", SourceProxmox, SourceStatus{LastSeen: now.Add(-90 * time.Second)}, nil, true},
+		{"proxmox by configuration", SourceProxmox, SourceStatus{LastSeen: now.Add(-90 * time.Second)}, configured, false},
+		{"vsphere by default", SourceVMware, SourceStatus{LastSeen: now.Add(-3 * time.Minute)}, nil, true},
+		{"vsphere by configuration", SourceVMware, SourceStatus{LastSeen: now.Add(-3 * time.Minute)}, configured, false},
+		{"agent without its own threshold", SourceAgent, SourceStatus{LastSeen: now.Add(-75 * time.Second)}, configured, true},
+		{"slow disk inventory", SourceProxmox, SourceStatus{LastSeen: now.Add(-5 * time.Minute), ExpectedUpdateIntervalSeconds: 600}, nil, false},
+		{"missed two disk inventory polls", SourceProxmox, SourceStatus{LastSeen: now.Add(-21 * time.Minute), ExpectedUpdateIntervalSeconds: 600}, nil, true},
+	} {
+		t.Run(tc.name, func(t *testing.T) {
+			status := map[DataSource]SourceStatus{tc.source: tc.sighting}
+			if got := metricSourceStale(now, status, tc.source, tc.thresholds); got != tc.wantStale {
+				t.Fatalf("metric merge stale = %v, want %v", got, tc.wantStale)
+			}
+
+			rr := NewRegistry(nil)
+			rr.resources["resource"] = &Resource{
+				ID:           "resource",
+				Status:       StatusOnline,
+				Sources:      []DataSource{tc.source},
+				SourceStatus: map[DataSource]SourceStatus{tc.source: tc.sighting},
+			}
+			rr.MarkStale(now, tc.thresholds)
+			if got := rr.resources["resource"].SourceStatus[tc.source].Status == "stale"; got != tc.wantStale {
+				t.Fatalf("stale pass stale = %v, want %v", got, tc.wantStale)
+			}
+		})
+	}
+}
+
 // Docker hosts hold a shorter lease than the registry's Docker stale
 // threshold, so a silent Docker host was offline for a while and then
 // flipped to warning once its sighting went stale.