feat(snapshot): quarantine deletes and collect stale blobs
- commit
- be1ed56542c93d6d14383356eeee8a632cb7dd8f
- author
- brice · · 2026-07-21 20:36 UTC
- parent
- 20864cb
- signature
- GPG Verified
CD88F9E564E0102646FB6077A1A07C7C7C3E349D· arch-gpg Verified for the committer email
5 files · +480 -90
M internal/adapters/snapshot/load.go +7 -0
| @@ -62,6 +62,13 @@ func (r *Repository) Load(ctx context.Context, name string) (ports.SnapshotGener | |||
| 62 | 62 | return ports.SnapshotGeneration{}, err | |
| 63 | 63 | } | |
| 64 | 64 | key := sessionKey(name) | |
| 65 | + | lock := r.sessionLock(key) | |
| 66 | + | lock.Lock() | |
| 67 | + | defer lock.Unlock() | |
| 68 | + | return r.load(ctx, name, key) | |
| 69 | + | } | |
| 70 | + | ||
| 71 | + | func (r *Repository) load(ctx context.Context, name, key string) (ports.SnapshotGeneration, error) { | |
| 65 | 72 | preferred, preferredDigest, err := r.readHead(key) | |
| 66 | 73 | if err != nil && !errors.Is(err, os.ErrNotExist) { | |
| 67 | 74 | preferred = 0 | |
| @@ -62,6 +62,13 @@ func (r *Repository) Load(ctx context.Context, name string) (ports.SnapshotGener | |||||
| 62 | return ports.SnapshotGeneration{}, err | 62 | return ports.SnapshotGeneration{}, err | ||
| 63 | } | 63 | } | ||
| 64 | key := sessionKey(name) | 64 | key := sessionKey(name) | ||
| 65 | + | lock := r.sessionLock(key) | |||
| 66 | + | lock.Lock() | |||
| 67 | + | defer lock.Unlock() | |||
| 68 | + | return r.load(ctx, name, key) | |||
| 69 | + | } | |||
| 70 | + | ||||
| 71 | + | func (r *Repository) load(ctx context.Context, name, key string) (ports.SnapshotGeneration, error) { | |||
| 65 | preferred, preferredDigest, err := r.readHead(key) | 72 | preferred, preferredDigest, err := r.readHead(key) | ||
| 66 | if err != nil && !errors.Is(err, os.ErrNotExist) { | 73 | if err != nil && !errors.Is(err, os.ErrNotExist) { | ||
| 67 | preferred = 0 | 74 | preferred = 0 | ||
A internal/adapters/snapshot/maintenance.go +301 -0
| @@ -0,0 +1,301 @@ | |||
| 1 | + | package snapshot | |
| 2 | + | ||
| 3 | + | import ( | |
| 4 | + | "context" | |
| 5 | + | "errors" | |
| 6 | + | "fmt" | |
| 7 | + | "os" | |
| 8 | + | "path/filepath" | |
| 9 | + | "strings" | |
| 10 | + | "time" | |
| 11 | + | ||
| 12 | + | "github.com/bnema/vev/internal/ports" | |
| 13 | + | codec "github.com/bnema/vev/internal/usecase/snapshot" | |
| 14 | + | ) | |
| 15 | + | ||
| 16 | + | const maintenanceBatch = 64 | |
| 17 | + | ||
| 18 | + | // Delete makes a session unavailable by durably moving it out of the canonical | |
| 19 | + | // namespace. Maintain reaps the private quarantine later; Delete never restores | |
| 20 | + | // a quarantined directory. | |
| 21 | + | func (r *Repository) Delete(ctx context.Context, name string) error { | |
| 22 | + | if err := ctx.Err(); err != nil { | |
| 23 | + | return err | |
| 24 | + | } | |
| 25 | + | key := sessionKey(name) | |
| 26 | + | lock := r.sessionLock(key) | |
| 27 | + | lock.Lock() | |
| 28 | + | defer lock.Unlock() | |
| 29 | + | ||
| 30 | + | canonical := r.sessionPath(key) | |
| 31 | + | if _, err := os.Lstat(canonical); errors.Is(err, os.ErrNotExist) { | |
| 32 | + | return nil | |
| 33 | + | } else if err != nil { | |
| 34 | + | return fmt.Errorf("stat snapshot session %q: %w", key, safeFilesystemError(err)) | |
| 35 | + | } | |
| 36 | + | quarantine := filepath.Join(filepath.Dir(canonical), ".deleting-"+key+"-"+fmt.Sprintf("%d", time.Now().UnixNano())) | |
| 37 | + | for attempt := 0; ; attempt++ { | |
| 38 | + | if _, err := os.Lstat(quarantine); errors.Is(err, os.ErrNotExist) { | |
| 39 | + | break | |
| 40 | + | } else if err != nil { | |
| 41 | + | return fmt.Errorf("stat deleting snapshot session %q: %w", key, safeFilesystemError(err)) | |
| 42 | + | } | |
| 43 | + | quarantine = filepath.Join(filepath.Dir(canonical), ".deleting-"+key+"-"+fmt.Sprintf("%d-%d", time.Now().UnixNano(), attempt)) | |
| 44 | + | } | |
| 45 | + | if err := r.rename(canonical, quarantine); err != nil { | |
| 46 | + | return fmt.Errorf("quarantine snapshot session %q: %w", key, safeFilesystemError(err)) | |
| 47 | + | } | |
| 48 | + | if err := r.syncDirectory(filepath.Dir(canonical)); err != nil { | |
| 49 | + | return fmt.Errorf("sync deleted snapshot session directory %q: %w", key, safeFilesystemError(err)) | |
| 50 | + | } | |
| 51 | + | return nil | |
| 52 | + | } | |
| 53 | + | ||
| 54 | + | // Maintain reaps a fixed number of stale entries. It serializes each session | |
| 55 | + | // with publication and loading so objects written by an in-flight publication | |
| 56 | + | // cannot be collected. | |
| 57 | + | func (r *Repository) Maintain(ctx context.Context) error { | |
| 58 | + | if err := ctx.Err(); err != nil { | |
| 59 | + | return err | |
| 60 | + | } | |
| 61 | + | budget := maintenanceBatch | |
| 62 | + | if err := r.removeTemps(ctx, r.dir, &budget); err != nil { | |
| 63 | + | return err | |
| 64 | + | } | |
| 65 | + | sessions := filepath.Join(r.dir, repositorySessionsDir) | |
| 66 | + | entries, err := os.ReadDir(sessions) | |
| 67 | + | if errors.Is(err, os.ErrNotExist) { | |
| 68 | + | return nil | |
| 69 | + | } | |
| 70 | + | if err != nil { | |
| 71 | + | return fmt.Errorf("read snapshot maintenance sessions: %w", safeFilesystemError(err)) | |
| 72 | + | } | |
| 73 | + | for _, entry := range entries { | |
| 74 | + | if budget == 0 { | |
| 75 | + | return nil | |
| 76 | + | } | |
| 77 | + | if err := ctx.Err(); err != nil { | |
| 78 | + | return err | |
| 79 | + | } | |
| 80 | + | path := filepath.Join(sessions, entry.Name()) | |
| 81 | + | if isQuarantine(entry.Name()) { | |
| 82 | + | if err := r.removeTree(path); err != nil && !errors.Is(err, os.ErrNotExist) { | |
| 83 | + | return fmt.Errorf("remove snapshot quarantine %q: %w", maintenancePath(r.dir, path), safeFilesystemError(err)) | |
| 84 | + | } | |
| 85 | + | if err := r.syncDirectory(sessions); err != nil { | |
| 86 | + | return fmt.Errorf("sync snapshot maintenance directory for %q: %w", maintenancePath(r.dir, path), safeFilesystemError(err)) | |
| 87 | + | } | |
| 88 | + | budget-- | |
| 89 | + | continue | |
| 90 | + | } | |
| 91 | + | if !entry.IsDir() || !canonicalSessionKey(entry.Name()) { | |
| 92 | + | continue | |
| 93 | + | } | |
| 94 | + | lock := r.sessionLock(entry.Name()) | |
| 95 | + | lock.Lock() | |
| 96 | + | err := r.maintainSession(ctx, entry.Name(), &budget) | |
| 97 | + | lock.Unlock() | |
| 98 | + | if err != nil { | |
| 99 | + | return err | |
| 100 | + | } | |
| 101 | + | } | |
| 102 | + | return nil | |
| 103 | + | } | |
| 104 | + | ||
| 105 | + | func isQuarantine(name string) bool { return strings.HasPrefix(name, ".deleting-") } | |
| 106 | + | ||
| 107 | + | func (r *Repository) maintainSession(ctx context.Context, key string, budget *int) error { | |
| 108 | + | if err := r.removeTemps(ctx, filepath.Join(r.sessionPath(key), repositoryGenerations), budget); err != nil { | |
| 109 | + | return err | |
| 110 | + | } | |
| 111 | + | objectRoot := filepath.Join(r.sessionPath(key), repositoryObjectsDir) | |
| 112 | + | shards, err := os.ReadDir(objectRoot) | |
| 113 | + | if err != nil && !errors.Is(err, os.ErrNotExist) { | |
| 114 | + | return fmt.Errorf("read snapshot object shards: %w", safeFilesystemError(err)) | |
| 115 | + | } | |
| 116 | + | for _, shard := range shards { | |
| 117 | + | if *budget == 0 { | |
| 118 | + | return nil | |
| 119 | + | } | |
| 120 | + | if shard.IsDir() { | |
| 121 | + | if err := r.removeTemps(ctx, filepath.Join(objectRoot, shard.Name()), budget); err != nil { | |
| 122 | + | return err | |
| 123 | + | } | |
| 124 | + | } | |
| 125 | + | } | |
| 126 | + | ||
| 127 | + | generations, err := r.generationNumbers(key) | |
| 128 | + | if err != nil { | |
| 129 | + | return fmt.Errorf("read snapshot generations: %w", safeFilesystemError(err)) | |
| 130 | + | } | |
| 131 | + | type manifestState struct { | |
| 132 | + | generation uint64 | |
| 133 | + | refs map[ports.SnapshotDigest]codec.ObjectRef | |
| 134 | + | complete bool | |
| 135 | + | } | |
| 136 | + | states := make([]manifestState, 0, len(generations)) | |
| 137 | + | complete := 0 | |
| 138 | + | for _, generation := range generations { | |
| 139 | + | if err := ctx.Err(); err != nil { | |
| 140 | + | return err | |
| 141 | + | } | |
| 142 | + | state := manifestState{generation: generation} | |
| 143 | + | data, err := readBounded(r.manifestPath(key, generation)) | |
| 144 | + | if err == nil { | |
| 145 | + | manifest, decodeErr := codec.UnmarshalManifest(data) | |
| 146 | + | if decodeErr == nil && manifest.Generation == generation && sessionKey(manifest.Name) == key { | |
| 147 | + | state.refs = manifestRefs(manifest) | |
| 148 | + | if state.refs != nil && withinGenerationBudget(len(data), state.refs) { | |
| 149 | + | _, loadErr := r.loadGeneration(ctx, manifest.Name, key, generation) | |
| 150 | + | state.complete = loadErr == nil | |
| 151 | + | } | |
| 152 | + | } | |
| 153 | + | } | |
| 154 | + | states = append(states, state) | |
| 155 | + | } | |
| 156 | + | ||
| 157 | + | retained := make(map[uint64]bool, 2) | |
| 158 | + | for _, state := range states { // generationNumbers is newest first. | |
| 159 | + | if state.complete && complete < 2 { | |
| 160 | + | retained[state.generation] = true | |
| 161 | + | complete++ | |
| 162 | + | } | |
| 163 | + | } | |
| 164 | + | referenced := make(map[ports.SnapshotDigest]struct{}) | |
| 165 | + | for _, state := range states { | |
| 166 | + | keep := retained[state.generation] || !state.complete | |
| 167 | + | if !keep && *budget > 0 { | |
| 168 | + | path := r.manifestPath(key, state.generation) | |
| 169 | + | if err := r.remove(path); err != nil && !errors.Is(err, os.ErrNotExist) { | |
| 170 | + | return fmt.Errorf("remove snapshot manifest %q: %w", maintenancePath(r.dir, path), safeFilesystemError(err)) | |
| 171 | + | } | |
| 172 | + | if err := r.syncDirectory(filepath.Dir(path)); err != nil { | |
| 173 | + | return fmt.Errorf("sync snapshot maintenance directory for %q: %w", maintenancePath(r.dir, path), safeFilesystemError(err)) | |
| 174 | + | } | |
| 175 | + | *budget-- | |
| 176 | + | continue | |
| 177 | + | } | |
| 178 | + | for digest := range state.refs { // incomplete manifests protect in-flight objects. | |
| 179 | + | referenced[digest] = struct{}{} | |
| 180 | + | } | |
| 181 | + | } | |
| 182 | + | return r.removeUnreferencedObjects(ctx, key, referenced, budget) | |
| 183 | + | } | |
| 184 | + | ||
| 185 | + | func (r *Repository) removeUnreferencedObjects(ctx context.Context, key string, referenced map[ports.SnapshotDigest]struct{}, budget *int) error { | |
| 186 | + | root := filepath.Join(r.sessionPath(key), repositoryObjectsDir) | |
| 187 | + | shards, err := os.ReadDir(root) | |
| 188 | + | if errors.Is(err, os.ErrNotExist) { | |
| 189 | + | return nil | |
| 190 | + | } | |
| 191 | + | if err != nil { | |
| 192 | + | return fmt.Errorf("read snapshot objects: %w", safeFilesystemError(err)) | |
| 193 | + | } | |
| 194 | + | for _, shard := range shards { | |
| 195 | + | if *budget == 0 { | |
| 196 | + | return nil | |
| 197 | + | } | |
| 198 | + | if !shard.IsDir() { | |
| 199 | + | continue | |
| 200 | + | } | |
| 201 | + | dir := filepath.Join(root, shard.Name()) | |
| 202 | + | objects, err := os.ReadDir(dir) | |
| 203 | + | if err != nil { | |
| 204 | + | return fmt.Errorf("read snapshot object shard: %w", safeFilesystemError(err)) | |
| 205 | + | } | |
| 206 | + | for _, object := range objects { | |
| 207 | + | if *budget == 0 { | |
| 208 | + | return nil | |
| 209 | + | } | |
| 210 | + | if err := ctx.Err(); err != nil { | |
| 211 | + | return err | |
| 212 | + | } | |
| 213 | + | digest, ok := parseObjectDigest(object.Name()) | |
| 214 | + | if !ok || object.IsDir() { | |
| 215 | + | continue | |
| 216 | + | } | |
| 217 | + | if _, used := referenced[digest]; used { | |
| 218 | + | continue | |
| 219 | + | } | |
| 220 | + | path := filepath.Join(dir, object.Name()) | |
| 221 | + | if err := r.remove(path); err != nil && !errors.Is(err, os.ErrNotExist) { | |
| 222 | + | return fmt.Errorf("remove snapshot object %q: %w", maintenancePath(r.dir, path), safeFilesystemError(err)) | |
| 223 | + | } | |
| 224 | + | if err := r.syncDirectory(dir); err != nil { | |
| 225 | + | return fmt.Errorf("sync snapshot maintenance directory for %q: %w", maintenancePath(r.dir, path), safeFilesystemError(err)) | |
| 226 | + | } | |
| 227 | + | *budget-- | |
| 228 | + | } | |
| 229 | + | } | |
| 230 | + | return nil | |
| 231 | + | } | |
| 232 | + | ||
| 233 | + | func (r *Repository) removeTemps(ctx context.Context, dir string, budget *int) error { | |
| 234 | + | if *budget == 0 { | |
| 235 | + | return nil | |
| 236 | + | } | |
| 237 | + | entries, err := os.ReadDir(dir) | |
| 238 | + | if errors.Is(err, os.ErrNotExist) { | |
| 239 | + | return nil | |
| 240 | + | } | |
| 241 | + | if err != nil { | |
| 242 | + | return fmt.Errorf("read snapshot maintenance directory: %w", safeFilesystemError(err)) | |
| 243 | + | } | |
| 244 | + | for _, entry := range entries { | |
| 245 | + | if *budget == 0 { | |
| 246 | + | return nil | |
| 247 | + | } | |
| 248 | + | if err := ctx.Err(); err != nil { | |
| 249 | + | return err | |
| 250 | + | } | |
| 251 | + | if entry.IsDir() || !strings.HasPrefix(entry.Name(), ".tmp-") { | |
| 252 | + | continue | |
| 253 | + | } | |
| 254 | + | path := filepath.Join(dir, entry.Name()) | |
| 255 | + | if err := r.remove(path); err != nil && !errors.Is(err, os.ErrNotExist) { | |
| 256 | + | return fmt.Errorf("remove snapshot maintenance file %q: %w", maintenancePath(r.dir, path), safeFilesystemError(err)) | |
| 257 | + | } | |
| 258 | + | if err := r.syncDirectory(dir); err != nil { | |
| 259 | + | return fmt.Errorf("sync snapshot maintenance directory for %q: %w", maintenancePath(r.dir, path), safeFilesystemError(err)) | |
| 260 | + | } | |
| 261 | + | *budget-- | |
| 262 | + | } | |
| 263 | + | return nil | |
| 264 | + | } | |
| 265 | + | ||
| 266 | + | func (r *Repository) removeTree(path string) error { | |
| 267 | + | if r.hooks.remove != nil { | |
| 268 | + | if err := r.hooks.remove(path); err != nil { | |
| 269 | + | return err | |
| 270 | + | } | |
| 271 | + | } | |
| 272 | + | return os.RemoveAll(path) | |
| 273 | + | } | |
| 274 | + | ||
| 275 | + | // maintenancePath avoids exposing the repository root while retaining a | |
| 276 | + | // path-safe, actionable location in errors. | |
| 277 | + | func maintenancePath(root, path string) string { | |
| 278 | + | rel, err := filepath.Rel(root, path) | |
| 279 | + | if err != nil || rel == "." || strings.HasPrefix(rel, ".."+string(filepath.Separator)) || rel == ".." { | |
| 280 | + | return filepath.Base(path) | |
| 281 | + | } | |
| 282 | + | return rel | |
| 283 | + | } | |
| 284 | + | ||
| 285 | + | type safePathError struct { | |
| 286 | + | op string | |
| 287 | + | cause error | |
| 288 | + | } | |
| 289 | + | ||
| 290 | + | func (e safePathError) Error() string { return e.op + ": " + e.cause.Error() } | |
| 291 | + | func (e safePathError) Unwrap() error { return e.cause } | |
| 292 | + | ||
| 293 | + | // safeFilesystemError preserves error matching without including a filesystem | |
| 294 | + | // path from terminal-facing error text. | |
| 295 | + | func safeFilesystemError(err error) error { | |
| 296 | + | var pathError *os.PathError | |
| 297 | + | if errors.As(err, &pathError) { | |
| 298 | + | return safePathError{op: pathError.Op, cause: pathError.Err} | |
| 299 | + | } | |
| 300 | + | return err | |
| 301 | + | } | |
| @@ -0,0 +1,301 @@ | |||||
| 1 | + | package snapshot | |||
| 2 | + | ||||
| 3 | + | import ( | |||
| 4 | + | "context" | |||
| 5 | + | "errors" | |||
| 6 | + | "fmt" | |||
| 7 | + | "os" | |||
| 8 | + | "path/filepath" | |||
| 9 | + | "strings" | |||
| 10 | + | "time" | |||
| 11 | + | ||||
| 12 | + | "github.com/bnema/vev/internal/ports" | |||
| 13 | + | codec "github.com/bnema/vev/internal/usecase/snapshot" | |||
| 14 | + | ) | |||
| 15 | + | ||||
| 16 | + | const maintenanceBatch = 64 | |||
| 17 | + | ||||
| 18 | + | // Delete makes a session unavailable by durably moving it out of the canonical | |||
| 19 | + | // namespace. Maintain reaps the private quarantine later; Delete never restores | |||
| 20 | + | // a quarantined directory. | |||
| 21 | + | func (r *Repository) Delete(ctx context.Context, name string) error { | |||
| 22 | + | if err := ctx.Err(); err != nil { | |||
| 23 | + | return err | |||
| 24 | + | } | |||
| 25 | + | key := sessionKey(name) | |||
| 26 | + | lock := r.sessionLock(key) | |||
| 27 | + | lock.Lock() | |||
| 28 | + | defer lock.Unlock() | |||
| 29 | + | ||||
| 30 | + | canonical := r.sessionPath(key) | |||
| 31 | + | if _, err := os.Lstat(canonical); errors.Is(err, os.ErrNotExist) { | |||
| 32 | + | return nil | |||
| 33 | + | } else if err != nil { | |||
| 34 | + | return fmt.Errorf("stat snapshot session %q: %w", key, safeFilesystemError(err)) | |||
| 35 | + | } | |||
| 36 | + | quarantine := filepath.Join(filepath.Dir(canonical), ".deleting-"+key+"-"+fmt.Sprintf("%d", time.Now().UnixNano())) | |||
| 37 | + | for attempt := 0; ; attempt++ { | |||
| 38 | + | if _, err := os.Lstat(quarantine); errors.Is(err, os.ErrNotExist) { | |||
| 39 | + | break | |||
| 40 | + | } else if err != nil { | |||
| 41 | + | return fmt.Errorf("stat deleting snapshot session %q: %w", key, safeFilesystemError(err)) | |||
| 42 | + | } | |||
| 43 | + | quarantine = filepath.Join(filepath.Dir(canonical), ".deleting-"+key+"-"+fmt.Sprintf("%d-%d", time.Now().UnixNano(), attempt)) | |||
| 44 | + | } | |||
| 45 | + | if err := r.rename(canonical, quarantine); err != nil { | |||
| 46 | + | return fmt.Errorf("quarantine snapshot session %q: %w", key, safeFilesystemError(err)) | |||
| 47 | + | } | |||
| 48 | + | if err := r.syncDirectory(filepath.Dir(canonical)); err != nil { | |||
| 49 | + | return fmt.Errorf("sync deleted snapshot session directory %q: %w", key, safeFilesystemError(err)) | |||
| 50 | + | } | |||
| 51 | + | return nil | |||
| 52 | + | } | |||
| 53 | + | ||||
| 54 | + | // Maintain reaps a fixed number of stale entries. It serializes each session | |||
| 55 | + | // with publication and loading so objects written by an in-flight publication | |||
| 56 | + | // cannot be collected. | |||
| 57 | + | func (r *Repository) Maintain(ctx context.Context) error { | |||
| 58 | + | if err := ctx.Err(); err != nil { | |||
| 59 | + | return err | |||
| 60 | + | } | |||
| 61 | + | budget := maintenanceBatch | |||
| 62 | + | if err := r.removeTemps(ctx, r.dir, &budget); err != nil { | |||
| 63 | + | return err | |||
| 64 | + | } | |||
| 65 | + | sessions := filepath.Join(r.dir, repositorySessionsDir) | |||
| 66 | + | entries, err := os.ReadDir(sessions) | |||
| 67 | + | if errors.Is(err, os.ErrNotExist) { | |||
| 68 | + | return nil | |||
| 69 | + | } | |||
| 70 | + | if err != nil { | |||
| 71 | + | return fmt.Errorf("read snapshot maintenance sessions: %w", safeFilesystemError(err)) | |||
| 72 | + | } | |||
| 73 | + | for _, entry := range entries { | |||
| 74 | + | if budget == 0 { | |||
| 75 | + | return nil | |||
| 76 | + | } | |||
| 77 | + | if err := ctx.Err(); err != nil { | |||
| 78 | + | return err | |||
| 79 | + | } | |||
| 80 | + | path := filepath.Join(sessions, entry.Name()) | |||
| 81 | + | if isQuarantine(entry.Name()) { | |||
| 82 | + | if err := r.removeTree(path); err != nil && !errors.Is(err, os.ErrNotExist) { | |||
| 83 | + | return fmt.Errorf("remove snapshot quarantine %q: %w", maintenancePath(r.dir, path), safeFilesystemError(err)) | |||
| 84 | + | } | |||
| 85 | + | if err := r.syncDirectory(sessions); err != nil { | |||
| 86 | + | return fmt.Errorf("sync snapshot maintenance directory for %q: %w", maintenancePath(r.dir, path), safeFilesystemError(err)) | |||
| 87 | + | } | |||
| 88 | + | budget-- | |||
| 89 | + | continue | |||
| 90 | + | } | |||
| 91 | + | if !entry.IsDir() || !canonicalSessionKey(entry.Name()) { | |||
| 92 | + | continue | |||
| 93 | + | } | |||
| 94 | + | lock := r.sessionLock(entry.Name()) | |||
| 95 | + | lock.Lock() | |||
| 96 | + | err := r.maintainSession(ctx, entry.Name(), &budget) | |||
| 97 | + | lock.Unlock() | |||
| 98 | + | if err != nil { | |||
| 99 | + | return err | |||
| 100 | + | } | |||
| 101 | + | } | |||
| 102 | + | return nil | |||
| 103 | + | } | |||
| 104 | + | ||||
| 105 | + | func isQuarantine(name string) bool { return strings.HasPrefix(name, ".deleting-") } | |||
| 106 | + | ||||
| 107 | + | func (r *Repository) maintainSession(ctx context.Context, key string, budget *int) error { | |||
| 108 | + | if err := r.removeTemps(ctx, filepath.Join(r.sessionPath(key), repositoryGenerations), budget); err != nil { | |||
| 109 | + | return err | |||
| 110 | + | } | |||
| 111 | + | objectRoot := filepath.Join(r.sessionPath(key), repositoryObjectsDir) | |||
| 112 | + | shards, err := os.ReadDir(objectRoot) | |||
| 113 | + | if err != nil && !errors.Is(err, os.ErrNotExist) { | |||
| 114 | + | return fmt.Errorf("read snapshot object shards: %w", safeFilesystemError(err)) | |||
| 115 | + | } | |||
| 116 | + | for _, shard := range shards { | |||
| 117 | + | if *budget == 0 { | |||
| 118 | + | return nil | |||
| 119 | + | } | |||
| 120 | + | if shard.IsDir() { | |||
| 121 | + | if err := r.removeTemps(ctx, filepath.Join(objectRoot, shard.Name()), budget); err != nil { | |||
| 122 | + | return err | |||
| 123 | + | } | |||
| 124 | + | } | |||
| 125 | + | } | |||
| 126 | + | ||||
| 127 | + | generations, err := r.generationNumbers(key) | |||
| 128 | + | if err != nil { | |||
| 129 | + | return fmt.Errorf("read snapshot generations: %w", safeFilesystemError(err)) | |||
| 130 | + | } | |||
| 131 | + | type manifestState struct { | |||
| 132 | + | generation uint64 | |||
| 133 | + | refs map[ports.SnapshotDigest]codec.ObjectRef | |||
| 134 | + | complete bool | |||
| 135 | + | } | |||
| 136 | + | states := make([]manifestState, 0, len(generations)) | |||
| 137 | + | complete := 0 | |||
| 138 | + | for _, generation := range generations { | |||
| 139 | + | if err := ctx.Err(); err != nil { | |||
| 140 | + | return err | |||
| 141 | + | } | |||
| 142 | + | state := manifestState{generation: generation} | |||
| 143 | + | data, err := readBounded(r.manifestPath(key, generation)) | |||
| 144 | + | if err == nil { | |||
| 145 | + | manifest, decodeErr := codec.UnmarshalManifest(data) | |||
| 146 | + | if decodeErr == nil && manifest.Generation == generation && sessionKey(manifest.Name) == key { | |||
| 147 | + | state.refs = manifestRefs(manifest) | |||
| 148 | + | if state.refs != nil && withinGenerationBudget(len(data), state.refs) { | |||
| 149 | + | _, loadErr := r.loadGeneration(ctx, manifest.Name, key, generation) | |||
| 150 | + | state.complete = loadErr == nil | |||
| 151 | + | } | |||
| 152 | + | } | |||
| 153 | + | } | |||
| 154 | + | states = append(states, state) | |||
| 155 | + | } | |||
| 156 | + | ||||
| 157 | + | retained := make(map[uint64]bool, 2) | |||
| 158 | + | for _, state := range states { // generationNumbers is newest first. | |||
| 159 | + | if state.complete && complete < 2 { | |||
| 160 | + | retained[state.generation] = true | |||
| 161 | + | complete++ | |||
| 162 | + | } | |||
| 163 | + | } | |||
| 164 | + | referenced := make(map[ports.SnapshotDigest]struct{}) | |||
| 165 | + | for _, state := range states { | |||
| 166 | + | keep := retained[state.generation] || !state.complete | |||
| 167 | + | if !keep && *budget > 0 { | |||
| 168 | + | path := r.manifestPath(key, state.generation) | |||
| 169 | + | if err := r.remove(path); err != nil && !errors.Is(err, os.ErrNotExist) { | |||
| 170 | + | return fmt.Errorf("remove snapshot manifest %q: %w", maintenancePath(r.dir, path), safeFilesystemError(err)) | |||
| 171 | + | } | |||
| 172 | + | if err := r.syncDirectory(filepath.Dir(path)); err != nil { | |||
| 173 | + | return fmt.Errorf("sync snapshot maintenance directory for %q: %w", maintenancePath(r.dir, path), safeFilesystemError(err)) | |||
| 174 | + | } | |||
| 175 | + | *budget-- | |||
| 176 | + | continue | |||
| 177 | + | } | |||
| 178 | + | for digest := range state.refs { // incomplete manifests protect in-flight objects. | |||
| 179 | + | referenced[digest] = struct{}{} | |||
| 180 | + | } | |||
| 181 | + | } | |||
| 182 | + | return r.removeUnreferencedObjects(ctx, key, referenced, budget) | |||
| 183 | + | } | |||
| 184 | + | ||||
| 185 | + | func (r *Repository) removeUnreferencedObjects(ctx context.Context, key string, referenced map[ports.SnapshotDigest]struct{}, budget *int) error { | |||
| 186 | + | root := filepath.Join(r.sessionPath(key), repositoryObjectsDir) | |||
| 187 | + | shards, err := os.ReadDir(root) | |||
| 188 | + | if errors.Is(err, os.ErrNotExist) { | |||
| 189 | + | return nil | |||
| 190 | + | } | |||
| 191 | + | if err != nil { | |||
| 192 | + | return fmt.Errorf("read snapshot objects: %w", safeFilesystemError(err)) | |||
| 193 | + | } | |||
| 194 | + | for _, shard := range shards { | |||
| 195 | + | if *budget == 0 { | |||
| 196 | + | return nil | |||
| 197 | + | } | |||
| 198 | + | if !shard.IsDir() { | |||
| 199 | + | continue | |||
| 200 | + | } | |||
| 201 | + | dir := filepath.Join(root, shard.Name()) | |||
| 202 | + | objects, err := os.ReadDir(dir) | |||
| 203 | + | if err != nil { | |||
| 204 | + | return fmt.Errorf("read snapshot object shard: %w", safeFilesystemError(err)) | |||
| 205 | + | } | |||
| 206 | + | for _, object := range objects { | |||
| 207 | + | if *budget == 0 { | |||
| 208 | + | return nil | |||
| 209 | + | } | |||
| 210 | + | if err := ctx.Err(); err != nil { | |||
| 211 | + | return err | |||
| 212 | + | } | |||
| 213 | + | digest, ok := parseObjectDigest(object.Name()) | |||
| 214 | + | if !ok || object.IsDir() { | |||
| 215 | + | continue | |||
| 216 | + | } | |||
| 217 | + | if _, used := referenced[digest]; used { | |||
| 218 | + | continue | |||
| 219 | + | } | |||
| 220 | + | path := filepath.Join(dir, object.Name()) | |||
| 221 | + | if err := r.remove(path); err != nil && !errors.Is(err, os.ErrNotExist) { | |||
| 222 | + | return fmt.Errorf("remove snapshot object %q: %w", maintenancePath(r.dir, path), safeFilesystemError(err)) | |||
| 223 | + | } | |||
| 224 | + | if err := r.syncDirectory(dir); err != nil { | |||
| 225 | + | return fmt.Errorf("sync snapshot maintenance directory for %q: %w", maintenancePath(r.dir, path), safeFilesystemError(err)) | |||
| 226 | + | } | |||
| 227 | + | *budget-- | |||
| 228 | + | } | |||
| 229 | + | } | |||
| 230 | + | return nil | |||
| 231 | + | } | |||
| 232 | + | ||||
| 233 | + | func (r *Repository) removeTemps(ctx context.Context, dir string, budget *int) error { | |||
| 234 | + | if *budget == 0 { | |||
| 235 | + | return nil | |||
| 236 | + | } | |||
| 237 | + | entries, err := os.ReadDir(dir) | |||
| 238 | + | if errors.Is(err, os.ErrNotExist) { | |||
| 239 | + | return nil | |||
| 240 | + | } | |||
| 241 | + | if err != nil { | |||
| 242 | + | return fmt.Errorf("read snapshot maintenance directory: %w", safeFilesystemError(err)) | |||
| 243 | + | } | |||
| 244 | + | for _, entry := range entries { | |||
| 245 | + | if *budget == 0 { | |||
| 246 | + | return nil | |||
| 247 | + | } | |||
| 248 | + | if err := ctx.Err(); err != nil { | |||
| 249 | + | return err | |||
| 250 | + | } | |||
| 251 | + | if entry.IsDir() || !strings.HasPrefix(entry.Name(), ".tmp-") { | |||
| 252 | + | continue | |||
| 253 | + | } | |||
| 254 | + | path := filepath.Join(dir, entry.Name()) | |||
| 255 | + | if err := r.remove(path); err != nil && !errors.Is(err, os.ErrNotExist) { | |||
| 256 | + | return fmt.Errorf("remove snapshot maintenance file %q: %w", maintenancePath(r.dir, path), safeFilesystemError(err)) | |||
| 257 | + | } | |||
| 258 | + | if err := r.syncDirectory(dir); err != nil { | |||
| 259 | + | return fmt.Errorf("sync snapshot maintenance directory for %q: %w", maintenancePath(r.dir, path), safeFilesystemError(err)) | |||
| 260 | + | } | |||
| 261 | + | *budget-- | |||
| 262 | + | } | |||
| 263 | + | return nil | |||
| 264 | + | } | |||
| 265 | + | ||||
| 266 | + | func (r *Repository) removeTree(path string) error { | |||
| 267 | + | if r.hooks.remove != nil { | |||
| 268 | + | if err := r.hooks.remove(path); err != nil { | |||
| 269 | + | return err | |||
| 270 | + | } | |||
| 271 | + | } | |||
| 272 | + | return os.RemoveAll(path) | |||
| 273 | + | } | |||
| 274 | + | ||||
| 275 | + | // maintenancePath avoids exposing the repository root while retaining a | |||
| 276 | + | // path-safe, actionable location in errors. | |||
| 277 | + | func maintenancePath(root, path string) string { | |||
| 278 | + | rel, err := filepath.Rel(root, path) | |||
| 279 | + | if err != nil || rel == "." || strings.HasPrefix(rel, ".."+string(filepath.Separator)) || rel == ".." { | |||
| 280 | + | return filepath.Base(path) | |||
| 281 | + | } | |||
| 282 | + | return rel | |||
| 283 | + | } | |||
| 284 | + | ||||
| 285 | + | type safePathError struct { | |||
| 286 | + | op string | |||
| 287 | + | cause error | |||
| 288 | + | } | |||
| 289 | + | ||||
| 290 | + | func (e safePathError) Error() string { return e.op + ": " + e.cause.Error() } | |||
| 291 | + | func (e safePathError) Unwrap() error { return e.cause } | |||
| 292 | + | ||||
| 293 | + | // safeFilesystemError preserves error matching without including a filesystem | |||
| 294 | + | // path from terminal-facing error text. | |||
| 295 | + | func safeFilesystemError(err error) error { | |||
| 296 | + | var pathError *os.PathError | |||
| 297 | + | if errors.As(err, &pathError) { | |||
| 298 | + | return safePathError{op: pathError.Op, cause: pathError.Err} | |||
| 299 | + | } | |||
| 300 | + | return err | |||
| 301 | + | } | |||
A internal/adapters/snapshot/maintenance_test.go +163 -0
| @@ -0,0 +1,163 @@ | |||
| 1 | + | package snapshot | |
| 2 | + | ||
| 3 | + | import ( | |
| 4 | + | "context" | |
| 5 | + | "errors" | |
| 6 | + | "os" | |
| 7 | + | "path/filepath" | |
| 8 | + | "testing" | |
| 9 | + | ) | |
| 10 | + | ||
| 11 | + | func TestRepositoryDeleteQuarantinesCanonicalSessionBeforeCleanup(t *testing.T) { | |
| 12 | + | repo := NewRepository(privateDir(t)) | |
| 13 | + | if err := repo.Publish(context.Background(), repositoryPublication(t, "named", 1, []byte("state"))); err != nil { | |
| 14 | + | t.Fatal(err) | |
| 15 | + | } | |
| 16 | + | if err := repo.Delete(context.Background(), "named"); err != nil { | |
| 17 | + | t.Fatal(err) | |
| 18 | + | } | |
| 19 | + | ||
| 20 | + | canonical := repo.sessionPath(sessionKey("named")) | |
| 21 | + | if _, err := os.Lstat(canonical); !errors.Is(err, os.ErrNotExist) { | |
| 22 | + | t.Fatalf("canonical session after Delete = %v, want not exist", err) | |
| 23 | + | } | |
| 24 | + | entries, err := os.ReadDir(filepath.Join(repo.dir, repositorySessionsDir)) | |
| 25 | + | if err != nil { | |
| 26 | + | t.Fatal(err) | |
| 27 | + | } | |
| 28 | + | if len(entries) != 1 || !isQuarantine(entries[0].Name()) || !entries[0].IsDir() { | |
| 29 | + | t.Fatalf("session entries after Delete = %#v, want one quarantine", entries) | |
| 30 | + | } | |
| 31 | + | if err := repo.Maintain(context.Background()); err != nil { | |
| 32 | + | t.Fatal(err) | |
| 33 | + | } | |
| 34 | + | entries, err = os.ReadDir(filepath.Join(repo.dir, repositorySessionsDir)) | |
| 35 | + | if err != nil { | |
| 36 | + | t.Fatal(err) | |
| 37 | + | } | |
| 38 | + | if len(entries) != 0 { | |
| 39 | + | t.Fatalf("session entries after Maintain = %#v, want none", entries) | |
| 40 | + | } | |
| 41 | + | } | |
| 42 | + | ||
| 43 | + | func TestRepositoryDeleteReturnsRootSyncFailureAfterQuarantine(t *testing.T) { | |
| 44 | + | repo := NewRepository(privateDir(t)) | |
| 45 | + | if err := repo.Publish(context.Background(), repositoryPublication(t, "named", 1, []byte("state"))); err != nil { | |
| 46 | + | t.Fatal(err) | |
| 47 | + | } | |
| 48 | + | injected := errors.New("sessions sync") | |
| 49 | + | repo.hooks.syncDirectory = func(string) error { return injected } | |
| 50 | + | if err := repo.Delete(context.Background(), "named"); !errors.Is(err, injected) { | |
| 51 | + | t.Fatalf("Delete error = %v, want root sync failure", err) | |
| 52 | + | } | |
| 53 | + | if _, err := os.Lstat(repo.sessionPath(sessionKey("named"))); !errors.Is(err, os.ErrNotExist) { | |
| 54 | + | t.Fatalf("canonical session after failed sync = %v, want not exist", err) | |
| 55 | + | } | |
| 56 | + | } | |
| 57 | + | ||
| 58 | + | func TestRepositoryDeleteWaitsForPublication(t *testing.T) { | |
| 59 | + | repo := NewRepository(privateDir(t)) | |
| 60 | + | if err := repo.Publish(context.Background(), repositoryPublication(t, "named", 1, []byte("first"))); err != nil { | |
| 61 | + | t.Fatal(err) | |
| 62 | + | } | |
| 63 | + | entered := make(chan struct{}) | |
| 64 | + | release := make(chan struct{}) | |
| 65 | + | repo.hooks.beforeHeadWrite = func(string) error { | |
| 66 | + | close(entered) | |
| 67 | + | <-release | |
| 68 | + | return nil | |
| 69 | + | } | |
| 70 | + | publishDone := make(chan error, 1) | |
| 71 | + | go func() { | |
| 72 | + | publishDone <- repo.Publish(context.Background(), repositoryPublication(t, "named", 2, []byte("second"))) | |
| 73 | + | }() | |
| 74 | + | <-entered | |
| 75 | + | deleteDone := make(chan error, 1) | |
| 76 | + | go func() { deleteDone <- repo.Delete(context.Background(), "named") }() | |
| 77 | + | select { | |
| 78 | + | case err := <-deleteDone: | |
| 79 | + | t.Fatalf("Delete returned before Publish released its lock: %v", err) | |
| 80 | + | default: | |
| 81 | + | } | |
| 82 | + | close(release) | |
| 83 | + | if err := <-publishDone; err != nil { | |
| 84 | + | t.Fatal(err) | |
| 85 | + | } | |
| 86 | + | if err := <-deleteDone; err != nil { | |
| 87 | + | t.Fatal(err) | |
| 88 | + | } | |
| 89 | + | if _, err := os.Lstat(repo.sessionPath(sessionKey("named"))); !errors.Is(err, os.ErrNotExist) { | |
| 90 | + | t.Fatalf("canonical session after concurrent Delete = %v, want not exist", err) | |
| 91 | + | } | |
| 92 | + | } | |
| 93 | + | ||
| 94 | + | func TestRepositoryMaintainRetainsNewestTwoCompleteGenerations(t *testing.T) { | |
| 95 | + | repo := NewRepository(privateDir(t)) | |
| 96 | + | for generation := uint64(1); generation <= 3; generation++ { | |
| 97 | + | if err := repo.Publish(context.Background(), repositoryPublication(t, "named", generation, []byte{byte(generation)})); err != nil { | |
| 98 | + | t.Fatal(err) | |
| 99 | + | } | |
| 100 | + | } | |
| 101 | + | if err := repo.Maintain(context.Background()); err != nil { | |
| 102 | + | t.Fatal(err) | |
| 103 | + | } | |
| 104 | + | key := sessionKey("named") | |
| 105 | + | for _, generation := range []uint64{2, 3} { | |
| 106 | + | if _, err := os.Lstat(repo.manifestPath(key, generation)); err != nil { | |
| 107 | + | t.Fatalf("retained manifest %d: %v", generation, err) | |
| 108 | + | } | |
| 109 | + | } | |
| 110 | + | if _, err := os.Lstat(repo.manifestPath(key, 1)); !errors.Is(err, os.ErrNotExist) { | |
| 111 | + | t.Fatalf("old manifest after Maintain = %v, want not exist", err) | |
| 112 | + | } | |
| 113 | + | for _, generation := range []uint64{2, 3} { | |
| 114 | + | if _, err := repo.Load(context.Background(), "named"); err != nil { | |
| 115 | + | t.Fatalf("Load after collecting generation %d: %v", generation, err) | |
| 116 | + | } | |
| 117 | + | } | |
| 118 | + | } | |
| 119 | + | ||
| 120 | + | func TestRepositoryMaintainPreservesIncompleteManifestReferences(t *testing.T) { | |
| 121 | + | repo := NewRepository(privateDir(t)) | |
| 122 | + | for generation := uint64(1); generation <= 3; generation++ { | |
| 123 | + | if err := repo.Publish(context.Background(), repositoryPublication(t, "named", generation, []byte{byte(generation)})); err != nil { | |
| 124 | + | t.Fatal(err) | |
| 125 | + | } | |
| 126 | + | } | |
| 127 | + | key := sessionKey("named") | |
| 128 | + | manifest := repositoryPublication(t, "named", 3, []byte{3}) | |
| 129 | + | if err := os.Remove(repo.objectPath(key, manifest.Objects[1].Digest)); err != nil { | |
| 130 | + | t.Fatal(err) | |
| 131 | + | } | |
| 132 | + | if err := repo.Maintain(context.Background()); err != nil { | |
| 133 | + | t.Fatal(err) | |
| 134 | + | } | |
| 135 | + | if _, err := os.Lstat(repo.manifestPath(key, 3)); err != nil { | |
| 136 | + | t.Fatalf("incomplete manifest removed: %v", err) | |
| 137 | + | } | |
| 138 | + | if _, err := os.Lstat(repo.objectPath(key, manifest.Objects[0].Digest)); err != nil { | |
| 139 | + | t.Fatalf("object referenced by incomplete manifest removed: %v", err) | |
| 140 | + | } | |
| 141 | + | } | |
| 142 | + | ||
| 143 | + | func TestRepositoryMaintainUsesFixedRemovalBatch(t *testing.T) { | |
| 144 | + | repo := NewRepository(privateDir(t)) | |
| 145 | + | if err := os.Mkdir(repo.dir, 0o700); err != nil { | |
| 146 | + | t.Fatal(err) | |
| 147 | + | } | |
| 148 | + | for i := 0; i < maintenanceBatch+1; i++ { | |
| 149 | + | if err := os.WriteFile(filepath.Join(repo.dir, ".tmp-"+string(rune('a'+i))), []byte("stale"), 0o600); err != nil { | |
| 150 | + | t.Fatal(err) | |
| 151 | + | } | |
| 152 | + | } | |
| 153 | + | if err := repo.Maintain(context.Background()); err != nil { | |
| 154 | + | t.Fatal(err) | |
| 155 | + | } | |
| 156 | + | entries, err := os.ReadDir(repo.dir) | |
| 157 | + | if err != nil { | |
| 158 | + | t.Fatal(err) | |
| 159 | + | } | |
| 160 | + | if len(entries) != 1 { | |
| 161 | + | t.Fatalf("remaining stale entries = %d, want 1 after one batch", len(entries)) | |
| 162 | + | } | |
| 163 | + | } | |
| @@ -0,0 +1,163 @@ | |||||
| 1 | + | package snapshot | |||
| 2 | + | ||||
| 3 | + | import ( | |||
| 4 | + | "context" | |||
| 5 | + | "errors" | |||
| 6 | + | "os" | |||
| 7 | + | "path/filepath" | |||
| 8 | + | "testing" | |||
| 9 | + | ) | |||
| 10 | + | ||||
| 11 | + | func TestRepositoryDeleteQuarantinesCanonicalSessionBeforeCleanup(t *testing.T) { | |||
| 12 | + | repo := NewRepository(privateDir(t)) | |||
| 13 | + | if err := repo.Publish(context.Background(), repositoryPublication(t, "named", 1, []byte("state"))); err != nil { | |||
| 14 | + | t.Fatal(err) | |||
| 15 | + | } | |||
| 16 | + | if err := repo.Delete(context.Background(), "named"); err != nil { | |||
| 17 | + | t.Fatal(err) | |||
| 18 | + | } | |||
| 19 | + | ||||
| 20 | + | canonical := repo.sessionPath(sessionKey("named")) | |||
| 21 | + | if _, err := os.Lstat(canonical); !errors.Is(err, os.ErrNotExist) { | |||
| 22 | + | t.Fatalf("canonical session after Delete = %v, want not exist", err) | |||
| 23 | + | } | |||
| 24 | + | entries, err := os.ReadDir(filepath.Join(repo.dir, repositorySessionsDir)) | |||
| 25 | + | if err != nil { | |||
| 26 | + | t.Fatal(err) | |||
| 27 | + | } | |||
| 28 | + | if len(entries) != 1 || !isQuarantine(entries[0].Name()) || !entries[0].IsDir() { | |||
| 29 | + | t.Fatalf("session entries after Delete = %#v, want one quarantine", entries) | |||
| 30 | + | } | |||
| 31 | + | if err := repo.Maintain(context.Background()); err != nil { | |||
| 32 | + | t.Fatal(err) | |||
| 33 | + | } | |||
| 34 | + | entries, err = os.ReadDir(filepath.Join(repo.dir, repositorySessionsDir)) | |||
| 35 | + | if err != nil { | |||
| 36 | + | t.Fatal(err) | |||
| 37 | + | } | |||
| 38 | + | if len(entries) != 0 { | |||
| 39 | + | t.Fatalf("session entries after Maintain = %#v, want none", entries) | |||
| 40 | + | } | |||
| 41 | + | } | |||
| 42 | + | ||||
| 43 | + | func TestRepositoryDeleteReturnsRootSyncFailureAfterQuarantine(t *testing.T) { | |||
| 44 | + | repo := NewRepository(privateDir(t)) | |||
| 45 | + | if err := repo.Publish(context.Background(), repositoryPublication(t, "named", 1, []byte("state"))); err != nil { | |||
| 46 | + | t.Fatal(err) | |||
| 47 | + | } | |||
| 48 | + | injected := errors.New("sessions sync") | |||
| 49 | + | repo.hooks.syncDirectory = func(string) error { return injected } | |||
| 50 | + | if err := repo.Delete(context.Background(), "named"); !errors.Is(err, injected) { | |||
| 51 | + | t.Fatalf("Delete error = %v, want root sync failure", err) | |||
| 52 | + | } | |||
| 53 | + | if _, err := os.Lstat(repo.sessionPath(sessionKey("named"))); !errors.Is(err, os.ErrNotExist) { | |||
| 54 | + | t.Fatalf("canonical session after failed sync = %v, want not exist", err) | |||
| 55 | + | } | |||
| 56 | + | } | |||
| 57 | + | ||||
| 58 | + | func TestRepositoryDeleteWaitsForPublication(t *testing.T) { | |||
| 59 | + | repo := NewRepository(privateDir(t)) | |||
| 60 | + | if err := repo.Publish(context.Background(), repositoryPublication(t, "named", 1, []byte("first"))); err != nil { | |||
| 61 | + | t.Fatal(err) | |||
| 62 | + | } | |||
| 63 | + | entered := make(chan struct{}) | |||
| 64 | + | release := make(chan struct{}) | |||
| 65 | + | repo.hooks.beforeHeadWrite = func(string) error { | |||
| 66 | + | close(entered) | |||
| 67 | + | <-release | |||
| 68 | + | return nil | |||
| 69 | + | } | |||
| 70 | + | publishDone := make(chan error, 1) | |||
| 71 | + | go func() { | |||
| 72 | + | publishDone <- repo.Publish(context.Background(), repositoryPublication(t, "named", 2, []byte("second"))) | |||
| 73 | + | }() | |||
| 74 | + | <-entered | |||
| 75 | + | deleteDone := make(chan error, 1) | |||
| 76 | + | go func() { deleteDone <- repo.Delete(context.Background(), "named") }() | |||
| 77 | + | select { | |||
| 78 | + | case err := <-deleteDone: | |||
| 79 | + | t.Fatalf("Delete returned before Publish released its lock: %v", err) | |||
| 80 | + | default: | |||
| 81 | + | } | |||
| 82 | + | close(release) | |||
| 83 | + | if err := <-publishDone; err != nil { | |||
| 84 | + | t.Fatal(err) | |||
| 85 | + | } | |||
| 86 | + | if err := <-deleteDone; err != nil { | |||
| 87 | + | t.Fatal(err) | |||
| 88 | + | } | |||
| 89 | + | if _, err := os.Lstat(repo.sessionPath(sessionKey("named"))); !errors.Is(err, os.ErrNotExist) { | |||
| 90 | + | t.Fatalf("canonical session after concurrent Delete = %v, want not exist", err) | |||
| 91 | + | } | |||
| 92 | + | } | |||
| 93 | + | ||||
| 94 | + | func TestRepositoryMaintainRetainsNewestTwoCompleteGenerations(t *testing.T) { | |||
| 95 | + | repo := NewRepository(privateDir(t)) | |||
| 96 | + | for generation := uint64(1); generation <= 3; generation++ { | |||
| 97 | + | if err := repo.Publish(context.Background(), repositoryPublication(t, "named", generation, []byte{byte(generation)})); err != nil { | |||
| 98 | + | t.Fatal(err) | |||
| 99 | + | } | |||
| 100 | + | } | |||
| 101 | + | if err := repo.Maintain(context.Background()); err != nil { | |||
| 102 | + | t.Fatal(err) | |||
| 103 | + | } | |||
| 104 | + | key := sessionKey("named") | |||
| 105 | + | for _, generation := range []uint64{2, 3} { | |||
| 106 | + | if _, err := os.Lstat(repo.manifestPath(key, generation)); err != nil { | |||
| 107 | + | t.Fatalf("retained manifest %d: %v", generation, err) | |||
| 108 | + | } | |||
| 109 | + | } | |||
| 110 | + | if _, err := os.Lstat(repo.manifestPath(key, 1)); !errors.Is(err, os.ErrNotExist) { | |||
| 111 | + | t.Fatalf("old manifest after Maintain = %v, want not exist", err) | |||
| 112 | + | } | |||
| 113 | + | for _, generation := range []uint64{2, 3} { | |||
| 114 | + | if _, err := repo.Load(context.Background(), "named"); err != nil { | |||
| 115 | + | t.Fatalf("Load after collecting generation %d: %v", generation, err) | |||
| 116 | + | } | |||
| 117 | + | } | |||
| 118 | + | } | |||
| 119 | + | ||||
| 120 | + | func TestRepositoryMaintainPreservesIncompleteManifestReferences(t *testing.T) { | |||
| 121 | + | repo := NewRepository(privateDir(t)) | |||
| 122 | + | for generation := uint64(1); generation <= 3; generation++ { | |||
| 123 | + | if err := repo.Publish(context.Background(), repositoryPublication(t, "named", generation, []byte{byte(generation)})); err != nil { | |||
| 124 | + | t.Fatal(err) | |||
| 125 | + | } | |||
| 126 | + | } | |||
| 127 | + | key := sessionKey("named") | |||
| 128 | + | manifest := repositoryPublication(t, "named", 3, []byte{3}) | |||
| 129 | + | if err := os.Remove(repo.objectPath(key, manifest.Objects[1].Digest)); err != nil { | |||
| 130 | + | t.Fatal(err) | |||
| 131 | + | } | |||
| 132 | + | if err := repo.Maintain(context.Background()); err != nil { | |||
| 133 | + | t.Fatal(err) | |||
| 134 | + | } | |||
| 135 | + | if _, err := os.Lstat(repo.manifestPath(key, 3)); err != nil { | |||
| 136 | + | t.Fatalf("incomplete manifest removed: %v", err) | |||
| 137 | + | } | |||
| 138 | + | if _, err := os.Lstat(repo.objectPath(key, manifest.Objects[0].Digest)); err != nil { | |||
| 139 | + | t.Fatalf("object referenced by incomplete manifest removed: %v", err) | |||
| 140 | + | } | |||
| 141 | + | } | |||
| 142 | + | ||||
| 143 | + | func TestRepositoryMaintainUsesFixedRemovalBatch(t *testing.T) { | |||
| 144 | + | repo := NewRepository(privateDir(t)) | |||
| 145 | + | if err := os.Mkdir(repo.dir, 0o700); err != nil { | |||
| 146 | + | t.Fatal(err) | |||
| 147 | + | } | |||
| 148 | + | for i := 0; i < maintenanceBatch+1; i++ { | |||
| 149 | + | if err := os.WriteFile(filepath.Join(repo.dir, ".tmp-"+string(rune('a'+i))), []byte("stale"), 0o600); err != nil { | |||
| 150 | + | t.Fatal(err) | |||
| 151 | + | } | |||
| 152 | + | } | |||
| 153 | + | if err := repo.Maintain(context.Background()); err != nil { | |||
| 154 | + | t.Fatal(err) | |||
| 155 | + | } | |||
| 156 | + | entries, err := os.ReadDir(repo.dir) | |||
| 157 | + | if err != nil { | |||
| 158 | + | t.Fatal(err) | |||
| 159 | + | } | |||
| 160 | + | if len(entries) != 1 { | |||
| 161 | + | t.Fatalf("remaining stale entries = %d, want 1 after one batch", len(entries)) | |||
| 162 | + | } | |||
| 163 | + | } | |||
M internal/adapters/snapshot/path.go +9 -0
| @@ -52,3 +52,12 @@ func parseGenerationFilename(name string) (uint64, bool) { | |||
| 52 | 52 | generation, err := strconv.ParseUint(n, 10, 64) | |
| 53 | 53 | return generation, err == nil && generationFilename(generation) == name && generation != 0 | |
| 54 | 54 | } | |
| 55 | + | ||
| 56 | + | func parseObjectDigest(name string) (ports.SnapshotDigest, bool) { | |
| 57 | + | var digest ports.SnapshotDigest | |
| 58 | + | if len(name) != hex.EncodedLen(len(digest)) || strings.ToLower(name) != name { | |
| 59 | + | return digest, false | |
| 60 | + | } | |
| 61 | + | _, err := hex.Decode(digest[:], []byte(name)) | |
| 62 | + | return digest, err == nil | |
| 63 | + | } | |
| @@ -52,3 +52,12 @@ func parseGenerationFilename(name string) (uint64, bool) { | |||||
| 52 | generation, err := strconv.ParseUint(n, 10, 64) | 52 | generation, err := strconv.ParseUint(n, 10, 64) | ||
| 53 | return generation, err == nil && generationFilename(generation) == name && generation != 0 | 53 | return generation, err == nil && generationFilename(generation) == name && generation != 0 | ||
| 54 | } | 54 | } | ||
| 55 | + | ||||
| 56 | + | func parseObjectDigest(name string) (ports.SnapshotDigest, bool) { | |||
| 57 | + | var digest ports.SnapshotDigest | |||
| 58 | + | if len(name) != hex.EncodedLen(len(digest)) || strings.ToLower(name) != name { | |||
| 59 | + | return digest, false | |||
| 60 | + | } | |||
| 61 | + | _, err := hex.Decode(digest[:], []byte(name)) | |||
| 62 | + | return digest, err == nil | |||
| 63 | + | } | |||
M internal/adapters/snapshot/repository.go +0 -90
| @@ -2,11 +2,6 @@ package snapshot | |||
| 2 | 2 | ||
| 3 | 3 | import ( | |
| 4 | 4 | "context" | |
| 5 | - | "errors" | |
| 6 | - | "fmt" | |
| 7 | - | "os" | |
| 8 | - | "path/filepath" | |
| 9 | - | "strings" | |
| 10 | 5 | "sync" | |
| 11 | 6 | ||
| 12 | 7 | "github.com/bnema/vev/internal/ports" | |
| @@ -59,91 +54,6 @@ func (r *Repository) sessionLock(key string) *sync.Mutex { | |||
| 59 | 54 | return lock.(*sync.Mutex) | |
| 60 | 55 | } | |
| 61 | 56 | ||
| 62 | - | func (r *Repository) Delete(ctx context.Context, name string) error { | |
| 63 | - | if err := ctx.Err(); err != nil { | |
| 64 | - | return err | |
| 65 | - | } | |
| 66 | - | key := sessionKey(name) | |
| 67 | - | lock := r.sessionLock(key) | |
| 68 | - | lock.Lock() | |
| 69 | - | defer lock.Unlock() | |
| 70 | - | ||
| 71 | - | path := r.sessionPath(key) | |
| 72 | - | if _, err := os.Lstat(path); errors.Is(err, os.ErrNotExist) { | |
| 73 | - | return nil | |
| 74 | - | } else if err != nil { | |
| 75 | - | return fmt.Errorf("stat snapshot session %q: %w", key, safeFilesystemError(err)) | |
| 76 | - | } | |
| 77 | - | if err := os.RemoveAll(path); err != nil { | |
| 78 | - | return fmt.Errorf("remove snapshot session %q: %w", key, safeFilesystemError(err)) | |
| 79 | - | } | |
| 80 | - | if err := r.syncDirectory(filepath.Join(r.dir, repositorySessionsDir)); err != nil { | |
| 81 | - | return fmt.Errorf("sync deleted snapshot session directory %q: %w", key, safeFilesystemError(err)) | |
| 82 | - | } | |
| 83 | - | return nil | |
| 84 | - | } | |
| 85 | - | ||
| 86 | - | // Maintain removes abandoned temporary and quarantine files. Immutable data is | |
| 87 | - | // deliberately never garbage-collected here because older generations need it. | |
| 88 | - | func (r *Repository) Maintain(ctx context.Context) error { | |
| 89 | - | if err := ctx.Err(); err != nil { | |
| 90 | - | return err | |
| 91 | - | } | |
| 92 | - | return filepath.WalkDir(r.dir, func(path string, entry os.DirEntry, err error) error { | |
| 93 | - | if err != nil { | |
| 94 | - | if errors.Is(err, os.ErrNotExist) { | |
| 95 | - | return nil | |
| 96 | - | } | |
| 97 | - | return fmt.Errorf("walk snapshot maintenance path %q: %w", maintenancePath(r.dir, path), safeFilesystemError(err)) | |
| 98 | - | } | |
| 99 | - | if err := ctx.Err(); err != nil { | |
| 100 | - | return err | |
| 101 | - | } | |
| 102 | - | if entry.IsDir() { | |
| 103 | - | return nil | |
| 104 | - | } | |
| 105 | - | base := entry.Name() | |
| 106 | - | if !strings.HasPrefix(base, ".tmp-") && !strings.HasPrefix(base, ".quarantine-") { | |
| 107 | - | return nil | |
| 108 | - | } | |
| 109 | - | if err := r.remove(path); err != nil && !errors.Is(err, os.ErrNotExist) { | |
| 110 | - | return fmt.Errorf("remove snapshot maintenance file %q: %w", maintenancePath(r.dir, path), safeFilesystemError(err)) | |
| 111 | - | } | |
| 112 | - | if err := r.syncDirectory(filepath.Dir(path)); err != nil { | |
| 113 | - | return fmt.Errorf("sync snapshot maintenance directory for %q: %w", maintenancePath(r.dir, path), safeFilesystemError(err)) | |
| 114 | - | } | |
| 115 | - | return nil | |
| 116 | - | }) | |
| 117 | - | } | |
| 118 | - | ||
| 119 | - | // maintenancePath avoids exposing the repository root while retaining a | |
| 120 | - | // path-safe, actionable location in errors. | |
| 121 | - | func maintenancePath(root, path string) string { | |
| 122 | - | rel, err := filepath.Rel(root, path) | |
| 123 | - | if err != nil || rel == "." || strings.HasPrefix(rel, ".."+string(filepath.Separator)) || rel == ".." { | |
| 124 | - | return filepath.Base(path) | |
| 125 | - | } | |
| 126 | - | return rel | |
| 127 | - | } | |
| 128 | - | ||
| 129 | - | type safePathError struct { | |
| 130 | - | op string | |
| 131 | - | cause error | |
| 132 | - | } | |
| 133 | - | ||
| 134 | - | func (e safePathError) Error() string { return e.op + ": " + e.cause.Error() } | |
| 135 | - | func (e safePathError) Unwrap() error { return e.cause } | |
| 136 | - | ||
| 137 | - | // safeFilesystemError preserves error matching without including a filesystem | |
| 138 | - | // path from an os.PathError in terminal-facing error text. | |
| 139 | - | func safeFilesystemError(err error) error { | |
| 140 | - | var pathError *os.PathError | |
| 141 | - | if errors.As(err, &pathError) { | |
| 142 | - | return safePathError{op: pathError.Op, cause: pathError.Err} | |
| 143 | - | } | |
| 144 | - | return err | |
| 145 | - | } | |
| 146 | - | ||
| 147 | 57 | func (r *Repository) LoadLegacy(ctx context.Context) ([]ports.LegacySnapshot, error) { | |
| 148 | 58 | if err := ctx.Err(); err != nil { | |
| 149 | 59 | return nil, err | |
| @@ -2,11 +2,6 @@ package snapshot | |||||
| 2 | 2 | ||||
| 3 | import ( | 3 | import ( | ||
| 4 | "context" | 4 | "context" | ||
| 5 | - | "errors" | |||
| 6 | - | "fmt" | |||
| 7 | - | "os" | |||
| 8 | - | "path/filepath" | |||
| 9 | - | "strings" | |||
| 10 | "sync" | 5 | "sync" | ||
| 11 | 6 | ||||
| 12 | "github.com/bnema/vev/internal/ports" | 7 | "github.com/bnema/vev/internal/ports" | ||
| @@ -59,91 +54,6 @@ func (r *Repository) sessionLock(key string) *sync.Mutex { | |||||
| 59 | return lock.(*sync.Mutex) | 54 | return lock.(*sync.Mutex) | ||
| 60 | } | 55 | } | ||
| 61 | 56 | ||||
| 62 | - | func (r *Repository) Delete(ctx context.Context, name string) error { | |||
| 63 | - | if err := ctx.Err(); err != nil { | |||
| 64 | - | return err | |||
| 65 | - | } | |||
| 66 | - | key := sessionKey(name) | |||
| 67 | - | lock := r.sessionLock(key) | |||
| 68 | - | lock.Lock() | |||
| 69 | - | defer lock.Unlock() | |||
| 70 | - | ||||
| 71 | - | path := r.sessionPath(key) | |||
| 72 | - | if _, err := os.Lstat(path); errors.Is(err, os.ErrNotExist) { | |||
| 73 | - | return nil | |||
| 74 | - | } else if err != nil { | |||
| 75 | - | return fmt.Errorf("stat snapshot session %q: %w", key, safeFilesystemError(err)) | |||
| 76 | - | } | |||
| 77 | - | if err := os.RemoveAll(path); err != nil { | |||
| 78 | - | return fmt.Errorf("remove snapshot session %q: %w", key, safeFilesystemError(err)) | |||
| 79 | - | } | |||
| 80 | - | if err := r.syncDirectory(filepath.Join(r.dir, repositorySessionsDir)); err != nil { | |||
| 81 | - | return fmt.Errorf("sync deleted snapshot session directory %q: %w", key, safeFilesystemError(err)) | |||
| 82 | - | } | |||
| 83 | - | return nil | |||
| 84 | - | } | |||
| 85 | - | ||||
| 86 | - | // Maintain removes abandoned temporary and quarantine files. Immutable data is | |||
| 87 | - | // deliberately never garbage-collected here because older generations need it. | |||
| 88 | - | func (r *Repository) Maintain(ctx context.Context) error { | |||
| 89 | - | if err := ctx.Err(); err != nil { | |||
| 90 | - | return err | |||
| 91 | - | } | |||
| 92 | - | return filepath.WalkDir(r.dir, func(path string, entry os.DirEntry, err error) error { | |||
| 93 | - | if err != nil { | |||
| 94 | - | if errors.Is(err, os.ErrNotExist) { | |||
| 95 | - | return nil | |||
| 96 | - | } | |||
| 97 | - | return fmt.Errorf("walk snapshot maintenance path %q: %w", maintenancePath(r.dir, path), safeFilesystemError(err)) | |||
| 98 | - | } | |||
| 99 | - | if err := ctx.Err(); err != nil { | |||
| 100 | - | return err | |||
| 101 | - | } | |||
| 102 | - | if entry.IsDir() { | |||
| 103 | - | return nil | |||
| 104 | - | } | |||
| 105 | - | base := entry.Name() | |||
| 106 | - | if !strings.HasPrefix(base, ".tmp-") && !strings.HasPrefix(base, ".quarantine-") { | |||
| 107 | - | return nil | |||
| 108 | - | } | |||
| 109 | - | if err := r.remove(path); err != nil && !errors.Is(err, os.ErrNotExist) { | |||
| 110 | - | return fmt.Errorf("remove snapshot maintenance file %q: %w", maintenancePath(r.dir, path), safeFilesystemError(err)) | |||
| 111 | - | } | |||
| 112 | - | if err := r.syncDirectory(filepath.Dir(path)); err != nil { | |||
| 113 | - | return fmt.Errorf("sync snapshot maintenance directory for %q: %w", maintenancePath(r.dir, path), safeFilesystemError(err)) | |||
| 114 | - | } | |||
| 115 | - | return nil | |||
| 116 | - | }) | |||
| 117 | - | } | |||
| 118 | - | ||||
| 119 | - | // maintenancePath avoids exposing the repository root while retaining a | |||
| 120 | - | // path-safe, actionable location in errors. | |||
| 121 | - | func maintenancePath(root, path string) string { | |||
| 122 | - | rel, err := filepath.Rel(root, path) | |||
| 123 | - | if err != nil || rel == "." || strings.HasPrefix(rel, ".."+string(filepath.Separator)) || rel == ".." { | |||
| 124 | - | return filepath.Base(path) | |||
| 125 | - | } | |||
| 126 | - | return rel | |||
| 127 | - | } | |||
| 128 | - | ||||
| 129 | - | type safePathError struct { | |||
| 130 | - | op string | |||
| 131 | - | cause error | |||
| 132 | - | } | |||
| 133 | - | ||||
| 134 | - | func (e safePathError) Error() string { return e.op + ": " + e.cause.Error() } | |||
| 135 | - | func (e safePathError) Unwrap() error { return e.cause } | |||
| 136 | - | ||||
| 137 | - | // safeFilesystemError preserves error matching without including a filesystem | |||
| 138 | - | // path from an os.PathError in terminal-facing error text. | |||
| 139 | - | func safeFilesystemError(err error) error { | |||
| 140 | - | var pathError *os.PathError | |||
| 141 | - | if errors.As(err, &pathError) { | |||
| 142 | - | return safePathError{op: pathError.Op, cause: pathError.Err} | |||
| 143 | - | } | |||
| 144 | - | return err | |||
| 145 | - | } | |||
| 146 | - | ||||
| 147 | func (r *Repository) LoadLegacy(ctx context.Context) ([]ports.LegacySnapshot, error) { | 57 | func (r *Repository) LoadLegacy(ctx context.Context) ([]ports.LegacySnapshot, error) { | ||
| 148 | if err := ctx.Err(); err != nil { | 58 | if err := ctx.Err(); err != nil { | ||
| 149 | return nil, err | 59 | return nil, err | ||