diff --git a/internal/storage/index/file/file.go b/internal/storage/index/file/file.go index 9cb9a31b..9d91e9f6 100644 --- a/internal/storage/index/file/file.go +++ b/internal/storage/index/file/file.go @@ -444,10 +444,17 @@ func (u *userIndex) folderVolatileDir(folder string) string { return filepath.Join(u.volatileDir, mailbox.FolderSubpath(u.driver, folder, folder, u.separator)) } -// withFolderRO reloads the folder state under the in-process mutex only and -// runs fn. No distributed lock is acquired — safe for read-only operations -// because the on-disk fileindex is append-only and a partially-written append -// cannot produce a torn read at the record boundary. +// withFolderRO reloads the folder state, then runs read-only fn against the +// settled in-memory snapshot under a shared lock. +// +// The reload is serialized against writers via the SAME cross-process lock the +// write path (withFolderLock) takes (#647): an unlocked reload could interleave +// with another process's lock-holding compaction (flush + truncateLog) and load +// a torn view into the shared in-memory folderState — which every subsequent, +// correctly locked write then trusts as a baseline, regressing the folder's +// NextUID. The distributed lock is held ONLY around reload(); fn then reads the +// settled snapshot under fs.mu.RLock without holding the exclusive resource, so +// concurrent readers do not serialize against each other beyond the reload. func (u *userIndex) withFolderRO(folderID uint64, fn func(*folderState) error) error { u.mu.Lock() fs, ok := u.open[folderID] @@ -455,19 +462,11 @@ func (u *userIndex) withFolderRO(folderID uint64, fn func(*folderState) error) e if !ok { return fmt.Errorf("fileindex: folder %d not open", folderID) } - // Brief exclusive lock to reload on-disk state (another process may - // have written since our last flush). Held only for the disk read. - // - // NOTE (#647): this reload is NOT serialized against u.b.locker, the - // cross-process distributed lock every write path (withFolderLock) takes - // before its own reload. A concurrent, lock-holding compaction on another - // process can interleave with this unlocked read and load a torn view into - // the shared in-memory folderState — which every subsequent, correctly - // locked write then trusts as a baseline. Passing locked=false so the - // reload debug breadcrumbs make this window visible in a live repro. - fs.mu.Lock() - err := fs.reload(false) - fs.mu.Unlock() + err := u.withDistLock(fs, func() error { + fs.mu.Lock() + defer fs.mu.Unlock() + return fs.reload(true) + }) if err != nil && !errors.Is(err, os.ErrNotExist) { return err } @@ -487,9 +486,26 @@ func (u *userIndex) withFolderRO(folderID uint64, fn func(*folderState) error) e // storage calls that touch the same key) does not deadlock against // itself. func (u *userIndex) withFolderLock(fs *folderState, fn func() error) error { - // Acquire the distributed lock BEFORE taking fs.mu so that a slow - // lock-wait (up to 35 s) does not block concurrent readers that only - // need fs.mu.RLock() via withFolderRO. + return u.withDistLock(fs, func() error { + fs.mu.Lock() + defer fs.mu.Unlock() + t1 := time.Now() + err := fn() + slog.Debug("fileindex: lock fn", + "user", u.username, "folder", fs.folder, + "fn_ms", time.Since(t1).Milliseconds()) + return err + }) +} + +// withDistLock runs fn while holding the cross-process index lock for fs.folder. +// It acquires the lock BEFORE fn touches fs.mu so a slow lock-wait (up to 35 s) +// does not block concurrent readers that only need fs.mu.RLock(). The +// HoldsResource() shortcut keeps it re-entrant: an outer caller that already +// holds the key (the POP3 QUIT pattern, or withFolderRO nested inside a locked +// write) runs fn without re-acquiring, so it cannot deadlock against itself. +// When no locker is wired (tests) fn runs unguarded. +func (u *userIndex) withDistLock(fs *folderState, fn func() error) error { if u.b.locker != nil { key := locks.MailboxKey(u.username, fs.folder) if !u.b.locker.HoldsResource(key) { @@ -500,21 +516,13 @@ func (u *userIndex) withFolderLock(fs *folderState, fn func() error) error { if err != nil { return fmt.Errorf("fileindex/lock %s: %w", fs.folder, err) } - lockWait := time.Since(t0) slog.Debug("fileindex: lock wait", "user", u.username, "folder", fs.folder, - "lock_wait_ms", lockWait.Milliseconds()) + "lock_wait_ms", time.Since(t0).Milliseconds()) defer func() { _ = u.b.locker.Unlock(ctx, lk.ID) }() } } - fs.mu.Lock() - defer fs.mu.Unlock() - t1 := time.Now() - err := fn() - slog.Debug("fileindex: lock fn", - "user", u.username, "folder", fs.folder, - "fn_ms", time.Since(t1).Milliseconds()) - return err + return fn() } // withTwoFolderLocks acquires the X locks for folderA and folderB diff --git a/internal/storage/integration/ro_reload_lock_test.go b/internal/storage/integration/ro_reload_lock_test.go new file mode 100644 index 00000000..a1924ae8 --- /dev/null +++ b/internal/storage/integration/ro_reload_lock_test.go @@ -0,0 +1,86 @@ +package integration_test + +import ( + "context" + "testing" + "time" + + "github.com/0kaba0hub/yarilo/internal/storage/index/file" + "github.com/0kaba0hub/yarilo/pkg/locks" + "github.com/0kaba0hub/yarilo/pkg/mailbox" +) + +// TestReadPathSerializesAgainstConcurrentLockHolder is the #647 regression: +// withFolderRO's reload must be serialized against writers via the same +// cross-process lock the write path takes. Before the fix the read path took +// only the in-process fs.mu and could interleave with another process's +// lock-holding compaction, poisoning the shared in-memory header (NextUID +// regression). Two clients on one embedded lock server stand in for two pods: +// client B holds the folder's X lock (an in-progress compaction), and a +// read-only op on the index wired to client A must block until B releases. +func TestReadPathSerializesAgainstConcurrentLockHolder(t *testing.T) { + sock := holdsTestSocket(t) + clientA := newHoldsClient(t, sock) + clientB := newHoldsClient(t, sock) + + const username = "dave@example.com" + home := t.TempDir() + user := &mailbox.UserInfo{Username: username, Home: home} + idxA := file.New(file.WithLocker(clientA)).OpenUser(user) + t.Cleanup(func() { _ = idxA.Close() }) + + folder, err := idxA.OpenFolder("INBOX", 1) + if err != nil { + t.Fatalf("open folder: %v", err) + } + uid, err := idxA.AllocateUID(folder.ID) + if err != nil { + t.Fatalf("allocate: %v", err) + } + if err := idxA.AppendMessage(folder.ID, &mailbox.MessageMeta{UID: uid, Filename: "1.eml"}); err != nil { + t.Fatalf("append: %v", err) + } + + // Client B (another pod) grabs the folder's X lock, standing in for an + // in-progress compaction holding it. + key := locks.MailboxKey(username, "INBOX") + ctx, cancel := context.WithTimeout(context.Background(), 15*time.Second) + defer cancel() + held, err := clientB.Lock(ctx, key, "podB/0/"+username, 30*time.Second) + if err != nil { + t.Fatalf("client B lock: %v", err) + } + if !clientB.HoldsResource(key) { + t.Fatal("client B does not hold the lock it just acquired") + } + + // A read-only op on idxA must block acquiring the same key that B holds. + done := make(chan error, 1) + started := make(chan struct{}) + go func() { + close(started) + _, e := idxA.GetMessages(folder.ID, mailbox.SeqSet{}) + done <- e + }() + <-started + + select { + case <-done: + t.Fatal("read-only GetMessages returned while another client held the folder lock — reload was NOT serialized (#647)") + case <-time.After(400 * time.Millisecond): + // Still blocked on the distributed lock — the fix is in effect. + } + + // Release B's lock; A's read must now complete promptly. + if err := clientB.Unlock(ctx, held.ID); err != nil { + t.Fatalf("client B unlock: %v", err) + } + select { + case e := <-done: + if e != nil { + t.Fatalf("GetMessages after lock release: %v", e) + } + case <-time.After(10 * time.Second): + t.Fatal("read-only GetMessages did not complete after the lock was released") + } +}