Merge reviewed Core guest metadata writer and shutdown safety

Change-source: pulse-maintainer

docs/release-control/v6/internal/subsystems/agent-lifecycle.md

@@ -7520,13 +7520,26 @@ 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. 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.
 
-`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

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

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

@@ -4581,27 +4581,45 @@ 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. 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
+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
 

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)
 	}
 

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)
+	}
+}

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,202 @@ 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
+	closeMu           sync.Mutex
+	closeAttempt      *guestMetadataCloseAttempt
 }
 
-// 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) {
+type guestMetadataCloseAttempt struct {
+	done chan struct{}
+	err  error
+}
+
+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 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.Lock()
+			if retry && s.lastWriteErr != nil && s.writerDone == nil {
+				s.queueSaveLocked()
+			}
+			done := s.writerDone
+			s.mu.Unlock()
+			if done != nil {
+				<-done
+			}
+			s.mu.RLock()
+			attempt.err = s.lastWriteErr
+			s.mu.RUnlock()
+			close(attempt.done)
+		}()
+	}
+	s.closeMu.Unlock()
+	timer := time.NewTimer(timeout)
+	defer timer.Stop()
+	select {
+	case <-attempt.done:
+		return attempt.err
+	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
@@ -113,6 +268,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 +292,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 +300,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 +342,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 +452,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 +474,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 +490,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 +528,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 {

internal/config/guest_metadata_writer_test.go

@@ -0,0 +1,396 @@
+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")
+	}
+}
+
+// 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")
+	}
+}

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)
+	}
+}

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)
+	}
+}

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

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")
 	}
 }
 

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) {

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
 	}
 }