Integrate reviewed complete-response guest lock checks

Accept the exact Core candidate and its retained parent/final race evidence. Reject incomplete authoritative VM configuration bodies before guest-agent dispatch and fresh telemetry publication; preserve existing canonical frontend and documentation work. Native PVE/HAOS-QGA thaw acceptance and containing-release follow-through remain separate.

Change-source: pulse-maintainer

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

@@ -44,6 +44,15 @@ backup), missing/malformed config or an unverifiable lock defers the command.
 Known locks in node inventory, cluster resources or detailed status suppress
 polling enrichment even when status fails and recent agent evidence exists.
 
+The VM config body must finish within the existing response-size and request
+bounds and contain exactly one complete JSON value. An unlocked JSON prefix
+does not establish lock clearance: a short Content-Length body, stalled suffix,
+extra JSON/garbage or chunked body exceeding 8 MiB yields `lock-unverified` both
+before dispatch and after a completed guest command. Post-command verification
+holds per-guest admission until body completion; its failure discards the
+payload. Complete responses at the limit, including trailing whitespace, remain
+valid. Ordinary PVE CPU/I/O observations remain independent of this deferral.
+
 Admission is held through the complete response, not just headers. A transport
 failure, timeout, wrong-command-ID response or incomplete body defers every
 command for that guest for a minute; callers neither immediately retry nor
@@ -63,6 +72,12 @@ resource lock with failed status → node fallback → unlocked resumption throu
 the real client/builders and History. The client controls cover every read,
 unknown lock state, diagnostic overlap, aliases/source/VM isolation, in-flight
 lock changes, incomplete bodies, cooldown and absence of failover replay.
+`TestGuestAgentConfigResponseMustBeComplete` covers every guest read before and
+after dispatch and subsequent healthy resumption; the stalled-prefix and
+post-config-body controls pin completion and cross-client admission.
+`TestUnverifiedVMConfigPreservesStatusCounterObservations` and the monitoring
+lifecycle pin live counter presence and rejection of cached disk/memory History
+refresh during an incomplete lock read, not just a decoder error.
 These reproduce missing Pulse safety controls, **not** the reported native
 PVE/QGA command-ID collision or successful backup thaw. HTTP lock checks cannot
 atomically reserve PVE's serial channel against a backup beginning afterwards,

internal/monitoring/guest_agent_backup_integration_test.go

@@ -26,7 +26,10 @@ func testGuestAgentBackupMonitoringLifecycle(t *testing.T) {
 		w.Header().Set("Content-Type", "application/json")
 		switch {
 		case strings.HasSuffix(r.URL.Path, "/config"):
-			if p == 1 || p == 2 {
+			if p == 4 {
+				w.Header().Set("Content-Length", "1000")
+				fmt.Fprint(w, `{"data":{}}`)
+			} else if p == 1 || p == 2 {
 				fmt.Fprint(w, `{"data":{"lock":"backup"}}`)
 			} else {
 				fmt.Fprint(w, `{"data":{}}`)
@@ -98,6 +101,22 @@ func testGuestAgentBackupMonitoringLifecycle(t *testing.T) {
 	memKey := guestMemoryCacheKey("fixture", "node", 105)
 	originalMetadata := m.guestMetadataCache[metadataKey]
 	originalMemory := m.vmAgentMemCache[memKey]
+	// A syntactically unlocked prefix is not a completed config observation.
+	// Warm guest caches must stay labelled last-known rather than becoming new
+	// disk/memory History samples when the lock response is incomplete.
+	phase.Store(4)
+	unverified, _ := build()
+	if unverified.ID != initial.ID || unverified.GuestAgentStatus != "deferred" || unverified.DiskStatusReason != "prev-lock-unverified" || unverified.Disk.Used != initial.Disk.Used || unverified.Memory.Used != initial.Memory.Used {
+		t.Fatalf("incomplete lock response lost truthful continuity: %#v", unverified)
+	}
+	if calls.Load() != before || !reflect.DeepEqual(m.guestMetadataCache[metadataKey], originalMetadata) || !reflect.DeepEqual(m.vmAgentMemCache[memKey], originalMemory) {
+		t.Fatal("incomplete lock response sent a guest command or renewed caches")
+	}
+	for metric, want := range map[string]int{"cpu": 2, "memory": 1, "memoryused": 1, "disk": 1} {
+		if got := len(m.metricsHistory.GetGuestMetrics(identity, metric, time.Hour)); got != want {
+			t.Errorf("incomplete lock %s history points = %d, want %d", metric, got, want)
+		}
+	}
 	phase.Store(1)
 	res.CPU = 0.2
 	locked, source := build()
@@ -113,7 +132,7 @@ func testGuestAgentBackupMonitoringLifecycle(t *testing.T) {
 	if !reflect.DeepEqual(m.guestMetadataCache[metadataKey], originalMetadata) || !reflect.DeepEqual(m.vmAgentMemCache[memKey], originalMemory) {
 		t.Fatal("lock renewed/replaced cached observations")
 	}
-	for metric, want := range map[string]int{"cpu": 2, "memory": 1, "memoryused": 1, "disk": 1} {
+	for metric, want := range map[string]int{"cpu": 3, "memory": 1, "memoryused": 1, "disk": 1} {
 		if got := len(m.metricsHistory.GetGuestMetrics(identity, metric, time.Hour)); got != want {
 			t.Errorf("%s history points = %d, want %d", metric, got, want)
 		}

pkg/proxmox/client.go

@@ -1596,11 +1596,18 @@ func (c *Client) GetVMConfig(ctx context.Context, node string, vmid int) (map[st
 	}
 	defer resp.Body.Close()
 
+	// Guest-agent admission uses this response as authoritative lock evidence.
+	// Decoding only the first JSON value can accept an unlocked prefix before a
+	// truncated, stalled, oversized or malformed response has been received.
+	body, err := readResponseBodyLimited(resp.Body)
+	if err != nil {
+		return nil, fmt.Errorf("read VM configuration: %w", err)
+	}
 	var result struct {
 		Data map[string]interface{} `json:"data"`
 	}
 
-	if err := json.NewDecoder(resp.Body).Decode(&result); err != nil {
+	if err := json.Unmarshal(body, &result); err != nil {
 		return nil, err
 	}
 

pkg/proxmox/guest_agent_lock_response_test.go

@@ -0,0 +1,164 @@
+package proxmox
+
+import (
+	"context"
+	"fmt"
+	"net/http"
+	"net/http/httptest"
+	"strings"
+	"sync/atomic"
+	"testing"
+	"time"
+)
+
+func TestGuestAgentConfigResponseMustBeComplete(t *testing.T) {
+	for _, test := range []struct {
+		name  string
+		write func(http.ResponseWriter)
+	}{
+		{"truncated", func(w http.ResponseWriter) {
+			w.Header().Set("Content-Length", "1000")
+			fmt.Fprint(w, `{"data":{}}`)
+		}},
+		{"trailing-json", func(w http.ResponseWriter) {
+			fmt.Fprint(w, `{"data":{}}{"data":{"lock":"backup"}}`)
+		}},
+		{"trailing-garbage", func(w http.ResponseWriter) {
+			fmt.Fprint(w, `{"data":{}}incomplete-response`)
+		}},
+		{"chunked-over-limit", func(w http.ResponseWriter) {
+			fmt.Fprint(w, `{"data":{}}`)
+			w.(http.Flusher).Flush() // No declared length: enforce the read boundary.
+			fmt.Fprint(w, strings.Repeat(" ", int(maxResponseBodyBytes)))
+		}},
+	} {
+		t.Run(test.name, func(t *testing.T) {
+			for name, read := range backupAgentReads() {
+				for _, postCommand := range []bool{false, true} {
+					t.Run(fmt.Sprintf("%s/post=%t", name, postCommand), func(t *testing.T) {
+						var configs, commands atomic.Int32
+						var healthy atomic.Bool
+						server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+							if strings.HasSuffix(r.URL.Path, "/config") {
+								if configs.Add(1) == 1 && postCommand || healthy.Load() {
+									fmt.Fprint(w, `{"data":{}}`)
+								} else {
+									test.write(w)
+								}
+								return
+							}
+							commands.Add(1)
+							backupAgentPayload(w, r)
+						}))
+						defer server.Close()
+						c := backupTestClient(t, server.URL)
+						if err := read(context.Background(), c, 105); GuestAgentDeferredReason(err) != "lock-unverified" {
+							t.Errorf("incomplete config was accepted: %v", err)
+						}
+						wantCommands := int32(0)
+						if postCommand {
+							wantCommands = 1
+						}
+						if got := commands.Load(); got != wantCommands {
+							t.Errorf("guest commands = %d, want %d", got, wantCommands)
+						}
+						// This config failure says nothing about the completed guest
+						// command. A later ordinary read must recheck the lock and resume.
+						healthy.Store(true)
+						if err := read(context.Background(), c, 105); err != nil {
+							t.Fatalf("complete unlocked config did not resume: %v", err)
+						}
+						if got := commands.Load(); got != wantCommands+1 {
+							t.Errorf("resumed guest commands = %d, want %d", got, wantCommands+1)
+						}
+					})
+				}
+			}
+		})
+	}
+}
+
+func TestGuestAgentConfigPrefixDoesNotAdmitCommand(t *testing.T) {
+	var commands atomic.Int32
+	server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		if strings.HasSuffix(r.URL.Path, "/config") {
+			fmt.Fprint(w, `{"data":{}}`)
+			w.(http.Flusher).Flush()
+			<-r.Context().Done() // The body never completes, despite a valid prefix.
+			return
+		}
+		commands.Add(1)
+		backupAgentPayload(w, r)
+	}))
+	defer server.Close()
+	ctx, cancel := context.WithTimeout(context.Background(), 500*time.Millisecond)
+	defer cancel()
+	if _, err := backupTestClient(t, server.URL).GetVMAgentInfo(ctx, "node", 105); GuestAgentDeferredReason(err) != "lock-unverified" {
+		t.Errorf("stalled config admitted a command: %v", err)
+	}
+	if got := commands.Load(); got != 0 {
+		t.Errorf("commands before config completion = %d, want zero", got)
+	}
+}
+
+func TestGuestAgentAdmissionIncludesPostConfigBody(t *testing.T) {
+	var configs, commands atomic.Int32
+	entered := make(chan struct{})
+	server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		if strings.HasSuffix(r.URL.Path, "/config") {
+			n := configs.Add(1)
+			fmt.Fprint(w, `{"data":{}}`)
+			if n == 2 {
+				w.(http.Flusher).Flush()
+				close(entered)
+				<-r.Context().Done()
+			}
+			return
+		}
+		commands.Add(1)
+		backupAgentPayload(w, r)
+	}))
+	defer server.Close()
+	c1, c2 := backupTestClient(t, server.URL), backupTestClient(t, server.URL)
+	ctx, cancel := context.WithTimeout(context.Background(), 500*time.Millisecond)
+	defer cancel()
+	done := make(chan error, 1)
+	go func() {
+		value, err := c1.GetVMFSInfo(ctx, "node", 105)
+		if value != nil {
+			t.Error("filesystem payload escaped the incomplete post-command lock check")
+		}
+		done <- err
+	}()
+	select {
+	case <-entered:
+	case err := <-done:
+		t.Fatalf("post-command config check never started: %v", err)
+	}
+	if _, err := c2.GetVMNetworkInterfaces(context.Background(), "node", 105); GuestAgentDeferredReason(err) != "agent-busy" {
+		t.Errorf("second client bypassed incomplete config admission: %v", err)
+	}
+	if err := <-done; GuestAgentDeferredReason(err) != "lock-unverified" {
+		t.Errorf("post-command config prefix accepted: %v", err)
+	}
+	if got := commands.Load(); got != 1 {
+		t.Errorf("commands during incomplete post-check = %d, want one", got)
+	}
+	if _, err := c2.GetVMFSInfo(context.Background(), "node", 105); err != nil {
+		t.Fatalf("admission not released after failed lock check: %v", err)
+	}
+}
+
+func TestVMConfigCompleteWhitespaceAtResponseLimit(t *testing.T) {
+	const config = `{"data":{"name":"fixture","agent":1}}`
+	server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		fmt.Fprint(w, config)
+		w.(http.Flusher).Flush()
+		fmt.Fprint(w, strings.Repeat(" ", int(maxResponseBodyBytes)-len(config)))
+	}))
+	defer server.Close()
+	got, err := backupTestClient(t, server.URL).GetVMConfig(context.Background(), "node", 105)
+	if err != nil || got["name"] != "fixture" || got["agent"] != float64(1) {
+		t.Fatalf("complete bounded config rejected or changed: %#v %v", got, err)
+	}
+}

pkg/proxmox/io_counters_test.go

@@ -1,7 +1,13 @@
 package proxmox
 
 import (
+	"context"
 	"encoding/json"
+	"fmt"
+	"net/http"
+	"net/http/httptest"
+	"strings"
+	"sync/atomic"
 	"testing"
 	"time"
 )
@@ -138,3 +144,35 @@ func TestBackupLocksRetainCumulativeCounterPresence(t *testing.T) {
 		})
 	}
 }
+
+func TestUnverifiedVMConfigPreservesStatusCounterObservations(t *testing.T) {
+	var commands atomic.Int32
+	server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		switch {
+		case strings.HasSuffix(r.URL.Path, "/config"):
+			w.Header().Set("Content-Length", "1000")
+			fmt.Fprint(w, `{"data":{}}`)
+		case strings.HasSuffix(r.URL.Path, "/status/current"):
+			fmt.Fprint(w, `{"data":{"status":"running","cpu":0.25,"diskread":0,"diskwrite":null,"netin":42}}`)
+		default:
+			commands.Add(1)
+			backupAgentPayload(w, r)
+		}
+	}))
+	defer server.Close()
+	c := backupTestClient(t, server.URL)
+	if _, err := c.GetVMFSInfo(context.Background(), "node", 105); GuestAgentDeferredReason(err) != "lock-unverified" {
+		t.Errorf("incomplete config did not defer the guest command: %v", err)
+	}
+	status, err := c.GetVMStatus(context.Background(), "node", 105)
+	if err != nil {
+		t.Fatal(err)
+	}
+	p := status.IOCounters.Effective()
+	if status.CPU != 0.25 || status.NetIn != 42 || status.DiskRead != 0 || status.ObservedAt.IsZero() || !p.DiskRead || !p.NetworkIn || p.DiskWrite || p.NetworkOut {
+		t.Fatalf("unverified guest lock suppressed/changed current PVE counters: %+v", status)
+	}
+	if got := commands.Load(); got != 0 {
+		t.Errorf("guest commands with incomplete config = %d, want zero", got)
+	}
+}