fix(snapshot): fill darwin maintenance batches

commit: ba65dc1faa81628913ac4d0d35bee1ce33ea65d7
author: brice · 3mo ago · 2026-07-22 06:04 UTC
parent: d8d4bf6
signature: GPG VerifiedCD88F9E564E0102646FB6077A1A07C7C7C3E349D · arch-gpg Verified for the committer email

3 files · +155 -12

M internal/adapters/snapshot/fd_count_darwin_test.go +12 -2


@@ -8,11 +8,21 @@ import (
8	8		)
9	9
10	10		// macOS exposes the calling process descriptor table through /dev/fd.
	11	+	// Readdirnames avoids os.ReadDir's fstatat calls for descriptors that can
	12	+	// disappear while the directory is being enumerated.
11	13		func openDescriptorCount(t *testing.T) int {
12	14		t.Helper()
13		-	entries, err := os.ReadDir("/dev/fd")
	15	+	dir, err := os.Open("/dev/fd")
14	16		if err != nil {
15	17		t.Fatal(err)
16	18		}
17		-	return len(entries)
	19	+	names, readErr := dir.Readdirnames(-1)
	20	+	closeErr := dir.Close()
	21	+	if readErr != nil {
	22	+	t.Fatal(readErr)
	23	+	}
	24	+	if closeErr != nil {
	25	+	t.Fatal(closeErr)
	26	+	}
	27	+	return len(names)
18	28		}

 )
 // macOS exposes the calling process descriptor table through /dev/fd.
+// Readdirnames avoids os.ReadDir's fstatat calls for descriptors that can
+// disappear while the directory is being enumerated.
 func openDescriptorCount(t *testing.T) int {
 	t.Helper()
-	entries, err := os.ReadDir("/dev/fd")
+	dir, err := os.Open("/dev/fd")
 	if err != nil {
 		t.Fatal(err)
 	}
-	return len(entries)
+	names, readErr := dir.Readdirnames(-1)
+	closeErr := dir.Close()
+	if readErr != nil {
+		t.Fatal(readErr)
+	}
+	if closeErr != nil {
+		t.Fatal(closeErr)
+	}
+	return len(names)
 }

M internal/adapters/snapshot/maintenance_cursor.go +17 -10


@@ -118,9 +118,16 @@ func maintenanceDirectoryError(operation string, err error) error {
118	118
119	119		// readMaintenanceDirent saves a resumable directory cursor. Linux d_off lets
120	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.
	121	+	// stores an entry count in the descriptor offset, so every returned buffer
	122	+	// must be drained before that descriptor-maintained count is saved.
123	123		func (r Repository) readMaintenanceDirent(dir string, limit int, cursor maintenanceCursor) (entries []maintenanceDirEntry, done bool, err error) {
	124	+	return r.readMaintenanceDirentWithDrain(dir, limit, cursor, drainMaintenanceDirentBatch(), directoryCookie)
	125	+	}
	126	+
	127	+	// readMaintenanceDirentWithDrain contains the shared traversal logic. The
	128	+	// explicit parameters permit platform-independent verification of Darwin's
	129	+	// buffer-draining semantics with deterministic fake directories.
	130	+	func (r Repository) readMaintenanceDirentWithDrain(dir string, limit int, cursor maintenanceCursor, drainBuffer bool, cookie func(maintenanceDirectory, *syscall.Dirent) (int64, error)) (entries []maintenanceDirEntry, done bool, err error) {
124	131		file, err := r.openMaintenanceDirectory(dir)
125	132		if err != nil {
126	133		return nil, false, maintenanceDirectoryError("open maintenance directory", err)

@@ -145,7 +152,7 @@ func (r Repository) readMaintenanceDirent(dir string, limit int, cursor mainte
145	152
146	153		entries = make([]maintenanceDirEntry, 0, limit)
147	154		buffer := make([]byte, maintenanceDirentBufferSize)
148		-	for len(entries) < limit \|\| drainMaintenanceDirentBatch() {
	155	+	for {
149	156		count, err := file.ReadDirent(buffer)
150	157		if err != nil {
151	158		return nil, false, maintenanceDirectoryError("read maintenance directory", err)

@@ -156,14 +163,14 @@ func (r Repository) readMaintenanceDirent(dir string, limit int, cursor mainte
156	163		if count == 0 {
157	164		return entries, true, nil
158	165		}
159		-	for data := buffer[:count]; len(data) != 0 && (len(entries) < limit \|\| drainMaintenanceDirentBatch()); {
	166	+	for data := buffer[:count]; len(data) != 0 && (len(entries) < limit \|\| drainBuffer); {
160	167		record, nameBytes, reclen, decodeErr := decodeMaintenanceDirent(data)
161	168		if decodeErr != nil {
162	169		return nil, false, decodeErr
163	170		}
164	171		data = data[reclen:]
165	172		// Advance past every raw record, including dot and disappeared entries.
166		-	offset, cookieErr := directoryCookie(file, record)
	173	+	offset, cookieErr := cookie(file, record)
167	174		if cookieErr != nil {
168	175		return nil, false, maintenanceDirectoryError("tell maintenance directory", cookieErr)
169	176		}

@@ -190,15 +197,15 @@ func (r Repository) readMaintenanceDirent(dir string, limit int, cursor mainte
190	197		}
191	198		entries = append(entries, maintenanceDirEntry{name: name, isDir: stat.Mode&syscall.S_IFMT == syscall.S_IFDIR})
192	199		}
193		-	if drainMaintenanceDirentBatch() {
194		-	if len(entries) > limit {
	200	+	if len(entries) >= limit {
	201	+	if drainBuffer && len(entries) > limit {
	202	+	// Darwin has already advanced the descriptor past this entire
	203	+	// buffer, so retain only this final buffer's excess.
195	204		cursor.pending = append(cursor.pending, entries[limit:]...)
196		-	entries = entries[:limit]
197	205		}
198		-	return entries, false, nil
	206	+	return entries[:limit], false, nil
199	207		}
200	208		}
201		-	return entries, false, nil
202	209		}
203	210
204	211		// decodeMaintenanceDirent copies a bounded raw record into aligned storage

 // readMaintenanceDirent saves a resumable directory cursor. Linux d_off lets
 // a large getdents batch stop at limit. Darwin's syscall.ReadDirent instead
-// stores an entry count in the descriptor offset, so its record-sized batch is
+// stores an entry count in the descriptor offset, so every returned buffer
-// always drained before that descriptor-maintained count is saved.
+// must be drained before that descriptor-maintained count is saved.
 func (r *Repository) readMaintenanceDirent(dir string, limit int, cursor *maintenanceCursor) (entries []maintenanceDirEntry, done bool, err error) {
+	return r.readMaintenanceDirentWithDrain(dir, limit, cursor, drainMaintenanceDirentBatch(), directoryCookie)
+}
+// readMaintenanceDirentWithDrain contains the shared traversal logic. The
+// explicit parameters permit platform-independent verification of Darwin's
+// buffer-draining semantics with deterministic fake directories.
+func (r *Repository) readMaintenanceDirentWithDrain(dir string, limit int, cursor *maintenanceCursor, drainBuffer bool, cookie func(maintenanceDirectory, *syscall.Dirent) (int64, error)) (entries []maintenanceDirEntry, done bool, err error) {
 	file, err := r.openMaintenanceDirectory(dir)
 	if err != nil {
 		return nil, false, maintenanceDirectoryError("open maintenance directory", err)
 	entries = make([]maintenanceDirEntry, 0, limit)
 	buffer := make([]byte, maintenanceDirentBufferSize)
-	for len(entries) < limit || drainMaintenanceDirentBatch() {
+	for {
 		count, err := file.ReadDirent(buffer)
 		if err != nil {
 			return nil, false, maintenanceDirectoryError("read maintenance directory", err)
 		if count == 0 {
 			return entries, true, nil
 		}
-		for data := buffer[:count]; len(data) != 0 && (len(entries) < limit || drainMaintenanceDirentBatch()); {
+		for data := buffer[:count]; len(data) != 0 && (len(entries) < limit || drainBuffer); {
 			record, nameBytes, reclen, decodeErr := decodeMaintenanceDirent(data)
 			if decodeErr != nil {
 				return nil, false, decodeErr
 			}
 			data = data[reclen:]
 			// Advance past every raw record, including dot and disappeared entries.
-			offset, cookieErr := directoryCookie(file, record)
+			offset, cookieErr := cookie(file, record)
 			if cookieErr != nil {
 				return nil, false, maintenanceDirectoryError("tell maintenance directory", cookieErr)
 			}
 			}
 			entries = append(entries, maintenanceDirEntry{name: name, isDir: stat.Mode&syscall.S_IFMT == syscall.S_IFDIR})
 		}
-		if drainMaintenanceDirentBatch() {
+		if len(entries) >= limit {
-			if len(entries) > limit {
+			if drainBuffer && len(entries) > limit {
+				// Darwin has already advanced the descriptor past this entire
+				// buffer, so retain only this final buffer's excess.
 				cursor.pending = append(cursor.pending, entries[limit:]...)
-				entries = entries[:limit]
 			}
-			return entries, false, nil
+			return entries[:limit], false, nil
 		}
 	}
-	return entries, false, nil
 }
 // decodeMaintenanceDirent copies a bounded raw record into aligned storage

M internal/adapters/snapshot/maintenance_test.go +126 -0


@@ -6,8 +6,10 @@ import (
6	6		"encoding/binary"
7	7		"errors"
8	8		"fmt"
	9	+	"io"
9	10		"os"
10	11		"path/filepath"
	12	+	"slices"
11	13		"strings"
12	14		"syscall"
13	15		"testing"

@@ -161,6 +163,130 @@ func (d fakeMaintenanceDirectory) ReadDirent(buffer []byte) (int, error) {
161	163		}
162	164		func (d fakeMaintenanceDirectory) Close() error { return d.closeErr }
163	165
	166	+	func TestReadMaintenanceDirentDrainsDarwinBatchesWithoutStarvation(t *testing.T) {
	167	+	dir := privateDir(t)
	168	+	if err := os.Mkdir(dir, 0o700); err != nil {
	169	+	t.Fatal(err)
	170	+	}
	171	+	batches := [][]string{
	172	+	{"one", "two"},
	173	+	{"three", "four", "five", "six"},
	174	+	}
	175	+	for _, batch := range batches {
	176	+	for _, name := range batch {
	177	+	if err := os.WriteFile(filepath.Join(dir, name), nil, 0o600); err != nil {
	178	+	t.Fatal(err)
	179	+	}
	180	+	}
	181	+	}
	182	+
	183	+	file := &fakeMultiBatchMaintenanceDirectory{batches: [][]byte{
	184	+	maintenanceDirentBatch(batches[0]...),
	185	+	maintenanceDirentBatch(batches[1]...),
	186	+	}}
	187	+	repo := NewRepository(dir)
	188	+	repo.hooks.openMaintenanceDirectory = func(string) (maintenanceDirectory, error) { return file, nil }
	189	+	cursor := &maintenanceCursor{}
	190	+
	191	+	entries, done, err := repo.readMaintenanceDirentWithDrain(dir, 4, cursor, true, maintenanceTestDirectoryCookie)
	192	+	if err != nil {
	193	+	t.Fatal(err)
	194	+	}
	195	+	if done {
	196	+	t.Fatal("readMaintenanceDirentWithDrain() done = true, want false before EOF")
	197	+	}
	198	+	if got, want := maintenanceEntryNames(entries), []string{"one", "two", "three", "four"}; !slices.Equal(got, want) {
	199	+	t.Fatalf("first entries = %v, want %v", got, want)
	200	+	}
	201	+	if got, want := maintenanceEntryNames(cursor.pending), []string{"five", "six"}; !slices.Equal(got, want) {
	202	+	t.Fatalf("pending entries = %v, want exactly final-buffer excess %v", got, want)
	203	+	}
	204	+	if got := file.reads; got != 2 {
	205	+	t.Fatalf("ReadDirent calls = %d, want 2 to fill the requested batch", got)
	206	+	}
	207	+
	208	+	id := maintenanceCursorID(dir, "test")
	209	+	repo.maintenanceCursors = map[string]*maintenanceCursor{id: cursor}
	210	+	entries, done, err = repo.readMaintenanceDir(dir, 4, "test")
	211	+	if err != nil {
	212	+	t.Fatal(err)
	213	+	}
	214	+	if done {
	215	+	t.Fatal("pending batch done = true, want false before EOF")
	216	+	}
	217	+	if got, want := maintenanceEntryNames(entries), []string{"five", "six"}; !slices.Equal(got, want) {
	218	+	t.Fatalf("pending entries = %v, want %v", got, want)
	219	+	}
	220	+	if got := file.reads; got != 2 {
	221	+	t.Fatalf("ReadDirent calls while delivering pending entries = %d, want 2", got)
	222	+	}
	223	+
	224	+	entries, done, err = repo.readMaintenanceDir(dir, 4, "test")
	225	+	if err != nil {
	226	+	t.Fatal(err)
	227	+	}
	228	+	if !done {
	229	+	t.Fatal("EOF batch done = false, want true")
	230	+	}
	231	+	if len(entries) != 0 {
	232	+	t.Fatalf("EOF entries = %v, want none", entries)
	233	+	}
	234	+	if _, ok := repo.maintenanceCursors[id]; ok {
	235	+	t.Fatal("EOF cursor retained after pending entries were delivered")
	236	+	}
	237	+	}
	238	+
	239	+	func maintenanceEntryNames(entries []maintenanceDirEntry) []string {
	240	+	names := make([]string, len(entries))
	241	+	for i, entry := range entries {
	242	+	names[i] = entry.name
	243	+	}
	244	+	return names
	245	+	}
	246	+
	247	+	func maintenanceDirentBatch(names ...string) []byte {
	248	+	nameOffset := int(unsafe.Offsetof(syscall.Dirent{}.Name))
	249	+	reclenOffset := unsafe.Offsetof(syscall.Dirent{}.Reclen)
	250	+	var data []byte
	251	+	for _, name := range names {
	252	+	record := make([]byte, nameOffset+len(name)+1)
	253	+	binary.NativeEndian.PutUint16(record[reclenOffset:], uint16(len(record)))
	254	+	copy(record[nameOffset:], name)
	255	+	data = append(data, record...)
	256	+	}
	257	+	return data
	258	+	}
	259	+
	260	+	type fakeMultiBatchMaintenanceDirectory struct {
	261	+	batches [][]byte
	262	+	index int
	263	+	reads int
	264	+	}
	265	+
	266	+	func (d *fakeMultiBatchMaintenanceDirectory) Seek(offset int64, whence int) (int64, error) {
	267	+	if whence == io.SeekCurrent {
	268	+	return int64(d.index), nil
	269	+	}
	270	+	d.index = int(offset)
	271	+	return offset, nil
	272	+	}
	273	+
	274	+	func (d *fakeMultiBatchMaintenanceDirectory) ReadDirent(buffer []byte) (int, error) {
	275	+	d.reads++
	276	+	if d.index == len(d.batches) {
	277	+	return 0, nil
	278	+	}
	279	+	batch := d.batches[d.index]
	280	+	d.index++
	281	+	return copy(buffer, batch), nil
	282	+	}
	283	+
	284	+	func (d *fakeMultiBatchMaintenanceDirectory) Close() error { return nil }
	285	+
	286	+	func maintenanceTestDirectoryCookie(file maintenanceDirectory, _ *syscall.Dirent) (int64, error) {
	287	+	return file.Seek(0, io.SeekCurrent)
	288	+	}
	289	+
164	290		func TestRepositoryDeleteQuarantinesCanonicalSessionBeforeCleanup(t *testing.T) {
165	291		repo := NewRepository(privateDir(t))
166	292		if err := repo.Publish(context.Background(), repositoryPublication(t, "named", 1, []byte("state"))); err != nil {


@@ -6,8 +6,10 @@ import (
6	"encoding/binary"	6		"encoding/binary"
7	"errors"	7		"errors"
8	"fmt"	8		"fmt"
		9	+	"io"
9	"os"	10		"os"
10	"path/filepath"	11		"path/filepath"
		12	+	"slices"
11	"strings"	13		"strings"
12	"syscall"	14		"syscall"
13	"testing"	15		"testing"

@@ -161,6 +163,130 @@ func (d fakeMaintenanceDirectory) ReadDirent(buffer []byte) (int, error) {
161	}	163		}
162	func (d fakeMaintenanceDirectory) Close() error { return d.closeErr }	164		func (d fakeMaintenanceDirectory) Close() error { return d.closeErr }
163		165
		166	+	func TestReadMaintenanceDirentDrainsDarwinBatchesWithoutStarvation(t *testing.T) {
		167	+	dir := privateDir(t)
		168	+	if err := os.Mkdir(dir, 0o700); err != nil {
		169	+	t.Fatal(err)
		170	+	}
		171	+	batches := [][]string{
		172	+	{"one", "two"},
		173	+	{"three", "four", "five", "six"},
		174	+	}
		175	+	for _, batch := range batches {
		176	+	for _, name := range batch {
		177	+	if err := os.WriteFile(filepath.Join(dir, name), nil, 0o600); err != nil {
		178	+	t.Fatal(err)
		179	+	}
		180	+	}
		181	+	}
		182	+
		183	+	file := &fakeMultiBatchMaintenanceDirectory{batches: [][]byte{
		184	+	maintenanceDirentBatch(batches[0]...),
		185	+	maintenanceDirentBatch(batches[1]...),
		186	+	}}
		187	+	repo := NewRepository(dir)
		188	+	repo.hooks.openMaintenanceDirectory = func(string) (maintenanceDirectory, error) { return file, nil }
		189	+	cursor := &maintenanceCursor{}
		190	+
		191	+	entries, done, err := repo.readMaintenanceDirentWithDrain(dir, 4, cursor, true, maintenanceTestDirectoryCookie)
		192	+	if err != nil {
		193	+	t.Fatal(err)
		194	+	}
		195	+	if done {
		196	+	t.Fatal("readMaintenanceDirentWithDrain() done = true, want false before EOF")
		197	+	}
		198	+	if got, want := maintenanceEntryNames(entries), []string{"one", "two", "three", "four"}; !slices.Equal(got, want) {
		199	+	t.Fatalf("first entries = %v, want %v", got, want)
		200	+	}
		201	+	if got, want := maintenanceEntryNames(cursor.pending), []string{"five", "six"}; !slices.Equal(got, want) {
		202	+	t.Fatalf("pending entries = %v, want exactly final-buffer excess %v", got, want)
		203	+	}
		204	+	if got := file.reads; got != 2 {
		205	+	t.Fatalf("ReadDirent calls = %d, want 2 to fill the requested batch", got)
		206	+	}
		207	+
		208	+	id := maintenanceCursorID(dir, "test")
		209	+	repo.maintenanceCursors = map[string]*maintenanceCursor{id: cursor}
		210	+	entries, done, err = repo.readMaintenanceDir(dir, 4, "test")
		211	+	if err != nil {
		212	+	t.Fatal(err)
		213	+	}
		214	+	if done {
		215	+	t.Fatal("pending batch done = true, want false before EOF")
		216	+	}
		217	+	if got, want := maintenanceEntryNames(entries), []string{"five", "six"}; !slices.Equal(got, want) {
		218	+	t.Fatalf("pending entries = %v, want %v", got, want)
		219	+	}
		220	+	if got := file.reads; got != 2 {
		221	+	t.Fatalf("ReadDirent calls while delivering pending entries = %d, want 2", got)
		222	+	}
		223	+
		224	+	entries, done, err = repo.readMaintenanceDir(dir, 4, "test")
		225	+	if err != nil {
		226	+	t.Fatal(err)
		227	+	}
		228	+	if !done {
		229	+	t.Fatal("EOF batch done = false, want true")
		230	+	}
		231	+	if len(entries) != 0 {
		232	+	t.Fatalf("EOF entries = %v, want none", entries)
		233	+	}
		234	+	if _, ok := repo.maintenanceCursors[id]; ok {
		235	+	t.Fatal("EOF cursor retained after pending entries were delivered")
		236	+	}
		237	+	}
		238	+
		239	+	func maintenanceEntryNames(entries []maintenanceDirEntry) []string {
		240	+	names := make([]string, len(entries))
		241	+	for i, entry := range entries {
		242	+	names[i] = entry.name
		243	+	}
		244	+	return names
		245	+	}
		246	+
		247	+	func maintenanceDirentBatch(names ...string) []byte {
		248	+	nameOffset := int(unsafe.Offsetof(syscall.Dirent{}.Name))
		249	+	reclenOffset := unsafe.Offsetof(syscall.Dirent{}.Reclen)
		250	+	var data []byte
		251	+	for _, name := range names {
		252	+	record := make([]byte, nameOffset+len(name)+1)
		253	+	binary.NativeEndian.PutUint16(record[reclenOffset:], uint16(len(record)))
		254	+	copy(record[nameOffset:], name)
		255	+	data = append(data, record...)
		256	+	}
		257	+	return data
		258	+	}
		259	+
		260	+	type fakeMultiBatchMaintenanceDirectory struct {
		261	+	batches [][]byte
		262	+	index int
		263	+	reads int
		264	+	}
		265	+
		266	+	func (d *fakeMultiBatchMaintenanceDirectory) Seek(offset int64, whence int) (int64, error) {
		267	+	if whence == io.SeekCurrent {
		268	+	return int64(d.index), nil
		269	+	}
		270	+	d.index = int(offset)
		271	+	return offset, nil
		272	+	}
		273	+
		274	+	func (d *fakeMultiBatchMaintenanceDirectory) ReadDirent(buffer []byte) (int, error) {
		275	+	d.reads++
		276	+	if d.index == len(d.batches) {
		277	+	return 0, nil
		278	+	}
		279	+	batch := d.batches[d.index]
		280	+	d.index++
		281	+	return copy(buffer, batch), nil
		282	+	}
		283	+
		284	+	func (d *fakeMultiBatchMaintenanceDirectory) Close() error { return nil }
		285	+
		286	+	func maintenanceTestDirectoryCookie(file maintenanceDirectory, _ *syscall.Dirent) (int64, error) {
		287	+	return file.Seek(0, io.SeekCurrent)
		288	+	}
		289	+
164	func TestRepositoryDeleteQuarantinesCanonicalSessionBeforeCleanup(t *testing.T) {	290		func TestRepositoryDeleteQuarantinesCanonicalSessionBeforeCleanup(t *testing.T) {
165	repo := NewRepository(privateDir(t))	291		repo := NewRepository(privateDir(t))
166	if err := repo.Publish(context.Background(), repositoryPublication(t, "named", 1, []byte("state"))); err != nil {	292		if err := repo.Publish(context.Background(), repositoryPublication(t, "named", 1, []byte("state"))); err != nil {