Backup evidence accessibility
++ Synthetic guest observations, no backup commands or collector reads. +
+ ++ {actions()} +
+From 244d322dcec6ff9c567ae7d0caa3c0571bd05dc0 Mon Sep 17 00:00:00 2001 From: "pulse-triage[bot]" <249995291+pulse-triage[bot]@users.noreply.github.com> Date: Mon, 5 Oct 2026 09:20:32 +0100 Subject: [PATCH 01/12] Coalesce guest identity writes and retain incomplete tenant shutdown Admit detached metadata in order, merge identity fields without restoring stale operator edits, and persist complete snapshots with one owned writer. Seal late writes and return shutdown failures before organisation data or offboarding is removed. Preserve existing drain budgets, file format, tenant boundaries and atomic persistence. Add controlled burst, ordering, write-error, blocked-close and real deletion controls. Contract-Neutral: Behavioural repair of existing JSON persistence and organisation deletion; no route, field, permission, identity format, schema or provider authority change. Shutdown failure uses the existing HTTP error envelope. Internal ownership methods and explanatory contract text change, not the public contract shape. Change-source: pulse-maintainer --- .../v6/internal/subsystems/agent-lifecycle.md | 23 +- .../v6/internal/subsystems/api-contracts.md | 14 + .../v6/internal/subsystems/monitoring.md | 51 ++- internal/api/org_handlers.go | 6 +- internal/api/org_shutdown_safety_test.go | 82 ++++ internal/config/guest_metadata.go | 244 +++++++++--- internal/config/guest_metadata_writer_test.go | 361 ++++++++++++++++++ internal/config/guest_writer_control_test.go | 100 +++++ .../guest_metadata_shutdown_test.go | 125 ++++++ internal/monitoring/monitor.go | 14 +- internal/monitoring/monitor_backups.go | 29 +- internal/monitoring/monitor_metadata_test.go | 21 +- internal/monitoring/multi_tenant_monitor.go | 113 +++--- 13 files changed, 1022 insertions(+), 161 deletions(-) create mode 100644 internal/api/org_shutdown_safety_test.go create mode 100644 internal/config/guest_metadata_writer_test.go create mode 100644 internal/config/guest_writer_control_test.go create mode 100644 internal/monitoring/guest_metadata_shutdown_test.go diff --git a/docs/release-control/v6/internal/subsystems/agent-lifecycle.md b/docs/release-control/v6/internal/subsystems/agent-lifecycle.md index 469e7d2737..6743cf3006 100644 --- a/docs/release-control/v6/internal/subsystems/agent-lifecycle.md +++ b/docs/release-control/v6/internal/subsystems/agent-lifecycle.md @@ -7520,13 +7520,24 @@ coarse activity class (`list`, `export`, `verify`, `summary`); the handler runs unchanged whether or not recording succeeds, and a router without persistence serves the request rather than failing it (`TestWithAuditReadActivity_NilPersistenceIsSafe`). -### Monitor shutdown drains queued guest metadata writes +### Monitor shutdown seals guest metadata admission and retains incomplete teardown + +`Monitor.Stop` seals guest identity/metadata mutations and waits under its +unchanged two-second metadata budget. One store-owned snapshot writer coalesces +changes admitted during filesystem I/O; an older queued identity cannot overwrite +an operator edit or reverse a newer accepted identity. `Close` reports both an +incomplete drain and a final persistence error, while the public void Stop logs +failure rather than claiming a successful stop. + +Tenant removal must first observe its monitoring loop exit and then a successful +metadata close. Failure retains the monitor, cancellation/done handles and +lifecycle guard: lazy recreation cannot start a second writer in the same +directory. `BeginTenantDeletion` returns that failure to the organisation +handler, which keeps persistence and offboarding callbacks untouched. Only +successful shutdown allows directory deletion. Other tenants keep their existing +independent runtime. Store drain does not establish independent guest thaw, +workload liveness or containment in installed software. -`Monitor.Stop` waits for in-flight `GuestMetadataStore` writes before closing -the metrics store, so a tenant monitor that has been stopped is guaranteed not -to write into its data directory afterwards. Tenant offboarding and any caller -that removes a tenant directory can rely on `Stop` having quiesced disk writes, -rather than racing a detached goroutine. ### Per-tenant resource stores are released on offboarding and shutdown `ResourceHandlers.getStore` opens a SQLite handle per org and caches it for the diff --git a/docs/release-control/v6/internal/subsystems/api-contracts.md b/docs/release-control/v6/internal/subsystems/api-contracts.md index 4cc867961f..90f36f2f4b 100644 --- a/docs/release-control/v6/internal/subsystems/api-contracts.md +++ b/docs/release-control/v6/internal/subsystems/api-contracts.md @@ -20,6 +20,20 @@ ## Purpose +### Organisation deletion retains data after incomplete monitoring shutdown + +Authenticated organisation-owner deletion waits for tenant-loop exit and sealed +guest metadata persistence before removing the directory or calling offboarding. +An incomplete drain or final write failure returns HTTP 503 with the existing +error envelope and `tenant_shutdown_incomplete`; the message says the data was +retained and exposes no filesystem path or underlying error. The runtime owner +and recreation guard stay retained. Default-organisation immutability, session +and membership checks, successful HTTP 204 and other tenants' access are +unchanged. `TestOrgDeleteRetainsDataWhenGuestMetadataPersistenceFails` exercises +a real confined rename failure through the ordinary handler and metadata store; +existing successful deletion and authorization controls remain required. + + ### Optional guest-memory observation reaches existing frontend consumers Frontend Memory and ResourceMetric model the existing optional server-owned diff --git a/docs/release-control/v6/internal/subsystems/monitoring.md b/docs/release-control/v6/internal/subsystems/monitoring.md index 48bbc86ac5..40cf4ace67 100644 --- a/docs/release-control/v6/internal/subsystems/monitoring.md +++ b/docs/release-control/v6/internal/subsystems/monitoring.md @@ -4581,27 +4581,42 @@ host dataset evidence is copied onto the matching provider-owned ZFS pool; provider health, scan, device, and error fields remain authoritative. When the provider cannot return pool detail, monitoring may synthesize only a minimal `UNKNOWN` pool so valid dataset evidence is still inspectable. -### Guest metadata writes are owned by the store and drained on shutdown +### Guest identity admission coalesces owned JSON snapshots -`persistGuestIdentity` no longer detaches its own goroutine per changed guest. -It calls `GuestMetadataStore.SetAsync`, which tracks the write on a WaitGroup -so `GuestMetadataStore.WaitForPendingWrites` can drain it. `Monitor.Stop` drains -before closing the metrics store, under a bounded timeout matching -`tenantMonitorShutdownTimeout` so a wedged store cannot hold up tenant teardown. +`persistGuestIdentity` calls `GuestMetadataStore.RememberIdentity`, which merges +only the last-known name/type under the store lock, preserves operator URLs, +descriptions, tags and notes, and never downgrades OCI. Admission is visible +before return. `SetAsync` deep-copies its full replacement at admission as well; +caller mutation and goroutine scheduling cannot reverse accepted update order. -Untracked writes were observable, not theoretical: a queued write could land -after the monitor stopped and after a tenant directory was being removed, -leaving a stray `guest_metadata.json.tmp` from the interrupted atomic write. -That is what made `TestHostedTenantAgentInstallTokenCannotReportToOtherTenant` -fail its `t.TempDir` cleanup with "directory not empty". -`TestGuestMetadataStore_WaitForPendingWritesDrainsQueuedWrites` and -`TestGuestMetadataStore_DataDirIsRemovableAfterDrain` pin the drain and fail if -`SetAsync` stops tracking its goroutine. +One lazy store-owned writer snapshots the complete metadata map. Filesystem I/O +does not hold the memory lock, so changes admitted during that I/O are coalesced +into the next complete snapshot. A separate write mutex serializes it with +synchronous Set/Delete/ReplaceAll/UpdateAll and legacy migration; an older +snapshot cannot resurrect a deleted or migrated entry or overwrite an operator +edit. The existing tenant-local JSON shape, private file permissions and atomic +temporary-file/rename path stay unchanged. A failed write is reported and not +spin-retried; a later ordinary identity poll can retry the unpersisted state. -Known and deliberately unchanged: each changed guest still triggers a full-file -save, so one poll cycle over N changed guests performs N marshals and N atomic -writes that serialize on the store mutex. Coalescing them is a behavioural -change beyond the shutdown defect. +`WaitForPendingWrites` reports quiescence, not durable success. `Close` seals +mutation admission and observes the background writer and any admitted +synchronous write within the unchanged caller budget. Timeout or final +persistence failure remains an error. Monitor Stop logs that failure instead of +claiming a clean stop; tenant removal retains the runtime owner and lifecycle +guard, and organisation deletion refuses to remove its directory. An unfinished +monitoring loop likewise prevents deletion. Shutdown does not cancel an admitted +filesystem write or silently discard metadata to obtain a timely result. + +`internal/config/guest_writer_control_test.go` exercises blocked snapshots at +25/100 updates and caller-copy ownership. `guest_metadata_writer_test.go` covers +ordered replacement, synchronous mutations, field-scoped identity, failure and +explicit recovery, bounded close with blocked synchronous/asynchronous I/O, +closed admission, real filesystem permissions and tenant separation. Existing +ten-second drain and cleanup controls remain unchanged. Monitoring shutdown +controls retain blocked writer/loop ownership; the real organisation-delete +handler returns its safe failure before offboarding or directory removal. +These are source controls, not an attribution of an earlier drain timeout, +native CPU/RSS relief or all-path guest-agent recovery. ### Monitoring projects canonical resource policy into alert evaluation diff --git a/internal/api/org_handlers.go b/internal/api/org_handlers.go index 0704db88ef..fbae571366 100644 --- a/internal/api/org_handlers.go +++ b/internal/api/org_handlers.go @@ -387,7 +387,11 @@ func (h *OrgHandlers) HandleDeleteOrg(w http.ResponseWriter, r *http.Request) { } if h.mtMonitor != nil { - h.mtMonitor.BeginTenantDeletion(orgID) + if err := h.mtMonitor.BeginTenantDeletion(orgID); err != nil { + log.Error().Err(err).Str("org_id", orgID).Msg("Organization shutdown incomplete; retaining data") + writeErrorResponse(w, http.StatusServiceUnavailable, "tenant_shutdown_incomplete", "Organization monitoring has not shut down cleanly; its data has been retained", nil) + return + } defer h.mtMonitor.FinishTenantDeletion(orgID) } diff --git a/internal/api/org_shutdown_safety_test.go b/internal/api/org_shutdown_safety_test.go new file mode 100644 index 0000000000..90ee7f0855 --- /dev/null +++ b/internal/api/org_shutdown_safety_test.go @@ -0,0 +1,82 @@ +package api + +import ( + "context" + "net/http" + "net/http/httptest" + "os" + "path/filepath" + "strings" + "testing" + "time" + + "github.com/rcourtman/pulse-go-rewrite/internal/config" + "github.com/rcourtman/pulse-go-rewrite/internal/models" + "github.com/rcourtman/pulse-go-rewrite/internal/monitoring" +) + +func TestOrgDeleteRetainsDataWhenGuestMetadataPersistenceFails(t *testing.T) { + t.Setenv("PULSE_DEV", "true") + previous := IsMultiTenantEnabled() + SetMultiTenantEnabled(true) + t.Cleanup(func() { SetMultiTenantEnabled(previous) }) + root := t.TempDir() + persistence := config.NewMultiTenantPersistence(root) + mtm := monitoring.NewMultiTenantMonitor(&config.Config{DataPath: root}, persistence, nil) + t.Cleanup(mtm.Stop) + h := NewOrgHandlers(persistence, mtm) + // Exercise the same session-owner setup as ordinary organisation creation. + create := withUser(httptest.NewRequest(http.MethodPost, "/api/orgs", strings.NewReader(`{"id":"retained","displayName":"Retained"}`)), "alice") + rec := httptest.NewRecorder() + h.HandleCreateOrg(rec, create) + if rec.Code != http.StatusCreated { + t.Fatalf("create: %d %s", rec.Code, rec.Body.String()) + } + org, err := persistence.LoadOrganization("retained") + if err != nil || org == nil { + t.Fatal(err) + } + monitor, err := mtm.GetMonitor("retained") + if err != nil { + t.Fatal(err) + } + orgPath := filepath.Join(root, "orgs", "retained") + // A real rename refusal, confined to this disposable test directory. + if err := os.Mkdir(filepath.Join(orgPath, "guest_metadata.json"), 0o700); err != nil { + t.Fatal(err) + } + if err := monitor.GuestMetadataStore().RememberIdentity("pve:node:100", "unsaved", "qemu"); err != nil { + t.Fatal(err) + } + if !monitor.GuestMetadataStore().WaitForPendingWrites(time.Second) { + t.Fatal("failed attempt did not quiesce") + } + called := false + h.SetOnDelete(func(context.Context, string) error { called = true; return nil }) + req := withUser(httptest.NewRequest(http.MethodDelete, "/api/orgs/retained", nil), "alice") + req.SetPathValue("id", "retained") + rec = httptest.NewRecorder() + h.HandleDeleteOrg(rec, req) + if rec.Code != http.StatusServiceUnavailable || !strings.Contains(rec.Body.String(), "tenant_shutdown_incomplete") { + t.Fatalf("unsafe deletion result: %d %s", rec.Code, rec.Body.String()) + } + if called { + t.Fatal("offboarding callback ran after incomplete shutdown") + } + if _, err := os.Stat(filepath.Join(orgPath, "org.json")); err != nil { + t.Fatalf("organisation data removed: %v", err) + } + if got, ok := mtm.PeekMonitor("retained"); !ok || got != monitor { + t.Fatal("failed writer owner lost") + } + if _, err := mtm.GetMonitor("retained"); err == nil { + t.Fatal("new monitor can write into retained shutdown path") + } + // Another tenant remains usable: the failed writer is not a global hold. + if err := persistence.SaveOrganization(&models.Organization{ID: "other"}); err != nil { + t.Fatal(err) + } + if _, err := persistence.GetPersistence("other"); err != nil { + t.Fatal(err) + } +} diff --git a/internal/config/guest_metadata.go b/internal/config/guest_metadata.go index 41e3bbae42..d68e281060 100644 --- a/internal/config/guest_metadata.go +++ b/internal/config/guest_metadata.go @@ -2,11 +2,13 @@ package config import ( "encoding/json" + "errors" "fmt" "os" "path/filepath" "strings" "sync" + "sync/atomic" "time" "github.com/rs/zerolog/log" @@ -31,49 +33,176 @@ type GuestMetadataStore struct { dataPath string fs FileSystem - // inflight tracks background Set calls started by SetAsync so shutdown can - // wait for them. Callers used to spawn their own detached goroutines, which - // meant a write could land after the monitor stopped and after the data - // directory was being torn down, leaving a stray guest_metadata.json.tmp. - inflight sync.WaitGroup + // writeMu serializes disk snapshots with synchronous mutations. Background + // I/O never holds mu, so a burst can accumulate in one follow-up snapshot. + // Always acquire writeMu before mu when both locks are needed. + writeMu sync.Mutex + revision uint64 + attemptedRevision uint64 + lastWriteErr error + writerDone chan struct{} + closed atomic.Bool + closeOnce sync.Once + closeDone chan struct{} + closeErr error } -// SetAsync persists metadata without blocking the caller, while keeping the -// write owned by the store so WaitForPendingWrites can drain it on shutdown. -func (s *GuestMetadataStore) SetAsync(guestID string, meta *GuestMetadata) { +var ErrGuestMetadataStoreClosed = errors.New("guest metadata store is closed") + +// SetAsync admits a detached replacement before returning. One store-owned +// writer persists the newest complete snapshot, coalescing changes that arrive +// during I/O. It does not start a goroutine or rewrite the file per guest. +func (s *GuestMetadataStore) SetAsync(guestID string, meta *GuestMetadata) error { if s == nil { + return nil + } + if meta == nil { + return fmt.Errorf("metadata cannot be nil") + } + clone := cloneGuestMetadata(meta) + clone.ID = guestID + s.mu.Lock() + defer s.mu.Unlock() + if s.closed.Load() { + return ErrGuestMetadataStoreClosed + } + s.metadata[guestID] = clone + s.queueSaveLocked() + return nil +} + +// RememberIdentity merges only monitor-owned identity fields at admission. +// A background name/type update must never restore an old URL, tag or note +// over a concurrent operator edit. OCI classification cannot be downgraded by +// a later poll lacking the classification evidence. +func (s *GuestMetadataStore) RememberIdentity(guestID, name, guestType string) error { + if s == nil { + return nil + } + guestType = strings.TrimSpace(guestType) + if guestType == "" { + return nil + } + s.mu.Lock() + defer s.mu.Unlock() + if s.closed.Load() { + return ErrGuestMetadataStoreClosed + } + meta := cloneGuestMetadata(s.metadata[guestID]) + if meta == nil { + meta = &GuestMetadata{ID: guestID, Tags: []string{}} + } + if meta.LastKnownType == "oci" { + guestType = "oci" + } + if meta.LastKnownName == name && meta.LastKnownType == guestType && s.lastWriteErr == nil { + return nil + } + meta.LastKnownName = name + meta.LastKnownType = guestType + s.metadata[guestID] = meta + s.queueSaveLocked() + return nil +} + +func (s *GuestMetadataStore) queueSaveLocked() { + s.revision++ + if s.writerDone != nil { return } - s.inflight.Add(1) - go func() { - defer s.inflight.Done() - if err := s.Set(guestID, meta); err != nil { - log.Error().Err(err).Str("guestID", guestID).Msg("failed to persist guest metadata") - } - }() + s.writerDone = make(chan struct{}) + go s.writePendingSnapshots() } -// WaitForPendingWrites blocks until background writes finish or the timeout -// elapses. It reports whether the store drained; a false result means a write -// may still be in flight and the data directory is not safe to remove. +func (s *GuestMetadataStore) writePendingSnapshots() { + for { + s.writeMu.Lock() + s.mu.Lock() + if s.attemptedRevision == s.revision { + close(s.writerDone) + s.writerDone = nil + s.mu.Unlock() + s.writeMu.Unlock() + return + } + revision := s.revision + data, err := json.Marshal(s.metadata) + s.mu.Unlock() + if err == nil { + err = persistMetadata(s.fs, s.dataPath, "guest_metadata.json", data) + } + s.mu.Lock() + s.attemptedRevision = revision + s.lastWriteErr = err + s.mu.Unlock() + s.writeMu.Unlock() + if err != nil { + // No unbounded automatic retry on a failing filesystem. A later + // admitted mutation (or ordinary identity poll) can try again. + log.Error().Err(err).Msg("failed to persist guest metadata snapshot") + } + } +} + +// WaitForPendingWrites reports quiescence, not persistence success. Producers +// must already be stopped if the caller plans to remove the data directory; +// Close seals admission as well. A timeout never makes teardown safe. func (s *GuestMetadataStore) WaitForPendingWrites(timeout time.Duration) bool { if s == nil { return true } - done := make(chan struct{}) - go func() { - s.inflight.Wait() - close(done) - }() + s.mu.RLock() + done := s.writerDone + s.mu.RUnlock() + if done == nil { + return true + } + timer := time.NewTimer(timeout) + defer timer.Stop() select { case <-done: return true - case <-time.After(timeout): + case <-timer.C: log.Warn().Dur("timeout", timeout).Msg("timed out waiting for guest metadata writes to drain") return false } } +// Close rejects further mutations, drains the owned writer within the supplied +// budget, and reports any final persistence error. Failure keeps the writer +// owned and must stop tenant-directory removal; it is not a successful stop. +func (s *GuestMetadataStore) Close(timeout time.Duration) error { + if s == nil { + return nil + } + // Seal before waiting for a synchronous mutation's I/O-held memory lock. + // One owned close observer, not one goroutine per waiter, covers both that + // mutation and the background writer without extending the caller's budget. + s.closeOnce.Do(func() { + s.closed.Store(true) + go func() { + s.mu.RLock() + done := s.writerDone + s.mu.RUnlock() + if done != nil { + <-done + } + s.mu.RLock() + s.closeErr = s.lastWriteErr + s.mu.RUnlock() + close(s.closeDone) + }() + }) + timer := time.NewTimer(timeout) + defer timer.Stop() + select { + case <-s.closeDone: + return s.closeErr + case <-timer.C: + return fmt.Errorf("guest metadata writes did not drain within %s", timeout) + } +} + func cloneGuestMetadata(meta *GuestMetadata) *GuestMetadata { if meta == nil { return nil @@ -94,9 +223,10 @@ func cloneGuestMetadata(meta *GuestMetadata) *GuestMetadata { // NewGuestMetadataStore creates a new metadata store func NewGuestMetadataStore(dataPath string, fs FileSystem) *GuestMetadataStore { store := &GuestMetadataStore{ - metadata: make(map[string]*GuestMetadata), - dataPath: dataPath, - fs: fs, + metadata: make(map[string]*GuestMetadata), + dataPath: dataPath, + fs: fs, + closeDone: make(chan struct{}), } if store.fs == nil { @@ -113,6 +243,16 @@ func NewGuestMetadataStore(dataPath string, fs FileSystem) *GuestMetadataStore { // Load reads metadata from disk func (s *GuestMetadataStore) Load() error { + s.writeMu.Lock() + defer s.writeMu.Unlock() + s.mu.Lock() + defer s.mu.Unlock() + if s.closed.Load() { + return ErrGuestMetadataStoreClosed + } + if s.revision != s.attemptedRevision || s.lastWriteErr != nil { + return fmt.Errorf("cannot reload guest metadata with unpersisted changes") + } filePath := filepath.Join(s.dataPath, "guest_metadata.json") log.Debug().Str("path", filePath).Msg("Loading guest metadata from disk") @@ -127,9 +267,6 @@ func (s *GuestMetadataStore) Load() error { return fmt.Errorf("failed to read metadata file: %w", err) } - s.mu.Lock() - defer s.mu.Unlock() - if err := json.Unmarshal(data, &s.metadata); err != nil { return fmt.Errorf("failed to unmarshal metadata: %w", err) } @@ -138,24 +275,16 @@ func (s *GuestMetadataStore) Load() error { return nil } -// save writes metadata to disk (must be called with lock held) +// save writes a synchronous mutation (writeMu and mu must both be held). func (s *GuestMetadataStore) save() error { - filePath := filepath.Join(s.dataPath, "guest_metadata.json") - - log.Debug().Str("path", filePath).Msg("Saving guest metadata to disk") - + s.revision++ data, err := json.Marshal(s.metadata) - if err != nil { - return fmt.Errorf("failed to marshal metadata: %w", err) + if err == nil { + err = persistMetadata(s.fs, s.dataPath, "guest_metadata.json", data) } - - if err := persistMetadata(s.fs, s.dataPath, "guest_metadata.json", data); err != nil { - return err - } - - log.Debug().Str("path", filePath).Int("entries", len(s.metadata)).Msg("Guest metadata saved successfully") - - return nil + s.attemptedRevision = s.revision + s.lastWriteErr = err + return err } // Get retrieves metadata for a guest @@ -188,8 +317,13 @@ func (s *GuestMetadataStore) GetWithLegacyMigration(guestID, instance, node stri // Helper to migrate a legacy ID to the new format migrate := func(legacyID string) *GuestMetadata { + s.writeMu.Lock() + defer s.writeMu.Unlock() s.mu.Lock() defer s.mu.Unlock() + if s.closed.Load() { + return nil + } if current := s.metadata[guestID]; current != nil { return cloneGuestMetadata(current) @@ -293,8 +427,13 @@ func (s *GuestMetadataStore) GetAll() map[string]*GuestMetadata { // Set updates or creates metadata for a guest func (s *GuestMetadataStore) Set(guestID string, meta *GuestMetadata) error { + s.writeMu.Lock() + defer s.writeMu.Unlock() s.mu.Lock() defer s.mu.Unlock() + if s.closed.Load() { + return ErrGuestMetadataStoreClosed + } if meta == nil { return fmt.Errorf("metadata cannot be nil") @@ -310,8 +449,13 @@ func (s *GuestMetadataStore) Set(guestID string, meta *GuestMetadata) error { // Delete removes metadata for a guest func (s *GuestMetadataStore) Delete(guestID string) error { + s.writeMu.Lock() + defer s.writeMu.Unlock() s.mu.Lock() defer s.mu.Unlock() + if s.closed.Load() { + return ErrGuestMetadataStoreClosed + } delete(s.metadata, guestID) @@ -321,8 +465,13 @@ func (s *GuestMetadataStore) Delete(guestID string) error { // ReplaceAll replaces all metadata entries and persists them to disk. func (s *GuestMetadataStore) ReplaceAll(metadata map[string]*GuestMetadata) error { + s.writeMu.Lock() + defer s.writeMu.Unlock() s.mu.Lock() defer s.mu.Unlock() + if s.closed.Load() { + return ErrGuestMetadataStoreClosed + } s.metadata = make(map[string]*GuestMetadata) @@ -354,8 +503,13 @@ func (s *GuestMetadataStore) UpdateAll( return nil } + s.writeMu.Lock() + defer s.writeMu.Unlock() s.mu.Lock() defer s.mu.Unlock() + if s.closed.Load() { + return ErrGuestMetadataStoreClosed + } working := make(map[string]*GuestMetadata, len(s.metadata)) for id, meta := range s.metadata { diff --git a/internal/config/guest_metadata_writer_test.go b/internal/config/guest_metadata_writer_test.go new file mode 100644 index 0000000000..d71b685f2d --- /dev/null +++ b/internal/config/guest_metadata_writer_test.go @@ -0,0 +1,361 @@ +package config + +import ( + "encoding/json" + "errors" + "fmt" + "os" + "path/filepath" + "reflect" + "sync" + "sync/atomic" + "testing" + "time" +) + +func guestWriterTestStore(t *testing.T) (*GuestMetadataStore, *controlledGuestWriterFS) { + t.Helper() + fs := &controlledGuestWriterFS{FileSystem: defaultFileSystem{}, entered: make(chan struct{}), release: make(chan struct{})} + store := NewGuestMetadataStore(t.TempDir(), fs) + var releaseOnce sync.Once + t.Cleanup(func() { + releaseOnce.Do(func() { + select { + case <-fs.release: + default: + close(fs.release) + } + }) + if !store.WaitForPendingWrites(time.Second) { + t.Error("writer did not exit during cleanup") + } + }) + if err := store.SetAsync("first", &GuestMetadata{LastKnownName: "first"}); err != nil { + t.Fatal(err) + } + select { + case <-fs.entered: + case <-time.After(time.Second): + t.Fatal("writer not started") + } + return store, fs +} +func capturedGuestSnapshot(t *testing.T, fs *controlledGuestWriterFS) map[string]*GuestMetadata { + t.Helper() + fs.mu.Lock() + data := append([]byte(nil), fs.data...) + fs.mu.Unlock() + var result map[string]*GuestMetadata + if err := json.Unmarshal(data, &result); err != nil { + t.Fatal(err) + } + return result +} +func TestGuestMetadataWriterOrderedReplacementsAndReadsDuringIO(t *testing.T) { + store, fs := guestWriterTestStore(t) + for i := 0; i < 100; i++ { + if err := store.SetAsync("same", &GuestMetadata{LastKnownName: fmt.Sprint(i)}); err != nil { + t.Fatal(err) + } + } + // This is an admission-state read, not a persistence-success assertion. + read := make(chan *GuestMetadata, 1) + go func() { read <- store.Get("same") }() + select { + case got := <-read: + if got.LastKnownName != "99" { + t.Fatalf("wrong admission order: %+v", got) + } + case <-time.After(time.Second): + t.Fatal("disk I/O blocked metadata reads") + } + close(fs.release) + if err := store.Close(time.Second); err != nil { + t.Fatal(err) + } + if got := capturedGuestSnapshot(t, fs)["same"]; got.LastKnownName != "99" { + t.Fatalf("older update persisted last: %+v", got) + } + fs.mu.Lock() + defer fs.mu.Unlock() + if fs.writes != 2 { + t.Fatalf("100 replacements during one blocked write caused %d writes", fs.writes) + } +} +func TestGuestMetadataWriterSynchronousMutationsCannotBeOverwritten(t *testing.T) { + for _, operation := range []string{"set", "delete", "replace", "update", "migration"} { + t.Run(operation, func(t *testing.T) { + store, fs := guestWriterTestStore(t) + id := "pve:node:100" + if err := store.SetAsync(id, &GuestMetadata{LastKnownName: "queued"}); err != nil { + t.Fatal(err) + } + done := make(chan error, 1) + go func() { + switch operation { + case "set": + done <- store.Set(id, &GuestMetadata{LastKnownName: "operator", Notes: []string{"new"}}) + case "delete": + done <- store.Delete(id) + case "replace": + done <- store.ReplaceAll(map[string]*GuestMetadata{"replacement": {LastKnownName: "replacement"}}) + case "update": + done <- store.UpdateAll(func(m map[string]*GuestMetadata) bool { m[id].Notes = []string{"edited"}; return true }) + case "migration": + if store.GetWithLegacyMigration("pve:other:100", "pve", "other", 100) == nil { + done <- errors.New("migration missing") + } else { + done <- nil + } + } + }() + close(fs.release) + select { + case err := <-done: + if err != nil { + t.Fatal(err) + } + case <-time.After(time.Second): + t.Fatal("synchronous mutation blocked") + } + if err := store.Close(time.Second); err != nil { + t.Fatal(err) + } + saved := capturedGuestSnapshot(t, fs) + if !reflect.DeepEqual(saved, store.GetAll()) { + t.Fatalf("disk snapshot differs after %s: saved=%+v memory=%+v", operation, saved, store.GetAll()) + } + switch operation { + case "set": + if saved[id].LastKnownName != "operator" || saved[id].Notes[0] != "new" { + t.Fatal("operator replacement lost") + } + case "delete": + if saved[id] != nil { + t.Fatal("deleted entry resurrected") + } + case "replace": + if len(saved) != 1 || saved["replacement"] == nil { + t.Fatal("replace resurrected old entries") + } + case "update": + if saved[id].Notes[0] != "edited" { + t.Fatal("update lost") + } + case "migration": + if saved[id] != nil || saved["pve:other:100"] == nil { + t.Fatal("old node resurrected") + } + } + }) + } +} +func TestGuestMetadataWriterIdentityPreservesOperatorFieldsAndOCI(t *testing.T) { + store := NewGuestMetadataStore(t.TempDir(), nil) + original := &GuestMetadata{CustomURL: "https://operator.example", Description: "operator", Tags: []string{"tag"}, Notes: []string{"note"}, LastKnownName: "old", LastKnownType: "oci"} + if err := store.Set("pve:node:100", original); err != nil { + t.Fatal(err) + } + if err := store.RememberIdentity("pve:node:100", "new", " lxc "); err != nil { + t.Fatal(err) + } + got := store.Get("pve:node:100") + if got.LastKnownName != "new" || got.LastKnownType != "oci" { + t.Fatalf("identity/OCI lost: %+v", got) + } + original.ID = "pve:node:100" + original.LastKnownName = "new" + if !reflect.DeepEqual(got, original) { + t.Fatalf("identity changed operator data: %+v", got) + } + if !store.WaitForPendingWrites(time.Second) { + t.Fatal("identity not drained") + } + before := store.revision + if err := store.RememberIdentity("pve:node:100", "new", "oci"); err != nil { + t.Fatal(err) + } + if err := store.RememberIdentity("pve:node:100", "ignored", " "); err != nil { + t.Fatal(err) + } + if store.revision != before { + t.Fatal("unchanged/empty type triggered a write") + } + if err := store.Close(time.Second); err != nil { + t.Fatal(err) + } + reloaded := NewGuestMetadataStore(store.dataPath, nil) + if !reflect.DeepEqual(reloaded.Get("pve:node:100"), got) { + t.Fatal("identity/operator fields not persisted") + } +} +func TestGuestMetadataWriterCloseBoundsBlockedAsyncAndSyncIO(t *testing.T) { + for _, synchronous := range []bool{false, true} { + t.Run(fmt.Sprint(synchronous), func(t *testing.T) { + fs := &controlledGuestWriterFS{FileSystem: defaultFileSystem{}, entered: make(chan struct{}), release: make(chan struct{})} + store := NewGuestMetadataStore(t.TempDir(), fs) + done := make(chan error, 1) + if synchronous { + go func() { done <- store.Set("first", &GuestMetadata{LastKnownName: "first"}) }() + } else { + done <- store.SetAsync("first", &GuestMetadata{LastKnownName: "first"}) + } + select { + case <-fs.entered: + case <-time.After(time.Second): + t.Fatal("write did not start") + } + t.Cleanup(func() { + select { + case <-fs.release: + default: + close(fs.release) + } + _ = store.Close(time.Second) + }) + start := time.Now() + if err := store.Close(10 * time.Millisecond); err == nil { + t.Fatal("blocked writer reported close success") + } + if elapsed := time.Since(start); elapsed > time.Second { + t.Fatalf("close exceeded predeclared bounded budget: %s", elapsed) + } + close(fs.release) + select { + case err := <-done: + if err != nil { + t.Fatal(err) + } + case <-time.After(time.Second): + t.Fatal("write still running") + } + if err := store.Close(time.Second); err != nil { + t.Fatal(err) + } + for _, mutation := range []func() error{ + func() error { return store.SetAsync("later", &GuestMetadata{}) }, + func() error { return store.RememberIdentity("later", "later", "qemu") }, + func() error { return store.Set("later", &GuestMetadata{}) }, + func() error { return store.Delete("first") }, + func() error { return store.ReplaceAll(nil) }, + func() error { return store.UpdateAll(func(map[string]*GuestMetadata) bool { return true }) }, + store.Load, + } { + if err := mutation(); !errors.Is(err, ErrGuestMetadataStoreClosed) { + t.Fatalf("closed admission accepted: %v", err) + } + } + if store.GetWithLegacyMigration("node:100", "pve", "node", 100) != nil { + t.Fatal("closed store migrated metadata") + } + if capturedGuestSnapshot(t, fs)["later"] != nil { + t.Fatal("post-close write") + } + }) + } +} + +type guestWriterFaultFS struct { + FileSystem + failWrite atomic.Bool + failRename atomic.Bool + writes atomic.Int32 +} + +func (f *guestWriterFaultFS) WriteFile(path string, data []byte, perm os.FileMode) error { + f.writes.Add(1) + if f.failWrite.Load() { + return errors.New("controlled write failure") + } + return f.FileSystem.WriteFile(path, data, perm) +} +func (f *guestWriterFaultFS) Rename(a, b string) error { + if f.failRename.Load() { + return errors.New("controlled rename failure") + } + return f.FileSystem.Rename(a, b) +} +func TestGuestMetadataWriterFailuresAreOwnedWithoutSpinRetry(t *testing.T) { + for _, phase := range []string{"write", "rename"} { + t.Run(phase, func(t *testing.T) { + fs := &guestWriterFaultFS{FileSystem: defaultFileSystem{}} + store := NewGuestMetadataStore(t.TempDir(), fs) + if err := store.Set("key", &GuestMetadata{LastKnownName: "original"}); err != nil { + t.Fatal(err) + } + fs.failWrite.Store(phase == "write") + fs.failRename.Store(phase == "rename") + if err := store.RememberIdentity("key", "changed", "qemu"); err != nil { + t.Fatal(err) + } + if !store.WaitForPendingWrites(time.Second) { + t.Fatal("failed attempt not quiescent") + } + if fs.writes.Load() != 2 { + t.Fatalf("failure retried automatically: %d", fs.writes.Load()) + } + if got := NewGuestMetadataStore(store.dataPath, nil).Get("key"); got.LastKnownName != "original" { + t.Fatal("failure replaced good disk snapshot") + } + fs.failWrite.Store(false) + fs.failRename.Store(false) + // A later ordinary poll retries the unpersisted identity, not a tight loop. + if err := store.RememberIdentity("key", "changed", "qemu"); err != nil { + t.Fatal(err) + } + if err := store.Close(time.Second); err != nil { + t.Fatal(err) + } + if fs.writes.Load() != 3 { + t.Fatalf("explicit recovery writes=%d", fs.writes.Load()) + } + if got := NewGuestMetadataStore(store.dataPath, nil).Get("key"); got.LastKnownName != "changed" { + t.Fatal("recovery not persisted") + } + fs.failWrite.Store(true) + failed := NewGuestMetadataStore(t.TempDir(), fs) + if err := failed.SetAsync("lost", &GuestMetadata{}); err != nil { + t.Fatal(err) + } + if err := failed.Close(time.Second); err == nil { + t.Fatal("failed final write reported durable close") + } + }) + } +} +func TestGuestMetadataWriterRealFilesystemTenantIsolation(t *testing.T) { + root := t.TempDir() + a := NewGuestMetadataStore(filepath.Join(root, "orgs", "a"), nil) + b := NewGuestMetadataStore(filepath.Join(root, "orgs", "b"), nil) + for i := 0; i < 100; i++ { + id := fmt.Sprintf("pve:node:%d", i) + if err := a.RememberIdentity(id, "a", "qemu"); err != nil { + t.Fatal(err) + } + if err := b.RememberIdentity(id, "b", "lxc"); err != nil { + t.Fatal(err) + } + } + for _, store := range []*GuestMetadataStore{a, b} { + if err := store.Close(10 * time.Second); err != nil { + t.Fatal(err) + } + reloaded := NewGuestMetadataStore(store.dataPath, nil) + if !reflect.DeepEqual(reloaded.GetAll(), store.GetAll()) { + t.Fatal("real filesystem lost entries") + } + info, err := os.Stat(filepath.Join(store.dataPath, "guest_metadata.json")) + if err != nil { + t.Fatal(err) + } + if info.Mode().Perm() != 0o600 { + t.Fatalf("unsafe file permissions: %v", info.Mode()) + } + if _, err := os.Stat(filepath.Join(store.dataPath, "guest_metadata.json.tmp")); !os.IsNotExist(err) { + t.Fatalf("temporary file survived close: %v", err) + } + } + if a.Get("pve:node:1").LastKnownName == b.Get("pve:node:1").LastKnownName { + t.Fatal("tenants share identity") + } +} diff --git a/internal/config/guest_writer_control_test.go b/internal/config/guest_writer_control_test.go new file mode 100644 index 0000000000..a1c2e9e7ee --- /dev/null +++ b/internal/config/guest_writer_control_test.go @@ -0,0 +1,100 @@ +package config + +import ( + "encoding/json" + "fmt" + "os" + "sync" + "testing" + "time" +) + +type controlledGuestWriterFS struct { + FileSystem + mu sync.Mutex + writes int + data []byte + entered chan struct{} + release chan struct{} +} + +func (f *controlledGuestWriterFS) MkdirAll(string, os.FileMode) error { return nil } +func (f *controlledGuestWriterFS) Rename(string, string) error { return nil } +func (f *controlledGuestWriterFS) WriteFile(_ string, data []byte, _ os.FileMode) error { + f.mu.Lock() + f.writes++ + first := f.writes == 1 + f.mu.Unlock() + if first { + close(f.entered) + <-f.release + } + f.mu.Lock() + f.data = append([]byte(nil), data...) + f.mu.Unlock() + return nil +} +func TestGuestWriterControlledBurst(t *testing.T) { + for _, count := range []int{25, 100} { + t.Run(fmt.Sprint(count), func(t *testing.T) { + fs := &controlledGuestWriterFS{FileSystem: defaultFileSystem{}, entered: make(chan struct{}), release: make(chan struct{})} + store := NewGuestMetadataStore(t.TempDir(), fs) + store.SetAsync("first", &GuestMetadata{LastKnownName: "first"}) + select { + case <-fs.entered: + case <-time.After(time.Second): + t.Fatal("first write did not start") + } + for i := 0; i < count; i++ { + store.SetAsync(fmt.Sprint(i), &GuestMetadata{LastKnownName: fmt.Sprint(i)}) + } + close(fs.release) + if !store.WaitForPendingWrites(10 * time.Second) { + t.Fatal("drain failed") + } + fs.mu.Lock() + writes := fs.writes + data := append([]byte(nil), fs.data...) + fs.mu.Unlock() + t.Logf("admitted=%d complete-file-writes=%d", count+1, writes) + if writes > 2 { + t.Errorf("one blocked snapshot plus one complete follow-up needed, got %d", writes) + } + var saved map[string]*GuestMetadata + if err := json.Unmarshal(data, &saved); err != nil { + t.Fatal(err) + } + if len(saved) != count+1 { + t.Fatalf("lost guests: got %d want %d", len(saved), count+1) + } + for i := 0; i < count; i++ { + if saved[fmt.Sprint(i)].LastKnownName != fmt.Sprint(i) { + t.Errorf("wrong guest %d", i) + } + } + }) + } +} +func TestGuestWriterCopiesAtAdmission(t *testing.T) { + fs := &controlledGuestWriterFS{FileSystem: defaultFileSystem{}, entered: make(chan struct{}), release: make(chan struct{})} + store := NewGuestMetadataStore(t.TempDir(), fs) + store.SetAsync("first", &GuestMetadata{LastKnownName: "first"}) + select { + case <-fs.entered: + case <-time.After(time.Second): + t.Fatal("first write did not start") + } + meta := &GuestMetadata{LastKnownName: "original", Tags: []string{"original"}, Notes: []string{"original"}} + store.SetAsync("second", meta) + meta.LastKnownName = "mutated" + meta.Tags[0] = "mutated" + meta.Notes[0] = "mutated" + close(fs.release) + if !store.WaitForPendingWrites(10 * time.Second) { + t.Fatal("drain failed") + } + got := store.Get("second") + if got.LastKnownName != "original" || got.Tags[0] != "original" || got.Notes[0] != "original" { + t.Fatalf("caller mutation reached stored metadata: %+v", got) + } +} diff --git a/internal/monitoring/guest_metadata_shutdown_test.go b/internal/monitoring/guest_metadata_shutdown_test.go new file mode 100644 index 0000000000..8c9ad302ee --- /dev/null +++ b/internal/monitoring/guest_metadata_shutdown_test.go @@ -0,0 +1,125 @@ +package monitoring + +import ( + "context" + "os" + "path/filepath" + "testing" + "time" + + "github.com/rcourtman/pulse-go-rewrite/internal/config" +) + +type blockedGuestMetadataFS struct { + config.FileSystem + entered chan struct{} + release chan struct{} +} + +func (f *blockedGuestMetadataFS) ReadFile(p string) ([]byte, error) { return os.ReadFile(p) } +func (f *blockedGuestMetadataFS) Stat(p string) (os.FileInfo, error) { return os.Stat(p) } +func (f *blockedGuestMetadataFS) MkdirAll(p string, m os.FileMode) error { return os.MkdirAll(p, m) } +func (f *blockedGuestMetadataFS) WriteFile(p string, b []byte, m os.FileMode) error { + close(f.entered) + <-f.release + return os.WriteFile(p, b, m) +} +func (f *blockedGuestMetadataFS) Rename(a, b string) error { return os.Rename(a, b) } + +func TestTenantDeletionRetainsBlockedGuestWriterOwnership(t *testing.T) { + path := t.TempDir() + fs := &blockedGuestMetadataFS{entered: make(chan struct{}), release: make(chan struct{})} + store := config.NewGuestMetadataStore(path, fs) + monitor := &Monitor{guestMetadataStore: store} + mtm := NewMultiTenantMonitor(&config.Config{DataPath: path}, nil, nil) + mtm.monitors["blocked"] = monitor + t.Cleanup(func() { + select { + case <-fs.release: + default: + close(fs.release) + } + _ = store.Close(time.Second) + mtm.Stop() + }) + persistGuestIdentity(store, "pve:node:100", "retained", "qemu") + select { + case <-fs.entered: + case <-time.After(time.Second): + t.Fatal("write did not start") + } + if err := mtm.BeginTenantDeletion("blocked"); err == nil { + t.Fatal("deletion accepted incomplete drain") + } + if retained, ok := mtm.PeekMonitor("blocked"); !ok || retained != monitor { + t.Fatal("in-flight writer owner was discarded") + } + mtm.FinishTenantDeletion("blocked") + if _, err := mtm.GetMonitor("blocked"); err == nil { + t.Fatal("deferred cleanup released unsafe lifecycle guard") + } + if _, err := os.Stat(path); err != nil { + t.Fatalf("data path not retained: %v", err) + } + close(fs.release) + if err := store.Close(time.Second); err != nil { + t.Fatal(err) + } + if err := mtm.BeginTenantDeletion("blocked"); err != nil { + t.Fatal(err) + } + if _, ok := mtm.PeekMonitor("blocked"); ok { + t.Fatal("completed shutdown not removed") + } + if err := store.RememberIdentity("late", "late", "qemu"); err == nil { + t.Fatal("stopped store accepted late identity") + } + if _, err := os.Stat(filepath.Join(path, "guest_metadata.json")); err != nil { + t.Fatal("accepted metadata not persisted") + } +} +func TestTenantDeletionRetainsUnfinishedMonitorLoop(t *testing.T) { + mtm := NewMultiTenantMonitor(&config.Config{}, nil, nil) + _, cancel := context.WithCancel(context.Background()) + done := make(chan struct{}) + monitor := &Monitor{} + mtm.monitors["loop"] = monitor + mtm.tenantCancel["loop"] = cancel + mtm.tenantDone["loop"] = done + t.Cleanup(func() { + select { + case <-done: + default: + close(done) + } + mtm.Stop() + }) + if err := mtm.BeginTenantDeletion("loop"); err == nil { + t.Fatal("unfinished monitoring loop admitted directory deletion") + } + if retained, ok := mtm.PeekMonitor("loop"); !ok || retained != monitor { + t.Fatal("unfinished monitoring loop lost owner") + } + if _, err := mtm.GetMonitor("loop"); err == nil { + t.Fatal("unfinished monitor replaced") + } + close(done) + if err := mtm.BeginTenantDeletion("loop"); err != nil { + t.Fatal(err) + } +} +func TestPersistGuestIdentityEditsAreImmediateAndFieldScoped(t *testing.T) { + store := config.NewGuestMetadataStore(t.TempDir(), nil) + if err := store.Set("pve:node:100", &config.GuestMetadata{CustomURL: "https://operator.example", Tags: []string{"important"}, Notes: []string{"keep"}}); err != nil { + t.Fatal(err) + } + persistGuestIdentity(store, "pve:node:100", "old", "qemu") + persistGuestIdentity(store, "pve:node:100", "new", "qemu") + got := store.Get("pve:node:100") + if got.LastKnownName != "new" || got.CustomURL != "https://operator.example" || got.Tags[0] != "important" || got.Notes[0] != "keep" { + t.Fatalf("producer corrupted metadata: %+v", got) + } + if err := store.Close(time.Second); err != nil { + t.Fatal(err) + } +} diff --git a/internal/monitoring/monitor.go b/internal/monitoring/monitor.go index 6e9091900e..6bab63084c 100644 --- a/internal/monitoring/monitor.go +++ b/internal/monitoring/monitor.go @@ -7598,6 +7598,12 @@ func monitorLastSeenUnix(value time.Time) int64 { const guestMetadataDrainTimeout = 2 * time.Second func (m *Monitor) Stop() { + if err := m.stop(); err != nil { + log.Error().Err(err).Msg("monitor stopped with incomplete guest metadata persistence; data directory must be retained") + } +} + +func (m *Monitor) stop() error { log.Info().Msg("stopping monitor") // A trailing agent-report refresh must not run against stores closed below. @@ -7620,9 +7626,7 @@ func (m *Monitor) Stop() { // Drain background guest-metadata writes before the data directory can be // torn down. Without this a queued write lands after shutdown and leaves a // stray guest_metadata.json.tmp behind. - if m.guestMetadataStore != nil { - m.guestMetadataStore.WaitForPendingWrites(guestMetadataDrainTimeout) - } + drainErr := m.guestMetadataStore.Close(guestMetadataDrainTimeout) // Close persistent metrics store (flushes buffered data) if m.metricsStore != nil { @@ -7633,7 +7637,11 @@ func (m *Monitor) Stop() { } } + if drainErr != nil { + return drainErr + } log.Info().Msg("monitor stopped") + return nil } // recordAuthFailure records an authentication failure for a node diff --git a/internal/monitoring/monitor_backups.go b/internal/monitoring/monitor_backups.go index 1dc59f1184..d5475857eb 100644 --- a/internal/monitoring/monitor_backups.go +++ b/internal/monitoring/monitor_backups.go @@ -872,33 +872,8 @@ func enrichWithPersistedMetadata(metadataStore *config.GuestMetadataStore, byVMI // persistGuestIdentity updates the metadata store with the last-known name and type for a guest func persistGuestIdentity(metadataStore *config.GuestMetadataStore, guestKey, name, guestType string) { - existing := metadataStore.Get(guestKey) - if existing == nil { - existing = &config.GuestMetadata{ - ID: guestKey, - Tags: []string{}, - } - } - - guestType = strings.TrimSpace(guestType) - if guestType == "" { - return - } - - // Never "downgrade" OCI containers back to LXC. OCI classification can be transiently - // unavailable if Proxmox config reads fail due to permissions or transient API errors. - if existing.LastKnownType == "oci" && guestType != "oci" { - guestType = existing.LastKnownType - } - - // Only update if the name or type has changed - if existing.LastKnownName != name || existing.LastKnownType != guestType { - existing.LastKnownName = name - existing.LastKnownType = guestType - // Save without blocking the monitor. The store owns the goroutine so - // Monitor.Stop can drain it; a detached goroutine here could write - // after shutdown, into a data directory that was already being removed. - metadataStore.SetAsync(guestKey, existing) + if err := metadataStore.RememberIdentity(guestKey, name, guestType); err != nil { + log.Error().Err(err).Str("guestKey", guestKey).Msg("failed to admit guest identity update") } } diff --git a/internal/monitoring/monitor_metadata_test.go b/internal/monitoring/monitor_metadata_test.go index 12f3d3e658..e63bd48fed 100644 --- a/internal/monitoring/monitor_metadata_test.go +++ b/internal/monitoring/monitor_metadata_test.go @@ -20,15 +20,6 @@ func TestPersistGuestIdentity_Concurrent(t *testing.T) { } defer os.RemoveAll(tmpDir) - // metadataFile := filepath.Join(tmpDir, "guest_metadata.json") // Actually NewGuestMetadataStore takes directory not file usually? No, it takes root dir? - // Let's verify standard usage. "store := NewGuestMetadataStore(tmpDir, nil)" - // Implementation: func NewGuestMetadataStore(dataPath string, fs FileSystem) - // Inside it does filepath.Join(dataPath, "guest_metadata.json") ? - // Let's re-read NewGuestMetadataStore in internal/config/guest_metadata.go via grep or similar if needed. - // But based on "config.NewGuestMetadataStore(metadataFile)" from my previous code failing, and grep showing "dataPath", it likely takes a Dir or File path. - // grep output: guestMetadataStore := NewGuestMetadataStore(dataPath, c.fs) - // Most likely directory. - store := config.NewGuestMetadataStore(tmpDir, nil) guestKey := "pve1:node1:100" @@ -36,8 +27,9 @@ func TestPersistGuestIdentity_Concurrent(t *testing.T) { // Test basic persistence persistGuestIdentity(store, guestKey, "VM 100", "qemu") - // Wait a bit since persistGuestIdentity is async - time.Sleep(50 * time.Millisecond) + if !store.WaitForPendingWrites(time.Second) { + t.Fatal("identity writer did not drain") + } meta := store.Get(guestKey) if meta == nil || meta.LastKnownName != "VM 100" || meta.LastKnownType != "qemu" { @@ -55,7 +47,9 @@ func TestPersistGuestIdentity_Concurrent(t *testing.T) { // Try to update to "lxc" persistGuestIdentity(store, guestKey, "VM 100", "lxc") - time.Sleep(50 * time.Millisecond) + if !store.WaitForPendingWrites(time.Second) { + t.Fatal("identity writer did not drain") + } meta = store.Get(guestKey) if meta.LastKnownType != "oci" { @@ -65,6 +59,9 @@ func TestPersistGuestIdentity_Concurrent(t *testing.T) { // Test persistence that shouldn't happen (no change) -> coverage of the if check // Should not trigger Set() persistGuestIdentity(store, guestKey, "VM 100", "oci") + if err := store.Close(time.Second); err != nil { + t.Fatal(err) + } } func TestEnrichWithPersistedMetadata_Detail(t *testing.T) { diff --git a/internal/monitoring/multi_tenant_monitor.go b/internal/monitoring/multi_tenant_monitor.go index 14a76d656b..248b842239 100644 --- a/internal/monitoring/multi_tenant_monitor.go +++ b/internal/monitoring/multi_tenant_monitor.go @@ -20,6 +20,7 @@ type MultiTenantMonitor struct { tenantCancel map[string]context.CancelFunc tenantDone map[string]chan struct{} tenantDeleting map[string]struct{} + tenantStopping map[string]struct{} persistence *config.MultiTenantPersistence baseConfig *config.Config wsHub *websocket.Hub @@ -37,6 +38,7 @@ func NewMultiTenantMonitor(baseCfg *config.Config, persistence *config.MultiTena tenantCancel: make(map[string]context.CancelFunc), tenantDone: make(map[string]chan struct{}), tenantDeleting: make(map[string]struct{}), + tenantStopping: make(map[string]struct{}), persistence: persistence, baseConfig: baseCfg, // Used as a template or for global settings wsHub: wsHub, @@ -294,47 +296,34 @@ func (mtm *MultiTenantMonitor) Stop() { log.Info().Msg("stopping MultiTenantMonitor and all tenant instances") mtm.mu.Lock() - mtm.globalCancel() - - monitors := make([]*Monitor, 0, len(mtm.monitors)) - cancels := make([]context.CancelFunc, 0, len(mtm.tenantCancel)) - doneSignals := make([]chan struct{}, 0, len(mtm.tenantDone)) - for orgID, monitor := range mtm.monitors { - if cancel := mtm.tenantCancel[orgID]; cancel != nil { - cancels = append(cancels, cancel) - } - if done := mtm.tenantDone[orgID]; done != nil { - doneSignals = append(doneSignals, done) - } - monitors = append(monitors, monitor) + if mtm.globalCancel != nil { + mtm.globalCancel() + } + orgIDs := make([]string, 0, len(mtm.monitors)) + for orgID := range mtm.monitors { + orgIDs = append(orgIDs, orgID) } - mtm.tenantCancel = make(map[string]context.CancelFunc) - mtm.tenantDone = make(map[string]chan struct{}) - mtm.monitors = make(map[string]*Monitor) mtm.mu.Unlock() - - for _, cancel := range cancels { - cancel() - } - for _, done := range doneSignals { - waitForTenantMonitorShutdown("", done) - } - for _, monitor := range monitors { - monitor.Stop() + for _, orgID := range orgIDs { + if err := mtm.removeTenant(orgID, false); err != nil { + log.Error().Err(err).Str("org_id", orgID).Msg("tenant shutdown incomplete; retaining runtime ownership and data") + } } } // RemoveTenant stops and removes a specific tenant's monitor. // Useful for offboarding or manual reloading. func (mtm *MultiTenantMonitor) RemoveTenant(orgID string) { - mtm.removeTenant(orgID, false) + if err := mtm.removeTenant(orgID, false); err != nil { + log.Error().Err(err).Str("org_id", orgID).Msg("tenant removal incomplete; retaining runtime ownership and data") + } } // BeginTenantDeletion prevents lazy monitor initialization while a tenant's // persistence is being removed. Call FinishTenantDeletion after the -// persistence operation completes, whether it succeeds or fails. -func (mtm *MultiTenantMonitor) BeginTenantDeletion(orgID string) { - mtm.removeTenant(orgID, true) +// persistence operation completes. An incomplete shutdown retains its guard. +func (mtm *MultiTenantMonitor) BeginTenantDeletion(orgID string) error { + return mtm.removeTenant(orgID, true) } // FinishTenantDeletion releases the temporary lifecycle guard installed by @@ -347,46 +336,70 @@ func (mtm *MultiTenantMonitor) FinishTenantDeletion(orgID string) { } mtm.mu.Lock() - delete(mtm.tenantDeleting, orgID) + // An incomplete removal still owns a runtime or writer. Do not release + // its guard merely because an offboarding caller ran deferred cleanup. + if _, retained := mtm.monitors[orgID]; !retained { + delete(mtm.tenantDeleting, orgID) + } mtm.mu.Unlock() } -func (mtm *MultiTenantMonitor) removeTenant(orgID string, deleting bool) { +func (mtm *MultiTenantMonitor) removeTenant(orgID string, deleting bool) error { orgID = strings.TrimSpace(orgID) if orgID == "" { - return + return nil } mtm.mu.Lock() - if deleting { - if mtm.tenantDeleting == nil { - mtm.tenantDeleting = make(map[string]struct{}) - } - mtm.tenantDeleting[orgID] = struct{}{} + if mtm.tenantDeleting == nil { + mtm.tenantDeleting = make(map[string]struct{}) } - monitor, exists := mtm.monitors[orgID] + if mtm.tenantStopping == nil { + mtm.tenantStopping = make(map[string]struct{}) + } + if _, stopping := mtm.tenantStopping[orgID]; stopping { + mtm.mu.Unlock() + return fmt.Errorf("organization %q shutdown is already in progress", orgID) + } + mtm.tenantDeleting[orgID] = struct{}{} + mtm.tenantStopping[orgID] = struct{}{} + monitor := mtm.monitors[orgID] cancel := mtm.tenantCancel[orgID] done := mtm.tenantDone[orgID] - delete(mtm.monitors, orgID) - delete(mtm.tenantCancel, orgID) - delete(mtm.tenantDone, orgID) mtm.mu.Unlock() - - if !exists { - return - } + defer func() { + mtm.mu.Lock() + delete(mtm.tenantStopping, orgID) + mtm.mu.Unlock() + }() log.Info().Str("org_id", orgID).Msg("stopping and removing tenant monitor") if cancel != nil { cancel() } - waitForTenantMonitorShutdown(orgID, done) - monitor.Stop() + if !waitForTenantMonitorShutdown(orgID, done) { + return fmt.Errorf("organization %q monitoring loop has not stopped", orgID) + } + if monitor != nil { + if err := monitor.stop(); err != nil { + return fmt.Errorf("organization %q guest metadata shutdown incomplete: %w", orgID, err) + } + } + + mtm.mu.Lock() + delete(mtm.monitors, orgID) + delete(mtm.tenantCancel, orgID) + delete(mtm.tenantDone, orgID) + if !deleting { + delete(mtm.tenantDeleting, orgID) + } + mtm.mu.Unlock() + return nil } -func waitForTenantMonitorShutdown(orgID string, done <-chan struct{}) { +func waitForTenantMonitorShutdown(orgID string, done <-chan struct{}) bool { if done == nil { - return + return true } timer := time.NewTimer(tenantMonitorShutdownTimeout) @@ -394,12 +407,14 @@ func waitForTenantMonitorShutdown(orgID string, done <-chan struct{}) { select { case <-done: + return true case <-timer.C: logger := log.Warn().Dur("timeout", tenantMonitorShutdownTimeout) if orgID != "" { logger = logger.Str("org_id", orgID) } logger.Msg("timed out waiting for tenant monitor loop to exit") + return false } } From 5e51069048165b93e3b776fbdbf16a7a0d536320 Mon Sep 17 00:00:00 2001 From: "pulse-triage[bot]" <249995291+pulse-triage[bot]@users.noreply.github.com> Date: Mon, 5 Oct 2026 09:31:01 +0100 Subject: [PATCH 02/12] Allow explicit recovery of retained guest metadata after failed close Keep admission sealed and reuse a still-running close observer. A later explicit shutdown attempt may persist retained state after a completed write failure; successful close remains idempotent. This avoids a permanent offboarding dead end after an ordinary filesystem repair without an automatic spin retry or overlapping writes. Contract-Neutral: Internal JSON writer recovery behaviour only; public routes, fields, permissions, identity format, private file mode and atomic persistence are unchanged. Change-source: pulse-maintainer --- .../v6/internal/subsystems/agent-lifecycle.md | 4 +- .../v6/internal/subsystems/monitoring.md | 5 +- internal/config/guest_metadata.go | 59 +++++++++++++------ internal/config/guest_metadata_writer_test.go | 35 +++++++++++ 4 files changed, 84 insertions(+), 19 deletions(-) diff --git a/docs/release-control/v6/internal/subsystems/agent-lifecycle.md b/docs/release-control/v6/internal/subsystems/agent-lifecycle.md index 6743cf3006..0e9db6c03d 100644 --- a/docs/release-control/v6/internal/subsystems/agent-lifecycle.md +++ b/docs/release-control/v6/internal/subsystems/agent-lifecycle.md @@ -7534,7 +7534,9 @@ metadata close. Failure retains the monitor, cancellation/done handles and lifecycle guard: lazy recreation cannot start a second writer in the same directory. `BeginTenantDeletion` returns that failure to the organisation handler, which keeps persistence and offboarding callbacks untouched. Only -successful shutdown allows directory deletion. Other tenants keep their existing +successful shutdown allows directory deletion. A later explicit shutdown attempt +may persist retained metadata after a completed write failure is repaired; +admission stays sealed and no in-flight attempt is replayed. Other tenants keep their existing independent runtime. Store drain does not establish independent guest thaw, workload liveness or containment in installed software. diff --git a/docs/release-control/v6/internal/subsystems/monitoring.md b/docs/release-control/v6/internal/subsystems/monitoring.md index 40cf4ace67..354ad8d5e2 100644 --- a/docs/release-control/v6/internal/subsystems/monitoring.md +++ b/docs/release-control/v6/internal/subsystems/monitoring.md @@ -4605,7 +4605,10 @@ persistence failure remains an error. Monitor Stop logs that failure instead of claiming a clean stop; tenant removal retains the runtime owner and lifecycle guard, and organisation deletion refuses to remove its directory. An unfinished monitoring loop likewise prevents deletion. Shutdown does not cancel an admitted -filesystem write or silently discard metadata to obtain a timely result. +filesystem write or silently discard metadata to obtain a timely result. A later +explicit close may retry a completed failed snapshot after filesystem repair; +admission stays sealed, successful close is idempotent, and no close retries +itself or overlaps a still-running attempt. `internal/config/guest_writer_control_test.go` exercises blocked snapshots at 25/100 updates and caller-copy ownership. `guest_metadata_writer_test.go` covers diff --git a/internal/config/guest_metadata.go b/internal/config/guest_metadata.go index d68e281060..cba3f71834 100644 --- a/internal/config/guest_metadata.go +++ b/internal/config/guest_metadata.go @@ -42,9 +42,13 @@ type GuestMetadataStore struct { lastWriteErr error writerDone chan struct{} closed atomic.Bool - closeOnce sync.Once - closeDone chan struct{} - closeErr error + closeMu sync.Mutex + closeAttempt *guestMetadataCloseAttempt +} + +type guestMetadataCloseAttempt struct { + done chan struct{} + err error } var ErrGuestMetadataStoreClosed = errors.New("guest metadata store is closed") @@ -177,27 +181,49 @@ func (s *GuestMetadataStore) Close(timeout time.Duration) error { } // Seal before waiting for a synchronous mutation's I/O-held memory lock. // One owned close observer, not one goroutine per waiter, covers both that - // mutation and the background writer without extending the caller's budget. - s.closeOnce.Do(func() { - s.closed.Store(true) + // mutation and the background writer within the caller's unchanged budget. + s.closed.Store(true) + s.closeMu.Lock() + attempt := s.closeAttempt + retry := false + if attempt != nil { + select { + case <-attempt.done: + // A later explicit shutdown/offboarding attempt may flush retained + // state after the filesystem is repaired. Never reopen admission, + // retry an in-flight write or spin-retry a failed close on its own. + if attempt.err != nil { + attempt = nil + retry = true + } + default: + } + } + if attempt == nil { + attempt = &guestMetadataCloseAttempt{done: make(chan struct{})} + s.closeAttempt = attempt go func() { - s.mu.RLock() + s.mu.Lock() + if retry && s.lastWriteErr != nil && s.writerDone == nil { + s.queueSaveLocked() + } done := s.writerDone - s.mu.RUnlock() + s.mu.Unlock() if done != nil { <-done } s.mu.RLock() - s.closeErr = s.lastWriteErr + attempt.err = s.lastWriteErr s.mu.RUnlock() - close(s.closeDone) + close(attempt.done) }() - }) + } + s.closeMu.Unlock() timer := time.NewTimer(timeout) defer timer.Stop() select { - case <-s.closeDone: - return s.closeErr + case <-attempt.done: + return attempt.err case <-timer.C: return fmt.Errorf("guest metadata writes did not drain within %s", timeout) } @@ -223,10 +249,9 @@ func cloneGuestMetadata(meta *GuestMetadata) *GuestMetadata { // NewGuestMetadataStore creates a new metadata store func NewGuestMetadataStore(dataPath string, fs FileSystem) *GuestMetadataStore { store := &GuestMetadataStore{ - metadata: make(map[string]*GuestMetadata), - dataPath: dataPath, - fs: fs, - closeDone: make(chan struct{}), + metadata: make(map[string]*GuestMetadata), + dataPath: dataPath, + fs: fs, } if store.fs == nil { diff --git a/internal/config/guest_metadata_writer_test.go b/internal/config/guest_metadata_writer_test.go index d71b685f2d..f4c78ce497 100644 --- a/internal/config/guest_metadata_writer_test.go +++ b/internal/config/guest_metadata_writer_test.go @@ -359,3 +359,38 @@ func TestGuestMetadataWriterRealFilesystemTenantIsolation(t *testing.T) { t.Fatal("tenants share identity") } } + +// Retention after a failed close is not a permanent offboarding dead end. +// A later explicit close may flush retained state after the filesystem recovers, +// without accepting new mutations or overlapping the original writer. +func TestGuestMetadataWriterExplicitCloseRetryPersistsRetainedState(t *testing.T) { + fs := &guestWriterFaultFS{FileSystem: defaultFileSystem{}} + fs.failWrite.Store(true) + store := NewGuestMetadataStore(t.TempDir(), fs) + if err := store.SetAsync("retained", &GuestMetadata{LastKnownName: "retained"}); err != nil { + t.Fatal(err) + } + if err := store.Close(time.Second); err == nil { + t.Fatal("first failed close reported success") + } + if fs.writes.Load() != 1 { + t.Fatalf("failed close retried without an explicit later call: %d", fs.writes.Load()) + } + if err := store.SetAsync("late", &GuestMetadata{}); !errors.Is(err, ErrGuestMetadataStoreClosed) { + t.Fatalf("failed close reopened admission: %v", err) + } + fs.failWrite.Store(false) + if err := store.Close(time.Second); err != nil { + t.Fatal(err) + } + if err := store.Close(time.Second); err != nil { + t.Fatal(err) + } + if fs.writes.Load() != 2 { + t.Fatalf("explicit recovery or successful idempotent close wrote unexpected snapshots: %d", fs.writes.Load()) + } + reloaded := NewGuestMetadataStore(store.dataPath, nil) + if reloaded.Get("retained") == nil || reloaded.Get("late") != nil { + t.Fatal("explicit close recovery lost retained state or wrote a late admission") + } +} From 13f6d107988b40f3395bf95aaee053a2378891d4 Mon Sep 17 00:00:00 2001 From: "pulse-triage[bot]" <249995291+pulse-triage[bot]@users.noreply.github.com> Date: Mon, 5 Oct 2026 09:38:34 +0100 Subject: [PATCH 03/12] Supply real GNU timeout to native macOS guidance checks Both macOS native jobs fail the copied temperature recipes before reading any sensor. Install coreutils into the workflow PATH and verify GNU timeout for every Unix runner. Exercise the actual workflow snippets with real GNU timeout and fail-closed install, missing-tool and unsupported-tool controls; leave the recipes and their TERM/KILL/privacy controls unchanged. Contract-Neutral: Native CI prerequisite and its fixture coverage only; no installer, runtime, API, permission or diagnostic contract changes. Change-source: pulse-maintainer --- .github/workflows/unified-agent-native.yml | 18 +++ .../native_guidance_tools_test.go | 153 ++++++++++++++++++ 2 files changed, 171 insertions(+) create mode 100644 scripts/installtests/native_guidance_tools_test.go diff --git a/.github/workflows/unified-agent-native.yml b/.github/workflows/unified-agent-native.yml index ca020f2c39..94ae17e0af 100644 --- a/.github/workflows/unified-agent-native.yml +++ b/.github/workflows/unified-agent-native.yml @@ -84,6 +84,24 @@ jobs: go-version-file: go.mod cache: true + - name: Supply GNU timeout for macOS guidance checks + if: runner.os == 'macOS' + shell: bash + run: | + set -euo pipefail + brew install coreutils + coreutils_bin="$(brew --prefix coreutils)/libexec/gnubin" + test -x "$coreutils_bin/timeout" + "$coreutils_bin/timeout" --version | grep -q 'GNU coreutils' + printf '%s\n' "$coreutils_bin" >> "$GITHUB_PATH" + + - name: Verify native Unix timeout prerequisite + if: matrix.unix + shell: bash + run: | + set -euo pipefail + timeout --version | grep -q 'GNU coreutils' + - name: Test native agent runtime run: >- go test diff --git a/scripts/installtests/native_guidance_tools_test.go b/scripts/installtests/native_guidance_tools_test.go new file mode 100644 index 0000000000..2ba6999c58 --- /dev/null +++ b/scripts/installtests/native_guidance_tools_test.go @@ -0,0 +1,153 @@ +//go:build !windows + +package installtests + +import ( + "context" + "fmt" + "os" + "os/exec" + "path/filepath" + "strings" + "testing" + "time" + + "gopkg.in/yaml.v3" +) + +type nativeGuidanceStep struct { + Name, If, Shell, Run string +} + +func nativeGuidanceToolSteps(content []byte) (nativeGuidanceStep, nativeGuidanceStep, error) { + var workflow struct { + Jobs map[string]struct{ Steps []nativeGuidanceStep } + } + if err := yaml.Unmarshal(content, &workflow); err != nil { + return nativeGuidanceStep{}, nativeGuidanceStep{}, err + } + steps := workflow.Jobs["native-agent"].Steps + bootstrapIndex, verifyIndex, installerIndex := -1, -1, -1 + for i, step := range steps { + switch step.Name { + case "Supply GNU timeout for macOS guidance checks": + bootstrapIndex = i + case "Verify native Unix timeout prerequisite": + verifyIndex = i + case "Test native Unix installer contracts": + installerIndex = i + } + } + if bootstrapIndex < 0 || verifyIndex <= bootstrapIndex || installerIndex <= verifyIndex { + return nativeGuidanceStep{}, nativeGuidanceStep{}, fmt.Errorf("native guidance needs GNU tool supply and verification before installer tests") + } + bootstrap, verify := steps[bootstrapIndex], steps[verifyIndex] + if bootstrap.If != "runner.os == 'macOS'" || verify.If != "matrix.unix" || steps[installerIndex].If != "matrix.unix" || bootstrap.Shell != "bash" || verify.Shell != "bash" { + return bootstrap, verify, fmt.Errorf("GNU tool preparation must cover macOS, verification and installer tests must cover every native Unix job") + } + return bootstrap, verify, nil +} + +func TestNativeGuidanceToolsSupplyAndVerifyRealGNUTimeout(t *testing.T) { + content, err := os.ReadFile(repoFile(".github", "workflows", "unified-agent-native.yml")) + if err != nil { + t.Fatal(err) + } + bootstrap, verify, err := nativeGuidanceToolSteps(content) + if err != nil { + t.Fatal(err) + } + realTimeout, err := exec.LookPath("timeout") + if err != nil { + t.Fatal("native Unix verification requires the real GNU timeout prerequisite") + } + for _, scenario := range []string{"real GNU tool", "install failure", "missing tool", "non-GNU tool"} { + t.Run(scenario, func(t *testing.T) { + root := t.TempDir() + bin, prefix := filepath.Join(root, "bin"), filepath.Join(root, "coreutils with spaces") + gnuBin := filepath.Join(prefix, "libexec", "gnubin") + for _, dir := range []string{bin, gnuBin} { + if err := os.MkdirAll(dir, 0700); err != nil { + t.Fatal(err) + } + } + // Only Homebrew's installer is simulated. The positive timeout is + // the real executable; the copied TERM/KILL recipes remain unchanged. + const brew = `#!/bin/sh +printf '%s\n' "$*" >> "$BREW_CALLS" +case "$*" in + 'install coreutils') exit "${BREW_EXIT:-0}" ;; + '--prefix coreutils') printf '%s\n' "$COREUTILS_PREFIX" ;; + *) exit 64 ;; +esac +` + if err := os.WriteFile(filepath.Join(bin, "brew"), []byte(brew), 0700); err != nil { + t.Fatal(err) + } + tool := filepath.Join(gnuBin, "timeout") + switch scenario { + case "real GNU tool", "install failure": + if err := os.Symlink(realTimeout, tool); err != nil { + t.Fatal(err) + } + case "non-GNU tool": + if err := os.WriteFile(tool, []byte("#!/bin/sh\nprintf 'BSD timeout\\n'\n"), 0700); err != nil { + t.Fatal(err) + } + } + pathFile, callsFile := filepath.Join(root, "github-path"), filepath.Join(root, "brew-calls") + installExit := "0" + if scenario == "install failure" { + installExit = "23" + } + env := append(os.Environ(), "PATH="+bin+":"+os.Getenv("PATH"), "GITHUB_PATH="+pathFile, + "COREUTILS_PREFIX="+prefix, "BREW_CALLS="+callsFile, "BREW_EXIT="+installExit) + ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second) + defer cancel() + cmd := exec.CommandContext(ctx, "bash", "-c", bootstrap.Run) + cmd.Env = env + output, runErr := cmd.CombinedOutput() + if ctx.Err() != nil { + t.Fatalf("prerequisite observation did not finish: %v", ctx.Err()) + } + if scenario != "real GNU tool" { + wantExit := 1 + if scenario == "install failure" { + wantExit = 23 + } + exit, ok := runErr.(*exec.ExitError) + if !ok || exit.ExitCode() != wantExit { + t.Fatalf("prerequisite exit %v, want %d: %s", runErr, wantExit, output) + } + if _, err := os.Stat(pathFile); !os.IsNotExist(err) { + t.Fatal("failed prerequisite exported an executable search path") + } + return + } + if runErr != nil { + t.Fatalf("GNU prerequisite failed: %v: %s", runErr, output) + } + path, err := os.ReadFile(pathFile) + if err != nil || string(path) != gnuBin+"\n" { + t.Fatal("bootstrap did not export the exact installed gnubin path") + } + calls, err := os.ReadFile(callsFile) + if err != nil || string(calls) != "install coreutils\n--prefix coreutils\n" { + t.Fatal("bootstrap changed its fixed Homebrew dependency") + } + cmd = exec.CommandContext(ctx, "bash", "-c", verify.Run) + cmd.Env = append(env, "PATH="+strings.TrimSpace(string(path))+":"+os.Getenv("PATH")) + if output, err := cmd.CombinedOutput(); err != nil { + t.Fatalf("next workflow step cannot use supplied GNU timeout: %v: %s", err, output) + } + cmd = exec.CommandContext(ctx, "bash", "-c", verify.Run) + cmd.Env = append(env, "PATH="+bin+":"+os.Getenv("PATH")) + if err := os.WriteFile(filepath.Join(bin, "timeout"), []byte("#!/bin/sh\nprintf 'BSD timeout\\n'\n"), 0700); err != nil { + t.Fatal(err) + } + if output, err := cmd.CombinedOutput(); err == nil { + t.Fatalf("Unix prerequisite probe accepted a non-GNU timeout: %s", output) + } + }) + } +} From 443faa53b6c9a651b856bbfd1846fa092918a06a Mon Sep 17 00:00:00 2001 From: "pulse-triage[bot]" <249995291+pulse-triage[bot]@users.noreply.github.com> Date: Mon, 5 Oct 2026 09:44:52 +0100 Subject: [PATCH 04/12] Keep macOS native utilities unchanged around timeout bootstrap Expose only the installed GNU gtimeout as timeout through a new runner-owned directory. Keep native BSD utilities on their original PATH so installing the missing dependency cannot silently change the other platform checks. The connected workflow test verifies the symlink identity and that the directory contains no other utilities. Contract-Neutral: Narrow the native CI dependency PATH and its proof; no shipped installer or runtime contract changes. Change-source: pulse-maintainer --- .github/workflows/unified-agent-native.yml | 10 +++++--- .../native_guidance_tools_test.go | 25 +++++++++++++------ 2 files changed, 24 insertions(+), 11 deletions(-) diff --git a/.github/workflows/unified-agent-native.yml b/.github/workflows/unified-agent-native.yml index 94ae17e0af..ec5fa82dc2 100644 --- a/.github/workflows/unified-agent-native.yml +++ b/.github/workflows/unified-agent-native.yml @@ -90,10 +90,12 @@ jobs: run: | set -euo pipefail brew install coreutils - coreutils_bin="$(brew --prefix coreutils)/libexec/gnubin" - test -x "$coreutils_bin/timeout" - "$coreutils_bin/timeout" --version | grep -q 'GNU coreutils' - printf '%s\n' "$coreutils_bin" >> "$GITHUB_PATH" + gnu_timeout="$(brew --prefix coreutils)/bin/gtimeout" + test -x "$gnu_timeout" + "$gnu_timeout" --version | grep -q 'GNU coreutils' + timeout_bin="$(mktemp -d "$RUNNER_TEMP/pulse-guidance-tools.XXXXXX")" + ln -s "$gnu_timeout" "$timeout_bin/timeout" + printf '%s\n' "$timeout_bin" >> "$GITHUB_PATH" - name: Verify native Unix timeout prerequisite if: matrix.unix diff --git a/scripts/installtests/native_guidance_tools_test.go b/scripts/installtests/native_guidance_tools_test.go index 2ba6999c58..c136e70d7f 100644 --- a/scripts/installtests/native_guidance_tools_test.go +++ b/scripts/installtests/native_guidance_tools_test.go @@ -65,8 +65,8 @@ func TestNativeGuidanceToolsSupplyAndVerifyRealGNUTimeout(t *testing.T) { t.Run(scenario, func(t *testing.T) { root := t.TempDir() bin, prefix := filepath.Join(root, "bin"), filepath.Join(root, "coreutils with spaces") - gnuBin := filepath.Join(prefix, "libexec", "gnubin") - for _, dir := range []string{bin, gnuBin} { + coreutilsBin, runnerTemp := filepath.Join(prefix, "bin"), filepath.Join(root, "runner temp") + for _, dir := range []string{bin, coreutilsBin, runnerTemp} { if err := os.MkdirAll(dir, 0700); err != nil { t.Fatal(err) } @@ -84,7 +84,7 @@ esac if err := os.WriteFile(filepath.Join(bin, "brew"), []byte(brew), 0700); err != nil { t.Fatal(err) } - tool := filepath.Join(gnuBin, "timeout") + tool := filepath.Join(coreutilsBin, "gtimeout") switch scenario { case "real GNU tool", "install failure": if err := os.Symlink(realTimeout, tool); err != nil { @@ -101,7 +101,7 @@ esac installExit = "23" } env := append(os.Environ(), "PATH="+bin+":"+os.Getenv("PATH"), "GITHUB_PATH="+pathFile, - "COREUTILS_PREFIX="+prefix, "BREW_CALLS="+callsFile, "BREW_EXIT="+installExit) + "COREUTILS_PREFIX="+prefix, "BREW_CALLS="+callsFile, "BREW_EXIT="+installExit, "RUNNER_TEMP="+runnerTemp) ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second) defer cancel() cmd := exec.CommandContext(ctx, "bash", "-c", bootstrap.Run) @@ -128,15 +128,26 @@ esac t.Fatalf("GNU prerequisite failed: %v: %s", runErr, output) } path, err := os.ReadFile(pathFile) - if err != nil || string(path) != gnuBin+"\n" { - t.Fatal("bootstrap did not export the exact installed gnubin path") + if err != nil { + t.Fatal(err) + } + timeoutBin := strings.TrimSuffix(string(path), "\n") + if filepath.Dir(timeoutBin) != runnerTemp || !strings.HasPrefix(filepath.Base(timeoutBin), "pulse-guidance-tools.") { + t.Fatal("bootstrap did not export its newly owned timeout directory") + } + entries, err := os.ReadDir(timeoutBin) + if err != nil || len(entries) != 1 || entries[0].Name() != "timeout" { + t.Fatal("bootstrap must not replace native macOS utilities other than timeout") + } + if target, err := os.Readlink(filepath.Join(timeoutBin, "timeout")); err != nil || target != tool { + t.Fatal("timeout must resolve to the exact installed GNU executable") } calls, err := os.ReadFile(callsFile) if err != nil || string(calls) != "install coreutils\n--prefix coreutils\n" { t.Fatal("bootstrap changed its fixed Homebrew dependency") } cmd = exec.CommandContext(ctx, "bash", "-c", verify.Run) - cmd.Env = append(env, "PATH="+strings.TrimSpace(string(path))+":"+os.Getenv("PATH")) + cmd.Env = append(env, "PATH="+timeoutBin+":"+os.Getenv("PATH")) if output, err := cmd.CombinedOutput(); err != nil { t.Fatalf("next workflow step cannot use supplied GNU timeout: %v: %s", err, output) } From 54de7bdca738ae9d69b1856befcd4d4a60fbfe5d Mon Sep 17 00:00:00 2001 From: "pulse-triage[bot]" <249995291+pulse-triage[bot]@users.noreply.github.com> Date: Mon, 5 Oct 2026 11:22:07 +0100 Subject: [PATCH 05/12] Capture fresh broadcast resources and history targets together Resolve canonical source mappings once per complete projection while preserving detached resource content and target selection after host coalescing. Bind resources and targets to the same locked generation, including absence and continuity overlays, and avoid another owned outer-slice copy. No timestamp cache or native CPU/RSS claim. Retain cold, clean and dirty full-content controls and paired complete-projection measurements. Contract-Neutral: Internal bulk-read and copy-ownership repair preserves existing public resource/history fields, source precedence, agent identity/lifecycle, tenant/revocation and safety contracts; no wire or permission delta. Change-source: pulse-maintainer --- .../v6/internal/subsystems/monitoring.md | 24 ++ .../internal/subsystems/unified-resources.md | 23 ++ .../broadcast_metrics_snapshot_test.go | 264 ++++++++++++++++++ .../monitoring/canonical_guardrails_test.go | 7 +- internal/monitoring/monitor.go | 83 ++++-- .../monitoring/monitor_host_agents_test.go | 4 + internal/unifiedresources/monitor_adapter.go | 12 + .../monitor_adapter_read_state_test.go | 37 +++ internal/unifiedresources/registry.go | 31 +- .../registry_projection_test.go | 99 +++++++ internal/unifiedresources/registry_test.go | 65 +++++ 11 files changed, 623 insertions(+), 26 deletions(-) create mode 100644 internal/monitoring/broadcast_metrics_snapshot_test.go create mode 100644 internal/unifiedresources/registry_projection_test.go diff --git a/docs/release-control/v6/internal/subsystems/monitoring.md b/docs/release-control/v6/internal/subsystems/monitoring.md index 354ad8d5e2..fd23f74cc1 100644 --- a/docs/release-control/v6/internal/subsystems/monitoring.md +++ b/docs/release-control/v6/internal/subsystems/monitoring.md @@ -17,6 +17,30 @@ ## Purpose +### Fresh broadcast projection owns one resource/target capture + +Canonical live, fallback and standalone-host continuity broadcast reads prefer +the adapter's detached bulk resource/metrics-target capture. After host +coalescing selects presentation identity, missing history targets are attached +from that same capture, including authoritative absence. Stores without this +optional capability retain the previous point-resolver fallback. Only the owned +outer coalesced slice is decorated in place; nested source payloads stay read-only. + +Every read still observes complete current resource data and derives live alert +health, metadata URL edits/clears and connected/ignored infrastructure. No +timestamp-only cache, telemetry omission, policy change or new wire field is +introduced. All existing catalogs, discovery/history coordinates, original +observation times, LastSeen, identity and tenant/revocation boundaries remain. + +`broadcast_metrics_snapshot_test.go` compares the complete served JSON with an +independent earlier projection across the metadata-heavy nine-provider demo, +typed-view and same-time dirty cases; it also covers captured target replacement +and continuity/metadata/live-alert boundaries. `monitor_host_agents_test.go` +retains its one-materialization and live-change regression. Benchmarks include +the exact parent's complete conversion and final JSON encoding, with cold, +list-only, typed-clean and dirty inputs; they do not establish installed CPU/RSS +relief, native identity refresh throughput or completion of #2199. + ### Report identity lookups do not rebuild unrelated Kubernetes metadata The producer's two temporary host/Docker-host admission registries use a diff --git a/docs/release-control/v6/internal/subsystems/unified-resources.md b/docs/release-control/v6/internal/subsystems/unified-resources.md index 829cedf05e..1bd25916bf 100644 --- a/docs/release-control/v6/internal/subsystems/unified-resources.md +++ b/docs/release-control/v6/internal/subsystems/unified-resources.md @@ -23,6 +23,29 @@ and sort the complete canonical change table while startup and ingestion wait. ## Purpose +### Broadcast bulk reads capture resources and history coordinates together + +`ListWithMetricsTargets` and the monitor adapter's matching optional read return +detached resources and a detached target map from one locked registry generation. +The existing source-mapping inversion runs once for this read when typed views +are dirty; it does not build all typed views or add a retained generation cache. +Clean typed views may supply their already-current inverse mapping. Legacy List, +GetAll and individual target reads preserve their shapes and semantics. + +The target map stays separate from resources until presentation host coalescing +selects the row identity. Missing coordinates remain missing in that capture; +they never fall through to a later live registry. Mapping changes with unchanged +LastSeen take effect on the next read. Returned resources, policy/identity blocks +and target values remain caller-owned across mutations and adapter replacement. +Projection cannot pair an earlier resource row with a later history identity. + +`registry_test.go`, `registry_projection_test.go` and +`monitor_adapter_read_state_test.go` exercise full resource/point-target parity, +same-time edits, concurrent mapping writes, detachments, empty/nil reads and +tenant separation. Connected monitoring proof covers all nine demo providers, +continuity overlays and complete served JSON. Paired cold/clean/dirty benchmarks +measure whole materialization, not a new point-lookup envelope or native relief. + Canonical resource History honours selected memory observation provenance and Proxmox VM filesystem deferrals instead of treating numeric carriers as fresh measurements. ### Canonical resource History cannot promote retained guest readings diff --git a/internal/monitoring/broadcast_metrics_snapshot_test.go b/internal/monitoring/broadcast_metrics_snapshot_test.go new file mode 100644 index 0000000000..4e61cc2f3d --- /dev/null +++ b/internal/monitoring/broadcast_metrics_snapshot_test.go @@ -0,0 +1,264 @@ +package monitoring + +import ( + "encoding/json" + "fmt" + "reflect" + "testing" + "time" + + "github.com/rcourtman/pulse-go-rewrite/internal/config" + "github.com/rcourtman/pulse-go-rewrite/internal/mock" + "github.com/rcourtman/pulse-go-rewrite/internal/models" + "github.com/rcourtman/pulse-go-rewrite/internal/unifiedresources" +) + +func projectionFixtureGraph(t testing.TB) mock.FixtureGraph { + t.Helper() + previousEnabled, previousConfig := mock.IsMockEnabled(), mock.GetConfig() + t.Cleanup(func() { + mustSetMockEnabled(t, false) + mock.SetMockConfig(previousConfig) + mustSetMockEnabled(t, previousEnabled) + }) + mustSetMockEnabled(t, false) + cfg := mock.DefaultConfig + cfg.UpdateInterval = 5 * time.Minute + mock.SetMockConfig(cfg) + mustSetMockEnabled(t, true) + graph := mock.CurrentFixtureGraph() + mustSetMockEnabled(t, false) + return graph +} + +func projectionFixtureRegistry(graph mock.FixtureGraph) *unifiedresources.ResourceRegistry { + rr := unifiedresources.NewRegistry(nil) + rr.IngestSnapshot(graph.State) + for _, source := range []unifiedresources.DataSource{unifiedresources.SourceTrueNAS, unifiedresources.SourceVMware, unifiedresources.SourceAvailability} { + rr.IngestRecords(source, graph.SupplementalRecords(source)) + } + return rr +} + +// The original full projection, including target resolution after host +// coalescing. Compare all wire fields/catalogues, not selected metric values. +func legacyFrontendProjectionForTest(m *Monitor, snapshot models.StateSnapshot, resources []unifiedresources.Resource, resolver MetricsTargetResourceStore, now time.Time) models.StateFrontend { + rows := unifiedresources.CoalescePresentationHostResources(resources) + rows = m.applyPersistedMetadataToUnifiedResources(rows) + rows = unifiedresources.AttachResourceHealth(rows, resourceHealthAlerts(snapshot.ActiveAlerts), now) + projected, catalogs := convertResourcesForBroadcastReference(rows, resolver) + out := snapshot.ToFrontend() + out.Resources = projected + out.CapabilityCatalog = catalogs.capabilities + out.PolicyCatalog = catalogs.policies + out.AISafeSummaryCatalog = catalogs.aiSafeSummaries + out.ConnectedInfrastructure = buildConnectedInfrastructure(rows, snapshot) + return out +} + +func TestBroadcastMetricsSnapshotConnectedContent(t *testing.T) { + graph := projectionFixtureGraph(t) + rr := projectionFixtureRegistry(graph) + adapter := unifiedresources.NewMonitorAdapter(rr) + store := &broadcastProjectionCountingStore{MonitorAdapter: adapter} + m := &Monitor{resourceStore: store} + resources := adapter.GetAll() + if len(resources) < 1000 { + t.Fatalf("demo fixture too small: %d", len(resources)) + } + sources := map[unifiedresources.DataSource]bool{} + for _, r := range resources { + for _, source := range r.Sources { + sources[source] = true + } + } + for _, source := range []unifiedresources.DataSource{unifiedresources.SourceProxmox, unifiedresources.SourceAgent, unifiedresources.SourceDocker, unifiedresources.SourcePBS, unifiedresources.SourcePMG, unifiedresources.SourceK8s, unifiedresources.SourceTrueNAS, unifiedresources.SourceVMware, unifiedresources.SourceAvailability} { + if !sources[source] { + t.Fatalf("demo omits provider %s", source) + } + } + assertEqual := func(stage string) { + t.Helper() + before, err := json.Marshal(adapter.GetAll()) + if err != nil { + t.Fatal(err) + } + want := legacyFrontendProjectionForTest(m, graph.State, adapter.GetAll(), adapter, time.Now().UTC()) + got := m.buildBroadcastFrontendStateFromSnapshot(graph.State) + wantJSON, err := json.Marshal(want) + if err != nil { + t.Fatal(err) + } + gotJSON, err := json.Marshal(got) + if err != nil { + t.Fatal(err) + } + if string(wantJSON) != string(gotJSON) { + t.Fatalf("%s: complete frontend/catalogue/infrastructure JSON differs", stage) + } + after, _ := json.Marshal(adapter.GetAll()) + if string(before) != string(after) { + t.Fatalf("%s: projection changed source data", stage) + } + t.Logf("%s: %d complete resources, %d frontend rows, %d wire bytes equal legacy projection", stage, len(adapter.GetAll()), len(got.Resources), len(gotJSON)) + } + assertEqual("cold-list-only") + rr.Workloads() + assertEqual("typed-views-clean") + // Same timestamps, new payload, policy and URL. No time-only invalidation. + resources = adapter.GetAll() + resources[0].Name = "changed-resource" + resources[0].Tags = []string{"customer-data"} + resources[0].CustomURL = "https://example.invalid/changed" + rr.IngestResources(resources) + assertEqual("dirty-same-time") +} + +func TestBroadcastMetricsSnapshotOwnsTargetsBeforeLiveReplacement(t *testing.T) { + m, adapter, _ := newReadStateCloneTestMonitor(t, 4) + view := m.currentUnifiedStateView() + if view.metricsTargets == nil { + t.Fatal("broadcast did not capture a bulk resolver") + } + old := view.resources[0] + want := adapter.MetricsTargetForResource(old.ID) + adapter.PopulateFromSnapshot(models.EmptyStateSnapshot()) + if got := view.metricsTargets.MetricsTargetForResource(old.ID); !reflect.DeepEqual(got, want) { + t.Fatal("captured resources were paired with a later generation's target") + } + if got := view.metricsTargets.MetricsTargetForResource("unknown"); got != nil { + t.Fatal("unknown target fell through to live data") + } + copyTarget := view.metricsTargets.MetricsTargetForResource(old.ID) + copyTarget.ResourceID = "caller-edit" + if got := view.metricsTargets.MetricsTargetForResource(old.ID); got.ResourceID == "caller-edit" { + t.Fatal("caller mutated the captured resolver") + } +} + +func TestBroadcastMetricsSnapshotContinuityAndLiveOverlays(t *testing.T) { + m, adapter, _ := newReadStateCloneTestMonitor(t, 2) + now := time.Now().UTC() + m.hostContinuityStore = config.NewHostContinuityStore(t.TempDir(), nil) + entry := config.HostContinuityEntry{HostID: "absent-host", Hostname: "absent.example", MachineID: "absent-machine", LastSeen: now.Add(-time.Hour)} + if err := m.hostContinuityStore.Upsert(entry); err != nil { + t.Fatal(err) + } + view := m.unifiedStateViewWithStandaloneHostContinuity(monitorUnifiedStateView{readState: adapter}) + if len(view.resources) != 3 || view.metricsTargets == nil { + t.Fatal("continuity projection omitted its resources or captured resolver") + } + var host unifiedresources.Resource + for _, r := range view.resources { + if r.Agent != nil && r.Agent.AgentID == entry.HostID { + host = r + } + if got, want := view.metricsTargets.MetricsTargetForResource(r.ID), broadcastMetricsTargetResolver(view.readState).MetricsTargetForResource(r.ID); !reflect.DeepEqual(got, want) { + t.Fatal("overlay target differs from overlay registry") + } + } + if host.ID == "" || host.Status == unifiedresources.StatusOnline || !host.LastSeen.Equal(entry.LastSeen) { + t.Fatal("continuity history was omitted or promoted to current") + } + m.hostMetadataStore = config.NewHostMetadataStore(t.TempDir(), nil) + for _, url := range []string{"https://example.invalid/operator", ""} { + if err := m.hostMetadataStore.Set(entry.HostID, &config.HostMetadata{CustomURL: url}); err != nil { + t.Fatal(err) + } + decorated := m.applyPersistedMetadataToUnifiedResources(view.resources) + for _, r := range decorated { + if r.ID == host.ID && r.CustomURL != url { + t.Fatal("operator URL change/clear was hidden by captured data") + } + } + } + critical := unifiedresources.EvaluateResourceHealth(host, []unifiedresources.ResourceHealthAlert{{ResourceID: host.ID, Level: "critical", Type: "cpu"}}, now) + if critical.Verdict != unifiedresources.HealthCritical { + t.Fatal("live alert was hidden by captured data") + } + if len(adapter.GetAll()) != 2 { + t.Fatal("continuity overlay wrote back to the live registry") + } +} + +var completeProjectionSink models.StateFrontend +var completeProjectionJSONSink []byte + +// Exact parent broadcast shape (one coalesce, copy-to-attach, then conversion), +// not the older two-coalesce content oracle above. The unchanged conversion, +// metadata/health decoration and snapshot projection are part of both costs. +func parentFrontendProjectionForTest(m *Monitor, snapshot models.StateSnapshot, resources []unifiedresources.Resource, resolver MetricsTargetResourceStore, now time.Time) models.StateFrontend { + rows := unifiedresources.CoalescePresentationHostResources(resources) + for i := range rows { + m.applyPersistedMetadataToUnifiedResource(&rows[i]) + health := unifiedresources.EvaluateResourceHealth(rows[i], resourceHealthAlerts(snapshot.ActiveAlerts), now) + rows[i].Health = &health + } + projected, catalogs := convertPresentationResourcesForBroadcast(attachBroadcastMetricsTargets(rows, resolver)) + out := snapshot.ToFrontend() + out.Resources = projected + out.CapabilityCatalog = catalogs.capabilities + out.PolicyCatalog = catalogs.policies + out.AISafeSummaryCatalog = catalogs.aiSafeSummaries + out.ConnectedInfrastructure = buildConnectedInfrastructure(rows, snapshot) + return out +} + +func BenchmarkBroadcastMetricsSnapshot(b *testing.B) { + graph := projectionFixtureGraph(b) + for _, mode := range []string{"cold", "list-clean", "views-clean", "dirty"} { + for _, encode := range []bool{false, true} { + stage := "projection" + if encode { + stage = "json-frame" + } + for _, bulk := range []bool{false, true} { + path := "point-targets" + if bulk { + path = "bulk-targets" + } + b.Run(fmt.Sprintf("%s/%s/%s", mode, stage, path), func(b *testing.B) { + rr := projectionFixtureRegistry(graph) + adapter := unifiedresources.NewMonitorAdapter(rr) + store := &broadcastProjectionCountingStore{MonitorAdapter: adapter} + m := &Monitor{resourceStore: store} + if mode != "cold" { + rr.List() + } + if mode == "views-clean" { + rr.Workloads() + } + b.ReportAllocs() + b.ResetTimer() + for i := 0; i < b.N; i++ { + if mode == "cold" || mode == "dirty" { + b.StopTimer() + if mode == "cold" { + rr = projectionFixtureRegistry(graph) + adapter = unifiedresources.NewMonitorAdapter(rr) + store.MonitorAdapter = adapter + } else { + changed := rr.List()[0] + changed.Tags = []string{fmt.Sprintf("edit-%d", i)} + rr.IngestResources([]unifiedresources.Resource{changed}) + } + b.StartTimer() + } + if bulk { + completeProjectionSink = m.buildBroadcastFrontendStateFromSnapshot(graph.State) + } else { + completeProjectionSink = parentFrontendProjectionForTest(m, graph.State, adapter.GetAll(), adapter, time.Now().UTC()) + } + if encode { + var err error + completeProjectionJSONSink, err = json.Marshal(completeProjectionSink) + if err != nil { + b.Fatal(err) + } + } + } + }) + } + } + } +} diff --git a/internal/monitoring/canonical_guardrails_test.go b/internal/monitoring/canonical_guardrails_test.go index 165f94a540..9e11cb4f7e 100644 --- a/internal/monitoring/canonical_guardrails_test.go +++ b/internal/monitoring/canonical_guardrails_test.go @@ -607,10 +607,11 @@ func TestBroadcastResourceProjectionCoalescesSplitHostIdentities(t *testing.T) { source := string(data) for _, snippet := range []string{ - "metricsTargetResolver := broadcastMetricsTargetResolver(unifiedView.readState)", + "metricsTargetResolver := unifiedView.metricsTargets", + "metricsTargetResolver = broadcastMetricsTargetResolver(unifiedView.readState)", "broadcastResources := unifiedresources.CoalescePresentationHostResources(unifiedView.resources)", - "broadcastFrontendResources, broadcastCatalogs := convertPresentationResourcesForBroadcast(", - "attachBroadcastMetricsTargets(broadcastResources, metricsTargetResolver)", + "broadcastFrontendResources, broadcastCatalogs := convertPresentationResourcesForBroadcast(broadcastResources)", + "attachBroadcastMetricsTargetsInPlace(broadcastResources, metricsTargetResolver)", "frontendState.CapabilityCatalog = broadcastCatalogs.capabilities", "frontendState.PolicyCatalog = broadcastCatalogs.policies", "frontendState.AISafeSummaryCatalog = broadcastCatalogs.aiSafeSummaries", diff --git a/internal/monitoring/monitor.go b/internal/monitoring/monitor.go index 6bab63084c..162d189642 100644 --- a/internal/monitoring/monitor.go +++ b/internal/monitoring/monitor.go @@ -4341,7 +4341,10 @@ func (m *Monitor) buildBroadcastFrontendStateFromSnapshot(snapshot models.StateS } } unifiedView := m.currentUnifiedStateView() - metricsTargetResolver := broadcastMetricsTargetResolver(unifiedView.readState) + metricsTargetResolver := unifiedView.metricsTargets + if metricsTargetResolver == nil { + metricsTargetResolver = broadcastMetricsTargetResolver(unifiedView.readState) + } broadcastResources := unifiedresources.CoalescePresentationHostResources(unifiedView.resources) // Coalescing owns the outer slice. Decorate that one projection in place, // not three full-resource copies; nested store data is still read-only. @@ -4352,9 +4355,8 @@ func (m *Monitor) buildBroadcastFrontendStateFromSnapshot(snapshot models.StateS health := unifiedresources.EvaluateResourceHealth(broadcastResources[i], healthAlerts, now) broadcastResources[i].Health = &health } - broadcastFrontendResources, broadcastCatalogs := convertPresentationResourcesForBroadcast( - attachBroadcastMetricsTargets(broadcastResources, metricsTargetResolver), - ) + attachBroadcastMetricsTargetsInPlace(broadcastResources, metricsTargetResolver) + broadcastFrontendResources, broadcastCatalogs := convertPresentationResourcesForBroadcast(broadcastResources) frontendState.Resources = broadcastFrontendResources frontendState.CapabilityCatalog = broadcastCatalogs.capabilities frontendState.PolicyCatalog = broadcastCatalogs.policies @@ -4871,23 +4873,50 @@ func (m *Monitor) RecentResourceChanges(resourceID string, since time.Time, limi } type monitorUnifiedStateView struct { - resources []unifiedresources.Resource - readState unifiedresources.ReadState - freshness time.Time + resources []unifiedresources.Resource + readState unifiedresources.ReadState + freshness time.Time + metricsTargets MetricsTargetResourceStore } type unifiedResourceReadStateLister interface { GetAll() []unifiedresources.Resource } +type unifiedResourceProjectionLister interface { + GetAllWithMetricsTargets() ([]unifiedresources.Resource, map[string]unifiedresources.MetricsTarget) +} + +// This resolver never falls through to the live store, including for an +// unknown ID. A missing target belongs to this captured generation too. +type projectionMetricsTargets map[string]unifiedresources.MetricsTarget + +func (targets projectionMetricsTargets) MetricsTargetForResource(id string) *unifiedresources.MetricsTarget { + target, ok := targets[unifiedresources.CanonicalResourceID(id)] + if !ok { + return nil + } + return &target +} + +func unifiedProjectionResources(lister unifiedResourceReadStateLister) ([]unifiedresources.Resource, MetricsTargetResourceStore) { + if projection, ok := lister.(unifiedResourceProjectionLister); ok { + resources, targets := projection.GetAllWithMetricsTargets() + return resources, projectionMetricsTargets(targets) + } + return lister.GetAll(), nil +} + func monitorUnifiedStateViewFromSnapshot(snapshot models.StateSnapshot) monitorUnifiedStateView { registry := unifiedresources.NewRegistry(nil) registry.IngestSnapshot(snapshot) adapter := unifiedresources.NewMonitorAdapter(registry) + resources, targets := unifiedProjectionResources(adapter) return monitorUnifiedStateView{ - resources: registry.List(), - readState: adapter, - freshness: snapshot.LastUpdate, + resources: resources, + readState: adapter, + freshness: snapshot.LastUpdate, + metricsTargets: targets, } } @@ -4895,10 +4924,12 @@ func monitorUnifiedStateViewFromResources(resources []unifiedresources.Resource, registry := unifiedresources.NewRegistry(nil) registry.IngestResources(resources) adapter := unifiedresources.NewMonitorAdapter(registry) + resources, targets := unifiedProjectionResources(adapter) return monitorUnifiedStateView{ - resources: registry.List(), - readState: adapter, - freshness: freshness, + resources: resources, + readState: adapter, + freshness: freshness, + metricsTargets: targets, } } @@ -4915,8 +4946,9 @@ func (m *Monitor) unifiedStateViewWithStandaloneHostContinuity(view monitorUnifi return view } - resources := lister.GetAll() + resources, targets := unifiedProjectionResources(lister) view.resources = resources + view.metricsTargets = targets if view.freshness.IsZero() { view.freshness = latestUnifiedResourceLastSeen(resources) } @@ -6479,17 +6511,24 @@ func attachBroadcastMetricsTargets( } out := make([]unifiedresources.Resource, len(resources)) - for i, resource := range resources { - if resource.MetricsTarget == nil { - resource.MetricsTarget = validBroadcastMetricsTarget( - resolver.MetricsTargetForResource(resource.ID), - ) - } - out[i] = resource - } + copy(out, resources) + attachBroadcastMetricsTargetsInPlace(out, resolver) return out } +// Only call with the owned outer slice of a coalesced broadcast projection. +// Nested resource data stays read-only, and the resolver returns owned targets. +func attachBroadcastMetricsTargetsInPlace(resources []unifiedresources.Resource, resolver MetricsTargetResourceStore) { + if resolver == nil { + return + } + for i := range resources { + if resources[i].MetricsTarget == nil { + resources[i].MetricsTarget = validBroadcastMetricsTarget(resolver.MetricsTargetForResource(resources[i].ID)) + } + } +} + func validBroadcastMetricsTarget( target *unifiedresources.MetricsTarget, ) *unifiedresources.MetricsTarget { diff --git a/internal/monitoring/monitor_host_agents_test.go b/internal/monitoring/monitor_host_agents_test.go index 12e934c142..695d475395 100644 --- a/internal/monitoring/monitor_host_agents_test.go +++ b/internal/monitoring/monitor_host_agents_test.go @@ -6718,6 +6718,10 @@ func (s *broadcastProjectionCountingStore) GetAll() []unifiedresources.Resource s.reads++ return s.MonitorAdapter.GetAll() } +func (s *broadcastProjectionCountingStore) GetAllWithMetricsTargets() ([]unifiedresources.Resource, map[string]unifiedresources.MetricsTarget) { + s.reads++ + return s.MonitorAdapter.GetAllWithMetricsTargets() +} func (*broadcastProjectionCountingStore) TryReplaceRegistryForRead(models.StateSnapshot, time.Duration, func() map[unifiedresources.DataSource][]unifiedresources.IngestRecord) bool { return false } diff --git a/internal/unifiedresources/monitor_adapter.go b/internal/unifiedresources/monitor_adapter.go index 28eda9d110..8bac41ceaa 100644 --- a/internal/unifiedresources/monitor_adapter.go +++ b/internal/unifiedresources/monitor_adapter.go @@ -496,6 +496,18 @@ func (a *MonitorAdapter) GetAll() []Resource { return registry.List() } +// GetAllWithMetricsTargets materializes one complete broadcast read, binding +// its detached target map to the same registry as its detached resource list. +// The adapter may publish a replacement immediately after this capture; neither +// half of the returned read then consults that newer generation. +func (a *MonitorAdapter) GetAllWithMetricsTargets() ([]Resource, map[string]MetricsTarget) { + registry := a.currentRegistry() + if registry == nil { + return nil, nil + } + return registry.ListWithMetricsTargets() +} + func (a *MonitorAdapter) unifiedAIAdapter() *UnifiedAIAdapter { registry := a.currentRegistry() if registry == nil { diff --git a/internal/unifiedresources/monitor_adapter_read_state_test.go b/internal/unifiedresources/monitor_adapter_read_state_test.go index 5eb791f3cf..7eefcb5b99 100644 --- a/internal/unifiedresources/monitor_adapter_read_state_test.go +++ b/internal/unifiedresources/monitor_adapter_read_state_test.go @@ -971,3 +971,40 @@ func TestHostContinuityOverlayReusedWithinRegistryGeneration(t *testing.T) { t.Fatal("overlay was reused past its maximum age") } } + +func TestMonitorAdapterProjectionSnapshotKeepsReplacedGeneration(t *testing.T) { + var absent *MonitorAdapter + rows, targets := absent.GetAllWithMetricsTargets() + if rows != nil || targets != nil { + t.Fatal("nil adapter invented a projection") + } + adapter := NewMonitorAdapter(NewRegistry(nil)) + snapshot := readRefreshSnapshot("before") + adapter.PopulateFromSnapshot(snapshot) + first, firstTargets := adapter.GetAllWithMetricsTargets() + snapshot.VMs[0].Name = "after" + snapshot.VMs[0].ID = "new-node:201" + snapshot.VMs[0].VMID = 201 + // Observation time is deliberately unchanged. + adapter.PopulateFromSnapshot(snapshot) + second, secondTargets := adapter.GetAllWithMetricsTargets() + if len(first) != 1 || len(second) != 1 || first[0].Name != "before" || second[0].Name != "after" { + t.Fatal("replacement lost complete fresh resource content") + } + if firstTargets[first[0].ID].ResourceID == secondTargets[second[0].ID].ResourceID { + t.Fatal("replacement retained old history coordinates") + } + for _, r := range second { + if want := adapter.MetricsTargetForResource(r.ID); want == nil || *want != secondTargets[r.ID] { + t.Fatal("target differs from the generation's point resolver") + } + } + other := NewMonitorAdapter(NewRegistry(nil)) + snapshot.VMs[0].ID = "other-tenant:501" + snapshot.VMs[0].Name = "other-tenant" + other.PopulateFromSnapshot(snapshot) + otherRows, otherTargets := other.GetAllWithMetricsTargets() + if reflect.DeepEqual(otherRows, second) || reflect.DeepEqual(otherTargets, secondTargets) { + t.Fatal("separate adapters shared a tenant projection") + } +} diff --git a/internal/unifiedresources/registry.go b/internal/unifiedresources/registry.go index 774f2d8930..34992538c1 100644 --- a/internal/unifiedresources/registry.go +++ b/internal/unifiedresources/registry.go @@ -1321,6 +1321,21 @@ func seededVMwareSourceID(resource *Resource) string { // the read lock and copy it without repeating identity/scope/policy derivation. // The typed views remain lazy, so a List-only consumer pays for no extra copy. func (rr *ResourceRegistry) List() []Resource { + resources, _ := rr.listMaterialized(false) + return resources +} + +// ListWithMetricsTargets captures detached resources and their history targets +// from the same registry generation. Resolve all source mappings in one pass, +// without forcing the unrelated typed views to be built. In particular a +// freshly ingested registry must not scan every mapping once per broadcast row. +// Targets are separate: host coalescing must select the presentation identity +// before a missing target is attached, just as with individual target lookups. +func (rr *ResourceRegistry) ListWithMetricsTargets() ([]Resource, map[string]MetricsTarget) { + return rr.listMaterialized(true) +} + +func (rr *ResourceRegistry) listMaterialized(withTargets bool) ([]Resource, map[string]MetricsTarget) { for { rr.mu.RLock() if rr.canonicalMetadataDirty { @@ -1332,13 +1347,27 @@ func (rr *ResourceRegistry) List() []Resource { // rather than copy a generation that is no longer materialized. continue } + var sourceTargets map[string][]SourceTarget + var targets map[string]MetricsTarget + if withTargets { + sourceTargets = rr.cachedSourceTargets + if rr.viewsDirty || sourceTargets == nil { + sourceTargets = rr.buildSourceTargetsIndexLocked() + } + targets = make(map[string]MetricsTarget, len(rr.resources)) + } out := make([]Resource, 0, len(rr.resources)) for _, r := range rr.resources { out = append(out, cloneMaterializedResource(r)) + if withTargets { + if target := rr.metricsTargetFromSourceTargets(r, sourceTargets[r.ID]); target != nil { + targets[r.ID] = *target + } + } } rr.mu.RUnlock() sortResourcesByName(out) - return out + return out, targets } } diff --git a/internal/unifiedresources/registry_projection_test.go b/internal/unifiedresources/registry_projection_test.go new file mode 100644 index 0000000000..ca9dda8383 --- /dev/null +++ b/internal/unifiedresources/registry_projection_test.go @@ -0,0 +1,99 @@ +package unifiedresources + +import ( + "fmt" + "sync" + "testing" +) + +// Exercise the capture while a writer changes the mapping and row together. +// Separate List/point reads can legitimately observe different generations; +// this paired bulk read cannot. The test never touches persistence or drivers. +func TestRegistryProjectionSnapshotConcurrentMappingEdits(t *testing.T) { + rr := NewRegistry(nil) + rr.IngestSnapshot(benchmarkVMState(1)) + id := rr.List()[0].ID + set := func(version int) { + rr.mu.Lock() + defer rr.mu.Unlock() + name := fmt.Sprintf("coordinate-%d", version) + rr.resources[id].Name = name + rr.bySource[SourceProxmox] = map[string]string{name: id} + rr.invalidateViewsLocked() + } + set(0) + var wg sync.WaitGroup + wg.Add(1) + go func() { + defer wg.Done() + for i := 0; i < 200; i++ { + set(i % 2) + } + }() + defer wg.Wait() + for i := 0; i < 200; i++ { + rows, targets := rr.ListWithMetricsTargets() + if len(rows) != 1 || rows[0].Name != targets[id].ResourceID { + t.Fatalf("mixed row/target generation: row=%q target=%q", rows[0].Name, targets[id].ResourceID) + } + } +} + +var projectionResourceSink []Resource +var projectionTargetSink map[string]MetricsTarget + +// Include the entire owned List and every history coordinate. These are not +// point-lookup timings, nor a replacement for the old clone-allocation envelope. +func BenchmarkRegistryProjectionSnapshot(b *testing.B) { + for _, count := range []int{0, 1, 1000} { + for _, mode := range []string{"cold", "list-clean", "views-clean", "dirty"} { + for _, bulk := range []bool{false, true} { + name := "point-targets" + if bulk { + name = "bulk-targets" + } + b.Run(fmt.Sprintf("%d/%s/%s", count, mode, name), func(b *testing.B) { + state := benchmarkVMState(count) + rr := NewRegistry(nil) + rr.IngestSnapshot(state) + if mode != "cold" { + rr.List() + } + if mode == "views-clean" { + rr.VMs() + } + b.ReportAllocs() + b.ResetTimer() + for i := 0; i < b.N; i++ { + if mode == "cold" { + b.StopTimer() + rr = NewRegistry(nil) + rr.IngestSnapshot(state) + b.StartTimer() + } + if mode == "dirty" { + rr.mu.Lock() + for _, r := range rr.resources { + r.Tags = []string{fmt.Sprintf("version-%d", i)} + break + } + rr.invalidateViewsLocked() + rr.mu.Unlock() + } + if bulk { + projectionResourceSink, projectionTargetSink = rr.ListWithMetricsTargets() + } else { + projectionResourceSink = rr.List() + projectionTargetSink = make(map[string]MetricsTarget, len(projectionResourceSink)) + for _, row := range projectionResourceSink { + if target := rr.MetricsTarget(row.ID); target != nil { + projectionTargetSink[row.ID] = *target + } + } + } + } + }) + } + } + } +} diff --git a/internal/unifiedresources/registry_test.go b/internal/unifiedresources/registry_test.go index e7490bef07..8eacc8f37b 100644 --- a/internal/unifiedresources/registry_test.go +++ b/internal/unifiedresources/registry_test.go @@ -6968,3 +6968,68 @@ func TestEvaluateResourceHealthKeepsStaleTelemetryBehindAWinningAlert(t *testing t.Fatalf("expected only the alert reason for a fresh source, got %+v", health.Reasons) } } + +// A List-only broadcast must capture history coordinates without constructing +// every typed view, and without trusting freshness timestamps as mutation keys. +func TestRegistryProjectionSnapshotDetachesAndTracksMappings(t *testing.T) { + rr := NewRegistry(nil) + rr.IngestSnapshot(benchmarkVMState(3)) + resources, targets := rr.ListWithMetricsTargets() + if !reflect.DeepEqual(resources, rr.List()) { + t.Fatal("bulk read changed resource content") + } + if !rr.viewsDirty || rr.cachedVMs != nil { + t.Fatal("projection forced unrelated typed views to materialize") + } + if len(targets) != len(resources) { + t.Fatalf("targets=%d resources=%d", len(targets), len(resources)) + } + for _, r := range resources { + if want := rr.MetricsTarget(r.ID); want == nil || *want != targets[r.ID] { + t.Fatalf("target for %s differs from point resolver", r.ID) + } + } + id := resources[0].ID + observedAt := resources[0].LastSeen + rr.mu.Lock() + for key, mapped := range rr.bySource[SourceProxmox] { + if mapped == id { + delete(rr.bySource[SourceProxmox], key) + } + } + rr.bySource[SourceProxmox]["new-history-coordinate"] = id + rr.resources[id].Tags = []string{"customer-data"} + rr.invalidateViewsLocked() + rr.mu.Unlock() + changed, updatedTargets := rr.ListWithMetricsTargets() + if updatedTargets[id].ResourceID != "new-history-coordinate" { + t.Fatal("same-time mapping change left an old target") + } + if targets[id].ResourceID == "new-history-coordinate" { + t.Fatal("later mapping edit mutated a prior target snapshot") + } + for _, r := range changed { + if r.ID == id { + if !r.LastSeen.Equal(observedAt) || r.Policy.Sensitivity != ResourceSensitivityRestricted { + t.Fatal("same-time metadata edit was stale") + } + r.Canonical.Aliases[0] = "caller-change" + r.Tags[0] = "caller-change" + r.Policy.Routing.Redact = append(r.Policy.Routing.Redact, ResourceRedactionPath) + } + } + again, againTargets := rr.ListWithMetricsTargets() + if !reflect.DeepEqual(againTargets, updatedTargets) { + t.Fatal("detached map mutation escaped") + } + for _, r := range again { + if r.ID == id && (r.Tags[0] != "customer-data" || r.Canonical.Aliases[0] == "caller-change") { + t.Fatal("caller changed live registry") + } + } + empty := NewRegistry(nil) + rows, emptyTargets := empty.ListWithMetricsTargets() + if len(rows) != 0 || len(emptyTargets) != 0 { + t.Fatal("empty projection invented resources or targets") + } +} From 32f09dc13d6b3afaa802da9dcad7bba2cd233b1f Mon Sep 17 00:00:00 2001 From: "pulse-triage[bot]" <249995291+pulse-triage[bot]@users.noreply.github.com> Date: Mon, 5 Oct 2026 11:30:00 +0100 Subject: [PATCH 06/12] Keep complete projection controls faithful to the parent path The independent full-state oracle retains the parent unified view watermark. The paired benchmark preprocesses active health alerts once per pass, like the exact parent, rather than adding per-row work to the baseline. Production projection and source fields are unchanged. Change-source: pulse-maintainer --- .../monitoring/broadcast_metrics_snapshot_test.go | 11 ++++++++++- 1 file changed, 10 insertions(+), 1 deletion(-) diff --git a/internal/monitoring/broadcast_metrics_snapshot_test.go b/internal/monitoring/broadcast_metrics_snapshot_test.go index 4e61cc2f3d..f774d14b72 100644 --- a/internal/monitoring/broadcast_metrics_snapshot_test.go +++ b/internal/monitoring/broadcast_metrics_snapshot_test.go @@ -84,6 +84,11 @@ func TestBroadcastMetricsSnapshotConnectedContent(t *testing.T) { t.Fatal(err) } want := legacyFrontendProjectionForTest(m, graph.State, adapter.GetAll(), adapter, time.Now().UTC()) + // The parent's unified view fills a missing store watermark from the + // listed resources. Preserve that full-state header in the oracle too. + if freshness := latestUnifiedResourceLastSeen(adapter.GetAll()); !freshness.IsZero() { + want.LastUpdate = freshness.UnixMilli() + } got := m.buildBroadcastFrontendStateFromSnapshot(graph.State) wantJSON, err := json.Marshal(want) if err != nil { @@ -189,9 +194,10 @@ var completeProjectionJSONSink []byte // metadata/health decoration and snapshot projection are part of both costs. func parentFrontendProjectionForTest(m *Monitor, snapshot models.StateSnapshot, resources []unifiedresources.Resource, resolver MetricsTargetResourceStore, now time.Time) models.StateFrontend { rows := unifiedresources.CoalescePresentationHostResources(resources) + healthAlerts := resourceHealthAlerts(snapshot.ActiveAlerts) for i := range rows { m.applyPersistedMetadataToUnifiedResource(&rows[i]) - health := unifiedresources.EvaluateResourceHealth(rows[i], resourceHealthAlerts(snapshot.ActiveAlerts), now) + health := unifiedresources.EvaluateResourceHealth(rows[i], healthAlerts, now) rows[i].Health = &health } projected, catalogs := convertPresentationResourcesForBroadcast(attachBroadcastMetricsTargets(rows, resolver)) @@ -201,6 +207,9 @@ func parentFrontendProjectionForTest(m *Monitor, snapshot models.StateSnapshot, out.PolicyCatalog = catalogs.policies out.AISafeSummaryCatalog = catalogs.aiSafeSummaries out.ConnectedInfrastructure = buildConnectedInfrastructure(rows, snapshot) + if freshness := latestUnifiedResourceLastSeen(resources); !freshness.IsZero() { + out.LastUpdate = freshness.UnixMilli() + } return out } From 8b12e0ee1b678dff4a251e73b6f54f1776c1dd34 Mon Sep 17 00:00:00 2001 From: "pulse-triage[bot]" <249995291+pulse-triage[bot]@users.noreply.github.com> Date: Mon, 5 Oct 2026 11:33:31 +0100 Subject: [PATCH 07/12] Pin live row health and clear across captured resource reads The full current-state regression now requires a critical alert to reach the same freshly projected row. Continuity controls compare a healthy VM, critical transition and clear instead of using an already-offline host as a vacuous health control. Change-source: pulse-maintainer --- .../monitoring/broadcast_metrics_snapshot_test.go | 13 ++++++++++--- internal/monitoring/monitor_host_agents_test.go | 4 ++++ 2 files changed, 14 insertions(+), 3 deletions(-) diff --git a/internal/monitoring/broadcast_metrics_snapshot_test.go b/internal/monitoring/broadcast_metrics_snapshot_test.go index f774d14b72..a6eef3d7af 100644 --- a/internal/monitoring/broadcast_metrics_snapshot_test.go +++ b/internal/monitoring/broadcast_metrics_snapshot_test.go @@ -177,9 +177,16 @@ func TestBroadcastMetricsSnapshotContinuityAndLiveOverlays(t *testing.T) { } } } - critical := unifiedresources.EvaluateResourceHealth(host, []unifiedresources.ResourceHealthAlert{{ResourceID: host.ID, Level: "critical", Type: "cpu"}}, now) - if critical.Verdict != unifiedresources.HealthCritical { - t.Fatal("live alert was hidden by captured data") + for _, r := range view.resources { + if r.Type != unifiedresources.ResourceTypeVM { + continue + } + baseline := unifiedresources.EvaluateResourceHealth(r, nil, now) + critical := unifiedresources.EvaluateResourceHealth(r, []unifiedresources.ResourceHealthAlert{{ResourceID: r.ID, Level: "critical", Type: "cpu"}}, now) + cleared := unifiedresources.EvaluateResourceHealth(r, nil, now) + if baseline.Verdict != unifiedresources.HealthOK || critical.Verdict != unifiedresources.HealthCritical || !reflect.DeepEqual(cleared, baseline) { + t.Fatal("live alert/clear was hidden by captured data") + } } if len(adapter.GetAll()) != 2 { t.Fatal("continuity overlay wrote back to the live registry") diff --git a/internal/monitoring/monitor_host_agents_test.go b/internal/monitoring/monitor_host_agents_test.go index 695d475395..b7d3708922 100644 --- a/internal/monitoring/monitor_host_agents_test.go +++ b/internal/monitoring/monitor_host_agents_test.go @@ -6763,6 +6763,10 @@ func TestBroadcastProjectionListsRegistryOnceAndKeepsLiveChanges(t *testing.T) { if row == nil || row.CPU.Current != 77 || len(row.Tags) != 1 || row.Tags[0] != "new-tag" { t.Fatalf("mutable row was stale: %#v", row) } + var health unifiedresources.ResourceHealth + if err := json.Unmarshal(row.Health, &health); err != nil || health.Verdict != unifiedresources.HealthCritical { + t.Fatalf("live critical alert did not reach the freshly projected row: %s (%v)", row.Health, err) + } if first.Resources[0].CPU.Current != oldCPU { t.Fatal("later projection mutated an accepted baseline") } From 02e6a90374416e0ffd05a363fd1178306d770506 Mon Sep 17 00:00:00 2001 From: "pulse-triage[bot]" <249995291+pulse-triage[bot]@users.noreply.github.com> Date: Mon, 5 Oct 2026 11:41:55 +0100 Subject: [PATCH 08/12] Make completed-backup evidence accessible in Workloads Use native read-only disclosure buttons for both backup entry points, with the shared Dialog preserving named completion evidence, focus and guest row action ownership. Keep existing freshness and uncertainty semantics, collection and backup operations unchanged. Include connected regression tests, the subsystem contract and inspected desktop/phone browser verification. Change-source: pulse-maintainer --- .../subsystems/performance-and-scalability.md | 23 ++ .../browser-tests/backup-details-access.cjs | 359 ++++++++++++++++++ .../browser-tests/backup-details-access.html | 12 + .../browser-tests/backup-details-access.tsx | 110 ++++++ .../5dc87cd447b452f0.json | 37 ++ .../src/components/Workloads/GuestRow.tsx | 2 + .../components/Workloads/GuestRowCells.tsx | 171 ++++++--- .../__tests__/BackupStatusAccess.test.tsx | 106 ++++++ .../__tests__/BackupStatusCell.test.tsx | 2 +- .../Workloads/__tests__/GuestRow.test.tsx | 25 ++ 10 files changed, 787 insertions(+), 60 deletions(-) create mode 100644 frontend-modern/browser-tests/backup-details-access.cjs create mode 100644 frontend-modern/browser-tests/backup-details-access.html create mode 100644 frontend-modern/browser-tests/backup-details-access.tsx create mode 100644 frontend-modern/browser-verification/5dc87cd447b452f0.json create mode 100644 frontend-modern/src/components/Workloads/__tests__/BackupStatusAccess.test.tsx diff --git a/docs/release-control/v6/internal/subsystems/performance-and-scalability.md b/docs/release-control/v6/internal/subsystems/performance-and-scalability.md index 5eaa84950b..8e1892536b 100644 --- a/docs/release-control/v6/internal/subsystems/performance-and-scalability.md +++ b/docs/release-control/v6/internal/subsystems/performance-and-scalability.md @@ -3737,3 +3737,26 @@ only explicit matching refreshes; the full `GuestDrawer.test.tsx` pins the unchanged cap and no live-recovery read. `guestHistorySegments.test.ts` pins complete point preservation at the existing request bound. Neither claims a timing benchmark, fleet CPU attribution or native incident relief. + + +### Accessible Workloads backup evidence + +Both the Backup column badge and its hidden-column name indicator are native +read-only disclosure buttons. Keyboard activation and an ordinary first touch +open the same named-guest completion evidence that fine-pointer hover shows. +Exact time and age use readable theme text rather than colour alone. The shared +Dialog owns focus, Escape/backdrop dismissal and background isolation, and +returns focus to the originating disclosure. Nested evidence activation must +not trigger the enclosing guest row. Touch targets are at least 44px; compact +desktop shields and the existing status colours remain. + +The evidence content is reactive to the same guest identity: running is not +completion, malformed/future times are unknown, and only the existing absence +sentinels mean no backup recorded. A removed row unmounts its disclosure and +restores dialog isolation. Opening details adds no API/history read, collector, +permission, backup/guest operation or polling timer. Dialog machinery and +content are mounted only when opened; existing freshness thresholds and source +projection are unchanged. `GuestRow.test.tsx` covers both nested row paths and +`BackupStatusAccess.test.tsx` covers semantics, focus and observation changes. +Browser proof exercises production rows with synthetic data, not native backup +or guest-recovery acceptance. diff --git a/frontend-modern/browser-tests/backup-details-access.cjs b/frontend-modern/browser-tests/backup-details-access.cjs new file mode 100644 index 0000000000..0fef9ecc26 --- /dev/null +++ b/frontend-modern/browser-tests/backup-details-access.cjs @@ -0,0 +1,359 @@ +const assert = require('node:assert/strict'); +const fs = require('node:fs'); +const path = require('node:path'); +const crypto = require('node:crypto'); +const { chromium, webkit } = require('playwright'); +const root = '/workspace'; +const frontend = path.join(root, 'frontend-modern'); +const output = path.join(root, 'tmp/backup-details-access/browser'); +const origin = 'http://127.0.0.1:5298'; +const runtime = [ + 'frontend-modern/src/components/Workloads/GuestRowCells.tsx', + 'frontend-modern/src/components/Workloads/GuestRow.tsx', +]; +const hash = (p) => crypto.createHash('sha256').update(fs.readFileSync(p)).digest('hex'); +(async () => { + fs.mkdirSync(output, { recursive: true }); + const version = require('playwright/package.json').version; + assert.equal( + version, + JSON.parse(fs.readFileSync(path.join(root, 'tests/integration/package-lock.json'))).packages[ + 'node_modules/@playwright/test' + ].version, + ); + const hashes = Object.fromEntries( + runtime.map((source) => [source, hash(path.join(root, source))]), + ); + process.chdir(frontend); + const { createServer } = await import( + path.join(frontend, 'node_modules/vite/dist/node/index.js') + ); + const server = await createServer({ + cacheDir: path.join(output, 'vite-cache'), + server: { host: '127.0.0.1', port: 5298, strictPort: true }, + }); + const results = []; + const captures = []; + const phases = []; + const started = Date.now(); + const mark = (stage) => { + const fact = { + stage, + elapsedMs: Date.now() - started, + memory: process.memoryUsage(), + resources: process.resourceUsage(), + }; + phases.push(fact); + console.log(JSON.stringify({ browserPhase: fact })); + }; + let browser; + let phase = 'server'; + try { + mark('listen-start'); + await server.listen(); + mark('listen-complete'); + // Supported Vite transform warmup, not a production build or app execution. + // Keep page navigation and the tool deadline unchanged; collect phases. + await server.warmupRequest('/browser-tests/backup-details-access.tsx'); + await server.warmupRequest('/src/index.css'); + mark('entry-css-warmup-complete'); + for (const [name, engine, phone, dark] of [ + ['desktop-light', chromium, false, false], + ['desktop-dark', chromium, false, true], + ['phone-light', webkit, true, false], + ['phone-dark', webkit, true, true], + ]) { + phase = name; + browser = await engine.launch( + engine === chromium + ? { headless: true, channel: 'chromium', args: ['--no-sandbox'] } + : { headless: true }, + ); + const page = await browser.newPage({ + viewport: { width: phone ? 390 : 1365, height: phone ? 844 : 900 }, + isMobile: phone, + hasTouch: phone, + locale: 'en-GB', + timezoneId: 'UTC', + }); + page.setDefaultTimeout(12000); + page.setDefaultNavigationTimeout(45000); + await page.clock.setFixedTime(new Date('2026-10-05T08:00:00Z')); + const errors = []; + const operations = []; + const reads = []; + const checks = []; + page.on('pageerror', (e) => errors.push(e.message)); + await page.route('**/*', (route) => { + const request = route.request(); + const url = new URL(request.url()); + assert.equal(url.origin, origin, 'no off-origin requests'); + if (!url.pathname.startsWith('/api/')) return route.continue(); + if (request.method() !== 'GET') operations.push(`${request.method()} ${url.pathname}`); + reads.push(url.pathname); + return route.fulfill({ + json: { success: true, data: [], alerts: [], config: null, anomalies: [] }, + }); + }); + if (dark) + await page.addInitScript(() => + document.addEventListener('DOMContentLoaded', () => + document.documentElement.classList.add('dark'), + ), + ); + mark(`${name}:navigate-start`); + await page.goto(`${origin}/browser-tests/backup-details-access.html`, { + waitUntil: 'domcontentloaded', + }); + mark(`${name}:domcontentloaded`); + const row = page.getByRole('region', { name: 'backup-vm row' }); + const badge = row.getByRole('button', { name: /^Backup status:/ }); + await badge.waitFor(); + const identity = await page.evaluate(() => window.__backupDetails.identity()); + const rowActionCount = async () => + Number(await page.getByRole('status', { name: 'Row action count' }).innerText()); + const activate = async (el) => (phone ? el.tap() : el.click()); + const capture = async (suffix) => { + const file = path.join(output, `${name}-${suffix}.png`); + await page.screenshot({ path: file, fullPage: true }); + captures.push({ + path: path.relative(root, file), + bytes: fs.statSync(file).size, + sha256: hash(file), + }); + }; + const dialog = page.getByRole('dialog', { name: /^Backup status details for / }); + const close = page.getByRole('button', { name: 'Close backup details' }); + const checkOpen = async (trigger) => { + await dialog.waitFor({ state: 'visible' }); + assert.equal(await trigger.getAttribute('aria-expanded'), 'true'); + assert.equal(await rowActionCount(), 0, 'backup action must not open the guest row'); + await close.waitFor(); + await page.waitForFunction( + () => document.activeElement?.getAttribute('aria-label') === 'Close backup details', + ); + // DOM visibility can precede the CSS entrance animation. Verify and + // capture the final user-visible panel, not its initial off-screen frame. + await dialog.evaluate(async (el) => { + await Promise.all( + el + .getAnimations({ subtree: true }) + .filter((a) => Number.isFinite(a.effect?.getTiming().iterations)) + .map((a) => a.finished.catch(() => {})), + ); + }); + const rect = await dialog.boundingBox(); + assert( + rect.x >= 0 && rect.x + rect.width <= (phone ? 390 : 1365), + 'dialog stays within viewport', + ); + assert( + rect.y >= 0 && rect.y + rect.height <= (phone ? 844 : 900), + 'settled panel stays vertically in view', + ); + assert( + await page.locator('main').evaluate((el) => el.closest('body > *').hasAttribute('inert')), + 'background isolated', + ); + }; + const checkClosed = async (trigger) => { + await dialog.waitFor({ state: 'hidden' }); + assert.equal(await trigger.getAttribute('aria-expanded'), 'false'); + assert( + await trigger.evaluate((el) => document.activeElement === el), + 'focus returns to source evidence', + ); + assert.equal( + await page.locator('main').evaluate((el) => el.closest('body > *').hasAttribute('inert')), + false, + ); + }; + assert.equal(await badge.getAttribute('aria-haspopup'), 'dialog'); + assert.equal((await badge.innerText()).trim(), '', 'fresh shield remains compact'); + if (!phone) { + await badge.hover(); + const tooltip = page.locator('[data-tooltip-portal="true"]'); + await tooltip.waitFor({ state: 'visible' }); + assert((await tooltip.innerText()).includes('Last completed backup')); + await page.getByRole('heading').hover(); + // Genuine sequential keyboard traversal, not synthetic click/focus. + await page.keyboard.press('Tab'); // column control + await page.keyboard.press('Tab'); // guest disclosure + await page.keyboard.press('Tab'); // backup disclosure + assert(await badge.evaluate((el) => document.activeElement === el)); + await page.keyboard.press('Enter'); + checks.push('desktop hover preserved and sequential Tab/Enter opens actual badge'); + } else { + const rect = await badge.boundingBox(); + assert(rect.width >= 44 && rect.height >= 44, 'touch disclosure is at least 44px'); + await activate(badge); + assert.equal( + await page.locator('[data-tooltip-portal="true"]').count(), + 0, + 'no synthetic hover overlay', + ); + checks.push('ordinary first tap opens 44px target, without hover UI'); + } + await checkOpen(badge); + assert((await dialog.innerText()).includes('backup-vm')); + assert((await dialog.innerText()).includes('Mon, 5 Oct 2026')); + assert((await dialog.innerText()).includes('06:00:00')); + assert((await dialog.innerText()).includes('2 hours ago')); + await capture('fresh-completed'); + await page.keyboard.press('Tab'); + assert( + await close.evaluate((el) => document.activeElement === el), + 'one-action dialog retains focus', + ); + await page.evaluate(() => + window.__backupDetails.update({ timestamp: '2026-10-05T06:00:00Z', running: true }), + ); + assert((await dialog.innerText()).includes('Backup running now')); + assert((await dialog.innerText()).includes('Last completed backup')); + assert.equal(await page.evaluate(() => window.__backupDetails.identity()), identity); + checks.push('same identity updates running while preserving prior completed observation'); + await capture('running-completed'); + await page.evaluate(() => + window.__backupDetails.update({ timestamp: '2026-10-05T08:01:00Z', running: true }), + ); + await dialog.getByText(/timestamp is in the future/).waitFor(); + assert(!(await dialog.innerText()).includes('Last completed backup')); + assert(!(await dialog.innerText()).includes('No backup has ever')); + await capture('future-running'); + await page.keyboard.press('Escape'); + await checkClosed(badge); + if (!phone) { + await page.keyboard.press('Space'); + await checkOpen(badge); + assert((await dialog.innerText()).includes('timestamp is in the future')); + await page.keyboard.press('Escape'); + await checkClosed(badge); + checks.push('native Space reopens current evidence and Escape restores focus'); + } + await page.getByRole('button', { name: 'Hide Backup column' }).click(); + const indicator = row.getByRole('button', { name: /^Backup running now/ }); + await activate(indicator); + await checkOpen(indicator); + assert((await dialog.innerText()).includes('timestamp is in the future')); + checks.push('hidden-column indicator exposes exactly the same uncertainty'); + await page.evaluate(() => + window.__backupDetails.update({ timestamp: 'bad-time', running: false }), + ); + await dialog.getByText(/invalid timestamp/).waitFor(); + await page.evaluate(() => window.__backupDetails.update({ timestamp: '', running: false })); + await dialog + .getByText('No backup has ever been recorded for this guest.', { exact: true }) + .waitFor(); + assert(!(await dialog.innerText()).includes('Invalid Date')); + assert(!(await dialog.innerText()).includes('NaN')); + checks.push('malformed then absence retain distinct truthful states'); + await activate(close); + await checkClosed(row.getByRole('button', { name: 'No backup found', exact: true })); + const ct = page + .getByRole('region', { name: 'backup-ct row' }) + .getByRole('button', { name: 'No backup found', exact: true }); + await activate(ct); + await checkOpen(ct); + assert((await dialog.innerText()).includes('backup-ct')); + assert( + !(await dialog.innerText()).includes('2 hours ago'), + 'adjacent guest never inherits another observation', + ); + await capture('ct-missing'); + await page.locator('[data-dialog-backdrop]').click({ position: { x: 3, y: 3 } }); + await checkClosed(ct); + await page.evaluate(() => + window.__backupDetails.update({ timestamp: '2026-10-05T06:00:00Z', running: false }), + ); + const recovered = row.getByRole('button', { name: /^Last backup:/ }); + await activate(recovered); + await checkOpen(recovered); + assert((await dialog.innerText()).includes('2 hours ago')); + await page.evaluate(() => window.__backupDetails.remove()); + await dialog.waitFor({ state: 'hidden' }); + assert.equal(await page.locator('[data-dialog-layer]').count(), 0); + assert.equal( + await page.locator('main').evaluate((el) => el.closest('body > *').hasAttribute('inert')), + false, + ); + checks.push( + 'recovery and actual row removal release dialog/inertness without extra operations', + ); + assert.equal(await rowActionCount(), 0); + await activate( + page.getByRole('region', { name: 'backup-ct row' }).getByText('backup-ct', { exact: true }), + ); + assert.equal(await rowActionCount(), 1, 'ordinary row action still works'); + assert.deepEqual(errors, []); + assert.deepEqual(operations, []); + assert( + !reads.some((p) => /backup|metrics|guest|resources/.test(p)), + 'opening evidence adds no collector/history read', + ); + results.push({ + name, + browser: browser.version(), + checks, + apiReads: reads, + operations, + errors, + finalRowActions: await rowActionCount(), + }); + mark(`${name}:checks-complete`); + await page.close(); + await browser.close(); + browser = undefined; + } + fs.writeFileSync( + path.join(output, 'result.json'), + JSON.stringify( + { + result: 'passed', + playwrightVersion: version, + runtime: hashes, + phases, + results, + captures, + }, + null, + 2, + ) + '\n', + ); + console.log( + JSON.stringify({ + result: 'passed', + cases: results.length, + checks: results.reduce((n, r) => n + r.checks.length, 0), + captures: captures.length, + hashes, + playwrightVersion: version, + }), + ); + } catch (e) { + fs.writeFileSync( + path.join(output, 'failure.json'), + JSON.stringify( + { + result: 'failed', + phase, + error: String(e), + runtime: hashes, + phases, + playwrightVersion: version, + captures, + results, + }, + null, + 2, + ) + '\n', + ); + throw e; + } finally { + if (browser) await browser.close(); + await server.close(); + mark('browser-server-closed'); + } +})().catch((e) => { + console.error(e); + process.exitCode = 1; +}); diff --git a/frontend-modern/browser-tests/backup-details-access.html b/frontend-modern/browser-tests/backup-details-access.html new file mode 100644 index 0000000000..488f6afbdc --- /dev/null +++ b/frontend-modern/browser-tests/backup-details-access.html @@ -0,0 +1,12 @@ + + +
+ + ++ Synthetic guest observations, no backup commands or collector reads. +
+ ++ {actions()} +
+{props.guestName}
+Backup running now…
+{info().ageFormatted}
++ {props.backupRunning + ? 'No completed backup yet - the first backup is running now.' + : 'No backup has ever been recorded for this guest.'} +
+