Let guest metadata writes finish before the monitor stops

persistGuestIdentity spawned a detached goroutine per changed guest to write
guest_metadata.json, with a comment noting it avoided blocking the monitor.
Nothing tracked those goroutines, so neither Monitor.Stop nor
MultiTenantMonitor.Stop could wait for them and a queued write could land after
shutdown. In hosted mode that means a write into a tenant directory that
offboarding is already removing, and a stray guest_metadata.json.tmp left
behind when the atomic write is interrupted.

The store now owns the goroutine. SetAsync tracks the write on a WaitGroup and
WaitForPendingWrites drains it under a bounded timeout matching
tenantMonitorShutdownTimeout, so a wedged store cannot hold up tenant teardown.
Monitor.Stop drains before closing the metrics store.

This is what made TestHostedTenantAgentInstallTokenCannotReportToOtherTenant
flaky: t.TempDir cleanup raced a queued write into orgs/client-b and failed
with "directory not empty". The test itself is unchanged, because it was never
a test bug. A goroutine dump at cleanup time showed the writers still live,
created by persistGuestIdentity, blocked on the store mutex.

Verified causally rather than by observation alone: the target test fails 0/4
with the drain removed and passes 8/8 with it, against 2/3 failures on the
unmodified baseline. The regression tests fail if SetAsync stops tracking its
goroutine.

Note for a future pass, deliberately not changed here: each changed guest still
triggers a full-file save, so one poll cycle over N changed guests does N
marshals and N atomic writes that serialize on the store mutex anyway. Fixing
that means coalescing at the call site and is a behavioural change beyond this
defect.

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

@@ -5811,3 +5811,10 @@ 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.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.

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

@@ -2767,3 +2767,24 @@ 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
+
+`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.
+
+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.
+
+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.

internal/config/guest_metadata.go

@@ -7,6 +7,7 @@ import (
 	"path/filepath"
 	"strings"
 	"sync"
+	"time"
 
 	"github.com/rs/zerolog/log"
 )
@@ -29,6 +30,48 @@ type GuestMetadataStore struct {
 	metadata map[string]*GuestMetadata // keyed by guest ID
 	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
+}
+
+// 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) {
+	if s == 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")
+		}
+	}()
+}
+
+// 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) WaitForPendingWrites(timeout time.Duration) bool {
+	if s == nil {
+		return true
+	}
+	done := make(chan struct{})
+	go func() {
+		s.inflight.Wait()
+		close(done)
+	}()
+	select {
+	case <-done:
+		return true
+	case <-time.After(timeout):
+		log.Warn().Dur("timeout", timeout).Msg("timed out waiting for guest metadata writes to drain")
+		return false
+	}
 }
 
 func cloneGuestMetadata(meta *GuestMetadata) *GuestMetadata {

internal/config/guest_metadata_test.go

@@ -6,6 +6,7 @@ import (
 	"path/filepath"
 	"sync"
 	"testing"
+	"time"
 )
 
 func TestGuestMetadataStore_Get(t *testing.T) {
@@ -756,3 +757,76 @@ func TestGuestMetadataStore_GetWithLegacyMigration_NodeMigrationDoesNotCrossGues
 		t.Error("unrelated entries must remain untouched")
 	}
 }
+
+// A queued async write must be observable on disk once the store reports it has
+// drained. Before the store owned these writes, callers detached their own
+// goroutines, so a write could land after shutdown and after the data directory
+// was being removed, leaving a stray guest_metadata.json.tmp behind.
+func TestGuestMetadataStore_WaitForPendingWritesDrainsQueuedWrites(t *testing.T) {
+	dataDir := t.TempDir()
+	store := NewGuestMetadataStore(dataDir, nil)
+
+	for i := range 25 {
+		store.SetAsync("node:10"+string(rune('a'+i%25)), &GuestMetadata{
+			ID:            "guest",
+			LastKnownName: "vm-" + string(rune('a'+i%25)),
+			LastKnownType: "qemu",
+		})
+	}
+
+	if !store.WaitForPendingWrites(10 * time.Second) {
+		t.Fatal("store did not drain within the timeout")
+	}
+
+	// Nothing may still be writing: a leftover .tmp means an atomic write was
+	// in flight when the store claimed to be drained.
+	entries, err := os.ReadDir(dataDir)
+	if err != nil {
+		t.Fatalf("read data dir: %v", err)
+	}
+	for _, entry := range entries {
+		if filepath.Ext(entry.Name()) == ".tmp" {
+			t.Fatalf("temporary file %q survived the drain; a write was still in flight", entry.Name())
+		}
+	}
+	if _, err := os.Stat(filepath.Join(dataDir, "guest_metadata.json")); err != nil {
+		t.Fatalf("guest metadata was not persisted: %v", err)
+	}
+}
+
+// The data directory must be removable immediately after a drain. This is the
+// exact property whose absence made TestHostedTenantAgentInstallTokenCannotReportToOtherTenant
+// fail its t.TempDir cleanup with "directory not empty".
+func TestGuestMetadataStore_DataDirIsRemovableAfterDrain(t *testing.T) {
+	root := t.TempDir()
+	dataDir := filepath.Join(root, "orgs", "client-b")
+	if err := os.MkdirAll(dataDir, 0o755); err != nil {
+		t.Fatalf("mkdir: %v", err)
+	}
+
+	store := NewGuestMetadataStore(dataDir, nil)
+	for i := range 25 {
+		store.SetAsync("node:20"+string(rune('a'+i%25)), &GuestMetadata{
+			ID:            "guest",
+			LastKnownName: "ct-" + string(rune('a'+i%25)),
+			LastKnownType: "lxc",
+		})
+	}
+
+	if !store.WaitForPendingWrites(10 * time.Second) {
+		t.Fatal("store did not drain within the timeout")
+	}
+	if err := os.RemoveAll(filepath.Join(root, "orgs")); err != nil {
+		t.Fatalf("org directory was not removable after drain: %v", err)
+	}
+}
+
+func TestGuestMetadataStore_WaitOnNilStoreAndIdleStoreReturnTrue(t *testing.T) {
+	var nilStore *GuestMetadataStore
+	if !nilStore.WaitForPendingWrites(time.Second) {
+		t.Fatal("nil store must report drained")
+	}
+	if !NewGuestMetadataStore(t.TempDir(), nil).WaitForPendingWrites(time.Second) {
+		t.Fatal("idle store must report drained")
+	}
+}

internal/monitoring/monitor.go

@@ -6782,6 +6782,11 @@ func monitorLastSeenUnix(value time.Time) int64 {
 
 // pollStorageBackupsWithNodes polls backups using a provided nodes list to avoid duplicate GetNodes calls
 // Stop gracefully stops the monitor
+// guestMetadataDrainTimeout bounds how long Stop waits for queued guest
+// metadata writes. It matches tenantMonitorShutdownTimeout so a wedged store
+// cannot hold up tenant teardown indefinitely.
+const guestMetadataDrainTimeout = 2 * time.Second
+
 func (m *Monitor) Stop() {
 	log.Info().Msg("stopping monitor")
 
@@ -6795,6 +6800,13 @@ func (m *Monitor) Stop() {
 		m.notificationMgr.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)
+	}
+
 	// Close persistent metrics store (flushes buffered data)
 	if m.metricsStore != nil {
 		if err := m.metricsStore.Close(); err != nil {

internal/monitoring/monitor_backups.go

@@ -862,12 +862,10 @@ func persistGuestIdentity(metadataStore *config.GuestMetadataStore, guestKey, na
 	if existing.LastKnownName != name || existing.LastKnownType != guestType {
 		existing.LastKnownName = name
 		existing.LastKnownType = guestType
-		// Save asynchronously to avoid blocking the monitor
-		go func() {
-			if err := metadataStore.Set(guestKey, existing); err != nil {
-				log.Error().Err(err).Str("guestKey", guestKey).Msg("failed to persist guest identity")
-			}
-		}()
+		// 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_host_agents_test.go

@@ -5,6 +5,7 @@ import (
 	"errors"
 	"fmt"
 	"net"
+	"os"
 	"strings"
 	"testing"
 	"time"
@@ -4904,3 +4905,42 @@ func TestAgentLXCFilesystemsExpireAndRejectAmbiguousNodeLink(t *testing.T) {
 		t.Fatalf("accepted empty inventory did not prune expired entries: %+v", monitor.proxmoxLXCFilesystemsCache)
 	}
 }
+
+// A stopped monitor must not write into its data directory afterwards.
+// Tenant offboarding and any caller that removes a monitor's data directory
+// depends on Stop having quiesced disk writes; before the guest metadata store
+// tracked its own async writes, a queued write could land after Stop and leave
+// a partially-written .tmp behind.
+func TestMonitorStopQuiescesMetadataWritesBeforeReturning(t *testing.T) {
+	dataPath := t.TempDir()
+	monitor := &Monitor{
+		state:              models.NewState(),
+		guestMetadataStore: config.NewGuestMetadataStore(dataPath, nil),
+	}
+
+	for i := range 40 {
+		monitor.guestMetadataStore.SetAsync(fmt.Sprintf("pve1:node1:%d", 100+i), &config.GuestMetadata{
+			ID:            fmt.Sprintf("pve1:node1:%d", 100+i),
+			LastKnownName: fmt.Sprintf("vm-%d", i),
+			LastKnownType: "qemu",
+		})
+	}
+
+	monitor.Stop()
+
+	entries, err := os.ReadDir(dataPath)
+	if err != nil {
+		t.Fatalf("read data dir: %v", err)
+	}
+	for _, entry := range entries {
+		if strings.HasSuffix(entry.Name(), ".tmp") {
+			t.Fatalf("write still in flight after Stop: %q", entry.Name())
+		}
+	}
+
+	// The directory must be removable immediately, which is the property tenant
+	// teardown relies on.
+	if err := os.RemoveAll(dataPath); err != nil {
+		t.Fatalf("data directory not removable after Stop: %v", err)
+	}
+}

internal/monitoring/pve_protection_observation_test.go

@@ -2,9 +2,12 @@ package monitoring
 
 import (
 	"errors"
+	"os"
+	"path/filepath"
 	"testing"
 	"time"
 
+	"github.com/rcourtman/pulse-go-rewrite/internal/config"
 	"github.com/rcourtman/pulse-go-rewrite/internal/operationaltrust"
 	"github.com/rcourtman/pulse-go-rewrite/internal/recovery"
 )
@@ -156,3 +159,52 @@ func TestIsPVEBackupPermissionError(t *testing.T) {
 		t.Fatal("transient timeout must not be classified as a permission failure")
 	}
 }
+
+// persistGuestIdentity runs inside the protection/backup read-state walk and
+// must hand its write to the store rather than detaching its own goroutine.
+// A detached write cannot be drained by Monitor.Stop, so it can land after
+// shutdown and after the data directory has started being removed.
+func TestPersistGuestIdentityQueuesADrainableWrite(t *testing.T) {
+	dataPath := t.TempDir()
+	store := config.NewGuestMetadataStore(dataPath, nil)
+
+	persistGuestIdentity(store, "pve1:node1:100", "web-01", "qemu")
+
+	if !store.WaitForPendingWrites(10 * time.Second) {
+		t.Fatal("persistGuestIdentity write was not drainable; the call site detached its own goroutine")
+	}
+	if _, err := os.Stat(filepath.Join(dataPath, "guest_metadata.json")); err != nil {
+		t.Fatalf("guest identity was not persisted after the drain: %v", err)
+	}
+	if meta := store.Get("pve1:node1:100"); meta == nil || meta.LastKnownName != "web-01" || meta.LastKnownType != "qemu" {
+		t.Fatalf("guest identity not recorded: %#v", meta)
+	}
+}
+
+// An unchanged identity must not queue a write at all, so a steady-state poll
+// cycle does no disk work.
+func TestPersistGuestIdentitySkipsUnchangedIdentities(t *testing.T) {
+	dataPath := t.TempDir()
+	store := config.NewGuestMetadataStore(dataPath, nil)
+
+	persistGuestIdentity(store, "pve1:node1:101", "db-01", "lxc")
+	if !store.WaitForPendingWrites(10 * time.Second) {
+		t.Fatal("initial write did not drain")
+	}
+	firstWrite, err := os.Stat(filepath.Join(dataPath, "guest_metadata.json"))
+	if err != nil {
+		t.Fatalf("stat after first write: %v", err)
+	}
+
+	persistGuestIdentity(store, "pve1:node1:101", "db-01", "lxc")
+	if !store.WaitForPendingWrites(10 * time.Second) {
+		t.Fatal("second call did not drain")
+	}
+	secondWrite, err := os.Stat(filepath.Join(dataPath, "guest_metadata.json"))
+	if err != nil {
+		t.Fatalf("stat after second call: %v", err)
+	}
+	if !firstWrite.ModTime().Equal(secondWrite.ModTime()) {
+		t.Error("unchanged identity triggered a redundant write")
+	}
+}