[Disk Manager] Fix style in filesystem traversal mechanism (#5142)

Author: jkuradobery

cloud/disk_manager/internal/pkg/dataplane/filesystem_snapshot/filesystem_traversal/traversal.go

@@ -94,17 +94,17 @@ func (t *FilesystemTraverser) Traverse(
 		logging.String("FILESYSTEM_ID", t.filesystemID),
 		logging.String("FILESYSTEM_CHECKPOINT_ID", t.filesystemCheckpointID),
 	)
+	defer close(t.processedNodeIDs)
+
 	if !t.rootNodeAlreadyScheduled {
 		err := t.storage.ScheduleRootNodeForListing(ctx, t.filesystemSnapshotID)
 		if err != nil {
 			return err
 		}
 
-		if t.stateSaver != nil {
-			err = t.stateSaver(ctx)
-			if err != nil {
-				return err
-			}
+		err = t.stateSaver(ctx)
+		if err != nil {
+			return err
 		}
 	}
 
@@ -126,7 +126,7 @@ func (t *FilesystemTraverser) Traverse(
 func (t *FilesystemTraverser) directoryScheduler(ctx context.Context) error {
 	defer close(t.scheduledNodes)
 
-	processedNodeIDs := make(map[uint64]struct{})
+	processingNodeIDs := make(map[uint64]struct{})
 	pendingNodes := make([]*storage.NodeQueueEntry, 0)
 
 	for {
@@ -136,34 +136,34 @@ func (t *FilesystemTraverser) directoryScheduler(ctx context.Context) error {
 			case <-ctx.Done():
 				return ctx.Err()
 			case nodeID := <-t.processedNodeIDs:
-				delete(processedNodeIDs, nodeID)
+				delete(processingNodeIDs, nodeID)
 			case t.scheduledNodes <- node:
 				pendingNodes = pendingNodes[:len(pendingNodes)-1]
-				processedNodeIDs[node.NodeID] = struct{}{}
+				processingNodeIDs[node.NodeID] = struct{}{}
 			}
 			continue
 		}
 
 		entries, err := t.storage.SelectNodesToList(
 			ctx,
 			t.filesystemSnapshotID,
-			processedNodeIDs,
+			processingNodeIDs,
 			t.selectNodesToListLimit,
 		)
 		if err != nil {
 			return err
 		}
 
 		if len(entries) == 0 {
-			if len(processedNodeIDs) == 0 {
+			if len(processingNodeIDs) == 0 {
 				logging.Info(ctx, "Traversal complete for %s", t.filesystemSnapshotID)
 				return nil
 			}
 			select {
 			case <-ctx.Done():
 				return ctx.Err()
 			case nodeID := <-t.processedNodeIDs:
-				delete(processedNodeIDs, nodeID)
+				delete(processingNodeIDs, nodeID)
 			}
 			continue
 		}
@@ -183,7 +183,12 @@ func (t *FilesystemTraverser) directoryLister(
 	onListedNodes OnListedNodesFunc,
 ) error {
 
-	session, err := t.client.CreateSession(ctx, t.filesystemID, t.filesystemCheckpointID, true)
+	session, err := t.client.CreateSession(
+		ctx,
+		t.filesystemID,
+		t.filesystemCheckpointID,
+		true,
+	)
 	if err != nil {
 		return err
 	}
@@ -232,7 +237,7 @@ func (t *FilesystemTraverser) listNode(
 
 	cookie := node.Cookie
 	for {
-		nodes, nextCookie, err := t.client.ListNodes(
+		children, nextCookie, err := t.client.ListNodes(
 			ctx,
 			session,
 			node.NodeID,
@@ -242,15 +247,15 @@ func (t *FilesystemTraverser) listNode(
 			return err
 		}
 
-		if len(nodes) > 0 {
-			err = onListedNodes(ctx, nodes, session, t.client)
+		if len(children) > 0 {
+			err = onListedNodes(ctx, children, session, t.client)
 			if err != nil {
 				return err
 			}
 		}
 
 		var childDirs []nfs.Node
-		for _, n := range nodes {
+		for _, n := range children {
 			// In case of filesystem scrubbing, ListNodes may return InvalidNodeID
 			// for nodes present in index tablet and absent in shard.
 			// See: https://github.com/ydb-platform/nbs/issues/5094
@@ -275,7 +280,7 @@ func (t *FilesystemTraverser) listNode(
 			return err
 		}
 
-		if nextCookie == "" {
+		if nextCookie == "" { // Listing of this node is finished
 			return nil
 		}
 

cloud/disk_manager/internal/pkg/dataplane/filesystem_snapshot/storage/mocks/storage_mock.go

@@ -151,13 +151,13 @@ func (s *StorageMock) SelectNodesToList(
 func (s *StorageMock) ScheduleChildNodesForListing(
 	ctx context.Context,
 	snapshotID string,
-	nodeID uint64,
+	parentNodeID uint64,
 	nextCookie string,
 	depth uint64,
 	children []nfs.Node,
 ) error {
 
-	args := s.Called(ctx, snapshotID, nodeID, nextCookie, depth, children)
+	args := s.Called(ctx, snapshotID, parentNodeID, nextCookie, depth, children)
 	return args.Error(0)
 }
 

cloud/disk_manager/internal/pkg/dataplane/filesystem_snapshot/storage/storage.go

@@ -118,7 +118,7 @@ type Storage interface {
 	ScheduleChildNodesForListing(
 		ctx context.Context,
 		snapshotID string,
-		nodeID uint64,
+		parentNodeID uint64,
 		nextCookie string,
 		depth uint64,
 		children []nfs.Node,

cloud/disk_manager/internal/pkg/dataplane/filesystem_snapshot/storage/storage_ydb.go

@@ -919,6 +919,7 @@ func (s *storageYDB) selectNodesToList(
 	}
 
 	// Exclude InvalidNodeID to avoid handling empty excludeNodeIDs slice case.
+	// List can't be empty in YDB query.
 	excludeNodeIDs = append(
 		excludeNodeIDs,
 		persistence.Uint64Value(
@@ -973,7 +974,7 @@ func (s *storageYDB) scheduleChildNodesForListing(
 	ctx context.Context,
 	session *persistence.Session,
 	snapshotID string,
-	nodeID uint64,
+	parentNodeID uint64,
 	nextCookie string,
 	depth uint64,
 	children []nfs.Node,
@@ -999,7 +1000,7 @@ func (s *storageYDB) scheduleChildNodesForListing(
 			where filesystem_snapshot_id = $snapshot_id and node_id = $node_id
 		`, s.tablesPath),
 			persistence.ValueParam("$snapshot_id", persistence.UTF8Value(snapshotID)),
-			persistence.ValueParam("$node_id", persistence.Uint64Value(nodeID)),
+			persistence.ValueParam("$node_id", persistence.Uint64Value(parentNodeID)),
 		)
 	} else {
 		_, err = tx.Execute(ctx, fmt.Sprintf(`
@@ -1014,7 +1015,7 @@ func (s *storageYDB) scheduleChildNodesForListing(
 			values ($snapshot_id, $node_id, $cookie, $depth)
 		`, s.tablesPath),
 			persistence.ValueParam("$snapshot_id", persistence.UTF8Value(snapshotID)),
-			persistence.ValueParam("$node_id", persistence.Uint64Value(nodeID)),
+			persistence.ValueParam("$node_id", persistence.Uint64Value(parentNodeID)),
 			persistence.ValueParam("$cookie", persistence.StringValue([]byte(nextCookie))),
 			persistence.ValueParam("$depth", persistence.Uint64Value(depth)),
 		)
@@ -1099,7 +1100,7 @@ func (s *storageYDB) SelectNodesToList(
 func (s *storageYDB) ScheduleChildNodesForListing(
 	ctx context.Context,
 	snapshotID string,
-	nodeID uint64,
+	parentNodeID uint64,
 	nextCookie string,
 	depth uint64,
 	children []nfs.Node,
@@ -1114,7 +1115,7 @@ func (s *storageYDB) ScheduleChildNodesForListing(
 				ctx,
 				session,
 				snapshotID,
-				nodeID,
+				parentNodeID,
 				nextCookie,
 				depth,
 				children,