+18 -4

bgs/admin.go

reviewed

··· 426 426 ctx, span := otel.Tracer("bgs").Start(context.Background(), "adminCompactAllRepos") 427 427 defer span.End() 428 428 429 429 - if err := bgs.runRepoCompaction(ctx); err != nil { 429 429 + var dry bool 430 430 + if strings.ToLower(e.QueryParam("dry")) == "true" { 431 431 + dry = true 432 432 + } 433 433 + 434 434 + lim := 50 435 435 + if limstr := e.QueryParam("limit"); limstr != "" { 436 436 + v, err := strconv.Atoi(limstr) 437 437 + if err != nil { 438 438 + return err 439 439 + } 440 440 + 441 441 + lim = v 442 442 + } 443 443 + 444 444 + stats, err := bgs.runRepoCompaction(ctx, lim, dry) 445 445 + if err != nil { 430 446 return fmt.Errorf("compaction run failed: %w", err) 431 447 } 432 448 433 433 - return e.JSON(200, map[string]any{ 434 434 - "success": "true", 435 435 - }) 449 449 + return e.JSON(200, stats) 436 450 } 437 451 438 452 func (bgs *BGS) handleAdminPostResyncPDS(e echo.Context) error {

+26 -5

bgs/bgs.go

reviewed

··· 1127 1127 return nil 1128 1128 } 1129 1129 1130 1130 - func (bgs *BGS) runRepoCompaction(ctx context.Context) error { 1130 1130 + type compactionStats struct { 1131 1131 + Completed map[models.Uid]*carstore.CompactionStats 1132 1132 + Targets []carstore.CompactionTarget 1133 1133 + } 1134 1134 + 1135 1135 + func (bgs *BGS) runRepoCompaction(ctx context.Context, lim int, dry bool) (*compactionStats, error) { 1131 1136 ctx, span := otel.Tracer("bgs").Start(ctx, "runRepoCompaction") 1132 1137 defer span.End() 1133 1138 1134 1134 - repos, err := bgs.repoman.CarStore().GetCompactionTargets(ctx) 1139 1139 + repos, err := bgs.repoman.CarStore().GetCompactionTargets(ctx, 50) 1135 1140 if err != nil { 1136 1136 - return fmt.Errorf("failed to get repos to compact: %w", err) 1141 1141 + return nil, fmt.Errorf("failed to get repos to compact: %w", err) 1142 1142 + } 1143 1143 + 1144 1144 + if lim > 0 && len(repos) > lim { 1145 1145 + repos = repos[:lim] 1146 1146 + } 1147 1147 + 1148 1148 + if dry { 1149 1149 + return &compactionStats{ 1150 1150 + Targets: repos, 1151 1151 + }, nil 1137 1152 } 1138 1153 1154 1154 + results := make(map[models.Uid]*carstore.CompactionStats) 1139 1155 for _, r := range repos { 1140 1140 - if _, err := bgs.repoman.CarStore().CompactUserShards(ctx, r.Usr); err != nil { 1156 1156 + st, err := bgs.repoman.CarStore().CompactUserShards(ctx, r.Usr) 1157 1157 + if err != nil { 1141 1158 log.Errorf("failed to compact shards for user %d: %s", r.Usr, err) 1142 1159 continue 1143 1160 } 1161 1161 + results[r.Usr] = st 1144 1162 } 1145 1163 1146 1146 - return nil 1164 1164 + return &compactionStats{ 1165 1165 + Targets: repos, 1166 1166 + Completed: results, 1167 1167 + }, nil 1147 1168 } 1148 1169 1149 1170 type repoHead struct {

+6 -4

carstore/bs.go

reviewed

··· 727 727 ctx, span := otel.Tracer("carstore").Start(ctx, "createBlockRefs") 728 728 defer span.End() 729 729 730 730 - if err := createInBatches(ctx, tx, brefs, 100); err != nil { 730 730 + if err := createInBatches(ctx, tx, brefs, 1000); err != nil { 731 731 return err 732 732 } 733 733 ··· 1150 1150 NumShards int 1151 1151 } 1152 1152 1153 1153 - func (cs *CarStore) GetCompactionTargets(ctx context.Context) ([]CompactionTarget, error) { 1153 1153 + func (cs *CarStore) GetCompactionTargets(ctx context.Context, shardCount int) ([]CompactionTarget, error) { 1154 1154 ctx, span := otel.Tracer("carstore").Start(ctx, "GetCompactionTargets") 1155 1155 defer span.End() 1156 1156 1157 1157 var targets []CompactionTarget 1158 1158 - if err := cs.meta.Raw(`select usr, count(*) as num_shards from car_shards group by usr having count(*) > 50 order by num_shards desc`).Scan(&targets).Error; err != nil { 1158 1158 + if err := cs.meta.Raw(`select usr, count(*) as num_shards from car_shards group by usr having count(*) > ? order by num_shards desc`, shardCount).Scan(&targets).Error; err != nil { 1159 1159 return nil, err 1160 1160 } 1161 1161 ··· 1186 1186 } 1187 1187 1188 1188 type CompactionStats struct { 1189 1189 + TotalRefs int `json:"totalRefs"` 1189 1190 StartShards int `json:"startShards"` 1190 1191 NewShards int `json:"newShards"` 1191 1192 SkippedShards int `json:"skippedShards"` ··· 1326 1327 1327 1328 stats := &CompactionStats{ 1328 1329 StartShards: len(shards), 1330 1330 + TotalRefs: len(brefs), 1329 1331 } 1330 1332 1331 1333 removedShards := make(map[uint]bool) ··· 1397 1399 } 1398 1400 } 1399 1401 1400 1400 - chunkSize := 100 1402 1402 + chunkSize := 500 1401 1403 for i := 0; i < len(staleToDelete); i += chunkSize { 1402 1404 sl := staleToDelete[i:] 1403 1405 if len(sl) > chunkSize {