detached 692d04f

fix(snapshot): clear orphaned checkpoint heads

commit
692d04fcf5b1daec220ef53685133efbe4cbf224
author
brice · · 2026-07-26 12:56 UTC
parent
434d046
signature
GPG VerifiedCD88F9E564E0102646FB6077A1A07C7C7C3E349D · arch-gpg Verified for the committer email
12 files · +75 -182
Unified
UnifiedSplit
D internal/adapters/snapshot/fd_count_darwin_test.go +0 -28
@@ -1,28 +0,0 @@
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-// Readdirnames avoids os.ReadDir's fstatat calls for descriptors that can
12-// disappear while the directory is being enumerated.
13-func openDescriptorCount(t *testing.T) int {
14- t.Helper()
15- dir, err := os.Open("/dev/fd")
16- if err != nil {
17- t.Fatal(err)
18- }
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)
28-}
@@ -1,28 +0,0 @@
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-// Readdirnames avoids os.ReadDir's fstatat calls for descriptors that can
12-// disappear while the directory is being enumerated.
13-func openDescriptorCount(t *testing.T) int {
14- t.Helper()
15- dir, err := os.Open("/dev/fd")
16- if err != nil {
17- t.Fatal(err)
18- }
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)
28-}
D internal/adapters/snapshot/fd_count_linux_test.go +0 -19
@@ -1,19 +0,0 @@
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-}
@@ -1,19 +0,0 @@
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/load.go +1 -49
@@ -6,7 +6,6 @@ import (
66 "encoding/binary"
77 "errors"
88 "fmt"
9- "io"
109 "os"
1110
1211 "github.com/bnema/vev/internal/domain"
@@ -14,54 +13,7 @@ import (
1413 codec "github.com/bnema/vev/internal/usecase/snapshot"
1514 )
1615
17-// Directory scans are deliberately finite. Snapshot directories are attacker
18-// controlled state, so callers get a retryable error instead of retaining an
19-// arbitrary number of names or spending an arbitrary amount of work.
20-const (
21- directoryTraversalBatch = 64
22- maxDirectoryTraversalEntries = 4096
23-)
24-
25-var (
26- ErrDirectoryTraversalBudget = errors.New("snapshot directory traversal budget exceeded")
27- ErrInvalidHEAD = errors.New("invalid snapshot HEAD")
28-)
29-
30-// walkDirectory is the common finite directory traversal primitive for reads.
31-// File.ReadDir maintains a cursor on one descriptor and returns at most one
32-// small batch, so no full directory enumeration is retained in memory.
33-func (r *Repository) walkDirectory(ctx context.Context, dir string, budget *int, visit func(os.DirEntry) error) (err error) {
34- file, err := r.openDirectory(dir)
35- if err != nil {
36- return err
37- }
38- defer func() {
39- if closeErr := file.Close(); err == nil && closeErr != nil {
40- err = closeErr
41- }
42- }()
43- for {
44- if err := ctx.Err(); err != nil {
45- return err
46- }
47- entries, readErr := file.ReadDir(directoryTraversalBatch)
48- for _, entry := range entries {
49- if *budget == 0 {
50- return ErrDirectoryTraversalBudget
51- }
52- *budget--
53- if err := visit(entry); err != nil {
54- return err
55- }
56- }
57- if errors.Is(readErr, io.EOF) {
58- return nil
59- }
60- if readErr != nil {
61- return readErr
62- }
63- }
64-}
16+var ErrInvalidHEAD = errors.New("invalid snapshot HEAD")
6517
6618 func (r *Repository) LoadCheckpoint(ctx context.Context, id domain.IncarnationID, name string, ref ports.CheckpointRef) (ports.SnapshotGeneration, error) {
6719 if err := ctx.Err(); err != nil {
@@ -6,7 +6,6 @@ import (
6 "encoding/binary"6 "encoding/binary"
7 "errors"7 "errors"
8 "fmt"8 "fmt"
9- "io"
10 "os"9 "os"
11 10
12 "github.com/bnema/vev/internal/domain"11 "github.com/bnema/vev/internal/domain"
@@ -14,54 +13,7 @@ import (
14 codec "github.com/bnema/vev/internal/usecase/snapshot"13 codec "github.com/bnema/vev/internal/usecase/snapshot"
15 )14 )
16 15
17-// Directory scans are deliberately finite. Snapshot directories are attacker16+var ErrInvalidHEAD = errors.New("invalid snapshot HEAD")
18-// controlled state, so callers get a retryable error instead of retaining an
19-// arbitrary number of names or spending an arbitrary amount of work.
20-const (
21- directoryTraversalBatch = 64
22- maxDirectoryTraversalEntries = 4096
23-)
24-
25-var (
26- ErrDirectoryTraversalBudget = errors.New("snapshot directory traversal budget exceeded")
27- ErrInvalidHEAD = errors.New("invalid snapshot HEAD")
28-)
29-
30-// walkDirectory is the common finite directory traversal primitive for reads.
31-// File.ReadDir maintains a cursor on one descriptor and returns at most one
32-// small batch, so no full directory enumeration is retained in memory.
33-func (r *Repository) walkDirectory(ctx context.Context, dir string, budget *int, visit func(os.DirEntry) error) (err error) {
34- file, err := r.openDirectory(dir)
35- if err != nil {
36- return err
37- }
38- defer func() {
39- if closeErr := file.Close(); err == nil && closeErr != nil {
40- err = closeErr
41- }
42- }()
43- for {
44- if err := ctx.Err(); err != nil {
45- return err
46- }
47- entries, readErr := file.ReadDir(directoryTraversalBatch)
48- for _, entry := range entries {
49- if *budget == 0 {
50- return ErrDirectoryTraversalBudget
51- }
52- *budget--
53- if err := visit(entry); err != nil {
54- return err
55- }
56- }
57- if errors.Is(readErr, io.EOF) {
58- return nil
59- }
60- if readErr != nil {
61- return readErr
62- }
63- }
64-}
65 17
66 func (r *Repository) LoadCheckpoint(ctx context.Context, id domain.IncarnationID, name string, ref ports.CheckpointRef) (ports.SnapshotGeneration, error) {18 func (r *Repository) LoadCheckpoint(ctx context.Context, id domain.IncarnationID, name string, ref ports.CheckpointRef) (ports.SnapshotGeneration, error) {
67 if err := ctx.Err(); err != nil {19 if err := ctx.Err(); err != nil {
M internal/adapters/snapshot/maintenance_delete.go +0 -1
@@ -22,7 +22,6 @@ func (r *Repository) DeleteIncarnation(ctx context.Context, id domain.Incarnatio
2222 }
2323 lock := r.lockSession(key)
2424 defer r.unlockSession(lock)
25- r.invalidateStorageEpoch(key)
2625
2726 path := r.sessionPath(id)
2827 if err := os.RemoveAll(path); err != nil {
@@ -22,7 +22,6 @@ func (r *Repository) DeleteIncarnation(ctx context.Context, id domain.Incarnatio
22 }22 }
23 lock := r.lockSession(key)23 lock := r.lockSession(key)
24 defer r.unlockSession(lock)24 defer r.unlockSession(lock)
25- r.invalidateStorageEpoch(key)
26 25
27 path := r.sessionPath(id)26 path := r.sessionPath(id)
28 if err := os.RemoveAll(path); err != nil {27 if err := os.RemoveAll(path); err != nil {
M internal/adapters/snapshot/maintenance_gc.go +32 -0
@@ -116,9 +116,41 @@ func (r *Repository) pruneGenerations(ctx context.Context, id domain.Incarnation
116116 }
117117 r.log.Info("snapshot_garbage_collected", "incarnation", id.String(), "action", "remove-generations", "removed", removed)
118118 }
119+ if err := r.removeHeadForDiscardedGenerations(id, kept); err != nil {
120+ return err
121+ }
119122 return r.sweepUnreferencedObjects(ctx, id, references)
120123 }
121124
125+// removeHeadForDiscardedGenerations removes a repository HEAD only when its
126+// generation was removed by the retention policy. In particular, an
127+// incarnation with no committed checkpoint retains no generations, so its
128+// crash-published HEAD cannot make a retry conflict with an orphan.
129+func (r *Repository) removeHeadForDiscardedGenerations(id domain.IncarnationID, kept map[uint64]struct{}) error {
130+ removeHead := len(kept) == 0
131+ if !removeHead {
132+ generation, _, err := r.readHead(id)
133+ if errors.Is(err, os.ErrNotExist) {
134+ return nil
135+ }
136+ if err != nil {
137+ return fmt.Errorf("read snapshot HEAD for %s: %w", id.String(), safeFilesystemError(err))
138+ }
139+ _, retained := kept[generation]
140+ removeHead = !retained
141+ }
142+ if !removeHead {
143+ return nil
144+ }
145+ if err := r.remove(r.headPath(id)); err != nil && !errors.Is(err, os.ErrNotExist) {
146+ return fmt.Errorf("remove snapshot HEAD for %s: %w", id.String(), safeFilesystemError(err))
147+ }
148+ if err := r.syncDirectory(r.sessionPath(id)); err != nil {
149+ return fmt.Errorf("sync snapshot incarnation for %s: %w", id.String(), safeFilesystemError(err))
150+ }
151+ return nil
152+}
153+
122154 func (r *Repository) referencesForGenerations(id domain.IncarnationID, entries []os.DirEntry, keep map[uint64]struct{}, committed domain.CheckpointRef) (map[ports.SnapshotDigest]struct{}, error) {
123155 references := make(map[ports.SnapshotDigest]struct{})
124156 committedFound := committed.Generation == 0
@@ -116,9 +116,41 @@ func (r *Repository) pruneGenerations(ctx context.Context, id domain.Incarnation
116 }116 }
117 r.log.Info("snapshot_garbage_collected", "incarnation", id.String(), "action", "remove-generations", "removed", removed)117 r.log.Info("snapshot_garbage_collected", "incarnation", id.String(), "action", "remove-generations", "removed", removed)
118 }118 }
119+ if err := r.removeHeadForDiscardedGenerations(id, kept); err != nil {
120+ return err
121+ }
119 return r.sweepUnreferencedObjects(ctx, id, references)122 return r.sweepUnreferencedObjects(ctx, id, references)
120 }123 }
121 124
125+// removeHeadForDiscardedGenerations removes a repository HEAD only when its
126+// generation was removed by the retention policy. In particular, an
127+// incarnation with no committed checkpoint retains no generations, so its
128+// crash-published HEAD cannot make a retry conflict with an orphan.
129+func (r *Repository) removeHeadForDiscardedGenerations(id domain.IncarnationID, kept map[uint64]struct{}) error {
130+ removeHead := len(kept) == 0
131+ if !removeHead {
132+ generation, _, err := r.readHead(id)
133+ if errors.Is(err, os.ErrNotExist) {
134+ return nil
135+ }
136+ if err != nil {
137+ return fmt.Errorf("read snapshot HEAD for %s: %w", id.String(), safeFilesystemError(err))
138+ }
139+ _, retained := kept[generation]
140+ removeHead = !retained
141+ }
142+ if !removeHead {
143+ return nil
144+ }
145+ if err := r.remove(r.headPath(id)); err != nil && !errors.Is(err, os.ErrNotExist) {
146+ return fmt.Errorf("remove snapshot HEAD for %s: %w", id.String(), safeFilesystemError(err))
147+ }
148+ if err := r.syncDirectory(r.sessionPath(id)); err != nil {
149+ return fmt.Errorf("sync snapshot incarnation for %s: %w", id.String(), safeFilesystemError(err))
150+ }
151+ return nil
152+}
153+
122 func (r *Repository) referencesForGenerations(id domain.IncarnationID, entries []os.DirEntry, keep map[uint64]struct{}, committed domain.CheckpointRef) (map[ports.SnapshotDigest]struct{}, error) {154 func (r *Repository) referencesForGenerations(id domain.IncarnationID, entries []os.DirEntry, keep map[uint64]struct{}, committed domain.CheckpointRef) (map[ports.SnapshotDigest]struct{}, error) {
123 references := make(map[ports.SnapshotDigest]struct{})155 references := make(map[ports.SnapshotDigest]struct{})
124 committedFound := committed.Generation == 0156 committedFound := committed.Generation == 0
M internal/adapters/snapshot/maintenance_gc_test.go +24 -0
@@ -13,6 +13,30 @@ import (
1313 "github.com/stretchr/testify/require"
1414 )
1515
16+func TestCollectGarbageClearsUncommittedCheckpointHead(t *testing.T) {
17+ t.Parallel()
18+ repository := NewRepository(privateDir(t))
19+ publication := repositoryPublication(t, "work", 1, []byte("first checkpoint"))
20+ require.NoError(t, repository.Publish(context.Background(), publication))
21+
22+ // Repository publication completed, but the catalogue commit did not.
23+ // Startup GC knows the incarnation but has no committed checkpoint for it.
24+ keep := map[domain.IncarnationID]domain.CheckpointRef{publication.IncarnationID: {}}
25+ require.NoError(t, repository.CollectGarbage(context.Background(), keep))
26+ _, err := os.Stat(repository.headPath(publication.IncarnationID))
27+ require.ErrorIs(t, err, os.ErrNotExist)
28+ require.Empty(t, garbageCollectionGenerations(t, repository, publication.IncarnationID))
29+
30+ // Retrying starts from an empty repository, then the resulting checkpoint
31+ // can be committed to the catalogue and restored.
32+ require.NoError(t, repository.Publish(context.Background(), publication))
33+ committed := domain.CheckpointRef{Generation: publication.Generation, ManifestDigest: sha256.Sum256(publication.Manifest)}
34+ keep[publication.IncarnationID] = committed
35+ require.NoError(t, repository.CollectGarbage(context.Background(), keep))
36+ _, err = repository.LoadCheckpoint(context.Background(), publication.IncarnationID, publication.Name, committed)
37+ require.NoError(t, err)
38+}
39+
1640 func TestCollectGarbage(t *testing.T) {
1741 t.Parallel()
1842 tests := []struct {
@@ -13,6 +13,30 @@ import (
13 "github.com/stretchr/testify/require"13 "github.com/stretchr/testify/require"
14 )14 )
15 15
16+func TestCollectGarbageClearsUncommittedCheckpointHead(t *testing.T) {
17+ t.Parallel()
18+ repository := NewRepository(privateDir(t))
19+ publication := repositoryPublication(t, "work", 1, []byte("first checkpoint"))
20+ require.NoError(t, repository.Publish(context.Background(), publication))
21+
22+ // Repository publication completed, but the catalogue commit did not.
23+ // Startup GC knows the incarnation but has no committed checkpoint for it.
24+ keep := map[domain.IncarnationID]domain.CheckpointRef{publication.IncarnationID: {}}
25+ require.NoError(t, repository.CollectGarbage(context.Background(), keep))
26+ _, err := os.Stat(repository.headPath(publication.IncarnationID))
27+ require.ErrorIs(t, err, os.ErrNotExist)
28+ require.Empty(t, garbageCollectionGenerations(t, repository, publication.IncarnationID))
29+
30+ // Retrying starts from an empty repository, then the resulting checkpoint
31+ // can be committed to the catalogue and restored.
32+ require.NoError(t, repository.Publish(context.Background(), publication))
33+ committed := domain.CheckpointRef{Generation: publication.Generation, ManifestDigest: sha256.Sum256(publication.Manifest)}
34+ keep[publication.IncarnationID] = committed
35+ require.NoError(t, repository.CollectGarbage(context.Background(), keep))
36+ _, err = repository.LoadCheckpoint(context.Background(), publication.IncarnationID, publication.Name, committed)
37+ require.NoError(t, err)
38+}
39+
16 func TestCollectGarbage(t *testing.T) {40 func TestCollectGarbage(t *testing.T) {
17 t.Parallel()41 t.Parallel()
18 tests := []struct {42 tests := []struct {
M internal/adapters/snapshot/publish.go +2 -3
@@ -56,9 +56,8 @@ func (r *Repository) Publish(ctx context.Context, publication ports.SnapshotPubl
5656 }
5757
5858 // Parent validation and all other publication checks above happen before
59- // creating or invalidating repository state. A rejected child must leave the
60- // authoritative checkpoint and its storage byte-for-byte unchanged.
61- r.invalidateStorageEpoch(key)
59+ // creating repository state. A rejected child must leave the authoritative
60+ // checkpoint and its storage byte-for-byte unchanged.
6261 if err := r.ensureSession(publication.IncarnationID); err != nil {
6362 return err
6463 }
@@ -56,9 +56,8 @@ func (r *Repository) Publish(ctx context.Context, publication ports.SnapshotPubl
56 }56 }
57 57
58 // Parent validation and all other publication checks above happen before58 // Parent validation and all other publication checks above happen before
59- // creating or invalidating repository state. A rejected child must leave the59+ // creating repository state. A rejected child must leave the authoritative
60- // authoritative checkpoint and its storage byte-for-byte unchanged.60+ // checkpoint and its storage byte-for-byte unchanged.
61- r.invalidateStorageEpoch(key)
62 if err := r.ensureSession(publication.IncarnationID); err != nil {61 if err := r.ensureSession(publication.IncarnationID); err != nil {
63 return err62 return err
64 }63 }
M internal/adapters/snapshot/repository.go +9 -49
@@ -21,13 +21,11 @@ type Repository struct {
2121 dir string
2222 log *slog.Logger
2323
24- // sessionStateMu owns both maps. A caller retains a reference before it
25- // waits on a session mutex, so an idle entry can never be removed while a
26- // waiter still holds its old mutex. Epoch entries share that lifetime.
24+ // sessionStateMu owns locks. A caller retains a reference before it waits
25+ // on a session mutex, so an idle entry can never be removed while a waiter
26+ // still holds its old mutex.
2727 sessionStateMu sync.Mutex
2828 locks map[string]*sessionMutex
29- storageEpochs map[string]uint64
30- nextEpoch uint64
3129 }
3230
3331 var _ ports.SnapshotRepository = (*Repository)(nil)
@@ -40,33 +38,15 @@ func NewRepository(dir string, logs ...*slog.Logger) *Repository {
4038 log = logs[0]
4139 }
4240 return &Repository{
43- dir: dir,
44- log: log,
45- locks: make(map[string]*sessionMutex),
46- storageEpochs: make(map[string]uint64),
41+ dir: dir,
42+ log: log,
43+ locks: make(map[string]*sessionMutex),
4744 }
4845 }
4946
50-// invalidateStorageEpoch must be called with lockSession(key) held. The
51-// registry mutex makes accesses for different session keys race-free too.
52-func (r *Repository) invalidateStorageEpoch(key string) {
53- r.sessionStateMu.Lock()
54- r.nextEpoch++
55- r.storageEpochs[key] = r.nextEpoch
56- r.sessionStateMu.Unlock()
57-}
58-
59-// storageEpoch must be called with lockSession(key) held.
60-func (r *Repository) storageEpoch(key string) uint64 {
61- r.sessionStateMu.Lock()
62- epoch := r.storageEpochs[key]
63- r.sessionStateMu.Unlock()
64- return epoch
65-}
66-
67-// sessionMutex is retained by each waiter and each resumable maintenance
68-// state. That reference is what makes eviction safe: a new caller cannot lock
69-// a replacement mutex until all users of the prior mutex have left it.
47+// sessionMutex is retained by each waiter. That reference makes eviction
48+// safe: a new caller cannot lock a replacement mutex until all users of the
49+// prior mutex have left it.
7050 type sessionMutex struct {
7151 key string
7252 mu sync.Mutex
@@ -91,18 +71,6 @@ func (r *Repository) unlockSession(lock *sessionMutex) {
9171 r.releaseSessionReference(lock)
9272 }
9373
94-func (r *Repository) retainSessionState(key string) *sessionMutex {
95- r.sessionStateMu.Lock()
96- lock := r.locks[key]
97- // sessionMaintenanceState is called while the active session lock is held.
98- if lock == nil {
99- panic("snapshot: retain missing session lock")
100- }
101- lock.references++
102- r.sessionStateMu.Unlock()
103- return lock
104-}
105-
10674 func (r *Repository) releaseSessionReference(lock *sessionMutex) {
10775 r.sessionStateMu.Lock()
10876 lock.references--
@@ -116,14 +84,6 @@ func (r *Repository) releaseSessionReference(lock *sessionMutex) {
11684 panic("snapshot: session lock registry mismatch")
11785 }
11886 delete(r.locks, lock.key)
119- delete(r.storageEpochs, lock.key)
12087 }
12188 r.sessionStateMu.Unlock()
12289 }
123-
124-// sessionStateCounts is test evidence that idle per-session state is evicted.
125-func (r *Repository) sessionStateCounts() (locks, epochs int) {
126- r.sessionStateMu.Lock()
127- defer r.sessionStateMu.Unlock()
128- return len(r.locks), len(r.storageEpochs)
129-}
@@ -21,13 +21,11 @@ type Repository struct {
21 dir string21 dir string
22 log *slog.Logger22 log *slog.Logger
23 23
24- // sessionStateMu owns both maps. A caller retains a reference before it24+ // sessionStateMu owns locks. A caller retains a reference before it waits
25- // waits on a session mutex, so an idle entry can never be removed while a25+ // on a session mutex, so an idle entry can never be removed while a waiter
26- // waiter still holds its old mutex. Epoch entries share that lifetime.26+ // still holds its old mutex.
27 sessionStateMu sync.Mutex27 sessionStateMu sync.Mutex
28 locks map[string]*sessionMutex28 locks map[string]*sessionMutex
29- storageEpochs map[string]uint64
30- nextEpoch uint64
31 }29 }
32 30
33 var _ ports.SnapshotRepository = (*Repository)(nil)31 var _ ports.SnapshotRepository = (*Repository)(nil)
@@ -40,33 +38,15 @@ func NewRepository(dir string, logs ...*slog.Logger) *Repository {
40 log = logs[0]38 log = logs[0]
41 }39 }
42 return &Repository{40 return &Repository{
43- dir: dir,41+ dir: dir,
44- log: log,42+ log: log,
45- locks: make(map[string]*sessionMutex),43+ locks: make(map[string]*sessionMutex),
46- storageEpochs: make(map[string]uint64),
47 }44 }
48 }45 }
49 46
50-// invalidateStorageEpoch must be called with lockSession(key) held. The47+// sessionMutex is retained by each waiter. That reference makes eviction
51-// registry mutex makes accesses for different session keys race-free too.48+// safe: a new caller cannot lock a replacement mutex until all users of the
52-func (r *Repository) invalidateStorageEpoch(key string) {49+// prior mutex have left it.
53- r.sessionStateMu.Lock()
54- r.nextEpoch++
55- r.storageEpochs[key] = r.nextEpoch
56- r.sessionStateMu.Unlock()
57-}
58-
59-// storageEpoch must be called with lockSession(key) held.
60-func (r *Repository) storageEpoch(key string) uint64 {
61- r.sessionStateMu.Lock()
62- epoch := r.storageEpochs[key]
63- r.sessionStateMu.Unlock()
64- return epoch
65-}
66-
67-// sessionMutex is retained by each waiter and each resumable maintenance
68-// state. That reference is what makes eviction safe: a new caller cannot lock
69-// a replacement mutex until all users of the prior mutex have left it.
70 type sessionMutex struct {50 type sessionMutex struct {
71 key string51 key string
72 mu sync.Mutex52 mu sync.Mutex
@@ -91,18 +71,6 @@ func (r *Repository) unlockSession(lock *sessionMutex) {
91 r.releaseSessionReference(lock)71 r.releaseSessionReference(lock)
92 }72 }
93 73
94-func (r *Repository) retainSessionState(key string) *sessionMutex {
95- r.sessionStateMu.Lock()
96- lock := r.locks[key]
97- // sessionMaintenanceState is called while the active session lock is held.
98- if lock == nil {
99- panic("snapshot: retain missing session lock")
100- }
101- lock.references++
102- r.sessionStateMu.Unlock()
103- return lock
104-}
105-
106 func (r *Repository) releaseSessionReference(lock *sessionMutex) {74 func (r *Repository) releaseSessionReference(lock *sessionMutex) {
107 r.sessionStateMu.Lock()75 r.sessionStateMu.Lock()
108 lock.references--76 lock.references--
@@ -116,14 +84,6 @@ func (r *Repository) releaseSessionReference(lock *sessionMutex) {
116 panic("snapshot: session lock registry mismatch")84 panic("snapshot: session lock registry mismatch")
117 }85 }
118 delete(r.locks, lock.key)86 delete(r.locks, lock.key)
119- delete(r.storageEpochs, lock.key)
120 }87 }
121 r.sessionStateMu.Unlock()88 r.sessionStateMu.Unlock()
122 }89 }
123-
124-// sessionStateCounts is test evidence that idle per-session state is evicted.
125-func (r *Repository) sessionStateCounts() (locks, epochs int) {
126- r.sessionStateMu.Lock()
127- defer r.sessionStateMu.Unlock()
128- return len(r.locks), len(r.storageEpochs)
129-}
M internal/adapters/snapshot/repository_test.go +0 -11
@@ -35,17 +35,6 @@ func loadPublication(ctx context.Context, repo *Repository, publication ports.Sn
3535 })
3636 }
3737
38-func loadGenerationCheckpoint(ctx context.Context, repo *Repository, id domain.IncarnationID, name string, generation uint64) (ports.SnapshotGeneration, error) {
39- manifest, err := repo.readBounded(repo.manifestPath(id, generation))
40- if err != nil {
41- return ports.SnapshotGeneration{}, err
42- }
43- return repo.LoadCheckpoint(ctx, id, name, ports.CheckpointRef{
44- Generation: generation,
45- ManifestDigest: codec.ManifestDigest(manifest),
46- })
47-}
48-
4938 func repositoryPublicationAfter(t *testing.T, repo *Repository, name string, generation uint64, payload []byte) ports.SnapshotPublication {
5039 t.Helper()
5140 return publicationWithCurrentParent(t, repo, repositoryPublication(t, name, generation, payload))
@@ -35,17 +35,6 @@ func loadPublication(ctx context.Context, repo *Repository, publication ports.Sn
35 })35 })
36 }36 }
37 37
38-func loadGenerationCheckpoint(ctx context.Context, repo *Repository, id domain.IncarnationID, name string, generation uint64) (ports.SnapshotGeneration, error) {
39- manifest, err := repo.readBounded(repo.manifestPath(id, generation))
40- if err != nil {
41- return ports.SnapshotGeneration{}, err
42- }
43- return repo.LoadCheckpoint(ctx, id, name, ports.CheckpointRef{
44- Generation: generation,
45- ManifestDigest: codec.ManifestDigest(manifest),
46- })
47-}
48-
49 func repositoryPublicationAfter(t *testing.T, repo *Repository, name string, generation uint64, payload []byte) ports.SnapshotPublication {38 func repositoryPublicationAfter(t *testing.T, repo *Repository, name string, generation uint64, payload []byte) ports.SnapshotPublication {
50 t.Helper()39 t.Helper()
51 return publicationWithCurrentParent(t, repo, repositoryPublication(t, name, generation, payload))40 return publicationWithCurrentParent(t, repo, repositoryPublication(t, name, generation, payload))
M internal/adapters/snapshot/safe_path.go +0 -13
@@ -103,19 +103,6 @@ func (r *Repository) repositoryRelative(path string) (string, bool) {
103103 return rel, true
104104 }
105105
106-func (r *Repository) stat(path string) (fi os.FileInfo, err error) {
107- rel, ok := r.repositoryRelative(path)
108- if !ok {
109- return nil, fmt.Errorf("snapshot path outside repository")
110- }
111- root, err := r.openRoot()
112- if err != nil {
113- return nil, err
114- }
115- defer func() { joinCloseError(&err, "close snapshot root", r.closeRoot(root)) }()
116- return root.Lstat(rel)
117-}
118-
119106 // createTempAt creates an exclusively owned temporary file in an already
120107 // validated directory. os.CreateTemp(path) would resolve that path again.
121108 func (r *Repository) createTempAt(dir string) (temp *os.File, err error) {
@@ -103,19 +103,6 @@ func (r *Repository) repositoryRelative(path string) (string, bool) {
103 return rel, true103 return rel, true
104 }104 }
105 105
106-func (r *Repository) stat(path string) (fi os.FileInfo, err error) {
107- rel, ok := r.repositoryRelative(path)
108- if !ok {
109- return nil, fmt.Errorf("snapshot path outside repository")
110- }
111- root, err := r.openRoot()
112- if err != nil {
113- return nil, err
114- }
115- defer func() { joinCloseError(&err, "close snapshot root", r.closeRoot(root)) }()
116- return root.Lstat(rel)
117-}
118-
119 // createTempAt creates an exclusively owned temporary file in an already106 // createTempAt creates an exclusively owned temporary file in an already
120 // validated directory. os.CreateTemp(path) would resolve that path again.107 // validated directory. os.CreateTemp(path) would resolve that path again.
121 func (r *Repository) createTempAt(dir string) (temp *os.File, err error) {108 func (r *Repository) createTempAt(dir string) (temp *os.File, err error) {
M internal/usecase/daemon/daemon.go +2 -3
@@ -131,9 +131,8 @@ type Daemon struct {
131131 maintenanceWorkerCancel context.CancelFunc
132132 maintenanceWorkerDone chan struct{}
133133 // restoreWorkerDone is the restoration goroutine's ownership signal. Startup
134- // restoration repairs HEADs, promotes fallbacks, and replaces catalogue
135- // records, so it is a durable writer and is guarded by snapshotWorkerMu with
136- // the other two.
134+ // restoration may repair a repository HEAD, so it is a durable writer and is
135+ // guarded by snapshotWorkerMu with the other two.
137136 restoreWorkerDone chan struct{}
138137 snapsEnabled bool
139138 noticeStore ports.NoticeStore
@@ -131,9 +131,8 @@ type Daemon struct {
131 maintenanceWorkerCancel context.CancelFunc131 maintenanceWorkerCancel context.CancelFunc
132 maintenanceWorkerDone chan struct{}132 maintenanceWorkerDone chan struct{}
133 // restoreWorkerDone is the restoration goroutine's ownership signal. Startup133 // restoreWorkerDone is the restoration goroutine's ownership signal. Startup
134- // restoration repairs HEADs, promotes fallbacks, and replaces catalogue134+ // restoration may repair a repository HEAD, so it is a durable writer and is
135- // records, so it is a durable writer and is guarded by snapshotWorkerMu with135+ // guarded by snapshotWorkerMu with the other two.
136- // the other two.
137 restoreWorkerDone chan struct{}136 restoreWorkerDone chan struct{}
138 snapsEnabled bool137 snapsEnabled bool
139 noticeStore ports.NoticeStore138 noticeStore ports.NoticeStore
M internal/usecase/daemon/snapshot_restore.go +5 -6
@@ -37,11 +37,10 @@ func (d *Daemon) closeRestoreDone() {
3737 }
3838
3939 // startSnapshotRestoration launches catalogue-driven restoration as a durable
40-// writer. Restoration repairs HEADs, promotes fallbacks, and replaces catalogue
41-// records, so lifecycle ownership must outlive it exactly as it outlives the
42-// snapshot and maintenance workers. The completion channel is registered before
43-// the goroutine starts, so shutdown can never begin waiting while this writer
44-// is still unregistered.
40+// writer. Restoration may repair HEADs, so lifecycle ownership must outlive it
41+// exactly as it outlives the snapshot and maintenance workers. The completion
42+// channel is registered before the goroutine starts, so shutdown can never
43+// begin waiting while this writer is still unregistered.
4544 func (d *Daemon) startSnapshotRestoration() {
4645 if d == nil {
4746 return
@@ -314,7 +313,7 @@ func (d *Daemon) persistAndRegisterRestoredSession(ctx context.Context, sess *se
314313 }
315314 }
316315 // Restoration is catalogue-authorized. Missing or unreadable authority is a
317- // hard failure; migration creates catalogue records before daemon startup.
316+ // hard failure.
318317 if d.persistEnabled {
319318 record, ok, err := d.catalogueRecord(sess.name)
320319 if err != nil {
@@ -37,11 +37,10 @@ func (d *Daemon) closeRestoreDone() {
37 }37 }
38 38
39 // startSnapshotRestoration launches catalogue-driven restoration as a durable39 // startSnapshotRestoration launches catalogue-driven restoration as a durable
40-// writer. Restoration repairs HEADs, promotes fallbacks, and replaces catalogue40+// writer. Restoration may repair HEADs, so lifecycle ownership must outlive it
41-// records, so lifecycle ownership must outlive it exactly as it outlives the41+// exactly as it outlives the snapshot and maintenance workers. The completion
42-// snapshot and maintenance workers. The completion channel is registered before42+// channel is registered before the goroutine starts, so shutdown can never
43-// the goroutine starts, so shutdown can never begin waiting while this writer43+// begin waiting while this writer is still unregistered.
44-// is still unregistered.
45 func (d *Daemon) startSnapshotRestoration() {44 func (d *Daemon) startSnapshotRestoration() {
46 if d == nil {45 if d == nil {
47 return46 return
@@ -314,7 +313,7 @@ func (d *Daemon) persistAndRegisterRestoredSession(ctx context.Context, sess *se
314 }313 }
315 }314 }
316 // Restoration is catalogue-authorized. Missing or unreadable authority is a315 // Restoration is catalogue-authorized. Missing or unreadable authority is a
317- // hard failure; migration creates catalogue records before daemon startup.316+ // hard failure.
318 if d.persistEnabled {317 if d.persistEnabled {
319 record, ok, err := d.catalogueRecord(sess.name)318 record, ok, err := d.catalogueRecord(sess.name)
320 if err != nil {319 if err != nil {

Keyboard shortcuts

Single-key shortcuts are on. ctrl k and esc work whatever this says.

ctrl k :
go to a section, branch, tag, file or repository
?
this list
j k
move the row cursor down / up
enter
open the row under the cursor; on Files, a directory opens or closes in place
↑ ↓ ← → home end
in the file tree (focus on a row): previous / next row, close / open a directory or go to its parent / first entry, first / last row. They never open a file
l h
on Files: open a directory or a file / close a directory or go up to its parent, then to the parent page. Elsewhere: into the side panel (on a merge request, its files) / back to the content; there j k move between its controls
g g G
first / last row
/
filter the list; on Files, find a file anywhere in the repository (esc clears, enter opens the best match)
b
switch branch or tag
g s g f g c g b g m g i g r g a
tabs: summary, files, commits, branches & tags, merge requests, issues, releases, actions (when available)
g h
home
w
wide mode: use the whole width of the screen, or not (remembered)
y
copy a permalink pinned to the commit
d u x
diff: split / unified / collapse or expand all
e E
diff: the whole file in view / every file, with the lines the hunks hide
n
merge request: next note in the diff
esc
close a dialog, leave a field

Branches 12

Tags 11

↑↓ moveenter openesc close