From e90a1a18519ecc4577e423d8a044357608888004 Mon Sep 17 00:00:00 2001 From: User Date: Fri, 21 Aug 2026 00:39:35 +0200 Subject: [PATCH] fix: synchronize DWH registry snapshots --- tools/dwh-auth/internal/registry/store.go | 70 ++++++++++------- .../dwh-auth/internal/registry/store_test.go | 46 +++++------ .../internal/securefile/securefile_linux.go | 12 ++- .../securefile/securefile_linux_test.go | 78 +++++++++++++++++++ 4 files changed, 157 insertions(+), 49 deletions(-) diff --git a/tools/dwh-auth/internal/registry/store.go b/tools/dwh-auth/internal/registry/store.go index e12aa4e0..a9c42a41 100644 --- a/tools/dwh-auth/internal/registry/store.go +++ b/tools/dwh-auth/internal/registry/store.go @@ -25,7 +25,10 @@ import ( "github.com/aritmolab/thothii/tools/dwh-auth/internal/securefile" ) -const maxRecordBytes = 4096 +const ( + maxRecordBytes = 4096 + registryLockName = ".writer.lock" +) // State describes which registry directory owns a public record. type State string @@ -155,12 +158,13 @@ func (s *Store) Find(keyID string) (record.Record, error) { if !validKeyID(keyID) { return record.Record{}, ErrNotFound } - if s == nil { - return record.Record{}, integrity(errors.New("uninitialized store")) - } - s.mu.RLock() - defer s.mu.RUnlock() - return s.findUnlocked(keyID) + var value record.Record + err := s.withReaderLock(func() error { + var findErr error + value, findErr = s.findUnlocked(keyID) + return findErr + }) + return value, err } func (s *Store) findUnlocked(keyID string) (record.Record, error) { @@ -182,12 +186,13 @@ func (s *Store) findUnlocked(keyID string) (record.Record, error) { // FindLegacy returns the sole active legacy_raw record. A revoked legacy record // wins; any multiple-legacy condition is an integrity fault. func (s *Store) FindLegacy() (record.Record, error) { - if s == nil { - return record.Record{}, integrity(errors.New("uninitialized store")) - } - s.mu.RLock() - defer s.mu.RUnlock() - return s.findLegacyUnlocked() + var value record.Record + err := s.withReaderLock(func() error { + var findErr error + value, findErr = s.findLegacyUnlocked() + return findErr + }) + return value, err } func (s *Store) findLegacyUnlocked() (record.Record, error) { @@ -217,12 +222,13 @@ func (s *Store) findLegacyUnlocked() (record.Record, error) { // List returns a stable, redacted inventory. A revoked record replaces any // same-key active record visible during a revoked-first transition. func (s *Store) List() ([]PublicRecord, error) { - if s == nil { - return nil, integrity(errors.New("uninitialized store")) - } - s.mu.RLock() - defer s.mu.RUnlock() - return s.listUnlocked() + var records []PublicRecord + err := s.withReaderLock(func() error { + var listErr error + records, listErr = s.listUnlocked() + return listErr + }) + return records, err } func (s *Store) listUnlocked() ([]PublicRecord, error) { @@ -304,12 +310,7 @@ func (s *Store) revokeUnlocked(keyID, reason string, at time.Time) error { // Check validates every record and protected directory without exposing any // digest data. func (s *Store) Check() error { - if s == nil { - return integrity(errors.New("uninitialized store")) - } - s.mu.RLock() - defer s.mu.RUnlock() - return s.checkUnlocked() + return s.withReaderLock(s.checkUnlocked) } func (s *Store) checkUnlocked() error { @@ -320,6 +321,23 @@ func (s *Store) checkUnlocked() error { return validateLegacyMultiplicity(active, revoked) } +func (s *Store) withReaderLock(fn func() error) error { + if s == nil { + return integrity(errors.New("uninitialized store")) + } + s.mu.RLock() + defer s.mu.RUnlock() + if s.root == nil || s.active == nil || s.revoked == nil { + return integrity(errors.New("uninitialized store")) + } + lock, err := s.root.LockShared(registryLockName) + if err != nil { + return integrity(err) + } + defer lock.Close() + return fn() +} + func (s *Store) withWriterLock(fn func() error) error { if s == nil { return integrity(errors.New("uninitialized store")) @@ -329,7 +347,7 @@ func (s *Store) withWriterLock(fn func() error) error { if s.root == nil || s.active == nil || s.revoked == nil { return integrity(errors.New("uninitialized store")) } - lock, err := s.root.Lock(".writer.lock") + lock, err := s.root.Lock(registryLockName) if err != nil { return integrity(err) } diff --git a/tools/dwh-auth/internal/registry/store_test.go b/tools/dwh-auth/internal/registry/store_test.go index 560a7fd3..3027e0d2 100644 --- a/tools/dwh-auth/internal/registry/store_test.go +++ b/tools/dwh-auth/internal/registry/store_test.go @@ -353,7 +353,7 @@ func TestConcurrentAddDoesNotOverwriteAndFindNeverReadsPartialRecord(t *testing. } }() } - if err := readStore.Revoke(first.KeyID, "synthetic rotation", first.CreatedAt.Add(time.Hour)); err != nil { + if err := left.Revoke(first.KeyID, "synthetic rotation", first.CreatedAt.Add(time.Hour)); err != nil { t.Fatalf("Revoke() during concurrent reads error = %v", err) } readers.Wait() @@ -597,11 +597,12 @@ func TestFindAndFindLegacyRejectExactExpiration(t *testing.T) { } } -func TestReadOperationsWaitAcrossRevokePublicationAndUnlink(t *testing.T) { +func TestCrossStoreReadersWaitAcrossRevokePublicationAndUnlink(t *testing.T) { root := t.TempDir() - store := openStore(t, root) + writer := openStore(t, root) + reader := openStore(t, root) legacy := syntheticLegacyRecord() - if err := store.Add(legacy); err != nil { + if err := writer.Add(legacy); err != nil { t.Fatalf("Add() error = %v", err) } @@ -610,17 +611,17 @@ func TestReadOperationsWaitAcrossRevokePublicationAndUnlink(t *testing.T) { writerDone := make(chan error, 1) revokedAt := legacy.CreatedAt.Add(time.Hour) go func() { - writerDone <- store.withWriterLock(func() error { + writerDone <- writer.withWriterLock(func() error { close(writerReady) <-releaseWriter revoked := revokedRecord(legacy, revokedAt, "synthetic rotation") - if err := store.writeRecord(store.revoked, revoked); err != nil { + if err := writer.writeRecord(writer.revoked, revoked); err != nil { return err } - if err := store.active.Remove(recordFileName(legacy.KeyID)); err != nil { + if err := writer.active.Remove(recordFileName(legacy.KeyID)); err != nil { return err } - return store.active.Sync() + return writer.active.Sync() }) }() <-writerReady @@ -633,10 +634,10 @@ func TestReadOperationsWaitAcrossRevokePublicationAndUnlink(t *testing.T) { name string run func() error }{ - {name: "Find", run: func() error { _, err := store.Find(legacy.KeyID); return err }}, - {name: "List", run: func() error { _, err := store.List(); return err }}, - {name: "Check", run: store.Check}, - {name: "FindLegacy", run: func() error { _, err := store.FindLegacy(); return err }}, + {name: "Find", run: func() error { _, err := reader.Find(legacy.KeyID); return err }}, + {name: "List", run: func() error { _, err := reader.List(); return err }}, + {name: "Check", run: reader.Check}, + {name: "FindLegacy", run: func() error { _, err := reader.FindLegacy(); return err }}, } results := make(chan readResult, len(readers)) var started sync.WaitGroup @@ -689,11 +690,12 @@ waitForReaders: } } -func TestScanReadersWaitForWriterTemporaryFile(t *testing.T) { +func TestCrossStoreScanReadersWaitForWriterTemporaryFile(t *testing.T) { root := t.TempDir() - store := openStore(t, root) + writer := openStore(t, root) + reader := openStore(t, root) legacy := syntheticLegacyRecord() - if err := store.Add(legacy); err != nil { + if err := writer.Add(legacy); err != nil { t.Fatalf("Add() error = %v", err) } @@ -702,8 +704,8 @@ func TestScanReadersWaitForWriterTemporaryFile(t *testing.T) { writerDone := make(chan error, 1) const temporary = ".tmp-regression" go func() { - writerDone <- store.withWriterLock(func() error { - file, err := store.active.CreateExclusive(temporary, 0o600) + writerDone <- writer.withWriterLock(func() error { + file, err := writer.active.CreateExclusive(temporary, 0o600) if err != nil { return err } @@ -712,10 +714,10 @@ func TestScanReadersWaitForWriterTemporaryFile(t *testing.T) { } close(writerReady) <-releaseWriter - if err := store.active.Remove(temporary); err != nil { + if err := writer.active.Remove(temporary); err != nil { return err } - return store.active.Sync() + return writer.active.Sync() }) }() <-writerReady @@ -728,9 +730,9 @@ func TestScanReadersWaitForWriterTemporaryFile(t *testing.T) { name string run func() error }{ - {name: "List", run: func() error { _, err := store.List(); return err }}, - {name: "Check", run: store.Check}, - {name: "FindLegacy", run: func() error { _, err := store.FindLegacy(); return err }}, + {name: "List", run: func() error { _, err := reader.List(); return err }}, + {name: "Check", run: reader.Check}, + {name: "FindLegacy", run: func() error { _, err := reader.FindLegacy(); return err }}, } results := make(chan readResult, len(readers)) var started sync.WaitGroup diff --git a/tools/dwh-auth/internal/securefile/securefile_linux.go b/tools/dwh-auth/internal/securefile/securefile_linux.go index 1f57fc65..2783ed6e 100644 --- a/tools/dwh-auth/internal/securefile/securefile_linux.go +++ b/tools/dwh-auth/internal/securefile/securefile_linux.go @@ -291,6 +291,16 @@ func (d *Dir) Names() ([]string, error) { // Lock opens or creates a protected 0600 lock file and acquires an exclusive // advisory lock. Close releases the lock and descriptor. func (d *Dir) Lock(name string) (*Lock, error) { + return d.lock(name, syscall.LOCK_EX) +} + +// LockShared opens or creates a protected 0600 lock file and acquires a shared +// advisory lock. Close releases the lock and descriptor. +func (d *Dir) LockShared(name string) (*Lock, error) { + return d.lock(name, syscall.LOCK_SH) +} + +func (d *Dir) lock(name string, lockMode int) (*Lock, error) { if err := validName(name); err != nil { return nil, err } @@ -327,7 +337,7 @@ func (d *Dir) Lock(name string) (*Lock, error) { _ = syscall.Close(fd) return nil, err } - if err := syscall.Flock(fd, syscall.LOCK_EX); err != nil { + if err := syscall.Flock(fd, lockMode); err != nil { _ = syscall.Close(fd) return nil, err } diff --git a/tools/dwh-auth/internal/securefile/securefile_linux_test.go b/tools/dwh-auth/internal/securefile/securefile_linux_test.go index 6a68140d..7f85a9a9 100644 --- a/tools/dwh-auth/internal/securefile/securefile_linux_test.go +++ b/tools/dwh-auth/internal/securefile/securefile_linux_test.go @@ -7,6 +7,7 @@ import ( "errors" "os" "path/filepath" + "syscall" "testing" ) @@ -204,6 +205,83 @@ func TestReadSecretAndCreateSecretUse0600AndExclusiveCreate(t *testing.T) { } } +func TestLockSharedAllowsReadersAndBlocksExclusiveWriter(t *testing.T) { + root := t.TempDir() + first, err := OpenDir(root) + if err != nil { + t.Fatalf("OpenDir(first) error = %v", err) + } + defer first.Close() + second, err := OpenDir(root) + if err != nil { + t.Fatalf("OpenDir(second) error = %v", err) + } + defer second.Close() + third, err := OpenDir(root) + if err != nil { + t.Fatalf("OpenDir(third) error = %v", err) + } + defer third.Close() + if first.fd == second.fd || second.fd == third.fd || first.fd == third.fd { + t.Fatal("OpenDir() reused a descriptor; want separate lock file descriptions") + } + + type sharedLocker interface { + LockShared(string) (*Lock, error) + } + firstLocker, ok := any(first).(sharedLocker) + if !ok { + t.Fatal("Dir lacks shared advisory locking") + } + secondLocker, ok := any(second).(sharedLocker) + if !ok { + t.Fatal("Dir lacks shared advisory locking") + } + + firstShared, err := firstLocker.LockShared("registry.lock") + if err != nil { + t.Fatalf("LockShared(first) error = %v", err) + } + defer firstShared.Close() + secondShared, err := secondLocker.LockShared("registry.lock") + if err != nil { + t.Fatalf("LockShared(second) error = %v", err) + } + defer secondShared.Close() + + probe, err := os.OpenFile(filepath.Join(root, "registry.lock"), os.O_RDWR, 0) + if err != nil { + t.Fatalf("OpenFile(lock probe) error = %v", err) + } + if err := syscall.Flock(int(probe.Fd()), syscall.LOCK_EX|syscall.LOCK_NB); err == nil { + _ = syscall.Flock(int(probe.Fd()), syscall.LOCK_UN) + _ = probe.Close() + t.Fatal("exclusive lock acquired while separate shared locks were held") + } else if !errors.Is(err, syscall.EWOULDBLOCK) && !errors.Is(err, syscall.EAGAIN) { + _ = probe.Close() + t.Fatalf("nonblocking exclusive lock error = %v, want EWOULDBLOCK", err) + } + if err := probe.Close(); err != nil { + t.Fatalf("Close(lock probe) error = %v", err) + } + + if err := secondShared.Close(); err != nil { + t.Fatalf("Close(second shared) error = %v", err) + } + secondShared = nil + if err := firstShared.Close(); err != nil { + t.Fatalf("Close(first shared) error = %v", err) + } + firstShared = nil + exclusive, err := third.Lock("registry.lock") + if err != nil { + t.Fatalf("Lock(exclusive after shared release) error = %v", err) + } + if err := exclusive.Close(); err != nil { + t.Fatalf("Close(exclusive) error = %v", err) + } +} + func writeFile(t *testing.T, path string, data []byte, mode os.FileMode) { t.Helper() if err := os.WriteFile(path, data, mode); err != nil {