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

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

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,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.
+`persistGuestIdentity` calls `GuestMetadataStore.RememberIdentity`, which merges
-It calls `GuestMetadataStore.SetAsync`, which tracks the write on a WaitGroup
+only the last-known name/type under the store lock, preserves operator URLs,
-so `GuestMetadataStore.WaitForPendingWrites` can drain it. `Monitor.Stop` drains
+descriptions, tags and notes, and never downgrades OCI. Admission is visible
-before closing the metrics store, under a bounded timeout matching
+before return. `SetAsync` deep-copies its full replacement at admission as well;
-`tenantMonitorShutdownTimeout` so a wedged store cannot hold up tenant teardown.
+caller mutation and goroutine scheduling cannot reverse accepted update order.
 
-Untracked writes were observable, not theoretical: a queued write could land
+One lazy store-owned writer snapshots the complete metadata map. Filesystem I/O
-after the monitor stopped and after a tenant directory was being removed,
+does not hold the memory lock, so changes admitted during that I/O are coalesced
-leaving a stray `guest_metadata.json.tmp` from the interrupted atomic write.
+into the next complete snapshot. A separate write mutex serializes it with
-That is what made `TestHostedTenantAgentInstallTokenCannotReportToOtherTenant`
+synchronous Set/Delete/ReplaceAll/UpdateAll and legacy migration; an older
-fail its `t.TempDir` cleanup with "directory not empty".
+snapshot cannot resurrect a deleted or migrated entry or overwrite an operator
-`TestGuestMetadataStore_WaitForPendingWritesDrainsQueuedWrites` and
+edit. The existing tenant-local JSON shape, private file permissions and atomic
-`TestGuestMetadataStore_DataDirIsRemovableAfterDrain` pin the drain and fail if
+temporary-file/rename path stay unchanged. A failed write is reported and not
-`SetAsync` stops tracking its goroutine.
+spin-retried; a later ordinary identity poll can retry the unpersisted state.
 
-Known and deliberately unchanged: each changed guest still triggers a full-file
+`WaitForPendingWrites` reports quiescence, not durable success. `Close` seals
-save, so one poll cycle over N changed guests performs N marshals and N atomic
+mutation admission and observes the background writer and any admitted
-writes that serialize on the store mutex. Coalescing them is a behavioural
+synchronous write within the unchanged caller budget. Timeout or final
-change beyond the shutdown defect.
+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
 

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,176 @@ type GuestMetadataStore struct {
 	dataPath string
 	fs       FileSystem
 
-	// inflight tracks background Set calls started by SetAsync so shutdown can
+	// writeMu serializes disk snapshots with synchronous mutations. Background
-	// wait for them. Callers used to spawn their own detached goroutines, which
+	// I/O never holds mu, so a burst can accumulate in one follow-up snapshot.
-	// meant a write could land after the monitor stopped and after the data
+	// Always acquire writeMu before mu when both locks are needed.
-	// directory was being torn down, leaving a stray guest_metadata.json.tmp.
+	writeMu           sync.Mutex
-	inflight sync.WaitGroup
+	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
+var ErrGuestMetadataStoreClosed = errors.New("guest metadata store is closed")
-// write owned by the store so WaitForPendingWrites can drain it on shutdown.
+
-func (s *GuestMetadataStore) SetAsync(guestID string, meta *GuestMetadata) {
+// 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)
+	s.writerDone = make(chan struct{})
-	go func() {
+	go s.writePendingSnapshots()
-		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")
-		}
-	}()
 }
 
-// WaitForPendingWrites blocks until background writes finish or the timeout
+func (s *GuestMetadataStore) writePendingSnapshots() {
-// elapses. It reports whether the store drained; a false result means a write
+	for {
-// may still be in flight and the data directory is not safe to remove.
+		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{})
+	s.mu.RLock()
-	go func() {
+	done := s.writerDone
-		s.inflight.Wait()
+	s.mu.RUnlock()
-		close(done)
+	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),
+		metadata:  make(map[string]*GuestMetadata),
-		dataPath: dataPath,
+		dataPath:  dataPath,
-		fs:       fs,
+		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")
+	s.revision++
-
-	log.Debug().Str("path", filePath).Msg("Saving guest metadata to disk")
-
 	data, err := json.Marshal(s.metadata)
-	if err != nil {
+	if err == nil {
-		return fmt.Errorf("failed to marshal metadata: %w", err)
+		err = persistMetadata(s.fs, s.dataPath, "guest_metadata.json", data)
 	}
-
+	s.attemptedRevision = s.revision
-	if err := persistMetadata(s.fs, s.dataPath, "guest_metadata.json", data); err != nil {
+	s.lastWriteErr = err
-		return err
+	return err
-	}
-
-	log.Debug().Str("path", filePath).Int("entries", len(s.metadata)).Msg("Guest metadata saved successfully")
-
-	return nil
 }
 
 // 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 {

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

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 {
+	drainErr := m.guestMetadataStore.Close(guestMetadataDrainTimeout)
-		m.guestMetadataStore.WaitForPendingWrites(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 err := metadataStore.RememberIdentity(guestKey, name, guestType); err != nil {
-	if existing == nil {
+		log.Error().Err(err).Str("guestKey", guestKey).Msg("failed to admit guest identity update")
-		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)
 	}
 }
 

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
+	if !store.WaitForPendingWrites(time.Second) {
-	time.Sleep(50 * time.Millisecond)
+		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()
+	if mtm.globalCancel != nil {
-
+		mtm.globalCancel()
-	monitors := make([]*Monitor, 0, len(mtm.monitors))
+	}
-	cancels := make([]context.CancelFunc, 0, len(mtm.tenantCancel))
+	orgIDs := make([]string, 0, len(mtm.monitors))
-	doneSignals := make([]chan struct{}, 0, len(mtm.tenantDone))
+	for orgID := range mtm.monitors {
-	for orgID, monitor := range mtm.monitors {
+		orgIDs = append(orgIDs, orgID)
-		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)
 	}
-	mtm.tenantCancel = make(map[string]context.CancelFunc)
-	mtm.tenantDone = make(map[string]chan struct{})
-	mtm.monitors = make(map[string]*Monitor)
 	mtm.mu.Unlock()
-
+	for _, orgID := range orgIDs {
-	for _, cancel := range cancels {
+		if err := mtm.removeTenant(orgID, false); err != nil {
-		cancel()
+			log.Error().Err(err).Str("org_id", orgID).Msg("tenant shutdown incomplete; retaining runtime ownership and data")
-	}
+		}
-	for _, done := range doneSignals {
-		waitForTenantMonitorShutdown("", done)
-	}
-	for _, monitor := range monitors {
-		monitor.Stop()
 	}
 }
 
 // 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.
+// persistence operation completes. An incomplete shutdown retains its guard.
-func (mtm *MultiTenantMonitor) BeginTenantDeletion(orgID string) {
+func (mtm *MultiTenantMonitor) BeginTenantDeletion(orgID string) error {
-	mtm.removeTenant(orgID, true)
+	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 {
-		if mtm.tenantDeleting == nil {
+		mtm.tenantDeleting = make(map[string]struct{})
-			mtm.tenantDeleting = make(map[string]struct{})
-		}
-		mtm.tenantDeleting[orgID] = 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()
-
+	defer func() {
-	if !exists {
+		mtm.mu.Lock()
-		return
+		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)
+	if !waitForTenantMonitorShutdown(orgID, done) {
-	monitor.Stop()
+		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
 	}
 }