From 625ac3766ac22ed0c4748bc304f0e7ada8e31bdd Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Alex=20S=C3=A1nchez?= Date: Fri, 10 Jul 2026 11:44:24 -0600 Subject: [PATCH 1/3] fix[backend](integrations): fixed plugins yaml files integration format to match event processor expected format --- .../integrations/repository/tenant_store.go | 31 +++++++++--- .../repository/tenant_store_test.go | 48 +++++++++++++++++++ 2 files changed, 73 insertions(+), 6 deletions(-) create mode 100644 backend/modules/integrations/repository/tenant_store_test.go diff --git a/backend/modules/integrations/repository/tenant_store.go b/backend/modules/integrations/repository/tenant_store.go index 26a8a5cb0..a504d3f9a 100644 --- a/backend/modules/integrations/repository/tenant_store.go +++ b/backend/modules/integrations/repository/tenant_store.go @@ -37,18 +37,31 @@ var pluginNames = map[string]string{ "SOPHOS": "sophos", } -func tenantFileName(module string) string { +func pluginKey(module string) string { name, ok := pluginNames[strings.ToUpper(module)] if !ok { name = strings.ToLower(module) } - return "system_plugins_" + name + ".yaml" + return name +} + +func tenantFileName(module string) string { + return "system_plugins_" + pluginKey(module) + ".yaml" } func (s *TenantStore) path(module string) string { return filepath.Join(s.dir, tenantFileName(module)) } +// pluginsFile is the on-disk wrapper: plugins..tenants: [...]. +type pluginsFile struct { + Plugins map[string]pluginEntry `yaml:"plugins"` +} + +type pluginEntry struct { + Tenants []domain.Tenant `yaml:"tenants"` +} + func (s *TenantStore) SetActiveByModule(_ context.Context, module string, active bool) error { if active { return nil @@ -70,18 +83,24 @@ func (s *TenantStore) Load(module string) ([]domain.Tenant, error) { if err != nil { return nil, err } - var tenants []domain.Tenant - if err := yaml.Unmarshal(data, &tenants); err != nil { + var pf pluginsFile + if err := yaml.Unmarshal(data, &pf); err != nil { return nil, err } - return tenants, nil + entry, ok := pf.Plugins[pluginKey(module)] + if !ok { + return []domain.Tenant{}, nil + } + return entry.Tenants, nil } func (s *TenantStore) Save(module string, tenants []domain.Tenant) error { if err := os.MkdirAll(s.dir, 0o755); err != nil { return err } - data, err := yaml.Marshal(tenants) + key := pluginKey(module) + pf := pluginsFile{Plugins: map[string]pluginEntry{key: {Tenants: tenants}}} + data, err := yaml.Marshal(pf) if err != nil { return err } diff --git a/backend/modules/integrations/repository/tenant_store_test.go b/backend/modules/integrations/repository/tenant_store_test.go new file mode 100644 index 000000000..b117a654e --- /dev/null +++ b/backend/modules/integrations/repository/tenant_store_test.go @@ -0,0 +1,48 @@ +package repository + +import ( + "os" + "strings" + "testing" + + "github.com/utmstack/utmstack/backend/modules/integrations/domain" +) + +func TestTenantStoreRoundtrip(t *testing.T) { + dir := t.TempDir() + s := NewTenantStore(dir) + + want := []domain.Tenant{ + {Name: "t1", Config: map[string]string{"access_key": "AK", "secret": "SK"}}, + {Name: "t2", Config: map[string]string{"access_key": "AK2"}}, + } + if err := s.Save("AWS_IAM_USER", want); err != nil { + t.Fatalf("Save: %v", err) + } + got, err := s.Load("AWS_IAM_USER") + if err != nil { + t.Fatalf("Load: %v", err) + } + if len(got) != len(want) { + t.Fatalf("len(got)=%d want %d", len(got), len(want)) + } + for i := range want { + if got[i].Name != want[i].Name { + t.Fatalf("[%d] name got %q want %q", i, got[i].Name, want[i].Name) + } + for k, v := range want[i].Config { + if got[i].Config[k] != v { + t.Fatalf("[%d] %s got %q want %q", i, k, got[i].Config[k], v) + } + } + } + + data, err := os.ReadFile(s.path("AWS_IAM_USER")) + if err != nil { + t.Fatalf("read: %v", err) + } + txt := string(data) + if !strings.Contains(txt, "plugins:") || !strings.Contains(txt, "aws:") || !strings.Contains(txt, "tenants:") { + t.Fatalf("unexpected file layout:\n%s", txt) + } +} From 54c767976076b5789da1bdb74d84213f1386e07a Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Alex=20S=C3=A1nchez?= Date: Fri, 10 Jul 2026 12:17:37 -0600 Subject: [PATCH 2/3] fix[backend](integrations): added lock files to avoid data duplication or ccorruption --- .../integrations/repository/tenant_store.go | 110 ++++++++++++------ .../repository/tenant_store_test.go | 61 ++++++++++ 2 files changed, 138 insertions(+), 33 deletions(-) diff --git a/backend/modules/integrations/repository/tenant_store.go b/backend/modules/integrations/repository/tenant_store.go index a504d3f9a..7381266a4 100644 --- a/backend/modules/integrations/repository/tenant_store.go +++ b/backend/modules/integrations/repository/tenant_store.go @@ -7,12 +7,46 @@ import ( "path/filepath" "strings" "sync" + "time" "github.com/utmstack/utmstack/backend/modules/integrations/connectors" "github.com/utmstack/utmstack/backend/modules/integrations/domain" "gopkg.in/yaml.v3" ) +// lockStaleness is how long a lock can sit untouched before we assume the +// writer crashed and steal it. Real writes are a few ms; anything past this is +// almost certainly a dead process. +const lockStaleness = 30 * time.Second + +// withFileLock acquires an exclusive cross-process lock via a .lock file, +// runs fn, then removes the lock. O_EXCL makes check-and-create atomic. If the +// existing lock is older than lockStaleness (writer crashed), it is stolen. +// ponytail: mtime-based staleness has a small race window if two processes +// steal simultaneously; upgrade to flock/fcntl if we ever see corruption. +func withFileLock(path string, fn func() error) error { + if err := os.MkdirAll(filepath.Dir(path), 0o755); err != nil { + return err + } + lock := path + ".lock" + for { + f, err := os.OpenFile(lock, os.O_CREATE|os.O_EXCL|os.O_WRONLY, 0o600) + if err == nil { + _ = f.Close() + defer os.Remove(lock) + return fn() + } + if !errors.Is(err, os.ErrExist) { + return err + } + if info, statErr := os.Stat(lock); statErr == nil && time.Since(info.ModTime()) > lockStaleness { + _ = os.Remove(lock) + continue + } + time.Sleep(50 * time.Millisecond) + } +} + var _ connectors.TenantRepository = (*TenantStore)(nil) type TenantStore struct { @@ -69,10 +103,12 @@ func (s *TenantStore) SetActiveByModule(_ context.Context, module string, active s.mu.Lock() defer s.mu.Unlock() - if err := os.Remove(s.path(module)); err != nil && !errors.Is(err, os.ErrNotExist) { - return err - } - return nil + return withFileLock(s.path(module), func() error { + if err := os.Remove(s.path(module)); err != nil && !errors.Is(err, os.ErrNotExist) { + return err + } + return nil + }) } func (s *TenantStore) Load(module string) ([]domain.Tenant, error) { @@ -95,9 +131,13 @@ func (s *TenantStore) Load(module string) ([]domain.Tenant, error) { } func (s *TenantStore) Save(module string, tenants []domain.Tenant) error { - if err := os.MkdirAll(s.dir, 0o755); err != nil { - return err - } + return withFileLock(s.path(module), func() error { + return s.saveLocked(module, tenants) + }) +} + +// saveLocked writes the file assuming the caller already holds the file lock. +func (s *TenantStore) saveLocked(module string, tenants []domain.Tenant) error { key := pluginKey(module) pf := pluginsFile{Plugins: map[string]pluginEntry{key: {Tenants: tenants}}} data, err := yaml.Marshal(pf) @@ -116,38 +156,42 @@ func (s *TenantStore) Upsert(module string, t domain.Tenant) error { s.mu.Lock() defer s.mu.Unlock() - tenants, err := s.Load(module) - if err != nil { - return err - } - for i := range tenants { - if tenants[i].Name == t.Name { - tenants[i] = t - return s.Save(module, tenants) + return withFileLock(s.path(module), func() error { + tenants, err := s.Load(module) + if err != nil { + return err } - } - return s.Save(module, append(tenants, t)) + for i := range tenants { + if tenants[i].Name == t.Name { + tenants[i] = t + return s.saveLocked(module, tenants) + } + } + return s.saveLocked(module, append(tenants, t)) + }) } func (s *TenantStore) Delete(module, name string) error { s.mu.Lock() defer s.mu.Unlock() - tenants, err := s.Load(module) - if err != nil { - return err - } - out := make([]domain.Tenant, 0, len(tenants)) - found := false - for _, t := range tenants { - if t.Name == name { - found = true - continue + return withFileLock(s.path(module), func() error { + tenants, err := s.Load(module) + if err != nil { + return err } - out = append(out, t) - } - if !found { - return domain.ErrTenantNotFound - } - return s.Save(module, out) + out := make([]domain.Tenant, 0, len(tenants)) + found := false + for _, t := range tenants { + if t.Name == name { + found = true + continue + } + out = append(out, t) + } + if !found { + return domain.ErrTenantNotFound + } + return s.saveLocked(module, out) + }) } diff --git a/backend/modules/integrations/repository/tenant_store_test.go b/backend/modules/integrations/repository/tenant_store_test.go index b117a654e..3ba9850d4 100644 --- a/backend/modules/integrations/repository/tenant_store_test.go +++ b/backend/modules/integrations/repository/tenant_store_test.go @@ -4,6 +4,7 @@ import ( "os" "strings" "testing" + "time" "github.com/utmstack/utmstack/backend/modules/integrations/domain" ) @@ -45,4 +46,64 @@ func TestTenantStoreRoundtrip(t *testing.T) { if !strings.Contains(txt, "plugins:") || !strings.Contains(txt, "aws:") || !strings.Contains(txt, "tenants:") { t.Fatalf("unexpected file layout:\n%s", txt) } + + // lockfile must be removed after Save + if _, err := os.Stat(s.path("AWS_IAM_USER") + ".lock"); !os.IsNotExist(err) { + t.Fatalf("lockfile still present after Save: err=%v", err) + } +} + +func TestWithFileLockStealsStale(t *testing.T) { + dir := t.TempDir() + target := dir + "/f.yaml" + lock := target + ".lock" + + if err := os.WriteFile(lock, nil, 0o600); err != nil { + t.Fatalf("preheld: %v", err) + } + old := time.Now().Add(-2 * lockStaleness) + if err := os.Chtimes(lock, old, old); err != nil { + t.Fatalf("chtimes: %v", err) + } + + start := time.Now() + if err := withFileLock(target, func() error { return nil }); err != nil { + t.Fatalf("withFileLock: %v", err) + } + if elapsed := time.Since(start); elapsed > 200*time.Millisecond { + t.Fatalf("stale lock not stolen fast enough: %v", elapsed) + } +} + +func TestWithFileLockWaits(t *testing.T) { + dir := t.TempDir() + target := dir + "/f.yaml" + lock := target + ".lock" + + // pre-hold the lock + if err := os.WriteFile(lock, nil, 0o600); err != nil { + t.Fatalf("preheld: %v", err) + } + + done := make(chan error, 1) + go func() { + done <- withFileLock(target, func() error { return nil }) + }() + + select { + case <-done: + t.Fatal("withFileLock returned while lock was held") + case <-time.After(120 * time.Millisecond): + } + + // release; now it should complete + _ = os.Remove(lock) + select { + case err := <-done: + if err != nil { + t.Fatalf("withFileLock: %v", err) + } + case <-time.After(500 * time.Millisecond): + t.Fatal("withFileLock did not return after lock released") + } } From 9074439592005ab34b6a175422619343533b947e Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Alex=20S=C3=A1nchez?= Date: Fri, 10 Jul 2026 12:28:05 -0600 Subject: [PATCH 3/3] fix[backend](integrations): used flock to handle locks with kernel locking --- .../integrations/repository/tenant_store.go | 40 ++++------ .../repository/tenant_store_test.go | 76 ++++++++++--------- 2 files changed, 54 insertions(+), 62 deletions(-) diff --git a/backend/modules/integrations/repository/tenant_store.go b/backend/modules/integrations/repository/tenant_store.go index 7381266a4..e9ac8a650 100644 --- a/backend/modules/integrations/repository/tenant_store.go +++ b/backend/modules/integrations/repository/tenant_store.go @@ -7,44 +7,30 @@ import ( "path/filepath" "strings" "sync" - "time" + "syscall" "github.com/utmstack/utmstack/backend/modules/integrations/connectors" "github.com/utmstack/utmstack/backend/modules/integrations/domain" "gopkg.in/yaml.v3" ) -// lockStaleness is how long a lock can sit untouched before we assume the -// writer crashed and steal it. Real writes are a few ms; anything past this is -// almost certainly a dead process. -const lockStaleness = 30 * time.Second - -// withFileLock acquires an exclusive cross-process lock via a .lock file, -// runs fn, then removes the lock. O_EXCL makes check-and-create atomic. If the -// existing lock is older than lockStaleness (writer crashed), it is stolen. -// ponytail: mtime-based staleness has a small race window if two processes -// steal simultaneously; upgrade to flock/fcntl if we ever see corruption. +// withFileLock acquires an exclusive advisory lock on .lock via flock, +// runs fn, releases the lock on close. The kernel drops the lock on fd close +// (including process death), so a crashed writer never leaves a stuck lock. +// Linux/BSD only; UTMStack runs in Linux containers. func withFileLock(path string, fn func() error) error { if err := os.MkdirAll(filepath.Dir(path), 0o755); err != nil { return err } - lock := path + ".lock" - for { - f, err := os.OpenFile(lock, os.O_CREATE|os.O_EXCL|os.O_WRONLY, 0o600) - if err == nil { - _ = f.Close() - defer os.Remove(lock) - return fn() - } - if !errors.Is(err, os.ErrExist) { - return err - } - if info, statErr := os.Stat(lock); statErr == nil && time.Since(info.ModTime()) > lockStaleness { - _ = os.Remove(lock) - continue - } - time.Sleep(50 * time.Millisecond) + lf, err := os.OpenFile(path+".lock", os.O_CREATE|os.O_RDWR, 0o600) + if err != nil { + return err + } + defer lf.Close() + if err := syscall.Flock(int(lf.Fd()), syscall.LOCK_EX); err != nil { + return err } + return fn() } var _ connectors.TenantRepository = (*TenantStore)(nil) diff --git a/backend/modules/integrations/repository/tenant_store_test.go b/backend/modules/integrations/repository/tenant_store_test.go index 3ba9850d4..a272bda21 100644 --- a/backend/modules/integrations/repository/tenant_store_test.go +++ b/backend/modules/integrations/repository/tenant_store_test.go @@ -3,6 +3,7 @@ package repository import ( "os" "strings" + "syscall" "testing" "time" @@ -46,43 +47,19 @@ func TestTenantStoreRoundtrip(t *testing.T) { if !strings.Contains(txt, "plugins:") || !strings.Contains(txt, "aws:") || !strings.Contains(txt, "tenants:") { t.Fatalf("unexpected file layout:\n%s", txt) } - - // lockfile must be removed after Save - if _, err := os.Stat(s.path("AWS_IAM_USER") + ".lock"); !os.IsNotExist(err) { - t.Fatalf("lockfile still present after Save: err=%v", err) - } -} - -func TestWithFileLockStealsStale(t *testing.T) { - dir := t.TempDir() - target := dir + "/f.yaml" - lock := target + ".lock" - - if err := os.WriteFile(lock, nil, 0o600); err != nil { - t.Fatalf("preheld: %v", err) - } - old := time.Now().Add(-2 * lockStaleness) - if err := os.Chtimes(lock, old, old); err != nil { - t.Fatalf("chtimes: %v", err) - } - - start := time.Now() - if err := withFileLock(target, func() error { return nil }); err != nil { - t.Fatalf("withFileLock: %v", err) - } - if elapsed := time.Since(start); elapsed > 200*time.Millisecond { - t.Fatalf("stale lock not stolen fast enough: %v", elapsed) - } } func TestWithFileLockWaits(t *testing.T) { dir := t.TempDir() target := dir + "/f.yaml" - lock := target + ".lock" - // pre-hold the lock - if err := os.WriteFile(lock, nil, 0o600); err != nil { - t.Fatalf("preheld: %v", err) + // pre-hold the flock via a peer fd + peer, err := os.OpenFile(target+".lock", os.O_CREATE|os.O_RDWR, 0o600) + if err != nil { + t.Fatalf("open: %v", err) + } + if err := syscall.Flock(int(peer.Fd()), syscall.LOCK_EX); err != nil { + t.Fatalf("flock: %v", err) } done := make(chan error, 1) @@ -92,18 +69,47 @@ func TestWithFileLockWaits(t *testing.T) { select { case <-done: - t.Fatal("withFileLock returned while lock was held") + t.Fatal("withFileLock returned while flock was held") case <-time.After(120 * time.Millisecond): } - // release; now it should complete - _ = os.Remove(lock) + if err := syscall.Flock(int(peer.Fd()), syscall.LOCK_UN); err != nil { + t.Fatalf("unlock: %v", err) + } + _ = peer.Close() + select { case err := <-done: if err != nil { t.Fatalf("withFileLock: %v", err) } case <-time.After(500 * time.Millisecond): - t.Fatal("withFileLock did not return after lock released") + t.Fatal("withFileLock did not return after flock released") + } +} + +// TestWithFileLockKernelReleasesOnClose proves the crash-recovery property: if +// the peer holding the flock closes its fd (equivalent to process death), the +// kernel releases the lock and a new acquirer proceeds without any TTL/steal. +func TestWithFileLockKernelReleasesOnClose(t *testing.T) { + dir := t.TempDir() + target := dir + "/f.yaml" + + peer, err := os.OpenFile(target+".lock", os.O_CREATE|os.O_RDWR, 0o600) + if err != nil { + t.Fatalf("open: %v", err) + } + if err := syscall.Flock(int(peer.Fd()), syscall.LOCK_EX); err != nil { + t.Fatalf("flock: %v", err) + } + // simulate crash: close fd without explicit unlock + _ = peer.Close() + + start := time.Now() + if err := withFileLock(target, func() error { return nil }); err != nil { + t.Fatalf("withFileLock: %v", err) + } + if elapsed := time.Since(start); elapsed > 100*time.Millisecond { + t.Fatalf("lock not released on close: %v", elapsed) } }