filer: list diretory pagination make up for the expired entries

weed/filer2/filer.go

@@ -234,15 +234,30 @@ func (f *Filer) FindEntry(ctx context.Context, p FullPath) (entry *Entry, err er
 
 }
 
-func (f *Filer) ListDirectoryEntries(ctx context.Context, p FullPath, startFileName string, inclusive bool, limit int) (entries []*Entry, expiredCount int, err error) {
+func (f *Filer) ListDirectoryEntries(ctx context.Context, p FullPath, startFileName string, inclusive bool, limit int) ([]*Entry, error) {
 	if strings.HasSuffix(string(p), "/") && len(p) > 1 {
 		p = p[0 : len(p)-1]
 	}
+
+	var makeupEntries []*Entry
+	entries, expiredCount, lastFileName, err := f.doListDirectoryEntries(ctx, p, startFileName, inclusive, limit)
+	for expiredCount > 0 && err == nil {
+		makeupEntries, expiredCount, lastFileName, err = f.doListDirectoryEntries(ctx, p, lastFileName, false, expiredCount)
+		if err == nil {
+			entries = append(entries, makeupEntries...)
+		}
+	}
+
+	return entries, err
+}
+
+func (f *Filer) doListDirectoryEntries(ctx context.Context, p FullPath, startFileName string, inclusive bool, limit int) (entries []*Entry, expiredCount int, lastFileName string, err error) {
 	listedEntries, listErr := f.store.ListDirectoryEntries(ctx, p, startFileName, inclusive, limit)
 	if listErr != nil {
-		return listedEntries, expiredCount, err
+		return listedEntries, expiredCount, "", listErr
 	}
 	for _, entry := range listedEntries {
+		lastFileName = entry.Name()
 		if entry.TtlSec > 0 {
 			if entry.Crtime.Add(time.Duration(entry.TtlSec) * time.Second).Before(time.Now()) {
 				f.store.DeleteEntry(ctx, p.Child(entry.Name()))

weed/filer2/filer_buckets.go

@@ -28,7 +28,7 @@ func (f *Filer) LoadBuckets(dirBucketsPath string) {
 
 	limit := math.MaxInt32
 
-	entries, _, err := f.ListDirectoryEntries(context.Background(), FullPath(dirBucketsPath), "", false, limit)
+	entries, err := f.ListDirectoryEntries(context.Background(), FullPath(dirBucketsPath), "", false, limit)
 
 	if err != nil {
 		glog.V(1).Infof("no buckets found: %v", err)

weed/filer2/filer_delete_entry.go

@@ -57,7 +57,7 @@ func (f *Filer) doBatchDeleteFolderMetaAndData(ctx context.Context, entry *Entry
 	lastFileName := ""
 	includeLastFile := false
 	for {
-		entries, _, err := f.ListDirectoryEntries(ctx, entry.FullPath, lastFileName, includeLastFile, PaginationSize)
+		entries, err := f.ListDirectoryEntries(ctx, entry.FullPath, lastFileName, includeLastFile, PaginationSize)
 		if err != nil {
 			glog.Errorf("list folder %s: %v", entry.FullPath, err)
 			return nil, fmt.Errorf("list folder %s: %v", entry.FullPath, err)

weed/filer2/leveldb/leveldb_store_test.go

@@ -48,14 +48,14 @@ func TestCreateAndFind(t *testing.T) {
 	}
 
 	// checking one upper directory
-	entries, _, _ := filer.ListDirectoryEntries(ctx, filer2.FullPath("/home/chris/this/is/one"), "", false, 100)
+	entries, _ := filer.ListDirectoryEntries(ctx, filer2.FullPath("/home/chris/this/is/one"), "", false, 100)
 	if len(entries) != 1 {
 		t.Errorf("list entries count: %v", len(entries))
 		return
 	}
 
 	// checking one upper directory
-	entries, _, _ = filer.ListDirectoryEntries(ctx, filer2.FullPath("/"), "", false, 100)
+	entries, _ = filer.ListDirectoryEntries(ctx, filer2.FullPath("/"), "", false, 100)
 	if len(entries) != 1 {
 		t.Errorf("list entries count: %v", len(entries))
 		return
@@ -75,7 +75,7 @@ func TestEmptyRoot(t *testing.T) {
 	ctx := context.Background()
 
 	// checking one upper directory
-	entries, _, err := filer.ListDirectoryEntries(ctx, filer2.FullPath("/"), "", false, 100)
+	entries, err := filer.ListDirectoryEntries(ctx, filer2.FullPath("/"), "", false, 100)
 	if err != nil {
 		t.Errorf("list entries: %v", err)
 		return

weed/filer2/leveldb2/leveldb2_store_test.go

@@ -48,14 +48,14 @@ func TestCreateAndFind(t *testing.T) {
 	}
 
 	// checking one upper directory
-	entries, _, _ := filer.ListDirectoryEntries(ctx, filer2.FullPath("/home/chris/this/is/one"), "", false, 100)
+	entries, _ := filer.ListDirectoryEntries(ctx, filer2.FullPath("/home/chris/this/is/one"), "", false, 100)
 	if len(entries) != 1 {
 		t.Errorf("list entries count: %v", len(entries))
 		return
 	}
 
 	// checking one upper directory
-	entries, _, _ = filer.ListDirectoryEntries(ctx, filer2.FullPath("/"), "", false, 100)
+	entries, _ = filer.ListDirectoryEntries(ctx, filer2.FullPath("/"), "", false, 100)
 	if len(entries) != 1 {
 		t.Errorf("list entries count: %v", len(entries))
 		return
@@ -75,7 +75,7 @@ func TestEmptyRoot(t *testing.T) {
 	ctx := context.Background()
 
 	// checking one upper directory
-	entries, _, err := filer.ListDirectoryEntries(ctx, filer2.FullPath("/"), "", false, 100)
+	entries, err := filer.ListDirectoryEntries(ctx, filer2.FullPath("/"), "", false, 100)
 	if err != nil {
 		t.Errorf("list entries: %v", err)
 		return

weed/server/filer_grpc_server.go

@@ -53,7 +53,7 @@ func (fs *FilerServer) ListEntries(req *filer_pb.ListEntriesRequest, stream file
 	lastFileName := req.StartFromFileName
 	includeLastFile := req.InclusiveStartFrom
 	for limit > 0 {
-		entries, expiredCount, err := fs.filer.ListDirectoryEntries(stream.Context(), filer2.FullPath(req.Directory), lastFileName, includeLastFile, paginationLimit)
+		entries, err := fs.filer.ListDirectoryEntries(stream.Context(), filer2.FullPath(req.Directory), lastFileName, includeLastFile, paginationLimit)
 
 		if err != nil {
 			return err
@@ -92,7 +92,7 @@ func (fs *FilerServer) ListEntries(req *filer_pb.ListEntriesRequest, stream file
 			}
 		}
 
-		if len(entries)+expiredCount < paginationLimit {
+		if len(entries) < paginationLimit {
 			break
 		}
 

weed/server/filer_grpc_server_rename.go

@@ -68,7 +68,7 @@ func (fs *FilerServer) moveFolderSubEntries(ctx context.Context, oldParent filer
 	includeLastFile := false
 	for {
 
-		entries, _, err := fs.filer.ListDirectoryEntries(ctx, currentDirPath, lastFileName, includeLastFile, 1024)
+		entries, err := fs.filer.ListDirectoryEntries(ctx, currentDirPath, lastFileName, includeLastFile, 1024)
 		if err != nil {
 			return err
 		}

weed/server/filer_server_handlers_read_dir.go

@@ -32,7 +32,7 @@ func (fs *FilerServer) listDirectoryHandler(w http.ResponseWriter, r *http.Reque
 
 	lastFileName := r.FormValue("lastFileName")
 
-	entries, expiredCount, err := fs.filer.ListDirectoryEntries(context.Background(), filer2.FullPath(path), lastFileName, false, limit)
+	entries, err := fs.filer.ListDirectoryEntries(context.Background(), filer2.FullPath(path), lastFileName, false, limit)
 
 	if err != nil {
 		glog.V(0).Infof("listDirectory %s %s %d: %s", path, lastFileName, limit, err)
@@ -40,7 +40,7 @@ func (fs *FilerServer) listDirectoryHandler(w http.ResponseWriter, r *http.Reque
 		return
 	}
 
-	shouldDisplayLoadMore := len(entries)+expiredCount == limit
+	shouldDisplayLoadMore := len(entries) == limit
 	if path == "/" {
 		path = ""
 	}