diff --git a/docs/release-control/v6/internal/subsystems/agent-lifecycle.md b/docs/release-control/v6/internal/subsystems/agent-lifecycle.md index 469e7d2737..6743cf3006 100644 --- a/docs/release-control/v6/internal/subsystems/agent-lifecycle.md +++ b/docs/release-control/v6/internal/subsystems/agent-lifecycle.md @@ -7520,13 +7520,24 @@ coarse activity class (`list`, `export`, `verify`, `summary`); the handler runs unchanged whether or not recording succeeds, and a router without persistence serves the request rather than failing it (`TestWithAuditReadActivity_NilPersistenceIsSafe`). -### Monitor shutdown drains queued guest metadata writes +### Monitor shutdown seals guest metadata admission and retains incomplete teardown + +`Monitor.Stop` seals guest identity/metadata mutations and waits under its +unchanged two-second metadata budget. One store-owned snapshot writer coalesces +changes admitted during filesystem I/O; an older queued identity cannot overwrite +an operator edit or reverse a newer accepted identity. `Close` reports both an +incomplete drain and a final persistence error, while the public void Stop logs +failure rather than claiming a successful stop. + +Tenant removal must first observe its monitoring loop exit and then a successful +metadata close. Failure retains the monitor, cancellation/done handles and +lifecycle guard: lazy recreation cannot start a second writer in the same +directory. `BeginTenantDeletion` returns that failure to the organisation +handler, which keeps persistence and offboarding callbacks untouched. Only +successful shutdown allows directory deletion. Other tenants keep their existing +independent runtime. Store drain does not establish independent guest thaw, +workload liveness or containment in installed software. -`Monitor.Stop` waits for in-flight `GuestMetadataStore` writes before closing -the metrics store, so a tenant monitor that has been stopped is guaranteed not -to write into its data directory afterwards. Tenant offboarding and any caller -that removes a tenant directory can rely on `Stop` having quiesced disk writes, -rather than racing a detached goroutine. ### Per-tenant resource stores are released on offboarding and shutdown `ResourceHandlers.getStore` opens a SQLite handle per org and caches it for the diff --git a/docs/release-control/v6/internal/subsystems/api-contracts.md b/docs/release-control/v6/internal/subsystems/api-contracts.md index 4cc867961f..90f36f2f4b 100644 --- a/docs/release-control/v6/internal/subsystems/api-contracts.md +++ b/docs/release-control/v6/internal/subsystems/api-contracts.md @@ -20,6 +20,20 @@ ## Purpose +### Organisation deletion retains data after incomplete monitoring shutdown + +Authenticated organisation-owner deletion waits for tenant-loop exit and sealed +guest metadata persistence before removing the directory or calling offboarding. +An incomplete drain or final write failure returns HTTP 503 with the existing +error envelope and `tenant_shutdown_incomplete`; the message says the data was +retained and exposes no filesystem path or underlying error. The runtime owner +and recreation guard stay retained. Default-organisation immutability, session +and membership checks, successful HTTP 204 and other tenants' access are +unchanged. `TestOrgDeleteRetainsDataWhenGuestMetadataPersistenceFails` exercises +a real confined rename failure through the ordinary handler and metadata store; +existing successful deletion and authorization controls remain required. + + ### Optional guest-memory observation reaches existing frontend consumers Frontend Memory and ResourceMetric model the existing optional server-owned diff --git a/docs/release-control/v6/internal/subsystems/monitoring.md b/docs/release-control/v6/internal/subsystems/monitoring.md index 48bbc86ac5..40cf4ace67 100644 --- a/docs/release-control/v6/internal/subsystems/monitoring.md +++ b/docs/release-control/v6/internal/subsystems/monitoring.md @@ -4581,27 +4581,42 @@ host dataset evidence is copied onto the matching provider-owned ZFS pool; provider health, scan, device, and error fields remain authoritative. When the provider cannot return pool detail, monitoring may synthesize only a minimal `UNKNOWN` pool so valid dataset evidence is still inspectable. -### Guest metadata writes are owned by the store and drained on shutdown +### Guest identity admission coalesces owned JSON snapshots -`persistGuestIdentity` no longer detaches its own goroutine per changed guest. -It calls `GuestMetadataStore.SetAsync`, which tracks the write on a WaitGroup -so `GuestMetadataStore.WaitForPendingWrites` can drain it. `Monitor.Stop` drains -before closing the metrics store, under a bounded timeout matching -`tenantMonitorShutdownTimeout` so a wedged store cannot hold up tenant teardown. +`persistGuestIdentity` calls `GuestMetadataStore.RememberIdentity`, which merges +only the last-known name/type under the store lock, preserves operator URLs, +descriptions, tags and notes, and never downgrades OCI. Admission is visible +before return. `SetAsync` deep-copies its full replacement at admission as well; +caller mutation and goroutine scheduling cannot reverse accepted update order. -Untracked writes were observable, not theoretical: a queued write could land -after the monitor stopped and after a tenant directory was being removed, -leaving a stray `guest_metadata.json.tmp` from the interrupted atomic write. -That is what made `TestHostedTenantAgentInstallTokenCannotReportToOtherTenant` -fail its `t.TempDir` cleanup with "directory not empty". -`TestGuestMetadataStore_WaitForPendingWritesDrainsQueuedWrites` and -`TestGuestMetadataStore_DataDirIsRemovableAfterDrain` pin the drain and fail if -`SetAsync` stops tracking its goroutine. +One lazy store-owned writer snapshots the complete metadata map. Filesystem I/O +does not hold the memory lock, so changes admitted during that I/O are coalesced +into the next complete snapshot. A separate write mutex serializes it with +synchronous Set/Delete/ReplaceAll/UpdateAll and legacy migration; an older +snapshot cannot resurrect a deleted or migrated entry or overwrite an operator +edit. The existing tenant-local JSON shape, private file permissions and atomic +temporary-file/rename path stay unchanged. A failed write is reported and not +spin-retried; a later ordinary identity poll can retry the unpersisted state. -Known and deliberately unchanged: each changed guest still triggers a full-file -save, so one poll cycle over N changed guests performs N marshals and N atomic -writes that serialize on the store mutex. Coalescing them is a behavioural -change beyond the shutdown defect. +`WaitForPendingWrites` reports quiescence, not durable success. `Close` seals +mutation admission and observes the background writer and any admitted +synchronous write within the unchanged caller budget. Timeout or final +persistence failure remains an error. Monitor Stop logs that failure instead of +claiming a clean stop; tenant removal retains the runtime owner and lifecycle +guard, and organisation deletion refuses to remove its directory. An unfinished +monitoring loop likewise prevents deletion. Shutdown does not cancel an admitted +filesystem write or silently discard metadata to obtain a timely result. + +`internal/config/guest_writer_control_test.go` exercises blocked snapshots at +25/100 updates and caller-copy ownership. `guest_metadata_writer_test.go` covers +ordered replacement, synchronous mutations, field-scoped identity, failure and +explicit recovery, bounded close with blocked synchronous/asynchronous I/O, +closed admission, real filesystem permissions and tenant separation. Existing +ten-second drain and cleanup controls remain unchanged. Monitoring shutdown +controls retain blocked writer/loop ownership; the real organisation-delete +handler returns its safe failure before offboarding or directory removal. +These are source controls, not an attribution of an earlier drain timeout, +native CPU/RSS relief or all-path guest-agent recovery. ### Monitoring projects canonical resource policy into alert evaluation diff --git a/internal/api/org_handlers.go b/internal/api/org_handlers.go index 0704db88ef..fbae571366 100644 --- a/internal/api/org_handlers.go +++ b/internal/api/org_handlers.go @@ -387,7 +387,11 @@ func (h *OrgHandlers) HandleDeleteOrg(w http.ResponseWriter, r *http.Request) { } if h.mtMonitor != nil { - h.mtMonitor.BeginTenantDeletion(orgID) + if err := h.mtMonitor.BeginTenantDeletion(orgID); err != nil { + log.Error().Err(err).Str("org_id", orgID).Msg("Organization shutdown incomplete; retaining data") + writeErrorResponse(w, http.StatusServiceUnavailable, "tenant_shutdown_incomplete", "Organization monitoring has not shut down cleanly; its data has been retained", nil) + return + } defer h.mtMonitor.FinishTenantDeletion(orgID) } diff --git a/internal/api/org_shutdown_safety_test.go b/internal/api/org_shutdown_safety_test.go new file mode 100644 index 0000000000..90ee7f0855 --- /dev/null +++ b/internal/api/org_shutdown_safety_test.go @@ -0,0 +1,82 @@ +package api + +import ( + "context" + "net/http" + "net/http/httptest" + "os" + "path/filepath" + "strings" + "testing" + "time" + + "github.com/rcourtman/pulse-go-rewrite/internal/config" + "github.com/rcourtman/pulse-go-rewrite/internal/models" + "github.com/rcourtman/pulse-go-rewrite/internal/monitoring" +) + +func TestOrgDeleteRetainsDataWhenGuestMetadataPersistenceFails(t *testing.T) { + t.Setenv("PULSE_DEV", "true") + previous := IsMultiTenantEnabled() + SetMultiTenantEnabled(true) + t.Cleanup(func() { SetMultiTenantEnabled(previous) }) + root := t.TempDir() + persistence := config.NewMultiTenantPersistence(root) + mtm := monitoring.NewMultiTenantMonitor(&config.Config{DataPath: root}, persistence, nil) + t.Cleanup(mtm.Stop) + h := NewOrgHandlers(persistence, mtm) + // Exercise the same session-owner setup as ordinary organisation creation. + create := withUser(httptest.NewRequest(http.MethodPost, "/api/orgs", strings.NewReader(`{"id":"retained","displayName":"Retained"}`)), "alice") + rec := httptest.NewRecorder() + h.HandleCreateOrg(rec, create) + if rec.Code != http.StatusCreated { + t.Fatalf("create: %d %s", rec.Code, rec.Body.String()) + } + org, err := persistence.LoadOrganization("retained") + if err != nil || org == nil { + t.Fatal(err) + } + monitor, err := mtm.GetMonitor("retained") + if err != nil { + t.Fatal(err) + } + orgPath := filepath.Join(root, "orgs", "retained") + // A real rename refusal, confined to this disposable test directory. + if err := os.Mkdir(filepath.Join(orgPath, "guest_metadata.json"), 0o700); err != nil { + t.Fatal(err) + } + if err := monitor.GuestMetadataStore().RememberIdentity("pve:node:100", "unsaved", "qemu"); err != nil { + t.Fatal(err) + } + if !monitor.GuestMetadataStore().WaitForPendingWrites(time.Second) { + t.Fatal("failed attempt did not quiesce") + } + called := false + h.SetOnDelete(func(context.Context, string) error { called = true; return nil }) + req := withUser(httptest.NewRequest(http.MethodDelete, "/api/orgs/retained", nil), "alice") + req.SetPathValue("id", "retained") + rec = httptest.NewRecorder() + h.HandleDeleteOrg(rec, req) + if rec.Code != http.StatusServiceUnavailable || !strings.Contains(rec.Body.String(), "tenant_shutdown_incomplete") { + t.Fatalf("unsafe deletion result: %d %s", rec.Code, rec.Body.String()) + } + if called { + t.Fatal("offboarding callback ran after incomplete shutdown") + } + if _, err := os.Stat(filepath.Join(orgPath, "org.json")); err != nil { + t.Fatalf("organisation data removed: %v", err) + } + if got, ok := mtm.PeekMonitor("retained"); !ok || got != monitor { + t.Fatal("failed writer owner lost") + } + if _, err := mtm.GetMonitor("retained"); err == nil { + t.Fatal("new monitor can write into retained shutdown path") + } + // Another tenant remains usable: the failed writer is not a global hold. + if err := persistence.SaveOrganization(&models.Organization{ID: "other"}); err != nil { + t.Fatal(err) + } + if _, err := persistence.GetPersistence("other"); err != nil { + t.Fatal(err) + } +} diff --git a/internal/config/guest_metadata.go b/internal/config/guest_metadata.go index 41e3bbae42..d68e281060 100644 --- a/internal/config/guest_metadata.go +++ b/internal/config/guest_metadata.go @@ -2,11 +2,13 @@ package config import ( "encoding/json" + "errors" "fmt" "os" "path/filepath" "strings" "sync" + "sync/atomic" "time" "github.com/rs/zerolog/log" @@ -31,49 +33,176 @@ type GuestMetadataStore struct { dataPath string fs FileSystem - // inflight tracks background Set calls started by SetAsync so shutdown can - // wait for them. Callers used to spawn their own detached goroutines, which - // meant a write could land after the monitor stopped and after the data - // directory was being torn down, leaving a stray guest_metadata.json.tmp. - inflight sync.WaitGroup + // writeMu serializes disk snapshots with synchronous mutations. Background + // I/O never holds mu, so a burst can accumulate in one follow-up snapshot. + // Always acquire writeMu before mu when both locks are needed. + writeMu sync.Mutex + revision uint64 + attemptedRevision uint64 + lastWriteErr error + writerDone chan struct{} + closed atomic.Bool + closeOnce sync.Once + closeDone chan struct{} + closeErr error } -// SetAsync persists metadata without blocking the caller, while keeping the -// write owned by the store so WaitForPendingWrites can drain it on shutdown. -func (s *GuestMetadataStore) SetAsync(guestID string, meta *GuestMetadata) { +var ErrGuestMetadataStoreClosed = errors.New("guest metadata store is closed") + +// SetAsync admits a detached replacement before returning. One store-owned +// writer persists the newest complete snapshot, coalescing changes that arrive +// during I/O. It does not start a goroutine or rewrite the file per guest. +func (s *GuestMetadataStore) SetAsync(guestID string, meta *GuestMetadata) error { if s == nil { + return nil + } + if meta == nil { + return fmt.Errorf("metadata cannot be nil") + } + clone := cloneGuestMetadata(meta) + clone.ID = guestID + s.mu.Lock() + defer s.mu.Unlock() + if s.closed.Load() { + return ErrGuestMetadataStoreClosed + } + s.metadata[guestID] = clone + s.queueSaveLocked() + return nil +} + +// RememberIdentity merges only monitor-owned identity fields at admission. +// A background name/type update must never restore an old URL, tag or note +// over a concurrent operator edit. OCI classification cannot be downgraded by +// a later poll lacking the classification evidence. +func (s *GuestMetadataStore) RememberIdentity(guestID, name, guestType string) error { + if s == nil { + return nil + } + guestType = strings.TrimSpace(guestType) + if guestType == "" { + return nil + } + s.mu.Lock() + defer s.mu.Unlock() + if s.closed.Load() { + return ErrGuestMetadataStoreClosed + } + meta := cloneGuestMetadata(s.metadata[guestID]) + if meta == nil { + meta = &GuestMetadata{ID: guestID, Tags: []string{}} + } + if meta.LastKnownType == "oci" { + guestType = "oci" + } + if meta.LastKnownName == name && meta.LastKnownType == guestType && s.lastWriteErr == nil { + return nil + } + meta.LastKnownName = name + meta.LastKnownType = guestType + s.metadata[guestID] = meta + s.queueSaveLocked() + return nil +} + +func (s *GuestMetadataStore) queueSaveLocked() { + s.revision++ + if s.writerDone != nil { return } - s.inflight.Add(1) - go func() { - defer s.inflight.Done() - if err := s.Set(guestID, meta); err != nil { - log.Error().Err(err).Str("guestID", guestID).Msg("failed to persist guest metadata") - } - }() + s.writerDone = make(chan struct{}) + go s.writePendingSnapshots() } -// WaitForPendingWrites blocks until background writes finish or the timeout -// elapses. It reports whether the store drained; a false result means a write -// may still be in flight and the data directory is not safe to remove. +func (s *GuestMetadataStore) writePendingSnapshots() { + for { + s.writeMu.Lock() + s.mu.Lock() + if s.attemptedRevision == s.revision { + close(s.writerDone) + s.writerDone = nil + s.mu.Unlock() + s.writeMu.Unlock() + return + } + revision := s.revision + data, err := json.Marshal(s.metadata) + s.mu.Unlock() + if err == nil { + err = persistMetadata(s.fs, s.dataPath, "guest_metadata.json", data) + } + s.mu.Lock() + s.attemptedRevision = revision + s.lastWriteErr = err + s.mu.Unlock() + s.writeMu.Unlock() + if err != nil { + // No unbounded automatic retry on a failing filesystem. A later + // admitted mutation (or ordinary identity poll) can try again. + log.Error().Err(err).Msg("failed to persist guest metadata snapshot") + } + } +} + +// WaitForPendingWrites reports quiescence, not persistence success. Producers +// must already be stopped if the caller plans to remove the data directory; +// Close seals admission as well. A timeout never makes teardown safe. func (s *GuestMetadataStore) WaitForPendingWrites(timeout time.Duration) bool { if s == nil { return true } - done := make(chan struct{}) - go func() { - s.inflight.Wait() - close(done) - }() + s.mu.RLock() + done := s.writerDone + s.mu.RUnlock() + if done == nil { + return true + } + timer := time.NewTimer(timeout) + defer timer.Stop() select { case <-done: return true - case <-time.After(timeout): + case <-timer.C: log.Warn().Dur("timeout", timeout).Msg("timed out waiting for guest metadata writes to drain") return false } } +// Close rejects further mutations, drains the owned writer within the supplied +// budget, and reports any final persistence error. Failure keeps the writer +// owned and must stop tenant-directory removal; it is not a successful stop. +func (s *GuestMetadataStore) Close(timeout time.Duration) error { + if s == nil { + return nil + } + // Seal before waiting for a synchronous mutation's I/O-held memory lock. + // One owned close observer, not one goroutine per waiter, covers both that + // mutation and the background writer without extending the caller's budget. + s.closeOnce.Do(func() { + s.closed.Store(true) + go func() { + s.mu.RLock() + done := s.writerDone + s.mu.RUnlock() + if done != nil { + <-done + } + s.mu.RLock() + s.closeErr = s.lastWriteErr + s.mu.RUnlock() + close(s.closeDone) + }() + }) + timer := time.NewTimer(timeout) + defer timer.Stop() + select { + case <-s.closeDone: + return s.closeErr + case <-timer.C: + return fmt.Errorf("guest metadata writes did not drain within %s", timeout) + } +} + func cloneGuestMetadata(meta *GuestMetadata) *GuestMetadata { if meta == nil { return nil @@ -94,9 +223,10 @@ func cloneGuestMetadata(meta *GuestMetadata) *GuestMetadata { // NewGuestMetadataStore creates a new metadata store func NewGuestMetadataStore(dataPath string, fs FileSystem) *GuestMetadataStore { store := &GuestMetadataStore{ - metadata: make(map[string]*GuestMetadata), - dataPath: dataPath, - fs: fs, + metadata: make(map[string]*GuestMetadata), + dataPath: dataPath, + fs: fs, + closeDone: make(chan struct{}), } if store.fs == nil { @@ -113,6 +243,16 @@ func NewGuestMetadataStore(dataPath string, fs FileSystem) *GuestMetadataStore { // Load reads metadata from disk func (s *GuestMetadataStore) Load() error { + s.writeMu.Lock() + defer s.writeMu.Unlock() + s.mu.Lock() + defer s.mu.Unlock() + if s.closed.Load() { + return ErrGuestMetadataStoreClosed + } + if s.revision != s.attemptedRevision || s.lastWriteErr != nil { + return fmt.Errorf("cannot reload guest metadata with unpersisted changes") + } filePath := filepath.Join(s.dataPath, "guest_metadata.json") log.Debug().Str("path", filePath).Msg("Loading guest metadata from disk") @@ -127,9 +267,6 @@ func (s *GuestMetadataStore) Load() error { return fmt.Errorf("failed to read metadata file: %w", err) } - s.mu.Lock() - defer s.mu.Unlock() - if err := json.Unmarshal(data, &s.metadata); err != nil { return fmt.Errorf("failed to unmarshal metadata: %w", err) } @@ -138,24 +275,16 @@ func (s *GuestMetadataStore) Load() error { return nil } -// save writes metadata to disk (must be called with lock held) +// save writes a synchronous mutation (writeMu and mu must both be held). func (s *GuestMetadataStore) save() error { - filePath := filepath.Join(s.dataPath, "guest_metadata.json") - - log.Debug().Str("path", filePath).Msg("Saving guest metadata to disk") - + s.revision++ data, err := json.Marshal(s.metadata) - if err != nil { - return fmt.Errorf("failed to marshal metadata: %w", err) + if err == nil { + err = persistMetadata(s.fs, s.dataPath, "guest_metadata.json", data) } - - if err := persistMetadata(s.fs, s.dataPath, "guest_metadata.json", data); err != nil { - return err - } - - log.Debug().Str("path", filePath).Int("entries", len(s.metadata)).Msg("Guest metadata saved successfully") - - return nil + s.attemptedRevision = s.revision + s.lastWriteErr = err + return err } // Get retrieves metadata for a guest @@ -188,8 +317,13 @@ func (s *GuestMetadataStore) GetWithLegacyMigration(guestID, instance, node stri // Helper to migrate a legacy ID to the new format migrate := func(legacyID string) *GuestMetadata { + s.writeMu.Lock() + defer s.writeMu.Unlock() s.mu.Lock() defer s.mu.Unlock() + if s.closed.Load() { + return nil + } if current := s.metadata[guestID]; current != nil { return cloneGuestMetadata(current) @@ -293,8 +427,13 @@ func (s *GuestMetadataStore) GetAll() map[string]*GuestMetadata { // Set updates or creates metadata for a guest func (s *GuestMetadataStore) Set(guestID string, meta *GuestMetadata) error { + s.writeMu.Lock() + defer s.writeMu.Unlock() s.mu.Lock() defer s.mu.Unlock() + if s.closed.Load() { + return ErrGuestMetadataStoreClosed + } if meta == nil { return fmt.Errorf("metadata cannot be nil") @@ -310,8 +449,13 @@ func (s *GuestMetadataStore) Set(guestID string, meta *GuestMetadata) error { // Delete removes metadata for a guest func (s *GuestMetadataStore) Delete(guestID string) error { + s.writeMu.Lock() + defer s.writeMu.Unlock() s.mu.Lock() defer s.mu.Unlock() + if s.closed.Load() { + return ErrGuestMetadataStoreClosed + } delete(s.metadata, guestID) @@ -321,8 +465,13 @@ func (s *GuestMetadataStore) Delete(guestID string) error { // ReplaceAll replaces all metadata entries and persists them to disk. func (s *GuestMetadataStore) ReplaceAll(metadata map[string]*GuestMetadata) error { + s.writeMu.Lock() + defer s.writeMu.Unlock() s.mu.Lock() defer s.mu.Unlock() + if s.closed.Load() { + return ErrGuestMetadataStoreClosed + } s.metadata = make(map[string]*GuestMetadata) @@ -354,8 +503,13 @@ func (s *GuestMetadataStore) UpdateAll( return nil } + s.writeMu.Lock() + defer s.writeMu.Unlock() s.mu.Lock() defer s.mu.Unlock() + if s.closed.Load() { + return ErrGuestMetadataStoreClosed + } working := make(map[string]*GuestMetadata, len(s.metadata)) for id, meta := range s.metadata { diff --git a/internal/config/guest_metadata_writer_test.go b/internal/config/guest_metadata_writer_test.go new file mode 100644 index 0000000000..d71b685f2d --- /dev/null +++ b/internal/config/guest_metadata_writer_test.go @@ -0,0 +1,361 @@ +package config + +import ( + "encoding/json" + "errors" + "fmt" + "os" + "path/filepath" + "reflect" + "sync" + "sync/atomic" + "testing" + "time" +) + +func guestWriterTestStore(t *testing.T) (*GuestMetadataStore, *controlledGuestWriterFS) { + t.Helper() + fs := &controlledGuestWriterFS{FileSystem: defaultFileSystem{}, entered: make(chan struct{}), release: make(chan struct{})} + store := NewGuestMetadataStore(t.TempDir(), fs) + var releaseOnce sync.Once + t.Cleanup(func() { + releaseOnce.Do(func() { + select { + case <-fs.release: + default: + close(fs.release) + } + }) + if !store.WaitForPendingWrites(time.Second) { + t.Error("writer did not exit during cleanup") + } + }) + if err := store.SetAsync("first", &GuestMetadata{LastKnownName: "first"}); err != nil { + t.Fatal(err) + } + select { + case <-fs.entered: + case <-time.After(time.Second): + t.Fatal("writer not started") + } + return store, fs +} +func capturedGuestSnapshot(t *testing.T, fs *controlledGuestWriterFS) map[string]*GuestMetadata { + t.Helper() + fs.mu.Lock() + data := append([]byte(nil), fs.data...) + fs.mu.Unlock() + var result map[string]*GuestMetadata + if err := json.Unmarshal(data, &result); err != nil { + t.Fatal(err) + } + return result +} +func TestGuestMetadataWriterOrderedReplacementsAndReadsDuringIO(t *testing.T) { + store, fs := guestWriterTestStore(t) + for i := 0; i < 100; i++ { + if err := store.SetAsync("same", &GuestMetadata{LastKnownName: fmt.Sprint(i)}); err != nil { + t.Fatal(err) + } + } + // This is an admission-state read, not a persistence-success assertion. + read := make(chan *GuestMetadata, 1) + go func() { read <- store.Get("same") }() + select { + case got := <-read: + if got.LastKnownName != "99" { + t.Fatalf("wrong admission order: %+v", got) + } + case <-time.After(time.Second): + t.Fatal("disk I/O blocked metadata reads") + } + close(fs.release) + if err := store.Close(time.Second); err != nil { + t.Fatal(err) + } + if got := capturedGuestSnapshot(t, fs)["same"]; got.LastKnownName != "99" { + t.Fatalf("older update persisted last: %+v", got) + } + fs.mu.Lock() + defer fs.mu.Unlock() + if fs.writes != 2 { + t.Fatalf("100 replacements during one blocked write caused %d writes", fs.writes) + } +} +func TestGuestMetadataWriterSynchronousMutationsCannotBeOverwritten(t *testing.T) { + for _, operation := range []string{"set", "delete", "replace", "update", "migration"} { + t.Run(operation, func(t *testing.T) { + store, fs := guestWriterTestStore(t) + id := "pve:node:100" + if err := store.SetAsync(id, &GuestMetadata{LastKnownName: "queued"}); err != nil { + t.Fatal(err) + } + done := make(chan error, 1) + go func() { + switch operation { + case "set": + done <- store.Set(id, &GuestMetadata{LastKnownName: "operator", Notes: []string{"new"}}) + case "delete": + done <- store.Delete(id) + case "replace": + done <- store.ReplaceAll(map[string]*GuestMetadata{"replacement": {LastKnownName: "replacement"}}) + case "update": + done <- store.UpdateAll(func(m map[string]*GuestMetadata) bool { m[id].Notes = []string{"edited"}; return true }) + case "migration": + if store.GetWithLegacyMigration("pve:other:100", "pve", "other", 100) == nil { + done <- errors.New("migration missing") + } else { + done <- nil + } + } + }() + close(fs.release) + select { + case err := <-done: + if err != nil { + t.Fatal(err) + } + case <-time.After(time.Second): + t.Fatal("synchronous mutation blocked") + } + if err := store.Close(time.Second); err != nil { + t.Fatal(err) + } + saved := capturedGuestSnapshot(t, fs) + if !reflect.DeepEqual(saved, store.GetAll()) { + t.Fatalf("disk snapshot differs after %s: saved=%+v memory=%+v", operation, saved, store.GetAll()) + } + switch operation { + case "set": + if saved[id].LastKnownName != "operator" || saved[id].Notes[0] != "new" { + t.Fatal("operator replacement lost") + } + case "delete": + if saved[id] != nil { + t.Fatal("deleted entry resurrected") + } + case "replace": + if len(saved) != 1 || saved["replacement"] == nil { + t.Fatal("replace resurrected old entries") + } + case "update": + if saved[id].Notes[0] != "edited" { + t.Fatal("update lost") + } + case "migration": + if saved[id] != nil || saved["pve:other:100"] == nil { + t.Fatal("old node resurrected") + } + } + }) + } +} +func TestGuestMetadataWriterIdentityPreservesOperatorFieldsAndOCI(t *testing.T) { + store := NewGuestMetadataStore(t.TempDir(), nil) + original := &GuestMetadata{CustomURL: "https://operator.example", Description: "operator", Tags: []string{"tag"}, Notes: []string{"note"}, LastKnownName: "old", LastKnownType: "oci"} + if err := store.Set("pve:node:100", original); err != nil { + t.Fatal(err) + } + if err := store.RememberIdentity("pve:node:100", "new", " lxc "); err != nil { + t.Fatal(err) + } + got := store.Get("pve:node:100") + if got.LastKnownName != "new" || got.LastKnownType != "oci" { + t.Fatalf("identity/OCI lost: %+v", got) + } + original.ID = "pve:node:100" + original.LastKnownName = "new" + if !reflect.DeepEqual(got, original) { + t.Fatalf("identity changed operator data: %+v", got) + } + if !store.WaitForPendingWrites(time.Second) { + t.Fatal("identity not drained") + } + before := store.revision + if err := store.RememberIdentity("pve:node:100", "new", "oci"); err != nil { + t.Fatal(err) + } + if err := store.RememberIdentity("pve:node:100", "ignored", " "); err != nil { + t.Fatal(err) + } + if store.revision != before { + t.Fatal("unchanged/empty type triggered a write") + } + if err := store.Close(time.Second); err != nil { + t.Fatal(err) + } + reloaded := NewGuestMetadataStore(store.dataPath, nil) + if !reflect.DeepEqual(reloaded.Get("pve:node:100"), got) { + t.Fatal("identity/operator fields not persisted") + } +} +func TestGuestMetadataWriterCloseBoundsBlockedAsyncAndSyncIO(t *testing.T) { + for _, synchronous := range []bool{false, true} { + t.Run(fmt.Sprint(synchronous), func(t *testing.T) { + fs := &controlledGuestWriterFS{FileSystem: defaultFileSystem{}, entered: make(chan struct{}), release: make(chan struct{})} + store := NewGuestMetadataStore(t.TempDir(), fs) + done := make(chan error, 1) + if synchronous { + go func() { done <- store.Set("first", &GuestMetadata{LastKnownName: "first"}) }() + } else { + done <- store.SetAsync("first", &GuestMetadata{LastKnownName: "first"}) + } + select { + case <-fs.entered: + case <-time.After(time.Second): + t.Fatal("write did not start") + } + t.Cleanup(func() { + select { + case <-fs.release: + default: + close(fs.release) + } + _ = store.Close(time.Second) + }) + start := time.Now() + if err := store.Close(10 * time.Millisecond); err == nil { + t.Fatal("blocked writer reported close success") + } + if elapsed := time.Since(start); elapsed > time.Second { + t.Fatalf("close exceeded predeclared bounded budget: %s", elapsed) + } + close(fs.release) + select { + case err := <-done: + if err != nil { + t.Fatal(err) + } + case <-time.After(time.Second): + t.Fatal("write still running") + } + if err := store.Close(time.Second); err != nil { + t.Fatal(err) + } + for _, mutation := range []func() error{ + func() error { return store.SetAsync("later", &GuestMetadata{}) }, + func() error { return store.RememberIdentity("later", "later", "qemu") }, + func() error { return store.Set("later", &GuestMetadata{}) }, + func() error { return store.Delete("first") }, + func() error { return store.ReplaceAll(nil) }, + func() error { return store.UpdateAll(func(map[string]*GuestMetadata) bool { return true }) }, + store.Load, + } { + if err := mutation(); !errors.Is(err, ErrGuestMetadataStoreClosed) { + t.Fatalf("closed admission accepted: %v", err) + } + } + if store.GetWithLegacyMigration("node:100", "pve", "node", 100) != nil { + t.Fatal("closed store migrated metadata") + } + if capturedGuestSnapshot(t, fs)["later"] != nil { + t.Fatal("post-close write") + } + }) + } +} + +type guestWriterFaultFS struct { + FileSystem + failWrite atomic.Bool + failRename atomic.Bool + writes atomic.Int32 +} + +func (f *guestWriterFaultFS) WriteFile(path string, data []byte, perm os.FileMode) error { + f.writes.Add(1) + if f.failWrite.Load() { + return errors.New("controlled write failure") + } + return f.FileSystem.WriteFile(path, data, perm) +} +func (f *guestWriterFaultFS) Rename(a, b string) error { + if f.failRename.Load() { + return errors.New("controlled rename failure") + } + return f.FileSystem.Rename(a, b) +} +func TestGuestMetadataWriterFailuresAreOwnedWithoutSpinRetry(t *testing.T) { + for _, phase := range []string{"write", "rename"} { + t.Run(phase, func(t *testing.T) { + fs := &guestWriterFaultFS{FileSystem: defaultFileSystem{}} + store := NewGuestMetadataStore(t.TempDir(), fs) + if err := store.Set("key", &GuestMetadata{LastKnownName: "original"}); err != nil { + t.Fatal(err) + } + fs.failWrite.Store(phase == "write") + fs.failRename.Store(phase == "rename") + if err := store.RememberIdentity("key", "changed", "qemu"); err != nil { + t.Fatal(err) + } + if !store.WaitForPendingWrites(time.Second) { + t.Fatal("failed attempt not quiescent") + } + if fs.writes.Load() != 2 { + t.Fatalf("failure retried automatically: %d", fs.writes.Load()) + } + if got := NewGuestMetadataStore(store.dataPath, nil).Get("key"); got.LastKnownName != "original" { + t.Fatal("failure replaced good disk snapshot") + } + fs.failWrite.Store(false) + fs.failRename.Store(false) + // A later ordinary poll retries the unpersisted identity, not a tight loop. + if err := store.RememberIdentity("key", "changed", "qemu"); err != nil { + t.Fatal(err) + } + if err := store.Close(time.Second); err != nil { + t.Fatal(err) + } + if fs.writes.Load() != 3 { + t.Fatalf("explicit recovery writes=%d", fs.writes.Load()) + } + if got := NewGuestMetadataStore(store.dataPath, nil).Get("key"); got.LastKnownName != "changed" { + t.Fatal("recovery not persisted") + } + fs.failWrite.Store(true) + failed := NewGuestMetadataStore(t.TempDir(), fs) + if err := failed.SetAsync("lost", &GuestMetadata{}); err != nil { + t.Fatal(err) + } + if err := failed.Close(time.Second); err == nil { + t.Fatal("failed final write reported durable close") + } + }) + } +} +func TestGuestMetadataWriterRealFilesystemTenantIsolation(t *testing.T) { + root := t.TempDir() + a := NewGuestMetadataStore(filepath.Join(root, "orgs", "a"), nil) + b := NewGuestMetadataStore(filepath.Join(root, "orgs", "b"), nil) + for i := 0; i < 100; i++ { + id := fmt.Sprintf("pve:node:%d", i) + if err := a.RememberIdentity(id, "a", "qemu"); err != nil { + t.Fatal(err) + } + if err := b.RememberIdentity(id, "b", "lxc"); err != nil { + t.Fatal(err) + } + } + for _, store := range []*GuestMetadataStore{a, b} { + if err := store.Close(10 * time.Second); err != nil { + t.Fatal(err) + } + reloaded := NewGuestMetadataStore(store.dataPath, nil) + if !reflect.DeepEqual(reloaded.GetAll(), store.GetAll()) { + t.Fatal("real filesystem lost entries") + } + info, err := os.Stat(filepath.Join(store.dataPath, "guest_metadata.json")) + if err != nil { + t.Fatal(err) + } + if info.Mode().Perm() != 0o600 { + t.Fatalf("unsafe file permissions: %v", info.Mode()) + } + if _, err := os.Stat(filepath.Join(store.dataPath, "guest_metadata.json.tmp")); !os.IsNotExist(err) { + t.Fatalf("temporary file survived close: %v", err) + } + } + if a.Get("pve:node:1").LastKnownName == b.Get("pve:node:1").LastKnownName { + t.Fatal("tenants share identity") + } +} diff --git a/internal/config/guest_writer_control_test.go b/internal/config/guest_writer_control_test.go new file mode 100644 index 0000000000..a1c2e9e7ee --- /dev/null +++ b/internal/config/guest_writer_control_test.go @@ -0,0 +1,100 @@ +package config + +import ( + "encoding/json" + "fmt" + "os" + "sync" + "testing" + "time" +) + +type controlledGuestWriterFS struct { + FileSystem + mu sync.Mutex + writes int + data []byte + entered chan struct{} + release chan struct{} +} + +func (f *controlledGuestWriterFS) MkdirAll(string, os.FileMode) error { return nil } +func (f *controlledGuestWriterFS) Rename(string, string) error { return nil } +func (f *controlledGuestWriterFS) WriteFile(_ string, data []byte, _ os.FileMode) error { + f.mu.Lock() + f.writes++ + first := f.writes == 1 + f.mu.Unlock() + if first { + close(f.entered) + <-f.release + } + f.mu.Lock() + f.data = append([]byte(nil), data...) + f.mu.Unlock() + return nil +} +func TestGuestWriterControlledBurst(t *testing.T) { + for _, count := range []int{25, 100} { + t.Run(fmt.Sprint(count), func(t *testing.T) { + fs := &controlledGuestWriterFS{FileSystem: defaultFileSystem{}, entered: make(chan struct{}), release: make(chan struct{})} + store := NewGuestMetadataStore(t.TempDir(), fs) + store.SetAsync("first", &GuestMetadata{LastKnownName: "first"}) + select { + case <-fs.entered: + case <-time.After(time.Second): + t.Fatal("first write did not start") + } + for i := 0; i < count; i++ { + store.SetAsync(fmt.Sprint(i), &GuestMetadata{LastKnownName: fmt.Sprint(i)}) + } + close(fs.release) + if !store.WaitForPendingWrites(10 * time.Second) { + t.Fatal("drain failed") + } + fs.mu.Lock() + writes := fs.writes + data := append([]byte(nil), fs.data...) + fs.mu.Unlock() + t.Logf("admitted=%d complete-file-writes=%d", count+1, writes) + if writes > 2 { + t.Errorf("one blocked snapshot plus one complete follow-up needed, got %d", writes) + } + var saved map[string]*GuestMetadata + if err := json.Unmarshal(data, &saved); err != nil { + t.Fatal(err) + } + if len(saved) != count+1 { + t.Fatalf("lost guests: got %d want %d", len(saved), count+1) + } + for i := 0; i < count; i++ { + if saved[fmt.Sprint(i)].LastKnownName != fmt.Sprint(i) { + t.Errorf("wrong guest %d", i) + } + } + }) + } +} +func TestGuestWriterCopiesAtAdmission(t *testing.T) { + fs := &controlledGuestWriterFS{FileSystem: defaultFileSystem{}, entered: make(chan struct{}), release: make(chan struct{})} + store := NewGuestMetadataStore(t.TempDir(), fs) + store.SetAsync("first", &GuestMetadata{LastKnownName: "first"}) + select { + case <-fs.entered: + case <-time.After(time.Second): + t.Fatal("first write did not start") + } + meta := &GuestMetadata{LastKnownName: "original", Tags: []string{"original"}, Notes: []string{"original"}} + store.SetAsync("second", meta) + meta.LastKnownName = "mutated" + meta.Tags[0] = "mutated" + meta.Notes[0] = "mutated" + close(fs.release) + if !store.WaitForPendingWrites(10 * time.Second) { + t.Fatal("drain failed") + } + got := store.Get("second") + if got.LastKnownName != "original" || got.Tags[0] != "original" || got.Notes[0] != "original" { + t.Fatalf("caller mutation reached stored metadata: %+v", got) + } +} diff --git a/internal/monitoring/guest_metadata_shutdown_test.go b/internal/monitoring/guest_metadata_shutdown_test.go new file mode 100644 index 0000000000..8c9ad302ee --- /dev/null +++ b/internal/monitoring/guest_metadata_shutdown_test.go @@ -0,0 +1,125 @@ +package monitoring + +import ( + "context" + "os" + "path/filepath" + "testing" + "time" + + "github.com/rcourtman/pulse-go-rewrite/internal/config" +) + +type blockedGuestMetadataFS struct { + config.FileSystem + entered chan struct{} + release chan struct{} +} + +func (f *blockedGuestMetadataFS) ReadFile(p string) ([]byte, error) { return os.ReadFile(p) } +func (f *blockedGuestMetadataFS) Stat(p string) (os.FileInfo, error) { return os.Stat(p) } +func (f *blockedGuestMetadataFS) MkdirAll(p string, m os.FileMode) error { return os.MkdirAll(p, m) } +func (f *blockedGuestMetadataFS) WriteFile(p string, b []byte, m os.FileMode) error { + close(f.entered) + <-f.release + return os.WriteFile(p, b, m) +} +func (f *blockedGuestMetadataFS) Rename(a, b string) error { return os.Rename(a, b) } + +func TestTenantDeletionRetainsBlockedGuestWriterOwnership(t *testing.T) { + path := t.TempDir() + fs := &blockedGuestMetadataFS{entered: make(chan struct{}), release: make(chan struct{})} + store := config.NewGuestMetadataStore(path, fs) + monitor := &Monitor{guestMetadataStore: store} + mtm := NewMultiTenantMonitor(&config.Config{DataPath: path}, nil, nil) + mtm.monitors["blocked"] = monitor + t.Cleanup(func() { + select { + case <-fs.release: + default: + close(fs.release) + } + _ = store.Close(time.Second) + mtm.Stop() + }) + persistGuestIdentity(store, "pve:node:100", "retained", "qemu") + select { + case <-fs.entered: + case <-time.After(time.Second): + t.Fatal("write did not start") + } + if err := mtm.BeginTenantDeletion("blocked"); err == nil { + t.Fatal("deletion accepted incomplete drain") + } + if retained, ok := mtm.PeekMonitor("blocked"); !ok || retained != monitor { + t.Fatal("in-flight writer owner was discarded") + } + mtm.FinishTenantDeletion("blocked") + if _, err := mtm.GetMonitor("blocked"); err == nil { + t.Fatal("deferred cleanup released unsafe lifecycle guard") + } + if _, err := os.Stat(path); err != nil { + t.Fatalf("data path not retained: %v", err) + } + close(fs.release) + if err := store.Close(time.Second); err != nil { + t.Fatal(err) + } + if err := mtm.BeginTenantDeletion("blocked"); err != nil { + t.Fatal(err) + } + if _, ok := mtm.PeekMonitor("blocked"); ok { + t.Fatal("completed shutdown not removed") + } + if err := store.RememberIdentity("late", "late", "qemu"); err == nil { + t.Fatal("stopped store accepted late identity") + } + if _, err := os.Stat(filepath.Join(path, "guest_metadata.json")); err != nil { + t.Fatal("accepted metadata not persisted") + } +} +func TestTenantDeletionRetainsUnfinishedMonitorLoop(t *testing.T) { + mtm := NewMultiTenantMonitor(&config.Config{}, nil, nil) + _, cancel := context.WithCancel(context.Background()) + done := make(chan struct{}) + monitor := &Monitor{} + mtm.monitors["loop"] = monitor + mtm.tenantCancel["loop"] = cancel + mtm.tenantDone["loop"] = done + t.Cleanup(func() { + select { + case <-done: + default: + close(done) + } + mtm.Stop() + }) + if err := mtm.BeginTenantDeletion("loop"); err == nil { + t.Fatal("unfinished monitoring loop admitted directory deletion") + } + if retained, ok := mtm.PeekMonitor("loop"); !ok || retained != monitor { + t.Fatal("unfinished monitoring loop lost owner") + } + if _, err := mtm.GetMonitor("loop"); err == nil { + t.Fatal("unfinished monitor replaced") + } + close(done) + if err := mtm.BeginTenantDeletion("loop"); err != nil { + t.Fatal(err) + } +} +func TestPersistGuestIdentityEditsAreImmediateAndFieldScoped(t *testing.T) { + store := config.NewGuestMetadataStore(t.TempDir(), nil) + if err := store.Set("pve:node:100", &config.GuestMetadata{CustomURL: "https://operator.example", Tags: []string{"important"}, Notes: []string{"keep"}}); err != nil { + t.Fatal(err) + } + persistGuestIdentity(store, "pve:node:100", "old", "qemu") + persistGuestIdentity(store, "pve:node:100", "new", "qemu") + got := store.Get("pve:node:100") + if got.LastKnownName != "new" || got.CustomURL != "https://operator.example" || got.Tags[0] != "important" || got.Notes[0] != "keep" { + t.Fatalf("producer corrupted metadata: %+v", got) + } + if err := store.Close(time.Second); err != nil { + t.Fatal(err) + } +} diff --git a/internal/monitoring/monitor.go b/internal/monitoring/monitor.go index 6e9091900e..6bab63084c 100644 --- a/internal/monitoring/monitor.go +++ b/internal/monitoring/monitor.go @@ -7598,6 +7598,12 @@ func monitorLastSeenUnix(value time.Time) int64 { const guestMetadataDrainTimeout = 2 * time.Second func (m *Monitor) Stop() { + if err := m.stop(); err != nil { + log.Error().Err(err).Msg("monitor stopped with incomplete guest metadata persistence; data directory must be retained") + } +} + +func (m *Monitor) stop() error { log.Info().Msg("stopping monitor") // A trailing agent-report refresh must not run against stores closed below. @@ -7620,9 +7626,7 @@ func (m *Monitor) Stop() { // Drain background guest-metadata writes before the data directory can be // torn down. Without this a queued write lands after shutdown and leaves a // stray guest_metadata.json.tmp behind. - if m.guestMetadataStore != nil { - m.guestMetadataStore.WaitForPendingWrites(guestMetadataDrainTimeout) - } + drainErr := m.guestMetadataStore.Close(guestMetadataDrainTimeout) // Close persistent metrics store (flushes buffered data) if m.metricsStore != nil { @@ -7633,7 +7637,11 @@ func (m *Monitor) Stop() { } } + if drainErr != nil { + return drainErr + } log.Info().Msg("monitor stopped") + return nil } // recordAuthFailure records an authentication failure for a node diff --git a/internal/monitoring/monitor_backups.go b/internal/monitoring/monitor_backups.go index 1dc59f1184..d5475857eb 100644 --- a/internal/monitoring/monitor_backups.go +++ b/internal/monitoring/monitor_backups.go @@ -872,33 +872,8 @@ func enrichWithPersistedMetadata(metadataStore *config.GuestMetadataStore, byVMI // persistGuestIdentity updates the metadata store with the last-known name and type for a guest func persistGuestIdentity(metadataStore *config.GuestMetadataStore, guestKey, name, guestType string) { - existing := metadataStore.Get(guestKey) - if existing == nil { - existing = &config.GuestMetadata{ - ID: guestKey, - Tags: []string{}, - } - } - - guestType = strings.TrimSpace(guestType) - if guestType == "" { - return - } - - // Never "downgrade" OCI containers back to LXC. OCI classification can be transiently - // unavailable if Proxmox config reads fail due to permissions or transient API errors. - if existing.LastKnownType == "oci" && guestType != "oci" { - guestType = existing.LastKnownType - } - - // Only update if the name or type has changed - if existing.LastKnownName != name || existing.LastKnownType != guestType { - existing.LastKnownName = name - existing.LastKnownType = guestType - // Save without blocking the monitor. The store owns the goroutine so - // Monitor.Stop can drain it; a detached goroutine here could write - // after shutdown, into a data directory that was already being removed. - metadataStore.SetAsync(guestKey, existing) + if err := metadataStore.RememberIdentity(guestKey, name, guestType); err != nil { + log.Error().Err(err).Str("guestKey", guestKey).Msg("failed to admit guest identity update") } } diff --git a/internal/monitoring/monitor_metadata_test.go b/internal/monitoring/monitor_metadata_test.go index 12f3d3e658..e63bd48fed 100644 --- a/internal/monitoring/monitor_metadata_test.go +++ b/internal/monitoring/monitor_metadata_test.go @@ -20,15 +20,6 @@ func TestPersistGuestIdentity_Concurrent(t *testing.T) { } defer os.RemoveAll(tmpDir) - // metadataFile := filepath.Join(tmpDir, "guest_metadata.json") // Actually NewGuestMetadataStore takes directory not file usually? No, it takes root dir? - // Let's verify standard usage. "store := NewGuestMetadataStore(tmpDir, nil)" - // Implementation: func NewGuestMetadataStore(dataPath string, fs FileSystem) - // Inside it does filepath.Join(dataPath, "guest_metadata.json") ? - // Let's re-read NewGuestMetadataStore in internal/config/guest_metadata.go via grep or similar if needed. - // But based on "config.NewGuestMetadataStore(metadataFile)" from my previous code failing, and grep showing "dataPath", it likely takes a Dir or File path. - // grep output: guestMetadataStore := NewGuestMetadataStore(dataPath, c.fs) - // Most likely directory. - store := config.NewGuestMetadataStore(tmpDir, nil) guestKey := "pve1:node1:100" @@ -36,8 +27,9 @@ func TestPersistGuestIdentity_Concurrent(t *testing.T) { // Test basic persistence persistGuestIdentity(store, guestKey, "VM 100", "qemu") - // Wait a bit since persistGuestIdentity is async - time.Sleep(50 * time.Millisecond) + if !store.WaitForPendingWrites(time.Second) { + t.Fatal("identity writer did not drain") + } meta := store.Get(guestKey) if meta == nil || meta.LastKnownName != "VM 100" || meta.LastKnownType != "qemu" { @@ -55,7 +47,9 @@ func TestPersistGuestIdentity_Concurrent(t *testing.T) { // Try to update to "lxc" persistGuestIdentity(store, guestKey, "VM 100", "lxc") - time.Sleep(50 * time.Millisecond) + if !store.WaitForPendingWrites(time.Second) { + t.Fatal("identity writer did not drain") + } meta = store.Get(guestKey) if meta.LastKnownType != "oci" { @@ -65,6 +59,9 @@ func TestPersistGuestIdentity_Concurrent(t *testing.T) { // Test persistence that shouldn't happen (no change) -> coverage of the if check // Should not trigger Set() persistGuestIdentity(store, guestKey, "VM 100", "oci") + if err := store.Close(time.Second); err != nil { + t.Fatal(err) + } } func TestEnrichWithPersistedMetadata_Detail(t *testing.T) { diff --git a/internal/monitoring/multi_tenant_monitor.go b/internal/monitoring/multi_tenant_monitor.go index 14a76d656b..248b842239 100644 --- a/internal/monitoring/multi_tenant_monitor.go +++ b/internal/monitoring/multi_tenant_monitor.go @@ -20,6 +20,7 @@ type MultiTenantMonitor struct { tenantCancel map[string]context.CancelFunc tenantDone map[string]chan struct{} tenantDeleting map[string]struct{} + tenantStopping map[string]struct{} persistence *config.MultiTenantPersistence baseConfig *config.Config wsHub *websocket.Hub @@ -37,6 +38,7 @@ func NewMultiTenantMonitor(baseCfg *config.Config, persistence *config.MultiTena tenantCancel: make(map[string]context.CancelFunc), tenantDone: make(map[string]chan struct{}), tenantDeleting: make(map[string]struct{}), + tenantStopping: make(map[string]struct{}), persistence: persistence, baseConfig: baseCfg, // Used as a template or for global settings wsHub: wsHub, @@ -294,47 +296,34 @@ func (mtm *MultiTenantMonitor) Stop() { log.Info().Msg("stopping MultiTenantMonitor and all tenant instances") mtm.mu.Lock() - mtm.globalCancel() - - monitors := make([]*Monitor, 0, len(mtm.monitors)) - cancels := make([]context.CancelFunc, 0, len(mtm.tenantCancel)) - doneSignals := make([]chan struct{}, 0, len(mtm.tenantDone)) - for orgID, monitor := range mtm.monitors { - if cancel := mtm.tenantCancel[orgID]; cancel != nil { - cancels = append(cancels, cancel) - } - if done := mtm.tenantDone[orgID]; done != nil { - doneSignals = append(doneSignals, done) - } - monitors = append(monitors, monitor) + if mtm.globalCancel != nil { + mtm.globalCancel() + } + orgIDs := make([]string, 0, len(mtm.monitors)) + for orgID := range mtm.monitors { + orgIDs = append(orgIDs, orgID) } - mtm.tenantCancel = make(map[string]context.CancelFunc) - mtm.tenantDone = make(map[string]chan struct{}) - mtm.monitors = make(map[string]*Monitor) mtm.mu.Unlock() - - for _, cancel := range cancels { - cancel() - } - for _, done := range doneSignals { - waitForTenantMonitorShutdown("", done) - } - for _, monitor := range monitors { - monitor.Stop() + for _, orgID := range orgIDs { + if err := mtm.removeTenant(orgID, false); err != nil { + log.Error().Err(err).Str("org_id", orgID).Msg("tenant shutdown incomplete; retaining runtime ownership and data") + } } } // RemoveTenant stops and removes a specific tenant's monitor. // Useful for offboarding or manual reloading. func (mtm *MultiTenantMonitor) RemoveTenant(orgID string) { - mtm.removeTenant(orgID, false) + if err := mtm.removeTenant(orgID, false); err != nil { + log.Error().Err(err).Str("org_id", orgID).Msg("tenant removal incomplete; retaining runtime ownership and data") + } } // BeginTenantDeletion prevents lazy monitor initialization while a tenant's // persistence is being removed. Call FinishTenantDeletion after the -// persistence operation completes, whether it succeeds or fails. -func (mtm *MultiTenantMonitor) BeginTenantDeletion(orgID string) { - mtm.removeTenant(orgID, true) +// persistence operation completes. An incomplete shutdown retains its guard. +func (mtm *MultiTenantMonitor) BeginTenantDeletion(orgID string) error { + return mtm.removeTenant(orgID, true) } // FinishTenantDeletion releases the temporary lifecycle guard installed by @@ -347,46 +336,70 @@ func (mtm *MultiTenantMonitor) FinishTenantDeletion(orgID string) { } mtm.mu.Lock() - delete(mtm.tenantDeleting, orgID) + // An incomplete removal still owns a runtime or writer. Do not release + // its guard merely because an offboarding caller ran deferred cleanup. + if _, retained := mtm.monitors[orgID]; !retained { + delete(mtm.tenantDeleting, orgID) + } mtm.mu.Unlock() } -func (mtm *MultiTenantMonitor) removeTenant(orgID string, deleting bool) { +func (mtm *MultiTenantMonitor) removeTenant(orgID string, deleting bool) error { orgID = strings.TrimSpace(orgID) if orgID == "" { - return + return nil } mtm.mu.Lock() - if deleting { - if mtm.tenantDeleting == nil { - mtm.tenantDeleting = make(map[string]struct{}) - } - mtm.tenantDeleting[orgID] = struct{}{} + if mtm.tenantDeleting == nil { + mtm.tenantDeleting = make(map[string]struct{}) } - monitor, exists := mtm.monitors[orgID] + if mtm.tenantStopping == nil { + mtm.tenantStopping = make(map[string]struct{}) + } + if _, stopping := mtm.tenantStopping[orgID]; stopping { + mtm.mu.Unlock() + return fmt.Errorf("organization %q shutdown is already in progress", orgID) + } + mtm.tenantDeleting[orgID] = struct{}{} + mtm.tenantStopping[orgID] = struct{}{} + monitor := mtm.monitors[orgID] cancel := mtm.tenantCancel[orgID] done := mtm.tenantDone[orgID] - delete(mtm.monitors, orgID) - delete(mtm.tenantCancel, orgID) - delete(mtm.tenantDone, orgID) mtm.mu.Unlock() - - if !exists { - return - } + defer func() { + mtm.mu.Lock() + delete(mtm.tenantStopping, orgID) + mtm.mu.Unlock() + }() log.Info().Str("org_id", orgID).Msg("stopping and removing tenant monitor") if cancel != nil { cancel() } - waitForTenantMonitorShutdown(orgID, done) - monitor.Stop() + if !waitForTenantMonitorShutdown(orgID, done) { + return fmt.Errorf("organization %q monitoring loop has not stopped", orgID) + } + if monitor != nil { + if err := monitor.stop(); err != nil { + return fmt.Errorf("organization %q guest metadata shutdown incomplete: %w", orgID, err) + } + } + + mtm.mu.Lock() + delete(mtm.monitors, orgID) + delete(mtm.tenantCancel, orgID) + delete(mtm.tenantDone, orgID) + if !deleting { + delete(mtm.tenantDeleting, orgID) + } + mtm.mu.Unlock() + return nil } -func waitForTenantMonitorShutdown(orgID string, done <-chan struct{}) { +func waitForTenantMonitorShutdown(orgID string, done <-chan struct{}) bool { if done == nil { - return + return true } timer := time.NewTimer(tenantMonitorShutdownTimeout) @@ -394,12 +407,14 @@ func waitForTenantMonitorShutdown(orgID string, done <-chan struct{}) { select { case <-done: + return true case <-timer.C: logger := log.Warn().Dur("timeout", tenantMonitorShutdownTimeout) if orgID != "" { logger = logger.Str("org_id", orgID) } logger.Msg("timed out waiting for tenant monitor loop to exit") + return false } }