fix(snapshot): budget temp shard traversal
- commit
- 5c87cd625b13473233e84fcf54a3cff6788a1098
- author
- brice · · 2026-07-22 03:26 UTC
- parent
- 26baad6
- signature
- GPG Verified
CD88F9E564E0102646FB6077A1A07C7C7C3E349D· arch-gpg Verified for the committer email
4 files · +97 -34
M internal/adapters/snapshot/load_bounds_test.go +7 -1
| @@ -75,10 +75,16 @@ func TestRepositoryMaintainDoesNotCollectBeyondRetainedMetadataBudget(t *testing | |||
| 75 | 75 | t.Fatal(err) | |
| 76 | 76 | } | |
| 77 | 77 | } | |
| 78 | - | for pass := 0; pass < 3; pass++ { | |
| 78 | + | for pass := 0; pass < maxMaintenanceMarkedGenerations*3; pass++ { | |
| 79 | 79 | if err := repo.Maintain(context.Background()); err != nil { | |
| 80 | 80 | t.Fatal(err) | |
| 81 | 81 | } | |
| 82 | + | if repo.maintenanceSessions[sessionKey("named")] == nil { | |
| 83 | + | break | |
| 84 | + | } | |
| 85 | + | if pass == maxMaintenanceMarkedGenerations*3-1 { | |
| 86 | + | t.Fatal("maintenance did not complete after bounded object-shard traversal") | |
| 87 | + | } | |
| 82 | 88 | } | |
| 83 | 89 | if _, err := os.Lstat(repo.manifestPath(sessionKey("named"), 1)); err != nil { | |
| 84 | 90 | t.Fatalf("old manifest collected after mark budget overflow: %v", err) | |
| @@ -75,10 +75,16 @@ func TestRepositoryMaintainDoesNotCollectBeyondRetainedMetadataBudget(t *testing | |||||
| 75 | t.Fatal(err) | 75 | t.Fatal(err) | ||
| 76 | } | 76 | } | ||
| 77 | } | 77 | } | ||
| 78 | - | for pass := 0; pass < 3; pass++ { | 78 | + | for pass := 0; pass < maxMaintenanceMarkedGenerations*3; pass++ { |
| 79 | if err := repo.Maintain(context.Background()); err != nil { | 79 | if err := repo.Maintain(context.Background()); err != nil { | ||
| 80 | t.Fatal(err) | 80 | t.Fatal(err) | ||
| 81 | } | 81 | } | ||
| 82 | + | if repo.maintenanceSessions[sessionKey("named")] == nil { | |||
| 83 | + | break | |||
| 84 | + | } | |||
| 85 | + | if pass == maxMaintenanceMarkedGenerations*3-1 { | |||
| 86 | + | t.Fatal("maintenance did not complete after bounded object-shard traversal") | |||
| 87 | + | } | |||
| 82 | } | 88 | } | ||
| 83 | if _, err := os.Lstat(repo.manifestPath(sessionKey("named"), 1)); err != nil { | 89 | if _, err := os.Lstat(repo.manifestPath(sessionKey("named"), 1)); err != nil { | ||
| 84 | t.Fatalf("old manifest collected after mark budget overflow: %v", err) | 90 | t.Fatalf("old manifest collected after mark budget overflow: %v", err) | ||
M internal/adapters/snapshot/maintenance_gc.go +28 -30
| @@ -268,39 +268,37 @@ func (r *Repository) removeObsoleteManifests(ctx context.Context, key string, st | |||
| 268 | 268 | return nil | |
| 269 | 269 | } | |
| 270 | 270 | ||
| 271 | - | // removeObjectTemps processes a single shard to completion before advancing | |
| 272 | - | // the shard-root cursor. Thus a hostile set of shards can retain at most the | |
| 273 | - | // root cursor and the one active shard cursor, rather than one cursor per | |
| 274 | - | // discovered shard. | |
| 271 | + | // removeObjectTemps performs at most one shard traversal or one root-entry | |
| 272 | + | // discovery per Maintain call. Empty attacker-controlled shards consume no | |
| 273 | + | // removal budget, so this fixed step ceiling prevents an unbounded scan of | |
| 274 | + | // them while retaining only the root cursor and one active shard cursor. | |
| 275 | 275 | func (r *Repository) removeObjectTemps(ctx context.Context, key string, state *sessionMaintenance, budget *int) (bool, error) { | |
| 276 | 276 | root := filepath.Join(r.sessionPath(key), repositoryObjectsDir) | |
| 277 | - | for *budget > 0 { | |
| 278 | - | if state.objectTempShard != "" { | |
| 279 | - | done, err := r.removeTemps(ctx, filepath.Join(root, state.objectTempShard), budget, "object-temps:"+key+":"+state.objectTempShard) | |
| 280 | - | if err != nil || !done { | |
| 281 | - | return false, err | |
| 282 | - | } | |
| 283 | - | // This releases the active shard cursor before the next root entry | |
| 284 | - | // is discovered. A completed small shard may therefore advance in | |
| 285 | - | // this call without accumulating continuation state. | |
| 286 | - | state.objectTempShard = "" | |
| 287 | - | continue | |
| 288 | - | } | |
| 289 | - | if state.objectTempsDone { | |
| 290 | - | return true, nil | |
| 291 | - | } | |
| 292 | - | entries, done, err := r.readMaintenanceDir(root, 1, "object-temps-shards:"+key) | |
| 293 | - | if errors.Is(err, os.ErrNotExist) { | |
| 294 | - | state.objectTempsDone = true | |
| 295 | - | return true, nil | |
| 296 | - | } | |
| 297 | - | if err != nil { | |
| 298 | - | return false, fmt.Errorf("read snapshot object shards: %w", safeFilesystemError(err)) | |
| 299 | - | } | |
| 300 | - | state.objectTempsDone = done | |
| 301 | - | if len(entries) != 0 && entries[0].isDir { | |
| 302 | - | state.objectTempShard = entries[0].name | |
| 277 | + | if *budget == 0 { | |
| 278 | + | return false, nil | |
| 279 | + | } | |
| 280 | + | if state.objectTempShard != "" { | |
| 281 | + | done, err := r.removeTemps(ctx, filepath.Join(root, state.objectTempShard), budget, "object-temps:"+key+":"+state.objectTempShard) | |
| 282 | + | if err != nil || !done { | |
| 283 | + | return false, err | |
| 303 | 284 | } | |
| 285 | + | state.objectTempShard = "" | |
| 286 | + | return state.objectTempsDone, nil | |
| 287 | + | } | |
| 288 | + | if state.objectTempsDone { | |
| 289 | + | return true, nil | |
| 290 | + | } | |
| 291 | + | entries, done, err := r.readMaintenanceDir(root, 1, "object-temps-shards:"+key) | |
| 292 | + | if errors.Is(err, os.ErrNotExist) { | |
| 293 | + | state.objectTempsDone = true | |
| 294 | + | return true, nil | |
| 295 | + | } | |
| 296 | + | if err != nil { | |
| 297 | + | return false, fmt.Errorf("read snapshot object shards: %w", safeFilesystemError(err)) | |
| 298 | + | } | |
| 299 | + | state.objectTempsDone = done | |
| 300 | + | if len(entries) != 0 && entries[0].isDir { | |
| 301 | + | state.objectTempShard = entries[0].name | |
| 304 | 302 | } | |
| 305 | 303 | return state.objectTempsDone && state.objectTempShard == "", nil | |
| 306 | 304 | } | |
| @@ -268,39 +268,37 @@ func (r *Repository) removeObsoleteManifests(ctx context.Context, key string, st | |||||
| 268 | return nil | 268 | return nil | ||
| 269 | } | 269 | } | ||
| 270 | 270 | ||||
| 271 | - | // removeObjectTemps processes a single shard to completion before advancing | 271 | + | // removeObjectTemps performs at most one shard traversal or one root-entry |
| 272 | - | // the shard-root cursor. Thus a hostile set of shards can retain at most the | 272 | + | // discovery per Maintain call. Empty attacker-controlled shards consume no |
| 273 | - | // root cursor and the one active shard cursor, rather than one cursor per | 273 | + | // removal budget, so this fixed step ceiling prevents an unbounded scan of |
| 274 | - | // discovered shard. | 274 | + | // them while retaining only the root cursor and one active shard cursor. |
| 275 | func (r *Repository) removeObjectTemps(ctx context.Context, key string, state *sessionMaintenance, budget *int) (bool, error) { | 275 | func (r *Repository) removeObjectTemps(ctx context.Context, key string, state *sessionMaintenance, budget *int) (bool, error) { | ||
| 276 | root := filepath.Join(r.sessionPath(key), repositoryObjectsDir) | 276 | root := filepath.Join(r.sessionPath(key), repositoryObjectsDir) | ||
| 277 | - | for *budget > 0 { | 277 | + | if *budget == 0 { |
| 278 | - | if state.objectTempShard != "" { | 278 | + | return false, nil |
| 279 | - | done, err := r.removeTemps(ctx, filepath.Join(root, state.objectTempShard), budget, "object-temps:"+key+":"+state.objectTempShard) | 279 | + | } |
| 280 | - | if err != nil || !done { | 280 | + | if state.objectTempShard != "" { |
| 281 | - | return false, err | 281 | + | done, err := r.removeTemps(ctx, filepath.Join(root, state.objectTempShard), budget, "object-temps:"+key+":"+state.objectTempShard) |
| 282 | - | } | 282 | + | if err != nil || !done { |
| 283 | - | // This releases the active shard cursor before the next root entry | 283 | + | return false, err |
| 284 | - | // is discovered. A completed small shard may therefore advance in | |||
| 285 | - | // this call without accumulating continuation state. | |||
| 286 | - | state.objectTempShard = "" | |||
| 287 | - | continue | |||
| 288 | - | } | |||
| 289 | - | if state.objectTempsDone { | |||
| 290 | - | return true, nil | |||
| 291 | - | } | |||
| 292 | - | entries, done, err := r.readMaintenanceDir(root, 1, "object-temps-shards:"+key) | |||
| 293 | - | if errors.Is(err, os.ErrNotExist) { | |||
| 294 | - | state.objectTempsDone = true | |||
| 295 | - | return true, nil | |||
| 296 | - | } | |||
| 297 | - | if err != nil { | |||
| 298 | - | return false, fmt.Errorf("read snapshot object shards: %w", safeFilesystemError(err)) | |||
| 299 | - | } | |||
| 300 | - | state.objectTempsDone = done | |||
| 301 | - | if len(entries) != 0 && entries[0].isDir { | |||
| 302 | - | state.objectTempShard = entries[0].name | |||
| 303 | } | 284 | } | ||
| 285 | + | state.objectTempShard = "" | |||
| 286 | + | return state.objectTempsDone, nil | |||
| 287 | + | } | |||
| 288 | + | if state.objectTempsDone { | |||
| 289 | + | return true, nil | |||
| 290 | + | } | |||
| 291 | + | entries, done, err := r.readMaintenanceDir(root, 1, "object-temps-shards:"+key) | |||
| 292 | + | if errors.Is(err, os.ErrNotExist) { | |||
| 293 | + | state.objectTempsDone = true | |||
| 294 | + | return true, nil | |||
| 295 | + | } | |||
| 296 | + | if err != nil { | |||
| 297 | + | return false, fmt.Errorf("read snapshot object shards: %w", safeFilesystemError(err)) | |||
| 298 | + | } | |||
| 299 | + | state.objectTempsDone = done | |||
| 300 | + | if len(entries) != 0 && entries[0].isDir { | |||
| 301 | + | state.objectTempShard = entries[0].name | |||
| 304 | } | 302 | } | ||
| 305 | return state.objectTempsDone && state.objectTempShard == "", nil | 303 | return state.objectTempsDone && state.objectTempShard == "", nil | ||
| 306 | } | 304 | } | ||
M internal/adapters/snapshot/maintenance_hardening_test.go +47 -0
| @@ -46,6 +46,53 @@ func TestRepositoryMaintainBoundsCursorsAcrossHostileObjectShards(t *testing.T) | |||
| 46 | 46 | } | |
| 47 | 47 | } | |
| 48 | 48 | ||
| 49 | + | func TestRepositoryMaintainBoundsEmptyObjectTempShards(t *testing.T) { | |
| 50 | + | repo := NewRepository(privateDir(t)) | |
| 51 | + | key := sessionKey("named") | |
| 52 | + | objects := filepath.Join(repo.sessionPath(key), repositoryObjectsDir) | |
| 53 | + | const shards = maintenanceBatch * 2 | |
| 54 | + | for i := 0; i < shards; i++ { | |
| 55 | + | if err := os.MkdirAll(filepath.Join(objects, fmt.Sprintf("%03d", i)), 0o700); err != nil { | |
| 56 | + | t.Fatal(err) | |
| 57 | + | } | |
| 58 | + | } | |
| 59 | + | ||
| 60 | + | openedShards := make(map[string]struct{}, shards) | |
| 61 | + | opensThisPass := 0 | |
| 62 | + | tempPass := false | |
| 63 | + | repo.hooks.openMaintenanceDirectory = func(dir string) (maintenanceDirectory, error) { | |
| 64 | + | if tempPass && filepath.Dir(dir) == objects { | |
| 65 | + | opensThisPass++ | |
| 66 | + | openedShards[dir] = struct{}{} | |
| 67 | + | } | |
| 68 | + | file, err := os.Open(dir) | |
| 69 | + | if err != nil { | |
| 70 | + | return nil, err | |
| 71 | + | } | |
| 72 | + | return osMaintenanceDirectory{file: file}, nil | |
| 73 | + | } | |
| 74 | + | ||
| 75 | + | for pass := 0; pass < shards*3; pass++ { | |
| 76 | + | opensThisPass = 0 | |
| 77 | + | tempPass = true | |
| 78 | + | if err := repo.Maintain(context.Background()); err != nil { | |
| 79 | + | t.Fatal(err) | |
| 80 | + | } | |
| 81 | + | tempPass = false | |
| 82 | + | if opensThisPass > 1 { | |
| 83 | + | t.Fatalf("empty object shards opened in maintenance pass %d = %d, want at most 1", pass, opensThisPass) | |
| 84 | + | } | |
| 85 | + | state := repo.maintenanceSessions[key] | |
| 86 | + | if state != nil && state.objectTempsDone { | |
| 87 | + | if got := len(openedShards); got != shards { | |
| 88 | + | t.Fatalf("empty object shards opened = %d, want %d", got, shards) | |
| 89 | + | } | |
| 90 | + | return | |
| 91 | + | } | |
| 92 | + | } | |
| 93 | + | t.Fatal("empty object temp shards were not eventually traversed") | |
| 94 | + | } | |
| 95 | + | ||
| 49 | 96 | func TestRepositoryMaintainBoundsAndResumesDeepWideQuarantine(t *testing.T) { | |
| 50 | 97 | repo := NewRepository(privateDir(t)) | |
| 51 | 98 | quarantine := filepath.Join(repo.dir, repositorySessionsDir, ".deleting-hostile") | |
| @@ -46,6 +46,53 @@ func TestRepositoryMaintainBoundsCursorsAcrossHostileObjectShards(t *testing.T) | |||||
| 46 | } | 46 | } | ||
| 47 | } | 47 | } | ||
| 48 | 48 | ||||
| 49 | + | func TestRepositoryMaintainBoundsEmptyObjectTempShards(t *testing.T) { | |||
| 50 | + | repo := NewRepository(privateDir(t)) | |||
| 51 | + | key := sessionKey("named") | |||
| 52 | + | objects := filepath.Join(repo.sessionPath(key), repositoryObjectsDir) | |||
| 53 | + | const shards = maintenanceBatch * 2 | |||
| 54 | + | for i := 0; i < shards; i++ { | |||
| 55 | + | if err := os.MkdirAll(filepath.Join(objects, fmt.Sprintf("%03d", i)), 0o700); err != nil { | |||
| 56 | + | t.Fatal(err) | |||
| 57 | + | } | |||
| 58 | + | } | |||
| 59 | + | ||||
| 60 | + | openedShards := make(map[string]struct{}, shards) | |||
| 61 | + | opensThisPass := 0 | |||
| 62 | + | tempPass := false | |||
| 63 | + | repo.hooks.openMaintenanceDirectory = func(dir string) (maintenanceDirectory, error) { | |||
| 64 | + | if tempPass && filepath.Dir(dir) == objects { | |||
| 65 | + | opensThisPass++ | |||
| 66 | + | openedShards[dir] = struct{}{} | |||
| 67 | + | } | |||
| 68 | + | file, err := os.Open(dir) | |||
| 69 | + | if err != nil { | |||
| 70 | + | return nil, err | |||
| 71 | + | } | |||
| 72 | + | return osMaintenanceDirectory{file: file}, nil | |||
| 73 | + | } | |||
| 74 | + | ||||
| 75 | + | for pass := 0; pass < shards*3; pass++ { | |||
| 76 | + | opensThisPass = 0 | |||
| 77 | + | tempPass = true | |||
| 78 | + | if err := repo.Maintain(context.Background()); err != nil { | |||
| 79 | + | t.Fatal(err) | |||
| 80 | + | } | |||
| 81 | + | tempPass = false | |||
| 82 | + | if opensThisPass > 1 { | |||
| 83 | + | t.Fatalf("empty object shards opened in maintenance pass %d = %d, want at most 1", pass, opensThisPass) | |||
| 84 | + | } | |||
| 85 | + | state := repo.maintenanceSessions[key] | |||
| 86 | + | if state != nil && state.objectTempsDone { | |||
| 87 | + | if got := len(openedShards); got != shards { | |||
| 88 | + | t.Fatalf("empty object shards opened = %d, want %d", got, shards) | |||
| 89 | + | } | |||
| 90 | + | return | |||
| 91 | + | } | |||
| 92 | + | } | |||
| 93 | + | t.Fatal("empty object temp shards were not eventually traversed") | |||
| 94 | + | } | |||
| 95 | + | ||||
| 49 | func TestRepositoryMaintainBoundsAndResumesDeepWideQuarantine(t *testing.T) { | 96 | func TestRepositoryMaintainBoundsAndResumesDeepWideQuarantine(t *testing.T) { | ||
| 50 | repo := NewRepository(privateDir(t)) | 97 | repo := NewRepository(privateDir(t)) | ||
| 51 | quarantine := filepath.Join(repo.dir, repositorySessionsDir, ".deleting-hostile") | 98 | quarantine := filepath.Join(repo.dir, repositorySessionsDir, ".deleting-hostile") | ||
M internal/adapters/snapshot/maintenance_test.go +15 -3
| @@ -178,10 +178,22 @@ func TestRepositoryMaintainRetainsNewestTwoCompleteGenerations(t *testing.T) { | |||
| 178 | 178 | t.Fatal(err) | |
| 179 | 179 | } | |
| 180 | 180 | } | |
| 181 | - | if err := repo.Maintain(context.Background()); err != nil { | |
| 182 | - | t.Fatal(err) | |
| 183 | - | } | |
| 184 | 181 | key := sessionKey("named") | |
| 182 | + | for pass := 0; pass < maintenanceBatch; pass++ { | |
| 183 | + | if err := repo.Maintain(context.Background()); err != nil { | |
| 184 | + | t.Fatal(err) | |
| 185 | + | } | |
| 186 | + | _, err := os.Lstat(repo.manifestPath(key, 1)) | |
| 187 | + | if errors.Is(err, os.ErrNotExist) { | |
| 188 | + | break | |
| 189 | + | } | |
| 190 | + | if err != nil { | |
| 191 | + | t.Fatalf("old manifest during Maintain = %v", err) | |
| 192 | + | } | |
| 193 | + | if pass == maintenanceBatch-1 { | |
| 194 | + | t.Fatal("old manifest was not eventually collected") | |
| 195 | + | } | |
| 196 | + | } | |
| 185 | 197 | for _, generation := range []uint64{2, 3} { | |
| 186 | 198 | if _, err := os.Lstat(repo.manifestPath(key, generation)); err != nil { | |
| 187 | 199 | t.Fatalf("retained manifest %d: %v", generation, err) | |
| @@ -178,10 +178,22 @@ func TestRepositoryMaintainRetainsNewestTwoCompleteGenerations(t *testing.T) { | |||||
| 178 | t.Fatal(err) | 178 | t.Fatal(err) | ||
| 179 | } | 179 | } | ||
| 180 | } | 180 | } | ||
| 181 | - | if err := repo.Maintain(context.Background()); err != nil { | |||
| 182 | - | t.Fatal(err) | |||
| 183 | - | } | |||
| 184 | key := sessionKey("named") | 181 | key := sessionKey("named") | ||
| 182 | + | for pass := 0; pass < maintenanceBatch; pass++ { | |||
| 183 | + | if err := repo.Maintain(context.Background()); err != nil { | |||
| 184 | + | t.Fatal(err) | |||
| 185 | + | } | |||
| 186 | + | _, err := os.Lstat(repo.manifestPath(key, 1)) | |||
| 187 | + | if errors.Is(err, os.ErrNotExist) { | |||
| 188 | + | break | |||
| 189 | + | } | |||
| 190 | + | if err != nil { | |||
| 191 | + | t.Fatalf("old manifest during Maintain = %v", err) | |||
| 192 | + | } | |||
| 193 | + | if pass == maintenanceBatch-1 { | |||
| 194 | + | t.Fatal("old manifest was not eventually collected") | |||
| 195 | + | } | |||
| 196 | + | } | |||
| 185 | for _, generation := range []uint64{2, 3} { | 197 | for _, generation := range []uint64{2, 3} { | ||
| 186 | if _, err := os.Lstat(repo.manifestPath(key, generation)); err != nil { | 198 | if _, err := os.Lstat(repo.manifestPath(key, generation)); err != nil { | ||
| 187 | t.Fatalf("retained manifest %d: %v", generation, err) | 199 | t.Fatalf("retained manifest %d: %v", generation, err) | ||