volume.fsck: batch and deduplicate filer entry deletions (#12480)

* volume.fsck: batch and deduplicate filer entry deletions In volume.fsck with -findMissingChunksInFiler and -reallyDeleteFilerEntries, deleting orphaned filer entries was slow and CPU intensive due to tiny worker channel buffers (128), lack of batch processing, redundant HTTP DELETE requests for multi-chunk files, and lock contention on parent directory tracking. Optimize by: - Adding startPurgeBatchWorkers with 4096-buffered channels to prevent producer stalls and batch opportunistic deletes up to 64 items. - Deduplicating missing chunks per file path during volume checks so that multi-chunk files are queued once. - Processing batches with in-batch path deduplication and aggregating parent directories before updating purgedDirs in a single lock acquisition. - Buffering temporary fid file writes (64KB) during the traversal to avoid millions of unbuffered syscalls. * volume.fsck: report a failed write of the filer fid files The fid files are written through a buffer now, so the last part of each one reaches the disk only in the deferred Flush. Its error was dropped: a full temp disk left a short fid file and the collection still succeeded, and the volume check would then take the chunks missing from it for orphans. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> * volume.fsck: retry a batched path whose earlier delete failed A path was marked done before its delete ran, so a second copy of it in the same batch was skipped after a transient failure and the entry stayed. Only skip paths already deleted. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> * volume.fsck: keep reporting a failed DELETE response read The non-verbose path now discards the body without checking the copy, so a connection cut after the headers went unreported. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> * volume.fsck: keep one purge worker starter and one delete function startPurgeWorkers and the single-path httpDelete were left only as wrappers the tests called. Give their names to the batch versions and adapt the tests. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> --------- Co-authored-by: Chris Lu <chrislusf@users.noreply.github.com> Co-authored-by: Claude Opus 5.5 <noreply@anthropic.com>
2026-11-08 21:08:27 +01:00 · 2026-11-08 05:03:52 -08:00
1 parent 73fa85f54f
commit f99cc6ca94
@@ -323,12 +323,23 @@ func (c *commandVolumeFsck) Do(args []string, commandEnv *CommandEnv, writer io.
 	return nil
 }

-func (c *commandVolumeFsck) collectFilerFileIdAndPaths(dataNodeVolumeIdToVInfo map[string]map[uint32]VInfo, purgeAbsent bool, queuePurge func(purgeItem), collectModifyFromAtNs int64, cutoffFromAtNs int64) error {
+func (c *commandVolumeFsck) collectFilerFileIdAndPaths(dataNodeVolumeIdToVInfo map[string]map[uint32]VInfo, purgeAbsent bool, queuePurge func(purgeItem), collectModifyFromAtNs int64, cutoffFromAtNs int64) (err error) {
 	if *c.verbose {
 		fmt.Fprintf(c.writer, "checking each file from filer path /...\n")
 	}

 	files := make(map[uint32]*os.File)
+	writers := make(map[uint32]*bufio.Writer)
+	defer func() {
+		for vid, f := range files {
+			if flushErr := writers[vid].Flush(); flushErr != nil && err == nil {
+				err = fmt.Errorf("failed to write file %s: %w", f.Name(), flushErr)
+			}
+			if closeErr := f.Close(); closeErr != nil && err == nil {
+				err = fmt.Errorf("failed to close file %s: %w", f.Name(), closeErr)
+			}
+		}
+	}()
 	for _, volumeIdToServer := range dataNodeVolumeIdToVInfo {
 		for vid := range volumeIdToServer {
 			if _, ok := files[vid]; ok {
@@ -339,13 +350,9 @@ func (c *commandVolumeFsck) collectFilerFileIdAndPaths(dataNodeVolumeIdToVInfo m
 				return fmt.Errorf("failed to create file %s: %v", getFilerFileIdFile(c.tempFolder, vid), openErr)
 			}
 			files[vid] = dst
+			writers[vid] = bufio.NewWriterSize(dst, 64*1024)
 		}
 	}
-	defer func() {
-		for _, f := range files {
-			f.Close()
-		}
-	}()

 	return doTraverseAndSaving(c.env, c.writer, "/", false, false,
 		func(ctx context.Context, entry *filer_pb.FullEntry, outputChan chan interface{}) (err error) {
@@ -415,21 +422,21 @@ func (c *commandVolumeFsck) collectFilerFileIdAndPaths(dataNodeVolumeIdToVInfo m
 			buffer := make([]byte, readbufferSize)
 			for item := range outputChan {
 				i := item.(*Item)
-				if f, ok := files[i.vid]; ok {
+				if w, ok := writers[i.vid]; ok {
 					util.Uint64toBytes(buffer, i.fileKey)
 					util.Uint32toBytes(buffer[8:], i.cookie)
 					util.Uint32toBytes(buffer[12:], uint32(len(i.path)))
-					if _, err := f.Write(buffer); err != nil {
+					if _, err := w.Write(buffer); err != nil {
 						return err
 					}
-					if _, err := f.Write([]byte(i.path)); err != nil {
+					if _, err := w.Write([]byte(i.path)); err != nil {
 						return err
 					}
 					util.Uint32toBytes(buffer[:4], uint32(len(i.hardLinkId)))
-					if _, err := f.Write(buffer[:4]); err != nil {
+					if _, err := w.Write(buffer[:4]); err != nil {
 						return err
 					}
-					if _, err := f.Write(i.hardLinkId); err != nil {
+					if _, err := w.Write(i.hardLinkId); err != nil {
 						return err
 					}
 				} else if *c.findMissingChunksInFiler {
@@ -690,8 +697,9 @@ func (c *commandVolumeFsck) readFilerFileIdFile(volumeId uint32, fn func(needleI
 	}
 	defer fp.Close()

-	br := bufio.NewReader(fp)
+	br := bufio.NewReaderSize(fp, 64*1024)
 	buffer := make([]byte, readbufferSize)
+	pathBuf := make([]byte, 512)
 	var readSize int
 	var readErr error
 	item := &Item{vid: volumeId}
@@ -709,15 +717,17 @@ func (c *commandVolumeFsck) readFilerFileIdFile(volumeId uint32, fn func(needleI
 		item.fileKey = util.BytesToUint64(buffer[:8])
 		item.cookie = util.BytesToUint32(buffer[8:12])
 		pathSize := util.BytesToUint32(buffer[12:16])
-		pathBytes := make([]byte, int(pathSize))
-		n, err := io.ReadFull(br, pathBytes)
+		if int(pathSize) > len(pathBuf) {
+			pathBuf = make([]byte, pathSize)
+		}
+		n, err := io.ReadFull(br, pathBuf[:pathSize])
 		if err != nil {
 			return fmt.Errorf("read fid path for volume %d,%x%08x: %w", volumeId, item.fileKey, item.cookie, err)
 		}
 		if n != int(pathSize) {
 			return fmt.Errorf("read fid path size mismatch for volume %d,%x%08x: got %d want %d", volumeId, item.fileKey, item.cookie, n, pathSize)
 		}
-		item.path = util.FullPath(pathBytes)
+		item.path = util.FullPath(string(pathBuf[:pathSize]))
 		item.hardLinkId = nil
 		if _, err := io.ReadFull(br, buffer[:4]); err != nil {
 			return fmt.Errorf("read fid hard link id size for volume %d,%x%08x: %w", volumeId, item.fileKey, item.cookie, err)
@@ -745,8 +755,13 @@ func (c *commandVolumeFsck) oneVolumeFileIdsCheckOneVolume(dataNodeId string, vo
 	if err = db.LoadFromIdx(getVolumeFileIdFile(c.tempFolder, dataNodeId, volumeId)); err != nil {
 		return
 	}
+	seenPaths := make(map[util.FullPath]struct{})
 	if err = c.readFilerFileIdFile(volumeId, func(needleId types.NeedleId, itemPath util.FullPath, hardLinkId []byte) {
 		if _, found := db.Get(needleId); !found {
+			if _, seen := seenPaths[itemPath]; seen {
+				return
+			}
+			seenPaths[itemPath] = struct{}{}
 			fmt.Fprintf(c.writer, "%s\n", itemPath)
 			if queuePurge != nil {
 				queuePurge(purgeItem{path: itemPath, hardLinkId: hardLinkId})
@@ -763,25 +778,44 @@ type purgeItem struct {
 	hardLinkId []byte
 }

-// startPurgeWorkers fans deletes across at least one worker and returns a
-// queue function plus a finish function that closes the channels and waits
-// for in-flight deletes. Paths that share a hard link id always land on the
-// same worker, so their shared hard-link counter updates stay serialized —
-// concurrent deletes of the same id could both read count N and write N-1,
-// leaving the shared record behind after every path is gone.
-func startPurgeWorkers(workers int, deleteFn func(util.FullPath)) (queue func(purgeItem), finish func()) {
+const (
+	purgeBatchSize         = 64
+	purgeChannelBufferSize = 4096
+)
+
+// startPurgeWorkers fans deletes across at least one worker, handing each
+// one batches of the paths already queued for it, and returns a queue
+// function plus a finish function that closes the channels and waits for
+// in-flight deletes. Paths that share a hard link id always land on the
+// same worker, so their shared hard-link counter updates stay serialized.
+func startPurgeWorkers(workers int, deleteFn func([]purgeItem)) (queue func(purgeItem), finish func()) {
 	if workers < 1 {
 		workers = 1
 	}
 	chans := make([]chan purgeItem, workers)
 	var wg sync.WaitGroup
 	for w := range chans {
-		chans[w] = make(chan purgeItem, 128)
+		chans[w] = make(chan purgeItem, purgeChannelBufferSize)
 		wg.Add(1)
 		go func(ch <-chan purgeItem) {
 			defer wg.Done()
+			batch := make([]purgeItem, 0, purgeBatchSize)
 			for it := range ch {
-				deleteFn(it.path)
+				batch = append(batch, it)
+			drain:
+				for len(batch) < purgeBatchSize {
+					select {
+					case next, ok := <-ch:
+						if !ok {
+							break drain
+						}
+						batch = append(batch, next)
+					default:
+						break drain
+					}
+				}
+				deleteFn(batch)
+				batch = batch[:0]
 			}
 		}(chans[w])
 	}
@@ -799,11 +833,38 @@ func startPurgeWorkers(workers int, deleteFn func(util.FullPath)) (queue func(pu
 		}
 }

-func (c *commandVolumeFsck) httpDelete(path util.FullPath) {
+func (c *commandVolumeFsck) httpDelete(items []purgeItem) {
+	if len(items) == 0 {
+		return
+	}
+	deleted := make(map[util.FullPath]struct{}, len(items))
+	localPurgedDirs := make(map[util.FullPath]struct{})
+
+	for _, it := range items {
+		if _, ok := deleted[it.path]; ok {
+			continue
+		}
+		if c.doHttpDelete(it.path) {
+			deleted[it.path] = struct{}{}
+			dir, _ := it.path.DirAndName()
+			localPurgedDirs[util.FullPath(dir)] = struct{}{}
+		}
+	}
+
+	if len(localPurgedDirs) > 0 {
+		c.purgedDirsLock.Lock()
+		for dir := range localPurgedDirs {
+			c.purgedDirs[dir] = struct{}{}
+		}
+		c.purgedDirsLock.Unlock()
+	}
+}
+
+func (c *commandVolumeFsck) doHttpDelete(path util.FullPath) bool {
 	req, err := http.NewRequest(http.MethodDelete, "", nil)
 	if err != nil {
 		fmt.Fprintf(c.writer, "HTTP delete request error: %v\n", err)
-		return
+		return false
 	}

 	req.URL = &url.URL{
@@ -824,28 +885,29 @@ func (c *commandVolumeFsck) httpDelete(path util.FullPath) {
 	resp, err := util_http.GetGlobalHttpClient().Do(req)
 	if err != nil {
 		fmt.Fprintf(c.writer, "DELETE fetch error: %v\n", err)
-		return
+		return false
 	}
 	defer resp.Body.Close()

-	_, err = io.ReadAll(resp.Body)
-	if err != nil {
-		fmt.Fprintf(c.writer, "DELETE response error: %v\n", err)
-	}
-
 	if *c.verbose {
+		body, err := io.ReadAll(resp.Body)
+		if err != nil {
+			fmt.Fprintf(c.writer, "DELETE response error: %v\n", err)
+		}
 		fmt.Fprintln(c.writer, "delete response Status : ", resp.Status)
 		fmt.Fprintln(c.writer, "delete response Headers : ", resp.Header)
+		if len(body) > 0 {
+			fmt.Fprintln(c.writer, "delete response Body : ", string(body))
+		}
+	} else if _, err := io.Copy(io.Discard, resp.Body); err != nil {
+		fmt.Fprintf(c.writer, "DELETE response error: %v\n", err)
 	}

 	if resp.StatusCode >= http.StatusMultipleChoices {
 		fmt.Fprintf(c.writer, "DELETE %s failed: %s\n", path, resp.Status)
-	} else {
-		dir, _ := path.DirAndName()
-		c.purgedDirsLock.Lock()
-		c.purgedDirs[util.FullPath(dir)] = struct{}{}
-		c.purgedDirsLock.Unlock()
+		return false
 	}
+	return true
 }

 // purgeEmptyDirectories removes the directories emptied by the purged entries, walking up while each parent is empty too.
@@ -64,8 +64,7 @@ func TestVolumeFsckHttpDeleteRecordsParentDirectory(t *testing.T) {
 		env:        &CommandEnv{option: &ShellOptions{FilerAddress: pb.ServerAddress(strings.TrimPrefix(filer.URL, "http://"))}},
 	}

-	c.httpDelete("/orphan/dir/deep/file.txt")
-	c.httpDelete("/locked/dir/file.txt")
+	c.httpDelete([]purgeItem{{path: "/orphan/dir/deep/file.txt"}, {path: "/locked/dir/file.txt"}})

 	if _, found := c.purgedDirs["/orphan/dir/deep"]; !found {
 		t.Errorf("expected /orphan/dir/deep to be recorded, got %v", c.purgedDirs)
@@ -267,10 +266,18 @@ func TestVolumeFsckPurgeEmptyDirectoriesKeepsChangedDirectory(t *testing.T) {

 // The purge pool may run at most `workers` deletes at once, and at least one
 // even when the flag is set to zero.
+func eachPurgePath(deleteFn func(util.FullPath)) func([]purgeItem) {
+	return func(batch []purgeItem) {
+		for _, it := range batch {
+			deleteFn(it.path)
+		}
+	}
+}
+
 func TestVolumeFsckPurgeWorkerLimit(t *testing.T) {
 	gate := make(chan struct{})
 	var inFlight, maxSeen, processed atomic.Int32
-	queue, finish := startPurgeWorkers(3, func(util.FullPath) {
+	queue, finish := startPurgeWorkers(3, eachPurgePath(func(util.FullPath) {
 		n := inFlight.Add(1)
 		for {
 			m := maxSeen.Load()
@@ -281,7 +288,7 @@ func TestVolumeFsckPurgeWorkerLimit(t *testing.T) {
 		<-gate
 		inFlight.Add(-1)
 		processed.Add(1)
-	})
+	}))

 	for i := 0; i < 10; i++ {
 		queue(purgeItem{path: util.FullPath(fmt.Sprintf("/p/%d", i))})
@@ -306,7 +313,7 @@ func TestVolumeFsckPurgeWorkerLimit(t *testing.T) {

 func TestVolumeFsckPurgeWorkersAtLeastOne(t *testing.T) {
 	var processed atomic.Int32
-	queue, finish := startPurgeWorkers(0, func(util.FullPath) { processed.Add(1) })
+	queue, finish := startPurgeWorkers(0, eachPurgePath(func(util.FullPath) { processed.Add(1) }))
 	for i := 0; i < 5; i++ {
 		queue(purgeItem{path: util.FullPath(fmt.Sprintf("/p/%d", i))})
 	}
@@ -322,7 +329,7 @@ func TestVolumeFsckPurgeWorkersAtLeastOne(t *testing.T) {
 func TestVolumeFsckPurgeHardLinksSerialized(t *testing.T) {
 	var inFlight, maxSeen, processed atomic.Int32
 	gate := make(chan struct{})
-	queue, finish := startPurgeWorkers(8, func(util.FullPath) {
+	queue, finish := startPurgeWorkers(8, eachPurgePath(func(util.FullPath) {
 		n := inFlight.Add(1)
 		for {
 			m := maxSeen.Load()
@@ -333,7 +340,7 @@ func TestVolumeFsckPurgeHardLinksSerialized(t *testing.T) {
 		<-gate
 		inFlight.Add(-1)
 		processed.Add(1)
-	})
+	}))
 	for i := 0; i < 8; i++ {
 		queue(purgeItem{path: util.FullPath(fmt.Sprintf("/hl/%d", i)), hardLinkId: []byte{9, 9, 9}})
 	}
@@ -514,3 +521,183 @@ func TestReadFilerFileIdFileRejectsTruncatedHardLinkId(t *testing.T) {
 		t.Fatal("expected an error for a truncated hard link id size")
 	}
 }
+
+func TestVolumeFsckPurgeBatchWorkersProcessesBatches(t *testing.T) {
+	var totalProcessed atomic.Int32
+	var maxBatchSeen atomic.Int32
+	var batchCount atomic.Int32
+
+	batchGate := make(chan struct{})
+	queue, finish := startPurgeWorkers(1, func(batch []purgeItem) {
+		<-batchGate
+		count := int32(len(batch))
+		totalProcessed.Add(count)
+		batchCount.Add(1)
+		for {
+			cur := maxBatchSeen.Load()
+			if count <= cur || maxBatchSeen.CompareAndSwap(cur, count) {
+				break
+			}
+		}
+	})
+
+	// Queue 10 items while worker is gated
+	for i := 0; i < 10; i++ {
+		queue(purgeItem{path: util.FullPath(fmt.Sprintf("/item/%d", i))})
+	}
+	close(batchGate)
+	finish()
+
+	if got := totalProcessed.Load(); got != 10 {
+		t.Fatalf("processed %d items, want 10", got)
+	}
+	if max := maxBatchSeen.Load(); max < 2 {
+		t.Fatalf("expected batch size >= 2 when draining queued items, got %d", max)
+	}
+}
+
+func TestVolumeFsckCheckOneVolumeDeduplicatesMissingChunksForSamePath(t *testing.T) {
+	tempDir := t.TempDir()
+	var verbose bool
+	var buf bytes.Buffer
+	c := &commandVolumeFsck{tempFolder: tempDir, verbose: &verbose, writer: &buf}
+
+	volumeId := uint32(8)
+	db := needle_map.NewMemDb()
+	if err := db.SaveToIdx(getVolumeFileIdFile(tempDir, "dn1", volumeId)); err != nil {
+		t.Fatalf("write empty volume idx: %v", err)
+	}
+	db.Close()
+
+	f, err := os.Create(getFilerFileIdFile(tempDir, volumeId))
+	if err != nil {
+		t.Fatalf("create fid file: %v", err)
+	}
+	writeFid := func(fileKey uint64, cookie uint32, p string, hardLinkId []byte) {
+		buffer := make([]byte, readbufferSize)
+		util.Uint64toBytes(buffer, fileKey)
+		util.Uint32toBytes(buffer[8:], cookie)
+		util.Uint32toBytes(buffer[12:], uint32(len(p)))
+		if _, err := f.Write(buffer); err != nil {
+			t.Fatalf("write fid header: %v", err)
+		}
+		if _, err := f.Write([]byte(p)); err != nil {
+			t.Fatalf("write fid path: %v", err)
+		}
+		util.Uint32toBytes(buffer[:4], uint32(len(hardLinkId)))
+		if _, err := f.Write(buffer[:4]); err != nil {
+			t.Fatalf("write fid hard link id size: %v", err)
+		}
+		if _, err := f.Write(hardLinkId); err != nil {
+			t.Fatalf("write fid hard link id: %v", err)
+		}
+	}
+
+	// Two chunks for the same file, both missing from volume idx
+	writeFid(101, 1, "/orphan/same_file.txt", nil)
+	writeFid(102, 1, "/orphan/same_file.txt", nil)
+	// Another file
+	writeFid(103, 1, "/orphan/another_file.txt", nil)
+	if err := f.Close(); err != nil {
+		t.Fatalf("close fid file: %v", err)
+	}
+
+	var queued []purgeItem
+	if err := c.oneVolumeFileIdsCheckOneVolume("dn1", volumeId, func(it purgeItem) {
+		queued = append(queued, it)
+	}); err != nil {
+		t.Fatalf("check volume: %v", err)
+	}
+
+	if len(queued) != 2 {
+		t.Fatalf("queued %d purges, want 2 (deduplicated): %v", len(queued), queued)
+	}
+	if queued[0].path != "/orphan/same_file.txt" || queued[1].path != "/orphan/another_file.txt" {
+		t.Fatalf("queued paths %v, want /orphan/same_file.txt and /orphan/another_file.txt", queued)
+	}
+}
+
+func TestVolumeFsckHttpDeleteBatchDeduplicatesAndRecordsDirs(t *testing.T) {
+	var requestCount, flakyCount atomic.Int32
+	filer := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		requestCount.Add(1)
+		if strings.HasPrefix(r.URL.Path, "/fail/") {
+			w.WriteHeader(http.StatusNotFound)
+			return
+		}
+		if strings.HasPrefix(r.URL.Path, "/flaky/") && flakyCount.Add(1) == 1 {
+			w.WriteHeader(http.StatusServiceUnavailable)
+			return
+		}
+		w.WriteHeader(http.StatusNoContent)
+	}))
+	defer filer.Close()
+
+	verbose := false
+	var output bytes.Buffer
+	c := &commandVolumeFsck{
+		verbose:    &verbose,
+		writer:     &output,
+		purgedDirs: make(map[util.FullPath]struct{}),
+		env:        &CommandEnv{option: &ShellOptions{FilerAddress: pb.ServerAddress(strings.TrimPrefix(filer.URL, "http://"))}},
+	}
+
+	items := []purgeItem{
+		{path: "/dir1/file1.txt"},
+		{path: "/dir1/file1.txt"}, // duplicate within batch
+		{path: "/dir2/file2.txt"},
+		{path: "/fail/file3.txt"},
+		{path: "/flaky/file4.txt"},
+		{path: "/flaky/file4.txt"}, // retried after the first delete failed
+	}
+
+	c.httpDelete(items)
+
+	if got := requestCount.Load(); got != 5 {
+		t.Fatalf("received %d requests, want 5", got)
+	}
+	if _, found := c.purgedDirs["/flaky"]; !found {
+		t.Errorf("expected /flaky to be recorded after the retried delete")
+	}
+	if _, found := c.purgedDirs["/dir1"]; !found {
+		t.Errorf("expected /dir1 to be recorded in purgedDirs")
+	}
+	if _, found := c.purgedDirs["/dir2"]; !found {
+		t.Errorf("expected /dir2 to be recorded in purgedDirs")
+	}
+	if _, found := c.purgedDirs["/fail"]; found {
+		t.Errorf("expected /fail not to be recorded in purgedDirs on 404")
+	}
+}
+
+func TestVolumeFsckCollectFilerFileIdsReportsFailedFlush(t *testing.T) {
+	if _, err := os.Stat("/dev/full"); err != nil {
+		t.Skip("needs /dev/full")
+	}
+	tempDir := t.TempDir()
+	if err := os.Symlink("/dev/full", getFilerFileIdFile(tempDir, 7)); err != nil {
+		t.Fatalf("link fid file to /dev/full: %v", err)
+	}
+	directory := &filer_pb.Entry{IsDirectory: true}
+	stub := &stubFilerServer{entries: map[string]*filer_pb.Entry{
+		"/":     directory,
+		"/file": {Chunks: []*filer_pb.FileChunk{{Size: 1, Fid: &filer_pb.FileId{VolumeId: 7, FileKey: 11, Cookie: 1}}}},
+	}}
+
+	verbose, findMissing, verify, force := false, false, false, false
+	c := &commandVolumeFsck{
+		tempFolder:               tempDir,
+		verbose:                  &verbose,
+		findMissingChunksInFiler: &findMissing,
+		verifyNeedle:             &verify,
+		forcePurging:             &force,
+		writer:                   io.Discard,
+		bucketsPath:              "/buckets",
+		env:                      startStubFiler(t, stub),
+	}
+
+	vinfo := map[string]map[uint32]VInfo{"dn1": {7: {}}}
+	if err := c.collectFilerFileIdAndPaths(vinfo, false, func(purgeItem) {}, 0, 0); err == nil {
+		t.Fatal("expected the failed fid file write to be reported")
+	}
+}