fix(snapshot): resume maintenance on darwin
- commit
- fb7c7c49827e2f733846f11e0ae766520cf94994
- author
- brice · · 2026-07-22 05:48 UTC
- parent
- 91368cb
- signature
- GPG Verified
CD88F9E564E0102646FB6077A1A07C7C7C3E349D· arch-gpg Verified for the committer email
9 files · +144 -36
M internal/adapters/snapshot/atomic.go +7 -2
| @@ -314,8 +314,13 @@ func (r *Repository) remove(path string) (err error) { | |||
| 314 | 314 | defer func() { | |
| 315 | 315 | joinCloseError(&err, "close snapshot parent directory", r.closeDescriptor(fd, "close snapshot parent directory")) | |
| 316 | 316 | }() | |
| 317 | - | if err := unix.Unlinkat(fd, name, 0); !errors.Is(err, unix.EISDIR) { | |
| 318 | - | return err | |
| 317 | + | unlinkErr := unix.Unlinkat(fd, name, 0) | |
| 318 | + | if unlinkErr == nil { | |
| 319 | + | return nil | |
| 320 | + | } | |
| 321 | + | var stat unix.Stat_t | |
| 322 | + | if err := unix.Fstatat(fd, name, &stat, unix.AT_SYMLINK_NOFOLLOW); err != nil || !directoryUnlinkRetry(unlinkErr, uint32(stat.Mode)) { | |
| 323 | + | return unlinkErr | |
| 319 | 324 | } | |
| 320 | 325 | return unix.Unlinkat(fd, name, unix.AT_REMOVEDIR) | |
| 321 | 326 | } | |
| @@ -314,8 +314,13 @@ func (r *Repository) remove(path string) (err error) { | |||||
| 314 | defer func() { | 314 | defer func() { | ||
| 315 | joinCloseError(&err, "close snapshot parent directory", r.closeDescriptor(fd, "close snapshot parent directory")) | 315 | joinCloseError(&err, "close snapshot parent directory", r.closeDescriptor(fd, "close snapshot parent directory")) | ||
| 316 | }() | 316 | }() | ||
| 317 | - | if err := unix.Unlinkat(fd, name, 0); !errors.Is(err, unix.EISDIR) { | 317 | + | unlinkErr := unix.Unlinkat(fd, name, 0) |
| 318 | - | return err | 318 | + | if unlinkErr == nil { |
| 319 | + | return nil | |||
| 320 | + | } | |||
| 321 | + | var stat unix.Stat_t | |||
| 322 | + | if err := unix.Fstatat(fd, name, &stat, unix.AT_SYMLINK_NOFOLLOW); err != nil || !directoryUnlinkRetry(unlinkErr, uint32(stat.Mode)) { | |||
| 323 | + | return unlinkErr | |||
| 319 | } | 324 | } | ||
| 320 | return unix.Unlinkat(fd, name, unix.AT_REMOVEDIR) | 325 | return unix.Unlinkat(fd, name, unix.AT_REMOVEDIR) | ||
| 321 | } | 326 | } | ||
M internal/adapters/snapshot/dirent_cookie_darwin.go +23 -2
| @@ -2,6 +2,27 @@ | |||
| 2 | 2 | ||
| 3 | 3 | package snapshot | |
| 4 | 4 | ||
| 5 | - | import "syscall" | |
| 5 | + | import ( | |
| 6 | + | "errors" | |
| 7 | + | "io" | |
| 8 | + | "syscall" | |
| 9 | + | "unsafe" | |
| 10 | + | ) | |
| 6 | 11 | ||
| 7 | - | func directoryCookie(dirent *syscall.Dirent) int64 { return int64(dirent.Seekoff) } | |
| 12 | + | // syscall.ReadDirent on Darwin implements Getdirentries by storing the number | |
| 13 | + | // of returned entries in the descriptor seek offset. Dirent.Seekoff is not | |
| 14 | + | // that cursor. A max-sized record buffer lets each call be drained before the | |
| 15 | + | // descriptor-maintained count is saved. | |
| 16 | + | const maintenanceDirentBufferSize = int(unsafe.Sizeof(syscall.Dirent{})) | |
| 17 | + | ||
| 18 | + | func directoryCookie(file maintenanceDirectory, _ *syscall.Dirent) (int64, error) { | |
| 19 | + | return file.Seek(0, io.SeekCurrent) | |
| 20 | + | } | |
| 21 | + | ||
| 22 | + | func drainMaintenanceDirentBatch() bool { return true } | |
| 23 | + | ||
| 24 | + | // Darwin reports EPERM, rather than EISDIR, when unlinkat without | |
| 25 | + | // AT_REMOVEDIR targets a directory. The mode check preserves EPERM for files. | |
| 26 | + | func directoryUnlinkRetry(err error, mode uint32) bool { | |
| 27 | + | return errors.Is(err, syscall.EPERM) && mode&syscall.S_IFMT == syscall.S_IFDIR | |
| 28 | + | } | |
| @@ -2,6 +2,27 @@ | |||||
| 2 | 2 | ||||
| 3 | package snapshot | 3 | package snapshot | ||
| 4 | 4 | ||||
| 5 | - | import "syscall" | 5 | + | import ( |
| 6 | + | "errors" | |||
| 7 | + | "io" | |||
| 8 | + | "syscall" | |||
| 9 | + | "unsafe" | |||
| 10 | + | ) | |||
| 6 | 11 | ||||
| 7 | - | func directoryCookie(dirent *syscall.Dirent) int64 { return int64(dirent.Seekoff) } | 12 | + | // syscall.ReadDirent on Darwin implements Getdirentries by storing the number |
| 13 | + | // of returned entries in the descriptor seek offset. Dirent.Seekoff is not | |||
| 14 | + | // that cursor. A max-sized record buffer lets each call be drained before the | |||
| 15 | + | // descriptor-maintained count is saved. | |||
| 16 | + | const maintenanceDirentBufferSize = int(unsafe.Sizeof(syscall.Dirent{})) | |||
| 17 | + | ||||
| 18 | + | func directoryCookie(file maintenanceDirectory, _ *syscall.Dirent) (int64, error) { | |||
| 19 | + | return file.Seek(0, io.SeekCurrent) | |||
| 20 | + | } | |||
| 21 | + | ||||
| 22 | + | func drainMaintenanceDirentBatch() bool { return true } | |||
| 23 | + | ||||
| 24 | + | // Darwin reports EPERM, rather than EISDIR, when unlinkat without | |||
| 25 | + | // AT_REMOVEDIR targets a directory. The mode check preserves EPERM for files. | |||
| 26 | + | func directoryUnlinkRetry(err error, mode uint32) bool { | |||
| 27 | + | return errors.Is(err, syscall.EPERM) && mode&syscall.S_IFMT == syscall.S_IFDIR | |||
| 28 | + | } | |||
M internal/adapters/snapshot/dirent_cookie_darwin_test.go +18 -4
| @@ -7,9 +7,23 @@ import ( | |||
| 7 | 7 | "testing" | |
| 8 | 8 | ) | |
| 9 | 9 | ||
| 10 | - | func TestDirectoryCookieUsesDarwinSeekOffset(t *testing.T) { | |
| 11 | - | dirent := syscall.Dirent{Seekoff: 42} | |
| 12 | - | if got := directoryCookie(&dirent); got != 42 { | |
| 13 | - | t.Fatalf("directoryCookie() = %d, want 42", got) | |
| 10 | + | func TestDirectoryCookieUsesDescriptorEntryCount(t *testing.T) { | |
| 11 | + | file := fakeMaintenanceDirectory{seekOffset: 42} | |
| 12 | + | dirent := syscall.Dirent{Seekoff: 99} | |
| 13 | + | got, err := directoryCookie(file, &dirent) | |
| 14 | + | if err != nil { | |
| 15 | + | t.Fatal(err) | |
| 16 | + | } | |
| 17 | + | if got != 42 { | |
| 18 | + | t.Fatalf("directoryCookie() = %d, want descriptor entry count 42", got) | |
| 19 | + | } | |
| 20 | + | } | |
| 21 | + | ||
| 22 | + | func TestDirectoryUnlinkRetryRequiresDarwinDirectoryEPERM(t *testing.T) { | |
| 23 | + | if directoryUnlinkRetry(syscall.EPERM, syscall.S_IFREG) { | |
| 24 | + | t.Fatal("directoryUnlinkRetry(EPERM, regular file) = true, want false") | |
| 25 | + | } | |
| 26 | + | if !directoryUnlinkRetry(syscall.EPERM, syscall.S_IFDIR) { | |
| 27 | + | t.Fatal("directoryUnlinkRetry(EPERM, directory) = false, want true") | |
| 14 | 28 | } | |
| 15 | 29 | } | |
| @@ -7,9 +7,23 @@ import ( | |||||
| 7 | "testing" | 7 | "testing" | ||
| 8 | ) | 8 | ) | ||
| 9 | 9 | ||||
| 10 | - | func TestDirectoryCookieUsesDarwinSeekOffset(t *testing.T) { | 10 | + | func TestDirectoryCookieUsesDescriptorEntryCount(t *testing.T) { |
| 11 | - | dirent := syscall.Dirent{Seekoff: 42} | 11 | + | file := fakeMaintenanceDirectory{seekOffset: 42} |
| 12 | - | if got := directoryCookie(&dirent); got != 42 { | 12 | + | dirent := syscall.Dirent{Seekoff: 99} |
| 13 | - | t.Fatalf("directoryCookie() = %d, want 42", got) | 13 | + | got, err := directoryCookie(file, &dirent) |
| 14 | + | if err != nil { | |||
| 15 | + | t.Fatal(err) | |||
| 16 | + | } | |||
| 17 | + | if got != 42 { | |||
| 18 | + | t.Fatalf("directoryCookie() = %d, want descriptor entry count 42", got) | |||
| 19 | + | } | |||
| 20 | + | } | |||
| 21 | + | ||||
| 22 | + | func TestDirectoryUnlinkRetryRequiresDarwinDirectoryEPERM(t *testing.T) { | |||
| 23 | + | if directoryUnlinkRetry(syscall.EPERM, syscall.S_IFREG) { | |||
| 24 | + | t.Fatal("directoryUnlinkRetry(EPERM, regular file) = true, want false") | |||
| 25 | + | } | |||
| 26 | + | if !directoryUnlinkRetry(syscall.EPERM, syscall.S_IFDIR) { | |||
| 27 | + | t.Fatal("directoryUnlinkRetry(EPERM, directory) = false, want true") | |||
| 14 | } | 28 | } | ||
| 15 | } | 29 | } | ||
M internal/adapters/snapshot/dirent_cookie_linux.go +14 -2
| @@ -2,6 +2,18 @@ | |||
| 2 | 2 | ||
| 3 | 3 | package snapshot | |
| 4 | 4 | ||
| 5 | - | import "syscall" | |
| 5 | + | import ( | |
| 6 | + | "errors" | |
| 7 | + | "syscall" | |
| 8 | + | ) | |
| 6 | 9 | ||
| 7 | - | func directoryCookie(dirent *syscall.Dirent) int64 { return dirent.Off } | |
| 10 | + | const maintenanceDirentBufferSize = 8192 | |
| 11 | + | ||
| 12 | + | // directoryCookie is d_off, the resumable getdents(2) cookie for Linux. | |
| 13 | + | func directoryCookie(_ maintenanceDirectory, dirent *syscall.Dirent) (int64, error) { | |
| 14 | + | return dirent.Off, nil | |
| 15 | + | } | |
| 16 | + | ||
| 17 | + | func drainMaintenanceDirentBatch() bool { return false } | |
| 18 | + | ||
| 19 | + | func directoryUnlinkRetry(err error, _ uint32) bool { return errors.Is(err, syscall.EISDIR) } | |
| @@ -2,6 +2,18 @@ | |||||
| 2 | 2 | ||||
| 3 | package snapshot | 3 | package snapshot | ||
| 4 | 4 | ||||
| 5 | - | import "syscall" | 5 | + | import ( |
| 6 | + | "errors" | |||
| 7 | + | "syscall" | |||
| 8 | + | ) | |||
| 6 | 9 | ||||
| 7 | - | func directoryCookie(dirent *syscall.Dirent) int64 { return dirent.Off } | 10 | + | const maintenanceDirentBufferSize = 8192 |
| 11 | + | ||||
| 12 | + | // directoryCookie is d_off, the resumable getdents(2) cookie for Linux. | |||
| 13 | + | func directoryCookie(_ maintenanceDirectory, dirent *syscall.Dirent) (int64, error) { | |||
| 14 | + | return dirent.Off, nil | |||
| 15 | + | } | |||
| 16 | + | ||||
| 17 | + | func drainMaintenanceDirentBatch() bool { return false } | |||
| 18 | + | ||||
| 19 | + | func directoryUnlinkRetry(err error, _ uint32) bool { return errors.Is(err, syscall.EISDIR) } | |||
M internal/adapters/snapshot/dirent_cookie_linux_test.go +14 -1
| @@ -9,7 +9,20 @@ import ( | |||
| 9 | 9 | ||
| 10 | 10 | func TestDirectoryCookieUsesLinuxOffset(t *testing.T) { | |
| 11 | 11 | dirent := syscall.Dirent{Off: 42} | |
| 12 | - | if got := directoryCookie(&dirent); got != 42 { | |
| 12 | + | got, err := directoryCookie(fakeMaintenanceDirectory{}, &dirent) | |
| 13 | + | if err != nil { | |
| 14 | + | t.Fatal(err) | |
| 15 | + | } | |
| 16 | + | if got != 42 { | |
| 13 | 17 | t.Fatalf("directoryCookie() = %d, want 42", got) | |
| 14 | 18 | } | |
| 15 | 19 | } | |
| 20 | + | ||
| 21 | + | func TestDirectoryUnlinkRetryRequiresLinuxEISDIR(t *testing.T) { | |
| 22 | + | if directoryUnlinkRetry(syscall.EPERM, syscall.S_IFDIR) { | |
| 23 | + | t.Fatal("directoryUnlinkRetry(EPERM, directory) = true, want false") | |
| 24 | + | } | |
| 25 | + | if !directoryUnlinkRetry(syscall.EISDIR, syscall.S_IFDIR) { | |
| 26 | + | t.Fatal("directoryUnlinkRetry(EISDIR, directory) = false, want true") | |
| 27 | + | } | |
| 28 | + | } | |
| @@ -9,7 +9,20 @@ import ( | |||||
| 9 | 9 | ||||
| 10 | func TestDirectoryCookieUsesLinuxOffset(t *testing.T) { | 10 | func TestDirectoryCookieUsesLinuxOffset(t *testing.T) { | ||
| 11 | dirent := syscall.Dirent{Off: 42} | 11 | dirent := syscall.Dirent{Off: 42} | ||
| 12 | - | if got := directoryCookie(&dirent); got != 42 { | 12 | + | got, err := directoryCookie(fakeMaintenanceDirectory{}, &dirent) |
| 13 | + | if err != nil { | |||
| 14 | + | t.Fatal(err) | |||
| 15 | + | } | |||
| 16 | + | if got != 42 { | |||
| 13 | t.Fatalf("directoryCookie() = %d, want 42", got) | 17 | t.Fatalf("directoryCookie() = %d, want 42", got) | ||
| 14 | } | 18 | } | ||
| 15 | } | 19 | } | ||
| 20 | + | ||||
| 21 | + | func TestDirectoryUnlinkRetryRequiresLinuxEISDIR(t *testing.T) { | |||
| 22 | + | if directoryUnlinkRetry(syscall.EPERM, syscall.S_IFDIR) { | |||
| 23 | + | t.Fatal("directoryUnlinkRetry(EPERM, directory) = true, want false") | |||
| 24 | + | } | |||
| 25 | + | if !directoryUnlinkRetry(syscall.EISDIR, syscall.S_IFDIR) { | |||
| 26 | + | t.Fatal("directoryUnlinkRetry(EISDIR, directory) = false, want true") | |||
| 27 | + | } | |||
| 28 | + | } | |||
A internal/adapters/snapshot/fd_count_darwin_test.go +18 -0
| @@ -0,0 +1,18 @@ | |||
| 1 | + | //go:build darwin | |
| 2 | + | ||
| 3 | + | package snapshot | |
| 4 | + | ||
| 5 | + | import ( | |
| 6 | + | "os" | |
| 7 | + | "testing" | |
| 8 | + | ) | |
| 9 | + | ||
| 10 | + | // macOS exposes the calling process descriptor table through /dev/fd. | |
| 11 | + | func openDescriptorCount(t *testing.T) int { | |
| 12 | + | t.Helper() | |
| 13 | + | entries, err := os.ReadDir("/dev/fd") | |
| 14 | + | if err != nil { | |
| 15 | + | t.Fatal(err) | |
| 16 | + | } | |
| 17 | + | return len(entries) | |
| 18 | + | } | |
| @@ -0,0 +1,18 @@ | |||||
| 1 | + | //go:build darwin | |||
| 2 | + | ||||
| 3 | + | package snapshot | |||
| 4 | + | ||||
| 5 | + | import ( | |||
| 6 | + | "os" | |||
| 7 | + | "testing" | |||
| 8 | + | ) | |||
| 9 | + | ||||
| 10 | + | // macOS exposes the calling process descriptor table through /dev/fd. | |||
| 11 | + | func openDescriptorCount(t *testing.T) int { | |||
| 12 | + | t.Helper() | |||
| 13 | + | entries, err := os.ReadDir("/dev/fd") | |||
| 14 | + | if err != nil { | |||
| 15 | + | t.Fatal(err) | |||
| 16 | + | } | |||
| 17 | + | return len(entries) | |||
| 18 | + | } | |||
A internal/adapters/snapshot/fd_count_linux_test.go +19 -0
| @@ -0,0 +1,19 @@ | |||
| 1 | + | //go:build linux | |
| 2 | + | ||
| 3 | + | package snapshot | |
| 4 | + | ||
| 5 | + | import ( | |
| 6 | + | "os" | |
| 7 | + | "testing" | |
| 8 | + | ) | |
| 9 | + | ||
| 10 | + | // openDescriptorCount checks the actual Linux descriptor table, not a mocked | |
| 11 | + | // lifecycle signal, so the maintenance leak assertion remains strict. | |
| 12 | + | func openDescriptorCount(t *testing.T) int { | |
| 13 | + | t.Helper() | |
| 14 | + | entries, err := os.ReadDir("/proc/self/fd") | |
| 15 | + | if err != nil { | |
| 16 | + | t.Fatal(err) | |
| 17 | + | } | |
| 18 | + | return len(entries) | |
| 19 | + | } | |
| @@ -0,0 +1,19 @@ | |||||
| 1 | + | //go:build linux | |||
| 2 | + | ||||
| 3 | + | package snapshot | |||
| 4 | + | ||||
| 5 | + | import ( | |||
| 6 | + | "os" | |||
| 7 | + | "testing" | |||
| 8 | + | ) | |||
| 9 | + | ||||
| 10 | + | // openDescriptorCount checks the actual Linux descriptor table, not a mocked | |||
| 11 | + | // lifecycle signal, so the maintenance leak assertion remains strict. | |||
| 12 | + | func openDescriptorCount(t *testing.T) int { | |||
| 13 | + | t.Helper() | |||
| 14 | + | entries, err := os.ReadDir("/proc/self/fd") | |||
| 15 | + | if err != nil { | |||
| 16 | + | t.Fatal(err) | |||
| 17 | + | } | |||
| 18 | + | return len(entries) | |||
| 19 | + | } | |||
M internal/adapters/snapshot/maintenance_cursor.go +25 -11
| @@ -50,12 +50,14 @@ func (r *Repository) readMaintenanceDir(dir string, n int, purpose string) ([]ma | |||
| 50 | 50 | id := maintenanceCursorID(dir, purpose) | |
| 51 | 51 | cursor := r.maintenanceCursors[id] | |
| 52 | 52 | if cursor != nil && len(cursor.pending) != 0 { | |
| 53 | - | entries := cursor.pending | |
| 54 | - | cursor.pending = nil | |
| 55 | - | if cursor.done { | |
| 53 | + | count := min(n, len(cursor.pending)) | |
| 54 | + | entries := cursor.pending[:count] | |
| 55 | + | cursor.pending = cursor.pending[count:] | |
| 56 | + | done := cursor.done && len(cursor.pending) == 0 | |
| 57 | + | if done { | |
| 56 | 58 | delete(r.maintenanceCursors, id) | |
| 57 | 59 | } | |
| 58 | - | return entries, cursor.done, nil | |
| 60 | + | return entries, done, nil | |
| 59 | 61 | } | |
| 60 | 62 | if cursor == nil { | |
| 61 | 63 | if len(r.maintenanceCursors) >= maxMaintenanceCursors { | |
| @@ -114,9 +116,10 @@ func maintenanceDirectoryError(operation string, err error) error { | |||
| 114 | 116 | return fmt.Errorf("%s: %w", operation, safeFilesystemError(err)) | |
| 115 | 117 | } | |
| 116 | 118 | ||
| 117 | - | // readMaintenanceDirent saves the OS-provided seek cookie from each record. | |
| 118 | - | // The field differs by platform; directoryCookie abstracts that layout while | |
| 119 | - | // syscall.Seek resumes the descriptor at the saved directory position. | |
| 119 | + | // readMaintenanceDirent saves a resumable directory cursor. Linux d_off lets | |
| 120 | + | // a large getdents batch stop at limit. Darwin's syscall.ReadDirent instead | |
| 121 | + | // stores an entry count in the descriptor offset, so its record-sized batch is | |
| 122 | + | // always drained before that descriptor-maintained count is saved. | |
| 120 | 123 | func (r *Repository) readMaintenanceDirent(dir string, limit int, cursor *maintenanceCursor) (entries []maintenanceDirEntry, done bool, err error) { | |
| 121 | 124 | file, err := r.openMaintenanceDirectory(dir) | |
| 122 | 125 | if err != nil { | |
| @@ -141,8 +144,8 @@ func (r *Repository) readMaintenanceDirent(dir string, limit int, cursor *mainte | |||
| 141 | 144 | } | |
| 142 | 145 | ||
| 143 | 146 | entries = make([]maintenanceDirEntry, 0, limit) | |
| 144 | - | buffer := make([]byte, 8192) | |
| 145 | - | for len(entries) < limit { | |
| 147 | + | buffer := make([]byte, maintenanceDirentBufferSize) | |
| 148 | + | for len(entries) < limit || drainMaintenanceDirentBatch() { | |
| 146 | 149 | count, err := file.ReadDirent(buffer) | |
| 147 | 150 | if err != nil { | |
| 148 | 151 | return nil, false, maintenanceDirectoryError("read maintenance directory", err) | |
| @@ -150,7 +153,7 @@ func (r *Repository) readMaintenanceDirent(dir string, limit int, cursor *mainte | |||
| 150 | 153 | if count == 0 { | |
| 151 | 154 | return entries, true, nil | |
| 152 | 155 | } | |
| 153 | - | for data := buffer[:count]; len(data) != 0 && len(entries) < limit; { | |
| 156 | + | for data := buffer[:count]; len(data) != 0 && (len(entries) < limit || drainMaintenanceDirentBatch()); { | |
| 154 | 157 | if len(data) < int(unsafe.Offsetof(syscall.Dirent{}.Name)) { | |
| 155 | 158 | return nil, false, syscall.EIO | |
| 156 | 159 | } | |
| @@ -162,7 +165,11 @@ func (r *Repository) readMaintenanceDirent(dir string, limit int, cursor *mainte | |||
| 162 | 165 | } | |
| 163 | 166 | data = data[reclen:] | |
| 164 | 167 | // Advance past every raw record, including dot and disappeared entries. | |
| 165 | - | cursor.offset = directoryCookie(record) | |
| 168 | + | offset, cookieErr := directoryCookie(file, record) | |
| 169 | + | if cookieErr != nil { | |
| 170 | + | return nil, false, maintenanceDirectoryError("tell maintenance directory", cookieErr) | |
| 171 | + | } | |
| 172 | + | cursor.offset = offset | |
| 166 | 173 | nameBytes := unsafe.Slice((*byte)(unsafe.Pointer(&record.Name[0])), reclen-nameOffset) | |
| 167 | 174 | if end := strings.IndexByte(string(nameBytes), 0); end >= 0 { | |
| 168 | 175 | nameBytes = nameBytes[:end] | |
| @@ -186,6 +193,13 @@ func (r *Repository) readMaintenanceDirent(dir string, limit int, cursor *mainte | |||
| 186 | 193 | } | |
| 187 | 194 | entries = append(entries, maintenanceDirEntry{name: name, isDir: stat.Mode&syscall.S_IFMT == syscall.S_IFDIR}) | |
| 188 | 195 | } | |
| 196 | + | if drainMaintenanceDirentBatch() { | |
| 197 | + | if len(entries) > limit { | |
| 198 | + | cursor.pending = append(cursor.pending, entries[limit:]...) | |
| 199 | + | entries = entries[:limit] | |
| 200 | + | } | |
| 201 | + | return entries, false, nil | |
| 202 | + | } | |
| 189 | 203 | } | |
| 190 | 204 | return entries, false, nil | |
| 191 | 205 | } | |
| @@ -50,12 +50,14 @@ func (r *Repository) readMaintenanceDir(dir string, n int, purpose string) ([]ma | |||||
| 50 | id := maintenanceCursorID(dir, purpose) | 50 | id := maintenanceCursorID(dir, purpose) | ||
| 51 | cursor := r.maintenanceCursors[id] | 51 | cursor := r.maintenanceCursors[id] | ||
| 52 | if cursor != nil && len(cursor.pending) != 0 { | 52 | if cursor != nil && len(cursor.pending) != 0 { | ||
| 53 | - | entries := cursor.pending | 53 | + | count := min(n, len(cursor.pending)) |
| 54 | - | cursor.pending = nil | 54 | + | entries := cursor.pending[:count] |
| 55 | - | if cursor.done { | 55 | + | cursor.pending = cursor.pending[count:] |
| 56 | + | done := cursor.done && len(cursor.pending) == 0 | |||
| 57 | + | if done { | |||
| 56 | delete(r.maintenanceCursors, id) | 58 | delete(r.maintenanceCursors, id) | ||
| 57 | } | 59 | } | ||
| 58 | - | return entries, cursor.done, nil | 60 | + | return entries, done, nil |
| 59 | } | 61 | } | ||
| 60 | if cursor == nil { | 62 | if cursor == nil { | ||
| 61 | if len(r.maintenanceCursors) >= maxMaintenanceCursors { | 63 | if len(r.maintenanceCursors) >= maxMaintenanceCursors { | ||
| @@ -114,9 +116,10 @@ func maintenanceDirectoryError(operation string, err error) error { | |||||
| 114 | return fmt.Errorf("%s: %w", operation, safeFilesystemError(err)) | 116 | return fmt.Errorf("%s: %w", operation, safeFilesystemError(err)) | ||
| 115 | } | 117 | } | ||
| 116 | 118 | ||||
| 117 | - | // readMaintenanceDirent saves the OS-provided seek cookie from each record. | 119 | + | // readMaintenanceDirent saves a resumable directory cursor. Linux d_off lets |
| 118 | - | // The field differs by platform; directoryCookie abstracts that layout while | 120 | + | // a large getdents batch stop at limit. Darwin's syscall.ReadDirent instead |
| 119 | - | // syscall.Seek resumes the descriptor at the saved directory position. | 121 | + | // stores an entry count in the descriptor offset, so its record-sized batch is |
| 122 | + | // always drained before that descriptor-maintained count is saved. | |||
| 120 | func (r *Repository) readMaintenanceDirent(dir string, limit int, cursor *maintenanceCursor) (entries []maintenanceDirEntry, done bool, err error) { | 123 | func (r *Repository) readMaintenanceDirent(dir string, limit int, cursor *maintenanceCursor) (entries []maintenanceDirEntry, done bool, err error) { | ||
| 121 | file, err := r.openMaintenanceDirectory(dir) | 124 | file, err := r.openMaintenanceDirectory(dir) | ||
| 122 | if err != nil { | 125 | if err != nil { | ||
| @@ -141,8 +144,8 @@ func (r *Repository) readMaintenanceDirent(dir string, limit int, cursor *mainte | |||||
| 141 | } | 144 | } | ||
| 142 | 145 | ||||
| 143 | entries = make([]maintenanceDirEntry, 0, limit) | 146 | entries = make([]maintenanceDirEntry, 0, limit) | ||
| 144 | - | buffer := make([]byte, 8192) | 147 | + | buffer := make([]byte, maintenanceDirentBufferSize) |
| 145 | - | for len(entries) < limit { | 148 | + | for len(entries) < limit || drainMaintenanceDirentBatch() { |
| 146 | count, err := file.ReadDirent(buffer) | 149 | count, err := file.ReadDirent(buffer) | ||
| 147 | if err != nil { | 150 | if err != nil { | ||
| 148 | return nil, false, maintenanceDirectoryError("read maintenance directory", err) | 151 | return nil, false, maintenanceDirectoryError("read maintenance directory", err) | ||
| @@ -150,7 +153,7 @@ func (r *Repository) readMaintenanceDirent(dir string, limit int, cursor *mainte | |||||
| 150 | if count == 0 { | 153 | if count == 0 { | ||
| 151 | return entries, true, nil | 154 | return entries, true, nil | ||
| 152 | } | 155 | } | ||
| 153 | - | for data := buffer[:count]; len(data) != 0 && len(entries) < limit; { | 156 | + | for data := buffer[:count]; len(data) != 0 && (len(entries) < limit || drainMaintenanceDirentBatch()); { |
| 154 | if len(data) < int(unsafe.Offsetof(syscall.Dirent{}.Name)) { | 157 | if len(data) < int(unsafe.Offsetof(syscall.Dirent{}.Name)) { | ||
| 155 | return nil, false, syscall.EIO | 158 | return nil, false, syscall.EIO | ||
| 156 | } | 159 | } | ||
| @@ -162,7 +165,11 @@ func (r *Repository) readMaintenanceDirent(dir string, limit int, cursor *mainte | |||||
| 162 | } | 165 | } | ||
| 163 | data = data[reclen:] | 166 | data = data[reclen:] | ||
| 164 | // Advance past every raw record, including dot and disappeared entries. | 167 | // Advance past every raw record, including dot and disappeared entries. | ||
| 165 | - | cursor.offset = directoryCookie(record) | 168 | + | offset, cookieErr := directoryCookie(file, record) |
| 169 | + | if cookieErr != nil { | |||
| 170 | + | return nil, false, maintenanceDirectoryError("tell maintenance directory", cookieErr) | |||
| 171 | + | } | |||
| 172 | + | cursor.offset = offset | |||
| 166 | nameBytes := unsafe.Slice((*byte)(unsafe.Pointer(&record.Name[0])), reclen-nameOffset) | 173 | nameBytes := unsafe.Slice((*byte)(unsafe.Pointer(&record.Name[0])), reclen-nameOffset) | ||
| 167 | if end := strings.IndexByte(string(nameBytes), 0); end >= 0 { | 174 | if end := strings.IndexByte(string(nameBytes), 0); end >= 0 { | ||
| 168 | nameBytes = nameBytes[:end] | 175 | nameBytes = nameBytes[:end] | ||
| @@ -186,6 +193,13 @@ func (r *Repository) readMaintenanceDirent(dir string, limit int, cursor *mainte | |||||
| 186 | } | 193 | } | ||
| 187 | entries = append(entries, maintenanceDirEntry{name: name, isDir: stat.Mode&syscall.S_IFMT == syscall.S_IFDIR}) | 194 | entries = append(entries, maintenanceDirEntry{name: name, isDir: stat.Mode&syscall.S_IFMT == syscall.S_IFDIR}) | ||
| 188 | } | 195 | } | ||
| 196 | + | if drainMaintenanceDirentBatch() { | |||
| 197 | + | if len(entries) > limit { | |||
| 198 | + | cursor.pending = append(cursor.pending, entries[limit:]...) | |||
| 199 | + | entries = entries[:limit] | |||
| 200 | + | } | |||
| 201 | + | return entries, false, nil | |||
| 202 | + | } | |||
| 189 | } | 203 | } | ||
| 190 | return entries, false, nil | 204 | return entries, false, nil | ||
| 191 | } | 205 | } | ||
M internal/adapters/snapshot/maintenance_test.go +6 -14
| @@ -97,13 +97,14 @@ func TestReadMaintenanceDirentRejectsRecordShorterThanNameOffset(t *testing.T) { | |||
| 97 | 97 | } | |
| 98 | 98 | ||
| 99 | 99 | type fakeMaintenanceDirectory struct { | |
| 100 | - | data []byte | |
| 101 | - | seekErr error | |
| 102 | - | readErr error | |
| 103 | - | closeErr error | |
| 100 | + | data []byte | |
| 101 | + | seekOffset int64 | |
| 102 | + | seekErr error | |
| 103 | + | readErr error | |
| 104 | + | closeErr error | |
| 104 | 105 | } | |
| 105 | 106 | ||
| 106 | - | func (d fakeMaintenanceDirectory) Seek(int64, int) (int64, error) { return 0, d.seekErr } | |
| 107 | + | func (d fakeMaintenanceDirectory) Seek(int64, int) (int64, error) { return d.seekOffset, d.seekErr } | |
| 107 | 108 | func (d fakeMaintenanceDirectory) ReadDirent(buffer []byte) (int, error) { | |
| 108 | 109 | if d.readErr != nil { | |
| 109 | 110 | return 0, d.readErr | |
| @@ -423,15 +424,6 @@ func TestRepositoryMaintainClosesContinuationDirectories(t *testing.T) { | |||
| 423 | 424 | } | |
| 424 | 425 | } | |
| 425 | 426 | ||
| 426 | - | func openDescriptorCount(t *testing.T) int { | |
| 427 | - | t.Helper() | |
| 428 | - | entries, err := os.ReadDir("/proc/self/fd") | |
| 429 | - | if err != nil { | |
| 430 | - | t.Fatal(err) | |
| 431 | - | } | |
| 432 | - | return len(entries) | |
| 433 | - | } | |
| 434 | - | ||
| 435 | 427 | func TestRepositoryMaintainBoundsQueuedShardNames(t *testing.T) { | |
| 436 | 428 | repo := NewRepository(privateDir(t)) | |
| 437 | 429 | key := sessionKey("named") | |
| @@ -97,13 +97,14 @@ func TestReadMaintenanceDirentRejectsRecordShorterThanNameOffset(t *testing.T) { | |||||
| 97 | } | 97 | } | ||
| 98 | 98 | ||||
| 99 | type fakeMaintenanceDirectory struct { | 99 | type fakeMaintenanceDirectory struct { | ||
| 100 | - | data []byte | 100 | + | data []byte |
| 101 | - | seekErr error | 101 | + | seekOffset int64 |
| 102 | - | readErr error | 102 | + | seekErr error |
| 103 | - | closeErr error | 103 | + | readErr error |
| 104 | + | closeErr error | |||
| 104 | } | 105 | } | ||
| 105 | 106 | ||||
| 106 | - | func (d fakeMaintenanceDirectory) Seek(int64, int) (int64, error) { return 0, d.seekErr } | 107 | + | func (d fakeMaintenanceDirectory) Seek(int64, int) (int64, error) { return d.seekOffset, d.seekErr } |
| 107 | func (d fakeMaintenanceDirectory) ReadDirent(buffer []byte) (int, error) { | 108 | func (d fakeMaintenanceDirectory) ReadDirent(buffer []byte) (int, error) { | ||
| 108 | if d.readErr != nil { | 109 | if d.readErr != nil { | ||
| 109 | return 0, d.readErr | 110 | return 0, d.readErr | ||
| @@ -423,15 +424,6 @@ func TestRepositoryMaintainClosesContinuationDirectories(t *testing.T) { | |||||
| 423 | } | 424 | } | ||
| 424 | } | 425 | } | ||
| 425 | 426 | ||||
| 426 | - | func openDescriptorCount(t *testing.T) int { | |||
| 427 | - | t.Helper() | |||
| 428 | - | entries, err := os.ReadDir("/proc/self/fd") | |||
| 429 | - | if err != nil { | |||
| 430 | - | t.Fatal(err) | |||
| 431 | - | } | |||
| 432 | - | return len(entries) | |||
| 433 | - | } | |||
| 434 | - | ||||
| 435 | func TestRepositoryMaintainBoundsQueuedShardNames(t *testing.T) { | 427 | func TestRepositoryMaintainBoundsQueuedShardNames(t *testing.T) { | ||
| 436 | repo := NewRepository(privateDir(t)) | 428 | repo := NewRepository(privateDir(t)) | ||
| 437 | key := sessionKey("named") | 429 | key := sessionKey("named") | ||