mptyl/ThothII
1
0
Fork 0
Code Issues 25 Pull Requests Actions Packages Projects Releases 1 Wiki Activity

fix: synchronize DWH registry snapshots

Browse Source
This commit is contained in:
User
2026-08-21 00:39:35 +02:00
parent 971a0e66b1
commit e90a1a1851
4 changed files with 157 additions and 49 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tools/dwh-auth/internal/registry/store.go
+44 -26
View File
@@ -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)
}
tools/dwh-auth/internal/registry/store_test.go
+24 -22
View File
@@ -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
tools/dwh-auth/internal/securefile/securefile_linux.go
+11 -1
View File
@@ -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
}
tools/dwh-auth/internal/securefile/securefile_linux_test.go
+78
View File
@@ -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 {