diff --git a/weed/storage/needle_map_sorted_file.go b/weed/storage/needle_map_sorted_file.go index 5a30525213..949ca91598 100644 --- a/weed/storage/needle_map_sorted_file.go +++ b/weed/storage/needle_map_sorted_file.go @@ -127,6 +127,9 @@ func (m *SortedFileNeedleMap) Delete(key NeedleId, offset Offset) error { return err } _, _, err = erasure_coding.SearchNeedleFromSortedIndex(f.file, m.dbFileSize, key, erasure_coding.MarkNeedleDeleted) + if err == nil { + m.logDelete(size) + } return err } diff --git a/weed/storage/needle_map_sorted_file_test.go b/weed/storage/needle_map_sorted_file_test.go index c9c1cc496f..e628ae4647 100644 --- a/weed/storage/needle_map_sorted_file_test.go +++ b/weed/storage/needle_map_sorted_file_test.go @@ -65,6 +65,21 @@ func TestSortedFileNeedleMap_DeleteAppendsTombstone(t *testing.T) { t.Fatalf("Delete: %v", err) } + if got, want := m.DeletedCount(), 1; got != want { + t.Fatalf("DeletedCount after delete: got %d, want %d", got, want) + } + if got, want := m.DeletedSize(), uint64(1024); got != want { + t.Fatalf("DeletedSize after delete: got %d, want %d", got, want) + } + + // Idempotent delete must not double count deletions. + if err := m.Delete(deletedKey, deletedOff); err != nil { + t.Fatalf("idempotent Delete: %v", err) + } + if got, want := m.DeletedCount(), 1; got != want { + t.Fatalf("DeletedCount after idempotent delete: got %d, want %d", got, want) + } + if got, want := fileSize(t, idxPath), wantIdxSize+int64(NeedleMapEntrySize); got != want { t.Fatalf("idx size after delete: got %d, want %d (front-overwrite regression)", got, want) } diff --git a/weed/storage/store_mark_readonly_can_delete_test.go b/weed/storage/store_mark_readonly_can_delete_test.go index 00376df8cf..40f536c732 100644 --- a/weed/storage/store_mark_readonly_can_delete_test.go +++ b/weed/storage/store_mark_readonly_can_delete_test.go @@ -152,3 +152,44 @@ func TestMountVolumeAnnouncesReadOnlyState(t *testing.T) { }) } } + +// After a restart a canDelete volume is served from .sdx; its deletes must +// still move the counters volume.deleteEmpty and vacuum read, and land them +// where the next reload would. +func TestMarkVolumeReadonlyCanDelete_CountsDeletesAfterRestart(t *testing.T) { + dir := t.TempDir() + store := newSingleDirStore(t, dir) + const vid = needle.VolumeId(13) + require.NoError(t, store.AddVolume(vid, "", NeedleMapInMemory, "000", "", 0, needle.GetCurrentVersion(), 0, types.HardDriveType, 0)) + + needles := []*needle.Needle{newRandomNeedle(1), newRandomNeedle(2), newRandomNeedle(3)} + for _, n := range needles { + _, err := store.WriteVolumeNeedle(vid, n, true, false) + require.NoError(t, err) + } + require.NoError(t, store.MarkVolumeReadonly(vid, true, true)) + store.Close() + + store2 := newSingleDirStore(t, dir) + v := store2.GetVolume(vid) + require.NotNil(t, v) + require.IsType(t, &SortedFileNeedleMap{}, v.nm) + for _, n := range needles { + _, err := store2.DeleteVolumeNeedle(vid, n) + require.NoError(t, err) + } + counters := func(v *Volume) [4]uint64 { + return [4]uint64{v.FileCount(), v.DeletedCount(), v.ContentSize(), v.DeletedSize()} + } + afterDeletes := counters(v) + require.Equal(t, uint64(len(needles)), afterDeletes[0]) + require.Equal(t, afterDeletes[0], afterDeletes[1], "an emptied volume must report every file deleted") + require.Equal(t, afterDeletes[2], afterDeletes[3]) + store2.Close() + + store3 := newSingleDirStore(t, dir) + v3 := store3.GetVolume(vid) + require.NotNil(t, v3) + require.Equal(t, afterDeletes, counters(v3)) + store3.Close() +}