this repo has no description
split up batches of deletes during compaction (#324)
turns out you cant pass infinitely sized arrays to your sql driver.
authored by
Whyrusleeping
and committed by
GitHub
e6006483
a6fcf91c
+60
-29
+60
-29
carstore/bs.go
+60
-29
carstore/bs.go
···
91
91
}
92
92
93
93
type staleRef struct {
94
-
ID uint `gorm:"primarykey"`
95
-
Cid models.DbCID `gorm:"index"`
96
-
Usr models.Uid
94
+
ID uint `gorm:"primarykey"`
95
+
Cid models.DbCID
96
+
Usr models.Uid `gorm:"index"`
97
97
}
98
98
99
99
type userView struct {
···
965
965
ctx, span := otel.Tracer("carstore").Start(ctx, "deleteShard")
966
966
defer span.End()
967
967
968
-
var ids []uint
969
-
for _, sh := range shs {
970
-
ids = append(ids, sh.ID)
971
-
}
968
+
deleteSlice := func(ctx context.Context, subs []*CarShard) error {
969
+
var ids []uint
970
+
for _, sh := range subs {
971
+
ids = append(ids, sh.ID)
972
+
}
972
973
973
-
txn := cs.meta.Begin()
974
+
txn := cs.meta.Begin()
974
975
975
-
if err := txn.Delete(&CarShard{}, "id in (?)", ids).Error; err != nil {
976
-
return err
977
-
}
976
+
if err := txn.Delete(&CarShard{}, "id in (?)", ids).Error; err != nil {
977
+
return err
978
+
}
978
979
979
-
if err := txn.Delete(&blockRef{}, "shard in (?)", ids).Error; err != nil {
980
-
return err
980
+
if err := txn.Delete(&blockRef{}, "shard in (?)", ids).Error; err != nil {
981
+
return err
982
+
}
983
+
984
+
if err := txn.Commit().Error; err != nil {
985
+
return err
986
+
}
987
+
988
+
var outErr error
989
+
for _, sh := range subs {
990
+
if err := cs.deleteShardFile(ctx, sh); err != nil {
991
+
if !os.IsNotExist(err) {
992
+
return err
993
+
}
994
+
outErr = err
995
+
}
996
+
}
997
+
998
+
return outErr
981
999
}
982
1000
983
-
if err := txn.Commit().Error; err != nil {
984
-
return err
985
-
}
1001
+
chunkSize := 100
1002
+
for i := 0; i < len(shs); i += chunkSize {
1003
+
sl := shs[i:]
1004
+
if len(sl) > chunkSize {
1005
+
sl = sl[:chunkSize]
1006
+
}
986
1007
987
-
var outErr error
988
-
for _, sh := range shs {
989
-
if err := cs.deleteShardFile(ctx, sh); err != nil {
990
-
if !os.IsNotExist(err) {
991
-
return err
992
-
}
993
-
outErr = err
1008
+
if err := deleteSlice(ctx, sl); err != nil {
1009
+
return err
994
1010
}
995
1011
}
996
1012
997
-
return outErr
1013
+
return nil
998
1014
}
999
1015
1000
1016
type shardStat struct {
···
1135
1151
span.SetAttributes(attribute.Int64("user", int64(user)))
1136
1152
1137
1153
var shards []CarShard
1138
-
if err := cs.meta.Find(&shards, "usr = ?", user).Error; err != nil {
1154
+
if err := cs.meta.WithContext(ctx).Find(&shards, "usr = ?", user).Error; err != nil {
1139
1155
return err
1140
1156
}
1141
1157
···
1150
1166
}
1151
1167
1152
1168
var brefs []blockRef
1153
-
if err := cs.meta.Debug().Raw(`select * from block_refs where shard in (?)`, shardIds).Scan(&brefs).Error; err != nil {
1169
+
if err := cs.meta.WithContext(ctx).Raw(`select * from block_refs where shard in (?)`, shardIds).Scan(&brefs).Error; err != nil {
1154
1170
return err
1155
1171
}
1156
1172
1157
1173
var staleRefs []staleRef
1158
-
if err := cs.meta.Debug().Find(&staleRefs, "usr = ?", user).Error; err != nil {
1174
+
if err := cs.meta.WithContext(ctx).Find(&staleRefs, "usr = ?", user).Error; err != nil {
1159
1175
return err
1160
1176
}
1161
1177
···
1244
1260
// now we need to delete the staleRefs we successfully cleaned up
1245
1261
// we can delete a staleRef if all the shards that have blockRefs with matching stale refs were processed
1246
1262
1263
+
return cs.deleteStaleRefs(ctx, brefs, staleRefs, removedShards)
1264
+
}
1265
+
1266
+
func (cs *CarStore) deleteStaleRefs(ctx context.Context, brefs []blockRef, staleRefs []staleRef, removedShards map[uint]bool) error {
1267
+
ctx, span := otel.Tracer("carstore").Start(ctx, "deleteStaleRefs")
1268
+
defer span.End()
1269
+
1247
1270
brByCid := make(map[cid.Cid][]blockRef)
1248
1271
for _, br := range brefs {
1249
1272
brByCid[br.Cid.CID] = append(brByCid[br.Cid.CID], br)
···
1265
1288
}
1266
1289
}
1267
1290
1268
-
if err := cs.meta.Delete(&staleRef{}, "id in (?)", staleToDelete).Error; err != nil {
1269
-
return err
1291
+
chunkSize := 100
1292
+
for i := 0; i < len(staleToDelete); i += chunkSize {
1293
+
sl := staleToDelete[i:]
1294
+
if len(sl) > chunkSize {
1295
+
sl = sl[:chunkSize]
1296
+
}
1297
+
1298
+
if err := cs.meta.Delete(&staleRef{}, "id in (?)", sl).Error; err != nil {
1299
+
return err
1300
+
}
1270
1301
}
1271
1302
1272
1303
return nil