From 9b77a4b6c6fafc8bd0015698d221b242bd2f4050 Mon Sep 17 00:00:00 2001 From: "pulse-triage[bot]" <249995291+pulse-triage[bot]@users.noreply.github.com> Date: Wed, 7 Oct 2026 19:39:32 +0100 Subject: [PATCH] Keep guest metadata fetch ownership through completion Honour cold-cache retry backoff and retain a per-key in-flight reservation until enrichment returns, rather than treating a timed hold as completion. Avoid duplicate command sequences and stale safety-state writers while preserving existing lock, uncertainty, observation and refresh policies. Cover blocked real HTTP reads, early deferral, slot cancellation and ordinary recovery. Contract-Neutral: Restore the existing per-key metadata reservation and retry scheduling semantics; no public API, configuration, guest command, lock, uncertainty fence, observation or numeric duration changes. Change-source: pulse-maintainer --- .../v6/internal/subsystems/monitoring.md | 17 ++ internal/monitoring/guest_metadata.go | 26 ++- .../guest_metadata_reservation_test.go | 205 ++++++++++++++++++ internal/monitoring/monitor.go | 1 + 4 files changed, 245 insertions(+), 4 deletions(-) create mode 100644 internal/monitoring/guest_metadata_reservation_test.go diff --git a/docs/release-control/v6/internal/subsystems/monitoring.md b/docs/release-control/v6/internal/subsystems/monitoring.md index 72365afee2..6aef8103c9 100644 --- a/docs/release-control/v6/internal/subsystems/monitoring.md +++ b/docs/release-control/v6/internal/subsystems/monitoring.md @@ -550,6 +550,23 @@ Local tests are not native freeze/thaw evidence; supplied platform/artifact provenance and mount completeness still need independent review. The tools add no runtime interface, recovery action or release gate. + +Metadata enrichment owns one fetch per existing `(instance, node, VMID)` cache +key until the whole fetch returns, including slot waiting and early deferral. +The configured hold remains retry timing, not proof that an in-flight fetch +completed. A cold cache cannot override another owner's reservation or an early +failure's backoff. A denied reservation returns only previously retained +identity (or none), without creating fresh evidence, a transport deferral or a +new retry deadline. Other keys remain independent; completion releases only +in-flight ownership and preserves normal refresh/backoff scheduling. + +`guest_metadata_reservation_test.go` exercises real HTTP enrichment with cold +and retained caches, active and expired scheduling holds, early unverified-lock +and cancelled-slot exits, independent VMs and ordinary resumption. The existing +shared Proxmox busy/cooldown fence, lock checks and no-replay policy are unchanged. +These source controls establish neither a reported freeze cause nor native +QEMU-only recovery. + ### Exact TrueNAS subscription termination — issue #2396 JSON-RPC stream readers and the subscription-acknowledgement wait recognise diff --git a/internal/monitoring/guest_metadata.go b/internal/monitoring/guest_metadata.go index ef2f8c8777..250526feee 100644 --- a/internal/monitoring/guest_metadata.go +++ b/internal/monitoring/guest_metadata.go @@ -97,6 +97,11 @@ func (m *Monitor) tryReserveGuestMetadataFetch(key string, now time.Time) bool { m.guestMetadataLimiterMu.Lock() defer m.guestMetadataLimiterMu.Unlock() + // A time-based hold limits retry rate, but is not completion evidence. + // Keep ownership until the fetch returns, even if its hold has expired. + if m.guestMetadataInFlight[key] { + return false + } if next, ok := m.guestMetadataLimiter[key]; ok && now.Before(next) { return false } @@ -104,10 +109,23 @@ func (m *Monitor) tryReserveGuestMetadataFetch(key string, now time.Time) bool { if hold <= 0 { hold = defaultGuestMetadataHold } + if m.guestMetadataLimiter == nil { + m.guestMetadataLimiter = make(map[string]time.Time) + } + if m.guestMetadataInFlight == nil { + m.guestMetadataInFlight = make(map[string]bool) + } m.guestMetadataLimiter[key] = now.Add(hold) + m.guestMetadataInFlight[key] = true return true } +func (m *Monitor) releaseGuestMetadataFetch(key string) { + m.guestMetadataLimiterMu.Lock() + delete(m.guestMetadataInFlight, key) + m.guestMetadataLimiterMu.Unlock() +} + func (m *Monitor) scheduleNextGuestMetadataFetch(key string, now time.Time) { if m == nil { return @@ -256,12 +274,12 @@ func (m *Monitor) fetchGuestAgentMetadata(ctx context.Context, client PVEClientI } reserved := m.tryReserveGuestMetadataFetch(key, now) - if !reserved && ok { + // An empty cache does not grant permission to bypass another fetch or the + // backoff from an early deferral. Return only existing identity, if any. + if !reserved { return cloneStringSlice(cached.ipAddresses), cloneGuestNetworkInterfaces(cached.networkInterfaces), cached.osName, cached.osVersion, cached.agentVersion, false } - if !reserved && !ok { - reserved = true - } + defer m.releaseGuestMetadataFetch(key) // Start with cached values as fallback in case new calls fail ipAddresses := cloneStringSlice(cached.ipAddresses) diff --git a/internal/monitoring/guest_metadata_reservation_test.go b/internal/monitoring/guest_metadata_reservation_test.go new file mode 100644 index 0000000000..832f6a549e --- /dev/null +++ b/internal/monitoring/guest_metadata_reservation_test.go @@ -0,0 +1,205 @@ +package monitoring + +import ( + "context" + "fmt" + "net/http" + "net/http/httptest" + "reflect" + "strings" + "sync/atomic" + "testing" + "time" + + "github.com/rcourtman/pulse-go-rewrite/pkg/proxmox" +) + +func metadataReservationClient(t *testing.T, handler http.Handler) *proxmox.Client { + t.Helper() + server := httptest.NewServer(handler) + t.Cleanup(server.Close) + client, err := proxmox.NewClient(proxmox.ClientConfig{Host: server.URL, TokenName: "fixture@pve!pulse", TokenValue: "fixture", Timeout: 5 * time.Second}) + if err != nil { + t.Fatal(err) + } + return client +} + +func metadataReservationPayload(w http.ResponseWriter, r *http.Request) { + switch { + case strings.HasSuffix(r.URL.Path, "/config"): + fmt.Fprint(w, `{"data":{}}`) + case strings.HasSuffix(r.URL.Path, "/network-get-interfaces"): + fmt.Fprint(w, `{"data":{"result":[{"name":"eth0","hardware-address":"02:00:00:00:00:01","ip-addresses":[{"ip-address":"192.0.2.20","prefix":24,"ip-address-type":"ipv4"}]}]}}`) + case strings.HasSuffix(r.URL.Path, "/get-osinfo"): + fmt.Fprint(w, `{"data":{"result":{"name":"FixtureOS","version":"2"}}}`) + case strings.HasSuffix(r.URL.Path, "/info"): + fmt.Fprint(w, `{"data":{"result":{"version":"2.0"}}}`) + default: + http.NotFound(w, r) + } +} + +func assertMetadataReservationReleased(t *testing.T, m *Monitor) { + t.Helper() + m.guestMetadataLimiterMu.Lock() + defer m.guestMetadataLimiterMu.Unlock() + if len(m.guestMetadataInFlight) != 0 { + t.Fatal("completed fetch retained in-flight ownership") + } +} + +func TestGuestMetadataReservationOwnsWholeHTTPFetch(t *testing.T) { + for _, seeded := range []bool{false, true} { + for _, expiredHold := range []bool{false, true} { + t.Run(fmt.Sprintf("cached-%t/hold-expired-%t", seeded, expiredHold), func(t *testing.T) { + key := guestMetadataCacheKey("reservation", "node", 105) + original := guestMetadataCacheEntry{} + m := &Monitor{guestMetadataLimiter: make(map[string]time.Time), guestMetadataSlots: make(chan struct{}, 2)} + if seeded { + original = metadataObservationFixture(time.Now().Add(-2 * guestMetadataCacheTTL)) + original.osInfoSkip = true + original.osInfoFailureCount = guestAgentOSInfoFailureThreshold + m.guestMetadataCache = map[string]guestMetadataCacheEntry{key: original} + } + entered, release := make(chan struct{}), make(chan struct{}) + var commands atomic.Int32 + client := metadataReservationClient(t, http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if strings.Contains(r.URL.Path, "/agent/") { + commands.Add(1) + } + if strings.Contains(r.URL.Path, "/qemu/105/") && strings.HasSuffix(r.URL.Path, "/network-get-interfaces") { + select { + case <-entered: + default: + close(entered) + } + <-release + } + metadataReservationPayload(w, r) + })) + status := &proxmox.VMStatus{Agent: proxmox.VMAgentField{Value: 1}} + done := make(chan bool, 1) + go func() { + _, _, _, _, _, deferred := m.fetchGuestAgentMetadata(context.Background(), client, "reservation", "node", "guest", 105, status, false) + done <- deferred + }() + t.Cleanup(func() { + select { + case <-release: + default: + close(release) + } + select { + case <-done: + case <-time.After(6 * time.Second): + t.Error("fetch did not finish during cleanup") + } + }) + select { + case <-entered: + case <-time.After(5 * time.Second): + t.Fatal("network command did not enter") + } + m.guestMetadataLimiterMu.Lock() + if expiredHold { + // Advance only test-owned scheduling state; do not shorten the + // Proxmox uncertainty fence or wait for a real guest timeout. + m.guestMetadataLimiter[key] = time.Now().Add(-time.Second) + } + before := m.guestMetadataLimiter[key] + m.guestMetadataLimiterMu.Unlock() + ips, ifaces, name, version, agent, deferred := m.fetchGuestAgentMetadata(context.Background(), client, "reservation", "node", "guest", 105, status, false) + if deferred || !reflect.DeepEqual(ips, original.ipAddresses) || !reflect.DeepEqual(ifaces, original.networkInterfaces) || name != original.osName || version != original.osVersion || agent != original.agentVersion { + t.Error("duplicate fetch invented a transport pause or changed retained identity") + } + m.guestMetadataLimiterMu.Lock() + unchanged := m.guestMetadataLimiter[key].Equal(before) + m.guestMetadataLimiterMu.Unlock() + if !unchanged || commands.Load() != 1 || len(m.guestMetadataSlots) != 1 { + t.Error("duplicate fetch changed retry timing, sent a command or consumed another slot") + } + // A different VM is not blocked by this metadata reservation. + _, _, _, _, otherAgent, otherDeferred := m.fetchGuestAgentMetadata(context.Background(), client, "reservation", "node", "other", 106, status, false) + if otherDeferred || otherAgent != "2.0" { + t.Error("reservation blocked another VM") + } + close(release) + select { + case deferred := <-done: + if deferred { + t.Error("owner did not complete normally") + } + // Leave a completed marker for the unconditional cleanup drain. + done <- false + case <-time.After(5 * time.Second): + t.Fatal("owner did not finish") + } + assertMetadataReservationReleased(t, m) + m.guestMetadataMu.Lock() + entry := m.guestMetadataCache[key] + entry.fetchedAt = time.Now().Add(-2 * guestMetadataCacheTTL) + m.guestMetadataCache[key] = entry + m.guestMetadataMu.Unlock() + m.guestMetadataLimiterMu.Lock() + m.guestMetadataLimiter[key] = time.Now().Add(-time.Second) + m.guestMetadataLimiterMu.Unlock() + _, _, _, _, agent, deferred = m.fetchGuestAgentMetadata(context.Background(), client, "reservation", "node", "guest", 105, status, false) + if deferred || agent != "2.0" || !m.hasRecentGuestMetadataEvidence("reservation", "node", 105, time.Now()) || len(m.guestMetadataSlots) != 0 { + t.Error("completed owner did not allow ordinary refresh") + } + assertMetadataReservationReleased(t, m) + }) + } + } +} + +func TestGuestMetadataColdCacheHonoursEarlyBackoff(t *testing.T) { + for _, mode := range []string{"unverified-lock", "cancelled-slot"} { + t.Run(mode, func(t *testing.T) { + key := guestMetadataCacheKey("backoff", "node", 105) + m := &Monitor{guestMetadataLimiter: make(map[string]time.Time), guestMetadataRetryBackoff: time.Minute} + var requests atomic.Int32 + var healthy atomic.Bool + client := metadataReservationClient(t, http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + requests.Add(1) + if !healthy.Load() && strings.HasSuffix(r.URL.Path, "/config") { + fmt.Fprint(w, `{"data":null}`) + return + } + metadataReservationPayload(w, r) + })) + status := &proxmox.VMStatus{Agent: proxmox.VMAgentField{Value: 1}} + ctx := context.Background() + if mode == "cancelled-slot" { + m.guestMetadataSlots = make(chan struct{}, 1) + m.guestMetadataSlots <- struct{}{} + cancelled, cancel := context.WithCancel(ctx) + cancel() + ctx = cancelled + } + _, _, _, _, _, deferred := m.fetchGuestAgentMetadata(ctx, client, "backoff", "node", "guest", 105, status, false) + if deferred != (mode == "unverified-lock") || len(m.guestMetadataCache) != 0 { + t.Fatal("early exit fabricated metadata or lost observed deferral") + } + assertMetadataReservationReleased(t, m) + if mode == "cancelled-slot" { + <-m.guestMetadataSlots + } + before, next := requests.Load(), m.guestMetadataLimiter[key] + ips, ifaces, name, version, agent, deferred := m.fetchGuestAgentMetadata(context.Background(), client, "backoff", "node", "guest", 105, status, false) + if requests.Load() != before || !m.guestMetadataLimiter[key].Equal(next) || len(m.guestMetadataCache) != 0 || len(ips)+len(ifaces) != 0 || name+version+agent != "" || deferred { + t.Error("cold cache bypassed retry backoff or invented guest evidence") + } + // No metadata from another installation can satisfy this key. + m.guestMetadataCache = map[string]guestMetadataCacheEntry{guestMetadataCacheKey("independent", "node", 105): {osName: "independent"}} + healthy.Store(true) + m.guestMetadataLimiter[key] = time.Now().Add(-time.Second) + _, _, _, _, agent, deferred = m.fetchGuestAgentMetadata(context.Background(), client, "backoff", "node", "guest", 105, status, false) + if deferred || agent != "2.0" || !m.hasRecentGuestMetadataEvidence("backoff", "node", 105, time.Now()) || !reflect.DeepEqual(m.guestMetadataCache[guestMetadataCacheKey("independent", "node", 105)], guestMetadataCacheEntry{osName: "independent"}) { + t.Error("healthy cold cache did not resume independently after backoff") + } + assertMetadataReservationReleased(t, m) + }) + } +} diff --git a/internal/monitoring/monitor.go b/internal/monitoring/monitor.go index 28744f4953..b651fd9444 100644 --- a/internal/monitoring/monitor.go +++ b/internal/monitoring/monitor.go @@ -1300,6 +1300,7 @@ type Monitor struct { guestMetadataCache map[string]guestMetadataCacheEntry guestMetadataLimiterMu sync.Mutex guestMetadataLimiter map[string]time.Time + guestMetadataInFlight map[string]bool // Guarded by guestMetadataLimiterMu; released only on fetch completion. guestMetadataSlots chan struct{} guestMetadataMinRefresh time.Duration guestMetadataRefreshJitter time.Duration