feat(catalog): add deleteInBatches loop helper
This commit is contained in:
@@ -90,3 +90,49 @@ func TestRetryOnDeadlockStopsOnCanceledContext(t *testing.T) {
|
||||
t.Fatalf("expected 1 call before cancel, got %d", calls)
|
||||
}
|
||||
}
|
||||
|
||||
func TestDeleteInBatchesLoopsUntilUnderBatchSize(t *testing.T) {
|
||||
withFastDeadlockRetry(t, 5)
|
||||
counts := []int64{5, 5, 2}
|
||||
idx := 0
|
||||
total, err := deleteInBatches(context.Background(), 5, func(context.Context) (int64, error) {
|
||||
n := counts[idx]
|
||||
idx++
|
||||
return n, nil
|
||||
})
|
||||
if err != nil {
|
||||
t.Fatalf("unexpected error: %v", err)
|
||||
}
|
||||
if total != 12 {
|
||||
t.Fatalf("expected total 12, got %d", total)
|
||||
}
|
||||
if idx != 3 {
|
||||
t.Fatalf("expected 3 batches, got %d", idx)
|
||||
}
|
||||
}
|
||||
|
||||
func TestDeleteInBatchesStopsImmediatelyWhenFirstBatchUnderSize(t *testing.T) {
|
||||
withFastDeadlockRetry(t, 5)
|
||||
calls := 0
|
||||
total, err := deleteInBatches(context.Background(), 5, func(context.Context) (int64, error) {
|
||||
calls++
|
||||
return 0, nil
|
||||
})
|
||||
if err != nil {
|
||||
t.Fatalf("unexpected error: %v", err)
|
||||
}
|
||||
if total != 0 || calls != 1 {
|
||||
t.Fatalf("expected total 0 and 1 call, got total=%d calls=%d", total, calls)
|
||||
}
|
||||
}
|
||||
|
||||
func TestDeleteInBatchesReturnsError(t *testing.T) {
|
||||
withFastDeadlockRetry(t, 5)
|
||||
sentinel := errors.New("delete failed")
|
||||
_, err := deleteInBatches(context.Background(), 5, func(context.Context) (int64, error) {
|
||||
return 0, sentinel
|
||||
})
|
||||
if !errors.Is(err, sentinel) {
|
||||
t.Fatalf("expected sentinel, got %v", err)
|
||||
}
|
||||
}
|
||||
|
||||
@@ -48,6 +48,31 @@ func retryOnDeadlock(ctx context.Context, op func() error) error {
|
||||
}
|
||||
}
|
||||
|
||||
// deleteInBatches repeatedly runs deleteBatch (each a single autocommit
|
||||
// statement) until a batch removes fewer than batchSize rows. Each batch is
|
||||
// retried on deadlock. It returns the total number of rows deleted.
|
||||
func deleteInBatches(
|
||||
ctx context.Context,
|
||||
batchSize int,
|
||||
deleteBatch func(ctx context.Context) (int64, error),
|
||||
) (int64, error) {
|
||||
var total int64
|
||||
for {
|
||||
var affected int64
|
||||
if err := retryOnDeadlock(ctx, func() error {
|
||||
n, e := deleteBatch(ctx)
|
||||
affected = n
|
||||
return e
|
||||
}); err != nil {
|
||||
return total, err
|
||||
}
|
||||
total += affected
|
||||
if affected < int64(batchSize) {
|
||||
return total, nil
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// Sentinel errors for folder repository operations.
|
||||
var (
|
||||
ErrFolderNotFound = errors.New("folder not found")
|
||||
|
||||
Reference in New Issue
Block a user