test(snapshot): verify filesystem incremental costs

commit: 70f8dc32389cccaaa290a54e2916ba7d2964a924
author: brice · 3mo ago · 2026-07-22 02:41 UTC
parent: 0e3ca7c
signature: GPG VerifiedCD88F9E564E0102646FB6077A1A07C7C7C3E349D · arch-gpg Verified for the committer email

2 files · +109 -16

M internal/adapters/snapshot/publish_efficiency_test.go +75 -10


@@ -19,20 +19,31 @@ func TestRepositoryPublishSkipsUnchangedTenThousandObjectHistory(t *testing.T) {
19	19		first, second := largeIncrementalPublications(t, "named", historyObjects)
20	20		seedCompletePublication(t, repo, first)
21	21
22		-	var reads, hashes, copies int
23		-	repo.hooks.beforeObjectRead = func(string) { reads++ }
24		-	repo.hooks.beforeObjectHash = func([]byte) { hashes++ }
25		-	repo.hooks.beforeObjectCopy = func([]byte) { copies++ }
26		-
	22	+	accounting := installFilesystemPublishAccounting(repo)
27	23		if err := repo.Publish(context.Background(), second); err != nil {
28	24		t.Fatal(err)
29	25		}
30	26		// Only the new mutable tail and visible objects need work. The 10k sealed
31	27		// objects are retained history supplied again by the producer, not input
32		-	// that Publish needs to read, hash, or copy.
33		-	if reads != 2 \|\| hashes != 2 \|\| copies != 2 {
34		-	t.Fatalf("unchanged 10k history reads/hashes/copies = %d/%d/%d, want 2/2/2", reads, hashes, copies)
	28	+	// that Publish needs to read, hash, copy, or write.
	29	+	requireFilesystemPublishAccounting(t, accounting, filesystemPublishAccounting{
	30	+	objectReads: 2, objectHashes: 2, objectCopies: 2,
	31	+	tempCreates: 4, tempWrites: 4, fileSyncs: 4,
	32	+	objectWrites: 2, manifestWrites: 1, headWrites: 1,
	33	+	immutableInstalls: 3, mutableRenames: 1, tempRemoves: 3, directorySyncs: 4,
	34	+	})
	35	+
	36	+	// Replaying the now-current generation still parses the 10k references, but
	37	+	// must not allocate one object-sized buffer per retained history entry.
	38	+	allocations := testing.AllocsPerRun(3, func() {
	39	+	if err := repo.Publish(context.Background(), second); err != nil {
	40	+	panic(err)
	41	+	}
	42	+	})
	43	+	if allocations > 300 {
	44	+	t.Fatalf("unchanged 10k history Publish allocations = %.0f, want <= 300; retained objects were likely copied", allocations)
35	45		}
	46	+
36	47		got, err := repo.Load(context.Background(), second.Name)
37	48		if err != nil {
38	49		t.Fatal(err)

@@ -42,6 +53,60 @@ func TestRepositoryPublishSkipsUnchangedTenThousandObjectHistory(t *testing.T) {
42	53		}
43	54		}
44	55
	56	+	type filesystemPublishAccounting struct {
	57	+	objectReads, objectHashes, objectCopies int
	58	+	tempCreates, tempWrites, fileSyncs int
	59	+	objectWrites, manifestWrites, headWrites int
	60	+	immutableInstalls, mutableRenames, tempRemoves, directorySyncs int
	61	+	}
	62	+
	63	+	func installFilesystemPublishAccounting(repo Repository) filesystemPublishAccounting {
	64	+	accounting := &filesystemPublishAccounting{}
	65	+	repo.hooks.beforeObjectRead = func(string) { accounting.objectReads++ }
	66	+	repo.hooks.beforeObjectHash = func([]byte) { accounting.objectHashes++ }
	67	+	repo.hooks.beforeObjectCopy = func([]byte) { accounting.objectCopies++ }
	68	+	repo.hooks.createTemp = func(string) error { accounting.tempCreates++; return nil }
	69	+	repo.hooks.writeTemp = func(string) error { accounting.tempWrites++; return nil }
	70	+	repo.hooks.syncFile = func(string) error { accounting.fileSyncs++; return nil }
	71	+	repo.hooks.beforeBlobWrite = func(string) error { accounting.objectWrites++; return nil }
	72	+	repo.hooks.beforeManifestWrite = func(string) error { accounting.manifestWrites++; return nil }
	73	+	repo.hooks.beforeHeadWrite = func(string) error { accounting.headWrites++; return nil }
	74	+	repo.hooks.installImmutable = func(string) error { accounting.immutableInstalls++; return nil }
	75	+	repo.hooks.rename = func(string) error { accounting.mutableRenames++; return nil }
	76	+	repo.hooks.remove = func(string) error { accounting.tempRemoves++; return nil }
	77	+	repo.hooks.syncDirectory = func(string) error { accounting.directorySyncs++; return nil }
	78	+	return accounting
	79	+	}
	80	+
	81	+	func requireFilesystemPublishAccounting(t testing.T, got filesystemPublishAccounting, want filesystemPublishAccounting) {
	82	+	t.Helper()
	83	+	if *got != want {
	84	+	t.Fatalf("filesystem Publish accounting = %+v, want %+v", *got, want)
	85	+	}
	86	+	}
	87	+
	88	+	func BenchmarkRepositoryPublishUnchangedTenThousandObjectHistory(b *testing.B) {
	89	+	first, second := largeIncrementalPublications(b, "named", 10_000)
	90	+	repo := NewRepository(filepath.Join(b.TempDir(), "vev"))
	91	+	seedCompletePublication(b, repo, first)
	92	+	accounting := installFilesystemPublishAccounting(repo)
	93	+
	94	+	b.ReportAllocs()
	95	+	b.ResetTimer()
	96	+	for b.Loop() {
	97	+	if err := repo.Publish(context.Background(), second); err != nil {
	98	+	b.Fatal(err)
	99	+	}
	100	+	}
	101	+	b.StopTimer()
	102	+	// The benchmark deliberately replays the current generation so it can run
	103	+	// repeatedly without fixture writes. Its bounds catch a regression that
	104	+	// reads, hashes, or copies all 10k retained objects on the fast path.
	105	+	if accounting.objectReads > 2 \|\| accounting.objectHashes > 2 \|\| accounting.objectCopies > 2 {
	106	+	b.Fatalf("unchanged 10k history per-run object work = %d/%d/%d, want <= 2/2/2", accounting.objectReads, accounting.objectHashes, accounting.objectCopies)
	107	+	}
	108	+	}
	109	+
45	110		func TestRepositoryPublishVerifiesNecessaryExistingObjectOnce(t *testing.T) {
46	111		repo := NewRepository(privateDir(t))
47	112		first := repositoryPublication(t, "named", 1, []byte("first"))

@@ -72,7 +137,7 @@ func TestRepositoryPublishVerifiesNecessaryExistingObjectOnce(t *testing.T) {
72	137		}
73	138		}
74	139
75		-	func largeIncrementalPublications(t *testing.T, name string, count int) (ports.SnapshotPublication, ports.SnapshotPublication) {
	140	+	func largeIncrementalPublications(t testing.TB, name string, count int) (ports.SnapshotPublication, ports.SnapshotPublication) {
76	141		t.Helper()
77	142		sealed := make([]codec.ObjectRef, 0, count)
78	143		objects := make([]ports.SnapshotObject, 0, count+2)

@@ -103,7 +168,7 @@ func largeIncrementalPublications(t *testing.T, name string, count int) (ports.S
103	168		return makePublication(1, "tail-1", "visible-1"), makePublication(2, "tail-2", "visible-2")
104	169		}
105	170
106		-	func seedCompletePublication(t testing.T, repo Repository, publication ports.SnapshotPublication) {
	171	+	func seedCompletePublication(t testing.TB, repo *Repository, publication ports.SnapshotPublication) {
107	172		t.Helper()
108	173		key := sessionKey(publication.Name)
109	174		if err := repo.ensureSession(key); err != nil {


@@ -19,20 +19,31 @@ func TestRepositoryPublishSkipsUnchangedTenThousandObjectHistory(t *testing.T) {
19		first, second := largeIncrementalPublications(t, "named", historyObjects)	19		first, second := largeIncrementalPublications(t, "named", historyObjects)
20		seedCompletePublication(t, repo, first)	20		seedCompletePublication(t, repo, first)
21			21
22	-	var reads, hashes, copies int	22	+	accounting := installFilesystemPublishAccounting(repo)
23	-	repo.hooks.beforeObjectRead = func(string) { reads++ }
24	-	repo.hooks.beforeObjectHash = func([]byte) { hashes++ }
25	-	repo.hooks.beforeObjectCopy = func([]byte) { copies++ }
26	-
27		if err := repo.Publish(context.Background(), second); err != nil {	23		if err := repo.Publish(context.Background(), second); err != nil {
28		t.Fatal(err)	24		t.Fatal(err)
29		}	25		}
30		// Only the new mutable tail and visible objects need work. The 10k sealed	26		// Only the new mutable tail and visible objects need work. The 10k sealed
31		// objects are retained history supplied again by the producer, not input	27		// objects are retained history supplied again by the producer, not input
32	-	// that Publish needs to read, hash, or copy.	28	+	// that Publish needs to read, hash, copy, or write.
33	-	if reads != 2 \|\| hashes != 2 \|\| copies != 2 {	29	+	requireFilesystemPublishAccounting(t, accounting, filesystemPublishAccounting{
34	-	t.Fatalf("unchanged 10k history reads/hashes/copies = %d/%d/%d, want 2/2/2", reads, hashes, copies)	30	+	objectReads: 2, objectHashes: 2, objectCopies: 2,
			31	+	tempCreates: 4, tempWrites: 4, fileSyncs: 4,
			32	+	objectWrites: 2, manifestWrites: 1, headWrites: 1,
			33	+	immutableInstalls: 3, mutableRenames: 1, tempRemoves: 3, directorySyncs: 4,
			34	+	})
			35	+
			36	+	// Replaying the now-current generation still parses the 10k references, but
			37	+	// must not allocate one object-sized buffer per retained history entry.
			38	+	allocations := testing.AllocsPerRun(3, func() {
			39	+	if err := repo.Publish(context.Background(), second); err != nil {
			40	+	panic(err)
			41	+	}
			42	+	})
			43	+	if allocations > 300 {
			44	+	t.Fatalf("unchanged 10k history Publish allocations = %.0f, want <= 300; retained objects were likely copied", allocations)
35		}	45		}
			46	+
36		got, err := repo.Load(context.Background(), second.Name)	47		got, err := repo.Load(context.Background(), second.Name)
37		if err != nil {	48		if err != nil {
38		t.Fatal(err)	49		t.Fatal(err)

@@ -42,6 +53,60 @@ func TestRepositoryPublishSkipsUnchangedTenThousandObjectHistory(t *testing.T) {
42		}	53		}
43		}	54		}
44			55
			56	+	type filesystemPublishAccounting struct {
			57	+	objectReads, objectHashes, objectCopies int
			58	+	tempCreates, tempWrites, fileSyncs int
			59	+	objectWrites, manifestWrites, headWrites int
			60	+	immutableInstalls, mutableRenames, tempRemoves, directorySyncs int
			61	+	}
			62	+
			63	+	func installFilesystemPublishAccounting(repo Repository) filesystemPublishAccounting {
			64	+	accounting := &filesystemPublishAccounting{}
			65	+	repo.hooks.beforeObjectRead = func(string) { accounting.objectReads++ }
			66	+	repo.hooks.beforeObjectHash = func([]byte) { accounting.objectHashes++ }
			67	+	repo.hooks.beforeObjectCopy = func([]byte) { accounting.objectCopies++ }
			68	+	repo.hooks.createTemp = func(string) error { accounting.tempCreates++; return nil }
			69	+	repo.hooks.writeTemp = func(string) error { accounting.tempWrites++; return nil }
			70	+	repo.hooks.syncFile = func(string) error { accounting.fileSyncs++; return nil }
			71	+	repo.hooks.beforeBlobWrite = func(string) error { accounting.objectWrites++; return nil }
			72	+	repo.hooks.beforeManifestWrite = func(string) error { accounting.manifestWrites++; return nil }
			73	+	repo.hooks.beforeHeadWrite = func(string) error { accounting.headWrites++; return nil }
			74	+	repo.hooks.installImmutable = func(string) error { accounting.immutableInstalls++; return nil }
			75	+	repo.hooks.rename = func(string) error { accounting.mutableRenames++; return nil }
			76	+	repo.hooks.remove = func(string) error { accounting.tempRemoves++; return nil }
			77	+	repo.hooks.syncDirectory = func(string) error { accounting.directorySyncs++; return nil }
			78	+	return accounting
			79	+	}
			80	+
			81	+	func requireFilesystemPublishAccounting(t testing.T, got filesystemPublishAccounting, want filesystemPublishAccounting) {
			82	+	t.Helper()
			83	+	if *got != want {
			84	+	t.Fatalf("filesystem Publish accounting = %+v, want %+v", *got, want)
			85	+	}
			86	+	}
			87	+
			88	+	func BenchmarkRepositoryPublishUnchangedTenThousandObjectHistory(b *testing.B) {
			89	+	first, second := largeIncrementalPublications(b, "named", 10_000)
			90	+	repo := NewRepository(filepath.Join(b.TempDir(), "vev"))
			91	+	seedCompletePublication(b, repo, first)
			92	+	accounting := installFilesystemPublishAccounting(repo)
			93	+
			94	+	b.ReportAllocs()
			95	+	b.ResetTimer()
			96	+	for b.Loop() {
			97	+	if err := repo.Publish(context.Background(), second); err != nil {
			98	+	b.Fatal(err)
			99	+	}
			100	+	}
			101	+	b.StopTimer()
			102	+	// The benchmark deliberately replays the current generation so it can run
			103	+	// repeatedly without fixture writes. Its bounds catch a regression that
			104	+	// reads, hashes, or copies all 10k retained objects on the fast path.
			105	+	if accounting.objectReads > 2 \|\| accounting.objectHashes > 2 \|\| accounting.objectCopies > 2 {
			106	+	b.Fatalf("unchanged 10k history per-run object work = %d/%d/%d, want <= 2/2/2", accounting.objectReads, accounting.objectHashes, accounting.objectCopies)
			107	+	}
			108	+	}
			109	+
45		func TestRepositoryPublishVerifiesNecessaryExistingObjectOnce(t *testing.T) {	110		func TestRepositoryPublishVerifiesNecessaryExistingObjectOnce(t *testing.T) {
46		repo := NewRepository(privateDir(t))	111		repo := NewRepository(privateDir(t))
47		first := repositoryPublication(t, "named", 1, []byte("first"))	112		first := repositoryPublication(t, "named", 1, []byte("first"))

@@ -72,7 +137,7 @@ func TestRepositoryPublishVerifiesNecessaryExistingObjectOnce(t *testing.T) {
72		}	137		}
73		}	138		}
74			139
75	-	func largeIncrementalPublications(t *testing.T, name string, count int) (ports.SnapshotPublication, ports.SnapshotPublication) {	140	+	func largeIncrementalPublications(t testing.TB, name string, count int) (ports.SnapshotPublication, ports.SnapshotPublication) {
76		t.Helper()	141		t.Helper()
77		sealed := make([]codec.ObjectRef, 0, count)	142		sealed := make([]codec.ObjectRef, 0, count)
78		objects := make([]ports.SnapshotObject, 0, count+2)	143		objects := make([]ports.SnapshotObject, 0, count+2)

@@ -103,7 +168,7 @@ func largeIncrementalPublications(t *testing.T, name string, count int) (ports.S
103		return makePublication(1, "tail-1", "visible-1"), makePublication(2, "tail-2", "visible-2")	168		return makePublication(1, "tail-1", "visible-1"), makePublication(2, "tail-2", "visible-2")
104		}	169		}
105			170
106	-	func seedCompletePublication(t testing.T, repo Repository, publication ports.SnapshotPublication) {	171	+	func seedCompletePublication(t testing.TB, repo *Repository, publication ports.SnapshotPublication) {
107		t.Helper()	172		t.Helper()
108		key := sessionKey(publication.Name)	173		key := sessionKey(publication.Name)
109		if err := repo.ensureSession(key); err != nil {	174		if err := repo.ensureSession(key); err != nil {

M internal/usecase/daemon/snapshot_scheduler_test.go +34 -6


@@ -174,9 +174,8 @@ func TestRenameRollbackAfterOldDeleteLeavesOldCoordinatorStopped(t *testing.T) {
174	174		state.mu.Unlock()
175	175		repository.EXPECT().Tombstone(mock.Anything, "work").Return(nil).Once()
176	176		repository.EXPECT().Delete(mock.Anything, "work").Return(nil).Once()
177		-	repository.EXPECT().Publish(mock.Anything, mock.Anything).Run(func(context.Context, ports.SnapshotPublication) {
178		-	t.Fatal("rollback resurrected the deleted old snapshot identity")
179		-	}).Maybe()
	177	+	// No Publish expectation is installed: any publication is an unexpected mock
	178	+	// call. The saver handshake below joins the scheduler before this test ends.
180	179
181	180		pty, release := newBlockingPTY(t)
182	181		defer release()

@@ -198,14 +197,43 @@ func TestRenameRollbackAfterOldDeleteLeavesOldCoordinatorStopped(t *testing.T) {
198	197		sess.snapshotMu.Unlock()
199	198		require.True(t, quarantined)
200	199
	200	+	// Drain stale producer notifications, then use the timer-reset handshake to
	201	+	// prove the saver consumed this wake and completed its scheduling pass. Stop
	202	+	// and join it before asserting: no publisher can start after this point.
	203	+	for {
	204	+	select {
	205	+	case <-d.snapshotWake:
	206	+	default:
	207	+	goto wakeDrained
	208	+	}
	209	+	}
	210	+
	211	+	wakeDrained:
	212	+	clock := newDeterministicSnapshotClock(time.Unix(100, 0))
	213	+	d.clock = clock
201	214		schedulerCtx, cancelScheduler := context.WithCancel(context.Background())
202		-	defer cancelScheduler()
203		-	go d.snapshotRepositorySaver(schedulerCtx)
	215	+	schedulerDone := make(chan struct{})
	216	+	go func() {
	217	+	d.snapshotRepositorySaver(schedulerCtx)
	218	+	close(schedulerDone)
	219	+	}()
	220	+	timer := clock.timer()
	221	+	require.Equal(t, 24*time.Hour, <-timer.resets)
204	222		select {
205	223		case d.snapshotWake <- struct{}{}:
206	224		default:
	225	+	t.Fatal("snapshot wake unexpectedly remained full")
	226	+	}
	227	+	require.Equal(t, 24*time.Hour, <-timer.resets, "scheduler did not process the rollback wake")
	228	+	cancelScheduler()
	229	+	select {
	230	+	case <-schedulerDone:
	231	+	case <-time.After(testWaitTimeout):
	232	+	t.Fatal("snapshot scheduler did not stop")
207	233		}
208		-	time.Sleep(50 * time.Millisecond)
	234	+	// The deterministic scheduler clock has no independently firing deadline;
	235	+	// restore the daemon's normal test clock before its session cleanup runs.
	236	+	d.clock = stubClock{}
209	237		}
210	238
211	239		func TestForcedSnapshotSchedulesSuccessorAfterRoutineInFlight(t *testing.T) {


@@ -174,9 +174,8 @@ func TestRenameRollbackAfterOldDeleteLeavesOldCoordinatorStopped(t *testing.T) {
174		state.mu.Unlock()	174		state.mu.Unlock()
175		repository.EXPECT().Tombstone(mock.Anything, "work").Return(nil).Once()	175		repository.EXPECT().Tombstone(mock.Anything, "work").Return(nil).Once()
176		repository.EXPECT().Delete(mock.Anything, "work").Return(nil).Once()	176		repository.EXPECT().Delete(mock.Anything, "work").Return(nil).Once()
177	-	repository.EXPECT().Publish(mock.Anything, mock.Anything).Run(func(context.Context, ports.SnapshotPublication) {	177	+	// No Publish expectation is installed: any publication is an unexpected mock
178	-	t.Fatal("rollback resurrected the deleted old snapshot identity")	178	+	// call. The saver handshake below joins the scheduler before this test ends.
179	-	}).Maybe()
180			179
181		pty, release := newBlockingPTY(t)	180		pty, release := newBlockingPTY(t)
182		defer release()	181		defer release()

@@ -198,14 +197,43 @@ func TestRenameRollbackAfterOldDeleteLeavesOldCoordinatorStopped(t *testing.T) {
198		sess.snapshotMu.Unlock()	197		sess.snapshotMu.Unlock()
199		require.True(t, quarantined)	198		require.True(t, quarantined)
200			199
			200	+	// Drain stale producer notifications, then use the timer-reset handshake to
			201	+	// prove the saver consumed this wake and completed its scheduling pass. Stop
			202	+	// and join it before asserting: no publisher can start after this point.
			203	+	for {
			204	+	select {
			205	+	case <-d.snapshotWake:
			206	+	default:
			207	+	goto wakeDrained
			208	+	}
			209	+	}
			210	+
			211	+	wakeDrained:
			212	+	clock := newDeterministicSnapshotClock(time.Unix(100, 0))
			213	+	d.clock = clock
201		schedulerCtx, cancelScheduler := context.WithCancel(context.Background())	214		schedulerCtx, cancelScheduler := context.WithCancel(context.Background())
202	-	defer cancelScheduler()	215	+	schedulerDone := make(chan struct{})
203	-	go d.snapshotRepositorySaver(schedulerCtx)	216	+	go func() {
			217	+	d.snapshotRepositorySaver(schedulerCtx)
			218	+	close(schedulerDone)
			219	+	}()
			220	+	timer := clock.timer()
			221	+	require.Equal(t, 24*time.Hour, <-timer.resets)
204		select {	222		select {
205		case d.snapshotWake <- struct{}{}:	223		case d.snapshotWake <- struct{}{}:
206		default:	224		default:
			225	+	t.Fatal("snapshot wake unexpectedly remained full")
			226	+	}
			227	+	require.Equal(t, 24*time.Hour, <-timer.resets, "scheduler did not process the rollback wake")
			228	+	cancelScheduler()
			229	+	select {
			230	+	case <-schedulerDone:
			231	+	case <-time.After(testWaitTimeout):
			232	+	t.Fatal("snapshot scheduler did not stop")
207		}	233		}
208	-	time.Sleep(50 * time.Millisecond)	234	+	// The deterministic scheduler clock has no independently firing deadline;
			235	+	// restore the daemon's normal test clock before its session cleanup runs.
			236	+	d.clock = stubClock{}
209		}	237		}
210			238
211		func TestForcedSnapshotSchedulesSuccessorAfterRoutineInFlight(t *testing.T) {	239		func TestForcedSnapshotSchedulesSuccessorAfterRoutineInFlight(t *testing.T) {