filer: fix prefixed listing when the resume point sorts below the prefix (#12049)

* test: prefixed directory listing with start below the prefix

A listing constrained to prefix "item-b" and resumed from "item" must
return item-b.versions. The seek-based stores return an empty page
because the iterator lands before the prefix range and breaks on the
first non-matching name.

Generated with [Devin](https://devin.ai)

Co-Authored-By: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com>

* filer: clamp prefixed-listing seek to the prefix range

Seek-based stores (leveldb, leveldb2, leveldb3, rocksdb, tikv, etcd)
started the scan at startFileName and stopped at the first key outside
the prefix. A startFileName below the prefix landed the iterator on a
non-matching sibling first, so the listing returned an empty page even
though matching entries followed. etcd had the mirror problem: a range
beginning below the prefix end leaked non-matching entries. Begin the
seek at max(startFileName, prefix), which is what foundationdb and the
sql stores already do.

Generated with [Devin](https://devin.ai)

Co-Authored-By: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com>

* filer(hbase): honor startFileName lower bound in prefixed listings

The scan starts at the prefix but only skipped the exact startFileName,
so entries sorting before it were returned again on resume.

Generated with [Devin](https://devin.ai)

Co-Authored-By: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com>

* filer: keep paging a prefixed listing while matches can still follow

The fallback for stores without native prefix support stopped paging as
soon as the last scanned name reached the prefix, dropping prefix
matches that continue on the next page.

Generated with [Devin](https://devin.ai)

Co-Authored-By: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com>

* filer: return the last delivered name from the prefixed-listing fallback

The fallback for stores without native prefix support pages raw entries
and filters client-side. Keep the scan cursor and the returned cursor
separate: resuming must continue after the last delivered match, not the
last scanned name, or matches still inside the current raw page are
skipped on the next page. Also keep paging while the scan cursor is
inside or below the prefix range instead of stopping at the first name
that reaches the prefix.

Generated with [Devin](https://devin.ai)

Co-Authored-By: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com>

* filer(hbase): start prefixed-listing scans at the resume key

Generated with [Devin](https://devin.ai)

Co-Authored-By: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com>

---------

Co-authored-by: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com>

weed/filer/etcd/etcd_store.go

@@ -181,7 +181,7 @@ func (store *EtcdStore) DeleteFolderChildren(ctx context.Context, fullpath weed_
 func (store *EtcdStore) ListDirectoryPrefixedEntries(ctx context.Context, dirPath weed_util.FullPath, startFileName string, includeStartFile bool, limit int64, prefix string, eachEntryFunc filer.ListEachEntryFunc) (lastFileName string, err error) {
 	directoryPrefix := genDirectoryKeyPrefix(dirPath, prefix)
 	lastFileStart := directoryPrefix
-	if startFileName != "" {
+	if startFileName > prefix {
 		lastFileStart = genDirectoryKeyPrefix(dirPath, startFileName)
 	}
 

weed/filer/filerstore_wrapper.go

@@ -400,46 +400,39 @@ func (fsw *FilerStoreWrapper) prefixFilterEntries(ctx context.Context, dirPath u
 		return actualStore.ListDirectoryEntries(ctx, dirPath, startFileName, includeStartFile, limit, eachEntryFunc)
 	}
 
-	var notPrefixed []*Entry
-	lastFileName, err = actualStore.ListDirectoryEntries(ctx, dirPath, startFileName, includeStartFile, limit, func(entry *Entry) (bool, error) {
-		notPrefixed = append(notPrefixed, entry)
-		return true, nil
-	})
-	if err != nil {
-		return
-	}
-
+	// lastFileName must stay at the last name delivered to eachEntryFunc; the
+	// raw scan cursor runs ahead over non-matching names and cannot be returned.
 	count := int64(0)
-	for count < limit && len(notPrefixed) > 0 {
-		for _, entry := range notPrefixed {
-			if strings.HasPrefix(entry.Name(), prefix) {
-				count++
-				res, resErr := eachEntryFunc(entry)
-
-				if resErr != nil {
-					err = fmt.Errorf("failed to process eachEntryFunc for entry %q: %w", entry.Name(), resErr)
-					return
-				}
-
-				if !res {
-					return
-				}
-				if count >= limit {
-					break
-				}
-			}
+	scanName, scanInclusive := startFileName, includeStartFile
+	for count < limit {
+		var page []*Entry
+		if scanName, err = actualStore.ListDirectoryEntries(ctx, dirPath, scanName, scanInclusive, limit, func(entry *Entry) (bool, error) {
+			page = append(page, entry)
+			return true, nil
+		}); err != nil {
+			return
 		}
-		if count < limit && lastFileName < prefix {
-			notPrefixed = notPrefixed[:0]
-			lastFileName, err = actualStore.ListDirectoryEntries(ctx, dirPath, lastFileName, false, limit, func(entry *Entry) (bool, error) {
-				notPrefixed = append(notPrefixed, entry)
-				return true, nil
-			})
-			if err != nil {
+		scanInclusive = false
+		if len(page) == 0 {
+			return
+		}
+		for _, entry := range page {
+			if !strings.HasPrefix(entry.Name(), prefix) {
+				continue
+			}
+			count++
+			res, resErr := eachEntryFunc(entry)
+			if resErr != nil {
+				err = fmt.Errorf("failed to process eachEntryFunc for entry %q: %w", entry.Name(), resErr)
 				return
 			}
-		} else {
-			break
+			lastFileName = entry.Name()
+			if !res || count >= limit {
+				return
+			}
+		}
+		if scanName >= prefix && !strings.HasPrefix(scanName, prefix) {
+			return
 		}
 	}
 	return

weed/filer/filerstore_wrapper_test.go

@@ -205,3 +205,41 @@ func TestFilerStoreWrapperRollbackSucceedsWithCancelledContext(t *testing.T) {
 	assert.NoError(t, wrapper.RollbackTransaction(cancelledCtx()))
 	assert.NoError(t, wrapper.RollbackTransaction(expiredCtx()))
 }
+
+type noPrefixListStore struct {
+	*stubFilerStore
+}
+
+func (s *noPrefixListStore) ListDirectoryPrefixedEntries(context.Context, util.FullPath, string, bool, int64, string, ListEachEntryFunc) (string, error) {
+	return "", ErrUnsupportedListDirectoryPrefixed
+}
+
+// The wrapper fallback must return the last delivered name as the cursor, not
+// the end of the raw page, and must keep paging while prefix matches can still
+// follow the raw cursor.
+func TestPrefixFilterEntriesCursorStaysAtLastDelivered(t *testing.T) {
+	store := &noPrefixListStore{newStubFilerStore()}
+	wrapper := NewFilerStoreWrapper(store)
+	ctx := context.Background()
+	for _, name := range []string{"a", "b1", "b2", "b3", "b4"} {
+		require.NoError(t, store.InsertEntry(ctx, &Entry{FullPath: util.FullPath("/d/" + name)}))
+	}
+
+	var delivered []string
+	lastFileName, err := wrapper.ListDirectoryPrefixedEntries(ctx, util.FullPath("/d"), "", false, 2, "b", func(entry *Entry) (bool, error) {
+		delivered = append(delivered, entry.Name())
+		return true, nil
+	})
+	require.NoError(t, err)
+	assert.Equal(t, []string{"b1", "b2"}, delivered)
+	assert.Equal(t, "b2", lastFileName)
+
+	delivered = delivered[:0]
+	lastFileName, err = wrapper.ListDirectoryPrefixedEntries(ctx, util.FullPath("/d"), lastFileName, false, 10, "b", func(entry *Entry) (bool, error) {
+		delivered = append(delivered, entry.Name())
+		return true, nil
+	})
+	require.NoError(t, err)
+	assert.Equal(t, []string{"b3", "b4"}, delivered)
+	assert.Equal(t, "b4", lastFileName)
+}

weed/filer/hbase/hbase_store.go

@@ -156,7 +156,11 @@ func (store *HbaseStore) ListDirectoryEntries(ctx context.Context, dirPath util.
 func (store *HbaseStore) ListDirectoryPrefixedEntries(ctx context.Context, dirPath util.FullPath, startFileName string, includeStartFile bool, limit int64, prefix string, eachEntryFunc filer.ListEachEntryFunc) (lastFileName string, err error) {
 	family := map[string][]string{store.cfMetaDir: {COLUMN_NAME}}
 	expectedPrefix := []byte(dirPath.Child(prefix))
-	scan, err := hrpc.NewScanRange(ctx, store.table, expectedPrefix, nil, hrpc.Families(family))
+	scanStart := expectedPrefix
+	if resumeKey := []byte(dirPath.Child(startFileName)); bytes.Compare(resumeKey, scanStart) > 0 {
+		scanStart = resumeKey
+	}
+	scan, err := hrpc.NewScanRange(ctx, store.table, scanStart, nil, hrpc.Families(family))
 	if err != nil {
 		return lastFileName, err
 	}
@@ -188,7 +192,7 @@ func (store *HbaseStore) ListDirectoryPrefixedEntries(ctx context.Context, dirPa
 
 		value := cell.Value
 
-		if fileName == startFileName && !includeStartFile {
+		if fileName < startFileName || (fileName == startFileName && !includeStartFile) {
 			continue
 		}
 

weed/filer/leveldb/leveldb_store.go

@@ -221,7 +221,7 @@ func (store *LevelDBStore) ListDirectoryPrefixedEntries(ctx context.Context, dir
 
 	directoryPrefix := genDirectoryKeyPrefix(dirPath, prefix)
 	lastFileStart := directoryPrefix
-	if startFileName != "" {
+	if startFileName > prefix {
 		lastFileStart = genDirectoryKeyPrefix(dirPath, startFileName)
 	}
 

weed/filer/leveldb/leveldb_store_test.go

@@ -110,3 +110,28 @@ func BenchmarkInsertEntry(b *testing.B) {
 		store.InsertEntry(ctx, entry)
 	}
 }
+
+func TestListDirectoryPrefixedEntries_StartBelowPrefix(t *testing.T) {
+	store := &LevelDBStore{}
+	store.initialize(t.TempDir())
+	defer store.Shutdown()
+
+	ctx := context.Background()
+	for _, name := range []string{"item-a", "item-a.versions", "item-b.versions"} {
+		if err := store.InsertEntry(ctx, &filer.Entry{FullPath: util.FullPath("/d/" + name)}); err != nil {
+			t.Fatalf("insert %s: %v", name, err)
+		}
+	}
+
+	var names []string
+	_, err := store.ListDirectoryPrefixedEntries(ctx, util.FullPath("/d"), "item", true, 100, "item-b", func(entry *filer.Entry) (bool, error) {
+		names = append(names, entry.Name())
+		return true, nil
+	})
+	if err != nil {
+		t.Fatalf("list: %v", err)
+	}
+	if len(names) != 1 || names[0] != "item-b.versions" {
+		t.Fatalf("want [item-b.versions], got %v", names)
+	}
+}

weed/filer/leveldb2/leveldb2_store.go

@@ -239,7 +239,7 @@ func (store *LevelDB2Store) ListDirectoryPrefixedEntries(ctx context.Context, di
 
 	directoryPrefix, partitionId := genDirectoryKeyPrefix(dirPath, prefix, store.dbCount)
 	lastFileStart := directoryPrefix
-	if startFileName != "" {
+	if startFileName > prefix {
 		lastFileStart, _ = genDirectoryKeyPrefix(dirPath, startFileName, store.dbCount)
 	}
 

weed/filer/leveldb2/leveldb2_store_test.go

@@ -151,6 +151,31 @@ func TestInitializeRejectsOversizedOption(t *testing.T) {
 	}
 }
 
+func TestListDirectoryPrefixedEntries_StartBelowPrefix(t *testing.T) {
+	store := &LevelDB2Store{}
+	store.initialize(t.TempDir(), 2)
+	defer store.Shutdown()
+
+	ctx := context.Background()
+	for _, name := range []string{"item-a", "item-a.versions", "item-b.versions"} {
+		if err := store.InsertEntry(ctx, &filer.Entry{FullPath: util.FullPath("/d/" + name)}); err != nil {
+			t.Fatalf("insert %s: %v", name, err)
+		}
+	}
+
+	var names []string
+	_, err := store.ListDirectoryPrefixedEntries(ctx, util.FullPath("/d"), "item", true, 100, "item-b", func(entry *filer.Entry) (bool, error) {
+		names = append(names, entry.Name())
+		return true, nil
+	})
+	if err != nil {
+		t.Fatalf("list: %v", err)
+	}
+	if len(names) != 1 || names[0] != "item-b.versions" {
+		t.Fatalf("want [item-b.versions], got %v", names)
+	}
+}
+
 func TestEmptyRoot(t *testing.T) {
 	testFiler := filer.NewFiler(pb.ServerDiscovery{}, nil, "", "", "", "", "", 255, nil)
 	dir := t.TempDir()

weed/filer/leveldb3/leveldb3_store.go

@@ -370,7 +370,7 @@ func (store *LevelDB3Store) ListDirectoryPrefixedEntries(ctx context.Context, di
 
 	directoryPrefix := genDirectoryKeyPrefix(shortPath, prefix)
 	lastFileStart := directoryPrefix
-	if startFileName != "" {
+	if startFileName > prefix {
 		lastFileStart = genDirectoryKeyPrefix(shortPath, startFileName)
 	}
 

weed/filer/leveldb3/leveldb3_store_test.go

@@ -182,3 +182,28 @@ func TestEmptyRoot(t *testing.T) {
 	}
 
 }
+
+func TestListDirectoryPrefixedEntries_StartBelowPrefix(t *testing.T) {
+	store := &LevelDB3Store{}
+	store.initialize(t.TempDir())
+	defer store.Shutdown()
+
+	ctx := context.Background()
+	for _, name := range []string{"item-a", "item-a.versions", "item-b.versions"} {
+		if err := store.InsertEntry(ctx, &filer.Entry{FullPath: util.FullPath("/d/" + name)}); err != nil {
+			t.Fatalf("insert %s: %v", name, err)
+		}
+	}
+
+	var names []string
+	_, err := store.ListDirectoryPrefixedEntries(ctx, util.FullPath("/d"), "item", true, 100, "item-b", func(entry *filer.Entry) (bool, error) {
+		names = append(names, entry.Name())
+		return true, nil
+	})
+	if err != nil {
+		t.Fatalf("list: %v", err)
+	}
+	if len(names) != 1 || names[0] != "item-b.versions" {
+		t.Fatalf("want [item-b.versions], got %v", names)
+	}
+}

weed/filer/rocksdb/rocksdb_store.go

@@ -243,7 +243,7 @@ func (store *RocksDBStore) ListDirectoryPrefixedEntries(ctx context.Context, dir
 
 	directoryPrefix := genDirectoryKeyPrefix(dirPath, prefix)
 	lastFileStart := directoryPrefix
-	if startFileName != "" {
+	if startFileName > prefix {
 		lastFileStart = genDirectoryKeyPrefix(dirPath, startFileName)
 	}
 

weed/filer/tikv/tikv_store.go

@@ -246,7 +246,7 @@ func (store *TikvStore) ListDirectoryPrefixedEntries(ctx context.Context, dirPat
 	lastFileName := ""
 	directoryPrefix := store.genDirectoryKeyPrefix(dirPath, prefix)
 	lastFileStart := directoryPrefix
-	if startFileName != "" {
+	if startFileName > prefix {
 		lastFileStart = store.genDirectoryKeyPrefix(dirPath, startFileName)
 	}